From e5b88af673f49b813b8abca5b285d6b50cc98f5c Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 6 Nov 2026 10:54:46 -0500 Subject: [PATCH] fix: Prevent TestDataV2 from briefly restoring an older flag when a client starts A TestDataV2 data source registered for changes when it was created, not when it started to sync. A change made between those two points was in the initial data and also waited in the queue. The sync then sent the queued change after the initial data, so the store went back to an older version until the later changes arrived. The sync now registers and takes the initial data in one step, under the lock that an update holds while it saves a change and copies the list of instances. Each change then reaches a synchronizer once, and in order. An initializer only fetches, so it no longer registers. Before this change, FDv2 never stopped its initializers, and each one stayed registered and queued every later change. --- .../test_data/test_data_source_v2.rb | 116 +++++++++--------- lib/ldclient-rb/integrations/test_data_v2.rb | 10 +- spec/integrations/test_data_v2_spec.rb | 33 ++++- 3 files changed, 102 insertions(+), 57 deletions(-) diff --git a/lib/ldclient-rb/impl/integrations/test_data/test_data_source_v2.rb b/lib/ldclient-rb/impl/integrations/test_data/test_data_source_v2.rb index b0b91071..990ef572 100644 --- a/lib/ldclient-rb/impl/integrations/test_data/test_data_source_v2.rb +++ b/lib/ldclient-rb/impl/integrations/test_data/test_data_source_v2.rb @@ -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 # @@ -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, @@ -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 - diff --git a/lib/ldclient-rb/integrations/test_data_v2.rb b/lib/ldclient-rb/integrations/test_data_v2.rb index eabd4778..e742d3d5 100644 --- a/lib/ldclient-rb/integrations/test_data_v2.rb +++ b/lib/ldclient-rb/integrations/test_data_v2.rb @@ -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 diff --git a/spec/integrations/test_data_v2_spec.rb b/spec/integrations/test_data_v2_spec.rb index fdfaf0f2..cb76d424 100644 --- a/spec/integrations/test_data_v2_spec.rb +++ b/spec/integrations/test_data_v2_spec.rb @@ -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 @@ -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 @@ -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))