Documentation
¶
Overview ¶
Package events implements the R2 dashboard's SSE broker (plan task t04-sse-broker): an in-process fan-out from event sources to per-client bounded buffers.
Sources are (a) the local Publisher seam — t06's filesystem watcher publishes here pre-R3 — and (b) an R3 EventBus attached via Broker.AttachBus (bridge.go), which binds ONLY to the D4.1 EventBus interface, never to a concrete backend, and assumes only the D4.2 G1–G4 floor (at-most-once best-effort delivery, best-effort per-topic order, bounded-buffer non-blocking publish, no cross-publish atomicity).
Backpressure implements spec OQ5's recommendation ("bounded buffer with disconnect-on-overflow, client refetches on reconnect"), which is also API.md §3.7's pinned resolution: each subscriber holds one bounded buffered channel, and EVERY drop is terminal — the first event that cannot be enqueued disconnects that subscriber (channel closed), so the SSE client reconnects and refetches. A live stream therefore never contains a silent hole: drop ⇒ disconnect ⇒ refetch, with no window in which a client keeps consuming past a lost event. Publishers never block (G3-compatible).
The broker also owns the store-cache eviction wiring point: on every pushed event (never on heartbeats) it invalidates the t02 read cache via the Evictor hooks (store.DiskStore's per-root Evict / whole-cache EvictAll) BEFORE fan-out, so a client refetching in reaction to the event always sees post-event state.
Anti-scope: no HTTP (t05 owns the SSE endpoint), no event production (t06 fswatch / R3 bus via t13), no replay (spec D2.2: reconnect+refetch).
Index ¶
- Constants
- Variables
- type Broker
- func (b *Broker) AttachBus(bus svcevents.EventBus, topics ...string) (func(), error)
- func (b *Broker) AttachR3Bus(bus svcevents.EventBus, opts ...R3Option) (func(), error)
- func (b *Broker) Close()
- func (b *Broker) Publish(topic string, payload any)
- func (b *Broker) Subscribe(ctx context.Context) (<-chan Event, func())
- func (b *Broker) SubscriberCount() int
- type Event
- type Evictor
- type IterLogRooter
- type Options
- type Publisher
- type R3Option
- type SessionResolver
Constants ¶
const ( // TopicIterationScored — a new iter-N.score.yaml sidecar appeared (spec R5). TopicIterationScored = "iteration.scored" // TopicSessionUpdated — a session score sidecar was rewritten (spec R6). TopicSessionUpdated = "session.updated" // TopicScoreRecomputed — an existing iteration score changed in place. TopicScoreRecomputed = "score.recomputed" // TopicRubricChanged — rubric recomputed under a NEW version (spec R7). TopicRubricChanged = "rubric.changed" // TopicHeartbeat — keepalive emitted by the broker itself. TopicHeartbeat = "heartbeat" )
Dashboard SSE topics — the API.md §3.7 event taxonomy carried in the SSE `event:` field and validated by schemas/dashboard-event.schema.json.
const ( // DefaultBuffer is the per-subscriber bounded channel capacity. DefaultBuffer = 64 // DefaultHeartbeat is the keepalive interval (API.md §3.7: 15s). DefaultHeartbeat = 15 * time.Second )
Defaults applied by New when the corresponding Options field is zero.
Variables ¶
var ErrClosed = errors.New("dashboard/events: broker closed")
ErrClosed is returned by AttachBus after the broker has been closed.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker fans events out to per-client bounded buffers. It implements Publisher. Construct with New; the zero value is not usable.
func (*Broker) AttachBus ¶
AttachBus subscribes the broker to the given topics on an R3 event bus and fans every received event out to the broker's SSE subscribers, running the store-cache eviction hook exactly as for local publishes.
Per the t04 rescope (§2A coherence), this binds ONLY to the D4.1 events.EventBus INTERFACE — never the concrete *events.Bus — and assumes only the D4.2 G1–G4 floor, so a config-selected backend (spec D4.4) can replace the builtin without touching this broker. Bridged events keep the source bus timestamp (normalized to the schema's whole-second UTC form).
The returned detach func unsubscribes every topic and is idempotent; Close also runs it. Attaching to a closed broker returns ErrClosed. Errors from bus.Subscribe (e.g. a closed bus) abort the attach and unwind any subscriptions already made.
func (*Broker) AttachR3Bus ¶
AttachR3Bus subscribes the broker to the R3 EventBus for every topic with a dashboard SSE surface, translates each received event into the API.md §3.7 taxonomy, evicts the affected store-cache scope, and fans the translated event out to SSE subscribers — the same eviction+fan-out path a local Publish takes.
The returned detach func unsubscribes every topic and is idempotent; Close also runs it. Attaching to a closed broker returns ErrClosed. An error from bus.Subscribe aborts the attach and unwinds any subscriptions already made.
func (*Broker) Close ¶
func (b *Broker) Close()
Close disconnects every subscriber, detaches any attached buses, stops the heartbeat, and waits for all broker goroutines to exit. Idempotent. Subsequent Publish calls are no-ops.
func (*Broker) Publish ¶
Publish implements Publisher: evicts the store cache for the event, then fans it out to every subscriber. Non-blocking for the publisher; delivery to any individual subscriber is at-most-once (G1). Publishing on a closed broker is a complete no-op (no eviction, no fan-out).
func (*Broker) Subscribe ¶
Subscribe registers a client and returns its bounded event stream plus an idempotent cancel func. The stream closes on cancel, on context cancellation, on broker Close, or the moment the client's buffer overflows (OQ5 disconnect-on-overflow — the client must then reconnect and refetch; no event is ever silently skipped on a live stream). Subscribing on a closed broker returns an already-closed stream and a no-op cancel.
func (*Broker) SubscriberCount ¶
SubscriberCount reports the live subscriber count (the health endpoint's subscriber_count field, API.md §3.6).
type Event ¶
type Event struct {
// Type is the topic (event taxonomy constant).
Type string `json:"type"`
// Seq is the monotonic per-connection sequence, starting at 0 and
// resetting on reconnect. It is NOT a durable cursor (no replay).
Seq uint64 `json:"seq"`
// TS is the publish time, stamped in UTC truncated to whole seconds
// so json.Marshal emits the schema's RFC3339 "Z" pattern verbatim.
TS time.Time `json:"ts"`
// Payload is the topic-specific thin key set; opaque to the broker.
Payload any `json:"payload"`
}
Event is one message delivered to a broker subscriber. Field names and JSON tags mirror schemas/dashboard-event.schema.json — t05 marshals this struct into the SSE frame's data: field (`event:` = Type, `id:` = Seq).
type Evictor ¶
type Evictor interface {
Evict(root string)
EvictAll()
}
Evictor is the t02 store-cache invalidation hook. *store.DiskStore satisfies it (per-root Evict + whole-cache EvictAll push hooks).
type IterLogRooter ¶
type IterLogRooter interface {
IterLogRoot() string
}
IterLogRooter is optionally implemented by event payloads that identify the iter-log root they concern. When a payload reports a root the broker evicts only that root's cache snapshot; otherwise it falls back to whole-cache eviction (correct but coarser).
type Options ¶
type Options struct {
// Buffer is the per-subscriber channel capacity; <=0 → DefaultBuffer.
Buffer int
// Heartbeat is the keepalive interval; 0 → DefaultHeartbeat,
// negative → heartbeats disabled (tests, embedded use).
Heartbeat time.Duration
// Evictor, when non-nil, receives cache invalidations before fan-out.
Evictor Evictor
}
Options configures a Broker. The zero value yields the defaults.
type Publisher ¶
Publisher is the broker's local publish seam. Pre-R3 producers (t06's filesystem watcher) hold this minimal shape; the r3-facing seam is the events.EventBus interface, bridged via AttachBus (wired in t13).
type R3Option ¶
type R3Option func(*r3Config)
R3Option configures AttachR3Bus.
func WithR3SessionResolver ¶
func WithR3SessionResolver(r SessionResolver) R3Option
WithR3SessionResolver sets the resolver used to fill the dashboard iteration.scored payload's session_id from the iteration's on-disk record. A nil resolver is ignored (the zero-value resolver yields "").
type SessionResolver ¶
SessionResolver resolves an iteration's dashboard session id from its iter-log directory and iteration number. R3's iteration.scored payload does not carry the session id (the dashboard event schema keys query invalidation on it), so the composing runtime injects a resolver — see internal/dashboard/server.Mount, which resolves it from the iter-<n>.yaml record. Best-effort: return "" when the session id cannot be determined.
Source Files