postgres

package

v0.1.0-beta.4 Latest Latest Go to latest Published: Aug 3, 2026 License: MIT Imports: 19 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/blissfulrays/pushq

Links

Open Source Insights

Documentation ¶

Overview ¶

Package postgres implements store.Store on PostgreSQL via pgx. Supports multiple server replicas: claims use FOR UPDATE SKIP LOCKED, dispatcher wakeups flow through LISTEN/NOTIFY, and the dispatcher role is serialized with an advisory-lock elector.

Index ¶

type Elector
- func (e *Elector) Acquire(ctx context.Context) (<-chan struct{}, error)
- func (e *Elector) Release()
type Store
- func Open(dsn string) (*Store, error)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type Elector ¶

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

Elector serializes the dispatcher role across replicas with a session-scoped Postgres advisory lock. Losing the connection loses the lock, which is exactly the failover behavior we want. Leases remain the correctness layer during the zombie window; this only prevents duplicate claim scans.

func (*Elector) Acquire ¶

func (e *Elector) Acquire(ctx context.Context) (<-chan struct{}, error)

Acquire blocks until the advisory lock is held. The returned channel closes if the lock connection later dies.

func (*Elector) Release ¶

func (e *Elector) Release()

Release drops the lock and closes the connection so a standby can take over immediately (blue/green handoff).

type Store ¶

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

func Open ¶

func Open(dsn string) (*Store, error)

func (*Store) ClaimTaskForRun ¶

func (s *Store) ClaimTaskForRun(ctx context.Context, now time.Time, leaseToken string, leaseGrace time.Duration, taskPK int64) (store.Task, bool, error)

func (*Store) ClaimTasks ¶

func (s *Store) ClaimTasks(ctx context.Context, now time.Time, leaseToken string, leaseGrace time.Duration, budgets []store.ClaimBudget) ([]store.Task, error)

ClaimTasks: one budgeted statement across all hot queues — a VALUES list CROSS JOIN LATERAL per-queue limited SKIP LOCKED select feeding an UPDATE.

func (*Store) CleanupTerminal ¶

func (s *Store) CleanupTerminal(ctx context.Context, now time.Time, succeededTTL, failedTTL, dedupWindow time.Duration, limit int) (int64, error)

func (*Store) Close ¶

func (s *Store) Close() error

func (*Store) CreateQueue ¶

func (s *Store) CreateQueue(ctx context.Context, u store.QueueUpsert, now time.Time) (store.Queue, error)

func (*Store) CreateTask ¶

func (s *Store) CreateTask(ctx context.Context, t store.NewTask, now time.Time) (store.Task, error)

func (*Store) CreateTasks ¶

func (s *Store) CreateTasks(ctx context.Context, ts []store.NewTask, now time.Time) ([]store.TaskResult, error)

func (*Store) DB ¶

func (s *Store) DB() *sql.DB

DB exposes the underlying pool (used by tests).

func (*Store) DeleteQueue ¶

func (s *Store) DeleteQueue(ctx context.Context, name string) error

func (*Store) DeleteTask ¶

func (s *Store) DeleteTask(ctx context.Context, queue, taskID string, now time.Time) error

func (*Store) Dialect ¶

func (s *Store) Dialect() string

func (*Store) Events ¶

func (s *Store) Events() <-chan store.Event

func (*Store) FinishAttempt ¶

func (s *Store) FinishAttempt(ctx context.Context, now time.Time, taskPK int64, leaseToken string, att store.Attempt, outcome store.Outcome, nextSchedule time.Time) (bool, error)

func (*Store) ForceRun ¶

func (s *Store) ForceRun(ctx context.Context, queue, taskID string, now time.Time) (store.Task, error)

func (*Store) GetQueue ¶

func (s *Store) GetQueue(ctx context.Context, name string) (store.Queue, error)

func (*Store) GetQueuesByID ¶

func (s *Store) GetQueuesByID(ctx context.Context, ids []int64) ([]store.Queue, error)

func (*Store) GetTask ¶

func (s *Store) GetTask(ctx context.Context, queue, taskID string, view store.View) (store.Task, error)

func (*Store) HardDeletePurged ¶

func (s *Store) HardDeletePurged(ctx context.Context, limit int) (int64, error)

func (*Store) ListAttempts ¶

func (s *Store) ListAttempts(ctx context.Context, queue, taskID string) ([]store.Attempt, error)

func (*Store) ListExpiredLeases ¶

func (s *Store) ListExpiredLeases(ctx context.Context, now time.Time, limit int) ([]store.Task, error)

func (*Store) ListQueues ¶

func (s *Store) ListQueues(ctx context.Context, opts store.ListQueuesOpts) ([]store.Queue, string, error)

func (*Store) ListTasks ¶

func (s *Store) ListTasks(ctx context.Context, opts store.ListTasksOpts) ([]store.Task, string, error)

func (*Store) Migrate ¶

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

func (*Store) NewElector ¶

func (s *Store) NewElector() *Elector

func (*Store) NextPending ¶

func (s *Store) NextPending(ctx context.Context, queueID int64, now time.Time) (*time.Time, error)

func (*Store) PatchQueue ¶

func (s *Store) PatchQueue(ctx context.Context, name string, p store.QueuePatch, now time.Time) (store.Queue, error)

func (*Store) Ping ¶

func (s *Store) Ping(ctx context.Context) error

func (*Store) PurgeQueue ¶

func (s *Store) PurgeQueue(ctx context.Context, name string, now time.Time) (store.Queue, error)

func (*Store) QueueStats ¶

func (s *Store) QueueStats(ctx context.Context, name, idPrefix string, now time.Time) (store.QueueStats, error)

func (*Store) SeedDispatch ¶

func (s *Store) SeedDispatch(ctx context.Context) ([]store.QueueSeed, error)

func (*Store) SetQueueState ¶

func (s *Store) SetQueueState(ctx context.Context, name string, state store.QueueState, now time.Time) (store.Queue, error)

func (*Store) UpsertQueue ¶

func (s *Store) UpsertQueue(ctx context.Context, u store.QueueUpsert, now time.Time) (store.Queue, store.UpsertOutcome, error)

Source Files ¶

View all Source files

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