Documentation
¶
Overview ¶
Package stream holds the SSE fan-out primitives shared by the streaming API: the Subscriber connection handle (subscriber.go), the Bucket fan-out primitive (bucket.go), the Heartbeater keepalive wheel (heartbeat.go) that nudges idle GET /v1/stream connections, and the event Hub (hub.go).
The Hub is the #294 delivery hot path: subscribers register under (topic, role), and each event is projected — column-filtered per the role's policy — and serialized ONCE per role, then pushed through the same Subscriber queue the keepalive wheel uses, so the handler drains both from a single byte-pump. That collapses the prior per-subscriber unmarshal/project/re-serialize into one pass per distinct (role, table) output shape. The one per-subscriber decision left is row-level security (#319): a role whose policy carries a row-filter shares the column projection but delivers each event only to the subscribers whose JWT claims admit the row.
Index ¶
Constants ¶
const ( KindKeepalive = "keepalive" // wheel keepalive comment KindEvent = "event" // live event off the hub KindReplay = "replay" // historical event from gap-fill on (re)connect )
Frame kinds for the "kind" attribute on the frame and byte counters.
Variables ¶
This section is empty.
Functions ¶
func NumericSpecOf ¶
func NumericSpecOf(st discovery.NumericStorage) policy.NumericSpec
NumericSpecOf renders discovery's storage classification as the policy evaluator's storage model. Exported so the tests/integration differential oracle builds specs through the very mapping production uses — one source, so the oracle can't keep validating a mapping the Hub no longer applies.
Types ¶
type Bucket ¶
type Bucket interface {
Add(sub *Subscriber)
Remove(sub *Subscriber)
Len() int
Push(f Frame)
Snapshot() []*Subscriber
}
Bucket is a concurrency-safe set of subscribers. Push fans one Frame out to every member fire-and-forget (the keepalive wheel and the Hub's no-row-filter fast path; Send itself counts any queue-full drop); Snapshot exposes the members so the event Hub can evaluate row visibility per subscriber before sending (and, later, evict).
type Frame ¶
type Frame struct {
Kind string // a Kind* constant: keepalive, event, or replay
Data []byte // the exact bytes written to the client
}
Frame is one ready-to-write SSE byte frame tagged with its kind, so the handler can label the write (frames_sent_total{kind=…}) at the moment it happens — a frame dropped by a full queue is never counted as sent. Producers (the keepalive wheel, the event Hub) fan frames in with Send; the handler writes Data verbatim.
type Heartbeater ¶
type Heartbeater struct {
// contains filtered or unexported fields
}
Heartbeater is a timing wheel that keepalives every live subscriber from one timer: subscribers spread across a ring of buckets, one bucket fired per tick, so each is nudged once per period while only ~1/buckets are written per tick — spreading the load instead of writing every connection at the same instant.
func NewHeartbeater ¶
func NewHeartbeater(period time.Duration, buckets int) *Heartbeater
NewHeartbeater builds the keepalive wheel. period is the effective per-connection interval (stream.keepalive_interval); buckets spreads that work across it, so the per-tick interval is period/buckets and one rotation spans the period. Non-positive inputs fall back to the package defaults.
func (*Heartbeater) Add ¶
func (hb *Heartbeater) Add(sub *Subscriber)
Add registers sub in the bucket that fires last (just behind the hand), giving it a full period before its first keepalive.
func (*Heartbeater) Len ¶
func (hb *Heartbeater) Len() int
Len reports the live subscriber count across the ring (for tests and metrics).
func (*Heartbeater) Remove ¶
func (hb *Heartbeater) Remove(sub *Subscriber)
Remove deregisters sub from its bucket. A no-op if it was never added, so the handler's deferred Remove is always safe.
func (*Heartbeater) Run ¶
func (hb *Heartbeater) Run(ctx context.Context)
Run drives the wheel until ctx is cancelled. Run it in its own goroutine for the lifetime of the server.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub fans live events out to SSE subscribers. Column projection is serialized ONCE per (topic, role) instead of once per subscriber — the #294/#353 lever — because column visibility depends solely on the role+table policy entry, never on JWT claims, so the projected frame is identical for every subscriber of a role.
Row-level security is the exception: a role's row-filter (RLS predicate) is resolved against each subscriber's claims, so two subscribers of the same role can be entitled to different rows. For a role that carries a row-filter, Broadcast therefore keeps the shared column projection but evaluates row visibility PER subscriber (ResolvedPermissions.RowVisible) before delivering — closing the query/stream RLS drift in #319. Roles without a row-filter keep the pure once-per-role fast path unchanged. See projectColumns.
func NewHub ¶
NewHub builds an event hub. A nil policy store passes every event through unfiltered (the unwired-tests case); a non-nil store whose Get returns nil is a total lockout (a deleted/absent policy denies everyone). A nil registry leaves every column's type unknown, so row-filter comparison degrades FAIL-CLOSED: equality/set predicates admit only a byte-identical value and ordering/!= admit nothing (see policy.ColumnKind); metric may be nil.
func (*Hub) Add ¶
func (h *Hub) Add(topic, role string, sub *Subscriber)
Add registers sub to receive events for (topic, role), creating the role bucket (and topic) on first use.
func (*Hub) Broadcast ¶
Broadcast projects raw — a published EventMessage JSON delivered on topic — and fans the finished SSE frame to each subscribed role's bucket. The column projection (decode, evaluate, marshal) happens once per distinct role. For a role that carries a row-level-security filter, that shared frame is still delivered only to the subscribers whose claims admit this row (evaluated per subscriber); a role without a filter takes the pure once-per-role fast path.
func (*Hub) Remove ¶
func (h *Hub) Remove(topic, role string, sub *Subscriber)
Remove deregisters sub from (topic, role), garbage-collecting the bucket and the topic once empty. A no-op if the registration is already gone, so the handler's deferred Remove is always safe.
func (*Hub) ReplayProjector ¶
ReplayProjector returns the projection function for one connection's gap-fill: each call projects a single replayed event for the connection's role+claims into a ready-to-write replay frame, or ok=false to skip it (denied table, invalid payload, or a row the claims aren't entitled to see). It is a Hub method so replay shares the Hub's policy store and schema registry with the live fan-out — the handler can't accidentally project replay against a different (or nil) policy. Replay is already per-connection, so row-level security evaluates against this connection's claims directly; the returned closure holds one policy snapshot for the whole gap-fill (matching Broadcast's one-snapshot-per-event — a reload landing mid-replay applies from the first live event) and caches the per-table column-kind lookup across the replay loop, so a large Last-Event-ID gap-fill doesn't pay a store read-lock plus a registry lookup and map build per event. The closure is for a single goroutine — each connection makes its own. The live path uses Broadcast.
type Metrics ¶
type Metrics struct {
// contains filtered or unexported fields
}
Metrics records SSE stream activity: how many streams are open, how long they last, and how much is written out. The zero value (a nil *Metrics) is a no-op, so the handler can hold one unconditionally and tests can skip wiring it.
func NewMetrics ¶
func NewMetrics() *Metrics
NewMetrics builds the SSE instruments on the global meter provider. Call it after observability.InitProvider, otherwise the instruments bind to the no-op default provider for the process lifetime.
func (*Metrics) ConnClosed ¶
ConnClosed records a stream closing and its total lifetime.
func (*Metrics) ConnOpened ¶
func (m *Metrics) ConnOpened()
ConnOpened records a newly established stream.
func (*Metrics) FrameDropped ¶
FrameDropped records one frame dropped because a subscriber's queue was full — the slow-consumer signal that was silent before #294. kind is a Kind* constant.
func (*Metrics) FrameSent ¶
FrameSent records one frame of the given kind (a Kind* constant) and its size in bytes.
func (*Metrics) RowWithheld ¶
RowWithheld records one event row withheld from one subscriber (live or replay) by the role's row-level-security filter, including fail-closed evaluations. Labeled by table and role (policy-bounded, not data-bounded) so an operator can tell "no matching rows" from "a misconfigured filter withholding everything".
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber is one SSE connection's outbound side: a queue of ready-to-write frames the HTTP handler drains to the client. Producers fan frames in via Send.
func NewSubscriber ¶
func NewSubscriber(claims map[string]any, m *Metrics) *Subscriber
NewSubscriber returns a Subscriber ready to register with a Heartbeater and the event Hub, carrying the connection's JWT claims (nil for a tokenless caller), deep-copied — see the claims field for the snapshot rationale — and the shared stream metrics (nil-safe) that Send counts drops on.
func (*Subscriber) Evicted ¶
func (s *Subscriber) Evicted() <-chan struct{}
Evicted is closed when the subscriber has been marked for disconnection. The handler selects on it to tear the connection down. Inert until the slow-consumer follow-up wires the threshold that closes it.
func (*Subscriber) Frames ¶
func (s *Subscriber) Frames() <-chan Frame
Frames is the queue of ready-to-write frames; the handler writes whatever arrives here to the client verbatim.
func (*Subscriber) Send ¶
func (s *Subscriber) Send(f Frame) bool
Send enqueues one frame without blocking, returning false if the queue is full (a slow consumer). The drop is counted here, labeled by the frame's kind, so no producer can forget to; callers may ignore the result (a keepalive that drops coalesces harmlessly — the full queue keeps the stream alive anyway).
Source Files