msgbus

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// DefaultByteArenaCapacity is the default capacity for the byte arena (1MB)
	DefaultByteArenaCapacity = 1024 * 1024

	// DefaultEventRingBufferSize is the default size for the event ring buffer
	DefaultEventRingBufferSize = 4096

	// DefaultOverflowDeadline is how long a critical-class producer waits for
	// ring/arena space before the process fails hard. Overflow of critical
	// topics means the dispatch loop has been stalled for this long — the
	// system can no longer guarantee order/balance state consistency.
	DefaultOverflowDeadline = 5 * time.Second
)
View Source 
const (
	// DefaultCommandRingBufferSize is the default size for the command ring buffer.
	// Smaller than events since commands are less frequent.
	DefaultCommandRingBufferSize = 1024

	// DefaultCommandArenaCapacity is the default capacity for the command byte arena (256KB).
	// Command payloads are typically small.
	DefaultCommandArenaCapacity = 256 * 1024
)
View Source 
const DefaultObserverInterval = time.Second

DefaultObserverInterval is the cadence at which the stats observer samples the overflow counters.

Variables

This section is empty.

Functions

This section is empty.

Types

type Command 

type Command struct {
	Ref       CommandRef
	CommandID uint64
	CreatedAt uint64
}

Command wraps a CommandRef with metadata. Commands are point-to-point: each command topic maps to exactly one processor.

type CommandProcessor 

type CommandProcessor func(cmd Command)

CommandProcessor is a function type for processing commands. Each command topic maps to exactly one CommandProcessor (point-to-point).

type CommandRef 

type CommandRef struct {
	CommandType command.CommandType
	Index       uint64 // offset in command arena
	Length      uint64 // size of data in bytes
	// contains filtered or unexported fields
}

CommandRef is a reference to command data stored in the command arena. CommandRefs are created by MsgBus.AllocateCmd, which also records the arena reservation so the space can be returned after processing.

type Config 

type Config struct {
	MsgLog MsgLogConfig `yaml:"msglog"`
}

Config contains msgbus configuration.

type Consumer 

type Consumer struct {
	Name     string               // unique identifier for this consumer
	Topics   map[event.Topic]bool // subscribed topics (nil or empty = all topics)
	Handler  EventHandler         // callback function for handling events
	Sequence uint64               // last processed event sequence (EventID)
}

Consumer represents a subscriber to the EventBus with optional topic filtering. Each consumer tracks its own sequence (last processed event ID) for coordinating arena memory release across multiple consumers.

func NewConsumer 

func NewConsumer(name string, topics []event.Topic, handler EventHandler) *Consumer

NewConsumer creates a new consumer with the given name, subscribed topics, and handler. If topics is nil or empty, the consumer will receive all event types.

func (*Consumer) ShouldHandle 

func (c *Consumer) ShouldHandle(topic event.Topic) bool

ShouldHandle returns true if this consumer should handle the given topic. Returns true for all topics if the consumer has no topic filter (empty Topics map).

type Event 

type Event struct {
	Ref       EventRef
	EventID   uint64
	CreatedAt uint64
	UpdatedAt uint64
}

Event wraps data with metadata. Data is embedded as a value type so Event and Data are pooled together (single allocation).

type EventBus 

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

EventBus is the central event distribution system. It uses a lock-free MPSC (Multiple Producer Single Consumer) ring buffer for events, allowing multiple goroutines to publish events concurrently while a single consumer (the dispatch loop) processes them.

Overflow policy (per topic class, see event.Topic.IsDroppable):

  • Critical topics (engine state, order lifecycle, execution, balance): Allocate/Publish spin (with runtime.Gosched escalation) until space frees; after the configured overflow deadline the process fails hard with a fatal log. Critical events are NEVER dropped.
  • Droppable topics (depth, tick, timer): after a bounded spin the event is dropped, the arena reservation is released, and the per-topic drop counter is incremented. Consumers recover via snapshot re-request (orderbook DepthID gap detection) or the next timer tick.

Deadlock note: a publisher running ON the dispatch goroutine (command processors, notifier calls from handlers) cannot be drained past a full ring, since it is itself the consumer. The overflow deadline converts that would-be deadlock into a detected fatal. Size the ring so that dispatch-thread publishers always have headroom.

Thread safety:

  • Allocate/Publish/Cancel can be called from multiple goroutines
  • Dispatch should be called from a single goroutine

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus creates a new EventBus with default capacity.

func NewEventBusWithCapacity 

func NewEventBusWithCapacity(arenaCapacity uint64) *EventBus

NewEventBusWithCapacity creates a new EventBus with the specified byte arena capacity.

func (*EventBus) Allocate 

func (e *EventBus) Allocate(topic event.Topic, size uint64) (EventRef, []byte, bool)

