celers-core

Core abstractions and traits for the CeleRS distributed task queue system.

Status: [Stable] — v0.3.0 (2026-07-13) — 438 tests

Overview

This crate provides the fundamental building blocks for CeleRS:

• Task trait: Type-safe executable task interface
• Broker trait: Abstract interface for message brokers (Redis, Postgres, AMQP, SQS, etc.)
• ResultBackend trait: Task result storage abstraction
• TaskExecutor / TaskRegistry: Task execution and registration
• Error types: Comprehensive error handling with retryability classification
• Batch operations: High-performance bulk enqueue/dequeue (10-100x faster)
• CeleryConfig: Full configuration with 23+ environment variables
• Router: Pattern-based task routing
• RateLimit, TimeLimit: Task rate and time limiting
• Revocation: Task cancellation and revocation
• Retry / Exception: Advanced retry strategies and exception policies
• Event / Control: Worker events and control commands
• Distributed Locks: DistributedLockBackend trait (implementations live downstream: InMemoryLockBackend in celers-beat, RedisLockBackend in celers-backend-redis, DbLockBackend in celers-backend-db)
• DAG support: Task dependency graphs with cycle detection
• Local development mode: InMemoryBroker / InMemoryResultBackend (full trait coverage, no external services) plus a CachingResultBackend<B> LRU wrapper fronting any ResultStore
• Distributed rate limiting: DistributedRateLimiter + DistributedRateLimitBackend trait (Redis-ready), TenantRateLimiter for per-tenant quotas, and TaskTypeCircuitBreakers for per-task-type circuit breaking
• Result lifecycle: tombstones (TombstoneRegistry / ResultExistence), result groups (ResultGroup), and per-task-type result TTL (ResultTtlConfig)
• Alerting: event snapshots (EventSnapshot) and a rule engine (AlertRule / AlertEvaluator) for failure-rate, queue-depth, and no-heartbeat alerts
• Task security: HMAC-SHA256 task signing (TaskSigner), argument sanitization (Sanitizer), and PII detection/masking (PiiDetector)

Features

• Type-safe task definitions
• Broker abstraction for multiple backends
• Task state machine (Pending → Running → Success/Failure/Retry)
• Batch operation support (10-100x performance improvement)
• Workflow support (Group, Chain, Chord)
• Async/await native
• Property-based testing coverage

Usage

use celers_core::{Broker, SerializedTask, TaskRegistry};

// Create a task
let task = SerializedTask::new(
    "my_task".to_string(),
    serde_json::to_vec(&args)?,
);

// Enqueue to broker
let task_id = broker.enqueue(task).await?;

// Batch enqueue for better performance
let task_ids = broker.enqueue_batch(tasks).await?;

// Dequeue and execute
if let Some(msg) = broker.dequeue().await? {
    let result = registry.execute(&msg.task).await?;
    broker.ack(&msg.task.metadata.id, msg.receipt_handle.as_deref()).await?;
}

Broker Trait

The core abstraction for message brokers:

#[async_trait]
pub trait Broker: Send + Sync {
    // Single operations
    async fn enqueue(&self, task: SerializedTask) -> Result<TaskId>;
    async fn dequeue(&self) -> Result<Option<BrokerMessage>>;
    async fn ack(&self, task_id: &TaskId, receipt_handle: Option<&str>) -> Result<()>;
    async fn reject(&self, task_id: &TaskId, receipt_handle: Option<&str>, requeue: bool) -> Result<()>;

    // Batch operations (10-100x faster)
    async fn enqueue_batch(&self, tasks: Vec<SerializedTask>) -> Result<Vec<TaskId>>;
    async fn dequeue_batch(&self, count: usize) -> Result<Vec<BrokerMessage>>;
    async fn ack_batch(&self, tasks: &[(TaskId, Option<String>)]) -> Result<()>;

    // Delayed execution (optional, default: execute immediately via enqueue())
    async fn enqueue_at(&self, task: SerializedTask, execute_at: i64) -> Result<TaskId>;
    async fn enqueue_after(&self, task: SerializedTask, delay_secs: u64) -> Result<TaskId>;

