mqlite

package module
v0.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 22, 2026 License: MIT Imports: 16 Imported by: 0

README

mqlite

mqlite

CI Go Reference Go Report Card codecov Release Go

A small, SQLite/Turso-backed online message queue with Azure Service Bus–style semantics — Peek-Lock, retries, DLQ, scheduling, dedup, GroupID ordering, topics — in a single pure-Go binary (no CGO).

Embed it like goqite, or serve it like a broker — the same engine. Start in-process (with same-DB transactional enqueue), and upgrade to a network broker with one line when you outgrow it.

Goal

mqlite aims to be the smallest reliable queue you can drop into a system — and to stay friendly to both humans and AI agents:

  • Lightweight & flexible. One pure-Go binary over a single SQLite file (or a remote Turso database) — no broker cluster, no ZooKeeper, no sidecar required. Embed it in your process or run it as a broker; move the storage from a :memory: test to a replicated Turso DB without changing a line of app code.
  • Reliable under concurrency. At-least-once delivery with fencing tokens, single-broker crash recovery, idempotent send/receive/settle, and O(log n) claims on a deep backlog — built to take high enqueue/dequeue throughput without losing or double-delivering messages.
  • Simple, unambiguous interface. Every operation is one plain HTTP POST with a JSON body — curl-able, trivially scriptable, and easy for an LLM or agent to drive — with exactly one settlement verb per outcome and no aliased calls.
  • Service Bus flavor, not a clone. Peek-Lock, GroupID sessions, DLQ, scheduling and dedup will feel familiar if you know Azure Service Bus — but the API is its own, shaped to be idiomatic Go and unambiguous rather than bug-for-bug compatible.

It targets most everyday queueing needs — background jobs, transactional outbox/event delivery, topic fan-out to subscriptions, and rate-limited pipelines — rather than competing with Kafka-scale streaming.

Why mqlite

  • One file, one binary. Local SQLite (modernc.org/sqlite, pure Go) or remote Turso/libSQL — the same SQL and semantics, selected by a connection string.
  • Service Bus semantics, honestly at-least-once. Peek-Lock + four-way settle (Complete/Abandon/Reject/Defer), visibility timeout with fencing tokens, delivery_count → DLQ, Renew, scheduled/deferred messages.
  • Approximate order by default, strict order opt-in. Plain queues are competing-consumer. Pick a queue-level ordering mode at create time: standard (default — per-GroupID FIFO with cross-group parallelism), group_fifo (same, but a GroupID is required on every message), or strict_fifo (single global head-of-line FIFO across the whole queue). GroupID is an ordering key (= SQS MessageGroupId / ASB SessionId) — not a consumer group; competing consumers just Receive the same queue.
  • curl-able contract. Every RPC is a plain HTTP POST to /mqlite.v1.<Service>/<Method> with a JSON body; the Go SDK is sugar on top.

Layout

Path What
engine/ The queue core (Store + Service Bus semantics). Transport-agnostic.
server/ Connect-style JSON-over-HTTP broker + Bearer-token auth.
wire/ The JSON contract shared by server and client (one source of truth).
. (package mqlite) Native Go SDK: remote Client, Embedded engine, Receiver.
cmd/mqlite/ Single-binary CLI: serve, send, receive, peek, …

Quickstart

Full copy-runnable programs (embedded, transactional outbox, remote consumer, curl): docs/examples.md.

1. Embedded (in-process — no broker, no HTTP, no second process)

mqlite's primary form is a library you embed directly in your Go process, exactly like goqite or using database/sql against SQLite. OpenEmbedded gives you the full queue — Send/Receive/Peek-Lock/DLQ/scheduling/topics — calling the storage engine in-process. There is no broker to run, no network hop, no JSON serialization, no extra daemon: your app and the queue are one binary backed by one SQLite (or Turso) database. You only start an HTTP server if you choose to (see §2) — the embedded path never opens a socket.

Single process, single writer. Embedded mode is one process owning one database. OpenEmbedded takes an exclusive lock on the DB file, so a second process — or a second OpenEmbedded on the same file — fails fast with ErrDBLocked instead of racing it (two writers would corrupt crash recovery and claim ordering). Sharing one file DB across processes is not supported: when you need multiple processes or hosts, run the broker (§2) and connect over HTTP — that single broker is the one writer. (:memory: is private per handle, and a remote Turso DB is serialized server-side, so neither takes the lock.)

ctx := context.Background()

// The whole MQ, in your process. file: local SQLite, or libsql://… for Turso.
eng, err := mqlite.OpenEmbedded(ctx, "file:./mq.db")
if err != nil { log.Fatal(err) }
defer eng.Close()

eng.CreateQueue(ctx, "orders", mqlite.QueueConfig{})