Allocate reserves arena space for an event of the given topic and returns a fully-populated EventRef plus the []byte slice to serialize into. After encoding, pass the EventRef to Publish (or Cancel if publishing is abandoned — the reservation must not leak).

Returns ok=false only for droppable topics when the arena stays full past the spin budget (the drop counter is incremented). Critical topics never fail: they wait, and fail hard after the overflow deadline.

func (*EventBus) Cancel 

func (e *EventBus) Cancel(ref EventRef)

Cancel releases the arena reservation of an allocated-but-never-published EventRef (e.g. after an encode error). Every Allocate must be balanced by exactly one Publish or Cancel.

func (*EventBus) ConsumerCount 

func (e *EventBus) ConsumerCount() int

ConsumerCount returns the number of registered consumers.

func (*EventBus) Dispatch 

func (e *EventBus) Dispatch() bool

Dispatch reads the next event from the ring buffer and dispatches it to all consumers whose topic filter matches. Returns true if an event was dispatched, false if the ring buffer is empty. After all consumers have processed the event, its arena reservation is released — event payload slices must not be retained past the handler.

func (*EventBus) DropCount 

func (e *EventBus) DropCount(topic event.Topic) uint64

DropCount returns the number of events dropped for the given topic. Only droppable-class topics can ever have a non-zero count.

func (*EventBus) MinSequence 

func (e *EventBus) MinSequence() uint64

MinSequence returns the minimum sequence across all consumers.

func (*EventBus) Poll 

func (e *EventBus) Poll(handler EventHandler) bool

Poll reads the next event from the ring buffer and calls the handler. The event's arena reservation is released after the handler returns.

func (*EventBus) Publish 

func (e *EventBus) Publish(ref EventRef) bool

Publish publishes an EventRef to the ring buffer. The caller must have serialized data into the arena buffer obtained via Allocate. This method is thread-safe and can be called from multiple goroutines.

Returns false only for droppable topics when the ring stays full past the spin budget; the event is dropped, its arena reservation released, and the drop counter incremented. Critical topics never fail (see EventBus doc).

func (*EventBus) ReadBuffer 

func (e *EventBus) ReadBuffer(offset, length uint64) []byte

ReadBuffer returns a []byte slice at the given offset/length for consumers to deserialize event data from. The slice is valid only until the event's reservation is released (i.e. within the dispatch handler).

func (*EventBus) Register 

func (e *EventBus) Register(name string, topics []event.Topic, handler EventHandler)

Register adds a consumer to the EventBus with optional topic filtering. If topics is nil or empty, the consumer will receive all topics. Consumers should be registered before calling Dispatch.

func (*EventBus) Release 

func (e *EventBus) Release()

Release updates minSequence to the minimum sequence across all consumers. This indicates the oldest event that is still being processed.

func (*EventBus) ReleaseArenas 

func (e *EventBus) ReleaseArenas()

ReleaseArenas is kept for backward compatibility. Arena space is released per event by Dispatch/Poll.

func (*EventBus) SetMsgLogger 

func (e *EventBus) SetMsgLogger(l *MsgLogger)

SetMsgLogger sets the message logger for persisting events as JSONL.

func (*EventBus) SetOverflowDeadline 

func (e *EventBus) SetOverflowDeadline(d time.Duration)

SetOverflowDeadline configures how long critical-class producers wait for space before failing hard. Must be called before concurrent publishing.

func (*EventBus) WaitCount 

func (e *EventBus) WaitCount(topic event.Topic) uint64

WaitCount returns the number of Allocate/Publish calls for the given topic that had to wait for ring or arena space.

type EventHandler 

type EventHandler func(event Event)

EventHandler is a function type for handling events

type EventPublisher 

type EventPublisher interface {
	Allocate(topic event.Topic, size uint64) (EventRef, []byte, bool)
	Publish(ref EventRef) bool
	Cancel(ref EventRef)
}

EventPublisher is the interface required by StateNotifier. It abstracts the event publishing functionality so that both EventBus and MsgBus can be used.

type EventRef 

type EventRef struct {
	Topic  event.Topic
	Index  uint64 // offset in arena
	Length uint64 // size of data in bytes
	// contains filtered or unexported fields
}

EventRef is a reference to event data stored in the event arena. EventRefs are created by EventBus.Allocate, which also records the arena reservation so the space can be returned after dispatch. Length may be shrunk by the producer before Publish if less data was written than allocated; the full reservation is still released.

type MsgBus 

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

MsgBus is the central message distribution system that supports both pub-sub events and point-to-point commands.

Event channel (pub-sub):

  • Uses MPSC (Multiple Producer Single Consumer) ring buffer
  • Multiple goroutines can publish events concurrently
  • Events are fan-out to all matching consumers

