fix(m12p2): invalidate SignalRanked top-K cache on CRDT reconciliation
apply_crdt_state force-sets accumulated scores outside apply_event_local, so it bypassed the single note_write() chokepoint that keeps the trending top-K cache fresh. A node going quiescent right after a partition heal kept serving the pre-reconciliation candidate set indefinitely. Invalidate the cache in apply_crdt_state too, with a regression test that reconciles a new high-score entity and requires it in the next candidate read. Also: extract the for_you/related ANN candidate cap to a named constant (ANN_PROFILE_CANDIDATE_LIMIT), keep the recall test's ef_search in lockstep with its construction default (honored since m12p3), and clarify the vector_search region doc (always null until m12p4 cross-shard reads).
This commit is contained in:
parent
bb21e69ae6
commit
da5d2d4d53
@ -292,7 +292,10 @@ pub struct SearchItem {
|
|||||||
pub struct VectorSearchResponse {
|
pub struct VectorSearchResponse {
|
||||||
/// Nearest items, ordered by ascending distance (closest first).
|
/// Nearest items, ordered by ascending distance (closest first).
|
||||||
pub items: Vec<VectorMatch>,
|
pub items: Vec<VectorMatch>,
|
||||||
/// Region the search was served from (cluster mode); `null` standalone.
|
/// Always `null`: the recall probe serves locally (standalone, or — in
|
||||||
|
/// cluster mode — merged across the node's hosted shard groups), so there is
|
||||||
|
/// no single serving region to report. Cross-region routing for this probe is
|
||||||
|
/// the m12p4 cross-shard read follow-up.
|
||||||
pub region: Option<String>,
|
pub region: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -277,6 +277,13 @@ const FOR_YOU_EXPLORATION: f64 = 0.1;
|
|||||||
/// Maximum items per creator in following results.
|
/// Maximum items per creator in following results.
|
||||||
const FOLLOWING_MAX_PER_CREATOR: usize = 3;
|
const FOLLOWING_MAX_PER_CREATOR: usize = 3;
|
||||||
|
|
||||||
|
/// Upper bound on the ANN candidate pool for the `for_you` and `related` profiles
|
||||||
|
/// (m12p2). Caps how many nearest neighbours the executor may draw from a single
|
||||||
|
/// query vector; the executor over-fetches `query.limit × ANN_OVERFETCH` within
|
||||||
|
/// this cap (see `query::executor::pipeline`). Generous enough that Stage 2/2.5
|
||||||
|
/// filtering and Stage 4 diversity have room without an unbounded beam.
|
||||||
|
const ANN_PROFILE_CANDIDATE_LIMIT: usize = 1000;
|
||||||
|
|
||||||
/// `for_you`: personalized home feed ranking.
|
/// `for_you`: personalized home feed ranking.
|
||||||
///
|
///
|
||||||
/// Combines interaction-weighted decay scores with exploration injection.
|
/// Combines interaction-weighted decay scores with exploration injection.
|
||||||
@ -292,7 +299,7 @@ fn for_you() -> RankingProfile {
|
|||||||
// candidate pool; the executor over-fetches `query.limit × 10` within it.
|
// candidate pool; the executor over-fetches `query.limit × 10` within it.
|
||||||
p.candidate_strategy = CandidateStrategy::Ann {
|
p.candidate_strategy = CandidateStrategy::Ann {
|
||||||
slot: "content".into(),
|
slot: "content".into(),
|
||||||
limit: 1000,
|
limit: ANN_PROFILE_CANDIDATE_LIMIT,
|
||||||
};
|
};
|
||||||
p.sort = Some(Sort::Hot { gravity: 1.5 });
|
p.sort = Some(Sort::Hot { gravity: 1.5 });
|
||||||
p.boosts = vec![
|
p.boosts = vec![
|
||||||
@ -358,7 +365,7 @@ fn related() -> RankingProfile {
|
|||||||
// Degrades to a scan when no `similar_to` is supplied (executor fallback).
|
// Degrades to a scan when no `similar_to` is supplied (executor fallback).
|
||||||
p.candidate_strategy = CandidateStrategy::Ann {
|
p.candidate_strategy = CandidateStrategy::Ann {
|
||||||
slot: "content".into(),
|
slot: "content".into(),
|
||||||
limit: 1000,
|
limit: ANN_PROFILE_CANDIDATE_LIMIT,
|
||||||
};
|
};
|
||||||
p.sort = Some(Sort::Hot { gravity: 1.2 });
|
p.sort = Some(Sort::Hot { gravity: 1.2 });
|
||||||
p.boosts = vec![
|
p.boosts = vec![
|
||||||
|
|||||||
@ -564,8 +564,10 @@ impl SignalLedger {
|
|||||||
// returning (satisfies clippy::significant_drop_tightening).
|
// returning (satisfies clippy::significant_drop_tightening).
|
||||||
drop(entry);
|
drop(entry);
|
||||||
// m12p2: invalidate the SignalRanked top-K cache (one relaxed atomic add).
|
// m12p2: invalidate the SignalRanked top-K cache (one relaxed atomic add).
|
||||||
// This is the single in-memory mutation chokepoint for every signal apply
|
// This is the single in-memory mutation chokepoint for the incremental
|
||||||
// path (record/replay/replication), so the cache can never miss a write.
|
// signal apply paths (record/replay/replication). The CRDT reconciliation
|
||||||
|
// path force-sets scores without funneling through here, so it invalidates
|
||||||
|
// separately — see `apply_crdt_state`.
|
||||||
self.hot_top_k.note_write();
|
self.hot_top_k.note_write();
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -740,6 +742,13 @@ impl SignalLedger {
|
|||||||
entry.warm.reconcile_to_count(state.total_count(), now_ns);
|
entry.warm.reconcile_to_count(state.total_count(), now_ns);
|
||||||
|
|
||||||
drop(entry);
|
drop(entry);
|
||||||
|
// m12p2: reconciliation force-set the accumulated score above, which can
|
||||||
|
// change SignalRanked top-K membership without going through
|
||||||
|
// `apply_event_local`. Invalidate the cache here too, or a node that goes
|
||||||
|
// quiescent right after reconciling a partition heal would keep serving a
|
||||||
|
// pre-reconciliation trending candidate set indefinitely (the small-ledger
|
||||||
|
// path treats an unchanged `dirty` counter as "exact regardless of clock").
|
||||||
|
self.hot_top_k.note_write();
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -563,6 +563,46 @@ mod reconcile_tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Regression (m12p2 W1): the `SignalRanked` top-K cache must reflect a
|
||||||
|
/// reordering caused by CRDT reconciliation. `apply_crdt_state` force-sets the
|
||||||
|
/// accumulated score outside `apply_event_local`, so before the fix it never
|
||||||
|
/// bumped the cache's `dirty` counter — a node quiescent after a partition
|
||||||
|
/// heal kept serving the pre-reconciliation candidate set. We populate the
|
||||||
|
/// cache (entity A only), reconcile a new high-score entity B, and require B
|
||||||
|
/// to appear in the very next candidate read.
|
||||||
|
#[test]
|
||||||
|
fn reconcile_invalidates_signal_ranked_cache() {
|
||||||
|
let ledger = SignalLedger::new(schema(), Box::new(NoopWalWriter));
|
||||||
|
let a = EntityId::new(1);
|
||||||
|
let b = EntityId::new(2);
|
||||||
|
let type_id = ledger.resolve_signal_type("view").unwrap();
|
||||||
|
|
||||||
|
// Local node has only A; populate (and cache) the SignalRanked top-K.
|
||||||
|
ledger
|
||||||
|
.record_signal("view", a, 1.0, Timestamp::now())
|
||||||
|
.unwrap();
|
||||||
|
let before = ledger.hot_top_k_candidates("view", 50);
|
||||||
|
assert!(before.contains(&a), "A must be a candidate after its write");
|
||||||
|
assert!(
|
||||||
|
!before.contains(&b),
|
||||||
|
"B has no signal yet, must be absent before reconciliation"
|
||||||
|
);
|
||||||
|
|
||||||
|
// A CRDT merge gives B a large undecayed score (lambda=0 ⇒ score == count).
|
||||||
|
let mut state = CrdtSignalState::new(0.0);
|
||||||
|
state.increment_bucket(ShardId(3), 10);
|
||||||
|
ledger.apply_crdt_state(b, type_id, &state).unwrap();
|
||||||
|
|
||||||
|
// The cache must have been invalidated by the reconciliation: B is now a
|
||||||
|
// candidate. Without the `note_write()` in `apply_crdt_state`, the stale
|
||||||
|
// cache would still return the A-only set and this would fail.
|
||||||
|
let after = ledger.hot_top_k_candidates("view", 50);
|
||||||
|
assert!(
|
||||||
|
after.contains(&b),
|
||||||
|
"B must surface as a candidate immediately after reconciliation"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/// Re-applying the same merged state is idempotent for the warm tier:
|
/// Re-applying the same merged state is idempotent for the warm tier:
|
||||||
/// the count does not grow on the second apply.
|
/// the count does not grow on the second apply.
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@ -362,13 +362,19 @@ fn usearch_recall_at_10k() {
|
|||||||
let k = 100;
|
let k = 100;
|
||||||
let num_queries: i32 = 10;
|
let num_queries: i32 = 10;
|
||||||
|
|
||||||
|
// Beam width = 4× k, the value the doc comment above calls out for recall@100
|
||||||
|
// > 0.95. Single source of truth for BOTH the construction default and the
|
||||||
|
// per-query override below: since m12p3 the `ef_search` arg to `search` is
|
||||||
|
// honored, so a smaller per-query value would genuinely lower recall — keep
|
||||||
|
// the two in lockstep here rather than re-introducing the old divergence.
|
||||||
|
let search_ef = 400;
|
||||||
let config = VectorIndexConfig {
|
let config = VectorIndexConfig {
|
||||||
dimensions: dim,
|
dimensions: dim,
|
||||||
metric: DistanceMetric::L2,
|
metric: DistanceMetric::L2,
|
||||||
quantization: QuantizationLevel::F32,
|
quantization: QuantizationLevel::F32,
|
||||||
connectivity: 16,
|
connectivity: 16,
|
||||||
ef_construction: 200,
|
ef_construction: 200,
|
||||||
ef_search: 400,
|
ef_search: search_ef,
|
||||||
};
|
};
|
||||||
let brute_config = f32_config(dim);
|
let brute_config = f32_config(dim);
|
||||||
let usearch = UsearchIndex::new(config).unwrap();
|
let usearch = UsearchIndex::new(config).unwrap();
|
||||||
@ -391,7 +397,7 @@ fn usearch_recall_at_10k() {
|
|||||||
for _ in 0..num_queries {
|
for _ in 0..num_queries {
|
||||||
let query = random_unit_vector(dim, &mut rng);
|
let query = random_unit_vector(dim, &mut rng);
|
||||||
let brute_results = brute.search(&query, k, 0).unwrap();
|
let brute_results = brute.search(&query, k, 0).unwrap();
|
||||||
let usearch_results = usearch.search(&query, k, 200).unwrap();
|
let usearch_results = usearch.search(&query, k, search_ef).unwrap();
|
||||||
|
|
||||||
let brute_ids: Vec<u64> = brute_results.iter().map(|r| r.id).collect();
|
let brute_ids: Vec<u64> = brute_results.iter().map(|r| r.id).collect();
|
||||||
let usearch_ids: Vec<u64> = usearch_results.iter().map(|r| r.id).collect();
|
let usearch_ids: Vec<u64> = usearch_results.iter().map(|r| r.id).collect();
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user