queueforge

QueueForge

A multi-tenant background-job queue for Go and PostgreSQL. Jobs live in your PostgreSQL database; you enqueue over an HTTP API (or from Go code in your own transaction) and run your task handlers in Go workers.

Status: pre-1.0, experimental. The core design (leases with fencing, FOR UPDATE SKIP LOCKED claiming, crash recovery) has an extensive test suite in the repository, but the maintainers' latest changes have been exercised with Docker deployment checks, PostgreSQL integration tests, chaos tests, and Kubernetes drills. It has not been validated as a multi-node production service or through an extended public deployment. Do not run business-critical workloads on it. Delivery is at-least-once: handlers must be idempotent.

Should you use it?

What it does not do: run Python or other-language tasks (handlers are Go only), exactly-once execution (nobody can promise that for side effects), multi-region or automatic database failover. Measured numbers in docs/benchmark-reports/report.md come from one shared vCPU and are not a scale claim.

Quick start (about 10 minutes)

Prerequisites: Docker, Go (version in go.mod).

# 1. PostgreSQL + API + the bundled sample worker (development passwords: never use in production)
docker compose -f deploy/docker-compose.yml up -d --build

# 2. Create a tenant and an API key (the secret is shown once)
docker compose -f deploy/docker-compose.yml exec api /app/admin create-tenant -name demo -rate 50 -burst 100 -pending 1000 -concurrent 4
docker compose -f deploy/docker-compose.yml exec api /app/admin create-api-key -tenant-id <TENANT_UUID> -name demo-key

# 3. Enqueue a job (the sample worker has an "echo" task)
export API_KEY='qf_live_...'
curl -s -X POST localhost:8080/v1/jobs -H "Authorization: Bearer $API_KEY" \
  -H "Idempotency-Key: demo-1" -H "Content-Type: application/json" \
  -d '{"task_type":"echo","payload":{"hello":"world"}}'

# 4. Watch it finish
curl -s localhost:8080/v1/jobs/<JOB_ID> -H "Authorization: Bearer $API_KEY"

Dashboard: http://localhost:8080/ui/ (paste your key).

Write your own tasks (Go)

Handlers run in a Go process you own. Import workerkit from your module:

package main

import (
	"context"
	"encoding/json"
	"log"

	"github.com/SGVARDHANVIKRANTHREDDY/queueforge/workerkit"
)

func main() {
	err := workerkit.Run(map[string]workerkit.Handler{
		"email.send": func(ctx context.Context, payload json.RawMessage) (json.RawMessage, error) {
			// Do the work. Respect ctx.Done(). Be idempotent: this can run more than once.
			// Return workerkit.Permanent(err) for failures that must not be retried.
			return json.RawMessage(`{"sent":true}`), nil
		},
	})
	if err != nil {
		log.Fatal(err)
	}
}

Configuration is via QF_* environment variables (see docs/DEPLOYMENT.md). A handler that panics is recovered and retried like any other failure.

Enqueue from your application

HTTP / curl: see api/openapi.yaml. Always send an Idempotency-Key; a retry with the same key and body returns the original job (HTTP 200, Idempotent-Replay: true), a different body is a 409.

Go

c := queueforge.New("https://queue.example.com", os.Getenv("QUEUEFORGE_KEY"))
job, err := c.Enqueue(ctx, queueforge.EnqueueRequest{TaskType: "email.send", Payload: payload})
jobs, err := c.EnqueueBatch(ctx, []queueforge.EnqueueRequest{...}, "") // atomic, up to 100

Python (enqueue and inspect only; tasks still run in Go workers)

from queueforge import QueueForgeClient
client = QueueForgeClient(base_url="https://queue.example.com", api_key="qf_live_...")
job = client.submit_job("email.send", {"to": "a@b.c"})          # idempotency key is automatic
client.submit_batch([{"task_type": "email.send", "payload": {}}])  # atomic, up to 100
result = client.poll_until_complete(job["id"])

In your own database transaction (Go, same PostgreSQL): the job commits or rolls back with your data.

enq := outbox.New(pool)
tx, _ := pool.Begin(ctx)
defer tx.Rollback(ctx)
// ... your INSERT/UPDATE ...
_, err := enq.EnqueueTx(ctx, tx, outbox.Params{TenantID: tenantID, TaskType: "email.send", Payload: p})
if err != nil { return err }
return tx.Commit(ctx)

Guarantees and limits (be precise)

• At-least-once execution. A job can run again after a crash or lost lease. Database writes from a worker that lost its lease are rejected (fencing), but external side effects are not undone.
• Submission is deduplicated by idempotency key plus a request hash.
• Every accepted job is stored by a PostgreSQL commit; durability is exactly your PostgreSQL's durability.
• Per tenant: enqueue rate limit, pending-job quota, concurrent-execution slots; queries are tenant-scoped.
• Single PostgreSQL primary: no built-in failover. Back up and run a restore drill (deploy/restore-drill.sh).
• Handlers must tolerate cancellation (ctx.Done()) and must not put secrets in error messages.

Operating it

Metrics, alerts and runbooks: deploy/alerts.yml, docs/runbooks/. Production on one VPS: docs/DEPLOYMENT.md. Kubernetes deployment: deploy/k8s/. Threat model: docs/THREAT_MODEL.md. Design decisions: docs/ADRs/ (ADR-001 to ADR-016).

Run the tests

export QF_DATABASE_URL='postgres://postgres:postgres@localhost:55432/queueforge?sslmode=disable'
go vet ./... && go test -race ./...                              # needs a reachable PostgreSQL 16
(cd sdk/python && python -m unittest discover -s tests -v)       # no database needed

Layout

cmd/{api,worker,admin} binaries; workerkit/ public worker entry point; outbox/ public transactional enqueue; sdk/{go,python} clients; internal/ implementation; migrations/ SQL; api/openapi.yaml; deploy/ Compose, Kubernetes, alerts; tests/ integration, property, chaos, security, load; bench/ comparisons.

License

QueueForge is released under the MIT License.