diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a9bb3a94..88609d86 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,6 +152,12 @@ jobs: make PG_CONFIG=$PG_CONFIG sudo make PG_CONFIG=$PG_CONFIG install + # Separate module, never installed: catalog_ael_boundary loads it into + # the *source* cluster to reproduce VACUUM's lock lifetime deterministically + - name: Build walshadow test-only PG module + working-directory: walshadow/pgext/test + run: make PG_CONFIG=/usr/lib/postgresql/${{ matrix.pg-major }}/bin/pg_config + # Each e2e test stages a source + shadow PG cluster on disk; the # runner's root fs is small, so point TMPDIR at the larger /mnt volume. - name: Set TMPDIR to /mnt diff --git a/pgext/overlay.c b/pgext/overlay.c index 0332c4ca..fedc33cc 100644 --- a/pgext/overlay.c +++ b/pgext/overlay.c @@ -10,9 +10,16 @@ * * Never opens the target relation. The replaying transaction holds * AccessExclusiveLock on it and standby lock replay is driven by the startup - * process, so relation_open would block against recovery. Catalogs only, and - * standby lock replay records AccessExclusiveLocks alone, so AccessShareLock - * on a catalog never conflicts with the DDL in flight. + * process, so relation_open would block against recovery. Catalogs only. + * + * AccessShareLock on a catalog does not conflict with the DDL in flight, but + * that is not the only lock replay can be holding: any unrelated source + * transaction that took AccessExclusiveLock on the catalog itself — VACUUM + * truncating a bloated pg_type is the one seen in practice — leaves the + * startup process holding it until that transaction's commit record is + * replayed, and the caller is withholding exactly that record. A pinned scan + * therefore takes the lock only if it is free, and reads without it otherwise; + * see `ws_overlay_scan`. * * An invalid top xid asks the same scan for the committed view, which is what * the daemon captures at a catalog commit. Same projections, same assembly on @@ -36,6 +43,7 @@ #include "catalog/pg_namespace.h" #include "catalog/pg_type.h" #include "libpq/pqformat.h" +#include "storage/lmgr.h" #include "storage/procarray.h" #include "utils/fmgroids.h" #include "utils/fmgrprotos.h" @@ -412,12 +420,74 @@ ws_scan_emit(Relation rel, const WsCatalogPlan *plan, Oid key, systable_endscan(scan); } +/* qsort/bsearch comparator over Oid */ +static int +ws_oid_cmp(const void *a, const void *b) +{ + Oid x = *(const Oid *) a; + Oid y = *(const Oid *) b; + + return (x > y) - (x < y); +} + +/* + * One heap pass, emitting rows whose key attribute is in `sorted` (or every + * row when `nsorted` is 0). Used only when the catalog's lock was unavailable: + * `systable_beginscan` with an index would `index_open` it under + * AccessShareLock, which is the wait this path exists to avoid, so the index + * is skipped and the oid list is applied to the heap tuple instead. + * + * `scanned` counts rows that passed the oid filter, so it stays comparable to + * the indexed path. + */ +static void +ws_scan_emit_lockfree(Relation rel, const WsCatalogPlan *plan, + const Oid *sorted, int nsorted, TransactionId top, + StringInfo out, WsScanStats *stats) +{ + SysScanDesc scan; + ScanKeyData skey[1]; + int nkeys = 0; + HeapTuple tup; + TupleDesc desc = RelationGetDescr(rel); + bool scoped = nsorted > 0; + + if (plan->min_attnum != 0) + ScanKeyInit(&skey[nkeys++], Anum_pg_attribute_attnum, + BTGreaterEqualStrategyNumber, F_INT2GE, + Int16GetDatum(plan->min_attnum)); + + scan = systable_beginscan(rel, InvalidOid, false, SnapshotAny, nkeys, skey); + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + if (scoped) + { + bool isnull; + Datum key = heap_getattr(tup, plan->keyattno, desc, &isnull); + Oid keyoid = DatumGetObjectId(key); + + if (isnull) + continue; + if (bsearch(&keyoid, sorted, (size_t) nsorted, sizeof(Oid), + ws_oid_cmp) == NULL) + continue; + } + stats->scanned++; + if (!ws_tuple_visible(tup->t_data, top, scoped, stats)) + continue; + stats->emitted++; + plan->emit(out, tup, desc); + } + systable_endscan(scan); +} + void ws_overlay_scan(WsCatalog cat, TransactionId top, const Oid *oids, int noids, - StringInfo out, WsScanStats *stats) + WsScanLock lock, StringInfo out, WsScanStats *stats) { WsCatalogPlan plan; bool rel_scoped; + bool locked; Relation rel; if (!ws_catalog_plan(cat, &plan)) @@ -429,9 +499,50 @@ ws_overlay_scan(WsCatalog cat, TransactionId top, const Oid *oids, int noids, * pg_namespace and pg_type have. The lock argument comes with the list, so * losing the list loses the argument too */ rel_scoped = AttributeNumberIsValid(plan.keyattno) && noids > 0; - rel = table_open(plan.relid, AccessShareLock); - if (rel_scoped) + /* + * See WsScanLock. A caller that named a replay position never waits here: + * the release for a lock replay is holding can be in the WAL that caller + * is withholding, so waiting deadlocks the daemon against its own shadow. + * Reading without the lock is licensed only once that position is known to + * be where replay is, which is what holds these pages still — together + * with the shadow being read-only, so no local backend can write either. + */ + if (lock == WS_SCAN_LOCK_WAIT) + { + LockRelationOid(plan.relid, AccessShareLock); + locked = true; + } + else + { + locked = ConditionalLockRelationOid(plan.relid, AccessShareLock); + if (!locked && lock == WS_SCAN_LOCK_NOWAIT) + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("walshadow: %u is locked and replay is not at the position the scan named", + plan.relid), + errdetail("Reading without the lock needs that position to hold; waiting for it can deadlock against withheld WAL."))); + if (!locked) + elog(DEBUG1, + "walshadow: pinned scan of %u read without AccessShareLock", + plan.relid); + } + rel = table_open(plan.relid, NoLock); + + if (!locked) + { + Oid *sorted = NULL; + + if (rel_scoped) + { + sorted = palloc_array(Oid, noids); + memcpy(sorted, oids, sizeof(Oid) * (size_t) noids); + qsort(sorted, (size_t) noids, sizeof(Oid), ws_oid_cmp); + } + ws_scan_emit_lockfree(rel, &plan, sorted, rel_scoped ? noids : 0, + top, out, stats); + } + else if (rel_scoped) { int i; @@ -441,5 +552,7 @@ ws_overlay_scan(WsCatalog cat, TransactionId top, const Oid *oids, int noids, else ws_scan_emit(rel, &plan, InvalidOid, top, false, out, stats); - table_close(rel, AccessShareLock); + table_close(rel, NoLock); + if (locked) + UnlockRelationOid(plan.relid, AccessShareLock); } diff --git a/pgext/test/Makefile b/pgext/test/Makefile new file mode 100644 index 00000000..90b2f131 --- /dev/null +++ b/pgext/test/Makefile @@ -0,0 +1,8 @@ +# Test-only loadable module; see wstest.c. Built by the integration tests +# that need VACUUM's lock lifetime without waiting on autovacuum. +# +# make -C pgext/test +MODULES = wstest +PG_CONFIG ?= pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) diff --git a/pgext/test/wstest.c b/pgext/test/wstest.c new file mode 100644 index 00000000..d56ab38d --- /dev/null +++ b/pgext/test/wstest.c @@ -0,0 +1,30 @@ +/* + * Test-only helper. Not part of the walshadow module: built and loaded only + * by the daemon's integration tests, never by a deployed shadow. + * + * Reproduces VACUUM's lock lifetime (vacuumlazy.c, lazy_truncate_heap): + * acquire AccessExclusiveLock on a relation, then release it source-side + * while the surrounding transaction stays open. The acquire emits + * XLOG_STANDBY_LOCK; the release emits nothing. A standby therefore keeps + * the replayed lock until the transaction's commit record arrives, which is + * the asymmetry tests need to drive deterministically. + */ +#include "postgres.h" + +#include "fmgr.h" +#include "storage/lmgr.h" + +PG_MODULE_MAGIC; + +PG_FUNCTION_INFO_V1(ws_test_lock_unlock_relation); + +Datum +ws_test_lock_unlock_relation(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + + LockRelationOid(relid, AccessExclusiveLock); + UnlockRelationOid(relid, AccessExclusiveLock); + + PG_RETURN_VOID(); +} diff --git a/pgext/walshadow.h b/pgext/walshadow.h index 7e3c0801..a0533a75 100644 --- a/pgext/walshadow.h +++ b/pgext/walshadow.h @@ -8,7 +8,7 @@ /* Bumped when a request or response layout changes, or when an op's reading * of an unchanged layout changes */ -#define WS_PROTO_VERSION 2 +#define WS_PROTO_VERSION 3 /* Bumped when any catalog projection changes shape */ #define WS_PROJECTION_VERSION 1 @@ -61,12 +61,34 @@ typedef struct WsScanStats extern int ws_overlay_ncols(WsCatalog cat); +/* + * How a scan may treat the target catalog's lock. + * + * A caller that named a replay position has withheld every successor byte to + * park the shadow there, so a lock this scan waits for can have its release + * sitting in that withheld WAL — waiting deadlocks the daemon against its own + * shadow. Such callers therefore never wait. Reading *without* the lock needs + * more than that: the position the caller named must be where replay actually + * is, which is what makes the pages stable. + */ +typedef enum WsScanLock +{ + /* No position named. Ordinary locking, and it may wait */ + WS_SCAN_LOCK_WAIT = 0, + /* Position named but replay is elsewhere, so the caller's guarantee does + * not hold: take the lock if free, error rather than wait or read past it */ + WS_SCAN_LOCK_NOWAIT = 1, + /* Position named and verified: take the lock if free, read without it + * otherwise */ + WS_SCAN_LOCK_PINNED = 2, +} WsScanLock; + /* * `top` invalid reads the committed view; an empty `oids` reads the whole * catalog, which is the only mode pg_namespace and pg_type have. */ extern void ws_overlay_scan(WsCatalog cat, TransactionId top, - const Oid *oids, int noids, + const Oid *oids, int noids, WsScanLock lock, StringInfo out, WsScanStats *stats); #endif /* WALSHADOW_H */ diff --git a/pgext/worker.c b/pgext/worker.c index 4f9fd40e..68f0128b 100644 --- a/pgext/worker.c +++ b/pgext/worker.c @@ -312,6 +312,8 @@ ws_handle_scan(StringInfo req, StringInfo resp) { WsCatalog cat = (WsCatalog) pq_getmsgbyte(req); TransactionId top = (TransactionId) pq_getmsgint(req, 4); + /* Replay position the caller parked at, 0 when it has not parked one */ + uint64 boundary = pq_getmsgint64(req); uint32 noids = pq_getmsgint(req, 4); int ncols = ws_overlay_ncols(cat); Oid *oids = NULL; @@ -319,6 +321,7 @@ ws_handle_scan(StringInfo req, StringInfo resp) WsScanStats stats = {0, 0, 0}; uint64 lsn_start; uint64 lsn_end; + WsScanLock lock; uint32 i; if (ncols < 0) @@ -340,12 +343,22 @@ ws_handle_scan(StringInfo req, StringInfo resp) initStringInfo(&rows); /* - * Caller asserts both LSNs equal the boundary it parked replay at. Equal - * but wrong is impossible: replay cannot rewind, and the daemon holds the - * successor bytes. + * Both LSNs go back for the caller to check against the boundary it parked + * replay at. Equal but wrong is impossible: replay cannot rewind, and the + * daemon holds the successor bytes. + * + * The pre-scan sample also decides how the scan may treat the catalog's + * lock: reading without it needs replay to actually be where the caller + * says, and that cannot be judged from the response. */ lsn_start = (uint64) GetXLogReplayRecPtr(NULL); - ws_overlay_scan(cat, top, oids, (int) noids, &rows, &stats); + if (boundary == 0) + lock = WS_SCAN_LOCK_WAIT; + else if (lsn_start == boundary) + lock = WS_SCAN_LOCK_PINNED; + else + lock = WS_SCAN_LOCK_NOWAIT; + ws_overlay_scan(cat, top, oids, (int) noids, lock, &rows, &stats); lsn_end = (uint64) GetXLogReplayRecPtr(NULL); pq_sendbyte(resp, WS_STATUS_OK); diff --git a/src/catalog/shadow_catalog.rs b/src/catalog/shadow_catalog.rs index f5b2dc30..7fdc7f9c 100644 --- a/src/catalog/shadow_catalog.rs +++ b/src/catalog/shadow_catalog.rs @@ -17,6 +17,7 @@ //! Single-database model: instance bound to one DB. use std::collections::BTreeSet; +use std::num::NonZeroU64; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant}; @@ -130,6 +131,15 @@ macro_rules! query_with_reconnect { }}; } +/// Replay position a name lookup is pinned to, plus whether the caller pinned +/// it (withholding successor WAL) or this read merely observed it. Only the +/// former may skip SQL. +#[derive(Clone, Copy)] +struct NamePin { + lsn: u64, + by_caller: bool, +} + impl ShadowCatalog { /// Connect over a libpq key=value conninfo. One-shot; wrap in /// [`with_transient_retry`] for retry-on-PG-coming-up. `conninfo` is stashed @@ -261,7 +271,14 @@ impl ShadowCatalog { } /// Wait until shadow's replay LSN ≥ `target`, returning the deciding poll's - /// LSN. `target = 0` returns on the first non-NULL LSN. + /// LSN. `target = 0` returns on the first non-zero LSN. + /// + /// Polls the bridge worker, not SQL. `SELECT pg_last_wal_replay_lsn()` + /// needs no lock of its own but its *parse* opens `pg_type`, so it blocks + /// whenever replay is holding an AccessExclusiveLock on a catalog — and the + /// release for such a lock can be in WAL this very wait is what unblocks. + /// The worker reads the position out of shared memory instead, and it is + /// long-lived, so it parses nothing here. pub async fn wait_for_replay(&mut self, target: u64) -> Result { if let Some(seen) = self.last_replay_lsn && seen >= target @@ -271,11 +288,10 @@ impl ShadowCatalog { } self.stats.replay_waits += 1; let start = Instant::now(); + let bridge = self.bridge.clone(); loop { - let row = self - .query_one_retry("SELECT pg_last_wal_replay_lsn()", &[]) - .await?; - let lsn = row.get::<_, Option>(0).map(u64::from); + // Worker reports 0 off a standby, same as the SQL function's NULL + let lsn = NonZeroU64::new(bridge.replay_lsn().await?).map(NonZeroU64::get); if let Some(lsn) = lsn { self.last_replay_lsn = Some(self.last_replay_lsn.map_or(lsn, |old| old.max(lsn))); if lsn >= target { @@ -301,18 +317,47 @@ impl ShadowCatalog { &mut self, oids: &[Oid], ) -> Result<(u64, Vec)> { - self.fetch_committed(Scope::Oids(oids)).await + self.fetch_committed(Scope::Oids(oids), None).await } /// Every eligible user relation: capture-all + descriptor-log boot seed. pub async fn fetch_all_descriptors(&mut self) -> Result<(u64, Vec)> { - self.fetch_committed(Scope::Eligible).await + self.fetch_committed(Scope::Eligible, None).await + } + + /// [`fetch_descriptors_batch`](Self::fetch_descriptors_batch) for a caller + /// that has parked replay at `boundary` and withheld every successor byte. + /// + /// Saying so is what keeps the read out of the deadlock: the worker checks + /// the position on both sides and may then read a catalog whose lock replay + /// is holding, and no step falls back to ordinary SQL, whose parse would + /// queue behind that same lock. + pub async fn fetch_descriptors_batch_at( + &mut self, + oids: &[Oid], + boundary: u64, + ) -> Result<(u64, Vec)> { + self.fetch_committed(Scope::Oids(oids), Some(boundary)) + .await + } + + /// [`fetch_all_descriptors`](Self::fetch_all_descriptors), pinned as in + /// [`fetch_descriptors_batch_at`](Self::fetch_descriptors_batch_at). + pub async fn fetch_all_descriptors_at( + &mut self, + boundary: u64, + ) -> Result<(u64, Vec)> { + self.fetch_committed(Scope::Eligible, Some(boundary)).await } /// Committed catalog at one replay position. - async fn fetch_committed(&mut self, scope: Scope<'_>) -> Result<(u64, Vec)> { + async fn fetch_committed( + &mut self, + scope: Scope<'_>, + boundary: Option, + ) -> Result<(u64, Vec)> { self.stats.fetches += 1; - let rows = self.committed_rows(scope).await?; + let rows = self.committed_rows(scope, boundary).await?; let replay_lsn = rows.replay_lsn; let db_node = self.current_db_oid().await?; let default_tablespace = self.default_tablespace_oid().await?; @@ -324,9 +369,20 @@ impl ShadowCatalog { /// hold; away from one it moves between requests, so no sequence of scans /// answers for a single position and the statement's one snapshot always /// does. - async fn committed_rows(&mut self, scope: Scope<'_>) -> Result { + async fn committed_rows( + &mut self, + scope: Scope<'_>, + boundary: Option, + ) -> Result { let bridge = self.bridge.clone(); - match self.scan_rows(&bridge, scope, 0, None).await { + let res = self.scan_rows(&bridge, scope, 0, boundary).await; + // A pinned read has no SQL fallback: the mirroring statement's parse + // opens pg_type, so it would queue behind exactly the recovery-held + // lock the pinned path exists to read past. Surface the error instead + if boundary.is_some() { + return res; + } + match res { Err(e) if worker_cannot_answer(&e) => self.mirror_rows(scope).await, other => other, } @@ -417,7 +473,10 @@ impl ShadowCatalog { &class.iter().map(|c| c.relnamespace).collect(), |r: NamespaceRow| (r.oid, r.nspname), top_xid, - pinned, + NamePin { + lsn: pinned, + by_caller: boundary.is_some(), + }, ) .await?; // DROP COLUMN zeroes atttypid, and no pg_type row for it is what leaves @@ -434,7 +493,10 @@ impl ShadowCatalog { .collect(), |r: TypeRow| (r.oid, r.typname), top_xid, - pinned, + NamePin { + lsn: pinned, + by_caller: boundary.is_some(), + }, ) .await?; @@ -459,14 +521,20 @@ impl ShadowCatalog { DescriptorRows::from_mirror(&rows) } - /// Oid → name for one whole-catalog projection, committed read first and - /// the overlay only for what it missed. + /// Oid → name for one whole-catalog projection. /// - /// The overlay scan behind it has no oid list and so no lock argument, and - /// refuses to answer while any foreign writer is mid-DDL. Running it only - /// for what the committed read lacks keeps that exposure to names the - /// requesting transaction created itself; a committed read never gets - /// there at all. + /// A pinned read (`caller_pinned`) goes to the worker and stops there. + /// The SQL it would otherwise try first resolves names out of `pg_type` and + /// `pg_namespace`, and its parse takes AccessShareLock on `pg_type` — the + /// lock replay can be holding on behalf of a transaction whose commit is in + /// the WAL the caller is withholding. That query is the one that hung the + /// daemon against its own shadow. + /// + /// Unpinned reads keep the committed-read-first order: the overlay scan + /// behind them has no oid list and so no lock argument, and refuses to + /// answer while any foreign writer is mid-DDL, so running it only for what + /// the committed read lacks keeps that exposure to names the requesting + /// transaction created itself. async fn resolve_names( &mut self, bridge: &Bridge, @@ -474,18 +542,21 @@ impl ShadowCatalog { wanted: &BTreeSet, name_of: fn(R) -> (Oid, String), top_xid: u32, - boundary: u64, + pin: NamePin, ) -> Result> { if wanted.is_empty() { return Ok(HashMap::new()); } - let list: Vec = wanted.iter().copied().collect(); - let rows = self.query_retry(sql, &[&list]).await?; - let mut names: HashMap = rows.iter().map(|r| (r.get(0), r.get(1))).collect(); - if names.len() == wanted.len() { - return Ok(names); + let mut names: HashMap = HashMap::new(); + if !pin.by_caller { + let list: Vec = wanted.iter().copied().collect(); + let rows = self.query_retry(sql, &[&list]).await?; + names.extend(rows.iter().map(|r| (r.get(0), r.get(1)))); + if names.len() == wanted.len() { + return Ok(names); + } } - let scan = bridge.scan_at(R::CATALOG, top_xid, &[], boundary).await?; + let scan = bridge.scan_at(R::CATALOG, top_xid, &[], pin.lsn).await?; for row in scan.parse::()? { let (oid, name) = name_of(row); if wanted.contains(&oid) { diff --git a/src/ops/bridge.rs b/src/ops/bridge.rs index bbb517b6..0085954e 100644 --- a/src/ops/bridge.rs +++ b/src/ops/bridge.rs @@ -21,7 +21,7 @@ use tokio::net::UnixStream; use tokio::sync::Mutex; /// Frame and op layouts. Must equal `WS_PROTO_VERSION` in `pgext/walshadow.h` -pub const PROTO_VERSION: u32 = 2; +pub const PROTO_VERSION: u32 = 3; /// Catalog column plans. Must equal `WS_PROJECTION_VERSION` pub const PROJECTION_VERSION: u32 = 1; @@ -264,15 +264,34 @@ impl Bridge { /// have. Losing the oid list loses the lock argument with it, so an /// uncommitted whole-catalog read fails rather than guess at a writer whose /// parentage standby `pg_subtrans` cannot resolve + /// + /// Unpinned: the worker locks the catalog the ordinary way. Callers holding + /// replay still use [`scan_at`](Self::scan_at) pub async fn scan( &self, cat: Catalog, top_xid: u32, oids: &[u32], ) -> Result { - let mut payload = Vec::with_capacity(9 + oids.len() * 4); + self.scan_inner(cat, top_xid, oids, 0).await + } + + /// `boundary` is the replay position the caller has parked the shadow at, + /// or `0` when it has not parked one. Naming it licenses the worker to read + /// a catalog whose lock replay is holding — necessary because that lock's + /// release can be in the WAL the caller is withholding — so the worker + /// re-checks the position before doing so. + async fn scan_inner( + &self, + cat: Catalog, + top_xid: u32, + oids: &[u32], + boundary: u64, + ) -> Result { + let mut payload = Vec::with_capacity(17 + oids.len() * 4); payload.push(cat as u8); payload.extend_from_slice(&top_xid.to_be_bytes()); + payload.extend_from_slice(&boundary.to_be_bytes()); payload.extend_from_slice(&(oids.len() as u32).to_be_bytes()); for oid in oids { payload.extend_from_slice(&oid.to_be_bytes()); @@ -335,20 +354,21 @@ impl Bridge { oids: &[u32], boundary: u64, ) -> Result { - let res = self.scan(cat, top_xid, oids).await?; + let res = self.scan_inner(cat, top_xid, oids, boundary).await?; self.pinned(res, boundary) } /// First scan of a read with no boundary of its own: whatever position it /// reports becomes the pin for the rest, so only a move inside this one - /// scan fails here + /// scan fails here. Sent unpinned — the caller has nothing to assert yet, + /// so the worker keeps normal locking for it pub async fn scan_pinning( &self, cat: Catalog, top_xid: u32, oids: &[u32], ) -> Result { - let res = self.scan(cat, top_xid, oids).await?; + let res = self.scan_inner(cat, top_xid, oids, 0).await?; let boundary = res.replay_lsn_start; self.pinned(res, boundary) } diff --git a/src/source/catalog_capture.rs b/src/source/catalog_capture.rs index 1af42a22..b5ab2979 100644 --- a/src/source/catalog_capture.rs +++ b/src/source/catalog_capture.rs @@ -432,10 +432,10 @@ impl CatalogCapture { let mut cat = self.catalog.lock().await; if info.capture_all { self.stats.capture_all_runs.fetch_add(1, Relaxed); - cat.fetch_all_descriptors().await + cat.fetch_all_descriptors_at(next_lsn).await } else { let oids: Vec = info.oids.iter().map(|a| a.oid).collect(); - cat.fetch_descriptors_batch(&oids).await + cat.fetch_descriptors_batch_at(&oids, next_lsn).await } } .map_err(|e| SinkError::Other(format!("descriptor capture at {commit_lsn:#X}: {e}")))?; diff --git a/tests/catalog_ael_boundary.rs b/tests/catalog_ael_boundary.rs new file mode 100644 index 00000000..09445647 --- /dev/null +++ b/tests/catalog_ael_boundary.rs @@ -0,0 +1,263 @@ +//! A recovery-held catalog `AccessExclusiveLock` must not wedge descriptor +//! capture. +//! +//! Source-side, `VACUUM`'s truncation step takes `AccessExclusiveLock` on the +//! catalog it is vacuuming and releases it before its transaction commits +//! (`vacuumlazy.c`, `lazy_truncate_heap`). A standby has no record of that +//! early release: `StandbyAcquireAccessExclusiveLock` holds the replayed lock +//! until the owning transaction's commit record arrives. So between the lock +//! record and that commit, the shadow's startup process holds AEL on a catalog +//! while other transactions keep committing catalog changes. +//! +//! walshadow stops publishing successor WAL at a catalog boundary and then +//! reads the shadow's catalog. If that read waits on the recovery-held lock, +//! the release it waits for is in the WAL it is withholding — a closed cycle. +//! `pg_type` makes it total: every statement resolves types at parse time, so +//! the entire shadow stops answering. +//! +//! The lock is manufactured with `pgext/test/wstest.so` rather than by waiting +//! for autovacuum, so the required ordering — `V` lock < `D` boundary < `V` +//! commit — is deterministic. + +#![cfg(target_os = "linux")] + +#[path = "common/inproc_harness.rs"] +mod fx; + +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use walshadow::mapping::NamespaceMapping; +use walshadow::record::{Record, RecordSink, SinkError}; +use walshadow::segment_sink::DirSegmentSink; +use walshadow::source_feed::{SourceEvent, SourceFeed, StandbyStatus}; +use walshadow::wal_stream::WalStream; + +/// Long enough that `V` is still open for the whole capture attempt. +const HOLD_SECS: u64 = 90; +/// Capture is a few local round trips; a fixed budget well above that turns +/// the deadlock into a test failure instead of a hung run. +const CAPTURE_BUDGET: Duration = Duration::from_secs(25); + +fn wstest_module() -> PathBuf { + let so = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("pgext/test") + .join("wstest.so"); + assert!( + so.is_file(), + "{} missing, run `make -C pgext/test`", + so.display() + ); + so +} + +/// Tracks per-record progress: `WalStream::dispatched_lsn` only moves at +/// segment boundaries, so it cannot tell a stalled pump from a few KB of WAL. +struct Progress<'s> { + inner: &'s mut fx::PipelineSinks, + max_next_lsn: u64, +} + +impl RecordSink for Progress<'_> { + fn on_record<'a>( + &'a mut self, + record: &'a Record<'a>, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async move { + self.inner.on_record(record).await?; + self.max_next_lsn = self.max_next_lsn.max(record.next_lsn); + Ok(()) + }) + } +} + +/// Pump until a record whose `next_lsn` reaches `target` has been *processed*. +/// Separate `&mut` fields rather than `&mut Pipeline` so borrows stay disjoint. +async fn pump_to_lsn( + feed: &mut SourceFeed, + stream: &mut WalStream, + sinks: &mut Progress<'_>, + segs: &mut DirSegmentSink, + buf: &mut Vec, + target: u64, +) -> Result<(), String> { + while sinks.max_next_lsn < target { + let next = tokio::time::timeout( + Duration::from_secs(2), + feed.next_event(StandbyStatus::collapsed(stream.dispatched_lsn()), buf), + ) + .await; + let chunk = match next { + Ok(Ok(SourceEvent::Wal(c))) => c, + Ok(Ok(_)) => break, + Ok(Err(e)) => return Err(format!("source feed: {e:#}")), + Err(_) => continue, + }; + stream + .push(chunk.start_lsn, chunk.data, sinks, segs) + .await + .map_err(|e| format!("push: {e}"))?; + } + Ok(()) +} + +/// Drive one catalog boundary through capture. `hold_lock` decides whether a +/// transaction leaves a recovery-held AEL on `pg_type` straddling it. +/// Returns how far capture got, or `Err` once the budget expires. +async fn run_boundary(hold_lock: bool) -> Result<(), String> { + let so = wstest_module(); + let ports = fx::Ports::alloc(); + let tmp = tempfile::tempdir().unwrap(); + + let schema = format!( + "CREATE TABLE demo (id int primary key, v text);\n\ + INSERT INTO demo VALUES (1, 'a');\n\ + CREATE FUNCTION ws_test_lock_unlock_relation(oid) RETURNS void \ + AS '{}', 'ws_test_lock_unlock_relation' LANGUAGE c;\n", + so.display(), + ); + + let ( + fx::BootstrappedClusters { + source, + shadow, + shadow_filter_dir, + }, + shadow_stream_state, + ) = fx::bootstrap_clusters(&tmp, &schema, ports.source, ports.shadow, ports.walsender).await; + let _keep_shadow = &shadow; + + let ch_tmp = tempfile::tempdir().unwrap(); + let ch = fx::ChServer::spawn(ch_tmp, ports.ch_tcp, ports.ch_http).expect("spawn ch"); + ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test") + .expect("create ch db"); + + let mut ddl_args = fx::DdlPipelineArgs::default(); + ddl_args.namespaces.insert( + "public".into(), + NamespaceMapping { + target_database: Some("walshadow_test".into()), + auto_create: true, + drop_table_strategy: None, + initial_load: None, + }, + ); + + let mut pipeline = fx::build_pipeline(fx::BuildPipelineArgs { + tmp: &tmp, + source: &source, + shadow: &shadow, + shadow_filter_dir: &shadow_filter_dir, + shadow_stream_state, + ch_database: "walshadow_test", + ch_tcp_port: ports.ch_tcp, + mappings: vec![], + app_name: "catalog-ael-boundary", + ddl: Some(ddl_args), + }) + .await; + + // Transaction V: take AEL on pg_type, release it source-side, stay open. + // The shadow keeps the replayed lock until V commits. + let v = hold_lock.then(|| { + fx::spawn_txn( + &source, + &format!( + "BEGIN;\n\ + SELECT ws_test_lock_unlock_relation('pg_catalog.pg_type'::regclass);\n\ + SELECT pg_sleep({HOLD_SECS});\n\ + COMMIT;\n" + ), + ) + }); + + // V parks in pg_sleep, so its lock record is written and its commit is + // not. The source-side lock is already gone, which is what lets D run. + if hold_lock { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let parked = source + .psql_one( + "SELECT count(*) FROM pg_stat_activity \ + WHERE state = 'active' AND query LIKE 'SELECT pg_sleep%'", + ) + .map_err(|e| format!("probe V: {e}"))?; + if parked.trim() == "1" { + break; + } + if Instant::now() >= deadline { + return Err("transaction V never parked".into()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + // Transaction D: an unrelated catalog mutation, so its boundary lands in + // the WAL between V's lock and V's commit. + source + .psql_one("ALTER TABLE demo ADD COLUMN w int") + .map_err(|e| format!("D ddl: {e}"))?; + let target = { + let s = source + .psql_one("SELECT pg_current_wal_insert_lsn()") + .map_err(|e| format!("insert lsn: {e}"))?; + walshadow::pg::parse_pg_lsn(&s).map_err(|e| format!("parse lsn: {e}"))? + }; + + let mut progress = Progress { + inner: &mut pipeline.sinks, + max_next_lsn: 0, + }; + let pumped = tokio::time::timeout( + CAPTURE_BUDGET, + pump_to_lsn( + &mut pipeline.feed, + &mut pipeline.stream, + &mut progress, + &mut pipeline.segment_sink, + &mut pipeline.chunk_buf, + target, + ), + ) + .await; + let reached = progress.max_next_lsn; + let captures = pipeline.desc_log.covered_through(); + drop(v); + + match pumped { + Err(_) => Err(format!( + "pump made no progress past {reached:#X} within {CAPTURE_BUDGET:?} \ + (target {target:#X}, descriptor log covered through {captures:#X})" + )), + Ok(Err(e)) => Err(e), + Ok(Ok(())) => Ok(()), + } +} + +/// Control: the identical boundary with no lock straddling it must capture +/// promptly. Guards the repro below against a harness-level stall. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn boundary_capture_completes_without_held_catalog_lock() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + run_boundary(false).await.expect("control boundary"); +} + +/// Regression: capture must not wait on a lock whose release is in the WAL +/// walshadow is withholding. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn boundary_capture_survives_recovery_held_catalog_lock() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + if let Err(e) = run_boundary(true).await { + panic!( + "boundary capture wedged behind a recovery-held AccessExclusiveLock \ + on pg_type: {e}" + ); + } +}