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

The lagging-first sort (#46) fixed one starvation and created another.
Sorted-but-uncapped iteration let the most-lagging group's chunk loop
run all the way to head before the next group got a read — and when
that group's flushes kept failing (portola OCC conflicts), its position
kept resetting and the loop relitigated the same range indefinitely.
Observed in prod: team-2's cursor frozen at 16967279 for 1.5h across
three pod restarts, applied_inserts=0, while team-50689 monopolized
every poll cycle.

Cap each group at _MAX_CHUNKS_PER_GROUP_PER_CYCLE (4) chunks per poll
cycle. Sorted order still gives the laggiest group the first turn each
cycle; the cap guarantees every other group also gets a turn in the
same cycle. 4 chunks x cdc_chunk_snapshots gives each group a per-cycle
read budget of ~7x the source arrival rate at current prod settings, so
lag still drains while fairness holds.

New test seeds one group 10k snapshots behind and one 100 behind, and
asserts the healthy group reads within the SAME cycle while the lagging
group reads exactly the cap. Verified to fail against the uncapped
loop.

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