blockqueue

package module
v0.3.1 Latest Latest
Warning

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

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

README

BlockQueue logo

BlockQueue

A durable, embeddable fan-out queue for Go, backed by SQLite or PostgreSQL.

Go Reference CI status Apache-2.0 license

BlockQueue stores one canonical message and creates one independently leased delivery for every subscriber. Use it as an imported Go package or run the optional HTTP service for cross-language clients.

flowchart LR
    P["Publisher"] --> T["Topic: orders"]
    T --> M[("Canonical message")]
    M --> D1["payment delivery"]
    M --> D2["email delivery"]
    M --> D3["analytics delivery"]
    D1 --> W1["payment workers"]
    D2 --> W2["email workers"]
    D3 --> W3["analytics workers"]

Why BlockQueue:

  • SQLite first: embed a durable queue without operating another service.
  • PostgreSQL scale-out: coordinate claims and schedules across processes.
  • Real fan-out: every subscriber owns independent retry and completion state.
  • Transactional APIs: commit application rows with publish or completion.
  • HTTP included: expose the same queue model to non-Go applications.

Turso/libSQL support is experimental. Delivery is at least once; handlers must be idempotent unless their database side effect and completion share one transaction.

Install

go get github.com/yudhasubki/blockqueue

The standalone server is optional:

go build -o blockqueue ./cmd/blockqueue

Quick start

This complete example creates one topic, publishes durably, claims the subscriber delivery, and ACKs the exact lease that was returned:

package main

import (
	"context"
	"log"
	"time"

	"github.com/yudhasubki/blockqueue"
	"github.com/yudhasubki/blockqueue/store/sqlite"
)

