queue

package
v0.51.0 Latest Latest
Warning

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

 Go to latest Published: Aug 5, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrQueueNotStream is returned when an exact stream publish targets a
	// queue that exists but is not configured as a stream.
	ErrQueueNotStream = errors.New("queue is not a stream")
	// ErrQueueNotDurable is returned when an exact durable publish targets an
	// ephemeral queue.
	ErrQueueNotDurable = errors.New("queue is not durable")
	// ErrQueueNotReserved is returned when an exact internal publish targets a
	// queue that is no longer protected as a statically reserved queue.
	ErrQueueNotReserved = errors.New("queue is not reserved")
	// ErrQueueMessageTooLarge is returned before append when an exact stream
	// publish exceeds the target queue's configured maximum message size.
	ErrQueueMessageTooLarge = errors.New("message exceeds queue maximum size")
	// ErrQueueNotProtected is returned when the exact durable stream publish
	// path is used for a queue without a registered immutable contract.
	ErrQueueNotProtected = errors.New("queue has no protected contract")
	// ErrCaptureStillRunning is returned by Stop when a capture worker is still
	// inside the queue store after the drain timeout. The manager's resources —
	// the queue store especially — must not be released while that is true.
	ErrCaptureStillRunning = errors.New("capture workers did not finish; queue resources left open")
	// ErrProtectedQueueMutation is returned when a create, update, or delete
	// would violate a registered queue contract.
	ErrProtectedQueueMutation = errors.New("protected queue mutation rejected")
	// ErrProtectedQueueContractDrift is returned when a protected queue's
	// persisted configuration no longer matches its registered contract.
	ErrProtectedQueueContractDrift = errors.New("protected queue contract drift")
	// ErrDurableSyncUnsupported is returned before append when the configured
	// queue store cannot establish a per-queue durability barrier.
	ErrDurableSyncUnsupported = errors.New("queue store does not support durable sync")
	// ErrDurableReplicatedStreamUnsupported prevents a false durability ACK in
	// clustered mode until the same barrier is carried through leader forwarding.
	ErrDurableReplicatedStreamUnsupported = errors.New("durable exact stream publish does not support replication")
)

Functions

func DefaultConsumerGroupID added in v0.50.0 

func DefaultConsumerGroupID(clientID string) string

DefaultConsumerGroupID returns the queue-mode consumer group used when a subscriber does not provide one explicitly.

func ValidateProtectedQueueContract added in v0.50.0 

func ValidateProtectedQueueContract(expected, persisted types.QueueConfig) error

ValidateProtectedQueueContract compares the persisted fields that define an exact internal publisher's safety and replay guarantees. MaxDepth is deliberately excluded because stream depth is not currently enforced.

Types

type ClientConnectionChecker 

type ClientConnectionChecker interface {
	IsClientConnected(clientID string) bool
}

ClientConnectionChecker is kept for deliverers where only live connections are valid queue delivery targets.

type ClientDeliveryTargetChecker 

type ClientDeliveryTargetChecker interface {
	HasDeliveryTarget(clientID string) bool
}

ClientDeliveryTargetChecker is optionally implemented by a Deliverer that can tell whether queue delivery has a valid target before queue cursors move. For MQTT, a disconnected persistent session is still a valid delivery target because the broker can enqueue QoS>0 messages for later delivery.

type Config 

type Config struct {
	// Consumer configuration
	VisibilityTimeout  time.Duration
	MaxDeliveryCount   int
	ClaimBatchSize     int
	AutoCommitInterval time.Duration

	// Delivery configuration
	DeliveryInterval  time.Duration
	DeliveryBatchSize int
	HeartbeatInterval time.Duration
	ConsumerTimeout   time.Duration

	// DLQ configuration
	DLQTopicPrefix string

	// Work stealing configuration
	StealInterval time.Duration
	StealEnabled  bool

	// PEL configuration
	MaxPELSize int

	// Retention configuration
	RetentionCheckInterval time.Duration

	// Capture dispatcher configuration. Topic capture runs off the publish
	// path so a stalled queue store cannot delay subscribers; these bound how
	// much unwritten capture is held and how long shutdown waits for it.
	CaptureWorkers      int
	CaptureQueueDepth   int
	CaptureDrainTimeout time.Duration

	// Replication/distribution configuration
	WritePolicy      WritePolicy
	DistributionMode DistributionMode

	// Queue configurations from main config
	QueueConfigs []types.QueueConfig

	// ProtectedQueueContracts are immutable runtime contracts for queues used
	// by exact internal publishers. Only explicitly listed queues are protected;
	// Reserved alone does not make a queue immutable.
	ProtectedQueueContracts []types.QueueConfig

	// OnConsumerRemoved is called when stale consumers are removed during
	// heartbeat cleanup. The callback receives the queue name, group ID,
	// and the list of removed consumer IDs (prefixed client IDs).
	// Must be non-blocking.
	OnConsumerRemoved func(queueName, groupID string, consumerIDs []string)
}

