diff --git a/Cargo.toml b/Cargo.toml
index 93bf735..8cd6a41 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,6 +17,7 @@ categories = ["database-implementations", "data-structures", "caching"]
 [features]
 perf_measurements = ["dep:performance_measurement", "dep:performance_measurement_codegen"]
 s3-support = ["dep:rusty-s3", "dep:url", "dep:reqwest", "dep:walkdir", "worktable_codegen/s3-support"]
+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 d2d09f3..f4a487c 100644
--- a/README.md
+++ b/README.md
@@ -46,6 +46,30 @@ S3 support layers *on top of* the disk engine rather than replacing it.
 worktable = { version = "0.9", features = ["s3-support"] }   # S3 sync, optional
 ```
 
+## 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 = "0.9", features = ["versioned-row-publication"] }
+```
+
+In this mode, 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
+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
+protocol and its scope.
+
 ## Relationship to `data_bucket`
 
 WorkTable is built on [`data_bucket`](https://crates.io/crates/data_bucket), which
@@ -396,4 +420,3 @@ enum WorkTableError
 
 Check out - [Examples](./examples)
 
-
diff --git a/benches/cases/unique_index.rs b/benches/cases/unique_index.rs
index 7eb69e7..acf75f5 100644
--- a/benches/cases/unique_index.rs
+++ b/benches/cases/unique_index.rs
@@ -1,6 +1,7 @@
 use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, black_box, criterion_group};
 use std::sync::Arc;
 use tokio::runtime::Runtime;
+use worktable::prelude::SelectQueryExecutor;
 
 use crate::common::*;
 
@@ -64,6 +65,26 @@ fn select_by_unique_index(c: &mut Criterion) {
     });
 }
 
+fn select_by_unique_index_range(c: &mut Criterion) {
+    let table = UniqueIndexWorkTable::default();
+
+    for i in 1..=1000i64 {
+        let row = UniqueIndexRow {
+            id: table.get_next_pk().into(),
+            test: i,
+            another: i as u64,
+        };
+        table.insert(row).unwrap();
+    }
+
+    c.bench_function("unique_index_select_by_test_range", |b| {
+        b.iter(|| {
+            let test = fastrand::i64(1..=1000);
+            black_box(table.select_by_test_range(test..=test).execute())
+        })
+    });
+}
+
 fn update(c: &mut Criterion) {
     let rt = Runtime::new().unwrap();
     let table = Arc::new(UniqueIndexWorkTable::default());
@@ -224,6 +245,7 @@ criterion_group! {
         insert,
         select_by_pk,
         select_by_unique_index,
+        select_by_unique_index_range,
         update,
         delete,
         upsert_insert,
diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml
index 05ace59..e6c5f78 100644
--- a/codegen/Cargo.toml
+++ b/codegen/Cargo.toml
@@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable"
 
 [features]
 s3-support = []
+versioned-row-publication = []
 
 [lib]
 name = "worktable_codegen"
diff --git a/codegen/src/generators/in_memory/queries/select.rs b/codegen/src/generators/in_memory/queries/select.rs
index a86ddea..7ba14ec 100644
--- a/codegen/src/generators/in_memory/queries/select.rs
+++ b/codegen/src/generators/in_memory/queries/select.rs
@@ -29,9 +29,13 @@ impl InMemoryGenerator {
                                                            #column_range_type,
                                                            #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let iter = self.0.primary_index.pk_map
                     .iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 SelectQueryBuilder::new(iter)
             }
diff --git a/codegen/src/generators/in_memory/table/impls.rs b/codegen/src/generators/in_memory/table/impls.rs
index b647c68..4f42537 100644
--- a/codegen/src/generators/in_memory/table/impls.rs
+++ b/codegen/src/generators/in_memory/table/impls.rs
@@ -110,9 +110,13 @@ impl InMemoryGenerator {
                     range.start_bound().map(|v| #primary_key_type::from(v.clone())),
                     range.end_bound().map(|v| #primary_key_type::from(v.clone())),
                 );
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.primary_index.pk_map
                     .range(converted_range)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 #pk_sorted_by
             }
@@ -262,6 +266,7 @@ impl InMemoryGenerator {
 
     fn gen_table_iter_inner(&self, func: TokenStream) -> TokenStream {
         quote! {
+            let _read_guard = self.0.data.read_guard();
             let first = self.0.primary_index.pk_map.iter().next().map(|(k, v)| (k.clone(), v.0));
             let Some((mut k, link)) = first else {
                 return Ok(())
diff --git a/codegen/src/generators/in_memory/table/index_fns.rs b/codegen/src/generators/in_memory/table/index_fns.rs
index 0a73cce..e82d75f 100644
--- a/codegen/src/generators/in_memory/table/index_fns.rs
+++ b/codegen/src/generators/in_memory/table/index_fns.rs
@@ -83,13 +83,39 @@ impl InMemoryGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-
-        Ok(quote! {
-            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+        let select = if cfg!(feature = "versioned-row-publication") {
+            quote! {
+                loop {
+                    let link: Link = self.0.indexes.#field_ident
+                        .get(#by)
+                        .map(|kv| kv.get().value.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
+                        .get(#by)
+                        .map(|kv| kv.get().value.into());
+                    if current_link == Some(link) {
+                        return None;
+                    }
+                }
+            }
+        } else {
+            quote! {
                 let link: Link = self.0.indexes.#field_ident.get(#by).map(|kv| kv.get().value.into())?;
                 let row = self.0.data.select_non_ghosted(link).ok()?;
                 #predicate_matches.then_some(row)
             }
+        };
+
+        Ok(quote! {
+            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+                let _read_guard = self.0.data.read_guard();
+                #select
+            }
         })
     }
 
@@ -121,10 +147,14 @@ impl InMemoryGenerator {
                                                                      #column_range_type,
                                                                      #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .get(#by)
                     .into_iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok())
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
                     .filter(move |r| &r.#row_field_ident == &by);
 
                 SelectQueryBuilder::new(rows)
@@ -143,21 +173,52 @@ impl InMemoryGenerator {
         let type_ = columns_map.get(i).ok_or(syn::Error::new(i.span(), "Row not found"))?;
         let fn_name = Ident::new(format!("select_by_{i}_range").as_str(), Span::mixed_site());
         let field_ident = &idx.name;
+        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_> },
-                quote! {
+                if revalidate {
+                    quote! {
+                        (
+                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
+                        )
+                    }
+                } else {
+                    quote! {
                     (
                         range.start_bound().map(|v| OrderedFloat(*v)),
                         range.end_bound().map(|v| OrderedFloat(*v)),
                     )
+                    }
                 },
             )
+        } else if revalidate {
+            (
+                quote! { std::ops::RangeBounds<#type_> },
+                quote! { predicate_range.clone() },
+            )
         } else {
             (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
+        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)
+                })
+            }
+        });
 
         Ok(quote! {
             pub fn #fn_name<R>(&self, range: R) -> SelectQueryBuilder<#row_ident,
@@ -167,9 +228,15 @@ impl InMemoryGenerator {
             where
                 R: #range_bounds
             {
+                #predicate_setup
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .range(#range_arg)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
+                    #predicate_filter;
 
                 SelectQueryBuilder::new_sorted(rows, #row_fields_ident::#column_pascal)
             }
diff --git a/codegen/src/generators/persist/queries/select.rs b/codegen/src/generators/persist/queries/select.rs
index 5db8e71..84501b2 100644
--- a/codegen/src/generators/persist/queries/select.rs
+++ b/codegen/src/generators/persist/queries/select.rs
@@ -29,9 +29,13 @@ impl PersistGenerator {
                                                            #column_range_type,
                                                            #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let iter = self.0.primary_index.pk_map
                     .iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 SelectQueryBuilder::new(iter)
             }
diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs
index eafdc3a..7481669 100644
--- a/codegen/src/generators/persist/table/impls.rs
+++ b/codegen/src/generators/persist/table/impls.rs
@@ -196,9 +196,13 @@ impl PersistGenerator {
                     range.start_bound().map(|v| #primary_key_type::from(v.clone())),
                     range.end_bound().map(|v| #primary_key_type::from(v.clone())),
                 );
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.primary_index.pk_map
                     .range(converted_range)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 #pk_sorted_by
             }
@@ -369,6 +373,7 @@ impl PersistGenerator {
 
     fn gen_table_iter_inner(&self, func: TokenStream) -> TokenStream {
         quote! {
+            let _read_guard = self.0.data.read_guard();
             let first = self.0.primary_index.pk_map.iter().next().map(|(k, v)| (k.clone(), v.0));
             let Some((mut k, link)) = first else {
                 return Ok(())
diff --git a/codegen/src/generators/persist/table/index_fns.rs b/codegen/src/generators/persist/table/index_fns.rs
index eb1ac14..8768de4 100644
--- a/codegen/src/generators/persist/table/index_fns.rs
+++ b/codegen/src/generators/persist/table/index_fns.rs
@@ -83,13 +83,39 @@ impl PersistGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-
-        Ok(quote! {
-            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+        let select = if cfg!(feature = "versioned-row-publication") {
+            quote! {
+                loop {
+                    let link: Link = self.0.indexes.#field_ident
+                        .get(#by)
+                        .map(|kv| kv.get().value.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
+                        .get(#by)
+                        .map(|kv| kv.get().value.into());
+                    if current_link == Some(link) {
+                        return None;
+                    }
+                }
+            }
+        } else {
+            quote! {
                 let link: Link = self.0.indexes.#field_ident.get(#by).map(|kv| kv.get().value.into())?;
                 let row = self.0.data.select_non_ghosted(link).ok()?;
                 #predicate_matches.then_some(row)
             }
+        };
+
+        Ok(quote! {
+            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+                let _read_guard = self.0.data.read_guard();
+                #select
+            }
         })
     }
 
@@ -121,10 +147,14 @@ impl PersistGenerator {
                                                                      #column_range_type,
                                                                      #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .get(#by)
                     .into_iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok())
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
                     .filter(move |r| &r.#row_field_ident == &by);
 
                 SelectQueryBuilder::new(rows)
@@ -143,21 +173,52 @@ impl PersistGenerator {
         let type_ = columns_map.get(i).ok_or(syn::Error::new(i.span(), "Row not found"))?;
         let fn_name = Ident::new(format!("select_by_{i}_range").as_str(), Span::mixed_site());
         let field_ident = &idx.name;
+        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_> },
-                quote! {
+                if revalidate {
+                    quote! {
+                        (
+                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
+                        )
+                    }
+                } else {
+                    quote! {
                     (
                         range.start_bound().map(|v| OrderedFloat(*v)),
                         range.end_bound().map(|v| OrderedFloat(*v)),
                     )
+                    }
                 },
             )
+        } else if revalidate {
+            (
+                quote! { std::ops::RangeBounds<#type_> },
+                quote! { predicate_range.clone() },
+            )
         } else {
             (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
+        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)
+                })
+            }
+        });
 
         Ok(quote! {
             pub fn #fn_name<R>(&self, range: R) -> SelectQueryBuilder<#row_ident,
@@ -167,9 +228,15 @@ impl PersistGenerator {
             where
                 R: #range_bounds
             {
+                #predicate_setup
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .range(#range_arg)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
+                    #predicate_filter;
 
                 SelectQueryBuilder::new_sorted(rows, #row_fields_ident::#column_pascal)
             }
diff --git a/codegen/src/generators/read_only/queries/select.rs b/codegen/src/generators/read_only/queries/select.rs
index 9e700ad..51dda41 100644
--- a/codegen/src/generators/read_only/queries/select.rs
+++ b/codegen/src/generators/read_only/queries/select.rs
@@ -29,9 +29,13 @@ impl ReadOnlyGenerator {
                                                            #column_range_type,
                                                            #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let iter = self.0.primary_index.pk_map
                     .iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 SelectQueryBuilder::new(iter)
             }
diff --git a/codegen/src/generators/read_only/table/impls.rs b/codegen/src/generators/read_only/table/impls.rs
index c1ffd3a..0d0cb5a 100644
--- a/codegen/src/generators/read_only/table/impls.rs
+++ b/codegen/src/generators/read_only/table/impls.rs
@@ -192,9 +192,13 @@ impl ReadOnlyGenerator {
                     range.start_bound().map(|v| #primary_key_type::from(v.clone())),
                     range.end_bound().map(|v| #primary_key_type::from(v.clone())),
                 );
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.primary_index.pk_map
                     .range(converted_range)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    });
 
                 #pk_sorted_by
             }
@@ -277,6 +281,7 @@ impl ReadOnlyGenerator {
 
     fn gen_table_iter_inner(&self, func: TokenStream) -> TokenStream {
         quote! {
+            let _read_guard = self.0.data.read_guard();
             let first = self.0.primary_index.pk_map.iter().next().map(|(k, v)| (k.clone(), v.0));
             let Some((mut k, link)) = first else {
                 return Ok(())
diff --git a/codegen/src/generators/read_only/table/index_fns.rs b/codegen/src/generators/read_only/table/index_fns.rs
index f626ad7..82738c7 100644
--- a/codegen/src/generators/read_only/table/index_fns.rs
+++ b/codegen/src/generators/read_only/table/index_fns.rs
@@ -83,13 +83,39 @@ impl ReadOnlyGenerator {
                 row.#row_field_ident.eq(&by)
             }
         };
-
-        Ok(quote! {
-            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+        let select = if cfg!(feature = "versioned-row-publication") {
+            quote! {
+                loop {
+                    let link: Link = self.0.indexes.#field_ident
+                        .get(#by)
+                        .map(|kv| kv.get().value.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
+                        .get(#by)
+                        .map(|kv| kv.get().value.into());
+                    if current_link == Some(link) {
+                        return None;
+                    }
+                }
+            }
+        } else {
+            quote! {
                 let link: Link = self.0.indexes.#field_ident.get(#by).map(|kv| kv.get().value.into())?;
                 let row = self.0.data.select_non_ghosted(link).ok()?;
                 #predicate_matches.then_some(row)
             }
+        };
+
+        Ok(quote! {
+            pub fn #fn_name(&self, by: #type_) -> Option<#row_ident> {
+                let _read_guard = self.0.data.read_guard();
+                #select
+            }
         })
     }
 
@@ -121,10 +147,14 @@ impl ReadOnlyGenerator {
                                                                      #column_range_type,
                                                                      #row_fields_ident>
             {
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .get(#by)
                     .into_iter()
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok())
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
                     .filter(move |r| &r.#row_field_ident == &by);
 
                 SelectQueryBuilder::new(rows)
@@ -143,21 +173,52 @@ impl ReadOnlyGenerator {
         let type_ = columns_map.get(i).ok_or(syn::Error::new(i.span(), "Row not found"))?;
         let fn_name = Ident::new(format!("select_by_{i}_range").as_str(), Span::mixed_site());
         let field_ident = &idx.name;
+        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_> },
-                quote! {
+                if revalidate {
+                    quote! {
+                        (
+                            predicate_range.0.as_ref().map(|v| OrderedFloat(*v)),
+                            predicate_range.1.as_ref().map(|v| OrderedFloat(*v)),
+                        )
+                    }
+                } else {
+                    quote! {
                     (
                         range.start_bound().map(|v| OrderedFloat(*v)),
                         range.end_bound().map(|v| OrderedFloat(*v)),
                     )
+                    }
                 },
             )
+        } else if revalidate {
+            (
+                quote! { std::ops::RangeBounds<#type_> },
+                quote! { predicate_range.clone() },
+            )
         } else {
             (quote! { std::ops::RangeBounds<#type_> }, quote! { range })
         };
+        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)
+                })
+            }
+        });
 
         Ok(quote! {
             pub fn #fn_name<R>(&self, range: R) -> SelectQueryBuilder<#row_ident,
@@ -167,9 +228,15 @@ impl ReadOnlyGenerator {
             where
                 R: #range_bounds
             {
+                #predicate_setup
+                let read_guard = self.0.data.read_guard();
                 let rows = self.0.indexes.#field_ident
                     .range(#range_arg)
-                    .filter_map(|(_, link)| self.0.data.select_non_ghosted(link.0).ok());
+                    .filter_map(move |(_, link)| {
+                        let _read_guard = &read_guard;
+                        self.0.data.select_non_ghosted(link.0).ok()
+                    })
+                    #predicate_filter;
 
                 SelectQueryBuilder::new_sorted(rows, #row_fields_ident::#column_pascal)
             }
diff --git a/docs/versioned-row-publication.md b/docs/versioned-row-publication.md
new file mode 100644
index 0000000..fbcc289
--- /dev/null
+++ b/docs/versioned-row-publication.md
@@ -0,0 +1,86 @@
+# Versioned row publication
+
+Status: feature-gated prototype behind `versioned-row-publication`.
+
+## Problem
+
+The original fast path deserializes directly from archived bytes in an
+`UnsafeCell` page while same-size updates mutate those bytes in place. Returning
+an owned row prevents references from escaping, but it does not make a read
+that overlaps a write data-race-free: the deserializer can still access bytes
+while a writer changes them. A sequence counter cannot repair this in Rust,
+because detecting a race after it happened does not make the racy byte access
+defined.
+
+Physical links introduce a second issue. A reader can obtain a link from an
+index, pause, and resume after delete or vacuum has reused that address for a
+different row. Predicate revalidation helps, but reclamation must also cover
+the interval from index lookup through acquisition of a stable row version.
+
+## Protocol
+
+With the feature enabled, `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
+  barrier.
+- A concurrent link map holds an immutable application-visible row version.
+  Each slot contains an `Arc<Row>` behind a short per-slot pointer lock and
+  atomic ghost, deleted, and vacuum lifecycle bits.
+
+The generated API follows these publication rules:
+
+1. **Read.** Acquire a read-grace guard before consulting an index. Resolve the
+   link, acquire an `Arc` to its complete published version, check lifecycle and
+   index predicates, clone the owned row, and release the guard. Unique and
+   primary-key point reads retry when the mapping swings to a replacement link
+   while it is being resolved. A reader never accesses mutable archived bytes
+   after a slot has been hydrated.
+2. **Insert.** Serialize the complete row and stage a ghosted version. Install
+   the primary and secondary indexes. Only after every checked index insert
+   succeeds does the lifecycle transition publish the version with release
+   ordering. Failed inserts retire an unpublished version.
+3. **Update.** Hold the generated row/field lock, mutate the archived image
+   under the page barrier, deserialize the completed wrapper, then replace the
+   immutable version. A concurrent reader can return the complete old version
+   or the complete new version, never a partially updated row.
+4. **Delete.** Remove index reachability, mark the version deleted, and retire
+   its physical link. The empty-link allocator cannot reuse it while a reader
+   that could have captured the old index entry remains active.
+5. **Vacuum.** Copy a complete row to a staged destination version, swing its
+   indexes, and retire the source publication and page. Retired links, slots,
+   and pages become reusable only after a read-side grace period.
+6. **Reload.** Persisted tables hydrate immutable slots lazily under the page
+   barrier. Subsequent generated reads use the published version map.
+
+The grace period is quiescent-state reclamation: a feature-only 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.
+
+## Guarantees and non-guarantees
+
+For generated table APIs in this mode:
+
+- reads do not race with mutation of archived page bytes;
+- a read returns a complete row version;
+- ghosted or deleted versions are not returned;
+- a retired physical link is not reused while a pre-existing generated read
+  can still resolve it; and
+- unique, non-unique point, and secondary-index range lookups revalidate each
+  resolved row predicate.
+
+This is not MVCC and does not add multi-operation transactions or snapshot
+range scans. A scan may include or omit a concurrently inserted or updated row.
+Point-read retry may starve under perpetual replacement churn.
+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. It adds one owned row copy plus slot/map
+metadata per live physical link, an atomic increment/decrement per generated
+read, a concurrent publication-map lookup, and writer-side page serialization.
+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.
diff --git a/src/in_memory/mod.rs b/src/in_memory/mod.rs
index 2b02cf0..0da4038 100644
--- a/src/in_memory/mod.rs
+++ b/src/in_memory/mod.rs
@@ -1,9 +1,11 @@
 mod data;
 mod empty_link_registry;
 mod pages;
+#[cfg(feature = "versioned-row-publication")]
+mod publication;
 mod row;
 
 pub use data::{DATA_INNER_LENGTH, Data, ExecutionError as DataExecutionError};
 pub use empty_link_registry::EmptyLinkRegistry;
-pub use pages::{DataPages, ExecutionError as PagesExecutionError};
-pub use row::{ArchivedRowWrapper, Query, RowWrapper, StorableRow};
+pub use pages::{DataPages, ExecutionError as PagesExecutionError, ReadGuard as DataPagesReadGuard};
+pub use row::{ArchivedRowWrapper, PublicationSafe, Query, RowWrapper, StorableRow};
diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs
index 08a41fb..475a3ba 100644
--- a/src/in_memory/pages.rs
+++ b/src/in_memory/pages.rs
@@ -1,5 +1,7 @@
 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")]
 use performance_measurement_codegen::performance_measurement;
@@ -10,7 +12,12 @@ 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;
 use std::{
     fmt::Debug,
     sync::Arc,
@@ -18,7 +25,11 @@ 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::{
         DATA_INNER_LENGTH, Data, DataExecutionError,
@@ -31,11 +42,78 @@ fn page_id_mapper(page_id: usize) -> usize {
     page_id - 1usize
 }
 
+/// `OffsetEqLink` already reduces publication keys to a trusted internal u64
+/// storage offset. Avoid hashing that offset again on every versioned read.
+#[cfg(feature = "versioned-row-publication")]
+#[derive(Default)]
+struct PublicationHasher(u64);
+
+#[cfg(feature = "versioned-row-publication")]
+impl Hasher for PublicationHasher {
+    fn finish(&self) -> u64 {
+        self.0
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        let mut hash = 0xcbf29ce484222325u64;
+        for byte in bytes {
+            hash ^= u64::from(*byte);
+            hash = hash.wrapping_mul(0x100000001b3);
+        }
+        self.0 = hash;
+    }
+
+    fn write_u64(&mut self, value: u64) {
+        self.0 = value;
+    }
+}
+
+#[cfg(feature = "versioned-row-publication")]
+type PublicationMap<Row, const DATA_LENGTH: usize> =
+    HashMap<OffsetEqLink<DATA_LENGTH>, Arc<PublishedRow<Row>>, BuildHasherDefault<PublicationHasher>>;
+
+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);
+    }
+}
+
 #[derive(Debug)]
 pub struct DataPages<Row, const DATA_LENGTH: usize = DATA_INNER_LENGTH>
 where
     Row: StorableRow,
 {
+    /// Immutable application-visible row versions. Published readers never
+    /// borrow the mutable archived page image.
+    #[cfg(feature = "versioned-row-publication")]
+    published_rows: RwLock<PublicationMap<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>>>,
+
     /// Pages vector. Currently, not lock free.
     pages: RwLock<Vec<Arc<Data<<Row as StorableRow>::WrappedRow, DATA_LENGTH>>>>,
 
@@ -66,8 +144,126 @@ 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() {
+            flags |= GHOSTED;
+        }
+        if row.is_deleted() {
+            flags |= DELETED;
+        }
+        if row.is_vacuumed() {
+            flags |= VACUUMED;
+        }
+        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();
+        let key = OffsetEqLink(link);
+
+        let mut published_rows = self.published_rows.write();
+        if let Some(slot) = published_rows.get(&key).cloned() {
+            drop(published_rows);
+            slot.replace(row, flags);
+        } else {
+            published_rows.insert(key, Arc::new(PublishedRow::new(row, flags)));
+        }
+    }
+
+    #[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>>> {
+        self.published_rows.read().get(&OffsetEqLink(link)).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:
+            Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
+    {
+        if let Some(slot) = self.published_slot(link) {
+            return Ok(slot);
+        }
+
+        let _page_access = self.page_access.read();
+        if let Some(slot) = self.published_slot(link) {
+            return Ok(slot);
+        }
+
+        let pages = self.pages.read();
+        let page = pages
+            .get(page_id_mapper(link.page_id.into()))
+            .ok_or(ExecutionError::PageNotFound(link.page_id))?;
+        let wrapped = page.get_row(link).map_err(ExecutionError::DataPageError)?;
+        let flags = Self::publication_flags(&wrapped);
+        let slot = Arc::new(PublishedRow::new(wrapped.get_inner(), flags));
+        let mut published_rows = self.published_rows.write();
+        Ok(published_rows.entry(OffsetEqLink(link)).or_insert(slot).clone())
+    }
+
+    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.active_readers.load(Ordering::SeqCst) != 0 {
+            return;
+        }
+
+        let mut retired_links = self.retired_links.lock();
+        let mut retired_pages = self.retired_pages.lock();
+        let mut retired_publications = self.retired_publications.lock();
+        if self.active_readers.load(Ordering::SeqCst) != 0 {
+            return;
+        }
+
+        let mut published_rows = self.published_rows.write();
+        for link in retired_links.drain(..) {
+            published_rows.remove(&OffsetEqLink(link));
+            self.empty_links.push(link);
+        }
+        for link in retired_publications.drain(..) {
+            published_rows.remove(&link);
+        }
+        drop(published_rows);
+        if !retired_pages.is_empty() {
+            let mut empty_pages = self.empty_pages.write();
+            empty_pages.extend(retired_pages.drain(..));
+        }
+    }
+
     pub fn new() -> Self {
         Self {
+            #[cfg(feature = "versioned-row-publication")]
+            published_rows: 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()),
             // 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()))]),
             empty_links: EmptyLinkRegistry::<DATA_LENGTH>::default(),
@@ -85,6 +281,18 @@ where
         } else {
             let last_page_id = vec.len();
             Self {
+                #[cfg(feature = "versioned-row-publication")]
+                published_rows: 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()),
                 pages: RwLock::new(vec),
                 empty_links: EmptyLinkRegistry::default(),
                 empty_pages: Default::default(),
@@ -97,13 +305,20 @@ where
 
     pub fn insert(&self, row: Row) -> Result<Link, ExecutionError>
     where
-        Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+        Row: Archive
+            + Clone
+            + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
     {
-        let general_row = <Row as StorableRow>::WrappedRow::from_inner(row);
+        let general_row = <Row as StorableRow>::WrappedRow::from_inner(row.clone());
+
+        #[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());
             let page = &pages[current_page];
@@ -113,6 +328,8 @@ where
                     if let Some(l) = left_link {
                         self.empty_links.push(l);
                     }
+                    #[cfg(feature = "versioned-row-publication")]
+                    self.stage_published_row(link, row.clone());
                     return Ok(link);
                 }
                 Err(e) => match e {
@@ -129,11 +346,19 @@ 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);
                 let page = &pages[current_page];
 
-                (page.save_row(&general_row), current_page)
+                let link = page.save_row(&general_row);
+                #[cfg(feature = "versioned-row-publication")]
+                if let Ok(saved_link) = &link {
+                    self.stage_published_row(*saved_link, row.clone());
+                }
+
+                (link, current_page)
             };
             match link {
                 Ok(link) => {
@@ -191,12 +416,17 @@ 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 = {
             let mut empty_pages = self.empty_pages.write();
             empty_pages.pop_front()
         };
 
         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());
             let page = pages[index].clone();
@@ -216,54 +446,104 @@ where
     #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "DataPages"))]
     pub fn select<L: Into<Link>>(&self, link: L) -> Result<Row, ExecutionError>
     where
-        Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+        Row: Archive
+            + Clone
+            + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
         let link = link.into();
-        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())
+        #[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())
+        }
     }
 
     pub fn select_non_ghosted(&self, link: Link) -> Result<Row, ExecutionError>
     where
-        Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+        Row: Archive
+            + Clone
+            + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        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);
+        #[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())
+        }
+
+        #[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())
         }
-        Ok(gen_row.get_inner())
     }
 
     pub fn select_non_vacuumed(&self, link: Link) -> Result<Row, ExecutionError>
     where
-        Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+        Row: Archive
+            + Clone
+            + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Portable + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        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);
+        #[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())
         }
-        if gen_row.is_vacuumed() {
-            return Err(ExecutionError::Vacuumed);
+
+        #[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())
         }
-        Ok(gen_row.get_inner())
     }
 
     #[cfg_attr(feature = "perf_measurements", performance_measurement(prefix_name = "DataPages"))]
@@ -272,6 +552,8 @@ 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
             .get::<usize>(page_id_mapper(link.page_id.into()))
@@ -287,18 +569,31 @@ where
     where
         Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <<Row as StorableRow>::WrappedRow as Archive>::Archived: Portable,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived:
+            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
             .get(page_id_mapper(link.page_id.into()))
             .ok_or(ExecutionError::PageNotFound(link.page_id))?;
-        let gen_row = unsafe {
-            page.get_mut_row_ref(link)
-                .map_err(ExecutionError::DataPageError)?
-                .unseal_unchecked()
+        let res = {
+            let gen_row = unsafe {
+                page.get_mut_row_ref(link)
+                    .map_err(ExecutionError::DataPageError)?
+                    .unseal_unchecked()
+            };
+            op(gen_row)
         };
-        let res = op(gen_row);
+
+        #[cfg(feature = "versioned-row-publication")]
+        {
+            let wrapped = page.get_row(link).map_err(ExecutionError::DataPageError)?;
+            self.publish_wrapped_row(link, wrapped);
+        }
+
         Ok(res)
     }
 
@@ -310,19 +605,24 @@ where
     /// - The operation does not cause data races or memory corruption.
     pub unsafe fn update<const N: usize>(&self, row: Row, link: Link) -> Result<Link, ExecutionError>
     where
-        Row: Archive,
+        Row: Archive + Clone,
         <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))?;
-        let gen_row = <Row as StorableRow>::WrappedRow::from_inner(row);
-        unsafe {
+        let gen_row = <Row as StorableRow>::WrappedRow::from_inner(row.clone());
+        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)
     }
 
     pub fn delete(&self, link: Link) -> Result<(), ExecutionError>
@@ -330,15 +630,26 @@ where
         Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
-        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + Portable
+            + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
         unsafe { self.with_mut_ref(link, |r| r.delete())? }
 
+        #[cfg(feature = "versioned-row-publication")]
+        {
+            self.retired_links.lock().push(link);
+            self.reclaim_retired();
+        }
+
+        #[cfg(not(feature = "versioned-row-publication"))]
         self.empty_links.push(link);
         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
             .get(page_id_mapper(link.page_id.into()))
@@ -348,7 +659,15 @@ 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")]
+            {
+                self.retired_pages.lock().push(page_id);
+                self.reclaim_retired();
+            }
+
+            #[cfg(not(feature = "versioned-row-publication"))]
             let mut g = self.empty_pages.write();
+            #[cfg(not(feature = "versioned-row-publication"))]
             g.push_back(page_id);
         }
     }
@@ -395,6 +714,8 @@ 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
             .iter()
@@ -402,6 +723,75 @@ where
             .collect()
     }
 
+    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
+            .get(page_id_mapper(page_id.into()))
+            .ok_or(ExecutionError::PageNotFound(page_id))?;
+        page.reset();
+
+        Ok(())
+    }
+
+    /// Copies a row to another page without exposing either mutable byte
+    /// image to application readers.
+    pub(crate) unsafe fn move_row_for_vacuum(
+        &self,
+        from_link: Link,
+        to_page_id: PageId,
+    ) -> Result<(Vec<u8>, Link), ExecutionError>
+    where
+        Row: Clone,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + 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
+            .get(page_id_mapper(from_link.page_id.into()))
+            .ok_or(ExecutionError::PageNotFound(from_link.page_id))?;
+        let to_page = pages
+            .get(page_id_mapper(to_page_id.into()))
+            .ok_or(ExecutionError::PageNotFound(to_page_id))?;
+
+        let raw_data = from_page
+            .get_raw_row(from_link)
+            .map_err(ExecutionError::DataPageError)?;
+        let archived = unsafe {
+            from_page
+                .get_mut_row_ref(from_link)
+                .map_err(ExecutionError::DataPageError)?
+                .unseal_unchecked()
+        };
+        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);
+            let new_wrapped = to_page.get_row(new_link).map_err(ExecutionError::DataPageError)?;
+            self.publish_wrapped_row(new_link, new_wrapped);
+        }
+
+        Ok((raw_data, new_link))
+    }
+
+    pub(crate) fn retire_published_link(&self, link: Link) {
+        #[cfg(feature = "versioned-row-publication")]
+        {
+            self.retired_publications.lock().push(OffsetEqLink(link));
+            self.reclaim_retired();
+        }
+
+        #[cfg(not(feature = "versioned-row-publication"))]
+        let _ = link;
+    }
+
     pub fn get_page_count(&self) -> usize {
         self.pages.read().len()
     }
@@ -466,7 +856,11 @@ 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;
 
     use parking_lot::RwLock;
@@ -603,6 +997,104 @@ 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();
+        let row = TestRow { a: 7, b: 9 };
+        let link = pages.insert(row).unwrap();
+
+        assert_eq!(pages.select_non_ghosted(link), Err(ExecutionError::Ghosted));
+        unsafe {
+            pages.with_mut_ref(link, |archived| archived.unghost()).unwrap();
+        }
+        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());
+        let link = pages.insert(TestRow { a: 0, b: 0 }).unwrap();
+        unsafe {
+            pages.with_mut_ref(link, |row| row.unghost()).unwrap();
+        }
+
+        let (first_field_written_tx, first_field_written_rx) = mpsc::channel();
+        let (finish_update_tx, finish_update_rx) = mpsc::channel();
+        let writer_pages = pages.clone();
+        let writer = thread::spawn(move || unsafe {
+            writer_pages
+                .with_mut_ref(link, |archived| {
+                    archived.inner.a = 1.into();
+                    first_field_written_tx.send(()).unwrap();
+                    finish_update_rx.recv().unwrap();
+                    archived.inner.b = 1.into();
+                })
+                .unwrap();
+        });
+
+        first_field_written_rx.recv().unwrap();
+        let (read_tx, read_rx) = mpsc::channel();
+        let reader_pages = pages.clone();
+        let reader = thread::spawn(move || {
+            read_tx.send(reader_pages.select_non_ghosted(link)).unwrap();
+        });
+
+        assert_eq!(
+            read_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
+            Ok(TestRow { a: 0, b: 0 }),
+            "reader must use the old immutable version instead of page bytes"
+        );
+
+        finish_update_tx.send(()).unwrap();
+        writer.join().unwrap();
+        reader.join().unwrap();
+        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();
+        let link = pages.insert(TestRow { a: 1, b: 1 }).unwrap();
+        unsafe {
+            pages.with_mut_ref(link, |row| row.unghost()).unwrap();
+        }
+        let old_slot = pages.published_slot(link).unwrap();
+        let old_version = old_slot.snapshot();
+
+        pages.delete(link).unwrap();
+        let reused_link = pages.insert(TestRow { a: 2, b: 2 }).unwrap();
+        assert_eq!(reused_link, link);
+        assert_eq!(pages.select_non_ghosted(reused_link), Err(ExecutionError::Ghosted));
+        unsafe {
+            pages.with_mut_ref(reused_link, |row| row.unghost()).unwrap();
+        }
+
+        assert_eq!(old_version.as_ref(), &TestRow { a: 1, b: 1 });
+        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();
+        let old_link = pages.insert(TestRow { a: 1, b: 1 }).unwrap();
+        unsafe {
+            pages.with_mut_ref(old_link, |row| row.unghost()).unwrap();
+        }
+
+        let read_guard = pages.read_guard();
+        pages.delete(old_link).unwrap();
+        let new_link = pages.insert(TestRow { a: 2, b: 2 }).unwrap();
+        assert_ne!(new_link, old_link, "retired link was reused by an active reader");
+
+        drop(read_guard);
+        pages.reclaim_retired();
+        assert!(pages.get_empty_links().contains(&old_link));
+    }
+
     #[test]
     fn select_non_vacuumed_returns_row_when_valid() {
         let pages = DataPages::<TestRow>::new();
diff --git a/src/in_memory/publication.rs b/src/in_memory/publication.rs
new file mode 100644
index 0000000..a0f0907
--- /dev/null
+++ b/src/in_memory/publication.rs
@@ -0,0 +1,51 @@
+use std::fmt::{Debug, Formatter};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU8, Ordering};
+
+use parking_lot::RwLock;
+
+pub(super) const GHOSTED: u8 = 1 << 0;
+pub(super) const DELETED: u8 = 1 << 1;
+pub(super) const VACUUMED: u8 = 1 << 2;
+
+/// One immutable application-visible row version plus atomic lifecycle bits.
+///
+/// Readers hold an `Arc` to a complete version, so replacing or retiring a
+/// version cannot invalidate an in-flight read. The short per-row lock only
+/// protects the `Arc` pointer; readers never access mutable archived bytes.
+pub(super) struct PublishedRow<Row> {
+    row: RwLock<Arc<Row>>,
+    flags: AtomicU8,
+}
+
+impl<Row> PublishedRow<Row> {
+    pub(super) fn new(row: Row, flags: u8) -> Self {
+        Self {
+            row: RwLock::new(Arc::new(row)),
+            flags: AtomicU8::new(flags),
+        }
+    }
+
+    pub(super) fn replace(&self, row: Row, flags: u8) {
+        *self.row.write() = Arc::new(row);
+        self.flags.store(flags, Ordering::Release);
+    }
+
+    pub(super) fn load(&self) -> (Arc<Row>, u8) {
+        let flags = self.flags.load(Ordering::Acquire);
+        let row = self.row.read().clone();
+        (row, flags)
+    }
+
+    pub(super) fn snapshot(&self) -> Arc<Row> {
+        self.row.read().clone()
+    }
+}
+
+impl<Row> Debug for PublishedRow<Row> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PublishedRow")
+            .field("flags", &self.flags.load(Ordering::Relaxed))
+            .finish_non_exhaustive()
+    }
+}
diff --git a/src/in_memory/row.rs b/src/in_memory/row.rs
index 2d75a8b..115b9b7 100644
--- a/src/in_memory/row.rs
+++ b/src/in_memory/row.rs
@@ -2,10 +2,22 @@ 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
-pub trait StorableRow {
+pub trait StorableRow: PublicationSafe {
     type WrappedRow: Archive + Debug;
 }
 
diff --git a/src/table/mod.rs b/src/table/mod.rs
index cb54aaa..298b403 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -23,7 +23,7 @@ use rkyv::ser::Serializer;
 use rkyv::ser::allocator::ArenaHandle;
 use rkyv::ser::sharing::Share;
 use rkyv::util::AlignedVec;
-use rkyv::{Archive, Deserialize, Serialize};
+use rkyv::{Archive, Deserialize, Portable, Serialize};
 use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -152,11 +152,30 @@ where
         <<Row as StorableRow>::WrappedRow as Archive>::Archived:
             Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
     {
-        let link: Option<Link> = self.primary_index.pk_map.get(&pk).map(|v| v.get().value.into());
-        if let Some(link) = link {
-            self.data.select_non_ghosted(link).ok()
-        } else {
-            None
+        let _read_guard = self.data.read_guard();
+        #[cfg(feature = "versioned-row-publication")]
+        {
+            loop {
+                let link: Link = self.primary_index.pk_map.get(&pk).map(|v| v.get().value.into())?;
+                if let Ok(row) = self.data.select_non_ghosted(link) {
+                    return Some(row);
+                }
+
+                let current_link: Option<Link> = self.primary_index.pk_map.get(&pk).map(|v| v.get().value.into());
+                if current_link == Some(link) {
+                    return None;
+                }
+            }
+        }
+
+        #[cfg(not(feature = "versioned-row-publication"))]
+        {
+            let link: Option<Link> = self.primary_index.pk_map.get(&pk).map(|v| v.get().value.into());
+            if let Some(link) = link {
+                self.data.select_non_ghosted(link).ok()
+            } else {
+                None
+            }
         }
     }
 
@@ -168,7 +187,9 @@ where
             + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
-        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + Portable
+            + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
         PrimaryKey: Clone,
         AvailableTypes: 'static,
         AvailableIndexes: AvailableIndex,
@@ -184,9 +205,9 @@ where
         if let Err(e) = self.indexes.save_row(row.clone(), link) {
             return match e {
                 IndexError::AlreadyExists { at, inserted_already } => {
-                    self.data.delete(link).map_err(WorkTableError::PagesError)?;
                     self.primary_index.remove(&pk, link);
                     self.indexes.delete_from_indexes(row, link, inserted_already)?;
+                    self.data.delete(link).map_err(WorkTableError::PagesError)?;
 
                     Err(WorkTableError::AlreadyExists(at.to_string_value()))
                 }
@@ -216,7 +237,9 @@ where
             + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
-        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + Portable
+            + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
         PrimaryKey: Clone,
         SecondaryEvents: Debug + Default + Clone + TableSecondaryIndexEventsOps<AvailableIndexes>,
         SecondaryIndexes: TableSecondaryIndex<Row, AvailableTypes, AvailableIndexes>
@@ -334,7 +357,9 @@ where
             + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
-        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + Portable
+            + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
         PrimaryKey: Clone,
         AvailableTypes: 'static,
         AvailableIndexes: Debug + AvailableIndex,
@@ -391,7 +416,9 @@ where
             + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
-        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper,
+        <<Row as StorableRow>::WrappedRow as Archive>::Archived: ArchivedRowWrapper
+            + Portable
+            + Deserialize<<Row as StorableRow>::WrappedRow, HighDeserializer<rkyv::rancor::Error>>,
         PrimaryKey: Clone,
         SecondaryEvents: Debug + Default + Clone + TableSecondaryIndexEventsOps<AvailableIndexes>,
         SecondaryIndexes: TableSecondaryIndex<Row, AvailableTypes, AvailableIndexes>
diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs
index 1cb3d8b..6fec037 100644
--- a/src/table/vacuum/vacuum.rs
+++ b/src/table/vacuum/vacuum.rs
@@ -207,13 +207,11 @@ where
     }
 
     fn free_page(&self, page_id: PageId) {
-        let p = self.data_pages.get_page(page_id).expect("should exist as called");
-        p.reset()
+        self.data_pages.reset_page(page_id).expect("should exist as called")
     }
 
     async fn move_data_from(&self, from: PageId, to: PageId) -> (bool, bool) {
         let to_page = self.data_pages.get_page(to).expect("should exist as link exists");
-        let from_page = self.data_pages.get_page(from).expect("should exist as link exists");
         let to_free_space = to_page.free_space();
 
         let page_start = OffsetEqLink::<_>(Link {
@@ -268,18 +266,13 @@ where
                 self.lock_manager.remove_with_lock_check(&pk);
                 continue;
             }
-            let raw_data = from_page
-                .get_raw_row(from_link.0)
-                .expect("link is not bigger than free offset");
-            unsafe {
+            let (raw_data, new_link) = unsafe {
                 self.data_pages
-                    .with_mut_ref(from_link.0, |r| r.set_in_vacuum_process())
-                    .expect("link should be valid")
-            }
-            let new_link = to_page
-                .save_raw_row(&raw_data)
-                .expect("page is not full as checked on links collection");
+                    .move_row_for_vacuum(from_link.0, to)
+                    .expect("links and destination capacity were checked")
+            };
             self.update_index_after_move(pk.clone(), from_link.0, new_link, raw_data);
+            self.data_pages.retire_published_link(from_link.0);
 
             lock.unlock();
             self.lock_manager.remove_with_lock_check(&pk);
diff --git a/tests/worktable/float.rs b/tests/worktable/float.rs
index 29747ee..576c707 100644
--- a/tests/worktable/float.rs
+++ b/tests/worktable/float.rs
@@ -54,6 +54,38 @@ 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();
+    let inside = TestFloatRow {
+        id: table.get_next_pk().into(),
+        test: 1,
+        another: 10.0,
+        exchange: "inside".to_string(),
+    };
+    let outside = TestFloatRow {
+        id: table.get_next_pk().into(),
+        test: 2,
+        another: 100.0,
+        exchange: "outside".to_string(),
+    };
+    table.insert(inside.clone()).unwrap();
+    table.insert(outside.clone()).unwrap();
+
+    let outside_link = table
+        .0
+        .primary_index
+        .pk_map
+        .get(&TestFloatPrimaryKey(outside.id))
+        .map(|entry| entry.get().value.0)
+        .unwrap();
+    TableIndex::insert(&table.0.indexes.another_idx, OrderedFloat(15.0), outside_link);
+
+    let rows = table.select_by_another_range(0.0..20.0).execute().unwrap();
+    assert_eq!(rows, vec![inside]);
+}
+
 #[test]
 fn select_all_range_float_test() {
     let table = TestFloatWorkTable::default();
diff --git a/tests/worktable/index/range.rs b/tests/worktable/index/range.rs
index 99591ce..39f98dd 100644
--- a/tests/worktable/index/range.rs
+++ b/tests/worktable/index/range.rs
@@ -32,6 +32,39 @@ worktable!(
     }
 );
 
+#[cfg(feature = "versioned-row-publication")]
+#[test]
+fn range_read_revalidates_each_resolved_row() {
+    let table = RangeTestWorkTable::default();
+    let inside = RangeTestRow {
+        id: table.get_next_pk().into(),
+        value: 10,
+        name: "inside".to_string(),
+    };
+    let outside = RangeTestRow {
+        id: table.get_next_pk().into(),
+        value: 100,
+        name: "outside".to_string(),
+    };
+    table.insert(inside.clone()).unwrap();
+    table.insert(outside.clone()).unwrap();
+
+    let outside_link = table
+        .0
+        .primary_index
+        .pk_map
+        .get(&RangeTestPrimaryKey(outside.id))
+        .map(|entry| entry.get().value.0)
+        .unwrap();
+
+    // Model the transient state where an index entry still falls inside the
+    // requested range but its completed row version has already moved out.
+    TableIndex::insert(&table.0.indexes.val_idx, 15, outside_link);
+
+    let rows = table.select_by_value_range(0..20).execute().unwrap();
+    assert_eq!(rows, vec![inside]);
+}
+
 #[test]
 fn test_range_select_basic() {
     let table = RangeTestWorkTable::default();