diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index a8562c0..21d5bf1 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -1157,6 +1157,19 @@ impl ApplyService { .into(), )); } + // Duplicate workflow names collide in the reconcile diff (keyed by + // `lower(name)`), silently dropping all but one. Reject them up front — + // case-insensitively, matching the diff key and the DB's partial-unique + // index on `(app_id, lower(name))`. + let mut seen_workflows: HashSet = HashSet::new(); + for w in &bundle.workflows { + if !seen_workflows.insert(w.name.to_lowercase()) { + return Err(ApplyError::Invalid(format!( + "duplicate workflow name `{}` (names are case-insensitive)", + w.name + ))); + } + } for w in &bundle.workflows { validate_workflow_definition(&w.name, &w.definition).map_err(ApplyError::Invalid)?; } @@ -4233,12 +4246,14 @@ fn validate_workflow_definition( ) })?; } - // DAG acyclicity via Kahn topological sort. + // DAG acyclicity via Kahn topological sort. Count DISTINCT deps: a step that + // lists the same dependency twice inflates its in-degree past what the + // single-decrement in the drain loop below can undo, which would otherwise + // report a spurious cycle for a valid DAG. let mut indeg: Map<&str, usize> = def.steps.iter().map(|s| (s.name.as_str(), 0)).collect(); for s in &def.steps { - for _dep in &s.depends_on { - *indeg.get_mut(s.name.as_str()).unwrap() += 1; - } + let distinct: BTreeSet<&str> = s.depends_on.iter().map(String::as_str).collect(); + *indeg.get_mut(s.name.as_str()).unwrap() += distinct.len(); } let mut queue: Vec<&str> = indeg .iter() @@ -6730,6 +6745,17 @@ mod tests { .contains("depends on itself")); } + #[test] + fn workflow_validation_tolerates_duplicate_depends_on() { + // A step listing the same dep twice is a valid DAG — the topo sort must + // not report a spurious cycle from the double-counted in-degree. + let dup_dep = wf_def(serde_json::json!([ + { "name": "a", "function": "fa" }, + { "name": "b", "function": "fb", "depends_on": ["a", "a"] }, + ])); + assert!(validate_workflow_definition("w", &dup_dep).is_ok()); + } + #[test] fn workflow_validation_detects_cycles() { let cyclic = wf_def(serde_json::json!([ diff --git a/crates/manager-core/src/workflow_expr.rs b/crates/manager-core/src/workflow_expr.rs index e7830aa..71275a6 100644 --- a/crates/manager-core/src/workflow_expr.rs +++ b/crates/manager-core/src/workflow_expr.rs @@ -11,7 +11,15 @@ //! (`== != < > <= >=`), primary (`( … )`, `exists `, literal, ``). //! Literals: numbers, `'…'`/`"…"` strings, `true`, `false`, `null`. //! A bare reference is truthy per [`is_truthy`]; a reference that doesn't -//! resolve is `null` (falsy) at run time — apply-time [`validate`] guards typos. +//! resolve is `null` (falsy) at run time. This is a DELIBERATE asymmetry with +//! [`workflow_template`], which HARD-FAILS a missing ref: a `when` is a +//! predicate (an absent value is "condition not met"), whereas an input +//! template must not silently substitute null. Apply-time [`validate`] guards +//! the *shape* of a ref (the root `input`/`steps..output` prefix and that +//! `` is a declared step), but it cannot check a deeper JSON path against +//! a runtime value — so a typo *past* the step name (e.g. `steps.a.output.typ` +//! for `…typo`) is not caught at apply and silently makes the step skip. Prefer +//! `exists ` to make "the value may be absent" explicit. //! //! [`workflow_template`]: crate::workflow_template diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs index 3f32eb1..36a82cd 100644 --- a/crates/manager-core/src/workflow_repo.rs +++ b/crates/manager-core/src/workflow_repo.rs @@ -433,10 +433,27 @@ pub fn compute_advance( } } } + // A run reaches here as Succeeded even if some steps failed under + // `on_error = continue` (a hard `fail` short-circuits above). Those + // steps contribute no output, so surface them in the run's `error` + // field — otherwise a partial failure is invisible at the run level. + let failed: Vec<&str> = def + .steps + .iter() + .filter(|s| status_of(&s.name) == StepStatus::Failed) + .map(|s| s.name.as_str()) + .collect(); + let run_error = (!failed.is_empty()).then(|| { + format!( + "run succeeded with {} continued failure(s): {}", + failed.len(), + failed.join(", ") + ) + }); return AdvancePlan { promote, run_status: RunStatus::Succeeded, - run_error: None, + run_error, run_output: Some(serde_json::Value::Object(out)), }; } @@ -533,8 +550,12 @@ struct ClaimedRow { /// Atomically claim one `ready`, due step across all active runs — the queue /// competing-consumer lease (`FOR UPDATE SKIP LOCKED`), so parallel workers /// (and, cluster mode later, parallel nodes) never grab the same step. The -/// claimed step flips to `running` with a fresh `claim_token`; its run flips -/// `pending → running`. Returns `None` when nothing is claimable. +/// claimed step flips to `running` with a fresh `claim_token` in one atomic +/// statement. A *separate*, non-transactional follow-up statement then reflects +/// the run as `pending → running` — this is a promptness convenience, not part +/// of the claim's atomicity: a crash in between is harmless (the token-gated +/// advance, and `reclaim`, both re-derive run status). Returns `None` when +/// nothing is claimable. pub async fn claim_ready_step(pool: &PgPool) -> Result, WorkflowRepoError> { let token = Uuid::new_v4(); let row: Option = sqlx::query_as( @@ -762,10 +783,12 @@ pub async fn advance_run_tx( RunStatus::Succeeded => { sqlx::query( "UPDATE workflow_runs \ - SET status = 'succeeded', output = $2, finished_at = NOW() WHERE id = $1", + SET status = 'succeeded', output = $2, error = $3, finished_at = NOW() \ + WHERE id = $1", ) .bind(run_id.into_inner()) .bind(plan.run_output) + .bind(plan.run_error) .execute(&mut **tx) .await?; } @@ -796,6 +819,13 @@ pub async fn advance_run_tx( /// Periodic safety net: a step leased by a crashed worker (still `running` /// with a `claim_token` past the visibility timeout) is re-armed `ready` so /// another worker retries it. Only touches steps of still-active runs. +/// +/// The crashed attempt produced no outcome, so it must NOT consume the retry +/// budget: `claim_ready_step` incremented `attempt` up front, so reclaim +/// *decrements* it (floored at 0) — the re-claim will re-increment, leaving the +/// effective attempt count unchanged. `next_attempt_at` is cleared so the +/// reclaimed step is immediately eligible. +/// /// Returns the number of steps reclaimed. pub async fn reclaim_stale_steps( pool: &PgPool, @@ -803,7 +833,8 @@ pub async fn reclaim_stale_steps( ) -> Result { let res = sqlx::query( "UPDATE workflow_run_steps s \ - SET status = 'ready', claim_token = NULL, claimed_at = NULL, updated_at = NOW() \ + SET status = 'ready', claim_token = NULL, claimed_at = NULL, \ + attempt = GREATEST(s.attempt - 1, 0), next_attempt_at = NULL, updated_at = NOW() \ FROM workflow_runs r \ WHERE s.run_id = r.id \ AND s.claim_token IS NOT NULL \ @@ -934,16 +965,23 @@ impl TryFrom for WorkflowRunStep { } /// All steps of a run, ordered by name (for the admin/CLI surface + tests). +/// App-scoped at the query (JOIN `workflow_runs`) so isolation is enforced here +/// rather than relying on the caller having pre-verified run ownership — a +/// run under another app returns no rows. pub async fn list_run_steps( pool: &PgPool, + app_id: AppId, run_id: WorkflowRunId, ) -> Result, WorkflowRepoError> { let rows: Vec = sqlx::query_as( - "SELECT id, run_id, step_name, status, attempt, max_attempts, output, error, \ - child_run_id \ - FROM workflow_run_steps WHERE run_id = $1 ORDER BY step_name", + "SELECT s.id, s.run_id, s.step_name, s.status, s.attempt, s.max_attempts, s.output, \ + s.error, s.child_run_id \ + FROM workflow_run_steps s \ + JOIN workflow_runs r ON r.id = s.run_id \ + WHERE s.run_id = $1 AND r.app_id = $2 ORDER BY s.step_name", ) .bind(run_id.into_inner()) + .bind(app_id.into_inner()) .fetch_all(pool) .await?; rows.into_iter().map(TryInto::try_into).collect() @@ -1105,10 +1143,19 @@ pub async fn resume_finished_children(pool: &PgPool) -> Result ("succeeded", r.child_output.clone(), None), + "canceled" => ( + "failed", + None, + Some("sub-workflow was canceled".to_string()), + ), + _ => ( "failed", None, Some( @@ -1116,7 +1163,7 @@ pub async fn resume_finished_children(pool: &PgPool) -> Result Router { struct WorkflowDto { name: String, enabled: bool, + /// Step count (kept for the dashboard's summary column). steps: usize, + /// The full stored definition, so `pic pull` can round-trip the + /// `[[workflows]]` manifest block (without it, a pull-then-apply silently + /// deletes every workflow). + definition: WorkflowDefinition, } #[derive(Serialize)] @@ -147,6 +152,7 @@ async fn list_workflows( name: w.name, enabled: w.enabled, steps: w.definition.steps.len(), + definition: w.definition, }) .collect(), )) @@ -171,11 +177,16 @@ async fn start_run_handler( ))); } let input = body.map_or(Value::Null, |Json(b)| b.input); + // An operator-initiated run has no parent execution to correlate to, so it + // roots a fresh execution tree. Log the id against the principal so the + // run's step logs remain traceable to who launched it (the SDK path threads + // `cx.root_execution_id` instead). + let root_execution_id = Uuid::new_v4(); let new = NewWorkflowRun { workflow_id: wf.id, app_id, input, - root_execution_id: Uuid::new_v4(), + root_execution_id, workflow_depth: 0, parent_run_id: None, parent_step_id: None, @@ -183,6 +194,11 @@ async fn start_run_handler( let run_id = start_run(&s.pool, new, &wf.definition) .await .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + tracing::info!( + %app_id, workflow = %name, run_id = %run_id.into_inner(), %root_execution_id, + user_id = %principal.user_id.into_inner(), + "admin started a workflow run" + ); // Read it back for the response (status = pending until the orchestrator ticks). let run = get_run(&s.pool, app_id, run_id) .await @@ -219,7 +235,7 @@ async fn get_run_detail( .await .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? .ok_or_else(|| WorkflowsApiError::NotFound(format!("run {run_id}")))?; - let steps = list_run_steps(&s.pool, run.id) + let steps = list_run_steps(&s.pool, app_id, run.id) .await .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; // Load the definition for the per-step `depends_on` DAG edges (they live on diff --git a/crates/manager-core/tests/workflow_orchestrator.rs b/crates/manager-core/tests/workflow_orchestrator.rs index 02cea49..c3e67a3 100644 --- a/crates/manager-core/tests/workflow_orchestrator.rs +++ b/crates/manager-core/tests/workflow_orchestrator.rs @@ -180,7 +180,7 @@ async fn seeds_roots_ready_and_completes_linear_chain() { .expect("start"); // Root `a` starts ready; b, c pending. - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); let by = |n: &str| steps.iter().find(|s| s.step_name == n).unwrap().status; assert_eq!(by("a"), StepStatus::Ready); assert_eq!(by("b"), StepStatus::Pending); @@ -327,7 +327,7 @@ async fn on_error_fail_fails_the_run() { let r = get_run(&pool, app, run).await.unwrap().unwrap(); assert_eq!(r.status, RunStatus::Failed); assert!(r.error.unwrap().contains('a')); - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); let b = steps.iter().find(|s| s.step_name == "b").unwrap(); assert_eq!(b.status, StepStatus::Pending); // never promoted } @@ -392,7 +392,7 @@ async fn double_complete_is_idempotent() { .unwrap(); assert_eq!(second, AdvanceResult::Stale); - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); assert_eq!(steps[0].output, Some(json!("first")), "first write wins"); let r = get_run(&pool, app, run).await.unwrap().unwrap(); assert_eq!(r.status, RunStatus::Succeeded); @@ -433,6 +433,9 @@ async fn stale_lease_is_reclaimed() { .unwrap() .expect("re-claim after reclaim"); assert_eq!(c2.step_name, "a"); + // The crashed attempt produced no outcome, so it must not consume the retry + // budget: after claim(→1), reclaim(→0), re-claim(→1) the attempt is 1, not 2. + assert_eq!(c2.attempt, 1, "reclaim must not burn the retry budget"); complete_step_and_advance(&pool, &c2, ok(json!("done"))) .await .unwrap(); @@ -757,7 +760,7 @@ async fn when_false_skips_step() { let r = get_run(&pool, app, run).await.unwrap().unwrap(); assert_eq!(r.status, RunStatus::Succeeded); - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); let by = |n: &str| steps.iter().find(|s| s.step_name == n).unwrap().status; assert_eq!(by("a"), StepStatus::Succeeded); assert_eq!(by("b"), StepStatus::Skipped); @@ -830,7 +833,7 @@ async fn missing_input_ref_fails_the_step() { "missing input ref fails the run" ); assert!(executor.body_for("fn_b").is_none(), "b never executes"); - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); let bstep = steps.iter().find(|s| s.step_name == "b").unwrap(); assert_eq!(bstep.status, StepStatus::Failed); assert!(bstep.error.as_deref().unwrap_or("").contains("input")); @@ -887,7 +890,7 @@ async fn nested_sub_workflow_runs_and_output_flows_to_parent() { // The child's output flowed into the parent step, then into `after`'s input. assert_eq!(executor.body_for("fn_after"), Some(json!({ "got": 42 }))); // The parent step "call" carries the child run's output. - let steps = list_run_steps(&pool, run).await.unwrap(); + let steps = list_run_steps(&pool, app, run).await.unwrap(); let call = steps.iter().find(|s| s.step_name == "call").unwrap(); assert_eq!(call.status, StepStatus::Succeeded); assert_eq!(call.output, Some(json!({ "leaf": { "answer": 42 } }))); @@ -971,7 +974,7 @@ async fn workflow_service_start_seeds_a_run() { .expect("start"); let r = get_run(&pool, app, run_id).await.unwrap().unwrap(); assert_eq!(r.input, json!({ "k": "v" })); - let steps = list_run_steps(&pool, run_id).await.unwrap(); + let steps = list_run_steps(&pool, app, run_id).await.unwrap(); assert_eq!(steps.len(), 1); assert_eq!(steps[0].step_name, "only"); assert_eq!(steps[0].status, StepStatus::Ready); // a root diff --git a/crates/picloud-cli/src/client.rs b/crates/picloud-cli/src/client.rs index 5ac8f66..b732b16 100644 --- a/crates/picloud-cli/src/client.rs +++ b/crates/picloud-cli/src/client.rs @@ -2266,6 +2266,15 @@ pub struct WorkflowDto { pub name: String, pub enabled: bool, pub steps: usize, + /// The full stored definition — present so `pic pull` can rebuild the + /// `[[workflows]]` manifest block. `#[serde(default)]` tolerates an older + /// server that predates the field (the definition is then empty). + #[serde(default = "empty_workflow_definition")] + pub definition: picloud_shared::WorkflowDefinition, +} + +fn empty_workflow_definition() -> picloud_shared::WorkflowDefinition { + picloud_shared::WorkflowDefinition { steps: Vec::new() } } #[derive(Debug, Deserialize)] diff --git a/crates/picloud-cli/src/cmds/pull.rs b/crates/picloud-cli/src/cmds/pull.rs index 62df6ed..7993330 100644 --- a/crates/picloud-cli/src/cmds/pull.rs +++ b/crates/picloud-cli/src/cmds/pull.rs @@ -18,8 +18,9 @@ use crate::client::{Client, VarOwnerArg}; use crate::config; use crate::manifest::{ CronTriggerSpec, DocsTriggerSpec, FilesTriggerSpec, KvTriggerSpec, Manifest, ManifestApp, - ManifestRoute, ManifestScript, ManifestSecrets, ManifestTriggers, PubsubTriggerSpec, - QueueTriggerSpec, MANIFEST_FILE, + ManifestRoute, ManifestScript, ManifestSecrets, ManifestTriggers, ManifestWorkflow, + ManifestWorkflowRetry, ManifestWorkflowStep, PubsubTriggerSpec, QueueTriggerSpec, + MANIFEST_FILE, }; use crate::output::{KvBlock, OutputMode}; @@ -45,6 +46,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> let app = client.apps_get(app_ident).await?; let scripts = client.scripts_list_by_app(app_ident).await?; let triggers = client.triggers_list(app_ident).await?.triggers; + let workflows = client.workflows_list(app_ident).await?; let secrets = client .secrets_list(VarOwnerArg::App(app_ident), None) .await? @@ -251,7 +253,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> }, vars: manifest_vars, suppress: crate::manifest::ManifestSuppress::default(), - workflows: Vec::new(), + workflows: workflows.iter().map(wire_workflow_to_manifest).collect(), }; std::fs::write(&manifest_path, manifest.to_toml()?) @@ -264,7 +266,8 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> .field("scripts", manifest.scripts.len().to_string()) .field("routes", manifest.routes.len().to_string()) .field("triggers", trigger_count(&manifest.triggers).to_string()) - .field("secrets", manifest.secrets.names.len().to_string()); + .field("secrets", manifest.secrets.names.len().to_string()) + .field("workflows", manifest.workflows.len().to_string()); block.print(mode); Ok(()) } @@ -273,6 +276,70 @@ fn trigger_count(t: &ManifestTriggers) -> usize { t.kv.len() + t.docs.len() + t.files.len() + t.cron.len() + t.pubsub.len() + t.queue.len() } +/// Rebuild a `[[workflows]]` manifest block from the server's stored definition +/// (the inverse of `plan::workflow_to_wire`). Without this a `pic pull` would +/// drop the app's workflows, and a subsequent `pic apply --prune` would delete +/// them from the server. +fn wire_workflow_to_manifest(w: &crate::client::WorkflowDto) -> ManifestWorkflow { + ManifestWorkflow { + name: w.name.clone(), + enabled: w.enabled, + steps: w + .definition + .steps + .iter() + .map(wire_step_to_manifest) + .collect(), + } +} + +fn wire_step_to_manifest(s: &picloud_shared::WorkflowStepDef) -> ManifestWorkflowStep { + ManifestWorkflowStep { + name: s.name.clone(), + function: s.function.clone(), + workflow: s.workflow.clone(), + // JSON step input → TOML. A whole-null input is the "absent" encoding; + // anything else round-trips through serde. A value TOML can't represent + // (e.g. a nested null) is dropped with a warning rather than aborting + // the whole pull. + input: if s.input.is_null() { + None + } else { + match toml::Value::try_from(&s.input) { + Ok(tv) => Some(tv), + Err(e) => { + eprintln!( + "warning: workflow step `{}` input not representable in TOML ({e}); \ + omitting it from the manifest", + s.name + ); + None + } + } + }, + depends_on: s.depends_on.clone(), + when: s.when.clone(), + retry: s.retry.as_ref().map(|r| ManifestWorkflowRetry { + max_attempts: r.max_attempts, + // Exponential is the default → omit; likewise the default base_ms. + backoff: match r.backoff { + picloud_shared::WorkflowBackoff::Exponential => None, + other => Some(other), + }, + base_ms: if r.base_ms == 500 { + None + } else { + Some(r.base_ms) + }, + }), + // Fail is the default → omit; Continue is explicit. + on_error: match s.on_error { + picloud_shared::OnError::Fail => None, + picloud_shared::OnError::Continue => Some(picloud_shared::OnError::Continue), + }, + } +} + /// True if `name` is safe to use as a single path component in `scripts/`. /// Rejects empty/over-long names, path separators, `.`/`..`, leading dots, /// and any deceptive display character — a server-returned name is otherwise diff --git a/crates/picloud-cli/src/manifest.rs b/crates/picloud-cli/src/manifest.rs index 3cea8d2..43ae035 100644 --- a/crates/picloud-cli/src/manifest.rs +++ b/crates/picloud-cli/src/manifest.rs @@ -651,9 +651,11 @@ pub struct ManifestWorkflowStep { pub when: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub retry: Option, - /// `fail` (default) or `continue`. + /// `fail` (default) or `continue`. Typed against the shared enum so a typo + /// (`on_error = "retry"`) fails at TOML parse with an "unknown variant" + /// message pointing at this step, rather than an opaque server-side error. #[serde(default, skip_serializing_if = "Option::is_none")] - pub on_error: Option, + pub on_error: Option, } /// Per-step retry policy in the manifest (mirrors the server `WorkflowRetry`). @@ -661,9 +663,10 @@ pub struct ManifestWorkflowStep { #[serde(deny_unknown_fields)] pub struct ManifestWorkflowRetry { pub max_attempts: u32, - /// `constant` | `linear` | `exponential` (default). + /// `constant` | `linear` | `exponential` (default). Typed against the shared + /// enum so a bad value fails at TOML parse, not opaquely on the server. #[serde(default, skip_serializing_if = "Option::is_none")] - pub backoff: Option, + pub backoff: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub base_ms: Option, } diff --git a/crates/picloud-cli/tests/workflows.rs b/crates/picloud-cli/tests/workflows.rs index 63e4dd2..2edc862 100644 --- a/crates/picloud-cli/tests/workflows.rs +++ b/crates/picloud-cli/tests/workflows.rs @@ -154,6 +154,118 @@ fn workflow_with_a_cycle_is_rejected() { assert!(err.contains("cycle"), "expected a cycle error, got:\n{err}"); } +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn workflow_duplicate_name_is_rejected() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let slug = common::unique_slug("wfdup"); + common::pic_as(&env) + .args(["apps", "create", &slug]) + .assert() + .success(); + let _guard = AppGuard::new(&env.url, &env.token, &slug); + + let dir = manifest_dir(); + seed_scripts(&dir); + // Two workflows whose names collide case-insensitively — the reconcile keys + // by lower(name), so this must be rejected before it silently drops one. + let dup = format!( + "[app]\nslug = \"{slug}\"\nname = \"WF Dup\"\n\n{SCRIPTS_BLOCK}\ + [[workflows]]\nname = \"Pipe\"\n\n\ + [[workflows.steps]]\nname = \"a\"\nfunction = \"validate\"\n\n\ + [[workflows]]\nname = \"pipe\"\n\n\ + [[workflows.steps]]\nname = \"a\"\nfunction = \"charge\"\n" + ); + let path = dir.path().join("picloud.toml"); + fs::write(&path, &dup).unwrap(); + + let out = common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .output() + .expect("apply"); + assert!(!out.status.success(), "duplicate names should be rejected"); + let err = String::from_utf8_lossy(&out.stderr); + assert!( + err.contains("duplicate workflow"), + "expected a duplicate-workflow error, got:\n{err}" + ); +} + +/// A workflow survives a `pull` → `plan` round-trip with no phantom drift. +/// Regression for the bug where `pic pull` dropped `[[workflows]]`, so a +/// subsequent `apply --prune` would silently delete them. +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn workflow_pull_round_trips_clean() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let slug = common::unique_slug("wfpull"); + common::pic_as(&env) + .args(["apps", "create", &slug]) + .assert() + .success(); + let _guard = AppGuard::new(&env.url, &env.token, &slug); + + let dir = manifest_dir(); + seed_scripts(&dir); + // Exercise the full step shape the pull mapper must round-trip: input + // template, when, depends_on, retry (with defaults omitted), on_error. + let manifest = format!( + "[app]\nslug = \"{slug}\"\nname = \"WF Pull\"\n\n{SCRIPTS_BLOCK}\ + [[workflows]]\nname = \"pipeline\"\n\n\ + [[workflows.steps]]\nname = \"v\"\nfunction = \"validate\"\n\ + input = {{ order = \"{{{{ input.order }}}}\" }}\n\n\ + [[workflows.steps]]\nname = \"c\"\nfunction = \"charge\"\n\ + depends_on = [\"v\"]\nwhen = \"steps.v.output.ok == true\"\n\ + on_error = \"continue\"\n[workflows.steps.retry]\nmax_attempts = 3\n" + ); + let path = dir.path().join("picloud.toml"); + fs::write(&path, &manifest).unwrap(); + common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .assert() + .success(); + + // Pull the live state into a fresh dir, then plan it back — the workflow + // must round-trip to a clean no-op (no create / update / delete). + let out = TempDir::new().expect("tempdir"); + common::pic_as(&env) + .args(["pull", &slug, "--dir"]) + .arg(out.path()) + .assert() + .success(); + let pulled = out.path().join("picloud.toml"); + let toml = fs::read_to_string(&pulled).unwrap(); + assert!( + toml.contains("[[workflows]]") && toml.contains("pipeline"), + "pulled manifest is missing the workflow:\n{toml}" + ); + + let plan = common::pic_as(&env) + .args(["plan", "--file"]) + .arg(&pulled) + .output() + .expect("plan"); + let stdout = String::from_utf8_lossy(&plan.stdout); + assert!( + plan.status.success(), + "plan failed: {}\n{stdout}", + String::from_utf8_lossy(&plan.stderr) + ); + // The workflow row must be a no-op; nothing created/updated/deleted. + assert!( + !stdout.contains("create") && !stdout.contains("update") && !stdout.contains("delete"), + "pull→plan should diff clean for workflows, got:\n{stdout}" + ); +} + /// M5 end-to-end: `pic workflows run` starts a run, the durable orchestrator /// (running inside the fixture's picloud) executes both steps, and /// `run-status` shows it succeeded. diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index c8a3658..21251be 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -120,4 +120,7 @@ pub use users::{ pub use validator::{ScriptValidator, ValidatedScript, ValidationError}; pub use vars::{NoopVarsService, VarsError, VarsService}; pub use version::{API_VERSION, PRODUCT_VERSION, SDK_VERSION, WIRE_VERSION}; -pub use workflow::{NoopWorkflowService, WorkflowError, WorkflowService}; +pub use workflow::{ + NoopWorkflowService, OnError, RunStatus, StepStatus, WorkflowBackoff, WorkflowDefinition, + WorkflowError, WorkflowRetry, WorkflowService, WorkflowStepDef, +};