From 4f4b1de269fc651c7a79a1ba0f09d4ed0c90bf96 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 19:24:13 +0200 Subject: [PATCH] fix: reset flag-called tracking when definitions reload --- .changeset/fresh-flag-reload-tracking.md | 5 + lib/posthog/client.rb | 18 +- lib/posthog/feature_flags.rb | 8 +- .../feature_flag_called_reload_spec.rb | 209 ++++++++++++++++++ 4 files changed, 233 insertions(+), 7 deletions(-) create mode 100644 .changeset/fresh-flag-reload-tracking.md create mode 100644 spec/posthog/feature_flag_called_reload_spec.rb diff --git a/.changeset/fresh-flag-reload-tracking.md b/.changeset/fresh-flag-reload-tracking.md new file mode 100644 index 0000000..5b4bf54 --- /dev/null +++ b/.changeset/fresh-flag-reload-tracking.md @@ -0,0 +1,5 @@ +--- +"posthog-ruby": patch +--- + +Reset feature flag event deduplication when local flag definitions are refreshed or discarded, allowing the next flag access to emit a fresh event. diff --git a/lib/posthog/client.rb b/lib/posthog/client.rb index 96ea536..fccbabc 100644 --- a/lib/posthog/client.rb +++ b/lib/posthog/client.rb @@ -191,6 +191,12 @@ def initialize(opts = {}) end end + # Initialize tracking before the poller can load definitions (including async loads). + @distinct_id_has_sent_flag_calls_mutex = Mutex.new + @distinct_id_has_sent_flag_calls = SizeLimitedHash.new(Defaults::MAX_HASH_SIZE) do |hash, key| + hash[key] = SizeLimitedArray.new(Defaults::MAX_HASH_SIZE) + end + unless @disabled @feature_flags_poller = FeatureFlagsPoller.new( @@ -203,15 +209,15 @@ def initialize(opts = {}) flag_definition_cache_provider: opts[:flag_definition_cache_provider], feature_flag_request_max_retries: opts[:feature_flag_request_max_retries], async_load: opts[:feature_flags_async_load] == true, - user_agent: @headers['User-Agent'] + user_agent: @headers['User-Agent'], + on_flag_definitions_updated: lambda { + @distinct_id_has_sent_flag_calls_mutex.synchronize do + @distinct_id_has_sent_flag_calls.clear + end + } ) end - @distinct_id_has_sent_flag_calls_mutex = Mutex.new - @distinct_id_has_sent_flag_calls = SizeLimitedHash.new(Defaults::MAX_HASH_SIZE) do |hash, key| - hash[key] = SizeLimitedArray.new(Defaults::MAX_HASH_SIZE) - end - @before_send = opts[:before_send] @is_server = opts.fetch(:is_server, true) != false @deprecation_emitted_for = Concurrent::Set.new diff --git a/lib/posthog/feature_flags.rb b/lib/posthog/feature_flags.rb index bae1698..cb3bd4a 100644 --- a/lib/posthog/feature_flags.rb +++ b/lib/posthog/feature_flags.rb @@ -45,6 +45,7 @@ class FeatureFlagsPoller # immediate first tick at construction, then the regular polling cadence, which keeps retrying until a # load succeeds. # @param user_agent [String] User-Agent header sent with feature flag requests. + # @param on_flag_definitions_updated [Proc, nil] Internal callback after definitions are applied or discarded. def initialize( polling_interval, secret_key, @@ -55,7 +56,8 @@ def initialize( flag_definition_cache_provider: nil, feature_flag_request_max_retries: nil, async_load: false, - user_agent: "posthog-ruby/#{PostHog::VERSION}" + user_agent: "posthog-ruby/#{PostHog::VERSION}", + on_flag_definitions_updated: nil ) @polling_interval = polling_interval || Defaults::FeatureFlags::POLLING_INTERVAL_SECONDS @secret_key = secret_key @@ -75,6 +77,7 @@ def initialize( @flag_definitions_loaded_at = Concurrent::AtomicReference.new(nil) @async_load = async_load @user_agent = user_agent + @on_flag_definitions_updated = on_flag_definitions_updated # Server-controlled gate for minimal `$feature_flag_called` events, read # from the top-level `minimal_flag_called_events` key of the local # evaluation definitions payload. false when the server does not send it. @@ -1217,6 +1220,7 @@ def _fetch_and_apply_flag_definitions # Handle quota limits with 402 status if res.is_a?(Hash) && res[:status] == 402 + definitions_were_loaded = definitions_loaded? logger.warn( '[FEATURE FLAGS] Feature flags quota limit exceeded - unsetting all local flags. ' \ 'Learn more about billing limits at https://posthog.com/docs/billing/limits-alerts' @@ -1229,6 +1233,7 @@ def _fetch_and_apply_flag_definitions @minimal_flag_called_events = false @loaded_flags_successfully_once.make_false @quota_limited.make_true + @on_flag_definitions_updated&.call if definitions_were_loaded return end @@ -1280,6 +1285,7 @@ def _apply_flag_definitions(data) logger.debug "Loaded #{@feature_flags.length} feature flags and #{@cohorts.length} cohorts" @flag_definitions_loaded_at.value = (Time.now.to_f * 1000).to_i @loaded_flags_successfully_once.make_true if @loaded_flags_successfully_once.false? + @on_flag_definitions_updated&.call end def _request_feature_flag_definitions(etag: nil) diff --git a/spec/posthog/feature_flag_called_reload_spec.rb b/spec/posthog/feature_flag_called_reload_spec.rb new file mode 100644 index 0000000..1d91970 --- /dev/null +++ b/spec/posthog/feature_flag_called_reload_spec.rb @@ -0,0 +1,209 @@ +# frozen_string_literal: true + +require 'spec_helper' + +module PostHog + describe 'feature flag called tracking across definition reloads' do + let(:definitions_endpoint) { 'https://us.i.posthog.com/flags/definitions?token=testsecret&send_cohorts=true' } + let(:flag_definition) do + { + id: 1, + key: 'beta-feature', + active: true, + version: 1, + filters: { groups: [{ properties: [], rollout_percentage: 100 }] } + } + end + let(:definitions) { { flags: [flag_definition] } } + let(:client) { @client || build_client } + let(:poller) { client.instance_variable_get(:@feature_flags_poller) } + + before do + stub_request(:get, definitions_endpoint).to_return(status: 200, body: definitions.to_json) + end + + after { @client&.shutdown } + + def build_client(**opts) + @client = Client.new( + api_key: API_KEY, + secret_key: API_KEY, + test_mode: true, + feature_flag_request_max_retries: 0, + **opts + ) + end + + def read_flag + client.get_feature_flag('beta-feature', 'user', only_evaluate_locally: true) + end + + def expect_single_event + expect(client.queued_messages).to eq(1) + expect(client.dequeue_last_message[:event]).to eq('$feature_flag_called') + end + + it 'emits once per user and flag response between successful manual reloads' do + 2.times { expect(read_flag).to be(true) } + expect_single_event + + client.reload_feature_flags + + 2.times { expect(read_flag).to be(true) } + expect_single_event + end + + it 'resets the shared tracker for both snapshot and single-flag access' do + snapshot = client.evaluate_flags('user', only_evaluate_locally: true) + expect(snapshot.get_flag('beta-feature')).to be(true) + expect(read_flag).to be(true) + expect_single_event + + client.reload_feature_flags + + expect(snapshot.get_flag('beta-feature')).to be(true) + expect(read_flag).to be(true) + expect_single_event + end + + it 'resets tracking on an automatic background refresh' do + refresh_allowed = Queue.new + requests = Concurrent::AtomicFixnum.new(0) + stub_request(:get, definitions_endpoint).to_return do + case requests.increment + when 1 + { status: 200, body: definitions.to_json } + when 2 + refresh_allowed.pop + { status: 200, body: { flags: [flag_definition.merge(version: 2)] }.to_json } + else + { status: 304, body: '' } + end + end + build_client(feature_flags_polling_interval: 0.1) + 2.times { expect(read_flag).to be(true) } + expect_single_event + + refresh_allowed << true + + eventually do + expect(read_flag).to be(true) + expect(client.queued_messages).to eq(1) + end + expect_single_event + expect(poller.feature_flags_by_key['beta-feature'][:version]).to eq(2) + 2.times { expect(read_flag).to be(true) } + expect(client.queued_messages).to eq(0) + ensure + refresh_allowed&.push(true) + end + + it 'resets tracking when definitions are applied from the external cache' do + provider = double( + 'cache provider', + should_fetch_flag_definitions?: true, + on_flag_definitions_received: nil, + shutdown: nil, + flag_definitions: definitions + ) + build_client(flag_definition_cache_provider: provider) + 2.times { expect(read_flag).to be(true) } + expect_single_event + allow(provider).to receive(:should_fetch_flag_definitions?).and_return(false) + + poller._load_feature_flags + + 2.times { expect(read_flag).to be(true) } + expect_single_event + expect(WebMock).to have_requested(:get, definitions_endpoint).once + end + + it 'resets tracking when an empty definitions response is applied' do + 2.times { client.get_feature_flag('missing', 'user', only_evaluate_locally: true) } + expect_single_event + stub_request(:get, definitions_endpoint).to_return(status: 200, body: { flags: [] }.to_json) + + client.reload_feature_flags + + 2.times { client.get_feature_flag('missing', 'user', only_evaluate_locally: true) } + expect_single_event + end + + it 'resets tracking when a quota-limited response discards definitions' do + 2.times { client.get_feature_flag('missing', 'user', only_evaluate_locally: true) } + expect_single_event + stub_request(:get, definitions_endpoint).to_return(status: 402, body: '{}') + + client.reload_feature_flags + + expect(client.feature_flags_loaded?).to be(false) + 2.times { client.get_feature_flag('missing', 'user', only_evaluate_locally: true) } + expect_single_event + end + + [ + { status: 304, body: '' }, + { status: 500, body: '{}' }, + { status: 200, body: '{}' }, + { status: 200, body: 'invalid json' } + ].each do |response| + it "preserves tracking when no definitions are applied (#{response})" do + 2.times { expect(read_flag).to be(true) } + expect_single_event + stub_request(:get, definitions_endpoint).to_return(response) + + client.reload_feature_flags + + 2.times { expect(read_flag).to be(true) } + expect(client.queued_messages).to eq(0) + end + end + + it 'preserves tracking when the reload request times out' do + expect(read_flag).to be(true) + expect_single_event + stub_request(:get, definitions_endpoint).to_timeout + + client.reload_feature_flags + + expect(read_flag).to be(true) + expect(client.queued_messages).to eq(0) + end + + it 'clears tracking under the same mutex used to suppress duplicates' do + expect(read_flag).to be(true) + tracker = client.instance_variable_get(:@distinct_id_has_sent_flag_calls) + mutex = client.instance_variable_get(:@distinct_id_has_sent_flag_calls_mutex) + allow(tracker).to receive(:clear).and_wrap_original do |clear| + expect(mutex.owned?).to be(true) + clear.call + end + + client.reload_feature_flags + + expect(tracker).to have_received(:clear).once + end + + it 'continues suppressing concurrent duplicate reads after reload' do + expect(read_flag).to be(true) + expect_single_event + client.reload_feature_flags + start = Queue.new + threads = Array.new(4) do + Thread.new do + start.pop + 5.times { read_flag } + end + end + threads.length.times { start << true } + threads.each do |thread| + expect(thread.join(2)).to eq(thread) + thread.value + end + + expect_single_event + ensure + threads&.each { |thread| thread.kill if thread.alive? } + end + end +end