vtq

vtq — Visibility Timeout Queue (SQLite)

Pure-Go, single-table job queue with SQS-style visibility timeout. One primitive, three patterns — just calibration.

         Publish ──► ┌──────────────┐
                     │  vtq_jobs    │
                     │              │  visible_at <= now()
         Claim  ◄─── │  id          │ ◄── consumer polls
                     │  payload     │
                     │  visible_at ─┤── invisible while held
                     │  attempts    │
                     └──────┬───────┘
                            │
              ┌─────────────┼─────────────┐
              ▼             ▼             ▼
           Ack           Nack         Timeout
         (delete)    (reappear now)  (reappear later)

Patterns

Quick start

db, _ := sql.Open("sqlite", "app.db")

q := vtq.New(db, vtq.Options{
    Queue:      "emails",
    Visibility: 30 * time.Second,
})
q.EnsureTable(ctx)

// Producer
q.Publish(ctx, "msg-001", []byte(`{"to":"a@b.com"}`))

// Consumer (blocks until ctx cancelled)
q.Run(ctx, func(ctx context.Context, job *vtq.Job) error {
    return sendEmail(job.Payload)
    // return nil  → Ack (deleted)
    // return err  → Nack (reappears immediately)
})

Leader election

q := vtq.New(db, vtq.Options{
    Queue:      "leader",
    Visibility: 10 * time.Second,
})
q.EnsureTable(ctx)

// Seed the token once.
q.Publish(ctx, "leader-token", nil)

// Each instance tries to claim.
for {
    job, _ := q.Claim(ctx)
    if job != nil {
        // I am the leader — renew before timeout.
        go keepAlive(ctx, q, job.ID, 5*time.Second)
        doLeaderWork(ctx)
        break
    }
    time.Sleep(time.Second)
}

func keepAlive(ctx context.Context, q *vtq.Q, id string, interval time.Duration) {
    for {
        select {
        case <-ctx.Done():
            return
        case <-time.After(interval):
            q.Extend(ctx, id, 10*time.Second)
        }
    }
}

API

Options

Schema

CREATE TABLE IF NOT EXISTS vtq_jobs (
    id          TEXT PRIMARY KEY,
    queue       TEXT NOT NULL DEFAULT '',
    payload     BLOB,
    visible_at  INTEGER NOT NULL DEFAULT 0,
    created_at  INTEGER NOT NULL,
    attempts    INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_vtq_visible ON vtq_jobs (queue, visible_at);

Combining with dbsync for HA

┌──────────┐  vtq: claim leader-token  ┌──────────┐
│ Instance │ ◄────────────────────────► │  SQLite   │
│ A (leader)                            │ vtq_jobs  │
│          │ ──── dbsync snapshots ───► │          │
└──────────┘                            └──────────┘
                                              ▲
┌──────────┐  vtq: claim → nil (blocked)      │
│ Instance │ ◄────────────────────────────────┘
│ B (standby)   reads via dbsync replica
└──────────┘
     │
     └── If A crashes → visibility expires → B claims → becomes leader