broker

Documentation

Overview

Package broker embeds the mochi-mqtt server and gives it Astarte MQTT v1 semantics (docs/DESIGN.md §3.1–§3.4): mTLS device identity with CN = "<realm>/<device_id>", the §3.2 ACL matrix, persistent sessions in a bbolt file (so session_present survives Astrate restarts), device lifecycle bookkeeping, and an inline publishing facade for the engine and AppEngine.

The broker does not interpret payloads. Every accepted device PUBLISH is handed to an Intake (implemented by internal/engine in M6) as an InboundMessage whose Ack callback releases the MQTT acknowledgment: for QoS >= 1 the PUBACK/PUBREC is withheld until Ack is called, which is how persistence-commit ordering (docs/DESIGN.md §5.3) and shard backpressure (§1.4) propagate to the device.

Index

• Constants
• Variables
• func SplitTopic(topic string) (realm, device, rest string, err error)
• type Broker
  • func New(ctx context.Context, cfg Config, st Store, intake Intake, sink LifecycleSink) (*Broker, error)
  • func (b *Broker) Close() error
  • func (b *Broker) DevAddr() string
  • func (b *Broker) Publisher() *Publisher
  • func (b *Broker) RefreshIntrospection(ctx context.Context, realm string, id deviceid.ID) error
  • func (b *Broker) ReloadRealms(ctx context.Context) error
  • func (b *Broker) SessionCount() int
  • func (b *Broker) Start() error
  • func (b *Broker) TLSAddr() string
• type Config
• type Identity
  • func ParseCN(cn string) (Identity, error)
  • func (i Identity) BaseTopic() string
  • func (i Identity) CN() string
• type InboundMessage
• type Intake
• type LifecycleEvent
• type LifecycleEventType
• type LifecycleSink
• type Publisher
  • func (p *Publisher) Publish(topic string, payload []byte, qos byte, retain bool, expiry time.Duration) error
• type Store

Constants

const (
	// DefaultTLSAddr is the standard Astarte broker port.
	DefaultTLSAddr = ":8883"
	// DefaultDevAddr is the plaintext development listener address, bound
	// only when InsecureDevMode is set.
	DefaultDevAddr = ":1883"
	// DefaultMaxPacketBytes caps inbound MQTT packets: the 64 KiB payload
	// bound (docs/DESIGN.md §3.5.3, §4.5) plus generous topic/header room.
	DefaultMaxPacketBytes = 128 * 1024
)

Config defaults.

Variables

var (
	// ErrBadCN reports a certificate CN (or claimed client ID) that is not
	// a well-formed "<realm>/<device_id>".
	ErrBadCN = errors.New("broker: malformed identity CN")
	// ErrBadTopic reports a topic that does not start with the
	// "<realm>/<device_id>" prefix scheme (docs/DESIGN.md §3.3).
	ErrBadTopic = errors.New("broker: malformed device topic")
)

Sentinel identity/topic errors.

Functions

func SplitTopic

func SplitTopic(topic string) (realm, device, rest string, err error)

SplitTopic splits a wire topic into its realm, device, and rest parts (docs/DESIGN.md §3.3 parsing note): "<realm>/<device_id>" yields an empty rest (the introspection topic); "<realm>/<device_id>/<rest>" yields the remainder verbatim (control suffix or "<interface_name><path>"). The realm and device segments are shape-checked; the rest is not interpreted.

Types

type Broker

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

Broker is the embedded MQTT broker (docs/ROADMAP.md §6 file 5.8): mochi server, Astarte hooks, persistent session store, and inline publisher.

func New

func New(ctx context.Context, cfg Config, st Store, intake Intake, sink LifecycleSink) (*Broker, error)

New assembles the broker: hooks registered, listeners bound (so the addresses are known), realm CA pools loaded. Call Start to begin serving. intake must be non-nil; sink may be nil.

func (*Broker) Close

func (b *Broker) Close() error

Close gracefully stops the broker: blocked publish acknowledgments are released (their messages stay unacknowledged on the devices, which re-send after reconnecting — at-least-once, docs/DESIGN.md §5.3), clients are disconnected, and the session store is flushed and closed.

Known upstream defect worked around here: mochi's attachClient registers on Listeners.ClientsWg AFTER the accept (server.go, present through mochi main), so a connection landing exactly at Close races that Add against CloseAll's Wait — flagged by the race detector, and in the worst case a "WaitGroup misuse" panic. Close therefore quiesces first: it stops each accept loop (passing a copy of mochi's own closeListenerClients callback, because the listener's once-guard would swallow the real one on the second Close), waits closeSettle for in-flight accepts to register, and only then calls Server.Close, whose Wait can no longer run concurrently with an Add. Revisit if the mochi pin ever moves past v2.7.9.

