Skip to content

fix: Prevent TestDataV2 from briefly restoring an older flag when a client starts #491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
keelerm84 merged 1 commit into from
Nov 6, 2026
Merged

fix: Prevent TestDataV2 from briefly restoring an older flag when a client starts #491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 61 additions & 55 deletions lib/ldclient-rb/impl/integrations/test_data/test_data_source_v2.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,6 @@ def initialize(test_data)
@closed = false
@update_queue = Queue.new
@lock = Mutex.new

# Always register for change notifications
@test_data.add_instance(self)
end

#
Expand All @@ -53,68 +50,25 @@ def name
# @return [LaunchDarkly::Interfaces::DataSystem::FetchResult]
#
def fetch(selector_store)
result =
begin
@lock.synchronize do
if @closed
next LaunchDarkly::Result.fail('TestDataV2 source has been closed')
end

# Get all current flags and segments from test data
init_data = @test_data.make_init_data
version = @test_data.get_version

# Build a full transfer changeset
builder = LaunchDarkly::Interfaces::DataSystem::ChangeSetBuilder.new
builder.start(LaunchDarkly::Interfaces::DataSystem::IntentCode::TRANSFER_FULL)

# Add all flags to the changeset
init_data[:flags].each do |key, flag_data|
builder.add_put(
LaunchDarkly::Interfaces::DataSystem::ObjectKind::FLAG,
key,
flag_data[:version] || 1,
flag_data
)
end

# Add all segments to the changeset
init_data[:segments].each do |key, segment_data|
builder.add_put(
LaunchDarkly::Interfaces::DataSystem::ObjectKind::SEGMENT,
key,
segment_data[:version] || 1,
segment_data
)
end

# Create selector for this version
selector = LaunchDarkly::Interfaces::DataSystem::Selector.new_selector(version.to_s, version)
change_set = builder.finish(selector)

basis = LaunchDarkly::Interfaces::DataSystem::Basis.new(change_set: change_set, persist: false, environment_id: nil)

LaunchDarkly::Result.success(basis)
end
rescue => e
LaunchDarkly::Result.fail("Error fetching test data: #{e.message}", e)
end

LaunchDarkly::Interfaces::DataSystem::FetchResult.new(result: result, fallback_to_fdv1: false)
fetch_result do
init_data = @test_data.make_init_data
[init_data, @test_data.get_version]
end
end

