fix(relay): bound storage conformance admission time

Justfile

@@ -437,12 +437,6 @@ test-unit:
         # `cargo test --workspace`; without this step a manifest edit that
         # diverges Rust from the corpus ships green.
         cargo nextest run -p buzz-agent --lib
-        # buzz-acp: the ACP harness. Its ~760 --lib tests are pure in-process
-        # unit tests whose fixtures spawn a local POSIX shell as a fake agent —
-        # no relay, no database, no network. Enumerated for the same reason as
-        # the crates above: nothing in CI runs `cargo test --workspace`, so
-        # until this line existed the harness that dispatches every agent turn
-        # had zero executed test coverage in CI on any platform.
         # buzz-agent: two infra-free concerns run together by executing the
         # whole crate (lib + integration tests), because nothing in CI runs
         # `cargo test --workspace`, so without this stanza neither its
@@ -496,9 +490,10 @@ test-unit:
         # unit job either.
         cargo nextest run -p buzz-relay --lib \
             -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/)'
-        # ACP author-gate and queue tests protect the trust boundary between
-        # relay events and agent prompts. They are infra-free; ignored lifecycle
-        # tests remain excluded and run in their dedicated integration lanes.        cargo nextest run -p buzz-acp --lib
+        # Real localhost HTTP tests for the startup storage admission deadline.
+        # Keep them in the infra-free gate; the broader Git suite uses MinIO.
+        cargo nextest run -p buzz-relay --lib \
+            -E 'test(/^api::git::store::probe_deadline::tests::/)'
     else
         ./scripts/run-tests.sh unit
     fi

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

@@ -32,6 +32,8 @@ use s3::error::S3Error;
 use s3::{Bucket, Region};
 use sha2::{Digest, Sha256};
 
+mod probe_deadline;
+
 /// Opaque object-store ETag (used for `If-Match` on pointer CAS).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ETag(pub String);
