Documentation
¶
Overview ¶
Package eventbus implements the in-memory invalidation broadcast bus described in docs/multi-device-storage.md §3.5 / §4.1.
The Hub publishes one Event per successful write (insert/update/delete on a domain table) so subscribed peers can drop the matching cache row and re-fetch. Events are best-effort: a slow subscriber whose buffer overflows is dropped, and the peer is expected to reconcile via the resync cursor (`GET /api/v1/changes?since=<seq>`). This is the design call-out in §3.5: "再同期用 cursor: invalidation broadcast の event drop に備え".
The bus is process-local. Cross-peer fan-out is the WebSocket layer's job (see internal/server/events_ws.go), which subscribes once per connected peer and forwards events frame-by-frame.
Concurrency model:
- Publish is non-blocking. It walks the subscriber list under a read lock and tries to enqueue into each subscriber's buffered channel; on a full channel the subscriber is terminated with ReasonOverflow and removed.
- Subscribe returns a *Subscription. The subscription's event channel is closed exactly once: when terminated for any reason (caller Cancel, ctx Done, overflow, or bus Close). The subscription's Context is cancelled at the same moment so any in-flight write derived from it unblocks immediately.
- Reason reports why a subscription ended: caller / ctx / overflow / closed. WebSocket handlers map this onto a close code so a peer can tell "drop, please resync" apart from "server going away" apart from "you cancelled".
- Close drains all subscribers and rejects further Publish/ Subscribe calls.
Buffer sizing: §3.5 promises invalidations are "best effort"; the server pairs this with an HTTP cursor for replay. A small buffer (default 64) is enough to ride out a brief WebSocket write stall; anything bigger just delays the inevitable cursor catch-up.
Index ¶
Constants ¶
const ( // ReasonCaller fires when the holder calls Subscription.Cancel(). // The peer initiated the close; no resync needed. ReasonCaller = "caller" // ReasonContext fires when the ctx passed to Subscribe is cancelled // (typically the HTTP request context — client disconnected). ReasonContext = "context" // ReasonOverflow fires when the per-subscription buffer filled and // Publish had to drop the subscriber. The peer MUST resync via // `GET /api/v1/changes?since=<seq>` to recover lost events. ReasonOverflow = "overflow" // ReasonClosed fires when the bus itself was Closed (server // shutdown). The peer should reconnect after the server returns. ReasonClosed = "closed" )
Reason for ending a subscription. Stable strings — the WebSocket handler maps these onto wire close codes, and changing them would break peer behavior expectations.
const DefaultBuffer = 64
DefaultBuffer is the per-subscriber channel capacity. A subscriber that falls more than this many events behind is dropped.
Variables ¶
var ErrClosed = errors.New("eventbus: closed")
ErrClosed is returned by Publish/Subscribe after Close.
Functions ¶
This section is empty.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is the in-memory broadcast hub. The zero value is NOT usable — call New.
func (*Bus) Close ¶
func (b *Bus) Close()
Close drains all subscribers (terminating each with ReasonClosed) and marks the bus closed. Subsequent Subscribe / Publish return ErrClosed. Idempotent.
func (*Bus) Dropped ¶
Dropped is the total number of subscribers evicted for buffer overflow since the bus was created. Useful for ops dashboards.
func (*Bus) Publish ¶
Publish broadcasts ev to every live subscriber. A full subscriber buffer means the subscriber is too slow — it is terminated with ReasonOverflow (the peer must resync via the cursor). Publish itself never blocks.
Publish on a closed bus returns ErrClosed; otherwise nil.
func (*Bus) Subscribe ¶
func (b *Bus) Subscribe(ctx context.Context) (*Subscription, error)
Subscribe registers a new subscription. ctx, when not nil, is wired to the subscription's lifetime: ctx.Done causes the subscription to terminate with ReasonContext. The watcher goroutine exits as soon as the subscription terminates for any reason (no leak even if ctx outlives the subscription).
func (*Bus) Subscribers ¶
Subscribers returns the current live subscriber count.
type Event ¶
type Event struct {
Table string `json:"table"`
ID string `json:"id"`
ETag string `json:"etag"`
Op string `json:"op"` // insert | update | delete
Seq int64 `json:"seq"`
TS int64 `json:"ts"` // unix millis
}
Event is the on-the-wire invalidation record. Fields mirror §4.1's `{table, id, etag, op}` plus `seq` for the resync cursor (peers ack up to a known seq) and `ts` (unix millis) for staleness debugging.
Seq is the global monotonic seq from store.NextGlobalSeq() — the SAME value the row was written with, so a peer can compare against its local cache's last-seen seq for that table without ambiguity.
ETag intentionally lacks `omitempty`: a delete is wire-distinguished from an update only by Op, and dropping the field would force peers to special-case missing-vs-empty. Always emit a string, always emit it.
type Subscription ¶
type Subscription struct {
// contains filtered or unexported fields
}
Subscription is the live handle returned by Bus.Subscribe.
Lifecycle:
- C() returns the event channel; closes once on termination.
- Context() returns a context that is cancelled at the same moment as C() closes. Use it as the parent of any write operation derived from the channel so the write unblocks immediately on overflow / shutdown.
- Reason() reports why the subscription ended; meaningful only after C() has closed (until then it returns "").
- Cancel() ends the subscription with ReasonCaller. Idempotent.
func (*Subscription) C ¶
func (s *Subscription) C() <-chan Event
C returns the event delivery channel. The channel is closed exactly once when the subscription ends — read until !ok, then call Reason() to learn why.
func (*Subscription) Cancel ¶
func (s *Subscription) Cancel()
Cancel ends the subscription with ReasonCaller. Idempotent.
Cancel routes through the bus's eviction path so the bus mutex serializes channel-close against an in-flight Publish: without that hop, Cancel could close the channel concurrently with Publish's send and trip "send on closed channel".
func (*Subscription) Context ¶
func (s *Subscription) Context() context.Context
Context returns a context that is cancelled when the subscription ends. Callers SHOULD use it as the parent of any blocking I/O that dispatches received events so the I/O unblocks promptly on overflow.
func (*Subscription) Reason ¶
func (s *Subscription) Reason() string
Reason returns the reason the subscription ended, or "" while live. One of the Reason* constants once terminated.
Source Files