fix(call-rate): honor RateLimit-Remaining when a reset header is present

[devin-ai-integration]

Summary

MovingWindowCallRatePolicy.update() silently dropped the rate-limit state reported by the API whenever the response carried both a remaining-calls header and a reset header — the exact shape Klaviyo (and many other APIs) return on every non-429 response. The both-present branch was a commented-out TODO, so every connector declaring an api_budget with moving-window policies was purely feed-forward: it throttled to manifest constants and only learned about upstream state from a 429 after the fact.

The single-header path was also effectively inert. items_to_add was assigned a comparison, not a count:

items_to_add = self._bucket.count() < self._bucket.rates[0].limit   # a bool
if items_to_add > 0:
    self._bucket.put(RateItem(..., weight=items_to_add))            # weight=True -> 1

so available_calls == 0 — which is what a 429 produces, see below — added a single dummy call instead of filling the bucket.

After this change, update() reacts to available_calls regardless of call_reset_ts, and fills the bucket until what it still allows equals what the API reports:

if available_calls is None:
    return
available_calls = max(0, available_calls)          # a negative header must not fail open
with self._limiter.lock:
    calls_left = self._calls_left(TimeClock().now())
    if calls_left is None:                         # no rate short enough to fill safely
        return
    items_to_add = calls_left - available_calls
    if items_to_add > 0 and not self._bucket.put(RateItem(...,  weight=items_to_add)):
        logger.warning(...)                        # put() inserts nothing if any rate rejects

_calls_left() is the new helper: for each configured rate it counts the items inside that rate's own interval (via binary_search, the same primitive InMemoryBucket.put uses) and returns the most constraining remainder. That matters for the burst+steady rate pairs connectors actually declare — comparing against rates[0].limit alone (the old code) would let a tighter long-window rate go unenforced, and bucket.count() counts the whole max-interval window rather than the rate's own.

The 429 path, and why the fill is capped

HttpAPIBudget.get_calls_left_from_response() returns 0 for any status in status_codes_for_ratelimit_hit (default [429]) even when the response carries no remaining header at all. So the fill path is reached by every connector with a moving-window policy, not only those whose API publishes RateLimit-*. Filling to zero then makes the next acquire_call sleep for the failing rate's entire interval, with nothing bounding it — 900 s for a single 100 per PT15M policy such as source-harvest's /reports/, which would breach the ≤600 s ceiling the linked issue requires (sources heartbeat at 5400 s) and would stack on top of the Retry-After backoff that connector already performs.

_calls_left() therefore ignores any rate whose window exceeds MAX_HEADER_DRIVEN_WAIT (10 minutes) and returns None when that leaves nothing eligible, in which case update() does not touch the bucket. A policy whose only window is longer than the cap keeps behaving exactly as it does today rather than parking a worker for the whole window.

Deliberately not changed:

• call_reset_ts stays unused, and is documented as such. A moving window has no reset point, so the window length remains the configured one and the only actionable signal is the number of calls left. This also sidesteps the fact that HttpAPIBudget.get_reset_ts_from_response() parses the reset header as an absolute epoch timestamp while several APIs (Klaviyo included) document it as seconds remaining — a real latent bug for FixedWindowCallRatePolicy users, but out of scope here and best fixed with an explicit semantics option.
• No new declarative schema fields, and no manifest change. Connectors that already declare an api_budget (e.g. source-klaviyo) pick this up once the CDK version ships in source-declarative-manifest.

Known limitation: the update applies to the most constraining rate

With several rates configured, min(calls_left) is normally the burst rate, while a RateLimit-Remaining header normally describes the coarsest window the API publishes. For every source-klaviyo policy (burst/second paired with steady/minute) that means the update only starts to bite in roughly the last 5-10 % of the steady window; over the earlier part of the minute the connector paces as it does today.

Applying the header to the coarsest rate instead is not a one-line change: InMemoryBucket.put() validates the weight against every rate and inserts nothing if any one fails, so filling a 150/min window on a policy that also declares 10/s is rejected outright. Doing it correctly requires dummy items carrying spread-out past timestamps merged into bucket.items, plus an explicit header→rate mapping to say which window the header describes. Both are left as follow-ups; this PR's scope is making the feedback arrive at all and making the 429 path safe.

