perf: keep fixed-width updates in place

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

@@ -75,7 +75,6 @@ impl InMemoryGenerator {
                         .unseal_unchecked()
                 };
 
-                let op_id = OperationId::Single(uuid::Uuid::now_v7());
                 #diff_process_insert
                 #persist_op
 
@@ -385,14 +384,9 @@ impl InMemoryGenerator {
                     return core::result::Result::Ok(());
                 }
             }
-        } else if self.columns.is_sized {
-            // Sized rows never change length; the caller's field swap is safe and
-            // no size check / reinsert is needed.
-            quote! {}
-        } else {
-            // Unsized rows where an updated column IS indexed: keep the original
-            // always-reinsert path so secondary-index maintenance and unique
-            // checks run through reinsert.
+        } else if touches_index {
+            // Updating an indexed column must keep the index-maintaining
+            // reinsert path, regardless of the row's storage shape.
             let row_updates = idents
                 .iter()
                 .map(|i| quote! { row_new.#i = row.#i.clone(); })
@@ -420,6 +414,11 @@ impl InMemoryGenerator {
                     return core::result::Result::Ok(());
                 }
             }
+        } else {
+            // Other columns make the row unsized, but this query updates only
+            // fixed-width, unindexed fields. The caller can safely swap those
+            // archived fields in place without rebuilding the full row.
+            quote! {}
         }
     }
 
@@ -605,14 +604,15 @@ impl InMemoryGenerator {
             })
             .collect::<Vec<_>>();
 
+        let archived_swap_is_safe = self.columns.is_sized || (unsized_fields.is_none() && idx_idents.is_none());
         let size_check = self.gen_size_check(unsized_fields, idents, idx_idents);
         let diff_process_insert = self.gen_process_diffs_insert_on_index(idents, idx_idents);
         let diff_process_remove = self.gen_process_diffs_remove_on_index(idx_idents);
         let persist_call = self.gen_persist_call();
         let persist_op = self.gen_persist_op();
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
 
-        let finish_update = if self.columns.is_sized {
+        let finish_update = if archived_swap_is_safe {
             quote! {
                 #diff_process_insert
                 #persist_op
@@ -653,7 +653,6 @@ impl InMemoryGenerator {
                 let mut bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&row).map_err(|_| WorkTableError::SerializeError)?;
                 let mut archived_row = unsafe { rkyv::access_unchecked_mut::<<#query_ident as rkyv::Archive>::Archived>(&mut bytes[..]).unseal_unchecked() };
 
-                let op_id = OperationId::Single(uuid::Uuid::now_v7());
                 #size_check
                 #finish_update
             }
@@ -784,7 +783,6 @@ impl InMemoryGenerator {
                     guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk));
                 }
 
-                let op_id = OperationId::Multi(uuid::Uuid::now_v7());
                 for pk in pks.into_iter() {
                     // Re-resolve and re-validate under the held lock. The
                     // query's lock set includes the predicate column, so the
@@ -931,7 +929,6 @@ impl InMemoryGenerator {
                     }
                 };
 
-                let op_id = OperationId::Single(uuid::Uuid::now_v7());
                 #size_check
                 #finish_update
             }

codegen/src/worktable/mod.rs

@@ -218,6 +218,43 @@ mod tests {
         }
     }
 
