test: Remove the poller race from the big segment status listener spec

spec/ldclient_listeners_spec.rb

@@ -13,8 +13,8 @@ module LaunchDarkly
       end
 
       it "sends status updates" do
-        store = MockBigSegmentStore.new
-        store.setup_metadata(Time.now)
+        store = GatedBigSegmentStore.new
+        store.queue_metadata(Time.now)
         big_segments_config = BigSegmentsConfig.new(
           store: store,
           status_poll_interval: 0.01
@@ -24,17 +24,29 @@ module LaunchDarkly
           expect(status1.available).to be(true)
           expect(status1.stale).to be(false)
 
+          # Status reads after the first come from the cache, so the second query comes from the
+          # poller. Observers have been notified of the first status by the time it starts, so the
+          # observer added below sees only later changes.
+          expect(store.wait_for_queries(2)).to be(true)
+
           statuses = Queue.new
           observer = SimpleObserver.adding_to_queue(statuses)
           client.big_segment_store_status_provider.add_observer(observer)
 
-          store.setup_metadata_error(StandardError.new("sorry"))
+          # Fail the waiting query, then wait for the poller to start the next one. It has notified
+          # observers of the failure by then.
+          store.queue_metadata_error(StandardError.new("sorry"))
+          expect(store.wait_for_queries(3)).to be(true)
 
+          # The observer should have received one status, reporting the store as unavailable.
+          expect(statuses.size).to eq(1)
           status2 = statuses.pop
           expect(status2.available).to be(false)
           expect(status2.stale).to be(false)
 
           expect(client.big_segment_store_status_provider.status).to eq(status2)
+        ensure
+          store.release
         end
       end
     end

spec/mock_components.rb

@@ -116,6 +116,87 @@ def setup_segment_for_context(user_key, segment, included)
     end
   end
 
+  #
+  # A Big Segment store whose metadata queries wait for the test to supply each result.
+  #
+  # The status poller notifies observers of one result before it starts its next query. Once a
+  # test sees a query start, every earlier result has been delivered, and the poller stays in that
+  # query until the test supplies another result.
+  #
+  # Call {#release} before closing the client. The poller cannot stop while a query is waiting.
+  #
+  class GatedBigSegmentStore
+    def initialize
+      @lock = Mutex.new
+      @changed = ConditionVariable.new
+      @results = []
+      @last_result = nil
+      @queries_started = 0
+      @released = false
+    end
+
+    def get_metadata
+      result = @lock.synchronize do
+        @queries_started += 1
+        @changed.broadcast
+        @changed.wait(@lock) while @results.empty? && !@released
+        @last_result = @results.shift unless @results.empty?
+        @last_result
+      end
+      raise result if result.is_a?(Exception)
+      result
+    end
+
+    def get_membership(_context_hash)
+      nil
+    end
+
+    def stop
+    end
+
+    # Supplies the result of the next metadata query.
+    def queue_metadata(last_up_to_date)
+      queue_result(Interfaces::BigSegmentStoreMetadata.new(last_up_to_date.to_f * 1000))
+    end
+
+    # Makes the next metadata query raise the given error.
+    def queue_metadata_error(ex)
+      queue_result(ex)
+    end
+
+    #
+    # Blocks until the store has received `count` metadata queries, or until the timeout passes.
+    #
+    # @return [Boolean] false if the queries did not start within the timeout
+    #
+    def wait_for_queries(count, timeout: 5)
+      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
+      @lock.synchronize do
+        while @queries_started < count
+          remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+          return false if remaining <= 0
+          @changed.wait(@lock, remaining)
+        end
+        true
+      end
+    end
+
+    # Stops queries from waiting. Waiting and later queries return the last result again.
+    def release
+      @lock.synchronize do
+        @released = true
+        @changed.broadcast
+      end
+    end
+
+    private def queue_result(result)
+      @lock.synchronize do
+        @results << result
+        @changed.broadcast
+      end
+    end
+  end
+
   class SimpleObserver
     def initialize(fn)
       @fn = fn