#
# Implementation of the Synchronizer.sync method.
#
# Yields updates as test data changes occur.
# Yields the current test data, and then an update for each later change.
#
# @param selector_store [LaunchDarkly::Interfaces::DataSystem::SelectorStore] Provides the Selector (unused for test data)
# @yield [LaunchDarkly::Interfaces::DataSystem::Update] Yields Update objects as synchronization progresses
# @return [void]
#
def sync(selector_store)
# First yield initial data
initial_fetch = fetch(selector_store)
# Subscribe and take the initial data in one step. Then the queue holds only the changes
# that the initial data does not hold.
initial_fetch = fetch_result { @test_data.subscribe(self) }
unless initial_fetch.success?
yield LaunchDarkly::Interfaces::DataSystem::Update.new(
state: LaunchDarkly::Interfaces::DataSource::Status::OFF,
Expand Down Expand Up @@ -287,9 +241,61 @@ def upsert_segment(segment_data)
end
end
end

#
# Builds a full transfer result from the data and the version that the block returns.
# The block runs while this instance holds its lock.
#
private def fetch_result
result =
begin
@lock.synchronize do
if @closed
next LaunchDarkly::Result.fail('TestDataV2 source has been closed')
end

init_data, version = yield

# Build a full transfer changeset
builder = LaunchDarkly::Interfaces::DataSystem::ChangeSetBuilder.new
builder.start(LaunchDarkly::Interfaces::DataSystem::IntentCode::TRANSFER_FULL)

# Add all flags to the changeset
init_data[:flags].each do |key, flag_data|
builder.add_put(
LaunchDarkly::Interfaces::DataSystem::ObjectKind::FLAG,
key,
flag_data[:version] || 1,
flag_data
)
end

# Add all segments to the changeset
init_data[:segments].each do |key, segment_data|
builder.add_put(
LaunchDarkly::Interfaces::DataSystem::ObjectKind::SEGMENT,
key,
segment_data[:version] || 1,
segment_data
)
end

# Create selector for this version
selector = LaunchDarkly::Interfaces::DataSystem::Selector.new_selector(version.to_s, version)
change_set = builder.finish(selector)

basis = LaunchDarkly::Interfaces::DataSystem::Basis.new(change_set: change_set, persist: false, environment_id: nil)

LaunchDarkly::Result.success(basis)
end
rescue => e
LaunchDarkly::Result.fail("Error fetching test data: #{e.message}", e)
end

LaunchDarkly::Interfaces::DataSystem::FetchResult.new(result: result, fallback_to_fdv1: false)
end
end
end
end
end
end

10 changes: 9 additions & 1 deletion lib/ldclient-rb/integrations/test_data_v2.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,10 +167,18 @@ def closed_instance(instance)
end

# @api private
#
# Adds an instance and takes the current data in one step. The instance then receives each
# change that the returned data does not hold, and no change that it holds.
#
# @param instance [LaunchDarkly::Impl::Integrations::TestData::TestDataSourceV2] the TestDataSourceV2 instance to add
def add_instance(instance)
# @return [Array(Hash, Integer)] the data, in the form of {#make_init_data}, and the next version
def subscribe(instance)
@lock.with_write_lock do
@instances.push(instance)
version = @version
@version += 1
[{ flags: @current_flags.dup, segments: @current_segments.dup }, version]
end
end

Expand Down
33 changes: 32 additions & 1 deletion spec/integrations/test_data_v2_spec.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ module Integrations
it 'sends a preconfigured flag to a running data source' do
td = TestDataV2.data_source
source = LaunchDarkly::Impl::Integrations::TestData::TestDataSourceV2.new(td)
td.subscribe(source)
expect(source).to receive(:upsert_flag).with(hash_including(key: 'myflag', version: 7))
td.use_preconfigured_flag({ key: 'myflag', version: 7, variations: ['a'] })
end
Expand Down Expand Up @@ -242,7 +243,7 @@ module Integrations

def expect_delivery_in_order
td = TestDataV2.data_source
td.add_instance(instance)
td.subscribe(instance)
first = Thread.new { yield td, 1 }
second = nil
begin
Expand Down Expand Up @@ -308,6 +309,36 @@ def expect_delivery_in_order
end

describe 'TestDataSourceV2' do
it 'does not deliver a change from before sync starts after the initial data' do
td = TestDataV2.data_source
source = LaunchDarkly::Impl::Integrations::TestData::TestDataSourceV2.new(td)
td.update(td.flag('myflag').variation_for_all(true))
td.update(td.flag('myflag').variation_for_all(false))

updates = Queue.new
sync_thread = Thread.new { source.sync(nil) { |update| updates << update } }
begin
initial = updates.pop(timeout: 1)
td.update(td.flag('myflag').variation_for_all(true))
following = updates.pop(timeout: 1)

expect(initial.change_set.changes.map(&:version)).to eq([2])
expect(following.change_set.changes.map(&:version)).to eq([3])
ensure
source.stop
sync_thread.join(1)
end
end

it 'does not receive changes after it is used as an initializer' do
td = TestDataV2.data_source
source = LaunchDarkly::Impl::Integrations::TestData::TestDataSourceV2.new(td)
source.fetch(nil)

expect(source).not_to receive(:upsert_flag)
td.update(td.flag('myflag').variation_for_all(true))
end

it 'includes both flags and segments in fetch' do
td = TestDataV2.data_source
td.update(td.flag('myflag').variation_for_all(true))
Expand Down
Loading