queue

package
v1.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Overview

Package queue is a durable, pluggable message queue with Kafka-shaped semantics: producers publish messages to a topic, and every consumer group subscribed to that topic receives every message, while the consumers inside one group compete for it.

Interfaces

Callers depend only on Queue. Two kinds of implementation sit behind it:

  • A broker (Kafka, SQS, ...) implements Queue directly and registers itself with Register. The framework never imports a broker client.
  • A database-backed queue implements the narrow Store SPI and lets the generic engine provide polling, concurrency, lease heartbeats, retries with backoff and dead-lettering. The built-in "logstore" type stores messages in the logstore database (SQLite, Postgres or ClickHouse); the "memory" type keeps them in process.

Delivery guarantees

Delivery is at least once. A handler that returns nil acknowledges the delivery; an error schedules a retry with exponential backoff; an error wrapped with Permanent, or the final allowed attempt, dead-letters it. A delivery whose lease is lost (the pod paused past the lease, or crashed) is redelivered to another consumer, so handlers must be idempotent. Delivery.DeliveryID is stable across redeliveries of the same message to the same group and is the intended deduplication key.

Ordering

Messages without a key are delivered in best-effort publish order. Messages that carry a Message.Key are delivered strictly in order per (topic, group, key): the next message for a key is not handed out until the previous one is acknowledged or dead-lettered, including while it waits out a retry backoff. Order among concurrent publishers of the same key is the order in which their writes commit.

Scheduling and limits

A message with Message.DeliverAt set is not handed out before that time; it runs on the first poll after it is due, so within the subscription's poll interval. It can be scheduled at most the configured retention ahead. A scheduled message that carries a key keeps its place in the key's order: later messages with the same key wait for it.

WithConcurrency bounds the handlers one instance runs for a subscription. WithMaxInFlight bounds a consumer group's deliveries in flight across every instance, counted in the store, so it holds however many instances consume the group.

Multiple instances

The Postgres and ClickHouse backends are safe to share between any number of Bifrost instances. Leases and retry times are computed on the database clock, every write after a claim is fenced on the claim's token, and per-key ordering is evaluated against committed state. ClickHouse has no row locks or transactions, so consuming on ClickHouse requires a shared distributed locker, through which claims, publishes and group registrations on a topic are also serialised. An instance without one can still publish, serialising only its own publishes, so per-key order between two such publishers is not guaranteed. The SQLite and memory backends are single-process only.

Index

Constants

View Source 
const (
	DefaultRetentionHours      = 168
	DefaultAckedRetentionHours = 24
	DefaultJanitorInterval     = "10m"
	DefaultMaxPayloadBytes     = 1 << 20
)

Engine defaults applied by EngineConfig.WithDefaults.

View Source 
const (
	HeaderDeadLetterSourceTopic = "x-queue-dlq-source-topic"
	HeaderDeadLetterGroup       = "x-queue-dlq-group"
	HeaderDeadLetterDeliveryID  = "x-queue-dlq-delivery-id"
	HeaderDeadLetterMessageID   = "x-queue-dlq-message-id"
	HeaderDeadLetterAttempts    = "x-queue-dlq-attempts"
	HeaderDeadLetterError       = "x-queue-dlq-error"
)

Headers added to the copy of a dead delivery published to a dead-letter topic.

View Source 
const (
	DefaultConcurrency   = 4
	DefaultBatchSize     = 32
	DefaultPollInterval  = time.Second
	DefaultLeaseDuration = 30 * time.Second
	DefaultMaxAttempts   = 5
	DefaultBackoffBase   = time.Second
	DefaultBackoffMax    = 5 * time.Minute
)

Subscription defaults applied by SubscribeOptions.WithDefaults.

View Source 
const MaxIDLength = 255

MaxIDLength bounds caller-supplied message IDs and keys, which are stored in indexed columns.

View Source 
const MaxNameLength = 200

MaxNameLength bounds topic and group names. Names become lock keys and index columns, so they are kept short.

Variables

