6385425a92
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
`score_hot` hardcoded `age_hours = 24.0`, so the divisor in `log10(max(views,1)) / (age_hours + 2)^gravity` was constant across the candidate set and `Sort::Hot` reduced EXACTLY to `log10(max(views, 1))` -- a view-count ranking wearing a recency sort's name. Four built-in profiles use it (`hot`, `for_you`, `following`, `brief`); anyone tuning `gravity` was tuning a no-op. The in-code comment justified this by saying a per-entity `created_at` lookup needs an `EntityId -> created_at_ns` reverse map that "is not built". That was stale, and it was the load-bearing claim: `created_at` has been materialized INTO item metadata on every write since `Items::metadata_with_created_at`, the executor has held an `EntityId -> metadata` map since M6p3, and the replication record carries the materialized map so replicas cannot diverge. No index, storage change, schema change or migration -- the scorer reads the map it already had, exactly the way `read_duration` does three lines away. `Sort::New` used `entity_id as f64`. Wrong twice: it assumed IDs are assigned in creation order, and it used the ID's MAGNITUDE as the base score, so on a catalog of N items the sort contributed ~N against a boost sum in single digits. Recency did not participate in the ranking, it annihilated every boost. Now negated age in hours -- same ordering, boost-comparable scale. Three more instances of the same defect class, found by auditing rather than assuming the report was complete: 1. Both age sorts were missing from `needs_metadata_for_sort`, so a profile with no session and no diversity never loaded the map the fix depends on. 2. Every metadata sort was DEAD on the SEARCH path. Its metadata pre-load was gated on `session_context.is_some()` and never consulted `profile.sort`, AND the `ProfileExecutor` it built never had `with_item_metadata` called at all -- the map it did compute went only to the keyword-hint argument, which the sort scorers do not read. `shortest`/`longest` scored NEG_INFINITY and the alphabetical sorts the missing-title sentinel, for every candidate, silently. 3. Under `ReducedCandidates` load the candidate cap kept the highest entity IDs, correct only while `Sort::New` meant "highest ID". Left alone it would discard the genuinely newest items BEFORE scoring -- wrong only when degraded, the hardest case to notice. Now keyed off the `created_at` index via the new `RangeIndex::top_n_descending`. The decision "which sorts read item metadata" now lives on `Sort` itself as an exhaustive match. It was a `matches!` in one executor while a second executor had its own different copy, which is precisely how a metadata-reading sort came to be omitted from both. MEASURED, not inferred: - Real server, 10 items, equal views, ages 2-20 days: before every score was 0.5 (all-equal set folded to the normalizer's midpoint) and the feed returned oldest-first forever; after, 1.0 -> 0.0 strictly descending, newest first. - `new` with zero signals returns the exact REVERSE of candidate-scan order. - `alphabetical_asc`, `shortest`, `longest` verified end to end with title and duration order both opposing entity id. - Metadata point-read cost at 2,000 candidates (the ceiling: `scan_candidates` caps at `max(limit*10, 200)` and `limit > 500` is rejected): 7.25ms, 3.6us per candidate. Guarded at 250ms. THE BUG REPORT'S CENTRAL PROMISE IS FALSE and the changelog says so. §7 claimed this fix lets a zero-signal corpus rank newest-first so a consumer could delete its workaround. It arithmetically cannot: the numerator `log10(max(views,1))` is exactly 0.0 for 0 OR 1 views, so the age divisor has nothing to scale and every candidate still ties -- confirmed on the live server, all ten scores 0.5. Age-awareness begins at the second view. Fixing cold-start needs recency to be ADDITIVE rather than a pure divisor, which reorders every existing Hot consumer, so it is a separate decision. `sort_hot_zero_view_corpus_still_ties_regardless_of_ age` pins the limit so it cannot be rediscovered by accident. Three existing tests asserted the old entity-ID behaviour. Inverted to assert real recency, not loosened -- and each fixture now makes id order and creation order DISAGREE, because an ordering assertion where the two candidate orderings agree is satisfied by the defect too. Three of my own new tests were vacuous for exactly that reason and were caught by mutation-testing; one was also flaky (it passed in a 12-test run and failed run alone, because retrieval order for exactly-tied vectors is not deterministic). Every new assertion is mutation-proven against the implementation it replaces. Full lib suite 2130 passed. Clippy 66 warnings vs 66 at baseline, zero added.
965 lines
46 KiB
Rust
965 lines
46 KiB
Rust
//! RETRIEVE query execution pipeline.
|
||
//!
|
||
//! Contains the `execute()` method that drives the 6-stage pipeline.
|
||
//! Stage 3 (signal scoring) is delegated to `stage3_score` in `mod.rs`.
|
||
|
||
#![allow(clippy::too_many_lines)]
|
||
|
||
use std::{
|
||
collections::{HashMap, HashSet},
|
||
time::Instant,
|
||
};
|
||
|
||
use super::{RetrieveExecutor, candidate_gen, post_filter, user_filter};
|
||
use crate::{
|
||
query::{
|
||
retrieve::{Cursor, QueryError, Results, Retrieve, RetrieveResult, Signal},
|
||
stats::QueryStats,
|
||
},
|
||
ranking::{
|
||
diversity::DiversitySelector,
|
||
profile::{CandidateStrategy, Sort},
|
||
},
|
||
schema::{EntityId, Timestamp},
|
||
storage::indexes::{
|
||
bitmap::BitmapIndex,
|
||
filter::{FilterEvaluator, FilterResult},
|
||
range::RangeIndex,
|
||
},
|
||
};
|
||
|
||
/// ANN candidate over-fetch multiplier (m12p2): fetch `limit * this` nearest
|
||
/// neighbours so Stage 2/2.5 filtering and Stage 4 diversity can trim back to
|
||
/// `limit` without starving. Mirrors the SEARCH ANN over-fetch.
|
||
const ANN_OVERFETCH: usize = 10;
|
||
|
||
/// Minimum ANN candidate count regardless of `limit`, so a tiny-limit feed still
|
||
/// seeds a usable candidate pool.
|
||
const ANN_CANDIDATE_FLOOR: usize = 200;
|
||
|
||
impl RetrieveExecutor<'_> {
|
||
/// Partition `candidates` so the `cap` NEWEST entities occupy `[0, cap)`.
|
||
///
|
||
/// Keyed off the `created_at` range index, read newest-first, then intersected
|
||
/// with the live candidate set — the index covers the whole universe while
|
||
/// `candidates` may already be narrowed, so an id from the index is kept only
|
||
/// if it is actually a candidate. Anything the index does not cover keeps its
|
||
/// relative order behind the ranked prefix.
|
||
///
|
||
/// Falls back to the historical descending-entity-ID partition when no
|
||
/// `created_at` index is wired: at this point in the pipeline item metadata has
|
||
/// not been loaded, so there is no other recency key available, and an
|
||
/// approximate survivor set beats discarding the newest items outright.
|
||
fn truncate_to_newest(&self, candidates: &mut [EntityId], cap: usize) {
|
||
let Some(index) = self.created_at_index else {
|
||
// No recency key available — preserve the documented approximation.
|
||
candidates.select_nth_unstable_by(cap - 1, |a, b| b.as_u64().cmp(&a.as_u64()));
|
||
return;
|
||
};
|
||
// Pull more than `cap` because the index spans the full universe and some
|
||
// of its newest entities may have been filtered out of `candidates`
|
||
// already. Bounded at 4x so a heavily-filtered query cannot walk the whole
|
||
// tree; if the oversample still comes up short the remainder is filled
|
||
// from the untouched tail below, which is the same set a blind truncate
|
||
// would have kept.
|
||
let newest = index.top_n_descending(cap.saturating_mul(4));
|
||
let mut rank: HashMap<u64, usize> = HashMap::with_capacity(newest.len());
|
||
for (pos, id) in newest.iter().enumerate() {
|
||
rank.entry(u64::from(*id)).or_insert(pos);
|
||
}
|
||
// Stable partition: ranked entities first in recency order, everything the
|
||
// index did not cover after them in its existing order.
|
||
candidates.sort_by_key(|eid| rank.get(&eid.as_u64()).copied().unwrap_or(usize::MAX));
|
||
}
|
||
|
||
/// Execute a RETRIEVE query through the 6-stage pipeline.
|
||
///
|
||
/// # Errors
|
||
///
|
||
/// Returns `QueryError` on validation failure, missing profile, or
|
||
/// unsupported candidate strategy.
|
||
#[allow(clippy::too_many_lines)]
|
||
pub fn execute(&self, query: &Retrieve) -> Result<Results, QueryError> {
|
||
let query_start = Instant::now();
|
||
let mut stats = QueryStats::new(query.profile.name.clone());
|
||
stats.degradation_level = self.degradation_level as u8;
|
||
|
||
// Validate the query against the profile registry.
|
||
query.validate(self.profile_registry)?;
|
||
|
||
// Compute the combined filter ONCE and reuse it across every stage.
|
||
// `combined_filter()` clones the entire `filters` Vec into a fresh
|
||
// `And` node on each call, and the pipeline consults it at 6+ points.
|
||
let combined_filter = query.combined_filter();
|
||
|
||
// Reject filter shapes with no sound semantics BEFORE any stage runs, so
|
||
// a bad query fails loudly instead of silently returning the inverse set
|
||
// (negated deferred post-filters) or AND-instead-of-OR (any deferred
|
||
// post-filter under an OR — they resolve to the full universe in Stage 2
|
||
// and are then applied conjunctively, so the OR would collapse to AND).
|
||
if let Some(ref filter_expr) = combined_filter {
|
||
user_filter::reject_negated_deferred_filters(filter_expr)?;
|
||
user_filter::reject_or_of_deferred_filters(filter_expr)?;
|
||
}
|
||
|
||
tracing::trace!(
|
||
profile = %query.profile.name,
|
||
limit = query.limit,
|
||
has_filter = combined_filter.is_some(),
|
||
has_diversity = query.diversity.is_some(),
|
||
"retrieve query start"
|
||
);
|
||
|
||
// Resolve the profile.
|
||
#[allow(clippy::option_if_let_else)]
|
||
let profile = match query.profile.version {
|
||
Some(v) => self
|
||
.profile_registry
|
||
.get_version(&query.profile.name, v)
|
||
.map_err(|_| QueryError::ProfileNotFound(query.profile.name.clone()))?,
|
||
None => self
|
||
.profile_registry
|
||
.get(&query.profile.name)
|
||
.map_err(|_| QueryError::ProfileNotFound(query.profile.name.clone()))?,
|
||
};
|
||
|
||
let mut warnings: Vec<String> = Vec::new();
|
||
|
||
// ── Stage 1: Candidate Generation ───────────────────────────────
|
||
// When for_creator is specified, restrict candidates to that creator's
|
||
// items using the CreatorItemsBitmap. This is O(k) where k = the
|
||
// creator's item count, avoiding the O(N) scan cap that would miss
|
||
// items at high IDs in a large catalog.
|
||
//
|
||
// The two `for_creator` arms below are kept SEPARATE on purpose. A single
|
||
// `if let creator_id && let creator_items` chain would make the whole
|
||
// condition false when `for_creator` is set but `creator_items` is absent,
|
||
// falling through to the full-universe scan and SILENTLY dropping the
|
||
// creator restriction (returning every creator's items). Instead, when the
|
||
// scope is requested but cannot be satisfied, we warn and return an EMPTY
|
||
// candidate set — the requested scope yields no items rather than the
|
||
// inverse (all items). Mirrors the warn-on-missing-index pattern in Stage 2.
|
||
let has_user_context = query.for_user.is_some();
|
||
let mut candidates = if let Some(creator_id) = query.for_creator {
|
||
if let Some(creator_items) = self.creator_items {
|
||
creator_items
|
||
.get(creator_id.as_u64())
|
||
.map_or_else(Vec::new, |bm| {
|
||
bm.iter()
|
||
.map(|id_u32| EntityId::new(u64::from(id_u32)))
|
||
.collect()
|
||
})
|
||
} else {
|
||
warnings.push(
|
||
"FOR CREATOR scope requested but the creator-items index is unavailable; \
|
||
returning no candidates (refusing to fall back to a full-universe scan \
|
||
that would ignore the creator restriction)"
|
||
.to_string(),
|
||
);
|
||
tracing::warn!(
|
||
creator_id = creator_id.as_u64(),
|
||
"for_creator scope unsatisfiable: creator_items index absent; returning empty"
|
||
);
|
||
Vec::new()
|
||
}
|
||
} else {
|
||
match &profile.candidate_strategy {
|
||
CandidateStrategy::Scan { .. } => {
|
||
candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
|
||
}
|
||
CandidateStrategy::SignalRanked { signal, .. } => {
|
||
// m12p2: candidates come from the cached per-signal-type top-K
|
||
// (O(K)). When the signal is absent from the schema, or no
|
||
// writes have landed yet, that yields nothing — fall back to a
|
||
// scan so the read still works (e.g. trending on a fresh corpus
|
||
// with no views yet) rather than returning an empty feed.
|
||
let ranked =
|
||
candidate_gen::signal_ranked_candidates(self.ledger, signal, query.limit);
|
||
if ranked.is_empty() {
|
||
warnings.push(format!(
|
||
"SignalRanked('{signal}') produced no candidates \
|
||
(signal absent or no writes yet); falling back to scan"
|
||
));
|
||
candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
|
||
} else {
|
||
ranked
|
||
}
|
||
}
|
||
CandidateStrategy::Ann {
|
||
limit: ann_limit, ..
|
||
} => {
|
||
// (`top_clusters` is consulted by the db layer when it resolves
|
||
// the fan-out set into `ann_query_vectors`; the executor just
|
||
// runs whatever vectors it was handed.)
|
||
// m12p2: ANN candidate generation in RETRIEVE — O(ef_search)
|
||
// nearest neighbours of the resolved query vector (the user's
|
||
// preference vector for `for_you`, the seed item's embedding
|
||
// for `similar_to`), over-fetched so Stage 2/2.5 + diversity
|
||
// have room. Degrades to a scan (with a caller- AND
|
||
// operator-visible note) when no registry/query vector is
|
||
// available — an anonymous read or a user with no preference
|
||
// vector yet — so the read keeps working instead of erroring.
|
||
// Over-fetch limit×10, floored so tiny feeds still seed a pool,
|
||
// and capped by the profile's declared `Ann.limit` so a huge
|
||
// page size can't drive an unbounded ANN beam.
|
||
let k = query
|
||
.limit
|
||
.saturating_mul(ANN_OVERFETCH)
|
||
.max(ANN_CANDIDATE_FLOOR)
|
||
.min((*ann_limit).max(ANN_CANDIDATE_FLOOR));
|
||
// Prefer the multi-vector fan-out set (warm `for_you`) when the
|
||
// db layer resolved one; fall back to the single query vector
|
||
// (`similar_to`, cold-start). Both are BORROWED — no per-request
|
||
// clone of the (up to 3 × dim) centroids on the serve path.
|
||
// `ann_candidates_multi` is byte-identical to the single search
|
||
// when the slice has ≤1 element, so a single-cluster user pays
|
||
// no fan-out overhead.
|
||
let single_fallback;
|
||
let fan_out: &[Vec<f32>] = match &self.ann_query_vectors {
|
||
Some(vs) if !vs.is_empty() => vs.as_slice(),
|
||
_ => match &self.ann_query_vector {
|
||
Some(v) => {
|
||
single_fallback = std::slice::from_ref(v);
|
||
single_fallback
|
||
}
|
||
None => &[],
|
||
},
|
||
};
|
||
if let Some(registry) = self.embedding_registry
|
||
&& !fan_out.is_empty()
|
||
{
|
||
let slot = self.item_embedding_slot.unwrap_or("content");
|
||
// ef_search=0 → the index default (ANN_DEFAULT_EF_SEARCH),
|
||
// matching the m12p2 single-vector path. TODO(tuning):
|
||
// thread the per-query ef_search the m12p3 knob resolves so
|
||
// the fan-out can trade beam width per cluster (the M-sweep
|
||
// open question). Pinned to the default for now.
|
||
let ann =
|
||
candidate_gen::ann_candidates_multi(registry, slot, fan_out, k, 0);
|
||
if ann.is_empty() {
|
||
warnings.push(
|
||
"ANN candidate generation returned no candidates; \
|
||
falling back to scan"
|
||
.to_string(),
|
||
);
|
||
tracing::warn!(
|
||
profile = %query.profile.name,
|
||
"ANN returned no candidates (empty/absent slot); scan fallback"
|
||
);
|
||
candidate_gen::scan_candidates(
|
||
self.universe,
|
||
query.limit,
|
||
has_user_context,
|
||
)
|
||
} else {
|
||
ann
|
||
}
|
||
} else {
|
||
warnings.push(
|
||
"ANN candidate strategy: no query vector resolvable \
|
||
(anonymous read or no preference vector yet); falling back to scan"
|
||
.to_string(),
|
||
);
|
||
tracing::debug!(
|
||
profile = %query.profile.name,
|
||
has_registry = self.embedding_registry.is_some(),
|
||
"ANN strategy without a query vector; scan fallback"
|
||
);
|
||
candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
|
||
}
|
||
}
|
||
CandidateStrategy::Relationship => {
|
||
// M3: source candidates from the user's followed creators.
|
||
// Uses the follows index in user_state + creator_items bitmap to build
|
||
// a candidate set from items produced by followed creators.
|
||
if let (Some(user_id), Some(user_state), Some(creator_items)) =
|
||
(query.for_user, self.user_state, self.creator_items)
|
||
{
|
||
let followed = user_state.followed_creators_vec(user_id);
|
||
if followed.is_empty() {
|
||
warnings.push(
|
||
"Relationship strategy: user follows no creators; falling back to scan"
|
||
.to_string(),
|
||
);
|
||
candidate_gen::scan_candidates(self.universe, query.limit, true)
|
||
} else {
|
||
let bitmap = creator_items.union_for(&followed);
|
||
if bitmap.is_empty() {
|
||
warnings.push(
|
||
"Relationship strategy: followed creators have no items; falling back to scan"
|
||
.to_string(),
|
||
);
|
||
candidate_gen::scan_candidates(self.universe, query.limit, true)
|
||
} else {
|
||
bitmap
|
||
.iter()
|
||
.map(|id_u32| EntityId::new(u64::from(id_u32)))
|
||
.collect()
|
||
}
|
||
}
|
||
} else {
|
||
warnings.push(
|
||
"Relationship strategy requires FOR USER clause; falling back to scan"
|
||
.to_string(),
|
||
);
|
||
candidate_gen::scan_candidates(self.universe, query.limit, false)
|
||
}
|
||
}
|
||
other => {
|
||
return Err(QueryError::UnsupportedStrategy(format!(
|
||
"{other:?} requires M4+ infrastructure"
|
||
)));
|
||
}
|
||
}
|
||
};
|
||
|
||
// Apply exclude list.
|
||
if !query.exclude.is_empty() {
|
||
let exclude_set: HashSet<u64> = query.exclude.iter().map(|id| id.as_u64()).collect();
|
||
candidates.retain(|id| !exclude_set.contains(&id.as_u64()));
|
||
}
|
||
|
||
// M7p2: ReducedCandidates — cap scan set to reduce Stage 3 scoring work.
|
||
//
|
||
// A blind `truncate(cap)` keeps the FIRST `cap` candidates in scan order.
|
||
// `scan_candidates` iterates the universe bitmap in ASCENDING entity-ID
|
||
// order, so a blind truncate keeps the LOWEST IDs. For a recency profile
|
||
// (`New`) the newest items are exactly the ones that should rank at the
|
||
// top, so a blind truncate drops the top-ranked items before scoring ever
|
||
// sees them and corrupts the ranking. We therefore truncate by the
|
||
// profile's primary ordering key. Other profiles score on signal state
|
||
// rather than scan position, so their truncation order is not load-bearing
|
||
// and a plain truncate is retained.
|
||
//
|
||
// The key for `New` is the `created_at` index, read newest-first. It used
|
||
// to be the entity ID, which was correct only while `Sort::New` itself
|
||
// ranked by descending ID; now that scoring reads the real `created_at`,
|
||
// an ID-keyed truncation would silently discard the genuinely newest items
|
||
// under load and leave the ranking wrong in a way visible only when
|
||
// degraded. With no `created_at` index wired there is no better key
|
||
// available at this point in the pipeline — metadata is not loaded until
|
||
// Stage 3 — so the ID heuristic remains the documented fallback.
|
||
if self.degradation_level.reduces_candidates() {
|
||
let cap = (query.limit * 4).max(100);
|
||
if candidates.len() > cap {
|
||
if matches!(profile.sort, Some(Sort::New)) {
|
||
self.truncate_to_newest(&mut candidates, cap);
|
||
}
|
||
candidates.truncate(cap);
|
||
}
|
||
}
|
||
|
||
stats.candidates_considered = candidates.len();
|
||
|
||
tracing::trace!(
|
||
candidates = candidates.len(),
|
||
"stage 1: candidates generated"
|
||
);
|
||
|
||
// ── Stage 2: Filter Evaluation ──────────────────────────────────
|
||
if let Some(ref filter_expr) = combined_filter {
|
||
// Resolve the universe up front, BEFORE building the evaluator. The
|
||
// `FilterEvaluator` resolves deferred variants
|
||
// (`MinSignal`/`MaxSignal`/`NearLocation`/`InCollection`/user-state)
|
||
// to `self.universe` and intersects every `And` child against it.
|
||
// That is only sound when the universe is a populated, trustworthy
|
||
// bitmap. When it is unavailable — not wired (`None`) or the read
|
||
// lock is poisoned — the empty fallback would make the intersection
|
||
// silently drop EVERY candidate (including ones that should pass)
|
||
// before the Stage 2.2/2.3/2.4 deferred post-filters ever run. The
|
||
// actual deferred filtering happens in those later stages, so the
|
||
// correct degraded behavior — mirroring the SEARCH pipeline's
|
||
// `apply_metadata_filter` (W28) — is to SKIP the index intersection
|
||
// entirely (retain all) and surface a warning, rather than collapse
|
||
// to an empty result set.
|
||
#[allow(clippy::option_if_let_else)]
|
||
let universe_guard = match self.universe {
|
||
Some(u) => {
|
||
if let Ok(guard) = u.read() {
|
||
Some(guard)
|
||
} else {
|
||
warnings.push(
|
||
"universe lock poisoned; metadata index pre-filter degraded — \
|
||
relying on post-filters"
|
||
.to_string(),
|
||
);
|
||
None
|
||
}
|
||
}
|
||
None => None,
|
||
};
|
||
|
||
if let Some(universe_ref) = universe_guard.as_deref() {
|
||
// Build a per-query FilterEvaluator from whatever indexes are
|
||
// available. When an index is `None` (no items have been written
|
||
// to it yet), the empty fallback causes filters on that field to
|
||
// match nothing -- which is the correct semantic: if no items
|
||
// have a "category" field indexed, then `CategoryEq("jazz")`
|
||
// should return zero results.
|
||
let empty_bitmap = BitmapIndex::new("_empty");
|
||
let empty_dur = RangeIndex::<u32>::new("_empty");
|
||
let empty_ts = RangeIndex::<u64>::new("_empty");
|
||
|
||
let cat = self.category_index.unwrap_or_else(|| {
|
||
tracing::debug!(
|
||
"category_index unavailable; CategoryEq/In filters will return no matches"
|
||
);
|
||
&empty_bitmap
|
||
});
|
||
let fmt = self.format_index.unwrap_or_else(|| {
|
||
tracing::debug!(
|
||
"format_index unavailable; FormatEq/In filters will return no matches"
|
||
);
|
||
&empty_bitmap
|
||
});
|
||
let cre = self.creator_index.unwrap_or(&empty_bitmap);
|
||
let tag = self.tag_index.unwrap_or(&empty_bitmap);
|
||
let dur = self.duration_index.unwrap_or(&empty_dur);
|
||
let ts = self.created_at_index.unwrap_or(&empty_ts);
|
||
|
||
let evaluator = FilterEvaluator::new(cat, fmt, cre, tag, dur, ts, universe_ref);
|
||
match evaluator.evaluate(filter_expr) {
|
||
FilterResult::Bitmap(bitmap) => {
|
||
// INCLUSION test against the match bitmap. The match bitmap
|
||
// holds only 32-bit ids, so a candidate id that overflows
|
||
// `u32` can never be a legitimate member and must be DROPPED
|
||
// — a raw `id as u32` truncation would alias `2^32 + k` onto
|
||
// the low slot `k`, which may be present in the bitmap, and
|
||
// wrongly RETAIN an item that does not satisfy the filter.
|
||
// `SignalRanked` candidates come straight from the ledger,
|
||
// which keys on full-u64 ids, so a `> u32::MAX` id is valid
|
||
// input that reaches here. Route through the checked
|
||
// `entity_as_u32` (drop-on-overflow), matching the
|
||
// InCollection/SocialGraph arms and the Stage 2.5 helper.
|
||
candidates.retain(|id| {
|
||
super::entity_as_u32(*id).is_some_and(|i| bitmap.contains(i))
|
||
});
|
||
}
|
||
FilterResult::Predicate(pred) => {
|
||
candidates.retain(|id| pred(id.as_u64()));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
tracing::trace!(candidates = candidates.len(), "stage 2: filter applied");
|
||
|
||
// ── 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 {
|
||
let ctx = post_filter::PostFilterCtx {
|
||
ledger: self.ledger,
|
||
collection_index: self.collection_index,
|
||
items_storage: self.items_storage,
|
||
degradation_level: self.degradation_level,
|
||
};
|
||
post_filter::apply_deferred_post_filters(filter_expr, &mut candidates, &ctx)?;
|
||
}
|
||
|
||
// ── Stage 2.5: User-Context Filtering (M3) ─────────────────────
|
||
// When FOR USER is specified, exclude seen/hidden items, items from
|
||
// blocked creators, and hard negatives, then apply user-state inclusion
|
||
// filters. The whole block is shared verbatim with the SEARCH pipeline
|
||
// via `super::apply_user_context_suppression` so the two surfaces cannot
|
||
// drift (a missing arm on one side silently leaks an item the other
|
||
// suppresses).
|
||
if let Some(user_id) = query.for_user {
|
||
super::apply_user_context_suppression(
|
||
&mut candidates,
|
||
user_id,
|
||
self.user_state,
|
||
self.creator_items,
|
||
self.hard_negatives,
|
||
combined_filter.as_ref(),
|
||
);
|
||
|
||
tracing::trace!(
|
||
candidates = candidates.len(),
|
||
user_id = user_id,
|
||
"stage 2.5: user-context filter applied"
|
||
);
|
||
}
|
||
|
||
stats.candidates_after_filter = candidates.len();
|
||
stats.filters_applied = query.filters.len();
|
||
|
||
// ── DateSaved validation (M6p3) ─────────────────────────────────
|
||
// DateSaved sort requires FOR USER context. Fail early with a clear
|
||
// error rather than silently returning bad ordering.
|
||
if matches!(profile.sort, Some(Sort::DateSaved)) && query.for_user.is_none() {
|
||
return Err(QueryError::InvalidFilter {
|
||
field: "sort".to_string(),
|
||
reason: "DateSaved sort requires FOR USER context".to_string(),
|
||
});
|
||
}
|
||
|
||
// ── Stage 3: Signal Scoring ─────────────────────────────────────
|
||
let scoring_start = Instant::now();
|
||
let now = Timestamp::now();
|
||
// `item_metadata` is the per-candidate metadata map Stage 3 pre-loaded
|
||
// once from `items_storage`. It is retained (not discarded) so Stage 4.5
|
||
// can resolve `creator_id` from it instead of re-reading storage per
|
||
// candidate inside `apply_notification_caps` (see the re-population below).
|
||
let (scored, item_metadata) = self.stage3_score(
|
||
&candidates,
|
||
profile,
|
||
query,
|
||
combined_filter.as_ref(),
|
||
now,
|
||
&mut stats,
|
||
&mut warnings,
|
||
)?;
|
||
let total_scored = scored.len();
|
||
stats.scoring_time_us = scoring_start.elapsed().as_micros() as u64;
|
||
tracing::trace!(scored = scored.len(), "stage 3: scored");
|
||
|
||
// ── Stage 4: Diversity Enforcement ──────────────────────────────
|
||
// Use explicit query diversity if set; otherwise fall back to the profile's
|
||
// DiversitySpec so that built-in profiles (e.g. notification: max_per_creator=1)
|
||
// are always enforced even when the caller does not set query.diversity.
|
||
let profile_diversity: Option<crate::ranking::diversity::DiversityConstraints> = {
|
||
let spec = &profile.diversity;
|
||
if spec.max_per_creator.is_some() || spec.format_mix_max_fraction.is_some() {
|
||
Some(crate::ranking::diversity::DiversityConstraints {
|
||
max_per_creator: spec.max_per_creator,
|
||
format_mix_max_fraction: spec.format_mix_max_fraction,
|
||
..crate::ranking::diversity::DiversityConstraints::default()
|
||
})
|
||
} else {
|
||
None
|
||
}
|
||
};
|
||
let effective_diversity = query.diversity.as_ref().or(profile_diversity.as_ref());
|
||
// M7p2: NoDiversity — skip diversity enforcement entirely under load.
|
||
let diversity_start = Instant::now();
|
||
let (final_candidates, constraints_satisfied) = if self.degradation_level.skips_diversity()
|
||
{
|
||
warnings.push("diversity enforcement skipped due to load degradation".to_string());
|
||
(scored, true)
|
||
} else if let Some(diversity) = effective_diversity {
|
||
// Pass the full scored set (not query.limit) so diversity constraints are
|
||
// applied before pagination slicing. This ensures the page boundary does
|
||
// not artificially inflate or reduce the diversity-enforced count.
|
||
let result = DiversitySelector::select(&scored, diversity, scored.len());
|
||
let satisfied = result.constraints_satisfied;
|
||
if !satisfied {
|
||
for v in &result.violations {
|
||
warnings.push(format!(
|
||
"diversity constraint '{}' relaxed: {}",
|
||
v.constraint, v.detail
|
||
));
|
||
}
|
||
}
|
||
(result.selected, satisfied)
|
||
} else {
|
||
(scored, true)
|
||
};
|
||
|
||
stats.diversity_time_us = diversity_start.elapsed().as_micros() as u64;
|
||
stats.candidates_after_diversity = final_candidates.len();
|
||
|
||
tracing::trace!(
|
||
final_count = final_candidates.len(),
|
||
"stage 4: diversity enforced"
|
||
);
|
||
|
||
// ── Stage 4.5: Notification Cap FILTER (M6p6) ───────────────────
|
||
// Trims the post-diversity set against the daily caps WITHOUT recording.
|
||
// Recording is deferred to Stage 5 so it counts only the delivered page,
|
||
// not items pagination drops (C-CRITICAL).
|
||
let final_candidates =
|
||
if let (Some(caps), Some(user_id)) = (query.notification_caps, query.for_user) {
|
||
let mut final_candidates = final_candidates;
|
||
// Backfill `creator_id` from the metadata Stage 3 already loaded so
|
||
// the per-creator cap check never re-reads storage per candidate.
|
||
// Stage 3 populates `creator_id` on the *scored* set, but only when
|
||
// the field was `None`; backfilling again here is cheap (a HashMap
|
||
// lookup) and closes the case where diversity selection or a later
|
||
// pass left a survivor with an unresolved creator. Without this, the
|
||
// `resolve_candidate_creator_id` fallback issues a storage `get()`
|
||
// for every still-unresolved candidate. Shares the
|
||
// one `backfill_creator_ids` helper with Stage 3 so the two passes
|
||
// cannot drift on the parse rule.
|
||
super::backfill_creator_ids(&mut final_candidates, &item_metadata);
|
||
self.apply_notification_caps(final_candidates, caps, user_id, &query.profile.name)
|
||
} else {
|
||
final_candidates
|
||
};
|
||
|
||
tracing::trace!(
|
||
final_count = final_candidates.len(),
|
||
"stage 4.5: notification cap filter applied (recording deferred to stage 5)"
|
||
);
|
||
|
||
// ── Stage 5: Result Assembly ────────────────────────────────────
|
||
// `Cursor::offset` is fallible: a keyset cursor used as an offset (or a
|
||
// 32-bit-overflowing value) is a loud `InvalidCursor` rather than a silent
|
||
// misread.
|
||
let offset = query.cursor.as_ref().map_or(Ok(0), Cursor::offset)?;
|
||
// Guard against cursor-offset overflow: an attacker-supplied cursor can
|
||
// carry an `offset` near `usize::MAX`, and `offset + limit` would wrap.
|
||
// `saturating_add` clamps to `usize::MAX`; the `.min(len)` then keeps the
|
||
// slice bounds valid and the empty-page branch below handles `offset >= len`.
|
||
let end = offset
|
||
.saturating_add(query.limit)
|
||
.min(final_candidates.len());
|
||
let page_slice = if offset < final_candidates.len() {
|
||
&final_candidates[offset..end]
|
||
} else {
|
||
&[]
|
||
};
|
||
|
||
// ── Notification delivery recording (post-pagination) ───────────────
|
||
// Record deliveries against ONLY the page actually returned to the caller,
|
||
// not the full post-diversity set. Stage 4.5 filtered without recording;
|
||
// this records the delivered page through the tracker's atomic
|
||
// check-and-record so the daily cap is consumed by exactly the
|
||
// notifications the caller receives — never items trimmed by pagination
|
||
// (C-CRITICAL: notification-cap recording must follow pagination). For
|
||
// non-notification profiles or no tracker this returns the page unchanged.
|
||
let recorded_page: Vec<crate::ranking::executor::ScoredCandidate> =
|
||
if let (Some(caps), Some(user_id)) = (query.notification_caps, query.for_user) {
|
||
self.record_notification_deliveries(page_slice, caps, user_id, &query.profile.name)
|
||
} else {
|
||
page_slice.to_vec()
|
||
};
|
||
let page: &[crate::ranking::executor::ScoredCandidate] = &recorded_page;
|
||
|
||
let items: Vec<RetrieveResult> = page
|
||
.iter()
|
||
.enumerate()
|
||
.map(|(i, c)| {
|
||
let signals: Vec<Signal> = c
|
||
.signal_snapshot
|
||
.iter()
|
||
.map(|(name, value)| Signal {
|
||
name: name.as_str().to_owned(),
|
||
value: *value,
|
||
source: "decay_score".to_string(),
|
||
})
|
||
.collect();
|
||
|
||
let reasons = crate::ranking::reason::select_top_reasons(c.reasons.clone());
|
||
|
||
RetrieveResult {
|
||
entity_id: c.entity_id,
|
||
score: c.score,
|
||
rank: offset + i + 1,
|
||
signals,
|
||
reasons,
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
// M2: offset-based pagination. The cursor encodes a list position rather
|
||
// than a (score, entity_id) keyset. This is stable only when the ranked
|
||
// list does not change between page requests (no concurrent signal writes
|
||
// during iteration). Replace with keyset pagination in M3.
|
||
let next_cursor = if end < final_candidates.len() {
|
||
Some(Cursor::from_offset(end))
|
||
} else {
|
||
None
|
||
};
|
||
|
||
tracing::trace!(
|
||
returned = items.len(),
|
||
has_next_cursor = next_cursor.is_some(),
|
||
"stage 5: results assembled"
|
||
);
|
||
|
||
stats.total_time_us = query_start.elapsed().as_micros() as u64;
|
||
|
||
Ok(Results {
|
||
items,
|
||
next_cursor,
|
||
total_candidates: total_scored,
|
||
constraints_satisfied,
|
||
warnings,
|
||
session_snapshot: self.session_snapshot.clone(),
|
||
degradation_level: self.degradation_level,
|
||
stats,
|
||
// Local queries carry no governance metadata; community-overlay
|
||
// queries (M9p2+) populate this.
|
||
policy_metadata: crate::query::retrieve::types::PolicyMetadata::default(),
|
||
})
|
||
}
|
||
}
|
||
|
||
// ── Tests ────────────────────────────────────────────────────────────────────
|
||
|
||
#[cfg(test)]
|
||
#[allow(clippy::unwrap_used)]
|
||
mod tests {
|
||
use std::time::Duration;
|
||
|
||
// `Retrieve`, `CandidateStrategy`, `EntityId`, and `Timestamp` are in scope
|
||
// via `use super::*`; `FilterExpr` is test-only here (the production pipeline
|
||
// body no longer names it directly), so it is imported explicitly to keep the
|
||
// module-level imports free of an otherwise-unused name.
|
||
use super::*;
|
||
use crate::{
|
||
ranking::{
|
||
profile::{DiversitySpec, RankingProfile},
|
||
registry::ProfileRegistry,
|
||
},
|
||
schema::{DecaySpec, EntityKind, SchemaBuilder, Window as RankWindow},
|
||
signals::{NoopWalWriter, SignalLedger},
|
||
storage::indexes::FilterExpr,
|
||
};
|
||
|
||
/// Schema with a single `view` signal so a `min_signal("view", …)` deferred
|
||
/// filter resolves and the `SignalRanked` strategy can generate candidates.
|
||
fn view_schema() -> crate::schema::Schema {
|
||
let mut builder = SchemaBuilder::new();
|
||
let _ = builder
|
||
.signal(
|
||
"view",
|
||
EntityKind::Item,
|
||
DecaySpec::Exponential {
|
||
half_life: Duration::from_secs(7 * 24 * 3600),
|
||
},
|
||
)
|
||
.windows(&[RankWindow::AllTime])
|
||
.add();
|
||
builder.build().unwrap()
|
||
}
|
||
|
||
/// Registry holding one `SignalRanked` profile. Unlike the built-in `Scan`
|
||
/// profiles, `SignalRanked` sources candidates from the ledger, NOT the
|
||
/// universe bitmap — exactly what the W28 "no universe wired" path needs so
|
||
/// candidate generation does not vanish before the deferred post-filter runs.
|
||
fn signal_ranked_registry() -> ProfileRegistry {
|
||
let mut reg = ProfileRegistry::new();
|
||
reg.register(RankingProfile {
|
||
name: "view_ranked".to_owned(),
|
||
version: 1,
|
||
candidate_strategy: CandidateStrategy::SignalRanked {
|
||
signal: "view".to_owned(),
|
||
window: RankWindow::AllTime,
|
||
},
|
||
boosts: vec![],
|
||
decay: None,
|
||
gates: vec![],
|
||
penalties: vec![],
|
||
excludes: vec![],
|
||
diversity: DiversitySpec::default(),
|
||
exploration: 0.0,
|
||
sort: None,
|
||
is_builtin: false,
|
||
})
|
||
.unwrap();
|
||
reg
|
||
}
|
||
|
||
/// W28 regression: a deferred `MinSignal` filter with NO universe wired must
|
||
/// still return candidates from `SignalRanked` retrieval and let the Stage
|
||
/// 2.3 post-filter trim them — NOT collapse to an empty set.
|
||
///
|
||
/// Before the fix, RETRIEVE Stage 2 fell back to an EMPTY `empty_universe`
|
||
/// when `self.universe` was `None`/poisoned, ran the `FilterEvaluator`
|
||
/// (which resolves `MinSignal` to that empty universe), and `candidates.retain`
|
||
/// dropped EVERY candidate before the post-filter could run. SEARCH already
|
||
/// skipped the index intersection in that case; this mirrors the same fix for
|
||
/// RETRIEVE.
|
||
#[test]
|
||
fn retrieve_min_signal_with_no_universe_still_postfilters() {
|
||
let schema = view_schema();
|
||
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
|
||
let profile_reg = signal_ranked_registry();
|
||
|
||
// Item 1 has 5 views (>= 3), item 2 has 1 view (< 3). Both have a `view`
|
||
// signal so both become SignalRanked candidates.
|
||
let now = Timestamp::now();
|
||
for _ in 0..5 {
|
||
ledger
|
||
.record_signal("view", EntityId::new(1), 1.0, now)
|
||
.unwrap();
|
||
}
|
||
ledger
|
||
.record_signal("view", EntityId::new(2), 1.0, now)
|
||
.unwrap();
|
||
|
||
// NOTE: `universe = None` — the exact condition that previously dropped
|
||
// every candidate in Stage 2.
|
||
let exec = RetrieveExecutor::new(
|
||
&ledger,
|
||
&profile_reg,
|
||
None, // category
|
||
None, // format
|
||
None, // creator
|
||
None, // tag
|
||
None, // duration
|
||
None, // created_at
|
||
None, // universe
|
||
);
|
||
let query = Retrieve::builder()
|
||
.profile("view_ranked")
|
||
.filter(FilterExpr::min_signal("view", 3.0))
|
||
.limit(10)
|
||
.build()
|
||
.unwrap();
|
||
|
||
let results = exec.execute(&query).unwrap();
|
||
let mut ids: Vec<u64> = results.items.iter().map(|r| r.entity_id.as_u64()).collect();
|
||
ids.sort_unstable();
|
||
assert_eq!(
|
||
ids,
|
||
vec![1],
|
||
"with no universe wired, min_signal(view,3) must keep item 1 and drop item 2 via the \
|
||
post-filter; before the fix the empty-universe intersection dropped BOTH"
|
||
);
|
||
}
|
||
|
||
/// Regression: the user-context suppression stage shared by RETRIEVE and
|
||
/// SEARCH (`super::apply_user_context_suppression`) must (a) exclude seen
|
||
/// items, (b) retain only saved items when a `Saved` filter is present, and
|
||
/// (c) treat a `> u32::MAX` id correctly under the u32 conversion — RETAIN it
|
||
/// against an exclusion bitmap (it can never be a member, and the old raw
|
||
/// `id as u32` truncation would have aliased it onto a low suppressed id and
|
||
/// dropped the wrong item).
|
||
#[test]
|
||
fn shared_suppression_excludes_seen_and_keeps_overflow_ids() {
|
||
use crate::entities::{HardNegIndex, UserStateIndex};
|
||
|
||
let user_state = UserStateIndex::new();
|
||
// User 42 has seen item 1.
|
||
user_state.mark_seen(42, 1);
|
||
let hard_neg = HardNegIndex::new();
|
||
|
||
// Candidate 1 is seen (must drop). Candidate 2 is unseen (must keep).
|
||
// Candidate `u64::from(u32::MAX) + 1` overflows u32: it can never be the
|
||
// seen bitmap's member, so it MUST be retained — a raw truncation would
|
||
// alias it onto slot 0 (== item 1, which IS seen) and wrongly drop it.
|
||
let overflow_id = u64::from(u32::MAX) + 1;
|
||
let mut candidates = vec![
|
||
EntityId::new(1),
|
||
EntityId::new(2),
|
||
EntityId::new(overflow_id),
|
||
];
|
||
|
||
crate::query::executor::apply_user_context_suppression(
|
||
&mut candidates,
|
||
42,
|
||
Some(&user_state),
|
||
None,
|
||
Some(&hard_neg),
|
||
None,
|
||
);
|
||
|
||
let mut ids: Vec<u64> = candidates.iter().map(|id| id.as_u64()).collect();
|
||
ids.sort_unstable();
|
||
assert_eq!(
|
||
ids,
|
||
vec![2, overflow_id],
|
||
"seen item 1 dropped; unseen item 2 and the overflow id (un-aliasable, so retained) kept"
|
||
);
|
||
}
|
||
|
||
/// Regression: a `Saved` inclusion filter routed through the shared
|
||
/// suppression helper keeps only saved items, and an overflow id is DROPPED
|
||
/// against the inclusion bitmap (it cannot satisfy membership).
|
||
#[test]
|
||
fn shared_suppression_saved_inclusion_drops_unsaved_and_overflow() {
|
||
use crate::entities::UserStateIndex;
|
||
|
||
let user_state = UserStateIndex::new();
|
||
user_state.add_save(42, 2); // only item 2 is saved
|
||
|
||
let overflow_id = u64::from(u32::MAX) + 1;
|
||
let mut candidates = vec![
|
||
EntityId::new(1),
|
||
EntityId::new(2),
|
||
EntityId::new(overflow_id),
|
||
];
|
||
|
||
crate::query::executor::apply_user_context_suppression(
|
||
&mut candidates,
|
||
42,
|
||
Some(&user_state),
|
||
None,
|
||
None,
|
||
Some(&FilterExpr::Saved(42)),
|
||
);
|
||
|
||
assert_eq!(
|
||
candidates,
|
||
vec![EntityId::new(2)],
|
||
"Saved(42) keeps only item 2; item 1 (unsaved) and the overflow id (cannot be an \
|
||
inclusion-bitmap member) are both dropped"
|
||
);
|
||
}
|
||
|
||
/// C-CRITICAL regression (Stage 2 inclusion intersection): a `SignalRanked`
|
||
/// candidate whose id is `(1<<32) + k` must be DROPPED by an index-backed
|
||
/// `CategoryEq` filter even when slot `k` IS present in the match bitmap. A raw
|
||
/// `id as u32` truncation would alias the overflow id onto slot `k` and wrongly
|
||
/// RETAIN it; the checked `entity_as_u32` (drop-on-overflow) prevents the leak.
|
||
#[test]
|
||
fn stage2_index_filter_drops_overflow_alias_candidate() {
|
||
use roaring::RoaringBitmap;
|
||
|
||
use crate::storage::indexes::bitmap::BitmapIndex;
|
||
|
||
let schema = view_schema();
|
||
let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
|
||
let profile_reg = signal_ranked_registry();
|
||
|
||
// Real low id `5` is in category "jazz"; the overflow id `(1<<32)+5`
|
||
// aliases onto slot 5 under truncation but must NOT be retained.
|
||
let k: u32 = 5;
|
||
let overflow_id = (1u64 << 32) + u64::from(k);
|
||
|
||
let now = Timestamp::now();
|
||
ledger
|
||
.record_signal("view", EntityId::new(u64::from(k)), 1.0, now)
|
||
.unwrap();
|
||
ledger
|
||
.record_signal("view", EntityId::new(overflow_id), 1.0, now)
|
||
.unwrap();
|
||
|
||
// Category index: slot 5 ∈ "jazz". The overflow item is NOT indexed under
|
||
// jazz; only its truncated alias (slot 5) would falsely match.
|
||
let category = BitmapIndex::new("category");
|
||
category.insert(k, "jazz");
|
||
|
||
// Universe must be Some so Stage 2 runs the index intersection. It only
|
||
// gates entry to the block; CategoryEq resolves to the category bitmap.
|
||
let mut universe_bm = RoaringBitmap::new();
|
||
universe_bm.insert(k);
|
||
let universe = std::sync::RwLock::new(universe_bm);
|
||
|
||
let exec = RetrieveExecutor::new(
|
||
&ledger,
|
||
&profile_reg,
|
||
Some(&category), // category
|
||
None, // format
|
||
None, // creator
|
||
None, // tag
|
||
None, // duration
|
||
None, // created_at
|
||
Some(&universe), // universe
|
||
);
|
||
let query = Retrieve::builder()
|
||
.profile("view_ranked")
|
||
.filter(FilterExpr::CategoryEq("jazz".into()))
|
||
.limit(10)
|
||
.build()
|
||
.unwrap();
|
||
|
||
let results = exec.execute(&query).unwrap();
|
||
let ids: Vec<u64> = results.items.iter().map(|r| r.entity_id.as_u64()).collect();
|
||
assert_eq!(
|
||
ids,
|
||
vec![u64::from(k)],
|
||
"only the genuine low id 5 may survive CategoryEq(jazz); the overflow id \
|
||
{overflow_id} must be dropped, not aliased onto slot {k} and retained"
|
||
);
|
||
}
|
||
}
|