diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs
index 2e4ce60a758..aa7f6401bac 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<Uuid>,
     /// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes.
     ttl_refresh_channels: BTreeSet<Uuid>,
     push_enqueue: Option<super::push_enqueue::PushEnqueue>,
@@ -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,43 @@ impl AdmittedTx {
         }
     }
 
+    /// 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<Uuid>) {
+        let operation = if needed && self.artifact_certification.is_none() {
+            let operation = Uuid::new_v4();
+            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(())
+    }
+
     /// 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..4f30c2c1bf6 100644
--- a/crates/buzz-db/src/store/artifact.rs
+++ b/crates/buzz-db/src/store/artifact.rs
@@ -184,15 +184,19 @@ impl Db {
                 }
             }
         }
-        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
new file mode 100644
index 00000000000..ff818cac77d
--- /dev/null
+++ b/crates/buzz-db/src/store/artifact_certification.rs
@@ -0,0 +1,208 @@
+//! Durable proof of the timestamp overlap used by incremental artifact reads.
+
+#[cfg(test)]
+use buzz_core::CommunityId;
+#[cfg(test)]
+use sqlx::PgPool;
+use sqlx::{pool::PoolConnection, PgConnection, Postgres};
+use std::time::Duration;
+use uuid::Uuid;
+
+use crate::{Db, Result};
+
+/// Timestamp overlap shared by writers and incremental readers.
+pub const OVERLAP_SECONDS: i64 = 120;
+const BATCH: i64 = 1_000;
+// Only recovery takes this lock. Serving writes do not use it.
+const RECOVERY_LOCK: i64 = 0x42555a5a415243;
+
+struct RecoveryConnection(Option<PoolConnection<Postgres>>);
+
+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<Uuid>,
+) -> Result<usize> {
+    let deadline = tokio::time::Instant::now() + Duration::from_millis(500);
+    let mut connection = RecoveryConnection(Some(
+        crate::observability::acquire_writer_until(
+            pool,
+            crate::observability::WriterOperation::Maintenance,
+            deadline,
+        )
+        .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'",
+        )
+        .execute(&mut *tx)
+        .await?;
+        match crate::deletion::DeletionStore::new(pool.clone())
+            .guard_transaction(&mut tx, community)
+            .await
+        {
+            Ok(()) => (),
+            Err(crate::DbError::AccessDenied(_)) => {
+                tx.rollback().await?;
+                return Ok(0);
+            }
+            Err(error) => return Err(error),
+        }
+        let completed = certify_pending(&mut tx, &[*community.as_uuid()], operation).await?;
+        tx.commit().await?;
+        Ok::<_, crate::DbError>(completed)
+    })
+    .await
+    .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))??;
+    connection.reuse();
+    Ok(completed)
+}
+
+async fn certify_pending(
+    connection: &mut PgConnection,
+    communities: &[Uuid],
+    operation: Option<Uuid>,
+) -> Result<usize> {
+    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 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') \
+             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(communities)
+    .bind(operation)
+    .bind(BATCH)
+    .bind(OVERLAP_SECONDS as f64)
+    .fetch_one(connection)
+    .await?;
+    Ok(completed as usize)
+}
+
+impl Db {
+    /// Recover one bounded batch across active communities with pending work.
+    pub async fn recover_artifact_certifications(&self) -> Result<usize> {
+        Ok(self.recover_artifact_certifications_after(None).await?.0)
+    }
+
+    /// 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<Uuid>,
+    ) -> Result<(usize, Option<Uuid>)> {
+        let deadline = tokio::time::Instant::now() + Duration::from_millis(500);
+        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 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<Uuid> = 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)
+            .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<Uuid> = 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<Uuid> = 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::<_, crate::DbError>((completed, next))
+        })
+        .await
+        .map_err(|_| crate::DbError::Sqlx(sqlx::Error::PoolTimedOut))??;
+        connection.reuse();
+        Ok(result)
+    }
+}
+
+#[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..73e8f2e36ac
--- /dev/null
+++ b/crates/buzz-db/src/store/artifact_certification_postgres_tests.rs
@@ -0,0 +1,477 @@
+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<Utc>, 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();
+    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();
+    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(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!(
+        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);
+
+    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(&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);
+}
+
+#[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, 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, 0);
+    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();
+}
+
+#[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);
+}
+
+#[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() {
+    println!("PERF source={}", env!("CARGO_MANIFEST_DIR"));
+    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 {
+                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 before = connections.load(Ordering::Relaxed);
+            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} connections={} elapsed_ms={:.3}", connections.load(Ordering::Relaxed)-before, start.elapsed().as_secs_f64()*1000.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..6fce8b2753a 100644
--- a/crates/buzz-db/src/store/artifact_postgres_tests.rs
+++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs
@@ -15,6 +15,162 @@ 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);
+}
+
+#[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())
@@ -823,3 +979,46 @@ 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() {
+    println!("PERF source={}", env!("CARGO_MANIFEST_DIR"));
+    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]);
+        }
+    }
+}
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 21cf59e46f4..7620b885ac4 100644
--- a/crates/buzz-db/src/store/channel.rs
+++ b/crates/buzz-db/src/store/channel.rs
@@ -163,8 +163,7 @@ pub async fn create_channel(
     .map_err(|e| listed_name_taken(e, Some(name)))?;
     ensure_listed_channel_name_unique(&mut tx, id).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
@@ -172,14 +171,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#"
@@ -407,8 +407,7 @@ async fn insert_channel_rows(
     if was_created {
         ensure_listed_channel_name_unique(tx, channel_id).await?;
         // Bootstrap the creator as owner.
-        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
@@ -416,14 +415,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()
@@ -435,19 +435,24 @@ async fn insert_channel_rows(
                     member.len()
                 )));
             }