Declarative-First Evaluation

The originating issue is on source-klaviyo, a manifest-only connector, so a custom Python component was evaluated and rejected. source-klaviyo already declares the right thing in its manifest — an HTTPAPIBudget with per-endpoint MovingWindowCallRatePolicy burst/steady rates, using the default ratelimit-remaining / ratelimit-reset header names, which match Klaviyo's headers (lookup is case-insensitive). None of the declarative building blocks (RecordFilter, AddFields/RemoveFields, DatetimeBasedCursor, DefaultPaginator, SubstreamPartitionRouter, requester error handlers, transformations, $ref overrides) can affect inter-request pacing — that is entirely the api_budget policy's job. The gap was therefore not in the manifest or in any connector-side component, but in the shared CDK policy the manifest already points at, so the fix belongs here. Net result: no connector custom component, and no manifest change either.

Behavior compatibility

• Non-429 responses from APIs that send neither header: available_calls is None → early return, unchanged.
• 429 responses: available_calls is 0, so the bucket is filled for every moving-window policy whose window is within the cap. Worst-case induced wait is that policy's own window, bounded by MAX_HEADER_DRIVEN_WAIT. Policies whose only window exceeds the cap are untouched.
• APIs that report more available calls than the configured rates allow: no-op. Updates can only lower the local allowance, so a manifest rate stricter than the API's own limit is still respected.
• A negative remaining header is clamped to 0 rather than overflowing every rate's headroom and silently inserting nothing.

Not a breaking change under the connector breaking-change checklist: no schema, spec, state, or emitted-data change — only request pacing. No connector version bump here either; this is CDK-only, and source-klaviyo picks it up when its source-declarative-manifest base image (pinned at 7.24.0) is bumped. It is, however, a fleet-wide pacing change: 36 connectors in airbytehq/airbyte declare a MovingWindowCallRatePolicy, and all of them reach the fill path via 429.

Reproduction

No live Klaviyo account was available (no private key), so this was not reproduced against the real API — it is verified statically and by unit tests. The gap is directly visible in the pre-change source: update() guarded on available_calls is not None and call_reset_ts is None, while HttpAPIBudget.update_from_response() passes both values whenever both headers are present, so Klaviyo's responses took the ignored path every time. test_update_available_calls_with_reset_ts reproduces that at the policy level (all 10 calls go through before the change), and TestHttpAPIBudget::test_update_from_response reproduces it end-to-end through a response object carrying Klaviyo-shaped headers.

Test Coverage

unit_tests/sources/streams/test_call_rate.py:

• test_update_available_calls_with_reset_ts — the both-headers-present combination now throttles.
• test_update_only_lowers_allowance — available_calls=50 against a limit of 10 is a no-op.
• test_update_is_noop_without_available_calls — headerless non-429 responses unaffected.
• test_update_respects_the_most_constraining_rate — 10/s + 5/min, so the long window binds from empty; asserts the failing rate is limit=5/1.0m and the wait is ~60 s. Both mutants of _calls_left (the old rates[0].limit - count() quantity, and calls_left[0] in place of min) fail this test.
• test_update_available_calls_zero_fills_bucket — covers the weight=True bug.
• test_update_ignores_rates_over_header_wait_cap — a 100 per PT15M policy is left untouched by available_calls=0.
• test_update_caps_to_eligible_rate — mixed 10/min + 100/15min; throttles on the eligible rate with a wait inside the cap.
• test_update_clamps_negative_available_calls — a negative header does not fail open.
• TestHttpAPIBudget::test_update_from_response — end-to-end with RateLimit-Remaining / -Reset / -Limit.
• TestHttpAPIBudget::test_update_from_429_response — the header-less 429 path that every fleet connector reaches; asserts the wait stays inside the cap.
• TestHttpAPIBudget::test_update_from_429_response_ignores_over_cap_policy — a 429 against an over-cap policy causes no stall.

