diff --git a/crates/manager-core/migrations/0072_execution_source_workflow.sql b/crates/manager-core/migrations/0072_execution_source_workflow.sql new file mode 100644 index 0000000..1257915 --- /dev/null +++ b/crates/manager-core/migrations/0072_execution_source_workflow.sql @@ -0,0 +1,18 @@ +-- v1.2 Workflows M2: the orchestrator executes each workflow step as a normal +-- function invocation and logs it, so `pic logs` can surface (and filter by) +-- workflow-driven runs — exactly as it does for cron/queue/invoke triggers. +-- +-- Add 'workflow' to the execution_logs.source CHECK (extends 0043). The column +-- default stays 'http'; this only widens the allowed set. Keep this list in +-- sync with `shared::ExecutionSource` (which gains the matching `Workflow` +-- variant) and `manager-core::OutboxSourceKind`. +-- +-- Postgres has no ALTER … ALTER CONSTRAINT for a CHECK, so drop + re-add. + +ALTER TABLE execution_logs DROP CONSTRAINT execution_logs_source_check; +ALTER TABLE execution_logs + ADD CONSTRAINT execution_logs_source_check + CHECK (source IN ( + 'http', 'kv', 'docs', 'dead_letter', 'cron', + 'files', 'pubsub', 'email', 'invoke', 'queue', 'workflow' + )); diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 62cff63..2080a60 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -109,6 +109,7 @@ pub mod vars_api; pub mod vars_repo; pub mod vars_service; pub mod workflow_expr; +pub mod workflow_orchestrator; pub mod workflow_repo; pub mod workflow_template; @@ -267,3 +268,4 @@ pub use users_service::{UsersServiceConfig, UsersServiceImpl}; pub use vars_api::{vars_router, VarsApiError, VarsApiState}; pub use vars_repo::{PostgresVarsRepo, VarOwner, VarRow, VarsRepo, VarsRepoError}; pub use vars_service::VarsServiceImpl; +pub use workflow_orchestrator::{spawn_workflow_orchestrator, WorkflowOrchestrator}; diff --git a/crates/manager-core/src/workflow_orchestrator.rs b/crates/manager-core/src/workflow_orchestrator.rs new file mode 100644 index 0000000..4ed5b0a --- /dev/null +++ b/crates/manager-core/src/workflow_orchestrator.rs @@ -0,0 +1,395 @@ +//! The durable workflow orchestrator (v1.2 Workflows, M2). +//! +//! A single dedicated background worker — mirroring [`cron_scheduler`] — that +//! advances every in-flight workflow run step-by-step, durably, surviving +//! restarts. It is deliberately NOT folded into `dispatcher.rs`: a per-*run* +//! graph advance is its own transaction with its own lease semantics. +//! +//! Each tick has two phases: +//! +//! **A. Claim + execute.** Up to a small batch of `ready`, due steps are +//! claimed with the same `FOR UPDATE SKIP LOCKED` competing-consumer lease +//! the queue uses ([`workflow_repo::claim_ready_step`]) — one execution-gate +//! permit acquired per step *before* the claim, so the shared gate bounds +//! real parallelism and no step is ever claimed without a slot to run it. +//! Each claimed step resolves its function by name (in the run's app scope), +//! builds an `ExecRequest`, and runs through the injected [`ExecutorClient`] +//! — the same executor path every other trigger uses. Claimed steps run +//! concurrently within the tick. +//! +//! **B. Advance.** The step outcome is written and the DAG advanced in one +//! token-gated transaction ([`workflow_repo::complete_step_and_advance`]): +//! pending steps whose dependencies are now satisfied flip to `ready`, and +//! the run's terminal status is recomputed. Fan-in falls out for free (a +//! join step stays `pending` until its last dependency flips it). A stale +//! worker (its lease reclaimed) matches zero rows and writes nothing. +//! +//! A second, slower cadence reclaims steps leased by a crashed worker +//! ([`workflow_repo::reclaim_stale_steps`]) — the durability safety net. +//! +//! M2 executes **function** steps only; `workflow`-kind (nested sub-workflow) +//! steps land in M4, and `when` conditions + input templating in M3. The seams +//! are present (the step target enum, `run_input`, `workflow_depth`). +//! +//! [`cron_scheduler`]: crate::cron_scheduler + +use std::sync::Arc; +use std::time::Duration as StdDuration; + +use chrono::{Duration as ChronoDuration, Utc}; +use sqlx::PgPool; +use tokio::sync::OwnedSemaphorePermit; + +use picloud_executor_core::{build_execution_log, ExecRequest, InvocationType}; +use picloud_orchestrator_core::{ExecutionGate, ExecutorClient, ScriptIdentity}; +use picloud_shared::workflow::{StepTarget, WorkflowBackoff, WorkflowStepDef}; +use picloud_shared::{ExecutionId, ExecutionLogSink, ExecutionSource, RequestId}; + +use crate::dispatcher::compute_backoff; +use crate::repo::ScriptRepository; +use crate::trigger_config::BackoffShape; +use crate::workflow_repo::{ + claim_ready_step, complete_step_and_advance, reclaim_stale_steps, ClaimedStep, StepOutcome, +}; + +/// Max steps claimed (and thus gate permits held) per orchestrator tick. Keeps +/// the workflow worker from monopolizing the gate it shares with HTTP + the +/// dispatcher; the tick cadence + reclaim pick up the rest. +const STEP_CLAIM_BATCH: usize = 16; + +/// Tunables, all env-overridable (`PICLOUD_WORKFLOW_*`), matching the +/// parse-warn-fallback pattern the rest of the crate uses. +#[derive(Debug, Clone, Copy)] +pub struct WorkflowOrchestratorConfig { + pub tick_interval_ms: u64, + pub reclaim_interval_ms: u64, + pub visibility_timeout_secs: u32, + pub retry_jitter_pct: u32, + pub step_timeout: StdDuration, + /// Nesting ceiling for sub-workflows (M4). Read now so the knob is stable. + pub max_depth: u32, +} + +impl Default for WorkflowOrchestratorConfig { + fn default() -> Self { + Self { + tick_interval_ms: 250, + reclaim_interval_ms: 30_000, + visibility_timeout_secs: 300, + retry_jitter_pct: 20, + step_timeout: StdDuration::from_secs(300), + max_depth: 8, + } + } +} + +impl WorkflowOrchestratorConfig { + #[must_use] + pub fn from_env() -> Self { + let d = Self::default(); + Self { + tick_interval_ms: env_u64("PICLOUD_WORKFLOW_TICK_INTERVAL_MS", d.tick_interval_ms), + reclaim_interval_ms: env_u64( + "PICLOUD_WORKFLOW_RECLAIM_INTERVAL_MS", + d.reclaim_interval_ms, + ), + visibility_timeout_secs: env_u32( + "PICLOUD_WORKFLOW_STEP_VISIBILITY_TIMEOUT_SECS", + d.visibility_timeout_secs, + ), + retry_jitter_pct: env_u32("PICLOUD_WORKFLOW_RETRY_JITTER_PCT", d.retry_jitter_pct), + step_timeout: StdDuration::from_secs(env_u64( + "PICLOUD_WORKFLOW_STEP_TIMEOUT_SEC", + d.step_timeout.as_secs(), + )), + max_depth: env_u32("PICLOUD_WORKFLOW_MAX_DEPTH", d.max_depth), + } + } +} + +fn env_u64(key: &str, default: u64) -> u64 { + match std::env::var(key) { + Ok(v) => v.trim().parse::().map_or_else( + |_| { + tracing::warn!(%key, value = %v, "ignoring invalid value; want a positive integer"); + default + }, + |n| if n > 0 { n } else { default }, + ), + Err(_) => default, + } +} + +fn env_u32(key: &str, default: u32) -> u32 { + u32::try_from(env_u64(key, u64::from(default))).unwrap_or(default) +} + +/// The orchestrator's collaborators. All the durable state lives behind the +/// pool free-fns in [`workflow_repo`](crate::workflow_repo); execution goes +/// through the injected [`ExecutorClient`], never `executor-core` directly. +pub struct WorkflowOrchestrator { + pub pool: PgPool, + pub scripts: Arc, + pub executor: Arc, + pub gate: Arc, + pub log_sink: Arc, + pub config: WorkflowOrchestratorConfig, +} + +impl WorkflowOrchestrator { + /// Spawn the advance loop + the reclaim task. Both run for the process + /// lifetime; returned handles are dropped on purpose (as with the + /// dispatcher + cron scheduler). + pub fn spawn(self) { + // Reclaim task — independent, slower cadence so it doesn't contend + // with the per-tick advance loop. + let reclaim_pool = self.pool.clone(); + let reclaim_interval = StdDuration::from_millis(self.config.reclaim_interval_ms.max(1_000)); + let visibility = self.config.visibility_timeout_secs; + tokio::spawn(async move { + let mut ticker = tokio::time::interval(reclaim_interval); + ticker.tick().await; + loop { + ticker.tick().await; + match reclaim_stale_steps(&reclaim_pool, visibility).await { + Ok(0) => {} + Ok(n) => { + tracing::info!(reclaimed = n, "workflow step visibility-timeout reclaim"); + } + Err(e) => tracing::warn!(?e, "workflow reclaim task errored"), + } + } + }); + + // Advance loop. + let interval = StdDuration::from_millis(self.config.tick_interval_ms.max(50)); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + ticker.tick().await; + loop { + ticker.tick().await; + self.tick_once().await; + } + }); + } + + /// One advance tick: claim a permit-bounded batch of ready steps, then run + /// them (execute + advance) concurrently. Public so integration tests can + /// drive the loop deterministically (one wave per call). + pub async fn tick_once(&self) { + use futures::stream::{self, StreamExt}; + + let mut batch: Vec<(ClaimedStep, OwnedSemaphorePermit)> = Vec::new(); + for _ in 0..STEP_CLAIM_BATCH { + // Acquire the gate slot BEFORE claiming so a claimed step always + // has somewhere to run; if the gate is saturated, stop this tick. + let Ok(permit) = self.gate.try_acquire() else { + break; + }; + match claim_ready_step(&self.pool).await { + Ok(Some(claimed)) => batch.push((claimed, permit)), + Ok(None) => { + drop(permit); + break; + } + Err(e) => { + drop(permit); + tracing::warn!(?e, "workflow claim errored"); + break; + } + } + } + if batch.is_empty() { + return; + } + // Each entry already holds a permit, so run them all concurrently. + stream::iter(batch) + .for_each_concurrent(None, |(claimed, permit)| async move { + self.run_step(claimed, permit).await; + }) + .await; + } + + /// Execute one claimed step, then write its outcome + advance the run. + async fn run_step(&self, claimed: ClaimedStep, permit: OwnedSemaphorePermit) { + let outcome = self.execute_step(&claimed).await; + // Release the gate slot before the (DB-only, fast) advance write. + drop(permit); + if let Err(e) = complete_step_and_advance(&self.pool, &claimed, outcome).await { + tracing::error!( + ?e, + run_id = %claimed.run_id, + step = %claimed.step_name, + "workflow step advance write failed" + ); + } + } + + /// Resolve + run a step's function, returning the outcome. Failures are + /// converted to [`StepOutcome::Failed`] carrying the retry delay derived + /// from the step's own retry policy (the repo decides retry-vs-terminal). + #[allow(clippy::too_many_lines)] + async fn execute_step(&self, claimed: &ClaimedStep) -> StepOutcome { + let Some(step) = claimed.step_def() else { + return self.fail( + None, + claimed, + "step not found in workflow definition".into(), + ); + }; + + let fn_name = match step.target() { + Some(StepTarget::Function(f)) => f.to_string(), + Some(StepTarget::Workflow(_)) => { + // Nested sub-workflows land in M4. + return self.fail( + Some(step), + claimed, + "sub-workflow steps not yet supported".into(), + ); + } + None => { + return self.fail( + Some(step), + claimed, + "step declares neither `function` nor `workflow`".into(), + ); + } + }; + + // Resolve the function by name in the RUN's app scope (an inherited + // group script is reachable, nearest-owner-wins) — never off a + // script-passed arg. cx isolation boundary preserved. + let script = match self + .scripts + .get_by_name_inherited(claimed.app_id, &fn_name) + .await + { + Ok(Some(s)) if s.enabled => s, + Ok(Some(_)) => { + return self.fail( + Some(step), + claimed, + format!("function {fn_name:?} is disabled"), + ); + } + Ok(None) => { + return self.fail( + Some(step), + claimed, + format!("function {fn_name:?} not found"), + ); + } + Err(e) => { + return self.fail(Some(step), claimed, format!("resolving {fn_name:?}: {e}")); + } + }; + + // M2 input: pass the step's static `input` if present, else the run + // input. M3 replaces this with `{{ … }}` template resolution against + // prior step outputs. + let body = if step.input.is_null() { + claimed.run_input.clone() + } else { + step.input.clone() + }; + + let execution_id = ExecutionId::new(); + let request_id = RequestId::new(); + let req = ExecRequest { + execution_id, + request_id, + script_id: script.id, + script_name: script.name.clone(), + invocation_type: InvocationType::Function, + path: format!("/workflow/{}", claimed.step_name), + method: String::new(), + headers: std::collections::BTreeMap::new(), + body: body.clone(), + params: std::collections::BTreeMap::new(), + query: std::collections::BTreeMap::new(), + rest: String::new(), + sandbox_overrides: script.sandbox, + app_id: claimed.app_id, + script_owner: script.owner(), + // Like invoke_async / HTTP outbox rows: steps run with no + // principal (the run's origin is forensic, not re-authenticated). + principal: None, + trigger_depth: 0, + root_execution_id: ExecutionId::from(claimed.root_execution_id), + is_dead_letter_handler: false, + event: None, + }; + let identity = ScriptIdentity { + script_id: script.id, + updated_at: script.updated_at, + }; + + let started = Utc::now(); + let outcome = self + .executor + .execute_with_identity(identity, &script.source, req, self.config.step_timeout) + .await; + let finished = Utc::now(); + + // Log the run so `pic logs` surfaces workflow steps (source=workflow). + let log = build_execution_log( + claimed.app_id, + script.id, + request_id, + format!("/workflow/{}", claimed.step_name), + std::collections::BTreeMap::new(), + body, + ExecutionSource::Workflow, + &outcome, + started, + finished, + ); + if let Err(e) = self.log_sink.record(log).await { + tracing::warn!(?e, step = %claimed.step_name, "workflow step log write failed"); + } + + match outcome { + Ok(resp) => StepOutcome::Succeeded(resp.body), + Err(err) => self.fail(Some(step), claimed, err.to_string()), + } + } + + /// Build a `Failed` outcome, computing the retry backoff from the step's + /// own retry policy (zero if it has none — the repo then treats the first + /// failure as terminal). + fn fail( + &self, + step: Option<&WorkflowStepDef>, + claimed: &ClaimedStep, + error: String, + ) -> StepOutcome { + let retry_delay = step + .and_then(|s| s.retry.as_ref()) + .map_or(ChronoDuration::zero(), |r| { + let ms = compute_backoff( + u32::try_from(claimed.attempt.max(1)).unwrap_or(1), + map_backoff(r.backoff), + r.base_ms, + self.config.retry_jitter_pct, + ); + ChronoDuration::milliseconds(i64::from(ms)) + }); + StepOutcome::Failed { error, retry_delay } + } +} + +/// Map the self-contained workflow backoff DTO onto the shared trigger backoff. +fn map_backoff(b: WorkflowBackoff) -> BackoffShape { + match b { + WorkflowBackoff::Constant => BackoffShape::Constant, + WorkflowBackoff::Linear => BackoffShape::Linear, + WorkflowBackoff::Exponential => BackoffShape::Exponential, + } +} + +/// Consume the orchestrator and spawn its background tasks (mirrors +/// [`spawn_cron_scheduler`](crate::spawn_cron_scheduler)). +pub fn spawn_workflow_orchestrator(orchestrator: WorkflowOrchestrator) { + orchestrator.spawn(); +} diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs index 7d1ed8f..3cc2153 100644 --- a/crates/manager-core/src/workflow_repo.rs +++ b/crates/manager-core/src/workflow_repo.rs @@ -6,8 +6,14 @@ //! declarative `apply` reconcile engine (Create/Update/Delete by `lower(name)`, //! all in one transaction) — mirroring `trigger_repo::insert_trigger_tx`. //! -//! Run/step state (`workflow_runs`, `workflow_run_steps`) is added in M2 with -//! the orchestrator. +//! Run/step state (`workflow_runs`, `workflow_run_steps`) lands in M2: the +//! durable orchestrator seeds a run (`start_run`), claims a `ready` step with +//! the same `FOR UPDATE SKIP LOCKED` lease `queue_repo` uses, and — in one +//! token-gated transaction — writes the step outcome and advances the DAG +//! (`complete_step_and_advance`). The graph-advance decision is a pure, +//! DB-free function ([`compute_advance`]) so it is unit-testable. + +use std::collections::BTreeMap; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -15,8 +21,10 @@ use sqlx::PgPool; use thiserror::Error; use uuid::Uuid; -use picloud_shared::workflow::WorkflowDefinition; -use picloud_shared::{AppId, WorkflowId}; +use picloud_shared::workflow::{ + OnError, RunStatus, StepStatus, WorkflowDefinition, WorkflowStepDef, +}; +use picloud_shared::{AppId, WorkflowId, WorkflowRunId, WorkflowRunStepId}; #[derive(Debug, Error)] pub enum WorkflowRepoError { @@ -227,3 +235,785 @@ pub async fn delete_workflow_tx( .await?; Ok(()) } + +// =========================================================================== +// M2 — run/step state + the durable orchestrator's persistence surface. +// =========================================================================== + +/// Everything needed to seed a fresh run (mirrors `NewQueueMessage`). The +/// orchestrator's M5 `workflow::start` and the M2 tests both build this. +#[derive(Debug, Clone)] +pub struct NewWorkflowRun { + pub workflow_id: WorkflowId, + pub app_id: AppId, + pub input: serde_json::Value, + /// Correlates every step execution in `execution_logs` under one id. + pub root_execution_id: Uuid, + /// 0 for a top-level run; a nested sub-workflow (M4) is parent + 1. + pub workflow_depth: i32, + pub parent_run_id: Option, + pub parent_step_id: Option, +} + +/// A run row, read for the admin/CLI surface + test assertions. +#[derive(Debug, Clone)] +pub struct WorkflowRun { + pub id: WorkflowRunId, + pub workflow_id: WorkflowId, + pub app_id: AppId, + pub status: RunStatus, + pub input: serde_json::Value, + pub output: Option, + pub error: Option, + pub root_execution_id: Uuid, + pub workflow_depth: i32, + pub parent_run_id: Option, + pub parent_step_id: Option, + pub started_at: Option>, + pub finished_at: Option>, + pub created_at: DateTime, +} + +/// A step row within a run (read for the admin/CLI surface + assertions). +#[derive(Debug, Clone)] +pub struct WorkflowRunStep { + pub id: WorkflowRunStepId, + pub run_id: WorkflowRunId, + pub step_name: String, + pub status: StepStatus, + pub attempt: i32, + pub max_attempts: i32, + pub output: Option, + pub error: Option, + pub child_run_id: Option, +} + +/// A claimed `ready` step, joined with its run + workflow definition so the +/// orchestrator has everything it needs to resolve, execute, and advance +/// without a second round trip. +#[derive(Debug, Clone)] +pub struct ClaimedStep { + pub step_id: WorkflowRunStepId, + pub run_id: WorkflowRunId, + pub step_name: String, + /// 1-indexed current attempt (the claim increments it). + pub attempt: i32, + pub max_attempts: i32, + pub claim_token: Uuid, + // ---- run context ---- + pub app_id: AppId, + pub workflow_id: WorkflowId, + pub root_execution_id: Uuid, + pub workflow_depth: i32, + pub run_input: serde_json::Value, + pub definition: WorkflowDefinition, +} + +impl ClaimedStep { + /// The definition of the step being executed. + #[must_use] + pub fn step_def(&self) -> Option<&WorkflowStepDef> { + self.definition + .steps + .iter() + .find(|s| s.name == self.step_name) + } +} + +/// The outcome the orchestrator hands back after executing a step. On +/// `Failed`, the repo decides retry-vs-terminal from `attempt < max_attempts`; +/// the orchestrator supplies the delay to use if a retry remains. +#[derive(Debug, Clone)] +pub enum StepOutcome { + Succeeded(serde_json::Value), + Failed { + error: String, + retry_delay: chrono::Duration, + }, +} + +/// What `complete_step_and_advance` did — surfaced mostly for tests/logging. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdvanceResult { + /// The claim token no longer matched (a stale worker); nothing written. + Stale, + /// The step failed but had attempts left; it was re-armed as `ready`. + Retried, + /// The step reached a terminal status and the run graph was advanced. + Advanced, +} + +// ---- the pure graph-advance decision (DB-free, unit-tested) -------------- + +/// The write plan `complete_step_and_advance` derives after a step reaches a +/// terminal status: which `pending` steps become `ready`, and the run's new +/// status/output/error. +#[derive(Debug, Clone, PartialEq)] +pub struct AdvancePlan { + pub promote: Vec, + pub run_status: RunStatus, + pub run_error: Option, + pub run_output: Option, +} + +/// Is `dep` satisfied for the purpose of unblocking a dependent? Succeeded and +/// skipped always satisfy; a `failed` step satisfies **only** when its +/// definition says `on_error = continue` (the run limps on). +fn dep_satisfied(status: StepStatus, on_error: OnError) -> bool { + match status { + StepStatus::Succeeded | StepStatus::Skipped => true, + StepStatus::Failed => matches!(on_error, OnError::Continue), + _ => false, + } +} + +/// Pure DAG advance. Given the definition and the current per-step +/// status/output snapshot, decide the promotions + terminal run state. +/// +/// - A `pending` step whose every `depends_on` is satisfied becomes `ready`. +/// - If any step `failed` with `on_error = fail`, the run fails immediately. +/// - Else if every step is terminal, the run succeeds; its output is the +/// object of `{ step_name: output }` over the succeeded steps. +/// - Otherwise the run is still `running`. +#[must_use] +pub fn compute_advance( + def: &WorkflowDefinition, + statuses: &BTreeMap, + outputs: &BTreeMap, +) -> AdvancePlan { + let by_name: BTreeMap<&str, &WorkflowStepDef> = + def.steps.iter().map(|s| (s.name.as_str(), s)).collect(); + let status_of = |name: &str| statuses.get(name).copied().unwrap_or(StepStatus::Pending); + + // Promote every pending step whose deps are all satisfied. + let mut promote = Vec::new(); + for step in &def.steps { + if status_of(&step.name) != StepStatus::Pending { + continue; + } + let ready = step.depends_on.iter().all(|dep| { + let d_on_err = by_name + .get(dep.as_str()) + .map_or(OnError::Fail, |s| s.on_error); + dep_satisfied(status_of(dep), d_on_err) + }); + if ready { + promote.push(step.name.clone()); + } + } + + // A hard failure (on_error = fail that exhausted) fails the whole run. + if let Some(failed) = def + .steps + .iter() + .find(|s| status_of(&s.name) == StepStatus::Failed && matches!(s.on_error, OnError::Fail)) + { + return AdvancePlan { + promote, + run_status: RunStatus::Failed, + run_error: Some(format!("step {:?} failed", failed.name)), + run_output: None, + }; + } + + // All steps terminal (accounting for the promotions we just queued — a + // promoted step is pending → ready, i.e. not terminal, so the run stays + // running whenever `promote` is non-empty). + let all_terminal = + promote.is_empty() && def.steps.iter().all(|s| status_of(&s.name).is_terminal()); + if all_terminal { + let mut out = serde_json::Map::new(); + for step in &def.steps { + if status_of(&step.name) == StepStatus::Succeeded { + if let Some(v) = outputs.get(&step.name) { + out.insert(step.name.clone(), v.clone()); + } + } + } + return AdvancePlan { + promote, + run_status: RunStatus::Succeeded, + run_error: None, + run_output: Some(serde_json::Value::Object(out)), + }; + } + + AdvancePlan { + promote, + run_status: RunStatus::Running, + run_error: None, + run_output: None, + } +} + +// ---- run seeding ---------------------------------------------------------- + +/// Seed a run: insert the `workflow_runs` row + one `workflow_run_steps` row +/// per definition step (root steps — no `depends_on` — start `ready`, the rest +/// `pending`). All in one transaction so a partially-seeded run is never +/// visible to the orchestrator's claim scan. +pub async fn start_run( + pool: &PgPool, + new: NewWorkflowRun, + definition: &WorkflowDefinition, +) -> Result { + let mut tx = pool.begin().await?; + let (run_id,): (Uuid,) = sqlx::query_as( + "INSERT INTO workflow_runs \ + (workflow_id, app_id, status, input, root_execution_id, \ + workflow_depth, parent_run_id, parent_step_id) \ + VALUES ($1, $2, 'pending', $3, $4, $5, $6, $7) RETURNING id", + ) + .bind(new.workflow_id.into_inner()) + .bind(new.app_id.into_inner()) + .bind(&new.input) + .bind(new.root_execution_id) + .bind(new.workflow_depth) + .bind(new.parent_run_id.map(WorkflowRunId::into_inner)) + .bind(new.parent_step_id.map(WorkflowRunStepId::into_inner)) + .fetch_one(&mut *tx) + .await?; + + for step in &definition.steps { + let status = if step.depends_on.is_empty() { + "ready" + } else { + "pending" + }; + let max_attempts = i32::try_from(step.max_attempts()).unwrap_or(1); + sqlx::query( + "INSERT INTO workflow_run_steps \ + (run_id, step_name, status, max_attempts) \ + VALUES ($1, $2, $3, $4)", + ) + .bind(run_id) + .bind(&step.name) + .bind(status) + .bind(max_attempts) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(run_id.into()) +} + +// ---- claim / complete / reclaim ------------------------------------------ + +#[derive(sqlx::FromRow)] +struct ClaimedRow { + id: Uuid, + run_id: Uuid, + step_name: String, + attempt: i32, + max_attempts: i32, + app_id: Uuid, + workflow_id: Uuid, + root_execution_id: Uuid, + workflow_depth: i32, + input: serde_json::Value, + definition: serde_json::Value, +} + +/// 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. +pub async fn claim_ready_step(pool: &PgPool) -> Result, WorkflowRepoError> { + let token = Uuid::new_v4(); + let row: Option = sqlx::query_as( + "WITH claimed AS ( \ + UPDATE workflow_run_steps \ + SET status = 'running', claim_token = $1, claimed_at = NOW(), \ + attempt = attempt + 1, updated_at = NOW() \ + WHERE id = ( \ + SELECT s.id FROM workflow_run_steps s \ + JOIN workflow_runs r ON r.id = s.run_id \ + WHERE s.status = 'ready' \ + AND (s.next_attempt_at IS NULL OR s.next_attempt_at <= NOW()) \ + AND r.status IN ('pending', 'running') \ + ORDER BY s.created_at \ + FOR UPDATE OF s SKIP LOCKED \ + LIMIT 1 \ + ) \ + RETURNING id, run_id, step_name, attempt, max_attempts \ + ) \ + SELECT c.id, c.run_id, c.step_name, c.attempt, c.max_attempts, \ + r.app_id, r.workflow_id, r.root_execution_id, r.workflow_depth, \ + r.input, w.definition \ + FROM claimed c \ + JOIN workflow_runs r ON r.id = c.run_id \ + JOIN workflows w ON w.id = r.workflow_id", + ) + .bind(token) + .fetch_optional(pool) + .await?; + + let Some(r) = row else { return Ok(None) }; + + // Reflect the run as running promptly (advance also does this, but a + // long-running first step would otherwise leave the run 'pending'). + sqlx::query( + "UPDATE workflow_runs SET status = 'running', started_at = COALESCE(started_at, NOW()) \ + WHERE id = $1 AND status = 'pending'", + ) + .bind(r.run_id) + .execute(pool) + .await?; + + let definition = serde_json::from_value(r.definition) + .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?; + Ok(Some(ClaimedStep { + step_id: r.id.into(), + run_id: r.run_id.into(), + step_name: r.step_name, + attempt: r.attempt, + max_attempts: r.max_attempts, + claim_token: token, + app_id: r.app_id.into(), + workflow_id: r.workflow_id.into(), + root_execution_id: r.root_execution_id, + workflow_depth: r.workflow_depth, + run_input: r.input, + definition, + })) +} + +#[derive(sqlx::FromRow)] +struct StepStateRow { + step_name: String, + status: String, + output: Option, +} + +/// Write a claimed step's outcome and advance the run graph — all in one +/// token-gated transaction. A stale worker (claim reclaimed out from under it) +/// matches zero rows and writes nothing (`AdvanceResult::Stale`). +/// +/// - `Succeeded` → the step is marked `succeeded` with its output, then the +/// graph advances (promotions + terminal run status). +/// - `Failed` with attempts left → the step is re-armed `ready` with a backoff +/// gate; the run keeps running (`Retried`, no advance). +/// - `Failed` exhausted → the step is marked `failed`, then the graph advances +/// (its `on_error` decides whether the run fails or limps on). +#[allow(clippy::too_many_lines)] +pub async fn complete_step_and_advance( + pool: &PgPool, + claimed: &ClaimedStep, + outcome: StepOutcome, +) -> Result { + let mut tx = pool.begin().await?; + + // 1. Token-gated step write. + match &outcome { + StepOutcome::Succeeded(out) => { + let res = sqlx::query( + "UPDATE workflow_run_steps \ + SET status = 'succeeded', output = $3, error = NULL, \ + claim_token = NULL, claimed_at = NULL, updated_at = NOW() \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(claimed.step_id.into_inner()) + .bind(claimed.claim_token) + .bind(out) + .execute(&mut *tx) + .await?; + if res.rows_affected() == 0 { + tx.rollback().await?; + return Ok(AdvanceResult::Stale); + } + } + StepOutcome::Failed { error, retry_delay } => { + if claimed.attempt < claimed.max_attempts { + let next = Utc::now() + *retry_delay; + let res = sqlx::query( + "UPDATE workflow_run_steps \ + SET status = 'ready', error = $3, claim_token = NULL, \ + claimed_at = NULL, next_attempt_at = $4, updated_at = NOW() \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(claimed.step_id.into_inner()) + .bind(claimed.claim_token) + .bind(error) + .bind(next) + .execute(&mut *tx) + .await?; + if res.rows_affected() == 0 { + tx.rollback().await?; + return Ok(AdvanceResult::Stale); + } + tx.commit().await?; + return Ok(AdvanceResult::Retried); + } + let res = sqlx::query( + "UPDATE workflow_run_steps \ + SET status = 'failed', error = $3, claim_token = NULL, \ + claimed_at = NULL, updated_at = NOW() \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(claimed.step_id.into_inner()) + .bind(claimed.claim_token) + .bind(error) + .execute(&mut *tx) + .await?; + if res.rows_affected() == 0 { + tx.rollback().await?; + return Ok(AdvanceResult::Stale); + } + } + } + + // 2. Advance — lock the run row so concurrent advances of the same run + // serialize (each does a full recompute, so this is idempotent). + let run: Option<(String,)> = + sqlx::query_as("SELECT status FROM workflow_runs WHERE id = $1 FOR UPDATE") + .bind(claimed.run_id.into_inner()) + .fetch_optional(&mut *tx) + .await?; + let Some((run_status,)) = run else { + tx.commit().await?; + return Ok(AdvanceResult::Advanced); + }; + if RunStatus::from_str(&run_status).is_some_and(RunStatus::is_terminal) { + // Another worker already finished the run; nothing to do. + tx.commit().await?; + return Ok(AdvanceResult::Advanced); + } + + let step_rows: Vec = sqlx::query_as( + "SELECT step_name, status, output FROM workflow_run_steps WHERE run_id = $1", + ) + .bind(claimed.run_id.into_inner()) + .fetch_all(&mut *tx) + .await?; + let mut statuses = BTreeMap::new(); + let mut outputs = BTreeMap::new(); + for r in step_rows { + if let Some(st) = StepStatus::from_str(&r.status) { + statuses.insert(r.step_name.clone(), st); + } + if let Some(o) = r.output { + outputs.insert(r.step_name, o); + } + } + + let plan = compute_advance(&claimed.definition, &statuses, &outputs); + + for name in &plan.promote { + sqlx::query( + "UPDATE workflow_run_steps \ + SET status = 'ready', next_attempt_at = NULL, updated_at = NOW() \ + WHERE run_id = $1 AND step_name = $2 AND status = 'pending'", + ) + .bind(claimed.run_id.into_inner()) + .bind(name) + .execute(&mut *tx) + .await?; + } + + match plan.run_status { + RunStatus::Succeeded => { + sqlx::query( + "UPDATE workflow_runs \ + SET status = 'succeeded', output = $2, finished_at = NOW() WHERE id = $1", + ) + .bind(claimed.run_id.into_inner()) + .bind(plan.run_output) + .execute(&mut *tx) + .await?; + } + RunStatus::Failed => { + sqlx::query( + "UPDATE workflow_runs \ + SET status = 'failed', error = $2, finished_at = NOW() WHERE id = $1", + ) + .bind(claimed.run_id.into_inner()) + .bind(plan.run_error) + .execute(&mut *tx) + .await?; + } + _ => { + sqlx::query( + "UPDATE workflow_runs \ + SET status = 'running', started_at = COALESCE(started_at, NOW()) WHERE id = $1", + ) + .bind(claimed.run_id.into_inner()) + .execute(&mut *tx) + .await?; + } + } + + tx.commit().await?; + Ok(AdvanceResult::Advanced) +} + +/// 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. +/// Returns the number of steps reclaimed. +pub async fn reclaim_stale_steps( + pool: &PgPool, + visibility_timeout_secs: u32, +) -> Result { + let res = sqlx::query( + "UPDATE workflow_run_steps s \ + SET status = 'ready', claim_token = NULL, claimed_at = NULL, updated_at = NOW() \ + FROM workflow_runs r \ + WHERE s.run_id = r.id \ + AND s.claim_token IS NOT NULL \ + AND s.status = 'running' \ + AND s.claimed_at < NOW() - ($1 || ' seconds')::INTERVAL \ + AND r.status IN ('pending', 'running')", + ) + .bind(i64::from(visibility_timeout_secs)) + .execute(pool) + .await?; + Ok(res.rows_affected()) +} + +// ---- run/step reads (admin surface + test assertions) -------------------- + +#[derive(sqlx::FromRow)] +struct RunRow { + id: Uuid, + workflow_id: Uuid, + app_id: Uuid, + status: String, + input: serde_json::Value, + output: Option, + error: Option, + root_execution_id: Uuid, + workflow_depth: i32, + parent_run_id: Option, + parent_step_id: Option, + started_at: Option>, + finished_at: Option>, + created_at: DateTime, +} + +impl TryFrom for WorkflowRun { + type Error = WorkflowRepoError; + fn try_from(r: RunRow) -> Result { + Ok(Self { + id: r.id.into(), + workflow_id: r.workflow_id.into(), + app_id: r.app_id.into(), + status: RunStatus::from_str(&r.status).ok_or_else(|| { + WorkflowRepoError::Serde(format!("bad run status {:?}", r.status)) + })?, + input: r.input, + output: r.output, + error: r.error, + root_execution_id: r.root_execution_id, + workflow_depth: r.workflow_depth, + parent_run_id: r.parent_run_id.map(Into::into), + parent_step_id: r.parent_step_id.map(Into::into), + started_at: r.started_at, + finished_at: r.finished_at, + created_at: r.created_at, + }) + } +} + +const RUN_COLS: &str = "id, workflow_id, app_id, status, input, output, error, \ + root_execution_id, workflow_depth, parent_run_id, parent_step_id, \ + started_at, finished_at, created_at"; + +/// Read a single run by id (app-scoped for isolation). +pub async fn get_run( + pool: &PgPool, + app_id: AppId, + run_id: WorkflowRunId, +) -> Result, WorkflowRepoError> { + let row: Option = sqlx::query_as(&format!( + "SELECT {RUN_COLS} FROM workflow_runs WHERE id = $1 AND app_id = $2" + )) + .bind(run_id.into_inner()) + .bind(app_id.into_inner()) + .fetch_optional(pool) + .await?; + row.map(TryInto::try_into).transpose() +} + +#[derive(sqlx::FromRow)] +struct StepRow { + id: Uuid, + run_id: Uuid, + step_name: String, + status: String, + attempt: i32, + max_attempts: i32, + output: Option, + error: Option, + child_run_id: Option, +} + +impl TryFrom for WorkflowRunStep { + type Error = WorkflowRepoError; + fn try_from(r: StepRow) -> Result { + Ok(Self { + id: r.id.into(), + run_id: r.run_id.into(), + step_name: r.step_name, + status: StepStatus::from_str(&r.status).ok_or_else(|| { + WorkflowRepoError::Serde(format!("bad step status {:?}", r.status)) + })?, + attempt: r.attempt, + max_attempts: r.max_attempts, + output: r.output, + error: r.error, + child_run_id: r.child_run_id.map(Into::into), + }) + } +} + +/// All steps of a run, ordered by name (for the admin/CLI surface + tests). +pub async fn list_run_steps( + pool: &PgPool, + 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", + ) + .bind(run_id.into_inner()) + .fetch_all(pool) + .await?; + rows.into_iter().map(TryInto::try_into).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use picloud_shared::workflow::WorkflowStepDef; + use serde_json::json; + + fn step(name: &str, deps: &[&str]) -> WorkflowStepDef { + WorkflowStepDef { + name: name.into(), + function: Some(format!("fn_{name}")), + workflow: None, + input: serde_json::Value::Null, + depends_on: deps.iter().map(|s| (*s).to_string()).collect(), + when: None, + retry: None, + on_error: OnError::Fail, + } + } + + fn def(steps: Vec) -> WorkflowDefinition { + WorkflowDefinition { steps } + } + + fn statuses(pairs: &[(&str, StepStatus)]) -> BTreeMap { + pairs.iter().map(|(n, s)| ((*n).to_string(), *s)).collect() + } + + #[test] + fn promotes_dependent_when_dep_succeeds() { + // a -> b. a succeeded, b still pending → b becomes ready. + let d = def(vec![step("a", &[]), step("b", &["a"])]); + let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Pending)]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert_eq!(plan.promote, vec!["b".to_string()]); + assert_eq!(plan.run_status, RunStatus::Running); + } + + #[test] + fn fan_in_waits_for_all_deps() { + // c depends on [a, b]. Only a done → c must NOT be promoted. + let d = def(vec![step("a", &[]), step("b", &[]), step("c", &["a", "b"])]); + let st = statuses(&[ + ("a", StepStatus::Succeeded), + ("b", StepStatus::Running), + ("c", StepStatus::Pending), + ]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert!(plan.promote.is_empty()); + assert_eq!(plan.run_status, RunStatus::Running); + + // Now both a and b done → c promoted. + let st = statuses(&[ + ("a", StepStatus::Succeeded), + ("b", StepStatus::Succeeded), + ("c", StepStatus::Pending), + ]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert_eq!(plan.promote, vec!["c".to_string()]); + } + + #[test] + fn all_terminal_succeeds_with_output_object() { + let d = def(vec![step("a", &[]), step("b", &["a"])]); + let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Succeeded)]); + let mut out = BTreeMap::new(); + out.insert("a".to_string(), json!({ "x": 1 })); + out.insert("b".to_string(), json!("done")); + let plan = compute_advance(&d, &st, &out); + assert!(plan.promote.is_empty()); + assert_eq!(plan.run_status, RunStatus::Succeeded); + assert_eq!( + plan.run_output, + Some(json!({ "a": { "x": 1 }, "b": "done" })) + ); + } + + #[test] + fn on_error_fail_exhausted_fails_the_run() { + let d = def(vec![step("a", &[]), step("b", &["a"])]); + // a failed (on_error = fail) → run fails, b never promoted. + let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert!(plan.promote.is_empty()); + assert_eq!(plan.run_status, RunStatus::Failed); + assert!(plan.run_error.is_some()); + } + + #[test] + fn on_error_continue_lets_dependents_proceed_and_run_succeed() { + let mut a = step("a", &[]); + a.on_error = OnError::Continue; + let d = def(vec![a, step("b", &["a"])]); + // a failed but continue → b promoted. + let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert_eq!(plan.promote, vec!["b".to_string()]); + assert_eq!(plan.run_status, RunStatus::Running); + + // a failed-continue + b succeeded → run succeeds (a contributes no output). + let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Succeeded)]); + let mut out = BTreeMap::new(); + out.insert("b".to_string(), json!(42)); + let plan = compute_advance(&d, &st, &out); + assert_eq!(plan.run_status, RunStatus::Succeeded); + assert_eq!(plan.run_output, Some(json!({ "b": 42 }))); + } + + #[test] + fn skipped_dep_satisfies_dependents() { + let d = def(vec![step("a", &[]), step("b", &["a"])]); + let st = statuses(&[("a", StepStatus::Skipped), ("b", StepStatus::Pending)]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert_eq!(plan.promote, vec!["b".to_string()]); + } + + #[test] + fn parallel_fan_out_promotes_all_ready_children() { + // root -> b, root -> c (diamond top). After root succeeds, both promote. + let d = def(vec![ + step("root", &[]), + step("b", &["root"]), + step("c", &["root"]), + ]); + let st = statuses(&[ + ("root", StepStatus::Succeeded), + ("b", StepStatus::Pending), + ("c", StepStatus::Pending), + ]); + let plan = compute_advance(&d, &st, &BTreeMap::new()); + assert_eq!(plan.promote, vec!["b".to_string(), "c".to_string()]); + assert_eq!(plan.run_status, RunStatus::Running); + } +} diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index be69dab..1cc8f8c 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/crates/manager-core/tests/expected_schema.txt @@ -873,7 +873,7 @@ constraints on email_trigger_details: [PRIMARY KEY] email_trigger_details_pkey: PRIMARY KEY (trigger_id) constraints on execution_logs: - [CHECK] execution_logs_source_check: CHECK ((source = ANY (ARRAY['http'::text, 'kv'::text, 'docs'::text, 'dead_letter'::text, 'cron'::text, 'files'::text, 'pubsub'::text, 'email'::text, 'invoke'::text, 'queue'::text]))) + [CHECK] execution_logs_source_check: CHECK ((source = ANY (ARRAY['http'::text, 'kv'::text, 'docs'::text, 'dead_letter'::text, 'cron'::text, 'files'::text, 'pubsub'::text, 'email'::text, 'invoke'::text, 'queue'::text, 'workflow'::text]))) [CHECK] execution_logs_status_check: CHECK ((status = ANY (ARRAY['success'::text, 'error'::text, 'timeout'::text, 'budget_exceeded'::text]))) [FOREIGN KEY] execution_logs_app_id_fk: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE [FOREIGN KEY] execution_logs_script_id_fkey: FOREIGN KEY (script_id) REFERENCES scripts(id) ON DELETE SET NULL @@ -1120,3 +1120,4 @@ constraints on workflows: 0069: email secret version 0070: admin session absolute expiry 0071: workflows + 0072: execution source workflow diff --git a/crates/manager-core/tests/workflow_orchestrator.rs b/crates/manager-core/tests/workflow_orchestrator.rs new file mode 100644 index 0000000..cee4ea1 --- /dev/null +++ b/crates/manager-core/tests/workflow_orchestrator.rs @@ -0,0 +1,537 @@ +//! v1.2 Workflows M2 — durable orchestrator integration tests. +//! +//! Two layers, both DB-gated (skip cleanly when `DATABASE_URL` is unset): +//! +//! * **State machine** — drive `claim_ready_step` → `complete_step_and_advance` +//! in a deterministic loop (no executor, feeding canned outcomes) to pin the +//! durable DAG advance: linear chains, parallel fan-out/fan-in, retry, +//! on_error fail vs continue, stale-token idempotency, and stale-lease +//! reclaim. +//! * **End-to-end** — a real orchestrator `tick()` with a fake `ExecutorClient` +//! and real `scripts` rows, asserting steps resolve + execute + advance to a +//! terminal run. + +#![allow(clippy::too_many_lines)] + +use std::sync::{Arc, LazyLock}; + +use async_trait::async_trait; +use picloud_executor_core::{ExecError, ExecRequest, ExecResponse, ExecStats}; +use picloud_manager_core::repo::PostgresScriptRepository; +use picloud_manager_core::workflow_orchestrator::{ + WorkflowOrchestrator, WorkflowOrchestratorConfig, +}; +use picloud_manager_core::workflow_repo::{ + claim_ready_step, complete_step_and_advance, get_run, list_run_steps, reclaim_stale_steps, + start_run, AdvanceResult, ClaimedStep, NewWorkflowRun, StepOutcome, +}; +use picloud_orchestrator_core::{ExecutionGate, ExecutorClient, ScriptIdentity}; +use picloud_shared::workflow::{ + OnError, RunStatus, StepStatus, WorkflowBackoff, WorkflowDefinition, WorkflowRetry, + WorkflowStepDef, +}; +use picloud_shared::{AppId, ExecutionLogSink, LogSinkError, WorkflowId}; +use serde_json::{json, Value}; +use sqlx::postgres::PgPoolOptions; +use sqlx::PgPool; +use std::time::Duration as StdDuration; +use uuid::Uuid; + +/// `claim_ready_step` is deliberately **global** (one production orchestrator +/// claims every app's ready steps). These tests share one DB, so a sibling +/// test's concurrent claim would steal steps from the run under assertion. This +/// process-wide lock serializes the claim-driving tests — each still creates +/// its own app/workflow/run, and each finishes with no `ready` steps, so once +/// serialized they don't interfere. (See the shared-DB test-harness note.) +static CLAIM_LOCK: LazyLock> = LazyLock::new(|| tokio::sync::Mutex::new(())); + +/// Acquire the serialization lock AND clear any leftover runs/steps so the +/// global `claim_ready_step`/`reclaim_stale_steps` see only this test's rows. +/// (The shared dev DB accumulates rows across runs; `workflow_runs` CASCADEs to +/// its steps.) Returns the held guard. +async fn lock_and_reset(pool: &PgPool) -> tokio::sync::MutexGuard<'static, ()> { + let guard = CLAIM_LOCK.lock().await; + sqlx::query("DELETE FROM workflow_runs") + .execute(pool) + .await + .expect("reset workflow_runs"); + guard +} + +async fn pool_or_skip() -> Option { + let Ok(url) = std::env::var("DATABASE_URL") else { + eprintln!("workflow_orchestrator: DATABASE_URL unset — skipping"); + return None; + }; + let pool = PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect to DATABASE_URL"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("apply migrations"); + Some(pool) +} + +fn step(name: &str, function: &str, deps: &[&str]) -> WorkflowStepDef { + WorkflowStepDef { + name: name.into(), + function: Some(function.into()), + workflow: None, + input: Value::Null, + depends_on: deps.iter().map(|s| (*s).to_string()).collect(), + when: None, + retry: None, + on_error: OnError::Fail, + } +} + +/// Insert an app + a workflow definition; return `(app_id, workflow_id)`. +async fn seed_workflow(pool: &PgPool, def: &WorkflowDefinition) -> (AppId, WorkflowId) { + let sfx = Uuid::new_v4().simple().to_string(); + let grp: (Uuid,) = + sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("wf-grp-{sfx}")) + .fetch_one(pool) + .await + .expect("insert group"); + let app: (Uuid,) = + sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id") + .bind(format!("wf-app-{sfx}")) + .bind(grp.0) + .fetch_one(pool) + .await + .expect("insert app"); + let def_json = serde_json::to_value(def).unwrap(); + let wf: (Uuid,) = sqlx::query_as( + "INSERT INTO workflows (app_id, name, definition) VALUES ($1, $2, $3) RETURNING id", + ) + .bind(app.0) + .bind(format!("wf-{sfx}")) + .bind(def_json) + .fetch_one(pool) + .await + .expect("insert workflow"); + (app.0.into(), wf.0.into()) +} + +fn new_run(app_id: AppId, wf_id: WorkflowId) -> NewWorkflowRun { + NewWorkflowRun { + workflow_id: wf_id, + app_id, + input: json!({ "seed": true }), + root_execution_id: Uuid::new_v4(), + workflow_depth: 0, + parent_run_id: None, + parent_step_id: None, + } +} + +/// Drive the run to a terminal state by claiming + completing steps, deciding +/// each outcome via `decide` (keyed on step name + current attempt). +async fn run_to_terminal(pool: &PgPool, mut decide: F) +where + F: FnMut(&ClaimedStep) -> StepOutcome, +{ + // Bounded so a bug can't spin forever. + for _ in 0..200 { + let Some(claimed) = claim_ready_step(pool).await.expect("claim") else { + break; + }; + let outcome = decide(&claimed); + complete_step_and_advance(pool, &claimed, outcome) + .await + .expect("advance"); + } +} + +fn ok(body: Value) -> StepOutcome { + StepOutcome::Succeeded(body) +} + +fn fail(msg: &str) -> StepOutcome { + StepOutcome::Failed { + error: msg.into(), + retry_delay: chrono::Duration::zero(), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn seeds_roots_ready_and_completes_linear_chain() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // a -> b -> c + let def = WorkflowDefinition { + steps: vec![ + step("a", "fn_a", &[]), + step("b", "fn_b", &["a"]), + step("c", "fn_c", &["b"]), + ], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + // Root `a` starts ready; b, c pending. + let steps = list_run_steps(&pool, 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); + assert_eq!(by("c"), StepStatus::Pending); + + run_to_terminal(&pool, |c| ok(json!({ "step": c.step_name }))).await; + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); + assert_eq!( + r.output, + Some(json!({ + "a": { "step": "a" }, + "b": { "step": "b" }, + "c": { "step": "c" }, + })) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn parallel_fan_out_and_fan_in() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // a -> {b, c} -> d (diamond). + let def = WorkflowDefinition { + steps: vec![ + step("a", "fn_a", &[]), + step("b", "fn_b", &["a"]), + step("c", "fn_c", &["a"]), + step("d", "fn_d", &["b", "c"]), + ], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + // After `a` completes, BOTH b and c must be claimable before d. + let a = claim_ready_step(&pool).await.unwrap().expect("claim a"); + assert_eq!(a.step_name, "a"); + complete_step_and_advance(&pool, &a, ok(json!("a"))) + .await + .unwrap(); + + let s1 = claim_ready_step(&pool) + .await + .unwrap() + .expect("first parallel"); + let s2 = claim_ready_step(&pool) + .await + .unwrap() + .expect("second parallel"); + let mut names = [s1.step_name.clone(), s2.step_name.clone()]; + names.sort(); + assert_eq!(names, ["b".to_string(), "c".to_string()]); + // d must NOT be claimable yet (fan-in waits). + assert!(claim_ready_step(&pool).await.unwrap().is_none()); + + complete_step_and_advance(&pool, &s1, ok(json!("x"))) + .await + .unwrap(); + // Still waiting on the other parallel branch. + assert!(claim_ready_step(&pool).await.unwrap().is_none()); + complete_step_and_advance(&pool, &s2, ok(json!("y"))) + .await + .unwrap(); + + let d = claim_ready_step(&pool) + .await + .unwrap() + .expect("d after fan-in"); + assert_eq!(d.step_name, "d"); + complete_step_and_advance(&pool, &d, ok(json!("d"))) + .await + .unwrap(); + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn retry_then_succeed() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let mut a = step("a", "fn_a", &[]); + a.retry = Some(WorkflowRetry { + max_attempts: 3, + backoff: WorkflowBackoff::Constant, + base_ms: 0, // no delay → immediately re-claimable + }); + let def = WorkflowDefinition { steps: vec![a] }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + // First claim fails (attempt 1) → retried. + let c1 = claim_ready_step(&pool).await.unwrap().expect("attempt 1"); + assert_eq!(c1.attempt, 1); + let res = complete_step_and_advance(&pool, &c1, fail("boom")) + .await + .unwrap(); + assert_eq!(res, AdvanceResult::Retried); + + // Second claim (attempt 2) succeeds → run done. + let c2 = claim_ready_step(&pool).await.unwrap().expect("attempt 2"); + assert_eq!(c2.attempt, 2); + complete_step_and_advance(&pool, &c2, ok(json!("recovered"))) + .await + .unwrap(); + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn on_error_fail_fails_the_run() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // a (fail) -> b; a has no retry and on_error=fail → run fails, b never runs. + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[]), step("b", "fn_b", &["a"])], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + run_to_terminal(&pool, |c| { + if c.step_name == "a" { + fail("kaboom") + } else { + ok(json!("b")) + } + }) + .await; + + 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 b = steps.iter().find(|s| s.step_name == "b").unwrap(); + assert_eq!(b.status, StepStatus::Pending); // never promoted +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn on_error_continue_run_succeeds() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let mut a = step("a", "fn_a", &[]); + a.on_error = OnError::Continue; + let def = WorkflowDefinition { + steps: vec![a, step("b", "fn_b", &["a"])], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + run_to_terminal(&pool, |c| { + if c.step_name == "a" { + fail("soft") + } else { + ok(json!("b-ran")) + } + }) + .await; + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!( + r.status, + RunStatus::Succeeded, + "continue lets the run finish" + ); + assert_eq!(r.output, Some(json!({ "b": "b-ran" }))); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn double_complete_is_idempotent() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[])], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + let c = claim_ready_step(&pool).await.unwrap().expect("claim"); + let first = complete_step_and_advance(&pool, &c, ok(json!("first"))) + .await + .unwrap(); + assert_eq!(first, AdvanceResult::Advanced); + + // Replaying the SAME claim (stale token, lease long gone) writes nothing. + let second = complete_step_and_advance(&pool, &c, ok(json!("SECOND"))) + .await + .unwrap(); + assert_eq!(second, AdvanceResult::Stale); + + let steps = list_run_steps(&pool, 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); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stale_lease_is_reclaimed() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[])], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + let c = claim_ready_step(&pool).await.unwrap().expect("claim"); + // Nothing else claimable while `a` is leased. + assert!(claim_ready_step(&pool).await.unwrap().is_none()); + + // Backdate the lease so it's past the (0s) visibility window. + sqlx::query( + "UPDATE workflow_run_steps SET claimed_at = NOW() - INTERVAL '1 hour' WHERE id = $1", + ) + .bind(c.step_id.into_inner()) + .execute(&pool) + .await + .unwrap(); + let reclaimed = reclaim_stale_steps(&pool, 0).await.unwrap(); + assert_eq!(reclaimed, 1); + + // Now claimable again (a fresh attempt). + let c2 = claim_ready_step(&pool) + .await + .unwrap() + .expect("re-claim after reclaim"); + assert_eq!(c2.step_name, "a"); + complete_step_and_advance(&pool, &c2, ok(json!("done"))) + .await + .unwrap(); + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); +} + +// ---- end-to-end: a real orchestrator tick with a fake executor ------------ + +struct FakeExecutor { + calls: std::sync::Mutex>, +} + +#[async_trait] +impl ExecutorClient for FakeExecutor { + async fn execute( + &self, + _source: &str, + req: ExecRequest, + _timeout: StdDuration, + ) -> Result { + self.calls.lock().unwrap().push(req.script_name.clone()); + Ok(ExecResponse { + status_code: 200, + headers: std::collections::BTreeMap::new(), + body: json!({ "ran": req.script_name }), + logs: vec![], + stats: ExecStats::default(), + }) + } + + async fn execute_with_identity( + &self, + _identity: ScriptIdentity, + source: &str, + req: ExecRequest, + timeout: StdDuration, + ) -> Result { + self.execute(source, req, timeout).await + } +} + +struct NoopLogSink; +#[async_trait] +impl ExecutionLogSink for NoopLogSink { + async fn record(&self, _log: picloud_shared::ExecutionLog) -> Result<(), LogSinkError> { + Ok(()) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn end_to_end_tick_executes_functions_and_completes() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // a -> b; both bound to real scripts (fn_a, fn_b) so name resolution works. + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[]), step("b", "fn_b", &["a"])], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + for fn_name in ["fn_a", "fn_b"] { + sqlx::query("INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2)") + .bind(fn_name) + .bind(app.into_inner()) + .execute(&pool) + .await + .expect("insert script"); + } + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + let executor = Arc::new(FakeExecutor { + calls: std::sync::Mutex::new(vec![]), + }); + let orch = WorkflowOrchestrator { + pool: pool.clone(), + scripts: Arc::new(PostgresScriptRepository::new(pool.clone())), + executor: executor.clone(), + gate: Arc::new(ExecutionGate::new(8)), + log_sink: Arc::new(NoopLogSink), + config: WorkflowOrchestratorConfig::default(), + }; + + // Drive ticks until the run is terminal (bounded). + for _ in 0..50 { + orch.tick_once().await; + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + if r.status.is_terminal() { + break; + } + } + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); + let calls = executor.calls.lock().unwrap().clone(); + assert!(calls.contains(&"fn_a".to_string())); + assert!(calls.contains(&"fn_b".to_string())); + assert_eq!( + r.output, + Some(json!({ "a": { "ran": "fn_a" }, "b": { "ran": "fn_b" } })) + ); +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index b8b21e8..99b0a24 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -486,7 +486,7 @@ pub async fn build_app( abandoned: abandoned_repo.clone(), principals, executor: executor.clone(), - gate, + gate: gate.clone(), log_sink: log_sink.clone(), inbox: inbox_resolver, queue: queue_repo.clone(), @@ -496,6 +496,20 @@ pub async fn build_app( } .spawn(); + // v1.2 Workflows (M2): the durable workflow orchestrator. Claims ready + // steps of in-flight runs, executes each through the same executor + gate + // the dispatcher uses, and advances the DAG — all durably, surviving + // restarts. Spawned here (before `executor`/`gate` are moved into the data + // plane) so it shares those Arcs. + picloud_manager_core::spawn_workflow_orchestrator(picloud_manager_core::WorkflowOrchestrator { + pool: pool.clone(), + scripts: script_repo.clone(), + executor: executor.clone(), + gate: gate.clone(), + log_sink: log_sink.clone(), + config: picloud_manager_core::workflow_orchestrator::WorkflowOrchestratorConfig::from_env(), + }); + let admin = AdminState { repo: script_repo.clone(), logs: log_repo, diff --git a/crates/shared/src/execution_log.rs b/crates/shared/src/execution_log.rs index 03b36d8..8e7267e 100644 --- a/crates/shared/src/execution_log.rs +++ b/crates/shared/src/execution_log.rs @@ -62,6 +62,7 @@ pub enum ExecutionSource { Email, Invoke, Queue, + Workflow, } impl ExecutionSource { @@ -78,6 +79,7 @@ impl ExecutionSource { Self::Email => "email", Self::Invoke => "invoke", Self::Queue => "queue", + Self::Workflow => "workflow", } } @@ -96,6 +98,7 @@ impl ExecutionSource { "email" => Some(Self::Email), "invoke" => Some(Self::Invoke), "queue" => Some(Self::Queue), + "workflow" => Some(Self::Workflow), _ => None, } } diff --git a/crates/shared/src/workflow.rs b/crates/shared/src/workflow.rs index 8f84662..990d940 100644 --- a/crates/shared/src/workflow.rs +++ b/crates/shared/src/workflow.rs @@ -181,11 +181,20 @@ impl StepStatus { /// A dependency is "satisfied" (a dependent may proceed) once it has /// succeeded OR been skipped (a false `when` — satisfied-but-empty). + /// NB: a `failed` step whose `on_error = continue` also satisfies its + /// dependents, but that depends on the step's *definition*, not its + /// status alone — the orchestrator's advance logic folds that in. #[must_use] pub fn is_satisfied(self) -> bool { matches!(self, Self::Succeeded | Self::Skipped) } + /// A step has reached a terminal status (no further transitions). + #[must_use] + pub fn is_terminal(self) -> bool { + matches!(self, Self::Succeeded | Self::Failed | Self::Skipped) + } + #[must_use] #[allow(clippy::should_implement_trait)] pub fn from_str(s: &str) -> Option {