Tasker

Workflow orchestration that meets your code where it lives.

Tasker is an open-source workflow orchestration engine built on PostgreSQL and PGMQ. You define workflows as task templates with ordered steps, implement handlers in Rust, Ruby, Python, or TypeScript, and the engine handles execution, retries, circuit breaking, and observability.

Your existing business logic — API calls, database operations, service integrations — becomes a distributed, event-driven, retryable workflow with minimal ceremony. No DSLs to learn, no framework rewrites. Just thin handler wrappers around code you already have.

Get Started

Choose Your Language

Tasker is polyglot from the ground up. The orchestration engine is Rust; workers can be any of four languages, all sharing the same core abstractions expressed idiomatically.

Each language guide covers installation, handler patterns, testing, and production considerations:

Rust · Ruby · Python · TypeScript

Explore the Documentation

For New Users

• Core Concepts — Tasks, steps, handlers, templates, and namespaces
• Choosing Your Package — Which package do you need?
• Quick Start — Running in 5 minutes

Architecture & Design

• Architecture Overview — System design and component interaction
• Design Principles — The tenets behind Tasker’s design decisions
• Architectural Decisions — ADRs documenting key technical choices

Operational Guides

• Handler Resolution — How Tasker finds and runs your handlers
• Retry Semantics — Retry strategies, backoff, and circuit breaking
• Batch Processing — Processing work in batches
• DLQ System — Dead letter queue for failed tasks
• Observability — Metrics, tracing, and logging

Reference

• Configuration Reference — All 246 parameters documented
• Worker API Convergence — Cross-language API alignment
• FFI Safety — How polyglot workers communicate safely

Framework Integrations

• Contrib Packages — Rails, FastAPI, Axum, and Bun integrations
• Example Workflows — E-commerce, ETL, and approval system patterns

Engineering Stories

A progressive-disclosure blog series teaching Tasker concepts through real-world scenarios. Each story follows an engineering team as they adopt workflow orchestration, with working code examples across all four languages.

Stories are being rewritten for the current Tasker architecture. View archive →

The Project

Tasker is open-source software (MIT license) built by an engineer who has spent years designing workflow systems at multiple organizations — and finally had the opportunity to build the one that was always in his head.

It’s not venture-backed. It’s not chasing a market. It’s a labor of love built for the engineering community.

Read the full story →

Source Repositories

Why Tasker

Last Updated: 2025-01-09 Audience: Engineers evaluating workflow orchestration tools Status: Pre-Alpha

The Story

Tasker is a labor of love.

Over the years, I’ve built workflow systems at multiple organizations—each time encountering the same fundamental challenges: orchestrating complex, multi-step processes with proper dependency management, ensuring idempotency, handling retries gracefully, and doing all of this in a way that doesn’t require teams to rewrite their existing business logic.

Each time, I’d design parts of the solution I wished we could build—but the investment was never justifiable. General-purpose workflow infrastructure rarely makes sense for a single company to build from scratch when there are urgent product features to ship. So I’d compromise, cobble together something workable, and move on.

Tasker represents the opportunity to finally build that system properly—the one that’s been evolving in my head for years. Not as a venture-backed startup chasing a market, but as open-source software built by someone who genuinely cares about the problem space and wants to give back to the engineering community.

The Landscape

Honesty is important, and so in full candor: Tasker is not solving an unsolved problem. The workflow orchestration space has mature, battle-tested options.

Apache Airflow

What it does well: Airflow is the industry standard for data pipeline orchestration. Born at Airbnb and now an Apache project with thousands of contributors, it excels at scheduled, batch-oriented workflows defined as Python DAGs. Its ecosystem of operators and integrations is unmatched—if you need to connect to a cloud service, there’s probably an Airflow provider for it.

When to choose it: You have scheduled ETL/ELT workloads, your team is Python-native, you need managed cloud options (AWS MWAA, Google Cloud Composer, Astronomer), and you value ecosystem breadth over ergonomic simplicity.

Honest comparison: Airflow’s 10+ years of production use across thousands of companies represents a level of battle-testing Tasker simply cannot match. If your primary use case is data pipeline orchestration with scheduled intervals, Airflow is likely the safer choice.

Temporal

What it does well: Temporal pioneered “durable execution”—workflows that automatically survive crashes, network failures, and infrastructure outages. It reconstructs application state transparently, letting developers write code as if failures don’t exist. The event history and replay capabilities are genuinely impressive.

When to choose it: You need long-running workflows (hours, days, or longer), your operations require true durability guarantees, you’re building microservice orchestration with complex saga patterns, or you need human-in-the-loop workflows with unbounded wait times.

Honest comparison: Temporal’s durable execution model is architecturally different from Tasker. If your workflows genuinely need to survive arbitrary failures mid-execution and resume from exact state, Temporal was purpose-built for this. Tasker provides resilience through retries and idempotent step execution, but doesn’t offer Temporal’s deterministic replay.

Prefect

What it does well: Prefect feels like “what if workflow orchestration were just Python decorators?” It emphasizes minimal boilerplate—add @flow and @task decorators to existing functions, and you have an orchestrated workflow. Prefect 3.0 embraces dynamic workflows with native Python control flow.

When to choose it: Your team is Python-native, you want the fastest path from script to production pipeline, you value simplicity and developer experience, or you’re doing ML/data science workflows where Jupyter-to-production is important.

Honest comparison: Prefect’s decorator-based ergonomics are genuinely excellent for Python-only teams. If your organization is homogeneously Python and you don’t need polyglot support, Prefect delivers a very clean experience.

Dagster

What it does well: Dagster introduced “software-defined assets” as first-class primitives—you define what data assets should exist and their dependencies, and the orchestrator figures out how to materialize them. This asset-centric model provides excellent lineage tracking and observability.

When to choose it: You’re building a data platform where understanding asset lineage is critical, you want a declarative approach focused on data products rather than task sequences, or you need strong dbt integration and data quality built into your orchestration layer.

Honest comparison: Dagster’s asset-centric philosophy is a genuinely different way of thinking about orchestration. If your mental model centers on “what data assets need to exist” rather than “what steps need to execute,” Dagster may be a better conceptual fit.

So Why Tasker?

Given this landscape, why build another workflow orchestrator?

Philosophy: Meet Teams Where They Are

Most workflow tools require you to think in their terms. Define your work as DAGs using their DSL. Adopt their scheduling model. Often, rewrite your business logic to fit their execution model.

Tasker takes a different approach: bring workflow orchestration to your existing code, rather than bringing your code to a workflow framework.

If your codebase already has reasonable SOLID characteristics—services with clear responsibilities, well-defined interfaces, operations that can be made idempotent—Tasker aims to turn that code into distributed, event-driven, retryable workflows with minimal ceremony.

This philosophy manifests in several ways:

Polyglot from the ground up. Tasker’s orchestration engine is written in Rust, but workers can be written in Ruby, Python, TypeScript, or native Rust. Each language implementation shares the same core abstractions—same handler signatures, same result factories, same patterns—expressed idiomatically for each language. This isn’t an afterthought; cross-language consistency is a core design principle.

Minimal migration burden. Your existing business logic—API calls, database operations, external service integrations—can become step handlers with thin wrappers. You don’t need to restructure your application around the orchestrator.

Framework-agnostic core. Tasker Core provides the fundamentals without framework opinions. Tasker Contrib then provides framework-specific integrations (Rails, FastAPI, Bun) for teams who want batteries-included ergonomics. Progressive disclosure: learn the core concepts first, add framework sugar when needed.

Architecture: Event-Driven with Resilience Built In

Tasker’s architecture reflects lessons learned from building distributed systems:

PostgreSQL-native by default. Everything flows through Postgres—task state, step queues (via PGMQ), event coordination (via LISTEN/NOTIFY). This isn’t because Postgres is trendy; it’s because many teams already have Postgres expertise and operational knowledge. Tasker works as a single-dependency system on PostgreSQL alone. For environments requiring higher throughput or existing RabbitMQ infrastructure, Tasker also supports RabbitMQ as an alternative messaging backend—switch with a configuration change.

Event-driven with polling fallback. Real-time responsiveness through Postgres notifications, but with polling as a reliability backstop. Events can be missed; polling ensures eventual consistency.

Defense in depth. Multiple overlapping protection layers provide robust idempotency without single-point dependency. Database-level atomicity, state machine guards, transaction boundaries, and application-level filtering each catch what others might miss.

Composition over inheritance. Handler capabilities are composed via mixins/traits, not class hierarchies. This enables selective capability inclusion, clear separation of concerns, and easier testing.

Performance: Fast by Default

Tasker is built in Rust not for marketing purposes, but because workflow orchestration has real performance implications. When you’re coordinating thousands of steps across distributed workers, overhead matters.

• Complex 7-step DAG workflows complete in under 133ms with push-based notifications
• Concurrent execution via work-stealing thread pools
• Lock-free channel-based internal coordination
• Zero-copy where possible in the FFI boundaries

The Honest Assessment

Tasker excels when:

• You need polyglot worker support across Ruby, Python, TypeScript, and Rust
• Your team already has Postgres expertise and wants to avoid additional infrastructure
• You want to bring orchestration to existing business logic rather than rewriting
• You value clean, consistent APIs across languages
• Performance matters and you’re willing to trade ecosystem breadth for it

Tasker may not be the right choice when:

• You need the battle-tested maturity and ecosystem of Airflow
• Your workflows require Temporal-style durable execution with deterministic replay
• You’re an all-Python team and Prefect’s ergonomics fit perfectly
• You’re building a data platform where asset-centric thinking (Dagster) is the right model
• You need managed cloud offerings with SLAs and enterprise support

What Tasker Is (and Isn’t)

Tasker Is:

• A workflow orchestration engine for step-based DAG execution with complex dependencies
• PostgreSQL-native with flexible messaging using PGMQ (default) or RabbitMQ
• Polyglot by design with first-class support for multiple languages
• Focused on developer experience for teams who want minimal intrusion
• Open source (MIT license) and built as a labor of love

Tasker Is Not:

• A data orchestration platform like Dagster with asset lineage and data quality primitives
• A durable execution engine like Temporal with deterministic replay and unlimited durability
• A scheduled job runner for simple cron-style workloads (use actual cron)
• A message bus for pure pub/sub fan-out (use Kafka or dedicated brokers)
• Enterprise software with commercial support, SLAs, or managed offerings

Current State

Tasker is pre-alpha software. This is important context:

What this means:

• The architecture is solidifying but breaking changes are expected
• Documentation is comprehensive but evolving
• There are no production deployments (that I know of) outside development
• You should not bet critical business processes on Tasker today

What this enables:

• Rapid iteration based on real feedback
• Willingness to break APIs to get the design right
• Focus on architectural correctness over backward compatibility
• Honest experimentation without legacy constraints

If you’re evaluating Tasker, I’d encourage you to explore it for non-critical workloads, provide feedback, and help shape what it becomes. If you need production-ready workflow orchestration today, please consider the established tools above—I genuinely recommend them for their respective strengths.

The Path Forward

Tasker is being built with care, not speed. The goal isn’t to capture market share or compete with well-funded companies. The goal is to create something genuinely useful—a workflow orchestration system that respects developers’ time and meets them where they are.

The codebase is open, the design decisions are documented, and contributions are welcome. This is software built by an engineer for engineers, not a product chasing metrics.

If that resonates with you, welcome. Let’s build something good together.

Related Documentation

• Tasker Core Tenets - The 10 foundational design principles
• Use Cases & Patterns - When and how to use Tasker
• Quick Start Guide - Get running in 5 minutes
• CHRONOLOGY - Development timeline and lessons learned

← Back to Documentation Hub

Getting Started

This section walks you from “what is Tasker?” to running your first workflow.

Path

1. Overview - What Tasker is and why it exists
2. Core Concepts - Tasks, steps, handlers, templates, and namespaces
3. Installation - Installing packages and running infrastructure
4. Choosing Your Package - Which package do you need?
5. Your First Handler - Write a step handler in your language
6. Your First Workflow - Define a template, submit a task, watch it run
7. Next Steps - Where to go from here

Overview

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Core Concepts

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Installation

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Choosing Your Package

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Your First Handler

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Your First Workflow

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Next Steps

This page will be written as part of the consumer documentation effort. See TAS-215 for details.

Tasker Core Architecture

This directory contains architectural reference documentation describing how Tasker Core’s components work together.

Documents

When to Read These

• Designing new features: Understand how components interact
• Debugging flow issues: Trace message paths through actors
• Understanding trade-offs: See why patterns were chosen
• Onboarding: Build mental model of the system

Related Documentation

• Principles - The “why” behind architectural decisions
• Guides - Practical “how-to” documentation
• CHRONOLOGY - Historical context for decisions

Actor-Based Architecture

Last Updated: 2025-12-04 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | Worker Actor Architecture | Events and Commands | States and Lifecycles

← Back to Documentation Hub

This document provides comprehensive documentation of the actor-based architecture in tasker-core, covering the lightweight Actor pattern that formalizes the relationship between Commands and Lifecycle Components. This architecture replaces imperative delegation with message-based actor coordination.

Overview

The tasker-core system implements a lightweight Actor pattern inspired by frameworks like Actix, but designed specifically for our orchestration needs without external dependencies. The architecture provides:

1. Actor Abstraction: Lifecycle components encapsulated as actors with clear lifecycle hooks
2. Message-Based Communication: Type-safe message handling via Handler trait
3. Central Registry: ActorRegistry for managing all orchestration actors
4. Service Decomposition: Focused components following single responsibility principle
5. Direct Integration: Command processor calls actors directly without wrapper layers

This architecture eliminates inconsistencies in lifecycle component initialization, provides type-safe message handling, and creates a clear separation between command processing and business logic execution.

Implementation Status

All phases implemented and production-ready: core abstractions, all 4 primary actors, message hydration, module reorganization, service decomposition, and direct actor integration.

Core Concepts

What is an Actor?

In the tasker-core context, an Actor is an encapsulated lifecycle component that:

• Manages its own state: Each actor owns its dependencies and configuration
• Processes messages: Responds to typed command messages via the Handler trait
• Has lifecycle hooks: Initialization (started) and cleanup (stopped) methods
• Is isolated: Actors communicate through message passing
• Is thread-safe: All actors are Send + Sync + ’static

Why Actors?

The previous architecture had several inconsistencies:

#![allow(unused)]
fn main() {
// OLD: Inconsistent initialization patterns
pub struct TaskInitializer {
    // Constructor pattern
}

pub struct TaskFinalizer {
    // Builder pattern with new()
}

pub struct StepEnqueuer {
    // Factory pattern with create()
}
}

The actor pattern provides consistency:

#![allow(unused)]
fn main() {
// NEW: Consistent actor pattern
impl OrchestrationActor for TaskRequestActor {
    fn name(&self) -> &'static str { "TaskRequestActor" }
    fn context(&self) -> &Arc<SystemContext> { &self.context }
    fn started(&mut self) -> TaskerResult<()> { /* initialization */ }
    fn stopped(&mut self) -> TaskerResult<()> { /* cleanup */ }
}
}

Actor vs Service

Services (underlying business logic):

• Encapsulate business logic
• Stateless operations on domain models
• Direct method invocation
• Examples: TaskFinalizer, StepEnqueuerService, OrchestrationResultProcessor

Actors (message-based coordination):

• Wrap services with message-based interface
• Manage service lifecycle
• Asynchronous message handling
• Examples: TaskRequestActor, ResultProcessorActor, StepEnqueuerActor, TaskFinalizerActor

The relationship:

#![allow(unused)]
fn main() {
pub struct TaskFinalizerActor {
    context: Arc<SystemContext>,
    service: TaskFinalizer,  // Wraps underlying service
}

impl Handler<FinalizeTaskMessage> for TaskFinalizerActor {
    type Response = FinalizationResult;

    async fn handle(&self, msg: FinalizeTaskMessage) -> TaskerResult<Self::Response> {
        // Delegates to service
        self.service.finalize_task(msg.task_uuid).await
    }
}
}

Actor Traits

OrchestrationActor Trait

The base trait for all orchestration actors, defined in tasker-orchestration/src/actors/traits.rs:

#![allow(unused)]
fn main() {
/// Base trait for all orchestration actors
///
/// Provides lifecycle management and context access for all actors in the
/// orchestration system. All actors must implement this trait to participate
/// in the actor registry and lifecycle management.
///
/// # Lifecycle
///
/// 1. **Construction**: Actor is created by ActorRegistry
/// 2. **Initialization**: `started()` is called during registry build
/// 3. **Operation**: Actor processes messages via Handler<M> implementations
/// 4. **Shutdown**: `stopped()` is called during registry shutdown
pub trait OrchestrationActor: Send + Sync + 'static {
    /// Returns the unique name of this actor
    ///
    /// Used for logging, metrics, and debugging. Should be a static string
    /// that clearly identifies the actor's purpose.
    fn name(&self) -> &'static str;

    /// Returns a reference to the system context
    ///
    /// Provides access to database pool, configuration, and other
    /// framework-level resources.
    fn context(&self) -> &Arc<SystemContext>;

    /// Called when the actor is started
    ///
    /// Perform any initialization work here, such as:
    /// - Setting up database connections
    /// - Loading configuration
    /// - Initializing caches
    ///
    /// # Errors
    ///
    /// Return an error if initialization fails. The actor will not be
    /// registered and the system will fail to start.
    fn started(&mut self) -> TaskerResult<()> {
        tracing::info!(actor = %self.name(), "Actor started");
        Ok(())
    }

    /// Called when the actor is stopped
    ///
    /// Perform any cleanup work here, such as:
    /// - Closing database connections
    /// - Flushing caches
    /// - Releasing resources
    ///
    /// # Errors
    ///
    /// Return an error if cleanup fails. Errors are logged but do not
    /// prevent other actors from shutting down.
    fn stopped(&mut self) -> TaskerResult<()> {
        tracing::info!(actor = %self.name(), "Actor stopped");
        Ok(())
    }
}
}

Key Design Decisions:

1. Send + Sync + ’static: Enables actors to be shared across threads
2. Default lifecycle hooks: Actors only override when needed
3. Context injection: All actors have access to SystemContext
4. Error handling: Lifecycle failures are TaskerResult for proper error propagation

Handler Trait

The message handling trait, enabling type-safe message processing:

#![allow(unused)]
fn main() {
/// Message handler trait for specific message types
///
/// Actors implement Handler<M> for each message type they can process.
/// This provides type-safe, asynchronous message handling with clear
/// input/output contracts.
#[async_trait]
pub trait Handler<M: Message>: OrchestrationActor {
    /// The response type returned by this handler
    type Response: Send;

    /// Handle a message asynchronously
    ///
    /// Process the message and return a response. This method should be
    /// idempotent where possible and handle errors gracefully.
    async fn handle(&self, msg: M) -> TaskerResult<Self::Response>;
}
}

Key Design Decisions:

1. async_trait: All message handling is asynchronous
2. Type safety: Message and Response types are checked at compile time
3. Multiple implementations: Actor can implement Handler for multiple message types
4. Error propagation: TaskerResult ensures proper error handling

Message Trait

The marker trait for command messages:

#![allow(unused)]
fn main() {
/// Marker trait for command messages
///
/// All messages sent to actors must implement this trait. The associated
/// `Response` type defines what the handler will return.
pub trait Message: Send + 'static {
    /// The response type for this message
    type Response: Send;
}
}

Key Design Decisions:

1. Marker trait: No methods, just type constraints
2. Associated type: Response type is part of the message definition
3. Send + ’static: Enables messages to cross thread boundaries

ActorRegistry

The central registry managing all orchestration actors, defined in tasker-orchestration/src/actors/registry.rs:

Purpose

The ActorRegistry serves as:

1. Single Source of Truth: All actors are registered here
2. Lifecycle Manager: Handles initialization and shutdown
3. Dependency Injection: Provides SystemContext to all actors
4. Type-Safe Access: Strongly-typed access to each actor

Structure

#![allow(unused)]
fn main() {
/// Registry managing all orchestration actors
///
/// The ActorRegistry holds Arc references to all actors in the system,
/// providing centralized access and lifecycle management.
#[derive(Clone)]
pub struct ActorRegistry {
    /// System context shared by all actors
    context: Arc<SystemContext>,

    /// Task request actor for processing task initialization requests
    pub task_request_actor: Arc<TaskRequestActor>,

    /// Result processor actor for processing step execution results
    pub result_processor_actor: Arc<ResultProcessorActor>,

    /// Step enqueuer actor for batch processing ready tasks
    pub step_enqueuer_actor: Arc<StepEnqueuerActor>,

    /// Task finalizer actor for task finalization with atomic claiming
    pub task_finalizer_actor: Arc<TaskFinalizerActor>,
}
}

Initialization

The build() method creates and initializes all actors:

#![allow(unused)]
fn main() {
impl ActorRegistry {
    pub async fn build(context: Arc<SystemContext>) -> TaskerResult<Self> {
        tracing::info!("Building ActorRegistry with actors");

        // Create shared StepEnqueuerService (used by multiple actors)
        let task_claim_step_enqueuer = StepEnqueuerService::new(context.clone()).await?;
        let task_claim_step_enqueuer = Arc::new(task_claim_step_enqueuer);

        // Create TaskRequestActor and its dependencies
        let task_initializer = Arc::new(TaskInitializer::new(
            context.clone(),
            task_claim_step_enqueuer.clone(),
        ));

        let task_request_processor = Arc::new(TaskRequestProcessor::new(
            context.message_client.clone(),
            context.task_handler_registry.clone(),
            task_initializer,
            TaskRequestProcessorConfig::default(),
        ));

        let mut task_request_actor = TaskRequestActor::new(context.clone(), task_request_processor);
        task_request_actor.started()?;
        let task_request_actor = Arc::new(task_request_actor);

        // Create ResultProcessorActor and its dependencies
        let task_finalizer = TaskFinalizer::new(context.clone(), task_claim_step_enqueuer.clone());
        let result_processor = Arc::new(OrchestrationResultProcessor::new(
            task_finalizer,
            context.clone(),
        ));

        let mut result_processor_actor =
            ResultProcessorActor::new(context.clone(), result_processor);
        result_processor_actor.started()?;
        let result_processor_actor = Arc::new(result_processor_actor);

        // Create StepEnqueuerActor using shared StepEnqueuerService
        let mut step_enqueuer_actor =
            StepEnqueuerActor::new(context.clone(), task_claim_step_enqueuer.clone());
        step_enqueuer_actor.started()?;
        let step_enqueuer_actor = Arc::new(step_enqueuer_actor);

        // Create TaskFinalizerActor using shared StepEnqueuerService
        let task_finalizer = TaskFinalizer::new(context.clone(), task_claim_step_enqueuer.clone());
        let mut task_finalizer_actor = TaskFinalizerActor::new(context.clone(), task_finalizer);
        task_finalizer_actor.started()?;
        let task_finalizer_actor = Arc::new(task_finalizer_actor);

        tracing::info!("✅ ActorRegistry built successfully with 4 actors");

        Ok(Self {
            context,
            task_request_actor,
            result_processor_actor,
            step_enqueuer_actor,
            task_finalizer_actor,
        })
    }
}
}

Shutdown

The shutdown() method gracefully stops all actors:

#![allow(unused)]
fn main() {
impl ActorRegistry {
    pub async fn shutdown(&mut self) {
        tracing::info!("Shutting down ActorRegistry");

        // Call stopped() on all actors in reverse initialization order
        if let Some(actor) = Arc::get_mut(&mut self.task_finalizer_actor) {
            if let Err(e) = actor.stopped() {
                tracing::error!(error = %e, "Failed to stop TaskFinalizerActor");
            }
        }

        if let Some(actor) = Arc::get_mut(&mut self.step_enqueuer_actor) {
            if let Err(e) = actor.stopped() {
                tracing::error!(error = %e, "Failed to stop StepEnqueuerActor");
            }
        }

        if let Some(actor) = Arc::get_mut(&mut self.result_processor_actor) {
            if let Err(e) = actor.stopped() {
                tracing::error!(error = %e, "Failed to stop ResultProcessorActor");
            }
        }

        if let Some(actor) = Arc::get_mut(&mut self.task_request_actor) {
            if let Err(e) = actor.stopped() {
                tracing::error!(error = %e, "Failed to stop TaskRequestActor");
            }
        }

        tracing::info!("✅ ActorRegistry shutdown complete");
    }
}
}

Implemented Actors

TaskRequestActor

Handles task initialization requests from external clients.

Location: tasker-orchestration/src/actors/task_request_actor.rs

Message: ProcessTaskRequestMessage

• Input: TaskRequestMessage with task details
• Response: Uuid of created task

Delegation: Wraps TaskRequestProcessor service

Purpose: Entry point for new workflow instances, coordinates task creation and initial step discovery.

ResultProcessorActor

Processes step execution results from workers.

Location: tasker-orchestration/src/actors/result_processor_actor.rs

Message: ProcessStepResultMessage

• Input: StepExecutionResult with execution outcome
• Response: () (unit type)

Delegation: Wraps OrchestrationResultProcessor service

Purpose: Handles step completion, coordinates task finalization when appropriate.

StepEnqueuerActor

Manages batch processing of ready tasks.

Location: tasker-orchestration/src/actors/step_enqueuer_actor.rs

Message: ProcessBatchMessage

• Input: Empty (uses system state)
• Response: StepEnqueuerServiceResult with batch stats

Delegation: Wraps StepEnqueuerService

Purpose: Discovers ready tasks and enqueues their executable steps.

TaskFinalizerActor

Handles task finalization with atomic claiming.

Location: tasker-orchestration/src/actors/task_finalizer_actor.rs

Message: FinalizeTaskMessage

• Input: task_uuid to finalize
• Response: FinalizationResult with action taken

Delegation: Wraps TaskFinalizer service (decomposed into focused components)

Purpose: Completes or fails tasks based on step execution results, prevents race conditions through atomic claiming.

Integration with Commands

Command Processor Integration

The command processor calls actors directly without intermediate wrapper layers:

#![allow(unused)]
fn main() {
// From: tasker-orchestration/src/orchestration/command_processor.rs

/// Handle task initialization using TaskRequestActor directly
async fn handle_initialize_task(
    &self,
    request: TaskRequestMessage,
) -> TaskerResult<TaskInitializeResult> {
    // Direct actor-based task initialization
    let msg = ProcessTaskRequestMessage { request };
    let task_uuid = self.actors.task_request_actor.handle(msg).await?;

    Ok(TaskInitializeResult::Success {
        task_uuid,
        message: "Task initialized successfully".to_string(),
    })
}

/// Handle step result processing using ResultProcessorActor directly
async fn handle_process_step_result(
    &self,
    step_result: StepExecutionResult,
) -> TaskerResult<StepProcessResult> {
    // Direct actor-based step result processing
    let msg = ProcessStepResultMessage {
        result: step_result.clone(),
    };

    match self.actors.result_processor_actor.handle(msg).await {
        Ok(()) => Ok(StepProcessResult::Success {
            message: format!(
                "Step {} result processed successfully",
                step_result.step_uuid
            ),
        }),
        Err(e) => Ok(StepProcessResult::Error {
            message: format!("Failed to process step result: {e}"),
        }),
    }
}

/// Handle task finalization using TaskFinalizerActor directly
async fn handle_finalize_task(&self, task_uuid: Uuid) -> TaskerResult<TaskFinalizationResult> {
    // Direct actor-based task finalization
    let msg = FinalizeTaskMessage { task_uuid };

    let result = self.actors.task_finalizer_actor.handle(msg).await?;

    Ok(TaskFinalizationResult::Success {
        task_uuid: result.task_uuid,
        final_status: format!("{:?}", result.action),
        completion_time: Some(chrono::Utc::now()),
    })
}
}

Design Evolution: Initially planned to use lifecycle_services/ as a wrapper layer between command processor and actors. After implementing Phase 7 service decomposition, we found direct actor calls were simpler and cleaner, so we removed the intermediate layer.

Service Decomposition (Phase 7)

Large services (800-900 lines) were decomposed into focused components following single responsibility principle:

TaskFinalizer Decomposition

task_finalization/ (848 lines → 6 files)
├── mod.rs                          # Public API and types
├── service.rs                      # Main TaskFinalizer service (~200 lines)
├── completion_handler.rs           # Task completion logic
├── event_publisher.rs              # Lifecycle event publishing
├── execution_context_provider.rs   # Context fetching
└── state_handlers.rs               # State-specific handling

StepEnqueuerService Decomposition

step_enqueuer_services/ (781 lines → 3 files)
├── mod.rs                          # Public API
├── service.rs                      # Main service (~250 lines)
├── batch_processor.rs              # Batch processing logic
└── state_handlers.rs               # State-specific processing

ResultProcessor Decomposition

result_processing/ (889 lines → 4 files)
├── mod.rs                          # Public API
├── service.rs                      # Main processor
├── metadata_processor.rs           # Metadata handling
├── error_handler.rs                # Error processing
└── result_validator.rs             # Result validation

Actor Lifecycle

Lifecycle Phases

┌─────────────────┐
│  Construction   │  ActorRegistry::build() creates actor instances
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Initialization  │  started() hook called on each actor
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   Operation     │  Actors process messages via Handler<M>::handle()
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│    Shutdown     │  stopped() hook called on each actor (reverse order)
└─────────────────┘

Example Actor Implementation

#![allow(unused)]
fn main() {
use tasker_orchestration::actors::{OrchestrationActor, Handler, Message};

// Define the actor
pub struct TaskFinalizerActor {
    context: Arc<SystemContext>,
    service: TaskFinalizer,
}

// Implement base actor trait
impl OrchestrationActor for TaskFinalizerActor {
    fn name(&self) -> &'static str {
        "TaskFinalizerActor"
    }

    fn context(&self) -> &Arc<SystemContext> {
        &self.context
    }

    fn started(&mut self) -> TaskerResult<()> {
        tracing::info!("TaskFinalizerActor starting");
        Ok(())
    }

    fn stopped(&mut self) -> TaskerResult<()> {
        tracing::info!("TaskFinalizerActor stopping");
        Ok(())
    }
}

// Define message type
pub struct FinalizeTaskMessage {
    pub task_uuid: Uuid,
}

impl Message for FinalizeTaskMessage {
    type Response = FinalizationResult;
}

// Implement message handler
#[async_trait]
impl Handler<FinalizeTaskMessage> for TaskFinalizerActor {
    type Response = FinalizationResult;

    async fn handle(&self, msg: FinalizeTaskMessage) -> TaskerResult<Self::Response> {
        tracing::debug!(
            actor = %self.name(),
            task_uuid = %msg.task_uuid,
            "Processing FinalizeTaskMessage"
        );

        // Delegate to service
        self.service.finalize_task(msg.task_uuid).await
            .map_err(|e| e.into())
    }
}
}

Benefits

1. Consistency

All lifecycle components follow the same pattern:

• Uniform initialization via started()
• Uniform cleanup via stopped()
• Uniform message handling via Handler<M>

2. Type Safety

Messages and responses are type-checked at compile time:

#![allow(unused)]
fn main() {
// Compile error if message/response types don't match
impl Handler<WrongMessage> for TaskFinalizerActor {
    type Response = WrongResponse;  // ❌ Won't compile
    // ...
}
}

3. Testability

• Clear message boundaries for mocking
• Isolated actor lifecycle for unit tests
• Type-safe message construction

4. Maintainability

• Clear separation of concerns
• Explicit message contracts
• Centralized lifecycle management
• Decomposed services (<300 lines per file)

5. Simplicity

• Direct actor calls (no wrapper layers)
• Pure routing in command processor
• Easy to trace message flow

Summary

The actor-based architecture provides a consistent, type-safe foundation for lifecycle component management in tasker-core. Key takeaways:

1. Lightweight Pattern: Actors wrap decomposed services, providing message-based interface
2. Lifecycle Management: Consistent initialization and shutdown via traits
3. Type Safety: Compile-time verification of message contracts
4. Service Decomposition: Focused components following single responsibility principle
5. Direct Integration: Command processor calls actors directly without intermediate wrappers
6. Production Ready: All phases complete, zero breaking changes, full test coverage

The architecture provides a solid foundation for future scalability and maintainability while maintaining the proven reliability of existing orchestration logic.

← Back to Documentation Hub

Backpressure Architecture

Last Updated: 2026-02-05 Audience: Architects, Developers, Operations Status: Active Related Docs: Worker Event Systems | MPSC Channel Guidelines

<- Back to Documentation Hub

This document provides the unified backpressure strategy for tasker-core, covering all system components from API ingestion through worker execution.

Core Principle

Step idempotency is the primary constraint. Any backpressure mechanism must ensure that step claiming, business logic execution, and result persistence remain stable and consistent. The system must gracefully degrade under load without compromising workflow correctness.

System Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                         BACKPRESSURE FLOW OVERVIEW                           │
└─────────────────────────────────────────────────────────────────────────────┘

                            ┌─────────────────┐
                            │  External Client │
                            └────────┬────────┘
                                     │
                    ┌────────────────▼────────────────┐
                    │  [1] API LAYER BACKPRESSURE      │
                    │  • Circuit breaker (503)         │
                    │  • System overload (503)         │
                    │  • Request validation            │
                    └────────────────┬────────────────┘
                                     │
                    ┌────────────────▼────────────────┐
                    │  [2] ORCHESTRATION BACKPRESSURE  │
                    │  • Command channel (bounded)     │
                    │  • Connection pool limits        │
                    │  • PGMQ depth checks             │
                    └────────────────┬────────────────┘
                                     │
                         ┌───────────┴───────────┐
                         │     PGMQ Queues       │
                         │  • Namespace queues   │
                         │  • Result queues      │
                         └───────────┬───────────┘
                                     │
                    ┌────────────────▼────────────────┐
                    │  [3] WORKER BACKPRESSURE         │
                    │  • Claim capacity check          │
                    │  • Semaphore-bounded handlers    │
                    │  • Completion channel bounds     │
                    └────────────────┬────────────────┘
                                     │
                    ┌────────────────▼────────────────┐
                    │  [4] RESULT FLOW BACKPRESSURE    │
                    │  • Completion channel bounds     │
                    │  • Domain event drop semantics   │
                    └─────────────────────────────────┘

Backpressure Points by Component

1. API Layer

The API layer provides backpressure through 503 responses with intelligent Retry-After headers.

Rate Limiting (429): Rate limiting is intentionally out of scope for tasker-core. This responsibility belongs to upstream infrastructure (API Gateways, NLB/ALB, service mesh). Tasker focuses on system health-based backpressure via 503 responses.

Response Codes:

• 200 OK - Request accepted
• 400 Bad Request - Invalid request format
• 503 Service Unavailable - System overloaded (includes Retry-After header)

503 Response Triggers:

1. Circuit Breaker Open: Database operations failing repeatedly
2. Queue Depth High (Planned): PGMQ namespace queues approaching capacity
3. Channel Saturation (Planned): Command channel buffer > 80% full

Retry-After Header Strategy:

503 Service Unavailable
Retry-After: {calculated_delay_seconds}

Calculation:
- Circuit breaker open: Use breaker timeout (default 30s)
- Queue depth high: Estimate based on processing rate
- Channel saturation: Short delay (5-10s) for buffer drain

Configuration:

# config/tasker/base/common.toml
[common.circuit_breakers.component_configs.web]
failure_threshold = 5      # Failures before opening
success_threshold = 2      # Successes in half-open to close
# timeout_seconds inherited from default_config (30s)

2. Orchestration Layer

The orchestration layer protects internal processing from command flooding.

Command Channel Backpressure:

Command Sender → [Bounded Channel] → Command Processor
                      │
                      └── If full: Block with timeout → Reject

Configuration:

# config/tasker/base/orchestration.toml
[orchestration.mpsc_channels.command_processor]
command_buffer_size = 5000

[orchestration.mpsc_channels.pgmq_events]
pgmq_event_buffer_size = 50000

3. Messaging Layer

The messaging layer provides the backbone between orchestration and workers. Provider-agnostic via MessageClient, supporting PGMQ (default) and RabbitMQ backends.