poetry run pytest unit_tests/sources/streams/test_call_rate.py → 49 passed. Budget tests in test_model_to_component_factory.py → 4 passed. ruff check/ruff format clean; mypy --config-file mypy.ini airbyte_cdk clean.

Resolves https://github.com/airbytehq/airbyte-internal-issues/issues/17029:

• https://github.com/airbytehq/airbyte-internal-issues/issues/17029

Link to Devin session: https://app.devin.ai/sessions/878b6be04bb648618e2fe0b6834a9151
Open in Devin Desktop: https://app.devin.ai/desktop/session/878b6be04bb648618e2fe0b6834a9151?variant=devin

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1787707125-movingwindow-honor-ratelimit-headers#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1787707125-movingwindow-honor-ratelimit-headers

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[Copilot]

Pull request overview

This PR fixes MovingWindowCallRatePolicy.update() so it properly synchronizes the local moving-window bucket with API-provided rate-limit feedback (notably when both “remaining” and “reset” headers are present), and adds unit tests covering the corrected behavior and HttpAPIBudget.update_from_response() integration.

Changes:

• Update moving-window rate-limit state from available_calls regardless of call_reset_ts, instead of silently ignoring the “both headers present” case.
• Add _calls_left() helper to compute remaining allowance across multiple configured rates (most constraining rate wins).
• Add new unit tests covering update semantics (both headers present, no-op cases, most-constraining rate behavior, and the zero-available-calls bucket fill case) plus an end-to-end HttpAPIBudget header-driven update test.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
airbyte_cdk/sources/streams/call_rate.py	Fixes moving-window update logic and introduces _calls_left() to reconcile bucket state with API-reported remaining calls.
unit_tests/sources/streams/test_call_rate.py	Adds unit tests validating the corrected moving-window update behavior and HttpAPIBudget.update_from_response() integration.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[devin-ai-integration]

🚫 Not fixing — I don't think this one holds, though it's a reasonable thing to check.

InMemoryBucket.put() expands weight into individual list entries rather than storing a single weighted item:

# pyrate_limiter/buckets/in_memory_bucket.py
def put(self, item: RateItem) -> bool:
    for rate in self.rates:
        lower_bound_idx = binary_search(self.items, item.timestamp - rate.interval)
        if lower_bound_idx >= 0:
            count_existing_items = len(self.items) - lower_bound_idx   # <- counts entries
            space_available = rate.limit - count_existing_items
        ...
    self.items.extend(item.weight * [item])                            # <- N entries for weight N

So a weight=5 dummy call (or a weight=5 try_acquire) becomes 5 entries in items, and len(items) is already the weighted count. _calls_left() deliberately mirrors put()'s own accounting — same binary_search on the same list, same len(items) - lower_bound_idx — so the two cannot disagree about how much room is left. Summing item.weight instead would double-count by a factor of the weight.

This policy always uses InMemoryBucket (self._bucket = InMemoryBucket(pyrate_rates) in MovingWindowCallRatePolicy.__init__), so there's no alternative backend where the expansion wouldn't hold. If pyrate-limiter ever switched to storing weighted items compactly, put() itself would break the same way and both would need updating together.

Happy to be overruled if a reviewer sees a bucket path I've missed.

[github-actions]

PyTest Results (Fast)

4 374 tests  +11   4 362 ✅ +10   9m 28s ⏱️ +41s
    1 suites ± 0      12 💤 + 1 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 3202424. ± Comparison against base commit 4855c2d.

This pull request skips 1 test.

unit_tests.sources.declarative.test_concurrent_declarative_source ‑ test_read_with_concurrent_and_synchronous_streams

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 377 tests  +11   4 365 ✅ +11   13m 41s ⏱️ -19s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 3202424. ± Comparison against base commit 4855c2d.

♻️ This comment has been updated with latest results.

[Patrick Nilan (pnilan)]

🤖 This comment was generated by an AI Agent.

Review: fix(call-rate): honor RateLimit-Remaining when a reset header is present

