diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ab88a45..6d8179a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,24 @@ All notable changes to MessageFoundry are documented here. The format follows ## [Unreleased] +### Fixed +- **The load harness's no-loss reconcile did not enforce the `read >= sent // 2` intake guarantee + 0.3.2 documented.** The unconfirmed-send excusal is capped at `max(connections, half the run)`, but + that `max()` takes the connection count as a *floor*, and every call site passes a connection count + — so on a short, low-rate step (connscale-smoke's N=100 cell: ~105 sends, 100 connections) the count + won the max() and the intake bound degraded to `read >= 5`, the very vacuity the cap exists to + prevent. Nothing clamped the excusal to `sent` either, so `timeouts > sent` degraded it to + `read >= 0`. The half-the-run cap still decides the systemic no-ACK verdict (0.3.2's de-flake is + unchanged), and an **unconditional intake floor** the excusal cannot lower now enforces + `read >= sent // 2` in all three reconcile copies, at every call site. The estate copy also gained + the honest-reporting branch its two siblings had: it previously printed `read>=sent, …` on a + bounded-excused run whose read was demonstrably below `sent`, and its over-budget detail string now + matches theirs — a test pins the three in step, since nothing enforced the claim that they were. + *Known gap, unfixed:* the systemic no-ACK verdict is still gated on the same capped budget, so a + dead ACK path that nonetheless delivered everything still passes when `connections >= sent`; an + intake floor cannot catch a fault whose signature is a high read with no ACKs. Bounding that arm + needs its own change. + ## [0.3.2] — 2026-07-28 — Early Access A patch release for one adopter-facing defect shipped in 0.3.1, plus two gates that were passing diff --git a/harness/load/connscale/runner.py b/harness/load/connscale/runner.py index a0b1cb18..a08fd2bb 100644 --- a/harness/load/connscale/runner.py +++ b/harness/load/connscale/runner.py @@ -715,8 +715,11 @@ def _build_record( ) -> ConnScaleRecord: c = metrics_counters.snapshot() base, final = poller.baseline, poller.final - # Budget = this step's connection count: at most ~one stranded in-flight per connection is a - # plausible teardown artifact; more is a systemic no-ACK fault the reconcile must fail. + # Budget = this step's connection count. NOT "~one stranded in-flight per connection" — that + # model was retired (this sender's `_inflight` is unbounded, so stranding scales with + # rate x ACK-latency): `_reconcile` treats the count only as a small-run FLOOR under its + # 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) in_pipeline_peak = max((s.in_pipeline for s in samples), default=0) @@ -821,21 +824,39 @@ def _reconcile( # a model this sender breaks (`_inflight` is an UNBOUNDED deque; open-loop sends are paced by the # offered rate, not an ACK slot), so genuine teardown stranding scales with rate x ACK-latency, # not the connection count. Bound it as a FRACTION instead — at most half the run, floored by the - # connection count — which keeps `read >= sent // 2` ALWAYS required. See harness/load/report.py's - # copy for the full rationale; the three copies are kept in step deliberately. + # connection count for tiny runs. + # + # That cap ALONE does NOT keep `read >= sent // 2` required, though the comment here used to claim + # it did: `max()` treats the connection count as a FLOOR, and THIS call site passes the step's + # `count` verbatim (_build_record above). A short, low-rate step sends barely more than one + # message per connection — connscale-smoke's N=100 cell is ~105 sends against budget + # max(100, 52) = 100 — so the count wins the max() and the bound collapses to `read >= 5`. So the + # 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 budget = max(unconfirmed_budget, sent // 2) over_budget = unconfirmed > budget excused = 0 if over_budget else unconfirmed 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 + # this a `timeouts > sent` run would pass at `read >= 0`). Zero at sent == 0, rounds DOWN on an + # odd `sent`, so it is never stricter than the documented bound. + floor_short = sent // 2 - read deliver_short = written - sink_received drained = backlog == 0 - ok = read_short <= 0 and deliver_short <= 0 and drained and not over_budget + ok = read_short <= 0 and floor_short <= 0 and deliver_short <= 0 and drained and not over_budget parts: list[str] = [] if read_short > 0: parts.append( f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)" ) + if floor_short > 0: + parts.append( + f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — " + f"the unconfirmed-send excusal cannot lower this floor" + ) if deliver_short > 0: parts.append( f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})" diff --git a/harness/load/estate/runner.py b/harness/load/estate/runner.py index c1ae832c..47b78c8a 100644 --- a/harness/load/estate/runner.py +++ b/harness/load/estate/runner.py @@ -356,6 +356,9 @@ def _build_record( ) -> EstateRecord: c = metrics.counters.snapshot() base, final = poller.baseline, poller.final + # Budget = the step's connection count — the small-run FLOOR under the reconcile's half-the-run + # excusal fraction, never a per-connection stranding allowance. `read >= sent // 2` holds here + # regardless of it (an estate step CAN send on the order of its connection count). no_loss = _reconcile(c, base, final, unconfirmed_budget=profile.count) in_pipeline_peak = max((s.in_pipeline for s in samples), default=0) read_per_s, written_per_s = _throughput_rates(samples) @@ -432,19 +435,32 @@ def _reconcile( # budget it is a systemic no-ACK fault. The bound is a FRACTION of the run (at most half), floored # by the connection count — NOT "~one per connection", a model this sender breaks because # `_inflight` is an unbounded deque. Mirrors the connscale/report reconciles; see report.py. + # + # The floor arm means the budget does NOT by itself keep half the run unexcusable: this call site + # passes `profile.count` (the connection count), and an estate step whose sends are of the same + # order as its connection count lets the count win the max(), degrading the intake bound to + # `read >= sent - count`. The `read >= sent // 2` guarantee is therefore enforced separately below. unconfirmed = c.timeouts budget = max(unconfirmed_budget, sent // 2) over_budget = unconfirmed > budget excused = 0 if over_budget else unconfirmed read_short = sent - excused - read + # Anti-vacuity floor the excusal cannot lower (see report.py's copy for the derivation): zero at + # sent == 0, rounds DOWN on an odd `sent`, and holds even when `timeouts > sent`. + floor_short = sent // 2 - read deliver_short = written - sink_received drained = backlog == 0 - ok = read_short <= 0 and deliver_short <= 0 and drained and not over_budget + ok = read_short <= 0 and floor_short <= 0 and deliver_short <= 0 and drained and not over_budget parts: list[str] = [] if read_short > 0: parts.append( f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)" ) + if floor_short > 0: + parts.append( + f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — " + f"the unconfirmed-send excusal cannot lower this floor" + ) if deliver_short > 0: parts.append( f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})" @@ -454,7 +470,16 @@ def _reconcile( if over_budget: parts.append( f"{unconfirmed} unconfirmed sends exceed the stranding budget " - f"({budget} = max(connections, half the run)) — systemic no-ACK fault" + f"({budget} = max(connections, half the run)) — systemic no-ACK fault " + f"(possible accepted-and-dropped); nothing excused" + ) + elif unconfirmed > 0 and read < sent: + # Honest reporting, as in the two sibling copies: without this branch a bounded-excused run + # reported the flat "read>=sent, ..." detail while read was demonstrably BELOW sent — a claim + # false on its own numbers. The gap is attributed, never silently absorbed. + parts.append( + f"{unconfirmed} unconfirmed send(s) (no ACK before connection close) " + f"not observed at intake — not counted as loss" ) detail = "; ".join(parts) if parts else "read>=sent, sink_received>=written, backlog drained" return NoLoss(ok, sent, read, written, sink_received, backlog, detail) diff --git a/harness/load/multishard.py b/harness/load/multishard.py index 926721fd..d63d8d2e 100644 --- a/harness/load/multishard.py +++ b/harness/load/multishard.py @@ -135,8 +135,9 @@ class MultiShardRecord: ack_p99_ms: float # Unconfirmed sends (in-flight at a connection close with no ACK seen). The reconcile excuses - # these from the intake bound only up to ~one per connection (engines × count); surfaced here so - # the tolerance width is visible on a PASSING record too. Default 0 for older artifacts. + # these from the intake bound up to max(total connections, half the run), and never below its + # `read >= sent // 2` intake floor; surfaced here so the tolerance width is visible on a PASSING + # record too. Default 0 for older artifacts. timeouts: int = 0 # --- per-engine disjoint-lane attribution (the no-cross-engine-steal proof) --- @@ -560,8 +561,9 @@ def _build_record( ) -> MultiShardRecord: c = metrics.counters.snapshot() base, final = poller.baseline, poller.final - # Budget = total connection count across every engine (at most ~one stranded in-flight per - # connection is a plausible teardown artifact; more is a systemic no-ACK fault). + # Budget = total connection count across every engine — a small-run FLOOR under the reconcile's + # half-the-run excusal fraction, not the retired "~one stranded in-flight per connection" model. + # Its independent intake floor keeps `read >= sent // 2` required whatever this value is. no_loss = _reconcile(c, base, final, unconfirmed_budget=engines * count_per_engine) in_pipeline_peak = max((s.in_pipeline for s in samples), default=0) # Aggregate achieved/delivered rate = the read/written delta across EXACTLY the hold window diff --git a/harness/load/report.py b/harness/load/report.py index 32e5fca6..236f350c 100644 --- a/harness/load/report.py +++ b/harness/load/report.py @@ -371,7 +371,9 @@ def build_report( slos.extend(_phase_slos(rec, profile.slo_for(rec.phase))) # Unconfirmed-send budget = the run's total client connection count (one pool of pool_size per - # target): at most ~one stranded in-flight frame per connection is a plausible teardown artifact. + # target). It is only the small-run FLOOR under `_reconcile`'s half-the-run excusal fraction — + # NOT "~one stranded in-flight frame per connection", a model retired because this sender's + # in-flight deque is unbounded. `_reconcile` requires `read >= sent // 2` regardless of it. no_loss = _reconcile( final_counters, poller, @@ -470,10 +472,15 @@ def _phase_slos(rec: PhaseRecord, slo: Slo) -> list[SloCheck]: # a single transport blip (one reconnect's failed open / stranded in-flights — client-side # noise, not loss) is >1%, so any sane threshold flips on one event. Below the floor the check # is not emitted at all (no verdict beats a noise-driven one); real load profiles run thousands - # of messages per phase and keep the gate. A mass reset/timeout FLOOD on a small phase is not - # un-gated by this floor: the reconcile's bounded unconfirmed-send budget fails zero_loss when - # timeouts exceed ~one per connection. (max_nak_rate below deliberately has no floor — a NAK - # is a deterministic engine verdict, not transport noise, so even one is signal.) + # of messages per phase and keep the gate. A mass reset/timeout FLOOD is not un-gated by this + # floor on the runs that need it: the reconcile fails zero_loss when timeouts exceed its + # stranding budget, or when intake drops below its unconditional `read >= sent // 2` floor — + # and on a CI smoke the whole run IS that phase. Mind the scope difference, though: the + # reconcile is computed ONCE over the run's final counters, so a flood confined to a sub-floor + # MEASURED phase inside a large multi-phase run is gated by neither. No shipped profile has + # such a phase today (reference's only sub-floor phase is `warmup`, which is unmeasured), so + # that is a known scope gap rather than a live hole. (max_nak_rate below deliberately has no + # floor — a NAK is a deterministic engine verdict, not transport noise, so even one is signal.) er = errs / sent out.append( SloCheck( @@ -585,25 +592,51 @@ def _reconcile( # zero-loss run (14 stranded of 90 against a budget of 4) and red the required windows-2025 leg. # # So bound it as a FRACTION of the run, floored by the caller's connection count for tiny runs: - # at most half the sends may be excused, which keeps `read >= sent // 2` ALWAYS required. A dead - # ACK path (`timeouts == sent`) still blows the cap and fails loudly, and a run that genuinely - # loses more than half its sends at intake is still caught. Observed teardown stranding is ~16%, - # so this is ~3x the worst seen — wide enough to stop flaking, far from vacuous. + # at most half the sends may be excused. Observed teardown stranding is ~16%, so half is ~3x the + # worst seen — wide enough to stop flaking, far from vacuous. A dead ACK path (`timeouts == sent`) + # blows that cap and fails loudly — but ONLY while the connection-count floor does not dominate: + # once `unconfirmed_budget >= sent` the max() forgives even a 100%-dead ACK path, and the intake + # floor below cannot catch that one either, because its signature is a HIGH read with no ACKs. + # Closing that half needs the floor ARM of the budget bounded; the floor below closes the INTAKE + # half only. Known gap, deliberately not fixed here — it would re-open PR #17's de-flake. + # + # That cap ALONE does NOT deliver `read >= sent // 2`, and the comment here used to claim it did. + # `max()` takes the connection count as a FLOOR, not a ceiling, and every call site passes a + # connection count (connscale and estate pass the step's `count` verbatim; this one passes + # pool_size x targets). A short, low-rate step sends barely more than one message per connection, + # so the count WINS the max() and the bound degrades to `read >= sent - connections` — at + # connscale-smoke's N=100 cell (~105 sends, budget max(100, 52) = 100) that is `read >= 5`, the + # very vacuity the cap exists to prevent. Nothing clamps `excused` to `sent` either, so + # `timeouts > sent` drives it to `read >= 0` outright. + # + # Hence the guarantee is enforced SEPARATELY below, as an intake floor the excusal cannot lower. unconfirmed = counters.timeouts budget = max(unconfirmed_budget, sent // 2) over_budget = unconfirmed > budget excused = 0 if over_budget else unconfirmed read_short = sent - excused - read + # The anti-vacuity guarantee, enforced independently of the excusal AND of `tolerance` — the + # tolerance is an operator knob on the SHORTFALL, not a licence to lower the floor that makes + # "at least half the run was demonstrably ingested" true at every call site. `sent // 2` is 0 at + # sent == 0 (a run that sent nothing clears it trivially) and rounds DOWN on an odd `sent`, so + # the floor is never stricter than the documented bound. + floor_short = sent // 2 - read deliver_short = written - sink_received read_ok = read_short <= tolerance + floor_ok = floor_short <= 0 deliver_ok = deliver_short <= tolerance drained = backlog == 0 - ok = read_ok and deliver_ok and drained and not over_budget + ok = read_ok and floor_ok and deliver_ok and drained and not over_budget parts: list[str] = [] if not read_ok: parts.append( f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)" ) + if not floor_ok: + parts.append( + f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — " + f"the unconfirmed-send excusal cannot lower this floor" + ) if not deliver_ok: parts.append( f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})" diff --git a/tests/test_harness_reconcile.py b/tests/test_harness_reconcile.py index ed339f27..34afadd3 100644 --- a/tests/test_harness_reconcile.py +++ b/tests/test_harness_reconcile.py @@ -9,19 +9,30 @@ NOTHING is excused. With ``timeouts == 0`` (every healthy run) the check is exactly as strict as ``read >= sent``. -The bound is ``max(unconfirmed_budget, sent // 2)``. It was ``unconfirmed_budget`` alone — "~one -stranded in-flight frame per connection" — until that model was found to be wrong for this sender: -``_inflight`` is an unbounded deque and open-loop sends are paced by the offered rate, not by an ACK -slot, so genuine teardown stranding scales with rate x ACK-latency rather than the connection count. -It false-failed a zero-loss run (14 stranded of 90 against a budget of 4) and red the required -windows-2025 leg. Capping at half the run keeps ``read >= sent // 2`` always required, so a dead ACK -path still fails loudly while ordinary teardown weather does not. +The systemic-fault threshold is ``max(unconfirmed_budget, sent // 2)``. It was ``unconfirmed_budget`` +alone — "~one stranded in-flight frame per connection" — until that model was found to be wrong for +this sender: ``_inflight`` is an unbounded deque and open-loop sends are paced by the offered rate, +not by an ACK slot, so genuine teardown stranding scales with rate x ACK-latency rather than the +connection count. It false-failed a zero-loss run (14 stranded of 90 against a budget of 4) and red +the required windows-2025 leg. + +That ``max()`` does NOT by itself keep ``read >= sent // 2`` required, though it was once documented +as doing so: the connection count is a FLOOR, not a ceiling, and every call site passes a connection +count (connscale and estate pass the step's ``count`` verbatim). Whenever the count exceeds half the +sends — the normal shape of a short, low-rate step, e.g. connscale-smoke's N=100 cell at ~105 sends +against a budget of ``max(100, 52) = 100`` — the bound degrades to ``read >= sent - connections``, +i.e. ``read >= 5``; and since nothing clamps the excusal to ``sent``, ``timeouts > sent`` degrades it +to ``read >= 0`` outright. So the guarantee is enforced SEPARATELY, as an unconditional intake floor +the excusal cannot lower: ``read >= sent // 2``, in all three copies, at every call site. These tests pin every edge the tolerance could silently widen through (the harness has caught real store bugs with this check — mf-load-test-harness — and that detection must survive the de-flake): -loss beyond the excusal, an over-budget flood (even with zero actual loss), and — mutation-tested — -that the tolerance applies to INTAKE ONLY: the delivery/backlog cases use shortfalls EXACTLY EQUAL to -the timeout count, so leaking the subtraction into either check flips the expected verdict. +loss beyond the excusal, an over-budget flood (even with zero actual loss), the floor-binding regime +(``unconfirmed_budget >= sent``, which no test reached before — every case here used a budget of 2 +against 36 sends, so ``sent // 2`` always won the max() and the floor arm was a surviving mutant), +and — mutation-tested — that the tolerance applies to INTAKE ONLY: the delivery/backlog cases use +shortfalls EXACTLY EQUAL to the timeout count, so leaking the subtraction into either check flips the +expected verdict. """ from __future__ import annotations @@ -31,6 +42,7 @@ from harness.load.connscale.runner import _reconcile as connscale_reconcile from harness.load.enginepoll import EnginePoller, EngineSample +from harness.load.estate.runner import _reconcile as estate_reconcile from harness.load.metrics import Counters, Histogram from harness.load.profile import Phase, Slo from harness.load.report import ( @@ -65,6 +77,14 @@ def _sample(*, read: int, written: int, pending: int = 0, inflight: int = 0) -> _BASE = _sample(read=0, written=0) _BUDGET = 2 # a tiny run's connection count +# connscale-smoke's real shape at the N=100 cell: a fixed_aggregate step offering 35 msg/s for 3s is +# ~105 sends against 100 connections, so `max(unconfirmed_budget, sent // 2)` is max(100, 52) = 100 — +# the CONNECTION COUNT wins the max() and the excusal alone would allow read >= 5. Every other test +# in this module uses a budget of 2 or 4 against 36/90 sends, where sent // 2 always wins, so this +# constant is the only thing that exercises the floor arm at all. +_SMOKE_SENT = 105 +_SMOKE_BUDGET = 100 # >= sent // 2 == 52: the regime where the budget stops bounding anything + # --- connscale _reconcile ------------------------------------------------------------------------ @@ -140,6 +160,68 @@ def test_connscale_reconcile_tolerance_is_intake_only() -> None: assert "not drained" in result2.detail +def test_connscale_reconcile_intake_floor_holds_when_the_budget_stops_bounding() -> None: + # The floor-binding regime, at connscale-smoke's own N=100 shape: 105 sends, 100 connections, a + # dead ACK path that also cost intake (read 5 of 105). The excusal ALONE passes this — 100 + # unconfirmed is not over max(100, 52), so read_short is 105 - 100 - 5 == 0 — which is exactly the + # `read >= 5` vacuity the cap was supposed to prevent. The unconditional floor is what fails it. + c = Counters(sent=_SMOKE_SENT, acked=5, timeouts=100, sink_received=5) + result = connscale_reconcile( + c, _BASE, _sample(read=5, written=5), unconfirmed_budget=_SMOKE_BUDGET + ) + assert not result.ok, result.detail + assert "intake floor 52" in result.detail + # And it is genuinely the FLOOR doing the work here, not the stranding budget: 100 unconfirmed is + # within max(100, 52), so the systemic-fault branch never fires. + assert "stranding budget" not in result.detail + + +def test_connscale_reconcile_teardown_stranding_still_passes_at_the_smoke_shape() -> None: + # The other side of the new floor, same 105-send / 100-connection shape: ordinary teardown + # stranding (17 of 105, ~16% — the rate x ACK-latency weather PR #17 de-flaked for) still + # reconciles clean, because the engine read every send that was not stranded and 88 >= 52. The + # floor must not reintroduce the flake it was added alongside. + c = Counters(sent=_SMOKE_SENT, acked=88, timeouts=17, sink_received=88) + result = connscale_reconcile( + c, _BASE, _sample(read=88, written=88), unconfirmed_budget=_SMOKE_BUDGET + ) + assert result.ok, result.detail + + +def test_connscale_reconcile_floor_is_exactly_half_rounded_down() -> None: + # Boundary pin on an ODD run size: the floor is sent // 2 (52 of 105), never ceil. Exactly at the + # floor passes; one below fails. Pins the rounding so a later `-(-sent // 2)` "tidy-up" — which + # would make the floor STRICTER than the documented `read >= sent // 2` — is caught here. + at_floor = Counters(sent=_SMOKE_SENT, acked=52, timeouts=53, sink_received=52) + assert connscale_reconcile( + at_floor, _BASE, _sample(read=52, written=52), unconfirmed_budget=_SMOKE_BUDGET + ).ok + below = Counters(sent=_SMOKE_SENT, acked=51, timeouts=54, sink_received=51) + result = connscale_reconcile( + below, _BASE, _sample(read=51, written=51), unconfirmed_budget=_SMOKE_BUDGET + ) + assert not result.ok + assert "intake floor 52" in result.detail + + +def test_connscale_reconcile_more_timeouts_than_sends_cannot_go_vacuous() -> None: + # `excused` is never clamped to `sent`, so timeouts > sent drives read_short NEGATIVE and the + # intake bound to `read >= 0` — total vacuity, not mere degradation — whenever the budget also + # covers it. The floor is the only thing standing between that state and a green zero-loss. + c = Counters(sent=10, acked=0, timeouts=50, sink_received=0) + result = connscale_reconcile(c, _BASE, _sample(read=0, written=0), unconfirmed_budget=100) + assert not result.ok, result.detail + assert "intake floor 5" in result.detail + + +def test_connscale_reconcile_empty_run_clears_the_floor_trivially() -> None: + # sent == 0 makes the floor 0, so a run that sent nothing is not failed BY the floor (it has + # nothing to read). Guards against a `sent // 2` -> `max(1, ...)` style edit failing empty runs. + assert connscale_reconcile( + Counters(), _BASE, _sample(read=0, written=0), unconfirmed_budget=_SMOKE_BUDGET + ).ok + + # --- load runner _reconcile ---------------------------------------------------------------------- @@ -252,6 +334,109 @@ def test_load_reconcile_tolerance_is_intake_only() -> None: assert "not drained" in result2.detail +def test_load_reconcile_intake_floor_holds_when_the_budget_stops_bounding() -> None: + # The load copy has the same hole: its budget is pool_size x targets, which for a small phase can + # exceed half the sends just as the connscale connection count does. 105 sent, budget 100, read 5: + # read_short is 0 under the excusal, so only the unconditional floor fails it. + c = Counters(sent=_SMOKE_SENT, acked=5, timeouts=100, sink_received=5) + result = load_reconcile( + c, _poller(_sample(read=5, written=5)), 1.0, tolerance=0, unconfirmed_budget=_SMOKE_BUDGET + ) + assert not result.ok, result.detail + assert "intake floor 52" in result.detail + assert "stranding budget" not in result.detail + + +def test_load_reconcile_teardown_stranding_still_passes_at_the_smoke_shape() -> None: + # ~16% teardown stranding at the same budget-dominated shape, fanned out 2x, still reconciles + # clean: 88 read >= the 52 floor and every non-stranded send was observed. + c = Counters(sent=_SMOKE_SENT, acked=88, timeouts=17, sink_received=176) + result = load_reconcile( + c, + _poller(_sample(read=88, written=176)), + 1.0, + tolerance=0, + unconfirmed_budget=_SMOKE_BUDGET, + ) + assert result.ok, result.detail + + +def test_load_reconcile_tolerance_cannot_lower_the_intake_floor() -> None: + # `tolerance` is an operator knob on the SHORTFALL; it is deliberately not applied to the floor, + # so no combination of tolerance and excusal can make "at least half the run was ingested" false + # while zero_loss reads green. A tolerance wide enough to cover the whole gap changes nothing. + c = Counters(sent=_SMOKE_SENT, acked=5, timeouts=100, sink_received=5) + result = load_reconcile( + c, _poller(_sample(read=5, written=5)), 1.0, tolerance=100, unconfirmed_budget=_SMOKE_BUDGET + ) + assert not result.ok, result.detail + assert "intake floor 52" in result.detail + + +# --- estate _reconcile --------------------------------------------------------------------------- + + +def test_estate_reconcile_intake_floor_holds_when_the_budget_stops_bounding() -> None: + # The third copy, kept in step: estate passes `profile.count` as the budget, so a step whose sends + # are of the same order as its connection count lands in the same degraded regime. Before the + # floor this returned ok=True — and (see below) said so in a detail claiming read >= sent. + c = Counters(sent=_SMOKE_SENT, acked=5, timeouts=100, sink_received=5) + result = estate_reconcile( + c, _BASE, _sample(read=5, written=5), unconfirmed_budget=_SMOKE_BUDGET + ) + assert not result.ok, result.detail + assert "intake floor 52" in result.detail + assert "stranding budget" not in result.detail + + +def test_estate_reconcile_teardown_stranding_still_passes_at_the_smoke_shape() -> None: + c = Counters(sent=_SMOKE_SENT, acked=88, timeouts=17, sink_received=88) + result = estate_reconcile( + c, _BASE, _sample(read=88, written=88), unconfirmed_budget=_SMOKE_BUDGET + ) + assert result.ok, result.detail + + +def test_estate_reconcile_excused_run_reports_the_gap_instead_of_claiming_read_ge_sent() -> None: + # This copy had no honest-reporting branch, so a bounded-excused run printed the flat + # "read>=sent, sink_received>=written, backlog drained" while read (88) was demonstrably below + # sent (105) — a statement false on its own numbers, on the operator-facing no-loss line. + c = Counters(sent=_SMOKE_SENT, acked=88, timeouts=17, sink_received=88) + result = estate_reconcile( + c, _BASE, _sample(read=88, written=88), unconfirmed_budget=_SMOKE_BUDGET + ) + assert result.ok, result.detail + assert "17 unconfirmed send(s)" in result.detail + assert "read>=sent" not in result.detail + + +def test_estate_reconcile_clean_run_still_reports_the_flat_claim() -> None: + # ...and when read really does cover sent, the flat claim is true and still emitted. + c = Counters(sent=36, acked=36, sink_received=36) + result = estate_reconcile(c, _BASE, _sample(read=36, written=36), unconfirmed_budget=_BUDGET) + assert result.ok + assert result.detail == "read>=sent, sink_received>=written, backlog drained" + + +# --- the three copies, kept in step -------------------------------------------------------------- + + +def test_the_three_reconcile_copies_emit_the_same_over_budget_detail() -> None: + # "The three copies are kept in step" is asserted in three code comments and the changelog, and + # nothing enforced it — so it had already drifted: estate's copy omitted the + # "(possible accepted-and-dropped); nothing excused" suffix its siblings carry, and the SAME + # systemic fault therefore read differently to an operator depending on which runner caught it. + c = Counters(sent=36, acked=6, timeouts=30, sink_received=36) + full = _sample(read=36, written=36) + details = { + connscale_reconcile(c, _BASE, full, unconfirmed_budget=_BUDGET).detail, + estate_reconcile(c, _BASE, full, unconfirmed_budget=_BUDGET).detail, + load_reconcile(c, _poller(full), 1.0, tolerance=0, unconfirmed_budget=_BUDGET).detail, + } + assert len(details) == 1, details + assert "systemic no-ACK fault (possible accepted-and-dropped); nothing excused" in details.pop() + + # --- rate-SLO sample floor -----------------------------------------------------------------------