Events & Background Jobs

maniflex offers two complementary mechanisms for work that happens outside the request pipeline: an event bus for lightweight domain-event fan-out, and a job queue for durable, retriable background work.

Event bus

The event bus lets pipeline middleware publish domain events that any number of subscribers consume independently. An events.Emit call on the DB-After step publishes user.created, order.placed, etc. to whichever bus is wired up:

import (
    "github.com/xaleel/maniflex/events"
    "github.com/xaleel/maniflex/events/redis"
)

bus := redis.New(redisClient, "myapp") // prefix namespaces the Redis stream keys
server.Pipeline.DB.Register(
    events.Emit(bus),
    maniflex.ForModel("Order"),
    maniflex.AtPosition(maniflex.After),
)

Publishing under a transaction

Emit never publishes before the write is durable. Which mechanism it uses depends on the bus:

The second row is the weaker guarantee of the two: the commit can succeed and the broker still be unreachable, and there is no record left to retry from. Use an outbox.Bus when losing an event is worse than storing one — see Example 3 for the pattern end to end.

If you register your own side effect from a middleware — a webhook, a cache invalidation — reach for ctx.AfterCommit rather than firing it inline:

ctx.AfterCommit(func() { go notify(orderID) })

It runs the callback immediately when no transaction is active, so it is safe to use unconditionally. It runs synchronously after the commit, so start a goroutine for anything slow.

Deferral needs an owner. AfterCommit can only queue a callback for someone who has promised to drain the queue, and that means a transaction the framework opened: WithTransaction on the pipeline, or maniflex.Batch anywhere — including inside a custom action, where the Service step never runs. Both drain on commit and drop on rollback, and both publish the queue on ctx.Ctx, so an Execute handed that same transaction queues onto it rather than firing on its own.

A transaction you open yourself with ctx.BeginTx cannot be drained: you call Commit, so only you know when it succeeded. AfterCommit then returns false and runs the callback inline, inside the open transaction — where a rollback can no longer take it back. It logs a warning saying so. Either let one of the two owners hold the transaction, or do the side effect yourself after your Commit returns.

Subscribers register a Subscription:

bus.Subscribe(ctx, events.Subscription{
    Patterns: []string{"order.*"},
    Handler:  func(ctx context.Context, e events.Event) error { /* ... */ return nil },
})

For WebSocket fan-out, connect a realtime.Hub to the bus — see Realtime / WebSockets.

What the payload carries

Event.Data is the written row, keyed by database column name — not by json name, and not the response shape. It is deliberately not the response projection: locale resolution and ctx.RedactResponseField masking are decisions made for one requesting caller, and an event is durable, replayable, and read by subscribers who never made that request.

Four kinds of column are stripped before the event leaves:

This matters because the payload is not a transient in-memory value: it is persisted verbatim to event_outbox.payload, replayed by the outbox relayer, pushed to every WebSocket and SSE client through the hub, and written to whatever the broker retains. An encrypted column emitted in the clear defeats the at-rest guarantee everywhere downstream at once.

maniflex.RedactRecord(model, row) applies the same exclusion set, if you build an event by hand (see the custom-action note below) or serialize ctx.DBResult in your own middleware.

Ordering

Event order is not guaranteed by default. Two events are delivered in the order they were produced only while nothing fails: a delivery that fails is retried after a backoff, and later events keep flowing past it in the meantime. For one record that means an update can be applied before the create it follows, or an older state can overwrite a newer one.

The outbox can preserve order per aggregate:

bus.Relay(outbox.RelayOptions{OrderedByKey: true})

A row is then held back while an older unshipped row shares its ordering key, which is the event’s Subject ("invoice/abc123" by default — the same value the Kafka adapter uses as its partition key). Ordering is per key, so an aggregate that is stuck holds up only its own events. Events with no Subject are never held: they name no aggregate, so there is nothing to order them against.

It is opt-in because it costs head-of-line blocking — while one row for an aggregate is failing, every later row for that aggregate waits with it, up to MaxAttempts and its backoff. Enabling it adds an ordering_key column; Migrate adds it to an existing table.

This covers the outbox only. No broker adapter serialises per key on the consumer side, so with Subscription.Concurrency above 1 two events for one record can be handled concurrently whatever order they arrived in. Kafka’s partition key gives per-partition ordering on the wire, not in the handler. Make handlers idempotent and safe to apply out of order, or set Concurrency: 1.

