fix(relay): serialize archived channel writes (NEXUS-4111)

crates/buzz-admin/src/main.rs

@@ -818,8 +818,14 @@ async fn reconcile_channels(
                     .tags(tags)
                     .sign_with_keys(&relay_keys)
                     .map_err(|e| anyhow::anyhow!("sign kind:39000: {e}"))?;
-                db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
-                    .await?;
+                db.replace_addressable_event(
+                    tenant.community(),
+                    &event,
+                    Some(channel.id),
+                    // Relay discovery state: archived channels still describe themselves.
+                    buzz_db::ArchiveGate::Exempt,
+                )
+                .await?;
             }
 
             // kind:39001 — admins
@@ -841,8 +847,14 @@ async fn reconcile_channels(
                     .tags(tags)
                     .sign_with_keys(&relay_keys)
                     .map_err(|e| anyhow::anyhow!("sign kind:39001: {e}"))?;
-                db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
-                    .await?;
+                db.replace_addressable_event(
+                    tenant.community(),
+                    &event,
+                    Some(channel.id),
+                    // Relay discovery state: archived channels still describe themselves.
+                    buzz_db::ArchiveGate::Exempt,
+                )
+                .await?;
             }
         }
 
@@ -862,8 +874,14 @@ async fn reconcile_channels(
                 .tags(tags)
                 .sign_with_keys(&relay_keys)
                 .map_err(|e| anyhow::anyhow!("sign kind:39002: {e}"))?;
-            db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
-                .await?;
+            db.replace_addressable_event(
+                tenant.community(),
+                &event,
+                Some(channel.id),
+                // Relay discovery state: archived channels still describe themselves.
+                buzz_db::ArchiveGate::Exempt,
+            )
+            .await?;
         }
 
         reconciled += 1;

crates/buzz-db/src/lib.rs

@@ -66,11 +66,11 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, artifact, artifact_sync, channel,
-    channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, lifecycle,
-    moderation, nip_fi_binding, operator_listener, partition, personal_read, product_feedback,
-    push, reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators,
-    reminder, replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion,
-    user_deletion_fences, workflow,
+    channel_members, community, deletion, dm, entity_deletion, event, event_deletion, feed,
+    git_repo, lifecycle, moderation, nip_fi_binding, operator_listener, partition, personal_read,
+    product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members,
+    relay_operators, reminder, replaceable, storage_accounting, thread, thread_window, usage, user,
+    user_deletion, user_deletion_fences, workflow,
 };
 
 pub use allowlist::AllowlistEntry;
@@ -83,6 +83,7 @@ pub use community::{
 pub use error::{DbError, Result};
 pub use event::{ChannelHeadPrecondition, ChannelHeadWriteStatus};
 pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT};
+pub use lifecycle::{ArchiveGate, WriteGates};
 pub use nip_fi_binding::{NipFiBindingRecord, NipFiUnbind, RowCapExceeded};
 pub use reaction::ReactionEventInsertOutcome;
 pub use reminder::DueReminder;

crates/buzz-db/src/runtime/observability.rs

@@ -1567,7 +1567,7 @@ mod tests {
             .with_inactive_community_fence(test_scope, |_| ())
             .await;
         let _ = writer_db.usage_community_count().await;
-        let _ = writer_db.reap_expired_ephemeral_channels().await;
+        let _ = writer_db.reap_expired_ephemeral_channels(1).await;
         let deletion_store = writer_db.deletion_store();
         let _ = deletion_store.reap_expired_serving_write_leases(1).await;
         let _ = deletion_store.serving_lease_stats().await;

crates/buzz-db/src/runtime/tests.rs

@@ -591,7 +591,13 @@ async fn migration_schema_database_guard_covers_legacy_writer_and_nip09_deletion
     let replace_c = c.clone();
     let replace_task = tokio::spawn(async move {
         replace_db
-            .replace_parameterized_event(community, &replace_c, &replace_d_tag, None)
+            .replace_parameterized_event(
+                community,
+                &replace_c,
+                &replace_d_tag,
+                None,
+                crate::ArchiveGate::Exempt,
+            )
             .await
     });
     tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -930,6 +936,7 @@ async fn insert_top_level(pool: &PgPool, community: Uuid, channel: Uuid, ev: &no
             depth: 0,
             broadcast: true,
         }),
+        crate::ArchiveGate::Exempt,
     )
     .await
     .expect("insert top-level event");
@@ -962,6 +969,7 @@ async fn insert_thread_reply(
             depth: 1,
             broadcast: false,
         }),
+        crate::ArchiveGate::Exempt,
     )
     .await
     .expect("insert reply");