@@ -101,21 +103,24 @@ pub enum StoreError {
 
 /// Configuration for `GitStore::run_conformance_probe`.
 ///
-/// Defaults: 32-way concurrency, 3 rounds. The probe is a deployment gate —
+/// Defaults: 32-way concurrency, 3 rounds, 120 seconds total. The probe is a deployment gate —
 /// run at startup, fail-closed. See `docs/git-on-object-storage.md` §Conformance.
 #[derive(Debug, Clone)]
 pub struct ProbeConfig {
     /// How many tasks race per round. Must be ≥ 2.
     pub race_width: usize,
     /// How many rounds to run each race phase.
     pub race_rounds: usize,
+    /// Deadline for the entire probe, including every race and cleanup request.
+    pub total_timeout: std::time::Duration,
 }
 
 impl Default for ProbeConfig {
     fn default() -> Self {
         Self {
             race_width: 32,
             race_rounds: 3,
+            total_timeout: std::time::Duration::from_secs(120),
         }
     }
 }
@@ -149,7 +154,7 @@ pub struct ProbeReport {
 #[derive(Debug, thiserror::Error)]
 #[error("conformance probe failed in phase '{phase}' (round {round}, key {key}): {reason}")]
 pub struct ProbeFailure {
-    /// One of `sequential`, `if_match_race`, `if_none_match_race`, `etag_consistency`.
+    /// One of `config`, `deadline`, `sequential`, `if_match_race`, `if_none_match_race`, `etag_consistency`.
     pub phase: &'static str,
     /// Round index (0-based) when this phase ran multiple rounds.
     pub round: usize,
@@ -573,7 +578,10 @@ impl GitStore {
     /// 4. **`etag_consistency`** — round-trip an ETag from `get_pointer` into
     ///    `put_pointer(IfMatch(...))` and assert `Won`. Tests that the token
     ///    is opaque and stable between read and CAS.
-    pub async fn run_conformance_probe(&self, cfg: ProbeConfig) -> Result<ProbeReport, StoreError> {
+    async fn run_conformance_probe_inner(
+        &self,
+        cfg: ProbeConfig,
+    ) -> Result<ProbeReport, StoreError> {
         use std::sync::Arc;
         if cfg.race_width < 2 || cfg.race_rounds == 0 {
             return Err(ProbeFailure {
@@ -1183,6 +1191,7 @@ mod probe {
             .run_conformance_probe(ProbeConfig {
                 race_width: 8,
                 race_rounds: 2,
+                ..ProbeConfig::default()
             })
             .await
             .expect("conformance probe");

crates/buzz-relay/src/api/git/store/probe_deadline.rs

@@ -0,0 +1,37 @@
+//! A total deadline around every request in the startup admission probe.
+
+use super::{GitStore, ProbeConfig, ProbeFailure, ProbeReport, StoreError};
+
+impl GitStore {
+    /// Admit the backend only when all conformance phases finish within the budget.
+    ///
+    /// Dropping the inner future cancels the pending request futures, including
+    /// the non-spawned racers in `join_all`. An unfinished racer is never treated
+    /// as an observed transport drop or a successful admission.
+    pub async fn run_conformance_probe(&self, cfg: ProbeConfig) -> Result<ProbeReport, StoreError> {
+        let timeout = cfg.total_timeout;
+        if timeout.is_zero() {
+            return Err(ProbeFailure {
+                phase: "config",
+                round: 0,
+                key: String::new(),
+                reason: "total_timeout must be greater than zero".into(),
+            }
+            .into());
+        }
+        tokio::time::timeout(timeout, self.run_conformance_probe_inner(cfg))
+            .await
+            .map_err(|_| ProbeFailure {
+                phase: "deadline",
+                round: 0,
+                key: String::new(),
+                reason: format!(
+                    "total probe deadline exceeded after {} ms; backend not admitted",
+                    timeout.as_millis()
+                ),
+            })?
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/buzz-relay/src/api/git/store/probe_deadline/tests.rs

@@ -0,0 +1,161 @@
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use axum::body::{Body, Bytes};
+use axum::extract::State;
+use axum::http::{header, HeaderMap, Method, Response, StatusCode, Uri};
+use axum::routing::any;
+use axum::Router;
+use tokio::sync::Mutex;
+
+use super::super::{GitStore, ProbeConfig, StoreError};
+
+#[derive(Default)]
+struct Backend {
+    objects: Mutex<HashMap<String, (Bytes, String)>>,
+    requests: AtomicUsize,
+    cas_requests: AtomicUsize,
+    stall_first_cas: bool,
+}
+
+async fn object(
+    State(backend): State<Arc<Backend>>,
+    method: Method,
+    uri: Uri,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Response<Body> {
+    backend.requests.fetch_add(1, Ordering::SeqCst);
+    if method == Method::PUT && headers.contains_key(header::IF_MATCH) {
+        let index = backend.cas_requests.fetch_add(1, Ordering::SeqCst);
+        if backend.stall_first_cas && index == 0 {
+            std::future::pending::<()>().await;
+        }
+    }
+    let mut objects = backend.objects.lock().await;
+    let key = uri.path().to_string();
+    let reply = |status, body, tag: Option<&str>| {
+        let mut response = Response::builder().status(status);
+        if let Some(tag) = tag {
+            response = response.header(header::ETAG, tag);
+        }
+        response.body(Body::from(body)).expect("fixture response")
+    };
+    match method {
+        Method::PUT => {
+            let existing = objects.get(&key);
+            if (headers.contains_key(header::IF_NONE_MATCH) && existing.is_some())
+                || headers.get(header::IF_MATCH).is_some_and(|condition| {
+                    existing.is_none_or(|(_, tag)| condition.as_bytes() != tag.as_bytes())
+                })
+            {
+                return reply(StatusCode::PRECONDITION_FAILED, Bytes::new(), None);
+            }
+            let tag = format!("\"{}\"", GitStore::digest_hex(&body));
+            objects.insert(key, (body, tag.clone()));
+            reply(StatusCode::OK, Bytes::new(), Some(&tag))
+        }
+        Method::GET => match objects.get(&key) {
+            Some((body, tag)) => reply(StatusCode::OK, body.clone(), Some(tag)),
+            None => reply(StatusCode::NOT_FOUND, Bytes::new(), None),
+        },
+        Method::DELETE => {
+            objects.remove(&key);
+            reply(StatusCode::NO_CONTENT, Bytes::new(), None)
+        }
+        _ => reply(StatusCode::METHOD_NOT_ALLOWED, Bytes::new(), None),
+    }
+}
+
+async fn start_backend(stall: bool) -> (GitStore, Arc<Backend>, tokio::task::JoinHandle<()>) {
+    let state = Arc::new(Backend {
+        stall_first_cas: stall,
+        ..Default::default()
+    });
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("local fixture listener");
+    let endpoint = format!("http://{}", listener.local_addr().expect("fixture address"));
+    let app = Router::new()
+        .route("/{*key}", any(object))
+        .with_state(Arc::clone(&state));
+    let server = tokio::spawn(async move {
+        axum::serve(listener, app).await.expect("fixture server");
+    });
+    let store = GitStore::new(
+        &endpoint,
+        "fixture-access",
+        "fixture-secret",
+        "probe-test",
+        "us-east-1",
+        buzz_media::config::S3AddressingStyle::Path,
+    )
+    .expect("fixture store");
+    (store, state, server)
+}
+
+#[tokio::test]
+async fn stalled_racer_fails_the_total_deadline_instead_of_admitting_backend() {
+    let (store, state, server) = start_backend(true).await;
+    let started = Instant::now();
+    let result = tokio::time::timeout(
+        Duration::from_secs(3),
+        store.run_conformance_probe(ProbeConfig {
+            race_width: 3,
+            race_rounds: 1,
+            total_timeout: Duration::from_millis(250),
+        }),
+    )
+    .await;
+    server.abort();
+    let _ = server.await;
+    let Err(StoreError::Probe(failure)) = result.expect("production deadline must finish first")
+    else {
+        panic!("a pending racer must never become successful admission");
+    };
+    assert!(started.elapsed() < Duration::from_secs(2));
+    assert_eq!(state.cas_requests.load(Ordering::SeqCst), 3);
+    assert_eq!(failure.phase, "deadline");
+    assert!(failure.reason.contains("backend not admitted"));
+    eprintln!("STALLED_S3_PROBE_DENIED: {failure}");
+}
+
+#[tokio::test]
+async fn responsive_backend_still_completes_all_conformance_phases() {
+    let (store, state, server) = start_backend(false).await;
+    let result = store
+        .run_conformance_probe(ProbeConfig {
+            race_width: 4,
+            race_rounds: 2,
+            // rust-s3 retries classified 412 responses once after a one-second
+            // backoff. Four race phases therefore need more than four seconds.
+            total_timeout: Duration::from_secs(10),
+        })
+        .await;
+    server.abort();
+    let _ = server.await;
+    let report = result.expect("responsive conditional-write backend");
+    assert_eq!(report.race_width, 4);
+    assert_eq!(report.race_rounds, 2);
+    assert_eq!(report.transport_drops, 0);
+    // Two four-writer races, six 412 retries, and two ETag consistency writes.
+    assert_eq!(state.cas_requests.load(Ordering::SeqCst), 16);
+    eprintln!("RESPONSIVE_S3_PROBE_ADMITTED: {report:?}");
+}
+
+#[tokio::test]
+async fn zero_deadline_is_invalid_without_sending_backend_requests() {
+    let (store, state, server) = start_backend(false).await;
+    let result = store
+        .run_conformance_probe(ProbeConfig {
+            total_timeout: Duration::ZERO,
+            ..ProbeConfig::default()
+        })
+        .await;
+    server.abort();
+    let _ = server.await;
+    assert_eq!(state.requests.load(Ordering::SeqCst), 0);
+    assert!(matches!(result, Err(StoreError::Probe(failure)) if failure.phase == "config"));
+}

crates/buzz-relay/src/main.rs

@@ -574,10 +574,12 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         let cfg = buzz_relay::api::git::store::ProbeConfig {
             race_width,
             race_rounds,
+            ..Default::default()
         };
         tracing::info!(
             race_width,
             race_rounds,
+            timeout_seconds = cfg.total_timeout.as_secs(),
             "running git object-store conformance probe (A3 gate)"
         );
         let report = state

scripts/run-tests.sh

@@ -96,7 +96,7 @@ run_unit_tests() {
     cargo test -p buzz-acp -- --nocapture
 
   # buzz-db migrator/lint unit tests (no infra): guard the embedded-migrator
-  # invariant (exactly the consolidated 0001; cutover/backfill stays an operator
+  # invariant (the complete checked-in additive migration set; cutover/backfill stays an operator
   # script, not startup state) and the tenant-scoping lints. The Postgres-backed
   # buzz-db tests are #[ignore]d; nothing here (or in integration mode below,
   # which runs `cargo test -p buzz-db` without --ignored) runs them — they need a
@@ -126,13 +126,6 @@ run_unit_tests() {
   run_test_step "buzz-agent unit tests" \
     cargo test -p buzz-agent --lib -- --nocapture
 
-  # buzz-acp harness unit tests: in-process, fixtures spawn a local POSIX shell
-  # as a fake agent (no relay, no database). Mirrors the nextest path in
-  # `just test-unit` — the two lists must stay in step.
-# ACP author-gate and queue tests are pure unit tests. Keep this fallback in
-  # step with `just test-unit`; ignored lifecycle tests run elsewhere.  run_test_step "buzz-acp unit tests" \
-    cargo test -p buzz-acp --lib -- --nocapture
-
   # Mirror the three infra-free relay handler modules in `just test-unit`'s
   # nextest expression. Keep the side-effects filter pinned to `::tests::` so
   # it does not select the sibling Postgres-backed test module.
@@ -144,6 +137,11 @@ run_unit_tests() {
 
   run_test_step "buzz-relay side-effects helper tests" \
     cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture
+
+  # Mirror the startup deadline regressions in the nextest lane. Their HTTP
+  # backend is bound to an ephemeral loopback port; no external services needed.
+  run_test_step "buzz-relay storage admission deadline tests" \
+    cargo test -p buzz-relay --lib api::git::store::probe_deadline::tests:: -- --nocapture
 }
 
 # ---- DB / integration tests (infra required) --------------------------------