core

Documentation

Overview

Package core defines notifier's internal persistence model.

Index

• Variables
• func Permanent(err error) error
• func Quarantine(err error) error
• func Retryable(err error) error
• func RetryableAfter(err error, delay time.Duration) error
• type ClaimRequest
• type DestinationID
• type Error
  • func (e *Error) Error() string
  • func (e *Error) Unwrap() error
• type Failure
  • func FailureOf(err error) (Failure, bool)
• type FailureScope
• type Item
• type LeaseToken
• type Op
• type Outcome
  • func (o Outcome) String() string
• type Plan
  • func NewPlan(policy Policy, destinations []DestinationID) Plan
  • func (p Plan) Destinations() []DestinationID
  • func (p Plan) ID() PlanID
  • func (p Plan) Policy() Policy
• type PlanID
• type Policy
• type Prober
• type Resolution
• type Store
• type Work
• type WorkID

Variables

var (
	ErrStoreStaleLeaseToken   = errors.New("store stale lease token")
	ErrStorePayloadConflict   = errors.New("store payload conflict")
	ErrStoreWorkDoesntExist   = errors.New("store work does not exist")
	ErrStoreInvalidTransition = errors.New("store invalid transition")
	ErrStoreBusy              = errors.New("store busy")
	ErrStoreUnavailable       = errors.New("store unavailable")
)

Only ErrStoreBusy and ErrStoreUnavailable are retried.

var (
	ErrRetryable  = errors.New("destination failure is retryable")
	ErrPermanent  = errors.New("destination failure is permanent")
	ErrQuarantine = errors.New("destination is unusable")
)

Sentinels errors.Is matches against the classification a destination applied via Retryable, Permanent, or Quarantine.

var ErrInvalidPlan = errors.New("invalid plan")

ErrInvalidPlan marks a plan that cannot be admitted, such as one with an empty ID.

Functions

func Permanent

func Permanent(err error) error

Permanent classifies err as a permanent failure for one delivery.

func Quarantine

func Quarantine(err error) error

Quarantine classifies err as a permanent destination-wide failure.

func Retryable

func Retryable(err error) error

Retryable classifies err as a retryable delivery-scoped failure.

func RetryableAfter

func RetryableAfter(err error, delay time.Duration) error

RetryableAfter classifies err with a provider-specified minimum retry delay.

Types

type ClaimRequest

type ClaimRequest struct {
	Plan            PlanID
	MaxWork         int
	MaxItemsPerWork int
	LeaseDuration   time.Duration
}

ClaimRequest bounds how much work to lease from a plan in one Claim call.

type DestinationID

type DestinationID string

DestinationID identifies one delivery destination within a plan.

type Error

type Error struct {
	Op      Op
	Subject string
	Err     error
}

Error attributes a failure to an operation and, optionally, a subject.

func (*Error) Error

func (e *Error) Error() string

func (*Error) Unwrap

func (e *Error) Unwrap() error

type Failure

type Failure struct {
	Permanent bool
	Scope     FailureScope
	// RetryAfter is the minimum delay the provider asked for, or zero.
	RetryAfter time.Duration
}

Failure is the classification FailureOf reads back off a destination error.

func FailureOf

func FailureOf(err error) (Failure, bool)

FailureOf reads the classification a destination applied to err. It reports false for an unclassified error, which the dispatcher then treats as retryable.

type FailureScope

type FailureScope int

FailureScope describes whether a failure affects one delivery or a destination.

const (
	FailureScopeUnknown FailureScope = iota
	FailureScopeDelivery
	FailureScopeDestination
)

FailureScope values, from unclassified to the delivery- and destination-wide scopes a failure can carry.

type Item

type Item[T any] struct {
	ID      int64
	Payload T
}

Item is one payload enqueued for delivery, keyed by ID for deduplication.

type LeaseToken

type LeaseToken string

LeaseToken fences a claimed batch so only its current holder can resolve it.

type Op

type Op string

Op names the operation an Error occurred during.

