baqueue

BaQueue Go

A Go implementation of the BaQueue library in the sibling baqueue-fastapi directory. It provides durable queues, retries, delayed jobs, batches, scheduling, worker supervision, maintenance, and the same live monitoring dashboard.

Requires Go 1.25 or newer. SQLite is the default and uses a pure-Go driver; no C compiler or external service is required. Install the module from GitHub:

go get github.com/basalam/baqueue-golang@main

Import it as baqueue "github.com/basalam/baqueue-golang". No tagged module release has been published yet. For local development in another application:

require github.com/basalam/baqueue-golang v0.0.0

replace github.com/basalam/baqueue-golang => ../baqeueu-go

Quick start

Run these from this directory in separate terminals:

go run ./examples/worker work -q emails --workers 3
go run ./examples/dispatch
go run ./cmd/baqueue dashboard

Workers, producers, and dashboard share .baqueue-go.db in their working directory. The dashboard is at http://127.0.0.1:9100. A self-contained in-memory demo is also available:

go run ./examples/dashboard

Use the library

ctx := context.Background()
q, err := baqueue.Open(ctx, baqueue.DefaultConfig())
if err != nil { return err }
defer q.Close(context.Background())

err = q.RegisterFunc("send_email", func(ctx context.Context, data map[string]any) (any, error) {
    return mailer.Send(ctx, data["to"].(string))
}, baqueue.OnQueue("emails"), baqueue.WithMaxAttempts(3))
if err != nil { return err }

id, err := q.Push(ctx, "send_email", map[string]any{"to": "a@example.com"})
id, err = q.Later(ctx, "send_email", time.Minute, map[string]any{"to": "b@example.com"})
ids, err := q.Bulk(ctx, []baqueue.JobRequest{
    {Name: "send_email", Data: map[string]any{"to": "c@example.com"}},
    {Name: "send_email", Data: map[string]any{"to": "d@example.com"}},
})

Check each returned error in application code. Dispatch options are OnQueue, WithMaxAttempts, WithBackoff, WithTimeout, and WithTags. Registered defaults are applied first, then dispatch options. Producers can dispatch an unregistered name; each consuming process must register the corresponding handler. Separate producers should set the same options explicitly, because policy is stored in the job at dispatch time.

Numeric values inside Data decode as json.Number, preserving large IDs without float64 rounding. Decode into an application struct or call Int64/Float64 as appropriate.

Implement Handler for a struct-based job. Optional SuccessHandler and FailureHandler interfaces provide OnSuccess and terminal OnFailure hooks. Python's decorators and dynamic imports map to RegisterFunc and Register in Go; Python code itself cannot execute inside a Go worker.

Feature parity

Go timeouts are cooperative: handlers must observe ctx.Done() and pass context to I/O. The worker does not abandon a still-running goroutine and immediately run the same job again. Use process isolation for untrusted or non-cooperative jobs. All goroutines share a heap, so memory_limit triggers garbage collection under pressure rather than pretending to enforce a per-worker memory limit. It is not a process memory cap.

Storage and delivery

config := baqueue.DefaultConfig()
config.Driver = baqueue.DriverConfig{Name: "redis", URL: "redis://localhost:6379/0"}
config.Prefix = "myapp:baqueue"
// Other choices:
// DriverConfig{Name: "sqlite", URL: "./queue.db"}
// DriverConfig{Name: "postgres", URL: "postgres://user:pass@localhost/db?sslmode=require"}
// DriverConfig{Name: "memory"}

Memory is only shared inside the current process. SQLite uses WAL, a busy timeout, and transactional claims across processes. PostgreSQL uses row locks with FOR UPDATE SKIP LOCKED. SQL tables have a baqueue_go_ prefix and store the full payload, including history; they are deliberately separate from the Python SQL schema. Opening Go against an existing Python database does not migrate its jobs.

Redis uses the Python key and payload layout. Use matching prefixes only when intentionally sharing storage, and separate queues by runtime. Go handlers can be registered under existing job_class strings. Do not let a Python worker consume a Go-only handler, or the reverse. Redis Cluster is not supported by this multi-key transaction layout. Use maxmemory-policy noeviction and appropriate Redis persistence; recovery cannot restore evicted keys or data lost by an unpersisted server restart.

A Redis claim watches the queue and hash, reads the head without removing it, then commits the payload/status/index updates and ready-entry removal together. Conflicting workers retry the transaction. A disconnect before EXEC leaves the entry ready; an uncertain successful EXEC leaves a processing record recoverable by the supervisor. Future-dated ready entries are restored to delayed storage. Stale processing attempts cannot complete or release a newer claim.

Delivery is at least once, not exactly once. Make externally visible job effects idempotent. Recovery retains the consumed attempt, max_attempts, backoff, and history. The default retry policy allows three total attempts; exponential delays start at 5 seconds, double per attempt, add up to 10% jitter, and cap at one hour. Explicit delays are seconds, for example Backoff{Delays: []float64{10,60,300}}. Manual retry preserves stored attempt counts and policy, as do release and stuck recovery. It intentionally permits another execution of a terminal failed job.

Workers, schedules, batches, and events

sc := baqueue.DefaultSupervisorConfig()
sc.Queues = []string{"emails", "payments"}
sc.MinWorkers, sc.MaxWorkers = 3, 10
sc.StuckProcessingSeconds = 3600
s := baqueue.NewSupervisor(q, sc)
s.Pruner = &baqueue.Pruner{Queue:q, Config:q.Config}
err = s.Run(ctx)

