dhara

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 31, 2026 License: MIT Imports: 21 Imported by: 0

README

dhara

CI Go Reference Go Version

Dhara is a distributed task queue library and service for Go, backed by PostgreSQL. It covers the full task lifecycle: enqueue, claim, retry, dead-letter. No external dependencies like Redis or RabbitMQ.

Use it two ways:

  • As an embeddable library: call client.Enqueue(...) from your application (even inside your own DB transaction) and run a dhara.Worker to execute tasks.
  • As pre-built services: cmd/server (HTTP API) and cmd/worker (task processor) are thin binaries assembled entirely from the library, so they can never drift from it.

Quick Start • Migrations • Design • Benchmarks • Configuration • API Reference

Architecture and Task Lifecycle

Dhara uses PostgreSQL as the single source of truth for task state and execution logs.

Dhara architecture diagram

The Task Lifecycle Flow
stateDiagram-v2
    [*] --> PENDING : Task Created (API / Client)
    PENDING --> RUNNING : Claimed by Worker (SKIP LOCKED)
    RUNNING --> COMPLETED : Execution Success
    RUNNING --> PENDING : Execution Failure / Stale Heartbeat (Retry with Jitter)
    RUNNING --> DEAD : Max Retries Exhausted
    PENDING --> CANCELED : Canceled via API
    RUNNING --> CANCELED : Canceled via API

Design Decisions

  1. Atomic concurrency with PostgreSQL (SKIP LOCKED)

    Task claiming uses transactional SELECT ... FOR UPDATE SKIP LOCKED instead of an external lock manager or broker. Multiple workers poll concurrently, each pending task is claimed by exactly one worker, and workers never block waiting on the same row.

  2. Failure recovery: heartbeats and the reaper

    Workers periodically update a heartbeat for the task they're running. A background reaper then finds tasks that have missed their heartbeat window and atomically requeues them (the attempt was already counted when the task was claimed), or moves them to a dead-letter state once retries are exhausted.

  3. Retries with full jitter

    Failed tasks retry on an exponential backoff with full jitter, which spreads retries across a safe window and keeps a thundering herd away from downstream services.

  4. Transactional Enqueue (EnqueueTx)

    EnqueueTx inserts the task inside your existing transaction, so it commits atomically with the business writes that produced it: the task never exists without its trigger, and never survives a rollback. Idempotency uses INSERT ... ON CONFLICT DO NOTHING, so a duplicate replay never aborts the caller's transaction.

  5. At-least-once delivery

    Tasks are guaranteed to run at least once, not exactly once. Failed attempts retry with backoff, stale heartbeats are reaped, and handlers still running during shutdown are cancelled: the interrupted attempt is recorded as a failure and retried, or dead-lettered if no attempts remain. A handler can therefore see the same task more than once, so it must be idempotent, for example by deduplicating on the task ID or an idempotency key. Enqueue-side idempotency prevents duplicate submission, not duplicate execution. Handlers should honor ctx.Done(); one that ignores cancellation delays shutdown until SHUTDOWN_TIMEOUT.

Benchmarks

Load tested with k6 against the HTTP API, backed by Postgres as the source of truth for what actually happened (task state, not just HTTP response codes).

Setup: 20 workers, realistic_work handler simulating 50-200ms of I/O, i3 (11th gen) / 8GB RAM / 500GB SSD, MaxConns=25.

Finding a real bottleneck

Predicted throughput from worker count × handler duration: ~160 tasks/sec. First measured result: ~20 tasks/sec.

The cause was a claim loop that only attempted one claim per poll tick, regardless of how fast a task actually finished:

select {
case <-ticker.C:
    processNext(ctx)
}

20 workers × 1 claim/sec (1s poll interval) = 20/sec, the math matched the measurement almost exactly. Fixed by claiming continuously while work is available, and only falling back to the poll interval once the queue is empty. Re-measured: ~148 tasks/sec, within a few percent of the original prediction.

Results
Submission rate Worker utilization p50 p95 p99 Task loss
100 tasks/sec ~67% (headroom) 181ms 299ms 380ms 0 / 6,001
~148 tasks/sec ~100% (saturation) 1.30s 2.03s 2.06s 0 / 9,001

Latency is end-to-end (completed_at - created_at), not just HTTP enqueue time, enqueue latency alone stayed under 10ms (p99) throughout every run, including the one bottlenecked at 20/sec, which is exactly why the completion-side bottleneck was invisible until checked directly against task state in Postgres rather than HTTP response times.