// produce
eng.SendOne(ctx, "orders", mqlite.OutMessage{Body: []byte("hello"), GroupID: "order-42"})

// consume (Peek-Lock): handle, then settle. at-least-once → handler must be idempotent.
msgs, _ := eng.Receive(ctx, "orders", mqlite.RecvOpts{Wait: 5 * time.Second})
for _, m := range msgs {
    if err := handle(m.Body); err != nil {
        _ = m.Abandon(ctx)   // release the lock → redelivered (or DLQ past max)
        continue
    }
    _ = m.Complete(ctx)      // remove it (idempotent under retries)
}

// or hands-off: a Receiver auto-settles (nil→Complete, err→Abandon) — still in-process.
eng.Receiver("orders", mqlite.WithConcurrency(4)).
    Run(ctx, func(ctx context.Context, m *mqlite.Message) error { return handle(m.Body) })

⭐ Transactional outbox — the embedded superpower. Because the queue lives in the same SQLite database as your application tables, you can enqueue a message in the same transaction as your business write. No dual-write, no outbox poller, no lost events: either both commit or neither does.

eng.Tx(ctx, func(tx *engine.EngineTx) error {
    tx.SQL().ExecContext(ctx, `INSERT INTO orders_tbl(id) VALUES (1)`)
    _, err := tx.SendOne("orders", engine.OutMessage{Body: []byte("order-created")})
    return err // commit both, or roll back both
})

Outgrow a single process? The same engine upgrades to a network broker with one call — eng.Serve(ctx, ":8080") — and remote clients speak the same semantics over HTTP (§2–§4). Embedded and broker are not two products; they are one engine.

2. Serve a broker
export MQLITE_DB="file:./mq.db"           # or libsql://<db>.turso.io
export MQLITE_DB_AUTH_TOKEN="<jwt>"        # only for remote Turso
export MQLITE_TOKENS="mqk_dev"             # accepted Bearer tokens
mqlite serve --addr :8080
3. Talk to it with curl
TOKEN=mqk_dev
# discovery: hit the root with no auth — a JSON card (name, version, status, endpoints)
curl http://127.0.0.1:8080/

# create the queue first — sending to a queue that doesn't exist is a 404
curl -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  --data '{"name":"orders","config":{}}' \
  http://127.0.0.1:8080/mqlite.v1.AdminService/CreateQueue

# send (body is base64)
curl -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  --data "{\"queue\":\"orders\",\"messages\":[{\"body\":\"$(printf hello | base64)\"}]}" \
  http://127.0.0.1:8080/mqlite.v1.QueueService/Send

# receive (long-poll 5s) → returns lock_token
curl -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  --data '{"queue":"orders","max_messages":1,"wait_time_ms":5000}' \
  http://127.0.0.1:8080/mqlite.v1.QueueService/Receive

Send response. /Send returns {"seq_numbers":[…]} — one per message, in order. A single message that hits a dedup conflict (same message_id, different body) is rejected with 409; in a multi-message batch the conflicting slot comes back as 0 (skipped, not enqueued) while the rest still commit.

Full endpoint, request/response, and error reference: docs/api-reference.md.

3b. Web UI (read-only ops panel)

The broker serves a read-only dashboard at http://<host>/ui — list queues with live counts, browse messages by state, and (the one write action) redrive a DLQ. The page loads without auth; its data calls use the Bearer token you paste in.

4. Or the Go SDK (remote)
cli, _ := mqlite.Open(ctx, "http://127.0.0.1:8080", mqlite.WithToken("mqk_dev"))
seq, _ := cli.SendOne(ctx, "orders", mqlite.OutMessage{Body: []byte("hi")})

// hands-off consumer: nil -> Complete, error -> Abandon, auto lock renewal
cli.Receiver("orders", mqlite.WithAutoRenew(), mqlite.WithConcurrency(4)).
    Run(ctx, func(ctx context.Context, m *mqlite.Message) error {
        return process(m.Body) // MUST be idempotent (at-least-once)
    })
5. CLI
mqlite create-queue orders --lock 30s --max-delivery 5 --dedup 10m --ordering strict_fifo
mqlite send orders "hello" --message-id m1 --group order-42 --reply-to replies
mqlite receive orders --wait 5s
mqlite peek orders --state dead_lettered
mqlite metrics orders
mqlite redrive orders --max 100        # DLQ → active
mqlite purge-dlq orders --older-than 24h # delete dead-lettered messages

Connection is read from the environment:

