fix(cluster): stop cluster-apply crash-loops from the recovery-sidecar trap

crates/omnigraph-cluster/Cargo.toml

@@ -10,8 +10,8 @@ documentation = "https://docs.rs/omnigraph-cluster"
 
 [features]
 # Fault-injection hooks for the apply protocol (crash-mid-apply, CAS-race
-# tests). Deliberately does NOT enable omnigraph/failpoints.
-failpoints = ["dep:fail", "fail/failpoints"]
+# tests), including cluster/engine boundary failures.
+failpoints = ["dep:fail", "fail/failpoints", "omnigraph/failpoints"]
 
 [dependencies]
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.7.0" }

crates/omnigraph-cluster/src/diff.rs

@@ -18,6 +18,7 @@ pub(crate) fn diff_resources(
                 disposition: None,
                 reason: None,
                 binding_change: false,
+                metadata_change: None,
                 migration: None,
             }),
             Some(before) if before != after => changes.push(PlanChange {
@@ -28,6 +29,7 @@ pub(crate) fn diff_resources(
                 disposition: None,
                 reason: None,
                 binding_change: false,
+                metadata_change: None,
                 migration: None,
             }),
             Some(_) => {}
@@ -43,6 +45,7 @@ pub(crate) fn diff_resources(
                 disposition: None,
                 reason: None,
                 binding_change: false,
+                metadata_change: None,
                 migration: None,
             });
         }
