engine

package
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: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ReasonMaxDeliveryCount = "MaxDeliveryCountExceeded"
	ReasonTTLExpired       = "TTLExpired"
	ReasonAppRequested     = "AppRequested"
)

Dead-letter reasons.

View Source 
const DefaultMaxMessageBytes = 1 << 20

DefaultMaxMessageBytes is the default body-size cap (1 MiB, design §14-Q7).

Variables

View Source 
var (
	ErrQueueNotFound         = errors.New("mqlite: queue not found")
	ErrLockLost              = errors.New("mqlite: lock lost or already settled")
	ErrDedupConflict         = errors.New("mqlite: dedup conflict (same id, different body)")
	ErrNotFound              = errors.New("mqlite: not found")
	ErrClosed                = errors.New("mqlite: engine closed")
	ErrMessageTooLarge       = errors.New("mqlite: message body exceeds max size")
	ErrNameConflict          = errors.New("mqlite: name already in use by another queue or topic")
	ErrGroupRequired         = errors.New("mqlite: group id required for group_fifo queue")
	ErrDBLocked              = errors.New("mqlite: database file is already open by another process")
	ErrSchemaVersionMismatch = errors.New("mqlite: database schema version is incompatible with this build")
	ErrInvalidFilter         = errors.New("mqlite: invalid subscription filter expression")
)

Sentinel errors. The server maps these onto Connect/HTTP error codes.

Functions

This section is empty.

Types

type Engine 

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

Engine is the embeddable queue core: a Store plus Service Bus semantics, independent of any network/transport (design §12.3). Both the in-process embedded API and the network broker drive this same type.

func Open 

func Open(ctx context.Context, opts Options) (*Engine, error)

Open opens the store (initializing the schema), performs single-broker crash recovery, and starts background maintenance loops.

func (*Engine) Abandon 

func (e *Engine) Abandon(ctx context.Context, queue string, seq int64, token string, delayMs int64) error

Abandon releases the lock and redelivers, or dead-letters if delivery_count has reached the queue's max (§8.5). delayMs applies exponential-backoff style re-visibility when requeued.

func (*Engine) Cancel 

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

Cancel deletes a not-yet-activated scheduled message by seq.

func (*Engine) Close 

func (e *Engine) Close() error

func (*Engine) Compact added in v0.1.1 

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

Compact reclaims free database pages to the OS (MQLITE-31). New local DBs use auto_vacuum=INCREMENTAL, so the default runs `PRAGMA incremental_vacuum` — bounded, no global lock, janitor-friendly. full=true runs a full `VACUUM`, which rewrites the whole file and holds a global write lock (a maintenance-window operation). Both are local-only — a remote Turso/libSQL store manages its own storage.

func (*Engine) Complete 

func (e *Engine) Complete(ctx context.Context, queue string, seq int64, token string) error

Complete removes a successfully-processed message (fencing on lock_token).

func (*Engine) CompleteBatch added in v0.1.1 

func (e *Engine) CompleteBatch(ctx context.Context, queue string, items []SettleItem) ([]SettleResult, error)

CompleteBatch completes many messages in one transaction / one round-trip — the broker-side fix for the drain N+1 (Receive returns a batch, but settling it one-by-one is one HTTP call per message). Each item is fenced on its own lock_token and recorded idempotently, exactly like Complete; a per-item failure (expired/wrong token) returns Ok=false rather than failing the whole batch.

func (*Engine) CreateQueue 

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

CreateQueue creates a queue (idempotent on name; updates config if exists).

func (*Engine) Defer 

func (e *Engine) Defer(ctx context.Context, queue string, seq int64, token string) error

Defer sets a message aside; it is later retrieved by seq via ReceiveDeferred.

func (*Engine) ListQueues 

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

ListQueues lists all queues/subscriptions.

func (*Engine) Peek 

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

Peek browses messages without locking or settling them (§7.3). Useful for triage and for recovering the seq_number of a deferred message.

func (*Engine) Purge 

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

Purge permanently deletes dead-lettered messages from a queue's DLQ (§7.3). Use after a redrive, or to discard poison messages that will never be reprocessed. opts.Max / opts.OlderThanMs scope the purge; both zero purges the whole DLQ. Returns the number of rows deleted.

func (*Engine) Receive 

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

Receive claims up to opts.MaxMessages messages (Peek-Lock by default), with long-poll up to opts.WaitMs (clamped to 20s, §11.3).

func (*Engine) ReceiveDeferred 

func (e *Engine) ReceiveDeferred(ctx context.Context, queue string, seqs ...int64) ([]*Message, error)

ReceiveDeferred locks previously-deferred messages by seq_number (§8.7).

func (*Engine) Redrive 

func (e *Engine) Redrive(ctx context.Context, dlq string, opts RedriveOptions) (int, error)

Redrive moves dead-lettered messages back to active for reprocessing (§11.2).

  • Same-queue redrive is an in-place UPDATE preserving rowid (FIFO replay order).
  • Cross-queue redrive re-INSERTs into the target (fresh rowid) and deletes the originals. The whole move is ONE transaction (atomic): a crash mid-redrive leaves every message either fully moved or fully in the DLQ — never half.
  • RatePerSec>0 throttles the move into per-second chunks (each chunk atomic), so draining a huge DLQ does not flood the target. RatePerSec<=0 moves everything in a single atomic transaction.

