fix: compact fragmented unsized index pages

src/index/unsized_node.rs

@@ -113,7 +113,27 @@ where
     }
 
     fn need_to_split(&self, _: usize, value: &T) -> bool {
-        let final_length = self.length + value.aligned_size();
+        let value_size = value.aligned_size();
+        let current_node_id_size = self.max().map(SizeMeasurable::aligned_size).unwrap_or(0);
+        let next_node_id_size = self
+            .max()
+            .map(|current_max| {
+                if value > current_max {
+                    value_size
+                } else {
+                    current_max.aligned_size()
+                }
+            })
+            .unwrap_or(value_size);
+        // `length` deliberately retains removed bytes until a rebuild. That
+        // makes this estimate conservative before reload; persisted pages are
+        // compacted separately when physical fragmentation survives reload.
+        let final_length = self
+            .length
+            .saturating_sub(current_node_id_size)
+            .saturating_add(next_node_id_size)
+            .saturating_add(value_size)
+            .saturating_add(UnsizedIndexPageUtility::<T>::slots_value_size());
         final_length >= self.length_capacity && self.inner.len() > 1
     }
 

src/persistence/space/index/unsized_.rs

@@ -7,7 +7,7 @@ use std::sync::atomic::{AtomicU32, Ordering};
 use data_bucket::page::PageId;
 use data_bucket::{
     GeneralHeader, GeneralPage, IndexPageUtility, IndexValue, Link, PageType, SizeMeasurable, SpaceId, SpaceInfoPage,
-    UnsizedIndexPage, VariableSizeMeasurable, parse_page, persist_page, persist_pages_batch,
+    UnsizedIndexPage, UnsizedIndexPageUtility, VariableSizeMeasurable, parse_page, persist_page, persist_pages_batch,
 };
 use eyre::eyre;
 use indexset::cdc::change::ChangeEvent;