Config holds configuration for the queue-based queue manager.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns default configuration.

type Deliverer 

type Deliverer interface {
	Deliver(ctx context.Context, clientID string, msg *storage.Message) error
}

Deliverer is the output boundary for queue message delivery. Protocol brokers implement this interface so the delivery engine can push messages to connected clients without knowing protocol details.

type DeliveryEngine 

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

DeliveryEngine claims messages from queues and routes them to local or remote consumers. It owns the scheduling loop and delivery state; the Manager delegates all delivery work here.

func NewDeliveryEngine 

func NewDeliveryEngine(
	queueStore storage.QueueStore,
	groupStore storage.ConsumerGroupStore,
	consumerMgr *consumer.Manager,
	local Deliverer,
	remote RemoteRouter,
	localNodeID string,
	distributionMode DistributionMode,
	batchSize int,
	logger *slog.Logger,
) *DeliveryEngine

NewDeliveryEngine creates a delivery engine. remote may be nil for single-node deployments.

func (*DeliveryEngine) DeliverAll 

func (e *DeliveryEngine) DeliverAll(ctx context.Context)

DeliverAll delivers messages for every queue (full sweep). Intended for tests and benchmarks that need synchronous delivery without the loop.

func (*DeliveryEngine) DeliverQueue 

func (e *DeliveryEngine) DeliverQueue(ctx context.Context, queueName string) bool

DeliverQueue delivers messages for a single queue by name. Returns true if any messages were delivered.

func (*DeliveryEngine) Schedule 

func (e *DeliveryEngine) Schedule(queueName string)

Schedule enqueues a queue name for delivery. Duplicate schedules for the same queue are coalesced until the queue is delivered.

func (*DeliveryEngine) ScheduleAll 

func (e *DeliveryEngine) ScheduleAll(ctx context.Context)

ScheduleAll lists all queues and schedules each for delivery.

func (*DeliveryEngine) Start 

func (e *DeliveryEngine) Start(ctx context.Context)

Start launches the delivery loop goroutine.

func (*DeliveryEngine) Stop 

func (e *DeliveryEngine) Stop()

Stop signals the delivery loop to exit and waits for it to finish.

func (*DeliveryEngine) Unschedule 

func (e *DeliveryEngine) Unschedule(queueName string)

Unschedule removes a queue from the dedup set. Called when a queue is deleted.

type DeliveryMessage 

type DeliveryMessage struct {
	ID          string
	Payload     []byte
	Topic       string
	Properties  map[string]string
	GroupID     string
	Offset      uint64
	DeliveredAt time.Time
	AckTopic    string
	NackTopic   string
	RejectTopic string
}

DeliveryMessage is the internal message format for queue delivery tracking.

type DeliveryTargetFunc 

type DeliveryTargetFunc func(ctx context.Context, clientID string, msg *storage.Message) error

DeliveryTargetFunc adapts a plain function to the DeliveryTarget interface.

func (DeliveryTargetFunc) Deliver 

func (f DeliveryTargetFunc) Deliver(ctx context.Context, clientID string, msg *storage.Message) error

type DistributionMode 

type DistributionMode string

DistributionMode controls how queue messages reach consumers across nodes. 

const (
	DistributionForward   DistributionMode = "forward"   // Forward publishes to nodes with consumers
	DistributionReplicate DistributionMode = "replicate" // Rely on Raft to replicate queue logs
)

type GroupOpForwarder 

