[codex] Fix atomic sync gates

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

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
 

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

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