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