@@ -82,6 +85,47 @@ pub(crate) fn append_policy_binding_changes(
             disposition: None,
             reason: None,
             binding_change: true,
+            metadata_change: Some(PlanMetadataChange::PolicyBindings),
+            migration: None,
+        });
+    }
+    changes.sort_by(|a, b| a.resource.cmp(&b.resource));
+}
+
+/// Metadata-only embedding provider changes: the provider digest is unchanged
+/// but the applied state predates storing the profile body needed by
+/// config-free serving. This mirrors policy binding backfill instead of
+/// hiding a serving-time failure behind a no-op plan.
+pub(crate) fn append_embedding_profile_changes(
+    changes: &mut Vec<PlanChange>,
+    prior_state: Option<&ClusterState>,
+    desired: &DesiredCluster,
+) {
+    let Some(state) = prior_state else {
+        return; // no state: provider Creates carry profiles already
+    };
+    for (address, desired_profile) in &desired.embedding_providers {
+        if changes
+            .iter()
+            .any(|change| change.resource.as_str() == address.as_str())
+        {
+            continue; // content change already covers it
+        }
+        let Some(entry) = state.applied_revision.resources.get(address) else {
+            continue; // not applied yet: the Create covers it
+        };
+        if entry.embedding_profile.as_ref() == Some(desired_profile) {
+            continue;
+        }
+        changes.push(PlanChange {
+            resource: address.clone(),
+            operation: PlanOperation::Update,
+            before_digest: Some(entry.digest.clone()),
+            after_digest: Some(entry.digest.clone()),
+            disposition: None,
+            reason: None,
+            binding_change: false,
+            metadata_change: Some(PlanMetadataChange::EmbeddingProfile),
             migration: None,
         });
     }

crates/omnigraph-cluster/src/lib.rs

@@ -33,9 +33,9 @@ use config::{
     validate_id, validate_query_source,
 };
 use diff::{
-    FailedGraphOrigin, ResourceKind, append_policy_binding_changes, approved_resources,
-    classify_changes, compute_approvals, compute_blast_radius, demote_dependents_of_failed_graphs,
-    diff_resources, resource_kind,
+    FailedGraphOrigin, ResourceKind, append_embedding_profile_changes,
+    append_policy_binding_changes, approved_resources, classify_changes, compute_approvals,
+    compute_blast_radius, demote_dependents_of_failed_graphs, diff_resources, resource_kind,
 };
 pub use serve::{
     ServingGraph, ServingPolicy, ServingQuery, ServingSnapshot, cluster_graph_ids,
@@ -183,6 +183,7 @@ pub async fn plan_config_dir(config_dir: impl AsRef<Path>) -> PlanOutput {
     };
     if !has_errors(&diagnostics) {
         append_policy_binding_changes(&mut changes, prior_state.as_ref(), &desired);
+        append_embedding_profile_changes(&mut changes, prior_state.as_ref(), &desired);
     }
     // Plan previews dispositions without sweeping; a pending recovery is
     // surfaced as the cluster_recovery_pending warning above instead.
@@ -404,6 +405,7 @@ pub async fn apply_config_dir_with_options(
     let prior_resources = state_resource_digests(&state);
     let mut changes = diff_resources(&prior_resources, &desired.resource_digests);
     append_policy_binding_changes(&mut changes, Some(&state), &desired);
+    append_embedding_profile_changes(&mut changes, Some(&state), &desired);
     let approval_artifacts = backend.list_approval_artifacts(&mut diagnostics).await;
     let approved = approved_resources(
         &approval_artifacts,
@@ -639,42 +641,9 @@ pub async fn apply_config_dir_with_options(
                 continue;
             }
         };
-        let observed_manifest_version = match db.snapshot_of(ReadTarget::branch("main")).await {
-            Ok(snapshot) => Some(snapshot.version()),
-            Err(_) => None,
-        };
-        let mut sidecar = RecoverySidecar {
-            schema_version: 1,
-            operation_id: Ulid::new().to_string(),
-            started_at: now_rfc3339(),
-            actor: options.actor.clone(),
-            kind: RecoverySidecarKind::SchemaApply,
-            graph_id: graph_id.clone(),
-            graph_uri: graph_uri.clone(),
-            observed_manifest_version,
-            expected_manifest_version: None,
-            desired_schema_digest: desired_graph.schema_digest.clone(),
-            state_cas_base: expected_cas.clone(),
-            approval_id: None,
-        };
-        let sidecar_path = match backend.write_recovery_sidecar(&sidecar).await {
-            Ok(path) => path,
-            Err(diagnostic) => {
-                diagnostics.push(diagnostic);
-                failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
-                graph_moving_aborted = true;
-                continue;
-            }
-        };
-        if let Err(diagnostic) = failpoints::maybe_fail("cluster_apply.before_schema_apply") {
-            // Simulated crash before the engine call: the sidecar stays; the
-            // sweep retires it next run (ledger still consistent with live).
-            diagnostics.push(diagnostic);
-            failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
-            graph_moving_aborted = true;
-            continue;
-        }
-        // Re-read + digest-verify the desired schema source under the lock.
+        // Re-read + digest-verify the desired schema source before the
+        // cluster sidecar exists. Parser/planner rejections cannot have
+        // moved graph state, so they must not leave recovery work behind.
         let schema_source = source_paths
             .get(schema_address(graph_id).as_str())
             .ok_or_else(|| {
@@ -708,12 +677,64 @@ pub async fn apply_config_dir_with_options(
             Ok(source) => source,
             Err(diagnostic) => {
                 diagnostics.push(diagnostic);
-                backend.delete_object(&sidecar_path).await; // nothing moved
                 failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
                 graph_moving_aborted = true;
                 continue;
             }
         };
+        if let Err(err) = db
+            .preview_schema_apply_with_options(&schema_source, SchemaApplyOptions::default())
+            .await
+        {
+            diagnostics.push(Diagnostic::error(
+                "schema_apply_failed",
+                schema_address(graph_id),
+                format!("schema apply is not supported on '{graph_uri}': {err}"),
+            ));
+            failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
+            graph_moving_aborted = true;
+            continue;
+        }
+        let observed_manifest_version = match db.snapshot_of(ReadTarget::branch("main")).await {
+            Ok(snapshot) => Some(snapshot.version()),
+            Err(_) => None,
+        };
+        let recorded_schema_digest = state
+            .applied_revision
+            .resources
+            .get(&schema_address(graph_id))
+            .map(|entry| entry.digest.clone());
+        let mut sidecar = RecoverySidecar {
+            schema_version: 1,
+            operation_id: Ulid::new().to_string(),
+            started_at: now_rfc3339(),
+            actor: options.actor.clone(),
+            kind: RecoverySidecarKind::SchemaApply,
+            graph_id: graph_id.clone(),
+            graph_uri: graph_uri.clone(),
+            observed_manifest_version,
+            expected_manifest_version: None,
+            desired_schema_digest: desired_graph.schema_digest.clone(),
+            state_cas_base: expected_cas.clone(),
+            approval_id: None,
+        };
+        let sidecar_path = match backend.write_recovery_sidecar(&sidecar).await {
+            Ok(path) => path,
+            Err(diagnostic) => {
+                diagnostics.push(diagnostic);
+                failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
+                graph_moving_aborted = true;
+                continue;
+            }
+        };
+        if let Err(diagnostic) = failpoints::maybe_fail("cluster_apply.before_schema_apply") {
+            // Simulated crash before the engine call: the sidecar stays; the
+            // sweep retires it next run (ledger still consistent with live).
+            diagnostics.push(diagnostic);
+            failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
+            graph_moving_aborted = true;
+            continue;
+        }
         // Soft drops only: allow_data_loss stays false until the approval
         // artifacts of stage 4C exist (RFC-004 §D4).
         match db
@@ -736,8 +757,15 @@ pub async fn apply_config_dir_with_options(
                     schema_address(graph_id),
                     format!("schema apply failed on '{graph_uri}': {err}"),
                 ));
-                // Sidecar stays; the sweep retires it (live digest unchanged
-                // == ledger consistent) or flags real movement.
+                if live_schema_matches_recorded_digest(
+                    &graph_uri,
+                    recorded_schema_digest.as_deref(),
+                    observed_manifest_version,
+                )
+                .await
+                {
+                    backend.delete_object(&sidecar_path).await;
+                }
                 failed_graphs.insert(graph_id.clone(), FailedGraphOrigin::SchemaApply);
                 graph_moving_aborted = true;
                 continue;
@@ -1022,6 +1050,7 @@ pub async fn apply_config_dir_with_options(
         &desired.resource_digests,
     );
     append_policy_binding_changes(&mut residual, Some(&new_state), &desired);
+    append_embedding_profile_changes(&mut residual, Some(&new_state), &desired);
     let converged = residual.is_empty();
     if converged {
         new_state.applied_revision.config_digest = Some(desired.config_digest.clone());
@@ -1939,6 +1968,29 @@ fn embedding_provider_digest(profile: &EmbeddingProviderConfig) -> String {
     sha256_hex(input.as_bytes())
 }
 
+async fn live_schema_matches_recorded_digest(
+    graph_uri: &str,
+    recorded_schema_digest: Option<&str>,
+    observed_manifest_version: Option<u64>,
+) -> bool {
+    let Some(recorded_schema_digest) = recorded_schema_digest else {
+        return false;
+    };
+    let Some(observed_manifest_version) = observed_manifest_version else {
+        return false;
+    };
+    let Ok(db) = Omnigraph::open_read_only(graph_uri).await else {
+        return false;
+    };
+    let Ok(snapshot) = db.snapshot_of(ReadTarget::branch("main")).await else {
+        return false;
+    };
+    if snapshot.version() != observed_manifest_version {
+        return false;
+    }
+    sha256_hex(db.schema_source().as_bytes()) == recorded_schema_digest
+}
+
 fn desired_config_digest(
     raw: &RawClusterConfig,
     resource_digests: &BTreeMap<String, String>,