airbytehq / airbytehq/airbyte-python-cdk

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

Đang mở
#1,151 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
airbyte-python-cdk/low-code bug community
Ngôn ngữ chính
Python
Star
26
Fork
53
Merge trung bình
2 ngày 6 giờ
Pull request đã merge (30 ngày)
10

Mô tả

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`:

```python
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`:

```python
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)

```python
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)
```

```python
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`:

```python
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`:

```python
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.

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Hướng nghiên cứu

Start in airbyte_cdk/sources/streams/call_rate.py, reading MovingWindowCallRatePolicy.update and HttpAPIBudget.get_reset_ts_from_response, then inspect declarative_component_schema.yaml and create_http_api_budget. Run the outlined tests in unit_tests/sources/streams/test_call_rate.py and the model-to-component factory tests. Done means remaining-call updates enforce the reported window, relative resets remain relative, epoch behavior is preserved, and manifest round-tripping passes.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
python
Lĩnh vực
api, backend, testing
Loại issue
Lỗi
Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức độ hoạt động
Sôi nổi
Độ rõ ràng
Đặc tả rõ ràng
Mức phù hợp với người mới
74/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.