fix: reset flag-called tracking when definitions reload

.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.

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

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)

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