Documentation
¶
Overview ¶
Package outbox is the transactional outbox (S19 8; design: zero-downtime-architecture.md section 2.3). A discrete side effect is recorded as a row in the same transaction as the change that causes it (Insert takes the caller's tx), and a sender claims, sends and marks it.
Delivery is at-least-once: a sender that crashes between the send and MarkSent leaves a claim that expires and the row is claimed again. Each effect therefore keeps its own claim as the dedupe (engagement_sends, left_open_alerts, tire_low_alerts, ota_notices, the live-activity claim), and each kind has a MaxAge after which a stale intent is dropped by the sender instead of being sent late.
Slice 1 ships the store only; nothing calls it yet. The table is service-only: the request role has no grant and the repair role cannot read it.
Index ¶
- Constants
- Variables
- func WithSuppressed(ctx context.Context) context.Context
- type Handler
- type Intent
- type Janitor
- type Purged
- type Row
- type Sender
- type Store
- func (s *Store) Claim(ctx context.Context, owner string, n int, lease time.Duration) ([]Row, error)
- func (s *Store) ClaimKinds(ctx context.Context, owner string, kinds []string, n int, lease time.Duration) ([]Row, error)
- func (s *Store) Drop(ctx context.Context, id int64, owner string) error
- func (s *Store) Insert(ctx context.Context, tx *sql.Tx, in Intent) (bool, error)
- func (s *Store) MarkSent(ctx context.Context, id int64, owner string) error
- func (s *Store) PurgeBatch(ctx context.Context, sentBefore, failedBefore time.Time, limit int) (Purged, error)
- func (s *Store) Release(ctx context.Context, id int64, owner string) error
Constants ¶
const ( SentTTL = 7 * 24 * time.Hour FailedTTL = 30 * 24 * time.Hour )
Retention, as the design fixes it: sent rows 7 days, failed rows 30 days. A pending row that nothing consumed within its kind's MaxAge plus PendingGrace is expired, so shadow rows written before any handler exists do not stay forever.
const ( KindCloseHook = "close_hook" // a drive or charge closed: weather, recap, engagement KindLiveActivity = "live_activity" // a Live Activity start or end (the periodic updates do not go through the outbox) KindPush = "push" KindEmail = "email" // account-level: no vehicle KindMilestone = "milestone" )
The kinds of discrete event the outbox carries. The set is fixed so a metric label stays bounded.
const ( // MaxAttempts is how many times a row is claimed before it is set aside as failed. MaxAttempts = 8 // MaxBackoff caps the wait after a failed send; the wait doubles from 30 s. MaxBackoff = time.Hour // MaxPayload is the payload cap, also a table CHECK. MaxPayload = 4096 )
const PendingGrace = 24 * time.Hour
PendingGrace is how long past its kind's MaxAge a never-consumed pending row stays before Purge expires it.
Variables ¶
var ErrMoot = errors.New("outbox: intent is moot")
ErrMoot is a Handler's answer when the intent no longer applies (the session is gone, the effect already happened): the row is set aside as failed without another try.
var ErrNotClaimed = errors.New("outbox: row is not claimed by this owner")
ErrNotClaimed is MarkSent's and Release's answer when the row is not claimed by that owner any more (or was already sent): the caller does not mark it.
var MaxAge = map[string]time.Duration{ KindCloseHook: 24 * time.Hour, KindLiveActivity: 5 * time.Minute, KindPush: time.Hour, KindEmail: 24 * time.Hour, KindMilestone: 24 * time.Hour, }
MaxAge is how old an intent may be and still be sent. A sender drops (marks failed) an older one rather than send it late.
Functions ¶
Types ¶
type Handler ¶
Handler performs one intent's side effect. It must be safe to run twice for one row (delivery is at-least-once): the effect keeps its own claim. nil marks the row sent, ErrMoot sets it aside, any other error releases it for a later try.
type Intent ¶
type Intent struct {
Key string
UserID uuid.UUID
VehicleID uuid.UUID // uuid.Nil for an account-level kind
Kind string
Payload json.RawMessage // ids and kinds only: never coordinates, addresses or names; empty means {}
}
Intent is one side effect to record.
type Janitor ¶
type Janitor struct {
Interval time.Duration // between runs
BatchSize int // rows per DELETE
MaxBatch int // batches per run (the next run continues)
Pause time.Duration // between batches
// contains filtered or unexported fields
}
Janitor purges the outbox on a schedule, in limited batches, under a singleton lock.
func NewJanitor ¶
func NewJanitor(db *sql.DB, store *Store, reg prometheus.Registerer) *Janitor
NewJanitor returns a janitor. reg may be nil (tests).
type Purged ¶
type Purged struct{ Sent, Failed, Expired int64 }
Purged is what one PurgeBatch removed, by class.
type Row ¶
type Row struct {
ID int64
Key string
UserID uuid.UUID
VehicleID uuid.UUID
Kind string
Payload json.RawMessage
CreatedAt time.Time
Attempts int
}
Row is a claimed intent.
type Sender ¶
type Sender struct {
// Interval is the wait between ticks, Batch the rows per tick, Lease how long a claim holds (it must outlast a handler).
Interval time.Duration
Batch int
Lease time.Duration
// contains filtered or unexported fields
}
Sender claims due rows of the kinds it has a handler for and runs the handlers. With no handler registered it claims nothing, so shadow rows written before any consumer exists stay untouched until Purge expires them.
func NewSender ¶
func NewSender(store *Store, reg prometheus.Registerer) *Sender
NewSender returns a sender on store. reg may be nil (tests).
func (*Sender) Once ¶
Once runs one tick and returns the rows it handled. Safe on any number of pods: a claim is SKIP LOCKED and leased.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store reads and writes the outbox table.
func NewStore ¶
func NewStore(db *sql.DB, reg prometheus.Registerer) *Store
NewStore returns a store on db. reg may be nil (tests).
func (*Store) Claim ¶
Claim leases up to n due rows to owner for lease. A row is due when it is not sent or failed and its claim is absent or expired (a Release pushes claimed_until out by its backoff, so a backed-off row is not due). attempts rises at the claim. A row already claimed MaxAttempts times is set aside as failed here and not returned, so a row that crashes its sender cannot come back forever on lease expiry.
func (*Store) ClaimKinds ¶
func (s *Store) ClaimKinds(ctx context.Context, owner string, kinds []string, n int, lease time.Duration) ([]Row, error)
ClaimKinds is Claim restricted to the given kinds. With none it claims nothing and does not query: a sender with no handler registered leaves every row alone (not claimed, not counted as an attempt).
func (*Store) Drop ¶
Drop sets a claimed row aside as failed without a send (a stale intent past its kind's MaxAge).
func (*Store) Insert ¶
Insert records in in tx, the caller's transaction (so it rolls back with the change that causes it). It reports whether a row was written: false for a key already recorded, and for a suppressed context.
func (*Store) MarkSent ¶
MarkSent records the send. It needs the row to be claimed by owner and not yet sent; it does not require the lease to be live, so a send that finishes just after its lease expired (and before anyone reclaimed the row) is recorded and not sent again.
func (*Store) PurgeBatch ¶
func (s *Store) PurgeBatch(ctx context.Context, sentBefore, failedBefore time.Time, limit int) (Purged, error)
PurgeBatch deletes at most limit rows, oldest id first: sent before sentBefore, failed before failedBefore, and pending rows (never sent, never failed) older than their kind's MaxAge plus PendingGrace, which nothing consumed (a shadow row with no handler, or an intent past any use). A short batch (Total < limit) means the table is clear of them; the caller loops with pauses so no statement runs long.
func (*Store) Release ¶
Release gives a row back after a failed send. claimed_by is cleared and claimed_until set to now() + backoff (30 s doubling per attempt, capped at MaxBackoff), which the claim query honours, so the row is not due until the backoff passes. A row at MaxAttempts is set aside as failed.
Source Files