Env Meaning
MQLITE_DB DB DSN: file:./mq.db, :memory:, or libsql://<db>.turso.io (embedded/serve)
MQLITE_DB_AUTH_TOKEN auth token for a remote libSQL/Turso DSN
MQLITE_ENDPOINT + MQLITE_TOKEN client mode: talk to a running broker (wins if set)
MQLITE_TOKENS comma-separated Bearer tokens that serve accepts
MQLITE_SYNC durability level: NORMAL (default) / FULL / OFF
MQLITE_DLQ_MAX_AGE · MQLITE_DLQ_MAX_COUNT · MQLITE_DLQ_MAX_BYTES broker DLQ retention (defaults 14d / 1,000,000 per queue, drop-oldest; byte cap off by default; MQLITE_DLQ_RETENTION=off disables)

The DB connection string is only ever read from the environment — it is never compiled into the binary. Copy .env.example → .env.local (gitignored).

6. MCP (drive it from an AI agent)

mqlite-mcp is a Model Context Protocol server that exposes the broker as agent tools (create_queue, send, receive, complete, …) — a thin, dependency-free stdio forwarder, in keeping with mqlite's "friendly to AI agents" goal. Point it at a broker with MQLITE_ENDPOINT + MQLITE_TOKEN. See docs/mcp.md.

Documentation

Use it — pick your interface:
CLI reference every command + flags; works embedded or against a broker
HTTP API reference endpoints, request/response, error codes
Subscription filters the expr topic-filter language + message env
MCP server drive mqlite from an AI agent
Examples copy-runnable Go (embedded · outbox · remote) + curl

Operate it: Deployment (Docker/Fly/systemd/Turso) · Observability (Prometheus/Grafana) · Retention · Turso

Understand it: Conformance / TCK · Resource profile · Benchmarks · Dependencies

Durability

mqlite stores everything in SQLite/libSQL with WAL journaling. State is durable across a normal process crash: on restart, in-flight (locked) messages are reclaimed and redelivered (at-least-once), and settled work stays settled.

The one knob is SQLite's synchronous level (database-wide — SQLite has no per-queue setting):

Level Guarantee Cost
NORMAL (default) Durable across an app/process crash. A sudden OS or power loss can drop the last few commits not yet checkpointed to the main DB file. Fast — fsync per WAL checkpoint, not per commit.
FULL Every commit is fsync'd before it returns — survives OS/power loss. Slower under high enqueue throughput.

Set it with the embedded option mqlite.WithSynchronous("FULL"), or MQLITE_SYNC=FULL for the serve/CLI path. Remote Turso DSNs ignore this — durability there is the server's responsibility.

What "no message loss" means here: mqlite is honestly at-least-once, not exactly-once. Under NORMAL, a power cut can lose a handful of just-enqueued messages that hadn't reached disk yet; choose FULL if that window is unacceptable. Once a message is durably written it is delivered at least once and never silently dropped.

Retention

Completed messages are deleted on settle and TTL'd messages expire, so a healthy queue's size tracks its in-flight backlog. The one sink that can grow without bound is the dead-letter queue — so serve bounds it by default: a background pass drops dead letters oldest-first once they are older than 14 days or beyond 1,000,000 per queue (MQLITE_DLQ_MAX_AGE / MQLITE_DLQ_MAX_COUNT; an optional per-queue byte cap MQLITE_DLQ_MAX_BYTES is off by default; MQLITE_DLQ_RETENTION=off disables all). Only the DLQ is ever touched — undelivered and in-flight work is never auto-deleted. The embedded library leaves this off unless you opt in with mqlite.WithDLQRetention(...). Freed pages are reused (the file plateaus at peak backlog rather than shrinking; VACUUM is manual). See docs/retention.md.

Docker

# --pull forces the latest golang:1.25 base (newest Go stdlib security patches) —
# use it for release builds so a cached old base layer can't ship known CVEs.
docker build --platform linux/amd64 --pull -t mqlite:0.1.0 .
docker run --platform linux/amd64 -p 8080:8080 -e MQLITE_TOKENS=mqk_dev mqlite:0.1.0
# remote Turso instead of the local volume:
docker run --platform linux/amd64 -p 8080:8080 \
  -e MQLITE_DB=libsql://<db>.turso.io -e MQLITE_DB_AUTH_TOKEN=<jwt> \
  -e MQLITE_TOKENS=mqk_dev mqlite:0.1.0

Footprint: ~11 MB static (CGO-free) binary, ~19 MB idle RSS, ~0.4 KB per message on disk — fits Fly's smallest 256 MB machine with room to spare. Measured numbers and Fly.io machine/volume sizing: docs/resource-profile.md.

Full copy-runnable recipes — Docker/GHCR, Fly.io (minimal-cost, scale-to-zero), systemd, and Turso — with auth and verification: docs/deployment.md. Wiring /metrics into Prometheus + Grafana (scrape config, PromQL, alerts): docs/observability.md.