Messaging Circuit Breaker: MessageClient wraps send/receive operations with circuit breaker protection. When the messaging provider (PGMQ or RabbitMQ) fails repeatedly, the breaker opens and returns MessagingError::CircuitBreakerOpen immediately, preventing slow timeouts from cascading into orchestration and worker processing loops. Ack/nack and health check operations bypass the breaker — ack/nack failures are safe (visibility timeout handles redelivery), and health check must work when the breaker is open to detect recovery. See Circuit Breakers for details.

Queue Depth Monitoring (Planned):

The system will work with PGMQ’s native capabilities rather than enforcing arbitrary limits. Queue depth monitoring provides visibility without hard rejection:

┌──────────────────────────────────────────────────────────────────────┐
│ QUEUE DEPTH STRATEGY                                                  │
├──────────────────────────────────────────────────────────────────────┤
│ Level    │ Depth Ratio │ Action                                       │
├──────────────────────────────────────────────────────────────────────┤
│ Normal   │ < 70%       │ Normal operation                             │
│ Warning  │ 70-85%      │ Log warning, emit metric                     │
│ Critical │ 85-95%      │ API returns 503 for new tasks                │
│ Overflow │ > 95%       │ API rejects all writes, alert operators      │
└──────────────────────────────────────────────────────────────────────┘

Note: Depth ratio = current_depth / configured_soft_limit
Soft limit is advisory, not a hard PGMQ constraint.

Portability Considerations:

• Queue depth semantics vary by backend (PGMQ vs RabbitMQ vs SQS)
• Configuration is backend-agnostic where possible
• Backend-specific tuning goes in backend-specific config sections

Configuration:

# config/tasker/base/common.toml
[common.queues]
default_visibility_timeout_seconds = 30

[common.queues.pgmq]
poll_interval_ms = 250

[common.queues.pgmq.queue_depth_thresholds]
critical_threshold = 500
overflow_threshold = 1000

# Messaging circuit breaker
[common.circuit_breakers.component_configs.messaging]
failure_threshold = 5      # Failures before opening
success_threshold = 2      # Successes in half-open to close
# timeout_seconds inherited from default_config (30s)

4. Worker Layer

The worker layer protects handler execution from saturation.

Handler Dispatch Flow:

Step Message
     │
     ▼
┌─────────────────┐
│ Capacity Check  │──── At capacity? ──── Leave in queue
│ (Planned)       │                       (visibility timeout)
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Acquire Permit  │
│ (Semaphore)     │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Execute Handler │
│ (with timeout)  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Release Permit  │──── BEFORE sending to completion channel
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Send Completion │
└─────────────────┘

Configuration:

# config/tasker/base/worker.toml
[worker.mpsc_channels.handler_dispatch]
dispatch_buffer_size = 1000
completion_buffer_size = 1000
max_concurrent_handlers = 10
handler_timeout_ms = 30000

5. Domain Events

Domain events use fire-and-forget semantics to avoid blocking the critical path.

Domain Event Flow:

Handler Complete
     │
     ├── Result → Completion Channel (blocking, must succeed)
     │
     └── Domain Events → try_send() → If full: DROP with metric
                                       │
                                       └── Step execution NOT affected

Segmentation of Responsibility

Orchestration System

The orchestration system must protect itself from:

1. Client overload: Too many /v1/tasks requests
2. Internal saturation: Command channel overflow
3. Database exhaustion: Connection pool depletion
4. Queue explosion: Unbounded PGMQ growth

Backpressure Response Hierarchy:

1. Return 503 to client with Retry-After (fastest, cheapest)
2. Block at command channel (internal buffering)
3. Soft-reject at queue depth threshold (503 to new tasks)
4. Circuit breaker opens (stop accepting work)

Worker System

The worker system must protect itself from:

1. Handler saturation: Too many concurrent handlers
2. FFI backlog: Ruby/Python handlers falling behind
3. Completion overflow: Results backing up
4. Step starvation: Claims outpacing processing

Backpressure Response Hierarchy:

1. Refuse step claim (leave in queue, visibility timeout)
2. Block at dispatch channel (internal buffering)
3. Block at completion channel (handler waits)
4. Circuit breaker opens (stop claiming work)

Step Idempotency Guarantees

Safe Backpressure Points

These backpressure points preserve step idempotency:

Unsafe Patterns (NEVER DO)

Idempotency Contract

┌─────────────────────────────────────────────────────────────────────────────┐
│                    STEP EXECUTION IDEMPOTENCY CONTRACT                       │
└─────────────────────────────────────────────────────────────────────────────┘

1. CLAIM: Atomic via pgmq_read_specific_message()
   ├── Only one worker can claim a message
   ├── Visibility timeout protects against worker crash
   └── If claim fails: Message stays in queue → another worker claims

2. EXECUTE: Handler invocation (FFI boundary critical - see below)
   ├── Handlers SHOULD be idempotent (business logic recommendation)
   ├── Timeout generates FAILURE result (not drop)
   ├── Panic generates FAILURE result (not drop)
   └── Error generates FAILURE result (not drop)

3. PERSIST: Result submission
   ├── Completion channel is bounded but BLOCKING
   ├── Result MUST reach orchestration (never dropped)
   └── If send fails: Step remains "in_progress" → recovered by orchestration

4. FINALIZE: Orchestration processes result
   ├── State transition is atomic
   ├── Duplicate results handled by state guards
   └── Idempotent: Same result processed twice = same outcome

FFI Boundary Idempotency Semantics

The FFI boundary (Rust → Ruby/Python handler) creates a critical demarcation for error classification:

┌─────────────────────────────────────────────────────────────────────────────┐
│                    FFI BOUNDARY ERROR CLASSIFICATION                         │
└─────────────────────────────────────────────────────────────────────────────┘

                           FFI BOUNDARY
                                │
    BEFORE FFI CROSSING         │         AFTER FFI CROSSING
    (System Layer)              │         (Business Logic Layer)
                                │
    ┌─────────────────────┐     │     ┌─────────────────────┐
    │ System errors are   │     │     │ System failures     │
    │ RETRYABLE:          │     │     │ are PERMANENT:      │
    │                     │     │     │                     │
    │ • Channel timeout   │     │     │ • Worker crash      │
    │ • Queue unavailable │     │     │ • FFI panic         │
    │ • Claim race lost   │     │     │ • Process killed    │
    │ • Network partition │     │     │                     │
    │ • Message malformed │     │     │ We cannot know if   │
    │                     │     │     │ business logic      │
    │ Step has NOT been   │     │     │ executed or not.    │
    │ handed to handler.  │     │     │                     │
    └─────────────────────┘     │     └─────────────────────┘
                                │
                                │     ┌─────────────────────┐
                                │     │ Developer errors    │
                                │     │ are TRUSTED:        │
                                │     │                     │
                                │     │ • RetryableError →  │
                                │     │   System retries    │
                                │     │                     │
                                │     │ • PermanentError →  │
                                │     │   Step fails        │
                                │     │                     │
                                │     │ Developer knows     │
                                │     │ their domain logic. │
                                │     └─────────────────────┘

Key Principles:

1. Before FFI: Any system error is safe to retry because no business logic has executed.

2. After FFI, system failure: If the worker crashes or FFI call fails after dispatch, we MUST treat it as permanent failure. We cannot know if the handler:

   • Never started (safe to retry)
   • Started but didn’t complete (unknown side effects)
   • Completed but didn’t return (work is done)

3. After FFI, developer error: Trust the developer’s classification:

   • RetryableError: Developer explicitly signals safe to retry (e.g., temporary API unavailable)
   • PermanentError: Developer explicitly signals not retriable (e.g., invalid data, business rule violation)

Implementation Guidance:

#![allow(unused)]
fn main() {
// BEFORE FFI - system error, retryable
match dispatch_to_handler(step).await {
    Err(DispatchError::ChannelFull) => StepResult::retryable("dispatch_channel_full"),
    Err(DispatchError::Timeout) => StepResult::retryable("dispatch_timeout"),
    Ok(ffi_handle) => {
        // AFTER FFI - different rules apply
        match ffi_handle.await {
            // System crash after FFI = permanent (unknown state)
            Err(FfiError::ProcessCrash) => StepResult::permanent("handler_crash"),
            Err(FfiError::Panic) => StepResult::permanent("handler_panic"),

            // Developer-returned errors = trust their classification
            Ok(HandlerResult::RetryableError(msg)) => StepResult::retryable(msg),
            Ok(HandlerResult::PermanentError(msg)) => StepResult::permanent(msg),
            Ok(HandlerResult::Success(data)) => StepResult::success(data),
        }
    }
}
}

Note: We RECOMMEND handlers be idempotent but cannot REQUIRE it—business logic is developer-controlled. The system provides visibility timeout protection and duplicate result handling, but ultimate idempotency responsibility lies with handler implementations.

Backpressure Decision Tree

Use this decision tree when designing new backpressure mechanisms:

                    ┌─────────────────────────┐
                    │ New Backpressure Point  │
                    └────────────┬────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │ Does this affect step   │
                    │ execution correctness?  │
                    └────────────┬────────────┘
                                 │
                   ┌─────────────┴─────────────┐
                   │                           │
                  Yes                          No
                   │                           │
                   ▼                           ▼
         ┌─────────────────┐         ┌─────────────────┐
         │ Can the work be │         │ Safe to drop    │
         │ retried safely? │         │ or timeout      │
         └────────┬────────┘         └─────────────────┘
                  │
        ┌─────────┴─────────┐
        │                   │
       Yes                  No
        │                   │
        ▼                   ▼
  ┌───────────┐      ┌───────────────┐
  │ Use block │      │ MUST NOT DROP │
  │ or reject │      │ Block until   │
  │ (retriable│      │ success       │
  │ error)    │      └───────────────┘
  └───────────┘

Configuration Reference

TOML Structure: Configuration files are organized as config/tasker/base/{common,worker,orchestration}.toml with environment overrides in config/tasker/environments/{test,development,production}/.

Complete Backpressure Configuration

# ════════════════════════════════════════════════════════════════════════════
# config/tasker/base/common.toml - Shared settings
# ════════════════════════════════════════════════════════════════════════════

# Circuit breaker defaults (inherited by all component breakers)
[common.circuit_breakers.default_config]
failure_threshold = 5      # Failures before opening
timeout_seconds = 30       # Time in open state before half-open
success_threshold = 2      # Successes in half-open to close

# Web/API database circuit breaker
[common.circuit_breakers.component_configs.web]
failure_threshold = 5
success_threshold = 2

# Messaging circuit breaker - PGMQ/RabbitMQ operations
[common.circuit_breakers.component_configs.messaging]
failure_threshold = 5
success_threshold = 2

# Queue configuration
[common.queues]
default_visibility_timeout_seconds = 30

[common.queues.pgmq]
poll_interval_ms = 250

[common.queues.pgmq.queue_depth_thresholds]
critical_threshold = 500
overflow_threshold = 1000

# ════════════════════════════════════════════════════════════════════════════
# config/tasker/base/orchestration.toml - Orchestration layer
# ════════════════════════════════════════════════════════════════════════════

[orchestration.mpsc_channels.command_processor]
command_buffer_size = 5000

[orchestration.mpsc_channels.pgmq_events]
pgmq_event_buffer_size = 50000

[orchestration.mpsc_channels.event_channel]
event_channel_buffer_size = 10000

# ════════════════════════════════════════════════════════════════════════════
# config/tasker/base/worker.toml - Worker layer
# ════════════════════════════════════════════════════════════════════════════

[worker.mpsc_channels.handler_dispatch]
dispatch_buffer_size = 1000        # Steps waiting for handler
completion_buffer_size = 1000      # Results waiting for orchestration
max_concurrent_handlers = 10       # Semaphore permits
handler_timeout_ms = 30000         # Max handler execution time

[worker.mpsc_channels.ffi_dispatch]
dispatch_buffer_size = 1000        # FFI events waiting for Ruby/Python
completion_timeout_ms = 30000      # Time to wait for FFI completion
starvation_warning_threshold_ms = 10000  # Warn if event waits this long

# Planned:
# claim_capacity_threshold = 0.8   # Refuse claims at 80% capacity

Monitoring and Alerting

See Backpressure Monitoring Runbook for:

• Key metrics to monitor
• Alerting thresholds
• Incident response procedures

Key Metrics Summary

Related Documentation

• Worker Event Systems - Dual-channel architecture
• MPSC Channel Guidelines - Channel creation guide
• MPSC Channel Tuning - Operational tuning
• Bounded MPSC Channels ADR

<- Back to Documentation Hub

Circuit Breakers

Last Updated: 2026-02-04 Audience: Architects, Operators, Developers Status: Active Related Docs: Backpressure Architecture | Observability | Operations: Backpressure Monitoring

<- Back to Documentation Hub

Circuit breakers provide fault isolation and cascade prevention across tasker-core. This document covers the circuit breaker architecture, implementations, configuration, and operational monitoring.

Core Concept

Circuit breakers prevent cascading failures by failing fast when a component is unhealthy. Instead of waiting for slow or failing operations to timeout, circuit breakers detect failure patterns and immediately reject calls, giving the downstream system time to recover.

State Machine

┌─────────────────────────────────────────────────────────────────────────────┐
│                     CIRCUIT BREAKER STATE MACHINE                            │
└─────────────────────────────────────────────────────────────────────────────┘

                    Success
                  ┌─────────┐
                  │         │
                  ▼         │
              ┌───────┐     │
      ───────>│CLOSED │─────┘
              └───┬───┘
                  │
                  │ failure_threshold
                  │ consecutive failures
                  │
                  ▼
              ┌───────┐
              │ OPEN  │◄─────────────────────┐
              └───┬───┘                      │
                  │                          │
                  │ timeout_seconds          │ Any failure
                  │ elapsed                  │ in half-open
                  │                          │
                  ▼                          │
            ┌──────────┐                     │
            │HALF-OPEN │─────────────────────┘
            └────┬─────┘
                 │
                 │ success_threshold
                 │ consecutive successes
                 │
                 ▼
            ┌───────┐
            │CLOSED │
            └───────┘

States:

• Closed: Normal operation. All calls allowed. Tracks consecutive failures.
• Open: Failing fast. All calls rejected immediately. Waiting for timeout.
• Half-Open: Testing recovery. Limited calls allowed. Single failure reopens.

Unified Trait: CircuitBreakerBehavior

All circuit breaker implementations share a common trait defined in tasker-shared/src/resilience/behavior.rs:

#![allow(unused)]
fn main() {
pub trait CircuitBreakerBehavior: Send + Sync + Debug {
    fn name(&self) -> &str;
    fn state(&self) -> CircuitState;
    fn should_allow(&self) -> bool;
    fn record_success(&self, duration: Duration);
    fn record_failure(&self, duration: Duration);
    fn is_healthy(&self) -> bool;
    fn force_open(&self);
    fn force_closed(&self);
    fn metrics(&self) -> CircuitBreakerMetrics;
}
}

Each specialized breaker wraps the generic CircuitBreaker (composition pattern) and implements this trait. This means:

• Consistent state machine behavior across all breakers
• Proper half-open → closed recovery via success_threshold
• Lock-free atomic state management
• Domain-specific methods remain as additional methods on each type

Circuit Breaker Implementations

Tasker-core has four circuit breaker implementations, each protecting specific components. All wrap the generic CircuitBreaker from tasker_shared::resilience:

1. Web Database Circuit Breaker

Purpose: Protects API endpoints from cascading database failures.

Scope: Independent from orchestration system’s internal operations.

Behavior:

• Opens when database queries fail repeatedly
• Returns 503 with Retry-After header when open
• Fast-fail rejection with atomic state management

Configuration (config/tasker/base/common.toml):

[common.circuit_breakers.component_configs.web]
failure_threshold = 5      # Consecutive failures before opening
success_threshold = 2      # Successes in half-open to fully close
# timeout_seconds inherited from default_config (30s)

Health Check Integration:

• Included in /health/ready endpoint
• State reported in /health/detailed response
• Metric: api_circuit_breaker_state (0=closed, 1=half-open, 2=open)

2. Task Readiness Circuit Breaker

Purpose: Protects fallback poller from database overload during polling cycles.

Scope: Independent from web circuit breaker, specific to task readiness queries.

Behavior:

• Opens when task readiness queries fail repeatedly
• Skips polling cycles when open (doesn’t fail-fast, just skips)
• Allows orchestration to continue processing existing work

Configuration (config/tasker/base/common.toml):

[common.circuit_breakers.component_configs.task_readiness]
failure_threshold = 10     # Higher threshold for polling
timeout_seconds = 60       # Longer recovery window
success_threshold = 3      # More successes needed for confidence

Why Separate from Web?:

• Different failure patterns (polling vs request-driven)
• Different recovery semantics (skip vs reject)
• Isolation prevents web failures from stopping polling (and vice versa)

3. FFI Completion Circuit Breaker

Purpose: Protects Ruby/Python worker completion channels from backpressure.

Scope: Worker-specific, protects FFI boundary.

Behavior:

• Latency-based: Treats slow sends (>100ms) as failures
• Opens when completion channel is consistently slow
• Prevents FFI threads from blocking on saturated channels
• Drops completions when open (with metrics), allowing handler threads to continue

Configuration (config/tasker/base/worker.toml):

[worker.circuit_breakers.ffi_completion_send]
failure_threshold = 5            # Slow sends before opening
recovery_timeout_seconds = 5     # Short recovery window
success_threshold = 2            # Successes to close
slow_send_threshold_ms = 100     # Latency threshold (100ms)

Why Latency-Based?:

• Slow channel sends indicate backpressure buildup
• Blocking FFI threads can cascade to Ruby/Python handler starvation
• Error-only detection misses slow-but-completing operations
• Latency detection catches degradation before total failure

Metrics:

• ffi_completion_slow_sends_total - Sends exceeding latency threshold
• ffi_completion_circuit_open_rejections_total - Rejections due to open circuit

4. Messaging Circuit Breaker

Purpose: Protects message queue operations from provider failures (PGMQ or RabbitMQ).

Scope: Integrated into MessageClient, shared across orchestration and worker messaging.

Behavior:

• Opens when send/receive operations fail repeatedly
• Protected operations: send_step_message, receive_step_messages, send_step_result, receive_step_results, send_task_request, receive_task_requests, send_task_finalization, receive_task_finalizations, send_message, receive_messages
• Unprotected operations (safe to fail or needed for recovery): ack_message, nack_message, extend_visibility, health_check, ensure_queue, queue stats
• Coordinates with visibility timeout for message safety
• Provider-agnostic: works with both PGMQ and RabbitMQ backends

Configuration (config/tasker/base/common.toml):

[common.circuit_breakers.component_configs.messaging]
failure_threshold = 5      # Failures before opening
success_threshold = 2      # Successes to close
# timeout_seconds inherited from default_config (30s)

Why ack/nack bypass the breaker?:

• Ack/nack failure causes message redelivery via visibility timeout, which is safe
• Health check must work when breaker is open to detect recovery
• Queue management is startup-only and should not be gated

Configuration Reference

Global Settings

[common.circuit_breakers.global_settings]
metrics_collection_interval_seconds = 30    # Metrics aggregation interval
min_state_transition_interval_seconds = 5.0 # Debounce for rapid transitions

Default Configuration

Applied to any circuit breaker without explicit configuration:

[common.circuit_breakers.default_config]
failure_threshold = 5      # 1-100 range
timeout_seconds = 30       # 1-300 range
success_threshold = 2      # 1-50 range

Component-Specific Overrides

# Task readiness (polling-specific)
[common.circuit_breakers.component_configs.task_readiness]
failure_threshold = 10
success_threshold = 3

# Messaging operations (PGMQ/RabbitMQ)
[common.circuit_breakers.component_configs.messaging]
failure_threshold = 5
success_threshold = 2

# Web/API database operations
[common.circuit_breakers.component_configs.web]
failure_threshold = 5
success_threshold = 2

Note: timeout_seconds is inherited from default_config for all component circuit breakers. The pgmq key is accepted as an alias for messaging for backward compatibility.

Worker-Specific Configuration

# FFI completion (latency-based)
[worker.circuit_breakers.ffi_completion_send]
failure_threshold = 5
recovery_timeout_seconds = 5
success_threshold = 2
slow_send_threshold_ms = 100

Environment Overrides

Different environments may need different thresholds:

Test (config/tasker/environments/test/common.toml):

[common.circuit_breakers.default_config]
failure_threshold = 2      # Faster failure detection
timeout_seconds = 5        # Quick recovery for tests
success_threshold = 1

Production (config/tasker/environments/production/common.toml):

[common.circuit_breakers.default_config]
failure_threshold = 10     # More tolerance for transient failures
timeout_seconds = 60       # Longer recovery window
success_threshold = 5      # More confidence before closing

Health Endpoint Integration

Circuit breaker states are exposed through health endpoints for monitoring and Kubernetes probes.

Orchestration Health (/health/detailed)

{
  "status": "healthy",
  "checks": {
    "circuit_breakers": {
      "status": "healthy",
      "message": "Circuit breaker state: Closed",
      "duration_ms": 1,
      "last_checked": "2025-12-10T10:00:00Z"
    }
  }
}

Worker Health (/health/detailed)

{
  "status": "healthy",
  "checks": {
    "circuit_breakers": {
      "status": "healthy",
      "message": "2 circuit breakers: 2 closed, 0 open, 0 half-open. Details: ffi_completion: closed (100 calls, 2 failures); task_readiness: closed (50 calls, 0 failures)",
      "duration_ms": 0,
      "last_checked": "2025-12-10T10:00:00Z"
    }
  }
}

Health Status Mapping

Monitoring and Alerting

Key Metrics

Prometheus Alerts

groups:
  - name: circuit_breakers
    rules:
      - alert: TaskerCircuitBreakerOpen
        expr: api_circuit_breaker_state == 2
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Circuit breaker is OPEN"
          description: "Circuit breaker {{ $labels.component }} has been open for >1 minute"

      - alert: TaskerCircuitBreakerHalfOpen
        expr: api_circuit_breaker_state == 1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Circuit breaker stuck in half-open"
          description: "Circuit breaker {{ $labels.component }} in half-open state >5 minutes"

      - alert: TaskerFFISlowSendsHigh
        expr: rate(ffi_completion_slow_sends_total[5m]) > 10
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "FFI completion channel experiencing backpressure"
          description: "Slow sends averaging >10/second, circuit breaker may open"

Grafana Dashboard Panels

Circuit Breaker State Timeline:

Panel: Time series
Query: api_circuit_breaker_state
Value mappings: 0=Closed (green), 1=Half-Open (yellow), 2=Open (red)

FFI Latency Percentiles:

Panel: Time series
Queries:
  - histogram_quantile(0.50, ffi_completion_send_duration_seconds_bucket)
  - histogram_quantile(0.95, ffi_completion_send_duration_seconds_bucket)
  - histogram_quantile(0.99, ffi_completion_send_duration_seconds_bucket)
Thresholds: 100ms warning, 500ms critical

Operational Procedures

When Circuit Breaker Opens

Immediate Actions:

1. Check database connectivity: pg_isready -h <host> -p 5432
2. Check connection pool status: /health/detailed endpoint
3. Review recent error logs for root cause
4. Monitor queue depth for message backlog

Recovery:

• Circuit automatically tests recovery after timeout_seconds
• No manual intervention needed for transient failures
• For persistent failures, fix underlying issue first

Escalation:

• If breaker stays open >5 minutes, escalate to database team
• If breaker oscillates (open/half-open/open), increase failure_threshold

Tuning Guidelines

Symptom: Breaker opens too frequently

• Increase failure_threshold
• Investigate root cause of failures
• Consider if failures are transient vs systemic

Symptom: Breaker stays open too long

• Decrease timeout_seconds
• Verify downstream system has recovered
• Check if success_threshold is too high

Symptom: FFI breaker opens unnecessarily

• Increase slow_send_threshold_ms
• Verify channel buffer sizes are adequate
• Check Ruby/Python handler throughput

Architecture Integration

Relationship to Backpressure

Circuit breakers are one layer of the broader backpressure strategy:

┌─────────────────────────────────────────────────────────────────────────────┐
│                        RESILIENCE LAYER STACK                                │
└─────────────────────────────────────────────────────────────────────────────┘

Layer 1: Circuit Breakers     → Fast-fail on component failure
Layer 2: Bounded Channels     → Backpressure on internal queues
Layer 3: Visibility Timeouts  → Message-level retry safety
Layer 4: Semaphore Limits     → Handler execution rate limiting
Layer 5: Connection Pools     → Database resource management

See Backpressure Architecture for the complete strategy.

Independence Principle

Each circuit breaker operates independently:

• Web breaker can be open while task readiness breaker is closed
• FFI breaker state doesn’t affect PGMQ breaker
• Prevents single failure mode from cascading across components
• Allows targeted recovery per component

Integration Points

Troubleshooting

Common Issues

Issue: Web circuit breaker flapping (open → half-open → open rapidly)

Diagnosis:

1. Check database query latency (slow queries can cause timeout failures)
2. Review connection pool saturation
3. Check if PostgreSQL is under memory pressure

Resolution:

• Increase failure_threshold if failures are transient
• Increase timeout_seconds to give more recovery time
• Fix underlying database performance issues

Issue: FFI completion circuit breaker opens during normal load

Diagnosis:

1. Check Ruby/Python handler execution time
2. Review completion channel buffer utilization
3. Verify worker concurrency settings

Resolution:

• Increase slow_send_threshold_ms if handlers are legitimately slow
• Increase channel buffer size in worker config
• Reduce handler concurrency if system is overloaded

Issue: Task readiness breaker open but web API working fine

Diagnosis:

• Task readiness queries may be slower/different than API queries
• Polling may hit database at different times (e.g., during maintenance)

Resolution:

• Independent breakers are working as designed
• Check specific task readiness query performance
• Consider database index optimization for readiness queries

Source Code Reference

Related Documentation

• Backpressure Architecture - Complete resilience strategy
• Operations: Backpressure Monitoring - Operational runbooks
• Operations: MPSC Channel Tuning - Channel capacity management
• Observability - Metrics and logging standards
• Configuration Management - TOML configuration reference

<- Back to Documentation Hub

Crate Architecture

Last Updated: 2026-01-15 Audience: Developers, Architects Status: Active Related Docs: Documentation Hub | Actor-Based Architecture | Events and Commands | Quick Start

← Back to Documentation Hub

Overview

Tasker Core is organized as a Cargo workspace with 7 member crates, each with a specific responsibility in the workflow orchestration system. This document explains the role of each crate, their inter-dependencies, and how they work together to provide a complete orchestration solution.

Design Philosophy

The crate structure follows these principles:

1. Separation of Concerns: Each crate has a well-defined responsibility
2. Minimal Dependencies: Crates depend on the minimum necessary dependencies
3. Shared Foundation: Common types and utilities in tasker-shared
4. Language Flexibility: Support for multiple worker implementations (Rust, Ruby, Python planned)
5. Production Ready: Workers and the orchestration system can be deployed and scaled independently

Workspace Structure

tasker-core/
├── tasker-pgmq/              # PGMQ wrapper with notification support
├── tasker-shared/            # Shared types, SQL functions, utilities
├── tasker-orchestration/     # Task coordination and lifecycle management
├── tasker-worker/            # Step execution and handler integration
├── tasker-client/            # API client library (REST + gRPC transport)
├── tasker-ctl/              # CLI binary (depends on tasker-client)
└── workers/
    ├── ruby/ext/tasker_core/ # Ruby FFI bindings
    └── rust/                 # Rust native worker

Crate Dependency Graph

┌─────────────────────────────────────────────────────────┐
│                   External Dependencies                 │
│  (sqlx, tokio, serde, pgmq, magnus, axum, etc.)       │
└─────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                    tasker-pgmq                          │
│  PGMQ wrapper with PostgreSQL LISTEN/NOTIFY            │
└─────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                    tasker-shared                        │
│  Core types, SQL functions, state machines             │
└─────────────────────────────────────────────────────────┘
                            │
               ┌────────────┴────────────┐
               │                         │
               ▼                         ▼
┌──────────────────────────┐  ┌──────────────────────────┐
│  tasker-orchestration    │  │    tasker-worker         │
│  Task coordination       │  │    Step execution        │
│  Lifecycle management    │  │    Handler integration   │
│  REST API                │  │    FFI support           │
└──────────────────────────┘  └──────────────────────────┘
               │                         │
               ▼                         │
┌──────────────────────────┐            │
│    tasker-client         │            │
│    API client library    │            │
│    REST + gRPC transport │            │
└──────────────────────────┘            │
               │                        │
               ▼                        │
┌──────────────────────────┐            │
│    tasker-ctl            │            │
│    CLI binary            │            │
└──────────────────────────┘            │
                                        │
               ┌────────────────────────┘
               │
      ┌────────┴────────┐
      ▼                 ▼
┌────────────┐  ┌────────────┐
│ workers/   │  │ workers/   │
│   ruby/    │  │   rust/    │
│   ext/     │  │            │
└────────────┘  └────────────┘

Core Crates

tasker-pgmq

Purpose: Wrapper around PostgreSQL Message Queue (PGMQ) with native PostgreSQL LISTEN/NOTIFY support

Location: tasker-pgmq/

Key Responsibilities:

• Wrap pgmq crate with notification capabilities
• Provide atomic pgmq_send_with_notify() operations
• Handle notification channel management
• Support namespace-aware queue naming

Public API:

#![allow(unused)]
fn main() {
pub struct PgmqClient {
    // Send message with atomic notification
    pub async fn send_with_notify<T>(&self, queue: &str, msg: T) -> Result<i64>;

    // Read message with visibility timeout
    pub async fn read<T>(&self, queue: &str, vt: i32) -> Result<Option<Message<T>>>;

    // Delete processed message
    pub async fn delete(&self, queue: &str, msg_id: i64) -> Result<bool>;
}
}

When to Use:

• When you need reliable message queuing with PostgreSQL
• When you need atomic send + notify operations
• When building event-driven systems on PostgreSQL

Dependencies:

• pgmq - Core PostgreSQL message queue functionality
• sqlx - Database connectivity
• tokio - Async runtime

tasker-shared

Purpose: Foundation crate containing all shared types, utilities, and SQL function interfaces

Location: tasker-shared/

Key Responsibilities:

• Core domain models (Task, WorkflowStep, TaskTransition, etc.)
• State machine implementations (Task + Step)
• SQL function executor and registry
• Database utilities and migrations
• Event system traits and types
• Messaging abstraction layer: Provider-agnostic messaging with PGMQ, RabbitMQ, and InMemory backends
• Factory system for testing
• Metrics and observability primitives

Public API:

#![allow(unused)]
fn main() {
// Core Models
pub mod models {
    pub struct Task { /* ... */ }
    pub struct WorkflowStep { /* ... */ }
    pub struct TaskTransition { /* ... */ }
    pub struct WorkflowStepTransition { /* ... */ }
}

// State Machines
pub mod state_machine {
    pub struct TaskStateMachine { /* ... */ }
    pub struct StepStateMachine { /* ... */ }
    pub enum TaskState { /* 12 states */ }
    pub enum WorkflowStepState { /* 9 states */ }
}

// SQL Functions
pub mod database {
    pub struct SqlFunctionExecutor { /* ... */ }
    pub async fn get_step_readiness_status(...) -> Result<Vec<StepReadinessStatus>>;
    pub async fn get_next_ready_tasks(...) -> Result<Vec<ReadyTaskInfo>>;
}

// Event System
pub mod event_system {
    pub trait EventDrivenSystem { /* ... */ }
    pub enum DeploymentMode { Hybrid, EventDrivenOnly, PollingOnly }
}

// Messaging
pub mod messaging {
    // Provider abstraction
    pub enum MessagingProvider { Pgmq, RabbitMq, InMemory }
    pub trait MessagingService { /* send_message, receive_messages, ack_message, ... */ }
    pub trait SupportsPushNotifications { /* subscribe, subscribe_many, requires_fallback_polling */ }
    pub enum MessageNotification { Available { ... }, Message(...) }

    // Domain client
    pub struct MessageClient { /* High-level queue operations */ }

    // Message types
    pub struct SimpleStepMessage { /* ... */ }
    pub struct TaskRequestMessage { /* ... */ }
    pub struct StepExecutionResult { /* ... */ }
}
}

When to Use:

• Always - This is the foundation for all other crates
• When you need core domain models
• When you need state machine logic
• When you need SQL function access
• When you need testing factories

Dependencies:

• tasker-pgmq - Message queue operations
• sqlx - Database operations
• serde - Serialization
• Many workspace-shared dependencies

Why It’s Separate:

• Eliminates circular dependencies between orchestration and worker
• Provides single source of truth for domain models
• Enables independent testing of core logic
• Allows multiple implementations (orchestration vs worker) to share code

tasker-orchestration

Purpose: Task coordination, lifecycle management, and orchestration REST API

Location: tasker-orchestration/

Key Responsibilities:

• Actor-based lifecycle coordination
• Task initialization and finalization
• Step discovery and enqueueing
• Result processing from workers
• Dynamic executor pool management
• Event-driven coordination
• REST API endpoints
• Health monitoring
• Metrics collection

Public API:

#![allow(unused)]
fn main() {
// Core orchestration
pub struct OrchestrationCore {
    pub async fn new() -> Result<Self>;
    pub async fn from_config(config: ConfigManager) -> Result<Self>;
}

// Actor-based coordination
pub mod actors {
    pub struct ActorRegistry { /* ... */ }
    pub struct TaskRequestActor { /* ... */ }
    pub struct ResultProcessorActor { /* ... */ }
    pub struct StepEnqueuerActor { /* ... */ }
    pub struct TaskFinalizerActor { /* ... */ }

    pub trait OrchestrationActor { /* ... */ }
    pub trait Handler<M: Message> { /* ... */ }
    pub trait Message { /* ... */ }
}

// Lifecycle services (wrapped by actors)
pub mod lifecycle {
    pub struct TaskInitializer { /* ... */ }
    pub struct StepEnqueuerService { /* ... */ }
    pub struct OrchestrationResultProcessor { /* ... */ }
    pub struct TaskFinalizer { /* ... */ }
}

// Message hydration (Phase 4)
pub mod hydration {
    pub struct StepResultHydrator { /* ... */ }
    pub struct TaskRequestHydrator { /* ... */ }
    pub struct FinalizationHydrator { /* ... */ }
}

// REST API (Axum)
pub mod web {
    // POST /v1/tasks
    pub async fn create_task(request: TaskRequest) -> Result<TaskResponse>;

    // GET /v1/tasks/{uuid}
    pub async fn get_task(uuid: Uuid) -> Result<TaskResponse>;

    // GET /health
    pub async fn health_check() -> Result<HealthResponse>;
}

// gRPC API (Tonic)
// Feature-gated behind `grpc-api`
pub mod grpc {
    pub struct GrpcServer { /* ... */ }
    pub struct GrpcState { /* wraps Arc<SharedApiServices> */ }

    pub mod services {
        pub struct TaskServiceImpl { /* 6 RPCs */ }
        pub struct StepServiceImpl { /* 4 RPCs */ }
        pub struct TemplateServiceImpl { /* 2 RPCs */ }
        pub struct HealthServiceImpl { /* 4 RPCs */ }
        pub struct AnalyticsServiceImpl { /* 2 RPCs */ }
        pub struct DlqServiceImpl { /* 6 RPCs */ }
        pub struct ConfigServiceImpl { /* 1 RPC */ }
    }

    pub mod interceptors {
        pub struct AuthInterceptor { /* Bearer token, API key */ }
    }
}

// Event systems
pub mod event_systems {
    pub struct OrchestrationEventSystem { /* ... */ }
    pub struct TaskReadinessEventSystem { /* ... */ }
}
}

Actor Architecture:

The orchestration crate implements a lightweight actor pattern for lifecycle component coordination:

• ActorRegistry: Manages all 4 orchestration actors with lifecycle hooks
• Message-Based Communication: Type-safe message handling via Handler<M> trait
• Service Decomposition: Large services decomposed into focused components (<300 lines per file)
• Direct Integration: Command processor calls actors directly without wrapper layers

