From 70408e90009f4c5f569dd7bb23d3638dc6b37a25 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Fri, 9 Oct 2026 21:52:31 +0900
Subject: [PATCH 1/6] fix(server): schedule extract requests that outwait the
 batch window

Refs #390

Starvation was cost-order, not inference time: cheaper items kept filling
max_batch_cost, so a high-cost extract request never reached the front even
though the GPU work itself is only seconds.
---
 .../sie_server/src/sie_server/core/batcher.py |  74 +++++++++--
 .../sie_server/tests/core/test_batcher.py     | 121 ++++++++++++++++++
 2 files changed, 186 insertions(+), 9 deletions(-)

diff --git a/packages/sie_server/src/sie_server/core/batcher.py b/packages/sie_server/src/sie_server/core/batcher.py
index c4a5cf4ca..8041f3058 100644
--- a/packages/sie_server/src/sie_server/core/batcher.py
+++ b/packages/sie_server/src/sie_server/core/batcher.py
@@ -9,7 +9,8 @@
 3. Accumulate while: total_cost < max_batch_cost AND
                      num_requests < max_batch_requests AND
                      elapsed < max_batch_wait_ms
-4. Sort by cost (reduces padding waste for text)
+4. Sort by cost (reduces padding waste for text). A request that has waited
+   max_batch_wait_ms is taken before cheaper items
 5. Split into optimal sub-batches respecting max_batch_cost
 6. Return one sub-batch per get_batch() call
 
@@ -25,8 +26,13 @@
 a long item's batch, and waits for at most one flagged item between two
 batches of its lane. A flagged item with k flagged items ahead of it waits
 for at most k + 1 batches of batchable items, however many keep arriving.
