From 30b403633b338aa752280a75ac9506dbbe614eb7 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Wed, 5 Aug 2026 23:10:21 +0000 Subject: [PATCH] feat(stage-e): add host-load soft brake, throttle on approach to the 7.0 ceiling operating_envelope.HOST_LOAD_CEILING is a binary cliff: the instant a fresh sample reads above 7.0, dispatch_micro_batch halts and requires a fresh owner action to resume. Two passes tripped a day apart on narrow overshoots (7.0796, then 7.17236328125 - 1.1% and 2.5% over) despite the box otherwise running well under load, each costing a stopped pass and a human interaction. Adds cardpicker.stage_e_load_brake: between the no-self-resume gate and the fresh envelope sample, re-samples os.getloadavg() independently and, while the reading sits between STAGE_E_HOST_LOAD_SOFT_CEILING (default 6.0) and the hard ceiling, sleeps with jitter and re-samples rather than proceeding straight to the trip check. A load above the hard ceiling stops braking at once and lets the envelope's own fresh sample trip honestly - the brake never itself decides trip and never suppresses a genuine breach. Cumulative wait past STAGE_E_LOAD_BRAKE_MAX_WAIT_S (default 240s) proceeds anyway, matching the ~4 time-constant decay window of the 1-minute load average it re-samples. All three settings default to values that make the brake active out of the box. DispatchOutcome gains load_brake_waits/load_brake_seconds, merged into PilotRunLedger.counters on every completed micro-batch, so the brake's effect is queryable rather than invisible. Tests (test_stage_e_load_brake.py) assert the specific wait counts and durations for each band, not just direction - including that the shipped defaults produce a reachable band end to end, and that a load above the ceiling never sleeps. Every new test's assertion was confirmed to fail when the behaviour it checks was manually removed (see the PR body's verification section). --- MPCAutofill/MPCAutofill/settings.py | 28 ++ MPCAutofill/cardpicker/stage_e_dispatch.py | 49 +++- MPCAutofill/cardpicker/stage_e_load_brake.py | 196 ++++++++++++++ .../tests/test_stage_e_load_brake.py | 246 ++++++++++++++++++ docs/features/stage-e-operations.md | 97 ++++++- 5 files changed, 598 insertions(+), 18 deletions(-) create mode 100644 MPCAutofill/cardpicker/stage_e_load_brake.py create mode 100644 MPCAutofill/cardpicker/tests/test_stage_e_load_brake.py diff --git a/MPCAutofill/MPCAutofill/settings.py b/MPCAutofill/MPCAutofill/settings.py index 356ee09ef..1c83f04e4 100755 --- a/MPCAutofill/MPCAutofill/settings.py +++ b/MPCAutofill/MPCAutofill/settings.py @@ -618,6 +618,34 @@ # own "placeholder, not invented precision" convention immediately above. STAGE_E_MAX_CONCURRENT_DISPATCHES = env.int("STAGE_E_MAX_CONCURRENT_DISPATCHES", default=2) +# Host-load soft brake (2026-08-05 - cardpicker/stage_e_load_brake.py's own module docstring +# carries the full mechanism). Delays a dispatch, +# never bypasses the hard ceiling: `STAGE_E_HOST_LOAD_SOFT_CEILING` opens the band below +# `operating_envelope.HOST_LOAD_CEILING` (7.0) in which a dispatch sleeps and re-samples instead +# of proceeding straight to the envelope check that would otherwise trip it. All three default to +# values that make the brake active out of the box - matching STAGE_E_MAX_CONCURRENT_DISPATCHES's +# own "safe by default, no opt-in required" convention above, since an instance maintainer running +# their own catalogue for the first time cannot know to enable a knob they don't know exists. +# +# STAGE_E_HOST_LOAD_SOFT_CEILING - ~85% of the hard 7.0 ceiling, read off the two load-average +# trips this brake exists to prevent (7.0796 and 7.17236328125 - both trips were already inside +# 15% of the bar when the pass that produced them was launched, well above this 6.0 line). +STAGE_E_HOST_LOAD_SOFT_CEILING = env.float("STAGE_E_HOST_LOAD_SOFT_CEILING", default=6.0) + +# STAGE_E_LOAD_BRAKE_INTERVAL_S - base sleep per brake iteration, before jitter +# (`stage_e_load_brake.run_load_brake`'s own `uniform(0.75, 1.5)` multiplier). 15s is short +# relative to the 1-minute load average it re-samples on each wake so the loop still notices a +# quick recovery, without being so short that a sustained band spends most of its time waking up +# rather than waiting. +STAGE_E_LOAD_BRAKE_INTERVAL_S = env.float("STAGE_E_LOAD_BRAKE_INTERVAL_S", default=15.0) + +# STAGE_E_LOAD_BRAKE_MAX_WAIT_S - the absolute bound on how long one dispatch call may spend +# braking before proceeding anyway. `os.getloadavg()`'s one-minute figure is an EWMA with a ~60s +# time constant, so 240s is ~4 time constants (~98% decay of a step change) - long enough for the +# brake to be more than a token gesture, short enough that a sustained real breach still reaches +# the envelope's own hard trip within one dispatch call rather than stalling indefinitely. +STAGE_E_LOAD_BRAKE_MAX_WAIT_S = env.float("STAGE_E_LOAD_BRAKE_MAX_WAIT_S", default=240.0) + # Persistent sweep cursor sizing (issue #458 - see cardpicker/stage_e_dispatch.py's # `_select_micro_batch` and docs/features/stage-e-operations.md's Phase 2 section for the full # design). Plain constants, not env-tunable (unlike STAGE_E_MICRO_BATCH_SIZE/ diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index c376c1e26..8c06fe5b4 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -118,6 +118,7 @@ from cardpicker.process_metrics import get_process_rss_mb from cardpicker.stage_e_batch_sizing import MODE_INCREMENTAL, resolve_micro_batch_size from cardpicker.stage_e_concurrency import try_acquire_dispatch_slot +from cardpicker.stage_e_load_brake import apply_load_brake from cardpicker.stage_e_signals import suppress_evidence_change_echo from cardpicker.utils import get_baked_git_sha from cardpicker.vote_write import purge_and_write_votes @@ -293,6 +294,15 @@ class DispatchOutcome: # selection). stage_c_backlog_found: int = 0 stage_c_backlog_wrapped: bool = False + # Stage E host-load soft brake (`stage_e_load_brake.py`) - + # how many times, and for how long in total, THIS dispatch call slept before its own fresh + # envelope sample below, because live load was in the soft-to-hard band. Zero/0.0 means either + # the brake never engaged (the common case) or this outcome was constructed before the brake + # ever ran (`halted-open-trip`, returned before the brake's insertion point) - both read the + # same as "no delay", which is the correct reading for a caller that only wants to know + # whether THIS call was delayed. + load_brake_waits: int = 0 + load_brake_seconds: float = 0.0 trip_id: Optional[str] = None @@ -1492,13 +1502,14 @@ def dispatch_micro_batch( unique per-attempt, per-batch ledger row while `run_id` keeps stamping every data row with the clean identity. When `ledger_run_id` is None this function behaves exactly as before. - Ordering: no-self-resume gate -> fresh envelope sample -> batch selection -> - concurrency-cap slot acquire (`cardpicker.stage_e_concurrency`) -> Stage C (sequential, per-card) - -> Stage D (AS-IS entry points, scoped) -> ledger write -> slot release. Every gate below returns - WITHOUT touching the DB (aside from the envelope check's own trip-persist side effect, and the - concurrency-cap check's own advisory-lock round trip, plus - 2026-07-25 - a throttled outcome's - `StageEThrottleCounter.record()` call, a single-row atomic counter update, never a growing - table) the instant it applies - a halted or throttled dispatch never partially starts Stage C. + Ordering: no-self-resume gate -> load brake (`cardpicker.stage_e_load_brake`) -> fresh envelope + sample -> batch selection -> concurrency-cap slot acquire (`cardpicker.stage_e_concurrency`) -> + Stage C (sequential, per-card) -> Stage D (AS-IS entry points, scoped) -> ledger write -> slot + release. Every gate below returns WITHOUT touching the DB (aside from the envelope check's own + trip-persist side effect, the concurrency-cap check's own advisory-lock round trip, plus - + 2026-07-25 - a throttled outcome's `StageEThrottleCounter.record()` call, a single-row atomic + counter update, never a growing table) the instant it applies - a halted or throttled dispatch + never partially starts Stage C. """ if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): @@ -1517,6 +1528,14 @@ def dispatch_micro_batch( ) return DispatchOutcome(status="halted-open-trip", run_id=run_id, trip_id=existing_trip.trip_id) + # LOAD BRAKE (`stage_e_load_brake.py`): a delay, never a + # bypass, sits here - after the no-self-resume gate so a braking process never blocks a check + # of an already-open trip, and before the fresh envelope sample below so the brake's own + # (possibly repeated) reads are never reused as that sample and can never suppress a genuine + # breach. Also before `try_acquire_dispatch_slot`, so a braking process holds no + # concurrency-cap slot while it waits. + brake_outcome = apply_load_brake() + signals = _sample_envelope_signals() fresh_trip = check_envelope(signals, run_id=run_id) if fresh_trip is not None: @@ -1526,7 +1545,13 @@ def dispatch_micro_batch( fresh_trip.detail, fresh_trip.trip_id, ) - return DispatchOutcome(status="halted-new-trip", run_id=run_id, trip_id=fresh_trip.trip_id) + return DispatchOutcome( + status="halted-new-trip", + run_id=run_id, + trip_id=fresh_trip.trip_id, + load_brake_waits=brake_outcome.waits, + load_brake_seconds=brake_outcome.seconds, + ) # BATCH SIZE (2026-07-29 - `cardpicker.stage_e_batch_sizing`'s own module docstring carries the # rule, the measurements it was read off, and the precedence order). An explicit `batch_size` @@ -1546,6 +1571,8 @@ def dispatch_micro_batch( run_id=run_id, stage_c_backlog_found=stage_c_fill.found, stage_c_backlog_wrapped=stage_c_fill.wrapped, + load_brake_waits=brake_outcome.waits, + load_brake_seconds=brake_outcome.seconds, ) # CONCURRENCY CAP (companion to PR #448's vote-collision fix - cardpicker.stage_e_concurrency's @@ -1573,6 +1600,8 @@ def dispatch_micro_batch( run_id=run_id, stage_c_backlog_found=stage_c_fill.found, stage_c_backlog_wrapped=stage_c_fill.wrapped, + load_brake_waits=brake_outcome.waits, + load_brake_seconds=brake_outcome.seconds, ) dispatch_run_id = run_id or f"stage-e-stream-{timezone.now().strftime('%Y%m%dT%H%M%S%f')}Z" @@ -1601,6 +1630,8 @@ def dispatch_micro_batch( card_ids=batch_ids, stage_c_backlog_found=stage_c_fill.found, stage_c_backlog_wrapped=stage_c_fill.wrapped, + load_brake_waits=brake_outcome.waits, + load_brake_seconds=brake_outcome.seconds, ) batch_start = time.monotonic() @@ -1648,6 +1679,8 @@ def dispatch_micro_batch( "stage_d_verdict_transfer_votes": outcome.stage_d_verdict_transfer_votes, "peak_rss_mb": peak_rss_mb, "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, + "load_brake_waits": outcome.load_brake_waits, + "load_brake_seconds": outcome.load_brake_seconds, }, ) ledger.save(update_fields=["status", "finished_at", "counters"]) diff --git a/MPCAutofill/cardpicker/stage_e_load_brake.py b/MPCAutofill/cardpicker/stage_e_load_brake.py new file mode 100644 index 000000000..7fca3f73e --- /dev/null +++ b/MPCAutofill/cardpicker/stage_e_load_brake.py @@ -0,0 +1,196 @@ +""" +Stage E host-load soft brake ("throttle on approach to 7.0") — added 2026-08-05, after +`operating_envelope.HOST_LOAD_CEILING = 7.0` tripped on a 1.1% overshoot +(load_avg 7.0796) and again 2026-08-05 on a 2.5% overshoot (7.17236328125), each costing a stopped +pass and a fresh human `resolve_envelope_trip` action (no self-resume — see that module's own +docstring). The ceiling is a binary cliff: `_bar_breach` trips the instant `load_avg > 7.0`, and +until this module existed there was no feedback from a live load reading anywhere in the dispatch +path except that trip — `stage_e_batch_sizing._duration_limit` derives its own contention term from +the CEILING, not a live sample, on purpose (reproducibility, see that function's own docstring), so +it cannot fill this gap. + +WHAT THIS IS NOT: this brake does not soften the 7.0 bar, does not touch `operating_envelope.py`, +and does not suppress a trip. A load that actually crosses 7.0 still reaches +`stage_e_dispatch._sample_envelope_signals`/`check_envelope` and trips exactly as before this +module existed — this module's only power is to DELAY a dispatch decision by re-sampling load in a +band BELOW the ceiling and sleeping while it stays there, so a pass under momentary contention slows +down instead of arriving at the ceiling at all. Called from `stage_e_dispatch.dispatch_micro_batch`, +between the no-self-resume gate and the fresh envelope sample — see that function's own docstring +for why that is the only correct insertion point (a braking process must hold no concurrency-cap +slot, and the brake's own sample must never be reused as the envelope's fresh sample). + +MECHANISM — three bands, `STAGE_E_HOST_LOAD_SOFT_CEILING` (default 6.0, ~85% of the hard 7.0) below +`operating_envelope.HOST_LOAD_CEILING` (7.0): + + * `load < soft` — proceed immediately. The common case; zero added cost. + * `soft <= load <= hard` — sleep, re-sample, repeat. + * `load > hard` — stop braking at once and let the caller's own fresh envelope sample trip + honestly. This module never itself decides "trip" — `brake_decision` reports the band, and the + caller (`dispatch_micro_batch`) is the one that acts on the caller's OWN sample moments later. + * cumulative wait > `STAGE_E_LOAD_BRAKE_MAX_WAIT_S` (default 240s) — proceed anyway. Best-effort, + and must never deadlock an unattended multi-hour run. + +WHY THIS ACTUALLY REDUCES LOAD: the load a pass generates is its own concurrency +(`settings.STAGE_E_MAX_CONCURRENT_DISPATCHES`). When every resident dispatcher is in the band, each +one pauses independently at its own next batch boundary — the resident process count falls, load +decays, and they resume. The pass self-throttles down to whatever concurrency fits under the +ceiling, continuously, instead of running at full concurrency until it hits the wall and halts. + +JITTER IS MANDATORY. Every dispatcher reads the same global `os.getloadavg()`. Without jitter they +would all back off and resume in lockstep — a sawtooth, and a thundering herd on every resume. Each +sleep is `interval_s * uniform(0.75, 1.5)`. + +`os.getloadavg()`'s 1-minute figure is an EWMA with a ~60s time constant — it shows only ~63% of a +step change after 60s. A single 15s sleep barely moves it; this is why the default max wait (240s, +~4 time constants, ~98% decay) is what it is, and why this module is written to expect a slow +response rather than tuned as though load reacted instantly to a paused process. + +TESTABILITY follows `EnvelopeSignals`' own precedent (`operating_envelope.py`): `brake_decision` is +plain data in, one of three strings out, no I/O. `run_load_brake` is a thin loop with the sampler, +the sleep function and the jitter function all INJECTED, so no test ever monkeypatches +`os.getloadavg` globally or actually sleeps. `apply_load_brake` is the only function that touches +`os`/`time.sleep`/Django settings, and it is a thin, deliberately untested-in-detail wrapper around +`run_load_brake` — see that function's own docstring for why it also can never raise. +""" + +import logging +import os +import random +import time +from dataclasses import dataclass +from typing import Callable, Optional + +from django.conf import settings + +from cardpicker.operating_envelope import HOST_LOAD_CEILING + +logger = logging.getLogger(__name__) + +# Defaults, active out of the box - an +# instance maintainer running their own catalogue for the first time gets the brake without having +# to know it exists or opt in, matching this project's standing "a knob that has to be set to do +# anything is the wrong polarity" posture (see settings.py's STAGE_E_MAX_CONCURRENT_DISPATCHES/ +# STAGE_E_STREAMING_ENABLED comments for the same convention applied elsewhere in this subsystem). +DEFAULT_SOFT_CEILING = 6.0 +DEFAULT_INTERVAL_S = 15.0 +DEFAULT_MAX_WAIT_S = 240.0 + +# `brake_decision`'s three possible answers - plain strings rather than an enum, matching +# `operating_envelope.EnvelopeTrip.Bar`'s own "match the caller's existing string convention" +# precedent for a small, stable, cross-module vocabulary. +WAIT = "wait" +PROCEED = "proceed" +TRIP = "trip" + + +def brake_decision(load_avg: Optional[float], soft_ceiling: float, hard_ceiling: float) -> str: + """ + The pure primitive - one load reading in, one band out, no I/O, no sleep, no settings read. + `load_avg=None` (a platform without a readable `os.getloadavg`, matching + `stage_e_dispatch._sample_envelope_signals`'s own documented convention) always returns + `PROCEED` - a brake that cannot see load must never block a dispatch on that account. + + Boundaries are both closed on the WAIT side, per the brief's own table: `load == soft_ceiling` + and `load == hard_ceiling` both wait, matching "soft <= load <= hard -> sleep, re-sample, + repeat" verbatim. Only `load > hard_ceiling` (strictly greater, matching + `operating_envelope._bar_breach`'s own `load_avg > HOST_LOAD_CEILING` check byte-for-byte) is + TRIP - this function must classify a genuine breach identically to the envelope primitive it + sits in front of, or the two could disagree about where 7.0 itself falls. + """ + if load_avg is None: + return PROCEED + if load_avg > hard_ceiling: + return TRIP + if load_avg >= soft_ceiling: + return WAIT + return PROCEED + + +@dataclass(frozen=True) +class BrakeOutcome: + """What one `run_load_brake` call did - zero waits/zero seconds is indistinguishable from + "the brake never engaged" and from "the brake was never reached", both of which are the + correct reading for a caller that only wants to know whether THIS call was delayed.""" + + waits: int = 0 + seconds: float = 0.0 + + +def run_load_brake( + sample: Callable[[], Optional[float]], + sleep: Callable[[float], None], + soft_ceiling: float, + hard_ceiling: float, + interval_s: float, + max_wait_s: float, + jitter: Callable[[], float] = lambda: random.uniform(0.75, 1.5), +) -> BrakeOutcome: + """ + The thin loop (module docstring's TESTABILITY section) - `sample`, `sleep` and `jitter` are all + injected so a test drives this deterministically without touching `os.getloadavg` or actually + sleeping. Re-samples via `sample()` on every iteration (never reuses a reading), matching the + brief's "must re-sample rather than reuse a stale sample" requirement. + + Stops WAITing the moment `brake_decision` reports anything other than `WAIT` (a `TRIP` reading + stops braking AT ONCE and returns immediately, letting the caller's own fresh envelope sample + trip honestly a moment later - this function never itself trips anything), or the moment + cumulative wait already spent would meet or exceed `max_wait_s` before another sleep - checked + BEFORE sleeping, not after, so the returned `seconds` can never itself exceed `max_wait_s`. + """ + waits = 0 + total_wait = 0.0 + while True: + decision = brake_decision(sample(), soft_ceiling, hard_ceiling) + if decision != WAIT: + return BrakeOutcome(waits=waits, seconds=round(total_wait, 3)) + if total_wait >= max_wait_s: + return BrakeOutcome(waits=waits, seconds=round(total_wait, 3)) + wait_for = interval_s * jitter() + sleep(wait_for) + waits += 1 + total_wait += wait_for + + +def _sample_load_avg() -> Optional[float]: + """Matches `stage_e_dispatch._sample_envelope_signals`'s own `os.getloadavg` try/except + convention byte-for-byte - `None` on a platform without it, never an exception.""" + try: + return os.getloadavg()[0] + except (OSError, AttributeError): + return None + + +def apply_load_brake() -> BrakeOutcome: + """ + THE entry point `dispatch_micro_batch` calls - reads the three settings, samples real load via + `os.getloadavg`, sleeps via real `time.sleep`, and drives `run_load_brake`. + + Wrapped in a bare `try/except Exception` that returns an unbraked `BrakeOutcome()` on ANY + error - matching `stage_e_batch_sizing.resolve_micro_batch_size`'s own stated posture ("a + typo'd env var must not be able to take the run down"). A malformed setting, a transient + `os.getloadavg` failure mid-loop, or anything else this function did not anticipate must + degrade to "dispatch proceeds as if unbraked", never to a broken pass - the brake is a + convenience the envelope's own hard ceiling does not depend on. + """ + try: + soft_ceiling = float(getattr(settings, "STAGE_E_HOST_LOAD_SOFT_CEILING", DEFAULT_SOFT_CEILING)) + interval_s = float(getattr(settings, "STAGE_E_LOAD_BRAKE_INTERVAL_S", DEFAULT_INTERVAL_S)) + max_wait_s = float(getattr(settings, "STAGE_E_LOAD_BRAKE_MAX_WAIT_S", DEFAULT_MAX_WAIT_S)) + outcome = run_load_brake( + sample=_sample_load_avg, + sleep=time.sleep, + soft_ceiling=soft_ceiling, + hard_ceiling=HOST_LOAD_CEILING, + interval_s=interval_s, + max_wait_s=max_wait_s, + ) + if outcome.waits: + logger.info( + "Stage E load brake engaged - %s wait(s), %.1fs total, before this dispatch", + outcome.waits, + outcome.seconds, + ) + return outcome + except Exception: + logger.exception("Stage E load brake failed - proceeding unbraked") + return BrakeOutcome() diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_load_brake.py b/MPCAutofill/cardpicker/tests/test_stage_e_load_brake.py new file mode 100644 index 000000000..d0a0c6968 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_stage_e_load_brake.py @@ -0,0 +1,246 @@ +""" +Tests for `cardpicker.stage_e_load_brake` (added 2026-08-05). + +`brake_decision` and `run_load_brake` never touch `os.getloadavg`/`time.sleep` - every case below +injects the sample/sleep/jitter it needs, matching that module's own `EnvelopeSignals`-precedent +testability design. `apply_load_brake` IS the thin wrapper that touches those real things, so its +own tests monkeypatch at the `os`/`time` boundary rather than reaching into the pure functions. + +No `db` fixture anywhere in this file - this module does no I/O of its own (its own module +docstring), and neither do these tests. +""" + +from typing import Callable, Iterator, List, Optional + +import pytest + +from django.test import override_settings + +from cardpicker.operating_envelope import HOST_LOAD_CEILING +from cardpicker.stage_e_load_brake import ( + DEFAULT_INTERVAL_S, + DEFAULT_SOFT_CEILING, + PROCEED, + TRIP, + WAIT, + BrakeOutcome, + apply_load_brake, + brake_decision, + run_load_brake, +) + + +class TestBrakeDecision: + def test_a_load_comfortably_below_soft_proceeds(self) -> None: + assert brake_decision(3.0, soft_ceiling=6.0, hard_ceiling=7.0) == PROCEED + + def test_the_soft_boundary_itself_waits(self) -> None: + assert brake_decision(6.0, soft_ceiling=6.0, hard_ceiling=7.0) == WAIT + + def test_the_approach_band_waits(self) -> None: + assert brake_decision(6.5, soft_ceiling=6.0, hard_ceiling=7.0) == WAIT + + def test_the_hard_boundary_itself_still_waits_not_trips(self) -> None: + # Only a load STRICTLY greater than the hard ceiling trips - matches + # `operating_envelope._bar_breach`'s own `load_avg > HOST_LOAD_CEILING` check exactly, so + # the two can never disagree about where 7.0 itself falls. + assert brake_decision(7.0, soft_ceiling=6.0, hard_ceiling=7.0) == WAIT + + def test_a_load_above_hard_trips(self) -> None: + assert brake_decision(7.1, soft_ceiling=6.0, hard_ceiling=7.0) == TRIP + + def test_an_unreadable_load_proceeds_rather_than_blocking(self) -> None: + # Matches `stage_e_dispatch._sample_envelope_signals`'s own "None means skip this bar" + # convention - a brake that cannot see load must never withhold a dispatch on that account. + assert brake_decision(None, soft_ceiling=6.0, hard_ceiling=7.0) == PROCEED + + def test_the_shipped_defaults_produce_a_band_that_is_actually_reachable(self) -> None: + # Not a static "soft < hard" assertion - drives `brake_decision` itself with the real + # module-level defaults at a load between them, so a future edit that moved the soft + # ceiling to or past 7.0 (making the band empty) fails THIS test, not just a comparison. + assert DEFAULT_SOFT_CEILING < HOST_LOAD_CEILING + midpoint = (DEFAULT_SOFT_CEILING + HOST_LOAD_CEILING) / 2 + assert brake_decision(midpoint, soft_ceiling=DEFAULT_SOFT_CEILING, hard_ceiling=HOST_LOAD_CEILING) == WAIT + + +def _fixed_jitter(value: float = 1.0) -> Callable[[], float]: + return lambda: value + + +def _sequence_sampler(values: List[Optional[float]]) -> Callable[[], Optional[float]]: + iterator: Iterator[Optional[float]] = iter(values) + + def _sample() -> Optional[float]: + return next(iterator) + + return _sample + + +def _getloadavg_sequence(one_minute_values: List[float]) -> Callable[[], "tuple[float, float, float]"]: + """Shapes a fixed sequence into `os.getloadavg()`'s own 3-tuple return type - only the first + element (the one-minute average) is ever read by this module.""" + iterator: Iterator[float] = iter(one_minute_values) + + def _sample() -> "tuple[float, float, float]": + value = next(iterator) + return (value, value, value) + + return _sample + + +class TestRunLoadBrake: + def test_a_load_below_soft_produces_no_waiting_and_never_sleeps(self) -> None: + sleeps: List[float] = [] + outcome = run_load_brake( + sample=lambda: 3.0, + sleep=sleeps.append, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=15.0, + max_wait_s=240.0, + jitter=_fixed_jitter(), + ) + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + assert sleeps == [] + + def test_a_load_in_the_approach_band_waits_the_specific_expected_amount(self) -> None: + # Two WAIT reads followed by a PROCEED read - the specific value asserted, not "is less + # than": exactly 2 waits, exactly 15.0 * fixed-jitter(1.0) * 2 = 30.0 seconds slept. + sleeps: List[float] = [] + outcome = run_load_brake( + sample=_sequence_sampler([6.5, 6.5, 5.0]), + sleep=sleeps.append, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=15.0, + max_wait_s=240.0, + jitter=_fixed_jitter(), + ) + assert outcome == BrakeOutcome(waits=2, seconds=30.0) + assert sleeps == [15.0, 15.0] + + def test_a_load_above_the_ceiling_stops_at_once_with_zero_wait(self) -> None: + # The brake must never delay - and so never mask - a genuine breach: a load already past + # the hard ceiling gets zero waits, zero seconds, and no sleep call at all. + sleeps: List[float] = [] + outcome = run_load_brake( + sample=lambda: 8.0, + sleep=sleeps.append, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=15.0, + max_wait_s=240.0, + jitter=_fixed_jitter(), + ) + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + assert sleeps == [] + + def test_cumulative_wait_past_the_bound_proceeds_anyway(self) -> None: + # Load never leaves the band on its own; only max_wait_s ends the loop. Traced by hand: + # iter 1 (total 0 < 150) sleeps 100 -> total 100; iter 2 (total 100 < 150) sleeps 100 -> + # total 200; iter 3 (total 200 >= 150) returns without a third sleep. + sleeps: List[float] = [] + outcome = run_load_brake( + sample=lambda: 6.5, + sleep=sleeps.append, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=100.0, + max_wait_s=150.0, + jitter=_fixed_jitter(), + ) + assert outcome == BrakeOutcome(waits=2, seconds=200.0) + assert sleeps == [100.0, 100.0] + + def test_it_resamples_on_every_iteration_rather_than_reusing_a_stale_reading(self) -> None: + calls = {"count": 0} + + def _sample() -> float: + calls["count"] += 1 + return 6.5 if calls["count"] < 3 else 5.0 + + outcome = run_load_brake( + sample=_sample, + sleep=lambda _seconds: None, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=1.0, + max_wait_s=240.0, + jitter=_fixed_jitter(), + ) + assert calls["count"] == 3 + assert outcome == BrakeOutcome(waits=2, seconds=2.0) + + def test_jitter_is_applied_to_every_sleep(self) -> None: + sleeps: List[float] = [] + outcome = run_load_brake( + sample=_sequence_sampler([6.5, 5.0]), + sleep=sleeps.append, + soft_ceiling=6.0, + hard_ceiling=7.0, + interval_s=10.0, + max_wait_s=240.0, + jitter=_fixed_jitter(0.75), + ) + assert sleeps == [7.5] + assert outcome == BrakeOutcome(waits=1, seconds=7.5) + + +class TestApplyLoadBrake: + def test_below_soft_with_real_settings_proceeds_without_sleeping(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("os.getloadavg", lambda: (3.0, 3.0, 3.0)) + sleeps: List[float] = [] + monkeypatch.setattr("time.sleep", sleeps.append) + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + assert sleeps == [] + + def test_the_shipped_defaults_actually_engage_the_brake_end_to_end(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Drives the real settings.py defaults (6.0 / 15 / 240) through the real wrapper, proving + # the shipped configuration - not a hand-picked test value - produces an engageable band. + monkeypatch.setattr("cardpicker.stage_e_load_brake.random.uniform", lambda a, b: 1.0) + monkeypatch.setattr("os.getloadavg", _getloadavg_sequence([6.5, 5.0])) + sleeps: List[float] = [] + monkeypatch.setattr("time.sleep", sleeps.append) + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=1, seconds=DEFAULT_INTERVAL_S) + assert sleeps == [DEFAULT_INTERVAL_S] + + def test_load_above_the_hard_ceiling_never_sleeps_even_with_real_settings( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr("os.getloadavg", lambda: (7.5, 7.5, 7.5)) + sleeps: List[float] = [] + monkeypatch.setattr("time.sleep", sleeps.append) + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + assert sleeps == [] + + def test_a_malformed_setting_degrades_to_unbraked_rather_than_raising(self) -> None: + # "a typo'd env var must not be able to take the run down" - matches + # `stage_e_batch_sizing.resolve_micro_batch_size`'s own stated posture for the same class + # of failure. + with override_settings(STAGE_E_HOST_LOAD_SOFT_CEILING="not-a-number"): + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + + def test_an_unreadable_loadavg_degrades_to_unbraked_rather_than_raising( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + def _raise() -> "tuple[float, float, float]": + raise OSError("no such syscall on this platform") + + monkeypatch.setattr("os.getloadavg", _raise) + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=0, seconds=0.0) + + def test_max_wait_setting_is_honoured_from_real_settings(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("cardpicker.stage_e_load_brake.random.uniform", lambda a, b: 1.0) + with override_settings(STAGE_E_LOAD_BRAKE_INTERVAL_S=100.0, STAGE_E_LOAD_BRAKE_MAX_WAIT_S=150.0): + monkeypatch.setattr("os.getloadavg", lambda: (6.5, 6.5, 6.5)) + sleeps: List[float] = [] + monkeypatch.setattr("time.sleep", sleeps.append) + outcome = apply_load_brake() + assert outcome == BrakeOutcome(waits=2, seconds=200.0) + assert sleeps == [100.0, 100.0] + assert outcome.seconds > 150.0 # proceeded past the bound rather than deadlocking on it diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 0ee316d14..627682081 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -446,19 +446,30 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ Phase 1's own review: no code path in `stage_e_dispatch.py` ever calls `acknowledge_trip` — resume is always `resolve_envelope_trip`'s own command, a fresh, explicit owner action (see the runbook above). -3. **Fresh envelope sample** — live host load (`os.getloadavg()`), this +3. **Load brake** (`cardpicker.stage_e_load_brake`, added 2026-08-05) — + before the fresh envelope sample below, re-samples `os.getloadavg()` on + its own account and, while the reading sits between + `settings.STAGE_E_HOST_LOAD_SOFT_CEILING` (default 6.0) and the hard + `operating_envelope.HOST_LOAD_CEILING` (7.0), sleeps and re-samples + rather than proceeding straight to the trip check. See "The host-load + soft brake" below for the full mechanism; the ordering guarantee that + matters here is that it runs after the no-self-resume gate (so it never + blocks a check of an already-open trip) and before both the fresh + envelope sample and the concurrency-cap slot acquire (so a braking + process holds no slot while it waits). +4. **Fresh envelope sample** — live host load (`os.getloadavg()`), this worker process's own RSS (`cardpicker.process_metrics.get_process_rss_mb`), and a rolling fetch-outcome window feed `check_envelope`. If THIS sample breaches a bar, a new trip is recorded and the call halts (`status="halted-new-trip"`) before touching Stage C/D at all. -4. **Micro-batch selection** — `_select_micro_batch` builds the card-id list: +5. **Micro-batch selection** — `_select_micro_batch` builds the card-id list: the triggering event's own card first (if any), filled up to `settings.STAGE_E_MICRO_BATCH_SIZE` from the Stage C backlog (cards lacking a full-manifest `ImageEvidence` row — the same shape `run_image_evidence_cohort.py`'s own resume filter uses, imported, not reimplemented) via the persistent sweep cursor described below (issue [#458](https://github.com/ProxyPrints/ProxyPrints.github.io/issues/458)). -5. **Concurrency-cap slot acquire** (companion change, 2026-07-24 — +6. **Concurrency-cap slot acquire** (companion change, 2026-07-24 — `cardpicker.stage_e_concurrency`) — refuses PROACTIVELY (`status="throttled-concurrency-cap"`, zero DB writes beyond the advisory-lock check itself) once `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` @@ -466,7 +477,7 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ this box's django-q2 worker processes. See "Concurrency cap" below for the full mechanism and the incident that motivated it — distinct from, and a proactive complement to, the envelope's own reactive host-load bar. -6. **Stage C** (COMPUTE sequential, per-card, not pooled — a micro-batch is +7. **Stage C** (COMPUTE sequential, per-card, not pooled — a micro-batch is far too small for BULK mode's process-pool concurrency to help; FETCH overlapped with compute since 2026-07-25, issue #472 — see "Evidence transfer and decoupled fetch-ahead" below) — the same @@ -477,19 +488,19 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ immediately and records a fresh trip (instant-pause bar) — in-flight, already-committed work stays committed; Stage D below still runs against whatever was reached ("in-flight work drains, nothing NEW starts"). -7. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ +8. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ `run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in the same escalation order every BULK-mode invocation already uses. Each of these already calls `resolve_and_persist_printing` internally for every card it touches — this is what satisfies §3 decision (4)'s "scoped incremental per-touch consensus recompute" with no separate consensus step in the dispatcher at all. -8. **Ledger write, then concurrency-cap slot release** — one `PilotRunLedger` +9. **Ledger write, then concurrency-cap slot release** — one `PilotRunLedger` row per micro-batch (see "Observability" below), then the slot acquired in - step 5 is released (always, including on an exception - see "Concurrency + step 6 is released (always, including on an exception - see "Concurrency cap" below). -Before adding a calculator to step 7, check which class its inference falls +Before adding a calculator to step 8, check which class its inference falls into — [`docs/theory.md`](../theory.md) §10a. Two of the three are safe to dispatch per batch and one is not, and the split is not about cost: a calculator whose conclusion is a **count over the population** (e.g. @@ -501,6 +512,69 @@ otherwise a card whose only sibling sits outside the batch silently loses it and nothing errors. PR #541's `run_d0_sibling_artist_propagation` scoping is the worked example of the second. +### The host-load soft brake (2026-08-05) + +`operating_envelope.HOST_LOAD_CEILING` (7.0) is a binary cliff: the instant +a fresh sample reads above it, the dispatch halts and demands a fresh owner +action to resume (no self-resume, above). Two real passes tripped on narrow +overshoots of that cliff a day apart (7.0796, then 7.17236328125 — 1.1% and +2.5% over) despite the box otherwise running comfortably under load, each +costing a stopped pipeline and a human interaction. +`cardpicker.stage_e_load_brake` adds a throttle on approach, upstream of +that cliff — it does not move the cliff itself, and every genuine breach +still halts exactly as before. + +**Mechanism.** Step 3 of the ordering above re-samples `os.getloadavg()` +independently of the envelope's own sample a moment later, and classifies +the reading into a band: + +- `load < STAGE_E_HOST_LOAD_SOFT_CEILING` (default 6.0) — proceed + immediately. The common case; zero added cost. +- `soft <= load <= HOST_LOAD_CEILING` — sleep, re-sample, repeat. +- `load > HOST_LOAD_CEILING` — stop braking at once; the envelope's own + fresh sample, checked a moment later, trips honestly. The brake never + itself decides "trip" and never suppresses a genuine breach. +- cumulative wait exceeds `STAGE_E_LOAD_BRAKE_MAX_WAIT_S` (default 240s) — + proceed anyway. Best-effort, and must never deadlock an unattended + multi-hour run. + +**Why it reduces load rather than just delaying it.** The load a pass +generates is mostly its own concurrency +(`settings.STAGE_E_MAX_CONCURRENT_DISPATCHES`). When every resident +dispatcher enters the band, each one pauses independently at its own next +batch boundary — the resident process count falls, load decays, and they +resume. The pass self-throttles down to whatever concurrency fits under the +ceiling, continuously, instead of running at full concurrency until it hits +the wall and halts. + +**Jitter is load-bearing.** Every dispatcher reads the same global +`os.getloadavg()`; without randomizing each sleep +(`interval * uniform(0.75, 1.5)`), every braking process would back off and +resume in lockstep — a sawtooth, and a thundering herd on every resume. + +**Settings** (all default to values that make the brake active out of the +box — no opt-in required): + +- `STAGE_E_HOST_LOAD_SOFT_CEILING` (default `6.0`) — the top of the band. +- `STAGE_E_LOAD_BRAKE_INTERVAL_S` (default `15`) — base sleep per + iteration, before jitter. +- `STAGE_E_LOAD_BRAKE_MAX_WAIT_S` (default `240`) — the absolute bound on + one dispatch call's cumulative brake time. + +**Seeing whether it engaged.** `DispatchOutcome.load_brake_waits`/ +`load_brake_seconds`, merged into `PilotRunLedger.counters` on every +completed micro-batch (see "Observability" below) — both `0`/`0.0` when the +brake never engaged, which is expected on a quiet box. A run showing +non-zero values under contention and zero when the box is quiet is the +brake behaving as designed, not a bug. + +**Failure posture.** Wrapped in a bare `try`/`except` that proceeds as if +unbraked on any error (a malformed setting, a transient `os.getloadavg` +failure) — matching `stage_e_batch_sizing.resolve_micro_batch_size`'s own +stated posture that a typo'd env var must not be able to take the run down. +The brake is a convenience; the envelope's own hard ceiling does not depend +on it. + ### Evidence transfer and decoupled fetch-ahead (issues #473 PR-2 and #472, 2026-07-25) Both landed together (owner-approved fold — "same function, one coherent @@ -959,8 +1033,11 @@ streamed micro-batch" below; non-zero occasionally is healthy, not a bug), `stage_d_illustration_votes`/`stage_d_illustration_already_voted` (2026-07-28, issue #507 — illustration deduction calculator wired into the streaming conveyor), `stage_d_slow_path_routed`, `elapsed_s`, `peak_rss_mb` (via the same -`process_metrics.get_process_rss_mb` Phase 1 wired in), and `lockout_trip_id` -(non-null only when a Google lockout tripped mid-batch). A halted call +`process_metrics.get_process_rss_mb` Phase 1 wired in), `lockout_trip_id` +(non-null only when a Google lockout tripped mid-batch), and +`load_brake_waits`/`load_brake_seconds` (2026-08-05 — see "The host-load +soft brake" below; both `0`/`0.0` whenever the brake never engaged for this +batch, which is the common case). A halted call (`disabled`/`halted-open-trip`/`halted-new-trip`) writes NO ledger row at all — a halted dispatch never partially starts, so there's nothing to record beyond the `EnvelopeTrip` row `check_envelope`