Reviewed at head 61c7434. Verdict: fix before merge. Not a breaking change — no public-API, manifest-contract, state, or emitted-data change, so no major bump and no cdk-migrations.md entry. But it is a silent, fleet-wide request-pacing change that needs a call-out, and there's one case that stalls a live connector for 15 minutes.

Findings: 1 × P0 · 3 × P1 · 3 × P2 · 3 × P3.

The core diagnosis is right, and worth saying first: the both-headers branch really was a commented-out TODO, items_to_add = self._bucket.count() < self._bucket.rates[0].limit really did assign a bool (so available_calls == 0 added one dummy call instead of filling the bucket), and this genuinely belongs in the CDK rather than in a connector. _calls_left also mirrors InMemoryBucket.put's own space computation correctly, including binary_search's -1 sentinel — I walked all four edge cases against the library. Four of the six new tests fail on 61c7434^ and pass on head, so they're real regression tests.

Measurements below were produced by loading each revision of call_rate.py as a module and driving HttpAPIBudget.update_from_response with real header shapes; fleet counts by parsing every api_budget block in airbytehq/airbyte.

---

Blast radius is wider than the PR body describes

36 connectors declare MovingWindowCallRatePolicy (61 policies; 53 literal-limit, 8 Jinja-interpolated) — including hubspot, stripe, zendesk-support, intercom, mailchimp, github, gitlab, google-ads, linkedin-ads, tiktok-marketing, amazon-seller-partner, klaviyo, sendgrid, freshdesk, harvest, linear.

All 36 are affected, not just the ones whose API sends the headers. status_codes_for_ratelimit_hit defaults to [429] (declarative_component_schema.yaml:1957-1963), and get_calls_left_from_response (call_rate.py:731-732) returns 0 on a rate-limit status even with no remaining header. So the fill path runs on every 429 fleet-wide. The body's "APIs that send neither header → early return, unchanged" is true only for non-429 responses.

Not affected — reachability disproved, not assumed: source-slack (components.py:217 overrides update_from_response entirely) and source-mixpanel's Python budget (plain APIBudget, whose update_from_response is pass at call_rate.py:631-637).

---

P0 — a single 429 parks a worker for the whole configured window, uncapped

airbyte_cdk/sources/streams/call_rate.py:508

self._bucket.put(RateItem(name="dummy", timestamp=now, weight=items_to_add))

On a 429, available_calls is 0, so the bucket fills to zero and the next acquire_call blocks for the failing rate's entire interval. _do_acquire (:671) does time.sleep(time_to_wait.total_seconds()), and LimiterMixin.send (:750) calls it with block=True, timeout=None. Nothing caps that sleep.

Scenario	base 61c7434^	head 61c7434
source-harvest /reports/ — 100/PT15M, 429 carrying Retry-After	100 further calls, then 900 s	0 further calls, 900.0 s
any connector — 429, no reset header, 100/PT15M	99 further calls, then 900 s	0 further calls, 900.0 s

source-harvest is the sharpest case. Its /reports/ policy is manifest.yaml:2659-2665 (100 per PT15M) and it maps ratelimit_reset_header: Retry-After (:2672), so update() was previously a complete no-op for it. On head one 429 produces a 15-minute block — and harvest already backs off on that same header via WaitTimeFromHeader (manifest.yaml:32-33), so the 900 s lands on top of the wait Harvest itself asked for.

This misses acceptance criterion 3 of the linked issue — "No header-driven local wait can exceed 600s" — which exists because of the 5,400 s source heartbeat. call_reset_ts, the value that would have bounded the wait correctly, is exactly what the PR discards at :491.

Second consequence: items_to_add is sized by min(calls_left) (normally the shortest window) but InMemoryBucket.put appends to one shared items list, so the dummies are charged against every rate and update() has no path that removes them. One isolated 429 on klaviyo's /api/events policy consumes 350 of the 3500-per-minute budget for a full minute, for a single failed request. (Under a sustained burst both builds fill the minute window identically, so recovery is unchanged — head just substitutes dummies for wasted 429s, which is better. The harm is concentrated in isolated blips.)

