From 0e0b02520f60a5830c4bcabee3e2e1a5047aca83 Mon Sep 17 00:00:00 2001
From: hMurderbot
 <219c58afff8ac2c76e4e527a329aa61c2ebbb896786b1916d4f9cca5140ecd04@buzz.block.builderlab.xyz>
Date: Tue, 6 Oct 2026 19:54:55 -0700
Subject: [PATCH] relay: prototype unread counts by relay-assigned sequence
 numbers

Not for merge. Side-by-side comparison with #8138: ingest numbers each
eligible message under a per-channel counter row, unread is the counter
less the reader's position, posting marks the author read, deleted
messages count until read past. Includes a shared benchmark test.

Signed-off-by: hMurderbot <219c58afff8ac2c76e4e527a329aa61c2ebbb896786b1916d4f9cca5140ecd04@buzz.block.builderlab.xyz>
---
 crates/buzz-db/src/store/deletion.rs          |   2 +
 crates/buzz-db/src/store/event.rs             |  73 ++++--
 .../personal_read/bench_postgres_tests.rs     | 228 ++++++++++++++++++
 .../src/store/personal_read/context.rs        |  26 +-
 .../src/store/personal_read/membership.rs     | 136 +++++++++--
 crates/buzz-db/src/store/personal_read/mod.rs |   5 +-
 .../src/store/personal_read/projection.rs     | 103 ++++----
 .../buzz-db/src/store/personal_read/writes.rs | 184 ++++++++++----
 migrations/0056_personal_read_state.sql       |  79 +++++-
 schema/schema.sql                             |  78 +++++-
 10 files changed, 737 insertions(+), 177 deletions(-)
 create mode 100644 crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs

diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs
index 26857a218b..f1fda470de 100644
--- a/crates/buzz-db/src/store/deletion.rs
+++ b/crates/buzz-db/src/store/deletion.rs
@@ -93,6 +93,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "moderation_reports",
     "parameterized_event_watermarks",
     "personal_read_accounts",
+    "personal_read_counters",
     "personal_read_frontiers",
     "pubkey_allowlist",
     "push_leases",
