queue

Queue

Queue is tango's built-in task queue, built on PostgreSQL. It provides type-safe, persistent task queues that run within the application — no external message broker required.

Origin: based on Backlite (MIT, originally SQLite-based), ported to PostgreSQL with pgx/v5, UUIDv7 primary keys, TIMESTAMPTZ columns, go-sqlbuilder query construction, and encoding/json/v2 payloads. Upstream is no longer tracked: the engine is owned and evolved by tango.

Features

• Embedded execution — workers run as goroutines inside your process, no separate service needed
• Type-safe queues — generic Queue[T Task] with compile-time type checking for payloads
• Persistent — tasks survive process restarts via PostgreSQL
• Automatic retries — configurable max attempts with a fixed backoff delay
• Atomic claims — tasks are claimed with a conditional UPDATE ... RETURNING; two dispatcher instances never execute the same task twice
• Delayed execution — schedule tasks for later or set a wait duration
• Graceful shutdown — workers finish in-flight tasks before stopping
• Completed task retention — configurable policies for retaining success/failure records
• Periodic cleanup — automatic deletion of expired completed tasks
• Transaction-aware — add tasks inside existing database transactions
• Status tracking — query task status by ID (pending / running / success / failure / not found)

Architecture

flowchart TB
    subgraph Client
        C[Client]
    end

    subgraph Dispatcher
        T[triggerer]
        F[fetcher]
        CL[cleaner]
        CH[(tasks channel)]
        W1[worker 1]
        WN[worker N]
    end

    subgraph PostgreSQL
        QT[(queue_tasks)]
        QTC[(queue_tasks_completed)]
    end

    subgraph Processors
        P1[Queue Processor]
    end

    C -->|Register queues| P1
    C -->|Add tasks| QT
    C -->|Start / Stop| T
    C -->|Start / Stop| F

    T -->|trigger| F
    F -->|claim & dispatch| CH
    CL -->|delete expired| QTC
    CH --> W1
    CH --> WN
    W1 -->|process| P1
    WN -->|process| P1
    P1 -->|success| QTC
    P1 -->|failed| QT

Goroutine model: each Start runs one generation of goroutines; a restart waits for the previous generation to drain, so restarts never race a shutdown in progress.

Requirements

• Go >= 1.27 (generics, stdlib uuid for UUIDv7 generation, encoding/json/v2)
• PostgreSQL >= 18 (native uuidv7() default in the schema)

Installation

This package lives inside the tango module and is not published. The app wires it in internal/registry: the client is built from deps.DB (the shared datastore.Store) with QUEUE_* config, the kernel module in this package (module.go) starts/stops the dispatcher, and features register their queues via deps.Queue.Register(...).

Schema is owned by the migrations (database/migrations/00010_create_queue_tables.sql) — run task db:migrate; the client never creates tables itself.

Quick Start

1. Define a Task

A task is any struct that implements the Task interface — just provide a Config() method that returns queue settings:

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/riipandi/tango/internal/queue"
)

// EmailTask represents an email to send.
type EmailTask struct {
    To      string `json:"to"`
    Subject string `json:"subject"`
    Body    string `json:"body"`
}

func (e EmailTask) Config() queue.QueueConfig {
    return queue.QueueConfig{
        Name:        "email",
        MaxAttempts: 3,
        Timeout:     30 * time.Second,
        Backoff:     10 * time.Second, // retry after 10s on failure
        Retention: &queue.Retention{
            Duration:   7 * 24 * time.Hour, // keep completed records for 7 days
            OnlyFailed: true,               // only retain failures
            Data: &queue.RetainData{
                OnlyFailed: true,
            },
        },
    }
}

2. Create the Client

store, err := datastore.New(context.Background(), datastore.Options{DSN: "postgresql://user:pass@localhost:5432/mydb"})
if err != nil {
    panic(err)
}
defer store.Close()

client, err := queue.NewClient(queue.ClientConfig{
    Store:           store,
    NumWorkers:      5,
    ReleaseAfter:    5 * time.Minute,
    CleanupInterval: 1 * time.Hour,
})
if err != nil {
    panic(err)
}

// ... register queues, add tasks, start ...

3. Register Queues

client.Register(queue.NewQueue[EmailTask](func(ctx context.Context, task EmailTask) error {
    fmt.Printf("Sending email to %s: %s\n", task.To, task.Subject)
    // ... send email ...
    return nil
}))

4. Add Tasks

// Immediate
ids, err := client.Add(EmailTask{
    To: "user@example.com", Subject: "Welcome", Body: "Hello!",
}).Save()

// Delayed
ids, err = client.Add(EmailTask{
    To: "user@example.com", Subject: "Reminder", Body: "Don't forget!",
}).Wait(1 * time.Hour).Save()

// At a specific time
ids, err = client.Add(EmailTask{
    To: "user@example.com", Subject: "Scheduled", Body: "This is scheduled.",
}).At(time.Now().Add(2 * time.Hour)).Save()

// Multiple tasks in one operation
ids, err = client.Add(
    EmailTask{To: "a@example.com", Subject: "Hi A"},
    EmailTask{To: "b@example.com", Subject: "Hi B"},
).Save()