func main() {
	ctx := context.Background()
	driver, err := sqlite.Open("queue.db", sqlite.Config{})
	if err != nil {
		log.Fatal(err)
	}

	queue := blockqueue.New(driver, blockqueue.Options{})
	if err := queue.Run(ctx); err != nil {
		log.Fatal(err)
	}
	defer queue.Close()

	topic := blockqueue.NewTopic("orders")
	subscriber := blockqueue.NewSubscriber(topic, "fulfillment", blockqueue.SubscriberOptions{
		MaxAttempts:        5,
		VisibilityDuration: "30s",
	})
	if err := queue.CreateTopic(ctx, topic, blockqueue.Subscribers{subscriber}); err != nil {
		log.Fatal(err)
	}

	receipt, err := queue.Publish(ctx, topic, blockqueue.Message{
		Message:        `{"order_id":"1022"}`,
		IdempotencyKey: "order-1022",
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("persisted %s", receipt.MessageID)

	deliveries, err := queue.ClaimWait(ctx, topic, subscriber.Name, 10, time.Minute)
	if err != nil {
		log.Fatal(err)
	}
	for _, delivery := range deliveries {
		log.Printf("processing %s", delivery.Message)
		if err := queue.AckDelivery(
			ctx, topic, subscriber.Name, delivery.ID, delivery.ReceiptToken,
		); err != nil {
			log.Fatal(err)
		}
	}
}

Publish waits for the canonical message and complete subscriber fan-out to commit. Use PublishAsync only when process-local admission is sufficient.

Managed Go workers

The worker package handles claim loops, bounded concurrency, heartbeats, panic recovery, retry, batched completion, and graceful drain:

type FulfillOrder struct {
	OrderID string `json:"order_id"`
}

runner, err := worker.NewJSON(
	queue,
	topic,
	subscriber.Name,
	worker.TypedHandlerFunc[FulfillOrder](func(
		ctx context.Context,
		job *worker.TypedJob[FulfillOrder],
	) error {
		return fulfill(ctx, job.Args.OrderID)
	}),
	worker.Options{Concurrency: 16},
)
if err != nil {
	log.Fatal(err)
}

if err := runner.Run(ctx); err != nil {
	log.Fatal(err)
}

Run several topic/subscriber workers under one lifecycle with worker.Group:

group, err := worker.NewGroup(paymentWorker, emailWorker, analyticsWorker)
if err != nil {
	log.Fatal(err)
}

if err := group.Run(ctx); err != nil {
	log.Fatal(err)
}

Each worker keeps its own concurrency limit. A terminal error from one worker cancels and drains its peers; cancelling ctx gracefully drains the whole group. For the same subscriber, prefer one worker with higher Concurrency.

Returning nil ACKs the job. Returning an error NACKs it using the subscriber retry policy. Use worker.RetryAfter for one custom delay and worker.CancelJob for a permanent outcome. See the runnable worker example.

Transactional completion

When application tables and BlockQueue share the same database, Job.CompleteTx commits the business update and ACK together:

return job.CompleteTx(ctx, nil, func(tx *sql.Tx) error {
	_, err := tx.ExecContext(ctx,
		"UPDATE orders SET fulfilled_at = ? WHERE id = ?",
		time.Now().UTC(), job.Args.OrderID,
	)
	return err
})

If that transaction rolls back, both the update and ACK roll back. Producer PublishTx provides the same atomic boundary for inserting application data and publishing later work. The consumer still runs in a separate transaction after publish commits.

See Concepts and the runnable transactional example. Remote HTTP requests cannot join a caller's database transaction; use the documented outbox pattern.

HTTP service

Copy config.yaml.example, then start the server:

./blockqueue migrate -config config.yaml
./blockqueue http -config config.yaml

Create a topic and publish durably:

curl -X POST http://127.0.0.1:8080/v1/topics \
  -H 'Content-Type: application/json' \
  -d '{
    "name":"orders",
    "subscribers":[{
      "name":"fulfillment",
      "option":{"max_attempts":5,"visibility_duration":"30s"}
    }]
  }'

curl -X POST \
  'http://127.0.0.1:8080/v1/topics/orders/messages?wait_for=commit' \
  -H 'Content-Type: application/json' \
  -d '{"message":"order-1022","idempotency_key":"order-1022"}'

HTTP publish is async by default and returns 202. Add ?wait_for=commit when the response must imply a committed message and fan-out. The OpenAPI 3.1 contract is served at /openapi.json; errors use application/problem+json.

The binary binds to 127.0.0.1 by default. Public deployments need an authenticated reverse proxy or private network.

Capabilities

Area Included
Publish Durable/async batch publish, idempotency, priority, headers, delay, absolute scheduling
Delivery Receipt leases, ACK, NACK, heartbeat, snooze, cancellation, DLQ and replay
Scheduler Five-field cron, IANA timezone, misfire recovery, overlap policy, run history
Transactions Publish and delivery completion with application rows in the same database
Operations /livez, /readyz, dashboard, Prometheus metrics, bounded retention and maintenance
API Importable Go packages and an HTTP /v1 contract with cursor pagination

Storage

Backend Status Coordination Strict default
SQLite Supported Embedded, single coordinated writer WAL + synchronous=FULL
PostgreSQL Supported Multi-process claims and scheduler leases TLS + synchronous_commit=on
Turso/libSQL Experimental Smoke-test scope Backend dependent

The database is authoritative. In-memory wakeups and PostgreSQL notifications only reduce latency. store.DurabilityBalanced is an explicit performance tradeoff and can lose the newest acknowledged writes after power loss or failover.

Documentation

Read When you need
Concepts Topic fan-out, delivery leases, transactions, topology, worker groups, and maintenance
Architecture Packages, locking, persistence, and multi-node algorithms
HTTP outbox Atomic business writes for remote HTTP publishers
Benchmarks Reproducible SQLite/PostgreSQL performance and acceptance runs
Changelog Release features, fixes, and explicitly documented breaking changes

There is one queue engine and one current HTTP contract at /v1. BlockQueue is pre-1.0; minor releases can contain explicitly documented breaking changes. The root package, worker, httpapi, store/sqlite, and store/postgres are supported APIs. store/turso remains experimental.

Development

go test ./...
go test -race ./...
go vet ./...

CI runs the shared SQLite/PostgreSQL contract suite, race detector, lint, staticcheck, vulnerability checks, and guarded benchmark smoke tests.

Report vulnerabilities through private vulnerability reporting. BlockQueue is licensed under the Apache License 2.0.

Documentation

Overview

Package blockqueue provides a durable, embeddable, at-least-once message queue for Go applications. A Queue stores one canonical message and one receipt-fenced delivery per subscriber, supporting fan-out, retries, delayed delivery, recurring schedules, cancellation, and dead-letter queues.

SQLite is suited to single-process deployments and PostgreSQL supports multi-process consumers. Publish and delivery completion can participate in caller-owned database transactions when application tables use the same database. The optional worker and httpapi packages provide a managed Go consumer runtime and a cross-language HTTP surface without changing the storage contract.

Delivery is at-least-once: handlers must make external side effects idempotent or commit them atomically with AckDeliveryTx.

Index

Examples

Constants

View Source 
const (
	// MaximumMessageBytes is the largest accepted UTF-8 message payload.
	MaximumMessageBytes = 1 << 20
	// MaximumHeadersBytes is the largest encoded headers object.
	MaximumHeadersBytes = 16 << 10
	// MaximumIdempotencyKeyBytes bounds a per-topic idempotency key.
	MaximumIdempotencyKeyBytes = 128
	// MaximumCorrelationIDBytes bounds an optional correlation identifier.
	MaximumCorrelationIDBytes = 255
	// MinimumPriority is the lowest accepted delivery priority.
	MinimumPriority = -1000
	// MaximumPriority is the highest accepted delivery priority.
	MaximumPriority = 1000
	// MaximumDeliveryLease is the longest claim or heartbeat lease.
	MaximumDeliveryLease = subscriberconfig.MaximumDeliveryLease
	// MinimumCheckpointInterval bounds automatic SQLite WAL checkpoints.
	MinimumCheckpointInterval = 30 * time.Second
)
View Source 
const (
	PublishStateAdmitted  = "admitted"
	PublishStatePersisted = "persisted"
	PublishStateStaged    = "staged"
)

Publish receipt states are stable public API values.

View Source 
const (
	DeliveryStatusPending    = persistence.DeliveryStatusPending
	DeliveryStatusDelivered  = persistence.DeliveryStatusDelivered
	DeliveryStatusProcessed  = persistence.DeliveryStatusProcessed
	DeliveryStatusDeadLetter = persistence.DeliveryStatusDeadLetter
	DeliveryStatusCancelled  = persistence.DeliveryStatusCancelled
)

Delivery states are shared by the database state machine and public API.

View Source 
const (
	ScheduleRunStatusRunning   = persistence.ScheduleRunStatusRunning
	ScheduleRunStatusCompleted = persistence.ScheduleRunStatusCompleted
	ScheduleRunStatusSkipped   = persistence.ScheduleRunStatusSkipped
	ScheduleRunStatusFailed    = persistence.ScheduleRunStatusFailed
)

Schedule run states are stable public API and persisted values.

View Source 
const (
	ScheduleMisfirePolicyFireOnce = persistence.ScheduleMisfirePolicyFireOnce
	ScheduleOverlapPolicySkip     = persistence.ScheduleOverlapPolicySkip
)

Schedule policies currently expose the supported v0.2 behavior.

View Source 
const DeliveryResultStatusFailed = "failed"

DeliveryResultStatusFailed is an operation result, not a persisted delivery state. The typed error returned by single-item methods remains authoritative.

View Source 
const MaxDeliveryTextBytes = persistence.MaxDeliveryTextBytes

MaxDeliveryTextBytes is the maximum persisted size of a NACK error or cancellation reason. Longer values are truncated on a valid UTF-8 boundary.

Variables

View Source 
var (
	ErrLeaseLost        = persistence.ErrLeaseLost
	ErrDeliveryNotFound = persistence.ErrDeliveryNotFound
	ErrInvalidReceipt   = persistence.ErrInvalidReceipt
	ErrResourcePaused   = errors.New("topic or subscriber is paused")
)
View Source 
var (
	ErrMigrationChecksum  = persistence.ErrMigrationChecksum
	ErrUnsupportedDialect = persistence.ErrUnsupportedDialect
)
View Source 
var (
	ErrTopicNotFound      = persistence.ErrTopicNotFound
	ErrQueueNotRunning    = errors.New("blockqueue is not running")
	ErrQueueStopping      = errors.New("blockqueue is stopping")
	ErrNoActiveSubscriber = persistence.ErrNoActiveSubscriber
	ErrInvalidPublish     = persistence.ErrInvalidPublish
	ErrInvalidCursor      = persistence.ErrInvalidCursor
	ErrInvalidTopic       = errors.New("invalid topic")
	ErrInvalidSubscriber  = errors.New("invalid subscriber")
	ErrResourceConflict   = persistence.ErrResourceConflict
)
View Source 
var (
	ErrScheduleNotFound  = persistence.ErrScheduleNotFound
	ErrScheduleVersion   = persistence.ErrScheduleVersion
	ErrScheduleOverlap   = persistence.ErrScheduleOverlap
	ErrScheduleLeaseLost = persistence.ErrScheduleLeaseLost
)
View Source 
var (
	ErrSubscriberNotFound = persistence.ErrSubscriberNotFound
	ErrSubscriberDeleted  = errors.New("subscriber was deleted")
)
View Source 
var (
	ErrWriterClosed          = persistence.ErrWriterClosed
	ErrPendingBudgetExceeded = errors.New("pending write budget exceeded")
	ErrWriterDrainTimeout    = errors.New("writer shutdown with unpersisted messages")
	ErrIdempotencyConflict   = persistence.ErrIdempotencyConflict
	ErrCommitUnknown         = errors.New("publish commit outcome unknown")
)
View Source 
var ErrDeliveryTerminal = persistence.ErrDeliveryTerminal

ErrDeliveryTerminal reports an attempted cancellation of a delivery that was already processed or dead-lettered.

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

ErrInvalidTransaction reports a missing or otherwise unusable caller transaction.

View Source 
var ErrTransactionCommitUnknown = errors.New("transaction commit outcome unknown")

ErrTransactionCommitUnknown means Commit returned a connection-level error for a caller-owned transaction. The database may have committed the application writes and queue changes; callers must reconcile and must not blindly repeat non-idempotent business operations.

Functions

func DashboardFS added in v0.2.0 

func DashboardFS() fs.FS

DashboardFS returns the embedded optional HTTP dashboard assets.

func Migrate added in v0.2.0 

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

Migrate installs the current schema and applies future ordered migrations. Migrate applies the embedded, checksummed schema for driver's backend. Queue.Run invokes it automatically; standalone deployments may call it explicitly before serving traffic.

Types

type BatchAckItem added in v0.2.0 

type BatchAckItem struct {
	MessageID    string
	ReceiptToken string
}

BatchAckItem identifies one receipt-fenced acknowledgement.

type BatchNackItem added in v0.2.0 

type BatchNackItem struct {
	MessageID    string
	ReceiptToken string
	RetryDelay   time.Duration
	Error        string
}

BatchNackItem identifies one receipt-fenced failure and optional retry delay.

type Clock added in v0.2.0 

type Clock interface {
	Now() time.Time
	After(time.Duration) <-chan time.Time
}

Clock supplies scheduler time and is injectable for deterministic tests.

type CommitUnknownError added in v0.2.0 

type CommitUnknownError struct {
	MessageIDs []string
	Cause      error
}

CommitUnknownError means the caller stopped waiting after admission. The writer still owns the messages and may already have committed them; callers can safely reconcile or retry using the included stable IDs/idempotency keys.

func (*CommitUnknownError) Error added in v0.2.0 

func (err *CommitUnknownError) Error() string

func (*CommitUnknownError) Is added in v0.2.0 

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

func (*CommitUnknownError) Unwrap added in v0.2.0 

func (err *CommitUnknownError) Unwrap() error

type Deliveries added in v0.2.0 

type Deliveries []Delivery

Deliveries is a collection of claimed or listed deliveries.

type Delivery added in v0.2.0 

type Delivery struct {
	ID             string            `json:"id"`
	Message        string            `json:"message"`
	Headers        map[string]string `json:"headers,omitempty"`
	CorrelationID  string            `json:"correlation_id,omitempty"`
	Status         string            `json:"status,omitempty"`
	DeliveryCount  int               `json:"delivery_count,omitempty"`
	FailureCount   int               `json:"failure_count,omitempty"`
	Priority       int               `json:"priority,omitempty"`
	ReceiptToken   string            `json:"receipt_token,omitempty"`
	LeaseExpiresAt *time.Time        `json:"lease_expires_at,omitempty"`
	VisibleAt      time.Time         `json:"visible_at"`
	CreatedAt      time.Time         `json:"created_at"`
	CancelledAt    *time.Time        `json:"cancelled_at,omitempty"`
	CancelReason   string            `json:"cancel_reason,omitempty"`
}

Delivery is one subscriber-specific view of a canonical message.

type DeliveryError added in v0.2.0 

type DeliveryError struct {
	ID           string    `db:"id" json:"id"`
	MessageID    string    `db:"message_id" json:"message_id"`
	SubscriberID string    `db:"subscriber_id" json:"subscriber_id"`
	FailureCount int       `db:"failure_count" json:"failure_count"`
	Error        string    `db:"error" json:"error"`
	FailedAt     time.Time `db:"failed_at" json:"failed_at"`
}

DeliveryError is an append-only record of one NACK or lease expiry. A DLQ replay resets the delivery failure count but does not erase prior records.

type DeliveryErrorPage added in v0.2.0 

type DeliveryErrorPage struct {
	Errors     []DeliveryError `json:"errors"`
	NextCursor string          `json:"next_cursor,omitempty"`
}

DeliveryErrorPage is a cursor-paginated delivery failure history.

type DeliveryPage added in v0.2.0 

type DeliveryPage struct {
	Messages   Deliveries `json:"messages"`
	NextCursor string     `json:"next_cursor,omitempty"`
}

DeliveryPage is one cursor-paginated page of active or dead-letter work.

type DeliveryResult added in v0.2.0 

type DeliveryResult struct {
	MessageID    string `json:"message_id"`
	SubscriberID string `json:"subscriber_id,omitempty"`
	Status       string `json:"status"`
	Error        string `json:"error,omitempty"`
}

DeliveryResult reports the per-item outcome of a batch operation.

type LifecycleState added in v0.2.0 

type LifecycleState uint32

LifecycleState describes whether a Queue accepts work or is shutting down. 

const (
	LifecycleNew LifecycleState = iota
	LifecycleRunning
	LifecycleStopping
	LifecycleStopped
)

Queue lifecycle states progress monotonically from new to stopped.

func (LifecycleState) String added in v0.2.0 

func (s LifecycleState) String() string

String returns the stable lowercase lifecycle name.

type Message added in v0.2.0 

type Message struct {
	Message        string            `json:"message"`
	Headers        map[string]string `json:"headers,omitempty"`
	CorrelationID  string            `json:"correlation_id,omitempty"`
	IdempotencyKey string            `json:"idempotency_key,omitempty"`
	Priority       int               `json:"priority,omitempty"`
	Delay          string            `json:"delay,omitempty"`
	ScheduleAt     string            `json:"schedule_at,omitempty"`
}

Message is a canonical publish request shared by all active subscribers.

type MessageDeliveryStatus added in v0.2.0 

type MessageDeliveryStatus struct {
	SubscriberID  string     `db:"subscriber_id" json:"subscriber_id"`
	Subscriber    string     `db:"subscriber" json:"subscriber"`
	Status        string     `db:"status" json:"status"`
	DeliveryCount int        `db:"delivery_count" json:"delivery_count"`
	FailureCount  int        `db:"failure_count" json:"failure_count"`
	VisibleAt     time.Time  `db:"visible_at" json:"visible_at"`
	ProcessedAt   *time.Time `db:"processed_at" json:"processed_at,omitempty"`
	CancelledAt   *time.Time `db:"cancelled_at" json:"cancelled_at,omitempty"`
	CancelReason  string     `db:"cancel_reason" json:"cancel_reason,omitempty"`
}

MessageDeliveryStatus describes one subscriber's delivery state.

type MessageStatus added in v0.2.0 

type MessageStatus struct {
	ID             string                  `json:"id"`
	TopicID        string                  `json:"topic_id"`
	Message        string                  `json:"message"`
	Headers        map[string]string       `json:"headers,omitempty"`
	CorrelationID  string                  `json:"correlation_id,omitempty"`
	IdempotencyKey string                  `json:"idempotency_key,omitempty"`
	Priority       int                     `json:"priority"`
	ScheduledAt    time.Time               `json:"scheduled_at"`
	CreatedAt      time.Time               `json:"created_at"`
	Deliveries     []MessageDeliveryStatus `json:"deliveries"`
}

MessageStatus is the canonical message and the current state of every subscriber delivery created with it.

type Options added in v0.2.0 

type Options struct {
	Writer               WriterOptions
	CheckpointInterval   time.Duration         // Default: 30s
	RetentionPeriod      time.Duration         // Default: 7d
	DeadLetterRetention  time.Duration         // Default: disabled; operators opt in explicitly
	ScheduleRunRetention time.Duration         // Default: 30d
	ShutdownTimeout      time.Duration         // Default: 30s for Close
	ReadinessBacklog     int64                 // Default: 90% of pending message budget
	Clock                Clock                 // Optional deterministic scheduler clock
	DisableMetrics       bool                  // Skip per-message metric updates on the hot path
	MetricRegisterer     prometheus.Registerer // Optional collector registry; defaults to Prometheus global registry
}

Options configures queue persistence, maintenance, shutdown, and metrics. Zero values select the documented production defaults.

type PublishReceipt added in v0.2.0 

type PublishReceipt struct {
	MessageID   string    `json:"message_id"`
	State       string    `json:"state"`
	Duplicate   *bool     `json:"duplicate"`
	ScheduledAt time.Time `json:"scheduled_at"`
}

PublishReceipt reports a stable message identity and persistence state.

type PublishReceipts added in v0.2.0 

type PublishReceipts []PublishReceipt

PublishReceipts contains one receipt per input message, in input order.

type Queue added in v0.2.0 

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

Queue is the import-first BlockQueue engine. It owns the supplied database driver from construction until shutdown.

func New 

func New(driver store.Driver, opt Options) *Queue

New constructs a queue that owns driver. Call Run before publishing and Shutdown when the application stops.

func NewChecked added in v0.3.1 

func NewChecked(driver store.Driver, opt Options) (*Queue, error)

NewChecked constructs a queue and returns metric registration failures. The caller retains ownership of driver when an error is returned.

func (*Queue) AckDelivery added in v0.2.0 

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

AckDelivery receipt-fences a successful lease transition to processed. Repeating the same successful receipt is idempotent.

func (*Queue) AckDeliveryTx added in v0.2.0 

func (q *Queue) AckDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, subscriber, messageID, receipt string) error