View Source 
var (
	// ErrClosed is returned by operations on a closed queue or subscription.
	ErrClosed = errors.New("queue: closed")
	// ErrUnsupported is returned when a backend cannot provide an operation or
	// a dependency is of a kind the queue cannot use.
	ErrUnsupported = errors.New("queue: unsupported")
	// ErrNoLocker is returned when a backend needs a distributed locker to
	// consume safely and none was supplied.
	ErrNoLocker = errors.New("queue: distributed locker required")
	// ErrInvalidName is returned for a malformed topic or group name.
	ErrInvalidName = errors.New("queue: invalid name")
	// ErrPayloadTooLarge is returned when a message payload exceeds the
	// configured maximum.
	ErrPayloadTooLarge = errors.New("queue: payload too large")
	// ErrScheduleTooFar is returned for a message scheduled further ahead than
	// the queue retains messages.
	ErrScheduleTooFar = errors.New("queue: scheduled too far ahead")
)
View Source 
var ErrDeadLetterLoop = errors.New("queue: dead letter topic is the subscribed topic")

ErrDeadLetterLoop is returned by Subscribe when the dead-letter topic is the subscribed topic: every dead-lettered copy would be consumed by the same subscription, fail again, and be dead-lettered again, forever.

Functions

func DeliveryID 

func DeliveryID(messageID, group string) string

DeliveryID returns the deterministic ID of messageID's delivery to group. Every backend derives delivery rows from it, so re-publishing a message or backfilling a group can never create a second delivery for the same pair.

func IsPermanent 

func IsPermanent(err error) bool

IsPermanent reports whether err, or any error it wraps, was marked with Permanent.

func NewProcessLocker 

func NewProcessLocker() logstore.DistributedLocker

NewProcessLocker returns a locker for single-instance deployments and tests. Use configstore.DistributedLockManager when instances share the queue.

func Permanent 

func Permanent(err error) error

Permanent wraps err so the delivery is dead-lettered without further retries. Permanent(nil) returns nil.

func Register 

func Register(t QueueType, f Factory)

Register makes a queue implementation available to NewQueue under t. It is meant to be called from an init function or at startup, and panics on an empty type, a nil factory, or a type registered twice, like database/sql's Register.

func ValidateName 

func ValidateName(kind, name string) error

ValidateName checks a topic or group name: 1 to MaxNameLength characters from [A-Za-z0-9._-].

Types

type ClaimRequest 

type ClaimRequest struct {
	Topic       string
	Group       string
	RunnerID    string        // recorded as the holder, for debugging
	Max         int           // deliveries to lease at most
	Lease       time.Duration // how long the leases last without renewal
	MaxInFlight int           // cap on the group's unexpired leases across all instances; 0 means none
}

ClaimRequest describes one claim.

type Claimed 

type Claimed struct {
	DeliveryID string   // deterministic, see DeliveryID
	ClaimToken string   // fences every write made under this claim
	Group      string   // consumer group the delivery belongs to
	Attempt    int      // 1-based attempt this claim represents
	Message    *Message // the delivered message
}

Claimed is a delivery leased to one consumer.

type Config 

type Config struct {
	Enabled bool      `json:"enabled"`
	Type    QueueType `json:"type"`
	Config  any       `json:"config,omitempty"` // *LogStoreQueueConfig, *MemoryQueueConfig, or json.RawMessage for registered types
}

Config selects and configures a queue implementation.

func (*Config) UnmarshalJSON 

func (c *Config) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes the type-specific config block into its typed struct. Types this package does not know keep the raw JSON, which the registered factory decodes itself.

type ConsumeChecker 

type ConsumeChecker interface {
	CanConsume() error
}

ConsumeChecker is implemented by stores that can only consume under some condition, such as having a distributed locker. Subscribe calls it before registering anything, so a misconfigured consumer fails at startup.

type Delivery 

type Delivery struct {
	Message
	DeliveryID string `json:"delivery_id"` // stable across redeliveries; use it to deduplicate
	Group      string `json:"group"`
	Attempt    int    `json:"attempt"` // 1-based
}

Delivery is a message handed to one consumer group's handler.

type Dependencies 

type Dependencies struct {
	LogStore logstore.LogStore          // backing store for QueueTypeLogStore
	Locker   logstore.DistributedLocker // cross-instance lock; required to consume on ClickHouse
}

Dependencies are the shared resources a queue implementation may build on. A queue never closes them.

type EngineConfig 