-When nothing pending is flagged, batch formation is exactly the cost-sorted
-greedy packing above.
+When nothing pending is flagged, requests younger than ``max_batch_wait_ms``
+are packed by that cost-sorted greedy algorithm. Once the oldest pending
+request has waited at least that long, it is taken before cheaper work fills
+the cost budget (#390); anything that does not fit in the remaining budget
+stays queued, and one request larger than ``max_batch_cost`` is still taken
+alone. Cheaper arrivals cannot skip that older request for longer than
+``max_batch_wait_ms`` plus one batch.
 
 Cost semantics vary by modality (modality-native units):
 - Text: cost = token count
@@ -322,11 +328,18 @@ async def submit_many(self, items: list[tuple[I, T]]) -> None:
 
     def _append_item(self, item: I, metadata: T) -> None:
         """Append item to pending list (caller must hold lock)."""
-        request = PendingRequest(item=item, metadata=metadata, runs_alone=getattr(item, "runs_alone", False) is True)
+        # Call time.monotonic at enqueue. The dataclass default holds the
+        # function captured at import, so a patched clock would not apply.
+        now = time.monotonic()
+        request = PendingRequest(
+            item=item,
+            metadata=metadata,
+            arrival_time=now,
+            runs_alone=getattr(item, "runs_alone", False) is True,
+        )
         self._pending.append(request)
         self._total_cost += item.cost
 
-        now = time.monotonic()
         self._last_submit_time = now
 
         # Track first request time for timeout
@@ -437,6 +450,31 @@ def _order_runs_alone_lane(self) -> None:
         if alone_turn:
             self._pending.insert(0, self._pending.pop(head_index))
 
+    def _promote_overdue_request(self) -> bool:
+        """Move the oldest pending request to the front once it has outwaited the batch window.
+
+        Cost-ascending order is unchanged while that request is younger than
+        ``max_batch_wait_ms``. After that, cheaper work must not keep filling
+        the cost budget ahead of it (#390). The caller has already sorted
+        ``_pending`` by cost. The existing take-loop then fills any remaining
+        budget, or takes this request alone when nothing else fits — including
+        a single request larger than ``max_batch_cost``.
+
+        Returns:
+            Whether the oldest request was moved to the front.
+        """
+        if len(self._pending) < 2:
+            return False
+        oldest = min(self._pending, key=lambda request: request.arrival_time)
+        waited_ms = (time.monotonic() - oldest.arrival_time) * 1000
+        if waited_ms < self._config.max_batch_wait_ms:
+            return False
+        oldest_index = next(index for index, request in enumerate(self._pending) if request is oldest)
+        if oldest_index == 0:
+            return False
+        self._pending.insert(0, self._pending.pop(oldest_index))
+        return True
+
     def _extract_batch(
         self,
         max_items: int | None = None,
@@ -446,11 +484,16 @@ def _extract_batch(
 
         Algorithm:
         1. Sort pending requests by cost (ascending)
-        2. Take requests until max_batch_cost would be exceeded
-        3. Keep remaining requests for next get_batch() call
+        2. If the oldest request has waited at least max_batch_wait_ms, move
+           it to the front so cheaper work cannot skip it (#390)
+        3. Take requests until max_batch_cost would be exceeded
+        4. Keep remaining requests for next get_batch() call
 
         This minimizes padding waste by grouping similar-cost sequences.
-        When a pending request is flagged ``runs_alone``, the order comes from
+        Young requests stay in cost order. An overdue oldest request is taken
+        before that fill; if it does not fit beside the cheaper items, the
+        take-loop keeps it alone instead of leaving it queued. When a pending
+        request is flagged ``runs_alone``, the order comes from
         ``_order_runs_alone_lane`` instead, and a flagged request is always
         taken on its own.
 
@@ -472,11 +515,15 @@ def _extract_batch(
         pending_before = len(self._pending)
         cost_before = self._total_cost
 
+        promoted_overdue = False
         if any(request.runs_alone for request in self._pending):
             self._order_runs_alone_lane()
         else:
-            # Sort pending by cost (ascending) for optimal batching
+            # Sort pending by cost (ascending) for optimal batching.
+            # Promote only after that sort, and only once the oldest request
+            # has outwaited the batch window (#390). runs_alone keeps its lane.
             self._pending.sort(key=lambda r: r.item.cost)
+            promoted_overdue = self._promote_overdue_request()
 
         # Greedily take items until we exceed max_batch_cost
         batch_items: list[I] = []
@@ -516,6 +563,15 @@ def _extract_batch(
                 hit_request_limit = True
                 break
 
+        # The overdue request was selected first so it could not be left behind
+        # a cheaper fill. Emit cost order so this batch still pads like any
+        # other cost-sorted sub-batch.
+        if promoted_overdue and take_count > 1:
+            pairs = list(zip(batch_items, batch_metadata, strict=True))
+            pairs.sort(key=lambda pair: pair[0].cost)
+            batch_items = [pair[0] for pair in pairs]
+            batch_metadata = [pair[1] for pair in pairs]
+
         lane = "alone" if take_count > 0 and self._pending[0].runs_alone else "batched"
         if take_count > 0:
             # Whose turn is next when both lanes are pending.
diff --git a/packages/sie_server/tests/core/test_batcher.py b/packages/sie_server/tests/core/test_batcher.py
index 87a418e31..80cb7b1ba 100644
--- a/packages/sie_server/tests/core/test_batcher.py
+++ b/packages/sie_server/tests/core/test_batcher.py
@@ -1274,3 +1274,124 @@ def extract() -> None:
         assert sorted(served) == sorted(submitted)
         assert not flagged_waits
         assert batcher.pending_cost == 0
+
+
+class _MonotonicClock:
+    """Stand-in for ``time.monotonic``. Advance it; do not sleep."""
+
+    def __init__(self, now: float) -> None:
+        self.now = now
+
+    def monotonic(self) -> float:
+        return self.now
+
+    def advance_ms(self, delay_ms: float) -> None:
+        self.now += delay_ms / 1000
+
+
+class TestOverdueRequestScheduling:
+    """A high-cost request is not starved by cheaper arrivals (#390).
+
+    Cost order still fills young batches. Once the oldest request has waited
+    ``max_batch_wait_ms``, the next batch takes it before cheaper work.
+    """
+
+    def _clock(self, monkeypatch: pytest.MonkeyPatch) -> _MonotonicClock:
+        clock = _MonotonicClock(10_000.0)
+        monkeypatch.setattr(time, "monotonic", clock.monotonic)
+        return clock
+
+    @pytest.mark.asyncio
+    async def test_young_cost_order_holds_then_overdue_request_is_taken(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Cheap fills skip an older expensive item only while it is still young.
+
+        Three immediate cost-ordered batches run before ``max_batch_wait_ms``.
+        After the wait elapses, the next batch includes that item and still
+        uses any remaining budget for a cheaper one that fits beside it.
+        """
+        real_monotonic = time.monotonic
+        clock = self._clock(monkeypatch)
+        wait_ms = 100.0
+        batcher: BatchFormer[TextPreparedItem, str] = BatchFormer(
+            BatchConfig(
+                max_batch_cost=10,
+                max_batch_requests=32,
+                max_batch_wait_ms=wait_ms,
+                coalesce_ms=1_000,
+                coalesce_ratio=1.0,
+            )
+        )
+        await batcher.submit(make_text_item(list(range(6))), "expensive")
+
+        for round_index in range(3):
+            clock.advance_ms(10)  # 10ms, 20ms, 30ms — still inside the 100ms window
+            for item_index in range(4):
+                await batcher.submit(make_text_item([1, 2, 3], item_index), f"cheap-{round_index}-{item_index}")
+            started = real_monotonic()
+            batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
+            assert real_monotonic() - started < 0.05
+            assert "expensive" not in batch.metadata
+            assert [item.cost for item in batch.items] == [3, 3, 3]
+            assert batch.total_cost == 9
+
+        assert any(request.metadata == "expensive" for request in batcher._pending)
+        assert any(str(request.metadata).startswith("cheap") for request in batcher._pending)
+
+        # More cheap work arrives after the window. It must not skip the old item.
+        clock.advance_ms(wait_ms)
+        for item_index in range(4):
+            await batcher.submit(make_text_item([1, 2, 3], item_index), f"late-{item_index}")
+
+        started = real_monotonic()
+        batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
+        assert real_monotonic() - started < 0.05
+        assert "expensive" in batch.metadata
+        assert [item.cost for item in batch.items] == [3, 6]
+        assert batch.total_cost == 9
+        assert any(str(request.metadata).startswith(("cheap", "late")) for request in batcher._pending)
+
+    @pytest.mark.asyncio
+    async def test_overdue_request_larger_than_the_cost_budget_is_taken_alone(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """An overdue request that does not fit beside cheaper items is not left queued.
+
+        The take-loop already returns one oversized item. Putting the overdue
+        request first uses that path instead of filling another cheap batch.
+        """
+        real_monotonic = time.monotonic
+        clock = self._clock(monkeypatch)
+        wait_ms = 100.0
+        batcher: BatchFormer[TextPreparedItem, str] = BatchFormer(
+            BatchConfig(
+                max_batch_cost=10,
+                max_batch_requests=32,
+                max_batch_wait_ms=wait_ms,
+                coalesce_ms=1_000,
+                coalesce_ratio=1.0,
+            )
+        )
+        await batcher.submit(make_text_item(list(range(30))), "expensive")
+
+        for round_index in range(3):
+            clock.advance_ms(10)
+            for item_index in range(3):
+                await batcher.submit(make_text_item([1, 2, 3, 4], item_index), f"cheap-{round_index}-{item_index}")
+            batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
+            assert "expensive" not in batch.metadata
+            assert [item.cost for item in batch.items] == [4, 4]
+            assert batch.total_cost == 8
+
+        assert any(request.metadata == "expensive" for request in batcher._pending)
+
+        clock.advance_ms(wait_ms)
+        for item_index in range(3):
+            await batcher.submit(make_text_item([1, 2, 3, 4], item_index), f"late-{item_index}")
+
+        started = real_monotonic()
+        batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
+        assert real_monotonic() - started < 0.05
+        assert batch.metadata == ["expensive"]
+        assert batch.total_cost == 30
+        assert batcher.pending_count > 0
+        assert all(request.metadata != "expensive" for request in batcher._pending)

From 3e62f0f301ed9a6cdeb2f069be82d60f2232aac3 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Fri, 9 Oct 2026 23:08:57 +0900
Subject: [PATCH 2/6] Revert "fix(server): schedule extract requests that
 outwait the batch window"

The 15ms promotion is reverted because it reschedules encode and score, not only extract.

This reverts commit 70408e90009f4c5f569dd7bb23d3638dc6b37a25.
---
 .../sie_server/src/sie_server/core/batcher.py |  74 ++---------
 .../sie_server/tests/core/test_batcher.py     | 121 ------------------
 2 files changed, 9 insertions(+), 186 deletions(-)

diff --git a/packages/sie_server/src/sie_server/core/batcher.py b/packages/sie_server/src/sie_server/core/batcher.py
index 8041f3058..c4a5cf4ca 100644
--- a/packages/sie_server/src/sie_server/core/batcher.py
+++ b/packages/sie_server/src/sie_server/core/batcher.py
@@ -9,8 +9,7 @@
 3. Accumulate while: total_cost < max_batch_cost AND
                      num_requests < max_batch_requests AND
                      elapsed < max_batch_wait_ms
-4. Sort by cost (reduces padding waste for text). A request that has waited
-   max_batch_wait_ms is taken before cheaper items
+4. Sort by cost (reduces padding waste for text)
 5. Split into optimal sub-batches respecting max_batch_cost
 6. Return one sub-batch per get_batch() call
 
@@ -26,13 +25,8 @@
 a long item's batch, and waits for at most one flagged item between two
 batches of its lane. A flagged item with k flagged items ahead of it waits
 for at most k + 1 batches of batchable items, however many keep arriving.
-When nothing pending is flagged, requests younger than ``max_batch_wait_ms``
-are packed by that cost-sorted greedy algorithm. Once the oldest pending
-request has waited at least that long, it is taken before cheaper work fills
-the cost budget (#390); anything that does not fit in the remaining budget
-stays queued, and one request larger than ``max_batch_cost`` is still taken
-alone. Cheaper arrivals cannot skip that older request for longer than
-``max_batch_wait_ms`` plus one batch.
+When nothing pending is flagged, batch formation is exactly the cost-sorted
+greedy packing above.
 
 Cost semantics vary by modality (modality-native units):
 - Text: cost = token count
@@ -328,18 +322,11 @@ async def submit_many(self, items: list[tuple[I, T]]) -> None:
 
     def _append_item(self, item: I, metadata: T) -> None:
         """Append item to pending list (caller must hold lock)."""
-        # Call time.monotonic at enqueue. The dataclass default holds the
-        # function captured at import, so a patched clock would not apply.
-        now = time.monotonic()
-        request = PendingRequest(
-            item=item,
-            metadata=metadata,
-            arrival_time=now,
-            runs_alone=getattr(item, "runs_alone", False) is True,
-        )
+        request = PendingRequest(item=item, metadata=metadata, runs_alone=getattr(item, "runs_alone", False) is True)
         self._pending.append(request)
         self._total_cost += item.cost
 
+        now = time.monotonic()
         self._last_submit_time = now
 
         # Track first request time for timeout
@@ -450,31 +437,6 @@ def _order_runs_alone_lane(self) -> None:
         if alone_turn:
             self._pending.insert(0, self._pending.pop(head_index))
 
-    def _promote_overdue_request(self) -> bool:
-        """Move the oldest pending request to the front once it has outwaited the batch window.
-
-        Cost-ascending order is unchanged while that request is younger than
-        ``max_batch_wait_ms``. After that, cheaper work must not keep filling
-        the cost budget ahead of it (#390). The caller has already sorted
-        ``_pending`` by cost. The existing take-loop then fills any remaining
-        budget, or takes this request alone when nothing else fits — including
-        a single request larger than ``max_batch_cost``.
-
-        Returns:
-            Whether the oldest request was moved to the front.
-        """
-        if len(self._pending) < 2:
-            return False
-        oldest = min(self._pending, key=lambda request: request.arrival_time)
-        waited_ms = (time.monotonic() - oldest.arrival_time) * 1000
-        if waited_ms < self._config.max_batch_wait_ms:
-            return False
-        oldest_index = next(index for index, request in enumerate(self._pending) if request is oldest)
-        if oldest_index == 0:
-            return False
-        self._pending.insert(0, self._pending.pop(oldest_index))
-        return True
-
     def _extract_batch(
         self,
         max_items: int | None = None,
@@ -484,16 +446,11 @@ def _extract_batch(
 
         Algorithm:
         1. Sort pending requests by cost (ascending)
-        2. If the oldest request has waited at least max_batch_wait_ms, move
-           it to the front so cheaper work cannot skip it (#390)
-        3. Take requests until max_batch_cost would be exceeded
-        4. Keep remaining requests for next get_batch() call
+        2. Take requests until max_batch_cost would be exceeded
+        3. Keep remaining requests for next get_batch() call
 
         This minimizes padding waste by grouping similar-cost sequences.
-        Young requests stay in cost order. An overdue oldest request is taken
-        before that fill; if it does not fit beside the cheaper items, the
-        take-loop keeps it alone instead of leaving it queued. When a pending
-        request is flagged ``runs_alone``, the order comes from
+        When a pending request is flagged ``runs_alone``, the order comes from
         ``_order_runs_alone_lane`` instead, and a flagged request is always
         taken on its own.
 
@@ -515,15 +472,11 @@ def _extract_batch(
         pending_before = len(self._pending)
         cost_before = self._total_cost
 
-        promoted_overdue = False
         if any(request.runs_alone for request in self._pending):
             self._order_runs_alone_lane()
         else:
-            # Sort pending by cost (ascending) for optimal batching.
-            # Promote only after that sort, and only once the oldest request
-            # has outwaited the batch window (#390). runs_alone keeps its lane.
+            # Sort pending by cost (ascending) for optimal batching
             self._pending.sort(key=lambda r: r.item.cost)
-            promoted_overdue = self._promote_overdue_request()
 
         # Greedily take items until we exceed max_batch_cost
         batch_items: list[I] = []
@@ -563,15 +516,6 @@ def _extract_batch(
                 hit_request_limit = True
                 break
 
-        # The overdue request was selected first so it could not be left behind
-        # a cheaper fill. Emit cost order so this batch still pads like any
-        # other cost-sorted sub-batch.
-        if promoted_overdue and take_count > 1:
-            pairs = list(zip(batch_items, batch_metadata, strict=True))
-            pairs.sort(key=lambda pair: pair[0].cost)
-            batch_items = [pair[0] for pair in pairs]
-            batch_metadata = [pair[1] for pair in pairs]
-
         lane = "alone" if take_count > 0 and self._pending[0].runs_alone else "batched"
         if take_count > 0:
             # Whose turn is next when both lanes are pending.
diff --git a/packages/sie_server/tests/core/test_batcher.py b/packages/sie_server/tests/core/test_batcher.py
index 80cb7b1ba..87a418e31 100644
--- a/packages/sie_server/tests/core/test_batcher.py
+++ b/packages/sie_server/tests/core/test_batcher.py
@@ -1274,124 +1274,3 @@ def extract() -> None:
         assert sorted(served) == sorted(submitted)
         assert not flagged_waits
         assert batcher.pending_cost == 0
-
-
-class _MonotonicClock:
-    """Stand-in for ``time.monotonic``. Advance it; do not sleep."""
-
-    def __init__(self, now: float) -> None:
-        self.now = now
-
-    def monotonic(self) -> float:
-        return self.now
-
-    def advance_ms(self, delay_ms: float) -> None:
-        self.now += delay_ms / 1000
-
-
-class TestOverdueRequestScheduling:
-    """A high-cost request is not starved by cheaper arrivals (#390).
-
-    Cost order still fills young batches. Once the oldest request has waited
-    ``max_batch_wait_ms``, the next batch takes it before cheaper work.
-    """
-
-    def _clock(self, monkeypatch: pytest.MonkeyPatch) -> _MonotonicClock:
-        clock = _MonotonicClock(10_000.0)
-        monkeypatch.setattr(time, "monotonic", clock.monotonic)
-        return clock
-
-    @pytest.mark.asyncio
-    async def test_young_cost_order_holds_then_overdue_request_is_taken(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        """Cheap fills skip an older expensive item only while it is still young.
-
-        Three immediate cost-ordered batches run before ``max_batch_wait_ms``.
-        After the wait elapses, the next batch includes that item and still
-        uses any remaining budget for a cheaper one that fits beside it.
-        """
-        real_monotonic = time.monotonic
-        clock = self._clock(monkeypatch)
-        wait_ms = 100.0
-        batcher: BatchFormer[TextPreparedItem, str] = BatchFormer(
-            BatchConfig(
-                max_batch_cost=10,
-                max_batch_requests=32,
-                max_batch_wait_ms=wait_ms,
-                coalesce_ms=1_000,
-                coalesce_ratio=1.0,
-            )
-        )
-        await batcher.submit(make_text_item(list(range(6))), "expensive")
-
-        for round_index in range(3):
-            clock.advance_ms(10)  # 10ms, 20ms, 30ms — still inside the 100ms window
-            for item_index in range(4):
-                await batcher.submit(make_text_item([1, 2, 3], item_index), f"cheap-{round_index}-{item_index}")
-            started = real_monotonic()
-            batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
-            assert real_monotonic() - started < 0.05
-            assert "expensive" not in batch.metadata
-            assert [item.cost for item in batch.items] == [3, 3, 3]
-            assert batch.total_cost == 9
-
-        assert any(request.metadata == "expensive" for request in batcher._pending)
-        assert any(str(request.metadata).startswith("cheap") for request in batcher._pending)
-
-        # More cheap work arrives after the window. It must not skip the old item.
-        clock.advance_ms(wait_ms)
-        for item_index in range(4):
-            await batcher.submit(make_text_item([1, 2, 3], item_index), f"late-{item_index}")
-
-        started = real_monotonic()
-        batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
-        assert real_monotonic() - started < 0.05
-        assert "expensive" in batch.metadata
-        assert [item.cost for item in batch.items] == [3, 6]
-        assert batch.total_cost == 9
-        assert any(str(request.metadata).startswith(("cheap", "late")) for request in batcher._pending)
-
-    @pytest.mark.asyncio
-    async def test_overdue_request_larger_than_the_cost_budget_is_taken_alone(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """An overdue request that does not fit beside cheaper items is not left queued.
-
-        The take-loop already returns one oversized item. Putting the overdue
-        request first uses that path instead of filling another cheap batch.
-        """
-        real_monotonic = time.monotonic
-        clock = self._clock(monkeypatch)
-        wait_ms = 100.0
-        batcher: BatchFormer[TextPreparedItem, str] = BatchFormer(
-            BatchConfig(
-                max_batch_cost=10,
-                max_batch_requests=32,
-                max_batch_wait_ms=wait_ms,
-                coalesce_ms=1_000,
-                coalesce_ratio=1.0,
-            )
-        )
-        await batcher.submit(make_text_item(list(range(30))), "expensive")
-
-        for round_index in range(3):
-            clock.advance_ms(10)
-            for item_index in range(3):
-                await batcher.submit(make_text_item([1, 2, 3, 4], item_index), f"cheap-{round_index}-{item_index}")
-            batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
-            assert "expensive" not in batch.metadata
-            assert [item.cost for item in batch.items] == [4, 4]
-            assert batch.total_cost == 8
-
-        assert any(request.metadata == "expensive" for request in batcher._pending)
-
-        clock.advance_ms(wait_ms)
-        for item_index in range(3):
-            await batcher.submit(make_text_item([1, 2, 3, 4], item_index), f"late-{item_index}")
-
-        started = real_monotonic()
-        batch = await asyncio.wait_for(batcher.get_batch(), timeout=1.0)
-        assert real_monotonic() - started < 0.05
-        assert batch.metadata == ["expensive"]
-        assert batch.total_cost == 30
-        assert batcher.pending_count > 0
-        assert all(request.metadata != "expensive" for request in batcher._pending)

From 9225a3aa9020bbcce899fa681115beca4ff8dbc6 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Fri, 9 Oct 2026 23:11:17 +0900
Subject: [PATCH 3/6] perf(gliclass): turn the adaptive batching controller off
 on gliclass-large-v1.0

Refs #390

This follows the #648 profile setting instead of a global batcher change.
---
 .../sie_server/models/knowledgator__gliclass-large-v1.0.yaml  | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml b/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
index cd3347987..cb28fd1ba 100644
--- a/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
+++ b/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
@@ -14,6 +14,10 @@ max_sequence_length: 1024  # the gliclass pipeline's default window, also used i
 profiles:
   default:
     max_batch_tokens: 16384
+    # The adaptive controller compares request latency, queue time included, with
+    # its target and shrinks GLiClass batches under load.
+    adaptive_batching:
+      enabled: false
     compute_precision: null
     adapter_path: sie_server.adapters.gliclass:GLiClassAdapter
     adapter_options:

From 12c36f4140002c71118f6c888ab00a6fdf98ba51 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Sun, 18 Oct 2026 17:57:51 +0900
Subject: [PATCH 4/6] revert(gliclass): drop the adaptive-batching opt-out

The L4 comparison showed that disabling the controller on
knowledgator/gliclass-large-v1.0 does not dispatch the over-budget
rows in #390. The profile matches main again.

Refs #390
---
 .../sie_server/models/knowledgator__gliclass-large-v1.0.yaml  | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml b/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
index cb28fd1ba..cd3347987 100644
--- a/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
+++ b/packages/sie_server/models/knowledgator__gliclass-large-v1.0.yaml
@@ -14,10 +14,6 @@ max_sequence_length: 1024  # the gliclass pipeline's default window, also used i
 profiles:
   default:
     max_batch_tokens: 16384
-    # The adaptive controller compares request latency, queue time included, with
-    # its target and shrinks GLiClass batches under load.
-    adaptive_batching:
-      enabled: false
     compute_precision: null
     adapter_path: sie_server.adapters.gliclass:GLiClassAdapter
     adapter_options:

From e77fcdf0b14a98b69a6fa2ae9c5d147d1c431316 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Sun, 18 Oct 2026 17:57:51 +0900
Subject: [PATCH 5/6] fix(server): serve an over-budget item between cheaper
 batches

Cost-sorted packing never reaches an item above the configured batch
cost while a cheaper item is waiting, so a cheap stream holds it until
the stream ends. Those items now take turns with one cheaper batch,
and the over-budget batch is that item alone. Items that fit the
configured cost stay on cost order. This does not promote by wait time.

Refs #390
---
 .../sie_server/src/sie_server/core/batcher.py |  91 +++++++-
 .../sie_server/tests/core/test_batcher.py     | 201 +++++++++++++++++-
 2 files changed, 283 insertions(+), 9 deletions(-)

diff --git a/packages/sie_server/src/sie_server/core/batcher.py b/packages/sie_server/src/sie_server/core/batcher.py
index c4a5cf4ca..a4adb2cf4 100644
--- a/packages/sie_server/src/sie_server/core/batcher.py
+++ b/packages/sie_server/src/sie_server/core/batcher.py
@@ -25,8 +25,24 @@
 a long item's batch, and waits for at most one flagged item between two
 batches of its lane. A flagged item with k flagged items ahead of it waits
 for at most k + 1 batches of batchable items, however many keep arriving.
-When nothing pending is flagged, batch formation is exactly the cost-sorted
-greedy packing above.
+When nothing pending is flagged and every pending item fits the configured
+``max_batch_cost``, batch formation is exactly the cost-sorted greedy
+packing above.
+
+An item whose cost is above that configured limit cannot share a batch
+formed at the limit. Sorted packing never selects it while a cheaper item
+is pending, so cheaper arrivals hold it until they stop (#390). Those
+items take turns with the cheaper lane: one cost-sorted batch of items
+that fit, then the oldest over-budget item by itself. The first batch
+comes from the lane of the oldest pending item. An over-budget item with
+k over-budget items ahead of it waits for at most k + 1 cheaper batches,
+however many cheaper items keep arriving. The adaptive controller may
+later change the live ``max_batch_cost``; lane membership stays on the
+budget ``BatchConfig`` was built with, so a raised live limit does not
+pull the item back into a cheaper pack. Items at or under the configured
+limit stay on cost order, including when cheaper items fill the live
+budget ahead of them. While a ``runs_alone`` item is pending, only that
+lane runs.
 
 Cost semantics vary by modality (modality-native units):
 - Text: cost = token count
@@ -177,6 +193,9 @@ def __init__(
             self.max_batch_cost = max_batch_tokens
         else:
             self.max_batch_cost = max_batch_cost
+        # The adaptive controller writes max_batch_cost. Over-budget lane
+        # membership stays on the budget this config was built with (#390).
+        self.configured_max_batch_cost = self.max_batch_cost
         self.max_batch_requests = max_batch_requests
         self.max_batch_wait_ms = max_batch_wait_ms
         self.coalesce_ms = coalesce_ms
@@ -229,6 +248,7 @@ def __init__(self, config: BatchConfig | None = None) -> None:
         self._first_request_time: float | None = None
         self._last_submit_time: float | None = None
         self._last_batch_alone: bool | None = None
+        self._last_over_budget_batch: bool | None = None
 
     @property
     def config(self) -> BatchConfig:
@@ -437,6 +457,51 @@ def _order_runs_alone_lane(self) -> None:
         if alone_turn:
             self._pending.insert(0, self._pending.pop(head_index))
 
+    def _order_over_budget_lane(self) -> bool:
+        """Order items that exceed the configured batch cost. Return True when this batch is one of them.
+
+        Packable items stay cost-sorted. Over-budget items stay in arrival
+        order and are not taken during a packable turn, so a raised live
+        ``max_batch_cost`` cannot fold one into the cheaper batch. On an
+        over-budget turn the oldest one moves to the front; the caller takes
+        that item alone.
+        """
+        budget = self._config.configured_max_batch_cost
+        has_over = False
+        has_pack = False
+        for request in self._pending:
+            if request.item.cost > budget:
+                has_over = True
+            else:
+                has_pack = True
+        if not has_over:
+            self._pending.sort(key=lambda request: request.item.cost)
+            return False
+
+        self._pending.sort(
+            key=lambda request: (
+                request.item.cost > budget,
+                request.arrival_time if request.item.cost > budget else request.item.cost,
+            )
+        )
+        if has_pack and not self._over_budget_turn(budget):
+            return False
+        head_index = next(index for index, request in enumerate(self._pending) if request.item.cost > budget)
+        if head_index != 0:
+            self._pending.insert(0, self._pending.pop(head_index))
+        return True
+
+    def _over_budget_turn(self, budget: int) -> bool:
+        """Whether the next batch belongs to the over-budget lane.
+
+        Both lanes are pending. The first batch follows the oldest item.
+        After that the lanes alternate.
+        """
+        if self._last_over_budget_batch is None:
+            oldest = min(self._pending, key=lambda request: request.arrival_time)
+            return oldest.item.cost > budget
+        return not self._last_over_budget_batch
+
     def _extract_batch(
         self,
         max_items: int | None = None,
@@ -472,11 +537,12 @@ def _extract_batch(
         pending_before = len(self._pending)
         cost_before = self._total_cost
 
-        if any(request.runs_alone for request in self._pending):
+        alone_lane = any(request.runs_alone for request in self._pending)
+        over_budget_turn = False
+        if alone_lane:
             self._order_runs_alone_lane()
         else:
-            # Sort pending by cost (ascending) for optimal batching
-            self._pending.sort(key=lambda r: r.item.cost)
+            over_budget_turn = self._order_over_budget_lane()
 
         # Greedily take items until we exceed max_batch_cost
         batch_items: list[I] = []
@@ -495,6 +561,10 @@ def _extract_batch(
             # unless it is the first item taken.
             if request.runs_alone and take_count > 0:
                 break
+            # A packable turn stops at the over-budget lane. A raised live
+            # budget must not fold that item into the cheaper batch.
+            if not over_budget_turn and take_count > 0 and request.item.cost > self._config.configured_max_batch_cost:
+                break
             # Would this item push us over the cost limit?
             # Always take at least one item (handles single large requests)
             if batch_cost + request.item.cost > self._config.max_batch_cost:
@@ -508,7 +578,7 @@ def _extract_batch(
             batch_cost += request.item.cost
             take_count += 1
 
-            if request.runs_alone:
+            if request.runs_alone or over_budget_turn:
                 break
 
             # Also respect max_batch_requests
@@ -516,10 +586,17 @@ def _extract_batch(
                 hit_request_limit = True
                 break
 
-        lane = "alone" if take_count > 0 and self._pending[0].runs_alone else "batched"
+        if take_count > 0 and self._pending[0].runs_alone:
+            lane = "alone"
+        elif over_budget_turn:
+            lane = "over_budget"
+        else:
+            lane = "batched"
         if take_count > 0:
             # Whose turn is next when both lanes are pending.
             self._last_batch_alone = lane == "alone"
+            if not alone_lane:
+                self._last_over_budget_batch = over_budget_turn
 
         # Remove taken items from pending
         self._pending = self._pending[take_count:]
diff --git a/packages/sie_server/tests/core/test_batcher.py b/packages/sie_server/tests/core/test_batcher.py
index 87a418e31..32c4e17ab 100644
--- a/packages/sie_server/tests/core/test_batcher.py
+++ b/packages/sie_server/tests/core/test_batcher.py
@@ -1092,7 +1092,11 @@ class TestRunsAloneLane:
 
     @pytest.mark.parametrize("seed", range(20))
     def test_unflagged_items_batch_exactly_as_cost_sorted_greedy_packing(self, seed: int) -> None:
-        """Without a flagged item pending, every batch matches the old algorithm."""
+        """Items that fit the configured budget match cost-sorted greedy packing.
+
+        An item above that budget uses the over-budget lane instead. This
+        control keeps every generated cost inside the budget.
+        """
         rng = random.Random(seed)  # noqa: S311 -- reproducible test data
         config = BatchConfig(max_batch_cost=rng.randint(1, 4_000), max_batch_requests=rng.randint(1, 8))
         batcher: BatchFormer[HasCost, int] = BatchFormer(config)
@@ -1101,7 +1105,7 @@ def test_unflagged_items_batch_exactly_as_cost_sorted_greedy_packing(self, seed:
 
         for _ in range(60):
             for _ in range(rng.randint(0, 5)):
-                cost = rng.randint(1, 1_500)
+                cost = rng.randint(1, config.max_batch_cost)
                 # Items with runs_alone=False and items without the attribute.
                 item: HasCost = _Audio(cost=cost) if rng.random() < 0.5 else ExtractPreparedItem(cost, 0)
                 batcher._append_item(item, next_id)
@@ -1274,3 +1278,196 @@ def extract() -> None:
         assert sorted(served) == sorted(submitted)
         assert not flagged_waits
         assert batcher.pending_cost == 0
+
+
+def _queue(batcher: BatchFormer[ExtractPreparedItem, str], cost: int, name: str, *, at: float) -> None:
+    """Queue an extract row as if it arrived at monotonic time ``at``."""
+    batcher._append_item(ExtractPreparedItem(cost, 0), name)
+    batcher._pending[-1].arrival_time = at
+
+
+class TestOverBudgetLane:
+    """Items above the configured batch cost take turns with cheaper items (#390).
+
+    Cost-sorted packing never reaches an item that does not fit the configured
+    limit while a cheaper item is pending. The cheaper lane keeps one batch,
+    then the oldest over-budget item is served alone. Items that fit stay on
+    cost order. Waiting is not itself a reason to move an item.
+    """
+
+    def test_over_budget_items_dispatch_while_cheaper_items_keep_arriving(self) -> None:
+        """24 rows above the GLiClass budget move while cheaper rows are still queued."""
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(
+            BatchConfig(max_batch_cost=16_384, max_batch_requests=64)
+        )
+        for index in range(24):
+            _queue(batcher, 22_355 + index, f"long-{index}", at=float(index))
+        for index in range(8):
+            _queue(batcher, 128, f"cheap-{index}", at=100.0 + index)
+
+        served: list[list[str]] = []
+        for index in range(48):
+            served.append(batcher._extract_batch().metadata)
+            _queue(batcher, 128, f"later-{index}", at=200.0 + index)
+
+        long_batches = [batch for batch in served if batch[0].startswith("long-")]
+        assert [batch[0] for batch in long_batches] == [f"long-{index}" for index in range(24)]
+        assert all(len(batch) == 1 for batch in long_batches)
+        cheap_batches = [batch for batch in served if not batch[0].startswith("long-")]
+        assert all(not name.startswith("long-") for batch in cheap_batches for name in batch)
+        cheap_before = 0
+        seen_long = 0
+        for batch in served:
+            if batch[0].startswith("long-"):
+                assert cheap_before <= seen_long + 1
+                seen_long += 1
+            else:
+                cheap_before += 1
+        assert batcher.pending_count > 0
+
+    def test_the_oldest_pending_item_decides_the_first_batch(self) -> None:
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        _queue(batcher, 250, "long", at=0.0)
+        _queue(batcher, 40, "cheap", at=1.0)
+
+        assert batcher._extract_batch().metadata == ["long"]
+        assert batcher._extract_batch().metadata == ["cheap"]
+
+    def test_over_budget_items_run_in_arrival_order(self) -> None:
+        """A newer, cheaper over-budget row does not pass an older one."""
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        _queue(batcher, 400, "older-larger", at=0.0)
+        _queue(batcher, 200, "newer-smaller", at=1.0)
+
+        assert batcher._extract_batch().metadata == ["older-larger"]
+        assert batcher._extract_batch().metadata == ["newer-smaller"]
+
+    def test_items_within_the_configured_budget_stay_on_cost_order(self) -> None:
+        """A row that fits is not pulled ahead because cheaper rows filled the budget."""
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(
+            BatchConfig(max_batch_cost=100, max_batch_requests=10)
+        )
+        _queue(batcher, 80, "larger", at=0.0)
+        for index in range(6):
+            _queue(batcher, 30, f"cheap-{index}", at=float(index + 1))
+
+        assert batcher._extract_batch().metadata == ["cheap-0", "cheap-1", "cheap-2"]
+        for index in range(3):
+            _queue(batcher, 30, f"more-{index}", at=10.0 + index)
+        second = batcher._extract_batch().metadata
+        assert "larger" not in second
+        assert all(name != "larger" for name in second)
+
+    def test_a_raised_live_budget_does_not_pack_the_over_budget_item_with_cheaper_ones(self) -> None:
+        config = BatchConfig(max_batch_cost=16_384)
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(config)
+        _queue(batcher, 100, "cheap", at=0.0)
+        _queue(batcher, 22_355, "long", at=1.0)
+        config.max_batch_cost = 65_536
+
+        assert config.configured_max_batch_cost == 16_384
+        assert batcher._extract_batch().metadata == ["cheap"]
+        assert batcher._extract_batch().metadata == ["long"]
+        assert batcher.pending_cost == 0
+
+    def test_turn_taking_keeps_pending_cost_consistent(self) -> None:
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        _queue(batcher, 40, "a", at=0.0)
+        _queue(batcher, 250, "b", at=1.0)
+        _queue(batcher, 40, "c", at=2.0)
+
+        assert batcher.pending_cost == 330
+        first = batcher._extract_batch()
+        assert first.metadata == ["a", "c"]
+        assert first.total_cost == 80
+        assert batcher.pending_cost == 250
+        second = batcher._extract_batch()
+        assert second.metadata == ["b"]
+        assert second.total_cost == 250
+        assert batcher.pending_count == 0
+        assert batcher.pending_cost == 0
+
+    @pytest.mark.parametrize("seed", range(20))
+    def test_over_budget_items_wait_for_at_most_one_extra_cheaper_batch(self, seed: int) -> None:
+        rng = random.Random(seed)  # noqa: S311 -- reproducible test data
+        budget = rng.randint(10, 500)
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(
+            BatchConfig(max_batch_cost=budget, max_batch_requests=rng.randint(1, 8))
+        )
+        served: list[str] = []
+        submitted: list[str] = []
+        waits: dict[str, list[int]] = {}
+        last_over: bool | None = None
+
+        def extract() -> None:
+            nonlocal last_over
+            pending = list(batcher._pending)
+            batch = batcher._extract_batch(max_items=rng.choice([None, 1, 3]))
+            assert batch.size > 0
+            over = batch.items[0].cost > budget
+            if over:
+                assert batch.size == 1
+            assert all((item.cost > budget) is over for item in batch.items)
+            has_over = any(request.item.cost > budget for request in pending)
+            has_pack = any(request.item.cost <= budget for request in pending)
+            if has_over and has_pack:
+                oldest_over = min(pending, key=lambda request: request.arrival_time).item.cost > budget
+                assert over is (oldest_over if last_over is None else not last_over)
+            last_over = over
+            if over:
+                ahead, cheap_batches = waits.pop(batch.metadata[0])
+                assert cheap_batches <= ahead + 1
+            else:
+                for wait in waits.values():
+                    wait[1] += 1
+            served.extend(batch.metadata)
+            assert batch.total_cost == sum(item.cost for item in batch.items)
+            assert batcher.pending_cost == sum(request.item.cost for request in batcher._pending)
+
+        now = 0.0
+        for index in range(200):
+            now += rng.uniform(0.0, 2.0)
+            if rng.random() < 0.7:
+                name = f"item-{index}"
+                if rng.random() < 0.25:
+                    waits[name] = [len(waits), 0]
+                    cost = budget + rng.randint(1, budget)
+                else:
+                    cost = rng.randint(1, budget)
+                _queue(batcher, cost, name, at=now)
+                submitted.append(name)
+            if batcher.pending_count and rng.random() < 0.55:
+                extract()
+        while batcher.pending_count:
+            extract()
+
+        assert sorted(served) == sorted(submitted)
+        assert not waits
+        assert batcher.pending_cost == 0
+
+    @pytest.mark.asyncio
+    async def test_try_drain_takes_the_over_budget_turn(self) -> None:
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        await batcher.submit_many(
+            [
+                (ExtractPreparedItem(40, 0), "cheap-a"),
+                (ExtractPreparedItem(250, 1), "long-a"),
+                (ExtractPreparedItem(40, 2), "cheap-b"),
+                (ExtractPreparedItem(180, 3), "long-b"),
+            ]
+        )
+        # submit_many stamps arrival in call order. Equal timestamps keep that order.
+        for index, request in enumerate(batcher._pending):
+            request.arrival_time = float(index)
+
+        first = await batcher.try_drain()
+        assert first is not None
+        assert first.metadata == ["cheap-a", "cheap-b"]
+        second = await batcher.try_drain()
+        assert second is not None
+        assert second.metadata == ["long-a"]
+        third = await batcher.try_drain(max_items=1)
+        assert third is not None
+        assert third.metadata == ["long-b"]
+        assert await batcher.try_drain() is None
+        assert batcher.pending_cost == 0

From a2da9d5049462746a980fc42fc9cb0b60e2a74fb Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Sun, 18 Oct 2026 18:26:01 +0900
Subject: [PATCH 6/6] fix(server): restart the over-budget lane when the queue
 is empty

A finished batch left the lane flag set, so the next queue served a
cheaper batch before an older over-budget item. Clearing the flag when
the queue is empty, including drain_pending, makes that queue follow
its oldest item.

Refs #390
---
 .../sie_server/src/sie_server/core/batcher.py |  6 +++-
 .../sie_server/tests/core/test_batcher.py     | 29 +++++++++++++++++++
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/packages/sie_server/src/sie_server/core/batcher.py b/packages/sie_server/src/sie_server/core/batcher.py
index a4adb2cf4..a8e8e0dd4 100644
--- a/packages/sie_server/src/sie_server/core/batcher.py
+++ b/packages/sie_server/src/sie_server/core/batcher.py
@@ -619,10 +619,12 @@ def _extract_batch(
             len(self._pending),
         )
 
-        # Reset timers if we've emptied the queue
+        # Reset timers if we've emptied the queue. The next queue follows its
+        # oldest item. The runs_alone lane keeps its turn across an empty queue.
         if not self._pending:
             self._first_request_time = None
             self._last_submit_time = None
+            self._last_over_budget_batch = None
 
         return FormattedBatch(
             items=batch_items,
@@ -660,6 +662,8 @@ def drain_pending(self) -> list[T]:
         self._total_cost = 0
         self._first_request_time = None
         self._last_submit_time = None
+        # Same restart as an empty extract. runs_alone still keeps its turn.
+        self._last_over_budget_batch = None
         return metadata
 
     async def try_drain(self, max_items: int | None = None) -> FormattedBatch[I, T] | None:
diff --git a/packages/sie_server/tests/core/test_batcher.py b/packages/sie_server/tests/core/test_batcher.py
index 32c4e17ab..98fed0694 100644
--- a/packages/sie_server/tests/core/test_batcher.py
+++ b/packages/sie_server/tests/core/test_batcher.py
@@ -1333,6 +1333,33 @@ def test_the_oldest_pending_item_decides_the_first_batch(self) -> None:
         assert batcher._extract_batch().metadata == ["long"]
         assert batcher._extract_batch().metadata == ["cheap"]
 
+    def test_an_empty_queue_restarts_the_over_budget_lane(self) -> None:
+        """The next queue follows its oldest item after an extract empties the queue."""
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        _queue(batcher, 250, "lone", at=0.0)
+        assert batcher._extract_batch().metadata == ["lone"]
+        assert batcher.pending_count == 0
+
+        _queue(batcher, 250, "over-budget", at=0.0)
+        _queue(batcher, 40, "cheap", at=1.0)
+
+        assert batcher._extract_batch().metadata == ["over-budget"]
+        assert batcher._extract_batch().metadata == ["cheap"]
+
+    def test_drain_pending_restarts_the_over_budget_lane(self) -> None:
+        """Draining the remainder of a queue clears the over-budget lane flag."""
+        batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
+        _queue(batcher, 250, "over-budget", at=0.0)
+        _queue(batcher, 40, "cheap", at=1.0)
+        assert batcher._extract_batch().metadata == ["over-budget"]
+
+        assert batcher.drain_pending() == ["cheap"]
+        assert batcher.pending_count == 0
+
+        _queue(batcher, 250, "next-over-budget", at=0.0)
+        _queue(batcher, 40, "next-cheap", at=1.0)
+        assert batcher._extract_batch().metadata == ["next-over-budget"]
+
     def test_over_budget_items_run_in_arrival_order(self) -> None:
         """A newer, cheaper over-budget row does not pass an older one."""
         batcher: BatchFormer[ExtractPreparedItem, str] = BatchFormer(BatchConfig(max_batch_cost=100))
@@ -1414,6 +1441,8 @@ def extract() -> None:
                 oldest_over = min(pending, key=lambda request: request.arrival_time).item.cost > budget
                 assert over is (oldest_over if last_over is None else not last_over)
             last_over = over
+            if batcher.pending_count == 0:
+                last_over = None
             if over:
                 ahead, cheap_batches = waits.pop(batch.metadata[0])
                 assert cheap_batches <= ahead + 1