eventbus

Documentation

Overview

Package eventbus provides an in-memory implementation of kernel/outbox Publisher and Subscriber for development and testing. It delivers messages synchronously and is not suitable for production use.

Package eventbus provides an in-memory implementation of kernel/outbox.Publisher and kernel/outbox.Subscriber for development and testing.

ref: ThreeDotsLabs/watermill message/message.go — Message model, Ack/Nack pattern Adopted: topic-based pub/sub, callback handler pattern. Deviated: in-memory channel-based delivery (at-most-once, lost on restart); built-in retry with exponential backoff (3 attempts) + dead letter slice.

Index

• type DeadLetter
• type InMemoryEventBus
  • func New(clk clock.Clock, opts ...Option) *InMemoryEventBus
  • func (b *InMemoryEventBus) Close(_ context.Context) error
  • func (b *InMemoryEventBus) DeadLetterLen() int
  • func (b *InMemoryEventBus) DrainDeadLetters() []DeadLetter
  • func (b *InMemoryEventBus) GuaranteesSerialInOrderDelivery() bool
  • func (b *InMemoryEventBus) Health() string
  • func (b *InMemoryEventBus) Publish(_ context.Context, topic string, payload []byte) error
  • func (b *InMemoryEventBus) Ready(sub outbox.Subscription) <-chan struct{}
  • func (b *InMemoryEventBus) Setup(_ context.Context, _ outbox.Subscription) error
  • func (b *InMemoryEventBus) StopIntake(_ context.Context) error
  • func (b *InMemoryEventBus) Subscribe(ctx context.Context, sub outbox.Subscription, handler outbox.SubscriberHandler) error
• type Option
  • func WithBufferSize(size int) Option

Types

type DeadLetter

type DeadLetter struct {
	Topic   string
	Entry   outbox.Entry
	LastErr error
}

DeadLetter represents a message that exhausted retries.

type InMemoryEventBus

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

InMemoryEventBus is a channel-based event bus for development and testing. It implements both outbox.Publisher and outbox.Subscriber. Semantics: at-most-once delivery (messages are lost on process restart).

ConsumerGroup dispatch:

• same consumerGroup on same topic: round-robin (competing consumers)
• different consumerGroups on same topic: each group gets a copy (fanout)
• empty consumerGroup: broadcast to every subscriber (backward compatible)

func New

func New(clk clock.Clock, opts ...Option) *InMemoryEventBus

New creates an InMemoryEventBus with the given clock and options.

func (*InMemoryEventBus) Close

func (b *InMemoryEventBus) Close(_ context.Context) error

Close terminates all subscriber goroutines and prevents new publishes. Safety: Close holds mu.Lock() for the full channel-closing loop, while Publish holds mu.RLock() while sending to subscriber channels. That lock ordering prevents Publish from sending to a closed subscriber channel.

ctx is intentionally not consumed: closing in-memory channels is O(1) and must always complete to avoid goroutine leaks. Intercepting ctx here would risk leaving subscriber goroutines permanently blocked on the channel read.

ref: kernel/lifecycle doc.go — "resources that must complete teardown unconditionally should ignore the ctx and document the reason".

func (*InMemoryEventBus) DeadLetterLen

func (b *InMemoryEventBus) DeadLetterLen() int

DeadLetterLen returns the number of dead letter messages.

func (*InMemoryEventBus) DrainDeadLetters

func (b *InMemoryEventBus) DrainDeadLetters() []DeadLetter

DrainDeadLetters returns and clears all dead letter messages.

func (*InMemoryEventBus) GuaranteesSerialInOrderDelivery

func (b *InMemoryEventBus) GuaranteesSerialInOrderDelivery() bool

GuaranteesSerialInOrderDelivery satisfies outbox.SerialInOrderGuarantor: the in-memory bus is the one transport that may carry an L3 projection subscription. Each subscription runs one consume goroutine that reads its buffered channel FIFO and invokes the handler synchronously (see Subscribe), so a single subscription's stream is delivered strictly serially and in order — the precondition projection exactly-once requires (ADR §6 row 4).

