fix(platform): re-inline sub-rules on redeploy and rollback (silent corruption)

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'
+    )
+);

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;

crates/ordo-platform/src/lib.rs

@@ -93,10 +93,7 @@ pub async fn connect_platform_store(config: &PlatformConfig) -> anyhow::Result<A
     Ok(Arc::new(PlatformStore::new(pool).await?))
 }
 
-pub async fn bootstrap_platform_store(
-    store: &Arc<PlatformStore>,
-    fail_active_executions: bool,
-) -> anyhow::Result<()> {
+pub async fn bootstrap_platform_store(store: &Arc<PlatformStore>) -> 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<PlatformStore>) -> 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<PlatformStore>) -> tokio::task::JoinHandle<()> {
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(60));

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?;

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;
         }