//! 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; }