Command channel (point-to-point):

  • Uses SPSC (Single Producer Single Consumer) ring buffer
  • Commands are produced on the dispatch thread (during event handling)
  • Each command topic maps to exactly one processor
  • Commands always have higher priority than events

Thread safety:

  • Publish/Allocate can be called from multiple goroutines (event channel)
  • Send/AllocateCmd should only be called from the dispatch thread (command channel)
  • Dispatch should be called from a single goroutine

func NewMsgBus 

func NewMsgBus() *MsgBus

NewMsgBus creates a new MsgBus with default capacities.

func NewMsgBusWithCapacity 

func NewMsgBusWithCapacity(eventArenaCapacity uint64) *MsgBus

NewMsgBusWithCapacity creates a new MsgBus with custom event arena capacity.

func (*MsgBus) Allocate 

func (m *MsgBus) Allocate(topic event.Topic, size uint64) (EventRef, []byte, bool)

Allocate reserves space in the event arena for an event of the given topic. Returns ok=false only for droppable topics under sustained overflow (see EventBus.Allocate).

func (*MsgBus) AllocateCmd 

func (m *MsgBus) AllocateCmd(cmdType command.CommandType, size uint64) (CommandRef, []byte)

AllocateCmd reserves space in the command arena for a command of the given type and returns a fully-populated CommandRef plus the []byte slice to serialize into. This should only be called from the dispatch thread.

The dispatch thread is both producer and consumer of the command arena, so exhaustion cannot be waited out — it is a fatal sizing error.

func (*MsgBus) Cancel 

func (m *MsgBus) Cancel(ref EventRef)

Cancel releases an allocated-but-never-published EventRef.

func (*MsgBus) ConsumerCount 

func (m *MsgBus) ConsumerCount() int

ConsumerCount returns the number of registered event consumers.

func (*MsgBus) Dispatch 

func (m *MsgBus) Dispatch() bool

Dispatch processes messages with command priority. It first drains ALL pending commands from the SPSC ring buffer, then processes one event from the MPSC ring buffer. Returns true if any work was done (command or event dispatched).

func (*MsgBus) DropCount 

func (m *MsgBus) DropCount(topic event.Topic) uint64

DropCount returns the number of dropped events for the given topic.

func (*MsgBus) GetTicker 

func (m *MsgBus) GetTicker() Ticker

GetTicker returns the attached Ticker, or nil if none is set.

func (*MsgBus) MinSequence 

func (m *MsgBus) MinSequence() uint64

MinSequence returns the minimum sequence across all event consumers.

func (*MsgBus) Poll 

func (m *MsgBus) Poll(handler EventHandler) bool

Poll reads the next event from the ring buffer and calls the handler. Returns true if an event was processed, false if the ring buffer is empty. This is a single-consumer convenience method (no topic filtering).

func (*MsgBus) Publish 

func (m *MsgBus) Publish(ref EventRef) bool

Publish publishes an EventRef to the event ring buffer. The caller should have already serialized data into the arena buffer obtained via Allocate. Thread-safe. Returns false only when a droppable-class event was dropped (see EventBus.Publish).

func (*MsgBus) ReadBuffer 

func (m *MsgBus) ReadBuffer(offset, length uint64) []byte

ReadBuffer returns a []byte slice from the event arena at the given offset/length for consumers to deserialize event data from.

func (*MsgBus) ReadCmdBuffer 

func (m *MsgBus) ReadCmdBuffer(offset, length uint64) []byte

ReadCmdBuffer returns a []byte slice from the command arena at the given offset/length for command processors to deserialize command data from.

func (*MsgBus) Register 

func (m *MsgBus) Register(name string, topics []event.Topic, handler EventHandler)

Register adds an event consumer to the MsgBus with optional topic filtering. If topics is nil or empty, the consumer will receive all event topics. Consumers should be registered before calling Dispatch.

func (*MsgBus) RegisterCommand 

func (m *MsgBus) RegisterCommand(cmdType command.CommandType, processor CommandProcessor)

RegisterCommand registers a processor for a specific command topic. Each command topic can have at most one handler (point-to-point). Panics if a processor is already registered for the given topic.

func (*MsgBus) Release 

func (m *MsgBus) Release()

Release updates minSequence for the event channel. This indicates the oldest event still being processed.

func (*MsgBus) ReleaseArenas 

func (m *MsgBus) ReleaseArenas()

ReleaseArenas is kept for backward compatibility with the event channel.

func (*MsgBus) Send 

func (m *MsgBus) Send(ref CommandRef)

Send sends a command to the SPSC command ring buffer. This should only be called from the dispatch thread (single producer).

