outbox

type BatchWriter ¶

type BatchWriter interface {
	Writer
	// WriteBatch persists multiple outbox entries atomically within the
	// caller's transaction scope. Implementations MUST validate entries
	// independently (defense-in-depth) even when called via
	// WriteBatchFallback, which only runs Entry.Validate(). If validation
	// fails for any entry, no entries are written.
	//
	// An empty entries slice is a no-op and returns nil.
	//
	// Implementations SHOULD use a single batch INSERT for efficiency.
	// The context MUST carry the caller's transaction (same requirement
	// as Writer.Write). All-or-nothing: either all entries are written
	// or none are (transaction rollback on failure).
	WriteBatch(ctx context.Context, entries []Entry) error
}

BatchWriter extends Writer with batch write support. Implementations that support batch operations SHOULD implement this interface for atomic multi-entry writes within a single transaction.

If an implementation does not support BatchWriter, callers can use WriteBatchFallback which auto-detects batch support and falls back to sequential Write calls.

ref: ThreeDotsLabs/watermill message/pubsub.go -- Publish(topic, ...msgs) variadic pattern. GoCell uses a separate interface instead to preserve the existing Writer contract and enable optimized multi-row INSERT.

type CellEmitter ¶

type CellEmitter interface {
	Emitter
	DurabilityReporter
	healthz.ProbeSet
	// contains filtered or unexported methods
}

CellEmitter is the only Emitter-shaped type that cells/<x> public With* Options (Cell.WithEmitter + every slice WithEmitter) may accept. The unexported sealedCellEmitter() method makes it unimplementable outside this package — kernel/outbox is the sole entry point via WrapEmitterForCell, which composition roots and the kernel-internal ResolveCellEmitter funnel call.

Unlike CellPublisher / CellWriter, CellEmitter additionally embeds DurabilityReporter and healthz.ProbeSet. Embedding the bare Emitter interface in internalCellEmitter promotes only Emit, so the inner *WriterEmitter.Durable() and *DirectEmitter.Probes() would be hidden behind the wrapper. Listing both in the CellEmitter contract makes the forwarding methods on internalCellEmitter COMPILE-MANDATORY (AI-HARD): omit Durable() or Probes() and internalCellEmitter no longer satisfies CellEmitter, so WrapEmitterForCell fails to compile. Without this a missing forward would silently break L2 durability detection (ResolveCellEmitter) and emitter health-probe registration (cell.RegisterEmitterHealthProbes).

AI-robust 评级：

• 字段/赋值层 + Durable()/Probes() forwarding：Hard（sealed marker 使外部 不可表达 internalCellEmitter 字面量；interface 嵌入使内部不可省略 forwarding）
• 公开 With* Option 签名层：Medium（CELL-RAW-INFRA-PUBLIC-OPTION-PARAM-01）
• Noop() 透传：Hard（SEALED-MARKER-NOOP-TRANSPARENCY-01 typed auto-discovery）

CellEmitter embeds Emitter so cells pass the wrapped value directly to slice service constructors / Emit call sites — those accept Emitter and the embedded interface satisfies them transparently, so internal API does not change.

ref: docs/architecture/202605101900-adr-cell-raw-infra-sealed-marker.md §D1

func DemoCellEmitter() CellEmitter

func NewDirectCellEmitter(
	p Publisher, mode DirectPublishFailureMode, mp metrics.Provider, clk clock.Clock, cellID string, opts ...DirectEmitterOption,
) (CellEmitter, error)

func ResolveCellEmitter(clk clock.Clock, in CellEmitterInputs) (CellEmitter, error)

func WrapEmitterForCell(e Emitter) CellEmitter

type CellEmitterInputs ¶

type CellEmitterInputs struct {
	EmitterConfig
	// PreResolved is the sealed emitter set directly via Cell.WithEmitter(e)
	// (wrapped at the composition root via WrapEmitterForCell). When non-nil,
	// ResolveCellEmitter skips ResolveEmitter and validates that durable mode
	// requires a durable PreResolved (PreResolved.Durable()==true).
	PreResolved CellEmitter
	// ConsistencyLevel is the owning Cell's consistency level; used to decide
	// whether the cellvocab.L2 non-durable Warn log fires.
	ConsistencyLevel cellvocab.Level
}

CellEmitterInputs bundles the Cell-side inputs for ResolveCellEmitter. Embeds EmitterConfig and adds the two knobs shared by every Cell's Init-time emitter resolution: the pre-resolved emitter (WithEmitter) and the Cell's consistency level (for the cellvocab.L2 non-durable Warn).

type CellPublisher ¶

type CellPublisher interface {
	Publisher
	// contains filtered or unexported methods
}

CellPublisher is the only Publisher-shaped type that cells/<x>/cell.go public With* Options may accept. The unexported sealedCellPublisher() method makes CellPublisher unimplementable outside this package — kernel/outbox is the sole entry point via WrapPublisherForCell, which composition roots must call.

AI-robust 评级：

• 字段/赋值层：Hard（sealed marker，外部不可表达 internalCellPublisher 字面量）
• 公开 With* Option 签名层：Medium（CELL-RAW-INFRA-PUBLIC-OPTION-PARAM-01 archtest type-aware 守）

双重防线，参见 ADR 202605101900-adr-cell-raw-infra-sealed-marker §D2。

CellPublisher embeds Publisher so cells can pass the wrapped value directly to outbox.NewDirectEmitter etc. — those constructors accept Publisher and the embedded interface satisfies them transparently, so internal kernel/outbox API does not change.

ref: docs/architecture/202605101900-adr-cell-raw-infra-sealed-marker.md §D1

func WrapPublisherForCell(p Publisher) CellPublisher

type CellWriter ¶

type CellWriter interface {
	Writer
	// contains filtered or unexported methods
}

CellWriter mirrors CellPublisher for the outbox.Writer side: cells/<x> public With* Options accept CellWriter; raw Writer is sealed off behind WrapWriterForCell, callable only from composition roots.

ref: docs/architecture/202605101900-adr-cell-raw-infra-sealed-marker.md §D1

func WrapWriterForCell(w Writer) CellWriter

type ClaimPolicy ¶

type ClaimPolicy uint8

ClaimPolicy controls ConsumerBase behavior when Claimer.Claim() fails. The zero value (ClaimPolicyFailClosed) is the safe default.

const (
	// ClaimPolicyFailClosed (default zero-value): retry Claim with exponential
	// backoff. Safe from duplicates, but consumption stops until the idempotency
	// backend recovers.
	ClaimPolicyFailClosed ClaimPolicy = iota

	// ClaimPolicyFailOpen: single Claim attempt; on error, proceed without
	// idempotency receipt. Avoids total consumer stall, but risks duplicate
	// processing during outage.
	ClaimPolicyFailOpen
)

func (p ClaimPolicy) String() string

func (p ClaimPolicy) Valid() bool

type ConsumerBase ¶

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

