core

Documentation

Overview

Package core holds the outbox domain: what a message is, how a failure is classified and how the next attempt is scheduled. It depends on nothing in this program, so the rules can be read — and tested — without a database or a broker in the way.

Index

• Variables
• func IsPermanent(err error) bool
• func IsUnavailable(err error) bool
• func Permanent(reason string, err error) error
• func Unavailable(reason string, err error) error
• type BackoffPolicy
  • func (p BackoffPolicy) Next(attempts int) time.Duration
• type Lease
• type Message
• type Outcome
• type PermanentError
  • func (e *PermanentError) Error() string
  • func (e *PermanentError) Unwrap() error
• type RetryLimits
• type Status
  • func (s Status) String() string
  • func (s Status) Valid() bool
• type Target
• type UnavailableError
  • func (e *UnavailableError) Error() string
  • func (e *UnavailableError) Unwrap() error

Variables

var (
	// ErrUnknownStream is returned when a message names a stream that is not
	// configured. It is permanent by construction: no amount of retrying will
	// make the stream appear.
	ErrUnknownStream = errors.New("unknown stream")
	// ErrUnknownDriver is returned when a configured stream points at a driver
	// that was never built.
	ErrUnknownDriver = errors.New("unknown driver")
	// ErrLeaseLost reports that a write-back matched fewer rows than it was
	// given, meaning the lease was reclaimed by another instance mid-flight.
	// It is not a failure of the message — the row belongs to someone else now.
	ErrLeaseLost = errors.New("lease lost")
)

Sentinel errors for conditions the dispatcher and store distinguish.

Functions

func IsPermanent

func IsPermanent(err error) bool

IsPermanent reports whether err — or anything it wraps — is permanent.

func IsUnavailable

func IsUnavailable(err error) bool

IsUnavailable reports whether err — or anything it wraps — is a failure to reach the broker.

A permanent error wins over this one: a payload the broker would reject stays permanent even if the connection also happened to drop, because the next attempt would reach the same conclusion. Callers classify permanence first for that reason.

func Permanent

func Permanent(reason string, err error) error

Permanent wraps err as a non-retryable failure.

func Unavailable

func Unavailable(reason string, err error) error

Unavailable wraps err as a failure to reach the broker.

Types

type BackoffPolicy

type BackoffPolicy struct {
	Base   time.Duration
	Max    time.Duration
	Jitter float64 // fraction of the delay, 0…1; 0.2 means ±20%
}

BackoffPolicy schedules the next attempt after a retryable failure.

Two properties are easy to leave out and expensive to be without:

• a ceiling. Unbounded doubling of a 60s base reaches days by the tenth attempt, long after anyone would consider the message live.
• jitter. Without it, every message that failed while a broker was down becomes due at the same instant when it comes back, and the recovered broker is met with the entire backlog at once.

func (BackoffPolicy) Next

func (p BackoffPolicy) Next(attempts int) time.Duration

Next returns the delay before attempt number attempts+1, where attempts is the number of attempts already made (so the first failure passes 1).

The result is Base * 2^(attempts-1), capped at Max, then spread by Jitter. It is never negative and never exceeds Max by more than the jitter fraction.

type Lease

type Lease struct {
	Token string
	Owner string
	Until time.Time
}

Lease identifies one claim. The token is what makes a write-back safe when several instances run: only the holder of the current token may finalize a row, so an instance whose lease expired mid-flight cannot overwrite the outcome recorded by whoever reclaimed the row.

type Message

type Message struct {
	ID       string
	Stream   string
	Topic    string
	Payload  []byte
	Headers  map[string]string
	Target   Target
	Attempts int

	// CreatedAt is read back from the database so delivery lag is measured
	// against the database clock rather than this process's. The two disagree,
	// and a lag metric that spans both absorbs the difference silently.
	CreatedAt time.Time
}

Message is one outbox row as the dispatcher sees it: everything needed to publish, plus the bookkeeping needed to write the outcome back.

type Outcome

type Outcome struct {
	ID string
	// Err is nil on success.
	Err error
	// Permanent marks a failure that retrying cannot fix, so the message goes
	// straight to StatusFailed instead of burning the attempt budget.
	Permanent bool
	// Deferred marks a failure to reach the broker at all. The message returns
	// to StatusPending without advancing its attempt counter: it was never
	// offered to the broker, so it should not be charged for the outage. See
	// UnavailableError.
	//
	// Permanent takes precedence — a message the broker would reject is failed
	// whether or not the connection also dropped.
	Deferred bool
	// Delay is how long to wait before the next attempt. Computed in Go rather
	// than in SQL so the backoff policy — including its jitter — stays
	// testable.
	Delay time.Duration
}

