[smg] import official wfaas crate to replace self managed one (#17724)
This commit is contained in:
Simo Lin
@@ -85,6 +85,7 @@ openai-protocol = { version = "1.0.0", features = ["axum"] }
|
||||
tool-parser = "1.0.0"
|
||||
llm-tokenizer = "1.0.0"
|
||||
smg-auth = "1.0.0"
|
||||
wfaas = "1.0.0"
|
||||
rustls = { version = "0.23", default-features = false, features = ["ring", "std"] }
|
||||
rustls-pemfile = "2.2"
|
||||
openssl = "0.10.73"
|
||||
@@ -153,10 +154,6 @@ tonic-v12 = { version = "0.12.3", package = "tonic" }
|
||||
serial_test = "3.0"
|
||||
rsa = { version = "0.9", features = ["sha2"] }
|
||||
|
||||
[[bench]]
|
||||
name = "special_token_search"
|
||||
harness = false
|
||||
path = "benches/special_token_search.rs"
|
||||
[[bench]]
|
||||
name = "wasm_middleware_latency"
|
||||
harness = false
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
use aho_corasick::AhoCorasick;
|
||||
use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
|
||||
|
||||
fn find_special_token_boundaries_naive(
|
||||
text: &str,
|
||||
special_tokens: &[String],
|
||||
) -> Vec<(usize, usize)> {
|
||||
let mut boundaries = Vec::new();
|
||||
for token in special_tokens {
|
||||
let mut start = 0;
|
||||
while let Some(pos) = text[start..].find(token) {
|
||||
let actual_pos = start + pos;
|
||||
boundaries.push((actual_pos, actual_pos + token.len()));
|
||||
start = actual_pos + token.len();
|
||||
}
|
||||
}
|
||||
boundaries.sort_by_key(|b| b.0);
|
||||
boundaries
|
||||
}
|
||||
|
||||
fn find_special_token_boundaries_aho(text: &str, ac: &AhoCorasick) -> Vec<(usize, usize)> {
|
||||
ac.find_iter(text)
|
||||
.map(|mat| (mat.start(), mat.end()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn bench_token_search_comparison(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("token_boundary_search");
|
||||
let text = "User: Hello! Assistant: How can I help you today? ".repeat(1000);
|
||||
|
||||
for token_count in [5, 50] {
|
||||
let special_tokens: Vec<String> = (0..token_count)
|
||||
.map(|i| format!("<|stop_sequence_{}|>", i))
|
||||
.collect();
|
||||
|
||||
let ac = AhoCorasick::new(&special_tokens).unwrap();
|
||||
group.throughput(Throughput::Bytes(text.len() as u64));
|
||||
|
||||
group.bench_function(format!("naive_tokens_{}", token_count), |b| {
|
||||
b.iter(|| {
|
||||
find_special_token_boundaries_naive(black_box(&text), black_box(&special_tokens))
|
||||
})
|
||||
});
|
||||
|
||||
group.bench_function(format!("aho_tokens_{}", token_count), |b| {
|
||||
b.iter(|| find_special_token_boundaries_aho(black_box(&text), black_box(&ac)))
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(benches, bench_token_search_comparison);
|
||||
criterion_main!(benches);
|
||||
@@ -19,4 +19,4 @@ pub use llm_tokenizer as tokenizer;
|
||||
pub use tool_parser;
|
||||
pub mod version;
|
||||
pub mod wasm;
|
||||
pub mod workflow;
|
||||
pub use wfaas as workflow;
|
||||
|
||||
@@ -1,364 +0,0 @@
|
||||
//! Workflow definition types
|
||||
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
fmt,
|
||||
sync::Arc,
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
|
||||
use super::{
|
||||
executor::StepExecutor,
|
||||
types::{FailureAction, RetryPolicy, StepId, WorkflowContext, WorkflowData, WorkflowId},
|
||||
};
|
||||
|
||||
/// A condition function that determines whether a step should run.
|
||||
pub type StepCondition<D> = Arc<dyn Fn(&WorkflowContext<D>) -> bool + Send + Sync>;
|
||||
|
||||
/// Errors that can occur during workflow validation
|
||||
#[derive(Debug, Clone, thiserror::Error)]
|
||||
pub enum ValidationError {
|
||||
/// A step depends on another step that doesn't exist
|
||||
#[error("Step '{step}' depends on non-existent step '{dependency}'")]
|
||||
MissingDependency { step: StepId, dependency: StepId },
|
||||
|
||||
/// A cycle was detected in the workflow DAG
|
||||
#[error("Cycle detected involving step '{0}'")]
|
||||
CycleDetected(StepId),
|
||||
}
|
||||
|
||||
/// Definition of a single step within a workflow
|
||||
pub struct StepDefinition<D: WorkflowData> {
|
||||
pub id: StepId,
|
||||
pub name: String,
|
||||
pub executor: Arc<dyn StepExecutor<D>>,
|
||||
pub retry_policy: Option<RetryPolicy>,
|
||||
pub timeout: Option<Duration>,
|
||||
pub on_failure: FailureAction,
|
||||
/// Dependencies that must ALL complete before this step runs
|
||||
pub depends_on: Vec<StepId>,
|
||||
/// Dependencies where ANY completing triggers this step (used with depends_on)
|
||||
pub depends_on_any: Vec<StepId>,
|
||||
/// Delay before starting the step (after dependencies satisfied)
|
||||
pub delay: Option<Duration>,
|
||||
/// Run step at or after this time (after dependencies satisfied)
|
||||
pub scheduled_at: Option<DateTime<Utc>>,
|
||||
/// Condition to evaluate; if false, step is skipped
|
||||
pub run_if: Option<StepCondition<D>>,
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> fmt::Debug for StepDefinition<D> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("StepDefinition")
|
||||
.field("id", &self.id)
|
||||
.field("name", &self.name)
|
||||
.field("retry_policy", &self.retry_policy)
|
||||
.field("timeout", &self.timeout)
|
||||
.field("on_failure", &self.on_failure)
|
||||
.field("depends_on", &self.depends_on)
|
||||
.field("depends_on_any", &self.depends_on_any)
|
||||
.field("delay", &self.delay)
|
||||
.field("scheduled_at", &self.scheduled_at)
|
||||
.field("run_if", &self.run_if.as_ref().map(|_| "<condition>"))
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> StepDefinition<D> {
|
||||
pub fn new(
|
||||
id: impl Into<String>,
|
||||
name: impl Into<String>,
|
||||
executor: Arc<dyn StepExecutor<D>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
id: StepId::new(id.into()),
|
||||
name: name.into(),
|
||||
executor,
|
||||
retry_policy: None,
|
||||
timeout: None,
|
||||
on_failure: FailureAction::FailWorkflow,
|
||||
depends_on: Vec::new(),
|
||||
depends_on_any: Vec::new(),
|
||||
delay: None,
|
||||
scheduled_at: None,
|
||||
run_if: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_retry(mut self, policy: RetryPolicy) -> Self {
|
||||
self.retry_policy = Some(policy);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.timeout = Some(timeout);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_failure_action(mut self, action: FailureAction) -> Self {
|
||||
self.on_failure = action;
|
||||
self
|
||||
}
|
||||
|
||||
/// Set dependencies for this step.
|
||||
/// The step will only run after ALL specified dependencies have completed successfully.
|
||||
/// Empty slice means no dependencies - step can run immediately in parallel with others.
|
||||
pub fn depends_on(mut self, deps: &[&str]) -> Self {
|
||||
self.depends_on = deps.iter().map(|s| StepId::new(*s)).collect();
|
||||
self
|
||||
}
|
||||
|
||||
/// Set "any of" dependencies for this step.
|
||||
/// The step will run when ANY of these dependencies complete (in addition to
|
||||
/// all `depends_on` dependencies).
|
||||
///
|
||||
/// **Combined semantics** (when both `depends_on` and `depends_on_any` are set):
|
||||
/// - ALL `depends_on` must complete successfully, AND
|
||||
/// - AT LEAST ONE `depends_on_any` must complete successfully
|
||||
///
|
||||
/// **Failure handling**:
|
||||
/// - For `depends_on`: if ANY fails, this step is blocked
|
||||
/// - For `depends_on_any`: only blocked if ALL fail (since we only need one)
|
||||
///
|
||||
/// **Skipped dependencies**: A skipped dependency (e.g., via `run_if`) counts as
|
||||
/// "completed" for dependency satisfaction purposes.
|
||||
pub fn depends_on_any(mut self, deps: &[&str]) -> Self {
|
||||
self.depends_on_any = deps.iter().map(|s| StepId::new(*s)).collect();
|
||||
self
|
||||
}
|
||||
|
||||
/// Set a delay before starting the step (after dependencies are satisfied).
|
||||
///
|
||||
/// If both `delay` and `scheduled_at` are set, the step will wait for the
|
||||
/// scheduled time AND THEN wait for the delay duration (they stack).
|
||||
pub fn with_delay(mut self, delay: Duration) -> Self {
|
||||
self.delay = Some(delay);
|
||||
self
|
||||
}
|
||||
|
||||
/// Schedule the step to run at or after the specified time.
|
||||
/// The step will wait until this time even if dependencies are satisfied earlier.
|
||||
///
|
||||
/// If the scheduled time is in the past when the step becomes ready, it will
|
||||
/// proceed immediately (a debug log is emitted).
|
||||
///
|
||||
/// If both `delay` and `scheduled_at` are set, the step will wait for the
|
||||
/// scheduled time AND THEN wait for the delay duration (they stack).
|
||||
pub fn scheduled_at(mut self, time: DateTime<Utc>) -> Self {
|
||||
self.scheduled_at = Some(time);
|
||||
self
|
||||
}
|
||||
|
||||
/// Set a condition for running this step.
|
||||
/// If the condition returns false, the step is skipped.
|
||||
///
|
||||
/// **Skipped step semantics**: When a step is skipped:
|
||||
/// - It is marked as `StepStatus::Skipped`
|
||||
/// - Downstream steps that depend on it (via `depends_on`) will consider it satisfied
|
||||
/// - This allows conditional branches without blocking the workflow
|
||||
///
|
||||
/// **Error handling**: If the context cannot be retrieved to evaluate the condition,
|
||||
/// the step will **fail** (not proceed blindly). This is a safety measure.
|
||||
///
|
||||
/// **Note**: The condition closure is not serializable, so workflows with `run_if`
|
||||
/// cannot be persisted and resumed from external storage.
|
||||
pub fn run_if<F>(mut self, condition: F) -> Self
|
||||
where
|
||||
F: Fn(&WorkflowContext<D>) -> bool + Send + Sync + 'static,
|
||||
{
|
||||
self.run_if = Some(Arc::new(condition));
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
/// Complete workflow definition
|
||||
pub struct WorkflowDefinition<D: WorkflowData> {
|
||||
pub id: WorkflowId,
|
||||
pub name: String,
|
||||
pub steps: Vec<StepDefinition<D>>,
|
||||
pub default_retry_policy: RetryPolicy,
|
||||
pub default_timeout: Duration,
|
||||
/// Pre-computed reverse dependencies: step_id -> indices of steps that depend on it
|
||||
reverse_deps: HashMap<StepId, Vec<usize>>,
|
||||
/// Pre-computed indices of steps with no dependencies (can start immediately)
|
||||
initial_step_indices: Vec<usize>,
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> fmt::Debug for WorkflowDefinition<D> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("WorkflowDefinition")
|
||||
.field("id", &self.id)
|
||||
.field("name", &self.name)
|
||||
.field("steps", &self.steps)
|
||||
.field("default_retry_policy", &self.default_retry_policy)
|
||||
.field("default_timeout", &self.default_timeout)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> WorkflowDefinition<D> {
|
||||
pub fn new(id: impl Into<String>, name: impl Into<String>) -> Self {
|
||||
Self {
|
||||
id: WorkflowId::new(id.into()),
|
||||
name: name.into(),
|
||||
steps: Vec::new(),
|
||||
default_retry_policy: RetryPolicy::default(),
|
||||
default_timeout: Duration::from_secs(300), // 5 minutes
|
||||
reverse_deps: HashMap::new(),
|
||||
initial_step_indices: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_step(mut self, step: StepDefinition<D>) -> Self {
|
||||
self.steps.push(step);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_default_retry(mut self, policy: RetryPolicy) -> Self {
|
||||
self.default_retry_policy = policy;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_default_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.default_timeout = timeout;
|
||||
self
|
||||
}
|
||||
|
||||
/// Get the retry policy for a step (step-specific or default)
|
||||
pub fn get_retry_policy<'a>(&'a self, step: &'a StepDefinition<D>) -> &'a RetryPolicy {
|
||||
step.retry_policy
|
||||
.as_ref()
|
||||
.unwrap_or(&self.default_retry_policy)
|
||||
}
|
||||
|
||||
/// Get the timeout for a step (step-specific or default)
|
||||
pub fn get_timeout(&self, step: &StepDefinition<D>) -> Duration {
|
||||
step.timeout.unwrap_or(self.default_timeout)
|
||||
}
|
||||
|
||||
/// Validate the workflow DAG structure and build dependency graph.
|
||||
/// Returns an error if:
|
||||
/// - A step depends on a non-existent step
|
||||
/// - There's a cycle in the dependencies
|
||||
///
|
||||
/// On success, pre-computes reverse dependencies for O(1) dependent lookup.
|
||||
#[must_use = "validation result should be checked"]
|
||||
pub fn validate(&mut self) -> Result<(), ValidationError> {
|
||||
// Build HashMap for O(1) lookup instead of O(n) linear search
|
||||
let steps_map: HashMap<&StepId, &StepDefinition<D>> =
|
||||
self.steps.iter().map(|s| (&s.id, s)).collect();
|
||||
|
||||
// Check all dependencies exist (both depends_on and depends_on_any)
|
||||
for step in &self.steps {
|
||||
for dep in &step.depends_on {
|
||||
if !steps_map.contains_key(dep) {
|
||||
return Err(ValidationError::MissingDependency {
|
||||
step: step.id.clone(),
|
||||
dependency: dep.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
for dep in &step.depends_on_any {
|
||||
if !steps_map.contains_key(dep) {
|
||||
return Err(ValidationError::MissingDependency {
|
||||
step: step.id.clone(),
|
||||
dependency: dep.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check for cycles using DFS (considers both dependency types)
|
||||
let mut visited = HashSet::new();
|
||||
let mut rec_stack = HashSet::new();
|
||||
|
||||
for step in &self.steps {
|
||||
if !visited.contains(&step.id)
|
||||
&& Self::has_cycle(&step.id, &steps_map, &mut visited, &mut rec_stack)
|
||||
{
|
||||
return Err(ValidationError::CycleDetected(step.id.clone()));
|
||||
}
|
||||
}
|
||||
|
||||
// Build reverse dependency map: for each step, which steps depend on it?
|
||||
// Include both depends_on and depends_on_any
|
||||
self.reverse_deps.clear();
|
||||
for (idx, step) in self.steps.iter().enumerate() {
|
||||
for dep_id in &step.depends_on {
|
||||
self.reverse_deps
|
||||
.entry(dep_id.clone())
|
||||
.or_default()
|
||||
.push(idx);
|
||||
}
|
||||
for dep_id in &step.depends_on_any {
|
||||
self.reverse_deps
|
||||
.entry(dep_id.clone())
|
||||
.or_default()
|
||||
.push(idx);
|
||||
}
|
||||
}
|
||||
|
||||
// Cache indices of steps with no dependencies (can start immediately)
|
||||
// A step with only depends_on_any still needs at least one to complete
|
||||
self.initial_step_indices = self
|
||||
.steps
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, s)| s.depends_on.is_empty() && s.depends_on_any.is_empty())
|
||||
.map(|(i, _)| i)
|
||||
.collect();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// DFS helper for cycle detection with O(1) HashMap lookup
|
||||
fn has_cycle<'a>(
|
||||
step_id: &'a StepId,
|
||||
steps_map: &HashMap<&'a StepId, &'a StepDefinition<D>>,
|
||||
visited: &mut HashSet<&'a StepId>,
|
||||
rec_stack: &mut HashSet<&'a StepId>,
|
||||
) -> bool {
|
||||
if rec_stack.contains(step_id) {
|
||||
return true; // Back edge found - cycle!
|
||||
}
|
||||
if visited.contains(step_id) {
|
||||
return false; // Already fully processed
|
||||
}
|
||||
|
||||
visited.insert(step_id);
|
||||
rec_stack.insert(step_id);
|
||||
|
||||
// O(1) lookup instead of linear search
|
||||
// Check both depends_on and depends_on_any for cycles
|
||||
if let Some(step) = steps_map.get(step_id) {
|
||||
for dep in &step.depends_on {
|
||||
if Self::has_cycle(dep, steps_map, visited, rec_stack) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
for dep in &step.depends_on_any {
|
||||
if Self::has_cycle(dep, steps_map, visited, rec_stack) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
rec_stack.remove(step_id);
|
||||
false
|
||||
}
|
||||
|
||||
/// Get indices of steps that depend on the given step
|
||||
pub fn get_dependent_indices(&self, step_id: &StepId) -> &[usize] {
|
||||
self.reverse_deps
|
||||
.get(step_id)
|
||||
.map(|v| v.as_slice())
|
||||
.unwrap_or(&[])
|
||||
}
|
||||
|
||||
/// Get indices of steps with no dependencies
|
||||
pub fn get_initial_step_indices(&self) -> &[usize] {
|
||||
&self.initial_step_indices
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,282 +0,0 @@
|
||||
//! Workflow event system for observability and monitoring
|
||||
|
||||
use std::{sync::Arc, time::Duration};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use super::types::{StepId, WorkflowId, WorkflowInstanceId};
|
||||
|
||||
/// Default timeout for subscriber event handlers
|
||||
const DEFAULT_SUBSCRIBER_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Events emitted by the workflow engine
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum WorkflowEvent {
|
||||
WorkflowStarted {
|
||||
instance_id: WorkflowInstanceId,
|
||||
definition_id: WorkflowId,
|
||||
},
|
||||
StepStarted {
|
||||
instance_id: WorkflowInstanceId,
|
||||
step_id: StepId,
|
||||
attempt: u32,
|
||||
},
|
||||
StepSucceeded {
|
||||
instance_id: WorkflowInstanceId,
|
||||
step_id: StepId,
|
||||
duration: Duration,
|
||||
},
|
||||
StepFailed {
|
||||
instance_id: WorkflowInstanceId,
|
||||
step_id: StepId,
|
||||
error: String,
|
||||
will_retry: bool,
|
||||
},
|
||||
StepRetrying {
|
||||
instance_id: WorkflowInstanceId,
|
||||
step_id: StepId,
|
||||
attempt: u32,
|
||||
delay: Duration,
|
||||
},
|
||||
WorkflowCompleted {
|
||||
instance_id: WorkflowInstanceId,
|
||||
duration: Duration,
|
||||
},
|
||||
WorkflowFailed {
|
||||
instance_id: WorkflowInstanceId,
|
||||
failed_step: StepId,
|
||||
error: String,
|
||||
},
|
||||
WorkflowCancelled {
|
||||
instance_id: WorkflowInstanceId,
|
||||
},
|
||||
}
|
||||
|
||||
/// Trait for subscribing to workflow events
|
||||
#[async_trait]
|
||||
pub trait EventSubscriber: Send + Sync {
|
||||
async fn on_event(&self, event: &WorkflowEvent);
|
||||
}
|
||||
|
||||
/// Event bus for publishing and subscribing to workflow events
|
||||
///
|
||||
/// # Subscriber Isolation
|
||||
///
|
||||
/// Each subscriber is notified in a separate spawned task with a timeout.
|
||||
/// This ensures that:
|
||||
/// - A slow subscriber doesn't block other subscribers
|
||||
/// - A panicking subscriber doesn't affect other subscribers
|
||||
/// - Event publishing returns quickly regardless of subscriber behavior
|
||||
pub struct EventBus {
|
||||
subscribers: Arc<RwLock<Vec<Arc<dyn EventSubscriber>>>>,
|
||||
/// Timeout for each subscriber's event handler
|
||||
subscriber_timeout: Duration,
|
||||
}
|
||||
|
||||
impl EventBus {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
subscribers: Arc::new(RwLock::new(Vec::new())),
|
||||
subscriber_timeout: DEFAULT_SUBSCRIBER_TIMEOUT,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create an EventBus with a custom subscriber timeout
|
||||
pub fn with_timeout(timeout: Duration) -> Self {
|
||||
Self {
|
||||
subscribers: Arc::new(RwLock::new(Vec::new())),
|
||||
subscriber_timeout: timeout,
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribe to workflow events
|
||||
pub async fn subscribe(&self, subscriber: Arc<dyn EventSubscriber>) {
|
||||
self.subscribers.write().await.push(subscriber);
|
||||
}
|
||||
|
||||
/// Unsubscribe from workflow events
|
||||
///
|
||||
/// Removes the subscriber by Arc pointer equality.
|
||||
/// Returns true if the subscriber was found and removed.
|
||||
pub async fn unsubscribe(&self, subscriber: &Arc<dyn EventSubscriber>) -> bool {
|
||||
let mut subs = self.subscribers.write().await;
|
||||
let len_before = subs.len();
|
||||
subs.retain(|s| !Arc::ptr_eq(s, subscriber));
|
||||
subs.len() < len_before
|
||||
}
|
||||
|
||||
/// Publish an event to all subscribers concurrently
|
||||
///
|
||||
/// Each subscriber is notified in a separate spawned task with a timeout.
|
||||
/// This method returns after spawning all notification tasks, without
|
||||
/// waiting for subscribers to complete (fire-and-forget).
|
||||
///
|
||||
/// Subscriber failures (timeout or panic) are logged but don't affect
|
||||
/// other subscribers or the caller.
|
||||
pub async fn publish(&self, event: WorkflowEvent) {
|
||||
let subscribers: Vec<_> = self.subscribers.read().await.iter().cloned().collect();
|
||||
let timeout = self.subscriber_timeout;
|
||||
|
||||
for (idx, subscriber) in subscribers.into_iter().enumerate() {
|
||||
let event = event.clone();
|
||||
tokio::spawn(async move {
|
||||
let result = tokio::time::timeout(timeout, subscriber.on_event(&event)).await;
|
||||
match result {
|
||||
Ok(()) => {}
|
||||
Err(_) => {
|
||||
warn!(
|
||||
subscriber_index = idx,
|
||||
timeout_secs = timeout.as_secs(),
|
||||
"Event subscriber timed out"
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Publish an event and wait for all subscribers to complete
|
||||
///
|
||||
/// Unlike `publish`, this method waits for all subscribers to finish
|
||||
/// (or timeout). Use this when you need to ensure all subscribers
|
||||
/// have processed the event before continuing.
|
||||
pub async fn publish_and_wait(&self, event: WorkflowEvent) {
|
||||
let subscribers: Vec<_> = self.subscribers.read().await.iter().cloned().collect();
|
||||
let timeout = self.subscriber_timeout;
|
||||
|
||||
let handles: Vec<_> = subscribers
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(idx, subscriber)| {
|
||||
let event = event.clone();
|
||||
tokio::spawn(async move {
|
||||
let result = tokio::time::timeout(timeout, subscriber.on_event(&event)).await;
|
||||
if result.is_err() {
|
||||
warn!(
|
||||
subscriber_index = idx,
|
||||
timeout_secs = timeout.as_secs(),
|
||||
"Event subscriber timed out"
|
||||
);
|
||||
}
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Wait for all spawned tasks, ignoring individual failures (panics)
|
||||
for handle in handles {
|
||||
let _ = handle.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for EventBus {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for EventBus {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("EventBus").finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
/// Logging subscriber that logs events using tracing
|
||||
pub struct LoggingSubscriber;
|
||||
|
||||
#[async_trait]
|
||||
impl EventSubscriber for LoggingSubscriber {
|
||||
async fn on_event(&self, event: &WorkflowEvent) {
|
||||
match event {
|
||||
WorkflowEvent::WorkflowStarted {
|
||||
instance_id,
|
||||
definition_id,
|
||||
} => {
|
||||
info!(
|
||||
instance_id = %instance_id,
|
||||
definition_id = %definition_id,
|
||||
"Workflow started"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::StepStarted {
|
||||
instance_id,
|
||||
step_id,
|
||||
attempt,
|
||||
} => {
|
||||
info!(
|
||||
instance_id = %instance_id,
|
||||
step_id = %step_id,
|
||||
attempt = attempt,
|
||||
"Step started"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::StepSucceeded {
|
||||
instance_id,
|
||||
step_id,
|
||||
duration,
|
||||
} => {
|
||||
info!(
|
||||
instance_id = %instance_id,
|
||||
step_id = %step_id,
|
||||
duration_ms = duration.as_millis(),
|
||||
"Step succeeded"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::StepFailed {
|
||||
instance_id,
|
||||
step_id,
|
||||
error,
|
||||
will_retry,
|
||||
} => {
|
||||
warn!(
|
||||
instance_id = %instance_id,
|
||||
step_id = %step_id,
|
||||
error = error,
|
||||
will_retry = will_retry,
|
||||
"Step failed"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::StepRetrying {
|
||||
instance_id,
|
||||
step_id,
|
||||
attempt,
|
||||
delay,
|
||||
} => {
|
||||
info!(
|
||||
instance_id = %instance_id,
|
||||
step_id = %step_id,
|
||||
attempt = attempt,
|
||||
delay_ms = delay.as_millis(),
|
||||
"Step retrying"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::WorkflowCompleted {
|
||||
instance_id,
|
||||
duration,
|
||||
} => {
|
||||
info!(
|
||||
instance_id = %instance_id,
|
||||
duration_ms = duration.as_millis(),
|
||||
"Workflow completed"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::WorkflowFailed {
|
||||
instance_id,
|
||||
failed_step,
|
||||
error,
|
||||
} => {
|
||||
error!(
|
||||
instance_id = %instance_id,
|
||||
failed_step = %failed_step,
|
||||
error = error,
|
||||
"Workflow failed"
|
||||
);
|
||||
}
|
||||
WorkflowEvent::WorkflowCancelled { instance_id } => {
|
||||
info!(instance_id = %instance_id, "Workflow cancelled");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,152 +0,0 @@
|
||||
//! Step executor trait and implementations
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use super::types::{StepResult, WorkflowContext, WorkflowData, WorkflowError, WorkflowResult};
|
||||
|
||||
/// Trait for executing individual workflow steps
|
||||
#[async_trait]
|
||||
pub trait StepExecutor<D: WorkflowData>: Send + Sync {
|
||||
/// Execute the step with the given context
|
||||
async fn execute(&self, context: &mut WorkflowContext<D>) -> WorkflowResult<StepResult>;
|
||||
|
||||
/// Check if an error is retry-able
|
||||
///
|
||||
/// Override this method to customize which errors should trigger retries.
|
||||
/// By default, all errors are considered retry-able.
|
||||
fn is_retryable(&self, _error: &WorkflowError) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Called when the step succeeds
|
||||
///
|
||||
/// This hook allows steps to perform cleanup or additional actions
|
||||
/// after successful execution.
|
||||
async fn on_success(&self, _context: &WorkflowContext<D>) -> WorkflowResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Called when the step fails after all retries
|
||||
///
|
||||
/// This hook allows steps to perform cleanup or compensation logic
|
||||
/// when the step cannot complete successfully.
|
||||
async fn on_failure(
|
||||
&self,
|
||||
_context: &WorkflowContext<D>,
|
||||
_error: &WorkflowError,
|
||||
) -> WorkflowResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Simple function-based step executor
|
||||
pub struct FunctionStep<D, F>
|
||||
where
|
||||
D: WorkflowData,
|
||||
F: Fn(
|
||||
&mut WorkflowContext<D>,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = WorkflowResult<StepResult>> + Send + '_>,
|
||||
> + Send
|
||||
+ Sync,
|
||||
{
|
||||
func: F,
|
||||
_phantom: std::marker::PhantomData<D>,
|
||||
}
|
||||
|
||||
impl<D, F> FunctionStep<D, F>
|
||||
where
|
||||
D: WorkflowData,
|
||||
F: Fn(
|
||||
&mut WorkflowContext<D>,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = WorkflowResult<StepResult>> + Send + '_>,
|
||||
> + Send
|
||||
+ Sync,
|
||||
{
|
||||
pub fn new(func: F) -> Self {
|
||||
Self {
|
||||
func,
|
||||
_phantom: std::marker::PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl<D, F> StepExecutor<D> for FunctionStep<D, F>
|
||||
where
|
||||
D: WorkflowData,
|
||||
F: Fn(
|
||||
&mut WorkflowContext<D>,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = WorkflowResult<StepResult>> + Send + '_>,
|
||||
> + Send
|
||||
+ Sync,
|
||||
{
|
||||
async fn execute(&self, context: &mut WorkflowContext<D>) -> WorkflowResult<StepResult> {
|
||||
(self.func)(context).await
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::*;
|
||||
use crate::workflow::types::WorkflowInstanceId;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
struct TestData {
|
||||
value: i32,
|
||||
}
|
||||
|
||||
impl WorkflowData for TestData {
|
||||
fn workflow_type() -> &'static str {
|
||||
"test"
|
||||
}
|
||||
}
|
||||
|
||||
struct TestStep {
|
||||
should_succeed: bool,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl StepExecutor<TestData> for TestStep {
|
||||
async fn execute(
|
||||
&self,
|
||||
_context: &mut WorkflowContext<TestData>,
|
||||
) -> WorkflowResult<StepResult> {
|
||||
if self.should_succeed {
|
||||
Ok(StepResult::Success)
|
||||
} else {
|
||||
Err(WorkflowError::StepFailed {
|
||||
step_id: crate::workflow::types::StepId::new("test"),
|
||||
message: "test error".to_string(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_step_executor_success() {
|
||||
let step = TestStep {
|
||||
should_succeed: true,
|
||||
};
|
||||
let mut context = WorkflowContext::new(WorkflowInstanceId::new(), TestData { value: 42 });
|
||||
|
||||
let result = step.execute(&mut context).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), StepResult::Success);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_step_executor_failure() {
|
||||
let step = TestStep {
|
||||
should_succeed: false,
|
||||
};
|
||||
let mut context = WorkflowContext::new(WorkflowInstanceId::new(), TestData { value: 42 });
|
||||
|
||||
let result = step.execute(&mut context).await;
|
||||
assert!(result.is_err());
|
||||
}
|
||||
}
|
||||
@@ -1,15 +0,0 @@
|
||||
//! Workflow engine for managing multi-step operations
|
||||
|
||||
mod definition;
|
||||
mod engine;
|
||||
mod event;
|
||||
mod executor;
|
||||
mod state;
|
||||
pub mod types;
|
||||
|
||||
pub use definition::{StepDefinition, ValidationError, WorkflowDefinition};
|
||||
pub use engine::WorkflowEngine;
|
||||
pub use event::{EventBus, EventSubscriber, LoggingSubscriber, WorkflowEvent};
|
||||
pub use executor::{FunctionStep, StepExecutor};
|
||||
pub use state::{InMemoryStore, StateStore};
|
||||
pub use types::*;
|
||||
@@ -1,204 +0,0 @@
|
||||
//! Workflow state management
|
||||
|
||||
use std::{collections::HashMap, marker::PhantomData, sync::Arc, time::Duration};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use super::types::{
|
||||
WorkflowContext, WorkflowData, WorkflowError, WorkflowInstanceId, WorkflowResult,
|
||||
WorkflowState, WorkflowStatus,
|
||||
};
|
||||
|
||||
/// Trait for workflow state persistence.
|
||||
///
|
||||
/// Implement this trait to provide custom storage backends (e.g., PostgreSQL, Redis).
|
||||
/// The default implementation is `InMemoryStore` which keeps state in memory.
|
||||
#[async_trait]
|
||||
pub trait StateStore<D: WorkflowData>: Send + Sync + Clone {
|
||||
/// Save workflow state
|
||||
async fn save(&self, state: WorkflowState<D>) -> WorkflowResult<()>;
|
||||
|
||||
/// Load workflow state by instance ID
|
||||
async fn load(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<WorkflowState<D>>;
|
||||
|
||||
/// Update workflow state using a closure
|
||||
async fn update<F>(&self, instance_id: WorkflowInstanceId, f: F) -> WorkflowResult<()>
|
||||
where
|
||||
F: FnOnce(&mut WorkflowState<D>) + Send;
|
||||
|
||||
/// Delete workflow state
|
||||
async fn delete(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<()>;
|
||||
|
||||
/// List all active workflows (Running or Pending)
|
||||
async fn list_active(&self) -> WorkflowResult<Vec<WorkflowState<D>>>;
|
||||
|
||||
/// List all workflows
|
||||
async fn list_all(&self) -> WorkflowResult<Vec<WorkflowState<D>>>;
|
||||
|
||||
/// Check if workflow is cancelled without loading full state
|
||||
async fn is_cancelled(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<bool>;
|
||||
|
||||
/// Clean up old completed/failed/cancelled workflows beyond a time threshold
|
||||
async fn cleanup_old_workflows(&self, ttl: Duration) -> usize;
|
||||
|
||||
/// Get just the workflow context without cloning the entire state
|
||||
async fn get_context(
|
||||
&self,
|
||||
instance_id: WorkflowInstanceId,
|
||||
) -> WorkflowResult<WorkflowContext<D>>;
|
||||
|
||||
/// Clean up a specific workflow immediately if it's in a terminal state
|
||||
/// Returns true if the workflow was removed, false otherwise
|
||||
async fn cleanup_if_terminal(&self, instance_id: WorkflowInstanceId) -> bool;
|
||||
}
|
||||
|
||||
/// In-memory state storage for workflow instances
|
||||
#[derive(Clone)]
|
||||
pub struct InMemoryStore<D: WorkflowData> {
|
||||
states: Arc<RwLock<HashMap<WorkflowInstanceId, WorkflowState<D>>>>,
|
||||
_phantom: PhantomData<D>,
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> InMemoryStore<D> {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
states: Arc::new(RwLock::new(HashMap::new())),
|
||||
_phantom: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Get count of workflows by status
|
||||
pub fn count_by_status(&self, status: WorkflowStatus) -> usize {
|
||||
self.states
|
||||
.read()
|
||||
.values()
|
||||
.filter(|s| s.status == status)
|
||||
.count()
|
||||
}
|
||||
|
||||
/// Get total count of all workflows
|
||||
pub fn count(&self) -> usize {
|
||||
self.states.read().len()
|
||||
}
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> Default for InMemoryStore<D> {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl<D: WorkflowData> StateStore<D> for InMemoryStore<D> {
|
||||
async fn save(&self, state: WorkflowState<D>) -> WorkflowResult<()> {
|
||||
self.states.write().insert(state.instance_id, state);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn load(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<WorkflowState<D>> {
|
||||
self.states
|
||||
.read()
|
||||
.get(&instance_id)
|
||||
.cloned()
|
||||
.ok_or(WorkflowError::NotFound(instance_id))
|
||||
}
|
||||
|
||||
async fn update<F>(&self, instance_id: WorkflowInstanceId, f: F) -> WorkflowResult<()>
|
||||
where
|
||||
F: FnOnce(&mut WorkflowState<D>) + Send,
|
||||
{
|
||||
let mut states = self.states.write();
|
||||
let state = states
|
||||
.get_mut(&instance_id)
|
||||
.ok_or(WorkflowError::NotFound(instance_id))?;
|
||||
f(state);
|
||||
state.updated_at = chrono::Utc::now();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<()> {
|
||||
self.states.write().remove(&instance_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_active(&self) -> WorkflowResult<Vec<WorkflowState<D>>> {
|
||||
let states = self.states.read();
|
||||
Ok(states
|
||||
.values()
|
||||
.filter(|s| matches!(s.status, WorkflowStatus::Running | WorkflowStatus::Pending))
|
||||
.cloned()
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn list_all(&self) -> WorkflowResult<Vec<WorkflowState<D>>> {
|
||||
let states = self.states.read();
|
||||
Ok(states.values().cloned().collect())
|
||||
}
|
||||
|
||||
async fn is_cancelled(&self, instance_id: WorkflowInstanceId) -> WorkflowResult<bool> {
|
||||
self.states
|
||||
.read()
|
||||
.get(&instance_id)
|
||||
.map(|s| s.status == WorkflowStatus::Cancelled)
|
||||
.ok_or(WorkflowError::NotFound(instance_id))
|
||||
}
|
||||
|
||||
async fn cleanup_old_workflows(&self, ttl: Duration) -> usize {
|
||||
let now = chrono::Utc::now();
|
||||
let mut states = self.states.write();
|
||||
let initial_count = states.len();
|
||||
|
||||
states.retain(|_, state| {
|
||||
// Keep active workflows
|
||||
if matches!(
|
||||
state.status,
|
||||
WorkflowStatus::Running | WorkflowStatus::Pending | WorkflowStatus::Paused
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// For terminal workflows, check age
|
||||
let age = now
|
||||
.signed_duration_since(state.updated_at)
|
||||
.to_std()
|
||||
.unwrap_or_default();
|
||||
age < ttl
|
||||
});
|
||||
|
||||
let removed_count = initial_count - states.len();
|
||||
if removed_count > 0 {
|
||||
tracing::info!(
|
||||
removed = removed_count,
|
||||
remaining = states.len(),
|
||||
"Cleaned up old workflow states"
|
||||
);
|
||||
}
|
||||
removed_count
|
||||
}
|
||||
|
||||
async fn get_context(
|
||||
&self,
|
||||
instance_id: WorkflowInstanceId,
|
||||
) -> WorkflowResult<WorkflowContext<D>> {
|
||||
self.states
|
||||
.read()
|
||||
.get(&instance_id)
|
||||
.map(|s| s.context.clone())
|
||||
.ok_or(WorkflowError::NotFound(instance_id))
|
||||
}
|
||||
|
||||
async fn cleanup_if_terminal(&self, instance_id: WorkflowInstanceId) -> bool {
|
||||
let mut states = self.states.write();
|
||||
if let Some(state) = states.get(&instance_id) {
|
||||
if matches!(
|
||||
state.status,
|
||||
WorkflowStatus::Completed | WorkflowStatus::Failed | WorkflowStatus::Cancelled
|
||||
) {
|
||||
states.remove(&instance_id);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
}
|
||||
@@ -1,277 +0,0 @@
|
||||
//! Core workflow types and definitions
|
||||
|
||||
use std::{collections::HashMap, fmt, time::Duration};
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::{de::DeserializeOwned, Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Trait for workflow data that can be passed through workflow steps.
|
||||
///
|
||||
/// Implementing this trait allows your data type to be used as the typed
|
||||
/// context for a workflow. The data must be serializable for state persistence.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```ignore
|
||||
/// #[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
/// pub struct MyWorkflowData {
|
||||
/// pub config: MyConfig,
|
||||
/// pub result: Option<MyResult>,
|
||||
/// #[serde(skip, default)]
|
||||
/// pub app_context: Option<Arc<AppContext>>,
|
||||
/// }
|
||||
///
|
||||
/// impl WorkflowData for MyWorkflowData {
|
||||
/// fn workflow_type() -> &'static str { "my_workflow" }
|
||||
/// }
|
||||
/// ```
|
||||
pub trait WorkflowData: Serialize + DeserializeOwned + Send + Sync + Clone + 'static {
|
||||
/// Human-readable name for logging and identification
|
||||
fn workflow_type() -> &'static str;
|
||||
}
|
||||
|
||||
/// Unique identifier for a workflow definition
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub struct WorkflowId(String);
|
||||
|
||||
impl WorkflowId {
|
||||
pub fn new(id: impl Into<String>) -> Self {
|
||||
Self(id.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for WorkflowId {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.0)
|
||||
}
|
||||
}
|
||||
|
||||
/// Unique identifier for a workflow instance
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub struct WorkflowInstanceId(Uuid);
|
||||
|
||||
impl WorkflowInstanceId {
|
||||
pub fn new() -> Self {
|
||||
Self(Uuid::new_v4())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for WorkflowInstanceId {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for WorkflowInstanceId {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.0)
|
||||
}
|
||||
}
|
||||
|
||||
/// Unique identifier for a workflow step
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub struct StepId(String);
|
||||
|
||||
impl StepId {
|
||||
pub fn new(id: impl Into<String>) -> Self {
|
||||
Self(id.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for StepId {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.0)
|
||||
}
|
||||
}
|
||||
|
||||
/// Retry policy configuration
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RetryPolicy {
|
||||
pub max_attempts: u32,
|
||||
pub backoff: BackoffStrategy,
|
||||
}
|
||||
|
||||
impl Default for RetryPolicy {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_attempts: 3,
|
||||
backoff: BackoffStrategy::Exponential {
|
||||
base: Duration::from_secs(1),
|
||||
max: Duration::from_secs(30),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Backoff strategy for retries
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum BackoffStrategy {
|
||||
/// Fixed delay between retries
|
||||
Fixed(Duration),
|
||||
/// Exponential backoff with base and max duration
|
||||
Exponential { base: Duration, max: Duration },
|
||||
/// Linear backoff with increment and max duration
|
||||
Linear { increment: Duration, max: Duration },
|
||||
}
|
||||
|
||||
/// Action to take when a step fails
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum FailureAction {
|
||||
/// Stop the entire workflow
|
||||
FailWorkflow,
|
||||
/// Skip this step and continue to the next
|
||||
ContinueNextStep,
|
||||
/// Keep retrying indefinitely until manual intervention
|
||||
RetryIndefinitely,
|
||||
}
|
||||
|
||||
/// Workflow execution status
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum WorkflowStatus {
|
||||
Pending,
|
||||
Running,
|
||||
Paused,
|
||||
Completed,
|
||||
Failed,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
/// Step execution status
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum StepStatus {
|
||||
Pending,
|
||||
Running,
|
||||
Succeeded,
|
||||
Failed,
|
||||
Retrying,
|
||||
Skipped,
|
||||
}
|
||||
|
||||
/// State of a workflow step
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct StepState {
|
||||
pub status: StepStatus,
|
||||
pub attempt: u32,
|
||||
pub last_error: Option<String>,
|
||||
pub started_at: Option<DateTime<Utc>>,
|
||||
pub completed_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
impl Default for StepState {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
status: StepStatus::Pending,
|
||||
attempt: 0,
|
||||
last_error: None,
|
||||
started_at: None,
|
||||
completed_at: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Workflow instance state
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(bound(
|
||||
serialize = "D: Serialize",
|
||||
deserialize = "D: serde::de::DeserializeOwned"
|
||||
))]
|
||||
pub struct WorkflowState<D: WorkflowData> {
|
||||
pub instance_id: WorkflowInstanceId,
|
||||
pub definition_id: WorkflowId,
|
||||
pub status: WorkflowStatus,
|
||||
pub current_step: Option<StepId>,
|
||||
pub step_states: HashMap<StepId, StepState>,
|
||||
pub context: WorkflowContext<D>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> WorkflowState<D> {
|
||||
pub fn new(instance_id: WorkflowInstanceId, definition_id: WorkflowId, data: D) -> Self {
|
||||
let now = Utc::now();
|
||||
Self {
|
||||
instance_id,
|
||||
definition_id,
|
||||
status: WorkflowStatus::Pending,
|
||||
current_step: None,
|
||||
step_states: HashMap::new(),
|
||||
context: WorkflowContext::new(instance_id, data),
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared context passed between workflow steps.
|
||||
///
|
||||
/// The context contains typed workflow data that is fully serializable,
|
||||
/// enabling state persistence and workflow recovery.
|
||||
///
|
||||
/// # Type Parameter
|
||||
///
|
||||
/// `D` - The workflow-specific data type implementing `WorkflowData`.
|
||||
/// This type holds all the state needed by workflow steps and must be
|
||||
/// serializable (except for fields marked with `#[serde(skip)]`).
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(bound(
|
||||
serialize = "D: Serialize",
|
||||
deserialize = "D: serde::de::DeserializeOwned"
|
||||
))]
|
||||
pub struct WorkflowContext<D: WorkflowData> {
|
||||
pub instance_id: WorkflowInstanceId,
|
||||
pub data: D,
|
||||
}
|
||||
|
||||
impl<D: WorkflowData> WorkflowContext<D> {
|
||||
pub fn new(instance_id: WorkflowInstanceId, data: D) -> Self {
|
||||
Self { instance_id, data }
|
||||
}
|
||||
}
|
||||
|
||||
/// Result returned by a step execution
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum StepResult {
|
||||
Success,
|
||||
Failure,
|
||||
Skip,
|
||||
}
|
||||
|
||||
/// Error kinds for workflow operations
|
||||
#[derive(Debug, Clone, thiserror::Error)]
|
||||
pub enum WorkflowError {
|
||||
#[error("Workflow not found: {0}")]
|
||||
NotFound(WorkflowInstanceId),
|
||||
|
||||
#[error("Workflow definition not found: {0}")]
|
||||
DefinitionNotFound(WorkflowId),
|
||||
|
||||
#[error("Step failed: {step_id} - {message}")]
|
||||
StepFailed { step_id: StepId, message: String },
|
||||
|
||||
#[error("Step timeout: {step_id}")]
|
||||
StepTimeout { step_id: StepId },
|
||||
|
||||
#[error("Workflow cancelled: {0}")]
|
||||
Cancelled(WorkflowInstanceId),
|
||||
|
||||
#[error("Invalid state transition: {from:?} -> {to:?}")]
|
||||
InvalidStateTransition {
|
||||
from: WorkflowStatus,
|
||||
to: WorkflowStatus,
|
||||
},
|
||||
|
||||
#[error("Context value not found: {0}")]
|
||||
ContextValueNotFound(String),
|
||||
|
||||
#[error("Type mismatch: expected {expected}, got {actual}")]
|
||||
TypeMismatch {
|
||||
expected: &'static str,
|
||||
actual: &'static str,
|
||||
},
|
||||
|
||||
#[error("Engine is shutting down, not accepting new workflows")]
|
||||
ShuttingDown,
|
||||
}
|
||||
|
||||
pub type WorkflowResult<T> = Result<T, WorkflowError>;
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user