At saturation, the jump in latency reflects queueing delay from running at the edge of worker capacity, not degraded correctness. Every run completed 100% of submitted tasks with zero loss, including a separate burst test that intentionally oversubmitted (120/sec against a then-20/sec ceiling, pre-fix) and fully recovered a 5,800-task backlog with zero losses once draining caught up.

Reproducing this
k6 run benchmarks/load-test.js
watch -n 5 'psql "$DHARA_DATABASE_URL" -f benchmarks/drain-check.sql'

See benchmarks/RESULTS.md for full methodology and raw data.

Quick Start

Prerequisites: Go 1.26+, Docker & Docker compose.

Option A: Use it as a library in your own app
go get github.com/md-talim/dhara

1. Create a client and enqueue tasks. A Client is lightweight and safe to construct anywhere (web handlers, background jobs, migrations). It starts zero goroutines.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/md-talim/dhara"
)

type EmailPayload struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
}

func main() {
	ctx := context.Background()

	pool, err := pgxpool.New(ctx, "postgres://dhara:dhara@localhost:5432/dhara?sslmode=disable")
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close() // you own the pool; the library never closes it

	// Apply migrations (embedded in the library, no MIGRATIONS_DIR needed)
	if _, err := dhara.Migrate(ctx, pool); err != nil {
		log.Fatal(err)
	}

	client := dhara.NewClient(pool)

	// Enqueue standalone, auto-wraps in its own transaction
	res, err := client.Enqueue(ctx, "send_email", EmailPayload{
		To:      "user@example.com",
		Subject: "Welcome!",
	})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("created task %s (duplicate=%v)\n", res.Task.ID, res.Duplicate)
}

2. Enqueue atomically inside your own transaction. The task is committed only if the surrounding transaction commits. Roll back the tx and the task never exists.

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

// Your business logic...
if _, err := tx.Exec(ctx, `INSERT INTO orders (id, total) VALUES ($1, $2)`, orderID, 42.00); err != nil {
	_ = tx.Rollback(ctx)
	log.Fatal(err)
}

// Task + order commit atomically
res, err := client.EnqueueTx(ctx, tx, "send_email", EmailPayload{
	To:      "user@example.com",
	Subject: "Order #" + orderID,
})
if err != nil {
	_ = tx.Rollback(ctx)
	log.Fatal(err)
}

if err := tx.Commit(ctx); err != nil {
	log.Fatal(err)
}

3. Run a worker to execute tasks. A Worker owns the goroutine pool, heartbeats, and the reaper. Start blocks until the context is cancelled, then cancels in-flight handlers and returns; interrupted attempts are recorded as failures and retried, or dead-lettered if their attempt budget is exhausted.

worker := dhara.NewWorker(pool,
	dhara.WithMaxWorkers(10),
	dhara.WithPollInterval(1*time.Second),
	dhara.WithHandlerTimeout(5*time.Minute),
)

worker.RegisterHandler("send_email", func(ctx context.Context, payload json.RawMessage) error {
	var p EmailPayload
	if err := json.Unmarshal(payload, &p); err != nil {
		return err
	}
	log.Printf("sending email to %s: %s", p.To, p.Subject)
	// do the work...
	return nil
})

ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()

if err := worker.Start(ctx); err != nil {
	log.Fatal(err)
}

Enqueue options. Enqueue and EnqueueTx accept variadic options:

client.Enqueue(ctx, "send_email", payload,
	dhara.WithIdempotencyKey("order-1234"), // replay-safe: same key + payload returns the existing task
	dhara.WithPriority(10),
	dhara.WithMaxRetries(3),
	dhara.WithRunAt(time.Now().Add(1*time.Hour)), // delay execution
)

Every enqueue returns an *EnqueueResult:

type EnqueueResult struct {
	Task      *Task // the created task, or the existing task on an idempotent replay
	Duplicate bool  // true if an existing (type, idempotency_key) task was returned instead
}
Option B: Run the whole stack in 5 minutes

Dhara ships as two independent binaries: the Server (HTTP API) and the Worker (task processor). They share the same database and can be scaled independently.

Easiest (one command): starts Postgres in Docker, waits for it, applies migrations, then runs the server and worker together. Ctrl-C stops both:

make dev

Or run the whole stack in containers with zero manual steps:

docker compose up --build

