Help us shape the future of Go! Take the 2026 Go Developer Survey

persistence

package
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: 29 Imported by: 0

Documentation

Overview

Package persistence owns BlockQueue's durable SQL state and backend dialect differences. The public blockqueue package remains the runtime/API facade; database drivers are supplied through the public store package.

Index

Constants

View Source
const (
	MaximumDeliveryLease = subscriberconfig.MaximumDeliveryLease
	MaxDeliveryTextBytes = 16 << 10
)
View Source
const (
	DeliveryStatusPending    = "pending"
	DeliveryStatusDelivered  = "delivered"
	DeliveryStatusProcessed  = "processed"
	DeliveryStatusDeadLetter = "dead_letter"
	DeliveryStatusCancelled  = "cancelled"
)

Delivery states are persisted values protected by schema constraints. Keep them centralized for Go-side comparisons and assignments. SQL predicates intentionally retain literals so both backends can match partial indexes.

View Source
const (
	ScheduleRunStatusRunning   = "running"
	ScheduleRunStatusCompleted = "completed"
	ScheduleRunStatusSkipped   = "skipped"
	ScheduleRunStatusFailed    = "failed"
)
View Source
const (
	ScheduleMisfirePolicyFireOnce = "fire_once"
	ScheduleOverlapPolicySkip     = "skip"
)
View Source
const (
	EventChannel        = "blockqueue_events"
	EventTopology       = "topology"
	EventScheduler      = "scheduler"
	EventDeliveryPrefix = "delivery:"
)

PostgreSQL notifications are wake-up hints. These values form the internal event protocol shared by persistence and the queue listener.

Variables

View Source
var (
	ErrTopicNotFound      = errors.New("topic not found")
	ErrNoActiveSubscriber = errors.New("topic has no active subscriber")
	ErrInvalidPublish     = errors.New("invalid publish request")
	ErrInvalidCursor      = errors.New("invalid pagination cursor")
	ErrResourceConflict   = errors.New("resource already exists")
	ErrSubscriberNotFound = errors.New("subscriber not found")

	ErrLeaseLost        = errors.New("delivery lease lost")
	ErrDeliveryNotFound = errors.New("delivery not found")
	ErrInvalidReceipt   = errors.New("receipt_token is required")
	ErrDeliveryTerminal = errors.New("delivery is already terminal")

	ErrIdempotencyConflict = errors.New("idempotency key conflicts with a different message")
	ErrWriterClosed        = errors.New("writer closed")

	ErrScheduleNotFound  = errors.New("schedule not found")
	ErrScheduleVersion   = errors.New("stale schedule version")
	ErrScheduleOverlap   = errors.New("previous schedule run is still active")
	ErrScheduleLeaseLost = errors.New("schedule lease lost")

	ErrUnsupportedDialect = errors.New("unsupported database dialect")
	ErrMigrationChecksum  = errors.New("migration checksum mismatch")
)

Functions

func Migrate

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

Migrate installs the v0.2 schema and applies future ordered migrations. v0.2 is a clean schema break: callers must provide a new database rather than a database created by v0.1.

Types

type BatchAckItem

type BatchAckItem struct {
	MessageID    string
	ReceiptToken string
}

type BatchNackItem

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

type CancellationRow

type CancellationRow = cancellationRow

type CheckpointMode

type CheckpointMode = sqliteCheckpointMode
const (
	CheckpointPassive  CheckpointMode = sqliteCheckpointPassive
	CheckpointTruncate CheckpointMode = sqliteCheckpointTruncate
)

type CheckpointResult

type CheckpointResult = sqliteCheckpointResult

type DeliveryError

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

type DeliveryRow

type DeliveryRow = deliveryRow

type MessageDeliveryStatus

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

type MessageStatus

type MessageStatus struct {
	ID             string
	TopicID        string
	Message        string
	Headers        map[string]string
	CorrelationID  string
	IdempotencyKey string
	Priority       int
	ScheduledAt    time.Time
	CreatedAt      time.Time
	Deliveries     []MessageDeliveryStatus
}

