event

package
v0.70.5 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package event provides Dingo's EventBus: an in-process publish/ subscribe primitive that lets components communicate without holding references to each other.

Components use typed events for asynchronous cross-component notifications. Synchronous state reads still use direct calls, callbacks, or narrow interfaces supplied by the node composition layer. This keeps event traffic explicit without forcing every query through the bus.

Publishing 

eventBus.Publish(
    chain.ChainForkEventType,
    event.NewEvent(chain.ChainForkEventType, chain.ChainForkEvent{...}),
)

Use PublishAsync for events that do not need to be delivered synchronously with the publisher's call stack, and PublishOrdered for those that additionally need to reach subscribers in the order they were published.

Ordering guarantees

Publish and PublishBlocking deliver on the caller's goroutine, so a single publisher's events reach each subscriber in call order.

PublishAsync does not preserve order. It hands the event to a shared queue drained by AsyncWorkerPoolSize workers that race each other into Publish, so two events enqueued in order can be delivered in either order -- observed as several inversions per few hundred events, even from one publishing goroutine. Use it only where subscribers treat each event independently.

PublishOrdered is the async path that does preserve order. Each event type gets its own FIFO drained by exactly one worker, so events published to one type arrive in publish order, and a slow subscriber delays only its own event type rather than every async event. The guarantee is per event type and only over publishes that are themselves sequenced: concurrent publishers still race to enqueue, and nothing is promised across different event types.

ledger.tx uses PublishOrdered. A subscriber deriving state from it can rely on a block's transactions arriving in index order, and on a rollback's undo events (Rollback: true) arriving before any transaction event the ledger emits afterwards. Forward Apply events are registered with the database transaction's AfterCommit hook, so a rollback or failed commit publishes none. See blinklabs-io/dingo#2287: while those undo events were emitted from a detached goroutine, a subscriber could apply an undo after the redo that followed it and stay wrong indefinitely.

A lane orders what reaches it; it cannot order two publishers racing to reach it. Where events for one stream come from more than one goroutine, as ledger.tx does, the publishers need their own happens-before -- see the ledger.tx section of ARCHITECTURE.md for how the rollback and block-apply paths establish theirs.

A healthy subscriber drains a full lane; the ordinary policy detaches a stalled one after the delivery timeout. A caller on a goroutine something else waits for before the bus stops must still use PublishOrderedContext and cancel that context when it needs a shorter bound than the subscriber- delivery timeout.

Delivery guarantees

The bus does not drop events for a live subscriber. When a subscriber's channel buffer or the shared async queue is full, the publisher waits for capacity rather than discarding the event, so ingestion slows instead of losing work that subscribers derive state from. A subscriber that remains full for the delivery timeout is detached by the ordinary subscription policy: events already accepted into its channel retain their order, while the event that cannot be accepted and later events continue only to healthy subscribers. A lossless owner can explicitly select the blocking policy when detaching its stream would make recovery unsafe. This bounds a dead ordinary subscriber's impact without trading unbounded memory for liveness. Stop, Close, and Unsubscribe also release publishers parked on a full buffer.

