Help us shape the future of Go! Take the 2026 Go Developer Survey

routing

package
v0.12.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 26, 2026 License: Apache-2.0 Imports: 14 Imported by: 0

Documentation

Overview

Package routing reads the durable routing and capacity records that Factory uses to find Hosts. Capacity advertisements are placement hints only; session ownership comes exclusively from the epoch-fenced registry.

Index

Constants

This section is empty.

Variables

View Source 
var ErrBindingsClosed = errors.New("routing: bindings are closed")

ErrBindingsClosed reports work asked of a closed table. A closed replica may not answer from routes it is giving up.

View Source 
var ErrDemandClosed = errors.New("routing: demand plane is closed")

ErrDemandClosed reports work asked of a closed demand plane.

View Source 
var ErrForeignRecord = fmt.Errorf("routing: a host record names another tenant or session: %w", delivery.ErrMalformed)

ErrForeignRecord reports a Host publication naming a tenant or session other than the one whose live tail it arrived on.

A record carries its own routing envelope, and the tail it arrives on names the session a viewer subscribed to. Only a faulty or compromised Host can make the two disagree, and forwarding such a record would put another session's -- or another TENANT's -- output in front of this session's viewers. It wraps delivery.ErrMalformed deliberately: the livetail drainer answers every refused record with the same repair (the tail is stopped, every viewer is reset from one durable tip, and the tail is re-bound), so a foreign record is a hole in the stream like any other, never a silent skip. Relay.ForeignRecords counts them.

View Source 
var ErrInvalidConfig = errors.New("routing: invalid directory configuration")

ErrInvalidConfig reports a Directory that cannot keep its advertised work bounds. Configuration is checked by NewDirectory before any store call.

View Source 
var ErrNoBinding = errors.New("routing: no local binding for session")

ErrNoBinding reports a delivery for a session this replica holds no demand for. A delivery does not open a route; see Bindings.Deliver.

View Source 
var ErrNoDemand = errors.New("routing: no local demand for session")

ErrNoDemand reports a release of demand nobody holds.

View Source 
var ErrNoHostBinding = errors.New("routing: no host binding for session")

ErrNoHostBinding reports a session this replica holds no HostBinding for.

View Source 
var ErrNoOwner = errors.New("routing: session has no live owner")

ErrNoOwner reports a session with no live registry owner. It is not a failure of this package and not something it may fix: a session needing an owner needs PLACEMENT, and a router that created one would be a second, unclaimed scaling authority beside internal/placement's reconciler.

View Source 
var ErrRelayClosed = errors.New("routing: relay is closed")

ErrRelayClosed reports work asked of a closed relay.

View Source 
var ErrUnexpectedControl = errors.New("routing: a host may not send a repair control")

ErrUnexpectedControl reports a repair control arriving FROM a Host.

A session.reset names a sequence the SENDER knows it forwarded in order, and on a DeliveryBinding that sender is this replica. A Host cannot know what this Factory delivered to a browser, so a control accepted from below would hand a client a coverage claim nobody is in a position to make.

View Source 
var ErrUnroutableOwner = errors.New("routing: registry owner cannot be bound")

ErrUnroutableOwner reports a registry owner whose ownership tuple Core will not carry. It fails here, naming the session, rather than inside the transport as a marshalling error naming nothing.

View Source 
var ErrWouldBlock = errors.New("routing: transport cannot take the record now")

ErrWouldBlock reports a transport that cannot take a record right now.

It is the ONE publish outcome that is not a failure: the record stays queued, and the queue -- not the transport -- is what bounds the backlog. Every other error means the ClientLink is gone, and the binding is dropped.

Functions

This section is empty.

Types

type Binder 