Tests

go test ./...                 # hermetic unit + invariant (TCK-style) tests
go test -bench=. ./engine     # embedded throughput benchmarks (msg/s)
go test ./engine -run TestMessageIntegrity      # no-loss + content-hash sweep (default 1k)
MQLITE_INTEGRITY_N=500000 go test ./engine -run TestMessageIntegrity  # large sweep
# live remote round-trip against your own Turso DB:
MQLITE_TEST_DB=libsql://<db>.turso.io MQLITE_TEST_DB_AUTH_TOKEN=<jwt> \
  go test ./engine -run TestTursoIntegration -v

A contiguous 1..N sequence is sent with random bodies (each hashed into a property) and consumed concurrently with redelivery stress; the test asserts every value arrives exactly once (no loss) and every body still matches its hash (no corruption). Local + cloud (Fly) throughput/memory/disk methodology and numbers: docs/benchmark.md. The large sweep also runs weekly in CI.

Status

v0.1 — the core is complete and tested (local SQLite + live Turso): hermetic unit + invariant (TCK-style) tests run in CI on every push, and live Turso/libSQL round-trips run in the nightly workflow. Not yet tagged for release.

License

MIT © mqlitehq

Documentation

Overview

Package mqlite is the native Go SDK for mqlite (design §17.1). The same method set drives two modes: 

Open         — a remote client talking to an mqlite broker over HTTP.
OpenEmbedded — the queue engine in-process (like goqite), with same-DB
               transactional enqueue and a one-line upgrade to a broker.

Settlement methods hang off *Message so the lock token never leaks into user code, and receive comes in two tiers: low-level Receive (you settle) and high-level Receiver.Run (handler returns nil -> Complete, error -> Abandon).

Example (Embedded)

Example_embedded runs the whole queue in-process — no broker, no HTTP — against a local SQLite file, and shows the single-process / single-writer guarantee: a second opener of the same DB is rejected with ErrDBLocked (MQLITE-6 / MQLITE-15). 

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"

	"github.com/mqlitehq/mqlite"
)

func main() {
	ctx := context.Background()
	dir, err := os.MkdirTemp("", "mqlite-example-*")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = os.RemoveAll(dir) }()
	dsn := "file:" + filepath.Join(dir, "mq.db")

	eng, err := mqlite.OpenEmbedded(ctx, dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = eng.Close() }()

	if err := eng.CreateQueue(ctx, "orders", mqlite.QueueConfig{}); err != nil {
		log.Fatal(err)
	}
	if _, err := eng.SendOne(ctx, "orders", mqlite.OutMessage{Body: []byte("order-42")}); err != nil {
		log.Fatal(err)
	}

	msgs, err := eng.Receive(ctx, "orders", mqlite.RecvOpts{})
	if err != nil {
		log.Fatal(err)
	}
	for _, m := range msgs {
		fmt.Printf("received: %s\n", m.Body)
		_ = m.Complete(ctx) // at-least-once; the handler must be idempotent
	}

	// Embedded mode is single-writer: a second process — or a second OpenEmbedded on
	// the same file — is rejected rather than racing the first.
	if _, err := mqlite.OpenEmbedded(ctx, dsn); errors.Is(err, mqlite.ErrDBLocked) {
		fmt.Println("second open rejected: single writer")
	}

}

Output:
received: order-42
second open rejected: single writer

Index

Examples

Constants

View Source 
const (
	OrderStandard   = engine.OrderStandard
	OrderGroupFIFO  = engine.OrderGroupFIFO
	OrderStrictFIFO = engine.OrderStrictFIFO
)
View Source 
const (
	Active       = engine.StateActive
	Locked       = engine.StateLocked
	Deferred     = engine.StateDeferred
	Scheduled    = engine.StateScheduled
	DeadLettered = engine.StateDeadLettered
)
View Source 
const TokenPrefix = "mqk_"

TokenPrefix is the conventional prefix for a broker auth token, so a token is recognizable at a glance (and greppable in logs/config). It is a convention, not a requirement — any non-empty string in MQLITE_TOKENS is accepted as a token.

Variables

View Source 
var (
	ErrLockLost        = engine.ErrLockLost
	ErrNotFound        = engine.ErrNotFound
	ErrQueueNotFound   = engine.ErrQueueNotFound
	ErrDedupConflict   = engine.ErrDedupConflict
	ErrMessageTooLarge = engine.ErrMessageTooLarge
	ErrNameConflict    = engine.ErrNameConflict
	ErrGroupRequired   = engine.ErrGroupRequired
	// ErrDBLocked is returned by OpenEmbedded when the local file DB is already open
	// in another process: embedded mode is single-process, single-writer (MQLITE-6).
	ErrDBLocked = engine.ErrDBLocked
)

