fix(git): recheck archive before pointer writes (NEXUS-4111)

crates/buzz-relay/src/api/git/cas_publish.rs

@@ -141,6 +141,11 @@ pub enum CasError {
     /// The push would make the repo exceed the relay's configured byte budget.
     #[error("resource limit: {0}")]
     ResourceLimit(String),
+
+    /// The caller's last check before the CAS refused the write. Pre-CAS:
+    /// the pointer was never written.
+    #[error("refused before CAS: {0}")]
+    Refused(&'static str),
 }
 
 /// Resource limits carried from hydration into the publish step.
@@ -994,6 +999,10 @@ fn record_compaction(
 /// at step 7 is the *only* writer serialization (`Inv_NoFork`). No
 /// advisory lock — adding one would hide exactly the interleavings the
 /// model proves safe.
+///
+/// `before_cas` is the caller's last check. It runs after every upload and
+/// immediately before the pointer CAS; an error refuses the write.
+#[allow(clippy::too_many_arguments)]
 pub async fn cas_publish(
     store: &GitStore,
     ctx: &TenantContext,
@@ -1002,6 +1011,7 @@ pub async fn cas_publish(
     repo: &str,
     parent_state: &ParentState,
     limits: PublishLimits,
+    before_cas: impl std::future::Future<Output = Result<(), CasError>>,
 ) -> Result<CasSuccess, CasError> {
     cas_publish_inner(
         store,
@@ -1014,10 +1024,12 @@ pub async fn cas_publish(
             limits,
             compaction_threshold: PACK_COMPACTION_THRESHOLD,
         },
+        before_cas,
     )
     .await
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn cas_publish_inner(
     store: &GitStore,
     ctx: &TenantContext,
@@ -1026,6 +1038,7 @@ async fn cas_publish_inner(
     repo: &str,
     parent_state: &ParentState,
     options: PublishOptions,
+    before_cas: impl std::future::Future<Output = Result<(), CasError>>,
 ) -> Result<CasSuccess, CasError> {
     let limits = options.limits;
     let pkey = pointer_key(ctx.community(), owner, repo);
@@ -1208,6 +1221,10 @@ async fn cas_publish_inner(
     };
     let manifest_digest = digest_from_manifest_key(&manifest_key)?;
 
+    // The caller's last check, after the uploads and immediately before the
+    // CAS (NEXUS-4111): the window it leaves is only the pointer PUT.
+    before_cas.await?;
+
     // Step 7: CAS the pointer.
     let precond = match &parent_state.if_match {
         Some(e) => Precond::IfMatch(e.clone()),
@@ -1772,6 +1789,7 @@ mod tests {
             repo,
             &parent_state,
             test_options,
+            std::future::ready(Ok(())),
         )
         .await
         .expect("compaction publish");
@@ -1791,6 +1809,7 @@ mod tests {
             repo,
             &parent_state,
             test_options,
+            std::future::ready(Ok(())),
         )
         .await
         .expect_err("stale parent must lose CAS");

crates/buzz-relay/src/api/git/settings.rs

@@ -28,7 +28,10 @@ use super::{
     manifest::{is_safe_refname, pointer_key, Manifest},
     manifest_event::{build_ref_state_event, RefStateInputs},
     store::{CasOutcome, ETag, GitStore, Precond},
-    transport::{authorize_git_read, deny_banned_git_principal, validate_repo_id},
+    transport::{
+        authorize_git_read, deny_banned_git_principal, repo_archive_state, validate_repo_id,
+        LastCheck,
+    },
 };
 use crate::{
     api::{api_error, bridge, relay_members},
@@ -102,6 +105,7 @@ impl DefaultBranchSnapshot {
         mut self,
         store: &GitStore,
         request: SetDefaultBranch,
+        last_check: impl std::future::Future<Output = Result<(), Response>>,
     ) -> Result<(Self, bool), Response> {
         if self.digest != request.expected_manifest {
             return Err(conflict());
@@ -140,6 +144,8 @@ impl DefaultBranchSnapshot {
                 .ok_or_else(|| backend("invalid manifest key"))?
                 .to_string();
         }
+        // The last archive check, immediately before the CAS (NEXUS-4111).
+        last_check.await?;
         // Even a no-op checks the observed ETag: concurrent deletion/push must
         // not be reported as a successful setting of a now-missing branch.
         match store
@@ -389,8 +395,25 @@ async fn set_default_branch(
         )
     })?;
     serving_write.verify().await.map_err(backend)?;
+    // It re-reads the binding and refuses with `authorize_management`'s
+    // outcomes.
+    let last_check = async {
+        #[cfg(test)]
+        crate::nip_fi_test_hooks::before_git_settings_last_check(auth.tenant.community()).await;
+        match repo_archive_state(&state.db, auth.tenant.community(), &owner, repo_name).await {
+            Ok(LastCheck::Bound(false)) => Ok(()),
+            Ok(LastCheck::Bound(true)) => Err(error(
+                StatusCode::FORBIDDEN,
+                "channel is archived (read-only)",
+            )),
+            Ok(LastCheck::Missing | LastCheck::NotBound | LastCheck::Broken) => {
+                Err(error(StatusCode::NOT_FOUND, "repository not found"))
+            }
+            Err(e) => Err(backend(e)),
+        }
+    };
     let (snapshot, changed) = serving_write
-        .protect(snapshot.set(&state.git_store, request))
+        .protect(snapshot.set(&state.git_store, request, last_check))
         .await
         .map_err(backend)??;
     let publication = async {

crates/buzz-relay/src/api/git/settings_tests.rs

@@ -1341,6 +1341,7 @@ mod external_infra {
                 &self.repo,
                 &super::super::super::cas_publish::ParentState::fresh(),
                 limits(0),
+                std::future::ready(Ok(())),
             )
             .await
             .unwrap();
@@ -1746,6 +1747,134 @@ mod external_infra {
         );
     }
 
+    /// NEXUS-4111, Git option A: the last check before the pointer CAS reads
+    /// the announcement again and the channel's archive state under its
+    /// gate. A change made after `authorize_management` passed meets that
+    /// check and gets `authorize_management`'s reply; nothing is written.
+    #[tokio::test]
+    #[ignore = "requires isolated Postgres, Redis and MinIO"]
+    async fn default_branch_last_archive_check_refuses() {
+        enum Change {
+            Bind(Option<&'static str>),
+            Delete,
+            Archive,
+        }
+        let f = Fixture::new().await;
+        let community = f.tenant.community();
+        let before = f.snapshot().await;
+        let channel = f.channel.to_string();
+        let mut at = nostr::Timestamp::now().as_secs() + 10;
+        let mut announce = |binding: Option<&str>| {
+            let mut tags = vec![
+                Tag::parse(["d", &f.repo]).unwrap(),
+                Tag::parse(["maintainers", &f.maintainer.public_key().to_hex()]).unwrap(),
+            ];
+            if let Some(binding) = binding {
+                tags.push(Tag::parse(["buzz-channel", binding]).unwrap());
+            }
+            at += 1;
+            EventBuilder::new(Kind::Custom(30617), "")
+                .tags(tags)
+                .custom_created_at(nostr::Timestamp::from(at))
+                .sign_with_keys(&f.owner)
+                .unwrap()
+        };
+        let exempt = buzz_db::ArchiveGate::Exempt;
+        for (change, status, reason) in [
+            (
+                Change::Bind(None),
+                StatusCode::NOT_FOUND,
+                "repository not found",
+            ),
+            (
+                Change::Bind(Some("not-a-uuid")),
+                StatusCode::NOT_FOUND,
+                "repository not found",
+            ),
+            (
+                Change::Delete,
+                StatusCode::NOT_FOUND,
+                "repository not found",
+            ),
+            (
+                Change::Archive,
+                StatusCode::FORBIDDEN,
+                "channel is archived (read-only)",
+            ),
+        ] {
+            let rebound = announce(Some(&channel));
+            let replaced = f
+                .state
+                .db
+                .replace_parameterized_event(community, &rebound, &f.repo, None, exempt);
+            assert!(replaced.await.unwrap().1, "bound again before the row");
+            let changed = match &change {
+                Change::Bind(binding) => Some(announce(*binding)),
+                Change::Delete | Change::Archive => None,
+            };
+            let (arrived, release) =
+                crate::nip_fi_test_hooks::git_settings_last_check_hook::arm(community);
+            let change_at_check = async {
+                let reached = tokio::time::timeout(std::time::Duration::from_secs(10), arrived);
+                reached
+                    .await
+                    .expect("the change reaches its last check")
+                    .unwrap();
+                match (&change, &changed) {
+                    (Change::Bind(_), Some(event)) => {
+                        let replaced = f
+                            .state
+                            .db
+                            .replace_parameterized_event(community, event, &f.repo, None, exempt);
+                        assert!(replaced.await.unwrap().1);
+                    }
+                    (Change::Delete, _) => {
+                        sqlx::query(
+                            "UPDATE events SET deleted_at = NOW() \
+                             WHERE community_id = $1 AND kind = 30617 AND d_tag = $2",
+                        )
+                        .bind(community.as_uuid())
+                        .bind(&f.repo)
+                        .execute(&f.pool)
+                        .await
+                        .unwrap();
+                    }
+                    _ => {
+                        sqlx::query(
+                            "UPDATE channels SET archived_at = NOW() \
+                             WHERE community_id = $1 AND id = $2",
+                        )
+                        .bind(community.as_uuid())
+                        .bind(f.channel)
+                        .execute(&f.pool)
+                        .await
+                        .unwrap();
+                    }
+                }
+                release.notify_one();
+            };
+            let (reply, ()) = tokio::join!(f.set(&f.owner, "main", None), change_at_check);
+            assert_eq!(
+                (reply.0, reply.1["error"].as_str()),
+                (status, Some(reason)),
+                "{reply:?}"
+            );
+            let after = f.snapshot().await;
+            assert_eq!(
+                (after.etag, after.digest),
+                (before.etag.clone(), before.digest.clone())
+            );
+        }
+        let notifications = buzz_db::EventQuery {
+            kinds: Some(vec![30618]),
+            d_tag: Some(f.repo.clone()),
+            global_only: true,
+            ..buzz_db::EventQuery::for_community(community)
+        };
+        let notifications = f.state.db.query_events(&notifications).await.unwrap();
+        assert!(notifications.is_empty(), "refused: {notifications:?}");
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated Postgres, Redis and MinIO"]
     async fn default_branch_push_races_and_fresh_clone() {
@@ -1760,6 +1889,7 @@ mod external_infra {
                     branch: "main".into(),
                     expected_manifest: old_digest.clone(),
                 },
+                std::future::ready(Ok(())),
             )
             .await
             .unwrap();
@@ -1771,6 +1901,7 @@ mod external_infra {
                     branch: "legacy".into(),
                     expected_manifest: old_digest,
                 },
+                std::future::ready(Ok(())),
             )
             .await
             .err()
@@ -1805,6 +1936,7 @@ mod external_infra {
             &f.repo,
             &parent,
             limits(push.hydrated_bytes()),
+            std::future::ready(Ok(())),
         )
         .await;
         assert!(matches!(
@@ -1832,6 +1964,7 @@ mod external_infra {
             &f.repo,
             &parent,
             limits(push.hydrated_bytes()),
+            std::future::ready(Ok(())),
         )
         .await
         .unwrap();
@@ -1842,7 +1975,8 @@ mod external_infra {
                     SetDefaultBranch {
                         branch: "main".into(),
                         expected_manifest: digest
-                    }
+                    },
+                    std::future::ready(Ok(())),
                 )
                 .await
                 .err()
@@ -1882,6 +2016,7 @@ mod external_infra {
             &f.repo,
             &parent,
             limits(push.hydrated_bytes()),
+            std::future::ready(Ok(())),
         )
         .await
         .unwrap();
@@ -1908,6 +2043,7 @@ mod external_infra {
             &f.repo,
             &parent,
             limits(push.hydrated_bytes()),
+            std::future::ready(Ok(())),
         )
         .await
         .unwrap();