queue

package

Go to main page

Versions in this module

v1

v1.5.1

Feb 16, 2026

Changes in this version

type PostgresStore

+ func (s *PostgresStore) CaptureBacklogTrendSample(at time.Time) error

+ func (s *PostgresStore) ListBacklogTrend(req BacklogTrendListRequest) (BacklogTrendListResponse, error)

v1.5.0

Feb 15, 2026

Changes in this version

+ type PostgresOption func(*PostgresStore)

+ func WithPostgresDLQRetention(maxAge time.Duration, maxDepth int) PostgresOption

+ func WithPostgresDeliveredRetention(maxAge time.Duration) PostgresOption

+ func WithPostgresNowFunc(now func() time.Time) PostgresOption

+ func WithPostgresPollInterval(d time.Duration) PostgresOption

+ func WithPostgresQueueLimits(maxDepth int, dropPolicy string) PostgresOption

+ func WithPostgresRetention(maxAge, pruneInterval time.Duration) PostgresOption

+ type PostgresStore struct

+ func NewPostgresStore(dsn string, opts ...PostgresOption) (*PostgresStore, error)

+ func (s *PostgresStore) Ack(leaseID string) error

+ func (s *PostgresStore) AckBatch(leaseIDs []string) (LeaseBatchResult, error)

+ func (s *PostgresStore) CancelMessages(req MessageCancelRequest) (MessageCancelResponse, error)

+ func (s *PostgresStore) CancelMessagesByFilter(req MessageManageFilterRequest) (MessageCancelResponse, error)

+ func (s *PostgresStore) Close() error

+ func (s *PostgresStore) DeleteDead(req DeadDeleteRequest) (DeadDeleteResponse, error)

+ func (s *PostgresStore) Dequeue(req DequeueRequest) (DequeueResponse, error)

+ func (s *PostgresStore) Enqueue(env Envelope) error

+ func (s *PostgresStore) Extend(leaseID string, extendBy time.Duration) error

+ func (s *PostgresStore) ListAttempts(req AttemptListRequest) (AttemptListResponse, error)

+ func (s *PostgresStore) ListDead(req DeadListRequest) (DeadListResponse, error)

+ func (s *PostgresStore) ListMessages(req MessageListRequest) (MessageListResponse, error)

+ func (s *PostgresStore) LookupMessages(req MessageLookupRequest) (MessageLookupResponse, error)

+ func (s *PostgresStore) MarkDead(leaseID string, reason string) error

+ func (s *PostgresStore) MarkDeadBatch(leaseIDs []string, reason string) (LeaseBatchResult, error)

+ func (s *PostgresStore) Nack(leaseID string, delay time.Duration) error

+ func (s *PostgresStore) NackBatch(leaseIDs []string, delay time.Duration) (LeaseBatchResult, error)

+ func (s *PostgresStore) RecordAttempt(attempt DeliveryAttempt) error

+ func (s *PostgresStore) RequeueDead(req DeadRequeueRequest) (DeadRequeueResponse, error)

+ func (s *PostgresStore) RequeueMessages(req MessageRequeueRequest) (MessageRequeueResponse, error)

+ func (s *PostgresStore) RequeueMessagesByFilter(req MessageManageFilterRequest) (MessageRequeueResponse, error)

+ func (s *PostgresStore) ResumeMessages(req MessageResumeRequest) (MessageResumeResponse, error)

+ func (s *PostgresStore) ResumeMessagesByFilter(req MessageManageFilterRequest) (MessageResumeResponse, error)

+ func (s *PostgresStore) RuntimeMetrics() StoreRuntimeMetrics

+ func (s *PostgresStore) Stats() (Stats, error)

+ type StoreCommonRuntimeMetrics struct

+ ErrorsTotal []StoreOperationErrorRuntimeMetric

+ OperationDurationSeconds []StoreOperationDurationRuntimeMetric

+ OperationTotal []StoreOperationCounterRuntimeMetric

+ type StoreOperationCounterRuntimeMetric struct

+ Operation string

+ Total int64

+ type StoreOperationDurationRuntimeMetric struct

+ DurationSeconds HistogramSnapshot

+ Operation string

+ type StoreOperationErrorRuntimeMetric struct

+ Kind string

+ Operation string

+ Total int64

type StoreRuntimeMetrics

+ Common StoreCommonRuntimeMetrics

v1.4.0

Feb 14, 2026

Changes in this version

+ type LeaseBatchConflict struct

+ Expired bool

+ LeaseID string

+ type LeaseBatchResult struct