+    #[test]
+    fn fixed_width_update_on_unsized_table_uses_archived_swap() {
+        let output = expand(quote! {
+            name: MixedWidthUpdate,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                payload: String,
+                balance: f64,
+            },
+            queries: {
+                update: {
+                    Balance(balance) by id,
+                }
+            }
+        })
+        .unwrap()
+        .to_string();
+
+        let update = output
+            .split("pub async fn update_balance")
+            .nth(1)
+            .expect("generated balance update");
+        assert!(
+            update.contains("data . with_mut_ref"),
+            "fixed-width unindexed field must update archived storage in place"
+        );
+        assert!(
+            !update.contains("self . reinsert"),
+            "an unrelated String column must not force a fixed-width update through reinsert"
+        );
+        assert!(
+            !update.contains("Uuid :: now_v7"),
+            "non-persistent updates must not generate an unused operation id"
+        );
+    }
+
     #[cfg(feature = "logical-index-persistence")]
     #[test]
     fn logical_persistence_wraps_only_default_wti_backends() {

src/table/vacuum/vacuum.rs

@@ -14,7 +14,7 @@ use rkyv::util::AlignedVec;
 use rkyv::{Archive, Deserialize, Serialize};
 
 use crate::in_memory::{ArchivedRowWrapper, DataPages, RowWrapper, StorableRow};
-use crate::lock::{Lock, LockMap, RowLock};
+use crate::lock::{Lock, LockGuard, LockMap, RowLock};
 use crate::prelude::{OffsetEqLink, TablePrimaryKey};
 use crate::vacuum::VacuumPersistence;
 use crate::vacuum::VacuumStats;
@@ -279,32 +279,48 @@ where
         drop(range);
 
         for (from_link, pk) in links {
-            let lock = self.full_row_lock(&pk).await;
-            let _mutation_guard = self.lock_manager.mutation_guard(&pk);
-            if self
-                .data_pages
-                .with_ref(from_link.0, |r| r.is_deleted())
-                .expect("link should be valid")
-            {
-                lock.unlock();
-                self.lock_manager.remove_with_lock_check(&pk);
-                continue;
-            }
-            let (raw_data, new_link) = unsafe {
-                self.data_pages
-                    .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);
+            self.move_candidate_if_current(from_link.0, pk, to).await?;
         }
 
         Ok((from_page_will_be_moved, to_page_will_be_filled))
     }
 
+    /// Moves a reverse-index candidate only if it is still the forward-index
+    /// location after the row lock is acquired.
+    ///
+    /// `move_data_from` snapshots the reverse index before taking per-row
+    /// locks. A concurrent reinsert may move the key and recycle the captured
+    /// slot for a different row in that interval. Revalidating under the row
+    /// lock prevents vacuum from publishing that replacement row under the
+    /// stale candidate's primary key.
+    async fn move_candidate_if_current(&self, from_link: Link, pk: PrimaryKey, to: PageId) -> eyre::Result<bool> {
+        let lock = self.full_row_lock(&pk).await;
+        let _guard = LockGuard::new_with_mutation(lock, self.lock_manager.clone(), pk.clone());
+
+        let current_link: Option<Link> = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into);
+        if current_link != Some(from_link) {
+            return Ok(false);
+        }
+
+        if self
+            .data_pages
+            .with_ref(from_link, |r| r.is_deleted())
+            .expect("a current primary-index link should be valid")
+        {
+            return Ok(false);
+        }
+
+        let (raw_data, new_link) = unsafe {
+            self.data_pages
+                .move_row_for_vacuum(from_link, to)
+                .expect("links and destination capacity were checked")
+        };
+        self.update_index_after_move(pk, from_link, new_link, raw_data)?;
+        self.data_pages.retire_published_link(from_link);
+
+        Ok(true)
+    }
+
     async fn full_row_lock(&self, pk: &PrimaryKey) -> Arc<Lock> {
         let lock_id = self.lock_manager.next_id();
         // One atomic acquire, no check-then-act: see LockMap::get_or_insert_with.
@@ -942,4 +958,74 @@ mod tests {
 
         drop(read_guard);
     }
+
+    #[tokio::test]
+    async fn vacuum_skips_a_stale_candidate_after_its_link_is_reused() {
+        let table = TestWorkTable::default();
+        let target = TestRow {
+            id: table.get_next_pk().into(),
+            test: 10,
+            another: 10,
+            exchange: "target00".to_string(),
+        };
+        let target_id = target.id;
+        table.insert(target).unwrap();
+
+        // Model the reverse-index snapshot taken before vacuum waits for the
+        // row lock.
+        let stale_link = table
+            .0
+            .primary_index
+            .pk_map
+            .get_value(&TestPrimaryKey::from(target_id))
+            .unwrap()
+            .0;
+
+        // A same-sized reinsert moves the target and retires its old slot.
+        let updated_target = TestRow {
+            id: target_id,
+            test: 11,
+            another: 11,
+            exchange: "updated0".to_string(),
+        };
+        table.update(updated_target.clone()).await.unwrap();
+        let current_target_link = table
+            .0
+            .primary_index
+            .pk_map
+            .get_value(&TestPrimaryKey::from(target_id))
+            .unwrap()
+            .0;
+        assert_ne!(current_target_link, stale_link);
+
+        // Reuse the retired physical slot for a different row. Without the
+        // post-lock forward-index check, vacuum would move this row and bind
+        // it to `target_id`.
+        let replacement = TestRow {
+            id: table.get_next_pk().into(),
+            test: 12,
+            another: 12,
+            exchange: "reused00".to_string(),
+        };
+        let replacement_id = replacement.id;
+        table.insert(replacement.clone()).unwrap();
+        let replacement_link = table
+            .0
+            .primary_index
+            .pk_map
+            .get_value(&TestPrimaryKey::from(replacement_id))
+            .unwrap()
+            .0;
+        assert_eq!(replacement_link, stale_link, "test setup must recycle the stale slot");
+
+        let destination = table.0.data.allocate_new_or_pop_free().id;
+        let moved = create_vacuum(&table)
+            .move_candidate_if_current(stale_link, TestPrimaryKey::from(target_id), destination)
+            .await
+            .unwrap();
+
+        assert!(!moved, "vacuum must reject a candidate whose key moved");
+        assert_eq!(table.select(target_id), Some(updated_target));
+        assert_eq!(table.select(replacement_id), Some(replacement));
+    }
 }

