From d5fdbf2a529a64fac82d3a5068c2c3d157624772 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Wed, 4 Nov 2026 01:21:14 +0000 Subject: [PATCH 1/2] sandbox-store: write the portability assignment renderers' statements whole and delete the renderers pplx-space-sandbox-portability still rendered SET-list fragments for the sandboxes row, its executions and its port exposures (clear/write/publish_clear_or_preserve_current_local_artifact_marker_assignments, suspended_k8s_finalization_sql and quarantined_k8s_finalization_sql with SuspendedK8sFinalizationSql and SuspendedTargetNodeAssignment). The data-store rules keep a statement whole at its query site and allow only a whole portability boolean predicate to be interpolated. Each statement that interpolated a renderer now spells it out at its query site in sandbox-store (drain, ingest, orphan, pod_tracking, repair, suspend) and in the cluster controller's pod-less drain give-up and started and failed-restorable repair stores. Statements whose only interpolations are portability predicates are built once in a LazyLock; the two proof shapes of pod_tracking's started recovery are two whole statements. The orphan guard and repair row-state guard fragments, which the reconciler scans also use, stay as they are. drain::revert_pod_less_resident_to_suspended returns whether the row flipped instead of the renderer. The sandbox_writes ratchet drops its two renderer-call patterns, which match nothing once the renderers are gone, and its comments about renderers; the statement-head pattern and every baseline cap are unchanged. Every statement text is unchanged: a Postgres statement log of the touched crates' tests on the base and on this change records the same statements, apart from sql! locator line numbers. --- ARCHITECTURE.toml | 22 +- architecture-baseline.toml | 5 +- .../sandbox/portability/src/lib.rs | 244 +----------------- crates/l30_storage/sandbox/store/src/drain.rs | 62 ++--- .../l30_storage/sandbox/store/src/ingest.rs | 90 ++++--- .../l30_storage/sandbox/store/src/orphan.rs | 102 +++++--- .../sandbox/store/src/pod_tracking.rs | 120 +++++---- .../l30_storage/sandbox/store/src/repair.rs | 80 +++--- .../l30_storage/sandbox/store/src/suspend.rs | 91 ++++--- .../src/failed_restorable_repair/store.rs | 30 +-- .../src/node_drain/store/pod_less_giveup.rs | 58 +++-- .../src/started_sandbox_repair/store.rs | 34 ++- .../server/src/node_local_backup_pending.rs | 6 +- 13 files changed, 396 insertions(+), 548 deletions(-) diff --git a/ARCHITECTURE.toml b/ARCHITECTURE.toml index 2a7e88c2a4..8c16d804ff 100644 --- a/ARCHITECTURE.toml +++ b/ARCHITECTURE.toml @@ -169,23 +169,14 @@ mode = "required" # production line that opens a statement writing the row (`UPDATE`, # `INSERT INTO`, `DELETE FROM`, `MERGE INTO` or `TRUNCATE` naming `sandboxes`, # in any case, schema-qualified or quoted, before any alias, including right -# after a `\n`, `\r` or `\t` escape in a one-line string), and each call to -# the sandbox-portability renderers that compose part of its SET list -# (`sandbox_assignments()`, as a method or a path call, and the -# `*_current_local_artifact_marker_assignments` functions, after punctuation, -# a line start or `return`, `match`, `if`, `in`, `else` or `while`), so a file -# with a composed `UPDATE sandboxes s SET {...}` counts the head and each -# renderer call. +# after a `\n`, `\r` or `\t` escape in a one-line string). # Not caught: a statement head split across lines (`UPDATE` and `sandboxes` on # different lines) or built at run time (`format!("UPDATE {table} ...")`); a # `COPY sandboxes FROM ...`; a statement in a non-`.rs` file, such as the # placement state machine's `include_str!` claim statements under # `postgres/sql/`; a head right after an escape other than `\n`, `\r` or -# `\t` (`\0`, `\x0a`); a renderer called through an alias or a `use ... as` -# rename, passed as a function value -# (`.map(SuspendedK8sFinalizationSql::sandbox_assignments)`), or called -# right after `break` or `yield`; database triggers; scripts and smoke -# harnesses outside workspace members' `src/`. +# `\t` (`\0`, `\x0a`); database triggers; scripts and smoke harnesses outside +# workspace members' `src/`. # False positives: the same text in a trailing `//` or `/* */` comment, in # `#[doc = "..."]`, in a SQL `--` comment or in an ordinary string such as an # error message ("failed to update sandboxes"); a statement on another @@ -204,16 +195,11 @@ mode = "required" # in a production file counts, so test-only writes go in a `*_tests` file. # A new writer outside the store fails, as does another write in a listed # file; so does splitting a listed file into modules, since `--renames` pairs -# only whole-file renames and the new files' entries are new. Portability -# calls its private `sandbox_assignments` helper bare; a path-qualified call to -# it there (`Self::...`, `super::...`) matches and fails. +# only whole-file renames and the new files' entries are new. [rules.pattern_ratchet.sandbox_writes] patterns = [ # A statement head on the row's table '''(?i)(?:\\[nrt]|\b)(?:UPDATE|DELETE\s+FROM|INSERT\s+INTO|MERGE\s+INTO|TRUNCATE(?:\s+TABLE)?)\s+(?:ONLY\s+)?(?:\\?"?public\\?"?\s*\.\s*)?\\?"?sandboxes\b''', - # Calls to the portability renderers of the row's assignments - '''(?:\.|::)\s*sandbox_assignments\s*\(''', - '''(?:^|[^\w\s]|\b(?:return|match|if|in|else|while)\s)\s*(?:\w+\s*::\s*)*(?:clear|write|publish_clear_or_preserve)_current_local_artifact_marker_assignments\s*\(''', ] allow = [ "crates/l30_storage/sandbox/store/**", diff --git a/architecture-baseline.toml b/architecture-baseline.toml index d6d3e54d7e..07c11c5423 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -43,9 +43,8 @@ "crates/l40_api/node/mgr/server/src/net/fallback.rs" = 1 "crates/l40_api/node/mgr/server/src/service/artifact_fs/receive.rs" = 4 -# Writers of the control-plane Postgres `sandboxes` row outside sandbox-store, -# and their calls to sandbox-portability's assignment renderers. They move into -# the store; lower each cap as they go. +# Writers of the control-plane Postgres `sandboxes` row outside sandbox-store. +# They move into the store; lower each cap as they go. [pattern_ratchet.sandbox_writes] "crates/l30_storage/placement/src/postgres.rs" = 1 "crates/l40_api/api/src/controllers/sandbox/responses.rs" = 1 diff --git a/crates/l30_storage/sandbox/portability/src/lib.rs b/crates/l30_storage/sandbox/portability/src/lib.rs index 843398c76f..dd689e0bea 100644 --- a/crates/l30_storage/sandbox/portability/src/lib.rs +++ b/crates/l30_storage/sandbox/portability/src/lib.rs @@ -439,8 +439,8 @@ pub fn sandbox_attached_anywhere(sandbox_alias: &str) -> String { /// it lasts, and a lagged sandbox can then be restored elsewhere. /// /// So the conjunct is not justified by "lag is same-node". It is justified -/// by where the marker can go next. -/// [`publish_clear_or_preserve_current_local_artifact_marker_assignments`] +/// by where the marker can go next. The node-published suspended +/// finalization (`pplx_space_sandbox_store::ingest::finalize_suspended`) /// moves the marker only for a sandbox some node is publishing a non-NULL /// artifact for. Reaching this delete means the node's *pending-backup /// report* no longer names the sandbox, which is weaker than the node @@ -1467,238 +1467,6 @@ pub fn failed_repair_revert_eligibility(sandbox_alias: &str) -> SuspendRevertEli suspend_revert_eligibility(sandbox_alias, marker_current, newest_current) } -/// Returns the shared SET-list fragment for clearing or preserving the current -/// local artifact marker while reverting a started sandbox back to suspended. -/// -/// `clear_marker_expr` must evaluate to TRUE only when the current marker is -/// stale and safe to clear; when it is FALSE, all marker columns are preserved. -pub fn clear_current_local_artifact_marker_assignments( - sandbox_alias: &str, - clear_marker_expr: &str, -) -> String { - format!( - "current_local_artifact_id = CASE WHEN {clear_marker_expr} THEN NULL ELSE {sandbox_alias}.current_local_artifact_id END,\n\ - current_local_artifact_kind = CASE WHEN {clear_marker_expr} THEN NULL ELSE {sandbox_alias}.current_local_artifact_kind END,\n\ - current_local_artifact_origin = CASE WHEN {clear_marker_expr} THEN NULL ELSE {sandbox_alias}.current_local_artifact_origin END" - ) -} - -pub fn write_current_local_artifact_marker_assignments( - artifact_id_expr: &str, - artifact_kind_expr: &str, - artifact_origin_expr: &str, -) -> String { - format!( - "current_local_artifact_id = {artifact_id_expr},\n\ - current_local_artifact_kind = {artifact_kind_expr},\n\ - current_local_artifact_origin = {artifact_origin_expr}" - ) -} - -/// Returns the shared SET-list fragment for choosing between a newly -/// published artifact marker, clearing a stale marker, and preserving the -/// current marker. -pub fn publish_clear_or_preserve_current_local_artifact_marker_assignments( - sandbox_alias: &str, - published_artifact_id_expr: &str, - published_artifact_kind_expr: &str, - published_artifact_origin_expr: &str, - clear_marker_expr: &str, -) -> String { - format!( - "current_local_artifact_id = CASE - WHEN {published_artifact_id_expr} IS NOT NULL THEN {published_artifact_id_expr} - WHEN {clear_marker_expr} THEN NULL - ELSE {sandbox_alias}.current_local_artifact_id - END, - current_local_artifact_kind = CASE - WHEN {published_artifact_id_expr} IS NOT NULL THEN {published_artifact_kind_expr} - WHEN {clear_marker_expr} THEN NULL - ELSE {sandbox_alias}.current_local_artifact_kind - END, - current_local_artifact_origin = CASE - WHEN {published_artifact_id_expr} IS NOT NULL THEN {published_artifact_origin_expr} - WHEN {clear_marker_expr} THEN NULL - ELSE {sandbox_alias}.current_local_artifact_origin - END" - ) -} - -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum SuspendedTargetNodeAssignment { - Preserve, - PreserveOrBackfillActive, - Clear, -} - -pub struct SuspendedK8sFinalizationSql { - sandbox_alias: String, - candidate_alias: String, - marker_assignments: String, - target_node_assignment: SuspendedTargetNodeAssignment, - lifecycle: SandboxLifecycle, -} - -impl SuspendedK8sFinalizationSql { - pub fn active_execution_assignments(&self, execution_alias: &str) -> String { - active_execution_assignments(execution_alias, self.lifecycle) - } - - pub fn open_port_exposure_assignments(&self) -> &'static str { - OPEN_PORT_EXPOSURE_ASSIGNMENTS - } - - pub fn sandbox_assignments(&self) -> String { - sandbox_assignments( - &self.sandbox_alias, - &self.marker_assignments, - self.target_node_assignment, - self.lifecycle, - ) - } - - pub fn retire_active_execution_cte(&self) -> String { - let active_execution_assignments = self.active_execution_assignments("e"); - let candidate_alias = &self.candidate_alias; - format!( - "retired_execution AS (\n\ - UPDATE sandbox_k8s_executions e\n\ - SET {active_execution_assignments}\n\ - FROM {candidate_alias}\n\ - WHERE e.sandbox_id = {candidate_alias}.id\n\ - AND e.active\n\ - RETURNING e.sandbox_id\n\ - )" - ) - } - - pub fn invalidate_open_port_exposures_cte(&self) -> String { - let open_port_exposure_assignments = self.open_port_exposure_assignments(); - let candidate_alias = &self.candidate_alias; - format!( - "reset_port_exposures AS (\n\ - UPDATE sandbox_port_exposures spe\n\ - SET {open_port_exposure_assignments}\n\ - FROM {candidate_alias}\n\ - WHERE spe.sandbox_id = {candidate_alias}.id\n\ - AND spe.desired_state = 'open'\n\ - RETURNING spe.sandbox_id\n\ - )" - ) - } -} - -/// Returns shared SQL fragments for projecting a Kubernetes sandbox as -/// suspended after a durable suspend handoff. -/// -/// `marker_assignments` supplies the caller-specific current artifact marker -/// policy: explicit suspend writes the just-published marker, while recovery -/// paths may preserve or clear an existing marker. -pub fn suspended_k8s_finalization_sql( - sandbox_alias: &str, - candidate_alias: &str, - marker_assignments: &str, - target_node_assignment: SuspendedTargetNodeAssignment, -) -> SuspendedK8sFinalizationSql { - k8s_finalization_sql( - sandbox_alias, - candidate_alias, - marker_assignments, - target_node_assignment, - SandboxLifecycle::Suspended, - ) -} - -/// Like [`suspended_k8s_finalization_sql`] but lands the sandbox in -/// `quarantined`, for finalizations that unplace a sandbox after authoritative -/// pod absence with no recoverable artifact. -pub fn quarantined_k8s_finalization_sql( - sandbox_alias: &str, - candidate_alias: &str, - marker_assignments: &str, - target_node_assignment: SuspendedTargetNodeAssignment, -) -> SuspendedK8sFinalizationSql { - k8s_finalization_sql( - sandbox_alias, - candidate_alias, - marker_assignments, - target_node_assignment, - SandboxLifecycle::Quarantined, - ) -} - -/// Private renderer shared by the two public constructors. The lifecycle is -/// deliberately not part of any public signature: only `suspended` and -/// `quarantined` are valid unplaced-terminal states, so exposing the full -/// `SandboxLifecycle` would let a caller render a contradictory row (e.g. -/// `started` with `placement = 'unplaced'`). The constructors fix the two -/// legal modes; the lifecycle string still comes from the shared -/// `SandboxLifecycle::as_str()` (no parallel vocabulary). -fn k8s_finalization_sql( - sandbox_alias: &str, - candidate_alias: &str, - marker_assignments: &str, - target_node_assignment: SuspendedTargetNodeAssignment, - lifecycle: SandboxLifecycle, -) -> SuspendedK8sFinalizationSql { - SuspendedK8sFinalizationSql { - sandbox_alias: sandbox_alias.to_string(), - candidate_alias: candidate_alias.to_string(), - marker_assignments: marker_assignments.to_string(), - target_node_assignment, - lifecycle, - } -} - -fn sandbox_assignments( - sandbox_alias: &str, - marker_assignments: &str, - target_node_assignment: SuspendedTargetNodeAssignment, - lifecycle: SandboxLifecycle, -) -> String { - let target_node_assignment = match target_node_assignment { - SuspendedTargetNodeAssignment::Preserve => None, - SuspendedTargetNodeAssignment::PreserveOrBackfillActive => Some(format!( - "target_node_id = COALESCE({sandbox_alias}.target_node_id, {sandbox_alias}.active_node_id)" - )), - SuspendedTargetNodeAssignment::Clear => Some( - "target_node_id = NULL,\n placement_changed_at = NULL".to_string(), - ), - }; - let mut sandbox_assignments = vec![ - format!("lifecycle = '{}'", lifecycle.as_str()), - "placement = 'unplaced'".to_string(), - "active_node_id = NULL".to_string(), - "command_ready_pending = FALSE".to_string(), - ]; - if let Some(target_node_assignment) = target_node_assignment { - sandbox_assignments.push(target_node_assignment); - } - if !marker_assignments.is_empty() { - sandbox_assignments.push(marker_assignments.to_string()); - } - sandbox_assignments.join(",\n ") -} - -fn active_execution_assignments(execution_alias: &str, lifecycle: SandboxLifecycle) -> String { - format!( - "active = FALSE,\n\ - state = '{}',\n\ - pod_ip = NULL,\n\ - terminated_at = COALESCE({execution_alias}.terminated_at, NOW()),\n\ - updated_at = NOW()", - lifecycle.as_str() - ) -} - -const OPEN_PORT_EXPOSURE_ASSIGNMENTS: &str = "apply_state = 'pending',\n\ - last_applied_node_id = NULL,\n\ - last_error = NULL,\n\ - last_bridge_status = NULL,\n\ - last_applied_bridge_generation = NULL,\n\ - last_observed_at = NULL,\n\ - updated_at = NOW()"; - /// Returns a SQL fragment evaluating to TRUE iff the snapshot the markerless /// restore selector would actually load is a ready, durable full `suspend` /// snapshot that provably loses no writes. @@ -2046,10 +1814,10 @@ fn inflight_snapshot_matches_current_suspend(sandbox_alias: &str) -> String { /// the artifact the node published, but only through /// [`node_published_suspend_evidence`], which requires the published origin to /// be `'suspend'` and its kind `'full'`, so it cannot carry a `'stop'` marker -/// either. It also finalizes through [`suspended_k8s_finalization_sql`], which -/// nulls `active_node_id` in the same statement, while this arm only reaches -/// rows whose `active_node_id` names the publishing node. So no current writer -/// produces a row this term spares. The schema has admitted a `'stop'` origin +/// either. Its statement (`pplx_space_sandbox_store::ingest::finalize_suspended`) +/// also nulls `active_node_id`, while this arm only reaches rows whose +/// `active_node_id` names the publishing node. So no current writer produces +/// a row this term spares. The schema has admitted a `'stop'` origin /// since the marker columns were added, so a row an earlier writer left /// pinned that way would be spared only if its pinned artifact is durable, /// which is the outcome this term exists for. diff --git a/crates/l30_storage/sandbox/store/src/drain.rs b/crates/l30_storage/sandbox/store/src/drain.rs index 372edcf932..40db045539 100644 --- a/crates/l30_storage/sandbox/store/src/drain.rs +++ b/crates/l30_storage/sandbox/store/src/drain.rs @@ -35,14 +35,14 @@ use pplx_space_common_core::types::{ PositiveDuration, SandboxId, SandboxLifecycle, }; +use std::sync::LazyLock; + use pplx_space_common_core::sql; use pplx_space_sandbox_portability::{ - SuspendedK8sFinalizationSql, SuspendedTargetNodeAssignment, - clear_current_local_artifact_marker_assignments, current_disk_artifact_backup_ready, - current_suspend_snapshot_ready, failed_repair_revert_eligibility, - inflight_suspend_upload_exists, latest_full_snapshot_ready, newest_full_snapshot_unambiguous, - pod_less_drain_revert_loses_no_newer_bytes, recoverable_suspend_snapshot_exists, - snapshot_durable_for_source_release, suspended_k8s_finalization_sql, + current_disk_artifact_backup_ready, current_suspend_snapshot_ready, + failed_repair_revert_eligibility, inflight_suspend_upload_exists, latest_full_snapshot_ready, + newest_full_snapshot_unambiguous, pod_less_drain_revert_loses_no_newer_bytes, + recoverable_suspend_snapshot_exists, snapshot_durable_for_source_release, }; use sqlx::{Postgres, Transaction}; @@ -1330,9 +1330,9 @@ pub async fn fail_resident_sandboxes_on_gone_node( /// [`failed_repair_revert_eligibility`] and both /// [`pod_less_drain_revert_loses_no_newer_bytes`] and /// [`newest_full_snapshot_unambiguous`] hold, all evaluated by this -/// statement: the same predicates the drain's precheck reads. Returns the -/// finalization when the row flipped, so the caller retires the executions -/// and resets the port exposures with the same assignments. +/// statement: the same predicates the drain's precheck reads. Returns whether +/// the row flipped; when it did, the caller retires the executions and resets +/// the port exposures in the same transaction. /// /// Writes an **inference**: a drain give-up, the control plane concluding /// from the apiserver, the node's teardown record and its own rows that @@ -1343,40 +1343,42 @@ pub async fn fail_resident_sandboxes_on_gone_node( pub async fn revert_pod_less_resident_to_suspended( tx: &mut Transaction<'_, Postgres>, sandbox_id: SandboxId, -) -> Result, sqlx::Error> { - let eligibility = failed_repair_revert_eligibility("s"); - let marker_assignments = - clear_current_local_artifact_marker_assignments("s", "pre.clear_marker"); - let finalization = suspended_k8s_finalization_sql( - "s", - "pre", - &marker_assignments, - SuspendedTargetNodeAssignment::Clear, - ); - let query = format!( - "WITH pre AS ( +) -> Result { + /// Built once: every input is a literal or a bind placeholder. + static QUERY: LazyLock = LazyLock::new(|| { + let eligibility = failed_repair_revert_eligibility("s"); + sql!(format!( + "WITH pre AS ( SELECT id, {clear_marker} AS clear_marker FROM sandboxes s WHERE s.id = $1 ) UPDATE sandboxes s - SET {assignments} + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = NULL, + placement_changed_at = NULL, + current_local_artifact_id = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_origin END FROM pre WHERE s.id = pre.id AND {eligible} AND {loses_no_newer_bytes} AND {unambiguous}", - clear_marker = eligibility.clear_marker, - assignments = finalization.sandbox_assignments(), - eligible = eligibility.eligible, - loses_no_newer_bytes = pod_less_drain_revert_loses_no_newer_bytes("s"), - unambiguous = newest_full_snapshot_unambiguous("s"), - ); - let reverted = sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) + clear_marker = eligibility.clear_marker, + eligible = eligibility.eligible, + loses_no_newer_bytes = pod_less_drain_revert_loses_no_newer_bytes("s"), + unambiguous = newest_full_snapshot_unambiguous("s"), + )) + }); + let reverted = sqlx::query(sqlx::AssertSqlSafe(QUERY.as_str())) .bind(sandbox_id) .execute(&mut **tx) .await? .rows_affected(); - Ok((reverted > 0).then_some(finalization)) + Ok(reverted > 0) } diff --git a/crates/l30_storage/sandbox/store/src/ingest.rs b/crates/l30_storage/sandbox/store/src/ingest.rs index 9acebef5fd..f1057c381a 100644 --- a/crates/l30_storage/sandbox/store/src/ingest.rs +++ b/crates/l30_storage/sandbox/store/src/ingest.rs @@ -34,11 +34,7 @@ use pplx_space_common_core::types::{ use std::sync::LazyLock; use pplx_space_common_core::sql; -use pplx_space_sandbox_portability::{ - SuspendedTargetNodeAssignment, node_published_execution_identity_matches, - publish_clear_or_preserve_current_local_artifact_marker_assignments, - suspended_k8s_finalization_sql, -}; +use pplx_space_sandbox_portability::node_published_execution_identity_matches; use sqlx::types::{Json, JsonValue}; use uuid::Uuid; @@ -183,8 +179,8 @@ pub struct SuspendedFinalizationCandidate { /// sandbox suspended, or reported its suspend backup ready. /// /// One statement retires the active execution, resets open port exposures -/// to pending and writes the suspended assignments -/// ([`suspended_k8s_finalization_sql`]) with the artifact marker published, +/// to pending and leaves the row `suspended` and unplaced, backfilling its +/// target node from its active node, with the artifact marker published, /// cleared or kept. A candidate with a published artifact finalizes only /// when its active execution is the exact one the node published from /// ([`node_published_execution_identity_matches`]); one without finalizes @@ -197,20 +193,6 @@ pub async fn finalize_suspended( executor: impl sqlx::PgExecutor<'_>, candidates: &[SuspendedFinalizationCandidate], ) -> Result, sqlx::Error> { - let artifact_marker_assignments = - publish_clear_or_preserve_current_local_artifact_marker_assignments( - "s", - "candidate.published_artifact_id", - "candidate.published_artifact_kind", - "candidate.published_artifact_origin", - "candidate.clear_marker", - ); - let finalization = suspended_k8s_finalization_sql( - "s", - "candidate", - &artifact_marker_assignments, - SuspendedTargetNodeAssignment::PreserveOrBackfillActive, - ); let candidate_ids = candidates .iter() .map(|candidate| candidate.sandbox_id) @@ -273,7 +255,16 @@ pub async fn finalize_suspended( .map(|identity| identity.pod_uid_was_missing) }) .collect::>(); - let candidate_cte = "candidate AS ( + /// Built once: every input is a literal or a bind placeholder. + static FINALIZATION_QUERY: LazyLock = LazyLock::new(|| { + let exact_execution_identity = node_published_execution_identity_matches( + "e", + "candidate.published_execution_generation", + "candidate.published_pod_uid", + "candidate.published_pod_uid_was_missing", + ); + sql!(format!( + "WITH candidate AS ( SELECT * FROM UNNEST( $1::UUID[], $2::BOOL[], $3::UUID[], $4::TEXT[], $5::TEXT[], @@ -284,18 +275,14 @@ pub async fn finalize_suspended( published_execution_generation, published_pod_uid, published_pod_uid_was_missing ) - )"; - let exact_execution_identity = node_published_execution_identity_matches( - "e", - "candidate.published_execution_generation", - "candidate.published_pod_uid", - "candidate.published_pod_uid_was_missing", - ); - let finalization_query = format!( - "WITH {candidate_cte}, + ), retired_execution AS ( UPDATE sandbox_k8s_executions e - SET {active_execution_assignments} + SET active = FALSE,\n\ + state = 'suspended',\n\ + pod_ip = NULL,\n\ + terminated_at = COALESCE(e.terminated_at, NOW()),\n\ + updated_at = NOW() FROM candidate WHERE e.sandbox_id = candidate.id AND e.active @@ -317,22 +304,45 @@ pub async fn finalize_suspended( ), reset_port_exposures AS ( UPDATE sandbox_port_exposures spe - SET {open_port_exposure_assignments} + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW() FROM eligible_candidate candidate WHERE spe.sandbox_id = candidate.id AND spe.desired_state = 'open' RETURNING spe.sandbox_id ) UPDATE sandboxes s - SET {sandbox_assignments} + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = COALESCE(s.target_node_id, s.active_node_id), + current_local_artifact_id = CASE + WHEN candidate.published_artifact_id IS NOT NULL THEN candidate.published_artifact_id + WHEN candidate.clear_marker THEN NULL + ELSE s.current_local_artifact_id + END, + current_local_artifact_kind = CASE + WHEN candidate.published_artifact_id IS NOT NULL THEN candidate.published_artifact_kind + WHEN candidate.clear_marker THEN NULL + ELSE s.current_local_artifact_kind + END, + current_local_artifact_origin = CASE + WHEN candidate.published_artifact_id IS NOT NULL THEN candidate.published_artifact_origin + WHEN candidate.clear_marker THEN NULL + ELSE s.current_local_artifact_origin + END FROM eligible_candidate candidate WHERE s.id = candidate.id - RETURNING s.id, candidate.clear_marker", - active_execution_assignments = finalization.active_execution_assignments("e"), - open_port_exposure_assignments = finalization.open_port_exposure_assignments(), - sandbox_assignments = finalization.sandbox_assignments(), - ); - sqlx::query_as(sqlx::AssertSqlSafe(sql!(finalization_query.as_str()))) + RETURNING s.id, candidate.clear_marker" + )) + }); + sqlx::query_as(sqlx::AssertSqlSafe(FINALIZATION_QUERY.as_str())) .bind(&candidate_ids) .bind(&clear_markers) .bind(&published_artifact_ids) diff --git a/crates/l30_storage/sandbox/store/src/orphan.rs b/crates/l30_storage/sandbox/store/src/orphan.rs index 1cb5f7e17c..837a977a63 100644 --- a/crates/l30_storage/sandbox/store/src/orphan.rs +++ b/crates/l30_storage/sandbox/store/src/orphan.rs @@ -6,8 +6,7 @@ //! read (`SELECT ... FOR UPDATE` on the same guard), the transaction, the //! reconcile-state cleanup, and the quarantine state and history rows. Each //! write here is one statement whose CTEs also retire the active execution -//! and reset open port exposures, as `pplx-space-sandbox-portability`'s -//! finalization renders them; splitting it would change what it does. +//! and reset open port exposures; splitting it would change what it does. //! //! The classification guard fragments ([`classify_guard_sql`], //! [`classify_guard_from_sql`] and [`classify_guard_lock_sql`]) are also @@ -17,11 +16,7 @@ use pplx_space_common_core::types::{NodeId, PodUid, SandboxId, SandboxLifecycle}; use pplx_space_common_core::sql; -use pplx_space_sandbox_portability::{ - SuspendedTargetNodeAssignment, clear_current_local_artifact_marker_assignments, - quarantined_k8s_finalization_sql, started_repair_revert_eligibility, - suspended_k8s_finalization_sql, -}; +use pplx_space_sandbox_portability::started_repair_revert_eligibility; use sqlx::{Postgres, Transaction}; /// The classification guard a statement here re-checks, with the values it @@ -93,17 +88,6 @@ pub async fn recover_orphan_to_suspended( let eligibility = started_repair_revert_eligibility("s"); let clear_marker = eligibility.clear_marker; let eligible = eligibility.eligible; - let artifact_marker_assignments = - clear_current_local_artifact_marker_assignments("s", "pre.clear_marker"); - let finalization = suspended_k8s_finalization_sql( - "s", - "updated_sandbox", - &artifact_marker_assignments, - SuspendedTargetNodeAssignment::Clear, - ); - let sandbox_assignments = finalization.sandbox_assignments(); - let retire_active_execution_cte = finalization.retire_active_execution_cte(); - let invalidate_open_port_exposures_cte = finalization.invalidate_open_port_exposures_cte(); let guard_sql = classify_guard_sql(guard); let guard_from_sql = classify_guard_from_sql(guard); let guard_lock_sql = classify_guard_lock_sql(guard); @@ -122,7 +106,15 @@ pub async fn recover_orphan_to_suspended( ), updated_sandbox AS ( UPDATE sandboxes s - SET {sandbox_assignments} + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = NULL, + placement_changed_at = NULL, + current_local_artifact_id = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_origin END FROM pre WHERE s.id = pre.id RETURNING s.id, @@ -130,8 +122,32 @@ pub async fn recover_orphan_to_suspended( pre.active_node_id, pre.clear_marker ), - {retire_active_execution_cte}, - {invalidate_open_port_exposures_cte} + retired_execution AS (\n\ + UPDATE sandbox_k8s_executions e\n\ + SET active = FALSE,\n\ + state = 'suspended',\n\ + pod_ip = NULL,\n\ + terminated_at = COALESCE(e.terminated_at, NOW()),\n\ + updated_at = NOW()\n\ + FROM updated_sandbox\n\ + WHERE e.sandbox_id = updated_sandbox.id\n\ + AND e.active\n\ + RETURNING e.sandbox_id\n\ + ), + reset_port_exposures AS (\n\ + UPDATE sandbox_port_exposures spe\n\ + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW()\n\ + FROM updated_sandbox\n\ + WHERE spe.sandbox_id = updated_sandbox.id\n\ + AND spe.desired_state = 'open'\n\ + RETURNING spe.sandbox_id\n\ + ) SELECT from_lifecycle, active_node_id, clear_marker FROM updated_sandbox" ); @@ -173,15 +189,6 @@ pub async fn quarantine_orphan( // exposure rows keep their `desired_state='open'` user intent and only // have the application observation reset, so the intent re-applies if an // operator later recovers the sandbox. - let finalization = quarantined_k8s_finalization_sql( - "s", - "updated_sandbox", - "", - SuspendedTargetNodeAssignment::Clear, - ); - let sandbox_assignments = finalization.sandbox_assignments(); - let retire_active_execution_cte = finalization.retire_active_execution_cte(); - let invalidate_open_port_exposures_cte = finalization.invalidate_open_port_exposures_cte(); let guard_sql = classify_guard_sql(guard); let guard_from_sql = classify_guard_from_sql(guard); let guard_lock_sql = classify_guard_lock_sql(guard); @@ -199,7 +206,12 @@ pub async fn quarantine_orphan( ), updated_sandbox AS ( UPDATE sandboxes s - SET {sandbox_assignments} + SET lifecycle = 'quarantined', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = NULL, + placement_changed_at = NULL FROM pre WHERE s.id = pre.id RETURNING s.id, @@ -207,8 +219,32 @@ pub async fn quarantine_orphan( pre.active_node_id, FALSE AS clear_marker ), - {retire_active_execution_cte}, - {invalidate_open_port_exposures_cte} + retired_execution AS (\n\ + UPDATE sandbox_k8s_executions e\n\ + SET active = FALSE,\n\ + state = 'quarantined',\n\ + pod_ip = NULL,\n\ + terminated_at = COALESCE(e.terminated_at, NOW()),\n\ + updated_at = NOW()\n\ + FROM updated_sandbox\n\ + WHERE e.sandbox_id = updated_sandbox.id\n\ + AND e.active\n\ + RETURNING e.sandbox_id\n\ + ), + reset_port_exposures AS (\n\ + UPDATE sandbox_port_exposures spe\n\ + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW()\n\ + FROM updated_sandbox\n\ + WHERE spe.sandbox_id = updated_sandbox.id\n\ + AND spe.desired_state = 'open'\n\ + RETURNING spe.sandbox_id\n\ + ) SELECT from_lifecycle, active_node_id, clear_marker FROM updated_sandbox" ); diff --git a/crates/l30_storage/sandbox/store/src/pod_tracking.rs b/crates/l30_storage/sandbox/store/src/pod_tracking.rs index b55ab710b7..8654541fb1 100644 --- a/crates/l30_storage/sandbox/store/src/pod_tracking.rs +++ b/crates/l30_storage/sandbox/store/src/pod_tracking.rs @@ -12,10 +12,12 @@ use pplx_space_common_core::types::{ExecutionGeneration, SandboxId}; +use std::sync::LazyLock; + use pplx_space_common_core::sql; use pplx_space_sandbox_portability::{ - clear_current_local_artifact_marker_assignments, failed_repair_revert_eligibility, - restore_retry_suspended_revert_snapshot_exists, started_suspend_revert_eligibility, + failed_repair_revert_eligibility, restore_retry_suspended_revert_snapshot_exists, + started_suspend_revert_eligibility, }; use sqlx::{Postgres, Transaction}; @@ -169,10 +171,35 @@ async fn recover_started_with_current_suspend_snapshot( sandbox_id: SandboxId, proof: StartedSuspendRecoveryProof, ) -> Result { - let (generation, execution_proof) = match proof { - StartedSuspendRecoveryProof::ExecutionRetired => ( - None, - "NOT EXISTS ( + /// Built once: every input is a literal or a bind placeholder. + static EXECUTION_RETIRED_QUERY: LazyLock = LazyLock::new(|| { + let eligibility = started_suspend_revert_eligibility("s"); + sql!(format!( + "WITH args AS ( + SELECT $2::BIGINT AS generation + ), + pre AS ( + SELECT id, + {clear_marker} AS clear_marker + FROM sandboxes s + WHERE s.id = $1 + ) + UPDATE sandboxes s + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + target_node_id = NULL, + placement_changed_at = NULL, + command_ready_pending = FALSE, + current_local_artifact_id = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_origin END + FROM pre, args + WHERE s.id = pre.id + AND s.lifecycle = 'started' + AND s.placement = 'unplaced' + AND s.active_node_id IS NULL + AND NOT EXISTS ( SELECT 1 FROM sandbox_k8s_executions e WHERE e.sandbox_id = s.id @@ -183,40 +210,17 @@ async fn recover_started_with_current_suspend_snapshot( FROM sandbox_k8s_executions e WHERE e.sandbox_id = s.id AND e.pod_ip IS NOT NULL - )", - ), - StartedSuspendRecoveryProof::FailedRecovery { generation } => ( - Some(generation.get()), - "EXISTS ( - SELECT 1 - FROM sandbox_k8s_executions e - WHERE e.sandbox_id = s.id - AND e.generation = args.generation - AND e.active - AND e.state = 'recovery_failed' - AND e.pod_uid IS NULL - AND e.pod_ip IS NULL - AND e.k8s_node_name IS NULL ) - AND NOT EXISTS ( - SELECT 1 - FROM sandbox_k8s_executions newer - WHERE newer.sandbox_id = s.id - AND newer.active - AND newer.generation > args.generation - )", - ), - }; - let eligibility = match proof { - StartedSuspendRecoveryProof::ExecutionRetired => started_suspend_revert_eligibility("s"), - StartedSuspendRecoveryProof::FailedRecovery { .. } => failed_repair_revert_eligibility("s"), - }; - let clear_marker = eligibility.clear_marker; - let eligible = eligibility.eligible; - let artifact_marker_assignments = - clear_current_local_artifact_marker_assignments("s", "pre.clear_marker"); - let query = format!( - "WITH args AS ( + AND {eligible}", + clear_marker = eligibility.clear_marker, + eligible = eligibility.eligible, + )) + }); + /// Built once: every input is a literal or a bind placeholder. + static FAILED_RECOVERY_QUERY: LazyLock = LazyLock::new(|| { + let eligibility = failed_repair_revert_eligibility("s"); + sql!(format!( + "WITH args AS ( SELECT $2::BIGINT AS generation ), pre AS ( @@ -232,16 +236,44 @@ async fn recover_started_with_current_suspend_snapshot( target_node_id = NULL, placement_changed_at = NULL, command_ready_pending = FALSE, - {artifact_marker_assignments} + current_local_artifact_id = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.clear_marker THEN NULL ELSE s.current_local_artifact_origin END FROM pre, args WHERE s.id = pre.id AND s.lifecycle = 'started' AND s.placement = 'unplaced' AND s.active_node_id IS NULL - AND {execution_proof} - AND {eligible}" - ); - let result = sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) + AND EXISTS ( + SELECT 1 + FROM sandbox_k8s_executions e + WHERE e.sandbox_id = s.id + AND e.generation = args.generation + AND e.active + AND e.state = 'recovery_failed' + AND e.pod_uid IS NULL + AND e.pod_ip IS NULL + AND e.k8s_node_name IS NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM sandbox_k8s_executions newer + WHERE newer.sandbox_id = s.id + AND newer.active + AND newer.generation > args.generation + ) + AND {eligible}", + clear_marker = eligibility.clear_marker, + eligible = eligibility.eligible, + )) + }); + let (generation, query) = match proof { + StartedSuspendRecoveryProof::ExecutionRetired => (None, &*EXECUTION_RETIRED_QUERY), + StartedSuspendRecoveryProof::FailedRecovery { generation } => { + (Some(generation.get()), &*FAILED_RECOVERY_QUERY) + } + }; + let result = sqlx::query(sqlx::AssertSqlSafe(query.as_str())) .bind(sandbox_id) .bind(generation) .execute(&mut **tx) diff --git a/crates/l30_storage/sandbox/store/src/repair.rs b/crates/l30_storage/sandbox/store/src/repair.rs index 8f44fcff3c..5f787d2cba 100644 --- a/crates/l30_storage/sandbox/store/src/repair.rs +++ b/crates/l30_storage/sandbox/store/src/repair.rs @@ -14,11 +14,12 @@ use pplx_space_common_core::types::{LocalArtifactId, PositiveDuration, SandboxId, SubstrateKind}; +use std::sync::LazyLock; + use pplx_space_common_core::sql; use pplx_space_sandbox_portability::{ - SuspendedTargetNodeAssignment, clear_current_local_artifact_marker_assignments, failed_repair_revert_eligibility, newest_suspend_snapshot_current, - started_repair_revert_eligibility, suspended_k8s_finalization_sql, + started_repair_revert_eligibility, }; use sqlx::{Postgres, Transaction}; @@ -80,24 +81,26 @@ pub async fn revert_started_to_suspended( min_age: PositiveDuration, newest_generation: i64, ) -> Result, sqlx::Error> { - let eligibility = started_repair_revert_eligibility("s"); - let artifact_marker_assignments = - clear_current_local_artifact_marker_assignments("s", "pre.will_clear"); - let finalization = suspended_k8s_finalization_sql( - "s", - "pre", - &artifact_marker_assignments, - SuspendedTargetNodeAssignment::Clear, - ); - let query = format!( - "WITH pre AS ( + /// Built once: every input is a literal or a bind placeholder. + static QUERY: LazyLock = LazyLock::new(|| { + let eligibility = started_repair_revert_eligibility("s"); + sql!(format!( + "WITH pre AS ( SELECT id, {clear_marker} AS will_clear FROM sandboxes s WHERE s.id = $1 ) UPDATE sandboxes s - SET {sandbox_assignments} + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = NULL, + placement_changed_at = NULL, + current_local_artifact_id = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_origin END FROM pre WHERE s.id = pre.id AND {guards} @@ -106,12 +109,12 @@ pub async fn revert_started_to_suspended( WHERE e.sandbox_id = s.id) = $3 AND {eligible} RETURNING pre.will_clear", - clear_marker = eligibility.clear_marker, - sandbox_assignments = finalization.sandbox_assignments(), - guards = started_repair_row_state_guards("$4", "$2"), - eligible = eligibility.eligible, - ); - sqlx::query_scalar(sqlx::AssertSqlSafe(sql!(query.as_str()))) + clear_marker = eligibility.clear_marker, + guards = started_repair_row_state_guards("$4", "$2"), + eligible = eligibility.eligible, + )) + }); + sqlx::query_scalar(sqlx::AssertSqlSafe(QUERY.as_str())) .bind(sandbox_id) .bind(duration_secs_i64(min_age)) .bind(newest_generation) @@ -195,22 +198,25 @@ pub async fn revert_failed_to_suspended( sandbox_id: SandboxId, min_age: PositiveDuration, ) -> Result, sqlx::Error> { - let eligibility = failed_repair_revert_eligibility("s"); - let marker_assignments = clear_current_local_artifact_marker_assignments("s", "pre.will_clear"); - let finalization = suspended_k8s_finalization_sql( - "s", - "pre", - &marker_assignments, - SuspendedTargetNodeAssignment::Clear, - ); - let query = format!( - "WITH pre AS ( + /// Built once: every input is a literal or a bind placeholder. + static QUERY: LazyLock = LazyLock::new(|| { + let eligibility = failed_repair_revert_eligibility("s"); + sql!(format!( + "WITH pre AS ( SELECT id, {clear} AS will_clear FROM sandboxes s WHERE s.id = $1 ) UPDATE sandboxes s - SET {sandbox_assignments}, + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + target_node_id = NULL, + placement_changed_at = NULL, + current_local_artifact_id = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN pre.will_clear THEN NULL ELSE s.current_local_artifact_origin END, restore_lease_expires_at = NULL, restore_lease_token = NULL FROM pre @@ -218,12 +224,12 @@ pub async fn revert_failed_to_suspended( AND {guards} AND {eligible} RETURNING pre.will_clear", - clear = eligibility.clear_marker, - sandbox_assignments = finalization.sandbox_assignments(), - guards = failed_repair_row_state_guards("$3", "$2"), - eligible = eligibility.eligible, - ); - sqlx::query_scalar(sqlx::AssertSqlSafe(sql!(query.as_str()))) + clear = eligibility.clear_marker, + guards = failed_repair_row_state_guards("$3", "$2"), + eligible = eligibility.eligible, + )) + }); + sqlx::query_scalar(sqlx::AssertSqlSafe(QUERY.as_str())) .bind(sandbox_id) .bind(duration_secs_i64(min_age)) .bind(SubstrateKind::Kubernetes) diff --git a/crates/l30_storage/sandbox/store/src/suspend.rs b/crates/l30_storage/sandbox/store/src/suspend.rs index e5f01459ac..4239e79c20 100644 --- a/crates/l30_storage/sandbox/store/src/suspend.rs +++ b/crates/l30_storage/sandbox/store/src/suspend.rs @@ -14,10 +14,6 @@ use pplx_space_common_core::types::{ }; use pplx_space_common_core::sql; -use pplx_space_sandbox_portability::{ - SuspendedTargetNodeAssignment, clear_current_local_artifact_marker_assignments, - suspended_k8s_finalization_sql, write_current_local_artifact_marker_assignments, -}; /// **Operation step.** Claim the suspend of the execution at `generation` /// tracking `pod_uid`: flip the row from `previous` to `suspending` while @@ -90,7 +86,7 @@ pub async fn record_committed_artifact( /// and unplaced, with `artifact_id` as its current local `suspend` artifact. /// /// One statement whose CTEs also write `sandbox_k8s_executions` and the port -/// exposures (`suspended_k8s_finalization_sql`). It has no lifecycle +/// exposures. It has no lifecycle /// predicate of its own and matches the row by id alone: it relies on the /// caller's sandbox and execution row locks and the re-checks it made under /// them in the same transaction. @@ -100,37 +96,55 @@ pub async fn finalize_with_committed_artifact( artifact_id: LocalArtifactId, artifact_kind: ArtifactKind, ) -> Result<(), sqlx::Error> { - let marker_assignments = write_current_local_artifact_marker_assignments("$2", "$3", "$4"); - let finalization = suspended_k8s_finalization_sql( - "s", - "candidate", - &marker_assignments, - SuspendedTargetNodeAssignment::Preserve, - ); - let query = format!( + sqlx::query(sql!( "WITH candidate AS MATERIALIZED ( SELECT id FROM sandboxes WHERE id = $1 ), - {retire_active_execution_cte}, - {invalidate_open_port_exposures_cte} + retired_execution AS (\n\ + UPDATE sandbox_k8s_executions e\n\ + SET active = FALSE,\n\ + state = 'suspended',\n\ + pod_ip = NULL,\n\ + terminated_at = COALESCE(e.terminated_at, NOW()),\n\ + updated_at = NOW()\n\ + FROM candidate\n\ + WHERE e.sandbox_id = candidate.id\n\ + AND e.active\n\ + RETURNING e.sandbox_id\n\ + ), + reset_port_exposures AS (\n\ + UPDATE sandbox_port_exposures spe\n\ + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW()\n\ + FROM candidate\n\ + WHERE spe.sandbox_id = candidate.id\n\ + AND spe.desired_state = 'open'\n\ + RETURNING spe.sandbox_id\n\ + ) UPDATE sandboxes s - SET {sandbox_assignments} + SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + current_local_artifact_id = $2,\n\ + current_local_artifact_kind = $3,\n\ + current_local_artifact_origin = $4 FROM candidate - WHERE s.id = candidate.id", - retire_active_execution_cte = finalization.retire_active_execution_cte(), - invalidate_open_port_exposures_cte = finalization.invalidate_open_port_exposures_cte(), - sandbox_assignments = finalization.sandbox_assignments(), - ); - - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(sandbox_id) - .bind(artifact_id.as_uuid()) - .bind(artifact_kind.as_str()) - .bind(ArtifactOrigin::Suspend.as_str()) - .execute(executor) - .await?; + WHERE s.id = candidate.id" + )) + .bind(sandbox_id) + .bind(artifact_id.as_uuid()) + .bind(artifact_kind.as_str()) + .bind(ArtifactOrigin::Suspend.as_str()) + .execute(executor) + .await?; Ok(()) } @@ -147,16 +161,15 @@ pub async fn finalize_missing_pod_from_current_snapshot( sandbox_id: SandboxId, clear_marker: bool, ) -> Result<(), sqlx::Error> { - let marker_assignments = clear_current_local_artifact_marker_assignments("s", "$2"); - let sandbox_assignments = suspended_k8s_finalization_sql( - "s", - "candidate", - &marker_assignments, - SuspendedTargetNodeAssignment::Preserve, - ) - .sandbox_assignments(); - let query = format!("UPDATE sandboxes s SET {sandbox_assignments} WHERE s.id = $1"); - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) + sqlx::query(sql!( + "UPDATE sandboxes s SET lifecycle = 'suspended', + placement = 'unplaced', + active_node_id = NULL, + command_ready_pending = FALSE, + current_local_artifact_id = CASE WHEN $2 THEN NULL ELSE s.current_local_artifact_id END,\n\ + current_local_artifact_kind = CASE WHEN $2 THEN NULL ELSE s.current_local_artifact_kind END,\n\ + current_local_artifact_origin = CASE WHEN $2 THEN NULL ELSE s.current_local_artifact_origin END WHERE s.id = $1" + )) .bind(sandbox_id) .bind(clear_marker) .execute(executor) diff --git a/crates/l40_api/cluster/controller/server/src/failed_restorable_repair/store.rs b/crates/l40_api/cluster/controller/server/src/failed_restorable_repair/store.rs index ab46cd71a4..49a0324622 100644 --- a/crates/l40_api/cluster/controller/server/src/failed_restorable_repair/store.rs +++ b/crates/l40_api/cluster/controller/server/src/failed_restorable_repair/store.rs @@ -6,9 +6,7 @@ use pplx_space_common_core::types::SubstrateKind; use pplx_space_common_core::sql; use pplx_space_common_core::types::PositiveDuration; -use pplx_space_sandbox_portability::{ - SuspendedTargetNodeAssignment, failed_repair_revert_eligibility, suspended_k8s_finalization_sql, -}; +use pplx_space_sandbox_portability::failed_repair_revert_eligibility; use pplx_space_sandbox_store::repair; use sqlx::PgPool; @@ -98,20 +96,20 @@ impl RepairDbStore { .execute(&mut *tx) .await?; - // Only the port-exposure assignments are read from this renderer; - // they do not depend on the marker policy the flip above applied. - let finalization = - suspended_k8s_finalization_sql("s", "pre", "", SuspendedTargetNodeAssignment::Clear); - let query = format!( + sqlx::query(sql!( "UPDATE sandbox_port_exposures - SET {assignments} - WHERE sandbox_id = $1 AND desired_state = 'open'", - assignments = finalization.open_port_exposure_assignments(), - ); - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(candidate.sandbox_id) - .execute(&mut *tx) - .await?; + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW() + WHERE sandbox_id = $1 AND desired_state = 'open'" + )) + .bind(candidate.sandbox_id) + .execute(&mut *tx) + .await?; tx.commit().await?; Ok(if cleared_marker { diff --git a/crates/l40_api/cluster/controller/server/src/node_drain/store/pod_less_giveup.rs b/crates/l40_api/cluster/controller/server/src/node_drain/store/pod_less_giveup.rs index 888c092dd7..9800d81ec2 100644 --- a/crates/l40_api/cluster/controller/server/src/node_drain/store/pod_less_giveup.rs +++ b/crates/l40_api/cluster/controller/server/src/node_drain/store/pod_less_giveup.rs @@ -119,8 +119,8 @@ use pplx_space_common_core::types::{ SubstrateKind, }; use pplx_space_sandbox_portability::{ - SuspendedK8sFinalizationSql, failed_repair_revert_eligibility, - newest_full_snapshot_unambiguous, pod_less_drain_revert_loses_no_newer_bytes, + failed_repair_revert_eligibility, newest_full_snapshot_unambiguous, + pod_less_drain_revert_loses_no_newer_bytes, }; use pplx_space_sandbox_store::drain; use sqlx::PgPool; @@ -644,23 +644,21 @@ async fn recover_pod_less_sandbox_locked( return Ok(PodLessRecovery::NotEligible(hold)); } - let Some(finalization) = - drain::revert_pod_less_resident_to_suspended(&mut tx, candidate.sandbox_id).await? - else { + if !drain::revert_pod_less_resident_to_suspended(&mut tx, candidate.sandbox_id).await? { // The guards just passed under the locks, so only a snapshot change // that the locks don't cover gets here. tx.rollback().await?; return Ok(PodLessRecovery::NotEligible(PodLessHold::SnapshotChanged)); - }; + } - retire_executions(&mut tx, candidate.sandbox_id, &finalization).await?; + retire_executions(&mut tx, candidate.sandbox_id).await?; WarmPoolStore::delete_claimed_slot_for_retired_execution(&mut tx, candidate.sandbox_id) .await .map_err(|err| match err { PoolError::Sql(err) => err, other => sqlx::Error::Protocol(format!("delete claimed pool slot: {other}")), })?; - reset_open_port_exposures(&mut tx, candidate.sandbox_id, &finalization).await?; + reset_open_port_exposures(&mut tx, candidate.sandbox_id).await?; // The observed guard tolerated a `failed`, pid-less record. Left behind, // it would keep a reverted row reading `failed` through the observed-state // overlay, and once the node is gone nothing would publish over it. @@ -679,19 +677,20 @@ async fn recover_pod_less_sandbox_locked( async fn retire_executions( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, sandbox_id: SandboxId, - finalization: &SuspendedK8sFinalizationSql, ) -> Result<(), sqlx::Error> { - let query = format!( + sqlx::query(sql!( "UPDATE sandbox_k8s_executions e - SET {assignments} + SET active = FALSE,\n\ + state = 'suspended',\n\ + pod_ip = NULL,\n\ + terminated_at = COALESCE(e.terminated_at, NOW()),\n\ + updated_at = NOW() WHERE e.sandbox_id = $1 - AND e.active", - assignments = finalization.active_execution_assignments("e"), - ); - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(sandbox_id) - .execute(&mut **tx) - .await?; + AND e.active" + )) + .bind(sandbox_id) + .execute(&mut **tx) + .await?; Ok(()) } @@ -706,18 +705,21 @@ pub(crate) fn is_deadlock(error: &sqlx::Error) -> bool { async fn reset_open_port_exposures( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, sandbox_id: SandboxId, - finalization: &SuspendedK8sFinalizationSql, ) -> Result<(), sqlx::Error> { - let query = format!( + sqlx::query(sql!( "UPDATE sandbox_port_exposures - SET {assignments} + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW() WHERE sandbox_id = $1 - AND desired_state = 'open'", - assignments = finalization.open_port_exposure_assignments(), - ); - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(sandbox_id) - .execute(&mut **tx) - .await?; + AND desired_state = 'open'" + )) + .bind(sandbox_id) + .execute(&mut **tx) + .await?; Ok(()) } diff --git a/crates/l40_api/cluster/controller/server/src/started_sandbox_repair/store.rs b/crates/l40_api/cluster/controller/server/src/started_sandbox_repair/store.rs index b8b181e71d..7ca5c8311a 100644 --- a/crates/l40_api/cluster/controller/server/src/started_sandbox_repair/store.rs +++ b/crates/l40_api/cluster/controller/server/src/started_sandbox_repair/store.rs @@ -6,10 +6,7 @@ use pplx_space_common_core::types::SubstrateKind; use pplx_space_common_core::sql; use pplx_space_common_core::types::PositiveDuration; -use pplx_space_sandbox_portability::{ - SuspendedK8sFinalizationSql, SuspendedTargetNodeAssignment, started_repair_revert_eligibility, - suspended_k8s_finalization_sql, -}; +use pplx_space_sandbox_portability::started_repair_revert_eligibility; use pplx_space_sandbox_store::repair; use sqlx::PgPool; @@ -173,11 +170,7 @@ impl RepairDbStore { return Ok(ApplyOutcome::SkippedRaced); }; - // Only the port-exposure assignments are read from this renderer; - // they do not depend on the marker policy the flip above applied. - let finalization = - suspended_k8s_finalization_sql("s", "pre", "", SuspendedTargetNodeAssignment::Clear); - self.reset_open_port_exposures(&mut tx, candidate.sandbox_id, &finalization) + self.reset_open_port_exposures(&mut tx, candidate.sandbox_id) .await?; tx.commit().await?; @@ -188,19 +181,22 @@ impl RepairDbStore { &self, tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, sandbox_id: SandboxId, - finalization: &SuspendedK8sFinalizationSql, ) -> Result<(), sqlx::Error> { - let query = format!( + sqlx::query(sql!( "UPDATE sandbox_port_exposures - SET {assignments} + SET apply_state = 'pending',\n\ + last_applied_node_id = NULL,\n\ + last_error = NULL,\n\ + last_bridge_status = NULL,\n\ + last_applied_bridge_generation = NULL,\n\ + last_observed_at = NULL,\n\ + updated_at = NOW() WHERE sandbox_id = $1 - AND desired_state = 'open'", - assignments = finalization.open_port_exposure_assignments(), - ); - sqlx::query(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(sandbox_id) - .execute(&mut **tx) - .await?; + AND desired_state = 'open'" + )) + .bind(sandbox_id) + .execute(&mut **tx) + .await?; Ok(()) } } diff --git a/crates/l40_api/internal/api/server/src/node_local_backup_pending.rs b/crates/l40_api/internal/api/server/src/node_local_backup_pending.rs index 1f86374ff9..080da0236d 100644 --- a/crates/l40_api/internal/api/server/src/node_local_backup_pending.rs +++ b/crates/l40_api/internal/api/server/src/node_local_backup_pending.rs @@ -337,9 +337,9 @@ pub(super) enum PendingLocalBackupCensus { /// setting `cleanup_required` /// (`sbx/mgr/store/src/db/local_artifacts.rs`'s `pin_suspended_artifact_for_cleanup`, a /// SQLite writer), and the teardown-suspend path pins one mid-transition, -/// and each of those eventually reaches this column through -/// `publish_clear_or_preserve_current_local_artifact_marker_assignments`. A -/// hand-kept status list that missed one would accept an incomplete report +/// and each of those eventually reaches this column through the node-published +/// suspended finalization (`pplx_space_sandbox_store::ingest::finalize_suspended`). +/// A hand-kept status list that missed one would accept an incomplete report /// and certify an empty queue. Reading the marker the server already stores /// is closed by construction and fails safe for states nobody has thought of /// yet. From 3a4a5ae70ce8fe6afbadfb53ca8ff5a1fac6bd2c Mon Sep 17 00:00:00 2001 From: ac-pplx Date: Wed, 4 Nov 2026 16:37:29 -0500 Subject: [PATCH 2/2] [skip ci] cr-test: preview only, do not merge