fix(cli): wait for accepted configuration previews

.github/workflows/tests-slow-nightly.yml

@@ -85,7 +85,7 @@ jobs:
             cli_cluster_e2e::cluster_e2e_live_apply_changes_schema_queries_and_adds_graph_without_restart \
             cli_cluster::core_live_apply_captures_server_file_root_without_local_storage_access \
             cli_cluster::managed_lifecycle_wait_reports_tombstone_and_checks_every_poll_identity \
-            cli_cluster::managed_apply_polls_original_delivery_without_resubmission_or_cancellation \
+            cli_cluster::managed_plan_and_apply_poll_original_identity_without_resubmission_or_cancellation \
             cli_cluster::managed_native_outcomes_preserve_activation_archive_and_uncertainty \
             cli_cluster::managed_poll_identity_mismatch_retains_original_delivery_and_native_ids \
             cli_cluster::managed_lifecycle_bad_acceptance_and_deadline_keep_recovery_identity \

changelog.d/managed-preview-polling.fixed.md

@@ -0,0 +1 @@
+- Managed plans now poll the original preview until ready. Timeouts retain its ID, revision and key; `--no-wait` returns on acceptance without cancelling work.

crates/omnigraph-cli/src/commands/cluster/args.rs

@@ -225,7 +225,7 @@ pub(crate) enum ClusterMembersCommand {
 
 #[derive(Debug, Default, Args)]
 pub(crate) struct ClusterRunArgs {
-    /// Return without waiting for completion. Managed apply acknowledges durable delivery reservation only.
+    /// Return without waiting for completion. Managed plan/apply acknowledge durable reservation only.
     #[arg(long)]
     pub(crate) no_wait: bool,
     /// For served apply or managed operations, bound the local wait in seconds (default 300, maximum 3600).

crates/omnigraph-cli/src/managed.rs

@@ -129,9 +129,11 @@ impl Output {
                     self.body["type"].as_str().unwrap_or("error")
                 )?;
                 for field in [
+                    "accepted_preview",
                     "accepted_deployment",
                     "requested_deployment_id",
                     "submission",
+                    "idempotency_key",
                 ] {
                     if let Some(value) = self.body.get(field) {
                         writeln!(stderr, "{field}: {value}")?;
@@ -140,6 +142,9 @@ impl Output {
                 if self.body["type"] == "output_delivery_failed" {
                     let recovery = recovery_evidence(&self.body);
                     for field in ["accepted_operation", "idempotency_key", "original_error"] {
+                        if field == "idempotency_key" && self.body.get(field).is_some() {
+                            continue;
+                        }
                         if let Some(value) = recovery.get(field) {
                             writeln!(stderr, "{field}: {value}")?;
                         }
@@ -192,6 +197,7 @@ fn accepted_recovery_output(body: &Value, recovery: Option<&Value>) -> Option<Va
     let mut output = Failure::output_delivery().body;
     let mut accepted = false;
     for (name, fields) in [
+        ("accepted_preview", &["preview_id", "revision"][..]),
         (
             "accepted_deployment",
             &["deployment_id", "native_deployment_id", "preview_id"][..],
@@ -253,6 +259,7 @@ fn recovery_evidence(body: &Value) -> Value {
         "title",
         "detail",
         "accepted_operation",
+        "accepted_preview",
         "accepted_deployment",
         "requested_deployment_id",
         "operation_id",

crates/omnigraph-cli/src/managed/deployment.rs

@@ -1,6 +1,8 @@
 //! Native managed previews and delivery: submit once, observe the original IDs.
 use super::*;
 
+mod preview;
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 struct Identity {
     deployment_id: String,
@@ -341,7 +343,7 @@ pub(super) async fn dispatch(
                 ));
             }
             let request = json!({"revision":revision});
-            let (body, key) = submit(
+            let submission = submit(
                 api,
                 &format!("{base}/plans"),
                 &request,
@@ -350,9 +352,7 @@ pub(super) async fn dispatch(
                 stderr,
             )
             .await?;
-            let exit = preview_exit(&body, context, &revision)
-                .map_err(|error| retain_submission(error, &key, &request))?;
-            Ok((body, exit))
+            preview::finish(api, context, submission, &revision, run, deadline, stderr).await
         }
         ClusterCommand::Apply { plan, run, .. } => {
             let plan = plan.as_deref().ok_or_else(|| {

crates/omnigraph-cli/src/managed/deployment/preview.rs

@@ -0,0 +1,197 @@
+//! Observe one accepted preview without submitting a replacement.
+use super::*;
+
+#[derive(Debug, PartialEq, Eq, Serialize)]
+struct Preview {
+    preview_id: String,
+    revision: String,
+    first_attempt_deadline: Option<String>,
+}
+
+impl Preview {
+    fn read(body: &Value, context: &Context, revision: &str) -> Result<Self> {
+        preview_exit(body, context, revision)?;
+        if !body["data"]["failure_code"].is_null()
+            && body["data"]["failure_code"].as_str().is_none_or(|code| {
+                code.is_empty()
+                    || code.len() > 128
+                    || !code.bytes().all(|byte| byte.is_ascii_graphic())
+            })
+        {
+            return Err(Failure::protocol());
+        }
+        let deadline = match &body["data"]["first_attempt_deadline"] {
+            Value::Null if body["data"]["state"] != "capturing" => None,
+            Value::String(value) => {
+                provider_timestamp(value)?;
+                Some(value.clone())
+            }
+            _ => return Err(Failure::protocol()),
+        };
+        Ok(Self {
+            preview_id: body["data"]["preview_id"]
+                .as_str()
+                .ok_or_else(Failure::protocol)?
+                .to_owned(),
+            revision: revision.to_owned(),
+            first_attempt_deadline: deadline,
+        })
+    }
+
+    fn verify(&self, body: &Value, context: &Context) -> Result<()> {
+        if Self::read(body, context, &self.revision)? != *self {
+            return Err(Failure::refused(
+                "context_mismatch",
+                "preview observation changed the original ID, revision or deadline",
+            ));
+        }
+        Ok(())
+    }
+
+    fn retain(&self, mut error: Failure, key: &str) -> Failure {
+        error.body["accepted_preview"] = serde_json::to_value(self).expect("preview identity");
+        error.body["idempotency_key"] = json!(key);
+        error
+    }
+
+    fn diagnostic(
+        &self,
+        stderr: &mut dyn Write,
+        key: &str,
+        message: std::fmt::Arguments<'_>,
+    ) -> Result<()> {
+        write_diagnostic(stderr, message).map_err(|error| self.retain(error, key))
+    }
+
+    fn timed_out(
+        &self,
+        mut body: Value,
+        key: &str,
+        stderr: &mut dyn Write,
+    ) -> Result<(Value, i32)> {
+        self.diagnostic(stderr, key, format_args!(
+            "local wait deadline reached; preview {} remains accepted; repeat `cluster plan --rev {} --idempotency-key KEY --target NAME` with the original key retained in the response; no work was cancelled or resubmitted\n",
+            self.preview_id, self.revision
+        ))?;
+        body["idempotency_key"] = json!(key);
+        Ok((body, 5))
+    }
+}
+
+pub(super) async fn finish(
+    api: &Api,
+    context: &Context,
+    submission: (Value, String),
+    revision: &str,
+    run: &ClusterRunArgs,
+    deadline: Instant,
+    stderr: &mut dyn Write,
+) -> Result<(Value, i32)> {
+    let (mut body, key) = submission;
+    let key = key.as_str();
+    let preview = Preview::read(&body, context, revision)
+        .map_err(|error| retain_submission(error, key, &json!({"revision":revision})))?;
+    if run.no_wait {
+        let exit = preview_exit(&body, context, revision)?;
+        if exit == 5 || (exit == 0 && !body["data"]["native_plan"].is_object()) {
+            body["idempotency_key"] = json!(key);
+        }
+        return Ok((body, if exit == 5 { 0 } else { exit }));
+    }
+    let path = format!(
+        "/v1/clusters/{}/plans/{}",
+        context.cluster, preview.preview_id
+    );
+    let mut observed = false;
+    let mut attempted_read = false;
+    loop {
+        preview
+            .verify(&body, context)
+            .map_err(|error| preview.retain(error, key))?;
+        let exit =
+            preview_exit(&body, context, revision).map_err(|error| preview.retain(error, key))?;
+        if exit == 1 {
+            let state = body["data"]["state"]
+                .as_str()
+                .ok_or_else(Failure::protocol)?;
+            let code = body["data"]["failure_code"]
+                .as_str()
+                .unwrap_or("preview_failed");
+            preview.diagnostic(
+                stderr,
+                key,
+                format_args!(
+                    "preview {}: {} ({}); a fresh plan requires a new idempotency key\n",
+                    preview.preview_id, state, code
+                ),
+            )?;
+            return Ok((body, exit));
+        }
+        if exit == 0 && body["data"]["native_plan"].is_object() {
+            return Ok((body, 0));
+        }
+        if exit == 0 && observed {
+            return Err(preview.retain(Failure::protocol(), key));
+        }
+        // A ready idempotent POST contains metadata only. Fetch its authorized
+        // plan immediately; pending previews use the normal bounded poll rate.
+        let next = if exit == 0 && !attempted_read {
+            Instant::now()
+        } else {
+            Instant::now() + POLL_INTERVAL
+        };
+        if next >= deadline {
+            tokio::time::sleep_until(deadline).await;
+            return preview.timed_out(body, key, stderr);
+        }
+        preview.diagnostic(
+            stderr,
+            key,
+            format_args!(
+                "preview {} at revision {}: {}\n",
+                preview.preview_id, preview.revision, body["data"]["state"]
+            ),
+        )?;
+        tokio::time::sleep_until(next).await;
+        attempted_read = true;
+        match tokio::time::timeout_at(deadline, api.raw(Method::GET, &path, None, None)).await {
+            Ok(Ok(response)) if response.status.is_success() => {
+                body = response.body;
+                observed = true;
+            }
+            Ok(Ok(response))
+                if response.status.is_server_error()
+                    || response.status == StatusCode::TOO_MANY_REQUESTS =>
+            {
+                preview.diagnostic(stderr, key, format_args!(
+                    "preview observation unavailable; retain the original ID and continue bounded read-only lookup\n"
+                ))?;
+            }
+            Ok(Ok(response)) => {
+                return Err(preview.retain(
+                    Failure {
+                        body: response.body,
+                        response_status: Some(response.status),
+                        exit: if response.status.is_client_error() {
+                            2
+                        } else {
+                            1
+                        },
+                    },
+                    key,
+                ));
+            }
+            Ok(Err(error)) if error.body["type"] == "transport_failed" => {
+                preview.diagnostic(
+                    stderr,
+                    key,
+                    format_args!(
+                        "preview observation transport unavailable; retain the original ID\n"
+                    ),
+                )?;
+            }
+            Ok(Err(error)) => return Err(preview.retain(error, key)),
+            Err(_) => return preview.timed_out(body, key, stderr),
+        }
+    }
+}

crates/omnigraph-cli/src/managed/tests.rs

@@ -201,45 +201,56 @@ fn managed_output_preserves_bytes_and_recovery_details_on_delivery_failure() {
         }
     }
 
-    // A lifecycle identity receives the same narrow fallback. Extra fields and
+    // Accepted identities receive the same narrow fallback. Extra fields and
     // reflected error text must never turn recovery into source/secret export.
-    let identity = json!({"cluster_id":"cluster-original","incarnation":"incarnation-original","operation_id":"operation-original","kind":"create"});
-    for fault in [OutputFault::PartialWrite, OutputFault::Flush] {
-        let mut body = json!({"type":"output_delivery_failed","detail":"private-error-content","accepted_operation":identity,"submission":{"idempotency_key":"original-key","request":{"source":"private-request-content"}},"access_token":"private-token-extra","data":{"files":{"secret.pg":"private-source-content"}}});
-        body["accepted_operation"]["token"] = json!("private-identity-extra");
-        let mut stdout = Vec::new();
-        assert_eq!(
-            Output::from_result(
-                Err(Failure {
-                    body,
-                    exit: 1,
-                    response_status: None
-                }),
-                false,
+    for (field, identity) in [
+        (
+            "accepted_operation",
+            json!({"cluster_id":"cluster-original","incarnation":"incarnation-original","operation_id":"operation-original","kind":"create"}),
+        ),
+        (
+            "accepted_preview",
+            json!({"preview_id":"preview-original","revision":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}),
+        ),
+    ] {
+        for fault in [OutputFault::PartialWrite, OutputFault::Flush] {
+            let mut body = json!({"type":"output_delivery_failed","detail":"private-error-content","submission":{"idempotency_key":"original-key","request":{"source":"private-request-content"}},"access_token":"private-token-extra","data":{"files":{"secret.pg":"private-source-content"}}});
+            body[field] = identity.clone();
+            body[field]["token"] = json!("private-identity-extra");
+            let mut stdout = Vec::new();
+            assert_eq!(
+                Output::from_result(
+                    Err(Failure {
+                        body,
+                        exit: 1,
+                        response_status: None
+                    }),
+                    false,
+                    1
+                )
+                .emit(&mut stdout, &mut FailedOutput::new(fault, 0))
+                .unwrap(),
                 1
+            );
+            let recovery: Value = serde_json::from_slice(&stdout).unwrap();
+            assert_eq!(recovery[field], identity);
+            assert_eq!(recovery["idempotency_key"], "original-key");
+            assert!(!String::from_utf8(stdout).unwrap().contains("private-"));
+
+            let mut stdout = Vec::new();
+            let error = Output::from_result(
+                Err(Failure::refused(
+                    "invalid_input",
+                    "no operation was accepted",
+                )),
+                false,
+                2,
             )
             .emit(&mut stdout, &mut FailedOutput::new(fault, 0))
-            .unwrap(),
-            1
-        );
-        let recovery: Value = serde_json::from_slice(&stdout).unwrap();
-        assert_eq!(recovery["accepted_operation"], identity);
-        assert_eq!(recovery["idempotency_key"], "original-key");
-        assert!(!String::from_utf8(stdout).unwrap().contains("private-"));
-
-        let mut stdout = Vec::new();
-        let error = Output::from_result(
-            Err(Failure::refused(
-                "invalid_input",
-                "no operation was accepted",
-            )),
-            false,
-            2,
-        )
-        .emit(&mut stdout, &mut FailedOutput::new(fault, 0))
-        .unwrap_err();
-        assert!(stdout.is_empty());
-        assert!(format!("{error:#}").contains("invalid_input"));
+            .unwrap_err();
+            assert!(stdout.is_empty());
+            assert!(format!("{error:#}").contains("invalid_input"));
+        }
     }
 }