-            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?;
         }
     }
 
@@ -1253,6 +1258,22 @@ async fn update_channel_on(
         set_parts.join(", ")
     );
 
+    // The current name is what a visibility or unlisted change that
+    // [`LISTED_NAME_INDEX`] refuses would have listed.
+    let (policy_changed, current_name) = if updates.visibility.is_some()
+        || updates.ttl_seconds.is_some()
+        || updates.unlisted.is_some()
+    {
+        sqlx::query_as::<_,(bool, String)>("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)), name 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
+            .map(|(changed, name)| (changed, Some(name)))?
+    } else {
+        (false, None)
+    };
+    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);
@@ -1280,25 +1301,13 @@ async fn update_channel_on(
     }
     q = q.bind(community_id.as_uuid());
     q = q.bind(channel_id);
+    q = q.bind(operation);
 
-    // The current name is what a visibility or unlisted change that
-    // [`LISTED_NAME_INDEX`] refuses would have listed.
-    let (policy_changed, current_name) = if updates.visibility.is_some()
-        || updates.ttl_seconds.is_some()
-        || updates.unlisted.is_some()
-    {
-        sqlx::query_as::<_,(bool, String)>("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)), name 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
-            .map(|(changed, name)| (changed, Some(name)))?
-    } else {
-        (false, None)
-    };
     let result = q
         .execute(tx.conn())
         .await
         .map_err(|e| listed_name_taken(e, updates.name.as_deref().or(current_name.as_deref())))?;
+
     if result.rows_affected() == 0 {
         return Err(DbError::ChannelNotFound(channel_id));
     }
@@ -1307,7 +1316,7 @@ async fn update_channel_on(
     }
     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())
@@ -1723,12 +1732,19 @@ async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> {
     }
     let channel_id = w.channel_id;
     let community_id = w.tx.community();
-    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=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 96501f0227c..702bd840c38 100644
--- a/crates/buzz-db/src/store/channel_members.rs
+++ b/crates/buzz-db/src/store/channel_members.rs
@@ -253,8 +253,7 @@ impl ChannelMembershipLocks<'_> {
     ) -> Result<()> {
         self.check_covers(channel_id)?;
         let community_id = self.tx.community();
-        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
@@ -262,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(())
     }
 }