type GroupOpForwarder interface {
	ForwardGroupOp(ctx context.Context, nodeID, queueName string, opData []byte) error
}

GroupOpForwarder forwards consumer group operations to a remote node.

type Manager 

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

Manager is the queue-based queue manager. It uses append-only logs with cursor-based consumer groups, NATS JetQueue-style.

func NewManager 

func NewManager(queueStore storage.QueueStore, groupStore storage.ConsumerGroupStore, dt Deliverer, config Config, logger *slog.Logger, cl cluster.Cluster) *Manager

NewManager creates a new queue-based queue manager. The cluster parameter is optional (nil for single-node mode).

func (*Manager) Ack 

func (m *Manager) Ack(ctx context.Context, queueName, messageID, groupID string) error

Ack acknowledges a message.

func (*Manager) CommitOffset 

func (m *Manager) CommitOffset(ctx context.Context, queueName, groupID string, offset uint64) error

CommitOffset explicitly commits an offset for a stream consumer group. Use when AutoCommit is disabled for manual commit control.

func (*Manager) CreateQueue 

func (m *Manager) CreateQueue(ctx context.Context, config types.QueueConfig) error

CreateQueue creates a new queue.

func (*Manager) DeleteQueue 

func (m *Manager) DeleteQueue(ctx context.Context, queueName string) error

DeleteQueue deletes a queue.

func (*Manager) DeliverQueueMessage 

func (m *Manager) DeliverQueueMessage(ctx context.Context, clientID string, msg *cluster.QueueMessage) error

DeliverQueueMessage implements cluster.QueueHandler.DeliverQueueMessage.

func (*Manager) Enqueue 

func (m *Manager) Enqueue(ctx context.Context, topic string, payload []byte, properties map[string]string) error

Enqueue is an alias for Publish for backward compatibility.

func (*Manager) EnqueueLocal 

func (m *Manager) EnqueueLocal(ctx context.Context, topic string, payload []byte, properties map[string]string) (string, error)

EnqueueLocal implements cluster.QueueHandler.EnqueueLocal.

func (*Manager) GetLag 

func (m *Manager) GetLag(ctx context.Context, queueName, groupID string) (uint64, error)

GetLag returns the lag for a consumer group.

func (*Manager) GetMetrics 

func (m *Manager) GetMetrics() consumer.Metrics

GetMetrics returns the current metrics snapshot.

func (*Manager) GetOrCreateQueue 

func (m *Manager) GetOrCreateQueue(ctx context.Context, queueName string, topics ...string) (*types.QueueConfig, error)

GetOrCreateQueue gets or creates a queue with default configuration.

func (*Manager) GetQueue 

func (m *Manager) GetQueue(ctx context.Context, queueName string) (*types.QueueConfig, error)

GetQueue returns the configuration for a queue.

func (*Manager) GetRaftManager 

func (m *Manager) GetRaftManager() *raft.Manager

GetRaftManager returns the Raft replication manager.

func (*Manager) GroupStore 

func (m *Manager) GroupStore() storage.ConsumerGroupStore

GroupStore returns the consumer group store used by the manager.

func (*Manager) HandleForwardedGroupOp 

func (m *Manager) HandleForwardedGroupOp(ctx context.Context, queueName string, opData []byte) error

HandleForwardedGroupOp implements cluster.QueueHandler.HandleForwardedGroupOp. It decodes a raft.Operation and applies it through the local coordinator.

func (*Manager) HandleQueuePublish 

func (m *Manager) HandleQueuePublish(ctx context.Context, publish types.PublishRequest, mode types.PublishMode) error

HandleQueuePublish implements cluster.QueueHandler.HandleQueuePublish.

func (*Manager) ListQueues 

func (m *Manager) ListQueues(ctx context.Context) ([]types.QueueConfig, error)

ListQueues returns all queue configurations.

func (*Manager) Nack 

func (m *Manager) Nack(ctx context.Context, queueName, messageID, groupID string) error

Nack negatively acknowledges a message.

func (*Manager) NarrowProtectedQueueContracts added in v0.50.0 

func (m *Manager) NarrowProtectedQueueContracts(contracts []types.QueueConfig) error

