feat(replica): portable heartbeat-token fence with snapshot-local reader routing

crates/buzz-db/Cargo.toml

@@ -21,6 +21,8 @@ tracing     = { workspace = true }
 thiserror   = { workspace = true }
 nostr       = { workspace = true }
 rand        = { workspace = true }
+metrics     = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true }
+metrics-util = { workspace = true }

crates/buzz-db/src/event.rs

@@ -316,6 +316,17 @@ pub async fn insert_event(
 /// Uses `QueryBuilder` for dynamic filter composition — avoids string concatenation
 /// while keeping all user values in bind parameters.
 pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result<Vec<StoredEvent>> {
+    let mut conn = pool.acquire().await?;
+    query_events_on(&mut conn, q).await
+}
+
+/// [`query_events`] on a specific session — the replica-routing path runs
+/// follow-up (aux) queries on the exact reader connection whose heartbeat
+/// observation proved coverage for the page they annotate.
+pub(crate) async fn query_events_on(
+    conn: &mut sqlx::PgConnection,
+    q: &EventQuery,
+) -> Result<Vec<StoredEvent>> {
     // Composite cursor requires both halves.
     if q.before_id.is_some() && q.until.is_none() {
         return Err(DbError::InvalidData(
@@ -538,7 +549,7 @@ pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result<Vec<StoredEve
     qb.push_bind(limit_val);
     qb.push(" OFFSET ").push_bind(offset_val);
 
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb.build().fetch_all(&mut *conn).await?;
 
     let mut out = Vec::with_capacity(rows.len());
     for row in rows {
@@ -596,6 +607,14 @@ pub(crate) fn row_to_stored_event(row: sqlx::postgres::PgRow) -> Result<Option<S
 ///
 /// Uses the same filter logic as `query_events` but returns only the count.
 pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result<i64> {
+    let mut conn = pool.acquire().await?;
+    count_events_on(&mut conn, q).await
+}
+
+/// [`count_events`] on a specific session — the replica-routing path runs
+/// the count on the exact reader connection whose heartbeat observation
+/// proved its predicate.
+pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuery) -> Result<i64> {
     // Empty list means "match nothing" — return 0 immediately.
     if q.kinds.as_deref().is_some_and(|k| k.is_empty()) {
         return Ok(0);
@@ -730,7 +749,7 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result<i64> {
         }
     }
 
-    let row = qb.build().fetch_one(pool).await?;
+    let row = qb.build().fetch_one(&mut *conn).await?;
     let cnt: i64 = row.try_get("cnt")?;
 
     Ok(cnt)
@@ -990,6 +1009,21 @@ pub async fn get_events_by_ids(
     pool: &PgPool,
     community_id: CommunityId,
     ids: &[&[u8]],
+) -> Result<Vec<StoredEvent>> {
+    if ids.is_empty() {
+        return Ok(vec![]);
+    }
+    let mut conn = pool.acquire().await?;
+    get_events_by_ids_on(&mut conn, community_id, ids).await
+}
+
+/// [`get_events_by_ids`] on a specific session — the replica-routing path
+/// runs the query on the exact reader connection whose heartbeat
+/// observation proved its predicate.
+pub(crate) async fn get_events_by_ids_on(
+    conn: &mut sqlx::PgConnection,
+    community_id: CommunityId,
+    ids: &[&[u8]],
 ) -> Result<Vec<StoredEvent>> {
     if ids.is_empty() {
         return Ok(vec![]);
@@ -1008,7 +1042,7 @@ pub async fn get_events_by_ids(
     }
     qb.push(")");
 
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb.build().fetch_all(&mut *conn).await?;
 
     let mut out = Vec::with_capacity(rows.len());
     for row in rows {

crates/buzz-db/src/feed.rs

@@ -132,6 +132,29 @@ pub async fn query_mentions(
     accessible_channel_ids: &[Uuid],
     since: Option<DateTime<Utc>>,
     limit: i64,
+) -> Result<Vec<StoredEvent>> {
+    let mut conn = pool.acquire().await?;
+    query_mentions_on(
+        &mut conn,
+        community,
+        pubkey_bytes,
+        accessible_channel_ids,
+        since,
+        limit,
+    )
+    .await
+}
+
+/// [`query_mentions`] on a specific session — the replica-routing path runs
+/// the query on the exact reader connection whose heartbeat observation
+/// proved its predicate.
+pub(crate) async fn query_mentions_on(
+    conn: &mut sqlx::PgConnection,
+    community: CommunityId,
+    pubkey_bytes: &[u8],
+    accessible_channel_ids: &[Uuid],
+    since: Option<DateTime<Utc>>,
+    limit: i64,
 ) -> Result<Vec<StoredEvent>> {
     let mut qb = build_mentions_query(
         community,
@@ -140,7 +163,7 @@ pub async fn query_mentions(
         since,
         limit,
     );
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb.build().fetch_all(&mut *conn).await?;
     collect_stored_events(rows)
 }
 
@@ -193,6 +216,27 @@ pub async fn query_needs_action(
     accessible_channel_ids: &[Uuid],
     since: Option<DateTime<Utc>>,
     limit: i64,
+) -> Result<Vec<StoredEvent>> {
+    let mut conn = pool.acquire().await?;
+    query_needs_action_on(
+        &mut conn,
+        community,
+        pubkey_bytes,
+        accessible_channel_ids,
+        since,
+        limit,
+    )
+    .await
+}
+
+/// [`query_needs_action`] on a specific session — see [`query_mentions_on`].
+pub(crate) async fn query_needs_action_on(
+    conn: &mut sqlx::PgConnection,
+    community: CommunityId,
+    pubkey_bytes: &[u8],
+    accessible_channel_ids: &[Uuid],
+    since: Option<DateTime<Utc>>,
+    limit: i64,
 ) -> Result<Vec<StoredEvent>> {
     let mut qb = build_needs_action_query(
         community,
@@ -201,7 +245,7 @@ pub async fn query_needs_action(
         since,
         limit,
     );
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb.build().fetch_all(&mut *conn).await?;
     collect_stored_events(rows)
 }
 
@@ -241,9 +285,21 @@ pub async fn query_activity(
     accessible_channel_ids: &[Uuid],
     since: Option<DateTime<Utc>>,
     limit: i64,
+) -> Result<Vec<StoredEvent>> {
+    let mut conn = pool.acquire().await?;
+    query_activity_on(&mut conn, community, accessible_channel_ids, since, limit).await
+}
+
+/// [`query_activity`] on a specific session — see [`query_mentions_on`].
+pub(crate) async fn query_activity_on(
+    conn: &mut sqlx::PgConnection,
+    community: CommunityId,
+    accessible_channel_ids: &[Uuid],
+    since: Option<DateTime<Utc>>,
+    limit: i64,
 ) -> Result<Vec<StoredEvent>> {
     let mut qb = build_activity_query(community, accessible_channel_ids, since, limit);
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb.build().fetch_all(&mut *conn).await?;
     collect_stored_events(rows)
 }