+ Conflicts []LeaseBatchConflict

+ Succeeded int

+ type LeaseBatchStore interface

+ AckBatch func(leaseIDs []string) (LeaseBatchResult, error)

+ MarkDeadBatch func(leaseIDs []string, reason string) (LeaseBatchResult, error)

+ NackBatch func(leaseIDs []string, delay time.Duration) (LeaseBatchResult, error)

type MemoryStore

+ func (s *MemoryStore) AckBatch(leaseIDs []string) (LeaseBatchResult, error)

+ func (s *MemoryStore) MarkDeadBatch(leaseIDs []string, reason string) (LeaseBatchResult, error)

+ func (s *MemoryStore) NackBatch(leaseIDs []string, delay time.Duration) (LeaseBatchResult, error)

type SQLiteStore

+ func (s *SQLiteStore) AckBatch(leaseIDs []string) (LeaseBatchResult, error)

+ func (s *SQLiteStore) MarkDeadBatch(leaseIDs []string, reason string) (LeaseBatchResult, error)

+ func (s *SQLiteStore) NackBatch(leaseIDs []string, delay time.Duration) (LeaseBatchResult, error)

v1.2.0

Feb 13, 2026

Changes in this version

+ var ErrMemoryPressure = errors.New("memory pressure")

type MemoryOption

+ func WithMemoryPressureLimits(retainedItems int, retainedBytes int64) MemoryOption

+ type MemoryPressureRuntimeMetrics struct

+ Active bool

+ Reason string

+ RejectTotal int64

+ RetainedBytes int64

+ RetainedBytesLimit int64

+ RetainedItemLimit int64

+ RetainedItems int64

+ type MemoryRuntimeMetrics struct

+ EvictionsTotalByReason map[string]int64

+ ItemsByState map[State]int64

+ Pressure MemoryPressureRuntimeMetrics

+ RetainedBytesByState map[State]int64

+ RetainedBytesTotal int64

type MemoryStore

+ func (s *MemoryStore) RuntimeMetrics() StoreRuntimeMetrics

type StoreRuntimeMetrics

+ Memory *MemoryRuntimeMetrics

v1.1.0

Feb 12, 2026

Changes in this version

+ type HistogramBucket struct

+ Count int64

+ Le float64

+ type HistogramSnapshot struct

+ Buckets []HistogramBucket

+ Count int64

+ Sum float64

+ type RuntimeMetricsProvider interface

+ RuntimeMetrics func() StoreRuntimeMetrics

type SQLiteOption

+ func WithSQLiteCheckpointInterval(d time.Duration) SQLiteOption

+ type SQLiteRuntimeMetrics struct

+ BusyTotal int64

+ CheckpointDurationSeconds HistogramSnapshot

+ CheckpointErrorTotal int64

+ CheckpointTotal int64

+ DequeueDurationSeconds HistogramSnapshot

+ RetryTotal int64

+ TxCommitTotal int64

+ TxRollbackTotal int64

+ WriteDurationSeconds HistogramSnapshot

type SQLiteStore

+ func (s *SQLiteStore) RuntimeMetrics() StoreRuntimeMetrics

+ type StoreRuntimeMetrics struct

+ Backend string

+ SQLite *SQLiteRuntimeMetrics

v1.0.3

Feb 10, 2026

v1.0.2

Feb 10, 2026

v1.0.1

Feb 10, 2026

Changes in this version

+ const MessageOrderAsc

+ const MessageOrderDesc

+ var ErrEnvelopeExists = errors.New("envelope already exists")

+ var ErrLeaseExpired = errors.New("lease expired")

+ var ErrLeaseNotFound = errors.New("lease not found")

+ var ErrQueueFull = errors.New("queue full")

+ func IDMutationTouchesManagedRoute(store Store, ids []string, allowedStates map[State]struct{}, ...) (bool, error)

+ type AttemptListRequest struct

+ Before time.Time

+ EventID string

+ Limit int

+ Outcome AttemptOutcome

+ Route string

+ Target string

+ type AttemptListResponse struct

+ Items []DeliveryAttempt

+ type AttemptOutcome string

+ const AttemptOutcomeAcked

+ const AttemptOutcomeDead

+ const AttemptOutcomeRetry

+ type BacklogBucket struct

+ EarliestQueuedNextRun time.Time

+ OldestQueuedAge time.Duration

+ OldestQueuedReceivedAt time.Time

+ Queued int

+ ReadyLag time.Duration

+ Route string

+ Target string

+ type BacklogTrendListRequest struct