See Actor-Based Architecture for comprehensive documentation.

When to Use:

• When you need to run the orchestration server
• When you need task coordination logic
• When building custom orchestration components
• When integrating with the REST API

Dependencies:

• tasker-shared - Core types and SQL functions
• tasker-pgmq - Message queuing
• axum - REST API framework
• tower-http - HTTP middleware

Deployment: Typically deployed as a server process (tasker-server binary)

Dual-Server Architecture:

Orchestration supports both REST and gRPC APIs running simultaneously via SharedApiServices:

#![allow(unused)]
fn main() {
pub struct SharedApiServices {
    pub security_service: Option<Arc<SecurityService>>,
    pub task_service: TaskService,
    pub step_service: StepService,
    pub health_service: HealthService,
    // ... other services
}

// Both APIs share the same service instances
AppState { services: Arc<SharedApiServices>, ... }   // REST
GrpcState { services: Arc<SharedApiServices>, ... }  // gRPC
}

Port Allocation:

• REST: 8080 (configurable)
• gRPC: 9190 (configurable)

tasker-worker

Purpose: Step execution, handler integration, and worker coordination

Location: tasker-worker/

Key Responsibilities:

• Claim steps from namespace queues
• Execute step handlers (Rust or FFI)
• Submit results to orchestration
• Template management and caching
• Event-driven step claiming
• Worker health monitoring
• FFI integration layer

Public API:

#![allow(unused)]
fn main() {
// Worker core
pub struct WorkerCore {
    pub async fn new(config: WorkerConfig) -> Result<Self>;
    pub async fn start(&mut self) -> Result<()>;
}

// Handler execution
pub mod handlers {
    pub trait StepHandler {
        async fn execute(&self, context: StepContext) -> Result<StepResult>;
    }
}

// Template management
pub mod task_template_manager {
    pub struct TaskTemplateManager {
        pub async fn load_templates(&mut self) -> Result<()>;
        pub fn get_template(&self, name: &str) -> Option<&TaskTemplate>;
    }
}

// Event systems
pub mod event_systems {
    pub struct WorkerEventSystem { /* ... */ }
}
}

When to Use:

• When you need to run a worker process
• When implementing custom step handlers
• When integrating with Ruby/Python handlers via FFI
• When building worker-specific tools

Dependencies:

• tasker-shared - Core types and messaging
• tasker-pgmq - Message queuing
• magnus (optional) - Ruby FFI bindings

Deployment: Deployed as worker processes, typically one per namespace or scaled horizontally

tasker-client

Purpose: Transport-agnostic API client library for REST and gRPC

Location: tasker-client/

Key Responsibilities:

• HTTP client for orchestration REST API
• gRPC client for orchestration gRPC API (feature-gated)
• Transport abstraction via unified client traits
• Configuration management and auth resolution
• Client-side request building

Public API:

#![allow(unused)]
fn main() {
// REST client
pub struct RestOrchestrationClient {
    pub async fn new(base_url: &str) -> Result<Self>;
    // Task, step, template, health operations
}

// gRPC client (feature-gated)
#[cfg(feature = "grpc")]
pub struct GrpcOrchestrationClient {
    pub async fn connect(endpoint: &str) -> Result<Self>;
    pub async fn connect_with_auth(endpoint: &str, auth: GrpcAuthConfig) -> Result<Self>;
    // Same operations as REST client
}

// Transport-agnostic client
pub enum UnifiedOrchestrationClient {
    Rest(Box<RestOrchestrationClient>),
    Grpc(Box<GrpcOrchestrationClient>),
}

// Client trait for transport abstraction
pub trait OrchestrationClient: Send + Sync {
    async fn create_task(&self, request: TaskRequest) -> Result<TaskResponse>;
    async fn get_task(&self, uuid: Uuid) -> Result<TaskResponse>;
    async fn list_tasks(&self, filters: TaskFilters) -> Result<Vec<TaskResponse>>;
    async fn health_check(&self) -> Result<HealthResponse>;
    // ... more operations
}
}

When to Use:

• When you need to interact with orchestration API from Rust
• When building integration tests
• When implementing client applications or FFI bindings
• When building UI frontends (TUI, web) that need API access

tasker-ctl

Purpose: Command-line interface for Tasker (split from tasker-client)

Location: tasker-ctl/

Key Responsibilities:

• CLI argument parsing and command dispatch (via clap)
• Task, worker, system, config, auth, and DLQ commands
• Configuration documentation generation (via askama, feature-gated)
• API key generation and management

CLI Tools:

# Task management
tasker-ctl task create --template linear_workflow
tasker-ctl task get <uuid>
tasker-ctl task list --namespace payments

# Health checks
tasker-ctl health

# Configuration docs generation
tasker-ctl docs generate

When to Use:

• When managing tasks from the command line
• When generating configuration documentation
• When performing administrative operations (auth, DLQ management)

Dependencies:

• reqwest - HTTP client
• clap - CLI argument parsing
• serde_json - JSON serialization

Worker Implementations

workers/ruby/ext/tasker_core

Purpose: Ruby FFI bindings enabling Ruby workers to execute Rust-orchestrated workflows

Location: workers/ruby/ext/tasker_core/

Key Responsibilities:

• Expose Rust worker functionality to Ruby via Magnus (FFI)
• Handle Ruby handler execution
• Manage Ruby <-> Rust type conversions
• Provide Ruby API for template registration
• FFI performance optimization

Ruby API:

# Worker bootstrap
result = TaskerCore::Worker::Bootstrap.start!

# Template registration (automatic)
# Ruby templates in workers/ruby/app/tasker/tasks/templates/

# Handler execution (automatic via FFI)
class MyHandler < TaskerCore::StepHandler
  def execute(context)
    # Step implementation
    { success: true, result: "done" }
  end
end

When to Use:

• When you have existing Ruby handlers
• When you need Ruby-specific libraries or gems
• When migrating from Ruby-based orchestration
• When team expertise is primarily Ruby

Dependencies:

• magnus - Ruby FFI bindings
• tasker-worker - Core worker logic
• Ruby runtime

Performance Considerations:

• FFI overhead: ~5-10ms per step (measured)
• Ruby GC can impact latency
• Thread-safe FFI calls via Ruby global lock
• Best for I/O-bound operations, not CPU-intensive

workers/rust

Purpose: Native Rust worker implementation for maximum performance

Location: workers/rust/

Key Responsibilities:

• Native Rust step handler execution
• Template definitions in Rust
• Direct integration with tasker-worker
• Maximum performance for CPU-intensive operations

Handler API:

#![allow(unused)]
fn main() {
// Define handler in Rust
pub struct MyHandler;

#[async_trait]
impl StepHandler for MyHandler {
    async fn execute(&self, context: StepContext) -> Result<StepResult> {
        // Step implementation
        Ok(StepResult::success(json!({"result": "done"})))
    }
}

// Register in template
pub fn register_template() -> TaskTemplate {
    TaskTemplate {
        name: "my_workflow",
        steps: vec![
            StepTemplate {
                name: "my_step",
                handler: Box::new(MyHandler),
                // ...
            }
        ]
    }
}
}

When to Use:

• When you need maximum performance
• For CPU-intensive operations
• When building new workflows in Rust
• When minimizing latency is critical

Dependencies:

• tasker-worker - Core worker logic
• tokio - Async runtime

Performance: Native Rust handlers have zero FFI overhead

Crate Relationships

How Crates Work Together

Task Creation Flow

Client Application
  ↓ [HTTP POST]
tasker-client
  ↓ [REST API]
tasker-orchestration::web
  ↓ [Task lifecycle]
tasker-orchestration::lifecycle::TaskInitializer
  ↓ [Uses]
tasker-shared::models::Task
tasker-shared::database::sql_functions
  ↓ [PostgreSQL]
Database + PGMQ

Step Execution Flow

tasker-orchestration::lifecycle::StepEnqueuer
  ↓ [pgmq_send_with_notify]
PGMQ namespace queue
  ↓ [pg_notify event]
tasker-worker::event_systems::WorkerEventSystem
  ↓ [Claims step]
tasker-worker::handlers::execute_handler
  ↓ [FFI or native]
workers/ruby or workers/rust
  ↓ [Returns result]
tasker-worker::orchestration_result_sender
  ↓ [pgmq_send_with_notify]
PGMQ orchestration_step_results queue
  ↓ [pg_notify event]
tasker-orchestration::lifecycle::ResultProcessor
  ↓ [Updates state]
tasker-shared::models::WorkflowStepTransition

Dependency Rationale

Why tasker-shared exists:

• Prevents circular dependencies (orchestration ↔ worker)
• Single source of truth for domain models
• Enables independent testing
• Allows SQL function reuse

Why workers are separate from tasker-worker:

• Language-specific implementations
• Independent deployment
• FFI boundary separation
• Multiple worker types supported

Why tasker-pgmq is separate:

• Reusable in other projects
• Focused responsibility
• Easy to test independently
• Can be published as separate crate

Building and Testing

Build All Crates

# Build everything with all features
cargo build --all-features

# Build specific crate
cargo build --package tasker-orchestration --all-features

# Build workspace root (minimal, mostly for integration)
cargo build

Test All Crates

# Test everything
cargo test --all-features

# Test specific crate
cargo test --package tasker-shared --all-features

# Test with database
DATABASE_URL="postgresql://..." cargo test --all-features

Feature Flags

# Root workspace features
[features]
benchmarks = [
    "tasker-shared/benchmarks",
    # ...
]
test-utils = [
    "tasker-orchestration/test-utils",
    "tasker-shared/test-utils",
    "tasker-worker/test-utils",
]

Migration Notes

Root Crate Being Phased Out

The root tasker-core crate (defined in the workspace root Cargo.toml) is being phased out:

• Current: Contains minimal code, mostly workspace configuration
• Future: Will be removed entirely, replaced by individual crates
• Impact: No functional impact, internal restructuring only
• Timeline: Complete when all functionality moved to member crates

Why: Cleaner workspace structure, better separation of concerns, easier to understand

Adding New Crates

When adding a new crate to the workspace:

1. Add to [workspace.members] in root Cargo.toml
2. Create crate: cargo new --lib tasker-new-crate
3. Add workspace dependencies to crate’s Cargo.toml
4. Update this documentation
5. Add to dependency graph above
6. Document public API

Best Practices

When to Create a New Crate

Create a new crate when:

• ✅ You have a distinct, reusable component
• ✅ You need independent versioning
• ✅ You want to reduce compile times
• ✅ You need isolation for testing
• ✅ You have language-specific implementations

Don’t create a new crate when:

• ❌ It’s tightly coupled to existing crates
• ❌ It’s only used in one place
• ❌ It would create circular dependencies
• ❌ It’s a small utility module

Dependency Management

• Use workspace dependencies: Define versions in root Cargo.toml
• Minimize dependencies: Only depend on what you need
• Version consistently: Use workspace = true in member crates
• Document dependencies: Explain why each dependency is needed

API Design

• Stable public API: Changes should be backward compatible
• Clear documentation: Every public item needs docs
• Examples in docs: Show how to use the API
• Error handling: Use Result with meaningful error types

Related Documentation

• Actor-Based Architecture - Actor pattern implementation in tasker-orchestration
• Messaging Abstraction - Provider-agnostic messaging
• Quick Start - Get running with the crates
• Events and Commands - How crates coordinate
• States and Lifecycles - State machines in tasker-shared
• Task Readiness & Execution - SQL functions in tasker-shared
• Archive: Ruby Integration Lessons - FFI patterns

← Back to Documentation Hub

Deployment Patterns and Configuration

Last Updated: 2026-01-15 Audience: Architects, Operators Status: Active Related Docs: Documentation Hub | Quick Start | Observability | Messaging Abstraction

← Back to Documentation Hub

Overview

Tasker Core supports three deployment modes, each optimized for different operational requirements and infrastructure constraints. This guide covers deployment patterns, configuration management, and production considerations.

Key Deployment Modes:

• Hybrid Mode (Recommended) - Event-driven with polling fallback
• EventDrivenOnly Mode - Pure event-driven for lowest latency
• PollingOnly Mode - Traditional polling for restricted environments

Messaging Backend Options:

• PGMQ (Default) - PostgreSQL-based, single infrastructure dependency
• RabbitMQ - AMQP broker, higher throughput for high-volume scenarios

Messaging Backend Selection

Tasker Core supports multiple messaging backends through a provider-agnostic abstraction layer. The choice of backend affects deployment architecture and operational requirements.

Backend Comparison

PGMQ (Default)

PostgreSQL Message Queue is the default backend, ideal for:

• Simpler deployments: Single database dependency
• Transactional workflows: Messages participate in PostgreSQL transactions
• Smaller to medium scale: Excellent for most workloads

Configuration:

# Default - no additional configuration needed
TASKER_MESSAGING_BACKEND=pgmq

Deployment Mode Interaction:

• Uses pg_notify for real-time notifications
• Fallback polling recommended for reliability
• Hybrid mode provides best balance

RabbitMQ

AMQP-based messaging for high-throughput scenarios:

• High-volume workloads: Better throughput characteristics
• Existing RabbitMQ infrastructure: Leverage existing investments
• Pure push delivery: No fallback polling required

Configuration:

TASKER_MESSAGING_BACKEND=rabbitmq
RABBITMQ_URL=amqp://user:password@rabbitmq:5672/%2F

Deployment Mode Interaction:

• EventDrivenOnly mode is natural fit (no fallback needed)
• Native push delivery via basic_consume()
• Protocol-guaranteed message delivery

Choosing a Backend

Decision Tree:
                              ┌─────────────────┐
                              │ Do you need the │
                              │ highest possible │
                              │ throughput?     │
                              └────────┬────────┘
                                       │
                            ┌──────────┴──────────┐
                            │                     │
                           Yes                    No
                            │                     │
                            ▼                     ▼
                   ┌────────────────┐   ┌────────────────┐
                   │ Do you have    │   │ Use PGMQ       │
                   │ existing       │   │ (simpler ops)  │
                   │ RabbitMQ?      │   └────────────────┘
                   └───────┬────────┘
                           │
                ┌──────────┴──────────┐
                │                     │
               Yes                    No
                │                     │
                ▼                     ▼
       ┌────────────────┐    ┌────────────────┐
       │ Use RabbitMQ   │    │ Evaluate       │
       └────────────────┘    │ operational    │
                             │ tradeoffs      │
                             └────────────────┘

Recommendation: Start with PGMQ. Migrate to RabbitMQ only when throughput requirements demand it.

Production Deployment Strategy: Mixed Mode Architecture

Important: In production-grade Kubernetes environments, you typically run multiple orchestration containers simultaneously with different deployment modes. This is not just about horizontal scaling with identical configurations—it’s about deploying containers with different coordination strategies to optimize for both throughput and reliability.

Recommended Production Pattern

High-Throughput + Safety Net Architecture:

# Most orchestration containers in EventDrivenOnly mode for maximum throughput
- EventDrivenOnly containers: 8-12 replicas (handles 80-90% of workload)
- PollingOnly containers: 2-3 replicas (safety net for missed events)

Why this works:

1. EventDrivenOnly containers handle the bulk of work with ~10ms latency
2. PollingOnly containers catch any events that might be missed during network issues or LISTEN/NOTIFY failures
3. Both sets of containers coordinate through atomic SQL operations (no conflicts)
4. Scale each mode independently based on throughput needs

Alternative: All-Hybrid Deployment

You can also deploy all containers in Hybrid mode and scale horizontally:

# All containers use Hybrid mode
- Hybrid containers: 10-15 replicas

This is simpler but less flexible. The mixed-mode approach lets you:

• Tune for specific workload patterns (event-heavy vs. polling-heavy)
• Adapt to infrastructure constraints (some networks better for events, others for polling)
• Optimize resource usage (EventDrivenOnly uses less CPU than Hybrid)
• Scale dimensions independently (scale up event listeners without scaling pollers)

Key Insight

The different deployment modes exist not just for config tuning, but to enable sophisticated deployment strategies where you mix coordination approaches across containers to meet production throughput and reliability requirements.

Deployment Mode Comparison

Hybrid Mode (Recommended)

Overview

Hybrid mode combines the best of both worlds: event-driven coordination for real-time performance with polling fallback for reliability.

How it works:

1. PostgreSQL LISTEN/NOTIFY provides real-time event notifications
2. If event listeners fail or lag, polling automatically takes over
3. System continuously monitors and switches between modes
4. No manual intervention required

Configuration

# config/tasker/base/orchestration.toml
[orchestration]
deployment_mode = "Hybrid"

[orchestration.hybrid]
# Event listener settings
enable_event_listeners = true
listener_reconnect_interval_ms = 5000
listener_health_check_interval_ms = 30000

# Polling fallback settings
enable_polling_fallback = true
polling_interval_ms = 1000
fallback_activation_threshold_ms = 5000

# Worker event settings
[orchestration.worker_events]
enable_worker_listeners = true
worker_listener_reconnect_ms = 5000

When to Use Hybrid Mode

Ideal for:

• Production deployments requiring high reliability
• Environments with occasional network instability
• Systems requiring both low latency and guaranteed delivery
• Multi-region deployments with variable network quality

Example: Production E-commerce Platform

# docker-compose.production.yml
version: '3.8'

services:
  orchestration:
    image: tasker-orchestration:latest
    environment:
      - TASKER_ENV=production
      - TASKER_DEPLOYMENT_MODE=Hybrid
      - DATABASE_URL=postgresql://tasker:${DB_PASSWORD}@postgres:5432/tasker_production
      - RUST_LOG=info
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '2'
          memory: 2G
        reservations:
          cpus: '1'
          memory: 1G
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 10s
      timeout: 5s
      retries: 3

  postgres:
    image: postgres:16
    environment:
      - POSTGRES_DB=tasker_production
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres-data:/var/lib/postgresql/data
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 8G

volumes:
  postgres-data:

Monitoring Hybrid Mode

Key Metrics:

#![allow(unused)]
fn main() {
// Hybrid mode health indicators
tasker_event_listener_active{mode="hybrid"} = 1           // Listener is active
tasker_event_listener_lag_ms{mode="hybrid"} < 100         // Event lag is acceptable
tasker_polling_fallback_active{mode="hybrid"} = 0         // Not in fallback mode
tasker_mode_switches_total{mode="hybrid"} < 10/hour       // Infrequent mode switching
}

Alert conditions:

• Event listener down for > 60 seconds
• Polling fallback active for > 5 minutes
• Mode switches > 20 per hour (indicates instability)

EventDrivenOnly Mode

Overview

EventDrivenOnly mode provides the lowest possible latency by relying entirely on PostgreSQL LISTEN/NOTIFY for coordination.

How it works:

1. Orchestration and workers establish persistent PostgreSQL connections
2. LISTEN on specific channels for events
3. Immediate notification on queue changes
4. No polling overhead or delay

Configuration

# config/tasker/base/orchestration.toml
[orchestration]
deployment_mode = "EventDrivenOnly"

[orchestration.event_driven]
# Listener configuration
listener_reconnect_interval_ms = 2000
listener_health_check_interval_ms = 15000
max_reconnect_attempts = 10

# Event channels
channels = [
    "pgmq_message_ready.orchestration",
    "pgmq_message_ready.*",
    "pgmq_queue_created"
]

# Connection pool for listeners
listener_pool_size = 5
connection_timeout_ms = 5000

When to Use EventDrivenOnly Mode

Ideal for:

• High-throughput, low-latency requirements
• Stable network environments
• Development and testing environments
• Systems with reliable PostgreSQL infrastructure

Not recommended for:

• Unstable network connections
• Environments with frequent PostgreSQL failovers
• Systems requiring guaranteed operation during network issues

Example: High-Performance Payment Processing

#![allow(unused)]
fn main() {
// Worker configuration for event-driven mode
use tasker_worker::WorkerConfig;

let config = WorkerConfig {
    deployment_mode: DeploymentMode::EventDrivenOnly,
    namespaces: vec!["payments".to_string()],
    event_driven_settings: EventDrivenSettings {
        listener_reconnect_interval_ms: 2000,
        health_check_interval_ms: 15000,
        max_reconnect_attempts: 10,
    },
    ..Default::default()
};

// Start worker with event-driven mode
let worker = WorkerCore::from_config(config).await?;
worker.start().await?;
}

Monitoring EventDrivenOnly Mode

Critical Metrics:

#![allow(unused)]
fn main() {
// Event-driven health indicators
tasker_event_listener_active{mode="event_driven"} = 1    // Must be 1
tasker_event_notifications_received_total                 // Should be > 0
tasker_event_processing_duration_seconds                  // Should be < 0.01
tasker_listener_reconnections_total                       // Should be low
}

Alert conditions:

• Event listener inactive
• No events received for > 60 seconds (when activity expected)
• Reconnections > 5 per hour

PollingOnly Mode

Overview

PollingOnly mode provides the most reliable operation in restricted or unstable network environments by using traditional polling.

How it works:

1. Orchestration and workers poll message queues at regular intervals
2. No dependency on persistent connections or LISTEN/NOTIFY
3. Configurable polling intervals for performance/resource trade-offs
4. Automatic retry and backoff on failures

Configuration

# config/tasker/base/orchestration.toml
[orchestration]
deployment_mode = "PollingOnly"

[orchestration.polling]
# Polling intervals
task_request_poll_interval_ms = 1000
step_result_poll_interval_ms = 500
finalization_poll_interval_ms = 2000

# Batch processing
batch_size = 10
max_messages_per_poll = 100

# Backoff on errors
error_backoff_base_ms = 1000
error_backoff_max_ms = 30000
error_backoff_multiplier = 2.0

When to Use PollingOnly Mode

Ideal for:

• Restricted network environments (firewalls blocking persistent connections)
• Environments with frequent PostgreSQL connection issues
• Systems prioritizing reliability over latency
• Legacy infrastructure with limited LISTEN/NOTIFY support

Not recommended for:

• High-frequency, low-latency requirements
• Systems with strict resource constraints
• Environments where polling overhead is problematic

Example: Batch Processing System

# config/tasker/environments/production/orchestration.toml
[orchestration]
deployment_mode = "PollingOnly"

[orchestration.polling]
# Longer intervals for batch processing
task_request_poll_interval_ms = 5000
step_result_poll_interval_ms = 2000
finalization_poll_interval_ms = 10000

# Large batches for efficiency
batch_size = 50
max_messages_per_poll = 500

Monitoring PollingOnly Mode

Key Metrics:

#![allow(unused)]
fn main() {
// Polling health indicators
tasker_polling_cycles_total                               // Should be increasing
tasker_polling_messages_processed_total                   // Should be > 0
tasker_polling_duration_seconds                           // Should be stable
tasker_polling_errors_total                               // Should be low
}

Alert conditions:

• Polling stopped (no cycles in last 60 seconds)
• Polling duration > 10x interval (indicates overload)
• Error rate > 5% of polling cycles

Configuration Management

Component-Based Configuration

Tasker Core uses a component-based TOML configuration system with environment-specific overrides.

Configuration Structure:

config/tasker/
├── base/                          # Base configuration (all environments)
│   ├── database.toml             # Database connection pool settings
│   ├── orchestration.toml        # Orchestration and deployment mode
│   ├── circuit_breakers.toml    # Circuit breaker thresholds
│   ├── executor_pools.toml      # Executor pool sizing
│   ├── pgmq.toml                # Message queue configuration
│   ├── query_cache.toml         # Query caching settings
│   └── telemetry.toml           # Metrics and logging
│
└── environments/                  # Environment-specific overrides
    ├── development/
    │   └── *.toml               # Development overrides
    ├── test/
    │   └── *.toml               # Test overrides
    └── production/
        └── *.toml               # Production overrides

Environment Detection

# Set environment via TASKER_ENV
export TASKER_ENV=production

# Validate configuration
cargo run --bin config-validator

# Expected output:
# ✓ Configuration loaded successfully
# ✓ Environment: production
# ✓ Deployment mode: Hybrid
# ✓ Database pool: 50 connections
# ✓ Circuit breakers: 10 configurations

Example: Production Configuration

# config/tasker/environments/production/orchestration.toml
[orchestration]
deployment_mode = "Hybrid"
max_concurrent_tasks = 1000
task_timeout_seconds = 3600

[orchestration.hybrid]
enable_event_listeners = true
enable_polling_fallback = true
polling_interval_ms = 2000
fallback_activation_threshold_ms = 10000

[orchestration.health]
health_check_interval_ms = 30000
unhealthy_threshold = 3
recovery_threshold = 2

# config/tasker/environments/production/database.toml
[database]
max_connections = 50
min_connections = 10
connection_timeout_ms = 5000
idle_timeout_seconds = 600
max_lifetime_seconds = 1800

[database.query_cache]
enabled = true
max_size = 1000
ttl_seconds = 300

# config/tasker/environments/production/circuit_breakers.toml
[circuit_breakers.database]
enabled = true
error_threshold = 5
timeout_seconds = 60
half_open_timeout_seconds = 30

[circuit_breakers.message_queue]
enabled = true
error_threshold = 10
timeout_seconds = 120
half_open_timeout_seconds = 60

Docker Compose Deployment

Development Setup

# docker-compose.yml
version: '3.8'

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: tasker
      POSTGRES_PASSWORD: tasker
      POSTGRES_DB: tasker_rust_test
    ports:
      - "5432:5432"
    volumes:
      - postgres-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U tasker"]
      interval: 5s
      timeout: 5s
      retries: 5

  orchestration:
    build:
      context: .
      target: orchestration
    environment:
      - TASKER_ENV=development
      - DATABASE_URL=postgresql://tasker:tasker@postgres:5432/tasker_rust_test
      - RUST_LOG=debug
    ports:
      - "8080:8080"
    depends_on:
      postgres:
        condition: service_healthy
    profiles:
      - server

  worker:
    build:
      context: .
      target: worker
    environment:
      - TASKER_ENV=development
      - DATABASE_URL=postgresql://tasker:tasker@postgres:5432/tasker_rust_test
      - RUST_LOG=debug
    ports:
      - "8081:8081"
    depends_on:
      postgres:
        condition: service_healthy
    profiles:
      - server

  ruby-worker:
    build:
      context: ./workers/ruby
      dockerfile: Dockerfile
    environment:
      - TASKER_ENV=development
      - DATABASE_URL=postgresql://tasker:tasker@postgres:5432/tasker_rust_test
      - RUST_LOG=debug
    ports:
      - "8082:8082"
    depends_on:
      postgres:
        condition: service_healthy
    profiles:
      - server

volumes:
  postgres-data:

Production Deployment

# docker-compose.production.yml
version: '3.8'

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: tasker
      POSTGRES_PASSWORD_FILE: /run/secrets/db_password
      POSTGRES_DB: tasker_production
    volumes:
      - postgres-data:/var/lib/postgresql/data
    deploy:
      placement:
        constraints:
          - node.labels.database == true
      resources:
        limits:
          cpus: '4'
          memory: 8G
        reservations:
          cpus: '2'
          memory: 4G
    secrets:
      - db_password

  orchestration:
    image: tasker-orchestration:${VERSION}
    environment:
      - TASKER_ENV=production
      - DATABASE_URL_FILE=/run/secrets/database_url
    deploy:
      replicas: 3
      update_config:
        parallelism: 1
        delay: 10s
        order: start-first
      rollback_config:
        parallelism: 0
        order: stop-first
      resources:
        limits:
          cpus: '2'
          memory: 2G
        reservations:
          cpus: '1'
          memory: 1G
    secrets:
      - database_url

  worker:
    image: tasker-worker:${VERSION}
    environment:
      - TASKER_ENV=production
      - DATABASE_URL_FILE=/run/secrets/database_url
    deploy:
      replicas: 5
      resources:
        limits:
          cpus: '1'
          memory: 1G
        reservations:
          cpus: '0.5'
          memory: 512M
    secrets:
      - database_url

secrets:
  db_password:
    external: true
  database_url:
    external: true

volumes:
  postgres-data:
    driver: local

Kubernetes Deployment

Mixed-Mode Production Deployment (Recommended)

This example demonstrates the recommended production pattern: multiple orchestration deployments with different modes.

# k8s/orchestration-event-driven.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: tasker-orchestration-event-driven
  namespace: tasker
  labels:
    app: tasker-orchestration
    mode: event-driven
spec:
  replicas: 10  # Majority of orchestration capacity
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 2
      maxUnavailable: 0
  selector:
    matchLabels:
      app: tasker-orchestration
      mode: event-driven
  template:
    metadata:
      labels:
        app: tasker-orchestration
        mode: event-driven
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: orchestration
        image: tasker-orchestration:1.0.0
        env:
        - name: TASKER_ENV
          value: "production"
        - name: DEPLOYMENT_MODE
          value: "EventDrivenOnly"  # High-throughput mode
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: tasker-secrets
              key: database-url
        - name: RUST_LOG
          value: "info"
        ports:
        - containerPort: 8080
          name: http
        resources:
          requests:
            cpu: 500m      # Lower CPU for event-driven
            memory: 512Mi
          limits:
            cpu: 1000m
            memory: 1Gi
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5

---
# k8s/orchestration-polling.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: tasker-orchestration-polling
  namespace: tasker
  labels:
    app: tasker-orchestration
    mode: polling
spec:
  replicas: 3  # Safety net for missed events
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: tasker-orchestration
      mode: polling
  template:
    metadata:
      labels:
        app: tasker-orchestration
        mode: polling
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: orchestration
        image: tasker-orchestration:1.0.0
        env:
        - name: TASKER_ENV
          value: "production"
        - name: DEPLOYMENT_MODE
          value: "PollingOnly"  # Reliability safety net
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: tasker-secrets
              key: database-url
        - name: RUST_LOG
          value: "info"
        ports:
        - containerPort: 8080
          name: http
        resources:
          requests:
            cpu: 750m      # Higher CPU for polling
            memory: 512Mi
          limits:
            cpu: 1500m
            memory: 1Gi
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5

---
# k8s/orchestration-service.yaml
apiVersion: v1
kind: Service
metadata:
  name: tasker-orchestration
  namespace: tasker
spec:
  selector:
    app: tasker-orchestration  # Matches BOTH deployments
  ports:
  - port: 8080
    targetPort: 8080
    protocol: TCP
    name: http
  type: ClusterIP

Key points about this mixed-mode deployment:

1. 10 EventDrivenOnly pods handle 80-90% of work with ~10ms latency
2. 3 PollingOnly pods catch anything missed by event listeners
3. Single service load balances across all 13 pods
4. No conflicts - atomic SQL operations prevent duplicate processing
5. Independent scaling - scale event-driven pods for throughput, polling pods for reliability

Single-Mode Orchestration Deployment

# k8s/orchestration-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: tasker-orchestration
  namespace: tasker
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: tasker-orchestration
  template:
    metadata:
      labels:
        app: tasker-orchestration
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: orchestration
        image: tasker-orchestration:1.0.0
        env:
        - name: TASKER_ENV
          value: "production"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: tasker-secrets
              key: database-url
        - name: RUST_LOG
          value: "info"
        ports:
        - containerPort: 8080
          name: http
          protocol: TCP
        resources:
          requests:
            cpu: 1000m
            memory: 1Gi
          limits:
            cpu: 2000m
            memory: 2Gi
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 2

---
apiVersion: v1
kind: Service
metadata:
  name: tasker-orchestration
  namespace: tasker
spec:
  selector:
    app: tasker-orchestration
  ports:
  - port: 8080
    targetPort: 8080
    protocol: TCP
    name: http
  type: ClusterIP

Worker Deployment

# k8s/worker-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: tasker-worker-payments
  namespace: tasker
spec:
  replicas: 5
  selector:
    matchLabels:
      app: tasker-worker
      namespace: payments
  template:
    metadata:
      labels:
        app: tasker-worker
        namespace: payments
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8081"
    spec:
      containers:
      - name: worker
        image: tasker-worker:1.0.0
        env:
        - name: TASKER_ENV
          value: "production"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: tasker-secrets
              key: database-url
        - name: RUST_LOG
          value: "info"
        - name: WORKER_NAMESPACES
          value: "payments"
        ports:
        - containerPort: 8081
          name: http
          protocol: TCP
        resources:
          requests:
            cpu: 500m
            memory: 512Mi
          limits:
            cpu: 1000m
            memory: 1Gi
        livenessProbe:
          httpGet:
            path: /health
            port: 8081
          initialDelaySeconds: 20
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8081
          initialDelaySeconds: 5
          periodSeconds: 5

---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: tasker-worker-payments
  namespace: tasker
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: tasker-worker-payments
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

Health Monitoring

Health Check Endpoints

Orchestration Health:

# Basic health check
curl http://localhost:8080/health

# Response:
{
  "status": "healthy",
  "database": "connected",
  "message_queue": "operational"
}

# Detailed health check
curl http://localhost:8080/health/detailed

# Response:
{
  "status": "healthy",
  "deployment_mode": "Hybrid",
  "event_listeners": {
    "active": true,
    "channels": 3,
    "lag_ms": 12
  },
  "polling": {
    "active": false,
    "fallback_triggered": false
  },
  "database": {
    "status": "connected",
    "pool_size": 50,
    "active_connections": 23
  },
  "circuit_breakers": {
    "database": "closed",
    "message_queue": "closed"
  },
  "executors": {
    "task_initializer": {
      "active": 3,
      "max": 10,
      "queue_depth": 5
    },
    "result_processor": {
      "active": 5,
      "max": 10,
      "queue_depth": 12
    }
  }
}

Worker Health:

# Worker health check
curl http://localhost:8081/health

# Response:
{
  "status": "healthy",
  "namespaces": ["payments", "inventory"],
  "active_executions": 8,
  "claimed_steps": 3
}

Kubernetes Probes

# Liveness probe - restart if unhealthy
livenessProbe:
  httpGet:
    path: /health
    port: 8080
  initialDelaySeconds: 30
  periodSeconds: 10
  timeoutSeconds: 5
  failureThreshold: 3

# Readiness probe - remove from load balancer if not ready
readinessProbe:
  httpGet:
    path: /health
    port: 8080
  initialDelaySeconds: 10
  periodSeconds: 5
  timeoutSeconds: 3
  failureThreshold: 2

gRPC Health Checks

Tasker Core exposes gRPC health endpoints alongside REST for Kubernetes gRPC health probes.

Port Allocation:

gRPC Health Endpoints:

# Using grpcurl
grpcurl -plaintext localhost:9190 tasker.v1.HealthService/CheckLiveness
grpcurl -plaintext localhost:9190 tasker.v1.HealthService/CheckReadiness
grpcurl -plaintext localhost:9190 tasker.v1.HealthService/CheckDetailedHealth

Kubernetes gRPC Probes (Kubernetes 1.24+):

# gRPC liveness probe
livenessProbe:
  grpc:
    port: 9190
    service: tasker.v1.HealthService
  initialDelaySeconds: 30
  periodSeconds: 10
  timeoutSeconds: 5
  failureThreshold: 3

# gRPC readiness probe
readinessProbe:
  grpc:
    port: 9190
    service: tasker.v1.HealthService
  initialDelaySeconds: 10
  periodSeconds: 5
  timeoutSeconds: 3
  failureThreshold: 2

Configuration (config/tasker/base/orchestration.toml):

[orchestration.grpc]
enabled = true
bind_address = "${TASKER_ORCHESTRATION_GRPC_BIND_ADDRESS:-0.0.0.0:9190}"
enable_reflection = true       # Service discovery via grpcurl
enable_health_service = true   # gRPC health checks

Scaling Patterns

Horizontal Scaling

Mixed-Mode Orchestration Scaling (Recommended)

Scale different deployment modes independently to optimize for throughput and reliability:

# Scale event-driven pods for throughput
kubectl scale deployment tasker-orchestration-event-driven --replicas=15 -n tasker

# Scale polling pods for reliability
kubectl scale deployment tasker-orchestration-polling --replicas=5 -n tasker

Scaling strategy by workload:

Single-Mode Orchestration Scaling

