From 07188f353b9ea550bd24a917c3e9bb63a3e5f172 Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Thu, 20 Aug 2026 15:49:01 -0500 Subject: [PATCH 1/2] wip(1292): in-flight intake-audit discriminator, committed under USAGE URGENT STOP NOT A FINISHED CHANGE AND NOT REVIEWED. Committed mid-flight because the usage ladder reached URGENT STOP, where the instruction is to commit whatever exists rather than risk losing a tree. A partial commit beats a lost one; an unreviewed commit on an unpushed branch costs nothing and can be amended. WHAT THIS IS: the BACKLOG #1292 store-side discriminator, being built to separate a harness reconciliation race from a real count-and-log intake loss. Its design stage returned confidence NOT-ESTABLISHED, which was the honest answer and the one I told the agent was acceptable -- the deliverable is an instrument that ATTRIBUTES the next occurrence, not a verdict on which branch is real. STATE, stated plainly so nobody mistakes this for finished work: - the implement stage was STILL RUNNING when this was committed - the ADVERSARIAL REVIEW STAGE HAD NOT RUN AT ALL - I have NOT verified any of it: no red-first, no ruff, no mypy, no test run by me - the two new modules (intake_audit.py, test_connscale_intake_audit.py) may be incomplete DO NOT LAND THIS. It needs the review stage, then my own verification, then almost certainly a correction commit -- the reviews of my last two commits each found a real defect, one of which let a genuine FD collapse pass. Assume the same here until shown otherwise. Also carried, unrelated and equally unverified: whatever the agent touched in driver.py, profile.py, report.py, runner.py, sender.py, docs/LOAD-TESTING.md and harness/load/__init__.py. The ledger is deliberately untouched, as it has been in every commit on this branch. --- docs/LOAD-TESTING.md | 26 + harness/load/__init__.py | 21 +- harness/load/connscale/driver.py | 7 + harness/load/connscale/intake_audit.py | 537 ++++++++++++++++++++ harness/load/connscale/profile.py | 8 + harness/load/connscale/report.py | 40 ++ harness/load/connscale/runner.py | 231 +++++++-- harness/load/sender.py | 24 +- tests/test_connscale_intake_audit.py | 659 +++++++++++++++++++++++++ tests/test_connscale_smoke.py | 52 ++ 10 files changed, 1572 insertions(+), 33 deletions(-) create mode 100644 harness/load/connscale/intake_audit.py create mode 100644 tests/test_connscale_intake_audit.py diff --git a/docs/LOAD-TESTING.md b/docs/LOAD-TESTING.md index 4a81e48d9..dc495d2da 100644 --- a/docs/LOAD-TESTING.md +++ b/docs/LOAD-TESTING.md @@ -297,6 +297,32 @@ 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 and every confirmed send has a row anyway, so the + shortfall is in the gauge (sample attribution or per-inbound sum coverage). A harness defect. + - `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 (its own read came back empty, was truncated, or + the send ledger was incomplete). 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 609db8ef0..a0838c787 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 0926dc426..9c86eed72 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 000000000..bedabb83d --- /dev/null +++ b/harness/load/connscale/intake_audit.py @@ -0,0 +1,537 @@ +# 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. Full control ids stay in the harness log line. +""" + +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 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_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)?""" + return self.verdict == VERDICT_INVARIANT_SUSPECT + + 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 judge( + ledger: IntakeLedger, + snapshot: StoreSnapshot, + *, + moment: str, + sent: int, + read_short: int, +) -> 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. the store sweep failed or was truncated -> PROBE_UNUSABLE. + 3. the store sweep read ZERO rows against a non-empty ledger -> PROBE_UNUSABLE. This is the + positive control, and it is 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. + 4. 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. + 5. only REJECT-ACKed sends unmatched -> CORRELATION_SUSPECT (see the module docstring). + 6. the ledger did not account for every send -> PROBE_UNUSABLE, because a null over a partial + ledger proves nothing. This is step 4's mirror image, and why the two are split rather than + both being checked up front. + 7. a shortfall with every confirmed send present -> SAMPLING_LAG: the gauge is short, not intake. + 8. otherwise INTAKE_COMPLETE. + """ + confirmed = ledger.confirmed + ledger_total = ledger.total + store_ids = snapshot.control_ids + + 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 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) + + 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} -- on a deployment the " + f"count-and-log invariant would not hold for those messages" + ), + ) + 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" + ) + if read_short > 0: + return _build( + moment=moment, + verdict=VERDICT_SAMPLING_LAG, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=(), + missing_rejected=(), + late_unconfirmed=late_unconfirmed, + detail=( + f"engine_read is short by {read_short} yet all {len(confirmed)} confirmed send(s) " + f"HAVE a messages row ({snapshot.total} row(s) present) -- the shortfall is in the " + f"engine_read gauge (sample attribution or per-inbound sum coverage), not in intake" + ), + ) + return _build( + moment=moment, + verdict=VERDICT_INTAKE_COMPLETE, + read_short=read_short, + sent=sent, + ledger=ledger, + snapshot=snapshot, + missing_accepted=(), + missing_rejected=(), + late_unconfirmed=late_unconfirmed, + detail=( + 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, +) -> 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) + 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 8edc56ef8..d4bd0fd8a 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 0c0a7c8ef..193d2666e 100644 --- a/harness/load/connscale/report.py +++ b/harness/load/connscale/report.py @@ -7,6 +7,11 @@ 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 stay in the harness log line. + 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 +27,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 +174,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 +260,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 +432,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 d4ca4fb78..dca0e112f 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 ( @@ -333,33 +335,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 +466,53 @@ 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 + ) + audit_live = intake_audit.not_run( + "no intake shortfall to attribute at this moment", moment=intake_audit.MOMENT_LIVE + ) + if ledger is not None and 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, + ) + # 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("intake audit disabled for this profile") + 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, + ) return _build_record( claim_mode=claim_mode, fuse_mode=fuse_mode, @@ -475,6 +526,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 +616,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 +838,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 +849,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,6 +935,57 @@ 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) ) @@ -889,14 +1034,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 +1311,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 6651d94d4..2fd32e1c2 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 000000000..4d2998aad --- /dev/null +++ b/tests/test_connscale_intake_audit.py @@ -0,0 +1,659 @@ +# 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 while a shortfall is reported -> 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, + 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",) + + +# --- 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 + assert ( + judge( + small, StoreSnapshot(frozenset(), 0), moment=MOMENT_LIVE, sent=4, read_short=0 + ).verdict + == VERDICT_PROBE_UNUSABLE + ) + + 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 f9668e3f9..90cbd8610 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. From d23a2b6bfee035b8e9443f9eadedb45a972c55a7 Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Thu, 20 Aug 2026 20:28:00 -0500 Subject: [PATCH 2/2] fix(1292): the intake audit could clear intake over an empty comparison, and blamed the gauge by guessing The adversarial review the wip commit said this needed has now run: six dimensions, each finding handed to a refuter. 17 findings raised, 8 survived, 9 refuted as overstated or wrong. Then a quality pass found a further defect in my own fix for one of them. Every fix below is verified red-first -- the guard is broken, the test is watched to fail, the guard restored. THE HIGH ONE. judge()'s vacuity guard tested `ledger.total` (confirmed + unconfirmed) while every finding branch consumes only `ledger.confirmed`. A step where nothing was ever ACK-confirmed therefore passed all six blindness guards and returned a CONCLUSIVE verdict -- "the shortfall is not in intake" -- computed over a ZERO-ELEMENT comparison, with a green SLO row and all four smoke assertions passing. The existing positive control guards the STORE side; nothing guarded the LEDGER side. It is reachable on the harness's own headline fault, because the runner's excusal clamps `excused` to 0 over budget. This is the blind-but-green direction the whole ordering exists to prevent, and it cleared intake on precisely the step the reconcile calls a possible accepted-and-dropped. THE ONE I GOT WRONG FIRST, RECORDED BECAUSE THE SHAPE MATTERS. SAMPLING_LAG named the `engine_read` gauge by pure elimination. My first correction inferred the missing state inside judge() by asking `read_short <= len(ledger.unconfirmed)`. That inference holds only OVER budget. IN budget `excused == unconfirmed`, so the unconfirmed sends are ALREADY subtracted out of `read_short` and any residue is confirmed sends the gauge did not count -- a real finding my predicate silenced. And over-budget needs `timeouts > 3/4 sent`, so the case I broke is the COMMON one. Measured against the real producer arithmetic (sent=100, timeouts=5, read=93): old predicate `0 < 2 <= 5` fires and silences; corrected predicate names the gauge. The fix is at the producer instead -- `_unexplained_shortfall` computes `sent - unconfirmed - read`, which means the same thing in both worlds, and judge() consumes it rather than reconstructing budget arithmetic it does not own. An absent value falls back to naming the gauge, so the unsafe direction is never the default. ALSO FIXED, all confirmed by the review: - engine_suspect now requires the POST-MORTEM moment. A live sweep pages over a store still being written and can manufacture INVARIANT_SUSPECT; the prose already hedged that, but the machine surface did not, and only a distant call site's choice of field kept it honest. - the live moment reports an absent row WITHOUT the engine conclusion (`_conclusion`), so a console line or JSON artifact cannot be quoted as an invariant violation the post-mortem does not support. - the live NOT_RUN reason said "no intake shortfall to attribute" when the audit was merely DISABLED -- a false statement about the run, in the field whose only job is attribution, on exactly the step someone opens the artifact to read. - two docstrings promised full control ids "stay in the harness log line". Nothing logs one. False in both directions: a reader finds none, and a maintainer reconciling prose against code would make it true by logging exactly what the PHI rule exists to keep out. TWO TESTS THAT PASSED WITH THE FEATURE BROKEN, which is the defect class this branch has now been corrected for three times. The overflow test asserted only the verdict against an empty snapshot, so a neighbouring guard produced the same PROBE_UNUSABLE and it passed with `if ledger.overflow:` deleted outright. CORRELATION_SUSPECT had a positive test but no negative control, so dropping the store-membership half of its predicate kept the suite green while turning every ordinary NAK into a standing finding. Both now assert the discriminating detail and both go red when broken. BEHAVIOUR CHANGE, STATED SO IT IS NOT A SURPRISE: a step where nothing was ever confirmed now returns PROBE_UNUSABLE, which is not `conclusive`, so `_assert_intake_audit` FAILS there. It previously passed with a vacuous clean verdict. That is the intended direction -- the item's own text says a probe that cannot answer is a defect to classify deliberately, not a gap to excuse -- but it does mean this test can now red where it was falsely green. Quality pass also collapsed three near-identical `_build` call sites into a `_matched` closure beside the existing `_unusable` one, hoisted the duplicated counts, and gave the two audit moments one spelling of the disabled reason so they cannot drift into disagreeing about why nothing ran. VERIFIED: ruff check + format, mypy strict, 33 intake-audit tests, 241 connscale tests, and the connscale end-to-end smoke (2 passed) -- which exercises the new verdict semantics for real rather than at unit level. Red-first on all six new or repaired guards, including one that reproduces the predicate bug I introduced. Glyph scan clean over 403 added lines against a 736-hit positive control. The ledger stays untouched, as on every commit on this branch. Co-Authored-By: Claude Opus 5 --- docs/LOAD-TESTING.md | 21 ++- harness/load/connscale/intake_audit.py | 186 ++++++++++++++++------ harness/load/connscale/report.py | 4 +- harness/load/connscale/runner.py | 51 ++++++- tests/test_connscale_intake_audit.py | 204 ++++++++++++++++++++++++- 5 files changed, 403 insertions(+), 63 deletions(-) diff --git a/docs/LOAD-TESTING.md b/docs/LOAD-TESTING.md index dc495d2da..525838ea8 100644 --- a/docs/LOAD-TESTING.md +++ b/docs/LOAD-TESTING.md @@ -306,17 +306,28 @@ headroom denominator (in + out events, not messages). Exit codes match `--load`. 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 and every confirmed send has a row anyway, so the - shortfall is in the gauge (sample attribution or per-inbound sum coverage). A harness defect. + - `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 (its own read came back empty, was truncated, or - the send ledger was incomplete). Deliberately **not** rendered as "everything is missing", and - deliberately **not** a pass either. + - `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 diff --git a/harness/load/connscale/intake_audit.py b/harness/load/connscale/intake_audit.py index bedabb83d..5238b36ae 100644 --- a/harness/load/connscale/intake_audit.py +++ b/harness/load/connscale/intake_audit.py @@ -38,7 +38,12 @@ 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. Full control ids stay in the harness log line. +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 @@ -65,6 +70,13 @@ #: 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" @@ -203,14 +215,23 @@ def conclusive(self) -> bool: 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)?""" - return self.verdict == VERDICT_INVARIANT_SUSPECT + """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.""" @@ -261,6 +282,27 @@ def not_run(reason: str, *, moment: str = MOMENT_POST_MORTEM) -> IntakeAudit: ) +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, @@ -268,6 +310,7 @@ def judge( 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. @@ -276,22 +319,33 @@ def judge( 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. the store sweep failed or was truncated -> PROBE_UNUSABLE. - 3. the store sweep read ZERO rows against a non-empty ledger -> PROBE_UNUSABLE. This is the - positive control, and it is 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. - 4. an ACCEPT-ACKed send with no row -> INVARIANT_SUSPECT. Checked BEFORE the partial-ledger guard + 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. - 5. only REJECT-ACKed sends unmatched -> CORRELATION_SUSPECT (see the module docstring). - 6. the ledger did not account for every send -> PROBE_UNUSABLE, because a null over a partial - ledger proves nothing. This is step 4's mirror image, and why the two are split rather than + 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. - 7. a shortfall with every confirmed send present -> SAMPLING_LAG: the gauge is short, not intake. - 8. otherwise INTAKE_COMPLETE. + 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( @@ -322,6 +376,22 @@ def _unusable(detail: str) -> IntakeAudit: 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: @@ -352,6 +422,25 @@ def _unusable(detail: str) -> IntakeAudit: ) 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 = ( "" @@ -370,8 +459,7 @@ def _unusable(detail: str) -> IntakeAudit: 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} -- on a deployment the " - f"count-and-log invariant would not hold for those messages" + f"its own store ({snapshot.total} row(s) present){partial} -- {_conclusion(moment)}" ), ) if missing_rejected: @@ -397,37 +485,37 @@ def _unusable(detail: str) -> IntakeAudit: 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 _build( - moment=moment, - verdict=VERDICT_SAMPLING_LAG, - read_short=read_short, - sent=sent, - ledger=ledger, - snapshot=snapshot, - missing_accepted=(), - missing_rejected=(), - late_unconfirmed=late_unconfirmed, - detail=( - f"engine_read is short by {read_short} yet all {len(confirmed)} confirmed send(s) " - f"HAVE a messages row ({snapshot.total} row(s) present) -- the shortfall is in the " - f"engine_read gauge (sample attribution or per-inbound sum coverage), not in intake" - ), + 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 _build( - moment=moment, - verdict=VERDICT_INTAKE_COMPLETE, - read_short=read_short, - sent=sent, - ledger=ledger, - snapshot=snapshot, - missing_accepted=(), - missing_rejected=(), - late_unconfirmed=late_unconfirmed, - detail=( - f"all {len(confirmed)} confirmed send(s) have a messages row; {snapshot.total} row(s) " - f"present, {late_unconfirmed} excused send(s) arrived anyway" - ), + 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", ) @@ -472,6 +560,7 @@ async def run_intake_audit( moment: str, sent: int, read_short: int, + unexplained_short: int | None = None, ) -> IntakeAudit: """Read the store once through ``reader`` and judge. @@ -482,7 +571,14 @@ async def run_intake_audit( 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) + 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 diff --git a/harness/load/connscale/report.py b/harness/load/connscale/report.py index 193d2666e..075e8670d 100644 --- a/harness/load/connscale/report.py +++ b/harness/load/connscale/report.py @@ -10,7 +10,9 @@ 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 stay in the harness log line. +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 diff --git a/harness/load/connscale/runner.py b/harness/load/connscale/runner.py index dca0e112f..d3c7c0386 100644 --- a/harness/load/connscale/runner.py +++ b/harness/load/connscale/runner.py @@ -89,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). @@ -479,22 +483,35 @@ async def _run_one_step( read_short = _read_shortfall( metrics.counters, poller.baseline, poller.final, unconfirmed_budget=count ) - audit_live = intake_audit.not_run( - "no intake shortfall to attribute at this moment", moment=intake_audit.MOMENT_LIVE - ) - if ledger is not None and read_short > 0: + # 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("intake audit disabled for this profile") + audit_final = intake_audit.not_run(_AUDIT_DISABLED) if ledger is not None: with contextlib.suppress(Exception): await node.stop() @@ -512,6 +529,7 @@ async def _run_one_step( 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, @@ -989,6 +1007,29 @@ def _read_shortfall( ) +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, diff --git a/tests/test_connscale_intake_audit.py b/tests/test_connscale_intake_audit.py index 4d2998aad..394bb3955 100644 --- a/tests/test_connscale_intake_audit.py +++ b/tests/test_connscale_intake_audit.py @@ -12,7 +12,9 @@ The three planted worlds mirror the three ways this can go, and they are deliberately not variations of one: -* the rows ARE all there while a shortfall is reported -> SAMPLING_LAG (harness/instrument) +* 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) @@ -37,6 +39,7 @@ VERDICT_NOT_RUN, VERDICT_PROBE_UNUSABLE, VERDICT_SAMPLING_LAG, + VERDICT_UNCONFIRMED_SHORTFALL, IntakeAudit, IntakeLedger, StoreSnapshot, @@ -199,6 +202,189 @@ def test_absent_rejected_message_is_correlation_suspect_not_loss() -> None: 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 ------------------------------------------------------------------------ @@ -261,12 +447,16 @@ def test_ledger_overflow_and_duplicates_are_unusable() -> None: for i in range(4): small.record_confirmed(f"C{i}", i, "AA", accepted=True) assert small.overflow == 2 - assert ( - judge( - small, StoreSnapshot(frozenset(), 0), moment=MOMENT_LIVE, sent=4, read_short=0 - ).verdict - == VERDICT_PROBE_UNUSABLE - ) + # 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)