Documentation
¶
Overview ¶
Package postgres is the PostgreSQL kind of the outbox store (stores/outbox): the relay's claim, lease and finalize operations over a product's own outbox table, named by configuration. Every product outbox table shares one column layout (below), so a service gets a production store by naming its table rather than re-writing the SKIP LOCKED claim per table.
The table must carry these columns:
id uuid, lane text, payload bytea, status text|enum ('pending','sent','parked'),
attempts int, next_attempt_at timestamptz, last_error text, created_at timestamptz,
sent_at timestamptz NULL
plus the key column (Config.KeyColumn), and optionally a tenant column and any attribute columns. Producers insert pending rows themselves, inside their business transaction; this store only drains them.
Every write (Claim, the finalizers, DeleteSentBefore) runs on the *sql.Tx a transaction manager (stores/sqltx) bound to the context, and fails INTERNAL without touching the database when none is bound, so a write never autocommits outside the workflow that owns it. Stats reads on the bound transaction when there is one.
Index ¶
- type Config
- type Store
- func (s *Store) Claim(ctx context.Context, lane string, limit int) ([]types.OutboxRecord, error)
- func (s *Store) DeleteSentBefore(ctx context.Context, cutoff time.Time) (int, error)
- func (s *Store) MarkSent(ctx context.Context, id uuid.UUID) error
- func (s *Store) Park(ctx context.Context, id uuid.UUID, lastErr string) error
- func (s *Store) Retry(ctx context.Context, id uuid.UUID, next time.Time, lastErr string) error
- func (s *Store) Stats(ctx context.Context, lane string) (types.OutboxStats, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// AttributeColumns maps each record attribute key to the column it is read from
// (e.g. {"event_type": "event_type"}); empty reads no attributes.
AttributeColumns map[string]string `yaml:"attribute_columns" mapstructure:"attribute_columns"`
// Table is the outbox table's name.
Table string `yaml:"table" mapstructure:"table"`
// KeyColumn is the column a record's Key is read from (as text); empty means
// "aggregate_id".
KeyColumn string `yaml:"key_column" mapstructure:"key_column"`
// TenantColumn is the column a record's Tenant is read from; empty means the table
// has none and every record's Tenant is "".
TenantColumn string `yaml:"tenant_column" mapstructure:"tenant_column"`
// Lease is how long a claimed but unfinalized row is hidden from another claim. It
// must outlast a relay pass, yet be short enough that a crashed relay's rows come
// back promptly; it must be positive.
Lease time.Duration `yaml:"lease" mapstructure:"lease"`
// MaxErrorLen caps the recorded last_error, so a long delivery error cannot fail
// the very update that records it; zero records it whole.
MaxErrorLen int `yaml:"max_error_len" mapstructure:"max_error_len"`
}
Config names a store's table and the columns that vary between outbox tables.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the OutboxStore over one PostgreSQL outbox table.
func New ¶
New returns the store for cfg over db. An invalid cfg is CodeInvalidInput; a nil db panics, as a wiring bug. It performs no I/O.
func (*Store) Claim ¶
Claim leases up to limit due pending rows of lane, oldest first; the returned Attempts counts this claim. A non-positive limit is CodeInvalidInput, since LIMIT 0 would silently claim nothing every pass.
func (*Store) DeleteSentBefore ¶
DeleteSentBefore purges sent rows whose sent_at is before cutoff and returns how many went; parked rows stay for an operator. The claim index covers only pending rows, so a retention sweep calls it to keep sent rows from accumulating.
func (*Store) MarkSent ¶
MarkSent moves a pending row to sent, stamping sent_at on the database clock and clearing last_error. A row already sent or parked is left unchanged.
func (*Store) Park ¶
Park takes a pending row out of rotation and records lastErr; the row stays for an operator to inspect. A row already sent is left unchanged.
Source Files