type Binder interface {
	Bind(ctx context.Context, endpoint sessionwire.InternalEndpoint, 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
}

Binder carries what this table decides. It is the HostLink pool's control surface stated in Core's vocabulary: nothing here names a transport type, because a binding is routing state that outlives any particular transport and because naming one would put the centrifuge dependency in this package's graph, where the boundary rules deliberately do not have it.

The pool's Bind takes its target as a struct; the endpoint is passed separately here so this seam names only Core. Composing the two is one adapter, and it is A9.1's along with the rest of the composition.

type Binding 

type Binding struct {
	Key                    BindingKey
	Endpoint               sessionwire.InternalEndpoint
	RuntimeCompatibilityID string
}

Binding is one live local route. It is comparable, so a caller can hold one and ask whether what it holds is still what the table has.

type BindingKey 

type BindingKey struct {
	TenantID       sessionwire.TenantID
	SessionID      sessionwire.SessionID
	HostID         sessionwire.HostID
	HostGeneration uint64
	LeaseEpoch     uint64
}

BindingKey identifies one Factory-local route, and identity is the WHOLE ownership tuple rather than the session. A4.3 step 1.

Each member earns its place by naming a way the route can become wrong without the session changing:

  • TenantID, because a SessionID is unique only within a tenant. A table keyed by session alone lets one tenant's route answer for another's.
  • HostID, because a session that moved is served by a different link.
  • HostGeneration, because a restarted Host has none of the state the old route assumed, and answers as a different Host.
  • LeaseEpoch, because the epoch is the fence: a bind naming a superseded epoch is exactly what the owning Host refuses.

So "invalidate the binding" and "the key changed" are one fact, and there is no second staleness rule that could disagree with the key.

type Bindings 

type Bindings struct {
	// contains filtered or unexported fields
}

Bindings is this replica's local routing table: which Host each session it is serving is bound to, and how many local subscribers want it.

It is LOCAL STATE ONLY (A4.3 step 3). Nothing here is written durably and nothing here is coordinated with another replica; a restart is a new value that knows nothing, and it rebuilds from exactly two sources — SessionStore, through the Resolver, and subscriber demand, through Acquire. That is why there is no reconstruction path to write: reconstruction is what the ordinary path already does.

A binding is a routing hint and never authority. The Host validates the tuple against its own durable lease, so the failure this table cannot have is delivering to a Host that will accept a route it should not — that failure is not available to it.

One stated limit, so it is not discovered later: invalidation is driven by an observation arriving at Observe. A lease that changed with nobody pushing an observation leaves a stale route in place until the owning Host refuses what it carries. That is the same fail-closed shape as everything else here — the Host refuses — but it is a refusal, not a repair. The repair is Relay's, in repair.go: it learns from BELOW that a route is unusable and asks Demand.Rebind for a fresh one rather than waiting for an ownership poll.

func NewBindings 

func NewBindings(resolver Resolver, binder Binder) (*Bindings, error)

NewBindings validates the composition before any call, so a missing seam is a composition error rather than a nil dereference on the first subscriber.

func (*Bindings) Acquire 

func (b *Bindings) Acquire(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) (Binding, error)

Acquire records one local subscriber's demand for a session and returns the route it should use, binding to the session's owner if this replica does not already hold one.

A binding already held is REUSED and costs no registry read (A4.3 step 2). That is what makes a per-session subscription cheap, and it is also why invalidation is a push: the route is not re-derived on use, so something has to say it is no longer current.

func (*Bindings) Binding 

func (b *Bindings) Binding(tenant sessionwire.TenantID, session sessionwire.SessionID) (Binding, bool)

Binding reports the route this replica currently holds for a session.

func (*Bindings) Close 

func (b *Bindings) Close(ctx context.Context) error

Close gives every route back to its Host and refuses later work.

Every route is attempted and every failure is joined, rather than returning at the first: one Host's failure must not strand the routes on the others, which are the ones a restarted replica would then conflict with.

There is no `if b.closed { return nil }` fast path, and its absence is deliberate: idempotence comes from EMPTYING the table, and a mutant that deleted such a guard survived the whole suite for exactly that reason. The HostLink pool removed the same redundant guard on the same evidence. A second Close unbinds nothing because there is nothing left to unbind.

func (*Bindings) Deliver 

func (b *Bindings) Deliver(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, delivery sessionwire.HostLinkCommandDelivery) error

Deliver wakes consumption of an already accepted command on this session's binding.

It requires demand and does not open a route (ErrNoBinding otherwise), which keeps the demand count the ONLY rule about a binding's lifetime. A caller delivering to a session nobody is watching must take demand FIRST -- and it must take it through Demand, not here.

THAT IS A CORRECTION, and the earlier wording is the hazard A7.3 settled. This paragraph used to say "brackets the delivery with Acquire and Release", which instructs a caller to become a SECOND holder of this table's demand. Release only unbinds on the last release, so while a second holder exists a release decrements without unbinding, route.bound stays true, and routeLocked hands the next caller a route nobody re-read from the registry. The invariant is one holder per session and it is Demand; see Demand.Rebind for the whole argument and TestTheRoutingTablesDemandIsHeldOnlyByTheDemandPlane for the guard. The pool's idle window is still what keeps a bracketed delivery from costing a dial per command.

A session whose binding was invalidated is rebound here first, so the command is REDELIVERED to the new owner (A4.3 step 2) rather than dropped with the durable inbox record left for nobody. Nothing here makes a command safe to repeat — that is the Host's lease and SessionStore's idempotency, and the delivery carries only the retry-stable public CommandID.

A delivery FAILURE is reported unchanged and changes nothing. This table cannot tell a lost connection from a Host's refusal, and dropping the route on either would turn one failure into a rebind storm without having delivered anything. Per-binding repair is Relay's, above the transport.

Deliver and the lock

The route is resolved -- and, if invalidated, rebound -- under the table's lock, so the bind coalescing mu's doc argues for still holds. The Host RPC then runs WITHOUT the lock. What can change after the unlock, and why each is safe:

  • The route is dropped or rebound (Observe, Release, a concurrent Deliver's rebind). The delivery names only (tenant, session, command); it carries no Host, generation or epoch of this table's, so it cannot re-assert a stale tuple. The Binder resolves the route it goes over AT CALL TIME under its own lock (the HostLink pool's DeliverCommand), so the hint reaches the pool's current route or is refused ErrUnknownBinding. That was already true with the lock held: the pool is shared with placement, which binds and unbinds through it directly.
  • The table closes. Close no longer waits for a delivery in flight; the delivery then meets the pool's own closed check or the unbound route, and its failure is reported. Nothing a stale delivery does writes to this table, so it cannot re-open a route on a closed one.
  • The link closes under it. That is the pool's and the link's concern; either reports an error.

None of this bypasses fencing, because fencing was never here: a delivery is a wake hint for an already committed record, and the Host applies what it reads from its own lease-fenced command stream.

func (*Bindings) Demand 

func (b *Bindings) Demand(tenant sessionwire.TenantID, session sessionwire.SessionID) int

Demand reports how many local subscribers hold a session.

func (*Bindings) Len 

func (b *Bindings) Len() int

Len reports how many sessions this replica is routing.

func (*Bindings) Observe 

func (b *Bindings) Observe(ctx context.Context, observed sessionwire.HostLinkRegistryObservation) bool

Observe applies a registry observation to the table and reports whether it invalidated a binding.

It DROPS the route rather than rebinding from the observation, and never creates one. An observation is a hint and the registry is the authority, so the next caller re-reads it; what the push is trusted for is only "what you hold is no longer current", which needs no trust at all. Subscriber demand survives, so the rebind happens on the next Acquire or Deliver.

An observation naming an OLDER lease epoch than the binding holds is stale and ignored. An observation at the same epoch naming a different Host or generation is not stale, it is a contradiction, and it invalidates: keeping a route the registry contradicts is how a command reaches a Host that is not the owner.

The unbind is best effort and its error is deliberately not reported. There is no answer a caller could act on — the local route is gone either way, for the reason Release states — and Observe's own answer is about this table.

func (*Bindings) Release 

func (b *Bindings) Release(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Release gives back one subscriber's demand. The LAST release unbinds: the route exists for the demand, so a route outliving it would hold a HostLink open for nobody.

A failed unbind still drops the local route, and the error is reported rather than acted on. The reason is the HostLink pool's, one level up: a bind is Factory-local state the Host validates against its own lease, so a route kept after a failure is kept forever — nothing above retries it — and the next bind would be refused as a conflict against a route nobody wants.

type Clock 

type Clock interface {
	// AfterFunc runs f after d and returns a stop function reporting whether
	// it prevented the call.
	AfterFunc(d time.Duration, f func()) (stop func() bool)
}

Clock is the time seam, and this package reads exactly one thing from it: when to run the next ownership poll.

It declares AfterFunc and not Now, for clientlink.Clock's reason: nothing here reads a wall clock, and a Now no caller called would be a method every composition supplies identically and no test could distinguish. factory.Clock is assignable to it without an adapter.

type Demand 

type Demand struct {
	// contains filtered or unexported fields
}

Demand is this replica's LOCAL subscriber demand, and the binding and polling that demand drives.

It sits directly on *Bindings rather than on an interface, and takes the resolver from it rather than being handed one. That is the single-authority property, by construction: the registry read a poll makes to decide whether a route is still current is THE SAME reader the table binds through, so there is no second ownership authority here that could disagree with the routing table about who owns a session.

What Acquire does NOT do, and why it is the point

A first subscriber that finds no owner is left UNBOUND and is not refused. Viewing is not a reason to make a cold session resident: a restore is a durable COMMAND, admitted through internal/admission by a caller who asked for one, and a viewer asked only to watch. Nothing on this path can admit anything -- the seams above are a routing table, a journal READ and a local publish -- and the module-wide guard in the root package holds that for the composition, where the seams alone cannot.

Locking

One mutex, held across the store and transport I/O a poll and an Acquire perform, which is Bindings' own trade at its own mu and is stated here for the same reason: it makes many subscribers arriving at once for one session cost one registry read, at the price of a slow read delaying work on an unrelated session. The lock order is Demand.mu then Bindings.mu and never the reverse -- nothing in Bindings names this type.

func NewDemand 

func NewDemand(bindings *Bindings, tips TipReader, hints Hinter, clock Clock, limits DemandLimits) (*Demand, error)

NewDemand validates the composition before any call, so a missing seam is a composition error rather than a nil dereference on the first subscriber.

func (*Demand) Acquire 

func (d *Demand) Acquire(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Acquire records one local subscriber for a session and, if it is the first, binds to the session's owner and starts polling.

It answers a LOCAL fact -- this replica now has a subscriber -- and that fact cannot fail. A session with no owner, an owner Core will not carry, a refused bind and a registry outage are all reported by returning nil with the session left unbound, because each of them is a routing outcome the poll retries and none of them is a reason to refuse a viewer. The one error is a closed plane, where the answer would be a lie.

A first subscriber that cannot be bound is served IMMEDIATELY with the tip hint, on the same path a poll uses, rather than waiting an interval for the first poll to notice. That is why serveLocked exists as one body: a viewer of a cold session and a viewer whose session went cold get the same treatment because they run the same statements.

The debounce that keeps a reconnecting browser from costing a rebind is the ClientLink edge's, and is deliberately not restated here: this plane is told about the first and the last subscriber, and A6.3 owns what "last" means.

func (*Demand) Close 

func (d *Demand) Close(ctx context.Context) error

Close stops every poll, gives every route back and refuses later work.

Every session is attempted and every failure joined rather than returning at the first, for Bindings.Close's reason: one Host's failure must not strand the routes on the others.

func (*Demand) Len 

func (d *Demand) Len() int

Len reports how many sessions this replica is watching.

func (*Demand) Rebind 

func (d *Demand) Rebind(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Rebind gives a watched session's route back and takes a fresh one IMMEDIATELY, without changing how many subscribers this replica holds.

It exists for internal/routing's repair plane, which learns from below that a route is unusable -- a physical HostLink closed, a route queue overflowed -- and may not wait an ownership poll to find out. It is deliberately the SAME two statements a poll runs, so a repair and a poll cannot repair differently.

THE INVARIANT THIS METHOD IS THE POINT OF: one demand holder per session

Bindings.Release only unbinds on the LAST release, because it COUNTS. This method is a release followed by an acquire, so it is fresh only while this type is the sole holder of the routing table's demand for a session: with a second holder the release would merely decrement, route.bound would stay true, and Bindings.routeLocked would hand the acquire back THE SAME STALE ROUTE without reading the registry at all -- which is precisely the state a repair exists to leave.

So the answer to "should Release become owner-aware" is no, and the reason is that owner-awareness would not fix it. Two holders means the route legitimately outlives one holder's release, so the stale-adoption window is a property of there being two, not of the counting. The repair plane therefore takes NO demand and asks here instead. TestASecondHolderOfTheRoutingTablesDemandMakesARebindStale is the reader for the hazard, and TestTheRoutingTablesDemandIsHeldOnlyByTheDemandPlane is the structural guard that keeps a second holder from appearing.

func (*Demand) Release 

func (d *Demand) Release(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Release gives back one subscriber. The LAST one stops the poll and gives the route back, because the route exists for the demand.

The unbind's failure IS reported here, unlike everywhere else in this file, and the difference is that this one has a caller. Bindings.Release drops the local route either way; what the error says is that a Host may still be holding a route this replica has forgotten, which is an operational fact somebody can act on.

func (*Demand) SetWatcher added in v0.4.0 

func (d *Demand) SetWatcher(w Watcher)

SetWatcher composes the watcher. It is a composition-time call, made before the first Acquire; a nil watcher restores the default, which does nothing.

func (*Demand) Watching 

func (d *Demand) Watching(tenant sessionwire.TenantID, session sessionwire.SessionID) int

Watching reports how many local subscribers this replica holds for a session.

type DemandLimits 

type DemandLimits struct {
	// OwnershipPollInterval is how long after one poll finishes the next is
	// armed. It is a gap between polls rather than a period, so a poll slower
	// than the interval delays the next one instead of overlapping with it.
	OwnershipPollInterval time.Duration

	// PollTimeout bounds ONE whole poll -- the registry read, the bind it may
	// provoke, the tip read and the publish together -- rather than each call
	// inside it. A poll runs on the clock's goroutine with no caller and no
	// request context to inherit, so this is the entire bound on it.
	PollTimeout time.Duration
}

DemandLimits bounds the background work one watched session may cost.

Both members are REQUIRED and positive. A zero poll interval is a busy loop against the registry and a zero timeout is an unbounded store call on a goroutine with no caller to abandon it, and neither is a sane default to infer for a composition that forgot to say.

func DefaultDemandLimits 

func DefaultDemandLimits() DemandLimits

DefaultDemandLimits returns bounded defaults for demand-driven binding.

Neither number is measured, and saying so is the point. The interval is the bounded delay specification section 10.5 promises a viewer on Factory A of work started through Factory B, and five seconds is a judgement about what a person watching a page will tolerate, not a figure derived from anything. The timeout is larger than the interval on purpose: a poll that is merely slow must not be abandoned in favour of a fresh one that will be just as slow.

func (DemandLimits) Validate 

func (l DemandLimits) Validate() error

Validate reports why these limits may not bound a demand plane.

type Directory 

type Directory struct {
	// contains filtered or unexported fields
}

Directory discovers placement candidates and the current owner of a session. It stores no routing state of its own.

func NewDirectory 

func NewDirectory(store Store, limits Limits) (*Directory, error)

NewDirectory validates all work bounds before constructing a Directory.

func (*Directory) Candidates 

func (d *Directory) Candidates(ctx context.Context, original sessionstore.ListCompatibleHostsRequest) (sessionstore.HostTargetPage, error)

Candidates returns compatible accepting Hosts in SessionStore's capacity order. The store owns target scoping, liveness, accepting-state filtering and generation fencing; this adapter does not restate those rules.

A lapsed row still occupies a ranked position. When a page reports one, Directory spends a bounded slice of one total due-page budget, then retries the caller's original position. It never follows the candidate continuation internally, so the continuation returned to the caller is always the honest continuation of the page it received.

func (*Directory) Owner 

func (d *Directory) Owner(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) (sessionwire.HostLinkRegistryObservation, bool, error)

Owner returns the live epoch-fenced registry observation for a session. Absence, expiry and a released tombstone all mean that no owner is routable; their different store codes remain an operational concern, not three kinds of ownership.

ABSENCE IS NOT ALWAYS A REGISTRY CODE. Outside the legacy single-tenant layout the store verifies a session's collision witnesses BEFORE it reads a record, so a session that was never created fails with *KeyspaceError binding_not_found and never reaches the registry at all. sessionstore's own noSuchSession names exactly that set, and internal/httpapi found the same thing the hard way: reading only the registry codes made every absent session a store FAILURE, which a caller answers by treating a routable session as unroutable -- or, here, by placing a session that has no record to place. Keyspace codes other than binding_not_found are real deployment faults and are propagated.

type Frame 

type Frame struct {
	Encoded            []byte
	CommittedAppendSeq uint64
	CoalesceKey        string
}

Frame is one record arriving on a session's live tail.

CommittedAppendSeq is the Host's own committed append sequence at the moment it sent this frame, and it is an INPUT rather than something derived from the record: a record cannot vouch for itself, and the whole content of the upper watermark bound is that somebody other than the record says how far the journal has been committed.

A DECLARED GAP. Core v0.9.1's session-channel records carry no member a Host could state its committed append sequence in. internal/realtime/livetail, which feeds this relay from the HostLink subscription since v0.4.0, passes the publication's own covered_through -- a vacuous fence, because Core requires covered_through == journal_seq -- until an additive Core member can tighten it. The honest reading of this field is "what the producer declares", and the relay fences against it rather than trusting the frame.

CoalesceKey is FACTORY-LOCAL and read only for an ephemeral record. Core gives an ephemeral publication no identity at all, so a path that coalesced without a declared key would be guessing that two opaque bodies are interchangeable; an empty key means this delta supersedes nothing.

type Hinter 

type Hinter interface {
	PublishJournalTip(ctx context.Context, hint sessionwire.JournalTip) error
}

Hinter publishes one repair hint to this replica's LOCAL subscribers.

Local is the whole of it. Specification section 10.5 gives version one no cross-node publication path, so a hint reaches the clients of this replica and no other; a second replica watching the same session polls and publishes its own. That is why the hint has to be repeatable rather than sequenced -- see Demand.hintLocked.

type Limits 

type Limits struct {
	CandidatePageLimit int
	ReconcilePageLimit int
	ReconcileMaxPages  int
	MaxCleanupAttempts int
}

Limits bounds both the foreground capacity query and the due cleanup it can provoke. ReconcileMaxPages is a total budget shared by every cleanup attempt, not a per-attempt allowance.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits returns bounded defaults for target discovery.

type LinkID 

type LinkID string

LinkID identifies one ClientLink connection on this replica. It is opaque here: a relay never parses it and never derives anything from it.

type Publisher 

type Publisher interface {
	// Publish hands one encoded session-channel record to one link. It returns
	// ErrWouldBlock when the transport cannot take it now.
	Publish(ctx context.Context, link LinkID, tenant sessionwire.TenantID, session sessionwire.SessionID, encoded []byte) error
	// CloseLink closes one ClientLink. Peer links continue.
	CloseLink(ctx context.Context, link LinkID, reason string) error
}

Publisher carries one record to ONE ClientLink, and closes one ClientLink.

Both methods name a link rather than a channel, because every repair in this file is per-link: closing a channel would take every session that link is watching, which is exactly the blast radius runbook A7.3 steps 3 and 4 forbid.

type Rebinder 

type Rebinder interface {
	Rebind(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
}

Rebinder re-establishes a session's route WITHOUT changing local demand.

*Demand implements it, and that is the whole reason this seam is spelled this way rather than as the routing table itself. See Demand.Rebind: the routing table's demand has exactly one holder per session, so a repair plane that took a unit of its own would make every rebind hand back the stale route it was repairing.

type Relay 

type Relay struct {
	// contains filtered or unexported fields
}

Relay is this replica's bounded delivery path: one route queue per HostBinding, one outbound queue per DeliveryBinding, and the repair each overflow owes.

The invariant, in one sentence

An enduring record is never discarded silently. Every bound in this file either evicts a record that carries no durable content, or reports an overflow that is repaired by telling the affected consumer -- and only the affected consumer -- where to read from.

Locking

One mutex for every session, held across queue work and the publishes a pump performs -- which makes a fan-out to thirty subscribers one pass rather than thirty interleaved ones -- and deliberately NOT across a HostBinding repair's tail stop, tip read, rebind or resume, nor Resync's tip read (see repair). Those are I/O against a store and a Host, and one session's slow Host held under this mutex stalled every other session's delivery (v0.4.0 quality gate F3). What is still held under it is a DeliveryBinding overflow's tip read (fanOutLocked), which a channel-wide publisher that never answers ErrWouldBlock does not reach in composition. Nothing in Demand or Bindings names this type, and no repair holds this mutex while calling them.

PRECONDITION: calls for ONE session are serialised by the caller

Because the mutex is released across a repair's I/O, this type no longer orders two calls for the same session against each other, and it relies on its caller to. Receive, Pump, HostLinkClosed, Resync and Forget for one (tenant, session) must never run concurrently: a second repair entered while the first is inside its tip read or rebind clears the first's stopped state early, admits a live frame between the two, and publishes a session.reset whose tip goes BACKWARDS (the v0.4.0 regate constructed E41, R41/77, E78, R41/45). Until d23cb6e the mutex serialised it; since then the precondition is load-bearing and is met by composition, not here: internal/realtime/livetail calls every one of these methods only from the session's single drainer goroutine (Plane.handle) and from Plane.Close after the drainers are stopped. Different sessions MAY run concurrently; that is the point of releasing the lock. A new caller that is not the plane's drainer must serialise per session itself.

func NewRelay 

func NewRelay(tips TipReader, rebinder Rebinder, tail Tail, publisher Publisher, limits RepairLimits) (*Relay, error)

NewRelay validates the composition before any call.

func (*Relay) Anchor added in v0.9.0 

func (r *Relay) Anchor(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Anchor records the durable tip a session's viewers are known to reach, as the point the live tail must continue: the caller calls it when a tail has gone live INSIDE the first viewer's subscribe (the start Gap 3 owes no reset), whose durable read comes after the tail was live and so reaches at least this tip. From then on a record past the tail's last position is probed (Receive) rather than assumed to continue it (I3.1 D2).

The tip is read with the lock RELEASED, as Resync reads it. A tip that cannot be read anchors nothing and is returned; the session keeps the original rule, which is the state before this method existed.

func (*Relay) Bindings 

func (r *Relay) Bindings(tenant sessionwire.TenantID, session sessionwire.SessionID) int

Bindings reports how many DeliveryBindings a session holds.

func (*Relay) Close 

func (r *Relay) Close()

Close refuses later work and drops every queue.

func (*Relay) ForeignRecords added in v0.8.1 

func (r *Relay) ForeignRecords() uint64

ForeignRecords reports how many Host publications this relay has refused with ErrForeignRecord since it was built. It never decreases. Any nonzero value means a Host sent a record for a session other than the tail it was sent on, which a correct Host never does.

func (*Relay) Forget added in v0.4.0 

func (r *Relay) Forget(tenant sessionwire.TenantID, session sessionwire.SessionID)

Forget drops a session's HostBinding and every queue it holds.

It is Open's inverse, for a session this replica has stopped watching; Close is the same thing for every session at once. Nothing queued is lost that anybody could read: a binding nobody holds has no consumer.

func (*Relay) HostLinkClosed 

func (r *Relay) HostLinkClosed(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

HostLinkClosed repairs a session whose physical HostLink went away.

It is the same repair a route-queue overflow takes, and that is deliberate: both mean the live tail stopped being a complete record of the session, and a second repair path would be a second place for the reset to be built differently.

func (*Relay) LastContiguous 

func (r *Relay) LastContiguous(tenant sessionwire.TenantID, session sessionwire.SessionID, link LinkID) (uint64, bool)

LastContiguous reports what one DeliveryBinding has vouched for.

func (*Relay) Open 

func (r *Relay) Open(tenant sessionwire.TenantID, session sessionwire.SessionID) error

Open creates this session's HostBinding and its bounded route queue.

func (*Relay) Pump 

func (r *Relay) Pump(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Pump moves one session's queued route records out to its DeliveryBindings and on to the transport.

It is a method rather than a goroutine for ReapIdle's reason: A9.1 owns the lifetime that would give a loop a start and a stop. What it buys a test is that the backlog is deterministic -- a queue fills because nothing pumped it, not because a scheduler was slow.

func (*Relay) Queued 

func (r *Relay) Queued(tenant sessionwire.TenantID, session sessionwire.SessionID, link LinkID) int

Queued reports how many records one DeliveryBinding is holding.

func (*Relay) Receive 

func (r *Relay) Receive(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, frame Frame) error

Receive takes one frame off a session's live tail onto its route queue.

It DISPATCHES on Core's own discriminator rather than on anything the caller asserts, so an unsequenced delta cannot be queued as durable data by a caller that mislabelled it. Three answers:

  • an enduring publication is parsed -- envelope only, bytes preserved -- and fenced against the frame's declared committed append sequence;
  • an ephemeral publication is queued as droppable, with the caller's declared coalesce key and no sequence check, because it has no sequence;
  • a repair control is REFUSED. See ErrUnexpectedControl.

A HostBinding overflow is repaired HERE and reported as nil: the overflow is handled, and the caller's next frame is the one that matters.

func (*Relay) Resync added in v0.4.0 

func (r *Relay) Resync(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error

Resync tells every DeliveryBinding of a session where to read from, because the live tail feeding it has just (re)started and whatever the Host published while it was not running is gone.

It is the half of a repair that does not touch the tail: no Stop, no rebind, no Resume. Gap 3 needs it because a tail can START over a hole -- a viewer watched a session this replica could not yet bind, a poll then bound it, and every record committed in between reached nobody -- and a Host keeps no history, so the only honest answer is a session.reset naming the one tip read here. It is built exactly as HostLinkClosed's is (resetBindingLocked, one tip for every binding), so the two cannot describe a gap differently.

The caller must call it AFTER the new tail is live: a reset sent before would have its consumer re-read the journal while records were still going nowhere.

func (*Relay) Subscribe 

func (r *Relay) Subscribe(tenant sessionwire.TenantID, session sessionwire.SessionID, link LinkID) error

Subscribe adds one ClientLink's DeliveryBinding to a session.

func (*Relay) Unsubscribe 

func (r *Relay) Unsubscribe(tenant sessionwire.TenantID, session sessionwire.SessionID, link LinkID)

Unsubscribe removes one DeliveryBinding. Its queued records go with it: a binding nobody holds has no consumer, and there is nothing durable to lose because everything still queued was never applied.

type RepairLimits 

type RepairLimits struct {
	// HostBindingQueue bounds one session's inbound route queue, in records.
	HostBindingQueue int
	// DeliveryBindingQueue bounds one client's outbound queue, in records.
	DeliveryBindingQueue int
}

RepairLimits bounds the two queues a session's delivery path holds.

The two are SEPARATE NUMBERS and must stay separate. A HostBinding's queue holds one session's inbound tail; a DeliveryBinding's queue holds one client's outbound copy of it, and a session with thirty subscribers holds thirty of the second and one of the first. A single constant would be either thirty times too much memory or a HostBinding that overflows before any client does -- and, as internal/httpapi's two page ceilings measured, a parameter every call site passes identically is untested by construction.

func DefaultRepairLimits 

func DefaultRepairLimits() RepairLimits

DefaultRepairLimits returns bounded defaults.

Neither number is measured and saying so is the point. 512 is wui's own DEFAULT_MAX_QUEUED_FRAMES, described there as roughly eight seconds of a fast token stream, and the inbound queue is twice it because one inbound record fans out to every subscriber and the inbound side must not be the first to give up. The RELATION is what is asserted; the numbers are pinned only as literals.

func (RepairLimits) Validate 

func (l RepairLimits) Validate() error

Validate reports why these limits may not bound a relay.

type Resolver 

type Resolver interface {
	Owner(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) (sessionwire.HostLinkRegistryObservation, bool, error)
}

Resolver reads the authoritative owner of a session. *Directory implements it, and that is the only implementation this module has: ownership comes from the epoch-fenced registry and from nowhere else.

type RouteReporter added in v0.4.0 

type RouteReporter interface {
	RouteHeld(tenant sessionwire.TenantID, session sessionwire.SessionID) bool
}

RouteReporter is an OPTIONAL Binder capability: whether the transport still holds a session's route. A table's binding is local state, and a transport shared with another writer -- the HostLink pool, which placement binds and unbinds through directly -- can drop the route under it. A Binder that reports it lets the ownership poll repair that; one that does not is trusted, as before.

type Store 

type Store interface {
	GetHostRegistration(context.Context, sessionstore.GetHostRegistrationRequest) (sessionstore.HostRegistrationEntry, error)
	ListCompatibleHosts(context.Context, sessionstore.ListCompatibleHostsRequest) (sessionstore.HostTargetPage, error)
	ReconcileHostTargets(context.Context, sessionstore.ReconcileHostTargetsRequest) (sessionstore.HostTargetReconcileResult, error)
}

Store is the SessionStore surface used by Directory. Keeping the registry read separate from capacity listing makes it impossible to turn an advertisement into a session ownership claim inside this package.

type Tail 

type Tail interface {
	// Stop ends the live tail for one session's binding.
	Stop(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) error
	// Resume restarts it after a sequence, which is the tip the repair captured.
	Resume(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, afterSeq uint64) error
}

Tail is the Host live tail's control surface for one session.

It is separate from Binder because the two have different lifetimes: a route is bound while a subscriber wants the session, and a tail is running only while this replica is keeping up with it. A repair stops the tail, repairs, and resumes from a sequence -- three things a bind cannot express.

type TipReader 

type TipReader interface {
	ReadPublicJournal(ctx context.Context, req sessionstore.ReadPublicJournalRequest) (sessionwire.JournalPage, error)
}

TipReader is the durable journal read this package uses to answer "how far has this session got", and NOTHING else.

It is the SessionStore domain method, declared here as the narrow interface THIS package calls, for internal/httpapi's SessionReader's reason. Only JournalPage.CapturedTip is read: the events a page carries are not looked at, are not forwarded, and are not authorized here -- a hint says how far the journal goes, and the client reads the range it is missing through the durable query plane, which authorizes it in its own right.

type Watcher added in v0.4.0 

type Watcher interface {
	// Watching is called for a session's FIRST local subscriber, before the
	// bind that subscriber provokes.
	Watching(tenant sessionwire.TenantID, session sessionwire.SessionID)
	// Served is called once that first serve has returned, bound or not.
	Served(tenant sessionwire.TenantID, session sessionwire.SessionID)
	// Unwatched is called when the session's last subscriber is gone, or the
	// plane closed.
	Unwatched(tenant sessionwire.TenantID, session sessionwire.SessionID)
}

Watcher is told when this replica starts and stops watching a session.

It exists for Gap 3's live tail, which needs to know one thing only this type can say: whether a tail went live INSIDE the first subscriber's own Acquire. A tail started there is gapless for every viewer -- the first viewer's subscribe has not been acknowledged yet, so its durable read comes after -- while a tail started by any later poll or rebind may have missed records every current viewer was relying on it for, and is owed a reset.

EVERY METHOD IS CALLED WITH THIS TYPE'S LOCK HELD, so an implementation must not block and must not call back into this package. The composition's is a few map writes under a leaf mutex.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.