const (
	OpRun           Op = "dispatcher run"
	OpEnqueue       Op = "dispatcher enqueue"
	OpClaim         Op = "dispatcher claim work"
	OpDrain         Op = "dispatcher drain"
	OpResolve       Op = "dispatcher resolve work"
	OpProbe         Op = "dispatcher probe destination"
	OpServiceSetup  Op = "service setup"
	OpServiceWorker Op = "service worker"
)

Operations an Error can be attributed to.

type Outcome

type Outcome int

Outcome is the result a resolution reports for a leased batch.

const (
	OutcomeUnknown Outcome = iota
	OutcomeDelivered
	OutcomeRetryableFailure
	OutcomeFailedPermanent

	OutcomeDeliveredUnrecorded
)

Outcome values, from unresolved through the three terminal dispositions a resolution can report.

func (Outcome) String

func (o Outcome) String() string

type Plan

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

Plan is the immutable destination snapshot a dispatcher delivers against. NewDispatcher derives it from the destinations it is given.

func NewPlan

func NewPlan(policy Policy, destinations []DestinationID) Plan

NewPlan derives a Plan's ID from its policy and destinations.

func (Plan) Destinations

func (p Plan) Destinations() []DestinationID

Destinations returns a clone, so no caller can rewrite a registered plan in place.

func (Plan) ID

func (p Plan) ID() PlanID

ID returns the plan's derived identity.

func (Plan) Policy

func (p Plan) Policy() Policy

Policy returns the plan's completion rule.

type PlanID

type PlanID string

PlanID identifies a registered plan, derived from its policy and destinations.

type Policy

type Policy int

Policy determines when a plan is considered complete across its destinations.

const (
	PolicyUnknown Policy = iota
	// PolicyAll requires every destination to deliver successfully.
	PolicyAll
	// PolicyFirstSuccess completes after the first successful destination.
	PolicyFirstSuccess
)

Policy values, from unset to the two completion rules NewPlan accepts.

type Prober

type Prober interface {
	Probe(ctx context.Context) error
}

Prober checks whether a destination is reachable before work is scheduled against it.

type Resolution

type Resolution struct {
	Work       WorkID
	Lease      LeaseToken
	Outcome    Outcome
	Scope      FailureScope
	RetryAfter time.Duration
	Failure    string
}

Resolution reports the outcome of a leased batch back to the store.

type Store

type Store[T any] interface {
	// Admit registers the plan on first sight, then creates one delivery
	// obligation per destination for each item. Item.ID is the deduplication
	// key: re-admitting a stored ID is a no-op, and reusing one with a different
	// payload must fail.
	Admit(ctx context.Context, plan Plan, items []Item[T]) error
	// Claim hands out up to MaxWork batches and marks them as taken; returning
	// fewer, including none, is normal. Each carries a LeaseUntil deadline,
	// after which the store offers it to someone else, and a fresh LeaseToken.
	Claim(ctx context.Context, request ClaimRequest) ([]Work[T], error)
	// Resolve applies a delivery outcome to one leased batch, rejecting any
	// resolution whose token is not the batch's current lease. Repeating an
	// identical resolution succeeds; contradicting an applied one fails.
	Resolve(ctx context.Context, resolution Resolution) error
	// QuarantineDestination blocks a destination from receiving new work and
	// terminalizes what it already holds, recording failure against it.
	QuarantineDestination(
		ctx context.Context,
		plan PlanID,
		destination DestinationID,
		failure string,
	) error
	// ActivateDestination lets a destination receive work again. Work
	// terminalized by an earlier quarantine stays terminal.
	ActivateDestination(
		ctx context.Context,
		plan PlanID,
		destination DestinationID,
	) error
	// PendingPlans returns every plan that still holds unfinished work,
	// including previous plans not part of this dispatcher's process
	PendingPlans(ctx context.Context) ([]PlanID, error)
}

Store persists plans, items, leases, outcomes, and destination states.

type Work

type Work[T any] struct {
	ID          WorkID
	Plan        PlanID
	Destination DestinationID
	Items       []Item[T]
	Attempt     int
	Lease       LeaseToken
	LeaseUntil  time.Time
}

Work is a leased batch of items claimed for delivery to one destination.

type WorkID

type WorkID string

WorkID identifies one claimed batch of work.