Idempotent delivery

Every broker adapter here is at-least-once, so a handler can see the same event twice. That is by design and not a rare edge: a consumer that crashes with work in flight replays it on restart, because the alternative — treating in-flight work as consumed — loses it. events.Dedupe wraps a handler to suppress the repeat:

store := events.NewSQLDedupeStore(db, "sqlite") // or NewInMemoryDedupeStore
store.Migrate(ctx)

bus.Subscribe(ctx, events.Subscription{
    Patterns: []string{"order.*"},
    Handler:  events.Dedupe(store)(myHandler),
})

The ID is claimed before the handler runs, so two workers handed the same event concurrently do not both process it. If the handler then returns an error, the claim is released, so the retry is not mistaken for a duplicate — a transient failure retries normally and only a genuine redelivery is dropped.

Releasing is an optional capability: a custom DedupeStore may also implement events.DedupeReleaser. Both bundled stores do. A store that does not cannot undo its claim, so a handler that fails transiently loses the event rather than retrying it — Dedupe logs a warning naming the store when you wrap one.

A claim outlives a process crash mid-handler: the ID stays recorded and the event is not reprocessed. InMemoryDedupeStore bounds this with its TTL; for SQLDedupeStore, prune event_dedupe on whatever window you can tolerate replaying.

Handler panics

A panic in a Subscription.Handler is recovered and turned into a failed attempt, so it flows into the same retry and dead-letter path as a returned error: a handler that panics once and then succeeds has delivered its event, and one that panics every time exhausts its attempts and dead-letters. Without that the panic unwound into the broker’s delivery goroutine, where nothing recovered it, and the Go runtime killed the process — one bad event type ending every other subscription and the HTTP server with it.

Each panicking attempt is logged at ERROR with the event type, its id, and the stack. That is deliberately louder than a returned error, which only WARNs until its attempts run out: an error says a delivery failed, a panic says the handler is broken.

Subscription.OnPanic is the programmatic signal, for counting or alerting without parsing logs:

events.Subscription{
    Patterns: []string{"invoice.*"},
    Handler:  handleInvoice,
    OnPanic: func(e events.Event, recovered any, stack []byte) {
        metrics.Inc("event_handler_panic", "type", e.Type)
    },
}

It fires once per panicking attempt — three times for a handler that panics through MaxRetry: 2 — and runs on the delivery goroutine, so it must not block. A panic inside the hook is not recovered again.

Dead-lettering

Set Subscription.DLQ (or RelayOptions.DLQType on the outbox relayer) to re-publish an event that exhausted its attempts under a separate type, through the same broker. Both paths produce the same payload:

Everything else is copied unchanged, so the dead-letter carries the same Data, Model, RecordID and TenantID as the event it came from.

A DLQ publish that itself fails is logged, and what happens to the event then depends on the adapter. The core withholds the acknowledgement, so a broker that acknowledges per message brings the event back and retries the dead-letter with it — events/redis and events/nats do this. events/kafka and events/rabbitmq acknowledge anyway and the event is gone; both have reasons, and both are in the delivery matrix with the rest of the row.

The outbox relayer keeps its row instead. The DLQ rides the same broker that just failed every delivery attempt, so “the dead-letter failed too” is the ordinary shape of an outage rather than an edge case. The row is retained and stays claimable, and each later poll retries delivery and then the dead-letter, until one is accepted. During a long outage the table therefore grows and drains again on recovery; that is the trade an outbox makes, and losing the event is the alternative. last_error records what happened, and retries back off to relayBackoff(MaxAttempts) so a dead broker is not hammered.

Setting no DLQType is still an opt-out: dead-lettering is disabled and the row is dropped once its attempts are spent, as documented on RelayOptions.DLQType.

An outbox row whose payload will not decode is dead-lettered immediately, without consuming its retry budget: decoding is deterministic, so a retry parses the same bytes and fails identically. That dead-letter is synthesised from the row itself — original_id is the outbox row id and original_type its type column, since there is no event to read them from — and carries the raw bytes as Data with DataType: application/octet-stream, because they are the only remaining evidence of what was written. The row is then marked shipped so the sweep can reclaim it; last_error records that it was resolved rather than delivered.

