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, a partition column and any attribute columns. Producers insert pending rows inside their business transaction, through Enqueue or their own INSERT; this store drains them.
With Config.PartitionColumn a lane is split into Config.Partitions partitions: a row's partition is types.OutboxPartition(key, partitions), written at insert (Enqueue computes it; a product's own INSERT must use the same function) and never recomputed, and Claim leases only its partition's rows, so one relay per partition drains a lane in parallel. Lowering Partitions strands rows in the dropped partitions until they are rewritten, so a reduction needs a migration; Stats counts them (Stranded) so a relay host can refuse to start over 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, partition, limit int) ([]types.OutboxRecord, error)
- func (s *Store) DeleteSentBefore(ctx context.Context, cutoff time.Time) (int, error)
- func (s *Store) Enqueue(ctx context.Context, rec *types.OutboxRecord) 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"`
// PartitionColumn is the integer column holding a row's partition; empty means the
// table has none and every row is in partition 0.
PartitionColumn string `yaml:"partition_column" mapstructure:"partition_column"`
// Partitions is how many partitions a lane is split into; zero means 1. More than
// one needs PartitionColumn.
Partitions int `yaml:"partitions" mapstructure:"partitions"`
// 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. cfg must not be nil.
func (*Store) Claim ¶
func (s *Store) Claim( ctx context.Context, lane string, partition, limit int, ) ([]types.OutboxRecord, error)
Claim leases up to limit due pending rows of lane's partition, oldest first; the returned Attempts counts this claim. A non-positive limit is CodeInvalidInput, since LIMIT 0 would silently claim nothing every pass, and so is a partition outside [0, partitions).
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) Enqueue ¶ added in v0.0.16
Enqueue writes rec as a pending row due now on the transaction bound to ctx; see interfaces.OutboxWriter. An empty Key, Tenant or attribute is NULL; with a partition column the row's partition is types.OutboxPartition(rec.Key, partitions). rec must not be nil.
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