postgres

package
v0.20.0 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DialectPostgres = sqlutil.DialectPostgres
	DialectSQLite   = sqlutil.DialectSQLite
)

Variables

View Source 
var ErrScheduleConflict = errors.New("recurring schedule definition conflicts")

ErrScheduleConflict means an immutable schedule ID was reused with a changed definition. Replace a schedule using a fresh ID and explicitly pause the old one.

Functions

func OccurrenceID added in v0.20.0 

func OccurrenceID(namespace, scheduleID string, at time.Time) string

OccurrenceID is stable across restarts, clock zones and pollers. Schedules are immutable, so namespace, schedule ID and planned instant fully identify a run.

Types

type Adapter 

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

Adapter bridges the storage implementation to queue interfaces.

func NewAdapter 

func NewAdapter(storage queue.Storage) *Adapter

NewAdapter builds a queue adapter from a storage implementation.

func (*Adapter) Dequeue 

func (a *Adapter) Dequeue(ctx context.Context) (queue.Delivery, error)

Dequeue returns a delivery wrapper when available.

func (*Adapter) Enqueue 

func (a *Adapter) Enqueue(ctx context.Context, msg *job.ExecutionMessage) (queue.EnqueueReceipt, error)

Enqueue forwards to the underlying storage.

func (*Adapter) EnqueueAfter added in v0.15.0 

func (a *Adapter) EnqueueAfter(ctx context.Context, msg *job.ExecutionMessage, delay time.Duration) (queue.EnqueueReceipt, error)

EnqueueAfter forwards delayed enqueue requests to storage when supported.

func (*Adapter) EnqueueAt added in v0.15.0 

func (a *Adapter) EnqueueAt(ctx context.Context, msg *job.ExecutionMessage, at time.Time) (queue.EnqueueReceipt, error)

EnqueueAt forwards scheduled enqueue requests to storage when supported.

func (*Adapter) GetDispatchStatus added in v0.16.0 

func (a *Adapter) GetDispatchStatus(ctx context.Context, dispatchID string) (queue.DispatchStatus, error)

GetDispatchStatus probes the underlying storage for dispatch lifecycle state.

type Dialect 

type Dialect = sqlutil.Dialect

Dialect controls SQL placeholder formatting.

type Occurrence added in v0.20.0 

type Occurrence struct {
	ID, Namespace, ScheduleID, AdmissionID string
	ScheduledAt                            time.Time
}

Occurrence is a durable admission receipt, independent of queue receipt expiry. AdmissionID identifies the queue dispatch or the application record published by an optional transaction-local publisher. It is not an execution receipt.

type OccurrencePublisher added in v0.20.0 

type OccurrencePublisher func(context.Context, *sql.Tx, Occurrence, *job.ExecutionMessage) (string, error)

OccurrencePublisher must publish on the supplied transaction and return a nonempty durable admission ID. It must not commit, open nested transactions or perform external I/O. Any error requires rollback by the caller of AdmitTx. This allows native authorized application acceptance instead of a second job.

type Option 

type Option func(*Storage)

Option configures the postgres storage adapter.

func WithClock 

func WithClock(now func() time.Time) Option

WithClock overrides the time source.

func WithDLQTableName 

func WithDLQTableName(name string) Option

WithDLQTableName sets the DLQ table name.

func WithDialect 

func WithDialect(dialect Dialect) Option

WithDialect sets the SQL placeholder style.

func WithIDFunc 

func WithIDFunc(fn func() string) Option

WithIDFunc overrides ID generation.

func WithStatusTTL added in v0.17.0 

func WithStatusTTL(ttl time.Duration) Option

WithStatusTTL sets dispatch status retention TTL.

func WithStatusTableName added in v0.17.0 

func WithStatusTableName(name string) Option

WithStatusTableName sets the dispatch status table name.

func WithTableName 

func WithTableName(name string) Option

WithTableName sets the primary table name.

func WithTokenFunc 

func WithTokenFunc(fn func() string) Option

WithTokenFunc overrides token generation.

func WithUseSkipLocked 

func WithUseSkipLocked(enabled bool) Option

WithUseSkipLocked toggles SKIP LOCKED usage in dequeue.

func WithVisibilityTimeout 

func WithVisibilityTimeout(timeout time.Duration) Option

WithVisibilityTimeout sets the lease timeout.

type RecurringSchedule added in v0.20.0 

type RecurringSchedule struct {
	Namespace, ID, Expression, TimeZone string
	StartAfter                          time.Time
	Message                             job.ExecutionMessage
}

RecurringSchedule identifies one immutable recurrence stream. StartAfter is exclusive; catch-up includes occurrences exactly equal to the supplied cutoff. Namespace must come from trusted application policy, not an untrusted request. TimeZone is an explicit IANA name (including UTC); inline TZ directives are not accepted. Message is the native command template, without a recurrence parameter.

type RecurringScheduler added in v0.20.0 

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

RecurringScheduler materializes due cron occurrences through the native SQL queue. It owns no goroutines, process-local locks or leadership state. Pollers discover due schedule IDs via Due and call Admit with a bounded catch-up limit. All pollers must share the same database, schemas and retained occurrence ledger.

func NewRecurringScheduler added in v0.20.0 

func NewRecurringScheduler(storage *Storage) (*RecurringScheduler, error)

func (*RecurringScheduler) Admit added in v0.20.0 

func (r *RecurringScheduler) Admit(ctx context.Context, namespace, id string, through time.Time, limit int) ([]Occurrence, error)

