Build a Persistent Job Queue in Go Step by Step Guide
Build a durable job queue in Go on an append-only log: enqueue, lease and ack, redelivery after a crash, compaction, fsync cost and tests.
Most programs that need background jobs start with a channel and a few goroutines, and that is the right start. The channel's limit is that it lives in memory. A deploy, a crash or an out-of-memory kill loses every job that was waiting. For emails, reports and webhooks that is often not acceptable. The next step is a queue that survives a restart, and there is a wide range of ways to get one: a Redis list, a table in the database you already run, a managed queue, a message broker. Before reaching for any of them, it is instructive to build the smallest durable queue yourself, because every real system is a variation on the same few ideas.
Tested with Go 1.25.6, macOS on Apple Silicon (arm64)
This article builds one on an append-only log, in about 150 lines of Go, and tests the failure cases that matter: a worker that dies, a process that crashes, a half-written record. It uses channels, mutexes, context and timers, and belongs to the concurrency patterns cluster. It also shows why you should, in most cases, not ship this code. Everything ran on Go 1.25.6.
What "persistent" and "queue" promise
Write the contract first, because the interesting part of a queue is what it does when something goes wrong.
Enqueue(payload)returns only after the job is durably recorded. If the process crashes right after, the job is still there on restart.Next(ctx)hands a job to a worker and leases it: for the lease duration nobody else gets it.Ack(id)marks the job done and removes it for good.- A job that is not acked before its lease expires becomes available again.
- After a crash and restart, every job that was not acked is pending again.
Together the last three give at-least-once delivery: a job is never lost, and it may run more than once. Exactly-once delivery is not something a queue alone can provide, since the worker could finish the work and die before the ack. The handler has to be idempotent: running it twice must be safe, by checking whether the effect already happened or by using a unique key in the downstream system. This is the cost of the guarantee and it is worth saying out loud, because a surprising number of production incidents are duplicated emails.
The log
The storage is a file of JSON lines. Two kinds of record are enough: add carries a job, and done says a job is finished. The state of the queue at any moment is "all add records without a matching done", and replaying the log from the start recomputes it:
package main
import (
"bufio"
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"sync"
"time"
)
type record struct {
Op string `json:"op"` // "add" or "done"
ID int64 `json:"id"`
Payload json.RawMessage `json:"payload,omitempty"`
}
// Job is one unit of work handed to a worker.
type Job struct {
ID int64
Payload json.RawMessage
}
type entry struct {
job Job
leasedUntil time.Time // zero means not leased; held in memory only
}
// Queue is a durable FIFO queue with leases.
type Queue struct {
mu sync.Mutex
path string
f *os.File
fsync bool
lease time.Duration
nextID int64
order []int64 // ids in arrival order
live map[int64]*entry
wake chan struct{} // closed and replaced whenever the state changes
}
// Open replays the log at path and returns the queue it describes.
func Open(path string, lease time.Duration, fsync bool) (*Queue, error) {
q := &Queue{path: path, lease: lease, fsync: fsync, live: map[int64]*entry{}, wake: make(chan struct{})}
if err := q.replay(); err != nil {
return nil, err
}
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return nil, err
}
q.f = f
return q, nil
}
func (q *Queue) replay() error {
f, err := os.Open(q.path)
if errors.Is(err, os.ErrNotExist) {
return nil
}
if err != nil {
return err
}
defer f.Close()
r := bufio.NewReader(f)
var good int64 // offset just past the last intact record
for {
line, err := r.ReadBytes('\n')
if err == io.EOF {
if len(line) > 0 { // a record cut off by a crash: drop it
return os.Truncate(q.path, good)
}
return nil
}
if err != nil {
return err
}
var rec record
if err := json.Unmarshal(bytes.TrimSpace(line), &rec); err != nil {
return fmt.Errorf("corrupt record at offset %d: %w", good, err)
}
q.apply(rec)
good += int64(len(line))
}
}
func (q *Queue) apply(rec record) {
switch rec.Op {
case "add":
q.live[rec.ID] = &entry{job: Job{ID: rec.ID, Payload: rec.Payload}}
q.order = append(q.order, rec.ID)
q.nextID = max(q.nextID, rec.ID)
case "done":
delete(q.live, rec.ID)
}
}replay shows the one rule about logs that is easy to get wrong. A crash in the middle of an append leaves a partial last line, and that is normal: the log is intact up to the last newline. Replay detects the missing newline at end of file, truncates to the last good offset, and carries on. A line that is complete but does not parse is different. It means the file was damaged some other way, and silently skipping it could hide lost data, so replay returns an error. The distinction is the difference between tolerating the failure you expect and masking the one you do not.
Writing records durably
Every state change is appended to the log before it is applied to memory. If the append fails, the in-memory state is not touched, so memory never gets ahead of the disk:
package main
import (
"context"
"encoding/json"
"errors"
"os"
"time"
)
var ErrClosed = errors.New("queue closed")
// append writes one record. The caller holds q.mu.
func (q *Queue) append(rec record) error {
line, err := json.Marshal(rec)
if err != nil {
return err
}
if _, err := q.f.Write(append(line, '\n')); err != nil {
return err
}
if q.fsync {
return q.f.Sync() // do not return until the kernel has put it on stable storage
}
return nil
}
// signal wakes every goroutine waiting in Next. The caller holds q.mu.
func (q *Queue) signal() {
close(q.wake)
q.wake = make(chan struct{})
}
// Enqueue durably adds a job and returns its id.
func (q *Queue) Enqueue(payload any) (int64, error) {
data, err := json.Marshal(payload)
if err != nil {
return 0, err
}
q.mu.Lock()
defer q.mu.Unlock()
if q.f == nil {
return 0, ErrClosed
}
id := q.nextID + 1
rec := record{Op: "add", ID: id, Payload: data}
if err := q.append(rec); err != nil {
return 0, err
}
q.apply(rec)
q.signal()
return id, nil
}
// Ack marks a job as finished for good.
func (q *Queue) Ack(id int64) error {
q.mu.Lock()
defer q.mu.Unlock()
if q.f == nil {
return ErrClosed
}
if _, ok := q.live[id]; !ok {
return nil // already acked; acking twice is harmless
}
rec := record{Op: "done", ID: id}
if err := q.append(rec); err != nil {
return err
}
q.apply(rec)
return nil
}
// Next blocks until a job is available and leases it to the caller.
func (q *Queue) Next(ctx context.Context) (Job, error) {
for {
q.mu.Lock()
if q.f == nil {
q.mu.Unlock()
return Job{}, ErrClosed
}
now := time.Now()
var soonest time.Time // when the earliest lease on a pending job runs out
for _, id := range q.order {
e, ok := q.live[id]
if !ok {
continue
}
if !e.leasedUntil.After(now) {
e.leasedUntil = now.Add(q.lease)
q.mu.Unlock()
return e.job, nil
}
if soonest.IsZero() || e.leasedUntil.Before(soonest) {
soonest = e.leasedUntil
}
}
wake := q.wake
q.mu.Unlock()
var expiry <-chan time.Time
if !soonest.IsZero() {
t := time.NewTimer(time.Until(soonest))
defer t.Stop()
expiry = t.C
}
select {
case <-wake: // something changed: look again
case <-expiry: // a lease ran out: the job is available again
case <-ctx.Done():
return Job{}, ctx.Err()
}
}
}
// Len reports how many jobs are not yet acked, leased or not.
func (q *Queue) Len() int {
q.mu.Lock()
defer q.mu.Unlock()
return len(q.live)
}
// Close stops the queue. Pending jobs stay in the log for the next Open.
func (q *Queue) Close() error {
q.mu.Lock()
defer q.mu.Unlock()
if q.f == nil {
return nil
}
err := q.f.Close()
q.f = nil
q.signal()
return err
}
// Compact rewrites the log with only the live jobs, so it does not grow without bound.
func (q *Queue) Compact() error {
q.mu.Lock()
defer q.mu.Unlock()
tmp := q.path + ".tmp"
out, err := os.Create(tmp)
if err != nil {
return err
}
for _, id := range q.order {
if e, ok := q.live[id]; ok {
line, _ := json.Marshal(record{Op: "add", ID: id, Payload: e.job.Payload})
out.Write(append(line, '\n'))
}
}
if err := out.Sync(); err != nil {
out.Close()
return err
}
out.Close()
if err := os.Rename(tmp, q.path); err != nil { // atomic replace
return err
}
q.f.Close()
f, err := os.OpenFile(q.path, os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return err
}
q.f = f
live := q.order[:0]
for _, id := range q.order {
if _, ok := q.live[id]; ok {
live = append(live, id)
}
}
q.order = live
return nil
}Several choices carry the correctness. Next loops: it scans for a job whose lease is zero or expired, and if there is none it waits on three things at once, a state-change broadcast, the earliest lease expiry, and the context. The broadcast is a channel that is closed and replaced, which wakes every waiter without a condition variable; waiters capture the current channel under the lock and select on it after unlocking, so a change between unlock and wait cannot be missed because it closes the very channel they hold. The scan is O(n) over pending jobs, fine for thousands and not for millions; a heap ordered by lease time fixes that. Ack is idempotent, because a worker may legitimately ack twice after a retry.
Leases live in memory only. After a crash, no lease survives, so every unacked job is immediately available again: that is exactly the at-least-once rule, and it means the log never records leases at all. The cost is that a crash makes in-flight work run again, including work that had finished and not yet been acked.
Compact writes a new file, syncs it, and renames it over the old one. The rename is atomic on POSIX file systems, so a crash leaves either the old log or the new one, never a mixture. A production version also syncs the containing directory so the rename itself is durable, and holds off writers (here, the mutex does) while it swaps.
Watching it work
The demo enqueues twelve jobs and runs three workers. Job 5 is made to "crash" the first time: its worker takes the lease and walks away without acking, as a dead process would. The queue should redeliver it after the lease expires. Then the process "crashes" with three jobs unprocessed, and a second Open on the same file must find exactly those three:
package main
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"slices"
"sync"
"time"
)
func main() {
dir, _ := os.MkdirTemp("", "queue")
defer os.RemoveAll(dir)
path := filepath.Join(dir, "jobs.log")
q, err := Open(path, 100*time.Millisecond, true)
if err != nil {
panic(err)
}
for i := 1; i <= 12; i++ {
q.Enqueue(map[string]int{"n": i})
}
var mu sync.Mutex
var processed []int
deliveries := map[int]int{}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
var wg sync.WaitGroup
for range 3 {
wg.Go(func() {
for {
job, err := q.Next(ctx)
if err != nil {
return
}
var p map[string]int
json.Unmarshal(job.Payload, &p)
mu.Lock()
deliveries[p["n"]]++
first := deliveries[p["n"]] == 1
mu.Unlock()
if p["n"] == 5 && first {
continue // simulate a worker that died: no ack, the lease will expire
}
q.Ack(job.ID)
mu.Lock()
processed = append(processed, p["n"])
done := len(processed) == 12
mu.Unlock()
if done {
cancel()
}
}
})
}
wg.Wait()
slices.Sort(processed)
fmt.Println("processed:", processed)
fmt.Println("job 5 was delivered", deliveries[5], "times (the first worker never acked)")
fmt.Println("pending after the run:", q.Len())
// Crash with three unprocessed jobs.
for i := 13; i <= 15; i++ {
q.Enqueue(map[string]int{"n": i})
}
q.Close() // stands in for a crash: nothing was acked
q2, err := Open(path, 100*time.Millisecond, true)
if err != nil {
panic(err)
}
fmt.Println("pending after reopening the log:", q2.Len())
before, _ := os.ReadFile(path)
q2.Compact()
after, _ := os.ReadFile(path)
fmt.Printf("log lines before compaction %d, after %d\n", bytesLines(before), bytesLines(after))
q2.Close()
fmt.Println("--- the cost of fsync ---")
measureFsync()
}
func bytesLines(b []byte) int {
n := 0
for _, c := range b {
if c == '\n' {
n++
}
}
return n
}processed: [1 2 3 4 5 6 7 8 9 10 11 12]
job 5 was delivered 2 times (the first worker never acked)
pending after the run: 0
pending after reopening the log: 3
log lines before compaction 27, after 3
--- the cost of fsync ---
fsync=false 300 enqueues took 1ms
fsync=true 300 enqueues took 1.427sThe twelve jobs were processed, job 5 was delivered twice, nothing was lost, and after the simulated crash the reopened queue held exactly the three jobs that had not been acked. Compaction shrank the log from every record ever written (adds and dones) to only the live jobs. The shape to remember is that all of those facts are consequences of two rules: write before apply, and leases are not durable.
Choosing the lease
The lease duration is the setting that matters most in production, and the trade-off is easy to state. A short lease redelivers a crashed worker's job quickly and risks redelivering a job that is merely slow, so two workers end up running it at once. A long lease avoids that and leaves a crashed worker's job waiting for the full duration before anyone retries it. Choose it from the handler's real runtime distribution: a few times the p99 is a reasonable start, and anything that can take much longer than the lease should extend it with a heartbeat while it works. Log every redelivery with the job id and delivery count. A rising redelivery rate is the earliest sign of a handler that has gotten slower than its lease.
The cost of durability
fsync is what makes Enqueue safe against power loss, and it is slow. Without it, the write sits in the operating system's page cache and a power cut can lose it. With it, each enqueue waits for the storage device. The function below enqueues 300 jobs with and without fsync:
package main
import (
"fmt"
"os"
"path/filepath"
"time"
)
func measureFsync() {
dir, _ := os.MkdirTemp("", "cost")
defer os.RemoveAll(dir)
for _, fsync := range []bool{false, true} {
q, _ := Open(filepath.Join(dir, fmt.Sprintf("j%v.log", fsync)), time.Second, fsync)
start := time.Now()
for i := range 300 {
q.Enqueue(i)
}
fmt.Printf("fsync=%-5v 300 enqueues took %v\n", fsync, time.Since(start).Round(time.Millisecond))
q.Close()
}
}The demo above calls it at the end, so its two lines are the last lines of the output.
Three hundred enqueues took 1 ms without fsync and about 1.4 seconds with it, roughly 4.7 ms per enqueue on this Mac, where Go's File.Sync asks the drive for a full flush. A page-cache write takes microseconds and a flush takes the storage device's latency, which on a Linux server with a typical SSD is closer to a millisecond or less per sync, and on a drive with a power-loss-protected cache can be far lower. The standard answer is a group commit: instead of syncing after every record, batch the records that arrive together and sync once for the whole batch, then release all of those callers. Under load, many enqueues share one flush and the cost per job drops by the batch size, while each caller still gets the durability guarantee. The change is local to append: callers send their record to a writer goroutine, which writes a batch, calls Sync once, and replies to each. It is a good exercise, and it is the main reason real queues are faster than this one.
The worker side
The queue is half the system; the loop that consumes it is the other half, and the choices in it decide whether the guarantees hold:
func worker(ctx context.Context, q *Queue, handle func(context.Context, Job) error) {
for {
job, err := q.Next(ctx)
if err != nil {
return // ctx cancelled or queue closed
}
jobCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
err = handle(jobCtx, job)
cancel()
if err != nil {
log.Printf("job %d failed: %v", job.ID, err)
continue // no ack: the lease expires and the job comes back
}
if err := q.Ack(job.ID); err != nil {
log.Printf("ack %d: %v", job.ID, err) // the job will run again; the handler must tolerate it
}
}
}Each handler call gets its own timeout, shorter than the lease; if the handler outlives the lease, another worker receives the same job while the first is still running, and the two collide. Set the lease to comfortably more than the longest legitimate handler (or extend the lease with a heartbeat method for long jobs). A failing job is not acked, which turns failure handling into the lease mechanism: it retries after the lease runs out. That works, and it is also the weakness of this design. A job that always fails retries forever, once per lease, forever. The fix is a delivery counter stored with the job, a maximum, and a dead-letter list where jobs go to wait for a human, which real queues provide and this teaching version omits.
The loop also shows where graceful shutdown enters. When the context is cancelled, Next returns, the worker exits after finishing its current job, and anything leased but unfinished is redelivered after the lease expires, which is the same recovery as a crash.
Tests for the failures
The tests that matter exercise crash recovery, redelivery and the torn write:
package main
import (
"context"
"os"
"path/filepath"
"testing"
"time"
)
func open(t *testing.T, path string, lease time.Duration) *Queue {
t.Helper()
q, err := Open(path, lease, false)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { q.Close() })
return q
}
func TestPendingJobsSurviveARestart(t *testing.T) {
path := filepath.Join(t.TempDir(), "q.log")
q := open(t, path, time.Second)
a, _ := q.Enqueue("a")
q.Enqueue("b")
q.Ack(a)
q.Close()
q2 := open(t, path, time.Second)
if q2.Len() != 1 {
t.Fatalf("after restart %d pending, want 1", q2.Len())
}
}
func TestExpiredLeaseIsRedelivered(t *testing.T) {
q := open(t, filepath.Join(t.TempDir(), "q.log"), 30*time.Millisecond)
q.Enqueue("work")
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
first, err := q.Next(ctx)
if err != nil {
t.Fatal(err)
}
second, err := q.Next(ctx) // blocks until the first lease expires
if err != nil {
t.Fatalf("job was not redelivered: %v", err)
}
if first.ID != second.ID {
t.Fatalf("redelivered job %d, want %d", second.ID, first.ID)
}
}
func TestAckedJobIsNotRedelivered(t *testing.T) {
q := open(t, filepath.Join(t.TempDir(), "q.log"), 20*time.Millisecond)
q.Enqueue("work")
ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
defer cancel()
job, _ := q.Next(ctx)
q.Ack(job.ID)
if _, err := q.Next(ctx); err == nil {
t.Fatal("an acked job came back")
}
}
func TestPartialLastLineIsDropped(t *testing.T) {
path := filepath.Join(t.TempDir(), "q.log")
q := open(t, path, time.Second)
q.Enqueue("whole")
q.Close()
f, _ := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
f.WriteString(`{"op":"add","id":2,"payl`) // a crash mid-write: no newline
f.Close()
q2 := open(t, path, time.Second)
if q2.Len() != 1 {
t.Fatalf("%d jobs after a torn write, want 1", q2.Len())
}
}
func TestCorruptMiddleRecordIsAnError(t *testing.T) {
path := filepath.Join(t.TempDir(), "q.log")
os.WriteFile(path, []byte("{\"op\":\"add\",\"id\":1}\nnot json\n{\"op\":\"add\",\"id\":2}\n"), 0o644)
if _, err := Open(path, time.Second, false); err == nil {
t.Fatal("a corrupt record in the middle must not be skipped silently")
}
}
func TestEveryJobIsDeliveredToOneWorker(t *testing.T) {
q := open(t, filepath.Join(t.TempDir(), "q.log"), time.Minute)
const n = 50
for i := range n {
q.Enqueue(i)
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
got := make(chan int64, n)
for range 5 {
go func() {
for {
job, err := q.Next(ctx)
if err != nil {
return
}
q.Ack(job.ID)
got <- job.ID
}
}()
}
seen := map[int64]bool{}
for range n {
select {
case id := <-got:
if seen[id] {
t.Fatalf("job %d delivered twice with a one minute lease", id)
}
seen[id] = true
case <-ctx.Done():
t.Fatalf("only %d of %d jobs delivered", len(seen), n)
}
}
}=== RUN TestPendingJobsSurviveARestart
--- PASS: TestPendingJobsSurviveARestart (0.00s)
=== RUN TestExpiredLeaseIsRedelivered
--- PASS: TestExpiredLeaseIsRedelivered (0.03s)
=== RUN TestAckedJobIsNotRedelivered
--- PASS: TestAckedJobIsNotRedelivered (0.15s)
=== RUN TestPartialLastLineIsDropped
--- PASS: TestPartialLastLineIsDropped (0.00s)
=== RUN TestCorruptMiddleRecordIsAnError
--- PASS: TestCorruptMiddleRecordIsAnError (0.00s)
=== RUN TestEveryJobIsDeliveredToOneWorker
--- PASS: TestEveryJobIsDeliveredToOneWorker (0.00s)
PASS
ok example.test/lab 1.453sThe torn-write test appends half a record by hand, which is a faithful model of a crash. The test for a corrupt middle line asserts the opposite behavior, an error. Both together pin the rule from the replay section. The redelivery tests use a short lease and real time; if you wanted them faster and fully deterministic, a fake clock for the lease checks would do it (fake clocks).
Group commit, sketched
The measurement says the way to make this fast is to share flushes. The change is to take appends out of the callers' hands and give them to one writer goroutine. Each caller sends its record together with a reply channel; the writer drains whatever is waiting, writes all the records, calls Sync once, and then replies to every caller in the batch:
type pending struct {
rec record
reply chan error
}
func (q *Queue) writer(in <-chan pending) {
for first := range in {
batch := []pending{first}
drain:
for len(batch) < 256 {
select {
case p := <-in:
batch = append(batch, p)
default:
break drain
}
}
var err error
for _, p := range batch {
line, _ := json.Marshal(p.rec)
if _, werr := q.f.Write(append(line, '\n')); werr != nil {
err = werr
}
}
if err == nil {
err = q.f.Sync() // one flush for the whole batch
}
for _, p := range batch {
p.reply <- err
}
}
}When load is light, each batch has one record and the cost is what it was. When load is heavy, a batch holds dozens, and the cost per record falls by that factor. Callers still return only after their record is on stable storage, so the durability promise is unchanged. This is the same idea as the write-ahead log commit in databases, and it is why a single disk can sustain thousands of durable commits per second.
When not to ship this
Everything above is a teaching queue, and the list of what it does not do is the list of reasons to use something else. It has no retry limit or dead-letter list, so a poison job that always fails loops forever. It has no scheduling (run at a time), priorities, or per-job timeouts. It lives in one process on one disk, with no replication and no way for a second machine to consume. Its scan is linear. Group commit is left as an exercise.
For real systems, the practical options in order of how often I would pick them: a table in the database you already operate, claimed with SELECT ... FOR UPDATE SKIP LOCKED in PostgreSQL (documented here), which gives transactional enqueue alongside your data and needs no new infrastructure; a Redis-based queue library when you already run Redis; a managed queue (SQS and its peers) when you want someone else to operate it; a broker such as NATS JetStream or RabbitMQ when you need routing and fan-out. What you take from this article is the vocabulary to evaluate them: what do they log before they acknowledge, what happens to a lease when a worker dies, how do they bound redelivery, and what do they ask of your handlers. Idempotency is always on that last list. The Go side of the worker, the loop that calls Next, runs the job under a context, and acks on success, is the worker pool pattern with a durable source, and graceful shutdown is what stops it cleanly without abandoning leased jobs. The reference for the durability call is the os.File.Sync documentation.