NarrowProtectedQueueContracts replaces the registry with an exact subset of the already-installed contracts without reading queue storage. It is used to remove stale contracts after another subsystem has atomically committed its new authorization snapshot; unlike ReplaceProtectedQueueContracts, this finalization step cannot fail because of storage I/O.

func (*Manager) ProtectedQueueContracts added in v0.50.0 

func (m *Manager) ProtectedQueueContracts() []types.QueueConfig

ProtectedQueueContracts returns a snapshot of the currently registered immutable queue contracts.

func (*Manager) Publish 

func (m *Manager) Publish(ctx context.Context, publish types.PublishRequest) error

Publish adds a message to all queues whose topic patterns match the topic. This is the NATS JetQueue-style "multi-queue" routing. The delivery engine routes appended records to remote consumers when needed.

func (*Manager) PublishToDurableStream added in v0.50.0 

func (m *Manager) PublishToDurableStream(ctx context.Context, queueName string, publish types.PublishRequest) error

PublishToDurableStream appends to exactly queueName and establishes a per-queue durability barrier before returning success. The target must be a reserved, durable, non-replicated stream. This method never performs topic fanout and never auto-creates a queue.

The contract snapshot is copied out of the registry before any storage work starts. Holding protectedQueuesMu across the append would let one fsync block every contract reload, and a reload waiting for the write lock would in turn stall every subsequent publish.

func (*Manager) PublishToMatchingQueues added in v0.51.0 

func (m *Manager) PublishToMatchingQueues(ctx context.Context, publish types.PublishRequest) error

PublishToMatchingQueues captures an ordinary pub/sub publish in existing queues whose configured topic patterns match it. Unlike Publish, it never auto-creates a queue when no pattern matches.

It resolves the matching queues on the caller's goroutine — that is an in-memory index lookup — and then hands the storage work to the capture dispatcher. Enqueueing never blocks, so a queue whose store stalls can no longer delay the subscribers of a matching topic or the publisher's acknowledgement.

The returned error therefore reports only what is known before the append is attempted: that the matching queues could not be resolved. An append that fails or is dropped afterwards is reported through queues.capture_failures and queues.capture_dropped, which is the only signal capture has.

func (*Manager) QueueStore 

func (m *Manager) QueueStore() storage.QueueStore

QueueStore returns the queue store used by the manager.

func (*Manager) Reject 

func (m *Manager) Reject(ctx context.Context, queueName, messageID, groupID, reason string) error

Reject rejects a message and moves it to DLQ.

func (*Manager) ReplaceProtectedQueueContracts added in v0.50.0 

func (m *Manager) ReplaceProtectedQueueContracts(ctx context.Context, contracts []types.QueueConfig) error

ReplaceProtectedQueueContracts validates and atomically replaces the immutable queue-contract registry. Queue mutations and exact publishes are blocked for the duration, so no operation can enter between persisted-state validation and the registry swap.

func (*Manager) SetRaftCoordinator 

func (m *Manager) SetRaftCoordinator(rc raft.QueueCoordinator)

SetRaftCoordinator sets queue-aware Raft coordinator.

func (*Manager) SetRaftManager 

func (m *Manager) SetRaftManager(rm *raft.Manager)

SetRaftManager sets the Raft replication manager.

func (*Manager) ShutdownComplete added in v0.51.0 

func (m *Manager) ShutdownComplete() bool

ShutdownComplete reports whether Stop finished with every capture worker out of the queue store.

It gates releasing anything the manager shares with those workers, the queue store above all. An append already in flight cannot be cancelled — the store takes no context — so Stop bounds its wait rather than hanging, and a worker may outlive it. Closing the store then would be a use-after-close on a segment that worker still holds. Callers that own such a resource must consult this and leak the handle instead; the process is exiting either way.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context) error

Start starts background workers.

func (*Manager) Stop 

func (m *Manager) Stop() error

Stop stops the manager and all workers. Stop shuts the manager down and reports whether that completed cleanly.

Capture drains first. Its workers append through the queue store, schedule delivery, and on replicated queues call the Raft coordinator, so those have to outlive the drain rather than the other way round.

A returned ErrCaptureStillRunning means a capture worker is still inside the queue store and could not be interrupted. Everything it touches — the store above all — must then be left alone: closing it underneath an in-flight append is a use-after-close, and leaking the handle into process exit is the cheaper outcome.