type EngineConfig struct {
	RetentionHours      int    `json:"retention_hours,omitempty"`       // how long dead-lettered deliveries are kept
	AckedRetentionHours int    `json:"acked_retention_hours,omitempty"` // how long a message is kept once every group has finished with it
	JanitorInterval     string `json:"janitor_interval,omitempty"`      // Go duration between purge passes
	MaxPayloadBytes     int    `json:"max_payload_bytes,omitempty"`     // per-message payload limit
}

EngineConfig tunes the generic engine shared by the database-backed and in-memory queues.

func (EngineConfig) Janitor 

func (c EngineConfig) Janitor() (time.Duration, error)

Janitor parses JanitorInterval.

func (EngineConfig) PurgePolicy 

func (c EngineConfig) PurgePolicy() PurgePolicy

PurgePolicy returns the configured retentions for the janitor.

func (EngineConfig) Retention 

func (c EngineConfig) Retention() time.Duration

Retention returns the configured retention as a duration.

func (EngineConfig) WithDefaults 

func (c EngineConfig) WithDefaults() EngineConfig

WithDefaults returns a copy of c with zero-value fields filled.

type Factory 

type Factory func(ctx context.Context, cfg *Config, deps Dependencies, logger schemas.Logger) (Queue, error)

Factory builds a queue from its config. cfg.Config holds the typed config for built-in types and a json.RawMessage for registered ones.

type Handler 

type Handler func(ctx context.Context, d *Delivery) error

Handler processes one delivery. Returning nil acknowledges it, returning an error schedules a retry, and returning an error wrapped with Permanent dead-letters it immediately. The context is cancelled when the subscription closes or when the delivery's lease is lost to another consumer.

type LogStoreQueueConfig 

type LogStoreQueueConfig struct {
	EngineConfig
}

LogStoreQueueConfig configures the logstore-backed queue.

type MemoryQueueConfig 

type MemoryQueueConfig struct {
	EngineConfig
}

MemoryQueueConfig configures the in-memory queue.

type Message 

type Message struct {
	ID          string            `json:"id"`                   // set by Publish when empty
	Topic       string            `json:"topic"`                // required
	Key         string            `json:"key"`                  // optional; non-empty means strict FIFO per (topic, group, key)
	Payload     []byte            `json:"payload"`              // opaque to the queue
	Headers     map[string]string `json:"headers"`              // optional metadata
	PublishedAt time.Time         `json:"published_at"`         // set by Publish
	DeliverAt   time.Time         `json:"deliver_at,omitempty"` // optional; not handed out before this time
}

Message is a unit of data published to a topic.

type Publisher 

type Publisher interface {
	// Publish durably stores msgs and fans each one out to every consumer group
	// registered on its topic. Publishing a message ID that already exists is a
	// no-op, so a producer may safely retry a failed Publish.
	Publish(ctx context.Context, msgs ...*Message) error
}

Publisher publishes messages to topics.

type PurgePolicy 

type PurgePolicy struct {
	DeadRetention  time.Duration // dead-lettered deliveries, kept for inspection
	AckedRetention time.Duration // messages no delivery needs any more, kept for late Earliest groups
}

PurgePolicy says how long finished rows are kept.

type Queue 

type Queue interface {
	Publisher
	Subscriber
	// Stats reports delivery counts for one consumer group on a topic.
	Stats(ctx context.Context, topic, group string) (Stats, error)
	// Ping checks that the backing store is reachable.
	Ping(ctx context.Context) error
	// Close closes every subscription and then the queue's own resources. It
	// never closes stores the queue was handed as dependencies.
	Close(ctx context.Context) error
}

Queue is the contract every queue implementation satisfies.

func NewQueue 

func NewQueue(ctx context.Context, cfg *Config, deps Dependencies, logger schemas.Logger) (Queue, error)

NewQueue builds the queue cfg selects. It returns nil, nil when cfg is nil or disabled, so callers can treat "no queue configured" uniformly.

func NewStoreQueue 

func NewStoreQueue(store Store, cfg EngineConfig, logger schemas.Logger) (Queue, error)

NewStoreQueue builds a Queue on top of store. It takes ownership of store and closes it from Queue.Close. External Store implementations use this to get the full engine without reimplementing it.

type QueueType 

type QueueType string