All commands are critical (order submit/cancel, reconciliation requests) and the dispatch thread cannot wait for itself to drain the ring, so a full command ring is a fatal sizing error rather than a silent drop.

func (*MsgBus) SetMsgLogger 

func (m *MsgBus) SetMsgLogger(l *MsgLogger)

SetMsgLogger sets the message logger for persisting events and commands as JSONL. It also forwards the logger to the EventBus for event logging.

func (*MsgBus) SetOverflowDeadline 

func (m *MsgBus) SetOverflowDeadline(d time.Duration)

SetOverflowDeadline configures how long critical-class event producers wait for ring/arena space before failing hard.

func (*MsgBus) SetTicker 

func (m *MsgBus) SetTicker(t Ticker)

SetTicker attaches a Ticker (e.g. core/clock.Clock) to the MsgBus. The ticker is driven by the dispatch loop via GetTicker().Tick(nowNs).

func (*MsgBus) StartObserver 

func (m *MsgBus) StartObserver(ctx context.Context, interval time.Duration)

StartObserver launches the low-frequency stats observer goroutine (P2-3).

High-frequency conditions on the hot path — event drops, overflow waits, unrouted commands — are recorded as atomic counters where they occur; this goroutine is the only place they reach the text log. Every interval it samples the counters and emits one Warn line if anything changed since the previous sample (silent otherwise, so a healthy system logs nothing).

The observer stops when ctx is cancelled. interval <= 0 selects DefaultObserverInterval.

func (*MsgBus) UnroutedCommandCount 

func (m *MsgBus) UnroutedCommandCount() uint64

UnroutedCommandCount returns the number of commands dispatched with no registered processor.

func (*MsgBus) WaitCount 

func (m *MsgBus) WaitCount(topic event.Topic) uint64

WaitCount returns the number of publish/allocate overflow waits for the given topic.

type MsgLogConfig 

type MsgLogConfig struct {
	Enabled bool              `yaml:"enabled"`
	File    rotate.FileConfig `yaml:"file"`
}

MsgLogConfig contains plaintext event/command JSONL logging configuration. Msglog never writes to stdout.

type MsgLogger 

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

MsgLogger writes plaintext JSONL event/command records to a single date+size rotated msg_*.jsonl stream. It is designed to be called from the single dispatch goroutine, so no locking is required.

func NewMsgLogger 

func NewMsgLogger(pol rotate.Policy) (*MsgLogger, error)

NewMsgLogger creates a MsgLogger backed by pol (ext should be "jsonl").

func (*MsgLogger) Close 

func (l *MsgLogger) Close()

Close syncs and closes the underlying writer.

func (*MsgLogger) LogCommand 

func (l *MsgLogger) LogCommand(cmd Command, payload []byte)

LogCommand writes a JSONL record for a command.

func (*MsgLogger) LogEvent 

func (l *MsgLogger) LogEvent(ev Event, payload []byte)

LogEvent writes a JSONL record for an event.

func (*MsgLogger) Sync 

func (l *MsgLogger) Sync() error

Sync fsyncs the active file when SyncPeriodic is configured.

type StateNotifier 

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

StateNotifier publishes engine state events to the event bus. It provides a convenient interface for engines to report their lifecycle states (Ready, Stop, Finished, Abnormal) through the event bus system.

State topics are critical-class: Allocate/Publish never drop them, so the ok returns below can only be false for droppable topics and are effectively always true here (kept for interface symmetry).

func NewStateNotifier 

func NewStateNotifier(publisher EventPublisher) *StateNotifier

NewStateNotifier creates a new StateNotifier with the given EventPublisher. Both *EventBus and *msgbus.MsgBus satisfy EventPublisher.

func (*StateNotifier) NotifyAbnormal 

func (n *StateNotifier) NotifyAbnormal(source common.EngineType, errorCode int, timestamp uint64)

NotifyAbnormal publishes an AbnormalEvent indicating an engine encountered an error.

func (*StateNotifier) NotifyFinished 

func (n *StateNotifier) NotifyFinished(source common.EngineType, timestamp uint64)

NotifyFinished publishes a FinishedEvent indicating an engine has finished.

func (*StateNotifier) NotifyReady 

func (n *StateNotifier) NotifyReady(source common.EngineType, timestamp uint64)

NotifyReady publishes a ReadyEvent indicating an engine is ready.

func (*StateNotifier) NotifyStop 

func (n *StateNotifier) NotifyStop(source common.EngineType, timestamp uint64)

NotifyStop publishes a StopEvent indicating an engine is stopping.

type Ticker 

type Ticker interface {
	Tick(nowNs uint64)
}

Ticker is the interface for a periodic timer that is driven by the dispatch loop. Implemented by core/clock.Clock.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.