Suggested fix — bound the header-driven fill so it can't induce a wait past the cap. Skipping rates whose window exceeds it leaves those connectors exactly as they behave today rather than stalling them:

MAX_HEADER_DRIVEN_WAIT_MS = 600_000

def _calls_left(self, now: int) -> int:
    items = self._bucket.items
    calls_left = []
    for rate in self._bucket.rates:
        if rate.interval > MAX_HEADER_DRIVEN_WAIT_MS:
            continue  # a full window here would sleep past the cap
        ...
    return min(calls_left) if calls_left else 0  # 0 => items_to_add <= 0 => no-op

Pairing that with a timeout on LimiterMixin.send's acquire_call would make the cap hold for the non-header path too — worth a follow-up issue either way.

Correction, so nobody chases it: an earlier draft of this review claimed source-linear reaches 2400 s. That was wrong. Because _calls_left takes min across rates, a long rate can only become failing_rate if it is the binding constraint — and linear's minute rate caps throughput below its hourly limit (40/min × 60 = 2400 < 2500 API-key; 80/min × 60 = 4800 < 5000 OAuth). Driving the limiter through 2 simulated hours of X-RateLimit-Requests-Remaining: 0 while respecting its own rates gives max wait 16.0 s, always on limit=40/1.0m. source-amplitude is safe for the same reason. The >600 s breach is confined to policies where a long window is its own binding constraint — in practice single-rate policies like harvest's, which is exactly 1 of the 53 literal-limit policies in the fleet.

---

P1 — min across rates makes the fix a near-no-op for source-klaviyo

call_rate.py:518 — return min(calls_left)

This is the right answer to "how many calls does the bucket still allow", but the wrong one for "how many calls does the window this header describes still allow". Issue #17029 says plainly that Klaviyo's headers "describe the steady (1-minute) window only" — and every Klaviyo policy pairs a small per-second burst rate with a large per-minute steady rate, so from an idle window the min is always the burst limit. items_to_add = burst_limit - available_calls is ≤ 0 for any available_calls >= burst_limit.

Sweeping RateLimit-Remaining downward against klaviyo's nine real policies (manifest.yaml:4499-4600) until the update first inserts an item:

policy	burst/s	steady/min	first engages at	share of the steady window
campaigns / profiles / templates / metrics	10	150	10	last 6.7 %
flows	3	60	3	last 5.0 %
events	350	3500	350	last 10.0 %
lists	75	700	75	last 10.7 %
lists_detailed	1	15	1	last 6.7 %
segments	75	750	75	last 10.0 %

Over the first ~90 % of Klaviyo's minute the connector paces exactly as it does today. The PR ships as a fix for source-klaviyo and delivers roughly a tenth of the intended throttling there.

Suggested fix — apply available_calls to the rate whose window the header describes, rather than to the min. call_reset_ts is the signal that identifies it. If its semantics are too ambiguous to rely on (your point about seconds-remaining vs epoch is fair), defaulting to the longest configured rate is still closer to correct than the min, since a remaining header essentially always describes the coarsest window an API publishes. A declarative field naming which rate the header maps to would settle it explicitly.

P1 — _calls_left is not covered by any test

call_rate.py:510-518, tests at unit_tests/sources/streams/test_call_rate.py:309-381

The new helper is the substance of the change, and every one of the six new tests passes unchanged when it's replaced by the old naive quantity:

Mutant applied to _calls_left	New tests that fail
replaced with rates[0].limit - bucket.count() (the old, wrong quantity)	0 / 6
calls_left[0] instead of min(calls_left)	0 / 6
control — unmutated head	0 / 6

Every multi-rate test starts from an empty bucket, where InMemoryBucket.__init__ has sorted rates ascending by interval — so rates[0] is the burst rate and is also the min. The two quantities only diverge once calls accumulate, or once a longer window carries the smaller limit.

Suggested test — invert the usual shape so the long window binds from empty:

