From 945e2d84eba8d8ff0ac2826d1c885481b47b3f93 Mon Sep 17 00:00:00 2001 From: Pama-Lee Date: Wed, 8 Jul 2026 23:35:19 +0800 Subject: [PATCH 1/5] fix(platform): allow rollback_failed in release status CHECK constraints The ReleaseExecutionStatus and ReleaseRequestStatus enums both include rollback_failed, but the CHECK constraints from 0008 never did. A failed rollback therefore could not persist its terminal status: the UPDATE hit the constraint, the error was swallowed, the execution stayed rollback_in_progress, and the worker re-claimed and re-ran the failing rollback every 2s poll, forever. Rebuild both CHECKs with the full status set the code writes. --- .../0022_rollback_failed_status.sql | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 crates/ordo-platform/migrations/0022_rollback_failed_status.sql diff --git a/crates/ordo-platform/migrations/0022_rollback_failed_status.sql b/crates/ordo-platform/migrations/0022_rollback_failed_status.sql new file mode 100644 index 00000000..c8156cf1 --- /dev/null +++ b/crates/ordo-platform/migrations/0022_rollback_failed_status.sql @@ -0,0 +1,43 @@ +-- The status enums in code (`ReleaseExecutionStatus`, `ReleaseRequestStatus`) both +-- include 'rollback_failed', but the CHECK constraints from 0008 never did. A failed +-- rollback therefore could not persist its terminal status: the UPDATE violated the +-- constraint, the error was swallowed, the execution stayed 'rollback_in_progress', +-- and the worker re-claimed and re-ran the failing rollback every poll, forever. +-- Rebuild both CHECKs with the full status set the code writes. + +ALTER TABLE release_executions +DROP CONSTRAINT IF EXISTS release_executions_status_check; + +ALTER TABLE release_executions +ADD CONSTRAINT release_executions_status_check CHECK ( + status IN ( + 'preparing', + 'waiting_start', + 'rolling_out', + 'paused', + 'verifying', + 'rollback_in_progress', + 'rollback_failed', + 'completed', + 'failed' + ) +); + +ALTER TABLE release_requests +DROP CONSTRAINT IF EXISTS release_requests_status_check; + +ALTER TABLE release_requests +ADD CONSTRAINT release_requests_status_check CHECK ( + status IN ( + 'draft', + 'pending_approval', + 'approved', + 'rejected', + 'cancelled', + 'executing', + 'completed', + 'failed', + 'rollback_failed', + 'rolled_back' + ) +); From 7978f65219e9f1f96cede3b7887e5e7f7298ea2d Mon Sep 17 00:00:00 2001 From: Pama-Lee Date: Wed, 8 Jul 2026 23:35:38 +0800 Subject: [PATCH 2/5] fix(platform): refuse concurrent rollouts of the same ruleset target The 0020 unique index guarantees at most one active execution per release request, but it cannot see across requests: every direct publish/redeploy mints a fresh internal request, so two concurrent publishes of the same ruleset (or a publish racing a governed execute) would still both roll out. Add a target-level guard - any active execution for the same project + ruleset + environment now returns 409 from both execute_release_request and the direct-publish enqueue path. Also map the 0020 unique-index violation to 409 instead of 500 for the loser of two concurrent executes of the same request. --- .../ordo-platform/src/release/executions.rs | 34 ++++++++++++++++++- crates/ordo-platform/src/ruleset_draft.rs | 17 ++++++++++ crates/ordo-platform/src/store/releases.rs | 32 +++++++++++++++++ 3 files changed, 82 insertions(+), 1 deletion(-) diff --git a/crates/ordo-platform/src/release/executions.rs b/crates/ordo-platform/src/release/executions.rs index 7210d53f..9e425921 100644 --- a/crates/ordo-platform/src/release/executions.rs +++ b/crates/ordo-platform/src/release/executions.rs @@ -17,6 +17,16 @@ fn execution_is_active(execution: &ReleaseExecution) -> bool { ) } +/// True when `err` is the `release_executions_one_active_per_request` unique-index +/// violation — the DB backstop for two concurrent executes of the same request. +/// The loser of that race should get a 409, not a 500. +fn is_active_execution_conflict(err: &anyhow::Error) -> bool { + err.downcast_ref::() + .and_then(|e| e.as_database_error()) + .and_then(|db| db.constraint()) + == Some("release_executions_one_active_per_request") +} + fn instance_is_terminal(status: &ReleaseInstanceStatus) -> bool { matches!( status, @@ -391,6 +401,22 @@ pub async fn execute_release_request( "Release execution is already in progress", )); } + // Cross-request guard: a *different* request (e.g. a direct publish's internal + // request) may already be rolling this ruleset out to this environment. + if state + .store + .has_active_release_execution_for_target( + &project_id, + &release.ruleset_name, + &release.environment_id, + ) + .await + .map_err(PlatformError::Internal)? + { + return Err(PlatformError::conflict( + "Another rollout of this ruleset to this environment is already in progress", + )); + } release_request_can_execute(&release.status, execution_attempts) .map_err(|err| PlatformError::conflict(err.to_string()))?; @@ -439,7 +465,13 @@ pub async fn execute_release_request( let execution = enqueue_release_execution(&state, &release, &env, &bound_servers, &actor, &claims.sub) .await - .map_err(PlatformError::Internal)?; + .map_err(|err| { + if is_active_execution_conflict(&err) { + PlatformError::conflict("Release execution is already in progress") + } else { + PlatformError::Internal(err) + } + })?; Ok(Json(execution)) } diff --git a/crates/ordo-platform/src/ruleset_draft.rs b/crates/ordo-platform/src/ruleset_draft.rs index 8a1acc93..a07c2b80 100644 --- a/crates/ordo-platform/src/ruleset_draft.rs +++ b/crates/ordo-platform/src/ruleset_draft.rs @@ -489,6 +489,23 @@ async fn enqueue_publish_release( bound_servers.push(server); } + // Each publish mints a fresh internal request, so the per-request unique index + // can't stop two concurrent publishes of the same ruleset from both rolling + // out. Refuse while any rollout of this ruleset to this environment is active. + // (Check-then-insert: a milliseconds-wide race window remains. Rule pushes are + // whole-ruleset overwrites, so the worst case is two rollouts racing and each + // server keeping whichever push lands last — never corrupted state.) + if state + .store + .has_active_release_execution_for_target(project_id, ruleset_name, &env.id) + .await + .map_err(PlatformError::Internal)? + { + return Err(PlatformError::conflict( + "A rollout of this ruleset to this environment is already in progress", + )); + } + let request = build_internal_publish_request( org_id, project_id, diff --git a/crates/ordo-platform/src/store/releases.rs b/crates/ordo-platform/src/store/releases.rs index 77dd910c..ab7ed550 100644 --- a/crates/ordo-platform/src/store/releases.rs +++ b/crates/ordo-platform/src/store/releases.rs @@ -934,6 +934,38 @@ impl PlatformStore { Ok(Some(item)) } + /// True when any release execution targeting the same rollout target + /// (project + ruleset + environment) is still active. The per-request unique + /// index (`release_executions_one_active_per_request`) can't see across + /// requests — each direct publish mints a fresh internal request — so this is + /// the guard that refuses a second concurrent rollout of one ruleset onto one + /// environment. + pub async fn has_active_release_execution_for_target( + &self, + project_id: &str, + ruleset_name: &str, + environment_id: &str, + ) -> Result { + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 + FROM release_executions e + JOIN release_requests r ON r.id = e.release_request_id + WHERE r.project_id = $1 + AND r.ruleset_name = $2 + AND r.environment_id = $3 + AND e.status IN ('preparing', 'waiting_start', 'rolling_out', 'paused', + 'verifying', 'rollback_in_progress') + )", + ) + .bind(project_id) + .bind(ruleset_name) + .bind(environment_id) + .fetch_one(&self.pool) + .await?; + Ok(exists) + } + pub async fn list_worker_claimable_release_executions( &self, limit: i64, From 5f42bf0d07b2e12e1169dd9027adf21cde5cd82e Mon Sep 17 00:00:00 2001 From: Pama-Lee Date: Wed, 8 Jul 2026 23:35:52 +0800 Subject: [PATCH 3/5] fix(platform): reconcile release state stranded by worker crashes The old safety net (fail_stuck_active_executions) was dead code: both bootstrap callers passed false, and enabling it would have been wrong anyway - it had no grace period and would kill in-flight rollouts on every restart, and its rollback_failed write predates the 0022 CHECK fix. Remove it. The poll loop already self-heals claimable executions (the session advisory lock releases when a dead worker's connection drops), but a terminal execution is not claimable, so a worker crash between settling the execution and settling the surrounding state stranded rows forever: - dispatched deployments whose execution already finished (studio polls them indefinitely) - executing requests whose latest execution already finished - executing requests that never got an execution row Add a 60s reconciliation task in the worker that repairs all three, with grace periods so it never races the worker's own settle writes. Completed-via-rollback executions are told apart from completed rollouts by their rollback_in_progress -> completed history row, so requests settle to rolled_back exactly as the worker would have written. Repairs are recorded in the request history and are idempotent, healing rows stranded before this task existed. --- .../src/bin/ordo-platform-worker.rs | 6 +- crates/ordo-platform/src/lib.rs | 72 ++++++-- crates/ordo-platform/src/main.rs | 2 +- crates/ordo-platform/src/store/releases.rs | 156 +++++++++++++++--- 4 files changed, 193 insertions(+), 43 deletions(-) diff --git a/crates/ordo-platform/src/bin/ordo-platform-worker.rs b/crates/ordo-platform/src/bin/ordo-platform-worker.rs index 4eddb13c..b6caceef 100644 --- a/crates/ordo-platform/src/bin/ordo-platform-worker.rs +++ b/crates/ordo-platform/src/bin/ordo-platform-worker.rs @@ -17,7 +17,8 @@ use axum::{ use clap::Parser; use ordo_platform::{ bootstrap_platform_store, build_app_state, config::PlatformConfig, connect_platform_store, - init_tracing, metrics, publish_existing_tenants, release, start_server_registry_maintenance, + init_tracing, metrics, publish_existing_tenants, release, start_release_reconciliation, + start_server_registry_maintenance, }; use tracing::{error, info}; @@ -50,8 +51,9 @@ async fn main() -> anyhow::Result<()> { metrics::init(); let store = connect_platform_store(&config).await?; - bootstrap_platform_store(&store, false).await?; + bootstrap_platform_store(&store).await?; start_server_registry_maintenance(store.clone()); + start_release_reconciliation(store.clone()); // Spawn the liveness/metrics server before entering the loop. let health_addr = config.worker_health_addr; diff --git a/crates/ordo-platform/src/lib.rs b/crates/ordo-platform/src/lib.rs index 2e267b7a..41cb6e03 100644 --- a/crates/ordo-platform/src/lib.rs +++ b/crates/ordo-platform/src/lib.rs @@ -93,10 +93,7 @@ pub async fn connect_platform_store(config: &PlatformConfig) -> anyhow::Result, - fail_active_executions: bool, -) -> anyhow::Result<()> { +pub async fn bootstrap_platform_store(store: &Arc) -> anyhow::Result<()> { let orgs = store.list_all_orgs().await.unwrap_or_default(); for org in &orgs { if let Err(e) = store.seed_system_roles(&org.id).await { @@ -127,20 +124,65 @@ pub async fn bootstrap_platform_store( Err(e) => tracing::warn!("fail_stuck_queued_deployments: {}", e), } - if fail_active_executions { - match store.fail_stuck_active_executions().await { - Ok(n) if n > 0 => tracing::warn!( - count = n, - "Marked stuck active release executions as failed on startup" - ), - Ok(_) => {} - Err(e) => tracing::warn!("fail_stuck_active_executions: {}", e), - } - } - Ok(()) } +/// Periodic repair of release state stranded by a worker crash. The poll loop +/// already self-heals claimable executions (the session-scoped advisory lock is +/// released when a dead worker's connection drops), but a *terminal* execution is +/// not claimable — if the worker died between settling the execution and settling +/// the surrounding state, nothing else ever repairs: +/// - `dispatched` deployment rows whose execution already finished, +/// - `executing` requests whose latest execution already finished, +/// - `executing` requests that never got an execution row at all. +/// +/// Spawned by the worker only. Every repair is idempotent, so it also heals rows +/// stranded before this task existed. +pub fn start_release_reconciliation(store: Arc) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(60)); + loop { + interval.tick().await; + let now = chrono::Utc::now(); + + match store.reconcile_dispatched_deployments().await { + Ok(n) if n > 0 => tracing::warn!( + count = n, + "Settled dispatched deployments whose execution already finished" + ), + Ok(_) => {} + Err(e) => tracing::warn!("reconcile_dispatched_deployments: {}", e), + } + + // Grace periods keep the reconciler far away from the worker's own + // settle writes; only rows the worker abandoned are old enough to match. + match store + .reconcile_stranded_executing_requests(now - chrono::Duration::minutes(5)) + .await + { + Ok(n) if n > 0 => tracing::warn!( + count = n, + "Repaired executing requests whose execution already finished" + ), + Ok(_) => {} + Err(e) => tracing::warn!("reconcile_stranded_executing_requests: {}", e), + } + + match store + .fail_executionless_executing_requests(now - chrono::Duration::minutes(15)) + .await + { + Ok(n) if n > 0 => tracing::warn!( + count = n, + "Failed executing requests that never got an execution" + ), + Ok(_) => {} + Err(e) => tracing::warn!("fail_executionless_executing_requests: {}", e), + } + } + }) +} + pub fn start_server_registry_maintenance(store: Arc) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(60)); diff --git a/crates/ordo-platform/src/main.rs b/crates/ordo-platform/src/main.rs index 4c5480f6..0c650f99 100644 --- a/crates/ordo-platform/src/main.rs +++ b/crates/ordo-platform/src/main.rs @@ -46,7 +46,7 @@ async fn main() -> anyhow::Result<()> { } let store = connect_platform_store(&config).await?; - bootstrap_platform_store(&store, false).await?; + bootstrap_platform_store(&store).await?; start_server_registry_maintenance(store.clone()); let state = build_app_state(config.clone(), store, false).await?; diff --git a/crates/ordo-platform/src/store/releases.rs b/crates/ordo-platform/src/store/releases.rs index ab7ed550..afb99b94 100644 --- a/crates/ordo-platform/src/store/releases.rs +++ b/crates/ordo-platform/src/store/releases.rs @@ -1021,33 +1021,139 @@ impl PlatformStore { Ok(count) } - /// On startup, mark any release execution stuck in an active non-terminal state as terminal. - /// These are executions where the platform spawned a background task that was killed mid-run. - /// Rollback flows become `rollback_failed`; rollout flows become `failed`. - pub async fn fail_stuck_active_executions(&self) -> Result { + /// Settle `dispatched` deployment rows whose backing release execution already + /// reached a terminal status — the worker died between finishing the execution + /// and flipping the deployment row, and a terminal execution is not claimable, + /// so nothing else ever settles them. A completed rollout settles to `success`; + /// a completed *rollback* (the execution's terminal history row transitioned + /// from `rollback_in_progress`) means the publish did not stick, so it settles + /// to `failed` like the other terminal states. Idempotent and race-free against + /// the worker's own settle: both write the same value derived from the same + /// execution status. + pub async fn reconcile_dispatched_deployments(&self) -> Result { let result = sqlx::query( - "WITH stuck AS ( - UPDATE release_executions - SET status = CASE - WHEN status = 'rollback_in_progress' THEN 'rollback_failed' - ELSE 'failed' - END, - finished_at = NOW() - WHERE status IN ('preparing', 'waiting_start', 'rolling_out', 'verifying', 'rollback_in_progress') - RETURNING release_request_id, status - ) - UPDATE release_requests - SET status = CASE - WHEN EXISTS ( - SELECT 1 FROM stuck - WHERE stuck.release_request_id = release_requests.id - AND stuck.status = 'rollback_failed' - ) THEN 'rollback_failed' - ELSE 'failed' - END - WHERE id IN (SELECT release_request_id FROM stuck) - AND status = 'executing'", + "UPDATE ruleset_deployments d + SET status = CASE + WHEN latest.status = 'completed' AND NOT EXISTS ( + SELECT 1 FROM release_request_history h + WHERE h.release_execution_id = latest.id + AND h.action = 'execution_status_changed' + AND h.from_status = 'rollback_in_progress' + AND h.to_status = 'completed' + ) THEN 'success' + ELSE 'failed' + END + FROM release_requests r + CROSS JOIN LATERAL ( + SELECT e.id, e.status + FROM release_executions e + WHERE e.release_request_id = r.id + ORDER BY e.started_at DESC + LIMIT 1 + ) latest + WHERE d.status = 'dispatched' + AND r.request_snapshot->>'publish_deployment_id' = d.id + AND latest.status IN ('completed', 'failed', 'rollback_failed')", + ) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Repair `executing` release requests whose *latest* execution reached a + /// terminal status more than a grace period ago (`finished_before`) — the + /// worker died between settling the execution and updating the request. The + /// request settles to the status the worker would have written, including + /// `rolled_back` when the completed execution arrived via a rollback, and the + /// transition is recorded in the request history. + pub async fn reconcile_stranded_executing_requests( + &self, + finished_before: chrono::DateTime, + ) -> Result { + let result = sqlx::query( + "WITH latest AS ( + SELECT DISTINCT ON (e.release_request_id) + e.release_request_id AS request_id, + e.id AS execution_id, + e.status AS exec_status, + e.finished_at + FROM release_executions e + JOIN release_requests r ON r.id = e.release_request_id + WHERE r.status = 'executing' + ORDER BY e.release_request_id, e.started_at DESC + ), + repaired AS ( + UPDATE release_requests r + SET status = CASE + WHEN l.exec_status = 'failed' THEN 'failed' + WHEN l.exec_status = 'rollback_failed' THEN 'rollback_failed' + WHEN EXISTS ( + SELECT 1 FROM release_request_history h + WHERE h.release_execution_id = l.execution_id + AND h.action = 'execution_status_changed' + AND h.from_status = 'rollback_in_progress' + AND h.to_status = 'completed' + ) THEN 'rolled_back' + ELSE 'completed' + END, + updated_at = NOW() + FROM latest l + WHERE r.id = l.request_id + AND r.status = 'executing' + AND l.exec_status IN ('completed', 'failed', 'rollback_failed') + AND l.finished_at IS NOT NULL + AND l.finished_at < $1 + RETURNING r.id, l.execution_id, r.status AS to_status + ) + INSERT INTO release_request_history + (id, release_request_id, release_execution_id, instance_id, scope, action, + actor_type, actor_id, actor_name, actor_email, from_status, to_status, + detail, created_at) + SELECT gen_random_uuid()::text, repaired.id, repaired.execution_id, NULL, + 'request', 'request_status_changed', + 'system', 'release_reconciler', 'release_reconciler', NULL, + 'executing', repaired.to_status, + jsonb_build_object('reason', 'reconciled_stranded_request'), NOW() + FROM repaired", + ) + .bind(finished_before) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Fail `executing` release requests that still have no execution row after a + /// grace period (`updated_before`) — the execute call flipped the request to + /// `executing` and then died (or the execution insert itself failed), so no + /// worker will ever pick it up. Failing it makes the request retryable. + pub async fn fail_executionless_executing_requests( + &self, + updated_before: chrono::DateTime, + ) -> Result { + let result = sqlx::query( + "WITH repaired AS ( + UPDATE release_requests r + SET status = 'failed', updated_at = NOW() + WHERE r.status = 'executing' + AND r.updated_at < $1 + AND NOT EXISTS ( + SELECT 1 FROM release_executions e + WHERE e.release_request_id = r.id + ) + RETURNING r.id + ) + INSERT INTO release_request_history + (id, release_request_id, release_execution_id, instance_id, scope, action, + actor_type, actor_id, actor_name, actor_email, from_status, to_status, + detail, created_at) + SELECT gen_random_uuid()::text, repaired.id, NULL, NULL, + 'request', 'request_status_changed', + 'system', 'release_reconciler', 'release_reconciler', NULL, + 'executing', 'failed', + jsonb_build_object('reason', 'reconciled_executionless_request'), NOW() + FROM repaired", ) + .bind(updated_before) .execute(&self.pool) .await?; Ok(result.rows_affected()) From 81a39d2635408640d257ab551ac52ef0739ab75b Mon Sep 17 00:00:00 2001 From: Pama-Lee Date: Thu, 9 Jul 2026 00:27:09 +0800 Subject: [PATCH 4/5] fix(platform): settle rollout when a bound server is deleted mid-flight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rolling_out / rollback_in_progress execution is worker-claimable, and build_rolling_context / build_rollback_context hard-errored when a row they depend on was gone (a bound server pruned as stale >30min or deleted by an admin, the request, environment, draft, or rollback snapshot). The spawned worker task only logged that error and dropped its advisory lock without writing any status, so the execution stayed claimable and was re-claimed and re-failed every 2s poll forever. The reconciler could not help — it only repairs executions that already reached a terminal status. Distinguish a permanent missing dependency (SetupDependencyMissing, from a None lookup) from a transient store error (propagated, retried next poll). On a permanent failure, settle the execution terminally (Failed, or RollbackFailed for a rollback) along with its request and any direct-publish deployment row, instead of looping. --- .../ordo-platform/src/release/executions.rs | 191 +++++++++++++++++- 1 file changed, 181 insertions(+), 10 deletions(-) diff --git a/crates/ordo-platform/src/release/executions.rs b/crates/ordo-platform/src/release/executions.rs index 9e425921..cc47c471 100644 --- a/crates/ordo-platform/src/release/executions.rs +++ b/crates/ordo-platform/src/release/executions.rs @@ -37,6 +37,126 @@ fn instance_is_terminal(status: &ReleaseInstanceStatus) -> bool { ) } +/// A permanent failure while assembling a deployment context: a row the execution +/// depends on (a bound server, the request, its environment, or the draft) is gone +/// and will not come back on its own. Distinguished from a transient store error +/// so the worker settles the execution terminally instead of re-claiming it every +/// poll forever — a claimable status plus a hard error is an infinite loop, since +/// the spawned task only logs and drops its lock without writing any status. +#[derive(Debug)] +struct SetupDependencyMissing(String); + +impl std::fmt::Display for SetupDependencyMissing { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for SetupDependencyMissing {} + +/// Either return the built context, or — when the failure is a permanent missing +/// dependency — settle the execution terminally and stop. Transient errors are +/// propagated so the next poll retries. +async fn settle_or_retry_setup_failure( + state: &AppState, + execution_id: &str, + is_rollback: bool, + err: anyhow::Error, +) -> anyhow::Result<()> { + if err.downcast_ref::().is_none() { + // Transient (e.g. a store/DB blip) — let the next poll re-claim and retry. + return Err(err); + } + warn!( + execution_id, + "release execution cannot proceed (dependency gone), settling as failed: {err}" + ); + fail_execution_terminally(state, execution_id, is_rollback, &err.to_string()).await; + Ok(()) +} + +/// Drive an execution to its terminal failed status (`Failed`, or `RollbackFailed` +/// for a rollback) together with its request and any direct-publish deployment row, +/// so a rollout whose target vanished doesn't strand as claimable. Best-effort: a +/// failed settle write is logged and left for the reconciler, never retried into a +/// loop. +async fn fail_execution_terminally( + state: &AppState, + execution_id: &str, + is_rollback: bool, + reason: &str, +) { + let system_actor = system_history_actor("release_rollout_worker"); + let Ok(Some(execution)) = state.store.get_release_execution(execution_id).await else { + return; + }; + let release_id = execution.request_id.clone(); + let (exec_terminal, request_terminal) = if is_rollback { + ( + ReleaseExecutionStatus::RollbackFailed, + ReleaseRequestStatus::RollbackFailed, + ) + } else { + (ReleaseExecutionStatus::Failed, ReleaseRequestStatus::Failed) + }; + + // Mark still-live instances failed so the UI doesn't show them mid-flight under + // a failed execution. Best-effort — an invalid transition is harmless here. + for instance in &execution.instances { + if !instance_is_terminal(&instance.status) { + let _ = update_instance_status_with_history( + state, + &release_id, + execution_id, + &instance.id, + ReleaseInstanceStatus::Failed, + Some(reason), + None, + &system_actor, + "instance_status_changed", + serde_json::json!({ "reason": "setup_dependency_missing" }), + ) + .await; + } + } + + if let Err(e) = update_release_execution_status_with_history( + state, + &release_id, + execution_id, + None, + exec_terminal, + None, + &system_actor, + serde_json::json!({ "reason": "setup_dependency_missing", "detail": reason }), + ) + .await + { + error!(execution_id, "Failed to settle stranded execution: {e}"); + } + if let Err(e) = set_release_request_status_with_history( + state, + &release_id, + ReleaseRequestStatus::Executing, + request_terminal, + &system_actor, + serde_json::json!({ "reason": "setup_dependency_missing", "detail": reason }), + ) + .await + { + error!(execution_id, "Failed to settle stranded request: {e}"); + } + + if let Ok(Some(release)) = state.store.get_release_request_by_id(&release_id).await { + if let Some(deployment_id) = &release.request_snapshot.publish_deployment_id { + let _ = state + .store + .update_deployment_status(deployment_id, DeploymentStatus::Failed) + .await; + } + } +} + async fn set_release_request_status_with_history( state: &AppState, release_request_id: &str, @@ -711,7 +831,12 @@ async fn run_claimed_release_execution(state: AppState, execution_id: &str) -> a match execution.status { ReleaseExecutionStatus::RollbackInProgress => { - let ctx = build_rollback_context(state, execution).await?; + let ctx = match build_rollback_context(state.clone(), execution).await { + Ok(ctx) => ctx, + Err(err) => { + return settle_or_retry_setup_failure(&state, execution_id, true, err).await + } + }; run_rollback_deployment(ctx).await; } ReleaseExecutionStatus::Preparing @@ -722,7 +847,12 @@ async fn run_claimed_release_execution(state: AppState, execution_id: &str) -> a { return Ok(()); } - let ctx = build_rolling_context(state, execution).await?; + let ctx = match build_rolling_context(state.clone(), execution).await { + Ok(ctx) => ctx, + Err(err) => { + return settle_or_retry_setup_failure(&state, execution_id, false, err).await + } + }; run_rolling_deployment(ctx).await; } _ => {} @@ -780,12 +910,16 @@ async fn build_rolling_context( .store .get_release_request_by_id(&execution.request_id) .await? - .ok_or_else(|| anyhow::anyhow!("Release request not found"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing("Release request not found".into())) + })?; let env = state .store .get_environment(&release.project_id, &release.environment_id) .await? - .ok_or_else(|| anyhow::anyhow!("Environment not found"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing("Environment not found".into())) + })?; execution.instances.sort_by(|left, right| { left.batch_index @@ -799,7 +933,12 @@ async fn build_rolling_context( .store .get_server(&instance.instance_id) .await? - .ok_or_else(|| anyhow::anyhow!("Bound server {} not found", instance.instance_id))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing(format!( + "Bound server {} not found", + instance.instance_id + ))) + })?; bound_servers.push(server); } @@ -810,7 +949,9 @@ async fn build_rolling_context( .store .get_draft_ruleset(&release.project_id, &release.ruleset_name) .await? - .ok_or_else(|| anyhow::anyhow!("Draft ruleset not found"))? + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing("Draft ruleset not found".into())) + })? .draft }; @@ -851,19 +992,27 @@ async fn build_rollback_context( .store .get_release_request_by_id(&execution.request_id) .await? - .ok_or_else(|| anyhow::anyhow!("Release request not found"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing("Release request not found".into())) + })?; let env = state .store .get_environment(&release.project_id, &release.environment_id) .await? - .ok_or_else(|| anyhow::anyhow!("Environment not found"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing("Environment not found".into())) + })?; let rollback_version = execution .instances .first() .map(|instance| instance.target_version.clone()) .or_else(|| release.version_diff.rollback_version.clone()) .or_else(|| release.rollback_version.clone()) - .ok_or_else(|| anyhow::anyhow!("Release request has no rollback version"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing( + "Release request has no rollback version".into(), + )) + })?; let rollback_deployment = state .store .list_deployments(&release.project_id, Some(&release.ruleset_name), 50) @@ -874,7 +1023,11 @@ async fn build_rollback_context( && deployment.version == rollback_version && deployment.status == DeploymentStatus::Success }) - .ok_or_else(|| anyhow::anyhow!("Rollback deployment snapshot not found"))?; + .ok_or_else(|| { + anyhow::Error::new(SetupDependencyMissing( + "Rollback deployment snapshot not found".into(), + )) + })?; execution.instances.sort_by(|left, right| { left.batch_index @@ -2975,6 +3128,24 @@ mod tests { assert!(normalized["steps"].get("done").is_some()); } + #[test] + fn setup_dependency_missing_is_distinguished_from_transient_errors() { + // The stranded-rollout fix hinges on telling a permanent missing dependency + // (settle the execution) apart from a transient store error (retry). Lock in + // that discrimination. + let permanent = anyhow::Error::new(SetupDependencyMissing( + "Bound server srv_x not found".into(), + )); + assert!(permanent.downcast_ref::().is_some()); + + let transient = anyhow::anyhow!("connection reset by peer"); + assert!(transient.downcast_ref::().is_none()); + + // A wrapped sqlx error (the shape a real store blip takes) is also transient. + let db: anyhow::Error = anyhow::Error::new(sqlx::Error::PoolClosed); + assert!(db.downcast_ref::().is_none()); + } + #[test] fn compute_batch_size_handles_each_strategy() { // No instances always yields a single (no-op) batch. From c45be00c3de3e0123de45cbaccb2981c8ed925ae Mon Sep 17 00:00:00 2001 From: Pama-Lee Date: Thu, 9 Jul 2026 00:36:12 +0800 Subject: [PATCH 5/5] fix(platform): scope cross-org server reads and release-execution reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three handlers took an id from the path and fetched by that id alone, after only a coarse auth check — and server ids are derivable (sha256 of the engine URL), so the ids are guessable across orgs: - get_server / get_server_metrics / get_server_health ignored the caller's claims entirely. Any authenticated user could read another org's engine (url, version, capabilities) and, via metrics/health, trigger a control-plane NATS RPC to it. Now an org-owned server requires the caller to be a member of that org; global (unowned) servers stay readable, matching delete_server's gating. - get_release_execution_for_request / list_release_execution_events checked the permission on the path org/project but then fetched by the unscoped release_id / execution_id. An admin in their own org could read another org's execution state and event stream with a guessed id. Both now resolve the release through the org+project scoped getter (404 on mismatch), and the events handler additionally verifies the execution belongs to that release. Also scope the engine-proxy fallback: when a project's environment has no bound server (the auto-created production env has none, so this is common), the fallback pool is now restricted to the caller's org plus deliberately-global servers, never another org's engine — previously it picked the most-recently-registered server anywhere, routing org A's eval traffic and ruleset payloads to org B's engine. --- crates/ordo-platform/src/proxy.rs | 30 ++++++++++------- .../ordo-platform/src/release/executions.rs | 33 +++++++++++++++++-- crates/ordo-platform/src/server_registry.rs | 28 ++++++++++++++-- 3 files changed, 73 insertions(+), 18 deletions(-) diff --git a/crates/ordo-platform/src/proxy.rs b/crates/ordo-platform/src/proxy.rs index 371af419..16cf2679 100644 --- a/crates/ordo-platform/src/proxy.rs +++ b/crates/ordo-platform/src/proxy.rs @@ -178,13 +178,9 @@ fn is_write_method(method: &Method) -> bool { /// 2. If the default env has a canary config and `rand < canary_percentage`, route to the canary env. /// 3. Otherwise route to the default env's server. /// 4. Fall back to the platform's configured `engine_url`. -pub(crate) async fn resolve_engine_url( - state: &AppState, - project_id: &str, - _org_id: &str, -) -> String { +pub(crate) async fn resolve_engine_url(state: &AppState, project_id: &str, org_id: &str) -> String { let Ok(Some(prod_env)) = state.store.get_default_environment(project_id).await else { - return resolve_fallback_engine_url(state).await; + return resolve_fallback_engine_url(state, org_id).await; }; // Canary check @@ -213,16 +209,26 @@ pub(crate) async fn resolve_engine_url( return url; } - resolve_fallback_engine_url(state).await + resolve_fallback_engine_url(state, org_id).await } -async fn resolve_fallback_engine_url(state: &AppState) -> String { +/// Pick a fallback engine when a project's environment has no usable bound server. +/// The candidate pool is restricted to this org's own servers plus deliberately +/// global (unowned) ones — never another org's engine. Otherwise, since the +/// auto-created `production` env has no server and thus commonly hits this path, +/// org A's eval traffic (its tenant id and, for ruleset writes, the full payload) +/// could be shipped to whichever engine registered most recently anywhere — e.g. +/// org B's. The platform's configured `engine_url` remains the final fallback. +async fn resolve_fallback_engine_url(state: &AppState, org_id: &str) -> String { if let Ok(servers) = state.store.list_servers(None).await { if let Some(server) = servers.into_iter().find(|server| { - matches!( - server.status, - crate::models::ServerStatus::Online | crate::models::ServerStatus::Degraded - ) + let same_org_or_global = + server.org_id.as_deref() == Some(org_id) || server.org_id.is_none(); + same_org_or_global + && matches!( + server.status, + crate::models::ServerStatus::Online | crate::models::ServerStatus::Degraded + ) }) { return server.url; } diff --git a/crates/ordo-platform/src/release/executions.rs b/crates/ordo-platform/src/release/executions.rs index cc47c471..91a43876 100644 --- a/crates/ordo-platform/src/release/executions.rs +++ b/crates/ordo-platform/src/release/executions.rs @@ -422,7 +422,7 @@ async fn update_batch_schedule_with_history( pub async fn list_release_execution_events( State(state): State, Extension(claims): Extension, - Path((org_id, project_id, _release_id, execution_id)): Path<(String, String, String, String)>, + Path((org_id, project_id, release_id, execution_id)): Path<(String, String, String, String)>, ) -> ApiResult>> { require_project_permission( &state, @@ -433,9 +433,26 @@ pub async fn list_release_execution_events( ) .await?; + // The permission check only proves the caller may view releases in this + // org/project; it does NOT prove `release_id`/`execution_id` belong here. + // Scope both to the path so a foreign (guessable) id can't be read. + let release = state + .store + .get_release_request(&org_id, &project_id, &release_id) + .await + .map_err(PlatformError::Internal)? + .ok_or_else(|| PlatformError::not_found("Release request not found"))?; + let execution = state + .store + .get_release_execution(&execution_id) + .await + .map_err(PlatformError::Internal)? + .filter(|execution| execution.request_id == release.id) + .ok_or_else(|| PlatformError::not_found("Release execution not found"))?; + let events = state .store - .list_release_execution_events(&execution_id) + .list_release_execution_events(&execution.id) .await .map_err(PlatformError::Internal)?; Ok(Json(events)) @@ -455,9 +472,19 @@ pub async fn get_release_execution_for_request( ) .await?; + // Scope release_id to this org/project — the permission check alone doesn't + // prove the (guessable) id belongs here, so resolve it through the org+project + // scoped getter, which returns None (→ 404) for a foreign release. + let release = state + .store + .get_release_request(&org_id, &project_id, &release_id) + .await + .map_err(PlatformError::Internal)? + .ok_or_else(|| PlatformError::not_found("Release request not found"))?; + let item = state .store - .find_release_execution_by_request_id(&release_id) + .find_release_execution_by_request_id(&release.id) .await .map_err(PlatformError::Internal)?; Ok(Json(item)) diff --git a/crates/ordo-platform/src/server_registry.rs b/crates/ordo-platform/src/server_registry.rs index 8288b4cd..0ead2532 100644 --- a/crates/ordo-platform/src/server_registry.rs +++ b/crates/ordo-platform/src/server_registry.rs @@ -277,10 +277,29 @@ pub async fn delete_connect_token( } } +/// Authorize a read of `server` by `user_id`. `server_id` is derived from the +/// engine URL (`derive_server_id` = sha256 of the normalized URL), so it is +/// guessable — a bare `get_server(id)` with no org check leaks another org's +/// engine (url, version, capabilities) and, for the metrics/health variants, +/// lets any authenticated user trigger a control-plane RPC to it. An org-owned +/// server therefore requires the caller to be a member of that org; a global +/// (unowned) server stays readable by any authenticated user, matching how +/// `delete_server` gates only when an org owns the server. +async fn authorize_server_read( + state: &AppState, + server: &ServerNode, + user_id: &str, +) -> ApiResult<()> { + if let Some(org_id) = &server.org_id { + load_org_and_check_role(state, org_id, user_id, Role::Viewer).await?; + } + Ok(()) +} + /// GET /api/v1/servers/:id pub async fn get_server( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(id): Path, ) -> ApiResult> { let server = state @@ -289,6 +308,7 @@ pub async fn get_server( .await .map_err(PlatformError::Internal)? .ok_or_else(|| PlatformError::not_found("Server not found"))?; + authorize_server_read(&state, &server, &claims.sub).await?; Ok(Json(server.into())) } @@ -358,7 +378,7 @@ fn rpc_body_as_text(body: &serde_json::Value) -> String { /// GET /api/v1/servers/:id/metrics — request server Prometheus metrics over NATS pub async fn get_server_metrics( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(id): Path, ) -> ApiResult { let server = state @@ -367,6 +387,7 @@ pub async fn get_server_metrics( .await .map_err(PlatformError::Internal)? .ok_or_else(|| PlatformError::not_found("Server not found"))?; + authorize_server_read(&state, &server, &claims.sub).await?; let rpc = request_server_rpc(&state, &server, "metrics").await?; let status = @@ -384,7 +405,7 @@ pub async fn get_server_metrics( /// GET /api/v1/servers/:id/health — request server health over NATS pub async fn get_server_health( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(id): Path, ) -> ApiResult> { let server = state @@ -393,6 +414,7 @@ pub async fn get_server_health( .await .map_err(PlatformError::Internal)? .ok_or_else(|| PlatformError::not_found("Server not found"))?; + authorize_server_read(&state, &server, &claims.sub).await?; match request_server_rpc(&state, &server, "health").await { Ok(rpc) => {