This builds and starts Postgres, runs migrations via a dedicated one-shot migrate service, then starts the server (http://localhost:8080) and the worker. docker compose down stops everything.

Interact with the queue:

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{"type":"echo","payload":{"message":"Hello, Dhara!"}}'

curl "http://localhost:8080/api/v1/tasks?limit=20"

The worker binary registers a few demo handlers: echo, send_email, always_fail, slow_task, realistic_work, so you can see retries, dead-lettering, and timeouts in action. To register your own task types, see Adding custom task types.

Migrations

Migrations are embedded in the library (go:embed), so users never need a MIGRATIONS_DIR or a separate migration step. Each migration ships as an up/down pair, and rollback runs the matching .down.sql.

There are three ways to migrate, in increasing order of control:

1. In your application (recommended for getting started). Call dhara.Migrate once before enqueuing:

res, err := dhara.Migrate(ctx, pool)
// res.Versions: the migration versions that were applied
// res.Skipped: how many were already applied and skipped

Roll back programmatically with dhara.MigrateDown(ctx, pool, steps): each step runs the corresponding .down.sql:

if _, err := dhara.MigrateDown(ctx, pool, 1); err != nil { // roll back 1 migration
	log.Fatal(err)
}

2. From the CLI (river-style, recommended for production). cmd/migrate is a tiny CLI over the same embedded migrations:

go run ./cmd/migrate           # apply all pending migrations (default: up)
go run ./cmd/migrate down      # roll back 1 migration (runs the .down.sql)
go run ./cmd/migrate down 3    # roll back 3 migrations

It reads DHARA_DATABASE_URL from the environment, just like the server and worker.

3. With your own migration files. If you maintain a fork of the schema, pass any fs.FS rooted at your migration directory (flat NNNNNN_name.up.sql / .down.sql files):

res, err := dhara.MigrateWith(ctx, pool, os.DirFS("./migrations"))
res, err = dhara.MigrateDownWith(ctx, pool, os.DirFS("./migrations"), 2)

Note: The library never auto-migrates by itself, DDL at app startup is a footgun in production. You choose when to run it: once in your app, via the CLI, or via AUTO_MIGRATE on the pre-built binaries.

Configuration

The pre-built binaries are configured via environment variables. Library users configure the same knobs with the dhara.With* option functions instead.

Variable Default Description
DHARA_DATABASE_URL Required PostgreSQL connection string
DHARA_MAX_CONNS 10 Maximum number of database connections in the pool
DHARA_MIN_CONNS 2 Minimum number of database connections in the pool
AUTO_MIGRATE true Automatically run embedded migrations on binary startup
PORT 8080 Port for the HTTP API server
WORKER_PREFIX dhara-worker Worker identity prefix
WORKER_COUNT 5 Size of the local concurrent worker pool
POLL_INTERVAL 1s How often workers poll for pending tasks
HEARTBEAT_INTERVAL 30s How often running tasks update their heartbeat
HANDLER_TIMEOUT 5m Maximum execution time limit for any single task
STUCK_THRESHOLD 5m How long a task can run without a heartbeat before it's considered stuck
REAPER_INTERVAL 30s How often the reaper scans for stuck tasks
BASE_BACKOFF 1s Initial retry backoff (exponential with full jitter)
MAX_BACKOFF 5m Maximum retry backoff
WORKER_METRICS_PORT 9090 Port for the worker's /metrics endpoint (0 disables it)
SHUTDOWN_TIMEOUT 30s Maximum time to wait for workers to exit after shutdown
LOG_LEVEL info Logging verbosity (debug, info, warn, error)
LOG_FORMAT text Structured log output style (text or json)

Durations accept either Go duration strings (30s, 5m) or plain integers interpreted as seconds.

API & Observability Reference

Task Management API
  • POST /api/v1/tasks: Enqueue a new task.
  • GET /api/v1/tasks: Query and list tasks with metadata.
  • GET /api/v1/tasks/{id}: Fetch detailed status of a specific task.
  • DELETE /api/v1/tasks/{id}: Cancel a PENDING task. Returns 409 Conflict if the task is currently RUNNING; terminal tasks are returned unchanged with 200.
  • POST /api/v1/tasks/{id}/retry: Manually trigger a retry for a DEAD task.

Every endpoint is a thin wrapper over the same dhara.Client methods the library exposes (Insert, ListTasks, GetTask, CancelTask, RetryTask), so the API and library share one code path by construction.

Orchestration Health Probes
  • GET /api/v1/livez: Liveness probe. Returns 200 OK when the process is up.
  • GET /api/v1/readyz: Readiness probe. Verifies database connectivity before routing traffic.
Prometheus Metrics

Metrics are process-local: each process exposes only what it produces. Scrape both endpoints and aggregate in PromQL; metric names are identical across processes.

Endpoint Process Exposes
GET /metrics (also /api/v1/metrics) API server Postgres queue gauges, tasks_enqueued_total, tasks_canceled_total
GET :9090/metrics (WORKER_METRICS_PORT) Worker tasks_completed_total, tasks_attempt_failures_total, tasks_retried_total, tasks_dead_total, tasks_reaped_total, reaper_failures_total, worker_claim_errors_total, workers_active, workers_inflight, task_execution_duration_seconds

The server endpoint always returns 200. If Postgres is unreachable, queue_metrics_up is 0 and the gauge series are omitted, while process counters keep flowing.

Library users can mount the same exposition directly:

mux.Handle("GET /metrics", client.MetricsHandler())        // queue gauges + enqueue/cancel
mux.Handle("GET /worker-metrics", worker.MetricsHandler()) // execution counters + latency

client.MetricsHandler includes the Postgres gauges; worker.MetricsHandler never touches the database. If both run in one process, mount both.

Useful queries:

sum(rate(tasks_enqueued_total[5m]))
sum(rate(tasks_completed_total[5m]))
sum(tasks_by_status{status="PENDING"})
histogram_quantile(0.95, sum by (le) (rate(task_execution_duration_seconds_bucket[5m])))

Adding custom task types

Handlers have the signature func(ctx context.Context, payload json.RawMessage) error, return nil for success, any error to trigger a retry (until max retries exhaust and the task goes DEAD).

Using the library: register a handler on your Worker:

worker.RegisterHandler("welcome_email", func(ctx context.Context, payload json.RawMessage) error {
	var p struct {
		To      string `json:"to"`
		Subject string `json:"subject"`
	}
	if err := json.Unmarshal(payload, &p); err != nil {
		return fmt.Errorf("invalid payload: %w", err)
	}

	log.Printf("sending welcome email to %s: %s", p.To, p.Subject)
	// perform task logic here...
	return nil
})

Using the binaries: add the same handler to the registration block in cmd/worker/main.go:

worker.RegisterHandler("echo", tasks.Echo)
worker.RegisterHandler("send_email", tasks.SendEmail)
worker.RegisterHandler("welcome_email", myWelcomeEmailHandler) // yours

Then submit tasks with that "type":

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{"type":"welcome_email","payload":{"to":"dev@example.com","subject":"Welcome to Dhara!"}}'

See internal/tasks/demo_handlers.go for more examples.

Documentation

Overview

Package dhara is a PostgreSQL-backed task queue for Go.

A Client enqueues tasks; a Worker claims and executes them. Postgres is the only dependency: task state, locking, retries and execution logs all live in the database.

Enqueue a task: 

client := dhara.NewClient(pool)
res, err := client.Enqueue(ctx, "send_email", payload)

Run a worker: 

w := dhara.NewWorker(pool, dhara.WithMaxWorkers(10))
w.RegisterHandler("send_email", handleEmail)
if err := w.Start(ctx); err != nil { // blocks until ctx is cancelled
	log.Fatal(err)
}

Delivery is at-least-once: a handler can run more than once after a retry, a stale-heartbeat reap, or a shutdown, so handlers must be idempotent. See the repository README for design notes and operational guidance.

Index

Constants

View Source 
const (
	DefaultPriority   = 0
	DefaultMaxRetries = 5
)

Defaults applied by Normalize when the caller leaves a field unset.

Variables

View Source 
var (
	// ErrTaskConflict is returned when an idempotency key is reused with a
	// different payload.
	ErrTaskConflict = dharatype.ErrTaskConflict

	// ErrTaskNotFound is returned when no task matches the given ID.
	ErrTaskNotFound = dharatype.ErrTaskNotFound

	// ErrTaskNotAvailable is returned by Claim when no task is ready.
	ErrTaskNotAvailable = dharatype.ErrTaskNotAvailable

	// ErrTaskOwnershipLost is returned when a worker tries to update a task
	// the reaper has already requeued.
	ErrTaskOwnershipLost = dharatype.ErrTaskOwnershipLost

	// ErrTaskNotDead is returned by RetryTask for a task that is not DEAD.
	ErrTaskNotDead = dharatype.ErrTaskNotDead
)

Errors returned by Client and Worker. Compare them with errors.Is.

View Source 
var MigrationFS fs.FS = mustSub(migrationsRoot, "migrations")

MigrationFS exposes the migrations embedded in the library, rooted at the migrations directory (flat NNNNNN_name.up.sql / .down.sql files).

Functions

This section is empty.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client enqueues and inspects tasks. It is safe for concurrent use, owns no goroutines, and keeps all state in Postgres.

func NewClient 

func NewClient(pool *pgxpool.Pool) *Client

NewClient returns a Client backed by pool. The caller owns the pool; the library never closes it.

func (*Client) CancelTask 

func (c *Client) CancelTask(ctx context.Context, id string) (*Task, error)

CancelTask cancels a PENDING task. A RUNNING task is returned unchanged so the caller can reject it; terminal tasks are returned as-is. Returns ErrTaskNotFound if the ID does not exist.

func (*Client) Enqueue 

func (c *Client) Enqueue(ctx context.Context, taskType string, payload any, opts ...EnqueueOption) (*EnqueueResult, error)

Enqueue marshals payload and inserts a task of the given type, applying opts.

func (*Client) EnqueueTx 

func (c *Client) EnqueueTx(ctx context.Context, tx pgx.Tx, taskType string, payload any, opts ...EnqueueOption) (*EnqueueResult, error)

EnqueueTx marshals payload and inserts the task inside tx, applying opts.

func (*Client) GetTask 

func (c *Client) GetTask(ctx context.Context, id string) (*Task, error)

GetTask returns the task with the given ID, or ErrTaskNotFound.

func (*Client) Insert 

func (c *Client) Insert(ctx context.Context, params InsertParams) (*EnqueueResult, error)

Insert normalizes and validates params, then enqueues the task in its own transaction. Use InsertTx to join an existing transaction.

func (*Client) InsertTx 

func (c *Client) InsertTx(ctx context.Context, tx pgx.Tx, params InsertParams) (*EnqueueResult, error)

InsertTx enqueues the task inside tx. The caller owns the transaction: the task commits or rolls back with it.

func (*Client) ListTasks 

func (c *Client) ListTasks(ctx context.Context, filter TaskListFilter) ([]Task, int, error)

ListTasks returns the tasks matching filter and the total count before pagination.

func (*Client) Metrics 

func (c *Client) Metrics() MetricsSnapshot

Metrics returns a point-in-time read of this client's process-local counters. Only the enqueue and cancel counters are non-zero on a Client.

func (*Client) MetricsHandler 

func (c *Client) MetricsHandler() http.Handler

MetricsHandler renders this client's process-local counters alongside the Postgres-derived queue gauges.

It always responds 200. When Postgres cannot be read, queue_metrics_up is 0 and the gauge series are omitted, so counters keep flowing during the outage you are trying to debug.

func (*Client) QueueMetrics 

func (c *Client) QueueMetrics(ctx context.Context) (*QueueMetrics, error)

QueueMetrics reads task counts by state from Postgres in a single query.

func (*Client) RetryTask 

func (c *Client) RetryTask(ctx context.Context, id string) (*Task, error)

RetryTask moves a DEAD task back to PENDING.

type EnqueueOption 

type EnqueueOption func(*InsertParams)

EnqueueOption customizes an Enqueue or EnqueueTx call.

func WithIdempotencyKey 

func WithIdempotencyKey(key string) EnqueueOption

WithIdempotencyKey makes submission replay-safe: the same (type, key) with the same payload returns the existing task. A different payload returns ErrTaskConflict.

func WithMaxRetries 

func WithMaxRetries(maxRetries int) EnqueueOption

WithMaxRetries sets the maximum number of attempts, including the first. Zero means one attempt and no retries. Valid range: 0..20.

func WithPriority 

func WithPriority(priority int) EnqueueOption

WithPriority sets claim order; higher values are claimed first. Valid range: 0..100.

func WithRunAt 

func WithRunAt(runAt time.Time) EnqueueOption

WithRunAt delays the task until runAt. It may not be more than 5 minutes in the past or more than 30 days in the future.

type EnqueueResult 

type EnqueueResult struct {
	// Task is the created task, or the existing task on an idempotent replay.
	Task *Task

	// Duplicate is true if an existing task with the same
	// (type, idempotency_key) was returned instead of inserting a new one.
	Duplicate bool
}

EnqueueResult is returned by every insert method.

type HandlerFunc 

type HandlerFunc = dharatype.HandlerFunc

HandlerFunc is the signature of a task handler. Returning a non-nil error schedules a retry, or dead-letters the task once its attempts are exhausted.

type InsertParams 

type InsertParams struct {
	Type           string
	Payload        json.RawMessage
	IdempotencyKey *string
	Priority       *int
	MaxRetries     *int
	RunAt          *time.Time
}

InsertParams is the shared input for creating a task. Pointer fields distinguish "not provided" (default applied) from an explicit zero value (e.g. MaxRetries: 0 means "never retry", while nil means the default 5).

func (*InsertParams) Normalize 

func (p *InsertParams) Normalize(now time.Time)

Normalize fills in defaults for unset fields: priority 0, max_retries 5, run_at now, payload {}.

func (*InsertParams) Validate 

func (p *InsertParams) Validate(now time.Time) error

Validate checks the (already normalized) params.

type MetricsSnapshot 

type MetricsSnapshot struct {
	TasksEnqueuedTotal       int64
	TasksCompletedTotal      int64
	TaskAttemptFailuresTotal int64
	TasksRetriedTotal        int64
	TasksDeadTotal           int64
	TasksCanceledTotal       int64
	TasksReapedTotal         int64
	ReaperFailuresTotal      int64
	ClaimErrorsTotal         int64
	WorkersActive            int64
	WorkersInflight          int64
}

MetricsSnapshot is a point-in-time read of one process's counters. The server and worker track disjoint sets: enqueue/cancel on the server, execution and reaper counters on the worker. Values reset on restart.

type MigrationResult 

type MigrationResult struct {
	Versions []string // versions applied or rolled back in this run
	Skipped  int      // versions already at the target state
}

MigrationResult reports what a migration run changed

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) (MigrationResult, error)