Custom actions emit manually. events.Emit runs on the DB step, which custom actions skip — so a server.Action handler never fires the middleware and must publish to the bus itself:

data, _ := json.Marshal(maniflex.RedactRecord(ctx.Model, ctx.DBResult))
err := bus.Publish(ctx.Ctx, events.Event{
    Type:     "order.cancelled",
    Model:    "Order",
    RecordID: orderID,
    Data:     data,
})

Nothing redacts a hand-built payload for you — marshal ctx.DBResult directly and you publish the plaintext of every encrypted column.

For the transactional outbox, publish inside the action’s own transaction so the event commits atomically with the write.

Adapter delivery matrix

Available adapters: events/redis, events/kafka, events/nats, events/rabbitmq. The in-process adapter (inproc.New() from github.com/xaleel/maniflex/events/inproc) ships in the core module for tests.

Every one is at-least-once, and the retry, backoff, panic, and dead-letter behaviour above is shared: it lives in events.DeliverWithRetry, not in the adapters. What follows is where they genuinely differ. The notes after the tables give the reasoning for each; these are the answers.

Ordering is the same everywhere: no adapter serialises per key on the consumer side, so Concurrency above 1 means two events for one record can be handled at once. See Ordering.

What happens to an event that is not delivered

The four outcomes of a delivery, and what each adapter does with them. The third column is the one that differs.

Redis and NATS follow the core rule: the safety net you asked for did not catch the event, so it is not acknowledged as though it had.

Kafka cannot. Its commits are cumulative, so a gap left by a consumer that keeps running stalls every later commit on that partition and grows its pending map without bound. That is an unbounded failure traded against one event whose dead-lettering had already failed too. During a shutdown there is no later commit to stall, which is why that column differs.

RabbitMQ’s reason used to be unbounded prefetch. Prefetch is bounded now, and the answer did not change — it got worse. Withheld messages come back only when the channel closes, so with prefetch N a run of N unsettled deliveries fills the window and the consumer receives nothing further for the life of the process. A handful of poison events would end consumption entirely, which is a heavier failure than losing those events. Per-message acknowledgement is what lets Redis and NATS withhold one message without blocking the next; AMQP’s prefetch window does not.

These are behaviours, not implementation details: a change to any cell above is marked (behaviour change) in the CHANGELOG. The Kafka and RabbitMQ settle rules are named functions with tests over each column, so they cannot move quietly. The rest of the rows are pinned by review — acknowledgement on those adapters goes through the broker client’s own message type rather than a seam a test can drive.

A consumer that cannot reach its broker now says so. The events/kafka and events/redis read loops retry forever — stopping would silently end consumption — but the retry is paced by an exponential, jittered backoff (100ms up to 30s) rather than the fixed one-second interval they used before. The jitter matters on recovery: without it every consumer in a fleet retries on the same tick and stampedes the broker the moment it comes back. Each failed read logs at WARN, escalating to ERROR once when the backoff first reaches its ceiling, so a sustained outage is findable without a long one burying the logs. The wait honours the context, so shutdown no longer blocks behind it. An idle stream is not a failure and does not advance the backoff. The policy is events.ReadBackoff if you need the same behaviour elsewhere.

events/kafka connects in plaintext unless told otherwise. Set Config.TLS and Config.SASL — build the mechanism with kafka-go’s own sasl/plain or sasl/scram — and both apply to publishing, consuming and topic creation, which each open their own connection. Managed clusters require at least one of the two. SASL/PLAIN sends credentials in the clear, so pair it with TLS.

inproc applies backpressure. Each subscription has a bounded queue (Options.QueueSize, default 1024) drained by Concurrency workers. Publish never blocks; it returns inproc.ErrQueueFull when a subscription is full, so a handler slower than the publish rate shows up as an error rather than as memory growth. events.Emit cannot return it to you — it publishes after the response — so it logs an ERROR naming the event instead.

events/redis reclaims abandoned messages. A consumer that dies mid-delivery leaves its messages pending; each consumer runs a periodic XAUTOCLAIM sweep to take them over (Redis 6.2+; see minimum versions). Tune Options.ClaimMinIdle (default 5m) above your slowest handler including retries — a message becomes claimable while its original consumer may still be working on it, so claiming early means delivering twice. Options.ConsumerName (default hostname+pid) must be stable across restarts: Redis never removes consumers from a group.