policy = MovingWindowCallRatePolicy(
    rates=[Rate(10, timedelta(seconds=1)), Rate(5, timedelta(minutes=1))], matchers=[]
)
policy.update(available_calls=1, call_reset_ts=None)
policy.try_acquire("call", weight=1)
with pytest.raises(CallRateLimitHit) as exc:
    policy.try_acquire("call", weight=1)
assert exc.value.rate == "limit=5/1.0m"
assert exc.value.time_to_wait.total_seconds() == pytest.approx(60, 0.1)

Verified: this fails against both mutants and passes against head. It also pins a path worth pinning — the naive version computes items_to_add = 9, which put() rejects on the 5/60s rate, silently dropping the item.

P1 — no test covers the 429 fallback

test_call_rate.py:363 (new TestHttpAPIBudget)

The new class covers only a 200 carrying RateLimit-Remaining. The path that reaches all 36 connectors — available_calls = 0 from status_code in status_codes_for_ratelimit_hit at :731-732 — is never exercised, and it's the path that produces the P0. The file at head contains one status_code reference (200) and zero occurrences of 429.

Arrange a MovingWindowCallRatePolicy(rates=[Rate(100, timedelta(minutes=15))]) in an HttpAPIBudget, feed a 429 with no remaining header, and assert both that the bucket empties and that the resulting time_to_wait stays within 600 s. That test fails today, which is the point.

---

P2 — a negative RateLimit-Remaining disables throttling entirely (fail-open)

call_rate.py:501, consumed by the discarded put return at :508

get_calls_left_from_response does a bare int(header) at :729 with no clamp. A negative value makes items_to_add exceed every rate's headroom, so put returns False and inserts nothing — while the logger.debug immediately above claims the allowance was adjusted.

header	parsed	items added by update()	calls then allowed
RateLimit-Remaining: 0	0	10	0
RateLimit-Remaining: -1	-1	0	10
RateLimit-Remaining: -5	-5	0	10

New to this PR — the old code only ever put weight=1, so put couldn't realistically fail. Reachability is candidate: the path from :729 is confirmed reachable, but no connector is proven to send a negative value. Fix: items_to_add = calls_left - max(0, available_calls), and log a warning if put ever returns False rather than discarding it.

P2 — two tests don't test what they're named for

• test_update_respects_the_most_constraining_rate (:338) uses Rate(3, 1s) + Rate(60, 1m) on an empty bucket, so the 60/min rate is never the binding constraint. Its assertion time_to_wait.total_seconds() <= 60 (:352) is satisfied by an actual wait of ~1.0 s — and it's the assertion the PR body offers as proof of the ≤600 s invariant.
• TestHttpAPIBudget::test_update_from_response uses a single rate (:365), where min is trivial, so it can't exercise the burst+steady shape every real api_budget declares.
• That 1-second window is also a live flake vector: both try_acquire calls must land within 1 s of the update(). Inserting time.sleep(1.1) reproduces the failure — pytest.raises gets no CallRateLimitHit.

Across all six new tests there is one assertion in total.

P2 — the PR body's compatibility matrix omits the 429 fallback

The release note is generated from the body and title, so as written a connector developer has no way to anticipate the P0. Please add the 429 row and state the worst-case wait per configured window. The fix title type is correct — it's the body, not the type, that needs to carry the warning.

---

P3

• call_rate.py:21 — from pyrate_limiter.utils import binary_search reaches past the package root into a helper pyrate_limiter uses internally. utils.py declares no __all__; the symbol is public only via from .utils import *. Contained by the ~3.1.0 pin, but from pyrate_limiter import binary_search alongside the existing import at :18 is the more stable path and costs nothing.
• call_rate.py:516-517 — put stores item.weight copies, so _calls_left returns weight units while available_calls counts requests, and :501 subtracts one from the other. source-amplitude declares weight: 60 / weight: 120 matchers on a 108000/PT1H policy. Inert today (amplitude declares no remaining header, so it only reaches :501 via the 429 path where the mismatch cancels) but live the moment a weighted policy meets a remaining header.
• call_rate.py:499 — leak() moved from the background thread (every rates[-1].interval * 2, i.e. 2 h for amplitude) to every response, inside self._limiter.lock. Both it and binary_search are O(n) in list copies; ~769 µs per response vs ~247 µs for the pre-existing per-acquire put. Sub-millisecond, but on a lock every worker shares. _calls_left already windows per rate, so the call is hygiene rather than correctness — consider dropping it or gating it on elapsed time.

