diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 10c8487bb87..5b43d443642 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -77,6 +77,10 @@ pub enum DbError { #[error("deletion safety error: {0}")] DeletionSafety(String), + /// An artifact sync page exceeds its resource limit. + #[error("artifact sync limit exceeded")] + ArtifactSyncLimit, + /// A complete read-state snapshot exceeds its bounded resource budget. #[error("read-state snapshot exceeds event or byte limit")] ReadStateSnapshotTooLarge, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9205bdc8992..8d012dfae7d 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -65,7 +65,7 @@ pub(crate) use runtime::{ sweep_admitted_communities, ReadSessionInner, RouteDecision, RoutePredicate, }; pub use store::{ - admin_moderation, allowlist, api_token, archived_identities, artifact, channel, + admin_moderation, allowlist, api_token, archived_identities, artifact, artifact_sync, channel, channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation, nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 7a1a551b7c0..3421a74c3ad 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -776,7 +776,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 75); + assert_eq!(migrations.len(), 78); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -2918,6 +2918,10 @@ mod postgres_tests { "relay_operator_audit", "entity_deletion_approvals", "entity_deletion_requests", + "artifact_heads", + "artifact_revisions", + "channel_members", + "communities", "entity_deletion_stage_history", "user_deletion_fences", ] { diff --git a/crates/buzz-db/src/store/artifact.rs b/crates/buzz-db/src/store/artifact.rs index a8a49ac602b..6f9f46eb174 100644 --- a/crates/buzz-db/src/store/artifact.rs +++ b/crates/buzz-db/src/store/artifact.rs @@ -185,14 +185,15 @@ impl Db { } } sqlx::query( - "INSERT INTO artifact_revisions (community_id,event_id,artifact_id) VALUES ($1,$2,$3)", + "INSERT INTO artifact_revisions (community_id,event_id,artifact_id,source_channel_id,modified_at) VALUES ($1,$2,$3,$4,clock_timestamp())", ) .bind(community.as_uuid()) .bind(event.id.as_bytes().as_slice()) .bind(env.id) + .bind(if env.op == ArtifactOp::Move { source } else { None }) .execute(tx.conn()) .await?; - sqlx::query("INSERT INTO artifact_heads (community_id,artifact_id,event_id,channel_id,artifact_type,root,deleted) VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (community_id,artifact_id) DO UPDATE SET event_id=EXCLUDED.event_id,channel_id=EXCLUDED.channel_id,root=EXCLUDED.root,deleted=EXCLUDED.deleted") + sqlx::query("INSERT INTO artifact_heads (community_id,artifact_id,event_id,channel_id,artifact_type,root,deleted,modified_at) VALUES ($1,$2,$3,$4,$5,$6,$7,clock_timestamp()) ON CONFLICT (community_id,artifact_id) DO UPDATE SET event_id=EXCLUDED.event_id,channel_id=EXCLUDED.channel_id,root=EXCLUDED.root,deleted=EXCLUDED.deleted,modified_at=clock_timestamp()") .bind(community.as_uuid()).bind(env.id).bind(event.id.as_bytes().as_slice()).bind(env.home).bind(&env.artifact_type).bind(&env.root).bind(env.op == ArtifactOp::Delete).execute(tx.conn()).await?; let (stored, _) = crate::event::insert_event_in_transaction(&mut tx, event, Some(env.home)).await?; diff --git a/crates/buzz-db/src/store/artifact_query.rs b/crates/buzz-db/src/store/artifact_query.rs index f359726d925..6dc38ba11ff 100644 --- a/crates/buzz-db/src/store/artifact_query.rs +++ b/crates/buzz-db/src/store/artifact_query.rs @@ -33,46 +33,7 @@ impl Db { } else { "SELECT e.id,e.pubkey,e.created_at,e.kind,e.tags,e.content,e.sig,e.received_at,e.channel_id FROM events e " }); - if query.view == ArtifactView::Current { - qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); - } - // A system channel, or an unlisted open channel the reader is not a - // member of, is read only by a query whose `#h` names it (NIP-CL); - // malformed `#h` values name no channel. - let named_channels: Vec = query - .tags - .iter() - .filter(|(name, _)| name == "h") - .flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok())) - .collect(); - // The explicit kind lets the community/kind index exclude chat before - // the join, even for an unfiltered current-state page. - qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (") - .push(crate::store::channel::legacy_channel_type_sql("c.channel_type")) - .push(" OR c.id=ANY(") - .push_bind(named_channels.clone()) - .push(")) AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY(") - .push_bind(named_channels) - .push("))) OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") - .push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))"); - for (name, values) in &query.tags { - // GIN-usable necessary prefilter; retain the positional recheck - // because JSON array containment alone also matches swapped values. - qb.push(" AND ("); - for (index, value) in values.iter().enumerate() { - if index != 0 { - qb.push(" OR "); - } - qb.push("e.tags @> ") - .push_bind(serde_json::json!([[name, value]])); - } - qb.push(")"); - qb.push(" AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) tag WHERE tag->>0=") - .push_bind(name) - .push(" AND tag->>1=ANY(") - .push_bind(values) - .push("))"); - } + predicates(&mut qb, community, reader, query); if count { let n: i64 = qb.build_query_scalar().fetch_one(&mut *tx).await?; tx.commit().await?; @@ -93,3 +54,68 @@ impl Db { Ok((events, 0)) } } + +/// Own the visibility and tag rules for artifact queries and current-ID reads. +pub(super) fn predicates( + qb: &mut QueryBuilder, + community: CommunityId, + reader: &[u8], + query: &ArtifactQuery, +) { + if query.view == ArtifactView::Current { + qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); + } + // System channels and unlisted open channels require an explicit `#h` + // unless the reader is a member of the unlisted channel (NIP-CL). + let named_channels: Vec = query + .tags + .iter() + .filter(|(name, _)| name == "h") + .flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok())) + .collect(); + // The explicit kind lets the community/kind index exclude chat before + // the join, even for an unfiltered current-state page. + qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (") + .push(crate::store::channel::legacy_channel_type_sql("c.channel_type")) + .push(" OR c.id=ANY(") + .push_bind(named_channels.clone()) + .push(")) AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY(") + .push_bind(named_channels) + .push("))) OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") + .push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))"); + tag_predicates(qb, query); + if query.view == ArtifactView::Current { + for (_, values) in query.tags.iter().filter(|(name, _)| name == "d") { + qb.push(" AND a.artifact_id=ANY(") + .push_bind( + values + .iter() + .filter_map(|id| id.parse::().ok()) + .collect::>(), + ) + .push(")"); + } + } +} + +/// Apply exact artifact tag predicates after access is resolved. +pub(super) fn tag_predicates(qb: &mut QueryBuilder, query: &ArtifactQuery) { + for (name, values) in &query.tags { + // GIN-usable necessary prefilter; retain the positional recheck + // because JSON array containment alone also matches swapped values. + qb.push(" AND ("); + for (index, value) in values.iter().enumerate() { + if index != 0 { + qb.push(" OR "); + } + qb.push("e.tags @> ") + .push_bind(serde_json::json!([[name, value]])); + } + qb.push(")"); + qb.push(" AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) tag WHERE tag->>0=") + .push_bind(name) + .push(" AND tag->>1=ANY(") + .push_bind(values) + .push("))"); + } +} diff --git a/crates/buzz-db/src/store/artifact_sync.rs b/crates/buzz-db/src/store/artifact_sync.rs new file mode 100644 index 00000000000..4da01728fd0 --- /dev/null +++ b/crates/buzz-db/src/store/artifact_sync.rs @@ -0,0 +1,205 @@ +//! Indexed artifact changes. Each page reads current authority on the writer. +use crate::{Db, DbError, Result}; +use buzz_core::{artifact::ArtifactQuery, CommunityId, StoredEvent}; +use chrono::{DateTime, Utc}; +use futures_util::TryStreamExt; +use serde::{Deserialize, Serialize}; +use sqlx::{QueryBuilder, Row}; +use uuid::Uuid; + +/// Maximum changes or requested identities in one page. +pub const PAGE_ROWS: usize = 100; + +/// A stable position within one timestamp window. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Position { + /// Database timestamp of the change. + pub at: DateTime, + /// Change source, which separates equal keys from different tables. + pub source: i32, + /// Stable identity within that source. + pub id: String, +} + +/// A changed identity that requires a current, permission-scoped read. +pub struct Candidate { + /// Position used after this candidate is processed. + pub position: Position, + /// Artifact to revalidate, absent for a removed membership. + pub artifact_id: Option, + /// Removed channel belonging to the authenticated reader's membership. + pub removed_channel: Option, +} + +impl Db { + /// Read server time and the community's access-policy recovery boundary. + pub async fn artifact_sync_boundary( + &self, + community: CommunityId, + ) -> Result<(DateTime, DateTime)> { + sqlx::query_as("SELECT clock_timestamp(),artifact_policy_modified_at FROM communities WHERE id=$1 AND deletion_state='active' AND deleted_at IS NULL") + .bind(community.as_uuid()).fetch_optional(&self.pool).await? + .ok_or_else(|| DbError::InvalidData("artifact sync community unavailable".into())) + } + + /// Find at most one page plus one look-ahead row, without a channel roster. + pub async fn artifact_sync_candidates( + &self, + community: CommunityId, + reader: &[u8], + since: Option>, + until: DateTime, + after: Option<&Position>, + query: &ArtifactQuery, + ) -> Result> { + let connection = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::SubscriptionHistory, + ) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + sqlx::query("SET LOCAL statement_timeout='2s'") + .execute(&mut *tx) + .await?; + let types = query + .tags + .iter() + .filter(|(n, _)| n == "type") + .flat_map(|(_, v)| v.iter().cloned()) + .collect::>(); + let named = query + .tags + .iter() + .filter(|(n, _)| n == "h") + .flat_map(|(_, v)| v.iter().filter_map(|id| id.parse::().ok())) + .collect::>(); + let rows = sqlx::query(CANDIDATES) + .bind(community.as_uuid()) + .bind(reader) + .bind(since) + .bind(until) + .bind(after.map(|p| p.at)) + .bind(after.map(|p| p.source)) + .bind(after.map(|p| p.id.as_str())) + .bind(types) + .bind(named) + .bind((PAGE_ROWS + 1) as i64) + .bind(buzz_core::channel::LEGACY_CHANNEL_TYPES.to_vec()) + .bind(query.tags.iter().any(|(name, _)| name == "h")) + .fetch_all(&mut *tx) + .await?; + let candidates = rows + .into_iter() + .map(|r| { + Ok(Candidate { + position: Position { + at: r.try_get("at")?, + source: r.try_get("source")?, + id: r.try_get("id")?, + }, + artifact_id: r.try_get("artifact_id")?, + removed_channel: r.try_get("removed_channel")?, + }) + }) + .collect::>>()?; + tx.commit().await?; + Ok(candidates) + } + + /// Return current matching events for bounded known identities. + /// Omitted identities have the same unavailable meaning for every cause. + pub async fn artifact_sync_current( + &self, + community: CommunityId, + reader: &[u8], + query: &ArtifactQuery, + ids: &[Uuid], + ) -> Result> { + if ids.len() > PAGE_ROWS { + return Err(DbError::ArtifactSyncLimit); + } + if ids.is_empty() { + return Ok(vec![]); + } + let connection = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::SubscriptionHistory, + ) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL statement_timeout='2s'") + .execute(&mut *tx) + .await?; + let mut qb = QueryBuilder::new("SELECT e.id,e.pubkey,e.created_at,e.kind,e.tags,e.content,e.sig,e.received_at,e.channel_id FROM events e "); + super::artifact_query::predicates(&mut qb, community, reader, query); + qb.push(" AND a.artifact_id=ANY(").push_bind(ids).push(")"); + qb.push(" AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.ttl_seconds IS NULL)"); + qb.push(" LIMIT ").push_bind(PAGE_ROWS as i64); + let mut events = Vec::new(); + let mut bytes = 0; + { + let mut rows = qb.build().fetch(&mut *tx); + while let Some(row) = rows.try_next().await? { + if let Some(event) = crate::event::row_to_stored_event(row)? { + bytes += serde_json::to_vec(&event.event) + .map_err(|e| DbError::InvalidData(e.to_string()))? + .len() + + 512; + if bytes > 8_388_608 { + return Err(DbError::ArtifactSyncLimit); + } + events.push(event); + } + } + } + tx.commit().await?; + Ok(events) + } +} + +// Keep candidate access rules aligned with artifact_query::predicates, which owns +// current-state visibility. This query also reads retained move and membership rows. +const CANDIDATES: &str = r#" +WITH readable AS NOT MATERIALIZED ( + SELECT c.id FROM channels c + WHERE c.community_id=$1 AND c.deleted_at IS NULL AND c.ttl_seconds IS NULL + AND (NOT $12::boolean OR c.id=ANY($9)) + AND (c.channel_type::text=ANY($11) OR c.id=ANY($9)) + AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY($9))) OR EXISTS ( + SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id + AND m.channel_id=c.id AND m.pubkey=$2 AND m.removed_at IS NULL)) +), changes AS ( + SELECT a.modified_at AS at,1 AS source,a.artifact_id::text AS id, + a.artifact_id,NULL::uuid AS removed_channel + FROM artifact_heads a + WHERE a.community_id=$1 AND a.modified_at >= COALESCE($3::timestamptz,'epoch'::timestamptz) + AND a.modified_at <= $4 AND a.channel_id IN (SELECT id FROM readable) + AND ($3::timestamptz IS NOT NULL OR NOT a.deleted) + AND (cardinality($8::text[])=0 OR a.artifact_type=ANY($8)) + UNION ALL + SELECT m.sync_modified_at,0,m.channel_id::text,NULL::uuid,m.channel_id + FROM channel_members m WHERE m.community_id=$1 AND m.pubkey=$2 + AND $3::timestamptz IS NOT NULL AND m.sync_modified_at >= $3 + AND m.sync_modified_at <= $4 AND m.removed_at IS NOT NULL + AND (NOT $12::boolean OR m.channel_id=ANY($9)) + AND NOT EXISTS (SELECT 1 FROM readable r WHERE r.id=m.channel_id) + UNION ALL + SELECT m.sync_modified_at,2,a.artifact_id::text,a.artifact_id,NULL::uuid + FROM channel_members m JOIN artifact_heads a + ON a.community_id=m.community_id AND a.channel_id=m.channel_id + WHERE m.community_id=$1 AND m.pubkey=$2 AND m.removed_at IS NULL + AND $3::timestamptz IS NOT NULL AND m.sync_modified_at >= $3 + AND m.sync_modified_at <= $4 AND a.channel_id IN (SELECT id FROM readable) + AND (cardinality($8::text[])=0 OR a.artifact_type=ANY($8)) + UNION ALL + SELECT DISTINCT r.modified_at,3,r.artifact_id::text,r.artifact_id,NULL::uuid + FROM artifact_revisions r WHERE r.community_id=$1 + AND r.source_channel_id IS NOT NULL AND r.source_channel_id IN (SELECT id FROM readable) + AND $3::timestamptz IS NOT NULL AND r.modified_at >= $3 AND r.modified_at <= $4 +) +SELECT * FROM changes WHERE $5::timestamptz IS NULL OR (at,source,id)>($5,$6,$7) +ORDER BY at,source,id LIMIT $10 +"#; diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 8907b224a93..cbe9ce2053f 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -162,6 +162,7 @@ pub async fn create_channel( INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) VALUES ($1, $2, $3, 'owner', $4) ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + sync_modified_at = CASE WHEN channel_members.removed_at IS NOT NULL THEN clock_timestamp() ELSE channel_members.sync_modified_at END, removed_at = NULL, removed_by = NULL, role = EXCLUDED.role @@ -403,6 +404,7 @@ async fn insert_channel_rows( INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) VALUES ($1, $2, $3, 'owner', $4) ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + sync_modified_at = CASE WHEN channel_members.removed_at IS NOT NULL THEN clock_timestamp() ELSE channel_members.sync_modified_at END, removed_at = NULL, removed_by = NULL, role = EXCLUDED.role @@ -1088,10 +1090,29 @@ async fn update_channel_on( q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); + let policy_changed = if updates.visibility.is_some() + || updates.ttl_seconds.is_some() + || updates.unlisted.is_some() + { + sqlx::query_scalar::<_,bool>("SELECT (($3::text IS NOT NULL AND visibility::text IS DISTINCT FROM $3) OR ($4 AND ttl_seconds IS DISTINCT FROM $5::int) OR ($6::bool IS NOT NULL AND unlisted IS DISTINCT FROM $6)) FROM channels WHERE community_id=$1 AND id=$2") + .bind(community_id.as_uuid()).bind(channel_id).bind(updates.visibility.as_deref()) + .bind(updates.ttl_seconds.is_some()).bind(updates.ttl_seconds.flatten()).bind(updates.unlisted) + .fetch_one(tx.conn()).await? + } else { + false + }; let result = q.execute(tx.conn()).await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + if policy_changed { + sqlx::query( + "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1", + ) + .bind(community_id.as_uuid()) + .execute(tx.conn()) + .await?; + } Ok(()) } @@ -1422,6 +1443,10 @@ async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> { .bind(channel_id) .execute(w.tx.conn()) .await?; + sqlx::query("UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1") + .bind(community_id.as_uuid()) + .execute(w.tx.conn()) + .await?; Ok(()) } diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 480054fa91f..050705356aa 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -256,6 +256,7 @@ impl ChannelMembershipLocks<'_> { INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) VALUES ($1, $2, $3, 'member'::member_role, $4) ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + sync_modified_at = CASE WHEN channel_members.removed_at IS NOT NULL THEN clock_timestamp() ELSE channel_members.sync_modified_at END, removed_at = NULL, removed_by = NULL, role = EXCLUDED.role @@ -932,6 +933,7 @@ async fn put_members_on( SELECT $1, $2, t.pubkey, t.role::member_role, $5 FROM UNNEST($3::bytea[], $4::text[]) AS t(pubkey, role) ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + sync_modified_at = CASE WHEN channel_members.removed_at IS NOT NULL THEN clock_timestamp() ELSE channel_members.sync_modified_at END, removed_at = NULL, removed_by = NULL, role = EXCLUDED.role @@ -1065,7 +1067,7 @@ pub async fn remove_member( let result = sqlx::query( r#" UPDATE channel_members - SET removed_at = NOW(), removed_by = $1 + SET removed_at = clock_timestamp(), sync_modified_at = clock_timestamp(), removed_by = $1 WHERE community_id = $2 AND channel_id = $3 AND pubkey = $4 AND removed_at IS NULL "#, ) diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 0c86e13f050..a9c93477ae6 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -192,6 +192,7 @@ pub async fn create_dm( INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) VALUES ($1, $2, $3, 'member', $4) ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + sync_modified_at = CASE WHEN channel_members.removed_at IS NOT NULL THEN clock_timestamp() ELSE channel_members.sync_modified_at END, removed_at = NULL, removed_by = NULL, role = EXCLUDED.role diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 151151f7385..c85fb6fd017 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1557,6 +1557,8 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx( let deleted = purged > 0 || result.rows_affected() > 0; if deleted { + sqlx::query("UPDATE artifact_heads SET modified_at=clock_timestamp() WHERE community_id=$1 AND artifact_id=(SELECT artifact_id FROM artifact_revisions WHERE community_id=$1 AND event_id=$2) AND event_id=$2") + .bind(community_id.as_uuid()).bind(event_id).execute(tx.conn()).await?; if let Some(pid) = parent_event_id { sqlx::query( "UPDATE thread_metadata \ diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d36fa4d0894..961d50bb179 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -83,6 +83,8 @@ pub mod workflow; pub mod artifact; mod artifact_query; +/// Timestamp-based incremental artifact reads. +pub mod artifact_sync; #[cfg(test)] mod artifact_postgres_tests; diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 19c2ab4c461..fc1bc93770a 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -721,7 +721,7 @@ pub async fn execute_kick_with_marker( let kick = sqlx::query( r#" UPDATE channel_members - SET removed_at = NOW(), removed_by = $1 + SET removed_at = clock_timestamp(), sync_modified_at = clock_timestamp(), removed_by = $1 WHERE community_id = $2 AND channel_id = $3 AND pubkey = $4 AND removed_at IS NULL "#, ) @@ -1713,7 +1713,7 @@ pub async fn deploy_kick_member( let result = sqlx::query( r#" UPDATE channel_members - SET removed_at = NOW(), removed_by = $1 + SET removed_at = clock_timestamp(), sync_modified_at = clock_timestamp(), removed_by = $1 WHERE community_id = $2 AND channel_id = $3 AND pubkey = $4 AND removed_at IS NULL "#, ) diff --git a/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs index fd06aa0e62f..7a8b8fb3442 100644 --- a/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs +++ b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs @@ -506,13 +506,12 @@ async fn apply_pgschema_to_existing_database(pool: &PgPool) { async fn migration_schema_0075_adds_invite_columns_without_rewriting_history() { let (pool, cases) = legacy_fixture().await; crate::migration::run_migrations(&pool).await.unwrap(); - assert_eq!( - sqlx::query_scalar::<_, i64>("SELECT max(version) FROM _sqlx_migrations WHERE success") - .fetch_one(&pool) - .await - .unwrap(), - 75 - ); + assert!(sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM _sqlx_migrations WHERE version=75 AND success)" + ) + .fetch_one(&pool) + .await + .unwrap()); assert_admission_column(&pool).await; assert_join_role_preserved(&pool, cases[0].community).await; assert_eq!( diff --git a/crates/buzz-relay/src/api/artifact.rs b/crates/buzz-relay/src/api/artifact.rs index bfe636c9086..c41ce5d15d0 100644 --- a/crates/buzz-relay/src/api/artifact.rs +++ b/crates/buzz-relay/src/api/artifact.rs @@ -47,6 +47,15 @@ async fn execute( "artifact queries require exactly one filter", )); } + if filters[0].get("artifact").and_then(Value::as_str) == Some("incremental") { + if count { + return Err(api_error( + StatusCode::BAD_REQUEST, + "incremental requires /query", + )); + } + return super::artifact_sync::query(state, tenant, reader, &filters[0]).await; + } let query = buzz_core::artifact::parse_query(&filters[0]) .map_err(|e| api_error(StatusCode::BAD_REQUEST, e))?; // The query's statement_timeout is the enforced execution budget. diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests.rs index 82d57dd2f19..2519510cf4c 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests.rs @@ -120,7 +120,7 @@ impl Fixture { .await .unwrap(); let status = response.status(); - let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024) + let bytes = axum::body::to_bytes(response.into_body(), 8 * 1024 * 1024) .await .unwrap(); (status, serde_json::from_slice(&bytes).unwrap()) @@ -486,3 +486,6 @@ async fn generic_d_lookup_matches_before_limit() { expected.sort(); assert_eq!(ids, expected, "{body}"); } + +#[path = "artifact_postgres_tests/sync_postgres_tests.rs"] +mod sync_postgres_tests; diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs new file mode 100644 index 00000000000..92461132597 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs @@ -0,0 +1,551 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_redaction_and_query_exit_return_removals() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let first = f.revision(id, "create", None); + let first = EventBuilder::new(Kind::Custom(45010), first.content.clone()) + .tags( + first + .tags + .iter() + .cloned() + .chain([Tag::parse(["project", "alpha"]).unwrap()]), + ) + .sign_with_keys(&f.owner) + .unwrap(); + accept(&f, &first).await; + let scope = json!({"artifact":"current","#type":["buzz.task"],"#project":["alpha"]}); + let (_, page) = f + .request("/query", json!([{"artifact":"incremental","query":scope}])) + .await; + assert_eq!(page["changes"][0]["type"], "upsert"); + let changed = f.revision(id, "update", Some(&first)); + accept(&f, &changed).await; + let (status, delta) = f + .request( + "/query", + json!([{"artifact":"incremental","query":scope,"checkpoint":page["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {delta}"); + assert_eq!( + delta["changes"], + json!([{ "type":"remove","artifactId":id }]) + ); + let before = sync(&f, None).await; + assert!(buzz_db::event::soft_delete_event_and_update_thread( + &f.pool, + f.community, + changed.id.as_bytes(), + None, + None + ) + .await + .unwrap()); + let after = sync(&f, Some(&before["checkpoint"])).await; + assert_eq!( + after["changes"], + json!([{ "type":"remove","artifactId":id }]) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_rejoin_replays_old_heads_without_resetting_on_task_edits() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let first = f.revision_as(&f.owner, f.private, id, "create", None); + accept(&f, &first).await; + sqlx::query("UPDATE artifact_heads SET modified_at='2000-01-01' WHERE community_id=$1") + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .unwrap(); + buzz_db::channel_members::add_member( + &f.pool, + f.community, + f.private, + f.peer.public_key().as_bytes(), + buzz_db::channel_members::MemberRole::Member, + Some(f.owner.public_key().as_bytes()), + ) + .await + .unwrap(); + let (_, page) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + buzz_db::channel_members::remove_member( + &f.pool, + f.community, + f.private, + f.peer.public_key().as_bytes(), + f.owner.public_key().as_bytes(), + ) + .await + .unwrap(); + buzz_db::channel_members::add_member( + &f.pool, + f.community, + f.private, + f.peer.public_key().as_bytes(), + buzz_db::channel_members::MemberRole::Member, + Some(f.owner.public_key().as_bytes()), + ) + .await + .unwrap(); + let (status, delta) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":page["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {delta}"); + assert_eq!(delta["changes"][0]["event"]["id"], first.id.to_hex()); + let changed = f.revision_as(&f.owner, f.private, id, "update", Some(&first)); + accept(&f, &changed).await; + let (status, delta) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":delta["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {delta}"); + assert_eq!(delta["changes"][0]["event"]["id"], changed.id.to_hex()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_ignores_unchanged_visibility_and_resets_on_ttl_changes() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let permanent_id = Uuid::new_v4(); + let permanent = f.revision(permanent_id, "create", None); + accept(&f, &permanent).await; + let first = sync(&f, None).await; + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + visibility: Some("open".into()), + name: Some("renamed".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + let renamed = sync(&f, Some(&first["checkpoint"])).await; + assert!(renamed["checkpoint"].is_string()); + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.private, + buzz_db::channel::ChannelUpdate { + ttl_seconds: Some(Some(3600)), + ..Default::default() + }, + ) + .await + .unwrap(); + let (status, reset) = f + .request( + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":renamed["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(reset["reason"], "policy_changed"); + let next = sync(&f, None).await; + assert!(next["checkpoint"].is_string()); + let expiring_id = Uuid::new_v4(); + let expiring = f.revision_as(&f.owner, f.private, expiring_id, "create", None); + accept(&f, &expiring).await; + for checkpoint in [None, Some(&next["checkpoint"])] { + let page = sync(&f, checkpoint).await; + assert_eq!(page["changes"].as_array().unwrap().len(), 1); + assert_eq!(page["changes"][0]["artifactId"], permanent_id.to_string()); + assert_eq!(page["changes"][0]["type"], "upsert"); + assert!(page["checkpoint"].is_string()); + } + let (status, current) = f + .request( + "/query", + json!([{ + "artifact":"incremental","action":"current","query":query(), + "artifactIds":[permanent_id,expiring_id] + }]), + ) + .await; + assert!(status.is_success(), "{status}: {current}"); + assert_eq!(current["changes"][0]["type"], "upsert"); + assert_eq!( + current["changes"][1], + json!({"type":"remove","artifactId":expiring_id}) + ); + let (status, named) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":{"artifact":"current","#h":[f.private]} + }]), + ) + .await; + assert!(status.is_success(), "{status}: {named}"); + assert_eq!(named["changes"], json!([])); + let moved = f.revision_as(&f.owner, f.private, permanent_id, "move", Some(&permanent)); + accept(&f, &moved).await; + let delta = sync(&f, Some(&next["checkpoint"])).await; + assert_eq!( + delta["changes"], + json!([{"type":"remove","artifactId":permanent_id}]) + ); + let (status, scoped) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":{"artifact":"current","#h":[f.home]}, + }]), + ) + .await; + assert!(status.is_success(), "{status}: {scoped}"); + assert_eq!(scoped["changes"], json!([])); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_shrinks_large_pages_and_current_batches_fail_explicitly() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let mut ids = Vec::new(); + for _ in 0..40 { + let id = Uuid::new_v4(); + let base = f.revision(id, "create", None); + let event = EventBuilder::new(Kind::Custom(45010), "x".repeat(260_000)) + .tags(base.tags.clone()) + .sign_with_keys(&f.owner) + .unwrap(); + accept(&f, &event).await; + ids.push(id); + } + let first = sync(&f, None).await; + assert_eq!(first["changes"].as_array().unwrap().len(), 20); + assert!(first["nextCursor"].is_string()); + let (status, last) = f + .request( + "/query", + json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {last}"); + assert_eq!(last["changes"].as_array().unwrap().len(), 20); + assert!(last["checkpoint"].is_string()); + let (status, body) = f + .request( + "/query", + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":ids}]), + ) + .await; + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!(body["code"], "incremental_limit"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_gate_controls_http_reads_and_capability() { + use tower::ServiceExt; + let _guard = SERIAL.lock().await; + let mut f = fixture().await; + let mut websocket_sync = Value::Null; + for enabled in [false, true] { + Arc::make_mut(&mut Arc::make_mut(&mut f.state).config).artifact_incremental_enabled = + enabled; + let response = crate::router::build_router(f.state.clone()) + .oneshot( + axum::http::Request::builder() + .uri("/info") + .header("host", &f.host) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert!(response.status().is_success()); + let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let info: Value = serde_json::from_slice(&bytes).unwrap(); + if enabled { + assert_eq!(info["artifacts"]["sync"], websocket_sync); + assert_eq!( + info["artifacts"]["incremental"], + json!({"version":1,"overlapSeconds":120, + "pageRows":100,"currentRows":100,"responseBytes":8388608, + "permanentChannelsOnly":true,"checkpointLifetimeSeconds":2592000}) + ); + } else { + websocket_sync = info["artifacts"]["sync"].clone(); + assert!(info["artifacts"].get("incremental").is_none()); + } + let (status, _) = f + .request( + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + assert_eq!( + status, + if enabled { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + } + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn incremental_resets_when_a_move_source_loses_access() { + let _guard = SERIAL.lock().await; + for delete_channel in [false, true] { + let f = fixture().await; + let id = Uuid::new_v4(); + let first = f.revision(id, "create", None); + accept(&f, &first).await; + let (_, read) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&first)); + accept(&f, &moved).await; + let heads: i64 = sqlx::query_scalar( + "SELECT count(*) FROM artifact_heads WHERE community_id=$1 AND channel_id=$2", + ) + .bind(f.community.as_uuid()) + .bind(f.home) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(heads, 0); + if delete_channel { + assert!( + buzz_db::channel::soft_delete_channel(&f.pool, f.community, f.home) + .await + .unwrap() + ); + } else { + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + } + let (status, delta) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":read["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(delta["reason"], "policy_changed"); + assert!(!delta.to_string().contains(&f.home.to_string())); + let (status, current) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id]}]), + ) + .await; + assert!(status.is_success()); + assert_eq!( + current["changes"], + json!([{"type":"remove","artifactId":id}]) + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_policy_update_can_pass_a_channel_key_share() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let mut tx = f.pool.begin().await.unwrap(); + sqlx::query("SELECT id FROM channels WHERE community_id=$1 AND id=$2 FOR KEY SHARE") + .bind(f.community.as_uuid()) + .bind(f.home) + .fetch_one(&mut *tx) + .await + .unwrap(); + tokio::time::timeout( + std::time::Duration::from_secs(2), + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ), + ) + .await + .expect("policy update must permit a foreign-key reader") + .unwrap(); + tx.rollback().await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_progress_replays_queries_with_escaped_values() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let scope = json!({"artifact":"current","#type":["buzz.task"],"#project":vec!["\u{1}".repeat(4096);15]}); + let (status, first) = f + .request("/query", json!([{"artifact":"incremental","query":scope}])) + .await; + assert!(status.is_success(), "{status}: {first}"); + assert!(first["checkpoint"].as_str().unwrap().len() < 200_000); + let (status, next) = f + .request( + "/query", + json!([{"artifact":"incremental","query":scope,"checkpoint":first["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {next}"); + assert_eq!(next["changes"], json!([])); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_current_intersects_known_ids_with_query_scope() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let included = Uuid::new_v4(); + let excluded = Uuid::new_v4(); + let event = f.revision(included, "create", None); + accept(&f, &event).await; + accept(&f, &f.revision(excluded, "create", None)).await; + let (status, body) = f + .request( + "/query", + json!([{"artifact":"incremental","action":"current", + "query":{"artifact":"current","#type":["buzz.task"],"#d":[included]}, + "artifactIds":[included,excluded]}]), + ) + .await; + assert!(status.is_success(), "{status}: {body}"); + assert_eq!(body["changes"][0]["artifactId"], included.to_string()); + assert_eq!(body["changes"][0]["event"]["id"], event.id.to_hex()); + assert_eq!( + body["changes"][1], + json!({"type":"remove","artifactId":excluded}) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn incremental_unlisted_channels_require_membership_or_named_scope() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let event = f.revision(id, "create", None); + accept(&f, &event).await; + let (_, before) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + assert_eq!(before["changes"][0]["event"]["id"], event.id.to_hex()); + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + unlisted: Some(true), + ..Default::default() + }, + ) + .await + .unwrap(); + let (status, reset) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":before["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(reset["reason"], "policy_changed"); + for (scope, visible) in [ + (query(), false), + ( + json!({"artifact":"current","#type":["buzz.task"],"#h":[f.home]}), + true, + ), + ] { + let (status, page) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":scope}]), + ) + .await; + assert!(status.is_success(), "{page}"); + assert_eq!( + page["changes"].as_array().unwrap().len(), + usize::from(visible) + ); + let (status, current) = f.request_as(&f.peer, "/query", json!([{"artifact":"incremental","action":"current","query":scope,"artifactIds":[id]}])).await; + assert!(status.is_success(), "{current}"); + assert_eq!( + current["changes"][0]["type"], + if visible { "upsert" } else { "remove" } + ); + } + assert_eq!( + sync(&f, None).await["changes"][0]["event"]["id"], + event.id.to_hex() + ); + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + unlisted: Some(false), + ..Default::default() + }, + ) + .await + .unwrap(); + let (_, listed) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + assert_eq!(listed["changes"][0]["event"]["id"], event.id.to_hex()); +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs new file mode 100644 index 00000000000..eab9488eb2a --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs @@ -0,0 +1,231 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_move_cursor_hides_private_and_redacted_revision_ids() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let (_, empty) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + let mut current = std::collections::HashMap::new(); + let mut private_ids = Vec::new(); + for _ in 0..101 { + let id = Uuid::new_v4(); + let created = f.revision(id, "create", None); + accept(&f, &created).await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&created)); + accept(&f, &moved).await; + private_ids.push(moved.id.to_hex()); + current.insert(id, moved); + } + let (&id, previous) = current.iter().next().unwrap(); + let back = f.revision(id, "move", Some(previous)); + accept(&f, &back).await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&back)); + accept(&f, &moved).await; + private_ids.push(moved.id.to_hex()); + current.insert(id, moved); + let stamp: chrono::DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(&f.pool) + .await + .unwrap(); + sqlx::query("UPDATE artifact_revisions SET modified_at=$2 WHERE community_id=$1") + .bind(f.community.as_uuid()) + .bind(stamp) + .execute(&f.pool) + .await + .unwrap(); + for redacted in [false, true] { + let (status, first) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":empty["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {first}"); + let cursor = first["nextCursor"].as_str().unwrap(); + let payload = cursor.split_once('.').unwrap().0; + let decoded = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload) + .unwrap(); + let decoded = String::from_utf8(decoded).unwrap(); + for private_id in &private_ids { + assert!( + !decoded.contains(private_id), + "cursor exposes a private revision ID (redacted={redacted})" + ); + } + assert_eq!(first["changes"].as_array().unwrap().len(), 100); + let (status, last) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"cursor":cursor}]), + ) + .await; + assert!(status.is_success(), "{status}: {last}"); + assert!(last["checkpoint"].is_string()); + let mut ids = std::collections::HashSet::new(); + for change in first["changes"] + .as_array() + .unwrap() + .iter() + .chain(last["changes"].as_array().unwrap()) + { + assert_eq!(change["type"], "remove"); + assert!(ids.insert(change["artifactId"].as_str().unwrap().to_owned())); + } + assert_eq!(ids.len(), 101); + if !redacted { + let boundary = first["changes"].as_array().unwrap().last().unwrap()["artifactId"] + .as_str() + .unwrap() + .parse::() + .unwrap(); + assert!(buzz_db::event::soft_delete_event_and_update_thread( + &f.pool, + f.community, + current[&boundary].id.as_bytes(), + None, + None + ) + .await + .unwrap()); + } + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_channel_scope_limits_candidates_and_keeps_source_removals() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + for _ in 0..101 { + accept( + &f, + &f.revision_as(&f.owner, f.private, Uuid::new_v4(), "create", None), + ) + .await; + } + let deleted_id = Uuid::new_v4(); + let deleted = f.revision(deleted_id, "create", None); + accept(&f, &deleted).await; + accept(&f, &f.revision(deleted_id, "delete", Some(&deleted))).await; + let id = Uuid::new_v4(); + let created = f.revision(id, "create", None); + accept(&f, &created).await; + let scoped_query = json!({"artifact":"current","#type":["buzz.task"],"#h":[f.home]}); + let (status, initial) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":scoped_query + }]), + ) + .await; + assert!(status.is_success(), "{status}: {initial}"); + assert_eq!(initial["changes"].as_array().unwrap().len(), 1); + assert_eq!(initial["changes"][0]["event"]["id"], created.id.to_hex()); + assert!(initial["nextCursor"].is_null()); + assert!(initial["checkpoint"].is_string()); + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&created)); + accept(&f, &moved).await; + let (status, delta) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":scoped_query,"checkpoint":initial["checkpoint"] + }]), + ) + .await; + assert!(status.is_success(), "{status}: {delta}"); + assert!(delta["changes"] + .as_array() + .unwrap() + .contains(&json!({"type":"remove","artifactId":id}))); + assert!(!delta.to_string().contains(&f.private.to_string())); + let restored = f.revision(id, "move", Some(&moved)); + accept(&f, &restored).await; + for channel in [f.home, f.private] { + sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role) VALUES($1,$2,$3,'owner')") + .bind(f.community.as_uuid()).bind(channel).bind(f.peer.public_key().as_bytes().as_slice()) + .execute(&f.pool).await.unwrap(); + } + buzz_db::channel_members::remove_member( + &f.pool, + f.community, + f.private, + f.owner.public_key().as_bytes(), + f.owner.public_key().as_bytes(), + ) + .await + .unwrap(); + let (status, returned) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":scoped_query,"checkpoint":delta["checkpoint"] + }]), + ) + .await; + assert!(status.is_success(), "{status}: {returned}"); + assert!(returned["changes"] + .as_array() + .unwrap() + .iter() + .any(|c| c["event"]["id"] == restored.id.to_hex())); + assert!(returned["changes"] + .as_array() + .unwrap() + .iter() + .all(|c| c["type"] != "removeChannel")); + buzz_db::channel_members::remove_member( + &f.pool, + f.community, + f.home, + f.owner.public_key().as_bytes(), + f.owner.public_key().as_bytes(), + ) + .await + .unwrap(); + let (status, removed) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":scoped_query,"checkpoint":returned["checkpoint"] + }]), + ) + .await; + assert!(status.is_success(), "{status}: {removed}"); + assert_eq!( + removed["changes"], + json!([{"type":"removeChannel","channelId":f.home}]) + ); + let (status, invalid_scope) = f + .request( + "/query", + json!([{ + "artifact":"incremental","query":{"artifact":"current","#h":["not-a-uuid"]} + }]), + ) + .await; + assert!(status.is_success(), "{status}: {invalid_scope}"); + assert_eq!(invalid_scope["changes"], json!([])); +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs new file mode 100644 index 00000000000..c3cc3d9f1d4 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs @@ -0,0 +1,79 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn incremental_resets_for_first_artifact_during_channel_transition() { + let _guard = SERIAL.lock().await; + for delete_channel in [false, true] { + let f = fixture().await; + let mut blocker = f.pool.begin().await.unwrap(); + let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *blocker) + .await + .unwrap(); + sqlx::query("SELECT id FROM communities WHERE id=$1 FOR NO KEY UPDATE") + .bind(f.community.as_uuid()) + .fetch_one(&mut *blocker) + .await + .unwrap(); + let pool = f.pool.clone(); + let community = f.community; + let home = f.home; + let transition = tokio::spawn(async move { + if delete_channel { + assert!( + buzz_db::channel::soft_delete_channel(&pool, community, home) + .await + .unwrap() + ); + } else { + buzz_db::channel::update_channel( + &pool, + community, + home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + } + }); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + assert!(!transition.is_finished(), "an empty channel must still advance the recovery marker"); + let blocked: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE $1=ANY(pg_blocking_pids(pid)) AND query LIKE 'UPDATE communities SET artifact_policy_modified_at=%')") + .bind(blocker_pid).fetch_one(&f.pool).await.unwrap(); + if blocked { break; } + tokio::task::yield_now().await; + } + }).await.expect("channel transition reaches the recovery marker"); + let id = Uuid::new_v4(); + let event = f.revision(id, "create", None); + tokio::time::timeout(std::time::Duration::from_secs(2), accept(&f, &event)) + .await + .expect("first artifact commits while the channel transition is uncommitted"); + let (status, first) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + assert!(status.is_success(), "{status}: {first}"); + assert_eq!(first["changes"][0]["event"]["id"], event.id.to_hex()); + blocker.commit().await.unwrap(); + transition.await.unwrap(); + let (status, next) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(next["reason"], "policy_changed"); + assert!(!next.to_string().contains(&f.home.to_string())); + } +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs new file mode 100644 index 00000000000..a5025b8dca3 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs @@ -0,0 +1,337 @@ +use super::*; +use axum::http::StatusCode; +#[path = "sync_access_postgres_tests.rs"] +mod access_postgres_tests; +#[path = "sync_cursor_postgres_tests.rs"] +mod cursor_postgres_tests; +#[path = "sync_policy_postgres_tests.rs"] +mod policy_postgres_tests; +#[path = "sync_scale_postgres_tests.rs"] +mod scale_postgres_tests; + +async fn fixture() -> Fixture { + let mut f = Fixture::new().await; + let mut state = (*f.state).clone(); + Arc::make_mut(&mut state.config).artifact_incremental_enabled = true; + f.state = Arc::new(state); + f +} +static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +fn query() -> Value { + json!({"artifact":"current","#type":["buzz.task"]}) +} +async fn sync(f: &Fixture, checkpoint: Option<&Value>) -> Value { + let mut request = json!({"artifact":"incremental","query":query()}); + if let Some(checkpoint) = checkpoint { + request["checkpoint"] = checkpoint.clone(); + } + let (status, body) = f.request("/query", json!([request])).await; + assert!(status.is_success(), "{status}: {body}"); + body +} +async fn accept(f: &Fixture, event: &Event) { + let envelope = buzz_core::artifact::validate(event).unwrap(); + let source = if envelope.op == buzz_core::artifact::ArtifactOp::Move { + f.state + .db + .artifact_home(f.community, envelope.id) + .await + .unwrap() + } else { + None + }; + assert!(matches!( + f.state + .db + .accept_artifact( + f.community, + event, + &envelope, + source, + &f.state.relay_keypair + ) + .await + .unwrap(), + buzz_db::artifact::ArtifactOutcome::Accepted(_) + )); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_recovers_late_commit_and_returns_explicit_deletion() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let first = sync(&f, None).await; + assert_eq!(first["changes"], json!([])); + let id = Uuid::new_v4(); + let event = f.revision(id, "create", None); + accept(&f, &event).await; + // Model a stamp assigned before the first read, followed by a later commit. + sqlx::query("UPDATE artifact_heads SET modified_at=$1::timestamptz-interval '119 seconds' WHERE community_id=$2 AND artifact_id=$3") + .bind(first["startedAt"].as_str().unwrap()).bind(f.community.as_uuid()).bind(id).execute(&f.pool).await.unwrap(); + let second = sync(&f, Some(&first["checkpoint"])).await; + assert!(second["changes"] + .as_array() + .unwrap() + .iter() + .any(|c| c["event"]["id"] == event.id.to_hex())); + let deleted = f.revision(id, "delete", Some(&event)); + accept(&f, &deleted).await; + let third = sync(&f, Some(&second["checkpoint"])).await; + assert!(third["changes"] + .as_array() + .unwrap() + .iter() + .any(|c| c == &json!({"type":"remove","artifactId":id}))); + assert_eq!(third["overlapSeconds"], 120); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_pages_equal_timestamps_and_replays_the_same_cursor() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + for _ in 0..103 { + accept(&f, &f.revision(Uuid::new_v4(), "create", None)).await; + } + sqlx::query("UPDATE artifact_heads SET modified_at='2000-01-01' WHERE community_id=$1") + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .unwrap(); + let first = sync(&f, None).await; + assert_eq!(first["changes"].as_array().unwrap().len(), 100); + assert!(first["checkpoint"].is_null()); + let request = json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]); + let (status, last) = f.request("/query", request.clone()).await; + assert!(status.is_success(), "{status}: {last}"); + assert_eq!(last["changes"].as_array().unwrap().len(), 3); + assert!(last["nextCursor"].is_null()); + assert!(last["checkpoint"].is_string()); + let (_, replay) = f.request("/query", request).await; + assert_eq!(last, replay); + let ids = first["changes"] + .as_array() + .unwrap() + .iter() + .chain(last["changes"].as_array().unwrap()) + .map(|c| c["artifactId"].as_str().unwrap()) + .collect::>(); + assert_eq!(ids.len(), 103); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_current_has_uniform_unavailable_and_rejects_duplicate_ids() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let event = f.revision_as(&f.owner, f.private, id, "create", None); + accept(&f, &event).await; + let missing = Uuid::new_v4(); + let body = json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id,missing,"local-only"]}]); + let (status, body) = f.request_as(&f.peer, "/query", body).await; + assert!(status.is_success(), "{status}: {body}"); + assert_eq!( + body["changes"], + json!([ + {"type":"remove","artifactId":id},{"type":"remove","artifactId":missing},{"type":"remove","artifactId":"local-only"} + ]) + ); + let (status, _) = f + .request( + "/query", + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id,id]}]), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + let (status, body) = f + .request( + "/query", + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id]}]), + ) + .await; + assert!(status.is_success(), "{status}: {body}"); + assert_eq!(body["changes"][0]["event"]["id"], event.id.to_hex()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_grant_finds_old_heads_and_removal_names_only_own_membership() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let event = f.revision_as(&f.owner, f.private, id, "create", None); + accept(&f, &event).await; + sqlx::query("UPDATE artifact_heads SET modified_at='2000-01-01' WHERE community_id=$1") + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .unwrap(); + let (_, first) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + assert_eq!(first["changes"], json!([])); + buzz_db::channel_members::add_member( + &f.pool, + f.community, + f.private, + f.peer.public_key().as_bytes(), + buzz_db::channel_members::MemberRole::Member, + Some(f.owner.public_key().as_bytes()), + ) + .await + .unwrap(); + let (_, granted) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), + ) + .await; + assert!( + granted["changes"] + .as_array() + .unwrap() + .iter() + .any(|c| c["event"]["id"] == event.id.to_hex()), + "{granted}" + ); + buzz_db::channel_members::remove_member( + &f.pool, + f.community, + f.private, + f.peer.public_key().as_bytes(), + f.owner.public_key().as_bytes(), + ) + .await + .unwrap(); + let (_, removed) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":granted["checkpoint"]}]), + ) + .await; + assert_eq!( + removed["changes"], + json!([{ "type":"removeChannel","channelId":f.private }]) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_move_to_hidden_destination_removes_the_source_view() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let id = Uuid::new_v4(); + let first = f.revision(id, "create", None); + accept(&f, &first).await; + let (_, read) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) + .await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&first)); + accept(&f, &moved).await; + let (_, delta) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":read["checkpoint"]}]), + ) + .await; + assert_eq!( + delta["changes"], + json!([{ "type":"remove","artifactId":id }]) + ); + assert!(!delta.to_string().contains(&f.private.to_string())); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_policy_changes_reset_without_hidden_channel_ids() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + for _ in 0..101 { + accept(&f, &f.revision(Uuid::new_v4(), "create", None)).await; + } + let first = sync(&f, None).await; + assert!(first["nextCursor"].is_string()); + buzz_db::channel::update_channel( + &f.pool, + f.community, + f.home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + let (status, body) = f + .request( + "/query", + json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!( + body, + json!({"code":"incremental_reset","error":"incremental_reset","reason":"policy_changed"}) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_progress_binds_reader_query_and_community() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let first = sync(&f, None).await; + for (reader, scope) in [ + (&f.peer, query()), + ( + &f.owner, + json!({"artifact":"current","#type":["other.task"]}), + ), + ] { + let (status, _) = f + .request_as( + reader, + "/query", + json!([{"artifact":"incremental","query":scope,"checkpoint":first["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + } + let mut other = fixture().await; + Arc::make_mut(&mut other.state).relay_keypair = f.state.relay_keypair.clone(); + let (status, _) = other + .request_as( + &f.owner, + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + let expired = crate::api::artifact_sync::expired_checkpoint_for_test( + &f.state, + first["checkpoint"].as_str().unwrap(), + ) + .unwrap(); + let (status, body) = f + .request( + "/query", + json!([{"artifact":"incremental","query":query(),"checkpoint":expired}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(body["reason"], "checkpoint_expired"); +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs new file mode 100644 index 00000000000..cf23203eaab --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs @@ -0,0 +1,132 @@ +use super::*; +use base64::Engine; +use sha2::Digest; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_downloads_ten_thousand_tasks_with_sixty_thousand_memberships() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let homes = (0..60_001) + .map(|n| if n == 0 { f.home } else { Uuid::new_v4() }) + .collect::>(); + for chunk in homes[1..].chunks(500) { + sqlx::query("INSERT INTO channels(community_id,id,name,visibility,channel_type,created_by,archived_at) SELECT $1,id,id::text,'private','stream',$2,'2000-01-01' FROM unnest($3::uuid[]) id") + .bind(f.community.as_uuid()).bind(f.owner.public_key().as_bytes().as_slice()).bind(chunk).execute(&f.pool).await.unwrap(); + sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role,sync_modified_at) SELECT $1,id,$2,'owner','2000-01-01' FROM unnest($3::uuid[]) id") + .bind(f.community.as_uuid()).bind(f.owner.public_key().as_bytes().as_slice()).bind(chunk).execute(&f.pool).await.unwrap(); + } + sqlx::query("UPDATE channel_members SET sync_modified_at='2000-01-01' WHERE community_id=$1 AND channel_id=ANY($2)") + .bind(f.community.as_uuid()).bind(vec![f.home,f.private]).execute(&f.pool).await.unwrap(); + sqlx::query("UPDATE channels SET archived_at=NULL WHERE community_id=$1 AND id=ANY($2)") + .bind(f.community.as_uuid()) + .bind(&homes[..10_000]) + .execute(&f.pool) + .await + .unwrap(); + let mut event_bytes = 0; + for chunk in homes[..10_050].chunks(500) { + let values=chunk.iter().map(|home| { + let id=Uuid::new_v4(); + let base=f.revision_as(&f.owner,*home,id,"create",None); + let event=EventBuilder::new(Kind::Custom(45010),json!({"notes":"x".repeat(4400)}).to_string()) + .tags(base.tags.clone()).sign_with_keys(&f.owner).unwrap(); + event_bytes=serde_json::to_vec(&event).unwrap().len(); + json!({"id":event.id.to_hex(),"artifact_id":id,"tags":event.tags,"content":event.content,"sig":event.sig.to_string(),"created_at":event.created_at.as_secs(),"channel_id":home}) + }).collect::>(); + sqlx::query("INSERT INTO events(community_id,id,pubkey,created_at,kind,tags,content,sig,channel_id) SELECT $1,decode(x.id,'hex'),$2,to_timestamp(x.created_at),45010,x.tags,x.content,decode(x.sig,'hex'),x.channel_id FROM jsonb_to_recordset($3) x(id text,tags jsonb,content text,sig text,created_at bigint,channel_id uuid)") + .bind(f.community.as_uuid()).bind(f.owner.public_key().as_bytes().as_slice()).bind(json!(values)).execute(&f.pool).await.unwrap(); + sqlx::query("INSERT INTO artifact_heads(community_id,artifact_id,event_id,channel_id,artifact_type,modified_at) SELECT $1,x.artifact_id,decode(x.id,'hex'),x.channel_id,'buzz.task','2000-01-01' FROM jsonb_to_recordset($2) x(id text,artifact_id uuid,channel_id uuid)") + .bind(f.community.as_uuid()).bind(json!(values)).execute(&f.pool).await.unwrap(); + } + sqlx::raw_sql("ANALYZE artifact_heads; ANALYZE channel_members; ANALYZE channels") + .execute(&f.pool) + .await + .unwrap(); + let mut other = (*bridge_handler_test_state().await.unwrap()).clone(); + let config = Arc::make_mut(&mut other.config); + config.require_auth_token = true; + config.artifact_incremental_enabled = true; + other.relay_keypair = f.state.relay_keypair.clone(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/query", listener.local_addr().unwrap()); + let router = crate::router::build_router(Arc::new(other)); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .unwrap(); + let started = std::time::Instant::now(); + let mut page = sync(&f, None).await; + let mut ids = std::collections::HashSet::new(); + let mut pages = 0; + loop { + pages += 1; + for change in page["changes"].as_array().unwrap() { + assert_eq!(change["type"], "upsert"); + ids.insert(change["artifactId"].as_str().unwrap().to_owned()); + } + if page["nextCursor"].is_null() { + break; + } + let (status, next) = signed_query( + &f, + &client, + &url, + json!([{"artifact":"incremental","query":query(),"cursor":page["nextCursor"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {next}"); + page = next; + } + let cold = started.elapsed(); + assert_eq!(ids.len(), 10_050); + assert_eq!(pages, 101); + let warm = std::time::Instant::now(); + let (status, delta) = signed_query( + &f, + &client, + &url, + json!([{"artifact":"incremental","query":query(),"checkpoint":page["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {delta}"); + assert_eq!(delta["changes"], json!([])); + assert!(delta["checkpoint"].is_string()); + eprintln!("10,050 tasks; 60,002 memberships; {event_bytes}-byte events; {pages} cold pages: {cold:?}; empty warm sync: {:?}",warm.elapsed()); + server.abort(); +} + +async fn signed_query( + f: &Fixture, + client: &reqwest::Client, + url: &str, + body: Value, +) -> (StatusCode, Value) { + let bytes = serde_json::to_vec(&body).unwrap(); + let auth = EventBuilder::new(Kind::Custom(27235), "") + .tags([ + Tag::parse(["u", &format!("https://{}/query", f.host)]).unwrap(), + Tag::parse(["method", "POST"]).unwrap(), + Tag::parse(["payload", &hex::encode(sha2::Sha256::digest(&bytes))]).unwrap(), + Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(), + ]) + .sign_with_keys(&f.owner) + .unwrap(); + let authorization = format!( + "Nostr {}", + base64::engine::general_purpose::STANDARD.encode(serde_json::to_vec(&auth).unwrap()) + ); + let response = client + .post(url) + .header("host", &f.host) + .header("content-type", "application/json") + .header("authorization", authorization) + .body(bytes) + .send() + .await + .unwrap(); + let status = StatusCode::from_u16(response.status().as_u16()).unwrap(); + let body = serde_json::from_slice(&response.bytes().await.unwrap()).unwrap(); + (status, body) +} diff --git a/crates/buzz-relay/src/api/artifact_sync.rs b/crates/buzz-relay/src/api/artifact_sync.rs new file mode 100644 index 00000000000..6ec686d3ff2 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_sync.rs @@ -0,0 +1,392 @@ +//! Incremental reads with signed progress and bounded current-identity reads. +use crate::state::AppState; +use axum::{http::StatusCode, Json}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; +use buzz_core::{ + artifact::{ArtifactQuery, ArtifactView}, + TenantContext, +}; +use buzz_db::artifact_sync::{Candidate, Position, PAGE_ROWS}; +use chrono::{DateTime, Duration, Utc}; +use hmac::{Hmac, KeyInit, Mac}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; +use uuid::Uuid; + +const OVERLAP: i64 = 120; +const CHECKPOINT_DAYS: i64 = 30; +const RESPONSE_BYTES: usize = 8_388_608; +static WORK: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4); +type Response = Result, (StatusCode, Json)>; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct Request { + artifact: String, + query: Value, + action: Option, + checkpoint: Option, + cursor: Option, + artifact_ids: Option>, +} + +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct Token { + version: u8, + community: Uuid, + reader: String, + query: String, + at: DateTime, + policy: DateTime, + since: Option>, + position: Option, + cursor: bool, +} + +fn error(status: StatusCode, code: &str) -> (StatusCode, Json) { + (status, Json(json!({"error":code,"code":code}))) +} +fn invalid() -> (StatusCode, Json) { + error(StatusCode::BAD_REQUEST, "incremental_invalid") +} +fn unavailable() -> (StatusCode, Json) { + error(StatusCode::SERVICE_UNAVAILABLE, "incremental_unavailable") +} +fn limit() -> (StatusCode, Json) { + error(StatusCode::PAYLOAD_TOO_LARGE, "incremental_limit") +} +fn reset(reason: &str) -> (StatusCode, Json) { + ( + StatusCode::CONFLICT, + Json(json!({"error":"incremental_reset","code":"incremental_reset","reason":reason})), + ) +} +fn db_error(e: buzz_db::DbError) -> (StatusCode, Json) { + match e { + buzz_db::DbError::ArtifactSyncLimit => limit(), + error => { + tracing::warn!(%error,"incremental artifact read failed"); + unavailable() + } + } +} + +/// Bounds advertised to clients that use the incremental protocol. +pub(crate) fn capability() -> Value { + json!({"version":1,"overlapSeconds":OVERLAP,"pageRows":PAGE_ROWS, + "currentRows":PAGE_ROWS,"responseBytes":RESPONSE_BYTES,"permanentChannelsOnly":true, + "checkpointLifetimeSeconds":CHECKPOINT_DAYS*86400}) +} + +fn mac(state: &AppState) -> Result, (StatusCode, Json)> { + let mut digest = Sha256::new(); + digest.update(b"buzz-artifact-incremental-v1\0"); + digest.update(state.relay_keypair.secret_key().as_secret_bytes()); + Hmac::::new_from_slice(&digest.finalize()).map_err(|_| unavailable()) +} +fn encode(state: &AppState, token: &Token) -> Result)> { + let data = serde_json::to_vec(token).map_err(|_| unavailable())?; + let mut signature = mac(state)?; + signature.update(&data); + let encoded = format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(&data), + URL_SAFE_NO_PAD.encode(signature.finalize().into_bytes()) + ); + if encoded.len() > 200_000 { + return Err(invalid()); + } + Ok(encoded) +} +fn decode(state: &AppState, value: &str) -> Result)> { + if value.len() > 200_000 { + return Err(invalid()); + } + let (data, signature) = value.split_once('.').ok_or_else(invalid)?; + let data = URL_SAFE_NO_PAD.decode(data).map_err(|_| invalid())?; + let signature = URL_SAFE_NO_PAD.decode(signature).map_err(|_| invalid())?; + let mut check = mac(state)?; + check.update(&data); + check.verify_slice(&signature).map_err(|_| invalid())?; + serde_json::from_slice(&data).map_err(|_| invalid()) +} + +/// Execute one incremental read without a manifest or server-side cursor state. +#[tracing::instrument(skip_all, fields(community_id = %tenant.community(), reader = %reader))] +pub(super) async fn query( + state: &Arc, + tenant: &TenantContext, + reader: &nostr::PublicKey, + filter: &Value, +) -> Response { + if !state.config.artifact_incremental_enabled { + return Err(unavailable()); + } + let _permit = WORK.try_acquire().map_err(|_| unavailable())?; + tokio::time::timeout( + std::time::Duration::from_secs(8), + execute(state, tenant, reader, filter), + ) + .await + .map_err(|_| unavailable())? +} + +async fn execute( + state: &Arc, + tenant: &TenantContext, + reader: &nostr::PublicKey, + filter: &Value, +) -> Response { + let request: Request = serde_json::from_value(filter.clone()).map_err(|_| invalid())?; + if request.artifact != "incremental" + || request.query.get("limit").is_some() + || request.query.get("offset").is_some() + { + return Err(invalid()); + } + let parsed = buzz_core::artifact::parse_query(&request.query).map_err(|_| invalid())?; + let query_binding = URL_SAFE_NO_PAD.encode(Sha256::digest( + serde_json::to_vec(&request.query).map_err(|_| invalid())?, + )); + if parsed.view != ArtifactView::Current { + return Err(invalid()); + } + let (now, policy) = state + .db + .artifact_sync_boundary(tenant.community()) + .await + .map_err(db_error)?; + if request.action.as_deref() == Some("current") { + if request.checkpoint.is_some() || request.cursor.is_some() { + return Err(invalid()); + } + let ids = request.artifact_ids.ok_or_else(invalid)?; + if ids.is_empty() + || ids.len() > PAGE_ROWS + || ids.iter().any(|id| id.is_empty() || id.len() > 256) + || ids.iter().collect::>().len() != ids.len() + { + return Err(invalid()); + } + let valid = ids + .iter() + .filter_map(|id| buzz_core::artifact::canonical_uuid(id).ok()) + .collect::>(); + let resolved = current(state, tenant, reader, &parsed, &valid).await?; + let resolved = resolved + .into_iter() + .filter_map(|value| { + value + .get("artifactId") + .and_then(Value::as_str) + .map(|id| (id.to_owned(), value.clone())) + }) + .collect::>(); + let changes = ids + .iter() + .map(|id| { + resolved + .get(id) + .cloned() + .unwrap_or_else(|| json!({"type":"remove","artifactId":id})) + }) + .collect::>(); + let response = json!({"version":1,"startedAt":now,"changes":changes}); + if serde_json::to_vec(&response) + .map_err(|_| unavailable())? + .len() + > RESPONSE_BYTES + { + return Err(limit()); + } + return Ok(Json(response)); + } + if request.action.is_some() + || request.artifact_ids.is_some() + || (request.checkpoint.is_some() && request.cursor.is_some()) + { + return Err(invalid()); + } + let encoded = request.cursor.as_ref().or(request.checkpoint.as_ref()); + let mut token = if let Some(encoded) = encoded { + let previous = decode(state, encoded)?; + if previous.version != 1 + || previous.community != *tenant.community().as_uuid() + || previous.reader != reader.to_hex() + || previous.query != query_binding + || previous.cursor != request.cursor.is_some() + || previous.at > now + { + return Err(invalid()); + } + if previous.at < now - Duration::days(CHECKPOINT_DAYS) { + return Err(reset("checkpoint_expired")); + } + if previous.policy != policy { + return Err(reset("policy_changed")); + } + if previous.cursor { + previous + } else { + Token { + since: Some(previous.at - Duration::seconds(OVERLAP)), + at: now, + position: None, + cursor: true, + ..previous + } + } + } else { + Token { + version: 1, + community: *tenant.community().as_uuid(), + reader: reader.to_hex(), + query: query_binding, + at: now, + policy, + since: None, + position: None, + cursor: true, + } + }; + let candidates = state + .db + .artifact_sync_candidates( + tenant.community(), + reader.as_bytes(), + token.since, + token.at, + token.position.as_ref(), + &parsed, + ) + .await + .map_err(db_error)?; + let mut selected = candidates.len().min(PAGE_ROWS); + let resolved = loop { + let ids = candidates[..selected] + .iter() + .filter_map(|c| c.artifact_id) + .collect::>() + .into_iter() + .collect::>(); + match current(state, tenant, reader, &parsed, &ids).await { + Err((StatusCode::PAYLOAD_TOO_LARGE, _)) if selected > 1 => selected /= 2, + result => break result?, + } + }; + let more = candidates.len() > selected; + let candidates = &candidates[..selected]; + let resolved = resolved + .into_iter() + .filter_map(|value| { + value + .get("artifactId") + .and_then(Value::as_str) + .map(|id| (id.to_owned(), value.clone())) + }) + .collect::>(); + let mut changes = Vec::new(); + let mut size = 0; + let mut processed = 0; + let mut emitted = HashSet::new(); + for candidate in candidates { + let change = change(candidate, &resolved)?; + let key = change.to_string(); + if !emitted.insert(key) { + token.position = Some(candidate.position.clone()); + processed += 1; + continue; + } + let bytes = serde_json::to_vec(&change) + .map_err(|_| unavailable())? + .len(); + // Leave room for the signed continuation and response fields. + if size + bytes > RESPONSE_BYTES - 262_144 { + if processed == 0 { + return Err(limit()); + } + break; + } + size += bytes; + changes.push(change); + token.position = Some(candidate.position.clone()); + processed += 1; + } + let (_, final_policy) = state + .db + .artifact_sync_boundary(tenant.community()) + .await + .map_err(db_error)?; + if final_policy != token.policy { + return Err(reset("policy_changed")); + } + let more = more || processed < candidates.len(); + let (cursor, checkpoint) = if more { + (Some(encode(state, &token)?), None) + } else { + token.cursor = false; + token.position = None; + token.since = None; + (None, Some(encode(state, &token)?)) + }; + Ok(Json( + json!({"version":1,"startedAt":token.at,"overlapSeconds":OVERLAP, + "changes":changes,"nextCursor":cursor,"checkpoint":checkpoint}), + )) +} + +async fn current( + state: &AppState, + tenant: &TenantContext, + reader: &nostr::PublicKey, + query: &ArtifactQuery, + ids: &[Uuid], +) -> Result, (StatusCode, Json)> { + let events = state + .db + .artifact_sync_current(tenant.community(), reader.as_bytes(), query, ids) + .await + .map_err(db_error)?; + let mut by_id = HashMap::new(); + for stored in events { + let env = buzz_core::artifact::validate(&stored.event).map_err(|_| unavailable())?; + by_id.insert(env.id, stored.event); + } + Ok(ids + .iter() + .map(|id| match by_id.remove(id) { + Some(event) => json!({"type":"upsert","artifactId":id,"event":event}), + None => json!({"type":"remove","artifactId":id}), + }) + .collect()) +} + +fn change( + candidate: &Candidate, + resolved: &HashMap, +) -> Result)> { + if let Some(channel) = candidate.removed_channel { + return Ok(json!({"type":"removeChannel","channelId":channel})); + } + let id = candidate.artifact_id.ok_or_else(unavailable)?; + resolved + .get(&id.to_string()) + .cloned() + .ok_or_else(unavailable) +} + +#[cfg(test)] +pub(super) fn expired_checkpoint_for_test( + state: &AppState, + value: &str, +) -> Result)> { + let mut token = decode(state, value)?; + token.at -= Duration::days(31); + encode(state, &token) +} diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index 19c9220e18a..129d7dd732c 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -869,3 +869,4 @@ mod parse_query_tests { } mod artifact; +pub(crate) mod artifact_sync; diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 79c439d8986..cbed06b641b 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -116,6 +116,8 @@ pub const MAX_DRAIN_JITTER_MS: u64 = 20_000; /// Relay runtime configuration, loaded from environment variables. #[derive(Debug, Clone)] pub struct Config { + /// Enable incremental artifact reads after every serving writer is upgraded. + pub artifact_incremental_enabled: bool, /// Address the relay HTTP/WebSocket server binds to. pub bind_addr: SocketAddr, /// Postgres database connection URL. @@ -797,6 +799,9 @@ impl Config { let require_relay_membership = std::env::var("BUZZ_REQUIRE_RELAY_MEMBERSHIP") .map(|v| v == "true" || v == "1") .unwrap_or(false); + let artifact_incremental_enabled = std::env::var("BUZZ_ARTIFACT_INCREMENTAL_ENABLED") + .map(|v| v == "true" || v == "1") + .unwrap_or(false); // Defaults true → single-pod (N=1) keeps today's huddle behavior. A // horizontally-scaled deployment sets this false; see the field doc. @@ -1452,6 +1457,7 @@ impl Config { partition_manager_advance_enabled, pubkey_allowlist_enabled, require_relay_membership, + artifact_incremental_enabled, huddle_audio_available, mesh, mesh_demo_echo, diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 390e73d0a3d..82e016c5dd9 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -391,6 +391,9 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st admin_api.as_deref(), state.config.klipy.as_ref().map(|_| "klipy"), ); + if state.config.artifact_incremental_enabled { + info.artifacts["incremental"] = crate::api::artifact_sync::capability(); + } if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { if state.config.buzz_v1_enabled { use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS}; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 80a4ab9001a..7b8b7174917 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -3372,8 +3372,16 @@ pub(crate) mod tests { #[ignore = "requires Postgres"] async fn a_lost_open_set_invalidation_grants_a_revoked_channel_only_until_max_age() { let database_url = crate::test_support::database_url(); - let writer_pod = test_state_with_database_url(&database_url).await; - let reader_pod = test_state_with_database_url(&database_url).await; + let writer_pod = test_state_with_database_url_and_pool_options( + &database_url, + sqlx::postgres::PgPoolOptions::new(), + ) + .await; + let reader_pod = test_state_with_database_url_and_pool_options( + &database_url, + sqlx::postgres::PgPoolOptions::new(), + ) + .await; let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); let community = writer_pod .db diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md new file mode 100644 index 00000000000..6af589b2bd5 --- /dev/null +++ b/docs/artifact-sync-deployment.md @@ -0,0 +1,70 @@ +# Incremental artifact reads deployment + +Incremental artifact reads use a 120-second overlap and database modification +timestamps. `BUZZ_ARTIFACT_INCREMENTAL_ENABLED` is off by default. Enable it only +after the schema, indexes, and every serving writer are upgraded. + +## Schema and indexes + +Migration `0077` adds columns with constant defaults, then changes the defaults +for future inserts. It does not rewrite existing rows or run a backfill. +Old rows remain at the epoch timestamp. An initial read includes them. +New application writers stamp every head change, grant, rejoin, removal, +redaction, and move source. Changes to channel listing also require recovery. Old application updates do not maintain all these +stamps, so mixed-version serving cannot advertise incremental correctness. + +For a populated database, first build the revision lookup index outside a +transaction, before migration `0076`: + +```sql +CREATE INDEX CONCURRENTLY idx_artifact_revisions_artifact + ON public.artifact_revisions (community_id, artifact_id); +``` + +Apply `0076_artifact_revision_lookup.sql` in a transaction after that prebuild. +Then apply `0077_artifact_sync_timestamps.sql` in a +transaction with its lock and statement budgets. The migration is idempotent +for this prebuild step. Then build these indexes outside a transaction: + +```sql +CREATE INDEX CONCURRENTLY artifact_heads_modified + ON public.artifact_heads (community_id, modified_at, artifact_id); +CREATE INDEX CONCURRENTLY artifact_heads_channel + ON public.artifact_heads (community_id, channel_id, artifact_id); +CREATE INDEX CONCURRENTLY channel_members_sync + ON public.channel_members (community_id, pubkey, sync_modified_at, channel_id); +CREATE INDEX CONCURRENTLY artifact_revisions_moves + ON public.artifact_revisions (community_id, modified_at, event_id) + WHERE source_channel_id IS NOT NULL; +``` + +Migration `0078` verifies each index definition and its valid, ready, and live +flags. It refuses a populated table whose index is missing. A fresh empty +database can build the indexes under the migration's short budgets. +Migration `0076` also verifies the exact definition and all three flags. +Remove an invalid index with `DROP INDEX CONCURRENTLY` before a new build. + +After all index builds succeed, run `buzz-admin migrate` with the target +`DATABASE_URL`. It verifies and records `0076`, reapplies the idempotent `0077`, +then verifies and records `0078`. Direct SQL application does not record the +migration ledger. + +## Enable and rollback + +Deploy the timestamp writers to every relay instance with the gate off. +Drain the old writers. All instances must use the same configured relay +signing key, because that key authenticates the progress tokens. Then enable +`BUZZ_ARTIFACT_INCREMENTAL_ENABLED=true` on the serving fleet. NIP-11 advertises +`artifacts.incremental` only when the gate is enabled. The endpoint refuses incremental +requests while the gate is off. + +Before a rollback to an old writer, disable the gate on every serving +instance. Before enabling it again, upgrade every writer and advance +`communities.artifact_policy_modified_at` for each affected active community +in a reviewed maintenance transaction under community deletion admission. +This forces clients to discover current records and revalidate saved IDs. +No local absence scan is permitted. + +The 120-second stamp-to-visibility bound remains an operating assumption. +Runtime measurements are tracked separately. This PR does not authorize +a deployment or a production maintenance write. diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index 3f0a8874996..ff0d62c36b4 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -106,3 +106,141 @@ Artifact revisions are state changes, not conversation messages. Receiving one M Clients validate content before interpreting it as a known type. Unknown types or malformed content render a safe fallback using the title, type, and home link. Editors MUST preserve unfamiliar fields and annotation tags or decline the edit. References may be unavailable or cyclic; clients must handle them safely without assuming relay validation. Artifact text is untrusted. A type name, relationship tag, or payload cannot authorize a privileged action. Moving, deleting, or revoking access cannot erase copies already downloaded. + +## Incremental reads + +Incremental reads maintain an eventually consistent local store. They do not prove a complete set at one database snapshot. Absence from a page never proves deletion or loss of access. + +This HTTP protocol uses `artifacts.incremental`. The WebSocket Sync subscription uses a separate `artifacts.sync` capability and defines its own complete-set guarantee. Clients select each protocol through its own capability. + +The relay advertises `artifacts.incremental` when incremental reads are enabled. +Each signed `POST /query` request contains one filter. The query uses the +same community, relay admission, channel access, and exact tag rules as an +ordinary artifact read. The existing `current` and `history` array responses +keep their format. +Unlisted open channels require membership or an explicit `#h` scope. +Known-ID reads do not bypass this rule. + +```json +[{"artifact":"incremental","query":{"artifact":"current","#type":["buzz.task"]}}] +``` + +The initial read pages through current identities. A later read supplies the +last completed `checkpoint`. A continuation supplies only `cursor`, with +the same `query`. The server signs these opaque strings. They bind the +community, reader, exact query, access-policy boundary, and database start +time. Tokens contain a fixed-size digest of the query. The cursor also binds +the last page position. Relay instances use the +same configured signing identity, so another instance can serve a continuation. +No server-side manifest or Redis allocation is required. + +```json +{ + "version": 1, + "startedAt": "2026-10-21T12:00:00.123456Z", + "overlapSeconds": 120, + "changes": [ + {"type":"upsert","artifactId":"22222222-2222-4222-8222-222222222222","event":{}}, + {"type":"remove","artifactId":"33333333-3333-4333-8333-333333333333"}, + {"type":"removeChannel","channelId":"44444444-4444-4444-8444-444444444444"} + ], + "nextCursor": null, + "checkpoint": "opaque" +} +``` + +`event` is the complete signed current revision. `remove` has one meaning: +the requested artifact is unavailable for this query. It does not distinguish +a missing identity, deletion, redaction, query exit, or denied access. +`removeChannel` names a removed membership retained for this exact reader. +It applies only to local rows whose current saved home is that channel. +An earlier move upsert to another home therefore survives a later source +channel removal. + +Each pass fixes `startedAt` from the database clock. A later pass reads from +the previous completed start time minus 120 seconds through its own start +time, inclusive. New head writes, move-source records, and membership changes +use database `clock_timestamp()`. The protocol assumes the delay from this +stamp to commit visibility stays within the accepted 120-second overlap. +That overlap is an operating assumption. It is not a measured upper bound. + +Pages order by `(change timestamp, source, stable identity)`. The source +separates membership removals, head changes, grants, and move-source records. +The identity breaks equal timestamp ties. It does not encode creation time. +The time window and page position are stable. The current contents can change +between requests. Replays must be idempotent. A later pass repairs changes +that move beyond the current window. + +Only the final page returns a checkpoint. The client saves it after every +page and local commit succeeds. A failed read keeps the prior checkpoint. +Neither an initial read nor delta completion establishes absence. The client +must not remove unseen rows. A fresh access grant replays older current +heads from that channel. A move retains its source channel in the existing +revision ledger, so a reader who cannot read the destination receives a +current revalidation result without destination metadata. + +### Current reads and write confirmation + +```json +[{"artifact":"incremental","action":"current","query":{"artifact":"current","#type":["buzz.task"]},"artifactIds":["22222222-2222-4222-8222-222222222222"]}] +``` + +The response has `version`, database `startedAt`, and `changes`. It contains +exactly one upsert or remove for each requested ID, in request order. +The request permits 1–100 unique nonempty strings, each at most 256 UTF-8 +bytes. A string that is not a canonical artifact UUID returns remove. +Duplicates and invalid bounds fail with `incremental_invalid`. The server does not +truncate the request. + +Clients serialize a full delta pass and its commits before a current-ID read. +Live revision pushes request a current-ID read; they do not replace confirmed +rows directly. A current-ID read that starts after publication acceptance +reads that commit or a later state on the writer. It can return a later head +or remove. An empty delta does not confirm an accepted edit. + +### Access recovery and bounds + +Relevant channel access changes atomically advance one community policy +stamp, including transitions on empty channels. Ordinary artifact writes do +not update this stamp. A different stamp +returns HTTP 409 `incremental_reset` with `reason: "policy_changed"`, without naming +private channels. A checkpoint older than 30 days returns the same code with +`reason: "checkpoint_expired"`. + +Recovery runs a new initial read to discover current readable records and +bounded current-ID reads for the IDs already saved locally. These reads +revalidate known IDs without a channel roster or an absence scan. Save new +progress only after discovery, revalidation, and local commits succeed. +Preserve preferences when a record becomes unavailable. + +The server retains task heads, membership removal rows, and move-source +ledger records. Removing this recovery history requires an explicit policy +reset before a client can resume. Whole-community deletion closes admission +and purges these rows through the existing deletion path. + +NIP-11 advertises: + +```json +{"version":1,"overlapSeconds":120,"pageRows":100,"currentRows":100,"responseBytes":8388608,"permanentChannelsOnly":true,"checkpointLifetimeSeconds":2592000} +``` + +Pages shrink to the response byte bound. A single oversized result returns +HTTP 413 `incremental_limit`. A current-ID batch can return that error; the client +can split the batch. No page or batch silently truncates its result. The +server permits four concurrent operations per process, a two-second SQL +statement budget, and an eight-second operation budget. There is no +identity-wide membership count limit on this path. + +Incremental reads include only matching readable artifacts in permanent channels. +TTL channels do not contribute discovery candidates. A known-ID read returns +`remove` for an artifact in a TTL channel. Moves into a TTL channel return +removals for the permanent source. An explicit `#h` scope limits candidates +to those channels, including retained move sources and membership removals. +Initial reads omit deleted heads. Later reads retain deletion removals. +HTTP 400 `incremental_invalid` +rejects invalid requests or tokens. HTTP 503 `incremental_unavailable` reports a +service or work-budget failure. These failures do not advance progress. + +The rollout gate is `BUZZ_ARTIFACT_INCREMENTAL_ENABLED`, off by default. Follow +[artifact-sync-deployment.md](../artifact-sync-deployment.md) before enabling +it. Incremental completion does not establish a live subscription to every channel. diff --git a/migrations/0076_artifact_revision_lookup.sql b/migrations/0076_artifact_revision_lookup.sql new file mode 100644 index 00000000000..707f3f89b1d --- /dev/null +++ b/migrations/0076_artifact_revision_lookup.sql @@ -0,0 +1,24 @@ +-- Populated databases prebuild this index concurrently. +-- See docs/artifact-sync-deployment.md. +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +DO $$ +BEGIN + IF to_regclass('public.idx_artifact_revisions_artifact') IS NULL THEN + IF EXISTS (SELECT 1 FROM public.artifact_revisions LIMIT 1) THEN + RAISE EXCEPTION 'prebuild idx_artifact_revisions_artifact concurrently; see docs/artifact-sync-deployment.md'; + END IF; + CREATE INDEX idx_artifact_revisions_artifact + ON public.artifact_revisions (community_id, artifact_id); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_index + WHERE indexrelid=to_regclass('public.idx_artifact_revisions_artifact') + AND indisvalid AND indisready AND indislive + AND pg_get_indexdef(indexrelid)= + 'CREATE INDEX idx_artifact_revisions_artifact ON public.artifact_revisions USING btree (community_id, artifact_id)' + ) THEN + RAISE EXCEPTION 'idx_artifact_revisions_artifact invalid or wrong definition; see docs/artifact-sync-deployment.md'; + END IF; +END $$; diff --git a/migrations/0077_artifact_sync_timestamps.sql b/migrations/0077_artifact_sync_timestamps.sql new file mode 100644 index 00000000000..b0df2f30102 --- /dev/null +++ b/migrations/0077_artifact_sync_timestamps.sql @@ -0,0 +1,11 @@ +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +ALTER TABLE artifact_heads ADD COLUMN IF NOT EXISTS modified_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch'; +ALTER TABLE artifact_heads ALTER COLUMN modified_at SET DEFAULT clock_timestamp(); +ALTER TABLE artifact_revisions ADD COLUMN IF NOT EXISTS source_channel_id UUID; +ALTER TABLE artifact_revisions ADD COLUMN IF NOT EXISTS modified_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch'; +ALTER TABLE artifact_revisions ALTER COLUMN modified_at SET DEFAULT clock_timestamp(); +ALTER TABLE channel_members ADD COLUMN IF NOT EXISTS sync_modified_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch'; +ALTER TABLE channel_members ALTER COLUMN sync_modified_at SET DEFAULT clock_timestamp(); +ALTER TABLE communities ADD COLUMN IF NOT EXISTS artifact_policy_modified_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch'; diff --git a/migrations/0078_artifact_sync_indexes.sql b/migrations/0078_artifact_sync_indexes.sql new file mode 100644 index 00000000000..18af6922132 --- /dev/null +++ b/migrations/0078_artifact_sync_indexes.sql @@ -0,0 +1,35 @@ +-- Populated databases prebuild these indexes concurrently after migration 0077. +-- See docs/artifact-sync-deployment.md. +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +DO $$ +DECLARE + item RECORD; + populated BOOLEAN; +BEGIN + FOR item IN SELECT * FROM (VALUES + ('artifact_heads_modified', 'artifact_heads', '(community_id, modified_at, artifact_id)', ''), + ('artifact_heads_channel', 'artifact_heads', '(community_id, channel_id, artifact_id)', ''), + ('channel_members_sync', 'channel_members', '(community_id, pubkey, sync_modified_at, channel_id)', ''), + ('artifact_revisions_moves', 'artifact_revisions', '(community_id, modified_at, event_id)', ' WHERE source_channel_id IS NOT NULL') + ) AS indexes(name, table_name, columns, predicate) + LOOP + IF to_regclass('public.' || item.name) IS NULL THEN + EXECUTE format('SELECT EXISTS(SELECT 1 FROM public.%I LIMIT 1)',item.table_name) INTO populated; + IF populated THEN + RAISE EXCEPTION 'prebuild % concurrently; see docs/artifact-sync-deployment.md', item.name; + END IF; + EXECUTE format('CREATE INDEX %I ON public.%I %s%s', item.name, item.table_name, item.columns, item.predicate); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_index WHERE indexrelid=to_regclass('public.' || item.name) + AND indisvalid AND indisready AND indislive + AND pg_get_indexdef(indexrelid)=format('CREATE INDEX %s ON public.%s USING btree %s%s', + item.name,item.table_name,item.columns, + CASE WHEN item.predicate='' THEN '' ELSE ' WHERE (source_channel_id IS NOT NULL)' END) + ) THEN + RAISE EXCEPTION '% is invalid or has a different definition', item.name; + END IF; + END LOOP; +END $$; diff --git a/schema/tables/public/artifact_heads.sql b/schema/tables/public/artifact_heads.sql index 168ec09d41f..7e73557b9b6 100644 --- a/schema/tables/public/artifact_heads.sql +++ b/schema/tables/public/artifact_heads.sql @@ -7,9 +7,12 @@ CREATE TABLE artifact_heads ( artifact_type TEXT NOT NULL, root BYTEA, deleted BOOLEAN NOT NULL DEFAULT false, + modified_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), PRIMARY KEY (community_id, artifact_id) ); CREATE INDEX artifact_heads_event ON artifact_heads (community_id, event_id); +CREATE INDEX artifact_heads_modified ON artifact_heads (community_id, modified_at, artifact_id); +CREATE INDEX artifact_heads_channel ON artifact_heads (community_id, channel_id, artifact_id); -- The relay does not expire events. Any future row retention or partition -- retirement must skip payloads referenced by `artifact_heads.event_id` diff --git a/schema/tables/public/artifact_revisions.sql b/schema/tables/public/artifact_revisions.sql index 57086018141..d3320c0bfff 100644 --- a/schema/tables/public/artifact_revisions.sql +++ b/schema/tables/public/artifact_revisions.sql @@ -4,5 +4,12 @@ CREATE TABLE artifact_revisions ( community_id UUID NOT NULL REFERENCES communities(id), event_id BYTEA NOT NULL CHECK (length(event_id) = 32), artifact_id UUID NOT NULL, + source_channel_id UUID, + modified_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), PRIMARY KEY (community_id, event_id) ); + +CREATE INDEX idx_artifact_revisions_artifact + ON artifact_revisions (community_id, artifact_id); +CREATE INDEX artifact_revisions_moves ON artifact_revisions (community_id, modified_at, event_id) + WHERE source_channel_id IS NOT NULL; diff --git a/schema/tables/public/channel_members.sql b/schema/tables/public/channel_members.sql index 7510e455de8..b0cd4e67926 100644 --- a/schema/tables/public/channel_members.sql +++ b/schema/tables/public/channel_members.sql @@ -11,6 +11,7 @@ CREATE TABLE channel_members ( removed_at TIMESTAMPTZ, removed_by BYTEA, hidden_at TIMESTAMPTZ, + sync_modified_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), PRIMARY KEY (community_id, channel_id, pubkey), FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id) ON DELETE CASCADE @@ -18,3 +19,4 @@ CREATE TABLE channel_members ( CREATE INDEX idx_channel_members_pubkey ON channel_members (community_id, pubkey) WHERE removed_at IS NULL; +CREATE INDEX channel_members_sync ON channel_members (community_id, pubkey, sync_modified_at, channel_id); diff --git a/schema/tables/public/communities.sql b/schema/tables/public/communities.sql index 06f74c3f585..7f9d6f09653 100644 --- a/schema/tables/public/communities.sql +++ b/schema/tables/public/communities.sql @@ -24,6 +24,7 @@ CREATE TABLE communities ( deletion_state TEXT NOT NULL DEFAULT 'active' CHECK (deletion_state IN ('active', 'quiescing', 'fenced', 'tombstone')), deletion_fence_generation BIGINT NOT NULL DEFAULT 0 CHECK (deletion_fence_generation >= 0), deleted_at TIMESTAMPTZ, + artifact_policy_modified_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch', CONSTRAINT chk_communities_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid) );