diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 70449bd547..1590f63fde 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1616,265 +1616,6 @@ impl Db { dm::list_hidden_dms(&self.pool, community_id, pubkey).await } - /// Insert thread metadata. - #[allow(clippy::too_many_arguments)] - #[datastore_span(name = "insert_thread_metadata", system = "postgresql")] - pub async fn insert_thread_metadata( - &self, - community_id: CommunityId, - event_id: &[u8], - event_created_at: DateTime, - channel_id: Uuid, - parent_event_id: Option<&[u8]>, - parent_event_created_at: Option>, - root_event_id: Option<&[u8]>, - root_event_created_at: Option>, - depth: i32, - broadcast: bool, - ) -> Result<()> { - thread::insert_thread_metadata( - &self.pool, - community_id, - event_id, - event_created_at, - channel_id, - parent_event_id, - parent_event_created_at, - root_event_id, - root_event_created_at, - depth, - broadcast, - ) - .await - } - - /// Fetch replies under a root event. - /// - /// Routing mirrors [`Db::get_channel_window_with_session`]: a head - /// fetch (`cursor: None`) is Predicate A (bounded staleness, gated by - /// the default-off head budget); cursor pages are Predicate B - /// (completeness). Thread pagination walks **forward** from oldest to - /// newest, so a cursor carries no upper bound — instead the served page - /// is post-verified against the wall the serving session proved: - /// - /// - an under-`limit` page is a candidate terminal page — the client - /// treats it as EOF, so it is re-run on the writer to keep the EOF - /// decision authoritative (a lagged replica could truncate the tail); - /// - a full page whose newest row exceeds the proved fence wall could - /// straddle a row the session has not replayed (commit order is not - /// `created_at` order), so it is also re-run on the writer. Only a - /// full page that sits entirely at or below the proved wall is served - /// from the replica. - /// - /// A head fetch routed under Predicate A skips the re-run: bounded - /// staleness (missing at most the freshest budget-window of replies) is - /// exactly the semantic the head gate accepts. - #[datastore_span(name = "get_thread_replies", system = "postgresql")] - pub async fn get_thread_replies( - &self, - community_id: CommunityId, - root_event_id: &[u8], - depth_limit: Option, - limit: u32, - cursor: Option<&[u8]>, - ) -> Result> { - let (path, predicate): (&'static str, RoutePredicate) = match cursor { - Some(_) => ( - "thread_cursor", - RoutePredicate::CoveredPostVerified { - proof: ChannelScoped::from_thread_metadata_join(), - }, - ), - None => ("thread_head", RoutePredicate::Bounded), - }; - if let RouteDecision::Replica(mut tx, entry, reason) = - self.route_read(path, predicate).await - { - match thread::get_thread_replies_on( - &mut tx, - community_id, - root_event_id, - depth_limit, - limit, - cursor, - ) - .await - { - Ok(replies) => { - if cursor.is_none() { - // Predicate A: bounded-stale head page, served as proved. - Self::record_route(path, "replica", reason); - return Ok(replies); - } - let full = replies.len() >= limit as usize; - let below_fence = replies - .last() - .is_some_and(|tail| tail.created_at <= entry.fence_wall); - if full && below_fence { - Self::record_route(path, "replica", reason); - return Ok(replies); - } - // Candidate terminal page, or page reaching above the - // proved wall — verify against the writer. Recorded as - // the request's ONLY route event: the replica leg was - // discarded, so counting it would overstate offload. - Self::record_route("thread_eof", "writer", "stale"); - } - Err(e) => { - // Mid-request replica failure (e.g. a hot-standby - // recovery conflict) fails closed to the writer. - tracing::warn!( - error = %e, - path, - "replica thread query failed; re-running on writer" - ); - Self::record_route(path, "writer", "replica_error"); - } - } - } - thread::get_thread_replies( - &self.pool, - community_id, - root_event_id, - depth_limit, - limit, - cursor, - ) - .await - } - - /// Fetch aggregated thread stats. - #[datastore_span(name = "get_thread_summary", system = "postgresql")] - pub async fn get_thread_summary( - &self, - community_id: CommunityId, - event_id: &[u8], - ) -> Result> { - thread::get_thread_summary(&self.pool, community_id, event_id).await - } - - /// One channel window: top-level rows + summaries + server `has_more`. - /// - /// Convenience wrapper over [`Db::get_channel_window_with_session`] for - /// callers with no follow-up queries; the serving session is released. - pub async fn get_channel_window( - &self, - community_id: CommunityId, - channel_id: Uuid, - limit: u32, - cursor: Option<(DateTime, Vec)>, - kind_filter: Option<&[u32]>, - ) -> Result { - self.get_channel_window_with_session(community_id, channel_id, limit, cursor, kind_filter) - .await - .map(|(window, _session)| window) - } - - /// [`Db::get_channel_window`], additionally returning the session that - /// served the page so request-scoped follow-ups (the aux closure) run on - /// the same proved connection. - /// - /// Routing: - /// - /// - **Cursor page** (Predicate B — completeness): scrolls *backward* - /// into history bounded above by the cursor timestamp (`created_at < - /// ts`, or `= ts` with the id tiebreak), so it may be served by a - /// replica session when one is configured AND that session **proves** - /// coverage of the cursor timestamp: the heartbeat token/epoch is - /// observed on the exact connection that will serve the page and - /// resolved against the fence's retained ring ([`replica_fence`]). - /// - **Head fetch** (Predicate A — bounded staleness): served by a - /// proved replica session only when the head gate is configured - /// ([`DbConfig::replica_read_max_age_ms`], default off) and the - /// proved entry is within the budget. This trades a bounded staleness - /// window (budget plus probe cadence) on the GET leg for writer - /// offload. NOTE: enabling the budget also breaks read-your-own-writes - /// on the GET leg; the client-side WS `since`-overlap union intended - /// to cover fresh events has NOT shipped yet — do not enable - /// `BUZZ_REPLICA_HEAD_MAX_AGE_SECS` until it has, proven by a - /// post-then-immediately-refetch test. - /// - /// Every failure fails closed to the writer and is recorded in - /// `buzz_db_route_decision`. - #[datastore_span(name = "get_channel_window", system = "postgresql")] - pub async fn get_channel_window_with_session( - &self, - community_id: CommunityId, - channel_id: Uuid, - limit: u32, - cursor: Option<(DateTime, Vec)>, - kind_filter: Option<&[u32]>, - ) -> Result<(thread::ChannelWindow, ReadSession)> { - let path: &'static str = if cursor.is_some() { - "channel_cursor" - } else { - "channel_head" - }; - match self - .route_read( - path, - RoutePredicate::from_channel_cursor(channel_id, &cursor), - ) - .await - { - RouteDecision::Replica(mut tx, _entry, reason) => { - match thread::get_channel_window_on( - &mut tx, - community_id, - channel_id, - limit, - cursor.clone(), - kind_filter, - ) - .await - { - Ok(window) => { - Self::record_route(path, "replica", reason); - return Ok(( - window, - ReadSession { - inner: ReadSessionInner::Replica { - tx, - writer: self.pool.clone(), - }, - }, - )); - } - Err(e) => { - // A mid-request replica failure (e.g. a hot-standby - // recovery conflict cancelling the held snapshot) - // fails closed to the writer: a stale-but-served - // page, never an error the writer could have - // answered. Dropping `tx` rolls the reader - // transaction back. - tracing::warn!( - error = %e, - path, - "replica window query failed; re-running on writer" - ); - Self::record_route(path, "writer", "replica_error"); - } - } - } - RouteDecision::Writer => {} - } - let window = thread::get_channel_window( - &self.pool, - community_id, - channel_id, - limit, - cursor, - kind_filter, - ) - .await?; - Ok(( - window, - ReadSession { - inner: ReadSessionInner::Writer(self.pool.clone()), - }, - )) - } - /// Shared route decision for one read: evaluate the predicate against a /// proved reader session and record the decision. Fail closed to the /// writer everywhere. @@ -1961,28 +1702,6 @@ impl Db { } } - /// Look up a single thread_metadata row by event_id. - #[datastore_span(name = "get_thread_metadata_by_event", system = "postgresql")] - pub async fn get_thread_metadata_by_event( - &self, - community_id: CommunityId, - event_id: &[u8], - ) -> Result> { - thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await - } - - /// Decrement reply counts. - #[datastore_span(name = "decrement_reply_count", system = "postgresql")] - pub async fn decrement_reply_count( - &self, - community_id: CommunityId, - parent_event_id: &[u8], - root_event_id: Option<&[u8]>, - ) -> Result<()> { - thread::decrement_reply_count(&self.pool, community_id, parent_event_id, root_event_id) - .await - } - /// Add (or re-activate) a reaction. #[datastore_span(name = "add_reaction", system = "postgresql")] pub async fn add_reaction( diff --git a/crates/buzz-db/src/thread.rs b/crates/buzz-db/src/thread.rs index 007677e258..d7a2d239ef 100644 --- a/crates/buzz-db/src/thread.rs +++ b/crates/buzz-db/src/thread.rs @@ -9,9 +9,14 @@ use chrono::{DateTime, Utc}; use sqlx::{PgPool, Row}; use uuid::Uuid; +use buzz_datastore_tracing::datastore_span; + use buzz_core::CommunityId; -use crate::{error::Result, event::row_to_stored_event}; +use crate::{ + error::Result, event::row_to_stored_event, route_proof::ChannelScoped, Db, ReadSession, + ReadSessionInner, RouteDecision, RoutePredicate, +}; // -- Structs ------------------------------------------------------------------ @@ -856,6 +861,296 @@ pub async fn get_thread_metadata_by_event( })) } +// -- Db API ------------------------------------------------------------------- + +impl Db { + /// Insert thread metadata. + #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "insert_thread_metadata", system = "postgresql")] + pub async fn insert_thread_metadata( + &self, + community_id: CommunityId, + event_id: &[u8], + event_created_at: DateTime, + channel_id: Uuid, + parent_event_id: Option<&[u8]>, + parent_event_created_at: Option>, + root_event_id: Option<&[u8]>, + root_event_created_at: Option>, + depth: i32, + broadcast: bool, + ) -> Result<()> { + crate::thread::insert_thread_metadata( + &self.pool, + community_id, + event_id, + event_created_at, + channel_id, + parent_event_id, + parent_event_created_at, + root_event_id, + root_event_created_at, + depth, + broadcast, + ) + .await + } + + /// Fetch replies under a root event. + /// + /// Routing mirrors [`Db::get_channel_window_with_session`]: a head + /// fetch (`cursor: None`) is Predicate A (bounded staleness, gated by + /// the default-off head budget); cursor pages are Predicate B + /// (completeness). Thread pagination walks **forward** from oldest to + /// newest, so a cursor carries no upper bound — instead the served page + /// is post-verified against the wall the serving session proved: + /// + /// - an under-`limit` page is a candidate terminal page — the client + /// treats it as EOF, so it is re-run on the writer to keep the EOF + /// decision authoritative (a lagged replica could truncate the tail); + /// - a full page whose newest row exceeds the proved fence wall could + /// straddle a row the session has not replayed (commit order is not + /// `created_at` order), so it is also re-run on the writer. Only a + /// full page that sits entirely at or below the proved wall is served + /// from the replica. + /// + /// A head fetch routed under Predicate A skips the re-run: bounded + /// staleness (missing at most the freshest budget-window of replies) is + /// exactly the semantic the head gate accepts. + #[datastore_span(name = "get_thread_replies", system = "postgresql")] + pub async fn get_thread_replies( + &self, + community_id: CommunityId, + root_event_id: &[u8], + depth_limit: Option, + limit: u32, + cursor: Option<&[u8]>, + ) -> Result> { + let (path, predicate): (&'static str, RoutePredicate) = match cursor { + Some(_) => ( + "thread_cursor", + RoutePredicate::CoveredPostVerified { + proof: ChannelScoped::from_thread_metadata_join(), + }, + ), + None => ("thread_head", RoutePredicate::Bounded), + }; + if let RouteDecision::Replica(mut tx, entry, reason) = + self.route_read(path, predicate).await + { + match crate::thread::get_thread_replies_on( + &mut tx, + community_id, + root_event_id, + depth_limit, + limit, + cursor, + ) + .await + { + Ok(replies) => { + if cursor.is_none() { + // Predicate A: bounded-stale head page, served as proved. + Self::record_route(path, "replica", reason); + return Ok(replies); + } + let full = replies.len() >= limit as usize; + let below_fence = replies + .last() + .is_some_and(|tail| tail.created_at <= entry.fence_wall); + if full && below_fence { + Self::record_route(path, "replica", reason); + return Ok(replies); + } + // Candidate terminal page, or page reaching above the + // proved wall — verify against the writer. Recorded as + // the request's ONLY route event: the replica leg was + // discarded, so counting it would overstate offload. + Self::record_route("thread_eof", "writer", "stale"); + } + Err(e) => { + // Mid-request replica failure (e.g. a hot-standby + // recovery conflict) fails closed to the writer. + tracing::warn!( + error = %e, + path, + "replica thread query failed; re-running on writer" + ); + Self::record_route(path, "writer", "replica_error"); + } + } + } + crate::thread::get_thread_replies( + &self.pool, + community_id, + root_event_id, + depth_limit, + limit, + cursor, + ) + .await + } + + /// Fetch aggregated thread stats. + #[datastore_span(name = "get_thread_summary", system = "postgresql")] + pub async fn get_thread_summary( + &self, + community_id: CommunityId, + event_id: &[u8], + ) -> Result> { + crate::thread::get_thread_summary(&self.pool, community_id, event_id).await + } + + /// One channel window: top-level rows + summaries + server `has_more`. + /// + /// Convenience wrapper over [`Db::get_channel_window_with_session`] for + /// callers with no follow-up queries; the serving session is released. + pub async fn get_channel_window( + &self, + community_id: CommunityId, + channel_id: Uuid, + limit: u32, + cursor: Option<(DateTime, Vec)>, + kind_filter: Option<&[u32]>, + ) -> Result { + self.get_channel_window_with_session(community_id, channel_id, limit, cursor, kind_filter) + .await + .map(|(window, _session)| window) + } + + /// [`Db::get_channel_window`], additionally returning the session that + /// served the page so request-scoped follow-ups (the aux closure) run on + /// the same proved connection. + /// + /// Routing: + /// + /// - **Cursor page** (Predicate B — completeness): scrolls *backward* + /// into history bounded above by the cursor timestamp (`created_at < + /// ts`, or `= ts` with the id tiebreak), so it may be served by a + /// replica session when one is configured AND that session **proves** + /// coverage of the cursor timestamp: the heartbeat token/epoch is + /// observed on the exact connection that will serve the page and + /// resolved against the fence's retained ring ([`crate::replica_fence`]). + /// - **Head fetch** (Predicate A — bounded staleness): served by a + /// proved replica session only when the head gate is configured + /// ([`crate::DbConfig::replica_read_max_age_ms`], default off) and the + /// proved entry is within the budget. This trades a bounded staleness + /// window (budget plus probe cadence) on the GET leg for writer + /// offload. NOTE: enabling the budget also breaks read-your-own-writes + /// on the GET leg; the client-side WS `since`-overlap union intended + /// to cover fresh events has NOT shipped yet — do not enable + /// `BUZZ_REPLICA_HEAD_MAX_AGE_SECS` until it has, proven by a + /// post-then-immediately-refetch test. + /// + /// Every failure fails closed to the writer and is recorded in + /// `buzz_db_route_decision`. + #[datastore_span(name = "get_channel_window", system = "postgresql")] + pub async fn get_channel_window_with_session( + &self, + community_id: CommunityId, + channel_id: Uuid, + limit: u32, + cursor: Option<(DateTime, Vec)>, + kind_filter: Option<&[u32]>, + ) -> Result<(crate::thread::ChannelWindow, ReadSession)> { + let path: &'static str = if cursor.is_some() { + "channel_cursor" + } else { + "channel_head" + }; + match self + .route_read( + path, + RoutePredicate::from_channel_cursor(channel_id, &cursor), + ) + .await + { + RouteDecision::Replica(mut tx, _entry, reason) => { + match crate::thread::get_channel_window_on( + &mut tx, + community_id, + channel_id, + limit, + cursor.clone(), + kind_filter, + ) + .await + { + Ok(window) => { + Self::record_route(path, "replica", reason); + return Ok(( + window, + ReadSession { + inner: ReadSessionInner::Replica { + tx, + writer: self.pool.clone(), + }, + }, + )); + } + Err(e) => { + // A mid-request replica failure (e.g. a hot-standby + // recovery conflict cancelling the held snapshot) + // fails closed to the writer: a stale-but-served + // page, never an error the writer could have + // answered. Dropping `tx` rolls the reader + // transaction back. + tracing::warn!( + error = %e, + path, + "replica window query failed; re-running on writer" + ); + Self::record_route(path, "writer", "replica_error"); + } + } + } + RouteDecision::Writer => {} + } + let window = crate::thread::get_channel_window( + &self.pool, + community_id, + channel_id, + limit, + cursor, + kind_filter, + ) + .await?; + Ok(( + window, + ReadSession { + inner: ReadSessionInner::Writer(self.pool.clone()), + }, + )) + } + + /// Look up a single thread_metadata row by event_id. + #[datastore_span(name = "get_thread_metadata_by_event", system = "postgresql")] + pub async fn get_thread_metadata_by_event( + &self, + community_id: CommunityId, + event_id: &[u8], + ) -> Result> { + crate::thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await + } + + /// Decrement reply counts. + #[datastore_span(name = "decrement_reply_count", system = "postgresql")] + pub async fn decrement_reply_count( + &self, + community_id: CommunityId, + parent_event_id: &[u8], + root_event_id: Option<&[u8]>, + ) -> Result<()> { + crate::thread::decrement_reply_count( + &self.pool, + community_id, + parent_event_id, + root_event_id, + ) + .await + } +} + #[cfg(test)] mod tests { use super::*; @@ -865,7 +1160,7 @@ mod tests { }; use nostr::{EventBuilder, Keys, Kind}; - const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 async fn setup_pool() -> PgPool { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")