From 81bb7b24e77a8280152e0aa1546d2057efb2619d Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 16:41:28 -0500 Subject: [PATCH] fix: rebuild item indexes from durable storage on restart After a pod restart, in-memory indexes (universe bitmap, category, format, creator, tags, duration, created_at) were empty because they were only populated by write_item_with_metadata() calls. This caused /health to report items: 0 and queries to return no results despite data persisting on disk in fjall storage. Add rebuild_item_indexes() which scans the items keyspace for Tag::Meta entries on startup and repopulates all indexes from stored metadata. Update m2_uat crash recovery test to assert indexes survive restart without rewriting items. Co-Authored-By: Claude Opus 4.6 --- tidal/src/db/open.rs | 16 +++++- tidal/src/db/state_rebuild.rs | 92 +++++++++++++++++++++++++++++++++++ tidal/tests/m2_uat.rs | 32 +++++++++--- 3 files changed, 132 insertions(+), 8 deletions(-) diff --git a/tidal/src/db/open.rs b/tidal/src/db/open.rs index 3d5f015..e9dab8f 100644 --- a/tidal/src/db/open.rs +++ b/tidal/src/db/open.rs @@ -22,7 +22,7 @@ use super::config::StorageMode; use super::storage_box::StorageBox; use super::wal_bridge::WalHandleWriter; -use super::state_rebuild::rebuild_entity_state; +use super::state_rebuild::{rebuild_entity_state, rebuild_item_indexes}; /// Bundle returned by [`TidalDb::open_with_schema`] to avoid a fragile tuple. /// @@ -177,6 +177,20 @@ impl super::TidalDb { let interaction_ledger = InteractionLedger::new(); rebuild_entity_state(&storage, &user_state, &creator_items, &interaction_ledger)?; + // Rebuild item indexes (universe bitmap + bitmap/range indexes) + // from persisted metadata so queries and /health work immediately. + let mut universe = universe; + rebuild_item_indexes( + &storage, + &mut universe, + &category_index, + &format_index, + &creator_index, + &tag_index, + &duration_index, + &created_at_index, + )?; + // Read preference vector dimensionality from the schema. let pref_dim = schema_def .embedding_slots() diff --git a/tidal/src/db/state_rebuild.rs b/tidal/src/db/state_rebuild.rs index b512cf5..e630d27 100644 --- a/tidal/src/db/state_rebuild.rs +++ b/tidal/src/db/state_rebuild.rs @@ -5,10 +5,14 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; +use roaring::RoaringBitmap; + use crate::cohort::CohortSignalLedger; use crate::query::suggest::SuggestionIndex; use crate::schema::{TidalError, Timestamp}; use crate::signals::{DEFAULT_MAX_SIGNAL_ENTRIES, SignalLedger, trim_cold_entries}; +use crate::storage::indexes::bitmap::BitmapIndex; +use crate::storage::indexes::range::RangeIndex; use crate::storage::{StorageEngine, Tag}; use super::metadata::deserialize_metadata; @@ -185,6 +189,94 @@ pub(super) fn rebuild_suggestion_index(storage: &StorageBox, suggestion_index: & } } +/// Rebuild item indexes (universe bitmap + bitmap/range indexes) from durable storage. +/// +/// Scans the items keyspace for `Tag::Meta` keys and populates the in-memory +/// universe bitmap and all six item indexes (category, format, creator, tags, +/// duration, `created_at`) from the persisted metadata. This ensures `/health` +/// reports the correct item count and queries work immediately after a restart +/// without rewriting items. +/// +/// Items without explicit `created_at` metadata will not have a `created_at` +/// range index entry restored — the original `Timestamp::now()` default was +/// only stored in the range index, not in metadata. +/// +/// For ephemeral mode the engine is empty, so this is a no-op. +#[allow(clippy::cast_possible_truncation, clippy::too_many_arguments)] +pub(super) fn rebuild_item_indexes( + storage: &StorageBox, + universe: &mut RoaringBitmap, + category_index: &BitmapIndex, + format_index: &BitmapIndex, + creator_index: &BitmapIndex, + tag_index: &BitmapIndex, + duration_index: &RangeIndex, + created_at_index: &RangeIndex, +) -> crate::Result<()> { + use crate::storage::keys::parse_key; + + let mut count = 0u64; + let scan_start = std::time::Instant::now(); + + for entry in storage.items_engine().scan_prefix(&[]) { + let (key, value) = entry.map_err(TidalError::from)?; + + if let Some((entity_id, Tag::Meta, _suffix)) = parse_key(&key) { + let id_u32 = entity_id.as_u64() as u32; + let meta = deserialize_metadata(&value); + + // Universe bitmap. + universe.insert(id_u32); + + // Bitmap indexes. + if let Some(val) = meta.get("category") { + category_index.insert(id_u32, val); + } + if let Some(val) = meta.get("format") { + format_index.insert(id_u32, val); + } + if let Some(val) = meta.get("creator_id") { + creator_index.insert(id_u32, val); + } + if let Some(tags) = meta.get("tags") { + for tag in tags.split(',') { + let tag = tag.trim(); + if !tag.is_empty() { + tag_index.insert(id_u32, tag); + } + } + } + + // Range indexes. + if let Some(val) = meta.get("duration") + && let Ok(dur) = val.parse::() + { + duration_index.insert(id_u32, dur); + } + if let Some(val) = meta.get("created_at") + && let Ok(ts) = val.parse::() + { + created_at_index.insert(id_u32, ts); + } + + count += 1; + if count.is_multiple_of(10_000) { + tracing::info!(rebuilt = count, "item index rebuild in progress"); + } + } + } + + if count > 0 { + tracing::info!( + items = count, + elapsed_ms = scan_start.elapsed().as_millis(), + "item indexes rebuilt from durable storage" + ); + } + + Ok(()) +} + /// Background thread body: checkpoint signal state to storage every 30 seconds. /// /// Checkpoints both the global signal ledger and the cohort signal ledger diff --git a/tidal/tests/m2_uat.rs b/tidal/tests/m2_uat.rs index 4046c28..60b5faf 100644 --- a/tidal/tests/m2_uat.rs +++ b/tidal/tests/m2_uat.rs @@ -415,9 +415,14 @@ fn milestone_2_uat() { .open() .expect("reopen failed (second session)"); - // Note: In-memory indexes (bitmaps, range) are NOT persisted in M2. - // After reopen, the universe is empty and queries return no results. - // Signal state IS persisted via WAL + checkpoint. + // Item indexes are now rebuilt from durable storage on restart. + // Universe bitmap, bitmap indexes, and range indexes should all + // be populated without rewriting items. + assert_eq!( + db2.item_count(), + 1_000, + "crash recovery: universe must contain 1K items after reopen (rebuilt from storage)" + ); // Verify signal recovery: decay score should survive. let score_after_reopen = db2 @@ -444,11 +449,24 @@ fn milestone_2_uat() { "crash recovery: share count for entity 42 should be >= 100, got {share_count_recovered}" ); - // Rewrite items so the in-memory indexes are repopulated, then re-query. - write_items(&db2, base_ns); - assert_eq!(db2.item_count(), 1_000, "universe must be repopulated"); + // Verify filtered query works without rewriting items — bitmap indexes + // were rebuilt from storage. + let query = Retrieve::builder() + .profile("hot") + .limit(10) + .filter(FilterExpr::CategoryEq("jazz".into())) + .build() + .expect("post-recovery filtered query build failed"); + let results = db2 + .retrieve(&query) + .expect("post-recovery filtered query failed"); + assert!( + !results.items.is_empty(), + "post-recovery filtered query must return results (indexes rebuilt from storage)" + ); + assert_score_descending(&results.items, "post-recovery hot+jazz"); - // Verify query still works after recovery + index repopulation. + // Verify unfiltered query also works. let query = Retrieve::builder() .profile("new") .limit(10)