events/redis trims its streams, and trimming loses events. Each stream is capped at Options.MaxLen (default 100,000). Streams are not queues — an entry stays after it is read — so an uncapped stream grows until Redis runs out of memory. But the cap is paid in events: trimming deletes the oldest entries without consulting consumer groups, so an entry a consumer has read but not yet acknowledged goes with them. The publisher gets no error and the consumer never learns it existed. Size MaxLen against how far behind you are willing to let a consumer fall, not against throughput; MaxLenUnlimited disables trimming if you would rather bound growth with a Redis maxmemory policy. Each event’s two writes — its own stream and the hub — go out as one MULTI/EXEC, so the two can never disagree about whether it happened.

events/rabbitmq reconnects only if it owns its connection. amqp091-go connections do not self-heal, so recovering from a drop means dialing a new one — which the bus can do only when it dialed the first:

bus, err := rabbitmq.NewWithDialer(func() (*amqp.Connection, error) {
    return amqp.Dial(os.Getenv("AMQP_URL"))
})

A dropped subscription then re-declares its exchange, queue, bindings and prefetch on a fresh channel and resumes, paced by Options.ReconnectBackoff. Unacked messages were requeued when the channel closed, so it resumes from them. Attempts are logged, escalating once to ERROR when the backoff reaches its ceiling.

New takes an *amqp.Connection you own and cannot replace it: a drop ends every subscription on it for the life of the process while the app keeps serving. That death is made loud — an ERROR naming the queue and Options.OnSubscriptionClosed — but nothing below it brings the subscription back. Prefer NewWithDialer for anything long-running.

Options.Prefetch bounds what the broker may push at one consumer, defaulting to the subscription’s Concurrency. Its Publish waits for a broker confirm, so a failed publish is reported rather than assumed delivered.

Broker adapters are nested modules. Adapters with heavy dependencies (e.g. NATS) ship as their own Go modules — go get github.com/xaleel/maniflex/events/nats — so the core module stays dependency-light. Pin each one explicitly in go.mod.

NATS: one bus binds one JetStream stream. nats.New(nc, stream) ties a bus to a single stream for both publish and subscribe, and JetStream rejects two streams whose subjects overlap. A service that publishes its own subjects while consuming another service’s subjects needs a stream-ownership decision — either a shared stream, or consume from the owning service’s stream. Create the stream yourself (scoped to your business subjects, not ">"); New does not create it.

NATS durable names changed, and Group now works. Subscription.Group becomes a JetStream queue group, so replicas sharing a Group share the work and each event is handled once by the group — what Group already meant on Kafka and Redis. Previously the adapter bound a durable with no queue group, which accepts exactly one subscription, so a second replica was refused outright. Durable names also gained a hash suffix ({group}-{subject}-{hash}) because the old form was not unique: . renders as _ and > as all, so invoice.* and invoice.all produced the same name and the second subscription was rejected with ErrSubjectMismatch.

Both are breaking for existing deployments. Consumers created by an earlier version are not reused: the old durables keep their position and go unconsumed, and the new ones start at the stream’s default delivery policy — which may replay retained events. Drain the old consumers before upgrading a busy deployment, then remove them (nats consumer ls <stream>, nats consumer rm).

Job queue

The jobs/ packages provide a producer/consumer queue with retries, dead-letter routing, and optional status persistence through the REST layer.

Adapters

All three share the same jobs.Queue and jobs.Source interfaces so swapping adapters is a one-line change.

jobs/sql on SQLite needs SQLite 3.35 or newer (March 2021 — both common Go drivers bundle something far newer). The claim is one UPDATE … RETURNING, so a worker receives exactly the rows it stamped. It was previously an UPDATE followed by a SELECT that re-found those rows by their lease_until timestamp — which is not a unique identifier, so two claims taken in the same clock tick each matched the other’s rows. That handed one job to several workers, and stranded others as running with an attempt already spent that no worker had ever received.

NotBefore is honoured to the nanosecond on SQLite. SQLite compares the timestamp columns as text, so the stored format has to sort the same way the instants do. Timestamps are written with a fixed-width fractional part for that reason; an earlier variable-width format could fire a scheduled job up to a second early or late whenever its NotBefore fell on a whole second.

