From a19284ef33cfce3c5bcdbb22a68ad76a43798f6c Mon Sep 17 00:00:00 2001 From: John Nunemaker Date: Wed, 8 Jul 2026 12:33:52 -0600 Subject: [PATCH] Fix atomic sync gates --- lib/flipper/adapters/poll.rb | 55 ++++++- .../adapters/sync/interval_synchronizer.rb | 43 ++++- spec/flipper/adapters/poll_spec.rb | 155 ++++++++++++++++++ .../sync/interval_synchronizer_spec.rb | 66 ++++++++ 4 files changed, 311 insertions(+), 8 deletions(-) diff --git a/lib/flipper/adapters/poll.rb b/lib/flipper/adapters/poll.rb index 20cd9e920..7b810ae88 100644 --- a/lib/flipper/adapters/poll.rb +++ b/lib/flipper/adapters/poll.rb @@ -17,7 +17,11 @@ class Poll def initialize(poller, adapter) @adapter = adapter @poller = poller + @pid = Process.pid @last_synced_at = 0 + @syncing = false + @sync_mutex = Mutex.new + @sync_condition = ConditionVariable.new # If the adapter is empty, we need to sync before starting the poller. # Yes, this will block the main thread, but that's better than thinking @@ -39,14 +43,59 @@ def initialize(poller, adapter) private def synced_adapter + reset_sync_state_if_forked @poller.start poller_last_synced_at = @poller.last_synced_at.value - if poller_last_synced_at > @last_synced_at - Flipper::Adapters::Sync::Synchronizer.new(@adapter, @poller.adapter).call - @last_synced_at = poller_last_synced_at + if claim_sync(poller_last_synced_at) + begin + Flipper::Adapters::Sync::Synchronizer.new(@adapter, @poller.adapter).call + complete_sync(poller_last_synced_at) + rescue + release_sync + raise + end end @adapter end + + def reset_sync_state_if_forked + return if @pid == Process.pid + + @pid = Process.pid + @syncing = false + @sync_mutex = Mutex.new + @sync_condition = ConditionVariable.new + end + + def claim_sync(poller_last_synced_at) + @sync_mutex.synchronize do + loop do + return false unless poller_last_synced_at > @last_synced_at + + unless @syncing + @syncing = true + return true + end + + @sync_condition.wait(@sync_mutex) + end + end + end + + def complete_sync(poller_last_synced_at) + @sync_mutex.synchronize do + @last_synced_at = poller_last_synced_at + @syncing = false + @sync_condition.broadcast + end + end + + def release_sync + @sync_mutex.synchronize do + @syncing = false + @sync_condition.broadcast + end + end end end end diff --git a/lib/flipper/adapters/sync/interval_synchronizer.rb b/lib/flipper/adapters/sync/interval_synchronizer.rb index f84309d74..6d91b37c4 100644 --- a/lib/flipper/adapters/sync/interval_synchronizer.rb +++ b/lib/flipper/adapters/sync/interval_synchronizer.rb @@ -21,22 +21,55 @@ def initialize(synchronizer, interval: nil) @interval = interval || DEFAULT_INTERVAL # TODO: add jitter to this so all processes booting at the same time # don't phone home at the same time. + @pid = Process.pid @last_sync_at = 0 + @syncing = false + @sync_mutex = Mutex.new end def call - return unless time_to_sync? + reset_sync_state_if_forked + return unless sync_needed? - @last_sync_at = now - @synchronizer.call + begin + @synchronizer.call + ensure + complete_sync + end nil end private - def time_to_sync? - seconds_since_last_sync = now - @last_sync_at + def reset_sync_state_if_forked + return if @pid == Process.pid + + @pid = Process.pid + @syncing = false + @sync_mutex = Mutex.new + end + + def sync_needed? + @sync_mutex.synchronize do + current_time = now + return false unless time_to_sync?(current_time) + return false if @syncing + + @last_sync_at = current_time + @syncing = true + true + end + end + + def complete_sync + @sync_mutex.synchronize do + @syncing = false + end + end + + def time_to_sync?(current_time) + seconds_since_last_sync = current_time - @last_sync_at seconds_since_last_sync >= @interval end diff --git a/spec/flipper/adapters/poll_spec.rb b/spec/flipper/adapters/poll_spec.rb index 2fe08fe75..0239b8738 100644 --- a/spec/flipper/adapters/poll_spec.rb +++ b/spec/flipper/adapters/poll_spec.rb @@ -38,4 +38,159 @@ expect(local_adapter.features).to eq(remote_adapter.features) end + + it "only synchronizes once per poller update when called concurrently" do + flipper = Flipper.new(local_adapter) + flipper.enable(:existing) + + get_all_calls = Concurrent::AtomicFixnum.new(0) + slow_remote_adapter = Class.new do + def initialize(result, get_all_calls) + @result = result + @get_all_calls = get_all_calls + end + + def get_all(**kwargs) + @get_all_calls.increment + sleep 0.05 + @result + end + end.new(local_adapter.get_all, get_all_calls) + + fake_poller = Struct.new(:last_synced_at, :adapter) do + def start + end + + def sync + raise "sync should not be called when the local adapter is not empty" + end + end.new(Concurrent::AtomicFixnum.new(1), slow_remote_adapter) + + instance = described_class.new(fake_poller, local_adapter) + threads = 10.times.map { Thread.new { instance.features } } + threads.each(&:join) + + expect(get_all_calls.value).to eq(1) + end + + it "waits for an in-flight poller update before returning the adapter" do + flipper = Flipper.new(local_adapter) + flipper.enable(:existing) + + remote = Flipper::Adapters::Memory.new(threadsafe: true) + Flipper.new(remote).enable(:updated) + + entered = Queue.new + release = Queue.new + slow_remote_adapter = Class.new do + def initialize(result, entered, release) + @result = result + @entered = entered + @release = release + end + + def get_all(**kwargs) + @entered << true + @release.pop + @result + end + end.new(remote.get_all, entered, release) + + fake_poller = Struct.new(:last_synced_at, :adapter) do + def start + end + + def sync + raise "sync should not be called when the local adapter is not empty" + end + end.new(Concurrent::AtomicFixnum.new(1), slow_remote_adapter) + + instance = described_class.new(fake_poller, local_adapter) + first_thread = Thread.new { instance.features } + entered.pop + + completed = Queue.new + second_thread = Thread.new { completed << instance.features } + sleep 0.05 + + expect(completed).to be_empty + + release << true + expect(first_thread.value).to eq(Set["updated"]) + expect(completed.pop).to eq(Set["updated"]) + second_thread.join + end + + it "retries a poller update after synchronization fails" do + flipper = Flipper.new(local_adapter) + flipper.enable(:existing) + + get_all_calls = Concurrent::AtomicFixnum.new(0) + flaky_remote_adapter = Class.new do + def initialize(result, get_all_calls) + @result = result + @get_all_calls = get_all_calls + end + + def get_all(**kwargs) + raise "transient failure" if @get_all_calls.increment == 1 + + @result + end + end.new(local_adapter.get_all, get_all_calls) + + fake_poller = Struct.new(:last_synced_at, :adapter) do + def start + end + + def sync + raise "sync should not be called when the local adapter is not empty" + end + end.new(Concurrent::AtomicFixnum.new(1), flaky_remote_adapter) + + instance = described_class.new(fake_poller, local_adapter) + + expect { instance.features }.to raise_error("transient failure") + instance.features + + expect(get_all_calls.value).to eq(2) + end + + it "resets in-flight synchronization state after a fork" do + flipper = Flipper.new(local_adapter) + flipper.enable(:existing) + + remote = Flipper::Adapters::Memory.new(threadsafe: true) + Flipper.new(remote).enable(:updated) + + get_all_calls = Concurrent::AtomicFixnum.new(0) + counting_remote_adapter = Class.new do + def initialize(result, get_all_calls) + @result = result + @get_all_calls = get_all_calls + end + + def get_all(**kwargs) + @get_all_calls.increment + @result + end + end.new(remote.get_all, get_all_calls) + + fake_poller = Struct.new(:last_synced_at, :adapter) do + def start + end + + def sync + raise "sync should not be called when the local adapter is not empty" + end + end.new(Concurrent::AtomicFixnum.new(1), counting_remote_adapter) + + instance = described_class.new(fake_poller, local_adapter) + instance.instance_variable_set(:@syncing, true) + + allow(Process).to receive(:pid).and_return(instance.instance_variable_get(:@pid) + 1) + + expect(instance.features).to eq(Set["updated"]) + expect(get_all_calls.value).to eq(1) + end end diff --git a/spec/flipper/adapters/sync/interval_synchronizer_spec.rb b/spec/flipper/adapters/sync/interval_synchronizer_spec.rb index e2076c26f..7408aeac2 100644 --- a/spec/flipper/adapters/sync/interval_synchronizer_spec.rb +++ b/spec/flipper/adapters/sync/interval_synchronizer_spec.rb @@ -30,4 +30,70 @@ subject.call expect(events.size).to be(1) end + + it "does not synchronize again while a claimed interval sync is in flight" do + entered = Queue.new + release = Queue.new + synchronizer = -> do + events << now + entered << true + release.pop + end + instance = described_class.new(synchronizer, interval: interval) + + allow(instance).to receive(:now).and_return(interval) + + first_thread = Thread.new { instance.call } + entered.pop + + threads = 10.times.map { Thread.new { instance.call } } + sleep 0.05 + + expect(events.size).to eq(1) + + release << true + ([first_thread] + threads).each(&:join) + + expect(events.size).to eq(1) + end + + it "does not synchronize again when the interval passes during an in-flight sync" do + current_time = interval + entered = Queue.new + release = Queue.new + synchronizer = -> do + events << current_time + entered << true + release.pop + end + instance = described_class.new(synchronizer, interval: interval) + + allow(instance).to receive(:now) { current_time } + + first_thread = Thread.new { instance.call } + entered.pop + + current_time += interval + second_thread = Thread.new { instance.call } + sleep 0.05 + + expect(events.size).to eq(1) + + release << true + [first_thread, second_thread].each(&:join) + + expect(events.size).to eq(1) + end + + it "resets in-flight synchronization state after a fork" do + instance = described_class.new(synchronizer, interval: interval) + instance.instance_variable_set(:@syncing, true) + + allow(instance).to receive(:now).and_return(interval) + allow(Process).to receive(:pid).and_return(instance.instance_variable_get(:@pid) + 1) + + instance.call + + expect(events.size).to eq(1) + end end