ConsumerBase wraps an EntryHandler with two-phase idempotency (Claim/Commit/Release) and exponential backoff retry. DLQ routing is handled by the broker via DLX (DispositionReject triggers Nack requeue=false).

Settlement flows to the Subscriber delivery loop via the second return value of SubscriberHandler: ConsumerBase.Wrap returns (DeliveryOutcome, Settlement) so that Commit/Release can be called after broker Ack/Nack without leaking idempotency types into business code. Business handlers only return the slim HandleResult; ConsumerBase lifts it to DeliveryOutcome, injecting ProcessReason (e.g. "retry_exhausted") as needed.

Lives in kernel/outbox rather than adapters/rabbitmq because the logic is broker-agnostic — it only depends on kernel/idempotency + outbox types — and is wired by runtime/bootstrap alongside any transport that speaks the Subscriber interface. Adapters (rabbitmq, nats, kafka) reuse this middleware unchanged.

Consumer: cg-{ConsumerGroup}-{topic} Idempotency key: {ConsumerGroup}:{event-id}, TTL 24h ACK timing: after business logic returns DispositionAck Retry: transient errors -> retry+backoff / permanent errors -> DispositionReject → DLX

ref: ThreeDotsLabs/watermill message/router.go — router-level retry/poison/dedup ref: MassTransit UseMessageRetry — pipeline middleware at receive endpoint ref: NATS JetStream consumer_config AckWait+MaxDeliver+BackOff — subscriber config

func NewConsumerBase(claimer idempotency.Claimer, config ConsumerBaseConfig, clk clock.Clock) (*ConsumerBase, error)

func (cb *ConsumerBase) AttachObserver(o ConsumerObserver) error

func (cb *ConsumerBase) IsConstructed() bool

func (cb *ConsumerBase) Wrap(sub Subscription, handler EntryHandler) SubscriberHandler

type ConsumerBaseConfig ¶

type ConsumerBaseConfig struct {
	// RetryCount is the maximum number of retries for transient errors.
	// Default: 3.
	RetryCount int

	// RetryBaseDelay is the initial delay for exponential backoff retries.
	// Default: 1s.
	RetryBaseDelay time.Duration

	// IdempotencyTTL is the TTL for idempotency keys (done-key TTL for Claimer).
	// Default: 24h (idempotency.DefaultTTL).
	IdempotencyTTL time.Duration

	// LeaseTTL is the processing-lease TTL for the Claimer backend.
	// If a consumer crashes mid-processing, the lease expires after this
	// duration, allowing another consumer to re-claim.
	// Default: 5m (idempotency.DefaultLeaseTTL). Only used with Claimer.
	LeaseTTL time.Duration

	// ClaimPolicy controls behavior when Claimer.Claim() fails due to
	// infrastructure errors (e.g., Redis down). See ClaimPolicyFailClosed
	// (default zero-value) and ClaimPolicyFailOpen for details.
	ClaimPolicy ClaimPolicy

	// ClaimRetryCount is the max number of Claim() attempts on the fail-closed
	// path before returning DispositionRequeue to the broker.
	// Default: falls back to RetryCount (3).
	ClaimRetryCount int

	// ClaimRetryBaseDelay is the initial backoff delay between Claim() retries.
	// Default: falls back to RetryBaseDelay (1s).
	ClaimRetryBaseDelay time.Duration

	// MaxRetryDelay caps the exponential backoff delay for both claimWithRetry
	// and retryLoop, preventing unbounded growth with large retry counts.
	// Default: 30s.
	MaxRetryDelay time.Duration

	// LeaseRenewalInterval is how often the lease renewal goroutine calls
	// receipt.Extend while the handler is running. Zero falls back to
	// LeaseTTL/3 so the lease is renewed well before it expires.
	// Set to a negative value to disable lease renewal entirely.
	LeaseRenewalInterval time.Duration
}

ConsumerBaseConfig configures ConsumerBase behavior.

func (c *ConsumerBaseConfig) SetDefaults()

type ConsumerObserver ¶

type ConsumerObserver interface {
	ObserveReject(ctx context.Context, cellID, topic, consumerGroup, reason string)
}

ConsumerObserver receives notifications from ConsumerBase when a delivery reaches a terminal Reject disposition. Implementations are provided by runtime/observability/metrics (OutboxRejectCollector) so kernel/outbox stays free of metric backend imports.

ObserveReject is called exactly once per terminal Reject (handler-explicit reject OR retry-budget exhaustion); cellID / topic / consumerGroup are the observability dimensions, reason is a closed-set string drawn from ConsumerRejectReason* constants.

Implementers should NOT include consumerGroup in metric label sets — it is provided here for slog/tracing dimensions but adding it as a Prometheus label expands cardinality multiplicatively (every consumer group × topic × cell × reason combination becomes a unique time-series). See runtime/observability/metrics.OutboxRejectCollector for the sanctioned label set {cell, topic, reason}.

Note: PendingDepthObserver is a SEPARATE interface defined in runtime/outbox (not here) covering the Relay path's pending-depth Gauge. The two interfaces serve different subsystems and are wired independently.

ref: kernel/outbox/relay_collector.go — RelayCollector is the analogous interface for the Relay path; same kernel-defined-runtime-implemented pattern keeps adapter-backend imports out of kernel.

type DeliveryOutcome ¶

type DeliveryOutcome struct {
	Disposition Disposition
	Err         error // optional: logged/observed; nil for success

	// ProcessReason is a low-cardinality kernel/process classification, such
	// as "retry_exhausted". Injected by ConsumerBase, never set by business
	// handlers.
	ProcessReason string

	// SettlementObservers are notified by subscribers after final broker
	// settlement. Appended by subscriber-layer middleware (e.g.
	// WrapConfigEventSubscriber) after ConsumerBase has resolved
	// retry/lease decisions.
	SettlementObservers []SettlementObserver
}

DeliveryOutcome is the subscriber-layer carrier produced by ConsumerBase.Wrap. It extends the slim HandleResult with kernel-injected ProcessReason and SettlementObservers so the subscriber (rabbitmq / eventbus / mqtt) can perform post-settlement notification without any of those concerns leaking into business handler code.

Business handlers return HandleResult. ConsumerBase.Wrap lifts HandleResult → DeliveryOutcome, injecting ProcessReason (e.g. "retry_exhausted") before returning SubscriberHandler to the subscriber layer.

type DemoTxRunner ¶

type DemoTxRunner struct{}

DemoTxRunner is the cell-boundary pass-through TxRunner installed at Cell Init() when the composition root has not provided a real persistence.TxRunner (publisher-only demo assemblies). It implements Nooper, so CheckNotNoop rejects it under DurabilityDurable mode — demo callers that forget to wire a real TxRunner surface an error at Init() time instead of silently losing cellvocab.L2 atomicity guarantees.

func (DemoTxRunner) Noop() bool

func (DemoTxRunner) RunInTx(ctx context.Context, fn func(context.Context) error) error