If using single deployment mode (Hybrid or EventDrivenOnly):

# Scale orchestration to 10 replicas (all same mode)
kubectl scale deployment tasker-orchestration --replicas=10 -n tasker

Key principles:

• Multiple orchestration instances process tasks independently
• Atomic finalization claiming prevents duplicate processing
• Load balancer distributes API requests across instances

Worker Scaling

Workers scale independently per namespace:

# Scale payment workers to 10 replicas
kubectl scale deployment tasker-worker-payments --replicas=10 -n tasker

• Each worker claims steps from namespace-specific queues
• No coordination required between workers
• Scale per namespace based on queue depth

Vertical Scaling

Resource Allocation:

# High-throughput orchestration
resources:
  requests:
    cpu: 2000m
    memory: 4Gi
  limits:
    cpu: 4000m
    memory: 8Gi

# Standard worker
resources:
  requests:
    cpu: 500m
    memory: 512Mi
  limits:
    cpu: 1000m
    memory: 1Gi

Auto-Scaling

HPA Configuration:

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: tasker-orchestration
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: tasker-orchestration
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: tasker_tasks_active
      target:
        type: AverageValue
        averageValue: "100"
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 50
        periodSeconds: 60
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 30

Production Considerations

Database Configuration

Connection Pooling:

# config/tasker/environments/production/database.toml
[database]
max_connections = 50              # Total pool size
min_connections = 10              # Minimum maintained connections
connection_timeout_ms = 5000      # Connection acquisition timeout
idle_timeout_seconds = 600        # Close idle connections after 10 minutes
max_lifetime_seconds = 1800       # Recycle connections after 30 minutes

Calculation:

Total DB Connections = (Orchestration Replicas × Pool Size) + (Worker Replicas × Pool Size)
Example: (3 × 50) + (10 × 20) = 350 connections

Ensure PostgreSQL max_connections > Total DB Connections + Buffer
Recommended: max_connections = 500 for above example

Circuit Breaker Tuning

# config/tasker/environments/production/circuit_breakers.toml
[circuit_breakers.database]
enabled = true
error_threshold = 5               # Open after 5 consecutive errors
timeout_seconds = 60              # Stay open for 60 seconds
half_open_timeout_seconds = 30    # Test recovery for 30 seconds

[circuit_breakers.message_queue]
enabled = true
error_threshold = 10
timeout_seconds = 120
half_open_timeout_seconds = 60

Executor Pool Sizing

# config/tasker/environments/production/executor_pools.toml
[executor_pools.task_initializer]
min_executors = 2
max_executors = 10
queue_high_watermark = 100
queue_low_watermark = 10

[executor_pools.result_processor]
min_executors = 5
max_executors = 20
queue_high_watermark = 200
queue_low_watermark = 20

[executor_pools.step_enqueuer]
min_executors = 3
max_executors = 15
queue_high_watermark = 150
queue_low_watermark = 15

Observability Integration

Prometheus Metrics:

# Prometheus scrape config
scrape_configs:
  - job_name: 'tasker-orchestration'
    kubernetes_sd_configs:
      - role: pod
        namespaces:
          names:
            - tasker
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
        action: keep
        regex: true
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
        action: replace
        target_label: __metrics_path__
        regex: (.+)
      - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port]
        action: replace
        regex: ([^:]+)(?::\d+)?;(\d+)
        replacement: $1:$2
        target_label: __address__

Key Alerts:

# alerts.yaml
groups:
  - name: tasker
    interval: 30s
    rules:
      - alert: TaskerOrchestrationDown
        expr: up{job="tasker-orchestration"} == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Tasker orchestration instance down"

      - alert: TaskerHighErrorRate
        expr: rate(tasker_step_errors_total[5m]) > 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High error rate in step execution"

      - alert: TaskerCircuitBreakerOpen
        expr: tasker_circuit_breaker_state{state="open"} == 1
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Circuit breaker {{ $labels.name }} is open"

      - alert: TaskerDatabasePoolExhausted
        expr: tasker_database_pool_active >= tasker_database_pool_max
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Database connection pool exhausted"

Migration Strategies

Migrating to Hybrid Mode

Step 1: Enable event listeners

# config/tasker/environments/production/orchestration.toml
[orchestration]
deployment_mode = "Hybrid"

[orchestration.hybrid]
enable_event_listeners = true
enable_polling_fallback = true    # Keep polling enabled during migration

Step 2: Monitor event listener health

# Check metrics for event listener stability
curl http://localhost:8080/health/detailed | jq '.event_listeners'

Step 3: Gradually reduce polling frequency

# Once event listeners are stable
[orchestration.hybrid]
polling_interval_ms = 5000        # Increase from 1000ms to 5000ms

Step 4: Validate performance

• Monitor latency metrics: tasker_step_discovery_duration_seconds
• Verify no missed events: tasker_polling_messages_found_total should be near zero

Rollback Plan

If event-driven mode fails:

# Immediate rollback to PollingOnly
[orchestration]
deployment_mode = "PollingOnly"

[orchestration.polling]
task_request_poll_interval_ms = 500    # Aggressive polling

Gradual rollback:

1. Increase polling frequency in Hybrid mode
2. Monitor for stability
3. Disable event listeners once polling is stable
4. Switch to PollingOnly mode

Troubleshooting

Event Listener Issues

Problem: Event listeners not receiving notifications

Diagnosis:

-- Check PostgreSQL LISTEN/NOTIFY is working
NOTIFY pgmq_message_ready, 'test';

# Check listener status
curl http://localhost:8080/health/detailed | jq '.event_listeners'

Solutions:

• Verify PostgreSQL version supports LISTEN/NOTIFY (9.0+)
• Check firewall rules allow persistent connections
• Increase listener_reconnect_interval_ms if connections drop frequently
• Switch to Hybrid or PollingOnly mode if issues persist

Polling Performance Issues

Problem: High CPU usage from polling

Diagnosis:

# Check polling frequency and batch sizes
curl http://localhost:8080/health/detailed | jq '.polling'

Solutions:

• Increase polling intervals
• Increase batch sizes to process more messages per poll
• Switch to Hybrid or EventDrivenOnly mode for better performance
• Scale horizontally to distribute polling load

Database Connection Exhaustion

Problem: “connection pool exhausted” errors

Diagnosis:

-- Check active connections
SELECT count(*) FROM pg_stat_activity WHERE datname = 'tasker_production';

-- Check max connections
SHOW max_connections;

Solutions:

• Increase max_connections in database.toml
• Increase PostgreSQL max_connections setting
• Reduce number of replicas
• Implement connection pooling at infrastructure level (PgBouncer)

Best Practices

Configuration Management

1. Use environment-specific overrides instead of modifying base configuration
2. Validate configuration with config-validator before deployment
3. Version control all configuration including environment overrides
4. Use secrets management for sensitive values (passwords, keys)

Deployment Strategy

1. Use mixed-mode architecture in production (EventDrivenOnly + PollingOnly)
   • Deploy 80-90% of orchestration pods in EventDrivenOnly mode for throughput
   • Deploy 10-20% of orchestration pods in PollingOnly mode as safety net
   • Single service load balances across all pods
2. Alternative: Deploy all pods in Hybrid mode for simpler configuration
   • Trade-off: Less tuning flexibility, slightly higher resource usage
3. Scale each mode independently based on workload characteristics
4. Monitor deployment mode metrics to adjust ratios over time
5. Test mixed-mode deployments in staging before production

Deployment Operations

1. Always test configuration changes in staging first
2. Use rolling updates with health checks to prevent downtime
3. Monitor deployment mode health during and after deployments
4. Keep polling capacity available even when event-driven is primary

Scaling Guidelines

1. Mixed-mode orchestration: Scale EventDrivenOnly and PollingOnly deployments independently
   • Scale event-driven pods based on throughput requirements
   • Scale polling pods based on reliability requirements
2. Single-mode orchestration: Scale based on API request rate and task initialization throughput
3. Workers: Scale based on namespace-specific queue depth
4. Database connections: Monitor and adjust pool sizes as replicas scale
5. Use HPA for automatic scaling based on CPU/memory and custom metrics

Observability

1. Enable comprehensive metrics in production
2. Set up alerts for circuit breaker states, connection pool exhaustion
3. Monitor deployment mode distribution in mixed-mode deployments
4. Track event listener lag in EventDrivenOnly and Hybrid modes
5. Monitor polling overhead to optimize resource usage
6. Track step execution latency per namespace and handler

Summary

Tasker Core’s flexible deployment modes enable sophisticated production architectures:

Deployment Modes

• Hybrid Mode: Event-driven with polling fallback in a single container
• EventDrivenOnly Mode: Maximum throughput with ~10ms latency
• PollingOnly Mode: Reliable safety net with traditional polling

Recommended Production Pattern

Mixed-Mode Architecture (recommended for production at scale):

• Deploy majority of orchestration pods in EventDrivenOnly mode for high throughput
• Deploy minority of orchestration pods in PollingOnly mode as reliability safety net
• Both deployments coordinate through atomic SQL operations with no conflicts
• Scale each mode independently based on workload characteristics

Alternative: Deploy all pods in Hybrid mode for simpler configuration with automatic fallback.

The key insight: deployment modes exist not just for configuration tuning, but to enable mixing coordination strategies across containers to meet production requirements for both throughput and reliability.

← Back to Documentation Hub

Next: Observability | Benchmarks | Quick Start

Domain Events Architecture

Last Updated: 2025-12-01 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | Events and Commands | Observability | States and Lifecycles

← Back to Documentation Hub

This document provides comprehensive documentation of the domain event system in tasker-core, covering event delivery modes, publisher patterns, subscriber implementations, and integration with the workflow orchestration system.

Overview

Domain Events vs System Events

The tasker-core system distinguishes between two types of events:

System events handle internal workflow coordination: task initialization, step enqueueing, result processing, and finalization. These are documented in Events and Commands.

Domain events enable business observability: payment processed, order fulfilled, inventory updated. Step handlers publish these events to enable external systems to react to business outcomes.

Key Design Principle: Non-Blocking Publication

Domain event publishing never fails the step. This is a fundamental design decision:

• Event publish errors are logged with warn! level
• Step execution continues regardless of publish outcome
• Business logic success is independent of event delivery
• A handler that successfully processes a payment should not fail if event publishing fails

#![allow(unused)]
fn main() {
// Event publishing is fire-and-forget
if let Err(e) = publisher.publish_event(event_name, payload, metadata).await {
    warn!(
        handler = self.handler_name(),
        event_name = event_name,
        error = %e,
        "Failed to publish domain event - step will continue"
    );
}
// Step continues regardless of publish result
}

Architecture

Data Flow

flowchart TB
    subgraph handlers["Step Handlers"]
        SH["Step Handler<br/>(Rust/Ruby)"]
    end

    SH -->|"publish_domain_event(name, payload)"| ER

    subgraph routing["Event Routing"]
        ER["EventRouter<br/>(Delivery Mode)"]
    end

    ER --> Durable
    ER --> Fast
    ER --> Broadcast

    subgraph modes["Delivery Modes"]
        Durable["Durable<br/>(PGMQ)"]
        Fast["Fast<br/>(In-Process)"]
        Broadcast["Broadcast<br/>(Both Paths)"]
    end

    Durable --> NEQ["Namespace<br/>Event Queue"]
    Fast --> IPB["InProcessEventBus"]
    Broadcast --> NEQ
    Broadcast --> IPB

    subgraph external["External Integration"]
        NEQ --> EC["External Consumer<br/>(Polling)"]
    end

    subgraph internal["Internal Subscribers"]
        IPB --> RS["Rust<br/>Subscribers"]
        IPB --> RFF["Ruby FFI<br/>Channel"]
    end

    style handlers fill:#e1f5fe
    style routing fill:#fff3e0
    style modes fill:#f3e5f5
    style external fill:#ffebee
    style internal fill:#e8f5e9

Component Summary

Delivery Modes

Overview

The domain event system supports three delivery modes, configured per event in YAML templates:

Durable Mode (PGMQ) - External Integration Boundary

Durable events define the integration boundary between Tasker and external systems. Events are published to namespace-specific PGMQ queues where external consumers can poll and process them.

Key Design Decision: Tasker does NOT consume durable events internally. PGMQ serves as a lightweight, PostgreSQL-native alternative to external message brokers (Kafka, AWS SNS/SQS, RabbitMQ). External systems or middleware proxies can:

• Poll PGMQ queues directly
• Forward events to Kafka, SNS/SQS, or other messaging systems
• Implement custom event processing pipelines

payment.processed → payments_domain_events (PGMQ queue) → External Systems
order.fulfilled   → fulfillment_domain_events (PGMQ queue) → External Systems

Characteristics:

• Persisted in PostgreSQL (survives restarts)
• For external consumer integration only
• No internal Tasker polling or subscription
• Consumer acknowledgment and retry handled by external consumers
• Ordered within namespace

Implementation:

#![allow(unused)]
fn main() {
// DomainEventPublisher routes durable events to PGMQ
pub async fn publish_event(
    &self,
    event_name: &str,
    payload: Value,
    metadata: EventMetadata,
) -> TaskerResult<()> {
    let queue_name = format!("{}_domain_events", metadata.namespace);
    let message = DomainEventMessage {
        event_name: event_name.to_string(),
        payload,
        metadata,
    };

    self.message_client
        .send_message(&queue_name, &message)
        .await
}
}

Fast Mode (In-Process) - Internal Subscriber Pattern

Fast events are the only delivery mode with internal subscriber support. Events are dispatched immediately to in-memory subscribers within the Tasker worker process.

#![allow(unused)]
fn main() {
// InProcessEventBus provides dual-path delivery
pub struct InProcessEventBus {
    event_sender: tokio::sync::broadcast::Sender<DomainEvent>,
    ffi_event_sender: Option<mpsc::Sender<DomainEvent>>,
}
}

Characteristics:

• Zero persistence overhead
• Sub-millisecond latency
• Lost on service restart
• Internal to Tasker process only
• Dual-path: Rust subscribers + Ruby FFI channel
• Non-blocking broadcast semantics

Dual-Path Architecture:

InProcessEventBus
       │
       ├──► tokio::broadcast::Sender ──► Rust Subscribers (EventRegistry)
       │
       └──► mpsc::Sender ──► Ruby FFI Channel ──► Ruby Event Handlers

Use Cases:

• Real-time metrics collection
• Internal logging and telemetry
• Secondary actions that are not business-critical parts of the Task -> WorkflowStep DAG hierarchy
• Example: DataDog, Sentry, NewRelic, PagerDuty, Salesforce, Slack, Zapier

Broadcast Mode - Internal + External Delivery

Broadcast mode delivers events to both paths simultaneously: the fast in-process bus for internal subscribers AND the durable PGMQ queue for external systems. This ensures internal subscribers receive the same event shape as external consumers.

#![allow(unused)]
fn main() {
// EventRouter handles broadcast semantics
async fn route_event(&self, event: DomainEvent, mode: EventDeliveryMode) {
    match mode {
        EventDeliveryMode::Durable => {
            self.durable_publisher.publish(event).await;
        }
        EventDeliveryMode::Fast => {
            self.in_process_bus.publish(event).await;
        }
        EventDeliveryMode::Broadcast => {
            // Send to both paths concurrently
            let (durable, fast) = tokio::join!(
                self.durable_publisher.publish(event.clone()),
                self.in_process_bus.publish(event)
            );
            // Log errors but don't fail
        }
    }
}
}

When to Use Broadcast:

• Internal subscribers need the same event that external systems receive
• Real-time internal metrics tracking for events also exported externally
• Audit logging both internally and to external compliance systems

Important: Data published via broadcast goes to BOTH the internal process AND the public PGMQ boundary. Do not use broadcast for sensitive internal-only data (use fast for those).

Publisher Patterns

Default Publisher (GenericStepEventPublisher)

The default publisher automatically handles event publication for step handlers:

#![allow(unused)]
fn main() {
pub struct GenericStepEventPublisher {
    router: Arc<EventRouter>,
    default_delivery_mode: EventDeliveryMode,
}

impl GenericStepEventPublisher {
    /// Publish event with metadata extracted from step context
    pub async fn publish(
        &self,
        step_data: &TaskSequenceStep,
        event_name: &str,
        payload: Value,
    ) -> TaskerResult<()> {
        let metadata = EventMetadata {
            task_uuid: step_data.task.task.task_uuid,
            step_uuid: Some(step_data.workflow_step.workflow_step_uuid),
            step_name: Some(step_data.workflow_step.name.clone()),
            namespace: step_data.task.namespace_name.clone(),
            correlation_id: step_data.task.task.correlation_id,
            fired_at: Utc::now(),
            fired_by: "generic_publisher".to_string(),
        };

        self.router.route_event(event_name, payload, metadata).await
    }
}
}

Custom Publishers

Custom publishers extend TaskerCore::DomainEvents::BasePublisher (Ruby) to provide specialized event handling with payload transformation, conditional publishing, and lifecycle hooks.

Real Example: PaymentEventPublisher (workers/ruby/spec/handlers/examples/domain_events/publishers/payment_event_publisher.rb):

# Custom publisher for payment-related domain events
# Demonstrates durable delivery mode with custom payload enrichment
module DomainEvents
  module Publishers
    class PaymentEventPublisher < TaskerCore::DomainEvents::BasePublisher
      # Must match the `publisher:` field in YAML
      def name
        'DomainEvents::Publishers::PaymentEventPublisher'
      end

      # Transform step result into payment event payload
      def transform_payload(step_result, event_declaration, step_context = nil)
        result = step_result[:result] || {}
        event_name = event_declaration[:name]

        if step_result[:success] && event_name&.include?('processed')
          build_success_payload(result, step_result, step_context)
        elsif !step_result[:success] && event_name&.include?('failed')
          build_failure_payload(result, step_result, step_context)
        else
          result
        end
      end

      # Determine if this event should be published
      def should_publish?(step_result, event_declaration, step_context = nil)
        result = step_result[:result] || {}
        event_name = event_declaration[:name]

        # For success events, verify we have transaction data
        if event_name&.include?('processed')
          return step_result[:success] && result[:transaction_id].present?
        end

        # For failure events, verify we have error info
        if event_name&.include?('failed')
          metadata = step_result[:metadata] || {}
          return !step_result[:success] && metadata[:error_code].present?
        end

        true  # Default: always publish
      end

      # Add execution metrics to event metadata
      def additional_metadata(step_result, event_declaration, step_context = nil)
        metadata = step_result[:metadata] || {}
        {
          execution_time_ms: metadata[:execution_time_ms],
          publisher_type: 'custom',
          publisher_name: name,
          payment_provider: metadata[:payment_provider]
        }
      end

      private

      def build_success_payload(result, step_result, step_context)
        {
          transaction_id: result[:transaction_id],
          amount: result[:amount],
          currency: result[:currency] || 'USD',
          payment_method: result[:payment_method] || 'credit_card',
          processed_at: result[:processed_at] || Time.now.iso8601,
          delivery_mode: 'durable',
          publisher: name
        }
      end
    end
  end
end

YAML Configuration for Custom Publisher:

steps:
  - name: process_payment
    publishes_events:
      - name: payment.processed
        condition: success
        delivery_mode: durable
        publisher: DomainEvents::Publishers::PaymentEventPublisher
      - name: payment.failed
        condition: failure
        delivery_mode: durable
        publisher: DomainEvents::Publishers::PaymentEventPublisher

YAML Event Declaration

Events are declared in task template YAML files using the publishes_events field at the step level:

# config/tasks/payments/credit_card_payment/1.0.0.yaml
name: credit_card_payment
namespace_name: payments
version: 1.0.0
description: Process credit card payments with validation and fraud detection

# Task-level domain events (optional)
domain_events: []

steps:
  - name: process_payment
    description: Process the payment transaction
    handler:
      callable: PaymentProcessing::StepHandler::ProcessPaymentHandler
      initialization:
        gateway_url: "${PAYMENT_GATEWAY_URL}"
    dependencies:
      - validate_payment
    retry:
      retryable: true
      limit: 3
      backoff: exponential
    timeout_seconds: 120

    # Step-level event declarations
    publishes_events:
      - name: payment.processed
        description: "Payment successfully processed"
        condition: success  # success, failure, retryable_failure, permanent_failure, always
        schema:
          type: object
          required: [transaction_id, amount]
          properties:
            transaction_id: { type: string }
            amount: { type: number }
        delivery_mode: broadcast  # durable, fast, or broadcast
        publisher: PaymentEventPublisher  # optional custom publisher

      - name: payment.failed
        description: "Payment processing failed permanently"
        condition: permanent_failure
        schema:
          type: object
          required: [error_code, reason]
          properties:
            error_code: { type: string }
            reason: { type: string }
        delivery_mode: durable

Publication Conditions:

• success: Publish only when step completes successfully
• failure: Publish on any step failure (backward compatible)
• retryable_failure: Publish only on retryable failures (step can be retried)
• permanent_failure: Publish only on permanent failures (exhausted retries or non-retryable)
• always: Publish regardless of step outcome

Event Declaration Fields:

• name: Event name in dotted notation (e.g., payment.processed)
• description: Human-readable description of when this event is published
• condition: When to publish (defaults to success)
• schema: JSON Schema for validating event payloads
• delivery_mode: Delivery mode (defaults to durable)
• publisher: Optional custom publisher class name

Subscriber Patterns

Subscriber patterns apply only to fast (in-process) events. Durable events are consumed by external systems, not by internal Tasker subscribers.

Rust Subscribers (InProcessEventBus)

Rust subscribers are registered with the InProcessEventBus using the EventHandler type. Subscribers are async closures that receive DomainEvent instances.

Real Example: Logging Subscriber (workers/rust/src/event_subscribers/logging_subscriber.rs):

#![allow(unused)]
fn main() {
use std::sync::Arc;
use tasker_shared::events::registry::EventHandler;
use tracing::info;

/// Create a logging subscriber that logs all events matching a pattern
pub fn create_logging_subscriber(prefix: &str) -> EventHandler {
    let prefix = prefix.to_string();

    Arc::new(move |event| {
        let prefix = prefix.clone();

        Box::pin(async move {
            let step_name = event.metadata.step_name.as_deref().unwrap_or("unknown");

            info!(
                prefix = %prefix,
                event_name = %event.event_name,
                event_id = %event.event_id,
                task_uuid = %event.metadata.task_uuid,
                step_name = %step_name,
                namespace = %event.metadata.namespace,
                correlation_id = %event.metadata.correlation_id,
                fired_at = %event.metadata.fired_at,
                "Domain event received"
            );

            Ok(())
        })
    })
}
}

Real Example: Metrics Collector (workers/rust/src/event_subscribers/metrics_subscriber.rs):

#![allow(unused)]
fn main() {
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

/// Collects metrics from domain events (thread-safe)
pub struct EventMetricsCollector {
    events_received: AtomicU64,
    success_events: AtomicU64,
    failure_events: AtomicU64,
    // ... additional fields
}

impl EventMetricsCollector {
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            events_received: AtomicU64::new(0),
            success_events: AtomicU64::new(0),
            failure_events: AtomicU64::new(0),
        })
    }

    /// Create an event handler for this collector
    pub fn create_handler(self: &Arc<Self>) -> EventHandler {
        let metrics = Arc::clone(self);

        Arc::new(move |event| {
            let metrics = Arc::clone(&metrics);

            Box::pin(async move {
                metrics.events_received.fetch_add(1, Ordering::Relaxed);

                if event.payload.execution_result.success {
                    metrics.success_events.fetch_add(1, Ordering::Relaxed);
                } else {
                    metrics.failure_events.fetch_add(1, Ordering::Relaxed);
                }

                Ok(())
            })
        })
    }

    pub fn events_received(&self) -> u64 {
        self.events_received.load(Ordering::Relaxed)
    }
}
}

Registration with InProcessEventBus:

#![allow(unused)]
fn main() {
use tasker_worker::worker::in_process_event_bus::InProcessEventBus;

let mut bus = InProcessEventBus::new(config);

// Subscribe to all events
bus.subscribe("*", create_logging_subscriber("[ALL]")).unwrap();

// Subscribe to specific patterns
bus.subscribe("payment.*", create_logging_subscriber("[PAYMENT]")).unwrap();

// Use metrics collector
let metrics = EventMetricsCollector::new();
bus.subscribe("*", metrics.create_handler()).unwrap();
}

Ruby Subscribers (BaseSubscriber)

Ruby subscribers extend TaskerCore::DomainEvents::BaseSubscriber and use the class-level subscribes_to pattern declaration.

Real Example: LoggingSubscriber (workers/ruby/spec/handlers/examples/domain_events/subscribers/logging_subscriber.rb):

# Example logging subscriber for fast/in-process domain events
module DomainEvents
  module Subscribers
    class LoggingSubscriber < TaskerCore::DomainEvents::BaseSubscriber
      # Subscribe to all events using pattern matching
      subscribes_to '*'

      # Handle any domain event by logging its details
      def handle(event)
        event_name = event[:event_name]
        metadata = event[:metadata] || {}

        logger.info "[LoggingSubscriber] Event: #{event_name}"
        logger.info "  Task: #{metadata[:task_uuid]}"
        logger.info "  Step: #{metadata[:step_name]}"
        logger.info "  Namespace: #{metadata[:namespace]}"
        logger.info "  Correlation: #{metadata[:correlation_id]}"
      end
    end
  end
end

Real Example: MetricsSubscriber (workers/ruby/spec/handlers/examples/domain_events/subscribers/metrics_subscriber.rb):

# Example metrics subscriber for fast/in-process domain events
module DomainEvents
  module Subscribers
    class MetricsSubscriber < TaskerCore::DomainEvents::BaseSubscriber
      subscribes_to '*'

      class << self
        attr_accessor :events_received, :success_events, :failure_events,
                      :events_by_namespace, :last_event_at

        def reset_counters!
          @mutex = Mutex.new
          @events_received = 0
          @success_events = 0
          @failure_events = 0
          @events_by_namespace = Hash.new(0)
          @last_event_at = nil
        end
      end

      reset_counters!

      def handle(event)
        event_name = event[:event_name]
        metadata = event[:metadata] || {}
        execution_result = event[:execution_result] || {}

        self.class.increment(:events_received)

        if execution_result[:success]
          self.class.increment(:success_events)
        else
          self.class.increment(:failure_events)
        end

        namespace = metadata[:namespace] || 'unknown'
        self.class.increment_hash(:events_by_namespace, namespace)
        self.class.set(:last_event_at, Time.now)
      end
    end
  end
end

Registration in Bootstrap:

# Register subscribers with the registry
registry = TaskerCore::DomainEvents::SubscriberRegistry.instance
registry.register(DomainEvents::Subscribers::LoggingSubscriber)
registry.register(DomainEvents::Subscribers::MetricsSubscriber)
registry.start_all!

# Later, query metrics
puts "Total events: #{DomainEvents::Subscribers::MetricsSubscriber.events_received}"
puts "By namespace: #{DomainEvents::Subscribers::MetricsSubscriber.events_by_namespace}"

External PGMQ Consumers (Durable Events)

Durable events are published to PGMQ queues for external consumption. Tasker does not provide internal consumers for these queues. External systems can consume events using:

1. Direct PGMQ Polling: Query pgmq.q_{namespace}_domain_events tables directly
2. PGMQ Client Libraries: Use pgmq client libraries in Python, Node.js, Go, etc.
3. Middleware Proxies: Build adapters that forward events to Kafka, SNS/SQS, etc.

Example: External Python Consumer:

import pgmq

# Connect to PGMQ
queue = pgmq.Queue("payments_domain_events", dsn="postgresql://...")

# Poll for events
while True:
    messages = queue.read(batch_size=50, vt=30)
    for msg in messages:
        process_event(msg.message)
        queue.delete(msg.msg_id)

Configuration

Domain event system configuration is part of the worker configuration in worker.toml files.

TOML Configuration

# config/tasker/base/worker.toml

# In-process event bus configuration for fast domain event delivery
[worker.mpsc_channels.in_process_events]
broadcast_buffer_size = 2000        # Channel capacity for broadcast events
log_subscriber_errors = true        # Log errors from event subscribers
dispatch_timeout_ms = 5000          # Timeout for event dispatch

# Domain Event System MPSC Configuration
[worker.mpsc_channels.domain_events]
command_buffer_size = 1000          # Channel capacity for domain event commands
shutdown_drain_timeout_ms = 5000    # Time to drain events on shutdown
log_dropped_events = true           # Log when events are dropped due to backpressure

# In-process event settings (part of worker event systems)
[worker.event_systems.worker.metadata.in_process_events]
ffi_integration_enabled = true      # Enable Ruby/Python FFI event channel
deduplication_cache_size = 10000    # Event deduplication cache size

Environment Overrides

Test Environment (config/tasker/environments/test/worker.toml):

# In-process event bus - smaller buffers for testing
[worker.mpsc_channels.in_process_events]
broadcast_buffer_size = 1000
log_subscriber_errors = true
dispatch_timeout_ms = 2000

# Domain Event System - smaller buffers for testing
[worker.mpsc_channels.domain_events]
command_buffer_size = 100
shutdown_drain_timeout_ms = 1000
log_dropped_events = true

[worker.event_systems.worker.metadata.in_process_events]
deduplication_cache_size = 1000

Production Environment (config/tasker/environments/production/worker.toml):

# In-process event bus - large buffers for production throughput
[worker.mpsc_channels.in_process_events]
broadcast_buffer_size = 5000
log_subscriber_errors = false       # Reduce log noise in production
dispatch_timeout_ms = 10000

# Domain Event System - large buffers for production throughput
[worker.mpsc_channels.domain_events]
command_buffer_size = 5000
shutdown_drain_timeout_ms = 10000
log_dropped_events = false          # Reduce log noise in production

Configuration Parameters

Integration with Step Execution

Event-Driven Domain Event Publishing

The worker uses an event-driven command pattern for step execution and domain event publishing. Nothing blocks - domain events are dispatched after successful orchestration notification using fire-and-forget semantics.

Flow (tasker-worker/src/worker/command_processor.rs):

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────────┐
│  FFI Handler    │────►│ Completion       │────►│ WorkerProcessor     │
│  (Ruby/Rust)    │     │ Channel          │     │ Command Loop        │
└─────────────────┘     └──────────────────┘     └──────────┬──────────┘
                                                            │
                        ┌───────────────────────────────────┴───────────────┐
                        │                                                   │
                        ▼                                                   ▼
              ┌─────────────────────┐                          ┌────────────────────┐
              │ 1. Send result to   │                          │ 2. Dispatch domain │
              │    orchestration    │──── on success ─────────►│    events          │
              │    (PGMQ)           │                          │    (fire-and-forget)│
              └─────────────────────┘                          └────────────────────┘

Implementation:

#![allow(unused)]
fn main() {
// tasker-worker/src/worker/command_processor.rs (lines 512-525)
// Worker command processor receives step completions via FFI channel
match self.handle_send_step_result(step_result.clone()).await {
    Ok(()) => {
        // Dispatch domain events AFTER successful orchestration notification.
        // Domain events are declarative of what HAS happened - the step is only
        // truly complete once orchestration has been notified successfully.
        // Fire-and-forget semantics (try_send) - never blocks the worker.
        self.dispatch_domain_events(&step_result, None);
        info!(
            worker_id = %self.worker_id,
            step_uuid = %step_result.step_uuid,
            "Step completion forwarded to orchestration successfully"
        );
    }
    Err(e) => {
        // Don't dispatch domain events - orchestration wasn't notified,
        // so the step isn't truly complete from the system's perspective
        error!(
            worker_id = %self.worker_id,
            step_uuid = %step_result.step_uuid,
            error = %e,
            "Failed to forward step completion to orchestration"
        );
    }
}
}

Domain Event Dispatch (fire-and-forget):

#![allow(unused)]
fn main() {
// tasker-worker/src/worker/command_processor.rs (lines 362-432)
fn dispatch_domain_events(&mut self, step_result: &StepExecutionResult, correlation_id: Option<Uuid>) {
    // Retrieve cached step context (stored when step was claimed)
    let task_sequence_step = match self.step_execution_contexts.remove(&step_result.step_uuid) {
        Some(ctx) => ctx,
        None => return, // No context = can't build events
    };

    // Build events from step definition's publishes_events declarations
    for event_def in &task_sequence_step.step_definition.publishes_events {
        // Check publication condition before building event
        if !event_def.should_publish(step_result.success) {
            continue; // Skip events whose condition doesn't match
        }

        let event = DomainEventToPublish {
            event_name: event_def.name.clone(),
            delivery_mode: event_def.delivery_mode,
            business_payload: step_result.result.clone(),
            metadata: EventMetadata { /* ... */ },
            task_sequence_step: task_sequence_step.clone(),
            execution_result: step_result.clone(),
        };
        domain_events.push(event);
    }

    // Fire-and-forget dispatch - try_send never blocks
    let dispatched = handle.dispatch_events(domain_events, publisher_name, correlation);

    if !dispatched {
        warn!(
            step_uuid = %step_result.step_uuid,
            "Domain event dispatch failed - channel full (events dropped)"
        );
    }
}
}

Key Design Decisions:

• Events only after orchestration success: Domain events are declarative of what HAS happened. If orchestration notification fails, the step isn’t truly complete from the system’s perspective.
• Fire-and-forget via try_send: Never blocks the worker command loop. If the channel is full, events are dropped and logged.
• Context caching: Step execution context is cached when the step is claimed, then retrieved for event building after completion.

Correlation ID Propagation

Domain events maintain correlation IDs for end-to-end distributed tracing. The correlation ID originates from the task and flows through all step executions and domain events.

EventMetadata Structure (tasker-shared/src/events/domain_events.rs):

#![allow(unused)]
fn main() {
pub struct EventMetadata {
    pub task_uuid: Uuid,
    pub step_uuid: Option<Uuid>,
    pub step_name: Option<String>,
    pub namespace: String,
    pub correlation_id: Uuid,      // From task for end-to-end tracing
    pub fired_at: DateTime<Utc>,
    pub fired_by: String,          // Publisher identifier (worker_id)
}
}

Getting Correlation ID via API:

Use the orchestration API to get the correlation ID for a task:

# Get task details including correlation_id
curl http://localhost:8080/v1/tasks/{task_uuid} | jq '.correlation_id'

# Response includes correlation_id
{
  "task_uuid": "0199c3e0-ccdb-7581-87ab-3f67daeaa4a5",
  "correlation_id": "0199c3e0-ccdb-7581-87ab-3f67daeaa4a5",
  "status": "complete",
  ...
}

Tracing Events in PGMQ:

# Find all durable events for a correlation ID
psql $DATABASE_URL -c "
  SELECT
    message->>'event_name' as event,
    message->'metadata'->>'step_name' as step,
    message->'metadata'->>'fired_at' as fired_at
  FROM pgmq.q_payments_domain_events
  WHERE message->'metadata'->>'correlation_id' = '0199c3e0-ccdb-7581-87ab-3f67daeaa4a5'
  ORDER BY message->'metadata'->>'fired_at';
"

Metrics and Observability

OpenTelemetry Metrics

Domain event publication emits OpenTelemetry counter metrics (tasker-shared/src/events/domain_events.rs:207-219):

#![allow(unused)]
fn main() {
// Metric emitted on every domain event publication
let counter = opentelemetry::global::meter("tasker")
    .u64_counter("tasker.domain_events.published.total")
    .with_description("Total number of domain events published")
    .build();

counter.add(1, &[
    opentelemetry::KeyValue::new("event_name", event_name.to_string()),
    opentelemetry::KeyValue::new("namespace", metadata.namespace.clone()),
]);
}

Prometheus Metrics Endpoint

The orchestration service exposes Prometheus-format metrics:

# Get Prometheus metrics from orchestration service
curl http://localhost:8080/metrics

# Get Prometheus metrics from worker service
curl http://localhost:8081/metrics

OpenTelemetry Tracing

Domain event publication is instrumented with tracing spans (tasker-shared/src/events/domain_events.rs:157-161):

