fix: concurrency correctness and release hardening

Cargo.toml

@@ -22,6 +22,8 @@ wti-hybrid-search = ["indexset/wt-slice-binary-search"]
 wti-predictable-search = ["indexset/custom-binary-search"]
 wti-std-search = ["indexset/std-binary-search"]
 wti-superslice-search = ["indexset/superslice-binary-search"]
+# Compatibility no-op: immutable row publication is mandatory for the safe
+# generated API, including `default-features = false` builds.
 versioned-row-publication = ["worktable_codegen/versioned-row-publication"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -32,14 +34,14 @@ arctic = { package = "arctic-wt", version = "0.1.4" }
 congee = { package = "congee-wt", version = "0.4.1" }
 convert_case = "0.6.0"
 crc32fast = "1.5.0"
-data_bucket = "=0.5.1"
+data_bucket = "=0.5.2"
 # data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "page_cdc_correction", version = "0.2.7" }
 # data_bucket = { path = "../DataBucket", version = "0.3.14" }
 derive_more = { version = "2.0.1", features = ["from", "error", "display", "debug", "into"] }
 eyre = "0.6.12"
 fastrand = "2.3.0"
 futures = "0.3.30"
-indexset = { package = "WorkTablesIndex", version = "=0.0.4", default-features = false, features = ["concurrent", "cdc", "multimap"] }
+indexset = { package = "WorkTablesIndex", version = "=0.0.5", default-features = false, features = ["concurrent", "cdc", "multimap"] }
 vanilla_indexset = { package = "indexset", version = "=0.15.0", features = ["concurrent", "cdc", "multimap"] }
 # indexset = { path = "../indexset", version = "0.15.0", features = ["concurrent", "cdc", "multimap"] }
 # indexset = { package = "wt-indexset", version = "=0.12.12", features = ["concurrent", "cdc", "multimap"] }

README.md

@@ -69,33 +69,32 @@ WorkTablesIndex uses its predictable branch-based node search by default in Work
 
 ## Concurrent read/write publication
 
-The default build preserves the existing lowest-latency page path and requires
-applications to exclude reads that overlap page-byte mutation. Applications
-that need generated reads to overlap updates, inserts, deletes, and vacuum can
-opt into immutable row-version publication:
-
-```toml
-[dependencies]
-worktable = { version = "=1.0.0-beta.2", features = ["versioned-row-publication"] }
-```
+Generated reads always use immutable row-version publication. This is also true
+for `default-features = false` builds: disabling a Cargo feature must not expose
+a safe API that can race deserialization against page-byte mutation. The former
+`versioned-row-publication` feature name remains accepted as a compatibility
+no-op for existing manifests.
 
 Generated point lookups use a strict backend-specific visibility contract by
-default. WorkTablesIndex 0.0.4 keeps the structural mapping pinned until its
+default. WorkTablesIndex 0.0.5 keeps the structural mapping pinned until its
 selected node is locked, making both hits and misses definitive; contended
 lookups release the structural guard before waiting and retry the mapping.
 Congee and Arctic use their native concurrent point lookups. The explicit
 vanilla `using indexset` backend remains experimental and is excluded from the
 stable concurrent-read contract because upstream IndexSet does not expose an
 equivalent validation primitive. This index-visibility contract is independent
-of the optional row publication mode above.
+of row publication.
 
-In this mode, generated reads acquire an immutable owned row version instead
+Generated reads acquire an immutable owned row version instead
 of borrowing the mutable archived page image. Writers replace a per-row version
 only after a complete page mutation, insert visibility is an atomic lifecycle
 transition after every index is installed, and deleted or relocated links are
 not reused until readers that could have captured them have drained. Page bytes
 remain the persistence image and are internally serialized; range queries are
-still non-snapshot reads. The mode intentionally trades memory, an atomic
+still non-snapshot reads. Archived-page mutations currently take one table-wide
+writer barrier, so even disjoint writes can serialize and latency-sensitive
+deployments should validate contention throughput and tail latency. The
+protocol intentionally trades memory, an atomic
 read-side grace-period counter, and publication bookkeeping for this stronger
 concurrent-read contract. See
 [`docs/versioned-row-publication.md`](docs/versioned-row-publication.md) for the

codegen/Cargo.toml

@@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable"
 
 [features]
 s3-support = []
+# Compatibility no-op retained for downstream manifests.
 versioned-row-publication = []
 
 [lib]

codegen/src/common/model/primary_key.rs

@@ -1,10 +1,10 @@
+use indexmap::IndexMap;
 use proc_macro2::{Ident, TokenStream};
-use std::collections::HashMap;
 
 #[derive(Debug, Clone)]
 pub struct PrimaryKey {
     pub ident: Ident,
-    pub values: HashMap<Ident, TokenStream>,
+    pub values: IndexMap<Ident, TokenStream>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq)]