jobs/sql recovers jobs from a crashed worker. A worker that dies after claiming a job leaves the row running with a lease nobody is renewing. Dequeue sweeps such rows before it claims, returning them to enqueued once their lease has lapsed, so the job is redelivered rather than stranded. The lease is a crash-detection window, not a job-duration limit — a live worker keeps renewing — so only a worker that has stopped renewing lets its jobs age out. A reclaim costs a retry attempt, which bounds a crash loop; a job whose attempts have already reached MaxRetry is dead-lettered by the sweep instead of being handed out again, with last_error recording the expired lease. That case is a poison pill: the worker running it died, so the next one likely dies the same way, and it never reaches the code that would otherwise dead-letter it. The sweep is rate-limited to roughly a tenth of the lease, so crash recovery lags the timeout slightly rather than costing a write on every poll.

jobs/redis recovers jobs from a crashed worker. A worker that dies after claiming a job but before completing it leaves the job in the consumer group’s pending list. Dequeue reclaims such entries (via XAUTOCLAIM) once they have been idle past Options.ReclaimMinIdle (default 5m), so the job is redelivered rather than lost. ReclaimMinIdle is a crash-detection window, not a job-duration limit: a live worker renews its hold on a long-running job through the worker’s lease-renewal loop, so only a worker that has stopped renewing — crashed or hung — lets its jobs age past the threshold. Give each worker a unique Options.ConsumerID (the default is maniflex-{hostname}-{pid}); two workers sharing one ID are a single consumer to Redis and share one pending list, which defeats per-worker recovery.

Delayed jobs are promoted once, even with many replicas. A job enqueued with NotBefore/EnqueueAt waits in a sorted set until due, then the promoter moves it to the stream. Every replica runs a promoter, but the move is a single atomic server-side script, so Redis serialises them: whichever replica runs first claims the due jobs and the rest find them already gone — a delayed job is delivered once, not once per replica, and a dropped connection cannot leave one half-moved.

Defining and enqueueing a job

import (
    "database/sql"

    "github.com/xaleel/maniflex/jobs"
    jobssql "github.com/xaleel/maniflex/jobs/sql"
)

// jobs/sql takes a database/sql handle, not the maniflex DB adapter.
db, _ := sql.Open("sqlite", "./app.db")
queue := jobssql.New(db)
if err := jobssql.Migrate(ctx, db, "sqlite"); err != nil { /* ... */ } // "postgres" on PG

Driver dialect. New detects whether the handle is Postgres or SQLite from the driver, recognising lib/pq and jackc/pgx. The dialect fixes both the SQL and the placeholder style ($1 vs ?), so a wrong guess fails outright rather than running slow. If you use a Postgres driver it does not recognise, state it explicitly with jobssql.New(db, jobssql.WithDriver("postgres")) — the same value you pass to Migrate.

Lanes and secrets

• Separate lanes: run an isolated queue on its own table so a type-restricted worker can’t interfere with other jobs. Pass WithTableName to both New and Migrate (indexes are renamed to match, so two queues share one DB):

  otp := jobssql.New(db, jobssql.WithTableName("otp_jobs"))
  jobssql.Migrate(ctx, db, "sqlite", jobssql.WithTableName("otp_jobs"))
  

  The name must be a plain SQL identifier ([A-Za-z_][A-Za-z0-9_]*). It is interpolated directly into every statement and into the migration DDL — a table reference cannot be bound as a parameter — so anything else is rejected: Migrate returns an error and New panics. Do not build it from user input.

• Encrypt payloads at rest: payloads are stored as cleartext JSON by default. Pass WithKeyProvider(kp, keyID) to encrypt the payload column with the same key machinery as mfx:"encrypted" struct fields — encryption.EnvKeyProvider or encryption.VaultKeyProvider:

  kp := &encryption.EnvKeyProvider{Prefix: "MYAPP_KEY"} // reads MYAPP_KEY_JOBS
  q := jobssql.New(db, jobssql.WithKeyProvider(kp, "jobs"))
  

  Stored values are enc:<base64(envelope)>, and the envelope carries the key id, so rotation works: point the queue at a new key id and jobs already in the queue still decrypt, as long as the provider can still resolve the id they were written under. Retire an old key only once no job holds a payload encrypted with it.

  The older WithPayloadCipher(cipher) (any Encrypt([]byte)/Decrypt([]byte) implementation, stored encq:) still works and is still read, but records no key id — so rotating its key strands every job still holding one. Prefer a key provider. Both may be set while migrating: new rows are written through the provider, existing encq: rows keep decrypting with the cipher.

  A payload that cannot be decrypted — key retired, option removed — is never handed to a handler as-is; the row is quarantined as dead with the reason recorded.