+ Limit int

+ Route string

+ Since time.Time

+ Target string

+ Until time.Time

+ type BacklogTrendListResponse struct

+ Items []BacklogTrendSample

+ Truncated bool

+ type BacklogTrendOperatorAction struct

+ AdminEndpoints []string

+ AlertRoutingKey string

+ Alerts []string

+ ID string

+ MCPTools []string

+ Playbook string

+ Severity string

+ Summary string

+ func (a BacklogTrendOperatorAction) Map() map[string]any

+ type BacklogTrendSample struct

+ CapturedAt time.Time

+ Dead int

+ Leased int

+ Queued int

+ type BacklogTrendSignalConfig struct

+ DeadShareHighMinTotal int

+ DeadShareHighPercent int

+ ExpectedCaptureInterval time.Duration

+ QueuedPressureLeasedMultiplier int

+ QueuedPressureMinTotal int

+ QueuedPressurePercent int

+ RecentSurgeMinDelta int

+ RecentSurgeMinTotal int

+ RecentSurgePercent int

+ StaleGraceFactor int

+ SustainedGrowthConsecutive int

+ SustainedGrowthMinDelta int

+ SustainedGrowthMinSamples int

+ Window time.Duration

+ func DefaultBacklogTrendSignalConfig() BacklogTrendSignalConfig

+ type BacklogTrendSignalOptions struct

+ Config BacklogTrendSignalConfig

+ ExpectedCaptureInterval time.Duration

+ Now time.Time

+ Window time.Duration

+ type BacklogTrendSignals struct

+ ActiveAlerts []string

+ BaselineTotal int

+ ConsecutiveIncreases int

+ DeadShareHigh bool

+ DeadSharePercent int

+ DeltaPercent int

+ DeltaTotal int

+ ExpectedCaptureInterval time.Duration

+ FreshnessSeconds int

+ GrowthRatePerMinute float64

+ LatestCapturedAt time.Time

+ LatestDead int

+ LatestLeased int

+ LatestQueued int

+ LatestTotal int

+ QueuedPressure bool

+ QueuedSharePercent int

+ RecentSurge bool

+ SampleCount int

+ SamplingStale bool

+ Since time.Time

+ Status string

+ SustainedGrowth bool

+ Truncated bool

+ Until time.Time

+ Window time.Duration

+ func AnalyzeBacklogTrendSignals(samples []BacklogTrendSample, truncated bool, opts BacklogTrendSignalOptions) BacklogTrendSignals

+ func (s BacklogTrendSignals) Map() map[string]any

+ func (s BacklogTrendSignals) OperatorActions() []BacklogTrendOperatorAction

+ type BacklogTrendStore interface

+ CaptureBacklogTrendSample func(at time.Time) error

+ ListBacklogTrend func(req BacklogTrendListRequest) (BacklogTrendListResponse, error)

+ type BatchEnqueuer interface

+ EnqueueBatch func(items []Envelope) (int, error)

+ type DeadDeleteRequest struct

+ IDs []string

+ type DeadDeleteResponse struct

+ Deleted int

+ type DeadListRequest struct

+ Before time.Time

+ IncludeHeaders bool

+ IncludePayload bool

+ IncludeTrace bool

+ Limit int

+ Route string

+ type DeadListResponse struct

+ Items []Envelope

+ type DeadRequeueRequest struct

+ IDs []string

+ type DeadRequeueResponse struct

+ Requeued int

+ type DeliveryAttempt struct

+ Attempt int

+ CreatedAt time.Time

+ DeadReason string

+ Error string

+ EventID string

+ ID string

+ Outcome AttemptOutcome

+ Route string

+ StatusCode int

+ Target string

+ type DequeueRequest struct

+ AllowLeased bool

+ Batch int

+ LeaseTTL time.Duration

+ MaxWait time.Duration

+ Now time.Time

+ Route string

+ Target string

+ type DequeueResponse struct

+ Items []Envelope

+ type Envelope struct

+ Attempt int

+ DeadReason string

+ Headers map[string]string

+ ID string

+ LeaseID string

+ LeaseUntil time.Time

+ NextRunAt time.Time

+ Payload []byte

+ ReceivedAt time.Time

+ Route string

+ SchemaVersion int

+ State State

+ Target string

+ Trace map[string]string

+ type MemoryOption func(*MemoryStore)

+ func WithDLQRetention(maxAge time.Duration, maxDepth int) MemoryOption

+ func WithDeliveredRetention(maxAge time.Duration) MemoryOption

