internal-api: recheck the failed projection's guards under the row lock

crates/libs/internal/api/server/AGENTS.md

@@ -19,14 +19,15 @@ Every table named here is in the control-plane Postgres store.
   registration commits separately. Keep receipts last, so their fencing sees
   the final evidence.
 - A new `public.sandboxes` lifecycle write from a node report rechecks, in
-  the write or under a `FOR UPDATE` row lock taken in the same transaction,
-  that `active_node_id` is the authenticated publisher and that the current
-  lifecycle is one it may leave, as the restored, stopped and quarantined
-  projections (`src/published_state.rs`) do. Existing exceptions: the failed
-  projection checks in a CTE and updates by id; delete receipts for stopped
-  or suspended rows with no `active_node_id` are fenced by the latest
-  execution.
-- New restoring-to-started convergence leaves rows with
+  the write or under a `FOR UPDATE` or `FOR NO KEY UPDATE` row lock taken in
+  the same transaction, that `active_node_id` is the authenticated publisher
+  and that the current lifecycle is one it may leave, as the restored,
+  stopped, failed and quarantined projections (`src/published_state.rs`) do.
+  Existing exceptions: activation-completion receipts recheck only the
+  lifecycle and are fenced by the execution attempt they release
+  (generation, attempt, `active`); delete receipts for stopped or suspended
+  rows with no `active_node_id` are fenced by the latest execution.
+- New restoring-to-target convergence leaves rows with
   `evacuation_restore_target` set to the node-drain redrive.
 - New certification columns advance only on the server's own verdict.
   `node_status_snapshots.local_backup_inventory_reported_at` is also

crates/libs/internal/api/server/src/published_state.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::types::{NodeId, SnapshotId};
 
 use pplx_space_common_core::sql;
 use std::collections::{BTreeMap, BTreeSet};
@@ -2968,44 +2968,8 @@
         "stopped",
     );
 
-    let failed_sandboxes = sqlx::query_as::<_, (Uuid, String)>(sql!(
-        "WITH input AS (
-             SELECT *
-             FROM jsonb_to_recordset($2::jsonb) AS input(
-                 sandbox_id UUID,
-                 status TEXT
-             )
-         ),
-         claimed AS (
-             SELECT s.id, s.lifecycle
-             FROM sandboxes s
-             JOIN input ON input.sandbox_id = s.id
-             WHERE s.active_node_id = $1
-               AND s.managed_by = 'k8s'
-               AND LOWER(TRIM(input.status)) = 'failed'
-               AND LOWER(TRIM(s.lifecycle)) NOT IN (
-                   'failed',
-                   'quarantined',
-                   'deleted',
-                   'cleanup_required',
-                   'deleting'
-               )
-         )
-         UPDATE sandboxes s
-         SET lifecycle = 'failed'
-         FROM claimed
-         WHERE s.id = claimed.id
-         RETURNING s.id, claimed.lifecycle"
-    ))
-    .bind(node_id)
-    .bind(rows_json.clone())
-    .fetch_all(&mut *conn)
-    .await
-    .map_err(|err| Status::internal(format!("mark failed sandbox lifecycle: {err}")))?;
-    let failed_sandboxes = failed_sandboxes
-        .into_iter()
-        .map(|(sandbox_id, lifecycle)| (SandboxId::from_uuid(sandbox_id), lifecycle))
-        .collect::<Vec<_>>();
+    let failed_sandboxes =
+        project_node_published_failed_sandboxes(conn, node_id, rows_json.clone()).await?;
     emit_node_published_state_events(
         node_id,
         &failed_sandboxes,
@@ -3380,6 +3344,63 @@
     })
 }
 
+/// Moves the reporting node's `k8s` rows that it published `failed` to
+/// Postgres `public.sandboxes.lifecycle = 'failed'`, returning each moved row
+/// with the lifecycle it left.
+///
+/// The claim locks before it checks, as the suspended projection does. A row a
+/// concurrent writer holds is read again at its committed version once the lock
+/// is granted, so a `deleting` or `quarantined` committed meanwhile, or a row
+/// that left this node, is refused rather than overwritten, and the returned
+/// from-state is the lifecycle the update actually replaced. `FOR NO KEY UPDATE`
+/// is the mode the `UPDATE` itself takes on the same rows, so taking it in the
+/// claim moves the lock earlier in the statement without adding a conflict.
+async fn project_node_published_failed_sandboxes(
+    conn: &mut sqlx::PgConnection,
+    node_id: NodeId,
+    rows_json: Json<serde_json::Value>,
+) -> Result<Vec<(SandboxId, String)>, Status> {
+    let failed_sandboxes = sqlx::query_as::<_, (Uuid, String)>(sql!(
+        "WITH input AS (
+             SELECT *
+             FROM jsonb_to_recordset($2::jsonb) AS input(
+                 sandbox_id UUID,
+                 status TEXT
+             )
+         ),
+         claimed AS (
+             SELECT s.id, s.lifecycle
+             FROM sandboxes s
+             JOIN input ON input.sandbox_id = s.id
+             WHERE s.active_node_id = $1
+               AND s.managed_by = 'k8s'
+               AND LOWER(TRIM(input.status)) = 'failed'
+               AND LOWER(TRIM(s.lifecycle)) NOT IN (
+                   'failed',
+                   'quarantined',
+                   'deleted',
+                   'cleanup_required',
+                   'deleting'
+               )
+             FOR NO KEY UPDATE OF s
+         )
+         UPDATE sandboxes s
+         SET lifecycle = 'failed'
+         FROM claimed
+         WHERE s.id = claimed.id
+         RETURNING s.id, claimed.lifecycle"
+    ))
+    .bind(node_id)
+    .bind(rows_json)
+    .fetch_all(&mut *conn)
+    .await
+    .map_err(|err| Status::internal(format!("mark failed sandbox lifecycle: {err}")))?;
+    Ok(failed_sandboxes
+        .into_iter()
+        .map(|(sandbox_id, lifecycle)| (SandboxId::from_uuid(sandbox_id), lifecycle))
+        .collect())
+}
+
 struct SuspendedProjectionOutcome {
     rows: Vec<(SandboxId, bool)>,
     recoveries: Vec<(SandboxId, SandboxLifecycle, SandboxQuarantineReasonCode)>,
@@ -3955,6 +3976,9 @@
     )
 }
 
