fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(workflows): close review gaps — pull round-trip, dup-names, reclaim budget

Browse Source 
Adversarial review of the v1.2 Workflows track surfaced two HIGH footguns and
several correctness/hardening gaps. Fixes:

HIGH
- pull round-trip: `pic pull` dropped `[[workflows]]`, so a later
  `apply --prune` silently deleted them. The list endpoint now returns the full
  `definition`; pull rebuilds the manifest block (inverse of workflow_to_wire).
- case-colliding names: two workflows differing only by case collided in the
  reconcile diff (keyed by lower(name)), silently dropping one. Rejected up
  front in validate_bundle_for.

MEDIUM
- reclaim retry budget: a crashed attempt (no outcome) consumed the retry
  budget. reclaim_stale_steps now decrements `attempt` (floored) and clears
  `next_attempt_at`, so a crash no longer counts as a failed try.
- on_error/backoff: typed the manifest fields against the shared enums so a
  bad value fails at TOML parse with a clear message, not an opaque 500.
- dedupe depends_on: a repeated dependency inflated the Kahn in-degree past the
  single decrement, reporting a spurious cycle for a valid DAG. Count distinct.
- canceled child: a canceled sub-workflow resolved the parent step with a
  message naming the cause instead of a generic "failed"; documented that a
  run-level cancel op is not yet supported.

LOW
- list_run_steps is now app-scoped at the query (JOIN workflow_runs) rather
  than relying on caller pre-verification.
- partial failure is surfaced: a run that succeeds with on_error=continue
  failures records them in the run's `error` field.
- admin-started runs log the root_execution_id against the principal.
- documented the deliberate when(missing→false) vs template(missing→fail)
  asymmetry; corrected the claim's atomicity comment.

Tests: unit (dedupe-deps), DB-gated reclaim-budget assertion, and two new CLI
journeys (duplicate-name rejection, pull→plan clean round-trip). fmt + clippy
-D warnings clean; 450 lib + 14 orchestrator DB + 5 workflow journeys pass;
schema snapshot unchanged (no migration).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-12 18:56:44 +02:00
parent bd9c3fa4f0
commit eaf5ace30f
10 changed files with 331 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
crates/manager-core/src/apply_service.rs
View File

@@ -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<String> = 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!([

10
crates/manager-core/src/workflow_expr.rs
View File

@@ -11,7 +11,15 @@
//! (`== != < > <= >=`), primary (`( … )`, `exists <ref>`, literal, `<ref>`).
//! 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.<name>.output` prefix and that
//! `<name>` 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 <ref>` to make "the value may be absent" explicit.
//!
//! [`workflow_template`]: crate::workflow_template

73
crates/manager-core/src/workflow_repo.rs
View File

@@ -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<Option<ClaimedStep>, WorkflowRepoError> {
let token = Uuid::new_v4();
let row: Option<ClaimedRow> = 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<u64, WorkflowRepoError> {
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<StepRow> 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<Vec<WorkflowRunStep>, WorkflowRepoError> {
let rows: Vec<StepRow> = 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<u64, WorkflowRepo
let mut resolved = 0u64;
for r in rows {
let mut tx = pool.begin().await?;
let (new_status, out, err) = if r.child_status == "succeeded" {
("succeeded", r.child_output.clone(), None)
} else {
(
// A child terminates as succeeded / failed / canceled. There is no
// distinct `canceled` step status (nor a run-cancel operation yet — the
// `canceled` run status is reserved for a future admin cancel), so a
// canceled child resolves the parent step to `failed` with a message
// that names the cause rather than masquerading as a generic failure.
let (new_status, out, err) = match r.child_status.as_str() {
"succeeded" => ("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<u64, WorkflowRepo
.clone()
.unwrap_or_else(|| "sub-workflow failed".to_string()),
),
)
),
};
let res = sqlx::query(
"UPDATE workflow_run_steps \

22
crates/manager-core/src/workflows_api.rs
View File

@@ -18,7 +18,7 @@ use axum::response::{IntoResponse, Json, Response};
use axum::routing::get;
use axum::{Extension, Router};
use chrono::{DateTime, Utc};
use picloud_shared::{AppId, Principal, WorkflowRunId};
use picloud_shared::{AppId, Principal, WorkflowDefinition, WorkflowRunId};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use uuid::Uuid;
@@ -58,7 +58,12 @@ pub fn workflows_router(state: WorkflowsState) -> 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