tasking

tasking — reliable async tasks & DAG workflows for Go

tasking is an embeddable Go library for running asynchronous background tasks and multi-step DAG workflows reliably — with retries, timeouts, cancellation, and a subscribable notification stream — surviving process restarts and worker crashes.

It is a library embedded by other applications: it orchestrates execution, but it does not own multi-tenancy or access control — those belong to the embedding application. Each component's DESIGN.md records the rationale for that boundary.

Use cases

• Background jobs — submit a unit of work and have it run reliably: immediate or scheduled one-shot, with per-attempt retry (exponential backoff), a per-attempt timeout, and cancellation.
• Multi-step pipelines — express work as a DAG of steps with dependency edges; the engine fans out steps that can run in parallel, dispatches each step once its parents finish, enforces a workflow-wide deadline, and lets you revive a failed workflow or cancel one (e.g. render → thumbnail → publish).
• Reacting to state changes — subscribe to task/workflow lifecycle events over Redis pub/sub.

Intentionally out of scope (belongs to the embedding application): multi-tenancy isolation, and access control. Periodic/recurring tasks are also not yet supported (only immediate and scheduled one-shot). See each component's DESIGN.md.

How the pieces fit

flowchart TD
    subgraph app["your application"]
      WF["workflow engine<br>(DAG orchestration + state)"]
      TASK["task engine<br>(execution: retry / timeout / cancel)"]
      NOTIFY["notify<br>(Redis pub/sub over the audit log)"]
    end
    DB[("db<br>durable state +<br>SystemEventAudit log")]

    WF -- "every step runs as a task<br>(__EXECUTE_WORKFLOW_STEP__)" --> TASK
    WF --> DB
    TASK --> DB
    DB -- "un-broadcast audit rows" --> NOTIFY
    NOTIFY -- "step results (creator channel)" --> WF

The workflow engine owns DAG orchestration and state; the task engine owns the execution of each individual step — per-attempt retry and timeout are the task engine's job. Every workflow step, whatever its type, is executed as an ordinary task under the reserved name __EXECUTE_WORKFLOW_STEP__, so the workflow engine inherits the task engine's reliability for free. Both engines write durable, creator-tagged SystemEventAudit rows in the same transaction as each state change; notify turns those rows into a best-effort Redis pub/sub stream — and the workflow scheduler is itself a notify subscriber, which is how step results reach it.

Shared reliability model. Across all three components, the database is the source of truth. Every IPC message is a best-effort poke that lets a component act sooner than its periodic maintenance sweep would; lose a poke and work is delayed, never lost. IPC rides crash-safe reliable Redis queues, handlers are idempotent, and unprocessable messages are quarantined rather than crash-looped.

Components

