feat: add executor tracing, auto-register primitives, and Default impls

- Add info!-level tracing to workflow executor: logs each execution
  round, each step run (with type and name), step completion, and
  workflow completion
- WorkflowHost.start() now auto-registers all built-in primitive step
  types so users don't need to register them manually
- Add #[derive(Default)] to all primitive steps and PollEndpointConfig
- Add tracing-subscriber to wfe crate for the pizza example
- Pizza example now shows full step-by-step execution logs
This commit is contained in:
2026-03-25 20:32:47 +00:00
parent 6d57f8ef22
commit 88fc6bf7ad
13 changed files with 70 additions and 3 deletions

View File

@@ -1,7 +1,7 @@
use std::sync::Arc;
use chrono::Utc;
use tracing::{debug, error, warn};
use tracing::{debug, error, info, warn};
use super::error_handler;
use super::result_processor;
@@ -106,6 +106,12 @@ impl WorkflowExecutor {
let mut execution_errors = Vec::new();
// 3. Find runnable execution pointers.
info!(
workflow_id,
definition_id = %workflow.workflow_definition_id,
pointers = workflow.execution_pointers.len(),
"Executing workflow"
);
let runnable_indices: Vec<usize> = workflow
.execution_pointers
.iter()
@@ -125,6 +131,14 @@ impl WorkflowExecutor {
.find(|s| s.id == step_id)
.ok_or(WfeError::StepNotFound(step_id))?;
info!(
workflow_id,
step_id,
step_type = %step.step_type,
step_name = step.name.as_deref().unwrap_or("(unnamed)"),
"Running step"
);
// b. Resolve the step body.
let mut step_body = step_registry
.resolve(&step.step_type)
@@ -156,6 +170,15 @@ impl WorkflowExecutor {
// Now we can mutate again since context is dropped.
match step_result {
Ok(result) => {
info!(
workflow_id,
step_id,
proceed = result.proceed,
has_sleep = result.sleep_for.is_some(),
has_event = result.event_name.is_some(),
has_branches = result.branch_values.is_some(),
"Step completed"
);
// e. Process the ExecutionResult.
// Extract workflow_id before mutable borrow.
let wf_id = workflow.id.clone();
@@ -225,6 +248,7 @@ impl WorkflowExecutor {
});
if all_done && workflow.status == WorkflowStatus::Runnable {
info!(workflow_id, "All pointers complete, workflow finished");
workflow.status = WorkflowStatus::Complete;
workflow.complete_time = Some(Utc::now());
}

View File

@@ -3,8 +3,9 @@ use std::time::Duration;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum HttpMethod {
#[default]
Get,
Post,
Put,
@@ -25,7 +26,13 @@ pub enum PollCondition {
BodyContains(String),
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
impl Default for PollCondition {
fn default() -> Self {
Self::StatusCode(200)
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct PollEndpointConfig {
/// URL template. Supports `{placeholder}` interpolation from workflow data.
pub url: String,

View File

@@ -4,6 +4,7 @@ use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};
/// A decision step that returns an outcome value for routing.
#[derive(Default)]
pub struct DecideStep {
pub expression_value: serde_json::Value,
}

View File

@@ -4,6 +4,7 @@ use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};
/// A no-op marker step indicating the end of a workflow branch.
#[derive(Default)]
pub struct EndStep;
#[async_trait]

View File

@@ -5,6 +5,7 @@ use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};
/// A conditional step that branches execution based on a boolean condition.
#[derive(Default)]
pub struct IfStep {
pub condition: bool,
}

View File

@@ -6,6 +6,7 @@ use crate::traits::step::{StepBody, StepExecutionContext};
/// A step that polls an external HTTP endpoint until a condition is met.
/// The actual HTTP polling is handled by the executor, not this step.
#[derive(Default)]
pub struct PollEndpointStep {
pub config: PollEndpointConfig,
}

View File

@@ -8,6 +8,7 @@ use crate::traits::step::{StepBody, StepExecutionContext};
/// A step that repeatedly schedules child execution at an interval
/// until a stop condition is met.
#[derive(Default)]
pub struct RecurStep {
pub interval: Duration,
pub stop_condition: bool,

View File

@@ -7,6 +7,7 @@ use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};
/// A step that schedules child execution after a delay.
#[derive(Default)]
pub struct ScheduleStep {
pub interval: Duration,
}

View File

@@ -6,6 +6,7 @@ use crate::traits::step::{StepBody, StepExecutionContext};
/// A container step that executes its children sequentially.
/// Completes when all children have finished.
#[derive(Default)]
pub struct SequenceStep;
#[async_trait]

View File

@@ -5,6 +5,7 @@ use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};
/// A looping step that repeats its children while a condition is true.
#[derive(Default)]
pub struct WhileStep {
pub condition: bool,
}