Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

blockqueue

package module

Go to main page

Versions in this module

v0

v0.2.0

Jul 11, 2026

Changes in this version

+ var ErrCommitUnknown = errors.New("publish commit outcome unknown")

+ var ErrDeliveryNotFound = errors.New("delivery not found")

+ var ErrDeliveryTerminal = errors.New("delivery is already terminal")

+ var ErrIdempotencyConflict = errors.New("idempotency key conflicts with a different message")

+ var ErrInvalidPublish = errors.New("invalid publish request")

+ var ErrInvalidReceipt = errors.New("receipt_token is required")

+ var ErrInvalidSubscriber = errors.New("invalid subscriber")

+ var ErrInvalidTopic = errors.New("invalid topic")

+ var ErrInvalidTransaction = errors.New("invalid blockqueue transaction")

+ var ErrLeaseLost = errors.New("delivery lease lost")

+ var ErrMigrationChecksum = errors.New("migration checksum mismatch")

+ var ErrNoActiveSubscriber = errors.New("topic has no active subscriber")

+ var ErrPendingBudgetExceeded = errors.New("pending write budget exceeded")

+ var ErrQueueNotRunning = errors.New("blockqueue is not running")

+ var ErrQueueStopping = errors.New("blockqueue is stopping")

+ var ErrResourceConflict = errors.New("resource already exists")

+ var ErrResourcePaused = errors.New("topic or subscriber is paused")

+ var ErrScheduleLeaseLost = errors.New("schedule lease lost")

+ var ErrScheduleNotFound = errors.New("schedule not found")

+ var ErrScheduleOverlap = errors.New("previous schedule run is still active")

+ var ErrScheduleVersion = errors.New("stale schedule version")

+ var ErrSubscriberDeleted = errors.New("subscriber was deleted")

+ var ErrSubscriberNotFound = errors.New("subscriber not found")

+ var ErrTopicNotFound = errors.New("topic not found")

+ var ErrUnsupportedDialect = errors.New("unsupported database dialect")

+ var ErrWriterClosed = errors.New("writer closed")

+ var ErrWriterDrainTimeout = errors.New("writer shutdown with unpersisted messages")

+ func DashboardFS() fs.FS

+ func Migrate(ctx context.Context, driver store.Driver) error

+ type BatchAckItem struct

+ MessageID string

+ ReceiptToken string

+ type BatchNackItem struct

+ Error string

+ MessageID string

+ ReceiptToken string

+ RetryDelay time.Duration

+ type Clock interface

+ After func(time.Duration) <-chan time.Time

+ Now func() time.Time

+ type CommitUnknownError struct

+ Cause error

+ MessageIDs []string

+ func (err *CommitUnknownError) Error() string

+ func (err *CommitUnknownError) Is(target error) bool

+ func (err *CommitUnknownError) Unwrap() error

+ type Deliveries []Delivery

+ type Delivery struct

+ CancelReason string

+ CancelledAt *time.Time

+ CorrelationID string

+ CreatedAt time.Time

+ DeliveryCount int

+ FailureCount int

+ Headers map[string]string

+ ID string

+ LeaseExpiresAt *time.Time

+ Message string

+ Priority int

+ ReceiptToken string

+ Status string

+ VisibleAt time.Time

+ type DeliveryError struct

+ Error string

+ FailedAt time.Time

+ FailureCount int

+ ID string

+ MessageID string

+ SubscriberID string

+ type DeliveryErrorPage struct

+ Errors []DeliveryError

+ NextCursor string

+ type DeliveryPage struct

+ Messages Deliveries

+ NextCursor string

+ type DeliveryResult struct

+ Error string

+ MessageID string

+ Status string

+ SubscriberID string

+ type LifecycleState uint32

+ const LifecycleNew

+ const LifecycleRunning

+ const LifecycleStopped

+ const LifecycleStopping

+ func (s LifecycleState) String() string

+ type Message struct

+ CorrelationID string

+ Delay string

+ Headers map[string]string

+ IdempotencyKey string

+ Message string

+ Priority int

+ ScheduleAt string

+ type MessageDeliveryStatus struct

+ CancelReason string

+ CancelledAt *time.Time

+ DeliveryCount int

+ FailureCount int

+ ProcessedAt *time.Time

+ Status string

+ Subscriber string

+ SubscriberID string

+ VisibleAt time.Time

+ type MessageStatus struct

+ CorrelationID string

+ CreatedAt time.Time

+ Deliveries []MessageDeliveryStatus

+ Headers map[string]string

+ ID string

+ IdempotencyKey string