Admit commits at most limit due occurrences, in chronological order, together with their native queue messages, statuses and cursor. Retry after any error; a lost commit response converges on the persisted cursor/ledger. Returned rows describe newly committed admission, not completion of worker side effects.

func (*RecurringScheduler) AdmitTx added in v0.20.0 

func (r *RecurringScheduler) AdmitTx(ctx context.Context, tx *sql.Tx, namespace, id string, through time.Time, limit int, publish OccurrencePublisher) ([]Occurrence, error)

AdmitTx uses a caller-owned transaction on the queue's exact database. The caller must roll back on error and must commit before exposing the receipts. A nonnil publisher replaces queue enqueue and must obey OccurrencePublisher. Trusted authorization/fencing can wrap this method and all application writes in the same transaction; no persisted actor hint is treated as authentication.

func (*RecurringScheduler) Create added in v0.20.0 

func (r *RecurringScheduler) Create(ctx context.Context, d RecurringSchedule) error

Create atomically installs a definition, or accepts an exact replay without rewinding the cursor or changing pause state. There is deliberately no delete or in-place edit API: deleting identity/ledger rows can repeat acknowledged work.

func (*RecurringScheduler) CreateTx added in v0.20.0 

func (r *RecurringScheduler) CreateTx(ctx context.Context, tx *sql.Tx, d RecurringSchedule) error

CreateTx installs or replays an immutable definition on the caller-owned queue database transaction. The caller must roll back on error and commit on success. Use this to fence host authorization and delegation persistence with creation.

func (*RecurringScheduler) Due added in v0.20.0 

func (r *RecurringScheduler) Due(ctx context.Context, namespace string, through time.Time, limit int) ([]string, error)

Due returns bounded schedule IDs ordered by oldest backlog. Callers control polling and fairness; a cutoff is a trusted scheduler clock, never actor input.

func (*RecurringScheduler) Migrate added in v0.20.0 

func (r *RecurringScheduler) Migrate(ctx context.Context) error

Migrate creates additive recurrence tables. Migrate the queue separately first.

func (*RecurringScheduler) SetPaused added in v0.20.0 

func (r *RecurringScheduler) SetPaused(ctx context.Context, namespace, id string, paused bool) error

SetPaused serializes with admission. Already committed queue work is not canceled. Resume catches up from the retained cursor, including paused times.

func (*RecurringScheduler) SetPausedTx added in v0.20.0 

func (r *RecurringScheduler) SetPausedTx(ctx context.Context, tx *sql.Tx, namespace, id string, paused bool) error

SetPausedTx serializes pause/resume with admission inside the caller-owned queue database transaction. Already admitted work is not canceled.

type Storage 

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

Storage implements queue.Storage backed by a SQL database.

func NewStorage 

func NewStorage(db *sql.DB, opts ...Option) *Storage

NewStorage builds a postgres storage adapter.

func (*Storage) Ack 

func (s *Storage) Ack(ctx context.Context, receipt queue.Receipt) error

Ack removes the message when it has been processed.

func (*Storage) Cleanup 

func (s *Storage) Cleanup(ctx context.Context) error

Cleanup drops queue tables (for tests).

func (*Storage) Database added in v0.20.0 

func (s *Storage) Database() *sql.DB

Database identifies the database required by a caller-owned enqueue transaction.

func (*Storage) Dequeue 

func (s *Storage) Dequeue(ctx context.Context) (*job.ExecutionMessage, queue.Receipt, error)

Dequeue leases the next available message.

func (*Storage) Enqueue 

func (s *Storage) Enqueue(ctx context.Context, msg *job.ExecutionMessage) (queue.EnqueueReceipt, error)

Enqueue stores the message for delivery.

func (*Storage) EnqueueAfter added in v0.15.0 

func (s *Storage) EnqueueAfter(ctx context.Context, msg *job.ExecutionMessage, delay time.Duration) (queue.EnqueueReceipt, error)

EnqueueAfter stores a message for delivery after the provided delay.

func (*Storage) EnqueueAt added in v0.15.0 

func (s *Storage) EnqueueAt(ctx context.Context, msg *job.ExecutionMessage, at time.Time) (queue.EnqueueReceipt, error)

EnqueueAt stores a message for delivery at the given time.

func (*Storage) EnqueueTx added in v0.20.0 

func (s *Storage) EnqueueTx(ctx context.Context, tx *sql.Tx, msg *job.ExecutionMessage, at time.Time) (queue.EnqueueReceipt, error)

EnqueueTx writes the message and acceptance status into the caller's SQL transaction. The caller must roll back on any error and commit before exposing the receipt. The transaction must belong to the configured queue database.

func (*Storage) ExtendLease added in v0.15.0 

func (s *Storage) ExtendLease(ctx context.Context, receipt queue.Receipt, ttl time.Duration) error

ExtendLease extends a delivery lease for long-running handlers.

func (*Storage) GetDispatchStatus added in v0.16.0 

func (s *Storage) GetDispatchStatus(ctx context.Context, dispatchID string) (queue.DispatchStatus, error)

GetDispatchStatus reads durable dispatch status records for a dispatch id.

func (*Storage) Migrate 

func (s *Storage) Migrate(ctx context.Context) error

Migrate creates the queue tables and indexes.

func (*Storage) Nack 

func (s *Storage) Nack(ctx context.Context, receipt queue.Receipt, opts queue.NackOptions) error

Nack requeues, delays, or dead-letters a message.

Source Files

View all Source files

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.