type PersistWriteResult

type PersistWriteResult struct {
	Duplicates  []bool
	ScheduledAt []time.Time
}

PersistWriteResult describes the storage-resolved identity of each request in input order. ScheduledAt is calculated from the database clock for immediate and relative-delay publishes, and is authoritative even when the caller owns the surrounding transaction and has not committed it yet.

type RetryPolicy

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"`
}

type Schedule

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

type ScheduleRun

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

type Store

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

Store is the queue engine's concrete persistence boundary. It deliberately avoids a broad repository interface: SQLite and PostgreSQL share the same transactional implementation, with only genuine SQL differences delegated to the internal dialect strategy.

func New

func New(driver store.Driver) *Store

func (*Store) AckDelivery

func (store *Store) AckDelivery(ctx context.Context, subscriberID uuid.UUID, messageID, receipt string) error

func (*Store) AckDeliveryWithTx

func (store *Store) AckDeliveryWithTx(ctx context.Context, tx *sql.Tx, subscriberID uuid.UUID, messageID, receipt string) error

func (*Store) BatchAckDeliveries

func (store *Store) BatchAckDeliveries(ctx context.Context, subscriberID uuid.UUID, requests []BatchAckItem) ([]error, error)

func (*Store) BatchNackDeliveries

func (store *Store) BatchNackDeliveries(ctx context.Context, subscriberID uuid.UUID, requests []BatchNackItem) ([]bool, []error, error)

func (*Store) BatchReplayDeadLetters

func (store *Store) BatchReplayDeadLetters(ctx context.Context, subscriberID uuid.UUID, messageIDs []string) ([]bool, error)

func (*Store) CancelClaimedDeliveryWithTx

func (store *Store) CancelClaimedDeliveryWithTx(ctx context.Context, tx *sql.Tx, subscriberID uuid.UUID, messageID, receipt, reason string) error

func (*Store) CancelDeliveryWithTx

func (store *Store) CancelDeliveryWithTx(ctx context.Context, tx *sql.Tx, subscriberID uuid.UUID, messageID, reason string) (string, error)

func (*Store) CancelMessageWithTx

func (store *Store) CancelMessageWithTx(ctx context.Context, tx *sql.Tx, topicID uuid.UUID, messageID, reason string) ([]CancellationRow, error)

func (*Store) CheckpointSQLite

func (store *Store) CheckpointSQLite(ctx context.Context, mode CheckpointMode) (CheckpointResult, error)

func (*Store) ClaimDeliveries

func (store *Store) ClaimDeliveries(ctx context.Context, subscriberID uuid.UUID, limit int, lease time.Duration) ([]DeliveryRow, error)

func (*Store) ClaimDueSchedule

func (store *Store) ClaimDueSchedule(ctx context.Context, owner string, now time.Time, lease time.Duration) (Schedule, bool, error)

func (*Store) Close

func (store *Store) Close() error

func (*Store) Conn

func (store *Store) Conn() *sqlx.DB

func (*Store) CreateSchedule

func (store *Store) CreateSchedule(ctx context.Context, schedule Schedule) error

func (*Store) CreateSubscribers

func (store *Store) CreateSubscribers(ctx context.Context, subscribers Subscribers) error

func (*Store) CreateTopic

func (store *Store) CreateTopic(ctx context.Context, topic Topic, subscribers Subscribers) error

func (*Store) DatabaseNow

func (store *Store) DatabaseNow(ctx context.Context) (time.Time, error)

func (*Store) DeleteSchedule

func (store *Store) DeleteSchedule(ctx context.Context, topicID uuid.UUID, scheduleID string) error

func (*Store) DeleteSubscriber

func (store *Store) DeleteSubscriber(ctx context.Context, topicID, subscriberID uuid.UUID, name string) error

func (*Store) DeleteTopic

func (store *Store) DeleteTopic(ctx context.Context, topicID uuid.UUID) error

