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 complex abstractions 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

• Example Apps & Integrations — Rails, FastAPI, Axum, and Bun integrations with working example projects

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-02-12 Audience: Engineers evaluating workflow orchestration tools Status: Early Release (0.1.x)

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 in early release (0.1.x). This is important context:

What this means:

• The architecture has solidified but breaking changes may still occur
• Documentation is comprehensive and continuously improving
• Early adopters are beginning to explore the system
• We follow semantic versioning and aim to communicate breaking changes clearly

Our commitment:

• Intentional about breaking changes—we weigh architectural correctness against user impact
• Migration guidance provided where practical
• Release notes document all significant changes
• Responsive to community feedback

If you’re evaluating Tasker, we encourage you to explore it, provide feedback, and help shape its direction. For production-critical workloads, evaluate whether the current stability level meets your needs, or consider the established tools above.

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

Contributing to Tasker

This guide covers development setup and workflow for contributing to tasker-core and tasker-contrib.

Contributing to tasker-core

Prerequisites

• Rust stable toolchain (via rustup)
• Docker Desktop (for PostgreSQL, RabbitMQ, and supporting services)
• cargo-make (cargo install cargo-make)
• Optional: Ruby 3.4+, Python 3.12+, Bun 1.x (for FFI worker development)

On macOS, the project includes a Brewfile that installs system dependencies (PostgreSQL 18, protobuf, LLVM, language toolchains).

Automated Setup

The fastest path is the automated setup script:

git clone https://github.com/tasker-systems/tasker-core.git
cd tasker-core

# Full setup — installs Homebrew deps, Rust tools, git hooks, worker deps
./bin/setup-dev.sh

# Or run targeted slices
./bin/setup-dev.sh --brew-only     # Homebrew bundle only
./bin/setup-dev.sh --cargo-only    # cargo-make, sqlx-cli, nextest
./bin/setup-dev.sh --hooks-only    # Git pre-commit hook
./bin/setup-dev.sh --check         # Audit what's installed

Starting Services

# Start PostgreSQL (with PGMQ), RabbitMQ, Dragonfly cache, Grafana LGTM
cargo make docker-up

Database Setup

cargo make db-setup    # Run migrations
cargo make db-reset    # Drop and recreate (clean slate)

Environment Configuration

Environment variables are assembled from modular files in config/dotenv/:

cargo make setup-env           # Standard test mode
cargo make setup-env-split     # Split database mode
cargo make setup-env-cluster   # Cluster testing mode

See config/dotenv/README.md for the full file structure and assembly order.

Key Commands

Always use --all-features when running cargo commands directly.

Test Tiers

Tests are organized into infrastructure levels via feature flags:

Run cargo make test-rust-unit for fast iteration. Run the full suite with cargo make test before opening a PR.

Worker Development

Tasker supports polyglot workers through FFI:

• Ruby via magnus — workers/ruby/
• Python via PyO3 — workers/python/
• TypeScript via napi-rs — workers/typescript/

Each worker directory has its own build and test commands. See the Worker Guides for language-specific details.

SQLx Query Cache

After modifying sqlx::query! macros or SQL schema, update the offline cache:

DATABASE_URL=postgresql://tasker:tasker@localhost:5432/tasker_rust_test \
  cargo sqlx prepare --workspace -- --all-targets --all-features

git add .sqlx/

Conventions

• Branch naming: username/ticket-id-short-description (e.g., jcoletaylor/tas-190-add-version-fields)
• Commit messages: type(scope): description (e.g., fix(orchestration): handle timeout in step enqueuer)
• Git hooks: Pre-commit runs cargo fmt on staged Rust files. Install with git config core.hooksPath .githooks or via setup-dev.sh.
• Lint suppression: Use #[expect(lint_name, reason = "...")] instead of #[allow]
• SQLx: Never use SQLX_OFFLINE=true — always export DATABASE_URL
• Public types: Must implement Debug
• Channels: All MPSC channels must be bounded and configured via TOML

Pull Request Process

1. Branch from main
2. Make focused changes (one logical change per PR)
3. Run cargo make check && cargo make test
4. Update documentation if your change affects public APIs or behavior
5. Open a PR against main

New functionality should include tests. Bug fixes should include a regression test.

Contributing to tasker-contrib

Prerequisites

• cargo-make (cargo install cargo-make)
• tasker-ctl binary (build from core or cargo install tasker-ctl)
• Docker and Docker Compose (for example app testing)
• Language toolchain for the area you’re working on (Ruby 3.3+, Python 3.12+, Bun 1.0+, or Rust stable)

Getting tasker-ctl

# Option A: Install from crates.io
cargo install tasker-ctl

# Option B: Build from local tasker-core (requires sibling checkout)
cargo make build-ctl

Adding a New Template

Each language plugin lives in {language}/tasker-cli-plugin/ and follows this structure:

{language}/tasker-cli-plugin/
+-- tasker-plugin.toml          # Plugin manifest
+-- templates/
    +-- step_handler/           # Template directory
    |   +-- template.toml       # Metadata and variables
    |   +-- files/              # Tera template files
    +-- step_handler_api/
    +-- task_template/

Steps to add a template:

1. Create the template directory under the appropriate plugin
2. Add template.toml with metadata and variable definitions
3. Add template files in files/ using Tera syntax (built-in helpers: snake_case, pascal_case)
4. Register the template in tasker-plugin.toml
5. Run cargo make test-templates to verify generation and syntax checking

Adding a New Example App

1. Create examples/{framework}-app/ with standard project structure
2. Add the app database to examples/init-db.sql
3. Add a cargo make test-example-{framework} task to Makefile.toml
4. Add the app to the test-examples dependencies in Makefile.toml
5. Add CI steps to .github/workflows/test-examples.yml

Validation Commands

Example App Infrastructure

The example apps share a Docker Compose stack in examples/:

cd examples
docker compose up -d

This starts PostgreSQL 18 (with PGMQ + app databases), Tasker orchestration (from published GHCR images), Dragonfly cache, and RabbitMQ. Wait for orchestration to be healthy before running tests:

curl -sf http://localhost:8080/health

Getting Help

• Discussions for questions
• Issues for bugs or feature requests
• Both projects follow the Contributor Covenant 3.0 code of conduct

Getting Started

Tasker is a distributed workflow orchestration system that coordinates complex, multi-step processes across services and languages. It provides:

• Task Orchestration — Define workflows as directed acyclic graphs (DAGs) with dependency management
• Multi-Language Support — Write handlers in Rust, Ruby, Python, or TypeScript
• Built-in Resilience — Automatic retries, error handling, and state persistence
• Event-Driven Architecture — PGMQ and RabbitMQ messaging for real-time observability

How Tasker Works

┌─────────────────────────────────────────────────────────────────────┐
│                        tasker-core (Rust)                           │
│  • REST + gRPC API for task submission                              │
│  • Workflow orchestration via lifecycle actors                      │
│  • Step execution and DAG dependency resolution                    │
│  • PostgreSQL state persistence                                     │
│  • Event publishing (PGMQ default, RabbitMQ optional)              │
└─────────────────────────────────────────────────────────────────────┘
                                  │
                    ┌─────────────┼─────────────┐
                    ▼             ▼             ▼
              ┌──────────┐ ┌──────────┐ ┌──────────┐
              │  Ruby    │ │  Python  │ │TypeScript│
              │ Workers  │ │ Workers  │ │ Workers  │
              └──────────┘ └──────────┘ └──────────┘

You define task templates (YAML DAGs) that describe what steps to run and their dependencies. You write step handlers in your preferred language that contain the business logic. Tasker’s orchestration engine executes the DAG — resolving dependencies, running independent steps in parallel, retrying failures, and persisting state.

Understand Tasker

Ready to Build?

Once you understand the concepts, head to Build Your First Project to set up your environment and write your first workflow.

If you prefer learning by example, the Quick Start gets you running a working app in 5 minutes.

Core Concepts

This page explains the fundamental building blocks of Tasker.

Tasks

A Task is a unit of work submitted to Tasker for execution. Tasks have:

• A task template that defines the workflow structure
• An initiator identifying the source (e.g., user:123, system:scheduler)
• A context containing input data and metadata
• A state managed by a 12-state machine (see below)

{
  "name": "order_fulfillment",
  "initiator": "api:checkout",
  "context": {
    "order_id": "ORD-12345",
    "customer_email": "customer@example.com"
  }
}

Task Templates

A Task Template is a YAML definition of a workflow. It specifies:

• Steps to execute
• Dependencies between steps (creating a DAG)
• Handler mappings connecting steps to your code

name: order_fulfillment
namespace_name: ecommerce
version: 1.0.0
steps:
  - name: validate_order
    handler:
      callable: OrderValidationHandler
    dependencies: []

  - name: reserve_inventory
    handler:
      callable: InventoryHandler
    dependencies:
      - validate_order

  - name: charge_payment
    handler:
      callable: PaymentHandler
    dependencies:
      - validate_order

Steps

A Step is a single operation within a workflow. Steps:

• Execute independently once dependencies are satisfied
• Can run in parallel when they have no mutual dependencies
• Return results that downstream steps can access
• Can be retried on failure

Task Lifecycle

Tasks progress through a multi-phase lifecycle managed by the orchestration actors:

Pending → Initializing → EnqueuingSteps → StepsInProcess → EvaluatingResults → Complete

The evaluating phase may loop back to enqueue more steps as dependencies are satisfied, wait for retries, or transition to terminal states (Complete, Error, Cancelled, ResolvedManually). Tasks support cancellation from any non-terminal state and manual resolution from BlockedByFailures.

Step Lifecycle

Steps follow a worker-to-orchestration handoff pattern through 10 states:

Pending → Enqueued → InProgress → EnqueuedForOrchestration → Complete

After a worker executes a step, the result is enqueued back to orchestration for processing. Steps can also transition through WaitingForRetry for automatic retry with backoff, or be cancelled, failed, or manually resolved.

For the full state machine diagrams and transition tables, see States and Lifecycles.

Step Handlers

A Step Handler is your code that executes a step’s business logic. The DSL approach declares what a handler receives — its inputs from the task context and results from upstream steps — and delegates to a service:

from tasker_core.step_handler.functional import inputs, step_handler
from app.services.types import EcommerceOrderInput
from app.services import ecommerce as svc

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    return svc.validate_cart_items(inputs.resolved_items)

The @inputs decorator extracts fields from the task context into a typed Pydantic model. The @depends_on decorator (shown below) does the same for upstream step results. Your handler function receives typed arguments instead of parsing raw JSON.

Class-based handlers (class MyHandler(StepHandler)) are also supported. See Class-Based Handlers.

Dependency Results

Steps can access typed results from their dependencies using @depends_on:

from tasker_core.step_handler.functional import depends_on, inputs, step_handler
from app.services.types import EcommerceOrderInput, EcommerceValidateCartResult