Re-exported sentinel errors so callers can use errors.Is on either mode.

Functions

func GenerateToken added in v0.1.1 

func GenerateToken() string

GenerateToken mints a fresh broker auth token: the "mqk_" prefix plus 128 bits of crypto/rand, hex-encoded (e.g. "mqk_" + 32 hex chars). Use it to seed MQLITE_TOKENS / WithTokens, or let `mqlite serve` call it automatically when no token is configured (secure by default). 128 bits is unguessable; bump the byte count here if a longer secret is ever wanted.

Types

type AbandonOpts 

type AbandonOpts struct {
	Delay time.Duration // re-hide for this long before redelivery (backoff)
}

AbandonOpts configures a Message.Abandon call.

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is a remote mqlite client (Connect-style JSON over HTTP).

func Open 

func Open(ctx context.Context, dsn string, opts ...Option) (*Client, error)

Open builds a client from a connection string: mqlite://<token>@host:port?tls=true

func (*Client) Cancel 

func (c *Client) Cancel(ctx context.Context, queue string, seq int64) error

Cancel deletes a not-yet-activated scheduled message.

func (*Client) Close 

func (c *Client) Close() error

Close releases client resources (no-op for HTTP).

func (*Client) CompleteBatch added in v0.1.1 

func (c *Client) CompleteBatch(ctx context.Context, queue string, msgs ...*Message) ([]SettleResult, error)

CompleteBatch completes many received messages in one round-trip — the fix for the drain N+1 (settling a received batch one HTTP call at a time). Messages should share one queue; a stale lock comes back as Ok=false rather than failing the batch.

func (*Client) CreateQueue 

func (c *Client) CreateQueue(ctx context.Context, name string, cfg QueueConfig) error

CreateQueue creates or updates a queue.

func (*Client) ListQueues 

func (c *Client) ListQueues(ctx context.Context) ([]QueueInfo, error)

ListQueues lists queues/subscriptions.

func (*Client) Peek 

func (c *Client) Peek(ctx context.Context, queue string, opts ...PeekOpts) ([]*PeekedMessage, error)

Peek browses without locking.

func (*Client) Purge 

func (c *Client) Purge(ctx context.Context, queue string, opts ...PurgeOpts) (int, error)

Purge permanently deletes dead-lettered messages (PurgeOpts scopes it; no opts purges the whole DLQ). Returns count deleted.

func (*Client) Receive 

func (c *Client) Receive(ctx context.Context, queue string, opts ...RecvOpts) ([]*Message, error)

Receive claims up to N messages (Peek-Lock by default), with optional long-poll. RecvOpts.Pick fetches previously-deferred messages by seq instead of claiming.

func (*Client) Receiver 

func (c *Client) Receiver(queue string, opts ...ReceiverOption) *Receiver

Receiver returns a stateful receive loop bound to this client.

func (*Client) Redrive 

func (c *Client) Redrive(ctx context.Context, dlqQueue string, opts ...RedriveOpts) (int, error)

Redrive moves dead-lettered messages back to active.

func (*Client) Send 

func (c *Client) Send(ctx context.Context, queue string, msgs ...OutMessage) ([]int64, error)

Send enqueues one or many messages in one request/transaction (use SendOne for a single seq return or At() scheduling).

func (*Client) SendOne 

func (c *Client) SendOne(ctx context.Context, queue string, m OutMessage, opts ...SendOpts) (int64, error)

SendOne enqueues one message and returns its seq. A dedup conflict (same id, different body) surfaces as ErrDedupConflict. SendOpts.At schedules delayed delivery.

func (*Client) Stats 

func (c *Client) Stats(ctx context.Context, queue string) (Metrics, error)

Stats returns counters for a queue.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, topic, name string, f *Filter) error

Subscribe registers a subscription under a topic with an optional filter.

type Embedded 

type Embedded struct {
	// contains filtered or unexported fields
}

Embedded runs the queue engine in-process (like goqite), exposing the same method set as the remote Client plus same-DB transactional enqueue (Tx) and a one-line upgrade to a network broker (Serve). See design §4.5.

func OpenEmbedded 

func OpenEmbedded(ctx context.Context, dbDSN string, opts ...EmbeddedOption) (*Embedded, error)

OpenEmbedded opens the engine on a DB DSN: file:./mq.db | :memory: | libsql://host

func (*Embedded) Cancel 

func (e *Embedded) Cancel(ctx context.Context, queue string, seq int64) error

Cancel deletes a not-yet-activated scheduled message.

func (*Embedded) Close 

func (e *Embedded) Close() error

Close stops background loops and closes the DB.

