diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b50917ec..2aed496d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -109,7 +109,7 @@ jobs: run: cargo build --workspace rust-test: - name: Rust tests (${{ matrix.shard }}) + name: Rust tests (pg${{ matrix.postgres }}, ${{ matrix.shard }}) needs: [rust-build] if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'full-ci') runs-on: ubuntu-latest @@ -123,6 +123,11 @@ jobs: strategy: fail-fast: false matrix: + # Postgres majors we support. The schema, its partitioning, and the + # planner behaviour the claim path depends on are all version- + # sensitive, so the whole sharded suite runs on each major rather + # than a smoke subset. + postgres: ["17", "18"] shard: - migrations-1 - migrations-2 @@ -132,7 +137,7 @@ jobs: - rest services: postgres: - image: postgres:17-alpine + image: postgres:${{ matrix.postgres }}-alpine env: POSTGRES_PASSWORD: postgres POSTGRES_DB: awa_test @@ -164,14 +169,18 @@ jobs: # canonical, so this leg re-runs it with the queue-storage engine active on a # fresh database (a separate DB avoids cross-engine job_unique_claims carryover). rust-test-queue-storage: - name: Rust tests (queue_storage) + name: Rust tests (queue_storage, pg${{ matrix.postgres }}) needs: [rust-build] if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'full-ci') runs-on: ubuntu-latest timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + postgres: ["17", "18"] services: postgres: - image: postgres:17-alpine + image: postgres:${{ matrix.postgres }}-alpine env: POSTGRES_PASSWORD: postgres POSTGRES_DB: awa_test @@ -274,10 +283,20 @@ jobs: # ─── Python (awa-python) ──────────────────────────────── python-build-test: - name: Python build + test + name: Python build + test (py${{ matrix.python }}) if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'full-ci') runs-on: ubuntu-latest timeout-minutes: 60 + strategy: + fail-fast: false + matrix: + # The newest two interpreters only. The wheel is abi3-py310 so one + # artifact covers the whole 3.10+ support window, and abi3 keeps the + # C API stable across it; these legs exercise the SDK's asyncio + # integration (pyo3-async-runtimes), which is where version skew + # actually lands — and it lands on the newest releases. When a new + # CPython ships, add it here and drop the oldest. + python: ["3.13", "3.14"] services: postgres: image: postgres:17-alpine @@ -300,14 +319,15 @@ jobs: - uses: Swatinem/rust-cache@v2 with: workspaces: "awa-python -> awa-python/target" + key: "py${{ matrix.python }}" - uses: actions/setup-python@v6 with: - python-version: "3.12" + python-version: "${{ matrix.python }}" - name: Install uv uses: astral-sh/setup-uv@v8.2.0 - name: Create venv and install deps run: | - uv venv .venv + uv venv .venv --python "${{ matrix.python }}" uv pip install maturin pytest pytest-asyncio asyncpg django greenlet psycopg psycopg-binary sqlalchemy - name: Build extension (maturin develop) run: .venv/bin/maturin develop diff --git a/AGENTS.md b/AGENTS.md index 91d03643..c31b8a2b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,6 +29,59 @@ DATABASE_URL=postgres://postgres:test@localhost:15432/awa_test uv run pytest tes Read the redirected log and check the real exit status; do not trust a green exit code from a piped or backgrounded command. +### Version matrix + +CI runs the sharded Rust suite and the queue-storage leg against **Postgres 17 +and 18** — both majors run the whole suite, because the schema, its +partitioning, and the planner behaviour the claim path depends on are all +version-sensitive. + +The Python suite runs on the **newest two CPython releases**. The support +window is whatever `requires-python` says (3.10+), and the `abi3-py310` wheel +covers all of it from one artifact, so the legs are not there to prove the +ABI — they exercise the asyncio integration (`pyo3-async-runtimes`), which is +where version skew actually lands. The `pyproject.toml` classifiers therefore +track the support window, not the test matrix. When a new CPython ships, add it +to the matrix and drop the oldest. + +### Contention-sensitive assertions in the nightly suites + +The nightly chaos and benchmark suites run on shared runners whose CPU +allocation varies run to run. Three assertion shapes are sensitive to that and +must go through `awa/tests/ci_timing.rs` rather than hard-coding a bound: + +| Shape | Helper | Failure if too tight | +| --- | --- | --- | +| Wall-clock wait for a state | `scaled_timeout` | Spurious timeout | +| `heartbeat_staleness` on a chaos client | `scaled_staleness` | The runtime rescues a *live* attempt and the test sees a genuine duplicate completion — a margin bug that reads as a correctness bug | +| "at least N of these happened" floor | `contention_floor` | Gate fires while every invariant it exists for is intact | + +All three only ever loosen a bound, and only when `CI` is set, so a local run +keeps the strict values and a real regression still fails fast on a developer +machine. `AWA_CHAOS_TIMEOUT_MULTIPLIER` overrides the factor (clamped to +`>= 1.0`). + +A minimum-progress floor must sit below the *observed* operating point, not the +nominal one. The receipt-plane rotation gate is the cautionary example (#399): +its floor came from the 1s rotate interval's tick rate, but rotation is driven +by the maintenance loop reaching a rotate decision, so the healthy steady state +sat flush against the floor and the gate fired at 41-vs-45 with every +architectural bound perfect. + +Measurement windows are *not* timeouts — `recv_until`'s duration defines what a +benchmark samples, so it stays unscaled. + +### Connection budget in the Python suite + +`awa-python/tests/conftest.py` defaults test clients to a small pool and fails +the session if backend count grows across it. Two conventions follow (#420): + +- A fixture that builds a client must `yield` it and close it in a `finally`, + never `return` it. Returning leaves pool teardown to GC timing, which parks + server connections for up to sqlx's 10-minute idle timeout. +- Pass `max_connections` explicitly only when the test is *about* pool sizing. + Otherwise take the conftest default, so one test cannot starve the next. + ## Schema Migrations Migrations are forward-only and must stay rolling-upgrade compatible. Version diff --git a/CHANGELOG.md b/CHANGELOG.md index ac25f35b..bc5801f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **Nightly flake gates now carry runner-contention margin ([#399](https://github.com/hardbyte/awa/issues/399), [#434](https://github.com/hardbyte/awa/issues/434)).** Four assertion shapes in the chaos and benchmark suites were tight enough that shared-runner CPU contention failed them while every invariant they exist for was intact, eroding the 14-consecutive-green-nightlies release gate. `awa/tests/ci_timing.rs` now holds the scaling for all of them, and it only ever loosens a bound, and only when `CI` is set: + - The mixed-fleet chaos test set `heartbeat_staleness` to 250ms against a 50ms heartbeat interval. Under contention a *live* worker's heartbeat missed that window, so the runtime correctly rescued a healthy attempt and the test saw a genuine duplicate completion — a margin problem that read as a correctness bug. Chaos clients scale the staleness window via `scaled_staleness` while leaving the heartbeat and rescue *intervals* at chaos cadence, so the rescue path is still exercised. + - That test's assert-after-drain step watched both completion streams for a fixed 250ms quiet window, the same race [#335](https://github.com/hardbyte/awa/issues/335) fixed in `test_weight_proportionality`: a duplicate arriving at 251ms was missed, and on a slow runner the window expired while work was still moving. It now waits for the queue itself to reach a terminal state — the authoritative drain signal — then drains both streams with `try_recv`, so duplicate detection no longer depends on wall-clock timing. + - The receipt-plane rotation floor (`>= 45` rotations in 180s) was derived from the 1s rotate interval's nominal tick rate, but rotation is driven by the maintenance loop reaching a rotate decision, so the healthy steady state is ~40-45 and the floor had no margin. It fired at 41 with zero dead rows in every claim and closure partition. The floor is now divided by the contention multiplier, keeping a genuine pinned-ring floor (a pinned ring shows ~0) with ~3x margin. + - `scheduling_benchmark_test.rs` had no contention scaling at all, so its `wait_for_leader`/`wait_for_dispatch` readiness gates and post-handler completion waits were hard 5s/30s deadlines. They now scale. Measurement windows (`recv_until`) deliberately do not — their duration defines what the benchmark samples. + +- **Python suite no longer exhausts the server's connection slots ([#420](https://github.com/hardbyte/awa/issues/420)).** Late in a full run, one or two tests ERRORed at *fixture setup* with `pool timed out while waiting for an open connection` — always ~15 minutes in, always rerun-clearable. `AsyncClient`/`Client` default to a 10-connection pool, and sqlx holds an idle connection for its 10-minute idle timeout, so pools from tests that finished minutes earlier kept slots parked while each new fixture asked for up to 10 more; when the server had none left, sqlx could not grow the pool and its 30s acquire timeout expired. The victim was therefore whichever test came next, not the test at fault. A new `awa-python/tests/conftest.py` caps what tests ask for, and two fixtures that built a client and `return`ed it — leaving pool teardown to GC timing rather than closing it — now yield and close deterministically. A session-scoped guard fails the run with a named backend count if a `return`-style fixture reappears. Verified by re-running the full suite against a deliberately constrained 30-connection server: the pre-fix run parked 11-19 of 30 backends on pools belonging to tests that had finished minutes earlier, and the post-fix run holds 3 and passes all 312. + - **`awa.jobs` is receipt-aware and no longer pathologically slow under backlog ([#422](https://github.com/hardbyte/awa/issues/422), migration v044).** The SQL-compat view had two defects found during the 0.6.1→0.7 upgrade rehearsal. *Performance:* the available-branch filter spelled `lane_seq >= sequence_next_value(claims.seq_name)` inline against the partitioned `ready_entries` scan; `sequence_next_value` is a VOLATILE PL/pgSQL function, so the planner could neither use the predicate as an index bound nor cache it across rows — every ready child was scanned in full and each surviving row paid a dynamic-SQL catalog round-trip (measured ~90ms → ~6ms for the branch on a 20k-row backlog; the gap grows with un-pruned sealed generations). v044 materializes the per-lane cursors once per statement — the same shape the claim path has used since v027/v039 — and reading each cursor once is also a more consistent snapshot than per-row evaluation. *Correctness:* receipt claims that never materialized into `leases` (row-local `lease_claims`, and by default compact `lease_claim_batches`) were invisible, so `SELECT state, count(*) FROM awa.jobs` reported `running=0` while workers held live work. The view now carries the open-receipt legs the admin surface shipped in #410 (`admin::state_counts` / `queue_storage::open_receipt_running_claims_sql`), projecting the claim ledger's own attempt / claim-time / deadline values and anti-joined against closures, closure batches, materialised leases, and terminal/deferred/DLQ supersession — a job reports at most one state outside the brief post-commit cursor-lag window the claim protocol itself has (identical to the admin surfaces; it self-heals on the lane's next claim). Rows previously returned are unchanged; the running rows are additive. **No 0.6.x backport is required:** released 0.6.x workers have no schema-version gate at startup, so both upgrade orderings keep working — only a ≤0.6.x *migrator* refuses via the #392 fail-safe, which is correct (past-v043 migrations are applied by the 0.7 binary). Verified by the compat matrix against pinned release artifacts (`scripts/compat-matrix.sh` now runs a `forward-0.6.6` leg alongside 0.6.2/0.6.0/0.5.7) plus a released-binary enqueue→claim→complete rehearsal on a v044 database. - **`awa migrate --extract-to` wrote only the first 15 migrations.** The target path was built by substituting the migration description into the filename, so v017 — whose description contains `/` — resolved to a nested directory that does not exist. The command aborted there, leaving a partial extraction that looked plausible but silently omitted two thirds of the schema, breaking the documented external-runner workflow for every range reaching v017. Only path separators are now substituted, so every filename this tool has already published stays byte-identical — re-extracting into an existing directory cannot leave two files for the same version. `V17` and `V21` are the only names that change, and neither could previously be written at all. All paths are computed and checked for collisions before anything is written, and a write failure names the file. A test asserts every migration is extracted with byte-identical SQL. @@ -25,6 +33,8 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Added +- **CI covers Postgres 18; the Python matrix moves to the newest two interpreters.** The sharded Rust suite and the queue-storage integration leg now run against Postgres 17 *and* 18 — the schema, its partitioning, and the planner behaviour the claim path depends on are all version-sensitive, so both majors run the whole suite rather than a smoke subset. (Postgres 18 was already in use by the telemetry-validation job and the Python nightly, unsystematically.) The Python suite moves from CPython 3.12 to 3.13 and 3.14: the `abi3-py310` wheel covers the whole `requires-python` window from one artifact, so the legs exist to exercise the asyncio integration where version skew actually lands rather than to re-prove the ABI on every supported interpreter. `pyproject.toml` now carries per-version classifiers for the support window it has always claimed. + - **Migration atomicity and idempotency are now enforced by tests, not convention.** Four no-database unit tests assert the properties the single-transaction runner depends on: every migration step is transaction-safe (no `CONCURRENTLY`, `VACUUM`, or statement-level `BEGIN`/`COMMIT`/`ROLLBACK`/`SAVEPOINT`), every top-level `CREATE` is guarded so a replay is a no-op, every migration records its own version exactly once with `ON CONFLICT DO NOTHING`, and versions stay unique and ordered up to `CURRENT_VERSION`. Integration tests prove the runtime behaviour end-to-end: an injected mid-run DDL failure rolls back every earlier step (fresh install and upgrade alike) and is never visible to another session, replaying the full ordered set over a migrated database leaves a byte-identical catalog, and applying every migration twice is indistinguishable from a clean install. `awa migrate` itself is covered by a new CLI suite spanning the applied run, the rendered SQL, and the extracted files. `migrations::MIGRATION_LOCK_KEY` (and its Python counterpart `awa.migration_lock_key()`) is public so emitted SQL takes the same lock as the runner instead of a copy that could drift. ### Changed diff --git a/awa-python/pyproject.toml b/awa-python/pyproject.toml index a37d83c7..f2eba701 100644 --- a/awa-python/pyproject.toml +++ b/awa-python/pyproject.toml @@ -17,6 +17,11 @@ classifiers = [ "Programming Language :: Rust", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "License :: OSI Approved :: MIT License", "License :: OSI Approved :: Apache Software License", "Topic :: Database", diff --git a/awa-python/tests/conftest.py b/awa-python/tests/conftest.py new file mode 100644 index 00000000..6e5aa9a6 --- /dev/null +++ b/awa-python/tests/conftest.py @@ -0,0 +1,119 @@ +"""Suite-wide fixtures and guards for the Python test session. + +#420: late in a full run, one or two tests ERROR at *fixture setup* with +`pool timed out while waiting for an open connection`, always ~15 minutes +in, and a rerun always passes. Two things feed it, both addressed here. + +**Peak connection demand.** `AsyncClient`/`Client` default to a +10-connection pool — a sensible application default, and far more than any +test needs. Re-running the suite against a `max_connections=30` server +makes the mechanism visible: 19 of 30 backends sit `idle`, held by pools +from tests that finished minutes earlier (sqlx keeps an idle connection for +its idle timeout, which defaults to 10 minutes). Each new fixture then +wants up to 10 more. When the server has none left, sqlx cannot grow the +pool and its 30s acquire timeout expires — which is the reported error, and +why the victim is whichever test came next rather than the test at fault. +`_TEST_POOL_MAX_CONNECTIONS` below caps what tests ask for. With the cap, the +same 312-test run on the same 30-connection server holds 3 backends where it +previously held 11-19, and passes. Callers that pass `max_connections` +explicitly are untouched. + +**Non-deterministic pool teardown.** Two fixtures built a client and +`return`ed it instead of yielding and closing it, leaving teardown to +whenever CPython collected the object. Those are fixed; +`connection_leak_guard` keeps them fixed. +""" + +import os + +import pytest + +import awa + +DATABASE_URL = os.environ.get( + "DATABASE_URL", "postgres://postgres:test@localhost:15432/awa_test" +) + +# Enough for a client running a small worker fleet: the LISTEN/NOTIFY +# connection is held for the client's lifetime, and the dispatcher, +# heartbeat, and maintenance loops each want one alongside the handler's +# own queries. The library default of 10 is right for applications and +# more than any test here needs. +_TEST_POOL_MAX_CONNECTIONS = 5 + + +def _cap_pool_size(cls: type) -> None: + """Default this client class's pool to a test-sized one. + + Wraps `__init__` rather than editing the 27 test modules that + construct clients. An explicit `max_connections` argument still wins, + so a test that deliberately exercises pool sizing is unaffected. + """ + original = cls.__init__ + + def __init__(self, database_url, max_connections=None, **kwargs): + if max_connections is None: + max_connections = _TEST_POOL_MAX_CONNECTIONS + original(self, database_url, max_connections, **kwargs) + + __init__.__wrapped__ = original + cls.__init__ = __init__ + + +for _cls in (awa.AsyncClient, awa.Client): + _cap_pool_size(_cls) + + +# Slack over the baseline. Backend teardown is asynchronous on the server +# side, so a just-closed pool can still be visible for a moment. Kept below +# `_TEST_POOL_MAX_CONNECTIONS` so one leaked pool cannot hide under it. +CONNECTION_LEAK_SLACK = 3 + + +def _backend_count() -> int | None: + """Backends on this database, or None if we cannot ask.""" + try: + client = awa.Client(DATABASE_URL, max_connections=1) + except Exception: + return None + try: + tx = client.transaction() + row = tx.fetch_one( + """ + SELECT count(*) AS n + FROM pg_stat_activity + WHERE datname = current_database() + AND pid <> pg_backend_pid() + """ + ) + tx.commit() + return int(row["n"]) + except Exception: + return None + finally: + client.close() + + +@pytest.fixture(scope="session", autouse=True) +def connection_leak_guard(): + baseline = _backend_count() + yield + if baseline is None: + return + + final = _backend_count() + if final is None: + return + + growth = final - baseline + print( + f"\n[connection-guard] backends at session start={baseline} " + f"end={final} growth={growth}" + ) + assert growth <= CONNECTION_LEAK_SLACK, ( + f"Connection leak: {growth} more backends at session end than at " + f"start (baseline={baseline}, final={final}, slack=" + f"{CONNECTION_LEAK_SLACK}). A fixture is building an awa client and " + f"returning it without closing it — use `yield c` with " + f"`await c.close()` / `c.close()` in a finally block." + ) diff --git a/awa-python/tests/mixed_fleet_helper.py b/awa-python/tests/mixed_fleet_helper.py index 01d64dc8..ece2aa9f 100644 --- a/awa-python/tests/mixed_fleet_helper.py +++ b/awa-python/tests/mixed_fleet_helper.py @@ -1,4 +1,5 @@ import asyncio +import math import os from dataclasses import dataclass @@ -16,6 +17,27 @@ class SimpleChaosJob: seq: int +def _staleness_ms(nominal_ms: int) -> int: + """Scale a heartbeat-staleness window by the runner-contention + multiplier the Rust side resolved and passed down. + + Unscaled, a contended runner can stall this process's event loop past + the window and have the runtime rescue a *live* attempt, producing a + genuine duplicate completion in a mixed-fleet test. Mirrors + `awa/tests/ci_timing.rs::scaled_staleness`, including its refusal of + non-finite or oversized values. + """ + raw = os.environ.get("AWA_CHAOS_TIMEOUT_MULTIPLIER") + try: + multiplier = float(raw) if raw is not None else 1.0 + except ValueError: + multiplier = 1.0 + if not math.isfinite(multiplier): + multiplier = 1.0 + multiplier = min(max(multiplier, 1.0), 100.0) + return int(nominal_ms * multiplier) + + async def main() -> None: database_url = os.environ["DATABASE_URL"] queue = os.environ["MIXED_QUEUE"] @@ -46,7 +68,7 @@ async def handle(job): heartbeat_interval_ms=50, promote_interval_ms=50, heartbeat_rescue_interval_ms=100, - heartbeat_staleness_ms=250, + heartbeat_staleness_ms=_staleness_ms(250), ) print(f"READY mode={mode} pid={os.getpid()}", flush=True) await asyncio.Event().wait() @@ -96,7 +118,7 @@ async def handle(job): await client.start( [queue_config], heartbeat_rescue_interval_ms=100, - heartbeat_staleness_ms=500, + heartbeat_staleness_ms=_staleness_ms(500), deadline_rescue_interval_ms=100, **start_kwargs, ) diff --git a/awa-python/tests/test_bridge.py b/awa-python/tests/test_bridge.py index 1abb6568..2b2dee11 100644 --- a/awa-python/tests/test_bridge.py +++ b/awa-python/tests/test_bridge.py @@ -45,9 +45,12 @@ class BridgePayment: def awa_client(): """Awa sync client for verifying job insertion.""" c = awa.Client(DATABASE_URL) - c.migrate() - reset_sync(c) - return c + try: + c.migrate() + reset_sync(c) + yield c + finally: + c.close() def _configure_django(): diff --git a/awa-python/tests/test_unique_insert.py b/awa-python/tests/test_unique_insert.py index 6e14d55a..04d0d184 100644 --- a/awa-python/tests/test_unique_insert.py +++ b/awa-python/tests/test_unique_insert.py @@ -21,12 +21,29 @@ @pytest.fixture async def client(): c = awa.AsyncClient(DATABASE_URL) - await c.migrate() - await reset_async(c) - tx = await c.transaction() - await tx.execute("DELETE FROM awa.jobs WHERE queue LIKE 'uniq_%'") - await tx.commit() - return c + # Everything after construction goes inside the try: a failure in + # migrate/reset would otherwise skip close() and park pooled + # connections, which is the exact failure this file's guard exists for. + try: + await c.migrate() + await reset_async(c) + tx = await c.transaction() + await tx.execute("DELETE FROM awa.jobs WHERE queue LIKE 'uniq_%'") + await tx.commit() + yield c + finally: + await c.close() + + +@pytest.fixture +def sync_client(): + c = awa.Client(DATABASE_URL) + try: + c.migrate() + reset_sync(c) + yield c + finally: + c.close() @dataclass @@ -292,11 +309,9 @@ async def test_no_unique_opts_allows_duplicates(client): # ── Sync variant ───────────────────────────────────────────────────── -def test_unique_insert_sync(): +def test_unique_insert_sync(sync_client): """Sync insert with unique_opts works.""" - c = awa.Client(DATABASE_URL) - c.migrate() - reset_sync(c) + c = sync_client tx = c.transaction() tx.execute("DELETE FROM awa.jobs WHERE queue = 'uniq_sync'") tx.commit() @@ -367,11 +382,9 @@ async def test_unique_insert_transaction_rollback(client): assert job.id > 0 -def test_unique_insert_in_sync_transaction(): +def test_unique_insert_in_sync_transaction(sync_client): """Sync transactional unique insert works.""" - c = awa.Client(DATABASE_URL) - c.migrate() - reset_sync(c) + c = sync_client tx = c.transaction() tx.execute("DELETE FROM awa.jobs WHERE queue = 'uniq_stx'") tx.commit() diff --git a/awa/tests/chaos_suite_test.rs b/awa/tests/chaos_suite_test.rs index c020d373..7d5a6a1c 100644 --- a/awa/tests/chaos_suite_test.rs +++ b/awa/tests/chaos_suite_test.rs @@ -20,6 +20,9 @@ use tokio::process::{Child, Command}; use tokio::sync::mpsc; use uuid::Uuid; +mod ci_timing; +use ci_timing::{chaos_timeout_multiplier, scaled_staleness, scaled_timeout}; + /// Serializes the chaos tests: the maintenance leader-election advisory lock /// is global per database, so any test's live client can hold leadership /// while another test waits for one of ITS clients to be elected. Running @@ -253,24 +256,6 @@ fn state_count(counts: &HashMap, state: &str) -> i64 { counts.get(state).copied().unwrap_or(0) } -fn chaos_timeout_multiplier() -> f64 { - if let Ok(raw) = std::env::var("AWA_CHAOS_TIMEOUT_MULTIPLIER") { - if let Ok(parsed) = raw.parse::() { - return parsed.max(1.0); - } - } - - if std::env::var_os("CI").is_some() { - 3.0 - } else { - 1.0 - } -} - -fn scaled_timeout(timeout: Duration) -> Duration { - timeout.mul_f64(chaos_timeout_multiplier()) -} - async fn wait_for_counts( pool: &sqlx::PgPool, queue: &str, @@ -886,6 +871,24 @@ impl PythonHelperProcess { } } + /// Close the helper's stdout and wait for the reader task to reach EOF. + /// + /// `stdout_lines.try_recv()` can report `Empty` while a line the helper + /// already flushed is still sitting in the OS pipe, because a spawned + /// task has to move it into the channel first. Killing the child closes + /// the write end — bytes already in the pipe stay readable — so once the + /// reader has seen EOF and dropped its sender, the channel holds every + /// line the helper emitted and nothing further can arrive. + async fn drain_stdout(&mut self) { + if self.child.id().is_some() { + let _ = self.child.kill().await; + let _ = self.child.wait().await; + } + // JoinHandle is Unpin, so this awaits completion without consuming it; + // `stop()` may still abort it afterwards. + let _ = (&mut self.stdout_reader).await; + } + async fn stop(mut self) { self.stdout_reader.abort(); if self.child.id().is_none() { @@ -930,6 +933,16 @@ async fn start_python_helper( .env("DATABASE_URL", database_url()) .env("MIXED_QUEUE", queue) .env("MIXED_MODE", mode) + // The helper configures its own client, so it needs the same + // staleness margin as the Rust side of a mixed-fleet test — a + // contended Python event loop can miss an unscaled window and have + // a live attempt rescued, which is the duplicate-completion flake + // from the Rust side all over again. Pass the *resolved* multiplier + // rather than relying on the child re-deriving it from `CI`. + .env( + "AWA_CHAOS_TIMEOUT_MULTIPLIER", + chaos_timeout_multiplier().to_string(), + ) .env("PYTHONUNBUFFERED", "1") .stdout(Stdio::piped()) .stderr(Stdio::inherit()); @@ -1155,7 +1168,7 @@ fn complete_client(pool: sqlx::PgPool, queue: &str) -> Client { .heartbeat_interval(Duration::from_millis(50)) .promote_interval(Duration::from_millis(50)) .heartbeat_rescue_interval(Duration::from_millis(100)) - .heartbeat_staleness(Duration::from_millis(250)) + .heartbeat_staleness(scaled_staleness(Duration::from_millis(250))) .leader_election_interval(Duration::from_millis(100)) .leader_check_interval(Duration::from_millis(100)) .register_worker(CompleteWorker) @@ -1802,7 +1815,7 @@ async fn test_mixed_rust_and_python_workers_share_same_queue() { .heartbeat_interval(Duration::from_millis(50)) .promote_interval(Duration::from_millis(50)) .heartbeat_rescue_interval(Duration::from_millis(100)) - .heartbeat_staleness(Duration::from_millis(250)) + .heartbeat_staleness(scaled_staleness(Duration::from_millis(250))) .leader_election_interval(Duration::from_millis(100)) .leader_check_interval(Duration::from_millis(100)) .register_worker(MixedFleetRustWorker { tx }) @@ -1915,38 +1928,62 @@ async fn test_mixed_rust_and_python_workers_share_same_queue() { } } - let quiet_deadline = tokio::time::sleep(scaled_timeout(Duration::from_millis(250))); - tokio::pin!(quiet_deadline); - loop { - tokio::select! { - marker = rx.recv() => { - let marker = marker.expect("Rust mixed-fleet receiver closed unexpectedly"); - assert!( - expected_markers.contains(&marker), - "Unexpected Rust marker processed after expected drain: {marker}" - ); - assert!( - completed_markers.insert(marker.clone()), - "Marker completed more than once after expected drain: {marker}" - ); - } - line = python_worker.stdout_lines.recv() => { - let line = line.expect("Python mixed-fleet worker stdout closed unexpectedly"); - if line.contains("COMPLETE mode=worker_chaos_probe") { - let marker = mixed_fleet_marker_from_line(&line) - .unwrap_or_else(|| panic!("Python completion line missing marker: {line}")) - .to_string(); - assert!( - expected_markers.contains(&marker), - "Unexpected Python marker processed after expected drain: {marker}" - ); - assert!( - completed_markers.insert(marker.clone()), - "Marker completed more than once after expected drain: {marker}" - ); - } - } - () = &mut quiet_deadline => break, + // Both fleets have reported every expected marker. Confirm the + // queue itself is terminal before asserting no marker completed + // twice: the previous shape watched the two completion streams for + // a fixed 250ms quiet window, which is an assert-after-drain race + // (#335, #399) — a duplicate arriving at 251ms was missed, and on a + // contended runner the window expired while work was still moving. + // Queue state is the authoritative drain signal, so wait on that, + // then drain whatever the streams have already buffered. + wait_for_counts( + &pool, + &queue, + |counts| { + state_count(counts, "completed") == expected_markers.len() as i64 + && state_count(counts, "running") == 0 + && state_count(counts, "available") == 0 + && state_count(counts, "retryable") == 0 + && state_count(counts, "scheduled") == 0 + }, + Duration::from_secs(30), + ) + .await; + + // The queue is terminal, so no further completion can be produced. + // Anything still in either stream is a duplicate or an unexpected + // marker, and both are real failures. + // + // The Rust worker sends on this channel from inside `perform`, before + // the completion is written, so a terminal queue means every send is + // already queued and `try_recv` sees it. + while let Ok(marker) = rx.try_recv() { + assert!( + expected_markers.contains(&marker), + "Unexpected Rust marker processed after queue drain: {marker}" + ); + assert!( + completed_markers.insert(marker.clone()), + "Marker completed more than once after queue drain: {marker}" + ); + } + // The Python helper needs the extra hop closed first: its line goes + // through an OS pipe and a reader task, neither of which is ordered + // against the database commit. + python_worker.drain_stdout().await; + while let Some(line) = python_worker.stdout_lines.recv().await { + if line.contains("COMPLETE mode=worker_chaos_probe") { + let marker = mixed_fleet_marker_from_line(&line) + .unwrap_or_else(|| panic!("Python completion line missing marker: {line}")) + .to_string(); + assert!( + expected_markers.contains(&marker), + "Unexpected Python marker processed after queue drain: {marker}" + ); + assert!( + completed_markers.insert(marker.clone()), + "Marker completed more than once after queue drain: {marker}" + ); } } @@ -2003,7 +2040,7 @@ async fn test_runtime_recovers_after_terminating_postgres_connections() { ) .heartbeat_interval(Duration::from_millis(50)) .heartbeat_rescue_interval(Duration::from_millis(100)) - .heartbeat_staleness(Duration::from_millis(250)) + .heartbeat_staleness(scaled_staleness(Duration::from_millis(250))) .promote_interval(Duration::from_millis(50)) .leader_election_interval(Duration::from_millis(100)) .leader_check_interval(Duration::from_millis(100)) @@ -2410,7 +2447,7 @@ async fn test_full_postgres_outage_recovers_with_metrics() { ) .heartbeat_interval(Duration::from_millis(50)) .heartbeat_rescue_interval(Duration::from_millis(100)) - .heartbeat_staleness(Duration::from_millis(250)) + .heartbeat_staleness(scaled_staleness(Duration::from_millis(250))) .promote_interval(Duration::from_millis(50)) .leader_election_interval(Duration::from_millis(100)) .leader_check_interval(Duration::from_millis(100)) diff --git a/awa/tests/ci_timing.rs b/awa/tests/ci_timing.rs new file mode 100644 index 00000000..3704a0ba --- /dev/null +++ b/awa/tests/ci_timing.rs @@ -0,0 +1,186 @@ +//! Shared CI contention scaling for the nightly chaos and benchmark suites. +//! +//! The nightly suites run on shared GitHub runners whose CPU allocation +//! varies run to run. Three assertion shapes are sensitive to that and +//! flaked repeatedly through 2026-07 (#399, #434): +//! +//! 1. **Wall-clock waits.** A test waits N seconds for a state the +//! runtime reaches in milliseconds when it has a core to itself. +//! Scale with [`scaled_timeout`]. +//! +//! 2. **Aggressive rescue cadences.** A chaos client sets +//! `heartbeat_staleness` a few multiples above `heartbeat_interval` +//! so its own backdating triggers rescue promptly. Under contention +//! a *live* worker's heartbeat can miss that window, so the runtime +//! correctly rescues a healthy attempt — and the test sees a genuine +//! duplicate completion, which reads as a correctness failure. Scale +//! with [`scaled_staleness`]: the interval stays fast so rescue is +//! still exercised, but the staleness window gains the same margin +//! as the waits around it. +//! +//! 3. **Minimum-progress floors.** A gate asserts "at least N of these +//! happened" to catch a stalled subsystem. The floor must sit far +//! below the *observed* operating point, not just below the nominal +//! one. Scale with [`contention_floor`]. +//! +//! Scaling only ever loosens a bound, and only on CI. A local run keeps +//! the strict values, so a real regression still fails fast on a +//! developer machine. Override with `AWA_CHAOS_TIMEOUT_MULTIPLIER`. +#![allow(dead_code)] + +use std::time::Duration; + +/// Upper bound on the override. `Duration::mul_f64` panics on a non-finite +/// or overflowing result, so an unbounded override would turn a typo into a +/// failed nightly rather than a loose bound. +const MAX_MULTIPLIER: f64 = 100.0; + +/// How much slack to give contention-sensitive bounds. `1.0` locally, +/// `3.0` on CI, or the `AWA_CHAOS_TIMEOUT_MULTIPLIER` override (clamped to +/// `1.0 ..= MAX_MULTIPLIER` so the override can only ever loosen, and can +/// never produce a duration `mul_f64` refuses). +pub fn chaos_timeout_multiplier() -> f64 { + let override_var = std::env::var("AWA_CHAOS_TIMEOUT_MULTIPLIER").ok(); + multiplier_from(override_var.as_deref(), std::env::var_os("CI").is_some()) +} + +/// The multiplier decision, with the environment passed in so it can be +/// tested without mutating process-global state. +fn multiplier_from(override_var: Option<&str>, is_ci: bool) -> f64 { + // `"inf"` and `"nan"` parse successfully as f64, and both reach + // `Duration::mul_f64`, which panics rather than saturating. Treat a + // non-finite override as no override at all. + if let Some(parsed) = override_var + .and_then(|raw| raw.parse::().ok()) + .filter(|parsed| parsed.is_finite()) + { + return parsed.clamp(1.0, MAX_MULTIPLIER); + } + + if is_ci { + 3.0 + } else { + 1.0 + } +} + +/// Grow a wait deadline by the contention multiplier. +pub fn scaled_timeout(timeout: Duration) -> Duration { + timeout.mul_f64(chaos_timeout_multiplier()) +} + +/// Grow a heartbeat-staleness window by the contention multiplier. +/// +/// Distinct from [`scaled_timeout`] only in intent: this one is passed to +/// `ClientBuilder::heartbeat_staleness`, where the cost of being too tight +/// is a spurious rescue of a live attempt rather than a timeout. Paired +/// heartbeat/rescue *intervals* are deliberately left unscaled so the +/// rescue path still runs at chaos cadence. +pub fn scaled_staleness(staleness: Duration) -> Duration { + staleness.mul_f64(chaos_timeout_multiplier()) +} + +/// Shrink a minimum-progress floor by the contention multiplier. +/// +/// Use for "this subsystem must have advanced at least N times" gates. +/// The returned floor is at least 1: the regression these gates exist to +/// catch is a fully stalled subsystem, so zero progress must still fail. +pub fn contention_floor(nominal: i64) -> i64 { + let scaled = (nominal as f64 / chaos_timeout_multiplier()).floor() as i64; + scaled.max(1) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn local_runs_keep_strict_bounds() { + assert_eq!(multiplier_from(None, false), 1.0); + } + + #[test] + fn ci_runs_get_margin() { + assert_eq!(multiplier_from(None, true), 3.0); + } + + #[test] + fn override_can_only_loosen() { + assert_eq!(multiplier_from(Some("0.1"), true), 1.0); + assert_eq!(multiplier_from(Some("-5"), true), 1.0); + assert_eq!(multiplier_from(Some("6"), false), 6.0); + } + + #[test] + fn unparseable_override_falls_back_to_the_environment() { + assert_eq!(multiplier_from(Some("banana"), true), 3.0); + assert_eq!(multiplier_from(Some(""), false), 1.0); + } + + #[test] + fn non_finite_override_falls_back_to_the_environment() { + // These all parse as f64 and would otherwise reach `mul_f64`, which + // panics on a non-finite result instead of saturating. + for raw in ["inf", "-inf", "infinity", "NaN", "nan"] { + assert_eq!(multiplier_from(Some(raw), true), 3.0, "raw={raw}"); + assert_eq!(multiplier_from(Some(raw), false), 1.0, "raw={raw}"); + } + } + + #[test] + fn oversized_override_is_clamped() { + assert_eq!(multiplier_from(Some("1e308"), true), MAX_MULTIPLIER); + assert_eq!(multiplier_from(Some("10000"), true), MAX_MULTIPLIER); + } + + #[test] + fn every_multiplier_produces_a_usable_duration() { + // `Duration::mul_f64` panics rather than saturating, so the clamp is + // what keeps a bad override from failing the nightly it was set to + // rescue. + for raw in [ + "inf", "-inf", "NaN", "1e308", "10000", "0", "-1", "banana", "", + ] { + for is_ci in [true, false] { + let m = multiplier_from(Some(raw), is_ci); + assert!(m.is_finite() && (1.0..=MAX_MULTIPLIER).contains(&m)); + // Would panic if the multiplier were unbounded. + let _ = Duration::from_secs(60).mul_f64(m); + } + } + } + + #[test] + fn scaling_a_timeout_grows_it() { + // 250ms staleness against a 50ms heartbeat interval is a 5x margin + // locally; on CI it becomes 15x, which is what stops a contended + // runner's live worker from being rescued as if it had died. + assert_eq!( + Duration::from_millis(250).mul_f64(multiplier_from(None, true)), + Duration::from_millis(750) + ); + } + + #[test] + fn progress_floor_keeps_margin_under_the_observed_rate() { + // The receipt gate's nominal floor is duration_secs / 4 = 45 for a + // 180s run. #399 saw a healthy run produce 41, so the CI floor has + // to sit well under that. + let ci_floor = (45.0_f64 / multiplier_from(None, true)).floor() as i64; + assert_eq!(ci_floor, 15); + assert!( + ci_floor < 41, + "floor must clear the observed operating point" + ); + } + + #[test] + fn progress_floor_never_reaches_zero() { + // Whatever the multiplier, a fully pinned ring (zero rotations) has + // to keep failing — that is the regression the gate exists for. + assert_eq!(contention_floor(45), contention_floor(45).max(1)); + let absurd = (45.0_f64 / multiplier_from(Some("10000"), true)).floor() as i64; + assert_eq!(absurd.max(1), 1); + assert_eq!(contention_floor(0), 1); + } +} diff --git a/awa/tests/receipt_plane_regression_gate.rs b/awa/tests/receipt_plane_regression_gate.rs index 48fcaa72..e8a9073c 100644 --- a/awa/tests/receipt_plane_regression_gate.rs +++ b/awa/tests/receipt_plane_regression_gate.rs @@ -60,6 +60,9 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use uuid::Uuid; +mod ci_timing; +use ci_timing::contention_floor; + fn database_url() -> String { env::var("DATABASE_URL") .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()) @@ -432,14 +435,23 @@ async fn test_receipt_plane_steady_state_bounds_under_load() { claim_peak_per_partition ); - // Invariants 3 & 4: rings advance during the run. Conservative - // expectation: `(duration_secs / rotate_interval_secs) / 4` so we - // tolerate slow CI runners. With rotate_interval=1s and a 180s - // duration, expected ≥45 rotations. - // rotate_interval is 1s above; expect at least 25% of the wall-clock - // ticks to actually rotate (a slow CI runner that misses some ticks - // shouldn't fire the gate, but a pinned ring will). - let expected_min_rotations = (duration_secs as i64) / 4; + // Invariants 3 & 4: rings advance during the run. The regression this + // catches is a *pinned* ring — one that stops rotating entirely — so + // the floor has to sit well below the observed operating point, not + // just below the nominal tick rate. + // + // The old floor was `duration_secs / 4` (45 rotations in 180s), + // derived from "25% of the 1s rotate_interval's wall-clock ticks". + // That model was wrong: rotation is driven by the maintenance loop + // reaching a rotate decision, not by the interval alone, so the + // healthy steady state is ~40-45 in 180s — flush against the floor. + // It duly fired on 2026-07-07 at 41 rotations with every + // architectural bound the gate exists for perfect (#399). + // + // Divide by the contention multiplier instead, so CI keeps a genuine + // pinned-ring floor (15 in 180s, ~3x margin under the observed rate) + // while a local run holds the strict value. + let expected_min_rotations = contention_floor((duration_secs as i64) / 4); assert!( queue_rotations >= expected_min_rotations, "queue ring rotated {} times in {}s; expected at least {} \ diff --git a/awa/tests/scheduling_benchmark_test.rs b/awa/tests/scheduling_benchmark_test.rs index 39bd6a4a..f409668d 100644 --- a/awa/tests/scheduling_benchmark_test.rs +++ b/awa/tests/scheduling_benchmark_test.rs @@ -20,6 +20,9 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; +mod ci_timing; +use ci_timing::scaled_timeout; + fn database_url() -> String { std::env::var("DATABASE_URL") .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()) @@ -77,6 +80,7 @@ async fn clean_cron_names(pool: &sqlx::PgPool, names: &[String]) { } async fn wait_for_leader(client: &Client, timeout: Duration) { + let timeout = scaled_timeout(timeout); let start = Instant::now(); loop { let health = client.health_check().await; @@ -92,6 +96,7 @@ async fn wait_for_leader(client: &Client, timeout: Duration) { } async fn wait_for_dispatch(client: &Client, timeout: Duration) { + let timeout = scaled_timeout(timeout); let start = Instant::now(); loop { let health = client.health_check().await; @@ -648,7 +653,7 @@ async fn recv_n( n: usize, timeout: Duration, ) -> Vec { - let deadline = Instant::now() + timeout; + let deadline = Instant::now() + scaled_timeout(timeout); let mut out = Vec::with_capacity(n); while out.len() < n { let now = Instant::now(); @@ -812,7 +817,7 @@ async fn test_runtime_completion_gap() { break; } assert!( - completion_start.elapsed() < Duration::from_secs(30), + completion_start.elapsed() < scaled_timeout(Duration::from_secs(30)), "Timed out waiting for completed rows after handlers returned" ); tokio::time::sleep(Duration::from_millis(20)).await; @@ -1510,7 +1515,7 @@ async fn run_scheduled_frontier_benchmark(queue: &str, total_jobs: i64, due_now: break completed; } assert!( - completion_start.elapsed() < Duration::from_secs(30), + completion_start.elapsed() < scaled_timeout(Duration::from_secs(30)), "Timed out waiting for completed rows after due jobs were picked up" ); tokio::time::sleep(Duration::from_millis(20)).await;