func (*Store) DialectError

func (store *Store) DialectError() error

func (*Store) ExtendDeliveryLease

func (store *Store) ExtendDeliveryLease(ctx context.Context, subscriberID uuid.UUID, messageID, receipt string, extension time.Duration) (time.Time, error)

func (*Store) FailScheduleOccurrence

func (store *Store) FailScheduleOccurrence(ctx context.Context, claimed Schedule, scheduledFor, nextRunAt time.Time, advance bool, owner, failure string) (ScheduleRun, error)

func (*Store) GetMessageStatus

func (store *Store) GetMessageStatus(ctx context.Context, topicID uuid.UUID, messageID string) (MessageStatus, error)

func (*Store) GetSchedule

func (store *Store) GetSchedule(ctx context.Context, topicID uuid.UUID, scheduleID string) (Schedule, error)

func (*Store) GetSubscribers

func (store *Store) GetSubscribers(ctx context.Context, filter SubscriberFilter) (Subscribers, error)

func (*Store) GetTopics

func (store *Store) GetTopics(ctx context.Context, filter TopicFilter) (Topics, error)

func (*Store) HasDeletedTopology

func (store *Store) HasDeletedTopology(ctx context.Context) (bool, error)

func (*Store) IncrementalVacuum

func (store *Store) IncrementalVacuum(ctx context.Context) error

func (*Store) ListDeliveries

func (store *Store) ListDeliveries(ctx context.Context, subscriberID uuid.UUID, deadLetter bool, limit int, cursor string) ([]DeliveryRow, error)

func (*Store) ListDeliveryErrors

func (store *Store) ListDeliveryErrors(ctx context.Context, subscriberID uuid.UUID, messageID string, limit int, cursor string) ([]DeliveryError, error)

func (*Store) ListScheduleRuns

func (store *Store) ListScheduleRuns(ctx context.Context, scheduleID string, limit int, before time.Time, beforeID string) ([]ScheduleRun, error)

func (*Store) ListSchedules

func (store *Store) ListSchedules(ctx context.Context, topicID uuid.UUID) ([]Schedule, error)

func (*Store) ListSchedulesPage

func (store *Store) ListSchedulesPage(ctx context.Context, topicID uuid.UUID, limit int, afterName, afterID string) ([]Schedule, error)

func (*Store) ListSubscriberStatuses

func (store *Store) ListSubscriberStatuses(ctx context.Context, topicID uuid.UUID, limit int, afterName, afterID string) ([]SubscriberStatusRow, error)

func (*Store) ListTopics

func (store *Store) ListTopics(ctx context.Context, limit int, afterName, afterID string) (Topics, error)

func (*Store) NackDelivery

func (store *Store) NackDelivery(ctx context.Context, subscriberID uuid.UUID, messageID, receipt string, delay time.Duration, errorText string) (bool, error)

func (*Store) NackDeliveryWithTx

func (store *Store) NackDeliveryWithTx(ctx context.Context, tx *sql.Tx, subscriberID uuid.UUID, messageID, receipt string, delay time.Duration, errorText string) (bool, error)

func (*Store) NextDeliveryWake

func (store *Store) NextDeliveryWake(ctx context.Context, subscriberID uuid.UUID) (time.Time, time.Time, bool, error)

func (*Store) NextLeaseExpiry

func (store *Store) NextLeaseExpiry(ctx context.Context) (time.Time, time.Time, bool, error)

func (*Store) NextScheduleDue

func (store *Store) NextScheduleDue(ctx context.Context, now time.Time) (time.Time, time.Time, bool, error)

func (*Store) PersistScheduleOccurrence

func (store *Store) PersistScheduleOccurrence(ctx context.Context, claimed Schedule, scheduledFor, nextRunAt time.Time, force, advance bool, owner string) (ScheduleRun, error)

func (*Store) PersistWriteRequests

func (store *Store) PersistWriteRequests(ctx context.Context, requests []WriteRequest) (PersistWriteResult, error)

