From 76be7b027308e58c2ff250b2eacb69b5c1168439 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 15:15:27 +0000 Subject: [PATCH] =?UTF-8?q?feat(pg-lo):=20the=20joint=20contribution=20?= =?UTF-8?q?=E2=80=94=20staged-put=20mechanics=20into=20the=20store-held=20?= =?UTF-8?q?tx=20(wave=206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wave-5 deferred design point, landed: under the consolidation posture (kv = postgres + large = pg-lo, same pg instance), a large entry's content reaches the store-held joint transaction the pin row commits in (ADR-012 §4's one-tx ruling). - The contribution's API home (the named design point) — decided: the joint staging-cursor variant. `PgLoBackend::put_joint_staged(&mut PgJointTx)` hands `PgLoJointStaging<'_>`, the staged put with inverted ownership: it borrows the held checkout, owns no checkout, opens no tx, never COMMITs, never ROLLBACKs (the driver owns termination, amendment A1) — no Drop, by design. The deciding facts: the large tier's put contract is the streaming cursor, the fleet put driver routes to the joint form as the non-fleet path routes to the staged form, and the unknown-length derivation pass reads back through the cursor's `ReadCursor` seam — none servable by a whole-value method. - The mechanics are the landed bodies re-homed: the 512 KiB `lowrite` knee (LOWRITE_CHUNK aggregation of the store core's 64 KiB windows), read-your-own-writes `lo_get` windows, the publish-time CAS (the key resolves at publish — the staged put's named design point), the companion-row insert, transactional LO creation (C6: the LO lives and dies with the store-held tx). The re-put form publishes an empty contribution — the staged LO unlinked inside the store-held tx, zero residue; a concurrent CAS miss surfaces the companion key's unique violation (the landed puts' posture). - Enabling affordance: `PgJointTx::client()` — the crate-internal statement surface every joint contribution (and the store-fleet-put driver's A1 termination design) rides through; deliberately not an abort verb. - Crate-internal seam: the public trait surface untouched, nothing of `EngineFleetState` on pg-lo (ADR-010 §2's correction), the engine-level held-tx staging cursor and the non-fleet put path byte-unchanged. Gated `feature = "postgres"`; both pg-lo pure builds compile. - Contract tests (`src/pg_lo.rs` mod joint, both engines on one dedicated database): one-tx publish (kv row + LO + companion row + pin row, byte-exact round-trips after `pin_commit`); forced failure mid-contribution → driver ROLLBACK → zero residue (no LO pages, no companion row, no pin row, no kv row); the CAS re-put form (empty contribution, upsert renewal); the 512 KiB knee (multi-chunk). Verification: `cargo test` (default) green; `--features "pg-lo,postgres"` green server-less (skipping) and against the harness server (pglo-poc, :15432 — the four joint tests + full pg_lo + conformance suites live); `--all-features` green; clippy `-D warnings` green (default / postgres / pg-lo+postgres / all-features / the two pg-lo pure builds); `cargo fmt --check` green; `cargo doc --no-deps` clean. The store-core driver that calls the contribution on the fleet put path is store-fleet-put. --- src/gc.rs | 9 +- src/pg_lo.rs | 611 +++++++++++++++++++++++++++++++++++- src/postgres.rs | 20 +- tasks/engine-pg-lo-joint.md | 100 +++++- 4 files changed, 730 insertions(+), 10 deletions(-) diff --git a/src/gc.rs b/src/gc.rs index 2b740cb..bc374b7 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -614,10 +614,13 @@ pub(crate) trait EngineFleetState: EngineWindowState { /// The entry half the engine contributes into the held tx: its put /// mechanics (row shape, CAS clause) under the same first-writer-wins /// semantics as the public `put` (ADR-010 §2). For the pg-lo - /// consolidation posture this clause is the kv host's row write — + /// consolidation posture this clause is the kv host's row write; /// the large entry's content reaches the same tx through that - /// engine's staged `lowrite` bodies (an engine-task contribution - /// against this `JointTx` type, per ADR-012 §4's one-tx ruling). + /// engine's joint contribution — the landed `put_joint_staged` + /// handout in `crate::pg_lo`, whose joint staging cursor writes + /// `lo_create`/chunked `lowrite` + the companion-row insert into + /// this tx type (an engine-task contribution against this + /// `JointTx` type, per ADR-012 §4's one-tx ruling). async fn joint_put( &self, tx: &mut Self::JointTx, diff --git a/src/pg_lo.rs b/src/pg_lo.rs index 3d54609..6f6c172 100644 --- a/src/pg_lo.rs +++ b/src/pg_lo.rs @@ -88,9 +88,48 @@ //! cost, bounded by the pool size. `#[cfg(test)] fail_commit` //! injection stands in for the media failure (the //! mem/sqlite/postgres pattern). **Scope boundary:** the *fleet* joint -//! form (wave 6's store-held tx, ADR-012 §4/§5's ruling that pg-lo -//! contributes LO writes + companion-row insert into it) is a wave-6 -//! design point; this is the engine-level staging cursor only. +//! form below is the store-held tx edition of this staged put; the +//! engine-level staging cursor stays the non-fleet put path's own. +//! +//! **The fleet joint form — the store-held tx (ADR-012 §4's one-tx +//! ruling, wave 6; feature `postgres`).** Under the consolidation +//! posture (kv = postgres + large = pg-lo, same pg instance), a large +//! entry's content reaches the *same store-held transaction* the pin +//! row commits in: `put_joint_staged` hands a `PgLoJointStaging` +//! cursor against the held `PgJointTx` — the postgres engine's held +//! checkout, wave 4's recorded shape — and `pin_commit` (the store's +//! terminal step) publishes entry + pin row in one step over the +//! shared instance. **Named design point — the contribution's API +//! home (decided):** the *joint staging-cursor variant*, not a +//! whole-value method taking `&mut PgJointTx` — the large tier's put +//! contract is the streaming staging cursor, the fleet put driver +//! routes to the joint form exactly as the non-fleet path routes to +//! the staged form (stage-while-hashing, the key resolving at stream +//! end), and the unknown-length flow's derivation pass re-reads the +//! staged artifact through the cursor's read seam — none of which a +//! whole-value method can serve. The cursor is the staged put above +//! with inverted ownership: it owns no checkout, opens no +//! transaction, never COMMITs, never ROLLBACKs (the store core's +//! driver owns termination on every path, amendment A1), and borrows +//! the held checkout for every statement; the staged-put mechanics +//! are the landed bodies re-homed — the 512 KiB `lowrite` knee, the +//! CAS existence check, the companion-row insert, transactional LO +//! creation (the LO lives and dies with the store-held tx, C6's +//! structural zero-orphan story), read-your-own-writes staging reads. +//! **The joint CAS form:** the key resolves at publish, so +//! first-writer-wins lands there — a re-put publishes an **empty +//! contribution** (the staged LO unlinked *inside the store-held tx*: +//! the committed tx carries nothing of the losing put, zero residue, +//! and the driver's `pin_commit` upsert-renews the pin row), while a +//! concurrent CAS miss surfaces the companion key's unique violation +//! (the landed puts' posture). The store's one-tx ruling is what the +//! consolidation admits: the same held tx also carries the kv host's +//! `joint_put` row write for entries that land on that tier, one tx +//! over one instance. The engine-level held-tx staging cursor above +//! is unchanged (the non-fleet put path keeps its own tx). Nothing of +//! `EngineFleetState` exists on this engine (ADR-010 §2's correction +//! of ADR-009's ownership statement): the contribution is +//! crate-internal mechanics the store core drives. //! //! **Delete = one transaction** (`lo_unlink` + companion-row delete) //! with the rows-affected discipline (`Err(Error::Missing)` when @@ -203,6 +242,8 @@ use crate::pg_common::{ build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64, time_to_nanos, PgOpts, DEFAULT_SCHEMA, }; +#[cfg(feature = "postgres")] +use crate::postgres::PgJointTx; use crate::Error; /// The `lowrite` statement chunk — POC #7 C2's measured knee (512 KiB @@ -422,6 +463,41 @@ impl PgLoBackend { } } +/// The fleet joint form's handout (feature `postgres`, ADR-012 §4's +/// one-tx ruling): hand a [`PgLoJointStaging`] cursor against the +/// store-held joint tx — the consolidation posture's one shared +/// instance is what makes the type shareable. `lo_create` + +/// `lo_open(INV_WRITE)` run inside the held checkout: the LO is born +/// inside the store's transaction, and every failure or the driver's +/// termination takes the pages with it (C6's structural zero-orphan +/// story). The cursor owns no checkout and no transaction — the store +/// core holds both from its `begin_joint_tx` to its `pin_commit` (the +/// ownership rule, ADR-010 §2; ADR-012 §4's one-tx ruling) — so the +/// handout opens nothing and terminates nothing: a handout error +/// returns with the held tx exactly as it arrived (open, uncommitted) +/// for the driver to finish. +#[cfg(feature = "postgres")] +impl PgLoBackend { + /// Hand out the joint staging cursor against the held joint tx. + pub(crate) async fn put_joint_staged<'a>( + &self, + tx: &'a mut PgJointTx, + ) -> Result, Error> { + let loid = lo_create(tx.client()).await?; + let fd = lo_open(tx.client(), loid, INV_WRITE).await?; + Ok(PgLoJointStaging { + tx, + loid, + fd, + table: self.table.clone(), + flushed_len: 0, + pending: Vec::new(), + #[cfg(test)] + fail_commit: self.fail_commit, + }) + } +} + /// The engine-owned bootstrap DDL — idempotent, no migration machinery. /// The C1 companion-table shape; column names quote-follow the alkstore /// hazard-list discipline (`"key"` quoted; `loid`/`size`/`committed_at` @@ -889,6 +965,173 @@ impl Drop for PgLoStaging { } } +/// The joint staging cursor — the staged put re-homed onto the +/// store-held joint tx (feature `postgres`, ADR-012 §4's one-tx +/// ruling; the named design point recorded in the module doc: the +/// joint staging-cursor variant, the large tier's put contract carried +/// into the consolidation posture). It borrows the held checkout — +/// everything it issues (`lowrite` statements, `lo_get` windows, the +/// CAS probe, the companion insert or the re-put `lo_unlink`) rides +/// the store's transaction; it holds no checkout of its own, opens +/// nothing, and terminates nothing: its error paths return the tx the +/// way they found it (open, uncommitted) for the driver to finish +/// (amendment A1's catch-propagate-terminate — the contribution never +/// commits, never rolls back the store's tx unilaterally). Dropping +/// the cursor without [`PgLoJointStaging::publish`] is likewise a +/// no-op on the tx — the driver owns termination on every path, and +/// the dropped cursor's staged pages die with that termination. +#[cfg(feature = "postgres")] +pub(crate) struct PgLoJointStaging<'a> { + /// The store-held joint tx this contribution rides (borrowed, never + /// owned — the ownership rule). + tx: &'a mut PgJointTx, + loid: u32, + fd: i32, + table: String, + /// Bytes flushed to the LO via `lowrite` — the read seam's LO part + /// boundary; the aggregation buffer holds everything above it. + flushed_len: u64, + pending: Vec, + /// Test-only mid-contribution failure injection (after the tail + /// flush, before the CAS probe), standing in for the media failure; + /// mirrors the engine-level knob so the joint form's failure + /// discipline is exercised against the real engine. + #[cfg(test)] + fail_commit: bool, +} + +#[cfg(feature = "postgres")] +impl PgLoJointStaging<'_> { + /// Flush full [`LOWRITE_CHUNK`] blocks out of the aggregation buffer + /// (the store core's 64 KiB put-loop windows aggregate here into the + /// C2 knee's statement size — the engine-level staged put's shape). + /// On failure the buffer is intact and `flushed_len` unclaimed — + /// the tx stays as the contribution found it. + async fn flush_blocks(&mut self) -> Result<(), Error> { + while self.pending.len() >= LOWRITE_CHUNK { + let block = self.pending[..LOWRITE_CHUNK].to_vec(); + lowrite_all(self.tx.client(), self.fd, &block).await?; + self.pending.drain(..LOWRITE_CHUNK); + self.flushed_len += LOWRITE_CHUNK as u64; + } + Ok(()) + } + + /// Flush the aggregation tail (< [`LOWRITE_CHUNK`]) — publish's + /// first mechanic, over the held checkout. + async fn flush_tail(&mut self) -> Result<(), Error> { + if self.pending.is_empty() { + return Ok(()); + } + let tail = std::mem::take(&mut self.pending); + lowrite_all(self.tx.client(), self.fd, &tail).await?; + self.flushed_len += tail.len() as u64; + Ok(()) + } + + /// Append a chunk to the contribution — the staged put's + /// aggregation shape (the store core's put loop hands 64 KiB + /// windows; they aggregate into [`LOWRITE_CHUNK`] `lowrite` + /// statements inside the held tx). + pub(crate) async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + if chunk.is_empty() { + return Ok(()); + } + self.pending.extend_from_slice(chunk); + self.flush_blocks().await + } + + /// Publish the contribution into the held tx: the tail's + /// aggregation flush, then the CAS existence check against the + /// companion table (the key resolves at publish — the staged put's + /// named design point: the derivation completes at stream end), and + /// either + /// + /// - the companion-row insert — `Ok(true)`, the fresh form — or + /// - the **re-put form**: the staged LO unlinked *inside the + /// store-held tx*, `Ok(false)` — an **empty contribution**: the + /// committed tx carries nothing of the losing put (zero residue), + /// and the driver's `pin_commit` then upsert-renews the pin row. + /// + /// A concurrent CAS miss surfaces the companion key's unique + /// violation as an error (the landed puts' posture: the check is + /// the fast path, the primary key is the hard dedup guarantee). + /// Every failure path — including the injected one — returns with + /// the tx exactly as the contribution found it: open, uncommitted, + /// un-rolled-back (the driver owns termination; no partial LO state + /// outlives its rollback — transactional LO creation, C6). + pub(crate) async fn publish(mut self, key: &[u8]) -> Result { + #[cfg(test)] + if self.fail_commit { + return Err(Error::Io( + "engine joint commit failure (pg-lo, injected)".to_owned(), + )); + } + self.flush_tail().await?; + let client = self.tx.client(); + let exists_sql = format!("SELECT 1 FROM {} WHERE key = $1", self.table); + let exists_stmt = client.prepare_cached(&exists_sql).await.map_err(err)?; + let present = client + .query_opt(&exists_stmt, &[&key]) + .await + .map_err(pg_err)? + .is_some(); + if present { + let unlink_stmt = client + .prepare_cached("SELECT lo_unlink($1)") + .await + .map_err(err)?; + client + .query_one(&unlink_stmt, &[&self.loid]) + .await + .map_err(pg_err)?; + return Ok(false); + } + let size = self.flushed_len as i64; + let committed_at = time_to_nanos(SystemTime::now()); + let insert_sql = format!( + "INSERT INTO {} (key, loid, size, committed_at) VALUES ($1, $2, $3, $4)", + self.table + ); + let insert_stmt = client.prepare_cached(&insert_sql).await.map_err(err)?; + client + .execute(&insert_stmt, &[&key, &self.loid, &size, &committed_at]) + .await + .map_err(pg_err)?; + Ok(true) + } +} + +#[cfg(feature = "postgres")] +impl ReadCursor for PgLoJointStaging<'_> { + fn len(&self) -> u64 { + self.flushed_len + self.pending.len() as u64 + } + + /// The staged artifact's read seam, inside the store-held tx + /// (read-your-own-writes: the LO's pages are visible to `lo_get` + /// from the tx that wrote them; reads above the flushed boundary + /// serve from the aggregation buffer) — the derivation pass's + /// instrument, the staged put's shape carried. + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.len() as usize); + if start == end { + return Ok(Vec::new()); + } + let client = self.tx.client(); + let lo_end = self.flushed_len.min(end as u64) as usize; + let mut out = Vec::with_capacity(end - start); + if (start as u64) < self.flushed_len { + out.extend(read_windows(client, self.loid, start as u64, lo_end as u64).await?); + } + if end > lo_end { + let pending_start = (start.max(lo_end)) - lo_end; + out.extend_from_slice(&self.pending[pending_start..(end - lo_end)]); + } + Ok(out) + } +} + impl Backend for PgLoBackend { type Cursor = PgLoCursor; type Staging = PgLoStaging; @@ -2464,4 +2707,366 @@ mod tests { "the shared harness database holds zero LO pages (per-test-database isolation)" ); } + + /// The fleet joint contribution's contract tests (feature + /// `postgres`; ADR-012 §4's one-tx ruling): the consolidation + /// posture — kv = postgres + large = pg-lo on one shared instance, + /// both engines opened through the default engine schema on the + /// same dedicated database — with the pg-lo engine's staged-put + /// mechanics riding the store-held `PgJointTx` beside the kv + /// host's `joint_put` row write, and `pin_commit` as the joint + /// terminal step. Failure forms play the store-core driver: the + /// contribution's error paths leave the held tx terminable, and + /// the driver-side rollback here is amendment A1's + /// catch-propagate-terminate exercised at the instrumentation + /// level the driver task designs on. + #[cfg(feature = "postgres")] + mod joint { + use super::*; + use crate::gc::{EngineFleetState, EngineWindowState, PinRow}; + use crate::postgres::PostgresBackend; + use crate::store::DEFAULT_PIN_TTL; + use crate::PinState; + use std::time::Duration; + + /// The consolidation shape's harness: both engines on one + /// dedicated database through the default engine schema (the + /// consolidation posture's shared-schema shape), the kv host's + /// GC schema ensured (`gc_pins` — the pin rows' home; fleet + /// pins live on the kv host, ADR-012 §5), one janitor holding + /// the whole database for both engines. + struct JointHarness { + lo: PgLoBackend, + kv: PostgresBackend, + _janitor: DatabaseJanitor, + } + + async fn harness_joint(tag: &str) -> Option { + let admin = harness_admin_dsn()?; + let db = fresh_db(tag); + create_db(&admin, &db).await; + let dsn = dsn_for_db(&db); + let lo = PgLoBackend::new(&dsn, PgOpts::default()) + .await + .expect("open the pg-lo engine"); + let kv = PostgresBackend::new(&dsn, PgOpts::default()) + .await + .expect("open the postgres engine"); + kv.ensure_gc_schema().await.expect("kv host gc schema"); + Some(JointHarness { + lo, + kv, + _janitor: DatabaseJanitor::new(&admin, &db), + }) + } + + fn pin_row(key: &[u8], owner: &[u8], expiry: SystemTime) -> PinRow { + PinRow { + key: key.to_vec(), + owner: owner.to_vec(), + expiry, + } + } + + /// The gc_pins rows as the engine stores them — the raw + /// encoding, read back through the shared instance (the probes + /// ride the pg-lo engine's pool over the kv host's table: one + /// instance, one schema). + async fn pin_rows(h: &JointHarness) -> Vec<(Vec, Vec, i64)> { + let client = h.lo.pool.get().await.expect("checkout"); + let sql = format!( + "SELECT key, owner, expiry FROM {} ORDER BY key, owner", + qualified_table(DEFAULT_SCHEMA, "gc_pins") + ); + let rows = client.query(sql.as_str(), &[]).await.expect("pin rows"); + rows.iter() + .map(|row| { + ( + row.get::<_, Vec>(0), + row.get::<_, Vec>(1), + row.get::<_, i64>(2), + ) + }) + .collect() + } + + /// The kv-tier entry's round-trip — byte-exact through the + /// postgres engine's cursor (the large-tier twin of the + /// parent's `stored_value`). + async fn kv_stored_value(h: &JointHarness, key: &[u8]) -> Option> { + let cursor = h.kv.get(key).await.expect("kv get succeeds")?; + Some(cursor.read_at(0, cursor.len()).await.expect("kv read")) + } + + /// The one-tx publish (the consolidation put's engine-level + /// form, store-fleet-put's driver shape): the store core opens + /// the joint tx on the kv host, the kv tier contributes its row + /// write, the large tier contributes LO + companion row through + /// the joint staging cursor, and `pin_commit` publishes entry + + /// pin row in one step — byte-exact round-trips after it, pin + /// row landed with the entry (pin-before-publish, one tx), and + /// exactly one LO with exactly its page count. + #[tokio::test] + async fn joint_one_tx_publishes_kv_entry_lo_entry_and_pin_row() { + let Some(h) = harness_joint("joint-one-tx").await else { + return; + }; + let lo_key = b"\x01joint-lo-key".to_vec(); + let kv_key = b"\x01joint-kv-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let mut tx = h.kv.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&h.kv, &mut tx, &kv_key, CONTENT) + .await + .unwrap(); + let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap(); + staged.append(CONTENT).await.unwrap(); + assert_eq!(staged.len() as usize, CONTENT.len(), "staged length"); + assert!( + staged.publish(&lo_key).await.unwrap(), + "the fresh contribution publishes" + ); + let expiry = SystemTime::now() + DEFAULT_PIN_TTL; + h.kv.pin_commit(tx, &[pin_row(&lo_key, &owner, expiry)]) + .await + .unwrap(); + + let cursor = + h.lo.get(&lo_key) + .await + .unwrap() + .expect("the LO entry published"); + assert_eq!(cursor.len(), CONTENT.len() as u64); + let (bytes, digest) = cursor.read_range(0, cursor.len()).await.unwrap(); + assert_eq!(bytes, CONTENT, "byte-exact LO round-trip after pin_commit"); + assert_eq!(digest, crate::backend::slice_digest(CONTENT)); + + assert!(h.kv.has(&kv_key).await.unwrap(), "the kv entry landed"); + assert_eq!( + kv_stored_value(&h, &kv_key).await.as_deref(), + Some(CONTENT), + "byte-exact kv round-trip after pin_commit" + ); + + assert_eq!( + EngineFleetState::pin_check(&h.kv, &lo_key).await.unwrap(), + PinState::Protected, + "the pin row landed with the entry in the one tx" + ); + assert_eq!( + pin_rows(&h).await, + vec![(lo_key, owner, time_to_nanos(expiry))], + "exactly the joint tx's pin row" + ); + assert_eq!( + catalog_loids(&h.lo).await.len(), + 1, + "exactly one LO for the large entry" + ); + assert_eq!( + catalog_rows(&h.lo).await, + pages_for(CONTENT.len()), + "exactly the entry's LO pages" + ); + } + + /// The joint form's failure discipline (the stage-hygiene + /// invariant's tx form, the wave-4 joint-atomicity test's + /// pg-lo edition): a forced failure mid-contribution — after + /// the tail flush, so LO pages are staged into the held tx — + /// returns with the tx terminable (never committed, never + /// rolled back by the contribution); the driver-side + /// termination (amendment A1's catch-propagate-terminate, the + /// ROLLBACK the driver owns on failure paths) aborts the whole + /// tx with zero residue: no LO pages, no companion row, no pin + /// row, no kv row. + #[tokio::test] + async fn joint_contribution_failure_terminates_to_zero_residue() { + let Some(h) = harness_joint("joint-fail").await else { + return; + }; + let failing_lo = PgLoBackend { + pool: h.lo.pool.clone(), + table: h.lo.table.clone(), + window_table: h.lo.window_table.clone(), + schema: h.lo.schema.clone(), + #[cfg(test)] + fail_commit: true, + }; + let lo_key = b"\x01joint-failure-key".to_vec(); + let kv_key = b"\x01joint-failure-kv-key".to_vec(); + + let mut tx = h.kv.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&h.kv, &mut tx, &kv_key, CONTENT) + .await + .unwrap(); + let mut staged = failing_lo + .put_joint_staged(&mut tx) + .await + .expect("the handout holds the held checkout"); + staged + .append(&pattern(2 * LOWRITE_CHUNK + 128)) + .await + .unwrap(); + let err = staged.publish(&lo_key).await.unwrap_err(); + assert!( + matches!(err, Error::Io(_)), + "the injected media failure surfaces as Io" + ); + + tx.client().batch_execute("ROLLBACK").await.unwrap(); + + assert!( + !h.lo.has(&lo_key).await.unwrap(), + "the aborted tx published no LO entry" + ); + assert!( + !h.kv.has(&kv_key).await.unwrap(), + "the aborted tx published no kv row" + ); + assert!( + h.lo.list().await.unwrap().is_empty(), + "zero residue in the companion table" + ); + assert_eq!( + catalog_rows(&h.lo).await, + 0, + "zero residue in the LO catalog — the staged pages died with the tx" + ); + assert_eq!( + EngineFleetState::pin_check(&h.kv, &lo_key).await.unwrap(), + PinState::Unprotected, + "no pin row survived" + ); + assert_eq!(pin_rows(&h).await, vec![], "zero residue in gc_pins"); + assert!( + h.kv.list().await.unwrap().is_empty(), + "zero residue in the blobs table itself" + ); + } + + /// The CAS re-put form: a second contribution of the same key + /// publishes an **empty contribution** (`Ok(false)` — the + /// staged LO unlinked inside the store-held tx), the tx commits + /// the pin row alone, and the pin row upsert-renews — with the + /// losing contribution's staged LO leaving zero residue. + #[tokio::test] + async fn joint_contribution_reput_commits_empty_and_upsert_renews() { + let Some(h) = harness_joint("joint-reput").await else { + return; + }; + let key = b"\x01joint-reput-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let mut tx = h.kv.begin_joint_tx().await.unwrap(); + let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap(); + staged.append(CONTENT).await.unwrap(); + assert!(staged.publish(&key).await.unwrap(), "fresh contribution"); + let expiry = SystemTime::now() + DEFAULT_PIN_TTL; + h.kv.pin_commit(tx, &[pin_row(&key, &owner, expiry)]) + .await + .unwrap(); + + let renewed = SystemTime::now() + DEFAULT_PIN_TTL + Duration::from_secs(7); + let mut tx = h.kv.begin_joint_tx().await.unwrap(); + let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap(); + staged.append(b"overwritten-by-joint-reput").await.unwrap(); + assert!( + !staged.publish(&key).await.unwrap(), + "the re-put publishes an empty contribution, never an overwrite" + ); + h.kv.pin_commit(tx, &[pin_row(&key, &owner, renewed)]) + .await + .unwrap(); + + assert_eq!( + stored_value(&h.lo, &key).await.as_deref(), + Some(CONTENT), + "first-writer-wins: the re-put does not overwrite" + ); + assert_eq!( + pin_rows(&h).await, + vec![(key, owner, time_to_nanos(renewed))], + "the re-put's pin row upsert renews the same instance's row" + ); + assert_eq!( + catalog_rows(&h.lo).await, + pages_for(CONTENT.len()), + "the losing contribution's staged LO left zero residue (unlinked in-tx)" + ); + assert_eq!( + catalog_loids(&h.lo).await.len(), + 1, + "exactly the first writer's LO remains" + ); + assert_eq!( + companion_loids_sorted(&h.lo).await.len(), + 1, + "the companion table holds exactly one row" + ); + } + + /// The 512 KiB knee rides: a multi-chunk contribution — 64 KiB + /// appends (the store core's put-loop window size) aggregating + /// into `LOWRITE_CHUNK` `lowrite` statements inside the held + /// tx — with the derivation pass reading the staged artifact + /// back through the read seam (including a read crossing the + /// flushed/pending aggregation boundary), the one-tx publish, + /// and the byte-exact committed round-trip. + #[tokio::test] + async fn joint_contribution_rides_the_512kib_knee_multichunk() { + let Some(h) = harness_joint("joint-knee").await else { + return; + }; + let content = pattern(1_500_000); + let key = b"\x01joint-knee-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let mut tx = h.kv.begin_joint_tx().await.unwrap(); + let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap(); + for chunk in content.chunks(64 * 1024) { + staged.append(chunk).await.unwrap(); + } + assert_eq!(staged.len() as usize, content.len(), "staged length"); + + let (bytes, digest) = staged.read_range(0, staged.len()).await.unwrap(); + assert_eq!(bytes, content, "staged artifact readable as a whole"); + assert_eq!(digest, crate::backend::slice_digest(&content)); + let boundary = LOWRITE_CHUNK as u64; + assert!( + staged.len() > boundary, + "the staged content crossed the first aggregation flush" + ); + let crossed = staged.read_at(boundary - 64, 256).await.unwrap(); + assert_eq!( + crossed, + content[(LOWRITE_CHUNK - 64)..(LOWRITE_CHUNK + 192)], + "a staged read crossing the flushed/pending boundary is byte-exact \ + (LO windows over committed-into-tx pages + the in-memory tail)" + ); + + assert!( + staged.publish(&key).await.unwrap(), + "the multi-chunk contribution publishes" + ); + let expiry = SystemTime::now() + DEFAULT_PIN_TTL; + h.kv.pin_commit(tx, &[pin_row(&key, &owner, expiry)]) + .await + .unwrap(); + + let cursor = + h.lo.get(&key) + .await + .unwrap() + .expect("the LO entry published"); + let (bytes, digest) = cursor.read_range(0, cursor.len()).await.unwrap(); + assert_eq!(bytes, content, "byte-exact multi-chunk round-trip"); + assert_eq!(digest, crate::backend::slice_digest(&content)); + assert_eq!( + catalog_rows(&h.lo).await, + pages_for(content.len()), + "exactly the entry's LO pages (C2's catalog math)" + ); + } + } } diff --git a/src/postgres.rs b/src/postgres.rs index 2a491fe..53ddd7e 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -767,7 +767,11 @@ impl Backend for PostgresBackend { /// resets transaction state, so the pool would hand the dirty session /// to the next checkout. The store core's joint-tx driver (wave 6) /// therefore owes pin_commit on **every** path from `begin_joint_tx` -/// onward; the recorded wave-4 review amendment names this duty. +/// onward; the recorded wave-4 review amendment names this duty. The +/// pg-lo joint contribution composes with the duty: its failure paths +/// never close the tx themselves — the contribution returns with the +/// tx open (no partial LO state that outlives the termination's +/// rollback; transactional LO creation), and the driver terminates. pub(crate) struct PgJointTx { client: Object, blobs_table: String, @@ -776,6 +780,20 @@ pub(crate) struct PgJointTx { fail_commit: bool, } +impl PgJointTx { + /// The held checkout, by reference — the statement surface every + /// joint contribution rides: the pg-lo consolidation contribution's + /// `lo_*` + companion clauses (the joint staging cursor's + /// statements) and the store-core driver's A1 termination mechanics + /// on its failure paths run through this checkout. Not an abort + /// verb: this exposes the session only — what the driver issues on + /// which path is the driver's termination design (amendment A1's + /// catch-propagate-terminate), never a seam-level commit/rollback. + pub(crate) fn client(&self) -> &Object { + &self.client + } +} + /// The window family (ADR-012 §5's postgres row): the fleet host's /// both-tables schema and the durable staged delete window. Same /// semantics as the sqlite engine's reference implementation — diff --git a/tasks/engine-pg-lo-joint.md b/tasks/engine-pg-lo-joint.md index 540c5cf..b6624b0 100644 --- a/tasks/engine-pg-lo-joint.md +++ b/tasks/engine-pg-lo-joint.md @@ -1,7 +1,7 @@ --- id: engine-pg-lo-joint name: the pg-lo joint contribution — staged-put mechanics into the store-held tx -status: pending +status: completed depends_on: [] scope: moderate risk: high @@ -138,8 +138,102 @@ the postgres host's clause — landed wave 4, unchanged here). ## Notes -> To be filled by implementation agent +> The contribution's API shape (the named design point) — **decided: the +> joint staging-cursor variant**, not a whole-value method taking +> `&mut PgJointTx`. Landed as `PgLoBackend::put_joint_staged(&self, tx: +> &mut PgJointTx) -> PgLoJointStaging<'_>` in `src/pg_lo.rs` (gated +> `feature = "postgres"` — `pg-lo` does not imply `postgres`, ADR-013 +> §4; both pure builds compile). The deciding facts: (1) the large +> tier's put contract is the streaming staging cursor +> (`StagingCursor`/`ReadCursor`), and store-fleet-put routes to the +> joint form "exactly as the non-fleet path routes to the staged form" +> — stage-while-hashing, the key resolving at stream end — so a +> whole-value method would force the driver to materialize every large +> entry, regressing the memory posture; (2) the unknown-length flow's +> derivation pass re-reads the staged artifact through the cursor's +> read seam, the engine-level `PgLoStaging` shape — a whole-value +> method cannot serve any of it; (3) the cursor is exactly "the staged +> put with inverted ownership" — the task description's own framing — +> the landed bodies re-homed, not re-written. Consequence recorded: +> the CAS existence check lands at **publish** (the key resolves +> there), the landed `PgLoStaging::publish` shape — a re-put publishes +> an **empty contribution** (the staged LO unlinked *inside the +> store-held tx*, so the committed tx carries nothing of the losing +> put; zero residue; the driver's `pin_commit` then upsert-renews the +> pin row), while a concurrent CAS miss surfaces the companion key's +> unique violation (the landed puts' posture: check = fast path, PK = +> hard guarantee). "Never `lo_create`" holds literally only in the +> engine-level whole-value put — the cursor form's handout must create +> the LO before the key exists — but the LO lives and dies with the +> store-held tx (C6, structural), so a losing put leaves zero residue +> either way (contract-tested: the re-put form's catalog stays at +> exactly the first writer's pages). +> +> **Ownership rule honored, literally:** `PgLoJointStaging` borrows the +> held checkout (`&'a mut PgJointTx`); it owns no checkout, opens no +> tx, never COMMITs, never ROLLBACKs — including on `Drop` (no Drop +> impl: the driver owns termination on every path, amendment A1). Its +> failure paths (statement failures, the injected media failure after +> the tail flush, the concurrent-PK race) return with the tx exactly +> as the contribution found it — open, uncommitted, un-rolled-back — +> composing with the driver's catch-propagate-terminate (A1) with no +> partial LO state that outlives the termination's rollback +> (transactional LO creation). The engine-level held-tx staging cursor +> and the non-fleet put path are byte-unchanged (wave-5 hygiene tests +> green untouched); nothing of `EngineFleetState` is implemented on +> pg-lo; the public trait surface untouched. +> +> **Enabling affordance (minimal, for the contribution and the next +> driver task):** `PgJointTx::client()` — a crate-internal accessor +> returning `&Object` (the field stays private), documented as the +> statement surface every joint contribution rides. The contract +> tests' driver-side ROLLBACK (the forced-failure form) and the future +> store-fleet-put driver's A1 termination design are the consumers; +> the accessor is deliberately *not* an abort verb (A1: an explicit +> abort verb would need a new ADR — the store core's driver issues its +> termination through the exposed checkout; that design decision +> remains the driver task's). +> +> **Tests** (`src/pg_lo.rs` `mod joint`, feature `postgres`-gated, the +> env-driven convention, per-test dedicated databases from the start — +> amendment A2's lesson honored; the probes ride the pg-lo engine's +> pool over the kv host's tables: one instance, one schema): the +> one-tx publish (kv host's `joint_put` row + LO content + companion +> row through the joint cursor + `pin_commit`'s pin row, byte-exact +> round-trips of both entries after commit, exactly one LO with +> exactly its page count); the forced-failure zero-residue form +> (injection after the tail flush, driver ROLLBACK, all four artifact +> families absent); the CAS re-put form (empty contribution, first +> writer wins, pin row upsert-renewed, zero losing-LO residue); the +> 512 KiB knee (64 KiB appends aggregating into multi-`LOWRITE_CHUNK` +> statements, boundary-crossing read-back, multi-chunk round-trip). ## Summary -> To be filled on completion \ No newline at end of file +The pg-lo engine's joint contribution landed: under the consolidation +posture (kv = postgres + large = pg-lo, same pg instance), a large +entry's content reaches the store-held joint transaction — +`put_joint_staged` hands the `PgLoJointStaging` staging cursor against +the postgres engine's held `PgJointTx` checkout; the cursor rides the +held tx for every statement (chunked 512 KiB `lowrite` aggregation, +read-your-own-writes `lo_get` windows for the derivation pass, the +publish-time CAS, the companion-row insert or the re-put's in-tx +`lo_unlink` empty-contribution form, transactional LO creation), and +the store's `pin_commit` — landed wave 4, unchanged — publishes entry + +pin row and commits in one step. The ownership rule is honored +structurally: the engine contributes mechanics into the tx it does not +own, never committing or rolling it back on any path (A1's discipline; +the `PgJointTx::client()` accessor is the checkout surface). The public +trait surface is untouched; nothing of `EngineFleetState` exists on +pg-lo; the engine-level held-tx staging cursor and the non-fleet put +path are byte-unchanged. Verified: `cargo test` (default) green; +`--features "pg-lo,postgres"` green server-less (skipping cleanly) and +against the harness server (pglo-poc, :15432 — live, all four joint +contract tests plus the full pg_lo and conformance suites green); +`--all-features` green; clippy `-D warnings` green (default, postgres, +pg-lo+postgres, all-features, and the pg-lo-pure builds +`--no-default-features --features "pg-lo,mem"` and +`--features "postgres,pg-lo"`); `cargo fmt --check` green; `cargo doc +--no-deps` clean. The store-core driver that calls this contribution on +the fleet put path is `store-fleet-put` (this task's scope boundary — +recorded in the module doc and here). \ No newline at end of file