AckDeliveryTx atomically acknowledges a lease in a caller-owned transaction.

func (*Queue) BatchAckDeliveries added in v0.2.0 

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

BatchAckDeliveries acknowledges items in one set-based transaction and returns an outcome for every request.

func (*Queue) BatchNackDeliveries added in v0.2.0 

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

BatchNackDeliveries records failures in one set-based transaction and returns an outcome for every request.

func (*Queue) BatchPublish added in v0.2.0 

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

BatchPublish validates the entire batch and waits for one atomic commit.

func (*Queue) BatchPublishAsync added in v0.2.0 

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

BatchPublishAsync validates and admits an entire batch without waiting for its database commit.

func (*Queue) BatchPublishDurable added in v0.2.0 

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

BatchPublishDurable is the explicit durable alias for BatchPublish.

func (*Queue) BatchPublishTx added in v0.2.0 

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

BatchPublishTx validates the complete batch before writing it to tx.

func (*Queue) CancelClaimedDelivery added in v0.3.0 

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

CancelClaimedDelivery terminally cancels only the delivery lease identified by receipt. A stale worker cannot cancel a newer redelivery.

func (*Queue) CancelClaimedDeliveryTx added in v0.3.0 

func (q *Queue) CancelClaimedDeliveryTx(
	ctx context.Context,
	tx *sql.Tx,
	topic Topic,
	subscriber, messageID, receipt, reason string,
) error

