Harden the ring-authority flip and release-gate rehearsal against lost concurrency races

CHANGELOG.md

@@ -10,7 +10,7 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### Fixed
 
-- **CLI help no longer prints callback signing secrets.** `awa serve --help` and `awa callbacks serve --help` identify `AWA_CALLBACK_HMAC_SECRET` without rendering its current value. Root help already keeps the manually resolved `DATABASE_URL` out of Clap output; regression coverage now protects all three credential-bearing help surfaces.
+- **The ring-authority flip no longer surfaces a raw deadlock when it loses a race to live traffic ([#480](https://github.com/hardbyte/awa/pull/480)).** The flip runs while the current fleet keeps claiming — by design — so `awa.flip_ring_authority`'s `claim_ring_slots` writes can lose a PostgreSQL deadlock cycle (SQLSTATE `40P01`) against a concurrent `claim_ready_runtime` call; the release-gate rehearsal reproduced exactly this under 2-core contention. The server-side function is one atomic transaction, so the detected deadlock rolls it back whole and re-running re-evaluates the refusal gate from scratch. `storage::flip_ring_authority` (the wrapper behind `awa storage flip-ring-authority` and the maintenance auto-flip) now retries only that error, bounded with backoff — the same policy `migrations::run` applies at its own atomic boundary. Refusals and every other error still return immediately, and the rehearsal's stale-heartbeat flip helper applies the same retry around its freshness-window variant. The wrapper's signature narrows from a generic executor to `&PgPool` (every in-repo caller already passed a pool; the retry needs a re-executable connection source). `awa serve --help` and `awa callbacks serve --help` identify `AWA_CALLBACK_HMAC_SECRET` without rendering its current value. Root help already keeps the manually resolved `DATABASE_URL` out of Clap output; regression coverage now protects all three credential-bearing help surfaces.
 
 - **Nightly flake gates now carry runner-contention margin ([#399](https://github.com/hardbyte/awa/issues/399), [#434](https://github.com/hardbyte/awa/issues/434)).** Four assertion shapes in the chaos and benchmark suites were tight enough that shared-runner CPU contention failed them while every invariant they exist for was intact, eroding the 14-consecutive-green-nightlies release gate. `awa/tests/ci_timing.rs` now holds the scaling for all of them, and it only ever loosens a bound, and only when `CI` is set:
   - The mixed-fleet chaos test set `heartbeat_staleness` to 250ms against a 50ms heartbeat interval. Under contention a *live* worker's heartbeat missed that window, so the runtime correctly rescued a healthy attempt and the test saw a genuine duplicate completion — a margin problem that read as a correctness bug. Chaos clients scale the staleness window via `scaled_staleness` while leaving the heartbeat and rescue *intervals* at chaos cadence, so the rescue path is still exercised.

awa-model/src/storage.rs

@@ -139,23 +139,56 @@ where
         .map_err(AwaError::from)
 }
 
+/// How many times [`flip_ring_authority`] re-runs the flip after losing a
+/// deadlock race. Matches the migration runner's bound for the same class of
+/// transient loss at an atomic transaction boundary.
+const FLIP_DEADLOCK_RETRIES: u32 = 5;
+
 /// Flip a schema's ring-cursor authority `columns -> ledger` (one-way,
 /// idempotent). Refuses unless `force` when a fresh-heartbeat runtime is not
 /// known to be flip-aware. Returns the resulting authority (`"ledger"`).
-pub async fn flip_ring_authority<'e, E>(
-    executor: E,
+///
+/// The flip runs while the current fleet keeps claiming, so its
+/// `claim_ring_slots` writes can lose a PostgreSQL deadlock race
+/// (SQLSTATE `40P01`) against a concurrent `claim_ready_runtime` call. The
+/// server-side function is a single atomic transaction, so a detected
+/// deadlock rolls it back whole and re-running re-evaluates the refusal gate
+/// from scratch. Retry that one error bounded with backoff — the same policy
+/// `migrations::run` applies at its own atomic boundary — so the operator's
+/// one-time cutover (and the maintenance auto-flip) doesn't surface a raw
+/// deadlock for a designed, recoverable loss. Refusals and every other error
+/// still return immediately.
+pub async fn flip_ring_authority(
+    pool: &PgPool,
     schema: &str,
     force: bool,
-) -> Result<String, AwaError>
-where
-    E: PgExecutor<'e>,
-{
-    sqlx::query_scalar::<_, String>("SELECT awa.flip_ring_authority($1, $2)")
-        .bind(schema)
-        .bind(force)
-        .fetch_one(executor)
-        .await
-        .map_err(AwaError::from)
+) -> Result<String, AwaError> {
+    let mut retry = 0;
+    loop {
+        match sqlx::query_scalar::<_, String>("SELECT awa.flip_ring_authority($1, $2)")
+            .bind(schema)
+            .bind(force)
+            .fetch_one(pool)
+            .await
+        {
+            Ok(authority) => return Ok(authority),
+            Err(sqlx::Error::Database(database))
+                if database.code().as_deref() == Some("40P01") && retry < FLIP_DEADLOCK_RETRIES =>
+            {
+                retry += 1;
+                let delay = std::time::Duration::from_millis(50 * (1 << (retry - 1)));
+                tracing::warn!(
+                    retry,
+                    max_retries = FLIP_DEADLOCK_RETRIES,
+                    delay_ms = delay.as_millis(),
+                    schema,
+                    "ring-authority flip lost a deadlock race to live traffic; retrying"
+                );
+                tokio::time::sleep(delay).await;
+            }
+            Err(error) => return Err(AwaError::from(error)),
+        }
+    }
 }
 
 fn queue_storage_schema_from_status(status: &StorageStatus) -> Option<String> {

awa/tests/rolling_upgrade_rehearsal_test.rs

@@ -770,13 +770,40 @@ async fn wait_for_mixed_fleet(
 /// is stale, while stamped current heartbeats may remain fresh. Mirrors the
 /// operator flow the migrate-first cell proved: refusal is asserted there;
 /// these cells assert a live stamped fleet does not block the flip.
+///
+/// The flip runs under live current traffic by design, so its
+/// `claim_ring_slots` writes can lose a deadlock race against a concurrent
+/// claim (observed under 2-core contention: SQLSTATE 40P01 at
+/// `flip_ring_authority` line 108 vs `claim_ready_runtime`). PostgreSQL
+/// detects the cycle and the flip rolls back atomically, so a bounded retry
+/// re-runs the whole gate — the same policy `migrations::run` applies at its
+/// atomic boundary (#433). Only 40P01 retries; a refusal or any other error
+/// still fails the cell.
 async fn flip_after_released_heartbeats_stale(pool: &sqlx::PgPool) {
     tokio::time::sleep(Duration::from_secs(1)).await;
-    let flipped: String = sqlx::query_scalar("SELECT awa.flip_ring_authority($1, FALSE, 0.5)")
-        .bind("awa")
-        .fetch_one(pool)
-        .await
-        .expect("flip after released heartbeats become stale");
+    let mut retry = 0;
+    let flipped: String = loop {
+        match sqlx::query_scalar("SELECT awa.flip_ring_authority($1, FALSE, 0.5)")
+            .bind("awa")
+            .fetch_one(pool)
+            .await
+        {
+            Ok(authority) => break authority,
+            Err(sqlx::Error::Database(database))
+                if database.code().as_deref() == Some("40P01") && retry < 5 =>
+            {
+                retry += 1;
+                let delay = Duration::from_millis(50 * (1 << (retry - 1)));
+                eprintln!(
+                    "[rehearsal] flip lost a deadlock race to live traffic \
+                     (retry {retry}/5, backing off {}ms)",
+                    delay.as_millis()
+                );
+                tokio::time::sleep(delay).await;
+            }
+            Err(error) => panic!("flip after released heartbeats become stale: {error}"),
+        }
+    };
     assert_eq!(flipped, "ledger");
 }
 
@@ -1015,6 +1042,45 @@ async fn test_migrate_first_deadline_rescue_resumes_with_current_leader() {
         .await
         .expect("load expired compact claim")
         .expect("expired compact claim must remain visible");
+    // This assert failed once in CI (nightly 33314091717, #434) with the job
+    // already Retryable, and the panic destroyed the evidence of *which* path
+    // rescued it: `load_job` only reports Retryable for a job with an open
+    // compact claim once a `deferred_jobs` row exists for the same
+    // `run_lease`, so the writer was a claims-aware rescue/supersession, not
+    // a producer insert. Before failing, dump the rows that identify the
+    // writer — the deferred row's `errors`/`attempt` and any closure carry
+    // each rescue path's distinct fingerprint.
+    if pending.state != JobState::Running || pending.attempt != 1 {
+        let deferred: Vec<serde_json::Value> = sqlx::query_scalar(
+            "SELECT to_jsonb(d) FROM awa.deferred_jobs AS d WHERE d.job_id = $1",
+        )
+        .bind(inserted.id)
+        .fetch_all(&pool)
+        .await
+        .unwrap_or_else(|e| vec![serde_json::json!({ "deferred_query_error": e.to_string() })]);
+        let closures: Vec<(i64, String)> = sqlx::query_as(
+            "SELECT run_lease, outcome FROM awa.lease_claim_closures WHERE job_id = $1",
+        )
+        .bind(inserted.id)
+        .fetch_all(&pool)
+        .await
+        .unwrap_or_else(|e| vec![(-1, e.to_string())]);
+        let batch_closures: Vec<serde_json::Value> = sqlx::query_scalar(
+            "SELECT to_jsonb(c) FROM awa.lease_claim_batches AS batches \
+             CROSS JOIN LATERAL unnest(batches.job_ids, batches.receipt_ids) AS items(job_id, receipt_id) \
+             JOIN awa.lease_claim_closure_batches AS c \
+               ON c.claim_slot = batches.claim_slot AND items.receipt_id <@ c.receipt_ranges \
+             WHERE items.job_id = $1",
+        )
+        .bind(inserted.id)
+        .fetch_all(&pool)
+        .await
+        .unwrap_or_else(|e| vec![serde_json::json!({ "batch_closure_query_error": e.to_string() })]);
+        eprintln!(
+            "[diag] job {} state={:?} attempt={} errors={:?}\n[diag] deferred={deferred:?}\n[diag] closures={closures:?}\n[diag] batch_closures={batch_closures:?}",
+            inserted.id, pending.state, pending.attempt, pending.errors
+        );
+    }
     assert_eq!(pending.state, JobState::Running);
     assert_eq!(pending.attempt, 1);
     let heartbeat_is_fresh: bool = sqlx::query_scalar(