func (*Embedded) Compact added in v0.1.1 

func (e *Embedded) Compact(ctx context.Context, full bool) error

Compact reclaims free DB pages to the OS: `PRAGMA incremental_vacuum` (no global lock; new local DBs default to auto_vacuum=INCREMENTAL), or a full `VACUUM` when full=true (rewrites the file, global lock — maintenance only). Local-only.

func (*Embedded) CompleteBatch added in v0.1.1 

func (e *Embedded) CompleteBatch(ctx context.Context, queue string, msgs ...*Message) ([]SettleResult, error)

CompleteBatch completes many received messages in one transaction (mirrors the remote Client method; for the in-process engine there is no round-trip to save, but it keeps the API symmetric and settles atomically).

func (*Embedded) CreateQueue 

func (e *Embedded) CreateQueue(ctx context.Context, name string, cfg QueueConfig) error

func (*Embedded) Engine 

func (e *Embedded) Engine() *engine.Engine

Engine exposes the underlying engine (advanced/embedded use).

func (*Embedded) ListQueues 

func (e *Embedded) ListQueues(ctx context.Context) ([]QueueInfo, error)

func (*Embedded) Peek 

func (e *Embedded) Peek(ctx context.Context, queue string, opts ...PeekOpts) ([]*PeekedMessage, error)

func (*Embedded) Purge 

func (e *Embedded) Purge(ctx context.Context, queue string, opts ...PurgeOpts) (int, error)

Purge permanently deletes dead-lettered messages (PurgeOpts scopes it; no opts purges the whole DLQ). Returns count deleted.

func (*Embedded) Receive 

func (e *Embedded) Receive(ctx context.Context, queue string, opts ...RecvOpts) ([]*Message, error)

func (*Embedded) Receiver 

func (e *Embedded) Receiver(queue string, opts ...ReceiverOption) *Receiver

Receiver returns a stateful receive loop bound to the embedded engine.

func (*Embedded) Redrive 

func (e *Embedded) Redrive(ctx context.Context, dlqQueue string, opts ...RedriveOpts) (int, error)

func (*Embedded) Send 

func (e *Embedded) Send(ctx context.Context, queue string, msgs ...OutMessage) ([]int64, error)

Send enqueues one or many messages in one transaction.

func (*Embedded) SendOne 

func (e *Embedded) SendOne(ctx context.Context, queue string, m OutMessage, opts ...SendOpts) (int64, error)

SendOne enqueues one message. A dedup conflict surfaces as ErrDedupConflict. SendOpts.At schedules delayed delivery.

func (*Embedded) Serve 

func (e *Embedded) Serve(ctx context.Context, addr string, opts ...ServeOption) error

Serve exposes this engine as an HTTP broker until ctx is canceled.

func (*Embedded) Stats 

func (e *Embedded) Stats(ctx context.Context, queue string) (Metrics, error)

func (*Embedded) Subscribe 

func (e *Embedded) Subscribe(ctx context.Context, topic, name string, f *Filter) error

func (*Embedded) Tx 

func (e *Embedded) Tx(ctx context.Context, fn func(*engine.EngineTx) error) error

Tx runs business writes and enqueues in one transaction (§4.5, embedded-only).

type EmbeddedOption 

type EmbeddedOption func(*embeddedConfig)

EmbeddedOption configures OpenEmbedded.

func WithClock 

func WithClock(now func() int64) EmbeddedOption

WithClock injects a test clock returning epoch ms.

func WithDBAuthToken 

func WithDBAuthToken(tok string) EmbeddedOption

WithDBAuthToken supplies the auth token for a remote libSQL/Turso DSN. The token is passed in by the caller (typically from an env var) — never compiled in.

func WithDLQRetention added in v0.1.1 

func WithDLQRetention(maxAge time.Duration, maxCount int, maxBytes int64) EmbeddedOption

WithDLQRetention bounds the dead-letter queue so a long-running broker can't grow without limit (MQLITE-21). A background pass drops dead letters oldest-first when they are older than maxAge, beyond maxCount per queue, or over maxBytes of body per queue; pass 0 to leave a dimension unbounded. ONLY the DLQ is affected — undelivered and in-flight messages are never deleted. Off by default for the embedded library; the `mqlite serve` broker enables a sane default (see cmd/mqlite).

func WithLogger 

func WithLogger(lg *slog.Logger) EmbeddedOption

WithLogger routes background-loop failures to lg (nil -> slog.Default()).

func WithMaxMessageBytes 

func WithMaxMessageBytes(n int64) EmbeddedOption

WithMaxMessageBytes caps message body size (0 -> 1 MiB default).

func WithSynchronous 

func WithSynchronous(mode string) EmbeddedOption