func (*Manager) Subscribe 

func (m *Manager) Subscribe(ctx context.Context, queueName, pattern string, clientID, groupID, proxyNodeID string) error

Subscribe adds a consumer to a stream with optional pattern matching.

func (*Manager) SubscribeExisting added in v0.50.0 

func (m *Manager) SubscribeExisting(ctx context.Context, queueName, pattern string, clientID, groupID, proxyNodeID string) error

SubscribeExisting adds a consumer to an existing queue without creating it.

func (*Manager) SubscribeExistingWithCursor added in v0.50.0 

func (m *Manager) SubscribeExistingWithCursor(ctx context.Context, queueName, pattern string, clientID, groupID, proxyNodeID string, cursor *types.CursorOption) error

SubscribeExistingWithCursor adds a consumer to an existing queue without creating the queue or changing its configured type.

func (*Manager) SubscribeWithCursor 

func (m *Manager) SubscribeWithCursor(ctx context.Context, queueName, pattern string, clientID, groupID, proxyNodeID string, cursor *types.CursorOption) error

SubscribeWithCursor adds a consumer with explicit cursor positioning.

func (*Manager) Unsubscribe 

func (m *Manager) Unsubscribe(ctx context.Context, queueName, pattern string, clientID, groupID string) error

Unsubscribe removes a consumer from a stream.

func (*Manager) UpdateConsumerHeartbeat 

func (m *Manager) UpdateConsumerHeartbeat(ctx context.Context, queueName, groupID, consumerID string) error

UpdateConsumerHeartbeat updates heartbeat for a specific consumer membership.

func (*Manager) UpdateHeartbeat 

func (m *Manager) UpdateHeartbeat(ctx context.Context, clientID string) error

UpdateHeartbeat updates the heartbeat for a consumer.

func (*Manager) UpdateQueue 

func (m *Manager) UpdateQueue(ctx context.Context, config types.QueueConfig) error

UpdateQueue updates an existing queue.

func (*Manager) ValidateProtectedQueueContracts added in v0.50.0 

func (m *Manager) ValidateProtectedQueueContracts(ctx context.Context) error

ValidateProtectedQueueContracts verifies every registered contract against the persisted queue configuration.

type MetricsProvider 

type MetricsProvider interface {
	GetMetrics() consumer.Metrics
	GetLag(ctx context.Context, queueName, groupID string) (uint64, error)
}

MetricsProvider exposes read-only queue delivery metrics.

type RemoteBatchRouter 

type RemoteBatchRouter interface {
	RouteQueueBatch(ctx context.Context, nodeID string, deliveries []cluster.QueueDelivery) error
}

type RemoteRouter 

type RemoteRouter interface {
	ListQueueConsumers(ctx context.Context, queueName string) ([]*cluster.QueueConsumerInfo, error)
	RouteQueueMessage(ctx context.Context, nodeID, clientID, queueName string, msg *cluster.QueueMessage) error
	UnregisterQueueConsumer(ctx context.Context, queueName, groupID, consumerID string) error
}

RemoteRouter is the subset of cluster.Cluster needed by the delivery engine for cross-node message routing.

type StreamCommitter 

type StreamCommitter interface {
	CommitOffset(ctx context.Context, queueName, groupID string, offset uint64) error
}

StreamCommitter exposes explicit commit control for stream consumer groups.

type WritePolicy 

type WritePolicy string

WritePolicy controls how non-leader nodes handle queue writes when Raft is enabled. 

const (
	WritePolicyLocal   WritePolicy = "local"   // Append locally (no Raft redirect)
	WritePolicyReject  WritePolicy = "reject"  // Reject writes on non-leaders
	WritePolicyForward WritePolicy = "forward" // Forward writes to the Raft leader
)

Source Files

View all Source files

Directories

Path Synopsis
Package consumer provides consumer group management with routing key filtering and work stealing support for the log-based queue model.
 Package consumer provides consumer group management with routing key filtering and work stealing support for the log-based queue model.
memory/log
Package log provides an in-memory implementation of the queue-based storage.
 Package log provides an in-memory implementation of the queue-based storage.

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.