@@ -3283,9 +3291,15 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() {
 
     // insert_event (single INSERT, autocommit): old channel row rejected.
     let old = signed_event_at(&author, "old-direct", now_secs - floor - 60);
-    let err = event::insert_event(&db.pool, cid, &old, Some(channel))
-        .await
-        .expect_err("armed pool must reject below-floor channel inserts");
+    let err = event::insert_event(
+        &db.pool,
+        cid,
+        &old,
+        Some(channel),
+        crate::ArchiveGate::Exempt,
+    )
+    .await
+    .expect_err("armed pool must reject below-floor channel inserts");
     assert!(
         err.to_string().contains("below the replica-fence floor"),
         "unexpected error: {err}"
@@ -3311,6 +3325,7 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() {
             depth: 0,
             broadcast: true,
         }),
+        crate::ArchiveGate::Exempt,
     )
     .await
     .expect_err("armed pool must reject below-floor thread-metadata inserts");
@@ -3321,9 +3336,15 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() {
 
     // Fresh events pass through both APIs.
     let fresh = signed_event_at(&author, "fresh-direct", now_secs);
-    event::insert_event(&db.pool, cid, &fresh, Some(channel))
-        .await
-        .expect("fresh insert passes the armed guard");
+    event::insert_event(
+        &db.pool,
+        cid,
+        &fresh,
+        Some(channel),
+        crate::ArchiveGate::Exempt,
+    )
+    .await
+    .expect("fresh insert passes the armed guard");
 
     // Close before the `scratch` guard force-drops the database on return.
     db.pool.close().await;

crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs

@@ -87,10 +87,16 @@ async fn thread_window_upper_fence_terminal_snapshot_and_fallback() {
         .sign_with_keys(&keys)
         .unwrap();
     for pool in [&writer, &replica] {
-        event::insert_event(pool, community, &edit, Some(channel))
-            .await
-            .unwrap();
-        event::insert_event(pool, community, &deletion, None)
+        event::insert_event(
+            pool,
+            community,
+            &edit,
+            Some(channel),
+            crate::ArchiveGate::Exempt,
+        )
+        .await
+        .unwrap();
+        event::insert_event(pool, community, &deletion, None, crate::ArchiveGate::Exempt)
             .await
             .unwrap();
     }

crates/buzz-db/src/store/admission_postgres_tests.rs

@@ -318,16 +318,24 @@ async fn workflow_writers_require_application_admission() {
             .expect_err("workflow create must reject"),
         );
         assert_fenced(
-            super::workflow::create_workflow_run(&db.pool, *community, Uuid::new_v4(), None, None)
-                .await
-                .expect_err("workflow run create must reject"),
+            super::workflow::create_workflow_run(
+                &db.pool,
+                *community,
+                Uuid::new_v4(),
+                None,
+                None,
+                crate::ArchiveGate::Exempt,
+            )
+            .await
+            .expect_err("workflow run create must reject"),
         );
         assert_fenced(
             super::workflow::claim_scheduled_workflow_fire(
                 &db.pool,
                 *community,
                 Uuid::new_v4(),
                 chrono::Utc::now(),
+                crate::ArchiveGate::Exempt,
             )
             .await
             .expect_err("schedule claim must reject"),
@@ -644,9 +652,14 @@ async fn personal_read_writer_requires_application_admission() {
     let event = EventBuilder::new(Kind::Custom(9), "read")
         .sign_with_keys(&Keys::generate())
         .expect("sign event");
-    db.insert_event(communities[0], &event, Some(channel.id))
-        .await
-        .expect("insert active event");
+    db.insert_event(
+        communities[0],
+        &event,
+        Some(channel.id),
+        crate::ArchiveGate::Exempt,
+    )
+    .await
+    .expect("insert active event");
     let intent = super::personal_read::ReadIntent::MarkThrough {
         target: super::personal_read::ReadTarget {
             channel_id: channel.id,
@@ -903,7 +916,7 @@ async fn channel_reap_admits_active_tenant_and_skips_quiescing_tenant() {
         .await
         .expect("seed channel");
     }
-    let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool)
+    let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool, 1000)
         .await
         .unwrap();
     assert_eq!(reaped.len(), 1);
@@ -1000,7 +1013,7 @@ async fn channel_reap_returns_committed_rows_after_earlier_tenant_error() {
         .await
         .expect("seed expired channel");
     }
-    let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool)
+    let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool, 1000)
         .await
         .expect("tenant failure must not stop the fleet sweep");
     assert_eq!(reaped.len(), 1);
@@ -1021,7 +1034,7 @@ async fn channel_reap_returns_committed_rows_after_earlier_tenant_error() {
     .execute(&db.pool)
     .await
     .unwrap();
-    let retried = super::channel::reap_expired_ephemeral_channels(&db.pool)
+    let retried = super::channel::reap_expired_ephemeral_channels(&db.pool, 1000)
         .await
         .unwrap();
     assert_eq!(retried.len(), 1);
@@ -1181,7 +1194,7 @@ async fn channel_reap_rechecks_admission_after_candidate_read() {
     .unwrap();
     let pool = db.pool.clone();
     let reaped = crate::test_support::fence_while_sweep_waits(community, async move {
-        super::channel::reap_expired_ephemeral_channels(&pool).await
+        super::channel::reap_expired_ephemeral_channels(&pool, 1000).await
     })
     .await
     .unwrap();

crates/buzz-db/src/store/artifact.rs

@@ -4,12 +4,13 @@
 //! before this transaction; here the relay only serializes each identity and
 //! compares the expected head.
 use crate::lifecycle::EdgeChange;
-use crate::{Db, DbError, Result};
+use crate::{ArchiveGate, Db, DbError, Result};
 use buzz_core::artifact::{ArtifactEnvelope, ArtifactOp};
 use buzz_core::lifecycle::{task_edge, TaskEdge, TASK_ARTIFACT_TYPE};
 use buzz_core::{CommunityId, StoredEvent};
 use nostr::{Event, EventBuilder, Keys, Kind, Tag};
 use sqlx::Row;
+use std::collections::BTreeSet;
 use uuid::Uuid;
 
 /// Acceptance outcome. Only a stale-`prev` conflict carries the current head,
@@ -101,6 +102,14 @@ impl Db {
     /// transaction, under its family locks. With `lifecycle_guard` off, a
     /// `parent` tag is refused unless the channel already has that projected
     /// parent, and a channel without a projection row keeps today's behavior.
+    ///
+    /// `gate` decides whether an archived channel refuses the revision (see
+    /// [`ArchiveGate`]). Under [`ArchiveGate::Refuse`] the revision takes the
+    /// gates of its home and, for a move, of `authorized_source`, after the
+    /// family locks of a task. An archived home fails with
+    /// [`DbError::AccessDenied`]; an archived move source answers like a
+    /// missing head.
+    #[allow(clippy::too_many_arguments)]
     pub async fn accept_artifact(
         &self,
         community: CommunityId,
@@ -109,6 +118,7 @@ impl Db {
         authorized_source: Option<Uuid>,
         relay_keys: &Keys,
         lifecycle_guard: bool,
+        gate: ArchiveGate,
     ) -> Result<ArtifactOutcome> {
         // A family root that moves before its lock is taken is retried once.
         for _ in 0..2 {
@@ -120,6 +130,7 @@ impl Db {
                     authorized_source,
                     relay_keys,
                     lifecycle_guard,
+                    gate,
                 )
                 .await?
             {
@@ -129,6 +140,7 @@ impl Db {
         Ok(ArtifactOutcome::Conflict("task family changed"))
     }
 
+    #[allow(clippy::too_many_arguments)]
     async fn accept_artifact_once(
         &self,
         community: CommunityId,
@@ -137,11 +149,20 @@ impl Db {
         authorized_source: Option<Uuid>,
         relay_keys: &Keys,
         lifecycle_guard: bool,
+        gate: ArchiveGate,
     ) -> Result<Option<ArtifactOutcome>> {
         let mut tx = self.begin_event_write_transaction(community).await?;
         sqlx::query("SET LOCAL statement_timeout='5s'")
             .execute(tx.conn())
             .await?;
+        // The channels this revision writes: its home and a move's source.
+        let mut gated = BTreeSet::new();
+        if gate == ArchiveGate::Refuse {
+            gated.insert(env.home);
+            if env.op == ArtifactOp::Move {
+                gated.extend(authorized_source);
+            }
+        }
         let is_task = env.artifact_type == TASK_ARTIFACT_TYPE;
         // A task channel, or any task revision once the guard is on, takes
         // the hierarchy path; the read is re-checked under the family lock.
@@ -154,15 +175,31 @@ impl Db {
                 Err(message) => return Ok(Some(ArtifactOutcome::Rejected(message))),
             };
             let author = event.pubkey.to_bytes();
-            match crate::lifecycle::lock_task_edge(&mut tx, edge.channel, edge.parent, &author)
-                .await?
+            match crate::lifecycle::lock_task_edge(
+                &mut tx,
+                edge.channel,
+                edge.parent,
+                &author,
+                &gated,
+            )
+            .await?
             {
                 Some(locks) => Some((edge, locks)),
                 None => return Ok(None),
             }
         } else {
+            crate::lifecycle::lock_shared_gates(&mut tx, &gated).await?;
             None
         };
+        let archived = crate::lifecycle::archived_among(&mut tx, &gated).await?;
+        if archived.contains(&env.home) {
+            return Err(DbError::AccessDenied(
+                crate::lifecycle::CHANNEL_ARCHIVED.into(),
+            ));
+        }
+        if !archived.is_empty() {
+            return Ok(Some(ArtifactOutcome::Conflict(HEAD_UNAVAILABLE)));
+        }
         // A coordinate lock handles the missing-row create race as well as edits.
         sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1,0))")
             .bind(format!("artifact:{}:{}", community.as_uuid(), env.id))