From 54e1e11bd7d61b45fba0852cef7a01ffa0c2c3fe Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 05:16:27 +0700 Subject: [PATCH] fix: consolidate release hardening and torn-store refusal Linearize PR #47 on the current master while preserving its exact reviewed tree. Includes release-hardening regression coverage, beta.4 metadata, and the accepted best-effort persistence durability boundary with typed torn-store rejection. --- Cargo.toml | 4 +- README.md | 69 +++++- codegen/Cargo.toml | 2 +- .../src/generators/in_memory/primary_key.rs | 3 + codegen/src/generators/mod.rs | 1 + codegen/src/generators/persist/primary_key.rs | 3 + codegen/src/generators/persist/table/impls.rs | 125 +++++++++- codegen/src/generators/persist/table/mod.rs | 43 ++++ codegen/src/generators/primary_key.rs | 45 ++++ .../src/generators/read_only/primary_key.rs | 3 + .../src/generators/read_only/table/impls.rs | 105 ++++++++- codegen/src/persist_table/generator/mod.rs | 3 + .../persist_table/generator/space_file/mod.rs | 37 ++- .../generator/space_file/worktable_impls.rs | 34 ++- codegen/src/persist_table/mod.rs | 2 +- codegen/src/persist_table/parser.rs | 43 +++- codegen/src/worktable/mod.rs | 23 ++ docs/crate.md | 114 +++++++++ docs/persistence-durability.md | 83 +++++++ docs/pr46-review-findings.md | 116 ++++++++++ src/in_memory/data.rs | 29 +++ src/in_memory/pages.rs | 28 +++ src/lib.rs | 13 +- src/persistence/engine.rs | 145 +++++++++++- src/persistence/error.rs | 45 ++++ src/persistence/mod.rs | 30 ++- src/persistence/space/data.rs | 10 +- src/persistence/task.rs | 18 ++ src/table/mod.rs | 70 +++++- src/table/vacuum/vacuum.rs | 7 + tests/persistence/mod.rs | 22 +- tests/persistence/read.rs | 18 +- tests/persistence/schema.rs | 90 ++++++++ tests/persistence/torn_shutdown.rs | 218 +++++++++++++----- tests/persistence/vacuum.rs | 10 + tests/worktable/borrowed_primary_key.rs | 81 +++++++ tests/worktable/custom_pk.rs | 15 ++ tests/worktable/leak_probe.rs | 76 ++++++ tests/worktable/mod.rs | 4 + tests/worktable/mutation_gate_deadlock.rs | 144 ++++++++++++ tests/worktable/update_in_place_unsized.rs | 24 +- tests/worktable/vacuum_no_row_loss.rs | 114 +++++++++ 42 files changed, 1912 insertions(+), 157 deletions(-) create mode 100644 codegen/src/generators/primary_key.rs create mode 100644 docs/crate.md create mode 100644 docs/persistence-durability.md create mode 100644 docs/pr46-review-findings.md create mode 100644 tests/persistence/schema.rs create mode 100644 tests/worktable/borrowed_primary_key.rs create mode 100644 tests/worktable/leak_probe.rs create mode 100644 tests/worktable/mutation_gate_deadlock.rs create mode 100644 tests/worktable/vacuum_no_row_loss.rs diff --git a/Cargo.toml b/Cargo.toml index 5d4b254c..59b8c5f4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ members = ["codegen", "examples", "performance_measurement", "performance_measur [package] name = "worktable" -version = "1.0.0-beta.3" +version = "1.0.0-beta.4" edition = "2024" authors = ["Handy-caT"] license = "MIT" @@ -61,7 +61,7 @@ tracing = "0.1" url = { version = "2", optional = true } uuid = { version = "1.24.0", features = ["v4", "v7"] } walkdir = { version = "2", optional = true } -worktable_codegen = { path = "codegen", version = "=1.0.0-beta.3" } +worktable_codegen = { path = "codegen", version = "=1.0.0-beta.4" } [dev-dependencies] chrono = "0.4.43" diff --git a/README.md b/README.md index a6608c26..b70f5c67 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ from a macro, and that persisting it is one feature flag away. ## Install ```sh -cargo add worktable@1.0.0-beta.3 +cargo add worktable@1.0.0-beta.4 ``` ## What you get @@ -34,6 +34,15 @@ cargo add worktable@1.0.0-beta.3 ## Persistence +> [!WARNING] +> WorkTable persistence is **best-effort, not crash-atomic database durability**. +> A successful mutation means the in-memory change was accepted and queued; it does +> not mean the change is on stable storage. `wait_for_ops()` and `close()` flush the +> persistence pipeline, but the current disk format has no transaction journal and +> does not `fsync` every batch. Process or power loss can therefore lose acknowledged +> changes. A torn store is refused with `PersistenceLoadError` rather than opened as +> plausible-but-invented rows. See the [durability and recovery contract](docs/persistence-durability.md). + Persistence is implemented, not planned. `PersistedWorkTable` and `PersistenceConfig` are exported from the crate root; the prelude carries `DiskPersistenceEngine`, `ReadOnlyPersistenceEngine`, the space and table-of-contents types, and the operation-log @@ -44,10 +53,10 @@ S3 support layers *on top of* the disk engine rather than replacing it. ```toml [dependencies] -worktable = { version = "=1.0.0-beta.3", features = ["s3-support"] } # S3 sync, optional +worktable = { version = "=1.0.0-beta.4", features = ["s3-support"] } # S3 sync, optional ``` -Persisted indexes default to WorkTablesIndex. Vanilla IndexSet can be selected explicitly with `using indexset` while retaining the existing disk/S3 representation. Congee and Arctic are explicitly memory-only and require `persist: false`. The full syntax and capability matrix are documented in [Per-index backends with `using`](docs/index-backend-dsl-proposal.md). +Persisted indexes default to WorkTablesIndex. Vanilla IndexSet can be selected explicitly with `using indexset` while retaining the existing disk/S3 representation. Congee and Arctic persistence is experimental and uses their native checkpoint/WAL adapters; declarations using either backend must state `persist: true` or `persist: false` explicitly. The full syntax and capability matrix are documented in [Per-index backends with `using`](docs/index-backend-dsl-proposal.md). ### Persistence lifecycle @@ -58,6 +67,12 @@ table.wait_for_ops().await?; // drain currently queued operations table.close().await?; // stop intake, drain, and join the engine task ``` +`wait_for_ops()` requires application-level writer quiescence if it is being used as +a shutdown boundary; it does not prevent another task from queueing later work. +Neither method is an `fsync` or a transaction commit. The exact guarantees and the +snapshot-restore/replay procedure are documented in +[`docs/persistence-durability.md`](docs/persistence-durability.md). + An unrecoverable event gap, queue-analysis error, batch-apply error, or engine-task failure moves persistence into a terminal failed state. The original error is returned to waiters, graceful close, and later mutation attempts; later durable @@ -65,6 +80,36 @@ operations are not applied after that failure. Dropping a busy table remains a last-resort diagnostic path, so applications should call `close()` during orderly shutdown. +These lifecycle calls are not a crash-durability guarantee: + +| Boundary | Current guarantee | +|---|---| +| Mutation returns | The in-memory change was accepted and its persistence operation was queued. | +| `wait_for_ops()` returns | The persistence engine completed the queued operations; no fsync or stable-storage guarantee is made. | +| `close()` returns | Intake stopped, the queue drained, and the engine task joined; no fsync guarantee is made. | +| Process crash / `SIGKILL` | Acknowledged rows may be lost and the file may be torn. | +| Power loss | No atomic-batch or stable-storage guarantee. | + +The 1.0 beta persistence tier is therefore best-effort rather than a substitute +for a crash-atomic embedded database. Applications requiring crash durability +need an external snapshot/rebuild strategy. A graceful persistence error is +terminal and surfaced consistently, but abrupt termination can currently leave +a partial multi-file batch. Loading audits archived rows plus primary and +secondary index consistency before exposing the table; torn state is refused as +`PersistenceLoadError` and must be restored or rebuilt as documented above. + +Generated persisted tables store row-schema, primary-key, and secondary-index +metadata in `SpaceInfo`. Existing legacy files whose schema metadata is +completely empty remain readable and are not rewritten merely by loading them; +their schema therefore cannot be validated. A non-empty schema mismatch is +rejected before rows are loaded. + +Persisted vacuum compacts the live in-memory layout and keeps disk indexes +consistent with moved rows, but it does not truncate `.wt.data`. Use +`persisted_data_file_size_bytes().await` on a generated persisted table to +observe physical growth and decide when to snapshot/rebuild or run future +offline compaction. + WorkTablesIndex uses its predictable branch-based node search by default in WorkTable. This avoids a measured regression for sequential numeric-key workloads. Alternative search policies remain compile-time feature gates: disable WorkTable's default features and enable one of `wti-hybrid-search`, `wti-std-search`, or `wti-superslice-search` (plus any other features such as `s3-support`). Prefer one search feature for an unambiguous build. If Cargo feature unification enables several, WorkTablesIndex applies the documented deterministic precedence rather than rejecting the graph. ## Concurrent read/write publication @@ -131,6 +176,9 @@ with persistence as an option rather than an assumption. structs that will be used for table logic. ```rust +use worktable::prelude::*; +use worktable::worktable; + worktable!( name: Test, columns: { @@ -181,6 +229,17 @@ Flags list: - `primary_key` flag and related to it. - `optional` flag. +Column modifiers are intentionally inline in the 1.0 DSL. A separate +`attributes` section is not supported; the macro emits an actionable diagnostic +if one is used. This keeps one canonical grammar through the 1.0 compatibility +freeze: + +```text +id: u64 primary_key autoincrement, +tenant: String primary_key, +nickname: String optional, +``` + #### `primary_key` flag declaration If user want to mark column as primary key `primary_key` flag is used. This flag can be used on multiple columns at a @@ -296,7 +355,7 @@ method for now is `select_by_`. It will be described below. There are some default query implementations that are available for all `WorkTable`'s: -- `select(&self, pk: PrimaryKey) -> Option<Row>`; +- `select(&self, pk: impl Into<PrimaryKey>) -> Option<Row>`; borrowed `String`, `str`, tuple, and generated primary-key forms are accepted; - `insert(&self, row: Row) -> Result<PrimaryKey, WorkTableError>`; - `upsert(&self, row: Row) -> Result<(), WorkTableError>`; - `update(&self, row: Row) -> Result<(), WorkTableError>`; @@ -305,7 +364,7 @@ There are some default query implementations that are available for all `WorkTab ### `queries` declaration -`indexes` field is used to define table's queries schema. Queries are used to update/select/delete data. +`queries` field is used to define table's queries schema. Queries are used to update/select/delete data. ``` queries: { diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index 3b864821..c7a55629 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "worktable_codegen" -version = "1.0.0-beta.3" +version = "1.0.0-beta.4" edition = "2024" license = "MIT" description = "Proc-macro companion crate for worktable: the worktable! macro and its derives." diff --git a/codegen/src/generators/in_memory/primary_key.rs b/codegen/src/generators/in_memory/primary_key.rs index 72ac82d5..389484a6 100644 --- a/codegen/src/generators/in_memory/primary_key.rs +++ b/codegen/src/generators/in_memory/primary_key.rs @@ -4,6 +4,7 @@ use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; use crate::generators::in_memory::InMemoryGenerator; use crate::generators::index_backend::primary_key_backend_impl; +use crate::generators::primary_key::gen_borrowed_primary_key_impl; use proc_macro2::{Ident, TokenStream}; use quote::quote; @@ -67,6 +68,7 @@ impl InMemoryGenerator { }; let (backend_derive, backend_impl) = primary_key_backend_impl(self.columns.primary_index_backend, &ident, types)?; + let borrowed_impl = gen_borrowed_primary_key_impl(&ident, types); Ok(quote! { #[derive( @@ -91,6 +93,7 @@ impl InMemoryGenerator { #[rkyv(derive(PartialEq, Eq, PartialOrd, Ord, Debug))] pub struct #ident(#(#types),*); + #borrowed_impl #backend_impl }) } diff --git a/codegen/src/generators/mod.rs b/codegen/src/generators/mod.rs index 6eed7863..83bc1ecb 100644 --- a/codegen/src/generators/mod.rs +++ b/codegen/src/generators/mod.rs @@ -1,4 +1,5 @@ pub mod in_memory; pub(crate) mod index_backend; pub mod persist; +pub(crate) mod primary_key; pub mod read_only; diff --git a/codegen/src/generators/persist/primary_key.rs b/codegen/src/generators/persist/primary_key.rs index 5929f0bf..550fce99 100644 --- a/codegen/src/generators/persist/primary_key.rs +++ b/codegen/src/generators/persist/primary_key.rs @@ -4,6 +4,7 @@ use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; use crate::generators::index_backend::primary_key_backend_impl; use crate::generators::persist::PersistGenerator; +use crate::generators::primary_key::gen_borrowed_primary_key_impl; use proc_macro2::{Ident, TokenStream}; use quote::quote; @@ -63,6 +64,7 @@ impl PersistGenerator { }; let (backend_derive, backend_impl) = primary_key_backend_impl(self.columns.primary_index_backend, &ident, types)?; + let borrowed_impl = gen_borrowed_primary_key_impl(&ident, types); Ok(quote! { #[derive( @@ -87,6 +89,7 @@ impl PersistGenerator { #[rkyv(derive(PartialEq, Eq, PartialOrd, Ord, Debug))] pub struct #ident(#(#types),*); + #borrowed_impl #backend_impl }) } diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index 1187d5c4..2dfa0856 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -1,9 +1,9 @@ use convert_case::{Case, Casing}; -use proc_macro2::{Ident, Span, TokenStream}; +use proc_macro2::{Ident, Literal, Span, TokenStream}; use quote::quote; use crate::common::model::GeneratorType; -use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; +use crate::common::name_generator::{WorktableNameGenerator, is_float, is_unsized_vec}; use crate::generators::persist::PersistGenerator; impl PersistGenerator { @@ -26,6 +26,7 @@ impl PersistGenerator { let count_fn = self.gen_table_count_fn(); let system_info_fn = self.gen_system_info_fn(); let vacuum_fn = self.gen_table_vacuum_fn(); + let validate_loaded_secondary_state_fn = self.gen_validate_loaded_secondary_state_fn(); quote! { #persisted_impl @@ -44,6 +45,96 @@ impl PersistGenerator { #iter_with_async_fn #system_info_fn #vacuum_fn + #validate_loaded_secondary_state_fn + } + } + } + + fn gen_validate_loaded_secondary_state_fn(&self) -> TokenStream { + if self.columns.indexes.is_empty() { + return quote! { + fn validate_loaded_secondary_state(&self, _path: &str) -> Result<(), PersistenceLoadError> { + Ok(()) + } + }; + } + + let expected_entries = self + .columns + .indexes + .iter() + .map(|(column, index)| { + let index_field = &index.name; + let row_field = &index.field; + let index_name = Literal::string(&index_field.to_string()); + let field_type = self + .columns + .columns_map + .get(column) + .expect("indexed column should exist") + .to_string(); + let key = if is_float(&field_type) { + quote! { OrderedFloat(row.#row_field) } + } else { + quote! { row.#row_field.clone() } + }; + + if index.is_unique { + quote! { + if self.0.indexes.#index_field.lookup_for_select(&#key).map(|link| link.0) != Some(offset_link.0) { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} does not reference primary key {primary_key:?}", #index_name), + )); + } + } + } else { + quote! { + if !self.0.indexes.#index_field + .get(&#key) + .any(|(_, candidate_link)| candidate_link.0 == offset_link.0) + { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} does not reference primary key {primary_key:?}", #index_name), + )); + } + } + } + }) + .collect::>(); + let entry_counts = self + .columns + .indexes + .values() + .map(|index| { + let index_field = &index.name; + let index_name = Literal::string(&index_field.to_string()); + quote! { + if self.0.indexes.#index_field.len() != primary_count { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} contains a different number of rows than the primary index", #index_name), + )); + } + } + }) + .collect::>(); + + quote! { + fn validate_loaded_secondary_state(&self, path: &str) -> Result<(), PersistenceLoadError> { + let primary_count = self.0.primary_index.pk_map.len(); + #(#entry_counts)* + for (primary_key, offset_link) in self.0.primary_index.pk_map.iter_values() { + let row = self.0.data.select_non_ghosted_checked(offset_link.0).map_err(|error| { + PersistenceLoadError::corrupt( + path, + format!("primary key {primary_key:?} references an invalid row: {error}"), + ) + })?; + #(#expected_entries)* + } + Ok(()) } } } @@ -123,7 +214,15 @@ impl PersistGenerator { + 'static, C: Clone + PersistenceConfig, { - async fn new(engine: E) -> eyre::Result { + async fn new(mut engine: E) -> eyre::Result { + let schema = Self::space_info_default().inner; + engine + .ensure_schema( + schema.row_schema, + schema.primary_key_fields, + schema.secondary_index_types, + ) + .await?; let mut inner = WorkTable::default(); inner.table_name = #table_name; #index_setup @@ -133,13 +232,23 @@ impl PersistGenerator { )) } - async fn load(engine: E) -> eyre::Result { - let table_path = engine.config().table_path(); - if !std::path::Path::new(table_path).exists() { + async fn load(mut engine: E) -> eyre::Result { + let schema = Self::space_info_default().inner; + engine + .validate_schema( + schema.row_schema, + schema.primary_key_fields, + schema.secondary_index_types, + ) + .await?; + let table_path = engine.config().table_path().to_owned(); + if !std::path::Path::new(&table_path).exists() { return Self::new(engine).await; }; - let space = #space_ident::parse_file(table_path).await?; - let table = space.into_worktable(engine).await; + let space = #space_ident::parse_file(&table_path) + .await + .map_err(|error| PersistenceLoadError::corrupt(&table_path, error))?; + let table = space.into_worktable(engine, &table_path).await?; Ok(table) } } diff --git a/codegen/src/generators/persist/table/mod.rs b/codegen/src/generators/persist/table/mod.rs index 339440fd..272fcb1c 100644 --- a/codegen/src/generators/persist/table/mod.rs +++ b/codegen/src/generators/persist/table/mod.rs @@ -126,9 +126,50 @@ impl PersistGenerator { worktables_node, )?; + let mut row_schema = self + .columns + .field_positions + .iter() + .map(|(name, position)| { + let type_name = self.columns.columns_map.get(name).expect("column exists").to_string(); + (*position, name, Literal::string(&type_name)) + }) + .collect::>(); + row_schema.sort_by_key(|(position, _, _)| *position); + let row_schema_names = row_schema.iter().map(|(_, name, _)| *name).collect::>(); + let row_schema_types = row_schema.iter().map(|(_, _, type_name)| type_name).collect::>(); + let primary_key_fields = &self.columns.primary_keys; + let secondary_indexes = self.columns.indexes.values().collect::>(); + let secondary_index_names = secondary_indexes.iter().map(|index| &index.name).collect::>(); + let secondary_index_types = secondary_indexes + .iter() + .map(|index| { + let type_name = self + .columns + .columns_map + .get(&index.field) + .expect("indexed column exists") + .to_string(); + Literal::string(&type_name) + }) + .collect::>(); + let schema_attribute = quote! { + #[table( + row_schema(#(#row_schema_names = #row_schema_types),*), + primary_key_fields(#(#primary_key_fields),*) + )] + }; + let secondary_schema_attribute = (!secondary_indexes.is_empty()).then(|| { + quote! { + #[table(secondary_index_types(#(#secondary_index_names = #secondary_index_types),*))] + } + }); + Ok(if self.config.as_ref().and_then(|c| c.page_size).is_some() { quote! { #derive + #schema_attribute + #secondary_schema_attribute pub struct #ident( WorkTable< #row_type, @@ -147,6 +188,8 @@ impl PersistGenerator { } else { quote! { #derive + #schema_attribute + #secondary_schema_attribute pub struct #ident( WorkTable< #row_type, diff --git a/codegen/src/generators/primary_key.rs b/codegen/src/generators/primary_key.rs new file mode 100644 index 00000000..931c8841 --- /dev/null +++ b/codegen/src/generators/primary_key.rs @@ -0,0 +1,45 @@ +use proc_macro2::{Ident, TokenStream}; +use quote::quote; +use syn::Index; + +pub(crate) fn gen_borrowed_primary_key_impl(ident: &Ident, types: &[&TokenStream]) -> TokenStream { + let from_key = quote! { + impl From<&#ident> for #ident { + fn from(value: &#ident) -> Self { + value.clone() + } + } + }; + + if types.len() == 1 { + let type_ = types[0]; + let from_str = (type_.to_string() == "String").then(|| { + quote! { + impl From<&str> for #ident { + fn from(value: &str) -> Self { + Self(value.to_owned()) + } + } + } + }); + quote! { + #from_key + impl From<&#type_> for #ident { + fn from(value: &#type_) -> Self { + Self(value.clone()) + } + } + #from_str + } + } else { + let positions = (0..types.len()).map(Index::from).collect::>(); + quote! { + #from_key + impl From<&(#(#types),*)> for #ident { + fn from(value: &(#(#types),*)) -> Self { + Self(#(value.#positions.clone()),*) + } + } + } + } +} diff --git a/codegen/src/generators/read_only/primary_key.rs b/codegen/src/generators/read_only/primary_key.rs index 2fe733ca..995298b4 100644 --- a/codegen/src/generators/read_only/primary_key.rs +++ b/codegen/src/generators/read_only/primary_key.rs @@ -3,6 +3,7 @@ use indexmap::IndexMap; use crate::common::model::{GeneratorType, PrimaryKey}; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; use crate::generators::index_backend::primary_key_backend_impl; +use crate::generators::primary_key::gen_borrowed_primary_key_impl; use crate::generators::read_only::ReadOnlyGenerator; use proc_macro2::{Ident, TokenStream}; @@ -63,6 +64,7 @@ impl ReadOnlyGenerator { }; let (backend_derive, backend_impl) = primary_key_backend_impl(self.columns.primary_index_backend, &ident, types)?; + let borrowed_impl = gen_borrowed_primary_key_impl(&ident, types); Ok(quote! { #[derive( @@ -87,6 +89,7 @@ impl ReadOnlyGenerator { #[rkyv(derive(PartialEq, Eq, PartialOrd, Ord, Debug))] pub struct #ident(#(#types),*); + #borrowed_impl #backend_impl }) } diff --git a/codegen/src/generators/read_only/table/impls.rs b/codegen/src/generators/read_only/table/impls.rs index 08c19fd7..79bfdb40 100644 --- a/codegen/src/generators/read_only/table/impls.rs +++ b/codegen/src/generators/read_only/table/impls.rs @@ -1,8 +1,8 @@ use convert_case::{Case, Casing}; -use proc_macro2::{Ident, Span, TokenStream}; +use proc_macro2::{Ident, Literal, Span, TokenStream}; use quote::quote; -use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; +use crate::common::name_generator::{WorktableNameGenerator, is_float, is_unsized_vec}; use crate::generators::read_only::ReadOnlyGenerator; impl ReadOnlyGenerator { @@ -25,6 +25,7 @@ impl ReadOnlyGenerator { let count_fn = self.gen_table_count_fn(); let system_info_fn = self.gen_system_info_fn(); let vacuum_fn = self.gen_table_vacuum_fn(); + let validate_loaded_secondary_state_fn = self.gen_validate_loaded_secondary_state_fn(); quote! { #persisted_impl @@ -43,6 +44,96 @@ impl ReadOnlyGenerator { #iter_with_async_fn #system_info_fn #vacuum_fn + #validate_loaded_secondary_state_fn + } + } + } + + fn gen_validate_loaded_secondary_state_fn(&self) -> TokenStream { + if self.columns.indexes.is_empty() { + return quote! { + fn validate_loaded_secondary_state(&self, _path: &str) -> Result<(), PersistenceLoadError> { + Ok(()) + } + }; + } + + let expected_entries = self + .columns + .indexes + .iter() + .map(|(column, index)| { + let index_field = &index.name; + let row_field = &index.field; + let index_name = Literal::string(&index_field.to_string()); + let field_type = self + .columns + .columns_map + .get(column) + .expect("indexed column should exist") + .to_string(); + let key = if is_float(&field_type) { + quote! { OrderedFloat(row.#row_field) } + } else { + quote! { row.#row_field.clone() } + }; + + if index.is_unique { + quote! { + if self.0.indexes.#index_field.lookup_for_select(&#key).map(|link| link.0) != Some(offset_link.0) { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} does not reference primary key {primary_key:?}", #index_name), + )); + } + } + } else { + quote! { + if !self.0.indexes.#index_field + .get(&#key) + .any(|(_, candidate_link)| candidate_link.0 == offset_link.0) + { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} does not reference primary key {primary_key:?}", #index_name), + )); + } + } + } + }) + .collect::>(); + let entry_counts = self + .columns + .indexes + .values() + .map(|index| { + let index_field = &index.name; + let index_name = Literal::string(&index_field.to_string()); + quote! { + if self.0.indexes.#index_field.len() != primary_count { + return Err(PersistenceLoadError::corrupt( + path, + format!("secondary index {} contains a different number of rows than the primary index", #index_name), + )); + } + } + }) + .collect::>(); + + quote! { + fn validate_loaded_secondary_state(&self, path: &str) -> Result<(), PersistenceLoadError> { + let primary_count = self.0.primary_index.pk_map.len(); + #(#entry_counts)* + for (primary_key, offset_link) in self.0.primary_index.pk_map.iter_values() { + let row = self.0.data.select_non_ghosted_checked(offset_link.0).map_err(|error| { + PersistenceLoadError::corrupt( + path, + format!("primary key {primary_key:?} references an invalid row: {error}"), + ) + })?; + #(#expected_entries)* + } + Ok(()) } } } @@ -113,12 +204,14 @@ impl ReadOnlyGenerator { } async fn load(engine: E) -> eyre::Result { - let table_path = engine.config().table_path(); - if !std::path::Path::new(table_path).exists() { + let table_path = engine.config().table_path().to_owned(); + if !std::path::Path::new(&table_path).exists() { return Self::new(engine).await; }; - let space = #space_ident::parse_file(table_path).await?; - let table = space.into_worktable(); + let space = #space_ident::parse_file(&table_path) + .await + .map_err(|error| PersistenceLoadError::corrupt(&table_path, error))?; + let table = space.into_worktable(&table_path)?; Ok(table) } } diff --git a/codegen/src/persist_table/generator/mod.rs b/codegen/src/persist_table/generator/mod.rs index 0ab25350..7baf87b6 100644 --- a/codegen/src/persist_table/generator/mod.rs +++ b/codegen/src/persist_table/generator/mod.rs @@ -14,6 +14,9 @@ pub struct PersistTableAttributes { pub pk_upstream: bool, pub pk_arctic: bool, pub pk_congee: bool, + pub row_schema: Vec<(String, String)>, + pub primary_key_fields: Vec, + pub secondary_index_types: Vec<(String, String)>, } pub struct Generator { diff --git a/codegen/src/persist_table/generator/space_file/mod.rs b/codegen/src/persist_table/generator/space_file/mod.rs index e3598653..8120e6eb 100644 --- a/codegen/src/persist_table/generator/space_file/mod.rs +++ b/codegen/src/persist_table/generator/space_file/mod.rs @@ -66,6 +66,17 @@ impl Generator { } else { quote! { self.primary_index.0.len() as u32 + self.primary_index.1.len() as u32 } }; + let row_schema = self.attributes.row_schema.iter().map(|(name, type_name)| { + quote! { (#name.to_string(), #type_name.to_string()) } + }); + let primary_key_fields = self + .attributes + .primary_key_fields + .iter() + .map(|name| quote! { #name.to_string() }); + let secondary_index_types = self.attributes.secondary_index_types.iter().map(|(name, type_name)| { + quote! { (#name.to_string(), #type_name.to_string()) } + }); quote! { fn get_primary_index_info(&self) -> eyre::Result>> { @@ -77,9 +88,9 @@ impl Generator { name: #literal_name.to_string(), pk_gen_state: (), empty_links_list: vec![], - primary_key_fields: vec![], - row_schema: vec![], - secondary_index_types: vec![], + primary_key_fields: vec![#(#primary_key_fields),*], + row_schema: vec![#(#row_schema),*], + secondary_index_types: vec![#(#secondary_index_types),*], }; let header = GeneralHeader { data_version: DATA_VERSION, @@ -200,7 +211,7 @@ impl Generator { if self.attributes.read_only { quote! { - pub fn into_worktable(self) -> #wt_ident { + pub fn into_worktable(self, path: &str) -> Result<#wt_ident, PersistenceLoadError> { let mut page_id = 1; let data = self.data.into_iter().map(|p| { let mut data = Data::from_data_page(p); @@ -227,12 +238,19 @@ impl Generator { pk_phantom: std::marker::PhantomData, }; - #wt_ident(table) + table.validate_persisted_state(path)?; + let worktable = #wt_ident(table); + worktable.validate_loaded_secondary_state(path)?; + Ok(worktable) } } } else { quote! { - pub async fn into_worktable(self, engine: E) -> #wt_ident + pub async fn into_worktable( + self, + engine: E, + path: &str, + ) -> Result<#wt_ident, PersistenceLoadError> where E: PersistenceEngine< <<#pk_type as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State, @@ -270,10 +288,13 @@ impl Generator { pk_phantom: std::marker::PhantomData, }; - #wt_ident( + table.validate_persisted_state(path)?; + let worktable = #wt_ident( table, #task_ident::run_engine(engine) - ) + ); + worktable.validate_loaded_secondary_state(path)?; + Ok(worktable) } } } diff --git a/codegen/src/persist_table/generator/space_file/worktable_impls.rs b/codegen/src/persist_table/generator/space_file/worktable_impls.rs index 579c24a2..0b69fe62 100644 --- a/codegen/src/persist_table/generator/space_file/worktable_impls.rs +++ b/codegen/src/persist_table/generator/space_file/worktable_impls.rs @@ -11,6 +11,7 @@ impl Generator { let persisted_pk_fn = self.gen_worktable_persisted_primary_key_fn(); let wait_for_ops_fn = self.gen_worktable_wait_for_ops_fn(); let close_fn = self.gen_worktable_close_fn(); + let persisted_data_file_size_fn = self.gen_persisted_data_file_size_fn(); quote! { impl #ident { @@ -18,6 +19,22 @@ impl Generator { #persisted_pk_fn #wait_for_ops_fn #close_fn + #persisted_data_file_size_fn + } + } + } + + fn gen_persisted_data_file_size_fn(&self) -> TokenStream { + if self.attributes.read_only { + quote! {} + } else { + quote! { + /// Returns the physical size of this table's `.wt.data` file. + /// Persisted vacuum currently compacts logical/in-memory pages + /// but does not truncate this file. + pub async fn persisted_data_file_size_bytes(&self) -> std::io::Result { + self.1.persisted_data_file_size_bytes().await + } } } } @@ -59,6 +76,17 @@ impl Generator { let pk = name_generator.get_primary_key_type_ident(); let literal_name = name_generator.get_work_table_literal_name(); let version_const = name_generator.get_version_const_ident(); + let row_schema = self.attributes.row_schema.iter().map(|(name, type_name)| { + quote! { (#name.to_string(), #type_name.to_string()) } + }); + let primary_key_fields = self + .attributes + .primary_key_fields + .iter() + .map(|name| quote! { #name.to_string() }); + let secondary_index_types = self.attributes.secondary_index_types.iter().map(|(name, type_name)| { + quote! { (#name.to_string(), #type_name.to_string()) } + }); quote! { pub fn space_info_default() -> GeneralPage::Generator as PrimaryKeyGeneratorState>::State>> { @@ -69,9 +97,9 @@ impl Generator { name: #literal_name.to_string(), pk_gen_state: <<#pk as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State::default(), empty_links_list: vec![], - primary_key_fields: vec![], - row_schema: vec![], - secondary_index_types: vec![], + primary_key_fields: vec![#(#primary_key_fields),*], + row_schema: vec![#(#row_schema),*], + secondary_index_types: vec![#(#secondary_index_types),*], }; let header = GeneralHeader { data_version: DATA_VERSION, diff --git a/codegen/src/persist_table/mod.rs b/codegen/src/persist_table/mod.rs index 530e6e3f..91d598bf 100644 --- a/codegen/src/persist_table/mod.rs +++ b/codegen/src/persist_table/mod.rs @@ -67,7 +67,7 @@ mod tests { "read_only should not generate PersistenceTask" ); assert!( - output.contains("fn into_worktable (self)"), + output.contains("fn into_worktable (self , path : & str)"), "read_only should have sync into_worktable without engine param" ); assert!( diff --git a/codegen/src/persist_table/parser.rs b/codegen/src/persist_table/parser.rs index 4a654ff2..26773d95 100644 --- a/codegen/src/persist_table/parser.rs +++ b/codegen/src/persist_table/parser.rs @@ -2,7 +2,7 @@ use crate::persist_table::generator::PersistTableAttributes; use proc_macro2::{Ident, Span, TokenStream}; use quote::ToTokens; use syn::spanned::Spanned; -use syn::{Attribute, ItemStruct}; +use syn::{Attribute, ItemStruct, LitStr}; pub struct Parser; @@ -32,6 +32,9 @@ impl Parser { pk_upstream: false, pk_arctic: false, pk_congee: false, + row_schema: vec![], + primary_key_fields: vec![], + secondary_index_types: vec![], }; for attr in attrs { @@ -57,6 +60,44 @@ impl Parser { res.pk_congee = true; return Ok(()); } + if meta.path.is_ident("row_schema") { + meta.parse_nested_meta(|field| { + let name = field + .path + .get_ident() + .ok_or_else(|| field.error("row schema field must be an identifier"))? + .to_string(); + let type_name = field.value()?.parse::()?.value(); + res.row_schema.push((name, type_name)); + Ok(()) + })?; + return Ok(()); + } + if meta.path.is_ident("primary_key_fields") { + meta.parse_nested_meta(|field| { + let name = field + .path + .get_ident() + .ok_or_else(|| field.error("primary key field must be an identifier"))? + .to_string(); + res.primary_key_fields.push(name); + Ok(()) + })?; + return Ok(()); + } + if meta.path.is_ident("secondary_index_types") { + meta.parse_nested_meta(|index| { + let name = index + .path + .get_ident() + .ok_or_else(|| index.error("secondary index name must be an identifier"))? + .to_string(); + let type_name = index.value()?.parse::()?.value(); + res.secondary_index_types.push((name, type_name)); + Ok(()) + })?; + return Ok(()); + } Ok(()) }) .expect("always ok even on unrecognized attrs"); diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs index 0097313f..fe993415 100644 --- a/codegen/src/worktable/mod.rs +++ b/codegen/src/worktable/mod.rs @@ -37,6 +37,12 @@ pub fn expand(input: TokenStream) -> syn::Result { "version must be specified before columns/indexes/queries/config", )); } + "attributes" => { + return Err(syn::Error::new( + ident.span(), + "a separate `attributes` section is not part of the 1.0 grammar; keep `primary_key`, `autoincrement`, `custom`, `optional`, and `using` inline on their column or index declarations", + )); + } _ => return Err(syn::Error::new(ident.span(), "Unexpected identifier")), } } @@ -137,6 +143,23 @@ mod tests { use super::expand; + #[test] + fn separate_attributes_section_has_an_actionable_1_0_diagnostic() { + let error = expand(quote! { + name: AttributesSection, + columns: { + id: u64 primary_key, + }, + attributes: { + id: primary_key, + }, + }) + .unwrap_err(); + + assert!(error.to_string().contains("not part of the 1.0 grammar")); + assert!(error.to_string().contains("keep `primary_key`")); + } + fn assert_composite_primary_key_field_order(output: proc_macro2::TokenStream) { let output = output.to_string(); let get_primary_key = output diff --git a/docs/crate.md b/docs/crate.md new file mode 100644 index 00000000..dc382cca --- /dev/null +++ b/docs/crate.md @@ -0,0 +1,114 @@ +# WorkTable + +WorkTable is a typed, macro-generated embedded table for Rust. It provides +primary and secondary indexes, generated CRUD/query methods, optional local or +S3-backed persistence, and per-table concurrency. It is not a SQL database and +does not provide multi-table transactions or multi-process access. + +## In-memory quick start + +```rust +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: User, + columns: { + id: u64 primary_key, + email: String, + }, + indexes: { + email_idx: email unique, + }, +); + +let table = UserWorkTable::default(); +let row = UserRow { + id: 1, + email: "person@example.com".to_owned(), +}; +table.insert(row.clone()).unwrap(); +assert_eq!(table.select(1), Some(row.clone())); +assert_eq!(table.select_by_email("person@example.com".to_owned()), Some(row)); +``` + +String and tuple primary keys accept borrowed forms, so callers do not need to +write an explicit clone merely to perform a lookup or delete. + +```rust +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: Account, + columns: { + tenant: String primary_key, + account: String primary_key, + enabled: bool, + }, +); + +let table = AccountWorkTable::default(); +let key = ("tenant-a".to_owned(), "account-1".to_owned()); +let row = AccountRow { + tenant: key.0.clone(), + account: key.1.clone(), + enabled: true, +}; +table.insert(row.clone()).unwrap(); +assert_eq!(table.select(&key), Some(row)); +``` + +## Persistence contract + +Persistence is an optional background write path. `insert`, `update`, and +`delete` returning means the in-memory mutation was accepted and its +persistence operation was queued; it does not mean the operation reached stable +storage. `wait_for_ops()` means the persistence engine completed the queued +operations. `close()` stops intake, drains, and joins the engine task. Neither +call currently issues an fsync or stable-storage guarantee, and neither makes +an in-place batch atomic against process death or power loss. + +A graceful persistence failure is terminal and is returned by later mutations, +`wait_for_ops()`, and `close()`. Abrupt termination can lose acknowledged rows, +leave a torn file, or produce bytes that pass structural validation but do not +represent a row that was written. Persistence is therefore best-effort in the +1.0 beta line; applications that require crash durability need an external +snapshot/rebuild strategy. + +Persisted `SpaceInfo` records the generated row schema, primary-key fields, and +secondary-index types. Existing legacy stores whose metadata is completely +empty remain readable but cannot be schema-validated, and loading them does not +rewrite the file. A non-empty schema mismatch is rejected before row loading. + +Persisted vacuum compacts the live in-memory layout and keeps persisted indexes +consistent with moved rows. It does not truncate `.wt.data`. Generated +persisted tables expose `persisted_data_file_size_bytes()` so operators can +observe physical growth and schedule replacement/offline compaction. + +## Column grammar + +The 1.0 grammar keeps column modifiers inline: + +```text +id: u64 primary_key autoincrement, +tenant: String primary_key, +nickname: String optional, +id: u64 primary_key using congee, +``` + +A separate `attributes` section is intentionally not part of the 1.0 grammar. +The macro rejects it with a migration-oriented diagnostic rather than silently +accepting a second spelling immediately before the DSL is frozen. + +## Concurrency boundary + +Generated reads use immutable row publication and a read grace period. Point +and range lookups return complete owned row versions, and retired links/pages +are not reused while a reader could still resolve them. This is not MVCC: a +range scan is not a snapshot, and disjoint archived-page writes currently share +a table-wide writer barrier. Direct use of low-level page mutation APIs is +outside the generated safe-API guarantee. + +See the repository README and `docs/versioned-row-publication.md` for backend, +query, and implementation details. diff --git a/docs/persistence-durability.md b/docs/persistence-durability.md new file mode 100644 index 00000000..9f2e3215 --- /dev/null +++ b/docs/persistence-durability.md @@ -0,0 +1,83 @@ +# Persistence durability and recovery + +WorkTable is an embedded in-memory table with optional **best-effort persistence**. +Its current local-disk and S3 paths are not a crash-atomic database: there is no +transaction journal spanning data, primary-index, and secondary-index files, and +ordinary disk batches are flushed but not synchronously committed to stable media. + +This is an explicit product boundary, not an implied durability guarantee. + +## Guarantee matrix + +| Boundary | Guaranteed | Not guaranteed | +|---|---|---| +| Mutation returns `Ok` | The in-memory mutation completed and its persistence operation was accepted by the running queue. | The bytes have reached the OS, disk, or S3. | +| `wait_for_ops()` returns `Ok` | The local persistence task reached an idle point: its queue and analyzer are empty and no batch is in flight. Errors observed by the worker are surfaced. | Intake is not closed; concurrent or later writers may queue more work. File `flush` is not `fsync`, and a multi-file batch is not crash-atomic. | +| `close()` returns `Ok` | Intake is closed, queued work is drained, and the persistence worker has joined without a reported error. | Power-loss durability or atomicity across data and index files. | +| Graceful process exit after `close()` | The WorkTable worker completed all writes it reported. | Survival of a subsequent power loss before the operating system commits buffered writes. | +| Process crash or `SIGKILL` | No row-fidelity guarantee for an interrupted batch. The next load either returns a state whose primary links and rows validate, or returns `PersistenceLoadError`. | Preservation of the latest acknowledged changes. | +| Power loss | The next load applies the same validation/refusal boundary. | Any acknowledged-change retention window; current batches do not call `fsync`. | +| S3 synchronization | Successful calls report completion of the configured upload path. | A transactionally consistent multi-file snapshot. Treat independently uploaded objects as best-effort unless an application-managed snapshot generation protects them. | + +Call `close()` during orderly shutdown. If `wait_for_ops()` is used before a +non-consuming shutdown path, stop application writers first; otherwise a writer can +enqueue new work after the task appears idle. + +## Load validation + +Loading an existing table performs an additional startup-only audit before the +persistence worker starts: + +- persisted archived rows referenced by the primary index must pass rkyv validation; +- each physical link must be in the initialized part of its page; +- each decoded row's primary key must equal the primary-index key; +- no two primary keys may reference the same physical link; and +- forward and reverse primary indexes must agree; and +- every secondary index must contain exactly one correct entry for each loaded row. + +Parsing failures and audit failures are returned as `PersistenceLoadError`. The public +`PersistedWorkTable::load` API still returns `eyre::Result`, so callers can identify the +typed outcome without string matching: + +```rust +match MyWorkTable::load(engine).await { + Ok(table) => use_table(table), + Err(report) => { + if let Some(corruption) = report.downcast_ref::() { + eprintln!("refusing {}: {}", corruption.path().display(), corruption.reason()); + restore_or_rebuild(corruption.path()); + } else { + return Err(report); + } + } +} +``` + +The audit is proportional to the number of primary-index entries. It runs only during +`load()` and adds no branch, lock, or scan to steady-state insert, select, update, or +delete paths. + +## Supported recovery procedure + +`PersistenceLoadError` is a refusal boundary. Do not continue writing to the rejected +directory and do not replace individual index or data files in place: the files are one +logical generation even though the format cannot commit them atomically. + +1. Stop every process that can write the table. +2. Preserve the rejected table directory for diagnosis. +3. Restore the **entire** table directory from one application-managed snapshot; or + create a new empty table directory and replay rows from an external authoritative + source or event log. +4. Open the restored/rebuilt directory and require `load()` to pass before serving it. + +WorkTable does not currently provide an in-place salvage tool that can prove which +side of a torn multi-file batch is authoritative. Full-directory restore or clean +replay is the supported recovery path. If neither exists, acknowledged data may be +unrecoverable under this best-effort contract. + +## When stronger durability is required + +Use a durable database or place WorkTable behind an authoritative log/snapshot system +when acknowledged writes must survive process or power loss. Making WorkTable itself +crash-atomic would require a separately designed and tested journal, shadow-page, or +generation-manifest protocol; it is not claimed by this contract. diff --git a/docs/pr46-review-findings.md b/docs/pr46-review-findings.md new file mode 100644 index 00000000..cae1954b --- /dev/null +++ b/docs/pr46-review-findings.md @@ -0,0 +1,116 @@ +# PR #46 review findings — status & fixes + +Fresh-eyes review of `fix/v1-blockers-consolidated` (the v1-blocker PR). This +document tracks each finding, whether it reproduced under test, and its fix. +Work lands on branch `fix/pr46-review-findings`. + +## F1 — Mutation gate held across `.await` (was rated P1) — NOT REPRODUCED + +`LockMap::mutation_guard` (`src/lock/map.rs`) is a blocking spin/yield ticket +lock. The generated `update`/`in_place`/`delete` paths keep the `MutationGuard` +inside `LockGuard` and hold it across `.await` (`update_with_guard(...).await`, +`reinsert(...).await`). The concern: two keys colliding on the same 1-of-64 +stripe, guard-holder parked at its await while the other task spins. + +**Result:** `tests/worktable/mutation_gate_deadlock.rs` reproduces the scenario +(colliding keys, single-worker and 2-worker runtimes, updates that await while +holding the gate) and **passes**. tokio schedules async tasks cooperatively and +the spinner falls back to `thread::yield_now()`, so the parked holder is still +polled to completion on the same thread. No livelock. + +**Disposition:** finding downgraded. The test is kept as a standing guard (with +timeouts, so it can never hang the harness) to catch a future regression that +would make the hazard real (e.g. a blocking, non-cooperative holder). + +## F2 — Vacuum can mark a live `page_from` empty (P1, data-loss) — NOT REPRODUCED + +`src/table/vacuum/vacuum.rs`: `page_from` is `mark_page_empty`'d unconditionally +after the inner loop, with no `page_from != page_to` guard. The concern: the +destination search falls through to `allocate_new_or_pop_free()` and returns a +page that ends up holding moved-in rows, which is then reclaimed. + +**Result:** `tests/worktable/vacuum_no_row_loss.rs` forces heavy cross-page +compaction (400 large rows, half deleted from many pages) with a concurrent +grace-period reader, then audits every survivor by primary key AND unique index +after vacuum quiesces. It **passes** — no row loss, no resurrection. The +grace-period deferral added in #46 (`allocate_new_or_pop_free` returns a temp +page instead of reusing an active source) appears to hold: `page_from` is not +handed back as a destination while it still holds live rows. + +**Disposition:** finding not reproduced at this scale; kept as a standing audit. +A `debug_assert!(page_from != page_to)` in the loop would make the invariant +explicit and cheap to enforce — recommended as a belt-and-suspenders follow-up. + +## F3 — Temp destination page mistracking (P2) + +A destination from `allocate_new_or_pop_free()` may be `mark_page_full`'d while +partially empty (wasted capacity) or dropped from all tracking sets (leaked). + +## F4 — Full-row unsized `update()` still always reinserts (P2, confirmed) + +`codegen/src/generators/in_memory/queries/update.rs` full-row path still has +`if true { reinsert }` for unsized rows. The custom-update path's `gen_size_check` +(`need_to_reinsert = true` initializer) is the other half. The overwrite perf +blocker is only partially documented; see `tests/worktable/update_in_place_unsized.rs`. + +### F4 — precise root cause of the corruption (why the one-liner isn't safe) + +The generated in-place update mutates a field with +`std::mem::swap(&mut archived.inner., &mut archived_row.)` inside +`with_mut_ref` (`archived` = the slot's bytes; `archived_row` = the freshly +serialized `bytes` buffer). WorkTable does **field-level** updates, so only the +changed field is swapped — the other fields are untouched. + +- For inline-representable fields (`u64`, and **short** strings ≤ + `rkyv::string::repr::INLINE_CAPACITY`), the archived value is stored inline, so + swapping the bytes is self-contained and correct. +- `ArchivedString` is a **union** (`ArchivedStringRepr`): a **long** string is + stored **out-of-line** as a *relative pointer + length*, with the character + bytes elsewhere in the buffer. `mem::swap` moves the relative pointer into the + slot, but the characters it points at live in `archived_row`'s buffer, which is + never written to the slot. The relative offset now points outside the slot → + reads come back as raw/garbage bytes. **This** is the corruption seen in + `update_parallel_more_strings` / `update_many_times` when the reinsert guard is + naively flipped to `false`. + +**Correct fix (not a one-liner; unsafe archived-memory work):** +`gen_size_check` should reinsert only when the new field value does **not** fit +its current slot region. When it fits (same-or-shorter serialized length, +including same-length), the in-place write for an out-of-line `String` must +overwrite the existing out-of-line byte region — e.g. via +`ArchivedStringRepr::as_bytes_seal` — instead of `mem::swap`ping the pointer. +Field-level semantics must be preserved (never overwrite the whole slot; only the +changed field's region). Because this manipulates archived memory directly, a +subtle error is silent data corruption, so it should not be rushed alongside a +release merge. `tests/worktable/update_in_place_unsized.rs` is the proof harness; +remove its `#[ignore]` once the in-place String write is correct. + +## F5 — 64-stripe gate serializes unrelated hot keys (P2, efficiency) + +`MUTATION_STRIPE_COUNT = 64`. Distinct hot keys colliding on a stripe are +serialized though they never touch the same row — relevant to the low-latency +claim; document the ceiling or scale it with expected concurrency. + +## F6 — Wedged stripe if a guard is leaked, no timeout (P2) + +`next_ticket`/`serving` are free-running with no timeout/poisoning; a leaked +`LockGuard` (mem::forget, Arc cycle) wedges 1/64 of keys forever with no +diagnostic. Low likelihood, but silent if it happens. + +## Leak root cause (the ~190GB orphaned test processes) — FOUND, already fixed + +The Activity Monitor screenshot showed two orphaned `mod-…` processes (WorkTable's +`tests/mod.rs` integration binary) at ~190GB each. Root cause: `vacuum_loop_test` +(`tests/worktable/vacuum.rs`) is a soak test that inserts a fresh row every 500µs +while vacuum reclaims outdated ones. If it runs unbounded (or vacuum can't keep up), +rows accumulate without limit and the harness process balloons. + +Already fixed on this PR branch by commit `111f61e` +("test: bound vacuum soak to prevent orphaned harnesses"): the test is now +`#[ignore]`d and hard-bounded to a 10-second `SOAK_DURATION` with an explicit +`stop_at` deadline on both the insert loop and the vacuum-observe loop. It can no +longer run forever or orphan the harness. + +Standing guard added: `tests/worktable/leak_probe.rs` asserts that 5000 same-key +updates keep physical page count bounded (reclamation keeps up) — passes, so the +reinsert-per-update path is not itself a leak. diff --git a/src/in_memory/data.rs b/src/in_memory/data.rs index 0202c015..789bb83b 100644 --- a/src/in_memory/data.rs +++ b/src/in_memory/data.rs @@ -204,6 +204,35 @@ impl Data { rkyv::deserialize::<_, rkyv::rancor::Error>(row).map_err(|_| ExecutionError::DeserializeError) } + /// Validates persisted bytes before deserializing them. + /// + /// The regular in-memory path only reads bytes written by WorkTable in the + /// same process. Loading persisted data is different: an abrupt shutdown + /// may leave a link or archived value only partially written. This method + /// is deliberately reserved for load-time validation so steady-state row + /// reads keep their existing cost. + pub fn get_row_checked(&self, link: Link) -> Result + where + Row: Archive, + ::Archived: Portable + + Deserialize> + + for<'a> rkyv::bytecheck::CheckBytes>, + { + let start = link.offset as usize; + let end = start + .checked_add(link.length as usize) + .ok_or(ExecutionError::InvalidLink)?; + let initialized = self.free_offset.load(Ordering::Acquire) as usize; + if link.length == 0 || end > initialized || end > DATA_LENGTH { + return Err(ExecutionError::InvalidLink); + } + + let inner_data = unsafe { &*self.inner_data.get() }; + let archived = rkyv::access::<::Archived, rkyv::rancor::Error>(&inner_data[start..end]) + .map_err(|_| ExecutionError::DeserializeError)?; + rkyv::deserialize::<_, rkyv::rancor::Error>(archived).map_err(|_| ExecutionError::DeserializeError) + } + pub fn get_raw_row(&self, link: Link) -> Result, ExecutionError> { let inner_data = unsafe { &mut *self.inner_data.get() }; let bytes = &mut inner_data[link.offset as usize..(link.offset + link.length) as usize]; diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index 987639f9..1aaa4e4f 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -505,6 +505,34 @@ where Ok(row.as_ref().clone()) } + /// Loads one persisted row through rkyv validation without publishing it. + /// + /// This is used by the table load audit before the persistence worker is + /// started. It is intentionally separate from the steady-state read path. + pub fn select_non_ghosted_checked(&self, link: Link) -> Result + where + <::WrappedRow as Archive>::Archived: Portable + + Deserialize<::WrappedRow, HighDeserializer> + + for<'a> rkyv::bytecheck::CheckBytes>, + { + let pages = self.pages.read(); + let page_id: usize = link.page_id.into(); + let page_index = page_id + .checked_sub(1) + .ok_or(ExecutionError::PageNotFound(link.page_id))?; + let page = pages + .get(page_index) + .ok_or(ExecutionError::PageNotFound(link.page_id))?; + let wrapped = page.get_row_checked(link).map_err(ExecutionError::DataPageError)?; + if wrapped.is_ghosted() { + return Err(ExecutionError::Ghosted); + } + if wrapped.is_deleted() { + return Err(ExecutionError::Deleted); + } + Ok(wrapped.get_inner()) + } + pub fn select_non_vacuumed(&self, link: Link) -> Result where Row: Archive diff --git a/src/lib.rs b/src/lib.rs index e1d4dbe5..5ba6ae0c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,5 @@ +#![doc = include_str!("../docs/crate.md")] + pub mod in_memory; mod index; pub mod lock; @@ -13,7 +15,7 @@ mod util; pub mod features; pub use index::*; -pub use persistence::{PersistedWorkTable, PersistenceConfig}; +pub use persistence::{PersistedWorkTable, PersistenceConfig, PersistenceLoadError}; pub use row::*; pub use table::*; @@ -34,10 +36,11 @@ pub mod prelude { pub use crate::persistence::{ AcknowledgeOperation, ArtPersistenceKey, DeleteOperation, DiskConfig, DiskPersistenceEngine, IndexTableOfContents, InsertOperation, Operation, OperationId, PersistedWorkTable, PersistenceConfig, - PersistenceEngine, PersistenceError, PersistenceResult, PersistenceState, PersistenceTask, - ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, SpaceDataOps, SpaceIndex, - SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events, + PersistenceEngine, PersistenceError, PersistenceLoadError, PersistenceResult, PersistenceState, + PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, SpaceDataOps, + SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, + map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, + validate_events, }; pub use crate::primary_key::{PrimaryKeyGenerator, PrimaryKeyGeneratorState, TablePrimaryKey}; pub use crate::table::select::{Order, QueryParams, SelectQueryBuilder, SelectQueryExecutor}; diff --git a/src/persistence/engine.rs b/src/persistence/engine.rs index 52bfefd9..6e929bc7 100644 --- a/src/persistence/engine.rs +++ b/src/persistence/engine.rs @@ -1,17 +1,49 @@ use std::fmt::Debug; use std::fs; +use std::future::Future; use std::hash::Hash; use std::marker::PhantomData; +use std::panic::{AssertUnwindSafe, resume_unwind}; use std::path::Path; -use futures::StreamExt; use futures::future::Either; use futures::stream::FuturesUnordered; +use futures::{FutureExt, StreamExt}; use crate::TableSecondaryIndexEventsOps; use crate::persistence::operation::{BatchOperation, Operation}; -use crate::persistence::{PersistenceConfig, PersistenceEngine, SpaceDataOps, SpaceIndexOps, SpaceSecondaryIndexOps}; -use crate::prelude::{PrimaryKeyGeneratorState, TablePrimaryKey}; +use crate::persistence::{ + PersistenceConfig, PersistenceEngine, PersistenceLoadError, SpaceDataOps, SpaceIndexOps, SpaceSecondaryIndexOps, +}; +use crate::prelude::{PrimaryKeyGeneratorState, TablePrimaryKey, WT_DATA_EXTENSION}; + +fn classify_existing_store_error(path: &str, existed: bool, result: eyre::Result) -> eyre::Result { + result.map_err(|error| { + if existed { + PersistenceLoadError::corrupt(path, format!("{error:#}")).into() + } else { + error + } + }) +} + +async fn load_store_component(path: &str, existed: bool, future: F) -> eyre::Result +where + F: Future>, +{ + match AssertUnwindSafe(future).catch_unwind().await { + Ok(result) => classify_existing_store_error(path, existed, result), + Err(payload) if existed => { + let reason = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or("persisted-state loader panicked"); + Err(PersistenceLoadError::corrupt(path, reason).into()) + } + Err(payload) => resume_unwind(payload), + } +} #[derive(Debug, Clone)] pub struct DiskConfig { @@ -72,6 +104,7 @@ pub struct DiskPersistenceEngine< pub data: SpaceData, pub primary_index: SpacePrimaryIndex, pub secondary_indexes: SpaceSecondaryIndexes, + created_data_file: bool, phantom_data: PhantomData<(PrimaryKey, SecondaryIndexEvents, PrimaryKeyGenState, AvailableIndexes)>, } @@ -110,16 +143,37 @@ where Self: Sized, { let table_path = Path::new(&config.tables_path); + let created_data_file = !table_path.join(WT_DATA_EXTENSION).exists(); if !table_path.exists() { fs::create_dir_all(table_path)?; } + let existed = !created_data_file; + + let data = load_store_component( + &config.tables_path, + existed, + SpaceData::from_table_files_path(config.tables_path.clone(), config.version), + ) + .await?; + let primary_index = load_store_component( + &config.tables_path, + existed, + SpacePrimaryIndex::primary_from_table_files_path(config.tables_path.clone(), config.version), + ) + .await?; + let secondary_indexes = load_store_component( + &config.tables_path, + existed, + SpaceSecondaryIndexes::from_table_files_path(config.tables_path.clone(), config.version), + ) + .await?; Ok(Self { config: config.clone(), - data: SpaceData::from_table_files_path(config.tables_path.clone(), config.version).await?, - primary_index: SpacePrimaryIndex::primary_from_table_files_path(config.tables_path.clone(), config.version) - .await?, - secondary_indexes: SpaceSecondaryIndexes::from_table_files_path(config.tables_path, config.version).await?, + data, + primary_index, + secondary_indexes, + created_data_file, phantom_data: PhantomData, }) } @@ -231,6 +285,83 @@ where Ok(()) } + async fn ensure_schema( + &mut self, + row_schema: Vec<(String, String)>, + primary_key_fields: Vec, + secondary_index_types: Vec<(String, String)>, + ) -> eyre::Result<()> { + let info = self.data.get_mut_info(); + let legacy_empty = info.inner.row_schema.is_empty() + && info.inner.primary_key_fields.is_empty() + && info.inner.secondary_index_types.is_empty(); + + if legacy_empty { + info.inner.row_schema = row_schema; + info.inner.primary_key_fields = primary_key_fields; + info.inner.secondary_index_types = secondary_index_types; + return self.data.save_info().await; + } + + if info.inner.row_schema != row_schema + || info.inner.primary_key_fields != primary_key_fields + || info.inner.secondary_index_types != secondary_index_types + { + return Err(eyre::eyre!( + "persisted schema mismatch for {}: stored row schema {:?}, primary key {:?}, indexes {:?}; generated row schema {:?}, primary key {:?}, indexes {:?}", + info.inner.name, + info.inner.row_schema, + info.inner.primary_key_fields, + info.inner.secondary_index_types, + row_schema, + primary_key_fields, + secondary_index_types, + )); + } + + Ok(()) + } + + async fn validate_schema( + &mut self, + row_schema: Vec<(String, String)>, + primary_key_fields: Vec, + secondary_index_types: Vec<(String, String)>, + ) -> eyre::Result<()> { + let info = self.data.get_mut_info(); + let legacy_empty = info.inner.row_schema.is_empty() + && info.inner.primary_key_fields.is_empty() + && info.inner.secondary_index_types.is_empty(); + + if legacy_empty { + if self.created_data_file { + info.inner.row_schema = row_schema; + info.inner.primary_key_fields = primary_key_fields; + info.inner.secondary_index_types = secondary_index_types; + return self.data.save_info().await; + } + return Ok(()); + } + + if info.inner.row_schema != row_schema + || info.inner.primary_key_fields != primary_key_fields + || info.inner.secondary_index_types != secondary_index_types + { + return Err(eyre::eyre!( + "persisted schema mismatch for {}: stored row schema {:?}, primary key {:?}, indexes {:?}; generated row schema {:?}, primary key {:?}, indexes {:?}", + info.inner.name, + info.inner.row_schema, + info.inner.primary_key_fields, + info.inner.secondary_index_types, + row_schema, + primary_key_fields, + secondary_index_types, + )); + } + + Ok(()) + } + fn config(&self) -> &DiskConfig { &self.config } diff --git a/src/persistence/error.rs b/src/persistence/error.rs index f8b48d5e..bbd98dd2 100644 --- a/src/persistence/error.rs +++ b/src/persistence/error.rs @@ -1,7 +1,52 @@ use std::error::Error; use std::fmt::{Display, Formatter}; +use std::path::{Path, PathBuf}; use std::sync::Arc; +/// A persisted table could not be loaded without risking invalid data. +/// +/// WorkTable persistence is best-effort rather than crash-atomic. Abrupt +/// process or power loss may therefore leave a partial batch on disk. `load` +/// reports that condition with this concrete error type instead of exposing +/// torn bytes as rows. Callers using the `eyre::Result`-based +/// [`PersistedWorkTable`](crate::persistence::PersistedWorkTable) API can +/// identify it with [`eyre::Report::downcast_ref`]. +#[derive(Debug)] +pub struct PersistenceLoadError { + path: PathBuf, + reason: String, +} + +impl PersistenceLoadError { + pub fn corrupt(path: impl AsRef, reason: impl Display) -> Self { + Self { + path: path.as_ref().to_path_buf(), + reason: reason.to_string(), + } + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn reason(&self) -> &str { + &self.reason + } +} + +impl Display for PersistenceLoadError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "torn or corrupt persisted table at {}: {}", + self.path.display(), + self.reason + ) + } +} + +impl Error for PersistenceLoadError {} + /// Terminal and lifecycle errors reported by a persistence task. #[derive(Debug)] pub enum PersistenceError { diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index 3cbaefaf..94917fe9 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -4,7 +4,7 @@ use crate::persistence::operation::BatchOperation; pub use engine::DiskConfig; pub use engine::DiskPersistenceEngine; -pub use error::{PersistenceError, PersistenceResult, PersistenceState}; +pub use error::{PersistenceError, PersistenceLoadError, PersistenceResult, PersistenceState}; pub use operation::{ AcknowledgeOperation, DeleteOperation, InsertOperation, Operation, OperationId, OperationType, UpdateOperation, validate_events, @@ -57,5 +57,33 @@ pub trait PersistenceEngine, ) -> impl Future> + Send; + /// Installs the generated table schema and rejects a non-empty schema that + /// belongs to a different table shape. + /// Custom engines may keep the default no-op when they do not expose + /// WorkTable data files. + fn ensure_schema( + &mut self, + _row_schema: Vec<(String, String)>, + _primary_key_fields: Vec, + _secondary_index_types: Vec<(String, String)>, + ) -> impl Future> + Send { + async { Ok(()) } + } + + /// Validates the generated schema while loading an existing store. + /// + /// Disk engines leave legacy stores with empty schema metadata unchanged, + /// so opening an old database does not mutate it as a side effect. A newly + /// created store may install its schema from this method. Custom engines + /// may keep the default no-op. + fn validate_schema( + &mut self, + _row_schema: Vec<(String, String)>, + _primary_key_fields: Vec, + _secondary_index_types: Vec<(String, String)>, + ) -> impl Future> + Send { + async { Ok(()) } + } + fn config(&self) -> &Self::Config; } diff --git a/src/persistence/space/data.rs b/src/persistence/space/data.rs index 55179727..1fbffeaa 100644 --- a/src/persistence/space/data.rs +++ b/src/persistence/space/data.rs @@ -1,4 +1,3 @@ -use std::future::Future; use std::io::SeekFrom; use std::path::Path; @@ -187,7 +186,12 @@ where &mut self.info } - fn save_info(&mut self) -> impl Future> + Send { - persist_page(&mut self.info, &mut self.data_file) + async fn save_info(&mut self) -> eyre::Result<()> { + persist_page(&mut self.info, &mut self.data_file).await?; + // A generated table may immediately reopen this file through a + // separate handle. Make the updated metadata visible before reporting + // success, just as `save_data` does for row bytes. + self.data_file.flush().await?; + Ok(()) } } diff --git a/src/persistence/task.rs b/src/persistence/task.rs index 30a35f05..578fe75f 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -603,6 +603,7 @@ pub struct PersistenceTask, analyzer_in_progress: Arc, lifecycle: Arc, + table_path: String, phantom_data: PhantomData, } @@ -662,6 +663,21 @@ impl self.lifecycle.state() } + /// Returns the current physical size of the table data file. + /// + /// This is intentionally separate from `VacuumStats`: vacuum currently + /// compacts and reuses in-memory pages but does not truncate `.wt.data`. + /// Operators can sample this value to observe physical growth. + pub async fn persisted_data_file_size_bytes(&self) -> std::io::Result { + tokio::fs::metadata(format!( + "{}/{}", + self.table_path.trim_end_matches('/'), + WT_DATA_EXTENSION + )) + .await + .map(|metadata| metadata.len()) + } + /// Returns a sink that lets vacuum queue persistence operations for row /// moves into this task's operation queue. pub fn vacuum_sink(&self) -> Arc> @@ -681,6 +697,7 @@ impl PrimaryKey: Clone + Debug + Send + Sync + 'static, AvailableIndexes: Copy + Clone + Debug + Hash + Eq + Send + Sync + 'static, { + let table_path = engine.config().table_path().to_owned(); let lifecycle = Arc::new(PersistenceLifecycle::new()); let queue = Arc::new(Queue::new(lifecycle.clone())); @@ -744,6 +761,7 @@ impl analyzer_inner_wt, analyzer_in_progress, lifecycle, + table_path, phantom_data: PhantomData, } } diff --git a/src/table/mod.rs b/src/table/mod.rs index d01b1604..3af0b187 100644 --- a/src/table/mod.rs +++ b/src/table/mod.rs @@ -3,13 +3,13 @@ pub mod system_info; pub mod vacuum; use crate::in_memory::{ArchivedRowWrapper, DataPages, RowWrapper, StorableRow}; -use crate::persistence::{AcknowledgeOperation, InsertOperation, Operation}; +use crate::persistence::{AcknowledgeOperation, InsertOperation, Operation, PersistenceLoadError}; use crate::prelude::{Link, LockMap, OperationId, PrimaryKeyGeneratorState}; use crate::primary_key::{PrimaryKeyGenerator, TablePrimaryKey}; use crate::util::OffsetEqLink; use crate::{ AvailableIndex, IndexError, IndexMap, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, - TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, convert_change_events, in_memory, + TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, UniqueIndex, convert_change_events, in_memory, }; use data_bucket::INNER_PAGE_SIZE; use derive_more::{Display, Error, From}; @@ -22,8 +22,10 @@ use rkyv::ser::allocator::ArenaHandle; use rkyv::ser::sharing::Share; use rkyv::util::AlignedVec; use rkyv::{Archive, Deserialize, Portable, Serialize}; +use std::collections::HashSet; use std::fmt::Debug; use std::marker::PhantomData; +use std::path::Path; use std::sync::Arc; use uuid::Uuid; @@ -123,6 +125,70 @@ where Row: StorableRow + Send + Clone + 'static, ::WrappedRow: RowWrapper, { + /// Audits the persisted primary-index/data boundary before a table is made + /// available to callers. + /// + /// This load-only scan prevents a torn index link from turning zeroed or + /// unrelated bytes into a plausible row. It deliberately does not run on + /// steady-state operations. + pub fn validate_persisted_state(&self, path: impl AsRef) -> Result<(), PersistenceLoadError> + where + <::WrappedRow as Archive>::Archived: Portable + + Deserialize<::WrappedRow, HighDeserializer> + + for<'a> rkyv::bytecheck::CheckBytes>, + { + let path = path.as_ref(); + let mut links = HashSet::with_capacity(self.primary_index.pk_map.len()); + + for (primary_key, offset_link) in self.primary_index.pk_map.iter_values() { + if !links.insert(offset_link) { + return Err(PersistenceLoadError::corrupt( + path, + format!("multiple primary keys reference physical link {:?}", offset_link.0), + )); + } + + let row = self.data.select_non_ghosted_checked(offset_link.0).map_err(|error| { + PersistenceLoadError::corrupt( + path, + format!("primary key {primary_key:?} references an invalid row: {error}"), + ) + })?; + if row.get_primary_key() != primary_key { + return Err(PersistenceLoadError::corrupt( + path, + format!("row at {:?} does not match primary key {primary_key:?}", offset_link.0), + )); + } + + let reverse_key = self.primary_index.reverse_pk_map.get_value(&offset_link); + if reverse_key.as_ref() != Some(&primary_key) { + return Err(PersistenceLoadError::corrupt( + path, + format!("reverse primary index does not match link {:?}", offset_link.0), + )); + } + } + + if self.primary_index.reverse_pk_map.len() != links.len() { + return Err(PersistenceLoadError::corrupt( + path, + "forward and reverse primary indexes contain different numbers of entries", + )); + } + + for (offset_link, primary_key) in self.primary_index.reverse_pk_map.iter_values() { + if self.primary_index.pk_map.get_value(&primary_key) != Some(offset_link) { + return Err(PersistenceLoadError::corrupt( + path, + format!("forward primary index does not match link {:?}", offset_link.0), + )); + } + } + + Ok(()) + } + pub fn get_next_pk(&self) -> PrimaryKey where PkGen: PrimaryKeyGenerator, diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index 4343b458..a2fced84 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -170,6 +170,13 @@ where // while a pre-existing reader is still active. self.data_pages.allocate_new_or_pop_free().id }; + // `page_from` is `mark_page_empty`'d after this loop; it must + // never also be the destination, or the post-loop reclamation + // would drop the rows just moved into it. (Review finding F2.) + debug_assert_ne!( + page_from, page_to, + "vacuum destination must differ from the source being reclaimed" + ); match self.move_data_from(page_from, page_to).await? { (true, true) => { // from moved fully and on to no more space diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index aa15496c..4425022b 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -9,6 +9,7 @@ mod failure; mod index_page; mod loaded_index_growth; mod read; +mod schema; mod space_index; mod sync; mod toc; @@ -55,24 +56,3 @@ worktable!( number: u64, } ); - -pub async fn get_empty_test_wt() -> TestPersistWorkTable { - let config = DiskConfig::new_with_table_name( - "tests/data", - TestPersistWorkTable::name_snake_case(), - TestPersistWorkTable::version(), - ); - let engine = TestPersistPersistenceEngine::new(config).await.unwrap(); - TestPersistWorkTable::new(engine).await.unwrap() -} - -pub async fn get_test_wt() -> TestPersistWorkTable { - let table = get_empty_test_wt().await; - - for i in 1..100 { - let row = TestPersistRow { another: i, id: i }; - table.insert(row).unwrap(); - } - - table -} diff --git a/tests/persistence/read.rs b/tests/persistence/read.rs index d98fc752..d56ec83c 100644 --- a/tests/persistence/read.rs +++ b/tests/persistence/read.rs @@ -4,8 +4,7 @@ use worktable::prelude::*; // TODO: Fix naming. use crate::persistence::{ - TEST_PERSIST_INNER_SIZE, TEST_PERSIST_PAGE_SIZE, TestPersistPersistenceEngine, TestPersistWorkTable, - get_empty_test_wt, get_test_wt, + TEST_PERSIST_INNER_SIZE, TEST_PERSIST_PAGE_SIZE, TestPersistPersistenceEngine, TestPersistRow, TestPersistWorkTable, }; use crate::remove_dir_if_exists; @@ -124,12 +123,11 @@ async fn test_space_parse() { ); let engine = TestPersistPersistenceEngine::new(config).await.unwrap(); let table = TestPersistWorkTable::load(engine).await.unwrap(); - let expected = get_test_wt().await; + let expected = (1..100) + .map(|id| TestPersistRow { id, another: id }) + .collect::>(); - assert_eq!( - table.select_all().execute().unwrap(), - expected.select_all().execute().unwrap() - ); + assert_eq!(table.select_all().execute().unwrap(), expected); } #[tokio::test] @@ -143,9 +141,5 @@ async fn test_space_parse_no_file() { ); let engine = TestPersistPersistenceEngine::new(config).await.unwrap(); let table = TestPersistWorkTable::load(engine).await.unwrap(); - let expected = get_empty_test_wt().await; - assert_eq!( - table.select_all().execute().unwrap(), - expected.select_all().execute().unwrap() - ); + assert!(table.select_all().execute().unwrap().is_empty()); } diff --git a/tests/persistence/schema.rs b/tests/persistence/schema.rs new file mode 100644 index 00000000..c815a18a --- /dev/null +++ b/tests/persistence/schema.rs @@ -0,0 +1,90 @@ +use tokio::fs::File; + +use super::*; +use crate::remove_dir_if_exists; + +worktable!( + name: SchemaMetadata, + persist: true, + columns: { + id: u64 primary_key autoincrement, + email: String, + score: i64, + }, + indexes: { + email_idx: email unique, + score_idx: score, + }, +); + +worktable!( + name: IncompatibleSchema, + persist: true, + columns: { + id: u64 primary_key autoincrement, + display_name: String, + }, +); + +#[tokio::test] +async fn generated_schema_is_persisted_and_mismatches_are_rejected() { + let root = "tests/data/persisted_schema_metadata"; + remove_dir_if_exists(root.to_owned()).await; + let table_path = format!("{root}/shared"); + let config = DiskConfig::new(root, &table_path, SchemaMetadataWorkTable::version()); + + let engine = SchemaMetadataPersistenceEngine::new(config.clone()).await.unwrap(); + let table = SchemaMetadataWorkTable::load(engine).await.unwrap(); + table.close().await.unwrap(); + + let mut file = File::open(format!("{table_path}/{}", WT_DATA_EXTENSION)).await.unwrap(); + let info = parse_page::, { PAGE_SIZE as u32 }>(&mut file, 0) + .await + .unwrap(); + assert_eq!( + info.inner.row_schema, + vec![ + ("id".to_owned(), "u64".to_owned()), + ("email".to_owned(), "String".to_owned()), + ("score".to_owned(), "i64".to_owned()), + ] + ); + assert_eq!(info.inner.primary_key_fields, vec!["id"]); + assert_eq!( + info.inner.secondary_index_types, + vec![ + ("email_idx".to_owned(), "String".to_owned()), + ("score_idx".to_owned(), "i64".to_owned()), + ] + ); + + let incompatible = DiskConfig::new(root, &table_path, IncompatibleSchemaWorkTable::version()); + let engine = IncompatibleSchemaPersistenceEngine::new(incompatible).await.unwrap(); + let error = IncompatibleSchemaWorkTable::load(engine).await.unwrap_err(); + assert!(error.to_string().contains("persisted schema mismatch")); +} + +#[tokio::test] +async fn loading_a_legacy_empty_schema_does_not_rewrite_the_file() { + let root = "tests/data/persisted_schema_legacy"; + remove_dir_if_exists(root.to_owned()).await; + let table_path = format!("{root}/shared"); + let config = DiskConfig::new(root, &table_path, SchemaMetadataWorkTable::version()); + + // Opening the raw engine bootstraps the same empty metadata written by + // pre-schema WorkTable releases. Dropping it before constructing a table + // makes the next engine observe an existing legacy file. + drop(SchemaMetadataPersistenceEngine::new(config.clone()).await.unwrap()); + + let engine = SchemaMetadataPersistenceEngine::new(config).await.unwrap(); + let table = SchemaMetadataWorkTable::load(engine).await.unwrap(); + table.close().await.unwrap(); + + let mut file = File::open(format!("{table_path}/{}", WT_DATA_EXTENSION)).await.unwrap(); + let info = parse_page::, { PAGE_SIZE as u32 }>(&mut file, 0) + .await + .unwrap(); + assert!(info.inner.row_schema.is_empty()); + assert!(info.inner.primary_key_fields.is_empty()); + assert!(info.inner.secondary_index_types.is_empty()); +} diff --git a/tests/persistence/torn_shutdown.rs b/tests/persistence/torn_shutdown.rs index 1c20cebe..f206bdb4 100644 --- a/tests/persistence/torn_shutdown.rs +++ b/tests/persistence/torn_shutdown.rs @@ -44,6 +44,16 @@ worktable!( const DIR: &str = "tests/data/torn_shutdown/persisted"; pub const WRITER_ENV: &str = "WT_TORN_SHUTDOWN_WRITER"; +static TORN_STORE_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +fn lock_torn_store_tests() -> std::sync::MutexGuard<'static, ()> { + // These process-level tests intentionally mutate the same persisted store. + // Cargo runs tests concurrently, so serialize the parent processes while + // still allowing each test's writer child to operate on that store. + TORN_STORE_TEST_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} fn key(i: u64) -> String { format!("msg-00000000-0000-4000-8000-{i:012}") @@ -57,14 +67,18 @@ fn row(i: u64) -> TornShutdownRow { } } -async fn open_table() -> TornShutdownWorkTable { +async fn try_open_table() -> eyre::Result { let config = DiskConfig::new_with_table_name( DIR, TornShutdownWorkTable::name_snake_case(), TornShutdownWorkTable::version(), ); - let engine = TornShutdownPersistenceEngine::new(config).await.unwrap(); - TornShutdownWorkTable::load(engine).await.unwrap() + let engine = TornShutdownPersistenceEngine::new(config).await?; + TornShutdownWorkTable::load(engine).await +} + +async fn open_table() -> TornShutdownWorkTable { + try_open_table().await.unwrap() } /// The writer half, run as a child process: appends rows forever without @@ -163,6 +177,11 @@ fn tear_the_store_repeatedly() { "writer round {round} was killed by a signal ({status}): the \ tear was read as data instead of refused. Child stderr:\n{stderr}" ); + assert!( + stderr.contains("torn or corrupt"), + "writer round {round} refused the store without the typed corruption message. \ + Child stderr:\n{stderr}" + ); continue; } }; @@ -176,10 +195,11 @@ fn tear_the_store_repeatedly() { /// The bar validated page reads meet TODAY: a store torn by mid-write kills /// never takes a process down with a signal. Every load either succeeds or /// refuses naming corruption, in the writer children and in this process. -/// What this bar does NOT include is row fidelity — see the ignored full-bar -/// test below for that. +/// What this bar does NOT include is row fidelity — the full Option B load +/// contract is exercised by the next test. #[test] fn test_torn_store_fails_clean_never_by_signal() { + let _test_guard = lock_torn_store_tests(); tear_the_store_repeatedly(); let outcome = std::panic::catch_unwind(|| { @@ -204,71 +224,145 @@ fn test_torn_store_fails_clean_never_by_signal() { drop(outcome); } -/// The boundary of the design, written down as a test. Persistence here is -/// best-effort by contract: consumers drain on every catchable exit, the -/// accepted loss window is the instant between in-memory and on-disk, and a -/// SIGKILL mid-write may cost data, with an index rebuild (worktable's -/// rebuild verbs, or a snapshot restore) as the recovery. This test states -/// what full crash-consistency WOULD look like: a killed store scans as a -/// consistent prefix, no phantom rows. Validated reads alone cannot meet it, -/// because a dangling index link into a zeroed region reads as a row of -/// empty fields that validates perfectly. It stays ignored as documentation -/// of the accepted risk, not as a demand: run it with -/// `cargo test -- --ignored test_store_survives_torn_shutdowns` if the -/// design contract ever changes. +/// Option B's load boundary: persistence is best-effort, so a SIGKILL may +/// lose acknowledged rows. The next load must nevertheless do exactly one of +/// two things: return a validated state containing no phantom rows, or return +/// the typed `PersistenceLoadError` that directs the caller to restore or +/// rebuild. A torn store must never become a live table with invented data. #[test] -#[ignore = "documents the accepted design boundary: SIGKILL mid-write may cost data; recovery is rebuild"] fn test_store_survives_torn_shutdowns() { + let _test_guard = lock_torn_store_tests(); tear_the_store_repeatedly(); - // The reckoning: load and scan the torn store IN THIS PROCESS, through - // an unwind boundary so a named corruption refusal counts as the fix - // working. What must not happen is the process dying of SIGBUS/UB (the - // harness reports that as the test binary dying), or the scan returning - // rows nobody wrote. - let outcome = std::panic::catch_unwind(|| { - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .enable_io() - .enable_time() - .build() - .unwrap(); - runtime.block_on(async { - let table = open_table().await; - let rows = table.select_all().execute().unwrap(); - let legal_projects: BTreeSet = (0..3).map(|p| format!("proj-{p:02}")).collect(); - for row in &rows { - assert!( - row.id.starts_with("msg-00000000-0000-4000-8000-"), - "scan returned a row no writer ever inserted (id {:?}): torn bytes \ + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + runtime.block_on(async { + match try_open_table().await { + Ok(table) => { + let rows = table.select_all().execute().unwrap(); + let legal_projects: BTreeSet = (0..3).map(|p| format!("proj-{p:02}")).collect(); + for row in &rows { + assert!( + row.id.starts_with("msg-00000000-0000-4000-8000-"), + "scan returned a row no writer ever inserted (id {:?}): torn bytes \ were read as data", - &row.id[..row.id.len().min(60)] - ); - assert!( - legal_projects.contains(&row.project_id), - "row {} carries project {:?}, which no writer ever wrote", - row.id, - row.project_id - ); + &row.id[..row.id.len().min(60)] + ); + assert!( + legal_projects.contains(&row.project_id), + "row {} carries project {:?}, which no writer ever wrote", + row.id, + row.project_id + ); + } + // And the survivor must still accept writes and a drain. + table.insert(row(9_000_000)).unwrap(); + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled appending to the survivor store") + .expect("persistence engine failed"); } - // And the survivor must still accept writes and a drain. - table.insert(row(9_000_000)).unwrap(); - timeout(Duration::from_secs(30), table.wait_for_ops()) - .await - .expect("persistence stalled appending to the survivor store") - .expect("persistence engine failed"); - }); + Err(error) => assert!( + error.downcast_ref::().is_some(), + "torn store was refused with an untyped error: {error:#}" + ), + } }); - if let Err(panic) = outcome { - let message = panic - .downcast_ref::() - .map(String::as_str) - .or_else(|| panic.downcast_ref::<&str>().copied()) - .unwrap_or("(non-string panic)"); +} + +#[test] +fn corrupted_row_is_refused_with_typed_load_error() { + let _test_guard = lock_torn_store_tests(); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + + let link = runtime.block_on(async { + remove_dir_if_exists(DIR.to_string()).await; + let table = open_table().await; + let primary_key = table.insert(row(7)).unwrap(); + let link = table.0.primary_index.pk_map.get_value(&primary_key).unwrap().0; + table.close().await.unwrap(); + link + }); + + let data_path = format!( + "{DIR}/{}/{}", + TornShutdownWorkTable::name_snake_case(), + WT_DATA_EXTENSION + ); + let page_id: u32 = link.page_id.into(); + let byte_offset = u64::from(page_id) * PAGE_SIZE as u64 + GENERAL_HEADER_SIZE as u64 + u64::from(link.offset); + { + use std::io::{Seek, SeekFrom, Write}; + + let mut data_file = std::fs::OpenOptions::new().write(true).open(data_path).unwrap(); + data_file.seek(SeekFrom::Start(byte_offset)).unwrap(); + data_file.write_all(&vec![0; link.length as usize]).unwrap(); + data_file.sync_all().unwrap(); + } + + runtime.block_on(async { + let error = match try_open_table().await { + Ok(_) => panic!("corrupted row was exposed as a live table"), + Err(error) => error, + }; + let typed = error + .downcast_ref::() + .expect("corrupt persisted row must return PersistenceLoadError"); + assert_eq!(typed.path(), std::path::Path::new(&format!("{DIR}/torn_shutdown"))); + assert!(!typed.reason().is_empty()); + }); +} + +#[test] +fn incomplete_secondary_index_is_refused_with_typed_load_error() { + let _test_guard = lock_torn_store_tests(); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(DIR.to_string()).await; + let table = open_table().await; + table.insert(row(11)).unwrap(); + table.close().await.unwrap(); + }); + + let secondary_path = format!( + "{DIR}/{}/project_idx{}", + TornShutdownWorkTable::name_snake_case(), + WT_INDEX_EXTENSION + ); + { + let secondary_file = std::fs::OpenOptions::new().write(true).open(secondary_path).unwrap(); + secondary_file.set_len(PAGE_SIZE as u64).unwrap(); + secondary_file.sync_all().unwrap(); + } + + runtime.block_on(async { + let error = match try_open_table().await { + Ok(_) => panic!("incomplete secondary index was exposed as a live table"), + Err(error) => error, + }; + let typed = error + .downcast_ref::() + .expect("incomplete secondary index must return PersistenceLoadError"); assert!( - message.contains("torn or corrupt"), - "the torn store failed without naming corruption: {message}" + typed.reason().contains("project_idx"), + "unexpected reason: {}", + typed.reason() ); - } + }); } /// The clean-shutdown sibling: many short load-append-drain-close sessions, diff --git a/tests/persistence/vacuum.rs b/tests/persistence/vacuum.rs index 5623c8e9..9ea77bcd 100644 --- a/tests/persistence/vacuum.rs +++ b/tests/persistence/vacuum.rs @@ -76,10 +76,20 @@ fn test_vacuum_on_persisted_table_survives_reload() { .await .expect("persistence should catch up before vacuum") .expect("persistence engine failed"); + let physical_bytes_before = table.persisted_data_file_size_bytes().await.unwrap(); let vacuum = table.vacuum(); let stats = vacuum.vacuum().await.unwrap(); assert!(stats.pages_freed > 0, "vacuum should have moved rows off a page"); + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence should catch up after vacuum") + .expect("persistence engine failed"); + let physical_bytes_after = table.persisted_data_file_size_bytes().await.unwrap(); + assert!( + physical_bytes_after >= physical_bytes_before, + "persisted vacuum is logical compaction and must not report implicit file truncation" + ); // Insert after vacuum: these operations carry event ids issued // after the moves, so if vacuum consumed ids without queueing the diff --git a/tests/worktable/borrowed_primary_key.rs b/tests/worktable/borrowed_primary_key.rs new file mode 100644 index 00000000..36acd65b --- /dev/null +++ b/tests/worktable/borrowed_primary_key.rs @@ -0,0 +1,81 @@ +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: BorrowedStringKey, + columns: { + id: String primary_key, + value: u64, + }, + queries: { + update: { + BorrowedValueById(value) by id, + } + in_place: { + BorrowedValueById(value) by id, + } + } +); + +worktable!( + name: BorrowedTupleKey, + columns: { + tenant: String primary_key, + record: String primary_key, + value: u64, + }, +); + +#[tokio::test] +async fn string_primary_key_accepts_borrowed_forms() { + let table = BorrowedStringKeyWorkTable::default(); + let id = "tenant".to_owned(); + let row = BorrowedStringKeyRow { + id: id.clone(), + value: 7, + }; + table.insert(row.clone()).unwrap(); + + assert_eq!(table.select(&id), Some(row.clone())); + assert_eq!(table.select(id.as_str()), Some(row.clone())); + + let generated = BorrowedStringKeyPrimaryKey::from(&id); + assert_eq!(table.select(&generated), Some(row)); + + table + .update_borrowed_value_by_id(BorrowedValueByIdQuery { value: 8 }, &id) + .await + .unwrap(); + table + .update_borrowed_value_by_id_in_place(|value| *value += 1, &id) + .await + .unwrap(); + assert_eq!(table.select(&id).unwrap().value, 9); + assert_eq!( + table + .select_by_pk_range(id.as_str()..=id.as_str()) + .execute() + .unwrap() + .len(), + 1 + ); + + table.delete(&id).await.unwrap(); + assert!(table.select(&id).is_none()); +} + +#[tokio::test] +async fn tuple_primary_key_accepts_a_borrowed_tuple() { + let table = BorrowedTupleKeyWorkTable::default(); + let key = ("tenant".to_owned(), "record".to_owned()); + let row = BorrowedTupleKeyRow { + tenant: key.0.clone(), + record: key.1.clone(), + value: 11, + }; + table.insert(row.clone()).unwrap(); + + assert_eq!(table.select(&key), Some(row)); + table.delete(&key).await.unwrap(); + assert!(table.select(&key).is_none()); +} diff --git a/tests/worktable/custom_pk.rs b/tests/worktable/custom_pk.rs index 66fa5fe4..1d66e2a4 100644 --- a/tests/worktable/custom_pk.rs +++ b/tests/worktable/custom_pk.rs @@ -62,3 +62,18 @@ fn test_custom_pk() { let pk = table.get_next_pk(); assert_eq!(pk, CustomId::from(0).into()); } + +#[tokio::test] +async fn borrowed_custom_primary_key_is_accepted() { + let table = TestWorkTable::default(); + let id = CustomId(42); + let row = TestRow { + id: id.clone(), + test: 7, + }; + table.insert(row.clone()).unwrap(); + + assert_eq!(table.select(&id), Some(row)); + table.delete(&id).await.unwrap(); + assert!(table.select(&id).is_none()); +} diff --git a/tests/worktable/leak_probe.rs b/tests/worktable/leak_probe.rs new file mode 100644 index 00000000..027b5f69 --- /dev/null +++ b/tests/worktable/leak_probe.rs @@ -0,0 +1,76 @@ +//! Leak probe (review follow-up): a long update-churn loop must not grow +//! storage without bound. Reinsert-per-update leaves dead slots / retired +//! publications; if reclamation never catches up, an update-heavy workload +//! balloons memory (the suspected cause of the hung, ballooning test process). +//! +//! Asserts on the table's own storage accounting (row/page counts) rather than +//! process RSS, so it is deterministic and cannot itself hang the harness. + +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: LeakProbe, + columns: { + id: u64 primary_key, + payload: String, + }, + queries: { + update: { + Payload(payload) by id, + } + } +); + +#[tokio::test] +async fn update_churn_does_not_grow_storage_unbounded() { + let table = LeakProbeWorkTable::default(); + table + .insert(LeakProbeRow { + id: 1, + payload: "0000".to_string(), + }) + .unwrap(); + + // One row, many same-length updates. Logical cardinality stays 1 the whole + // time; only physical slot churn happens. + let pages_after_warmup = { + for i in 0..100u64 { + table + .update_payload( + PayloadQuery { + payload: format!("{:04}", i % 10000), + }, + 1, + ) + .await + .unwrap(); + } + table.0.data.get_bytes().len() + }; + + for i in 0..5_000u64 { + table + .update_payload( + PayloadQuery { + payload: format!("{:04}", i % 10000), + }, + 1, + ) + .await + .unwrap(); + } + + let pages_after_churn = table.0.data.get_bytes().len(); + + // Logical row count is unchanged. + assert_eq!(table.count(), 1, "cardinality drifted under update churn"); + + // Physical page count must not grow ~linearly with the number of updates. + // Allow generous slack, but 5000 updates growing pages by thousands is a + // reclamation leak, not normal slack. + assert!( + pages_after_churn <= pages_after_warmup + 8, + "update churn leaked pages: {pages_after_warmup} -> {pages_after_churn} over 5000 same-key updates" + ); +} diff --git a/tests/worktable/mod.rs b/tests/worktable/mod.rs index ac946814..5a111ca3 100644 --- a/tests/worktable/mod.rs +++ b/tests/worktable/mod.rs @@ -1,6 +1,7 @@ mod array; mod base; mod bench; +mod borrowed_primary_key; mod config; mod count; mod custom_pk; @@ -9,7 +10,9 @@ mod float; mod in_place; mod index; mod index_backends; +mod leak_probe; mod lock_order; +mod mutation_gate_deadlock; mod nid; mod option; mod tuple_primary_key; @@ -18,4 +21,5 @@ mod update_in_place_unsized; mod upsert; mod uuid; mod vacuum; +mod vacuum_no_row_loss; mod with_enum; diff --git a/tests/worktable/mutation_gate_deadlock.rs b/tests/worktable/mutation_gate_deadlock.rs new file mode 100644 index 00000000..97c1209c --- /dev/null +++ b/tests/worktable/mutation_gate_deadlock.rs @@ -0,0 +1,144 @@ +//! Guard test for the synchronous mutation gate (`LockMap::mutation_guard`). +//! +//! The gate is a blocking spin/yield ticket lock, and the generated +//! `update`/`in_place`/`delete` paths hold the resulting `MutationGuard` inside +//! the `LockGuard` **across `.await`** (e.g. `update_with_guard(...).await`, +//! `reinsert(...).await`). A review flagged this as a possible livelock/deadlock +//! when two keys collide on the same 1-of-64 stripe on a constrained runtime. +//! +//! These tests exercise exactly that scenario (colliding keys, single-worker and +//! 2-worker runtimes) and currently PASS: because tokio schedules async tasks +//! cooperatively and the spinner falls back to `thread::yield_now()`, the parked +//! guard-holder is still polled to completion on the same thread, so forward +//! progress holds. They are kept as a standing guard so a future change to the +//! gate (or to spinning under a blocking holder) that DOES introduce the hazard +//! is caught. Every case is wrapped in `tokio::time::timeout`, so if the hazard +//! ever appears it surfaces as a FAILED assertion — never a harness-hanging, +//! memory-ballooning process. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::time::timeout; +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: GateBench, + columns: { + id: u64 primary_key, + val: u64, + }, + queries: { + update: { + Val(val) by id, + } + } +); + +/// Find two distinct keys that land on the same mutation stripe. The stripe is +/// `DefaultHasher(key) % 64`; brute-force a colliding pair so the test does not +/// depend on gate internals beyond the documented stripe count. +fn colliding_keys() -> (u64, u64) { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + const STRIPES: u64 = 64; + let stripe = |k: u64| { + let mut h = DefaultHasher::new(); + GateBenchPrimaryKey::from(k).hash(&mut h); + h.finish() % STRIPES + }; + let first = 1u64; + let target = stripe(first); + for k in 2..100_000u64 { + if stripe(k) == target { + return (first, k); + } + } + panic!("no colliding key pair found"); +} + +/// Two same-stripe keys updated concurrently must both make progress. On a +/// single-worker runtime, a gate held across `.await` cannot: the parked holder +/// has no thread to resume on while the other task spins. +#[test] +fn concurrent_same_stripe_updates_do_not_deadlock() { + let (a, b) = colliding_keys(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async move { + let table = Arc::new(GateBenchWorkTable::default()); + table.insert(GateBenchRow { id: a, val: 0 }).unwrap(); + table.insert(GateBenchRow { id: b, val: 0 }).unwrap(); + + let ta = { + let table = table.clone(); + tokio::spawn(async move { + for i in 0..500u64 { + table.update_val(ValQuery { val: i }, a).await.unwrap(); + } + }) + }; + let tb = { + let table = table.clone(); + tokio::spawn(async move { + for i in 0..500u64 { + table.update_val(ValQuery { val: i }, b).await.unwrap(); + } + }) + }; + + let joined = async { + ta.await.unwrap(); + tb.await.unwrap(); + }; + timeout(Duration::from_secs(20), joined) + .await + .expect("same-stripe concurrent updates deadlocked (gate held across .await)"); + + assert_eq!(table.select(a).unwrap().val, 499); + assert_eq!(table.select(b).unwrap().val, 499); + }); +} + +/// Same hazard on a small multi-worker pool: enough same-stripe tasks await +/// while holding the gate that every worker ends up spinning on `serving`. +#[test] +fn many_same_stripe_updates_do_not_starve_worker_pool() { + let (a, b) = colliding_keys(); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async move { + let table = Arc::new(GateBenchWorkTable::default()); + for k in [a, b] { + table.insert(GateBenchRow { id: k, val: 0 }).unwrap(); + } + + let mut handles = Vec::new(); + for worker in 0..8u64 { + let table = table.clone(); + let key = if worker % 2 == 0 { a } else { b }; + handles.push(tokio::spawn(async move { + for i in 0..300u64 { + table.update_val(ValQuery { val: i }, key).await.unwrap(); + } + })); + } + + let joined = async { + for h in handles { + h.await.unwrap(); + } + }; + timeout(Duration::from_secs(30), joined) + .await + .expect("same-stripe pool starved (gate spin held across .await)"); + }); +} diff --git a/tests/worktable/update_in_place_unsized.rs b/tests/worktable/update_in_place_unsized.rs index 453f287b..d7e51136 100644 --- a/tests/worktable/update_in_place_unsized.rs +++ b/tests/worktable/update_in_place_unsized.rs @@ -17,15 +17,21 @@ //! field changed size. //! //! ## Why the obvious fix is not enough (do not just flip the initializer) -//! Setting the initializer to `false` correctly lets same-length updates skip -//! reinsert — but the in-place archived write in this custom-update path then -//! CORRUPTS variable-length rows in existing tests -//! (`worktable::unsized_::update_parallel_more_strings`, `update_many_times`, -//! `in_place::test_update_in_place_and_update_unsized_multithread`): reads come -//! back as raw archived bytes. So a real fix must make the in-place write of an -//! (even equal-length) archived `String` field safe in this path, not merely -//! change when the fast path is taken. That is a storage/codegen change beyond a -//! one-liner; tracked here so the fix has a proof. +//! Setting the initializer to `false` lets same-length updates skip reinsert — +//! but the in-place write then CORRUPTS long strings. The generated field write +//! is `mem::swap(&mut archived.inner., &mut archived_row.)`. +//! `ArchivedString` is a union: short strings (<= rkyv INLINE_CAPACITY) are +//! inline, so the swap is self-contained; LONG strings are out-of-line — a +//! relative pointer + length whose characters live in `archived_row`'s buffer. +//! Swapping only the pointer into the slot leaves it pointing at bytes that were +//! never written to the slot → reads come back as raw archived bytes (see +//! `worktable::unsized_::update_parallel_more_strings`, `update_many_times`, +//! `in_place::test_update_in_place_and_update_unsized_multithread`). +//! +//! A real fix must overwrite the existing out-of-line byte region in place +//! (e.g. `ArchivedStringRepr::as_bytes_seal`) when the new value fits, reinserting +//! only when it doesn't — preserving field-level semantics. Unsafe archived-memory +//! work; a subtle error is silent corruption. See docs/pr46-review-findings.md (F4). //! //! The observable is the row's physical `Link`: an in-place update keeps it, a //! reinsert changes it. Remove `#[ignore]` when the in-place path is fixed. diff --git a/tests/worktable/vacuum_no_row_loss.rs b/tests/worktable/vacuum_no_row_loss.rs new file mode 100644 index 00000000..98159338 --- /dev/null +++ b/tests/worktable/vacuum_no_row_loss.rs @@ -0,0 +1,114 @@ +//! Regression (review finding F2): vacuum must never lose a surviving row. +//! +//! In the compaction loop, `page_from` is `mark_page_empty`'d unconditionally +//! after its rows are moved, with no `page_from != page_to` guard. If the +//! destination search falls through to `allocate_new_or_pop_free()` and hands +//! back a page that ends up holding moved-in rows (e.g. the source page itself, +//! or a temp page later reclaimed), marking that page empty can drop live rows +//! once the read grace period ends. +//! +//! This test forces heavy cross-page compaction (large rows, half deleted from +//! many pages) while a concurrent reader repeatedly resolves rows, then audits +//! that EVERY surviving row — by primary key AND by secondary index — is still +//! present and correct after vacuum quiesces. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use worktable::prelude::*; +use worktable::vacuum::{VacuumManager, VacuumManagerConfig}; +use worktable_codegen::worktable; + +worktable!( + name: VacuumLoss, + columns: { + id: u64 primary_key autoincrement, + value: i64, + data: String + }, + indexes: { + value_idx: value unique, + data_idx: data, + } +); + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn vacuum_never_loses_surviving_rows() { + let config = VacuumManagerConfig { + check_interval: Duration::from_millis(5), + ..Default::default() + }; + let vacuum_manager = Arc::new(VacuumManager::with_config(config)); + let table = Arc::new(VacuumLossWorkTable::default()); + + // Large rows so only a couple fit per page -> deleting half fragments many + // pages and forces the compaction loop to move rows across pages. + let mut all: HashMap = HashMap::new(); + for i in 0..400i64 { + let row = VacuumLossRow { + id: table.get_next_pk().into(), + value: i, + data: format!("{i:04}-{}", "d".repeat(4_000)), + }; + table.insert(row.clone()).unwrap(); + all.insert(row.id, row); + } + + // Delete every other row to leave fragmented source pages. + let mut survivors: HashMap = HashMap::new(); + let mut ids: Vec = all.keys().copied().collect(); + ids.sort_unstable(); + for (n, id) in ids.iter().enumerate() { + if n % 2 == 0 { + table.delete(*id).await.unwrap(); + } else { + survivors.insert(*id, all[id].clone()); + } + } + + let vacuum = table.vacuum(); + vacuum_manager.register(vacuum); + let handle = vacuum_manager.run_vacuum_task(); + + // Concurrent reader active across the whole vacuum: repeatedly resolve + // survivors so a grace period is live while pages are being moved/retired. + let reader_table = table.clone(); + let reader_survivors: Vec = survivors.keys().copied().collect(); + let reader = tokio::spawn(async move { + for _ in 0..50 { + for id in &reader_survivors { + let _ = reader_table.select(*id); + } + tokio::task::yield_now().await; + } + }); + + reader.await.unwrap(); + // Let vacuum run a few more cycles, then stop it and let grace periods drain. + tokio::time::sleep(Duration::from_millis(200)).await; + handle.abort(); + tokio::time::sleep(Duration::from_millis(100)).await; + + // FULL AUDIT: every survivor must still be present and correct, by pk and + // by both secondary indexes. A lost row (None) is the data-loss bug. + for (id, expected) in &survivors { + let by_pk = table.select(*id); + assert_eq!( + by_pk.as_ref(), + Some(expected), + "row {id} lost or corrupted after vacuum (by primary key)" + ); + let by_value = table.select_by_value(expected.value); + assert_eq!( + by_value.as_ref(), + Some(expected), + "row {id} unreachable via unique value index after vacuum" + ); + } + + // Deleted rows must stay gone. + for id in ids.iter().enumerate().filter(|(n, _)| n % 2 == 0).map(|(_, id)| *id) { + assert_eq!(table.select(id), None, "deleted row {id} resurrected by vacuum"); + } +}