• Unhandled types are requeued, not killed: a worker that lacks a handler for a job’s type now requeues it (so another worker can claim it) instead of dead-lettering it — safe for a type-restricted worker sharing a table.

• A row that will not decode is quarantined, not fatal: if a job’s stored payload cannot be decoded or decrypted — a rotated cipher key, a corrupted value — that one row is marked dead with the reason in its last_error, and the rest of the batch it was claimed with dispatches normally. Previously the whole Dequeue failed, and because the claim had already committed, every good job claimed alongside it was stranded as running. Quarantined rows stay visible through Inspector.List/Get (with an empty payload) so you can see what happened and re-enqueue if the cause was recoverable.

• Visibility timeout: a claimed job is invisible to other workers until its lease expires, after which another Dequeue may reclaim it. The default is 5 minutes; WithLeaseDuration(d) changes it. It must exceed how long a handler runs, or a still-running job is reclaimed and executed a second time — but a long handler does not need a large value if the worker renews the lease, which it does automatically. Renewal only ever extends: a renewal horizon shorter than the current lease leaves it alone, so renewing can never make a job reclaimable sooner than the timeout promises.

  q := jobssql.New(db, jobssql.WithLeaseDuration(30*time.Minute))
  

  The timeout is therefore also the crash-recovery window: it is how long a dead worker’s jobs wait before another one picks them up. Trading it off against handler duration is the whole decision — too short re-runs live work, too long delays recovery. Renewal is what lets you keep it short.

// Inside a pipeline middleware or action handler:
id, err := queue.Enqueue(ctx, jobs.Job{
    Type:     "send_receipt",
    ActorID:  ctx.Auth.UserID,
    TenantID: ctx.Auth.TenantID,
    Payload:  json.RawMessage(`{"order_id":"abc"}`),
})

Fields worth knowing:

The Worker

import "github.com/xaleel/maniflex/jobs"

w, err := jobs.NewWorker(jobs.WorkerConfig{
    Source:   queue.(jobs.Source),
    Handlers: map[string]jobs.Handler{
        "send_receipt": func(ctx context.Context, j jobs.Job) (jobs.Result, error) {
            var p struct{ OrderID string `json:"order_id"` }
            json.Unmarshal(j.Payload, &p)
            return jobs.Result{}, mailer.SendReceipt(ctx, p.OrderID)
        },
    },
    Concurrency: 8,        // goroutines; default = GOMAXPROCS
    Logger:      slog.Default(),
})

ctx, cancel := context.WithCancel(context.Background())
go w.Run(ctx)

// On shutdown:
cancel()
w.Shutdown(shutdownCtx)

Cancelling the run context does not orphan a finished job. When you cancel() to stop the worker, a handler already running is interrupted through its context — but once a handler returns, the worker records the outcome (ack, retry, or dead-letter) on a context detached from the cancellation, so a job that just succeeded is acknowledged and not redelivered on the next start. Those writes are still bounded, so a hung queue backend cannot hold Shutdown open indefinitely.

Migrate before you launch background goroutines. server.Go(fn) (and a bare go w.Run(ctx)) starts running immediately, but AutoMigrate only runs inside Start(). A worker that touches a table before Start() migrates it races table creation. When you launch workers yourself, call server.MigrateOnly(ctx) after SetDB and before starting them so the tables exist.

Result carries an optional URL (pre-signed storage URL for file outputs) and Output (small structured JSON). Both are surfaced through the status model below.

StatusModel — REST polling

Mount the status model once, alongside other model registrations:

import jobsmaniflex "github.com/xaleel/maniflex/jobs/maniflex"

sink, queue, err := jobsmaniflex.Mount(server, rawQueue)
if err != nil { log.Fatal(err) }

// Pass sink to the worker:
w, _ := jobs.NewWorker(jobs.WorkerConfig{
    Source:  queue.(jobs.Source),
    Status:  sink,
    Handlers: handlers,
})

Mount registers a StatusModel (table job_statuses) and returns:

