diff --git a/crates/buzz-relay/src/api/git/cas_publish.rs b/crates/buzz-relay/src/api/git/cas_publish.rs index 50bb36d818c..40721e717dc 100644 --- a/crates/buzz-relay/src/api/git/cas_publish.rs +++ b/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>, ) -> Result { 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>, ) -> Result { 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"); diff --git a/crates/buzz-relay/src/api/git/settings.rs b/crates/buzz-relay/src/api/git/settings.rs index f11e11026b1..4dead443c68 100644 --- a/crates/buzz-relay/src/api/git/settings.rs +++ b/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>, ) -> 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 { diff --git a/crates/buzz-relay/src/api/git/settings_tests.rs b/crates/buzz-relay/src/api/git/settings_tests.rs index 30a2a65f1e4..555f5922686 100644 --- a/crates/buzz-relay/src/api/git/settings_tests.rs +++ b/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(¬ifications).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(); diff --git a/crates/buzz-relay/src/api/git/transport.rs b/crates/buzz-relay/src/api/git/transport.rs index 3e75c20caf2..6922a61fed4 100644 --- a/crates/buzz-relay/src/api/git/transport.rs +++ b/crates/buzz-relay/src/api/git/transport.rs @@ -726,6 +726,67 @@ fn read_role_allows(role: Option<&str>) -> bool { } } +/// A repository's state at the last check before a pointer CAS (NEXUS-4111). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum LastCheck { + /// No current kind:30617 for this repo. + Missing, + /// The announcement has no `buzz-channel` tag. + NotBound, + /// The first `buzz-channel` tag is not a UUID. + Broken, + /// Bound to a channel; `true` when that channel is archived. + Bound(bool), +} + +/// Read the repo's current kind:30617 again, resolve its binding, and read +/// the bound channel's archive state under its archive gate. +/// +/// The gate read is a short admitted transaction that takes the channel's +/// gate shared (see `buzz_db::lifecycle`), so an archive that holds the gate +/// is waited for and then seen. It does no I/O under the gate. A write that +/// passes may still land after a later archive commits, until its pointer +/// PUT completes: see `docs/git-on-object-storage.md`. +pub(super) async fn repo_archive_state( + db: &buzz_db::Db, + community: buzz_core::CommunityId, + owner_hex: &str, + repo_name: &str, +) -> buzz_db::Result { + let owner = hex::decode(owner_hex) + .map_err(|_| buzz_db::DbError::InvalidData("invalid repo owner".into()))?; + let query = buzz_db::EventQuery { + kinds: Some(vec![30617]), + pubkey: Some(owner), + d_tag: Some(repo_name.to_string()), + global_only: true, + limit: Some(1), + ..buzz_db::EventQuery::for_community(community) + }; + let Some(repo) = db.query_events(&query).await?.pop() else { + return Ok(LastCheck::Missing); + }; + let channel = match resolve_repo_binding(&repo.event) { + RepoBinding::NotBound => return Ok(LastCheck::NotBound), + RepoBinding::Broken => return Ok(LastCheck::Broken), + RepoBinding::Bound(channel) => channel, + }; + let mut tx = db.begin_event_write_transaction(community).await?; + let gated = + buzz_db::lifecycle::lock_write_gates(&mut tx, buzz_db::ArchiveGate::Refuse, &[channel]) + .await; + tx.rollback().await?; + match gated { + Ok(_) => Ok(LastCheck::Bound(false)), + Err(buzz_db::DbError::AccessDenied(reason)) + if reason == buzz_db::lifecycle::CHANNEL_ARCHIVED => + { + Ok(LastCheck::Bound(true)) + } + Err(error) => Err(error), + } +} + #[derive(Deserialize)] /// Query parameters for the `info/refs` endpoint. pub struct InfoRefsQuery { @@ -1952,6 +2013,9 @@ struct FinalizePushHooks { post_cas_gate: Option>, #[cfg(test)] fail_ref_state_insert: bool, + /// Pauses after the last archive check passes, before the pointer CAS. + #[cfg(test)] + pre_cas_gate: Option>, } #[cfg(test)] @@ -2042,6 +2106,32 @@ async fn finalize_push_inner( .into_response(); } + // The last archive check, immediately before the CAS. It re-reads the + // binding and refuses with the pre-receive's outcomes (policy.rs). + let last_check = async { + let community = ctx.tenant.community(); + let refused = match repo_archive_state(&state.db, community, &ctx.owner, &ctx.repo_id).await + { + Ok(LastCheck::NotBound | LastCheck::Bound(false)) => None, + Ok(LastCheck::Bound(true)) => Some("channel is archived (read-only)"), + Ok(LastCheck::Missing) => Some("repository not found"), + Ok(LastCheck::Broken) => Some("invalid channel binding"), + Err(error) => { + error!(owner = %ctx.owner, repo = %ctx.repo, %error, "push: last archive check failed"); + Some("internal error") + } + }; + if let Some(reason) = refused { + return Err(CasError::Refused(reason)); + } + #[cfg(test)] + if let Some(gate) = &hooks.pre_cas_gate { + gate.reached.notify_one(); + gate.resume.notified().await; + } + Ok(()) + }; + // Step 7 (CAS). The PushContext binds `parent_state` (observed at // hydrate) to the CAS predicate here — no re-reading of the pointer // between hydrate and CAS. Observe serving-lease loss throughout the @@ -2058,6 +2148,7 @@ async fn finalize_push_inner( max_pack_bytes: state.config.git_max_pack_bytes, max_repo_bytes: state.config.git_max_repo_bytes, }, + last_check, ); let success = match serving_write.protect(publish).await { Ok(result) => match result { @@ -2095,6 +2186,10 @@ async fn finalize_push_inner( ) .into_response(); } + Err(CasError::Refused(reason)) => { + warn!(owner = %ctx.owner, repo = %ctx.repo, reason, "push refused at the last check before CAS"); + return (StatusCode::FORBIDDEN, reason).into_response(); + } Err(CasError::ResourceLimit(e)) => { warn!( owner = %ctx.owner, @@ -2700,8 +2795,8 @@ mod track_c_tests { .id; let (request, claim) = approved_deletion(&state, &host).await; let scratch = TempDir::new().expect("scratch"); - let owner = format!("owner-{}", uuid::Uuid::new_v4().simple()); let repo = format!("repo-{}", uuid::Uuid::new_v4().simple()); + let owner = announce(&state, community, &Keys::generate(), &repo, None, 1).await; let ctx = pushed_context( &state, community, @@ -2716,6 +2811,7 @@ mod track_c_tests { let hooks = FinalizePushHooks { post_cas_gate: Some(Arc::clone(&gate)), fail_ref_state_insert: false, + pre_cas_gate: None, }; let finalize_state = Arc::clone(&state); let finalize = @@ -2793,12 +2889,14 @@ mod track_c_tests { .expect("create test community") .id; let scratch = TempDir::new().expect("scratch"); + let repo = format!("repo-{}", uuid::Uuid::new_v4().simple()); + let owner = announce(&state, community, &Keys::generate(), &repo, None, 1).await; let ctx = pushed_context( &state, community, &host, - format!("owner-{}", uuid::Uuid::new_v4().simple()), - format!("repo-{}", uuid::Uuid::new_v4().simple()), + owner, + repo, Keys::generate().public_key(), scratch.path(), ) @@ -2806,6 +2904,7 @@ mod track_c_tests { let hooks = FinalizePushHooks { post_cas_gate: None, fail_ref_state_insert: true, + pre_cas_gate: None, }; let response = finalize_push_inner(&state, ctx, &hooks).await; @@ -2820,6 +2919,274 @@ mod track_c_tests { pool.close().await; } + /// Announce `repo` as `owner` at `created_at`, replacing any earlier + /// announcement, with an optional `buzz-channel` value. Returns the + /// owner's hex key. + async fn announce( + state: &AppState, + community: CommunityId, + owner: &Keys, + repo: &str, + binding: Option<&str>, + created_at: u64, + ) -> String { + let mut tags = vec![Tag::parse(["d", repo]).expect("d tag")]; + if let Some(binding) = binding { + tags.push(Tag::parse(["buzz-channel", binding]).expect("binding tag")); + } + let event = EventBuilder::new(Kind::Custom(30617), "") + .tags(tags) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(owner) + .expect("sign 30617"); + let exempt = buzz_db::ArchiveGate::Exempt; + let replaced = state + .db + .replace_parameterized_event(community, &event, repo, None, exempt); + assert!( + replaced.await.expect("announce").1, + "the announcement is current" + ); + owner.public_key().to_hex() + } + + /// One community, one repo owner, and fresh repos and channels for the + /// last-archive-check journey (NEXUS-4111, Git option A). + struct LastCheckFixture { + state: Arc, + pool: sqlx::PgPool, + community: CommunityId, + host: String, + owner: Keys, + clock: std::sync::atomic::AtomicU64, + } + + impl LastCheckFixture { + async fn new() -> Self { + let (state, pool) = finalize_test_state().await; + let host = format!("git-last-check-{}.example", uuid::Uuid::new_v4().simple()); + let community = state.db.ensure_configured_community(&host).await; + let community = community.expect("create test community").id; + let owner = Keys::generate(); + let owner_key = owner.public_key().to_bytes(); + let user = state.db.ensure_user(community, &owner_key).await; + user.expect("owner user"); + Self { + state, + pool, + community, + host, + owner, + clock: std::sync::atomic::AtomicU64::new(1), + } + } + + async fn channel(&self) -> uuid::Uuid { + let channel = uuid::Uuid::new_v4(); + self.state + .db + .create_channel_with_id( + self.community, + channel, + &format!("git-last-check-{channel}"), + buzz_db::channel::ChannelType::Stream, + buzz_db::channel::ChannelVisibility::Open, + None, + &self.owner.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + channel + } + + async fn archive(&self, channel: uuid::Uuid) { + sqlx::query( + "UPDATE channels SET archived_at = NOW() WHERE community_id = $1 AND id = $2", + ) + .bind(self.community.as_uuid()) + .bind(channel) + .execute(&self.pool) + .await + .expect("archive"); + } + + async fn announce(&self, repo: &str, binding: Option<&str>) { + let at = self + .clock + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + announce(&self.state, self.community, &self.owner, repo, binding, at).await; + } + + /// A new repo announced with `binding`, and its pushed workspace. + async fn pushed(&self, binding: Option<&str>) -> (String, TempDir, PushContext) { + let repo = format!("repo-{}", uuid::Uuid::new_v4().simple()); + self.announce(&repo, binding).await; + let scratch = TempDir::new().expect("scratch"); + let ctx = pushed_context( + &self.state, + self.community, + &self.host, + self.owner.public_key().to_hex(), + repo.clone(), + self.owner.public_key(), + scratch.path(), + ) + .await; + (repo, scratch, ctx) + } + + /// Whether the repo's pointer exists, and its count of kind:30618. + async fn published(&self, repo: &str) -> (bool, usize) { + let owner = self.owner.public_key().to_hex(); + let key = crate::api::git::manifest::pointer_key(self.community, &owner, repo); + let pointer = self.state.git_store.get_pointer(&key).await; + let mut query = buzz_db::event::EventQuery::for_community(self.community); + query.kinds = Some(vec![30_618]); + query.d_tag = Some(repo.to_string()); + let events = self.state.db.query_events(&query).await; + ( + pointer.expect("read pointer").is_some(), + events.expect("query 30618").len(), + ) + } + } + + /// Finalize on its own task (a debug-build finalize future is too large + /// for the test thread's stack) and read its status and body. + async fn finalize_reply( + state: Arc, + ctx: PushContext, + hooks: FinalizePushHooks, + ) -> (StatusCode, String) { + let finalize = tokio::spawn(async move { + let response = finalize_push_inner(&state, ctx, &hooks).await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024).await; + let body = String::from_utf8_lossy(&body.expect("response body")).into_owned(); + (status, body) + }); + finalize.await.expect("finalize task") + } + + /// The last archive check runs immediately before the pointer CAS. It + /// re-reads the announcement, keeps the pre-receive's outcomes, and waits + /// for an archive that holds the channel's gate. A push whose check + /// passed before the archive committed still lands: the documented + /// exception. + async fn finalize_push_last_archive_check_refuses_or_lands() { + const ARCHIVED: &str = "channel is archived (read-only)"; + let f = LastCheckFixture::new().await; + let no_hooks = FinalizePushHooks::default; + let refused = |reason: &str| (StatusCode::FORBIDDEN, reason.to_string()); + + // An unbound repo has no channel to check and proceeds. + let (repo, _scratch, ctx) = f.pushed(None).await; + let reply = finalize_reply(f.state.clone(), ctx, no_hooks()).await; + assert_eq!(reply.0, StatusCode::OK, "{reply:?}"); + assert_eq!(f.published(&repo).await, (true, 1)); + + // A1: the channel is archived after the pre-receive passed. + let archived = f.channel().await; + let (repo, _scratch, ctx) = f.pushed(Some(&archived.to_string())).await; + f.archive(archived).await; + let reply = finalize_reply(f.state.clone(), ctx, no_hooks()).await; + assert_eq!(reply, refused(ARCHIVED)); + assert_eq!(f.published(&repo).await, (false, 0)); + + // A4: a re-bind into an archived channel commits before the check. + let active = f.channel().await; + let (repo, _scratch, ctx) = f.pushed(Some(&active.to_string())).await; + f.announce(&repo, Some(&archived.to_string())).await; + let reply = finalize_reply(f.state.clone(), ctx, no_hooks()).await; + assert_eq!(reply, refused(ARCHIVED)); + assert_eq!(f.published(&repo).await, (false, 0)); + + // A5: a deleted or broken announcement keeps its pre-receive refusal. + let (repo, _scratch, ctx) = f.pushed(Some(&active.to_string())).await; + sqlx::query( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = 30617 AND d_tag = $2", + ) + .bind(f.community.as_uuid()) + .bind(&repo) + .execute(&f.pool) + .await + .expect("delete the announcement"); + let reply = finalize_reply(f.state.clone(), ctx, no_hooks()).await; + assert_eq!(reply, refused("repository not found")); + assert_eq!(f.published(&repo).await, (false, 0)); + let (repo, _scratch, ctx) = f.pushed(Some(&active.to_string())).await; + f.announce(&repo, Some("not-a-uuid")).await; + let reply = finalize_reply(f.state.clone(), ctx, no_hooks()).await; + assert_eq!(reply, refused("invalid channel binding")); + assert_eq!(f.published(&repo).await, (false, 0)); + + // A3: an archive that holds the gate exclusive is waited for, then + // seen. A second session holds it; the check queues SHARED on it. + let gated = f.channel().await; + let (repo, _scratch, ctx) = f.pushed(Some(&gated.to_string())).await; + let gate = format!("buzz_channel_gate:{}:{gated}", f.community.as_uuid()); + let mut holder = f.pool.begin().await.expect("holder"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(&gate) + .execute(&mut *holder) + .await + .expect("hold the gate exclusive"); + sqlx::query("UPDATE channels SET archived_at = NOW() WHERE community_id = $1 AND id = $2") + .bind(f.community.as_uuid()) + .bind(gated) + .execute(&mut *holder) + .await + .expect("archive in the holder"); + let push = tokio::spawn(finalize_reply(f.state.clone(), ctx, no_hooks())); + let queued = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let modes: Vec = sqlx::query_scalar( + "SELECT mode FROM pg_locks \ + WHERE locktype = 'advisory' AND NOT granted AND objsubid = 1 \ + AND ((classid::bigint << 32) | objid::bigint) = hashtextextended($1, 0)", + ) + .bind(&gate) + .fetch_all(&f.pool) + .await + .expect("inspect gate waiters"); + if modes == ["ShareLock"] { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }); + queued + .await + .expect("the last check queues SHARED on the gate"); + holder.commit().await.expect("commit the archive"); + assert_eq!(push.await.expect("push task"), refused(ARCHIVED)); + assert_eq!(f.published(&repo).await, (false, 0)); + + // The exception: the check passed, then the archive committed, and + // the pointer write still lands with its kind:30618. + let late = f.channel().await; + let (repo, _scratch, ctx) = f.pushed(Some(&late.to_string())).await; + let paused = Arc::new(PostCasGate::default()); + let hooks = FinalizePushHooks { + pre_cas_gate: Some(Arc::clone(&paused)), + ..FinalizePushHooks::default() + }; + let push = tokio::spawn(finalize_reply(f.state.clone(), ctx, hooks)); + let reached = tokio::time::timeout(Duration::from_secs(10), paused.reached.notified()); + reached.await.expect("the push reaches its last check"); + f.archive(late).await; + paused.resume.notify_one(); + let reply = push.await.expect("push task"); + assert_eq!(reply.0, StatusCode::OK, "{reply:?}"); + assert_eq!(f.published(&repo).await, (true, 1)); + + let LastCheckFixture { state, pool, .. } = f; + drop(state); + pool.close().await; + } + mod external_infra_minio_tests { #[tokio::test] #[ignore = "requires Postgres and MinIO"] @@ -2838,6 +3205,12 @@ mod track_c_tests { async fn finalize_push_db_failure_after_cas_is_not_success_and_releases_lease() { super::finalize_push_db_failure_after_cas_is_not_success_and_releases_lease().await; } + + #[tokio::test] + #[ignore = "requires Postgres and MinIO"] + async fn finalize_push_last_archive_check_refuses_or_lands() { + super::finalize_push_last_archive_check_refuses_or_lands().await; + } } /// A gzip-encoded request body is transparently inflated before it diff --git a/crates/buzz-relay/src/nip_fi_test_hooks.rs b/crates/buzz-relay/src/nip_fi_test_hooks.rs index 9bcbd3ec439..654d1e1764c 100644 --- a/crates/buzz-relay/src/nip_fi_test_hooks.rs +++ b/crates/buzz-relay/src/nip_fi_test_hooks.rs @@ -215,6 +215,10 @@ make_hook!(auto_join_done_hook, after_auto_join); // shared membership check passes, immediately before the fresh ban reread. // A ban or closed pool here makes only the reread refuse. make_hook!(http_ban_recheck_hook, before_http_ban_recheck); +// `before_git_settings_last_check`: fires in the Git default-branch change +// after the manifest upload, immediately before the last archive check and +// the pointer CAS. An archive or a re-announcement here meets that check. +make_hook!(git_settings_last_check_hook, before_git_settings_last_check); // `before_write_ban_check`: fires in ingest's write-restriction gates (plain // and cached) immediately before the restriction read. // `before_command_ban_check`: fires in the moderation and relay-admin command diff --git a/docs/git-on-object-storage.md b/docs/git-on-object-storage.md index f97cbff5de6..30d477256bb 100644 --- a/docs/git-on-object-storage.md +++ b/docs/git-on-object-storage.md @@ -164,10 +164,12 @@ HTTP API. apply. Direct signer authority does not depend on an optional owner's channel membership. A stored agent-owner relationship is an ownership lookup, not a live delegation credential. -- Authorization is **admission-time**: later membership/maintainer removal, - rebinding or archival does not cancel an already admitted write. This is - narrower than the vision's instantaneous-revocation aspiration. The shared - serving-write lease separately fences community deletion through the write. +- Authorization is **admission-time**: later membership/maintainer removal, or + rebinding to another active channel, does not cancel an already admitted + write. This is narrower than the vision's instantaneous-revocation + aspiration. Archive and the announcement's current binding are checked again + immediately before the CAS (§Channel archive). The shared serving-write lease + separately fences community deletion through the write. GET returns `{branch, head, manifest}`. POST accepts only `{branch, expected_manifest}` (maximum 4096 bytes) and returns the same snapshot @@ -188,6 +190,52 @@ Deploy relay support **before** using an updated CLI against that relay. This requires no schema migration and changes no live repository merely by deploying. See the [CLI examples](../crates/buzz-cli/README.md#default-branch). +### Channel archive + +This applies to repositories hosted by the relay. It does not apply to local +clones or to repositories on other hosts, such as GitHub. Archive does not +strictly freeze Git: one exception applies (below). + +- **Last check.** A push and a default-branch change each run a last check + after their uploads and immediately before the pointer CAS (§Push step 7). + The check re-reads the current kind:30617 announcement and resolves its + binding. For a bound channel, it takes that channel's write gate shared in a + short database transaction with no object-store call. An archive that is + running holds the gate exclusive, so the check waits for it and then refuses. +- **Outcomes.** A refused write does not change the pointer and stores no + kind:30618. + + | Announcement at the last check | Push | Default-branch change | + |---|---|---| + | Bound to an archived channel | 403 `channel is archived (read-only)` | 403, same reason | + | Deleted | 403 `repository not found` | 404 `repository not found` | + | Invalid binding | 403 `invalid channel binding` | 404 `repository not found` | + | Not bound | Proceeds | 404 `repository not found` | + | Bound to an active channel | Proceeds | Proceeds | + + For a bound repository, the check's transaction uses the same admission as + event writes, so it also refuses the write while the community is being + deleted. A database error refuses the write. +- **Exception: late write.** The check and the CAS are not atomic. A write + whose last check passed before the archive committed may still apply its + pointer write after the commit. It is not undone, and its kind:30618 is + stored. The S3 client library allows about 121 seconds for that write: a + 60-second request, its one built-in re-send after 1 second, and a second + 60-second request. That is a client timeout budget, not a server-side bound: + the object store may apply a received write after the client gives up. Every + write whose last check starts after the archive commits is refused. +- **Announcements stay ungated.** Archive does not gate kind:30617 + announcements or their deletions. A repository may be bound to an archived + channel; its pushes are then refused. A repository may be moved out of an + archived channel, or unbound, and then accepts pushes under its new binding. + +**Possible future improvement.** A strict storage fence would close the +late-write exception. Archive would move the pointer of each repository bound +to the archived channels inside its transaction, so a CAS with an older ETag +fails. Announcement changes would also take the gates of the old and new +channel. The cost is that archive then calls the object store, so archive +depends on object-storage availability and latency. This is not implemented. + ## Axioms The protocol's safety is proved *relative to* the following properties of the