diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index ec2ebd2254..91026d8210 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -456,6 +456,16 @@ pub async fn insert_event_in_transaction( .await?; } crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?; + if let Some(channel) = channel_id { + crate::personal_read::record_timeline_arrival( + tx, + community_id, + event, + channel, + result.0.received_at, + ) + .await?; + } } Ok(result) } @@ -1899,6 +1909,24 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .await?; } crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?; + // Last, so the channel row lock is held only until commit. + let timeline = match &thread_meta { + None => channel_id, + Some(meta) if meta.parent_event_id.is_none() || (meta.depth == 1 && meta.broadcast) => { + Some(meta.channel_id) + } + Some(_) => None, + }; + if let Some(channel) = timeline { + crate::personal_read::record_timeline_arrival( + tx, + community_id, + event, + channel, + received_at, + ) + .await?; + } } Ok(( diff --git a/crates/buzz-db/src/store/personal_read/membership.rs b/crates/buzz-db/src/store/personal_read/membership.rs index a1732dd456..daa79bb9c3 100644 --- a/crates/buzz-db/src/store/personal_read/membership.rs +++ b/crates/buzz-db/src/store/personal_read/membership.rs @@ -121,3 +121,31 @@ pub(crate) async fn record_message( .await?; Ok(()) } + +/// Note that an eligible timeline message (top-level, or a broadcast reply) +/// arrived in `channel`. Run it as the ingest transaction's last statement: +/// concurrent posts to one channel queue on this row lock until commit. +/// `GREATEST` makes their order irrelevant. Key-share locks from inserts that +/// reference the channel do not conflict with this non-key update. +pub(crate) async fn record_timeline_arrival( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + event: &Event, + channel: Uuid, + received_at: DateTime, +) -> Result<()> { + if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) { + return Ok(()); + } + sqlx::query( + "UPDATE channels SET last_timeline_received_at = + GREATEST(last_timeline_received_at, $3) + WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(received_at) + .execute(&mut **tx) + .await?; + Ok(()) +} diff --git a/crates/buzz-db/src/store/personal_read/mod.rs b/crates/buzz-db/src/store/personal_read/mod.rs index 79657bf565..7507244122 100644 --- a/crates/buzz-db/src/store/personal_read/mod.rs +++ b/crates/buzz-db/src/store/personal_read/mod.rs @@ -12,7 +12,7 @@ mod model; mod projection; mod writes; -pub(crate) use membership::{record_message, Place}; +pub(crate) use membership::{record_message, record_timeline_arrival, Place}; pub use model::*; #[cfg(test)] diff --git a/crates/buzz-db/src/store/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs index bf8da6ba65..0aac481022 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -639,6 +639,74 @@ async fn posting_marks_read_and_read_through_skips_deleted_and_auxiliary() { assert!(!empty.unread && empty.mentions == 0 && empty.read_through_id.is_none()); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_unread_follows_timeline_arrivals_only() { + let (db, pool, community, channel, actor, root) = fixture().await; + let base = now(); + let other = Keys::generate(); + let last = |pool: PgPool| async move { + sqlx::query_scalar::<_, Option>>( + "SELECT last_timeline_received_at FROM channels WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(&pool) + .await + .unwrap() + }; + assert_eq!( + apply(&db, community, &actor, mark(channel, None, &root)).await, + IntentOutcome::Applied + ); + let caught_up = last(pool.clone()).await; + assert!(caught_up.is_some(), "the fixture's message set it"); + assert!(!sidebar(&db, community, &actor).await.unread); + + // A thread reply is not on the timeline: the channel stays read. + reply( + &db, + community, + channel, + &root, + &other, + base + 1, + vec![], + false, + ) + .await; + assert_eq!(last(pool.clone()).await, caught_up); + assert!(!sidebar(&db, community, &actor).await.unread); + + // A reaction (not an eligible kind) doesn't move it; a broadcast reply, + // which shows on the timeline, does. + post_kind(&db, community, channel, &other, 7, base + 2, vec![]).await; + assert_eq!(last(pool.clone()).await, caught_up); + let broadcast = reply( + &db, + community, + channel, + &root, + &other, + base + 3, + vec![], + true, + ) + .await; + assert!(last(pool.clone()).await > caught_up); + assert!(sidebar(&db, community, &actor).await.unread); + + // Deleting the only unread timeline message leaves the channel read: the + // probe confirms what the arrival time suggests. + sqlx::query("UPDATE events SET deleted_at=clock_timestamp() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(broadcast.id.as_bytes().as_slice()) + .execute(&pool) + .await + .unwrap(); + assert!(!sidebar(&db, community, &actor).await.unread); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn top_level_message_with_depth_zero_metadata_marks_read_and_follows() { diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index 5c2fb57e20..bb30411f7d 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -84,8 +84,11 @@ impl Db { // cover them all, so a follow that survived leaving does not bring // back replies from the absence. // - // - unread: the first timeline message past the position - // (idx_events_community_channel_created); + // - unread: nothing when the channel's last timeline arrival is at + // or before the position (one comparison, so a caught-up reader + // never walks replies); otherwise the first timeline message past + // it (idx_events_community_channel_created), which rules out a + // deleted latest message; // - mentions: the actor's mentions in that scope that arrived past // the position (idx_event_mentions_scope_received); // - read_through_id: the newest timeline message, by author time, that @@ -100,6 +103,7 @@ impl Db { r#"WITH roster AS MATERIALIZED ( SELECT c.id, c.name, c.channel_type::text AS channel_type, c.archived_at IS NOT NULL AS archived, cm.hidden_at IS NOT NULL AS hidden, + c.last_timeline_received_at, GREATEST(cf.through_timestamp, cm.joined_at, s.started) AS position, GREATEST(cf.threads_through_timestamp, cm.joined_at, s.started) AS threads_floor FROM channel_members cm JOIN channels c @@ -114,7 +118,7 @@ impl Db { ORDER BY c.id LIMIT $5 ) SELECT r.id, r.name, r.channel_type, r.archived, r.hidden, - EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW} + COALESCE(r.last_timeline_received_at > r.position, false) AND EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW} WHERE e.community_id=$1 AND e.channel_id=r.id AND {UNREAD_EVENT} AND {ON_TIMELINE}) AS unread, (SELECT count(*) FROM event_mentions m JOIN events e ON e.community_id=$1 diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index 218090a302..58a5965cfa 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -215,8 +215,11 @@ device-local. - 20 sidebar rows, 100 intents, 20 contexts / 100 selectors per request. - 64 KiB write body; 16 KiB context URL; 1 MiB serialized API response. -- `unread` is an existence probe that stops at the first message past the - position, so a caught-up reader examines almost nothing. Counting is forward +- A channel whose last timeline arrival (`channels.last_timeline_received_at`) + is at or before the position is read with one comparison, without walking + past replies. Otherwise `unread` is an existence probe that stops at the + first message past the position, which also rules out a deleted latest + message. Counting is forward from the position: work grows with what is unread, not with history. - Mention counts walk only your mentions in that scope past the position (`idx_event_mentions_scope_received`). @@ -228,15 +231,19 @@ device-local. containment, not a production capacity claim. Apply migration 0056 (or the equivalent desired schema). It creates two -private tables, adds `thread_metadata.last_reply_received_at` and +private tables, adds `thread_metadata.last_reply_received_at`, +`channels.last_timeline_received_at` and `event_mentions.received_at`/`root_id` (NULL for rows from before it, which no position can reach), and one index on `event_mentions`. Ingest writes the author's own position and follow rows in the same transaction as the message. During a rolling deploy, relay processes still on the old code write `event_mentions` rows without `received_at`/`root_id` and do not mark posting -read. Those mentions never count, and an author's own message from such a -process can show unread to them until they next read the scope. +read. Those mentions never count, an author's own message from such a +process can show unread to them until they next read the scope, and a +channel or thread whose only new messages came through such a process shows +read until its next message through new code (the arrival columns are not +updated). Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and `/buzz/v1/me/read-state`, plus database pool/statement metrics. No payload, diff --git a/migrations/0056_personal_read_state.sql b/migrations/0056_personal_read_state.sql index cc0629af0e..4ea1b7e881 100644 --- a/migrations/0056_personal_read_state.sql +++ b/migrations/0056_personal_read_state.sql @@ -41,6 +41,14 @@ CREATE TABLE personal_read_frontiers ( -- with nothing new since the reader's position without reading replies. ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ; +-- Arrival time of a channel's latest eligible timeline message (top-level or +-- broadcast reply): the sidebar answers "anything new?" for a caught-up +-- reader with one comparison instead of walking past replies. Unindexed so +-- updates stay HOT. NULL for channels with no timeline message since this +-- migration, which no position can be behind: every position is floored at +-- a started_at set after it. +ALTER TABLE channels ADD COLUMN last_timeline_received_at TIMESTAMPTZ; + -- Arrival time and read scope of each mention, so the sidebar counts a -- reader's unread mentions in one channel timeline or thread from its -- position. root_id is the thread a reply belongs to, NULL when the mention diff --git a/schema/schema.sql b/schema/schema.sql index 1a94faf2da..4c4378b66e 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -101,6 +101,8 @@ CREATE TABLE channels ( participant_hash BYTEA, ttl_seconds INT, ttl_deadline TIMESTAMPTZ, + -- Arrival of the latest eligible timeline message; see 0056. + last_timeline_received_at TIMESTAMPTZ, PRIMARY KEY (community_id, id), CONSTRAINT chk_channels_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid) );