From 43ea0f00bf6007a3ea07b351dc13a57ed2b0defc Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:08:38 -0800 Subject: [PATCH 01/20] fix(relay): certify artifact sync visibility durably Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/runtime/admitted_tx.rs | 14 ++ crates/buzz-db/src/runtime/migration.rs | 3 +- crates/buzz-db/src/store/artifact.rs | 1 + .../src/store/artifact_certification.rs | 100 +++++++++++ .../artifact_certification_postgres_tests.rs | 160 ++++++++++++++++++ .../src/store/artifact_postgres_tests.rs | 49 ++++++ crates/buzz-db/src/store/artifact_sync.rs | 2 + crates/buzz-db/src/store/channel.rs | 11 +- crates/buzz-db/src/store/channel_members.rs | 3 + crates/buzz-db/src/store/deletion.rs | 9 + crates/buzz-db/src/store/dm.rs | 1 + crates/buzz-db/src/store/event.rs | 3 + crates/buzz-db/src/store/mod.rs | 1 + .../buzz-db/src/store/relay_admin_actions.rs | 2 + .../sync_policy_postgres_tests.rs | 39 +++++ crates/buzz-relay/src/api/artifact_sync.rs | 4 +- .../buzz-relay/src/artifact_certification.rs | 30 ++++ crates/buzz-relay/src/lib.rs | 2 + crates/buzz-relay/src/main.rs | 2 + docs/artifact-sync-deployment.md | 50 +++++- .../0081_artifact_sync_certification.sql | 15 ++ schema/schema.sql | 1 + .../tables/public/artifact_sync_pending.sql | 12 ++ 23 files changed, 506 insertions(+), 8 deletions(-) create mode 100644 crates/buzz-db/src/store/artifact_certification.rs create mode 100644 crates/buzz-db/src/store/artifact_certification_postgres_tests.rs create mode 100644 crates/buzz-relay/src/artifact_certification.rs create mode 100644 migrations/0081_artifact_sync_certification.sql create mode 100644 schema/tables/public/artifact_sync_pending.sql diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index 2e4ce60a758..d646d8b827b 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -72,6 +72,7 @@ use crate::Result; pub struct AdmittedTx { tx: Transaction<'static, Postgres>, community: CommunityId, + artifact_certification: Option, /// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes. ttl_refresh_channels: BTreeSet, push_enqueue: Option, @@ -89,6 +90,7 @@ impl AdmittedTx { Ok(Self { tx, community, + artifact_certification: None, ttl_refresh_channels: BTreeSet::new(), push_enqueue: None, push_events: Vec::new(), @@ -108,6 +110,7 @@ impl AdmittedTx { Ok(Self { tx, community: lease.community_id, + artifact_certification: None, ttl_refresh_channels: BTreeSet::new(), push_enqueue: None, push_events: Vec::new(), @@ -131,6 +134,17 @@ impl AdmittedTx { } } + /// Record work before the first change that affects incremental artifact reads. + pub(crate) async fn record_artifact_change(&mut self) -> Result<()> { + if self.artifact_certification.is_none() { + let operation = Uuid::new_v4(); + sqlx::query("INSERT INTO artifact_sync_pending (community_id,operation_id,earliest_at) VALUES ($1,$2,clock_timestamp())") + .bind(self.community.as_uuid()).bind(operation).execute(&mut *self.tx).await?; + self.artifact_certification = Some(operation); + } + Ok(()) + } + /// The community this transaction was admitted for. pub fn community(&self) -> CommunityId { self.community diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index c78a99eab5b..48709754b43 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -777,7 +777,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 80); + assert_eq!(migrations.len(), 81); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -2893,6 +2893,7 @@ mod postgres_tests { "entity_deletion_requests", "artifact_heads", "artifact_revisions", + "artifact_sync_pending", "channel_members", "communities", "entity_deletion_stage_history", diff --git a/crates/buzz-db/src/store/artifact.rs b/crates/buzz-db/src/store/artifact.rs index 6f9f46eb174..0fc29562e48 100644 --- a/crates/buzz-db/src/store/artifact.rs +++ b/crates/buzz-db/src/store/artifact.rs @@ -184,6 +184,7 @@ impl Db { } } } + tx.record_artifact_change().await?; sqlx::query( "INSERT INTO artifact_revisions (community_id,event_id,artifact_id,source_channel_id,modified_at) VALUES ($1,$2,$3,$4,clock_timestamp())", ) diff --git a/crates/buzz-db/src/store/artifact_certification.rs b/crates/buzz-db/src/store/artifact_certification.rs new file mode 100644 index 00000000000..db7b14e01e2 --- /dev/null +++ b/crates/buzz-db/src/store/artifact_certification.rs @@ -0,0 +1,100 @@ +//! Durable proof of the timestamp overlap used by incremental artifact reads. + +use buzz_core::CommunityId; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::{Db, Result}; + +/// Timestamp overlap shared by writers and incremental readers. +pub const OVERLAP_SECONDS: i64 = 120; +const BATCH: i64 = 100; + +pub(crate) async fn certify( + pool: &PgPool, + community: CommunityId, + operation: Option, +) -> Result { + let Some(mut tx) = crate::begin_community_sweep_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await? + else { + return Ok(0); + }; + let completed = certify_pending(&mut tx, operation).await?; + tx.commit().await?; + Ok(completed) +} + +async fn certify_pending(tx: &mut crate::AdmittedTx, operation: Option) -> Result { + let community = tx.community(); + sqlx::query("SET LOCAL statement_timeout='2s'") + .execute(tx.conn()) + .await?; + let pending: Vec<(Uuid, chrono::DateTime)> = sqlx::query_as( + "SELECT operation_id,earliest_at FROM artifact_sync_pending \ + WHERE community_id=$1 AND ($2::uuid IS NULL OR operation_id=$2) \ + ORDER BY earliest_at,operation_id LIMIT $3 FOR UPDATE SKIP LOCKED", + ) + .bind(community.as_uuid()) + .bind(operation) + .bind(BATCH) + .fetch_all(tx.conn()) + .await?; + if pending.is_empty() { + return Ok(0); + } + // This clock read follows a read that sees the committed pending records. + // A delayed recovery makes the proof conservative and requires a reset. + let now: chrono::DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(tx.conn()) + .await?; + if pending.iter().any(|(_, earliest)| { + *earliest > now || now - *earliest >= chrono::Duration::seconds(OVERLAP_SECONDS) + }) { + sqlx::query("UPDATE communities SET artifact_policy_modified_at=GREATEST(clock_timestamp(),artifact_policy_modified_at + interval '1 microsecond') WHERE id=$1") + .bind(community.as_uuid()) + .execute(tx.conn()) + .await?; + } + let operations: Vec<_> = pending.iter().map(|(id, _)| *id).collect(); + sqlx::query("DELETE FROM artifact_sync_pending WHERE community_id=$1 AND operation_id=ANY($2)") + .bind(community.as_uuid()) + .bind(&operations) + .execute(tx.conn()) + .await?; + Ok(pending.len()) +} + +impl Db { + /// Recover one bounded batch of committed artifact certifications. + /// Multiple relay instances can run this method at the same time. + pub async fn recover_artifact_certifications(&self) -> Result { + let communities: Vec = sqlx::query_scalar( + "SELECT DISTINCT community_id FROM \ + (SELECT p.community_id FROM artifact_sync_pending p JOIN communities c ON c.id=p.community_id WHERE c.deletion_state='active' AND c.deleted_at IS NULL ORDER BY p.earliest_at,p.community_id,p.operation_id LIMIT $1) p", + ) + .bind(BATCH) + .fetch_all(&self.pool) + .await?; + let mut completed = 0; + for id in communities { + let community = CommunityId::from_uuid(id); + match certify(&self.pool, community, None).await { + Ok(count) => completed += count, + Err(error) => { + tracing::warn!(%error, community_id = %community, + "artifact certification requires recovery"); + } + } + } + Ok(completed) + } +} + +#[cfg(test)] +#[path = "artifact_certification_postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs new file mode 100644 index 00000000000..f2fae8700ca --- /dev/null +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -0,0 +1,160 @@ +use super::*; +use chrono::{DateTime, Utc}; + +async fn fixture() -> (Db, CommunityId) { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") { + crate::migration::run_migrations(&pool).await.unwrap(); + } + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community.as_uuid()) + .bind(format!("certification-{}.test", community.as_uuid())) + .execute(&pool) + .await + .unwrap(); + (Db::from_pool(pool), community) +} + +async fn begin(db: &Db, community: CommunityId) -> crate::AdmittedTx { + db.begin_event_write_transaction(community).await.unwrap() +} + +async fn state(db: &Db, community: CommunityId) -> (DateTime, i64) { + sqlx::query_as("SELECT artifact_policy_modified_at,(SELECT count(*) FROM artifact_sync_pending WHERE community_id=$1) FROM communities WHERE id=$1") + .bind(community.as_uuid()).fetch_one(db.pool()).await.unwrap() +} + +async fn age(tx: &mut crate::AdmittedTx, seconds: i32) { + let community = tx.community(); + sqlx::query("UPDATE artifact_sync_pending SET earliest_at=clock_timestamp()-make_interval(secs=>$2) WHERE community_id=$1") + .bind(community.as_uuid()).bind(seconds as f64).execute(tx.conn()).await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn rollback_and_lost_commit_reply_leave_correct_recovery_work() { + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + assert_eq!(state(&db, community).await, initial); + drop(tx); + assert_eq!(state(&db, community).await, initial); + + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + tx.record_artifact_change().await.unwrap(); + age(&mut tx, 121).await; + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); + // The writer can lose its commit reply and never return to retry. + tx.commit().await.unwrap(); + assert_eq!(state(&db, community).await, (initial.0, 1)); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); + let completed = state(&db, community).await; + assert!(completed.0 > initial.0); + assert_eq!(completed.1, 0); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); + assert_eq!(state(&db, community).await, completed); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn visibility_within_overlap_preserves_the_policy_stamp() { + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + tx.commit().await.unwrap(); + assert_eq!(certify(db.pool(), community, None).await.unwrap(), 1); + assert_eq!(state(&db, community).await, initial); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn certification_crash_rolls_back_reset_and_completion_together() { + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + age(&mut tx, 121).await; + tx.commit().await.unwrap(); + + let mut recovery = begin(&db, community).await; + assert_eq!(certify_pending(&mut recovery, None).await.unwrap(), 1); + assert_eq!(state(&db, community).await, (initial.0, 1)); + drop(recovery); + let (a, b) = tokio::join!( + certify(db.pool(), community, None), + certify(db.pool(), community, None), + ); + assert_eq!(a.unwrap() + b.unwrap(), 1); + let completed = state(&db, community).await; + assert!(completed.0 > initial.0); + assert_eq!(completed.1, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failed_reset_keeps_pending_work_and_retry_is_idempotent() { + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + age(&mut tx, 121).await; + tx.commit().await.unwrap(); + let operation: Uuid = + sqlx::query_scalar("SELECT operation_id FROM artifact_sync_pending WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(db.pool()) + .await + .unwrap(); + let mut blocker = db.pool().begin().await.unwrap(); + sqlx::query("SELECT id FROM communities WHERE id=$1 FOR NO KEY UPDATE") + .bind(community.as_uuid()) + .execute(&mut *blocker) + .await + .unwrap(); + assert!(certify(db.pool(), community, Some(operation)) + .await + .is_err()); + assert_eq!(state(&db, community).await, (initial.0, 1)); + blocker.rollback().await.unwrap(); + assert_eq!( + certify(db.pool(), community, Some(operation)) + .await + .unwrap(), + 1 + ); + let completed = state(&db, community).await; + assert_eq!( + certify(db.pool(), community, Some(operation)) + .await + .unwrap(), + 0 + ); + assert_eq!(state(&db, community).await, completed); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn recovery_handles_clock_reversal_and_leaves_deleting_communities() { + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + age(&mut tx, -121).await; + tx.commit().await.unwrap(); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); + assert!(state(&db, community).await.0 > initial.0); + + sqlx::query("UPDATE communities SET deletion_state='quiescing' WHERE id=$1") + .bind(community.as_uuid()) + .execute(db.pool()) + .await + .unwrap(); + assert!(db.begin_event_write_transaction(community).await.is_err()); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); +} diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index f858deeb182..5001ed4d945 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -15,6 +15,55 @@ struct Fixture { peer: Keys, relay: Keys, } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn artifact_writes_record_one_certification_before_their_timestamps() { + let f = Fixture::new().await; + let id = Uuid::new_v4(); + let mut previous = None; + let mut home = f.a; + for op in ["create", "update", "move", "delete", "restore"] { + let source = if op == "move" { Some(home) } else { None }; + if op == "move" { + home = f.b; + } + let event = f.revision(id, op, home, previous.as_ref(), &f.owner, vec![]); + assert!(matches!( + f.accept(&event, source).await, + ArtifactOutcome::Accepted(_) + )); + let (count, early): (i64, bool) = sqlx::query_as( + "SELECT count(*),bool_and(p.earliest_at <= LEAST(h.modified_at,r.modified_at)) \ + FROM artifact_sync_pending p JOIN artifact_heads h USING(community_id) \ + JOIN artifact_revisions r ON r.community_id=h.community_id AND r.event_id=h.event_id \ + WHERE p.community_id=$1 AND h.artifact_id=$2", + ) + .bind(f.community.as_uuid()) + .bind(id) + .fetch_one(&f.db.pool) + .await + .unwrap(); + assert_eq!((count, early), (1, true), "{op}"); + assert!(matches!( + f.accept(&event, source).await, + ArtifactOutcome::Duplicate + )); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + previous = Some(event); + } + let event = previous.unwrap(); + crate::event::soft_delete_event_and_update_thread( + &f.db.pool, + f.community, + event.id.as_bytes(), + None, + None, + ) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); +} impl Fixture { async fn new() -> Self { let pool = PgPool::connect(&crate::test_support::database_url()) diff --git a/crates/buzz-db/src/store/artifact_sync.rs b/crates/buzz-db/src/store/artifact_sync.rs index 4da01728fd0..281398cdd1e 100644 --- a/crates/buzz-db/src/store/artifact_sync.rs +++ b/crates/buzz-db/src/store/artifact_sync.rs @@ -7,6 +7,8 @@ use serde::{Deserialize, Serialize}; use sqlx::{QueryBuilder, Row}; use uuid::Uuid; +pub use super::artifact_certification::OVERLAP_SECONDS; + /// Maximum changes or requested identities in one page. pub const PAGE_ROWS: usize = 100; diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 36ae4a1dc6b..5572152faf4 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -161,6 +161,7 @@ pub async fn create_channel( .execute(tx.conn()) .await?; + tx.record_artifact_change().await?; sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) @@ -403,6 +404,7 @@ async fn insert_channel_rows( if was_created { // Bootstrap the creator as owner. + tx.record_artifact_change().await?; sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) @@ -431,6 +433,7 @@ async fn insert_channel_rows( member.len() ))); } + tx.record_artifact_change().await?; sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) @@ -1204,13 +1207,16 @@ async fn update_channel_on( } else { false }; + if policy_changed { + tx.record_artifact_change().await?; + } 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", + "UPDATE communities SET artifact_policy_modified_at=GREATEST(clock_timestamp(),artifact_policy_modified_at + interval '1 microsecond') WHERE id=$1", ) .bind(community_id.as_uuid()) .execute(tx.conn()) @@ -1564,12 +1570,13 @@ async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> { } let channel_id = w.channel_id; let community_id = w.tx.community(); + w.tx.record_artifact_change().await?; sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2") .bind(community_id.as_uuid()) .bind(channel_id) .execute(w.tx.conn()) .await?; - sqlx::query("UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1") + sqlx::query("UPDATE communities SET artifact_policy_modified_at=GREATEST(clock_timestamp(),artifact_policy_modified_at + interval '1 microsecond') WHERE id=$1") .bind(community_id.as_uuid()) .execute(w.tx.conn()) .await?; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 2c139228a8e..faecc1d00cd 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -253,6 +253,7 @@ impl ChannelMembershipLocks<'_> { ) -> Result<()> { self.check_covers(channel_id)?; let community_id = self.tx.community(); + self.tx.record_artifact_change().await?; sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) @@ -1015,6 +1016,7 @@ async fn put_members_on( let mut written: std::collections::HashMap, MemberRecord> = std::collections::HashMap::with_capacity(write_keys.len()); if !write_keys.is_empty() { + tx.record_artifact_change().await?; let rows = sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) @@ -1152,6 +1154,7 @@ pub async fn remove_member( } } + tx.record_artifact_change().await?; let result = sqlx::query( r#" UPDATE channel_members diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 9c1eb2f433b..5501e1818e6 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -81,6 +81,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "nip_fi_profile_heads", "api_tokens", "archived_identities", + "artifact_sync_pending", "artifact_heads", "artifact_revisions", "audit_log", @@ -151,6 +152,7 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[ "push_leases", "relay_invites", "delivery_log", + "artifact_sync_pending", "artifact_heads", "artifact_revisions", "events", @@ -8628,6 +8630,12 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); let (request, inventory) = inventoried_request(&db, &store).await; let community = *request.community_id.as_uuid(); seed_relay_admin_rows(&db, community).await; + let mut artifact_write = db + .begin_event_write_transaction(request.community_id) + .await + .unwrap(); + artifact_write.record_artifact_change().await.unwrap(); + artifact_write.commit().await.unwrap(); let bystander = db .ensure_configured_community(&format!("bystander-{}.example", Uuid::new_v4().simple())) .await @@ -8699,6 +8707,7 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .await .expect("bindings"); let deleted = store.purge_postgres(&token).await.expect("purge postgres"); + assert_eq!(deleted.get("artifact_sync_pending"), Some(&1)); let retained: Uuid = sqlx::query_scalar( "SELECT target_community_id FROM entity_deletion_stage_history WHERE id = $1", diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index ebdbc45d271..30eeba033c7 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -187,6 +187,7 @@ pub async fn create_dm( // Add all participants as members with role='member'. for pk in participants { + tx.record_artifact_change().await?; sqlx::query( r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index bf80e302d44..3986ec90486 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1557,6 +1557,9 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx( let deleted = purged > 0 || result.rows_affected() > 0; if deleted { + if target.is_some_and(|(kind, _)| kind == 45010) { + tx.record_artifact_change().await?; + } 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 { diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index e71f292ef27..cceff28f22b 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -82,6 +82,7 @@ pub mod workflow; /// NIP-AR artifact lifecycle and durable delivery. pub mod artifact; +mod artifact_certification; mod artifact_query; /// Timestamp-based incremental artifact reads. pub mod artifact_sync; diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index f1fa4af9beb..4ab1135f0af 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -718,6 +718,7 @@ pub async fn execute_kick_with_marker( crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id) .await?; + tx.record_artifact_change().await?; let kick = sqlx::query( r#" UPDATE channel_members @@ -1726,6 +1727,7 @@ pub async fn deploy_kick_member( .await?; // Use a direct UPDATE to avoid the tenant ownership check in channel::remove_member. // This is the deployment-authority primitive: no actor role check. + tx.record_artifact_change().await?; let result = sqlx::query( r#" UPDATE channel_members 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 index c3cc3d9f1d4..cd93f169d7e 100644 --- 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 @@ -1,5 +1,44 @@ use super::*; +#[tokio::test] +#[ignore = "requires Postgres"] +async fn durable_certification_resets_existing_checkpoints_after_the_overlap() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let first = sync(&f, None).await; + let id = Uuid::new_v4(); + let event = f.revision(id, "create", None); + accept(&f, &event).await; + // Model a commit whose original stamp predates the checkpoint overlap. + sqlx::query("UPDATE artifact_heads SET modified_at=$1::timestamptz-interval '121 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(); + sqlx::query("UPDATE artifact_sync_pending SET earliest_at=$1::timestamptz-interval '121 seconds' WHERE community_id=$2") + .bind(first["startedAt"].as_str().unwrap()).bind(f.community.as_uuid()).execute(&f.pool).await.unwrap(); + let missed = sync(&f, Some(&first["checkpoint"])).await; + assert_eq!(missed["changes"], json!([])); + assert_eq!( + f.state.db.recover_artifact_certifications().await.unwrap(), + 1 + ); + for checkpoint in [&first["checkpoint"], &missed["checkpoint"]] { + let (status, body) = f + .request( + "/query", + json!([{ + "artifact":"incremental", "query":query(), "checkpoint":checkpoint + }]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(body["reason"], "policy_changed"); + } + let recovered = sync(&f, None).await; + assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); + let (status, ordinary) = f.request("/query", json!([query()])).await; + assert!(status.is_success()); + assert!(ordinary.to_string().contains(&event.id.to_hex())); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn incremental_resets_for_first_artifact_during_channel_transition() { diff --git a/crates/buzz-relay/src/api/artifact_sync.rs b/crates/buzz-relay/src/api/artifact_sync.rs index 6ec686d3ff2..e46aee2958a 100644 --- a/crates/buzz-relay/src/api/artifact_sync.rs +++ b/crates/buzz-relay/src/api/artifact_sync.rs @@ -6,7 +6,7 @@ use buzz_core::{ artifact::{ArtifactQuery, ArtifactView}, TenantContext, }; -use buzz_db::artifact_sync::{Candidate, Position, PAGE_ROWS}; +use buzz_db::artifact_sync::{Candidate, Position, OVERLAP_SECONDS, PAGE_ROWS}; use chrono::{DateTime, Duration, Utc}; use hmac::{Hmac, KeyInit, Mac}; use serde::{Deserialize, Serialize}; @@ -18,7 +18,7 @@ use std::{ }; use uuid::Uuid; -const OVERLAP: i64 = 120; +const OVERLAP: i64 = OVERLAP_SECONDS; const CHECKPOINT_DAYS: i64 = 30; const RESPONSE_BYTES: usize = 8_388_608; static WORK: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4); diff --git a/crates/buzz-relay/src/artifact_certification.rs b/crates/buzz-relay/src/artifact_certification.rs new file mode 100644 index 00000000000..ad70ddc8a29 --- /dev/null +++ b/crates/buzz-relay/src/artifact_certification.rs @@ -0,0 +1,30 @@ +//! Recovery of committed artifact work that has no visibility proof. + +use std::{sync::Arc, time::Duration}; + +use crate::state::AppState; + +/// Retry pending certifications without a client retry or an enabled read gate. +pub async fn run(state: Arc) { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + if state + .shutting_down + .load(std::sync::atomic::Ordering::Acquire) + { + return; + } + match tokio::time::timeout( + Duration::from_secs(5), + state.db.recover_artifact_certifications(), + ) + .await + { + Ok(Ok(_)) => {} + Ok(Err(error)) => tracing::warn!(%error, "artifact certification recovery failed"), + Err(_) => tracing::warn!("artifact certification recovery reached its time limit"), + } + } +} diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 69f3432431f..b0d593c857b 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -21,6 +21,8 @@ pub(crate) mod nip_fi_upgrade; /// REST API route handlers. pub mod api; +/// Durable recovery for incremental artifact reads. +pub mod artifact_certification; /// WebSocket audio relay for huddle voice channels. pub mod audio; /// Relay configuration from environment variables. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index f48926f72d2..9ac8d8a2730 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1462,6 +1462,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { }); } + tokio::spawn(buzz_relay::artifact_certification::run(Arc::clone(&state))); + // NIP-AR Sync: a possible loss on the Redis link resyncs every sync. tokio::spawn(buzz_relay::handlers::artifact_sync::watch_link(Arc::clone( &state, diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index 6af589b2bd5..a187a81a459 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -65,6 +65,50 @@ 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. +Without durable certification on every serving writer, the 120-second +stamp-to-visibility bound remains an operating assumption. Runtime measurements +are tracked separately. + +## Durable certification + +Migration `0081` adds `artifact_sync_pending`. It is an expand migration. +It adds no trigger, function, foreign key, or backfill. Old relays ignore this +table. New relays require it before they accept relevant writes. With startup +migrations enabled, the older binary still has the SQLx `VersionMissing` +restriction described in `docs/database.md`. + +Each supported writer records one pending operation in its admitted transaction. +The operation has a unique UUID. Its timestamp precedes every relevant sync +stamp in that transaction. Repeated changes in the transaction reuse the record. +Rollback removes both the changes and the record. An uncertain commit leaves +both committed or neither committed. Recovery does not require a caller retry. + +Every relay starts a recovery worker, even when the incremental gate is off. +The worker reads bounded batches. It locks pending rows with `SKIP LOCKED`. +After it observes a committed record, it reads database time. If less than +120 seconds separate that time from the early timestamp, the overlap covers +visibility of the write. A clock reversal or a delay of at least 120 seconds +requires a reset. A delayed recovery can cause an unnecessary reset. + +Recovery advances `artifact_policy_modified_at` only when it requires a reset. +It removes the pending rows in that same transaction. A crash before this +commit leaves the pending rows for retry. A retry after the commit finds no +work. Ordinary writes do not acquire a shared counter or policy row lock. + +Existing clients use the existing `policy_changed` reset. They start discovery +again and revalidate saved IDs. The request, response, cursor, checkpoint, and +capability formats stay at version 1. No client change is required. A reset +covers the community because that is the existing policy boundary. + +Full coverage requires all serving writers to use this release. The covered +paths are artifact create, update, move, delete, restore, and redaction; +channel creation; membership grants, rejoins, and removals; channel visibility, +listing, TTL, and deletion changes; DM creation; and admin member removals. +Archive state and hidden DMs do not change artifact access. Incremental reads +exclude TTL channels. Whole-community deletion stops reads and purges pending +records. Operator SQL and mixed old writers cannot provide this guarantee. +Keep the gate off during a mixed rollout. Use the existing reset procedure +before enablement or after a rollback to an old writer. + +This change does not authorize deployment, gate enablement, or a production +maintenance write. diff --git a/migrations/0081_artifact_sync_certification.sql b/migrations/0081_artifact_sync_certification.sql new file mode 100644 index 00000000000..9361a2e25f9 --- /dev/null +++ b/migrations/0081_artifact_sync_certification.sql @@ -0,0 +1,15 @@ +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +CREATE TABLE artifact_sync_pending ( + community_id UUID NOT NULL, + operation_id UUID NOT NULL, + earliest_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (community_id, operation_id) +); + +CREATE INDEX artifact_sync_pending_recovery + ON artifact_sync_pending (earliest_at, community_id, operation_id); + +CREATE INDEX artifact_sync_pending_community + ON artifact_sync_pending (community_id, earliest_at, operation_id); diff --git a/schema/schema.sql b/schema/schema.sql index 710345e5275..825b248f7af 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -141,6 +141,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/storage_accounting_snapshots.sql \i tables/public/artifact_heads.sql \i tables/public/artifact_revisions.sql +\i tables/public/artifact_sync_pending.sql -- partitions \i partitions/events.sql diff --git a/schema/tables/public/artifact_sync_pending.sql b/schema/tables/public/artifact_sync_pending.sql new file mode 100644 index 00000000000..6a207e87d77 --- /dev/null +++ b/schema/tables/public/artifact_sync_pending.sql @@ -0,0 +1,12 @@ +CREATE TABLE artifact_sync_pending ( + community_id UUID NOT NULL, + operation_id UUID NOT NULL, + earliest_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (community_id, operation_id) +); + +CREATE INDEX artifact_sync_pending_recovery + ON artifact_sync_pending (earliest_at, community_id, operation_id); + +CREATE INDEX artifact_sync_pending_community + ON artifact_sync_pending (community_id, earliest_at, operation_id); From 3ea1f6bb1daf55fd70854972d1ede74141f5efb4 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:14:15 -0800 Subject: [PATCH 02/20] test(relay): cover artifact access certification Signed-off-by: Daniel Rouleau --- .../src/store/artifact_postgres_tests.rs | 107 ++++++++++++++++++ .../buzz-db/src/store/relay_admin_actions.rs | 7 ++ docs/nips/NIP-AR.md | 17 ++- 3 files changed, 126 insertions(+), 5 deletions(-) diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index 5001ed4d945..ac2f13624b5 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -64,6 +64,113 @@ async fn artifact_writes_record_one_certification_before_their_timestamps() { .unwrap(); assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn access_writers_record_certification_for_grants_and_removals() { + use crate::channel_members::{add_member, remove_member, MemberRole}; + let f = Fixture::new().await; + let owner = f.owner.public_key().to_bytes(); + let peer = f.peer.public_key().to_bytes(); + for _ in 0..2 { + add_member( + &f.db.pool, + f.community, + f.b, + &peer, + MemberRole::Member, + Some(&owner), + ) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + remove_member(&f.db.pool, f.community, f.b, &peer, &owner) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + } + let mut tx = + f.db.begin_event_write_transaction(f.community) + .await + .unwrap(); + crate::channel_members::lock_channel_memberships_in_transaction(&mut tx, &[f.b]) + .await + .unwrap() + .insert_auto_membership(f.b, &peer, &owner) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + crate::relay_admin_actions::deploy_kick_member(&f.db.pool, f.community, f.b, &peer, &owner) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + crate::dm::create_dm(&f.db.pool, f.community, &[&owner, &peer], &owner) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_writers_record_certification_for_creation_and_policy() { + use buzz_core::channel::{ChannelType, ChannelVisibility}; + let f = Fixture::new().await; + let owner = f.owner.public_key().to_bytes(); + crate::channel::create_channel( + &f.db.pool, + f.community, + "certified", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + &[], + false, + ) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + crate::channel::create_channel_with_id( + &f.db.pool, + f.community, + Uuid::new_v4(), + "certified-id", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + &[], + ) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + for updates in [ + crate::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + crate::channel::ChannelUpdate { + unlisted: Some(true), + ..Default::default() + }, + crate::channel::ChannelUpdate { + ttl_seconds: Some(Some(3600)), + ..Default::default() + }, + ] { + crate::channel::update_channel(&f.db.pool, f.community, f.a, updates) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); + } + crate::channel::soft_delete_channel(&f.db.pool, f.community, f.a) + .await + .unwrap(); + assert_eq!(f.db.recover_artifact_certifications().await.unwrap(), 1); +} impl Fixture { async fn new() -> Self { let pool = PgPool::connect(&crate::test_support::database_url()) diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 4ab1135f0af..7cc6fdca734 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -3905,6 +3905,13 @@ mod postgres_tests { matches!(r1, KickWithMarkerResult::Removed), "first kick must be Removed" ); + let pending: i64 = + sqlx::query_scalar("SELECT count(*) FROM artifact_sync_pending WHERE community_id=$1") + .bind(community_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(pending, 1); // step_marker must now be set. let rec = get_action(&pool, action_id) diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index ce829897a3b..421e8cab2c3 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -290,9 +290,15 @@ 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. +use database `clock_timestamp()`. A relay with durable certification records +pending work in the same transaction, before the first sync stamp. After +commit, it proves visibility within the 120-second overlap or requests a +policy reset. Recovery finishes pending work after a crash without a caller +retry. The reset and completion of certification commit together. + +Certification requires every serving writer to record pending work. Without +that coverage, the stamp-to-visibility bound is an operating assumption. +Clients need no additional request or response fields for certification. Pages order by `(change timestamp, source, stable identity)`. The source separates membership removals, head changes, grants, and move-source records. @@ -331,8 +337,9 @@ 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 +stamp, including transitions on empty channels. Certification also advances +this stamp when it cannot prove visibility within the overlap. Ordinary +artifact acceptance does 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"`. From 4fc917a24387b666dc8a57d5e17afc22fef75197 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:17:35 -0800 Subject: [PATCH 03/20] test(db): use deletion authority in certification test Signed-off-by: Daniel Rouleau --- .../src/store/artifact_certification_postgres_tests.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index f2fae8700ca..d473e43f0fc 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -150,11 +150,19 @@ async fn recovery_handles_clock_reversal_and_leaves_deleting_communities() { assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); assert!(state(&db, community).await.0 > initial.0); + let mut pending = begin(&db, community).await; + pending.record_artifact_change().await.unwrap(); + pending.commit().await.unwrap(); + let mut deletion = db.pool().begin().await.unwrap(); + sqlx::query("SELECT set_config('buzz.deletion_executor_community',$1,true),set_config('buzz.deletion_fence_generation','0',true)") + .bind(community.as_uuid().to_string()).execute(&mut *deletion).await.unwrap(); sqlx::query("UPDATE communities SET deletion_state='quiescing' WHERE id=$1") .bind(community.as_uuid()) - .execute(db.pool()) + .execute(&mut *deletion) .await .unwrap(); + deletion.commit().await.unwrap(); assert!(db.begin_event_write_transaction(community).await.is_err()); assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); + assert_eq!(state(&db, community).await.1, 1); } From cd83d5ceb9954410cc4badcdc1cbbc3758a759d8 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:21:12 -0800 Subject: [PATCH 04/20] test(relay): exercise certification after restart Signed-off-by: Daniel Rouleau --- .../tests/artifact_certification_e2e.rs | 256 ++++++++++++++++++ 1 file changed, 256 insertions(+) create mode 100644 crates/buzz-relay/tests/artifact_certification_e2e.rs diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs new file mode 100644 index 00000000000..1cd9b6d2a27 --- /dev/null +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -0,0 +1,256 @@ +//! Signed HTTP coverage against a relay process and an isolated database. + +use base64::{engine::general_purpose::STANDARD, Engine}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::{net::TcpListener, process::Command, time::Duration}; +use uuid::Uuid; + +struct Relay(std::process::Child); +impl Drop for Relay { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +async fn post( + client: &reqwest::Client, + base: &str, + keys: &Keys, + path: &str, + body: Value, +) -> (u16, Value) { + let url = format!("{base}{path}"); + let body = serde_json::to_vec(&body).unwrap(); + let auth = EventBuilder::new(Kind::HttpAuth, "") + .tags([ + Tag::parse(["u", url.as_str()]).unwrap(), + Tag::parse(["method", "POST"]).unwrap(), + Tag::parse(["payload", &hex::encode(Sha256::digest(&body))]).unwrap(), + Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(), + ]) + .sign_with_keys(keys) + .unwrap(); + let response = client + .post(url) + .header( + "authorization", + format!( + "Nostr {}", + STANDARD.encode(serde_json::to_vec(&auth).unwrap()) + ), + ) + .header("content-type", "application/json") + .body(body) + .send() + .await + .unwrap(); + let status = response.status().as_u16(); + let body = response.json().await.unwrap(); + (status, body) +} + +async fn publish(client: &reqwest::Client, base: &str, keys: &Keys, event: &Event) { + let (status, body) = post(client, base, keys, "/events", json!(event)).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(body["accepted"], true, "{body}"); +} + +#[tokio::test] +#[ignore = "requires an empty local PostgreSQL database and Redis"] +async fn relay_recovers_certification_after_restart_without_a_writer_retry() { + let database = std::env::var("BUZZ_CERTIFICATION_E2E_DATABASE_URL").unwrap(); + let pool = sqlx::PgPool::connect(&database).await.unwrap(); + let name: String = sqlx::query_scalar("SELECT current_database()") + .fetch_one(&pool) + .await + .unwrap(); + assert!( + name.starts_with("buzz_cert_"), + "requires an isolated test database" + ); + let empty: bool = sqlx::query_scalar("SELECT to_regclass('public.communities') IS NULL") + .fetch_one(&pool) + .await + .unwrap(); + assert!(empty, "requires an empty test database"); + let listeners: Vec<_> = (0..3) + .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) + .collect(); + let ports: Vec<_> = listeners + .iter() + .map(|s| s.local_addr().unwrap().port()) + .collect(); + drop(listeners); + let base = format!("http://127.0.0.1:{}", ports[0]); + let relay_keys = Keys::generate(); + let owner = Keys::generate(); + let log = tempfile::NamedTempFile::new().unwrap(); + let spawn = |enabled: bool| { + Relay( + Command::new(env!("CARGO_BIN_EXE_buzz-relay")) + .env("DATABASE_URL", &database) + .env("BUZZ_BIND_ADDR", format!("127.0.0.1:{}", ports[0])) + .env("BUZZ_HEALTH_PORT", ports[1].to_string()) + .env("BUZZ_METRICS_PORT", ports[2].to_string()) + .env("RELAY_URL", format!("ws://127.0.0.1:{}", ports[0])) + .env( + "BUZZ_RELAY_PRIVATE_KEY", + relay_keys.secret_key().to_secret_hex(), + ) + .env("RELAY_OWNER_PUBKEY", owner.public_key().to_hex()) + .env("BUZZ_REQUIRE_AUTH_TOKEN", "true") + .env("BUZZ_REQUIRE_RELAY_MEMBERSHIP", "false") + .env("BUZZ_AUTO_MIGRATE", "true") + .env("BUZZ_ARTIFACT_INCREMENTAL_ENABLED", enabled.to_string()) + .env("BUZZ_PUSH_ENABLED", "false") + .env("BUZZ_AUDIT_ENABLED", "false") + .stdout(log.reopen().unwrap()) + .stderr(log.reopen().unwrap()) + .spawn() + .unwrap(), + ) + }; + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .unwrap(); + let mut relay = spawn(true); + ready(&client, &base, &log).await; + let channel = Uuid::new_v4(); + let create = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["name", &format!("certification-{channel}")]).unwrap(), + Tag::parse(["visibility", "open"]).unwrap(), + Tag::parse(["channel_type", "stream"]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + publish(&client, &base, &owner, &create).await; + let query = json!({"artifact":"current","#type":["buzz.task"]}); + let (_, first) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert!(first["checkpoint"].is_string(), "{first}"); + let artifact = Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(45010), "task") + .tags([ + Tag::parse(["ar", "1"]).unwrap(), + Tag::parse(["d", &artifact.to_string()]).unwrap(), + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["type", "buzz.task"]).unwrap(), + Tag::parse(["title", "Certification"]).unwrap(), + Tag::parse(["op", "create"]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + publish(&client, &base, &owner, &event).await; + drop(relay); + + // The fixture models a committed write outside the overlap, with no caller. + let mut tx = pool.begin().await.unwrap(); + let community: Uuid = + sqlx::query_scalar("SELECT community_id FROM artifact_heads WHERE artifact_id=$1") + .bind(artifact) + .fetch_one(&mut *tx) + .await + .unwrap(); + sqlx::query("UPDATE artifact_heads SET modified_at=$1::timestamptz-interval '121 seconds' WHERE community_id=$2 AND artifact_id=$3") + .bind(first["startedAt"].as_str().unwrap()).bind(community).bind(artifact).execute(&mut *tx).await.unwrap(); + sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) VALUES($1,$2,$3::timestamptz-interval '121 seconds')") + .bind(community).bind(Uuid::new_v4()).bind(first["startedAt"].as_str().unwrap()).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + + relay = spawn(true); + ready(&client, &base, &log).await; + let mut reset = false; + for _ in 0..30 { + let (status, body) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query, "checkpoint":first["checkpoint"] + }]), + ) + .await; + if status == 409 { + assert_eq!(body["reason"], "policy_changed"); + reset = true; + break; + } + assert_eq!(status, 200, "{body}"); + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(reset, "recovery must reset without a writer retry"); + let (_, recovered) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); + let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; + assert_eq!(status, 200); + assert!(ordinary.to_string().contains(&event.id.to_hex())); + let remaining: i64 = + sqlx::query_scalar("SELECT count(*) FROM artifact_sync_pending WHERE community_id=$1") + .bind(community) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(remaining, 0); + drop(relay); + + let _relay = spawn(false); + ready(&client, &base, &log).await; + let (status, body) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert_eq!(status, 503); + assert_eq!(body["code"], "incremental_unavailable"); + let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; + assert_eq!(status, 200); + assert!(ordinary.to_string().contains(&event.id.to_hex())); + println!("PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads, gate off"); +} + +async fn ready(client: &reqwest::Client, base: &str, log: &tempfile::NamedTempFile) { + for _ in 0..150 { + if client + .get(format!("{base}/health")) + .send() + .await + .is_ok_and(|r| r.status().is_success()) + { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!( + "relay startup failed: {}", + std::fs::read_to_string(log.path()).unwrap() + ); +} From f07d9de23c7ae1f04b21eef4a537abe47f5502f5 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:29:57 -0800 Subject: [PATCH 05/20] fix(relay): bound and rotate certification recovery Signed-off-by: Daniel Rouleau --- .../src/store/artifact_certification.rs | 116 +++-- .../artifact_certification_postgres_tests.rs | 76 ++- .../sync_policy_postgres_tests.rs | 2 + .../buzz-relay/src/artifact_certification.rs | 44 +- .../tests/artifact_certification_e2e.rs | 466 +++++++++--------- docs/artifact-sync-deployment.md | 7 +- 6 files changed, 436 insertions(+), 275 deletions(-) diff --git a/crates/buzz-db/src/store/artifact_certification.rs b/crates/buzz-db/src/store/artifact_certification.rs index db7b14e01e2..552e2a822e7 100644 --- a/crates/buzz-db/src/store/artifact_certification.rs +++ b/crates/buzz-db/src/store/artifact_certification.rs @@ -2,6 +2,7 @@ use buzz_core::CommunityId; use sqlx::PgPool; +use std::time::Duration; use uuid::Uuid; use crate::{Db, Result}; @@ -15,23 +16,42 @@ pub(crate) async fn certify( community: CommunityId, operation: Option, ) -> Result { - let Some(mut tx) = crate::begin_community_sweep_transaction( - pool, - community, - crate::observability::WriterOperation::Maintenance, - ) - .await? - else { - return Ok(0); - }; - let completed = certify_pending(&mut tx, operation).await?; - tx.commit().await?; - Ok(completed) + let deadline = tokio::time::Instant::now() + Duration::from_millis(500); + tokio::time::timeout_at(deadline, async { + let mut connection = crate::observability::acquire_writer_until( + pool, + crate::observability::WriterOperation::Maintenance, + deadline, + ) + .await?; + // A cancelled commit has uncertain protocol state. Never reuse its connection. + connection.close_on_drop(); + sqlx::query("SET statement_timeout='400ms'") + .execute(&mut *connection) + .await?; + let tx = sqlx::Transaction::begin(connection, None).await?; + let mut tx = match crate::AdmittedTx::admit( + tx, + &crate::deletion::DeletionStore::new(pool.clone()), + community, + ) + .await + { + Ok(tx) => tx, + Err(crate::DbError::AccessDenied(_)) => return Ok(0), + Err(error) => return Err(error), + }; + let completed = certify_pending(&mut tx, operation).await?; + tx.commit().await?; + Ok(completed) + }) + .await + .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))? } async fn certify_pending(tx: &mut crate::AdmittedTx, operation: Option) -> Result { let community = tx.community(); - sqlx::query("SET LOCAL statement_timeout='2s'") + sqlx::query("SET LOCAL statement_timeout='400ms'") .execute(tx.conn()) .await?; let pending: Vec<(Uuid, chrono::DateTime)> = sqlx::query_as( @@ -70,28 +90,64 @@ async fn certify_pending(tx: &mut crate::AdmittedTx, operation: Option) -> } impl Db { - /// Recover one bounded batch of committed artifact certifications. - /// Multiple relay instances can run this method at the same time. + /// Recover one bounded batch in the first active community with pending work. pub async fn recover_artifact_certifications(&self) -> Result { - let communities: Vec = sqlx::query_scalar( - "SELECT DISTINCT community_id FROM \ - (SELECT p.community_id FROM artifact_sync_pending p JOIN communities c ON c.id=p.community_id WHERE c.deletion_state='active' AND c.deleted_at IS NULL ORDER BY p.earliest_at,p.community_id,p.operation_id LIMIT $1) p", + Ok(self.recover_artifact_certifications_after(None).await?.0) + } + + /// Recover at most 100 records in the next community after `after`. + /// The returned UUID advances the sweep even when that community is blocked. + /// `None` means that the sweep reaches its end and can start again. + pub async fn recover_artifact_certifications_after( + &self, + after: Option, + ) -> Result<(usize, Option)> { + let deadline = tokio::time::Instant::now() + Duration::from_millis(500); + let mut connection = crate::observability::acquire_writer_until( + &self.pool, + crate::observability::WriterOperation::Maintenance, + deadline, ) - .bind(BATCH) - .fetch_all(&self.pool) .await?; - let mut completed = 0; - for id in communities { - let community = CommunityId::from_uuid(id); - match certify(&self.pool, community, None).await { - Ok(count) => completed += count, - Err(error) => { - tracing::warn!(%error, community_id = %community, - "artifact certification requires recovery"); - } + let next = tokio::time::timeout_at(deadline, async { + let mut tx = sqlx::Connection::begin(&mut *connection).await?; + sqlx::query("SET LOCAL statement_timeout='400ms'") + .execute(&mut *tx) + .await?; + let next = sqlx::query_scalar::<_, Uuid>( + "SELECT DISTINCT p.community_id FROM artifact_sync_pending p \ + JOIN communities c ON c.id=p.community_id \ + WHERE c.deletion_state='active' AND c.deleted_at IS NULL \ + AND ($1::uuid IS NULL OR p.community_id>$1) \ + ORDER BY p.community_id LIMIT 1", + ) + .bind(after) + .fetch_optional(&mut *tx) + .await?; + tx.commit().await?; + Ok::<_, sqlx::Error>(next) + }) + .await; + let next = match next { + Ok(result) => result?, + Err(_) => { + drop(connection.detach()); + return Err(sqlx::Error::PoolTimedOut.into()); + } + }; + drop(connection); + let Some(id) = next else { + return Ok((0, None)); + }; + let community = CommunityId::from_uuid(id); + match certify(&self.pool, community, None).await { + Ok(count) => Ok((count, Some(id))), + Err(error) => { + tracing::warn!(%error, community_id = %community, + "artifact certification requires recovery"); + Ok((0, Some(id))) } } - Ok(completed) } } diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index d473e43f0fc..b9f8b1e91a4 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -46,8 +46,8 @@ async fn rollback_and_lost_commit_reply_leave_correct_recovery_work() { let mut tx = begin(&db, community).await; tx.record_artifact_change().await.unwrap(); - tx.record_artifact_change().await.unwrap(); age(&mut tx, 121).await; + tx.record_artifact_change().await.unwrap(); assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); // The writer can lose its commit reply and never return to retry. tx.commit().await.unwrap(); @@ -166,3 +166,77 @@ async fn recovery_handles_clock_reversal_and_leaves_deleting_communities() { assert_eq!(db.recover_artifact_certifications().await.unwrap(), 0); assert_eq!(state(&db, community).await.1, 1); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn blocked_community_does_not_exclude_other_pending_work() { + let (db, a) = fixture().await; + let (_, b) = fixture().await; + let mut communities = [a, b]; + communities.sort_by_key(|c| *c.as_uuid()); + for community in communities { + let mut tx = begin(&db, community).await; + sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) SELECT $1,gen_random_uuid(),clock_timestamp()-interval '121 seconds' FROM generate_series(1,101)") + .bind(community.as_uuid()).execute(tx.conn()).await.unwrap(); + tx.commit().await.unwrap(); + } + let mut blocker = db.pool().begin().await.unwrap(); + sqlx::query("SELECT id FROM communities WHERE id=$1 FOR NO KEY UPDATE") + .bind(communities[0].as_uuid()) + .execute(&mut *blocker) + .await + .unwrap(); + let (completed, after) = db + .recover_artifact_certifications_after(None) + .await + .unwrap(); + assert_eq!(completed, 0); + assert_eq!(after, Some(*communities[0].as_uuid())); + let (completed, after) = db + .recover_artifact_certifications_after(after) + .await + .unwrap(); + assert_eq!(completed, 100); + assert_eq!(after, Some(*communities[1].as_uuid())); + assert_eq!(state(&db, communities[0]).await.1, 101); + assert_eq!(state(&db, communities[1]).await.1, 1); + blocker.rollback().await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn certification_closes_a_connection_after_a_blocked_reset() { + let (db, community) = fixture().await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + age(&mut tx, 121).await; + tx.commit().await.unwrap(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + let mut blocker = db.pool().begin().await.unwrap(); + sqlx::query("SELECT id FROM communities WHERE id=$1 FOR NO KEY UPDATE") + .bind(community.as_uuid()) + .execute(&mut *blocker) + .await + .unwrap(); + let start = tokio::time::Instant::now(); + assert!(certify(&pool, community, None).await.is_err()); + assert!(start.elapsed() < std::time::Duration::from_secs(1)); + let next: i32 = tokio::time::timeout( + std::time::Duration::from_secs(1), + sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&pool), + ) + .await + .unwrap() + .unwrap(); + assert_ne!(pid, next); + assert_eq!(state(&db, community).await.1, 1); + blocker.rollback().await.unwrap(); +} 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 index cd93f169d7e..bfc9cc98d9f 100644 --- 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 @@ -5,6 +5,8 @@ use super::*; async fn durable_certification_resets_existing_checkpoints_after_the_overlap() { let _guard = SERIAL.lock().await; let f = fixture().await; + sqlx::query("UPDATE channel_members SET sync_modified_at=clock_timestamp()-interval '1 day' WHERE community_id=$1") + .bind(f.community.as_uuid()).execute(&f.pool).await.unwrap(); let first = sync(&f, None).await; let id = Uuid::new_v4(); let event = f.revision(id, "create", None); diff --git a/crates/buzz-relay/src/artifact_certification.rs b/crates/buzz-relay/src/artifact_certification.rs index ad70ddc8a29..28ad1543c3c 100644 --- a/crates/buzz-relay/src/artifact_certification.rs +++ b/crates/buzz-relay/src/artifact_certification.rs @@ -8,23 +8,37 @@ use crate::state::AppState; pub async fn run(state: Arc) { let mut interval = tokio::time::interval(Duration::from_secs(1)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut after = None; loop { interval.tick().await; - if state - .shutting_down - .load(std::sync::atomic::Ordering::Acquire) - { - return; - } - match tokio::time::timeout( - Duration::from_secs(5), - state.db.recover_artifact_certifications(), - ) - .await - { - Ok(Ok(_)) => {} - Ok(Err(error)) => tracing::warn!(%error, "artifact certification recovery failed"), - Err(_) => tracing::warn!("artifact certification recovery reached its time limit"), + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let mut progress = false; + for _ in 0..100 { + if state + .shutting_down + .load(std::sync::atomic::Ordering::Acquire) + { + return; + } + if tokio::time::Instant::now() >= deadline { + break; + } + match state.db.recover_artifact_certifications_after(after).await { + Ok((completed, next)) => { + after = next; + progress |= completed > 0; + if next.is_none() { + if !progress { + break; + } + progress = false; + } + } + Err(error) => { + tracing::warn!(%error, "artifact certification recovery failed"); + break; + } + } } } } diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs index 1cd9b6d2a27..1fe33bf8547 100644 --- a/crates/buzz-relay/tests/artifact_certification_e2e.rs +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -1,256 +1,266 @@ //! Signed HTTP coverage against a relay process and an isolated database. -use base64::{engine::general_purpose::STANDARD, Engine}; -use nostr::{Event, EventBuilder, Keys, Kind, Tag}; -use serde_json::{json, Value}; -use sha2::{Digest, Sha256}; -use std::{net::TcpListener, process::Command, time::Duration}; -use uuid::Uuid; +mod external_infra_process { + use base64::{engine::general_purpose::STANDARD, Engine}; + use nostr::{Event, EventBuilder, Keys, Kind, Tag}; + use serde_json::{json, Value}; + use sha2::{Digest, Sha256}; + use std::{net::TcpListener, process::Command, time::Duration}; + use uuid::Uuid; -struct Relay(std::process::Child); -impl Drop for Relay { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); + struct Relay(std::process::Child); + impl Drop for Relay { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } } -} -async fn post( - client: &reqwest::Client, - base: &str, - keys: &Keys, - path: &str, - body: Value, -) -> (u16, Value) { - let url = format!("{base}{path}"); - let body = serde_json::to_vec(&body).unwrap(); - let auth = EventBuilder::new(Kind::HttpAuth, "") - .tags([ - Tag::parse(["u", url.as_str()]).unwrap(), - Tag::parse(["method", "POST"]).unwrap(), - Tag::parse(["payload", &hex::encode(Sha256::digest(&body))]).unwrap(), - Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(), - ]) - .sign_with_keys(keys) - .unwrap(); - let response = client - .post(url) - .header( - "authorization", - format!( - "Nostr {}", - STANDARD.encode(serde_json::to_vec(&auth).unwrap()) - ), - ) - .header("content-type", "application/json") - .body(body) - .send() - .await - .unwrap(); - let status = response.status().as_u16(); - let body = response.json().await.unwrap(); - (status, body) -} - -async fn publish(client: &reqwest::Client, base: &str, keys: &Keys, event: &Event) { - let (status, body) = post(client, base, keys, "/events", json!(event)).await; - assert_eq!(status, 200, "{body}"); - assert_eq!(body["accepted"], true, "{body}"); -} + async fn post( + client: &reqwest::Client, + base: &str, + keys: &Keys, + path: &str, + body: Value, + ) -> (u16, Value) { + let url = format!("{base}{path}"); + let body = serde_json::to_vec(&body).unwrap(); + let auth = EventBuilder::new(Kind::HttpAuth, "") + .tags([ + Tag::parse(["u", url.as_str()]).unwrap(), + Tag::parse(["method", "POST"]).unwrap(), + Tag::parse(["payload", &hex::encode(Sha256::digest(&body))]).unwrap(), + Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(), + ]) + .sign_with_keys(keys) + .unwrap(); + let response = client + .post(url) + .header( + "authorization", + format!( + "Nostr {}", + STANDARD.encode(serde_json::to_vec(&auth).unwrap()) + ), + ) + .header("content-type", "application/json") + .body(body) + .send() + .await + .unwrap(); + let status = response.status().as_u16(); + let body = response.json().await.unwrap(); + (status, body) + } -#[tokio::test] -#[ignore = "requires an empty local PostgreSQL database and Redis"] -async fn relay_recovers_certification_after_restart_without_a_writer_retry() { - let database = std::env::var("BUZZ_CERTIFICATION_E2E_DATABASE_URL").unwrap(); - let pool = sqlx::PgPool::connect(&database).await.unwrap(); - let name: String = sqlx::query_scalar("SELECT current_database()") - .fetch_one(&pool) - .await - .unwrap(); - assert!( - name.starts_with("buzz_cert_"), - "requires an isolated test database" - ); - let empty: bool = sqlx::query_scalar("SELECT to_regclass('public.communities') IS NULL") - .fetch_one(&pool) - .await - .unwrap(); - assert!(empty, "requires an empty test database"); - let listeners: Vec<_> = (0..3) - .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) - .collect(); - let ports: Vec<_> = listeners - .iter() - .map(|s| s.local_addr().unwrap().port()) - .collect(); - drop(listeners); - let base = format!("http://127.0.0.1:{}", ports[0]); - let relay_keys = Keys::generate(); - let owner = Keys::generate(); - let log = tempfile::NamedTempFile::new().unwrap(); - let spawn = |enabled: bool| { - Relay( - Command::new(env!("CARGO_BIN_EXE_buzz-relay")) - .env("DATABASE_URL", &database) - .env("BUZZ_BIND_ADDR", format!("127.0.0.1:{}", ports[0])) - .env("BUZZ_HEALTH_PORT", ports[1].to_string()) - .env("BUZZ_METRICS_PORT", ports[2].to_string()) - .env("RELAY_URL", format!("ws://127.0.0.1:{}", ports[0])) - .env( - "BUZZ_RELAY_PRIVATE_KEY", - relay_keys.secret_key().to_secret_hex(), - ) - .env("RELAY_OWNER_PUBKEY", owner.public_key().to_hex()) - .env("BUZZ_REQUIRE_AUTH_TOKEN", "true") - .env("BUZZ_REQUIRE_RELAY_MEMBERSHIP", "false") - .env("BUZZ_AUTO_MIGRATE", "true") - .env("BUZZ_ARTIFACT_INCREMENTAL_ENABLED", enabled.to_string()) - .env("BUZZ_PUSH_ENABLED", "false") - .env("BUZZ_AUDIT_ENABLED", "false") - .stdout(log.reopen().unwrap()) - .stderr(log.reopen().unwrap()) - .spawn() - .unwrap(), - ) - }; - let client = reqwest::Client::builder() - .timeout(Duration::from_secs(5)) - .build() - .unwrap(); - let mut relay = spawn(true); - ready(&client, &base, &log).await; - let channel = Uuid::new_v4(); - let create = EventBuilder::new(Kind::Custom(9007), "") - .tags([ - Tag::parse(["h", &channel.to_string()]).unwrap(), - Tag::parse(["name", &format!("certification-{channel}")]).unwrap(), - Tag::parse(["visibility", "open"]).unwrap(), - Tag::parse(["channel_type", "stream"]).unwrap(), - ]) - .sign_with_keys(&owner) - .unwrap(); - publish(&client, &base, &owner, &create).await; - let query = json!({"artifact":"current","#type":["buzz.task"]}); - let (_, first) = post( - &client, - &base, - &owner, - "/query", - json!([{ - "artifact":"incremental", "query":query - }]), - ) - .await; - assert!(first["checkpoint"].is_string(), "{first}"); - let artifact = Uuid::new_v4(); - let event = EventBuilder::new(Kind::Custom(45010), "task") - .tags([ - Tag::parse(["ar", "1"]).unwrap(), - Tag::parse(["d", &artifact.to_string()]).unwrap(), - Tag::parse(["h", &channel.to_string()]).unwrap(), - Tag::parse(["type", "buzz.task"]).unwrap(), - Tag::parse(["title", "Certification"]).unwrap(), - Tag::parse(["op", "create"]).unwrap(), - ]) - .sign_with_keys(&owner) - .unwrap(); - publish(&client, &base, &owner, &event).await; - drop(relay); + async fn publish(client: &reqwest::Client, base: &str, keys: &Keys, event: &Event) { + let (status, body) = post(client, base, keys, "/events", json!(event)).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(body["accepted"], true, "{body}"); + } - // The fixture models a committed write outside the overlap, with no caller. - let mut tx = pool.begin().await.unwrap(); - let community: Uuid = - sqlx::query_scalar("SELECT community_id FROM artifact_heads WHERE artifact_id=$1") - .bind(artifact) - .fetch_one(&mut *tx) + #[tokio::test] + #[ignore = "requires a local relay network process, empty PostgreSQL database, and Redis"] + async fn relay_recovers_certification_after_restart_without_a_writer_retry() { + let database = std::env::var("BUZZ_CERTIFICATION_E2E_DATABASE_URL").unwrap(); + let pool = sqlx::PgPool::connect(&database).await.unwrap(); + let name: String = sqlx::query_scalar("SELECT current_database()") + .fetch_one(&pool) + .await + .unwrap(); + assert!( + name.starts_with("buzz_cert_"), + "requires an isolated test database" + ); + let empty: bool = sqlx::query_scalar("SELECT to_regclass('public.communities') IS NULL") + .fetch_one(&pool) .await .unwrap(); - sqlx::query("UPDATE artifact_heads SET modified_at=$1::timestamptz-interval '121 seconds' WHERE community_id=$2 AND artifact_id=$3") + assert!(empty, "requires an empty test database"); + let listeners: Vec<_> = (0..3) + .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) + .collect(); + let ports: Vec<_> = listeners + .iter() + .map(|s| s.local_addr().unwrap().port()) + .collect(); + drop(listeners); + let base = format!("http://127.0.0.1:{}", ports[0]); + let relay_keys = Keys::generate(); + let owner = Keys::generate(); + let log = tempfile::NamedTempFile::new().unwrap(); + let spawn = |enabled: bool| { + Relay( + Command::new(env!("CARGO_BIN_EXE_buzz-relay")) + .env("DATABASE_URL", &database) + .env("BUZZ_BIND_ADDR", format!("127.0.0.1:{}", ports[0])) + .env("BUZZ_HEALTH_PORT", ports[1].to_string()) + .env("BUZZ_METRICS_PORT", ports[2].to_string()) + .env("RELAY_URL", format!("ws://127.0.0.1:{}", ports[0])) + .env( + "BUZZ_RELAY_PRIVATE_KEY", + relay_keys.secret_key().to_secret_hex(), + ) + .env("RELAY_OWNER_PUBKEY", owner.public_key().to_hex()) + .env("BUZZ_REQUIRE_AUTH_TOKEN", "true") + .env("BUZZ_REQUIRE_RELAY_MEMBERSHIP", "false") + .env("BUZZ_AUTO_MIGRATE", "true") + .env("BUZZ_ARTIFACT_INCREMENTAL_ENABLED", enabled.to_string()) + .env("BUZZ_PUSH_ENABLED", "false") + .env("BUZZ_AUDIT_ENABLED", "false") + .stdout(log.reopen().unwrap()) + .stderr(log.reopen().unwrap()) + .spawn() + .unwrap(), + ) + }; + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .unwrap(); + let mut relay = spawn(true); + ready(&client, &base, &log).await; + let channel = Uuid::new_v4(); + let create = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["name", &format!("certification-{channel}")]).unwrap(), + Tag::parse(["visibility", "open"]).unwrap(), + Tag::parse(["channel_type", "stream"]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + publish(&client, &base, &owner, &create).await; + let query = json!({"artifact":"current","#type":["buzz.task"]}); + let (_, first) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert!(first["checkpoint"].is_string(), "{first}"); + let artifact = Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(45010), "task") + .tags([ + Tag::parse(["ar", "1"]).unwrap(), + Tag::parse(["d", &artifact.to_string()]).unwrap(), + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["type", "buzz.task"]).unwrap(), + Tag::parse(["title", "Certification"]).unwrap(), + Tag::parse(["op", "create"]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + publish(&client, &base, &owner, &event).await; + drop(relay); + + // The fixture models a committed write outside the overlap, with no caller. + let mut tx = pool.begin().await.unwrap(); + let community: Uuid = + sqlx::query_scalar("SELECT community_id FROM artifact_heads WHERE artifact_id=$1") + .bind(artifact) + .fetch_one(&mut *tx) + .await + .unwrap(); + sqlx::query("UPDATE artifact_heads SET modified_at=$1::timestamptz-interval '121 seconds' WHERE community_id=$2 AND artifact_id=$3") .bind(first["startedAt"].as_str().unwrap()).bind(community).bind(artifact).execute(&mut *tx).await.unwrap(); - sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) VALUES($1,$2,$3::timestamptz-interval '121 seconds')") - .bind(community).bind(Uuid::new_v4()).bind(first["startedAt"].as_str().unwrap()).execute(&mut *tx).await.unwrap(); - tx.commit().await.unwrap(); + sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) SELECT $1,gen_random_uuid(),$2::timestamptz-interval '121 seconds' FROM generate_series(1,500)") + .bind(community).bind(first["startedAt"].as_str().unwrap()).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); - relay = spawn(true); - ready(&client, &base, &log).await; - let mut reset = false; - for _ in 0..30 { - let (status, body) = post( + relay = spawn(true); + ready(&client, &base, &log).await; + let mut reset = false; + for _ in 0..30 { + let (status, body) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query, "checkpoint":first["checkpoint"] + }]), + ) + .await; + if status == 409 { + assert_eq!(body["reason"], "policy_changed"); + reset = true; + break; + } + assert_eq!(status, 200, "{body}"); + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(reset, "recovery must reset without a writer retry"); + let (_, recovered) = post( &client, &base, &owner, "/query", json!([{ - "artifact":"incremental", "query":query, "checkpoint":first["checkpoint"] + "artifact":"incremental", "query":query }]), ) .await; - if status == 409 { - assert_eq!(body["reason"], "policy_changed"); - reset = true; - break; - } - assert_eq!(status, 200, "{body}"); - tokio::time::sleep(Duration::from_millis(100)).await; - } - assert!(reset, "recovery must reset without a writer retry"); - let (_, recovered) = post( - &client, - &base, - &owner, - "/query", - json!([{ - "artifact":"incremental", "query":query - }]), - ) - .await; - assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); - let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; - assert_eq!(status, 200); - assert!(ordinary.to_string().contains(&event.id.to_hex())); - let remaining: i64 = - sqlx::query_scalar("SELECT count(*) FROM artifact_sync_pending WHERE community_id=$1") + assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); + let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; + assert_eq!(status, 200); + assert!(ordinary.to_string().contains(&event.id.to_hex())); + let mut remaining = 500; + for _ in 0..15 { + remaining = sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM artifact_sync_pending WHERE community_id=$1", + ) .bind(community) .fetch_one(&pool) .await .unwrap(); - assert_eq!(remaining, 0); - drop(relay); + if remaining == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert_eq!(remaining, 0); + drop(relay); - let _relay = spawn(false); - ready(&client, &base, &log).await; - let (status, body) = post( - &client, - &base, - &owner, - "/query", - json!([{ - "artifact":"incremental", "query":query - }]), - ) - .await; - assert_eq!(status, 503); - assert_eq!(body["code"], "incremental_unavailable"); - let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; - assert_eq!(status, 200); - assert!(ordinary.to_string().contains(&event.id.to_hex())); - println!("PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads, gate off"); -} + let _relay = spawn(false); + ready(&client, &base, &log).await; + let (status, body) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert_eq!(status, 503); + assert_eq!(body["code"], "incremental_unavailable"); + let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; + assert_eq!(status, 200); + assert!(ordinary.to_string().contains(&event.id.to_hex())); + println!("PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads, gate off"); + } -async fn ready(client: &reqwest::Client, base: &str, log: &tempfile::NamedTempFile) { - for _ in 0..150 { - if client - .get(format!("{base}/health")) - .send() - .await - .is_ok_and(|r| r.status().is_success()) - { - return; + async fn ready(client: &reqwest::Client, base: &str, log: &tempfile::NamedTempFile) { + for _ in 0..150 { + if client + .get(format!("{base}/health")) + .send() + .await + .is_ok_and(|r| r.status().is_success()) + { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; } - tokio::time::sleep(Duration::from_millis(100)).await; + panic!( + "relay startup failed: {}", + std::fs::read_to_string(log.path()).unwrap() + ); } - panic!( - "relay startup failed: {}", - std::fs::read_to_string(log.path()).unwrap() - ); } diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index a187a81a459..9e843326d0c 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -84,7 +84,12 @@ Rollback removes both the changes and the record. An uncertain commit leaves both committed or neither committed. Recovery does not require a caller retry. Every relay starts a recovery worker, even when the incremental gate is off. -The worker reads bounded batches. It locks pending rows with `SKIP LOCKED`. +The worker reads bounded batches. It rotates across communities, including +after a failed attempt. It locks pending rows with `SKIP LOCKED`. Each tick +has a five-second work budget and at most 100 batches. Each batch has at most +100 records. A blocked community cannot exclude other communities. Each +operation has a short deadline. Recovery closes certification connections +instead of returning an uncertain connection to the pool. After it observes a committed record, it reads database time. If less than 120 seconds separate that time from the early timestamp, the overlap covers visibility of the write. A clock reversal or a delay of at least 120 seconds From 47a52192cb99f00558576656464bfee43865a48a Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:33:16 -0800 Subject: [PATCH 06/20] test(db): drop the acknowledgement after commit Signed-off-by: Daniel Rouleau --- .../artifact_certification_postgres_tests.rs | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index b9f8b1e91a4..dff6cb244c3 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -240,3 +240,85 @@ async fn certification_closes_a_connection_after_a_blocked_reset() { assert_eq!(state(&db, community).await.1, 1); blocker.rollback().await.unwrap(); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn lost_commit_acknowledgement_recovers_the_committed_operation() { + use sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgSslMode}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + + let (db, community) = fixture().await; + let initial = state(&db, community).await; + let options: PgConnectOptions = crate::test_support::database_url().parse().unwrap(); + let upstream = (options.get_host().to_owned(), options.get_port()); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let proxy = tokio::spawn(async move { + let (client, _) = listener.accept().await.unwrap(); + let server = TcpStream::connect(upstream).await.unwrap(); + let (mut client_read, mut client_write) = client.into_split(); + let (mut server_read, mut server_write) = server.into_split(); + let forward = + tokio::spawn(async move { tokio::io::copy(&mut client_read, &mut server_write).await }); + loop { + let mut header = [0u8; 5]; + server_read.read_exact(&mut header).await.unwrap(); + let length = u32::from_be_bytes(header[1..].try_into().unwrap()) as usize; + assert!((4..=1_048_576).contains(&length)); + let mut body = vec![0; length - 4]; + server_read.read_exact(&mut body).await.unwrap(); + if header[0] == b'C' && body == b"COMMIT\0" { + forward.abort(); + let _ = forward.await; + return; + } + client_write.write_all(&header).await.unwrap(); + client_write.write_all(&body).await.unwrap(); + } + }); + let pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + options + .host("127.0.0.1") + .port(port) + .ssl_mode(PgSslMode::Disable), + ) + .await + .unwrap(); + let writer = Db::from_pool(pool); + let mut tx = begin(&writer, community).await; + tx.record_artifact_change().await.unwrap(); + age(&mut tx, 121).await; + let operation: Uuid = + sqlx::query_scalar("SELECT operation_id FROM artifact_sync_pending WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(tx.conn()) + .await + .unwrap(); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(3), tx.commit()) + .await + .unwrap() + .is_err() + ); + proxy.await.unwrap(); + assert_eq!(state(&db, community).await, (initial.0, 1)); + assert_eq!( + certify(db.pool(), community, Some(operation)) + .await + .unwrap(), + 1 + ); + let completed = state(&db, community).await; + assert!(completed.0 > initial.0); + assert_eq!(completed.1, 0); + assert_eq!( + certify(db.pool(), community, Some(operation)) + .await + .unwrap(), + 0 + ); + assert_eq!(state(&db, community).await, completed); +} From add9862d816e1328c5f6c23d566b5d2e2f8bc19e Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:38:15 -0800 Subject: [PATCH 07/20] test(relay): isolate artifact recovery from git storage Signed-off-by: Daniel Rouleau --- .../tests/artifact_certification_e2e.rs | 30 ++++++++++--------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs index 1fe33bf8547..2e4bfb168a0 100644 --- a/crates/buzz-relay/tests/artifact_certification_e2e.rs +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -90,6 +90,7 @@ mod external_infra_process { let owner = Keys::generate(); let log = tempfile::NamedTempFile::new().unwrap(); let spawn = |enabled: bool| { + let output = log.reopen().unwrap(); Relay( Command::new(env!("CARGO_BIN_EXE_buzz-relay")) .env("DATABASE_URL", &database) @@ -108,6 +109,7 @@ mod external_infra_process { .env("BUZZ_ARTIFACT_INCREMENTAL_ENABLED", enabled.to_string()) .env("BUZZ_PUSH_ENABLED", "false") .env("BUZZ_AUDIT_ENABLED", "false") + .env("BUZZ_GIT_CONFORMANCE_PROBE", "false") .stdout(log.reopen().unwrap()) .stderr(log.reopen().unwrap()) .spawn() @@ -195,20 +197,6 @@ mod external_infra_process { tokio::time::sleep(Duration::from_millis(100)).await; } assert!(reset, "recovery must reset without a writer retry"); - let (_, recovered) = post( - &client, - &base, - &owner, - "/query", - json!([{ - "artifact":"incremental", "query":query - }]), - ) - .await; - assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); - let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; - assert_eq!(status, 200); - assert!(ordinary.to_string().contains(&event.id.to_hex())); let mut remaining = 500; for _ in 0..15 { remaining = sqlx::query_scalar::<_, i64>( @@ -224,6 +212,20 @@ mod external_infra_process { tokio::time::sleep(Duration::from_millis(100)).await; } assert_eq!(remaining, 0); + let (_, recovered) = post( + &client, + &base, + &owner, + "/query", + json!([{ + "artifact":"incremental", "query":query + }]), + ) + .await; + assert_eq!(recovered["changes"][0]["event"]["id"], event.id.to_hex()); + let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; + assert_eq!(status, 200); + assert!(ordinary.to_string().contains(&event.id.to_hex())); drop(relay); let _relay = spawn(false); From 4c684fa7cacd55f0b1a2bfaa09a01e60c7a167bc Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:47:44 -0800 Subject: [PATCH 08/20] test(relay): share the process log handle Signed-off-by: Daniel Rouleau --- crates/buzz-relay/tests/artifact_certification_e2e.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs index 2e4bfb168a0..4540a030fbd 100644 --- a/crates/buzz-relay/tests/artifact_certification_e2e.rs +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -110,8 +110,8 @@ mod external_infra_process { .env("BUZZ_PUSH_ENABLED", "false") .env("BUZZ_AUDIT_ENABLED", "false") .env("BUZZ_GIT_CONFORMANCE_PROBE", "false") - .stdout(log.reopen().unwrap()) - .stderr(log.reopen().unwrap()) + .stdout(output.try_clone().unwrap()) + .stderr(output) .spawn() .unwrap(), ) From e37a75085e76f19bca45c150e1d29d88a14a6b80 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Mon, 2 Nov 2026 08:52:24 -0800 Subject: [PATCH 09/20] refactor(db): keep the certification indexes scoped Signed-off-by: Daniel Rouleau --- docs/artifact-sync-deployment.md | 3 ++- migrations/0081_artifact_sync_certification.sql | 3 --- schema/tables/public/artifact_sync_pending.sql | 3 --- 3 files changed, 2 insertions(+), 7 deletions(-) diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index 9e843326d0c..bed57e40d46 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -73,7 +73,8 @@ are tracked separately. Migration `0081` adds `artifact_sync_pending`. It is an expand migration. It adds no trigger, function, foreign key, or backfill. Old relays ignore this -table. New relays require it before they accept relevant writes. With startup +table for serving writes. Their deletion catalog fails closed until the new +relay deploys. New relays require the table before they accept relevant writes. With startup migrations enabled, the older binary still has the SQLx `VersionMissing` restriction described in `docs/database.md`. diff --git a/migrations/0081_artifact_sync_certification.sql b/migrations/0081_artifact_sync_certification.sql index 9361a2e25f9..dedef560b94 100644 --- a/migrations/0081_artifact_sync_certification.sql +++ b/migrations/0081_artifact_sync_certification.sql @@ -8,8 +8,5 @@ CREATE TABLE artifact_sync_pending ( PRIMARY KEY (community_id, operation_id) ); -CREATE INDEX artifact_sync_pending_recovery - ON artifact_sync_pending (earliest_at, community_id, operation_id); - CREATE INDEX artifact_sync_pending_community ON artifact_sync_pending (community_id, earliest_at, operation_id); diff --git a/schema/tables/public/artifact_sync_pending.sql b/schema/tables/public/artifact_sync_pending.sql index 6a207e87d77..08688be294c 100644 --- a/schema/tables/public/artifact_sync_pending.sql +++ b/schema/tables/public/artifact_sync_pending.sql @@ -5,8 +5,5 @@ CREATE TABLE artifact_sync_pending ( PRIMARY KEY (community_id, operation_id) ); -CREATE INDEX artifact_sync_pending_recovery - ON artifact_sync_pending (earliest_at, community_id, operation_id); - CREATE INDEX artifact_sync_pending_community ON artifact_sync_pending (community_id, earliest_at, operation_id); From 4e811df0bb619d2bc4cc101741e01667666ad4f4 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:09:21 -0800 Subject: [PATCH 10/20] test(db): measure artifact write latency Signed-off-by: Daniel Rouleau --- .../src/store/artifact_postgres_tests.rs | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index ac2f13624b5..e4bb54d9b41 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -979,3 +979,45 @@ async fn sync_snapshot_pages_are_one_cut_and_head_includes_deletions() { .unwrap() .is_none()); } + +#[tokio::test] +#[ignore = "requires Postgres; performance measurement"] +async fn measure_artifact_write_latency() { + use std::{sync::Arc, time::Instant}; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(64) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let fixture = Arc::new(Fixture::on(Db::from_pool(pool)).await); + for concurrency in [1, 8, 32] { + for repetition in 0..3 { + let mut workers = tokio::task::JoinSet::new(); + let start = Instant::now(); + for _ in 0..concurrency { + let f = fixture.clone(); + workers.spawn(async move { + let mut samples = Vec::new(); + for _ in 0..32 { + let event = + f.revision(Uuid::new_v4(), "create", f.a, None, &f.owner, vec![]); + let start = Instant::now(); + assert!(matches!( + f.accept(&event, None).await, + ArtifactOutcome::Accepted(_) + )); + samples.push(start.elapsed().as_secs_f64() * 1000.0); + } + samples + }); + } + let mut samples = Vec::new(); + while let Some(result) = workers.join_next().await { + samples.extend(result.unwrap()); + } + let elapsed = start.elapsed().as_secs_f64(); + samples.sort_by(f64::total_cmp); + println!("PERF write concurrency={concurrency} repetition={repetition} count={} elapsed_s={elapsed:.4} ops_s={:.1} p50_ms={:.3} p95_ms={:.3} p99_ms={:.3}", samples.len(), samples.len() as f64 / elapsed, samples[samples.len()/2], samples[samples.len()*95/100], samples[samples.len()*99/100]); + } + } +} From 0d470068d00900b2524f189b95b80ad332be3f8e Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:09:54 -0800 Subject: [PATCH 11/20] perf(db): batch durable artifact certification Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/runtime/admitted_tx.rs | 38 ++- crates/buzz-db/src/store/artifact.rs | 23 +- .../src/store/artifact_certification.rs | 241 +++++++++++------- .../artifact_certification_postgres_tests.rs | 153 ++++++++++- crates/buzz-db/src/store/channel.rs | 102 ++++---- crates/buzz-db/src/store/channel_members.rs | 66 ++--- crates/buzz-db/src/store/dm.rs | 21 +- crates/buzz-db/src/store/event.rs | 15 +- .../buzz-db/src/store/relay_admin_actions.rs | 44 ++-- docs/artifact-sync-deployment.md | 13 +- docs/nips/NIP-AR.md | 3 +- 11 files changed, 477 insertions(+), 242 deletions(-) diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index d646d8b827b..aa7f6401bac 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -134,14 +134,40 @@ impl AdmittedTx { } } - /// Record work before the first change that affects incremental artifact reads. - pub(crate) async fn record_artifact_change(&mut self) -> Result<()> { - if self.artifact_certification.is_none() { + /// Add one durable record to an existing mutation without another SQL call. + pub(crate) fn artifact_change_query( + &mut self, + sql: &str, + community_parameter: usize, + operation_parameter: usize, + needed: bool, + ) -> (String, Option) { + let operation = if needed && self.artifact_certification.is_none() { let operation = Uuid::new_v4(); - sqlx::query("INSERT INTO artifact_sync_pending (community_id,operation_id,earliest_at) VALUES ($1,$2,clock_timestamp())") - .bind(self.community.as_uuid()).bind(operation).execute(&mut *self.tx).await?; self.artifact_certification = Some(operation); - } + Some(operation) + } else { + None + }; + // The transaction clock is a conservative bound for NOW() defaults. + // CTE execution order cannot move this lower bound past the change. + let sql = format!( + "WITH buzz_artifact_certification AS (INSERT INTO artifact_sync_pending \ + (community_id,operation_id,earliest_at) \ + SELECT ${community_parameter},${operation_parameter}::uuid,LEAST(transaction_timestamp(),statement_timestamp()) \ + WHERE ${operation_parameter}::uuid IS NOT NULL) {sql}" + ); + (sql, operation) + } + + #[cfg(test)] + pub(crate) async fn record_artifact_change(&mut self) -> Result<()> { + let (sql, operation) = self.artifact_change_query("SELECT $1::uuid", 1, 2, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(self.community.as_uuid()) + .bind(operation) + .execute(self.conn()) + .await?; Ok(()) } diff --git a/crates/buzz-db/src/store/artifact.rs b/crates/buzz-db/src/store/artifact.rs index 0fc29562e48..4f30c2c1bf6 100644 --- a/crates/buzz-db/src/store/artifact.rs +++ b/crates/buzz-db/src/store/artifact.rs @@ -184,16 +184,19 @@ impl Db { } } } - tx.record_artifact_change().await?; - sqlx::query( - "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?; + let (sql, operation) = tx.artifact_change_query("INSERT INTO artifact_revisions (community_id,event_id,artifact_id,source_channel_id,modified_at) VALUES ($1,$2,$3,$4,clock_timestamp())", 1, 5, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(env.id) + .bind(if env.op == ArtifactOp::Move { + source + } else { + None + }) + .bind(operation) + .execute(tx.conn()) + .await?; 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, _) = diff --git a/crates/buzz-db/src/store/artifact_certification.rs b/crates/buzz-db/src/store/artifact_certification.rs index 552e2a822e7..dd9bb89c65f 100644 --- a/crates/buzz-db/src/store/artifact_certification.rs +++ b/crates/buzz-db/src/store/artifact_certification.rs @@ -1,7 +1,7 @@ //! Durable proof of the timestamp overlap used by incremental artifact reads. use buzz_core::CommunityId; -use sqlx::PgPool; +use sqlx::{pool::PoolConnection, PgConnection, PgPool, Postgres}; use std::time::Duration; use uuid::Uuid; @@ -9,145 +9,190 @@ use crate::{Db, Result}; /// Timestamp overlap shared by writers and incremental readers. pub const OVERLAP_SECONDS: i64 = 120; -const BATCH: i64 = 100; +const BATCH: i64 = 1_000; +// Only recovery takes this lock. Serving writes do not use it. +const RECOVERY_LOCK: i64 = 0x42555a5a415243; +struct RecoveryConnection(Option>); + +impl RecoveryConnection { + fn conn(&mut self) -> Result<&mut PgConnection> { + self.0 + .as_deref_mut() + .ok_or_else(|| sqlx::Error::PoolClosed.into()) + } + + fn reuse(mut self) { + drop(self.0.take()); + } +} + +impl Drop for RecoveryConnection { + fn drop(&mut self) { + // Cancellation or an uncertain commit must not return this connection. + if let Some(connection) = self.0.take() { + drop(connection.detach()); + } + } +} + +#[cfg(test)] pub(crate) async fn certify( pool: &PgPool, community: CommunityId, operation: Option, ) -> Result { let deadline = tokio::time::Instant::now() + Duration::from_millis(500); - tokio::time::timeout_at(deadline, async { - let mut connection = crate::observability::acquire_writer_until( + let mut connection = RecoveryConnection(Some( + crate::observability::acquire_writer_until( pool, crate::observability::WriterOperation::Maintenance, deadline, ) - .await?; - // A cancelled commit has uncertain protocol state. Never reuse its connection. - connection.close_on_drop(); - sqlx::query("SET statement_timeout='400ms'") - .execute(&mut *connection) - .await?; - let tx = sqlx::Transaction::begin(connection, None).await?; - let mut tx = match crate::AdmittedTx::admit( - tx, - &crate::deletion::DeletionStore::new(pool.clone()), - community, + .await?, + )); + let completed = tokio::time::timeout_at(deadline, async { + let mut tx = sqlx::Connection::begin(connection.conn()?).await?; + sqlx::raw_sql( + "SET TRANSACTION ISOLATION LEVEL READ COMMITTED; SET LOCAL statement_timeout='400ms'", ) - .await + .execute(&mut *tx) + .await?; + match crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community) + .await { - Ok(tx) => tx, - Err(crate::DbError::AccessDenied(_)) => return Ok(0), + Ok(()) => (), + Err(crate::DbError::AccessDenied(_)) => { + tx.rollback().await?; + return Ok(0); + } Err(error) => return Err(error), - }; - let completed = certify_pending(&mut tx, operation).await?; + } + let completed = certify_pending(&mut tx, &[*community.as_uuid()], operation).await?; tx.commit().await?; - Ok(completed) + Ok::<_, crate::DbError>(completed) }) .await - .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))? + .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))??; + connection.reuse(); + Ok(completed) } -async fn certify_pending(tx: &mut crate::AdmittedTx, operation: Option) -> Result { - let community = tx.community(); - sqlx::query("SET LOCAL statement_timeout='400ms'") - .execute(tx.conn()) - .await?; - let pending: Vec<(Uuid, chrono::DateTime)> = sqlx::query_as( - "SELECT operation_id,earliest_at FROM artifact_sync_pending \ - WHERE community_id=$1 AND ($2::uuid IS NULL OR operation_id=$2) \ - ORDER BY earliest_at,operation_id LIMIT $3 FOR UPDATE SKIP LOCKED", +async fn certify_pending( + connection: &mut PgConnection, + communities: &[Uuid], + operation: Option, +) -> Result { + let completed: i64 = sqlx::query_scalar( + "WITH pending AS MATERIALIZED ( \ + SELECT p.community_id,p.operation_id,p.earliest_at \ + FROM artifact_sync_pending p \ + WHERE p.community_id=ANY($1) AND ($2::uuid IS NULL OR p.operation_id=$2) \ + ORDER BY p.community_id,p.earliest_at,p.operation_id \ + LIMIT $3 FOR UPDATE OF p SKIP LOCKED \ + ), observed AS MATERIALIZED ( \ + SELECT clock_timestamp() AS now FROM (SELECT count(*) FROM pending) ready \ + ), resets AS ( \ + UPDATE communities c SET artifact_policy_modified_at=GREATEST( \ + clock_timestamp(),c.artifact_policy_modified_at + interval '1 microsecond') \ + WHERE c.id IN (SELECT p.community_id FROM pending p CROSS JOIN observed o \ + WHERE p.earliest_at>o.now OR o.now-p.earliest_at>=make_interval(secs=>$4)) \ + RETURNING c.id \ + ), completed AS ( \ + DELETE FROM artifact_sync_pending p USING pending claimed \ + WHERE p.community_id=claimed.community_id AND p.operation_id=claimed.operation_id \ + RETURNING p.operation_id \ + ) SELECT count(*) FROM completed", ) - .bind(community.as_uuid()) + .bind(communities) .bind(operation) .bind(BATCH) - .fetch_all(tx.conn()) + .bind(OVERLAP_SECONDS as f64) + .fetch_one(connection) .await?; - if pending.is_empty() { - return Ok(0); - } - // This clock read follows a read that sees the committed pending records. - // A delayed recovery makes the proof conservative and requires a reset. - let now: chrono::DateTime = sqlx::query_scalar("SELECT clock_timestamp()") - .fetch_one(tx.conn()) - .await?; - if pending.iter().any(|(_, earliest)| { - *earliest > now || now - *earliest >= chrono::Duration::seconds(OVERLAP_SECONDS) - }) { - sqlx::query("UPDATE communities SET artifact_policy_modified_at=GREATEST(clock_timestamp(),artifact_policy_modified_at + interval '1 microsecond') WHERE id=$1") - .bind(community.as_uuid()) - .execute(tx.conn()) - .await?; - } - let operations: Vec<_> = pending.iter().map(|(id, _)| *id).collect(); - sqlx::query("DELETE FROM artifact_sync_pending WHERE community_id=$1 AND operation_id=ANY($2)") - .bind(community.as_uuid()) - .bind(&operations) - .execute(tx.conn()) - .await?; - Ok(pending.len()) + Ok(completed as usize) } impl Db { - /// Recover one bounded batch in the first active community with pending work. + /// Recover one bounded batch across active communities with pending work. pub async fn recover_artifact_certifications(&self) -> Result { Ok(self.recover_artifact_certifications_after(None).await?.0) } - /// Recover at most 100 records in the next community after `after`. - /// The returned UUID advances the sweep even when that community is blocked. - /// `None` means that the sweep reaches its end and can start again. + /// Recover at most 1,000 records across the next communities after `after`. + /// The cursor advances even when a community is blocked. A competing worker + /// leaves the cursor unchanged. `None` marks the end of a sweep. pub async fn recover_artifact_certifications_after( &self, after: Option, ) -> Result<(usize, Option)> { let deadline = tokio::time::Instant::now() + Duration::from_millis(500); - let mut connection = crate::observability::acquire_writer_until( - &self.pool, - crate::observability::WriterOperation::Maintenance, - deadline, - ) - .await?; - let next = tokio::time::timeout_at(deadline, async { - let mut tx = sqlx::Connection::begin(&mut *connection).await?; - sqlx::query("SET LOCAL statement_timeout='400ms'") + let mut connection = RecoveryConnection(Some( + crate::observability::acquire_writer_until( + &self.pool, + crate::observability::WriterOperation::Maintenance, + deadline, + ) + .await?, + )); + let result = tokio::time::timeout_at(deadline, async { + let mut tx = sqlx::Connection::begin(connection.conn()?).await?; + sqlx::raw_sql("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; SET LOCAL statement_timeout='400ms'") .execute(&mut *tx) .await?; - let next = sqlx::query_scalar::<_, Uuid>( - "SELECT DISTINCT p.community_id FROM artifact_sync_pending p \ - JOIN communities c ON c.id=p.community_id \ - WHERE c.deletion_state='active' AND c.deleted_at IS NULL \ - AND ($1::uuid IS NULL OR p.community_id>$1) \ - ORDER BY p.community_id LIMIT 1", + let elected: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(RECOVERY_LOCK) + .fetch_one(&mut *tx) + .await?; + if !elected { + tx.rollback().await?; + return Ok((0, after)); + } + let candidates: Vec = sqlx::query_scalar( + "SELECT DISTINCT community_id FROM ( \ + SELECT p.community_id FROM artifact_sync_pending p \ + JOIN communities c ON c.id=p.community_id \ + WHERE c.deletion_state='active' AND c.deleted_at IS NULL \ + AND ($1::uuid IS NULL OR p.community_id>$1) \ + ORDER BY p.community_id,p.earliest_at,p.operation_id LIMIT $2 \ + ) batch ORDER BY community_id", ) .bind(after) - .fetch_optional(&mut *tx) + .bind(BATCH) + .fetch_all(&mut *tx) + .await?; + let next = candidates.last().copied(); + let admitted: Vec = sqlx::query_scalar( + "SELECT id FROM unnest($1::uuid[]) id \ + WHERE pg_try_advisory_xact_lock_shared(community_deletion_lock_key(id)) \ + ORDER BY id", + ) + .bind(&candidates) + .fetch_all(&mut *tx) + .await?; + // Read lifecycle state in a new snapshot after deletion admission. + let active: Vec = sqlx::query_scalar( + "SELECT id FROM communities WHERE id=ANY($1) \ + AND deletion_state='active' AND deleted_at IS NULL \ + AND current_setting('transaction_isolation')='read committed' \ + ORDER BY id FOR NO KEY UPDATE SKIP LOCKED", + ) + .bind(&admitted) + .fetch_all(&mut *tx) .await?; + let completed = if active.is_empty() { + 0 + } else { + certify_pending(&mut tx, &active, None).await? + }; tx.commit().await?; - Ok::<_, sqlx::Error>(next) + Ok::<_, crate::DbError>((completed, next)) }) - .await; - let next = match next { - Ok(result) => result?, - Err(_) => { - drop(connection.detach()); - return Err(sqlx::Error::PoolTimedOut.into()); - } - }; - drop(connection); - let Some(id) = next else { - return Ok((0, None)); - }; - let community = CommunityId::from_uuid(id); - match certify(&self.pool, community, None).await { - Ok(count) => Ok((count, Some(id))), - Err(error) => { - tracing::warn!(%error, community_id = %community, - "artifact certification requires recovery"); - Ok((0, Some(id))) - } - } + .await + .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))??; + connection.reuse(); + Ok(result) } } diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index dff6cb244c3..fbaa01cdfa5 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -83,7 +83,12 @@ async fn certification_crash_rolls_back_reset_and_completion_together() { tx.commit().await.unwrap(); let mut recovery = begin(&db, community).await; - assert_eq!(certify_pending(&mut recovery, None).await.unwrap(), 1); + assert_eq!( + certify_pending(recovery.conn(), &[*community.as_uuid()], None) + .await + .unwrap(), + 1 + ); assert_eq!(state(&db, community).await, (initial.0, 1)); drop(recovery); let (a, b) = tokio::join!( @@ -190,16 +195,10 @@ async fn blocked_community_does_not_exclude_other_pending_work() { .recover_artifact_certifications_after(None) .await .unwrap(); - assert_eq!(completed, 0); - assert_eq!(after, Some(*communities[0].as_uuid())); - let (completed, after) = db - .recover_artifact_certifications_after(after) - .await - .unwrap(); - assert_eq!(completed, 100); + assert_eq!(completed, 101); assert_eq!(after, Some(*communities[1].as_uuid())); assert_eq!(state(&db, communities[0]).await.1, 101); - assert_eq!(state(&db, communities[1]).await.1, 1); + assert_eq!(state(&db, communities[1]).await.1, 0); blocker.rollback().await.unwrap(); } @@ -322,3 +321,139 @@ async fn lost_commit_acknowledgement_recovers_the_committed_operation() { ); assert_eq!(state(&db, community).await, completed); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn recovery_reuses_confirmed_connections_and_elects_one_worker() { + let (db, community) = fixture().await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + tx.commit().await.unwrap(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + let recovery = Db::from_pool(pool.clone()); + let mut election = db.pool().begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(RECOVERY_LOCK) + .execute(&mut *election) + .await + .unwrap(); + assert_eq!( + recovery + .recover_artifact_certifications_after(None) + .await + .unwrap(), + (0, None) + ); + assert_eq!(state(&db, community).await.1, 1); + election.rollback().await.unwrap(); + assert_eq!(recovery.recover_artifact_certifications().await.unwrap(), 1); + assert_eq!(recovery.recover_artifact_certifications().await.unwrap(), 0); + let next: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(pid, next); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn cancellation_discards_the_recovery_connection() { + let (db, community) = fixture().await; + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + tx.commit().await.unwrap(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + let mut block = db.pool().begin().await.unwrap(); + sqlx::query("LOCK TABLE artifact_sync_pending IN ACCESS EXCLUSIVE MODE") + .execute(&mut *block) + .await + .unwrap(); + let recovery = Db::from_pool(pool.clone()); + let task = tokio::spawn(async move { recovery.recover_artifact_certifications().await }); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE pid=$1 AND wait_event_type='Lock')") + .bind(pid).fetch_one(db.pool()).await.unwrap(); + if waiting { break; } + tokio::task::yield_now().await; + } + }).await.unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + block.rollback().await.unwrap(); + let next: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + assert_ne!(pid, next); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn deletion_admission_does_not_block_other_communities() { + let (db, a) = fixture().await; + let (_, b) = fixture().await; + for community in [a, b] { + let mut tx = begin(&db, community).await; + tx.record_artifact_change().await.unwrap(); + tx.commit().await.unwrap(); + } + let mut deletion = db.pool().begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(a.as_uuid()) + .execute(&mut *deletion) + .await + .unwrap(); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); + assert_eq!(state(&db, a).await.1, 1); + assert_eq!(state(&db, b).await.1, 0); + deletion.rollback().await.unwrap(); + assert_eq!(db.recover_artifact_certifications().await.unwrap(), 1); +} + +#[tokio::test] +#[ignore = "requires Postgres; performance measurement"] +async fn measure_certification_backlog() { + let (db, _) = fixture().await; + for communities in [1, 100] { + for repetition in 0..3 { + for _ in 0..communities { + let (_, community) = fixture().await; + sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) SELECT $1,gen_random_uuid(),clock_timestamp() FROM generate_series(1,$2)") + .bind(community.as_uuid()).bind(10_000 / communities).execute(db.pool()).await.unwrap(); + } + let start = std::time::Instant::now(); + let mut after = None; + let mut completed = 0; + let mut calls = 0; + while completed < 10_000 { + let (count, next) = db + .recover_artifact_certifications_after(after) + .await + .unwrap(); + completed += count; + after = next; + calls += 1; + assert!(calls < 30_000); + } + println!("PERF recovery communities={communities} repetition={repetition} count={completed} calls={calls} elapsed_ms={:.3}", start.elapsed().as_secs_f64()*1000.0); + } + } +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 5572152faf4..b9427e8d6a7 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -161,9 +161,7 @@ pub async fn create_channel( .execute(tx.conn()) .await?; - tx.record_artifact_change().await?; - sqlx::query( - r#" + let (sql, operation) = tx.artifact_change_query(r#" 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 @@ -171,14 +169,15 @@ pub async fn create_channel( removed_at = NULL, removed_by = NULL, role = EXCLUDED.role - "#, - ) - .bind(community_id.as_uuid()) - .bind(id) - .bind(created_by) - .bind(created_by) - .execute(tx.conn()) - .await?; + "#, 1, 5, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(id) + .bind(created_by) + .bind(created_by) + .bind(operation) + .execute(tx.conn()) + .await?; let row = sqlx::query( r#" @@ -404,9 +403,7 @@ async fn insert_channel_rows( if was_created { // Bootstrap the creator as owner. - tx.record_artifact_change().await?; - sqlx::query( - r#" + let (sql, operation) = tx.artifact_change_query(r#" 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 @@ -414,14 +411,15 @@ async fn insert_channel_rows( removed_at = NULL, removed_by = NULL, role = EXCLUDED.role - "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(created_by) - .bind(created_by) - .execute(tx.conn()) - .await?; + "#, 1, 5, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(created_by) + .bind(created_by) + .bind(operation) + .execute(tx.conn()) + .await?; for member in members .iter() @@ -433,20 +431,24 @@ async fn insert_channel_rows( member.len() ))); } - tx.record_artifact_change().await?; - sqlx::query( + let (sql, operation) = tx.artifact_change_query( r#" 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 NOTHING "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(member) - .bind(created_by) - .execute(tx.conn()) - .await?; + 1, + 5, + true, + ); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(member) + .bind(created_by) + .bind(operation) + .execute(tx.conn()) + .await?; } } @@ -1168,6 +1170,19 @@ async fn update_channel_on( set_parts.join(", ") ); + 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 (sql, operation) = + tx.artifact_change_query(&sql, param_idx, channel_param_idx + 1, policy_changed); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)); if let Some(ref name) = updates.name { q = q.bind(name); @@ -1195,21 +1210,8 @@ async fn update_channel_on( } q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); + q = q.bind(operation); - 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 - }; - if policy_changed { - tx.record_artifact_change().await?; - } let result = q.execute(tx.conn()).await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); @@ -1570,10 +1572,16 @@ async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> { } let channel_id = w.channel_id; let community_id = w.tx.community(); - w.tx.record_artifact_change().await?; - sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2") + let (sql, operation) = w.tx.artifact_change_query( + "UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2", + 1, + 3, + true, + ); + sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(community_id.as_uuid()) .bind(channel_id) + .bind(operation) .execute(w.tx.conn()) .await?; sqlx::query("UPDATE communities SET artifact_policy_modified_at=GREATEST(clock_timestamp(),artifact_policy_modified_at + interval '1 microsecond') WHERE id=$1") diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index faecc1d00cd..94d690ceb2b 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -253,9 +253,7 @@ impl ChannelMembershipLocks<'_> { ) -> Result<()> { self.check_covers(channel_id)?; let community_id = self.tx.community(); - self.tx.record_artifact_change().await?; - sqlx::query( - r#" + let (sql, operation) = self.tx.artifact_change_query(r#" 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 @@ -263,14 +261,15 @@ impl ChannelMembershipLocks<'_> { removed_at = NULL, removed_by = NULL, role = EXCLUDED.role - "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(pubkey) - .bind(invited_by) - .execute(self.tx.conn()) - .await?; + "#, 1, 5, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .bind(invited_by) + .bind(operation) + .execute(self.tx.conn()) + .await?; Ok(()) } } @@ -1016,9 +1015,7 @@ async fn put_members_on( let mut written: std::collections::HashMap, MemberRecord> = std::collections::HashMap::with_capacity(write_keys.len()); if !write_keys.is_empty() { - tx.record_artifact_change().await?; - let rows = sqlx::query( - r#" + let (sql, operation) = tx.artifact_change_query(r#" INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) SELECT $1, $2, t.pubkey, t.role::member_role, $5 FROM UNNEST($3::bytea[], $4::text[]) AS t(pubkey, role) @@ -1028,15 +1025,16 @@ async fn put_members_on( removed_by = NULL, role = EXCLUDED.role RETURNING channel_id, pubkey, role::text AS role, joined_at, invited_by, removed_at - "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(&write_keys) - .bind(&write_roles) - .bind(invited_by) - .fetch_all(tx.conn()) - .await?; + "#, 1, 6, true); + let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(&write_keys) + .bind(&write_roles) + .bind(invited_by) + .bind(operation) + .fetch_all(tx.conn()) + .await?; for row in rows { let member = row_to_member_record(row)?; written.insert(member.pubkey.clone(), member); @@ -1154,20 +1152,24 @@ pub async fn remove_member( } } - tx.record_artifact_change().await?; - let result = sqlx::query( + let (sql, operation) = tx.artifact_change_query( r#" UPDATE channel_members 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 "#, - ) - .bind(actor_pubkey) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(pubkey) - .execute(tx.conn()) - .await?; + 2, + 5, + true, + ); + let result = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(actor_pubkey) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .bind(operation) + .execute(tx.conn()) + .await?; if result.rows_affected() == 0 { return Err(DbError::MemberNotFound(channel_id)); diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 30eeba033c7..c05371eca60 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -187,9 +187,7 @@ pub async fn create_dm( // Add all participants as members with role='member'. for pk in participants { - tx.record_artifact_change().await?; - sqlx::query( - r#" + let (sql, operation) = tx.artifact_change_query(r#" 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 @@ -197,14 +195,15 @@ pub async fn create_dm( removed_at = NULL, removed_by = NULL, role = EXCLUDED.role - "#, - ) - .bind(community_id.as_uuid()) - .bind(id) - .bind(*pk) - .bind(created_by) - .execute(tx.conn()) - .await?; + "#, 1, 5, true); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(id) + .bind(*pk) + .bind(created_by) + .bind(operation) + .execute(tx.conn()) + .await?; } let row = sqlx::query( diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 3986ec90486..22fa05c3a21 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1557,11 +1557,16 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx( let deleted = purged > 0 || result.rows_affected() > 0; if deleted { - if target.is_some_and(|(kind, _)| kind == 45010) { - tx.record_artifact_change().await?; - } - 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?; + let (sql, operation) = tx.artifact_change_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", + 1, 3, target.is_some_and(|(kind, _)| kind == 45010), + ); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community_id.as_uuid()) + .bind(event_id) + .bind(operation) + .execute(tx.conn()) + .await?; if let Some(pid) = parent_event_id { sqlx::query( "UPDATE thread_metadata \ diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 7cc6fdca734..3813ff19314 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -718,20 +718,24 @@ pub async fn execute_kick_with_marker( crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id) .await?; - tx.record_artifact_change().await?; - let kick = sqlx::query( + let (sql, operation) = tx.artifact_change_query( r#" UPDATE channel_members 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 "#, - ) - .bind(actor_pubkey) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(target_pubkey) - .execute(tx.conn()) - .await?; + 2, + 5, + true, + ); + let kick = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(actor_pubkey) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(target_pubkey) + .bind(operation) + .execute(tx.conn()) + .await?; if kick.rows_affected() == 0 { tx.rollback().await?; @@ -1727,20 +1731,24 @@ pub async fn deploy_kick_member( .await?; // Use a direct UPDATE to avoid the tenant ownership check in channel::remove_member. // This is the deployment-authority primitive: no actor role check. - tx.record_artifact_change().await?; - let result = sqlx::query( + let (sql, operation) = tx.artifact_change_query( r#" UPDATE channel_members 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 "#, - ) - .bind(actor_pubkey) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(target_pubkey) - .execute(tx.conn()) - .await?; + 2, + 5, + true, + ); + let result = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(actor_pubkey) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(target_pubkey) + .bind(operation) + .execute(tx.conn()) + .await?; tx.commit().await?; if result.rows_affected() > 0 { diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index bed57e40d46..1602a6617c1 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -79,8 +79,9 @@ migrations enabled, the older binary still has the SQLx `VersionMissing` restriction described in `docs/database.md`. Each supported writer records one pending operation in its admitted transaction. -The operation has a unique UUID. Its timestamp precedes every relevant sync -stamp in that transaction. Repeated changes in the transaction reuse the record. +The operation has a unique UUID. The existing mutation statement also inserts the record, with no extra SQL +round trip. Its timestamp uses the earlier transaction or statement clock as +a conservative bound for the relevant sync stamps. Repeated changes in the transaction reuse the record. Rollback removes both the changes and the record. An uncertain commit leaves both committed or neither committed. Recovery does not require a caller retry. @@ -88,9 +89,11 @@ Every relay starts a recovery worker, even when the incremental gate is off. The worker reads bounded batches. It rotates across communities, including after a failed attempt. It locks pending rows with `SKIP LOCKED`. Each tick has a five-second work budget and at most 100 batches. Each batch has at most -100 records. A blocked community cannot exclude other communities. Each -operation has a short deadline. Recovery closes certification connections -instead of returning an uncertain connection to the pool. +1,000 records across communities in one transaction. Only one worker processes +a batch across the fleet. The election lock applies only to recovery. Serving +writes do not take it. A blocked community cannot exclude other communities. +Each operation has a short deadline. Recovery reuses a connection after a +confirmed commit. Errors or cancellation close that connection. After it observes a committed record, it reads database time. If less than 120 seconds separate that time from the early timestamp, the overlap covers visibility of the write. A clock reversal or a delay of at least 120 seconds diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index 421e8cab2c3..85153287f94 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -291,7 +291,8 @@ 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()`. A relay with durable certification records -pending work in the same transaction, before the first sync stamp. After +pending work in the same statement as a relevant change. The record uses +a conservative timestamp from the start of the transaction or statement. After commit, it proves visibility within the 120-second overlap or requests a policy reset. Recovery finishes pending work after a crash without a caller retry. The reset and completion of certification commit together. From a251c7cba42d8266c873a40aabb4d9693a9a534c Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:11:07 -0800 Subject: [PATCH 12/20] test(db): count recovery connection setup Signed-off-by: Daniel Rouleau --- .../artifact_certification_postgres_tests.rs | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index fbaa01cdfa5..0a9761422be 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -431,7 +431,23 @@ async fn deletion_admission_does_not_block_other_communities() { #[tokio::test] #[ignore = "requires Postgres; performance measurement"] async fn measure_certification_backlog() { - let (db, _) = fixture().await; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + let (_, _) = fixture().await; + let connections = Arc::new(AtomicUsize::new(0)); + let observed = connections.clone(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(10) + .after_connect(move |_, _| { + observed.fetch_add(1, Ordering::Relaxed); + Box::pin(async { Ok(()) }) + }) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool); for communities in [1, 100] { for repetition in 0..3 { for _ in 0..communities { @@ -439,6 +455,7 @@ async fn measure_certification_backlog() { sqlx::query("INSERT INTO artifact_sync_pending(community_id,operation_id,earliest_at) SELECT $1,gen_random_uuid(),clock_timestamp() FROM generate_series(1,$2)") .bind(community.as_uuid()).bind(10_000 / communities).execute(db.pool()).await.unwrap(); } + let before = connections.load(Ordering::Relaxed); let start = std::time::Instant::now(); let mut after = None; let mut completed = 0; @@ -453,7 +470,7 @@ async fn measure_certification_backlog() { calls += 1; assert!(calls < 30_000); } - println!("PERF recovery communities={communities} repetition={repetition} count={completed} calls={calls} elapsed_ms={:.3}", start.elapsed().as_secs_f64()*1000.0); + println!("PERF recovery communities={communities} repetition={repetition} count={completed} calls={calls} connections={} elapsed_ms={:.3}", connections.load(Ordering::Relaxed)-before, start.elapsed().as_secs_f64()*1000.0); } } } From 9f8e4e072be4194ef2690073dc41482cecd3c1aa Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:12:19 -0800 Subject: [PATCH 13/20] fix(db): scope certification test imports Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/store/artifact_certification.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/artifact_certification.rs b/crates/buzz-db/src/store/artifact_certification.rs index dd9bb89c65f..2135de1c806 100644 --- a/crates/buzz-db/src/store/artifact_certification.rs +++ b/crates/buzz-db/src/store/artifact_certification.rs @@ -1,7 +1,10 @@ //! Durable proof of the timestamp overlap used by incremental artifact reads. +#[cfg(test)] use buzz_core::CommunityId; -use sqlx::{pool::PoolConnection, PgConnection, PgPool, Postgres}; +#[cfg(test)] +use sqlx::PgPool; +use sqlx::{pool::PoolConnection, PgConnection, Postgres}; use std::time::Duration; use uuid::Uuid; From 8a8f4baee4f58ded8bf29aaa0c8e48ed9014720e Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:13:47 -0800 Subject: [PATCH 14/20] perf(db): end recovery early when no work remains Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/store/artifact_certification.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/artifact_certification.rs b/crates/buzz-db/src/store/artifact_certification.rs index 2135de1c806..ff818cac77d 100644 --- a/crates/buzz-db/src/store/artifact_certification.rs +++ b/crates/buzz-db/src/store/artifact_certification.rs @@ -95,7 +95,7 @@ async fn certify_pending( ORDER BY p.community_id,p.earliest_at,p.operation_id \ LIMIT $3 FOR UPDATE OF p SKIP LOCKED \ ), observed AS MATERIALIZED ( \ - SELECT clock_timestamp() AS now FROM (SELECT count(*) FROM pending) ready \ + SELECT clock_timestamp() AS now FROM pending HAVING count(*) >= 0 \ ), resets AS ( \ UPDATE communities c SET artifact_policy_modified_at=GREATEST( \ clock_timestamp(),c.artifact_policy_modified_at + interval '1 microsecond') \ @@ -165,6 +165,10 @@ impl Db { .bind(BATCH) .fetch_all(&mut *tx) .await?; + if candidates.is_empty() { + tx.commit().await?; + return Ok((0, None)); + } let next = candidates.last().copied(); let admitted: Vec = sqlx::query_scalar( "SELECT id FROM unnest($1::uuid[]) id \ From 399067a924ac2ceb2394ee60c135e42773600761 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:17:09 -0800 Subject: [PATCH 15/20] fix(relay): stop a tick when recovery is busy Signed-off-by: Daniel Rouleau --- crates/buzz-relay/src/artifact_certification.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/buzz-relay/src/artifact_certification.rs b/crates/buzz-relay/src/artifact_certification.rs index 28ad1543c3c..2a85a254c99 100644 --- a/crates/buzz-relay/src/artifact_certification.rs +++ b/crates/buzz-relay/src/artifact_certification.rs @@ -25,6 +25,9 @@ pub async fn run(state: Arc) { } match state.db.recover_artifact_certifications_after(after).await { Ok((completed, next)) => { + if completed == 0 && next == after { + break; + } after = next; progress |= completed > 0; if next.is_none() { From 93b5a686ac1f349c2b1fc4d9bf11427655fc11ab Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:18:38 -0800 Subject: [PATCH 16/20] docs: trim the artifact rollout notes Signed-off-by: Daniel Rouleau --- docs/artifact-sync-deployment.md | 1 - 1 file changed, 1 deletion(-) diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index 7bb8bbc6cba..ecdfe8785fe 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -111,4 +111,3 @@ before enablement or after a rollback to an old writer. This change does not authorize deployment, gate enablement, or a production maintenance write. - From 9f6342e3a7f931948f17da7707811caade8aeea2 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:20:44 -0800 Subject: [PATCH 17/20] test(db): identify the measured source checkout Signed-off-by: Daniel Rouleau --- .../buzz-db/src/store/artifact_certification_postgres_tests.rs | 1 + crates/buzz-db/src/store/artifact_postgres_tests.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs index 0a9761422be..73e8f2e36ac 100644 --- a/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs @@ -431,6 +431,7 @@ async fn deletion_admission_does_not_block_other_communities() { #[tokio::test] #[ignore = "requires Postgres; performance measurement"] async fn measure_certification_backlog() { + println!("PERF source={}", env!("CARGO_MANIFEST_DIR")); use std::sync::{ atomic::{AtomicUsize, Ordering}, Arc, diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index e4bb54d9b41..6fce8b2753a 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -983,6 +983,7 @@ async fn sync_snapshot_pages_are_one_cut_and_head_includes_deletions() { #[tokio::test] #[ignore = "requires Postgres; performance measurement"] async fn measure_artifact_write_latency() { + println!("PERF source={}", env!("CARGO_MANIFEST_DIR")); use std::{sync::Arc, time::Instant}; let pool = sqlx::postgres::PgPoolOptions::new() .max_connections(64) From f4ba2b40fe64f64c6dfa9e02eec48c8ee7138749 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:24:02 -0800 Subject: [PATCH 18/20] fix(db): use the storage test scratch fixture Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/store/storage_accounting.rs | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/crates/buzz-db/src/store/storage_accounting.rs b/crates/buzz-db/src/store/storage_accounting.rs index 12f88bba044..c594577ece6 100644 --- a/crates/buzz-db/src/store/storage_accounting.rs +++ b/crates/buzz-db/src/store/storage_accounting.rs @@ -261,12 +261,8 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn usage_delivery_reads_snapshot_time_and_community_states() { - let admin = PgPoolOptions::new() - .max_connections(1) - .connect(&crate::test_support::database_url()) - .await - .expect("connect admin"); - let (pool, name) = create_scratch_db(&admin).await; + let (pool, _scratch) = + ScratchDb::migrated(&crate::test_support::database_url(), "storage_accounting").await; let db = Db::from_pool(pool.clone()); assert_eq!( @@ -331,10 +327,5 @@ mod postgres_tests { drop(leader); drop(db); pool.close().await; - let _ = sqlx::query(sqlx::AssertSqlSafe(format!( - "DROP DATABASE IF EXISTS {name} WITH (FORCE)" - ))) - .execute(&admin) - .await; } } From 660f131a0ccf863cf7eb6e1b7d0b7b95c0666aa0 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:36:01 -0800 Subject: [PATCH 19/20] test(relay): follow the current incremental read contract Signed-off-by: Daniel Rouleau --- .../tests/artifact_certification_e2e.rs | 30 ++++--------------- docs/artifact-sync-deployment.md | 9 +++--- 2 files changed, 10 insertions(+), 29 deletions(-) diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs index 4540a030fbd..a07176a18b6 100644 --- a/crates/buzz-relay/tests/artifact_certification_e2e.rs +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -89,7 +89,7 @@ mod external_infra_process { let relay_keys = Keys::generate(); let owner = Keys::generate(); let log = tempfile::NamedTempFile::new().unwrap(); - let spawn = |enabled: bool| { + let spawn = || { let output = log.reopen().unwrap(); Relay( Command::new(env!("CARGO_BIN_EXE_buzz-relay")) @@ -106,7 +106,6 @@ mod external_infra_process { .env("BUZZ_REQUIRE_AUTH_TOKEN", "true") .env("BUZZ_REQUIRE_RELAY_MEMBERSHIP", "false") .env("BUZZ_AUTO_MIGRATE", "true") - .env("BUZZ_ARTIFACT_INCREMENTAL_ENABLED", enabled.to_string()) .env("BUZZ_PUSH_ENABLED", "false") .env("BUZZ_AUDIT_ENABLED", "false") .env("BUZZ_GIT_CONFORMANCE_PROBE", "false") @@ -120,7 +119,7 @@ mod external_infra_process { .timeout(Duration::from_secs(5)) .build() .unwrap(); - let mut relay = spawn(true); + let mut relay = spawn(); ready(&client, &base, &log).await; let channel = Uuid::new_v4(); let create = EventBuilder::new(Kind::Custom(9007), "") @@ -174,7 +173,7 @@ mod external_infra_process { .bind(community).bind(first["startedAt"].as_str().unwrap()).execute(&mut *tx).await.unwrap(); tx.commit().await.unwrap(); - relay = spawn(true); + relay = spawn(); ready(&client, &base, &log).await; let mut reset = false; for _ in 0..30 { @@ -226,26 +225,9 @@ mod external_infra_process { let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; assert_eq!(status, 200); assert!(ordinary.to_string().contains(&event.id.to_hex())); - drop(relay); - - let _relay = spawn(false); - ready(&client, &base, &log).await; - let (status, body) = post( - &client, - &base, - &owner, - "/query", - json!([{ - "artifact":"incremental", "query":query - }]), - ) - .await; - assert_eq!(status, 503); - assert_eq!(body["code"], "incremental_unavailable"); - let (status, ordinary) = post(&client, &base, &owner, "/query", json!([query])).await; - assert_eq!(status, 200); - assert!(ordinary.to_string().contains(&event.id.to_hex())); - println!("PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads, gate off"); + println!( + "PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads" + ); } async fn ready(client: &reqwest::Client, base: &str, log: &tempfile::NamedTempFile) { diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index ecdfe8785fe..d80e6229c3e 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -75,7 +75,7 @@ a conservative bound for the relevant sync stamps. Repeated changes in the trans Rollback removes both the changes and the record. An uncertain commit leaves both committed or neither committed. Recovery does not require a caller retry. -Every relay starts a recovery worker, even when the incremental gate is off. +Every relay starts a recovery worker. The worker reads bounded batches. It rotates across communities, including after a failed attempt. It locks pending rows with `SKIP LOCKED`. Each tick has a five-second work budget and at most 100 batches. Each batch has at most @@ -106,8 +106,7 @@ listing, TTL, and deletion changes; DM creation; and admin member removals. Archive state and hidden DMs do not change artifact access. Incremental reads exclude TTL channels. Whole-community deletion stops reads and purges pending records. Operator SQL and mixed old writers cannot provide this guarantee. -Keep the gate off during a mixed rollout. Use the existing reset procedure -before enablement or after a rollback to an old writer. +The full guarantee starts after old writers drain. Use the existing reset +procedure after the fleet upgrade or a rollback to an old writer. -This change does not authorize deployment, gate enablement, or a production -maintenance write. +This change does not authorize deployment or a production maintenance write. From 5b93ee50a8b05bf373494f84d06e848668f8d86d Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Tue, 3 Nov 2026 16:37:56 -0800 Subject: [PATCH 20/20] test(relay): retain the restarted process guard Signed-off-by: Daniel Rouleau --- crates/buzz-relay/tests/artifact_certification_e2e.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs index a07176a18b6..aecd28d78bc 100644 --- a/crates/buzz-relay/tests/artifact_certification_e2e.rs +++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs @@ -119,7 +119,7 @@ mod external_infra_process { .timeout(Duration::from_secs(5)) .build() .unwrap(); - let mut relay = spawn(); + let relay = spawn(); ready(&client, &base, &log).await; let channel = Uuid::new_v4(); let create = EventBuilder::new(Kind::Custom(9007), "") @@ -173,7 +173,7 @@ mod external_infra_process { .bind(community).bind(first["startedAt"].as_str().unwrap()).execute(&mut *tx).await.unwrap(); tx.commit().await.unwrap(); - relay = spawn(); + let _relay = spawn(); ready(&client, &base, &log).await; let mut reset = false; for _ in 0..30 {