CancelClaimedDeliveryTx is CancelClaimedDelivery within a caller-owned transaction.

func (*Queue) CancelDelivery added in v0.2.0 

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

CancelDelivery terminally cancels one subscriber delivery. Repeating a successful cancellation is idempotent.

func (*Queue) CancelDeliveryTx added in v0.2.0 

func (q *Queue) CancelDeliveryTx(ctx context.Context, tx *sql.Tx, topic Topic, subscriber, messageID, reason string) error

CancelDeliveryTx is CancelDelivery within a caller-owned transaction.

func (*Queue) CancelMessage added in v0.2.0 

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

CancelMessage cancels every pending or delivered subscriber delivery for a canonical message and returns the resulting state per subscriber.

func (*Queue) CancelMessageTx added in v0.2.0 

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

CancelMessageTx is CancelMessage within a caller-owned transaction.

func (*Queue) Claim added in v0.2.0 

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

Claim atomically leases visible deliveries in canonical priority order. Every redelivery receives a fresh receipt token.

func (*Queue) ClaimWait added in v0.2.0 

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

ClaimWait claims immediately available work or long-polls without a polling backoff. Its timer follows the earliest pending visibility or expired lease deadline stored in the database and also wakes on hints or ctx cancellation.

func (*Queue) Close added in v0.2.0 