codegen/src/generators/in_memory/primary_key.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use indexmap::IndexMap;
 
 use crate::common::model::{GeneratorType, PrimaryKey};
 use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec};
@@ -27,7 +27,7 @@ impl InMemoryGenerator {
                         .clone(),
                 )
             })
-            .collect::<HashMap<_, _>>();
+            .collect::<IndexMap<_, _>>();
 
         let def = self.gen_primary_key_type()?;
         let impl_ = self.gen_table_primary_key_impl()?;

codegen/src/generators/in_memory/queries/delete.rs

@@ -46,7 +46,7 @@ impl InMemoryGenerator {
             {
                 let pk: #pk_ident = pk.into();
                 let op_lock = { #full_row_lock };
-                let _guard = LockGuard::new(
+                let _guard = LockGuard::new_with_mutation(
                     op_lock,
                     self.0.lock_manager.clone(),
                     pk.clone(),
@@ -69,6 +69,7 @@ impl InMemoryGenerator {
             where #pk_ident: From<Pk>
             {
                 let pk: #pk_ident = pk.into();
+                let _mutation_guard = self.0.lock_manager.mutation_guard(&pk);
                 #delete_logic
                 core::result::Result::Ok(())
             }
@@ -118,7 +119,12 @@ impl InMemoryGenerator {
                         return Err(e);
                     }
                 };
-                let row = self.0.select(pk.clone()).unwrap();
+                // Low-level staged or hydrated state can publish index
+                // reachability before clearing the row's ghost bit. Ordinary
+                // insert shares this delete's per-key mutation gate, but keep
+                // this boundary defensive instead of panicking on a hidden
+                // version.
+                let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
                 #process
             }
         } else {
@@ -129,7 +135,7 @@ impl InMemoryGenerator {
                         .get_value(&pk)
                         .map(Into::into)
                         .ok_or(WorkTableError::NotFound)?;
-                let row = self.0.select(pk.clone()).unwrap();
+                let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
                 #process
             }
         }

codegen/src/generators/in_memory/queries/in_place.rs

@@ -106,7 +106,7 @@ impl InMemoryGenerator {
             {
                 let pk: #pk_type = by.into();
                 let op_lock = { #custom_lock };
-                let _guard = LockGuard::new(
+                let _guard = LockGuard::new_with_mutation(
                     op_lock,
                     self.0.lock_manager.clone(),
                     pk.clone(),

codegen/src/generators/in_memory/queries/update.rs

@@ -62,7 +62,7 @@ impl InMemoryGenerator {
                 if true {
                     drop(_guard);
                     let op_lock = { #full_row_lock };
-                    let _guard = LockGuard::new(
+                    let _guard = LockGuard::new_with_mutation(
                         op_lock,
                         self.0.lock_manager.clone(),
                         pk.clone(),
@@ -85,7 +85,7 @@ impl InMemoryGenerator {
             pub async fn update(&self, row: #row_ident) -> core::result::Result<(), WorkTableError> {
                 let pk = row.get_primary_key();
                 let op_lock = { #full_row_lock };
-                let guard = LockGuard::new(
+                let guard = LockGuard::new_with_mutation(
                     op_lock,
                     self.0.lock_manager.clone(),
                     pk.clone(),
@@ -247,7 +247,7 @@ impl InMemoryGenerator {
                 if need_to_reinsert {
                     drop(_guard);
                     let op_lock = { #full_row_lock };
-                    let _guard = LockGuard::new(
+                    let _guard = LockGuard::new_with_mutation(
                         op_lock,
                         self.0.lock_manager.clone(),
                         pk.clone(),
@@ -465,7 +465,7 @@ impl InMemoryGenerator {
             {
                 let pk: #pk_ident = pk.into();
                 let op_lock = { #custom_lock };
-                let _guard = LockGuard::new(
+                let _guard = LockGuard::new_with_mutation(
                     op_lock,
                     self.0.lock_manager.clone(),
                     pk.clone(),
@@ -549,11 +549,12 @@ impl InMemoryGenerator {
                 let mut need_to_reinsert = true;
                 #(#fields_check)*
                 if need_to_reinsert {
+                    drop(_mutation_guard);
                     let old_guard = guards.remove(&pk).expect("guard should exist for this pk");
                     drop(old_guard);
 
                     let op_lock = { #full_row_lock };
-                    let _guard = LockGuard::new(
+                    let _guard = LockGuard::new_with_mutation(
                         op_lock,
                         self.0.lock_manager.clone(),
                         pk.clone(),
@@ -638,6 +639,7 @@ impl InMemoryGenerator {
                     if self.0.data.select_non_ghosted(link)?.#by_field != by {
                         continue;
                     }
+                    let _mutation_guard = self.0.lock_manager.mutation_guard(&pk);
                     let mut bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&row)
                         .map_err(|_| WorkTableError::SerializeError)?;
 
@@ -725,7 +727,7 @@ impl InMemoryGenerator {
                 let pk = self.0.data.select_non_ghosted(link)?.get_primary_key().clone();
 
                 let op_lock = { #custom_lock };
-                let _guard = LockGuard::new(
+                let _guard = LockGuard::new_with_mutation(
                     op_lock,
                     self.0.lock_manager.clone(),
                     pk.clone(),

codegen/src/generators/in_memory/table/impls.rs

@@ -160,11 +160,11 @@ impl InMemoryGenerator {
             /// Inserts the row if its primary key is absent, updates it
             /// otherwise.
             ///
-            /// A definitely absent key takes the same optimistic lock-free
-            /// insert path as `insert`. Existing keys and insertion collisions
-            /// acquire one full-row lock across the repeated existence check
-            /// and selected mutation, so upserts, updates, and deletes on the
-            /// same key cannot invalidate that decision.
+            /// A definitely absent key takes the same optimistic synchronous
+            /// insert path as `insert`. Insert and generated locked mutations
+            /// share a per-key mutation gate; existing keys and insertion
+            /// collisions also acquire the full-row lock across the repeated
+            /// existence check and selected mutation.
             pub async fn upsert(&self, row: #row_type) -> core::result::Result<(), WorkTableError> {
                 let pk = row.get_primary_key();
                 if !self.0.primary_index.pk_map.contains_key(&pk) {
@@ -174,9 +174,14 @@ impl InMemoryGenerator {
                         core::result::Result::Err(e) => return core::result::Result::Err(e),
                     }
                 }
+                // Retries only fire when an existence flip invalidated the
+                // optimistic decision (NotFound / row-absent). The FIFO
+                // per-key mutation gate guarantees forward progress; retain a
+                // bounded scheduler backoff for repeated decision races.
+                let mut backoff_spins: u32 = 0;
                 loop {
                     let op_lock = { #full_row_lock };
-                    let guard = LockGuard::new(
+                    let guard = LockGuard::new_with_mutation(
                         op_lock,
                         self.0.lock_manager.clone(),
                         pk.clone(),
@@ -185,11 +190,15 @@ impl InMemoryGenerator {
                     let result = if self.0.primary_index.pk_map.contains_key(&pk) {
                         self.update_with_guard(row.clone(), guard).await
                     } else {
+                        // `insert` acquires the same per-key mutation gate as
+                        // `guard`; release the row operation before entering
+                        // the synchronous insertion protocol, then retry the
+                        // locked decision if another writer won the race.
+                        drop(guard);
                         match self.insert(row.clone()) {
                             core::result::Result::Ok(_) => core::result::Result::Ok(()),
-                            core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => {
-                                self.update_with_guard(row.clone(), guard).await
-                            }
+                            core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) =>
+                                core::result::Result::Err(WorkTableError::NotFound),
                             core::result::Result::Err(e) => core::result::Result::Err(e),
                         }
                     };
@@ -199,7 +208,19 @@ impl InMemoryGenerator {
                         core::result::Result::Err(WorkTableError::PagesError(e)) if e.is_row_absent() => {}
                         other => return other,
                     }
-                    tokio::task::yield_now().await;
+                    if backoff_spins < 8 {
+                        backoff_spins = backoff_spins.saturating_add(1);
+                        tokio::task::yield_now().await;
+                    } else {
+                        // Cap the exponent BEFORE shifting: `1u64 << 64` panics
+                        // (overflow) in debug/test builds. Clamp the shift to a
+                        // 256µs ceiling and saturate the counter so a long
+                        // starvation streak can never overflow.
+                        let exponent = core::cmp::min(backoff_spins - 8, 8);
+                        let micros = core::cmp::min(1u64 << exponent, 256);
+                        backoff_spins = backoff_spins.saturating_add(1);
+                        tokio::time::sleep(std::time::Duration::from_micros(micros)).await;
+                    }
                 }
             }
         }