@step_handler("process_payment")
@depends_on(cart_result=("validate_cart", EcommerceValidateCartResult))
@inputs(EcommerceOrderInput)
def process_payment(
    cart_result: EcommerceValidateCartResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.process_payment(
        payment_token=inputs.payment_token,
        total=cart_result.total or 0.0,
    )

Each @depends_on entry maps a parameter name to a ("step_name", ResultModel) tuple. Tasker resolves the upstream step’s result and deserializes it into the model, so your handler receives a typed object — not a raw dict.

Workflow Steps

A Workflow Step is a special step that delegates to another task template, creating a sub-workflow:

steps:
  - name: process_line_items
    handler:
      callable: WorkflowHandler
    dependencies:
      - validate_order

Error Handling

Tasker distinguishes between error types:

from tasker_core.errors import PermanentError, RetryableError

def call(self, context):
    if invalid_input:
        raise PermanentError(message="Invalid order ID format", error_code="INVALID_ID")
    if service_unavailable:
        raise RetryableError(message="Payment gateway timeout", error_code="GATEWAY_TIMEOUT")

Next Steps

• Handler Types — The four handler types and when to use each
• Your First Handler — Write your first step handler
• Your First Workflow — Create a complete workflow

Handler Types

Tasker provides four handler types that cover the most common workflow patterns. The DSL approach lets you declare what a handler receives — typed inputs and dependency results — while your business logic stays in your service layer.

Cross-Language Availability

Rust provides only the base Step Handler trait, composing capability traits instead. See Rust’s Handler Architecture below.

Step Handler (DSL)

The base handler type. All other types extend it.

When to use: General-purpose business logic — database operations, calculations, transformations, service calls, or anything that takes input and produces output.

Python

from tasker_core.step_handler.functional import inputs, step_handler
from app.services.types import EcommerceOrderInput
from app.services import ecommerce as svc

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    return svc.validate_cart_items(inputs.resolved_items)

Ruby

extend TaskerCore::StepHandler::Functional

ValidateCartHandler = step_handler(
  'Ecommerce::StepHandlers::ValidateCartHandler',
  inputs: Types::Ecommerce::OrderInput
) do |inputs:, context:|
  Ecommerce::Service.validate_cart_items(cart_items: inputs.cart_items)
end

TypeScript

import { defineHandler } from '@tasker-systems/tasker';
import * as svc from '../services/ecommerce';

export const ValidateCartHandler = defineHandler(
  'Ecommerce.StepHandlers.ValidateCartHandler',
  { inputs: { cartItems: 'cart_items' } },
  async ({ cartItems }) => svc.validateCartItems(cartItems as CartItem[]),
);

For the class-based alternative, see Class-Based Handlers.

Generate with tasker-ctl:

tasker-ctl template generate step_handler \
  --plugin tasker-contrib-python \
  --param name=ProcessPayment

Available for all four languages: tasker-contrib-rails, tasker-contrib-python, tasker-contrib-typescript, tasker-contrib-rust.

See it in action: All five workflows in the example apps use step handlers. Start with the e-commerce checkout (Post 01) for the simplest example.

Next: Your First Handler walks through writing and registering a step handler end-to-end.

What the DSL Composes

The DSL builds a typed method signature from two sources:

Both are injected as function parameters. Your handler receives typed objects — not raw dicts or JSON — and delegates to a service function that contains the actual business logic.

Here’s a handler that uses both:

@step_handler("create_order")
@depends_on(
    cart_result=("validate_cart", EcommerceValidateCartResult),
    payment_result=("process_payment", EcommerceProcessPaymentResult),
    inventory_result=("update_inventory", EcommerceUpdateInventoryResult),
)
@inputs(EcommerceOrderInput)
def create_order(
    cart_result: EcommerceValidateCartResult,
    payment_result: EcommerceProcessPaymentResult,
    inventory_result: EcommerceUpdateInventoryResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.create_order(
        cart=cart_result, payment=payment_result,
        inventory=inventory_result, customer_email=inputs.customer_email,
    )

The handler declares what it needs; Tasker resolves how to get it.

Type System by Language

Each language uses its native type system for input and result models:

Python (Pydantic BaseModel):

class EcommerceOrderInput(BaseModel):
    items: list[dict[str, Any]] | None = None        # submitted as "items"
    cart_items: list[dict[str, Any]] | None = None    # or "cart_items"
    customer_email: str | None = None
    payment_token: str | None = None

    @property
    def resolved_items(self) -> list[dict[str, Any]]:
        """Accept either field name from the task context."""
        return self.items or self.cart_items or []

Ruby (Dry::Struct):

module Types
  module Ecommerce
    class OrderInput < Types::InputStruct
      attribute :cart_items, Types::Array.of(Types::Hash).optional
      attribute :customer_email, Types::String.optional
      attribute :payment_token, Types::String.optional
    end
  end
end

TypeScript (interfaces):

interface CartItem {
  sku: string;
  name: string;
  price: number;
  quantity: number;
}

Specialized Handler Patterns

API Handler

Adds HTTP client methods with built-in error classification. The APIMixin provides self.get(), self.post(), etc. with automatic retryable/permanent error detection.

When to use: Calling external APIs where you need to distinguish retryable errors (5xx, timeouts) from permanent errors (4xx).

API handlers currently use the class-based pattern with mixin composition. See Class-Based Handlers — API Handler for the full pattern.

Decision Handler

Adds workflow routing methods. decision_success() activates downstream steps by name; skip_branches() when no steps should execute.

When to use: Conditional branching — when the next steps depend on runtime data.

from tasker_core.step_handler.functional import decision_handler

@decision_handler("order_routing")
def order_routing(context: StepContext):
    order_type = context.get_input("order_type")
    if order_type == "premium":
        return ["validate_premium", "process_premium"]
    return ["standard_processing"]

See Conditional Workflows for decision handler patterns in depth.

Batchable Handler

Adds batch processing for splitting large workloads into parallel cursor-based batches.

When to use: Processing large datasets where you want to divide work across multiple parallel workers.

Workflow pattern: Analyzer → parallel Workers → optional Aggregator.

Batchable handlers currently use the class-based pattern due to their stateful nature (cursor management, batch context). See Class-Based Handlers — Batchable Handler for the full pattern, and Batch Processing for the production guide.

Task Templates

All handler types are wired together using YAML task template definitions. A task template defines the DAG — which steps to run, their dependencies, and which handlers to invoke.

name: order_processing
namespace: ecommerce
version: "1.0.0"
description: "Order processing workflow"

step_templates:
  - name: validate_order
    description: "Validate the incoming order"
    handler:
      callable: ValidateOrderHandler
      initialization: {}
    depends_on_step_name: []
    retry:
      max_attempts: 3
      backoff_strategy: exponential
      backoff_base_seconds: 2

Generate with tasker-ctl:

tasker-ctl template generate task_template \
  --plugin tasker-contrib-python \
  --param name=OrderProcessing \
  --param namespace=ecommerce

Task templates are language-agnostic — the same YAML structure works across all four languages. The handler.callable field maps to the handler’s registered name or class path.

For a complete walkthrough of building a multi-step workflow with templates, see Your First Workflow.

Rust’s Handler Architecture

Rust provides RustStepHandler as its single handler trait — but this is not a limitation. The Rust worker crate defines capability traits in handler_capabilities.rs that Rust handlers compose directly:

A Rust handler implements RustStepHandler and adds any capability traits it needs. This is idiomatic Rust — trait composition instead of class inheritance. For a complex example that combines multiple capabilities, see diamond_decision_batch.rs in the Rust worker crate.

In fact, the Rust batch_processing module is the foundation that Python, Ruby, and TypeScript access through FFI. The specialized handler types in those languages are ergonomic wrappers around the Rust implementation — Rust developers work with the underlying traits directly.

Choosing Your Package

Tasker supports multiple languages through FFI bindings. Each language package provides the same core capabilities with idiomatic APIs.

Language Guides

How to Choose

Use Rust

• Need maximum performance with zero-overhead abstractions
• Want compile-time type safety and memory safety guarantees
• Are building native Tasker extensions or the orchestration system itself
• Prefer direct API access without FFI overhead

Use Ruby

• Have an existing Rails or Ruby application
• Prefer Ruby’s expressive DSL capabilities
• Value rapid development with convention-over-configuration
• Want seamless integration with Ruby ecosystem gems

Use Python

• Are building data pipelines or ML workflows
• Want async/await support with asyncio, aiohttp, etc.
• Need integration with the Python data science ecosystem
• Prefer Python’s clean syntax and type hints

Use TypeScript

• Are building Node.js or Bun applications
• Want strong typing with TypeScript’s type system
• Need to integrate with existing JavaScript ecosystems
• Prefer modern async/await patterns

Package Architecture

All language packages share the same architecture:

┌─────────────────────────────────────────────────────────────┐
│                    Your Application                          │
├─────────────────────────────────────────────────────────────┤
│  Language Package (tasker-rb, tasker-py, @tasker/tasker)   │
├─────────────────────────────────────────────────────────────┤
│  FFI Layer (Magnus/PyO3/NAPI)                               │
├─────────────────────────────────────────────────────────────┤
│  tasker-worker (Rust core)                                   │
├─────────────────────────────────────────────────────────────┤
│  tasker-client (API client)                                  │
└─────────────────────────────────────────────────────────────┘

This means:

• Same core logic — All packages use the same Rust implementation
• Same features — Handler registration, client SDK, event system
• Cross-language consistency — get_input(), get_dependency_result(), etc. work the same

Quick Installation

# Rust
cargo add tasker-worker tasker-client

# Ruby
gem install tasker-rb

# Python
pip install tasker-py

# TypeScript/JavaScript
npm install @tasker-systems/tasker

See the individual language guides for detailed setup and examples.

Example Apps & Framework Integrations

Tasker Contrib provides two things for each supported language:

1. CLI plugin templates — code generators for tasker-ctl that scaffold handlers, task definitions, and infrastructure configuration
2. Example applications — fully working apps that demonstrate real-world workflow patterns against published Tasker packages

The Integration Pattern

All four framework integrations follow the same pattern:

1. Bootstrap — tasker-ctl init creates project structure with infrastructure config
2. Create — tasker-ctl template generate scaffolds handlers and task templates
3. Process — Your handlers receive StepContext, execute business logic, return StepHandlerResult
4. Query — Use the Tasker client SDK to submit tasks and check status

The framework integration layer is intentionally thin: it translates your framework’s idioms (Rails generators, FastAPI dependency injection, Bun middleware) into Tasker concepts without inventing new abstractions.

Available Integrations

Each plugin provides templates for all handler types: step, API, decision, and batchable (Rust provides step handler only).

The Apps

Workflow Patterns

All four apps implement the same five workflow patterns, progressing from simple to complex:

Each workflow demonstrates a specific DAG pattern. The Engineering Stories series teaches these patterns through progressive narrative — start with Post 01 and work forward.

Shared Infrastructure

All example apps share a single docker-compose.yml that provides:

• PostgreSQL with PGMQ extensions — state persistence and default message queue
• Tasker orchestration engine — published GHCR image, handles DAG execution
• RabbitMQ — optional message broker for event publishing
• Dragonfly — Redis-compatible cache

cd examples
docker compose up -d

# Wait for orchestration to be healthy
curl -sf http://localhost:8080/health

Each app gets its own database (example_rails, example_fastapi, example_bun, example_axum) created by the shared init-db.sql script.

Running an Example

Python (FastAPI)

cd examples/fastapi-app
uv sync
uv run uvicorn app.main:app --port 8083

Ruby (Rails)

cd examples/rails-app
bundle install
bin/rails server -p 8082

TypeScript (Bun/Hono)

cd examples/bun-app
bun install
bun run dev

Rust (Axum)

cd examples/axum-app
cargo run

What to Study

Each app demonstrates the same concepts in its framework’s idioms. Comparing across languages is the fastest way to understand Tasker’s cross-language handler contract:

• Handler registration — How each framework discovers and registers step handlers
• Context access — get_input(), get_dependency_result(), and step configuration
• Error handling — PermanentError vs RetryableError patterns
• Task templates — Identical YAML DAG definitions across all four apps
• Testing — Each app has integration tests that submit tasks and verify results

Getting Started

• Quick Start — Clone an example app and run it in 5 minutes
• Using tasker-ctl — Bootstrap a project with the CLI tool
• Choosing Your Package — Which language SDK fits your project

Source Repository

All example apps live in the tasker-systems/tasker-contrib repository under examples/.

Build Your First Project

This section walks you from zero to a running Tasker workflow. Choose your path based on how you like to learn:

Quick Start

Quick Start — Two ways to get running fast:

• Path A: Clone an example app (5 min) — Docker Compose up, pick a framework, curl an endpoint
• Path B: Bootstrap with tasker-ctl (10 min) — Initialize a project, generate handlers, configure, and run

Step-by-Step Guides

If you prefer a guided walkthrough:

1. Installation — Install language packages and run Tasker infrastructure
2. Using tasker-ctl — Initialize projects, generate handlers from templates, manage configuration
3. Your First Handler — Write a step handler in your preferred language
4. Your First Workflow — Define a task template, submit a task, watch it execute

Language Guides

Comprehensive setup and API reference for each supported language:

• Ruby — Ruby workers with tasker-rb
• Python — Python workers with tasker-py
• TypeScript — TypeScript workers with @tasker-systems/tasker
• Rust — Native Rust workers with tasker-worker

What’s Next

After building your first project, see Next Steps for where to go from here — example apps, engineering stories, architecture deep-dives, and production operations.

Quick Start

Two paths to a running Tasker workflow. Pick the one that fits your style.

Path A: Clone an Example App (5 minutes)

The fastest way to see Tasker in action. The example apps provide fully working projects in all four languages, running against published packages via Docker Compose.

Prerequisites

• Docker and Docker Compose
• curl (or any HTTP client)

1. Clone tasker-contrib

git clone https://github.com/tasker-systems/tasker-contrib.git
cd tasker-contrib/examples

2. Start the infrastructure

docker compose up -d

This starts PostgreSQL (with PGMQ), the Tasker orchestration engine, RabbitMQ, and Dragonfly (cache). All services use published GHCR images — no local builds needed.

Apple Silicon: The compose file includes platform: linux/amd64 for Tasker images. Ensure “Use Rosetta” is enabled in Docker Desktop.

3. Wait for orchestration to be healthy

# Retry until healthy (up to 60 seconds on first pull)
until curl -sf http://localhost:8080/health > /dev/null; do
  echo "Waiting for orchestration..."
  sleep 5
done
echo "Orchestration is healthy"

4. Pick a framework and run it

Each app has its own setup instructions. For example, with Ruby (Rails):

cd rails-app
bundle install
bin/rails db:create db:migrate
bin/rails server -p 3000

Or with Python (FastAPI):

cd fastapi-app
uv sync
uv run alembic upgrade head
uv run uvicorn app.main:app --port 8000

5. Submit a task

# Submit an e-commerce order processing task
curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ecommerce_order_processing",
    "namespace": "ecommerce_rb",
    "version": "1.0.0",
    "initiator": "quickstart",
    "source_system": "cli",
    "reason": "Quick-start verification",
    "context": {
      "cart_items": [
        {"sku": "WIDGET-001", "name": "Widget", "quantity": 2, "unit_price": 29.99}
      ],
      "customer_email": "test@example.com"
    }
  }'

The orchestration engine coordinates the workflow — validating the cart, processing payment, reserving inventory, creating the order, and sending confirmation — with each step handled by your chosen framework’s app.

What you just ran

Each example app implements four real-world workflow patterns:

See the Example Apps page for full details, or read the Engineering Stories for narrative walkthroughs.

Path B: Bootstrap with tasker-ctl (10 minutes)

Build a project from scratch using Tasker’s CLI tool. This path generates handler scaffolding, task templates, and infrastructure configuration — everything you need to start writing business logic.

Prerequisites

• Docker and Docker Compose (for Tasker infrastructure)
• Rust toolchain (for installing tasker-ctl)
• Your preferred language runtime (Python, Ruby, Bun/Node, or Rust)

1. Install and initialize

cargo install tasker-ctl

mkdir my-tasker-project && cd my-tasker-project
tasker-ctl init
tasker-ctl remote update

This creates a .tasker-ctl.toml configured with the tasker-contrib remote, then fetches the community templates to your local cache.

2. See what’s available

tasker-ctl template list

Templates are organized by language and type:

Filter by language to see just your stack:

tasker-ctl template list --language python

3. Generate a handler

tasker-ctl template generate step_handler \
  --language python \
  --param name=ProcessOrder

This generates process_order_handler.py and tests/test_process_order_handler.py using the DSL pattern — a decorated function with typed inputs that delegates to a service. See Your First Handler for a walkthrough of the generated code.

4. Generate a task template

tasker-ctl template generate task_template \
  --language python \
  --param name=OrderProcessing \
  --param namespace=default \
  --param handler_callable=handlers.process_order_handler.ProcessOrderHandler

This generates order_processing.yaml — a task definition with one step wired to your handler. Edit it to add more steps and build a DAG.

5. Generate infrastructure

# Docker Compose stack (PostgreSQL + orchestration, optionally RabbitMQ + Dragonfly)
tasker-ctl template generate docker_compose \
  --plugin tasker-contrib-ops \
  --param name=myproject

# TOML configuration files (from tasker-contrib/config/tasker base configs)
tasker-ctl config generate --remote tasker-contrib \
  --context orchestration --environment development --output config/orchestration.toml
tasker-ctl config generate --remote tasker-contrib \
  --context worker --environment development --output config/worker.toml

6. Start infrastructure and submit

docker compose up -d

# Wait for health
until curl -sf http://localhost:8080/health > /dev/null; do sleep 5; done

# Submit a task
curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "name": "order_processing",
    "namespace": "default",
    "version": "1.0.0",
    "initiator": "quickstart",
    "source_system": "cli",
    "reason": "First task",
    "context": {"order_id": "ORD-001"}
  }'

