From 6942446710ff420cf6f7a76e6794d9c0add124dc Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 23 Sep 2026 00:00:53 +0200 Subject: [PATCH] feat(scan): re-ask a killed portion remainder instead of dropping it The portioner iterated a fixed pre-split list, so a killed portion left its unanswered ids behind: on the plugin the list ran out at 1,263 s of a 3,600 s budget with those remainders lost, and completion sat far below what the budget could buy. It is now a work queue. A portion that comes back short pushes its unanswered ids to the front, to be re-asked at the budget the kill just halved; `fit` splits an over-budget head, so heads shrink toward single requests, and a single request a whole ceiling could not answer is dropped rather than looped on forever. A control asserts both the re-ask and that termination. This is the last piece of portioning-over-cap: streaming keeps what a kill finished, single-attempt timing keeps the sizer honest, coverage tells a too-big portion from a slow one, and the queue keeps working while budget and work remain. 1,425 tests pass, nineteen skipped. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- src/openultrasast/model/scan.py | 40 +++++++++++++++++++++------------ tests/test_cpg_backend.py | 22 ++++++++++++++++++ 2 files changed, 48 insertions(+), 14 deletions(-) diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index dd42160..a2b110f 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -904,27 +904,39 @@ def _batched( ) merged: dict[str, list[object]] = {} answered_any = False - for index, chunk in enumerate(chunks): - # Checked per CHUNK, not per kind. A first attempt checked between kinds, and one kind is a single - # iteration that runs every chunk inside it -- so taint spent the whole budget and the reserve only - # stopped the cheap kinds behind it. The granularity of a budget check has to match the granularity - # of the work it guards. + # A QUEUE, not a fixed list, because a killed portion leaves a remainder that must be re-asked at the now + # smaller budget rather than dropped. On the plugin the fixed list ran out at 1,263 s of a 3,600 s budget + # with the killed portions' remainders lost -- completion far below what the budget could buy. The queue + # keeps working while there is time and something left to ask; `fit` sizes the head against the current + # budget, and a portion that comes back short pushes its unanswered ids back for a smaller retry. + queue: list[dict[str, Mapping[str, object]]] = list(chunks) + while queue: + # Checked per portion, not per kind: one kind runs every portion inside a single loop iteration, so a + # per-kind check would let taint spend the whole budget before the reserve looked. if not within(): - logger.info("cpg %s: %d batch(es) left unasked to keep time for arbitration", kind, len(chunks) - index) + left = sum(len(c) for c in queue) + logger.info("cpg %s: %d request(s) left unasked to keep time for arbitration", kind, left) break - chunk = sizer.fit(chunk, pending=chunks[index + 1 :]) + chunk = sizer.fit(queue.pop(0), pending=queue) started = time.monotonic() answer = batch(kind, chunk) # WHOLE, not merely non-empty. A killed portion streams some answers and returns them, which is the # point -- but its time is the ceiling, not the cost of the work, so fitting a per-visit rate from it - # is meaningless. Only a portion that answered every id it was asked gives a clean sample; a short - # one halves the budget, because it was too big for the ceiling. - whole = answer is not None and all(rid in answer for rid in chunk) + # is meaningless. Only a portion that answered every id it was asked gives a clean sample. + answered_ids = set(answer or ()) + whole = answer is not None and all(rid in answered_ids for rid in chunk) sizer.observe(chunk, time.monotonic() - started, answered=whole) - if answer is None: - continue - answered_any = True - merged.update(answer) + if answer is not None: + answered_any = True + merged.update(answer) + # The unanswered remainder goes back to the front, to be re-asked at the now-smaller budget: either + # progress was made and the rest is a genuine remainder, or nothing answered and `fit` will split the + # portion smaller next time. Termination is guaranteed because a portion that answers nothing halves + # the budget, `fit` caps the head at the budget, so heads shrink to single requests; and a SINGLE + # request a whole ceiling could not answer is dropped rather than looped on forever. + remainder = {rid: req for rid, req in chunk.items() if rid not in answered_ids} + if remainder and len(chunk) > 1: + queue.insert(0, remainder) return merged if answered_any else None diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py index 7d12bde..e94f44b 100644 --- a/tests/test_cpg_backend.py +++ b/tests/test_cpg_backend.py @@ -1311,3 +1311,25 @@ def test_the_portioner_asks_single_attempts_and_halves_on_a_short_answer(tmp_pat assert once_calls, "the single-attempt path was not used" # The portion that came back short of its ids was reported as not-whole (False) to the sizer. assert False in seen, seen + + +def test_the_queue_re_asks_a_killed_remainder_and_terminates() -> None: + """A killed portion's unanswered ids are re-asked at a smaller budget; a lone hard request is dropped.""" + from openultrasast.model import scan + + attempts: list[list[str]] = [] + + def batch(kind: str, requests: dict[str, dict[str, object]]) -> dict[str, list[object]] | None: + attempts.append(sorted(requests)) + # "hard" never answers, whatever size it is asked in; everything else answers. + answered = {rid: [] for rid in requests if rid != "hard"} + return answered or None + + weights = {"a": 500, "b": 500, "hard": 500, "c": 500} + requests = {rid: {} for rid in weights} + merged = scan._batched(batch, "taint", requests, weights=weights) + assert merged is not None + assert {"a", "b", "c"} <= set(merged) and "hard" not in merged, merged + # It must stop: "hard" is asked, dropped once it is alone, and never re-queued forever. + assert len(attempts) < 40, len(attempts) + assert attempts.count(["hard"]) <= 1, attempts