#![allow(unused)]
fn main() {
#[instrument(skip(self, payload, metadata), fields(
    event_name = %event_name,
    namespace = %metadata.namespace,
    correlation_id = %metadata.correlation_id
))]
pub async fn publish_event(
    &self,
    event_name: &str,
    payload: DomainEventPayload,
    metadata: EventMetadata,
) -> Result<Uuid, DomainEventError> {
    // ... implementation with debug! and info! logs including correlation_id
}
}

Grafana Query Examples

Loki Query - Domain Events by Correlation ID:

{service_name="tasker-worker"} |= "Domain event published" | json | correlation_id = "0199c3e0-ccdb-7581-87ab-3f67daeaa4a5"

Loki Query - All Domain Event Publications:

{service_name=~"tasker.*"} |= "Domain event" | json | line_format "{{.event_name}} - {{.namespace}} - {{.correlation_id}}"

Tempo Query - Trace by Correlation ID:

{resource.service.name="tasker-worker"} && {span.correlation_id="0199c3e0-ccdb-7581-87ab-3f67daeaa4a5"}

Prometheus Query - Event Publication Rate by Namespace:

sum by (namespace) (rate(tasker_domain_events_published_total[5m]))

Prometheus Query - Event Publication Rate by Event Name:

topk(10, sum by (event_name) (rate(tasker_domain_events_published_total[5m])))

Structured Log Fields

Domain event logs include structured fields for querying:

Best Practices

1. Choose the Right Delivery Mode

Key Decision Criteria:

• Need internal Tasker subscribers? → Use fast or broadcast
• Need external system integration? → Use durable or broadcast
• Internal-only, sensitive data? → Use fast (never reaches PGMQ boundary)

2. Design Event Payloads

Do:

#![allow(unused)]
fn main() {
json!({
    "transaction_id": "TXN-123",
    "amount": 99.99,
    "currency": "USD",
    "timestamp": "2025-12-01T10:00:00Z",
    "idempotency_key": step_uuid
})
}

Don’t:

#![allow(unused)]
fn main() {
json!({
    "data": "payment processed",  // No structure
    "info": full_database_record  // Too much data
})
}

3. Handle Subscriber Failures Gracefully

#![allow(unused)]
fn main() {
#[async_trait]
impl EventSubscriber for MySubscriber {
    async fn handle(&self, event: &DomainEvent) -> TaskerResult<()> {
        // Wrap in timeout
        match timeout(Duration::from_secs(5), self.process(event)).await {
            Ok(result) => result,
            Err(_) => {
                warn!(event = %event.name, "Subscriber timeout");
                Ok(()) // Don't fail the dispatch
            }
        }
    }
}
}

4. Use Correlation IDs for Debugging

#![allow(unused)]
fn main() {
// Always include correlation ID in logs
info!(
    correlation_id = %event.metadata.correlation_id,
    event_name = %event.name,
    namespace = %event.metadata.namespace,
    "Processing domain event"
);
}

Related Documentation

• Events and Commands: events-and-commands.md - System event architecture
• Observability: observability/README.md - Metrics and monitoring
• States and Lifecycles: states-and-lifecycles.md - Task/step state machines

This domain event architecture provides a flexible, reliable foundation for business observability in the tasker-core workflow orchestration system.

Events and Commands Architecture

Last Updated: 2026-01-15 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | Actor-Based Architecture | Messaging Abstraction | States and Lifecycles | Deployment Patterns

← Back to Documentation Hub

This document provides comprehensive documentation of the event-driven and command pattern architecture in tasker-core, covering the unified event system foundation, orchestration and worker implementations, and the flow of tasks and steps through the system.

Overview

The tasker-core system implements a sophisticated hybrid architecture that combines:

1. Event-Driven Systems: Real-time coordination using PostgreSQL LISTEN/NOTIFY and PGMQ notifications
2. Command Pattern: Async command processors using tokio mpsc channels for orchestration and worker operations
3. Hybrid Deployment Modes: PollingOnly, EventDrivenOnly, and Hybrid modes with fallback polling
4. Queue-Based Communication: Provider-agnostic message queues (PGMQ or RabbitMQ) for reliable step execution and result processing

This architecture eliminates polling complexity while maintaining resilience through fallback mechanisms and provides horizontal scaling capabilities with atomic operation guarantees.

Event System Foundation

EventDrivenSystem Trait

The foundation of the event architecture is defined in tasker-shared/src/event_system/event_driven.rs with the EventDrivenSystem trait:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait EventDrivenSystem: Send + Sync {
    type SystemId: Send + Sync + Clone + fmt::Display + fmt::Debug;
    type Event: Send + Sync + Clone + fmt::Debug;
    type Config: Send + Sync + Clone;
    type Statistics: EventSystemStatistics + Send + Sync + Clone;

    // Core lifecycle methods
    async fn start(&mut self) -> Result<(), DeploymentModeError>;
    async fn stop(&mut self) -> Result<(), DeploymentModeError>;
    fn is_running(&self) -> bool;

    // Event processing
    async fn process_event(&self, event: Self::Event) -> Result<(), DeploymentModeError>;

    // Monitoring and health
    async fn health_check(&self) -> Result<DeploymentModeHealthStatus, DeploymentModeError>;
    fn statistics(&self) -> Self::Statistics;

    // Configuration
    fn deployment_mode(&self) -> DeploymentMode;
    fn config(&self) -> &Self::Config;
}
}

Deployment Modes

The system supports three deployment modes for different operational requirements:

PollingOnly Mode

• Traditional polling-based coordination
• No event listeners or real-time notifications
• Reliable fallback for environments with networking restrictions
• Higher latency but guaranteed operation

EventDrivenOnly Mode

• Pure event-driven coordination using PostgreSQL LISTEN/NOTIFY
• Real-time response to database changes
• Lowest latency for step discovery and task coordination
• Requires reliable PostgreSQL connections

Hybrid Mode

• Primary event-driven coordination with polling fallback
• Best of both worlds: real-time when possible, reliable when needed
• Automatic fallback during connection issues
• Production-ready with resilience guarantees

Selecting a Deployment Mode

The Tasker system is built with the expectation of distributed deployment with multiple instances of both orchestration core servers and worker servers operating simultaneously. The goal of separating deployment mode is to enable different deployments to scale up event driven only processing nodes to meet demand, while having polling only nodes at a reasonable fallback polling interval and batch size. It is also to deploy in hybrid mode and control these on an instance over instance level.

Event Types and Sources

Queue-Level Events (Provider-Agnostic)

The system supports multiple messaging backends through MessageNotification:

#![allow(unused)]
fn main() {
pub enum MessageNotification {
    /// Signal-only notification (PGMQ style)
    /// Indicates a message is available but requires separate fetch
    Available {
        queue_name: String,
        msg_id: Option<i64>,
    },

    /// Full message notification (RabbitMQ style)
    /// Contains the complete message payload
    Message(QueuedMessage<Vec<u8>>),
}
}

Event Sources by Provider:

Common Event Types:

• Step Results: Worker completion notifications
• Task Requests: New task initialization requests
• Message Ready Events: Queue message availability notifications
• Transport: Provider-agnostic via MessagingProvider.subscribe_many()

Command Pattern Architecture

Command Processor Pattern

Both orchestration and worker systems implement the command pattern to replace complex polling-based coordinators:

Benefits:

• No Polling Loops (Except where intended for fallback): Pure tokio mpsc command processing
• Simplified Architecture: ~100 lines vs 1000+ lines of complex systems
• Race Condition Prevention: Atomic operations through proper delegation
• Observability Preservation: Maintains metrics through delegated components

Command Flow Patterns

Both systems follow consistent command processing patterns:

sequenceDiagram
    participant Client
    participant CommandChannel
    participant Processor
    participant Delegate
    participant Response

    Client->>CommandChannel: Send Command + ResponseChannel
    CommandChannel->>Processor: Receive Command
    Processor->>Delegate: Delegate to Business Logic Component
    Delegate-->>Processor: Return Result
    Processor->>Response: Send Result via ResponseChannel
    Response-->>Client: Receive Result

Orchestration Event Systems

OrchestrationEventSystem

Implemented in tasker-orchestration/src/orchestration/event_systems/orchestration_event_system.rs:

#![allow(unused)]
fn main() {
pub struct OrchestrationEventSystem {
    system_id: String,
    deployment_mode: DeploymentMode,
    queue_listener: Option<OrchestrationQueueListener>,
    fallback_poller: Option<OrchestrationFallbackPoller>,
    context: Arc<SystemContext>,
    orchestration_core: Arc<OrchestrationCore>,
    command_sender: mpsc::Sender<OrchestrationCommand>,
    // ... statistics and state
}
}

Orchestration Command Types

The command processor handles both full-message and signal-only notification types:

#![allow(unused)]
fn main() {
pub enum OrchestrationCommand {
    // Task lifecycle
    InitializeTask { request: TaskRequestMessage, resp: CommandResponder<TaskInitializeResult> },
    ProcessStepResult { result: StepExecutionResult, resp: CommandResponder<StepProcessResult> },
    FinalizeTask { task_uuid: Uuid, resp: CommandResponder<TaskFinalizationResult> },

    // Full message processing (RabbitMQ style - MessageNotification::Message)
    // Used when provider delivers complete message payload
    ProcessStepResultFromMessage { queue_name: String, message: QueuedMessage, resp: CommandResponder<StepProcessResult> },
    InitializeTaskFromMessage { queue_name: String, message: QueuedMessage, resp: CommandResponder<TaskInitializeResult> },
    FinalizeTaskFromMessage { queue_name: String, message: QueuedMessage, resp: CommandResponder<TaskFinalizationResult> },

    // Signal-only processing (PGMQ style - MessageNotification::Available)
    // Used when provider sends notification that requires separate fetch
    ProcessStepResultFromMessageEvent { message_event: MessageReadyEvent, resp: CommandResponder<StepProcessResult> },
    InitializeTaskFromMessageEvent { message_event: MessageReadyEvent, resp: CommandResponder<TaskInitializeResult> },
    FinalizeTaskFromMessageEvent { message_event: MessageReadyEvent, resp: CommandResponder<TaskFinalizationResult> },

    // Task readiness (database events)
    ProcessTaskReadiness { task_uuid: Uuid, namespace: String, priority: i32, ready_steps: i32, triggered_by: String, resp: CommandResponder<TaskReadinessResult> },

    // System operations
    GetProcessingStats { resp: CommandResponder<OrchestrationProcessingStats> },
    HealthCheck { resp: CommandResponder<SystemHealth> },
    Shutdown { resp: CommandResponder<()> },
}
}

Command Routing by Notification Type:

• MessageNotification::Message -> *FromMessage commands (immediate processing)
• MessageNotification::Available -> *FromMessageEvent commands (requires fetch)

Orchestration Queue Architecture

The orchestration system coordinates multiple queue types:

1. orchestration_step_results: Step completion results from workers
2. orchestration_task_requests: New task initialization requests
3. orchestration_task_finalization: Task finalization notifications
4. Namespace Queues: Per-namespace step queues (e.g., fulfillment_queue, inventory_queue)

TaskReadinessEventSystem

Handles database-level events for task readiness using PostgreSQL LISTEN/NOTIFY:

#![allow(unused)]
fn main() {
pub struct TaskReadinessEventSystem {
    system_id: String,
    deployment_mode: DeploymentMode,
    listener: Option<TaskReadinessListener>,
    fallback_poller: Option<TaskReadinessFallbackPoller>,
    context: Arc<SystemContext>,
    command_sender: mpsc::Sender<OrchestrationCommand>,
    // ... configuration and statistics
}
}

PGMQ Notification Channels:

• pgmq_message_ready.orchestration: Orchestration queue messages ready (task requests, step results, finalizations)
• pgmq_message_ready.{namespace}: Worker namespace queue messages ready (e.g., payments, fulfillment, linear_workflow)
• pgmq_message_ready: Global channel for all queue messages (fallback)
• pgmq_queue_created: Queue creation notifications

Unified Event Coordination

The UnifiedEventCoordinator demonstrates coordinated management of multiple event systems:

#![allow(unused)]
fn main() {
pub struct UnifiedEventCoordinator {
    orchestration_system: OrchestrationEventSystem,
    task_readiness_fallback: FallbackPoller,
    deployment_mode: DeploymentMode,
    health_monitor: EventSystemHealthMonitor,
    // ... coordination logic
}
}

Coordination Features:

• Shared Command Channel: Both systems send commands to same orchestration processor
• Health Monitoring: Unified health checking across all event systems
• Deployment Mode Management: Synchronized mode changes
• Statistics Aggregation: Combined metrics from all systems

Worker Event Systems

WorkerEventSystem

Implemented in tasker-worker/src/worker/event_systems/worker_event_system.rs:

#![allow(unused)]
fn main() {
pub struct WorkerEventSystem {
    system_id: String,
    deployment_mode: DeploymentMode,
    queue_listeners: HashMap<String, WorkerQueueListener>,
    fallback_pollers: HashMap<String, WorkerFallbackPoller>,
    context: Arc<SystemContext>,
    command_sender: mpsc::Sender<WorkerCommand>,
    // ... statistics and configuration
}
}

Worker Command Types

#![allow(unused)]
fn main() {
pub enum WorkerCommand {
    // Step execution
    ExecuteStep { message: PgmqMessage<SimpleStepMessage>, queue_name: String, resp: CommandResponder<()> },
    ExecuteStepWithCorrelation { message: PgmqMessage<SimpleStepMessage>, queue_name: String, correlation_id: Uuid, resp: CommandResponder<()> },

    // Result processing
    SendStepResult { result: StepExecutionResult, resp: CommandResponder<()> },
    ProcessStepCompletion { step_result: StepExecutionResult, correlation_id: Option<Uuid>, resp: CommandResponder<()> },

    // Event integration
    ExecuteStepFromMessage { queue_name: String, message: PgmqMessage, resp: CommandResponder<()> },
    ExecuteStepFromEvent { message_event: MessageReadyEvent, resp: CommandResponder<()> },

    // System operations
    GetWorkerStatus { resp: CommandResponder<WorkerStatus> },
    SetEventIntegration { enabled: bool, resp: CommandResponder<()> },
    GetEventStatus { resp: CommandResponder<EventIntegrationStatus> },
    RefreshTemplateCache { namespace: Option<String>, resp: CommandResponder<()> },
    HealthCheck { resp: CommandResponder<WorkerHealthStatus> },
    Shutdown { resp: CommandResponder<()> },
}
}

Worker Queue Architecture

Workers monitor namespace-specific queues for step execution as Custom Namespace Queues that are dynamically configured per deployment

Example queues:

1. fulfillment_queue: All fulfillment namespace steps
2. inventory_queue: All inventory namespace steps
3. notifications_queue: All notification namespace steps
4. payment_queue: All payment processing steps

Event Flow and System Interactions

Complete Task Execution Flow

sequenceDiagram
    participant Client
    participant Orchestration
    participant TaskDB
    participant StepQueue
    participant Worker
    participant ResultQueue

    %% Task Initialization
    Client->>Orchestration: TaskRequestMessage (via pgmq_send_with_notify)
    Orchestration->>TaskDB: Create Task + Steps

    %% Step Discovery and Enqueueing (Event-Driven or Fallback Polling)
    Orchestration->>StepQueue: pgmq_send_with_notify(ready steps)
    StepQueue-->>Worker: pg_notify('pgmq_message_ready.{namespace}')

    %% Step Execution
    Worker->>StepQueue: pgmq.read() to claim step
    Worker->>Worker: Execute Step Handler
    Worker->>ResultQueue: pgmq_send_with_notify(StepExecutionResult)
    ResultQueue-->>Orchestration: pg_notify('pgmq_message_ready.orchestration')

    %% Result Processing
    Orchestration->>Orchestration: ProcessStepResult Command
    Orchestration->>TaskDB: Update Step State
    Note over Orchestration: Fallback poller discovers ready steps if events missed

    %% Task Completion
    Note over Orchestration: All Steps Complete
    Orchestration->>Orchestration: FinalizeTask Command
    Orchestration->>TaskDB: Mark Task Complete
    Orchestration-->>Client: Task Completed

Event-Driven Step Discovery

sequenceDiagram
    participant Worker
    participant PostgreSQL
    participant PgmqNotify
    participant OrchestrationListener
    participant StepEnqueuer

    Worker->>PostgreSQL: pgmq_send_with_notify('orchestration_step_results', result)
    PostgreSQL->>PostgreSQL: Atomic: pgmq.send() + pg_notify()
    PostgreSQL->>PgmqNotify: NOTIFY 'pgmq_message_ready.orchestration'
    PgmqNotify->>OrchestrationListener: MessageReadyEvent
    OrchestrationListener->>StepEnqueuer: ProcessStepResult Command
    StepEnqueuer->>PostgreSQL: Query ready steps, enqueue via pgmq_send_with_notify()

Hybrid Mode Operation

stateDiagram-v2
    [*] --> EventDriven

    EventDriven --> Processing : Event Received
    Processing --> EventDriven : Success
    Processing --> PollingFallback : Event Failed

    PollingFallback --> FallbackPolling : Start Polling
    FallbackPolling --> EventDriven : Connection Restored
    FallbackPolling --> Processing : Poll Found Work

    EventDriven --> HealthCheck : Periodic Check
    HealthCheck --> EventDriven : Healthy
    HealthCheck --> PollingFallback : Event Issues Detected

Queue Architecture and Message Flow

PGMQ Integration

The system uses PostgreSQL Message Queue (PGMQ) for reliable message delivery:

Queue Types and Purposes

Message Processing Patterns

Event-Driven Processing:

1. Message arrives in PGMQ queue
2. PostgreSQL triggers pg_notify with MessageReadyEvent
3. Event system receives notification
4. System processes message via command pattern
5. Message deleted after successful processing

Polling-Based Processing (Fallback):

1. Periodic queue polling (configurable interval)
2. Fetch available messages in batches
3. Process messages via command pattern
4. Delete processed messages

Circuit Breaker Integration

All PGMQ operations are protected by circuit breakers:

#![allow(unused)]
fn main() {
pub struct UnifiedPgmqClient {
    standard_client: Box<dyn PgmqClientTrait + Send + Sync>,
    protected_client: Option<ProtectedPgmqClient>,
    circuit_breaker_enabled: bool,
}
}

Circuit Breaker Features:

• Automatic Protection: Failure detection and circuit opening
• Configurable Thresholds: Error rate and timeout configuration
• Seamless Fallback: Automatic switching between standard and protected clients
• Recovery Detection: Automatic circuit closing when service recovers

Statistics and Monitoring

Event System Statistics

Both orchestration and worker event systems implement comprehensive statistics:

#![allow(unused)]
fn main() {
pub trait EventSystemStatistics {
    fn events_processed(&self) -> u64;
    fn events_failed(&self) -> u64;
    fn processing_rate(&self) -> f64;         // events/second
    fn average_latency_ms(&self) -> f64;
    fn deployment_mode_score(&self) -> f64;   // 0.0-1.0 effectiveness
    fn success_rate(&self) -> f64;            // derived: processed/(processed+failed)
}
}

Health Monitoring

Deployment Mode Health Status

#![allow(unused)]
fn main() {
pub enum DeploymentModeHealthStatus {
    Healthy,                    // All systems operational
    Degraded { reason: String },// Some issues but functional
    Unhealthy { reason: String },// Significant issues
    Critical { reason: String }, // System failure imminent
}
}

Health Check Integration

• Event System Health: Connection status, processing latency, error rates
• Command Processor Health: Queue backlog, processing timeout detection
• Database Health: Connection pool status, query performance
• Circuit Breaker Status: Circuit state, failure rates, recovery status

Metrics Collection

Key metrics collected across the system:

Orchestration Metrics

• Task Initialization Rate: Tasks/minute initialized
• Step Enqueueing Rate: Steps/minute enqueued to worker queues
• Result Processing Rate: Results/minute processed from workers
• Task Completion Rate: Tasks/minute completed successfully
• Error Rates: Failures by operation type and cause

Worker Metrics

• Step Execution Rate: Steps/minute executed
• Handler Performance: Execution time by handler type
• Queue Processing: Messages claimed/processed by queue
• Result Submission Rate: Results/minute sent to orchestration
• FFI Integration: Event correlation and handler communication stats

Error Handling and Resilience

Error Categories

The system handles multiple error categories with appropriate strategies:

Transient Errors

• Database Connection Issues: Circuit breaker protection + retry with exponential backoff
• Queue Processing Failures: Message retry with backoff, poison message detection
• Network Interruptions: Automatic fallback to polling mode

Permanent Errors

• Invalid Message Format: Dead letter queue for manual analysis
• Handler Execution Failures: Step failure state with retry limits
• Configuration Errors: System startup prevention with clear error messages

System Errors

• Resource Exhaustion: Graceful degradation and load shedding
• Component Crashes: Automatic restart with state recovery
• Data Corruption: Transaction rollback and consistency validation

Fallback Mechanisms

Event System Fallbacks

1. Event-Driven -> Polling: Automatic fallback when event connection fails
2. Real-time -> Batch: Switch to batch processing during high load
3. Primary -> Secondary: Database failover support for high availability

Command Processing Fallbacks

1. Async -> Sync: Degraded operation for critical operations
2. Distributed -> Local: Local processing when coordination fails
3. Optimistic -> Pessimistic: Conservative processing during uncertainty

Configuration Management

Event System Configuration

Event systems are configured via TOML with environment overrides:

# config/tasker/base/event_systems.toml
[orchestration_event_system]
system_id = "orchestration-events"
deployment_mode = "Hybrid"
health_monitoring_enabled = true
health_check_interval = "30s"
max_concurrent_processors = 10
processing_timeout = "100ms"

[orchestration_event_system.queue_listener]
enabled = true
batch_size = 50
poll_interval = "1s"
connection_timeout = "5s"

[orchestration_event_system.fallback_poller]
enabled = true
poll_interval = "5s"
batch_size = 20
max_retry_attempts = 3

[task_readiness]
enabled = true
polling_interval_seconds = 30

[orchestration_event_system]
system_id = "orchestration-events"
deployment_mode = "Hybrid"
# PGMQ channels handled by listeners, not direct postgres channels
supported_namespaces = ["orchestration"]

Runtime Configuration Changes

Certain configuration changes can be applied at runtime:

• Deployment Mode Switching: EventDrivenOnly <-> Hybrid <-> PollingOnly
• Event Integration Toggle: Enable/disable event processing
• Health Check Intervals: Adjust monitoring frequency
• Circuit Breaker Thresholds: Modify failure detection sensitivity

Integration Points

State Machine Integration

Event systems integrate tightly with the state machines documented in states-and-lifecycles.md:

1. Task State Changes: Event systems react to task transitions
2. Step State Changes: Step completion triggers task readiness checks
3. Event Generation: State transitions generate events for system coordination
4. Atomic Operations: Event processing maintains state machine consistency

Database Integration

Event systems coordinate with PostgreSQL at multiple levels:

1. LISTEN/NOTIFY: Real-time notifications for database changes
2. PGMQ Integration: Reliable message queues built on PostgreSQL
3. Transaction Coordination: Event processing within database transactions
4. SQL Functions: Database functions generate events and notifications

External System Integration

The event architecture supports integration with external systems:

1. Webhook Events: HTTP callbacks for external system notifications
2. Message Bus Integration: Apache Kafka, RabbitMQ, etc. for enterprise messaging
3. Monitoring Integration: Prometheus, DataDog, etc. for metrics export
4. API Integration: REST and GraphQL APIs for external coordination

Actor Integration

Overview

The tasker-core system implements a lightweight Actor pattern that formalizes the relationship between Commands and Lifecycle Components. This architecture provides a consistent, type-safe foundation for orchestration component management with all lifecycle operations coordinated through actors.

Status: Complete (Phases 1-7) - Production ready

For comprehensive actor documentation, see Actor-Based Architecture.

Actor Pattern Basics

The actor pattern introduces three core traits:

1. OrchestrationActor: Base trait for all actors with lifecycle hooks
2. Handler: Message handling trait for type-safe command processing
3. Message: Marker trait for command messages

#![allow(unused)]
fn main() {
// Actor definition
pub struct TaskFinalizerActor {
    context: Arc<SystemContext>,
    service: TaskFinalizer,
}

// Message definition
pub struct FinalizeTaskMessage {
    pub task_uuid: Uuid,
}

impl Message for FinalizeTaskMessage {
    type Response = FinalizationResult;
}

// Message handler
#[async_trait]
impl Handler<FinalizeTaskMessage> for TaskFinalizerActor {
    type Response = FinalizationResult;

    async fn handle(&self, msg: FinalizeTaskMessage) -> TaskerResult<Self::Response> {
        self.service.finalize_task(msg.task_uuid).await
            .map_err(|e| e.into())
    }
}
}

Integration with Command Processor

The actor pattern integrates seamlessly with the command processor through direct actor calls:

#![allow(unused)]
fn main() {
// From: tasker-orchestration/src/orchestration/command_processor.rs

async fn handle_finalize_task(&self, task_uuid: Uuid) -> TaskerResult<TaskFinalizationResult> {
    // Direct actor-based task finalization
    let msg = FinalizeTaskMessage { task_uuid };
    let result = self.actors.task_finalizer_actor.handle(msg).await?;

    Ok(TaskFinalizationResult::Success {
        task_uuid: result.task_uuid,
        final_status: format!("{:?}", result.action),
        completion_time: Some(chrono::Utc::now()),
    })
}

async fn handle_process_step_result(
    &self,
    step_result: StepExecutionResult,
) -> TaskerResult<StepProcessResult> {
    // Direct actor-based step result processing
    let msg = ProcessStepResultMessage {
        result: step_result.clone(),
    };

    match self.actors.result_processor_actor.handle(msg).await {
        Ok(()) => Ok(StepProcessResult::Success {
            message: format!(
                "Step {} result processed successfully",
                step_result.step_uuid
            ),
        }),
        Err(e) => Ok(StepProcessResult::Error {
            message: format!("Failed to process step result: {e}"),
        }),
    }
}
}

Event → Command → Actor Flow

The complete event-to-actor flow:

┌──────────────┐
│ PGMQ Message │ Message arrives in queue
└──────┬───────┘
       │
       ▼
┌──────────────────┐
│  Event Listener  │ EventDrivenSystem processes notification
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│ Command Channel  │ Send command to processor via tokio::mpsc
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│ Command Processor│ Convert command to actor message
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│  Actor Registry  │ Route message to appropriate actor
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│ Handler<M>::     │ Actor processes message
│    handle()      │ Delegates to underlying service
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│  Response        │ Return result to command processor
└──────────────────┘

ActorRegistry and Lifecycle

The ActorRegistry manages all 4 orchestration actors and integrates with the system lifecycle:

#![allow(unused)]
fn main() {
// During system startup
let context = Arc::new(SystemContext::with_pool(pool).await?);
let actors = ActorRegistry::build(context).await?;  // Calls started() on all actors

// During operation
let msg = FinalizeTaskMessage { task_uuid };
let result = actors.task_finalizer_actor.handle(msg).await?;

// During shutdown
actors.shutdown().await;  // Calls stopped() on all actors in reverse order
}

Current Actors:

• TaskRequestActor: Handles task initialization requests
• ResultProcessorActor: Processes step execution results
• StepEnqueuerActor: Manages batch processing of ready tasks
• TaskFinalizerActor: Handles task finalization with atomic claiming

Benefits for Event-Driven Architecture

The actor pattern enhances the event-driven architecture by providing:

1. Type Safety: Compile-time verification of message contracts
2. Consistency: Uniform lifecycle management across all components
3. Testability: Clear message boundaries for isolated testing
4. Observability: Actor-level metrics and tracing
5. Evolvability: Easy to add new message handlers and actors

Implementation Status

The actor integration is complete:

1. Phase 1 ✅: Actor infrastructure and test harness

   • OrchestrationActor, Handler, Message traits
   • ActorRegistry structure

2. Phase 2-3 ✅: All 4 primary actors implemented

   • TaskRequestActor, ResultProcessorActor
   • StepEnqueuerActor, TaskFinalizerActor

3. Phase 4-6 ✅: Message hydration and module reorganization

   • Hydration layer for PGMQ messages
   • Clean module organization

4. Phase 7 ✅: Service decomposition

   • Large services decomposed into focused components
   • All files <300 lines following single responsibility principle

5. Cleanup ✅: Direct actor integration

   • Command processor calls actors directly
   • Removed intermediate wrapper layers
   • Production-ready implementation

Service Decomposition

Large services (800-900 lines) were decomposed into focused components:

TaskFinalizer (848 → 6 files):

• service.rs: Main TaskFinalizer (~200 lines)
• completion_handler.rs: Task completion logic
• event_publisher.rs: Lifecycle event publishing
• execution_context_provider.rs: Context fetching
• state_handlers.rs: State-specific handling

StepEnqueuerService (781 → 3 files):

• service.rs: Main service (~250 lines)
• batch_processor.rs: Batch processing logic
• state_handlers.rs: State-specific processing

ResultProcessor (889 → 4 files):

• service.rs: Main processor
• metadata_processor.rs: Metadata handling
• error_handler.rs: Error processing
• result_validator.rs: Result validation

This comprehensive event and command architecture, now enhanced with the actor pattern, provides the foundation for scalable, reliable, and maintainable workflow orchestration in the tasker-core system while maintaining the flexibility to operate in diverse deployment environments.

Idempotency and Atomicity Guarantees

Last Updated: 2025-01-19 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | States and Lifecycles | Events and Commands | Task Readiness & Execution

← Back to Documentation Hub

Overview

Tasker Core is designed for distributed orchestration with multiple orchestrator instances processing tasks concurrently. This document explains the defense-in-depth approach that ensures safe concurrent operation without race conditions, data corruption, or lost work.

The system provides idempotency and atomicity through four overlapping protection layers:

1. Database Atomicity: PostgreSQL constraints, row locking, and compare-and-swap operations
2. State Machine Guards: Current-state validation before all transitions
3. Transaction Boundaries: All-or-nothing semantics for complex operations
4. Application Logic: State-based filtering and idempotent patterns

These layers work together to ensure that operations can be safely retried, multiple orchestrators can process work concurrently, and crashes don’t leave the system in an inconsistent state.

Core Protection Mechanisms

Layer 1: Database Atomicity

PostgreSQL provides fundamental atomic guarantees through several mechanisms:

Unique Constraints

Purpose: Prevent duplicate creation of entities

Key Constraints:

• tasker.tasks.identity_hash (UNIQUE) - Prevents duplicate task creation from identical requests
• tasker.task_namespaces.name (UNIQUE) - Namespace name uniqueness
• tasker.named_tasks (namespace_id, name, version) (UNIQUE) - Task template uniqueness
• tasker.named_steps.system_name (UNIQUE) - Step handler uniqueness

Example Protection:

#![allow(unused)]
fn main() {
// Two orchestrators receive identical TaskRequestMessage
// Orchestrator A creates task first -> commits successfully
// Orchestrator B attempts to create -> unique constraint violation
// Result: Exactly one task created, error cleanly handled
}

See Task Initialization for details on how this protects task creation.

Row-Level Locking

Purpose: Prevent concurrent modifications to the same database row

Locking Patterns:

1. FOR UPDATE - Exclusive lock, blocks concurrent transactions

   -- Used in: transition_task_state_atomic()
   SELECT * FROM tasker.tasks WHERE task_uuid = $1 FOR UPDATE;
   -- Blocks until transaction commits or rolls back
   

2. FOR UPDATE SKIP LOCKED - Lock-free work distribution

   -- Used in: get_next_ready_tasks()
   SELECT * FROM tasker.tasks
   WHERE state = ANY($1)
   FOR UPDATE SKIP LOCKED
   LIMIT $2;
   -- Each orchestrator gets different tasks, no blocking
   

Example Protection:

#![allow(unused)]
fn main() {
// Scenario: Two orchestrators attempt state transition on same task
// Orchestrator A: BEGIN; SELECT FOR UPDATE; UPDATE state; COMMIT;
// Orchestrator B: BEGIN; SELECT FOR UPDATE (BLOCKS until A commits)
//                 UPDATE fails due to state validation
// Result: Only one transition succeeds, no race condition
}

Compare-and-Swap Semantics

Purpose: Validate expected state before making changes

Pattern: All state transitions validate current state in the same transaction as the update

-- From transition_task_state_atomic()
UPDATE tasker.tasks
SET state = $new_state, updated_at = NOW()
WHERE task_uuid = $uuid
  AND state = $expected_current_state  -- Critical: CAS validation
RETURNING *;

Example Protection:

#![allow(unused)]
fn main() {
// Orchestrator A and B both think task is in "Pending" state
// A transitions: WHERE state = 'Pending' -> succeeds, now "Initializing"
// B transitions: WHERE state = 'Pending' -> returns 0 rows (fails gracefully)
// Result: Atomic transition, no invalid state
}

See SQL Function Architecture for more on database-level guarantees.

Layer 2: State Machine Guards

Purpose: Enforce valid state transitions through application-level validation

Both task and step state machines validate current state before allowing transitions. This provides protection even when database constraints alone wouldn’t catch invalid operations.

Task State Machine

Defined in tasker-shared/src/state_machine/task_state_machine.rs, the TaskStateMachine validates:

1. Current state retrieval: Always fetch latest state from database
2. Event applicability: Check if event is valid for current state
3. Terminal state protection: Cannot transition from Complete/Error/Cancelled
4. Ownership tracking: Processor UUID tracked for audit (not enforced after ownership removal)

Example Protection:

#![allow(unused)]
fn main() {
// TaskStateMachine prevents invalid transitions
let mut state_machine = TaskStateMachine::new(task, context);

// Attempt to mark complete when still processing
let result = state_machine.transition(TaskEvent::MarkComplete).await;
// Result: Error - cannot mark complete while steps are in progress

// Current state validation prevents:
// - Completing tasks with pending steps
// - Re-initializing completed tasks
// - Transitioning from terminal states
}

See States and Lifecycles for complete state machine documentation.

Workflow Step State Machine

Defined in tasker-shared/src/state_machine/step_state_machine.rs, the StepStateMachine ensures:

1. Execution claiming: Only Pending/Enqueued steps can transition to InProgress
2. Completion validation: Only InProgress steps can be marked complete
3. Retry eligibility: Validates max_attempts and backoff timing

Example Protection:

#![allow(unused)]
fn main() {
// Worker attempts to claim already-processing step
let mut step_machine = StepStateMachine::new(step.into(), context);

match step_machine.current_state().await {
    WorkflowStepState::InProgress => {
        // Already being processed by another worker
        return Ok(false); // Cannot claim
    }
    WorkflowStepState::Pending | WorkflowStepState::Enqueued => {
        // Attempt atomic transition
        step_machine.transition(StepEvent::Start).await?;
    }
}
}

This prevents:

• Multiple workers executing the same step concurrently
• Marking steps complete that weren’t started
• Retrying steps that exceeded max_attempts

Layer 3: Transaction Boundaries

Purpose: Ensure all-or-nothing semantics for multi-step operations

Critical operations wrap multiple database changes in a single transaction, ensuring atomic completion or full rollback on failure.

Task Initialization Transaction

Task creation involves multiple dependent entities that must all succeed or all fail:

#![allow(unused)]
fn main() {
// From TaskInitializer.initialize_task()
let mut tx = pool.begin().await?;

// 1. Create or find namespace (find-or-create is idempotent)
let namespace = NamespaceResolver::resolve_namespace(&mut tx, namespace_name).await?;

// 2. Create or find named task
let named_task = NamespaceResolver::resolve_named_task(&mut tx, namespace, task_name).await?;

// 3. Create task record
let task = create_task(&mut tx, named_task.uuid, context).await?;

// 4. Create all workflow steps and edges
let (step_count, step_mapping) = WorkflowStepBuilder::create_workflow_steps(
    &mut tx, task.uuid, template
).await?;

// 5. Initialize state machine
StateInitializer::initialize_task_state(&mut tx, task.uuid).await?;

// ALL OR NOTHING: Commit entire transaction
tx.commit().await?;
}

