messaging

README

messaging

Kafka messaging plus transactional outbox and inbox (idempotent-consumer) patterns, structured as multiple Go modules in one repository.

Modules

Module	Path	Depends on
root	github.com/zuksmaq/messaging	—
kafka	github.com/zuksmaq/messaging/kafka	root
outbox	github.com/zuksmaq/messaging/outbox	root
inbox	github.com/zuksmaq/messaging/inbox	root
integration	github.com/zuksmaq/messaging/integration	all four

The kafka module is split into packages: kafka holds the wire formats and connection settings both sides share, kafka/producer and kafka/consumer each own a Config and a New constructor (ADR 0008).

The integration module ships no library code: it exists so the end-to-end test of outbox → Kafka → Runner → inbox can depend on every module at once without any of them depending on the others.

See messaging-handoff.md for the full design record: decisions, invariants, open items, and rejected alternatives.

Development

go work use .   # workspace already committed as go.work

CI builds each module with GOWORK=off to catch un-tagged local-only changes before release.

Documentation

Index

• Constants
• Variables
• type Consumer
• type DeliveryStatus
  • func (s DeliveryStatus) String() string
• type Header
• type Message
• type ProducedMessage
• type Producer
• type ReceivedMessage
  • func (m ReceivedMessage[K, V]) EventID() string
  • func (m ReceivedMessage[K, V]) Tombstone() bool

Constants

const EventIDHeader = "event-id"

EventIDHeader is the header key carrying the producer-assigned idempotency id. The outbox relay stamps it from the outbox row id; consumers use it as the inbox de-duplication key.

Variables

var (
	// ErrSerialization indicates a value could not be encoded for
	// the wire.
	ErrSerialization = errors.New("serialization failed")

	// ErrDeserialization indicates a received value could not be
	// decoded from the wire.
	ErrDeserialization = errors.New("deserialization failed")

	// ErrSchemaRegistryRequired indicates the configured wire format
	// requires a schema registry client that was not provided.
	ErrSchemaRegistryRequired = errors.New("schema registry required")

	// ErrInvalidConfig indicates a config's Validate method rejected
	// it.
	ErrInvalidConfig = errors.New("invalid configuration")

	// ErrBroker indicates the underlying broker client returned an
	// error producing or consuming a message.
	ErrBroker = errors.New("broker error")
)

Sentinel errors for the categories callers need to branch on via errors.Is/errors.As. Wrap these with context as they travel up (fmt.Errorf("...: %w", err)); never compare with ==.

Types

type Consumer

type Consumer[K, V any] interface {
	// Consume blocks until the next message is available.
	Consume(ctx context.Context) (ReceivedMessage[K, V], error)

	// Commit advances the consumer's offset past msg.
	Commit(ctx context.Context, msg ReceivedMessage[K, V]) error
}

Consumer reads keyed messages from a broker. Auto-commit is intentionally absent from the contract: offsets advance only when the caller calls Commit after handling a message.

type DeliveryStatus

type DeliveryStatus int

DeliveryStatus reports how durably a produced message was persisted by the broker.

const (
	// NotPersisted means the broker gave no acknowledgement that the
	// message was written.
	NotPersisted DeliveryStatus = iota
	// PossiblyPersisted means the broker acknowledged the write but
	// durability is not guaranteed (e.g. a single in-sync replica).
	PossiblyPersisted
	// Persisted means the broker acknowledged the write with the
	// durability guarantees required by the producer's configuration.
	Persisted
)

func (DeliveryStatus) String

func (s DeliveryStatus) String() string

String returns the human-readable name of the delivery status.

type Header

type Header struct {
	Key   string
	Value []byte
}

Header is a single broker message header.

type Message

type Message[K, V any] struct {
	Key     K
	Value   V
	Headers map[string][]byte
}

Message is the logical content of a produced or received record: a key, a value, and broker headers.

type ProducedMessage

type ProducedMessage struct {
	Topic     string
	Partition int32
	Offset    int64
	Status    DeliveryStatus
}

ProducedMessage is the broker's acknowledgement of a produced message.

type Producer

type Producer[K, V any] interface {
	// Produce publishes a message and blocks until the broker
	// acknowledges it, returning the assigned coordinates and
	// DeliveryStatus.
	Produce(ctx context.Context, topic string, key K, value V, headers map[string][]byte) (ProducedMessage, error)

	// Close flushes any buffered messages and releases the
	// underlying broker connection.
	Close(ctx context.Context) error
}

Producer publishes keyed messages to a broker. Produce awaits the broker's acknowledgement before returning; implementations must never fire-and-forget.

type ReceivedMessage

type ReceivedMessage[K, V any] struct {
	Message[K, V]

	// RawKey and RawValue are the key and value exactly as the broker
	// delivered them, before deserialization. They survive a
	// deserialization failure, so a poison-message policy can forward
	// the original bytes to a dead-letter topic even when Key and Value
	// are still zero.
	RawKey   []byte
	RawValue []byte

	// HeaderList is every header as delivered, in order, preserving
	// duplicate keys that the Headers convenience map (last-wins)
	// collapses.
	HeaderList []Header

	Topic     string
	Partition int32
	Offset    int64
	Timestamp time.Time
}

ReceivedMessage is a message read from the broker, carrying its content plus the coordinates the broker assigned it.

func (ReceivedMessage[K, V]) EventID

func (m ReceivedMessage[K, V]) EventID() string

EventID returns the EventIDHeader value, or "" if absent.

func (ReceivedMessage[K, V]) Tombstone

func (m ReceivedMessage[K, V]) Tombstone() bool

Tombstone reports whether this message is a tombstone (a nil value), used by log-compacted topics to mark a key for deletion.