func (q *Queue) Close()

Close shuts down with Options.ShutdownTimeout and discards the returned error. Use Shutdown when the caller must observe an incomplete drain.

func (*Queue) CreateSchedule added in v0.2.0 

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

CreateSchedule persists a recurring cron publish and computes its next run.

func (*Queue) CreateSubscribers added in v0.2.0 

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

CreateSubscribers atomically adds subscribers to an existing topic.

func (*Queue) CreateTopic added in v0.2.0 

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

CreateTopic atomically persists a topic and its initial subscribers.

func (*Queue) DeleteSchedule added in v0.2.0 

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

DeleteSchedule removes one recurring schedule.

func (*Queue) DeleteSubscriber added in v0.2.0 

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

DeleteSubscriber logically removes one subscriber after fencing earlier admissions. Physical deliveries are reclaimed asynchronously.

func (*Queue) DeleteTopic added in v0.2.0 

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

DeleteTopic logically removes a topic after fencing earlier admissions. Physical rows are reclaimed asynchronously in bounded chunks.

func (*Queue) DeliveryErrors added in v0.2.0 

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

DeliveryErrors returns append-only NACK and lease-expiry history newest first. The cursor is opaque and scoped to the selected delivery.

func (*Queue) ExtendLease added in v0.2.0 

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

ExtendLease moves an active receipt's expiry forward using database time.

func (*Queue) GetMessageStatus added in v0.2.0 

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

GetMessageStatus returns a canonical message and all of its delivery states.

func (*Queue) GetSchedule added in v0.2.0 

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

GetSchedule returns one schedule by ID within topic.

func (*Queue) GetSubscribersStatus added in v0.2.0 

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

GetSubscribersStatus returns all subscriber queue-depth summaries.

func (*Queue) GetTopic added in v0.2.0 

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

GetTopic reads one active topic from the immutable runtime registry.

func (*Queue) GetTopics added in v0.2.0 

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

GetTopics returns all topics matching filter. Prefer ListTopics for bounded operator-facing enumeration.

func (*Queue) ListDeliveries added in v0.2.0 

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

ListDeliveries returns a cursor page of active deliveries or, when deadLetter is true, dead-lettered deliveries.

func (*Queue) ListSchedules added in v0.2.0 

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

ListSchedules returns every schedule for topic. Prefer ListSchedulesPage for bounded operator-facing enumeration.

func (*Queue) ListSchedulesPage added in v0.3.0 

func (q *Queue) ListSchedulesPage(ctx context.Context, topic Topic, limit int, cursor string) (SchedulePage, error)

ListSchedulesPage returns a bounded cursor page ordered by name and ID.

func (*Queue) ListSubscriberStatuses added in v0.3.0 

func (q *Queue) ListSubscriberStatuses(ctx context.Context, topic Topic, limit int, cursor string) (SubscriberStatusPage, error)

ListSubscriberStatuses returns a bounded cursor page ordered by name and ID.

func (*Queue) ListTopics added in v0.3.0 

func (q *Queue) ListTopics(ctx context.Context, limit int, cursor string) (TopicPage, error)

ListTopics returns a bounded cursor page ordered by name and ID.

func (*Queue) Live added in v0.2.0 

func (q *Queue) Live() bool

Live reports whether the queue has not reached the stopped state.

func (*Queue) NackDelivery added in v0.2.0 

func (q *Queue) NackDelivery(ctx context.Context, topic Topic, subscriber, messageID, receipt string, retryDelay time.Duration, errorText string) error

NackDelivery records one failed attempt. A zero retryDelay applies the subscriber retry policy; a stale receipt returns ErrLeaseLost.

func (*Queue) NackDeliveryTx added in v0.2.0 

func (q *Queue) NackDeliveryTx(
	ctx context.Context,
	tx *sql.Tx,
	topic Topic,
	subscriber, messageID, receipt string,
	retryDelay time.Duration,
	errorText string,
) error

NackDeliveryTx atomically records a failed lease in a caller-owned transaction. A zero retryDelay selects the subscriber retry policy.

func (*Queue) PauseSchedule added in v0.2.0 

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

PauseSchedule changes whether future occurrences may be claimed.

