From a1c69256963e564fda9a92e24f04f76af8288a3f Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 2 Sep 2026 11:44:20 -0500 Subject: [PATCH 1/2] fix: Wait for status listener delivery in data source specs --- spec/impl/data_source/polling_spec.rb | 15 +++++-- spec/spec_helper.rb | 57 +++++++++++++++++++++++++-- 2 files changed, 65 insertions(+), 7 deletions(-) diff --git a/spec/impl/data_source/polling_spec.rb b/spec/impl/data_source/polling_spec.rb index 3dcfdc30..c61ed3a1 100644 --- a/spec/impl/data_source/polling_spec.rb +++ b/spec/impl/data_source/polling_spec.rb @@ -78,8 +78,11 @@ def with_processor(store, initialize_to_valid = false) expect(store.get(Impl::DataStore::FEATURES, "flagkey")).to eq(flag) expect(store.get(Impl::DataStore::SEGMENTS, "segkey")).to eq(segment) - expect(listener.statuses.count).to eq(1) - expect(listener.statuses[0].state).to eq(Interfaces::DataSource::Status::VALID) + # The poll thread sets the ready event before it publishes the VALID + # status, so wait for the listener to receive it. + statuses = listener.wait_for_status + expect(statuses.count).to eq(1) + expect(statuses[0].state).to eq(Interfaces::DataSource::Status::VALID) end end end @@ -166,9 +169,13 @@ def verify_recoverable_http_error(status) expect(finished).to be false expect(processor.initialized?).to be false - expect(listener.statuses.count).to eq(2) + # The ready event is never set for a recoverable error, so the wait + # above only passes time. Wait for the poll thread to publish the + # INTERRUPTED status before asserting on it. + statuses = listener.wait_for_count(2) + expect(statuses.count).to eq(2) - s = listener.statuses[1] + s = statuses[1] expect(s.state).to eq(Interfaces::DataSource::Status::INTERRUPTED) expect(s.last_error.status_code).to eq(status) end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 83f40de9..f953022b 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -37,15 +37,66 @@ def update(status) end end +# +# A test listener that records every event it receives. +# +# A data source can deliver events from its own thread. A spec that starts a +# data source and then reads `statuses` at once can run before the event +# arrives. Use `wait_for_count` or `wait_for_status` to block until the events +# you expect have arrived, then assert on the returned array. +# class ListenerSpy - attr_reader :statuses - def initialize + @mutex = Mutex.new + @condition = ConditionVariable.new @statuses = [] end + # + # Returns a copy of the events received so far. + # + # @return [Array] + # + def statuses + @mutex.synchronize { @statuses.dup } + end + def update(status) - @statuses << status + @mutex.synchronize do + @statuses << status + @condition.broadcast + end + end + + # + # Blocks until at least `count` events have arrived, or until the timeout + # passes. Returns a copy of the events received so far. The caller must + # still assert on the result; this method does not fail on timeout. + # + # @param count [Integer] the number of events to wait for + # @param timeout [Numeric] the maximum time to wait, in seconds + # @return [Array] + # + def wait_for_count(count, timeout: 2) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + @mutex.synchronize do + while @statuses.count < count + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + break if remaining <= 0 + @condition.wait(@mutex, remaining) + end + @statuses.dup + end + end + + # + # Blocks until at least one event has arrived, or until the timeout passes. + # + # @param timeout [Numeric] the maximum time to wait, in seconds + # @return [Array] + # + def wait_for_status(timeout: 2) + wait_for_count(1, timeout: timeout) end end From b3fcc6565d0fa67502df482b90a39d26894dd44d Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 9 Sep 2026 16:57:25 -0500 Subject: [PATCH 2/2] fix: Publish the data source status before releasing ready waiters FDv1 polling and the FDv2 synchronizer loop both set the ready event before publishing the VALID data source status. A caller that returned from start could therefore read a status that did not yet reflect the successful poll. Both now publish the status first and set the ready event afterwards, matching the OFF path fixed in #429. The polling spec no longer needs to wait for the VALID status, because the broadcaster notifies listeners inline on the poll thread and the ready event is now set after that. The recoverable-error spec keeps its wait: that path publishes INTERRUPTED and never sets the ready event, so there is nothing to synchronize against. --- lib/ldclient-rb/impl/data_source/polling.rb | 14 +++++++--- lib/ldclient-rb/impl/data_system/fdv2.rb | 12 +++++---- spec/impl/data_source/polling_spec.rb | 26 +++++++++++++++---- spec/impl/data_system/fdv2_datasystem_spec.rb | 19 ++++++++++++++ spec/spec_helper.rb | 14 ++-------- 5 files changed, 59 insertions(+), 26 deletions(-) diff --git a/lib/ldclient-rb/impl/data_source/polling.rb b/lib/ldclient-rb/impl/data_source/polling.rb index 6f01138b..2298fa22 100644 --- a/lib/ldclient-rb/impl/data_source/polling.rb +++ b/lib/ldclient-rb/impl/data_source/polling.rb @@ -38,14 +38,20 @@ def poll begin all_data, headers = request_all_data DataSource.record_environment_id(@config.data_source_update_sink, headers) + newly_initialized = false if all_data update_sink_or_data_store.init(all_data) - if @initialized.make_true - @config.logger.info { "[LDClient] Polling connection initialized" } - @ready.set - end + newly_initialized = @initialized.make_true end @config.data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::VALID, nil) + + if newly_initialized + @config.logger.info { "[LDClient] Polling connection initialized" } + # Publish the VALID status before releasing anyone waiting on the + # ready event, so a client that returns from start can rely on the + # data source status already reflecting the successful poll. + @ready.set + end rescue JSON::ParserError => e @config.logger.error { "[LDClient] JSON parsing failed for polling response." } error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new( diff --git a/lib/ldclient-rb/impl/data_system/fdv2.rb b/lib/ldclient-rb/impl/data_system/fdv2.rb index 15994d7e..c077544e 100644 --- a/lib/ldclient-rb/impl/data_system/fdv2.rb +++ b/lib/ldclient-rb/impl/data_system/fdv2.rb @@ -473,15 +473,17 @@ def consume_synchronizer_results(synchronizer, check_recovery: false) # Handle the update @store.apply(update.change_set, true) if update.change_set - # Set ready event on valid update - if update.state == LaunchDarkly::Interfaces::DataSource::Status::VALID - @ready_event.set - record_environment_id(update.environment_id) - end + valid = update.state == LaunchDarkly::Interfaces::DataSource::Status::VALID + record_environment_id(update.environment_id) if valid # Update status @data_source_status_provider.update_status(update.state, update.error) + # Publish the status before releasing anyone waiting on the ready + # event, so a client that returns from start can rely on the data + # source status already reflecting the update. + @ready_event.set if valid + return SyncResult::FDV1 if update.fallback_to_fdv1 return SyncResult::REMOVE if update.state == LaunchDarkly::Interfaces::DataSource::Status::OFF diff --git a/spec/impl/data_source/polling_spec.rb b/spec/impl/data_source/polling_spec.rb index c61ed3a1..4dc8ada8 100644 --- a/spec/impl/data_source/polling_spec.rb +++ b/spec/impl/data_source/polling_spec.rb @@ -78,11 +78,27 @@ def with_processor(store, initialize_to_valid = false) expect(store.get(Impl::DataStore::FEATURES, "flagkey")).to eq(flag) expect(store.get(Impl::DataStore::SEGMENTS, "segkey")).to eq(segment) - # The poll thread sets the ready event before it publishes the VALID - # status, so wait for the listener to receive it. - statuses = listener.wait_for_status - expect(statuses.count).to eq(1) - expect(statuses[0].state).to eq(Interfaces::DataSource::Status::VALID) + expect(listener.statuses.count).to eq(1) + expect(listener.statuses[0].state).to eq(Interfaces::DataSource::Status::VALID) + end + end + + it 'publishes the valid status before releasing ready waiters' do + allow(requestor).to receive(:request_all_data).and_return(all_data) + store = InMemoryFeatureStore.new + with_processor(store) do |processor| + # The broadcaster notifies listeners inline on the poll thread, so a + # listener that sees the ready event already set proves the status was + # published too late. + ready = processor.instance_variable_get(:@ready) + ready_set_when_notified = nil + status_broadcaster.add_listener(CallbackListener.new(->(_status) { ready_set_when_notified = ready.set? })) + + config = processor.instance_variable_get(:@config) + processor.start.wait + + expect(ready_set_when_notified).to be false + expect(config.data_source_update_sink.current_status.state).to eq(Interfaces::DataSource::Status::VALID) end end end diff --git a/spec/impl/data_system/fdv2_datasystem_spec.rb b/spec/impl/data_system/fdv2_datasystem_spec.rb index 0f2759bc..1fc7f6dc 100644 --- a/spec/impl/data_system/fdv2_datasystem_spec.rb +++ b/spec/impl/data_system/fdv2_datasystem_spec.rb @@ -697,6 +697,25 @@ def build(_sdk_key, _config) FDv2.new(sdk_key, LaunchDarkly::Config.new(logger: logger), data_system_config) end end + + describe "data source status" do + it "publishes the valid status before releasing ready waiters" do + td = LaunchDarkly::Integrations::TestDataV2.data_source + td.update(td.flag("flagkey").on(true)) + + data_system_config = LaunchDarkly::DataSystem::ConfigBuilder.new + .synchronizers([td.test_data_ds_builder]) + .build + + fdv2 = FDv2.new(sdk_key, config, data_system_config) + + ready_event = fdv2.start + expect(ready_event.wait(2)).to be true + expect(fdv2.data_source_status_provider.status.state).to eq(LaunchDarkly::Interfaces::DataSource::Status::VALID) + + fdv2.stop + end + end end end end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index f953022b..33426f11 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -42,8 +42,8 @@ def update(status) # # A data source can deliver events from its own thread. A spec that starts a # data source and then reads `statuses` at once can run before the event -# arrives. Use `wait_for_count` or `wait_for_status` to block until the events -# you expect have arrived, then assert on the returned array. +# arrives. Use `wait_for_count` to block until the events you expect have +# arrived, then assert on the returned array. # class ListenerSpy def initialize @@ -88,16 +88,6 @@ def wait_for_count(count, timeout: 2) @statuses.dup end end - - # - # Blocks until at least one event has arrived, or until the timeout passes. - # - # @param timeout [Numeric] the maximum time to wait, in seconds - # @return [Array] - # - def wait_for_status(timeout: 2) - wait_for_count(1, timeout: timeout) - end end