WithSynchronous sets the local SQLite PRAGMA synchronous level — the durability vs throughput knob (MQLITE-7). "NORMAL" (default) is fast and durable across a process crash, but a sudden OS/power loss can drop the last few not-yet- checkpointed commits; "FULL" fsyncs every commit (safer, slower). Database-wide (SQLite has no per-queue sync); ignored for remote Turso DSNs.

func WithoutBackground 

func WithoutBackground() EmbeddedOption

WithoutBackground disables the maintenance loops (tests drive them manually).

type Filter 

type Filter = engine.Filter

Filter is a subscription filter (equality-AND + subject prefix).

type Message 

type Message struct {
	SequenceNumber int64
	Body           []byte
	MessageID      string
	GroupID        string
	CorrelationID  string
	ReplyTo        string
	Subject        string
	ContentType    string
	Properties     map[string]string
	DeliveryCount  int
	EnqueuedAt     time.Time
	LockedUntil    time.Time
	// contains filtered or unexported fields
}

Message is a delivered message handle. The lock token is held internally and never exposed, so settlement can't be misrouted (§17.1).

func (*Message) Abandon 

func (m *Message) Abandon(ctx context.Context, opts ...AbandonOpts) error

Abandon releases the lock for immediate redelivery (optionally after a delay).

func (*Message) Complete 

func (m *Message) Complete(ctx context.Context) error

Complete removes a successfully-processed message.

func (*Message) Defer 

func (m *Message) Defer(ctx context.Context) error

Defer sets the message aside, to be retrieved later by SequenceNumber.

func (*Message) Reject 

func (m *Message) Reject(ctx context.Context, opts ...RejectOpts) error

Reject moves the message to the dead-letter state (optionally with a reason).

func (*Message) Renew 

func (m *Message) Renew(ctx context.Context) error

Renew extends the lock lease (for long-running handlers).

type Metrics 

type Metrics = engine.Metrics

Metrics mirrors engine.Metrics with the same fields.

type Option 

type Option func(*config)

Option configures Open.

func WithHTTPClient 

func WithHTTPClient(h *http.Client) Option

WithHTTPClient supplies a custom *http.Client (e.g. for TLS config).

func WithToken 

func WithToken(tok string) Option

WithToken overrides the token from the connection string.

type OrderingMode 

type OrderingMode = engine.OrderingMode

OrderingMode mirrors engine.OrderingMode for queue-level delivery ordering.

type OutMessage 

type OutMessage struct {
	Body      []byte
	MessageID string // dedup/idempotency key; empty -> body SHA-256 when dedup on
	// GroupID is an ordering/partition key (= SQS MessageGroupId, ASB SessionId):
	// same GroupID = strict in-order (FIFO per group); empty = own group (max
	// parallelism). NOT a consumer group — competing consumers just Receive the
	// same queue; peek-lock gives each message to exactly one.
	GroupID       string
	CorrelationID string
	ReplyTo       string // = ASB ReplyTo; opaque address the consumer should reply to
	Subject       string // = ASB Label
	ContentType   string
	Properties    map[string]string // custom KV (headers)
	TTL           time.Duration     // 0 -> queue default
}

OutMessage is a message to send. Body is opaque; broker never parses it.

type PeekOpts 

type PeekOpts struct {
	From  int64 // start browsing at this seq
	State State // filter by state
	Max   int   // cap results
}

PeekOpts configures a Peek call.

type PeekedMessage 

type PeekedMessage struct {
	SequenceNumber int64
	State          State
	Body           []byte
	MessageID      string
	GroupID        string
	CorrelationID  string
	ReplyTo        string
	Subject        string
	ContentType    string
	Properties     map[string]string
	DeliveryCount  int
	EnqueuedAt     time.Time
	VisibleAt      time.Time
	LockedUntil    time.Time
}

PeekedMessage is a read-only browse result.

type PurgeOpts 

type PurgeOpts struct {
	Max       int           // cap how many messages are deleted
	OlderThan time.Duration // only delete messages older than this
}

PurgeOpts configures a Purge call.

type QueueConfig 

type QueueConfig struct {
	LockDuration       time.Duration
	MaxDeliveryCount   int
	DefaultTTL         time.Duration
	DeadLetterOnExpire *bool
	DedupWindow        time.Duration
	Ordering           OrderingMode // "" -> standard
	// Per-queue DLQ retention overrides (MQLITE-29). For each: 0 -> inherit the
	// broker default; >0 -> this queue's drop-oldest bound; a negative value ->
	// explicitly unbounded (opt out of the default).
	DLQMaxAge   time.Duration
	DLQMaxCount int
	DLQMaxBytes int64
}

QueueConfig configures a queue (durations as time.Duration).