Example Protection:

#![allow(unused)]
fn main() {
// Scenario: Task creation partially fails
// - Namespace created ✓
// - Named task created ✓
// - Task record created ✓
// - Workflow steps: Cycle detected ✗ (error thrown)
// Result: tx.rollback() -> ALL changes reverted, clean failure
}

Cycle Detection Enforcement

Workflow dependencies are validated during task initialization to prevent circular references:

#![allow(unused)]
fn main() {
// From WorkflowStepBuilder::create_step_dependencies()
for dependency in &step_definition.dependencies {
    let from_uuid = step_mapping[dependency];
    let to_uuid = step_mapping[&step_definition.name];

    // Check for self-reference
    if from_uuid == to_uuid {
        return Err(CycleDetected { from, to });
    }

    // Check for path that would create cycle
    if WorkflowStepEdge::would_create_cycle(pool, from_uuid, to_uuid).await? {
        return Err(CycleDetected { from, to });
    }

    // Safe to create edge
    WorkflowStepEdge::create_with_transaction(&mut tx, edge).await?;
}
}

This prevents invalid DAG structures from ever being persisted to the database.

Layer 4: Application Logic Patterns

Purpose: Implement idempotent patterns at the application level

Beyond database and state machine protections, application code uses several patterns to ensure safe retry and concurrent operation.

Find-or-Create Pattern

Used for entities that should be unique but may be created by multiple orchestrators:

#![allow(unused)]
fn main() {
// From NamespaceResolver
pub async fn resolve_namespace(
    tx: &mut Transaction<'_, Postgres>,
    name: &str,
) -> Result<TaskNamespace> {
    // Try to find existing
    if let Some(namespace) = TaskNamespace::find_by_name(pool, name).await? {
        return Ok(namespace);
    }

    // Create if not found
    match TaskNamespace::create_with_transaction(tx, NewTaskNamespace { name }).await {
        Ok(namespace) => Ok(namespace),
        Err(sqlx::Error::Database(e)) if is_unique_violation(&e) => {
            // Another orchestrator created it between our find and create
            // Re-query to get the one that won the race
            TaskNamespace::find_by_name(pool, name).await?
                .ok_or(Error::NotFound)
        }
        Err(e) => Err(e),
    }
}
}

Why This Works:

• First attempt: Finds existing → idempotent
• Create attempt: Unique constraint prevents duplicates
• Retry after unique violation: Gets the winner → idempotent
• Result: Exactly one namespace, regardless of concurrent attempts

State-Based Filtering

Operations filter by state to naturally deduplicate work:

#![allow(unused)]
fn main() {
// From StepEnqueuerService
// Only enqueue steps in specific states
let ready_steps = steps.iter()
    .filter(|step| matches!(
        step.state,
        WorkflowStepState::Pending | WorkflowStepState::WaitingForRetry
    ))
    .collect();

// Skip steps already:
// - Enqueued (another orchestrator handled it)
// - InProgress (worker is executing)
// - Complete (already done)
// - Error (terminal state)
}

Example Protection:

#![allow(unused)]
fn main() {
// Scenario: Orchestrator crash mid-batch
// Before crash: Enqueued steps 1-5 of 10
// After restart: Process task again
// State filtering:
//   - Steps 1-5: state = Enqueued → skip
//   - Steps 6-10: state = Pending → enqueue
// Result: Each step enqueued exactly once
}

State-Before-Queue Pattern

Ensures workers only see steps in correct state:

#![allow(unused)]
fn main() {
// 1. Commit state transition to database FIRST
step_state_machine.transition(StepEvent::Enqueue).await?;
// Step now in Enqueued state in database

// 2. THEN send PGMQ notification
pgmq_client.send_with_notify(queue_name, step_message).await?;

// Worker receives notification and:
// - Queries database for step
// - Sees state = Enqueued (committed)
// - Can safely claim and execute
}

Why Order Matters:

#![allow(unused)]
fn main() {
// Wrong order (queue-before-state):
// 1. Send PGMQ message
// 2. Worker receives immediately
// 3. Worker queries database → state still Pending
// 4. Worker might skip or fail to claim
// 5. State transition commits

// Correct order (state-before-queue):
// 1. State transition commits
// 2. Send PGMQ message
// 3. Worker receives
// 4. Worker queries → state correctly Enqueued
// 5. Worker can claim
}

See Events and Commands for event system details.

Component-by-Component Guarantees

Task Initialization Idempotency

Component: TaskRequestActor and TaskInitializer service Operation: Creating a new task from a template File: tasker-orchestration/src/orchestration/lifecycle/task_initialization/

Protection Mechanisms

1. Identity Hash Unique Constraint

   #![allow(unused)]
   fn main() {
   // Tasks are identified by hash of (namespace, task_name, context)
   let identity_hash = calculate_identity_hash(namespace, name, context);
   
   NewTask {
       identity_hash,  // Unique constraint prevents duplicates
       named_task_uuid,
       context,
       // ...
   }
   }

2. Transaction Atomicity

   • All entities created in single transaction
   • Namespace, named task, task, workflow steps, edges
   • Cycle detection validates DAG before committing
   • Any failure rolls back everything

3. Find-or-Create for Shared Entities

   • Namespaces can be created by any orchestrator
   • Named tasks shared across workflow instances
   • Named steps reused across tasks

Concurrent Scenario

Two orchestrators receive identical TaskRequestMessage:

T0: Orchestrator A begins transaction
T1: Orchestrator B begins transaction
T2: A creates namespace "payments"
T3: B attempts to create namespace "payments"
T4: A creates task with identity_hash "abc123"
T5: B attempts to create task with identity_hash "abc123"
T6: A commits successfully ✓
T7: B attempts commit → unique constraint violation on identity_hash
T8: B transaction rolled back

Result:

• Exactly one task created
• No partial state in database
• Orchestrator B receives clear error
• Retry-safe: B can check if task exists and return it

Cycle Detection

Prevents invalid workflow definitions:

#![allow(unused)]
fn main() {
// Template defines: A depends on B, B depends on C, C depends on A
// During initialization:
//   - Create steps A, B, C
//   - Create edge A -> B (valid)
//   - Create edge B -> C (valid)
//   - Attempt edge C -> A
//     - would_create_cycle() returns true
//     - Error: CycleDetected
//   - Transaction rolled back
// Result: Invalid workflow rejected, no partial data
}

See tasker-shared/src/models/core/workflow_step_edge.rs:236-270 for cycle detection implementation.

Step Enqueueing Idempotency

Component: StepEnqueuerActor and StepEnqueuerService Operation: Enqueueing ready workflow steps to worker queues File: tasker-orchestration/src/orchestration/lifecycle/step_enqueuer_services/

Multi-Layer Protection

1. SQL-Level Row Locking

   -- get_next_ready_tasks() uses SKIP LOCKED
   SELECT task_uuid FROM tasker.tasks
   WHERE state = ANY($states)
   FOR UPDATE SKIP LOCKED  -- Prevents concurrent claiming
   LIMIT $batch_size;
   

   Each orchestrator gets different tasks, no overlap

2. State Machine Compare-and-Swap

   #![allow(unused)]
   fn main() {
   // Only transition if task in expected state
   state_machine.transition(TaskEvent::EnqueueSteps(uuids)).await?;
   // Fails if another orchestrator already transitioned
   }

3. Step State Filtering

   #![allow(unused)]
   fn main() {
   // Only enqueue steps in specific states
   let enqueueable = steps.filter(|s| matches!(
       s.state,
       WorkflowStepState::Pending | WorkflowStepState::WaitingForRetry
   ));
   }

4. State-Before-Queue Ordering

   #![allow(unused)]
   fn main() {
   // 1. Commit step state to Enqueued
   step.transition(StepEvent::Enqueue).await?;
   
   // 2. Send PGMQ message
   pgmq.send_with_notify(queue, message).await?;
   }

Concurrent Scenario

Two orchestrators discover the same ready steps:

T0: Orchestrator A queries get_next_ready_tasks(batch=100)
T1: Orchestrator B queries get_next_ready_tasks(batch=100)
T2: A gets tasks [1,2,3] (locked by A's transaction)
T3: B gets tasks [4,5,6] (different rows, SKIP LOCKED)
T4: A enqueues steps for tasks 1,2,3
T5: B enqueues steps for tasks 4,5,6
T6: Both commit successfully

Result: No overlap, each task processed once

Orchestrator Crash Mid-Batch:

T0: Orchestrator A gets task 1 with steps [A, B, C, D]
T1: A enqueues steps A, B to "payments_queue"
T2: A crashes before processing steps C, D
T3: Task 1 state still EnqueuingSteps
T4: Orchestrator B picks up task 1 (A's transaction rolled back)
T5: B queries steps for task 1
T6: Steps A, B have state = Enqueued → skip
T7: Steps C, D have state = Pending → enqueue

Result: Steps A, B enqueued once, C, D recovered and enqueued

Result Processing Idempotency

Component: ResultProcessorActor and OrchestrationResultProcessor Operation: Processing step execution results from workers File: tasker-orchestration/src/orchestration/lifecycle/result_processing/

Protection Mechanisms

1. State Guard Validation

   #![allow(unused)]
   fn main() {
   // TaskCoordinator validates step state before processing result
   let current_state = step_state_machine.current_state().await?;
   
   match current_state {
       WorkflowStepState::InProgress => {
           // Valid: step is being processed
           step_state_machine.transition(StepEvent::Complete).await?;
       }
       WorkflowStepState::Complete => {
           // Idempotent: already processed this result
           return Ok(AlreadyComplete);
       }
       _ => {
           // Invalid state for result processing
           return Err(InvalidState);
       }
   }
   }

2. Atomic State Transitions

   • Step result processing uses compare-and-swap
   • Task state transitions validate current state
   • All updates in same transaction as state check

3. Ownership Removed

   • Processor UUID tracked for audit only
   • Not enforced for transitions
   • Any orchestrator can process results
   • Enables recovery after crashes

Concurrent Scenario

Worker submits result, orchestrator crashes, retry arrives:

T0: Worker completes step A, submits result to orchestration_step_results queue
T1: Orchestrator A pulls message, begins processing
T2: A transitions step A to Complete
T3: A begins task state evaluation
T4: A crashes before deleting PGMQ message
T5: PGMQ visibility timeout expires → message reappears
T6: Orchestrator B pulls same message
T7: B queries step A state → Complete
T8: B returns early (idempotent, already processed)
T9: B deletes PGMQ message

Result: Step processed exactly once, retry is harmless

Before Ownership Removal (Ownership Enforced):

// Orchestrator A owned task in EvaluatingResults state
// A crashes
// B receives retry
// B checks: task.processor_uuid != B.uuid
// Error: Ownership violation → TASK STUCK

After Ownership Removal (Ownership Audit-Only):

// Orchestrator A owned task in EvaluatingResults state
// A crashes
// B receives retry
// B checks: current task state (no ownership check)
// B processes successfully → TASK RECOVERS

See the Ownership Removal ADR for full analysis.

Task Finalization Idempotency

Component: TaskFinalizerActor and TaskFinalizer service Operation: Finalizing task to terminal state File: tasker-orchestration/src/orchestration/lifecycle/task_finalization/

Current Protection (Sufficient for Recovery)

1. State Guard Protection

   #![allow(unused)]
   fn main() {
   // TaskFinalizer checks current task state
   let context = ExecutionContextProvider::fetch(task_uuid).await?;
   
   match context.should_finalize() {
       true => {
           // Transition to Complete
           task_state_machine.transition(TaskEvent::MarkComplete).await?;
       }
       false => {
           // Not ready to finalize (steps still pending)
           return Ok(NotReady);
       }
   }
   }

2. Idempotent for Recovery

   #![allow(unused)]
   fn main() {
   // Scenario: Orchestrator crashes during finalization
   // - Task state already Complete → state guard returns early
   // - Task state still StepsInProcess → retry succeeds
   // Result: Recovery works, final state reached
   }

Concurrent Scenario (Not Graceful)

Two orchestrators attempt finalization simultaneously:

T0: Orchestrators A and B both receive finalization trigger
T1: A checks: all steps complete → proceed
T2: B checks: all steps complete → proceed
T3: A transitions task to Complete (succeeds)
T4: B attempts transition to Complete
T5: State guard: task already Complete
T6: B receives StateMachineError (invalid transition)

Result:

• ✓ Task finalized exactly once (correct)
• ✓ No data corruption
• ⚠️ Orchestrator B gets error (not graceful)

Future Enhancement: Atomic Finalization Claiming

Atomic claiming would make concurrent finalization graceful:

-- Proposed claim_task_for_finalization() function
UPDATE tasker.tasks
SET finalization_claimed_at = NOW(),
    finalization_claimed_by = $processor_uuid
WHERE task_uuid = $uuid
  AND state = 'StepsInProcess'
  AND finalization_claimed_at IS NULL
RETURNING *;

With atomic finalization claiming:

T0: Orchestrators A and B both receive finalization trigger
T1: A calls claim_task_for_finalization() → succeeds
T2: B calls claim_task_for_finalization() → returns 0 rows
T3: A proceeds with finalization
T4: B returns early (silent success, already claimed)

This enhancement is deferred (implementation not yet scheduled).

SQL Function Atomicity

File: tasker-shared/src/database/sql/ Documented: Task Readiness & Execution

Atomic State Transitions

Function: transition_task_state_atomic() Protection: Compare-and-swap with row locking

-- Atomic state transition with validation
UPDATE tasker.tasks
SET state = $new_state,
    updated_at = NOW()
WHERE task_uuid = $uuid
  AND state = $expected_current_state  -- CAS: only if state matches
FOR UPDATE;  -- Lock prevents concurrent modifications

Key Guarantees:

• Returns 0 rows if state doesn’t match → safe retry
• Row lock prevents concurrent transitions
• Processor UUID tracked for audit, not enforced

Work Distribution Without Contention

Function: get_next_ready_tasks() Protection: Lock-free claiming via SKIP LOCKED

SELECT task_uuid, correlation_id, state
FROM tasker.tasks
WHERE state = ANY($processable_states)
  AND (
    state NOT IN ('WaitingForRetry') OR
    last_retry_at + retry_interval < NOW()
  )
ORDER BY
  CASE state
    WHEN 'Pending' THEN 1
    WHEN 'WaitingForRetry' THEN 2
    ELSE 3
  END,
  created_at ASC
FOR UPDATE SKIP LOCKED  -- Skip locked rows, no blocking
LIMIT $batch_size;

Key Guarantees:

• Each orchestrator gets different tasks
• No blocking or contention
• Dynamic priority (Pending before WaitingForRetry)
• Prevents task starvation

Step Readiness with Dependency Validation

Function: get_step_readiness_status() Protection: Validates dependencies in single query

WITH step_dependencies AS (
  SELECT COUNT(*) as total_deps,
         SUM(CASE WHEN dep_step.state = 'Complete' THEN 1 ELSE 0 END) as completed_deps
  FROM tasker.workflow_step_edges e
  JOIN tasker.workflow_steps dep_step ON e.from_step_uuid = dep_step.uuid
  WHERE e.to_step_uuid = $step_uuid
)
SELECT
  CASE
    WHEN total_deps = completed_deps THEN 'Ready'
    WHEN step.state = 'Error' AND step.attempts < step.max_attempts THEN 'WaitingForRetry'
    ELSE 'Blocked'
  END as readiness
FROM step_dependencies, tasker.workflow_steps step
WHERE step.uuid = $step_uuid;

Key Guarantees:

• Atomic dependency check
• Handles retry logic with backoff
• Prevents premature execution

Cycle Detection

Function: WorkflowStepEdge::would_create_cycle() (Rust, uses SQL) Protection: Recursive CTE path traversal

WITH RECURSIVE step_path AS (
  -- Base: Start from proposed destination
  SELECT from_step_uuid, to_step_uuid, 1 as depth
  FROM tasker.workflow_step_edges
  WHERE from_step_uuid = $proposed_to

  UNION ALL

  -- Recursive: Follow edges
  SELECT sp.from_step_uuid, wse.to_step_uuid, sp.depth + 1
  FROM step_path sp
  JOIN tasker.workflow_step_edges wse ON sp.to_step_uuid = wse.from_step_uuid
  WHERE sp.depth < 100  -- Prevent infinite recursion
)
SELECT COUNT(*) as has_path
FROM step_path
WHERE to_step_uuid = $proposed_from;

Returns: True if adding edge would create cycle

Enforcement: Called by WorkflowStepBuilder during task initialization

• Self-reference check: from_uuid == to_uuid
• Path check: Would adding edge create cycle?
• Error before commit: Transaction rolled back on cycle

See tasker-orchestration/src/orchestration/lifecycle/task_initialization/workflow_step_builder.rs for enforcement.

Cross-Cutting Scenarios

Multiple Orchestrators Processing Same Task

Scenario: Load balancer distributes work to multiple orchestrators

Protection:

1. Work Distribution:

   -- Each orchestrator gets different tasks via SKIP LOCKED
   Orchestrator A: Tasks [1, 2, 3]
   Orchestrator B: Tasks [4, 5, 6]
   

2. State Transitions:

   #![allow(unused)]
   fn main() {
   // Both attempt to transition same task (shouldn't happen, but...)
   A: transition(Pending -> Initializing) → succeeds
   B: transition(Pending -> Initializing) → fails (state already changed)
   }

3. Step Enqueueing:

   #![allow(unused)]
   fn main() {
   // Task in EnqueuingSteps state
   A: Processes task, enqueues steps A, B
   B: Cannot claim task (state not in processable states)
   // OR if B claims during transition:
   B: Filters steps by state → A, B already Enqueued, skips them
   }

Result: No duplicate work, clean coordination

Orchestrator Crashes and Recovers

Scenario: Orchestrator crashes mid-operation, another takes over

During Task Initialization

Before ownership removal:
T0: Orchestrator A initializes task 1
T1: Task transitions to Initializing (processor_uuid = A)
T2: A crashes
T3: Task stuck in Initializing forever (ownership blocks recovery)

After ownership removal:
T0: Orchestrator A initializes task 1
T1: Task transitions to Initializing (processor_uuid = A for audit)
T2: A crashes
T3: Orchestrator B picks up task 1
T4: B transitions Initializing -> EnqueuingSteps (succeeds, no ownership check)
T5: Task recovers automatically

During Step Enqueueing

T0: Orchestrator A enqueues steps [A, B] of task 1
T1: A crashes before committing
T2: Transaction rolls back
T3: Steps A, B remain in Pending state
T4: Orchestrator B picks up task 1
T5: B enqueues steps A, B (state still Pending)
T6: No duplicate work

During Result Processing

T0: Worker completes step A
T1: Orchestrator A receives result, transitions step to Complete
T2: A crashes before updating task state
T3: PGMQ message visibility timeout expires
T4: Orchestrator B receives same result message
T5: B queries step A → already Complete
T6: B skips processing (idempotent)
T7: B evaluates task state, continues workflow

Result: Complete recovery, no manual intervention

Retry After Transient Failure

Scenario: Database connection lost during operation

#![allow(unused)]
fn main() {
// Orchestrator attempts task initialization
let result = task_initializer.initialize(request).await;

match result {
    Err(TaskInitializationError::Database(_)) => {
        // Transient failure (connection lost)
        // Retry same request
        let retry_result = task_initializer.initialize(request).await;

        // Possibilities:
        // 1. Succeeds: Transaction completed before connection lost
        //    → identity_hash unique constraint prevents duplicate
        //    → Get existing task
        // 2. Succeeds: Transaction rolled back
        //    → Create task successfully
        // 3. Fails: Different error
        //    → Handle appropriately
    }
    Ok(task) => { /* Success */ }
}
}

Key Pattern: Operations are designed to be retry-safe

• Database constraints prevent duplicates
• State guards prevent invalid transitions
• Find-or-create handles concurrent creation

PGMQ Message Duplicate Delivery

Scenario: PGMQ message processed twice due to visibility timeout

#![allow(unused)]
fn main() {
// Worker completes step, sends result
pgmq.send("orchestration_step_results", result).await?;

// Orchestrator A receives message
let message = pgmq.read("orchestration_step_results").await?;

// A processes result
result_processor.process(message.payload).await?;

// A about to delete message, crashes
// Message visibility timeout expires → message reappears

// Orchestrator B receives same message
let duplicate = pgmq.read("orchestration_step_results").await?;

// B processes result
// State machine checks: step already Complete
// Returns early (idempotent)
result_processor.process(duplicate.payload).await?; // Harmless

// B deletes message
pgmq.delete(duplicate.msg_id).await?;
}

Protection:

• State guards: Check current state before processing
• Idempotent handlers: Safe to process same message multiple times
• Message deletion: Only after confirmed processing

See Events and Commands for PGMQ architecture.

Multi-Instance Validation

The defense-in-depth architecture was validated through comprehensive multi-instance cluster testing. This section documents the validation results and confirms the effectiveness of the protection mechanisms.

Test Configuration

• Orchestration Instances: 2 (ports 8080, 8081)
• Worker Instances: 2 per type (Rust: 8100-8101, Ruby: 8200-8201, Python: 8300-8301, TypeScript: 8400-8401)
• Total Services: 10 concurrent instances
• Database: Shared PostgreSQL with PGMQ messaging

Validation Results

What Was Validated

1. Concurrent Task Creation

   • Tasks created through different orchestration instances
   • No duplicate tasks or UUIDs
   • All tasks complete successfully
   • State consistent across all instances

2. Work Distribution

   • SKIP LOCKED distributes tasks without overlap
   • Multiple workers claim different steps
   • No duplicate step processing

3. State Machine Guards

   • Invalid transitions rejected at state machine layer
   • Compare-and-swap prevents concurrent modifications
   • Terminal states protected from re-entry

4. Transaction Boundaries

   • All-or-nothing semantics maintained under load
   • No partial task initialization observed
   • Crash recovery works correctly

5. Cross-Instance Consistency

   • Task state queries return same result from any instance
   • Step state transitions visible immediately to all instances
   • No stale reads observed

Protection Layer Effectiveness

Intermittent Failures Analysis

Three tests showed intermittent failures under heavy parallelization:

• Root Cause: Database connection pool exhaustion when running 1600+ tests in parallel
• Evidence: Failures occurred only at high parallelism (>4 threads), not with serialized execution
• Classification: Resource contention, NOT race conditions
• Mitigation: Nextest configured with test-threads = 1 for multi_instance tests

Key Finding: No race conditions were detected. All intermittent failures traced to resource limits.

Domain Event Tests

21 tests were excluded from cluster mode using #[cfg(not(feature = "test-cluster"))]:

• Reason: Domain event tests verify in-process event delivery (publish/subscribe within single process)
• Behavior in Cluster: Events published in one instance aren’t delivered to subscribers in another instance
• Status: Working as designed - these tests run correctly in single-instance CI

Stress Test Results

Rapid Task Burst Test:

• 25 tasks created in <1 second
• All tasks completed successfully
• No duplicate UUIDs
• Creation rate: ~50 tasks/second sustained

Round-Robin Distribution Test:

• Tasks distributed evenly across orchestration instances
• Load balancing working correctly
• No single-instance bottleneck

Recommendations Validated

The following architectural decisions were validated by cluster testing:

1. Ownership Removal: Processor UUID as audit-only (not enforced) enables automatic recovery
2. SKIP LOCKED Pattern: Effective for contention-free work distribution
3. State-Before-Queue Pattern: Prevents workers from seeing uncommitted state
4. Find-or-Create Pattern: Handles concurrent entity creation correctly

Future Enhancements Identified

Testing identified one P2 improvement opportunity:

Atomic Finalization Claiming

• Current: Second orchestrator gets StateMachineError during concurrent finalization
• Proposed: Transaction-based locking for graceful handling
• Priority: P2 (operational improvement, correctness already ensured)

Running Cluster Validation

To reproduce the validation:

# Setup cluster environment
cargo make setup-env-cluster

# Start full cluster
cargo make cluster-start-all

# Run all tests including cluster tests
cargo make test-rust-all

# Stop cluster
cargo make cluster-stop

See Cluster Testing Guide for detailed instructions.

Design Principles

Defense in Depth

The system intentionally provides multiple overlapping protection layers rather than relying on a single mechanism. This ensures:

1. Resilience: If one layer fails (e.g., application bug), others prevent corruption
2. Clear Semantics: Each layer has a specific purpose and failure mode
3. Ease of Reasoning: Developers can understand guarantees at each level
4. Graceful Degradation: System remains safe even under partial failures

Fail-Safe Defaults

When in doubt, the system errs on the side of caution:

• State transitions fail if current state doesn’t match → prevents invalid states
• Unique constraints fail creation → prevents duplicates
• Row locks block concurrent access → prevents race conditions
• Cycle detection fails initialization → prevents invalid workflows

Better to fail cleanly than to corrupt data.

Retry Safety

All critical operations are designed to be safely retryable:

• Idempotent: Same operation, repeated → same outcome
• State-Based: Check current state before acting
• Atomic: All-or-nothing commits
• No Side Effects: Operations don’t accumulate partial state

This enables:

• Automatic retry after transient failures
• Duplicate message handling
• Recovery after crashes
• Horizontal scaling without coordination overhead

Audit Trail Without Enforcement

Ownership Decision: Track ownership for observability, don’t enforce for correctness

#![allow(unused)]
fn main() {
// Processor UUID recorded in all transitions
pub struct TaskTransition {
    pub task_uuid: Uuid,
    pub from_state: TaskState,
    pub to_state: TaskState,
    pub processor_uuid: Uuid,  // For audit and debugging
    pub event: String,
    pub timestamp: DateTime<Utc>,
}

// But NOT enforced in transition logic
impl TaskStateMachine {
    pub async fn transition(&mut self, event: TaskEvent) -> Result<TaskState> {
        // ✅ Tracks processor UUID
        // ❌ Does NOT require ownership match
        // Reason: Enables recovery after crashes
    }
}
}

Why This Works:

• State guards provide correctness (current state validation)
• Processor UUID provides observability (who did what when)
• No ownership blocking means automatic recovery
• Full audit trail for debugging and monitoring

Implementation Checklist

When implementing new orchestration operations, ensure:

Database Layer

• Unique constraints for entities that must be singular
• FOR UPDATE locking for state transitions
• FOR UPDATE SKIP LOCKED for work distribution
• Compare-and-swap (CAS) in UPDATE WHERE clauses
• Transaction wrapping for multi-step operations

State Machine Layer

• Current state retrieval before transitions
• Event applicability validation
• Terminal state protection
• Error handling for invalid transitions

Application Layer

• Find-or-create pattern for shared entities
• State-based filtering before processing
• State-before-queue ordering for events
• Idempotent message handlers

Testing

• Concurrent operation tests (multiple orchestrators)
• Crash recovery tests (mid-operation failures)
• Retry safety tests (duplicate message handling)
• Race condition tests (timing-dependent scenarios)

Related Documentation

Core Architecture

• States and Lifecycles - Dual state machine architecture
• Events and Commands - Event-driven coordination patterns
• Actor-Based Architecture - Orchestration actor pattern
• Task Readiness & Execution - SQL functions and execution logic

Implementation Details

• Ownership Removal ADR - Processor UUID ownership removal decision

Multi-Instance Validation

• Cluster Testing Guide - Running multi-instance cluster tests

Testing

• Comprehensive Lifecycle Testing - Testing patterns including concurrent scenarios

← Back to Documentation Hub

Messaging Abstraction Architecture

Last Updated: 2026-01-15 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | Events and Commands | Deployment Patterns | Crate Architecture

<- Back to Documentation Hub

Overview

The provider-agnostic messaging abstraction enables Tasker Core to support multiple messaging backends through a unified interface. This architecture allows switching between PGMQ (PostgreSQL Message Queue) and RabbitMQ without changes to business logic.

Key Benefits:

• Zero handler changes required: Switching providers requires only configuration changes
• Provider-specific optimizations: Each backend can leverage its native strengths
• Testability: In-memory provider for fast unit testing
• Gradual migration: Systems can transition between providers incrementally

Core Concepts

Message Delivery Models

Different messaging providers have fundamentally different delivery models:

PGMQ (Signal-Only):

• pg_notify sends a signal that a message exists
• Worker must fetch the message after receiving the signal
• Fallback polling catches missed signals

RabbitMQ (Full Message Push):

• basic_consume() delivers complete messages
• No separate fetch required
• Protocol guarantees delivery

Architecture Layers

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Application Layer                                  │
│  (Orchestration, Workers, Event Systems)                                    │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    │ Uses MessageClient
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                           MessageClient                                      │
│  Domain-level facade with queue classification                              │
│  Location: tasker-shared/src/messaging/client.rs                           │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    │ Delegates to
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         MessagingProvider Enum                               │
│  Runtime dispatch without trait objects (zero-cost abstraction)             │
│  Location: tasker-shared/src/messaging/service/provider.rs                 │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    │               │               │
                    ▼               ▼               ▼
            ┌───────────┐   ┌───────────┐   ┌───────────┐
            │   PGMQ    │   │ RabbitMQ  │   │ InMemory  │
            │ Provider  │   │ Provider  │   │ Provider  │
            └───────────┘   └───────────┘   └───────────┘

Core Traits and Types

MessagingService Trait

Location: tasker-shared/src/messaging/service/traits.rs

The foundational trait defining queue operations:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait MessagingService: Send + Sync {
    // Queue lifecycle
    async fn create_queue(&self, name: &str) -> Result<(), MessagingError>;
    async fn delete_queue(&self, name: &str) -> Result<(), MessagingError>;
    async fn queue_exists(&self, name: &str) -> Result<bool, MessagingError>;
    async fn list_queues(&self) -> Result<Vec<String>, MessagingError>;

    // Message operations
    async fn send_message(&self, queue: &str, payload: &[u8]) -> Result<i64, MessagingError>;
    async fn send_message_with_delay(&self, queue: &str, payload: &[u8], delay_seconds: i64) -> Result<i64, MessagingError>;
    async fn receive_messages(&self, queue: &str, limit: i32, visibility_timeout: i32) -> Result<Vec<QueuedMessage<Vec<u8>>>, MessagingError>;
    async fn ack_message(&self, queue: &str, msg_id: i64) -> Result<(), MessagingError>;
    async fn nack_message(&self, queue: &str, msg_id: i64) -> Result<(), MessagingError>;

    // Provider information
    fn provider_name(&self) -> &'static str;
}
}

SupportsPushNotifications Trait

Location: tasker-shared/src/messaging/service/traits.rs

Extends MessagingService with push notification capabilities:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait SupportsPushNotifications: MessagingService {
    /// Subscribe to messages on a single queue
    fn subscribe(&self, queue_name: &str)
        -> Result<Pin<Box<dyn Stream<Item = MessageNotification> + Send>>, MessagingError>;

    /// Subscribe to messages on multiple queues
    fn subscribe_many(&self, queue_names: &[String])
        -> Result<Pin<Box<dyn Stream<Item = MessageNotification> + Send>>, MessagingError>;

    /// Whether this provider requires fallback polling for reliability
    fn requires_fallback_polling(&self) -> bool;

    /// Suggested polling interval if fallback is needed
    fn fallback_polling_interval(&self) -> Option<Duration>;

    /// Whether this provider supports fetching by message ID
    fn supports_fetch_by_message_id(&self) -> bool;
}
}

MessageNotification Enum

Location: tasker-shared/src/messaging/service/traits.rs

Abstracts the two notification models:

#![allow(unused)]
fn main() {
pub enum MessageNotification {
    /// Signal-only notification (PGMQ style)
    /// Indicates a message is available but requires separate fetch
    Available {
        queue_name: String,
        msg_id: Option<i64>,
    },

    /// Full message notification (RabbitMQ style)
    /// Contains the complete message payload
    Message(QueuedMessage<Vec<u8>>),
}
}

Provider Implementations

PGMQ Provider

Location: tasker-shared/src/messaging/service/providers/pgmq.rs

PostgreSQL-based message queue with LISTEN/NOTIFY integration:

#![allow(unused)]
fn main() {
impl SupportsPushNotifications for PgmqMessagingService {
    fn subscribe_many(&self, queue_names: &[String])
        -> Result<Pin<Box<dyn Stream<Item = MessageNotification> + Send>>, MessagingError>
    {
        // Uses PgmqNotifyListener for pg_notify subscription
        // Returns MessageNotification::Available (signal-only) for large messages
        // Returns MessageNotification::Message for small messages (<7KB)
    }

    fn requires_fallback_polling(&self) -> bool {
        true  // pg_notify can miss signals during connection issues
    }

    fn supports_fetch_by_message_id(&self) -> bool {
        true  // PGMQ supports read_specific_message()
    }
}
}

Characteristics:

• Uses PostgreSQL for storage and delivery
• pg_notify for real-time notifications
• Fallback polling required for reliability
• Supports visibility timeout for message claiming

RabbitMQ Provider

Location: tasker-shared/src/messaging/service/providers/rabbitmq.rs

AMQP-based message broker with native push delivery:

#![allow(unused)]
fn main() {
impl SupportsPushNotifications for RabbitMqMessagingService {
    fn subscribe_many(&self, queue_names: &[String])
        -> Result<Pin<Box<dyn Stream<Item = MessageNotification> + Send>>, MessagingError>
    {
        // Uses lapin basic_consume() for native push delivery
        // Always returns MessageNotification::Message (full payload)
    }

    fn requires_fallback_polling(&self) -> bool {
        false  // AMQP protocol guarantees delivery
    }

    fn supports_fetch_by_message_id(&self) -> bool {
        false  // RabbitMQ doesn't support fetch-by-ID
    }
}
}

Characteristics:

• Native push delivery via AMQP protocol
• No fallback polling needed
• Higher throughput for high-volume scenarios
• Requires separate infrastructure (RabbitMQ server)

InMemory Provider

Location: tasker-shared/src/messaging/service/providers/in_memory.rs

In-process message queue for testing:

#![allow(unused)]
fn main() {
impl SupportsPushNotifications for InMemoryMessagingService {
    fn requires_fallback_polling(&self) -> bool {
        false  // In-memory is reliable within process
    }
}
}

Use Cases:

• Unit testing without external dependencies
• Integration testing with controlled timing
• Development environments

MessagingProvider Enum

Location: tasker-shared/src/messaging/service/provider.rs

Enum dispatch pattern for runtime provider selection without trait objects:

#![allow(unused)]
fn main() {
pub enum MessagingProvider {
    Pgmq(PgmqMessagingService),
    RabbitMq(RabbitMqMessagingService),
    InMemory(InMemoryMessagingService),
}

impl MessagingProvider {
    /// Delegate all MessagingService methods to the underlying provider
    pub async fn send_message(&self, queue: &str, payload: &[u8]) -> Result<i64, MessagingError> {
        match self {
            Self::Pgmq(p) => p.send_message(queue, payload).await,
            Self::RabbitMq(p) => p.send_message(queue, payload).await,
            Self::InMemory(p) => p.send_message(queue, payload).await,
        }
    }

    /// Subscribe to push notifications
    pub fn subscribe_many(&self, queue_names: &[String])
        -> Result<Pin<Box<dyn Stream<Item = MessageNotification> + Send>>, MessagingError>
    {
        match self {
            Self::Pgmq(p) => p.subscribe_many(queue_names),
            Self::RabbitMq(p) => p.subscribe_many(queue_names),
            Self::InMemory(p) => p.subscribe_many(queue_names),
        }
    }

    /// Check if fallback polling is required
    pub fn requires_fallback_polling(&self) -> bool {
        match self {
            Self::Pgmq(p) => p.requires_fallback_polling(),
            Self::RabbitMq(p) => p.requires_fallback_polling(),
            Self::InMemory(p) => p.requires_fallback_polling(),
        }
    }
}
}

Benefits:

• Zero-cost abstraction (no vtable indirection)
• Exhaustive match ensures all providers handled
• Easy to add new providers

MessageClient Facade

Location: tasker-shared/src/messaging/client.rs

Domain-level facade providing high-level queue operations:

