feat(pg-lo): the joint contribution — staged-put mechanics into the store-held tx (wave 6)
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.
This commit is contained in:
glm-5.3-flash committed
1 parent
b6b353242b
commit
76be7b0273
4 files changed
+730
-10
No files matched your search
@@ -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,
|
||||
|
||||
+608
-3
@@ -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<PgLoJointStaging<'a>, 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<u8>,
|
||||
/// 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<bool, Error> {
|
||||
#[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<Vec<u8>, 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<JointHarness> {
|
||||
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<u8>, Vec<u8>, 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<u8>>(0),
|
||||
row.get::<_, Vec<u8>>(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<Vec<u8>> {
|
||||
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)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
+19
-1
@@ -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 —
|
||||
|
||||
@@ -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
|
||||
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).
|
||||
Reference in new issue
Block a user