func (*Queue) PauseSubscriber added in v0.2.0 

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

PauseSubscriber stops new claims for one subscriber while retaining work.

func (*Queue) PauseTopic added in v0.2.0 

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

PauseTopic stops new claims while continuing to accept publishes.

func (*Queue) Publish added in v0.2.0 

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

Publish is durable by default. A nil error means the canonical message and every subscriber delivery row committed successfully.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/yudhasubki/blockqueue"
	"github.com/yudhasubki/blockqueue/store/sqlite"
)

func main() {
	queue, topic := newExampleQueue()
	defer queue.Close()

	receipt, err := queue.Publish(context.Background(), topic, blockqueue.Message{
		Message:        `{"order_id":"order-1022"}`,
		IdempotencyKey: "order-1022",
	})
	if err != nil {
		panic(err)
	}

	fmt.Println(receipt.State, *receipt.Duplicate)
}

func newExampleQueue() (*blockqueue.Queue, blockqueue.Topic) {
	driver, err := sqlite.Open(":memory:", sqlite.Config{})
	if err != nil {
		panic(err)
	}
	queue := blockqueue.New(driver, blockqueue.Options{DisableMetrics: true})
	if err := queue.Run(context.Background()); err != nil {
		panic(err)
	}
	topic := blockqueue.NewTopic("orders")
	subscriber := blockqueue.NewSubscriber(topic, "fulfillment", blockqueue.SubscriberOptions{})
	if err := queue.CreateTopic(context.Background(), topic, blockqueue.Subscribers{subscriber}); err != nil {
		panic(err)
	}
	return queue, topic
}

Output:
persisted false

func (*Queue) PublishAsync added in v0.2.0 

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

PublishAsync returns the message identity at admission time. It is useful to Go callers that need the same receipt exposed by HTTP 202 responses.

func (*Queue) PublishDurable added in v0.2.0 

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

PublishDurable waits until the message and all subscriber delivery rows are committed. Duplicate is definitive in the returned receipt.

func (*Queue) PublishTx added in v0.2.0 

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

PublishTx stages a canonical message and its fan-out in caller-owned tx. The caller owns commit or rollback; staged rows are not claimable before a successful commit.

func (*Queue) Ready added in v0.2.0 

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

Ready checks lifecycle, persistence, maintenance health, and backlog limits.

func (*Queue) ReplayDeadLetters added in v0.2.0 

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

ReplayDeadLetters returns selected terminal deliveries to pending and resets their failure budget while retaining prior error history.

func (*Queue) ResumeSubscriber added in v0.2.0 

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

ResumeSubscriber allows claims and wakes the selected subscriber.

func (*Queue) ResumeTopic added in v0.2.0 

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

ResumeTopic allows claims and wakes waiting subscribers.

func (*Queue) Run added in v0.2.0 

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

Run validates durable state and builds the complete runtime snapshot before starting background workers.

func (*Queue) RunScheduleNow added in v0.2.0 

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

RunScheduleNow publishes an immediate occurrence. force bypasses overlap protection but preserves occurrence idempotency.

func (*Queue) ScheduleRunHistory added in v0.2.0 

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

ScheduleRunHistory returns newest occurrences first using an opaque cursor.

func (*Queue) Shutdown added in v0.2.0 

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

Shutdown stops admission, drains the writer, stops listeners and maintenance workers, performs the final checkpoint, and closes the database driver.

func (*Queue) SnoozeDelivery added in v0.2.0 

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

SnoozeDelivery returns an active lease to pending without recording a failure or consuming the subscriber's failure budget.

func (*Queue) SnoozeDeliveryTx added in v0.2.0 

func (q *Queue) SnoozeDeliveryTx(
	ctx context.Context,
	tx *sql.Tx,
	topic Topic,
	subscriber, messageID, receipt string,
	delay time.Duration,
) (time.Time, error)

SnoozeDeliveryTx is SnoozeDelivery within a caller-owned transaction.

func (*Queue) State added in v0.2.0 

func (q *Queue) State() LifecycleState

State returns the queue's current lifecycle state.

func (*Queue) UpdateSchedule added in v0.2.0 

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

UpdateSchedule replaces a schedule when expectedVersion matches.

func (*Queue) WithTx added in v0.2.0 

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

WithTx runs fn in a transaction owned by the queue. It is the preferred way to atomically change application tables and publish or complete deliveries: the queue can notify local waiters only after commit succeeds. A context cancellation or deadline reported by Commit is outcome-unknown because the server may have committed before the client stopped waiting; fn is never retried.

Example 
package main

import (
	"context"
	"database/sql"
	"fmt"

	"github.com/yudhasubki/blockqueue"
	"github.com/yudhasubki/blockqueue/store/sqlite"
)

func main() {
	queue, topic := newExampleQueue()
	defer queue.Close()

	ctx := context.Background()
	var publishState string
	err := queue.WithTx(ctx, nil, func(tx *sql.Tx) error {
		if _, err := tx.ExecContext(ctx, `
			CREATE TABLE orders (id TEXT PRIMARY KEY, status TEXT NOT NULL)
		`); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx,
			"INSERT INTO orders (id, status) VALUES (?, ?)", "order-1022", "pending"); err != nil {
			return err
		}
		receipt, err := queue.PublishTx(ctx, tx, topic, blockqueue.Message{
			Message:        `{"order_id":"order-1022"}`,
			IdempotencyKey: "fulfill-order-1022",
		})
		publishState = receipt.State
		return err
	})

	fmt.Println(publishState, err == nil)
}

func newExampleQueue() (*blockqueue.Queue, blockqueue.Topic) {
	driver, err := sqlite.Open(":memory:", sqlite.Config{})
	if err != nil {
		panic(err)
	}
	queue := blockqueue.New(driver, blockqueue.Options{DisableMetrics: true})
	if err := queue.Run(context.Background()); err != nil {
		panic(err)
	}
	topic := blockqueue.NewTopic("orders")
	subscriber := blockqueue.NewSubscriber(topic, "fulfillment", blockqueue.SubscriberOptions{})
	if err := queue.CreateTopic(context.Background(), topic, blockqueue.Subscribers{subscriber}); err != nil {
		panic(err)
	}
	return queue, topic
}