---

Notes on CI and validation

• Check: destination-motherduck is red, but it's baseline — also red on #1134, #1135, #1138 and #1139 with PermissionError(13, 'Permission denied'). Not attributable to this diff.
• Validate PR title did not evaluate — both its steps are gated on draft == false. Judged manually: fix is the right type.
• Skipped Check: source-google-drive did not run (it prints as pass).
• Test Connectors is not evidence of fleet safety here. Of the four connectors that ran, only source-intercom declares a MovingWindowCallRatePolicy — and Intercom sends X-RateLimit-Remaining, which doesn't match the default ratelimit-remaining header name, so the header path is never entered.

Worth a /prerelease plus a pinned run on source-harvest (the P0) or source-klaviyo (the originating connector) before this merges — those are the two that actually exercise the touched path.

One follow-up, not a finding here

update() now runs on nearly every response, and a successful put() clears InMemoryBucket.failing_rate; AbstractBucket.waiting() short-circuits to 0 when that's None. So a thread that has just caught BucketFullException (lock released, then re-taken at :465) can read time_to_wait == 0 and hot-retry. I reproduced this under contention — but at a comparable rate on the base commit (1104/2442 vs 924/2783), so it's pre-existing, not introduced here. Flagging it as a follow-up issue rather than a finding against this PR, since this change does widen how often update() reaches a successful put() in production.

[devin-ai-integration]

👍 On it. Thanks — this is a genuinely better review than the change deserved, and the P0 is real. I'm fixing P0, both P2 correctness items, all three P3s, and the whole test-coverage gap. I'm not taking the P1 min-across-rates change on my own judgment; that one needs your call, and I've laid out why below.

Fixing now

P0 — uncapped fill on 429. Accepted, and my PR body's ≤600 s claim was wrong: I asserted it from a test where the long rate was never the binding constraint, which is exactly your P2. Taking your suggested shape — rates whose window exceeds the cap are skipped, and if that leaves no eligible rate the update is a no-op, so source-harvest behaves exactly as it does today instead of parking for 900 s:

MAX_HEADER_DRIVEN_WAIT = timedelta(minutes=10)

def _calls_left(self, now: int) -> Optional[int]:
    calls_left = []
    for rate in self._bucket.rates:
        if rate.interval > self._max_header_driven_wait_ms:
            continue  # a full window here would sleep past the cap
        ...
    return min(calls_left) if calls_left else None   # None => no-op

P2 — negative RateLimit-Remaining fails open. Accepted; clamping with max(0, available_calls). And you're right that discarding put()'s return is new harm in this diff — the old code only ever put weight=1, so it couldn't fail. Adding a logger.warning on a rejected put so it can't silently contradict the debug line above it.

P3s. All three: from pyrate_limiter import binary_search (confirmed exported at the package root, so no reason to reach into .utils); dropping the per-response leak() — _calls_left already windows per rate and put() does its own per-rate binary_search, so it was buying nothing for an O(n) copy under the shared lock; and documenting the weight-vs-request-count mismatch at the subtraction rather than pretending the units match.

Tests. Your mutation result is the finding I'm most annoyed I handed you — six tests, zero of which pin the helper that is the substance of the change. Taking your inverted-rate test verbatim, fixing test_update_respects_the_most_constraining_rate so the long rate actually binds, adding the 429-through-HttpAPIBudget case (asserting both that the bucket empties and that the wait stays under the cap), adding the long-window-policy case that must stay untouched, and getting the 1-second windows out of the timing assertions so they can't flake.

🙋 Needs your decision — P1 (min across rates)

Your diagnosis is right and I'm not going to argue the numbers: for every Klaviyo policy the min is the burst rate, so the fix engages only in the last ~5-10 % of the steady window. Where I'd push back is on the suggested remedy — defaulting to the longest rate is not a one-line change, and done naively it makes things worse rather than partially-working.

