diff --git a/tasks/suite-column-postgres.md b/tasks/suite-column-postgres.md index b1f0bf5..69be62d 100644 --- a/tasks/suite-column-postgres.md +++ b/tasks/suite-column-postgres.md @@ -1,7 +1,7 @@ --- id: suite-column-postgres name: Conformance suite — the postgres column (the SKIP-posture column) -status: pending +status: completed depends_on: [store-postgres-construction] scope: moderate risk: medium @@ -121,8 +121,78 @@ server**). The column runs under `--features postgres` (feature ## Notes -> To be filled by implementation agent +> Implemented in `tests/conformance/postgres_column.rs` + `main.rs` + +> `README.md` (wave 4, following `suite-column-sqlite` — the +> SQL-backed structural twin — and the alkstore pg-factory pattern). +> Recorded design points, as resolved: +> +> - **Factories as specced** — `DualTierPostgresFactory` (kv = postgres +> + large = local at the column-declared `TUNED_THRESHOLD` = 1024, +> window delay `WINDOW_DELAY` = `Duration::ZERO`, the test tuning) +> and `KvOnlyPostgresFactory` (`KV_ONLY_THRESHOLD` = 1024, the +> rejection row's host). Both hold the DSN, a schema sequence +> counter, and the per-instance schema registry (`Mutex>, +> poisoned-lock via `unwrap_or_else(|e| e.into_inner())` per +> convention); `open()` mints `alkblobs_suite_{tag}_{pid}_{nanos}_{seq}` +> — unique per call, racing opens included — plus a fresh private +> temp root for the large tier (dual-tier; kv-only needs none), +> tracked per-instance. Teardown `DROP SCHEMA IF EXISTS "…" CASCADE` +> over exactly the minted schemas (one admin connection, then the +> drops) plus `remove_dir_all` over the vended roots (NotFound +> tolerated) — idempotent, owned, panicking-row-safe. +> - **SKIP posture as specced** — `harness_dsn()` rides +> `ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` (host required, non-empty; +> port/user/password/db defaulting to the harness server's values — +> :15432, `postgres`/`poc`, db `blobs` — the src/postgres.rs engine +> tests' exact convention, the alkblobs adaptation of the alkstore +> harness convention), `harness_ready()` probes with a 3 s-timed +> admin connection, every runner gates on it first, and the skip +> note prints once per skipped test. The acceptance run below is +> against the harness server — a skip never substituted. +> - **The medium oracle, adapted by medium** — an independent admin +> connection per count (`SELECT COUNT(*) FROM "schema".blobs` / +> `gc_window`) vs the root's committed files; `schema_exists` (the +> `pg_namespace` probe) instruments the structural teardown +> assertions. +> - **Rows** — the sqlite column's inventory mirrored point for point +> (the schedule-relative rows, the dispatch rows with the oracle's +> medium swapped, the virgin-store row read against a fresh schema, +> the structural factory tests adapted to schemas: two opens mint +> distinct schemas and share nothing, teardown-against-a-panicking +> row really drops the schemas); stamps cite the postgres rows of +> ADR-007/012 (ADR-007 §5 replacing the sqlite arm's citations). +> - **One cleanup caught in review-of-diff**: the `dual_tier` open +> tracks the temp *base* dir and constructs the store against +> `base.join("local-root")` — teardown removes the base directory +> wholesale (the root's parent), so the site removal is one +> operation like the sqlite column's. +> - The `pglo-poc` container was already running in the dev +> environment (verified with `docker ps` before implementation); no +> environment changes were needed. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-12 (wave 4). The postgres column — the suite's +> first SKIP-posture column — landed green against the harness server +> (`pglo-poc`, :15432): 17 conformance tests (the mirrored +> form-independent reference rows — INV-1 round-trips, INV-6 stat +> discipline, the `NoLivenessSources`/`ProtectFailed` abort shapes, +> the kv-only rejection row — the durable-schedule sweep rows (INV-2 +> and INV-3 schedule-relative readings, the two-sweep exact-count row +> over the full protection set, the sweep-vs-put race across the +> window delay, the holder-stages-only lease shape, INV-5 list +> correctness), the dispatch rows with the placement oracle adapted to +> the pg medium, and the virgin-store row), every row ADR-stamped and +> traced to the backlog. Factory isolation (fresh schema per open, +> registry-tracked) and idempotent `DROP SCHEMA … CASCADE` teardown +> are structurally exercised (panicking-row-safe; `pg_namespace` +> verified empty after runs — the shared server carries no residue). +> README amended: the postgres rows table in §Columns beyond the +> reference set, the column plan's SKIP-posture note naming the +> landed column, and current state (four columns). Verification: +> `cargo test` (default) 69/19/3 ok — the column compiles out; +> `cargo test --features postgres` 102/36/3 ok green against the +> server, and green skipping server-less (17 skip-noted columns rows; +> default-build gates unaffected); `cargo clippy --all-targets -- -D +> warnings` green (default, `--features postgres`, `--all-features`); +> `cargo fmt --check` green. \ No newline at end of file diff --git a/tests/conformance/README.md b/tests/conformance/README.md index 6e85aa9..4b00a73 100644 --- a/tests/conformance/README.md +++ b/tests/conformance/README.md @@ -295,19 +295,62 @@ legal degenerate tuning, `Duration::ZERO` — execute-on-next-sweep): the schedule the durable rows assert against is exactly the next-sweep executor form. +**The postgres column's own rows** (tests/conformance/postgres_column.rs, +the first SKIP-posture column — the fleet-capable-kv shape, `kv = +postgres` + `large = local`, feature `postgres` default-off, +ADR-007/012 §5): the **durable-schedule sweep rows** mirror the sqlite +column's family (§The schedule-relative reading for SQL-backed +columns; ADR-012 §6.7 — the same window delay tuning, `0` — +execute-on-next-sweep) and the dispatch rows with the placement +oracle adapted by medium again — kv entries live as rows in the pg +schema's `blobs` table, large entries under the local root (an +independent connection counting the schema's rows vs the root's +committed files; the column's own rows also watch the schedule +itself, the schema's `gc_window` rows). The form-independent +reference rows mirror verbatim: the INV-1 round-trip family and the +INV-6 stat-discipline row against the dual-tier factory, the +`NoLivenessSources`/`ProtectFailed` abort shapes (form-independent — +abort before staging), and the kv-only rejection row against a +`kv-only(postgres)` factory (threshold-agnostic by discovery, +unchanged; the threshold materializable per the configuration +contract). Connection settings ride the environment +(`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`) and the runners gate +behind a reachability probe; the factories follow the alkstore +pg-factory pattern (alkstore-postgres `contract_suite.rs`): a fresh +engine-owned schema per `open()`, tracked per-instance, teardown +`DROP SCHEMA IF EXISTS … CASCADE` over exactly the minted schemas +(idempotent, owned, panicking-row-safe — `CASCADE` removes +dependents server-side while the abandoned store's pooled +connections fail closed on next use). + +| Row (`postgres_column.rs`) | Traces to | +|---|---| +| `dedup_identical_content_is_one_pool_entry` | INV-2 (same-path and cross-path legs; counts read against the durable schedule) | +| `stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success` | INV-3 (both put paths, both failure modes; schedule-relative counts) | +| `two_sweep_exact_counts_over_the_full_protection_set` | INV-4/INV-5 (the full protection set across two sweeps — staged this sweep, executed by the next; the both-tiers form) | +| `a_put_landing_during_the_window_survives_the_executing_sweep` | INV-4 (the sweep-vs-put race across the window delay; execution-time re-arbitration) | +| `the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only` | INV-4 (the `SweeperLock` abort shape — abort before staging; the holder's durable-form completion: stages only) | +| `list_correctness_is_observable_through_durable_schedule_outcomes` | INV-5 (exact counts, never list comparison) | +| `routing_is_a_pure_function_of_length_across_put_paths_and_re_puts` | ADR-003 §Dispatch policy (purity/no-migration); ADR-012 §6.3; ADR-012 §6.7 (schedule-relative dedup legs); ADR-007 §5; INV-7 facade-contract row | +| `threshold_boundary_lengths_route_deterministically` | ADR-003 §Dispatch policy (at/below/above, tuned + shipped-default thresholds); ADR-012 §6.3; ADR-007 §5 | +| `kv_tier_misses_fall_through_to_the_large_tier_per_op` | ADR-012 §6.1 (fall-through per op); ADR-012 §6.5 ruling 5 (miss shapes); ADR-006; ADR-012 §6.3; ADR-007 §5 | +| `virgin_store_reads_are_no_ops_on_a_fresh_schema` | store-api.md §Error model (virgin-store semantics); ADR-012 §6.5; ADR-007 §5 (the constructor always ensures the schema — the facade's fresh-schema form of the tolerance) | + ## Column plan **mem is the contract-reference engine** (ADR-011 §2): wave 2 lands its column first (the first column binds the `Store` facade type), and every other engine's rows mirror mem's column — an addition to mem's column propagates as a requirement to all other columns (INV-7). The -pg columns (postgres, pg-lo) additionally run the SKIP posture above; -the fleet-form rows of INV-4 are engine-gated to the postgres column -and first run in wave 6. +pg columns (postgres, pg-lo) additionally run the SKIP posture above — +the **postgres column is the posture's first landed host** (server-less +runs skip cleanly; its wave-4 acceptance is the green run against the +harness server); the fleet-form rows of INV-4 are engine-gated to the +postgres column and first run in wave 6. ## Current state -The harness is green on **three columns** (waves 2–3). The **mem +The harness is green on **four columns** (waves 2–4). 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 @@ -361,3 +404,30 @@ default build: every `cargo test` run exercises it. The sqlite engine's no-tables-yet read tolerance is engine-instrumented (src/sqlite.rs's tests; the constructor-vs-lazy timing design point there), the column's virgin-store row being its facade form. + +The **postgres column** (the suite's **first SKIP-posture column** — +the fleet-capable-kv shape, `kv = postgres` + `large = local`, feature +`postgres` **default-off**, ADR-007/013 §4, landed wave 4) mirrors the +form-independent reference rows (INV-1, INV-6, the +`NoLivenessSources`/`ProtectFailed` abort shapes, and the kv-only +rejection row against `kv-only(postgres)`) and grows the sqlite +column's durable-schedule sweep rows plus the dispatch rows with the +placement oracle adapted by medium (an independent connection counts +the pg schema's `blobs` rows vs the root's committed files; the +column's own rows watch the schedule's `gc_window` rows) and the +virgin-store row — the recorded inventory is in §Columns beyond the +reference set. Its factories declare the column-tuned threshold and +the test tuning of the window delay (`Duration::ZERO` — +execute-on-next-sweep) and follow the alkstore pg-factory pattern: a +fresh engine-owned schema per `open()` (unique per call, seeded per +tag/pid/time — racing opens included), tracked per-instance in the +schema registry, teardown `DROP SCHEMA IF EXISTS … CASCADE` over +exactly the minted schemas plus removal of every vended temp root +(idempotent, owned, panicking-row-safe; each runner owns its +instance). The column **skips cleanly server-less** under the +reachability probe (`ALKBLOBS_PG_HOST` gating; the remaining +settings default to the harness server's values) — a skip is not a +failure — and its acceptance is the green run **against the harness +server** (dockerized `pglo-poc`, `postgres:16-alpine`, :15432, db +`blobs`, verified live 2026-10-12; the alkstore harness's same +server). diff --git a/tests/conformance/main.rs b/tests/conformance/main.rs index 9f87773..0e801bf 100644 --- a/tests/conformance/main.rs +++ b/tests/conformance/main.rs @@ -30,7 +30,10 @@ //! shape, `kv = sqlite` + `large = local`), which mirrors the //! form-independent reference rows and grows the durable-schedule //! sweep rows (README §The schedule-relative reading for SQL-backed -//! columns). +//! columns); [`postgres_column`] — the suite's first SKIP-posture +//! column (the fleet-capable-kv shape, `kv = postgres` + `large = +//! `local`, feature `postgres` default-off), which mirrors the same +//! sets against the harness server 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. @@ -45,6 +48,8 @@ mod factory; mod local_column; #[cfg(feature = "mem")] mod mem_column; +#[cfg(feature = "postgres")] +mod postgres_column; mod rows; #[cfg(feature = "sqlite")] mod sqlite_column; diff --git a/tests/conformance/postgres_column.rs b/tests/conformance/postgres_column.rs new file mode 100644 index 0000000..d54f612 --- /dev/null +++ b/tests/conformance/postgres_column.rs @@ -0,0 +1,2076 @@ +//! The **postgres column** — the suite's first SKIP-posture column: +//! the reference row set run against the fleet-capable-kv shape +//! (`kv = postgres` + `large = local`; feature `postgres` is +//! default-off, ADR-007, ADR-012 §5), extended with the +//! **durable-schedule sweep rows** and the dispatch rows (README +//! §Columns beyond the reference set; INV-7: every engine's rows +//! mirror the reference column). +//! +//! The SKIP posture (README §The SKIP posture) is a factory-owner +//! responsibility here: the connection settings ride the environment +//! (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` — host required; the +//! rest default to the harness server's values), the runners gate +//! every row behind a **reachability probe**, and when no server is +//! reachable the column skips cleanly — a skip is not a failure. The +//! acceptance run is green **against the harness server** (dockerized +//! `pglo-poc`, `postgres:16-alpine`, :15432, db `blobs`) — a skip +//! never substitutes for it. The same server hosts the sqlite and +//! local columns' neighbors' needs; nothing is skipped server-less +//! except this column's rows. +//! +//! The sweep rows are **schedule-relative on this column** (the +//! README §The schedule-relative reading for SQL-backed columns is +//! the recorded amendment, citing ADR-012 §6.7): the durable form +//! stages its candidates this sweep and executes them by the next, +//! so the exact counts here are asserted against the staged schedule. +//! The form-independent rows mirror verbatim ([`crate::rows`]: the +//! round-trip family, the stat-discipline row, the no-sources and +//! failed-protect abort shapes, and the kv-only rejection row against +//! the kv-only factory); the schedule-relative properties grew their +//! own rows below (dedup and stage hygiene observe their invariants +//! through sweep outcomes, so they read against the schedule too). +//! Mem-only never mirrors here: the mode is not constructible +//! without mem. +//! +//! The factories (README §The factory contract; the alkstore +//! pg-factory pattern — alkstore-postgres `contract_suite.rs` — +//! reused): `open()` mints a **fresh engine-owned schema per call** — +//! unique per call (seeded per tag/pid/time so two factory instances +//! never share a schema name, racing opens included), tracked +//! per-instance in the schema registry — and vends the concrete +//! [`alkblobs::Store`] facade over it plus a fresh private temp +//! directory for the large engine's root, so two opens never share +//! state even on the shared server. Teardown `DROP SCHEMA IF EXISTS +//! … CASCADE` over exactly the schemas this factory instance minted +//! (idempotent, owned — never the shared server's other schemas; +//! safe against an abandoned store: `CASCADE` removes the dependents +//! server-side while the store's pooled connections fail closed on +//! next use) plus the removal of every temp root it vended. The +//! sqlite column's instance-tracks-its-own-site pattern over schemas +//! (plus roots): every runner constructs its own factory, so +//! concurrent rows never touch each other's schemas. +//! +//! The dual-tier dispatch rows mirror the sqlite column's family with +//! the placement oracle adapted by medium again: kv entries live as +//! **rows in the pg schema's `blobs` table**, large entries under the +//! local root — the oracle distinguishes by medium (an independent +//! connection counting the schema's rows vs the root's committed +//! files), and watches the schedule itself (the schema's `gc_window` +//! rows). + +#![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; + +use tokio_postgres::NoTls; + +use crate::factory::StoreFactory; +use crate::rows::{ + self, committed_file_count, patterned, slice_digest, staging_file_count, Failing, LatchProtect, + SetProtect, SetSource, +}; +use alkblobs::{Error, Key, PgOpts, Store, StoreMode, DEFAULT_DISPATCH_THRESHOLD}; + +// ------------------------------------------------------------------ +// The factories +// ------------------------------------------------------------------ + +/// The column's declared dual-tier threshold — small, so the shared +/// content matrix and the race fixtures cross it and both tiers' +/// bodies are exercised (sub-threshold rows land in the pg schema; +/// over-threshold rows stage-commit into the local root). +pub(crate) const TUNED_THRESHOLD: u64 = 1024; + +/// The column's declared kv-only rejection threshold — small, so every +/// leg of the 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 durable +/// schedule's staging sweep stages the dead set; the *next* sweep is +/// its executor, which is exactly the schedule the sweep rows assert +/// against. +pub(crate) const WINDOW_DELAY: Duration = Duration::ZERO; + +/// One vended backing site: the engine-owned schema this open minted +/// plus a fresh private temporary directory holding the large +/// engine's root. +#[derive(Clone)] +pub(crate) struct Site { + /// The server's DSN — the oracle's independent connections and the + /// kv tier's shared medium address. Two opens never share a + /// schema; isolation is schema-scoped, not server-scoped. + dsn: String, + /// The engine-owned schema this open minted — the kv tier's + /// medium (the dispatch oracle's kv-side counter rides the + /// schema's table rows). + schema: String, + /// The whole private directory — what teardown removes (the local + /// root's parent; kv-only sites leave it empty and still remove + /// it). + base: PathBuf, + /// The local root — the large tier's medium (the dispatch + /// oracle's file-side counter rides [`committed_file_count`]). + root: PathBuf, +} + +impl Site { + /// The schema-qualified pool table's count via the oracle. + pub(crate) fn schema(&self) -> &str { + &self.schema + } + + /// The local root path — for the file oracle's callers. + pub(crate) fn root(&self) -> &Path { + &self.root + } + + fn remove_site(self) -> Result<(), Error> { + match std::fs::remove_dir_all(&self.base) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(Error::Io(e.to_string())), + } + } +} + +/// A fresh private temp directory for one open's local root. +fn fresh_root(tag: &str) -> PathBuf { + static SEQ: AtomicU64 = AtomicU64::new(0); + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let base = std::env::temp_dir().join(format!( + "alkblobs-postgres-column-{tag}-{}-{:x}-{seq}", + std::process::id(), + nanos + )); + std::fs::create_dir_all(&base).expect("temp site created"); + base +} + +/// The harness connection settings ride the environment +/// (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` — the alkstore harness +/// convention). `ALKBLOBS_PG_HOST` set (non-empty) is the probe's +/// "configured" answer; port/user/password/db default to the harness +/// server's values (`pglo-poc`: :15432, `postgres`/`poc`, db +/// `blobs`). Absent → the caller skips. +pub(crate) fn harness_dsn() -> Option { + let host = std::env::var("ALKBLOBS_PG_HOST") + .ok() + .filter(|v| !v.is_empty())?; + let port = std::env::var("ALKBLOBS_PG_PORT") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "15432".to_owned()); + let user = std::env::var("ALKBLOBS_PG_USER") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "postgres".to_owned()); + let password = std::env::var("ALKBLOBS_PG_PASSWORD") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "poc".to_owned()); + let db = std::env::var("ALKBLOBS_PG_DB") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "blobs".to_owned()); + Some(format!( + "host={host} port={port} user={user} password={password} dbname={db}" + )) +} + +/// The skip note — the SKIP posture's clean-skip print, exactly once +/// per skipped test. +fn skip_note() { + eprintln!("skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"); +} + +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) +} + +/// The server reachability probe: the SKIP posture's gate the runners +/// check before their rows run. One cheap admin connection per test; +/// `teardown` reconnects independently (no shared admin client +/// lifetime). +pub(crate) async fn harness_ready() -> bool { + let Some(dsn) = harness_dsn() else { + skip_note(); + return false; + }; + match tokio::time::timeout(Duration::from_secs(3), admin_connect(&dsn)).await { + Ok(Ok(_)) => true, + Ok(Err(e)) => { + eprintln!("skipping (harness server unreachable): {e}"); + false + } + Err(_) => { + eprintln!("skipping (harness server unreachable: connect timeout)"); + false + } + } +} + +/// 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 name, racing opens included. +fn fresh_schema_name(tag: &str, seq: &AtomicU64) -> String { + format!( + "alkblobs_suite_{}_{}_{}_{}", + tag, + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0), + seq.fetch_add(1, Ordering::Relaxed), + ) +} + +/// The dispatch-coverage factory: `kv = postgres` + `large = local` at +/// the threshold this column declares and the test tuning of the +/// window delay (see [`TUNED_THRESHOLD`], [`WINDOW_DELAY`], and the +/// default-threshold sibling `DualTierPostgresFactory::new( +/// DEFAULT_DISPATCH_THRESHOLD)` the boundary row runs against). +/// +/// **Why a fresh instance per runner (not a shared static):** teardown +/// drops the schemas *this instance* minted (and removes the roots it +/// vended); a shared instance's teardown would race concurrently +/// running rows' sites. Every runner constructs its own and owns its +/// lifecycle. +pub(crate) struct DualTierPostgresFactory { + dsn: String, + /// The declared dispatch threshold — the routing boundary this + /// column's rows run at (constructor-tunable; ADR-003). Declared + /// configuration, like the local column's (the dual-tier store + /// never rejects, so there is nothing to probe). + threshold: u64, + /// The declared window delay — the durable schedule's cadence knob + /// the sweep rows run at (the test tuning: execute-on-next-sweep). + window_delay: Duration, + schema_seq: AtomicU64, + /// Every schema this instance minted, for its own teardown. + schemas: Mutex>, + /// Every temp site this instance vended, for its own teardown. + sites: Mutex>, +} + +impl DualTierPostgresFactory { + pub(crate) fn new(threshold: u64) -> DualTierPostgresFactory { + DualTierPostgresFactory { + dsn: harness_dsn().expect("the harness server is configured for the column"), + threshold, + window_delay: WINDOW_DELAY, + schema_seq: AtomicU64::new( + (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0)) + ^ ((std::process::id() as u64) << 8), + ), + schemas: Mutex::new(Vec::new()), + sites: Mutex::new(Vec::new()), + } + } + + /// Open + track the vended site: the shape the column's own rows + /// use (they observe the backing medium, so they need the site + /// alongside the store). + pub(crate) async fn open_tracked(&self) -> Result<(Store, Site), 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 = fresh_root("dual-tier"); + let site = Site { + dsn: self.dsn.clone(), + schema: schema.clone(), + root: base.join("local-root"), + base, + }; + let store = Store::dual_tier_postgres_local( + &self.dsn, + opts(schema), + site.root.clone(), + self.threshold, + self.window_delay, + ) + .await?; + self.sites + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(site.base.clone()); + Ok((store, site)) + } + + /// The schemas this instance minted, for the structural tests' + /// teardown assertions. + pub(crate) fn tracked_schemas(&self) -> Vec { + self.schemas + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + + /// The DSN the factory opens against. + pub(crate) fn dsn(&self) -> &str { + &self.dsn + } +} + +/// The kv-only(postgres) factory (ADR-011 §4; ADR-012 §6.2): no large +/// tier; the threshold is the put-time rejection point the kv-only +/// row discovers by probing. The fleet host's kv-only shape. +pub(crate) struct KvOnlyPostgresFactory { + dsn: String, + /// The rejection point this column's configuration declares. The + /// kv-only row discovers it behaviorally and requires it + /// materializable for its boundary legs. + pub(crate) threshold: u64, + window_delay: Duration, + schema_seq: AtomicU64, + schemas: Mutex>, +} + +impl KvOnlyPostgresFactory { + pub(crate) fn new(threshold: u64) -> KvOnlyPostgresFactory { + KvOnlyPostgresFactory { + dsn: harness_dsn().expect("the harness server is configured for the column"), + threshold, + window_delay: WINDOW_DELAY, + schema_seq: AtomicU64::new( + (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0)) + ^ ((std::process::id() as u64) << 8), + ), + schemas: Mutex::new(Vec::new()), + } + } +} + +/// The factories' schema-name policy, in engine-opts form: the +/// constructor's `PgOpts` with exactly the factory's minted schema +/// and the ship defaults for the rest (pool size, +/// `synchronous_commit` — the alkstore factory's shape: only the +/// schema name is factory policy). +fn opts(schema: String) -> PgOpts { + PgOpts { + schema, + ..PgOpts::default() + } +} + +impl StoreFactory for DualTierPostgresFactory { + fn name(&self) -> &'static str { + "dual-tier postgres+local" + } + + async fn open(&self) -> Result { + self.open_tracked().await.map(|(store, _)| store) + } + + 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 { + match std::fs::remove_dir_all(&base) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(Error::Io(e.to_string())), + } + } + Ok(()) + } +} + +impl StoreFactory for KvOnlyPostgresFactory { + fn name(&self) -> &'static str { + "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()); + let store = + Store::kv_only_postgres(&self.dsn, opts(schema), self.threshold, self.window_delay) + .await?; + Ok(store) + } + + 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(()) + } +} + +// ------------------------------------------------------------------ +// The medium oracle (the dispatch rows' instrument, adapted from the +// sqlite column's): both tiers are on the server's disk and the local +// root now, so the oracle distinguishes by medium — an independent +// connection counting the schema's tables vs the root's committed +// files. +// ------------------------------------------------------------------ + +/// Rows in one of the schema's tables — the kv tier's pool (`blobs`) +/// or the durable window's staged candidates (`gc_window`) — counted +/// from the oracle's own independent connection to the shared server. +async fn pg_table_rows(site: &Site, table: &str) -> usize { + let client = admin_connect(&site.dsn) + .await + .expect("the oracle connection opens"); + let sql = format!("SELECT COUNT(*) FROM \"{}\".{table}", site.schema()); + let row = client + .query_one(&sql, &[]) + .await + .expect("the oracle query succeeds"); + let count: i64 = row.get(0); + count.max(0) as usize +} + +/// The pg kv tier's pool residency — exactly one row per committed +/// kv-tier entry (the CAS table's rows). +async fn pg_pool_rows(site: &Site) -> usize { + pg_table_rows(site, "blobs").await +} + +/// The durable window's staged-candidate residency — the schedule +/// observer the durable rows assert against (staged rows land here; +/// the executing sweep's take empties them). +async fn pg_window_rows(site: &Site) -> usize { + pg_table_rows(site, "gc_window").await +} + +/// Whether the engine-owned schema still exists server-side — the +/// structural teardown tests' instrument (`pg_namespace`, the same +/// probe the alkstore factory shape uses). +async fn schema_exists(dsn: &str, schema: &str) -> bool { + let client = admin_connect(dsn) + .await + .expect("the oracle connection opens"); + let count: i64 = client + .query_one( + "SELECT count(*) FROM pg_namespace WHERE nspname = $1", + &[&schema], + ) + .await + .expect("the oracle query succeeds") + .get(0); + count > 0 +} + +// ------------------------------------------------------------------ +// The column's own rows — the durable-schedule sweep family, the +// schedule-relative readings of the sweep-counting INV rows, and the +// dispatch forms. Every row takes this column's concrete factory (it +// observes the backing medium), is ADR-stamped per the convention, +// and traces its family. +// ------------------------------------------------------------------ + +/// INV-2 on the durable schedule: identical content put twice — +/// through the same put path and across the two paths — is one pool +/// entry (one row in the schema), and the re-put's `Pin` protects +/// exactly as a fresh put's does. One-pool-entry is asserted through +/// durable-schedule outcomes (the INV-5 posture): the pins held stage +/// nothing; once unpinned, sweep 1 stages exactly the one entry and +/// removes nothing, and its executor sweep deletes exactly that +/// staged set — never a list comparison, and never a single-sweep +/// count (README §The schedule-relative reading for SQL-backed +/// columns). +/// +/// Contract stamp: store-api.md §Invariants 2; ADR-003 (whole-value +/// put is CAS first-writer-wins); ADR-005 (a re-put refcounts the +/// existing entry under the same arbitration lock); ADR-008 §2 (the +/// staged re-arbitrated delete); ADR-012 §6.7 (the executor is the +/// next sweep — the schedule these counts read against). +pub async fn dedup_identical_content_is_one_pool_entry(factory: &DualTierPostgresFactory) { + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let content = b"deduplicated content".to_vec(); + let canonical = Key::derive(&content, content.len() as u64); + + let (key, first) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fresh put publishes"); + let (reput, second) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the same-path re-put publishes"); + assert_eq!(key, canonical); + assert_eq!(reput, canonical, "one derivation, one address"); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + report.candidates_staged, 0, + "one pool entry, both refcounts composing under the same lock" + ); + assert_eq!( + pg_pool_rows(&site).await, + 1, + "one pool entry, both refcounts — one row in the schema" + ); + drop(first); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the re-put's pin protects exactly as a fresh put's" + ); + drop(second); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 0), + "exactly one pool entry all along — staged this sweep, not executed" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.deleted, report.cancelled_by_arbitration), + (1, 0), + "the executing sweep deleted exactly the staged entry" + ); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + assert_eq!(pg_pool_rows(&site).await, 0, "the executed delete landed"); + + let cross = b"deduplicated across the put paths".to_vec(); + let cross_canonical = Key::derive(&cross, cross.len() as u64); + let (key, first) = store + .put(Some(cross.len() as u64), Cursor::new(cross.as_slice())) + .await + .expect("the known-length put publishes"); + let (reput, second) = store + .put(None, Cursor::new(cross.as_slice())) + .await + .expect("the unknown-length re-put publishes"); + assert_eq!(key, cross_canonical); + assert_eq!(reput, cross_canonical, "one pool entry across the paths"); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 0, "both refcounts compose"); + drop(second); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the re-put's pin protects exactly as a fresh put's (other drop order)" + ); + drop(first); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 0), + "exactly one pool entry all along" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "executed exactly the staged set"); + assert!(!store.has(&cross_canonical.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-3 on the durable schedule: stage hygiene — every put-path +/// failure leaves zero residue and every success leaves exactly one +/// committed entry, over both put paths and both injectable failure +/// modes (a mid-stream stream failure and a declared-length +/// mismatch). The kv tier's stage form is degenerate (nothing stages +/// outside a flush), so residue is the pool itself — the schema's +/// rows, which the failures never touch — observed through +/// durable-schedule sweep outcomes: zero staged after every failure, +/// and a success staged exactly once, then executed exactly once. +/// +/// Contract stamp: store-api.md §Invariants 3; ADR-010 §1 (the +/// staged-put contract: commit publishes, every failure path +/// discards); ADR-003 (stage-then-commit, the kv form degenerate); +/// ADR-012 §6.7 (the schedule-relative reading the success counts +/// follow). +pub async fn stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success( + factory: &DualTierPostgresFactory, +) { + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let content = patterned(200, 167); + let canonical = Key::derive(&content, content.len() as u64); + + let err = store + .put(Some(content.len() as u64), Failing::new(&content, 40)) + .await + .expect_err("the failed mid-stream put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + let err = store + .put(None, Failing::new(&content, 40)) + .await + .expect_err("the failed unknown-length put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + let err = store + .put(Some(100), Cursor::new(content.as_slice())) + .await + .expect_err("the over-fed mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + let err = store + .put(Some(400), Cursor::new(content.as_slice())) + .await + .expect_err("the under-fed mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "zero residue after every failure, both put paths, both failure modes" + ); + assert_eq!( + pg_pool_rows(&site).await, + 0, + "the pool table holds zero entries after every failure" + ); + + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the known-length put publishes"); + assert_eq!(key, canonical); + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 0), + "exactly one committed entry staged on success" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.deleted, 1, + "the executing sweep removed exactly that one entry" + ); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + + let (key, pin) = store + .put(None, Cursor::new(content.as_slice())) + .await + .expect("the unknown-length put publishes"); + assert_eq!(key, canonical); + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 0), + "exactly one committed entry staged on the other put path too" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "and executed exactly that staged set"); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the durable-schedule happy shape: the full protection set +/// spanning both tiers (a source-live kv entry, a pinned large-tier +/// entry, a protect-named large-tier entry) plus one dead entry per +/// tier — sweep 1 stages exactly the two dead candidates and deletes +/// nothing (staging is not executing: the pool and the medium hold +/// everything and the window holds the staged rows); sweep 2 executes +/// exactly that staged set (`deleted + cancelled == N` against the +/// staged set), and the protected entries stay byte-identical. The +/// exact staged candidates also make this the INV-5 posture's +/// both-tiers form: list correctness (complete and well-formed across +/// both tiers' enumerations) is observable only through these +/// outcomes. The medium oracle distinguishes the tiers (the schema's +/// pool rows vs the root's committed files) and watches the schedule +/// itself (the window's rows). +/// +/// Contract stamp: store-api.md §Invariants 4 (exact counts with +/// correct liveness) and §Invariants 5 (list correctness through +/// sweep outcomes); ADR-005 (three liveness marks; delete windows +/// with arbitrate-at-delete-time); ADR-008 §2 (the staged +/// re-arbitrated delete; the constructor-tunable delay); ADR-012 §1 +/// (sweeps are never errors — the report is the outcome record); +/// ADR-012 §6.7 (the executor is the next sweep — the schedule these +/// counts read against); ADR-007 §5 (postgres, the fleet host's kv +/// engine — the exact-count sweep-outcome gate against dockerized +/// postgres). +pub async fn two_sweep_exact_counts_over_the_full_protection_set( + factory: &DualTierPostgresFactory, +) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + let source = SetSource::register(&store); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let source_live = patterned((threshold - 1) as usize, 191); + let (key_live, _) = store + .put( + Some(source_live.len() as u64), + Cursor::new(source_live.as_slice()), + ) + .await + .expect("the source-live entry publishes"); + source.add(key_live.clone()); + + let pinned = patterned(2 * threshold as usize + 7, 173); + let (key_pinned, pin_pinned) = store + .put(None, Cursor::new(pinned.as_slice())) + .await + .expect("the pinned entry publishes"); + + let protected = patterned(2 * threshold as usize + 11, 167); + let (key_protected, _) = store + .put( + Some(protected.len() as u64), + Cursor::new(protected.as_slice()), + ) + .await + .expect("the protect-named entry publishes"); + protect.set_keys(vec![key_protected.clone()]); + + let dead_kv = patterned((threshold - 3) as usize, 149); + let (key_dead_kv, pin) = store + .put(Some(dead_kv.len() as u64), Cursor::new(dead_kv.as_slice())) + .await + .expect("the kv-tier dead entry publishes"); + drop(pin); + let dead_large = patterned(2 * threshold as usize + 13, 151); + let (key_dead_large, pin) = store + .put(None, Cursor::new(dead_large.as_slice())) + .await + .expect("the large-tier dead entry publishes"); + drop(pin); + + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (2, 3), + "before the sweep, both tiers hold their entries exactly \ + (two pool rows in the schema, three committed files \ + under the root)" + ); + + let first = store.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), + "sweep 1 staged exactly the two dead entries and deleted \ + nothing — staging is not executing" + ); + assert_eq!( + pg_window_rows(&site).await, + 2, + "the staged set waits in the durable window" + ); + for (key, content) in [ + (&key_live, &source_live), + (&key_pinned, &pinned), + (&key_protected, &protected), + ] { + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the protected entry is present"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content.as_slice(), + "the protected entry is byte-identical after the staging sweep" + ); + } + assert!( + store.has(&key_live.as_bytes()).await.unwrap() + && store.has(&key_pinned.as_bytes()).await.unwrap() + && store.has(&key_protected.as_bytes()).await.unwrap(), + "the staged sweep deleted nothing: every entry intact" + ); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + second.candidates_staged, 0, + "nothing newly dead since the staging sweep" + ); + assert_eq!( + second.deleted + second.cancelled_by_arbitration, + 2, + "sweep 2 executed exactly the staged set" + ); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (2, 0), + "with correct liveness, every executed candidate deleted" + ); + assert!(!store.has(&key_dead_kv.as_bytes()).await.unwrap()); + assert!(!store.has(&key_dead_large.as_bytes()).await.unwrap()); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (1, 2), + "the executed deletes landed on both media exactly" + ); + assert_eq!( + pg_window_rows(&site).await, + 0, + "the executed rows left the durable window" + ); + + drop(pin_pinned); + let third = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (third.candidates_staged, third.deleted), + (1, 0), + "with the pin dropped, the formerly pinned entry stages" + ); + let fourth = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + fourth.deleted, 1, + "and the next sweep executes it — exactly one deletion" + ); + assert!(!store.has(&key_pinned.as_bytes()).await.unwrap()); + assert!(store.has(&key_live.as_bytes()).await.unwrap()); + assert!(store.has(&key_protected.as_bytes()).await.unwrap()); + assert_eq!( + committed_file_count(site.root()), + 1, + "exactly the protect-named entry's file remains" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the race, in its durable-schedule form: a put landing +/// *during the window* survives the executing sweep — the re-put's +/// pin is minted (pin-before-publish) between the staging sweep and +/// the executing sweep, so the execution-time re-arbitration cancels +/// the pinned candidate while the unpinned twin deletes; once the pin +/// drops, the twin re-stages and executes exactly once. Both put +/// paths race in-body over both tiers (a kv twin and a large twin). +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-005 (pin-before- +/// publish ordering; the delete window closes the sweep-vs-put race); +/// ADR-008 §2 (the staged re-arbitrated delete — the race holds +/// across the window delay); ADR-012 §6.7 (execution-time +/// re-arbitration is the executor sweep's posture). +pub async fn a_put_landing_during_the_window_survives_the_executing_sweep( + factory: &DualTierPostgresFactory, +) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let rounds: [(String, Vec, bool, bool); 2] = [ + ( + "the kv twin (known-path re-put)".to_owned(), + patterned((threshold - 1) as usize, 191), + true, + true, + ), + ( + "the large twin (unknown-path re-put)".to_owned(), + patterned(2 * threshold as usize + 5, 173), + false, + false, + ), + ]; + for (index, (tag, content, original_known, reput_known)) in rounds.iter().enumerate() { + let len = content.len() as u64; + let twin = Key::derive(content, len); + let witness = format!("the dead witness of round {index}").into_bytes(); + let witness_key = Key::derive(&witness, witness.len() as u64); + + let original = if *original_known { Some(len) } else { None }; + let (_, pin) = store + .put(original, Cursor::new(content.as_slice())) + .await + .expect("the twin publishes"); + drop(pin); + let (_, pin) = store + .put(None, Cursor::new(witness.as_slice())) + .await + .expect("the witness publishes"); + drop(pin); + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + ( + staged.aborted, + staged.candidates_staged, + staged.deleted, + staged.cancelled_by_arbitration + ), + (None, 2, 0, 0), + "{tag}: the staging sweep stages exactly the two dead \ + twins and deletes nothing" + ); + + // The put lands during the window (the re-put dedups into the + // existing twin and refcounts its pin): protected across the + // delay, pin-before-publish. + let reput = if *reput_known { Some(len) } else { None }; + let (_, pin) = store + .put(reput, Cursor::new(content.as_slice())) + .await + .expect("the re-put publishes"); + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(executed.aborted, None, "{tag}"); + assert_eq!( + executed.candidates_staged, 0, + "{tag}: nothing newly dead since the staging sweep" + ); + assert_eq!( + executed.deleted + executed.cancelled_by_arbitration, + 2, + "{tag}: the executing sweep ran exactly the staged set" + ); + assert_eq!(executed.deleted, 1, "{tag}: the unpinned witness deleted"); + assert_eq!( + executed.cancelled_by_arbitration, 1, + "{tag}: the pinned twin's candidate cancelled at execution time" + ); + assert!( + !store.has(&witness_key.as_bytes()).await.unwrap(), + "{tag}: the unpinned twin is gone" + ); + let (got_len, blob) = store + .get(&twin.as_bytes()) + .await + .expect("get succeeds") + .expect("{tag}: the put landing during the window survived"); + assert_eq!(got_len, len, "{tag}"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content.as_slice(), + "{tag}: the surviving twin is byte-identical" + ); + drop(pin); + + let restaged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (restaged.candidates_staged, restaged.deleted), + (1, 0), + "{tag}: the pin dropped — the twin re-stages" + ); + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "{tag}: the re-staged candidate executes exactly once" + ); + assert!(!store.has(&twin.as_bytes()).await.unwrap()); + assert_eq!( + pg_window_rows(&site).await, + 0, + "{tag}: the window holds nothing after the executed round" + ); + } + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, abort shape: the sweep is single-flight on the durable form +/// too — the in-process sweep lease (ADR-013 §2's single-node analog). +/// The contended second sweeper aborts with `SweeperLock` in a +/// completed `Ok(SweepReport)` — nothing executed, nothing staged, +/// nothing deleted (abort before staging: form-independent) — while +/// the lease holder's sweep completes by **staging only**: its staged +/// set waits in the window for the next sweep to execute (the +/// durable form's schedule-relative replacement for the reference +/// row's inline holder-completion leg). +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-012 §1 (abort is +/// report data, never an error); ADR-013 §2 (the named exclusive +/// sweep lease — one sweeper per pool, holdership ends at release); +/// ADR-012 §6.7 (the executor is the next sweep — the holder stages, +/// it does not execute). +pub async fn the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only( + factory: &DualTierPostgresFactory, +) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + store.register_protect_callback(Arc::new(LatchProtect { + entered: entered_tx, + release: parking_lot::Mutex::new(Some(release_rx)), + })); + + let content = patterned((threshold - 1) as usize, 167); + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the dead witness publishes"); + drop(pin); + + // The lease holder sweeps on its own task (its protect consult + // blocks that task's thread mid-callback); this task contends and + // then releases the holder. + let store = Arc::new(store); + let holder = tokio::spawn({ + let store = Arc::clone(&store); + async move { store.sweep().await } + }); + + while entered_rx.recv_timeout(Duration::from_millis(50)).is_err() { + tokio::task::yield_now().await; + } + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + second.aborted, + Some(alkblobs::GcAbortCause::SweeperLock), + "the contended sweeper aborts as report data" + ); + assert_eq!( + ( + second.candidates_staged, + second.deleted, + second.cancelled_by_arbitration + ), + (0, 0, 0), + "the contended sweep executes, stages, and deletes nothing" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "the contended sweep deleted nothing" + ); + assert_eq!( + pg_window_rows(&site).await, + 0, + "the aborted sweep staged nothing into the window" + ); + release_tx.send(()).ok(); + + let report = holder + .await + .expect("the holder task joined") + .expect("sweeps are never errors"); + assert_eq!(report.aborted, None, "the lease holder completes"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 0), + "the holder stages exactly the dead set — it does not execute \ + (the durable form's schedule)" + ); + assert_eq!( + pg_window_rows(&site).await, + 1, + "the holder's staged set waits in the durable window" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "the holder staged, did not execute: the entry is intact" + ); + + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "the next sweep is the executor of the staged set" + ); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-5 on the durable schedule: `list()` correctness is observable +/// only through GC — the enumeration is complete and well-formed in +/// both directions, proven by exact durable-schedule outcomes: the +/// staging sweep stages exactly the dead set (neither more nor fewer +/// candidates), the executing sweep deletes exactly that staged set, +/// and a liveness flip re-stages exactly the newly dead entry. Never +/// a direct list comparison. +/// +/// Contract stamp: store-api.md §Invariants 5; ADR-003 (`list()` is +/// complete and well-formed by contract — load-bearing for GC); +/// ADR-005 (the sweep's live-set diff rides `list()`); ADR-012 §6.7 +/// (the executor is the next sweep — the schedule these counts read +/// against). +pub async fn list_correctness_is_observable_through_durable_schedule_outcomes( + factory: &DualTierPostgresFactory, +) { + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + let source = SetSource::register(&store); + + let mut keys = Vec::new(); + for (index, content) in [ + b"kept live by the source, entry one".to_vec(), + b"kept live by the source, entry two".to_vec(), + b"dead at the first staging, entry three".to_vec(), + b"dead at the first staging, entry four".to_vec(), + b"dead at the first staging, entry five".to_vec(), + ] + .into_iter() + .enumerate() + { + let declared = if index % 2 == 0 { + Some(content.len() as u64) + } else { + None + }; + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("the entry publishes"); + drop(pin); + keys.push(key); + } + source.add(keys[0].clone()); + source.add(keys[1].clone()); + + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + (first.candidates_staged, first.deleted), + (3, 0), + "the enumeration staged exactly the three dead entries — \ + complete and well-formed in both directions" + ); + for key in &keys { + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "staging deleted nothing: every enumerated entry is intact" + ); + } + assert_eq!( + pg_window_rows(&site).await, + 3, + "the staged schedule holds the three candidates" + ); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (second.candidates_staged, second.deleted), + (0, 3), + "the executing sweep removed exactly the staged set" + ); + assert!(store.has(&keys[0].as_bytes()).await.unwrap()); + assert!(store.has(&keys[1].as_bytes()).await.unwrap()); + for key in &keys[2..] { + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + source.remove(&keys[1]); + let third = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (third.candidates_staged, third.deleted), + (1, 0), + "the newly dead entry is staged exactly" + ); + let fourth = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(fourth.deleted, 1, "and executed exactly by the next sweep"); + assert!(!store.has(&keys[1].as_bytes()).await.unwrap()); + assert!(store.has(&keys[0].as_bytes()).await.unwrap()); + + source.remove(&keys[0]); + let fifth = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(fifth.candidates_staged, 1, "the last entry stages exactly"); + let sixth = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(sixth.deleted, 1, "and executes exactly"); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Dispatch row: **routing is a pure function of length** — at, below, +/// and over the declared threshold, the same content lands on exactly +/// the same tier through both put paths (same placement every time; +/// nothing moves, nothing duplicates), identical content through both +/// paths is one pool entry, and every read is byte-identical from +/// wherever that tier holds it. +/// +/// Tier placement is observed through the backing medium (the module +/// doc's oracle, adapted): kv entries hold a row in the pg schema and +/// none under the root; large entries hold a committed file under the +/// root and no pool row — the facade cannot and must not expose tier +/// residency (it is engine-blind). +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row); +/// ADR-003 §Dispatch policy (routing is a pure function of length; +/// no migration; CAS first-writer-wins makes re-puts one entry); +/// ADR-005 (re-put refcounts); ADR-007 §5 (postgres, the fleet host's +/// kv engine); ADR-012 §6.3 (the dual-tier dispatch-coverage shape); +/// ADR-012 §6.7 (the schedule-relative cleanup the dedup legs read +/// against). +pub async fn routing_is_a_pure_function_of_length_across_put_paths_and_re_puts( + factory: &DualTierPostgresFactory, +) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + for content in [ + patterned((threshold - 1) as usize, 191), + patterned(threshold as usize, 173), + patterned((threshold + 1) as usize, 167), + patterned(4 * threshold as usize + 9, 151), + ] { + let len = content.len() as u64; + let canonical = Key::derive(&content, len); + let expected_rows = if len > threshold { 0 } else { 1 }; + let expected_files = if len > threshold { 1 } else { 0 }; + + for (path_tag, declared) in [("the known path", Some(len)), ("the unknown path", None)] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!( + key, canonical, + "the canonical derivation addresses the put on {path_tag}" + ); + assert_eq!( + pg_pool_rows(&site).await, + expected_rows, + "length {len} on {path_tag}: the placement is exactly the \ + routing function's answer and never moves or duplicates \ + (the pg medium holds {expected_rows} kv rows)" + ); + assert_eq!( + committed_file_count(site.root()), + expected_files, + "length {len} on {path_tag}: the large-tier medium holds \ + {expected_files} committed files" + ); + assert_eq!( + staging_file_count(site.root()), + 0, + "a published put leaves no stage file on {path_tag}" + ); + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!( + got_len, len, + "the read serves the entry from its placed tier" + ); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "the entry is byte-identical from its placed tier across paths" + ); + drop(pin); + } + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + staged.candidates_staged, 1, + "identical content through both put paths is one pool entry" + ); + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "the executing sweep removed exactly the one entry (durable \ + schedule: staged this sweep, executed by the next)" + ); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (0, 0), + "both media are back to zero" + ); + } + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Dispatch row: **threshold boundary lengths route deterministically** +/// — at/below/above the factory's declared threshold, on both put +/// paths, the placement is the routing function's answer every time +/// (the medium oracle distinguishes the schema's rows from the root's +/// committed files); and the default-threshold boundary re-routes +/// with the tuning (the constructor-tuned threshold deterministically +/// places the default boundary's length on the tier its own threshold +/// names). Run against two factories by the runner — the column's +/// tuned threshold and the shipped default — the same row shape at +/// two declared thresholds being the tunable re-route demonstration. +/// The durable cleanup: the staged sweep stages exactly the six +/// entries' dead set and the next sweep executes it. +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row); +/// ADR-003 §Dispatch policy (default 128 KiB at-or-below routes kv; +/// constructor-tunable without API change); ADR-007 §5 (postgres, the +/// fleet host's kv engine); ADR-012 §6.3 (the dual-tier +/// dispatch-coverage shape); ADR-012 §6.7 (schedule-relative cleanup). +pub async fn threshold_boundary_lengths_route_deterministically(factory: &DualTierPostgresFactory) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let boundary = [ + (threshold - 1, 191u32, false), + (threshold, 173, false), + (threshold, 167, false), + (threshold + 1, 151, true), + (threshold + 1, 139, true), + ]; + let mut expected_rows = 0u64; + let mut expected_files = 0u64; + for (len, modulus, lands_large) in boundary { + let content = patterned(len as usize, modulus); + for declared in [Some(content.len() as u64), None] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!( + key, + Key::derive(&content, len), + "the canonical derivation addresses the put" + ); + assert_eq!( + pg_pool_rows(&site).await as u64, + expected_rows + if lands_large { 0 } else { 1 }, + "boundary length {len}: at/below the threshold lands kv \ + (one row in the schema), over it lands large (a \ + committed file instead), both put paths" + ); + assert_eq!( + committed_file_count(site.root()) as u64, + expected_files + if lands_large { 1 } else { 0 }, + "boundary length {len}: the large-tier medium matches the \ + routing answer" + ); + drop(pin); + } + if lands_large { + expected_files += 1; + } else { + expected_rows += 1; + } + } + + // The default boundary re-routes with the constructor's tuning: + // the shipped default's length belongs to the tier this + // configuration's own threshold puts it on. + let at_default = patterned(DEFAULT_DISPATCH_THRESHOLD as usize, 199); + let default_lands_large = DEFAULT_DISPATCH_THRESHOLD > threshold; + let (key, pin) = store + .put( + Some(DEFAULT_DISPATCH_THRESHOLD), + Cursor::new(at_default.as_slice()), + ) + .await + .expect("the default-boundary put publishes"); + assert_eq!( + key, + Key::derive(&at_default, DEFAULT_DISPATCH_THRESHOLD), + "the canonical derivation addresses the put" + ); + assert_eq!( + pg_pool_rows(&site).await as u64, + expected_rows + if default_lands_large { 0 } else { 1 }, + "the default boundary's length routes to the tier this \ + configuration's tuning names (the pg medium's answer)" + ); + assert_eq!( + committed_file_count(site.root()) as u64, + expected_files + if default_lands_large { 1 } else { 0 }, + "the default boundary's length routes to the tier this \ + configuration's tuning names (the large medium's answer)" + ); + drop(pin); + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + staged.candidates_staged, 6, + "each leg's content was exactly one pool entry" + ); + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 6, + "the executing sweep removed every entry, tier notwithstanding" + ); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (0, 0), + "both media are back to zero" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Dispatch row: **a kv-tier miss falls through to the large tier on +/// every key-addressed op** — content the routing function placed on +/// the large tier (one committed file under the root; the pg medium +/// holds no pool row — the kv tier misses) serves `get`, `stat`, +/// `has`, and `read_range` whole, exact, and digest-correct; the +/// mirror-position leg proves the kv tier is the *first* consult +/// (sub-threshold content, one schema row and no committed file, +/// reads still served); and both-tier misses resolve per the error +/// model's per-op shapes (reached here by the durable schedule's +/// staged-then-executed deletion of the subject entry). +/// +/// Contract stamp: store-api.md §Read path (fall-through consult +/// order); ADR-012 §6.1 (every key-addressed op consults kv first, +/// then large; the general rule); ADR-012 §6.5 ruling 5 (both-tier +/// miss shapes: `None` on get/stat, `Err(Missing)` on read_range); +/// ADR-006 (the slice digest); ADR-007 §5 (postgres, the fleet host's +/// kv engine); ADR-012 §6.3 (the dispatch-coverage shape). +pub async fn kv_tier_misses_fall_through_to_the_large_tier_per_op( + factory: &DualTierPostgresFactory, +) { + let threshold = factory.threshold; + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + // Over-threshold content — the kv tier misses, the large tier + // serves every op. Both put paths hold the shape. + for (content, declared) in [ + ( + patterned(3 * threshold as usize + 41, 181), + Some(3 * threshold + 41), + ), + (patterned(3 * threshold as usize + 43, 197), None), + ] { + let len = content.len() as u64; + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (0, 1), + "the content lives on exactly the large tier — the kv tier \ + misses" + ); + + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the large tier served the kv miss"); + assert_eq!(got_len, len); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "the fall-through read is byte-identical" + ); + assert_eq!( + store.stat(&key.as_bytes()).await.expect("stat succeeds"), + Some(alkblobs::EntryMeta { + len, + algorithm: key.algorithm(), + }), + "stat falls through and carries the length plus algorithm" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "has falls through" + ); + let mid = len / 2; + let (slice, digest) = store + .read_range(&key.as_bytes(), mid, 17) + .await + .expect("read_range falls through"); + assert_eq!( + slice, + content[(mid as usize)..(mid as usize + 17)], + "the range slice is exact" + ); + assert_eq!( + digest, + slice_digest(&slice), + "the slice digest is SHA-256 over exactly the returned bytes" + ); + + // Sweep the entry away (the pin dropped): the miss shapes now + // resolve per op — the large tier misses too. The durable + // schedule: staged this sweep, executed by the next. + drop(pin); + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!((staged.candidates_staged, staged.deleted), (1, 0)); + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "exactly the one entry died across the two sweeps" + ); + assert!( + store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "a both-tier miss on get is Ok(None)" + ); + assert_eq!( + store.stat(&key.as_bytes()).await.expect("stat succeeds"), + None, + "a both-tier miss on stat is Ok(None)" + ); + assert!( + !store.has(&key.as_bytes()).await.unwrap(), + "a both-tier miss on has is false" + ); + assert_eq!( + store + .read_range(&key.as_bytes(), 0, 4) + .await + .expect_err("the both-tier miss errs"), + Error::Missing, + "a both-tier miss on read_range is Err(Missing)" + ); + } + + // The mirror-position leg: sub-threshold content — the kv tier is + // the first consult and serves before the large tier is reached + // (one row in the schema; no committed file ever lands). + let small = patterned((threshold - 1) as usize, 149); + let (key, pin) = store + .put(Some(small.len() as u64), Cursor::new(small.as_slice())) + .await + .expect("the sub-threshold put publishes"); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (1, 0), + "the kv tier holds the sub-threshold entry — the large tier was \ + never consulted on the write, never reached on the read" + ); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the kv tier served"); + assert_eq!(len, small.len() as u64); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + small + ); + + drop(pin); + + // Both-tier miss shapes for a key that never existed at all, per + // op, over the fresh both-tier store. + let absent = b"never put here".to_vec(); + let absent = Key::derive(&absent, absent.len() as u64); + assert!( + store + .get(&absent.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "never-put: get is Ok(None)" + ); + assert_eq!( + store.stat(&absent.as_bytes()).await.expect("stat succeeds"), + None, + "never-put: stat is Ok(None)" + ); + assert!( + !store.has(&absent.as_bytes()).await.unwrap(), + "never-put: has is false" + ); + assert_eq!( + store + .read_range(&absent.as_bytes(), 0, 4) + .await + .expect_err("never-put read_range errs"), + Error::Missing, + "never-put: read_range is Err(Missing)" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Virgin-store row: **reads are no-ops on a fresh schema** — an +/// empty `list`, absent reads per op (`None` on get/stat, `false` on +/// has, `Err(Missing)` on read_range), and nothing on the fresh +/// medium itself (zero pool rows, zero window rows, zero committed +/// files). The constructor always ensures the schema (the engine's +/// recorded posture — `ensure_gc_schema` runs at construction, both +/// GC tables in the fresh schema), so the facade's fresh-schema form +/// of the tolerance is this row's assertions. +/// +/// Contract stamp: store-api.md §Error model (virgin-store semantics: +/// reads on a fresh store see absent/empty, never errors); ADR-012 +/// §6.5 ruling 5 (the per-op miss shapes); ADR-007 §5 (the postgres +/// engine — the schema-ensuring construction posture and the +/// exact-count gate's host). +pub async fn virgin_store_reads_are_no_ops_on_a_fresh_schema(factory: &DualTierPostgresFactory) { + let (store, site) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated store"); + assert_eq!( + store.mode(), + StoreMode::DualTier, + "the column's factory is the dispatch-coverage shape" + ); + + let absent = b"nothing was ever put".to_vec(); + let absent = Key::derive(&absent, absent.len() as u64); + + assert_eq!( + store.list().await.expect("list succeeds"), + Vec::::new(), + "the virgin pool lists empty" + ); + assert!( + store + .get(&absent.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "absent get is Ok(None)" + ); + assert_eq!( + store.stat(&absent.as_bytes()).await.expect("stat succeeds"), + None, + "absent stat is Ok(None)" + ); + assert!( + !store.has(&absent.as_bytes()).await.unwrap(), + "absent has is false" + ); + assert_eq!( + store + .read_range(&absent.as_bytes(), 0, 4) + .await + .expect_err("absent read_range errs"), + Error::Missing, + "absent read_range is Err(Missing)" + ); + assert_eq!( + ( + pg_pool_rows(&site).await, + pg_window_rows(&site).await, + committed_file_count(site.root()) + ), + (0, 0, 0), + "the fresh medium holds nothing (pool, durable window, and \ + committed tree all empty)" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +// ------------------------------------------------------------------ +// The runners — the column's tests: the form-independent reference +// rows mirrored verbatim against the dual-tier factory (INV-1, INV-6, +// and the two no-op abort shapes), the durable-schedule rows above, +// the kv-only rejection row against the kv-only factory (unchanged — +// threshold-agnostic by discovery), and the factory contract's +// structural shapes. Every runner gates behind the reachability +// probe — the SKIP posture: server-less runs skip cleanly, a skip is +// not a failure; acceptance is the green run against the harness +// server. +// ------------------------------------------------------------------ + +#[tokio::test] +async fn postgres_inv1_round_trip_puts_and_gets_are_byte_identical() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + rows::round_trip_puts_and_gets_are_byte_identical(&factory).await; +} + +#[tokio::test] +async fn postgres_inv2_dedup_identical_content_is_one_pool_entry() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + dedup_identical_content_is_one_pool_entry(&factory).await; +} + +#[tokio::test] +async fn postgres_inv3_stage_hygiene_zero_residue_one_entry_on_success() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(&factory).await; +} + +#[tokio::test] +async fn postgres_inv4_two_sweep_exact_counts_over_the_full_protection_set() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + two_sweep_exact_counts_over_the_full_protection_set(&factory).await; +} + +#[tokio::test] +async fn postgres_inv4_a_put_landing_during_the_window_survives_the_executing_sweep() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + a_put_landing_during_the_window_survives_the_executing_sweep(&factory).await; +} + +#[tokio::test] +async fn postgres_inv4_sweep_without_liveness_sources_aborts_and_deletes_nothing() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + rows::sweep_without_liveness_sources_aborts_and_deletes_nothing(&factory).await; +} + +#[tokio::test] +async fn postgres_inv4_sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + rows::sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing(&factory).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn postgres_inv4_the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only(&factory).await; +} + +#[tokio::test] +async fn postgres_inv5_list_correctness_through_durable_schedule_outcomes() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + list_correctness_is_observable_through_durable_schedule_outcomes(&factory).await; +} + +#[tokio::test] +async fn postgres_inv6_stat_reports_len_and_algorithm_without_fetching_content() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + rows::stat_reports_len_and_algorithm_without_fetching_content(&factory).await; +} + +#[tokio::test] +async fn postgres_virgin_store_reads_are_no_ops_on_a_fresh_schema() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + virgin_store_reads_are_no_ops_on_a_fresh_schema(&factory).await; +} + +#[tokio::test] +async fn postgres_kv_only_mode_rejects_over_threshold_puts_at_put_time() { + if !harness_ready().await { + return; + } + let factory = KvOnlyPostgresFactory::new(KV_ONLY_THRESHOLD); + rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&factory).await; +} + +#[tokio::test] +async fn dispatch_routing_is_a_pure_function_of_length_across_put_paths_and_re_puts() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + routing_is_a_pure_function_of_length_across_put_paths_and_re_puts(&factory).await; +} + +#[tokio::test] +async fn dispatch_threshold_boundary_lengths_route_deterministically() { + if !harness_ready().await { + return; + } + let tuned = DualTierPostgresFactory::new(TUNED_THRESHOLD); + threshold_boundary_lengths_route_deterministically(&tuned).await; + let default = DualTierPostgresFactory::new(DEFAULT_DISPATCH_THRESHOLD); + threshold_boundary_lengths_route_deterministically(&default).await; +} + +#[tokio::test] +async fn dispatch_kv_tier_misses_fall_through_to_the_large_tier_per_op() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + kv_tier_misses_fall_through_to_the_large_tier_per_op(&factory).await; +} + +// ------------------------------------------------------------------ +// The factory contract's structural shapes (the mem and sqlite +// columns' pattern, carried to the first SKIP-posture factory): +// fresh isolated opens (a fresh schema per open, nothing shared — two +// opens never observe one another on the shared server) and teardown +// against a panicking row (the schemas really dropped, idempotent). +// Not conformance rows — no backlog entry — the factory contract's +// own wiring-level assertions. +// ------------------------------------------------------------------ + +/// Every `open()` mints a fresh, isolated schema: no shared state +/// between islands, observable on the facade *and* on the fresh media +/// (each open's schema and local root are private and start empty). +#[tokio::test] +async fn factory_open_vends_fresh_isolated_postgres_stores() { + if !harness_ready().await { + return; + } + let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD); + + let (a, site_a) = factory.open_tracked().await.expect("open vends a store"); + let (b, site_b) = factory.open_tracked().await.expect("open vends another"); + assert_ne!( + site_a.schema(), + site_b.schema(), + "two opens never share a schema" + ); + assert_ne!( + site_a.root(), + site_b.root(), + "two opens never share a local root" + ); + + let content_a = patterned(2 * TUNED_THRESHOLD as usize + 5, 167); + let content_b = patterned(TUNED_THRESHOLD as usize - 5, 191); + let (key_a, _) = a + .put( + Some(content_a.len() as u64), + Cursor::new(content_a.as_slice()), + ) + .await + .expect("island a's put publishes"); + let (key_b, _) = b + .put(None, Cursor::new(content_b.as_slice())) + .await + .expect("island b's put publishes"); + + assert!(a.has(&key_a.as_bytes()).await.unwrap()); + assert!( + !a.has(&key_b.as_bytes()).await.unwrap(), + "island a sees nothing of island b" + ); + assert!(b.has(&key_b.as_bytes()).await.unwrap()); + assert!( + !b.has(&key_a.as_bytes()).await.unwrap(), + "island b sees nothing of island a" + ); + assert_eq!( + ( + pg_pool_rows(&site_a).await, + committed_file_count(site_a.root()) + ), + (0, 1), + "island a's media hold exactly its own entry (a's content routes large)" + ); + assert_eq!( + ( + pg_pool_rows(&site_b).await, + committed_file_count(site_b.root()) + ), + (1, 0), + "island b's media hold exactly its own entry (b's content routes kv)" + ); + + let (c, site_c) = factory.open_tracked().await.expect("a third vends fresh"); + assert_eq!( + ( + pg_pool_rows(&site_c).await, + committed_file_count(site_c.root()) + ), + (0, 0), + "a third open's media are virgin" + ); + assert!( + !c.has(&key_a.as_bytes()).await.unwrap(), + "a third open shares nothing with the first" + ); + assert!( + !c.has(&key_b.as_bytes()).await.unwrap(), + "a third open shares nothing with the second" + ); + + let schemas = factory.tracked_schemas(); + assert_eq!(schemas.len(), 3, "one minted schema per open"); + assert_eq!( + schemas.len(), + schemas + .iter() + .collect::>() + .len(), + "no two opens share a schema name" + ); + + factory + .teardown() + .await + .expect("teardown drops the schemas and removes the roots"); + factory.teardown().await.expect("teardown repeats cleanly"); + for schema in &schemas { + assert!( + !schema_exists(factory.dsn(), schema).await, + "teardown dropped the schema: {schema}" + ); + } +} + +/// A failing assertion panics the row before teardown runs — and +/// teardown must make sense over the abandoned instance anyway: it is +/// idempotent against the panicked row's path, the minted schemas are +/// really gone server-side (`DROP SCHEMA … CASCADE` over exactly the +/// registry's list; `CASCADE` removes the dependents while the +/// abandoned store's pooled connections fail closed on next use), and +/// the next open mints a fresh schema on a fresh (empty) medium +/// (README §The factory contract). +#[tokio::test] +async fn 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(DualTierPostgresFactory::new(TUNED_THRESHOLD)); + + let task = { + let factory = Arc::clone(&factory); + tokio::spawn(async move { + let (store, _) = factory.open_tracked().await.expect("open vends the store"); + 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 instance"); + factory.teardown().await.expect("teardown is idempotent"); + for schema in &schemas { + assert!( + !schema_exists(factory.dsn(), schema).await, + "teardown dropped the panicked row's schema: {schema}" + ); + } + + let (store, site) = factory + .open_tracked() + .await + .expect("the next open still vends a store"); + assert!( + !store.has(&abandoned.as_bytes()).await.unwrap(), + "no residue from the panicked row — isolation is factory-owned" + ); + assert_eq!( + (pg_pool_rows(&site).await, committed_file_count(site.root())), + (0, 0), + "and its media carry nothing of the abandoned run" + ); + factory + .teardown() + .await + .expect("teardown drops the last schema"); +}