Output:
staged true

func (*Queue) WriterHealthy added in v0.2.0 

func (q *Queue) WriterHealthy() bool

WriterHealthy reports whether persistence is currently accepting progress.

type RetryPolicy added in v0.2.0 

type RetryPolicy struct {
	InitialDelay  string  `json:"initial_delay,omitempty"`
	MaxDelay      string  `json:"max_delay,omitempty"`
	Multiplier    float64 `json:"multiplier,omitempty"`
	Jitter        float64 `json:"jitter,omitempty"`
	DisableJitter bool    `json:"disable_jitter,omitempty"`
}

RetryPolicy controls the delay applied after a NACK or expired lease. Empty fields use the documented exponential-backoff defaults.

type Schedule added in v0.2.0 

type Schedule struct {
	ID             string         `db:"id" json:"id"`
	TopicID        string         `db:"topic_id" json:"topic_id"`
	Name           string         `db:"name" json:"name"`
	CronExpression string         `db:"cron_expression" json:"cron"`
	Timezone       string         `db:"timezone" json:"timezone"`
	Message        string         `db:"message" json:"message"`
	Headers        string         `db:"headers" json:"-"`
	CorrelationID  sql.NullString `db:"correlation_id" json:"-"`
	Priority       int            `db:"priority" json:"priority"`
	MisfirePolicy  string         `db:"misfire_policy" json:"misfire_policy"`
	OverlapPolicy  string         `db:"overlap_policy" json:"overlap_policy"`
	Paused         bool           `db:"paused" json:"paused"`
	Version        int            `db:"version" json:"version"`
	NextRunAt      time.Time      `db:"next_run_at" json:"next_run_at"`
	OwnerID        sql.NullString `db:"owner_id" json:"-"`
	LeaseExpiresAt sql.NullTime   `db:"lease_expires_at" json:"-"`
	FencingToken   int64          `db:"fencing_token" json:"-"`
	CreatedAt      time.Time      `db:"created_at" json:"created_at"`
	UpdatedAt      time.Time      `db:"updated_at" json:"updated_at"`
	// contains filtered or unexported fields
}

Schedule is the persisted scheduler definition and its next occurrence.

func (Schedule) MarshalJSON added in v0.2.0 

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

MarshalJSON exposes decoded headers and optional correlation data.

func (Schedule) PublicHeaders added in v0.2.0 

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

PublicHeaders decodes the schedule's persisted JSON headers.

type ScheduleInput added in v0.2.0 

type ScheduleInput struct {
	Name           string            `json:"name"`
	CronExpression string            `json:"cron"`
	Timezone       string            `json:"timezone,omitempty"`
	Message        string            `json:"message"`
	Headers        map[string]string `json:"headers,omitempty"`
	CorrelationID  string            `json:"correlation_id,omitempty"`
	Priority       int               `json:"priority,omitempty"`
	MisfirePolicy  string            `json:"misfire_policy,omitempty"`
	OverlapPolicy  string            `json:"overlap_policy,omitempty"`
}

ScheduleInput defines a recurring five-field cron publish.

type SchedulePage added in v0.3.0 

type SchedulePage struct {
	Schedules  []Schedule `json:"schedules"`
	NextCursor string     `json:"next_cursor,omitempty"`
}

SchedulePage is one cursor-paginated page of schedules.

type ScheduleRun added in v0.2.0 

type ScheduleRun struct {
	ID           string         `db:"id" json:"id"`
	ScheduleID   string         `db:"schedule_id" json:"schedule_id"`
	MessageID    sql.NullString `db:"message_id" json:"-"`
	ScheduledFor time.Time      `db:"scheduled_for" json:"scheduled_for"`
	StartedAt    time.Time      `db:"started_at" json:"started_at"`
	FinishedAt   sql.NullTime   `db:"finished_at" json:"-"`
	Status       string         `db:"status" json:"status"`
	Error        sql.NullString `db:"error" json:"-"`
	CreatedAt    time.Time      `db:"created_at" json:"created_at"`
}

ScheduleRun records one scheduled occurrence and its terminal outcome.

func (ScheduleRun) MarshalJSON added in v0.2.0 

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

MarshalJSON exposes nullable run fields as optional JSON strings.

type ScheduleRunPage added in v0.2.0 

type ScheduleRunPage struct {
	Runs       []ScheduleRun `json:"runs"`
	NextCursor string        `json:"next_cursor,omitempty"`
}

ScheduleRunPage is one cursor-paginated page of occurrence history.

type Subscriber added in v0.2.0 

type Subscriber struct {
	ID        uuid.UUID         `db:"id" json:"id"`
	TopicID   uuid.UUID         `db:"topic_id" json:"topic_id"`
	TopicName string            `db:"topic_name" json:"topic_name,omitempty"`
	Name      string            `db:"name" json:"name"`
	Options   SubscriberOptions `db:"option" json:"options"`
	Paused    bool              `db:"paused" json:"paused"`
	CreatedAt time.Time         `db:"created_at" json:"created_at"`
	DeletedAt *time.Time        `db:"deleted_at" json:"deleted_at,omitempty"`
}

Subscriber defines one independently leased delivery stream for a topic.

func NewSubscriber added in v0.2.0 

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

NewSubscriber creates a normalized subscriber value with a new UUID.

type SubscriberOptions added in v0.2.0 

type SubscriberOptions struct {
	MaxAttempts        int         `json:"max_attempts"`
	VisibilityDuration string      `json:"visibility_duration"`
	DequeueBatchSize   int         `json:"dequeue_batch_size,omitempty"`
	RetryPolicy        RetryPolicy `json:"retry_policy,omitempty"`
}

SubscriberOptions controls retries, lease visibility, and claim batching. Zero fields are normalized to safe defaults when the subscriber is created.

func (*SubscriberOptions) Scan added in v0.2.0 

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