+#[cfg(test)]
+#[path = "published_state_lifecycle_race_tests.rs"]
+mod published_state_lifecycle_race_tests;
 #[cfg(test)]
 #[path = "published_state_tests.rs"]
 mod published_state_tests;

crates/libs/internal/api/server/src/published_state_lifecycle_race_tests.rs

@@ -0,0 +1,199 @@
+//! The `failed` node-report projection racing another writer of the same
+//! control-plane Postgres `public.sandboxes` row. Each case holds the other
+//! writer's row lock, proves the projection waits on it (`pg_blocking_pids`),
+//! commits the write, and asserts the row the projection leaves and the
+//! transitions it reports. The node-local sbx-mgr SQLite store is not involved.
+
+use std::time::Duration;
+
+use pplx_space_common_core::sql;
+use pplx_space_common_core::types::{NodeId, SandboxId};
+use pplx_space_db_test_fixtures::{insert_ready_worker_node, pg_test_db_with_max_connections};
+use pplx_space_test_env::TestResult;
+use sqlx::types::Json;
+use sqlx::{PgPool, Postgres, Transaction};
+
+use super::project_node_published_failed_sandboxes;
+
+/// Bounds both the wait for the projection to block and its completion.
+const RACE_BUDGET: Duration = Duration::from_secs(10);
+
+/// The persisted Postgres `public.sandboxes.lifecycle` and `active_node_id`.
+async fn row(pool: &PgPool, sandbox_id: SandboxId) -> TestResult<(String, Option<NodeId>)> {
+    Ok(sqlx::query_as(sql!(
+        "SELECT lifecycle, active_node_id FROM sandboxes WHERE id = $1"
+    ))
+    .bind(sandbox_id)
+    .fetch_one(pool)
+    .await?)
+}
+
+/// A `k8s` row placed on `node_id` in `lifecycle`.
+async fn insert_k8s_sandbox(
+    pool: &PgPool,
+    node_id: NodeId,
+    lifecycle: &str,
+) -> TestResult<SandboxId> {
+    let sandbox_id = SandboxId::new();
+    sqlx::query(sql!(
+        "INSERT INTO sandboxes (
+             id, user_id, org_id, lifecycle, placement, active_node_id,
+             target_node_id, managed_by, network_policy_override_json,
+             network_policy_generation
+         )
+         VALUES (
+             $1,
+             '00000000-0000-0000-0000-000000000010',
+             '00000000-0000-0000-0000-000000000001',
+             $2, 'on_node', $3, $3, 'k8s', $4, 1
+         )"
+    ))
+    .bind(sandbox_id)
+    .bind(lifecycle)
+    .bind(node_id)
+    .bind(Json(pplx_space_net_common::NetworkPolicyV1::default()))
+    .execute(pool)
+    .await?;
+    Ok(sandbox_id)
+}
+
+/// Opens the conflicting writer's transaction and returns it with its backend
+/// pid; the caller's statements on it hold their row locks until it ends.
+async fn begin_holder(pool: &PgPool) -> TestResult<(Transaction<'static, Postgres>, i32)> {
+    let mut holder = pool.begin().await?;
+    let pid = sqlx::query_scalar(sql!("SELECT pg_backend_pid()"))
+        .fetch_one(&mut *holder)
+        .await?;
+    Ok((holder, pid))
+}
+
+/// Whether backend `waiter` is waiting on a lock `holder` holds.
+async fn is_blocked_by(pool: &PgPool, waiter: i32, holder: i32) -> TestResult<bool> {
+    Ok(
+        sqlx::query_scalar(sql!("SELECT $1 = ANY(pg_blocking_pids($2))"))
+            .bind(holder)
+            .bind(waiter)
+            .fetch_one(pool)
+            .await?,
+    )
+}
+
+async fn wait_until_blocked(pool: &PgPool, waiter: i32, holder: i32) -> TestResult {
+    tokio::time::timeout(RACE_BUDGET, async {
+        while !is_blocked_by(pool, waiter, holder).await? {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+        TestResult::Ok(())
+    })
+    .await
+    .map_err(|_| format!("backend {waiter} never waited on backend {holder}"))?
+}
+
+/// The concurrent writer that commits while the `failed` projection waits.
+#[derive(Clone, Copy, Debug)]
+enum FailedRace {
+    /// A user DELETE: lock the row, then persist `deleting` intent.
+    Delete,
+    /// A quarantine hold committed for the row.
+    Quarantine,
+    /// A suspend that finalized and unplaced the row.
+    SuspendUnplaced,
+    /// A pause: the row still qualifies for `failed`.
+    Pause,
+}
+
+impl FailedRace {
+    async fn write(
+        self,
+        holder: &mut Transaction<'_, Postgres>,
+        sandbox_id: SandboxId,
+    ) -> TestResult {
+        if matches!(self, Self::Delete) {
+            sqlx::query(sql!(
+                "SELECT lifecycle FROM sandboxes WHERE id = $1 FOR UPDATE"
+            ))
+            .bind(sandbox_id)
+            .execute(&mut **holder)
+            .await?;
+        }
+        match self {
+            Self::Delete => sqlx::query(sql!(
+                "UPDATE sandboxes SET lifecycle = 'deleting' WHERE id = $1"
+            )),
+            Self::Quarantine => sqlx::query(sql!(
+                "UPDATE sandboxes SET lifecycle = 'quarantined' WHERE id = $1"
+            )),
+            Self::SuspendUnplaced => sqlx::query(sql!(
+                "UPDATE sandboxes
+                    SET lifecycle = 'suspended', placement = 'unplaced', active_node_id = NULL
+                  WHERE id = $1"
+            )),
+            Self::Pause => sqlx::query(sql!(
+                "UPDATE sandboxes SET lifecycle = 'paused' WHERE id = $1"
+            )),
+        }
+        .bind(sandbox_id)
+        .execute(&mut **holder)
+        .await?;
+        Ok(())
+    }
+}
+
+/// A `failed` report for a `started` row reaches the row while another writer
+/// holds it. The projection decides on what that writer commits: a `deleting`,
+/// a `quarantined` or a row unplaced as `suspended` stands, and a `paused` row
+/// still becomes `failed`, reported as leaving `paused`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_projection_decides_on_the_row_committed_while_it_waits() -> TestResult {
+    let db = pg_test_db_with_max_connections(4).await?;
+    let pool = &db.pool;
+    let node_id = NodeId::new();
+    insert_ready_worker_node(pool, node_id).await?;
+
+    let cases = [
+        (FailedRace::Delete, "deleting", Some(node_id), None),
+        (FailedRace::Quarantine, "quarantined", Some(node_id), None),
+        (FailedRace::SuspendUnplaced, "suspended", None, None),
+        (FailedRace::Pause, "failed", Some(node_id), Some("paused")),
+    ];
+    let mut mismatches = Vec::new();
+    for (race, expected_lifecycle, expected_node, expected_from) in cases {
+        let sandbox_id = insert_k8s_sandbox(pool, node_id, "started").await?;
+        let (mut holder, holder_pid) = begin_holder(pool).await?;
+        race.write(&mut holder, sandbox_id).await?;
+
+        let mut projection_tx = pool.begin().await?;
+        let waiter_pid: i32 = sqlx::query_scalar(sql!("SELECT pg_backend_pid()"))
+            .fetch_one(&mut *projection_tx)
+            .await?;
+        let rows_json = Json(serde_json::json!([
+            { "sandbox_id": sandbox_id.to_string(), "status": "failed" }
+        ]));
+        let projection = tokio::spawn(async move {
+            let moved =
+                project_node_published_failed_sandboxes(&mut projection_tx, node_id, rows_json)
+                    .await?;
+            projection_tx.commit().await?;
+            TestResult::<Vec<_>>::Ok(moved)
+        });
+        wait_until_blocked(pool, waiter_pid, holder_pid).await?;
+        holder.commit().await?;
+        let moved = tokio::time::timeout(RACE_BUDGET, projection)
+            .await
+            .map_err(|_| format!("{race:?}: projection did not finish"))???;
+
+        let expected_moved = expected_from
+            .map(|from| vec![(sandbox_id, from.to_owned())])
+            .unwrap_or_default();
+        let (lifecycle, active_node_id) = row(pool, sandbox_id).await?;
+        let actual = (lifecycle.as_str(), active_node_id, moved);
+        let expected = (expected_lifecycle, expected_node, expected_moved);
+        if actual != expected {
+            mismatches.push(format!(
+                "{race:?}: (lifecycle, active node, reported moves) {actual:?}, expected {expected:?}"
+            ));
+        }
+    }
+    assert!(mismatches.is_empty(), "{mismatches:#?}");
+    Ok(())
+}