Worker

Worker System Protocol

Background task processing system with queues, priorities, and retry logic. Provides a robust foundation for async task execution similar to:

Sidekiq (Ruby)
Celery (Python)
Bull/BullMQ (Node.js)
AWS SQS/Lambda

Features:

Task queues with priorities
Task scheduling and retry logic
Batch processing
Dead letter queues
Task monitoring and logging

@example Basic task

const task: Task = {
  id: 'task-123',
  type: 'send_email',
  payload: { to: 'user@example.com', subject: 'Welcome' },
  queue: 'notifications',
  priority: 5
};

Source: packages/spec/src/system/worker.zod.ts

TypeScript Usage

import { BatchProgressSchema, QueueConfigSchema, TaskSchema, TaskExecutionResultSchema, TaskPriority, TaskRetryPolicySchema, TaskStatus, WorkerStatsSchema } from '@objectstack/spec/system';
import type { BatchProgress, QueueConfig, Task, TaskExecutionResult, TaskPriority, TaskRetryPolicy, TaskStatus, WorkerStats } from '@objectstack/spec/system';

// Validate data
const result = BatchProgressSchema.parse(data);

BatchProgress

Properties

Property	Type	Required	Description
batchId	`string`	✅	Batch job identifier
total	`integer`	✅	Total number of items
processed	`integer`	✅	Items processed
succeeded	`integer`	✅	Items succeeded
failed	`integer`	✅	Items failed
percentage	`number`	✅	Progress percentage
status	`Enum<'pending' \| 'running' \| 'completed' \| 'failed' \| 'cancelled'>`	✅	Batch status
startedAt	`string`	optional	When batch started
completedAt	`string`	optional	When batch completed

QueueConfig

Properties

Property	Type	Required	Description
name	`string`	✅	Queue name (snake_case)
concurrency	`integer`	✅	Max concurrent task executions
rateLimit	`{ max: integer; duration: integer }`	optional	Rate limit configuration
defaultRetryPolicy	`{ maxRetries: integer; backoffStrategy: Enum<'fixed' \| 'linear' \| 'exponential'>; initialDelayMs: integer; maxDelayMs: integer; … }`	optional	Default retry policy for tasks
deadLetterQueue	`string`	optional	Dead letter queue name
priority	`integer`	✅	Queue priority (lower = higher priority)
autoScale	`{ enabled: boolean; minWorkers: integer; maxWorkers: integer; scaleUpThreshold: integer; … }`	optional	Auto-scaling configuration

Task

Properties

Property	Type	Required	Description
id	`string`	✅	Unique task identifier
type	`string`	✅	Task type (snake_case)
payload	`any`	✅	Task payload data
queue	`string`	✅	Queue name
priority	`Enum<'critical' \| 'high' \| 'normal' \| 'low' \| 'background'>`	✅	Task priority level
retryPolicy	`{ maxRetries: integer; backoffStrategy: Enum<'fixed' \| 'linear' \| 'exponential'>; initialDelayMs: integer; maxDelayMs: integer; … }`	optional	Retry policy configuration
timeoutMs	`integer`	optional	Task timeout in milliseconds
scheduledAt	`string`	optional	ISO 8601 datetime to execute task
attempts	`integer`	✅	Number of execution attempts
status	`Enum<'pending' \| 'queued' \| 'processing' \| 'completed' \| 'failed' \| 'cancelled' \| 'timeout' \| 'dead'>`	✅	Current task status
metadata	`{ createdAt?: string; updatedAt?: string; createdBy?: string; tags?: string[] }`	optional	Task metadata

TaskExecutionResult

Properties

Property	Type	Required	Description
taskId	`string`	✅	Task identifier
status	`Enum<'pending' \| 'queued' \| 'processing' \| 'completed' \| 'failed' \| 'cancelled' \| 'timeout' \| 'dead'>`	✅	Execution status
result	`any`	optional	Execution result data
error	`{ message: string; stack?: string; code?: string }`	optional	Error details if failed
durationMs	`integer`	optional	Execution duration in milliseconds
startedAt	`string`	✅	When execution started
completedAt	`string`	optional	When execution completed
attempt	`integer`	✅	Attempt number (1-indexed)
willRetry	`boolean`	✅	Whether task will be retried

TaskPriority

Allowed Values

critical
high
normal
low
background

TaskRetryPolicy

Properties

Property	Type	Required	Description
maxRetries	`integer`	✅	Maximum retry attempts
backoffStrategy	`Enum<'fixed' \| 'linear' \| 'exponential'>`	✅	Backoff strategy between retries
initialDelayMs	`integer`	✅	Initial retry delay in milliseconds
maxDelayMs	`integer`	✅	Maximum retry delay in milliseconds
backoffMultiplier	`number`	✅	Multiplier for exponential backoff

TaskStatus

Allowed Values

pending
queued
processing
completed
failed
cancelled
timeout
dead

WorkerStats

Properties

Property	Type	Required	Description
workerName	`string`	✅	Worker name
totalProcessed	`integer`	✅	Total tasks processed
succeeded	`integer`	✅	Successful tasks
failed	`integer`	✅	Failed tasks
active	`integer`	✅	Currently active tasks
avgExecutionMs	`number`	optional	Average execution time in milliseconds
uptimeMs	`integer`	✅	Worker uptime in milliseconds
queues	`Record<string, { pending: integer; active: integer; completed: integer; failed: integer }>`	optional	Per-queue statistics

On this page