type QueueInfo 

type QueueInfo = engine.QueueInfo

QueueInfo mirrors engine.QueueInfo.

type Receiver 

type Receiver struct {
	// contains filtered or unexported fields
}

Receiver is a stateful receive loop. handler returning nil auto-Completes; returning error auto-Abandons (§17.1).

func (*Receiver) Close 

func (r *Receiver) Close() error

Close releases receiver resources (no-op).

func (*Receiver) Run 

func (r *Receiver) Run(ctx context.Context, handler func(context.Context, *Message) error) error

Run pulls and processes messages until ctx is canceled.

type ReceiverOption 

type ReceiverOption func(*receiverConfig)

ReceiverOption configures a Receiver.

func WithAutoRenew 

func WithAutoRenew() ReceiverOption

WithAutoRenew keeps locks alive with a background heartbeat for long handlers.

func WithConcurrency 

func WithConcurrency(n int) ReceiverOption

WithConcurrency sets how many messages are processed in parallel.

func WithPrefetch 

func WithPrefetch(n int) ReceiverOption

WithPrefetch sets how many messages to claim per receive (defaults to concurrency).

type RecvOpts 

type RecvOpts struct {
	Max        int           // most messages to claim (0 -> 1)
	Wait       time.Duration // long-poll wait (0 -> don't wait); capped at 20s
	AtMostOnce bool          // receive-and-delete (no lock, no settle)
	Attempt    string        // idempotent-receive key; a retry replays the same batch
	Pick       []int64       // fetch these deferred seqs by seq instead of claiming
}

RecvOpts configures a Receive call.

type RedriveOpts 

type RedriveOpts struct {
	To        string        // target queue (empty -> back to source)
	Max       int           // cap how many messages move
	OlderThan time.Duration // only move messages older than this
	Rate      int           // throughput limit per second
}

RedriveOpts configures a Redrive call.

type RejectOpts 

type RejectOpts struct {
	Reason string // dead-letter reason
	Detail string // dead-letter description
}

RejectOpts configures a Message.Reject call.

type SendOpts 

type SendOpts struct {
	At time.Time // schedule delivery for t (zero -> immediate)
}

SendOpts configures a SendOne call.

type ServeOption 

type ServeOption func(*serveConfig)

ServeOption configures Serve.

func WithTokenCSV 

func WithTokenCSV(csv string) ServeOption

WithTokenCSV sets accepted Bearer tokens from a comma-separated string (env-friendly).

func WithTokens 

func WithTokens(tokens ...string) ServeOption

WithTokens sets the accepted Bearer tokens for the broker.

func WithVersion added in v0.1.1 

func WithVersion(v string) ServeOption

WithVersion sets the version string the broker reports on its open "/" discovery endpoint (typically the CLI/build version).

type SettleResult added in v0.1.1 

type SettleResult struct {
	SequenceNumber int64
	Ok             bool
}

SettleResult is the per-message outcome of CompleteBatch (Ok=false = the lock was lost/expired for that message — not a fatal error for the batch).

type State 

type State = engine.State

State mirrors engine.State for Peek filtering.

Source Files

View all Source files

Directories

Path Synopsis
cmd
mqlite command
Command mqlite is the single-binary CLI: run a broker, or produce/consume/ administer queues against a local DB (embedded) or a remote broker (client).
 Command mqlite is the single-binary CLI: run a broker, or produce/consume/ administer queues against a local DB (embedded) or a remote broker (client).
mqlite-mcp command
Command mqlite-mcp is a Model Context Protocol (MCP) server that exposes the mqlite broker as agent tools.
 Command mqlite-mcp is a Model Context Protocol (MCP) server that exposes the mqlite broker as agent tools.
Package server exposes an mqlite Engine as a Connect-style JSON-over-HTTP broker (design §7).
 Package server exposes an mqlite Engine as a Connect-style JSON-over-HTTP broker (design §7).
test
bench command
Command mqlite-bench stress-tests the local SQLite engine (embedded, no HTTP) across several high-frequency request patterns, and attributes per-scenario CPU and disk I/O via /proc/self (Linux/Docker).
 Command mqlite-bench stress-tests the local SQLite engine (embedded, no HTTP) across several high-frequency request patterns, and attributes per-scenario CPU and disk I/O via /proc/self (Linux/Docker).
sdkcheck command
Command sdkcheck is the SDK end-to-end suite (the "用 SDK" path).
 Command sdkcheck is the SDK end-to-end suite (the "用 SDK" path).
Package wire defines the JSON-over-HTTP contract shared by the mqlite broker (server) and the Go SDK client.
 Package wire defines the JSON-over-HTTP contract shared by the mqlite broker (server) and the Go SDK client.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.