diff --git a/tasks/suite-column-fleet.md b/tasks/suite-column-fleet.md index d24138a..4af0be7 100644 --- a/tasks/suite-column-fleet.md +++ b/tasks/suite-column-fleet.md @@ -1,7 +1,7 @@ --- id: suite-column-fleet name: Conformance suite — the fleet column (exact-count fleet sweep outcomes under concurrency) -status: pending +status: completed depends_on: [store-fleet-put, store-fleet-sweep] scope: moderate risk: medium @@ -98,21 +98,21 @@ instances over one shared pool**. ## Acceptance Criteria -- [ ] The fleet column green **against the harness server** +- [x] The fleet column green **against the harness server** (`pglo-poc`, :15432): the fleet-validity rows, the joint-tx rows, the TTL/renewal rows, the exact-count fleet sweep rows (two-instance survival, lease contention, cross-instance arbitration, the race across the delay, the maintenance step, the fleet liveness row), and the kv-only re-routing row — every row ADR-stamped and traceable to the backlog -- [ ] The fleet factory contract holds: one schema, two+ instances +- [x] The fleet factory contract holds: one schema, two+ instances over it (distinct owner ids), teardown drops exactly the minted schemas, panicking-row-safe — structurally exercised -- [ ] Server-less runs skip cleanly (the reachability probe; the +- [x] Server-less runs skip cleanly (the reachability probe; the default-build gates stay green with the column skipped) -- [ ] README amendments recorded (the fleet rows table; current +- [x] README amendments recorded (the fleet rows table; current state) -- [ ] `cargo test` (default) green — the column compiles out; +- [x] `cargo test` (default) green — the column compiles out; `cargo test --features postgres` green server-less (skipping) and green against the server at acceptance; `cargo test --all-features` green (the consolidation factory's @@ -143,8 +143,67 @@ instances over one shared pool**. ## Notes -> To be filled by implementation agent +Landed as `tests/conformance/fleet_column.rs` (cfg `feature = +"postgres"`; its consolidation pieces cfg the `postgres`+`pg-lo` +pair), wired in `tests/conformance/main.rs`, README amended (§Columns +beyond the reference set's fleet rows table, §Column plan, §Current +state — six columns). Three factories: `FleetDualTierFactory` (one +minted schema, two instances over it — each its own owner id and +private local root riding `FleetSharedMedia::asserted`), +`FleetKvOnlyFactory` (the re-routing client shape; implements +`StoreFactory` so the reference kv-only rejection row mirrors +verbatim), `FleetConsolidationFactory` (a dedicated database, kv + pg-lo +schemas, two fleet instances; the pg-lo column's LO-isolation corollary +— LO reaping, `WITH (FORCE)` drop, drop-guard janitors; a `TrackedDb` +constructor was added to pg_lo_column.rs since the janitor field is +private). The oracle reuses the postgres column's skip probe +(`harness_ready`) and observes the durable state over independent +connections (pool/window/pin-row counts, schema/table existence, the +embedder-table instruments, and the manual-expiry instrument for the +abandonment shape). Column tuning: window delay `0`, TEST_TTL = 900 ms +(TTL/3 ≈ 300 ms), waits are the expiry cadence itself or the manual +instrument — no sleep-based race assertions. + +Recorded deviations from the task text, none blocking: + +- The composite predicate's *structural* non-fleet-valid shapes (kv ≠ + postgres, mem anywhere, `local` undeclared) have no fleet + constructor to reach through (the recorded enforcement map) — the + column asserts the constructor-set shape in prose + the runtime + rejects that do exist (zero TTL, single-connection pool, the + same-instance probe's failure posture). A `TrackedDb::new` + constructor was added rather than widening the field's visibility. +- The pg-lo **cross-instance mismatch** leg is unexercisable against + the one-instance harness server (the probe reads + `pg_control_system()` off one data directory); the row tests the + probe's errors-closed failure posture (an unreachable large + instance rejects at the constructor) and records the mismatch leg — + the same posture the engine suite's tests carry. +- The renewal row's "renewed == at_drop" compare reads at_drop *after* + the drop and asserts `>=` — one TTL/3 tick may legitimately land + between the pre-drop read and the drop; the frozen-after-drop + assertion (EXPIRY_HORIZON, `==`) is the never-leaks proof. ## Summary -> To be filled on completion \ No newline at end of file +The fleet column is green: the wave's gate rows run against the +harness server under `--features postgres` (17 rows + 2 structural +factory tests) and under `--all-features` (the consolidation rows +join) — the fleet-validity rows, the joint-tx rows (both-or-neither, +the upsert-renew, second-row composition, the byte-exact +consolidation round-trip), the TTL/renewal rows (cadence, +never-resurrect, paused-holder + recover), the exact-count fleet sweep +rows (two-instance survival with exact counts, lease contention as +report data, cross-instance arbitration until expiry, the race closed +by the row, the maintenance step's exact reap, the fleet liveness verb +with the schema-ownership rule), and the kv-only re-routing row — +every row ADR-stamped and traced. The fleet factory contract holds +structurally (one schema, two instances, distinct owner ids, owned +idempotent teardown, panicking-row-safe); server-less runs skip +cleanly; the default build compiles the column out. Verification: +`cargo test` (default) green; `cargo test --features postgres` green +server-less (skipping) and green against `pglo-poc` (:15432); +`cargo test --all-features` green against the server (125 conformance +tests + the full library suite, 226 tests); `cargo clippy --all-targets +-- -D warnings` and `cargo fmt --check` green on the default, postgres, +and all-features variants. \ No newline at end of file diff --git a/tests/conformance/README.md b/tests/conformance/README.md index 4813b24..e829c45 100644 --- a/tests/conformance/README.md +++ b/tests/conformance/README.md @@ -402,6 +402,64 @@ inside the dedicated database before it goes. | `virgin_store_reads_are_no_ops_on_a_fresh_schema` | store-api.md §Error model (virgin-store semantics); ADR-012 §6.5; ADR-009 §2 (the companion-table bootstrap); ADR-012 §5 (the mem-pglo row asserts the window on the pg-lo host — `gc_window` present-and-empty, `gc_pins` never) | | `the_window_lives_on_the_kv_schema_and_the_pg_lo_schema_stays_dormant` (postgres-pglo only) | the dormancy row: ADR-012 §5 (the one-window-host rule's kv-host arm — the pg-lo window clauses dormant, no `gc_window` in the pg-lo schema; the schedule observed on the kv host); ADR-008 §2; ADR-012 §6.7 | +**The fleet column's own rows** +(tests/conformance/fleet_column.rs, the suite's third SKIP-posture +column — the **fleet column**, the exact-count fleet sweep outcomes +under concurrency, wave 6's gate, landed wave 6): the column runs +under `--features postgres` (fleet mode requires kv = postgres — the +composite fleet-validity predicate's kv clause, ADR-012 §3), rides the +SKIP posture (it reuses the postgres column's reachability probe; +server-less runs skip cleanly; acceptance is green **against the +harness server**), and extends the factory contract to the **fleet +shape**: a fleet factory mints **one schema** and vends **two `Store` +instances over it**, each `fleet: true`, each with its own +store-instance id (the instances share the pool; the pin rows compose +by owner — the fleet's actual definition, ADR-011/ADR-012 §3). +Teardown drops exactly the minted schemas (`DROP SCHEMA … CASCADE`, +the postgres column's registry pattern — idempotent, owned, +panicking-row-safe), removes every vended temp root, and — the +consolidation shape — reaps the dedicated database's LOs before +dropping it (the pg-lo column's LO-isolation corollary). **The test +tuning** (the column declares its tuning, the constructor parameters' +purpose): window delay `0` (execute-on-next-sweep) and a **short test +pin TTL** (the fleet constructors' `pin_ttl` parameter — the +renewal/expiry rows need observable cadences; the TTL is constructor +configuration, never API). Where a row needs ordering, the window +delay and the TTL tunings provide it, or the independent connection's +**manual-expiry instrument** (the abandonment shape the maintenance +step reaps — the fleet engine suite's recorded shape); no sleep-based +race assertions — the lease/arbitration outcomes are decided by the +server, not by timing. + +Three factories: `FleetDualTierFactory` (kv = postgres + large = +local, each instance with its own private root riding the operator's +`FleetSharedMedia` declaration), `FleetKvOnlyFactory` (the re-routing +posture's client shape, ADR-008 §3 — the mirrored kv-only rejection +row's factory form) and — gated on the `postgres` + `pg-lo` feature +pair — `FleetConsolidationFactory` (kv = postgres + large = pg-lo, the +same pg instance; a dedicated database per open, the LO-isolation +corollary's teardown). + +| Row (`fleet_column.rs`) | Traces to | +|---|---| +| `every_fleet_valid_configuration_constructs` | the fleet-validity rows: ADR-012 §3 (the composite fleet-validity predicate at the constructor — kv-only, dual-tier declared-shared-media, and two-instances-over-one-pool with distinct owner ids, each exercised); ADR-008 §3 (the declared-shared-media arm; the re-routing arm); ADR-011 §4 (the mode vocabulary); INV-7 facade-contract row | +| `the_consolidation_fleet_constructs_over_one_shared_instance` (postgres+pg-lo only) | the fleet-validity rows' consolidation form: ADR-012 §4 (the same-instance arm — the `system_identifier` probe passing by construction); ADR-009 §2 (the companion table); ADR-012 §3 | +| `the_constructor_entry_guards_reject_the_degenerate_tunings` | the constructor's errors-closed rejects: ADR-008 §2 (a zero pin TTL degenerates the TTL/3 cadence — validated at entry); ADR-013 §2 (the fleet pool-sizing guard: a single-connection pool would self-deadlock the sweep lease); ADR-008 §3 (the sealed `FleetSharedMedia` — `local` undeclared has no constructor to reach it through; the structural clauses carry no fleet constructor) | +| `the_consolidation_same_instance_probe_errors_closed` (postgres+pg-lo only) | the same-instance probe's failure posture: ADR-012 §4 (any probe failure is a constructor error; the mismatching-instances leg needs two servers — recorded unexercisable on the one-instance harness) | +| `the_joint_entry_and_pin_tx_commits_both_or_neither` | the joint-tx rows: store-api.md §Invariants 4 (a visible pool entry is never unprotected — the joint commit is the mechanism) and §Invariants 3 (zero residue on failed puts); ADR-008 §2 (the joint entry+pin tx, pin-before-publish); ADR-010 §2 (the store core holds the tx end to end); ADR-012 §3/§4 | +| `a_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row` | store-api.md §Invariants 2 (the fleet's cross-instance dedup leg); ADR-003 (CAS first-writer-wins); ADR-005 (the re-put refcounts); ADR-010 §2 (protection composes by row); ADR-008 §2 (the row's `(key, owner)` identity, the upsert-renew) | +| `the_pg_lo_consolidation_round_trip_is_byte_exact_over_the_shared_instance` (postgres+pg-lo only) | the joint-tx rows' consolidation form (one tx over the shared instance, byte-exact large-entry form, read back from the second instance): store-api.md §Invariants 1 and §Invariants 4; ADR-009 §2; ADR-008 §2; ADR-012 §4 | +| `the_pin_guards_ttl_third_renewal_moves_the_rows_expiry` | the TTL/renewal rows: ADR-008 §2 (the guard renews its row every TTL/3; a dropped guard stops its duty — no leaked tasks, the row freezes); ADR-012 §2 (the TTL lives with the store's renewal schedule — this column's declared tuning) | +| `an_expired_row_is_never_renewed_back_to_life` | the never-resurrect semantics through the facade: ADR-008 §2 (an expired row is never revived, even under a live guard whose renewal is still running); ADR-012 §2 | +| `a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers` | the delete-then-recover backstop: ADR-008 §2 (the expiry semantics verbatim — a paused holder's row abandons at expiry, the delete proceeds, a byte-identical re-put recovers); ADR-012 §2 | +| `two_instance_exact_counts_instance_a_pins_instance_b_sweeps` | INV-4's engine-gated fleet form (the wave's gate): store-api.md §Invariants 4 (DB-backed pins + the advisory-locked sweep, the exact-count sweep-outcome shape against dockerized postgres) and §Invariants 5 (list correctness through outcomes); ADR-008 §2 (the SQL arbitration composes behind the in-process domain — a's foreign row cancels b's candidate at the execution-time re-arbitration); ADR-012 §1 (the exact report fields); ADR-012 §6.7 (the executor is the next sweep); ADR-013 §2; ADR-007 §5 | +| `two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data` | INV-4 (the fleet-form `SweeperLock` shape): ADR-013 §2 (the SQL lease — one sweeper per pool across instances; the second aborts as report data with nothing staged, executed, or deleted); ADR-012 §1 (abort is report data); ADR-012 §6.7 (the holder stages only — the executor is the next sweep) | +| `a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry` | INV-4 (the cross-instance arbitration row): ADR-008 §2 (the SQL-level arbitration on the direct delete path folds un-expired pin rows of *any* instance into one answer); ADR-012 §2 (the expiry semantics — the refusal boundary is exactly the row's expiry); ADR-005 (the facade's `GcRefuse`, nothing deleted) | +| `a_put_landing_during_the_fleet_window_survives_the_executing_sweep` | INV-4 (the sweep-vs-put race's fleet form, closed by the row): ADR-008 §2 (pin-before-publish; the window delay separates staging from execution); ADR-012 §6.7 (the execution-time re-arbitration is the executor sweep's posture); ADR-012 §2 | +| `the_maintenance_step_reaps_exactly_the_expired_rows` | INV-4 (the maintenance-step row): ADR-008 §2 (the sweeper's pin-reap — the GC-of-GC cleanup; expired rows exactly, un-expired rows survive; the reap runs before the overdue execution); ADR-012 §6.7; ADR-012 §1 | +| `an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned` | the fleet liveness row: ADR-012 §6.6 (the fleet liveness verb — the same verb with a fleet-shaped argument; idempotent re-registration; malformed references error); ADR-008 §2 (the schema-ownership rule — the table is the embedder's, consulted at arbitration time, never enumerated); ADR-012 §6.4 (the deregistration correction flow) | +| `fleet_kv_only_mode_rejects_over_threshold_puts_at_put_time` (runner) | the kv-only re-routing row: the reference set's row mirrored verbatim against the fleet kv-only factory (threshold-agnostic by discovery, unchanged) — ADR-012 §6.2's machinery, the fleet edition (ADR-008 §3's re-routing posture's client shape; store-api.md §Invariants 7: the mode is constructible, the row runs) | + ## Column plan **mem is the contract-reference engine** (ADR-011 §2): wave 2 lands @@ -415,12 +473,14 @@ harness server) and the **pg-lo column is the second** (landed wave 5; its two factories gate on the `mem`+`pg-lo` and `postgres`+`pg-lo` feature pairs respectively — the first column whose factories need different feature sets — with the column's full run `--all-features` -against the harness server; the fleet-form rows of INV-4 are -engine-gated to the postgres column and first run in wave 6). +against the harness server). The **fleet column is the third** (wave +6's gate): the engine-gated fleet-form rows of INV-4 run under +`--features postgres` against two store instances over one shared +pool, their consolidation shape gated on the `postgres`+`pg-lo` pair. ## Current state -The harness is green on **five columns** (waves 2–5). The **mem +The harness is green on **six columns** (waves 2–6). The **mem column** is the contract-reference column: the factory binding is the concrete `alkblobs::Store` facade (see §The factory contract — the binding demonstrator remains in `binding.rs`), the reference row set @@ -542,3 +602,57 @@ structural (the engine suite's `shared_harness_db_holds_zero_lo_residue` gate measures it) and the structural teardown assertions probe zero orphan residue inside the dedicated database before it goes. + +The **fleet column** (the suite's **third SKIP-posture column** — +the fleet column, the exact-count fleet sweep outcomes under +concurrency, wave 6's gate, feature `postgres` default-off, its +consolidation factory also gated on the `postgres`+`pg-lo` feature +pair, landed wave 6) grows the **fleet column** of the suite +(README §Columns beyond the reference set — the recorded inventory is +the fleet rows table there): the fleet-validity rows (the composite +fleet-validity predicate at the constructor — every fleet-valid +configuration constructs, the constructor's entry guards reject the +degenerate tunings, and the consolidation shapes gate on the feature +pair), the **joint entry+pin tx rows** (the independent connection +sees both or neither; the dedup re-put upsert-renews the row; a second +instance's re-put composes a second row — protection by row; the pg-lo +consolidation round-trip byte-exact over the shared instance), the +**TTL/renewal rows** (the guard's TTL/3 renewal moves the row's expiry +— the short test TTL makes the cadence observable; the never-resurrect +semantics through the facade; a paused holder loses protection exactly +at expiry with the delete-then-recover backstop), and the **exact-count +fleet sweep rows** — the wave's gate, INV-4's engine-gated fleet form: +the two-instance exact-count shape (instance A pins, instance B +sweeps — the pinned entry survives via a's row at the execution-time +SQL re-arbitration, the unpinned twins delete, counts exact against +the staged schedule), the lease contention row (two instances sweep +concurrently — exactly one sweeps, the other aborts `SweeperLock` as +report data with nothing deleted), the cross-instance arbitration row +(a foreign owner's un-expired pin row refuses a direct delete — +`GcRefuse` — until the row's expiry), the race across the window delay +in its fleet form, the maintenance-step row (expired pin rows reaped +exactly), and the fleet liveness row (a test-owned embedder table +protects through the registration verb — the schema-ownership rule +asserted — with the deregistration correction flow). The kv-only +re-routing row mirrors the reference set's rejection row verbatim +against the fleet kv-only factory (the re-routing posture's client +shape, ADR-008 §3). Its factories follow the postgres column's +registry pattern extended to the fleet shape (one minted schema, two +instances over it, each with its own store-instance id — the pin rows +compose by owner; the dual-tier instances carry private local roots +riding the operator's `FleetSharedMedia` declaration; the +consolidation factory follows the pg-lo column's LO-isolation +corollary: a dedicated database per open, LO reaping, `WITH (FORCE)` +drop, drop-guard janitors). Connection settings ride the environment +(`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`) and the runners gate behind +the reachability probe (the postgres column's — a skip is never a +failure), and its acceptance is the green run **against the harness +server** (dockerized `pglo-poc`, :15432 — verified live 2026-10-12) +under `--features postgres` and the consolidation rows under +`--all-features`. The test tuning: window delay `0` +(execute-on-next-sweep) and a short test pin TTL (the fleet +constructors' `pin_ttl` parameter — the column declares its tuning); +where a row needs ordering, the window delay and the TTL tunings +provide it, or the independent connection's manual-expiry instrument — +no sleep-based race assertions, the lease/arbitration outcomes being +decided by the server, not by timing. diff --git a/tests/conformance/fleet_column.rs b/tests/conformance/fleet_column.rs new file mode 100644 index 0000000..718e884 --- /dev/null +++ b/tests/conformance/fleet_column.rs @@ -0,0 +1,2428 @@ +//! The **fleet column** — the suite's third SKIP-posture column: the +//! exact-count fleet sweep outcomes under concurrency, run against +//! live postgres. This is the wave-6 gate (docs/plans/implementation.md +//! §Wave 6): INV-4's **engine-gated fleet form** (store-api.md +//! §Invariants 4; ADR-008 §2; ADR-012 §1/§4; ADR-013 §2) — the +//! architecture's most intricate state machine (the fleet GC protocol: +//! the joint entry+pin tx, the pin rows' TTL/renewal/expiry lives, the +//! sweep lease, the SQL-level arbitration, the maintenance step, the +//! fleet liveness verb) held under two store instances over one shared +//! pool. +//! +//! The column runs under `--features postgres` — fleet mode requires +//! kv = postgres, the composite fleet-validity predicate's kv clause +//! (ADR-012 §3) — and rides the SKIP posture: every runner gates behind +//! the reachability probe (`crate::postgres_column::harness_ready`, +//! the postgres column's probe — the column reuses the posture's +//! instrument), server-less runs skip cleanly, and acceptance is green +//! **against the harness server** (dockerized `pglo-poc`, +//! `postgres:16-alpine`, :15432, db `blobs`). +//! +//! **The fleet factory** (the postgres column's schema-per-open +//! pattern adapted to the fleet shape; README §The factory contract): +//! `open_fleet()` mints **one schema** and vends **two `Store` +//! instances over it**, each `fleet: true` (the `_fleet` constructors, +//! each carrying its own store-instance id — the instances share the +//! pool; the pin rows compose by owner, `PinRow`'s `(key, owner)` row +//! identity, ADR-011/ADR-012 §3's fleet definition). The dual-tier +//! fleet factory gives each instance its own private local root (the +//! large tier rides the operator's [`FleetSharedMedia`::asserted] +//! declaration — the factory owns the isolated-medium form of it); the +//! kv-only fleet factory is the **re-routing posture's client shape** +//! (ADR-008 §3) and the consolidation fleet factory (kv = postgres + +//! large = pg-lo, the same pg instance) is gated on the `postgres` + +//! `pg-lo` feature pair. Teardown drops exactly the minted schemas +//! (`DROP SCHEMA … CASCADE`, the postgres column's registry pattern — +//! idempotent, owned, panicking-row-safe), removes every vended temp +//! root, and — the consolidation shape — reaps the dedicated +//! database's LOs before dropping it (`WITH (FORCE)`, the pg-lo +//! column's LO-isolation corollary). +//! +//! **The test tuning** (the column declares its tuning, the constructor +//! parameters' purpose): window delay `Duration::ZERO` +//! (execute-on-next-sweep — the durable schedule's degenerate legal +//! tuning) and a **short test pin TTL** ([`TEST_TTL`], the fleet +//! constructors' `pin_ttl` parameter — the renewal/expiry rows need +//! observable cadences; the TTL is constructor configuration, never +//! API, ADR-008 §2). Where a row needs ordering the window delay and +//! the TTL tunings provide it; the waits the rows take are the expiry +//! mechanism's own cadence (deterministic by construction — a dropped +//! holder's row expires at its minted TTL) or the independent +//! connection's manual-expiry instrument (the abandonment the +//! maintenance step reaps — the same instrument the fleet engine +//! suite's tests carry). No sleep-based race assertions: the +//! lease/arbitration outcomes are decided by the server, not by timing. +//! +//! The concurrency rows are the acceptance here (the wave's gate): +//! the **two-instance exact-count shape** (instance A pins, instance B +//! sweeps — A's row cancels B's candidate at the execution-time SQL +//! re-arbitration), the **lease contention row** (two instances sweep +//! concurrently — exactly one sweeps, the other aborts `SweeperLock` +//! as report data with nothing deleted), the **cross-instance +//! arbitration row** (a foreign owner's un-expired pin row refuses a +//! direct delete — `GcRefuse` — and stops after expiry), the **race +//! across the window delay in its fleet form** (a put landing during +//! the delay — its row committed before its entry is visible — +//! survives the next sweep's execution-time re-arbitration), the +//! **maintenance-step row** (expired pin rows reaped exactly; un- +//! expired rows survive), and the **fleet liveness row** (a test-owned +//! embedder table's rows protect through the registration verb — the +//! schema-ownership rule: the engine never created the table). The +//! joint-tx rows, the TTL/renewal rows (through the public `put` and +//! the pin guard), the fleet-validity rows (the predicate at the +//! constructor), and the kv-only re-routing row (the reference set's +//! rejection row mirrored verbatim against the fleet kv-only factory) +//! complete the column. + +#![allow(dead_code)] + +use std::io::Cursor; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use tokio_postgres::NoTls; + +use crate::factory::StoreFactory; +use crate::postgres_column::harness_ready; +use crate::rows::{committed_file_count, patterned, LatchProtect, SetSource}; +use alkblobs::{ + Error, FleetSharedMedia, GcAbortCause, Key, LivenessRegistration, PgOpts, Store, StoreMode, +}; + +// ------------------------------------------------------------------ +// The column's declared tuning +// ------------------------------------------------------------------ + +/// The column's declared dual-tier threshold — small, so the rows' +/// contents land sub-threshold (kv tier: the pg schema's pool rows; +/// the pin rows' medium) and the consolidation row's large entry lands +/// over it. +pub(crate) const TUNED_THRESHOLD: u64 = 1024; + +/// The column's declared kv-only rejection threshold — small, so every +/// leg of the mirrored rejection row stays trivially materializable. +pub(crate) const KV_ONLY_THRESHOLD: u64 = 1024; + +/// The **test tuning of the window delay** (ADR-008 §2's legal +/// degenerate tuning): `0` — execute-on-next-sweep. The fleet sweep's +/// staging sweeps the dead set; the *next* sweep is its executor — +/// exactly the schedule the exact-count rows assert against. +pub(crate) const WINDOW_DELAY: Duration = Duration::ZERO; + +/// The **short test pin TTL** — the fleet constructors' `pin_ttl` +/// parameter, the column's declared tuning (ADR-008 §2: fleet +/// constructor parameters, not API). Short enough that the TTL/3 +/// cadence (≈300 ms) and the expiry horizon are observable within a +/// test's runtime; long enough that a held guard's renewing row never +/// lapses under a sweep. +pub(crate) const TEST_TTL: Duration = Duration::from_millis(900); +/// One renewal cadence plus margin: any interval this long admits at +/// least one TTL/3 tick (TEST_TTL / 3 ≈ 300 ms). +pub(crate) const ONE_CADENCE: Duration = Duration::from_millis(600); +/// Beyond the short TTL: a renewal stopped at its guard's drop leaves +/// the row expired at every read this long after. +pub(crate) const EXPIRY_HORIZON: Duration = Duration::from_millis(1300); + +fn wall_nanos() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as i64) + .unwrap_or(0) +} + +// ------------------------------------------------------------------ +// The factories — one minted schema, two (or more) `Store` instances +// over it, each `fleet: true` +// ------------------------------------------------------------------ + +/// A fresh engine-owned schema name — unique per call, seeded per +/// (tag, pid, nanos, factory-internal sequence) so two factory +/// instances never share a schema, racing opens included (the postgres +/// column's naming policy). +fn fresh_schema_name(tag: &str, seq: &AtomicU64) -> String { + let seq = seq.fetch_add(1, Ordering::Relaxed); + format!( + "alkblobs_fleet_{}_{}_{}_{}", + tag, + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0), + seq, + ) +} + +fn seq_seed() -> u64 { + (std::time::SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0)) + ^ ((std::process::id() as u64) << 8) +} + +/// The factories' engine-opts shape: exactly the minted schema name and +/// the ship defaults for the rest (the alkstore factory's policy — only +/// the schema name is factory policy). +fn opts(schema: String) -> PgOpts { + PgOpts { + schema, + ..PgOpts::default() + } +} + +/// A fresh private temp directory holding one instance's local root. +fn fresh_root(tag: &str, seq: &AtomicU64) -> PathBuf { + let seq = seq.fetch_add(1, Ordering::Relaxed); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let base = std::env::temp_dir().join(format!( + "alkblobs-fleet-column-{tag}-{}-{nanos:x}-{seq}", + std::process::id(), + )); + std::fs::create_dir_all(&base).expect("temp site created"); + base +} + +fn remove_dir(base: &Path) -> Result<(), Error> { + match std::fs::remove_dir_all(base) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(Error::Io(e.to_string())), + } +} + +/// The shared backing site of one open's fleet: the engine-owned +/// schema this open minted (the kv tier's medium, the pin rows', the +/// window's, the sweep lease's, and the advisory lock's — one shared +/// engine over one schema) and the per-instance local roots vended +/// with it. +#[derive(Clone)] +pub(crate) struct FleetSite { + /// The server's DSN — the oracle's independent connections address + /// the shared medium through it. + dsn: String, + /// The minted schema — the fleet's shared pool (the pin rows + /// compose by owner inside it). + schema: String, + /// The per-instance local roots (the dual-tier fleet's large-tier + /// media; empty on the kv-only factory). + roots: Vec, +} + +impl FleetSite { + pub(crate) fn schema(&self) -> &str { + &self.schema + } + + pub(crate) fn dsn(&self) -> &str { + &self.dsn + } + + pub(crate) fn root(&self, instance: usize) -> &Path { + &self.roots[instance] + } +} + +/// The fleet dual-tier factory: `kv = postgres` + `large = local` at +/// the column's tuned threshold, `fleet: true` — one minted schema, +/// two vended instances over it, each with its own store-instance id +/// (the pin rows compose by owner) and its own private local root (the +/// operator's shared-media declaration asserted at each call site). +/// +/// **The fleet factory contract** (README §The factory contract, the +/// postgres column's pattern adapted to the fleet shape): one minted +/// schema per `open_fleet()`, tracked per-instance in the schema +/// registry; every vended instance shares the pool (the fleet's +/// definition — two store instances of one store shape over one shared +/// engine); teardown `DROP SCHEMA IF EXISTS … CASCADE` over exactly +/// the minted schemas plus removal of every vended root (idempotent, +/// owned, panicking-row-safe — each runner constructs its own factory +/// instance and owns its lifecycle, so concurrent rows never touch +/// each other's schemas). +pub(crate) struct FleetDualTierFactory { + dsn: String, + /// The declared dispatch threshold — the routing boundary the + /// rows' contents sit against (the shared matrix lands sub- + /// threshold: kv tier, the pin rows' medium). + threshold: u64, + schema_seq: AtomicU64, + schemas: Mutex>, + sites: Mutex>, +} + +impl FleetDualTierFactory { + pub(crate) fn new(threshold: u64) -> FleetDualTierFactory { + FleetDualTierFactory { + dsn: crate::postgres_column::harness_dsn() + .expect("the harness server is configured for the column"), + threshold, + schema_seq: AtomicU64::new(seq_seed()), + schemas: Mutex::new(Vec::new()), + sites: Mutex::new(Vec::new()), + } + } + + /// The declared dispatch threshold, for the rows that size their + /// contents against it. + pub(crate) fn threshold(&self) -> u64 { + self.threshold + } + + /// Mint one schema, vend two fleet instances over it (each with + /// its own store-instance id and its own private local root), and + /// track the site: the column's rows' instrument. + pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, FleetSite), Error> { + let schema = fresh_schema_name("dual-tier", &self.schema_seq); + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(schema.clone()); + let base_a = fresh_root("dual-tier-a", &self.schema_seq); + let base_b = fresh_root("dual-tier-b", &self.schema_seq); + let root_a = base_a.join("local-root"); + let root_b = base_b.join("local-root"); + let store_a = Store::dual_tier_postgres_local_fleet( + &self.dsn, + opts(schema.clone()), + root_a.clone(), + self.threshold, + WINDOW_DELAY, + TEST_TTL, + FleetSharedMedia::asserted(), + ) + .await?; + let store_b = Store::dual_tier_postgres_local_fleet( + &self.dsn, + opts(schema.clone()), + root_b.clone(), + self.threshold, + WINDOW_DELAY, + TEST_TTL, + FleetSharedMedia::asserted(), + ) + .await?; + self.sites + .lock() + .unwrap_or_else(|e| e.into_inner()) + .extend([base_a, base_b]); + Ok(( + store_a, + store_b, + FleetSite { + dsn: self.dsn.clone(), + schema, + roots: vec![root_a, root_b], + }, + )) + } + + pub(crate) fn tracked_schemas(&self) -> Vec { + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + + /// The DSN the factory opens against — the structural tests' + /// probe target. + pub(crate) fn dsn(&self) -> &str { + &self.dsn + } + + pub(crate) async fn teardown(&self) -> Result<(), Error> { + let schemas = self + .schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect::>(); + let sites = self + .sites + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect::>(); + if !schemas.is_empty() { + let client = admin_connect(&self.dsn).await?; + for schema in &schemas { + let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE"); + client + .batch_execute(&sql) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + } + } + for base in &sites { + remove_dir(base)?; + } + Ok(()) + } +} + +/// The fleet kv-only factory: no large tier — the **re-routing +/// posture's client shape** (ADR-008 §3), `fleet: true`, the column's +/// kv-only threshold. [`StoreFactory`]'s `open()` vends one isolated +/// fleet store over a freshly minted schema (the mirroring rejection +/// row's factory form); [`Self::open_fleet`] vends **two** instances +/// over one minted schema (the fleet contract's structural test's +/// shape). +pub(crate) struct FleetKvOnlyFactory { + dsn: String, + /// The rejection point this column's configuration declares — the + /// mirrored row discovers it behaviorally (threshold-agnostic by + /// discovery) and requires it materializable for the boundary legs. + pub(crate) threshold: u64, + schema_seq: AtomicU64, + schemas: Mutex>, +} + +impl FleetKvOnlyFactory { + pub(crate) fn new(threshold: u64) -> FleetKvOnlyFactory { + FleetKvOnlyFactory { + dsn: crate::postgres_column::harness_dsn() + .expect("the harness server is configured for the column"), + threshold, + schema_seq: AtomicU64::new(seq_seed()), + schemas: Mutex::new(Vec::new()), + } + } + + /// One minted schema, two fleet instances over it — the fleet + /// factory contract's structural shape on this factory too. + pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, FleetSite), Error> { + let schema = fresh_schema_name("kv-only", &self.schema_seq); + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(schema.clone()); + let store_a = Store::kv_only_postgres_fleet( + &self.dsn, + opts(schema.clone()), + self.threshold, + WINDOW_DELAY, + TEST_TTL, + ) + .await?; + let store_b = Store::kv_only_postgres_fleet( + &self.dsn, + opts(schema.clone()), + self.threshold, + WINDOW_DELAY, + TEST_TTL, + ) + .await?; + Ok(( + store_a, + store_b, + FleetSite { + dsn: self.dsn.clone(), + schema, + roots: Vec::new(), + }, + )) + } + + pub(crate) fn tracked_schemas(&self) -> Vec { + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + + pub(crate) fn dsn(&self) -> &str { + &self.dsn + } + + pub(crate) async fn teardown(&self) -> Result<(), Error> { + let schemas = self + .schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect::>(); + if !schemas.is_empty() { + let client = admin_connect(&self.dsn).await?; + for schema in &schemas { + let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE"); + client + .batch_execute(&sql) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + } + } + Ok(()) + } +} + +impl StoreFactory for FleetKvOnlyFactory { + fn name(&self) -> &'static str { + "fleet kv-only postgres" + } + + async fn open(&self) -> Result { + let schema = fresh_schema_name("kv-only", &self.schema_seq); + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(schema.clone()); + Store::kv_only_postgres_fleet( + &self.dsn, + opts(schema), + self.threshold, + WINDOW_DELAY, + TEST_TTL, + ) + .await + } + + async fn teardown(&self) -> Result<(), Error> { + self.teardown_impl().await + } +} + +impl FleetKvOnlyFactory { + async fn teardown_impl(&self) -> Result<(), Error> { + let schemas = self + .schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect::>(); + if !schemas.is_empty() { + let client = admin_connect(&self.dsn).await?; + for schema in &schemas { + let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE"); + client + .batch_execute(&sql) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + } + } + Ok(()) + } +} + +// ------------------------------------------------------------------ +// The medium oracle — the independent connection (outside every +// store's pool, seeing only durable state): the pool's rows, the pin +// rows' `(key, owner, expiry)` triples, the window's staged +// candidates, the schema/table existence probes, the liveness-table +// instruments, and the manual-expiry instrument (the abandonment the +// maintenance step reaps — the fleet engine suite's recorded shape). +// ------------------------------------------------------------------ + +async fn admin_connect(dsn: &str) -> Result { + let (client, connection) = tokio_postgres::connect(dsn, NoTls) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + tokio::spawn(async move { + let _ = connection.await; + }); + Ok(client) +} + +/// One independent connection's probe — the durable-state instrument. +struct Probe { + client: tokio_postgres::Client, + _conn: tokio::task::JoinHandle<()>, +} + +async fn connect_probe(dsn: &str) -> Probe { + let (client, connection) = tokio_postgres::connect(dsn, NoTls) + .await + .expect("the independent connection connects"); + Probe { + client, + _conn: tokio::spawn(async move { + let _ = connection.await; + }), + } +} + +impl Probe { + async fn scalar(&self, sql: &str) -> i64 { + let row = self + .client + .query_one(sql, &[]) + .await + .expect("the oracle query succeeds"); + row.get::<_, i64>(0) + } + + /// The kv tier's pool residency — one row per committed kv-tier + /// entry (the schema's `blobs` table, the shared pool's). + async fn pool_rows(&self, schema: &str) -> usize { + let sql = format!("SELECT COUNT(*) FROM \"{schema}\".blobs"); + self.scalar(&sql).await.max(0) as usize + } + + /// The durable window's staged-candidate residency — the schedule + /// observer (staged rows land here; the executing sweep's take + /// empties them). + async fn window_rows(&self, schema: &str) -> usize { + let sql = format!("SELECT COUNT(*) FROM \"{schema}\".gc_window"); + self.scalar(&sql).await.max(0) as usize + } + + /// The pin rows in their stored encoding, `(key, owner, expiry)` + /// sorted by `(key, owner)` — the pool's durable protection set + /// (cross-instance by construction: every instance's rows compose + /// in this one table). + async fn pin_rows(&self, schema: &str) -> Vec<(Vec, Vec, i64)> { + let sql = + format!("SELECT key, owner, expiry FROM \"{schema}\".gc_pins ORDER BY key, owner"); + let rows = self + .client + .query(sql.as_str(), &[]) + .await + .expect("the oracle query succeeds"); + rows.iter() + .map(|row| { + ( + row.get::<_, Vec>(0), + row.get::<_, Vec>(1), + row.get::<_, i64>(2), + ) + }) + .collect() + } + + /// Whether the engine-owned schema still exists server-side — the + /// structural teardown tests' instrument (`pg_namespace`). + async fn schema_exists(&self, schema: &str) -> bool { + let count: i64 = self + .client + .query_one( + "SELECT count(*) FROM pg_namespace WHERE nspname = $1", + &[&schema], + ) + .await + .expect("the oracle query succeeds") + .get(0); + count > 0 + } + + /// Whether a table exists in a schema — the schema-ownership + /// rule's instrument (the embedder's liveness table is the + /// embedder's: the engine never creates, alters, or drops it). + async fn table_exists(&self, schema: &str, table: &str) -> bool { + let row = self + .client + .query_one( + "SELECT EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = $1 AND tablename = $2)", + &[&schema, &table], + ) + .await + .expect("the oracle query succeeds"); + row.get::<_, bool>(0) + } + + /// Create the embedder's liveness table — **test-owned DDL**, the + /// instrument that keeps the schema-ownership rule honest (the + /// table exists because this probe made it, never because an + /// engine did). + async fn create_liveness_table(&self, schema: &str, table: &str) { + let sql = format!("CREATE TABLE \"{schema}\".\"{table}\" (key bytea PRIMARY KEY)"); + self.client + .execute(sql.as_str(), &[]) + .await + .expect("the embedder DDL succeeds"); + } + + async fn insert_live_row(&self, schema: &str, table: &str, key: &[u8]) { + let sql = format!("INSERT INTO \"{schema}\".\"{table}\" (key) VALUES ($1)"); + self.client + .execute(sql.as_str(), &[&key]) + .await + .expect("the live row inserts"); + } + + /// The manual-expiry instrument: move every pin row for `key` to an + /// already-past expiry — the abandonment shape (a holder whose + /// renewal stopped), immediately durable, no timing reliance. The + /// row stops protecting at once (`arbitrate` requires + /// `expiry > now`); the fleet sweep's maintenance step reaps it. + async fn expire_pin_rows(&self, schema: &str, key: &[u8]) { + let expired = wall_nanos() - 1_000_000_000; + let sql = format!("UPDATE \"{schema}\".gc_pins SET expiry = $1 WHERE key = $2"); + self.client + .execute(sql.as_str(), &[&expired, &key]) + .await + .expect("the manual expiry succeeds"); + } + + /// The one pin row's expiry (by key), when exactly one exists. + async fn pin_expiry(&self, schema: &str, key: &[u8]) -> Option { + let sql = format!("SELECT expiry FROM \"{schema}\".gc_pins WHERE key = $1"); + let rows = self + .client + .query(sql.as_str(), &[&key]) + .await + .expect("the oracle query succeeds"); + rows.into_iter().next().map(|row| row.get::<_, i64>(0)) + } +} + +// ------------------------------------------------------------------ +// The fleet-validity rows — the composite fleet-validity predicate at +// the constructor, its clause-per-row form through the public +// constructors. +// ------------------------------------------------------------------ + +/// The composite fleet-validity predicate at the constructor: every +/// fleet-valid configuration constructs — the kv-only fleet client +/// (large = none, the re-routing posture's arm), the dual-tier fleet +/// (large = local on the declared shared media), and — gated — the +/// consolidation fleet (large = pg-lo, the same pg instance). Each +/// construct carries its mode, the minted schema is the fleet's one +/// shared pool, and a second vended instance constructs over the same +/// pool with its own store-instance id (the two instances' pin rows +/// compose by owner — the fleet's definition, exercised). +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row); +/// ADR-011 §4 (the constructor-mode vocabulary); ADR-012 §3 (the +/// declaration — the composite fleet-validity predicate at the +/// constructor); ADR-008 §3 (the declared-shared-media arm; the +/// re-routing arm); ADR-012 §4 (the same-instance arm, ADR-012 §2's +/// TTL parameter). +pub async fn every_fleet_valid_configuration_constructs( + kv_only: &FleetKvOnlyFactory, + dual_tier: &FleetDualTierFactory, +) { + let (a, b, site) = kv_only + .open_fleet() + .await + .expect("the kv-only fleet constructs (large = none, the re-routing arm)"); + assert_eq!(a.mode(), StoreMode::KvOnly); + assert_eq!(b.mode(), StoreMode::KvOnly); + let content = b"the kv-only fleet client's entry".to_vec(); + let (_, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the kv-only fleet put publishes"); + drop(pin); + let (_, re_pin) = b + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the second instance's fleet re-put publishes"); + drop(re_pin); + let probe = connect_probe(site.dsn()).await; + assert_eq!( + probe.pool_rows(&site.schema).await, + 1, + "the fleet re-put dedups — one pool entry across the two instances" + ); + let rows = probe.pin_rows(&site.schema).await; + assert_eq!(rows.len(), 2, "two instances, two owner rows on one pool"); + assert_ne!( + rows[0].1, rows[1].1, + "the two instances carry distinct store-instance ids — the pin \ + rows compose by owner" + ); + kv_only + .teardown() + .await + .expect("teardown drops exactly the minted schemas"); + assert!( + !probe.schema_exists(&site.schema).await, + "the fleet factory's schema is gone after teardown" + ); + + let (a, b, site) = dual_tier + .open_fleet() + .await + .expect("the dual-tier fleet constructs (large = local on the declared shared media)"); + assert_eq!(a.mode(), StoreMode::DualTier); + assert_eq!(b.mode(), StoreMode::DualTier); + let content = patterned((dual_tier.threshold() - 1) as usize, 191); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the dual-tier fleet put publishes"); + drop(pin); + let (_, re_pin) = b + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the second instance's fleet re-put publishes"); + drop(re_pin); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the second instance sees the first's entry — one fleet, one pool" + ); + let probe = connect_probe(site.dsn()).await; + assert_eq!(probe.pool_rows(&site.schema).await, 1, "one entry row"); + let rows = probe.pin_rows(&site.schema).await; + assert_eq!(rows.len(), 2, "two owner rows"); + assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids"); + dual_tier + .teardown() + .await + .expect("teardown drops exactly the minted schemas"); +} + +/// The consolidation fleet constructs — `kv = postgres` + `large = +/// pg-lo`, **the same pg instance** (the composite fleet-validity +/// predicate's runtime-checkable arm: the `system_identifier` probe, +/// server-side, immune to DSN aliases). Two instances over the one +/// shared instance, each with its own store-instance id; the pin rows +/// live on the kv host, the companion rows on the pg-lo host. +/// +/// Contract stamp: ADR-012 §4 (the consolidation posture — the +/// same-instance clause the constructor probes); ADR-009 §2 (the +/// companion table); ADR-012 §3 (the declaration); ADR-012 §2 (the TTL +/// parameter). +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +pub async fn the_consolidation_fleet_constructs_over_one_shared_instance( + consolidation: &FleetConsolidationFactory, +) { + let (a, b, site) = consolidation + .open_fleet() + .await + .expect("the consolidation fleet constructs over one shared instance"); + assert_eq!(a.mode(), StoreMode::DualTier); + assert_eq!(b.mode(), StoreMode::DualTier); + let content = patterned(2 * TUNED_THRESHOLD as usize + 7, 173); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the consolidation fleet put publishes"); + drop(pin); + let (_, re_pin) = b + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the second instance's fleet re-put publishes"); + drop(re_pin); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the second instance sees the first's entry — one fleet, one \ + shared instance" + ); + let probe = connect_probe(site.dsn()).await; + assert_eq!( + probe.schema_scalar(site.kv_schema(), "blobs").await, + 0, + "the over-threshold entry rides the large tier — the kv pool \ + holds no entry row" + ); + let rows = probe.pin_rows(site.kv_schema()).await; + assert_eq!(rows.len(), 2, "two instances' pin rows on the kv host"); + assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids"); + consolidation + .teardown() + .await + .expect("teardown reaps the LOs, drops the schemas and the database"); +} + +/// The constructor's entry guards — the rejects that error **closed, +/// before any connection attempt**: a `Duration::ZERO` pin TTL +/// (`Duration::ZERO` would degenerate the TTL/3 renewal cadence and +/// expire every pin at once — validated at entry on every `_fleet` +/// constructor) and a single-connection pool (the sweep lease holds +/// its checkout for the sweep's duration and the body's other clauses +/// ride further checkouts under it — the fleet pool-sizing guard, +/// ADR-013 §2). The predicate's structural non-fleet-valid shapes — +/// kv ≠ postgres, mem anywhere, `local` without the shared-media +/// declaration — carry **no fleet constructor to reach through** (the +/// recorded enforcement map: the `_fleet` constructors are named only +/// for the shapes the predicate accepts), so those clauses have no +/// runtime form here beyond the constructors' very existence. +/// +/// Contract stamp: ADR-012 §3 (the declaration at the constructor); +/// ADR-008 §2 (the TTL's expiry semantics — zero degenerates the +/// cadence); ADR-013 §2 (the lease's hold across the sweep body — the +/// pool-sizing guard); ADR-008 §3 (the shared-media declaration — the +/// sealed type the call site names; `local` undeclared has no +/// constructor to reach it through). +pub async fn the_constructor_entry_guards_reject_the_degenerate_tunings() { + let Some(dsn) = crate::postgres_column::harness_dsn() else { + return; + }; + let schema = "no_such_schema_zero_ttl"; + let err = Store::kv_only_postgres_fleet( + &dsn, + opts(schema.to_owned()), + KV_ONLY_THRESHOLD, + WINDOW_DELAY, + Duration::ZERO, + ) + .await + .err() + .expect("a zero pin TTL rejects at the kv-only fleet constructor"); + assert!( + err.to_string().contains("must be positive"), + "the zero-TTL rejection is the constructor's, spelled for the \ + embedder: {err}" + ); + let err = Store::dual_tier_postgres_local_fleet( + &dsn, + opts(schema.to_owned()), + std::env::temp_dir().join("alkblobs-fleet-column-zero-ttl-root"), + TUNED_THRESHOLD, + WINDOW_DELAY, + Duration::ZERO, + FleetSharedMedia::asserted(), + ) + .await + .err() + .expect("a zero pin TTL rejects at the dual-tier fleet constructor"); + assert!( + err.to_string().contains("must be positive"), + "the zero-TTL rejection is the constructor's: {err}" + ); + let err = Store::kv_only_postgres_fleet( + &dsn, + PgOpts { + schema: schema.to_owned(), + max_size: 1, + ..PgOpts::default() + }, + KV_ONLY_THRESHOLD, + WINDOW_DELAY, + TEST_TTL, + ) + .await + .err() + .expect("a single-connection pool rejects at the fleet constructor"); + assert!( + err.to_string().contains("at least 2 connections"), + "the pool-sizing guard is the rejection: {err}" + ); +} + +/// The consolidation fleet's same-instance probe **errors closed** (the +/// runtime-checkable arm's failure posture): pointing the pg-lo large +/// engine at an unreachable instance is a constructor error — any +/// probe failure rejects, never a silent pass. The probe's +/// *mismatching-instances* leg itself needs two postgres instances; +/// the harness server is one instance, so that leg is unexercisable +/// here (recorded — the engine suite's tests carry the same-instance +/// posture; the cross-instance mismatch is the constructor's recorded +/// shape). +/// +/// Contract stamp: ADR-012 §4 (the same-instance clause — the probe +/// reads `pg_control_system()` and any failure, including an +/// unreachable instance, is a constructor error); ADR-012 §3. +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +pub async fn the_consolidation_same_instance_probe_errors_closed(dsn: &str) { + let unreachable = "host=no-such-host.invalid port=1 user=x password=x dbname=x"; + let err = Store::dual_tier_postgres_pglo_fleet( + dsn, + opts("no_such_schema_probe_closed".to_owned()), + unreachable, + opts("no_such_schema_probe_closed".to_owned()), + TUNED_THRESHOLD, + WINDOW_DELAY, + TEST_TTL, + ) + .await + .err() + .expect("the unreachable large instance rejects at the constructor"); + assert!( + matches!(err, Error::Io(_)), + "the probe failure surfaces as the operational constructor error \ + it is (the probe errors closed): {err}" + ); +} + +// ------------------------------------------------------------------ +// The joint entry+pin tx rows — through the public `put`, the fleet +// put path's atomicity visible to an independent connection. +// ------------------------------------------------------------------ + +/// The fleet put path's joint transaction, through the public `put`: +/// the entry row and its pin row commit **together** — an independent +/// connection sees both or neither. The both-halves: instance a's put +/// publishes one pool row and one pin row (the pin row's expiry ahead +/// of `now` — the TTL the store resolved), and the second instance +/// sees the entry at once (one fleet, one pool). The neither-half: a +/// rejecting put (declared-length mismatch, the staged-put contract's +/// failure mode) leaves zero pool rows and zero pin rows. +/// +/// Contract stamp: store-api.md §Invariants 4 (a visible pool entry is +/// never unprotected — the joint commit is the mechanism; an in-flight +/// put is never deleted by a sweep started before it committed); +/// store-api.md §Invariants 3 (zero residue on failed puts); ADR-008 +/// §2 (the joint entry+pin tx, pin-before-publish); ADR-010 §2 (the +/// ownership statement — the store core holds the tx end to end); +/// ADR-012 §3/§4 (the fleet declaration; the put path's wiring). +pub async fn the_joint_entry_and_pin_tx_commits_both_or_neither(factory: &FleetDualTierFactory) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&a); + + let content = patterned((factory.threshold() - 1) as usize, 191); + let canonical = Key::derive(&content, content.len() as u64); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fleet put publishes"); + + assert_eq!(key, canonical, "the canonical derivation addresses the put"); + let probe = connect_probe(site.dsn()).await; + assert_eq!( + ( + probe.pool_rows(&site.schema).await, + probe.pin_rows(&site.schema).await.len() + ), + (1, 1), + "the independent connection sees both — the entry row and its \ + pin row committed together" + ); + assert!( + probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .unwrap() + > wall_nanos(), + "the pin row's expiry is ahead of now — the TTL the store resolved" + ); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the second instance sees the entry — one fleet, one pool" + ); + + // The neither-half: the declared-length mismatch rejects before + // any entry or pin row exists (a failed put leaves nothing). + let err = a + .put( + Some(content.len() as u64 + 400), + Cursor::new(content.as_slice()), + ) + .await + .expect_err("the over-fed mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + assert_eq!( + ( + probe.pool_rows(&site.schema).await, + probe.pin_rows(&site.schema).await.len() + ), + (1, 1), + "the failed put left neither — the pool and the pin set are \ + exactly the successful put's" + ); + + drop(pin); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// The dedup re-put under fleet, through the public `put`: the same +/// instance's re-put commits the pin row alone (the joint tx's row +/// upsert-renewing `(key, owner)` — the wave-4 recorded shape; no new +/// entry row), and a second instance's re-put composes a **second** +/// pin row — protection by row (ADR-010 §2's refcount-in-rows +/// reading), one pool entry throughout. +/// +/// Contract stamp: store-api.md §Invariants 2 (dedup: identical +/// content twice is one pool entry — the fleet's cross-instance leg); +/// ADR-003 (whole-value put is CAS first-writer-wins); ADR-005 (a +/// re-put refcounts the existing entry); ADR-010 §2 (the pin rows +/// compose by owner); ADR-008 §2 (the joint tx's row upsert; the +/// `(key, owner)` row identity). +pub async fn a_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row( + factory: &FleetDualTierFactory, +) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&a); + + let content = patterned((factory.threshold() - 1) as usize, 167); + let canonical = Key::derive(&content, content.len() as u64); + + let (_, pin_one) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the first put publishes"); + let probe = connect_probe(site.dsn()).await; + assert_eq!(probe.pool_rows(&site.schema).await, 1, "one entry row"); + let first_rows = probe.pin_rows(&site.schema).await; + assert_eq!(first_rows.len(), 1, "instance a's one pin row"); + drop(pin_one); + + let (key_again, pin_two) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the dedup re-put publishes"); + assert_eq!(key_again, canonical, "one derivation, one address"); + assert_eq!( + probe.pool_rows(&site.schema).await, + 1, + "the dedup re-put committed the pin row alone — no new entry row" + ); + let renewed_rows = probe.pin_rows(&site.schema).await; + assert_eq!( + renewed_rows.len(), + 1, + "the same instance's row, upsert-renewed" + ); + assert!( + renewed_rows[0].2 > first_rows[0].2, + "the row upsert-renews (key, owner): the expiry moved forward" + ); + drop(pin_two); + + // The second instance's re-put: a different pin row, the same one + // pool entry. + let (key_third, pin_three) = b + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the second instance's re-put publishes"); + assert_eq!(key_third, canonical); + assert_eq!( + probe.pool_rows(&site.schema).await, + 1, + "still one pool entry across instances" + ); + let rows = probe.pin_rows(&site.schema).await; + assert_eq!(rows.len(), 2, "protection composes by row — two owners"); + assert_ne!(rows[0].1, rows[1].1, "two store-instance ids, two rows"); + drop(pin_three); + + factory.teardown().await.expect("teardown is idempotent"); +} + +/// A guard's TTL/3 renewal duty, through the facade: while the guard +/// lives, the pin row's expiry moves forward every TTL/3 (the short +/// test TTL makes the cadence observable over an independent +/// connection); the dropped guard's renewal is stopped with its life — +/// no leaked tasks, no renewal past the guard's life — and the row +/// freezes. +/// +/// Contract stamp: ADR-008 §2 (the pin guard renews its row every +/// TTL/3; a dropped guard stops its duty); ADR-012 §2 (the TTL lives +/// with the store's renewal schedule — the constructor parameter this +/// column tunes). +pub async fn the_pin_guards_ttl_third_renewal_moves_the_rows_expiry( + factory: &FleetDualTierFactory, +) { + let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet"); + + let content = patterned((factory.threshold() - 1) as usize, 149); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fleet put publishes"); + let probe = connect_probe(site.dsn()).await; + let initial = probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the put's pin row"); + tokio::time::sleep(ONE_CADENCE).await; + let renewed = probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the row"); + assert!( + renewed > initial, + "the guard's TTL/3 duty moved the row's expiry forward \ + (initial {initial}, renewed {renewed})" + ); + drop(pin); + let at_drop = probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the row"); + assert!( + at_drop >= renewed, + "the row's expiry moved forward until the drop and never \ + backwards (one TTL/3 tick may land between the read and the \ + drop)" + ); + tokio::time::sleep(EXPIRY_HORIZON).await; + assert_eq!( + at_drop, + probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the row"), + "the dropped guard's renewal is stopped — no leaked tasks, no \ + renewal past the guard's life" + ); + + factory.teardown().await.expect("teardown is idempotent"); +} + +/// The pin row's **never-resurrect semantics, through the facade**: an +/// expired row is never renewed back to life — even under a *live* +/// guard whose renewal duty is still running. The independent +/// connection's manual-expiry instrument creates the abandoned shape +/// (a row past its expiry under a live holder); after at least one +/// TTL/3 tick the row holds exactly the expired value — the guard's +/// renewal observed the expired slot and the never-resurrect predicate +/// (`pin_renew`'s `expiry > now` clause) kept it dead. The row's +/// protection is over; the delete-then-recover backstop's territory. +/// +/// Contract stamp: ADR-008 §2 (the never-resurrect semantics — an +/// expired row is never revived; the expiry semantics verbatim); +/// ADR-012 §2 (the TTL/renewal schedule's expiry reading). +pub async fn an_expired_row_is_never_renewed_back_to_life(factory: &FleetDualTierFactory) { + let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet"); + + let content = patterned((factory.threshold() - 1) as usize, 151); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fleet put publishes"); + + let probe = connect_probe(site.dsn()).await; + probe.expire_pin_rows(&site.schema, &key.as_bytes()).await; + let expired = probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the abandoned row"); + assert!( + expired <= wall_nanos(), + "the instrument left the row expired (a live guard holds it)" + ); + + tokio::time::sleep(ONE_CADENCE).await; + assert_eq!( + expired, + probe + .pin_expiry(&site.schema, &key.as_bytes()) + .await + .expect("the row"), + "the expired row was never renewed back to life — the live \ + guard's renewal never resurrects" + ); + + drop(pin); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// A paused holder (renewal stopped) loses protection exactly at +/// expiry, and the delete-then-recover backstop recovers: the dropped +/// guard's row freezes at its remaining TTL; while un-expired, the +/// (foreign-instance) delete is refused — `GcRefuse`, the row's +/// once-protection — and nothing is deleted; past expiry the delete +/// proceeds; and a byte-identical re-put recovers the entry (one new +/// pin row, the pool's own protection restored). +/// +/// Contract stamp: ADR-008 §2 (the expiry semantics verbatim: a pin +/// whose holder paused past TTL abandons the row; the entry becomes +/// deleteable; recovery is the byte-identical re-put); ADR-012 §2 +/// (the TTL schedule's reading); store-api.md §Invariants 4 (correct +/// liveness: the refuse/proceed boundary is exactly the expiry). +pub async fn a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers( + factory: &FleetDualTierFactory, +) { + let (a, b, _site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&a); + + let content = patterned((factory.threshold() - 1) as usize, 137); + let canonical = Key::derive(&content, content.len() as u64); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fleet put publishes"); + + // Renewal stopped: the guard's drop pauses the holder (the row's + // remaining TTL is its last protection). + drop(pin); + let err = b + .delete(&key.as_bytes()) + .await + .expect_err("the un-expired row still protects"); + assert_eq!( + err, + Error::GcRefuse, + "the row's once-protection holds while un-expired (a foreign \ + instance's read of the pool's pin set)" + ); + assert!(a.has(&key.as_bytes()).await.unwrap(), "nothing was deleted"); + + tokio::time::sleep(EXPIRY_HORIZON).await; + b.delete(&key.as_bytes()) + .await + .expect("past expiry, the paused holder's protection is gone"); + assert!(!a.has(&key.as_bytes()).await.unwrap()); + + // The delete-then-recover backstop: a byte-identical re-put + // recovers the entry. + let (recovered, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the recovering re-put publishes"); + assert_eq!(recovered, canonical, "the same canonical address"); + let (got_len, blob) = a + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry recovered"); + assert_eq!(got_len, content.len() as u64); + assert_eq!( + blob.read_at(0, got_len).await.expect("the window reads"), + content.as_slice(), + "the recovered entry is byte-identical" + ); + drop(pin); + + factory.teardown().await.expect("teardown is idempotent"); +} + +// ------------------------------------------------------------------ +// The exact-count fleet sweep rows — INV-4's engine-gated fleet form, +// the wave's gate. Every row readsits counts against the durable +// schedule (staged this sweep, executed by the next — the window +// delay `0` tuning), through the pool/window/pin-row oracle. +// ------------------------------------------------------------------ + +/// The wave's gate — INV-4's **two-instance exact-count shape**: +/// instance a pins an entry (its pin row — the joint tx's row, its +/// guard held), instance b sweeps. The mark phases stage exactly the +/// dead set (the SQL liveness is never enumerated into the mark set — +/// a's row is not b's in-process state), the executing sweep deletes +/// the unpinned twins exactly and cancels exactly the pinned entry at +/// the execution-time SQL re-arbitration (a's un-expired row, a +/// foreign owner from b's seat), and the pinned entry stays byte- +/// identical. Counts exact: staged 3, executed 2+1, re-staged 1. +/// +/// Contract stamp: store-api.md §Invariants 4 (sweep safety's +/// engine-gated fleet form — DB-backed pins + the advisory-locked +/// sweep, the exact-count sweep-outcome shape against dockerized +/// postgres) and §Invariants 5 (list correctness through outcomes); +/// ADR-008 §2 (the fleet sweep body; the SQL arbitration composes +/// behind the in-process domain); ADR-012 §1 (abort is report data; +/// the exact report fields); ADR-012 §6.7 (the executor is the next +/// sweep); ADR-013 §2 (the sweep lease's single-flight); ADR-007 §5 +/// (postgres, the fleet host — the exact-count gate against +/// dockerized postgres). +pub async fn two_instance_exact_counts_instance_a_pins_instance_b_sweeps( + factory: &FleetDualTierFactory, +) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&b); + + let pinned = patterned((factory.threshold() - 1) as usize, 179); + let (key_pinned, pin) = a + .put(Some(pinned.len() as u64), Cursor::new(pinned.as_slice())) + .await + .expect("instance a's fleet put publishes — its pin row with it"); + + let dead = patterned((factory.threshold() - 3) as usize, 173); + let (key_dead, pin_dead) = b + .put(Some(dead.len() as u64), Cursor::new(dead.as_slice())) + .await + .expect("the twin publishes"); + drop(pin_dead); + let dead_too = patterned((factory.threshold() - 5) as usize, 181); + let (key_dead_too, pin_dead_too) = b + .put( + Some(dead_too.len() as u64), + Cursor::new(dead_too.as_slice()), + ) + .await + .expect("the other twin publishes"); + drop(pin_dead_too); + + // The twins' holders abandoned their rows: the expired shape (the + // manual-expiry instrument — the abandonment the maintenance step + // reaps; no timing reliance). + let probe = connect_probe(site.dsn()).await; + probe + .expire_pin_rows(&site.schema, &key_dead.as_bytes()) + .await; + probe + .expire_pin_rows(&site.schema, &key_dead_too.as_bytes()) + .await; + + assert_eq!( + ( + probe.pool_rows(&site.schema).await, + probe.pin_rows(&site.schema).await.len() + ), + (3, 3), + "three entries; three pin rows on the one shared pool (one per \ + instance's put, a's renewing)" + ); + + let first = b.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + ( + first.candidates_staged, + first.deleted, + first.cancelled_by_arbitration + ), + (3, 0, 0), + "b's mark staged exactly the three dead-by-b's-seat candidates \ + (a's row is not b's enumerable live state) and deleted nothing" + ); + assert_eq!( + probe.window_rows(&site.schema).await, + 3, + "the staged set waits in the durable window" + ); + assert_eq!( + probe.pin_rows(&site.schema).await.len(), + 1, + "the maintenance step reaped exactly the twins' expired rows — \ + a's renewing row survives" + ); + + let second = b.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + second.candidates_staged, 1, + "the survivor re-staged (a's row is still not enumerable live \ + state)" + ); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (2, 1), + "the executing sweep deleted exactly the two unpinned twins and \ + cancelled exactly the pinned entry — a's un-expired row at the \ + SQL re-arbitration" + ); + assert!(!b.has(&key_dead.as_bytes()).await.unwrap()); + assert!(!b.has(&key_dead_too.as_bytes()).await.unwrap()); + assert!(a.has(&key_pinned.as_bytes()).await.unwrap()); + assert!( + b.has(&key_pinned.as_bytes()).await.unwrap(), + "the fleet's shared storage" + ); + let (got_len, blob) = b + .get(&key_pinned.as_bytes()) + .await + .expect("get succeeds") + .expect("the pinned entry survived"); + assert_eq!(got_len, pinned.len() as u64); + assert_eq!( + blob.read_at(0, got_len).await.expect("the window reads"), + pinned.as_slice(), + "the surviving pinned entry is byte-identical" + ); + + drop(pin); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the lease contention row: two instances sweep **concurrently** +/// — exactly one sweeps; the contended second aborts `SweeperLock` as +/// report data with nothing staged, nothing executed, nothing deleted +/// (abort before staging: form-independent), while the holder — whose +/// protect consult latched mid-body — completes by staging only (its +/// staged set waits for the next sweep to execute). The SQL lease is +/// the fleet host's advisory-locked single-flight across instances; +/// the in-process flag composes beside it. +/// +/// Contract stamp: store-api.md §Invariants 4 (aborting sources delete +/// nothing); ADR-013 §2 (the named exclusive sweep lease — one sweeper +/// per pool across instances, atomic test-and-set, holdership ends at +/// release; the fleet pool-sizing guard's semantics); ADR-012 §1 (the +/// contended sweep's abort is report data); ADR-012 §6.7 (the holder +/// stages only — the executor is the next sweep). +pub async fn two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data( + factory: &FleetDualTierFactory, +) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&a); + SetSource::register(&b); + + let dead = patterned((factory.threshold() - 1) as usize, 191); + let (key, pin) = a + .put(Some(dead.len() as u64), Cursor::new(dead.as_slice())) + .await + .expect("the dead witness publishes"); + drop(pin); + let probe = connect_probe(site.dsn()).await; + probe.expire_pin_rows(&site.schema, &key.as_bytes()).await; + + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + a.register_protect_callback(Arc::new(LatchProtect { + entered: entered_tx, + release: parking_lot::Mutex::new(Some(release_rx)), + })); + + let a = Arc::new(a); + let holder = tokio::spawn({ + let a = Arc::clone(&a); + async move { a.sweep().await } + }); + while entered_rx.recv_timeout(Duration::from_secs(5)).is_err() { + tokio::task::yield_now().await; + } + + let contended = b.sweep().await.expect("sweeps are never errors"); + assert_eq!( + contended.aborted, + Some(GcAbortCause::SweeperLock), + "the contended second instance aborts as report data" + ); + assert_eq!( + ( + contended.candidates_staged, + contended.deleted, + contended.cancelled_by_arbitration + ), + (0, 0, 0), + "the contended sweep staged, executed, and deleted nothing" + ); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the contended sweep deleted nothing" + ); + assert_eq!( + probe.window_rows(&site.schema).await, + 0, + "the aborted sweep staged nothing into the window" + ); + release_tx.send(()).ok(); + + let held = holder + .await + .expect("the holder task joined") + .expect("sweeps are never errors"); + assert_eq!(held.aborted, None, "the lease holder completes"); + assert_eq!( + (held.candidates_staged, held.deleted), + (1, 0), + "the holder stages exactly the dead set — it does not execute" + ); + assert_eq!( + probe.window_rows(&site.schema).await, + 1, + "the holder's staged set waits in the durable window" + ); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the holder staged, did not execute" + ); + + let executed = b.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "the released lease's next sweep executes the staged set" + ); + assert!(!b.has(&key.as_bytes()).await.unwrap()); + + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the cross-instance arbitration row: a foreign owner's +/// un-expired pin row refuses instance b's direct delete +/// (`GcRefuse` — nothing deleted); the row's holder stops renewing +/// (the guard's drop — the paused holder), the short test TTL expires +/// the row, and the delete proceeds. The direct path's SQL-level +/// arbitration is the fleet-authoritative half of the two-domain +/// composition: b's own in-process domain has none of a's state, and +/// a's row alone decides the refusal. +/// +/// Contract stamp: store-api.md §Invariants 4 (with correct liveness +/// the delete is refused — the refusal boundary is exactly the row's +/// expiry); ADR-008 §2 (the SQL-level arbitration on the direct +/// delete path folds un-expired pin rows of *any* instance into one +/// answer); ADR-012 §2 (the expiry semantics — `expiry > now` is +/// live); ADR-005 (the facade owns the refusal — `GcRefuse`, +/// nothing deleted). +pub async fn a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry( + factory: &FleetDualTierFactory, +) { + let (a, b, _site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&b); + + let content = patterned((factory.threshold() - 1) as usize, 163); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("instance a pins via its fleet put"); + + let err = b.delete(&key.as_bytes()).await.expect_err( + "the foreign owner's un-expired pin row refuses instance b's \ + delete", + ); + assert_eq!(err, Error::GcRefuse, "the refusal is the facade's, as data"); + assert!(a.has(&key.as_bytes()).await.unwrap(), "nothing was deleted"); + assert!( + b.has(&key.as_bytes()).await.unwrap(), + "the shared storage is untouched" + ); + + drop(pin); + tokio::time::sleep(EXPIRY_HORIZON).await; + b.delete(&key.as_bytes()).await.expect( + "after the row expires (renewal stopped, short TTL), the \ + delete proceeds", + ); + assert!(!a.has(&key.as_bytes()).await.unwrap()); + + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the race across the window delay in its fleet form: an +/// entry staged by the first sweep, a put landing **during the +/// window** — instance b's fleet re-put dedups into the staged entry +/// and its pin row commits before anything further could resolve the +/// candidate — and the executing sweep's SQL re-arbitration cancels +/// the candidate, never the entry. The surviving entry is byte- +/// identical; once the foreign guard's row expires (renewal stopped, +/// short TTL), the next sweep deletes exactly it. +/// +/// Contract stamp: store-api.md §Invariants 4 (an in-flight/just- +/// landed put is never deleted by a sweep started before it +/// committed — the race's fleet form is closed by the row); ADR-008 +/// §2 (pin-before-publish; the window delay separates staging from +/// execution); ADR-012 §6.7 (the execution-time re-arbitration is the +/// executor sweep's posture); ADR-012 §2 (the expiry reading the +/// final delete rides). +pub async fn a_put_landing_during_the_fleet_window_survives_the_executing_sweep( + factory: &FleetDualTierFactory, +) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&a); + + let content = patterned((factory.threshold() - 1) as usize, 191); + let canonical = Key::derive(&content, content.len() as u64); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the twin publishes"); + drop(pin); + let probe = connect_probe(site.dsn()).await; + probe.expire_pin_rows(&site.schema, &key.as_bytes()).await; + + let staged = a.sweep().await.expect("sweeps are never errors"); + assert_eq!(staged.aborted, None); + assert_eq!( + (staged.candidates_staged, staged.deleted), + (1, 0), + "the staging sweep stages exactly the abandoned twin and \ + deletes nothing" + ); + + // The put lands during the window: instance b's fleet re-put + // dedups into the staged entry — its pin row commits (the joint + // tx, row-alone on the dedup leg) before anything resolves the + // candidate. + let (_, foreign_pin) = b + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("instance b's fleet put lands during the window"); + + let executed = a.sweep().await.expect("sweeps are never errors"); + assert_eq!(executed.aborted, None); + assert_eq!( + (executed.deleted, executed.cancelled_by_arbitration), + (0, 1), + "the executing sweep's SQL re-arbitration cancelled the \ + candidate — b's un-expired row" + ); + assert_eq!( + executed.candidates_staged, 1, + "the survivor re-staged (b's row is not a's enumerable live state)" + ); + let (got_len, blob) = a + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the put landing during the window survived"); + assert_eq!(got_len, content.len() as u64); + assert_eq!( + blob.read_at(0, got_len).await.expect("the window reads"), + content.as_slice(), + "the survivor is byte-identical" + ); + + drop(foreign_pin); + tokio::time::sleep(EXPIRY_HORIZON).await; + let reaped = a.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (reaped.deleted, reaped.candidates_staged), + (1, 0), + "after the foreign row expired (renewal stopped, short TTL) \ + and the maintenance step reaped it, the survivor executed \ + exactly once" + ); + assert!( + !a.has(&key.as_bytes()).await.unwrap(), + "the address is the canonical one" + ); + assert_eq!(canonical.as_bytes(), key.as_bytes()); + + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the maintenance-step row: the fleet sweep's pin-reap (the +/// GC-of-GC cleanup) deletes **exactly** the expired pin rows — the +/// abandoned shape (renewal stopped past expiry) is reaped, the +/// un-expired row survives — and the reaped row's candidate executes +/// while the surviving row's candidate is cancelled at the +/// execution-time SQL re-arbitration (the reap runs before the +/// overdue execution: every later arbitration sees the post-reap +/// protection state). +/// +/// Contract stamp: store-api.md §Invariants 4 (exact counts with +/// correct liveness — the maintenance step is the pin-set's sweep); +/// ADR-008 §2 (the sweeper's maintenance step: an expired pin is +/// abandoned and reaped; the GC-of-GC cleanup); ADR-012 §6.7 (the +/// executor is the next sweep — the counts' schedule); +/// ADR-012 §1 (the exact report fields). +pub async fn the_maintenance_step_reaps_exactly_the_expired_rows(factory: &FleetDualTierFactory) { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + SetSource::register(&b); + + let doomed = b"whose pin is abandoned".to_vec(); + let (key_doomed, pin_one) = a + .put(Some(doomed.len() as u64), Cursor::new(doomed.as_slice())) + .await + .expect("the doomed entry pins"); + let survivor = b"whose pin renews".to_vec(); + let (key_survivor, pin_two) = a + .put( + Some(survivor.len() as u64), + Cursor::new(survivor.as_slice()), + ) + .await + .expect("the survivor entry pins"); + + let probe = connect_probe(site.dsn()).await; + probe + .expire_pin_rows(&site.schema, &key_doomed.as_bytes()) + .await; + + let first = b.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + ( + first.candidates_staged, + first.deleted, + first.cancelled_by_arbitration + ), + (2, 0, 0), + "both entries marked (the in-process live set is b's, empty); \ + nothing overdue yet" + ); + let rows = probe.pin_rows(&site.schema).await; + assert_eq!( + rows.len(), + 1, + "the maintenance step reaped exactly the expired row — one \ + row survives" + ); + assert_eq!( + rows[0].0, + key_survivor.as_bytes(), + "the un-expired row survives" + ); + + let second = b.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (1, 1), + "the reaped row's candidate executed; the survivor's cancelled \ + at the SQL re-arbitration" + ); + assert_eq!(second.candidates_staged, 1, "the survivor re-staged"); + assert!(a.has(&key_survivor.as_bytes()).await.unwrap()); + assert!(!a.has(&key_doomed.as_bytes()).await.unwrap()); + + drop(pin_one); + drop(pin_two); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// The **fleet liveness row**: a test-owned embedder table's rows +/// protect fleet-wide through the fleet registration verb — the SQL +/// arbitration folds the table into the answer on the sweep's +/// execution-time re-arbitration and on the direct-delete path alike — +/// and the schema-ownership rule holds: the table exists because the +/// test made it (the engine never creates, alters, or drops it), the +/// table is never enumerated into the mark set (consulted at +/// arbitration time), a malformed reference errors at registration, +/// re-registration is idempotent, and the deregistration composes +/// (the correction flow: the token's removal lets the delete proceed). +/// +/// Contract stamp: store-api.md §Invariants 4 (correct liveness: a +/// registered fleet liveness table's rows protect); ADR-012 §6.6 (the +/// fleet liveness verb — the same verb with a fleet-shaped argument; +/// registration records the reference); ADR-008 §2 (the SQL protocol +/// sees tables, never callbacks — the schema-ownership rule: the +/// store reads, never owns). +pub async fn an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned( + factory: &FleetDualTierFactory, +) { + let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet"); + + let probe = connect_probe(site.dsn()).await; + probe + .create_liveness_table(&site.schema, "embedder_live") + .await; + let reference = format!("{}.embedder_live", site.schema); + + let registration = a + .register_liveness_source(LivenessRegistration::FleetTable(reference.clone())) + .expect("the embedder table reference registers"); + let re_registered = a + .register_liveness_source(LivenessRegistration::FleetTable(reference.clone())) + .expect("re-registration is idempotent"); + assert_eq!( + registration, re_registered, + "re-registering the same reference returns the original token" + ); + let malformed = a + .register_liveness_source(LivenessRegistration::FleetTable( + "bad..reference".to_owned(), + )) + .expect_err("a malformed reference errors at registration"); + assert!( + malformed + .to_string() + .contains("invalid liveness table reference"), + "the malformed reference surfaces the engine error: {malformed}" + ); + assert!( + probe.table_exists(&site.schema, "embedder_live").await + && !probe.table_exists(&site.schema, "no_such_table").await, + "the schema-ownership rule: the embedder's table is the \ + embedder's, and no engine ever created one at registration \ + time" + ); + + let protected = b"live in the embedder's table".to_vec(); + let (key_protected, pin_one) = a + .put( + Some(protected.len() as u64), + Cursor::new(protected.as_slice()), + ) + .await + .expect("the protected entry pins"); + drop(pin_one); + let dead = b"dead beside the table's row".to_vec(); + let (key_dead, pin_two) = a + .put(Some(dead.len() as u64), Cursor::new(dead.as_slice())) + .await + .expect("the dead entry pins"); + drop(pin_two); + probe + .expire_pin_rows(&site.schema, &key_protected.as_bytes()) + .await; + probe + .expire_pin_rows(&site.schema, &key_dead.as_bytes()) + .await; + probe + .insert_live_row(&site.schema, "embedder_live", &key_protected.as_bytes()) + .await; + + let first = a.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + (first.candidates_staged, first.deleted), + (2, 0), + "both marked — the table is never enumerated into the mark set" + ); + let second = a.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (1, 1), + "the dead entry executed; the table-protected entry cancelled \ + at the execution-time SQL re-arbitration" + ); + assert!(a.has(&key_protected.as_bytes()).await.unwrap()); + assert!(!a.has(&key_dead.as_bytes()).await.unwrap()); + + assert_eq!( + a.delete(&key_protected.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "the liveness table protects the direct delete path too" + ); + + assert!( + a.unregister_liveness_source(registration), + "the fleet token deregisters once" + ); + assert!( + !a.unregister_liveness_source(re_registered), + "the idempotent re-registration consumed one token only" + ); + a.delete(&key_protected.as_bytes()) + .await + .expect("the correction flow: deregistered, the delete proceeds"); + + factory.teardown().await.expect("teardown is idempotent"); +} + +// ------------------------------------------------------------------ +// The structural factory tests — the fleet factory contract's own +// wiring-level assertions (not conformance rows: no backlog entry). +// ------------------------------------------------------------------ + +/// The fleet factory contract, structurally: one minted schema, two +/// instances over it, distinct store-instance ids observable in the +/// pin rows' owner bytes, distinct vended roots, teardown dropping +/// exactly the minted schema and removing every vend, idempotent. +/// Run on both factories: the dual-tier fleet (kv = postgres + large +/// = local) and the kv-only fleet (the re-routing client shape). +#[tokio::test] +async fn fleet_factory_one_schema_vends_instances_with_distinct_owner_ids() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + { + let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet"); + assert_eq!( + factory.tracked_schemas().len(), + 1, + "one minted schema per open" + ); + assert_ne!( + site.root(0), + site.root(1), + "two instances, two private roots" + ); + let one = b"instance a's entry".to_vec(); + let two = b"instance b's entry".to_vec(); + let (_, pin_a) = a + .put(Some(one.len() as u64), Cursor::new(one.as_slice())) + .await + .expect("a's put publishes"); + let (_, pin_b) = b + .put(Some(two.len() as u64), Cursor::new(two.as_slice())) + .await + .expect("b's put publishes"); + drop(pin_a); + drop(pin_b); + let probe = connect_probe(site.dsn()).await; + let rows = probe.pin_rows(&site.schema).await; + assert_eq!(rows.len(), 2, "two instances, two pin rows"); + assert_ne!( + rows[0].1, rows[1].1, + "two store-instance ids — the pin rows compose by owner" + ); + assert_eq!( + committed_file_count(site.root(0)), + 0, + "the kv-tier entries left no large-tier files" + ); + } + factory.teardown().await.expect("teardown drops the schema"); + factory.teardown().await.expect("teardown repeats cleanly"); + let probe = connect_probe(factory.dsn()).await; + for schema in factory.tracked_schemas() { + assert!( + !probe.schema_exists(&schema).await, + "teardown dropped exactly the minted schema: {schema}" + ); + } + + let kv_only = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD); + { + let (a, b, site) = kv_only + .open_fleet() + .await + .expect("open vends the kv-only fleet"); + assert_eq!( + kv_only.tracked_schemas().len(), + 1, + "one minted schema per open" + ); + assert_eq!(a.mode(), StoreMode::KvOnly); + assert_eq!(b.mode(), StoreMode::KvOnly); + let one = b"instance a's kv entry".to_vec(); + let two = b"instance b's kv entry".to_vec(); + let (_, pin_a) = a + .put(Some(one.len() as u64), Cursor::new(one.as_slice())) + .await + .expect("a's put publishes"); + let (_, pin_b) = b + .put(Some(two.len() as u64), Cursor::new(two.as_slice())) + .await + .expect("b's put publishes"); + drop(pin_a); + drop(pin_b); + let probe = connect_probe(site.dsn()).await; + let rows = probe.pin_rows(&site.schema).await; + assert_eq!(rows.len(), 2, "two instances, two pin rows"); + assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids"); + } + kv_only.teardown().await.expect("teardown drops the schema"); + let probe = connect_probe(kv_only.dsn()).await; + for schema in kv_only.tracked_schemas() { + assert!( + !probe.schema_exists(&schema).await, + "teardown dropped the kv-only fleet's schema: {schema}" + ); + } +} + +/// A failing assertion panics the row before teardown runs — and the +/// fleet factory's teardown must make sense over the abandoned fleet +/// anyway: the minted schema is really gone server-side (`DROP SCHEMA +/// … CASCADE` over exactly the registry's list — CASCADE removes the +/// dependents while the abandoned instances' pooled connections fail +/// closed on next use), and the next open mints a fresh schema on a +/// fresh (empty) pool (README §The factory contract). +#[tokio::test] +async fn fleet_factory_teardown_is_safe_against_a_panicking_row() { + if !harness_ready().await { + return; + } + let content = patterned(TUNED_THRESHOLD as usize - 7, 173); + let abandoned = Key::derive(&content, content.len() as u64); + let factory = Arc::new(FleetDualTierFactory::new(TUNED_THRESHOLD)); + + let task = { + let factory = Arc::clone(&factory); + tokio::spawn(async move { + let (store, _, _) = factory.open_fleet().await.expect("open vends the fleet"); + let (_, pin) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the put publishes"); + drop(pin); + panic!("a failing assertion panics the row (the factory contract's early exit)"); + }) + }; + assert!( + task.await.is_err(), + "the row's panic surfaced as its failure" + ); + + let schemas = factory.tracked_schemas(); + assert!(!schemas.is_empty(), "the panicked row minted a schema"); + factory + .teardown() + .await + .expect("teardown makes sense over the abandoned fleet"); + factory.teardown().await.expect("teardown is idempotent"); + let probe = connect_probe(factory.dsn()).await; + for schema in &schemas { + assert!( + !probe.schema_exists(schema).await, + "teardown dropped the panicked row's schema: {schema}" + ); + } + + let (store, _, site) = factory + .open_fleet() + .await + .expect("the next open still vends a fleet"); + assert!( + !store.has(&abandoned.as_bytes()).await.unwrap(), + "no residue from the panicked run — isolation is factory-owned" + ); + assert_eq!( + probe.pool_rows(&site.schema).await, + 0, + "the fresh pool is empty" + ); + factory + .teardown() + .await + .expect("teardown drops the last schema"); +} + +// ------------------------------------------------------------------ +// The consolidation fleet factory — kv = postgres + large = pg-lo on +// one shared pg instance (the composite fleet-validity predicate's +// same-instance arm), gated on the `postgres` + `pg-lo` feature pair. +// ------------------------------------------------------------------ +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +mod consolidation { + use super::*; + use crate::pg_lo_column::{ + admin_dsn, create_db, db_dsn, drop_db, fresh_db_name, fresh_schema_name, reap_lo_schema, + TrackedDb, + }; + + /// One open's tracked consolidation site: the dedicated database + /// (LOs are database-level — the C7 lesson), the kv schema (the + /// fleet host: pin rows, window, lease) and the pg-lo schema (the + /// companion table) minted inside it. + #[derive(Clone)] + pub(crate) struct ConsolidationSite { + dsn: String, + db: String, + kv_schema: String, + lo_schema: String, + } + + impl ConsolidationSite { + pub(crate) fn dsn(&self) -> &str { + &self.dsn + } + + pub(crate) fn kv_schema(&self) -> &str { + &self.kv_schema + } + + pub(crate) fn lo_schema(&self) -> &str { + &self.lo_schema + } + + pub(crate) fn db(&self) -> &str { + &self.db + } + } + + /// The consolidation fleet factory: one dedicated database per + /// `open_fleet()`, two engine schemas minted inside it, two fleet + /// instances (`dual_tier_postgres_pglo_fleet` — the same-instance + /// clause's probe passing by construction: one instance, two + /// schemas), teardown reaping the LOs (via the companion table — + /// the contract authority), dropping exactly the minted schemas, + /// dropping the database (`WITH (FORCE)`), with a drop-guard + /// janitor per open backing the abandoned paths (the pg-lo + /// column's LO-isolation corollary, adapted to the fleet shape). + pub(crate) struct FleetConsolidationFactory { + dsn: String, + seq: AtomicU64, + dbs: Mutex>, + } + + impl FleetConsolidationFactory { + pub(crate) fn new() -> FleetConsolidationFactory { + FleetConsolidationFactory { + dsn: crate::postgres_column::harness_dsn() + .expect("the harness server is configured for the column"), + seq: AtomicU64::new(seq_seed()), + dbs: Mutex::new(Vec::new()), + } + } + + /// One dedicated database, two schemas, two fleet instances + /// over them. + pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, ConsolidationSite), Error> { + let maintenance = admin_dsn(&self.dsn); + let db = fresh_db_name("fleet-consolidation", &self.seq); + create_db(&maintenance, &db).await?; + let dsn = db_dsn(&self.dsn, &db); + let kv_schema = fresh_schema_name("kv", &self.seq); + let lo_schema = fresh_schema_name("lo", &self.seq); + self.dbs + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(TrackedDb::new( + db.clone(), + dsn.clone(), + vec![kv_schema.clone(), lo_schema.clone()], + vec![lo_schema.clone()], + &maintenance, + )); + let store_a = Store::dual_tier_postgres_pglo_fleet( + &dsn, + opts(kv_schema.clone()), + &dsn, + opts(lo_schema.clone()), + TUNED_THRESHOLD, + WINDOW_DELAY, + TEST_TTL, + ) + .await?; + let store_b = Store::dual_tier_postgres_pglo_fleet( + &dsn, + opts(kv_schema.clone()), + &dsn, + opts(lo_schema.clone()), + TUNED_THRESHOLD, + WINDOW_DELAY, + TEST_TTL, + ) + .await?; + Ok(( + store_a, + store_b, + ConsolidationSite { + dsn, + db, + kv_schema, + lo_schema, + }, + )) + } + + pub(crate) async fn teardown(&self) -> Result<(), Error> { + let mut dbs = self + .dbs + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + .collect::>(); + let maintenance = admin_dsn(&self.dsn); + for tracked in dbs.iter_mut() { + if let Ok(client) = admin_connect(&tracked.dsn).await { + for schema in &tracked.lo_schemas { + reap_lo_schema(&client, schema).await; + } + for schema in &tracked.schemas { + let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE"); + let _ = client.batch_execute(&sql).await; + } + } + drop_db(&maintenance, &tracked.name).await?; + TrackedDb::disarm(tracked); + } + Ok(()) + } + } +} + +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +use consolidation::FleetConsolidationFactory; + +impl Probe { + /// The schema-qualified pool table's count via the probe (the + /// consolidation rows' kv-side oracle: the kv schema's `blobs`). + async fn schema_scalar(&self, schema: &str, table: &str) -> usize { + let sql = format!("SELECT COUNT(*) FROM \"{schema}\".\"{table}\""); + self.scalar(&sql).await.max(0) as usize + } +} + +// ------------------------------------------------------------------ +// The consolidation round-trip row — the joint tx over the shared +// instance, byte-exact large-entry form (the feature pair's home). +// ------------------------------------------------------------------ + +/// The pg-lo consolidation round-trip **over the shared instance, two +/// instances**: instance a's fleet put routes over the threshold to +/// the large tier — the joint tx carries the LO writes, the companion +/// row insert, and the pin row in **one transaction over the shared +/// instance** (the consolidation posture's whole point), and instance +/// b reads the entry whole: byte-exact, stat-correct. The mediums' +/// oracle legs: one companion row + its LO on the pg-lo medium, one +/// pin row per put on the kv host, zero pool rows (the large tier +/// holds it). +/// +/// Contract stamp: store-api.md §Invariants 1 (the round-trip's +/// consolidation leg — byte-identical) and §Invariants 4 (the joint +/// tx holds on every fleet-valid configuration); ADR-009 §2 (the +/// companion table — the addressing authority; delete = lo_unlink + +/// row delete); ADR-008 §2 (the joint entry+pin tx); ADR-012 §4 (the +/// consolidation posture: one tx over the shared instance; the +/// same-instance clause); ADR-012 §3 (the declaration). +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +pub async fn the_pg_lo_consolidation_round_trip_is_byte_exact_over_the_shared_instance( + factory: &FleetConsolidationFactory, +) { + let (a, b, site) = factory + .open_fleet() + .await + .expect("open vends the consolidation fleet"); + SetSource::register(&a); + + let content = patterned(2 * 1024 + 7, 167); + let canonical = Key::derive(&content, content.len() as u64); + let (key, pin) = a + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fleet put routes to the consolidation joint tx"); + + assert_eq!(key, canonical, "the canonical derivation addresses the put"); + let probe = connect_probe(site.dsn()).await; + assert_eq!( + probe.schema_scalar(site.kv_schema(), "blobs").await, + 0, + "the over-threshold entry rides the large tier — the kv pool \ + holds none" + ); + assert_eq!( + probe.schema_scalar(site.lo_schema(), "lo_entries").await, + 1, + "the joint tx's companion row is on the LO medium — one row" + ); + assert_eq!( + probe.pin_rows(site.kv_schema()).await.len(), + 1, + "the joint tx's pin row is on the kv host — both halves of the \ + one tx, over the one shared instance" + ); + + let (got_len, blob) = b + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("instance b reads the entry from the shared instance"); + assert_eq!(got_len, content.len() as u64); + assert_eq!( + blob.read_at(0, got_len).await.expect("the windows read"), + content.as_slice(), + "instance b's read is byte-exact — the large-entry form \ + round-trips over the shared instance" + ); + assert_eq!( + b.stat(&key.as_bytes()).await.expect("stat succeeds"), + Some(alkblobs::EntryMeta { + len: content.len() as u64, + algorithm: key.algorithm(), + }), + "stat carries the length plus algorithm on the consolidation \ + shape" + ); + + drop(pin); + factory.teardown().await.expect("teardown is idempotent"); +} + +// ------------------------------------------------------------------ +// The runners — the column's tests. Every one gates behind the +// reachability probe (the SKIP posture, shared with the postgres +// column): server-less runs skip cleanly, a skip is not a failure; +// the acceptance run is green against the harness server. +// ------------------------------------------------------------------ + +fn harness_dsn() -> Option { + crate::postgres_column::harness_dsn() +} + +#[tokio::test] +async fn fleet_validity_every_fleet_valid_configuration_constructs() { + if !harness_ready().await { + return; + } + let kv_only = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD); + let dual_tier = FleetDualTierFactory::new(TUNED_THRESHOLD); + every_fleet_valid_configuration_constructs(&kv_only, &dual_tier).await; +} + +#[tokio::test] +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +async fn fleet_validity_the_consolidation_fleet_constructs() { + if !harness_ready().await { + return; + } + let consolidation = FleetConsolidationFactory::new(); + the_consolidation_fleet_constructs_over_one_shared_instance(&consolidation).await; +} + +#[tokio::test] +async fn fleet_validity_the_constructor_entry_guards_reject_the_degenerate_tunings() { + if !harness_ready().await { + return; + } + the_constructor_entry_guards_reject_the_degenerate_tunings().await; +} + +#[tokio::test] +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +async fn fleet_validity_the_consolidation_same_instance_probe_errors_closed() { + if !harness_ready().await { + return; + } + let Some(dsn) = harness_dsn() else { + return; + }; + the_consolidation_same_instance_probe_errors_closed(&dsn).await; +} + +#[tokio::test] +async fn fleet_put_the_joint_entry_and_pin_tx_commits_both_or_neither() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + the_joint_entry_and_pin_tx_commits_both_or_neither(&factory).await; +} + +#[tokio::test] +async fn fleet_put_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + a_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row(&factory).await; +} + +#[tokio::test] +#[cfg(all(feature = "postgres", feature = "pg-lo"))] +async fn fleet_put_the_consolidation_round_trip_is_byte_exact_over_the_shared_instance() { + if !harness_ready().await { + return; + } + let factory = FleetConsolidationFactory::new(); + the_pg_lo_consolidation_round_trip_is_byte_exact_over_the_shared_instance(&factory).await; +} + +#[tokio::test] +async fn fleet_ttl_the_pin_guards_ttl_third_renewal_moves_the_rows_expiry() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + the_pin_guards_ttl_third_renewal_moves_the_rows_expiry(&factory).await; +} + +#[tokio::test] +async fn fleet_ttl_an_expired_row_is_never_renewed_back_to_life() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + an_expired_row_is_never_renewed_back_to_life(&factory).await; +} + +#[tokio::test] +async fn fleet_ttl_a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers(&factory).await; +} + +#[tokio::test] +async fn fleet_sweep_two_instance_exact_counts_instance_a_pins_instance_b_sweeps() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + two_instance_exact_counts_instance_a_pins_instance_b_sweeps(&factory).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fleet_sweep_two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data(&factory).await; +} + +#[tokio::test] +async fn fleet_sweep_a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry(&factory).await; +} + +#[tokio::test] +async fn fleet_sweep_a_put_landing_during_the_fleet_window_survives_the_executing_sweep() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + a_put_landing_during_the_fleet_window_survives_the_executing_sweep(&factory).await; +} + +#[tokio::test] +async fn fleet_sweep_the_maintenance_step_reaps_exactly_the_expired_rows() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + the_maintenance_step_reaps_exactly_the_expired_rows(&factory).await; +} + +#[tokio::test] +async fn fleet_sweep_an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned() { + if !harness_ready().await { + return; + } + let factory = FleetDualTierFactory::new(TUNED_THRESHOLD); + an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned(&factory).await; +} + +#[tokio::test] +async fn fleet_kv_only_mode_rejects_over_threshold_puts_at_put_time() { + if !harness_ready().await { + return; + } + let factory = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD); + crate::rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&factory).await; +} diff --git a/tests/conformance/main.rs b/tests/conformance/main.rs index 9325dc3..ee423dd 100644 --- a/tests/conformance/main.rs +++ b/tests/conformance/main.rs @@ -38,7 +38,13 @@ //! LO-medium column, feature `pg-lo` default-off, its two factories //! spanning feature pairs `mem`+`pg-lo` and `postgres`+`pg-lo`), which //! runs the same sets against the LO medium and skips cleanly -//! server-less. +//! server-less; [`fleet_column`] — the suite's third SKIP-posture +//! column (the fleet column — the exact-count fleet sweep outcomes +//! under concurrency, wave 6's gate, feature `postgres`-gated, its +//! consolidation factory also gated on `pg-lo`), which runs the +//! fleet-validity rows, the joint-tx rows, the TTL/renewal rows, and +//! the engine-gated fleet-form sweep rows against two store instances +//! over one shared pool and skips cleanly server-less. //! //! Items beyond the tests below are the later columns' surface, hence //! the scoped `#![allow(dead_code)]` — it shrinks as columns land. @@ -49,6 +55,8 @@ mod backlog; #[cfg(feature = "mem")] mod binding; mod factory; +#[cfg(feature = "postgres")] +mod fleet_column; #[cfg(feature = "mem")] mod local_column; #[cfg(feature = "mem")] diff --git a/tests/conformance/pg_lo_column.rs b/tests/conformance/pg_lo_column.rs index 4042223..2d7a760 100644 --- a/tests/conformance/pg_lo_column.rs +++ b/tests/conformance/pg_lo_column.rs @@ -408,6 +408,26 @@ pub(crate) struct TrackedDb { } impl TrackedDb { + /// Register one tracked database — the constructor the columns' + /// factories share (the janitor field is private; the factory + /// records the maintenance DSN and the database's name, and the + /// janitor backs every abandoned exit path). + pub(crate) fn new( + name: String, + dsn: String, + schemas: Vec, + lo_schemas: Vec, + maintenance_dsn: &str, + ) -> TrackedDb { + TrackedDb { + janitor: Some(DbJanitor::new(maintenance_dsn, &name)), + name, + dsn, + schemas, + lo_schemas, + } + } + /// Disarm the janitor (teardown took over the disposal). pub(crate) fn disarm(&mut self) { self.janitor = None;