+ Message string

+ Priority int

+ ScheduledAt time.Time

+ TopicID string

+ type Options struct

+ CheckpointInterval time.Duration

+ Clock Clock

+ DeadLetterRetention time.Duration

+ DisableMetrics bool

+ MetricRegisterer prometheus.Registerer

+ ReadinessBacklog int64

+ RetentionPeriod time.Duration

+ ScheduleRunRetention time.Duration

+ ShutdownTimeout time.Duration

+ Writer WriterOptions

+ type PublishReceipt struct

+ Duplicate *bool

+ MessageID string

+ ScheduledAt time.Time

+ State string

+ type PublishReceipts []PublishReceipt

+ type Queue struct

+ func (q *Queue) AckDelivery(ctx context.Context, topic Topic, subscriber, messageID, receipt string) error

+ func (q *Queue) AckDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, ...) error

+ func (q *Queue) BatchAckDeliveries(ctx context.Context, topic Topic, subscriber string, requests []BatchAckItem) []DeliveryResult

+ func (q *Queue) BatchNackDeliveries(ctx context.Context, topic Topic, subscriber string, requests []BatchNackItem) []DeliveryResult

+ func (q *Queue) BatchPublish(ctx context.Context, topic Topic, requests []Message) (PublishReceipts, error)

+ func (q *Queue) BatchPublishAsync(ctx context.Context, topic Topic, requests []Message) (PublishReceipts, error)

+ func (q *Queue) BatchPublishDurable(ctx context.Context, topic Topic, requests []Message) (PublishReceipts, error)

+ func (q *Queue) BatchPublishTx(ctx context.Context, tx *sql.Tx, topic Topic, requests []Message) (PublishReceipts, error)

+ func (q *Queue) CancelDelivery(ctx context.Context, topic Topic, subscriber, messageID, reason string) error

+ func (q *Queue) CancelDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, ...) error

+ func (q *Queue) CancelMessage(ctx context.Context, topic Topic, messageID, reason string) ([]DeliveryResult, error)

+ func (q *Queue) CancelMessageTx(ctx context.Context, tx *sql.Tx, topic Topic, messageID, reason string) ([]DeliveryResult, error)

+ func (q *Queue) Claim(ctx context.Context, topic Topic, subscriber string, limit int, ...) (Deliveries, error)

+ func (q *Queue) ClaimWait(ctx context.Context, topic Topic, subscriber string, limit int, ...) (Deliveries, error)

+ func (q *Queue) Close()

+ func (q *Queue) CreateSchedule(ctx context.Context, topic Topic, input ScheduleInput) (Schedule, error)

+ func (q *Queue) CreateSubscribers(ctx context.Context, topic Topic, subscribers Subscribers) error

+ func (q *Queue) CreateTopic(ctx context.Context, topic Topic, subscribers Subscribers) error

+ func (q *Queue) DeleteSchedule(ctx context.Context, topic Topic, scheduleID string) error

+ func (q *Queue) DeleteSubscriber(ctx context.Context, topic Topic, subscriber string) error

+ func (q *Queue) DeleteTopic(ctx context.Context, topic Topic) error

+ func (q *Queue) DeliveryErrors(ctx context.Context, topic Topic, subscriber, messageID string, limit int, ...) (DeliveryErrorPage, error)

+ func (q *Queue) ExtendLease(ctx context.Context, topic Topic, subscriber, messageID, receipt string, ...) (time.Time, error)

+ func (q *Queue) GetMessageStatus(ctx context.Context, topic Topic, messageID string) (MessageStatus, error)

+ func (q *Queue) GetSchedule(ctx context.Context, topic Topic, scheduleID string) (Schedule, error)

+ func (q *Queue) GetSubscribersStatus(ctx context.Context, topic Topic) (SubscriberStatuses, error)

+ func (q *Queue) GetTopic(topicName string) (Topic, bool)

+ func (q *Queue) GetTopics(ctx context.Context, filter TopicFilter) (Topics, error)

+ func (q *Queue) ListDeliveries(ctx context.Context, topic Topic, subscriber string, deadLetter bool, ...) (DeliveryPage, error)

+ func (q *Queue) ListSchedules(ctx context.Context, topic Topic) ([]Schedule, error)

+ func (q *Queue) Live() bool

+ func (q *Queue) NackDelivery(ctx context.Context, topic Topic, subscriber, messageID, receipt string, ...) error

+ func (q *Queue) NackDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, ...) error

+ func (q *Queue) PauseSchedule(ctx context.Context, topic Topic, scheduleID string, paused bool) error

