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' + ) +); 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/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 7210d53f..f3541002 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, @@ -27,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, @@ -292,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, @@ -303,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)) @@ -325,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)) @@ -391,6 +548,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 +612,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)) } @@ -679,7 +858,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 @@ -690,7 +874,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; } _ => {} @@ -748,12 +937,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 @@ -767,7 +960,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); } @@ -778,7 +976,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 }; @@ -819,19 +1019,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) @@ -842,7 +1050,31 @@ 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(), + )) + })?; + + // Re-inline sub-rules into the target snapshot before dispatching the rollback. + // A direct publish's deployment row stores the un-inlined authored draft, so + // rolling it back verbatim would ship a dangling SubRule reference (silently + // dropped at studio→engine conversion). Inlining is idempotent, so an + // already-inlined snapshot (governed releases) is untouched. A sub-rule that no + // longer exists makes the rollback unassemblable — a permanent failure — so map + // it to SetupDependencyMissing to settle as RollbackFailed instead of looping. + let snapshot = crate::ruleset_draft::inline_sub_rules_into_draft( + &state, + &release.org_id, + &release.project_id, + rollback_deployment.snapshot, + ) + .await + .map_err(|e| { + anyhow::Error::new(SetupDependencyMissing(format!( + "Rollback snapshot could not be assembled: {e}" + ))) + })?; execution.instances.sort_by(|left, right| { left.batch_index @@ -862,7 +1094,7 @@ async fn build_rollback_context( rollback_version, env, instances: execution.instances, - snapshot: rollback_deployment.snapshot, + snapshot, strategy: execution.strategy, actor: system_history_actor("release_worker"), release_note: Some(format!("Rollback for release {}", release_request_id)), @@ -2943,6 +3175,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. diff --git a/crates/ordo-platform/src/ruleset_draft.rs b/crates/ordo-platform/src/ruleset_draft.rs index 8a1acc93..af920f5b 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, @@ -687,6 +704,20 @@ pub async fn redeploy( .map_err(PlatformError::Internal)? .ok_or_else(|| PlatformError::not_found("Environment not found"))?; + // Re-inline sub-rules before rolling the snapshot out again. A direct publish + // stores the *authored* (un-inlined) draft in its deployment row, and older + // rows predate inlining entirely, so dispatching the stored snapshot verbatim + // would ship a ruleset whose SubRule steps reference a graph that isn't there + // (studio→engine conversion drops the dangling reference silently). Inlining is + // idempotent — an already-inlined graph is left untouched (see the + // `sub_rules.contains_key` guard in `inline_sub_rules_with_manifest`) — so this + // heals old/direct-publish rows and no-ops on already-inlined ones. Do it before + // creating the row so a now-missing sub-rule fails fast without orphaning a + // `Dispatched` deployment. + let inlined = + inline_sub_rules_into_draft(&state, &org_id, &project_id, original.snapshot.clone()) + .await?; + let dep_id = Uuid::new_v4().to_string(); let deployment = RulesetDeployment { id: dep_id.clone(), @@ -696,7 +727,7 @@ pub async fn redeploy( ruleset_name: ruleset_name.clone(), version: original.version.clone(), release_note: req.release_note.clone(), - snapshot: original.snapshot.clone(), + snapshot: inlined.clone(), deployed_at: Utc::now(), deployed_by: Some(claims.sub.clone()), status: DeploymentStatus::Dispatched, @@ -707,16 +738,16 @@ pub async fn redeploy( .await .map_err(PlatformError::Internal)?; - // The stored deployment snapshot already contains inlined sub-rules — roll it out - // through the release engine, exactly like a fresh publish (async, worker-settled). - // On enqueue failure, settle the `Dispatched` row to `Failed` so it doesn't orphan. + // Roll it out through the release engine, exactly like a fresh publish (async, + // worker-settled). On enqueue failure, settle the `Dispatched` row to `Failed` + // so it doesn't orphan. if let Err(e) = enqueue_publish_release( &state, &org_id, &project_id, &ruleset_name, &original.version, - original.snapshot.clone(), + inlined, &dep_id, &env, &claims, 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) => { diff --git a/crates/ordo-platform/src/store/releases.rs b/crates/ordo-platform/src/store/releases.rs index 77dd910c..afb99b94 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, @@ -989,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())