Task Engine

Event-Driven Distributed Task Processing Engine
Celery-like developer ergonomics with zero-extra-infrastructure on PostgreSQL, or enterprise-grade sub-millisecond throughput on Redis Streams.

Highlights

• Zero-Extra-Infrastructure Mode: Run background tasks, delayed jobs, and retries natively on PostgreSQL using ACID transactional guarantees and FOR UPDATE SKIP LOCKED. No Redis or RabbitMQ required for small-to-medium deployments.
• Enterprise High-Throughput Broker: Scale to millions of tasks with Redis Streams consumer groups (XREADGROUP, XACK) with sub-millisecond dispatch.
• Dual Language SDKs: Full-featured clients for Python (task-engine) and TypeScript / Node.js (@task-engine/sdk).
• Celery-like Ergonomics: Decorate functions with @engine.task or @task, dispatch with .delay() or .apply_async(), and track results via AsyncResult.
• Distributed Cron & Scheduling: Crontab and interval scheduling with automatic leader election via PostgreSQL advisory locks.
• Resilience & Fault Tolerance: Heartbeat leases (lease_expires_at), automatic re-queueing of crashed worker tasks, exponential backoff retries with jitter, and dead-letter queue (DLQ) re-driving.
• Multi-Tenancy & Security: First-class tenant isolation (X-Tenant-Id), JWT & API Key authentication, and token bucket rate limiting.
• Full Observability: Prometheus metrics (/metrics), OpenTelemetry distributed tracing, structured JSON logs, and real-time WebSocket event feeds.
• Real-time Web Console: Next.js 14 dashboard for queue monitoring, worker management, live metrics, and DLQ replay.

Quickstart

Python Quickstart (task-engine)

pip install task-engine

Define a background task in tasks.py:

from task_engine import TaskEngine, TaskContext

engine = TaskEngine.from_env()


@engine.task(queue="notifications", priority=3, max_retries=3)
async def send_welcome_email(payload: dict, context: TaskContext) -> dict:
    print(f"Executing task {context.task_id} for tenant: {context.tenant_id}")
    return {"status": "sent", "email": payload["email"]}

Start the worker:

task-engine worker -A tasks --queues notifications --concurrency 4

Submit and await results programmatically:

from tasks import send_welcome_email

result = send_welcome_email.delay({"email": "alice@example.com"})
print(f"Task ID: {result.id}")

output = result.get(timeout=30)
print(f"Task finished: {output}")

TypeScript / Node.js Quickstart (@task-engine/sdk)

npm install @task-engine/sdk
# or
pnpm add @task-engine/sdk

Submit tasks and poll results with complete type safety:

import { TaskEngine } from "@task-engine/sdk";

const engine = new TaskEngine({
  apiUrl: process.env.TASK_ENGINE_API_URL || "http://localhost:8000",
  apiKey: process.env.TASK_ENGINE_API_KEY,
  tenantId: "default",
});

async function run() {
  const result = await engine.submitTask({
    taskType: "notifications.send_welcome_email",
    payload: { email: "bob@example.com" },
    queue: "notifications",
  });

  console.log(`Submitted: ${result.id}`);
  const output = await result.get({ timeoutMs: 15000 });
  console.log("Result:", output);
}

run();

Subscribe to real-time events over WebSockets:

import { TaskEngineWebSocket } from "@task-engine/sdk";

const ws = new TaskEngineWebSocket({
  apiUrl: "http://localhost:8000",
  channel: "tasks",
});

ws.on("task.completed", (event) => console.log("Completed:", event.data));
ws.connect();

CLI Command Reference

# Start background worker process
task-engine worker -A tasks --queues default,notifications --concurrency 10

# Start distributed cron scheduler daemon
task-engine scheduler -A tasks

# Apply database migrations
task-engine migrate

# Start web dashboard and API server
task-engine ui --port 8000

# Submit ad-hoc tasks from terminal
task-engine submit notifications.send_welcome_email -p '{"email":"user@test.com"}'

# Query task lifecycle and execution output
task-engine status <task-id>

Documentation Hub

Comprehensive engineering guides, specifications, and runbooks:

• SDK Developer Guides:
  • Python SDK Guide
  • TypeScript SDK Guide
  • SDK Changelog
• API Specifications:
  • OpenAPI 3.1 Schema (openapi.json)
  • Postman v2.1 Collection (postman_collection.json)
• Infrastructure & Operations:
  • Configuration Reference
  • Cluster Operations Guide
  • Troubleshooting Guide
• Operational Runbooks:
  • Runbook Index
  • RB-001: Worker Crash Storm
  • RB-002: Redis Outage
  • RB-003: Postgres Outage
  • RB-004: DLQ Replay
  • RB-005: Queue Overload
  • RB-006: Scheduler Stuck
  • RB-007: Outbox Backlog
  • RB-008: Migration Rollback
  • RB-009: Credential Rotation
  • RB-010: Disaster Recovery
  • RB-011: Capacity Expansion
• Architecture Decision Records:
  • ADR Index
  • ADR-001: Project Structure
  • ADR-002: Configuration & Secrets
  • ADR-003: Data Model & Persistence
  • ADR-004: Core Services & Runtime
  • ADR-005: API Surface & Schemas
  • ADR-006: Redis Streams Broker Adapter
  • ADR-007: Observability (Metrics, Tracing, Logging)
  • ADR-008: Security, RBAC & Multi-Tenancy
  • ADR-009: Test Strategy & Validation

Verification & Testing

Python Test Suite

uv run pytest tests/
uv run ruff check .
uv run mypy src/

TypeScript SDK Suite

cd packages/sdk-ts
pnpm test
pnpm run typecheck
pnpm run build

License

This project is licensed under the MIT License.