func (*Engine) Reject 

func (e *Engine) Reject(ctx context.Context, queue string, seq int64, token, reason, desc string) error

Reject moves a locked message to the dead-letter state with a reason.

func (*Engine) Remote 

func (e *Engine) Remote() bool

Remote reports whether the underlying store is a remote Turso/libSQL DB.

func (*Engine) Renew 

func (e *Engine) Renew(ctx context.Context, queue string, seq int64, token string) error

Renew extends the lock lease by the queue's lock duration (§11.3).

func (*Engine) RunMaintenanceOnce 

func (e *Engine) RunMaintenanceOnce(ctx context.Context)

RunMaintenanceOnce runs every maintenance pass synchronously. Tests with DisableBackground use this to drive time-based transitions deterministically.

func (*Engine) Schedule 

func (e *Engine) Schedule(ctx context.Context, queue string, m OutMessage, atMs int64) (int64, error)

Schedule enqueues a message that becomes visible at `atMs` (epoch ms) (§8.7). As with SendOne, a dedup conflict surfaces as ErrDedupConflict while a publish that no subscription accepts is a no-op returning (0, nil).

func (*Engine) Send 

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

Send enqueues one or many messages in one transaction (§11.3 Batch).

func (*Engine) SendOne 

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

SendOne enqueues one message, returning its seq_number. A dedup conflict (same id, different body) surfaces as ErrDedupConflict. Publishing to a topic that no subscription filter accepts is a valid no-op: it returns (0, nil).

func (*Engine) Stats 

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

Stats returns pgmq-style counters for a queue (§7.3).

func (*Engine) Subscribe 

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

Subscribe registers subscription `name` under `topic`, creating the subscription's backing queue and the fan-out mapping (§10.1).

func (*Engine) Tx 

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

Tx runs fn inside one transaction. If fn returns nil the transaction commits (and long-poll waiters for any written queue are notified); otherwise it rolls back.

type EngineTx 

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

EngineTx is a transaction handle for same-DB transactional enqueue (§4.5). Business-table writes and queue enqueues commit together: "business success ⇔ message enqueued" — a natural outbox with no distributed-commit dilemma. Only valid in embedded mode against a local file / single libSQL connection.

func (*EngineTx) SQL 

func (t *EngineTx) SQL() *sql.Tx

SQL returns the underlying *sql.Tx so callers can run their business writes in the same transaction as the enqueue.

func (*EngineTx) SendOne 

func (t *EngineTx) SendOne(queue string, m OutMessage) (int64, error)

SendOne enqueues a message inside the open transaction.

type Filter 

type Filter struct {
	Expr string `json:"expr,omitempty"`
}

Filter is a subscription filter: a single expr-lang boolean predicate over a message. An empty Expr matches every message. The expression is type-checked and compiled once at Subscribe (a bad one is rejected with ErrInvalidFilter) and run per message at publish (fan-out). The language surface — the variables a filter sees, the helpers, and the fail-closed evaluator — lives in filter.go.

This replaces the earlier equality-AND + subject-prefix form: a filter is now just `subject_parts[0] == "orders" && properties["tier"] == "gold"` (MQLITE-17).

type Message 

type Message struct {
	SeqNumber     int64
	Body          []byte
	MessageID     string
	GroupID       string
	CorrelationID string
	ReplyTo       string
	Subject       string
	ContentType   string
	Properties    map[string]string
	DeliveryCount int
	EnqueuedAtMs  int64
	LockedUntilMs int64
	LockToken     string // fencing token; echoed back on settle
}

Message is a delivered message carrying the lock token (Peek-Lock).

type Metrics 

type Metrics struct {
	Queue              string
	Active             int64
	Locked             int64
	Deferred           int64
	Scheduled          int64
	DeadLettered       int64
	Total              int64
	OldestMessageAgeMs int64
}

Metrics mirrors pgmq-style queue counters (§7.3 Stats).

type Options 

type Options struct {
	DB                string       // DB DSN: file:./mq.db | :memory: | libsql://host
	AuthToken         string       // injected from env; never compiled in
	Now               func() int64 // test clock (epoch ms)
	DisableBackground bool         // skip reaper/scheduler loops (tests)
	Synchronous       string       // local SQLite PRAGMA synchronous: NORMAL(default)|FULL|OFF
	MaxMessageBytes   int64        // reject bodies larger than this; 0 -> 1 MiB (§11.4)
	// DLQ retention bounds (MQLITE-21): a background pass drops dead letters
	// oldest-first past these. ONLY state='dead_lettered' is touched; 0 = that
	// dimension unbounded. Defaults are applied by the broker, not the engine.
	DLQMaxAgeMs int64        // dead letters older than this (by enqueued_at) are dropped
	DLQMaxCount int          // keep at most this many dead letters per queue (drop oldest)
	DLQMaxBytes int64        // cap total dead-letter body bytes per queue (drop oldest)
	Logger      *slog.Logger // background-loop failures log here; nil -> slog.Default()
}