+ func (q *Queue) PauseSubscriber(ctx context.Context, topic Topic, subscriber string) error

+ func (q *Queue) PauseTopic(ctx context.Context, topic Topic) error

+ func (q *Queue) Publish(ctx context.Context, topic Topic, request Message) (PublishReceipt, error)

+ func (q *Queue) PublishAsync(ctx context.Context, topic Topic, request Message) (PublishReceipt, error)

+ func (q *Queue) PublishDurable(ctx context.Context, topic Topic, request Message) (PublishReceipt, error)

+ func (q *Queue) PublishTx(ctx context.Context, tx *sql.Tx, topic Topic, request Message) (PublishReceipt, error)

+ func (q *Queue) Ready(ctx context.Context) bool

+ func (q *Queue) ReplayDeadLetters(ctx context.Context, topic Topic, subscriber string, messageIDs []string) []DeliveryResult

+ func (q *Queue) ResumeSubscriber(ctx context.Context, topic Topic, subscriber string) error

+ func (q *Queue) ResumeTopic(ctx context.Context, topic Topic) error

+ func (q *Queue) Run(ctx context.Context) error

+ func (q *Queue) RunScheduleNow(ctx context.Context, topic Topic, scheduleID string, force bool) (ScheduleRun, error)

+ func (q *Queue) ScheduleRunHistory(ctx context.Context, topic Topic, scheduleID string, limit int, cursor string) (ScheduleRunPage, error)

+ func (q *Queue) Shutdown(ctx context.Context) error

+ func (q *Queue) SnoozeDelivery(ctx context.Context, topic Topic, subscriber, messageID, receipt string, ...) (time.Time, error)

+ func (q *Queue) SnoozeDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, ...) (time.Time, error)

+ func (q *Queue) State() LifecycleState

+ func (q *Queue) UpdateSchedule(ctx context.Context, topic Topic, scheduleID string, expectedVersion int, ...) (Schedule, error)

+ func (q *Queue) WithTx(ctx context.Context, options *sql.TxOptions, fn func(*sql.Tx) error) error

+ func (q *Queue) WriterHealthy() bool

+ type RetryPolicy struct

+ InitialDelay string

+ Jitter float64

+ MaxDelay string

+ Multiplier float64

+ type Schedule struct

+ CorrelationID sql.NullString

+ CreatedAt time.Time

+ CronExpression string

+ FencingToken int64

+ Headers string

+ ID string

+ LeaseExpiresAt sql.NullTime

+ Message string

+ MisfirePolicy string

+ Name string

+ NextRunAt time.Time

+ OverlapPolicy string

+ OwnerID sql.NullString

+ Paused bool

+ Priority int

+ Timezone string

+ TopicID string

+ UpdatedAt time.Time

+ Version int

+ func (schedule Schedule) MarshalJSON() ([]byte, error)

+ func (schedule Schedule) PublicHeaders() map[string]string

+ type ScheduleInput struct

+ CorrelationID string

+ CronExpression string

+ Headers map[string]string

+ Message string

+ MisfirePolicy string

+ Name string

+ OverlapPolicy string

+ Priority int

+ Timezone string

+ type ScheduleRun struct

+ CreatedAt time.Time

+ Error sql.NullString

+ FinishedAt sql.NullTime

+ ID string

+ MessageID sql.NullString

+ ScheduleID string

+ ScheduledFor time.Time

+ StartedAt time.Time

+ Status string

+ func (run ScheduleRun) MarshalJSON() ([]byte, error)

+ type ScheduleRunPage struct

+ NextCursor string

+ Runs []ScheduleRun

+ type Subscriber struct

+ CreatedAt time.Time

+ DeletedAt *time.Time

+ ID uuid.UUID

+ Name string

+ Options SubscriberOptions

+ Paused bool

+ TopicID uuid.UUID

+ TopicName string

+ func NewSubscriber(topic Topic, name string, options SubscriberOptions) Subscriber

+ type SubscriberOptions struct

+ DequeueBatchSize int

+ MaxAttempts int

+ RetryPolicy RetryPolicy

+ VisibilityDuration string

+ func (options *SubscriberOptions) Scan(source any) error

+ func (options SubscriberOptions) Value() (driver.Value, error)

+ type SubscriberStatus struct

+ Name string

+ TopicID uuid.UUID

+ UnackedMessage int

+ UnpublishedMessage int

+ type SubscriberStatuses []SubscriberStatus

+ type Subscribers []Subscriber

+ type Topic struct

+ CreatedAt time.Time

