outbox

Documentation

Overview

Package outbox provides transport-neutral durable event publication contracts and an explicit at-least-once dispatcher.

Index

• Variables
• type ClaimRequest
• type Completion
• type Delivery
  • func NewDelivery(message Message, receipt string, attempt int) (Delivery, error)
  • func (delivery Delivery) Attempt() int
  • func (delivery Delivery) Message() Message
  • func (delivery Delivery) Receipt() string
• type Dispatcher
  • func NewDispatcher(store Store, publisher Publisher, options Options, observers ...Observer) (*Dispatcher, error)
  • func (dispatcher *Dispatcher) RunOnce(ctx context.Context) (Result, error)
• type FailureDelay
• type Message
  • func NewMessage(spec MessageSpec) (Message, error)
  • func (message Message) ContentType() string
  • func (message Message) ID() string
  • func (message Message) Module() string
  • func (message Message) OccurredAt() time.Time
  • func (message Message) Payload() []byte
  • func (message Message) Topic() string
• type MessageSpec
• type Observation
• type Observer
• type Options
• type Publisher
• type Release
• type Result
• type SQLStatements
• type SQLStore
  • func NewSQLStore(executor data.Executor, statements SQLStatements) (*SQLStore, error)
  • func (store *SQLStore) Claim(ctx context.Context, request ClaimRequest) (deliveries []Delivery, resultErr error)
  • func (store *SQLStore) Complete(ctx context.Context, completion Completion) error
  • func (store *SQLStore) Enqueue(ctx context.Context, executor data.Executor, message Message) error
  • func (store *SQLStore) Release(ctx context.Context, release Release) error
• type Store

Variables

var ErrPublisherPanicked = errors.New("outbox publisher panicked")

ErrPublisherPanicked identifies an observed publisher panic. RunOnce reports it and then re-panics with the original value.

Types

type ClaimRequest

type ClaimRequest struct {
	Owner string
	Now   time.Time
	Lease time.Duration
	Limit int
}

ClaimRequest asks a store to atomically lease the oldest available messages.

type Completion

type Completion struct {
	Owner   string
	Receipt string
}

Completion identifies one lease that was published successfully.

type Delivery

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

Delivery is an immutable leased message returned by a Store.

func NewDelivery

func NewDelivery(message Message, receipt string, attempt int) (Delivery, error)

NewDelivery validates and freezes one store lease.

func (Delivery) Attempt

func (delivery Delivery) Attempt() int

Attempt returns the one-based delivery attempt.

func (Delivery) Message

func (delivery Delivery) Message() Message

Message returns the immutable leased event.

func (Delivery) Receipt

func (delivery Delivery) Receipt() string

Receipt returns the store-owned opaque lease receipt.

type Dispatcher

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

Dispatcher claims, publishes, and completes durable messages.

func NewDispatcher

func NewDispatcher(
	store Store,
	publisher Publisher,
	options Options,
	observers ...Observer,
) (*Dispatcher, error)

NewDispatcher validates and freezes one dispatch worker.

func (*Dispatcher) RunOnce

func (dispatcher *Dispatcher) RunOnce(ctx context.Context) (Result, error)

RunOnce performs one deterministic claim batch. Publishing is at least once: a successful publish followed by a completion failure can be delivered again.

type FailureDelay

type FailureDelay func(Delivery) time.Duration

FailureDelay computes when a failed lease becomes available again.

type Message

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

Message is an immutable serialized event prepared for durable storage.

func NewMessage

func NewMessage(spec MessageSpec) (Message, error)

NewMessage validates and freezes one serialized event. IDs are caller-owned idempotency keys and must be unique within the application outbox.

func (Message) ContentType

func (message Message) ContentType() string

ContentType returns the normalized payload media type.

func (Message) ID

func (message Message) ID() string

ID returns the caller-owned idempotency key.

func (Message) Module

func (message Message) Module() string

Module returns the publishing module identity.

func (Message) OccurredAt

func (message Message) OccurredAt() time.Time

OccurredAt returns the UTC event occurrence time.

func (Message) Payload

func (message Message) Payload() []byte

Payload returns a defensive copy of the serialized event.

func (Message) Topic

func (message Message) Topic() string

Topic returns the stable event contract identity.

type MessageSpec

type MessageSpec struct {
	ID          string
	Topic       string
	Module      string
	ContentType string
	Payload     []byte
	OccurredAt  time.Time
}

MessageSpec is the inspectable input to NewMessage.

type Observation

type Observation struct {
	Topic     string
	Module    string
	Attempt   int
	Duration  time.Duration
	Published bool
	Completed bool
	Released  bool
	Err       error
	Panicked  bool
}

Observation contains bounded metadata and no payload or lease receipt.

type Observer

type Observer func(context.Context, Observation)

Observer receives completed delivery attempts synchronously.

type Options

type Options struct {
	Owner        string
	BatchSize    int
	Lease        time.Duration
	Clock        func() time.Time
	FailureDelay FailureDelay
}

Options configures one instance-owned dispatcher.

type Publisher

type Publisher interface {
	Publish(context.Context, Message) error
}

Publisher sends one message to an external transport. Implementations must use Message.ID as the downstream idempotency key.

type Release

type Release struct {
	Owner       string
	Receipt     string
	AvailableAt time.Time
}

Release makes one failed delivery available after an explicit delay.

type Result

type Result struct {
	Claimed   int
	Published int
	Completed int
	Released  int
}

Result summarizes one bounded dispatch pass.

type SQLStatements

type SQLStatements struct {
	Insert   string
	Claim    string
	Complete string
	Release  string
}

SQLStatements supplies dialect-owned, fixed SQL for the outbox protocol. Statement text is trusted startup configuration, never request input.

type SQLStore

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

SQLStore implements Store using standard database/sql contracts.

func NewSQLStore

func NewSQLStore(
	executor data.Executor,
	statements SQLStatements,
) (*SQLStore, error)

NewSQLStore validates and freezes one driver-neutral SQL store. Construction performs no database operation.

func (*SQLStore) Claim

func (store *SQLStore) Claim(
	ctx context.Context,
	request ClaimRequest,
) (deliveries []Delivery, resultErr error)

Claim atomically leases messages through the configured statement. Arguments are owner, current time, lease expiry, and limit. Rows must return ID, topic, module, content type, payload, occurrence time, receipt, and attempt.

func (*SQLStore) Complete

func (store *SQLStore) Complete(ctx context.Context, completion Completion) error

Complete removes or marks one published lease using owner and receipt.

func (*SQLStore) Enqueue

func (store *SQLStore) Enqueue(
	ctx context.Context,
	executor data.Executor,
	message Message,
) error

Enqueue inserts one message through the supplied application transaction. Arguments are ID, topic, module, content type, payload, and occurrence time.

func (*SQLStore) Release

func (store *SQLStore) Release(ctx context.Context, release Release) error

Release clears one failed lease and sets its next availability.

type Store

type Store interface {
	Enqueue(context.Context, data.Executor, Message) error
	Claim(context.Context, ClaimRequest) ([]Delivery, error)
	Complete(context.Context, Completion) error
	Release(context.Context, Release) error
}

Store owns durable persistence and lease transitions. Enqueue must use the supplied executor so application state and its event can commit atomically.