What you just built

Path B gives you project scaffolding — handler code, task template YAML, and infrastructure config. To wire a handler into a running worker, you’ll need to integrate it with a framework (Rails, FastAPI, Bun, or Axum) that starts a Tasker worker at boot. See the language guides for that next step, or study the example apps for complete working implementations.

Next Steps

• Your First Handler — Detailed walkthrough of handler anatomy
• Your First Workflow — Build a multi-step DAG with dependencies
• Using tasker-ctl — Full CLI reference for project scaffolding

Installation

This guide covers installing Tasker components for development.

Prerequisites

• Docker and Docker Compose V2 (for Tasker infrastructure)
• Rust toolchain (for installing tasker-ctl)
• A language runtime for your workers: Ruby 3.2+, Python 3.10+, Bun 1.0+ (or Node 18+), or Rust 1.75+

Install tasker-ctl

tasker-ctl is the CLI tool for scaffolding projects, generating handlers, and managing configuration:

cargo install tasker-ctl

Verify the installation:

tasker-ctl --version
# tasker-ctl 0.1.4

Apple Silicon note: The published Docker images on GHCR are currently x86_64 only. On Apple Silicon Macs, enable “Use Rosetta for x86_64/amd64 emulation” in Docker Desktop settings, or ensure your docker-compose.yml includes platform: linux/amd64 on Tasker service containers.

Installing Worker Packages

Install the package for your language of choice:

Ruby

gem install tasker-rb

Or add to your Gemfile:

gem 'tasker-rb', '~> 0.1.5'

Python

pip install tasker-py

Or with uv:

uv add tasker-py

TypeScript / JavaScript

bun add @tasker-systems/tasker

Or with npm:

npm install @tasker-systems/tasker

Rust

Add to your Cargo.toml:

[dependencies]
tasker-worker = "0.1.5"
tasker-client = "0.1.5"

Infrastructure with Docker Compose

Tasker requires PostgreSQL (with the PGMQ extension) and an orchestration service. The fastest way to get these running is Docker Compose.

You can generate a compose file with tasker-ctl:

tasker-ctl init
tasker-ctl remote update
tasker-ctl template generate docker_compose \
  --plugin tasker-contrib-ops \
  --param name=myproject

Or use the pre-configured stack from the example apps:

git clone https://github.com/tasker-systems/tasker-contrib.git
cd tasker-contrib/examples
docker compose up -d

This starts PostgreSQL (with PGMQ), the Tasker orchestration engine, RabbitMQ, and Dragonfly (cache). The orchestration API is available at http://localhost:8080.

Verify services are running

curl -sf http://localhost:8080/health

Configuration

Tasker uses environment variables and TOML configuration files. Key environment variables:

# Database connection (required)
export DATABASE_URL="postgresql://tasker:tasker@localhost:5432/tasker"

# Orchestration API URL (for client SDKs and tasker-ctl)
export ORCHESTRATION_URL="http://localhost:8080"

# Messaging backend: "pgmq" (default, uses PostgreSQL) or "rabbitmq"
export TASKER_MESSAGING_BACKEND="pgmq"

For full configuration management, see Configuration Management or generate annotated config files with tasker-ctl config generate.

Next Steps

• Quick Start — Two paths to a running workflow
• Using tasker-ctl — Project scaffolding and template generation
• Your First Handler — Write your first step handler

Getting Started with tasker-ctl

tasker-ctl is the command-line tool for managing Tasker workflows, generating project scaffolding, and working with configuration. This guide covers the developer-facing features for bootstrapping new projects.

Installation

Install from crates.io:

cargo install tasker-ctl

Or with cargo-binstall (uses prebuilt binaries when available):

cargo binstall tasker-ctl

Initialize Your Project

Run tasker-ctl init to create a .tasker-ctl.toml configuration file in your project directory:

tasker-ctl init

This creates a .tasker-ctl.toml pre-configured with the tasker-contrib remote, which provides community templates for all supported languages and default configuration files.

To skip the tasker-contrib remote (e.g., if you only use private templates):

tasker-ctl init --no-contrib

Fetch Remote Templates

After initialization, fetch the remote templates:

tasker-ctl remote update

This clones the configured remotes to a local cache (~/.cache/tasker-ctl/remotes/). Subsequent fetches only pull changes. The cache is checked for freshness automatically and warnings are shown when it becomes stale (default: 24 hours).

Browse Templates

List all available templates:

tasker-ctl template list

Filter by language:

tasker-ctl template list --language ruby
tasker-ctl template list --language python

Get detailed information about a template:

tasker-ctl template info step_handler --language ruby

Generate Code from Scaffolding Templates

Generate a step handler from a scaffolding template:

tasker-ctl template generate step_handler \
  --language ruby \
  --param name=ProcessPayment \
  --output ./app/handlers/

This creates handler files using the naming conventions and patterns for your chosen language. Template parameters (like name) are transformed automatically — ProcessPayment becomes process_payment for file names and ProcessPaymentHandler for class names.

Generate Typed Code from Task Templates

The generate command reads your task template YAML files and produces typed code — result models and handler scaffolds — in any supported language. This keeps your handler code aligned with the schemas defined in your templates.

tasker-ctl generate <COMMAND>

Commands:
  types    Generate typed result models from step result_schema definitions
  handler  Generate handler scaffolds with typed dependency wiring

Generate Types

Generate typed result models from the result_schema defined on each step in a task template:

tasker-ctl generate types \
  --template config/tasker/templates/ecommerce_order_processing.yaml \
  --language typescript

This reads each step’s result_schema and produces language-idiomatic types. For TypeScript, you get Zod schemas with inferred types:

export const EcommerceValidateCartResultSchema = z.object({
  free_shipping: z.boolean(),
  item_count: z.number().int(),
  subtotal: z.number(),
  tax: z.number(),
  total: z.number(),
  validated_items: z.array(EcommerceValidateCartResultValidatedItemsSchema),
  validation_id: z.string(),
  // ...
}).passthrough();

export type EcommerceValidateCartResult = z.infer<typeof EcommerceValidateCartResultSchema>;

For Python, you get Pydantic models:

class EcommerceValidateCartResult(BaseModel):
    free_shipping: bool
    item_count: int
    subtotal: float
    total: float
    validated_items: list[EcommerceValidateCartResultValidatedItems]
    validation_id: str

To generate types for a single step:

tasker-ctl generate types \
  --template config/tasker/templates/ecommerce_order_processing.yaml \
  --language python \
  --step validate_cart

Supported languages: typescript (ts), python (py), ruby (rb), rust (rs).

Generate Handlers

Generate handler scaffolds with typed dependency wiring:

tasker-ctl generate handler \
  --template config/tasker/templates/ecommerce_order_processing.yaml \
  --language typescript \
  --step process_payment

The generator reads the step’s dependencies from the template and wires them into the handler scaffold:

export const ProcessPaymentHandler = defineHandler(
  'Ecommerce::StepHandlers::ProcessPaymentHandler',
  {
    depends: {
      validateCartResult: 'validate_cart'
    },
  },
  async ({ validateCartResult, context }) => {
    // validateCartResult: ValidateCartResult (typed)
    // TODO: implement handler logic
    return {
      amount_charged: 0,
      authorization_code: "",
      currency: "",
      // ...
    };
  }
);

The return value stub matches the step’s result_schema, so you can fill in real logic and the shape is already correct.

To generate handlers for all steps at once, omit --step. Add --with-tests to also generate test scaffolds:

tasker-ctl generate handler \
  --template config/tasker/templates/ecommerce_order_processing.yaml \
  --language typescript \
  --with-tests

Use --output to write to a file instead of stdout:

tasker-ctl generate types \
  --template config/tasker/templates/ecommerce_order_processing.yaml \
  --language typescript \
  --output src/services/types.ts

