mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(relay): record when a read-state account starts (#8157)
## Goal Add `started_at` to read-state accounts from the day the table is created. This prevents users seeing ALL their channels as unread when they first install a client update that starts using these APIs --------- Signed-off-by: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz> Signed-off-by: ART <ea39c1c8acf57180161ef4ba4ed68abf8589f5291956b909a5aa991fc66664ca@buzz.block.builderlab.xyz> Signed-off-by: hMurderbot <219c58afff8ac2c76e4e527a329aa61c2ebbb896786b1916d4f9cca5140ecd04@buzz.block.builderlab.xyz> Co-authored-by: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz> Co-authored-by: hMurderbot <219c58afff8ac2c76e4e527a329aa61c2ebbb896786b1916d4f9cca5140ecd04@buzz.block.builderlab.xyz> Co-authored-by: ART <ea39c1c8acf57180161ef4ba4ed68abf8589f5291956b909a5aa991fc66664ca@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -705,12 +705,17 @@ mod postgres_tests {
|
|||||||
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
|
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
|
||||||
migrations.sort_by_key(|migration| migration.version);
|
migrations.sort_by_key(|migration| migration.version);
|
||||||
|
|
||||||
assert_eq!(migrations.len(), 56);
|
assert_eq!(migrations.len(), 57);
|
||||||
assert_eq!(migrations[55].version, 56);
|
assert_eq!(migrations[55].version, 56);
|
||||||
assert!(migrations[55]
|
assert!(migrations[55]
|
||||||
.sql
|
.sql
|
||||||
.as_str()
|
.as_str()
|
||||||
.contains("CREATE TABLE personal_read_accounts"));
|
.contains("CREATE TABLE personal_read_accounts"));
|
||||||
|
assert_eq!(migrations[56].version, 57);
|
||||||
|
assert!(migrations[56]
|
||||||
|
.sql
|
||||||
|
.as_str()
|
||||||
|
.contains("ALTER TABLE personal_read_accounts ADD COLUMN started_at"));
|
||||||
assert_eq!(migrations[48].version, 49);
|
assert_eq!(migrations[48].version, 49);
|
||||||
assert_eq!(migrations[49].version, 50);
|
assert_eq!(migrations[49].version, 50);
|
||||||
assert_eq!(migrations[50].version, 51);
|
assert_eq!(migrations[50].version, 51);
|
||||||
@@ -2210,9 +2215,30 @@ mod postgres_tests {
|
|||||||
.sql
|
.sql
|
||||||
.as_ref(),
|
.as_ref(),
|
||||||
);
|
);
|
||||||
|
// 0057 adds personal_read_accounts.started_at after 0056's CREATE TABLE.
|
||||||
|
let started_at_column = "started_at timestamptz, ";
|
||||||
|
assert!(MIGRATOR
|
||||||
|
.iter()
|
||||||
|
.find(|m| m.version == 57)
|
||||||
|
.expect("personal read start migration")
|
||||||
|
.sql
|
||||||
|
.as_str()
|
||||||
|
.contains("ADD COLUMN started_at TIMESTAMPTZ"));
|
||||||
|
assert!(schema
|
||||||
|
.tables
|
||||||
|
.get("personal_read_accounts")
|
||||||
|
.expect("schema.sql personal read accounts")
|
||||||
|
.contains(started_at_column));
|
||||||
for (table, definition) in personal.tables {
|
for (table, definition) in personal.tables {
|
||||||
|
let in_schema = schema.tables.get(&table).map(|schema_definition| {
|
||||||
|
if table == "personal_read_accounts" {
|
||||||
|
schema_definition.replacen(started_at_column, "", 1)
|
||||||
|
} else {
|
||||||
|
schema_definition.clone()
|
||||||
|
}
|
||||||
|
});
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
schema.tables.get(&table),
|
in_schema.as_ref(),
|
||||||
Some(&definition),
|
Some(&definition),
|
||||||
"personal read table {table} differs"
|
"personal read table {table} differs"
|
||||||
);
|
);
|
||||||
|
|||||||
@@ -311,7 +311,16 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
|
|||||||
production_result.is_ok(),
|
production_result.is_ok(),
|
||||||
"production migrator must preserve ingestion progress: {production_result:?}"
|
"production migrator must preserve ingestion progress: {production_result:?}"
|
||||||
);
|
);
|
||||||
assert_eq!(version, 56);
|
let latest = sqlx::migrate::Migrator::new(
|
||||||
|
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../migrations"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.iter()
|
||||||
|
.map(|m| m.version)
|
||||||
|
.max()
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(version, latest);
|
||||||
assert_eq!(final_oid, oid, "prebuild must not be replaced");
|
assert_eq!(final_oid, oid, "prebuild must not be replaced");
|
||||||
assert_eq!(count, 4, "all writer witnesses must persist");
|
assert_eq!(count, 4, "all writer witnesses must persist");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -39,7 +39,9 @@ impl Db {
|
|||||||
.await?;
|
.await?;
|
||||||
writes::deadlines(&mut tx).await?;
|
writes::deadlines(&mut tx).await?;
|
||||||
let actor_bytes = actor.to_bytes();
|
let actor_bytes = actor.to_bytes();
|
||||||
let account = read_account(&mut tx, retention_seconds).await?;
|
let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?;
|
||||||
|
// Every frontier is floored at the account's start.
|
||||||
|
let floor = account.floor();
|
||||||
let mut contexts = Vec::with_capacity(queries.len());
|
let mut contexts = Vec::with_capacity(queries.len());
|
||||||
// Replies whose state turns on conversation membership, by position.
|
// Replies whose state turns on conversation membership, by position.
|
||||||
let mut pending = Vec::new();
|
let mut pending = Vec::new();
|
||||||
@@ -128,7 +130,7 @@ impl Db {
|
|||||||
account.cutoff_ms,
|
account.cutoff_ms,
|
||||||
) {
|
) {
|
||||||
MessageReadState::NotCounted
|
MessageReadState::NotCounted
|
||||||
} else if prefix.is_some_and(|p| received <= p) {
|
} else if received <= prefix.map_or(floor, |p| p.max(floor)) {
|
||||||
MessageReadState::Read
|
MessageReadState::Read
|
||||||
} else {
|
} else {
|
||||||
let reason = classification::reason(
|
let reason = classification::reason(
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
use chrono::{DateTime, Utc};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@@ -56,6 +57,18 @@ pub struct ReadAccount {
|
|||||||
pub retention_seconds: u32,
|
pub retention_seconds: u32,
|
||||||
/// Read-time author-time cutoff (Unix milliseconds), not a discard boundary.
|
/// Read-time author-time cutoff (Unix milliseconds), not a discard boundary.
|
||||||
pub cutoff_ms: i64,
|
pub cutoff_ms: i64,
|
||||||
|
/// Arrival floor of every frontier: the account's first applied read
|
||||||
|
/// intent. `None` until then, and nothing counts as unread. Not on the wire.
|
||||||
|
#[serde(skip)]
|
||||||
|
pub started_at: Option<DateTime<Utc>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ReadAccount {
|
||||||
|
/// The arrival every frontier is floored at: the start, or the end of time
|
||||||
|
/// for an account that has not started.
|
||||||
|
pub(super) fn floor(&self) -> DateTime<Utc> {
|
||||||
|
self.started_at.unwrap_or(DateTime::<Utc>::MAX_UTC)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Maximum channel summaries in one sidebar page.
|
/// Maximum channel summaries in one sidebar page.
|
||||||
|
|||||||
@@ -34,11 +34,13 @@ impl World {
|
|||||||
.await
|
.await
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.id;
|
.id;
|
||||||
|
let actor = Keys::generate();
|
||||||
|
super::postgres_tests::start_before_everything(&pool, community, &actor.public_key()).await;
|
||||||
Self {
|
Self {
|
||||||
db,
|
db,
|
||||||
pool,
|
pool,
|
||||||
community,
|
community,
|
||||||
actor: Keys::generate(),
|
actor,
|
||||||
peer: Keys::generate(),
|
peer: Keys::generate(),
|
||||||
now: Timestamp::now().as_secs(),
|
now: Timestamp::now().as_secs(),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,9 +38,28 @@ pub(super) async fn fixture() -> (Db, PgPool, CommunityId, Uuid, Keys, nostr::Ev
|
|||||||
db.insert_event(community, &event, Some(channel))
|
db.insert_event(community, &event, Some(channel))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
start_before_everything(&pool, community, &actor.public_key()).await;
|
||||||
(db, pool, community, channel, actor, event)
|
(db, pool, community, channel, actor, event)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Start an account before any fixture arrival, so unread semantics can be
|
||||||
|
/// tested without the start floor covering every message.
|
||||||
|
pub(super) async fn start_before_everything(
|
||||||
|
pool: &PgPool,
|
||||||
|
community: CommunityId,
|
||||||
|
actor: &nostr::PublicKey,
|
||||||
|
) {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO personal_read_accounts (community_id,actor,started_at)
|
||||||
|
VALUES ($1,$2,'1900-01-01T00:00:00Z')",
|
||||||
|
)
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(actor.to_bytes().as_slice())
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
/// Move every community message past the default horizon by author time, the
|
/// Move every community message past the default horizon by author time, the
|
||||||
/// only clock the unread window reads.
|
/// only clock the unread window reads.
|
||||||
async fn expire(pool: &PgPool, community: CommunityId) {
|
async fn expire(pool: &PgPool, community: CommunityId) {
|
||||||
@@ -253,6 +272,7 @@ async fn personal_read_sidebar_window_budget_counts_boundary_and_ineligible_tail
|
|||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn personal_read_sidebar_is_read_only_and_does_not_wait_for_account() {
|
async fn personal_read_sidebar_is_read_only_and_does_not_wait_for_account() {
|
||||||
let (db, pool, community, channel, actor, event) = fixture().await;
|
let (db, pool, community, channel, actor, event) = fixture().await;
|
||||||
|
unstart(&pool, community).await;
|
||||||
db.personal_read_sidebar(
|
db.personal_read_sidebar(
|
||||||
community,
|
community,
|
||||||
&actor.public_key(),
|
&actor.public_key(),
|
||||||
@@ -540,6 +560,7 @@ async fn personal_read_latest_old_message_is_not_an_empty_channel() {
|
|||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn personal_read_frontier_is_monotonic_and_rejects_malformed_anchors() {
|
async fn personal_read_frontier_is_monotonic_and_rejects_malformed_anchors() {
|
||||||
let (db, pool, community, channel, actor, event) = fixture().await;
|
let (db, pool, community, channel, actor, event) = fixture().await;
|
||||||
|
unstart(&pool, community).await;
|
||||||
let target = ReadTarget {
|
let target = ReadTarget {
|
||||||
channel_id: channel,
|
channel_id: channel,
|
||||||
root_id: None,
|
root_id: None,
|
||||||
@@ -626,6 +647,189 @@ async fn personal_read_frontier_is_monotonic_and_rejects_malformed_anchors() {
|
|||||||
assert!(sparse.is_none());
|
assert!(sparse.is_none());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Undo the fixture's start, for tests of what creates an account.
|
||||||
|
async fn unstart(pool: &PgPool, community: CommunityId) {
|
||||||
|
sqlx::query("DELETE FROM personal_read_accounts WHERE community_id=$1")
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn started_at(
|
||||||
|
pool: &PgPool,
|
||||||
|
community: CommunityId,
|
||||||
|
actor: &nostr::PublicKey,
|
||||||
|
) -> Option<chrono::DateTime<chrono::Utc>> {
|
||||||
|
sqlx::query_scalar(
|
||||||
|
"SELECT started_at FROM personal_read_accounts WHERE community_id=$1 AND actor=$2",
|
||||||
|
)
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(actor.to_bytes().as_slice())
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[ignore = "requires Postgres"]
|
||||||
|
async fn personal_read_first_applied_intent_starts_the_account_once() {
|
||||||
|
let (db, pool, community, channel, _, event) = fixture().await;
|
||||||
|
let intent = ReadIntent::MarkThrough {
|
||||||
|
target: ReadTarget {
|
||||||
|
channel_id: channel,
|
||||||
|
root_id: None,
|
||||||
|
},
|
||||||
|
message_id: event.id.to_hex(),
|
||||||
|
};
|
||||||
|
// The fixture actor is already started; the channel is open to anyone.
|
||||||
|
let actor = Keys::generate().public_key();
|
||||||
|
assert!(sqlx::query_scalar::<_, Vec<u8>>(
|
||||||
|
"SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2"
|
||||||
|
)
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(actor.to_bytes().as_slice())
|
||||||
|
.fetch_optional(&pool)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.is_none());
|
||||||
|
for _ in 0..2 {
|
||||||
|
assert_eq!(
|
||||||
|
db.apply_personal_read_intent(community, &actor, &intent)
|
||||||
|
.await
|
||||||
|
.unwrap(),
|
||||||
|
IntentOutcome::Applied
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let first = started_at(&pool, community, &actor).await;
|
||||||
|
assert!(
|
||||||
|
first.is_some(),
|
||||||
|
"the first applied intent starts the account"
|
||||||
|
);
|
||||||
|
|
||||||
|
// An account can exist before its actor starts; the first intent starts
|
||||||
|
// it, and later intents never move the start.
|
||||||
|
let pending = Keys::generate().public_key();
|
||||||
|
sqlx::query("INSERT INTO personal_read_accounts (community_id,actor) VALUES ($1,$2)")
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(pending.to_bytes().as_slice())
|
||||||
|
.execute(&pool)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(started_at(&pool, community, &pending).await, None);
|
||||||
|
db.apply_personal_read_intent(community, &pending, &intent)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let pending_start = started_at(&pool, community, &pending).await;
|
||||||
|
assert!(pending_start >= first);
|
||||||
|
db.apply_personal_read_intent(community, &pending, &intent)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(started_at(&pool, community, &pending).await, pending_start);
|
||||||
|
assert_eq!(started_at(&pool, community, &actor).await, first);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[ignore = "requires Postgres"]
|
||||||
|
async fn personal_read_counts_only_arrivals_after_the_account_starts() {
|
||||||
|
let (db, pool, community, channel, actor, _) = fixture().await;
|
||||||
|
unstart(&pool, community).await;
|
||||||
|
let unread = |page: &SidebarPage, id: Uuid| {
|
||||||
|
let row = page.channels.iter().find(|c| c.channel_id == id).unwrap();
|
||||||
|
match row.unread {
|
||||||
|
ReadCount::Exact { value } => value,
|
||||||
|
_ => panic!("expected an exact count"),
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let sidebar = || async {
|
||||||
|
db.personal_read_sidebar(
|
||||||
|
community,
|
||||||
|
&actor.public_key(),
|
||||||
|
DEFAULT_RETENTION_SECONDS,
|
||||||
|
20,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
};
|
||||||
|
// Before the first read intent nothing counts, not the 30-day window.
|
||||||
|
assert_eq!(unread(&sidebar().await, channel), 0);
|
||||||
|
|
||||||
|
// The first intent, in another channel, starts the account everywhere.
|
||||||
|
let other = db
|
||||||
|
.create_channel(
|
||||||
|
community,
|
||||||
|
"elsewhere",
|
||||||
|
ChannelType::Stream,
|
||||||
|
ChannelVisibility::Open,
|
||||||
|
None,
|
||||||
|
&actor.public_key().to_bytes(),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.id;
|
||||||
|
let elsewhere = EventBuilder::new(Kind::Custom(9), "elsewhere")
|
||||||
|
.sign_with_keys(&Keys::generate())
|
||||||
|
.unwrap();
|
||||||
|
db.insert_event(community, &elsewhere, Some(other))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
db.apply_personal_read_intent(
|
||||||
|
community,
|
||||||
|
&actor.public_key(),
|
||||||
|
&ReadIntent::MarkThrough {
|
||||||
|
target: ReadTarget {
|
||||||
|
channel_id: other,
|
||||||
|
root_id: None,
|
||||||
|
},
|
||||||
|
message_id: elsewhere.id.to_hex(),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
// A channel never marked starts caught up: earlier arrivals stay read.
|
||||||
|
assert_eq!(unread(&sidebar().await, channel), 0);
|
||||||
|
|
||||||
|
// Arrivals after the start count.
|
||||||
|
let later = EventBuilder::new(Kind::Custom(9), "after the start")
|
||||||
|
.sign_with_keys(&Keys::generate())
|
||||||
|
.unwrap();
|
||||||
|
db.insert_event(community, &later, Some(channel))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(unread(&sidebar().await, channel), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[ignore = "requires Postgres"]
|
||||||
|
async fn personal_read_unstarted_account_is_exactly_caught_up_past_the_scan_cap() {
|
||||||
|
let (db, pool, community, channel, actor, _) = fixture().await;
|
||||||
|
unstart(&pool, community).await;
|
||||||
|
let last = add_history(&db, community, channel, MAX_UNREAD_SCAN + 44).await;
|
||||||
|
let page = db
|
||||||
|
.personal_read_sidebar(
|
||||||
|
community,
|
||||||
|
&actor.public_key(),
|
||||||
|
DEFAULT_RETENTION_SECONDS,
|
||||||
|
20,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let row = &page.channels[0];
|
||||||
|
// The start floor covers rows the cap left unexamined: zero, not unknown.
|
||||||
|
assert!(matches!(row.unread, ReadCount::Exact { value: 0 }));
|
||||||
|
assert!(matches!(row.attention, ReadCount::Exact { value: 0 }));
|
||||||
|
assert!(row.threads.items.is_empty());
|
||||||
|
assert!(row.threads.complete);
|
||||||
|
assert_eq!(
|
||||||
|
row.latest_message_id.as_deref(),
|
||||||
|
Some(last.id.to_hex().as_str())
|
||||||
|
);
|
||||||
|
assert!(row.latest_message_complete);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn personal_read_channel_and_thread_never_inherit_each_other() {
|
async fn personal_read_channel_and_thread_never_inherit_each_other() {
|
||||||
@@ -777,7 +981,7 @@ async fn personal_read_contexts_bound_selectors_and_use_only_matching_frontiers(
|
|||||||
.fetch_one(&pool)
|
.fetch_one(&pool)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(accounts, 0, "context GET must never create authority");
|
assert_eq!(accounts, 1, "context GET must never create authority");
|
||||||
db.apply_personal_read_intent(
|
db.apply_personal_read_intent(
|
||||||
community,
|
community,
|
||||||
&actor.public_key(),
|
&actor.public_key(),
|
||||||
@@ -824,17 +1028,26 @@ async fn personal_read_contexts_bound_selectors_and_use_only_matching_frontiers(
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(page["contexts"][1]["messages"][1]["status"], "read");
|
assert_eq!(page["contexts"][1]["messages"][1]["status"], "read");
|
||||||
let other = Keys::generate();
|
let other = Keys::generate();
|
||||||
let page = serde_json::to_value(
|
let other_page = || async {
|
||||||
db.personal_read_contexts(
|
serde_json::to_value(
|
||||||
community,
|
db.personal_read_contexts(
|
||||||
&other.public_key(),
|
community,
|
||||||
DEFAULT_RETENTION_SECONDS,
|
&other.public_key(),
|
||||||
&queries,
|
DEFAULT_RETENTION_SECONDS,
|
||||||
|
&queries,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap(),
|
||||||
)
|
)
|
||||||
.await
|
.unwrap()
|
||||||
.unwrap(),
|
};
|
||||||
)
|
// Nothing counts before an account starts.
|
||||||
.unwrap();
|
let page = other_page().await;
|
||||||
|
assert_eq!(page["contexts"][0]["messages"][0]["status"], "read");
|
||||||
|
assert_eq!(page["contexts"][1]["messages"][1]["status"], "read");
|
||||||
|
// Frontiers are per actor: the actor's marks never read for another.
|
||||||
|
start_before_everything(&pool, community, &other.public_key()).await;
|
||||||
|
let page = other_page().await;
|
||||||
assert_eq!(page["contexts"][0]["messages"][0]["status"], "unread");
|
assert_eq!(page["contexts"][0]["messages"][0]["status"], "unread");
|
||||||
assert_eq!(page["contexts"][1]["messages"][1]["status"], "unread");
|
assert_eq!(page["contexts"][1]["messages"][1]["status"], "unread");
|
||||||
assert!(db
|
assert!(db
|
||||||
|
|||||||
@@ -77,7 +77,7 @@ impl Db {
|
|||||||
writes::deadlines(&mut tx).await?;
|
writes::deadlines(&mut tx).await?;
|
||||||
sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?;
|
sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?;
|
||||||
let actor_bytes = actor.to_bytes();
|
let actor_bytes = actor.to_bytes();
|
||||||
let account = read_account(&mut tx, retention_seconds).await?;
|
let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?;
|
||||||
// The inner event LIMIT is deliberately before eligibility filtering.
|
// The inner event LIMIT is deliberately before eligibility filtering.
|
||||||
// This bounds rows/joins even with long deleted or self-authored runs.
|
// This bounds rows/joins even with long deleted or self-authored runs.
|
||||||
// Aggregate equivalent eligible evidence before transfer. Multiplicity
|
// Aggregate equivalent eligible evidence before transfer. Multiplicity
|
||||||
@@ -132,10 +132,11 @@ impl Db {
|
|||||||
), classified AS (
|
), classified AS (
|
||||||
SELECT e.*, tm.root_event_id AS root, tm.parent_event_id AS parent,
|
SELECT e.*, tm.root_event_id AS root, tm.parent_event_id AS parent,
|
||||||
COALESCE(tm.root_event_id<>e.id,false) AS is_reply,
|
COALESCE(tm.root_event_id<>e.id,false) AS is_reply,
|
||||||
COALESCE(e.received_at <=
|
-- Every frontier is floored at the account's start.
|
||||||
|
COALESCE(e.received_at <= GREATEST(COALESCE($10::timestamptz,'infinity'),
|
||||||
CASE WHEN tm.root_event_id IS NOT NULL AND tm.root_event_id<>e.id
|
CASE WHEN tm.root_event_id IS NOT NULL AND tm.root_event_id<>e.id
|
||||||
THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp)
|
THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp)
|
||||||
ELSE cf.through_timestamp END,false) AS covered
|
ELSE cf.through_timestamp END),false) AS covered
|
||||||
FROM (SELECT * FROM candidates ORDER BY created_at DESC,id LIMIT $8-1) e
|
FROM (SELECT * FROM candidates ORDER BY created_at DESC,id LIMIT $8-1) e
|
||||||
LEFT JOIN thread_metadata tm ON tm.community_id=$1 AND tm.channel_id=r.id
|
LEFT JOIN thread_metadata tm ON tm.community_id=$1 AND tm.channel_id=r.id
|
||||||
AND tm.event_created_at=e.created_at AND tm.event_id=e.id
|
AND tm.event_created_at=e.created_at AND tm.event_id=e.id
|
||||||
@@ -189,8 +190,12 @@ impl Db {
|
|||||||
.ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?)
|
.ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?)
|
||||||
.bind((MAX_UNREAD_SCAN+1) as i64)
|
.bind((MAX_UNREAD_SCAN+1) as i64)
|
||||||
.bind(only)
|
.bind(only)
|
||||||
|
.bind(account.started_at)
|
||||||
.fetch_all(&mut *tx).await?;
|
.fetch_all(&mut *tx).await?;
|
||||||
let has_more = rows.len() > limit;
|
let has_more = rows.len() > limit;
|
||||||
|
// An unstarted account has read every arrival, scanned or not, so its
|
||||||
|
// counts are exactly zero whatever the scan cap left unexamined.
|
||||||
|
let started = account.started_at.is_some();
|
||||||
let mut channels = Vec::new();
|
let mut channels = Vec::new();
|
||||||
let mut pending = Vec::new();
|
let mut pending = Vec::new();
|
||||||
for row in rows.into_iter().take(limit) {
|
for row in rows.into_iter().take(limit) {
|
||||||
@@ -198,7 +203,8 @@ impl Db {
|
|||||||
let evidence = evidence
|
let evidence = evidence
|
||||||
.as_array()
|
.as_array()
|
||||||
.ok_or_else(|| DbError::InvalidData("invalid sidebar evidence".into()))?;
|
.ok_or_else(|| DbError::InvalidData("invalid sidebar evidence".into()))?;
|
||||||
let complete = row.try_get::<i64, _>("scanned")? <= MAX_UNREAD_SCAN as i64;
|
let evidence = if started { evidence.as_slice() } else { &[] };
|
||||||
|
let complete = !started || row.try_get::<i64, _>("scanned")? <= MAX_UNREAD_SCAN as i64;
|
||||||
let channel_type: String = row.try_get("channel_type")?;
|
let channel_type: String = row.try_get("channel_type")?;
|
||||||
let mut unread = 0;
|
let mut unread = 0;
|
||||||
let mut attention = 0;
|
let mut attention = 0;
|
||||||
@@ -379,20 +385,28 @@ pub(super) fn summarize(
|
|||||||
ThreadSummaries { items, complete }
|
ThreadSummaries { items, complete }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Read-time horizon only: frontier state is not discarded on expiry.
|
/// Read-time horizon and the account's start, in the caller's snapshot.
|
||||||
|
/// Frontier state is not discarded on expiry.
|
||||||
pub(super) async fn read_account(
|
pub(super) async fn read_account(
|
||||||
conn: &mut PgConnection,
|
conn: &mut PgConnection,
|
||||||
|
community: CommunityId,
|
||||||
|
actor: &[u8],
|
||||||
retention_seconds: u32,
|
retention_seconds: u32,
|
||||||
) -> Result<ReadAccount> {
|
) -> Result<ReadAccount> {
|
||||||
let cutoff: DateTime<Utc> = sqlx::query_scalar(
|
let row = sqlx::query(
|
||||||
"SELECT date_trunc('milliseconds',transaction_timestamp()-make_interval(secs=>$1::double precision))",
|
"SELECT date_trunc('milliseconds',transaction_timestamp()-make_interval(secs=>$1::double precision)) AS cutoff,
|
||||||
|
(SELECT started_at FROM personal_read_accounts WHERE community_id=$2 AND actor=$3) AS started_at",
|
||||||
)
|
)
|
||||||
.bind(f64::from(retention_seconds))
|
.bind(f64::from(retention_seconds))
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(actor)
|
||||||
.fetch_one(conn)
|
.fetch_one(conn)
|
||||||
.await?;
|
.await?;
|
||||||
|
let cutoff: DateTime<Utc> = row.try_get("cutoff")?;
|
||||||
Ok(ReadAccount {
|
Ok(ReadAccount {
|
||||||
retention_seconds,
|
retention_seconds,
|
||||||
cutoff_ms: cutoff.timestamp_millis(),
|
cutoff_ms: cutoff.timestamp_millis(),
|
||||||
|
started_at: row.try_get("started_at")?,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -23,15 +23,17 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Serialize private frontier writes, never shared conversation rows.
|
/// Start the account if this is the actor's first read intent, then serialize
|
||||||
|
/// private frontier writes, never shared conversation rows.
|
||||||
pub(super) async fn lock_account(
|
pub(super) async fn lock_account(
|
||||||
conn: &mut PgConnection,
|
conn: &mut PgConnection,
|
||||||
community: CommunityId,
|
community: CommunityId,
|
||||||
actor: &[u8],
|
actor: &[u8],
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
"INSERT INTO personal_read_accounts (community_id,actor) VALUES ($1,$2)
|
"INSERT INTO personal_read_accounts (community_id,actor,started_at) VALUES ($1,$2,now())
|
||||||
ON CONFLICT (community_id,actor) DO NOTHING",
|
ON CONFLICT (community_id,actor) DO UPDATE SET started_at=now()
|
||||||
|
WHERE personal_read_accounts.started_at IS NULL",
|
||||||
)
|
)
|
||||||
.bind(community.as_uuid())
|
.bind(community.as_uuid())
|
||||||
.bind(actor)
|
.bind(actor)
|
||||||
|
|||||||
@@ -56,6 +56,24 @@ async fn request(
|
|||||||
(status, serde_json::from_slice(&bytes).unwrap())
|
(status, serde_json::from_slice(&bytes).unwrap())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Start a reader's account before any fixture arrival. Nothing counts as
|
||||||
|
/// unread until an account starts.
|
||||||
|
async fn start_before_everything(
|
||||||
|
state: &crate::state::AppState,
|
||||||
|
community: buzz_core::CommunityId,
|
||||||
|
key: &Keys,
|
||||||
|
) {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO personal_read_accounts (community_id,actor,started_at)
|
||||||
|
VALUES ($1,$2,'1900-01-01T00:00:00Z')",
|
||||||
|
)
|
||||||
|
.bind(community.as_uuid())
|
||||||
|
.bind(key.public_key().to_bytes().as_slice())
|
||||||
|
.execute(state.db.pool())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn accessory_router_signed_url_body_replay_and_actor_boundary() {
|
async fn accessory_router_signed_url_body_replay_and_actor_boundary() {
|
||||||
@@ -217,6 +235,7 @@ async fn accessory_router_signed_url_body_replay_and_actor_boundary() {
|
|||||||
.map(|b| format!("%{b:02X}"))
|
.map(|b| format!("%{b:02X}"))
|
||||||
.collect();
|
.collect();
|
||||||
let path = format!("/buzz/v1/me/read-state?targets={targets}");
|
let path = format!("/buzz/v1/me/read-state?targets={targets}");
|
||||||
|
start_before_everything(&state, community, &other).await;
|
||||||
for (key, status) in [(&actor, "read"), (&other, "unread")] {
|
for (key, status) in [(&actor, "read"), (&other, "unread")] {
|
||||||
let auth = proof(key, &host, &path, "GET", None);
|
let auth = proof(key, &host, &path, "GET", None);
|
||||||
let page = request(state.clone(), &host, &path, "GET", Some(&auth), b"").await;
|
let page = request(state.clone(), &host, &path, "GET", Some(&auth), b"").await;
|
||||||
@@ -275,6 +294,7 @@ async fn accessory_context_get_signed_query_and_independent_batch_outcomes() {
|
|||||||
"/buzz/v1/me/read-state?targets={}",
|
"/buzz/v1/me/read-state?targets={}",
|
||||||
encode(&targets.to_string())
|
encode(&targets.to_string())
|
||||||
);
|
);
|
||||||
|
start_before_everything(&state, community, &actor).await;
|
||||||
let auth = proof(&actor, &host, &path, "GET", None);
|
let auth = proof(&actor, &host, &path, "GET", None);
|
||||||
let result = request(state.clone(), &host, &path, "GET", Some(&auth), b"").await;
|
let result = request(state.clone(), &host, &path, "GET", Some(&auth), b"").await;
|
||||||
assert_eq!(result.0, StatusCode::OK, "{}", result.1);
|
assert_eq!(result.0, StatusCode::OK, "{}", result.1);
|
||||||
@@ -609,6 +629,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) {
|
|||||||
.sign_with_keys(signer)
|
.sign_with_keys(signer)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let path = format!("/buzz/v1/me/sidebar?channel_ids={channel}");
|
let path = format!("/buzz/v1/me/sidebar?channel_ids={channel}");
|
||||||
|
start_before_everything(&state, community, &reader).await;
|
||||||
for (event, key, count) in [(&message, &author, 1), (&deletion, signer, 0)] {
|
for (event, key, count) in [(&message, &author, 1), (&deletion, signer, 0)] {
|
||||||
let body = serde_json::to_vec(event).unwrap();
|
let body = serde_json::to_vec(event).unwrap();
|
||||||
let auth = proof(key, &host, "/events", "POST", Some(&body));
|
let auth = proof(key, &host, "/events", "POST", Some(&body));
|
||||||
|
|||||||
@@ -229,9 +229,12 @@ examined: the 4,096 most recent events by author time inside the horizon. So
|
|||||||
marking through it reads everything counted. When the horizon holds no message,
|
marking through it reads everything counted. When the horizon holds no message,
|
||||||
it is the last to arrive among the channel's 256 most recent events.
|
it is the last to arrive among the channel's 256 most recent events.
|
||||||
|
|
||||||
There is no import of earlier client read state: an account starts with no
|
There is no import of earlier client read state. The relay records the
|
||||||
frontiers, and the horizon bounds what that can show as unread. Manual unread
|
account's `started_at` at its first applied read intent and never moves it.
|
||||||
remains device-local.
|
Every frontier is floored at `started_at`: until the account starts nothing
|
||||||
|
counts as unread, and from then it starts caught up, so a context it has never
|
||||||
|
marked counts only arrivals after the start. Manual unread remains
|
||||||
|
device-local.
|
||||||
|
|
||||||
## Bounds and deployment
|
## Bounds and deployment
|
||||||
|
|
||||||
@@ -251,10 +254,11 @@ remains device-local.
|
|||||||
shared eight-second intent-processing deadline after admission. Limits are
|
shared eight-second intent-processing deadline after admission. Limits are
|
||||||
containment, not a production capacity claim.
|
containment, not a production capacity claim.
|
||||||
|
|
||||||
Apply migration 0056 (or the equivalent desired schema). It creates two empty
|
Apply migrations 0056 and 0057 (or the equivalent desired schema). 0056
|
||||||
private tables and no index on `events`: both sidebar scans are served by the
|
creates two empty private tables and no index on `events`, and 0057 adds the
|
||||||
existing `idx_events_community_channel_created`. No new per-message ingest write
|
nullable `personal_read_accounts.started_at` column. Both sidebar scans are
|
||||||
path or stored unread counters are introduced.
|
served by the existing `idx_events_community_channel_created`. No new
|
||||||
|
per-message ingest write path or stored unread counters are introduced.
|
||||||
|
|
||||||
Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and
|
Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and
|
||||||
`/buzz/v1/me/read-state`, plus database pool/statement metrics. Inspect exact /
|
`/buzz/v1/me/read-state`, plus database pool/statement metrics. Inspect exact /
|
||||||
@@ -292,10 +296,8 @@ is no new public account export/reset endpoint. Operator-assisted erasure/export
|
|||||||
must use the established authenticated operational process and explicitly scope
|
must use the established authenticated operational process and explicitly scope
|
||||||
both community and actor; never equate the read-time horizon with data erasure.
|
both community and actor; never equate the read-time horizon with data erasure.
|
||||||
|
|
||||||
Migration 0056 must be applied before this relay serves, enabled or not: started
|
There is no down migration, and disabling the API is not a rollback. A relay
|
||||||
without it and with auto-migration off, the relay stops before readiness. There
|
built before 0056 that restarts with `BUZZ_AUTO_MIGRATE=true` (the Helm default)
|
||||||
is no down migration, and disabling the API is not a rollback. A relay built
|
|
||||||
before 0056 that restarts with `BUZZ_AUTO_MIGRATE=true` (the Helm default)
|
|
||||||
refuses to start on the migrated schema. Whole-community deletion run from a
|
refuses to start on the migrated schema. Whole-community deletion run from a
|
||||||
build before 0056 rejects the two new tables. A deletion approved on the
|
build before 0056 rejects the two new tables. A deletion approved on the
|
||||||
earlier schema and not yet fenced fails structural revalidation after 0056:
|
earlier schema and not yet fenced fails structural revalidation after 0056:
|
||||||
|
|||||||
4
migrations/0057_personal_read_started_at.sql
Normal file
4
migrations/0057_personal_read_started_at.sql
Normal file
@@ -0,0 +1,4 @@
|
|||||||
|
-- started_at is the actor's first read intent. NULL until then: an account
|
||||||
|
-- can exist before its actor has started. Read positions are floored at it,
|
||||||
|
-- so a started account begins caught up and an unstarted one counts nothing.
|
||||||
|
ALTER TABLE personal_read_accounts ADD COLUMN started_at TIMESTAMPTZ;
|
||||||
@@ -2,8 +2,11 @@
|
|||||||
-- A frontier is the relay arrival time (events.received_at) of the message a
|
-- A frontier is the relay arrival time (events.received_at) of the message a
|
||||||
-- context was read through; the unread cutoff alone is signed event time. An
|
-- context was read through; the unread cutoff alone is signed event time. An
|
||||||
-- empty root_id denotes a channel frontier.
|
-- empty root_id denotes a channel frontier.
|
||||||
|
-- started_at is the actor's first read intent. NULL until then: an account
|
||||||
|
-- can exist before its actor has started.
|
||||||
CREATE TABLE personal_read_accounts (
|
CREATE TABLE personal_read_accounts (
|
||||||
community_id UUID NOT NULL REFERENCES communities(id),
|
community_id UUID NOT NULL REFERENCES communities(id),
|
||||||
actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
|
actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
|
||||||
|
started_at TIMESTAMPTZ,
|
||||||
PRIMARY KEY (community_id, actor)
|
PRIMARY KEY (community_id, actor)
|
||||||
);
|
);
|
||||||
|
|||||||
Reference in New Issue
Block a user