Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Service ¶
type Service struct {
// CH indexes per-id state. Kept exported for the lifecycle tests
// in this package that assert reclamation behaviour.
CH map[string]*idGroup
// contains filtered or unexported fields
}
Service is the verifier's SSE fanout layer. Public shape unchanged from the original per-process broadcaster:
- OpenListener(id) returns a channel the HTTP handler reads to deliver one SSE message to the client.
- CloseListener(id, listener) tears down a single reader.
- Submit(id, msg) delivers msg to every live listener on id.
Internally the message bus is a pubsub.PubSub (MemoryPubSub by default; Redis / Valkey when HA is configured). Multiple SSE listeners on one node for the same session share a single backend subscription via a ref-counted idGroup; a fan-out goroutine forwards each decoded message from that subscription to every local listener channel.
func New ¶
New builds a Service wired to a MemoryPubSub. Standalone deployments use this constructor; HA deployments should prefer NewWithBus so the backend matches cfg.Common.HA.PubSub.
func NewWithBus ¶ added in v0.7.21
func NewWithBus(_ context.Context, cfg *model.Cfg, log *logger.Log, bus pubsub.PubSub) (*Service, error)
NewWithBus builds a Service around the supplied pub/sub bus. The caller owns bus.Close - Service.Close only tears down its own subscriptions, not the bus itself, so one bus may back multiple Services if that ever becomes useful.
func (*Service) Close ¶
Close terminates every outstanding group and closes every live listener channel so handlers blocked on <-listener wake up with a closed channel instead of hanging. Groups are closed FIRST so each fan-out goroutine exits and deliver is quiescent before any listener channel is closed - otherwise a backend message in flight could select the send branch on a channel we are about to close here and panic. The bus itself is NOT closed here - its lifecycle is owned by the caller that constructed it (today: cmd/verifier).
func (*Service) CloseListener ¶
CloseListener removes one listener registration. The last close on an id reclaims the group and its backend subscription, so a stream of transient ids cannot inflate CH. Idempotent and safe to call after Service.Close has already closed the listener channel: a handler's deferred CloseListener runs in both the normal and the shutdown-race path.
func (*Service) HealthProbe ¶ added in v0.7.21
HealthProbe satisfies the status.Prober contract by delegating to the backing bus. For MemoryPubSub this is always healthy; for the RESP backends it is a PING, so a dead Redis/Valkey surfaces in /health rather than only through silent cross-node delivery failures.
func (*Service) OpenListener ¶
OpenListener registers a new SSE listener for id. If this is the first listener on this node for the id, a backend subscription is opened and a fan-out goroutine is started that decodes backend payloads and forwards them to every local listener channel.
The backend Subscribe handshake runs OUTSIDE s.mu so a slow or hung RESP server cannot block other OpenListener / CloseListener / Service.Close callers for the duration of its network timeout.
The s.CH lookup and g.register of the fast path run under a single s.mu critical section, so a last-listener CloseListener detaching g from s.CH cannot interleave between them. If a concurrent OpenListener wins the race to install a new group, or the group we found was reclaimed before we could register, OpenListener retries the whole lookup instead of attaching to a detached group.
Source Files