5. Start the Dispatcher

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

client.Start(ctx)

// ... your application runs ...

// Graceful shutdown — waits for in-flight tasks to complete
client.Stop(context.Background())

Configuration Reference

ClientConfig

QueueConfig

Retention

RetainData

API Reference

NewClient(cfg ClientConfig) (*Client, error)

Creates a new queue client. Validates config and initializes the internal dispatcher.

(*Client).Register(queue Queue)

Registers a queue so tasks can be added to it. Panics if a queue with the same name is registered twice, or if the queue name is empty. Must be called before Start().

(*Client).Add(tasks ...Task) *TaskAddOp

Starts an operation to add one or more tasks. Returns a fluent builder:

// Immediate
ids, err := client.Add(myTask).Save()

// Delayed
ids, err := client.Add(myTask).Wait(5 * time.Minute).Save()

// Scheduled
ids, err := client.Add(myTask).At(futureTime).Save()

// With context
ids, err := client.Add(myTask).Ctx(requestCtx).Save()

// Inside a transaction
tx, _ := pool.Begin(ctx)
ids, err := client.Add(myTask).Executor(tx).Save()
tx.Commit(ctx)
client.Notify() // required when using Executor()

(*Client).Start(ctx context.Context)

Starts the dispatcher background goroutines. The provided context controls the main lifecycle — cancelling it triggers shutdown.

(*Client).Stop(ctx context.Context) bool

Attempts graceful shutdown. Waits up to the context deadline. Returns true if all workers finished their tasks, false if the context was cancelled first.

stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer stopCancel()
if !client.Stop(stopCtx) {
    log.Warn("some tasks did not finish gracefully")
}

(*Client).Status(ctx context.Context, taskID string) (TaskStatus, error)

Returns the current status of a task by ID:

(*Client).Notify()

Notifies the dispatcher that a new task was added. Only required when adding tasks inside an external transaction (via TaskAddOp.Executor()), because the dispatcher cannot observe transaction commits.

(*Client).Flush(ctx context.Context) (int64, error)

Deletes all pending (unclaimed) tasks and returns how many were removed. Claimed tasks — in flight or awaiting release — are untouched and will finish their lifecycle normally.

removed, err := client.Flush(ctx)

(*Client).FlushCompleted(ctx context.Context) (int64, error)

Deletes all completed task records, bypassing retention expiry, and returns how many were removed.

removed, err := client.FlushCompleted(ctx)

FromContext(ctx context.Context) *Client

Retrieves the client from a processor context, allowing processors to enqueue follow-up tasks:

client.Register(queue.NewQueue[OrderTask](func(ctx context.Context, task OrderTask) error {
    queue.FromContext(ctx).Add(EmailTask{To: task.Email}).Save()
    return nil
}))

Database Schema

Two tables, created by migration database/migrations/00010_create_queue_tables.sql:

queue_tasks

Index: idx_queue_tasks_fetch on (wait_until ASC, id ASC) WHERE wait_until IS NOT NULL

queue_tasks_completed

Index: idx_queue_tasks_completed_expires on (expires_at) WHERE expires_at IS NOT NULL

Transaction Support

Tasks can be added inside an existing database transaction. The task only becomes visible to the dispatcher after the transaction commits. You must call Notify() after committing:

tx, err := pool.Begin(ctx)
if err != nil {
    return err
}

if _, err = tx.Exec(ctx, "INSERT INTO orders (...) VALUES (...)"); err != nil {
    _ = tx.Rollback(ctx)
    return err
}

ids, err := client.Add(EmailTask{
    To:      "customer@example.com",
    Subject: "Order confirmed",
}).Executor(tx).Save()
if err != nil {
    _ = tx.Rollback(ctx)
    return err
}

if err := tx.Commit(ctx); err != nil {
    return err
}
client.Notify()

Graceful Shutdown

ctx, cancel := context.WithCancel(context.Background())
client.Start(ctx)

// ... application lifetime ...

cancel() // signal shutdown

stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer stopCancel()
if client.Stop(stopCtx) {
    log.Println("All workers finished gracefully")
} else {
    log.Warn("Shutdown timed out — some tasks may be re-queued")
}

Error Handling

• Processor panics are recovered and logged; the task is marked as failed
• Max attempts exceeded: task is moved to queue_tasks_completed with the error message
• Backoff: failed tasks are re-queued with wait_until set to now + Backoff
• Stuck tasks: ReleaseAfter reclaims tasks whose claimed_at expired; the claim wins atomically, so a task is never executed twice across dispatcher instances
• Unregistered queue: the task is discarded with an error log instead of crashing the worker

Testing

Tests run against a real Postgres (testcontainers, Postgres 18) using the shared pkg/testutils.StartPostgres helper; helpers in helpers_test.go manage per-test cleanup.

go test ./internal/queue/
go test -race ./internal/queue/

Design Decisions

Credits

Based on Backlite by Mike Stefanello, adapted for PostgreSQL with pgx/v5.