diff --git a/Cargo.toml b/Cargo.toml
index 36903ce7..3e1798bf 100644
--- a/Cargo.toml
+++ b/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
diff --git a/README.md b/README.md
index 85390100..f19a35b1 100644
--- a/README.md
+++ b/README.md
@@ -69,15 +69,11 @@ 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
@@ -87,15 +83,15 @@ 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. 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
diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml
index 1add3e47..c32ce6f2 100644
--- a/codegen/Cargo.toml
+++ b/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]
diff --git a/codegen/src/generators/in_memory/queries/delete.rs b/codegen/src/generators/in_memory/queries/delete.rs
index fad4375f..b405e837 100644
--- a/codegen/src/generators/in_memory/queries/delete.rs
+++ b/codegen/src/generators/in_memory/queries/delete.rs
@@ -118,7 +118,11 @@ impl InMemoryGenerator {
                         return Err(e);
                     }
                 };
-                let row = self.0.select(pk.clone()).unwrap();
+                // A lock-free insert publishes index reachability before it
+                // clears the staged row's ghost bit. Treat that window as an
+                // absent row: this delete linearizes before the insert's
+                // publication instead of panicking on the hidden version.
+                let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
                 #process
             }
         } else {
@@ -129,7 +133,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
             }
         }
diff --git a/codegen/src/generators/in_memory/table/index_fns.rs b/codegen/src/generators/in_memory/table/index_fns.rs
index 68670b69..9f30ad7f 100644
--- a/codegen/src/generators/in_memory/table/index_fns.rs
+++ b/codegen/src/generators/in_memory/table/index_fns.rs
@@ -83,36 +83,26 @@ impl InMemoryGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-        let select = if cfg!(feature = "versioned-row-publication") {
-            quote! {
-                for _ in 0..64 {
-                    let link: Link = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into)?;
-                    if let Ok(row) = self.0.data.select_non_ghosted(link) {
-                        if #predicate_matches {
-                            return Some(row);
-                        }
-                    }
-
-                    let current_link: Option<Link> = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into);
-                    if current_link == Some(link) {
-                        return None;
-                    }
-                    std::hint::spin_loop();
-                }
-                None
-            }
-        } else {
-            quote! {
+        let select = quote! {
+            for _ in 0..64 {
                 let link: Link = self.0.indexes.#field_ident
                     .lookup_for_select(#by)
                     .map(Into::into)?;
-                let row = self.0.data.select_non_ghosted(link).ok()?;
-                #predicate_matches.then_some(row)
+                if let Ok(row) = self.0.data.select_non_ghosted(link) {
+                    if #predicate_matches {
+                        return Some(row);
+                    }
+                }
+
+                let current_link: Option<Link> = self.0.indexes.#field_ident
+                    .lookup_for_select(#by)
+                    .map(Into::into);
+                if current_link == Some(link) {
+                    return None;
+                }
+                std::hint::spin_loop();
             }
+            None
         };
 
         Ok(quote! {
@@ -182,33 +172,21 @@ impl InMemoryGenerator {
         let row_field_ident = &idx.field;
         let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site());
 
-        let revalidate = cfg!(feature = "versioned-row-publication");
         let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) {
             (
                 quote! { std::ops::RangeBounds<#type_> },
-                if revalidate {
-                    quote! {
-                        (
-                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
-                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
-                        )
-                    }
-                } else {
-                    quote! {
+                quote! {
                     (
-                        range.start_bound().map(|v| OrderedFloat(*v)),
-                        range.end_bound().map(|v| OrderedFloat(*v)),
+                        predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                        predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
                     )
-                    }
                 },
             )
-        } else if revalidate {
+        } else {
             (
                 quote! { std::ops::RangeBounds<#type_> },
                 quote! { predicate_range.clone() },
             )
-        } else {
-            (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
         let (index_range, select_row) = if idx.is_unique {
             (
@@ -231,21 +209,17 @@ impl InMemoryGenerator {
                 },
             )
         };
-        let predicate_setup = revalidate.then(|| {
-            quote! {
-                let predicate_range = (
-                    range.start_bound().cloned(),
-                    range.end_bound().cloned(),
-                );
-            }
-        });
-        let predicate_filter = revalidate.then(|| {
-            quote! {
-                .filter(move |row| {
-                    std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
-                })
-            }
-        });
+        let predicate_setup = quote! {
+            let predicate_range = (
+                range.start_bound().cloned(),
+                range.end_bound().cloned(),
+            );
+        };
+        let predicate_filter = quote! {
+            .filter(move |row| {
+                std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
+            })
+        };
 
         Ok(quote! {
             pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident,
diff --git a/codegen/src/generators/persist/queries/delete.rs b/codegen/src/generators/persist/queries/delete.rs
index 0c3e2290..f1908e84 100644
--- a/codegen/src/generators/persist/queries/delete.rs
+++ b/codegen/src/generators/persist/queries/delete.rs
@@ -111,7 +111,11 @@ impl PersistGenerator {
                         return Err(e);
                     }
                 };
-                let row = self.0.select(pk.clone()).unwrap();
+                // A lock-free insert publishes index reachability before it
+                // clears the staged row's ghost bit. Treat that window as an
+                // absent row: this delete linearizes before the insert's
+                // publication instead of panicking on the hidden version.
+                let row = self.0.select(pk.clone()).ok_or(WorkTableError::NotFound)?;
                 #process
             }
         } else {
@@ -122,7 +126,7 @@ impl PersistGenerator {
                         .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
             }
         }
diff --git a/codegen/src/generators/persist/table/index_fns.rs b/codegen/src/generators/persist/table/index_fns.rs
index bf99fbe6..0259e010 100644
--- a/codegen/src/generators/persist/table/index_fns.rs
+++ b/codegen/src/generators/persist/table/index_fns.rs
@@ -83,36 +83,26 @@ impl PersistGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-        let select = if cfg!(feature = "versioned-row-publication") {
-            quote! {
-                for _ in 0..64 {
-                    let link: Link = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into)?;
-                    if let Ok(row) = self.0.data.select_non_ghosted(link) {
-                        if #predicate_matches {
-                            return Some(row);
-                        }
-                    }
-
-                    let current_link: Option<Link> = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into);
-                    if current_link == Some(link) {
-                        return None;
-                    }
-                    std::hint::spin_loop();
-                }
-                None
-            }
-        } else {
-            quote! {
+        let select = quote! {
+            for _ in 0..64 {
                 let link: Link = self.0.indexes.#field_ident
                     .lookup_for_select(#by)
                     .map(Into::into)?;
-                let row = self.0.data.select_non_ghosted(link).ok()?;
-                #predicate_matches.then_some(row)
+                if let Ok(row) = self.0.data.select_non_ghosted(link) {
+                    if #predicate_matches {
+                        return Some(row);
+                    }
+                }
+
+                let current_link: Option<Link> = self.0.indexes.#field_ident
+                    .lookup_for_select(#by)
+                    .map(Into::into);
+                if current_link == Some(link) {
+                    return None;
+                }
+                std::hint::spin_loop();
             }
+            None
         };
 
         Ok(quote! {
@@ -182,33 +172,21 @@ impl PersistGenerator {
         let row_field_ident = &idx.field;
         let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site());
 
-        let revalidate = cfg!(feature = "versioned-row-publication");
         let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) {
             (
                 quote! { std::ops::RangeBounds<#type_> },
-                if revalidate {
-                    quote! {
-                        (
-                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
-                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
-                        )
-                    }
-                } else {
-                    quote! {
+                quote! {
                     (
-                        range.start_bound().map(|v| OrderedFloat(*v)),
-                        range.end_bound().map(|v| OrderedFloat(*v)),
+                        predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                        predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
                     )
-                    }
                 },
             )
-        } else if revalidate {
+        } else {
             (
                 quote! { std::ops::RangeBounds<#type_> },
                 quote! { predicate_range.clone() },
             )
-        } else {
-            (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
         let (index_range, select_row) = if idx.is_unique {
             (
@@ -231,21 +209,17 @@ impl PersistGenerator {
                 },
             )
         };
-        let predicate_setup = revalidate.then(|| {
-            quote! {
-                let predicate_range = (
-                    range.start_bound().cloned(),
-                    range.end_bound().cloned(),
-                );
-            }
-        });
-        let predicate_filter = revalidate.then(|| {
-            quote! {
-                .filter(move |row| {
-                    std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
-                })
-            }
-        });
+        let predicate_setup = quote! {
+            let predicate_range = (
+                range.start_bound().cloned(),
+                range.end_bound().cloned(),
+            );
+        };
+        let predicate_filter = quote! {
+            .filter(move |row| {
+                std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
+            })
+        };
 
         Ok(quote! {
             pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident,
diff --git a/codegen/src/generators/read_only/table/index_fns.rs b/codegen/src/generators/read_only/table/index_fns.rs
index 42f9ff92..23b33dfb 100644
--- a/codegen/src/generators/read_only/table/index_fns.rs
+++ b/codegen/src/generators/read_only/table/index_fns.rs
@@ -83,36 +83,26 @@ impl ReadOnlyGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-        let select = if cfg!(feature = "versioned-row-publication") {
-            quote! {
-                for _ in 0..64 {
-                    let link: Link = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into)?;
-                    if let Ok(row) = self.0.data.select_non_ghosted(link) {
-                        if #predicate_matches {
-                            return Some(row);
-                        }
-                    }
-
-                    let current_link: Option<Link> = self.0.indexes.#field_ident
-                        .lookup_for_select(#by)
-                        .map(Into::into);
-                    if current_link == Some(link) {
-                        return None;
-                    }
-                    std::hint::spin_loop();
-                }
-                None
-            }
-        } else {
-            quote! {
+        let select = quote! {
+            for _ in 0..64 {
                 let link: Link = self.0.indexes.#field_ident
                     .lookup_for_select(#by)
                     .map(Into::into)?;
-                let row = self.0.data.select_non_ghosted(link).ok()?;
-                #predicate_matches.then_some(row)
+                if let Ok(row) = self.0.data.select_non_ghosted(link) {
+                    if #predicate_matches {
+                        return Some(row);
+                    }
+                }
+
+                let current_link: Option<Link> = self.0.indexes.#field_ident
+                    .lookup_for_select(#by)
+                    .map(Into::into);
+                if current_link == Some(link) {
+                    return None;
+                }
+                std::hint::spin_loop();
             }
+            None
         };
 
         Ok(quote! {
@@ -182,33 +172,21 @@ impl ReadOnlyGenerator {
         let row_field_ident = &idx.field;
         let column_pascal = Ident::new(&i.to_string().to_case(Case::Pascal), Span::mixed_site());
 
-        let revalidate = cfg!(feature = "versioned-row-publication");
         let (range_bounds, range_arg) = if is_float(type_.to_string().as_str()) {
             (
                 quote! { std::ops::RangeBounds<#type_> },
-                if revalidate {
-                    quote! {
-                        (
-                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
-                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
-                        )
-                    }
-                } else {
-                    quote! {
+                quote! {
                     (
-                        range.start_bound().map(|v| OrderedFloat(*v)),
-                        range.end_bound().map(|v| OrderedFloat(*v)),
+                        predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                        predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
                     )
-                    }
                 },
             )
-        } else if revalidate {
+        } else {
             (
                 quote! { std::ops::RangeBounds<#type_> },
                 quote! { predicate_range.clone() },
             )
-        } else {
-            (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
         let (index_range, select_row) = if idx.is_unique {
             (
@@ -231,21 +209,17 @@ impl ReadOnlyGenerator {
                 },
             )
         };
-        let predicate_setup = revalidate.then(|| {
-            quote! {
-                let predicate_range = (
-                    range.start_bound().cloned(),
-                    range.end_bound().cloned(),
-                );
-            }
-        });
-        let predicate_filter = revalidate.then(|| {
-            quote! {
-                .filter(move |row| {
-                    std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
-                })
-            }
-        });
+        let predicate_setup = quote! {
+            let predicate_range = (
+                range.start_bound().cloned(),
+                range.end_bound().cloned(),
+            );
+        };
+        let predicate_filter = quote! {
+            .filter(move |row| {
+                std::ops::RangeBounds::contains(&predicate_range, &row.#row_field_ident)
+            })
+        };
 
         Ok(quote! {
             pub fn #fn_name<'a, R>(&'a self, range: R) -> SelectQueryBuilder<#row_ident,
diff --git a/docs/index-backend-dsl-proposal.md b/docs/index-backend-dsl-proposal.md
index 72e3f94b..732fbe43 100644
--- a/docs/index-backend-dsl-proposal.md
+++ b/docs/index-backend-dsl-proposal.md
@@ -164,16 +164,20 @@ definitive; a contended lookup drops the structural guard before waiting and
 then retries the mapping. Vanilla IndexSet does not expose a comparable
 structural validation primitive; `using indexset` therefore remains
 experimental and is excluded from concurrent correctness and published
-performance claims. This WorkTablesIndex visibility guarantee is independent of
-`versioned-row-publication`, which addresses concurrent page bytes, ghost
-publication, and reclamation rather than index routing.
+performance claims. This WorkTablesIndex visibility guarantee composes with
+the mandatory immutable row-publication protocol, which addresses concurrent
+page bytes, ghost publication, and reclamation rather than index routing.
 
 ### Congee
 
 - Point lookup and mutation call Congee directly.
 - WorkTable links do not fit in Congee's one-word payload. The adapter stores an `Arc` pointer, so inserts allocate and reads clone the `Arc` before copying the link.
 - Ordered reads use Congee's native range scan and materialize only the requested key interval into a `Vec`; a full iteration is therefore O(n) with one result allocation, while a narrow range no longer dumps, re-probes, and sorts the whole tree.
-- With `persist: true`, mutations additionally take a key-striped sequencing lock before producing one logical WAL event. Memory-only Congee does not pay that cost.
+- Memory-only Congee point reads remain native and concurrent. Mutations use a
+  WorkTable adapter mutex because congee-wt 0.4.1 can otherwise lose disjoint
+  structural insert/remove updates. With `persist: true`, the persistence layer
+  additionally takes a key-striped sequencing lock before producing one logical
+  WAL event.
 
 ### Arctic
 
diff --git a/docs/versioned-row-publication.md b/docs/versioned-row-publication.md
index 9d79dd2b..08ee2139 100644
--- a/docs/versioned-row-publication.md
+++ b/docs/versioned-row-publication.md
@@ -1,6 +1,7 @@
 # Versioned row publication
 
-Status: feature-gated prototype behind `versioned-row-publication`.
+Status: mandatory for generated safe APIs. The former
+`versioned-row-publication` Cargo feature is retained as a compatibility no-op.
 
 ## Problem
 
@@ -19,7 +20,7 @@ the interval from index lookup through acquisition of a stable row version.
 
 ## Protocol
 
-With the feature enabled, `DataPages` maintains two representations:
+`DataPages` maintains two representations:
 
 - Archived page bytes are the compact persistence and mutation image. All
   accesses that can overlap a mutation are serialized by an internal page
@@ -63,7 +64,7 @@ The generated API follows these publication rules:
    admitting write traffic. Subsequent generated reads use the published
    version map.
 
-The grace period is quiescent-state reclamation: a feature-only atomic counter
+The grace period is quiescent-state reclamation: an atomic counter
 tracks generated reads, and retirement queues are drained when that counter is
 zero. `Arc` ownership independently keeps a version alive after a reader has
 acquired it.
@@ -102,16 +103,13 @@ than spinning without a bound. The guarantee also does not cover callers that
 bypass generated table methods and directly invoke low-level `Data` page
 mutation APIs.
 
-## Cost model and rollout
-
-The feature is off by default. Cargo features unify across a dependency graph,
-so any dependency enabling it enables it for every WorkTable consumer in that
-build. It adds one owned row copy plus slot/map
-metadata per live physical link, an atomic increment/decrement per generated
-read, a sharded publication-map lookup, and writer-side page serialization.
-The index-visibility algorithm is always active and separate from row
-publication: WorkTablesIndex acquires the selected node while its structural
-mapping is pinned on the uncontended path, and may retry after node contention.
-Those costs are inappropriate to impose silently on latency-sensitive users.
-The default path remains unchanged; benchmark results for both modes must be
-reported before this feature is proposed for default enablement.
+## Cost model
+
+The protocol adds one owned row copy plus slot/map metadata per live physical
+link, an atomic increment/decrement per generated read, a sharded publication
+map lookup, and writer-side page serialization. These costs are mandatory: the
+previous fast path allowed safe generated reads to race mutation of archived
+bytes, and performance cannot justify undefined behavior. The index-visibility
+algorithm is separate: WorkTablesIndex acquires the selected node while its
+structural mapping is pinned on the uncontended path, and may retry after node
+contention.
diff --git a/src/in_memory/mod.rs b/src/in_memory/mod.rs
index 0da40382..fefa6add 100644
--- a/src/in_memory/mod.rs
+++ b/src/in_memory/mod.rs
@@ -1,7 +1,6 @@
 mod data;
 mod empty_link_registry;
 mod pages;
-#[cfg(feature = "versioned-row-publication")]
 mod publication;
 mod row;
 
diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs
index c832d944..58c80b6e 100644
--- a/src/in_memory/pages.rs
+++ b/src/in_memory/pages.rs
@@ -1,6 +1,5 @@
 use data_bucket::page::PageId;
 use derive_more::{Display, Error, From};
-#[cfg(feature = "versioned-row-publication")]
 use parking_lot::Mutex;
 use parking_lot::RwLock;
 #[cfg(feature = "perf_measurements")]
@@ -12,13 +11,10 @@ use rkyv::{
     ser::{Serializer, allocator::ArenaHandle, sharing::Share},
     util::AlignedVec,
 };
-#[cfg(feature = "versioned-row-publication")]
 use std::collections::HashMap;
 use std::collections::VecDeque;
-#[cfg(feature = "versioned-row-publication")]
 use std::hash::{BuildHasherDefault, Hasher};
 use std::marker::PhantomData;
-#[cfg(feature = "versioned-row-publication")]
 use std::sync::atomic::AtomicUsize;
 use std::{
     fmt::Debug,
@@ -27,10 +23,8 @@ use std::{
 };
 
 use crate::in_memory::empty_link_registry::EmptyLinkRegistry;
-#[cfg(feature = "versioned-row-publication")]
 use crate::in_memory::publication::{DELETED, GHOSTED, PublishedRow, VACUUMED};
 use crate::prelude::ArchivedRowWrapper;
-#[cfg(feature = "versioned-row-publication")]
 use crate::util::OffsetEqLink;
 use crate::{
     in_memory::{
@@ -44,12 +38,9 @@ fn page_id_mapper(page_id: usize) -> usize {
     page_id - 1usize
 }
 
-#[cfg(feature = "versioned-row-publication")]
 const PUBLICATION_SHARD_COUNT: usize = 64;
-#[cfg(feature = "versioned-row-publication")]
 const RETIREMENT_BACKLOG_WARN_AT: usize = 1_024;
 
-#[cfg(feature = "versioned-row-publication")]
 fn mix_publication_offset(mut value: u64) -> u64 {
     value ^= value >> 30;
     value = value.wrapping_mul(0xbf58_476d_1ce4_e5b9);
@@ -62,17 +53,14 @@ fn mix_publication_offset(mut value: u64) -> u64 {
 /// storage offset. Avalanche that offset so both hash-table bucket bits and
 /// SIMD control bits remain distributed for aligned, monotonically allocated
 /// row positions.
-#[cfg(feature = "versioned-row-publication")]
 struct PublicationHasher(u64);
 
-#[cfg(feature = "versioned-row-publication")]
 impl Default for PublicationHasher {
     fn default() -> Self {
         Self(0xcbf2_9ce4_8422_2325)
     }
 }
 
-#[cfg(feature = "versioned-row-publication")]
 impl Hasher for PublicationHasher {
     fn finish(&self) -> u64 {
         self.0
@@ -92,20 +80,16 @@ impl Hasher for PublicationHasher {
     }
 }
 
-#[cfg(feature = "versioned-row-publication")]
 type PublicationMap<Row, const DATA_LENGTH: usize> =
     HashMap<OffsetEqLink<DATA_LENGTH>, Arc<PublishedRow<Row>>, BuildHasherDefault<PublicationHasher>>;
 
-#[cfg(feature = "versioned-row-publication")]
 type PublicationShards<Row, const DATA_LENGTH: usize> =
     [RwLock<PublicationMap<Row, DATA_LENGTH>>; PUBLICATION_SHARD_COUNT];
 
-#[cfg(feature = "versioned-row-publication")]
 fn publication_shard<const DATA_LENGTH: usize>(key: &OffsetEqLink<DATA_LENGTH>) -> usize {
     mix_publication_offset(key.absolute_index()) as usize & (PUBLICATION_SHARD_COUNT - 1)
 }
 
-#[cfg(feature = "versioned-row-publication")]
 fn queue_retirement<T>(queue: &Mutex<Vec<T>>, pending_retirements: &AtomicUsize, queue_name: &'static str, value: T) {
     let mut queue = queue.lock();
     queue.push(value);
@@ -121,19 +105,17 @@ fn queue_retirement<T>(queue: &Mutex<Vec<T>>, pending_retirements: &AtomicUsize,
 }
 
 pub struct ReadGuard<'a> {
-    #[cfg(feature = "versioned-row-publication")]
     active_readers: &'a AtomicU64,
     marker: PhantomData<&'a ()>,
 }
 
 impl Drop for ReadGuard<'_> {
     fn drop(&mut self) {
-        #[cfg(feature = "versioned-row-publication")]
         self.active_readers.fetch_sub(1, Ordering::SeqCst);
     }
 }
 
-/// Page storage and, when enabled, immutable row publication.
+/// Page storage with immutable row publication.
 ///
 /// # Versioned-publication synchronization
 ///
@@ -155,31 +137,24 @@ where
 {
     /// Immutable application-visible row versions. Published readers never
     /// borrow the mutable archived page image.
-    #[cfg(feature = "versioned-row-publication")]
     published_rows: PublicationShards<Row, DATA_LENGTH>,
 
     /// Protects the mutable page image used by writers, vacuum, and
     /// persistence. Application reads use `published_rows` after hydration.
-    #[cfg(feature = "versioned-row-publication")]
     page_access: RwLock<()>,
 
     /// Read-side grace period protecting the interval from index lookup until
     /// an immutable row version has been acquired.
-    #[cfg(feature = "versioned-row-publication")]
     active_readers: AtomicU64,
 
-    #[cfg(feature = "versioned-row-publication")]
     retired_links: Mutex<Vec<Link>>,
 
-    #[cfg(feature = "versioned-row-publication")]
     retired_pages: Mutex<Vec<PageId>>,
 
-    #[cfg(feature = "versioned-row-publication")]
     retired_publications: Mutex<Vec<OffsetEqLink<DATA_LENGTH>>>,
 
     /// Avoids taking all retirement-queue mutexes on mutations when there is
     /// no reclamation work pending.
-    #[cfg(feature = "versioned-row-publication")]
     pending_retirements: AtomicUsize,
 
     /// Pages vector. Currently, not lock free.
@@ -212,7 +187,6 @@ where
     Row: StorableRow,
     <Row as StorableRow>::WrappedRow: RowWrapper<Row>,
 {
-    #[cfg(feature = "versioned-row-publication")]
     fn publication_flags(row: &<Row as StorableRow>::WrappedRow) -> u8 {
         let mut flags = 0;
         if row.is_ghosted() {
@@ -227,7 +201,6 @@ where
         flags
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     fn publish_wrapped_row(&self, link: Link, wrapped: <Row as StorableRow>::WrappedRow) {
         let flags = Self::publication_flags(&wrapped);
         let row = wrapped.get_inner();
@@ -242,19 +215,16 @@ where
         }
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     fn stage_published_row(&self, link: Link, row: Row) {
         let wrapped = <Row as StorableRow>::WrappedRow::from_inner(row);
         self.publish_wrapped_row(link, wrapped);
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     fn published_slot(&self, link: Link) -> Option<Arc<PublishedRow<Row>>> {
         let key = OffsetEqLink(link);
         self.published_rows[publication_shard(&key)].read().get(&key).cloned()
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     fn published_slot_or_hydrate(&self, link: Link) -> Result<Arc<PublishedRow<Row>>, ExecutionError>
     where
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
@@ -282,17 +252,14 @@ where
     }
 
     pub fn read_guard(&self) -> ReadGuard<'_> {
-        #[cfg(feature = "versioned-row-publication")]
         self.active_readers.fetch_add(1, Ordering::SeqCst);
 
         ReadGuard {
-            #[cfg(feature = "versioned-row-publication")]
             active_readers: &self.active_readers,
             marker: PhantomData,
         }
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     fn reclaim_retired(&self) {
         if self.pending_retirements.load(Ordering::Acquire) == 0 {
             return;
@@ -325,19 +292,12 @@ where
 
     pub fn new() -> Self {
         Self {
-            #[cfg(feature = "versioned-row-publication")]
             published_rows: std::array::from_fn(|_| RwLock::new(PublicationMap::default())),
-            #[cfg(feature = "versioned-row-publication")]
             page_access: RwLock::new(()),
-            #[cfg(feature = "versioned-row-publication")]
             active_readers: AtomicU64::new(0),
-            #[cfg(feature = "versioned-row-publication")]
             retired_links: Mutex::new(Vec::new()),
-            #[cfg(feature = "versioned-row-publication")]
             retired_pages: Mutex::new(Vec::new()),
-            #[cfg(feature = "versioned-row-publication")]
             retired_publications: Mutex::new(Vec::new()),
-            #[cfg(feature = "versioned-row-publication")]
             pending_retirements: AtomicUsize::new(0),
             // We are starting ID's from `1` because `0`'s page in file is info page.
             pages: RwLock::new(vec![Arc::new(Data::new(1.into()))]),
@@ -356,19 +316,12 @@ where
         } else {
             let last_page_id = vec.len();
             Self {
-                #[cfg(feature = "versioned-row-publication")]
                 published_rows: std::array::from_fn(|_| RwLock::new(PublicationMap::default())),
-                #[cfg(feature = "versioned-row-publication")]
                 page_access: RwLock::new(()),
-                #[cfg(feature = "versioned-row-publication")]
                 active_readers: AtomicU64::new(0),
-                #[cfg(feature = "versioned-row-publication")]
                 retired_links: Mutex::new(Vec::new()),
-                #[cfg(feature = "versioned-row-publication")]
                 retired_pages: Mutex::new(Vec::new()),
-                #[cfg(feature = "versioned-row-publication")]
                 retired_publications: Mutex::new(Vec::new()),
-                #[cfg(feature = "versioned-row-publication")]
                 pending_retirements: AtomicUsize::new(0),
                 pages: RwLock::new(vec),
                 empty_links: EmptyLinkRegistry::default(),
@@ -388,16 +341,11 @@ where
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
     {
-        #[cfg(feature = "versioned-row-publication")]
         let general_row = <Row as StorableRow>::WrappedRow::from_inner(row.clone());
-        #[cfg(not(feature = "versioned-row-publication"))]
-        let general_row = <Row as StorableRow>::WrappedRow::from_inner(row);
 
-        #[cfg(feature = "versioned-row-publication")]
         self.reclaim_retired();
 
         if let Some(link) = self.empty_links.pop_max() {
-            #[cfg(feature = "versioned-row-publication")]
             let _page_access = self.page_access.write();
             let pages = self.pages.read();
             let current_page: usize = page_id_mapper(link.page_id.into());
@@ -408,7 +356,6 @@ where
                     if let Some(l) = left_link {
                         self.empty_links.push(l);
                     }
-                    #[cfg(feature = "versioned-row-publication")]
                     self.stage_published_row(link, row);
                     return Ok(link);
                 }
@@ -426,7 +373,6 @@ where
 
         loop {
             let (link, tried_page) = {
-                #[cfg(feature = "versioned-row-publication")]
                 let _page_access = self.page_access.write();
                 let pages = self.pages.read();
                 let current_page = page_id_mapper(self.current_page_id.load(Ordering::Acquire) as usize);
@@ -436,7 +382,6 @@ where
             };
             match link {
                 Ok(link) => {
-                    #[cfg(feature = "versioned-row-publication")]
                     self.stage_published_row(link, row);
                     self.row_count.fetch_add(1, Ordering::Relaxed);
                     return Ok(link);
@@ -492,7 +437,6 @@ where
     /// Allocates a new page or reuses a free page from `empty_pages`.
     /// Does **NOT** set the page as `current`.
     pub fn allocate_new_or_pop_free(&self) -> Arc<Data<<Row as StorableRow>::WrappedRow, DATA_LENGTH>> {
-        #[cfg(feature = "versioned-row-publication")]
         self.reclaim_retired();
 
         let page_id = {
@@ -501,7 +445,6 @@ where
         };
 
         if let Some(page_id) = page_id {
-            #[cfg(feature = "versioned-row-publication")]
             let _page_access = self.page_access.write();
             let pages = self.pages.read();
             let index = page_id_mapper(page_id.into());
@@ -529,21 +472,8 @@ where
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
         let link = link.into();
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            let slot = self.published_slot_or_hydrate(link)?;
-            Ok(slot.snapshot().as_ref().clone())
-        }
-
-        #[cfg(not(feature = "versioned-row-publication"))]
-        {
-            let pages = self.pages.read();
-            let page = pages
-                .get(page_id_mapper(link.page_id.into()))
-                .ok_or(ExecutionError::PageNotFound(link.page_id))?;
-            let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?;
-            Ok(gen_row.get_inner())
-        }
+        let slot = self.published_slot_or_hydrate(link)?;
+        Ok(slot.snapshot().as_ref().clone())
     }
 
     pub fn select_non_ghosted(&self, link: Link) -> Result<Row, ExecutionError>
@@ -554,31 +484,15 @@ where
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            let slot = self.published_slot_or_hydrate(link)?;
-            let (row, flags) = slot.load();
-            if flags & GHOSTED != 0 {
-                return Err(ExecutionError::Ghosted);
-            }
-            if flags & DELETED != 0 {
-                return Err(ExecutionError::Deleted);
-            }
-            Ok(row.as_ref().clone())
+        let slot = self.published_slot_or_hydrate(link)?;
+        let (row, flags) = slot.load();
+        if flags & GHOSTED != 0 {
+            return Err(ExecutionError::Ghosted);
         }
-
-        #[cfg(not(feature = "versioned-row-publication"))]
-        {
-            let pages = self.pages.read();
-            let page = pages
-                .get(page_id_mapper(link.page_id.into()))
-                .ok_or(ExecutionError::PageNotFound(link.page_id))?;
-            let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?;
-            if gen_row.is_ghosted() {
-                return Err(ExecutionError::Ghosted);
-            }
-            Ok(gen_row.get_inner())
+        if flags & DELETED != 0 {
+            return Err(ExecutionError::Deleted);
         }
+        Ok(row.as_ref().clone())
     }
 
     pub fn select_non_vacuumed(&self, link: Link) -> Result<Row, ExecutionError>
@@ -589,37 +503,18 @@ where
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            let slot = self.published_slot_or_hydrate(link)?;
-            let (row, flags) = slot.load();
-            if flags & GHOSTED != 0 {
-                return Err(ExecutionError::Ghosted);
-            }
-            if flags & VACUUMED != 0 {
-                return Err(ExecutionError::Vacuumed);
-            }
-            if flags & DELETED != 0 {
-                return Err(ExecutionError::Deleted);
-            }
-            Ok(row.as_ref().clone())
+        let slot = self.published_slot_or_hydrate(link)?;
+        let (row, flags) = slot.load();
+        if flags & GHOSTED != 0 {
+            return Err(ExecutionError::Ghosted);
         }
-
-        #[cfg(not(feature = "versioned-row-publication"))]
-        {
-            let pages = self.pages.read();
-            let page = pages
-                .get(page_id_mapper(link.page_id.into()))
-                .ok_or(ExecutionError::PageNotFound(link.page_id))?;
-            let gen_row = page.get_row(link).map_err(ExecutionError::DataPageError)?;
-            if gen_row.is_ghosted() {
-                return Err(ExecutionError::Ghosted);
-            }
-            if gen_row.is_vacuumed() {
-                return Err(ExecutionError::Vacuumed);
-            }
-            Ok(gen_row.get_inner())
+        if flags & VACUUMED != 0 {
+            return Err(ExecutionError::Vacuumed);
         }
+        if flags & DELETED != 0 {
+            return Err(ExecutionError::Deleted);
+        }
+        Ok(row.as_ref().clone())
     }
 
     #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "DataPages"))]
@@ -628,7 +523,6 @@ where
         Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         Op: Fn(&<<Row as StorableRow>::WrappedRow as Archive>::Archived) -> Res,
     {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.read();
         let pages = self.pages.read();
         let page = pages
@@ -649,7 +543,6 @@ where
             Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
         Op: FnMut(&mut <<Row as StorableRow>::WrappedRow as Archive>::Archived) -> Res,
     {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.write();
         let pages = self.pages.read();
         let page = pages
@@ -664,7 +557,6 @@ where
             op(gen_row)
         };
 
-        #[cfg(feature = "versioned-row-publication")]
         {
             let wrapped = page.get_row(link).map_err(ExecutionError::DataPageError)?;
             self.publish_wrapped_row(link, wrapped);
@@ -685,21 +577,16 @@ where
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
     {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.write();
         let pages = self.pages.read();
         let page = pages
             .get(page_id_mapper(link.page_id.into()))
             .ok_or(ExecutionError::PageNotFound(link.page_id))?;
-        #[cfg(feature = "versioned-row-publication")]
         let gen_row = <Row as StorableRow>::WrappedRow::from_inner(row.clone());
-        #[cfg(not(feature = "versioned-row-publication"))]
-        let gen_row = <Row as StorableRow>::WrappedRow::from_inner(row);
         let result = unsafe {
             page.save_row_by_link(&gen_row, link)
                 .map_err(ExecutionError::DataPageError)
         }?;
-        #[cfg(feature = "versioned-row-publication")]
         self.stage_published_row(link, row);
         Ok(result)
     }
@@ -715,19 +602,12 @@ where
     {
         unsafe { self.with_mut_ref(link, |r| r.delete())? }
 
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            queue_retirement(&self.retired_links, &self.pending_retirements, "links", link);
-            self.reclaim_retired();
-        }
-
-        #[cfg(not(feature = "versioned-row-publication"))]
-        self.empty_links.push(link);
+        queue_retirement(&self.retired_links, &self.pending_retirements, "links", link);
+        self.reclaim_retired();
         Ok(())
     }
 
     pub fn select_raw(&self, link: Link) -> Result<Vec<u8>, ExecutionError> {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.read();
         let pages = self.pages.read();
         let page = pages
@@ -738,17 +618,8 @@ where
 
     pub fn mark_page_empty(&self, page_id: PageId) {
         if u32::from(page_id) != self.current_page_id.load(Ordering::Acquire) {
-            #[cfg(feature = "versioned-row-publication")]
-            {
-                queue_retirement(&self.retired_pages, &self.pending_retirements, "pages", page_id);
-                self.reclaim_retired();
-            }
-
-            #[cfg(not(feature = "versioned-row-publication"))]
-            {
-                let mut g = self.empty_pages.write();
-                g.push_back(page_id);
-            }
+            queue_retirement(&self.retired_pages, &self.pending_retirements, "pages", page_id);
+            self.reclaim_retired();
         }
     }
 
@@ -794,7 +665,6 @@ where
     }
 
     pub fn get_bytes(&self) -> Vec<([u8; DATA_LENGTH], u32)> {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.read();
         let pages = self.pages.read();
         pages
@@ -804,7 +674,6 @@ where
     }
 
     pub(crate) fn reset_page(&self, page_id: PageId) -> Result<(), ExecutionError> {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.write();
         let pages = self.pages.read();
         let page = pages
@@ -837,7 +706,6 @@ where
             + Portable
             + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        #[cfg(feature = "versioned-row-publication")]
         let _page_access = self.page_access.write();
         let pages = self.pages.read();
         let from_page = pages
@@ -859,7 +727,6 @@ where
         archived.set_in_vacuum_process();
         let new_link = to_page.save_raw_row(&raw_data).map_err(ExecutionError::DataPageError)?;
 
-        #[cfg(feature = "versioned-row-publication")]
         {
             let old_wrapped = from_page.get_row(from_link).map_err(ExecutionError::DataPageError)?;
             self.publish_wrapped_row(from_link, old_wrapped);
@@ -871,19 +738,13 @@ where
     }
 
     pub(crate) fn retire_published_link(&self, link: Link) {
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            queue_retirement(
-                &self.retired_publications,
-                &self.pending_retirements,
-                "publications",
-                OffsetEqLink(link),
-            );
-            self.reclaim_retired();
-        }
-
-        #[cfg(not(feature = "versioned-row-publication"))]
-        let _ = link;
+        queue_retirement(
+            &self.retired_publications,
+            &self.pending_retirements,
+            "publications",
+            OffsetEqLink(link),
+        );
+        self.reclaim_retired();
     }
 
     pub fn get_page_count(&self) -> usize {
@@ -950,10 +811,8 @@ mod tests {
     use std::collections::HashSet;
     use std::sync::Arc;
     use std::sync::atomic::{AtomicBool, Ordering};
-    #[cfg(feature = "versioned-row-publication")]
     use std::sync::mpsc;
     use std::thread;
-    #[cfg(feature = "versioned-row-publication")]
     use std::time::Duration;
     use std::time::Instant;
 
@@ -1091,7 +950,6 @@ mod tests {
         assert_eq!(res.err(), Some(PagesExecutionError::Ghosted))
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     #[test]
     fn versioned_insert_stays_hidden_until_unghost() {
         let pages = DataPages::<TestRow>::new();
@@ -1105,7 +963,6 @@ mod tests {
         assert_eq!(pages.select_non_ghosted(link), Ok(row));
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     #[test]
     fn versioned_reader_observes_old_row_while_page_update_is_incomplete() {
         let pages = Arc::new(DataPages::<TestRow>::new());
@@ -1147,7 +1004,6 @@ mod tests {
         assert_eq!(pages.select_non_ghosted(link), Ok(TestRow { a: 1, b: 1 }));
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     #[test]
     fn retired_version_survives_link_reuse_for_in_flight_reader() {
         let pages = DataPages::<TestRow>::new();
@@ -1170,7 +1026,6 @@ mod tests {
         assert_eq!(pages.select_non_ghosted(reused_link), Ok(TestRow { a: 2, b: 2 }));
     }
 
-    #[cfg(feature = "versioned-row-publication")]
     #[test]
     fn read_grace_period_prevents_link_aba() {
         let pages = DataPages::<TestRow>::new();
diff --git a/src/in_memory/row.rs b/src/in_memory/row.rs
index 115b9b73..89eedaa1 100644
--- a/src/in_memory/row.rs
+++ b/src/in_memory/row.rs
@@ -2,18 +2,10 @@ use std::fmt::Debug;
 
 use rkyv::Archive;
 
-#[cfg(feature = "versioned-row-publication")]
 pub trait PublicationSafe: Send + Sync + 'static {}
 
-#[cfg(feature = "versioned-row-publication")]
 impl<T: Send + Sync + 'static> PublicationSafe for T {}
 
-#[cfg(not(feature = "versioned-row-publication"))]
-pub trait PublicationSafe {}
-
-#[cfg(not(feature = "versioned-row-publication"))]
-impl<T> PublicationSafe for T {}
-
 /// Common trait for the `Row`s that can be stored on the [`Data`] page.
 ///
 /// [`Data`]: crate::in_memory::data::Data
diff --git a/src/index/congee.rs b/src/index/congee.rs
index f2e41846..06e8a4d5 100644
--- a/src/index/congee.rs
+++ b/src/index/congee.rs
@@ -6,6 +6,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use congee::{CongeeRaw, DefaultAllocator};
+use parking_lot::Mutex;
 
 use super::UniqueIndex;
 
@@ -47,6 +48,10 @@ impl_congee_key!(u64);
 /// reclamation pattern used by Congee's own `CongeeArc` implementation.
 pub struct CongeeIndex<K, V> {
     inner: CongeeRaw<usize, usize>,
+    // congee-wt 0.4.1 can lose disjoint insert/remove mutations when their
+    // structural updates overlap. Keep point reads native and concurrent, but
+    // serialize mutations until the backend offers the required visibility.
+    mutation: Mutex<()>,
     len: AtomicUsize,
     marker: std::marker::PhantomData<(K, V)>,
 }
@@ -72,6 +77,7 @@ where
         };
         Self {
             inner: CongeeRaw::new_with_drainer(DefaultAllocator {}, drainer),
+            mutation: Mutex::new(()),
             len: AtomicUsize::new(0),
             marker: std::marker::PhantomData,
         }
@@ -190,6 +196,7 @@ where
         let len = inner.keys().len();
         Ok(Self {
             inner,
+            mutation: Mutex::new(()),
             len: AtomicUsize::new(len),
             marker: std::marker::PhantomData,
         })
@@ -224,6 +231,7 @@ where
 
     #[inline]
     fn insert_value(&self, key: K, value: V) -> Option<V> {
+        let _mutation = self.mutation.lock();
         let guard = self.inner.pin();
         let pointer = Arc::into_raw(Arc::new(value)).expose_provenance();
         match self.inner.insert(key.into_congee(), pointer, &guard) {
@@ -242,6 +250,7 @@ where
 
     #[inline]
     fn insert_value_checked(&self, key: K, value: V) -> Option<()> {
+        let _mutation = self.mutation.lock();
         let guard = self.inner.pin();
         let pointer = Arc::into_raw(Arc::new(value)).expose_provenance();
         let result = self
@@ -269,6 +278,7 @@ where
 
     #[inline]
     fn remove_value(&self, key: &K) -> Option<(K, V)> {
+        let _mutation = self.mutation.lock();
         let guard = self.inner.pin();
         let pointer = self.inner.remove(&key.into_congee(), &guard)?;
         self.len.fetch_sub(1, Ordering::Relaxed);
diff --git a/src/index/unique.rs b/src/index/unique.rs
index d4ba8a01..55e559f3 100644
--- a/src/index/unique.rs
+++ b/src/index/unique.rs
@@ -292,9 +292,22 @@ mod tests {
             threads.push(std::thread::spawn(move || {
                 for sequence in 0..1_000_u64 {
                     let key = worker * 1_000 + sequence;
-                    assert_eq!(index.insert_value_checked(key, key + 1), Some(()));
-                    assert_eq!(index.get_value(&key), Some(key + 1));
-                    assert_eq!(index.remove_value(&key), Some((key, key + 1)));
+                    let backend = std::any::type_name::<I>();
+                    assert_eq!(
+                        index.insert_value_checked(key, key + 1),
+                        Some(()),
+                        "backend={backend}, key={key}, operation=insert"
+                    );
+                    assert_eq!(
+                        index.get_value(&key),
+                        Some(key + 1),
+                        "backend={backend}, key={key}, operation=get"
+                    );
+                    assert_eq!(
+                        index.remove_value(&key),
+                        Some((key, key + 1)),
+                        "backend={backend}, key={key}, operation=remove"
+                    );
                 }
             }));
         }
diff --git a/src/table/mod.rs b/src/table/mod.rs
index aabf6d28..dc55f52a 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -140,32 +140,19 @@ where
             Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
         let _read_guard = self.data.read_guard();
-        #[cfg(feature = "versioned-row-publication")]
-        {
-            for _ in 0..64 {
-                let link = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into)?;
-                if let Ok(row) = self.data.select_non_ghosted(link) {
-                    return Some(row);
-                }
-
-                let current_link: Option<Link> = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into);
-                if current_link == Some(link) {
-                    return None;
-                }
-                std::hint::spin_loop();
+        for _ in 0..64 {
+            let link = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into)?;
+            if let Ok(row) = self.data.select_non_ghosted(link) {
+                return Some(row);
             }
-            None
-        }
 
-        #[cfg(not(feature = "versioned-row-publication"))]
-        {
-            let link = self.primary_index.pk_map.lookup_for_select(&pk).map(|value| value.0);
-            if let Some(link) = link {
-                self.data.select_non_ghosted(link).ok()
-            } else {
-                None
+            let current_link: Option<Link> = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into);
+            if current_link == Some(link) {
+                return None;
             }
+            std::hint::spin_loop();
         }
+        None
     }
 
     #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "WorkTable"))]
diff --git a/tests/worktable/float.rs b/tests/worktable/float.rs
index 576c707b..a924d343 100644
--- a/tests/worktable/float.rs
+++ b/tests/worktable/float.rs
@@ -54,7 +54,6 @@ fn unique_float_point_read_revalidates_the_returned_row() {
     assert_eq!(table.select_by_value(second.value), Some(second));
 }
 
-#[cfg(feature = "versioned-row-publication")]
 #[test]
 fn float_range_read_revalidates_each_resolved_row() {
     let table = TestFloatWorkTable::default();
diff --git a/tests/worktable/index/range.rs b/tests/worktable/index/range.rs
index 19497bac..d9ac9d71 100644
--- a/tests/worktable/index/range.rs
+++ b/tests/worktable/index/range.rs
@@ -32,7 +32,6 @@ worktable!(
     }
 );
 
-#[cfg(feature = "versioned-row-publication")]
 #[tokio::test]
 async fn idle_select_builder_does_not_pin_retired_links() {
     let table = UniqueRangeTestWorkTable::default();
@@ -62,7 +61,6 @@ async fn idle_select_builder_does_not_pin_retired_links() {
     drop(idle_query);
 }
 
-#[cfg(feature = "versioned-row-publication")]
 #[test]
 fn range_read_revalidates_each_resolved_row() {
     let table = RangeTestWorkTable::default();
diff --git a/tests/worktable/upsert.rs b/tests/worktable/upsert.rs
index 7ba61a3b..f5e55a09 100644
--- a/tests/worktable/upsert.rs
+++ b/tests/worktable/upsert.rs
@@ -76,6 +76,33 @@ async fn raw_insert_delete_churn_never_panics_or_stalls() {
     }
 }
 
+/// Pins the exact publication schedule that used to let delete unwrap a
+/// ghosted row: data and primary-index reachability exist, but insert has not
+/// yet cleared the lifecycle bit. Delete must linearize before publication and
+/// leave the staged insert intact.
+#[tokio::test]
+async fn delete_during_insert_publication_window_returns_not_found() {
+    let table = UpsertChurnWorkTable::default();
+    const KEY: u64 = 7;
+    let row = UpsertChurnRow { id: KEY, val: 11 };
+
+    let link = table.0.data.insert(row.clone()).unwrap();
+    assert!(
+        table
+            .0
+            .primary_index
+            .insert_checked(UpsertChurnPrimaryKey::from(KEY), link)
+            .is_some()
+    );
+
+    assert!(matches!(table.delete(KEY).await, Err(WorkTableError::NotFound)));
+
+    unsafe {
+        table.0.data.with_mut_ref(link, |staged| staged.unghost()).unwrap();
+    }
+    assert_eq!(table.select(KEY), Some(row));
+}
+
 async fn churn_run(churn_flips: u64, upserts_per_task: u64) {
     #[allow(non_snake_case)]
     let CHURN_FLIPS = churn_flips;