fix(data): chunk poll_once's catch-up request under Coinbase's candle cap

keel/data/market_feed.py

@@ -7,13 +7,21 @@
 Only *closed* candles are ever fetched or persisted: the candle currently forming as of
 `now_ts` has OHLC values that are still changing, so both `backfill` and `poll_once` stop at
 the most recently closed candle boundary for each granularity.
+
+`poll_once`'s catch-up range is requested in windows of at most `MAX_CANDLES_PER_REQUEST`
+candles, not as one single request: Coinbase rejects any range over ~350 candles with
+`400 INVALID_ARGUMENT`. Without chunking, a product that ever falls behind that cap (e.g. after
+downtime) would fail every poll forever -- the request only gets bigger over time, never
+smaller -- so it can never self-heal. Chunking makes catch-up always succeed eventually,
+however far behind a product has fallen.
 """
 
 from __future__ import annotations
 
 import time
 from typing import TYPE_CHECKING
 
+from keel.data.history import MAX_CANDLES_PER_REQUEST
 from keel.types import Candle, Granularity
 
 if TYPE_CHECKING:
@@ -116,6 +124,46 @@ def backfill(
     return total_written
 
 
+def _poll_catch_up(
+    client: CoinbaseClient,
+    repo: Repository,
+    product_id: str,
+    granularity: Granularity,
+    gran_sec: int,
+    fetch_start: int,
+    latest_closed: int,
+    last_ts: int | None,
+) -> int:
+    """Fetch and upsert `[fetch_start, latest_closed]`, chunked under the venue's candle cap.
+
+    Mirrors `history._fill_forward`'s windowing idiom: page forward in windows of at most
+    `MAX_CANDLES_PER_REQUEST` candles each, upserting per window for incremental durability
+    (if a later window raises, earlier windows are already persisted). An empty window does
+    *not* stop the loop -- a mid-history hole must not block catch-up of newer candles.
+    """
+    total_written = 0
+    seen: set[int] = set()
+    window_start = fetch_start
+    while window_start <= latest_closed:
+        # `- 1` keeps each inclusive [window_start, window_end] range at exactly
+        # MAX_CANDLES_PER_REQUEST candles (an inclusive range of `+ step` would be one over).
+        window_end = min(
+            latest_closed, window_start + (MAX_CANDLES_PER_REQUEST - 1) * gran_sec
+        )
+        fetched = client.get_candles(product_id, granularity, window_start, window_end)
+        new_candles: list[Candle] = [
+            c
+            for c in fetched
+            if c.ts <= latest_closed and (last_ts is None or c.ts > last_ts) and c.ts not in seen
+        ]
+        if new_candles:
+            seen.update(c.ts for c in new_candles)
+            total_written += repo.upsert_candles(product_id, granularity, new_candles)
+        window_start = window_end + gran_sec
+
+    return total_written
+
+
 def poll_once(
     client: CoinbaseClient,
     repo: Repository,
@@ -127,7 +175,11 @@ def poll_once(
     """Fetch and upsert any newly closed candles since the last poll.
 
     For each `(product, granularity)` pair, only candles strictly newer than the latest one
-    already stored (and no later than the most recently closed candle) are written.
+    already stored (and no later than the most recently closed candle) are written. The
+    catch-up range is requested in windows of at most `MAX_CANDLES_PER_REQUEST` candles (see
+    module docstring) so a product that has fallen far behind -- even hundreds of candles, as
+    happened in production -- still fully catches up in one call to `poll_once`, instead of
+    failing outright and staying stuck.
 
     Returns the total number of new candle rows written.
     """
@@ -145,14 +197,9 @@ def poll_once(
                 continue
 
             fetch_start = last_ts + gran_sec if last_ts is not None else latest_closed
-            fetched = client.get_candles(product_id, granularity, fetch_start, latest_closed)
-            new_candles: list[Candle] = [
-                c
-                for c in fetched
-                if c.ts <= latest_closed and (last_ts is None or c.ts > last_ts)
-            ]
-            if new_candles:
-                total_written += repo.upsert_candles(product_id, granularity, new_candles)
+            total_written += _poll_catch_up(
+                client, repo, product_id, granularity, gran_sec, fetch_start, latest_closed, last_ts
+            )
 
     return total_written
 

tests/data/test_market_feed.py

@@ -12,6 +12,7 @@
 import pytest
 
 from keel.data.db import connect, migrate
+from keel.data.history import MAX_CANDLES_PER_REQUEST
 from keel.data.market_feed import backfill, is_fresh, poll_once
 from keel.data.repository import Repository
 from keel.types import Candle, Granularity
@@ -23,6 +24,13 @@
 WINDOW_START = NOW - HISTORY_DAYS * 86400  # already hour-aligned
 EXPECTED_TS = list(range(WINDOW_START, LATEST_CLOSED + 1, GRAN_SEC))  # 48 hourly candles
 
+# A stale-poll scenario mirroring the real ZEC-USD production failure: the last stored
+# candle is STALE_HOURS behind the most recently closed one, well over Coinbase's
+# ~350-candle-per-request cap, so a correct `poll_once` must page the catch-up in windows.
+STALE_HOURS = 552  # the real ZEC-USD gap
+STALE_LAST_TS = LATEST_CLOSED - STALE_HOURS * GRAN_SEC  # the stale last-stored candle
+STALE_FULL_TS = list(range(STALE_LAST_TS, LATEST_CLOSED + 1, GRAN_SEC))  # seed + all catch-up
+
 
 def _candle(ts: int, price: str = "100") -> Candle:
     p = Decimal(price)
@@ -64,6 +72,13 @@ def _full_series(product_id: str = "BTC-USD") -> dict[tuple[str, Granularity], l
     return {(product_id, Granularity.ONE_HOUR): [_candle(ts) for ts in ts_values]}
 
 
+def _stale_series(product_id: str = "BTC-USD") -> dict[tuple[str, Granularity], list[Candle]]:
+    # Contiguous hourly run from the stale last-stored candle through LATEST_CLOSED, plus
+    # the still-forming candle at ts=NOW, matching `_full_series`'s convention.
+    ts_values = STALE_FULL_TS + [NOW]
+    return {(product_id, Granularity.ONE_HOUR): [_candle(ts) for ts in ts_values]}
+
+
 # -- backfill -----------------------------------------------------------------
 
 
@@ -180,6 +195,104 @@ def test_poll_once_starts_from_scratch_when_repo_is_empty(repo):
     assert [c.ts for c in stored] == [LATEST_CLOSED]
 
 
+def test_poll_once_uses_a_single_request_for_a_small_gap(repo):
+    # repo already has everything except the last 2 closed candles -- unchanged small-gap path
+    have = EXPECTED_TS[:-2]
+    repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(ts) for ts in have])
+    client = FakeClient(_full_series())
+
+    poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW)
+
+    assert len(client.calls) == 1
+
+
+def test_poll_once_chunks_a_gap_larger_than_the_coinbase_cap(repo):
+    """A stale last candle (552h behind, like real ZEC-USD) must still be fully caught up."""
+    repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)])
+    client = FakeClient(_stale_series())
+
+    written = poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW)
+
+    assert written == STALE_HOURS
+    assert len(client.calls) > 1
+    stored = repo.get_candles("BTC-USD", Granularity.ONE_HOUR)
+    assert [c.ts for c in stored] == STALE_FULL_TS
+    stored_ts = {c.ts for c in stored}
+    assert NOW not in stored_ts  # still-forming candle never persisted
+
+
+def test_poll_once_never_requests_more_than_the_candle_cap(repo):
+    repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)])
+    client = FakeClient(_stale_series())
+
+    poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW)
+
+    assert client.calls
+    for _, _, start, end in client.calls:
+        assert start <= end
+        candle_count = (end - start) // GRAN_SEC + 1
+        assert candle_count <= MAX_CANDLES_PER_REQUEST
+
+
+def test_poll_once_chunk_windows_are_contiguous_and_non_overlapping(repo):
+    repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)])
+    client = FakeClient(_stale_series())
+
+    poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW)
+
+    # a gap this large can only tile into >1 window under the Coinbase candle cap
+    assert len(client.calls) > 1
+    fetch_start = STALE_LAST_TS + GRAN_SEC
+    assert client.calls[0][2] == fetch_start
+    assert client.calls[-1][3] == LATEST_CLOSED
+    for previous, current in zip(client.calls, client.calls[1:]):
+        assert current[2] == previous[3] + GRAN_SEC
+
+
+def test_poll_once_keeps_going_past_an_empty_window_so_a_mid_history_hole_cannot_wedge_catch_up(
+    repo,
+):
+    """A window that returns zero candles must not stop catch-up.
+
+    `_poll_catch_up`'s docstring calls this out explicitly: an empty window means a
+    mid-history hole (a stretch the venue genuinely has no candles for), not proof that
+    there's nothing further to fetch. If the loop broke on an empty window -- the way
+    `history._fill_backward` does -- catch-up would wedge at the hole forever and never
+    reach fresh data beyond it. That asymmetry with `_fill_backward` is intentional, not an
+    oversight: `_fill_backward` walks backward through history it may legitimately exhaust,
+    so stopping at "no more data" is correct there. `_poll_catch_up` walks forward toward
+    *now*, where there is always more recent data past any hole, so it must keep paging.
+
+    The fixture spans a gap wider than the Coinbase cap (`STALE_HOURS`, as in the chunking
+    tests above), so catch-up must issue more than one windowed request. The fake client is
+    configured to have real candles for *none* of the first window's range -- an empty
+    window, simulating the hole -- and only serves candles for a later window. A future
+    refactor that adds `if not fetched: break` to `_poll_catch_up` makes this test fail: the
+    loop would stop after the first (empty) window and never reach the later, real candles.
+    """
+    repo.upsert_candles("BTC-USD", Granularity.ONE_HOUR, [_candle(STALE_LAST_TS)])
+    fetch_start = STALE_LAST_TS + GRAN_SEC
+    first_window_end = min(
+        LATEST_CLOSED, fetch_start + (MAX_CANDLES_PER_REQUEST - 1) * GRAN_SEC
+    )
+    assert first_window_end < LATEST_CLOSED, "fixture must span >1 window for this test to hold"
+    later_ts = list(range(first_window_end + GRAN_SEC, LATEST_CLOSED + 1, GRAN_SEC))
+    client = FakeClient(
+        {("BTC-USD", Granularity.ONE_HOUR): [_candle(ts) for ts in later_ts + [NOW]]}
+    )
+
+    written = poll_once(client, repo, ["BTC-USD"], [Granularity.ONE_HOUR], now_ts=NOW)
+
+    # the empty first window must not have stopped the loop before a second request
+    assert len(client.calls) > 1
+    assert written == len(later_ts)
+    stored_ts = {c.ts for c in repo.get_candles("BTC-USD", Granularity.ONE_HOUR)}
+    # the pre-existing seed candle plus exactly the later window's candles -- nothing from
+    # the empty (hole) window, since it had nothing to write
+    assert stored_ts == {STALE_LAST_TS, *later_ts}
+    assert LATEST_CLOSED in stored_ts  # catch-up reached fresh data past the hole
+
+
 # -- is_fresh -------------------------------------------------------------------