@@ -1017,8 +1017,7 @@ async fn put_members_on(
     let mut written: std::collections::HashMap<Vec<u8>, MemberRecord> =
         std::collections::HashMap::with_capacity(write_keys.len());
     if !write_keys.is_empty() {
-        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 +1027,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,19 +1154,24 @@ pub async fn remove_member(
         }
     }
 
-    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/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..c05371eca60 100644
--- a/crates/buzz-db/src/store/dm.rs
+++ b/crates/buzz-db/src/store/dm.rs
@@ -187,8 +187,7 @@ pub async fn create_dm(
 
     // Add all participants as members with role='member'.
     for pk in participants {
-        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
@@ -196,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 30371af4bb3..9d2a772cc94 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -1625,8 +1625,16 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx(
     let deleted = purged > 0 || result.rows_affected() > 0;
 
     if deleted {
-        sqlx::query("UPDATE artifact_heads SET modified_at=clock_timestamp() WHERE community_id=$1 AND artifact_id=(SELECT artifact_id FROM artifact_revisions WHERE community_id=$1 AND event_id=$2) AND event_id=$2")
-            .bind(community_id.as_uuid()).bind(event_id).execute(tx.conn()).await?;
+        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/mod.rs b/crates/buzz-db/src/store/mod.rs
index bd2d709151b..b4a31fe1a64 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..3813ff19314 100644
--- a/crates/buzz-db/src/store/relay_admin_actions.rs
+++ b/crates/buzz-db/src/store/relay_admin_actions.rs
@@ -718,19 +718,24 @@ pub async fn execute_kick_with_marker(
     crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id)
         .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?;
@@ -1726,19 +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.
-    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 {
@@ -3903,6 +3913,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/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..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
@@ -1,5 +1,46 @@
 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;
+    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);
+    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 8b50f748939..c5ce1779754 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..2a85a254c99
--- /dev/null
+++ b/crates/buzz-relay/src/artifact_certification.rs
@@ -0,0 +1,47 @@
+//! 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<AppState>) {
+    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;
+        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)) => {
+                    if completed == 0 && next == after {
+                        break;
+                    }
+                    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/src/lib.rs b/crates/buzz-relay/src/lib.rs
index f21d0181af8..164ab6fc691 100644
--- a/crates/buzz-relay/src/lib.rs
+++ b/crates/buzz-relay/src/lib.rs
@@ -23,6 +23,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 f5fe7d0913c..e92db384c31 100644
--- a/crates/buzz-relay/src/main.rs
+++ b/crates/buzz-relay/src/main.rs
@@ -1470,6 +1470,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/crates/buzz-relay/tests/artifact_certification_e2e.rs b/crates/buzz-relay/tests/artifact_certification_e2e.rs
new file mode 100644
index 00000000000..aecd28d78bc
--- /dev/null
+++ b/crates/buzz-relay/tests/artifact_certification_e2e.rs
@@ -0,0 +1,250 @@
+//! Signed HTTP coverage against a relay process and an isolated database.
+
+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();
+        }
+    }
+
+    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 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();
+        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 = || {
+            let output = log.reopen().unwrap();
+            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_PUSH_ENABLED", "false")
+                    .env("BUZZ_AUDIT_ENABLED", "false")
+                    .env("BUZZ_GIT_CONFORMANCE_PROBE", "false")
+                    .stdout(output.try_clone().unwrap())
+                    .stderr(output)
+                    .spawn()
+                    .unwrap(),
+            )
+        };
+        let client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(5))
+            .build()
+            .unwrap();
+        let relay = spawn();
+        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) 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();
+
+        let _relay = spawn();
+        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 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();
+            if remaining == 0 {
+                break;
+            }
+            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()));
+        println!(
+            "PASS signed HTTP, durable restart recovery, existing checkpoints, ordinary reads"
+        );
+    }
+
+    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()
+        );
+    }
+}
diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md
index e4bf6abb82e..d80e6229c3e 100644
--- a/docs/artifact-sync-deployment.md
+++ b/docs/artifact-sync-deployment.md
@@ -55,5 +55,58 @@ All instances must use the same configured relay signing key, because that key
 authenticates the progress tokens. NIP-11 advertises `artifacts.incremental`.
 Every serving writer must maintain the modification timestamps.
 
-The 120-second stamp-to-visibility bound remains an operating assumption.
-Runtime measurements are tracked separately.
+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 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`.
+
+Each supported writer records one pending operation in its admitted transaction.
+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.
+
+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
+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
+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.
+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 or a production maintenance write.
diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md
index da0c36a17db..7c99bd8d77b 100644
--- a/docs/nips/NIP-AR.md
+++ b/docs/nips/NIP-AR.md
@@ -290,9 +290,16 @@ 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 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.
+
+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 +338,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"`.
diff --git a/migrations/0081_artifact_sync_certification.sql b/migrations/0081_artifact_sync_certification.sql
new file mode 100644
index 00000000000..dedef560b94
--- /dev/null
+++ b/migrations/0081_artifact_sync_certification.sql
@@ -0,0 +1,12 @@
+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_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..08688be294c
--- /dev/null
+++ b/schema/tables/public/artifact_sync_pending.sql
@@ -0,0 +1,9 @@
+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_community
+    ON artifact_sync_pending (community_id, earliest_at, operation_id);