type DirectEmitter ¶

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

DirectEmitter emits by wrapping entries in the v1 wire envelope and calling Publisher.Publish directly.

func NewDirectEmitter(
	p Publisher, mode DirectPublishFailureMode, mp metrics.Provider, clk clock.Clock, cellID string, opts ...DirectEmitterOption,
) (*DirectEmitter, error)

func (*DirectEmitter) Durable() bool

func (e *DirectEmitter) Emit(ctx context.Context, entry Entry) error

func (e *DirectEmitter) Probes() []healthz.Probe

type DirectEmitterOption ¶

type DirectEmitterOption func(*directEmitterOptions)

DirectEmitterOption configures NewDirectEmitter.

func WithFailOpenRateThreshold(ratio float64) DirectEmitterOption

func WithLogger(l *slog.Logger) DirectEmitterOption

type DirectPublishFailureMode ¶

type DirectPublishFailureMode int

DirectPublishFailureMode controls direct publisher error handling.

const (
	DirectPublishFailClosed DirectPublishFailureMode = iota + 1
	DirectPublishFailOpen
)

type DiscardPublisher ¶

type DiscardPublisher struct {
	// Logger is used for discard warnings. If nil, slog.Default() is used.
	Logger *slog.Logger
	// contains filtered or unexported fields
}

DiscardPublisher is an explicit publisher sink for tests and demos. Unlike NoopWriter, it affects direct-publish flows rather than durable outbox writes. It is an explicit opt-in sink, not a default runtime fallback.

Use DiscardPublisher when:

• Unit testing Cells that require an outbox.Publisher dependency
• Running demo/example code without a message broker

Publish logs a structured warning and discards the payload. The warning ensures discard behavior is visible in logs.

ref: go-logr zero-value safe (if sink == nil), slog DiscardHandler

func (d *DiscardPublisher) Close(_ context.Context) error

func (d *DiscardPublisher) DiscardCount() uint64

func (*DiscardPublisher) Noop() bool

func (d *DiscardPublisher) Publish(_ context.Context, topic string, _ []byte) error

type Disposition ¶

type Disposition uint8

Disposition describes the broker-level action for a consumed message.

• DispositionAck: message processed successfully (or duplicate); broker may discard.
• DispositionRequeue: temporary failure / shutdown; broker should redeliver.
• DispositionReject: permanent failure; broker routes to dead-letter exchange.

const (
	// DispositionAck indicates the message was processed successfully (or is a
	// safe-to-skip duplicate); broker may discard.
	//
	// IMPORTANT: iota+1 ensures the zero value (0) is NOT a valid Disposition.
	// A forgotten/uninitialised HandleResult.Disposition will NOT silently ACK.
	DispositionAck     Disposition = iota + 1 // = 1
	DispositionRequeue                        // NACK+requeue -- transient / shutdown
	DispositionReject                         // NACK+no-requeue -- permanent failure -> DLX
)

func (d Disposition) String() string

func (d Disposition) Valid() bool

type DurabilityMode ¶

type DurabilityMode int

DurabilityMode declares whether an assembly runs in demo or durable mode. Cells use this to reject noop/test implementations at Init() time, preventing "pseudo-success" assemblies in production.

The zero value is intentionally invalid (unset), forcing callers to explicitly choose DurabilityDemo or DurabilityDurable. ref: Vault StoredKeysInvalid=0, gRPC InvalidSecurityLevel=0, net/http SameSite iota+1

const (
	// DurabilityDemo allows noop implementations (NoopWriter, NoopTxRunner,
	// DiscardPublisher). Used by examples/ and unit tests.
	DurabilityDemo DurabilityMode = iota + 1

	// DurabilityDurable rejects noop implementations at Init() time.
	// Used by production assemblies (e.g., cmd/corebundle).
	DurabilityDurable
)

func (m DurabilityMode) String() string

type DurabilityReporter ¶

type DurabilityReporter interface {
	Durable() bool
}

DurabilityReporter is an optional interface Emitter implementations may expose so callers (typically Cell boundaries) can query whether this emitter is backed by durable (transactional outbox) sinks. Emitters that do not implement DurabilityReporter are treated as non-durable by callers — the safe default for direct-publish and noop paths.

ref: kernel/cell.EmitterOutcome.Durable — the primary consumer; Cells use the reported value to decide whether optional slices (e.g. rbacassign) upgrade from L0 to L2. ref: github.com/ThreeDotsLabs/watermill message/router.go — `disabledPublisher` pattern; an explicit typed indicator lets callers branch on capability without runtime type switches.

type Emitter ¶

type Emitter interface {
	Emit(ctx context.Context, entry Entry) error
}

Emitter emits an outbox entry either by writing it to a durable outbox or by directly publishing its canonical wire envelope.

Implementations may optionally satisfy DurabilityReporter to expose whether their sink is backed by durable (transactional outbox) storage. Callers that need to decide L2/L0 slice upgrades should use ReportDurable, which returns false for any Emitter that does not implement the optional interface.

func NewNoopEmitter() Emitter

type EmitterConfig ¶

type EmitterConfig struct {
	CellID            string
	Mode              DurabilityMode
	Publisher         Publisher
	OutboxWriter      Writer
	TxRunner          persistence.TxRunner
	Logger            *slog.Logger
	DirectPublishMode DirectPublishFailureMode
	// MetricsProvider is REQUIRED when DurabilityDemo mode resolves to a DirectEmitter.
	MetricsProvider metrics.Provider
}

EmitterConfig bundles the inputs needed to resolve an outbox.Emitter for a Cell according to DurabilityMode rules.

ResolveEmitter accepts nil TxRunner and treats it as absent (no pairing with a real writer is possible). Callers in Durable mode must inject a real TxRunner explicitly; Demo mode tolerates nil and any Nooper implementation.

MetricsProvider is required for DirectEmitter resolution paths. Pass metrics.NopProvider{} explicitly in tests.

The clock is NOT a field here — it is a mandatory positional parameter to ResolveEmitter (see ADR 202605270000-adr-clock-positional-injection-funnel).

ref: outbox.CheckNotNoop — sibling durability guard.

type EmitterOutcome ¶

type EmitterOutcome struct {
	Emitter Emitter
	Durable bool
}

EmitterOutcome reports the resolved emitter and whether it is durable (backed by a real writer+txRunner). Cells use Durable to decide whether optional slices (e.g., rbacassign) can activate L2 outbox publication.

func ResolveEmitter(clk clock.Clock, cfg EmitterConfig) (EmitterOutcome, error)

type Entry ¶

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

Entry represents a single outbox record to be published.

