Documentation
¶
Overview ¶
Package livetail carries a watched session's live output from the Host that owns it to the ClientLink viewers watching it on this replica. It is Gap 3.
What it composes, and what it adds ¶
Almost nothing here is new. routing.Relay already owns the bounded queues, the contiguity tracking, the session.reset and the A7.3 repair; routing.Demand already decides when a watched session is bound; the HostLink pool already holds the one connection per Host. What was missing is three edges:
- a HostLink SUBSCRIPTION to the session's channel, made after a successful bind on the same connection (hostlink.Subscriber) -- Plane.Bind;
- a ClientLink PUBLISH to `session:{tenant}:{session}`, the channel a viewer already subscribes to (Viewers) -- Plane.Publish;
- something to drive the Relay between them -- the per-session mailbox.
The rule the whole package serves ¶
A Host keeps NO history. Whatever it publishes while this replica is not subscribed is gone, and a viewer does not know. So every time a session's tail STARTS -- except the one start that happens inside the first viewer's own subscribe, which that viewer's durable read covers -- every viewer is sent a session.reset, AFTER the new tail is live, so its repair reads the journal only once nothing further can be missed. And every time a tail STOPS without being asked to, the Relay's HostLinkClosed repair runs: it resets every viewer and re-binds, and re-binding re-subscribes.
Reconnect ordering ¶
centrifuge-go resubscribes on its own after a reconnect, before anything is re-bound, and a Host refuses that (MaySubscribe needs the bind on the SAME connection). The link withdraws every subscription on the drop and tells this package Ended; this package runs the repair, whose re-bind fails fast while the link reconnects; the link then says Restored once the new connection has negotiated, and this package re-binds -- which subscribes -- and the new tail going live sends the reset. Re-bind, then subscribe, then reset: in that order, owned here.
Locking ¶
Plane.mu is a LEAF: nothing is called while it is held except starting a goroutine. That is what lets the transport's callbacks (hostlink sinks) and routing.Demand (Watcher, called under Demand's lock) reach it without ever being part of a cycle. The Relay is called only from a session's drainer goroutine, which holds no lock of this package while it does so -- and that is ALSO what satisfies routing.Relay's precondition that calls for one session are serialised (it releases its own lock across a repair's I/O and no longer orders them itself). pushLocked starts a drainer only when none is running, so one session never has two; keep it that way.
# What IS held across a Host round trip, stated because an earlier version # of this comment claimed nothing was
routing.Demand's one mutex is held across a bind: Demand.Acquire, a poll and Demand.Rebind all bind under it, and Plane.Bind -- the Binder -- makes the bind RPC and then waits for the subscribe's answer. So a Host slow to answer delays other sessions' BINDS and polls on this replica. It does not delay their DELIVERY: a drainer's Receive and Pump take only the Relay's mutex, and since v0.4.0's fix round the Relay no longer holds that across a repair's rebind (routing.Relay.repair). The pool's mutex is held across a bind RPC too, as it always was (hostlink.Pool.Bind).
Index ¶
- Variables
- type Config
- type Links
- type Plane
- func (p *Plane) Attach(relay Relay, rebinder routing.Rebinder)
- func (p *Plane) Bind(ctx context.Context, endpoint sessionwire.InternalEndpoint, ...) error
- func (p *Plane) Close(ctx context.Context) error
- func (p *Plane) CloseLink(_ context.Context, link routing.LinkID, _ string) error
- func (p *Plane) DeliverCommand(ctx context.Context, tenant sessionwire.TenantID, ...) error
- func (p *Plane) DroppedEphemerals() uint64
- func (p *Plane) Publish(_ context.Context, _ routing.LinkID, tenant sessionwire.TenantID, ...) error
- func (p *Plane) PublishJournalTip(_ context.Context, hint sessionwire.JournalTip) error
- func (p *Plane) Resume(ctx context.Context, tenant sessionwire.TenantID, ...) error
- func (p *Plane) RouteHeld(tenant sessionwire.TenantID, session sessionwire.SessionID) bool
- func (p *Plane) Served(tenant sessionwire.TenantID, session sessionwire.SessionID)
- func (p *Plane) Stop(_ context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
- func (p *Plane) Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
- func (p *Plane) Unwatched(tenant sessionwire.TenantID, session sessionwire.SessionID)
- func (p *Plane) Watching(tenant sessionwire.TenantID, session sessionwire.SessionID)
- type Relay
- type Viewers
Constants ¶
This section is empty.
Variables ¶
var ErrInvalidConfig = errors.New("livetail: invalid configuration")
ErrInvalidConfig is the class of every New rejection.
var ErrNoViewers = errors.New("livetail: no ClientLink is running")
ErrNoViewers reports a publish with no ClientLink to publish through: the node is not running yet, or has stopped.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Links is the HostLink pool. Required.
Links Links
// Viewers supplies the running ClientLink, or nil when there is none.
// Required.
Viewers func() Viewers
// MailboxLimit bounds one session's undelivered publications. Ephemeral
// arrivals drop at capacity; non-ephemeral publications (enduring, or any
// record the Relay will refuse) evict queued ephemerals. Control events
// are not bounded. An overflow with no ephemeral victim loses the tail
// and triggers repair. Required, positive.
MailboxLimit int
// EventTimeout bounds the relay work one event may cost: a tip read, a
// rebind and the publishes. Required, positive.
EventTimeout time.Duration
// Logger receives what a drainer could not do. Nil takes slog.Default.
Logger *slog.Logger
}
Config composes a Plane.
type Links ¶
type Links interface {
Bind(ctx context.Context, target hostlink.Target, req sessionwire.HostLinkBindRequest) error
Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
DeliverCommand(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, delivery sessionwire.HostLinkCommandDelivery) error
Subscribe(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, sink hostlink.SessionSink) error
Unsubscribe(tenant sessionwire.TenantID, session sessionwire.SessionID)
RouteFor(tenant sessionwire.TenantID, session sessionwire.SessionID) (sessionwire.HostID, bool)
}
Links is the HostLink pool, narrowed to what this package calls.
type Plane ¶
type Plane struct {
// contains filtered or unexported fields
}
Plane is the live-tail composition. See the package documentation.
It implements routing.Binder, routing.Tail, routing.Publisher, routing.Hinter and routing.Watcher, and those are the seams the composition hands it to; it holds the Relay and the Rebinder it drives, set by Attach.
func (*Plane) Attach ¶
Attach supplies the Relay this plane drives and the Rebinder a Restored link asks for. They are late because each is built from this plane: the Relay takes it as its Tail and Publisher, and the demand plane takes it as its Hinter and Watcher. It must be called before the first viewer.
func (*Plane) Bind ¶
func (p *Plane) Bind(ctx context.Context, endpoint sessionwire.InternalEndpoint, req sessionwire.HostLinkBindRequest) error
Bind binds a session's route and then subscribes to its channel, in that order and over the same link.
The order is the Host's rule: host v0.2.1 accepts a subscribe only from a connection that already holds the bind for that channel. A subscribe that fails undoes the bind and fails the whole call, so a route this replica holds is always one carrying a live tail -- a route without one would read, to every viewer, exactly like an idle session.
The endpoint is the owner's advertised BASE; the pool derives the tenant's address. A base that cannot carry this tenant's address (Gap 1, hostlink.ErrNoTenantEndpoint) leaves the session unbound, as every failed bind does, and is logged with Core's code, because the routing plane's next poll will meet the same answer and nothing else would ever say why a watched session stays silent.
func (*Plane) Close ¶
Close stops accepting events, waits for every drainer to finish or ctx to end, and closes the Relay.
func (*Plane) CloseLink ¶
CloseLink makes every viewer of the session repair from the durable journal. It is reached when the Relay cannot tell them where to read from.
func (*Plane) DeliverCommand ¶
func (p *Plane) DeliverCommand(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, delivery sessionwire.HostLinkCommandDelivery) error
DeliverCommand is the pool's, unchanged.
func (*Plane) DroppedEphemerals ¶ added in v0.12.1
DroppedEphemerals counts mailbox admission drops and evictions. It does not inspect or log publication bodies.
func (*Plane) Publish ¶
func (p *Plane) Publish(_ context.Context, _ routing.LinkID, tenant sessionwire.TenantID, session sessionwire.SessionID, encoded []byte) error
Publish hands one record to every viewer of the session. It never answers routing.ErrWouldBlock: the transport's per-connection queue is the bound, and a viewer that falls behind it is disconnected and repairs.
func (*Plane) PublishJournalTip ¶
func (p *Plane) PublishJournalTip(_ context.Context, hint sessionwire.JournalTip) error
PublishJournalTip is routing.Hinter: the unbound-session hint, published to the same channel.
func (*Plane) Resume ¶
func (p *Plane) Resume(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, _ uint64) error
Resume makes sure a bound session has a tail. The Relay calls it after its rebind, and the rebind's own Bind has normally subscribed already, in which case this finds the subscription and does nothing. A session the rebind could not bind is left alone: its tail starts when a poll or a Restored link binds it, and that start sends the reset.
func (*Plane) RouteHeld ¶
func (p *Plane) RouteHeld(tenant sessionwire.TenantID, session sessionwire.SessionID) bool
RouteHeld is routing.RouteReporter: whether the pool still holds the session's route. Placement binds and unbinds through the same pool, so its transient unbind can drop a route a viewer's binding relies on; reporting it lets the next ownership poll re-bind instead of trusting the table forever.
func (*Plane) Served ¶
func (p *Plane) Served(tenant sessionwire.TenantID, session sessionwire.SessionID)
Served closes it.
func (*Plane) Stop ¶
func (p *Plane) Stop(_ context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
Stop ends the session's tail. Anything its sink reports from now on is dropped.
func (*Plane) Unbind ¶
func (p *Plane) Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
Unbind stops the session's tail and then gives the route back. Stopping it is not optional: a Host never unsubscribes a Factory -- not on unbind, not when it invalidates the session's routes -- so a tail left behind would keep publishing into a route this replica no longer holds.
A failure on a link whose connection is gone (hostlink.ConnectionGone) is classified routing.ErrRouteGone, with the transport's cause kept: the Host dropped the route with the connection, which the routing plane's shutdown paths count as released rather than as a failure.
func (*Plane) Unwatched ¶
func (p *Plane) Unwatched(tenant sessionwire.TenantID, session sessionwire.SessionID)
Unwatched forgets the session: its tail's events are dropped from now on, and the drainer drops its Relay state after anything already queued.
func (*Plane) Watching ¶
func (p *Plane) Watching(tenant sessionwire.TenantID, session sessionwire.SessionID)
Watching opens the window in which a tail going live is the first viewer's.
type Relay ¶
type Relay interface {
Open(tenant sessionwire.TenantID, session sessionwire.SessionID) error
Subscribe(tenant sessionwire.TenantID, session sessionwire.SessionID, link routing.LinkID) error
Receive(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, frame routing.Frame) error
Pump(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
HostLinkClosed(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
Resync(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
Anchor(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
Forget(tenant sessionwire.TenantID, session sessionwire.SessionID)
Close()
}
Relay is routing.Relay, narrowed to what this package calls.
type Viewers ¶
type Viewers interface {
PublishSession(tenant sessionwire.TenantID, session sessionwire.SessionID, encoded []byte) error
CloseSession(tenant sessionwire.TenantID, session sessionwire.SessionID)
}
Viewers is the ClientLink side: publish to a session's viewers, or make them all repair. clientlink.Handler implements it; the composition supplies it late, through a func, because the node is started after composition.
Source Files