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 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
with self._limiter.lock:
    now = TimeClock().now()
    self._bucket.leak(now)
    items_to_add = self._calls_left(now) - available_calls
    if items_to_add > 0:
        self._bucket.put(RateItem(name="dummy", timestamp=now, weight=items_to_add))

_calls_left() is the new helper: for each configured rate it counts the items inside that rate's own interval (via pyrate_limiter.utils.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.

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. Connectors that already declare an api_budget (e.g. source-klaviyo) pick this up with no manifest change once the CDK version ships in source-declarative-manifest.

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

• APIs that send neither header: available_calls is None → early return, unchanged.
• 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.
• APIs that send both headers: previously ignored, now throttled. This is the intended fix; it only ever slows a connector down toward what the API says is left.
• The derived wait is still bounded by the configured rate intervals, never by a header value, so no header-driven local wait can exceed the connector's own window (the ≤600s invariant the originating issue requires). Asserted in test_update_respects_the_most_constraining_rate.

Not a breaking change under the connector breaking-change checklist: no schema, spec, state, or stream changes — only rate-limiting timing. 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.

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 — six new tests:

• 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 APIs unaffected.
• test_update_respects_the_most_constraining_rate — burst (3/s) + steady (60/m); also asserts the wait stays within the configured window.
• test_update_available_calls_zero_fills_bucket — covers the weight=True bug above.
• TestHttpAPIBudget::test_update_from_response — end-to-end through HttpAPIBudget.update_from_response with RateLimit-Remaining / -Reset / -Limit, proving a declared api_budget lowers its allowance from headers alone.

poetry run pytest unit_tests/sources/streams/test_call_rate.py → 44 passed. ruff check/ruff format clean; mypy airbyte_cdk/sources/streams/call_rate.py 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

[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 369 tests  +6   4 358 ✅ +6   10m 45s ⏱️ + 1m 58s
    1 suites ±0      11 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 61c7434. ± Comparison against base commit 4855c2d.

[github-actions]

PyTest Results (Full)

4 372 tests  +6   4 360 ✅ +6   13m 55s ⏱️ -5s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 61c7434. ± Comparison against base commit 4855c2d.