Generate Configuration

Generate a deployable configuration file from the base + environment configs:

# From local config directory
tasker-ctl config generate \
  --context orchestration \
  --environment production \
  --output config/orchestration.toml

# From a remote (tasker-contrib provides default configs)
tasker-ctl config generate \
  --remote tasker-contrib \
  --context orchestration \
  --environment development \
  --output config/orchestration.toml

The config generate command merges base configuration with environment-specific overrides and strips documentation metadata, producing a clean deployment-ready TOML file.

Manage Remotes

tasker-ctl remote list                    # Show configured remotes and cache status
tasker-ctl remote add my-templates URL    # Add a new remote
tasker-ctl remote update                  # Fetch latest for all remotes
tasker-ctl remote update tasker-contrib   # Fetch a specific remote
tasker-ctl remote remove my-templates     # Remove a remote and its cache

Typical Workflow

A new project typically follows this sequence:

# 1. Initialize CLI configuration
tasker-ctl init

# 2. Fetch community templates
tasker-ctl remote update

# 3. Scaffold a step handler from a template
tasker-ctl template generate step_handler --language python --param name=ValidateOrder

# 4. Scaffold a task template
tasker-ctl template generate task_template --language python \
  --param name=OrderProcessing \
  --param namespace=default \
  --param handler_callable=handlers.validate_order_handler.ValidateOrderHandler

# 5. Generate typed code from your task template
tasker-ctl generate types --template config/tasker/templates/order_processing.yaml --language python
tasker-ctl generate handler --template config/tasker/templates/order_processing.yaml --language python

# 6. Generate infrastructure (uses --plugin for ops templates)
tasker-ctl template generate docker_compose --plugin tasker-contrib-ops --param name=myproject

# 7. Generate environment-specific config (merges base + environment overrides)
tasker-ctl config generate --remote tasker-contrib \
  --context worker --environment development --output config/worker.toml

Note: Language templates use --language to select the plugin (e.g., --language ruby selects tasker-contrib-rails). Ops templates use --plugin tasker-contrib-ops directly since they are language-independent.

Next Steps

• Your First Handler — Write a step handler from scratch
• Your First Workflow — Define a task template and run it
• Configuration Management — Understanding the TOML config structure
• tasker-ctl Architecture — How the CLI is built

Your First Handler

This guide walks you through writing your first step handler using the DSL.

What is a Handler?

A Step Handler is your code that executes business logic for a single workflow step. With the DSL, a handler declares what it receives — typed inputs from the task context and typed results from upstream steps — and delegates to a service function. Handlers are thin wrappers: Tasker handles sequencing, retries, and error classification; your service layer handles the business logic.

You can generate a handler from a template with tasker-ctl:

tasker-ctl template generate step_handler --language python --param name=ProcessOrder

Or write one from scratch using the patterns below.

The DSL Approach

Every handler follows the same three-layer pattern:

1. Type definition — the contract (what the handler receives and returns)
2. Handler declaration — the DSL wiring (which step, which inputs, which dependencies)
3. Service delegation — one-line call to your business logic

Python

# app/services/types.py — the contract
class EcommerceOrderInput(BaseModel):
    items: list[dict[str, Any]] | None = None        # submitted as "items"
    cart_items: list[dict[str, Any]] | None = None    # or "cart_items"
    customer_email: str | None = None
    payment_token: str | None = None

    @property
    def resolved_items(self) -> list[dict[str, Any]]:
        """Accept either field name from the task context."""
        return self.items or self.cart_items or []

# app/handlers/ecommerce.py — the handler
from tasker_core.step_handler.functional import inputs, step_handler
from app.services.types import EcommerceOrderInput
from app.services import ecommerce as svc

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    return svc.validate_cart_items(inputs.resolved_items)

The @step_handler decorator registers this function under the name validate_cart — the exact string that must appear in the handler.callable field of the task template YAML. The @inputs decorator tells Tasker to extract the task context into an EcommerceOrderInput Pydantic model. The function body is a single service call.

Ruby

# app/services/types.rb — the contract
module Types
  module Ecommerce
    class OrderInput < Types::InputStruct
      attribute :cart_items, Types::Array.of(Types::Hash).optional
      attribute :customer_email, Types::String.optional
    end
  end
end

# app/handlers/ecommerce/step_handlers/validate_cart_handler.rb — the handler
module Ecommerce
  module StepHandlers
    extend TaskerCore::StepHandler::Functional

    ValidateCartHandler = step_handler(
      'Ecommerce::StepHandlers::ValidateCartHandler',
      inputs: Types::Ecommerce::OrderInput
    ) do |inputs:, context:|
      Ecommerce::Service.validate_cart_items(cart_items: inputs.cart_items)
    end
  end
end

Ruby uses step_handler as a method that takes a block. The inputs: keyword argument receives a Dry::Struct instance with typed attributes.

TypeScript

// src/services/types.ts — the contract
export interface CartItem {
  sku: string;
  name: string;
  price: number;
  quantity: number;
}

// src/handlers/ecommerce.ts — the handler
import { defineHandler } from '@tasker-systems/tasker';
import * as svc from '../services/ecommerce';

export const ValidateCartHandler = defineHandler(
  'Ecommerce.StepHandlers.ValidateCartHandler',
  { inputs: { cartItems: 'cart_items' } },
  async ({ cartItems }) => svc.validateCartItems(cartItems as CartItem[]),
);

TypeScript uses defineHandler as a factory function. The inputs config maps camelCase parameter names to snake_case YAML field names.

Rust

Rust uses the RustStepHandler trait directly — this is Rust’s only handler pattern. There is no DSL equivalent, by design.

#![allow(unused)]
fn main() {
use anyhow::Result;
use async_trait::async_trait;
use serde_json::json;
use std::time::Instant;
use tasker_shared::messaging::StepExecutionResult;
use tasker_shared::types::TaskSequenceStep;
use tasker_worker_rust::{success_result, RustStepHandler};
use tasker_worker_rust::step_handlers::StepHandlerConfig;

pub struct ProcessOrderHandler {
    config: StepHandlerConfig,
}

#[async_trait]
impl RustStepHandler for ProcessOrderHandler {
    fn new(config: StepHandlerConfig) -> Self {
        Self { config }
    }

    fn name(&self) -> &str {
        "process_order"
    }

    async fn call(
        &self,
        step_data: &TaskSequenceStep,
    ) -> Result<StepExecutionResult> {
        let start = Instant::now();
        let _input_data = &step_data.task.context;

        let result_data = json!({
            "processed": true,
            "handler": "process_order"
        });

        let duration_ms = start.elapsed().as_millis() as i64;

        Ok(success_result(
            step_data.workflow_step.workflow_step_uuid,
            result_data,
            duration_ms,
            None,
        ))
    }
}
}

Reading the DSL

Each language’s DSL has the same three concepts:

The handler function always receives context as its last parameter — a StepContext with execution metadata. Most handlers don’t need it directly, but it’s available for advanced patterns.

Registering Handlers

Handlers are resolved by matching the handler.callable field in task template YAML against the name you registered with the DSL. The mechanism is identical across all languages — the callable is an opaque key that must match between registration and template.

Each language follows its own naming convention:

The registry doesn’t enforce a format — any string works as long as the DSL registration and YAML callable match. Qualified names are recommended because they prevent collisions across modules and align with how each language’s class-based fallback resolver works. Short names like validate_cart are equally valid and resolve reliably through exact-match (the highest-priority resolver). See Handler Resolution for the full resolver chain architecture, fallback behavior, and conventions.

Class-Based Alternative

If you prefer class inheritance, all handler types support a class-based pattern where you extend StepHandler and implement call(context). See Class-Based Handlers for the full reference.

See It in Action

The example apps implement step handlers for four real-world workflows in all four languages. Compare the same handler across Rails, FastAPI, Bun, and Axum to see how each framework’s idioms map to the Tasker contract.

Next Steps

• Your First Workflow — Connect handlers into a multi-step DAG
• Language guides: Ruby | Python | TypeScript | Rust

Your First Workflow

This guide walks you through creating a complete workflow with multiple steps. We’ll use the e-commerce order processing pattern from the example apps, demonstrating parallel execution and typed dependency injection.

This walkthrough uses Python. See the Ruby, TypeScript, and Rust guides for language-specific examples.

What is a Workflow?

A Workflow is a directed acyclic graph (DAG) of steps defined in a task template YAML file. Steps execute when their dependencies are satisfied, enabling parallel execution where possible.

Example: Order Processing

Let’s build an order processing workflow with five steps. After validating the cart, payment processing and inventory reservation happen in parallel — they’re independent operations that don’t need each other’s results. Once both succeed, we create the order record and send the confirmation:

           ┌──────────────┐
           │ validate_cart │
           └──────┬───────┘
                  │
        ┌─────────┴─────────┐
        ▼                   ▼
┌──────────────┐   ┌──────────────┐
│   process    │   │   update     │
│   payment    │   │  inventory   │
└──────┬───────┘   └──────┬───────┘
        │                  │
        └────────┬─────────┘
                 ▼
        ┌──────────────┐
        │ create_order │
        └──────┬───────┘
               │
               ▼
        ┌──────────────┐
        │    send      │
        │ confirmation │
        └──────────────┘

This is a real-world pattern: payment authorization and inventory reservation are calls to different external systems. Running them in parallel reduces total checkout time. The order record isn’t created until both succeed, and the confirmation email isn’t sent until the order exists.

Step 1: Define the Task Template

Create a YAML file defining the workflow structure. You can generate a starter template with tasker-ctl:

tasker-ctl template generate task_template \
  --language python \
  --param name=EcommerceOrderProcessing \
  --param namespace=ecommerce \
  --param handler_callable=handlers.ecommerce.ValidateCartHandler

Then extend the generated single-step template into the full DAG:

# config/tasker/templates/ecommerce_order_processing.yaml
name: ecommerce_order_processing
namespace_name: ecommerce
version: "1.0.0"
description: "E-commerce checkout: validate → (payment ‖ inventory) → order → confirm"

steps:
  - name: validate_cart
    description: "Validate cart items, check availability, calculate totals"
    handler:
      callable: validate_cart
    dependencies: []
    retry:
      retryable: true
      max_attempts: 2
      backoff: exponential

  - name: process_payment
    description: "Authorize payment through payment gateway"
    handler:
      callable: process_payment
    dependencies:
      - validate_cart
    retry:
      retryable: true
      max_attempts: 3
      backoff: exponential

  - name: update_inventory
    description: "Reserve inventory for order items"
    handler:
      callable: update_inventory
    dependencies:
      - validate_cart
    retry:
      retryable: true
      max_attempts: 3
      backoff: exponential

  - name: create_order
    description: "Create order record from payment and inventory results"
    handler:
      callable: create_order
    dependencies:
      - process_payment
      - update_inventory
    retry:
      retryable: true
      max_attempts: 2
      backoff: exponential

  - name: send_confirmation
    description: "Send order confirmation email to customer"
    handler:
      callable: send_confirmation
    dependencies:
      - create_order
    retry:
      retryable: true
      max_attempts: 2
      backoff: exponential

input_schema:
  type: object
  required:
    - cart_items
    - customer_email
  properties:
    cart_items:
      type: array
      items:
        type: object
        required: [sku, name, quantity, unit_price]
        properties:
          sku:
            type: string
          name:
            type: string
          quantity:
            type: integer
          unit_price:
            type: number
    customer_email:
      type: string
      format: email
    payment_token:
      type: string

Key YAML Fields

How Dependencies Create the DAG

The dependencies field defines the execution graph:

• validate_cart has no dependencies — it runs first
• process_payment and update_inventory both depend only on validate_cart — they run in parallel once validation completes
• create_order depends on both process_payment and update_inventory — it waits for both to complete (convergence point)
• send_confirmation depends on create_order — it runs last

Tasker resolves these dependencies automatically. You declare what depends on what, and the engine figures out what can run in parallel.

YAML Dependencies vs Handler Dependencies

The YAML dependencies field and the handler’s @depends_on decorator serve different purposes:

• YAML dependencies define the DAG shape — which steps must complete before this step starts. These are proximal (direct predecessors only). create_order lists process_payment and update_inventory because it must wait for both.