func (*Broker) DevAddr

func (b *Broker) DevAddr() string

DevAddr returns the bound plaintext listener address, or "" when insecure_dev_mode is off.

func (*Broker) Publisher

func (b *Broker) Publisher() *Publisher

Publisher returns the inline publishing facade for the engine/AppEngine.

func (*Broker) RefreshIntrospection

func (b *Broker) RefreshIntrospection(ctx context.Context, realm string, id deviceid.ID) error

RefreshIntrospection reloads a connected device's introspection-derived ACL state. The engine calls it after persisting a new introspection (docs/ROADMAP.md §7.2 file 6.7). Unknown or disconnected devices are a no-op: their state loads fresh on the next connect or delivery.

func (*Broker) ReloadRealms

func (b *Broker) ReloadRealms(ctx context.Context) error

ReloadRealms rebuilds the per-realm client-CA pools from the store. Realm CRUD (M7 housekeeping) calls it after creating, deleting, or re-keying a realm; new TLS handshakes pick the change up immediately.

func (*Broker) SessionCount

func (b *Broker) SessionCount() int

SessionCount returns the number of live authenticated device sessions (docs/DESIGN.md §5.2: the broker-sessions observability gauge reads it).

func (*Broker) Start

func (b *Broker) Start() error

Start restores persisted session state and begins serving the bound listeners. It does not block.

func (*Broker) TLSAddr

func (b *Broker) TLSAddr() string

TLSAddr returns the bound mTLS listener address (useful with ":0").

type Config

type Config struct {
	// TLSAddr is the mTLS listener address (default DefaultTLSAddr).
	TLSAddr string
	// ServerTLSCert is the broker's server-side TLS identity, required for
	// the TLS listener. Deployments issue it from the realm CA or any CA
	// the device fleet trusts (docs/DESIGN.md §4.4 flow C delivers ca_crt).
	ServerTLSCert tls.Certificate
	// InsecureDevMode additionally binds a plaintext listener that
	// authenticates by claimed client ID alone — local development only
	// (docs/DESIGN.md §3.1).
	InsecureDevMode bool
	// DevAddr is the plaintext listener address (default DefaultDevAddr;
	// ignored unless InsecureDevMode).
	DevAddr string
	// SessionStorePath is the bbolt file persisting sessions across
	// restarts (docs/DESIGN.md §3.1). Required.
	SessionStorePath string
	// EnforceLatestCert rejects connections presenting a certificate older
	// than the device's latest issuance (pairing.enforce_latest_cert,
	// docs/DESIGN.md §4.3).
	EnforceLatestCert bool
	// MaxPacketBytes caps inbound MQTT packet size (default
	// DefaultMaxPacketBytes).
	MaxPacketBytes uint32
	// Logger receives broker and hook logs (default slog.Default()).
	Logger *slog.Logger
}

Config carries the broker's operational knobs (TOML wiring lands in M8).

type Identity

type Identity struct {
	// Realm is the device's realm name.
	Realm string
	// DeviceID is the 128-bit Astarte device ID.
	DeviceID deviceid.ID
}

Identity is an authenticated device: the parsed form of the certificate CN "<realm>/<device_id>" (docs/DESIGN.md §3.1, §4.3).

func ParseCN

func ParseCN(cn string) (Identity, error)

ParseCN parses a certificate CN (equivalently: a claimed MQTT client ID) of the form "<realm>/<device_id>". The realm must be a valid Astarte realm name and the device part a 22-character base64url 128-bit device ID; anything else fails with ErrBadCN.

func (Identity) BaseTopic

func (i Identity) BaseTopic() string

BaseTopic returns the device's base MQTT topic B = "<realm>/<device_id>" (docs/DESIGN.md §3.2). Identical to CN; named separately for call-site clarity.

func (Identity) CN

func (i Identity) CN() string

CN renders the identity back to its canonical "<realm>/<device_id>" form — the certificate CN, the required MQTT client ID, and the device's base topic, which are all the same string by design.

type InboundMessage

