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 <noreply@anthropic.com>
This commit is contained in:
parent
889c746bc1
commit
81bb7b24e7
@ -22,7 +22,7 @@ use super::config::StorageMode;
|
|||||||
use super::storage_box::StorageBox;
|
use super::storage_box::StorageBox;
|
||||||
use super::wal_bridge::WalHandleWriter;
|
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.
|
/// Bundle returned by [`TidalDb::open_with_schema`] to avoid a fragile tuple.
|
||||||
///
|
///
|
||||||
@ -177,6 +177,20 @@ impl super::TidalDb {
|
|||||||
let interaction_ledger = InteractionLedger::new();
|
let interaction_ledger = InteractionLedger::new();
|
||||||
rebuild_entity_state(&storage, &user_state, &creator_items, &interaction_ledger)?;
|
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.
|
// Read preference vector dimensionality from the schema.
|
||||||
let pref_dim = schema_def
|
let pref_dim = schema_def
|
||||||
.embedding_slots()
|
.embedding_slots()
|
||||||
|
|||||||
@ -5,10 +5,14 @@ use std::sync::Arc;
|
|||||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use roaring::RoaringBitmap;
|
||||||
|
|
||||||
use crate::cohort::CohortSignalLedger;
|
use crate::cohort::CohortSignalLedger;
|
||||||
use crate::query::suggest::SuggestionIndex;
|
use crate::query::suggest::SuggestionIndex;
|
||||||
use crate::schema::{TidalError, Timestamp};
|
use crate::schema::{TidalError, Timestamp};
|
||||||
use crate::signals::{DEFAULT_MAX_SIGNAL_ENTRIES, SignalLedger, trim_cold_entries};
|
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 crate::storage::{StorageEngine, Tag};
|
||||||
|
|
||||||
use super::metadata::deserialize_metadata;
|
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<u32>,
|
||||||
|
created_at_index: &RangeIndex<u64>,
|
||||||
|
) -> 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::<u32>()
|
||||||
|
{
|
||||||
|
duration_index.insert(id_u32, dur);
|
||||||
|
}
|
||||||
|
if let Some(val) = meta.get("created_at")
|
||||||
|
&& let Ok(ts) = val.parse::<u64>()
|
||||||
|
{
|
||||||
|
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.
|
/// Background thread body: checkpoint signal state to storage every 30 seconds.
|
||||||
///
|
///
|
||||||
/// Checkpoints both the global signal ledger and the cohort signal ledger
|
/// Checkpoints both the global signal ledger and the cohort signal ledger
|
||||||
|
|||||||
@ -415,9 +415,14 @@ fn milestone_2_uat() {
|
|||||||
.open()
|
.open()
|
||||||
.expect("reopen failed (second session)");
|
.expect("reopen failed (second session)");
|
||||||
|
|
||||||
// Note: In-memory indexes (bitmaps, range) are NOT persisted in M2.
|
// Item indexes are now rebuilt from durable storage on restart.
|
||||||
// After reopen, the universe is empty and queries return no results.
|
// Universe bitmap, bitmap indexes, and range indexes should all
|
||||||
// Signal state IS persisted via WAL + checkpoint.
|
// 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.
|
// Verify signal recovery: decay score should survive.
|
||||||
let score_after_reopen = db2
|
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}"
|
"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.
|
// Verify filtered query works without rewriting items — bitmap indexes
|
||||||
write_items(&db2, base_ns);
|
// were rebuilt from storage.
|
||||||
assert_eq!(db2.item_count(), 1_000, "universe must be repopulated");
|
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()
|
let query = Retrieve::builder()
|
||||||
.profile("new")
|
.profile("new")
|
||||||
.limit(10)
|
.limit(10)
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user