The guarantee is scoped to a SINGLE subscriber on a (consumerGroup, topic): a projection's group is "<cellID>-<projectionID>" with exactly one subscription registered by the bootstrap drain, so the precondition holds. It does NOT extend to multiple competing subscribers in one group, which roundRobin dispatches across independent goroutines (irrelevant to projections, which never share a group).

func (*InMemoryEventBus) Health

func (b *InMemoryEventBus) Health() string

Health returns the current status of the event bus. Returns "healthy" when the bus is open, "closed" when it has been shut down.

func (*InMemoryEventBus) Publish

func (b *InMemoryEventBus) Publish(_ context.Context, topic string, payload []byte) error

Publish sends payload to subscribers of the given topic.

Contract (P1-14 follow-up): payload MUST be a v1 wire envelope (produced by outbox.MarshalEnvelope for relay-driven paths, or outbox.MarshalDirectEnvelope for direct-publish paths such as demo-mode L2 cells and L4 cells without outbox writer). Raw business payloads are rejected fail-closed.

ConsumerGroup dispatch:

• For each named consumer group: pick ONE subscriber via round-robin
• For the empty-group ("") bucket: send to ALL subscribers (broadcast)

Non-blocking: if a subscriber's buffer is full, the message is dropped (logged at Error level — indicates message loss and impacts correctness).

The bus unwraps the envelope so subscribers always see the business payload in Entry.Payload, matching the semantics of the RabbitMQ subscriber path.

Regression guard: before this contract, demo-mode and L4 direct publishes sent raw business bytes; with envelope enforcement added in P1-14 A1/A2 such bytes were silently dead-lettered + nil returned, causing complete event loss (symptom: ssobff walkthrough audit-entries assertion fails because auditcore never received session.created / user.created events). Returning an explicit error makes producer-side contract violations loud instead of silent; the dead-letter slice is retained as a diagnostic trail.

ref: Watermill poison-queue middleware — undecodable messages → DLX, main route cleared; K8s workqueue fail-closed semantics.

func (*InMemoryEventBus) Ready

func (b *InMemoryEventBus) Ready(sub outbox.Subscription) <-chan struct{}

Ready implements outbox.Subscriber. Returns a channel that closes once Subscribe has been called for the given subscription (i.e., the subscription is registered and ready to receive messages). This prevents the publish-before-subscribe race in tests that use waitForSubscription.

The key is sub.ConsumerGroup + "|" + sub.Topic so that different consumer groups on the same topic each get an independent ready signal.

func (*InMemoryEventBus) Setup

func (b *InMemoryEventBus) Setup(_ context.Context, _ outbox.Subscription) error

Setup implements outbox.Subscriber. InMemoryEventBus requires no topology pre-declaration; returns nil immediately.

func (*InMemoryEventBus) StopIntake

func (b *InMemoryEventBus) StopIntake(_ context.Context) error

StopIntake satisfies outbox.SubscriberIntakeStopper. InMemoryEventBus does not have a broker-side intake to stop — messages flow directly via in-process channels. This no-op implementation exists so that runtime/eventrouter's type assertion `subscriber.(SubscriberIntakeStopper)` succeeds uniformly across in-memory and broker-backed configurations.

StopIntake is idempotent and safe to call concurrently with Subscribe/Publish.

func (*InMemoryEventBus) Subscribe

func (b *InMemoryEventBus) Subscribe(ctx context.Context, sub outbox.Subscription, handler outbox.SubscriberHandler) error

Subscribe registers an EntryHandler for the given subscription. It blocks until ctx is canceled or the bus is closed.

Consumer: cg-eventbus-{sub.ConsumerGroup}-{sub.Topic} Idempotency key: N/A (in-memory, no persistence) ACK timing: after handler returns DispositionAck Retry: transient errors -> retry 3x with exponential backoff / permanent -> dead letter

sub.ConsumerGroup selects the dispatch mode:

• non-empty: messages are load-balanced among subscribers in the same group
• empty: each subscriber receives every message (broadcast / fanout)

type Option

type Option func(*InMemoryEventBus)

Option configures the InMemoryEventBus.

func WithBufferSize

func WithBufferSize(size int) Option

WithBufferSize sets the channel buffer size per subscriber. Default is 256.