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 bounded

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 — a hardcoded timedelta(minutes=10) on the policy class — 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 bound keeps behaving exactly as it does today rather than parking a worker for the whole window. The constructor signature is unchanged, so no call site or manifest is affected.

Relationship to #1154

#1154 landed on main while this was open and fixes a subset of the same ground: the weight=True bug, and a _get_available_calls() helper — but only inside the old available_calls == 0 and call_reset_ts is None branch, leaving the both-headers TODO (the actual issue here) in place. update() here supersedes that branch, so the merge (36dd8ce8) drops _get_available_calls() in favour of _calls_left(), which is the same most-constraining-rate computation plus the wait bound. Two of #1154's tests asserted behaviour this PR deliberately changes and were adapted rather than deleted:

• test_update_positive_available_calls_is_noop → test_update_does_not_raise_allowance. A positive remaining count is no longer a no-op (that is the fix); what still holds is that the API reporting more headroom than the configured rates allow changes nothing. Its second half — update(available_calls=0, call_reset_ts=<ts>) leaving the bucket untouched — is exactly the bug, and is now covered with the opposite expectation by test_update_available_calls_with_reset_ts.
• test_update_available_calls_zero_multiple_rates → test_update_skipped_when_over_cap_rate_blocks_fill, same rates (10/PT10M, 3/PT10S, 2/PT1H). See the limitation below.

Known limitation: an over-cap rate can block the fill entirely

A rate longer than MAX_HEADER_DRIVEN_WAIT is excluded from _calls_left(), but InMemoryBucket.put() still validates the dummy weight against every rate. So on a policy that mixes short rates with, say, 2 per PT1H, a fill sized for the short rates is rejected outright and the bucket is left unchanged (with the could not adjust available calls warning). That is the intended trade: the only way to make the fill land would be to clamp it to the long rate's headroom, which fills that rate to its limit and parks the worker for the full hour — precisely what the bound exists to prevent. Such a policy keeps pacing to its configured rates, as it does today.

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 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 11 per-endpoint MovingWindowCallRatePolicy entries and status_codes_for_ratelimit_hit: [429], using the default ratelimit-remaining / ratelimit-reset header names, which match Klaviyo's headers (lookup is case-insensitive). That YAML is instantiated into the classes in airbyte_cdk/sources/streams/call_rate.py by ModelToComponentFactory.create_moving_window_call_rate_policy, so the connector already executes the code changed here. 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, no manifest change, and no new declarative surface — declarative_component_schema.yaml and its generated model are untouched.

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 bound. Worst-case induced wait is that policy's own window, bounded by MAX_HEADER_DRIVEN_WAIT (10 min). Policies whose only window exceeds the bound 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_skipped_when_over_cap_rate_blocks_fill — the mixed short/PT1H policy from fix: exhaust MovingWindowCallRatePolicy bucket when API reports 0 available calls #1154; the fill is rejected and the bucket stays as configured.
• test_update_does_not_raise_allowance — available_calls=50 against 10/PT1M with 3 calls used is a no-op.
• test_update_caps_to_eligible_rate — mixed 10/min + 100/15min; throttles on the eligible rate with a wait inside the bound.
• 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 bound.
• TestHttpAPIBudget::test_update_from_429_response_ignores_over_cap_policy — a 429 against an over-bound policy causes no stall.

pytest unit_tests/sources/streams/test_call_rate.py → 54 passed. Full test_model_to_component_factory.py → 232 passed. Declarative schema/model consistency tests pass. ruff check/ruff format clean; mypy --config-file mypy.ini airbyte_cdk clean.

Follow-ups

• Making the bound configurable per connector was implemented and then reverted at pnilan's request, to keep this PR to the header-adherence fix. The declarative field (max_header_driven_wait on MovingWindowCallRatePolicy) is in 0a6c8e66's parent if it's wanted later.
• Applying the remaining count to the coarsest configured rate rather than the most constraining one (see the limitation above).
• HttpAPIBudget.get_reset_ts_from_response() parsing the reset header as an epoch timestamp when several APIs document it as seconds remaining, which affects FixedWindowCallRatePolicy.

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.

[github-actions]

PyTest Results (Fast)

4 685 tests  +11   4 674 ✅ +12   11m 7s ⏱️ + 1m 15s
    1 suites ± 0      11 💤  -  1 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36dd8ce. ± Comparison against base commit d8d8e6f.