Scan implements sql.Scanner for a persisted JSON options value.

func (SubscriberOptions) Value added in v0.2.0 

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

Value implements driver.Valuer using the normalized JSON representation.

type SubscriberQueueStats 

type SubscriberQueueStats struct {
	Pending   int `db:"pending"`
	Delivered int `db:"delivered"`
}

SubscriberQueueStats contains persisted pending and leased delivery counts.

type SubscriberStatus added in v0.2.0 

type SubscriberStatus struct {
	TopicID            uuid.UUID `json:"topic_id"`
	Name               string    `json:"name"`
	UnpublishedMessage int       `json:"unpublished_message"`
	UnackedMessage     int       `json:"unacked_message"`
}

SubscriberStatus summarizes pending and currently leased work.

type SubscriberStatusPage added in v0.3.0 

type SubscriberStatusPage struct {
	Subscribers SubscriberStatuses `json:"subscribers"`
	NextCursor  string             `json:"next_cursor,omitempty"`
}

SubscriberStatusPage is one cursor-paginated subscriber status page.

type SubscriberStatuses added in v0.2.0 

type SubscriberStatuses []SubscriberStatus

SubscriberStatuses is a collection of subscriber status summaries.

type Subscribers added in v0.2.0 

type Subscribers []Subscriber

Subscribers is a collection of topic subscribers.

type Topic added in v0.2.0 

type Topic struct {
	ID        uuid.UUID  `db:"id" json:"id"`
	Name      string     `db:"name" json:"name"`
	Paused    bool       `db:"paused" json:"paused"`
	CreatedAt time.Time  `db:"created_at" json:"created_at"`
	DeletedAt *time.Time `db:"deleted_at" json:"deleted_at,omitempty"`
}

Topic identifies a fan-out stream. Names are unique among active topics.

func NewTopic added in v0.2.0 

func NewTopic(name string) Topic

NewTopic creates a topic value with a new UUID.

type TopicFilter added in v0.2.0 

type TopicFilter struct {
	Names       []string
	WithDeleted bool
}

TopicFilter selects topics for GetTopics.

type TopicPage added in v0.3.0 

type TopicPage struct {
	Topics     Topics `json:"topics"`
	NextCursor string `json:"next_cursor,omitempty"`
}

TopicPage is one cursor-paginated page of topics.

type Topics added in v0.2.0 

type Topics []Topic

Topics is a collection of queue topics.

type TransactionCommitUnknownError added in v0.3.0 

type TransactionCommitUnknownError struct {
	Cause error
}

TransactionCommitUnknownError reports that a caller-owned transaction may have committed even though Commit returned an error.

func (*TransactionCommitUnknownError) Error added in v0.3.0 

func (err *TransactionCommitUnknownError) Error() string

func (*TransactionCommitUnknownError) Is added in v0.3.0 

func (err *TransactionCommitUnknownError) Is(target error) bool

func (*TransactionCommitUnknownError) Unwrap added in v0.3.0 

func (err *TransactionCommitUnknownError) Unwrap() error

type WriterOptions added in v0.2.0 

type WriterOptions struct {
	BatchSize          int
	FlushInterval      time.Duration
	MaxPendingMessages int64
	MaxPendingBytes    int64
	RetryMin           time.Duration
	RetryMax           time.Duration
}

WriterOptions controls batching and the weighted pending budget.

func DefaultWriterOptions added in v0.2.0 

func DefaultWriterOptions() WriterOptions

DefaultWriterOptions returns the bounded production writer defaults.

Source Files

View all Source files

Directories

Path Synopsis
cmd
blockqueue command
example
basic command
Package main demonstrates using BlockQueue as an imported Go library.
 Package main demonstrates using BlockQueue as an imported Go library.
transactional command
Package main demonstrates transactional enqueueing and transactional delivery completion with BlockQueue and an application table in one SQLite database.
 Package main demonstrates transactional enqueueing and transactional delivery completion with BlockQueue and an application table in one SQLite database.
worker command
Package main demonstrates the typed BlockQueue worker runtime.
 Package main demonstrates the typed BlockQueue worker runtime.
Package httpapi exposes BlockQueue as a versioned, cross-language HTTP API.
 Package httpapi exposes BlockQueue as a versioned, cross-language HTTP API.
internal
headercodec
Package headercodec owns the persisted JSON representation of message headers.
 Package headercodec owns the persisted JSON representation of message headers.
persistence
Package persistence owns BlockQueue's durable SQL state and backend dialect differences.
 Package persistence owns BlockQueue's durable SQL state and backend dialect differences.
subscriberconfig
Package subscriberconfig owns subscriber defaults and validation shared by the runtime registry and persistence layer.
 Package subscriberconfig owns subscriber defaults and validation shared by the runtime registry and persistence layer.
testdb
Package testdb provides guarded, schema-isolated PostgreSQL databases for integration tests and benchmarks.
 Package testdb provides guarded, schema-isolated PostgreSQL databases for integration tests and benchmarks.
textlimit
Package textlimit provides UTF-8-safe bounds for diagnostic text stored by BlockQueue.
 Package textlimit provides UTF-8-safe bounds for diagnostic text stored by BlockQueue.
pkg
metric
Package metric contains the Prometheus collectors used by BlockQueue's queue and worker runtimes.
 Package metric contains the Prometheus collectors used by BlockQueue's queue and worker runtimes.
Package store defines the small database boundary used by BlockQueue.
 Package store defines the small database boundary used by BlockQueue.
postgres
Package postgres provides the production PostgreSQL storage driver.
 Package postgres provides the production PostgreSQL storage driver.
sqlite
Package sqlite provides the production SQLite storage driver.
 Package sqlite provides the production SQLite storage driver.
turso
Package turso provides experimental libSQL/Turso storage support.
 Package turso provides experimental libSQL/Turso storage support.
Package worker provides a bounded, lease-aware consumer runtime for BlockQueue.
 Package worker provides a bounded, lease-aware consumer runtime for BlockQueue.

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.