QueueType names a queue implementation. Built-in types are listed below; any other type must be added with Register before NewQueue can build it. 

const (
	// QueueTypeLogStore stores messages in the logstore database (SQLite,
	// Postgres or ClickHouse).
	QueueTypeLogStore QueueType = "logstore"
	// QueueTypeMemory keeps messages in process. It is for tests and
	// single-instance deployments; messages do not survive a restart.
	QueueTypeMemory QueueType = "memory"
)

func Registered 

func Registered() []QueueType

Registered returns the registered queue types, sorted.

type StartFrom 

type StartFrom string

StartFrom selects which messages a consumer group receives when it is registered for the first time. It has no effect on a group that already exists. 

const (
	// StartFromLatest delivers only messages published after the group is
	// registered, matching Kafka's "latest" offset reset.
	StartFromLatest StartFrom = "latest"
	// StartFromEarliest also delivers every message still retained on the topic.
	StartFromEarliest StartFrom = "earliest"
)

type Stats 

type Stats struct {
	Pending int64 `json:"pending"` // waiting to be claimed: due, in retry backoff, or queued behind their key
	Leased  int64 `json:"leased"`  // currently held by a consumer
	Dead    int64 `json:"dead"`    // dead-lettered and not yet purged
}

Stats is a point-in-time count of one consumer group's deliveries.

type Store 

type Store interface {
	// EnsureGroup registers group on topic. When the group is new and startFrom
	// is StartFromEarliest, it also creates deliveries for every retained
	// message on the topic. Registering an existing group is a no-op.
	EnsureGroup(ctx context.Context, topic, group string, startFrom StartFrom) error
	// Append stores msgs, which already carry IDs and publish times, and
	// creates one pending delivery per group registered on each message's
	// topic. It is idempotent: messages and deliveries that already exist are
	// left untouched, including their state.
	Append(ctx context.Context, msgs []*Message) error
	// Claim leases up to req.Max due deliveries of req.Group on req.Topic. A
	// delivery is due when its message's DeliverAt and any retry backoff have
	// passed, or its previous lease expired, and, when it has a key, no earlier
	// delivery with the same key is still pending or leased. With
	// req.MaxInFlight set, the group's unexpired leases across every instance
	// never exceed it.
	Claim(ctx context.Context, req ClaimRequest) ([]*Claimed, error)
	// Ack completes deliveries in one round trip and returns the delivery IDs
	// that were still held; the rest had lost their lease and are untouched.
	Ack(ctx context.Context, cs []*Claimed) ([]string, error)
	// Retry counts a failed attempt and makes the delivery due again after
	// backoff. It reports false when c no longer holds it.
	Retry(ctx context.Context, c *Claimed, backoff time.Duration, lastErr string) (bool, error)
	// Kill counts a failed attempt and dead-letters the delivery. It reports
	// false when c no longer holds it.
	Kill(ctx context.Context, c *Claimed, lastErr string) (bool, error)
	// Extend renews the leases of cs for lease and returns the delivery IDs
	// still held. The caller treats every other delivery as lost.
	Extend(ctx context.Context, cs []*Claimed, lease time.Duration) ([]string, error)
	// Release gives deliveries back without counting an attempt, so another
	// consumer can claim them immediately.
	Release(ctx context.Context, cs []*Claimed) error
	// Purge deletes up to batch rows of each kind policy says have outlived
	// their use, returning how many it removed.
	Purge(ctx context.Context, policy PurgePolicy, batch int) (int64, error)
	// Stats counts group's deliveries on topic.
	Stats(ctx context.Context, topic, group string) (Stats, error)
	// Ping checks that the store is reachable.
	Ping(ctx context.Context) error
	// Close releases the store's own resources. It must not close a database
	// handle the store borrowed.
	Close(ctx context.Context) error
}

Store is the storage SPI behind the generic queue engine. A backend that implements it gets polling, concurrency, lease heartbeats, retries and dead-lettering from the engine for free.

Durations, never wall-clock instants, cross this interface: each store turns them into deadlines on its own clock. Shared backends use the database clock, which is what keeps leases correct when instances' clocks disagree.

Every operation after Claim is fenced on Claimed.ClaimToken. When the lease behind a token has expired and the delivery was claimed again, operations with the old token change nothing and report it as not held.