• sink — a jobs.StatusSink to pass to WorkerConfig.Status; the worker writes a row for every lifecycle transition.
• queue — a wrapped jobs.Queue; every Enqueue call creates an initial enqueued status row so clients can poll immediately.

The REST layer exposes these endpoints automatically (no extra code):

GET  /api/job_statuses           list (filterable, paginated)
GET  /api/job_statuses/:id       single row
POST /api/job_statuses           → 405 (worker-only)

A typical client flow after an action returns {"job_id": "abc"}:

GET /api/job_statuses/abc
→ {"data": {"status": "enqueued", ...}}

GET /api/job_statuses/abc   (poll until done)
→ {"data": {"status": "succeeded", "result_url": "https://...", "completed_at": "..."}}

Status values: enqueued → running → succeeded | failed | dead | cancelled.

Scope

The built-in force-filter restricts the list to the caller’s own actor_id (and tenant_id when set); callers with the admin role see everything. Override the role name with MountOptions.AdminRole.

A request with no authenticated caller is refused with 401. These rows are per-actor, so there is no scope to apply without an identity — and answering without one would return every actor’s and every tenant’s job metadata. This means the endpoints are only useful behind whatever authentication your app installs; a nil ctx.Auth is treated as a misconfiguration, not as permission.

Atomic enqueue with jobs/sql

When jobs/sql is the adapter and a maniflex.WithTransaction middleware is active, queue.Enqueue runs its INSERT through the same *sql.Tx:

// Service step:
server.Pipeline.Service.Register(func(ctx *maniflex.ServerContext, next func() error) error {
    if err := next(); err != nil {  // DB write commits first
        return err
    }
    _, err := queue.Enqueue(ctx.Ctx, jobs.Job{
        Type:    "reconcile_inventory",
        Payload: json.RawMessage(`{"product_id":"` + productID + `"}`),
    })
    return err
}, maniflex.ForModel("Order"), maniflex.AtPosition(maniflex.After))

If the transaction rolls back, the job row never appears. If the process crashes after commit, the job row is durable and the worker will pick it up. This eliminates the “DB committed but job lost” race that an in-memory queue cannot prevent.

GroupKey — serialised execution

Set GroupKey to ensure at most one job for a given key runs at a time:

queue.Enqueue(ctx, jobs.Job{
    Type:     "generate_payroll",
    GroupKey: "tenant:" + tenantID,  // one payroll run per tenant at a time
})

jobs/inproc tracks running keys in memory. jobs/sql enforces the key on two levels: the claim query ranks candidates with ROW_NUMBER() OVER (PARTITION BY group_key) and takes only the top row per key, so a single Dequeue — however large its batch — never starts two jobs of one key; and a partial unique index on (group_key) WHERE status = 'running' makes a second running job of a key impossible even across two workers claiming at the same instant, which the query alone cannot prevent on Postgres. An empty GroupKey opts out of serialisation entirely, so unkeyed jobs run fully in parallel.

Upgrade note: the partial unique index is created by Migrate. If a queue already contains two running jobs for one key — the very bug this closes — creating the index fails and the migration stops. Drain or clear the duplicate running rows, then migrate.

Retry and dead-letter

When a handler returns an error the worker re-queues the job after an exponential backoff (base 1 s, cap 5 min). After Job.MaxRetry attempts the job is marked dead and the status row records the final error. Set WorkerConfig.DLQType to route dead jobs to a separate handler for inspection or alerting.

Job.Backoff overrides the policy per job:

jobs.Job{
    Type:     "sync_ledger",
    MaxRetry: 20,
    Backoff:  jobs.ExponentialBackoff{Base: time.Minute, Max: time.Hour},
}

Both fields take a default when left at zero: Base zero means 1 s, and Max zero means uncapped — the delay keeps doubling and saturates at the largest representable time.Duration rather than overflowing. A Base of zero used to mean no delay at all, so a literal that set only Max silently retried in a tight loop; use jobs.FixedBackoff{} if that is what you want. Delays are clamped rather than allowed to wrap, so a long MaxRetry with a coarse Base cannot produce a negative delay and an immediate re-run.

