diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py index 9f7d82d..c5f49ad 100644 --- a/src/openultrasast/cpg/backend.py +++ b/src/openultrasast/cpg/backend.py @@ -194,6 +194,10 @@ Runner = Callable[..., subprocess.CompletedProcess[str]] STREAM_ID = "id" STREAM_ROWS = "rows" PARTIAL = "__partial__" +# What each streamed answer cost the engine, as `[{"id": ..., "ms": ...}]` under a reserved key beside the +# census. It is a list so every layer that merges answers per key -- shards concatenate, partitions collect -- +# carries it without knowing what it is, and every consumer that reads answers by request id ignores it. +TIMING = "__timing__" def extract_payload(stdout: str) -> object | None: @@ -254,6 +258,11 @@ def _streamed(body: str, *, complete: bool) -> dict[str, object] | None: elif STREAM_ID in document and STREAM_ROWS in document: answers[str(document[STREAM_ID])] = document[STREAM_ROWS] saw_stream = True + cost = document.get("ms") + if isinstance(cost, (int, float)) and not isinstance(cost, bool): + timing = answers.setdefault(TIMING, []) + if isinstance(timing, list): + timing.append({"id": str(document[STREAM_ID]), "ms": float(cost)}) if not saw_stream: return None if not complete: @@ -1144,7 +1153,7 @@ class JoernBackend: logger.info( "cpg batch %s: %d of %d answered before the kill; asking the other %d in parts", query, - len([k for k in merged if k != "__census__"]), + len([k for k in merged if k not in ("__census__", TIMING)]), len(requests), len(remaining), ) @@ -1249,7 +1258,7 @@ class JoernBackend: logger.warning("cpg batch %s returned no parseable payload", query) return None if parsed.get(PARTIAL): - answered = sum(1 for rid, rows in parsed.items() if rid != PARTIAL and isinstance(rows, list) and rid != "__census__") + answered = sum(1 for rid, rows in parsed.items() if rid not in (PARTIAL, TIMING, "__census__") and isinstance(rows, list)) logger.info("cpg batch %s was killed with %d of %d answer(s) already streamed", query, answered, len(requests)) # A request the engine did not answer is absent, not empty: the caller must be able to tell them apart. return {str(rid): list(rows) for rid, rows in parsed.items() if rid != PARTIAL and isinstance(rows, list)} diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc index 382a88a..f6a80af 100644 --- a/src/openultrasast/cpg/queries/taint.sc +++ b/src/openultrasast/cpg/queries/taint.sc @@ -886,6 +886,12 @@ println(ujson.write(ujson.Obj("__census__" -> census))) System.out.flush() parsed.foreach { case (id, req) => + // What THIS answer cost, measured where it is spent. The driver sizes its portions from it, learns the + // fixed start as the wall time the answers do not account for, and -- when a portion is killed -- knows + // exactly which request was running, because every one before it reported its own time. A per-portion + // wall clock cannot tell one 90 s request from thirty 3 s ones, and fitting a single rate to both + // collapsed the sizer to three sink visits per JVM start on the plugin. + val started = System.nanoTime() def field(name: String): String = req.obj.get(name).map(_.str).getOrElse("") val paramSrc = req.obj.get("parameterSources").map(_.str).getOrElse("false") val fieldSrc = req.obj.get("fieldSources").map(_.str).getOrElse("true") @@ -910,7 +916,7 @@ field("contextEvidence") ): _* ) - println(ujson.write(ujson.Obj("id" -> id, "rows" -> rows))) + println(ujson.write(ujson.Obj("id" -> id, "rows" -> rows, "ms" -> ((System.nanoTime() - started) / 1000000L).toDouble))) System.out.flush() } // The census rides the stream as its FIRST line rather than costing its own invocation, because JVM diff --git a/src/openultrasast/model/partitions.py b/src/openultrasast/model/partitions.py index 972dd8c..f6099b7 100644 --- a/src/openultrasast/model/partitions.py +++ b/src/openultrasast/model/partitions.py @@ -11,7 +11,7 @@ from dataclasses import dataclass from pathlib import Path from typing import Any -from ..cpg.backend import _DeadlineExpired, _remove_owned_tree +from ..cpg.backend import TIMING, _DeadlineExpired, _remove_owned_tree from ..preprocess import LANGUAGE_BY_EXTENSION from .contracts import ExecutionBudget from .layout import is_test_path, is_vendored, layout_facts @@ -125,6 +125,9 @@ class PartitionGraph: for rid, rows in raw.items(): if rid in selected and isinstance(rows, list): answers[rid] = self._normalize(rows) + elif rid == TIMING and isinstance(rows, list): + # Carried, not normalized: it names request ids and costs, never a path. + answers.setdefault(TIMING, []).extend(rows) elif rid == "__census__" and isinstance(rows, list): expected = set(next(p.paths for p in self.partitions if p.language == language)) normalized = self._normalize(rows) diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index a2b110f..c940392 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -27,7 +27,7 @@ from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any -from ..cpg.backend import CpgResult +from ..cpg.backend import TIMING, CpgResult from .candidates import enumerate_candidates from .config_value import request_params as config_params from .contracts import ( @@ -407,6 +407,7 @@ def _scan_repository_impl( # per family put a ten-line file at four minutes and a thousand regions at roughly fifty hours -- while # the queries themselves are milliseconds once the CPG is loaded. rows_by_id: dict[str, list[object]] = {} + too_expensive_ids: set[str] = set() census_reported = False query_started = time.monotonic() pruned = sum(q.reason == "tier_zero" for q in scope.deferred) @@ -432,13 +433,31 @@ def _scan_repository_impl( # portion and collapsed the sizer. `run_batch_once` is one invocation whose kill returns what it # streamed. A backend without it falls back to `run_batch`, unchanged. once = getattr(cpg, "run_batch_once", None) + expensive: dict[str, float] = {} answered_batch = _batched( once if callable(once) else batch, kind, requests, weights=_sink_weights(work, evidence_by_pair), + families=_request_families(work), within=lambda: _within_deadline(execution_budget, reserve=reserve), + too_expensive=expensive, ) + if expensive: + # Asked ALONE and still over the ceiling: this question costs more than one invocation may + # spend, which is a fact about the question and not a transient failure. Reported apart from + # `query_failed`, so a reader can tell a scan that ran out of luck from one that met a question + # it cannot afford. + too_expensive_ids.update(expensive) + degradations.append( + { + "stage": "model", + "reason": "query_too_expensive", + "kind": kind, + "requests": len(expensive), + "max_seconds": max(expensive.values()), + } + ) if answered_batch is None: # The engine could not answer. That is NOT an empty result, and recording it is what keeps a # failed scan from reading as a clean repository: on a 117k-line PHP checkout every query @@ -472,7 +491,7 @@ def _scan_repository_impl( census_reported = True valid = {rid: rows for rid, rows in answered_batch.items() if rid in requests and isinstance(rows, list)} rows_by_id.update(valid) - missing = len(requests) - len(valid) + missing = len(requests) - len(valid) - len(expensive) if missing: degradations.append({"stage": "model", "reason": "query_failed", "kind": kind, "requests": missing}) else: # a backend without batching still works, one call at a time @@ -579,7 +598,11 @@ def _scan_repository_impl( outcomes[rid] = QuestionOutcome( question.identity, "not_arbitrated" if has_answer else "unanswered", - "deadline_exhausted" if not _within_deadline(execution_budget) else "query_unanswered", + "query_too_expensive" + if rid in too_expensive_ids + else "deadline_exhausted" + if not _within_deadline(execution_budget) + else "query_unanswered", _rows_json(rows_by_id[rid]) if has_answer else None, ) if question.identity in context_gaps and outcomes[rid].status == "completed": @@ -844,6 +867,11 @@ def _sink_weights( return weights +def _request_families(work: Sequence[tuple[str, ScanRegion, ArbiterSpec]]) -> dict[str, str]: + """The family of each request id: the unit the sizer learns a cost rate for.""" + return {rid: str(getattr(spec, "family", "")) for rid, _region, spec in work} + + def _sized_batches(requests: Mapping[str, Mapping[str, object]], weights: Mapping[str, int]) -> list[dict[str, Mapping[str, object]]]: """Chunks whose implied sink work stays under the budget, in the order the ranker gave. @@ -878,127 +906,259 @@ def _batched( requests: Mapping[str, Mapping[str, object]], *, weights: Mapping[str, int], + families: Mapping[str, str] | None = None, within: Callable[[], bool] = lambda: True, + too_expensive: dict[str, float] | None = None, ) -> dict[str, list[object]] | None: - """Ask one kind in chunks the engine can afford, and merge what answered. + """Ask one kind in portions the engine can afford, and merge what answered. - A chunk that fails leaves its own requests unanswered rather than discarding the answers beside it, which - is the same contract the backend's split retry keeps. Everything answered is returned; only a kind where - no chunk answered at all is reported as unavailable. + A portion that fails leaves its own requests unanswered rather than discarding the answers beside it. + Everything answered is returned; only a kind where no portion answered at all is reported as unavailable. + A request that could not be answered even alone, and spent more than a portion's target trying, is + recorded in ``too_expensive`` with the seconds it took -- a different fact from a failure, and reported + as one. """ chunks = _sized_batches(requests, weights) - if len(chunks) <= 1: - answered: dict[str, list[object]] | None = batch(kind, requests) - return answered - # The budget adapts to what the engine actually does. The constant it starts from was read off one - # subject, and a graph where sinks are cheaper or dearer would either waste portions or blow them. After - # each portion the observed cost per sink visit, less the fixed start, resizes the rest toward a target - # duration -- so a kill becomes something the sizing avoids rather than something the ceiling handles. - sizer = _PortionSizer(weights) - logger.info( - "cpg %s split into %d batches by sink weight (%d request(s), %d sink visit(s))", - kind, - len(chunks), - len(requests), - sum(weights.get(rid, 1) for rid in requests), - ) + # The cost model is MEASURED, per family. The engine reports what each answer cost, so the sizer learns + # the fixed start as the wall time the answers do not account for and a seconds-per-sink-visit rate for + # each family from its own answers. One global rate fitted to whole portions could not tell one 90 s + # request from thirty 3 s ones: on the plugin it fitted the tail, fell to three visits per portion, and + # never grew back, because a fast portion under an assumed 70 s start read as no information at all. + sizer = _PortionSizer(weights, families) + if len(chunks) > 1: + logger.info( + "cpg %s split into %d batches by sink weight (%d request(s), %d sink visit(s))", + kind, + len(chunks), + len(requests), + sum(weights.get(rid, 1) for rid in requests), + ) merged: dict[str, list[object]] = {} answered_any = False - # 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. + # A QUEUE, not a fixed list, because a killed portion leaves a remainder that must be re-asked rather than + # dropped. The queue keeps working while there is time and something left to ask; `fit` sizes the head + # against the current cost model. queue: list[dict[str, Mapping[str, object]]] = list(chunks) - while queue: + # Requests that were RUNNING when a portion was killed. The engine answers in order and streams each answer + # as it finishes, so after a kill the first id without an answer is the one that held the process at the + # ceiling, and every id after it never started. Those are not evidence of anything and go straight back; + # the culprit is asked alone, after everything else, so an expensive question cannot hold the queue. + # + # This is the difference between a TRANSIENT kill and a genuinely expensive question. The earlier queue + # re-asked the whole remainder at half the size, and on the plugin that meant the same expensive + # output-encoding request killing portion after portion while the cheap ones behind it waited. + isolated: list[str] = [] + while queue or isolated: # 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(): - left = sum(len(c) for c in queue) + left = sum(len(c) for c in queue) + len(isolated) logger.info("cpg %s: %d request(s) left unasked to keep time for arbitration", kind, left) break - chunk = sizer.fit(queue.pop(0), pending=queue) + alone = not queue + chunk = {isolated[0]: requests[isolated.pop(0)]} if alone else 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. - 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) + seconds = time.monotonic() - started + timing = _answer_timing(answer) + answered_ids = {rid for rid in (answer or ()) if rid in chunk} + remainder = [rid for rid in chunk if rid not in answered_ids] 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) + merged.update({rid: rows for rid, rows in answer.items() if rid in chunk}) + if "__census__" in answer and "__census__" not in merged: + merged["__census__"] = answer["__census__"] + # A kill is a partial answer from an engine that LOADED: the census arrived, so the process was + # working through requests in order. Nothing at all -- no census -- is a failure to start or a crash, + # and says nothing about which request was at fault. + killed = answer is not None and bool(remainder) + culprit = remainder[0] if killed else None + sizer.observe(chunk, seconds, answered=answered_ids, timing=timing, culprit=culprit) + if not remainder: + continue + if alone or len(chunk) == 1: + # Asked alone and still unanswered. Termination rests here: a single request is never re-queued. + if seconds >= PORTION_TARGET_SECONDS and too_expensive is not None: + too_expensive[remainder[0]] = round(seconds, 1) + logger.info("cpg %s: request %s unanswered when asked alone (%.0fs)", kind, remainder[0], seconds) + continue + if culprit is not None: + isolated.append(culprit) + rest = {rid: chunk[rid] for rid in remainder if rid != culprit} + logger.info( + "cpg %s: portion killed on request %s after %d of %d answered; %d re-queued, the culprit asked alone later", + kind, + culprit, + len(answered_ids), + len(chunk), + len(rest), + ) + else: + # Nothing answered and no census: the sizer has already made this family dearer, so `fit` will cut + # the portion smaller next time. Heads shrink to single requests, and a single request is dropped. + rest = {rid: chunk[rid] for rid in remainder} + if rest: + queue.insert(0, rest) return merged if answered_any else None +def _answer_timing(answer: Mapping[str, object] | None) -> dict[str, float]: + """Seconds per answered request id, from the engine's own streamed cost; empty when it reported none.""" + timing: dict[str, float] = {} + raw = answer.get(TIMING) if isinstance(answer, Mapping) else None + if not isinstance(raw, list): + return timing + for entry in raw: + if isinstance(entry, Mapping) and isinstance(entry.get("ms"), (int, float)): + rid = str(entry.get("id", "")) + # Shards each answer the same id; its cost is what all of them spent. + timing[rid] = timing.get(rid, 0.0) + float(entry["ms"]) / 1000.0 + return timing + + # What one portion should take: long enough that the fixed JVM start is a minority of it, short enough that # a mis-estimate is bounded and a kill costs little. PORTION_TARGET_SECONDS = 120.0 -PORTION_FIXED_SECONDS = 70.0 # measured start-and-load on a 4 MB graph -PORTION_MIN_INFORMATIVE_SECONDS = 10.0 # less work than this above the fixed start says nothing about cost per visit -PORTION_MAX_GROWTH = 2.0 # a budget may double per portion, never more +# A PRIOR for the start-and-load, used until one whole portion has shown the real one: its wall time less the +# time its answers report. Measured 70 s on a 4 MB graph once, and 25 s on the plugin in the trace that +# showed the constant was wrong -- which is why it is a prior and not a fact. +PORTION_FIXED_SECONDS = 70.0 +# A family with no answers yet is costed at the dearest rate seen so far, never below this. Pessimism is the +# cheaper mistake: an over-estimate costs one extra JVM start, an under-estimate costs a ceiling. +PORTION_PRIOR_SECONDS_PER_VISIT = (PORTION_TARGET_SECONDS - PORTION_FIXED_SECONDS) / MAX_SINK_VISITS_PER_BATCH +# The least the engine's work may be budgeted per portion, whatever the fixed start is measured at. +PORTION_MIN_WORK_SECONDS = 10.0 class _PortionSizer: - """Resize the remaining portions from the cost the engine has shown so far. + """Size portions in predicted SECONDS, from what the engine has shown each family costs. - Sizing starts from `MAX_SINK_VISITS_PER_BATCH`, and the observed seconds per sink visit replaces that - guess as soon as one portion has answered. A portion that did not answer is evidence too, of the - harshest kind: the budget is halved rather than fitted, because the true cost is unknown and above the - ceiling. `pending` is edited in place, so a portion cut down to fit puts its remainder back at the front - of the queue rather than losing it. + Every answered request contributes its own reported cost to its family's rate (seconds per sink visit). + A whole portion also shows the fixed start: its wall time less what its answers account for. A killed + portion's culprit contributes a LOWER BOUND -- it ran for at least the time the others do not explain -- + so its family becomes dearer and the next portion of that family is cut to fit. An engine that reports no + costs is costed from the portion's wall time, spread over its requests by prediction. """ - def __init__(self, weights: Mapping[str, int]) -> None: + def __init__(self, weights: Mapping[str, int], families: Mapping[str, str] | None = None) -> None: self.weights = weights - self.budget = float(MAX_SINK_VISITS_PER_BATCH) + self.families = families or {} + self.fixed = PORTION_FIXED_SECONDS + self._fixed_seen: list[float] = [] + self.spent: dict[str, float] = {} + self.visited: dict[str, int] = {} + self.penalty: dict[str, float] = {} def weight(self, chunk: Mapping[str, object]) -> int: return sum(self.weights.get(rid, 1) for rid in chunk) + def rate(self, family: str) -> float: + """Seconds per sink visit for a family: its own, else the dearest known, else the prior.""" + known = [self.spent[f] / self.visited[f] for f in self.visited if self.visited[f] > 0] + base = self.spent[family] / self.visited[family] if self.visited.get(family) else max([PORTION_PRIOR_SECONDS_PER_VISIT, *known]) + return base * self.penalty.get(family, 1.0) + + def cost(self, rid: str) -> float: + return self.rate(self.families.get(rid, "")) * self.weights.get(rid, 1) + + @property + def allowance(self) -> float: + """Predicted engine seconds one portion may carry, once the fixed start is paid.""" + return max(PORTION_MIN_WORK_SECONDS, PORTION_TARGET_SECONDS - self.fixed) + def fit( self, chunk: dict[str, Mapping[str, object]], *, pending: list[dict[str, Mapping[str, object]]] ) -> dict[str, Mapping[str, object]]: - """Shrink this portion to the budget, or grow it from the ones behind it while it still fits.""" - if self.weight(chunk) > self.budget and len(chunk) > 1: - kept: dict[str, Mapping[str, object]] = {} - carried = 0 - for rid, req in chunk.items(): - if kept and carried + self.weights.get(rid, 1) > self.budget: - break - kept[rid] = req - carried += self.weights.get(rid, 1) - pending.insert(0, {rid: req for rid, req in chunk.items() if rid not in kept}) - return kept - while pending and self.weight(chunk) + self.weight(pending[0]) <= self.budget: - chunk = {**chunk, **pending.pop(0)} - return chunk + """Cut this portion to the allowance, or grow it from the ones behind it while it still fits. - def observe(self, chunk: Mapping[str, object], seconds: float, *, answered: bool) -> None: - visits = max(1, self.weight(chunk)) - if not answered: - self.budget = max(1.0, self.budget / 2) - logger.info("cpg portion of %d sink visit(s) did not answer in %.0fs; budget halved to %d", visits, seconds, int(self.budget)) - return - work = seconds - PORTION_FIXED_SECONDS - # A portion that finished inside the fixed start carries no information about cost per visit -- 3 s - # of work over 199 visits read as 0.015 s each and the next budget grew thirteen-fold to 3,404, which - # the ceiling then killed. Measured on the plugin. So a sample below the floor is not extrapolated - # from, and growth is capped at doubling per step: the sizing may be wrong, but never by more than - # one factor of two per portion, which a kill can then undo. - if work >= PORTION_MIN_INFORMATIVE_SECONDS: - fitted = (PORTION_TARGET_SECONDS - PORTION_FIXED_SECONDS) / (work / visits) - self.budget = max(1.0, min(fitted, self.budget * PORTION_MAX_GROWTH)) - logger.info("cpg portion of %d sink visit(s) answered in %.0fs; next budget %d visit(s)", visits, seconds, int(self.budget)) + Always at least one request, however dear: the scan's job is to report that it is expensive, not to + decide it is unaskable. The rest of a cut portion goes to the FRONT of the queue, in ranker order. + """ + kept: dict[str, Mapping[str, object]] = {} + carried = 0.0 + for rid, req in chunk.items(): + if kept and carried + self.cost(rid) > self.allowance: + break + kept[rid] = req + carried += self.cost(rid) + rest = {rid: req for rid, req in chunk.items() if rid not in kept} + if rest: + pending.insert(0, rest) + return kept + # Grown id by id, not chunk by chunk: the queue's head is often one large chunk -- the remainder of the + # ranker's first cut -- and waiting for the whole of it to fit kept portions at one request after a + # collapse even once the engine had shown each costs a fraction of a second. + while pending: + head = pending[0] + taken = [] + for rid in head: + if carried + self.cost(rid) > self.allowance: + break + taken.append(rid) + carried += self.cost(rid) + for rid in taken: + kept[rid] = head.pop(rid) + if head: + break + pending.pop(0) + return kept + + def _learn(self, rid: str, seconds: float) -> None: + family = self.families.get(rid, "") + self.spent[family] = self.spent.get(family, 0.0) + max(0.0, seconds) + self.visited[family] = self.visited.get(family, 0) + self.weights.get(rid, 1) + + def observe( + self, + chunk: Mapping[str, object], + seconds: float, + *, + answered: set[str] | frozenset[str], + timing: Mapping[str, float] | None = None, + culprit: str | None = None, + ) -> None: + timing = timing or {} + done = [rid for rid in chunk if rid in answered] + whole = len(done) == len(chunk) + reported = all(rid in timing for rid in done) + if done and reported: + for rid in done: + self._learn(rid, timing[rid]) + explained = sum(timing[rid] for rid in done) + if whole: + # The fixed start is what the answers do not explain. The median of what has been seen, so one + # slow start (a cold page cache, a busy host) does not resize every portion after it. + self._fixed_seen.append(max(0.0, seconds - explained)) + self.fixed = sorted(self._fixed_seen)[len(self._fixed_seen) // 2] + elif done: + # No per-answer costs: spread the portion's work over its requests in proportion to prediction. + predicted = {rid: self.cost(rid) for rid in done} + total = sum(predicted.values()) or 1.0 + work = max(0.0, seconds - self.fixed) + if whole and seconds < self.fixed: + self.fixed = seconds # a portion cannot finish before its own start: the prior was too high + for rid in done: + self._learn(rid, work * predicted[rid] / total) + explained = work + else: + explained = 0.0 + if culprit is not None: + # It ran at least as long as nothing else accounts for, and did not finish. A lower bound, which is + # exactly what the next portion of its family should be sized against. + self._learn(culprit, seconds - self.fixed - (explained if reported else 0.0)) + elif not done: + # Nothing answered and no census: the engine said nothing about cost, only that this was too much. + for family in {self.families.get(rid, "") for rid in chunk}: + self.penalty[family] = self.penalty.get(family, 1.0) * 2 + logger.info( + "cpg portion of %d request(s), %d sink visit(s): %d answered in %.0fs; fixed start %.0fs, allowance %.0fs", + len(chunk), + self.weight(chunk), + len(done), + seconds, + self.fixed, + self.allowance, + ) def _collect( diff --git a/src/openultrasast/push/reuse.py b/src/openultrasast/push/reuse.py index dccdac2..cb72dc1 100644 --- a/src/openultrasast/push/reuse.py +++ b/src/openultrasast/push/reuse.py @@ -11,7 +11,7 @@ from pathlib import Path from typing import Any from openultrasast.cpg.artifact import GraphArtifact, digest_value -from openultrasast.cpg.backend import CpgResult, JoernBackend +from openultrasast.cpg.backend import TIMING, CpgResult, JoernBackend from openultrasast.model.contracts import ExecutionBudget from openultrasast.model.regions import ScanRegion from openultrasast.push.cache import ArtifactCache, SemanticKeys @@ -74,7 +74,11 @@ class ReusingBackend: return {key: normalize(item) for key, item in value.items()} return value - def batch(kind: str, requests: Mapping[str, Mapping[str, object]]) -> dict[str, list[object]] | None: + def cached(engine: Any) -> Any: + return lambda kind, requests: batch(kind, requests, engine=engine) + + def batch(kind: str, requests: Mapping[str, Mapping[str, object]], *, engine: Any = None) -> dict[str, list[object]] | None: + engine = engine or result.run_batch answers: dict[str, list[object]] = {} missing = {} keys = {} @@ -96,8 +100,8 @@ class ReusingBackend: else: missing[rid] = request if missing and time.monotonic() < budget.deadline_monotonic: - raw: Any = result.run_batch(kind, missing) if result.run_batch else None - if raw is None and result.run_batch is None: + raw: Any = engine(kind, missing) if engine else None + if raw is None and engine is None: raw = {rid: result.run(kind, params) for rid, params in missing.items()} if isinstance(raw, dict): normalized = normalize(raw) @@ -106,6 +110,10 @@ class ReusingBackend: if not valid: return answers or None answers["__census__"] = census + # What the engine spent on the ids it actually ran; a cached answer cost nothing and + # reports no time, so the portioner learns from real work only. + if isinstance(raw.get(TIMING), list): + answers[TIMING] = raw[TIMING] for rid in missing: rows = normalized.get(rid) if isinstance(rows, list): @@ -120,7 +128,15 @@ class ReusingBackend: rows = batch(kind, {"one": params}) return rows.get("one") if rows is not None else None - return replace(result, run=run, run_batch=batch) + # BOTH batch paths go through the cache. The scan's portioner asks `run_batch_once`, and wrapping only + # `run_batch` let every portioned query bypass reuse -- and the source-root normalization with it, so a + # pre-push scan answered from the engine with absolute scratch paths in its rows. + return replace( + result, + run=run, + run_batch=cached(result.run_batch), + run_batch_once=cached(result.run_batch_once) if result.run_batch_once is not None else None, + ) @staticmethod def _census(value: object, artifact: GraphArtifact) -> bool: diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 9f019a5..738a1ad 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -435,6 +435,10 @@ _COVERAGE_DEGRADATIONS: dict[str, str] = { ), "files_unparsed": "{count} file(s) could not be parsed and are absent from the graph: {files}.", "query_failed": "A `{kind}` query failed for {requests} region(s), so those regions were not decided.", + "query_too_expensive": ( + "{requests} `{kind}` question(s) could not be answered even when asked alone, spending up to " + "{max_seconds} s each, so those regions were not decided. This is the cost of the question, not a transient failure." + ), "budget_exhausted": "The model-call budget ran out with {regions_unasked} region(s) still unasked.", "regions_truncated": ( "Only the **{examined} highest-ranked of {total} regions** were examined. Nothing above says anything " diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py index e94f44b..b7ebc52 100644 --- a/tests/test_cpg_backend.py +++ b/tests/test_cpg_backend.py @@ -1271,46 +1271,63 @@ def test_a_killed_batch_retries_only_the_remainder(tmp_path: Path) -> None: assert all("a" not in retry for retry in calls[1:]), calls -def test_the_portioner_asks_single_attempts_and_halves_on_a_short_answer(tmp_path: Path) -> None: - """A killed portion returns what it streamed; the sizer must read that as too big, not as slow. +def test_the_portioner_asks_single_attempts_and_isolates_the_culprit(tmp_path: Path) -> None: + """A killed portion's running request is asked alone, AFTER the rest; the ids behind it go straight back. - Timing `run_batch` folds its internal split-retry into the number -- one portion read 953 s that way and - the sizer fitted a per-visit rate from it and collapsed. The portioner asks `run_batch_once` instead, and - a portion that came back short of the ids it asked halves the budget rather than fitting from a ceiling. + The engine answers in order and streams as it goes, so after a kill the first unanswered id is the one + that held the ceiling and everything after it never started. Re-asking them together at half the size is + what let one expensive request kill portion after portion on the plugin. """ from openultrasast.cpg.backend import CpgResult from openultrasast.model import scan - once_calls: list[list[str]] = [] + calls: list[list[str]] = [] def once(kind: str, requests: dict[str, dict[str, object]]) -> dict[str, list[object]] | None: - once_calls.append(sorted(requests)) - # Answer everything except a designated heavy id, as a kill would: it streamed the rest and died. - return {rid: [] for rid in requests if rid != "heavy"} + calls.append(list(requests)) + answer: dict[str, list[object]] = {"__census__": [{"methods": "3"}]} + for rid in requests: + if rid == "heavy": + return answer # killed while answering heavy: nothing after it started + answer[rid] = [] + return answer result = CpgResult(cpg_path=tmp_path / "c.bin", run=lambda k, p: None, run_batch=lambda k, r: None, run_batch_once=once) + weights = dict.fromkeys(["a", "heavy", "b", "c"], 1) + expensive: dict[str, float] = {} + merged = scan._batched(result.run_batch_once, "taint", {rid: {} for rid in weights}, weights=weights, too_expensive=expensive) + assert merged is not None and {"a", "b", "c"} <= set(merged) and "heavy" not in merged + assert calls[0] == ["a", "heavy", "b", "c"] + assert calls[1] == ["b", "c"], "the requests behind the culprit were not re-asked straight away" + assert calls[-1] == ["heavy"] and calls.count(["heavy"]) == 1, calls + assert not expensive, "a lone failure faster than a portion's target is not 'too expensive'" - seen: list[bool] = [] - real_observe = scan._PortionSizer.observe - def spy(self, chunk, seconds, *, answered): # type: ignore[no-untyped-def] - seen.append(answered) - return real_observe(self, chunk, seconds, answered=answered) +def test_a_single_portion_kill_still_re_asks_its_remainder() -> None: + """A scan small enough for one portion used to take one attempt and lose whatever a kill cut off.""" + from openultrasast.model import scan - # Two portions over the budget so the portioner actually splits and observes. - weights = {"heavy": 500, "a": 400, "b": 400} - requests = {rid: {} for rid in weights} - monkey = scan._PortionSizer.observe - scan._PortionSizer.observe = spy # type: ignore[method-assign] - try: - merged = scan._batched(result.run_batch_once, "taint", requests, weights=weights) - finally: - scan._PortionSizer.observe = monkey # type: ignore[method-assign] + calls: list[list[str]] = [] - assert merged is not None and "a" in merged and "b" in merged and "heavy" not in merged - 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 once(kind: str, requests: dict[str, dict[str, object]]) -> dict[str, list[object]] | None: + calls.append(list(requests)) + if len(calls) == 1: + return {"__census__": [], "a": []} # killed on b + return {"__census__": [], **{rid: [] for rid in requests}} + + merged = scan._batched(once, "taint", {"a": {}, "b": {}, "c": {}}, weights={"a": 1, "b": 1, "c": 1}) + assert merged is not None and {"a", "b", "c"} <= set(merged), (merged, calls) + + +def test_streamed_costs_survive_every_layer() -> None: + """The engine's per-answer cost is parsed, carried beside the census, and never read as a request.""" + from openultrasast.cpg.backend import BEGIN, END, TIMING, extract_payload + from openultrasast.model import scan + + body = f'{BEGIN}\n{{"__census__": []}}\n{{"id": "0", "rows": [], "ms": 1500}}\n{{"id": "1", "rows": [1], "ms": 20}}\n{END}\n' + parsed = extract_payload(body) + assert isinstance(parsed, dict) and parsed[TIMING] == [{"id": "0", "ms": 1500.0}, {"id": "1", "ms": 20.0}] + assert scan._answer_timing(parsed) == {"0": 1.5, "1": 0.02} def test_the_queue_re_asks_a_killed_remainder_and_terminates() -> None: diff --git a/tests/test_cpg_queries_compile.py b/tests/test_cpg_queries_compile.py index 9db5370..35ecb0f 100644 --- a/tests/test_cpg_queries_compile.py +++ b/tests/test_cpg_queries_compile.py @@ -621,3 +621,8 @@ def test_the_taint_query_streams_one_line_per_request(tmp_path: Path) -> None: assert json.loads(lines[0]).keys() == {"__census__"} assembled = extract_payload(done.stdout) assert assembled is not None and set(assembled) >= {"0", "1", "__census__"} and "__partial__" not in assembled + # Every answer reports its own cost, which is what the portioner sizes from and how a kill names the + # request that was running. Absent, the sizer falls back to guessing from wall time. + answers = [json.loads(line) for line in lines[1:]] + assert all(isinstance(a.get("ms"), (int, float)) and a["ms"] >= 0 for a in answers), answers + assert [t["id"] for t in assembled["__timing__"]] == ["0", "1"] diff --git a/tests/test_model_scope.py b/tests/test_model_scope.py index 3f20903..85f5cb9 100644 --- a/tests/test_model_scope.py +++ b/tests/test_model_scope.py @@ -322,26 +322,56 @@ def test_expired_planning_keeps_census_without_more_fact_reads(tmp_path, monkeyp assert "deadline_exhausted" in result.scope.unresolved_boundaries -def test_a_portion_sizer_shrinks_to_budget_and_learns_from_what_answered() -> None: - """Portions follow the engine's observed cost; a portion that did not answer halves the budget.""" +def test_a_portion_sizer_learns_measured_costs_per_family() -> None: + """Portions are sized in predicted seconds from what each family's answers reported costing.""" from openultrasast.model import scan - weights = {"w": 500, "x": 400, "y": 100, "z": 100} - sizer = scan._PortionSizer(weights) - sizer.budget = 600.0 - pending = [{"y": {}}, {"z": {}}] - fitted = sizer.fit({"w": {}, "x": {}}, pending=pending) - assert sorted(fitted) == ["w"] and pending[0] == {"x": {}}, (fitted, pending) # x pushed to the front, not lost - grown = sizer.fit({"y": {}}, pending=[{"z": {}}]) - assert sorted(grown) == ["y", "z"] - sizer.observe({"w": {}}, 300.0, answered=False) - assert sizer.budget == 300.0 - sizer.observe({"y": {}, "z": {}}, scan.PORTION_FIXED_SECONDS + 20.0, answered=True) # 0.1 s per visit - assert 450 < sizer.budget < 550, sizer.budget - # A portion that finished inside the fixed start says nothing: the budget must not move on it. - before = sizer.budget - sizer.observe({"y": {}}, scan.PORTION_FIXED_SECONDS + 1.0, answered=True) - assert sizer.budget == before - # And a genuinely cheap portion may double the budget, never more: 0.01 s per visit would fit 5,000. - sizer.observe({"w": {}}, scan.PORTION_FIXED_SECONDS + 5.0 + 5.0, answered=True) - assert sizer.budget == before * scan.PORTION_MAX_GROWTH, sizer.budget + weights = {"i1": 10, "i2": 10, "o1": 10, **{f"x{i}": 10 for i in range(8)}} + families = {"i1": "injection", "i2": "injection", "o1": "output_encoding", **{f"x{i}": "output_encoding" for i in range(8)}} + sizer = scan._PortionSizer(weights, families) + # A whole portion: answers explain 2 s, the other 23 s is the fixed start -- not the 70 s prior. + sizer.observe({"i1": {}, "i2": {}}, 25.0, answered={"i1", "i2"}, timing={"i1": 1.0, "i2": 1.0}) + assert sizer.fixed == 23.0 + assert abs(sizer.rate("injection") - 0.1) < 1e-9 + # A family never seen is costed at the dearest rate known, never below the prior. + assert sizer.rate("output_encoding") >= scan.PORTION_PRIOR_SECONDS_PER_VISIT + sizer.observe({"o1": {}}, 43.0, answered={"o1"}, timing={"o1": 20.0}) + assert abs(sizer.rate("output_encoding") - 2.0) < 1e-9 + assert abs(sizer.rate("injection") - 0.1) < 1e-9, "one family's cost leaked into another's" + # A portion is cut where the predicted seconds run out: 97 s of allowance holds four 20 s requests. + pending: list[dict[str, dict[str, object]]] = [] + fitted = sizer.fit({f"x{i}": {} for i in range(8)}, pending=pending) + assert len(fitted) == 4 and len(pending[0]) == 4, (fitted, pending) + + +def test_a_fast_portion_grows_the_next_one_after_a_collapse() -> None: + """The trace that motivated this: after two kills the budget fell to 3 visits and never came back. + + Portions of 3 visits finished in 27 s. Under an assumed 70 s start that was 'no information', so the + budget could only shrink. Measured, 27 s is almost all start, the work is cheap, and the next portion + must carry far more. + """ + from openultrasast.model import scan + + weights = {f"r{i}": 3 for i in range(200)} + families = dict.fromkeys(weights, "injection") + sizer = scan._PortionSizer(weights, families) + sizer.observe({"r0": {}}, 27.0, answered={"r0"}, timing={"r0": 0.6}) + pending = [{f"r{i}": {} for i in range(2, 200)}] + fitted = sizer.fit({"r1": {}}, pending=pending) + assert len(fitted) >= 100, len(fitted) + + +def test_a_killed_culprit_makes_its_family_dearer_and_nothing_else() -> None: + """The request running at the kill ran at least as long as nothing else explains.""" + from openultrasast.model import scan + + weights = {"a": 5, "b": 5, "c": 5} + families = {"a": "injection", "b": "output_encoding", "c": "output_encoding"} + sizer = scan._PortionSizer(weights, families) + sizer.fixed = 25.0 + sizer.observe({"a": {}, "b": {}, "c": {}}, 300.0, answered={"a"}, timing={"a": 5.0}, culprit="b") + # b ran at least 300 - 25 - 5 = 270 s over 5 visits. + assert sizer.rate("output_encoding") >= 54.0 + assert abs(sizer.rate("injection") - 1.0) < 1e-9 + assert sizer.fixed == 25.0, "a killed portion must not be read as a fixed-start sample" diff --git a/tests/test_push_reuse.py b/tests/test_push_reuse.py index f611eb8..2ac2bc1 100644 --- a/tests/test_push_reuse.py +++ b/tests/test_push_reuse.py @@ -87,3 +87,37 @@ def test_fresh_missing_census_cannot_borrow_a_cached_query_census(tmp_path): assert "q" in result.run_batch("taint", {"q": {"file": "a.js"}}) answer = result.run_batch("taint", {"q": {"file": "a.js"}, "fresh": {"file": "a.js", "callDepth": 3}}) assert "q" in answer and "fresh" not in answer + + +def test_the_single_attempt_path_is_reused_and_normalized_too(tmp_path): + """The portioner asks `run_batch_once`; wrapping only `run_batch` let every portioned query skip reuse.""" + from openultrasast.cpg.artifact import GraphArtifact, GraphCensus, GraphIdentity + from openultrasast.cpg.backend import TIMING + + identity = GraphIdentity( + "source", "declarations", "exclude", "repository", "javascript", "engine", "frontend", "version", ("dataflowOss",), () + ) + artifact = GraphArtifact(identity, "graph-bytes", 10, GraphCensus(1, 1, 1), "complete", (), "/old", ("a.js",)) + cache = ArtifactCache(tmp_path / "cache", max_bytes=100000) + semantics = SemanticKeys("facts", "queries", "config", "ranking", "admission", "disabled", "empty", "advisory") + wrapper = ReusingBackend(JoernBackend(), cache, semantics, declarations="d", exclusions="e") + census = [{"files": 1, "methods": 1, "file_names": ["a.js"]}] + once_calls = [] + + def once(kind, requests): + once_calls.append(dict(requests)) + return { + **{rid: [{"file": "/old/a.js"}] for rid in requests}, + "__census__": census, + TIMING: [{"id": rid, "ms": 5.0} for rid in requests], + } + + graph = CpgResult(tmp_path / "graph", lambda *_: None, lambda k, r: None, run_batch_once=once) + result = wrapper._answers(graph, artifact, ExecutionBudget(time.monotonic() + 5, 0.2)) + assert result.run_batch_once is not None + first = result.run_batch_once("taint", {"q": {"file": "a.js"}}) + assert first["q"] == [{"file": "a.js"}], "the source root was not stripped on the single-attempt path" + assert first[TIMING] == [{"id": "q", "ms": 5.0}] + second = result.run_batch_once("taint", {"q": {"file": "a.js"}}) + assert second["q"] == [{"file": "a.js"}] and len(once_calls) == 1, "the single-attempt path bypassed the cache" + assert TIMING not in second, "a cached answer cost the engine nothing and must report no time"