+ func WithNowFunc(now func() time.Time) MemoryOption

+ func WithQueueLimits(maxDepth int, dropPolicy string) MemoryOption

+ func WithQueueRetention(maxAge, pruneInterval time.Duration) MemoryOption

+ type MemoryStore struct

+ func NewMemoryStore(opts ...MemoryOption) *MemoryStore

+ func (s *MemoryStore) Ack(leaseID string) error

+ func (s *MemoryStore) CancelMessages(req MessageCancelRequest) (MessageCancelResponse, error)

+ func (s *MemoryStore) CancelMessagesByFilter(req MessageManageFilterRequest) (MessageCancelResponse, error)

+ func (s *MemoryStore) CaptureBacklogTrendSample(at time.Time) error

+ func (s *MemoryStore) DeleteDead(req DeadDeleteRequest) (DeadDeleteResponse, error)

+ func (s *MemoryStore) Dequeue(req DequeueRequest) (DequeueResponse, error)

+ func (s *MemoryStore) Enqueue(env Envelope) error

+ func (s *MemoryStore) EnqueueBatch(items []Envelope) (int, error)

+ func (s *MemoryStore) Extend(leaseID string, extendBy time.Duration) error

+ func (s *MemoryStore) ListAttempts(req AttemptListRequest) (AttemptListResponse, error)

+ func (s *MemoryStore) ListBacklogTrend(req BacklogTrendListRequest) (BacklogTrendListResponse, error)

+ func (s *MemoryStore) ListDead(req DeadListRequest) (DeadListResponse, error)

+ func (s *MemoryStore) ListMessages(req MessageListRequest) (MessageListResponse, error)

+ func (s *MemoryStore) LookupMessages(req MessageLookupRequest) (MessageLookupResponse, error)

+ func (s *MemoryStore) MarkDead(leaseID string, reason string) error

+ func (s *MemoryStore) Nack(leaseID string, delay time.Duration) error

+ func (s *MemoryStore) RecordAttempt(attempt DeliveryAttempt) error

+ func (s *MemoryStore) RequeueDead(req DeadRequeueRequest) (DeadRequeueResponse, error)

+ func (s *MemoryStore) RequeueMessages(req MessageRequeueRequest) (MessageRequeueResponse, error)

+ func (s *MemoryStore) RequeueMessagesByFilter(req MessageManageFilterRequest) (MessageRequeueResponse, error)

+ func (s *MemoryStore) ResumeMessages(req MessageResumeRequest) (MessageResumeResponse, error)

+ func (s *MemoryStore) ResumeMessagesByFilter(req MessageManageFilterRequest) (MessageResumeResponse, error)

+ func (s *MemoryStore) Stats() (Stats, error)

+ type MessageCancelRequest struct

+ IDs []string

+ type MessageCancelResponse struct

+ Canceled int

+ Matched int

+ PreviewOnly bool

+ type MessageListRequest struct

+ Before time.Time

+ IncludeHeaders bool

+ IncludePayload bool

+ IncludeTrace bool

+ Limit int

+ Order string

+ Route string

+ State State

+ Target string

+ type MessageListResponse struct

+ Items []Envelope

+ type MessageLookupItem struct

+ ID string

+ Route string

+ State State

+ type MessageLookupRequest struct

+ IDs []string

+ type MessageLookupResponse struct

+ Items []MessageLookupItem

+ type MessageManageFilterRequest struct

+ Before time.Time

+ Limit int

+ PreviewOnly bool

+ Route string

+ State State

+ Target string

+ type MessageRequeueRequest struct

+ IDs []string

+ type MessageRequeueResponse struct

+ Matched int

+ PreviewOnly bool

+ Requeued int

+ type MessageResumeRequest struct

+ IDs []string

+ type MessageResumeResponse struct

+ Matched int

+ PreviewOnly bool

+ Resumed int

+ type SQLiteOption func(*SQLiteStore)

+ func WithSQLiteDLQRetention(maxAge time.Duration, maxDepth int) SQLiteOption

+ func WithSQLiteDeliveredRetention(maxAge time.Duration) SQLiteOption

+ func WithSQLiteNowFunc(now func() time.Time) SQLiteOption

+ func WithSQLitePollInterval(d time.Duration) SQLiteOption

+ func WithSQLiteQueueLimits(maxDepth int, dropPolicy string) SQLiteOption

+ func WithSQLiteRetention(maxAge, pruneInterval time.Duration) SQLiteOption

+ type SQLiteStore struct