tests/worktable/update_in_place_unsized.rs

@@ -44,10 +44,12 @@ macro_rules! unsized_in_place_suite {
                 columns: {
                     id: u64 primary_key using $using,
                     payload: String,
+                    balance: f64,
                 },
                 queries: {
                     update: {
                         Payload(payload) by id,
+                        Balance(balance) by id,
                     }
                 }
             );
@@ -72,6 +74,7 @@ macro_rules! unsized_in_place_suite {
                     .insert(UnsizedUpdateRow {
                         id: 1,
                         payload: "abcdefgh".to_string(), // 8 bytes
+                        balance: 1.0,
                     })
                     .unwrap();
 
@@ -105,6 +108,7 @@ macro_rules! unsized_in_place_suite {
                     .insert(UnsizedUpdateRow {
                         id: 1,
                         payload: "abcdefghij".to_string(),
+                        balance: 1.0,
                     })
                     .unwrap();
 
@@ -146,6 +150,7 @@ macro_rules! unsized_in_place_suite {
                     .insert(UnsizedUpdateRow {
                         id: 1,
                         payload: "0000".to_string(),
+                        balance: 1.0,
                     })
                     .unwrap();
 
@@ -201,6 +206,33 @@ macro_rules! unsized_in_place_suite {
                     format!("{:04}", (20_000u64 - 1) % 10000)
                 );
             }
+
+            #[tokio::test]
+            async fn fixed_width_update_on_unsized_row_stays_in_place() {
+                let table = UnsizedUpdateWorkTable::default();
+                table
+                    .insert(UnsizedUpdateRow {
+                        id: 1,
+                        payload: "out-of-line payload that must remain unchanged".to_string(),
+                        balance: 1.0,
+                    })
+                    .unwrap();
+                let before = link_of(&table, 1);
+
+                table
+                    .update_balance(BalanceQuery { balance: 42.5 }, 1)
+                    .await
+                    .unwrap();
+
+                let after = link_of(&table, 1);
+                let row = table.select(1).unwrap();
+                assert_eq!(before, after);
+                assert_eq!(row.balance, 42.5);
+                assert_eq!(
+                    row.payload,
+                    "out-of-line payload that must remain unchanged"
+                );
+            }
         }
     };
 }