feat: Retry indefinitely after a data source failure instead of stopping permanently

contract-tests/service.rb

@@ -55,6 +55,8 @@
       'flag-change-listeners',
       'flag-value-change-listeners',
       'fdv1-fallback',
+      'retry-conformance-fdv1-streaming',
+      'retry-conformance-fdv1-polling',
     ],
   }.to_json
 end

launchdarkly-server-sdk.gemspec

@@ -42,13 +42,13 @@ Gem::Specification.new do |spec|
 
   spec.add_runtime_dependency "benchmark", "~> 0.1", ">= 0.1.1"
   spec.add_runtime_dependency "concurrent-ruby", "~> 1.1"
-  spec.add_runtime_dependency "ld-eventsource", "2.6.0"
+  # Keep this an exact version: the SDK depends on this library's behavior, so test
+  # each new version with the SDK before release.
+  spec.add_runtime_dependency "ld-eventsource", "3.0.1"
   spec.add_runtime_dependency "observer", "~> 0.1.2"
   spec.add_runtime_dependency "openssl", ">= 3.1.2", "< 5.0"
   spec.add_runtime_dependency "semantic", "~> 1.6"
   spec.add_runtime_dependency "zlib", "~> 3.1" unless RUBY_PLATFORM == "java"
-  # Please keep ld-eventsource dependency as an exact version so that bugfixes to
-  # that LD library are always associated with a new SDK version.
 
   spec.add_runtime_dependency "http", ">= 4.4.0", "< 7.0.0"
   spec.add_runtime_dependency "json", ">= 2.3", "< 4"

lib/ldclient-rb/impl/data_source.rb

@@ -138,6 +138,9 @@ def update_status(new_state, new_error)
           @mutex.synchronize do
             old_status = @current_status
 
+            # OFF is final: no status is reported after it.
+            return if old_status.state == LaunchDarkly::Interfaces::DataSource::Status::OFF
+
             if new_state == LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED && old_status.state == LaunchDarkly::Interfaces::DataSource::Status::INITIALIZING
               # See {LaunchDarkly::Interfaces::DataSource::UpdateSink#update_status} for more information
               new_state = LaunchDarkly::Interfaces::DataSource::Status::INITIALIZING

lib/ldclient-rb/impl/data_source/polling.rb

@@ -1,5 +1,6 @@
 require "ldclient-rb/impl/data_source"
 require "ldclient-rb/impl/repeating_task"
+require "ldclient-rb/impl/retry_state"
 require "ldclient-rb/impl/util"
 
 require "concurrent/atomics"
@@ -16,7 +17,8 @@ def initialize(config, requestor)
           @initialized = Concurrent::AtomicBoolean.new(false)
           @started = Concurrent::AtomicBoolean.new(false)
           @ready = Concurrent::Event.new
-          @task = Impl::RepeatingTask.new(@config.poll_interval, 0, -> { self.poll }, @config.logger, 'LD/PollingDataSource')
+          @retry_state = Impl::RetryState.for_polling(@config.poll_interval, @config.logger)
+          @task = Impl::RepeatingTask.new(-> { @retry_state.next_delay }, 0, -> { self.poll }, @config.logger, 'LD/PollingDataSource')
         end
 
         def initialized?
@@ -43,14 +45,16 @@ def poll
               update_sink_or_data_store.init(all_data)
               newly_initialized = @initialized.make_true
             end
+            @retry_state.record_success
             @config.data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::VALID, nil)
 
             if newly_initialized
               @config.logger.info { "[LDClient] Polling connection initialized" }
               @ready.set
             end
           rescue JSON::ParserError => e
-            @config.logger.error { "[LDClient] JSON parsing failed for polling response." }
+            @retry_state.record_failure(:normal)
+            @config.logger.error { "[LDClient] JSON parsing failed for polling response - #{Util.retry_message(@retry_state.next_delay)}" }
             error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(
               LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA,
               0,
@@ -59,25 +63,16 @@ def poll
             )
             @config.data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, error_info)
           rescue Impl::DataSource::UnexpectedResponseError => e