@@ -113,6 +114,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
 /// Foreign-key-safe child-before-parent order for the PostgreSQL purge.
 pub const PURGE_SCOPED_TABLES: &[&str] = &[
     "personal_read_frontiers",
+    "personal_read_counters",
     "personal_read_accounts",
     "workflow_approvals",
     "scheduled_workflow_fires",
diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs
index a6d7cb8516..7acbe3fbdf 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -477,10 +477,14 @@ async fn insert_event_on(
     let received_at = Utc::now();
     let d_tag = extract_d_tag(event);
     let not_before = extract_not_before(event);
+    // No thread metadata: on the timeline.
+    let numbers =
+        crate::personal_read::reserve(&mut *connection, community_id, event, channel_id, true)
+            .await?;
     let result = sqlx::query(
         r#"
-        INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+        INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, channel_seq, timeline_seq)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
         ON CONFLICT DO NOTHING
         "#,
     )
@@ -496,10 +500,16 @@ async fn insert_event_on(
     .bind(channel_id)
     .bind(d_tag.as_deref())
     .bind(not_before)
-    .execute(connection)
+    .bind(numbers.map(|n| n.channel))
+    .bind(numbers.map(|n| n.timeline))
+    .execute(&mut *connection)
     .await?;
 
     let was_inserted = result.rows_affected() > 0;
+    if let (true, Some(n), Some(channel)) = (was_inserted, numbers, channel_id) {
+        crate::personal_read::commit(connection, community_id, event, channel, n, received_at)
+            .await?;
+    }
 
     Ok((
         StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
@@ -1580,10 +1590,17 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
     let d_tag = extract_d_tag(event);
     let not_before = extract_not_before(event);
 
+    let on_timeline = thread_meta
+        .as_ref()
+        .is_none_or(|m| m.parent_event_id.is_none() || (m.depth == 1 && m.broadcast));
+    let numbers =
+        crate::personal_read::reserve(&mut **tx, community_id, event, channel_id, on_timeline)
+            .await?;
+
     let result = sqlx::query(
         r#"
-        INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+        INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, channel_seq, timeline_seq)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
         ON CONFLICT DO NOTHING
         "#,
     )
@@ -1599,12 +1616,18 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
     .bind(channel_id)
     .bind(d_tag.as_deref())
     .bind(not_before)
+    .bind(numbers.map(|n| n.channel))
+    .bind(numbers.map(|n| n.timeline))
     .execute(&mut **tx)
     .await?;
 
     let was_inserted = result.rows_affected() > 0;
 
     if was_inserted {
+        if let (Some(n), Some(channel)) = (numbers, channel_id) {
+            crate::personal_read::commit(&mut **tx, community_id, event, channel, n, received_at)
+                .await?;
+        }
         if let Some(ref meta) = thread_meta {
             let broadcast_val: bool = meta.broadcast;
 
@@ -1696,29 +1719,45 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
                     .execute(&mut **tx)
                     .await?;
 
+                    // An eligible reply takes the thread's next number unless it
+                    // counts on the timeline instead (broadcast depth-1).
+                    let counted = numbers.filter(|_| !(meta.depth == 1 && meta.broadcast));
+                    let mut thread_seq = None;
                     if let Some(root_id) = meta.root_event_id {
-                        sqlx::query(
+                        thread_seq = sqlx::query_scalar::<_, i64>(
                             r#"
                             UPDATE thread_metadata
-                            SET descendant_count = descendant_count + 1
+                            SET descendant_count = descendant_count + 1,
+                                reply_seq = reply_seq + $3::int,
+                                reply_channel_seq = CASE WHEN $3 = 1 THEN $4 ELSE reply_channel_seq END,
+                                reply_last_id = CASE WHEN $3 = 1 THEN $5 ELSE reply_last_id END
                             WHERE community_id = $1 AND event_id = $2
+                            RETURNING reply_seq
                             "#,
                         )
                         .bind(community_id.as_uuid())
                         .bind(root_id)
-                        .execute(&mut **tx)
+                        .bind(i32::from(counted.is_some()))
+                        .bind(counted.map(|n| n.channel))
+                        .bind(meta.event_id)
+                        .fetch_optional(&mut **tx)
                         .await?;
                     }
                     let root = meta.root_event_id.unwrap_or(pid);
-                    crate::personal_read::record_reply(
-                        tx,
-                        community_id,
-                        event,
-                        meta.channel_id,
-                        root,
-                        received_at,
-                    )
-                    .await?;
+                    if let (Some(n), Some(seq)) = (numbers, thread_seq) {
+                        crate::personal_read::record_reply(
+                            tx,
+                            community_id,
+                            event,
+                            meta.channel_id,
+                            root,
+                            n,
+                            seq,
+                            counted.is_some(),
+                            received_at,
+                        )
+                        .await?;
+                    }
                 }
             }
         }
diff --git a/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs
new file mode 100644
index 0000000000..816fb62d3c
--- /dev/null
+++ b/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs
@@ -0,0 +1,228 @@
+//! Side-by-side benchmark (not for merge). Identical file in both branches.
+//! BENCH_THREADS (default 5000) followed threads, 3 replies each.
+use super::postgres_tests::{apply, channel_read, mention};
+use super::*;
+use crate::{
+    channel::{ChannelType, ChannelVisibility},
+    Db,
+};
+use buzz_core::CommunityId;
+use nostr::{EventBuilder, Keys, Kind, Tag};
+use sqlx::postgres::PgPoolOptions;
+use std::time::{Duration, Instant};
+use uuid::Uuid;
+
+fn ts() -> u64 {
+    nostr::Timestamp::now().as_secs()
+}
+
+fn ev(author: &Keys, at: u64, tags: Vec<Tag>) -> nostr::Event {
+    EventBuilder::new(Kind::Custom(9), format!("bench {at} {}", Uuid::new_v4()))
+        .tags(tags)
+        .custom_created_at(nostr::Timestamp::from(at))
+        .sign_with_keys(author)
+        .unwrap()
+}
+
+async fn top(db: &Db, c: CommunityId, ch: Uuid, e: &nostr::Event) -> Duration {
+    let t = Instant::now();
+    db.insert_event(c, e, Some(ch)).await.unwrap();
+    t.elapsed()
+}
+
+async fn rep(db: &Db, c: CommunityId, ch: Uuid, root: &nostr::Event, e: &nostr::Event) -> Duration {
+    let stamp = |e: &nostr::Event| {
+        chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap()
+    };
+    let t = Instant::now();
+    db.insert_event_with_thread_metadata(
+        c,
+        e,
+        Some(ch),
+        Some(crate::event::ThreadMetadataParams {
+            event_id: e.id.as_bytes(),
+            event_created_at: stamp(e),
+            channel_id: ch,
+            parent_event_id: Some(root.id.as_bytes()),
+            parent_event_created_at: Some(stamp(root)),
+            root_event_id: Some(root.id.as_bytes()),
+            root_event_created_at: Some(stamp(root)),
+            depth: 1,
+            broadcast: false,
+        }),
+    )
+    .await
+    .unwrap();
+    t.elapsed()
+}
+
+fn stats(label: &str, mut v: Vec<Duration>) {
+    v.sort();
+    let p = |q: f64| v[((v.len() as f64 - 1.0) * q) as usize].as_secs_f64() * 1000.0;
+    println!(
+        "BENCH {label}: n={} p50={:.3}ms p95={:.3}ms p99={:.3}ms max={:.3}ms",
+        v.len(),
+        p(0.5),
+        p(0.95),
+        p(0.99),
+        p(1.0)
+    );
+}
+
+async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) {
+    let mut v = Vec::new();
+    let mut last = None;
+    for _ in 0..30 {
+        let t = Instant::now();
+        let page = db
+            .personal_read_sidebar(c, &actor.public_key(), 20, None)
+            .await
+            .unwrap();
+        v.push(t.elapsed());
+        last = Some(page);
+    }
+    let page = last.unwrap();
+    let busy = &page.channels.iter().max_by_key(|r| r.unread).unwrap();
+    println!(
+        "BENCH {label}: busiest unread={} attention={} threads={} channels={}",
+        busy.unread,
+        busy.attention,
+        busy.unread_thread_count,
+        page.channels.len()
+    );
+    stats(&format!("sidebar {label}"), v);
+}
+
+#[tokio::test]
+#[ignore = "benchmark"]
+async fn bench_unread() {
+    let threads: usize = std::env::var("BENCH_THREADS")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(5000);
+    let pool = PgPoolOptions::new()
+        .max_connections(40)
+        .connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    let db = Db::from_pool(pool.clone());
+    let c = db
+        .ensure_configured_community(&format!("bench-{}.local", Uuid::new_v4()))
+        .await
+        .unwrap()
+        .id;
+    let actor = Keys::generate();
+    let mut channels = Vec::new();
+    for i in 0..20 {
+        channels.push(
+            db.create_channel(
+                c,
+                &format!("bench {i}"),
+                ChannelType::Stream,
+                ChannelVisibility::Open,
+                None,
+                &actor.public_key().to_bytes(),
+                None,
+            )
+            .await
+            .unwrap()
+            .id,
+        );
+    }
+    let busy = channels[0];
+    let others: Vec<Keys> = (0..8).map(|_| Keys::generate()).collect();
+    let base = ts() - 3600;
+    // A first message in every channel, and the first intent starts the actor.
+    let mut firsts = Vec::new();
+    for ch in &channels {
+        let e = ev(&others[0], base, vec![]);
+        top(&db, c, *ch, &e).await;
+        firsts.push(e);
+    }
+    assert_eq!(
+        apply(&db, c, &actor, channel_read(busy, &firsts[0])).await,
+        IntentOutcome::Applied
+    );
+
+    // Phase 1: followed threads, serial ingest.
+    let (mut roots_t, mut reps_t) = (Vec::new(), Vec::new());
+    let mut roots = Vec::new();
+    let mut latest = firsts[0].clone();
+    for i in 0..threads {
+        let at = base + 1 + (i as u64 % 3000);
+        let root = ev(&others[i % 8], at, vec![]);
+        roots_t.push(top(&db, c, busy, &root).await);
+        let mine = ev(&actor, at, vec![]);
+        reps_t.push(rep(&db, c, busy, &root, &mine).await);
+        for j in 1..3 {
+            let r = ev(&others[(i + j) % 8], at, vec![]);
+            reps_t.push(rep(&db, c, busy, &root, &r).await);
+            latest = r;
+        }
+        roots.push(root);
+    }
+    stats("ingest top-level serial", roots_t);
+    stats("ingest reply serial", reps_t);
+    assert_eq!(
+        apply(&db, c, &actor, channel_read(busy, &latest)).await,
+        IntentOutcome::Applied
+    );
+    for (ch, e) in channels.iter().zip(&firsts).skip(1) {
+        apply(&db, c, &actor, channel_read(*ch, e)).await;
+    }
+    sqlx::query("ANALYZE").execute(&pool).await.unwrap();
+    sidebar_bench(&db, c, &actor, &format!("caught-up {threads} threads")).await;
+
+    // Phase 2: a 20k timeline backlog with 12 mentions.
+    let now = ts();
+    let mut t2 = Vec::new();
+    for i in 0..20_000u64 {
+        let tags = if i % 1667 == 0 {
+            mention(&actor)
+        } else {
+            vec![]
+        };
+        let e = ev(&others[(i % 8) as usize], now - 600 + i / 40, tags);
+        t2.push(top(&db, c, busy, &e).await);
+    }
+    stats("ingest top-level serial (busy)", t2);
+    sqlx::query("ANALYZE").execute(&pool).await.unwrap();
+    sidebar_bench(&db, c, &actor, "20k timeline backlog").await;
+
+    // Phase 3: 500 followed threads get a new reply.
+    for root in roots.iter().take(500) {
+        let r = ev(&others[3], ts(), vec![]);
+        rep(&db, c, busy, root, &r).await;
+    }
+    sidebar_bench(&db, c, &actor, "20k backlog + 500 unread threads").await;
+
+    // Concurrency: 16 writers into one channel, then into 16 channels.
+    for (label, spread) in [("same channel", false), ("16 channels", true)] {
+        let started = Instant::now();
+        let mut handles = Vec::new();
+        for w in 0..16usize {
+            let db = db.clone();
+            let author = others[w % 8].clone();
+            let ch = if spread { channels[w + 1] } else { busy };
+            handles.push(tokio::spawn(async move {
+                let mut v = Vec::new();
+                for i in 0..300u64 {
+                    let e = ev(&author, ts() - (i % 30), vec![]);
+                    v.push(top(&db, c, ch, &e).await);
+                }
+                v
+            }));
+        }
+        let mut all = Vec::new();
+        for h in handles {
+            all.extend(h.await.unwrap());
+        }
+        let n = all.len();
+        let secs = started.elapsed().as_secs_f64();
+        println!(
+            "BENCH concurrent {label}: {n} inserts in {secs:.2}s = {:.0}/s",
+            n as f64 / secs
+        );
+        stats(&format!("ingest concurrent {label}"), all);
+    }
+}
diff --git a/crates/buzz-db/src/store/personal_read/context.rs b/crates/buzz-db/src/store/personal_read/context.rs
index 97fe7ebf0f..9d4825eeca 100644
--- a/crates/buzz-db/src/store/personal_read/context.rs
+++ b/crates/buzz-db/src/store/personal_read/context.rs
@@ -2,7 +2,6 @@
 use super::{classification, model::*, writes};
 use crate::{observability, Db, DbError, Result};
 use buzz_core::CommunityId;
-use chrono::{DateTime, Utc};
 use sqlx::{Acquire, Row};
 use std::collections::HashMap;
 use uuid::Uuid;
@@ -47,17 +46,15 @@ impl Db {
                 contexts.push(ContextState::Unavailable);
                 continue;
             };
-            // The positions the sidebar counts against, both floored at the
-            // account's start and absent before it: the channel's, never
-            // before joining and absent for a non-member; and the thread's,
-            // which exists only for the actor's threads and includes any
-            // whole-channel cut.
+            // The channel-wide cuts the sidebar subtracts against, absent
+            // before the account starts: the channel's for a member, and the
+            // thread's, which exists only for the actor's threads. A cut in
+            // channel numbering orders every message in either context.
             let positions = sqlx::query(
                 "SELECT CASE WHEN m.joined_at IS NOT NULL AND s.started IS NOT NULL
-                        THEN GREATEST(cf.through_timestamp, m.joined_at, s.started) END AS channel,
+                        THEN COALESCE(cf.through_channel_seq, 0) END AS channel,
                     CASE WHEN tf.root_id IS NOT NULL AND s.started IS NOT NULL
-                        THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp, s.started)
-                        END AS thread
+                        THEN tf.through_channel_seq END AS thread
                  FROM (SELECT (SELECT started_at FROM personal_read_accounts
                     WHERE community_id=$1 AND actor=$2) AS started) s
                  LEFT JOIN channel_members m ON m.community_id=$1 AND m.channel_id=$3
@@ -73,15 +70,15 @@ impl Db {
             .bind(&root)
             .fetch_one(&mut *tx)
             .await?;
-            let channel_position: Option<DateTime<Utc>> = positions.try_get("channel")?;
-            let thread_position: Option<DateTime<Utc>> = positions.try_get("thread")?;
+            let channel_position: Option<i64> = positions.try_get("channel")?;
+            let thread_position: Option<i64> = positions.try_get("thread")?;
             let ids: Vec<Vec<u8>> = query
                 .message_ids
                 .iter()
                 .filter_map(|id| writes::event_id(id))
                 .collect();
             let rows = sqlx::query(
-                "SELECT encode(e.id,'hex') AS id,e.kind,e.received_at,
+                "SELECT encode(e.id,'hex') AS id,e.kind,e.channel_seq,
                     e.deleted_at IS NOT NULL AS deleted,e.pubkey=$3 AS own,e.tags,
                     tm.root_event_id,tm.depth,tm.broadcast,c.channel_type::text AS channel_type
                  FROM events e JOIN channels c ON c.community_id=e.community_id AND c.id=e.channel_id
@@ -133,7 +130,8 @@ impl Db {
                             } else {
                                 thread_position
                             };
-                            let received: DateTime<Utc> = row.try_get("received_at")?;
+                            // Unnumbered messages predate numbering: read.
+                            let seq: Option<i64> = row.try_get("channel_seq")?;
                             let tags: Vec<Vec<String>> =
                                 serde_json::from_value(row.try_get("tags")?).unwrap_or_default();
                             let reason = classification::reason(
@@ -143,7 +141,7 @@ impl Db {
                             );
                             match position {
                                 None => MessageReadState::NotCounted,
-                                Some(p) if received <= p => MessageReadState::Read,
+                                Some(p) if seq.is_none_or(|n| n <= p) => MessageReadState::Read,
                                 Some(_) if on_timeline => MessageReadState::Unread { reason },
                                 Some(_) => MessageReadState::Unread {
                                     reason: match reason {
diff --git a/crates/buzz-db/src/store/personal_read/membership.rs b/crates/buzz-db/src/store/personal_read/membership.rs
index 8b2db9fb54..488a4b75b5 100644
--- a/crates/buzz-db/src/store/personal_read/membership.rs
+++ b/crates/buzz-db/src/store/personal_read/membership.rs
@@ -1,58 +1,157 @@
-//! Thread membership. A thread frontier row exists exactly for the actor's
-//! threads, and only ingest creates one: read intents move existing rows.
+//! Ingest's share of read state: numbering each eligible message under its
+//! channel's counter, posting marking the author's position, and thread
+//! membership. A thread frontier row exists exactly for the actor's threads,
+//! and only ingest creates one: read intents move existing rows.
 use super::model::ELIGIBLE_KINDS;
 use crate::Result;
 use buzz_core::CommunityId;
 use chrono::{DateTime, Utc};
 use nostr::Event;
-use sqlx::{Postgres, Transaction};
+use sqlx::{PgConnection, Postgres, Row, Transaction};
 use uuid::Uuid;
 
+/// The numbers a message takes if it is stored.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct Numbers {
+    /// The channel's timeline count once stored: its own number on the timeline.
+    pub timeline: i64,
+    /// Its number among every eligible message in the channel.
+    pub channel: i64,
+    /// Shown on the timeline: top-level, or a broadcast depth-1 reply.
+    pub on_timeline: bool,
+}
+
+/// Lock the channel's counter until commit and return the numbers an eligible
+/// message would take. The lock is what makes numbers follow commit order.
+pub(crate) async fn reserve(
+    conn: &mut PgConnection,
+    community: CommunityId,
+    event: &Event,
+    channel: Option<Uuid>,
+    on_timeline: bool,
+) -> Result<Option<Numbers>> {
+    let Some(channel) = channel else {
+        return Ok(None);
+    };
+    if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
+        return Ok(None);
+    }
+    let row = sqlx::query(
+        "INSERT INTO personal_read_counters (community_id, channel_id) VALUES ($1, $2)
+         ON CONFLICT (community_id, channel_id)
+            DO UPDATE SET channel_seq = personal_read_counters.channel_seq
+         RETURNING timeline_seq, channel_seq",
+    )
+    .bind(community.as_uuid())
+    .bind(channel)
+    .fetch_one(&mut *conn)
+    .await?;
+    Ok(Some(Numbers {
+        timeline: row.try_get::<i64, _>("timeline_seq")? + i64::from(on_timeline),
+        channel: row.try_get::<i64, _>("channel_seq")? + 1,
+        on_timeline,
+    }))
+}
+
+/// Advance the counter for a stored message. Posting reads the author's
+/// timeline through it once the author has started.
+pub(crate) async fn commit(
+    conn: &mut PgConnection,
+    community: CommunityId,
+    event: &Event,
+    channel: Uuid,
+    numbers: Numbers,
+    received_at: DateTime<Utc>,
+) -> Result<()> {
+    sqlx::query(
+        "UPDATE personal_read_counters SET timeline_seq=$3, channel_seq=$4, latest_id=$5
+         WHERE community_id=$1 AND channel_id=$2",
+    )
+    .bind(community.as_uuid())
+    .bind(channel)
+    .bind(numbers.timeline)
+    .bind(numbers.channel)
+    .bind(event.id.as_bytes().as_slice())
+    .execute(&mut *conn)
+    .await?;
+    if numbers.on_timeline {
+        sqlx::query(
+            "INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+                through_seq, through_channel_seq, through_timestamp)
+             SELECT $1, $2, $3, $4, $5, $6 FROM personal_read_accounts
+             WHERE community_id=$1 AND actor=$2 AND started_at IS NOT NULL
+             ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+                through_seq=GREATEST(personal_read_frontiers.through_seq, EXCLUDED.through_seq),
+                through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq,
+                    EXCLUDED.through_channel_seq),
+                through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
+                    EXCLUDED.through_timestamp)",
+        )
+        .bind(community.as_uuid())
+        .bind(event.pubkey.to_bytes().as_slice())
+        .bind(channel)
+        .bind(numbers.timeline)
+        .bind(numbers.channel)
+        .bind(received_at)
+        .execute(&mut *conn)
+        .await?;
+    }
+    Ok(())
+}
+
 /// Record who a newly stored reply makes a thread member: its author, the
 /// root's author, members it mentions, and in a DM (at most nine members)
-/// every member. The author's row starts at the reply, which it has read;
-/// everyone else's starts just before it, so the reply is unread. Existing
-/// rows are left alone.
+/// every member. `thread_seq` is the thread's count once the reply is stored.
+/// The author reads through the reply (posting marks read, also on an
+/// existing row); everyone else's new row starts just before it, so the reply
+/// is unread. Other existing rows are left alone.
 pub(crate) async fn record_reply(
     tx: &mut Transaction<'_, Postgres>,
     community: CommunityId,
     event: &Event,
     channel: Uuid,
     root: &[u8],
+    numbers: Numbers,
+    thread_seq: i64,
+    counts_in_thread: bool,
     received_at: DateTime<Utc>,
 ) -> Result<()> {
-    if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
-        return Ok(());
-    }
     let mentioned: Vec<Vec<u8>> = event
         .tags
         .public_keys()
         .map(|key| key.to_bytes().to_vec())
         .collect();
+    let before = thread_seq - i64::from(counts_in_thread);
     // Both inserts' fence and foreign-key checks run at the end of the
     // statement, after the account rows exist.
     sqlx::query(
         "WITH joined AS (
-            SELECT $4::bytea AS actor, $6::timestamptz AS through
+            SELECT $4::bytea AS actor, $7::bigint AS seq, $8::bigint AS cseq, $6::timestamptz AS at
             UNION ALL
-            SELECT pubkey, $6-interval '1 microsecond' FROM events
+            SELECT pubkey, $9, $8-1, $6-interval '1 microsecond' FROM events
             WHERE community_id=$1 AND channel_id=$2 AND id=$3
             UNION ALL
-            SELECT cm.pubkey, $6-interval '1 microsecond'
+            SELECT cm.pubkey, $9, $8-1, $6-interval '1 microsecond'
             FROM channel_members cm JOIN channels c
                 ON c.community_id=cm.community_id AND c.id=cm.channel_id
             WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
                 AND (cm.pubkey=ANY($5) OR c.channel_type='dm')
          ), members AS (
-            SELECT DISTINCT ON (actor) actor, through FROM joined ORDER BY actor, through DESC
+            SELECT DISTINCT ON (actor) actor, seq, cseq, at FROM joined ORDER BY actor, cseq DESC
          ), accounts AS (
             INSERT INTO personal_read_accounts (community_id, actor)
             SELECT $1, actor FROM members ON CONFLICT DO NOTHING
          )
-         INSERT INTO personal_read_frontiers
-            (community_id, actor, channel_id, root_id, through_timestamp)
-         SELECT $1, actor, $2, $3, through FROM members
-         ON CONFLICT DO NOTHING",
+         INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id,
+            through_seq, through_channel_seq, through_timestamp)
+         SELECT $1, actor, $2, $3, seq, cseq, at FROM members
+         ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+            through_seq=GREATEST(personal_read_frontiers.through_seq, EXCLUDED.through_seq),
+            through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq,
+                EXCLUDED.through_channel_seq),
+            through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
+                EXCLUDED.through_timestamp)
+         WHERE personal_read_frontiers.actor=$4",
     )
     .bind(community.as_uuid())
     .bind(channel)