type InboundMessage struct {
	// Realm is the device's realm name (first topic segment).
	Realm string
	// DeviceID is the publishing device, parsed from the connection's
	// certificate CN.
	DeviceID deviceid.ID
	// Topic is the full topic the device published to,
	// "<realm>/<device_id>[/<rest>]".
	Topic string
	// Payload is the raw message body (BSON, JSON, control bytes, or empty
	// for property unset — classification happens in the engine).
	Payload []byte
	// QoS is the publish quality of service as received (0..2).
	QoS byte
	// ReceivedAt is the broker reception timestamp (used as the fallback
	// datastream timestamp when the payload carries none).
	ReceivedAt time.Time
	// Ack releases the broker-held acknowledgment for this message. For
	// QoS >= 1 the device's PUBACK (or PUBREC) is not sent until Ack is
	// called — the engine calls it after the persistence batch commits
	// (docs/DESIGN.md §5.3). Ack is never nil, is safe to call multiple
	// times, and must eventually be called for every QoS >= 1 message the
	// intake accepts, or the publishing device stalls (that stall is the
	// designed backpressure path, §1.4).
	Ack func()
}

InboundMessage is one device PUBLISH delivered to the engine intake (docs/ROADMAP.md §6 file 5.1). Payload is an independent copy: it remains valid after the broker recycles its packet buffers.

type Intake

type Intake interface {
	Submit(InboundMessage)
}

Intake consumes accepted device publishes. internal/engine implements it (M6); tests use recorders. Submit may block — a full engine shard blocking the broker's per-client read loop is the §1.4 backpressure contract.

type LifecycleEvent

type LifecycleEvent struct {
	// Type is the event discriminator.
	Type LifecycleEventType
	// Realm is the device's realm name.
	Realm string
	// DeviceID is the device.
	DeviceID deviceid.ID
	// RemoteIP is the peer address for connects; the zero Addr for
	// disconnects.
	RemoteIP netip.Addr
	// At is the event instant.
	At time.Time
}

LifecycleEvent is one device connect/disconnect observation (docs/DESIGN.md §3.1 — the work astarte_vmq_plugin does upstream).

type LifecycleEventType

type LifecycleEventType string

LifecycleEventType discriminates LifecycleEvent values. The constants double as the Astarte trigger event names fed to the M6 trigger engine.

const (
	// EventDeviceConnected fires after a device session is established.
	EventDeviceConnected LifecycleEventType = "device_connected"
	// EventDeviceDisconnected fires after a device connection ends (not on
	// session takeover: the device is still connected, on a new channel).
	EventDeviceDisconnected LifecycleEventType = "device_disconnected"
)

type LifecycleSink

type LifecycleSink interface {
	OnLifecycleEvent(LifecycleEvent)
}

LifecycleSink receives lifecycle events (the device_connected / device_disconnected trigger feed). Implementations must not block: they are invoked from broker connection goroutines.

type Publisher

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

Publisher is the inline-client facade for server→device messages (docs/ROADMAP.md §6 file 5.7). Publishes bypass the ACL (docs/DESIGN.md §3.2) and are delivered — or queued on the persistent session, with optional per-message expiry — exactly as device-side subscriptions dictate. The engine (M6) and AppEngine (M7) consume it.

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, payload []byte, qos byte, retain bool, expiry time.Duration) error

Publish sends a server-side message. qos is clamped to 0..2; retain marks the message retained on its topic (server-owned properties, docs/DESIGN.md §3.4); a positive expiry bounds how long the message may wait in an offline device's queue (datastream expiry, §2.5 — rounded up to whole seconds, the MQTT expiry granularity). Zero expiry means the broker default (mochi's maximum message expiry, 24h).

type Store

type Store interface {
	// ListRealms feeds the per-realm client-CA pools.
	ListRealms(ctx context.Context) ([]store.Realm, error)
	// GetDevice authenticates connections and loads introspection for ACLs.
	GetDevice(ctx context.Context, realmID int16, id deviceid.ID) (*store.Device, error)
	// GetInterface resolves the ownership of introspected interfaces.
	GetInterface(ctx context.Context, realmID int16, name string, major int) (*store.StoredInterface, error)
	// SetDeviceConnected records a connection on the device row.
	SetDeviceConnected(ctx context.Context, realmID int16, id deviceid.ID, at time.Time, ip netip.Addr) error
	// SetDeviceDisconnected records a disconnection on the device row.
	SetDeviceDisconnected(ctx context.Context, realmID int16, id deviceid.ID, at time.Time) error
}

Store is the persistence surface the broker consumes (hexagonal-lite, docs/DESIGN.md §1.3). *store.Store satisfies it; tests substitute fakes.