event

type BackpressurePolicy ¶ added in v0.1.12

type BackpressurePolicy int

BackpressurePolicy controls what happens when a subscription's buffer is already full at Publish time.

const (
	// DropNewest drops the incoming envelope and keeps existing buffered
	// items. Default policy: prioritises older, presumably already-observed
	// events.
	DropNewest BackpressurePolicy = iota

	// DropOldest drops the oldest buffered envelope to make room for the
	// new one. Use when the latest state is more valuable than history.
	DropOldest

	// Block makes Publish wait until either the buffer has room or the
	// publishing context is cancelled. Use sparingly: a slow subscriber
	// will back-pressure every publisher.
	Block
)

func (p BackpressurePolicy) String() string

type Bus ¶ added in v0.1.12

type Bus interface {
	// Publish delivers env to every matching subscription. Implementations
	// must:
	//   - fill env.ID and env.Time when zero;
	//   - return ErrBusClosed once Close has been invoked;
	//   - guarantee that, on return, the envelope has been enqueued for
	//     every matching subscription, or dropped according to that
	//     subscription's BackpressurePolicy.
	//
	// Cross-process implementations (added in later commits) may relax
	// the on-return delivery guarantee to "fire-and-forget"; that
	// relaxation must be documented at the implementation level.
	Publish(ctx context.Context, env Envelope) error

	// Subscribe creates a new subscription matching pattern. The returned
	// Subscription is closed automatically when ctx is cancelled.
	// pattern is validated and a malformed value returns an error.
	Subscribe(ctx context.Context, pattern Pattern, opts ...SubOption) (Subscription, error)

	// Close shuts down the bus. After Close returns, every subscription
	// channel will be closed and subsequent Publish / Subscribe calls
	// return ErrBusClosed. Close is idempotent.
	Close() error
}

Bus is a publish-subscribe channel for Envelopes routed by Subject / Pattern.

type DropReason ¶ added in v0.1.12

type DropReason int

DropReason explains why an envelope was discarded for a particular subscription.

const (
	// DropReasonBufferFull indicates the subscription buffer was full and
	// the active BackpressurePolicy chose to drop.
	DropReasonBufferFull DropReason = iota

	// DropReasonClosed indicates Publish raced with subscription close;
	// the envelope was not delivered.
	DropReasonClosed
)

func (r DropReason) String() string

type Envelope ¶ added in v0.1.12

type Envelope struct {
	ID      string            `json:"id"`
	Subject Subject           `json:"subject"`
	Time    time.Time         `json:"time"`
	Source  string            `json:"source,omitempty"`
	TraceID string            `json:"trace_id,omitempty"`
	SpanID  string            `json:"span_id,omitempty"`
	Headers map[string]string `json:"headers,omitempty"`
	Payload json.RawMessage   `json:"payload,omitempty"`
}

Envelope is the cross-process-friendly carrier for a single event.

Every field serialises cleanly to JSON. Payload is stored as json.RawMessage so:

1. in-memory and remote bus implementations behave identically (bytes in, bytes out — no any-typed surprises);
2. server-side persistence and SSE forwarding can avoid an extra round of marshal/unmarshal;
3. consumers decide when to decode (and into what concrete type).

Use NewEnvelope / MustEnvelope to construct envelopes; use Decode to extract the payload into a typed Go value.

func MustEnvelope(ctx context.Context, subject Subject, payload any) Envelope

func NewEnvelope(ctx context.Context, subject Subject, payload any) (Envelope, error)

func (e Envelope) ActorID() string

func (e Envelope) Decode(out any) error

func (e Envelope) GraphID() string

func (e Envelope) Header(key string) string

func (e Envelope) NodeID() string

func (e Envelope) RunID() string

func (e *Envelope) SetActorID(id string)

func (e *Envelope) SetGraphID(id string)

func (e *Envelope) SetHeader(key, value string)

func (e *Envelope) SetNodeID(id string)

func (e *Envelope) SetRunID(id string)

func (e *Envelope) SetTenant(id string)

func (e Envelope) Tenant() string

type MemoryBus ¶

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

MemoryBus is the in-process Bus implementation.

Concurrency model:

• subscribers map is guarded by mu;
• Publish takes RLock for the route-and-enqueue scan;
• Subscribe / Close / removeSub take the write lock;
• in-flight Publish calls are tracked by inflight (sync.WaitGroup) so Close can wait for them to drain before any subscriber channel is closed — this avoids the "send on closed channel" race that bit earlier per-subscription bus implementations.

Hot-path lock-freedom:

• Observer callbacks are deferred to a local slice and executed after the RLock has been released.

Route cache:

• routeCache memoises subject → []*memSub so a hot subject avoids the O(N) match loop. Guarded by routeCacheMu (a separate mutex from b.mu so cache writes don't compete with the publish RLock fan-out).
• Lock order when both are held: ALWAYS b.mu first, then routeCacheMu. Subscribe / Close / removeSub clear the cache while holding b.mu.Lock(); Publish reads/writes the cache while holding b.mu.RLock(). This ordering prevents a writer's clear from racing a publisher's lookup of stale entries.

func NewMemoryBus(opts ...MemoryBusOption) *MemoryBus

func (b *MemoryBus) Close() error

func (b *MemoryBus) Dropped() int64

func (b *MemoryBus) Publish(ctx context.Context, env Envelope) error

func (b *MemoryBus) Subscribe(ctx context.Context, pattern Pattern, opts ...SubOption) (Subscription, error)

type MemoryBusOption ¶

type MemoryBusOption func(*MemoryBus)

MemoryBusOption configures a MemoryBus at construction time.

func WithObserver(o Observer) MemoryBusOption

func WithRouteCacheSize(n int) MemoryBusOption

type NoopBus ¶

type NoopBus struct{}

NoopBus is a Bus that discards every Publish and yields a closed channel for every Subscribe. Useful for tests and as a default value to keep nil-checks out of producer code.

func NewNoopBus() NoopBus

func (NoopBus) Close() error

func (NoopBus) Publish(context.Context, Envelope) error

func (NoopBus) Subscribe(context.Context, Pattern, ...SubOption) (Subscription, error)

type Observer ¶ added in v0.1.12

type Observer interface {
	// OnPublish fires once per Publish call, before any delivery decision.
	OnPublish(env Envelope)
	// OnDeliver fires after a successful enqueue into a subscription's
	// channel.
	OnDeliver(subID SubscriptionID, env Envelope)
	// OnDrop fires when an envelope is discarded for a specific
	// subscription, identifying the reason.
	OnDrop(subID SubscriptionID, env Envelope, reason DropReason)
}

Observer is the bus-level instrumentation hook. It carries the OnPublish / OnDeliver / OnDrop lifecycle for every envelope, giving callers a single place to wire metrics / tracing / drop accounting.

Concurrency contract — every Bus implementation must guarantee:

• Observer methods are invoked outside any bus-internal lock.
• Implementations must return promptly. They MUST NOT call back into Bus methods (Publish / Subscribe / Close); doing so risks deadlock and the bus is allowed to detect such calls and panic in debug builds.
• Observer methods may be invoked concurrently from multiple goroutines; implementations must be safe for concurrent use.

Ordering:

• For a single Publish call, OnPublish is invoked exactly once, followed by zero or more OnDeliver / OnDrop calls (one per matching subscription).
• The relative order of OnDeliver / OnDrop calls within one Publish reflects the bus's internal subscription scan and is therefore implementation-defined. MemoryBus, for example, memoises the match result per Subject so a hot subject sees a stable callback order across publishes — observers MUST NOT rely on that order being either stable or random.

Observer is intentionally an interface (not a func) so a single hook can correlate the three lifecycle moments without ad-hoc closures.

type Pattern ¶ added in v0.1.12

type Pattern string

Pattern is a Subject matcher using NATS-style wildcards:

• matches exactly one segment > matches one or more trailing segments (must be the last segment)

Examples:

graph.run.r1.>             every event for run r1
graph.run.*.node.*.start   every node start across runs
kanban.>                   every kanban event

Pattern matching is case-sensitive.

func (p Pattern) Matches(s Subject) bool

func (p Pattern) Validate() error

type SubOption ¶ added in v0.1.12

type SubOption func(*subOptions)

SubOption configures a Subscription at creation time.

func WithBackpressure(p BackpressurePolicy) SubOption

func WithBufferSize(n int) SubOption

func WithPredicate(fn func(Envelope) bool) SubOption

type Subject ¶ added in v0.1.12

type Subject string

Subject is a dot-delimited routing key, e.g.:

graph.run.r1.start
graph.run.r1.node.n1.complete
kanban.board.b1.update

Segments are separated by '.'. A segment must not contain '.', '*' or '>' and must not be empty.

func (s Subject) Validate() error

type Subscription ¶

type Subscription interface {
	// ID is unique within the originating bus instance.
	ID() SubscriptionID
	// C returns the envelope channel. The channel is closed exactly once,
	// either when Close is invoked or when the parent bus is closed.
	C() <-chan Envelope
	// Close cancels the subscription. Idempotent.
	Close() error
}

Subscription is an active receive-side handle.

type SubscriptionID ¶ added in v0.1.12

type SubscriptionID string

SubscriptionID identifies a subscription within a single Bus instance. It is opaque to consumers; bus implementations choose the format.