Migrate applies all pending migrations using embedded migration files

func MigrateDown 

func MigrateDown(ctx context.Context, pool *pgxpool.Pool, steps int) (MigrationResult, error)

MigrateDown rolls back the given number of migrations using the embedded files. Each rollback runs the matching .down.sql

func MigrateDownWith 

func MigrateDownWith(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS, steps int) (MigrationResult, error)

MigrateDownWith rolls back migrations from a custom FS.

func MigrateWith 

func MigrateWith(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) (MigrationResult, error)

MigrateWith applies pending migrations from a custom FS (e.g. your own os.DirFS when you maintain your own migration files).

type QueueMetrics 

type QueueMetrics struct {
	TasksPending        int64
	TasksRunning        int64
	TasksCompleted      int64
	TasksCanceled       int64
	TasksDead           int64
	TasksPendingReady   int64
	TasksPendingDelayed int64
}

QueueMetrics is a point-in-time snapshot of task counts by state.

type Task 

type Task struct {
	ID             uuid.UUID  `json:"id"`
	Type           string     `json:"type"`
	Status         TaskStatus `json:"status"`
	Priority       int        `json:"priority"`
	Attempts       int        `json:"attempts"`
	MaxRetries     int        `json:"max_retries"`
	RunAt          time.Time  `json:"run_at"`
	IdempotencyKey *string    `json:"idempotency_key,omitempty"`
	StartedAt      *time.Time `json:"started_at,omitempty"`
	CompletedAt    *time.Time `json:"completed_at,omitempty"`
	LastError      *string    `json:"last_error,omitempty"`
	CreatedAt      time.Time  `json:"created_at"`
	UpdatedAt      time.Time  `json:"updated_at"`
}