• Handler @depends_on declares which step results the handler needs injected as typed parameters. These can reference any ancestor step — not just direct predecessors. Tasker makes all ancestor results available in the step context.

In the create_order handler below, notice that @depends_on references validate_cart even though the YAML only lists process_payment and update_inventory as dependencies. The handler can access validate_cart’s result because it’s a transitive ancestor — Tasker has already executed it earlier in the DAG.

Step 2: Define Your Types

Before writing handlers, define the types that describe what flows between steps. These are Pydantic models — the same types the DSL uses to inject inputs and dependency results:

# app/services/types.py
from pydantic import BaseModel
from typing import Any

class EcommerceOrderInput(BaseModel):
    items: list[dict[str, Any]] | None = None        # submitted as "items"
    cart_items: list[dict[str, Any]] | None = None    # or "cart_items"
    customer_email: str | None = None
    payment_token: str | None = None

    @property
    def resolved_items(self) -> list[dict[str, Any]]:
        """Accept either field name from the task context."""
        return self.items or self.cart_items or []

class EcommerceValidateCartResult(BaseModel):
    validated_items: list[dict[str, Any]] | None = None
    item_count: int | None = None
    subtotal: float | None = None
    tax: float | None = None
    total: float | None = None

class EcommerceProcessPaymentResult(BaseModel):
    payment_id: str | None = None
    transaction_id: str | None = None
    amount_charged: float | None = None
    status: str | None = None

class EcommerceUpdateInventoryResult(BaseModel):
    total_items_reserved: int | None = None
    inventory_log_id: str | None = None

class EcommerceCreateOrderResult(BaseModel):
    order_id: str | None = None
    customer_email: str | None = None
    total: float | None = None
    status: str | None = None

All fields are optional with None defaults. This is intentional — task context may not include every field, and upstream step results may vary. The type system provides structure and IDE autocomplete without brittle required-field failures.

Step 3: Implement Handlers

With types defined, the handlers are short — each one declares what it receives and delegates to a service function:

# app/handlers/ecommerce.py
from tasker_core.step_handler.functional import depends_on, inputs, step_handler
from tasker_core.types import StepContext
from app.services import ecommerce as svc
from app.services.types import (
    EcommerceCreateOrderResult,
    EcommerceOrderInput,
    EcommerceProcessPaymentResult,
    EcommerceUpdateInventoryResult,
    EcommerceValidateCartResult,
)

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    return svc.validate_cart_items(inputs.resolved_items)

