diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index c146199..b0176d8 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -11,8 +11,8 @@ permissions: jobs: sdk-spec: # Keep the ref in `uses:` and `spec-ref` in sync. - uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.1 + uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.9.0 with: - spec-ref: v0.7.1 - tags: core or events or local-evaluation + spec-ref: v0.9.0 + tags: core events local-evaluation language: ruby diff --git a/lib/flagsmith.rb b/lib/flagsmith.rb index 59bf142..6f8104d 100644 --- a/lib/flagsmith.rb +++ b/lib/flagsmith.rb @@ -50,6 +50,10 @@ class Client # rubocop:disable Metrics/ClassLength # # flag = flagsmith.get_experiment_flag('foo', 'identifier', plan: 'premium') # flagsmith.track_event('purchase', identifier: 'identifier', value: 42) + # + # Close the client before the application stops, so that its buffered events are sent: + # + # flagsmith.close # Available Configs. # @@ -221,8 +225,7 @@ def get_value_for_identity(feature_name, user_id = nil, default: nil) # Get a flag for an identity, taking the same traits as get_identity_flags, and record the # identity's exposure to the variant it was served when it is in the feature's running experiment. - # Requires enable_events. Exposure needs remote evaluation: the environment document used by local - # evaluation and offline handlers carries no experiment data, so no exposure is recorded from it. + # Requires enable_events. def get_experiment_flag(feature_name, identifier, **traits) require_events! flag = get_identity_flags(identifier, **traits)[feature_name] @@ -260,6 +263,21 @@ def track_exposure_event(feature_name, identifier:, value: nil, traits: nil, met metadata: metadata) end + # Send the events tracked so far, and return once each of them has been delivered, dropped, or kept for the + # next flush, retries included. Requires enable_events. + def flush_events + require_events! + event_processor.flush + end + + # Stop the client's background work, and send the buffered events, waiting for at most +timeout_seconds+. + # Events not sent by then are lost. + def close(timeout_seconds: Flagsmith::EventProcessor::CLOSE_TIMEOUT_SECONDS) + @environment_data_polling_manager&.stop + @realtime_client&.running = false + @event_processor&.stop(timeout_seconds) + end + def get_identity_segments(identifier, traits = {}) raise Flagsmith::ClientError, 'Local evaluation or offline handler is required to obtain identity segments.' unless environment @@ -283,7 +301,6 @@ def require_events! def exposure_skip_reason(flag) return 'feature not found' unless flag.is_a?(Flagsmith::Flags::Flag) return 'feature is disabled' unless flag.enabled? - return 'flags come from the environment document, which carries no experiment data' if environment return 'identity is not in a running experiment' unless flag.experiment&.in_experiment? 'flag has no variant' if flag.variant.nil? @@ -300,6 +317,7 @@ def environment_flags_from_document # rubocop:disable Metrics/MethodLength Flagsmith::Flags::Collection.from_evaluation_result( evaluation_result, + identity: false, analytics_processor: analytics_processor, default_flag_handler: default_flag_handler, offline_handler: offline_handler diff --git a/lib/flagsmith/engine/features/models.rb b/lib/flagsmith/engine/features/models.rb index de7ae0f..2a84bd6 100644 --- a/lib/flagsmith/engine/features/models.rb +++ b/lib/flagsmith/engine/features/models.rb @@ -8,7 +8,7 @@ module Engine class FeatureState include Flagsmith::Engine::Utils::HashFunc - attr_reader :feature, :enabled, :django_id, :uuid, :feature_segment + attr_reader :feature, :enabled, :django_id, :uuid, :feature_segment, :experiment attr_accessor :multivariate_feature_state_values def initialize(params = {}) @@ -19,6 +19,7 @@ def initialize(params = {}) @feature_state_value = params.fetch(:feature_state_value, nil) @uuid = params.fetch(:uuid, SecureRandom.uuid) @multivariate_feature_state_values = params.fetch(:multivariate_feature_state_values, []) + @experiment = params.fetch(:experiment, nil) end attr_writer :feature_state_value @@ -66,14 +67,12 @@ def higher_segment_priority?(other) class << self def build(json) multivariate_feature_state_values = build_multivariate_values(json[:multivariate_feature_state_values]) - attributes = json.slice(:uuid, :enabled, :django_id, :feature_state_value) - .merge(feature: Flagsmith::Engine::Feature.build(json[:feature])) - .merge(multivariate_feature_state_values: multivariate_feature_state_values) - if json.key?(:feature_segment) && !json[:feature_segment].nil? - attributes = attributes.merge( - feature_segment: Flagsmith::Engine::Features::Segment.new(json[:feature_segment]) - ) - end + attributes = json.slice(:uuid, :enabled, :django_id, :feature_state_value).merge( + feature: Flagsmith::Engine::Feature.build(json[:feature]), + multivariate_feature_state_values: multivariate_feature_state_values, + experiment: json.dig(:metadata, :experiment) + ) + attributes[:feature_segment] = Flagsmith::Engine::Features::Segment.new(json[:feature_segment]) unless json[:feature_segment].nil? new(**attributes) end @@ -113,16 +112,17 @@ def build(json) module Features # MultivariateFeatureOptionModel class MultivariateOption - attr_reader :value, :id + attr_reader :value, :id, :key - def initialize(value:, id: nil) + def initialize(value:, id: nil, key: nil) @value = value @id = id + @key = key end class << self def build(json) - new(**json.slice(:id, :value)) + new(**json.slice(:id, :value, :key)) end end end diff --git a/lib/flagsmith/engine/mappers/environment.rb b/lib/flagsmith/engine/mappers/environment.rb index 4931ace..f5e5920 100644 --- a/lib/flagsmith/engine/mappers/environment.rb +++ b/lib/flagsmith/engine/mappers/environment.rb @@ -26,22 +26,30 @@ def self.build_feature_hash(feature_state) name: feature_state.feature.name, enabled: feature_state.enabled, value: feature_state.get_value, - metadata: { id: feature_state.feature.id } + metadata: build_feature_metadata(feature_state) } add_variants_to_feature(feature_hash, feature_state) add_priority_to_feature(feature_hash, feature_state) feature_hash end + # The running experiment a feature state belongs to, if any, is passed through to its flag + def self.build_feature_metadata(feature_state) + metadata = { id: feature_state.feature.id } + metadata[:experiment] = feature_state.experiment if feature_state.experiment + metadata + end + def self.add_variants_to_feature(feature_hash, feature_state) return unless feature_state.multivariate_feature_state_values&.any? feature_hash[:variants] = feature_state.multivariate_feature_state_values.map do |mv| { value: mv.multivariate_feature_option.value, + key: mv.multivariate_feature_option.key, weight: mv.percentage_allocation, priority: mv.id || uuid_to_big_int(mv.mv_fs_value_uuid) - } + }.compact end end diff --git a/lib/flagsmith/sdk/event_processor.rb b/lib/flagsmith/sdk/event_processor.rb index 9958ee9..176a0cb 100644 --- a/lib/flagsmith/sdk/event_processor.rb +++ b/lib/flagsmith/sdk/event_processor.rb @@ -26,26 +26,36 @@ def self.to_a # # Events are sent every +flush_interval_seconds+, as soon as +max_buffer_items+ are buffered, # or on #flush. Sending events never raises into the application: failures are logged. + # + # A batch is sent up to +MAX_ATTEMPTS+ times, one batch at a time, backing off with full jitter between + # attempts. A batch whose last attempt fails transiently is kept for the next flush, unless that attempt got + # a 500. A batch the events API rejects is dropped, and an events API refusing the environment key stops + # event tracking. When the buffer is full, the oldest event is dropped. While a failed batch is kept, a full + # buffer waits for the next flush rather than starting another round of attempts. + # + # Once stopped, tracking and flushing do nothing. class EventProcessor # rubocop:disable Metrics/ClassLength - include Flagsmith::SDK::Intervals - FLAG_EXPOSURE_EVENT = '$flag_exposure' ENDPOINT = 'v1/events' FLUSH_INTERVAL_SECONDS = 10 MAX_BUFFER_ITEMS = 1000 REQUEST_TIMEOUT_SECONDS = 3 - RETRIABLE_STATUSES = [500, 502, 503, 504].freeze + MAX_ATTEMPTS = 3 + RETRY_BASE_SECONDS = 1 + CLOSE_TIMEOUT_SECONDS = 5 + # Retried along with every server error + RETRIED_STATUSES = [408, 429].freeze + REFUSED_STATUSES = [401, 403].freeze - def initialize(config, flush_interval_seconds: FLUSH_INTERVAL_SECONDS, max_buffer_items: MAX_BUFFER_ITEMS) + def initialize(config, flush_interval_seconds: FLUSH_INTERVAL_SECONDS, max_buffer_items: MAX_BUFFER_ITEMS, + retry_base_seconds: RETRY_BASE_SECONDS) @logger = config.logger @flush_interval_seconds = flush_interval_seconds @max_buffer_items = max_buffer_items + @retry_base_seconds = retry_base_seconds @connection = build_connection(config) - @events = [] - @buffered_exposures = Set.new - @mutex = Mutex.new - @flush_mutex = Mutex.new - @flush_pending = false + initialize_buffer + initialize_state end def track_event(event, identifier: nil, value: nil, traits: nil, metadata: nil) @@ -58,51 +68,203 @@ def track_exposure_event(feature_name, identifier:, value: nil, traits: nil, met identifier: identifier, value: value, traits: traits, metadata: metadata)) end + # Sends the events tracked so far, and returns once each of them has been delivered, dropped, or kept for + # the next flush, retries included. Events another flush sent while this one waited are not sent again. def flush - drain + request = @mutex.synchronize { flush_request unless stopping?('flushing events') } + send_batch(request) if request end def start - return if @interval + return if @interval || @stopping - @interval = set_interval(@flush_interval_seconds) { flush } + @interval = Thread.new { send_batch until wait(@flush_interval_seconds) { @stopping } } RunningEventProcessors.add(self) end - def stop - @flush_mutex.synchronize { clear_interval(@interval) } if @interval # never interrupts a batch being sent - @interval = nil + # Stops flushing every interval, and sends the buffered events, waiting for at most +timeout_seconds+. + # Events not sent by then are not retried, and no batch is taken to send once it returns. + def stop(timeout_seconds = CLOSE_TIMEOUT_SECONDS) RunningEventProcessors.delete(self) - flush + change_state { @stopping = true } + @interval = nil + final_flush = Thread.new { send_batch } + @logger.warn "Gave up sending events after #{timeout_seconds} seconds while closing" unless final_flush.join(timeout_seconds) + change_state { @closed = true } end private + def initialize_buffer + @events = [] + @buffered_exposures = Set.new + # Events tracked so far, how many of them were in a batch once sent, and the batches sent + @tracked_count = 0 + @attempted_count = 0 + @settled_count = 0 + @flush_pending = false + # Whether the buffer holds a batch whose attempts all failed + @holding_failed_batch = false + end + + def initialize_state + @mutex = Mutex.new + # Held while a batch is in flight, so that one batch is sent at a time + @flush_mutex = Mutex.new + @state_changed = ConditionVariable.new + @refused = false + @stopping = false + @closed = false + end + # Only the flush scheduled by a full buffer clears +@flush_pending+, once it holds the buffer, so at most one # such flush is waiting at a time. - def scheduled_flush - drain { @flush_pending = false } + def scheduled_flush(request) + send_batch(request) { @flush_pending = false } + end + + # What a flush requested now asks for: the events tracked so far, unless a batch settled since, including + # one in flight now, sent them + def flush_request + [@tracked_count, @settled_count] end - def drain + # Sends the buffered events as one batch. Given a +flush_request+, does nothing if a batch settled since the + # request sent every event tracked up to it. + def send_batch(...) @flush_mutex.synchronize do - batch = @mutex.synchronize do - buffered = @events - @events = [] - @buffered_exposures.clear - yield if block_given? - buffered + batch, attempted_count = take_batch(...) + settle(batch, deliver(batch), attempted_count) if batch + end + rescue StandardError => e + @logger.error "Unable to send events to the Flagsmith events API: #{e}" + end + + def take_batch(request = nil) + @mutex.synchronize do + yield if block_given? + next if @closed || @events.empty? || sent_since?(request) + + @holding_failed_batch = false + [clear_buffer, @tracked_count] + end + end + + def sent_since?(request) + return false unless request + + tracked_count, settled_count = request + @settled_count > settled_count && @attempted_count >= tracked_count + end + + # Returns the events the buffer held + def clear_buffer + events = @events + @events = [] + @buffered_exposures.clear + events + end + + def settle(batch, outcome, attempted_count) + @mutex.synchronize do + @attempted_count = attempted_count + @settled_count += 1 + hold_failed_batch(batch) if outcome == :kept + next unless outcome == :refused + + @refused = true + clear_buffer + end + end + + # Keeps a failed batch ahead of the events tracked since, with one of each exposure + def hold_failed_batch(batch) + @buffered_exposures.clear + @events = (batch + @events).last(@max_buffer_items).reject { |event| duplicate_exposure?(event) } + @holding_failed_batch = true + end + + # Returns :delivered, :dropped, :kept or :refused + def deliver(batch) + body = { events: batch }.to_json + status = nil + (1..MAX_ATTEMPTS).each do |attempt| + return keep(batch, 'the client is closed') if attempt > 1 && wait(backoff(attempt)) { @closed } + + status = post(body, batch, attempt) + outcome = outcome_of(status, batch) + return outcome if outcome + + log_failed_attempt(batch, attempt, "status #{status}") if status + end + status == 500 ? drop(batch, 'its last attempt got 500') : keep(batch, 'its last attempt failed') + end + + # Full jitter: any wait up to the backoff, which doubles from +@retry_base_seconds+ + def backoff(attempt) + rand * @retry_base_seconds * (2**(attempt - 2)) + end + + # Returns the response status, or nil when the request got none + def post(body, batch, attempt) + @connection.post(ENDPOINT, body).status + rescue Faraday::Error => e + log_failed_attempt(batch, attempt, e) + nil + end + + def log_failed_attempt(batch, attempt, failure) + @logger.warn "Unable to send #{batch.length} events to the Flagsmith events API, " \ + "attempt #{attempt} of #{MAX_ATTEMPTS}: #{failure}" + end + + # Returns the outcome of a batch given its response status, or nil to try again + def outcome_of(status, batch) + return nil if status.nil? || status >= 500 || RETRIED_STATUSES.include?(status) + return :delivered if (200..299).cover?(status) + + if REFUSED_STATUSES.include?(status) + @logger.error "The Flagsmith events API refused the environment key with #{status}, stopping event tracking" + return :refused + end + + drop(batch, "the events API rejected it with #{status}") + end + + def drop(batch, reason) + @logger.warn "Dropping #{batch.length} events: #{reason}" + :dropped + end + + def keep(batch, reason) + @logger.warn "Keeping #{batch.length} events for the next flush: #{reason}" + :kept + end + + def change_state + @mutex.synchronize do + yield + @state_changed.broadcast + end + end + + # Waits for at most +seconds+, until the block is true. Returns the block's value. + def wait(seconds) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + seconds + @mutex.synchronize do + until yield + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + break if remaining <= 0 + + @state_changed.wait(@mutex, remaining) end - send_events(batch) unless batch.empty? + yield end end def build_connection(config) Faraday.new(url: config.events_api_url, headers: headers(config)) do |f| f.options.timeout = REQUEST_TIMEOUT_SECONDS - f.response :raise_error - f.request :retry, max: 1, interval: 1, methods: %i[post], retry_statuses: RETRIABLE_STATUSES, - exceptions: [*Faraday::Retry::Middleware::DEFAULT_EXCEPTIONS, Faraday::ConnectionFailed] end end @@ -135,29 +297,56 @@ def trait_values(traits) end def buffer(event) - schedule_flush = @mutex.synchronize do - next false if duplicate_exposure?(event) + request = @mutex.synchronize do + next if stopping?("tracking #{event[:event]} event") || refused?(event) || duplicate_exposure?(event) + drop_oldest if @events.length >= @max_buffer_items @events << event - next false if @flush_pending || @events.length < @max_buffer_items + @tracked_count += 1 + next unless full_buffer_flush_due? @flush_pending = true + flush_request end - Thread.new { scheduled_flush } if schedule_flush + Thread.new { scheduled_flush(request) } if request + end + + # A full buffer is flushed at once, unless such a flush is already waiting, or it holds a failed batch, + # which waits for the next flush. + def full_buffer_flush_due? + !@flush_pending && !@holding_failed_batch && @events.length >= @max_buffer_items + end + + def stopping?(action) + return false unless @stopping + + @logger.debug "Not #{action}: the client is closed" + true + end + + def refused?(event) + return false unless @refused + + @logger.debug "Not tracking #{event[:event]} event: the Flagsmith events API refused the environment key" + true + end + + def drop_oldest + dropped = @events.shift + @buffered_exposures.delete(exposure_key(dropped)) if dropped[:event] == FLAG_EXPOSURE_EVENT + @logger.warn "Event buffer is full, dropping the oldest #{dropped[:event]} event" + end + + def exposure_key(event) + event.values_at(:feature_name, :identifier, :value) end def duplicate_exposure?(event) return false unless event[:event] == FLAG_EXPOSURE_EVENT - return false if @buffered_exposures.add?(event.values_at(:feature_name, :identifier, :value)) + return false if @buffered_exposures.add?(exposure_key(event)) @logger.debug "Skipping duplicate #{FLAG_EXPOSURE_EVENT} event for feature #{event[:feature_name]}" true end - - def send_events(events) - @connection.post(ENDPOINT, { events: events }.to_json) - rescue StandardError => e - @logger.warn "Unable to send #{events.length} events to the Flagsmith events API, dropping them: #{e}" - end end end diff --git a/lib/flagsmith/sdk/models/flags.rb b/lib/flagsmith/sdk/models/flags.rb index 722db4a..2177e61 100644 --- a/lib/flagsmith/sdk/models/flags.rb +++ b/lib/flagsmith/sdk/models/flags.rb @@ -30,7 +30,7 @@ def initialize(enabled:, value:) end end - # The running experiment an identity's flag was evaluated in, as returned by the API + # The running experiment an identity's flag was evaluated in, as returned by the API or the environment document Experiment = Struct.new(:id, :name, :in_experiment, keyword_init: true) do alias_method :in_experiment?, :in_experiment @@ -192,7 +192,9 @@ def from_api(json_data, **args) ) end - def from_evaluation_result(evaluation_result, **args) + # Only an identity is enrolled in an experiment, as with remote evaluation: environment flags, + # evaluated without one, are built with +identity: false+ and carry no experiment. + def from_evaluation_result(evaluation_result, identity: true, **args) to_flag_object = lambda { |flag_result, acc| id = flag_result.dig(:metadata, :id) if id.nil? @@ -200,19 +202,21 @@ def from_evaluation_result(evaluation_result, **args) "FlagResult metadata.id is missing for feature \"#{flag_result[:name]}\". This indicates a bug in the SDK, please report it." end - acc[flag_result[:name]] = Collection.map_evaluated_flag_to_flag_result(flag_result, id) + acc[flag_result[:name]] = Collection.map_evaluated_flag_to_flag_result(flag_result, id, identity: identity) } new(evaluation_result[:flags].each_value.each_with_object({}, &to_flag_object), **args) end - def map_evaluated_flag_to_flag_result(flag_result, id) + def map_evaluated_flag_to_flag_result(flag_result, id, identity: true) Flagsmith::Flags::Flag.new( feature_name: flag_result[:name], enabled: flag_result[:enabled], value: flag_result[:value], feature_id: id, - reason: flag_result[:reason] + reason: flag_result[:reason], + variant: flag_result[:variant], + experiment: (Experiment.from_api(flag_result[:metadata]) if identity) ) end diff --git a/sdk-spec/steps.rb b/sdk-spec/steps.rb index cd8581c..a38be88 100644 --- a/sdk-spec/steps.rb +++ b/sdk-spec/steps.rb @@ -31,6 +31,14 @@ def create_client_with_events(environment_key, api_url, events_api_url) ) end +# Create the SDK client with local evaluation and events enabled +def create_client_with_local_evaluation_and_events(environment_key, api_url, events_api_url) + Flagsmith::Client.new( + environment_key: environment_key, api_url: api_url, enable_local_evaluation: true, + enable_events: true, events_api_url: events_api_url + ) +end + # When the application gets the environment flags def get_environment_flags(flagsmith) as_spec_flags(flagsmith.get_environment_flags) @@ -55,3 +63,18 @@ def track_event(flagsmith, event, identifier, value, metadata) def track_exposure_event(flagsmith, feature, identifier, value, metadata) flagsmith.track_exposure_event(feature, identifier: identifier, value: value, metadata: metadata) end + +# When the application tracks events "event_{first}" to "event_{last}" for identity "{identifier}" +def track_numbered_events(flagsmith, first, last, identifier) + (first..last).each { |number| flagsmith.track_event("event_#{number}", identifier: identifier) } +end + +# When the application flushes events +def flush_events(flagsmith) + flagsmith.flush_events +end + +# When the application closes the client with a timeout of {timeout} seconds +def close_client(flagsmith, timeout) + flagsmith.close(timeout_seconds: timeout) +end diff --git a/spec/sdk/event_processor_spec.rb b/spec/sdk/event_processor_spec.rb index 55cc71d..027ef9c 100644 --- a/spec/sdk/event_processor_spec.rb +++ b/spec/sdk/event_processor_spec.rb @@ -14,7 +14,7 @@ let(:sent_batches) { [] } let(:max_buffer_items) { 1000 } - subject { described_class.new(config, max_buffer_items: max_buffer_items) } + subject { described_class.new(config, max_buffer_items: max_buffer_items, retry_base_seconds: 0) } before do allow(Faraday).to receive(:new).and_wrap_original do |original, *args, **options, &block| @@ -32,6 +32,17 @@ def respond_to_events_with(status) end end + def reset_connection_to_events + stubs.post('/v1/events') do |env| + sent_batches << { headers: env.request_headers, body: JSON.parse(env.body) } + raise Faraday::ConnectionFailed, 'Connection reset by peer' + end + end + + def sent_events + sent_batches.map { |batch| batch[:body]['events'].map { |event| event['event'] } } + end + it 'sends a tracked event to the events API with its details' do # Given respond_to_events_with(202) @@ -128,6 +139,55 @@ def respond_to_events_with(status) expect(sent_batches[0][:body]['events'].length).to eq(2) end + it 'sends a kept batch with the next flush, even with no events tracked since' do + # Given + 3.times { respond_to_events_with(503) } + respond_to_events_with(202) + subject.track_event('checkout') + subject.flush + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[checkout]] * 4) + end + + it 'sends an exposure once when tracked again while its batch is kept' do + # Given + 3.times { respond_to_events_with(503) } + respond_to_events_with(202) + subject.track_exposure_event('banner', identifier: 'user_123', value: 'treatment') + subject.flush + + # When + subject.track_exposure_event('banner', identifier: 'user_123', value: 'treatment') + subject.flush + + # Then + expect(sent_events.last).to eq(%w[$flag_exposure]) + end + + it 'sends an exposure once when tracked again while its batch was in flight, then kept' do + # Given + stubs.post('/v1/events') do |env| + sent_batches << { body: JSON.parse(env.body) } + sleep 0.05 + [sent_batches.length > 3 ? 202 : 503, {}, ''] + end + subject.track_exposure_event('banner', identifier: 'user_123', value: 'treatment') + first_flush = Thread.new { subject.flush } + Timeout.timeout(5) { sleep 0.01 while sent_batches.empty? } + subject.track_exposure_event('banner', identifier: 'user_123', value: 'treatment') + first_flush.join + + # When + subject.flush + + # Then + expect(sent_events.last).to eq(%w[$flag_exposure]) + end + it 'sends nothing when no events were tracked' do # Given / When subject.flush @@ -171,29 +231,244 @@ def respond_to_events_with(status) it 'drops events the events API rejects, without raising' do # Given respond_to_events_with(400) - subject.track_event('checkout_completed') + respond_to_events_with(202) + subject.track_event('add_to_cart') + expect { subject.flush }.not_to raise_error + subject.track_event('checkout') # When - expect { subject.flush }.not_to raise_error + subject.flush # Then - expect(sent_batches.length).to eq(1) - expect(logger).to have_received(:warn).with(/\AUnable to send 1 events to the Flagsmith events API, dropping them: /) + expect(sent_events).to eq([%w[add_to_cart], %w[checkout]]) + expect(logger).to have_received(:warn).with('Dropping 1 events: the events API rejected it with 400') end - it 'sends events again once when the events API fails' do + it 'sends a batch again until the events API accepts it' do # Given respond_to_events_with(503) + reset_connection_to_events respond_to_events_with(202) - subject.track_event('checkout_completed') + subject.track_event('checkout') # When subject.flush # Then - expect(sent_batches.length).to eq(2) - expect(sent_batches[1][:body]).to eq(sent_batches[0][:body]) - expect(logger).not_to have_received(:warn) + expect(sent_events).to eq([%w[checkout], %w[checkout], %w[checkout]]) + expect(logger).to have_received(:warn).with( + 'Unable to send 1 events to the Flagsmith events API, attempt 1 of 3: status 503' + ) + expect(logger).to have_received(:warn).with( + 'Unable to send 1 events to the Flagsmith events API, attempt 2 of 3: Connection reset by peer' + ) + end + + it 'sends a batch again when the events API is throttling' do + # Given + respond_to_events_with(429) + respond_to_events_with(202) + subject.track_event('checkout') + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[checkout], %w[checkout]]) + end + + it 'backs off with full jitter, doubling from the retry base' do + # Given + processor = described_class.new(config, retry_base_seconds: 1) + allow(processor).to receive(:rand).and_return(1.0, 0.5) + + # When / Then + expect(processor.send(:backoff, 2)).to eq(1.0) + expect(processor.send(:backoff, 3)).to eq(1.0) + end + + it 'keeps a batch that keeps failing for the next flush, ahead of events tracked since' do + # Given + 3.times { respond_to_events_with(503) } + respond_to_events_with(202) + subject.track_event('add_to_cart') + subject.flush + subject.track_event('checkout') + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[add_to_cart]] * 3 + [%w[add_to_cart checkout]]) + expect(logger).to have_received(:warn).with('Keeping 1 events for the next flush: its last attempt failed') + end + + it 'drops a batch whose last attempt gets a server error' do + # Given + 2.times { respond_to_events_with(503) } + respond_to_events_with(500) + respond_to_events_with(202) + subject.track_event('add_to_cart') + subject.flush + subject.track_event('checkout') + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[add_to_cart]] * 3 + [%w[checkout]]) + expect(logger).to have_received(:warn).with('Dropping 1 events: its last attempt got 500') + end + + it 'keeps a batch whose last attempt fails otherwise, after server errors' do + # Given + 2.times { respond_to_events_with(500) } + respond_to_events_with(503) + respond_to_events_with(202) + subject.track_event('add_to_cart') + subject.flush + subject.track_event('checkout') + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[add_to_cart]] * 3 + [%w[add_to_cart checkout]]) + end + + [401, 403].each do |status| + it "stops tracking events once the events API refuses the environment key with #{status}" do + # Given + respond_to_events_with(status) + subject.track_event('add_to_cart') + subject.flush + + # When + subject.track_event('checkout') + subject.flush + + # Then + expect(sent_events).to eq([%w[add_to_cart]]) + expect(logger).to have_received(:error).with( + "The Flagsmith events API refused the environment key with #{status}, stopping event tracking" + ) + end + end + + context 'when the buffer is full of events the events API failed' do + let(:max_buffer_items) { 2 } + + it 'drops the oldest event' do + # Given + 3.times { respond_to_events_with(503) } + respond_to_events_with(202) + subject.track_event('first') + subject.track_event('second') + subject.flush + + # When + subject.track_event('third') + subject.flush + + # Then + expect(sent_events.last).to eq(%w[second third]) + expect(sent_events.flatten.count('first')).to eq(3) + expect(logger).to have_received(:warn).with('Event buffer is full, dropping the oldest first event') + end + + it 'waits for the next flush rather than attempting the kept batch again' do + # Given + 3.times { respond_to_events_with(503) } + subject.track_event('first') + subject.track_event('second') + subject.flush + allow(Thread).to receive(:new).and_call_original + + # When + subject.track_event('third') + subject.track_event('fourth') + + # Then + expect(Thread).not_to have_received(:new) + expect(sent_events.length).to eq(3) + end + + it 'does not send again a batch in flight when the flush was requested' do + # Given + stubs.post('/v1/events') do |env| + sent_batches << { body: JSON.parse(env.body) } + sleep 0.05 + [sent_batches.length > 3 ? 202 : 503, {}, ''] + end + subject.track_event('first') + subject.track_event('second') + Timeout.timeout(5) { sleep 0.01 while sent_batches.empty? } + + # When + subject.flush + + # Then + expect(sent_events).to eq([%w[first second]] * 3) + end + + it 'sends a full buffer without waiting again once the kept batch is sent' do + # Given + 3.times { respond_to_events_with(503) } + 2.times { respond_to_events_with(202) } + subject.track_event('first') + subject.track_event('second') + subject.flush + Timeout.timeout(5) { sleep 0.01 while subject.instance_variable_get(:@flush_pending) } + subject.flush + allow(Thread).to receive(:new).and_call_original + + # When + subject.track_event('third') + subject.track_event('fourth') + + # Then + expect(Thread).to have_received(:new).once + Timeout.timeout(5) { sleep 0.01 until sent_events.last == %w[third fourth] } + expect(sent_events).to eq([%w[first second]] * 4 + [%w[third fourth]]) + end + end + + context 'when stopped' do + before do + respond_to_events_with(202) + subject.start + subject.stop + end + + it 'tracks no more events' do + # Given / When + subject.track_event('checkout') + subject.send(:send_batch) + + # Then + expect(sent_batches).to be_empty + expect(logger).to have_received(:debug).with('Not tracking checkout event: the client is closed') + end + + it 'flushes nothing' do + # Given + subject.instance_variable_get(:@events) << { event: 'checkout' } + + # When + subject.flush + + # Then + expect(sent_batches).to be_empty + expect(logger).to have_received(:debug).with('Not flushing events: the client is closed') + end + + it 'does not start again' do + # Given / When + subject.start + + # Then + expect(Flagsmith::RunningEventProcessors.to_a).not_to include(subject) + end end it 'sends the remaining events when stopped' do @@ -209,6 +484,46 @@ def respond_to_events_with(status) expect(sent_batches.length).to eq(1) end + it 'returns from stopping within its timeout while the events API does not answer, and retries no more' do + # Given + stubs.post('/v1/events') do |env| + sent_batches << { body: JSON.parse(env.body) } + sleep 0.5 + [503, {}, ''] + end + subject.track_event('checkout') + + # When + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + subject.stop(0.1) + + # Then + expect(Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at).to be < 0.4 + sleep 0.6 + expect(sent_batches.length).to eq(1) + expect(logger).to have_received(:warn).with('Gave up sending events after 0.1 seconds while closing') + end + + it 'takes no batch to send once stopping has returned' do + # Given + stubs.post('/v1/events') do |env| + sent_batches << { body: JSON.parse(env.body) } + sleep 0.3 + [202, {}, ''] + end + subject.track_event('first') + Thread.new { subject.flush } + Timeout.timeout(5) { sleep 0.01 while sent_batches.empty? } + subject.track_event('second') + + # When + subject.stop(0.05) + + # Then + sleep 0.5 + expect(sent_events).to eq([%w[first]]) + end + it 'is stopped at exit while it is running' do # Given / When subject.start diff --git a/spec/sdk/experiments_spec.rb b/spec/sdk/experiments_spec.rb index a938253..e888d4d 100644 --- a/spec/sdk/experiments_spec.rb +++ b/spec/sdk/experiments_spec.rb @@ -135,24 +135,72 @@ end context 'when the flags come from the environment document' do + let(:environment_document) do + JSON.parse(File.read('spec/sdk/fixtures/experiment_environment.json'), symbolize_names: true) + end + subject do + allow(api_client).to receive(:get).with('environment-document/') + .and_return(OpenStruct.new(body: environment_document)) described_class.new( - environment_key: 'ser.environment-key', enable_events: true, logger: logger, - offline_handler: Flagsmith::OfflineHandlers::LocalFileHandler.new('spec/sdk/fixtures/environment.json') + environment_key: 'ser.environment-key', enable_events: true, enable_local_evaluation: true, logger: logger ) end - it 'returns the identity flag and records no exposure' do + it 'records the exposure of an identity the running experiment enrolls to its variant' do # Given / When - flag = subject.get_experiment_flag('some_feature', 'user_123') + flag = subject.get_experiment_flag('checkout', 'user_123', plan: 'premium') # Then - expect(flag.value).to eq('some-value') - expect(subject.event_processor).not_to have_received(:track_exposure_event) - expect(logger).to have_received(:debug).with( - 'Not sending $flag_exposure for feature some_feature: ' \ - 'flags come from the environment document, which carries no experiment data.' + expect(flag.value).to eq('new checkout') + expect(flag.variant).to eq('treatment') + expect(flag.experiment).to eq( + Flagsmith::Flags::Experiment.new(id: 42, name: 'Checkout experiment', in_experiment: true) + ) + expect(subject.event_processor).to have_received(:track_exposure_event).with( + 'checkout', identifier: 'user_123', value: 'treatment', traits: { plan: 'premium' }, metadata: { experiment_id: 42 } + ) + end + + it 'records no exposure of an identity the running experiment does not enroll' do + # Given / When + flag = subject.get_experiment_flag('checkout', 'user_123', plan: 'free') + + # Then + expect(flag.value).to eq('control') + expect(flag.experiment).to eq( + Flagsmith::Flags::Experiment.new(id: 42, name: 'Checkout experiment', in_experiment: false) ) + expect(subject.event_processor).not_to have_received(:track_exposure_event) + end + + it 'records no exposure of an identity with an identity override' do + # Given / When + flag = subject.get_experiment_flag('checkout', 'user_overridden') + + # Then + expect(flag.value).to eq('override') + expect(flag.experiment).to be_nil + expect(subject.event_processor).not_to have_received(:track_exposure_event) + end + + it 'records no exposure for a feature without a running experiment' do + # Given / When + flag = subject.get_experiment_flag('plain', 'user_123', plan: 'premium') + + # Then + expect(flag.value).to eq('hello') + expect(flag.experiment).to be_nil + expect(subject.event_processor).not_to have_received(:track_exposure_event) + end + + it 'serves environment flags without an experiment' do + # Given / When + flag = subject.get_environment_flags.get_flag('checkout') + + # Then + expect(flag.value).to eq('control') + expect(flag.experiment).to be_nil end end @@ -223,6 +271,62 @@ end end + describe '#flush_events' do + it 'flushes the tracked events' do + # Given + allow(subject.event_processor).to receive(:flush) + + # When + subject.flush_events + + # Then + expect(subject.event_processor).to have_received(:flush) + end + + context 'when events are not enabled' do + let(:enable_events) { false } + + it 'raises an error' do + # Given / When / Then + expect { subject.flush_events } + .to raise_error(Flagsmith::ClientError, 'The enable_events config param is required to send events.') + end + end + end + + describe '#close' do + it 'stops the event processor within the timeout' do + # Given + allow(subject.event_processor).to receive(:stop) + + # When + subject.close(timeout_seconds: 2) + + # Then + expect(subject.event_processor).to have_received(:stop).with(2) + end + + it 'stops the event processor within 5 seconds by default' do + # Given + allow(subject.event_processor).to receive(:stop) + + # When + subject.close + + # Then + expect(subject.event_processor).to have_received(:stop).with(5) + end + + context 'when events are not enabled' do + let(:enable_events) { false } + + it 'closes without error' do + # Given / When / Then + expect { subject.close }.not_to raise_error + end + end + end + describe '#event_processor' do it 'sends events to the configured events API' do # Given / When diff --git a/spec/sdk/fixtures/experiment_environment.json b/spec/sdk/fixtures/experiment_environment.json new file mode 100644 index 0000000..38c122c --- /dev/null +++ b/spec/sdk/fixtures/experiment_environment.json @@ -0,0 +1,84 @@ +{ + "api_key": "B62qaMZNwfiqT76p38ggrQ", + "name": "Test environment", + "project": { + "name": "Test project", + "organisation": { + "feature_analytics": false, + "name": "Test Org", + "id": 1, + "persist_trait_data": true, + "stop_serving_flags": false + }, + "id": 1, + "hide_disabled_flags": false, + "segments": [ + { + "name": "Checkout experiment rollout", + "id": 7, + "rules": [ + { + "type": "ALL", + "conditions": [{ "property_": "plan", "operator": "EQUAL", "value": "premium" }], + "rules": [] + } + ], + "feature_states": [ + { + "django_id": 2, + "featurestate_uuid": "c2c2c2c2-c2c2-c2c2-c2c2-c2c2c2c2c2c2", + "feature": { "id": 10, "name": "checkout", "type": "MULTIVARIATE" }, + "feature_segment": { "priority": 0 }, + "enabled": true, + "feature_state_value": "control", + "multivariate_feature_state_values": [ + { + "id": 3, + "multivariate_feature_option": { "id": 3, "value": "new checkout", "key": "treatment" }, + "percentage_allocation": 100, + "mv_fs_value_uuid": "3e3e3e3e-3e3e-3e3e-3e3e-3e3e3e3e3e3e" + } + ], + "metadata": { "experiment": { "id": 42, "name": "Checkout experiment", "in_experiment": true } } + } + ] + } + ] + }, + "id": 1, + "feature_states": [ + { + "django_id": 1, + "featurestate_uuid": "c1c1c1c1-c1c1-c1c1-c1c1-c1c1c1c1c1c1", + "feature": { "id": 10, "name": "checkout", "type": "MULTIVARIATE" }, + "enabled": true, + "feature_state_value": "control", + "multivariate_feature_state_values": [], + "metadata": { "experiment": { "id": 42, "name": "Checkout experiment", "in_experiment": false } } + }, + { + "django_id": 4, + "featurestate_uuid": "c4c4c4c4-c4c4-c4c4-c4c4-c4c4c4c4c4c4", + "feature": { "id": 11, "name": "plain", "type": "STANDARD" }, + "enabled": true, + "feature_state_value": "hello", + "multivariate_feature_state_values": [] + } + ], + "identity_overrides": [ + { + "identifier": "user_overridden", + "identity_uuid": "0f21cde8-63c5-4e50-baca-87897fa6cd01", + "environment_api_key": "B62qaMZNwfiqT76p38ggrQ", + "identity_features": [ + { + "featurestate_uuid": "c5c5c5c5-c5c5-c5c5-c5c5-c5c5c5c5c5c5", + "feature": { "id": 10, "name": "checkout", "type": "MULTIVARIATE" }, + "enabled": true, + "feature_state_value": "override", + "multivariate_feature_state_values": [] + } + ] + } + ] +}