func (*Store) PersistWriteRequestsWithTx

func (store *Store) PersistWriteRequestsWithTx(ctx context.Context, tx *sql.Tx, requests []WriteRequest) (PersistWriteResult, error)

func (*Store) PruneDeadLetters

func (store *Store) PruneDeadLetters(ctx context.Context, retention time.Duration) error

func (*Store) PruneDeletedTopology

func (store *Store) PruneDeletedTopology(ctx context.Context, budget time.Duration) (int64, bool, bool, error)

func (*Store) PruneProcessedMessages

func (store *Store) PruneProcessedMessages(ctx context.Context, retention time.Duration) error

func (*Store) PruneScheduleRuns

func (store *Store) PruneScheduleRuns(ctx context.Context, retention time.Duration) error

func (*Store) ReapExpiredDeliveries

func (store *Store) ReapExpiredDeliveries(ctx context.Context, limit int) (int64, error)

func (*Store) ReplayDeadLetter

func (store *Store) ReplayDeadLetter(ctx context.Context, subscriberID uuid.UUID, messageID string) (bool, error)

func (*Store) ScheduleNameExists

func (store *Store) ScheduleNameExists(ctx context.Context, topicID uuid.UUID, name string) (bool, error)

func (*Store) SetSchedulePaused

func (store *Store) SetSchedulePaused(ctx context.Context, topicID uuid.UUID, scheduleID string, paused bool) error

func (*Store) SetSubscriberPaused

func (store *Store) SetSubscriberPaused(ctx context.Context, subscriberID uuid.UUID, paused bool) error

func (*Store) SetTopicPaused

func (store *Store) SetTopicPaused(ctx context.Context, topicID uuid.UUID, paused bool) error

func (*Store) SnoozeDeliveryWithTx

func (store *Store) SnoozeDeliveryWithTx(ctx context.Context, tx *sql.Tx, subscriberID uuid.UUID, messageID, receipt string, delay time.Duration) (time.Time, error)

func (*Store) StatementCacheLen

func (store *Store) StatementCacheLen() int

func (*Store) SupportsSQLiteMaintenance

func (store *Store) SupportsSQLiteMaintenance() bool

func (*Store) TopicSubscriberQueueStats

func (store *Store) TopicSubscriberQueueStats(ctx context.Context, topicID uuid.UUID) (map[uuid.UUID]SubscriberQueueStats, error)

func (*Store) TryMaintenanceLeadership

func (store *Store) TryMaintenanceLeadership(ctx context.Context) (bool, func() error, error)

func (*Store) UpdateSchedule

func (store *Store) UpdateSchedule(ctx context.Context, topicID uuid.UUID, scheduleID string, expectedVersion int, schedule Schedule) error

type Subscriber

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

type SubscriberFilter

type SubscriberFilter struct {
	TopicIDs    []uuid.UUID
	Names       []string
	WithDeleted bool
}

type SubscriberOptions

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"`
}

func (*SubscriberOptions) Scan

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

func (SubscriberOptions) Value

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

type SubscriberQueueStats

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

type SubscriberStatusRow

type SubscriberStatusRow struct {
	ID        string `db:"id"`
	Name      string `db:"name"`
	Pending   int    `db:"pending"`
	Delivered int    `db:"delivered"`
}

type Subscribers

type Subscribers []Subscriber

type Topic

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

type TopicFilter

type TopicFilter struct {
	Names       []string
	WithDeleted bool
}

type Topics

type Topics []Topic

type WriteRequest

type WriteRequest struct {
	TopicID         uuid.UUID
	MessageID       string
	Message         string
	Headers         []byte
	CorrelationID   string
	IdempotencyKey  string
	IdempotencyHash string
	ScheduleMode    string
	ScheduleDelay   time.Duration
	Priority        int
	VisibleAt       time.Time
	CreatedAt       time.Time
}

Jump to

Keyboard shortcuts

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