jordan/tidaldb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: M0-M10 remediation — deferred post-filters, WAL hardening, docker build fixes

Browse Source 
- Extract shared deferred post-filter (InCollection/MinSignal/MaxSignal/
  NearLocation/SocialGraph) into query/executor/post_filter.rs so RETRIEVE and
  SEARCH apply identical predicates; SEARCH previously ignored four variants
- Harden WAL writer/compaction/dedup paths
- Brute-force vector store fixes + tests
- Governance community ledger updates
- Pin docker builders to bookworm (glibc/libmvec match) and add
  protobuf-compiler for tidal-net's tonic-build
- Add M0-M10 code-review document
This commit is contained in:
jx12n 2026-06-07 20:47:17 -06:00
parent 3bcfb3c576
commit 69ae6e64a1
17 changed files with 2982 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docker/cluster/Dockerfile
View File

@ -3,7 +3,9 @@
# Runs a simulated 3-region cluster (us-east, eu-west, ap-south) behind a # Runs a simulated 3-region cluster (us-east, eu-west, ap-south) behind a
# single HTTP surface on port 9500. See docs/runbooks/cluster.md for the # single HTTP surface on port 9500. See docs/runbooks/cluster.md for the
# operational API. # operational API.
FROM rust:1.91 AS builder # Pin the builder to bookworm so its glibc matches the bookworm-slim runtime;
# the default tag tracks trixie, whose `libmvec.so.1` is absent on bookworm.
FROM rust:1.91-bookworm AS builder
ARG DEBIAN_FRONTEND=noninteractive ARG DEBIAN_FRONTEND=noninteractive
@ -23,7 +25,9 @@ COPY applications/iknowyou/engine/Cargo.toml applications/iknowyou/engine/Cargo.
# Copy full workspace. # Copy full workspace.
COPY . . COPY . .
RUN apt-get update && apt-get install -y g++ && rm -rf /var/lib/apt/lists/* # g++ builds the usearch C++ HNSW core; protobuf-compiler (protoc) is required by
# tidal-net's build script (tonic-build compiles the WAL-shipping .proto).
RUN apt-get update && apt-get install -y g++ protobuf-compiler && rm -rf /var/lib/apt/lists/*
RUN cargo build -p tidal-server --release RUN cargo build -p tidal-server --release
FROM debian:bookworm-slim FROM debian:bookworm-slim

10
docker/deploy/Dockerfile
View File

@ -1,6 +1,12 @@
FROM rust:1.91-slim AS builder # Pin the builder to bookworm so its glibc matches the bookworm-slim runtime.
# The default slim tag tracks Debian trixie, whose vectorized-math `libmvec.so.1`
# is absent on bookworm and aborts the binary at startup.
FROM rust:1.91-slim-bookworm AS builder
WORKDIR /build WORKDIR /build
RUN apt-get update && apt-get install -y pkg-config libssl-dev g++ && rm -rf /var/lib/apt/lists/* # protobuf-compiler (protoc) is required by tidal-net's build script (tonic-build
# compiles the WAL-shipping .proto); tidal-server depends on tidal-net for gRPC
# cluster replication. g++ builds the usearch C++ HNSW core.
RUN apt-get update && apt-get install -y pkg-config libssl-dev g++ protobuf-compiler && rm -rf /var/lib/apt/lists/*
# Copy workspace manifests first for layer caching. # Copy workspace manifests first for layer caching.
COPY Cargo.toml Cargo.lock ./ COPY Cargo.toml Cargo.lock ./

9
docker/standalone/Dockerfile
View File

@ -1,6 +1,13 @@
FROM rust:1.91 AS builder # Pin the builder to bookworm so its glibc matches the bookworm-slim runtime;
# the default tag tracks trixie, whose `libmvec.so.1` is absent on bookworm.
FROM rust:1.91-bookworm AS builder
WORKDIR /app WORKDIR /app
# protobuf-compiler (protoc) is required by tidal-net's build script (tonic-build
# compiles the WAL-shipping .proto); tidal-server depends on tidal-net for gRPC
# cluster replication. The full rust image already ships g++/pkg-config/libssl.
RUN apt-get update && apt-get install -y --no-install-recommends protobuf-compiler && rm -rf /var/lib/apt/lists/*
# Copy workspace manifests first for layer caching. # Copy workspace manifests first for layer caching.
COPY Cargo.toml Cargo.lock ./ COPY Cargo.toml Cargo.lock ./
COPY tidal/Cargo.toml tidal/Cargo.toml COPY tidal/Cargo.toml tidal/Cargo.toml

1754
docs/reviews/M0-M10-code-review.md Normal file
View File

File diff suppressed because it is too large Load Diff

8
tidal/src/db/state_rebuild.rs
View File

@ -442,11 +442,15 @@ pub(super) fn run_checkpoint_thread(
// Compact WAL segments covered by the checkpoint. // Compact WAL segments covered by the checkpoint.
// This runs AFTER the checkpoint is durable, so deleted // This runs AFTER the checkpoint is durable, so deleted
// segments are guaranteed to be redundant. // segments are guaranteed to be redundant. The WAL writer thread
// is still live here, so we MUST use the online variant, which
// never deletes the segment the writer is appending to (deleting
// it would unlink the inode out from under the open FD and lose
// post-checkpoint writes on Linux).
if let Some(ref dir) = wal_dir if let Some(ref dir) = wal_dir
&& seq > 0 && seq > 0
{ {
match crate::wal::compaction::compact_wal(dir, seq) { match crate::wal::compaction::compact_wal_online(dir, seq) {
Ok(result) => { Ok(result) => {
#[cfg(feature = "metrics")] #[cfg(feature = "metrics")]
{ {

132
tidal/src/governance/community_ledger.rs
View File

@ -26,7 +26,7 @@ use crate::{
ledger::types::EntitySignalEntry, ledger::types::EntitySignalEntry,
warm::BucketedCounter, warm::BucketedCounter,
}, },
storage::{StorageEngine, Tag, encode_key, entity_tag_prefix}, storage::{StorageEngine, Tag, WriteBatch, encode_key, entity_tag_prefix},
}; };
/// The writer that produced a contribution. /// The writer that produced a contribution.
@ -401,27 +401,38 @@ impl CommunityLedger {
/// ///
/// Returns `TidalError::Storage` on a write failure. /// Returns `TidalError::Storage` on a write failure.
pub fn checkpoint(&self, storage: &dyn StorageEngine) -> crate::Result<()> { pub fn checkpoint(&self, storage: &dyn StorageEngine) -> crate::Result<()> {
// Clear any stale entries first so a purged contributor does not linger. // This ledger is the SOLE durable copy of per-contributor state — it is
// not rebuildable from the WAL — so the old→new snapshot swap MUST be
// all-or-nothing. The previous implementation issued individual
// `delete()`s of every stale row followed by individual `put()`s of the
// current cells; a crash or I/O error between the deletes and the puts
// left the durable snapshot partially erased, silently losing community
// contributions on a normal shutdown-checkpoint. We stage every
// stale-key delete and every current-cell put into a single WriteBatch
// and commit once, matching the cohort/co-engagement checkpoints, so the
// backend applies the whole replacement atomically.
let prefix = entity_tag_prefix(EntityId::new(0), Tag::Community); let prefix = entity_tag_prefix(EntityId::new(0), Tag::Community);
let stale: Vec<_> = storage let mut batch = WriteBatch::new();
.scan_prefix(&prefix)
.flatten() // Stage deletion of any stale entries so a purged contributor does not
.map(|(k, _)| k) // linger after the swap.
.collect(); for item in storage.scan_prefix(&prefix) {
for k in stale { let (key, _) = item.map_err(TidalError::from)?;
storage.delete(&k).map_err(TidalError::from)?; batch.delete(key);
} }
// Stage the current snapshot.
for cell in &self.cells { for cell in &self.cells {
let (community, entity, type_id) = *cell.key(); let (community, entity, type_id) = *cell.key();
for ((user, writer, epoch), entry) in &cell.value().contributors { for ((user, writer, epoch), entry) in &cell.value().contributors {
let suffix = let suffix =
encode_contributor_suffix(community, *user, *epoch, entity, type_id, writer); encode_contributor_suffix(community, *user, *epoch, entity, type_id, writer);
let key = encode_key(EntityId::new(0), Tag::Community, &suffix); let key = encode_key(EntityId::new(0), Tag::Community, &suffix);
storage batch.put(key, serialize_entry(entity, type_id, entry));
.put(&key, &serialize_entry(entity, type_id, entry))
.map_err(TidalError::from)?;
} }
} }
storage.write_batch(batch).map_err(TidalError::from)?;
Ok(()) Ok(())
} }
@ -869,6 +880,103 @@ mod tests {
); );
} }
/// Count the durable `Tag::Community` rows in a backend — i.e. how many
/// contributor cells the checkpoint actually persisted.
fn durable_community_row_count(storage: &crate::storage::InMemoryBackend) -> usize {
let prefix = entity_tag_prefix(EntityId::new(0), Tag::Community);
storage.scan_prefix(&prefix).flatten().count()
}
/// Total number of in-memory contributor cells across every community cell.
fn in_memory_contributor_count(l: &CommunityLedger) -> usize {
l.cells.iter().map(|c| c.value().contributors.len()).sum()
}
#[test]
fn checkpoint_persists_exactly_the_in_memory_cells() {
// The durable snapshot must contain one row per in-memory contributor
// cell — no more, no fewer. This is the all-or-nothing-swap invariant:
// every cell is persisted and nothing stale lingers.
use crate::storage::InMemoryBackend;
let storage = InMemoryBackend::new();
let l = ledger();
let c = CommunityId(1);
let like = l.resolve_signal_type("like").unwrap();
let save = l.resolve_signal_type("save").unwrap();
let ts = Timestamp::now().as_nanos();
// Two contributors across two items and two signal types.
for (user, item, tid) in [
(UserId(1), EntityId::new(10), like),
(UserId(1), EntityId::new(20), save),
(UserId(2), EntityId::new(10), like),
] {
l.record(c, item, tid, user, "agent", 0, 1.0, ts);
}
l.checkpoint(&storage).unwrap();
let durable = durable_community_row_count(&storage);
let in_mem = in_memory_contributor_count(&l);
assert_eq!(
durable, in_mem,
"durable rows ({durable}) must exactly equal in-memory cells ({in_mem})"
);
assert_eq!(durable, 3, "expected 3 contributor rows persisted");
}
#[test]
fn recheckpoint_after_purge_replaces_snapshot_atomically() {
// After a purge shrinks the in-memory state, a re-checkpoint must leave
// the durable snapshot with *exactly* the surviving cells — no stale
// rows for the purged contributor. Because the swap is a single atomic
// WriteBatch (stale deletes + current puts), the durable row count
// always matches the in-memory cell count, never the pre-purge count.
use crate::storage::InMemoryBackend;
let storage = InMemoryBackend::new();
let l = ledger();
let c = CommunityId(1);
let like = l.resolve_signal_type("like").unwrap();
let ts = Timestamp::now().as_nanos();
// Three contributors, then checkpoint.
for u in 1..=3u64 {
l.record(c, EntityId::new(5), like, UserId(u), "", 0, 1.0, ts);
}
l.checkpoint(&storage).unwrap();
assert_eq!(durable_community_row_count(&storage), 3);
// Purge user 2, then re-checkpoint. The durable snapshot must shed the
// purged row and hold only the two survivors.
let removed = l.purge_contributor(c, UserId(2), 0..=u32::MAX);
assert_eq!(removed, 1);
l.checkpoint(&storage).unwrap();
let durable = durable_community_row_count(&storage);
assert_eq!(
durable,
in_memory_contributor_count(&l),
"re-checkpoint must leave exactly the surviving cells"
);
assert_eq!(
durable, 2,
"purged contributor's stale row must not linger after re-checkpoint"
);
// And a fresh restore reflects exactly the post-purge state.
let restored = ledger();
restored.restore(&storage);
assert_eq!(
restored
.windowed_count(c, EntityId::new(5), "like", Window::AllTime)
.unwrap(),
2,
"restored aggregate must reflect the purge, not the pre-purge snapshot"
);
}
#[test] #[test]
fn apply_purge_tombstone_purges_and_sets_watermark() { fn apply_purge_tombstone_purges_and_sets_watermark() {
use crate::governance::tombstone::PurgeTombstone; use crate::governance::tombstone::PurgeTombstone;

1
tidal/src/query/executor/mod.rs
View File

@ -15,6 +15,7 @@
pub mod candidate_gen; pub mod candidate_gen;
pub mod personalization; pub mod personalization;
pub mod post_filter;
pub mod social_filter; pub mod social_filter;
pub mod user_filter; pub mod user_filter;

139
tidal/src/query/executor/pipeline.rs
View File

@ -9,9 +9,8 @@ use std::{collections::HashSet, time::Instant};
use roaring::RoaringBitmap; use roaring::RoaringBitmap;
use super::{RetrieveExecutor, candidate_gen, social_filter, user_filter}; use super::{RetrieveExecutor, candidate_gen, post_filter, user_filter};
use crate::{ use crate::{
db::deserialize_metadata as deserialize_item_metadata,
query::{ query::{
retrieve::{Cursor, QueryError, Results, Retrieve, RetrieveResult, Signal}, retrieve::{Cursor, QueryError, Results, Retrieve, RetrieveResult, Signal},
stats::QueryStats, stats::QueryStats,
@ -21,14 +20,11 @@ use crate::{
profile::{CandidateStrategy, Sort}, profile::{CandidateStrategy, Sort},
}, },
schema::{EntityId, Timestamp}, schema::{EntityId, Timestamp},
storage::{ storage::indexes::{
Tag, encode_key,
indexes::{
bitmap::BitmapIndex, bitmap::BitmapIndex,
filter::{FilterEvaluator, FilterExpr, FilterResult}, filter::{FilterEvaluator, FilterExpr, FilterResult},
range::RangeIndex, range::RangeIndex,
}, },
},
}; };
impl RetrieveExecutor<'_> { impl RetrieveExecutor<'_> {
@ -239,110 +235,22 @@ impl RetrieveExecutor<'_> {
tracing::trace!(candidates = candidates.len(), "stage 2: filter applied"); tracing::trace!(candidates = candidates.len(), "stage 2: filter applied");
// ── Stage 2.2: InCollection Post-Filter (M6p4) ────────────────── // ── Stage 2.2/2.3/2.4: Deferred Post-Filters ────────────────────
// InCollection (M6p4), MinSignal/MaxSignal thresholds (M6p3), and
// NearLocation geo (M6p3). These evaluate to the full universe at the
// FilterEvaluator level and must be applied here against the live
// collection index, ledger, and item metadata. Shared verbatim with the
// SEARCH pipeline via `post_filter::apply_deferred_post_filters` so the
// two surfaces cannot drift (a missing arm there silently leaked items
// that violated the filter).
if let Some(ref filter_expr) = combined_filter { if let Some(ref filter_expr) = combined_filter {
let coll_filters = user_filter::extract_collection_filters(filter_expr); let ctx = post_filter::PostFilterCtx {
if !coll_filters.is_empty() ledger: self.ledger,
&& let Some(coll_idx) = self.collection_index collection_index: self.collection_index,
{ items_storage: self.items_storage,
for cf in &coll_filters { degradation_level: self.degradation_level,
if let FilterExpr::InCollection(cid) = cf {
if let Some(bm) = coll_idx.bitmap(*cid) {
candidates.retain(|id| bm.contains(id.as_u64() as u32));
} else {
// Collection not found or empty -- retain nothing.
candidates.clear();
}
}
}
tracing::trace!(
candidates = candidates.len(),
"stage 2.2: InCollection filters applied"
);
}
}
// ── Stage 2.3: Signal Threshold Post-Filters (M6p3) ─────────────
if let Some(ref filter_expr) = combined_filter {
let sig_filters = user_filter::extract_signal_threshold_filters(filter_expr);
if !sig_filters.is_empty() {
for sf in &sig_filters {
// An unknown signal name in a threshold filter must fail the
// query loudly rather than silently retaining/dropping every
// candidate, so the fallible `read_agg` is evaluated up front
// and its error mapped to a `QueryError`.
let (signal, threshold, keep_if_ge) = match sf {
FilterExpr::MinSignal { signal, threshold } => (signal, *threshold, true),
FilterExpr::MaxSignal { signal, threshold } => (signal, *threshold, false),
_ => continue,
}; };
let mut kept = Vec::with_capacity(candidates.len()); post_filter::apply_deferred_post_filters(filter_expr, &mut candidates, &ctx)?;
for &eid in &candidates {
let val = crate::ranking::executor::helpers::read_agg(
eid,
signal,
&crate::ranking::profile::SignalAgg::Value,
crate::schema::Window::AllTime,
self.ledger,
self.degradation_level,
)
.map_err(|e| QueryError::InvalidFilter {
field: signal.clone(),
reason: e.to_string(),
})?;
if (keep_if_ge && val >= threshold) || (!keep_if_ge && val <= threshold) {
kept.push(eid);
}
}
candidates = kept;
}
tracing::trace!(
candidates = candidates.len(),
"stage 2.3: signal threshold filters applied"
);
}
}
// ── Stage 2.4: Geographic Post-Filter (M6p3) ────────────────────
if let Some(ref filter_expr) = combined_filter {
let geo_filters = user_filter::extract_geo_filters(filter_expr);
if !geo_filters.is_empty()
&& let Some(storage) = self.items_storage
{
candidates.retain(|&eid| {
let key = encode_key(eid, Tag::Meta, b"");
let meta = storage
.get(&key)
.ok()
.flatten()
.map(|b| deserialize_item_metadata(&b))
.unwrap_or_default();
geo_filters.iter().all(|gf| match gf {
FilterExpr::NearLocation {
lat,
lng,
radius_km,
} => {
let item_lat = meta.get("latitude").and_then(|v| v.parse::<f64>().ok());
let item_lng =
meta.get("longitude").and_then(|v| v.parse::<f64>().ok());
match (item_lat, item_lng) {
(Some(ilat), Some(ilng)) => {
crate::ranking::executor::formulas::haversine_km(
*lat, *lng, ilat, ilng,
) <= *radius_km
}
_ => false,
}
}
_ => true,
})
});
tracing::trace!(
candidates = candidates.len(),
"stage 2.4: geographic filters applied"
);
}
} }
// ── Stage 2.5: User-Context Filtering (M3) ───────────────────── // ── Stage 2.5: User-Context Filtering (M3) ─────────────────────
@ -400,13 +308,16 @@ impl RetrieveExecutor<'_> {
depth, depth,
} => { } => {
// Social graph: constrain to items from followed creators // Social graph: constrain to items from followed creators
// (depth=1) or expanded community items (depth=2). // (depth=1) or expanded community items (depth=2). Shared
if let Some(ci) = self.creator_items { // with the SEARCH pipeline so neither surface silently
let sg_bitmap = social_filter::social_graph_bitmap( // ignores a `social_graph` filter.
*uid, *depth, user_state, ci, post_filter::apply_social_graph_filter(
*uid,
*depth,
&mut candidates,
user_state,
self.creator_items,
); );
candidates.retain(|id| sg_bitmap.contains(id.as_u64() as u32));
}
} }
// Other variants are not user-state inclusion filters. // Other variants are not user-state inclusion filters.
_ => {} _ => {}

249
tidal/src/query/executor/post_filter.rs Normal file
View File

@ -0,0 +1,249 @@
//! Shared deferred post-filter application for RETRIEVE and SEARCH.
//!
//! The deferred filter variants — `InCollection`, `MinSignal`/`MaxSignal`,
//! `NearLocation`, and `SocialGraph` — evaluate to the *full universe* at the
//! [`FilterEvaluator`] level (see [`crate::storage::indexes::filter::expr`]).
//! They carry no index-backed predicate and must be applied as **post-filters**
//! after candidate generation/retrieval, against the live ledger, item
//! metadata, collection index, and social graph.
//!
//! Both the RETRIEVE pipeline (Stages 2.2/2.3/2.4 and the `SocialGraph` arm of
//! 2.5) and the SEARCH pipeline (Stage 2.5) apply the *same* predicates. Before
//! this module existed, SEARCH applied only the user-state inclusion filters
//! (`Saved`/`Liked`/`InProgress`) and silently ignored the four deferred
//! variants above — a well-formed SEARCH with `min_signal`/`near_location`/
//! `in_collection`/`social_graph` returned items that violated the filter, with
//! no error and nothing logged. Extracting the predicates here means RETRIEVE
//! and SEARCH cannot drift again: both call one function.
//!
//! [`FilterEvaluator`]: crate::storage::indexes::filter::FilterEvaluator
use crate::{
entities::{CreatorItemsBitmap, UserStateIndex},
load::DegradationLevel,
query::{
executor::{social_filter, user_filter},
retrieve::QueryError,
},
schema::EntityId,
signals::SignalLedger,
storage::{StorageEngine, Tag, encode_key, indexes::filter::FilterExpr},
};
/// Borrowed infrastructure needed to apply the index/ledger/metadata-backed
/// deferred post-filters (`InCollection`, `MinSignal`/`MaxSignal`,
/// `NearLocation`).
///
/// The collection index and item storage are optional because a filter is only
/// enforced when the corresponding index/store is wired. When that handle is
/// missing the corresponding filter is a no-op (the candidate set is left
/// unchanged) — matching the RETRIEVE pipeline's pre-extraction behavior
/// byte-for-byte and the graceful-degradation contract the rest of the executor
/// follows. The real `TidalDb` paths always wire the collection index and item
/// storage, so the no-op branch is only reachable from minimally-constructed
/// executors (e.g. unit tests that exercise other stages).
///
/// `SocialGraph` is *not* applied through this context: it lives in each
/// pipeline's user-context block via [`apply_social_graph_filter`], so its
/// `UserStateIndex`/`CreatorItemsBitmap` handles are passed there directly.
pub(crate) struct PostFilterCtx<'a> {
/// Signal ledger — required for `MinSignal`/`MaxSignal`. Always present.
pub ledger: &'a SignalLedger,
/// Collection index — required for `InCollection`.
pub collection_index: Option<&'a crate::entities::CollectionIndex>,
/// Item metadata storage — required for `NearLocation` (lat/lng lookup).
pub items_storage: Option<&'a dyn StorageEngine>,
/// Load-degradation level, threaded into ledger reads for `MinSignal`/`MaxSignal`.
pub degradation_level: DegradationLevel,
}
/// Apply the `InCollection`, `MinSignal`/`MaxSignal`, and `NearLocation`
/// deferred post-filters to `candidates`, in that order.
///
/// This is the shared core of RETRIEVE Stages 2.2/2.3/2.4 and the equivalent
/// SEARCH stage. The `SocialGraph` arm is intentionally **not** applied here:
/// it lives inside each pipeline's user-context block (after the
/// `Saved`/`Liked`/`InProgress` inclusion arms) so the RETRIEVE ordering stays
/// byte-for-byte identical. Call [`apply_social_graph_filter`] from that block.
///
/// # Errors
///
/// Returns [`QueryError::InvalidFilter`] if a `MinSignal`/`MaxSignal` filter
/// names a signal the schema does not define — the same loud-failure contract
/// RETRIEVE has: an unknown signal must fail the query, never silently retain or
/// drop every candidate.
pub(crate) fn apply_deferred_post_filters(
filter_expr: &FilterExpr,
candidates: &mut Vec<EntityId>,
ctx: &PostFilterCtx<'_>,
) -> Result<(), QueryError> {
apply_in_collection(filter_expr, candidates, ctx.collection_index);
apply_signal_thresholds(filter_expr, candidates, ctx.ledger, ctx.degradation_level)?;
apply_near_location(filter_expr, candidates, ctx.items_storage);
Ok(())
}
/// RETRIEVE Stage 2.2 / SEARCH equivalent: retain only items in every named
/// collection. A missing/empty *collection* retains nothing for that filter —
/// an `InCollection(c)` for a collection with no items matches the empty set,
/// never the full set. An absent *index handle* is a no-op (matching RETRIEVE's
/// pre-extraction behavior); the real DB path always wires it.
fn apply_in_collection(
filter_expr: &FilterExpr,
candidates: &mut Vec<EntityId>,
collection_index: Option<&crate::entities::CollectionIndex>,
) {
let coll_filters = user_filter::extract_collection_filters(filter_expr);
if coll_filters.is_empty() {
return;
}
let Some(coll_idx) = collection_index else {
// No collection index wired: leave candidates unchanged, matching the
// RETRIEVE pipeline's `if !empty && let Some(idx)` guard byte-for-byte.
return;
};
for cf in &coll_filters {
if let FilterExpr::InCollection(cid) = cf {
if let Some(bm) = coll_idx.bitmap(*cid) {
// SAFETY: all entity IDs stored in RoaringBitmaps are 32-bit;
// truncation is intentional and correct throughout the executor.
candidates.retain(|id| bm.contains(id.as_u64() as u32));
} else {
// Collection not found or empty -- retain nothing.
candidates.clear();
}
}
}
tracing::trace!(
candidates = candidates.len(),
"post-filter: InCollection applied"
);
}
/// RETRIEVE Stage 2.3 / SEARCH equivalent: retain only items whose `AllTime`
/// windowed count for the named signal satisfies the threshold (>= for
/// `MinSignal`, <= for `MaxSignal`). Thresholds are conjunctive.
///
/// # Errors
///
/// Returns [`QueryError::InvalidFilter`] when the signal name is not in the
/// schema (`read_agg` propagates the schema error); this fails the query loudly
/// rather than silently retaining or dropping every candidate.
fn apply_signal_thresholds(
filter_expr: &FilterExpr,
candidates: &mut Vec<EntityId>,
ledger: &SignalLedger,
degradation_level: DegradationLevel,
) -> Result<(), QueryError> {
let sig_filters = user_filter::extract_signal_threshold_filters(filter_expr);
if sig_filters.is_empty() {
return Ok(());
}
for sf in &sig_filters {
let (signal, threshold, keep_if_ge) = match sf {
FilterExpr::MinSignal { signal, threshold } => (signal, *threshold, true),
FilterExpr::MaxSignal { signal, threshold } => (signal, *threshold, false),
_ => continue,
};
let mut kept = Vec::with_capacity(candidates.len());
for &eid in &*candidates {
let val = crate::ranking::executor::helpers::read_agg(
eid,
signal,
&crate::ranking::profile::SignalAgg::Value,
crate::schema::Window::AllTime,
ledger,
degradation_level,
)
.map_err(|e| QueryError::InvalidFilter {
field: signal.clone(),
reason: e.to_string(),
})?;
if (keep_if_ge && val >= threshold) || (!keep_if_ge && val <= threshold) {
kept.push(eid);
}
}
*candidates = kept;
}
tracing::trace!(
candidates = candidates.len(),
"post-filter: signal thresholds applied"
);
Ok(())
}
/// RETRIEVE Stage 2.4 / SEARCH equivalent: retain only items within `radius_km`
/// of every `NearLocation` filter, via Haversine distance against the item's
/// `latitude`/`longitude` metadata. Items missing coordinates (or with no
/// metadata) are excluded. An absent *item storage* handle is a no-op (matching
/// RETRIEVE's pre-extraction behavior); the real DB path always wires it.
fn apply_near_location(
filter_expr: &FilterExpr,
candidates: &mut Vec<EntityId>,
items_storage: Option<&dyn StorageEngine>,
) {
let geo_filters = user_filter::extract_geo_filters(filter_expr);
if geo_filters.is_empty() {
return;
}
let Some(storage) = items_storage else {
// No item storage wired: leave candidates unchanged, matching the
// RETRIEVE pipeline's `if !empty && let Some(storage)` guard.
return;
};
candidates.retain(|&eid| {
let key = encode_key(eid, Tag::Meta, b"");
let meta = storage
.get(&key)
.ok()
.flatten()
.map(|b| crate::db::deserialize_metadata(&b))
.unwrap_or_default();
geo_filters.iter().all(|gf| match gf {
FilterExpr::NearLocation {
lat,
lng,
radius_km,
} => {
let item_lat = meta.get("latitude").and_then(|v| v.parse::<f64>().ok());
let item_lng = meta.get("longitude").and_then(|v| v.parse::<f64>().ok());
match (item_lat, item_lng) {
(Some(ilat), Some(ilng)) => {
crate::ranking::executor::formulas::haversine_km(*lat, *lng, ilat, ilng)
<= *radius_km
}
_ => false,
}
}
_ => true,
})
});
tracing::trace!(
candidates = candidates.len(),
"post-filter: NearLocation applied"
);
}
/// Apply the `SocialGraph` inclusion arm: when the filter contains a
/// `SocialGraph { user_id, depth }` and both `user_state` and `creator_items`
/// are wired, retain only items reachable from the user's social graph.
///
/// Separated from [`apply_deferred_post_filters`] so each pipeline can call it
/// at the exact position inside its user-context block (after the
/// `Saved`/`Liked`/`InProgress` arms), preserving RETRIEVE's ordering.
pub(crate) fn apply_social_graph_filter(
user_id: u64,
depth: u8,
candidates: &mut Vec<EntityId>,
user_state: &UserStateIndex,
creator_items: Option<&CreatorItemsBitmap>,
) {
let Some(ci) = creator_items else {
return;
};
let sg_bitmap = social_filter::social_graph_bitmap(user_id, depth, user_state, ci);
// SAFETY: all entity IDs stored in RoaringBitmaps are 32-bit; truncation is
// intentional and correct throughout the executor (a >u32 id can never be a
// bitmap member, so it correctly fails this inclusion test).
candidates.retain(|id| sg_bitmap.contains(id.as_u64() as u32));
}

347
tidal/src/query/search/executor/pipeline.rs
View File

@ -23,8 +23,7 @@ use roaring::RoaringBitmap;
use super::{ use super::{
super::{ super::{
executor_helpers::{ executor_helpers::{
build_user_context, evaluate_metadata_filter, extract_user_state_filters, build_user_context, evaluate_metadata_filter, has_creator_metadata_filter,
has_creator_metadata_filter,
}, },
scope::ScopeResolver, scope::ScopeResolver,
types::{Search, SearchResultItem, SearchResults}, types::{Search, SearchResultItem, SearchResults},
@ -162,6 +161,9 @@ impl SearchExecutor<'_> {
self.apply_metadata_filter(query, &mut candidates); self.apply_metadata_filter(query, &mut candidates);
self.apply_creator_metadata_filter(query, &mut candidates); self.apply_creator_metadata_filter(query, &mut candidates);
// ── Stage 2.2/2.3/2.4: Deferred Post-Filters ────────────────────────
self.apply_deferred_post_filters(query, &mut candidates)?;
// ── Stage 2.5: User-Context Filtering ──────────────────────────────── // ── Stage 2.5: User-Context Filtering ────────────────────────────────
self.apply_user_context_filter(query, &mut candidates); self.apply_user_context_filter(query, &mut candidates);
@ -438,6 +440,38 @@ impl SearchExecutor<'_> {
} }
/// Stage 2.5: user-context filtering (suppression + inclusion). /// Stage 2.5: user-context filtering (suppression + inclusion).
/// Apply the deferred post-filters (`InCollection`, `MinSignal`/`MaxSignal`,
/// `NearLocation`) that evaluate to the full universe at the
/// `FilterEvaluator` level and so must run as Stage 2.x post-filters. Shared
/// verbatim with the RETRIEVE pipeline via
/// `post_filter::apply_deferred_post_filters` — before SEARCH called this, it
/// silently ignored these filters and returned items that violated them.
///
/// # Errors
///
/// Propagates `QueryError` from the shared applier (e.g. an unknown signal
/// name referenced by a Min/MaxSignal threshold).
fn apply_deferred_post_filters(
&self,
query: &Search,
candidates: &mut Vec<EntityId>,
) -> Result<(), QueryError> {
if let Some(ref filter_expr) = query.combined_filter() {
let ctx = crate::query::executor::post_filter::PostFilterCtx {
ledger: self.ledger,
collection_index: self.collection_index,
items_storage: self.items_storage,
degradation_level: self.degradation_level,
};
crate::query::executor::post_filter::apply_deferred_post_filters(
filter_expr,
candidates,
&ctx,
)?;
}
Ok(())
}
fn apply_user_context_filter(&self, query: &Search, candidates: &mut Vec<EntityId>) { fn apply_user_context_filter(&self, query: &Search, candidates: &mut Vec<EntityId>) {
let Some(user_id) = query.for_user else { let Some(user_id) = query.for_user else {
return; return;
@ -465,9 +499,13 @@ impl SearchExecutor<'_> {
} }
} }
// User-state inclusion filters (Saved, Liked, InProgress). // User-state inclusion filters (Saved, Liked, InProgress, SocialGraph).
// Routed through the canonical RETRIEVE extractor (which includes
// SocialGraph) so both surfaces share one collector and cannot drift.
if let Some(filter_expr) = query.combined_filter() { if let Some(filter_expr) = query.combined_filter() {
for uf in extract_user_state_filters(&filter_expr) { for uf in
crate::query::executor::user_filter::extract_user_state_filters(&filter_expr)
{
match uf { match uf {
FilterExpr::Saved(uid) => { FilterExpr::Saved(uid) => {
let saved = user_state.saved_bitmap(*uid); let saved = user_state.saved_bitmap(*uid);
@ -485,6 +523,23 @@ impl SearchExecutor<'_> {
user_state.is_in_progress(*uid, id.as_u64(), *threshold) user_state.is_in_progress(*uid, id.as_u64(), *threshold)
}); });
} }
FilterExpr::SocialGraph {
user_id: uid,
depth,
} => {
// Constrain to items from the user's social graph
// (followed creators at depth=1, community at
// depth=2). Shared with the RETRIEVE pipeline so
// neither surface silently ignores `social_graph`.
// Before this arm, the `_ => {}` below swallowed it.
crate::query::executor::post_filter::apply_social_graph_filter(
*uid,
*depth,
candidates,
user_state,
self.creator_items,
);
}
_ => {} _ => {}
} }
} }
@ -778,6 +833,40 @@ mod tests {
) )
} }
/// Like [`vector_executor`] but wires the `universe` bitmap. A FILTER clause
/// makes Stage 2's `FilterEvaluator` return `self.universe` for deferred
/// variants (InCollection/MinSignal/…), so the deferred-post-filter tests
/// must populate the universe with the retrieved item ids or the metadata
/// stage would intersect against an empty universe and drop everything.
fn filtered_executor<'a>(
ledger: &'a SignalLedger,
profile_reg: &'a ProfileRegistry,
registry: &'a RwLock<EmbeddingSlotRegistry>,
universe: &'a RwLock<RoaringBitmap>,
) -> SearchExecutor<'a> {
SearchExecutor::new(
ledger,
profile_reg,
None, // text_index
Some(registry), // embedding_registry
None, // category
None, // format
None, // creator
None, // tag
None, // duration
None, // created_at
Some(universe), // universe
)
}
/// Universe bitmap holding the two ANN-retrieved item ids (1 and 2).
fn two_item_universe() -> RwLock<RoaringBitmap> {
let mut bm = RoaringBitmap::new();
bm.insert(1);
bm.insert(2);
RwLock::new(bm)
}
/// CRITICAL regression: the top SEARCH result must be the most *relevant* /// CRITICAL regression: the top SEARCH result must be the most *relevant*
/// item (closest vector), not the most-*liked* one. Before the fix the fused /// item (closest vector), not the most-*liked* one. Before the fix the fused
/// relevance score was discarded and ordering collapsed to signal popularity. /// relevance score was discarded and ordering collapsed to signal popularity.
@ -915,4 +1004,254 @@ mod tests {
assert!(results.items.is_empty()); assert!(results.items.is_empty());
assert!(results.next_cursor.is_none()); assert!(results.next_cursor.is_none());
} }
// ── Deferred post-filter regression coverage ─────────────────────────────
//
// SEARCH previously applied only Saved/Liked/InProgress and silently ignored
// MinSignal/MaxSignal/NearLocation/InCollection/SocialGraph: a well-formed
// SEARCH with those filters returned items that VIOLATED the filter, with no
// error. These tests prove SEARCH now EXCLUDES the violating items, exactly
// like RETRIEVE. Each retrieves two equidistant items via the ANN path and
// asserts the filter drops the one that does not satisfy it.
// `EntityId`, `Timestamp`, `Tag`, `encode_key`, and `FilterExpr` are already
// in scope via `use super::*`; only the entities and storage backend below
// are new for these tests.
use crate::{
entities::{
CollectionId, CollectionIndex, CreatorItemsBitmap, HardNegIndex, InteractionLedger,
UserStateIndex,
},
storage::{StorageEngine, memory::InMemoryBackend},
};
/// Items-only length-prefixed metadata encoding that `db::deserialize_metadata`
/// (used by the NearLocation post-filter) reads back. Mirrors
/// `db::metadata::serialize_metadata`, which is not visible outside `db`.
fn encode_meta(pairs: &[(&str, &str)]) -> Vec<u8> {
let mut buf = Vec::new();
buf.extend_from_slice(&(pairs.len() as u32).to_le_bytes());
for (k, v) in pairs {
buf.extend_from_slice(&(k.len() as u32).to_le_bytes());
buf.extend_from_slice(k.as_bytes());
buf.extend_from_slice(&(v.len() as u32).to_le_bytes());
buf.extend_from_slice(v.as_bytes());
}
buf
}
/// Two equidistant items on the unit vector so the ANN path retrieves both;
/// the deferred filter, not relevance, decides which survives.
fn two_item_registry() -> RwLock<EmbeddingSlotRegistry> {
RwLock::new(registry_with_slot(
"content",
&[(1, [1.0, 0.0]), (2, [1.0, 0.0])],
))
}
fn ids_of(results: &SearchResults) -> Vec<u64> {
let mut v: Vec<u64> = results.items.iter().map(|r| r.entity_id.as_u64()).collect();
v.sort_unstable();
v
}
#[test]
fn search_min_signal_excludes_under_threshold() {
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
// Item 1 has 5 likes (>= 3), item 2 has 1 like (< 3).
let now = Timestamp::now();
for _ in 0..5 {
ledger
.record_signal("like", EntityId::new(1), 1.0, now)
.unwrap();
}
ledger
.record_signal("like", EntityId::new(2), 1.0, now)
.unwrap();
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.filter(FilterExpr::min_signal("like", 3.0))
.limit(10)
.build()
.unwrap();
let results = exec.execute(&query).unwrap();
assert_eq!(
ids_of(&results),
vec![1],
"min_signal(like,3) must drop item 2 (1 like); before the fix both were returned"
);
}
#[test]
fn search_max_signal_excludes_over_threshold() {
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
// Item 1 has 1 view (<= 2), item 2 has 5 views (> 2).
let now = Timestamp::now();
ledger
.record_signal("view", EntityId::new(1), 1.0, now)
.unwrap();
for _ in 0..5 {
ledger
.record_signal("view", EntityId::new(2), 1.0, now)
.unwrap();
}
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.filter(FilterExpr::max_signal("view", 2.0))
.limit(10)
.build()
.unwrap();
let results = exec.execute(&query).unwrap();
assert_eq!(
ids_of(&results),
vec![1],
"max_signal(view,2) must drop item 2 (5 views); before the fix both were returned"
);
}
#[test]
fn search_unknown_signal_threshold_errors_loudly() {
// An unknown signal name in a threshold must fail the query, never
// silently retain or drop every candidate.
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.filter(FilterExpr::min_signal("not_a_signal", 1.0))
.limit(10)
.build()
.unwrap();
let err = exec.execute(&query).expect_err("unknown signal must error");
assert!(matches!(err, QueryError::InvalidFilter { .. }));
}
#[test]
fn search_near_location_excludes_out_of_radius() {
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
// Item 1 in NYC, item 2 in LA.
let storage = InMemoryBackend::new();
let put_geo = |id: u64, lat: &str, lng: &str| {
let key = encode_key(EntityId::new(id), Tag::Meta, b"");
storage
.put(&key, &encode_meta(&[("latitude", lat), ("longitude", lng)]))
.unwrap();
};
put_geo(1, "40.7128", "-74.0060");
put_geo(2, "34.0522", "-118.2437");
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe)
.with_items_storage(&storage);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.filter(FilterExpr::near_location(40.7128, -74.0060, 50.0))
.limit(10)
.build()
.unwrap();
let results = exec.execute(&query).unwrap();
assert_eq!(
ids_of(&results),
vec![1],
"near_location(NYC,50km) must drop item 2 (LA); before the fix both were returned"
);
}
#[test]
fn search_in_collection_excludes_outside_collection() {
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
// Collection 7 contains item 1 only.
let collections = CollectionIndex::new();
let cid = CollectionId::new(7);
collections.add_item(cid, 1);
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe)
.with_collection_index(&collections);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.filter(FilterExpr::in_collection(cid))
.limit(10)
.build()
.unwrap();
let results = exec.execute(&query).unwrap();
assert_eq!(
ids_of(&results),
vec![1],
"in_collection(7) must drop item 2 (not in collection); before the fix both returned"
);
}
#[test]
fn search_social_graph_scopes_to_followed_creator_items() {
let schema = test_schema();
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
let profile_reg = setup_registry();
let registry = two_item_registry();
// User 42 follows creator 10; creator 10 owns item 1 only.
let user_state = UserStateIndex::new();
user_state.add_follow(42, 10);
user_state.add_creator_follower(10, 42);
let creator_items = CreatorItemsBitmap::new();
creator_items.add_item(10, 1);
let hard_neg = HardNegIndex::new();
let interaction = InteractionLedger::new();
let universe = two_item_universe();
let exec = filtered_executor(&ledger, &profile_reg, &registry, &universe)
.with_user_context(&user_state, &hard_neg, &interaction, &creator_items);
let query = Search::builder()
.vector(vec![1.0f32, 0.0])
.using_profile("search")
.for_user(42)
.filter(FilterExpr::social_graph(42, 1))
.limit(10)
.build()
.unwrap();
let results = exec.execute(&query).unwrap();
assert_eq!(
ids_of(&results),
vec![1],
"social_graph(42,1) must drop item 2 (not from a followed creator); before the fix \
the SocialGraph arm was swallowed and both were returned"
);
}
} }

34
tidal/src/query/search/executor_helpers.rs
View File

@ -1,9 +1,14 @@
//! Helper functions extracted from the SEARCH executor to keep file size //! Helper functions extracted from the SEARCH executor to keep file size
//! under the 600-line limit (`CODING_GUIDELINES` §9). //! under the 600-line limit (`CODING_GUIDELINES` §9).
//! //!
//! - User-state filter extraction helpers
//! - `build_user_context` free function (was a private method on `SearchExecutor`) //! - `build_user_context` free function (was a private method on `SearchExecutor`)
//! - Creator metadata filter predicates //! - Creator metadata filter predicates
//!
//! User-state filter extraction lives in
//! [`crate::query::executor::user_filter::extract_user_state_filters`]; both the
//! RETRIEVE and SEARCH pipelines call that single collector so they cannot
//! drift on which user-state variants (Saved/Liked/InProgress/SocialGraph) are
//! recognized.
use std::collections::HashMap; use std::collections::HashMap;
@ -14,33 +19,6 @@ use crate::{
storage::indexes::filter::FilterExpr, storage::indexes::filter::FilterExpr,
}; };
// ── User-state filter extraction ─────────────────────────────────────────────
fn collect_user_state_filters<'a>(expr: &'a FilterExpr, out: &mut Vec<&'a FilterExpr>) {
match expr {
FilterExpr::Saved(_) | FilterExpr::Liked(_) | FilterExpr::InProgress { .. } => {
out.push(expr);
}
FilterExpr::And(children) | FilterExpr::Or(children) => {
for child in children {
collect_user_state_filters(child, out);
}
}
// Do NOT recurse into Not: a deferred variant under Not has no positive
// application. Such queries are rejected up front by
// `crate::query::executor::user_filter::reject_negated_deferred_filters`,
// so extracting the inner variant here would invert the requested set.
// `Not` and every other node contribute nothing here.
_ => {}
}
}
pub(super) fn extract_user_state_filters(expr: &FilterExpr) -> Vec<&FilterExpr> {
let mut result = Vec::new();
collect_user_state_filters(expr, &mut result);
result
}
// ── UserContext builder ─────────────────────────────────────────────────────── // ── UserContext builder ───────────────────────────────────────────────────────
/// Build a `UserContext` for personalized scoring. /// Build a `UserContext` for personalized scoring.

57
tidal/src/storage/vector/brute/mod.rs
View File

@ -63,6 +63,7 @@ pub fn l2_distance_sq(a: &[f32], b: &[f32]) -> f32 {
/// All reads take a shared lock; all writes take an exclusive lock. This is /// All reads take a shared lock; all writes take an exclusive lock. This is
/// acceptable because brute-force is not a hot-path production index -- it is /// acceptable because brute-force is not a hot-path production index -- it is
/// used for correctness baselines, small datasets, and tests. /// used for correctness baselines, small datasets, and tests.
#[derive(Debug)]
pub struct BruteForceIndex { pub struct BruteForceIndex {
vectors: RwLock<HashMap<VectorId, Vec<f32>>>, vectors: RwLock<HashMap<VectorId, Vec<f32>>>,
config: VectorIndexConfig, config: VectorIndexConfig,
@ -268,12 +269,57 @@ impl BruteForceIndex {
))); )));
} }
// Read count // Read count.
//
// SECURITY: `count` comes straight off disk and is fully
// corruption/attacker-controlled. Every byte of arithmetic below uses
// checked operations and is bounded against the *actual* buffer length
// BEFORE any allocation or loop. Without this, a hostile `count` would
// (a) overflow `count * bytes_per_vector` in release builds, wrapping to
// a tiny `expected_size` that slips past the truncation guard, then
// (b) drive `HashMap::with_capacity(count)` into a multi-exabyte alloc
// (process abort) and an out-of-bounds read loop (panic). A corrupt
// *derived* index is recoverable (rebuildable from the entity store), so
// it must surface as `CorruptedIndex`, never crash the recovery path.
let count = read_u64_le(data, 9) as usize; let count = read_u64_le(data, 9) as usize;
// Validate total size // Bytes consumed per stored vector: id (8) + floats (dims * 4).
let bytes_per_vector = 8 + dims * 4; // id (8) + floats (dims * 4) // Guard `dims` (also from disk-ish via header; already matched against
let expected_size = HEADER_SIZE + count * bytes_per_vector; // config above, but compute defensively) so the multiply cannot wrap.
let float_bytes = dims
.checked_mul(4)
.ok_or_else(|| VectorError::CorruptedIndex(format!("dimensions too large: {dims}")))?;
let bytes_per_vector = 8usize.checked_add(float_bytes).ok_or_else(|| {
VectorError::CorruptedIndex(format!("per-vector size overflow at dims {dims}"))
})?;
debug_assert!(
bytes_per_vector >= 8,
"bytes_per_vector includes the 8-byte id and cannot be zero"
);
// Cap `count` by how many vectors the body could actually contain. This
// makes the subsequent `with_capacity` and read loop safe regardless of
// the on-disk value: a damaged header can never request more entries
// than the buffer can hold.
let body_len = data.len() - HEADER_SIZE; // safe: data.len() >= HEADER_SIZE checked above
let max_count = body_len / bytes_per_vector; // bytes_per_vector >= 8, no div-by-zero
if count > max_count {
return Err(VectorError::CorruptedIndex(format!(
"implausible vector count: header claims {count}, but the {body_len}-byte \
body holds at most {max_count} vectors of {bytes_per_vector} bytes each"
)));
}
// Validate the exact declared size (now provably non-overflowing because
// `count <= max_count` and `max_count * bytes_per_vector <= body_len`).
let expected_size = count
.checked_mul(bytes_per_vector)
.and_then(|body| body.checked_add(HEADER_SIZE))
.ok_or_else(|| {
VectorError::CorruptedIndex(format!(
"index size overflow: count {count} * {bytes_per_vector} + {HEADER_SIZE}"
))
})?;
if data.len() < expected_size { if data.len() < expected_size {
return Err(VectorError::CorruptedIndex(format!( return Err(VectorError::CorruptedIndex(format!(
"file truncated: expected at least {expected_size} bytes, got {}", "file truncated: expected at least {expected_size} bytes, got {}",
@ -281,7 +327,8 @@ impl BruteForceIndex {
))); )));
} }
// Parse vectors // Parse vectors. `count` is now bounded by the buffer, so `with_capacity`
// cannot over-allocate and the read loop cannot index out of bounds.
let mut vectors = HashMap::with_capacity(count); let mut vectors = HashMap::with_capacity(count);
let mut offset = HEADER_SIZE; let mut offset = HEADER_SIZE;
for _ in 0..count { for _ in 0..count {

113
tidal/src/storage/vector/brute/tests.rs
View File

@ -262,6 +262,119 @@ fn vector_index_config_defaults() {
assert_eq!(config.ef_search, 200); assert_eq!(config.ef_search, 200);
} }
// -----------------------------------------------------------------------
// Corruption / hostile-input deserialization
// -----------------------------------------------------------------------
/// Build a minimal valid 17-byte brute-force index header for `dims`
/// dimensions claiming `count` stored vectors. The returned buffer contains
/// *only* the header -- callers append (or omit) body bytes to construct
/// truncated or torn inputs.
fn make_header(dims: u32, count: u64) -> Vec<u8> {
let mut buf = Vec::with_capacity(17);
buf.extend_from_slice(MAGIC); // [0..4]
buf.push(FORMAT_VERSION); // [4]
buf.extend_from_slice(&dims.to_le_bytes()); // [5..9]
buf.extend_from_slice(&count.to_le_bytes()); // [9..17]
buf
}
#[test]
fn deserialize_rejects_enormous_count_without_panicking() {
// A header claiming u64::MAX vectors with no body. Pre-fix this wrapped
// `count * bytes_per_vector`, slipped past the truncation guard, then
// aborted in `HashMap::with_capacity` / panicked in the read loop.
let config = test_config(); // dims = 3
let data = make_header(config.dimensions as u32, u64::MAX);
let result = BruteForceIndex::deserialize(&data, &config);
assert!(
matches!(result, Err(VectorError::CorruptedIndex(_))),
"enormous count must be a recoverable CorruptedIndex, got {result:?}"
);
}
#[test]
fn deserialize_rejects_count_exceeding_body() {
// Header claims 1000 vectors but supplies a body big enough for only one.
let config = test_config(); // dims = 3 => bytes_per_vector = 8 + 12 = 20
let mut data = make_header(config.dimensions as u32, 1000);
data.extend_from_slice(&[0u8; 20]); // exactly one vector's worth of body
let result = BruteForceIndex::deserialize(&data, &config);
match result {
Err(VectorError::CorruptedIndex(msg)) => {
assert!(
msg.contains("implausible vector count"),
"expected count-cap message, got: {msg}"
);
}
other => panic!("expected CorruptedIndex, got {other:?}"),
}
}
#[test]
fn deserialize_rejects_torn_body() {
// Header honestly claims 2 vectors (40 bytes of body) but the body is torn
// mid-vector (only 25 bytes present). This must be rejected with a typed
// CorruptedIndex error, never read OOB. The count-cap guard catches it first:
// a 25-byte body holds at most 1 whole 20-byte vector, so a claimed count of 2
// is already implausible — which is exactly the bound that prevents the OOB
// read the torn tail would otherwise cause.
let config = test_config(); // dims = 3 => bytes_per_vector = 20
let mut data = make_header(config.dimensions as u32, 2);
data.extend_from_slice(&[0u8; 25]); // 25 < 40: torn second vector
let result = BruteForceIndex::deserialize(&data, &config);
match result {
Err(VectorError::CorruptedIndex(msg)) => {
assert!(
msg.contains("implausible vector count") || msg.contains("truncated"),
"expected a size/count corruption message, got: {msg}"
);
}
other => panic!("expected CorruptedIndex on torn body, got {other:?}"),
}
}
#[test]
fn deserialize_rejects_overflowing_dimensions() {
// A dims value whose `* 4` overflows `usize` must be rejected, not panic.
// We point `config.dimensions` at the same huge value so the dimension
// check passes and we reach the per-vector size arithmetic.
let huge_dims = usize::MAX;
let config = VectorIndexConfig {
dimensions: huge_dims,
..test_config()
};
// Header stores dims as a u32, so truncation means the on-file dims won't
// equal `huge_dims`; that mismatch is itself a CorruptedIndex, which is
// still the correct (non-panicking) outcome. Assert no panic + an error.
let data = make_header(huge_dims as u32, 1);
let result = BruteForceIndex::deserialize(&data, &config);
assert!(
matches!(result, Err(VectorError::CorruptedIndex(_))),
"overflowing dims must yield CorruptedIndex, got {result:?}"
);
}
#[test]
fn deserialize_accepts_well_formed_buffer() {
// Round-trip sanity: a buffer produced by `save` with a real body still
// deserializes after the hardening. Guards the fix against over-rejection.
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("ok.bfvi");
let config = test_config();
let index = BruteForceIndex::new(config.clone());
index.insert(7, &[1.0, 2.0, 3.0]).unwrap();
index.insert(8, &[4.0, 5.0, 6.0]).unwrap();
index.save(&path).unwrap();
let data = std::fs::read(&path).unwrap();
let loaded = BruteForceIndex::deserialize(&data, &config).unwrap();
assert_eq!(loaded.len(), 2);
}
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
// Property tests // Property tests
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------

136
tidal/src/wal/compaction.rs
View File

@ -26,6 +26,19 @@ pub struct CompactionResult {
/// checkpoint; it is NOT deleted (conservative: we replay a few extra events /// checkpoint; it is NOT deleted (conservative: we replay a few extra events
/// rather than risk losing uncovered events). /// rather than risk losing uncovered events).
/// ///
/// # Live-writer hazard — use [`compact_wal_online`] instead while the WAL is open
///
/// This function does NOT account for the segment a running writer thread is
/// actively appending to. The writer holds one segment open and only rotates at
/// the size threshold, so after any burst the live segment's `first_seq` is well
/// below the materialized checkpoint — and this function would `remove_file` it
/// out from under the writer's open FD. On Linux the writer keeps appending to
/// the now-unlinked inode and every post-checkpoint, already-fsync'd write is
/// silently lost on the next open. Therefore this entry point is only safe to
/// call once the writer thread has been joined (i.e. after
/// `WalHandle::shutdown()`); the periodic-checkpoint path must use
/// [`compact_wal_online`].
///
/// # Safety invariant /// # Safety invariant
/// ///
/// The caller must ensure the checkpoint at `checkpoint_seq` has been durably /// The caller must ensure the checkpoint at `checkpoint_seq` has been durably
@ -48,21 +61,71 @@ pub struct CompactionResult {
/// if an error is encountered mid-way -- this is safe (see invariant above). /// if an error is encountered mid-way -- this is safe (see invariant above).
pub fn compact_wal(wal_dir: &Path, checkpoint_seq: u64) -> Result<CompactionResult, WalError> { pub fn compact_wal(wal_dir: &Path, checkpoint_seq: u64) -> Result<CompactionResult, WalError> {
let segments = segment::list_segments(wal_dir)?; let segments = segment::list_segments(wal_dir)?;
compact_segments(wal_dir, &segments, checkpoint_seq)
}
/// Online-safe compaction: identical to [`compact_wal`] except it NEVER deletes
/// the segment the WAL writer thread is (or may be) actively appending to.
///
/// `list_segments` returns segments sorted by ascending `first_seq`, and the
/// writer always appends to — and only ever rotates to a strictly *higher* —
/// the segment with the largest `first_seq` (see `WalHandle::open`, which opens
/// `segments.last()`). So the live segment is always the maximum-`first_seq`
/// segment on disk. We clamp the deletion floor to that segment's `first_seq`,
/// guaranteeing it survives even when the checkpoint has advanced past its
/// `first_seq` — the common case, since the active segment starts before a
/// checkpoint but still holds uncheckpointed tail events. Deleting it would
/// unlink the inode out from under the writer's open FD and, on Linux, silently
/// lose every post-checkpoint, already-fsync'd write that followed.
///
/// Safe to run concurrently with the writer: a rotation during compaction only
/// creates an even-higher segment (untouched here) and seals the old active one
/// (now safely deletable on the *next* compaction) — never a lost write.
///
/// Use this from the periodic-checkpoint path while the writer is live; use
/// [`compact_wal`] only after `WalHandle::shutdown()` has joined the writer.
///
/// # Errors
///
/// Returns `WalError::Io` on filesystem failure.
pub fn compact_wal_online(
wal_dir: &Path,
checkpoint_seq: u64,
) -> Result<CompactionResult, WalError> {
let segments = segment::list_segments(wal_dir)?;
// The live segment is the maximum-`first_seq` segment; never delete at or
// above it. Clamp the floor so the writer's open segment is always preserved.
let floor = match segments.iter().map(|(first_seq, _)| *first_seq).max() {
Some(active_first_seq) => checkpoint_seq.min(active_first_seq),
None => checkpoint_seq,
};
compact_segments(wal_dir, &segments, floor)
}
/// Delete every listed segment whose `first_seq` is strictly less than `floor`,
/// then fsync the directory so the unlinks are durable.
///
/// `floor` is the exclusive lower bound: a segment is kept iff
/// `first_seq >= floor`. [`compact_wal`] passes `checkpoint_seq` directly;
/// [`compact_wal_online`] passes `min(checkpoint_seq, active_first_seq)` to
/// protect the live segment.
fn compact_segments(
wal_dir: &Path,
segments: &[(u64, std::path::PathBuf)],
floor: u64,
) -> Result<CompactionResult, WalError> {
let total_before = segments.len(); let total_before = segments.len();
let mut deleted = 0usize; let mut deleted = 0usize;
let mut bytes_reclaimed = 0u64; let mut bytes_reclaimed = 0u64;
for (seg_first_seq, seg_path) in &segments { for (seg_first_seq, seg_path) in segments {
// Only delete segments whose first_seq is strictly less than the if *seg_first_seq < floor {
// checkpoint sequence. Segments starting at or after checkpoint_seq
// may contain events that are not yet covered by the checkpoint.
if *seg_first_seq < checkpoint_seq {
// SAFETY: TOCTOU gap between metadata() and remove_file() is benign. // SAFETY: TOCTOU gap between metadata() and remove_file() is benign.
// `list_segments()` only returns files that exist on disk at scan time. // `list_segments()` only returns files that exist on disk at scan time.
// If another process deleted this segment between the scan and now, // If another process deleted this segment between the scan and now,
// remove_file() returns an error which compact_wal propagates -- but // remove_file() returns an error which propagates -- but TidalDb
// TidalDb enforces single-process-per-directory ownership (see Config::lock_dir), // enforces single-process-per-directory ownership (see Config::lock_dir),
// so this cannot happen in production. The gap is a safe no-op under the // so this cannot happen in production. The gap is a safe no-op under the
// single-owner invariant. // single-owner invariant.
let file_size = std::fs::metadata(seg_path).map(|m| m.len()).unwrap_or(0); let file_size = std::fs::metadata(seg_path).map(|m| m.len()).unwrap_or(0);
@ -93,7 +156,7 @@ pub fn compact_wal(wal_dir: &Path, checkpoint_seq: u64) -> Result<CompactionResu
deleted, deleted,
bytes_reclaimed, bytes_reclaimed,
remaining, remaining,
checkpoint_seq, floor,
"WAL compaction complete" "WAL compaction complete"
); );
@ -229,4 +292,61 @@ mod tests {
assert_eq!(result.segments_deleted, 1); // seq=1 < 100 assert_eq!(result.segments_deleted, 1); // seq=1 < 100
assert_eq!(result.segments_remaining, 1); // seq=100 preserved assert_eq!(result.segments_remaining, 1); // seq=100 preserved
} }
#[test]
fn online_never_deletes_active_segment_even_when_fully_checkpointed() {
// BLOCKER regression: the live (max-first_seq) segment must survive even
// when the checkpoint has advanced far past its first_seq — otherwise the
// writer's open FD points at an unlinked inode and acknowledged, fsync'd
// writes are silently lost on Linux.
let dir = tempfile::tempdir().unwrap();
for &seq in &[1u64, 100] {
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, seq, 1024).unwrap();
}
// Checkpoint way past both segments. Shutdown-compaction would delete both;
// online compaction must preserve the active segment (first_seq = 100).
let result = compact_wal_online(dir.path(), 1_000).unwrap();
assert_eq!(
result.segments_deleted, 1,
"only the non-active old segment"
);
assert_eq!(result.segments_remaining, 1);
let remaining = list_segments(dir.path()).unwrap();
assert_eq!(remaining[0].0, 100, "active segment preserved");
}
#[test]
fn online_preserves_single_active_segment() {
let dir = tempfile::tempdir().unwrap();
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, 50, 1024).unwrap();
// Even with a checkpoint past it, the sole (active) segment is never deleted.
let result = compact_wal_online(dir.path(), 1_000).unwrap();
assert_eq!(result.segments_deleted, 0);
assert_eq!(result.segments_remaining, 1);
}
#[test]
fn online_matches_shutdown_compaction_for_older_segments() {
// When older segments sit below both the checkpoint and the active
// segment, online compaction reclaims exactly the same ones.
let dir = tempfile::tempdir().unwrap();
for &seq in &[1u64, 50, 100, 200] {
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, seq, 1024).unwrap();
}
// floor = min(100, max_first_seq=200) = 100 -> delete first_seq < 100.
let result = compact_wal_online(dir.path(), 100).unwrap();
assert_eq!(result.segments_deleted, 2);
assert_eq!(result.segments_remaining, 2);
let remaining = list_segments(dir.path()).unwrap();
assert_eq!(remaining[0].0, 100);
assert_eq!(remaining[1].0, 200);
}
#[test]
fn online_empty_dir_is_noop() {
let dir = tempfile::tempdir().unwrap();
let result = compact_wal_online(dir.path(), 100).unwrap();
assert_eq!(result.segments_deleted, 0);
assert_eq!(result.segments_remaining, 0);
}
} }

28
tidal/src/wal/dedup.rs
View File

@ -81,6 +81,34 @@ impl DedupWindow {
false false
} }
/// Check whether `event` is already in the dedup window WITHOUT recording it.
///
/// Used by the WAL writer to split duplicate *detection* from dedup-state
/// *insertion*. An event must only be marked as seen once its batch is
/// durably persisted (via [`record`](Self::record)); otherwise a batch that
/// fails to flush and is then retried would be silently suppressed as a
/// phantom duplicate — a lost, acknowledged-intent write. Rotates the buffers
/// by time/size exactly like [`is_duplicate`](Self::is_duplicate).
pub fn contains(&mut self, event: &EventRecord) -> bool {
self.maybe_rotate();
let hash = event_content_hash(event);
self.current.contains(&hash) || self.previous.contains(&hash)
}
/// Record `event` as seen.
///
/// Call this ONLY after the batch containing `event` has been durably
/// fsynced, so a transient flush failure leaves the event un-recorded and a
/// retry is accepted rather than dropped. Honours the per-window size cap
/// like the rest of the window (forces an early rotation at the cap).
pub fn record(&mut self, event: &EventRecord) {
if self.current.len() >= self.max_entries {
self.force_rotate();
}
let hash = event_content_hash(event);
self.current.insert(hash);
}
/// Populate the dedup window from replayed events during recovery. /// Populate the dedup window from replayed events during recovery.
/// ///
/// This inserts all provided events into the current window so that /// This inserts all provided events into the current window so that

62
tidal/src/wal/writer.rs
View File

@ -266,9 +266,20 @@ fn partition_dedup(
) { ) {
let mut kept_events: Vec<EventRecord> = Vec::new(); let mut kept_events: Vec<EventRecord> = Vec::new();
let mut kept_replies: Vec<crossbeam::channel::Sender<Result<u64, WalError>>> = Vec::new(); let mut kept_replies: Vec<crossbeam::channel::Sender<Result<u64, WalError>>> = Vec::new();
// Hashes kept *within this batch*, so two identical events in the same drained
// batch still dedup against each other even though the durable dedup window is
// only updated AFTER a successful flush (see `run_writer`). This preserves
// intra-batch suppression without violating the "mark as seen only once
// durable" invariant.
let mut batch_seen: std::collections::HashSet<u128> = std::collections::HashSet::new();
for (event, reply) in batch { for (event, reply) in batch {
if dedup.is_duplicate(&event) { let hash = format::event_content_hash(&event);
// `dedup.contains` CHECKS membership without recording; `batch_seen.insert`
// returns false when this exact event already appeared earlier in this
// batch. Recording into `dedup` happens in `run_writer`, only after the
// batch is durably persisted.
if dedup.contains(&event) || !batch_seen.insert(hash) {
// Duplicate: notify with the dedup sentinel (seq=0) immediately. // Duplicate: notify with the dedup sentinel (seq=0) immediately.
let _ = reply.send(Ok(0)); let _ = reply.send(Ok(0));
} else { } else {
@ -295,13 +306,22 @@ fn partition_dedup(
/// 3. Deduplicate events, encode the batch, write to segment, fsync. /// 3. Deduplicate events, encode the batch, write to segment, fsync.
/// 4. Send sequence numbers back to all waiting callers. /// 4. Send sequence numbers back to all waiting callers.
/// ///
/// # Resilience
///
/// A flush failure in the **steady-state loop** does NOT terminate the writer.
/// `flush_batch` notifies every waiting caller with the error (so they can
/// retry), the batch's events are left un-recorded in the dedup window (so a
/// retry is accepted, not suppressed as a phantom duplicate), and the loop keeps
/// serving — a transient I/O fault (ENOSPC, EINTR, NFS blip) must never convert
/// into a permanent write outage by dropping the command channel.
///
/// # Errors /// # Errors
/// ///
/// Returns `WalError::Io` on filesystem failure during batch writes or fsync. /// Only the shutdown drain / final fsync propagate an error out of this function
/// Returns `WalError::Corruption` if batch encoding fails (should not happen /// (the WAL is closing anyway, and callers were already notified). The encoding
/// under normal operation — `effective_batch_size` clamps the drain limit to /// path cannot fail under normal operation — `effective_batch_size` clamps the
/// `MAX_EVENTS_PER_BATCH` so an oversized `batch_size` can never produce an /// drain limit to `MAX_EVENTS_PER_BATCH` so an oversized `batch_size` can never
/// unencodable batch). /// produce an unencodable batch.
/// ///
/// # Panics /// # Panics
/// ///
@ -402,7 +422,35 @@ pub fn run_writer(
let (kept_events, kept_replies) = partition_dedup(&mut dedup, batch.drain(..)); let (kept_events, kept_replies) = partition_dedup(&mut dedup, batch.drain(..));
if !kept_events.is_empty() { if !kept_events.is_empty() {
next_seq = flush_batch(&mut segment, config, next_seq, &kept_events, kept_replies)?; match flush_batch(&mut segment, config, next_seq, &kept_events, kept_replies) {
Ok(seq) => {
next_seq = seq;
// Record events as seen ONLY now that the batch is durably
// persisted. Recording earlier (in `partition_dedup`) would
// suppress a legitimate retry of a batch that failed to flush.
for event in &kept_events {
dedup.record(event);
}
}
Err(e) => {
// A transient I/O fault (a brief ENOSPC that an operator then
// clears, an EINTR, an NFS/network-storage blip) must NOT tear
// down the writer thread — that would drop the command channel
// and wedge every future write forever, since the WAL is the
// source of truth for all entity/signal/relationship writes.
// `flush_batch` has already notified each waiting caller with
// the error, so they can retry; we keep `next_seq` unchanged
// (the failed batch's sequence range is free for the retry),
// leave the events UNRECORDED in the dedup window so the retry
// is accepted, and keep serving.
tracing::error!(
error = %e,
seq = next_seq,
events = kept_events.len(),
"wal: batch flush failed; callers notified to retry, writer continuing"
);
}
}
} }
if shutdown_requested { if shutdown_requested {

75
tidal/src/wal/writer_tests.rs
View File

@ -298,6 +298,81 @@ fn shutdown_drain_notifies_callers_on_write_error() {
); );
} }
#[test]
fn steady_state_flush_error_does_not_kill_writer_and_allows_retry() {
// BLOCKER + wal-3 regression: a transient flush failure in the STEADY-STATE
// loop must NOT terminate the writer thread, and the failed event must NOT be
// recorded in the dedup window — so an identical retry is accepted and
// durably persisted (a real, non-zero seq), never silently swallowed as a
// duplicate (the seq=0 sentinel).
//
// batch_size = 1 makes each event its own batch, so the retry lands in a
// SEPARATE flush from the failed first attempt while we drive run_writer
// synchronously on this thread (where the thread-local fault hook applies).
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let (tx, rx) = bounded(16);
let segment = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 16 * 1024 * 1024)
.expect("open should succeed");
let dedup = DedupWindow::new(Duration::from_secs(30));
let config = WriterConfig {
dir: dir.path().to_path_buf(),
segment_size: 16 * 1024 * 1024,
batch_size: 1, // one event per batch, so attempt and retry flush separately
batch_timeout: Duration::from_millis(10),
dedup_window: Duration::from_secs(30),
session_journal_path: None,
shard_id: ShardId::SINGLE,
region_id: RegionId::SINGLE,
};
let event = make_event(99);
// First attempt: will hit the injected one-shot flush failure.
let (reply_tx1, reply_rx1) = bounded(1);
tx.send(WalCommand::Append {
event: event.clone(),
reply: reply_tx1,
})
.expect("send should succeed");
// Retry of the SAME event: must be accepted (not deduped) and persisted.
let (reply_tx2, reply_rx2) = bounded(1);
tx.send(WalCommand::Append {
event,
reply: reply_tx2,
})
.expect("send should succeed");
tx.send(WalCommand::Shutdown).expect("send should succeed");
drop(tx);
// Arm the one-shot fault on THIS thread, then run the writer synchronously
// here so the fault fires inside the first steady-state flush_batch.
arm_flush_failure();
let result = run_writer(&rx, &config, segment, 1, dedup);
// BLOCKER: the writer survived the transient error and exited cleanly.
assert!(
result.is_ok(),
"writer must survive a transient steady-state flush error, got {result:?}"
);
// First attempt observed the error (so the caller knows to retry).
let r1 = reply_rx1.recv().expect("first caller must be notified");
assert!(
matches!(r1, Err(WalError::Io(_))),
"first attempt must see the flush error, got {r1:?}"
);
// wal-3: the retry was NOT suppressed as a duplicate — it got a real,
// non-zero sequence number, meaning it was durably persisted.
let r2 = reply_rx2
.recv()
.expect("retry caller must be notified")
.expect("retry must persist successfully");
assert!(
r2 > 0,
"retry must get a real sequence number, not the dedup sentinel 0; got {r2}"
);
}
#[test] #[test]
fn writer_assigns_monotonic_sequences() { fn writer_assigns_monotonic_sequences() {
let dir = tempfile::tempdir().expect("tempdir creation should succeed"); let dir = tempfile::tempdir().expect("tempdir creation should succeed");