postgres

Documentation

Overview

Package postgres is a persistent, multi-process rerun.Store backed by PostgreSQL via the pure-Go, cgo-free driver github.com/lib/pq. Its lease is a session-scoped advisory lock, so a lease held by a worker that dies is released automatically when that worker's connection drops — no expiry, no heartbeat, no reaper.

Index

• type Store
  • func New(dsn string) *Store
  • func (s *Store) Acquire(ctx context.Context, runID string) (io.Closer, bool, error)
  • func (s *Store) Append(ctx context.Context, runID string, l rerun.Log) error
  • func (s *Store) CancelRequested(ctx context.Context, runID string) (bool, error)
  • func (s *Store) Close() error
  • func (s *Store) Create(ctx context.Context, r rerun.Run) error
  • func (s *Store) Finish(ctx context.Context, runID string, st rerun.Status) error
  • func (s *Store) Incomplete(ctx context.Context) ([]rerun.Run, error)
  • func (s *Store) LoadLogs(ctx context.Context, runID string) ([]rerun.Log, error)
  • func (s *Store) PopSignal(ctx context.Context, runID, name string) ([]byte, bool, error)
  • func (s *Store) PushSignal(ctx context.Context, runID, name string, payload []byte) error
  • func (s *Store) RequestCancel(ctx context.Context, runID string) error

Types

type Store

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

Store is a Postgres-backed rerun.Store.

func New

func New(dsn string) *Store

New opens the database at dsn and migrates it to the latest schema. Unlike the SQLite store it does not cap the pool: Postgres handles concurrency, and the lease's exclusivity depends on a competing Acquire landing on a second connection so it observes the lock as held. A v0.1 database is adopted and upgraded in place.

func (*Store) Acquire

func (s *Store) Acquire(ctx context.Context, runID string) (io.Closer, bool, error)

Acquire takes a session-scoped advisory lock on a dedicated connection. pg_try_advisory_lock is non-blocking, so a competing worker (a different session) observes acquired=false. If this worker's process dies the connection drops and Postgres releases the lock automatically, which is what makes a crashed run recoverable across machines.

func (*Store) Append

func (s *Store) Append(ctx context.Context, runID string, l rerun.Log) error

Append inserts a journal row, returning rerun.ErrSeqConflict on a duplicate (run, seq) position.

func (*Store) CancelRequested

func (s *Store) CancelRequested(ctx context.Context, runID string) (bool, error)

CancelRequested reports whether a cancel has been recorded for the run.

func (*Store) Close

func (s *Store) Close() error

Close releases the underlying connection pool.

func (*Store) Create

func (s *Store) Create(ctx context.Context, r rerun.Run) error

Create inserts a new run row, returning rerun.ErrRunExists on a duplicate ID.

func (*Store) Finish

func (s *Store) Finish(ctx context.Context, runID string, st rerun.Status) error

Finish updates a run's status.

func (*Store) Incomplete

func (s *Store) Incomplete(ctx context.Context) ([]rerun.Run, error)

Incomplete returns every run still Pending or Running.

func (*Store) LoadLogs

func (s *Store) LoadLogs(ctx context.Context, runID string) ([]rerun.Log, error)

LoadLogs returns a run's journal entries ordered by sequence.

func (*Store) PopSignal

func (s *Store) PopSignal(ctx context.Context, runID, name string) ([]byte, bool, error)

PopSignal removes and returns the oldest queued signal for (runID, name).

func (*Store) PushSignal

func (s *Store) PushSignal(ctx context.Context, runID, name string, payload []byte) error

PushSignal and PopSignal implement rerun.Signaler: a durable, per-run, per-name FIFO mailbox. A signal delivered before the workflow reaches Wait waits in the table until Wait pops it, and survives a crash in between.

func (*Store) RequestCancel

func (s *Store) RequestCancel(ctx context.Context, runID string) error

RequestCancel and CancelRequested implement rerun.Canceller: a durable cancel flag so Cancel can reach a run executing in another process.