fix(datasource): dedupe the pylon saturation warning (EXT-13)

The warning fired once per request let through past max_wait, and the
condition it reports is a saturation that lasts, so it put a line on
every request it described: 1080 lines for a stream 5x over a 120/min
budget for two minutes, burying the first one, which is the only one the
operator needed. It now says its piece once per endpoint per window, and
says so in the line itself, so one line does not read as one request.
A stream 25x over budget for five minutes goes from 15000 lines to 5.

RateLimiter keeps its bookings ordered by inserting into place rather
than appending and sorting. The list is sorted, so the expired bookings
are its leading run and the new one has a single position: pruning is a
prefix drop and the insert is a binary search, where the sort was
O(n log n) per request held under the mutex every endpoint shares. With
20000 bookings in the window an acquire costs 1.9us rather than 620us.
Verified behaviour-preserving against the previous scheduler over 50
randomised trials of limit, window and max_wait: identical sleeps, and
the bookings sorted at every step.

DEFAULT_MAX_WAIT said the region it smooths is "a stream a few percent
over budget", which is optimistic. The waits accumulate rather than
settling, so the bound is crossed sooner the further over budget the
stream sits, and a bypass books inside the window and brings the next
one closer. Measured against a 120/min endpoint: 1% over throttles its
first thousand requests then lets one in twelve through, 5% over gives
up after two hundred and lets half through, and past 10% over the bound
is crossed as the window fills and almost nothing is throttled. The
comment now says that.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: christophebrun-forest

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/rate_limiter.rb

@@ -26,12 +26,18 @@ class RateLimiter
     # once per attempt, on top of the backoff `retry` waits itself. None of it
     # runs under the Faraday timeout, which only covers the adapter.
     #
-    # The bound leaves a narrow operating region, and it is worth being plain
-    # about it: a wait fits under it only while the window has been full for less
-    # than this long, so what gets smoothed is a stream a few percent over
-    # budget. A burst arriving at once books its next slot a whole window out and
-    # goes straight through. Under real saturation the 429 retry is the defence,
-    # not this.
+    # The region this smooths is narrow, and it is worth being plain about it.
+    # On a stream over budget the waits accumulate rather than settling, so the
+    # bound is crossed sooner the further over it sits: measured against a
+    # 120/min endpoint, a stream 1% over budget throttles its first thousand
+    # requests and then lets roughly one in twelve through unthrottled, one 5%
+    # over gives up after two hundred and lets half through, and past ~10% over
+    # the bound is crossed as soon as the window fills — request 121 — after
+    # which almost nothing is throttled at all. A burst arriving at once books
+    # its next slot a whole window out and goes straight through from the first
+    # request past the budget. Under real saturation the 429 retry is the
+    # defence, not this; what this buys is the region just over budget, and a
+    # budget the code knows rather than one it discovers as a 429.
     DEFAULT_MAX_WAIT = 5.0
 
     attr_reader :max_wait, :window
