diff --git a/crates/libs/internal/api/server/AGENTS.md b/crates/libs/internal/api/server/AGENTS.md index 47c9105ec7..5a98bcc91a 100644 --- a/crates/libs/internal/api/server/AGENTS.md +++ b/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 diff --git a/crates/libs/internal/api/server/src/published_state.rs b/crates/libs/internal/api/server/src/published_state.rs index e6e4f5dd13..10d6de944e 100644 --- a/crates/libs/internal/api/server/src/published_state.rs +++ b/crates/libs/internal/api/server/src/published_state.rs @@ -2968,44 +2968,8 @@ pub async fn publish_sandbox_states( "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::>(); + 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 @@ pub async fn publish_sandbox_states( }) } +/// 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, +) -> Result, 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 @@ fn snapshot_locations_json(snapshot: &SnapshotDetail) -> serde_json::Value { ) } +#[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; diff --git a/crates/libs/internal/api/server/src/published_state_lifecycle_race_tests.rs b/crates/libs/internal/api/server/src/published_state_lifecycle_race_tests.rs new file mode 100644 index 0000000000..435e212d32 --- /dev/null +++ b/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)> { + 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 { + 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 { + 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::>::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(()) +}