@@ -60,6 +159,9 @@ pub(crate) async fn record_reply(
     .bind(event.pubkey.to_bytes().as_slice())
     .bind(&mentioned)
     .bind(received_at)
+    .bind(thread_seq)
+    .bind(numbers.channel)
+    .bind(before)
     .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 0c517f20ed..d8e7d749e3 100644
--- a/crates/buzz-db/src/store/personal_read/mod.rs
+++ b/crates/buzz-db/src/store/personal_read/mod.rs
@@ -11,7 +11,7 @@ mod model;
 mod projection;
 mod writes;
 
-pub(crate) use membership::record_reply;
+pub(crate) use membership::{commit, record_reply, reserve};
 pub use model::*;
 
 #[cfg(test)]
@@ -25,3 +25,6 @@ mod threads_postgres_tests;
 
 #[cfg(test)]
 mod arrival_postgres_tests;
+
+#[cfg(test)]
+mod bench_postgres_tests;
diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs
index ea47e9bd67..c2c87fc071 100644
--- a/crates/buzz-db/src/store/personal_read/projection.rs
+++ b/crates/buzz-db/src/store/personal_read/projection.rs
@@ -71,89 +71,78 @@ impl Db {
             .await?;
         writes::deadlines(&mut tx).await?;
         sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?;
-        // Every position is floored at the account's start (infinity before
-        // the actor's first intent, so nothing counts). A channel's is its
-        // frontier, never earlier than joining: a new member starts caught
-        // up. A thread row exists only for the actor's threads (see
-        // `membership`), and a whole-channel cut covers them all. Each count
-        // ranges over author time from the position less the arrival skew
-        // (served by idx_events_community_channel_created or
-        // idx_thread_metadata_window), filters on arrival, and stops at the
-        // cap: rows the filters reject are read but bound nothing. The latest message is the last arrival among
-        // the newest messages and everything counted, so marking through it
-        // reads every counted message whatever its author time.
+        // Unread is subtraction: the channel's counter less the actor's
+        // position (starting and joining write one, so a started member
+        // without a row has read nothing), and each of the actor's threads' reply count less its
+        // position. Nothing counts before the account starts. Attention needs
+        // the messages themselves, so it walks forward from the position
+        // through the numbered messages after it, only when something is
+        // unread, and stops at the cap. A whole-channel mark through the
+        // counter's latest message reads every counted message.
         let sql = format!(
             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,
-                    GREATEST(cf.through_timestamp, cm.joined_at, s.started) AS position,
-                    GREATEST(cf.threads_through_timestamp, s.started) AS threads_floor
+                    s.started IS NOT NULL AS started,
+                    COALESCE(k.timeline_seq, 0) AS timeline_seq,
+                    COALESCE(cf.through_seq, 0) AS through_seq,
+                    COALESCE(cf.through_channel_seq, 0) AS through_channel_seq,
+                    COALESCE(cf.through_timestamp, '-infinity') AS position,
+                    encode(k.latest_id, 'hex') AS latest_id
                 FROM channel_members cm JOIN channels c
                     ON c.community_id=cm.community_id AND c.id=cm.channel_id
-                CROSS JOIN (SELECT COALESCE((SELECT started_at FROM personal_read_accounts
-                    WHERE community_id=$1 AND actor=$2), 'infinity') AS started) s
+                CROSS JOIN (SELECT (SELECT started_at FROM personal_read_accounts
+                    WHERE community_id=$1 AND actor=$2) AS started) s
+                LEFT JOIN personal_read_counters k ON k.community_id=cm.community_id
+                    AND k.channel_id=c.id
                 LEFT JOIN personal_read_frontiers cf ON cf.community_id=cm.community_id
                     AND cf.actor=cm.pubkey AND cf.channel_id=c.id AND cf.root_id=''::bytea
                 WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL
                     AND c.deleted_at IS NULL AND ($3::uuid IS NULL OR c.id>$3)
                     AND ($4::uuid[] IS NULL OR c.id=ANY($4))
                 ORDER BY c.id LIMIT $5
+             ), counted AS (
+                SELECT r.*, CASE WHEN r.started
+                    THEN GREATEST(r.timeline_seq - r.through_seq, 0) ELSE 0 END AS unread
+                FROM roster r
              )
              SELECT r.id, r.name, r.channel_type, r.archived, r.hidden,
-                (SELECT v.id FROM (VALUES (latest.arrival, latest.id),
-                        (timeline.arrival, timeline.id), (threads.arrival, threads.id)) v(arrival, id)
-                    WHERE v.id IS NOT NULL ORDER BY v.arrival DESC, v.id LIMIT 1) AS latest_message_id,
-                latest.at AS latest_message_at,
-                COALESCE(timeline.unread, 0) AS unread, COALESCE(timeline.attention, 0) AS attention,
+                r.latest_id AS latest_message_id, latest.at AS latest_message_at,
+                LEAST(r.unread, $9)::int AS unread,
+                CASE WHEN r.channel_type='dm' THEN LEAST(r.unread, $9)::int
+                    ELSE COALESCE(attention.n, 0) END AS attention,
                 COALESCE(threads.items, '[]') AS threads
-             FROM roster r
+             FROM counted r
              LEFT JOIN LATERAL (
-                SELECT (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
-                    max(received_at) AS arrival, max(extract(epoch FROM created_at)::bigint) AS at
-                FROM (SELECT id,created_at,received_at FROM events
+                SELECT max(extract(epoch FROM created_at)::bigint) AS at
+                FROM (SELECT created_at FROM events
                     WHERE community_id=$1 AND channel_id=r.id AND kind=ANY($6) AND deleted_at IS NULL
                     ORDER BY created_at DESC LIMIT $7) newest
              ) latest ON true
              LEFT JOIN LATERAL (
-                SELECT count(*)::int AS unread, count(*) FILTER (WHERE directed)::int AS attention,
-                    (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
-                    max(received_at) AS arrival
-                FROM (
-                    SELECT e.id, e.received_at, r.channel_type='dm' OR {DIRECTED_TAGS} AS directed
-                    FROM events e
+                SELECT count(*)::int AS n FROM (
+                    SELECT 1 FROM events e
                     LEFT JOIN thread_metadata tm ON tm.community_id=$1
                         AND tm.event_created_at=e.created_at AND tm.event_id=e.id
-                    WHERE e.community_id=$1 AND e.channel_id=r.id
-                        AND e.created_at >= r.position-$8 AND e.received_at > r.position
-                        AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL AND {ON_TIMELINE}
+                    WHERE r.unread > 0 AND r.channel_type<>'dm'
+                        AND e.community_id=$1 AND e.channel_id=r.id
+                        AND e.created_at >= r.position-$8 AND e.channel_seq > r.through_channel_seq
+                        AND e.kind=ANY($6) AND e.pubkey<>$2 AND {ON_TIMELINE} AND {DIRECTED_TAGS}
                     ORDER BY e.created_at LIMIT $9
-                ) counted
-             ) timeline ON true
+                ) directed
+             ) attention ON true
              LEFT JOIN LATERAL (
                 SELECT jsonb_agg(jsonb_build_object('root_id',encode(tf.root_id,'hex'),
-                        'unread',n.unread,'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items,
-                    (array_agg(n.id ORDER BY n.arrival DESC,n.id))[1] AS id, max(n.arrival) AS arrival
+                        'unread',LEAST(root.reply_seq - tf.through_seq, $9),
+                        'latest_reply_id',encode(root.reply_last_id,'hex'),
+                        'latest_reply_at',extract(epoch FROM last.event_created_at)::bigint)) AS items
                 FROM personal_read_frontiers tf
-                CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, r.threads_floor) AS position) p
-                CROSS JOIN LATERAL (
-                    SELECT count(*)::int AS unread,
-                        (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
-                        max(extract(epoch FROM created_at)::bigint) AS at,
-                        max(received_at) AS arrival
-                    FROM (
-                        SELECT e.id, e.created_at, e.received_at
-                        FROM thread_metadata tm JOIN events e ON e.community_id=$1
-                            AND e.created_at=tm.event_created_at AND e.id=tm.event_id
-                        WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
-                            AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
-                            AND tm.event_created_at >= p.position-$8
-                            AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2
-                            AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
-                        ORDER BY tm.event_created_at LIMIT $9
-                    ) counted
-                ) n
-                WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
-                    AND tf.root_id<>''::bytea AND n.unread > 0
+                JOIN thread_metadata root ON root.community_id=$1 AND root.event_id=tf.root_id
+                    AND root.depth=0
+                CROSS JOIN LATERAL (SELECT event_created_at FROM thread_metadata
+                    WHERE community_id=$1 AND event_id=root.reply_last_id LIMIT 1) last
+                WHERE r.started AND tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
+                    AND tf.root_id<>''::bytea AND root.reply_seq > tf.through_seq
              ) threads ON true
              ORDER BY r.id"#
         )
diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs
index a4940cb5fb..41913085fc 100644
--- a/crates/buzz-db/src/store/personal_read/writes.rs
+++ b/crates/buzz-db/src/store/personal_read/writes.rs
@@ -26,28 +26,60 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> {
 /// Start the account if this is the actor's first read intent, then serialize
 /// private frontier writes, never shared conversation rows. NO KEY UPDATE
 /// leaves ingest's foreign-key checks (which create thread rows) unblocked.
+/// Starting reads every joined channel and existing thread through its
+/// current count.
 pub(super) async fn lock_account(
     conn: &mut PgConnection,
     community: CommunityId,
     actor: &[u8],
 ) -> Result<()> {
-    sqlx::query(
+    let started = sqlx::query(
         "INSERT INTO personal_read_accounts (community_id,actor,started_at) VALUES ($1,$2,now())
         ON CONFLICT (community_id,actor) DO UPDATE SET started_at=now()
-        WHERE personal_read_accounts.started_at IS NULL",
+        WHERE personal_read_accounts.started_at IS NULL RETURNING actor",
     )
     .bind(community.as_uuid())
     .bind(actor)
-    .execute(&mut *conn)
-    .await?;
+    .fetch_optional(&mut *conn)
+    .await?
+    .is_some();
     sqlx::query(
         "SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2
          FOR NO KEY UPDATE",
     )
     .bind(community.as_uuid())
     .bind(actor)
-    .fetch_one(conn)
+    .fetch_one(&mut *conn)
     .await?;
+    if started {
+        sqlx::query(
+            "INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+                through_seq, through_channel_seq, through_timestamp)
+             SELECT $1, $2, cm.channel_id, COALESCE(k.timeline_seq,0), COALESCE(k.channel_seq,0), now()
+             FROM channel_members cm LEFT JOIN personal_read_counters k
+                ON k.community_id=cm.community_id AND k.channel_id=cm.channel_id
+             WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL
+             ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+                through_seq=EXCLUDED.through_seq, through_channel_seq=EXCLUDED.through_channel_seq,
+                through_timestamp=EXCLUDED.through_timestamp",
+        )
+        .bind(community.as_uuid())
+        .bind(actor)
+        .execute(&mut *conn)
+        .await?;
+        sqlx::query(
+            "UPDATE personal_read_frontiers tf SET through_seq=root.reply_seq,
+                through_channel_seq=COALESCE(root.reply_channel_seq, tf.through_channel_seq),
+                through_timestamp=now()
+             FROM thread_metadata root
+             WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.root_id<>''::bytea
+                AND root.community_id=$1 AND root.event_id=tf.root_id AND root.depth=0",
+        )
+        .bind(community.as_uuid())
+        .bind(actor)
+        .execute(&mut *conn)
+        .await?;
+    }
     Ok(())
 }
 
@@ -87,6 +119,9 @@ async fn access(
 struct Message {
     id: Vec<u8>,
     received_at: DateTime<Utc>,
+    /// Numbers, absent for messages stored before numbering.
+    timeline_seq: Option<i64>,
+    channel_seq: Option<i64>,
     /// Canonical thread root, when the message is a reply.
     thread: Option<Vec<u8>>,
     /// Shown on the channel timeline: not a reply, or a broadcast depth-1 reply.
@@ -103,7 +138,8 @@ async fn message(
     kinds: Option<&[i32]>,
 ) -> Result<Option<Message>> {
     let row = sqlx::query(
-        "SELECT e.id, e.received_at, tm.root_event_id, tm.depth, tm.broadcast
+        "SELECT e.id, e.received_at, e.timeline_seq, e.channel_seq,
+             tm.root_event_id, tm.depth, tm.broadcast
          FROM events e LEFT JOIN thread_metadata tm ON tm.community_id=e.community_id
              AND tm.event_created_at=e.created_at AND tm.event_id=e.id AND tm.channel_id=e.channel_id
          WHERE e.community_id=$1 AND e.channel_id=$2 AND e.id=$3
@@ -119,6 +155,8 @@ async fn message(
         Ok(Message {
             on_timeline: thread.is_none() || broadcast_reply,
             received_at: row.try_get("received_at")?,
+            timeline_seq: row.try_get("timeline_seq")?,
+            channel_seq: row.try_get("channel_seq")?,
             thread,
             id,
         })
@@ -126,25 +164,6 @@ async fn message(
     .transpose()
 }
 
-/// An eligible-kind message's arrival time, deleted or not, without tags.
-async fn anchor_received_at(
-    conn: &mut PgConnection,
-    community: CommunityId,
-    channel: Uuid,
-    id: &[u8],
-) -> Result<Option<DateTime<Utc>>> {
-    Ok(sqlx::query_scalar(
-        "SELECT received_at FROM events
-         WHERE community_id=$1 AND channel_id=$2 AND id=$3 AND kind=ANY($4) LIMIT 1",
-    )
-    .bind(community.as_uuid())
-    .bind(channel)
-    .bind(id)
-    .bind(ELIGIBLE_KINDS.as_slice())
-    .fetch_optional(conn)
-    .await?)
-}
-
 pub(super) async fn valid_target(
     conn: &mut PgConnection,
     community: CommunityId,
@@ -174,6 +193,17 @@ pub(super) async fn valid_target(
     Ok(Some(root))
 }
 
+/// A thread's count through channel cut `cut`: its counted replies numbered
+/// at or before it. Equal to the count ingest assigned, since both follow the
+/// channel's numbering. The root's latest reply answers the usual case.
+const THREAD_SEQ_AT: &str = "CASE WHEN root.reply_channel_seq IS NULL
+        OR root.reply_channel_seq <= $cut THEN root.reply_seq
+    ELSE (SELECT count(*) FROM thread_metadata tm JOIN events e ON e.community_id=tm.community_id
+        AND e.created_at=tm.event_created_at AND e.id=tm.event_id
+        WHERE tm.community_id=root.community_id AND tm.root_event_id=root.event_id
+            AND tm.event_id<>root.event_id AND NOT (tm.depth=1 AND tm.broadcast)
+            AND e.kind=ANY($kinds) AND e.channel_seq <= $cut) END";
+
 pub(super) async fn apply(
     conn: &mut PgConnection,
     community: CommunityId,
@@ -199,20 +229,29 @@ pub(super) async fn apply(
             else {
                 return Ok(IntentOutcome::Blocked);
             };
+            let (Some(timeline_seq), Some(channel_seq)) = (msg.timeline_seq, msg.channel_seq)
+            else {
+                // Stored before numbering: already read.
+                return Ok(IntentOutcome::Applied);
+            };
             if root.is_empty() {
                 if !msg.on_timeline {
                     return Ok(IntentOutcome::Blocked);
                 }
                 sqlx::query(
-                    "INSERT INTO personal_read_frontiers
-                     (community_id, actor, channel_id, root_id, through_timestamp)
-                     VALUES ($1,$2,$3,''::bytea,$4)
-                     ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
-                     SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $4)",
+                    "INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+                        through_seq, through_channel_seq, through_timestamp)
+                     VALUES ($1,$2,$3,$4,$5,$6)
+                     ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+                        through_seq=GREATEST(personal_read_frontiers.through_seq, $4),
+                        through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq, $5),
+                        through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $6)",
                 )
                 .bind(community.as_uuid())
                 .bind(actor)
                 .bind(target.channel_id)
+                .bind(timeline_seq)
+                .bind(channel_seq)
                 .bind(msg.received_at)
                 .execute(&mut *conn)
                 .await?;
@@ -222,18 +261,27 @@ pub(super) async fn apply(
                 }
                 // Reading a thread never joins it: only the actor's threads
                 // have a row to advance.
-                sqlx::query(
-                    "UPDATE personal_read_frontiers
-                     SET through_timestamp=GREATEST(through_timestamp, $5)
-                     WHERE community_id=$1 AND actor=$2 AND channel_id=$3 AND root_id=$4",
+                let sql = format!(
+                    "UPDATE personal_read_frontiers tf SET
+                        through_seq=GREATEST(tf.through_seq, {THREAD_SEQ_AT}),
+                        through_channel_seq=GREATEST(tf.through_channel_seq, $5),
+                        through_timestamp=GREATEST(tf.through_timestamp, $6)
+                     FROM thread_metadata root
+                     WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=$3 AND tf.root_id=$4
+                        AND root.community_id=$1 AND root.event_id=$4 AND root.depth=0"
                 )
-                .bind(community.as_uuid())
-                .bind(actor)
-                .bind(target.channel_id)
-                .bind(&root)
-                .bind(msg.received_at)
-                .execute(&mut *conn)
-                .await?;
+                .replace("$cut", "$5")
+                .replace("$kinds", "$7");
+                sqlx::query(sqlx::AssertSqlSafe(sql))
+                    .bind(community.as_uuid())
+                    .bind(actor)
+                    .bind(target.channel_id)
+                    .bind(&root)
+                    .bind(channel_seq)
+                    .bind(msg.received_at)
+                    .bind(ELIGIBLE_KINDS.as_slice())
+                    .execute(&mut *conn)
+                    .await?;
             }
         }
         ReadIntent::MarkChannelRead {
@@ -246,19 +294,55 @@ pub(super) async fn apply(
             if !access(conn, community, actor, *channel_id).await? {
                 return Ok(IntentOutcome::Blocked);
             }
-            // Only the anchor's arrival matters: ancestry cannot change which
+            // Only the anchor's numbers matter: ancestry cannot change which
             // messages a whole-channel cut covers.
-            let Some(through) = anchor_received_at(conn, community, *channel_id, &id).await? else {
+            let Some(msg) =
+                message(conn, community, *channel_id, &id, Some(&ELIGIBLE_KINDS)).await?
+            else {
                 return Ok(IntentOutcome::Blocked);
             };
+            let (Some(timeline_seq), Some(cut)) = (msg.timeline_seq, msg.channel_seq) else {
+                return Ok(IntentOutcome::Applied);
+            };
             sqlx::query(
-                "INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id,
-                    through_timestamp, threads_through_timestamp) VALUES ($1,$2,$3,''::bytea,$4,$4)
-                 ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
-                 SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $4),
-                    threads_through_timestamp=GREATEST(personal_read_frontiers.threads_through_timestamp, $4)",
-            ).bind(community.as_uuid()).bind(actor).bind(channel_id).bind(through)
-                .execute(&mut *conn).await?;
+                "INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+                    through_seq, through_channel_seq, through_timestamp)
+                 VALUES ($1,$2,$3,$4,$5,$6)
+                 ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+                    through_seq=GREATEST(personal_read_frontiers.through_seq, $4),
+                    through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq, $5),
+                    through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $6)",
+            )
+            .bind(community.as_uuid())
+            .bind(actor)
+            .bind(channel_id)
+            .bind(timeline_seq)
+            .bind(cut)
+            .bind(msg.received_at)
+            .execute(&mut *conn)
+            .await?;
+            // The cut covers every one of the actor's threads in the channel.
+            let sql = format!(
+                "UPDATE personal_read_frontiers tf SET
+                    through_seq=GREATEST(tf.through_seq, {THREAD_SEQ_AT}),
+                    through_channel_seq=GREATEST(tf.through_channel_seq, $4),
+                    through_timestamp=GREATEST(tf.through_timestamp, $5)
+                 FROM thread_metadata root
+                 WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=$3
+                    AND tf.root_id<>''::bytea AND root.community_id=$1
+                    AND root.event_id=tf.root_id AND root.depth=0"
+            )
+            .replace("$cut", "$4")
+            .replace("$kinds", "$6");
+            sqlx::query(sqlx::AssertSqlSafe(sql))
+                .bind(community.as_uuid())
+                .bind(actor)
+                .bind(channel_id)
+                .bind(cut)
+                .bind(msg.received_at)
+                .bind(ELIGIBLE_KINDS.as_slice())
+                .execute(&mut *conn)
+                .await?;
         }
     }
     Ok(IntentOutcome::Applied)
diff --git a/migrations/0056_personal_read_state.sql b/migrations/0056_personal_read_state.sql
index aa06f16489..574c53841d 100644
--- a/migrations/0056_personal_read_state.sql
+++ b/migrations/0056_personal_read_state.sql
@@ -1,13 +1,23 @@
 -- Private accessory read progress. Never included in Nostr event queries.
--- A frontier is the relay arrival time (events.received_at) of the message a
--- context was read through, never signed event time, which the sender chooses.
--- An empty root_id covers only the channel timeline; a root-specific frontier
--- covers that thread, without inheritance.
--- threads_through_timestamp is the only cross-context cut: an explicit
--- whole-channel read that also covers every thread in that channel.
--- started_at is the actor's first read intent and the floor of every
--- position: until then nothing counts, and from then the actor starts caught
--- up. Ingest creates accounts for thread membership without starting them.
+--
+-- Unread is counted by subtraction over relay-assigned message numbers.
+-- Ingest numbers every eligible message in its channel under one locked
+-- counter row:
+--   events.channel_seq   every eligible message in the channel, in commit order;
+--   events.timeline_seq  the channel's timeline count once the message is
+--                        stored (its own number when it is on the timeline);
+--   thread_metadata.reply_seq on a root: the thread's eligible replies so far,
+--                        broadcast depth-1 replies excluded (they count on the
+--                        timeline), with reply_channel_seq and reply_last_id
+--                        the latest one's.
+-- A position is the numbers a context was read through. Unread is the latest
+-- number minus the position, so every numbered message counts for every
+-- reader: posting marks the author's position, and deleted messages count
+-- until read past.
+--
+-- started_at is the actor's first read intent: until then nothing counts.
+-- Starting writes a caught-up position for every joined channel and thread,
+-- and joining a channel afterwards writes one at its current count.
 CREATE TABLE personal_read_accounts (
     community_id UUID NOT NULL REFERENCES communities(id),
     actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
@@ -15,15 +25,32 @@ CREATE TABLE personal_read_accounts (
     PRIMARY KEY (community_id, actor)
 );
 
+CREATE TABLE personal_read_counters (
+    community_id UUID NOT NULL,
+    channel_id UUID NOT NULL,
+    timeline_seq BIGINT NOT NULL DEFAULT 0,
+    channel_seq BIGINT NOT NULL DEFAULT 0,
+    -- The message numbered channel_seq: marking the channel read through it
+    -- reads everything, so a badge has an anchor without a scan.
+    latest_id BYTEA,
+    PRIMARY KEY (community_id, channel_id),
+    FOREIGN KEY (community_id, channel_id)
+        REFERENCES channels (community_id, id) ON DELETE CASCADE
+);
+
+-- An empty root_id is the channel timeline. A thread row exists exactly for
+-- the actor's threads. through_seq is in the context's own numbering
+-- (timeline or thread); through_channel_seq is the channel-wide cut it was
+-- read through. through_timestamp is the anchor's arrival, used only to
+-- bound the mention scan.
 CREATE TABLE personal_read_frontiers (
     community_id UUID NOT NULL,
     actor BYTEA NOT NULL,
     channel_id UUID NOT NULL,
     root_id BYTEA NOT NULL DEFAULT ''::bytea CHECK (octet_length(root_id) IN (0, 32)),
+    through_seq BIGINT NOT NULL,
+    through_channel_seq BIGINT NOT NULL,
     through_timestamp TIMESTAMPTZ NOT NULL,
-    -- Whole-channel cut covering every thread; channel rows only.
-    threads_through_timestamp TIMESTAMPTZ
-        CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea),
     PRIMARY KEY (community_id, actor, channel_id, root_id),
     FOREIGN KEY (community_id, actor)
         REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE,
@@ -31,7 +58,35 @@ CREATE TABLE personal_read_frontiers (
         REFERENCES channels (community_id, id) ON DELETE CASCADE
 );
 
+ALTER TABLE events ADD COLUMN channel_seq BIGINT, ADD COLUMN timeline_seq BIGINT;
+ALTER TABLE thread_metadata ADD COLUMN reply_seq BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN reply_channel_seq BIGINT, ADD COLUMN reply_last_id BYTEA;
 
+-- Joining (or rejoining) a channel after starting begins caught up.
+CREATE FUNCTION personal_read_on_join() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF NEW.removed_at IS NULL AND (TG_OP = 'INSERT' OR OLD.removed_at IS NOT NULL) THEN
+        INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+            through_seq, through_channel_seq, through_timestamp)
+        SELECT NEW.community_id, NEW.pubkey, NEW.channel_id,
+            COALESCE(k.timeline_seq, 0), COALESCE(k.channel_seq, 0), now()
+        FROM personal_read_accounts a
+        LEFT JOIN personal_read_counters k
+            ON k.community_id = NEW.community_id AND k.channel_id = NEW.channel_id
+        WHERE a.community_id = NEW.community_id AND a.actor = NEW.pubkey
+            AND a.started_at IS NOT NULL
+        ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+            through_seq = EXCLUDED.through_seq,
+            through_channel_seq = EXCLUDED.through_channel_seq,
+            through_timestamp = EXCLUDED.through_timestamp;
+    END IF;
+    RETURN NULL;
+END $$;
+
+CREATE TRIGGER trg_channel_members_personal_read_join
+    AFTER INSERT OR UPDATE OF removed_at ON channel_members
+    FOR EACH ROW EXECUTE FUNCTION personal_read_on_join();
 
 SELECT attach_community_write_fence('personal_read_accounts');
+SELECT attach_community_write_fence('personal_read_counters');
 SELECT attach_community_write_fence('personal_read_frontiers');
diff --git a/schema/schema.sql b/schema/schema.sql
index a54f877671..cfcbafcfa8 100644
--- a/schema/schema.sql
+++ b/schema/schema.sql
@@ -194,12 +194,25 @@ CREATE UNIQUE INDEX idx_users_okta ON users (community_id, okta_user_id)
     WHERE okta_user_id IS NOT NULL;
 
 -- Private accessory read progress. Never included in Nostr event queries.
--- A frontier is the relay arrival time (events.received_at) of the message a
--- context was read through, never signed event time. An empty root_id denotes
--- a channel frontier.
--- started_at is the actor's first read intent and the floor of every
--- position: until then nothing counts, and from then the actor starts caught
--- up. Ingest creates accounts for thread membership without starting them.
+--
+-- Unread is counted by subtraction over relay-assigned message numbers.
+-- Ingest numbers every eligible message in its channel under one locked
+-- counter row:
+--   events.channel_seq   every eligible message in the channel, in commit order;
+--   events.timeline_seq  the channel's timeline count once the message is
+--                        stored (its own number when it is on the timeline);
+--   thread_metadata.reply_seq on a root: the thread's eligible replies so far,
+--                        broadcast depth-1 replies excluded (they count on the
+--                        timeline), with reply_channel_seq and reply_last_id
+--                        the latest one's.
+-- A position is the numbers a context was read through. Unread is the latest
+-- number minus the position, so every numbered message counts for every
+-- reader: posting marks the author's position, and deleted messages count
+-- until read past.
+--
+-- started_at is the actor's first read intent: until then nothing counts.
+-- Starting writes a caught-up position for every joined channel and thread,
+-- and joining a channel afterwards writes one at its current count.
 CREATE TABLE personal_read_accounts (
     community_id UUID NOT NULL REFERENCES communities(id),
     actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
@@ -207,15 +220,32 @@ CREATE TABLE personal_read_accounts (
     PRIMARY KEY (community_id, actor)
 );
 
+CREATE TABLE personal_read_counters (
+    community_id UUID NOT NULL,
+    channel_id UUID NOT NULL,
+    timeline_seq BIGINT NOT NULL DEFAULT 0,
+    channel_seq BIGINT NOT NULL DEFAULT 0,
+    -- The message numbered channel_seq: marking the channel read through it
+    -- reads everything, so a badge has an anchor without a scan.
+    latest_id BYTEA,
+    PRIMARY KEY (community_id, channel_id),
+    FOREIGN KEY (community_id, channel_id)
+        REFERENCES channels (community_id, id) ON DELETE CASCADE
+);
+
+-- An empty root_id is the channel timeline. A thread row exists exactly for
+-- the actor's threads. through_seq is in the context's own numbering
+-- (timeline or thread); through_channel_seq is the channel-wide cut it was
+-- read through. through_timestamp is the anchor's arrival, used only to
+-- bound the mention scan.
 CREATE TABLE personal_read_frontiers (
     community_id UUID NOT NULL,
     actor BYTEA NOT NULL,
     channel_id UUID NOT NULL,
     root_id BYTEA NOT NULL DEFAULT ''::bytea CHECK (octet_length(root_id) IN (0, 32)),
+    through_seq BIGINT NOT NULL,
+    through_channel_seq BIGINT NOT NULL,
     through_timestamp TIMESTAMPTZ NOT NULL,
-    -- Whole-channel cut covering every thread; channel rows only.
-    threads_through_timestamp TIMESTAMPTZ
-        CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea),
     PRIMARY KEY (community_id, actor, channel_id, root_id),
     FOREIGN KEY (community_id, actor)
         REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE,
@@ -223,6 +253,31 @@ CREATE TABLE personal_read_frontiers (
         REFERENCES channels (community_id, id) ON DELETE CASCADE
 );
 
+-- Joining (or rejoining) a channel after starting begins caught up.
+CREATE FUNCTION personal_read_on_join() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF NEW.removed_at IS NULL AND (TG_OP = 'INSERT' OR OLD.removed_at IS NOT NULL) THEN
+        INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
+            through_seq, through_channel_seq, through_timestamp)
+        SELECT NEW.community_id, NEW.pubkey, NEW.channel_id,
+            COALESCE(k.timeline_seq, 0), COALESCE(k.channel_seq, 0), now()
+        FROM personal_read_accounts a
+        LEFT JOIN personal_read_counters k
+            ON k.community_id = NEW.community_id AND k.channel_id = NEW.channel_id
+        WHERE a.community_id = NEW.community_id AND a.actor = NEW.pubkey
+            AND a.started_at IS NOT NULL
+        ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
+            through_seq = EXCLUDED.through_seq,
+            through_channel_seq = EXCLUDED.through_channel_seq,
+            through_timestamp = EXCLUDED.through_timestamp;
+    END IF;
+    RETURN NULL;
+END $$;
+
+CREATE TRIGGER trg_channel_members_personal_read_join
+    AFTER INSERT OR UPDATE OF removed_at ON channel_members
+    FOR EACH ROW EXECUTE FUNCTION personal_read_on_join();
+
 -- ── Events (partitioned by month on created_at) ──────────────────────────────
 -- Conformance: "Channel-less global events and DMs". `community_id` leads the
 -- PK and every hot-path index. Partition stays BY RANGE (created_at) — the
@@ -563,6 +618,10 @@ CREATE TABLE thread_metadata (
     FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id)
 );
 
+ALTER TABLE events ADD COLUMN channel_seq BIGINT, ADD COLUMN timeline_seq BIGINT;
+ALTER TABLE thread_metadata ADD COLUMN reply_seq BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN reply_channel_seq BIGINT, ADD COLUMN reply_last_id BYTEA;
+
 CREATE INDEX idx_thread_metadata_parent ON thread_metadata (community_id, parent_event_id);
 CREATE INDEX idx_thread_metadata_root ON thread_metadata (community_id, root_event_id);
 CREATE INDEX idx_thread_metadata_window
@@ -1820,6 +1879,7 @@ SELECT attach_community_write_fence('moderation_actions');
 SELECT attach_community_write_fence('moderation_reports');
 SELECT attach_community_write_fence('parameterized_event_watermarks');
 SELECT attach_community_write_fence('personal_read_accounts');
+SELECT attach_community_write_fence('personal_read_counters');
 SELECT attach_community_write_fence('personal_read_frontiers');
 SELECT attach_community_write_fence('pubkey_allowlist');
 SELECT attach_community_write_fence('push_leases');