Outcome is the result of publishing one message, on its way back to the database.

type PermanentError

type PermanentError struct {
	Reason string
	Err    error
}

PermanentError marks a publish failure that retrying cannot fix. The dispatcher sends such a message straight to StatusFailed instead of spending the whole attempt budget on it: retrying an unroutable message or an unknown stream five times, each after a longer backoff, reaches the same conclusion an hour later.

Permanent, in practice: an unknown stream or driver, a broker rejecting the payload as too large, an unroutable publish (RabbitMQ basic.return), an unknown topic with auto-creation disabled, and authentication failures. Everything else is retryable.

func (*PermanentError) Error

func (e *PermanentError) Error() string

func (*PermanentError) Unwrap

func (e *PermanentError) Unwrap() error

type RetryLimits

type RetryLimits struct {
	// MaxAttempts is how many times a broker may reject a message before it is
	// given up on.
	MaxAttempts int
	// MaxDefer bounds how long an unreachable broker may hold a message back
	// before it fails anyway, measured from the first deferral rather than from
	// the row's creation — an old message meeting its first outage has waited
	// no time at all.
	//
	// Zero means unbounded: the message waits for as long as the broker is
	// down, and the backlog age is what raises the alarm. That is the default,
	// because a message failed by a timeout is worth less than one that is
	// merely late, and because failing it makes an outage into an operator's
	// problem twice.
	MaxDefer time.Duration
}

RetryLimits bounds how long the dispatcher keeps trying, along the two axes that fail for different reasons.

type Status

type Status int16

Status is the lifecycle state of an outbox row. It is stored as a SMALLINT with a CHECK constraint rather than as text: the set is closed, and the schema should say so.

const (
	// StatusPending — ready to be claimed once available_at has passed.
	StatusPending Status = 0
	// StatusProcessing — claimed by an instance holding a lease.
	StatusProcessing Status = 1
	// StatusSent — accepted by the broker. Terminal.
	StatusSent Status = 2
	// StatusFailed — attempts exhausted, or a permanent error. Terminal until
	// an explicit requeue.
	StatusFailed Status = 3
)

func (Status) String

func (s Status) String() string

func (Status) Valid

func (s Status) Valid() bool

Valid reports whether s is one of the four defined statuses.

type Target

type Target struct {
	// Key is the partition key. Kafka uses it; RabbitMQ ignores it.
	Key string `json:"key,omitempty"`
	// Version, when above zero, appends a "vN" suffix to the effective topic
	// name. It lived in a dedicated SMALLINT column before, which read as if
	// it versioned the row rather than the topic name.
	Version int `json:"version,omitempty"`
	// Exchange and RoutingKey are RabbitMQ-only. Empty Exchange means the
	// default exchange, and an empty RoutingKey means the topic name, so a
	// message that says nothing about routing still goes somewhere sensible.
	Exchange   string `json:"exchange,omitempty"`
	RoutingKey string `json:"routing_key,omitempty"`
}

Target is the routing envelope stored in the target JSONB column. Unknown keys are preserved by the database and ignored here; that is the extension point.

type UnavailableError

type UnavailableError struct {
	Reason string
	Err    error
}

UnavailableError marks a failure to reach the broker at all: no live connection, a channel closed underneath the publish, a confirmation that never arrived because the socket had gone. The broker never saw the message, so it never refused it — and charging the message an attempt for that spends its budget on somebody else's outage. At the default backoff the whole budget is gone in fifteen minutes, so a twenty-minute restart leaves a table full of failed rows that only ever needed to wait.

A message failing this way returns to pending on the ordinary backoff with its attempt counter untouched, and is marked deferred until it either goes through or exceeds DispatchConfig.MaxDefer. What raises the alarm is the age of the backlog and outbox_messages_deferred_total, which is the pair an operator should be woken by in any case.

The classification has to be conservative in one specific direction: a per-message problem mistaken for an outage never advances its counter and so never reaches failed. When a driver cannot tell the two apart, the retryable answer is the safe one.

func (*UnavailableError) Error

func (e *UnavailableError) Error() string

func (*UnavailableError) Unwrap

func (e *UnavailableError) Unwrap() error