+            kind = Util.classify_http_status(e.status)
+            @retry_state.record_failure(kind)
             error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(
               LaunchDarkly::Interfaces::DataSource::ErrorInfo::ERROR_RESPONSE, e.status, nil, Time.now)
-            message = Util.http_error_message(e.status, "polling request", "will retry")
-            @config.logger.error { "[LDClient] #{message}" }
-
-            if Util.http_error_recoverable?(e.status)
-              @config.data_source_update_sink&.update_status(
-                LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED,
-                error_info
-              )
-            else
-              # Publish the OFF 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 failure.
-              stop_with_error_info error_info
-              @ready.set  # if client was waiting on us, make it stop waiting - has no effect if already set
-            end
+            message = Util.http_error_retry_message(e.status, "polling request", @retry_state.next_delay)
+            @config.logger.public_send(Util.failure_log_level(kind)) { "[LDClient] #{message}" }
+            @config.data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, error_info)
           rescue StandardError => e
-            Impl::Util.log_exception(@config.logger, "Exception while polling", e)
+            @retry_state.record_failure(:normal)
+            Impl::Util.log_exception(@config.logger, "Exception while polling - #{Util.retry_message(@retry_state.next_delay)}", e)
             @config.data_source_update_sink&.update_status(
               LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED,
               LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(LaunchDarkly::Interfaces::DataSource::ErrorInfo::UNKNOWN, 0, e.to_s, Time.now)

lib/ldclient-rb/impl/data_source/stream.rb

@@ -1,5 +1,6 @@
 require "ldclient-rb/impl/data_source"
 require "ldclient-rb/impl/model/serialization"
+require "ldclient-rb/impl/retry_state"
 require "ldclient-rb/impl/util"
 require "ldclient-rb/in_memory_store"
 
@@ -31,6 +32,8 @@ def initialize(sdk_key, config, diagnostic_accumulator = nil)
           @started = Concurrent::AtomicBoolean.new(false)
           @stopped = Concurrent::AtomicBoolean.new(false)
           @ready = Concurrent::Event.new
+          @stop_event = Concurrent::Event.new
+          @retry_state = Impl::RetryState.for_streaming(@config.initial_reconnect_delay, @config.logger)
           @connection_attempt_start_time = 0
           @connection_attempt_started_monotonic = nil
         end
@@ -51,47 +54,19 @@ def start
             connect_timeout: @config.connect_timeout,
             logger: @config.logger,
             socket_factory: @config.socket_factory,
-            reconnect_time: @config.initial_reconnect_delay,
+            # The SDK waits in its error handler before each reconnect, so the SSE client adds no delay.
+            reconnect_time: 0,
           }
           log_connection_started
 
           uri = Impl::Util.add_payload_filter_key(@config.stream_uri + "/all", @config)
           @es = SSE::Client.new(uri, **opts) do |conn|
             conn.on_connect { |response_headers| DataSource.record_environment_id(@data_source_update_sink, response_headers) }
             conn.on_event { |event| process_message(event) }
-            conn.on_error { |err|
-              log_connection_result(false)
-              case err
-              when SSE::Errors::HTTPStatusError
-                status = err.status
-                error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(
-                  LaunchDarkly::Interfaces::DataSource::ErrorInfo::ERROR_RESPONSE, status, nil, Time.now)
-                message = Util.http_error_message(status, "streaming connection", "will retry")
-                @config.logger.error { "[LDClient] #{message}" }
-
-                if Util.http_error_recoverable?(status)
-                  @data_source_update_sink&.update_status(
-                    LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED,
-                    error_info
-                  )
-                else
-                  @ready.set  # if client was waiting on us, make it stop waiting - has no effect if already set
-                  stop_with_error_info error_info
-                end
-              when SSE::Errors::HTTPContentTypeError, SSE::Errors::HTTPProxyError, SSE::Errors::ReadTimeoutError
-                @data_source_update_sink&.update_status(
-                  LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED,
-                  LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(LaunchDarkly::Interfaces::DataSource::ErrorInfo::NETWORK_ERROR, 0, err.to_s, Time.now)
-                )
-
-              else
-                @data_source_update_sink&.update_status(
-                  LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED,
-                  LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(LaunchDarkly::Interfaces::DataSource::ErrorInfo::UNKNOWN, 0, err.to_s, Time.now)
-                )
-              end
-            }
+            conn.on_error { |err| handle_error(err) }
           end