The practical consequence is that a slow subscriber backpressures its publishers until it drains or is detached. A publisher must not hold a lock that a subscriber of the same event acquires: once the buffer fills, the subscriber waits for the lock and the publisher waits for the capacity the subscriber would free, and neither proceeds until the delivery bound detaches that subscriber. Queue such events and publish them after releasing the lock (see ledger's pendingPublishes). Subscribers that take a channel from Subscribe must drain it for as long as they hold the subscription and must Unsubscribe when they stop. A delivery parked for a long time is reported by the event_delivery_blocked_total metric and an "event delivery stalled" warning.

Subscribing 

eventBus.SubscribeFunc(chain.ChainForkEventType, func(evt event.Event) {
    e, ok := evt.Data.(chain.ChainForkEvent)
    if !ok { return }
    // handle e
})

The bus runs a pool of async worker goroutines (default 4) to dispatch subscribers. Subscriber callbacks must be non-blocking; if a callback needs to do real work, push it onto its own goroutine. A slow subscriber backpressures the bus and delays delivery of unrelated events: the async workers are a shared pool, so a subscriber that parks them holds up every async event type. PublishOrdered's per-type lanes are exempt from that specific coupling -- a parked lane worker holds up only its own event type -- but they are single workers, so a slow subscriber stalls that type more readily than four shared workers would.

Event type constants live alongside the package that owns the event: ChainForkEventType in chain, ChainSwitchEventType in chainselection, PeerEligibilityChangedEventType in peergov, etc.

Index

Constants

View Source 
const (
	ChainsyncResyncReasonLocalTipPlateau                   = "local_tip_plateau"
	ChainsyncResyncReasonPostPlateauRealign                = "post_plateau_realign"
	ChainsyncResyncReasonRollbackAhead                     = "rollback point ahead of local tip"
	ChainsyncResyncReasonRollbackNotFound                  = "rollback point not found"
	ChainsyncResyncReasonRollbackLoop                      = "rollback loop detected"
	ChainsyncResyncReasonPersistentFork                    = "persistent chain fork"
	ChainsyncResyncReasonRollbackExceedsK                  = "rollback exceeds security parameter K"
	ChainsyncResyncReasonRollbackExceedsMithril            = "rollback exceeds Mithril trust boundary"
	ChainsyncResyncReasonPeerTipBehindMithril              = "peer tip behind Mithril trust boundary"
	ChainsyncResyncReasonForkResolutionExceedsK            = "fork resolution exceeds security parameter K"
	ChainsyncResyncReasonLocalLedgerRollback               = "local ledger rollback"
	ChainsyncResyncReasonLiveTxValidationRecovery          = "live tx validation recovery"
	ChainsyncResyncReasonDeterministicTxValidationRecovery = "deterministic tx validation recovery"
	ChainsyncResyncReasonReplayRecoveryNonConverging       = "replay tx validation recovery not converging"
	ChainsyncResyncReasonChainSwitchCursorAhead            = "chain switch cursor ahead of local tip"
	ChainsyncResyncReasonBlockfetchTimeoutRetryFailed      = "blockfetch timeout retry failed on all available connections"
	ChainsyncResyncReasonBlockfetchRangeUnavailable        = "blockfetch could not obtain the queued header range"
	ChainsyncResyncReasonHeaderValidationRecovery          = "deferred header validation recovery"
	ChainsyncResyncReasonForkQueueOverflowRestartFailed    = "failed to restart blockfetch after fork-resolution header-queue overflow"
	// ChainsyncResyncReasonFutureHeaderAdmissionRecovery re-intersects the
	// ChainSync mini-protocol after a resolvable header was deliberately dropped
	// outside the permitted clock-skew window. It is not a peer-fault signal and
	// does not require a fresh connection or peer cooldown.
	ChainsyncResyncReasonFutureHeaderAdmissionRecovery = "future header admission recovery"
)
View Source 
const (
	// EventQueueSize is the high-burst buffer used by subscribers that may
	// receive bulk-sync bursts (e.g. chainsync/blockfetch ingest in the
	// ledger). Subscribers opt in to this size via the *WithBuffer
	// variants. Sized to absorb the worst case from #1556 / #1914. Buffer
	// size no longer decides whether events survive — a full buffer
	// backpressures the publisher rather than dropping (#2932) — it decides
	// how large a burst passes through without slowing ingestion.
	EventQueueSize = 100000
	// DefaultSubscriberBuffer is the per-subscriber channel buffer used by
	// Subscribe/SubscribeFunc when no explicit size is requested. Most
	// subscribers (peergov, governance, async housekeeping, etc.) only
	// receive sparse traffic and do not need 100k slots; sizing the
	// default down keeps idle steady-state heap small while leaving the
	// burst headroom available to opt-in callers via SubscribeWithBuffer
	// / SubscribeFuncWithBuffer. See blinklabs-io/dingo#2106.
	DefaultSubscriberBuffer = 1024
	AsyncQueueSize          = 1000
	AsyncWorkerPoolSize     = 4
	RemoteDeliverTimeout    = 5 * time.Second
)
View Source 
const BlockForgedEventType = EventType("block.forged")

BlockForgedEventType is the event type for locally forged blocks

View Source 
const ChainsyncResyncEventType = EventType("chainsync.resync")

ChainsyncResyncEventType is the event type emitted when a chainsync re-sync is required (e.g. persistent fork detected or rollback exceeds the security parameter).

View Source 
const EpochNonceReadyEventType = EventType("epoch.nonce_ready")

EpochNonceReadyEventType is emitted once the current epoch has advanced past the randomness stabilisation cutoff, which means the next epoch's nonce is now stable and its leader schedule can be precomputed.

View Source 
const EpochTransitionEventType = EventType("epoch.transition")

EpochTransitionEventType is the event type for epoch transitions

View Source 
const HardForkEventType = EventType("hardfork.transition")

HardForkEventType is the event type for hard fork (era transition) events

View Source 
const OrderedQueueSize = 10000

OrderedQueueSize is the per-event-type buffer behind PublishOrdered. It is larger than AsyncQueueSize because an ordered lane is drained by exactly one worker rather than by the shared pool, so it has to absorb the same bursts with less drain throughput. The ledger publishes ledger.tx from an after-commit callback, and a publisher that parks delays the block-apply pipeline even though the transaction is already durable, so the buffer is sized to swallow a bulk-sync batch's transactions rather than to bound memory tightly. It is still bounded: past this point the publisher waits, exactly as it already did on a full shared async queue.

Variables

View Source 
var ErrEventBusStopped = errors.New("event bus stopped")

ErrEventBusStopped is returned by PublishBlocking when the EventBus is stopping or closed before or during delivery.

View Source 
var ErrEventSubscriberStalled = errors.New("event subscriber stalled")

ErrEventSubscriberStalled is returned by PublishBlocking when an ordinary in-memory subscriber stays full past the delivery bound and is detached.

Functions

This section is empty.

Types

type BlockForgedEvent added in v0.22.0 

type BlockForgedEvent struct {
	// Slot is the slot number where the block was forged
	Slot uint64
	// BlockNumber is the block height in the chain
	BlockNumber uint64
	// BlockHash is the hash of the forged block
	BlockHash []byte
	// TxCount is the number of transactions included in the block
	TxCount uint
	// BlockSize is the size of the block in bytes
	BlockSize uint
	// Timestamp is when the block was forged
	Timestamp time.Time
}

BlockForgedEvent is emitted when the node successfully forges a new block. This event is published after the block has been added to the local chain via chain.AddBlock(), which triggers automatic propagation to connected peers.

type ChainsyncResyncEvent added in v0.22.0 

type ChainsyncResyncEvent struct {
	ConnectionId ouroboros.ConnectionId
	Reason       string
	Point        ocommon.Point
}

ChainsyncResyncEvent carries the connection ID that should be re-synced.

type EpochNonceReadyEvent added in v0.27.3 

type EpochNonceReadyEvent struct {
	CurrentEpoch uint64
	ReadyEpoch   uint64
	CutoffSlot   uint64
}

EpochNonceReadyEvent signals that the next epoch's nonce is stable.

type EpochTransitionEvent added in v0.22.0 

type EpochTransitionEvent struct {
	PreviousEpoch uint64
	NewEpoch      uint64
	BoundarySlot  uint64
	EpochNonce    []byte
	// ProtocolVersion is the protocol major version active for the boundary.
	ProtocolVersion uint
	// SnapshotSlot is the slot at which snapshot should be taken (typically boundary - 1)
	SnapshotSlot uint64
}

EpochTransitionEvent is emitted when the chain crosses an epoch boundary

type Event 

type Event struct {
	Timestamp time.Time
	Data      any
	Type      EventType
}

func NewEvent 

func NewEvent(eventType EventType, eventData any) Event

type EventBus 

type EventBus struct {
	Logger *slog.Logger
	// contains filtered or unexported fields
}

func NewEventBus 

func NewEventBus(
	promRegistry prometheus.Registerer,
	logger *slog.Logger,
) *EventBus

NewEventBus creates a new EventBus with async worker pool

func (*EventBus) Close added in v0.27.2 

func (e *EventBus) Close()

Close permanently shuts down the EventBus and its worker pool. Unlike Stop, Close does not restart async workers, so the EventBus cannot be reused.

func (*EventBus) HasSubscribers added in v0.25.1 

func (e *EventBus) HasSubscribers(eventType EventType) bool

func (*EventBus) Publish 

func (e *EventBus) Publish(eventType EventType, evt Event)

Publish allows a producer to send an event of a particular type to all subscribers

func (*EventBus) PublishAsync added in v0.21.0 

func (e *EventBus) PublishAsync(eventType EventType, evt Event) bool

PublishAsync enqueues an event for asynchronous delivery to all subscribers. It hands the event to the shared async queue and returns without waiting for subscriber delivery. Use this for events that do not need to be delivered synchronously with the publisher's call stack.

When the queue is full the caller waits for space rather than losing the event, so a backlog slows producers instead of discarding work. Returns false only when the EventBus is stopped or closed.

func (*EventBus) PublishBlocking added in v0.47.0 

func (e *EventBus) PublishBlocking(eventType EventType, evt Event) error

PublishBlocking delivers an event to all subscribers without dropping in-memory channel events when subscriber buffers are full. This should be reserved for ordering-critical streams where loss is worse than applying producer backpressure. It returns ErrEventBusStopped when the bus is stopping or closed before or during delivery.

func (*EventBus) PublishOrdered added in v0.70.1 

func (e *EventBus) PublishOrdered(eventType EventType, evt Event) bool

PublishOrdered enqueues an event for asynchronous delivery that preserves publisher order. Events published to the same event type are delivered to each subscriber in the order they were published, which PublishAsync does not promise.

Ordering holds per event type, and only among publishes that are themselves ordered: two goroutines publishing concurrently to one type still race to enqueue, and nothing is promised across different event types. A single producer sequence -- a ledger rollback's transaction undo events followed by the next block's transaction events -- is exactly the case this is for.

Like PublishAsync it does not drop for a live subscriber: a full lane makes the publisher wait for capacity rather than discarding the event. A stalled ordinary subscriber is detached after the delivery timeout, which lets its lane make progress for healthy subscribers; a lossless subscription instead remains blocked until lifecycle cancellation. Shutdown also releases the wait. Returns false when the EventBus is stopped or closed.

Each event type gets its own lane, so a slow subscriber delays only its own event type instead of holding up every async event as it would on the shared pool.

func (*EventBus) PublishOrderedContext added in v0.70.1 

func (e *EventBus) PublishOrderedContext(
	ctx context.Context,
	eventType EventType,
	evt Event,
) bool

PublishOrderedContext is PublishOrdered that also abandons the publish when ctx is done. An ordinary stalled subscriber is detached after the delivery timeout, while a lossless subscriber waits for lifecycle cancellation. A caller on a shutdown-critical goroutine -- one something else waits for before the EventBus itself stops, such as a LedgerState the node closes while keeping the bus running for a live restore -- must pass a context it cancels when it needs a shorter bound.

Abandoning is not a drop in the delivery-guarantee sense: the event was never accepted, and the false return says so.

func (*EventBus) RegisterSubscriber added in v0.18.0 

func (e *EventBus) RegisterSubscriber(
	eventType EventType,
	sub Subscriber,
) EventSubscriberId

RegisterSubscriber allows external adapters (e.g., network-backed subscribers) to register with the EventBus. It returns the assigned subscriber id. Returns 0 if the EventBus is stopped or closed.

func (*EventBus) Stop added in v0.18.0 

func (e *EventBus) Stop()

Stop closes all subscriber channels and clears the subscribers map. This ensures that SubscribeFunc goroutines exit cleanly during shutdown. The EventBus can still be reused after Stop() is called.

func (*EventBus) Subscribe 

func (e *EventBus) Subscribe(
	eventType EventType,
) (EventSubscriberId, <-chan Event)

Subscribe allows a consumer to receive events of a particular type via a channel. Returns (0, nil) if the EventBus is stopped or closed.

func (*EventBus) SubscribeFunc 

func (e *EventBus) SubscribeFunc(
	eventType EventType,
	handlerFunc EventHandlerFunc,
) EventSubscriberId

SubscribeFunc allows a consumer to receive events of a particular type via a callback function. Returns 0 if the EventBus is stopped or closed.

func (*EventBus) SubscribeFuncStrict added in v0.70.5 

func (e *EventBus) SubscribeFuncStrict(
	eventType EventType,
	buffer int,
	backpressurePolicy SubscriberBackpressurePolicy,
	handlerFunc EventHandlerFunc,
	onPanic EventPanicHandler,
) EventSubscriberId

SubscribeFuncStrict is like SubscribeFuncWithBufferPolicy, but for handlers that implement state-machine logic where silently continuing past a failed event is unsafe -- e.g. a handler that mutates state derived from a monotonic stream of chain events, where the next event's correctness depends on the previous one having actually been applied. A handler panic is never absorbed the way it is for a plain SubscribeFunc handler:

  • it is still recovered and logged, so it cannot crash the node;
  • onPanic (if non-nil) is invoked with the event and the recovered panic value, on the subscription's own dispatch goroutine, so the owning component can react (log with its own context, alert, mark itself degraded);
  • the subscription is then torn down and its dispatch goroutine exits, so no further event is delivered into (and silently dropped by) a handler that just proved its state may be inconsistent.

backpressurePolicy is independent of the panic behavior above and must be chosen deliberately: SubscriberBackpressureDetach can silently drop this subscription on ordinary backpressure alone, with the same permanent loss of future events as a panic would cause, since nothing here re-subscribes. Use SubscriberBackpressureBlock for an ordering-critical stream whose omission would leave the owning component's state stale in a way it cannot safely recover from -- chainselection.NewChainSelector's PeerRollbackEvent subscription is the concrete example.

Use SubscribeFunc/SubscribeFuncWithBuffer instead for handlers that are safe to keep retrying on the next event -- internal/dblifecycle.Manager is the concrete example this distinction exists for: a panic in one epoch's snapshot attempt must not stop automatic snapshots for every later epoch, so it deliberately relies on the plain SubscribeFunc behavior instead.

func (*EventBus) SubscribeFuncWithBuffer added in v0.41.0 

func (e *EventBus) SubscribeFuncWithBuffer(
	eventType EventType,
	buffer int,
	handlerFunc EventHandlerFunc,
) EventSubscriberId

SubscribeFuncWithBuffer is like SubscribeFunc but lets the caller pick the per-subscriber channel buffer. See SubscribeWithBuffer for details.

func (*EventBus) SubscribeFuncWithBufferPolicy added in v0.70.2 

func (e *EventBus) SubscribeFuncWithBufferPolicy(
	eventType EventType,
	buffer int,
	backpressurePolicy SubscriberBackpressurePolicy,
	handlerFunc EventHandlerFunc,
) EventSubscriberId

SubscribeFuncWithBufferPolicy is SubscribeFuncWithBuffer with an explicit stalled subscriber policy. Use SubscriberBackpressureBlock only when detaching the subscriber would leave its owning component unable to recover safely.

func (*EventBus) SubscribeWithBuffer added in v0.41.0 

func (e *EventBus) SubscribeWithBuffer(
	eventType EventType,
	buffer int,
) (EventSubscriberId, <-chan Event)

SubscribeWithBuffer is like Subscribe but lets the caller pick the per-subscriber channel buffer. Use this for subscribers that need to tolerate bursts larger than DefaultSubscriberBuffer (e.g. chainsync or blockfetch ingest during bulk catch-up). A non-positive buffer falls back to DefaultSubscriberBuffer.

func (*EventBus) SubscribeWithBufferPolicy added in v0.70.2 

func (e *EventBus) SubscribeWithBufferPolicy(
	eventType EventType,
	buffer int,
	backpressurePolicy SubscriberBackpressurePolicy,
) (EventSubscriberId, <-chan Event)

SubscribeWithBufferPolicy is SubscribeWithBuffer with an explicit stalled subscriber policy. Use SubscriberBackpressureBlock only when detaching the subscriber would leave its owning component unable to recover safely.

func (*EventBus) Unsubscribe 

func (e *EventBus) Unsubscribe(eventType EventType, subId EventSubscriberId)

Unsubscribe stops delivery of events for a particular type for an existing subscriber

func (*EventBus) UnsubscribeAndWait added in v0.69.0 

func (e *EventBus) UnsubscribeAndWait(
	eventType EventType,
	subId EventSubscriberId,
)

UnsubscribeAndWait is like Unsubscribe, but for SubscribeFunc/ SubscribeFuncWithBuffer subscribers it additionally blocks until that subscriber's dispatch goroutine has fully exited -- including finishing any handler call already in flight when this is called. Plain Subscribe/ SubscribeWithBuffer subscribers have no bus-owned goroutine, so this behaves exactly like Unsubscribe for them.

Use this wherever a caller unsubscribes and then, in the same teardown sequence, mutates or discards state that the handler closure reads without its own synchronization (e.g. a component field nilled out right after Close()) -- plain Unsubscribe only stops *future* deliveries, so a handler goroutine that already dequeued an event can still be executing concurrently with that teardown. Do not call this from within the subscriber's own handler: waiting for a goroutine to exit from inside that same goroutine deadlocks forever.

Safe to call concurrently with a plain Unsubscribe for the same subId (e.g. from two different teardown paths racing each other): whichever call actually removes the e.subscribers entry, this one still finds and waits on the subscriber via channelSubsById, so the race cannot turn this into a no-op.

func (*EventBus) UnsubscribeAndWaitContext added in v0.70.0 

func (e *EventBus) UnsubscribeAndWaitContext(
	ctx context.Context,
	eventType EventType,
	subId EventSubscriberId,
) error

UnsubscribeAndWaitContext is UnsubscribeAndWait with the wait for an in-flight handler bounded by ctx, returning ctx.Err() if the handler is still running when ctx is done.

The unsubscribe itself is never skipped or bounded: it is synchronous and runs even for an already-canceled ctx, so future deliveries stop either way and only the wait can be cut short. A caller that gets a non-nil error therefore knows the handler may still be executing concurrently with whatever it does next, which is the whole reason to wait -- prefer UnsubscribeAndWait wherever there is no deadline to honor.

Use this from a shutdown path that must respect a deadline: plain UnsubscribeAndWait blocks for as long as the handler runs, which lets a single stuck handler overrun a bounded shutdown.

type EventHandlerFunc 

type EventHandlerFunc func(Event)

type EventPanicHandler added in v0.70.5 

type EventPanicHandler func(Event, any)

EventPanicHandler is invoked by SubscribeFuncStrict, on the subscription's own dispatch goroutine, after a handler panic has been recovered and logged. It receives the event that was being processed and the recovered panic value.

type EventSubscriberId 

type EventSubscriberId int

type EventType 

type EventType string

type HardForkEvent added in v0.22.0 

type HardForkEvent struct {
	// Slot is the slot at which the hard fork takes effect
	Slot uint64
	// EpochNo is the epoch number where the hard fork occurs
	EpochNo uint64
	// FromEra is the era ID before the hard fork
	FromEra uint
	// ToEra is the era ID after the hard fork
	ToEra uint
	// OldMajorVersion is the protocol major version before
	OldMajorVersion uint
	// OldMinorVersion is the protocol minor version before
	OldMinorVersion uint
	// NewMajorVersion is the protocol major version after
	NewMajorVersion uint
	// NewMinorVersion is the protocol minor version after
	NewMinorVersion uint
}

HardForkEvent is emitted when a hard fork (era transition) occurs at an epoch boundary due to a protocol version change

type Subscriber added in v0.18.0 

type Subscriber interface {
	Deliver(Event) error
	Close()
}

Subscriber is a delivery abstraction that allows the EventBus to deliver events to in-memory channels and to network-backed subscribers via the same interface. Implementations must ensure Close() is idempotent and safe to call multiple times.

type SubscriberBackpressurePolicy added in v0.70.2 

type SubscriberBackpressurePolicy uint8

SubscriberBackpressurePolicy decides what happens when a channel-backed subscriber stays full past the delivery timeout. 

const (
	// SubscriberBackpressureDetach removes a subscriber that is no longer
	// making progress. It is the default for ordinary asynchronous consumers.
	SubscriberBackpressureDetach SubscriberBackpressurePolicy = iota
	// SubscriberBackpressureBlock keeps a lossless subscriber attached until it
	// drains or normal lifecycle cancellation closes it. Use this for a stream
	// whose omission would make the owning component unable to recover safely.
	SubscriberBackpressureBlock
)

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.