type SubscribeOption 

type SubscribeOption func(*SubscribeOptions)

SubscribeOption configures a subscription.

func WithBackoff 

func WithBackoff(base, max time.Duration) SubscribeOption

WithBackoff sets the exponential retry backoff's first delay and cap.

func WithBatchSize 

func WithBatchSize(n int) SubscribeOption

WithBatchSize sets the maximum number of deliveries claimed per poll.

func WithConcurrency 

func WithConcurrency(n int) SubscribeOption

WithConcurrency sets how many handlers run at once.

func WithDeadLetterTopic 

func WithDeadLetterTopic(topic string) SubscribeOption

WithDeadLetterTopic publishes a copy of every dead-lettered delivery to topic.

func WithLeaseDuration 

func WithLeaseDuration(d time.Duration) SubscribeOption

WithLeaseDuration sets how long a claim survives without a heartbeat.

func WithMaxAttempts 

func WithMaxAttempts(n int) SubscribeOption

WithMaxAttempts sets how many attempts a delivery gets before it is dead-lettered.

func WithMaxInFlight 

func WithMaxInFlight(n int) SubscribeOption

WithMaxInFlight caps how many of the group's deliveries run at once across every instance consuming it, on top of each instance's own Concurrency. Every subscriber of the group should use the same value. Zero means no cap.

func WithPollInterval 

func WithPollInterval(d time.Duration) SubscribeOption

WithPollInterval sets the wait between polls that found nothing.

func WithStartFrom 

func WithStartFrom(s StartFrom) SubscribeOption

WithStartFrom selects which messages a newly registered group receives.

type SubscribeOptions 

type SubscribeOptions struct {
	Concurrency     int           // handlers running at once; also the max in-flight deliveries
	BatchSize       int           // max deliveries claimed per poll
	PollInterval    time.Duration // wait between polls that found nothing
	LeaseDuration   time.Duration // how long a claim is held without a heartbeat
	MaxAttempts     int           // attempts before a delivery is dead-lettered
	BackoffBase     time.Duration // first retry delay
	BackoffMax      time.Duration // retry delay cap
	DeadLetterTopic string        // optional topic that receives a copy of dead deliveries
	MaxInFlight     int           // cap on the group's deliveries in flight across all instances; 0 means none
	StartFrom       StartFrom     // applies only when the group is first registered
}

SubscribeOptions tunes one subscription.

func NewSubscribeOptions 

func NewSubscribeOptions(opts ...SubscribeOption) (SubscribeOptions, error)

NewSubscribeOptions applies opts over the defaults and validates the result.

func (SubscribeOptions) Backoff 

func (o SubscribeOptions) Backoff(attempt int, rnd func() float64) time.Duration

Backoff returns the delay before retrying a delivery whose attempt-th try failed: BackoffBase doubled per attempt, capped at BackoffMax, with "equal jitter" so the result lies in [d/2, d]. rnd returns a value in [0, 1); nil uses math/rand.

func (SubscribeOptions) WithDefaults 

func (o SubscribeOptions) WithDefaults() SubscribeOptions

WithDefaults returns a copy of o with zero-value fields filled.

type Subscriber 

type Subscriber interface {
	// Subscribe registers group on topic (if it is not registered yet) and starts
	// delivering its messages to h until the returned Subscription is closed.
	Subscribe(ctx context.Context, topic, group string, h Handler, opts ...SubscribeOption) (Subscription, error)
}

Subscriber attaches handlers to topics as members of a consumer group.

type Subscription 

type Subscription interface {
	// Close stops claiming new deliveries, waits for in-flight handlers until
	// ctx is done, and releases any leases still held so other consumers can
	// take them over immediately.
	Close(ctx context.Context) error
}

Subscription is a running consumer.

type WakeSource 

type WakeSource interface {
	// Listen calls wake with a topic each time a message is published to it,
	// until ctx is done. It returns ErrUnsupported when the store has no way
	// to listen.
	Listen(ctx context.Context, wake func(topic string)) error
}

WakeSource is implemented by stores that can report publishes made by other instances. The engine runs Listen for the queue's lifetime and wakes local subscriptions of each reported topic. It is a latency optimisation only: polling still finds every message if notifications are lost.

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.