+          # stop can run before @es is set, and then it has no client to close.
+          @es.close if @stopped.value
 
           @ready
         end
@@ -106,10 +81,73 @@ def stop
         # @param [LaunchDarkly::Interfaces::DataSource::ErrorInfo, nil] error_info
         #
         def stop_with_error_info(error_info = nil)
-          if @stopped.make_true
-            @es.close
-            @data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::OFF, error_info)
-            @config.logger.info { "[LDClient] Stream connection stopped" }
+          return unless @stopped.make_true
+
+          @stop_event.set
+          begin
+            @es&.close
+          rescue StandardError => e
+            Impl::Util.log_exception(@config.logger, "Error while closing stream connection", e)
+          end
+          @data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::OFF, error_info)
+          @config.logger.info { "[LDClient] Stream connection stopped" }
+        end
+
+        def handle_error(err)
+          status = 0
+          message = err.to_s
+          kind = :normal
+          level = :warn
+
+          case err
+          when SSE::Errors::HTTPStatusError
+            status = err.status
+            message = nil
+            kind = Util.classify_http_status(status)
+            level = Util.failure_log_level(kind)
+            error_kind = LaunchDarkly::Interfaces::DataSource::ErrorInfo::ERROR_RESPONSE
+            description = Util.http_error_description(status, "streaming connection")
+          when JSON::ParserError
+            level = :error
+            error_kind = LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA
+            description = "Invalid data on streaming connection: #{err}"
+          when SSE::Errors::StreamClosedByServerError
+            error_kind = LaunchDarkly::Interfaces::DataSource::ErrorInfo::NETWORK_ERROR
+            description = "The server closed the streaming connection"
+          when SSE::Errors::HTTPContentTypeError, SSE::Errors::HTTPProxyError, SSE::Errors::ReadTimeoutError
+            error_kind = LaunchDarkly::Interfaces::DataSource::ErrorInfo::NETWORK_ERROR
+            description = "Error on streaming connection: #{err}"
+          else
+            error_kind = LaunchDarkly::Interfaces::DataSource::ErrorInfo::UNKNOWN
+            description = "Error on streaming connection: #{err}"
+          end
+
+          error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(error_kind, status, message, Time.now)
+          handle_failure(kind, error_info, level, description)
+        end
+
+        #
+        # Records a failure, logs and reports it, and waits before the next attempt. The SSE client calls this on
+        # its worker thread before it reconnects, so the wait is the reconnect delay, and {#stop} ends it at once.
+        #
+        # @param kind [Symbol] `:normal` or `:unexpected`
+        # @param error_info [LaunchDarkly::Interfaces::DataSource::ErrorInfo]
+        # @param level [Symbol] `:error` or `:warn`
+        # @param description [String] what failed, for the log message
+        #
+        def handle_failure(kind, error_info, level, description)
+          return if @stopped.value
+
+          @retry_state.record_failure(kind)
+          delay = @retry_state.next_delay
+          begin
+            log_connection_result(false)
+            @config.logger.public_send(level) { "[LDClient] #{description} - #{Util.retry_message(delay)}" }
+            @data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, error_info)
+          ensure
+            # The SSE client reconnects as soon as this returns, also when it returns by raising.
+            @stop_event.wait([delay, Util::TIMEOUT_MAX].min)
+            log_connection_started
           end
         end
 
@@ -131,6 +169,7 @@ def update_sink_or_data_store
         def process_message(message)
           log_connection_result(true)
           method = message.type
+          known_type = [PUT, PATCH, DELETE].include?(method)
           @config.logger.debug { "[LDClient] Stream received #{method} message: #{message.data}" }
 
           begin