@@ -59,6 +59,32 @@ where
         + Debug
         + for<'a> rkyv::bytecheck::CheckBytes<rkyv::api::high::HighValidator<'a, rancor::Error>>,
 {
+    fn compact_page_if_needed(page: &mut UnsizedIndexPage<T, DATA_LENGTH>) -> eyre::Result<()> {
+        let persisted_size =
+            UnsizedIndexPageUtility::<T>::persisted_size(page.slots_size as usize, page.node_id_size as usize)
+                + page.last_value_offset as usize;
+        if persisted_size <= DATA_LENGTH as usize {
+            return Ok(());
+        }
+
+        // Removed variable-width entries leave holes at the page tail. After
+        // reload the in-memory node contains only live values, so it cannot
+        // see that physical fragmentation. Compact before the growing slot
+        // directory can overlap the next tail value.
+        // An empty page cannot pass the occupancy guard above, which also
+        // protects `UnsizedIndexPage::rebuild` from its non-empty assumption.
+        page.rebuild();
+        let compacted_size =
+            UnsizedIndexPageUtility::<T>::persisted_size(page.slots_size as usize, page.node_id_size as usize)
+                + page.last_value_offset as usize;
+        if compacted_size > DATA_LENGTH as usize {
+            return Err(eyre!(
+                "unsized index page requires {compacted_size} bytes after compaction, but its capacity is {DATA_LENGTH}"
+            ));
+        }
+        Ok(())
+    }
+
     pub async fn new<S: AsRef<str>>(index_file_path: S, space_id: SpaceId, version: u32) -> eyre::Result<Self> {
         let space_index = SpaceIndex::<T, DATA_LENGTH>::new(index_file_path, space_id, version).await?;
         Ok(Self {
@@ -138,6 +164,32 @@ where
             key: value.key.clone(),
             link: value.value,
         };
+        let value_size = index_value.aligned_size();
+        let future_node_id_size = if node_id.key < value.key {
+            value_size
+        } else {
+            utility.node_id_size as usize
+        };
+        let future_utility_size =
+            UnsizedIndexPageUtility::<T>::persisted_size(utility.slots_size as usize + 1, future_node_id_size);
+        if future_utility_size + utility.last_value_offset as usize + value_size > DATA_LENGTH as usize {
+            let mut page =
+                parse_page::<UnsizedIndexPage<T, DATA_LENGTH>, DATA_LENGTH>(&mut self.index_file, page_id.into())
+                    .await?;
+            page.inner.apply_change_event(ChangeEvent::InsertAt {
+                // The page mutation ignores event ids; this synthetic event
+                // exists only to reuse the same insertion accounting.
+                event_id: 0.into(),
+                max_value: node_id,
+                value,
+                index,
+            })?;
+            Self::compact_page_if_needed(&mut page.inner)?;
+            let changed_node_id =
+                (page.inner.node_id.key != utility.node_id.key).then(|| Pair::from(page.inner.node_id.clone()));
+            persist_page(&mut page, &mut self.index_file).await?;
+            return Ok(changed_node_id);
+        }
         let previous_offset = utility.last_value_offset;
         let value_offset = UnsizedIndexPage::<T, DATA_LENGTH>::persist_value(
             &mut self.index_file,
@@ -449,6 +501,10 @@ where
             }
         }
 
+        for page in pages.values_mut() {
+            Self::compact_page_if_needed(&mut page.inner)?;
+        }
+
         self.table_of_contents.persist(&mut self.index_file).await?;
         persist_pages_batch(pages.values().cloned().collect(), &mut self.index_file).await?;
         // The batch's last page write is a buffered `write_all`; flush so the

tests/persistence/sync/string_secondary_index.rs

@@ -29,6 +29,101 @@ worktable! (
     }
 );
 
+worktable! (
+    name: FragmentedStringSecondary,
+    persist: true,
+    columns: {
+        id: u64 primary_key autoincrement,
+        project_id: String,
+    },
+    indexes: {
+        project_idx: project_id,
+    },
+);
+
+#[test]
+fn fragmented_string_index_compacts_after_restart_before_appending() {
+    let path = "tests/data/unsized_secondary_sync/fragmented_restart";
+    let config = DiskConfig::new_with_table_name(
+        path,
+        FragmentedStringSecondaryWorkTable::name_snake_case(),
+        FragmentedStringSecondaryWorkTable::version(),
+    );
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    runtime.block_on(async {
+        remove_dir_if_exists(path.to_string()).await;
+        let project_id = "proj-fdfb706b-72bc-4b0c-a96e-6b6235be6fb4".to_string();
+        let mut inserted_ids = Vec::new();
+
+        {
+            let engine = FragmentedStringSecondaryPersistenceEngine::new(config.clone())
+                .await
+                .unwrap();
+            let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap();
+            for _ in 0..211 {
+                let row = FragmentedStringSecondaryRow {
+                    id: table.get_next_pk().0,
+                    project_id: project_id.clone(),
+                };
+                inserted_ids.push(row.id);
+                table.insert(row).unwrap();
+            }
+            table.wait_for_ops().await.unwrap();
+            for id in inserted_ids.iter().take(37).copied() {
+                table.delete(id).await.unwrap();
+            }
+            table.wait_for_ops().await.unwrap();
+        }
+
+        {
+            let engine = FragmentedStringSecondaryPersistenceEngine::new(config.clone())
+                .await
+                .unwrap();
+            let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap();
+            for _ in 0..20 {
+                let row = FragmentedStringSecondaryRow {
+                    id: table.get_next_pk().0,
+                    project_id: project_id.clone(),
+                };
+                table.insert(row).unwrap();
+            }
+            table.wait_for_ops().await.unwrap();
+        }
+
+        {
+            let engine = FragmentedStringSecondaryPersistenceEngine::new(config).await.unwrap();
+            let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap();
+            assert_eq!(table.select_all().execute().unwrap().len(), 194);
+            assert_eq!(table.select_by_project_id(project_id).execute().unwrap().len(), 194);
+        }
+
+        let index_path = format!("{path}/fragmented_string_secondary/project_idx.wt.idx");
+        let mut index_file = tokio::fs::File::open(index_path).await.unwrap();
+        let page = parse_page::<UnsizedIndexPage<String, { INNER_PAGE_SIZE as u32 }>, { INNER_PAGE_SIZE as u32 }>(
+            &mut index_file,
+            2,
+        )
+        .await
+        .unwrap();
+        let utility_size = worktable::data_bucket::UnsizedIndexPageUtility::<String>::persisted_size(
+            page.inner.slots_size as usize,
+            page.inner.node_id_size as usize,
+        );
+        assert!(
+            utility_size + page.inner.last_value_offset as usize <= INNER_PAGE_SIZE,
+            "slot directory must not overlap values stored at the page tail"
+        );
+
+        remove_dir_if_exists(path.to_string()).await;
+    });
+}
+
 #[test]
 fn test_space_insert_sync() {
     let config = DiskConfig::new_with_table_name(