Entry is sealed construction (issue #1229): all fields are unexported and the only ways to obtain an Entry are NewEntry (the producer constructor), UnmarshalEnvelope (the wire-decode funnel), and EntryScan.ToEntry (the storage-adapter reconstruction funnel) — all three in package kernel/outbox. A composite literal `outbox.Entry{...}` outside this package is a compile error (type-system Hard), which is the point: OccurredAt is always clock-derived, Observability and Principal are always ctx-injected at the trust boundary, and no producer can forge or omit them. Consumers read state through the value-receiver getters below.

ref: OpenTelemetry SpanContext / OIDC Principal — typed carriers of trace and principal identity, kept distinct from producer-owned business Metadata.

func NewEntry(clk clock.Clock, ctx context.Context, eventType string, payload []byte, opts ...EntryOption) (Entry, error)

func UnmarshalEnvelope(topic string, raw []byte) (Entry, error)

func (e Entry) AggregateID() string

func (e Entry) AggregateType() string

func (e Entry) CreatedAt() time.Time

func (e Entry) EventID() string

func (e Entry) EventType() string

func (e Entry) FailurePolicy() FailurePolicy

func (e Entry) ID() string

func (e Entry) Metadata() map[string]string

func (e Entry) Observability() ObservabilityMetadata

func (e Entry) OccurredAt() time.Time

func (e Entry) Payload() []byte

func (e Entry) Principal() PrincipalMetadata

func (e Entry) RestoreContext(ctx context.Context) context.Context

func (e Entry) RoutingTopic() string

func (e Entry) Stream() string

func (e Entry) Topic() string

func (e Entry) Validate() error

type EntryHandler ¶

type EntryHandler func(context.Context, Entry) HandleResult

EntryHandler is the business handler signature. Business handlers return a HandleResult that declares the intended broker disposition. Settlement is not visible to business handlers — it is delivered via SubscriberHandler, which is the Subscriber-layer interface (not business layer).

type EntryOption ¶

type EntryOption func(*Entry)

EntryOption customizes an Entry under construction. Options may set the aggregate identity, explicit topic, business metadata, failure policy, and (for domain-time / reconstruction needs) the createdAt / occurredAt stamps.

There is deliberately no WithPrincipal / WithObservability option: those two families are injected from the construction context by NewEntry alone, so a producer cannot forge or override them — the trust boundary is the ctx, not the call site.

func WithAggregateID(id string) EntryOption

func WithAggregateType(t string) EntryOption

func WithCreatedAt(t time.Time) EntryOption

func WithFailurePolicy(p FailurePolicy) EntryOption

func WithID(id string) EntryOption

func WithMetadata(m map[string]string) EntryOption

func WithOccurredAt(t time.Time) EntryOption

func WithTopic(t string) EntryOption

type EntryScan ¶

type EntryScan struct {
	ID            string
	AggregateID   string
	AggregateType string
	EventType     string
	Topic         string
	Payload       []byte
	CreatedAt     time.Time
	OccurredAt    time.Time
	Metadata      map[string]string
	Observability ObservabilityMetadata
	Principal     PrincipalMetadata
	FailurePolicy FailurePolicy
}

EntryScan is the sanctioned reconstruction funnel for storage adapters that rebuild an Entry from persisted columns (the relay claim path scans DB rows into it, then calls ToEntry). It exists because Entry fields are unexported (sealed construction, issue #1229): an adapter in another package cannot scan `rows.Scan(&entry.id, ...)` nor build a composite literal, so the kernel exposes this typed, validated rebuild path instead.

EntryScan deliberately carries exported fields — it is a scan target, not a producer surface. It has NO wire role (wireMessage remains the only json.Unmarshal funnel) and NO clock/ctx injection (the persisted row already carries createdAt/occurredAt/observability/principal). ToEntry runs the full Entry.Validate, so a corrupt row cannot reconstruct an invalid Entry.

EntryScan is asserted to be the only exported Entry-reconstruction mirror by archtest OUTBOX-ENTRY-SEALED-CONSTRUCTION-01; any second exported struct that reconstitutes an Entry would defeat the seal.

func (s EntryScan) ToEntry() (Entry, error)

type FailurePolicy ¶

type FailurePolicy int

FailurePolicy expresses how an Emitter handles publisher-side failures for a particular Entry. Cells default their DirectEmitter to DirectPublishFailClosed (k8s apiserver audit model); individual entries opt into FailOpen for observational / non-critical sinks.

Reserved for direct-publish paths: WriterEmitter (transactional outbox) surfaces errors through the surrounding transaction and does not consult FailurePolicy — the field is ignored there by design.

const (
	// FailurePolicyDefault falls through to Emitter ctor default. Zero value.
	FailurePolicyDefault FailurePolicy = iota
	// FailurePolicyFailOpen drops on publisher failure (log + counter), returns nil.
	FailurePolicyFailOpen
	// FailurePolicyFailClosed surfaces publisher failure to caller.
	FailurePolicyFailClosed
)

func (p FailurePolicy) Resolve(ctorDefault DirectPublishFailureMode) DirectPublishFailureMode

type HandleResult ¶

type HandleResult struct {
	Disposition Disposition
	Err         error // optional: logged/observed; nil for success
}

HandleResult carries the business handler's processing outcome. Business handlers return HandleResult via EntryHandler — it contains only what the handler author decides: the disposition and an optional error.

ProcessReason and SettlementObservers are subscriber-layer concerns and live in DeliveryOutcome (the type ConsumerBase.Wrap produces). Business handlers never need to set those fields; ConsumerBase injects ProcessReason (e.g. "retry_exhausted") and the subscriber layer appends SettlementObservers.

func Ack() HandleResult

func Reject(err error) HandleResult

func Requeue(err error) HandleResult

type NoopRelayCollector ¶

type NoopRelayCollector struct{}

NoopRelayCollector is a no-op implementation of RelayCollector. Used when metrics collection is disabled (nil Metrics in RelayConfig).

func (NoopRelayCollector) RecordBatchSize(_ context.Context, _ int)

func (NoopRelayCollector) RecordCleanup(_ context.Context, _, _ int64)

func (NoopRelayCollector) RecordPollCycle(_ context.Context, _ PollCycleResult)

func (NoopRelayCollector) RecordReclaim(_ context.Context, _ int64)

type NoopWriter ¶

type NoopWriter struct{}

NoopWriter is an explicit outbox writer sink for tests and demos. It validates entries like a real writer, then discards them instead of persisting anything. It is not a production durability mechanism.

Use NoopWriter when:

• Unit testing Cells that require an outbox.Writer dependency
• Running demo/example code without a database

NoopWriter still validates entries via Entry.Validate(), unlike a zero-value or nil writer. This catches schema errors during development.

func (NoopWriter) Noop() bool

func (NoopWriter) Write(_ context.Context, entry Entry) error

func (NoopWriter) WriteBatch(_ context.Context, entries []Entry) error

type Nooper ¶

type Nooper interface {
	Noop() bool
}

Nooper is a marker interface for test/demo-only implementations. Types that implement Nooper are rejected by CheckNotNoop when the assembly runs in DurabilityDurable mode.

Kernel noop types (outbox.NoopWriter, outbox.DiscardPublisher) implement this interface; cells in Demo mode may register their own Nooper TxRunners.

type NopConsumerObserver ¶

type NopConsumerObserver struct{}

NopConsumerObserver is the default ConsumerObserver used when no metrics provider is wired. ObserveReject is a no-op.

func (NopConsumerObserver) ObserveReject(_ context.Context, _, _, _, _ string)

type ObservabilityMetadata ¶

type ObservabilityMetadata struct {
	TraceID       idutil.SafeID `json:"traceId,omitempty"`
	TraceParent   string        `json:"traceParent,omitempty"`
	RequestID     idutil.SafeID `json:"requestId,omitempty"`
	CorrelationID idutil.SafeID `json:"correlationId,omitempty"`
}

ObservabilityMetadata carries cross-async tracing context that the gocell observability bridge owns. Producers MUST NOT populate these fields directly — the writer bridge (InjectObservabilityFromContext) fills them from context at persistence time. Consumer middleware (RestoreToContext) reads them back into handler context.

The typed field replaces the pre-PR246-FU1 string-key metadata bridge (Merge*/IsReserved*) which allowed producers to forge reserved keys via entry.Metadata["trace_id"] = "evil" before the observability layer got a chance to overwrite them. With a typed field the forgery surface is removed by construction — the observability system and the business metadata map no longer share a key namespace.

ref: OpenTelemetry SpanContext -- typed carrier of trace identity across transport boundaries, kept distinct from application attributes. Adopted: separate struct for system-owned fields vs. producer-owned Metadata map. Deviated: only 4 fields (no sampled flag, no traceFlags struct) to keep the async boundary narrow; TraceParent is the W3C canonical form.

TraceID/RequestID/CorrelationID use idutil.SafeID — UnmarshalJSON fail-closes on unsafe characters at wire boundary (CWE-117). TraceParent keeps `string`: its 55-byte W3C format has its own validator (validTraceParent) and is not in the IsSafeID character set.

func ContextObservability(ctx context.Context) ObservabilityMetadata

func (o ObservabilityMetadata) IsZero() bool

func (o ObservabilityMetadata) RestoreToContext(ctx context.Context) context.Context

func (o ObservabilityMetadata) Validate() error

type PermanentError ¶

type PermanentError struct {
	Err error
}

PermanentError wraps an error to indicate it should not be retried and should be routed to the dead-letter queue. This is a domain concept alongside Disposition and HandleResult.

ref: Temporal SDK temporal.ApplicationError (NonRetryable flag in SDK core); Watermill delegates error classification to middleware -- GoCell makes it explicit at the kernel level so WrapLegacyHandler and InMemoryEventBus can detect it without depending on adapter-layer types.

func NewPermanentError(err error) *PermanentError

func (e *PermanentError) Error() string

func (e *PermanentError) Unwrap() error

type PollCycleResult ¶

type PollCycleResult struct {
	// Published / Retried / Dead are the canonical outcomes from the publish
	// and writeback phases. Skipped covers `MarkPublished updated=false`
	// (the entry was reclaimed mid-flight before MarkPublished could win).
	// Lost covers the same condition for failure writebacks (Mark{Retry,Dead}
	// updated=false): the lease lost mid-flight while the publisher was
	// reporting an error, so the failure must NOT be counted as retried/dead
	// — the new lease owner will report the canonical outcome.
	Published, Retried, Dead, Skipped, Lost int
	ClaimDur, PublishDur, WriteBackDur      time.Duration
}

PollCycleResult captures the outcome of a single relay poll cycle. Used by RelayCollector.RecordPollCycle to avoid a long parameter list and to support future extensions without breaking the interface.

type PrincipalMetadata ¶

type PrincipalMetadata struct {
	ActorID   idutil.SafeID `json:"actorId,omitempty"`
	SubjectID idutil.SafeID `json:"subjectId,omitempty"`
	TenantID  idutil.SafeID `json:"tenantId,omitempty"`
	SessionID idutil.SafeID `json:"sessionId,omitempty"`
}

PrincipalMetadata carries cross-async principal identity that the gocell observability bridge owns. Producers MUST NOT populate these fields directly — NewEntry fills them from the construction context (the single injection trust boundary). Consumer middleware (RestoreToContext) reads them back into handler context. It mirrors ObservabilityMetadata field-for -field (sealed-construction sibling, issue #1229).

OAuth/OIDC semantics:

• ActorID = impersonator — the principal actually triggering the action (token "act.sub" claim). In non-impersonation flows equals SubjectID.
• SubjectID = subject-of-record — the user the action is performed on behalf of (token "sub" claim).
• TenantID = tenant boundary identifier (multi-tenant deployments).
• SessionID = session identifier (server-side session binding).

The typed field prevents producers from forging principal IDs via entry.Metadata["actor_id"] = "evil" — the two namespaces are physically separate, and Entry.Validate rejects ReservedMetadataKeys to keep producers honest at write time. Mirror of ObservabilityMetadata's rationale for trace identity.

ref: OpenID Connect Core 1.0 §5.1 ("sub"), RFC 8693 §4.1 ("act") — typed carrier of principal identity, kept distinct from observability (trace) and business (metadata) namespaces.

All four fields use idutil.SafeID — UnmarshalJSON fail-closes on unsafe characters at wire boundary (CWE-117 defense).

The field set, JSON tags, and method set are frozen by archtest PRINCIPAL-SEALED-FIELD-FROZEN-01.

func ContextPrincipal(ctx context.Context) PrincipalMetadata

func (p PrincipalMetadata) IsZero() bool

func (p PrincipalMetadata) RestoreToContext(ctx context.Context) context.Context

func (p PrincipalMetadata) Validate() error

type ProviderRelayCollectorConfig ¶

type ProviderRelayCollectorConfig struct {
	// PollBuckets overrides DefaultRelayPollBuckets; zero value uses defaults.
	PollBuckets []float64
	// BatchBuckets overrides DefaultRelayBatchBuckets; zero value uses defaults.
	BatchBuckets []float64
}

ProviderRelayCollectorConfig customizes metric naming / bucketing. Zero value is acceptable and produces defaults.

type Publisher ¶

type Publisher interface {
	Publish(ctx context.Context, topic string, payload []byte) error
	// Close terminates the publisher and releases resources. The ctx parameter
	// allows callers to share a shutdown budget (e.g., bootstrap shutCtx).
	Close(ctx context.Context) error
}

Publisher sends events to the message broker.

ref: uber-go/fx app.go Lifecycle.Append OnStop(ctx) — ContextCloser pattern adopted so bootstrap teardown can share a unified shutCtx budget across all managed publishers.

type Relay ¶

type Relay interface {
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
}

Relay polls unpublished outbox entries and publishes them.

type RelayCollector ¶

type RelayCollector interface {
	// RecordPollCycle records a completed poll cycle with outcome counts and
	// per-phase durations. Called once per pollOnce invocation after writeBack.
	RecordPollCycle(ctx context.Context, r PollCycleResult)

	// RecordBatchSize records the number of entries claimed in a poll cycle.
	// Called even when the batch is empty (size=0) to capture idle cycles.
	RecordBatchSize(ctx context.Context, size int)

	// RecordReclaim records the number of stale entries reclaimed back to
	// pending (or dead-lettered). Called once per reclaimStale tick **only
	// when count > 0** — idle reclaim sweeps are not observed (the metric
	// is a recovery counter, not a tick frequency gauge). This contrasts
	// with RecordBatchSize which is also called on size==0 cycles so
	// dashboards can detect a totally idle relay.
	RecordReclaim(ctx context.Context, count int64)

	// RecordCleanup records the number of entries removed during periodic
	// cleanup, split by original status (published vs dead-lettered).
	RecordCleanup(ctx context.Context, publishedDeleted, deadDeleted int64)
}

RelayCollector records outbox relay operational metrics. Implementations must be safe for concurrent use. Zero counts are valid inputs; implementations should handle them gracefully (e.g. skip counter increments for zero values).

The interface is intentionally in kernel/outbox (not runtime/) so that adapters/postgres can depend on it without pulling in runtime/ packages.

ref: Temporal client.Options{MetricsHandler} — inject-at-construction pattern ref: Watermill components/metrics — publish_time_seconds, subscriber_messages_received_total ref: Debezium JMX — MilliSecondsBehindSource, max.batch.size, DLQ count

func NewProviderRelayCollector(p metrics.Provider, cellID string, opts ...ProviderRelayCollectorConfig) (RelayCollector, error)

type SerialInOrderGuarantor ¶

type SerialInOrderGuarantor interface {
	GuaranteesSerialInOrderDelivery() bool
}

SerialInOrderGuarantor is an optional Subscriber-implementer extension contract. Business handlers must never reference this interface.

A Subscriber that implements it and returns true ASSERTS that it delivers a single subscription's stream strictly serially and in order: at most one delivery is dispatched at a time and the next is handed off only after the previous handler returns. This is the precondition L3 CQRS projection subscriptions REQUIRE — projection exactly-once rests on a monotonic checkpoint (applyOne skips any event whose stream position ≤ the stored checkpoint), which is only SOUND under serial in-order delivery. Under concurrent delivery (e.g. the AMQP subscriber dispatching one goroutine per delivery with prefetch>1) a higher position can commit the checkpoint before a lower position is applied, silently dropping the lower event's distinct apply (a projection gap). See kernel/projection/doc.go "Ordering precondition" and ADR docs/architecture/202605261620-adr-cqrs-projection-lifecycle-harness.md §6 threat row 4.

Scope of the guarantee: it holds for a SINGLE subscriber on a given (consumerGroup, topic). A projection's consumer group is derived as "<cellID>-<projectionID>" and the bootstrap drain registers exactly one subscription for it, so the single-subscriber precondition is structurally satisfied. The guarantee does NOT extend to multiple competing subscribers sharing one consumer group (e.g. the in-memory bus round-robins across them on independent goroutines).

fail-closed by absence: a Subscriber that does NOT implement this interface is treated as NOT serial. The projection drain rejects wiring a projection onto such a transport. A concurrent transport therefore opts OUT simply by not implementing the method — and any future transport that forgets to implement it is auto-rejected for projections rather than silently unsafe.

Guarded by archtest PROJECTION-SERIAL-DELIVERY-ENFORCEMENT-01, which freezes this method set and pins the implementer set to exactly the qualifying transports. A new transport that implements this marker MUST also be added to that archtest's implementer golden (serialGuarantorSoleImpl), or sub-rule B fails CI — by design, so adding a serial transport is a deliberate update.

type Settlement ¶

type Settlement interface {
	Commit(ctx context.Context) error
	Release(ctx context.Context) error
}

Settlement is the broker-side commit/release handle that ConsumerBase delivers alongside HandleResult via SubscriberHandler. Subscriber implementations (adapters/rabbitmq, runtime/eventbus) call Commit before broker Ack and Release after broker Nack to finalize idempotency state.

Business handlers MUST NOT see Settlement — it is part of SubscriberHandler (framework Subscriber-side hand-off), not EntryHandler (business return signature). The compile-time separation supersedes the prior HANDLER-RECEIPT-WRITE-01 archtest gate, which guarded a HandleResult.Receipt field that no longer exists (see 029 #12 K#12 PR-V1-OUTBOX-RECEIPT-EXTRACT).

kernel/idempotency.Receipt implicitly satisfies Settlement: its Commit and Release methods match this interface, and its Extend method is hidden in the Settlement view (renewal stays inside ConsumerBase).

ref: nats-io/nats.go jetstream/message.go Msg interface — settle ops

(Ack/Nak/Term/InProgress) abstracted into a single interface

ref: IBM/sarama consumer_group.go ConsumeClaim(session ConsumerGroupSession,

claim ConsumerGroupClaim) — settle handle as explicit method parameter
rather than a ctx side-channel

type SettlementObservation ¶

type SettlementObservation struct {
	Entry         Entry
	Disposition   Disposition
	Result        SettlementResult
	ProcessReason string
	Err           error
}

SettlementObservation is emitted by subscribers after the final delivery settlement action is known.

type SettlementObserver ¶

type SettlementObserver interface {
	ObserveSettlement(context.Context, SettlementObservation)
}

SettlementObserver records a post-settlement observation. Implementations must be non-blocking or bounded; subscriber delivery loops call observers on the hot path after broker settlement.

type SettlementObserverFunc ¶

type SettlementObserverFunc func(context.Context, SettlementObservation)

SettlementObserverFunc adapts a function into a SettlementObserver.

func (f SettlementObserverFunc) ObserveSettlement(ctx context.Context, obs SettlementObservation)

type SettlementResult ¶

type SettlementResult string

SettlementResult describes whether the subscriber completed the final broker settlement action. It intentionally observes the boundary after Commit, Ack/Nack, and receipt release decisions, not the handler's process result.

const (
	SettlementResultSuccess        SettlementResult = "success"
	SettlementResultRetryExhausted SettlementResult = "retry_exhausted"
	SettlementResultCommitFailed   SettlementResult = "commit_failed"
	SettlementResultAckFailed      SettlementResult = "ack_failed"
	SettlementResultNackFailed     SettlementResult = "nack_failed"
)

func RejectSettlementResult(outcome DeliveryOutcome) SettlementResult

type State ¶

type State uint8

State is the publication state of a transactional-outbox entry.

Single-source role ¶

State is the single Go-side source of the wire/DB status strings: String() produces the exact value bound into the `status` column, so adapters/postgres and runtime/outbox must use the enum rather than bare string literals (enforced by the OUTBOX-STATE-LITERAL-BAN-01 archtest).

Transition table ¶

stateTransitions documents the complete legal state graph and is validated by the OUTBOX-STATE-TRANSITION-COMPLETENESS-01 archtest (every State const must be a key). It is NOT the runtime enforcement source for SQL operations:

• relay settlement (writeBackOne / handleFailedEntry): three paths where the relay explicitly decides claiming→published, claiming→dead, or claiming→pending. Each calls Transition as a Mark↔target cross-check: a function that calls MarkPublished must also Transition to StatePublished, catching copy-paste mismatches at CI time (OUTBOX-STATE-TRANSITION-GUARD-01). These calls are not correctness guards — the constant arguments mean they can never fail at runtime — but they serve as machine-readable pairing assertions.

• store ClaimPending (pending→claiming) and ReclaimStale (claiming→pending/dead): no Mark* pairing exists for these transitions. Their correctness is enforced by SQL CAS predicates (WHERE status='claiming'/'pending'), conformance behavior tests, and LITERAL-BAN. Duplicating Transition calls there would add no pairing-check value and would be misleading.

• SQL-text ↔ table-edge machine binding: SQL args are ...any, so the wire string cannot be type-constrained at the binding site. This is a true Medium ceiling, explicitly accepted; no fragile static SQL↔edge parser is attempted.

ref: kernel/saga/status.go (transition-table + Transition pattern).

const (
	// StatePending: awaiting publish (initial state and the retry back-off
	// state).
	//
	// IMPORTANT: iota+1 ensures the zero value (0) is NOT a valid State, so a
	// forgotten/uninitialised State will not silently appear as pending.
	StatePending   State = iota + 1 // = 1
	StateClaiming                   // locked by a relay instance, publish in progress
	StatePublished                  // delivered to broker (terminal)
	StateDead                       // exceeded MaxAttempts, needs manual intervention (terminal)
)

func ParseState(s string) (State, error)

func (s State) CanTransitionTo(target State) bool

func (s State) IsTerminal() bool

func (s State) String() string

func (s State) Valid() bool

func (s State) ValidTransitions() []State

type Subscriber ¶

type Subscriber interface {
	// Setup pre-declares broker topology (exchanges, queues, bindings) for the
	// given subscription before Subscribe is called. Callers SHOULD await Ready
	// before publishing to ensure messages are queued deterministically.
	// In-memory implementations MUST return nil immediately.
	Setup(ctx context.Context, sub Subscription) error

	// Ready returns a channel that is closed when the subscription is ready to
	// consume. In-memory implementations SHOULD return an already-closed channel.
	Ready(sub Subscription) <-chan struct{}

	// Subscribe registers a handler for the given subscription and blocks until
	// ctx is canceled or an unrecoverable error occurs.
	//
	// Subscription.ConsumerGroup identifies the logical consumer group.
	// Subscribers sharing the same group compete for messages (load-balanced);
	// different groups each receive a full copy (fanout).
	//
	// handler is SubscriberHandler so the Subscriber can receive Settlement
	// alongside DeliveryOutcome without idempotency types leaking into business
	// code. Callers that hold an EntryHandler and want the full business
	// pipeline (middleware chain + ConsumerBase idempotency) should use
	// SubscriberWithMiddleware.SubscribeEntry instead of lifting manually.
	Subscribe(ctx context.Context, sub Subscription, handler SubscriberHandler) error

	// Close terminates all active subscriptions and releases resources.
	// The ctx parameter allows callers to share a shutdown budget.
	//
	// ref: uber-go/fx app.go Lifecycle.Append OnStop(ctx context.Context) error
	// ref: ThreeDotsLabs/watermill message/pubsub.go Subscriber.Close()
	Close(ctx context.Context) error
}

Subscriber consumes events from a topic.

ref: ThreeDotsLabs/watermill message/pubsub.go Subscriber interface Adopted: Close() for clean shutdown; topic-based subscription model. Deviated: callback-based EntryHandler instead of channel-based (<-chan *Message) to align with GoCell's ConsumerBase pattern and simplify consumer lifecycle. Extended: Setup/Ready split mirrors Watermill Router's setup-before-run pattern, eliminating the 500ms startup-timeout heuristic in eventrouter (Commit 3).

ref: Kafka sarama ConsumerGroup -- consumerGroup isolates consumption; same group competes, different groups each get a full copy (fanout). ref: go-micro broker.SubscribeOptions.Queue -- same concept, different name.

type SubscriberHandler ¶

type SubscriberHandler func(ctx context.Context, entry Entry) (DeliveryOutcome, Settlement)

SubscriberHandler is the Subscriber-layer handler type that ConsumerBase.Wrap returns and Subscriber.Subscribe accepts. It extends EntryHandler with a Settlement return value so Subscriber implementations can call Settlement.Commit before broker Ack and Settlement.Release after broker Nack without any idempotency types leaking into business code.

The return type is DeliveryOutcome (not HandleResult) because ConsumerBase.Wrap injects kernel-level ProcessReason ("retry_exhausted") and subscriber-layer SettlementObservers that business handlers must not set. Business handlers return slim HandleResult; ConsumerBase lifts it to DeliveryOutcome.

Settlement may be nil when ConsumerBase has no idempotency state (fail-open claim error, ClaimDone, ClaimBusy short-circuit). Subscribers MUST nil-check before calling Commit/Release.

Business handlers use EntryHandler — they never see Settlement. The type separation provides compile-time enforcement without archtest gates.

ref: IBM/sarama consumer_group.go ConsumeClaim(session ConsumerGroupSession,

claim ConsumerGroupClaim) — settle handle as explicit method parameter

ref: nats-io/nats.go jetstream/message.go Msg interface (Ack/Nak/Term)

type SubscriberIntakeStopper ¶

type SubscriberIntakeStopper interface {
	StopIntake(ctx context.Context) error
}

SubscriberIntakeStopper is a Subscriber-implementer extension contract. Business handlers should never reference this interface.

Subscribers that implement it can stop accepting new deliveries while still processing in-flight ones, enabling a two-phase drain during shutdown (stop intake → wait in-flight handlers). Router.Close calls StopIntake on subscribers implementing this interface before canceling the run context.

Subscribers that do not implement this interface will fall back to the legacy single-phase close behavior (cancel context only).

StopIntake MUST be idempotent. It SHOULD be safe to call concurrently with in-flight Subscribe calls.

Errors are best-effort: callers (e.g., eventrouter.Router.Close) log a warning and proceed to context cancellation regardless. Return an error only for caller-observable failures that operators would want to see in logs.

ref: ThreeDotsLabs/watermill message/router.go (closingInProgressCh two-phase barrier) ref: uber-go/fx app.go shutdown semantics (run ctx vs stop ctx separation)

type SubscriberWithMiddleware ¶

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

SubscriberWithMiddleware wraps a Subscriber with a business middleware chain and a required ConsumerBase for idempotency/retry.

The SubscribeEntry method is the primary entry point for callers that hold an EntryHandler (e.g., eventrouter.Router). It orchestrates:

1. Business middleware chain (EntryHandler → EntryHandler): applied via slices.Backward(s.Middleware) — [0] is outermost (first to see each delivery, last to return), [len-1] is innermost (adjacent to ConsumerBase). This is the opposite of chi/Kratos forward composition where [0] is applied last.
2. ConsumerBase.Wrap (EntryHandler → SubscriberHandler): injects idempotency Claim/Commit/Release and retry logic. SubscribeEntry fails fast when ConsumerBase is nil.
3. Observability restore (entry.Observability → ctx): built-in OUTERMOST wrapper applied inside Inner.Subscribe so every layer sees a ctx populated with trace_id/request_id/correlation_id.
4. Inner.Subscribe: the actual broker subscription.

Observability restoration is a paired invariant with Entry.InjectObservability FromContext on the producer side — there is no kill-switch. Raw delivery without restore (integration testing only) should invoke Inner.Subscribe directly.

SubscriberWithMiddleware does NOT implement the Subscriber interface: it exposes only SubscribeEntry (EntryHandler) as the single public entry point. This prevents the lift→discard footgun where callers could assign *SubscriberWithMiddleware to outbox.Subscriber and bypass the business middleware chain. Adapter-layer tests that need raw SubscriberHandler delivery must call the inner subscriber directly.

Construction is funneled through NewSubscriberWithMiddleware: fields are unexported so struct literals cannot create an invalid (nil-inner / nil-ConsumerBase) instance. This mirrors the OUTBOX-SERVICE-01 ctor fail-fast pattern that 12 outbox-bound services already follow.

ref: ThreeDotsLabs/watermill message/router.go — handleMessage applies middleware then calls handler; Ack/Nack monopoly stays in router. ref: go-kratos/kratos — middleware chain on business Request/Reply; transport/grpc monopolizes gRPC status. ref: IBM/sarama consumer_group.go — ConsumeClaim owns MarkMessage; business handler in ConsumeClaim body never calls MarkMessage directly.

func NewSubscriberWithMiddleware(inner Subscriber, cb *ConsumerBase, mw ...SubscriptionMiddleware) (*SubscriberWithMiddleware, error)

func (s *SubscriberWithMiddleware) Close(ctx context.Context) error

func (s *SubscriberWithMiddleware) Ready(sub Subscription) <-chan struct{}

func (s *SubscriberWithMiddleware) Setup(ctx context.Context, sub Subscription) error

func (s *SubscriberWithMiddleware) StopIntake(ctx context.Context) error

func (s *SubscriberWithMiddleware) SubscribeEntry(ctx context.Context, sub Subscription, h EntryHandler) error

type Subscription ¶

type Subscription struct {
	// Topic is the broker routing key. Required.
	Topic string

	// ConsumerGroup is the logical consumer group identity. Required.
	// It forms the idempotency key namespace: "{ConsumerGroup}:{entry.ID}".
	// Subscribers sharing the same ConsumerGroup compete for messages;
	// different groups each receive a full copy (fanout).
	ConsumerGroup string

	// CellID is the observability owner label. Required.
	//
	// CellID is the single source of truth for the cell owning this
	// subscription — metrics, slog fields, and trace span attributes use it
	// to route subscriber telemetry to the right owner. The value is set by
	// codegen (contractgen NewSubscription / cellgen cell.tmpl) from the
	// cell's metadata.ID at compile time; there is no runtime fallback to
	// ConsumerGroup. ConsumerGroup is a broker partition key + idempotency
	// namespace, conceptually orthogonal to ownership.
	//
	// ref: ADR docs/architecture/202605111000-adr-subscription-cellid-mandatory.md
	CellID string

	// SliceID is an optional observability owner label. Codegen fills it
	// from slice.yaml contractUsages[role=subscribe] via cellgen when a cell
	// wants per-slice consumer metrics.
	SliceID string

	// ContractID/Kind/Transport identify the contract bound to this
	// subscription. They are intentionally primitive strings rather than a
	// contractspec.ContractSpec to keep kernel/outbox independent of
	// kernel/wrapper (wrapper already imports outbox for EntryHandler).
	//
	// Runtime eventrouter fills these fields for AddContractHandler
	// registrations so subscription middleware can install the contract span
	// outside ConsumerBase while still after observability metadata restore.
	ContractID        string
	ContractKind      string
	ContractTransport string
}

Subscription describes the full identity of a single subscription intent. It is the first-class object passed through the middleware chain, ensuring every layer (idempotency, observability, retry) sees Topic, ConsumerGroup, CellID, and SliceID without information loss at middleware boundaries.

ref: ThreeDotsLabs/watermill message/router.go handler context injection ref: MassTransit ConsumeContext — full identity traverses the entire pipeline

func (s Subscription) IdempotencyNamespace() string

func (s Subscription) ObservabilityID() string

func (s Subscription) Validate() error

type SubscriptionMiddleware ¶

type SubscriptionMiddleware func(sub Subscription, next EntryHandler) EntryHandler

SubscriptionMiddleware is the event-consumer middleware type. It carries the full Subscription identity (Topic + ConsumerGroup + CellID), so middleware can route, log, and observe per-subscription rather than per-topic.

Apply in order: [0] is outermost, [len-1] is innermost.

The next handler is EntryHandler (business signature). Middleware authors work entirely in the business domain and do not see Settlement — it is an internal concern of the Subscriber layer (adapters/rabbitmq, runtime/eventbus). ConsumerBase is field-injected into SubscriberWithMiddleware and applied as an explicit EntryHandler→SubscriberHandler conversion boundary after the business middleware chain, not as a SubscriptionMiddleware entry.

ref: ThreeDotsLabs/watermill message/router.go — HandlerMiddleware operates on Message→Message; Ack/Nack is the router's exclusive responsibility. ref: go-kratos/kratos middleware — middleware operates on Request→Reply; gRPC status (settle equivalent) is the transport's exclusive responsibility.

type Writer ¶

type Writer interface {
	// Write persists an outbox entry atomically with the caller's business
	// state. Write MUST be invoked from within an active transaction; the
	// implementation extracts the tx from ctx via TxFromContext(ctx). Calling
	// Write outside of a persistence.TxRunner.RunInTx scope is a programming
	// error — implementations return an errcode error with KindInternal
	// (e.g. adapters/postgres returns ErrAdapterPGNoTx) rather than silently
	// writing without transactional guarantees.
	//
	// ref: nikolayk812/pgx-outbox writer.go -- explicit tx parameter +
	// ErrTxNil guard for the same MUST-have-tx contract.
	Write(ctx context.Context, entry Entry) error
}

Writer writes outbox entries within a transaction. The implementation MUST ensure the outbox write is atomic with the business state write (same DB transaction).

type WriterEmitter ¶

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

WriterEmitter emits by writing entries to the transactional outbox.

func NewWriterEmitter(w Writer) (*WriterEmitter, error)

func (e *WriterEmitter) Durable() bool

func (e *WriterEmitter) Emit(ctx context.Context, entry Entry) error