diff --git a/lib/ldclient-rb/impl/data_system/fdv2.rb b/lib/ldclient-rb/impl/data_system/fdv2.rb index 63ec1a90..7850abfb 100644 --- a/lib/ldclient-rb/impl/data_system/fdv2.rb +++ b/lib/ldclient-rb/impl/data_system/fdv2.rb @@ -34,7 +34,8 @@ module SyncResult REMOVE = :remove # Switch to the FDv1 protocol fallback. - # Replaces the synchronizer list with the FDv1 fallback synchronizer. + # FDv2 is stopped, but its synchronizers are retained so that it can be + # retried once the fallback TTL elapses. FDV1 = :fdv1 end @@ -115,6 +116,10 @@ def initialize(sdk_key, config, data_system_config) @active_synchronizer = nil @threads = [] + # TTL, in seconds, of the most recent FDv1 Fallback Directive. Only + # accessed from the main and synchronizer threads, which run in turn. + @fdv1_fallback_ttl = nil + # Track configuration @configured_with_data_sources = (@data_system_config.initializers && !@data_system_config.initializers.empty?) || !@synchronizer_builders.empty? @@ -228,10 +233,12 @@ def run_main_loop ) # Run initializers first. If an initializer signals the - # server-directed FDv1 Fallback Directive, switch terminally to - # the FDv1 Fallback Synchronizer (or transition to OFF if none - # is configured) before entering the synchronizer phase. - if run_initializers + # server-directed FDv1 Fallback Directive, the synchronizer phase + # starts in FDv1 fallback. If there are no FDv2 synchronizers to + # recover to, switch to the FDv1 Fallback Synchronizer (or + # transition to OFF if none is configured). + if run_initializers && @synchronizer_builders.empty? + @fdv1_fallback_ttl = nil if @fdv1_fallback_synchronizer_builder @logger.warn { "[LDClient] Falling back to FDv1 protocol" } @synchronizer_builders = [@fdv1_fallback_synchronizer_builder] @@ -278,6 +285,7 @@ def run_initializers fetch_result = initializer.fetch(@store) fallback = fetch_result.fallback_to_fdv1 + @fdv1_fallback_ttl = LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(fetch_result.fdv1_fallback_ttl) if fallback basis_result = fetch_result.result if basis_result.success? @@ -356,6 +364,16 @@ def synchronizer_loop begin while !@stop_event.set? && current_index < @synchronizer_builders.length + unless @fdv1_fallback_ttl.nil? + ttl = @fdv1_fallback_ttl + @fdv1_fallback_ttl = nil + break unless run_fdv1_fallback(ttl) + + @logger.info { "[LDClient] FDv1 fallback TTL elapsed; attempting to recover to FDv2" } + current_index = 0 + next + end + synchronizer_builder = @synchronizer_builders[current_index] is_primary = current_index == 0 @@ -376,22 +394,8 @@ def synchronizer_loop case sync_result when SyncResult::FDV1 - if @fdv1_fallback_synchronizer_builder - @logger.warn { "[LDClient] Falling back to FDv1 protocol" } - @synchronizer_builders = [@fdv1_fallback_synchronizer_builder] - current_index = 0 - next - end - # No FDv1 fallback configured: the data system must HALT - # rather than fall through to the next FDv2 synchronizer. - # Continuing to retry would reopen the connection that just - # delivered the directive. - @logger.warn { "[LDClient] Synchronizer requested FDv1 fallback but none configured; halting data system" } - @data_source_status_provider.update_status( - LaunchDarkly::Interfaces::DataSource::Status::OFF, - @data_source_status_provider.status.last_error - ) - break + # The fallback runs before the next FDv2 synchronizer is built. + next when SyncResult::RECOVER @logger.info { "[LDClient] Recovery condition met, returning to primary synchronizer" } current_index = 0 @@ -430,16 +434,71 @@ def synchronizer_loop end end + # + # Stays off FDv2 until the FDv1 fallback TTL elapses, running the FDv1 + # Fallback Synchronizer in the meantime if one is configured. FDv2's + # synchronizers are retained so they can be retried afterwards. + # + # @param ttl [Integer] seconds to stay off FDv2 + # @return [Boolean] true when the TTL elapsed; false if the data system was stopped + # + def run_fdv1_fallback(ttl) + deadline = Impl::Util.monotonic_seconds + ttl + + if @fdv1_fallback_synchronizer_builder + @logger.warn { "[LDClient] Falling back to FDv1 protocol; will retry FDv2 in #{ttl}s" } + + loop do + remaining = deadline - Impl::Util.monotonic_seconds + break if @stop_event.set? || remaining <= 0 + + @lock.synchronize do + sync = @fdv1_fallback_synchronizer_builder.build(@sdk_key, @config) + if sync.respond_to?(:set_diagnostic_accumulator) && @diagnostic_accumulator + sync.set_diagnostic_accumulator(@diagnostic_accumulator) + end + @active_synchronizer = sync + end + + sync_result = consume_synchronizer_results(@active_synchronizer, recover_after: remaining) + break if sync_result == SyncResult::RECOVER + + if sync_result == SyncResult::REMOVE + @logger.warn { "[LDClient] FDv1 fallback synchronizer failed; will retry FDv2 when the fallback TTL elapses" } + break + end + end + else + @logger.warn { "[LDClient] Synchronizer requested FDv1 fallback but none configured; will retry FDv2 in #{ttl}s" } + @data_source_status_provider.update_status( + LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, + @data_source_status_provider.status.last_error + ) + # Nothing will deliver data until FDv2 is retried, so do not make + # the application wait for it. + @ready_event.set + end + + # The FDv1 synchronizer cannot itself direct a fallback. + @fdv1_fallback_ttl = nil + + remaining = deadline - Impl::Util.monotonic_seconds + @stop_event.wait(remaining) if remaining > 0 + !@stop_event.set? + end + # # Consume results from a synchronizer until a condition is met or it fails. # # @param synchronizer [Object] The synchronizer # @param check_recovery [Boolean] Whether to check recovery condition (healthy too long) + # @param recover_after [Float, nil] If set, return {SyncResult::RECOVER} after this many seconds # @return [Symbol] One of {SyncResult::FALLBACK}, {SyncResult::RECOVER}, {SyncResult::REMOVE}, or {SyncResult::FDV1} # - def consume_synchronizer_results(synchronizer, check_recovery: false) + def consume_synchronizer_results(synchronizer, check_recovery: false, recover_after: nil) action_queue = Queue.new timer = LaunchDarkly::Impl::RepeatingTask.new(10, 10, -> { action_queue.push("check") }, @logger, "FDv2-sync-cond-timer") + recovery_task = (Concurrent::ScheduledTask.execute(recover_after) { action_queue.push("recover") } unless recover_after.nil?) # Start reader thread sync_reader = Thread.new do @@ -461,6 +520,7 @@ def consume_synchronizer_results(synchronizer, check_recovery: false) if update.is_a?(String) break if update == "quit" + return SyncResult::RECOVER if update == "recover" if update == "check" # Check condition periodically. One atomic read pairs the status with @@ -487,7 +547,10 @@ def consume_synchronizer_results(synchronizer, check_recovery: false) @ready_event.set if valid - return SyncResult::FDV1 if update.fallback_to_fdv1 + if update.fallback_to_fdv1 + @fdv1_fallback_ttl = LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(update.fdv1_fallback_ttl) + return SyncResult::FDV1 + end return SyncResult::REMOVE if update.state == LaunchDarkly::Interfaces::DataSource::Status::OFF end @@ -497,6 +560,7 @@ def consume_synchronizer_results(synchronizer, check_recovery: false) ensure synchronizer.stop timer.stop + recovery_task&.cancel sync_reader.join(0.5) if sync_reader.alive? end diff --git a/lib/ldclient-rb/impl/data_system/polling.rb b/lib/ldclient-rb/impl/data_system/polling.rb index 8c5d1266..527ce213 100644 --- a/lib/ldclient-rb/impl/data_system/polling.rb +++ b/lib/ldclient-rb/impl/data_system/polling.rb @@ -20,6 +20,42 @@ module DataSystem LD_ENVID_HEADER = "X-LD-EnvID" LD_FD_FALLBACK_HEADER = "X-LD-FD-Fallback" + LD_FD_FALLBACK_TTL_HEADER = "X-LD-FD-Fallback-TTL" + + # Used whenever the service supplies no usable FDv1 fallback TTL. + DEFAULT_FDV1_FALLBACK_TTL = 60 * 60 + + # + # Returns how long, in seconds, to stay on FDv1 after a fallback + # directive before attempting to recover to FDv2. A service-supplied TTL + # in (0, 1 hour] is used as-is. Anything else (missing, malformed, zero, + # negative, or too large) uses the 1 hour default, minus a uniformly + # random jitter of up to half of it, so that SDKs which fell back + # together do not all retry at once. + # + # @param ttl [Integer, String, nil] The TTL from the + # X-LD-FD-Fallback-TTL header or goodbye.protocolFallbackTTL + # @return [Integer] + # + def self.fdv1_fallback_ttl(ttl) + ttl = Integer(ttl, 10) if ttl.is_a?(String) && ttl.match?(/\A\d+\z/) + return ttl if ttl.is_a?(Integer) && ttl > 0 && ttl <= DEFAULT_FDV1_FALLBACK_TTL + + DEFAULT_FDV1_FALLBACK_TTL - Random.rand(0..DEFAULT_FDV1_FALLBACK_TTL / 2) + end + + # + # Returns the FDv1 fallback TTL carried on the response headers. See + # {.fdv1_fallback_ttl}. + # + # @param headers [#[], Hash, nil] + # @return [Integer] + # + def self.fdv1_fallback_ttl_from_headers(headers) + value = lookup_header(headers, LD_FD_FALLBACK_TTL_HEADER) + value = value.first if value.is_a?(Array) + fdv1_fallback_ttl(value) + end # # Reports whether the response headers signal that the SDK should fall @@ -128,6 +164,7 @@ def sync(ss) until @stop.set? result = @requester.fetch(ss.selector) fallback = LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(result.headers) + fallback_ttl = (LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers(result.headers) if fallback) envid = LaunchDarkly::Impl::DataSystem.lookup_header(result.headers, LD_ENVID_HEADER) if !result.success? @@ -149,7 +186,8 @@ def sync(ss) state: LaunchDarkly::Interfaces::DataSource::Status::OFF, error: error_info, environment_id: envid, - fallback_to_fdv1: true + fallback_to_fdv1: true, + fdv1_fallback_ttl: fallback_ttl ) break end @@ -168,7 +206,8 @@ def sync(ss) state: LaunchDarkly::Interfaces::DataSource::Status::OFF, error: error_info, environment_id: envid, - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) break end @@ -186,7 +225,8 @@ def sync(ss) state: LaunchDarkly::Interfaces::DataSource::Status::OFF, error: error_info, environment_id: envid, - fallback_to_fdv1: true + fallback_to_fdv1: true, + fdv1_fallback_ttl: fallback_ttl ) break end @@ -202,7 +242,8 @@ def sync(ss) state: LaunchDarkly::Interfaces::DataSource::Status::VALID, change_set: result.value, environment_id: envid, - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) end @@ -231,8 +272,10 @@ def stop # Default to false so the rescue clause has a defined value even when an # exception is raised before the header read below has had a chance to run. fallback = false + fallback_ttl = nil result = @requester.fetch(ss.selector) fallback = LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(result.headers) + fallback_ttl = LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers(result.headers) if fallback unless result.success? if result.exception.is_a?(LaunchDarkly::Impl::DataSource::UnexpectedResponseError) @@ -243,13 +286,15 @@ def stop @logger.warn { "[LDClient] #{http_error_message_result}" } if Impl::Util.http_error_recoverable?(status_code) return LaunchDarkly::Interfaces::DataSystem::FetchResult.new( result: LaunchDarkly::Result.fail(http_error_message_result, result.exception), - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) end return LaunchDarkly::Interfaces::DataSystem::FetchResult.new( result: LaunchDarkly::Result.fail(result.error || 'Failed to request payload', result.exception), - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) end @@ -265,7 +310,8 @@ def stop LaunchDarkly::Interfaces::DataSystem::FetchResult.new( result: LaunchDarkly::Result.success(basis), - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) rescue => e # An exception can fire after we have already read the response @@ -277,7 +323,8 @@ def stop @logger.debug { "[LDClient] Exception trace: #{e.backtrace}" } LaunchDarkly::Interfaces::DataSystem::FetchResult.new( result: LaunchDarkly::Result.fail(msg, e), - fallback_to_fdv1: fallback + fallback_to_fdv1: fallback, + fdv1_fallback_ttl: fallback_ttl ) end end diff --git a/lib/ldclient-rb/impl/data_system/protocolv2.rb b/lib/ldclient-rb/impl/data_system/protocolv2.rb index ef6b5ad3..b71bcad7 100644 --- a/lib/ldclient-rb/impl/data_system/protocolv2.rb +++ b/lib/ldclient-rb/impl/data_system/protocolv2.rb @@ -151,11 +151,24 @@ class Goodbye # @return [String] The reason for goodbye attr_reader :reason + # @return [Object, nil] The raw protocolFallbackTTL value, in seconds. When the + # field is present, the server is directing the SDK to fall back to FDv1. + attr_reader :protocol_fallback_ttl + # # @param reason [String] The reason for goodbye + # @param protocol_fallback_ttl [Object, nil] The raw protocolFallbackTTL value # - def initialize(reason:) + def initialize(reason:, protocol_fallback_ttl: nil) @reason = reason + @protocol_fallback_ttl = protocol_fallback_ttl + end + + # + # @return [Boolean] true if the server is directing the SDK to fall back to FDv1. + # + def fallback_to_fdv1? + !@protocol_fallback_ttl.nil? end # @@ -164,9 +177,9 @@ def initialize(reason:) # @return [Hash] # def to_h - { - reason: @reason, - } + h = { reason: @reason } + h[:protocolFallbackTTL] = @protocol_fallback_ttl unless @protocol_fallback_ttl.nil? + h end # @@ -181,7 +194,7 @@ def self.from_h(data) raise ArgumentError, "Missing required fields in Goodbye" if reason.nil? - new(reason: reason) + new(reason: reason, protocol_fallback_ttl: data[:protocolFallbackTTL]) end end diff --git a/lib/ldclient-rb/impl/data_system/streaming.rb b/lib/ldclient-rb/impl/data_system/streaming.rb index ae37bcae..0b4b2dcb 100644 --- a/lib/ldclient-rb/impl/data_system/streaming.rb +++ b/lib/ldclient-rb/impl/data_system/streaming.rb @@ -92,6 +92,7 @@ def sync(ss) # on_event, and on_error on the same SSE worker thread, so # reads and writes here are single-threaded by construction. fdv1_fallback_pending = false + fdv1_fallback_ttl = nil base_uri = @http_config.base_uri + FDV2_STREAMING_ENDPOINT headers = Impl::Util.default_http_headers(@sdk_key, @config) @@ -110,13 +111,17 @@ def sync(ss) # Per-environment identifier: server sends it on every connect, # but it never changes once known so only assign once. envid ||= LaunchDarkly::Impl::DataSystem.lookup_header(headers, LD_ENVID_HEADER) - fdv1_fallback_pending = true if LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(headers) + if LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(headers) + fdv1_fallback_pending = true + fdv1_fallback_ttl = LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers(headers) + end end end client.on_event do |event| begin - update = process_message(event, change_set_builder, envid, fdv1_fallback_pending: fdv1_fallback_pending) + update = process_message(event, change_set_builder, envid, + fdv1_fallback_pending: fdv1_fallback_pending, fdv1_fallback_ttl: fdv1_fallback_ttl) next unless update log_connection_result(true) @@ -124,11 +129,12 @@ def sync(ss) yield update - # When the FDv1 Fallback Directive rode along on a Valid update, close - # the stream so the primary synchronizer is stopped once the directive - # engages. process_message marks the Update with fallback_to_fdv1 only - # on payloads that complete a transfer, so the consumer has already - # applied the ChangeSet by the time we get here. + # When the FDv1 Fallback Directive rode along on a Valid update, or + # arrived in a goodbye event, close the stream so the primary + # synchronizer is stopped once the directive engages. process_message + # marks a Valid Update with fallback_to_fdv1 only on payloads that + # complete a transfer, so the consumer has already applied the + # ChangeSet by the time we get here. if update.fallback_to_fdv1 fdv1_fallback_pending = false stop @@ -169,13 +175,17 @@ def sync(ss) client.on_error do |error| log_connection_result(false) fallback = false + fallback_ttl = nil if error.respond_to?(:headers) && error.headers envid ||= LaunchDarkly::Impl::DataSystem.lookup_header(error.headers, LD_ENVID_HEADER) - fallback = true if LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(error.headers) + if LaunchDarkly::Impl::DataSystem.fdv1_fallback_requested?(error.headers) + fallback = true + fallback_ttl = LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers(error.headers) + end end - update = handle_error(error, envid, fallback) + update = handle_error(error, envid, fallback, fallback_ttl) yield update if update end @@ -218,9 +228,11 @@ def stop # or PAYLOAD_TRANSFERRED) is marked with fallback_to_fdv1: true so # the consumer can engage the FDv1 Fallback Synchronizer after # applying the in-flight ChangeSet. + # @param fdv1_fallback_ttl [Integer, nil] the TTL, in seconds, that + # accompanied the pending directive. # @return [LaunchDarkly::Interfaces::DataSystem::Update, nil] # - private def process_message(message, change_set_builder, envid, fdv1_fallback_pending: false) + private def process_message(message, change_set_builder, envid, fdv1_fallback_pending: false, fdv1_fallback_ttl: nil) event_type = message.type # Handle heartbeat @@ -240,7 +252,8 @@ def stop return LaunchDarkly::Interfaces::DataSystem::Update.new( state: LaunchDarkly::Interfaces::DataSource::Status::VALID, environment_id: envid, - fallback_to_fdv1: fdv1_fallback_pending + fallback_to_fdv1: fdv1_fallback_pending, + fdv1_fallback_ttl: fdv1_fallback_ttl ) end nil @@ -258,7 +271,14 @@ def stop when LaunchDarkly::Interfaces::DataSystem::EventName::GOODBYE goodbye = LaunchDarkly::Impl::DataSystem::ProtocolV2::Goodbye.from_h(JSON.parse(message.data, symbolize_names: true)) @logger.info { "[LDClient] SSE server received goodbye: #{goodbye.reason}" } - nil + return nil unless goodbye.fallback_to_fdv1? + + LaunchDarkly::Interfaces::DataSystem::Update.new( + state: LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, + environment_id: envid, + fallback_to_fdv1: true, + fdv1_fallback_ttl: LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(goodbye.protocol_fallback_ttl) + ) when LaunchDarkly::Interfaces::DataSystem::EventName::ERROR error = LaunchDarkly::Impl::DataSystem::ProtocolV2::Error.from_h(JSON.parse(message.data, symbolize_names: true)) @@ -276,7 +296,8 @@ def stop state: LaunchDarkly::Interfaces::DataSource::Status::VALID, change_set: change_set, environment_id: envid, - fallback_to_fdv1: fdv1_fallback_pending + fallback_to_fdv1: fdv1_fallback_pending, + fdv1_fallback_ttl: fdv1_fallback_ttl ) else @@ -291,9 +312,10 @@ def stop # @param error [Exception] # @param envid [String, nil] # @param fallback [Boolean] + # @param fallback_ttl [Integer, nil] # @return [LaunchDarkly::Interfaces::DataSystem::Update, nil] # - private def handle_error(error, envid, fallback) + private def handle_error(error, envid, fallback, fallback_ttl = nil) return nil if @stopped.set? update = nil @@ -312,6 +334,7 @@ def stop state: LaunchDarkly::Interfaces::DataSource::Status::OFF, error: error_info, fallback_to_fdv1: true, + fdv1_fallback_ttl: fallback_ttl, environment_id: envid ) stop diff --git a/lib/ldclient-rb/interfaces/data_system.rb b/lib/ldclient-rb/interfaces/data_system.rb index ef517a85..cd44c602 100644 --- a/lib/ldclient-rb/interfaces/data_system.rb +++ b/lib/ldclient-rb/interfaces/data_system.rb @@ -527,19 +527,25 @@ class Update # @return [String, nil] The environment ID attr_reader :environment_id + # @return [Integer, nil] How long, in seconds, to stay on FDv1 before attempting to + # recover to FDv2, when {#fallback_to_fdv1} is set. If nil, a default is used. + attr_reader :fdv1_fallback_ttl + # # @param state [Symbol] The data source state ({LaunchDarkly::Interfaces::DataSource::Status}) # @param change_set [ChangeSet, nil] The change set # @param error [LaunchDarkly::Interfaces::DataSource::ErrorInfo, nil] Error information # @param fallback_to_fdv1 [Boolean] Whether to fall back to FDv1 # @param environment_id [String, nil] The environment ID + # @param fdv1_fallback_ttl [Integer, nil] How long, in seconds, to stay on FDv1 # - def initialize(state:, change_set: nil, error: nil, fallback_to_fdv1: false, environment_id: nil) + def initialize(state:, change_set: nil, error: nil, fallback_to_fdv1: false, environment_id: nil, fdv1_fallback_ttl: nil) @state = state @change_set = change_set @error = error @fallback_to_fdv1 = fallback_to_fdv1 @environment_id = environment_id + @fdv1_fallback_ttl = fdv1_fallback_ttl end end @@ -560,13 +566,19 @@ class FetchResult # @return [Boolean] Whether the server has instructed the SDK to fall back to the FDv1 protocol. attr_reader :fallback_to_fdv1 + # @return [Integer, nil] How long, in seconds, to stay on FDv1 before attempting to + # recover to FDv2, when {#fallback_to_fdv1} is set. If nil, a default is used. + attr_reader :fdv1_fallback_ttl + # # @param result [LaunchDarkly::Result] A Result containing either a Basis or an error. # @param fallback_to_fdv1 [Boolean] Whether to fall back to FDv1. + # @param fdv1_fallback_ttl [Integer, nil] How long, in seconds, to stay on FDv1. # - def initialize(result:, fallback_to_fdv1: false) + def initialize(result:, fallback_to_fdv1: false, fdv1_fallback_ttl: nil) @result = result @fallback_to_fdv1 = fallback_to_fdv1 + @fdv1_fallback_ttl = fdv1_fallback_ttl end # @return [Boolean] true when the underlying Result was successful. diff --git a/spec/impl/data_system/fdv2_datasystem_spec.rb b/spec/impl/data_system/fdv2_datasystem_spec.rb index b9bc4dd0..9a7367f5 100644 --- a/spec/impl/data_system/fdv2_datasystem_spec.rb +++ b/spec/impl/data_system/fdv2_datasystem_spec.rb @@ -519,7 +519,7 @@ def with_data_system(initializers, synchronizers) end describe "stays on FDv1 after fallback" do - it "does not retry FDv2 after falling back to FDv1" do + it "does not retry FDv2 before the fallback TTL elapses" do mock_primary = double("primary_synchronizer") allow(mock_primary).to receive(:name).and_return("mock-primary") allow(mock_primary).to receive(:stop) @@ -548,7 +548,7 @@ def with_data_system(initializers, synchronizers) # Give it time to settle sleep 1.0 - # Primary should only be called once (not retried after fallback) + # Primary should only be called once (the default TTL is at least 30 minutes) expect(mock_primary).to have_received(:sync).once # Verify FDv1 is serving data @@ -675,6 +675,121 @@ def build(_sdk_key, _config) end end + describe "FDv1 fallback TTL" do + # Synchronizer that requests FDv1 fallback with a 1 second TTL on its + # first sync, and reports valid data afterwards. + class FallbackOnceSynchronizer + def initialize(events) + @events = events + @syncs = 0 + @recovered = Concurrent::Event.new + end + + attr_reader :recovered + + def name + "fallback-once" + end + + def sync(_selector_store) + @syncs += 1 + @stopped = Concurrent::Event.new + @events << :fdv2_sync + if @syncs == 1 + yield LaunchDarkly::Interfaces::DataSystem::Update.new( + state: LaunchDarkly::Interfaces::DataSource::Status::OFF, + fallback_to_fdv1: true, + fdv1_fallback_ttl: 1 + ) + return + end + + yield LaunchDarkly::Interfaces::DataSystem::Update.new( + state: LaunchDarkly::Interfaces::DataSource::Status::VALID + ) + @recovered.set + @stopped.wait + end + + def stop + @stopped&.set + end + end + + class RecordingFDv1Synchronizer + def initialize(events) + @events = events + @stopped = Concurrent::Event.new + end + + def name + "recording-fdv1" + end + + def sync(_selector_store) + @events << :fdv1_sync + yield LaunchDarkly::Interfaces::DataSystem::Update.new( + state: LaunchDarkly::Interfaces::DataSource::Status::VALID + ) + @stopped.wait + end + + def stop + @events << :fdv1_stop + @stopped.set + end + end + + it "recovers to FDv2 after the TTL when no FDv1 fallback synchronizer is configured" do + events = Queue.new + + primary = FallbackOnceSynchronizer.new(events) + + data_system_config = LaunchDarkly::DataSystem::ConfigBuilder.new + .initializers(nil) + .synchronizers([MockBuilder.new(primary)]) + .build + + valid_status = Concurrent::Event.new + listener = Object.new + listener.define_singleton_method(:update) do |status| + valid_status.set if status.state == LaunchDarkly::Interfaces::DataSource::Status::VALID + end + + fdv2 = FDv2.new(sdk_key, config, data_system_config) + fdv2.data_source_status_provider.add_listener(listener) + + ready_event = fdv2.start + expect(ready_event.wait(2)).to be true + expect(primary.recovered.wait(3)).to be true + expect(valid_status.wait(2)).to be true + + fdv2.stop + end + + it "stops the FDv1 fallback synchronizer before recovering to FDv2" do + events = Queue.new + + primary = FallbackOnceSynchronizer.new(events) + + fdv1 = RecordingFDv1Synchronizer.new(events) + + data_system_config = LaunchDarkly::DataSystem::ConfigBuilder.new + .initializers(nil) + .synchronizers([MockBuilder.new(primary)]) + .fdv1_compatible_synchronizer(MockBuilder.new(fdv1)) + .build + + fdv2 = FDv2.new(sdk_key, config, data_system_config) + + fdv2.start + expect(primary.recovered.wait(3)).to be true + fdv2.stop + + expect(Array.new(4) { events.pop }).to eq([:fdv2_sync, :fdv1_sync, :fdv1_stop, :fdv2_sync]) + end + end + describe "payload filter key" do let(:data_system_config) do LaunchDarkly::DataSystem::ConfigBuilder.new diff --git a/spec/impl/data_system/polling_initializer_spec.rb b/spec/impl/data_system/polling_initializer_spec.rb index a4fddfba..194979c7 100644 --- a/spec/impl/data_system/polling_initializer_spec.rb +++ b/spec/impl/data_system/polling_initializer_spec.rb @@ -55,6 +55,36 @@ def [](name) end end + RSpec.describe ".fdv1_fallback_ttl" do + [1, 60, 3600, "1", "120", "3600"].each do |ttl| + it "uses the valid TTL #{ttl.inspect} unchanged" do + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(ttl)).to eq(Integer(ttl)) + end + end + + [nil, 0, -1, 3601, 1.5, "", "0", "-5", "3601", "1.5", "abc"].each do |ttl| + it "uses a jittered 1 hour default for the invalid TTL #{ttl.inspect}" do + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(ttl)).to be_between(1800, 3600) + end + end + + it "jitters the default by subtracting up to 30 minutes" do + allow(Random).to receive(:rand).with(0..1800).and_return(0, 1800) + + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(nil)).to eq(3600) + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl(nil)).to eq(1800) + end + + it "reads the TTL from the X-LD-FD-Fallback-TTL header" do + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers({ 'X-LD-FD-Fallback-TTL' => '120' })).to eq(120) + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers({ 'x-ld-fd-fallback-ttl' => ['120'] })).to eq(120) + end + + it "uses the default when the X-LD-FD-Fallback-TTL header is absent" do + expect(LaunchDarkly::Impl::DataSystem.fdv1_fallback_ttl_from_headers({})).to be_between(1800, 3600) + end + end + RSpec.describe PollingDataSource do let(:logger) { double("Logger", info: nil, warn: nil, error: nil, debug: nil) } diff --git a/spec/impl/data_system/polling_synchronizer_spec.rb b/spec/impl/data_system/polling_synchronizer_spec.rb index b2538843..2936c181 100644 --- a/spec/impl/data_system/polling_synchronizer_spec.rb +++ b/spec/impl/data_system/polling_synchronizer_spec.rb @@ -518,6 +518,33 @@ def selector expect(off.environment_id).to eq('test-env-503') end + it "captures the fallback TTL header" do + headers_error = { + LD_FD_FALLBACK_HEADER => 'true', + LD_FD_FALLBACK_TTL_HEADER => '120', + } + failure = LaunchDarkly::Result.fail( + "error for test", + LaunchDarkly::Impl::DataSource::UnexpectedResponseError.new(503), + headers_error + ) + + synchronizer = PollingDataSource.new( + 0.01, + ListBasedRequester.new([failure]), + logger + ) + + updates = [] + synchronizer.sync(MockSelectorStore.new(LaunchDarkly::Interfaces::DataSystem::Selector.no_selector)) do |update| + updates << update + end + + expect(updates.length).to eq(1) + expect(updates[0].fallback_to_fdv1).to eq(true) + expect(updates[0].fdv1_fallback_ttl).to eq(120) + end + it "captures envid from generic error with headers" do builder = LaunchDarkly::Interfaces::DataSystem::ChangeSetBuilder.new builder.start(LaunchDarkly::Interfaces::DataSystem::IntentCode::TRANSFER_FULL) diff --git a/spec/impl/data_system/streaming_synchronizer_spec.rb b/spec/impl/data_system/streaming_synchronizer_spec.rb index 513c5cf7..3556dfe0 100644 --- a/spec/impl/data_system/streaming_synchronizer_spec.rb +++ b/spec/impl/data_system/streaming_synchronizer_spec.rb @@ -356,8 +356,8 @@ def initialize(type, data = nil) it "does not stamp fallback_to_fdv1 on intermediate (non-payload-completing) events" do # The directive is only meaningful once it can ride along on a Valid result that - # the consumer can act on. PUT_OBJECT, DELETE_OBJECT, GOODBYE, ERROR don't - # produce Updates -- intermediate events return nil even when the directive is + # the consumer can act on. PUT_OBJECT, DELETE_OBJECT, ERROR, and a GOODBYE without a + # TTL don't produce Updates -- intermediate events return nil even when the directive is # pending. The signal arrives with the next payload-completing event. put = LaunchDarkly::Impl::DataSystem::ProtocolV2::PutObject.new( version: 100, @@ -374,6 +374,51 @@ def initialize(type, data = nil) expect(update).to be_nil end + + it "carries the header TTL on the payload-completing update" do + server_intent = LaunchDarkly::Interfaces::DataSystem::ServerIntent.new( + payload: LaunchDarkly::Interfaces::DataSystem::Payload.new( + id: "id", + target: 300, + code: LaunchDarkly::Interfaces::DataSystem::IntentCode::TRANSFER_NONE, + reason: "up-to-date" + ) + ) + + update = synchronizer.send(:process_message, MockSSEEvent.new( + LaunchDarkly::Interfaces::DataSystem::EventName::SERVER_INTENT, + JSON.generate(server_intent.to_h) + ), change_set_builder, envid, fdv1_fallback_pending: true, fdv1_fallback_ttl: 120) + + expect(update.fallback_to_fdv1).to eq(true) + expect(update.fdv1_fallback_ttl).to eq(120) + end + + it "signals FDv1 fallback with the goodbye's protocolFallbackTTL" do + goodbye = LaunchDarkly::Impl::DataSystem::ProtocolV2::Goodbye.new( + reason: "fallback", + protocol_fallback_ttl: 120 + ) + + update = synchronizer.send(:process_message, MockSSEEvent.new( + LaunchDarkly::Interfaces::DataSystem::EventName::GOODBYE, + JSON.generate(goodbye.to_h) + ), change_set_builder, envid) + + expect(update.state).to eq(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED) + expect(update.fallback_to_fdv1).to eq(true) + expect(update.fdv1_fallback_ttl).to eq(120) + end + + it "uses the jittered default for an invalid goodbye protocolFallbackTTL" do + update = synchronizer.send(:process_message, MockSSEEvent.new( + LaunchDarkly::Interfaces::DataSystem::EventName::GOODBYE, + JSON.generate({ reason: "fallback", protocolFallbackTTL: 0 }) + ), change_set_builder, envid) + + expect(update.fallback_to_fdv1).to eq(true) + expect(update.fdv1_fallback_ttl).to be_between(1800, 3600) + end end describe '#handle_error' do