@step_handler("process_payment")
@depends_on(cart_result=("validate_cart", EcommerceValidateCartResult))
@inputs(EcommerceOrderInput)
def process_payment(
    cart_result: EcommerceValidateCartResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.process_payment(
        payment_token=inputs.payment_token,
        total=cart_result.total or 0.0,
    )

@step_handler("update_inventory")
@depends_on(cart_result=("validate_cart", EcommerceValidateCartResult))
def update_inventory(cart_result: EcommerceValidateCartResult, context: StepContext):
    return svc.update_inventory(cart_result.validated_items or [])

@step_handler("create_order")
@depends_on(
    cart_result=("validate_cart", EcommerceValidateCartResult),
    payment_result=("process_payment", EcommerceProcessPaymentResult),
    inventory_result=("update_inventory", EcommerceUpdateInventoryResult),
)
@inputs(EcommerceOrderInput)
def create_order(
    cart_result: EcommerceValidateCartResult,
    payment_result: EcommerceProcessPaymentResult,
    inventory_result: EcommerceUpdateInventoryResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.create_order(
        cart=cart_result, payment=payment_result,
        inventory=inventory_result, customer_email=inputs.customer_email,
    )

@step_handler("send_confirmation")
@depends_on(order_result=("create_order", EcommerceCreateOrderResult))
@inputs(EcommerceOrderInput)
def send_confirmation(
    order_result: EcommerceCreateOrderResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.send_confirmation(
        order=order_result, customer_email=inputs.customer_email,
    )

That’s the entire handler file — five handlers in about 50 lines. The service functions (svc.validate_cart_items, svc.process_payment, etc.) contain your actual business logic. Tasker doesn’t care what happens inside them — it cares about the handler’s typed signature and the result it returns.

Anatomy of a Handler with Dependencies

Look at create_order — the convergence point where three parallel branches meet:

@step_handler("create_order")
@depends_on(
    cart_result=("validate_cart", EcommerceValidateCartResult),       # ① upstream step + type
    payment_result=("process_payment", EcommerceProcessPaymentResult), # ② another upstream step
    inventory_result=("update_inventory", EcommerceUpdateInventoryResult),
)
@inputs(EcommerceOrderInput)                                          # ③ task context
def create_order(
    cart_result: EcommerceValidateCartResult,   # injected as typed Pydantic model
    payment_result: EcommerceProcessPaymentResult,
    inventory_result: EcommerceUpdateInventoryResult,
    inputs: EcommerceOrderInput,                # task context as typed model
    context: StepContext,                       # execution metadata
):
    return svc.create_order(...)               # ④ delegate to service

1. Each @depends_on entry maps a parameter name to a ("step_name", ResultModel) tuple
2. Tasker resolves the upstream step’s result dict and deserializes it into the Pydantic model
3. @inputs does the same for the task context
4. The handler function receives fully typed objects and passes them to the service

Step 4: Submit a Task

Submit a task via the REST API:

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ecommerce_order_processing",
    "namespace": "ecommerce",
    "version": "1.0.0",
    "initiator": "api:checkout",
    "source_system": "web",
    "reason": "New order received",
    "context": {
      "customer_email": "customer@example.com",
      "payment_token": "tok_test_success",
      "cart_items": [
        {"sku": "WIDGET-A", "name": "Widget A", "quantity": 2, "unit_price": 29.99},
        {"sku": "GADGET-B", "name": "Gadget B", "quantity": 1, "unit_price": 49.99}
      ]
    }
  }'

Or with tasker-ctl:

tasker-ctl task create \
  --name ecommerce_order_processing \
  --namespace ecommerce \
  --input '{"customer_email": "customer@example.com", "cart_items": [{"sku": "WIDGET-A", "name": "Widget A", "quantity": 2, "unit_price": 29.99}]}'

Execution Flow

When this task runs:

1. validate_cart executes first (no dependencies)
2. process_payment and update_inventory execute in parallel (both depend only on validate_cart)
3. create_order executes after both parallel steps complete (convergence)
4. send_confirmation executes after create_order completes

The total execution time is determined by the longest path through the DAG, not the sum of all steps. If payment takes 2 seconds and inventory takes 1 second, step 3 begins at the 2-second mark — the inventory result is already waiting.

Your Services, Tasker’s Orchestration

Notice what the handlers don’t contain: no tax calculations, no payment gateway logic, no inventory reservation algorithms. That business logic lives in your service layer (app/services/ecommerce.py), where it can be tested independently and reused outside of Tasker.

The handlers are thin wrappers that declare their typed signature and delegate. Tasker brings workflow orchestration to your existing codebase — it manages the DAG, sequencing, retries, and error classification. Your services do what they’ve always done.

See It in Action

The example apps implement this e-commerce workflow (and three others) in all four languages — Rails, FastAPI, Bun, and Axum. Each app is a fully working project you can clone and run with Docker Compose.

The example apps also include more complex DAG patterns:

• Data Pipeline — Three parallel extract branches, each feeding its own transform, converging at aggregation (8 steps)
• Microservices — User registration with parallel billing and preferences setup (5 steps, diamond pattern)
• Cross-Namespace — Customer success workflow that delegates to a payments namespace (namespace isolation)

Next Steps

• Language guides: Ruby | Python | TypeScript | Rust
• Architecture Overview — Understand lifecycle actors and DAG execution
• Handler Types — API, Decision, and Batchable handler patterns

Building with Ruby

This guide covers building Tasker step handlers with Ruby using the tasker-rb gem in a Rails application.

Quick Start

Add the gem to your Gemfile:

gem 'tasker-rb'

Generate a step handler with tasker-ctl:

tasker-ctl template generate step_handler \
  --language ruby \
  --param name=ValidateCart \
  --param module_name=Ecommerce

This creates a DSL-style handler with typed inputs that delegates to a service method.

Writing a Handler (DSL)

Every handler follows the three-layer pattern: type definition, handler declaration, service delegation.

# app/services/types.rb — the contract
module Types
  module Ecommerce
    class OrderInput < Types::InputStruct
      attribute :cart_items, Types::Array.optional
      attribute :customer_email, Types::String.optional
      attribute :payment_info, Types::Hash.optional
      attribute :shipping_address, Types::Hash.optional
    end
  end
end

# app/handlers/ecommerce/step_handlers/validate_cart_handler.rb — the handler
module Ecommerce
  module StepHandlers
    extend TaskerCore::StepHandler::Functional

    ValidateCartHandler = step_handler(
      'Ecommerce::StepHandlers::ValidateCartHandler',
      inputs: Types::Ecommerce::OrderInput
    ) do |inputs:, context:|
      Ecommerce::Service.validate_cart_items(
        cart_items: inputs.cart_items,
        customer_email: inputs.customer_email,
      )
    end
  end
end

The step_handler method registers a handler and takes a block. The inputs: keyword argument receives a Dry::Struct instance with typed, optional attributes. The block body is a single service call.

Type System

Ruby handlers use Dry::Struct for both input and result types.

Input types extend Types::InputStruct — a base class where all attributes are optional and omittable, so missing keys don’t raise:

module Types
  module Ecommerce
    class OrderInput < Types::InputStruct
      attribute :cart_items, Types::Array.optional
      attribute :customer_email, Types::String.optional
      attribute :payment_info, Types::Hash.optional
      attribute :shipping_address, Types::Hash.optional
    end
  end
end

Result types extend Types::ResultStruct — similar to InputStruct but describing what a handler returns (used by downstream depends_on):

module Types
  module Ecommerce
    class ValidateCartResult < Types::ResultStruct
      attribute :validated_items, Types::Array
      attribute :item_count, Types::Integer
      attribute :subtotal, Types::Float
      attribute :tax, Types::Float
      attribute :total, Types::Float
    end
  end
end

Both InputStruct and ResultStruct support string and symbol key access (e.g., result['user_id'] and result[:user_id]) and nested access via dig.

Accessing Task Context

The inputs: config extracts the full task context into a typed Dry::Struct instance. Fields are matched by name from the submitted JSON:

ValidateCartHandler = step_handler(
  'Ecommerce::StepHandlers::ValidateCartHandler',
  inputs: Types::Ecommerce::OrderInput
) do |inputs:, context:|
  # inputs.cart_items, inputs.customer_email, etc. are typed attributes
  Ecommerce::Service.validate_cart_items(cart_items: inputs.cart_items)
end

The context: keyword provides execution metadata (task UUID, step UUID, step config) but most handlers don’t need it directly.

Working with Dependencies

The depends_on: config injects typed results from upstream steps. Each entry maps a keyword argument name to a ['step_name', ResultModel] pair:

ProcessPaymentHandler = step_handler(
  'Ecommerce::StepHandlers::ProcessPaymentHandler',
  depends_on: { cart_total: ['validate_cart', Types::Ecommerce::ValidateCartResult] },
  inputs: Types::Ecommerce::OrderInput
) do |cart_total:, inputs:, context:|
  Ecommerce::Service.process_payment(
    payment_info: inputs.payment_info,
    total: cart_total&.total,
  )
end

Handlers can reference any ancestor step in the DAG — not just direct predecessors. Here’s a convergence handler that accesses three upstream steps:

CreateOrderHandler = step_handler(
  'Ecommerce::StepHandlers::CreateOrderHandler',
  depends_on: {
    cart_validation: ['validate_cart', Types::Ecommerce::ValidateCartResult],
    payment_result: ['process_payment', Types::Ecommerce::ProcessPaymentResult],
    inventory_result: ['update_inventory', Types::Ecommerce::UpdateInventoryResult],
  },
  inputs: Types::Ecommerce::OrderInput
) do |cart_validation:, payment_result:, inventory_result:, inputs:, context:|
  Ecommerce::Service.create_order(
    cart_validation: cart_validation,
    payment_result: payment_result,
    inventory_result: inventory_result,
    customer_email: inputs.customer_email,
    shipping_address: inputs.shipping_address,
  )
end

Multi-Step Example: Data Pipeline

The data pipeline workflow demonstrates a parallel DAG — three independent extract branches, each feeding its own transform, converging at aggregation:

extract_sales    extract_inventory    extract_customers
     │                  │                    │
     ▼                  ▼                    ▼
transform_sales  transform_inventory  transform_customers
     │                  │                    │
     └──────────────────┼────────────────────┘
                        ▼
               aggregate_metrics
                        │
                        ▼
              generate_insights

Ruby handlers follow the same concise pattern:

module DataPipeline
  module StepHandlers
    extend TaskerCore::StepHandler::Functional

    # Extract — no dependencies, runs in parallel
    ExtractSalesDataHandler = step_handler(
      'DataPipeline::StepHandlers::ExtractSalesDataHandler',
      inputs: Types::DataPipeline::PipelineInput
    ) do |inputs:, context:|
      DataPipeline::Service.extract_sales_data(
        source: inputs.source,
        date_range_start: inputs.date_range_start,
      )
    end

    # Transform — depends on one extract branch
    TransformSalesHandler = step_handler(
      'DataPipeline::StepHandlers::TransformSalesHandler',
      depends_on: { sales_data: ['extract_sales_data', Types::DataPipeline::ExtractSalesResult] }
    ) do |sales_data:, context:|
      DataPipeline::Service.transform_sales(sales_data: sales_data)
    end

    # Aggregate — converges three transform branches
    AggregateMetricsHandler = step_handler(
      'DataPipeline::StepHandlers::AggregateMetricsHandler',
      depends_on: {
        sales: ['transform_sales', Types::DataPipeline::TransformSalesResult],
        inventory: ['transform_inventory', Types::DataPipeline::TransformInventoryResult],
        customers: ['transform_customers', Types::DataPipeline::TransformCustomersResult],
      }
    ) do |sales:, inventory:, customers:, context:|
      DataPipeline::Service.aggregate_metrics(
        sales: sales, inventory: inventory, customers: customers,
      )
    end
  end
end

Error Handling

Raise typed exceptions to control retry behavior:

# Permanent error — will NOT be retried
raise TaskerCore::Errors::PermanentError.new(
  'Payment declined: insufficient funds',
  error_code: 'PAYMENT_DECLINED'
)

# Retryable error — will be retried per the step's retry config
raise TaskerCore::Errors::RetryableError.new(
  'Payment gateway temporarily unavailable'
)

Error codes (like PAYMENT_DECLINED, EMPTY_CART) are included in the step result for observability and debugging.

Testing

DSL handlers are constants holding callable blocks — test by invoking the service functions directly or by using RSpec mocks:

RSpec.describe 'Ecommerce::StepHandlers::ValidateCartHandler' do
  let(:inputs) do
    Types::Ecommerce::OrderInput.new(
      cart_items: [{ 'sku' => 'SKU-001', 'name' => 'Widget', 'quantity' => 2, 'unit_price' => 29.99 }],
      customer_email: 'test@example.com'
    )
  end

  it 'delegates to the service' do
    expect(Ecommerce::Service).to receive(:validate_cart_items)
      .with(cart_items: inputs.cart_items, customer_email: inputs.customer_email)
      .and_return({ validated_items: [], total: 64.79 })

    context = instance_double(TaskerCore::Types::StepContext)
    result = Ecommerce::StepHandlers::ValidateCartHandler.call(inputs: inputs, context: context)

    expect(result[:total]).to eq(64.79)
  end
end

For handlers with dependencies, construct result models directly:

let(:cart) { Types::Ecommerce::ValidateCartResult.new(total: 64.79, validated_items: []) }
let(:payment) { Types::Ecommerce::ProcessPaymentResult.new(payment_id: 'pay_abc') }
let(:inventory) { Types::Ecommerce::UpdateInventoryResult.new(inventory_log_id: 'log_123') }
let(:inputs) { Types::Ecommerce::OrderInput.new(customer_email: 'test@example.com') }

it 'creates order from upstream data' do
  expect(Ecommerce::Service).to receive(:create_order)
    .and_return({ order_id: 'ORD-001' })

  context = instance_double(TaskerCore::Types::StepContext)
  result = Ecommerce::StepHandlers::CreateOrderHandler.call(
    cart_validation: cart, payment_result: payment,
    inventory_result: inventory, inputs: inputs, context: context,
  )

  expect(result[:order_id]).to eq('ORD-001')
end

Because handlers delegate to service methods, you can also test the services directly without any Tasker infrastructure.

Handler Variants

API Handler

Adds HTTP client methods with built-in error classification. Uses TaskerCore::StepHandler::Mixins::API with the class-based pattern. See Class-Based Handlers — API Handler.

Decision Handler

Adds workflow routing with decision_success() for activating downstream step sets. Uses TaskerCore::StepHandler::Mixins::Decision with the class-based pattern. See Conditional Workflows.

Batchable Handler

Adds batch processing with Analyzer/Worker pattern using TaskerCore::StepHandler::Mixins::Batchable. See Class-Based Handlers — Batchable Handler and Batch Processing.

Class-Based Alternative

If you prefer class inheritance, all handler types support a class-based pattern where you inherit from TaskerCore::StepHandler::Base and implement call(context). See Class-Based Handlers for the full reference.

Next Steps

• Your First Workflow — Build a multi-step DAG end-to-end
• Architecture — System design details
• Rails example app — Complete working implementation

Building with Python

This guide covers building Tasker step handlers with Python using the tasker_core package in a FastAPI application.

Quick Start

Install the package:

pip install tasker-py
# Or with uv (recommended)
uv add tasker-py

Generate a step handler with tasker-ctl:

tasker-ctl template generate step_handler \
  --language python \
  --param name=ValidateCart \
  --param module_name=handlers.ecommerce

This creates a DSL-style handler with typed inputs that delegates to a service function.

Writing a Handler (DSL)

Every handler follows the three-layer pattern: type definition, handler declaration, service delegation.

# app/services/types.py — the contract
from pydantic import BaseModel
from typing import Any

class EcommerceOrderInput(BaseModel):
    items: list[dict[str, Any]] | None = None
    cart_items: list[dict[str, Any]] | None = None
    customer_email: str | None = None
    payment_token: str | None = None

    @property
    def resolved_items(self) -> list[dict[str, Any]]:
        """Accept either field name from the task context."""
        return self.items or self.cart_items or []

# app/handlers/ecommerce.py — the handler
from tasker_core.step_handler.functional import inputs, step_handler
from app.services.types import EcommerceOrderInput
from app.services import ecommerce as svc

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    return svc.validate_cart_items(inputs.resolved_items)

The @step_handler decorator registers this function as the handler for the validate_cart step. The @inputs decorator tells Tasker to extract the task context into a Pydantic model. The function body is a single service call.

Type System

Python handlers use Pydantic BaseModel for both input and result types. The DSL deserializes JSON into these models automatically.

Input types receive the task context:

class EcommerceOrderInput(BaseModel):
    items: list[dict[str, Any]] | None = None
    cart_items: list[dict[str, Any]] | None = None
    payment_token: str | None = None
    customer_email: str | None = None

    @property
    def resolved_items(self) -> list[dict[str, Any]]:
        """Accept either field name from the task context."""
        return self.items or self.cart_items or []

Result types describe what a handler returns (used by downstream @depends_on):

class EcommerceValidateCartResult(BaseModel):
    validated_items: list[dict[str, Any]] | None = None
    item_count: int | None = None
    subtotal: float | None = None
    tax: float | None = None
    total: float | None = None

All fields are optional with None defaults. This is intentional — task context may not include every field, and upstream results may vary. The type system provides structure and IDE autocomplete without brittle required-field failures.

Validation with @model_validator:

from pydantic import model_validator

class ValidateRefundRequestInput(BaseModel):
    ticket_id: str | None = None
    order_ref: str | None = None
    refund_amount: float | None = None

    @property
    def resolved_ticket_id(self) -> str | None:
        return self.ticket_id or self.order_ref

    @model_validator(mode='after')
    def check_required_fields(self) -> 'ValidateRefundRequestInput':
        if not self.resolved_ticket_id:
            raise PermanentError(
                message="ticket_id or order_ref is required",
                error_code="MISSING_TICKET_ID",
            )
        return self

Accessing Task Context

The @inputs(Model) decorator extracts the full task context into a typed Pydantic model. Fields are matched by name from the submitted JSON:

@step_handler("validate_cart")
@inputs(EcommerceOrderInput)
def validate_cart(inputs: EcommerceOrderInput, context: StepContext):
    # inputs.cart_items, inputs.customer_email, etc. are typed fields
    return svc.validate_cart_items(inputs.resolved_items)

The context parameter provides execution metadata (task UUID, step UUID, step config) but most handlers don’t need it directly.

Working with Dependencies

The @depends_on decorator injects typed results from upstream steps. Each entry maps a parameter name to a ("step_name", ResultModel) tuple:

@step_handler("process_payment")
@depends_on(cart_result=("validate_cart", EcommerceValidateCartResult))
@inputs(EcommerceOrderInput)
def process_payment(
    cart_result: EcommerceValidateCartResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.process_payment(
        payment_token=inputs.payment_token,
        total=cart_result.total or 0.0,
    )

Handlers can reference any ancestor step in the DAG — not just direct predecessors. Tasker makes all ancestor results available. Here’s a convergence handler that accesses three upstream steps:

@step_handler("create_order")
@depends_on(
    cart_result=("validate_cart", EcommerceValidateCartResult),
    payment_result=("process_payment", EcommerceProcessPaymentResult),
    inventory_result=("update_inventory", EcommerceUpdateInventoryResult),
)
@inputs(EcommerceOrderInput)
def create_order(
    cart_result: EcommerceValidateCartResult,
    payment_result: EcommerceProcessPaymentResult,
    inventory_result: EcommerceUpdateInventoryResult,
    inputs: EcommerceOrderInput,
    context: StepContext,
):
    return svc.create_order(
        cart=cart_result, payment=payment_result,
        inventory=inventory_result, customer_email=inputs.customer_email,
    )

Multi-Step Example: Data Pipeline

The data pipeline workflow demonstrates a parallel DAG — three independent extract branches, each feeding its own transform, converging at aggregation:

extract_sales    extract_inventory    extract_customers
     │                  │                    │
     ▼                  ▼                    ▼
transform_sales  transform_inventory  transform_customers
     │                  │                    │
     └──────────────────┼────────────────────┘
                        ▼
               aggregate_metrics
                        │
                        ▼
              generate_insights

The handlers are just as concise as the e-commerce ones:

from app.services import data_pipeline as svc
from app.services.types import (
    DataPipelineInput,
    PipelineExtractSalesResult,
    PipelineTransformSalesResult,
    PipelineTransformInventoryResult,
    PipelineTransformCustomersResult,
    PipelineAggregateMetricsResult,
)

# Extract — no dependencies, runs in parallel
@step_handler("extract_sales_data")
@inputs(DataPipelineInput)
def extract_sales_data(inputs: DataPipelineInput, context: StepContext):
    return svc.extract_sales_data(
        source=inputs.source,
        date_range_start=inputs.date_range_start,
        date_range_end=inputs.date_range_end,
        granularity=inputs.granularity,
    )

# Transform — depends on one extract branch
@step_handler("transform_sales")
@depends_on(sales_data=("extract_sales_data", PipelineExtractSalesResult))
def transform_sales(sales_data: PipelineExtractSalesResult, context: StepContext):
    return svc.transform_sales(sales_data=sales_data)

# Aggregate — converges three transform branches
@step_handler("aggregate_metrics")
@depends_on(
    sales_transform=("transform_sales", PipelineTransformSalesResult),
    traffic_transform=("transform_inventory", PipelineTransformInventoryResult),
    inventory_transform=("transform_customers", PipelineTransformCustomersResult),
)
def aggregate_metrics(
    sales_transform: PipelineTransformSalesResult,
    traffic_transform: PipelineTransformInventoryResult,
    inventory_transform: PipelineTransformCustomersResult,
    context: StepContext,
):
    return svc.aggregate_metrics(
        sales_transform=sales_transform,
        traffic_transform=traffic_transform,
        inventory_transform=inventory_transform,
    )

Eight handlers, eight service delegations. The pipeline DAG runs three extract steps in parallel, feeds each into a transform, then converges at aggregation and insight generation.

Error Handling

Raise PermanentError or RetryableError from your handler or service functions:

from tasker_core.errors import PermanentError, RetryableError

# Non-retryable validation failure
raise PermanentError(
    message="Payment declined: insufficient funds",
    error_code="PAYMENT_DECLINED",
)

# Retryable transient failure
raise RetryableError(
    message="Payment gateway returned an error, will retry",
    error_code="GATEWAY_ERROR",
)

Pydantic @model_validator errors are also caught and converted to PermanentError automatically — invalid input data won’t be retried.

Testing

DSL handlers are plain functions — test them by calling the function directly with mocked inputs:

from unittest.mock import MagicMock, patch

def test_validate_cart():
    context = MagicMock()
    # Mock the inputs that @inputs would inject
    inputs = EcommerceOrderInput(
        cart_items=[{"sku": "SKU-001", "name": "Widget", "quantity": 2, "unit_price": 29.99}]
    )

    with patch("app.services.ecommerce.validate_cart_items") as mock_svc:
        mock_svc.return_value = {"validated_items": [], "total": 64.79}
        result = validate_cart(inputs=inputs, context=context)

    mock_svc.assert_called_once()
    assert result["total"] == 64.79

For handlers with dependencies, construct the result models directly:

def test_create_order():
    context = MagicMock()
    cart = EcommerceValidateCartResult(total=64.79, validated_items=[])
    payment = EcommerceProcessPaymentResult(payment_id="pay_abc", transaction_id="txn_xyz")
    inventory = EcommerceUpdateInventoryResult(inventory_log_id="log_123")
    inputs = EcommerceOrderInput(customer_email="test@example.com")

    with patch("app.services.ecommerce.create_order") as mock_svc:
        mock_svc.return_value = {"order_id": "ORD-001"}
        result = create_order(
            cart_result=cart, payment_result=payment,
            inventory_result=inventory, inputs=inputs, context=context,
        )

    assert result["order_id"] == "ORD-001"

Because handlers delegate to service functions, you can also test the services directly without any Tasker infrastructure.

Handler Variants

API Handler

Adds HTTP client methods with built-in error classification. Currently uses the class-based pattern with APIMixin. See Class-Based Handlers — API Handler.

Decision Handler

Adds workflow routing. The DSL provides @decision_handler:

from tasker_core.step_handler.functional import decision_handler

@decision_handler("order_routing")
def order_routing(context: StepContext):
    order_type = context.get_input("order_type")
    if order_type == "premium":
        return ["validate_premium", "process_premium"]
    return ["standard_processing"]

See Conditional Workflows for decision handler patterns.

Batchable Handler

Adds batch processing for splitting large workloads. Uses the class-based pattern due to its stateful nature (cursor management, batch context). See Class-Based Handlers — Batchable Handler and Batch Processing.

Class-Based Alternative

If you prefer class inheritance, all handler types support a class-based pattern where you extend StepHandler and implement call(context). See Class-Based Handlers for the full reference.

Next Steps

• Your First Workflow — Build a multi-step DAG end-to-end
• Architecture — System design details
• FastAPI example app — Complete working implementation

Building with TypeScript

This guide covers building Tasker step handlers with TypeScript using the @tasker-systems/tasker package in a Bun application.

Quick Start

Install the package:

bun add @tasker-systems/tasker
# Or with npm
npm install @tasker-systems/tasker

Generate a step handler with tasker-ctl:

tasker-ctl template generate step_handler \
  --language typescript \
  --param name=ValidateCart

This creates a DSL-style handler with typed inputs that delegates to a service function.

Writing a Handler (DSL)

Every handler follows the three-layer pattern: type definition, handler declaration, service delegation.

// src/services/types.ts — the contract
export interface CartItem {
  sku: string;
  name: string;
  price: number;
  quantity: number;
}

// src/handlers/ecommerce.ts — the handler
import { defineHandler } from '@tasker-systems/tasker';
import type { CartItem } from '../services/types';
import * as svc from '../services/ecommerce';

export const ValidateCartHandler = defineHandler(
  'Ecommerce.StepHandlers.ValidateCartHandler',
  { inputs: { cartItems: 'cart_items' } },
  async ({ cartItems }) => svc.validateCartItems(cartItems as CartItem[] | undefined),
);

The defineHandler factory registers a handler by name. The inputs config maps camelCase parameter names to snake_case YAML field names. The async callback receives typed arguments and delegates to a service function.

Type System

TypeScript handlers use standard interfaces for both input and result types. The DSL injects values from the task context and upstream results as plain objects matching these interfaces.

Input types:

export interface CartItem {
  sku: string;
  name: string;
  price: number;
  quantity: number;
}

export interface PaymentInfo {
  method: string;
  card_last_four?: string;
  token: string;
  amount: number;
}

Result types describe what a handler returns (used by downstream depends):

export interface EcommerceValidateCartResult {
  [key: string]: unknown;
  validated_items: CartItem[];
  item_count: number;
  subtotal: number;
  tax: number;
  total: number;
}

export interface EcommerceProcessPaymentResult {
  [key: string]: unknown;
  payment_id: string;
  transaction_id: string;
  amount_charged: number;
  status: string;
}

The [key: string]: unknown index signature allows result objects to carry additional fields without type errors when accessing known fields.

Accessing Task Context

The inputs config in defineHandler extracts fields from the task context. Each entry maps a camelCase parameter name to the snake_case field name in the submitted JSON:

export const ValidateCartHandler = defineHandler(
  'Ecommerce.StepHandlers.ValidateCartHandler',
  { inputs: { cartItems: 'cart_items' } },
  async ({ cartItems }) => svc.validateCartItems(cartItems as CartItem[] | undefined),
);

The callback receives cartItems directly — no need to parse raw JSON.

Working with Dependencies

The depends config injects results from upstream steps. Each entry maps a camelCase parameter name to the upstream step name:

export const ProcessPaymentHandler = defineHandler(
  'Ecommerce.StepHandlers.ProcessPaymentHandler',
  {
    depends: { cartResult: 'validate_cart' },
    inputs: { paymentInfo: 'payment_info' },
  },
  async ({ cartResult, paymentInfo }) =>
    svc.processPayment(
      cartResult as Record<string, unknown>,
      paymentInfo as PaymentInfo | undefined,
    ),
);

Handlers can reference any ancestor step in the DAG — not just direct predecessors. Here’s a convergence handler that accesses three upstream steps plus task inputs:

export const CreateOrderHandler = defineHandler(
  'Ecommerce.StepHandlers.CreateOrderHandler',
  {
    depends: {
      cartResult: 'validate_cart',
      paymentResult: 'process_payment',
      inventoryResult: 'update_inventory',
    },
    inputs: { customerEmail: 'customer_email' },
  },
  async ({ cartResult, paymentResult, inventoryResult, customerEmail }) =>
    svc.createOrder(
      cartResult as Record<string, unknown>,
      paymentResult as Record<string, unknown>,
      inventoryResult as Record<string, unknown>,
      customerEmail as string | undefined,
    ),
);

Multi-Step Example: Data Pipeline

The data pipeline workflow demonstrates a parallel DAG — three independent extract branches, each feeding its own transform, converging at aggregation:

extract_sales    extract_inventory    extract_customers
     │                  │                    │
     ▼                  ▼                    ▼
transform_sales  transform_inventory  transform_customers
     │                  │                    │
     └──────────────────┼────────────────────┘
                        ▼
               aggregate_metrics
                        │
                        ▼
              generate_insights

TypeScript handlers follow the same concise pattern:

import { defineHandler } from '@tasker-systems/tasker';
import type {
  PipelineExtractSalesResult,
  PipelineTransformSalesResult,
  PipelineTransformInventoryResult,
  PipelineTransformCustomersResult,
} from '../services/types';
import * as svc from '../services/data_pipeline';

// Extract — no dependencies, runs in parallel
export const ExtractSalesDataHandler = defineHandler(
  'DataPipeline.StepHandlers.ExtractSalesDataHandler',
  { inputs: { source: 'source', dateRangeStart: 'date_range_start' } },
  async ({ source, dateRangeStart }) =>
    svc.extractSalesData(source as string, dateRangeStart as string | undefined),
);

// Transform — depends on one extract branch
export const TransformSalesHandler = defineHandler(
  'DataPipeline.StepHandlers.TransformSalesHandler',
  { depends: { salesData: 'extract_sales_data' } },
  async ({ salesData }) =>
    svc.transformSales(salesData as PipelineExtractSalesResult),
);

// Aggregate — converges three transform branches
export const AggregateMetricsHandler = defineHandler(
  'DataPipeline.StepHandlers.AggregateMetricsHandler',
  {
    depends: {
      salesTransform: 'transform_sales',
      inventoryTransform: 'transform_inventory',
      customersTransform: 'transform_customers',
    },
  },
  async ({ salesTransform, inventoryTransform, customersTransform }) =>
    svc.aggregateMetrics(
      salesTransform as PipelineTransformSalesResult,
      inventoryTransform as PipelineTransformInventoryResult,
      customersTransform as PipelineTransformCustomersResult,
    ),
);

Error Handling

Throw PermanentError or RetryableError from your handler or service functions:

import { PermanentError, RetryableError } from '@tasker-systems/tasker';

// Non-retryable validation failure
throw new PermanentError('Payment declined: insufficient funds', 'PAYMENT_DECLINED');

// Retryable transient failure
throw new RetryableError('Payment gateway temporarily unavailable', 'GATEWAY_ERROR');

Testing

DSL handlers are exported constants — test them with Vitest by calling the handler’s async callback directly, or by testing the service functions:

import { describe, it, expect, vi } from 'vitest';
import * as svc from '../services/ecommerce';

describe('ValidateCartHandler', () => {
  it('delegates to service', async () => {
    const mockResult = { validated_items: [], total: 64.79 };
    vi.spyOn(svc, 'validateCartItems').mockResolvedValue(mockResult);

    const cartItems = [{ sku: 'SKU-001', name: 'Widget', price: 29.99, quantity: 2 }];
    const result = await svc.validateCartItems(cartItems);

    expect(result.total).toBe(64.79);
  });
});

For handlers with dependencies, test the service functions with typed arguments:

describe('CreateOrderHandler', () => {
  it('creates order from upstream data', async () => {
    const mockResult = { order_id: 'ORD-001' };
    vi.spyOn(svc, 'createOrder').mockResolvedValue(mockResult);

    const result = await svc.createOrder(
      { total: 64.79, validated_items: [] },
      { payment_id: 'pay_abc', transaction_id: 'txn_xyz' },
      { inventory_log_id: 'log_123' },
      'test@example.com',
    );

    expect(result.order_id).toBe('ORD-001');
  });
});

Because handlers delegate to service functions, you can test the services directly without any Tasker infrastructure.

Handler Variants

API Handler

Adds HTTP client methods with error classification using the native fetch API. Extends ApiHandler with the class-based pattern. See Class-Based Handlers — API Handler.

Decision Handler

Adds workflow routing with decisionSuccess() for activating downstream step sets. Extends DecisionHandler with the class-based pattern. See Conditional Workflows.

Batchable Handler

Adds batch processing with Analyzer/Worker pattern using BatchableStepHandler. See Class-Based Handlers — Batchable Handler and Batch Processing.

Class-Based Alternative

If you prefer class inheritance, all handler types support a class-based pattern where you extend StepHandler and implement async call(context). See Class-Based Handlers for the full reference.

Next Steps

• Your First Workflow — Build a multi-step DAG end-to-end
• Architecture — System design details
• Bun example app — Complete working implementation

Building with Rust

This guide covers building Tasker step handlers with native Rust using the tasker-worker and tasker-shared crates in an Axum application.

Quick Start

Add dependencies to your Cargo.toml:

[dependencies]
tasker-worker = "0.1"
tasker-shared = "0.1"
tasker-client = "0.1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
async-trait = "0.1"

Generate a step handler with tasker-ctl:

tasker-ctl template generate step_handler \
  --language rust \
  --param name=ValidateCart

This creates a handler implementing the StepHandler trait.

No DSL equivalent: Rust uses the StepHandler trait directly — the trait IS the pattern. Python, Ruby, and TypeScript have DSL wrappers for ergonomics, but Rust’s trait system provides the same structure natively. For the DSL approach in other languages, see Python, Ruby, or TypeScript.

Writing a Step Handler

The Axum example app uses plain functions wrapped in a handler registry. This is the recommended pattern for Rust handlers — write business logic as standalone functions, then register them with the worker.

Standalone Functions

Each handler function takes the task context and/or dependency results and returns Result<Value, String>:

#![allow(unused)]
fn main() {
use serde_json::{json, Value};
use std::collections::HashMap;

/// Step 1: Validate cart items, calculate totals.
/// No dependencies — receives task context only.
pub fn validate_cart(context: &Value) -> Result<Value, String> {
    let cart_items: Vec<CartItem> =
        serde_json::from_value(context.get("cart_items").cloned().unwrap_or(json!([])))
            .map_err(|e| format!("Invalid cart_items format: {}", e))?;

    if cart_items.is_empty() {
        return Err("Cart cannot be empty".to_string());
    }

    // Business logic: validate items, calculate pricing...
    let mut subtotal = 0.0_f64;
    for item in &cart_items {
        subtotal += item.price * item.quantity as f64;
    }
    let tax = (subtotal * 0.08 * 100.0).round() / 100.0;
    let total = ((subtotal + tax) * 100.0).round() / 100.0;

    Ok(json!({
        "validated_items": cart_items,
        "subtotal": subtotal,
        "tax": tax,
        "total": total,
        "item_count": cart_items.len()
    }))
}
}

Functions with Dependencies

Handlers that need upstream step results receive a HashMap<String, Value>:

#![allow(unused)]
fn main() {
/// Step 2: Process payment using cart total from validate_cart.
pub fn process_payment(
    context: &Value,
    dependency_results: &HashMap<String, Value>,
) -> Result<Value, String> {
    let token = context
        .get("payment_token")
        .and_then(|v| v.as_str())
        .unwrap_or("tok_test_success");

    let cart_result = dependency_results
        .get("validate_cart")
        .ok_or("Missing validate_cart dependency result")?;
    let cart_total = cart_result
        .get("total")
        .and_then(|v| v.as_f64())
        .unwrap_or(0.0);

    // Business logic: process payment...
    Ok(json!({
        "transaction_id": format!("txn_{}", uuid::Uuid::new_v4()),
        "amount_charged": cart_total,
        "status": "completed"
    }))
}
}

Some handlers only need dependency results (no task context):

#![allow(unused)]
fn main() {
/// Step 3: Reserve inventory based on validated cart items.
pub fn update_inventory(
    dependency_results: &HashMap<String, Value>,
) -> Result<Value, String> {
    let cart_result = dependency_results
        .get("validate_cart")
        .ok_or("Missing validate_cart dependency result")?;

    let validated_items = cart_result
        .get("validated_items")
        .and_then(|v| v.as_array())
        .ok_or("Missing validated_items in cart result")?;

    // Business logic: create inventory reservations...
    Ok(json!({
        "total_items_reserved": validated_items.len(),
        "status": "reserved"
    }))
}
}

Handler Registry

Plain functions are bridged to the StepHandler trait via a FunctionHandler wrapper and registered in a StepHandlerRegistry. The registry matches the handler.callable field from task template YAML:

#![allow(unused)]
fn main() {
use std::sync::{Arc, RwLock};
use std::collections::HashMap;
use async_trait::async_trait;
use tasker_worker::worker::handlers::{StepHandler, StepHandlerRegistry};
use tasker_shared::types::base::TaskSequenceStep;
use tasker_shared::messaging::StepExecutionResult;

pub struct AxumHandlerRegistry {
    handlers: RwLock<HashMap<String, Arc<dyn StepHandler>>>,
}

impl AxumHandlerRegistry {
    pub fn new() -> Self {
        let registry = Self { handlers: RwLock::new(HashMap::new()) };
        // Register all handlers — callable names match YAML handler.callable
        registry.register_fn("ecommerce_validate_cart",
            Box::new(|ctx, _deps| handlers::ecommerce::validate_cart(ctx)));
        registry.register_fn("ecommerce_process_payment",
            Box::new(|ctx, deps| handlers::ecommerce::process_payment(ctx, deps)));
        // ... more handlers
        registry
    }
}
}

The FunctionHandler wrapper extracts context and dependency results from the TaskSequenceStep and calls the plain function:

#![allow(unused)]
fn main() {
#[async_trait]
impl StepHandler for FunctionHandler {
    async fn call(&self, step: &TaskSequenceStep) -> TaskerResult<StepExecutionResult> {
        let context = step.task.task.context.clone()
            .unwrap_or_else(|| Value::Object(Default::default()));
        let dep_results: HashMap<String, Value> = step.dependency_results
            .iter()
            .map(|(name, result)| (name.clone(), result.result.clone()))
            .collect();

        match (self.handler_fn)(&context, &dep_results) {
            Ok(result) => Ok(StepExecutionResult::success(
                step.workflow_step.workflow_step_uuid,
                result, elapsed_ms, None,
            )),
            Err(err) => Ok(StepExecutionResult::failure(
                step.workflow_step.workflow_step_uuid,
                err, None, None, false, elapsed_ms, None,
            )),
        }
    }
}
}

Accessing Task Context

In standalone functions, context is a &Value — use serde_json accessors:

#![allow(unused)]
fn main() {
// Read a string field with a default
let customer_email = context
    .get("customer_email")
    .and_then(|v| v.as_str())
    .unwrap_or("unknown@example.com");

// Deserialize into a typed struct
#[derive(Debug, Deserialize)]
struct CartItem {
    product_id: i64,
    quantity: i64,
}

let cart_items: Vec<CartItem> =
    serde_json::from_value(context.get("cart_items").cloned().unwrap_or(json!([])))
        .map_err(|e| format!("Invalid cart_items: {}", e))?;
}

Accessing Dependency Results

Dependency results are a HashMap<String, Value> mapping step names to their result JSON:

#![allow(unused)]
fn main() {
// Get a single upstream result
let cart_result = dependency_results
    .get("validate_cart")
    .ok_or("Missing validate_cart dependency")?;
let total = cart_result.get("total").and_then(|v| v.as_f64()).unwrap_or(0.0);

// Combine results from multiple upstream steps (convergence)
let payment_result = dependency_results
    .get("process_payment")
    .ok_or("Missing process_payment dependency")?;
let inventory_result = dependency_results
    .get("update_inventory")
    .ok_or("Missing update_inventory dependency")?;
}

Error Handling

Return Err(String) from standalone functions. The FunctionHandler wrapper converts errors to StepExecutionResult::failure:

#![allow(unused)]
fn main() {
// Validation error (permanent — will not retry)
if cart_items.is_empty() {
    return Err("Cart cannot be empty".to_string());
}

// Business logic error with context
return Err(format!(
    "Insufficient stock for {}: requested {}, available {}",
    product.name, requested, available
));
}

For finer control over retryability, use StepExecutionResult directly in a StepHandler implementation:

#![allow(unused)]
fn main() {
// Non-retryable error
Ok(StepExecutionResult::failure(
    step.workflow_step.workflow_step_uuid,
    "Invalid order data".to_string(),
    Some("VALIDATION_ERROR".to_string()),
    Some("ValidationError".to_string()),
    false,  // not retryable
    duration_ms,
    None,
))

// Retryable transient error
Ok(StepExecutionResult::failure(
    step.workflow_step.workflow_step_uuid,
    "Payment gateway unreachable".to_string(),
    Some("GATEWAY_ERROR".to_string()),
    Some("NetworkError".to_string()),
    true,  // retryable
    duration_ms,
    None,
))
}

Task Template Configuration

Generate a task template with tasker-ctl:

tasker-ctl template generate task_template \
  --language rust \
  --param name=EcommerceOrderProcessing \
  --param namespace=ecommerce \
  --param handler_callable=ecommerce_validate_cart

Rust handler callables use the snake_case names registered in the handler registry (e.g., ecommerce_validate_cart, ecommerce_process_payment).

Testing

Test standalone handler functions directly with serde_json values:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::collections::HashMap;

    #[test]
    fn test_validate_cart_success() {
        let context = json!({
            "cart_items": [
                {"product_id": 1, "quantity": 2},
                {"product_id": 2, "quantity": 1}
            ]
        });

        let result = validate_cart(&context).unwrap();

        assert!(result.get("total").unwrap().as_f64().unwrap() > 0.0);
    }

    #[test]
    fn test_validate_cart_empty() {
        let context = json!({"cart_items": []});

        let result = validate_cart(&context);

        assert!(result.is_err());
        assert!(result.unwrap_err().contains("empty"));
    }

    #[test]
    fn test_process_payment_with_dependency() {
        let context = json!({
            "payment_token": "tok_test_success"
        });

        let mut deps = HashMap::new();
        deps.insert("validate_cart".to_string(), json!({
            "total": 64.79,
            "validated_items": []
        }));

        let result = process_payment(&context, &deps).unwrap();

        assert_eq!(result["status"], "completed");
        assert_eq!(result["amount_charged"], 64.79);
    }
}
}

Capability Traits

Beyond the base StepHandler, the worker crate defines capability traits in handler_capabilities.rs for specialized patterns:

A Rust handler implements StepHandler and adds any capability traits it needs — this is idiomatic Rust trait composition. For a complex example combining multiple capabilities, see diamond_decision_batch.rs in the Rust worker crate.

The Rust batch_processing module is the foundation that Python, Ruby, and TypeScript access through FFI. The specialized handler types in those languages are ergonomic wrappers — Rust developers work with the underlying traits directly.

Next Steps

• Your First Workflow — Build a multi-step DAG end-to-end
• Architecture — System design details
• Axum example app — Complete working implementation

Next Steps

You’ve built your first handler and workflow. Here’s where to go from here.

See Complete Applications

The Example Apps demonstrate five real-world workflow patterns implemented in all four languages. Each app is a fully working project you can clone, run, and study.

Learn Through Stories

The Engineering Stories series teaches Tasker concepts through progressive scenarios — from a simple e-commerce checkout to multi-team namespace isolation. Each story builds on the previous one.

Go Deeper

Architecture & Design

• Architecture Overview — System design, lifecycle actors, and DAG execution
• Worker Architecture — How workers process steps across languages
• Event System — Pub/sub and observability events

Operations

• Backpressure Architecture — Monitor and tune backpressure
• Configuration Management — Database connection and pool management
• Observability — Metrics, tracing, and logging
• Auth & Security — Authentication and authorization

Reference

• Configuration Reference — All configuration options
• API Reference — REST and gRPC API documentation
• Handler Types — API, Decision, and Batchable handler patterns

CLI Tooling

• Using tasker-ctl — Initialize projects, generate scaffolding, manage remotes
• tasker-ctl Architecture — How the CLI tool is built

Contributing

• Contributing to Tasker — Development setup, workflow, and PR process for tasker-core and tasker-contrib
• GitHub Issues — Report bugs or request features

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<M> 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<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?

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<M> 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<M> 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) => StepExecutionResult::retryable("dispatch_channel_full"),
    Err(DispatchError::Timeout) => StepExecutionResult::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) => StepExecutionResult::permanent("handler_crash"),
            Err(FfiError::Panic) => StepExecutionResult::permanent("handler_panic"),

            // Developer-returned errors = trust their classification
            Ok(HandlerResult::RetryableError(msg)) => StepExecutionResult::retryable(msg),
            Ok(HandlerResult::PermanentError(msg)) => StepExecutionResult::permanent(msg),
            Ok(HandlerResult::Success(data)) => StepExecutionResult::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 RustStepHandler {
        async fn call(&self, step_data: &TaskSequenceStep) -> Result<StepExecutionResult>;
    }
}

// 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::Base
  def call(context)
    # Step implementation
    success(result: { status: '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 RustStepHandler for MyHandler {
    async fn call(&self, step_data: &TaskSequenceStep) -> Result<StepExecutionResult> {
        // Step implementation
        Ok(StepExecutionResult::success_from_json(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<M>: 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<M>, 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