@@ -164,16 +203,11 @@ def process_message(message)
               @config.logger.warn { "[LDClient] Unknown message received: #{method}" }
             end
 
+            # Only a data payload shows that the stream is healthy.
+            @retry_state.record_success if known_type
             @data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::VALID, nil)
-          rescue JSON::ParserError => e
+          rescue JSON::ParserError
             @config.logger.error { "[LDClient] JSON parsing failed for method #{method}. Ignoring event." }
-            error_info = LaunchDarkly::Interfaces::DataSource::ErrorInfo.new(
-              LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA,
-              0,
-              e.to_s,
-              Time.now
-            )
-            @data_source_update_sink&.update_status(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, error_info)
 
             # Re-raise the exception so the SSE implementation can catch it and restart the stream.
             raise

lib/ldclient-rb/impl/data_system/streaming.rb

@@ -334,7 +334,8 @@ def stop
 
             @logger.warn { "[LDClient] #{error_info.message}" }
 
-          when SSE::Errors::HTTPContentTypeError, SSE::Errors::HTTPProxyError, SSE::Errors::ReadTimeoutError
+          when SSE::Errors::HTTPContentTypeError, SSE::Errors::HTTPProxyError, SSE::Errors::ReadTimeoutError,
+               SSE::Errors::StreamClosedByServerError
             @logger.warn { "[LDClient] Network error on stream connection: #{error}, will retry" }
 
             update = LaunchDarkly::Interfaces::DataSystem::Update.new(

lib/ldclient-rb/impl/repeating_task.rb

@@ -5,19 +5,26 @@
 module LaunchDarkly
   module Impl
     #
-    # Runs a task again and again on a worker thread, with a fixed interval
-    # between runs.
+    # Runs a task again and again on a worker thread.
+    #
+    # The interval is read after each run, and the wait starts when the run
+    # returns.
     #
     # The worker waits on an event instead of calling `sleep`, so `stop` can
     # wake it at once even if `stop` runs before the worker starts waiting.
     #
     # @private
     #
     class RepeatingTask
+      # The wait, in seconds, when a callable interval raises or returns a value that is not a usable
+      # number. It must not be 0: a broken callable would then run the task back to back, which for
+      # polling is a tight loop of HTTP requests.
+      FALLBACK_INTERVAL = 1
+
       attr_reader :name
 
       #
-      # @param interval [Numeric] seconds between the start of one run and the start of the next
+      # @param interval [Numeric, #call] seconds between runs, or an object that returns them
       # @param start_delay [Numeric, nil] seconds to wait before the first run
       # @param task [Proc] the code to run
       # @param logger [Logger]
@@ -39,22 +46,32 @@ def start
           @stop_event.wait(@start_delay) unless @start_delay.nil? || @start_delay == 0
 
           until @stopped.value do
-            started_at = Impl::Util.monotonic_seconds
             begin
               @task.call
             rescue => e
               Impl::Util.log_exception(@logger, "Uncaught exception from repeating task", e)
             end
-            # The run must be measured on the monotonic clock: a wall-clock step would
-            # make this wait negative (immediate re-run) or stretch it by the step size.
-            delta = @interval - (Impl::Util.monotonic_seconds - started_at)
-            @stop_event.wait(delta) if delta > 0
+            delay = next_delay
+            @stop_event.wait([delay, Util::TIMEOUT_MAX].min) if delay > 0
           end
         end
 
         @worker.name = @name
       end
 
+      private def next_delay
+        return @interval if @interval.is_a?(Numeric)
+
+        delay = @interval.call
+        return delay if delay.is_a?(Numeric) && delay >= 0
+
+        @logger.error { "[LDClient] Repeating task #{@name} got an invalid interval #{delay.inspect}; waiting #{FALLBACK_INTERVAL}s" }
+        FALLBACK_INTERVAL
+      rescue => e
+        Impl::Util.log_exception(@logger, "Exception while reading the interval of repeating task #{@name}", e)
+        FALLBACK_INTERVAL
+      end
+
       #
       # Stops the worker thread and waits for it to finish.
       #