Jobs of an unhandled type. A worker that dequeues a job whose Type it has no handler for does not fail or drop it — a type-restricted worker sharing a queue with others must let a job pass to the worker that does handle it. It requeues the job instead, without spending a retry attempt, so the job’s budget is preserved for its real handler. To stop a type that no worker handles from bouncing forever, the worker counts these requeues in a header and dead-letters the job once it reaches WorkerConfig.MaxUnhandledRequeues (default 20) — surfacing the misconfiguration rather than storming. This requires the queue to implement jobs.Requeuer (all three built-in adapters do); a custom adapter that does not falls back to the older unbounded Nack.

The status row follows the job either way, so it never reads as executing on a worker that has already let it go. Through Requeuer it returns to enqueued, since a requeue spends no attempt; through the Nack fallback it becomes failed — or dead once the budget is spent — because that is what Nack does. The row is written only after the queue write succeeds: if the requeue itself fails the job is still held, and running is then the truthful status.

Cancellation

When the inner queue implements jobs.Cancellable (both jobs/inproc and jobs/sql do), the wrapped queue returned by Mount also implements it:

c := queue.(jobs.Cancellable)
c.Cancel(ctx, jobID)   // marks the job cancelled in the queue and updates the status row

Only jobs that have not yet started can be cancelled; a running job must finish or fail before the status row moves.

The two adapters retain a cancelled job differently: jobs/sql keeps the row with status='cancelled', while jobs/inproc drops the entry — as it does for a succeeded or dead-lettered job — so its own Inspector reports only live jobs. Either way the job_statuses row created by Mount is the durable record of the outcome, which is what clients poll.

Completion events (optional)

Set WorkerConfig.EventBus to publish job.{type}.completed and job.{type}.failed events on every terminal transition. Pair with a realtime.Hub to push completion notifications to connected clients without polling:

w, _ := jobs.NewWorker(jobs.WorkerConfig{
    // ...
    EventBus: bus,   // any value implementing Publish(ctx, type, payload) error
})

Scheduled jobs with jobs/cron

jobs/cron provides a minimal ticker that calls Queue.EnqueueAt on a fixed interval. It does not offer durable cron (if a replica is down at fire time the tick is missed); for durable scheduling, combine jobs/sql with a next_fire_at column in your model. For field-based transitions (auto-publish, auto-expire), see Scheduled Fields & Runner.

Schedules are fixed intervals (Every), not cron expressions:

import (
    "time"

    "github.com/xaleel/maniflex/jobs"
    "github.com/xaleel/maniflex/jobs/cron"
)

cr := cron.New(queue, nil) // nil logger → slog.Default()
cr.Add(cron.Entry{
    Every: 24 * time.Hour,
    Job:   jobs.Job{Type: "daily_report"},
})
cr.Start(ctx) // returns immediately; cr.Stop() halts the tickers

Start is idempotent — a duplicate call is a safe no-op, never a second set of tickers. Add must come before it: entries registered after Start are not scheduled. A Scheduler is not reusable, so Start after Stop stays stopped.

Running cron on more than one replica

A Scheduler ticks in its own process and knows nothing about its peers, so three replicas each running one enqueue daily_report three times a day. Run the Scheduler in exactly one process, or pass a cron.Locker and let the replicas elect a single winner per interval:

cr := cron.New(queue, nil, cron.WithLocker(myLocker))

// Locker is the whole interface. Return true to exactly one caller per key.
type Locker interface {
    Acquire(ctx context.Context, key string, ttl time.Duration) (bool, error)
}

The framework ships no implementation — use whatever the deployment already runs. A Redis one is three lines:

func (l redisLocker) Acquire(ctx context.Context, key string, ttl time.Duration) (bool, error) {
    return l.client.SetNX(ctx, key, "1", ttl).Result()
}

Keys look like cron|daily_report|24h0m0s|1784678400 and carry the fire time truncated to Every, so replicas that started at different moments still agree on which interval a tick belongs to — one ticking at 00:03 and another at 00:47 contend for the same midnight key. Entry.Name overrides the daily_report segment; set it when two entries share both a Type and an interval.

Two properties worth knowing:

• There is no release. The lock marks the interval as claimed, not the work as in progress, so it must outlive the firing — let it expire with its TTL (which is Every). Deleting it early lets the next replica to tick within the same interval fire again.
• A Locker error fails open and the job fires, matching idempotency. A lock backend outage produces visible duplicates rather than a nightly job that quietly never ran.