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