Options configures Open.

type OrderingMode 

type OrderingMode string

OrderingMode is a queue-level delivery-ordering policy. 

OrderStandard   — best-effort FIFO per group; ungrouped messages are each
                  their own group (max parallelism). The default.
OrderGroupFIFO  — strict in-order delivery within a group; GroupID is
                  required on every message (claim behaves like standard).
OrderStrictFIFO — strict single-flight global FIFO: the queue head blocks
                  the whole queue until it is settled (no grouping).

const (
	OrderStandard   OrderingMode = "standard"
	OrderGroupFIFO  OrderingMode = "group_fifo"
	OrderStrictFIFO OrderingMode = "strict_fifo"
)

type OutMessage 

type OutMessage struct {
	Body      []byte
	MessageID string // dedup / idempotency key; empty -> body SHA-256 used when dedup on
	// GroupID is an ordering / partition key (= SQS MessageGroupId, ASB SessionId):
	// messages sharing a GroupID are delivered strictly in-order (FIFO per group);
	// empty -> the message is its own group (max parallelism). It is NOT a consumer
	// group — competing consumers just Receive the same queue and peek-lock hands
	// each message to exactly one of them.
	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), JSON-encoded; broker does not interpret
	TTLMs         int64             // 0 -> use queue default
}

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

type PeekOptions 

type PeekOptions struct {
	FromSeq int64
	State   State // empty -> any state
	Max     int
}

PeekOptions controls a Peek call.

type PeekedMessage 

type PeekedMessage struct {
	SeqNumber             int64
	State                 State
	Body                  []byte
	MessageID             string
	GroupID               string
	CorrelationID         string
	ReplyTo               string
	Subject               string
	ContentType           string
	Properties            map[string]string
	DeliveryCount         int
	EnqueuedAtMs          int64
	VisibleAtMs           int64
	LockedUntilMs         int64
	DeadLetterReason      string
	DeadLetterDescription string
}

PeekedMessage is a read-only browse result (no lock, cannot be settled).

type QueueConfig 

type QueueConfig struct {
	Kind               string       // "queue" (default) or "subscription"
	LockDurationMs     int64        // Peek-Lock default lock duration; 0 -> 30000
	MaxDeliveryCount   int          // 0 -> 10
	DefaultTTLMs       int64        // 0 -> unlimited
	DeadLetterOnExpire *bool        // nil -> true
	DedupWindowMs      int64        // 0 -> dedup disabled
	Ordering           OrderingMode // "" -> standard
	// Per-queue DLQ retention overrides (MQLITE-29). For each: 0 -> inherit the
	// broker/engine default; >0 -> this queue's own drop-oldest bound; -1 ->
	// explicitly unbounded (opt out of the default).
	DLQMaxAgeMs int64 // dead letters older than this (by enqueued_at) are dropped
	DLQMaxCount int   // keep at most this many dead letters in this queue
	DLQMaxBytes int64 // cap total dead-letter body bytes in this queue
}

QueueConfig configures a queue or subscription (entity-level defaults). Zero values mean "use the documented default".

type QueueInfo 

type QueueInfo struct {
	Name             string
	Kind             string
	LockDurationMs   int64
	MaxDeliveryCount int
	DefaultTTLMs     int64
	DedupWindowMs    int64
}

QueueInfo describes a queue for listing.

type ReceiveMode 

type ReceiveMode int

ReceiveMode selects Peek-Lock (default) or Receive-and-Delete (at-most-once fast path). 

const (
	PeekLock ReceiveMode = iota
	ReceiveAndDelete
)

type ReceiveOptions 

type ReceiveOptions struct {
	MaxMessages int
	WaitMs      int64 // long-poll up to 20000
	Mode        ReceiveMode
	// AttemptID, when set, makes Receive idempotent under client retries: a retry
	// with the same id replays the same batch (same lock tokens) instead of
	// claiming new messages / burning delivery_count (SQS ReceiveRequestAttemptId).
	AttemptID string
}

ReceiveOptions controls a Receive call.

type RedriveOptions 

type RedriveOptions struct {
	Target      string // empty -> back to source queue (in-place); else cross-queue re-INSERT
	Max         int
	OlderThanMs int64
	RatePerSec  int
}

RedriveOptions controls a Redrive call (§11.2).

type SettleItem added in v0.1.1 

type SettleItem struct {
	SeqNumber int64
	LockToken string
}

SettleItem identifies one message to settle in a batch (fenced on LockToken).

type SettleResult added in v0.1.1 

type SettleResult struct {
	SeqNumber int64
	Ok        bool // true = settled (or an idempotent replay of an already-settled token)
}

SettleResult is the per-item outcome of a batch settle.

type State 

type State string

State is the lifecycle state of a message (design §6). 

const (
	StateActive       State = "active"
	StateLocked       State = "locked"
	StateDeferred     State = "deferred"
	StateScheduled    State = "scheduled"
	StateCompleted    State = "completed"
	StateDeadLettered State = "dead_lettered"
)

Source Files

View all Source files

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.