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

postgres

package
v2.15.1 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func TruncateForTest 

func TruncateForTest(t *testing.T, s *Store)

TruncateForTest clears all queue/attempt/trend/secret tables on the given store. It is exported solely for cross-package integration tests and panics on any error so callers don't have to plumb the *testing.T deeper.

Types

type Option 

type Option func(*Store)

func WithAttemptsRetention added in v2.11.0 

func WithAttemptsRetention(maxAge time.Duration, maxRows int) Option

WithAttemptsRetention bounds delivery-attempt history by age and by row count. Both dimensions are optional; zero disables that half.

func WithDLQRetention 

func WithDLQRetention(maxAge time.Duration, maxDepth int) Option

func WithDeliveredRetention 

func WithDeliveredRetention(maxAge time.Duration) Option

func WithNowFunc 

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

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

func WithQueueLimits 

func WithQueueLimits(maxDepth int, dropPolicy string) Option

func WithRetention 

func WithRetention(maxAge, pruneInterval time.Duration) Option

type Store 

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

func NewStore 

func NewStore(dsn string, opts ...Option) (*Store, error)

func (*Store) Ack 

func (s *Store) Ack(leaseID string) error

func (*Store) AckBatch 

func (s *Store) AckBatch(leaseIDs []string) (queue.LeaseBatchResult, error)

func (*Store) CancelMessages 

func (s *Store) CancelMessages(req queue.MessageCancelRequest) (queue.MessageCancelResponse, error)

func (*Store) CancelMessagesByFilter 

func (s *Store) CancelMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageCancelResponse, error)

func (*Store) CaptureBacklogTrendSample 

func (s *Store) CaptureBacklogTrendSample(at time.Time) error

func (*Store) Close 

func (s *Store) Close() error

func (*Store) Delete 

func (s *Store) Delete(poolName, id string) (bool, error)

func (*Store) DeleteDead 

func (s *Store) DeleteDead(req queue.DeadDeleteRequest) (queue.DeadDeleteResponse, error)

func (*Store) Dequeue 

func (s *Store) Dequeue(req queue.DequeueRequest) (queue.DequeueResponse, error)

func (*Store) DequeueContext added in v2.11.0 

func (s *Store) DequeueContext(dequeueCtx context.Context, req queue.DequeueRequest) (queue.DequeueResponse, error)

DequeueContext is Dequeue with the caller's context honoured: a consumer that has gone away neither keeps a goroutine parked for the rest of max_wait nor receives a lease it can never settle.

func (*Store) Enqueue 

func (s *Store) Enqueue(env queue.Envelope) error

func (*Store) EnqueueBatch added in v2.10.0 

func (s *Store) EnqueueBatch(items []queue.Envelope) (int, error)

EnqueueBatch implements queue.BatchEnqueuer: every item is committed or none is.

Postgres did not implement the interface, so internal/admin fell back to a per-item loop — meaning the batch-publish atomicity guarantee silently degraded to best-effort on the recommended backend. A 50-item publish failing at item 30 left items 0–29 committed and visible while returning an error, so the client retried the whole batch and those 30 were delivered twice.

func (*Store) Extend 

func (s *Store) Extend(leaseID string, extendBy time.Duration) error

func (*Store) LeaseRoutes added in v2.11.0 

func (s *Store) LeaseRoutes(leaseIDs []string) (map[string]string, error)

LeaseRoutes reports the route each known lease belongs to.

func (*Store) ListAll 

func (s *Store) ListAll() ([]secrets.Record, error)

func (*Store) ListAttempts 

func (s *Store) ListAttempts(req queue.AttemptListRequest) (queue.AttemptListResponse, error)

func (*Store) ListBacklogTrend 

func (s *Store) ListBacklogTrend(req queue.BacklogTrendListRequest) (queue.BacklogTrendListResponse, error)

func (*Store) ListByPool 

func (s *Store) ListByPool(poolName string) ([]secrets.Record, error)

func (*Store) ListDead 

func (s *Store) ListDead(req queue.DeadListRequest) (queue.DeadListResponse, error)

func (*Store) ListMessages 

func (s *Store) ListMessages(req queue.MessageListRequest) (queue.MessageListResponse, error)

func (*Store) LookupMessages 

func (s *Store) LookupMessages(req queue.MessageLookupRequest) (queue.MessageLookupResponse, error)

func (*Store) MarkDead 

func (s *Store) MarkDead(leaseID string, reason string) error

func (*Store) MarkDeadBatch 

func (s *Store) MarkDeadBatch(leaseIDs []string, reason string) (queue.LeaseBatchResult, error)

func (*Store) Nack 

func (s *Store) Nack(leaseID string, delay time.Duration) error

func (*Store) NackBatch 

func (s *Store) NackBatch(leaseIDs []string, delay time.Duration) (queue.LeaseBatchResult, error)

func (*Store) NotifyCh 

func (s *Store) NotifyCh() <-chan struct{}

NotifyCh returns a channel that is closed when new items become available. After the channel fires, callers must call NotifyCh again to get the next signal.

func (*Store) RecordAttempt 

func (s *Store) RecordAttempt(attempt queue.DeliveryAttempt) error

func (*Store) RequeueDead 

func (s *Store) RequeueDead(req queue.DeadRequeueRequest) (queue.DeadRequeueResponse, error)

func (*Store) RequeueMessages 

func (s *Store) RequeueMessages(req queue.MessageRequeueRequest) (queue.MessageRequeueResponse, error)

func (*Store) RequeueMessagesByFilter 

func (s *Store) RequeueMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageRequeueResponse, error)

func (*Store) ResumeMessages 

func (s *Store) ResumeMessages(req queue.MessageResumeRequest) (queue.MessageResumeResponse, error)

func (*Store) ResumeMessagesByFilter 

func (s *Store) ResumeMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageResumeResponse, error)

func (*Store) RuntimeMetrics 

func (s *Store) RuntimeMetrics() queue.StoreRuntimeMetrics

func (*Store) Stats 

func (s *Store) Stats() (queue.Stats, error)

func (*Store) Upsert 

func (s *Store) Upsert(rec secrets.Record) error

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.