InMemoryBucket.put() validates the incoming weight against every rate and inserts nothing if any one fails:

if space_available < item.weight:
    self.failing_rate = rate
    return False

So for klaviyo campaigns (10/s + 150/min) with RateLimit-Remaining: 30, targeting the minute rate wants 120 dummies, the 1-second rate offers 10, and put rejects the lot — a total no-op, plus a spurious failing_rate. Filling a coarse window without tripping a fine one requires the dummies to carry spread-out past timestamps, i.e. synthesising plausible request history and merging it into bucket.items in sorted order. That's implementable (evenly spacing D dummies across the target window keeps every short-window check satisfied, and it recovers allowance gradually rather than in a cliff, which is the better moving-window behaviour anyway) — but it is a materially bigger change to code that paces 36 connectors, and it needs the header→rate mapping to be explicit, which is your declarative-field suggestion.

So: (a) land this PR with the P0/P2/P3/test fixes and the tail-of-window limitation stated plainly in the body and release note, then do the spread-fill plus a declarative ratelimit_remaining_applies_to field as a follow-up; or (b) you want the spread-fill in this PR and I build it here. I lean (a) — the 429 bool bug is worth shipping on its own and (b) shouldn't ride along with it — but it's your codebase and your fleet, and if (a) means the Klaviyo issue stays effectively open then (b) is the honest answer. Tell me which and I'll execute it.

On your other two notes: agreed the Retry-After double-wait on harvest is nasty (the cap makes it moot for that policy, since no eligible rate remains); and agreed on the failing_rate/waiting() hot-retry being pre-existing — thank you for actually measuring it on the base commit instead of filing it against this diff. I'll open follow-up issues for that, for the LimiterMixin.send timeout, and for the RateLimit-Reset seconds-vs-epoch parsing this PR sidesteps, once you've picked (a) or (b). Prerelease against source-klaviyo/source-harvest before merge: agreed, and it's more meaningful after the decision above.

[devin-ai-integration]

☑️ Resolved in 3202424.

• P0 (uncapped fill after 429): _calls_left() now skips any rate whose window exceeds MAX_HEADER_DRIVEN_WAIT (10 min) and returns None when nothing is eligible; update() then leaves the bucket alone. A single 100 per PT15M policy (Harvest) is untouched by a 429 instead of parking a worker for 900 s.
• P2: negative remaining clamped with max(0, ...) (it previously failed open — put() rejected the oversized weight and inserted nothing while the debug line claimed an adjustment); put()'s return value is now checked and a rejection logs a warning.
• P3: dropped the per-response bucket.leak(now), and binary_search now comes from the package root instead of pyrate_limiter.utils.
• Tests: test_update_respects_the_most_constraining_rate reshaped to 10/s + 5/min so the long window binds from empty, asserting rate == "limit=5/1.0m" and ~60 s. I ran your two mutants — rates[0].limit - bucket.count() and calls_left[0] for min(...) — and both now fail that test (0/6 before). Added coverage for the over-cap policy, the mixed eligible/over-cap policy, the negative header, and both 429 paths (header-less 429 on an eligible policy, and on an over-cap policy). 49 passed; ruff and repo-wide mypy clean.
• The two limitations you identified — most-constraining-rate targeting, and entries-are-weight-units vs. requests — are now stated in the update() docstring and in the PR body, so the code and the description agree.

P1 (apply the header to the coarsest rate) is still your call and is not in this commit. As laid out in my earlier reply, the naive version is a regression: put() validates the weight against every rate and inserts nothing if any rejects, so targeting Klaviyo's 150/min on a policy that also declares 10/s is refused outright. Doing it properly needs spread-timestamp dummies merged into bucket.items plus a declarative ratelimit_remaining_applies_to-style mapping. My preference remains landing this as-is with the tail-of-window limitation documented and doing the spread-fill as a follow-up, but say the word and I'll build it here instead.

https://app.devin.ai/sessions/878b6be04bb648618e2fe0b6834a9151