@@ -44,16 +50,17 @@ def initialize(limits: RateLimits, window: WINDOW, max_wait: DEFAULT_MAX_WAIT, c
       @sleeper = sleeper || ->(seconds) { sleep(seconds) }
       @mutex   = Mutex.new
       @slots   = {}
+      @warned  = {}
     end
 
     # Blocks until the endpoint has room, then returns. Called once per attempt,
     # retries included: a replayed request spends the budget a first one did.
     def acquire(method, path)
       rule = @limits.for(method, path)
-      wait = @mutex.synchronize { reserve(rule) }
-      return if wait <= 0
+      wait, warn = @mutex.synchronize { reserve(rule) }
 
-      return warn_saturated(rule, wait) if wait > @max_wait
+      warn_saturated(rule, wait) if warn
+      return if wait <= 0 || wait > @max_wait
 
       @sleeper.call(wait)
     end
@@ -67,31 +74,48 @@ def acquire(method, path)
     # What is recorded is the moment the request will be made, not the moment it
     # was asked for, so concurrent callers each take a distinct slot and spread
     # out instead of all waking onto the same one.
+    #
+    # Returns the wait the caller owes and whether this is the bypass worth a log
+    # line — both settled here, the second being shared state like the first.
     def reserve(rule)
       taken = (@slots[rule.name] ||= [])
       now = @clock.call
-      taken.reject! { |at| at <= now - @window }
+      # The list is kept ordered, so the expired bookings are its leading run
+      # and the index below reads as the limit-th most recent one.
+      taken.shift(taken.bsearch_index { |at| at > now - @window } || taken.size)
 
       slot = taken.size < rule.limit ? now : taken[taken.size - rule.limit] + @window
       wait = slot - now
       # Past the bound the request goes out now, so the slot it books is now:
       # recording the one it declined to wait for would meter a request nobody
       # ever made and push the whole window further out.
-      taken << (wait > @max_wait ? now : slot)
-      # A booking of `now` lands before slots already reserved further out, and
-      # the index above only reads as the limit-th most recent booking on an
-      # ordered list: unsorted, a later caller reads the wrong one and lets a
-      # request through against a window that had a slot for it.
-      taken.sort!
+      insert(taken, wait > @max_wait ? now : slot)
+
+      [wait, wait > @max_wait && first_warning?(rule, now)]
+    end
+
+    def insert(taken, booking)
+      taken.insert(taken.bsearch_index { |at| at >= booking } || taken.size, booking)
+    end
+
+    # One line per endpoint per window. What the warning reports is a saturation
+    # that lasts, so a line per request puts one on every request it describes —
+    # a thousand of them for a couple of minutes over budget, burying the first,
+    # which is the only one the operator needed.
+    def first_warning?(rule, now)
+      last = @warned[rule.name]
+      return false if last && now - last < @window
 
-      wait
+      @warned[rule.name] = now
+      true
     end
 
     def warn_saturated(rule, wait)
       ForestAdminDatasourcePylon.logger.warn(
         "[forest_admin_datasource_pylon] #{rule.name} is at its budget of #{rule.limit} requests per " \
         "#{@window.round}s; the next slot is #{wait.round(1)}s out, past the #{@max_wait.round(1)}s this waits. " \
-        'Letting the request through — Pylon may answer 429, which the client retries.'
+        'Letting the request through — Pylon may answer 429, which the client retries. Further requests over ' \
+        "this budget are let through too, and this says so once per #{@window.round}s."
       )
     end
   end

packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/rate_limiter_spec.rb

@@ -69,6 +69,7 @@ def at(time)
 
     describe 'when the wait would exceed max_wait' do
       it 'lets the request through rather than queueing behind the window' do
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
         subject = limiter(max_wait: 1.0)
         limit.times { subject.acquire(:get, '/things') }
 
@@ -88,6 +89,36 @@ def at(time)
           .to have_received(:warn).with(%r{get /things is at its budget of 2 requests per 6s})
       end
 
+      # What the warning reports is a saturation that lasts, so a line per request
+      # puts one on every request it describes: a stream 10% over a 120/min
+      # budget crosses the bound as the window fills and then bypasses almost
+      # everything, which is a thousand identical lines for a couple of minutes.
+      it 'warns once per window rather than once per request' do
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
+        subject = limiter(max_wait: 1.0)
+        limit.times { subject.acquire(:get, '/things') }
+
+        5.times { subject.acquire(:get, '/things') }
+
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).once
+      end
+
+      # Deduplicated, not silenced: a saturation still going a window later is
+      # news again, and the operator reading one line has to be able to tell the
+      # burst that passed from the stream that did not.
+      it 'warns again once a window has gone by' do
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
+        subject = limiter(max_wait: 1.0)
+        limit.times { subject.acquire(:get, '/things') }
+        subject.acquire(:get, '/things')
+
+        at(7.0)
+        limit.times { subject.acquire(:get, '/things') }
+        subject.acquire(:get, '/things')
+
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).twice
+      end
+
       # The slot it declined to wait for must not be booked: recording it would
       # meter a request nobody made and push every later slot further out, so one
       # burst past the bound would keep the window saturated indefinitely.
@@ -143,6 +174,17 @@ def at(time)
 
         expect(slept).to be_empty
       end
+
+      # Each endpoint reports its own saturation too: one silenced by another's
+      # line would have the operator narrow the wrong budget.
+      it 'warns for each endpoint that saturates' do
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
+        subject = limiter(max_wait: 1.0)
+
+        ['/things', '/others'].each { |path| (limit + 1).times { subject.acquire(:get, path) } }
+
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).twice
+      end
     end
   end