#![allow(unused)]
fn main() {
pub struct MessageClient {
    provider: Arc<MessagingProvider>,
    classifier: QueueClassifier,
}

impl MessageClient {
    /// Send a step message to the appropriate namespace queue
    pub async fn send_step_message(
        &self,
        namespace: &str,
        step: &SimpleStepMessage,
    ) -> Result<i64, MessagingError> {
        let queue_name = self.classifier.step_queue_for_namespace(namespace);
        let payload = serde_json::to_vec(step)?;
        self.provider.send_message(&queue_name, &payload).await
    }

    /// Send a step result to the orchestration queue
    pub async fn send_step_result(
        &self,
        result: &StepExecutionResult,
    ) -> Result<i64, MessagingError> {
        let queue_name = self.classifier.orchestration_results_queue();
        let payload = serde_json::to_vec(result)?;
        self.provider.send_message(&queue_name, &payload).await
    }

    /// Access the underlying provider for advanced operations
    pub fn provider(&self) -> &MessagingProvider {
        &self.provider
    }
}
}

Event System Integration

Provider-Agnostic Queue Listeners

Both orchestration and worker queue listeners use provider.subscribe_many():

#![allow(unused)]
fn main() {
// tasker-orchestration/src/orchestration/orchestration_queues/listener.rs
impl OrchestrationQueueListener {
    pub async fn start(&mut self) -> Result<(), MessagingError> {
        let queues = vec![
            self.classifier.orchestration_results_queue(),
            self.classifier.orchestration_requests_queue(),
            self.classifier.orchestration_finalization_queue(),
        ];

        // Provider-agnostic subscription
        let stream = self.provider.subscribe_many(&queues)?;

        // Process notifications
        while let Some(notification) = stream.next().await {
            match notification {
                MessageNotification::Available { queue_name, msg_id } => {
                    // PGMQ style: send event command to fetch message
                    self.send_event_command(queue_name, msg_id).await;
                }
                MessageNotification::Message(msg) => {
                    // RabbitMQ style: send message command with full payload
                    self.send_message_command(msg).await;
                }
            }
        }
    }
}
}

Deployment Mode Selection

Event systems select the appropriate mode based on provider capabilities:

#![allow(unused)]
fn main() {
// Determine effective deployment mode for this provider
let effective_mode = deployment_mode.effective_for_provider(provider.provider_name());

match effective_mode {
    DeploymentMode::EventDrivenOnly => {
        // Start queue listener only (no fallback poller)
        // RabbitMQ typically uses this mode
    }
    DeploymentMode::Hybrid => {
        // Start both listener and fallback poller
        // PGMQ uses this mode for reliability
    }
    DeploymentMode::PollingOnly => {
        // Start fallback poller only
        // For restricted environments
    }
}
}

Command Routing

Dual Command Variants

Command processors handle both notification types:

#![allow(unused)]
fn main() {
pub enum OrchestrationCommand {
    // For full message notifications (RabbitMQ)
    ProcessStepResultFromMessage {
        queue_name: String,
        message: QueuedMessage<Vec<u8>>,
        resp: CommandResponder<StepProcessResult>,
    },

    // For signal-only notifications (PGMQ)
    ProcessStepResultFromMessageEvent {
        message_event: MessageReadyEvent,
        resp: CommandResponder<StepProcessResult>,
    },
}
}

Routing Logic:

• MessageNotification::Message -> ProcessStepResultFromMessage
• MessageNotification::Available -> ProcessStepResultFromMessageEvent

Type-Safe Channel Wrappers

NewType wrappers for MPSC channels prevent accidental misuse:

Orchestration Channels

Location: tasker-orchestration/src/orchestration/channels.rs

#![allow(unused)]
fn main() {
/// Strongly-typed sender for orchestration commands
#[derive(Debug, Clone)]
pub struct OrchestrationCommandSender(pub(crate) mpsc::Sender<OrchestrationCommand>);

/// Strongly-typed receiver for orchestration commands
#[derive(Debug)]
pub struct OrchestrationCommandReceiver(pub(crate) mpsc::Receiver<OrchestrationCommand>);

/// Strongly-typed sender for orchestration notifications
#[derive(Debug, Clone)]
pub struct OrchestrationNotificationSender(pub(crate) mpsc::Sender<OrchestrationNotification>);

/// Strongly-typed receiver for orchestration notifications
#[derive(Debug)]
pub struct OrchestrationNotificationReceiver(pub(crate) mpsc::Receiver<OrchestrationNotification>);
}

Worker Channels

Location: tasker-worker/src/worker/channels.rs

#![allow(unused)]
fn main() {
/// Strongly-typed sender for worker commands
#[derive(Debug, Clone)]
pub struct WorkerCommandSender(pub(crate) mpsc::Sender<WorkerCommand>);

/// Strongly-typed receiver for worker commands
#[derive(Debug)]
pub struct WorkerCommandReceiver(pub(crate) mpsc::Receiver<WorkerCommand>);
}

Channel Factory

#![allow(unused)]
fn main() {
pub struct ChannelFactory;

impl ChannelFactory {
    /// Create type-safe orchestration command channel pair
    pub fn orchestration_command_channel(buffer_size: usize)
        -> (OrchestrationCommandSender, OrchestrationCommandReceiver)
    {
        let (tx, rx) = mpsc::channel(buffer_size);
        (OrchestrationCommandSender(tx), OrchestrationCommandReceiver(rx))
    }
}
}

Benefits:

• Compile-time prevention of channel misuse
• Self-documenting function signatures
• Zero runtime overhead (NewTypes compile away)

Configuration

Provider Selection

# config/dotenv/test.env
# Valid values: pgmq (default), rabbitmq
TASKER_MESSAGING_BACKEND=pgmq

# RabbitMQ connection (only used when backend=rabbitmq)
RABBITMQ_URL=amqp://tasker:tasker@localhost:5672/%2F

Provider-Specific Settings

# config/tasker/base/common.toml
[pgmq]
visibility_timeout_seconds = 60
max_message_size_bytes = 1048576
batch_size = 100

[rabbitmq]
prefetch_count = 100
connection_timeout_seconds = 30
heartbeat_seconds = 60

Migration Guide

Switching from PGMQ to RabbitMQ

1. Deploy RabbitMQ infrastructure
2. Update configuration:

   export TASKER_MESSAGING_BACKEND=rabbitmq
   export RABBITMQ_URL=amqp://user:pass@rabbitmq:5672/%2F
   

3. Restart services - No code changes required

Gradual Migration

For zero-downtime migration:

1. Deploy new services with RabbitMQ configuration
2. Gradually shift traffic to new services
3. Monitor for any issues
4. Decommission PGMQ-based services

Testing

Provider-Agnostic Tests

Most tests should use InMemoryMessagingService for speed:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_step_execution() {
    let provider = MessagingProvider::InMemory(InMemoryMessagingService::new());
    let client = MessageClient::new(Arc::new(provider));

    // Test with in-memory provider
    client.send_step_message("payments", &step_msg).await.unwrap();
}
}

Provider-Specific Tests

For integration tests that need specific provider behavior:

#![allow(unused)]
fn main() {
#[tokio::test]
#[cfg(feature = "integration-tests")]
async fn test_pgmq_notifications() {
    let provider = MessagingProvider::Pgmq(PgmqMessagingService::new(pool).await?);
    // Test PGMQ-specific behavior
}
}

Best Practices

1. Use MessageClient for Application Code

#![allow(unused)]
fn main() {
// Good: Use domain-level facade
let client = context.message_client();
client.send_step_result(&result).await?;

// Avoid: Direct provider access unless necessary
let provider = context.messaging_provider();
provider.send_message("queue", &payload).await?;
}

2. Handle Both Notification Types

#![allow(unused)]
fn main() {
match notification {
    MessageNotification::Available { queue_name, msg_id } => {
        // Signal-only: need to fetch message
    }
    MessageNotification::Message(msg) => {
        // Full message: can process immediately
    }
}
}

3. Respect Provider Capabilities

#![allow(unused)]
fn main() {
if provider.supports_fetch_by_message_id() {
    // Can use read_specific_message()
} else {
    // Must use alternative approach
}
}

4. Configure Fallback Appropriately

#![allow(unused)]
fn main() {
if provider.requires_fallback_polling() {
    // Start fallback poller for reliability
}
}

Related Documentation

• Events and Commands - Command pattern details
• Deployment Patterns - Deployment modes and configuration
• Worker Event Systems - Worker event architecture
• Crate Architecture - Workspace structure

<- Back to Documentation Hub

Next: Events and Commands | Deployment Patterns

States and Lifecycles

Last Updated: 2025-10-10 Audience: All Status: Active Related Docs: Documentation Hub | Events and Commands | Task Readiness & Execution

← Back to Documentation Hub

This document provides comprehensive documentation of the state machine architecture in tasker-core, covering both task and workflow step lifecycles, their state transitions, and the underlying persistence mechanisms.

Overview

The tasker-core system implements a sophisticated dual-state-machine architecture:

1. Task State Machine: Manages overall workflow orchestration with 12 comprehensive states
2. Workflow Step State Machine: Manages individual step execution with 8 states including orchestration queuing

Both state machines work in coordination to provide atomic, auditable, and resilient workflow execution with proper event-driven communication between orchestration and worker systems.

Task State Machine Architecture

Task State Definitions

The task state machine implements 12 comprehensive states as defined in tasker-shared/src/state_machine/states.rs:

Initial States

• Pending: Created but not started (default initial state)
• Initializing: Discovering initial ready steps and setting up task context

Active Processing States

• EnqueuingSteps: Actively enqueuing ready steps to worker queues
• StepsInProcess: Steps are being processed by workers (orchestration monitoring)
• EvaluatingResults: Processing results from completed steps and determining next actions

Waiting States

• WaitingForDependencies: No ready steps, waiting for dependencies to be satisfied
• WaitingForRetry: Waiting for retry timeout before attempting failed steps again
• BlockedByFailures: Has failures that prevent progress (manual intervention may be needed)

Terminal States

• Complete: All steps completed successfully (terminal)
• Error: Task failed permanently (terminal)
• Cancelled: Task was cancelled (terminal)
• ResolvedManually: Manually resolved by operator (terminal)

Task State Properties

Each state has key properties that drive system behavior:

#![allow(unused)]
fn main() {
impl TaskState {
    pub fn is_terminal(&self) -> bool         // Cannot transition further
    pub fn requires_ownership(&self) -> bool  // Processor ownership required
    pub fn is_active(&self) -> bool          // Currently being processed  
    pub fn is_waiting(&self) -> bool         // Waiting for external conditions
    pub fn can_be_processed(&self) -> bool   // Available for orchestration pickup
}
}

Ownership-Required States: Initializing, EnqueuingSteps, StepsInProcess, EvaluatingResults Processable States: Pending, WaitingForDependencies, WaitingForRetry

Task Lifecycle Flow

stateDiagram-v2
    [*] --> Pending
    
    %% Initial Flow
    Pending --> Initializing : Start
    
    %% From Initializing
    Initializing --> EnqueuingSteps : ReadyStepsFound(count)
    Initializing --> Complete : NoStepsFound
    Initializing --> WaitingForDependencies : NoDependenciesReady
    
    %% Processing Flow
    EnqueuingSteps --> StepsInProcess : StepsEnqueued(uuids)
    EnqueuingSteps --> Error : EnqueueFailed(error)
    
    StepsInProcess --> EvaluatingResults : AllStepsCompleted
    StepsInProcess --> EvaluatingResults : StepCompleted(uuid)
    StepsInProcess --> WaitingForRetry : StepFailed(uuid)
    
    %% Result Evaluation
    EvaluatingResults --> Complete : AllStepsSuccessful
    EvaluatingResults --> EnqueuingSteps : ReadyStepsFound(count)
    EvaluatingResults --> WaitingForDependencies : NoDependenciesReady
    EvaluatingResults --> BlockedByFailures : PermanentFailure(error)
    
    %% Waiting States
    WaitingForDependencies --> EvaluatingResults : DependenciesReady
    WaitingForRetry --> EnqueuingSteps : RetryReady
    
    %% Problem Resolution
    BlockedByFailures --> Error : GiveUp
    BlockedByFailures --> ResolvedManually : ManualResolution
    
    %% Cancellation (from any non-terminal state)
    Pending --> Cancelled : Cancel
    Initializing --> Cancelled : Cancel
    EnqueuingSteps --> Cancelled : Cancel
    StepsInProcess --> Cancelled : Cancel
    EvaluatingResults --> Cancelled : Cancel
    WaitingForDependencies --> Cancelled : Cancel
    WaitingForRetry --> Cancelled : Cancel
    BlockedByFailures --> Cancelled : Cancel
    
    %% Legacy Support
    Error --> Pending : Reset
    
    %% Terminal States
    Complete --> [*]
    Error --> [*]
    Cancelled --> [*]
    ResolvedManually --> [*]

Task Event System

Task state transitions are driven by events defined in tasker-shared/src/state_machine/events.rs:

Lifecycle Events

• Start: Begin task processing
• Cancel: Cancel task execution
• GiveUp: Abandon task (BlockedByFailures -> Error)
• ManualResolution: Manually resolve task

Discovery Events

• ReadyStepsFound(count): Ready steps discovered during initialization/evaluation
• NoStepsFound: No steps defined - task can complete immediately
• NoDependenciesReady: Dependencies not satisfied - wait required
• DependenciesReady: Dependencies now ready - can proceed

Processing Events

• StepsEnqueued(vec<Uuid>): Steps successfully queued for workers
• EnqueueFailed(error): Failed to enqueue steps
• StepCompleted(uuid): Individual step completed
• StepFailed(uuid): Individual step failed
• AllStepsCompleted: All current batch steps finished
• AllStepsSuccessful: All steps completed successfully

System Events

• PermanentFailure(error): Unrecoverable failure
• RetryReady: Retry timeout expired
• Timeout: Operation timeout occurred
• ProcessorCrashed: Processor became unavailable

Processor Ownership

The task state machine implements processor ownership for active states to prevent race conditions:

#![allow(unused)]
fn main() {
// Ownership validation in task_state_machine.rs
if target_state.requires_ownership() {
    let current_processor = self.get_current_processor().await?;
    TransitionGuard::check_ownership(target_state, current_processor, self.processor_uuid)?;
}
}

Ownership Rules:

• States requiring ownership: Initializing, EnqueuingSteps, StepsInProcess, EvaluatingResults
• Processor UUID stored in tasker.task_transitions.processor_uuid column
• Atomic ownership claiming prevents concurrent processing
• Ownership validated on each transition attempt

Workflow Step State Machine Architecture

Step State Definitions

The workflow step state machine implements 9 states for individual step execution:

Processing Pipeline States

• Pending: Initial state when step is created
• Enqueued: Queued for processing but not yet claimed by worker
• InProgress: Currently being executed by a worker
• EnqueuedForOrchestration: Worker completed, queued for orchestration processing
• EnqueuedAsErrorForOrchestration: Worker failed, queued for orchestration error processing

Waiting States

• WaitingForRetry: Step failed with retryable error, waiting for backoff period before retry

Terminal States

• Complete: Step completed successfully (after orchestration processing)
• Error: Step failed permanently (non-retryable or max retries exceeded)
• Cancelled: Step was cancelled
• ResolvedManually: Step was manually resolved by operator

State Machine Evolution

Previously, the Error state was used for both retryable and permanent failures. The introduction of WaitingForRetry created a semantic change:

• Before: Error = any failure (retryable or permanent)
• After: Error = permanent failure only, WaitingForRetry = retryable failure awaiting backoff

This change required updates to:

1. get_step_readiness_status() to recognize WaitingForRetry as a ready-eligible state
2. get_task_execution_context() to properly detect blocked vs recovering tasks
3. Error classification logic to distinguish permanent from retryable errors

Step State Properties

#![allow(unused)]
fn main() {
impl WorkflowStepState {
    pub fn is_terminal(&self) -> bool                    // No further transitions
    pub fn is_error(&self) -> bool                       // In error state (may allow retry)
    pub fn is_active(&self) -> bool                      // Being processed by worker
    pub fn is_in_processing_pipeline(&self) -> bool     // In execution pipeline
    pub fn is_ready_for_claiming(&self) -> bool         // Available for worker claim
    pub fn satisfies_dependencies(&self) -> bool        // Can satisfy other step dependencies
}
}

Step Lifecycle Flow

stateDiagram-v2
    [*] --> Pending

    %% Main Execution Path
    Pending --> Enqueued : Enqueue
    Enqueued --> InProgress : Start (worker claims)
    InProgress --> EnqueuedForOrchestration : EnqueueForOrchestration(success)
    EnqueuedForOrchestration --> Complete : Complete(results) [orchestration]

    %% Error Handling Path
    InProgress --> EnqueuedAsErrorForOrchestration : EnqueueForOrchestration(error)
    EnqueuedAsErrorForOrchestration --> WaitingForRetry : WaitForRetry(error) [retryable]
    EnqueuedAsErrorForOrchestration --> Error : Fail(error) [permanent/max retries]

    %% Retry Path
    WaitingForRetry --> Pending : Retry (after backoff)

    %% Legacy Direct Path (deprecated)
    InProgress --> Complete : Complete(results) [direct - legacy]
    InProgress --> Error : Fail(error) [direct - legacy]

    %% Legacy Backward Compatibility
    Pending --> InProgress : Start [legacy]

    %% Direct Failure Paths (error before worker processing)
    Pending --> Error : Fail(error)
    Enqueued --> Error : Fail(error)

    %% Cancellation Paths
    Pending --> Cancelled : Cancel
    Enqueued --> Cancelled : Cancel
    InProgress --> Cancelled : Cancel
    EnqueuedForOrchestration --> Cancelled : Cancel
    EnqueuedAsErrorForOrchestration --> Cancelled : Cancel
    WaitingForRetry --> Cancelled : Cancel
    Error --> Cancelled : Cancel

    %% Manual Resolution (from any state)
    Pending --> ResolvedManually : ResolveManually
    Enqueued --> ResolvedManually : ResolveManually
    InProgress --> ResolvedManually : ResolveManually
    EnqueuedForOrchestration --> ResolvedManually : ResolveManually
    EnqueuedAsErrorForOrchestration --> ResolvedManually : ResolveManually
    WaitingForRetry --> ResolvedManually : ResolveManually
    Error --> ResolvedManually : ResolveManually

    %% Terminal States
    Complete --> [*]
    Error --> [*]
    Cancelled --> [*]
    ResolvedManually --> [*]

Step Event System

Step transitions are driven by StepEvent types:

Processing Events

• Enqueue: Queue step for worker processing
• Start: Begin step execution (worker claims step)
• EnqueueForOrchestration(results): Worker completes, queues for orchestration
• Complete(results): Mark step complete (from orchestration or legacy direct)
• Fail(error): Mark step as permanently failed
• WaitForRetry(error): Mark step for retry after backoff

Control Events

• Cancel: Cancel step execution
• ResolveManually: Manual operator resolution
• Retry: Retry step from WaitingForRetry or Error state

Step Execution Flow Integration

The step state machine integrates tightly with the task state machine:

1. Task Discovers Ready Steps: TaskEvent::ReadyStepsFound(count) -> Task moves to EnqueuingSteps
2. Steps Get Enqueued: StepEvent::Enqueue -> Steps move to Enqueued state
3. Workers Claim Steps: StepEvent::Start -> Steps move to InProgress
4. Workers Complete Steps: StepEvent::EnqueueForOrchestration(results) -> Steps move to EnqueuedForOrchestration
5. Orchestration Processes Results: StepEvent::Complete(results) -> Steps move to Complete
6. Task Evaluates Progress: TaskEvent::StepCompleted(uuid) -> Task moves to EvaluatingResults
7. Task Completes or Continues: Based on remaining steps -> Task moves to Complete or back to EnqueuingSteps

Guard Conditions and Validation

Both state machines implement comprehensive guard conditions in tasker-shared/src/state_machine/guards.rs:

Task Guards

TransitionGuard

• Validates all task state transitions
• Prevents invalid state combinations
• Enforces terminal state immutability
• Supports legacy transition compatibility

Ownership Validation

• Checks processor ownership for ownership-required states
• Prevents concurrent task processing
• Allows ownership claiming for unowned tasks

Step Guards

StepDependenciesMetGuard

• Validates all step dependencies are satisfied
• Delegates to WorkflowStep::dependencies_met()
• Prevents premature step execution

StepNotInProgressGuard

• Ensures step is not already being processed
• Prevents duplicate worker claims
• Validates step availability

Retry Guards

• StepCanBeRetriedGuard: Validates step is in Error state
• Checks retry limits and conditions
• Prevents infinite retry loops

Orchestration Guards

• StepCanBeEnqueuedForOrchestrationGuard: Step must be InProgress
• StepCanBeCompletedFromOrchestrationGuard: Step must be EnqueuedForOrchestration
• StepCanBeFailedFromOrchestrationGuard: Step must be EnqueuedForOrchestration

Persistence Layer Architecture

Delegation Pattern

The persistence layer in tasker-shared/src/state_machine/persistence.rs implements a delegation pattern to the model layer:

#![allow(unused)]
fn main() {
// TaskTransitionPersistence -> TaskTransition::create() & TaskTransition::get_current()
// StepTransitionPersistence -> WorkflowStepTransition::create() & WorkflowStepTransition::get_current()
}

Benefits:

• No SQL duplication between state machine and models
• Atomic transaction handling in models
• Single source of truth for database operations
• Independent testability of model methods

Transition Storage

Task Transitions (tasker.task_transitions)

CREATE TABLE tasker.task_transitions (
  task_transition_uuid UUID PRIMARY KEY DEFAULT uuid_generate_v7(),
  task_uuid UUID NOT NULL,
  to_state VARCHAR NOT NULL,
  from_state VARCHAR,
  processor_uuid UUID,           -- Ownership tracking
  metadata JSONB,
  sort_key INTEGER NOT NULL,
  most_recent BOOLEAN DEFAULT false,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

Step Transitions (tasker.workflow_step_transitions)

CREATE TABLE tasker.workflow_step_transitions (
  workflow_step_transition_uuid UUID PRIMARY KEY DEFAULT uuid_generate_v7(),
  workflow_step_uuid UUID NOT NULL,
  to_state VARCHAR NOT NULL,
  from_state VARCHAR,
  metadata JSONB,
  sort_key INTEGER NOT NULL,
  most_recent BOOLEAN DEFAULT false,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

Current State Resolution

Both transition models implement efficient current state resolution:

#![allow(unused)]
fn main() {
// O(1) current state lookup using most_recent flag
TaskTransition::get_current(pool, task_uuid) -> Option<TaskTransition>
WorkflowStepTransition::get_current(pool, step_uuid) -> Option<WorkflowStepTransition>
}

Performance Optimization:

• most_recent = true flag on latest transition only
• Indexed queries: (task_uuid, most_recent) WHERE most_recent = true
• Atomic flag updates during transition creation

Atomic Transitions with Ownership

Atomic transitions with processor ownership:

#![allow(unused)]
fn main() {
impl TaskTransitionPersistence {
    pub async fn transition_with_ownership(
        &self,
        task_uuid: Uuid,
        from_state: TaskState,
        to_state: TaskState, 
        processor_uuid: Uuid,
        metadata: Option<Value>,
        pool: &PgPool,
    ) -> PersistenceResult<bool>
}
}

Atomicity Guarantees:

• Single database transaction for state change
• Processor UUID stored in dedicated column
• most_recent flag updated atomically
• Race condition prevention through database constraints

Action System

Both state machines execute actions after successful transitions:

Task Actions

1. PublishTransitionEventAction: Publishes task state change events
2. UpdateTaskCompletionAction: Updates task completion status
3. ErrorStateCleanupAction: Performs error state cleanup

Step Actions

1. PublishTransitionEventAction: Publishes step state change events
2. UpdateStepResultsAction: Updates step results and execution data
3. TriggerStepDiscoveryAction: Triggers task-level step discovery
4. ErrorStateCleanupAction: Performs step error cleanup

Actions execute sequentially after transition persistence, ensuring consistency.

State Machine Integration Points

Task <-> Step Coordination

1. Step Discovery: Task initialization discovers ready steps
2. Step Enqueueing: Task enqueues discovered steps to worker queues
3. Progress Monitoring: Task monitors step completion via events
4. Result Processing: Task processes step results and discovers next steps
5. Completion Detection: Task completes when all steps are complete

Event-Driven Communication

• pg_notify: PostgreSQL notifications for real-time coordination
• Event Publishers: Publish state transition events to event system
• Event Subscribers: React to state changes across system boundaries
• Queue Integration: Provider-agnostic message queues (PGMQ or RabbitMQ) for worker communication

Worker Integration

• Step Claiming: Workers claim Enqueued steps from queues
• Progress Updates: Workers transition steps to InProgress
• Result Submission: Workers submit results via EnqueueForOrchestration
• Orchestration Processing: Orchestration processes results and completes steps

This sophisticated state machine architecture provides the foundation for reliable, auditable, and scalable workflow orchestration in the tasker-core system.

Step Result Audit System

The step result audit system provides SOC2-compliant audit trails for workflow step execution results, enabling complete attribution tracking for compliance and debugging.

Audit Table Design

The tasker.workflow_step_result_audit table stores lightweight references with attribution data:

CREATE TABLE tasker.workflow_step_result_audit (
    workflow_step_result_audit_uuid UUID PRIMARY KEY DEFAULT uuid_generate_v7(),
    workflow_step_uuid UUID NOT NULL REFERENCES tasker.workflow_steps,
    workflow_step_transition_uuid UUID NOT NULL REFERENCES tasker.workflow_step_transitions,
    task_uuid UUID NOT NULL REFERENCES tasker.tasks,
    recorded_at TIMESTAMP NOT NULL DEFAULT NOW(),

    -- Attribution (NEW data not in transitions)
    worker_uuid UUID,
    correlation_id UUID,

    -- Extracted scalars for indexing/filtering
    success BOOLEAN NOT NULL,
    execution_time_ms BIGINT,

    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    UNIQUE (workflow_step_uuid, workflow_step_transition_uuid)
);

Design Principles

1. No Data Duplication: Full execution results already exist in tasker.workflow_step_transitions.metadata. The audit table stores references only.

2. Attribution Capture: The audit system captures NEW attribution data:

   • worker_uuid: Which worker instance processed the step
   • correlation_id: Distributed tracing identifier for request correlation

3. Indexed Scalars: Success and execution time are extracted for efficient filtering without JSON parsing.

4. SQL Trigger: A database trigger (trg_step_result_audit) guarantees audit record creation when workers persist results, ensuring SOC2 compliance.

Attribution Flow

Attribution data flows through the system via TransitionContext:

#![allow(unused)]
fn main() {
// Worker creates attribution context
let context = TransitionContext::with_worker(
    worker_uuid,
    Some(correlation_id),
);

// Context is merged into transition metadata
state_machine.transition_with_context(event, Some(context)).await?;

// SQL trigger extracts attribution from metadata
-- In trigger:
-- v_worker_uuid := (NEW.metadata->>'worker_uuid')::UUID;
-- v_correlation_id := (NEW.metadata->>'correlation_id')::UUID;
}

Trigger Behavior

The create_step_result_audit trigger fires on transitions to:

• enqueued_for_orchestration: Successful step completion
• enqueued_as_error_for_orchestration: Failed step completion

These states represent when workers persist execution results, creating the audit trail.

Querying Audit History

Via API

GET /v1/tasks/{task_uuid}/workflow_steps/{step_uuid}/audit

Returns audit records with full transition details via JOIN, ordered by recorded_at DESC.

Via Client

#![allow(unused)]
fn main() {
let audit_history = client.get_step_audit_history(task_uuid, step_uuid).await?;
for record in audit_history {
    println!("Worker: {:?}, Success: {}, Time: {:?}ms",
        record.worker_uuid,
        record.success,
        record.execution_time_ms
    );
}
}

Via Model

#![allow(unused)]
fn main() {
// Get audit history for a step with full transition details
let history = WorkflowStepResultAudit::get_audit_history(&pool, step_uuid).await?;

// Get all audit records for a task
let task_history = WorkflowStepResultAudit::get_task_audit_history(&pool, task_uuid).await?;

// Query by worker for attribution investigation
let worker_records = WorkflowStepResultAudit::get_by_worker(&pool, worker_uuid, Some(100)).await?;

// Query by correlation ID for distributed tracing
let correlated = WorkflowStepResultAudit::get_by_correlation_id(&pool, correlation_id).await?;
}

Indexes for Common Query Patterns

The audit table includes optimized indexes:

• idx_audit_step_uuid: Primary query - get audit history for a step
• idx_audit_task_uuid: Get all audit records for a task
• idx_audit_recorded_at: Time-range queries for SOC2 audit reports
• idx_audit_worker_uuid: Attribution investigation (partial index)
• idx_audit_correlation_id: Distributed tracing queries (partial index)
• idx_audit_success: Success/failure filtering

Historical Data

The migration includes a backfill for existing transitions. Historical records will have NULL attribution (worker_uuid, correlation_id) since that data wasn’t captured before the audit system was introduced.

Worker Actor-Based Architecture

Last Updated: 2025-12-04 Audience: Architects, Developers Status: Active Related Docs: Documentation Hub | Actor-Based Architecture | Events and Commands

<- Back to Documentation Hub

This document provides comprehensive documentation of the worker actor-based architecture in tasker-worker, covering the lightweight Actor pattern that mirrors the orchestration architecture for step execution and worker coordination.

Overview

The tasker-worker system implements a lightweight Actor pattern that mirrors the orchestration architecture, providing:

1. Actor Abstraction: Worker components encapsulated as actors with clear lifecycle hooks
2. Message-Based Communication: Type-safe message handling via Handler<M> trait
3. Central Registry: WorkerActorRegistry for managing all worker actors
4. Service Decomposition: Focused services following single responsibility principle
5. Lock-Free Statistics: AtomicU64 counters for hot-path performance
6. Direct Integration: Command processor routes to actors without wrapper layers

This architecture provides consistency between orchestration and worker systems, enabling clearer code organization and improved maintainability.

Implementation Status

Complete: All phases implemented and production-ready

• Phase 1: Core abstractions (traits, registry, lifecycle management)
• Phase 2: Service decomposition from 1575 LOC command_processor.rs
• Phase 3: All 5 primary actors implemented
• Phase 4: Command processor refactored to pure routing (~200 LOC)
• Phase 5: Stateless service design eliminating lock contention
• Cleanup: Lock-free AtomicU64 statistics, shared event system

Current State: Production-ready actor-based worker with 5 actors managing all step execution operations.

Core Concepts

What is a Worker Actor?

In the tasker-worker context, a Worker Actor is an encapsulated step execution component that:

• Manages its own state: Each actor owns its dependencies and configuration
• Processes messages: Responds to typed command messages via the Handler<M> trait
• Has lifecycle hooks: Initialization (started) and cleanup (stopped) methods
• Is isolated: Actors communicate through message passing
• Is thread-safe: All actors are Send + Sync + 'static

Why Actors for Workers?

The previous architecture had a monolithic command processor:

#![allow(unused)]
fn main() {
// OLD: 1575 LOC monolithic command processor
pub struct WorkerProcessor {
    // All logic mixed together
    // RwLock contention on hot path
    // Two-phase initialization complexity
}
}

The actor pattern provides:

#![allow(unused)]
fn main() {
// NEW: Pure routing command processor (~200 LOC)
impl ActorCommandProcessor {
    async fn handle_command(&self, command: WorkerCommand) -> bool {
        match command {
            WorkerCommand::ExecuteStep { message, queue_name, resp } => {
                let msg = ExecuteStepMessage { message, queue_name };
                let result = self.actors.step_executor_actor.handle(msg).await;
                let _ = resp.send(result);
                true
            }
            // ... pure routing, no business logic
        }
    }
}
}

Actor vs Service

Services (underlying business logic):

• Encapsulate step execution logic
• Stateless operations on step data
• Direct method invocation
• Examples: StepExecutorService, FFICompletionService, WorkerStatusService

Actors (message-based coordination):

• Wrap services with message-based interface
• Manage service lifecycle
• Asynchronous message handling
• Examples: StepExecutorActor, FFICompletionActor, WorkerStatusActor

The relationship:

#![allow(unused)]
fn main() {
pub struct StepExecutorActor {
    context: Arc<SystemContext>,
    service: Arc<StepExecutorService>,  // Wraps underlying service
}

#[async_trait]
impl Handler<ExecuteStepMessage> for StepExecutorActor {
    async fn handle(&self, msg: ExecuteStepMessage) -> TaskerResult<bool> {
        // Delegates to stateless service
        self.service.execute_step(msg.message, &msg.queue_name).await
    }
}
}

Worker Actor Traits

WorkerActor Trait

The base trait for all worker actors, defined in tasker-worker/src/worker/actors/traits.rs:

#![allow(unused)]
fn main() {
/// Base trait for all worker actors
///
/// Provides lifecycle management and context access for all actors in the
/// worker system. All actors must implement this trait to participate
/// in the actor registry and lifecycle management.
pub trait WorkerActor: Send + Sync + 'static {
    /// Returns the unique name of this actor
    fn name(&self) -> &'static str;

    /// Returns a reference to the system context
    fn context(&self) -> &Arc<SystemContext>;

    /// Called when the actor is started
    fn started(&mut self) -> TaskerResult<()> {
        tracing::info!(actor = %self.name(), "Actor started");
        Ok(())
    }

    /// Called when the actor is stopped
    fn stopped(&mut self) -> TaskerResult<()> {
        tracing::info!(actor = %self.name(), "Actor stopped");
        Ok(())
    }
}
}

Handler Trait

The message handling trait, enabling type-safe message processing:

#![allow(unused)]
fn main() {
/// Message handler trait for specific message types
#[async_trait]
pub trait Handler<M: Message>: WorkerActor {
    /// Handle a message asynchronously
    async fn handle(&self, msg: M) -> TaskerResult<M::Response>;
}
}

Message Trait

The marker trait for command messages:

#![allow(unused)]
fn main() {
/// Marker trait for command messages
pub trait Message: Send + 'static {
    /// The response type for this message
    type Response: Send;
}
}

WorkerActorRegistry

The central registry managing all worker actors, defined in tasker-worker/src/worker/actors/registry.rs:

Structure

#![allow(unused)]
fn main() {
/// Registry managing all worker actors
#[derive(Clone)]
pub struct WorkerActorRegistry {
    /// System context shared by all actors
    context: Arc<SystemContext>,

    /// Worker ID for this registry
    worker_id: String,

    /// Step executor actor for step execution    pub step_executor_actor: Arc<StepExecutorActor>,

    /// FFI completion actor for handling step completions    pub ffi_completion_actor: Arc<FFICompletionActor>,

    /// Template cache actor for template management    pub template_cache_actor: Arc<TemplateCacheActor>,

    /// Domain event actor for event dispatching    pub domain_event_actor: Arc<DomainEventActor>,

    /// Worker status actor for health and status    pub worker_status_actor: Arc<WorkerStatusActor>,
}
}

Initialization

All dependencies required at construction time (no two-phase initialization):

#![allow(unused)]
fn main() {
impl WorkerActorRegistry {
    pub async fn build(
        context: Arc<SystemContext>,
        worker_id: String,
        task_template_manager: Arc<TaskTemplateManager>,
        event_publisher: WorkerEventPublisher,
        domain_event_handle: DomainEventSystemHandle,
    ) -> TaskerResult<Self> {
        // Create actors with all dependencies upfront
        let mut step_executor_actor = StepExecutorActor::new(
            context.clone(),
            worker_id.clone(),
            task_template_manager.clone(),
            event_publisher,
            domain_event_handle,
        );

        // Call started() lifecycle hook
        step_executor_actor.started()?;

        // ... create other actors ...

        Ok(Self {
            context,
            worker_id,
            step_executor_actor: Arc::new(step_executor_actor),
            // ...
        })
    }
}
}

Implemented Actors

StepExecutorActor

Handles step execution from PGMQ messages and events.

Location: tasker-worker/src/worker/actors/step_executor_actor.rs