+ DeletedAt *time.Time

+ ID uuid.UUID

+ Name string

+ Paused bool

+ func NewTopic(name string) Topic

+ type TopicFilter struct

+ Names []string

+ WithDeleted bool

+ type Topics []Topic

+ type WriterOptions struct

+ BatchSize int

+ FlushInterval time.Duration

+ MaxPendingBytes int64

+ MaxPendingMessages int64

+ RetryMax time.Duration

+ RetryMin time.Duration

+ func DefaultWriterOptions() WriterOptions

v0.1.1

Jan 10, 2026

Changes in this version

+ var ErrJobNotFound = errors.New("job not found")

+ var ErrListenerDeleted = errors.New("listener was deleted")

+ var ErrListenerNotFound = errors.New("listener not found")

+ var ErrListenerRetryMessageNotFound = errors.New("error ack message. message_id not found")

+ var ErrListenerShutdown = errors.New("listener shutdown")

+ var ErrMessageNotFound = errors.New("message not found")

+ func InvalidateSubscriberCache(topicId uuid.UUID)

+ type BlockQueue struct

+ Opt BlockQueueOption

+ func New[V chan bqio.ResponseMessages](driver Driver, opt BlockQueueOption) *BlockQueue[V]

+ func (q *BlockQueue[V]) Ack(ctx context.Context, topic core.Topic, subscriberName, messageId string) error

+ func (q *BlockQueue[V]) AddJob(ctx context.Context, topic core.Topic, subscribers core.Subscribers) error

+ func (q *BlockQueue[V]) AddSubscriber(ctx context.Context, topic core.Topic, subscribers core.Subscribers) error

+ func (q *BlockQueue[V]) Close()

+ func (q *BlockQueue[V]) DeleteJob(topic core.Topic) error

+ func (q *BlockQueue[V]) DeleteSubscriber(ctx context.Context, topic core.Topic, subcriber string) error

+ func (q *BlockQueue[V]) GetSubscribersStatus(ctx context.Context, topic core.Topic) (bqio.SubscriberMessages, error)

+ func (q *BlockQueue[V]) GetTopics(ctx context.Context, filter core.FilterTopic) (core.Topics, error)

+ func (q *BlockQueue[V]) Publish(ctx context.Context, topic core.Topic, request bqio.Publish) error

+ func (q *BlockQueue[V]) Read(ctx context.Context, topic core.Topic, subscriber string) (bqio.ResponseMessages, error)

+ func (q *BlockQueue[V]) Run(ctx context.Context) error

+ type BlockQueueOption struct

+ CheckpointInterval time.Duration

+ WriteBufferConfig WriteBufferConfig

+ type Driver interface

+ Close func() error

+ Conn func() *sqlx.DB

+ type Http struct

+ Stream *BlockQueue[chan io.ResponseMessages]

+ func (h *Http) Router() http.Handler

+ type Job struct

+ Id uuid.UUID

+ Name string

+ type Listener struct

+ Id string

+ JobId string

+ PriorityQueue *pqueue.PriorityQueue[V]

+ SubscriberId uuid.UUID

+ type MessageCounter struct

+ Name string

+ UnackMessage int

+ UnpublishMessage int

+ type SubscriberInfo struct

+ DequeueBatchSize int

+ Id uuid.UUID

+ MaxAttempts int

+ MaxBackoff time.Duration

+ Name string

+ PollInterval time.Duration

+ VisibilityDuration time.Duration

+ type SubscriberMessage struct

+ CreatedAt time.Time

+ Id int64

+ Message string

+ MessageId string

+ RetryCount int

+ Status string

+ SubscriberId uuid.UUID

+ TopicId uuid.UUID

+ VisibleAt time.Time

+ type SubscriberMessages []SubscriberMessage

+ type SubscriberQueueStats struct

+ Delivered int

+ Pending int

+ type WriteBuffer struct

+ func NewWriteBuffer(ctx context.Context, database *db, config WriteBufferConfig) *WriteBuffer

+ func (w *WriteBuffer) Close()

+ func (w *WriteBuffer) Enqueue(topicId uuid.UUID, messageId, message string)

+ type WriteBufferConfig struct

+ BatchSize int

+ BufferSize int

+ FlushInterval time.Duration

+ func DefaultWriteBufferConfig() WriteBufferConfig

v0.0.5-beta

Apr 5, 2025

v0.0.4-beta

Jul 25, 2024

v0.0.3-beta

Jun 18, 2024

v0.0.2-beta

Dec 18, 2023

v0.0.1-beta

Dec 11, 2023