diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 54db344..d15c2f7 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -2050,13 +2050,28 @@ impl ShardReplica { signal: &str, entity: EntityId, weight: f64, + user_id: Option, + creator_id: Option, ) -> Result { if !self.is_leader() { return Err(self.not_leader()); } let db = self.db()?; - db.signal_staged(signal, entity, weight, Timestamp::now()) - .map_err(ServerError::Tidal) + // Always the context-carrying stage: it degrades to the plain staged + // write when both ids are `None`, so there is no second code path to + // keep in step. Using `signal_staged` here is what silently discarded + // `user_id`/`creator_id` on every clustered write while still + // answering 204 — no hard negatives, no seen tracking, no interaction + // weight, no preference vector, and no wire evidence of the loss. + db.signal_with_context_staged( + signal, + entity, + weight, + Timestamp::now(), + user_id, + creator_id, + ) + .map_err(ServerError::Tidal) } /// Complete a staged signal write: block until the leader group-commit @@ -7486,6 +7501,8 @@ pub async fn write_signal( let signal = req.signal; let entity = EntityId::new(req.entity_id); let weight = req.weight; + let user_id = req.user_id; + let creator_id = req.creator_id; // Two-phase write (m11p1): STAGE on the write pool (microseconds; the // bounded queue keeps the 429 admission semantics), then COMPLETE — the // group-commit fsync wait — on the blocking pool, freeing the pool worker @@ -7506,7 +7523,8 @@ pub async fn write_signal( let ticket = state .write_pool .submit(move || { - let staged = state_for_job.stage_signal_local(&signal, entity, weight)?; + let staged = + state_for_job.stage_signal_local(&signal, entity, weight, user_id, creator_id)?; Ok(StagedWriteTicket::new(staged, Arc::clone(&state_for_job))) }) .await @@ -9513,6 +9531,8 @@ pub async fn sharded_write_signal( let entity = EntityId::new(req.entity_id); let signal = req.signal.clone(); let weight = req.weight; + let user_id = req.user_id; + let creator_id = req.creator_id; sharded_write_route( &state, &headers, @@ -9520,8 +9540,15 @@ pub async fn sharded_write_signal( "/sharded/signals", &req, move || { - db.signal(&signal, entity, weight, Timestamp::now()) - .map_err(ServerError::Tidal) + db.signal_with_context( + &signal, + entity, + weight, + Timestamp::now(), + user_id, + creator_id, + ) + .map_err(ServerError::Tidal) }, StatusCode::NO_CONTENT, ) diff --git a/tidal-server/src/cluster/routes.rs b/tidal-server/src/cluster/routes.rs index 4900b15..fca6127 100644 --- a/tidal-server/src/cluster/routes.rs +++ b/tidal-server/src/cluster/routes.rs @@ -395,6 +395,30 @@ pub async fn write_embedding( Ok(StatusCode::NO_CONTENT) } +/// Refuse a signal whose originating context this router cannot honour. +/// +/// `SimulatedCluster::write_signal` and `scatter_gather::sharded_write_signal` +/// both apply `(signal, entity, weight)` and nothing else, so `user_id` / +/// `creator_id` would be accepted and thrown away — no hard negatives, no seen +/// tracking, no interaction weight, no preference vector, and a 204 claiming +/// success. Fail closed instead, naming the route that does support it. +/// +/// # Errors +/// +/// `ServerError::BadRequest` when either context id is present. +fn reject_unsupported_signal_context(req: &SignalRequest) -> Result<()> { + if req.user_id.is_some() || req.creator_id.is_some() { + return Err(ServerError::BadRequest( + "single-process cluster mode cannot record signal context: \ + user_id/creator_id are unsupported on this router. Run the \ + multi-process cluster (`--region`, the deployed RF3 topology), \ + which applies context via signal_with_context_staged." + .to_owned(), + )); + } + Ok(()) +} + /// Record a signal on the leader region and eagerly ship it to followers. /// /// Returns `204 No Content` once the signal is **durably applied on the leader** @@ -421,6 +445,13 @@ pub async fn write_signal( State(state): State>, Json(req): Json, ) -> std::result::Result { + // The simulated relay behind single-process mode applies (signal, entity, + // weight) only, so it CANNOT honour originating context. Refuse rather than + // accept-and-discard: a 204 on a dropped `user_id` is the failure mode that + // made a clustered deployment look healthy while learning nothing. The + // production multi-process path (`build_region_router`) carries context + // properly — see `ClusterNode::stage_signal_local`. + reject_unsupported_signal_context(&req).map_err(ClusterAppError)?; // write_signal ships to followers over gRPC (a blocking `runtime.block_on`), // so it must run off the async reactor AND off any thread carrying a runtime // handle — hand it to the runtime-free write pool. A saturated pool yields @@ -658,6 +689,9 @@ pub async fn sharded_write_signal( Json(req): Json, ) -> std::result::Result { require_local_ack(&headers, "/sharded/signals").map_err(ClusterAppError)?; + // Same reason as `write_signal`: the scatter-gather write applies + // (signal, entity, weight) only and would discard context behind a 204. + reject_unsupported_signal_context(&req).map_err(ClusterAppError)?; let shards = state.shard_ids().map_err(ClusterAppError)?; // Offload the blocking single-shard signal write off the reactor (see // [`sharded_create_item`]). diff --git a/tidal-server/src/dto.rs b/tidal-server/src/dto.rs index 489acbd..9743166 100644 --- a/tidal-server/src/dto.rs +++ b/tidal-server/src/dto.rs @@ -73,10 +73,15 @@ pub struct SignalRequest { /// Signal weight applied to the running decay score. #[schema(example = 1.0)] pub weight: f64, - /// Optional originating user context (standalone path only). + /// Optional originating user context. + /// + /// Honoured by standalone and by the deployed multi-process cluster. The + /// experimental single-process cluster router refuses a request carrying + /// it rather than dropping it (`/signals` → 400). #[serde(default)] pub user_id: Option, - /// Optional originating creator context (standalone path only). + /// Optional originating creator context. Same support matrix as + /// [`Self::user_id`]; drives the `(user, creator)` interaction weight. #[serde(default)] pub creator_id: Option, } diff --git a/tidal/src/db/signals.rs b/tidal/src/db/signals.rs index 5af716e..7b68a68 100644 --- a/tidal/src/db/signals.rs +++ b/tidal/src/db/signals.rs @@ -18,12 +18,32 @@ use crate::{ signals::StagedLedgerApply, }; +/// The originating user/creator context of a signal write. +/// +/// Carried on a [`StagedSignal`] so the two-phase path applies the SAME +/// side effects as [`TidalDb::signal_with_context`] — see +/// [`TidalDb::apply_signal_context`]. Owns its signal-type name because the +/// staged write outlives the borrow of the request that produced it. +#[derive(Debug)] +struct SignalContext { + signal_type: String, + entity_id: EntityId, + weight: f64, + timestamp: Timestamp, + for_user: Option, + creator_id: Option, +} + /// A staged signal write on a [`TidalDb`]: admission-checked and WAL-submitted, /// durability and in-memory fold pending. Created by -/// [`TidalDb::signal_staged`]; completed by [`wait`](Self::wait). +/// [`TidalDb::signal_staged`] or +/// [`TidalDb::signal_with_context_staged`]; completed by [`wait`](Self::wait). #[derive(Debug)] pub struct StagedSignal { staged: StagedLedgerApply, + /// `Some` only for a `signal_with_context_staged` write. `None` leaves + /// `wait` byte-for-byte equivalent to the pre-context behaviour. + context: Option, #[cfg(feature = "metrics")] write_start: std::time::Instant, } @@ -33,6 +53,12 @@ impl StagedSignal { /// in-memory aggregate (identical end state to a completed /// [`TidalDb::signal`] call, including the write-latency metrics). /// + /// For a write staged with originating context, the user/creator side + /// effects (hard negatives, seen, interaction weight, preference vector, + /// cohort and community forwarding) are applied here, AFTER durability — + /// the same order [`TidalDb::signal_with_context`] uses, so a crash can + /// never leave a side effect whose base signal was never logged. + /// /// Returns the event's assigned WAL seqno — the replicated-stream /// position quorum acks gate on (m11p3). `0` = suppressed by the dedup /// window (an identical record is already durable; its quorum status is @@ -47,6 +73,19 @@ impl StagedSignal { pub fn wait(self, db: &TidalDb) -> crate::Result { let result = db.ledger()?.complete_staged(self.staged); + if result.is_ok() + && let Some(ctx) = self.context + { + db.apply_signal_context( + &ctx.signal_type, + ctx.entity_id, + ctx.weight, + ctx.timestamp, + ctx.for_user, + ctx.creator_id, + ); + } + #[cfg(feature = "metrics")] if result.is_ok() { use std::sync::atomic::Ordering; @@ -277,7 +316,74 @@ impl TidalDb { weight: f64, timestamp: Timestamp, ) -> crate::Result { - self.require_writeable("signal_staged")?; + self.stage_signal_inner( + "signal_staged", + signal_type, + entity_id, + weight, + timestamp, + None, + ) + } + + /// Two-phase counterpart of + /// [`signal_with_context`](Self::signal_with_context). + /// + /// Staging validates and submits the base signal; [`StagedSignal::wait`] + /// makes it durable and THEN applies the user/creator side effects. This is + /// the write a replicated cluster leader needs: `signal_with_context` would + /// serialise every context-carrying write on its own fsync, and + /// `signal_staged` silently discards the context. + /// + /// # Errors + /// + /// Same admission errors as [`signal_staged`](Self::signal_staged), plus + /// `InvalidInput` if `for_user` is set and `entity_id` exceeds the `u32` + /// item-universe limit (see `signal_with_context` for why that aliasing is + /// rejected up front rather than truncated). + pub fn signal_with_context_staged( + &self, + signal_type: &str, + entity_id: EntityId, + weight: f64, + timestamp: Timestamp, + for_user: Option, + creator_id: Option, + ) -> crate::Result { + Self::validate_context_entity(entity_id, for_user)?; + // No context at all ⇒ no side effects to apply; take the plain path so + // `wait` does not carry a pointless allocation per write. + let context = (for_user.is_some() || creator_id.is_some()).then(|| SignalContext { + signal_type: signal_type.to_owned(), + entity_id, + weight, + timestamp, + for_user, + creator_id, + }); + self.stage_signal_inner( + "signal_with_context_staged", + signal_type, + entity_id, + weight, + timestamp, + context, + ) + } + + /// Shared staging body for [`signal_staged`](Self::signal_staged) and + /// [`signal_with_context_staged`](Self::signal_with_context_staged): one + /// admission sequence, one `StagedSignal` construction. + fn stage_signal_inner( + &self, + op: &'static str, + signal_type: &str, + entity_id: EntityId, + weight: f64, + timestamp: Timestamp, + context: Option, + ) -> crate::Result { + self.require_writeable(op)?; Self::validate_signal_weight(weight)?; self.check_write_backpressure()?; @@ -290,6 +396,7 @@ impl TidalDb { Ok(StagedSignal { staged, + context, #[cfg(feature = "metrics")] write_start, }) @@ -648,16 +755,39 @@ impl TidalDb { creator_id: Option, ) -> crate::Result<()> { self.require_writeable("signal_with_context")?; + Self::validate_context_entity(entity_id, for_user)?; - // When a `for_user` identity is present this call narrows the item id to - // its u32 slot and writes it into DURABLE Tag::HardNeg / Tag::UserState - // rows. A bare `as u32` truncation would silently alias two items whose - // ids share their low 32 bits — a permanent cross-id collision in the - // hard-negative / seen / saved / liked correctness primitives that - // survives restart. Reject an over-range id up front (mirroring - // `write_item_with_metadata`) so a colliding durable row never lands on - // disk. Done before the base `signal()` so a rejected write leaves no - // trace at all. + // Record the base signal. + self.signal(signal_type, entity_id, weight, timestamp)?; + + self.apply_signal_context( + signal_type, + entity_id, + weight, + timestamp, + for_user, + creator_id, + ); + Ok(()) + } + + /// Reject an item id that cannot round-trip through the `u32` item slot. + /// + /// When a `for_user` identity is present the write narrows the item id to + /// its u32 slot and stores it in DURABLE `Tag::HardNeg` / `Tag::UserState` + /// rows. A bare `as u32` truncation would silently alias two items whose + /// ids share their low 32 bits — a permanent cross-id collision in the + /// hard-negative / seen / saved / liked correctness primitives that + /// survives restart. Rejecting up front (mirroring + /// `write_item_with_metadata`) keeps a colliding durable row off disk, and + /// doing it BEFORE the base signal leaves a rejected write with no trace at + /// all. + /// + /// # Errors + /// + /// `TidalError::InvalidInput` if `for_user` is set and `entity_id` exceeds + /// `u32::MAX`. + fn validate_context_entity(entity_id: EntityId, for_user: Option) -> crate::Result<()> { if for_user.is_some() && entity_id.as_u64() > u64::from(u32::MAX) { let raw = entity_id.as_u64(); return Err(TidalError::invalid_input(format!( @@ -666,10 +796,30 @@ impl TidalDb { u32::MAX ))); } + Ok(()) + } - // Record the base signal. - self.signal(signal_type, entity_id, weight, timestamp)?; - + /// Apply the user/creator side effects of a signal whose base write is + /// already durable. + /// + /// Shared by [`signal_with_context`](Self::signal_with_context) and the + /// two-phase [`StagedSignal::wait`] so the replicated cluster path and the + /// standalone path cannot diverge — the divergence this replaced silently + /// dropped `user_id`/`creator_id` on every clustered write while still + /// answering 204. + /// + /// Infallible by construction: every step is either in-memory or a + /// best-effort durable write that logs its own failure. The base signal has + /// already succeeded, so a side-effect failure must not retract it. + fn apply_signal_context( + &self, + signal_type: &str, + entity_id: EntityId, + weight: f64, + timestamp: Timestamp, + for_user: Option, + creator_id: Option, + ) { // pg1: record user's most recent signal timestamp for staleness/feedback-loop tracking. #[cfg(feature = "metrics")] if let Some(user_id) = for_user { @@ -757,8 +907,6 @@ impl TidalDb { // 8. Community forwarding (M9): forward to opted-in community aggregates. self.try_community_forwarding(signal_type, entity_id, weight, timestamp, user_id); } - - Ok(()) } /// Persist the current `(user, creator)` interaction weight as a durable diff --git a/tidal/tests/staged_signal_context.rs b/tidal/tests/staged_signal_context.rs new file mode 100644 index 0000000..d4604bd --- /dev/null +++ b/tidal/tests/staged_signal_context.rs @@ -0,0 +1,181 @@ +//! The two-phase signal write must carry originating context. +//! +//! `signal_staged` applies `(signal, entity, weight)` and nothing else. The +//! replicated cluster leader used it for every `POST /signals`, so a clustered +//! deployment accepted each behavioural signal, answered `204`, and silently +//! discarded `user_id`/`creator_id` — no hard negatives, no seen tracking, no +//! interaction weight, no preference vector, and no wire evidence of the loss. +//! `TidalDb::signal_with_context_staged` is the fix; these tests pin it. +//! +//! The shape is DIFFERENTIAL on purpose: each test drives the same signals +//! through the synchronous `signal_with_context` and through the staged path, +//! then asserts the two databases reach the same observable state. A staged +//! path that drops context fails these on the assertions rather than needing a +//! hand-written expected value, which is exactly the check the original bug got +//! past. + +use std::time::Duration; + +use tidaldb::TidalDb; +use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, Timestamp, Window}; + +const USER: u64 = 77; +const CREATOR: u64 = 900; +const ITEM: u64 = 4_242; + +fn schema() -> Schema { + let mut builder = SchemaBuilder::new(); + // `dislike` is a hard-negative signal; `like` is positive engagement. Both + // branches of the context dispatch need coverage. + for (name, half_life_days) in [("like", 14_u64), ("dislike", 1)] { + let _ = builder + .signal( + name, + EntityKind::Item, + DecaySpec::Exponential { + half_life: Duration::from_secs(half_life_days * 24 * 3600), + }, + ) + .windows(&[Window::TwentyFourHours, Window::SevenDays, Window::AllTime]) + .velocity(true) + .add(); + } + builder.build().expect("schema must be valid") +} + +fn open_db() -> TidalDb { + TidalDb::builder() + .ephemeral() + .with_schema(schema()) + .open() + .expect("db open") +} + +/// `(is_negative, is_seen, interaction_score)` — the durable, user-scoped +/// consequences of a context-carrying signal. +fn observed(db: &TidalDb, now_ns: u64) -> (bool, bool, f64) { + let item_slot = u32::try_from(ITEM).expect("test item id fits u32"); + ( + db.hard_negatives().is_negative(USER, item_slot), + db.user_state().is_seen(USER, item_slot), + db.interaction_ledger().score(USER, CREATOR, now_ns), + ) +} + +#[test] +fn staged_context_matches_synchronous_for_hard_negative() { + let ts = Timestamp::now(); + let sync_db = open_db(); + sync_db + .signal_with_context("dislike", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR)) + .expect("synchronous context write"); + + let staged_db = open_db(); + staged_db + .signal_with_context_staged("dislike", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR)) + .expect("stage") + .wait(&staged_db) + .expect("complete"); + + let now = ts.as_nanos(); + let sync = observed(&sync_db, now); + let staged = observed(&staged_db, now); + + // Positive control: the synchronous path really does record all three, so a + // passing comparison cannot be two empty states agreeing with each other. + assert!(sync.0, "synchronous dislike must record a hard negative"); + assert!(sync.1, "synchronous dislike must mark the item seen"); + assert!( + sync.2 > 0.0, + "synchronous dislike must record interaction weight" + ); + + assert_eq!( + staged.0, sync.0, + "staged write dropped the hard negative — user_id did not reach the engine" + ); + assert_eq!( + staged.1, sync.1, + "staged write dropped seen tracking — user_id did not reach the engine" + ); + assert!( + (staged.2 - sync.2).abs() < 1e-9, + "staged write dropped interaction weight: staged={} sync={}", + staged.2, + sync.2 + ); +} + +#[test] +fn staged_context_matches_synchronous_for_positive_engagement() { + let ts = Timestamp::now(); + let sync_db = open_db(); + sync_db + .signal_with_context("like", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR)) + .expect("synchronous context write"); + + let staged_db = open_db(); + staged_db + .signal_with_context_staged("like", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR)) + .expect("stage") + .wait(&staged_db) + .expect("complete"); + + let now = ts.as_nanos(); + let sync = observed(&sync_db, now); + let staged = observed(&staged_db, now); + + // A like is not a hard negative — pinned so a future dispatch change cannot + // start hiding liked items without failing here. + assert!(!sync.0, "a like must not record a hard negative"); + assert!(sync.1, "a like must mark the item seen"); + assert!(sync.2 > 0.0, "a like must record interaction weight"); + + assert_eq!(staged, sync, "staged like diverged from synchronous like"); +} + +#[test] +fn staged_write_without_context_still_records_the_base_signal() { + let ts = Timestamp::now(); + let db = open_db(); + db.signal_with_context_staged("like", ITEM.into(), 1.0, ts, None, None) + .expect("stage") + .wait(&db) + .expect("complete"); + + let item_slot = u32::try_from(ITEM).expect("test item id fits u32"); + // No context supplied ⇒ no user-scoped side effects, and in particular no + // attribution to a user that was never named. + assert!(!db.user_state().is_seen(USER, item_slot)); + assert!(!db.hard_negatives().is_negative(USER, item_slot)); + assert!( + db.read_decay_score(ITEM.into(), "like", 0) + .expect("read score") + .is_some_and(|score| score > 0.0), + "the base signal must still be recorded without context" + ); +} + +#[test] +fn staged_context_rejects_an_item_id_past_the_u32_universe() { + let db = open_db(); + // The u32 item-slot guard must fire at STAGING, before anything is written — + // a durable Tag::HardNeg row keyed on a truncated id is a permanent + // cross-item collision that survives restart. + let over_range = u64::from(u32::MAX) + 1; + let err = db + .signal_with_context_staged( + "dislike", + over_range.into(), + 1.0, + Timestamp::now(), + Some(USER), + None, + ) + .expect_err("an over-range item id with a user context must be rejected"); + let message = err.to_string(); + assert!( + message.contains("u32 item-universe limit"), + "unexpected error: {message}" + ); +}