Raise the stuck timeout above the longest legitimate handler runtime. Give supervisors distinct names when monitoring multiple processes. The standard CLI starts automatic pruning once, alongside its supervisor group. Default retention: completed 5 seconds, failed/cancelled 1 day, metrics 7 days; prune interval 5 seconds. Nonpositive retention disables pruning for that category. Legacy positive prune_*_hours fields override seconds.

scheduler, err := baqueue.NewScheduler(q, []baqueue.ScheduleEntry{
    {JobClass:"cleanup", Cron:"0 * * * *", Queue:"maintenance"},
    {JobClass:"heartbeat", Every:30, Queue:"maintenance"},
})
if err != nil { return err }
err = scheduler.Run(ctx)

batch, err := q.Batch([]baqueue.JobRequest{
    {Name:"send_email", Data:map[string]any{"to":"a@example.com"}},
    {Name:"send_email", Data:map[string]any{"to":"b@example.com"}},
}).Name("newsletter").Then("all_done").Catch("any_failed").Finally("finished").Dispatch(ctx)

Schedules use UTC, five-field cron expressions, or interval seconds. As in the Python scheduler, the first tick dispatches immediately. Run one scheduler per schedule set; schedules are not a distributed leader-election mechanism. Batch callback names persist in storage and therefore work when producer and workers are separate processes. Counters are atomic; callbacks use deterministic IDs. Terminal job state and the member's first batch outcome commit together. Manual retry does not reopen that member's original batch outcome or count it twice. Callback dispatch has persistent completion flags and deterministic job IDs; the supervisor retries unfinished dispatch through ReconcileBatches. Callbacks still require idempotent external effects under at-least-once delivery.

Events: job.pushed, job.started, job.completed, job.failed, job.retrying, batch.dispatched, batch.completed, batch.failed, worker.started, worker.stopped, supervisor.started, supervisor.stopped, queue.pruned, and schedule.dispatched. q.Events.On(name, fn) returns an unsubscribe function. Listeners run locally, synchronously, in registration order. Listener errors and panics are logged without changing the job outcome.

Embed the dashboard

handler := dashboard.New(q.Driver,
    dashboard.WithContext(applicationContext),
    dashboard.WithPrefix("/baqueue"),
    dashboard.WithAuth(yourAuthenticationMiddleware),
)
mux.Handle("/baqueue", handler)
mux.Handle("/baqueue/", handler)

Use standard net/http, Gin WrapH, or Echo WrapHandler; no particular Go web framework is required. All API and WebSocket paths honor the prefix. Authentication wraps every route, including assets and WebSockets. The dashboard is administrative: use application authentication and a trusted network. The CLI binds loopback by default. WebSockets accept same-origin requests unless explicitly configured. The original UI loads Alpine.js from jsDelivr and fonts from Google Fonts, so a browser needs access to those hosts.

CLI and configuration

go run ./cmd/baqueue --config config.json status
go run ./examples/worker work --driver sqlite -q emails --workers 3 --max-workers 10
go run ./examples/worker schedule
go run ./cmd/baqueue dashboard --host 127.0.0.1 --port 9100
go run ./cmd/baqueue prune --status completed --hours 24 --yes
go run ./cmd/baqueue retry-failed --queue emails --tag newsletter --yes
go run ./cmd/baqueue reconcile-indexes --driver redis --driver-url redis://localhost:6379/0
go run ./cmd/baqueue reconcile-pending --driver redis --driver-url redis://localhost:6379/0

The standalone CLI cannot dynamically import Go handlers. Build your own executable with cli.Run(ctx, os.Args[1:], os.Stdout, register) as in examples/worker. Configuration field names match Python; load partial JSON with LoadConfig to retain defaults. See config.example.json. Driver URLs and credentials should come from application configuration, not source control.

driver.options supports Redis username, password, db, max_connections (pool_size), socket_timeout, socket_connect_timeout, read_timeout, write_timeout, and pool_timeout. PostgreSQL supports max_size (max_open_conns), min_size (max_idle_conns), and command_timeout (seconds, mapped to server statement timeout). Idle connections are retained lazily, not pre-opened. SQLite accepts URI pragma options in its URL. Unsupported options fail explicitly. SQL isolation can use separate database files or PostgreSQL search_path schemas; prefix controls Redis namespaces.

Verification

go test ./...
go vet ./...
go build ./...
docker compose -f compose.test.yml up -d --wait

PowerShell integration environment:

$env:BAQUEUE_REDIS_URL = 'redis://127.0.0.1:16389/0'
$env:BAQUEUE_POSTGRES_URL = 'postgres://postgres:baqueue_test@127.0.0.1:15489/baqueue_test?sslmode=disable'
go test ./... -count=1

Integration tests use uniquely named Redis prefixes and PostgreSQL schemas. Without the environment variables they skip external services. SQLite and memory always run. Use go test -race ./... with a supported C compiler, or a Linux Go container. Never point fault-injection tests at production.

go run ./cmd/stress --url redis://127.0.0.1:16389/0 --jobs 1000 --workers 8
# In another terminal, while the stress scenario runs:
docker compose -f compose.test.yml restart redis

See docs/verification.md for the recorded validation and CHANGELOG.md for the initial port scope. Licensed under MIT.