diff --git a/.changeset/bounded-mvcc-maintenance.md b/.changeset/bounded-mvcc-maintenance.md new file mode 100644 index 000000000..26f3145a6 --- /dev/null +++ b/.changeset/bounded-mvcc-maintenance.md @@ -0,0 +1,12 @@ +--- +"@reddb-io/cli": patch +--- + +Reclaim obsolete table history in bounded background batches for writable +standalone embedded runtimes. Avoid copying row payloads during VACUUM and +planner statistics collection, and bound cardinality and histogram working sets. +Rebalance deleted B-tree pages iteratively so cascading merges do not retain +page buffers from every level on a maintenance worker's stack. +Reuse trees after completed physical delete batches instead of re-encoding +unchanged survivors at the next checkpoint, with revision checks that retain +full validation after untracked edits. diff --git a/.changeset/checkpoint-bulk-row-publication.md b/.changeset/checkpoint-bulk-row-publication.md new file mode 100644 index 000000000..921842eac --- /dev/null +++ b/.changeset/checkpoint-bulk-row-publication.md @@ -0,0 +1,5 @@ +--- +"@reddb-io/cli": patch +--- + +Fence bulk row insertion through its WAL capture or append so a checkpoint cannot publish an unfinished batch that starts during serialization. Add forced-crash regressions for explicit transactions and acknowledged autocommit batches. diff --git a/.changeset/checkpoint-concurrent-row-publication.md b/.changeset/checkpoint-concurrent-row-publication.md new file mode 100644 index 000000000..53d13ddb2 --- /dev/null +++ b/.changeset/checkpoint-concurrent-row-publication.md @@ -0,0 +1,7 @@ +--- +"@reddb-io/cli": patch +--- + +Fence row mutations and delete stamps through their WAL append against checkpoint +publication. Transactions starting during a checkpoint cannot leak uncommitted +rows, and concurrent acknowledged autocommit updates retain their recovery WAL. diff --git a/.changeset/checkpoint-unfinished-transactions.md b/.changeset/checkpoint-unfinished-transactions.md new file mode 100644 index 000000000..5a875e7f5 --- /dev/null +++ b/.changeset/checkpoint-unfinished-transactions.md @@ -0,0 +1,8 @@ +--- +"@reddb-io/cli": patch +--- + +Exclude unfinished row versions from crash-recoverable checkpoints and retain +their committed preimages. Invalidate checkpoint reuse when transaction outcomes +change, and fence commit WAL publication with visibility so a concurrent +checkpoint cannot discard a durable commit's recovery records. diff --git a/.changeset/reopened-positional-row-updates.md b/.changeset/reopened-positional-row-updates.md new file mode 100644 index 000000000..8b76ba82f --- /dev/null +++ b/.changeset/reopened-positional-row-updates.md @@ -0,0 +1,7 @@ +--- +"@reddb-io/cli": patch +--- + +Restore shared column names on owned reads of reopened positional rows. Preserve +untouched values during SQL updates and include positional rows when rebuilding +secondary indexes without changing the file format or copying payloads again. diff --git a/.changeset/transaction-pager-flush.md b/.changeset/transaction-pager-flush.md new file mode 100644 index 000000000..3a64fb67f --- /dev/null +++ b/.changeset/transaction-pager-flush.md @@ -0,0 +1,5 @@ +--- +"@reddb-io/cli": patch +--- + +Keep captured transaction inserts and row upserts out of published B-tree pages until their committed WAL is durable. This prevents incidental pager flushes from recovering unfinished inserts or delete stamps after a crash. Add crash regressions for INSERT, bulk INSERT, UPDATE and DELETE with open, rolled-back and committed transactions. diff --git a/.red/adr/0014-mvcc-history-store-and-transaction-recovery.md b/.red/adr/0014-mvcc-history-store-and-transaction-recovery.md index 93cc487b4..b6778c964 100644 --- a/.red/adr/0014-mvcc-history-store-and-transaction-recovery.md +++ b/.red/adr/0014-mvcc-history-store-and-transaction-recovery.md @@ -49,8 +49,8 @@ This ADR ratifies the implementation contract for the first rollout: paths explicitly adopt the resolver. - RedDB implements snapshot isolation with first-committer-wins write conflict detection. It does not implement serializable isolation or SSI. -- The first rollout includes manual `VACUUM` / GC only. It does not include an - autovacuum daemon. +- The original rollout includes manual `VACUUM` / GC only. The bounded + standalone maintenance evolution below adds automatic table-history cleanup. - Current secondary indexes plus MVCC recheck and fallback are the first correctness target. Historical secondary indexes are deferred. - Full multi-model adoption is deferred behind table-row correctness. @@ -431,3 +431,40 @@ duplicating rules. (the baseline); the new non-table PATCH path is flag-gated. For a non-versioned document the two verbs therefore differ in history accrual (invisible without AS OF, which is rejected for non-versioned). Align if uniform behavior is wanted. + +## Evolution: bounded standalone maintenance (2026-10-20) + +Writable standalone embedded runtimes reclaim obsolete table versions through +an existing 200 ms maintenance tick. Each tick inspects at most 4,096 physical +positions and reclaims at most 256 versions. These limits include empty or +retired scan positions. Source cursors use weak segment references so a slow +sweep does not retain retired payloads across consolidation. A subsequent sweep +visits replacement segments and writes appended after cursor capture. + +Reclamation uses the same transaction, statement and external-pin horizons as +manual VACUUM. Aborted creators can be removed, while aborted deletions never +make a live version eligible. Transaction outcomes remain retained until a +separate global reference proof permits pruning them. Physical deletes and +WAL publication share the checkpoint fence; a busy checkpoint defers a tick. +Collection replacement invalidates its old cursor. Failed batches retry with +backoff and rate-limited diagnostics. + +`runtime.mvcc.autovacuum.enabled` defaults to `true` in this supported scope. +`runtime.mvcc.autovacuum.scan_rows_per_tick` and +`runtime.mvcc.autovacuum.reclaim_rows_per_tick` can reduce their respective +budgets; values are clamped to 1..4096 and 1..256. Read-only runtimes and the +serverless, primary-replica and cluster deploy profiles do not enable this local collector: they require their +own write-authority and distributed history-retention contract first. + +Manual VACUUM borrows headers instead of cloning a collection's payloads. +Statistics collection also borrows rows, caps exact fingerprint cardinality at +30,000 entries before using HyperLogLog, and caps each column's histogram +sample at 1 MiB. Values over 4 KiB leave min/max bounds unknown. Statistics are +planner estimates and must not be used as proof that a data range is empty. + +Resource sketch: a collector tick buffers at most 256 source IDs and 256 +reclamation IDs, plus bounded lookup scratch (about 6 KiB total), rather than +N rows multiplied by payload width. Cursor metadata scales with segment count. +Statistics use bounded cardinality and sample buffers per column; neither path +makes a second full collection of row payloads. Explicit VACUUM still traverses +the target collection; the automatic collector resumes between ticks. diff --git a/.red/context/persistence.md b/.red/context/persistence.md index f30bae639..032b42638 100644 --- a/.red/context/persistence.md +++ b/.red/context/persistence.md @@ -27,11 +27,29 @@ Part of the [glossary map](../CONTEXT-MAP.md). The storage engine shared by **ev ## Checkpoint reuse +Checkpoints exclude row creators whose transactions are active or aborted and +clear their delete stamps on committed preimages. Live transaction state stays +unchanged. Transaction visibility revisions invalidate reuse after begin, commit +or rollback; equivalent exclusion sets can retain the existing proof. Commit +WAL publication and visibility changes share the checkpoint publication fence. +While store WAL actions are captured for a transaction, row inserts and +upserts stay in the live segment and deferred WAL, leaving published B-tree +pages unchanged. Incidental pager flushes cannot publish unfinished inserts +or delete stamps; a successful commit makes the captured WAL batch durable. +Bulk inserts hold the checkpoint publication fence before installing their +segment rows through WAL capture or append, including transactions that begin +after checkpoint visibility has already been frozen. + +Owned row reads restore missing positional column names from the collection's +shared schema before mutation or index backfill. Explicit row schemas and named +fields remain authoritative. This changes neither the stored record format nor +the raw segment revisions used by checkpoint reuse. + Paged checkpoints reuse unchanged collection trees using volatile segment mutation -revisions and manager identity. Aborted transaction visibility, pending vault +revisions and manager identity. Active and aborted transaction visibility, pending vault versions and record format also participate in reuse admission. Visibility changes outside a collection are checked against borrowed row headers -without serializing its payloads. Each checkpoint shares one frozen aborted-XID +without serializing its payloads. Each checkpoint shares one frozen excluded-XID set across collections for both reuse admission and row normalization. WAL replay and every entity or metadata mutation invalidate the corresponding generation. New mutation paths must advance the segment revision before exposing mutable data. @@ -64,6 +82,11 @@ visibility changes and versioned rows retain conservative validation. Durable replay reconciliation seeds the current raw tree generation before runtime visibility normalization is attached. +Completed WAL-backed physical delete batches may advance that proof when every +tree removal succeeds and the segment revision delta accounts for exactly the +deleted rows. Earlier untracked edits, changed segment topology, pending vault +versions and format changes retain full checkpoint validation. + Changed checkpoints encode ordered borrowed rows in batches capped at 1 MiB or 512 records (a single larger value keeps the existing value-size contract). Comparison with a published tree is streamed; a replacement copies the already @@ -131,3 +154,9 @@ The runtime rejects incomplete remote snapshots before advertising success. - **LSN-pinned analytical scan** — read protocol for the columnar analytics projection. An analytical read is an ordinary MVCC read: it pins the current snapshot/LSN, scans columnar segments up to the last materialized LSN, and concatenates the un-materialized tail (rows past the last checkpoint boundary) through the normal row read path. There is a single consistency class — analytical reads are never a separate stale tier — and `AS OF` time-travel composes on top by pinning a historical LSN instead of the current one. - **Store fork** — metadata-level fork of an entire store: a new operational manifest generation pinned to a fork LSN, sharing the parent's immutable artifacts and writing its own private WAL from that point forward. Creation is an O(metadata) operation, never a physical copy of the dataset. Deliberately distinct from the VCS data model's *branch* (#1567 vocabulary): a store fork is storage/deploy mechanics for experiment-and-discard workflows (e.g. testing a risky migration or letting an agent run against production-shaped data), not version-control semantics over collection contents. Mechanically, a fork is a lazy in-place restore-to-LSN riding the operational backup/restore boundary (ADR 0043): immutable artifacts (append-only segments, checkpoint artifacts, columnar projection segments) are shared by manifest reference; mutable collection files are hydrated lazily (replay/CoW) the first time the fork touches them, so creation stays O(metadata) and cost is deferred to what the fork actually writes. Lifecycle is diverge-and-discard: a fork ends by being discarded, kept as an independent store, or promoted to new primary (the recovery/PITR path) — there is no merge-back of divergent stores at the storage layer; merge semantics belong to the VCS data model, which has vocabulary for conflicts. - **WAL spool** — versioned (v2) write-ahead log records used for replication streaming. Includes magic, version byte, lsn, timestamp, payload-len, payload, crc32. `sync_all` after every append. (The replica-side consumer is the **logical change applier** — see [Primary-replica](primary-replica.md) / [Clustering](clustering.md).) + +Row mutation batches and tombstone batches take the publication fence before +changing live segments and retain it through WAL capture or append. Fencing only +COMMIT is insufficient: a transaction may begin after checkpoint visibility was +frozen, and an autocommit's recovery records must not be truncated before its +mutations belong to the published generation. diff --git a/crates/reddb-server/src/application/ports_impls_entity.rs b/crates/reddb-server/src/application/ports_impls_entity.rs index 4613dc4b8..77c611b5a 100644 --- a/crates/reddb-server/src/application/ports_impls_entity.rs +++ b/crates/reddb-server/src/application/ports_impls_entity.rs @@ -1368,6 +1368,10 @@ impl RedDBRuntime { } let store = self.db().store(); + // Fence the in-memory change and its WAL append together. Otherwise a + // checkpoint can copy a new transaction's rows or truncate a concurrent + // autocommit's WAL before those rows belong to its frozen generation. + let _publication = store.vault_publication_lock.lock(); let collection = &applied[0].collection; let Some(manager) = store.get_collection(collection) else { return Err(crate::RedDBError::NotFound(format!( diff --git a/crates/reddb-server/src/runtime.rs b/crates/reddb-server/src/runtime.rs index 7d4cdbdcb..852364d70 100644 --- a/crates/reddb-server/src/runtime.rs +++ b/crates/reddb-server/src/runtime.rs @@ -1571,6 +1571,7 @@ mod mutation_index_topology_tests; #[cfg(test)] mod mutation_memory_tests; pub(crate) mod mvcc_lifecycle; +pub(crate) mod mvcc_maintenance; pub(crate) mod node_load_telemetry; pub(crate) mod occupancy_sampler; pub(crate) mod primary_queue_store; diff --git a/crates/reddb-server/src/runtime/impl_core.rs b/crates/reddb-server/src/runtime/impl_core.rs index ee2bb6723..5942b34ec 100644 --- a/crates/reddb-server/src/runtime/impl_core.rs +++ b/crates/reddb-server/src/runtime/impl_core.rs @@ -1457,9 +1457,9 @@ impl RedDBRuntime { .read() .get(&conn_id) .is_some_and(|writes| !writes.is_empty()); - let vault_store = self.inner.db.store(); + let store = self.inner.db.store(); let _vault_guard = - has_vault_writes.then(|| vault_store.vault_publication_lock.lock()); + has_vault_writes.then(|| store.vault_publication_lock.lock()); let context = self.inner.transaction_state.commit(conn_id, |ctx| { let mut own_xids = std::collections::HashSet::new(); own_xids.insert(ctx.xid); @@ -1510,8 +1510,14 @@ impl RedDBRuntime { undo?; return Err(err); } + // Validate before taking the publication fence: + // conflict rollback can acquire index topology. + // Retain the fence through WAL and visibility so a + // checkpoint cannot discard a durable commit's WAL. + let publication = store.vault_publication_lock.lock(); self.restore_pending_write_stamps(conn_id); if let Err(err) = self.flush_pending_store_wal_actions(conn_id) { + drop(publication); let undo = self.revive_pending_versioned_updates(conn_id); self.revive_pending_tombstones(conn_id); self.discard_pending_queue_dedup(conn_id); @@ -1521,7 +1527,7 @@ impl RedDBRuntime { undo?; return Err(err); } - Ok(()) + Ok(publication) }); if context.is_err() { self.discard_pending_vault_writes(conn_id)?; diff --git a/crates/reddb-server/src/runtime/impl_dml_batch.rs b/crates/reddb-server/src/runtime/impl_dml_batch.rs index 316720aae..130f46193 100644 --- a/crates/reddb-server/src/runtime/impl_dml_batch.rs +++ b/crates/reddb-server/src/runtime/impl_dml_batch.rs @@ -26,6 +26,9 @@ impl RedDBRuntime { } let store = self.db().store(); + // Keep tombstone stamps and their WAL in one publication interval, + // including writers that start after checkpoint visibility is frozen. + let _publication = store.vault_publication_lock.lock(); let Some(manager) = store.get_collection(collection) else { return Ok((0, vec![])); }; diff --git a/crates/reddb-server/src/runtime/impl_kv/transactions.rs b/crates/reddb-server/src/runtime/impl_kv/transactions.rs index d974c50ac..1d75c9d91 100644 --- a/crates/reddb-server/src/runtime/impl_kv/transactions.rs +++ b/crates/reddb-server/src/runtime/impl_kv/transactions.rs @@ -383,17 +383,21 @@ mod tests { .unwrap(); let vault_store = runtime.inner.db.store(); let _publication = vault_store.vault_publication_lock.lock(); - let _: RedDBResult<_> = runtime.inner.transaction_state.commit(121, |context| { - runtime.prepare_pending_vault_writes(121, context)?; - // Exercise both an incidental pager flush and a full checkpoint - // exactly after materialization, before COMMIT's WAL append. - runtime.inner.db.store().pager().unwrap().flush().unwrap(); - runtime.checkpoint()?; - if !operational { - runtime.publish_serverless_generation()?; - } - std::process::exit(0); - }); + let _: RedDBResult<_> = + runtime + .inner + .transaction_state + .commit(121, |context| -> RedDBResult<()> { + runtime.prepare_pending_vault_writes(121, context)?; + // Exercise both an incidental pager flush and a full checkpoint + // exactly after materialization, before COMMIT's WAL append. + runtime.inner.db.store().pager().unwrap().flush().unwrap(); + runtime.checkpoint()?; + if !operational { + runtime.publish_serverless_generation()?; + } + std::process::exit(0); + }); panic!("crash injection did not exit"); } let directory = tempfile::tempdir().unwrap(); diff --git a/crates/reddb-server/src/runtime/impl_lifecycle.rs b/crates/reddb-server/src/runtime/impl_lifecycle.rs index 5effb7a81..161e579a2 100644 --- a/crates/reddb-server/src/runtime/impl_lifecycle.rs +++ b/crates/reddb-server/src/runtime/impl_lifecycle.rs @@ -915,6 +915,8 @@ impl RedDBRuntime { let tick = std::time::Duration::from_millis(200); let work_interval = std::time::Duration::from_secs(60); let mut last_work = std::time::Instant::now(); + let mut mvcc_maintenance = + super::mvcc_maintenance::MvccMaintenance::default(); loop { std::thread::park_timeout(tick); if stopped.load(AtomicOrdering::Acquire) { @@ -929,6 +931,11 @@ impl RedDBRuntime { let _stats = inner.db.store().context_index().stats(); last_work = std::time::Instant::now(); } + let rt = RedDBRuntime { + background_tasks: None, + inner, + }; + rt.reclaim_history_tick(&mut mvcc_maintenance); } }, ); diff --git a/crates/reddb-server/src/runtime/mvcc_maintenance.rs b/crates/reddb-server/src/runtime/mvcc_maintenance.rs new file mode 100644 index 000000000..e2f5a281b --- /dev/null +++ b/crates/reddb-server/src/runtime/mvcc_maintenance.rs @@ -0,0 +1,94 @@ +//! Paced table-history maintenance for standalone embedded runtimes. +use std::time::{Duration, Instant}; + +use super::RedDBRuntime; +use crate::storage::profile::DeployProfile; +use crate::storage::unified::store::MvccReclaimCursor; + +#[derive(Default)] +pub(super) struct MvccMaintenance { + current: Option, + after_collection: Option, + retry_after: Option, + last_warning: Option, +} + +impl RedDBRuntime { + pub(super) fn reclaim_history_tick(&self, state: &mut MvccMaintenance) { + let options = self.inner.db.options(); + // Replicated profiles need a distributed reader/replica low-water + // mark before local physical history can be reclaimed automatically. + if options.read_only + || options.storage_profile.deploy_profile != DeployProfile::Embedded + || options.replication.role != crate::replication::ReplicationRole::Standalone + || self.inner.lifecycle.phase() != super::lifecycle::Phase::Ready + || !self.config_bool("runtime.mvcc.autovacuum.enabled", true) + { + state.current = None; + return; + } + if state.retry_after.is_some_and(|at| Instant::now() < at) { + return; + } + let store = self.inner.db.store(); + if state + .current + .as_ref() + .is_some_and(|cursor| store.get_collection(&cursor.collection).is_none()) + { + state.current = None; + } + if state.current.is_none() { + let mut collections = store.list_collections(); + collections.sort_unstable(); + if collections.is_empty() { + return; + } + let index = state.after_collection.as_ref().map_or(0, |after| { + collections.partition_point(|name| name <= after) % collections.len() + }); + state.current = store.mvcc_reclaim_cursor(collections[index].clone()); + } + let Some(cursor) = state.current.as_mut() else { + return; + }; + let scan_budget = self + .config_u64("runtime.mvcc.autovacuum.scan_rows_per_tick", 4096) + .clamp(1, 4096) as usize; + let reclaim_budget = self + .config_u64("runtime.mvcc.autovacuum.reclaim_rows_per_tick", 256) + .clamp(1, 256) as usize; + let result = store.reclaim_mvcc_tick( + cursor, + &self.snapshot_manager(), + self.mvcc_vacuum_cutoff_xid(), + scan_budget, + reclaim_budget, + ); + match result { + Ok(reclaimed) => { + state.retry_after = None; + if reclaimed > 0 { + tracing::debug!(collection = %cursor.collection, reclaimed_versions = reclaimed, + "paced MVCC history reclamation"); + } + if cursor.finished() { + state.after_collection = Some(cursor.collection.clone()); + state.current = None; + } + } + Err(error) => { + state.retry_after = Some(Instant::now() + Duration::from_secs(5)); + if state + .last_warning + .is_none_or(|at| at.elapsed() >= Duration::from_secs(60)) + { + tracing::warn!(%error, "MVCC history reclamation deferred"); + state.last_warning = Some(Instant::now()); + } + // An unsuccessful WAL-backed batch is not recorded as progress. + state.current = None; + } + } + } +} diff --git a/crates/reddb-server/src/runtime/transaction_state.rs b/crates/reddb-server/src/runtime/transaction_state.rs index ddb41a8da..ccea4e362 100644 --- a/crates/reddb-server/src/runtime/transaction_state.rs +++ b/crates/reddb-server/src/runtime/transaction_state.rs @@ -52,20 +52,25 @@ impl TransactionState { xid } - pub(crate) fn commit( + pub(crate) fn commit( &self, connection_id: u64, - prepare: impl FnOnce(&TxnContext) -> Result<(), E>, + prepare: impl FnOnce(&TxnContext) -> Result, ) -> Result, E> { self.local_tenants.write().remove(&connection_id); let Some(context) = self.contexts.write().remove(&connection_id) else { return Ok(None); }; - if let Err(error) = prepare(&context) { - self.rollback_context(&context); - return Err(error); - } + // A writer can return its publication guard after validation and WAL + // append. Keep it through the transaction's visibility transition. + let _publication = match prepare(&context) { + Ok(publication) => publication, + Err(error) => { + self.rollback_context(&context); + return Err(error); + } + }; for (_, xid) in &context.savepoints { self.snapshot_manager.commit(*xid); diff --git a/crates/reddb-server/src/storage/engine/btree/impl.rs b/crates/reddb-server/src/storage/engine/btree/impl.rs index b2eb10781..2fa900886 100644 --- a/crates/reddb-server/src/storage/engine/btree/impl.rs +++ b/crates/reddb-server/src/storage/engine/btree/impl.rs @@ -961,14 +961,24 @@ impl BTree { Ok((new_page, separator)) } - fn rebalance_leaf(&self, leaf_id: u32, path: Vec) -> BTreeResult<()> { + fn rebalance_leaf(&self, leaf_id: u32, mut path: Vec) -> BTreeResult<()> { + // Finish the leaf step before ascending so its page buffers do not + // accumulate with interior buffers on ordinary worker stacks. + if let Some(parent_id) = self.rebalance_leaf_step(leaf_id, &path)? { + path.pop(); + self.rebalance_interior(parent_id, path)?; + } + Ok(()) + } + + fn rebalance_leaf_step(&self, leaf_id: u32, path: &[u32]) -> BTreeResult> { if path.is_empty() { - return Ok(()); + return Ok(None); } let root_id = self.root_page_id(); if leaf_id == root_id { - return Ok(()); + return Ok(None); } let mut leaf = self.pager.read_page(leaf_id)?; @@ -1000,7 +1010,7 @@ impl BTree { } if leaf_entries_size(&leaf_entries) >= min_bytes { - return Ok(()); + return Ok(None); } if child_index > 0 { @@ -1037,7 +1047,7 @@ impl BTree { self.pager.write_page(parent_id, parent)?; } - return Ok(()); + return Ok(None); } } @@ -1076,7 +1086,7 @@ impl BTree { self.pager.write_page(parent_id, parent)?; } - return Ok(()); + return Ok(None); } } @@ -1107,9 +1117,7 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - let mut parent_path = path; - parent_path.pop(); - return self.rebalance_interior(parent_id, parent_path); + return Ok(Some(parent_id)); } if child_index + 1 < parent_children.len() { @@ -1139,15 +1147,25 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - let mut parent_path = path; - parent_path.pop(); - return self.rebalance_interior(parent_id, parent_path); + return Ok(Some(parent_id)); } + Ok(None) + } + + fn rebalance_interior(&self, mut node_id: u32, mut path: Vec) -> BTreeResult<()> { + // Each completed step releases its page buffers before the next level. + while let Some(parent_id) = self.rebalance_interior_step(node_id, &mut path)? { + node_id = parent_id; + } Ok(()) } - fn rebalance_interior(&self, node_id: u32, mut path: Vec) -> BTreeResult<()> { + fn rebalance_interior_step( + &self, + node_id: u32, + path: &mut Vec, + ) -> BTreeResult> { let root_id = self.root_page_id(); let mut node = self.pager.read_page(node_id)?; let (mut node_keys, mut node_children) = read_interior_keys_children(&node)?; @@ -1163,16 +1181,16 @@ impl BTree { )) })? = next_root; } - return Ok(()); + return Ok(None); } if interior_entries_size(&node_keys) >= min_bytes { - return Ok(()); + return Ok(None); } let parent_id = match path.pop() { Some(id) => id, - None => return Ok(()), + None => return Ok(None), }; let mut parent = self.pager.read_page(parent_id)?; let (mut parent_keys, mut parent_children) = read_interior_keys_children(&parent)?; @@ -1216,7 +1234,7 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - return Ok(()); + return Ok(None); } } } @@ -1249,7 +1267,7 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - return Ok(()); + return Ok(None); } } } @@ -1274,7 +1292,7 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - return self.rebalance_interior(parent_id, path); + return Ok(Some(parent_id)); } if child_index + 1 < parent_children.len() { @@ -1297,9 +1315,9 @@ impl BTree { parent.update_checksum(); self.pager.write_page(parent_id, parent)?; - return self.rebalance_interior(parent_id, path); + return Ok(Some(parent_id)); } - Ok(()) + Ok(None) } } diff --git a/crates/reddb-server/src/storage/query/planner/stats_catalog.rs b/crates/reddb-server/src/storage/query/planner/stats_catalog.rs index acc673c6c..e32b28fae 100644 --- a/crates/reddb-server/src/storage/query/planner/stats_catalog.rs +++ b/crates/reddb-server/src/storage/query/planner/stats_catalog.rs @@ -1,4 +1,5 @@ use std::collections::{BTreeSet, HashMap, HashSet}; +use std::hash::{DefaultHasher, Hash, Hasher}; use std::sync::Arc; use super::cost::{ColumnStats, TableStats}; @@ -14,6 +15,8 @@ const DEFAULT_ROWS_PER_PAGE: u64 = 100; const DEFAULT_SAMPLE_TARGET: usize = 30_000; const DEFAULT_BUCKETS: usize = 100; const DEFAULT_MCV_SIZE: usize = 100; +const MAX_STAT_VALUE_BYTES: usize = 4096; +const MAX_HISTOGRAM_SAMPLE_BYTES: usize = 1024 * 1024; #[derive(Debug, Clone)] pub(crate) struct AnalyzedColumnStats { @@ -44,12 +47,43 @@ pub(crate) struct PersistedPlannerStats { pub(crate) fn analyze_collection(store: &UnifiedStore, table: &str) -> Option { let manager = store.get_collection(table)?; - let entities = manager - .query_all(|_| true) + let mut row_count = 0usize; + let mut total_bytes = 0usize; + let mut names = BTreeSet::new(); + manager.for_each_entity(|entity| { + row_count += 1; + for_each_entity_field(entity, |name, value| { + if !names.contains(name) { + names.insert(name.to_string()); + } + total_bytes = total_bytes.saturating_add(name.len() + approximate_value_size(value)); + }); + true + }); + let samples = sample_indices(row_count, DEFAULT_SAMPLE_TARGET); + let columns = names .into_iter() - .map(|entity| (entity.id, fields_from_entity(&entity))) - .collect::>(); - Some(analyze_entity_fields(table, &entities)) + .map(|column| { + analyze_column_values(&column, &samples, |visit| { + manager.for_each_entity(|entity| { + let value = match &entity.data { + EntityData::Row(row) => row.get_field(&column), + EntityData::Node(node) => node.properties.get(&column), + _ => None, + }; + visit(value); + true + }); + }) + }) + .collect(); + Some(AnalyzedTableStats { + table: table.to_string(), + row_count: row_count as u64, + avg_row_size: total_bytes.checked_div(row_count).unwrap_or(0) as u64, + page_count: (row_count as u64 / DEFAULT_ROWS_PER_PAGE).max(1), + columns, + }) } pub(crate) fn analyze_entity_fields( @@ -262,20 +296,57 @@ fn analyze_column( column: &str, entities: &[(EntityId, Vec<(String, Value)>)], sample_indices: &[usize], +) -> AnalyzedColumnStats { + analyze_column_values(column, sample_indices, |visit| { + for (_, fields) in entities { + visit(field_value(fields, column)); + } + }) +} + +fn analyze_column_values( + column: &str, + sample_indices: &[usize], + scan: impl FnOnce(&mut dyn FnMut(Option<&Value>)), ) -> AnalyzedColumnStats { let mut null_count = 0u64; - let mut distinct = HashSet::new(); + let mut distinct = Some(HashSet::::new()); + let mut cardinality = crate::storage::primitives::HyperLogLog::new(); let mut min_key: Option = None; let mut min_value: Option = None; let mut max_key: Option = None; let mut max_value: Option = None; + let mut bounds_fit = true; + let mut sample_for_hist = Vec::new(); + let mut sample_counts = HashMap::::new(); + let mut histogram_bytes = 0usize; + let mut row = 0usize; - for (_, fields) in entities { - match field_value(fields, column) { + scan(&mut |value| { + let sampled = sample_indices.binary_search(&row).is_ok(); + row += 1; + match value { Some(Value::Null) | None => null_count += 1, Some(value) => { - distinct.insert(value.clone()); - if let Some(key) = value_to_canonical_key(value) { + let mut hash = DefaultHasher::new(); + value.hash(&mut hash); + let fingerprint = hash.finish(); + cardinality.add(&fingerprint.to_le_bytes()); + if let Some(values) = distinct.as_mut() { + values.insert(fingerprint); + if values.len() > DEFAULT_SAMPLE_TARGET { + distinct = None; + } + } + let bytes = approximate_value_size(value); + if bytes > MAX_STAT_VALUE_BYTES { + bounds_fit = false; + min_key = None; + min_value = None; + max_key = None; + max_value = None; + } + if let Some(key) = bounds_fit.then(|| value_to_canonical_key(value)).flatten() { if min_key.as_ref().is_none_or(|current| key < *current) { min_key = Some(key.clone()); min_value = Some(value.clone()); @@ -285,24 +356,16 @@ fn analyze_column( max_value = Some(value.clone()); } } + if sampled && bytes <= MAX_HISTOGRAM_SAMPLE_BYTES.saturating_sub(histogram_bytes) { + if let Some(hist_value) = value_to_histogram_value(value) { + histogram_bytes += bytes; + sample_for_hist.push(hist_value.clone()); + *sample_counts.entry(hist_value).or_insert(0) += 1; + } + } } } - } - - let mut sample_for_hist = Vec::new(); - let mut sample_counts = HashMap::::new(); - for idx in sample_indices { - if let Some(value) = entities - .get(*idx) - .and_then(|(_, fields)| field_value(fields, column)) - .filter(|value| !matches!(value, Value::Null)) - { - if let Some(hist_value) = value_to_histogram_value(value) { - sample_for_hist.push(hist_value.clone()); - *sample_counts.entry(hist_value).or_insert(0) += 1; - } - } - } + }); let histogram = if sample_for_hist.is_empty() { None @@ -330,7 +393,7 @@ fn analyze_column( AnalyzedColumnStats { name: column.to_string(), - distinct_count: distinct.len() as u64, + distinct_count: distinct.map_or_else(|| cardinality.count(), |values| values.len() as u64), null_count, min_value, max_value, @@ -366,33 +429,26 @@ fn sample_indices(total_rows: usize, sample_target: usize) -> Vec { reservoir.into_sorted_indices() } -fn fields_from_entity(entity: &UnifiedEntity) -> Vec<(String, Value)> { +fn for_each_entity_field(entity: &UnifiedEntity, mut visit: impl FnMut(&str, &Value)) { match &entity.data { - EntityData::Row(row) => named_fields_from_row(row), - EntityData::Node(node) => node - .properties - .iter() - .map(|(key, value)| (key.clone(), value.clone())) - .collect(), - _ => Vec::new(), - } -} - -fn named_fields_from_row(row: &RowData) -> Vec<(String, Value)> { - if let Some(named) = &row.named { - return named - .iter() - .map(|(key, value)| (key.clone(), value.clone())) - .collect(); - } - if let Some(schema) = &row.schema { - return schema - .iter() - .zip(row.columns.iter()) - .map(|(key, value)| (key.clone(), value.clone())) - .collect(); + EntityData::Row(row) => { + if let Some(named) = &row.named { + for (name, value) in named { + visit(name, value); + } + } else if let Some(schema) = &row.schema { + for (name, value) in schema.iter().zip(&row.columns) { + visit(name, value); + } + } + } + EntityData::Node(node) => { + for (name, value) in &node.properties { + visit(name, value); + } + } + _ => {} } - Vec::new() } fn field_value<'a>(fields: &'a [(String, Value)], column: &str) -> Option<&'a Value> { @@ -673,3 +729,91 @@ fn approximate_value_size(value: &Value) -> usize { Value::IpAddr(_) => 16, } } + +#[cfg(test)] +mod streaming_tests { + use super::*; + + #[test] + fn small_column_statistics_preserve_counts_and_bounds() { + let rows = (0..10) + .map(|id| { + ( + EntityId::new(id + 1), + vec![ + ("id".to_string(), Value::Integer(id as i64)), + ( + "optional".to_string(), + if id % 2 == 0 { + Value::Null + } else { + Value::Integer(1) + }, + ), + ], + ) + }) + .collect::>(); + let stats = analyze_entity_fields("small_rows", &rows); + assert_eq!(stats.row_count, 10); + let ids = stats + .columns + .iter() + .find(|column| column.name == "id") + .expect("id stats"); + assert_eq!(ids.distinct_count, 10); + assert_eq!(ids.min_value, Some(Value::Integer(0))); + assert_eq!(ids.max_value, Some(Value::Integer(9))); + let optional = stats + .columns + .iter() + .find(|column| column.name == "optional") + .expect("optional stats"); + assert_eq!(optional.null_count, 5); + assert_eq!(optional.distinct_count, 1); + } + + #[test] + fn high_cardinality_uses_the_bounded_estimator() { + let rows = (0..50_000) + .map(|id| { + ( + EntityId::new(id + 1), + vec![("id".to_string(), Value::Integer(id as i64))], + ) + }) + .collect::>(); + let stats = analyze_entity_fields("distinct_rows", &rows); + let estimate = stats.columns[0].distinct_count; + assert!( + (47_500..=52_500).contains(&estimate), + "unexpected cardinality estimate {estimate}" + ); + } + + #[test] + fn wide_values_leave_bounds_unknown_and_samples_bounded() { + let value = Value::text("x".repeat(8192)); + let rows = (0..1000) + .map(|id| { + ( + EntityId::new(id + 1), + vec![("wide".to_string(), value.clone())], + ) + }) + .collect::>(); + let stats = analyze_entity_fields("wide_rows", &rows); + let column = &stats.columns[0]; + assert!(column.min_value.is_none()); + assert!(column.max_value.is_none()); + assert_eq!(column.distinct_count, 1); + assert!( + column + .histogram + .as_ref() + .expect("histogram has a bounded sample") + .total_count + <= 128 + ); + } +} diff --git a/crates/reddb-server/src/storage/transaction/snapshot.rs b/crates/reddb-server/src/storage/transaction/snapshot.rs index 5b11c7237..cb13fd0c9 100644 --- a/crates/reddb-server/src/storage/transaction/snapshot.rs +++ b/crates/reddb-server/src/storage/transaction/snapshot.rs @@ -133,7 +133,7 @@ impl TxnContext { /// `begin()/commit()` pair (which also leaves the xid in neither set). pub struct SnapshotManager { next_xid: AtomicU64, - aborted_revision: AtomicU64, + visibility_revision: AtomicU64, state: parking_lot::RwLock, /// Reservation window for the autocommit pool. A single /// `parking_lot::Mutex` protects two `u64`s — `next` (next xid to @@ -212,7 +212,7 @@ impl SnapshotManager { Self { // Start at 1 so xid=0 keeps its pre-MVCC "everyone sees it" meaning. next_xid: AtomicU64::new(1), - aborted_revision: AtomicU64::new(0), + visibility_revision: AtomicU64::new(0), state: parking_lot::RwLock::new(ManagerState::default()), // Pool starts empty — first caller triggers a refill. autocommit_pool: parking_lot::Mutex::new(AutocommitPool::default()), @@ -240,6 +240,7 @@ impl SnapshotManager { let mut state = self.state.write(); let xid = self.next_xid.fetch_add(1, Ordering::Relaxed); state.active.insert(xid); + self.visibility_revision.fetch_add(1, Ordering::Release); xid } @@ -339,12 +340,13 @@ impl SnapshotManager { /// future snapshots; earlier snapshots keep their own view. pub fn commit(&self, xid: Xid) { let mut state = self.state.write(); - state.active.remove(&xid); + let was_active = state.active.remove(&xid); state.snapshot_horizons.remove(&xid); // Also clear from aborted set in case of prior rollback_to call // that touched this xid (defensive; normally a no-op). - if state.aborted.remove(&xid) { - self.aborted_revision.fetch_add(1, Ordering::Release); + let was_aborted = state.aborted.remove(&xid); + if was_active || was_aborted { + self.visibility_revision.fetch_add(1, Ordering::Release); } if let Some(ssi) = state.serializable.get_mut(&xid) { ssi.committed = true; @@ -400,9 +402,10 @@ impl SnapshotManager { pub fn rollback(&self, xid: Xid) { let mut state = self.state.write(); state.snapshot_horizons.remove(&xid); - state.active.remove(&xid); - if state.aborted.insert(xid) { - self.aborted_revision.fetch_add(1, Ordering::Release); + let was_active = state.active.remove(&xid); + let newly_aborted = state.aborted.insert(xid); + if was_active || newly_aborted { + self.visibility_revision.fetch_add(1, Ordering::Release); } state.serializable.remove(&xid); for ssi in state.serializable.values_mut() { @@ -539,12 +542,13 @@ impl SnapshotManager { self.state.read().aborted.contains(&xid) } - pub(crate) fn checkpoint_aborted_state(&self) -> (u64, HashSet) { + /// Unfinished writers must be rolled back in a crash-recoverable checkpoint, + /// without changing their live transaction state or their readers. + pub(crate) fn checkpoint_excluded_state(&self) -> (u64, HashSet) { let state = self.state.read(); - ( - self.aborted_revision.load(Ordering::Acquire), - state.aborted.clone(), - ) + let mut excluded = state.aborted.clone(); + excluded.extend(state.active.iter().copied()); + (self.visibility_revision.load(Ordering::Acquire), excluded) } /// Is this xid still active? @@ -647,7 +651,7 @@ impl SnapshotManager { let previous = aborted.len(); aborted.retain(|&x| x >= below || pinned.contains_key(&x)); if previous != aborted.len() { - self.aborted_revision.fetch_add(1, Ordering::Release); + self.visibility_revision.fetch_add(1, Ordering::Release); } } diff --git a/crates/reddb-server/src/storage/unified/manager.rs b/crates/reddb-server/src/storage/unified/manager.rs index f1ee4f586..5b172172a 100644 --- a/crates/reddb-server/src/storage/unified/manager.rs +++ b/crates/reddb-server/src/storage/unified/manager.rs @@ -22,8 +22,8 @@ use std::sync::{Arc, Weak}; use super::entity::{EntityId, EntityKind, UnifiedEntity}; use super::metadata::{Metadata, MetadataFilter}; use super::segment::{ - GraphEntityKind, GrowingSegment, SegmentConfig, SegmentError, SegmentId, SegmentState, - SegmentStats, UnifiedSegment, ZoneColPred, ZoneColPredKind, + GraphEntityKind, GrowingSegment, SegmentConfig, SegmentError, SegmentId, SegmentScanCursor, + SegmentState, SegmentStats, UnifiedSegment, ZoneColPred, ZoneColPredKind, }; use super::visibility_map::VisibilityMap; use crate::runtime::mvcc::{entity_visible_with_context, SnapshotContext}; @@ -185,6 +185,19 @@ struct PhysicalRootCacheEntry { root: u64, } +/// Resumable physical maintenance positions. Weak sources let consolidation +/// release retired payloads; a later sweep revisits any replacement segments. +pub(crate) struct PhysicalScanCursor { + sources: Vec<(Weak>, SegmentScanCursor)>, + source: usize, +} + +impl PhysicalScanCursor { + pub(crate) fn finished(&self) -> bool { + self.source == self.sources.len() + } +} + /// Segment manager for a collection pub struct SegmentManager { /// Statement-scoped constraint admission. Reentrant because SQL upserts @@ -289,6 +302,20 @@ impl SegmentManager { self.column_schema.read().clone() } + // Positional file records omit the shared column names. Restore them on + // owned reads before callers update fields or rebuild secondary indexes. + // Existing per-row names remain authoritative; segment contents stay raw. + fn restore_owned_row_schemas<'a>(&self, entities: impl Iterator) { + let schema = self.column_schema(); + for entity in entities { + if let super::entity::EntityData::Row(row) = &mut entity.data { + if row.named.is_none() && row.schema.is_none() { + row.schema = schema.clone(); + } + } + } + } + pub(crate) fn set_column_schema_if_empty(&self, columns: Vec) { if columns.is_empty() { return; @@ -596,7 +623,9 @@ impl SegmentManager { /// Get an entity by ID — scans growing then sealed segments. pub fn get(&self, id: EntityId) -> Option { - self.get_with(id, Clone::clone) + let mut entity = self.get_with(id, Clone::clone)?; + self.restore_owned_row_schemas(std::iter::once(&mut entity)); + Some(entity) } /// Inspect under the segment lock, for example to bound a payload clone. @@ -773,6 +802,7 @@ impl SegmentManager { } if remaining.is_empty() { + self.restore_owned_row_schemas(out.iter_mut().flatten()); return out; } @@ -793,6 +823,8 @@ impl SegmentManager { }); } + drop(sealed); + self.restore_owned_row_schemas(out.iter_mut().flatten()); out } @@ -1938,6 +1970,60 @@ impl SegmentManager { }); } + pub(crate) fn physical_scan_cursor(&self) -> PhysicalScanCursor { + let growing = self.growing.read(); + let sealed = self.sealed.read(); + PhysicalScanCursor { + sources: growing + .iter() + .chain(sealed.iter()) + .map(|segment| (Arc::downgrade(segment), segment.read().scan_cursor())) + .collect(), + source: 0, + } + } + + /// At most one small ID batch; every inspected slot or retired source + /// consumes work, including deleted entries that produce no candidate. + pub(crate) fn physical_scan_batch( + &self, + cursor: &mut PhysicalScanCursor, + work_left: &mut usize, + ) -> Vec { + let mut ids = Vec::with_capacity(super::segment::SCAN_BATCH_SIZE.min(*work_left)); + while !cursor.finished() && *work_left > 0 { + let (source, position) = &mut cursor.sources[cursor.source]; + if position.finished() { + *work_left -= 1; + cursor.source += 1; + continue; + } + let Some(segment) = source.upgrade() else { + *work_left -= 1; + cursor.source += 1; + continue; + }; + segment.read().scan_batch( + position, + &mut || { + if *work_left == 0 { + return false; + } + *work_left -= 1; + true + }, + &mut |id| ids.push(id), + ); + if position.finished() { + cursor.source += 1; + } + if !ids.is_empty() { + break; + } + } + ids + } + /// Capture segment membership and append boundaries, then hydrate bounded /// batches. The consumer runs outside topology and segment locks, so scoring /// and RLS may re-enter storage and writers can progress between batches. @@ -2061,7 +2147,9 @@ impl SegmentManager { where F: Fn(&UnifiedEntity) -> bool + Sync, { - self.query_all_in_segments(|_| true, filter) + let mut entities = self.query_all_in_segments(|_| true, filter); + self.restore_owned_row_schemas(entities.iter_mut()); + entities } fn query_all_in_segments( diff --git a/crates/reddb-server/src/storage/unified/store.rs b/crates/reddb-server/src/storage/unified/store.rs index 3a9639e01..1aafb2ed6 100644 --- a/crates/reddb-server/src/storage/unified/store.rs +++ b/crates/reddb-server/src/storage/unified/store.rs @@ -291,8 +291,8 @@ impl StoreStats { struct CheckpointVersion { manager: std::sync::Weak, segments: Vec<(u64, u64)>, - aborted_revision: u64, - aborted: Arc>, + visibility_revision: u64, + excluded_xids: Arc>, pending_vault: Arc>, format_version: u32, } @@ -423,6 +423,8 @@ mod impl_native_c; mod impl_pages; mod impl_vault_import; mod keyed_write; +mod mvcc_reclaim; +pub(crate) use mvcc_reclaim::MvccReclaimCursor; mod native_helpers; pub use self::builder::EntityBuilder; diff --git a/crates/reddb-server/src/storage/unified/store/impl_entities.rs b/crates/reddb-server/src/storage/unified/store/impl_entities.rs index 09a2ffeb1..3066d7ecf 100644 --- a/crates/reddb-server/src/storage/unified/store/impl_entities.rs +++ b/crates/reddb-server/src/storage/unified/store/impl_entities.rs @@ -83,7 +83,7 @@ impl UnifiedStore { .collect(); serialized.sort_by(|a, b| a.0.cmp(&b.0)); - if !skip_btree { + if !skip_btree && !Self::deferred_store_wal_capture_active() { let Some(pager) = &self.pager else { let records: Vec> = serialized.into_iter().map(|(_id, record)| record).collect(); @@ -165,7 +165,7 @@ impl UnifiedStore { // u64 IDs encoded as big-endian — lex order = numeric order. serialized.sort_by(|a, b| a.0.cmp(&b.0)); - if !skip_btree { + if !skip_btree && !Self::deferred_store_wal_capture_active() { let Some(pager) = &self.pager else { let records: Vec> = serialized.into_iter().map(|(_id, record)| record).collect(); @@ -561,9 +561,14 @@ impl UnifiedStore { self.update_graph_label_index(collection, label, id); } - // Also insert into B-tree index if pager is active + // Captured transactions stay WAL-only until commit; pager flushes must + // never make an unfinished row durable. let mut registry_dirty = false; - if let Some(pager) = &self.pager { + if let Some(pager) = self + .pager + .as_ref() + .filter(|_| !Self::deferred_store_wal_capture_active()) + { if let Some(entity) = manager.get(id) { let mut btree_indices = self.btree_indices.write(); let btree = btree_indices @@ -625,6 +630,9 @@ impl UnifiedStore { collection: &str, mut entities: Vec, ) -> Result, StoreError> { + // A transaction can begin after checkpoint visibility is frozen. + // Fence the entire row batch, including its captured or committed WAL. + let _publication = self.vault_publication_lock.lock(); // REDDB_BULK_TIMING=1 prints a per-call breakdown of the bulk // insert path to stderr. Off by default — used by the reddb // benchmark harness to locate ingest bottlenecks. @@ -784,7 +792,7 @@ impl UnifiedStore { let mut t_btree_lock = std::time::Duration::ZERO; let mut t_btree_insert = std::time::Duration::ZERO; let mut t_flush = std::time::Duration::ZERO; - if !skip_btree { + if !skip_btree && !Self::deferred_store_wal_capture_active() { if let (Some(pager), Some(batch)) = (&self.pager, serialized.as_ref()) { let t0 = std::time::Instant::now(); let mut btree_indices = self.btree_indices.write(); @@ -890,7 +898,7 @@ impl UnifiedStore { // both or neither. Otherwise the rebuild re-inserts this row from the // manager and one of the two B-tree inserts fails on a duplicate key, // or lands mid-rebuild and tears the tree. - let paged_btree = if persist_record { + let paged_btree = if persist_record && !Self::deferred_store_wal_capture_active() { self.get_or_create_btree(collection) } else { None @@ -1053,22 +1061,24 @@ impl UnifiedStore { let manager = self .get_collection(collection) .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?; - let entities = - manager.query_all(|entity| matches!(entity.kind, EntityKind::TableRow { .. })); let mut logical = HashMap::::new(); - for entity in &entities { + manager.for_each_entity(|entity| { + if !matches!(entity.kind, EntityKind::TableRow { .. }) { + return true; + } let entry = logical.entry(entity.logical_id()).or_insert((false, 0)); if entity.xmax == 0 { entry.0 = true; } entry.1 = entry.1.max(entity.xmin); - } + true + }); let mut stats = MvccVacuumStats::default(); let mut reclaim_ids = Vec::new(); - for entity in entities { - if entity.xmax == 0 { - continue; + manager.for_each_entity(|entity| { + if !matches!(entity.kind, EntityKind::TableRow { .. }) || entity.xmax == 0 { + return true; } stats.scanned_versions += 1; let (has_live_version, max_xmin) = logical @@ -1092,7 +1102,8 @@ impl UnifiedStore { stats.retained_history_versions += 1; } } - } + true + }); if !reclaim_ids.is_empty() { self.delete_batch(collection, &reclaim_ids)?; @@ -1261,6 +1272,9 @@ impl UnifiedStore { if ids.is_empty() { return Ok(Vec::new()); } + // Physical reclamation must publish its mutation and WAL record under + // the same fence as checkpoint snapshots and WAL truncation. + let _publication = self.vault_publication_lock.lock(); // Sharded invalidation. The bounded LRU's shard write lock is // only acquired when the shard actually carries one of these ids, @@ -1272,19 +1286,22 @@ impl UnifiedStore { .get_collection(collection) .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?; + let before = self.pager.as_ref().map(|_| manager.checkpoint_revisions()); let deleted_ids = manager.delete_batch(ids)?; if deleted_ids.is_empty() { return Ok(deleted_ids); } let mut registry_dirty = false; + let mut updated_tree = false; if self.pager.is_some() { let btree_indices = self.btree_indices.read(); if let Some(btree) = btree_indices.get(collection) { + updated_tree = true; let root_before = btree.root_page_id(); for id in &deleted_ids { let key = id.raw().to_be_bytes(); - let _ = btree.delete(&key); + updated_tree &= matches!(btree.delete(&key), Ok(true)); } registry_dirty = root_before != btree.root_page_id(); } @@ -1304,6 +1321,11 @@ impl UnifiedStore { .collect::>(); self.finish_paged_write(actions)?; + if updated_tree { + if let Some(before) = before { + self.note_paged_delete(collection, &manager, &before, &deleted_ids); + } + } Ok(deleted_ids) } diff --git a/crates/reddb-server/src/storage/unified/store/impl_file.rs b/crates/reddb-server/src/storage/unified/store/impl_file.rs index f4d75a0b5..c64d88270 100644 --- a/crates/reddb-server/src/storage/unified/store/impl_file.rs +++ b/crates/reddb-server/src/storage/unified/store/impl_file.rs @@ -218,6 +218,10 @@ impl UnifiedStore { pub(crate) fn to_binary_dump_bytes(&self) -> Vec { let _vault_guard = self.vault_publication_lock.lock(); let pending_vault_versions = self.pending_vault_versions.read(); + let (_, excluded) = self.snapshot_manager.get().map_or_else( + || (0, Default::default()), + |manager| manager.checkpoint_excluded_state(), + ); let mut buf = Vec::new(); // Version 9 includes explicit table-row logical identity plus MVCC @@ -238,24 +242,16 @@ impl UnifiedStore { return false; } !matches!(entity.data, EntityData::Row(_) | EntityData::Vector(_)) - || !self - .snapshot_manager - .get() - .is_some_and(|manager| manager.is_aborted(entity.xmin)) + || !excluded.contains(&entity.xmin) }); reddb_file::encode_native_dump_collection_header(&mut buf, name, entities.len() as u32); // V7+: serialize entity+metadata as a length-prefixed record. // Each record: [u32 len][serialize_entity_record bytes] for mut entity in entities { - // Rolled-back deletions must not become committed tombstones - // when a new runtime reconstructs its xid floor from the dump. - if matches!(entity.data, EntityData::Row(_)) - && self - .snapshot_manager - .get() - .is_some_and(|manager| manager.is_aborted(entity.xmax)) - { + // Unfinished and rolled-back deletions cannot become committed + // tombstones when recovery reconstructs the xid floor. + if matches!(entity.data, EntityData::Row(_)) && excluded.contains(&entity.xmax) { entity.xmax = 0; } let metadata = manager.get_metadata(entity.id); diff --git a/crates/reddb-server/src/storage/unified/store/impl_pages.rs b/crates/reddb-server/src/storage/unified/store/impl_pages.rs index e9dcd616d..a0550e2b9 100644 --- a/crates/reddb-server/src/storage/unified/store/impl_pages.rs +++ b/crates/reddb-server/src/storage/unified/store/impl_pages.rs @@ -72,7 +72,8 @@ fn checkpoint_visibility_matches( current: &CheckpointVersion, manager: &SegmentManager, ) -> bool { - if previous.aborted_revision == current.aborted_revision + if (previous.visibility_revision == current.visibility_revision + || previous.excluded_xids == current.excluded_xids) && previous.pending_vault == current.pending_vault { return true; @@ -84,15 +85,15 @@ fn checkpoint_visibility_matches( let skipped = |version: &CheckpointVersion| { version.pending_vault.contains(&entity.id) || (matches!(entity.data, EntityData::Row(_) | EntityData::Vector(_)) - && version.aborted.contains(&entity.xmin)) + && version.excluded_xids.contains(&entity.xmin)) }; let previous_skipped = skipped(previous); let current_skipped = skipped(current); matches = previous_skipped == current_skipped && (current_skipped || !matches!(entity.data, EntityData::Row(_)) - || previous.aborted.contains(&entity.xmax) - == current.aborted.contains(&entity.xmax)); + || previous.excluded_xids.contains(&entity.xmax) + == current.excluded_xids.contains(&entity.xmax)); matches }); matches @@ -535,7 +536,54 @@ impl UnifiedStore { && previous.segments == before && previous.format_version == self.format_version() && !previous.pending_vault.contains(&id) - && !previous.aborted.contains(&0) + && !previous.excluded_xids.contains(&0) + { + previous.segments = after; + } + } + + /// Successful physical deletes already changed both the manager and tree. + /// Reuse remains valid only when their exact revision delta accounts for + /// this batch, with no intervening edits or segment replacement. + pub(crate) fn note_paged_delete( + &self, + collection: &str, + manager: &Arc, + before: &[(u64, u64)], + deleted: &[EntityId], + ) { + let after = manager.checkpoint_revisions(); + if before.len() != after.len() { + return; + } + let mut changes = 0_u64; + for ((old_id, old_revision), (new_id, new_revision)) in before.iter().zip(&after) { + if old_id != new_id { + return; + } + let Some(delta) = new_revision.checked_sub(*old_revision) else { + return; + }; + let Some(total) = changes.checked_add(delta) else { + return; + }; + changes = total; + } + let Ok(expected) = u64::try_from(deleted.len()) else { + return; + }; + if changes != expected { + return; + } + let mut versions = self.checkpoint_versions.write(); + let Some(previous) = versions.get_mut(collection) else { + return; + }; + if previous.manager.ptr_eq(&Arc::downgrade(manager)) + && previous.segments == before + && previous.format_version == self.format_version() + && !deleted.iter().any(|id| previous.pending_vault.contains(id)) + && !previous.excluded_xids.contains(&0) { previous.segments = after; } @@ -855,8 +903,8 @@ impl UnifiedStore { CheckpointVersion { manager: Arc::downgrade(manager), segments: manager.checkpoint_revisions(), - aborted_revision: 0, - aborted: Arc::default(), + visibility_revision: 0, + excluded_xids: Arc::default(), pending_vault: Arc::default(), format_version: self.format_version(), }, @@ -1032,8 +1080,8 @@ impl UnifiedStore { CheckpointVersion { manager: Arc::downgrade(manager), segments: manager.checkpoint_revisions(), - aborted_revision: 0, - aborted: Arc::default(), + visibility_revision: 0, + excluded_xids: Arc::default(), pending_vault: Arc::default(), format_version, }, @@ -1244,17 +1292,17 @@ impl UnifiedStore { let previous_versions = self.checkpoint_versions.read(); let mut checkpoint_versions = HashMap::with_capacity(collections.len()); let pending_vault = Arc::new(pending_vault_versions.clone()); - let (aborted_revision, aborted) = self.snapshot_manager.get().map_or_else( + let (visibility_revision, excluded) = self.snapshot_manager.get().map_or_else( || (0, Default::default()), - |manager| manager.checkpoint_aborted_state(), + |manager| manager.checkpoint_excluded_state(), ); - let aborted = Arc::new(aborted); + let excluded = Arc::new(excluded); for (name, manager) in collections.iter() { let version = CheckpointVersion { manager: Arc::downgrade(manager), segments: manager.checkpoint_revisions(), - aborted_revision, - aborted: Arc::clone(&aborted), + visibility_revision, + excluded_xids: Arc::clone(&excluded), pending_vault: Arc::clone(&pending_vault), format_version: self.format_version(), }; @@ -1282,12 +1330,13 @@ impl UnifiedStore { CHECKPOINT_SERIALIZED_ROWS.with(|count| count.set(count.get() + 1)); if pending_vault_versions.contains(&entity.id) || (matches!(entity.data, EntityData::Row(_) | EntityData::Vector(_)) - && aborted.contains(&entity.xmin)) + && excluded.contains(&entity.xmin)) { return None; } let mut normalized = std::borrow::Cow::Borrowed(entity); - if matches!(entity.data, EntityData::Row(_)) && aborted.contains(&entity.xmax) { + if matches!(entity.data, EntityData::Row(_)) && excluded.contains(&entity.xmax) + { normalized.to_mut().xmax = 0; } Some(Self::serialize_entity_record( diff --git a/crates/reddb-server/src/storage/unified/store/mvcc_reclaim.rs b/crates/reddb-server/src/storage/unified/store/mvcc_reclaim.rs new file mode 100644 index 000000000..255c37fef --- /dev/null +++ b/crates/reddb-server/src/storage/unified/store/mvcc_reclaim.rs @@ -0,0 +1,184 @@ +//! Bounded physical table-history reclamation; payloads stay borrowed. +use super::*; +use crate::storage::transaction::SnapshotManager; +use crate::storage::unified::manager::PhysicalScanCursor; + +pub(crate) struct MvccReclaimCursor { + pub(crate) collection: String, + manager: Arc, + scan: PhysicalScanCursor, + invalidated: bool, +} + +impl MvccReclaimCursor { + pub(crate) fn finished(&self) -> bool { + self.invalidated || self.scan.finished() + } +} + +impl UnifiedStore { + pub(crate) fn mvcc_reclaim_cursor(&self, collection: String) -> Option { + let manager = self.get_collection(&collection)?; + let scan = manager.physical_scan_cursor(); + Some(MvccReclaimCursor { + collection, + manager, + scan, + invalidated: false, + }) + } + + /// A busy checkpoint leaves the cursor untouched. Each successful tick + /// inspects at most `scan_budget` positions and deletes at most + /// `reclaim_budget` physical versions through the ordinary WAL-backed path. + pub(crate) fn reclaim_mvcc_tick( + &self, + cursor: &mut MvccReclaimCursor, + snapshots: &SnapshotManager, + cutoff: u64, + scan_budget: usize, + reclaim_budget: usize, + ) -> Result { + let Some(_publication) = self.vault_publication_lock.try_lock() else { + return Ok(0); + }; + let Some(manager) = self.get_collection(&cursor.collection) else { + cursor.invalidated = true; + return Ok(0); + }; + if !Arc::ptr_eq(&manager, &cursor.manager) { + cursor.invalidated = true; + return Ok(0); + } + let mut scan_left = scan_budget; + let mut reclaim = Vec::with_capacity(reclaim_budget); + while scan_left > 0 && reclaim.len() < reclaim_budget && !cursor.finished() { + let work = scan_left.min(reclaim_budget - reclaim.len()); + let mut work_left = work; + let ids = manager.physical_scan_batch(&mut cursor.scan, &mut work_left); + scan_left -= work - work_left; + manager.for_each_id(&ids, |_, entity| { + if !matches!(entity.kind, EntityKind::TableRow { .. }) { + return; + } + let aborted_insert = entity.xmin != 0 && snapshots.is_aborted(entity.xmin); + let obsolete = + entity.xmax != 0 && entity.xmax < cutoff && !snapshots.is_aborted(entity.xmax); + if aborted_insert || obsolete { + reclaim.push(entity.id); + } + }); + } + Ok(self.delete_batch(&cursor.collection, &reclaim)?.len()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{RedDBOptions, RedDBRuntime}; + + fn fixture() -> RedDBRuntime { + let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime opens"); + runtime + .execute_query("SET CONFIG runtime.mvcc.autovacuum.enabled = false") + .expect("disable automatic ticks"); + runtime + .execute_query("CREATE TABLE reclaim_rows (id INTEGER, n INTEGER)") + .expect("create table"); + runtime + } + + #[test] + fn collection_replacement_finishes_its_stale_cursor() { + let runtime = fixture(); + runtime + .execute_query("INSERT INTO reclaim_rows (id, n) VALUES (1, 0)") + .expect("seed row"); + let store = runtime.db().store(); + let mut cursor = store + .mvcc_reclaim_cursor("reclaim_rows".to_string()) + .expect("open cursor"); + runtime + .execute_query("DROP TABLE reclaim_rows") + .expect("drop source"); + runtime + .execute_query("CREATE TABLE reclaim_rows (id INTEGER, n INTEGER)") + .expect("replace source"); + runtime + .execute_query("INSERT INTO reclaim_rows (id, n) VALUES (2, 1)") + .expect("seed replacement"); + let reclaimed = store + .reclaim_mvcc_tick( + &mut cursor, + &runtime.snapshot_manager(), + runtime.mvcc_vacuum_cutoff_xid(), + 4096, + 256, + ) + .expect("invalidate stale cursor"); + assert_eq!(reclaimed, 0); + assert!( + cursor.finished(), + "collector would remain stuck on the replaced collection" + ); + assert_eq!( + runtime + .execute_query("SELECT n FROM reclaim_rows WHERE id = 2") + .expect("replacement survives") + .result + .records + .len(), + 1 + ); + } + + #[test] + fn reclamation_resumes_without_exceeding_its_delete_budget() { + let runtime = fixture(); + let values = (0..128) + .map(|id| format!("({id}, 0)")) + .collect::>() + .join(","); + runtime + .execute_query(&format!("INSERT INTO reclaim_rows (id, n) VALUES {values}")) + .expect("seed rows"); + runtime + .execute_query("UPDATE reclaim_rows SET n = 1") + .expect("create history"); + let store = runtime.db().store(); + let mut cursor = store + .mvcc_reclaim_cursor("reclaim_rows".to_string()) + .expect("open cursor"); + let mut total = 0; + for _ in 0..64 { + let reclaimed = store + .reclaim_mvcc_tick( + &mut cursor, + &runtime.snapshot_manager(), + runtime.mvcc_vacuum_cutoff_xid(), + 32, + 8, + ) + .expect("bounded tick succeeds"); + assert!(reclaimed <= 8); + total += reclaimed; + if cursor.finished() { + break; + } + } + assert!(cursor.finished()); + assert_eq!(total, 128); + assert_eq!( + runtime + .execute_query("SELECT COUNT(*) AS n FROM reclaim_rows WHERE n = 1") + .expect("live rows survive") + .result + .records[0] + .get("n") + .expect("count exists") + .to_string(), + "128" + ); + } +} diff --git a/crates/reddb-server/tests/automatic_history_reclamation.rs b/crates/reddb-server/tests/automatic_history_reclamation.rs new file mode 100644 index 000000000..139695d44 --- /dev/null +++ b/crates/reddb-server/tests/automatic_history_reclamation.rs @@ -0,0 +1,225 @@ +//! Idle maintenance must reclaim obsolete versions without an explicit VACUUM. +use std::time::{Duration, Instant}; + +use reddb_server::{RedDBOptions, RedDBRuntime}; + +fn physical_rows(runtime: &RedDBRuntime, table: &str) -> usize { + let mut count = 0; + runtime + .db() + .store() + .get_collection(table) + .expect("fixture collection exists") + .for_each_entity(|_| { + count += 1; + true + }); + count +} + +fn await_history_reclamation(runtime: &RedDBRuntime, table: &str, maximum: usize) { + let deadline = Instant::now() + Duration::from_secs(15); + while physical_rows(runtime, table) > maximum && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + let physical = physical_rows(runtime, table); + assert!( + physical <= maximum, + "idle maintenance retained {physical} versions, expected <= {maximum}" + ); +} + +fn value(runtime: &RedDBRuntime, connection: u64) -> String { + use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; + set_current_connection_id(connection); + let result = runtime.execute_query("SELECT n FROM reader_rows WHERE id = 1"); + clear_current_connection_id(); + result.expect("read row").result.records[0] + .get("n") + .expect("value exists") + .to_string() +} + +#[test] +fn fixed_logical_data_does_not_retain_every_update_forever() { + let runtime = + RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("in-memory runtime opens"); + runtime + .execute_query("CREATE TABLE churn_rows (id INTEGER, n INTEGER)") + .expect("create churn table"); + let values = (0..128) + .map(|id| format!("({id}, 0)")) + .collect::>() + .join(","); + runtime + .execute_query(&format!("INSERT INTO churn_rows (id, n) VALUES {values}")) + .expect("seed fixed logical data"); + for _ in 0..10 { + runtime + .execute_query("UPDATE churn_rows SET n = n + 1") + .expect("update fixed logical data"); + } + let manager = runtime + .db() + .store() + .get_collection("churn_rows") + .expect("churn table exists"); + let deadline = Instant::now() + Duration::from_secs(15); + while manager.query_all(|_| true).len() > 256 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + let physical = manager.query_all(|_| true).len(); + assert!( + physical <= 256, + "128 live rows retained {physical} physical versions after idle maintenance" + ); + let result = runtime + .execute_query("SELECT COUNT(*) AS total FROM churn_rows WHERE n = 10") + .expect("read current values"); + assert_eq!( + result.result.records[0] + .get("total") + .expect("count exists") + .to_string(), + "128" + ); +} + +#[test] +fn automatic_reclamation_preserves_an_active_snapshot_reader() { + use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; + let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime opens"); + runtime + .execute_query("CREATE TABLE reader_rows (id INTEGER, n INTEGER)") + .expect("create table"); + runtime + .execute_query("INSERT INTO reader_rows (id, n) VALUES (1, 0)") + .expect("seed row"); + set_current_connection_id(42); + let result = runtime.execute_query("BEGIN ISOLATION LEVEL SNAPSHOT"); + clear_current_connection_id(); + result.expect("reader begins"); + assert_eq!(value(&runtime, 42), "0"); + for _ in 0..20 { + runtime + .execute_query("UPDATE reader_rows SET n = n + 1 WHERE id = 1") + .expect("update row"); + } + std::thread::sleep(Duration::from_millis(600)); + assert_eq!( + value(&runtime, 42), + "0", + "background cleanup changed reader visibility" + ); + set_current_connection_id(42); + let result = runtime.execute_query("COMMIT"); + clear_current_connection_id(); + result.expect("reader completes"); + await_history_reclamation(&runtime, "reader_rows", 1); + assert_eq!(value(&runtime, 1), "20"); +} + +#[test] +fn disabled_collector_preserves_history_and_can_resume() { + let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime opens"); + runtime + .execute_query("SET CONFIG runtime.mvcc.autovacuum.enabled = false") + .expect("disable collector"); + runtime + .execute_query("CREATE TABLE reader_rows (id INTEGER, n INTEGER)") + .expect("create table"); + runtime + .execute_query("INSERT INTO reader_rows (id, n) VALUES (1, 0)") + .expect("seed row"); + runtime + .execute_query("UPDATE reader_rows SET n = 1") + .expect("update row"); + std::thread::sleep(Duration::from_millis(600)); + assert_eq!(physical_rows(&runtime, "reader_rows"), 2); + runtime + .execute_query("SET CONFIG runtime.mvcc.autovacuum.enabled = true") + .expect("enable collector"); + await_history_reclamation(&runtime, "reader_rows", 1); + assert_eq!(value(&runtime, 1), "1"); +} + +#[test] +fn indexed_rows_survive_reclamation_checkpoint_and_reopen() { + let directory = tempfile::tempdir().expect("fixture directory"); + let path = directory.path().join("reclaimed.rdb"); + { + let runtime = + RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("runtime opens"); + runtime + .execute_query("CREATE TABLE reader_rows (id INTEGER, n INTEGER)") + .expect("create table"); + runtime + .execute_query("CREATE INDEX reader_rows_id ON reader_rows (id)") + .expect("create index"); + runtime + .execute_query("INSERT INTO reader_rows (id, n) VALUES (1, 0)") + .expect("seed row"); + for _ in 0..20 { + runtime + .execute_query("UPDATE reader_rows SET n = n + 1 WHERE id = 1") + .expect("update row"); + } + await_history_reclamation(&runtime, "reader_rows", 1); + assert_eq!(value(&runtime, 1), "20"); + runtime.flush().expect("checkpoint reclaimed state"); + } + let restored = + RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("reopen checkpoint"); + assert_eq!(value(&restored, 1), "20"); + assert_eq!(physical_rows(&restored, "reader_rows"), 1); +} + +#[test] +fn aborted_indexed_writes_remain_invisible_and_the_key_can_be_reused() { + use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; + let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime opens"); + runtime + .execute_query("CREATE TABLE reader_rows (id INTEGER, n INTEGER)") + .expect("create table"); + runtime + .execute_query("CREATE UNIQUE INDEX reader_rows_id ON reader_rows (id)") + .expect("create index"); + set_current_connection_id(42); + runtime + .execute_query("BEGIN") + .expect("begin aborted insert"); + runtime + .execute_query("INSERT INTO reader_rows (id, n) VALUES (1, 99)") + .expect("insert tentative row"); + let result = runtime.execute_query("ROLLBACK"); + clear_current_connection_id(); + result.expect("rollback insert"); + await_history_reclamation(&runtime, "reader_rows", 0); + assert!(runtime + .execute_query("SELECT n FROM reader_rows WHERE id = 1") + .expect("read after rollback") + .result + .records + .is_empty()); + runtime + .execute_query("INSERT INTO reader_rows (id, n) VALUES (1, 7)") + .expect("reuse rolled-back unique key"); + set_current_connection_id(42); + runtime + .execute_query("BEGIN") + .expect("begin aborted update"); + runtime + .execute_query("UPDATE reader_rows SET id = 2, n = 8 WHERE id = 1") + .expect("change indexed key tentatively"); + let result = runtime.execute_query("ROLLBACK"); + clear_current_connection_id(); + result.expect("rollback indexed update"); + await_history_reclamation(&runtime, "reader_rows", 1); + assert_eq!(value(&runtime, 1), "7"); + assert!(runtime + .execute_query("SELECT n FROM reader_rows WHERE id = 2") + .expect("aborted key remains absent") + .result + .records + .is_empty()); +} diff --git a/crates/reddb-server/tests/btree_delete_stack.rs b/crates/reddb-server/tests/btree_delete_stack.rs new file mode 100644 index 000000000..f6f51f43f --- /dev/null +++ b/crates/reddb-server/tests/btree_delete_stack.rs @@ -0,0 +1,112 @@ +//! Cascading B-tree merges must fit the maintenance worker's ordinary stack. +use std::process::Command; +use std::sync::Arc; + +use reddb_server::storage::engine::{BTree, PageType, Pager}; + +fn fixture_key(id: u32) -> Vec { + let mut key = vec![b'x'; 1024]; + key[..4].copy_from_slice(&id.to_be_bytes()); + key +} + +#[test] +fn cascading_deletes_fit_a_two_mib_worker_stack() { + const CHILD: &str = "REDDB_BTREE_STACK_TEST_CHILD"; + const DIRECTORY: &str = "REDDB_BTREE_STACK_TEST_DIRECTORY"; + if let Ok(order) = std::env::var(CHILD) { + let directory = + std::path::PathBuf::from(std::env::var_os(DIRECTORY).expect("child fixture path")); + assert_eq!( + std::fs::read(directory.join("owned-stack-fixture")).expect("fixture owner marker"), + b"btree-stack-v1" + ); + let path = directory.join("tree.rdb"); + let (pager, tree) = std::thread::Builder::new() + .stack_size(8 * 1024 * 1024) + .spawn(move || { + let pager = Arc::new(Pager::open_default(&path).expect("fixture pager opens")); + let tree = BTree::new(pager.clone()); + let rows = (0..32768) + .map(|id| (fixture_key(id), id.to_be_bytes().to_vec())) + .collect::>(); + tree.bulk_insert_sorted(&rows).expect("seed a deep tree"); + let mut page = tree.root_page_id(); + let mut levels = 0; + for _ in 0..16 { + let current = pager.read_page(page).expect("read fixture level"); + levels += 1; + if current.page_type().expect("valid page type") == PageType::BTreeLeaf { + break; + } + page = current.right_child(); + } + assert!( + levels >= 5, + "fixture must exercise cascading interior merges" + ); + (pager, tree) + }) + .expect("seed worker starts") + .join() + .expect("seed worker completes"); + std::thread::Builder::new() + .name("small-stack-delete".to_string()) + .stack_size(2 * 1024 * 1024) + .spawn(move || { + let ids: Vec = if order == "ascending" { + (0..32767).collect() + } else { + (1..32768).rev().collect() + }; + for id in ids { + assert!(tree + .delete(&fixture_key(id)) + .expect("cascading delete succeeds")); + } + let remaining = if order == "ascending" { 32767u32 } else { 0u32 }; + assert_eq!(tree.count().expect("count remaining keys"), 1); + assert_eq!( + tree.get(&fixture_key(remaining)) + .expect("read remaining key"), + Some(remaining.to_be_bytes().to_vec()) + ); + assert!(tree + .delete(&fixture_key(remaining)) + .expect("remove final root")); + assert!(tree.is_empty()); + drop(tree); + drop(pager); + }) + .expect("small-stack worker starts") + .join() + .expect("small-stack worker completes"); + return; + } + // A regression aborts only its child, so the parent can report the failure + // and remove its own temporary fixture instead of leaving test databases. + for order in ["ascending", "descending"] { + let directory = tempfile::tempdir().expect("isolated fixture directory"); + std::fs::write( + directory.path().join("owned-stack-fixture"), + b"btree-stack-v1", + ) + .expect("mark fixture owner"); + let result = Command::new(std::env::current_exe().expect("test executable")) + .args([ + "--exact", + "cascading_deletes_fit_a_two_mib_worker_stack", + "--nocapture", + ]) + .env(CHILD, order) + .env(DIRECTORY, directory.path()) + .output() + .expect("run isolated stack regression"); + drop(directory); + assert!( + result.status.success(), + "{order} deletion failed: {}", + String::from_utf8_lossy(&result.stderr) + ); + } +} diff --git a/crates/reddb-server/tests/checkpoint_after_physical_delete.rs b/crates/reddb-server/tests/checkpoint_after_physical_delete.rs new file mode 100644 index 000000000..c1a9e518a --- /dev/null +++ b/crates/reddb-server/tests/checkpoint_after_physical_delete.rs @@ -0,0 +1,144 @@ +//! Physical deletion already updates the paged tree; unchanged survivors need no re-encoding. +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +use reddb_server::storage::{EntityId, UnifiedEntity, UnifiedStore, UnifiedStoreConfig}; +use reddb_types::Value; + +struct TrackedAllocator; + +#[global_allocator] +static ALLOCATOR: TrackedAllocator = TrackedAllocator; +static TRACK: AtomicBool = AtomicBool::new(false); +static ALLOCATED: AtomicUsize = AtomicUsize::new(0); + +fn account(bytes: usize) { + if TRACK.load(Ordering::Relaxed) { + ALLOCATED.fetch_add(bytes, Ordering::Relaxed); + } +} + +// Count allocator traffic only within the measured checkpoint. Delegation keeps +// System's layouts, ownership and failure behavior unchanged. +unsafe impl GlobalAlloc for TrackedAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc(layout) }; + if !pointer.is_null() { + account(layout.size()); + } + pointer + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc_zeroed(layout) }; + if !pointer.is_null() { + account(layout.size()); + } + pointer + } + + unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) { + unsafe { System.dealloc(pointer, layout) }; + } + + unsafe fn realloc(&self, pointer: *mut u8, layout: Layout, size: usize) -> *mut u8 { + let replacement = unsafe { System.realloc(pointer, layout, size) }; + if !replacement.is_null() { + account(size); + } + replacement + } +} + +#[test] +fn physical_delete_checkpoint_does_not_reencode_wide_survivors() { + let directory = tempfile::tempdir().expect("isolated fixture"); + let path = directory.path().join("data.rdb"); + let store = UnifiedStore::open_with_config( + &path, + UnifiedStoreConfig::default().with_page_cache_slots(8), + ) + .expect("open paged store"); + store.create_collection("rows").expect("create collection"); + for id in 1..=128 { + store + .insert( + "rows", + UnifiedEntity::table_row( + EntityId::new(id), + "rows", + id, + vec![Value::text("x".repeat(64 * 1024))], + ), + ) + .expect("seed wide row"); + } + store.persist().expect("initial checkpoint"); + assert_eq!( + store + .delete_batch( + "rows", + &[EntityId::new(1), EntityId::new(64), EntityId::new(128)] + ) + .expect("physical deletes"), + [EntityId::new(1), EntityId::new(64), EntityId::new(128)] + ); + + ALLOCATED.store(0, Ordering::Relaxed); + TRACK.store(true, Ordering::Relaxed); + let checkpoint = store.persist(); + TRACK.store(false, Ordering::Relaxed); + checkpoint.expect("checkpoint after physical deletion"); + let allocated = ALLOCATED.load(Ordering::Relaxed); + eprintln!("checkpoint_after_delete_allocated_bytes={allocated}"); + assert!( + allocated < 4 * 1024 * 1024, + "checkpoint re-encoded unchanged wide survivors: {allocated} allocated bytes" + ); + + // A direct manager edit is not covered by the preceding paged generation. + // A later physical delete must not hide that edit by advancing its proof. + store + .get_collection("rows") + .expect("collection") + .update(UnifiedEntity::table_row( + EntityId::new(2), + "rows", + 2, + vec![Value::text("changed")], + )) + .expect("untracked manager edit"); + assert_eq!( + store + .delete_batch("rows", &[EntityId::new(3)]) + .expect("delete after untracked edit"), + [EntityId::new(3)] + ); + store + .persist() + .expect("checkpoint must retain untracked edit"); + drop(store); + let reopened = UnifiedStore::open(&path).expect("reopen checkpoint"); + assert_eq!( + reopened.get_collection("rows").expect("collection").count(), + 124 + ); + for id in [1, 3, 64, 128] { + assert!(reopened.get("rows", EntityId::new(id)).is_none()); + } + assert_eq!( + reopened + .get("rows", EntityId::new(2)) + .expect("changed row") + .data + .as_row() + .expect("row") + .columns, + [Value::text("changed")] + ); + let survivor = reopened.get("rows", EntityId::new(4)).expect("survivor"); + assert_eq!( + survivor.data.as_row().expect("row").columns, + [Value::text("x".repeat(64 * 1024))] + ); +} diff --git a/crates/reddb-server/tests/checkpoint_concurrent_transaction.rs b/crates/reddb-server/tests/checkpoint_concurrent_transaction.rs new file mode 100644 index 000000000..e012ca0b7 --- /dev/null +++ b/crates/reddb-server/tests/checkpoint_concurrent_transaction.rs @@ -0,0 +1,212 @@ +//! Checkpoint publication must fence row mutations as well as commit visibility. +use std::path::Path; +use std::process::{Child, Command}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant}; + +use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; +use reddb_server::{RedDBOptions, RedDBRuntime, StorageDeployPreset}; + +const CHILD_PATH: &str = "REDDB_TEST_CONCURRENT_CHECKPOINT_PATH"; + +fn open(path: &Path) -> RedDBRuntime { + RedDBRuntime::with_options( + RedDBOptions::persistent(path).with_memory_budget(2 * 1024 * 1024 * 1024), + ) + .expect("open runtime") +} + +fn sql(runtime: &RedDBRuntime, query: &str) { + runtime.execute_query(query).expect("fixture SQL"); +} + +fn checkpoint_child(path: &Path, explicit: bool, insert_batch: bool) { + let runtime = RedDBRuntime::with_options( + RedDBOptions::persistent(path) + .with_storage_profile(StorageDeployPreset::Serverless.selection()) + .expect("paged profile"), + ) + .expect("paged runtime"); + runtime.checkpoint().expect("initialize file"); + drop(runtime); + let runtime = Arc::new(open(path)); + sql( + &runtime, + "SET CONFIG runtime.mvcc.autovacuum.enabled = false", + ); + let payload = "x".repeat(131072); + for table in 0..8 { + sql( + &runtime, + &format!("CREATE TABLE pad_{table} (id INTEGER,n INTEGER,payload TEXT)"), + ); + for id in 0..16 { + sql( + &runtime, + &format!("INSERT INTO pad_{table} (id,n,payload) VALUES ({id},0,'{payload}')"), + ); + } + } + for table in 0..16 { + sql( + &runtime, + &format!("CREATE TABLE target_{table} (id INTEGER,n INTEGER)"), + ); + sql( + &runtime, + &format!("INSERT INTO target_{table} (id,n) VALUES (1,0)"), + ); + } + runtime.checkpoint().expect("seed checkpoint"); + for table in 0..8 { + sql(&runtime, &format!("UPDATE pad_{table} SET n=1")); + } + let pager = Arc::clone(runtime.db().store().pager().expect("paged store")); + let pages = pager.page_count().expect("page count"); + let finished = Arc::new(AtomicBool::new(false)); + let checkpoint = { + let runtime = Arc::clone(&runtime); + let finished = Arc::clone(&finished); + thread::spawn(move || { + runtime.checkpoint().expect("concurrent checkpoint"); + finished.store(true, Ordering::SeqCst); + }) + }; + // Page allocation precedes the final flush. File length changes too late + // to establish that a transaction begins during checkpoint serialization. + let deadline = Instant::now() + Duration::from_secs(30); + while pager.page_count().expect("checkpoint progress") <= pages { + assert!( + !finished.load(Ordering::SeqCst), + "checkpoint finished before allocation window" + ); + assert!(Instant::now() < deadline, "checkpoint allocation deadline"); + thread::yield_now(); + } + set_current_connection_id(42); + if explicit { + sql(&runtime, "BEGIN"); + } + for table in 0..16 { + sql( + &runtime, + &if insert_batch { + format!("INSERT INTO target_{table} (id,n) VALUES (2,9),(3,10)") + } else { + format!("UPDATE target_{table} SET n=9 WHERE id=1") + }, + ); + } + clear_current_connection_id(); + checkpoint.join().expect("checkpoint worker"); + std::fs::write(path.with_extension("ready"), b"ready").expect("crash marker"); + loop { + thread::sleep(Duration::from_secs(1)); + } +} + +struct CrashChild(Child); +impl Drop for CrashChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn crash_case(name: &str, explicit: bool, insert_batch: bool) { + if let Some(path) = std::env::var_os(CHILD_PATH) { + checkpoint_child(Path::new(&path), explicit, insert_batch); + unreachable!(); + } + let directory = tempfile::tempdir().expect("isolated fixture"); + let path = directory.path().join("data.rdb"); + let mut child = CrashChild( + Command::new(std::env::current_exe().expect("test executable")) + .args(["--exact", name, "--nocapture"]) + .env(CHILD_PATH, &path) + .spawn() + .expect("checkpoint child"), + ); + let deadline = Instant::now() + Duration::from_secs(60); + while !path.with_extension("ready").exists() { + assert!( + child.0.try_wait().expect("child status").is_none(), + "child exited early" + ); + assert!(Instant::now() < deadline, "child deadline"); + thread::sleep(Duration::from_millis(10)); + } + child.0.kill().expect("force process termination"); + assert!(!child.0.wait().expect("wait for child").success()); + let restored = open(&path); + let expected = if explicit { "0" } else { "9" }; + for table in 0..16 { + if insert_batch { + let result = restored + .execute_query(&format!("SELECT id,n FROM target_{table} ORDER BY id")) + .expect("recovered inserted rows"); + let expected = if explicit { + vec![(1, 0)] + } else { + vec![(1, 0), (2, 9), (3, 10)] + }; + assert_eq!( + result.result.records.len(), + expected.len(), + "target_{table}" + ); + for (row, (id, n)) in result.result.records.iter().zip(expected) { + assert_eq!(row.get("id").expect("id").to_string(), id.to_string()); + assert_eq!(row.get("n").expect("n").to_string(), n.to_string()); + } + continue; + } + let result = restored + .execute_query(&format!("SELECT n FROM target_{table} WHERE id=1")) + .expect("recovered row"); + assert_eq!(result.result.records.len(), 1); + assert_eq!( + result.result.records[0].get("n").expect("n").to_string(), + expected, + "target_{table}" + ); + } +} + +#[test] +fn transactions_started_during_checkpoint_do_not_become_committed_after_crash() { + crash_case( + "transactions_started_during_checkpoint_do_not_become_committed_after_crash", + true, + false, + ); +} + +#[test] +fn autocommit_updates_started_during_checkpoint_survive_crash() { + crash_case( + "autocommit_updates_started_during_checkpoint_survive_crash", + false, + false, + ); +} + +#[test] +fn unfinished_bulk_inserts_started_during_checkpoint_do_not_survive_crash() { + crash_case( + "unfinished_bulk_inserts_started_during_checkpoint_do_not_survive_crash", + true, + true, + ); +} + +#[test] +fn autocommit_bulk_inserts_started_during_checkpoint_survive_crash() { + crash_case( + "autocommit_bulk_inserts_started_during_checkpoint_survive_crash", + false, + true, + ); +} diff --git a/crates/reddb-server/tests/checkpoint_open_transaction.rs b/crates/reddb-server/tests/checkpoint_open_transaction.rs new file mode 100644 index 000000000..8f1b8ad58 --- /dev/null +++ b/crates/reddb-server/tests/checkpoint_open_transaction.rs @@ -0,0 +1,205 @@ +//! A checkpoint must not commit an unfinished transaction across a process crash. +use std::path::Path; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; +use reddb_server::{RedDBOptions, RedDBRuntime, StorageDeployPreset}; + +const CHILD_PATH: &str = "REDDB_TEST_OPEN_TRANSACTION_CHECKPOINT_PATH"; + +fn open(path: &Path) -> RedDBRuntime { + RedDBRuntime::with_options(RedDBOptions::persistent(path)).expect("open runtime") +} + +fn seed(path: &Path, paged: bool) { + if paged { + let runtime = RedDBRuntime::with_options( + RedDBOptions::persistent(path) + .with_storage_profile(StorageDeployPreset::Serverless.selection()) + .expect("paged profile"), + ) + .expect("paged runtime"); + runtime.checkpoint().expect("initialize file"); + } + let runtime = open(path); + for sql in [ + "CREATE TABLE tx_rows (id INTEGER, n INTEGER)", + "CREATE UNIQUE INDEX tx_rows_id ON tx_rows (id)", + "INSERT INTO tx_rows (id,n) VALUES (1,7),(4,8)", + ] { + runtime + .execute_query(sql) + .expect("seed transaction fixture"); + } + runtime.checkpoint().expect("committed seed"); +} + +fn verify_values(runtime: &RedDBRuntime, first_value: i64) { + let rows = runtime + .execute_query("SELECT id,n FROM tx_rows ORDER BY id") + .expect("original rows"); + assert_eq!(rows.result.records.len(), 2); + for (row, (id, value)) in rows.result.records.iter().zip([(1, first_value), (4, 8)]) { + assert_eq!(row.get("id").expect("id").to_string(), id.to_string()); + assert_eq!(row.get("n").expect("n").to_string(), value.to_string()); + assert_eq!( + runtime + .execute_query(&format!("SELECT n FROM tx_rows WHERE id={id}")) + .expect("indexed original") + .result + .records + .len(), + 1 + ); + } +} + +struct CrashChild(Child); + +impl Drop for CrashChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn crash_case(name: &str, mutation: &str, commit: bool, expected_value: i64) { + if let Some(path) = std::env::var_os(CHILD_PATH) { + let path = Path::new(&path); + let runtime = open(path); + set_current_connection_id(42); + runtime + .execute_query("BEGIN") + .expect("begin unfinished transaction"); + runtime + .execute_query(mutation) + .expect("uncommitted mutation"); + clear_current_connection_id(); + runtime + .checkpoint() + .expect("checkpoint while transaction is open"); + if commit { + set_current_connection_id(42); + runtime + .execute_query("COMMIT") + .expect("durable commit after checkpoint"); + clear_current_connection_id(); + } + std::fs::write(path.with_extension("ready"), b"checkpoint completed") + .expect("crash marker"); + loop { + std::thread::sleep(Duration::from_secs(1)); + } + } + for paged in [false, true] { + run_crash_case(name, paged, expected_value); + } +} + +fn run_crash_case(name: &str, paged: bool, expected_value: i64) { + let directory = tempfile::tempdir().expect("isolated fixture"); + let path = directory.path().join("data.rdb"); + seed(&path, paged); + let mut child = CrashChild( + Command::new(std::env::current_exe().expect("test executable")) + .args(["--exact", name, "--nocapture"]) + .env(CHILD_PATH, &path) + .spawn() + .expect("spawn checkpoint child"), + ); + let deadline = Instant::now() + Duration::from_secs(30); + while !path.with_extension("ready").exists() && Instant::now() < deadline { + assert!( + child.0.try_wait().expect("child status").is_none(), + "checkpoint child exited early" + ); + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + path.with_extension("ready").exists(), + "checkpoint child timed out" + ); + child.0.kill().expect("force process termination"); + let status = child.0.wait().expect("wait for termination"); + assert!(!status.success()); + let restored = open(&path); + verify_values(&restored, expected_value); + restored.checkpoint().expect("checkpoint recovered state"); + drop(restored); + verify_values(&open(&path), expected_value); +} + +#[test] +fn unfinished_insert_does_not_survive_checkpoint_and_crash() { + crash_case( + "unfinished_insert_does_not_survive_checkpoint_and_crash", + "INSERT INTO tx_rows (id,n) VALUES (3,99)", + false, + 7, + ); +} + +#[test] +fn unfinished_update_does_not_survive_checkpoint_and_crash() { + crash_case( + "unfinished_update_does_not_survive_checkpoint_and_crash", + "UPDATE tx_rows SET id=2,n=99 WHERE id=1", + false, + 7, + ); +} + +#[test] +fn unfinished_delete_does_not_survive_checkpoint_and_crash() { + crash_case( + "unfinished_delete_does_not_survive_checkpoint_and_crash", + "DELETE FROM tx_rows WHERE id=4", + false, + 7, + ); +} + +#[test] +fn transaction_can_commit_after_its_versions_were_excluded_from_checkpoint() { + for paged in [false, true] { + commit_after_checkpoint(paged); + } +} + +fn commit_after_checkpoint(paged: bool) { + let directory = tempfile::tempdir().expect("isolated fixture"); + let path = directory.path().join("data.rdb"); + seed(&path, paged); + { + let runtime = open(&path); + set_current_connection_id(42); + runtime.execute_query("BEGIN").expect("begin"); + runtime + .execute_query("UPDATE tx_rows SET n=9 WHERE id=1") + .expect("update"); + clear_current_connection_id(); + runtime.checkpoint().expect("checkpoint unfinished version"); + set_current_connection_id(42); + runtime + .execute_query("COMMIT") + .expect("commit after checkpoint"); + clear_current_connection_id(); + runtime.checkpoint().expect("checkpoint committed version"); + } + let restored = open(&path); + let rows = restored + .execute_query("SELECT n FROM tx_rows WHERE id=1") + .expect("committed value"); + assert_eq!(rows.result.records[0].get("n").expect("n").to_string(), "9"); +} + +#[test] +fn commit_after_checkpoint_survives_crash_without_another_checkpoint() { + crash_case( + "commit_after_checkpoint_survives_crash_without_another_checkpoint", + "UPDATE tx_rows SET n=9 WHERE id=1", + true, + 9, + ); +} diff --git a/crates/reddb-server/tests/pager_flush_transaction.rs b/crates/reddb-server/tests/pager_flush_transaction.rs new file mode 100644 index 000000000..b969be0de --- /dev/null +++ b/crates/reddb-server/tests/pager_flush_transaction.rs @@ -0,0 +1,160 @@ +//! Incidental pager flushes must not publish unfinished transaction changes. +use std::path::Path; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use reddb_server::runtime::mvcc::{clear_current_connection_id, set_current_connection_id}; +use reddb_server::{RedDBOptions, RedDBRuntime, StorageDeployPreset}; + +const CHILD_PATH: &str = "REDDB_TEST_TRANSACTION_FLUSH_PATH"; +const CHILD_END: &str = "REDDB_TEST_TRANSACTION_FLUSH_END"; + +fn open(path: &Path) -> RedDBRuntime { + RedDBRuntime::with_options(RedDBOptions::persistent(path)).expect("open runtime") +} + +fn seed(path: &Path) { + let runtime = RedDBRuntime::with_options( + RedDBOptions::persistent(path) + .with_storage_profile(StorageDeployPreset::Serverless.selection()) + .expect("paged profile"), + ) + .expect("paged runtime"); + for sql in [ + "CREATE TABLE tx_rows (id INTEGER, n INTEGER)", + "CREATE UNIQUE INDEX tx_rows_id ON tx_rows (id)", + "INSERT INTO tx_rows (id,n) VALUES (1,7),(4,8)", + ] { + runtime.execute_query(sql).expect("seed fixture"); + } + runtime.checkpoint().expect("committed seed"); +} + +fn verify(runtime: &RedDBRuntime, expected: &[(i64, i64)]) { + let rows = runtime + .execute_query("SELECT id,n FROM tx_rows ORDER BY id") + .expect("read recovered rows"); + assert_eq!(rows.result.records.len(), expected.len()); + for (row, (id, value)) in rows.result.records.iter().zip(expected) { + assert_eq!(row.get("id").expect("id").to_string(), id.to_string()); + assert_eq!(row.get("n").expect("n").to_string(), value.to_string()); + let indexed = runtime + .execute_query(&format!("SELECT n FROM tx_rows WHERE id={id}")) + .expect("indexed recovered row"); + assert_eq!(indexed.result.records.len(), 1); + assert_eq!( + indexed.result.records[0].get("n").expect("n").to_string(), + value.to_string() + ); + } +} + +struct CrashChild(Child); +impl Drop for CrashChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn crash_case(name: &str, mutation: &str, committed: &[(i64, i64)]) { + if let Some(path) = std::env::var_os(CHILD_PATH) { + let path = Path::new(&path); + let runtime = open(path); + set_current_connection_id(42); + runtime.execute_query("BEGIN").expect("begin"); + runtime.execute_query(mutation).expect("mutation"); + clear_current_connection_id(); + let flush = || { + runtime + .db() + .store() + .pager() + .expect("paged fixture") + .flush() + .expect("incidental flush"); + }; + flush(); + let end = std::env::var(CHILD_END).expect("transaction outcome"); + if !end.is_empty() { + set_current_connection_id(42); + runtime.execute_query(&end).expect("end transaction"); + clear_current_connection_id(); + flush(); + } + std::fs::write(path.with_extension("ready"), b"flushed").expect("crash marker"); + loop { + std::thread::sleep(Duration::from_secs(1)); + } + } + for end in ["", "ROLLBACK", "COMMIT"] { + let directory = tempfile::tempdir().expect("isolated fixture"); + let path = directory.path().join("data.rdb"); + seed(&path); + let mut child = CrashChild( + Command::new(std::env::current_exe().expect("test executable")) + .args(["--exact", name, "--nocapture"]) + .env(CHILD_PATH, &path) + .env(CHILD_END, end) + .spawn() + .expect("spawn crash child"), + ); + let deadline = Instant::now() + Duration::from_secs(30); + while !path.with_extension("ready").exists() && Instant::now() < deadline { + assert!( + child.0.try_wait().expect("child status").is_none(), + "child exited early" + ); + std::thread::sleep(Duration::from_millis(20)); + } + assert!(path.with_extension("ready").exists(), "child timed out"); + child.0.kill().expect("force termination"); + assert!(!child.0.wait().expect("wait for termination").success()); + let expected = if end == "COMMIT" { + committed + } else { + &[(1, 7), (4, 8)] + }; + let runtime = open(&path); + verify(&runtime, expected); + runtime.checkpoint().expect("checkpoint recovered state"); + drop(runtime); + verify(&open(&path), expected); + } +} + +#[test] +fn insert_survives_incidental_flush_only_after_commit() { + crash_case( + "insert_survives_incidental_flush_only_after_commit", + "INSERT INTO tx_rows (id,n) VALUES (3,99)", + &[(1, 7), (3, 99), (4, 8)], + ); +} + +#[test] +fn bulk_insert_survives_incidental_flush_only_after_commit() { + crash_case( + "bulk_insert_survives_incidental_flush_only_after_commit", + "INSERT INTO tx_rows (id,n) VALUES (3,99),(5,100)", + &[(1, 7), (3, 99), (4, 8), (5, 100)], + ); +} + +#[test] +fn update_survives_incidental_flush_only_after_commit() { + crash_case( + "update_survives_incidental_flush_only_after_commit", + "UPDATE tx_rows SET id=2,n=99 WHERE id=1", + &[(2, 99), (4, 8)], + ); +} + +#[test] +fn delete_survives_incidental_flush_only_after_commit() { + crash_case( + "delete_survives_incidental_flush_only_after_commit", + "DELETE FROM tx_rows WHERE id=4", + &[(1, 7)], + ); +} diff --git a/crates/reddb-server/tests/positional_row_update_reopen.rs b/crates/reddb-server/tests/positional_row_update_reopen.rs new file mode 100644 index 000000000..e90b45bb2 --- /dev/null +++ b/crates/reddb-server/tests/positional_row_update_reopen.rs @@ -0,0 +1,93 @@ +//! Reopened positional rows must retain untouched fields during SQL updates. +use std::path::Path; + +use reddb_server::{RedDBOptions, RedDBRuntime, StorageDeployPreset}; + +fn open(path: &Path, initial: bool) -> RedDBRuntime { + let options = RedDBOptions::persistent(path); + RedDBRuntime::with_options(if initial { + options + .with_storage_profile(StorageDeployPreset::Serverless.selection()) + .expect("paged fixture") + } else { + options + }) + .expect("runtime opens") +} + +fn seed(path: &Path) { + { + let runtime = open(path, true); + runtime.checkpoint().expect("initialize paged file"); + } + let runtime = open(path, false); + for query in [ + "CREATE TABLE updated_rows (id INTEGER, n INTEGER, payload TEXT)", + "CREATE UNIQUE INDEX rows_id ON updated_rows (id)", + "INSERT INTO updated_rows (id,n,payload) VALUES (0,0,'zero'),(1,0,'one'),(2,0,'two')", + "UPDATE updated_rows SET n = 1", + "DELETE FROM updated_rows WHERE id < 2", + "INSERT INTO updated_rows (id,n,payload) VALUES (0,1,'zero'),(1,1,'one')", + ] { + runtime.execute_query(query).expect("seed fixture"); + } + runtime.checkpoint().expect("checkpoint reinserted rows"); +} + +fn verify(runtime: &RedDBRuntime, revision: i64) { + let rows = runtime + .execute_query("SELECT id,n,payload FROM updated_rows") + .expect("full row scan"); + assert_eq!(rows.result.records.len(), 3); + for (id, payload) in [(0, "zero"), (1, "one"), (2, "two")] { + let row = rows + .result + .records + .iter() + .find(|row| { + row.get("id") + .is_some_and(|value| value.to_string() == id.to_string()) + }) + .expect("untouched row ID must remain present"); + assert_eq!(row.get("n").expect("n").to_string(), revision.to_string()); + assert_eq!( + row.get("payload").expect("payload").as_text(), + Some(payload) + ); + let indexed = runtime + .execute_query(&format!("SELECT payload FROM updated_rows WHERE id = {id}")) + .expect("unique index lookup"); + assert_eq!(indexed.result.records.len(), 1); + assert_eq!( + indexed.result.records[0] + .get("payload") + .expect("indexed payload") + .as_text(), + Some(payload) + ); + } +} + +fn exercise(update: &str) { + let directory = tempfile::tempdir().expect("fixture directory"); + let path = directory.path().join("data.rdb"); + seed(&path); + { + let runtime = open(&path, false); + verify(&runtime, 1); + runtime.execute_query(update).expect("update reopened rows"); + verify(&runtime, 2); + runtime.checkpoint().expect("checkpoint updated rows"); + } + verify(&open(&path, false), 2); +} + +#[test] +fn constant_update_preserves_reopened_positional_fields_and_indexes() { + exercise("UPDATE updated_rows SET n = 2"); +} + +#[test] +fn expression_update_preserves_reopened_positional_fields_and_indexes() { + exercise("UPDATE updated_rows SET n = n + 1"); +} diff --git a/crates/reddb-server/tests/vacuum_memory.rs b/crates/reddb-server/tests/vacuum_memory.rs new file mode 100644 index 000000000..66d335f5b --- /dev/null +++ b/crates/reddb-server/tests/vacuum_memory.rs @@ -0,0 +1,111 @@ +//! VACUUM must not clone live row payloads just to inspect MVCC headers. +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use reddb_server::{RedDBOptions, RedDBRuntime}; +use reddb_types::Value; + +struct TrackedAllocator; + +#[global_allocator] +static ALLOCATOR: TrackedAllocator = TrackedAllocator; +static LIVE_BYTES: AtomicUsize = AtomicUsize::new(0); +static PEAK_BYTES: AtomicUsize = AtomicUsize::new(0); + +fn account_allocated(bytes: usize) { + let live = LIVE_BYTES.fetch_add(bytes, Ordering::Relaxed) + bytes; + PEAK_BYTES.fetch_max(live, Ordering::Relaxed); +} + +// Test-only accounting delegates every allocation to System with its original +// layout; it never allocates bookkeeping objects or changes pointer ownership. +unsafe impl GlobalAlloc for TrackedAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc(layout) }; + if !pointer.is_null() { + account_allocated(layout.size()); + } + pointer + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc_zeroed(layout) }; + if !pointer.is_null() { + account_allocated(layout.size()); + } + pointer + } + + unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) { + unsafe { System.dealloc(pointer, layout) }; + LIVE_BYTES.fetch_sub(layout.size(), Ordering::Relaxed); + } + + unsafe fn realloc(&self, pointer: *mut u8, layout: Layout, size: usize) -> *mut u8 { + let replacement = unsafe { System.realloc(pointer, layout, size) }; + if !replacement.is_null() { + if size >= layout.size() { + account_allocated(size - layout.size()); + } else { + LIVE_BYTES.fetch_sub(layout.size() - size, Ordering::Relaxed); + } + } + replacement + } +} + +#[test] +fn vacuum_does_not_materialize_wide_live_payloads() { + let runtime = + RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("in-memory fixture opens"); + runtime + .execute_query("SET CONFIG runtime.result_cache.enabled = false") + .expect("disable result caching for the allocation measurement"); + runtime + .execute_query("CREATE TABLE wide_rows (id INTEGER, payload JSON)") + .expect("create fixture table"); + let payload = format!("{{\"value\":\"{}\"}}", "x".repeat(16 * 1024)).into_bytes(); + for start in (0_i64..5_000).step_by(100) { + let mut params = Vec::with_capacity(200); + let mut values = Vec::with_capacity(100); + for id in start..start + 100 { + params.push(Value::Integer(id)); + params.push(Value::Json(payload.clone())); + values.push(format!("(${}, ${})", params.len() - 1, params.len())); + } + runtime + .execute_query_with_params( + &format!( + "INSERT INTO wide_rows (id, payload) VALUES {}", + values.join(",") + ), + ¶ms, + ) + .expect("insert fixture batch"); + if (start + 100) % 1_000 == 0 { + eprintln!("seeded_rows={}", start + 100); + } + } + + let baseline = LIVE_BYTES.load(Ordering::Relaxed); + PEAK_BYTES.store(baseline, Ordering::Relaxed); + runtime + .execute_query("VACUUM wide_rows") + .expect("vacuum succeeds"); + let extra_peak = PEAK_BYTES.load(Ordering::Relaxed).saturating_sub(baseline); + eprintln!("vacuum_extra_peak_bytes={extra_peak}"); + assert!( + extra_peak < 16 * 1024 * 1024, + "header inspection allocated {extra_peak} extra bytes" + ); + let count = runtime + .execute_query("SELECT COUNT(*) AS total FROM wide_rows") + .expect("count survives vacuum"); + assert_eq!( + count.result.records[0] + .get("total") + .expect("count exists") + .to_string(), + "5000" + ); +}