Task is the public representation of a task. It intentionally omits the payload: the API never returns it, and callers that inserted the task already know what they put in.

type TaskListFilter 

type TaskListFilter = dharatype.TaskListFilter

TaskListFilter selects and paginates tasks for Client.ListTasks. The zero value lists all tasks with the default limit.

type TaskListResult 

type TaskListResult struct {
	Tasks []Task
	Total int
}

TaskListResult is returned by list operations.

type TaskStatus 

type TaskStatus string

TaskStatus is the lifecycle state of a task. 

const (
	// TaskStatusPending means the task is waiting to be claimed.
	TaskStatusPending TaskStatus = "PENDING"
	// TaskStatusRunning means a worker has claimed and is executing the task.
	TaskStatusRunning TaskStatus = "RUNNING"
	// TaskStatusCompleted means the handler returned nil.
	TaskStatusCompleted TaskStatus = "COMPLETED"
	// TaskStatusCanceled means the task was canceled before completing.
	TaskStatusCanceled TaskStatus = "CANCELED"
	// TaskStatusDead means every attempt failed, or no handler was registered.
	TaskStatusDead TaskStatus = "DEAD"
)

type ValidationError 

type ValidationError = dharatype.ValidationError

ValidationError wraps invalid InsertParams or enqueue options. Use errors.As to inspect the cause.

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

