diff --git a/docs/LOAD-TESTING.md b/docs/LOAD-TESTING.md index 4a81e48d..525838ea 100644 --- a/docs/LOAD-TESTING.md +++ b/docs/LOAD-TESTING.md @@ -297,6 +297,43 @@ headroom denominator (in + out events, not messages). Exit codes match `--load`. backlog stayed low) so engine numbers are never silently the harness's own ceiling. - The `zero_loss` gate is **exact** by default (no message may be lost). At-least-once re-deliveries (`sink_received > engine_written`) are reported as a count and are *not* treated as loss. +- **`intake_audit` (connection-scale runs only) is the per-MESSAGE companion to `zero_loss`, and it + answers a question `zero_loss` cannot ask.** `zero_loss` compares COUNTS, so its + `engine_read N < confirmed sent M (lost K on intake)` reads identically whether the engine lost an + acknowledged message or the harness's own `engine_read` gauge was short — and `engine_read` is + itself a `COUNT(*)` sampled through two HTTP layers, so a second count could not separate them. + The audit records each send's control id as its response frame comes back and then asks the step's + own store, per message, whether that row is there. Its verdict is on the console, in the JSON + artifact under `records[].intake_audit`, and appended to a failing `no_loss` detail: + - `INTAKE_COMPLETE` — every confirmed send has a row. + - `SAMPLING_LAG` — a shortfall was reported, every confirmed send has a row anyway, **and the + shortfall is larger than the never-confirmed sends can account for**, so the unexplained + remainder is in the gauge (sample attribution or per-inbound sum coverage). A harness defect. + - `UNCONFIRMED_SHORTFALL` — the shortfall is no larger than the set of sends the harness never got + a response frame for, so it implicates **neither** intake **nor** the gauge. Split out from + `SAMPLING_LAG` because the excusal clamps its allowance to zero once the unconfirmed count + exceeds its budget, and the shortfall then consists of sends the engine may never have received; + blaming the gauge for those accused an instrument that was exactly right, and contradicted the + `no_loss` line this verdict is appended to, which already calls that step a systemic no-ACK fault. + - `INVARIANT_SUSPECT` — a send the engine accept-ACKed has no row in its own stopped, committed + store. The engine branch: on a deployment an acknowledged message would be lost at intake. The + verdict names the sequence numbers, so it is reproducible rather than statistical. + - `CORRELATION_SUSPECT` — only *rejected* sends are unmatched. Not an engine finding: several NAK + paths record their row with a NULL control id, so a rejected message is expected to be + unmatchable by control id. + - `PROBE_UNUSABLE` — the audit could not answer. At least: its own read came back empty or + truncated, the send ledger was incomplete, or **nothing was ever confirmed**, which leaves the + compared set empty. That last one is the ledger-side positive control and it is not redundant + with the store-side one — a ledger holding only unconfirmed sends is non-empty by total and + still compares nothing, so without it the audit returned a conclusive "not in intake" computed + over zero elements. Deliberately **not** rendered as "everything is missing", and deliberately + **not** a pass either. + + It runs at two moments: LIVE (engine still up, only on a shortfall) and POST-MORTEM (engine + stopped, always). The post-mortem one is authoritative — a live read can be explained away as + early sampling; a read of a stopped engine's store cannot. Set `intake_audit = false` in the + `[connscale]` profile to skip it on a heavy operator sweep; it is on by default, because a check + an operator has to remember to enable is absent on exactly the run that needed it. ## Known limitations diff --git a/harness/load/__init__.py b/harness/load/__init__.py index 609db8ef..a0838c78 100644 --- a/harness/load/__init__.py +++ b/harness/load/__init__.py @@ -8,10 +8,23 @@ fan-out and times each message end-to-end; an engine poller samples the HTTP API for throughput, backlog, and drain. See ``docs/LOAD-TESTING.md``. -Like :mod:`harness.scenarios`, this package imports no PySide6 and never imports the engine's -``pipeline``/``store``/``config`` internals — only the **pure** surfaces the harness is allowed to -use: the MLLP framing primitives (:mod:`messagefoundry.transports.mllp`), the parsing library, the -generators, and the HTTP :class:`~messagefoundry.apiclient.EngineClient`. +Like :mod:`harness.scenarios`, this package imports no PySide6, and drives the engine through the +**pure** surfaces a client is allowed to use: the MLLP framing primitives +(:mod:`messagefoundry.transports.mllp`), the parsing library, the generators, and the HTTP +:class:`~messagefoundry.apiclient.EngineClient`. + +**The store carve-out, and it is not the client rule being bent.** The rigs that OWN the engine +subprocess they measure — they spawn it, hand it a store, and stop it — are test rigs rather than +clients, and some of their jobs are only doable against the store directly. In +:mod:`harness.load.connscale` that is at least emptying a shared server store between sweep steps +(``runner._reset_server_store``) and the BACKLOG #1292 intake audit's per-message read +(``runner._store_reader``); :mod:`harness.load.shardcert` provisions its own store the same way. +Each goes through the ``Store`` protocol via ``open_store``, lazily imported inside the function so +the import graph of everything else is unchanged, and each is a read/reset path on a store the rig +itself provisioned — never a shortcut around the API for something the API could answer. (Separately +and harmlessly, several modules import the ``AckMode`` enum from ``config``; that is a value type, +not engine state.) The Qt-free client rule itself is unchanged: nothing here imports PySide6, and +the monitoring path is still the HTTP API. """ from __future__ import annotations diff --git a/harness/load/connscale/driver.py b/harness/load/connscale/driver.py index 0926dc42..9c86eed7 100644 --- a/harness/load/connscale/driver.py +++ b/harness/load/connscale/driver.py @@ -22,6 +22,7 @@ import asyncio +from harness.load.connscale.intake_audit import IntakeLedger from harness.load.corpus import Corpus, Outgoing from harness.load.correlator import Correlator from harness.load.metrics import LiveMetrics @@ -50,6 +51,7 @@ def __init__( correlator: Correlator, metrics: LiveMetrics, queue_max: int = 256, + ledger: IntakeLedger | None = None, ) -> None: if count < 1: raise ValueError("connection count must be >= 1") @@ -57,6 +59,10 @@ def __init__( self._base_port = base_port self._count = count self._m = metrics + # BACKLOG #1292: ONE ledger SHARED across the N connections, exactly as the correlator and + # LiveMetrics are. The reconcile it discriminates aggregates across all N, so a per-connection + # ledger would answer a narrower question than the assertion it has to explain. + self._ledger = ledger # One persistent, pipelined connection per inbound port; expect_ack so each send→ACK is timed. self._conns = [ PersistentConnection( @@ -66,6 +72,7 @@ def __init__( metrics, expect_ack=True, queue_max=queue_max, + ledger=ledger, ) for i in range(count) ] diff --git a/harness/load/connscale/intake_audit.py b/harness/load/connscale/intake_audit.py new file mode 100644 index 00000000..5238b36a --- /dev/null +++ b/harness/load/connscale/intake_audit.py @@ -0,0 +1,633 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""The intake audit (BACKLOG #1292) -- a PER-MESSAGE discriminator for an ``engine_read`` shortfall. + +WHAT IT IS FOR. ``harness.load.connscale.runner._reconcile`` fails a step with +``engine_read {read} < confirmed sent {sent - excused} (lost N on intake)``. That message is a +COUNT-vs-COUNT comparison and it cannot be attributed after the fact: the shortfall reads identically +whether the engine lost an acknowledged message (the count-and-log invariant, a real defect) or the +harness's own ``engine_read`` gauge was sampled early / summed short (an instrument defect). This +module asks a DIFFERENT question -- *is THIS message's row there* -- so that the two separate. + +WHY NOT ANOTHER COUNT. ``engine_read`` is ALREADY ``COUNT(*) FROM messages`` for the run's channels, +sampled through two HTTP layers: the store's ``_collect_connection_metrics`` -> the engine's +``connection_metrics_view`` -> ``GET /connections``'s ``read`` field -> ``enginepoll``'s re-sum of the +per-inbound rows. A probe that counted rows would re-derive the number already under dispute. This +one compares SETS of control ids. + +WHAT THE SENDER CONTRIBUTES. :class:`IntakeLedger` is filled by +:class:`~harness.load.sender.PersistentConnection` at the two points where a send LEAVES ``_inflight``: + +* CONFIRMED -- a response frame was read back for it (``_on_ack``), carrying its MSA-1 code and + whether that code was an accept. Measured on this rig the accounting identity + ``sent == acked + nak + timeouts`` holds exactly, so the confirmed set is precisely + ``sent - excused`` -- the same quantity ``_reconcile`` bounds. Matching the reconcile's own + arithmetic is the point; keying on ``acked`` alone would answer a neighbouring question. +* UNCONFIRMED -- still outstanding when the connection closed (``_fail_inflight``), which the + reconcile EXCUSES. Reported separately, and the subset that turns out to be in the store + (``late_unconfirmed_total``) is the honest measure of how loose that excusal is. + +WHAT IT DOES NOT ASSERT. A confirmed send whose MSA-1 was a REJECT and whose row is absent is NOT an +engine finding, and is reported as :data:`VERDICT_CORRELATION_SUSPECT` rather than as loss. Several of +the NAK limbs in ``pipeline/wiring_runner.py`` write their ``messages`` row with ``control_id=None`` +(the decode-error, NUL, parse-failure and oversize paths record the row BEFORE anything parsed an +MSH-10), so such a row EXISTS but is unfindable by control id. A rejected message is therefore +expected to be unmatchable here, and only the ACCEPTED-and-absent set carries the count-and-log +invariant. + +PHI. ``report.py`` states the rule for this artifact family: metrics and metadata only, never message +bodies and never control-id lists. So the audit reports SEQUENCE NUMBERS (dense integers minted by +the harness's own counter, meaningless outside the run) and the DISTINCT MSA-1 codes involved -- both +sufficient to act on, neither a control-id list. **Control ids are not emitted ANYWHERE -- not to the +artifact, not to the log.** Said explicitly because the earlier wording here promised they "stay in +the harness log line", which was false in both directions: a reader who went looking for them found +none, and a maintainer reconciling the prose against the code would have made it true by logging +them, which is precisely what the rule above forbids. A run is reproduced from the seqs and the +profile, never from an identifier list. +""" + +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import Any, Final + +log = logging.getLogger(__name__) + +#: When the audit was taken. Running BOTH is what removes the ambiguity: ``live`` still has the engine +#: up (so "we sampled too early" is available as an explanation), ``post_mortem`` runs against the +#: stopped, committed store (so it is not). +MOMENT_LIVE: Final = "live" +MOMENT_POST_MORTEM: Final = "post_mortem" + +#: The audit did not run (disabled by profile, or the live moment was not triggered). +VERDICT_NOT_RUN: Final = "NOT_RUN" +#: The probe itself could not answer. NEVER read as loss -- see :func:`judge` for the ordering rule. +VERDICT_PROBE_UNUSABLE: Final = "PROBE_UNUSABLE" +#: Every confirmed send has a row, and the reconcile saw no shortfall either. +VERDICT_INTAKE_COMPLETE: Final = "INTAKE_COMPLETE" +#: A shortfall was reported, yet every confirmed send HAS a row -> the ``engine_read`` gauge, not the +#: engine, is short. A harness/instrument defect (sample attribution or sum coverage). +VERDICT_SAMPLING_LAG: Final = "SAMPLING_LAG" +#: A shortfall was reported and it is wholly accounted for by sends that were NEVER CONFIRMED, so it +#: implicates neither intake nor the gauge. Split out from :data:`VERDICT_SAMPLING_LAG` because the +#: runner's excusal clamps ``excused`` to 0 once the unconfirmed count exceeds its budget, and the +#: shortfall handed here then consists of sends the engine may never have seen. Blaming the gauge for +#: those named an instrument that was exactly right, and CONTRADICTED the reconcile text this verdict +#: is appended to -- which already calls that step a systemic no-ACK fault. +VERDICT_UNCONFIRMED_SHORTFALL: Final = "UNCONFIRMED_SHORTFALL" +#: A send the engine ACCEPT-ACKed has no ``messages`` row -> the count-and-log invariant would be +#: broken. The engine branch, and the only one that justifies the P1. +VERDICT_INVARIANT_SUSPECT: Final = "INVARIANT_SUSPECT" +#: Only REJECT-ACKed sends are unmatched -> the harness's frame-to-message correspondence, or the +#: ``control_id=None`` NAK limbs above. Not an engine finding. +VERDICT_CORRELATION_SUSPECT: Final = "CORRELATION_SUSPECT" + +#: How many sends one ledger holds before it stops recording and declares itself overflowed. A ledger +#: that silently stopped recording would render as a clean audit, so overflow is a PROBE_UNUSABLE +#: input, not a shrug. +DEFAULT_LEDGER_CAPACITY: Final = 500_000 + +#: How many sequence numbers a verdict names. The COUNT is always exact and reported beside the +#: sample, so a truncated list never understates the finding. +SAMPLE_CAP: Final = 32 + +_PAGE = 1000 # rows per list_messages page during the store sweep + + +@dataclass(frozen=True) +class ConfirmedSend: + """One send for which the harness READ A RESPONSE FRAME back, with what that frame said.""" + + seq: int + code: str # MSA-1 verbatim ("" when the frame carried no parsable MSA-1) + accepted: bool # the sender's OWN accept decision, passed in rather than re-derived here + + +class IntakeLedger: + """Per-message record of what the sender observed, keyed by control id (MSH-10). + + Written only from the event loop (``PersistentConnection._on_ack`` / ``_fail_inflight``), so no + locking. Optional on the connection and ``None`` by default -- the same seam ``tracker`` uses -- + so the steady-state write path is unchanged when no audit is wanted. + """ + + __slots__ = ("_capacity", "_confirmed", "_duplicates", "_overflow", "_unconfirmed") + + def __init__(self, *, capacity: int = DEFAULT_LEDGER_CAPACITY) -> None: + if capacity < 1: + raise ValueError("ledger capacity must be >= 1") + self._capacity = capacity + self._confirmed: dict[str, ConfirmedSend] = {} + self._unconfirmed: dict[str, int] = {} + self._overflow = 0 + self._duplicates = 0 + + def record_confirmed(self, control_id: str, seq: int, code: str, *, accepted: bool) -> None: + """A response frame was read for ``control_id``. ``accepted`` is the SENDER's decision.""" + if self._reject(control_id): + return + self._confirmed[control_id] = ConfirmedSend(seq, code, accepted) + + def record_unconfirmed(self, control_id: str, seq: int) -> None: + """``control_id`` was still in flight when its connection closed (the reconcile excuses it).""" + if self._reject(control_id): + return + self._unconfirmed[control_id] = seq + + def _reject(self, control_id: str) -> bool: + """Refuse a record, counting WHY. Both counters feed PROBE_UNUSABLE rather than being + absorbed: a ledger that quietly stopped recording, or one whose keys are not unique, produces + a clean-looking set comparison that means nothing.""" + if len(self._confirmed) + len(self._unconfirmed) >= self._capacity: + self._overflow += 1 + return True + if control_id in self._confirmed or control_id in self._unconfirmed: + self._duplicates += 1 + return True + return False + + @property + def confirmed(self) -> Mapping[str, ConfirmedSend]: + return self._confirmed + + @property + def unconfirmed(self) -> Mapping[str, int]: + return self._unconfirmed + + @property + def total(self) -> int: + """Sends accounted for. Compared against ``sent`` -- a mismatch means the ledger is partial, + which invalidates a NULL result (though not a positive one).""" + return len(self._confirmed) + len(self._unconfirmed) + + @property + def overflow(self) -> int: + return self._overflow + + @property + def duplicates(self) -> int: + return self._duplicates + + +@dataclass(frozen=True) +class StoreSnapshot: + """What one read of the step's own store returned. + + ``error`` and ``truncated`` are carried BESIDE the data, never folded into it: an empty + ``control_ids`` is produced identically by a working sweep of an empty store and by a broken + query, and those warrant opposite verdicts. + """ + + control_ids: frozenset[str] + total: int # COUNT(*) of the messages table -- the sweep's positive control + truncated: bool = False + error: str | None = None + + +StoreReader = Callable[[], Awaitable[StoreSnapshot]] + + +@dataclass(frozen=True) +class IntakeAudit: + """One audit: the verdict, its inputs, and enough detail to act on without re-running.""" + + moment: str + verdict: str + read_short: int # the reconcile shortfall this audit was taken against + sent: int + confirmed_total: int + unconfirmed_total: int + store_total: int + missing_accepted_total: int + missing_rejected_total: int + late_unconfirmed_total: int + missing_accepted_seqs: tuple[int, ...] = () # bounded sample; the totals above are exact + missing_rejected_seqs: tuple[int, ...] = () + missing_codes: tuple[str, ...] = () # DISTINCT MSA-1 codes across the missing set, sorted + detail: str = "" + + @property + def conclusive(self) -> bool: + """Did this audit actually answer the question? PROBE_UNUSABLE and NOT_RUN did not, and must + never be read as a pass.""" + return self.verdict in ( + VERDICT_INTAKE_COMPLETE, + VERDICT_SAMPLING_LAG, + VERDICT_UNCONFIRMED_SHORTFALL, + VERDICT_INVARIANT_SUSPECT, + VERDICT_CORRELATION_SUSPECT, + ) + + @property + def engine_suspect(self) -> bool: + """Is this the branch that implicates the ENGINE (vs the harness or the probe)? + + THE MOMENT IS PART OF THE CLAIM, not a caveat on it. A LIVE sweep pages over a store still + being written and can miss a row that is present, so a live INVARIANT_SUSPECT is not by + itself an engine finding -- ``_conclusion`` already says so in the prose. Requiring the + post-mortem here makes the machine surface agree with that text BY CONSTRUCTION rather than + by the SLO gate happening to read the post-mortem field, which is an invariant maintained by + a distant call site and would break silently if another reader picked the live one. + """ + return self.verdict == VERDICT_INVARIANT_SUSPECT and self.moment == MOMENT_POST_MORTEM + + def summary(self) -> str: + """One line a CI reader can act on without re-running anything.""" + return ( + f"intake audit [{self.moment}] {self.verdict}: {self.detail} " + f"(sent={self.sent} confirmed={self.confirmed_total} " + f"unconfirmed={self.unconfirmed_total} store_rows={self.store_total} " + f"missing_accepted={self.missing_accepted_total} " + f"missing_rejected={self.missing_rejected_total} " + f"late_unconfirmed={self.late_unconfirmed_total} " + f"seqs={list(self.missing_accepted_seqs)} codes={list(self.missing_codes)})" + ) + + def to_json_dict(self) -> dict[str, object]: + return { + "moment": self.moment, + "verdict": self.verdict, + "read_short": self.read_short, + "sent": self.sent, + "confirmed": self.confirmed_total, + "unconfirmed": self.unconfirmed_total, + "store_total": self.store_total, + "missing_accepted": self.missing_accepted_total, + "missing_rejected": self.missing_rejected_total, + "late_unconfirmed": self.late_unconfirmed_total, + # Sequence numbers, not control ids (PHI rule, see the module docstring). Bounded sample. + "missing_accepted_seqs": list(self.missing_accepted_seqs), + "missing_rejected_seqs": list(self.missing_rejected_seqs), + "missing_codes": list(self.missing_codes), + "detail": self.detail, + } + + +def not_run(reason: str, *, moment: str = MOMENT_POST_MORTEM) -> IntakeAudit: + """The audit was not taken. Distinct from a clean audit AND from an unusable one.""" + return IntakeAudit( + moment=moment, + verdict=VERDICT_NOT_RUN, + read_short=0, + sent=0, + confirmed_total=0, + unconfirmed_total=0, + store_total=0, + missing_accepted_total=0, + missing_rejected_total=0, + late_unconfirmed_total=0, + detail=reason, + ) + + +def _conclusion(moment: str) -> str: + """What an accept-ACKed-but-absent row is ALLOWED to conclude, which depends on the moment. + + Only the post-mortem may state the engine finding. ``sweep_store`` pages with ``ORDER BY + received_at DESC`` + OFFSET, so a row committed while a LIVE sweep is walking shifts the window + and a genuinely present row can go unread -- the module docstring records this as known and + deliberate, and it manufactures exactly this verdict. The live text therefore reports the same + observation without the conclusion, so a console line or a JSON artifact cannot be quoted as an + invariant violation the post-mortem beside it does not support. + """ + if moment == MOMENT_POST_MORTEM: + return ( + "the engine was STOPPED and its store committed when this was read, so on a deployment " + "the count-and-log invariant would not hold for those messages" + ) + return ( + "NOT an engine finding on its own: this LIVE read pages over a store still being written " + "and can miss a row that is present, so it stands only if the post-mortem reproduces it" + ) + + +def judge( + ledger: IntakeLedger, + snapshot: StoreSnapshot, + *, + moment: str, + sent: int, + read_short: int, + unexplained_short: int | None = None, +) -> IntakeAudit: + """Turn a ledger + one store read into a verdict. Pure -- the whole decision table, unit-testable. + + THE ORDERING IS THE DESIGN, not an accident of writing. A POSITIVE finding is self-evidencing; a + NULL is printed identically by every silent instrument failure, so each way the probe can be + blind is ruled out BEFORE a null is allowed to mean anything: + + 1. the sender-side ledger is overflowed / non-unique / empty -> PROBE_UNUSABLE. + 2. NOTHING was ever confirmed -> PROBE_UNUSABLE. The ledger-side positive control, and separate + from step 1 on purpose: the compared set is ``confirmed``, so a ledger holding only + unconfirmed sends is non-empty by ``total`` and still compares NOTHING. + 3. the store sweep failed or was truncated -> PROBE_UNUSABLE. + 4. the store sweep read ZERO rows against a non-empty ledger -> PROBE_UNUSABLE. The store-side + positive control, checked HERE so a broken query renders as "unusable" and never as "every + message is missing" -- the worst possible false positive to hang a P1 on. + 5. an ACCEPT-ACKed send with no row -> INVARIANT_SUSPECT. Checked BEFORE the partial-ledger guard + below: a short ledger under-reports, so a finding inside it is still a real finding. + 6. only REJECT-ACKed sends unmatched -> CORRELATION_SUSPECT (see the module docstring). + 7. the ledger did not account for every send -> PROBE_UNUSABLE, because a null over a partial + ledger proves nothing. This is step 5's mirror image, and why the two are split rather than + both being checked up front. + 8. a shortfall with NOTHING left unexplained once the never-confirmed sends are set aside -> + UNCONFIRMED_SHORTFALL. ``unexplained_short`` is supplied by the PRODUCER, which alone knows + whether its excusal was clamped; this step must not infer it from the unconfirmed count, + because in-budget those sends are already subtracted out of ``read_short`` and the guess + silences a real gauge finding on the common path. + 9. a shortfall with an unexplained remainder, every confirmed send present -> SAMPLING_LAG: that + remainder is in the gauge, not intake. + 10. otherwise INTAKE_COMPLETE. + """ + confirmed = ledger.confirmed + unconfirmed_count = len(ledger.unconfirmed) + ledger_total = ledger.total + store_ids = snapshot.control_ids + unexplained = read_short if unexplained_short is None else unexplained_short + + def _unusable(detail: str) -> IntakeAudit: + return _build( + moment=moment, + verdict=VERDICT_PROBE_UNUSABLE, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=(), + missing_rejected=(), + late_unconfirmed=0, + detail=detail, + ) + + if ledger.overflow: + return _unusable( + f"the send ledger overflowed after {ledger_total} entries ({ledger.overflow} send(s) " + f"unrecorded), so an absent control id cannot be told from an unrecorded one" + ) + if ledger.duplicates: + return _unusable( + f"{ledger.duplicates} duplicate control id(s) reached the ledger -- the ids are not " + f"unique this run, so set membership does not identify a message" + ) + if ledger_total == 0 and sent > 0: + return _unusable( + f"the send ledger recorded NOTHING against {sent} counted send(s) -- the sender-side " + f"instrument did not run, so a clean set comparison here would be vacuous" + ) + if not confirmed and sent > 0: + # THE POSITIVE CONTROL FOR THE LEDGER SIDE, and it must test `confirmed` rather than + # `ledger_total`: every finding branch below iterates `confirmed` and NOTHING reads + # `unconfirmed` except as a count, so a ledger holding only unconfirmed sends compares an + # EMPTY set and every verdict it could reach would be true of nothing. The guard above does + # not cover this -- it fires only when the ledger is empty outright. Reachable on the + # harness's own headline fault: when the runner's excusal goes over budget it clamps + # `excused` to 0, so a step where no send was ever ACKed arrives here with a large + # `read_short`, and without this it returned a CONCLUSIVE "not in intake" over zero + # elements -- clearing intake on exactly the step the reconcile calls a possible + # accepted-and-dropped. + return _unusable( + f"NO send was ever confirmed against {sent} counted send(s) ({len(ledger.unconfirmed)} " + f"unconfirmed) -- the compared set is empty, so no verdict here could distinguish a " + f"clean intake from a lost one" + ) + if snapshot.error is not None: + return _unusable(f"the store sweep failed: {snapshot.error}") + if snapshot.truncated: + return _unusable( + f"the store sweep was truncated at {len(store_ids)} of {snapshot.total} row(s) -- the " + f"unread remainder is indistinguishable from absence" + ) + if snapshot.total == 0 and ledger_total > 0: + return _unusable( + f"the store sweep read 0 row(s) against {ledger_total} accounted send(s) -- the query " + f"answered nothing rather than the store being empty; reported as unusable, NOT as " + f"{ledger_total} lost messages" + ) + + missing_accepted = tuple( + sorted( + (rec.seq, rec.code) + for cid, rec in confirmed.items() + if rec.accepted and cid not in store_ids + ) + ) + missing_rejected = tuple( + sorted( + (rec.seq, rec.code) + for cid, rec in confirmed.items() + if not rec.accepted and cid not in store_ids + ) + ) + late_unconfirmed = sum(1 for cid in ledger.unconfirmed if cid in store_ids) + + def _matched(verdict: str, detail: str) -> IntakeAudit: + """The three MATCHED outcomes -- every confirmed send accounted for -- differ only in verdict + and prose. Collapsed for the same reason ``_unusable`` above is: three adjacent hand-rolled + blocks differing in one constant make a divergence in a copied argument read as normal, and + that divergence is not hypothetical here (``_unusable`` deliberately passes + ``late_unconfirmed=0`` while these pass the computed value).""" + return _build( + moment=moment, + verdict=verdict, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=(), + missing_rejected=(), + late_unconfirmed=late_unconfirmed, + detail=detail, + ) + + if missing_accepted: + partial = ( + "" + if ledger_total == sent + else f" (the ledger accounted {ledger_total} of {sent} send(s), so this is a LOWER BOUND)" + ) + return _build( + moment=moment, + verdict=VERDICT_INVARIANT_SUSPECT, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=missing_accepted, + missing_rejected=missing_rejected, + late_unconfirmed=late_unconfirmed, + detail=( + f"{len(missing_accepted)} send(s) the engine ACCEPT-ACKed have no messages row in " + f"its own store ({snapshot.total} row(s) present){partial} -- {_conclusion(moment)}" + ), + ) + if missing_rejected: + return _build( + moment=moment, + verdict=VERDICT_CORRELATION_SUSPECT, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=(), + missing_rejected=missing_rejected, + late_unconfirmed=late_unconfirmed, + detail=( + f"{len(missing_rejected)} REJECT-ACKed send(s) are unmatched and no accepted send " + f"is -- not an engine finding: a rejected message may be recorded with a NULL " + f"control id, and the harness pops response frames strictly FIFO" + ), + ) + if ledger_total != sent: + return _unusable( + f"the send ledger accounted {ledger_total} of {sent} counted send(s) -- a clean set " + f"comparison over a partial ledger cannot exclude a loss among the " + f"{sent - ledger_total} it never saw" + ) + # NAMING THE GAUGE IS A POSITIVE CLAIM, so it is made only for the part of the shortfall no + # excusal can forgive -- and that part is COMPUTED BY THE PRODUCER, never inferred here. The + # runner's excusal clamps `excused` to 0 over budget and then hands on a bare int, so + # `read_short` alone cannot say which world it describes: in-budget the unconfirmed sends are + # already subtracted out of it, over-budget they are still inside it. Guessing from the + # unconfirmed COUNT gets the common in-budget case backwards and silences a real gauge finding. + # `None` means the producer did not say; then every missing message is the gauge's to answer + # for, which is the pre-existing behaviour and errs toward a HARNESS finding rather than + # toward silence. + if read_short > 0 and unexplained <= 0: + return _matched( + VERDICT_UNCONFIRMED_SHORTFALL, + f"engine_read is short by {read_short}, and once the {unconfirmed_count} never-" + f"confirmed send(s) are set aside NOTHING is left unaccounted for -- so the shortfall " + f"implicates NEITHER intake NOR the engine_read gauge. All {len(confirmed)} confirmed " + f"send(s) have a messages row ({snapshot.total} row(s) present, {late_unconfirmed} " + f"unconfirmed send(s) arrived anyway)", + ) + if read_short > 0: + return _matched( + VERDICT_SAMPLING_LAG, + f"engine_read is short by {read_short}, of which {unexplained} remain(s) unaccounted " + f"for after the {unconfirmed_count} never-confirmed send(s) are set aside, yet all " + f"{len(confirmed)} confirmed send(s) HAVE a messages row ({snapshot.total} row(s) " + f"present) -- that remainder is in the engine_read gauge (sample attribution or " + f"per-inbound sum coverage), not in intake", + ) + return _matched( + VERDICT_INTAKE_COMPLETE, + f"all {len(confirmed)} confirmed send(s) have a messages row; {snapshot.total} row(s) " + f"present, {late_unconfirmed} excused send(s) arrived anyway", + ) + + +def _build( + *, + moment: str, + verdict: str, + read_short: int, + sent: int, + ledger: IntakeLedger, + snapshot: StoreSnapshot, + missing_accepted: tuple[tuple[int, str], ...], + missing_rejected: tuple[tuple[int, str], ...], + late_unconfirmed: int, + detail: str, +) -> IntakeAudit: + # "(none)" rather than "" so an unparsable MSA-1 is a NAMED cause in the artifact instead of an + # empty string a reader would take for a serialization gap. + codes = sorted({code or "(none)" for _seq, code in (*missing_accepted, *missing_rejected)}) + return IntakeAudit( + moment=moment, + verdict=verdict, + read_short=read_short, + sent=sent, + confirmed_total=len(ledger.confirmed), + unconfirmed_total=len(ledger.unconfirmed), + store_total=snapshot.total, + missing_accepted_total=len(missing_accepted), + missing_rejected_total=len(missing_rejected), + late_unconfirmed_total=late_unconfirmed, + missing_accepted_seqs=tuple(seq for seq, _code in missing_accepted[:SAMPLE_CAP]), + missing_rejected_seqs=tuple(seq for seq, _code in missing_rejected[:SAMPLE_CAP]), + missing_codes=tuple(codes), + detail=detail, + ) + + +async def run_intake_audit( + ledger: IntakeLedger, + reader: StoreReader, + *, + moment: str, + sent: int, + read_short: int, + unexplained_short: int | None = None, +) -> IntakeAudit: + """Read the store once through ``reader`` and judge. + + A reader failure becomes a PROBE_UNUSABLE snapshot rather than an exception: the audit is an + instrument, and an instrument must never fail the run it was added to diagnose. + """ + try: + snapshot = await reader() + except Exception as exc: # noqa: BLE001 - any reader failure is a probe outcome, not a run failure + snapshot = StoreSnapshot(frozenset(), 0, error=f"{type(exc).__name__}: {exc}") + audit = judge( + ledger, + snapshot, + moment=moment, + sent=sent, + read_short=read_short, + unexplained_short=unexplained_short, + ) + if audit.verdict != VERDICT_INTAKE_COMPLETE: + log.warning("%s", audit.summary()) + return audit + + +async def sweep_store(store: Any, *, row_cap: int) -> StoreSnapshot: + """Collect every ``control_id`` in ``store``'s ``messages`` table, unfiltered and paged. + + UNFILTERED BY CHANNEL, DELIBERATELY. The connscale runner gives each SQLite step its own DB file + and empties the shared server store before each step, so this store holds exactly this step's + rows. Filtering by the LIVE inbound registry -- the natural-looking choice -- would reproduce the + exact blind spot the audit exists to detect: the API emits a ``read`` figure only for a channel + still present in ``rr.registry.inbound``, so a row whose channel left the registry (the mid-hold + reload probe) is committed but uncounted. An unfiltered sweep SEES that row. + + ``row_cap`` bounds the work: a table larger than the cap is reported TRUNCATED rather than + partially swept, because an unread remainder is indistinguishable from absence. + + ``control_id`` is stored in CLEARTEXT (``_insert_message`` ciphers only raw/error/summary/ + metadata), so this reads an encrypted store unchanged. Typed against the ``Store`` PROTOCOL + surface (``count_messages``/``list_messages``) rather than a backend, so SQLite and the two + server backends go down one path. + + KNOWN AND DELIBERATE: the paging is ``ORDER BY received_at DESC`` + OFFSET, so a row committed + WHILE the sweep is walking lands at offset 0 and shifts the window, which can drop the last page's + final row. That is why :data:`MOMENT_POST_MORTEM` -- taken after the engine process has exited, so + no insert is possible -- is the authoritative moment and the one the CI assertion reads. A LIVE + sweep can therefore report a message it did not actually miss; the live/post-mortem DELTA is + diagnostic rather than a defect, and a live finding that the post-mortem does not reproduce is + itself evidence about timing. + """ + total = int(await store.count_messages()) + if total > row_cap: + return StoreSnapshot(frozenset(), total, truncated=True) + ids: set[str] = set() + offset = 0 + while offset < total: + rows = await store.list_messages(limit=_PAGE, offset=offset) + if not rows: + # Fewer rows than COUNT(*) promised. Report TRUNCATED rather than returning a short set + # that would read as absence for every row the sweep never reached. + return StoreSnapshot(frozenset(ids), total, truncated=True) + for row in rows: + # Unguarded on purpose: all three backends project `control_id` in `list_messages`. If + # one ever stopped, the KeyError becomes a PROBE_UNUSABLE snapshot in `run_intake_audit` + # -- which is the right verdict for a probe that cannot read its own key, and far better + # than a `.get()` default that would render every row as an absent control id. + cid = row["control_id"] + if cid: + ids.add(str(cid)) + offset += len(rows) + return StoreSnapshot(frozenset(ids), total) diff --git a/harness/load/connscale/profile.py b/harness/load/connscale/profile.py index 8edc56ef..d4bd0fd8 100644 --- a/harness/load/connscale/profile.py +++ b/harness/load/connscale/profile.py @@ -78,6 +78,7 @@ "base_port", "transform", "reload_probe", + "intake_audit", "store_backend", "corpus_count_per_trigger", "correlator_capacity", @@ -157,6 +158,12 @@ class ConnScaleProfile: # variance from a single run: the runner loops each (claim_mode, fuse, sweep_mode, count) cell # ``trials`` times as distinct steps, and build_fuse_comparison aggregates the repeats by key. trials: int = 1 + # BACKLOG #1292: run the PER-MESSAGE intake audit alongside the count-based no-loss reconcile. + # Default ON, because the whole point of the item is that a shortfall in CI cannot be attributed + # after the fact -- an audit an operator has to remember to enable would be absent on exactly the + # run that needed it. It costs one paged sweep of the step's own store per step, so the heavy + # operator sweeps (N=1500, long holds) can turn it off. + intake_audit: bool = True def modes(self) -> tuple[str, ...]: """The sweep modes to run (``both`` expands to both, in a stable order).""" @@ -308,6 +315,7 @@ def _profile_from_data(data: dict[str, Any], *, where: str) -> ConnScaleProfile: ), transform=transform, reload_probe=_opt_bool(cs, "reload_probe", f"{where} [connscale]", default=False), + intake_audit=_opt_bool(cs, "intake_audit", f"{where} [connscale]", default=True), store_backend=store_backend, corpus_count_per_trigger=_opt_int( cs, "corpus_count_per_trigger", f"{where} [connscale]", default=20, minimum=1 diff --git a/harness/load/connscale/report.py b/harness/load/connscale/report.py index 0c0a7c8e..075e8670 100644 --- a/harness/load/connscale/report.py +++ b/harness/load/connscale/report.py @@ -7,6 +7,13 @@ reconcile, plus an SLO verdict. **Metrics + metadata only** — never message bodies or control-id lists (PHI rule). Pure + deterministic, so it unit-tests without a live run. +That rule is why the BACKLOG #1292 intake audit reports **sequence numbers**, not the control ids it +actually matched on: a seq is a dense integer minted by the harness's own counter and meaningless +outside the run, so it identifies the message for a follow-up without putting a list of message +identifiers into a shared artifact. The control ids are emitted NOWHERE -- not here and not to the +log; the previous wording sent readers to a log line that never carried them, and invited a +maintainer to make the sentence true by logging exactly what this rule exists to keep out. + The thundering-herd measurement is reported **explicitly and separated** (critic must-change #3): the ``fixed_aggregate`` sweep (constant R across N) IS the herd measurement, so the report carries the ``empty_claims_wake_fanout``-per-second slope vs N AS the wake-fanout cost, kept DISTINCT from the @@ -22,6 +29,13 @@ from typing import TYPE_CHECKING from harness._spreadsheet import SPREADSHEET_FORMULA_TRIGGERS, spreadsheet_safe +from harness.load.connscale.intake_audit import ( + MOMENT_LIVE, + VERDICT_INTAKE_COMPLETE, + VERDICT_NOT_RUN, + IntakeAudit, + not_run, +) if TYPE_CHECKING: from harness.load.connscale.compare import ( @@ -162,6 +176,19 @@ class ConnScaleRecord: fd_probe_ticks: int = 0 fd_probe_degraded_ticks: int = 0 fd_probe_degraded: tuple[str, ...] = () + # --- BACKLOG #1292: the intake audit, the PER-MESSAGE discriminator for a no_loss shortfall --- + # `no_loss` compares COUNTS, and a shortfall in it reads identically whether the engine lost an + # acknowledged message or the `engine_read` gauge was short. These carry the per-message verdict + # that separates the two. `intake_audit` is the POST-MORTEM one (taken against the stopped, + # committed store, so sampling timing cannot explain it) and is the authoritative field; + # `intake_audit_live` is the one taken while the engine was still up, and runs only on a + # shortfall -- the DELTA between them is what says sample-lag vs sum-coverage. Both default to a + # NOT_RUN verdict so an older artifact / a record built without the audit deserializes unchanged + # and never reads as a clean pass it did not earn. + intake_audit: IntakeAudit = field(default_factory=lambda: not_run("audit not wired")) + intake_audit_live: IntakeAudit = field( + default_factory=lambda: not_run("audit not wired", moment=MOMENT_LIVE) + ) def to_json_dict(self) -> dict[str, object]: return { @@ -235,6 +262,12 @@ def to_json_dict(self) -> dict[str, object]: "degraded": list(self.fd_probe_degraded), }, }, + # BACKLOG #1292. Sequence numbers and MSA-1 codes only -- never control ids, per the + # module docstring's metadata-only rule. + "intake_audit": { + "post_mortem": self.intake_audit.to_json_dict(), + "live": self.intake_audit_live.to_json_dict(), + }, "wall5_reload": {"seconds": self.reload_seconds}, "wall6_ack_ms": { "p50": round(self.ack_p50_ms, 3), @@ -401,6 +434,15 @@ def render_console(self) -> str: f"fd probe: {r.sweep_mode}@N={r.count} -- {r.fd_probe_degraded_ticks} of " f"{r.fd_probe_ticks} tick(s) measured nothing [{causes}]" ) + # BACKLOG #1292: the per-message attribution, printed whenever it says anything beyond "clean". + # A `no_loss` shortfall renders as a bare count in the table above, which is exactly the + # unattributable failure this exists to replace -- so the verdict goes on the console beside + # it, not only in the JSON artifact. + for r in self.records: + for audit in (r.intake_audit_live, r.intake_audit): + if audit.verdict in (VERDICT_INTAKE_COMPLETE, VERDICT_NOT_RUN): + continue + lines.append(f"{r.sweep_mode}@N={r.count} -- {audit.summary()}") lines.append("") lines.append("SLOs:") if not self.slos: diff --git a/harness/load/connscale/runner.py b/harness/load/connscale/runner.py index d4ca4fb7..d3c7c038 100644 --- a/harness/load/connscale/runner.py +++ b/harness/load/connscale/runner.py @@ -30,10 +30,11 @@ import tempfile import time from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import Any +from harness.load.connscale import intake_audit from harness.load.connscale.compare import ( ClaimModeComparison, FuseModeComparison, @@ -42,6 +43,7 @@ build_fuse_comparison, ) from harness.load.connscale.driver import ConnScaleDriver +from harness.load.connscale.intake_audit import IntakeAudit, IntakeLedger, StoreReader from harness.load.connscale.probe import FdSampler, ProcSample, time_reload from harness.load.connscale.profile import ConnScaleProfile from harness.load.connscale.report import ( @@ -87,6 +89,10 @@ _STOP_GRACE = 5.0 _SETTLE = 0.5 # let final ACKs/arrivals settle before the truly-final engine sample _HEALTH_TIMEOUT = 30.0 +# One spelling, because both audit moments report it and they must not drift into disagreeing about +# why nothing ran -- a reader comparing the two moments of a disabled step reads the difference as +# meaningful. +_AUDIT_DISABLED = "intake audit disabled for this profile" _PORTS_READY_TIMEOUT = 60.0 # waiting for the engine to report all N inbound rows (N can be large) # A single trivial ADT type — the connscale graph routes every message identically, so the mix only # needs to drive ONE generated type (the wall is per-connection machinery, not message-type spread). @@ -333,33 +339,35 @@ async def _run_one_step( if profile.store_backend is None: db_dir = tempfile.mkdtemp(prefix="mefor-connscale-") db_path = str(Path(db_dir) / f"{tag}.db") - node = EngineNode( - tag, - api_port, - env=_node_env( - base_env, - claim_mode=claim_mode, - fuse_mode=fuse_mode, - batch_mode=batch_mode, - count=count, - base_port=profile.base_port, - transform=profile.transform, - sink_host=sink_host, - sink_port=sink_port, - sink_ports=sink_ports, - install_executor_shim=install_executor_shim, - db_path=db_path, - ), - config_dir=_CONFIG_DIR, - cwd=cwd, + # Captured rather than inlined into EngineNode: the BACKLOG #1292 intake audit opens THIS step's + # store afterwards, and it must resolve the same MEFOR_STORE_* the engine itself was given (the + # per-step SQLite file, or the shared server connection) instead of re-deriving them. + node_env = _node_env( + base_env, + claim_mode=claim_mode, + fuse_mode=fuse_mode, + batch_mode=batch_mode, + count=count, + base_port=profile.base_port, + transform=profile.transform, + sink_host=sink_host, + sink_port=sink_port, + sink_ports=sink_ports, + install_executor_shim=install_executor_shim, + db_path=db_path, ) + node = EngineNode(tag, api_port, env=node_env, config_dir=_CONFIG_DIR, cwd=cwd) poller = EnginePoller(node.url, token=None, origin=time.perf_counter()) + # BACKLOG #1292: the per-message send ledger the intake audit reads. None disables the audit + # wholesale (the sender's write path is then byte-identical to pre-#1292). + ledger = IntakeLedger() if profile.intake_audit else None driver = ConnScaleDriver( host=sink_host, base_port=profile.base_port, count=count, correlator=correlator, metrics=metrics, + ledger=ledger, ) fd_sampler: FdSampler | None = None samples: list[EngineSample] = [] @@ -462,6 +470,67 @@ async def _run_one_step( final = await poller.sample_once() if final is not None: samples.append(final) + + # --- BACKLOG #1292: the intake audit, at its TWO moments ------------------------------- + # MOMENT 1, LIVE (engine still up), and only on a shortfall: does every message the harness + # read a response frame for actually HAVE a row right now? If it does, nothing was lost and + # the shortfall is in the `engine_read` gauge. Gated on the shortfall because on a clean step + # it would only re-confirm what MOMENT 2 confirms anyway, at a page sweep per step. + # `poller.baseline`/`poller.final`, NOT the local `final`: those are the exact two samples + # `_build_record` hands `_reconcile`, and the audit has to be triggered by the shortfall the + # step will actually REPORT. The two diverge on the drain-timeout path, where the local + # `final` can be None while the poller still holds an earlier sample. + read_short = _read_shortfall( + metrics.counters, poller.baseline, poller.final, unconfirmed_budget=count + ) + # The part no excusal forgives, computed HERE because only this side knows whether the + # excusal was clamped. The audit must not re-derive it -- see `_unexplained_shortfall`. + unexplained_short = _unexplained_shortfall(metrics.counters, poller.baseline, poller.final) + # The two NOT_RUN reasons answer different questions, and each is stated at the branch it + # describes. The disabled one has to be reachable on a step that DID have a shortfall: "no + # shortfall to attribute" printed on a failing step is a false statement about the run, in + # the one field whose entire job is attribution, on exactly the step someone opens the + # artifact to read. + if ledger is None: + audit_live = intake_audit.not_run(_AUDIT_DISABLED, moment=intake_audit.MOMENT_LIVE) + elif read_short > 0: + audit_live = await intake_audit.run_intake_audit( + ledger, + _store_reader(node_env, metrics.counters.sent), + moment=intake_audit.MOMENT_LIVE, + sent=metrics.counters.sent, + read_short=read_short, + unexplained_short=unexplained_short, + ) + else: + audit_live = intake_audit.not_run( + "no intake shortfall to attribute at this moment", + moment=intake_audit.MOMENT_LIVE, + ) + # MOMENT 2, POST-MORTEM. Stop the engine FIRST, so the store is committed and quiesced: with + # no process running, "we sampled too early" is no longer available as an explanation, which + # is what separates outcome 1 (sample lag) from outcome 2 (sum coverage). `stop()` is + # idempotent, so the `finally` below still runs it on every other path. + audit_final = intake_audit.not_run(_AUDIT_DISABLED) + if ledger is not None: + with contextlib.suppress(Exception): + await node.stop() + if node.alive: + # Refuse to CALL it a post-mortem when it would not be one. A read taken while the + # engine is still running answers the LIVE question, and labelling it post_mortem + # would destroy the one distinction the second moment exists to make. + audit_final = intake_audit.not_run( + "the engine did not stop, so a post-mortem read would not be post-mortem" + ) + else: + audit_final = await intake_audit.run_intake_audit( + ledger, + _store_reader(node_env, metrics.counters.sent), + moment=intake_audit.MOMENT_POST_MORTEM, + sent=metrics.counters.sent, + read_short=read_short, + unexplained_short=unexplained_short, + ) return _build_record( claim_mode=claim_mode, fuse_mode=fuse_mode, @@ -475,6 +544,8 @@ async def _run_one_step( samples=samples, drain_seconds=drain_seconds, reload_seconds=reload_seconds, + audit_live=audit_live, + audit_final=audit_final, ) finally: with contextlib.suppress(Exception): @@ -563,6 +634,37 @@ def _node_env( return env +def _store_reader(node_env: Mapping[str, str], sent: int) -> StoreReader: + """A one-shot reader over THIS step's own store, for the BACKLOG #1292 intake audit. + + Goes through the ``Store`` protocol (``open_store`` -> ``count_messages``/``list_messages``) so + SQLite and the two server backends take one path, and resolves its settings from the SAME env the + engine subprocess was given. It deliberately does NOT go through ``GET /messages``: that route is + ``require_phi_read`` and charges a per-actor anti-automation budget, so a per-message sweep would + be 429'd -- and it would also read the numbers through the very API layer under suspicion. + + The engine-package import sits INSIDE the call, mirroring ``_reset_server_store`` directly above: + this rig OWNS the engine subprocess and inspects its store, which is why it is the one part of + ``harness/load`` that reaches past the HTTP API at all. + """ + # 4x the sends plus a floor: `messages` holds one row per RECEIVED message and this store is + # exclusive to the step, so a table meaningfully larger than the run means the assumption is + # wrong -- report TRUNCATED rather than sweep an unbounded table. + row_cap = max(4 * sent + 1000, 10_000) + + async def _read() -> intake_audit.StoreSnapshot: + from messagefoundry.config.settings import load_settings + from messagefoundry.store.base import open_store + + store = await open_store(load_settings(environ=node_env).store) + try: + return await intake_audit.sweep_store(store, row_cap=row_cap) + finally: + await store.close() + + return _read + + async def _reset_server_store(backend: str, env: Mapping[str, str]) -> tuple[int, int]: """Empty the pipeline tables of the SHARED server store before a step, so every (mode, count) step is apples-to-apples (the pooled arm never inherits the per_lane arm's rows). Opens a short-lived @@ -754,6 +856,8 @@ def _build_record( samples: list[EngineSample], drain_seconds: float | None, reload_seconds: float | None, + audit_live: IntakeAudit | None = None, + audit_final: IntakeAudit | None = None, ) -> ConnScaleRecord: c = metrics_counters.snapshot() base, final = poller.baseline, poller.final @@ -763,6 +867,14 @@ def _build_record( # half-the-run fraction, and its separate intake floor keeps `read >= sent // 2` required here # even when `count` exceeds half the step's sends (the short-hold smoke cells). no_loss = _reconcile(c, base, final, unconfirmed_budget=count) + live = audit_live if audit_live is not None else intake_audit.not_run("audit not wired") + post = audit_final if audit_final is not None else intake_audit.not_run("audit not wired") + # BACKLOG #1292: ATTRIBUTE the reconcile's own failure text, never soften it. `ok` is untouched -- + # a genuine invariant failure still fails the step on the count check exactly as before -- and the + # audit verdict is APPENDED so a CI reader gets the attribution in the same message that + # currently gives them only a number they cannot act on. + if not no_loss.ok and post.verdict != intake_audit.VERDICT_NOT_RUN: + no_loss = replace(no_loss, detail=f"{no_loss.detail}; {post.summary()}") in_pipeline_peak = max((s.in_pipeline for s in samples), default=0) # Wall #1: executor saturation (None when the shim isn't installed → all-None samples). @@ -841,9 +953,83 @@ def _build_record( working_set_peak_bytes=proc.working_set_peak_bytes, fuse_thread_hops=fuse_mode, batch_handoff_statements=batch_mode, + intake_audit=post, + intake_audit_live=live, + ) + + +@dataclass(frozen=True) +class _Excusal: + """How many unconfirmed sends the intake bound forgives this step, and whether that broke.""" + + unconfirmed: int + budget: int + excused: int + over_budget: bool + + +def _excusal(c: Counters, *, unconfirmed_budget: int) -> _Excusal: + """THE definition of the unconfirmed-send excusal, extracted so ``_reconcile`` and the BACKLOG + #1292 intake audit compute the SAME ``sent - excused``. + + The audit exists to explain the ``engine_read {read} < confirmed sent {sent - excused}`` message, + so it has to be triggered by that exact quantity. A second copy of this arithmetic beside it + would let the audit fire on a shortfall the reconcile does not report, or stay silent on one it + does -- either way attributing the wrong failure. Behaviour is verbatim what ``_reconcile`` + computed inline before the extraction. + """ + unconfirmed = c.timeouts + # Three quarters, not half — half was sized against a 16% worst-observed and windows-2025 has + # since produced 51% on a lossless run, failing `main` at 9b03057f by ONE message. `excused` is + # clamped rather than zeroed so an over-budget failure stops claiming intake loss it cannot show. + # `ok` still requires `not over_budget`, so the verdict is unchanged. Full rationale: report.py. + budget = max(unconfirmed_budget, 3 * c.sent // 4) + over_budget = unconfirmed > budget + return _Excusal(unconfirmed, budget, 0 if over_budget else unconfirmed, over_budget) + + +def _read_shortfall( + c: Counters, + base: EngineSample | None, + final: EngineSample | None, + *, + unconfirmed_budget: int, +) -> int: + """``confirmed sent - engine_read`` -- the shortfall ``_reconcile`` reports as intake loss, and + the trigger for the LIVE intake audit. 0 when the engine gauges are unavailable (there is then no + shortfall to attribute; ``_reconcile`` fails the step on its own for that).""" + if base is None or final is None: + return 0 + return ( + c.sent + - _excusal(c, unconfirmed_budget=unconfirmed_budget).excused + - (final.read - base.read) ) +def _unexplained_shortfall( + c: Counters, base: EngineSample | None, final: EngineSample | None +) -> int: + """The part of the shortfall NO excusal can forgive -- ``sent - unconfirmed - engine_read``. + + THE PRODUCER OWNS THIS BECAUSE ONLY THE PRODUCER CAN COMPUTE IT. ``_read_shortfall`` subtracts + ``excused``, which ``_excusal`` CLAMPS TO 0 over budget, and it then returns a bare int -- so the + clamp state is destroyed one line after being computed. A consumer handed only that int cannot + tell the two worlds apart: in-budget the unconfirmed sends are ALREADY subtracted out, so a + residual shortfall is a genuine gauge finding, while over-budget the same number silently + contains them. Comparing the shortfall against the unconfirmed COUNT to guess which world it is + gets the common (in-budget) case backwards and would silence a real gauge finding. + + Subtracting the UNCLAMPED ``c.timeouts`` makes the quantity mean the same thing in both worlds, + so the audit never has to reconstruct budget arithmetic it does not own. Uses ``c.timeouts`` -- + the same input ``_excusal`` calls ``unconfirmed`` -- rather than the ledger, keeping one + definition of the population. + """ + if base is None or final is None: + return 0 + return c.sent - c.timeouts - (final.read - base.read) + + def _reconcile( c: Counters, base: EngineSample | None, @@ -889,14 +1075,13 @@ def _reconcile( # guarantee is enforced SEPARATELY below as an intake floor the excusal cannot lower. See # harness/load/report.py's copy for the full rationale; the three copies are kept in step # deliberately. - unconfirmed = c.timeouts - # Three quarters, not half — half was sized against a 16% worst-observed and windows-2025 has - # since produced 51% on a lossless run, failing `main` at 9b03057f by ONE message. `excused` is - # clamped rather than zeroed so an over-budget failure stops claiming intake loss it cannot show. - # `ok` still requires `not over_budget`, so the verdict is unchanged. Full rationale: report.py. - budget = max(unconfirmed_budget, 3 * sent // 4) - over_budget = unconfirmed > budget - excused = 0 if over_budget else unconfirmed + ex = _excusal(c, unconfirmed_budget=unconfirmed_budget) + unconfirmed, budget, excused, over_budget = ( + ex.unconfirmed, + ex.budget, + ex.excused, + ex.over_budget, + ) read_short = sent - excused - read # The anti-vacuity guarantee, independent of the excusal: at least half the sends must be # observed at intake whatever the budget forgives (nothing clamps `excused` to `sent`, so without @@ -1167,6 +1352,37 @@ def _evaluate_slos(profile: ConnScaleProfile, records: list[ConnScaleRecord]) -> if slo.zero_loss: all_ok = all(r.no_loss.ok for r in records) out.append(SloCheck("zero_loss", True, all_ok, all_ok)) + if profile.intake_audit: + # BACKLOG #1292, and DELIBERATELY NOT folded into zero_loss: this is a per-MESSAGE check and + # zero_loss is a per-COUNT one, so they can disagree, and each disagreement is informative. + # It is strictly ADDITIONAL -- it can fail a step whose counts reconciled, because a + # confirmed-then-absent message that happened to be excused as a timeout passes the count + # check today. PROBE_UNUSABLE does NOT fail here: an unanswerable instrument is not evidence + # of a defect in either direction, and it is reported as its own observation instead. + suspect = [r for r in records if r.intake_audit.engine_suspect] + # THE SCOPE TRAVELS WITH THE VERDICT, and that is not decoration here. `_evaluate_slos` is + # SHARED with the batch-box aggregate (batchbox.py), whose records are folded from + # remote-driver reports and carry a NOT_RUN audit by construction -- those driver processes + # poll a REMOTE engine and have no store to read. A bare "clean" there would be a green + # earned by nothing at all. So the observation always names how many steps were actually + # audited, and zero-audited says so in those words instead of passing itself off as a finding + # of no defect. The per-step guarantee is asserted in tests/test_connscale_smoke.py, where + # the audit genuinely runs; this line is the operator-facing summary of it. + audited = sum(1 for r in records if r.intake_audit.conclusive) + if suspect: + observed = "; ".join( + f"{r.sweep_mode}@N={r.count} {r.intake_audit.summary()}" for r in suspect + ) + elif audited == 0: + observed = ( + f"NOT AUDITED -- 0 of {len(records)} step(s) carry an intake audit, so this says " + f"nothing about per-message intake" + ) + else: + observed = f"clean ({audited} of {len(records)} step(s) audited)" + out.append( + SloCheck("intake_audit", "no accept-ACKed message absent", observed, not suspect) + ) if slo.max_drain_seconds is not None: worst = max( (r.drain_seconds for r in records if r.drain_seconds is not None), diff --git a/harness/load/sender.py b/harness/load/sender.py index 6651d94d..2fd32e1c 100644 --- a/harness/load/sender.py +++ b/harness/load/sender.py @@ -22,6 +22,7 @@ from collections import deque from collections.abc import Callable +from harness.load.connscale.intake_audit import IntakeLedger from harness.load.corpus import Outgoing from harness.load.correlator import Correlator from harness.load.failover_track import FailoverTracker @@ -66,6 +67,7 @@ def __init__( expect_ack: bool = True, queue_max: int = 1000, tracker: FailoverTracker | None = None, + ledger: IntakeLedger | None = None, ) -> None: self._host = host self._port = port @@ -73,6 +75,16 @@ def __init__( self._m = metrics self._expect_ack = expect_ack self._tracker = tracker # failover-only: record which seqs the engine accept-ACKed + # BACKLOG #1292 intake audit: an optional PER-MESSAGE record of how each send left `_inflight` + # (a response frame was read, or the connection closed on it). Same opt-in seam as `tracker` + # above, and None by default, so the steady-state write path is unchanged when nothing wants + # it. Meaningful only with `expect_ack` -- see `_write_loop`. + self._ledger = ledger + if ledger is not None and not expect_ack: + # Refuse rather than fill a ledger that can only ever be empty: with no ACK expected no + # response frame is ever read, so every send would be unaccounted and the audit's set + # comparison would be vacuous. Loud here beats a clean-looking verdict over nothing. + raise ValueError("an intake ledger requires expect_ack (it records response frames)") self._queue: asyncio.Queue[_Job] = asyncio.Queue(maxsize=queue_max) self._inflight: deque[tuple[int, int, str, OnDone | None]] = deque() self._stop = asyncio.Event() @@ -211,7 +223,15 @@ def _on_ack(self, ack: bytes) -> None: # MLLP ACKs are in-order per connection (the engine ACKs on receipt in send order). _seq, send_ns, _cid, on_done = self._inflight.popleft() self._m.ack.record(float(ack_ns - send_ns)) - if _ack_code(ack) in _ACCEPT: + # ONE accept decision, made here and PASSED to the ledger rather than re-derived beside it: + # a second copy of `_ACCEPT` in the audit module is a two-place constant, and the two + # disagreeing would silently move a message between the engine-suspect and harness-suspect + # buckets -- the exact attribution the audit exists to get right. + code = _ack_code(ack) + accepted = code in _ACCEPT + if self._ledger is not None: + self._ledger.record_confirmed(_cid, _seq, code, accepted=accepted) + if accepted: self._m.counters.acked += 1 if self._tracker is not None: # An accept-ACK means the engine durably committed this to the ingress stage (ACK-on- @@ -228,6 +248,8 @@ def _fail_inflight(self) -> None: return for _seq, _send_ns, _cid, on_done in self._inflight: self._m.counters.timeouts += 1 + if self._ledger is not None: + self._ledger.record_unconfirmed(_cid, _seq) if on_done is not None: on_done() self._inflight.clear() diff --git a/tests/test_connscale_intake_audit.py b/tests/test_connscale_intake_audit.py new file mode 100644 index 00000000..394bb395 --- /dev/null +++ b/tests/test_connscale_intake_audit.py @@ -0,0 +1,849 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""The BACKLOG #1292 intake audit -- the per-message discriminator for an ``engine_read`` shortfall. + +THE DEFECT UNDER TEST is an ATTRIBUTION defect, not a counting one. ``connscale``'s no-loss reconcile +fails with ``engine_read {read} < confirmed sent {sent - excused} (lost N on intake)``, and that +sentence is produced identically by an engine that lost an acknowledged message and by a harness +gauge that was sampled early or summed short. So the assertions here are about which VERDICT a given +world produces, and the decisive test is that three worlds which are indistinguishable to the count +check produce three DIFFERENT verdicts here. + +The three planted worlds mirror the three ways this can go, and they are deliberately not variations +of one: + +* the rows ARE all there, a shortfall is reported, and + part of it survives setting the never-confirmed + sends aside -> SAMPLING_LAG (harness/instrument) +* an ACCEPT-ACKed row is genuinely absent -> INVARIANT_SUSPECT (the engine branch) +* the probe's own read comes back empty -> PROBE_UNUSABLE (the null guard) + +The third is not optional. Without it a broken query renders as "every message is missing", which is +the worst possible false positive to hang a P1 on -- a catastrophic-looking engine finding produced +entirely by the instrument. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest + +from harness.load.connscale.intake_audit import ( + MOMENT_LIVE, + MOMENT_POST_MORTEM, + VERDICT_CORRELATION_SUSPECT, + VERDICT_INTAKE_COMPLETE, + VERDICT_INVARIANT_SUSPECT, + VERDICT_NOT_RUN, + VERDICT_PROBE_UNUSABLE, + VERDICT_SAMPLING_LAG, + VERDICT_UNCONFIRMED_SHORTFALL, + IntakeAudit, + IntakeLedger, + StoreSnapshot, + judge, + not_run, + run_intake_audit, + sweep_store, +) +from harness.load.connscale.profile import load_connscale_profile_text +from harness.load.connscale.report import ConnScaleRecord, SloCheck +from harness.load.connscale.runner import ( + _build_record, + _evaluate_slos, + _read_shortfall, + _reconcile, +) +from harness.load.enginepoll import EnginePoller, EngineSample +from harness.load.metrics import Counters, Histogram +from messagefoundry.store.store import MessageStore + + +def _ledger(*, accepted: int = 3, rejected: int = 0, unconfirmed: int = 0) -> IntakeLedger: + """A ledger shaped like one real step: ``accepted`` AA sends, ``rejected`` AE sends, and + ``unconfirmed`` sends stranded at a connection close.""" + led = IntakeLedger() + seq = 0 + for _ in range(accepted): + led.record_confirmed(f"CID{seq:04d}", seq, "AA", accepted=True) + seq += 1 + for _ in range(rejected): + led.record_confirmed(f"CID{seq:04d}", seq, "AE", accepted=False) + seq += 1 + for _ in range(unconfirmed): + led.record_unconfirmed(f"CID{seq:04d}", seq) + seq += 1 + return led + + +def _all_ids(led: IntakeLedger) -> frozenset[str]: + return frozenset({*led.confirmed, *led.unconfirmed}) + + +# --- PLANT A: the rows are all there, yet the count check reported a shortfall ------------------- + + +def test_plant_a_full_store_with_shortfall_is_sampling_lag() -> None: + led = _ledger(accepted=5) + snap = StoreSnapshot(_all_ids(led), total=5) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=2) + + assert audit.verdict == VERDICT_SAMPLING_LAG + # Nothing is claimed lost -- the whole point is that the shortfall is in the gauge. + assert audit.missing_accepted_total == 0 + assert audit.read_short == 2 and audit.store_total == 5 + assert "engine_read gauge" in audit.detail + assert not audit.engine_suspect and audit.conclusive + + +# --- PLANT B: an accept-ACKed message is genuinely absent --------------------------------------- + + +def test_plant_b_absent_accepted_message_is_invariant_suspect_and_names_it() -> None: + led = _ledger(accepted=5) + present = frozenset(cid for cid in led.confirmed if cid != "CID0002") + snap = StoreSnapshot(present, total=4) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=1) + + assert audit.verdict == VERDICT_INVARIANT_SUSPECT + assert audit.engine_suspect + assert audit.missing_accepted_total == 1 + # NAMED, so the finding is reproducible rather than statistical -- by SEQUENCE NUMBER, because + # the artifact rule for this family forbids control-id lists. + assert audit.missing_accepted_seqs == (2,) + assert audit.missing_codes == ("AA",) + assert "count-and-log invariant" in audit.detail + + +def test_plant_b_verdict_differs_from_plant_a_on_the_same_shortfall() -> None: + """THE ITEM, in one assertion: two worlds the count check cannot tell apart. + + Both have ``sent=5`` and a reported shortfall, so both produce the SAME + ``engine_read ... < confirmed sent ...`` message today. The audit separates them, and separates + them into the two branches that have opposite owners. + """ + led = _ledger(accepted=5) + lag = judge( + led, StoreSnapshot(_all_ids(led), 5), moment=MOMENT_POST_MORTEM, sent=5, read_short=1 + ) + loss = judge( + led, + StoreSnapshot(frozenset(c for c in led.confirmed if c != "CID0000"), 4), + moment=MOMENT_POST_MORTEM, + sent=5, + read_short=1, + ) + assert lag.read_short == loss.read_short == 1 # identical to the count check + assert lag.verdict != loss.verdict + assert (lag.engine_suspect, loss.engine_suspect) == (False, True) + + +# --- PLANT C: the probe itself read nothing ------------------------------------------------------ + + +def test_plant_c_empty_store_read_is_unusable_not_total_loss() -> None: + """A NULL NEEDS A MECHANISM. An empty read is what a broken query returns, and it is also what an + empty store returns; the two warrant opposite verdicts, so the empty read is refused rather than + rendered as the catastrophic reading.""" + led = _ledger(accepted=5) + snap = StoreSnapshot(frozenset(), total=0) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=5) + + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert not audit.engine_suspect + # The decisive assertion: it did NOT report five lost messages. + assert audit.missing_accepted_total == 0 + assert "NOT as 5 lost messages" in audit.detail + + +def test_the_three_plants_yield_three_distinct_verdicts() -> None: + """Two plants that agree are not two directions. Enumerated, so a future edit that collapses two + of these paths into one fails here rather than quietly halving the instrument.""" + led = _ledger(accepted=4) + verdicts = { + judge( + led, StoreSnapshot(_all_ids(led), 4), moment=MOMENT_POST_MORTEM, sent=4, read_short=1 + ).verdict, + judge( + led, + StoreSnapshot(frozenset(list(led.confirmed)[1:]), 3), + moment=MOMENT_POST_MORTEM, + sent=4, + read_short=1, + ).verdict, + judge( + led, StoreSnapshot(frozenset(), 0), moment=MOMENT_POST_MORTEM, sent=4, read_short=1 + ).verdict, + } + assert verdicts == {VERDICT_SAMPLING_LAG, VERDICT_INVARIANT_SUSPECT, VERDICT_PROBE_UNUSABLE} + + +# --- the fourth outcome: a rejected send is not an engine finding -------------------------------- + + +def test_absent_rejected_message_is_correlation_suspect_not_loss() -> None: + """Several NAK limbs record their ``messages`` row with a NULL control id (they run before an + MSH-10 has been parsed), so a rejected send is EXPECTED to be unmatchable by control id. Reading + that as intake loss would manufacture a P1 out of correct engine behaviour.""" + led = _ledger(accepted=3, rejected=1) + present = frozenset(cid for cid, rec in led.confirmed.items() if rec.accepted) + snap = StoreSnapshot(present, total=3) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=4, read_short=1) + + assert audit.verdict == VERDICT_CORRELATION_SUSPECT + assert not audit.engine_suspect + assert audit.missing_rejected_total == 1 and audit.missing_accepted_total == 0 + assert audit.missing_codes == ("AE",) + + +def test_a_rejected_send_that_IS_stored_is_not_a_correlation_finding() -> None: + """The negative control for the branch above, and it is the one that pins the MEMBERSHIP test. + + The NULL-control-id NAK limbs are only SOME of them: a limb that rejects AFTER parsing MSH-10 + writes a row that DOES carry the id, and the sweep finds it. Without this, dropping the + ``cid not in store_ids`` half of the predicate -- leaving a bare ``not rec.accepted`` -- kept the + whole suite green while turning every ordinary NAK into a standing CORRELATION_SUSPECT that + ``report.py`` prints on each clean step, and which the smoke assertion cannot catch because that + verdict is conclusive and not engine_suspect. This is the false-alarm direction. + """ + led = _ledger(accepted=3, rejected=1) + audit = judge( + led, StoreSnapshot(_all_ids(led), total=4), moment=MOMENT_POST_MORTEM, sent=4, read_short=0 + ) + + assert audit.verdict == VERDICT_INTAKE_COMPLETE + assert audit.missing_rejected_total == 0 and audit.missing_accepted_total == 0 + + +# --- the LEDGER-side positive control, and the shortfall it must not misattribute ---------------- + + +def test_a_ledger_with_nothing_confirmed_is_unusable_not_clean() -> None: + """THE BLIND-BUT-GREEN CASE, and the one the store-side control could not see. + + ``confirmed`` is the only set compared; ``unconfirmed`` is read as a count and never searched. + So a step where no send was ever ACKed compares an EMPTY set, and every guard keyed on + ``ledger.total`` -- which counts both -- waves it through. It is reachable on the harness's own + headline fault: the runner's excusal clamps ``excused`` to 0 once the unconfirmed count exceeds + its budget, so such a step arrives here with a large ``read_short``. + + Before the ledger-side control this returned a CONCLUSIVE verdict stating the shortfall was + "not in intake" -- computed over zero elements, rendering a green SLO row, and passing all four + smoke assertions -- on precisely the step the reconcile calls a possible accepted-and-dropped. + """ + led = _ledger(accepted=0, unconfirmed=5) + snap = StoreSnapshot(frozenset({"OTHER0", "OTHER1", "OTHER2"}), total=3) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=2) + + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert "NO send was ever confirmed" in audit.detail + # The properties that actually protect the run: an empty comparison must not be readable as an + # answer, and must never clear the engine. + assert not audit.conclusive + assert not audit.engine_suspect + + +def test_a_shortfall_inside_the_unconfirmed_set_does_not_accuse_the_gauge() -> None: + """A shortfall made ENTIRELY of never-confirmed sends implicates neither intake nor the gauge. + + ``_excusal`` clamps ``excused`` to 0 when the unconfirmed count exceeds its budget, so + ``read_short`` then carries sends the engine may never have received. Calling that SAMPLING_LAG + accused an ``engine_read`` gauge that matched the store exactly, and the sentence was appended + verbatim to the reconcile's own "systemic no-ACK fault" line -- one failure message contradicting + itself and pointing at an enginepoll bug that does not exist. + """ + led = _ledger(accepted=2, unconfirmed=8) + stored = frozenset(led.confirmed) + # The OVER-BUDGET world: excusal clamped to 0, so read_short = sent - read = 8, and once the 8 + # never-confirmed sends are set aside nothing is unexplained (10 - 8 - 2 = 0). + audit = judge( + led, + StoreSnapshot(stored, total=2), + moment=MOMENT_POST_MORTEM, + sent=10, + read_short=8, + unexplained_short=0, + ) + + assert audit.verdict == VERDICT_UNCONFIRMED_SHORTFALL + assert audit.conclusive and not audit.engine_suspect + # The regression guard keys on the ACCUSATION, not on the words "engine_read gauge" -- this + # detail names the gauge inside a NEGATION ("implicates NEITHER intake NOR the engine_read + # gauge"), so a bare substring test would answer a different question than the one asked. + # "sample attribution" is the diagnosis unique to SAMPLING_LAG, and is what must be absent. + assert "sample attribution" not in audit.detail + assert "never-confirmed" in audit.detail + + +def test_a_shortfall_larger_than_the_unconfirmed_set_still_names_the_gauge() -> None: + """The complement, so the split above cannot be satisfied by never returning SAMPLING_LAG. + + One more missing than the never-confirmed sends can account for, with every confirmed send + present, leaves a remainder nothing else explains -- and THAT is a real gauge finding. + """ + led = _ledger(accepted=2, unconfirmed=8) + audit = judge( + led, + StoreSnapshot(frozenset(led.confirmed), total=2), + moment=MOMENT_POST_MORTEM, + sent=10, + read_short=9, + unexplained_short=1, + ) + + assert audit.verdict == VERDICT_SAMPLING_LAG + assert "sample attribution" in audit.detail + + +def test_an_IN_BUDGET_shortfall_is_a_gauge_finding_even_though_sends_went_unconfirmed() -> None: + """THE REGRESSION THAT AN INFERRED PREDICATE GETS BACKWARDS, and it is the COMMON path. + + A first cut at the split above asked ``read_short <= len(ledger.unconfirmed)`` and read a True as + "the excusal was clamped". That inference only holds OVER budget. In budget ``excused == + unconfirmed``, so the never-confirmed sends are ALREADY subtracted out of ``read_short`` and any + residue is confirmed sends the gauge did not count -- a genuine SAMPLING_LAG. Because + over-budget needs ``timeouts > 3/4 sent``, the in-budget world here is the ordinary one, so the + inferred predicate silenced a real gauge finding on the path most runs take. + + Numbers are the real arithmetic: sent=100, timeouts=5, engine_read=93. ``_excusal`` is in budget + (5 <= max(24, 75)) so ``excused``=5 and ``read_short`` = 100-5-93 = 2, while the unconfirmed + count is 5 -- and 2 <= 5, which is exactly the shape the bad predicate accepted. The producer's + ``unexplained`` = 100-5-93 = 2 is positive, so the gauge is correctly named. + """ + led = _ledger(accepted=95, unconfirmed=5) + audit = judge( + led, + StoreSnapshot(frozenset(led.confirmed), total=95), + moment=MOMENT_POST_MORTEM, + sent=100, + read_short=2, + unexplained_short=2, + ) + + assert audit.verdict == VERDICT_SAMPLING_LAG + assert "sample attribution" in audit.detail + + +def test_the_unexplained_remainder_comes_from_the_producer_not_the_ledger() -> None: + """The two worlds are INDISTINGUISHABLE from inside judge(), which is why it must not guess. + + Identical ledger, identical store, identical ``read_short`` -- only the producer's + ``unexplained_short`` differs, and the verdict flips. That is the whole argument for passing it: + no function of the ledger alone could separate these two. + """ + + def _verdict(unexplained: int) -> str: + led = _ledger(accepted=2, unconfirmed=8) + return judge( + led, + StoreSnapshot(frozenset(led.confirmed), total=2), + moment=MOMENT_POST_MORTEM, + sent=10, + read_short=8, + unexplained_short=unexplained, + ).verdict + + assert _verdict(0) == VERDICT_UNCONFIRMED_SHORTFALL + assert _verdict(3) == VERDICT_SAMPLING_LAG + + +def test_only_the_post_mortem_moment_states_the_engine_conclusion() -> None: + """The same absent row concludes DIFFERENT things at the two moments, and the text must say so. + + ``sweep_store`` pages ``ORDER BY received_at DESC`` + OFFSET, so a row committed while a LIVE + sweep walks shifts the window and a genuinely present row can go unread -- documented as known + and deliberate, and it manufactures exactly this verdict. The machine gates already read only the + post-mortem, but the live detail is printed to the console and stored in the JSON artifact, where + an unhedged "the count-and-log invariant would not hold" is quotable as an engine finding that + the post-mortem beside it may not support. + """ + led = _ledger(accepted=3) + snap = StoreSnapshot(frozenset(list(led.confirmed)[1:]), total=2) + + live = judge(led, snap, moment=MOMENT_LIVE, sent=3, read_short=1) + post = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=3, read_short=1) + + # Both SEE it -- the hedge must not suppress the finding, only its conclusion. + assert live.verdict == post.verdict == VERDICT_INVARIANT_SUSPECT + assert live.missing_accepted_total == post.missing_accepted_total == 1 + + assert "count-and-log invariant would not hold" in post.detail + assert "STOPPED" in post.detail + assert "count-and-log invariant would not hold" not in live.detail + assert "post-mortem reproduces it" in live.detail + + # The MACHINE surface must agree with the prose by construction, not because the SLO gate + # happens to read the post-mortem field. A live sweep can manufacture this verdict; only the + # post-mortem may carry it into `engine_suspect`, which is what fails the run. + assert post.engine_suspect and not live.engine_suspect + + +# --- the positive controls ------------------------------------------------------------------------ + + +def test_clean_run_reports_late_unconfirmed_as_its_positive_control() -> None: + """``late_unconfirmed`` proves the sweep sees BEYOND the confirmed set: an excused send that + nevertheless arrived. A sweep that only ever returned the confirmed ids would score clean here + and would be blind to exactly the messages the reconcile forgives.""" + led = _ledger(accepted=3, unconfirmed=1) + snap = StoreSnapshot(_all_ids(led), total=4) + + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=4, read_short=0) + + assert audit.verdict == VERDICT_INTAKE_COMPLETE + assert audit.late_unconfirmed_total == 1 + assert audit.unconfirmed_total == 1 and audit.confirmed_total == 3 + assert audit.store_total == 4 + + +def test_empty_ledger_against_real_sends_is_unusable() -> None: + """The sender-side positive control. An audit over a ledger that recorded nothing is vacuously + clean, so it is refused.""" + audit = judge( + IntakeLedger(), StoreSnapshot(frozenset(), 0), moment=MOMENT_LIVE, sent=7, read_short=1 + ) + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert "recorded NOTHING against 7" in audit.detail + + +def test_zero_send_step_is_complete_not_unusable() -> None: + """A step that sent nothing has nothing to audit and is not an instrument failure.""" + audit = judge( + IntakeLedger(), StoreSnapshot(frozenset(), 0), moment=MOMENT_LIVE, sent=0, read_short=0 + ) + assert audit.verdict == VERDICT_INTAKE_COMPLETE + + +# --- the partial-ledger split: a positive finding survives it, a null does not -------------------- + + +def test_partial_ledger_makes_a_NULL_unusable() -> None: + led = _ledger(accepted=3) + snap = StoreSnapshot(_all_ids(led), total=3) + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=0) + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert "accounted 3 of 5" in audit.detail + + +def test_partial_ledger_does_not_suppress_a_POSITIVE_finding() -> None: + """The mirror image, and the reason the two guards sit on opposite sides of the finding checks: a + short ledger under-reports, so a message inside it that is genuinely absent is still absent.""" + led = _ledger(accepted=3) + snap = StoreSnapshot(frozenset(list(led.confirmed)[1:]), total=2) + audit = judge(led, snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=3) + assert audit.verdict == VERDICT_INVARIANT_SUSPECT + assert "LOWER BOUND" in audit.detail + + +def test_ledger_overflow_and_duplicates_are_unusable() -> None: + small = IntakeLedger(capacity=2) + for i in range(4): + small.record_confirmed(f"C{i}", i, "AA", accepted=True) + assert small.overflow == 2 + # The snapshot trips NO OTHER GUARD -- rows present, nothing truncated, no error, and every id + # the ledger did manage to record IS in the store -- and the DETAIL is asserted, not just the + # verdict. Both matter: with the overflow guard deleted this world still reaches + # PROBE_UNUSABLE via the partial-ledger guard ("accounted 2 of 4"), so a verdict-only assertion + # passed with the guard under test entirely removed. Overflow is step 1 of the blindness + # ordering, and a step whose test cannot fail is not covering it. + snap = StoreSnapshot(frozenset(small.confirmed), total=len(small.confirmed)) + overflowed = judge(small, snap, moment=MOMENT_LIVE, sent=4, read_short=0) + assert overflowed.verdict == VERDICT_PROBE_UNUSABLE + assert "overflowed" in overflowed.detail + + dup = IntakeLedger() + dup.record_confirmed("SAME", 0, "AA", accepted=True) + dup.record_confirmed("SAME", 1, "AA", accepted=True) + assert dup.duplicates == 1 + audit = judge( + dup, StoreSnapshot(frozenset({"SAME"}), 1), moment=MOMENT_LIVE, sent=2, read_short=0 + ) + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert "duplicate control id" in audit.detail + + +def test_not_run_is_neither_a_pass_nor_a_finding() -> None: + audit = not_run("audit disabled") + assert audit.verdict == VERDICT_NOT_RUN + assert not audit.conclusive and not audit.engine_suspect + + +# --- run_intake_audit: a broken reader is a probe outcome, never a run failure -------------------- + + +def test_reader_exception_becomes_probe_unusable() -> None: + led = _ledger(accepted=2) + + async def _boom() -> StoreSnapshot: + raise RuntimeError("no such table: messages") + + audit = asyncio.run( + run_intake_audit(led, _boom, moment=MOMENT_POST_MORTEM, sent=2, read_short=2) + ) + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert "RuntimeError" in audit.detail and "no such table" in audit.detail + assert audit.missing_accepted_total == 0 + + +# --- the REAL store sweep, against a real SQLite store ------------------------------------------- + + +def test_sweep_reads_control_ids_from_a_real_store(tmp_path: Path) -> None: + """RUN THE THING: the reader against a real ``MessageStore``, not a stub. + + Includes its own positive control -- a control id that was never inserted must NOT come back -- + because a sweep that returned everything asked of it would pass a membership test without ever + querying anything. + """ + + async def go() -> None: + store = await MessageStore.open(tmp_path / "sweep.db") + try: + for i in range(3): + await store.enqueue_ingress( + channel_id="IB_CS_00000", + raw=f"MSH|^~\\&|A|B|C|D|20260101||ADT^A01|SWEEP{i:04d}|P|2.5\r", + control_id=f"SWEEP{i:04d}", + message_type="ADT^A01", + ) + snap = await sweep_store(store, row_cap=1000) + assert snap.total == 3 and not snap.truncated and snap.error is None + assert snap.control_ids == {"SWEEP0000", "SWEEP0001", "SWEEP0002"} + assert "SWEEP9999" not in snap.control_ids # the sweep discriminates, it does not echo + + # And the cap: a table bigger than the cap is TRUNCATED, never a short set that would + # read as absence for every row the sweep did not reach. + capped = await sweep_store(store, row_cap=2) + assert capped.truncated and capped.total == 3 and capped.control_ids == frozenset() + assert ( + judge( + _ledger(accepted=3), capped, moment=MOMENT_POST_MORTEM, sent=3, read_short=1 + ).verdict + == VERDICT_PROBE_UNUSABLE + ) + finally: + await store.close() + + asyncio.run(go()) + + +def test_sweep_of_an_empty_real_store_is_refused_by_judge(tmp_path: Path) -> None: + """The end-to-end null guard: a REAL sweep of a REAL empty store returns the same empty set a + broken query would, and ``judge`` refuses it rather than reporting total loss.""" + + async def go() -> None: + store = await MessageStore.open(tmp_path / "empty.db") + try: + snap = await sweep_store(store, row_cap=1000) + assert snap.total == 0 and snap.control_ids == frozenset() and snap.error is None + audit = judge( + _ledger(accepted=2), snap, moment=MOMENT_POST_MORTEM, sent=2, read_short=2 + ) + assert audit.verdict == VERDICT_PROBE_UNUSABLE + assert audit.missing_accepted_total == 0 + finally: + await store.close() + + asyncio.run(go()) + + +# --- the SENDER seam, driven over a real socket --------------------------------------------------- + + +def test_sender_ledger_records_both_exits_over_a_real_socket() -> None: + """RUN THE THING at the other end: a real :class:`PersistentConnection` against a real MLLP + listener that ACKs two frames, NAKs one, and then closes on a fourth without answering. + + All three ledger states have to be reachable from the actual sender, not just constructible: the + audit's arithmetic assumes CONFIRMED is exactly ``sent - excused``, and that assumption is only + worth anything if ``_on_ack`` and ``_fail_inflight`` both feed it. + """ + from harness.load.corpus import Outgoing + from harness.load.correlator import Correlator + from harness.load.metrics import Counters, Histogram, LiveMetrics + from harness.load.sender import PersistentConnection + from messagefoundry.transports.mllp import MLLPDecoder, frame + + ledger = IntakeLedger() + metrics = LiveMetrics(Counters(), Histogram(), Histogram()) + + async def go() -> None: + answered = 0 + + async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + nonlocal answered + decoder = MLLPDecoder() + while True: + chunk = await reader.read(65536) + if not chunk: + break + for _msg in decoder.feed(chunk): + answered += 1 + if answered > 3: + # The fourth frame is swallowed and the socket dropped: the send is left in + # `_inflight` and must land in the ledger as UNCONFIRMED. + writer.close() + return + code = "AA" if answered <= 2 else "AE" + writer.write( + frame(f"MSH|^~\\&|E|E|H|H|20260101||ACK|A{answered}|P|2.5\rMSA|{code}|X\r") + ) + await writer.drain() + + server = await asyncio.start_server(handle, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + correlator = Correlator(1000, metrics) + conn = PersistentConnection( + "127.0.0.1", port, correlator, metrics, expect_ack=True, ledger=ledger + ) + conn.start() + for i in range(4): + await conn.submit( + Outgoing( + seq=i, + code="ADT", + control_id=f"LG{i:04d}", + payload=f"MSH|^~\\&|A|B|C|D|20260101||ADT^A01|LG{i:04d}|P|2.5\r", + ) + ) + # Give the exchange time to complete, then stop (which sweeps whatever is still in flight). + for _ in range(200): + if ledger.total >= 4: + break + await asyncio.sleep(0.01) + await conn.stop(0.2) + server.close() + await server.wait_closed() + + asyncio.run(go()) + + accepted = {cid for cid, rec in ledger.confirmed.items() if rec.accepted} + rejected = {cid for cid, rec in ledger.confirmed.items() if not rec.accepted} + assert accepted == {"LG0000", "LG0001"}, ledger.confirmed + assert rejected == {"LG0002"}, ledger.confirmed + assert set(ledger.unconfirmed) == {"LG0003"}, dict(ledger.unconfirmed) + # The accounting identity the audit's arithmetic rests on, measured rather than assumed. + c = metrics.counters + assert c.sent == c.acked + c.nak + c.timeouts == ledger.total == 4 + assert not ledger.overflow and not ledger.duplicates + + +def test_ledger_requires_expect_ack() -> None: + """A ledger with no response frames to record can only ever be empty, and an empty ledger scores + vacuously clean. Refused loudly at construction instead.""" + from harness.load.correlator import Correlator + from harness.load.metrics import Counters, Histogram, LiveMetrics + from harness.load.sender import PersistentConnection + + metrics = LiveMetrics(Counters(), Histogram(), Histogram()) + with pytest.raises(ValueError, match="expect_ack"): + PersistentConnection( + "127.0.0.1", + 1, + Correlator(10, metrics), + metrics, + expect_ack=False, + ledger=IntakeLedger(), + ) + + +# --- the runner wiring: one definition of the shortfall, and it reaches the artifact -------------- + + +def _sample(read: int) -> EngineSample: + return EngineSample( + elapsed_s=0.0, + pending=0, + inflight=0, + done=0, + dead=0, + read=read, + written=0, + out_dead=0, + queue_depth=0, + in_pipeline=0, + db_size_bytes=0, + journal_mode="wal", + synchronous="normal", + uptime_s=0.0, + ) + + +def test_read_shortfall_is_the_same_number_the_reconcile_prints() -> None: + """ONE DEFINITION, asserted rather than assumed. + + The audit exists to explain ``engine_read N < confirmed sent M (lost K on intake)``, so it has to + fire on exactly that ``K``. A second copy of the unconfirmed-send excusal beside it would let the + audit trigger on a shortfall the step does not report, or stay silent on one it does -- and either + way it would be attributing the wrong failure. The excusal is deliberately non-trivial here + (``timeouts`` inside the budget, so some sends ARE excused), so a version that ignored it would + not agree by accident. + """ + c = Counters(sent=40, timeouts=4) + base, final = _sample(0), _sample(30) + + short = _read_shortfall(c, base, final, unconfirmed_budget=8) + no_loss = _reconcile(c, base, final, unconfirmed_budget=8) + + assert short == 6 # 40 sent - 4 excused - 30 read + assert not no_loss.ok + # The number the failing message actually carries, read back out of the message itself. + assert f"(lost {short} on intake)" in no_loss.detail + assert "confirmed sent 36" in no_loss.detail + + +def test_read_shortfall_is_zero_without_engine_gauges() -> None: + """No samples means no shortfall to ATTRIBUTE. ``_reconcile`` fails the step on its own for that, + and the audit must not invent a finding out of a missing measurement.""" + assert _read_shortfall(Counters(sent=10), None, _sample(0), unconfirmed_budget=1) == 0 + assert _read_shortfall(Counters(sent=10), _sample(0), None, unconfirmed_budget=1) == 0 + + +def _record_with(audit: IntakeAudit, *, read: int, sent: int) -> ConnScaleRecord: + poller = EnginePoller("http://127.0.0.1:1", token=None, origin=0.0) + poller._samples = [_sample(0), _sample(read)] + return _build_record( + claim_mode="per_lane", + fuse_mode=False, + batch_mode=False, + mode="fixed_aggregate", + count=4, + aggregate_rate=10.0, + metrics_counters=Counters(sent=sent), + ack_hist=Histogram(), + poller=poller, + samples=[], + drain_seconds=1.0, + reload_seconds=None, + audit_live=not_run("not triggered", moment=MOMENT_LIVE), + audit_final=audit, + ) + + +def test_a_failing_reconcile_carries_the_audit_verdict_into_its_own_message() -> None: + """The deliverable: a CI reader gets the attribution WITHOUT re-running anything. + + ``no_loss.ok`` is untouched -- the count check still fails the step exactly as before -- but its + detail, which is what the smoke's assertion message prints, now says WHICH branch it was. + """ + led = _ledger(accepted=4) + audit = judge( + led, + StoreSnapshot(frozenset(list(led.confirmed)[1:]), 3), + moment=MOMENT_POST_MORTEM, + sent=4, + read_short=1, + ) + rec = _record_with(audit, read=3, sent=4) + + assert rec.no_loss.ok is False # unchanged: the count check still fails + assert "lost 1 on intake" in rec.no_loss.detail # the original message survives verbatim + assert "INVARIANT_SUSPECT" in rec.no_loss.detail # and now says which branch + assert "seqs=[0]" in rec.no_loss.detail + assert rec.intake_audit is audit and rec.intake_audit.engine_suspect + assert "intake_audit" in rec.to_json_dict() + + +def test_a_passing_reconcile_is_left_byte_identical() -> None: + """No verdict is appended to a detail that reports no problem: the audit rides in its own field + and on the console, and a clean step's ``no_loss`` string is unchanged from pre-#1292.""" + led = _ledger(accepted=4) + audit = judge( + led, StoreSnapshot(_all_ids(led), 4), moment=MOMENT_POST_MORTEM, sent=4, read_short=0 + ) + rec = _record_with(audit, read=4, sent=4) + assert rec.no_loss.ok + assert rec.no_loss.detail == "read>=sent, sink_received>=written, backlog drained" + + +# --- the SLO: a green must not be earned by an audit that never ran ------------------------------ + + +def _profile(intake_audit: bool = True) -> object: + flag = "true" if intake_audit else "false" + return load_connscale_profile_text( + "[connscale]\n" + 'name = "slo-it"\n' + "counts = [4]\n" + "base_port = 41000\n" + "aggregate_rate = 10.0\n" + f"intake_audit = {flag}\n" + "\n" + "[connscale.slo]\n" + "zero_loss = false\n" + ) + + +def _slo(record: ConnScaleRecord, *, enabled: bool = True) -> SloCheck | None: + checks = _evaluate_slos(_profile(enabled), [record]) # type: ignore[arg-type] + return next((c for c in checks if c.name == "intake_audit"), None) + + +def test_slo_fails_on_a_suspect_record_and_names_the_sequence_numbers() -> None: + led = _ledger(accepted=4) + audit = judge( + led, + StoreSnapshot(frozenset(list(led.confirmed)[1:]), 3), + moment=MOMENT_POST_MORTEM, + sent=4, + read_short=1, + ) + check = _slo(_record_with(audit, read=3, sent=4)) + assert check is not None and not check.ok + assert "INVARIANT_SUSPECT" in str(check.observed) and "seqs=[0]" in str(check.observed) + + +def test_slo_states_its_scope_rather_than_claiming_a_bare_clean() -> None: + """A GREEN THAT MEANS LESS, headed off. ``_evaluate_slos`` is shared with the batch-box aggregate, + whose records carry a NOT_RUN audit by construction (its driver processes poll a REMOTE engine and + have no store to read). A bare "clean" there would be a pass earned by nothing, so the observation + always names how many steps were actually audited -- and zero-audited says so in those words.""" + led = _ledger(accepted=4) + clean = judge( + led, StoreSnapshot(_all_ids(led), 4), moment=MOMENT_POST_MORTEM, sent=4, read_short=0 + ) + audited = _slo(_record_with(clean, read=4, sent=4)) + assert audited is not None and audited.ok + assert str(audited.observed) == "clean (1 of 1 step(s) audited)" + + never = _slo(_record_with(not_run("audit not wired"), read=4, sent=4)) + assert never is not None and never.ok # not a FAILURE -- but it must not read as a finding + assert "NOT AUDITED" in str(never.observed) and "0 of 1" in str(never.observed) + + +def test_slo_is_absent_when_the_profile_turns_the_audit_off() -> None: + led = _ledger(accepted=4) + clean = judge( + led, StoreSnapshot(_all_ids(led), 4), moment=MOMENT_POST_MORTEM, sent=4, read_short=0 + ) + assert _slo(_record_with(clean, read=4, sent=4), enabled=False) is None + + +# --- the sweep's short-page limb ----------------------------------------------------------------- + + +def test_sweep_reports_truncated_when_a_page_returns_fewer_rows_than_counted() -> None: + """``COUNT(*)`` promised more rows than the pages delivered. Returning the short set would read as + absence for every row the sweep never reached -- the same catastrophic false positive the + empty-read guard exists for, one page further in.""" + + class _ShortStore: + async def count_messages(self) -> int: + return 5 + + async def list_messages(self, *, limit: int, offset: int) -> list[dict[str, object]]: + return [{"control_id": "A"}] if offset == 0 else [] + + snap = asyncio.run(sweep_store(_ShortStore(), row_cap=100)) + assert snap.truncated and snap.total == 5 and snap.control_ids == frozenset({"A"}) + verdict = judge( + _ledger(accepted=5), snap, moment=MOMENT_POST_MORTEM, sent=5, read_short=4 + ).verdict + assert verdict == VERDICT_PROBE_UNUSABLE diff --git a/tests/test_connscale_smoke.py b/tests/test_connscale_smoke.py index f9668e3f..90cbd861 100644 --- a/tests/test_connscale_smoke.py +++ b/tests/test_connscale_smoke.py @@ -23,6 +23,7 @@ import pytest +from harness.load.connscale.intake_audit import MOMENT_POST_MORTEM from harness.load.connscale.probe import ProbeDegraded from harness.load.connscale.profile import load_connscale_profile_text from harness.load.connscale.report import ConnScaleRecord, ConnScaleReport @@ -173,6 +174,44 @@ def _assert_fd_probe(records: Sequence[ConnScaleRecord]) -> None: ) +def _assert_intake_audit(records: Sequence[ConnScaleRecord]) -> None: + """Assert the BACKLOG #1292 discriminator on every step, in the order its verdicts matter. + + Four properties, each pinning a different way this could go quietly wrong: + + 1. NO step is ``engine_suspect``. This is the finding the item is about: the engine framed an + accept-ACK for a message and its own stopped, committed store has no row for it. + 2. EVERY step's audit is CONCLUSIVE. A PROBE_UNUSABLE result is not a pass -- it means the + instrument could not answer, and an instrument that silently stops answering leaves the + original unattributable failure in place while looking green. + 3. EVERY step's sweep read rows (``store_total > 0``) against a step that sent messages. This is + the positive control: a set comparison against an empty set is clean for the wrong reason. + 4. The authoritative audit is the POST-MORTEM one. A live read could be explained away as early + sampling; a read of a stopped engine's store cannot, and mislabelling one as the other would + destroy the only distinction the second moment exists to make. + """ + for r in records: + audit = r.intake_audit + assert not audit.engine_suspect, ( + f"COUNT-AND-LOG INVARIANT -- {r.sweep_mode}@N={r.count}: {audit.summary()}. The engine " + f"accept-ACKed those sends and its own stopped, committed store has no messages row for " + f"them, so a deploying site would be able to lose an acknowledged message at intake. " + f"The sequence numbers above name them; this is reproducible, not statistical." + ) + assert audit.conclusive, ( + f"INTAKE AUDIT COULD NOT ANSWER -- {r.sweep_mode}@N={r.count}: {audit.summary()}. The " + f"discriminator is the whole point of this step's no-loss coverage, so a probe that did " + f"not run or could not read is reported as a failure rather than tolerated: tolerating " + f"it restores exactly the unattributable red this check exists to replace." + ) + assert audit.moment == MOMENT_POST_MORTEM, audit + assert audit.store_total > 0, ( + f"INTAKE AUDIT POSITIVE CONTROL -- {r.sweep_mode}@N={r.count} sent {r.sent} message(s) " + f"and the store sweep read {audit.store_total} row(s): {audit.summary()}. A clean set " + f"comparison over an empty read says nothing about intake." + ) + + # --- the one expensive run, shared by every property below ---------------------------------------- @@ -259,6 +298,19 @@ def test_no_loss_reconciles_at_every_step(smoke_report: ConnScaleReport) -> None assert r.no_loss.ok, (r.sweep_mode, r.count, r.no_loss.detail) +def test_no_accept_acked_message_is_absent_from_the_stopped_engines_store( + smoke_report: ConnScaleReport, +) -> None: + """BACKLOG #1292 -- the PER-MESSAGE intake audit, asserted INDEPENDENTLY of the count check above. + + Strictly ADDITIONAL, not a restatement: the count check compares totals and forgives an + unconfirmed send, so a message the engine ACCEPT-ACKed and then has no row for passes it whenever + that message also happened to be excused as a timeout. This asserts the thing the count check + cannot: that no accept-ACKed message is absent from the engine's own committed store. + """ + _assert_intake_audit(smoke_report.records) + + def test_the_fd_and_empty_claim_curves_are_monotonic_in_n(smoke_report: ConnScaleReport) -> None: """Curve monotonicity smoke (a LOOSE >= per mode; CI runners are noisy): FD count + empty-claims at N=24 >= N=12. Asserted via the report's monotonicity SLOs.