`MovingWindowCallRatePolicy.update` ignores the remaining-calls header (bool `items_to_add`), and `HttpAPIBudget` parses relative-seconds reset headers as epoch timestamps

[bazarnov]

Found while certifying source-pipedrive (airbytehq/airbyte#85812). Verified against CDK 7.28.3 (the source-declarative-manifest:7.28.3 base image) and 7.23.6; line numbers are for 7.28.3 unless marked.

Summary

Two defects in airbyte_cdk/sources/streams/call_rate.py make HTTPAPIBudget.ratelimit_remaining_header and ratelimit_reset_header nearly useless for the moving-window policy and actively harmful for the fixed-window policy when the vendor sends a relative reset value.

Code (7.28.3; identical lines in 7.23.6)

airbyte_cdk/sources/streams/call_rate.py:476-496, MovingWindowCallRatePolicy.update:

476    def update(
477        self, available_calls: Optional[int], call_reset_ts: Optional[datetime.datetime]
478    ) -> None:
...
485        if (
486            available_calls is not None and call_reset_ts is None
487        ):  # we do our best to sync buckets with API
488            if available_calls == 0:
489                with self._limiter.lock:
490                    items_to_add = self._bucket.count() < self._bucket.rates[0].limit
491                    if items_to_add > 0:
492                        now: int = TimeClock().now()  # type: ignore[no-untyped-call]
493                        self._bucket.put(RateItem(name="dummy", timestamp=now, weight=items_to_add))

• Line 488: only available_calls == 0 is handled; remaining: 3 with a local bucket that thinks 10 calls are free is ignored.
• Line 490: items_to_add is the boolean count < limit, so line 493 puts a single RateItem with weight=True (== 1) instead of filling the window. After remaining: 0 the bucket still believes limit - count - 1 calls are available.
• Line 486: any response that also carries a reset header (call_reset_ts is not None) skips the block entirely, so a vendor sending both headers gets no adjustment at all.

airbyte_cdk/sources/streams/call_rate.py:696-703, HttpAPIBudget.get_reset_ts_from_response:

696    def get_reset_ts_from_response(
697        self, response: requests.Response
698    ) -> Optional[datetime.datetime]:
699        if response.headers.get(self._ratelimit_reset_header):
700            return datetime.datetime.fromtimestamp(
701                int(response.headers[self._ratelimit_reset_header])
702            )
703        return None

The header is always treated as epoch seconds. Pipedrive documents x-ratelimit-reset as "the remaining window before the rate limit resets" (seconds, 2-second window; https://pipedrive.readme.io/docs/core-api-concepts-rate-limiting), and other APIs use the same relative convention (Retry-After style). A value of 1 becomes 1970-01-01 00:00:01.

The manifest schema (declarative_component_schema.yaml:1947-1951) only says the header "indicates when the rate limit resets", with no way to declare the format; create_http_api_budget (model_to_component_factory.py:4527-4540) passes the header names straight through.

Observed (self-check against 7.23.6 and 7.28.3)

p = MovingWindowCallRatePolicy(rates=[Rate(limit=10, interval=timedelta(minutes=1))], matchers=[])
p.update(available_calls=3, call_reset_ts=None); p._bucket.count()   # 0   (expected 7)
p.update(available_calls=0, call_reset_ts=None); p._bucket.count()   # 1   (expected 10)

p = FixedWindowCallRatePolicy(next_reset_ts=now + 2s, period=2s, call_limit=2, matchers=[])
b = HttpAPIBudget(policies=[p], ratelimit_reset_header="x-ratelimit-reset", ratelimit_remaining_header="x-ratelimit-remaining")
# response headers: x-ratelimit-reset: 1, x-ratelimit-remaining: 0
b.get_reset_ts_from_response(r)            # 1970-01-01 00:00:01
p.try_acquire(...); p.try_acquire(...)     # 2 of 2 used
p.update(available_calls=0, call_reset_ts=b.get_reset_ts_from_response(r))
p.try_acquire(...)                          # allowed: _update_current_window sees now > 1970 and zeroes _calls_num

With a relative reset header the fixed-window policy resets its counter on every response, so the budget never throttles; the moving-window policy under-counts by limit - 1 after a remaining: 0 response.

Expected behaviour

• MovingWindowCallRatePolicy.update(available_calls=n, ...) leaves exactly n calls available in the window: it should add max(0, limit - n - bucket.count()) dummy weight, for any n, not just 0, and regardless of whether a reset timestamp was also parsed.
• HttpAPIBudget should be able to read a reset header that is a relative number of seconds and convert it to an absolute timestamp (now + seconds), keeping epoch seconds as the default for backward compatibility.

Proposed fix

MovingWindowCallRatePolicy.update:

if available_calls is not None:
    with self._limiter.lock:
        limit = self._bucket.rates[0].limit
        items_to_add = limit - available_calls - self._bucket.count()
        if items_to_add > 0:
            now: int = TimeClock().now()
            self._bucket.put(RateItem(name="dummy", timestamp=now, weight=items_to_add))

(call_reset_ts stays unused by this policy, which matches the existing TODO; keep the guard available_calls is not None only.)

HttpAPIBudget:

def __init__(self, ..., ratelimit_reset_header_format: Literal["epoch_seconds", "relative_seconds"] = "epoch_seconds", **kwargs): ...

def get_reset_ts_from_response(self, response):
    raw = response.headers.get(self._ratelimit_reset_header)
    if not raw:
        return None
    value = int(float(raw))
    if self._ratelimit_reset_header_format == "relative_seconds":
        return datetime.datetime.now() + datetime.timedelta(seconds=value)
    return datetime.datetime.fromtimestamp(value)

plus the matching optional ratelimit_reset_header_format field on HTTPAPIBudget in declarative_component_schema.yaml and create_http_api_budget. A defensive alternative that needs no schema change: treat values smaller than, say, one year in seconds as relative, but an explicit option is clearer.

Unit-test outline (unit_tests/sources/streams/test_call_rate.py)

1. test_moving_window_update_fills_window_to_remaining: Rate(limit=10, 1 min), update(available_calls=3, call_reset_ts=None) >> bucket.count() == 7; three try_acquire succeed, the fourth raises CallRateLimitHit.
2. test_moving_window_update_remaining_zero_blocks_next_call: update(available_calls=0, ...) >> bucket.count() == 10 and the next try_acquire raises CallRateLimitHit (today it succeeds).
3. test_moving_window_update_with_reset_ts_still_applies_remaining: update(available_calls=0, call_reset_ts=now + 2s) >> window full.
4. test_moving_window_update_never_exceeds_limit: update(available_calls=0) twice >> bucket.count() == 10, no BucketFullException.
5. test_http_api_budget_relative_reset_header: HttpAPIBudget(ratelimit_reset_header="x-ratelimit-reset", ratelimit_reset_header_format="relative_seconds"), header x-ratelimit-reset: 2 >> get_reset_ts_from_response within now + 2s +- 1s; default format still returns fromtimestamp(int(header)).
6. test_fixed_window_relative_reset_does_not_reset_counter: fixed-window call_limit=2, headers x-ratelimit-remaining: 0, x-ratelimit-reset: 1 in relative mode >> third try_acquire raises CallRateLimitHit with time_to_wait <= 1s.
7. Manifest round-trip in unit_tests/sources/declarative/parsers/test_model_to_component_factory.py: HTTPAPIBudget with ratelimit_reset_header_format: relative_seconds builds an HttpAPIBudget with the option set.