Worker claims and executes tasks. Register handlers before calling Start.

func NewWorker 

func NewWorker(pool *pgxpool.Pool, opts ...WorkerOption) *Worker

NewWorker returns a Worker that claims tasks from pool. It starts no goroutines until Start is called.

func (*Worker) Metrics 

func (w *Worker) Metrics() MetricsSnapshot

Metrics returns a point-in-time read of this worker's process-local counters.

func (*Worker) MetricsHandler 

func (w *Worker) MetricsHandler() http.Handler

MetricsHandler renders the worker's process-local counters, gauges and execution-latency histogram. It never touches Postgres, so it stays scrapeable while the database is down.

func (*Worker) RegisterHandler 

func (w *Worker) RegisterHandler(taskType string, handler HandlerFunc)

RegisterHandler associates taskType with handler. It panics if called after Start.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) error

Start runs the worker pool and blocks until ctx is cancelled or Stop is called. In-flight handlers are cancelled, their outcomes persisted, and the call returns after the pool drains or WithShutdownTimeout elapses. It returns an error if the worker was already started or the drain timed out.

func (*Worker) Stop 

func (w *Worker) Stop()

Stop requests a graceful shutdown. It is safe to call more than once.

type WorkerOption 

type WorkerOption func(*workerConfig)

WorkerOption configures a Worker.

