Skip to content

fix: make FDv1 fallback TTL-based and never terminal #469

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

Draft
beekld wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix: make FDv1 fallback TTL-based and never terminal #469

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
110 changes: 87 additions & 23 deletions lib/ldclient-rb/impl/data_system/fdv2.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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?
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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?
Expand Down Expand Up @@ -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

Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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

Expand Down
63 changes: 55 additions & 8 deletions lib/ldclient-rb/impl/data_system/polling.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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?
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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)
Expand All @@ -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

Expand All @@ -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
Expand All @@ -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
Expand Down
23 changes: 18 additions & 5 deletions lib/ldclient-rb/impl/data_system/protocolv2.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

#
Expand All @@ -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

#
Expand All @@ -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

Expand Down
Loading
Loading