diff --git a/Cargo.lock b/Cargo.lock index 2b9ab95..9b380c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3659,6 +3659,7 @@ dependencies = [ "rustix 1.1.3", "serde", "serde_json", + "smallvec", "tantivy", "tempfile", "thiserror 2.0.18", diff --git a/docs/reviews/perf-sweep-2026-06-13.md b/docs/reviews/perf-sweep-2026-06-13.md index ab1a749..3ef7618 100644 --- a/docs/reviews/perf-sweep-2026-06-13.md +++ b/docs/reviews/perf-sweep-2026-06-13.md @@ -680,3 +680,31 @@ fallacy the relabel removes; the open-loop tidal-stress run is the cited tail au **Remaining:** Waves 2–7 (allocation kill → de-clone → storage/transport → threading → WAL/ship CPU → enforcement). Each lands against the benches above as the before/after gate. + +## Wave 2 — source-level allocation kill (IN PROGRESS 2026-06-13) + +### T1 — signal_snapshot carrier (DONE, verified) +Replaced `ScoredCandidate.signal_snapshot: Vec<(String,f64)>` with +`SmallVec<[(SignalKey,f64); 4]>` where `SignalKey = Static(&'static str) | Owned(Arc)`: +- All compile-time-constant labels (sort bases, `relevance`, `co_engagement`, + `preference_affinity`) → `Static` — **zero allocation**, pointer-copy clone. +- Dynamic `{signal}_boost`/`_penalty`/`_decay` labels → built **once per query** + in a `RuleLabels` hoist (was `format!` per-candidate-per-rule) and shared by + `Arc` — refcount-bump clone, not a string copy. Cohort rescore hoisted the same way. +- `scored` accumulator pre-sized to `candidates.len()`. +- Owned `String`s rebuilt only at the two response-assembly sites (`≤ limit` items). +- Files: `ranking/executor/{context,mod,scoring,helpers}.rs`, `query/executor/{helpers,pipeline,candidate_gen}.rs`, `query/search/executor/pipeline.rs`, `tidal/Cargo.toml` (smallvec promoted from lock). + +**Correctness:** byte-identical — full 1896-test lib suite green (existing tests assert snapshot keys AND scores). +**Measured win** (real before/after, `cargo bench --bench ranking`, committed-base vs working-tree): +`score_200_hot` 23.89→21.04 µs (**−11.9%**), `score_200_trending` 27.66→25.85 µs (**−6.5%**), +`score_200_full_pipeline` 27.88→26.97 µs (**−3.3%**). Win scales with per-candidate snapshot +width; the personalized `for_you` path (boosts+penalties+decay) benefits most and is not yet benched. + +### T2 — per-term DashMap lookup collapse (NEXT, not started) +Group each candidate's exclude/gate/boost/penalty/decay terms by `SignalTypeId`, take ONE +`entries.get()` per distinct type (held `Ref` serves all that type's terms), collapsing +(E+G+B+P+sort+decay) shard-lock+hash+lookups per candidate to T distinct types. HIGH correctness +risk (short-circuit order, per-term degradation-window substitution, the gates-propagate / +boosts-swallow `UnknownSignalType` split, None→default mapping) — gated by an A/B property test +asserting identical ScoredCandidate (score + snapshot) across random profiles before the loop is switched. diff --git a/tidal/Cargo.toml b/tidal/Cargo.toml index cbbdd15..bcc5f03 100644 --- a/tidal/Cargo.toml +++ b/tidal/Cargo.toml @@ -21,6 +21,12 @@ lru = "0.12" fs4 = "0.8" rand = "0.9" roaring = "0.10" +# Inline storage for the per-candidate ranking signal-snapshot (ranking/executor): +# the common formula-sort breadth fits inline, so the hot scoring loop builds the +# explain-ability breakdown with zero heap allocation. Already in the lock +# transitively (tantivy/fjall), so no new dependency surface — promoted to a +# direct dep here, matching the arc-swap/blake3/base64 pattern. +smallvec = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" diff --git a/tidal/benches/diversity.rs b/tidal/benches/diversity.rs index 4d47362..93c261e 100644 --- a/tidal/benches/diversity.rs +++ b/tidal/benches/diversity.rs @@ -21,7 +21,7 @@ fn make_200_candidates(n_creators: usize) -> Vec { entity_id: EntityId::new(i as u64 + 1), #[allow(clippy::cast_precision_loss)] score: (200 - i) as f64 / 200.0, - signal_snapshot: vec![], + signal_snapshot: smallvec::SmallVec::new(), creator_id: Some(EntityId::new((i % n_creators) as u64 + 1)), format: Some(if i % 3 == 0 { "video".into() diff --git a/tidal/src/query/executor/candidate_gen.rs b/tidal/src/query/executor/candidate_gen.rs index 30dc7cb..1665eec 100644 --- a/tidal/src/query/executor/candidate_gen.rs +++ b/tidal/src/query/executor/candidate_gen.rs @@ -8,7 +8,7 @@ use std::{collections::HashSet, sync::RwLock}; use roaring::RoaringBitmap; use crate::{ - ranking::executor::ScoredCandidate, + ranking::executor::{ScoredCandidate, SignalSnapshot}, schema::{EntityId, Timestamp}, signals::SignalLedger, }; @@ -176,7 +176,7 @@ pub(crate) fn inject_exploration( scored.push(ScoredCandidate { entity_id, score: 0.0, - signal_snapshot: vec![], + signal_snapshot: SignalSnapshot::new(), creator_id: None, format: None, }); @@ -189,7 +189,10 @@ pub(crate) fn inject_exploration( #[allow(clippy::unwrap_used)] mod tests { use super::*; - use crate::{ranking::executor::ScoredCandidate, schema::EntityId}; + use crate::{ + ranking::executor::{ScoredCandidate, SignalSnapshot}, + schema::EntityId, + }; #[test] fn exploration_injects_random_candidates() { @@ -200,7 +203,7 @@ mod tests { // i ∈ 1..=5 — exactly representable in f64; fixture score only. #[allow(clippy::cast_precision_loss)] score: (i as f64).mul_add(-0.1, 1.0), - signal_snapshot: vec![], + signal_snapshot: SignalSnapshot::new(), creator_id: None, format: None, }) @@ -244,7 +247,7 @@ mod tests { .map(|i| ScoredCandidate { entity_id: EntityId::new(i), score: 1.0, - signal_snapshot: vec![], + signal_snapshot: SignalSnapshot::new(), creator_id: None, format: None, }) @@ -265,7 +268,7 @@ mod tests { .map(|i| ScoredCandidate { entity_id: EntityId::new(i), score: 1.0, - signal_snapshot: vec![], + signal_snapshot: SignalSnapshot::new(), creator_id: None, format: None, }) diff --git a/tidal/src/query/executor/helpers.rs b/tidal/src/query/executor/helpers.rs index 273a089..a4aa267 100644 --- a/tidal/src/query/executor/helpers.rs +++ b/tidal/src/query/executor/helpers.rs @@ -2,11 +2,15 @@ //! //! Contains cohort rescoring, notification cap enforcement, and related helpers. +use std::sync::Arc; + +use smallvec::SmallVec; + use super::RetrieveExecutor; use crate::{ db::deserialize_metadata as deserialize_item_metadata, query::retrieve::QueryError, - ranking::executor::ScoredCandidate, + ranking::executor::{ScoredCandidate, SignalKey, SignalSnapshot}, schema::EntityId, storage::{Tag, encode_key}, }; @@ -242,14 +246,22 @@ impl RetrieveExecutor<'_> { return Ok(()); } + // Cohort snapshot keys are the cohort boost signal names — invariant + // across candidates, so intern them once into shared `Arc`s here rather + // than re-cloning each name per candidate (mirrors the global path's + // `RuleLabels` hoist). + let cohort_labels: Vec> = boosts + .iter() + .map(|b| Arc::from(b.signal.as_str())) + .collect(); for candidate in scored.iter_mut() { let mut cohort_score = 0.0; // Rebuild the signal snapshot from the cohort-scoped values so the // reported `RetrieveResult.signals` reflect the cohort scope the score // was derived from, not the stale global-ledger snapshot the main path // left on the candidate (C-CRITICAL: snapshot consistency). - let mut cohort_snapshot: Vec<(String, f64)> = Vec::with_capacity(boosts.len()); - for boost in boosts { + let mut cohort_snapshot: SignalSnapshot = SmallVec::with_capacity(boosts.len()); + for (i, boost) in boosts.iter().enumerate() { let value = match &boost.agg { SignalAgg::Value => { // Degrade an unknown signal to 0 (the ledger has no @@ -304,7 +316,7 @@ impl RetrieveExecutor<'_> { } }; cohort_score += value * boost.weight; - cohort_snapshot.push((boost.signal.clone(), value)); + cohort_snapshot.push((SignalKey::Owned(Arc::clone(&cohort_labels[i])), value)); } candidate.score = cohort_score; candidate.signal_snapshot = cohort_snapshot; @@ -543,7 +555,7 @@ mod tests { ScoredCandidate { entity_id: EntityId::new(id), score, - signal_snapshot: Vec::new(), + signal_snapshot: SignalSnapshot::new(), creator_id: None, format: None, } @@ -779,7 +791,7 @@ mod tests { let mk = |id: u64, creator: u64| ScoredCandidate { entity_id: EntityId::new(id), score: 1.0, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: Some(EntityId::new(creator)), format: None, }; diff --git a/tidal/src/query/executor/pipeline.rs b/tidal/src/query/executor/pipeline.rs index abb49da..51f2c25 100644 --- a/tidal/src/query/executor/pipeline.rs +++ b/tidal/src/query/executor/pipeline.rs @@ -508,7 +508,7 @@ impl RetrieveExecutor<'_> { .signal_snapshot .iter() .map(|(name, value)| Signal { - name: name.clone(), + name: name.as_str().to_owned(), value: *value, source: "decay_score".to_string(), }) diff --git a/tidal/src/query/executor/tests_part2.rs b/tidal/src/query/executor/tests_part2.rs index 733a808..f68d96d 100644 --- a/tidal/src/query/executor/tests_part2.rs +++ b/tidal/src/query/executor/tests_part2.rs @@ -352,14 +352,14 @@ fn cohort_rescore_normalizes_score_to_unit_range_and_reorders() { ScoredCandidate { entity_id: EntityId::new(1), score: 0.9, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: None, format: None, }, ScoredCandidate { entity_id: EntityId::new(2), score: 0.1, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: None, format: None, }, @@ -591,7 +591,7 @@ fn notification_caps_limit_total_and_per_creator() { let mk = |id: u64, creator: u64| ScoredCandidate { entity_id: EntityId::new(id), score: 1.0, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: Some(EntityId::new(creator)), format: None, }; diff --git a/tidal/src/query/search/executor/pipeline.rs b/tidal/src/query/search/executor/pipeline.rs index 2a46875..65594c0 100644 --- a/tidal/src/query/search/executor/pipeline.rs +++ b/tidal/src/query/search/executor/pipeline.rs @@ -773,7 +773,7 @@ impl SearchExecutor<'_> { .signal_snapshot .iter() .map(|(name, value)| Signal { - name: name.clone(), + name: name.as_str().to_owned(), value: *value, source: "decay_score".to_string(), }) diff --git a/tidal/src/ranking/diversity/selector.rs b/tidal/src/ranking/diversity/selector.rs index 4621042..aaeffff 100644 --- a/tidal/src/ranking/diversity/selector.rs +++ b/tidal/src/ranking/diversity/selector.rs @@ -324,7 +324,7 @@ mod tests { ScoredCandidate { entity_id: EntityId::new(id), score, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: creator.map(EntityId::new), format: format.map(String::from), } @@ -506,7 +506,7 @@ mod tests { candidates.push(ScoredCandidate { entity_id, score, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: Some(EntityId::new(creator as u64 + 1)), format: Some("video".into()), }); @@ -521,7 +521,7 @@ mod tests { .map(|i| ScoredCandidate { entity_id: EntityId::new(i as u64 + 1), score: (n - i) as f64, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: Some(EntityId::new(i as u64 + 1)), format: Some(if i % 2 == 0 { "video".into() @@ -537,7 +537,7 @@ mod tests { .map(|i| ScoredCandidate { entity_id: EntityId::new(i as u64 + 1), score: (n - i) as f64, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: Some(EntityId::new(1)), format: Some("video".into()), }) diff --git a/tidal/src/ranking/executor/context.rs b/tidal/src/ranking/executor/context.rs index d94507a..d30bb37 100644 --- a/tidal/src/ranking/executor/context.rs +++ b/tidal/src/ranking/executor/context.rs @@ -5,6 +5,9 @@ //! single query, and `ScoredCandidate` is the output of the scoring pipeline. use std::collections::HashMap; +use std::sync::Arc; + +use smallvec::SmallVec; use crate::schema::EntityId; @@ -35,6 +38,82 @@ pub struct UserContext { pub preference_boosts: HashMap, } +// -- Signal-snapshot key ------------------------------------------------------ + +/// A key in a candidate's [`ScoredCandidate::signal_snapshot`]. +/// +/// The snapshot exists only for response-time explain-ability, yet historically +/// every contributing signal allocated an owned `String` **per candidate** — +/// `format!("{}_boost", signal)` / `"relevance".to_string()` ran inside the +/// scoring hot loop, and ~80–90 % of those strings died unread because diversity +/// and pagination clone the whole `ScoredCandidate` and only the returned page is +/// ever serialized. `SignalKey` removes that cascade: +/// +/// - [`SignalKey::Static`] holds a compile-time-constant label (`"view"`, +/// `"relevance"`, …) — zero allocation, a pointer copy on clone. +/// - [`SignalKey::Owned`] holds a runtime label (`"{signal}_boost"`, a cohort +/// signal name) built **once per rule per query** and shared across every +/// candidate by `Arc` — a refcount bump on clone, never a string copy. +/// +/// Owned `String`s are rebuilt only at the two response-assembly sites, for the +/// `≤ limit` items that actually cross the wire, so the JSON `signals[].name` +/// stays byte-for-byte identical. +#[derive(Debug, Clone)] +pub enum SignalKey { + /// A compile-time-constant label. Zero-allocation; clone is a pointer copy. + Static(&'static str), + /// A runtime label built once per rule and shared by `Arc`; clone is a + /// refcount bump. + Owned(Arc), +} + +impl SignalKey { + /// Borrow the label as a `&str` (the form the response assembly materializes + /// into an owned `String`). + #[inline] + #[must_use] + pub fn as_str(&self) -> &str { + match self { + Self::Static(s) => s, + Self::Owned(s) => s, + } + } +} + +impl std::fmt::Display for SignalKey { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +impl From<&'static str> for SignalKey { + fn from(s: &'static str) -> Self { + Self::Static(s) + } +} + +impl PartialEq for SignalKey { + fn eq(&self, other: &str) -> bool { + self.as_str() == other + } +} + +impl PartialEq<&str> for SignalKey { + fn eq(&self, other: &&str) -> bool { + self.as_str() == *other + } +} + +/// A candidate's per-signal score breakdown for response explain-ability. +/// +/// Inline capacity 4 covers the common formula-sort snapshot width (the base +/// sort labels — `TopWindow` emits 4) so `trending` / `hot` / `top_window` +/// queries never touch the heap; richer personalized profiles spill to a single +/// heap allocation, still with no per-key string allocation. The element order +/// is the contribution order (sort base → relevance → boosts → co-engagement → +/// penalties → decay) and is preserved verbatim into the response `signals` array. +pub type SignalSnapshot = SmallVec<[(SignalKey, f64); 4]>; + // -- Scored candidate --------------------------------------------------------- /// A candidate entity with its computed score and signal snapshot. @@ -46,7 +125,7 @@ pub struct UserContext { pub struct ScoredCandidate { pub entity_id: EntityId, pub score: f64, - pub signal_snapshot: Vec<(String, f64)>, + pub signal_snapshot: SignalSnapshot, /// Creator ID for diversity enforcement (populated in m2p4). pub creator_id: Option, /// Content format for diversity enforcement (populated in m2p4). diff --git a/tidal/src/ranking/executor/helpers.rs b/tidal/src/ranking/executor/helpers.rs index 529bde9..066ff7e 100644 --- a/tidal/src/ranking/executor/helpers.rs +++ b/tidal/src/ranking/executor/helpers.rs @@ -397,7 +397,7 @@ mod tests { let mut candidates = vec![ScoredCandidate { entity_id: EntityId::new(1), score: 42.0, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: None, format: None, }]; @@ -411,7 +411,7 @@ mod tests { ScoredCandidate { entity_id: EntityId::new(id), score, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: None, format: None, } @@ -477,7 +477,7 @@ mod tests { .map(|i| ScoredCandidate { entity_id: EntityId::new(i), score: f64::NAN, - signal_snapshot: vec![], + signal_snapshot: crate::ranking::executor::SignalSnapshot::new(), creator_id: None, format: None, }) diff --git a/tidal/src/ranking/executor/mod.rs b/tidal/src/ranking/executor/mod.rs index 1683e8e..2215a9a 100644 --- a/tidal/src/ranking/executor/mod.rs +++ b/tidal/src/ranking/executor/mod.rs @@ -20,8 +20,9 @@ pub mod helpers; // Re-export all public items so that `crate::ranking::executor::Foo` paths continue to work. use std::collections::HashMap; +use std::sync::Arc; -pub use context::{ScoredCandidate, UserContext}; +pub use context::{ScoredCandidate, SignalKey, SignalSnapshot, UserContext}; use formulas::{INTERACTION_BOOST_WEIGHT, PREFERENCE_BOOST_WEIGHT, RELEVANCE_BASE_WEIGHT}; use helpers::{normalize, passes_excludes, passes_gates, read_agg_for_sort}; @@ -34,6 +35,41 @@ use crate::{ signals::SignalLedger, }; +/// Per-query snapshot labels for the dynamic boost / penalty / decay rules. +/// +/// The `{signal}_boost` / `{signal}_penalty` / `{signal}_decay` labels depend +/// only on the profile, so they are invariant across candidates and built **once +/// per query** here, then shared onto every candidate's snapshot by `Arc` — a +/// refcount bump in place of the old per-`(candidate, rule)` `format!` heap +/// allocation. `boosts` / `penalties` are parallel by index to `profile.boosts` +/// / `profile.penalties`; `decay` is `Some` exactly when `profile.decay` is. +struct RuleLabels { + boosts: Vec>, + penalties: Vec>, + decay: Option>, +} + +impl RuleLabels { + fn build(profile: &RankingProfile) -> Self { + Self { + boosts: profile + .boosts + .iter() + .map(|b| Arc::from(format!("{}_boost", b.signal))) + .collect(), + penalties: profile + .penalties + .iter() + .map(|p| Arc::from(format!("{}_penalty", p.signal))) + .collect(), + decay: profile + .decay + .as_ref() + .map(|d| Arc::from(format!("{}_decay", d.signal))), + } + } +} + /// Neutralize a `NaN` score at *construction*, before it can reach the sort. /// /// Boost arithmetic (additive session / interaction / co-engagement terms over @@ -322,7 +358,7 @@ impl<'a> ProfileExecutor<'a> { if let Some(&cosine) = user_ctx.preference_boosts.get(&entity_id.as_u64()) { preference_boost = cosine * PREFERENCE_BOOST_WEIGHT; if preference_boost != 0.0 { - snapshot.push(("preference_affinity".to_string(), preference_boost)); + snapshot.push((SignalKey::Static("preference_affinity"), preference_boost)); } } interaction_boost + preference_boost @@ -403,7 +439,7 @@ impl<'a> ProfileExecutor<'a> { mut extra_boost: F, ) -> crate::Result> where - F: FnMut(EntityId, &mut Vec<(String, f64)>) -> f64, + F: FnMut(EntityId, &mut SignalSnapshot) -> f64, { let empty: HashMap = HashMap::new(); // Lowercase the session keywords ONCE, before the candidate loop, rather @@ -415,7 +451,13 @@ impl<'a> ProfileExecutor<'a> { // "now" so the whole result set is consistent and re-running the same // query against an unchanged ledger is byte-identical (Accuracy-W). let now_ns = now.as_nanos(); - let mut scored: Vec = Vec::new(); + // Dynamic snapshot labels (`{signal}_boost`/`_penalty`/`_decay`) are + // invariant across candidates — build them once here, not per candidate. + let labels = RuleLabels::build(profile); + // Pre-size to the candidate count: every candidate that survives + // exclude/gate pushes exactly one entry, so this is the exact upper bound + // and removes the Vec's growth reallocations on the scoring hot path. + let mut scored: Vec = Vec::with_capacity(candidates.len()); for &entity_id in candidates { // Stage 2: hard exclusion -- remove content the user must never see. if !passes_excludes(entity_id, &profile.excludes, self.ledger, now_ns)? { @@ -426,7 +468,7 @@ impl<'a> ProfileExecutor<'a> { continue; } let (raw, mut snapshot) = - self.compute_raw_score(entity_id, profile, now, retrieval_scores)?; + self.compute_raw_score(entity_id, profile, now, retrieval_scores, &labels)?; let metadata = item_metadata.get(&entity_id.as_u64()).map_or(&empty, |m| m); let session_boost = session_ctx.map_or(0.0, |ctx| { Self::session_boost( @@ -552,7 +594,8 @@ impl<'a> ProfileExecutor<'a> { profile: &RankingProfile, now: Timestamp, retrieval_scores: Option<&HashMap>, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + labels: &RuleLabels, + ) -> crate::Result<(f64, SignalSnapshot)> { let now_ns = now.as_nanos(); let (sort_base, mut snapshot) = self.score_by_sort(entity_id, profile.sort.as_ref(), now)?; @@ -567,7 +610,7 @@ impl<'a> ProfileExecutor<'a> { let weighted = RELEVANCE_BASE_WEIGHT * relevance; base += weighted; if weighted != 0.0 { - snapshot.push(("relevance".to_string(), weighted)); + snapshot.push((SignalKey::Static("relevance"), weighted)); } } @@ -584,7 +627,7 @@ impl<'a> ProfileExecutor<'a> { // than their sort formula, so they are NOT subsumed and DO apply on top of // the sort base — preserving each profile's intended engagement weighting. let mut boost_sum = 0.0; - for b in &profile.boosts { + for (i, b) in profile.boosts.iter().enumerate() { if sort_subsumes_boost(profile.sort.as_ref(), &b.signal, &b.agg, b.window) { continue; } @@ -599,7 +642,7 @@ impl<'a> ProfileExecutor<'a> { )?; let weighted = b.weight * val; if weighted != 0.0 { - snapshot.push((format!("{}_boost", b.signal), weighted)); + snapshot.push((SignalKey::Owned(Arc::clone(&labels.boosts[i])), weighted)); } boost_sum += weighted; } @@ -620,7 +663,7 @@ impl<'a> ProfileExecutor<'a> { { let boost = f64::from(co_eng.score(seed, entity_id)) * 0.3; if boost != 0.0 { - snapshot.push(("co_engagement".to_string(), boost)); + snapshot.push((SignalKey::Static("co_engagement"), boost)); } boost } else { @@ -635,7 +678,7 @@ impl<'a> ProfileExecutor<'a> { // subsumed by a sort and always apply (a penalty is a complementary // demotion overlay, not a re-statement of the sort formula). let mut penalty_sum = 0.0; - for p in &profile.penalties { + for (i, p) in profile.penalties.iter().enumerate() { let val = read_agg_for_sort( entity_id, &p.signal, @@ -647,7 +690,10 @@ impl<'a> ProfileExecutor<'a> { )?; let weighted = p.weight * val; if weighted != 0.0 { - snapshot.push((format!("{}_penalty", p.signal), -weighted)); + snapshot.push(( + SignalKey::Owned(Arc::clone(&labels.penalties[i])), + -weighted, + )); } penalty_sum += weighted; } @@ -665,7 +711,7 @@ impl<'a> ProfileExecutor<'a> { // in `[0, 1]` blends the factor toward 1.0 so `weight = 0` is a no-op and // `weight = 1` applies the full decay. `half_life_secs` is informational // here -- the half-life is already encoded in the signal's decay rate. - let decayed = if let Some(decay) = &profile.decay { + let decayed = if let (Some(decay), Some(decay_label)) = (&profile.decay, &labels.decay) { let recency = read_agg_for_sort( entity_id, &decay.signal, @@ -678,7 +724,7 @@ impl<'a> ProfileExecutor<'a> { .clamp(0.0, 1.0); let factor = decay.weight.mul_add(recency, 1.0 - decay.weight); if (factor - 1.0).abs() > f64::EPSILON { - snapshot.push((format!("{}_decay", decay.signal), factor)); + snapshot.push((SignalKey::Owned(Arc::clone(decay_label)), factor)); } additive * factor } else { diff --git a/tidal/src/ranking/executor/scoring.rs b/tidal/src/ranking/executor/scoring.rs index 7a34572..cbd00b7 100644 --- a/tidal/src/ranking/executor/scoring.rs +++ b/tidal/src/ranking/executor/scoring.rs @@ -3,8 +3,10 @@ //! All methods in this module are part of the `ProfileExecutor` implementation. //! Extracted from `mod.rs` to keep file sizes manageable. +use smallvec::smallvec; + use super::{ - ProfileExecutor, + ProfileExecutor, SignalKey, SignalSnapshot, formulas::{ controversial_score, hidden_gems_score, hot_score, shuffle_quality_score, shuffle_quality_weight, shuffle_random, trending_score, @@ -78,7 +80,7 @@ impl ProfileExecutor<'_> { entity_id: EntityId, sort: Option<&Sort>, now: Timestamp, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { // Capture the query clock ONCE so every per-candidate ledger read below // ages to the same "now" (reproducible scores; one fewer syscall per read). let now_ns = now.as_nanos(); @@ -87,7 +89,7 @@ impl ProfileExecutor<'_> { Some(Sort::Trending) => self.score_trending(entity_id, now_ns), Some(Sort::Controversial) => self.score_controversial(entity_id, now_ns), Some(Sort::HiddenGems) => self.score_hidden_gems(entity_id, now_ns), - Some(Sort::Shuffle) => Ok((self.score_shuffle(entity_id, now_ns)?, vec![])), + Some(Sort::Shuffle) => Ok((self.score_shuffle(entity_id, now_ns)?, smallvec![])), Some(Sort::New) => { // M2 limitation: entity metadata (`created_at`) is not accessible from the // executor. Entity ID is used as a proxy for recency -- ranks higher IDs @@ -99,7 +101,7 @@ impl ProfileExecutor<'_> { // precision loss for very large IDs, which is acceptable for ranking). #[allow(clippy::cast_precision_loss)] let score = entity_id.as_u64() as f64; - Ok((score, vec![])) + Ok((score, smallvec![])) } Some(Sort::TopWindow { window }) => self.score_top_window(entity_id, *window, now_ns), Some(Sort::MostViewed { window }) => { @@ -136,17 +138,21 @@ impl ProfileExecutor<'_> { )?; Ok(( view_vel + like_vel, - vec![ - ("view_velocity".to_string(), view_vel), - ("like_velocity".to_string(), like_vel), + smallvec![ + (SignalKey::Static("view_velocity"), view_vel), + (SignalKey::Static("like_velocity"), like_vel), ], )) } Some(Sort::Rising) => self.score_rising(entity_id, now_ns), - Some(Sort::AlphabeticalAsc) => Ok((self.score_alphabetical_asc(entity_id), vec![])), - Some(Sort::AlphabeticalDesc) => Ok((self.score_alphabetical_desc(entity_id), vec![])), - Some(Sort::Shortest) => Ok((self.score_shortest(entity_id), vec![])), - Some(Sort::Longest) => Ok((self.score_longest(entity_id), vec![])), + Some(Sort::AlphabeticalAsc) => { + Ok((self.score_alphabetical_asc(entity_id), smallvec![])) + } + Some(Sort::AlphabeticalDesc) => { + Ok((self.score_alphabetical_desc(entity_id), smallvec![])) + } + Some(Sort::Shortest) => Ok((self.score_shortest(entity_id), smallvec![])), + Some(Sort::Longest) => Ok((self.score_longest(entity_id), smallvec![])), Some(Sort::MostCommented { window }) => { self.single_signal_score(entity_id, "comment", &SignalAgg::Value, *window, now_ns) } @@ -160,8 +166,8 @@ impl ProfileExecutor<'_> { Window::AllTime, now_ns, ), - Some(Sort::DateSaved) => Ok((self.score_date_saved(entity_id), vec![])), - None => Ok((0.0, vec![])), + Some(Sort::DateSaved) => Ok((self.score_date_saved(entity_id), smallvec![])), + None => Ok((0.0, smallvec![])), } } @@ -176,7 +182,7 @@ impl ProfileExecutor<'_> { entity_id: EntityId, gravity: f64, now: Timestamp, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let views = read_agg_for_sort( entity_id, "view", @@ -197,7 +203,7 @@ impl ProfileExecutor<'_> { let age_hours = 24.0_f64; Ok(( hot_score(views, age_hours, gravity), - vec![("view".to_string(), views)], + smallvec![(SignalKey::Static("view"), views)], )) } @@ -275,11 +281,11 @@ impl ProfileExecutor<'_> { fn single_signal_score( &self, entity_id: EntityId, - signal: &str, + signal: &'static str, agg: &SignalAgg, window: Window, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let val = read_agg_for_sort( entity_id, signal, @@ -289,14 +295,14 @@ impl ProfileExecutor<'_> { self.degradation_level, now_ns, )?; - Ok((val, vec![(signal.to_string(), val)])) + Ok((val, smallvec![(SignalKey::Static(signal), val)])) } fn score_trending( &self, entity_id: EntityId, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { // M6: social-graph-scoped trending. When a social subgraph and // per-user signal index are available, compute aggregate velocity // across the subgraph users instead of using the global ledger. @@ -334,9 +340,9 @@ impl ProfileExecutor<'_> { }); return Ok(( trending_score(view_vel, share_vel), - vec![ - ("view_velocity".to_string(), view_vel), - ("share_velocity".to_string(), share_vel), + smallvec![ + (SignalKey::Static("view_velocity"), view_vel), + (SignalKey::Static("share_velocity"), share_vel), ], )); } @@ -362,9 +368,9 @@ impl ProfileExecutor<'_> { )?; Ok(( trending_score(view_vel, share_vel), - vec![ - ("view_velocity".to_string(), view_vel), - ("share_velocity".to_string(), share_vel), + smallvec![ + (SignalKey::Static("view_velocity"), view_vel), + (SignalKey::Static("share_velocity"), share_vel), ], )) } @@ -373,7 +379,7 @@ impl ProfileExecutor<'_> { &self, entity_id: EntityId, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let pos = read_agg_for_sort( entity_id, "like", @@ -394,7 +400,10 @@ impl ProfileExecutor<'_> { )?; Ok(( controversial_score(pos, neg), - vec![("like".to_string(), pos), ("dislike".to_string(), neg)], + smallvec![ + (SignalKey::Static("like"), pos), + (SignalKey::Static("dislike"), neg) + ], )) } @@ -402,7 +411,7 @@ impl ProfileExecutor<'_> { &self, entity_id: EntityId, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let quality = read_agg_for_sort( entity_id, "completion", @@ -423,9 +432,9 @@ impl ProfileExecutor<'_> { )?; Ok(( hidden_gems_score(quality, view_count), - vec![ - ("completion".to_string(), quality), - ("view".to_string(), view_count), + smallvec![ + (SignalKey::Static("completion"), quality), + (SignalKey::Static("view"), view_count), ], )) } @@ -435,7 +444,7 @@ impl ProfileExecutor<'_> { entity_id: EntityId, window: Window, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let views = read_agg_for_sort( entity_id, "view", @@ -477,11 +486,11 @@ impl ProfileExecutor<'_> { 0.3, likes.mul_add(0.3, shares.mul_add(0.2, completion * views * 0.1)), ), - vec![ - ("view".to_string(), views), - ("like".to_string(), likes), - ("share".to_string(), shares), - ("completion".to_string(), completion), + smallvec![ + (SignalKey::Static("view"), views), + (SignalKey::Static("like"), likes), + (SignalKey::Static("share"), shares), + (SignalKey::Static("completion"), completion), ], )) } @@ -490,7 +499,7 @@ impl ProfileExecutor<'_> { &self, entity_id: EntityId, now_ns: u64, - ) -> crate::Result<(f64, Vec<(String, f64)>)> { + ) -> crate::Result<(f64, SignalSnapshot)> { let short = read_agg_for_sort( entity_id, "view", @@ -516,9 +525,9 @@ impl ProfileExecutor<'_> { }; Ok(( score, - vec![ - ("view_velocity_1h".to_string(), short), - ("view_velocity_24h".to_string(), long), + smallvec![ + (SignalKey::Static("view_velocity_1h"), short), + (SignalKey::Static("view_velocity_24h"), long), ], )) }