This pull request removes 2 and adds 13 tests. Note that renamed tests count towards both.

unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_zero_multiple_rates
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_positive_available_calls_is_noop

unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_429_response
unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_429_response_ignores_over_cap_policy
unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_response
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_with_reset_ts
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_zero_fills_bucket
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_caps_to_eligible_rate
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_clamps_negative_available_calls
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_does_not_raise_allowance
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_ignores_rates_over_header_wait_cap
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_is_noop_without_available_calls
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 688 tests  +11   4 676 ✅ +11   9m 57s ⏱️ - 3m 57s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36dd8ce. ± Comparison against base commit d8d8e6f.

This pull request removes 2 and adds 13 tests. Note that renamed tests count towards both.

unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_zero_multiple_rates
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_positive_available_calls_is_noop

unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_429_response
unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_429_response_ignores_over_cap_policy
unit_tests.sources.streams.test_call_rate.TestHttpAPIBudget ‑ test_update_from_response
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_with_reset_ts
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_available_calls_zero_fills_bucket
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_caps_to_eligible_rate
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_clamps_negative_available_calls
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_does_not_raise_allowance
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_ignores_rates_over_header_wait_cap
unit_tests.sources.streams.test_call_rate.TestMovingWindowCallRatePolicy ‑ test_update_is_noop_without_available_calls
…

♻️ This comment has been updated with latest results.

[Patrick Nilan (pnilan)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/34370352578

[devin-ai-integration]

Prerelease from 155da840 published successfully (run) — PyPI and SDM both green, manifest-server and Builder-bump steps skipped as usual for a prerelease:

• airbyte-cdk==7.28.2.post4.dev34370352578 (PyPI)
• airbyte/source-declarative-manifest:7.28.2.post4.dev34370352578 (sha256:8b54068febbed74c769af9a3eb256bcada77ab3ad803f90d74d94c53572dd7af)

If you're pointing source-klaviyo at that image, what this build changes is only pacing, so the signal is in timing rather than in records: expect fewer 429s and request gaps appearing after a low RateLimit-Remaining, with identical output. Worth knowing about two limits while you read the results — the header update currently applies to the burst rate rather than Klaviyo's PT1M rate (the open item above), so on a healthy sync it may barely engage; and the 10-minute default cap means any policy configured with a window longer than that is left alone.

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

[Patrick Nilan (pnilan)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/34371430943

[Patrick Nilan (pnilan)]

/ai-resolve-conflicts

[devin-ai-integration]

☑️ Resolved in 36dd8ce8 — merged main into the branch; no conflict markers remain and CI is running.

The conflict was with #1154, which landed the same weight=True fix and a _get_available_calls() helper, but only inside the old available_calls == 0 and call_reset_ts is None branch — the both-headers TODO that this PR is about was left in place. update() here supersedes that branch, so _get_available_calls() is dropped in favour of _calls_left() (same most-constraining-rate computation, plus the 10-minute wait bound).

Two of #1154's tests asserted behaviour this PR deliberately changes, so I adapted rather than deleted them:

• test_update_positive_available_calls_is_noop → test_update_does_not_raise_allowance. A positive remaining count is no longer a no-op — that is the fix. What still holds is that the API reporting more headroom than the configured rates allow changes nothing. Its second half (update(available_calls=0, call_reset_ts=<ts>) leaving the bucket untouched) is exactly the bug, and is covered with the opposite expectation by test_update_available_calls_with_reset_ts.
• test_update_available_calls_zero_multiple_rates → test_update_skipped_when_over_cap_rate_blocks_fill, same rates (10/PT10M, 3/PT10S, 2/PT1H), but asserting the bucket is left untouched rather than filled to the 2/PT1H minimum.

That last one is a real behaviour difference worth your eyes. _calls_left() excludes the 1-hour rate because it is over the cap, but InMemoryBucket.put() still validates the dummy weight against every rate — so a fill sized for the short rates is rejected outright and the policy keeps pacing as configured. The alternative would be clamping the fill to the long rate's headroom, which fills that rate to its limit and parks the worker for the full hour — precisely what the cap exists to prevent. I took the no-op as the safer trade; it is documented in the PR body under "Known limitation: an over-cap rate can block the fill entirely". Happy to revisit if you'd rather the fill land.

Local run after the merge: 54 call-rate tests, 232 factory tests, ruff and mypy clean.

[Patrick Nilan (pnilan)]

Closing -- need to scope a solution out instead of having a reactive devin fix