Messages:

• ExecuteStepMessage - Execute step from raw data
• ExecuteStepWithCorrelationMessage - Execute with FFI correlation
• ExecuteStepFromPgmqMessage - Execute from PGMQ message
• ExecuteStepFromEventMessage - Execute from event notification

Delegation: Wraps StepExecutorService (stateless, no locks)

Purpose: Central coordinator for all step execution, handles claiming, handler invocation, and result construction.

FFICompletionActor

Handles step completion results from FFI handlers.

Location: tasker-worker/src/worker/actors/ffi_completion_actor.rs

Messages:

• SendStepResultMessage - Send result to orchestration
• ProcessStepCompletionMessage - Process completion with correlation

Delegation: Wraps FFICompletionService

Purpose: Forwards step execution results to orchestration queue, manages correlation for async FFI handlers.

TemplateCacheActor

Manages task template caching and refresh.

Location: tasker-worker/src/worker/actors/template_cache_actor.rs

Messages:

• RefreshTemplateCacheMessage - Refresh cache for namespace

Delegation: Wraps TaskTemplateManager

Purpose: Maintains handler template cache for efficient step execution.

DomainEventActor

Dispatches domain events after step completion.

Location: tasker-worker/src/worker/actors/domain_event_actor.rs

Messages:

• DispatchDomainEventsMessage - Dispatch events for completed step

Delegation: Wraps DomainEventSystemHandle

Purpose: Fire-and-forget domain event dispatch (never blocks step completion).

WorkerStatusActor

Provides worker health and status reporting.

Location: tasker-worker/src/worker/actors/worker_status_actor.rs

Messages:

• GetWorkerStatusMessage - Get current worker status
• HealthCheckMessage - Perform health check
• GetEventStatusMessage - Get event integration status
• SetEventIntegrationMessage - Enable/disable event integration

Features:

• Lock-free statistics via AtomicStepExecutionStats
• AtomicU64 counters for total_executed, total_succeeded, total_failed
• Average execution time computed on read from sum / count

Purpose: Real-time health monitoring and statistics without lock contention.

Lock-Free Statistics

The WorkerStatusActor uses atomic counters for lock-free statistics on the hot path:

#![allow(unused)]
fn main() {
/// Lock-free step execution statistics using atomic counters
#[derive(Debug)]
pub struct AtomicStepExecutionStats {
    total_executed: AtomicU64,
    total_succeeded: AtomicU64,
    total_failed: AtomicU64,
    total_execution_time_ms: AtomicU64,
}

impl AtomicStepExecutionStats {
    /// Record a successful step execution (lock-free)
    #[inline]
    pub fn record_success(&self, execution_time_ms: u64) {
        self.total_executed.fetch_add(1, Ordering::Relaxed);
        self.total_succeeded.fetch_add(1, Ordering::Relaxed);
        self.total_execution_time_ms.fetch_add(execution_time_ms, Ordering::Relaxed);
    }

    /// Record a failed step execution (lock-free)
    #[inline]
    pub fn record_failure(&self) {
        self.total_executed.fetch_add(1, Ordering::Relaxed);
        self.total_failed.fetch_add(1, Ordering::Relaxed);
    }

    /// Get a snapshot of current statistics
    pub fn snapshot(&self) -> StepExecutionStats {
        let total_executed = self.total_executed.load(Ordering::Relaxed);
        let total_time = self.total_execution_time_ms.load(Ordering::Relaxed);
        let average_execution_time_ms = if total_executed > 0 {
            total_time as f64 / total_executed as f64
        } else {
            0.0
        };
        StepExecutionStats {
            total_executed,
            total_succeeded: self.total_succeeded.load(Ordering::Relaxed),
            total_failed: self.total_failed.load(Ordering::Relaxed),
            average_execution_time_ms,
        }
    }
}
}

Benefits:

• Zero lock contention on step completion (every step calls record_success or record_failure)
• Sub-microsecond overhead per operation
• Consistent averages computed from totals

Integration with Commands

ActorCommandProcessor

The command processor provides pure routing to actors:

#![allow(unused)]
fn main() {
impl ActorCommandProcessor {
    async fn handle_command(&self, command: WorkerCommand) -> bool {
        match command {
            // Step Execution Commands -> StepExecutorActor
            WorkerCommand::ExecuteStep { message, queue_name, resp } => {
                let msg = ExecuteStepMessage { message, queue_name };
                let result = self.actors.step_executor_actor.handle(msg).await;
                let _ = resp.send(result);
                true
            }

            // Completion Commands -> FFICompletionActor
            WorkerCommand::SendStepResult { result, resp } => {
                let msg = SendStepResultMessage { result };
                let send_result = self.actors.ffi_completion_actor.handle(msg).await;
                let _ = resp.send(send_result);
                true
            }

            // Status Commands -> WorkerStatusActor
            WorkerCommand::HealthCheck { resp } => {
                let result = self.actors.worker_status_actor.handle(HealthCheckMessage).await;
                let _ = resp.send(result);
                true
            }

            WorkerCommand::Shutdown { resp } => {
                let _ = resp.send(Ok(()));
                false  // Exit command loop
            }
        }
    }
}
}

FFI Completion Flow

Domain events are dispatched after successful orchestration notification:

#![allow(unused)]
fn main() {
async fn handle_ffi_completion(&self, step_result: StepExecutionResult) {
    // Record stats (lock-free)
    if step_result.success {
        self.actors.worker_status_actor
            .record_success(step_result.metadata.execution_time_ms as f64).await;
    } else {
        self.actors.worker_status_actor.record_failure().await;
    }

    // Send to orchestration FIRST
    let msg = SendStepResultMessage { result: step_result.clone() };
    match self.actors.ffi_completion_actor.handle(msg).await {
        Ok(()) => {
            // Domain events dispatched AFTER successful orchestration notification
            // Fire-and-forget - never blocks the worker
            self.actors.step_executor_actor
                .dispatch_domain_events(step_result.step_uuid, &step_result, None).await;
        }
        Err(e) => {
            // Don't dispatch domain events - orchestration wasn't notified
            tracing::error!("Failed to forward step completion to orchestration");
        }
    }
}
}

Service Decomposition

Large services were decomposed from the monolithic command processor:

StepExecutorService

services/step_execution/
├── mod.rs                  # Public API
├── service.rs              # StepExecutorService (~250 lines)
├── step_claimer.rs         # Step claiming logic
├── handler_invoker.rs      # Handler invocation
└── result_builder.rs       # Result construction

Key Design: Completely stateless service using &self methods. Wrapped in Arc<StepExecutorService> without any locks.

FFICompletionService

services/ffi_completion/
├── mod.rs                  # Public API
├── service.rs              # FFICompletionService
└── result_sender.rs        # Orchestration result sender

WorkerStatusService

services/worker_status/
├── mod.rs                  # Public API
└── service.rs              # WorkerStatusService

Key Architectural Decisions

1. Stateless Services

Services use &self methods with no mutable state:

#![allow(unused)]
fn main() {
impl StepExecutorService {
    pub async fn execute_step(
        &self,  // Immutable reference
        message: PgmqMessage<SimpleStepMessage>,
        queue_name: &str,
    ) -> TaskerResult<bool> {
        // Stateless execution - no mutable state
    }
}
}

Benefits:

• Zero lock contention
• Maximum concurrency per worker
• Simplified reasoning about state

2. Constructor-Based Dependency Injection

All dependencies required at construction time:

#![allow(unused)]
fn main() {
pub async fn new(
    context: Arc<SystemContext>,
    worker_id: String,
    task_template_manager: Arc<TaskTemplateManager>,
    event_publisher: WorkerEventPublisher,        // Required
    domain_event_handle: DomainEventSystemHandle, // Required
) -> TaskerResult<Self>
}

Benefits:

• Compiler enforces complete initialization
• No “partially initialized” states
• Clear dependency graph

3. Shared Event System

Event publisher and subscriber share the same WorkerEventSystem:

#![allow(unused)]
fn main() {
let shared_event_system = event_system
    .unwrap_or_else(|| Arc::new(WorkerEventSystem::new()));
let event_publisher =
    WorkerEventPublisher::with_event_system(worker_id.clone(), shared_event_system.clone());

// Enable subscriber with same shared system
processor.enable_event_subscriber(Some(shared_event_system)).await;
}

Benefits:

• FFI handlers reliably receive step execution events
• No isolated event systems causing silent failures

4. Graceful Degradation

Domain events never fail step completion:

#![allow(unused)]
fn main() {
// dispatch_domain_events returns () not TaskerResult<()>
// Errors logged but never propagated
pub async fn dispatch_domain_events(
    &self,
    step_uuid: Uuid,
    result: &StepExecutionResult,
    metadata: Option<HashMap<String, serde_json::Value>>,
) {
    // Fire-and-forget with error logging
    // Channel full? Log and continue
    // Dispatch error? Log and continue
}
}

Comparison with Orchestration Actors

Benefits

1. Consistency with Orchestration

Same patterns and traits as orchestration actors:

• Identical Handler<M> trait interface
• Similar registry lifecycle management
• Consistent message-based communication

2. Zero Lock Contention

• Stateless services eliminate RwLock on hot path
• AtomicU64 counters for statistics
• Maximum concurrent step execution

3. Type Safety

Messages and responses checked at compile time:

#![allow(unused)]
fn main() {
// Compile error if types don't match
impl Handler<ExecuteStepMessage> for StepExecutorActor {
    async fn handle(&self, msg: ExecuteStepMessage) -> TaskerResult<bool> {
        // Must return bool, not something else
    }
}
}

4. Testability

• Clear message boundaries for mocking
• Isolated actor lifecycle for unit tests
• 119 unit tests, 73 E2E tests passing

5. Maintainability

• 1575 LOC -> ~200 LOC command processor
• Focused services (<300 lines per file)
• Clear separation of concerns

Detailed Analysis

For design rationale, see the Worker Decomposition ADR.

Summary

The worker actor-based architecture provides a consistent, type-safe foundation for step execution in tasker-worker. Key takeaways:

1. Mirrors Orchestration: Same patterns as orchestration actors for consistency
2. Lock-Free Performance: Stateless services and AtomicU64 counters
3. Type Safety: Compile-time verification of message contracts
4. Pure Routing: Command processor delegates without business logic
5. Graceful Degradation: Domain events never fail step completion
6. Production Ready: 119 unit tests, 73 E2E tests, full regression coverage

The architecture provides a solid foundation for high-throughput step execution while maintaining the proven reliability of the orchestration system.

<- Back to Documentation Hub

Worker Event Systems Architecture

Last Updated: 2026-01-15 Audience: Architects, Developers Status: Active Related Docs: Worker Actors | Events and Commands | Messaging Abstraction

<- Back to Documentation Hub

This document provides comprehensive documentation of the worker event system architecture in tasker-worker, covering the dual-channel event pattern, domain event publishing, and FFI integration.

Overview

The worker event system implements a dual-channel architecture for non-blocking step execution:

1. WorkerEventSystem: Receives step execution events via provider-agnostic subscriptions
2. HandlerDispatchService: Fire-and-forget handler invocation with bounded concurrency
3. CompletionProcessorService: Routes results back to orchestration
4. DomainEventSystem: Fire-and-forget domain event publishing

Messaging Backend Support: The worker event system supports multiple messaging backends (PGMQ, RabbitMQ) through a provider-agnostic abstraction. See Messaging Abstraction for details.

This architecture enables true parallel handler execution while maintaining strict ordering guarantees for domain events.

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────┐
│                           WORKER EVENT FLOW                                  │
└─────────────────────────────────────────────────────────────────────────────┘

                    MessagingProvider (PGMQ or RabbitMQ)
                                  │
                                  │ provider.subscribe_many()
                                  ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         WorkerEventSystem                                    │
│  ┌──────────────────────┐    ┌──────────────────────┐                       │
│  │  WorkerQueueListener │    │  WorkerFallbackPoller │                      │
│  │  (provider-agnostic) │    │  (PGMQ only)          │                      │
│  └──────────┬───────────┘    └──────────┬───────────┘                       │
│             │                           │                                    │
│             └───────────┬───────────────┘                                    │
│                         │                                                    │
│                         ▼                                                    │
│   MessageNotification::Message → ExecuteStepFromMessage (RabbitMQ)          │
│   MessageNotification::Available → ExecuteStepFromEvent (PGMQ)              │
└─────────────────────────────────────────────────────────────────────────────┘
                                  │
                                  ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                      ActorCommandProcessor                                   │
│                              │                                               │
│                              ▼                                               │
│                      StepExecutorActor                                       │
│                              │                                               │
│                              │ claim step, send to dispatch channel          │
│                              ▼                                               │
└─────────────────────────────────────────────────────────────────────────────┘
                                  │
                    ┌─────────────┴─────────────┐
                    │                           │
           Rust Workers               FFI Workers (Ruby/Python)
                    │                           │
                    ▼                           ▼
┌───────────────────────────────┐   ┌───────────────────────────────┐
│   HandlerDispatchService      │   │     FfiDispatchChannel        │
│                               │   │                               │
│   dispatch_receiver           │   │   pending_events HashMap      │
│         │                     │   │         │                     │
│         ▼                     │   │         ▼                     │
│   [Semaphore] N permits       │   │   poll_step_events()          │
│         │                     │   │         │                     │
│         ▼                     │   │         ▼                     │
│   handler.call()              │   │   Ruby/Python handler         │
│         │                     │   │         │                     │
│         ▼                     │   │         ▼                     │
│   PostHandlerCallback         │   │   complete_step_event()       │
│         │                     │   │         │                     │
│         ▼                     │   │         ▼                     │
│   completion_sender           │   │   PostHandlerCallback         │
│                               │   │         │                     │
└───────────────┬───────────────┘   │         ▼                     │
                │                   │   completion_sender           │
                │                   │                               │
                │                   └───────────────┬───────────────┘
                │                                   │
                └───────────────┬───────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                    CompletionProcessorService                                │
│                              │                                               │
│                              ▼                                               │
│                    FFICompletionService                                      │
│                              │                                               │
│                              ▼                                               │
│               orchestration_step_results queue                               │
└─────────────────────────────────────────────────────────────────────────────┘
                                │
                                ▼
                         Orchestration

Core Components

1. WorkerEventSystem

Location: tasker-worker/src/worker/event_systems/worker_event_system.rs

Implements the EventDrivenSystem trait for worker namespace queue processing. Supports three deployment modes with provider-agnostic message handling:

Provider-Specific Behavior:

• PGMQ: Uses MessageNotification::Available (signal-only), requires fallback polling
• RabbitMQ: Uses MessageNotification::Message (full payload), no fallback needed

Key Features:

• Unified configuration via WorkerEventSystemConfig
• Atomic statistics with AtomicU64 counters
• Converts WorkerNotification to WorkerCommand for processing

#![allow(unused)]
fn main() {
// Worker notification to command conversion (provider-agnostic)
match notification {
    // RabbitMQ style - full message delivered
    WorkerNotification::Message(msg) => {
        command_sender.send(WorkerCommand::ExecuteStepFromMessage {
            queue_name: msg.queue_name.clone(),
            message: msg,
            resp: resp_tx,
        }).await;
    }
    // PGMQ style - signal-only, requires fetch
    WorkerNotification::Event(WorkerQueueEvent::StepMessage(msg_event)) => {
        command_sender.send(WorkerCommand::ExecuteStepFromEvent {
            message_event: msg_event,
            resp: resp_tx,
        }).await;
    }
    // ...
}
}

2. HandlerDispatchService

Location: tasker-worker/src/worker/handlers/dispatch_service.rs

Non-blocking handler dispatch with bounded parallelism.

Architecture:

dispatch_receiver → [Semaphore] → handler.call() → [callback] → completion_sender
                         │                              │
                         └─→ Bounded to N concurrent    └─→ Domain events
                              tasks

Key Design Decisions:

1. Semaphore-Bounded Concurrency: Limits concurrent handlers to prevent resource exhaustion
2. Permit Release Before Send: Prevents backpressure cascade
3. Post-Handler Callback: Domain events fire only after result is committed

#![allow(unused)]
fn main() {
tokio::spawn(async move {
    let permit = semaphore.acquire().await?;

    let result = execute_with_timeout(&registry, &msg, timeout).await;

    // Release permit BEFORE sending to completion channel
    drop(permit);

    // Send result FIRST
    sender.send(result.clone()).await?;

    // Callback fires AFTER result is committed
    if let Some(cb) = callback {
        cb.on_handler_complete(&step, &result, &worker_id).await;
    }
});
}

Error Handling:

Handler Resolution

Before handler execution, the dispatch service resolves the handler using a resolver chain pattern:

HandlerDefinition                    ResolverChain                    Handler
     │                                    │                              │
     │  callable: "process_payment"       │                              │
     │  method: "refund"                  │                              │
     │  resolver: null                    │                              │
     │                                    │                              │
     ├───────────────────────────────────►│                              │
     │                                    │                              │
     │                    ┌───────────────┴───────────────┐              │
     │                    │ ExplicitMappingResolver (10)  │              │
     │                    │ can_resolve? ─► YES           │              │
     │                    │ resolve() ─────────────────────────────────►│
     │                    └───────────────────────────────┘              │
     │                                                                   │
     │                    ┌───────────────────────────────┐              │
     │                    │ MethodDispatchWrapper         │              │
     │                    │ (if method != "call")         │◄─────────────┤
     │                    └───────────────────────────────┘              │

Built-in Resolvers:

Method Dispatch: When handler.method is specified and not "call", a MethodDispatchWrapper is applied to invoke the specified method instead of the default call() method.

See Handler Resolution Guide for complete documentation.

3. FfiDispatchChannel

Location: tasker-worker/src/worker/handlers/ffi_dispatch_channel.rs

Pull-based polling interface for FFI workers (Ruby, Python). Enables language-specific handlers without complex FFI memory management.

Flow:

Rust                           Ruby/Python
  │                                 │
  │  dispatch(step)                 │
  │ ──────────────────────────────► │
  │                                 │ pending_events.insert()
  │                                 │
  │  poll_step_events()             │
  │ ◄────────────────────────────── │
  │                                 │
  │                                 │ handler.call()
  │                                 │
  │  complete_step_event(result)    │
  │ ◄────────────────────────────── │
  │                                 │
  │  PostHandlerCallback            │
  │  completion_sender.send()       │
  │                                 │

Key Features:

• Thread-safe pending events map with lock poisoning recovery
• Configurable completion timeout (default 30s)
• Starvation detection and warnings
• Fire-and-forget callbacks via runtime_handle.spawn()

4. CompletionProcessorService

Location: tasker-worker/src/worker/handlers/completion_processor.rs

Receives completed step results and routes to orchestration queue via FFICompletionService.

completion_receiver → CompletionProcessorService → FFICompletionService → orchestration_step_results

Note: Currently processes completions sequentially. Parallel processing is planned as a future enhancement.

5. DomainEventSystem

Location: tasker-worker/src/worker/event_systems/domain_event_system.rs

Async system for fire-and-forget domain event publishing.

Architecture:

command_processor.rs                  DomainEventSystem
      │                                     │
      │ try_send(command)                   │ spawn process_loop()
      ▼                                     ▼
mpsc::Sender<DomainEventCommand>  →  mpsc::Receiver
                                            │
                                            ▼
                                    EventRouter → PGMQ / InProcess

Key Design:

• try_send() never blocks - if channel is full, events are dropped with metrics
• Background task processes commands asynchronously
• Graceful shutdown drains fast events up to configurable timeout
• Three delivery modes: Durable (PGMQ), Fast (in-process), Broadcast

Shared Event Abstractions

EventDrivenSystem Trait

Location: tasker-shared/src/event_system/event_driven.rs

Unified trait for all event-driven systems:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait EventDrivenSystem: Send + Sync {
    type SystemId: Send + Sync + Clone;
    type Event: Send + Sync + Clone;
    type Config: Send + Sync + Clone;
    type Statistics: EventSystemStatistics;

    fn system_id(&self) -> Self::SystemId;
    fn deployment_mode(&self) -> DeploymentMode;
    fn is_running(&self) -> bool;

    async fn start(&mut self) -> Result<(), DeploymentModeError>;
    async fn stop(&mut self) -> Result<(), DeploymentModeError>;
    async fn process_event(&self, event: Self::Event) -> Result<(), DeploymentModeError>;
    async fn health_check(&self) -> Result<DeploymentModeHealthStatus, DeploymentModeError>;

    fn statistics(&self) -> Self::Statistics;
    fn config(&self) -> &Self::Config;
}
}

Deployment Modes

Location: tasker-shared/src/event_system/deployment.rs

#![allow(unused)]
fn main() {
pub enum DeploymentMode {
    PollingOnly,      // Traditional polling, no events
    EventDrivenOnly,  // Pure event-driven, no polling
    Hybrid,           // Event-driven with polling fallback
}
}

PostHandlerCallback Trait

Location: tasker-worker/src/worker/handlers/dispatch_service.rs

Extensibility point for post-handler actions:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait PostHandlerCallback: Send + Sync + 'static {
    /// Called after a handler completes
    async fn on_handler_complete(
        &self,
        step: &TaskSequenceStep,
        result: &StepExecutionResult,
        worker_id: &str,
    );

    /// Name of this callback for logging purposes
    fn name(&self) -> &str;
}
}

Implementations:

• NoOpCallback: Default no-operation callback
• DomainEventCallback: Publishes domain events to DomainEventSystem

Configuration

Worker Event System

# config/tasker/base/event_systems.toml
[event_systems.worker]
system_id = "worker-event-system"
deployment_mode = "Hybrid"

[event_systems.worker.metadata.listener]
retry_interval_seconds = 5
max_retry_attempts = 3
event_timeout_seconds = 60
batch_processing = true
connection_timeout_seconds = 30

[event_systems.worker.metadata.fallback_poller]
enabled = true
polling_interval_ms = 100
batch_size = 10
age_threshold_seconds = 30
max_age_hours = 24
visibility_timeout_seconds = 60

Handler Dispatch

# config/tasker/base/worker.toml
[worker.mpsc_channels.handler_dispatch]
dispatch_buffer_size = 1000
completion_buffer_size = 1000
max_concurrent_handlers = 10
handler_timeout_ms = 30000

[worker.mpsc_channels.ffi_dispatch]
dispatch_buffer_size = 1000
completion_timeout_ms = 30000
starvation_warning_threshold_ms = 10000
callback_timeout_ms = 5000
completion_send_timeout_ms = 10000

Integration with Worker Actors

The event systems integrate with the worker actor architecture:

WorkerEventSystem
       │
       ▼
ActorCommandProcessor
       │
       ├──► StepExecutorActor ──► dispatch_sender
       │
       ├──► FFICompletionActor ◄── completion_receiver
       │
       └──► DomainEventActor ◄── PostHandlerCallback

See Worker Actors Documentation for actor details.

Event Flow Guarantees

Ordering Guarantee

Domain events fire AFTER result is committed to completion channel:

handler.call()
    → result committed to completion_sender
    → PostHandlerCallback.on_handler_complete()
    → domain events dispatched

This eliminates race conditions where downstream systems see events before orchestration processes results.

Idempotency Guarantee

State machine guards prevent duplicate execution:

1. Step claimed atomically via transition_step_state_atomic()
2. State guards reject duplicate claims
3. Results are deduplicated by completion channel

Fire-and-Forget Guarantee

Domain event failures never fail step completion:

#![allow(unused)]
fn main() {
// DomainEventCallback
pub async fn on_handler_complete(&self, step, result, worker_id) {
    // dispatch_events uses try_send() - never blocks
    // If channel full, events dropped with metrics
    // Step completion is NOT affected
    self.handle.dispatch_events(events, publisher_name, correlation_id);
}
}

Monitoring

Key Metrics

Health Checks

#![allow(unused)]
fn main() {
async fn health_check(&self) -> Result<DeploymentModeHealthStatus, DeploymentModeError> {
    if self.is_running.load(Ordering::Acquire) {
        Ok(DeploymentModeHealthStatus::Healthy)
    } else {
        Ok(DeploymentModeHealthStatus::Critical)
    }
}
}

Backpressure Handling

The worker event system implements multiple backpressure mechanisms to ensure graceful degradation under load while preserving step idempotency.

Backpressure Points

┌─────────────────────────────────────────────────────────────────────────────┐
│                      WORKER BACKPRESSURE FLOW                                │
└─────────────────────────────────────────────────────────────────────────────┘

[1] Step Claiming
    │
    ├── Planned: Capacity check before claiming
    │   └── If at capacity: Leave message in queue (visibility timeout)
    │
    ▼
[2] Handler Dispatch Channel (Bounded)
    │
    ├── dispatch_buffer_size = 1000
    │   └── If full: Sender blocks until space available
    │
    ▼
[3] Semaphore-Bounded Execution
    │
    ├── max_concurrent_handlers = 10
    │   └── If permits exhausted: Task waits for permit
    │
    ├── CRITICAL: Permit released BEFORE sending to completion channel
    │   └── Prevents backpressure cascade
    │
    ▼
[4] Completion Channel (Bounded)
    │
    ├── completion_buffer_size = 1000
    │   └── If full: Handler task blocks until space available
    │
    ▼
[5] Domain Events (Fire-and-Forget)
    │
    └── try_send() semantics
        └── If channel full: Events DROPPED (step execution unaffected)

Handler Dispatch Backpressure

The HandlerDispatchService uses semaphore-bounded parallelism:

#![allow(unused)]
fn main() {
// Permit acquisition blocks if all permits in use
let permit = semaphore.acquire().await?;

let result = execute_with_timeout(&registry, &msg, timeout).await;

// CRITICAL: Release permit BEFORE sending to completion channel
// This prevents backpressure cascade where full completion channel
// holds permits, starving new handler execution
drop(permit);

// Now send to completion channel (may block if full)
sender.send(result).await?;
}

Why permit release before send matters:

• If completion channel is full, handler task blocks on send
• If permit is held during block, no new handlers can start
• By releasing permit first, new handlers can start even if completions are backing up

FFI Dispatch Backpressure

The FfiDispatchChannel handles backpressure for Ruby/Python workers:

Starvation Detection:

[worker.mpsc_channels.ffi_dispatch]
starvation_warning_threshold_ms = 10000  # Warn if event waits > 10s

Domain Event Drop Semantics

Domain events use try_send() and are explicitly designed to be droppable:

#![allow(unused)]
fn main() {
// Domain events fire AFTER result is committed
// They are non-critical and use fire-and-forget semantics
match event_sender.try_send(event) {
    Ok(()) => { /* Event dispatched */ }
    Err(TrySendError::Full(_)) => {
        // Event dropped - step execution NOT affected
        warn!("Domain event dropped: channel full");
        metrics.increment("domain_events_dropped");
    }
}
}

Why this is safe: Domain events are informational. Dropping them does not affect step execution correctness. The step result is already committed to the completion channel before domain events fire.

Step Claiming Backpressure (Planned)

Future enhancement: Workers will check capacity before claiming steps:

#![allow(unused)]
fn main() {
// Planned implementation
fn should_claim_step(&self) -> bool {
    let available = self.semaphore.available_permits();
    let threshold = self.config.claim_capacity_threshold;  // e.g., 0.8
    let max = self.config.max_concurrent_handlers;

    available as f64 / max as f64 > (1.0 - threshold)
}
}

If at capacity:

• Worker does NOT acknowledge the PGMQ message
• Message returns to queue after visibility timeout
• Another worker (or same worker later) claims it

Idempotency Under Backpressure

All backpressure mechanisms preserve step idempotency:

Critical Rule: A claimed step MUST produce a result (success or failure). Backpressure may delay but never drop step execution.

For comprehensive backpressure strategy, see Backpressure Architecture.

Best Practices

1. Choose Deployment Mode

• Production: Use Hybrid for reliability with event-driven performance
• Development: Use EventDrivenOnly for fastest iteration
• Restricted environments: Use PollingOnly when pg_notify unavailable

2. Tune Concurrency

[worker.mpsc_channels.handler_dispatch]
max_concurrent_handlers = 10  # Start here, increase based on monitoring

Monitor:

• Semaphore wait times
• Handler execution latency
• Completion channel saturation

3. Configure Timeouts

handler_timeout_ms = 30000        # Match your slowest handler
completion_timeout_ms = 30000     # FFI completion timeout
callback_timeout_ms = 5000        # Domain event callback timeout

4. Monitor Starvation

For FFI workers, monitor pending event age:

# Ruby
metrics = Tasker.ffi_dispatch_metrics
if metrics[:oldest_pending_age_ms] > 10000
  warn "FFI polling falling behind"
end

Related Documentation

• Messaging Abstraction - Provider-agnostic messaging
• Backpressure Architecture - Unified backpressure strategy
• Worker Actor-Based Architecture - Actor pattern implementation
• Events and Commands - Command pattern details
• Dual-Channel Event System ADR - Dual-channel event system decision
• FFI Callback Safety - FFI guidelines
• RCA: Parallel Execution Timing Bugs - Lessons learned
• Backpressure Monitoring Runbook - Metrics and alerting

<- Back to Documentation Hub

Tasker Core Guides

This directory contains practical how-to guides for working with Tasker Core.

Documents

When to Read These

• Getting started: Begin with Quick Start
• Implementing features: Check Use Cases and Patterns
• Handling errors: See Retry Semantics and DLQ System
• Processing data: Review Batch Processing
• Deploying: Consult Configuration Management

Related Documentation

• Architecture - The “what” - system structure
• Principles - The “why” - design philosophy
• Workers - Language-specific handler development

API Security Guide

API-level security for orchestration (8080) and worker (8081) endpoints using JWT bearer tokens and API key authentication with permission-based access control.

Security is disabled by default for backward compatibility. Enable it explicitly in configuration.

See also: Auth Documentation Hub for architecture overview, Permissions for route mapping, Configuration for full reference, Testing for E2E test patterns.

Quick Start

1. Generate Keys

cargo run --bin tasker-ctl -- auth generate-keys --output-dir ./keys

2. Generate a Token

cargo run --bin tasker-ctl -- auth generate-token \
  --private-key ./keys/jwt-private-key.pem \
  --permissions "tasks:create,tasks:read,tasks:list,steps:read" \
  --subject my-service \
  --expiry-hours 24

3. Enable Auth in Configuration

In config/tasker/base/orchestration.toml:

[auth]
enabled = true
jwt_public_key_path = "./keys/jwt-public-key.pem"
jwt_issuer = "tasker-core"
jwt_audience = "tasker-api"

4. Use the Token

export TASKER_AUTH_TOKEN=<generated-token>
cargo run --bin tasker-ctl -- task list

Or with curl:

curl -H "Authorization: Bearer $TASKER_AUTH_TOKEN" http://localhost:8080/v1/tasks

Permission Vocabulary

Wildcards

• tasks:* - All task permissions
• steps:* - All step permissions
• dlq:* - All DLQ permissions
• * - All permissions (superuser)

Show All Permissions

cargo run --bin tasker-ctl -- auth show-permissions

Configuration Reference

Server-Side (orchestration.toml / worker.toml)

[auth]
enabled = true

# JWT Configuration
jwt_issuer = "tasker-core"
jwt_audience = "tasker-api"
jwt_token_expiry_hours = 24

# Key Configuration (one of these):
jwt_public_key_path = "./keys/jwt-public-key.pem"   # File path (preferred)
jwt_public_key = "-----BEGIN RSA PUBLIC KEY-----..." # Inline PEM
# Or set env: TASKER_JWT_PUBLIC_KEY_PATH

# JWKS (for dynamic key rotation)
jwt_verification_method = "jwks"  # "public_key" (default) or "jwks"
jwks_url = "https://auth.example.com/.well-known/jwks.json"
jwks_refresh_interval_seconds = 3600

# Permission validation
permissions_claim = "permissions"   # JWT claim containing permissions
strict_validation = true            # Reject tokens with unknown permissions
log_unknown_permissions = true

# API Key Authentication
api_key_header = "X-API-Key"
api_keys_enabled = true

[[auth.api_keys]]
key = "sk-prod-key-1"
permissions = ["tasks:read", "tasks:list", "steps:read"]
description = "Read-only monitoring service"

[[auth.api_keys]]
key = "sk-admin-key"
permissions = ["*"]
description = "Admin key"

Client-Side (Environment Variables)

Priority: endpoint-specific token > global token > API key > config file.

JWT Token Structure

{
  "sub": "my-service",
  "iss": "tasker-core",
  "aud": "tasker-api",
  "iat": 1706000000,
  "exp": 1706086400,
  "permissions": [
    "tasks:create",
    "tasks:read",
    "tasks:list",
    "steps:read"
  ],
  "worker_namespaces": []
}

Common Role Patterns

Read-only operator:

permissions: ["tasks:read", "tasks:list", "steps:read", "dlq:read", "dlq:stats"]

Task submitter:

permissions: ["tasks:create", "tasks:read", "tasks:list"]

Ops admin:

permissions: ["tasks:*", "steps:*", "dlq:*", "system:*"]

Worker service:

permissions: ["worker:config_read", "worker:templates_read"]

Superuser:

permissions: ["*"]

Public Endpoints

These endpoints never require authentication:

• GET /health - Basic health check
• GET /health/detailed - Detailed health
• GET /metrics - Prometheus metrics

API Key Authentication

API keys are validated against a configured registry. Each key has its own set of permissions.

# Using API key
curl -H "X-API-Key: sk-prod-key-1" http://localhost:8080/v1/tasks

API keys are simpler than JWTs but have limitations:

• No expiration (rotate by removing from config)
• No claims beyond permissions
• Best for service-to-service communication with static permissions

Error Responses

401 Unauthorized (Missing/Invalid Credentials)

{
  "error": "unauthorized",
  "message": "Missing authentication credentials"
}

403 Forbidden (Insufficient Permissions)

{
  "error": "forbidden",
  "message": "Missing required permission: tasks:create"
}

Migration Guide: Disabled to Enabled

1. Generate keys and distribute the public key to server config
2. Generate tokens for each service/user with appropriate permissions
3. Set enabled = true in auth config
4. Deploy - services without valid tokens will get 401 responses
5. Monitor the tasker.auth.failures.total metric for issues

All endpoints remain accessible without auth when enabled = false.

Observability

Structured Logs

• info on successful authentication (subject, method)
• warn on authentication failure (error details)
• warn on permission denial (subject, required permission)

Prometheus Metrics

CLI Auth Commands

# Generate RSA key pair
tasker-ctl auth generate-keys [--output-dir ./keys] [--key-size 2048]

# Generate JWT token
tasker-ctl auth generate-token \
  --permissions tasks:create,tasks:read \
  --subject my-service \
  --private-key ./keys/jwt-private-key.pem \
  --expiry-hours 24

# List all permissions
tasker-ctl auth show-permissions

# Validate a token
tasker-ctl auth validate-token \
  --token <JWT> \
  --public-key ./keys/jwt-public-key.pem

gRPC Authentication

gRPC endpoints support the same authentication methods as REST, using gRPC metadata instead of HTTP headers.

gRPC Ports

Bearer Token (gRPC)

# Using grpcurl with Bearer token
grpcurl -plaintext \
  -H "Authorization: Bearer $TASKER_AUTH_TOKEN" \
  localhost:9190 tasker.v1.TaskService/ListTasks

API Key (gRPC)

# Using grpcurl with API key
grpcurl -plaintext \
  -H "X-API-Key: sk-prod-key-1" \
  localhost:9190 tasker.v1.TaskService/ListTasks

gRPC Client Configuration

#![allow(unused)]
fn main() {
use tasker_client::grpc_clients::{OrchestrationGrpcClient, GrpcAuthConfig};

// With API key
let client = OrchestrationGrpcClient::connect_with_auth(
    "http://localhost:9190",
    GrpcAuthConfig::ApiKey("sk-prod-key-1".to_string()),
).await?;

// With Bearer token
let client = OrchestrationGrpcClient::connect_with_auth(
    "http://localhost:9190",
    GrpcAuthConfig::Bearer("eyJ...".to_string()),
).await?;
}

gRPC Error Codes