    // Management
    async fn queue_size(&self) -> Result<usize>;
    async fn cancel(&self, task_id: &TaskId) -> Result<bool>;
}

Task Types

SerializedTask

pub struct SerializedTask {
    pub metadata: TaskMetadata,
    pub payload: Vec<u8>,       // JSON-serialized arguments
}

impl SerializedTask {
    pub fn new(name: String, payload: Vec<u8>) -> Self;
    pub fn with_priority(self, priority: i32) -> Self;
    pub fn with_max_retries(self, max_retries: u32) -> Self;
    pub fn with_timeout(self, timeout_secs: u64) -> Self;
}

TaskMetadata

pub struct TaskMetadata {
    pub id: Uuid,
    pub name: String,
    pub state: TaskState,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub max_retries: u32,
    pub timeout_secs: Option<u64>,
    pub priority: i32,

    // Workflow support
    pub group_id: Option<Uuid>,   // Group ID for parallel execution
    pub chord_id: Option<Uuid>,    // Chord ID for barrier synchronization
    pub on_success_link: Option<String>,  // Next task to chain-enqueue on success
    pub dependencies: HashSet<TaskId>,    // Tasks that must complete before this one
}

TaskState

pub enum TaskState {
    Pending,
    Running,
    Retrying(u32),  // retry count
    Success,
    Failure,
    Cancelled,
}

Task Registry

Execute registered tasks by name:

use celers_core::{Task, TaskRegistry};

// Implement `Task` for your task type (register() takes a value, not a closure)
struct MyTask;

#[async_trait::async_trait]
impl Task for MyTask {
    type Input = MyInput;
    type Output = MyOutput;

    async fn execute(&self, input: Self::Input) -> celers_core::Result<Self::Output> {
        // Task implementation
        Ok(output)
    }

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

let registry = TaskRegistry::new();

// Register by value — the task name comes from `Task::name()`
registry.register(MyTask).await;

// Execute by name
let result = registry.execute(&task).await?;

Error Handling

pub enum CelersError {
    Serialization(String),
    Deserialization(String),
    Broker(String),
    TaskNotFound(String),
    TaskExecution(String),
    TaskRevoked(TaskId),
    Timeout(String),
    InvalidStateTransition { from: String, to: String },
    Configuration(String),
    Io(std::io::Error),
    Other(String),
}

impl CelersError {
    pub const fn is_retryable(&self) -> bool;
    pub const fn category(&self) -> &'static str;
}

pub type Result<T> = std::result::Result<T, CelersError>;

Performance

Batch Operations

Batch operations provide significant performance improvements:

Best Practices

// Good: Batch enqueue for bulk operations
let tasks = create_many_tasks(100);
broker.enqueue_batch(tasks).await?;

// Less efficient: Individual enqueue
for task in tasks {
    broker.enqueue(task).await?;
}

Workflow Support

TaskMetadata includes fields for Canvas workflow primitives:

// Group tasks
task.metadata.group_id = Some(group_id);

// Chord tasks (map-reduce)
task.metadata.chord_id = Some(chord_id);

See celers-canvas for high-level workflow APIs.

Configuration

CeleryConfig supports 23+ environment variables including:

• CELERY_BROKER_URL, CELERY_RESULT_BACKEND
• CELERY_TASK_SERIALIZER, CELERY_RESULT_SERIALIZER
• CELERY_TASK_DEFAULT_QUEUE, CELERY_TASK_ROUTES
• CELERY_WORKER_CONCURRENCY, CELERY_TASK_SOFT_TIME_LIMIT, CELERY_TASK_TIME_LIMIT
• And more — use CeleryConfig::from_env() or ConfigValidation::validate_detailed()

Dependencies

[dependencies]
async-trait = "0.1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4", "serde"] }
chrono = { version = "0.4", features = ["serde"] }
tracing = "0.1"
regex = "1"
rand = "0.10"
num_cpus = "1"

See Also

• celers-broker-redis: Redis broker with batch operations
• celers-broker-postgres: PostgreSQL broker implementation
• celers-worker: Worker runtime for executing tasks
• celers-canvas: High-level workflow APIs (Chain, Group, Chord)

License

Apache-2.0