Fix WebSocket synchronization and reliability

README.md

@@ -223,13 +223,12 @@ client.flush
 
 Call `track` after evaluating the related experiment flag. `numeric_value` defaults to `1.0`.
 
-### Thread safety and shutdown
+### Thread safety
 
 - Share one client across application threads; do not create a client per request.
 - Flag evaluation performs no network I/O.
 - Event enqueue is non-blocking and bounded; overload increments `dropped_events`.
 - Status and flag callbacks execute outside internal locks.
-- `close` is thread-safe and idempotent, flushes accepted events, stops background workers, and prevents new events.
 - Public client methods contain ordinary internal failures and return fallbacks or `false`.
 
 ## Supported Ruby versions

lib/featbit.rb

@@ -8,7 +8,7 @@
 require_relative "featbit/data_store"
 require_relative "featbit/evaluator"
 require_relative "featbit/event_processor"
-require_relative "featbit/web_socket_data_synchronizer"
+require_relative "featbit/data_sync/web_socket_data_synchronizer"
 require_relative "featbit/client"
 
 module FeatBit

lib/featbit/client.rb

@@ -123,8 +123,7 @@ def remove_flag_change_listener(id)
 
     def close
       @lifecycle_mutex.synchronize do
-        return @close_result unless @close_result.nil?
-        return true if @closed
+        return @close_result == true if @closed
 
         @closed = true
       end

lib/featbit/data_sync/closable_web_socket_client.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+require "timeout"
+require "websocket-client-simple"
+
+module FeatBit
+  # websocket-client-simple has no abort API. Keep its internal cleanup here,
+  # after the connector has exited and SDK callbacks have drained.
+  class ClosableWebSocketClient < WebSocket::Client::Simple::Client
+    CLOSE_TIMEOUT = 1.0
+
+    def close(drain: false)
+      # The library also calls close from its own I/O error callbacks. Only the
+      # SDK owner may drain/kill the reader, after application callbacks return.
+      unless drain
+        @closed = true
+        return true
+      end
+
+      begin
+        Timeout.timeout(CLOSE_TIMEOUT) { super() }
+      rescue IOError, SystemCallError, OpenSSL::SSL::SSLError, Timeout::Error
+        # A failed close handshake is harmless if the ensure cleanup succeeds.
+      ensure
+        @closed = true
+        begin
+          @socket&.close
+          @socket = nil
+        ensure
+          @thread&.kill
+          @thread&.join unless Thread.current.equal?(@thread)
+        end
+      end
+      true
+    end
+  end
+end

lib/featbit/data_sync/synchronization_result.rb

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module FeatBit
+  class SynchronizationResult
+    attr_reader :valid, :changed
+
+    def initialize(valid:, changed:)
+      @valid = valid
+      @changed = changed
+      freeze
+    end
+
+    def valid? = valid
+    def changed? = changed
+
+    INVALID = new(valid: false, changed: false)
+    UNCHANGED = new(valid: true, changed: false)
+    CHANGED = new(valid: true, changed: true)
+
+    def self.valid(changed:)
+      changed ? CHANGED : UNCHANGED
+    end
+  end
+end

lib/featbit/data_sync/web_socket_close_policy.rb

@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+require_relative "../status"
+
+module FeatBit
+  class WebSocketClosePolicy
+    SERVER_REJECTED_CODE = 4003
+
+    def initialize(status_provider)
+      @status_provider = status_provider
+      @mutex = Mutex.new
+      @rejected = false
+    end
+
+    def close_frame?(event)
+      event.respond_to?(:type) && event.type.to_sym == :close
+    rescue StandardError
+      false
+    end
+
+    def reject?(event)
+      return false unless close_code(event) == SERVER_REJECTED_CODE
+
+      @mutex.synchronize { @rejected = true }
+      reason = close_reason(event)
+      message = "WebSocket connection rejected by server (4003)"
+      message = "#{message}: #{reason}" unless reason.empty?
+      @status_provider.update(Status::FAILED, message: message)
+      true
+    end
+
+    def rejected?
+      @mutex.synchronize { @rejected }
+    end
+
+    private
+
+    def close_code(event)
+      return event.code.to_i if event.respond_to?(:code) && event.code
+      return event[:code].to_i if event.is_a?(Hash) && event[:code]
+
+      event["code"].to_i if event.is_a?(Hash) && event["code"]
+    end
+
+    def close_reason(event)
+      value = if event.respond_to?(:reason) && event.reason
+                event.reason
+              elsif event.respond_to?(:data) && event.data
+                event.data
+              end
+      value.to_s
+    end
+  end
+end

lib/featbit/data_sync/web_socket_connection_attempt.rb

@@ -0,0 +1,95 @@
+# frozen_string_literal: true
+
+require "timeout"
+
+module FeatBit
+  class WebSocketConnectionAttempt
+    attr_reader :socket
+
+    def initialize(timeout:, clock:, lifecycle:, closer:)
+      @deadline = clock.call + timeout
+      @clock = clock
+      @mutex = Mutex.new
+      @condition = ConditionVariable.new
+      @result = nil
+      @lifecycle = lifecycle
+      @closer = closer
+      @finished = false
+    end
+
+    def connect(connector, url, headers)
+      @connector_thread = Thread.new do
+        configured = false
+        @socket = connector.call(url, headers) do |connected_socket|
+          @socket = connected_socket
+          yield connected_socket
+          configured = true
+        end
+        yield @socket unless configured
+      rescue StandardError => e
+        @connect_error = e
+      end
+      until @connector_thread.join(0.01)
+        break if @lifecycle.stopped? || @clock.call >= @deadline
+      end
+      return @socket if @lifecycle.stopped?
+
+      raise Timeout::Error, "WebSocket connection timed out" if @connector_thread.alive?
+      raise @connect_error if @connect_error
+
+      @socket
+    end
+
+    def cleanup
+      # Cancellation is confined to the owned connector, never an application
+      # callback. Joining before close prevents a late connector creating a socket.
+      @connector_thread&.kill if @connector_thread&.alive?
+      @connector_thread&.join
+      @closer.call(@socket)
+    end
+
+    def signal(result)
+      @mutex.synchronize do
+        @finished = true unless result == :opened
+        return if @result
+
+        @result = result
+        @condition.broadcast
+      end
+    end
+
+    def wait(stopped:)
+      @mutex.synchronize do
+        until @result || stopped.call
+          remaining = @deadline - @clock.call
+          return :timeout unless remaining.positive?
+
+          @condition.wait(@mutex, [remaining, 0.05].min)
+        end
+        stopped.call ? :stopped : @result
+      end
+    end
+
+    def monitor(stopped:, ping:, interval:)
+      next_ping = @clock.call + interval
+      until stopped.call || @mutex.synchronize { @finished } || socket_closed?
+        if @clock.call >= next_ping
+          ping.call(@socket)
+          next_ping = @clock.call + interval
+        end
+        sleep(0.05)
+      end
+    end
+
+    private
+
+    def socket_closed?
+      return @socket.closed? if @socket.respond_to?(:closed?)
+      return !@socket.open? if @socket.respond_to?(:open?)
+
+      false
+    rescue StandardError
+      true
+    end
+  end
+end