func WithBaseBackoff 

func WithBaseBackoff(d time.Duration) WorkerOption

WithBaseBackoff sets the initial retry delay. Default 1s.

func WithHandlerTimeout 

func WithHandlerTimeout(d time.Duration) WorkerOption

WithHandlerTimeout bounds a single handler invocation. Default 5m.

func WithHeartbeatInterval 

func WithHeartbeatInterval(d time.Duration) WorkerOption

WithHeartbeatInterval sets how often running tasks refresh their lock. Keep it well below WithStuckThreshold. Default 30s.

func WithLogger 

func WithLogger(l *slog.Logger) WorkerOption

WithLogger sets the structured logger. Default slog.Default().

func WithMaxBackoff 

func WithMaxBackoff(d time.Duration) WorkerOption

WithMaxBackoff caps the retry delay. Default 5m.

func WithMaxWorkers 

func WithMaxWorkers(n int) WorkerOption

WithMaxWorkers sets the number of concurrent handler goroutines. Default 5.

func WithPollInterval 

func WithPollInterval(d time.Duration) WorkerOption

WithPollInterval sets how long an idle worker waits before polling again. Default 1s.

func WithReaperInterval 

func WithReaperInterval(d time.Duration) WorkerOption

WithReaperInterval sets how often stale RUNNING tasks are scanned. Default 30s.

func WithShutdownTimeout 

func WithShutdownTimeout(d time.Duration) WorkerOption

WithShutdownTimeout bounds how long Start waits for in-flight handlers after its context is cancelled. Default 30s.

func WithStuckThreshold 

func WithStuckThreshold(d time.Duration) WorkerOption

WithStuckThreshold sets how long a task may go without a heartbeat before the reaper requeues it. Default 5m.

func WithWorkerPrefix 

func WithWorkerPrefix(prefix string) WorkerOption

WithWorkerPrefix sets the identity prefix for worker IDs and reaper logs. Default "dhara-worker".

Source Files

View all Source files

Directories

Path Synopsis
cmd
migrate command
server command
worker command
Package dharatype holds the shared data types used across the dhara library, the core engine, and the API server.
 Package dharatype holds the shared data types used across the dhara library, the core engine, and the API server.
internal
config
ctxlog
db
logging
metrics
store
tasks
testutil
Package testutil provides shared helpers for dhara's test suites.
 Package testutil provides shared helpers for dhara's test suites.
worker

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.