+ func NewSQLiteStore(dbPath string, opts ...SQLiteOption) (*SQLiteStore, error)

+ func (s *SQLiteStore) Ack(leaseID string) error

+ func (s *SQLiteStore) CancelMessages(req MessageCancelRequest) (MessageCancelResponse, error)

+ func (s *SQLiteStore) CancelMessagesByFilter(req MessageManageFilterRequest) (MessageCancelResponse, error)

+ func (s *SQLiteStore) CaptureBacklogTrendSample(at time.Time) error

+ func (s *SQLiteStore) Close() error

+ func (s *SQLiteStore) DeleteDead(req DeadDeleteRequest) (DeadDeleteResponse, error)

+ func (s *SQLiteStore) Dequeue(req DequeueRequest) (DequeueResponse, error)

+ func (s *SQLiteStore) Enqueue(env Envelope) error

+ func (s *SQLiteStore) EnqueueBatch(items []Envelope) (int, error)

+ func (s *SQLiteStore) Extend(leaseID string, extendBy time.Duration) error

+ func (s *SQLiteStore) ListAttempts(req AttemptListRequest) (AttemptListResponse, error)

+ func (s *SQLiteStore) ListBacklogTrend(req BacklogTrendListRequest) (BacklogTrendListResponse, error)

+ func (s *SQLiteStore) ListDead(req DeadListRequest) (DeadListResponse, error)

+ func (s *SQLiteStore) ListMessages(req MessageListRequest) (MessageListResponse, error)

+ func (s *SQLiteStore) LookupMessages(req MessageLookupRequest) (MessageLookupResponse, error)

+ func (s *SQLiteStore) MarkDead(leaseID string, reason string) error

+ func (s *SQLiteStore) Nack(leaseID string, delay time.Duration) error

+ func (s *SQLiteStore) RecordAttempt(attempt DeliveryAttempt) error

+ func (s *SQLiteStore) RequeueDead(req DeadRequeueRequest) (DeadRequeueResponse, error)

+ func (s *SQLiteStore) RequeueMessages(req MessageRequeueRequest) (MessageRequeueResponse, error)

+ func (s *SQLiteStore) RequeueMessagesByFilter(req MessageManageFilterRequest) (MessageRequeueResponse, error)

+ func (s *SQLiteStore) ResumeMessages(req MessageResumeRequest) (MessageResumeResponse, error)

+ func (s *SQLiteStore) ResumeMessagesByFilter(req MessageManageFilterRequest) (MessageResumeResponse, error)

+ func (s *SQLiteStore) Stats() (Stats, error)

+ type State string

+ const StateCanceled

+ const StateDead

+ const StateDelivered

+ const StateLeased

+ const StateQueued

+ type Stats struct

+ ByState map[State]int

+ EarliestQueuedNextRun time.Time

+ OldestQueuedAge time.Duration

+ OldestQueuedReceivedAt time.Time

+ ReadyLag time.Duration

+ TopQueued []BacklogBucket

+ Total int

+ type Store interface

+ Ack func(leaseID string) error

+ CancelMessages func(req MessageCancelRequest) (MessageCancelResponse, error)

+ CancelMessagesByFilter func(req MessageManageFilterRequest) (MessageCancelResponse, error)

+ DeleteDead func(req DeadDeleteRequest) (DeadDeleteResponse, error)

+ Dequeue func(req DequeueRequest) (DequeueResponse, error)

+ Enqueue func(env Envelope) error

+ Extend func(leaseID string, extendBy time.Duration) error

+ ListAttempts func(req AttemptListRequest) (AttemptListResponse, error)

+ ListDead func(req DeadListRequest) (DeadListResponse, error)

+ ListMessages func(req MessageListRequest) (MessageListResponse, error)

+ LookupMessages func(req MessageLookupRequest) (MessageLookupResponse, error)

+ MarkDead func(leaseID string, reason string) error

+ Nack func(leaseID string, delay time.Duration) error

+ RecordAttempt func(attempt DeliveryAttempt) error

+ RequeueDead func(req DeadRequeueRequest) (DeadRequeueResponse, error)

+ RequeueMessages func(req MessageRequeueRequest) (MessageRequeueResponse, error)

+ RequeueMessagesByFilter func(req MessageManageFilterRequest) (MessageRequeueResponse, error)

+ ResumeMessages func(req MessageResumeRequest) (MessageResumeResponse, error)

+ ResumeMessagesByFilter func(req MessageManageFilterRequest) (MessageResumeResponse, error)

+ Stats func() (Stats, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL