Merge pull request #48 from PostHog/jakob/viaduck-round-robin-reads

main: cap per-group chunk reads per poll cycle (round-robin fairness)

Author: jghoman

tests/unit/test_main.py

@@ -1030,6 +1030,62 @@ def fake_read(src_table, *, after_snapshot, end_snapshot, filter_expr=None):
     )
 
 
+def test_poll_cycle_lagging_group_cannot_monopolize_reads():
+    """The per-cycle chunk cap turns lagging-first priority into round-robin:
+    a deeply-lagging group reads at most _MAX_CHUNKS_PER_GROUP_PER_CYCLE
+    chunks per cycle, then the healthy group gets its turn. Without the cap
+    (the first sort-only fix), the lagging group's chunk loop ran to head —
+    or forever when its flushes kept failing and its position kept resetting
+    — and the healthy destination's cursor froze (observed on portola:
+    team-2 starved for 1.5h while team-50689 relitigated the same range).
+    """
+    from viaduck.main import _MAX_CHUNKS_PER_GROUP_PER_CYCLE
+
+    # dest-lagging is 10,000 snapshots behind; dest-healthy is 100 behind.
+    # Uncapped, dest-lagging's group would issue 100 chunk reads before
+    # dest-healthy saw a single one.
+    delivery = _make_delivery({"dest-lagging": 0, "dest-healthy": 9_900})
+    router = MagicMock()
+    cfg = _make_cfg([("dest-lagging", "lag"), ("dest-healthy", "ok")])
+    cfg.poll.cdc_chunk_snapshots = 100
+
+    read_calls: list[tuple[int, int]] = []
+
+    def fake_read(src_table, *, after_snapshot, end_snapshot, filter_expr=None):
+        read_calls.append((after_snapshot, end_snapshot))
+        return pa.table({"company": pa.array([], type=pa.string())})
+
+    router.build_filter_expr.return_value = None
+
+    with (
+        patch("viaduck.main.source.current_snapshot_id", return_value=10_000),
+        patch("viaduck.main.source.read_cdc", side_effect=fake_read),
+    ):
+        _poll_cycle(
+            MagicMock(),
+            delivery,
+            MagicMock(),
+            router,
+            cfg,
+            ["dest-lagging", "dest-healthy"],
+            {"lag": "dest-lagging", "ok": "dest-healthy"},
+            key_columns=[],
+            mode="append_only",
+        )
+
+    # The lagging group read exactly the cap, no more.
+    lagging_reads = [c for c in read_calls if c[0] < 9_900]
+    assert len(lagging_reads) == _MAX_CHUNKS_PER_GROUP_PER_CYCLE, (
+        f"lagging group should read exactly the per-cycle cap, got {len(lagging_reads)}: {lagging_reads}"
+    )
+    # The healthy group STILL got its read this cycle — the load-bearing
+    # assertion. Under the uncapped loop this list is empty.
+    healthy_reads = [c for c in read_calls if c[0] >= 9_900]
+    assert healthy_reads == [(9_900, 10_000)], f"healthy group must read within the same cycle, got {healthy_reads}"
+    # Order: lagging first (priority preserved), healthy after.
+    assert read_calls[0][0] == 0 and read_calls[-1][0] == 9_900
+
+
 def test_poll_cycle_snapshot_at_zero():
     """Source snapshot 0 with positions at 0: nothing to read, triggers run."""
     delivery = _make_delivery({"dest-1": 0})

viaduck/main.py

@@ -69,6 +69,15 @@
 
 log = logging.getLogger(__name__)
 
+# Per-cycle read budget per cursor group. Bounds how many CDC chunks one
+# group may read before the poll cycle moves to the next group — the fairness
+# valve that keeps a deeply-lagging (or flush-failing, position-resetting)
+# destination from monopolizing the poll thread while healthy peers starve.
+# 4 chunks × cdc_chunk_snapshots=50 = 200 snapshots per group per cycle,
+# ~7× the source's per-cycle arrival rate — groups drain lag while every
+# other group still gets a turn each cycle.
+_MAX_CHUNKS_PER_GROUP_PER_CYCLE = 4
+
 
 def _start_progress_heartbeat(
     label: str,
@@ -806,14 +815,22 @@ def _poll_cycle(src_table, delivery, dest_pool, router, cfg, assigned_ids, rv_to
             epochs = {d: epoch for d, (_pos, epoch) in plan.items()}
             groups = _group_by_cursor(positions, assigned_ids)
 
-            # Iterate lowest cursor first: the most-lagging group reads before
-            # a caught-up peer's turn can trip the buffer watermark. Under the
-            # prior insertion-order iteration, a destination whose cursor was
-            # further behind was silently starved whenever the first-iterated
-            # group filled the buffer mid-chunk — the outer loop moved to the
-            # next group and immediately hit `should_pause_reads()`, breaking
-            # out without reading anything. Sorting flips the priority so the
-            # laggiest destination gets first shot at each poll's read budget.
+            # Iterate lowest cursor first — the most-lagging group gets the
+            # first turn each cycle — but cap the chunks each group may read
+            # per cycle so a deeply-lagging group cannot monopolize the poll
+            # thread. Both halves are load-bearing:
+            #   - Insertion-order iteration (pre-sort) starved a lagging
+            #     destination whose config index was after a caught-up peer:
+            #     the peer's read filled the buffer and the lagging group's
+            #     first watermark check bounced it out with zero reads.
+            #   - Sorted-but-uncapped iteration (the first fix) starved the
+            #     HEALTHY destination instead: the lagging group's chunk loop
+            #     ran to head — or forever, when its flushes kept failing and
+            #     the position kept resetting — so the caught-up group never
+            #     got a read. Observed on portola: team-2's cursor frozen for
+            #     1.5h while team-50689 relitigated the same range.
+            # The cap turns strict priority into a round-robin with a
+            # lagging-first bias: every group makes progress every cycle.
             for start_snap, dest_ids in sorted(groups.items()):
                 if start_snap >= current_id:
                     continue  # already read through the current snapshot
@@ -828,7 +845,9 @@ def _poll_cycle(src_table, delivery, dest_pool, router, cfg, assigned_ids, rv_to
                 chunk_size = cfg.poll.cdc_chunk_snapshots
                 chunk_start = start_snap
                 routing_error = False
-                while chunk_start < current_id:
+                chunks_this_group = 0
+                while chunk_start < current_id and chunks_this_group < _MAX_CHUNKS_PER_GROUP_PER_CYCLE:
+                    chunks_this_group += 1
                     if delivery.should_pause_reads():
                         _log_watermark_paused("mid-chunk")
                         break