Underpinning all three: the db package (db.Client and transaction support — persistence and the SystemEventAudit log) and the models package (configuration structs, wire types, and the engine's state enums).

Sitting above them, the root tasking package provides one-call constructors for the wiring below — NewPersistenceClient, NewTaskWorkerReceiver, NewTaskScheduler, NewTaskClient, DefineTaskQueueForWorkflowStep, NewWorkflowScheduler, NewWorkflowClient, and NewNotificationProducer. Each fills in the standard Redis IPC and executor factories; drop to the per-package constructors (task.NewReceiver, workflow.NewClient, …) when you need to substitute one, as tests do.

Wiring it together

A single embedding application typically stands up all three components, using the root package's constructors. The snippet below is otherwise complete — errors are discarded into _ purely to keep the composition readable.

The composition point to notice is DefineTaskQueueForWorkflowStep: it builds the workflow Step Runner and hands it back as an ordinary task queue config, running under the reserved task name __EXECUTE_WORKFLOW_STEP__. That queue joins your own in one []models.TaskQueueConfig, which is how the workflow engine inherits the task engine for execution, retry and timeout. Processors are supplied declaratively — each is declared on its task, inside the queue that runs it — so there is no runtime registration call: the receiver builds each queue's task name → processor map from the config and hands it to that queue's executor.

// --- One persistence client, shared by every component in this process ---
dbClient, _ := tasking.NewPersistenceClient(postgres.Open(dsn), logger.Warn)

// --- Workflow step handlers: one per step Type ---
stepHandlers := map[string]models.WorkflowStepProcessor{
    "render-html":     renderHandler{},
    "make-thumbnails": thumbnailHandler{},
    "push-cdn":        cdnHandler{},
}

// --- Where workflow plugs into task: this builds the Step Runner and returns an ordinary task
//     queue which runs it under the reserved name __EXECUTE_WORKFLOW_STEP__ ---
workflowQueue, _ := tasking.DefineTaskQueueForWorkflowStep(
    "workflow-queue", 4 /* workers */, 16 /* buffered requests */, dbClient, stepHandlers,
)

// --- Your own queues, plus the workflow one. Task names must be unique across the whole set ---
taskQueues := []models.TaskQueueConfig{
    {
        Name:           "default-queue",
        Workers:        4,
        BufferRequests: 16,
        SupportedTasks: []models.TaskDefinition{
            {
                TaskName:  "resize-image",
                Retry:     models.RetryParam{InitialDelaySec: 5, MaxRetries: 3},
                Processor: resizeProcessor{},
            },
        },
    },
    workflowQueue,
}

// --- Task Receiver: runs the tasks. A nil OnFatal keeps the default (log and terminate) ---
receiver, _ := tasking.NewTaskWorkerReceiver(ctx, models.TaskReceiverConfig{
    Name: "worker-01", SchedulerQueue: "task-scheduler", TaskQueues: taskQueues,
}, dbClient, redisClient, nil)
_ = receiver.Initialize(ctx, nil) // MUST run before Start: reconciles buffered work after a crash
_ = receiver.Start(ctx)
defer receiver.Stop(ctx)

// --- Task Scheduler: the single writer of task state. It gets the SAME taskQueues — that is what
//     tells it which queue to poke for __EXECUTE_WORKFLOW_STEP__ and for your own task names ---
taskScheduler, _ := tasking.NewTaskScheduler(ctx, models.TaskSchedulerConfig{
    MaintenanceTimerIntSecs: 30, SchedulerQueue: "task-scheduler", TaskQueues: taskQueues,
}, dbClient, redisClient, nil)
_ = taskScheduler.Start(ctx)
defer taskScheduler.Stop(ctx)

// --- notify Producer: broadcasts audit rows. EmitCreator is REQUIRED for workflow feedback — the
//     workflow scheduler subscribes on notify:creator:<engine-creator>, and without it every step
//     sits in RUNNING until its deadline expires ---
producer, _ := tasking.NewNotificationProducer(ctx, models.NotificationProducerConfig{
    PollIntervalSecs: 5, BatchSize: 100, EmitCreator: true,
}, dbClient, redisClient)
_ = producer.Start(ctx)
defer producer.Stop(ctx)

// --- Workflow Scheduler: single writer of workflow state. It builds its own task client (to
//     dispatch steps) and notify consumer (to receive their results) internally ---
wfScheduler, _ := tasking.NewWorkflowScheduler(
    ctx,
    models.WorkflowSchedulerConfig{
        MaintenanceTimerIntSecs: 30, SchedulerQueue: "workflow-scheduler",
    },
    "task-scheduler", // the task scheduler's queue, where step tasks are submitted
    dbClient, redisClient, nil,
)
_ = wfScheduler.Start(ctx)
defer wfScheduler.Stop(ctx)

// --- Submit workflows. The step Types listed here must match the handler map above ---
wfClient, _ := tasking.NewWorkflowClient(
    ctx, "my-app", "my-app",
    models.WorkflowClientConfig{SchedulerQueue: "workflow-scheduler"},
    []string{"render-html", "make-thumbnails", "push-cdn"},
    dbClient, redisClient,
)
wf, _ := wfClient.DefineAndRunWorkflow(ctx, workflow.DefineWorkflowParams{ /* ... */ }, nil)
_ = wf

// --- Submit ordinary tasks. Only needed if your app dispatches non-workflow work; the workflow
//     scheduler does not use this client ---
taskClient, _ := tasking.NewTaskClient(
    ctx, "my-app", "my-app",
    models.TaskClientConfig{SchedulerQueue: "task-scheduler", TaskQueues: taskQueues},
    dbClient, redisClient,
)
_ = taskClient

Routing. Because a workflow step is an ordinary task, the queue returned by DefineTaskQueueForWorkflowStep must reach both the receiver's TaskReceiverConfig and the scheduler's TaskSchedulerConfig — the receiver runs the step tasks, the scheduler decides which queue to dispatch them to. Give it to the receiver alone and steps are defined but never routed, so workflows stall with nothing reported. See task/README.md and workflow/README.md for the full flow.

Requirements & getting started

• Module: github.com/alwitt/tasking (Go 1.26+).
• PostgreSQL-compatible database — durable state and the SystemEventAudit log, via the db package.
• Redis — IPC message queues (engine coordination) and pub/sub (notifications).

Build and test targets live in the Makefile.

License

Released under the MIT License.