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

hostlink

package
v0.16.1 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package hostlink holds Factory's client side of the Factory-Host connection.

The interfaces here are the ones THIS package calls. Unlike the authenticator, authorizer and store seams, the dialer is NOT part of Factory's public option surface: the HostLink protocol is internal to this module, and the boundary rules already confine its engine to internal/realtime. What composition configures from outside is the pool's limits, not its transport.

What this package is and is not

Pool holds at most one physical connection per Host and multiplexes every session binding to that Host over it. It owns the CONTROL plane -- bind, unbind, attach, command delivery, and the capacity and registry observations a Host pushes -- and, since v0.4.0, a session's LIVE TAIL: Pool.Subscribe subscribes sessionwire.HostLinkChannel on the link that holds the session's bind (subscribe.go). The per-binding queues and the backpressure repair live ABOVE this package, in internal/realtime/delivery and internal/routing's Relay, composed by internal/realtime/livetail. It does not decide WHICH Host a session belongs to; that is A7.2's demand-driven binding, which calls Bind and Unbind.

Index

Constants

View Source 
const (
	// PushTypeCapacity carries a HostLinkCapacityReport.
	PushTypeCapacity = "host.capacity"
	// PushTypeRegistry carries a HostLinkRegistryObservation.
	PushTypeRegistry = "host.registry"
)

The HostLink control vocabulary is CORE'S, pinned here at core v0.9.1.

sessionwire/v1's hostlink_framing.go names the reserved RPC methods (HostLinkMethodBind, HostLinkMethodUnbind, HostLinkMethodAttach and the two drain methods), the channel prefix, and HostLinkChannel(tenant, session). This package declares no method constant of its own any more, and the reason is what the previous constants recorded as a gap: Factory sent "hostlink.command" while Host reserved only bind/unbind/drain and resolved EVERY other method as a channel name, so every command delivery Factory made would have been refused. There is no command method. COMMAND DELIVERY'S RPC METHOD IS THE SESSION'S CHANNEL, HostLinkChannel(tenant, session), and the body is the HostLinkCommandDelivery -- which is why Link.DeliverCommand takes the tenant and session beside the record.

The strings are still pinned as absolute literals by TestTheWireVocabularyIsPinnedToItsLiterals, now against Core's constants and Core's channel helper rather than against locals, and for the same reason: a fixture built from the constant under test pins nothing.

The asynchronous messages a Host pushes.

Core's framing covers RPC methods and the session channel; it names no push discriminator, so the {type, data} envelope below is still Factory's half of the protocol and the stand-in node in the tests implements exactly it. They are async messages rather than channel publications on purpose. A channel would put control-plane observations behind the subscription machinery; there is exactly one connection per Host and every observation on it is for this replica, so a discriminated message on that connection is the whole requirement.

That reasoning is stronger for a REGISTRY observation than for a CAPACITY report, and the difference is recorded rather than answered. A registry observation is per-session and per-route, so it genuinely concerns only the replica holding the route. A capacity report is a Host-level advertisement every Factory replica wants: published to a channel, the broker would fan one message out to every subscribed replica instead of the Host calling Client.Send once per connected replica, and two replicas on one Host is already a measured case rather than a hypothetical one. A Host writer has a real argument here. It is not taken now because Core names no channel for it -- HostLinkChannel is per session -- and because the Observer seam absorbs the change: moving capacity to a publication changes how a link is wired to the transport and nothing above it.

View Source 
const ClientName = "factory-hostlink"

ClientName is the connection label this replica presents to a Host.

It is a DIAGNOSTIC and carries no authority: the Host authenticates the service token and authorizes each bind against its own durable lease. It is deliberately one constant rather than per-replica, which is what the transport's own documentation asks for.

Variables

View Source 
var ErrAttachMismatch = errors.New("hostlink: attach observation does not match the request")

ErrAttachMismatch reports an accepted attach whose observation does not name the session, the Host or the Host incarnation the request did. The pool refuses to hand it to a caller, because the caller's next act is to bind with its epoch, and a bind built from an observation of something else is a route to nowhere that the Host will refuse for a reason reading as a lease problem.

View Source 
var ErrBindingConflict = errors.New("hostlink: session is bound to another host")

ErrBindingConflict reports a session already bound to a different Host.

View Source 
var ErrCommandUndelivered = errors.New("hostlink: command was not delivered")

ErrCommandUndelivered reports that a committed command was NOT handed to a Host.

It is the distinction runbook A7.1 step 3 turns on. The inbox record is already committed when DeliverCommand runs, so the caller's next decision is whether the command is finished. A transport failure means the Host has never seen it and the record stays PENDING for a later attempt; a Host refusal -- reported as *HostRefusal, and deliberately NOT wrapped in this -- means the Host answered and the caller may act on that answer. Collapsing the two would either terminate a command nobody received or retry one that was refused.

View Source 
var ErrDialFailed = errors.New("hostlink: dial failed")

ErrDialFailed is the class of every dial rejection that is not the dialer's own composition error.

View Source 
var ErrHostFailed = errors.New("hostlink: the host answered with a failure carrying no Core code")

ErrHostFailed is the class of every HostFailure.

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

ErrInvalidConfig is the class of every NewPool rejection.

View Source 
var ErrLinkClosed = errors.New("hostlink: link is closed")

ErrLinkClosed reports a call on a link this replica has closed: by the pool's shutdown, the idle reaper, an eviction, or a handshake it refused. It is TERMINAL -- a closed link never answers again -- and it is returned before anything is sent.

View Source 
var ErrLinkLimit = errors.New("hostlink: link limit reached")

ErrLinkLimit reports a bind that would open a link past MaxLinks.

View Source 
var ErrLinkReconnecting = errors.New("hostlink: link is reconnecting")

ErrLinkReconnecting reports a reserved HostLink operation refused locally because the link is between a dropped connection and the next Host reply, so there is no capability set to admit it against. It is a TRANSIENT, deliberately distinct from ErrUnsupportedMethod: the Host may well advertise the method, and a caller that read "did not advertise" during a 250ms reconnect would remake a placement decision over a perfectly good Host. Retry after the reconnect settles; session-channel delivery is not gated and is queued by the transport instead.

View Source 
var ErrMalformedAttachReply = errors.New("hostlink: attach reply is not a Core record")

ErrMalformedAttachReply reports an attach reply that is neither of the two records Core allows: an observation or a HostLinkError. An EMPTY reply is one of these -- an accepted attach must say at which lease epoch the session is now resident, and a caller told nothing has no epoch it may bind with.

View Source 
var ErrNoTenantEndpoint = errors.New("hostlink: the host's advertised base cannot carry this tenant's address")

ErrNoTenantEndpoint reports a Host whose advertised base cannot carry one tenant's HostLink address: Core's sessionwire.HostLinkEndpoint refused to derive it. It is a PER-TENANT fact about one Host -- a tenant too long for the address that base leaves room for (too_long), a tenant Core will not route (unroutable_tenant), or a base that is not a bare base at all (base_names_tenant, base_not_bare), which refuses every tenant -- and it is decided BEFORE anything is dialled, so it costs no connection.

A caller treats it as "this Host cannot serve this tenant": placement skips the candidate for that tenant and asks the next one, and a routing bind leaves the session unbound for the next poll. It is never a reason to abandon the Host for another tenant.

View Source 
var ErrPoolClosed = errors.New("hostlink: pool is closed")

ErrPoolClosed reports work asked of a pool that has been shut down.

View Source 
var ErrSubscribeRefused = errors.New("hostlink: the host did not accept the session subscription")

ErrSubscribeRefused reports a session subscription the Host did not accept, or that ended before it was ever live.

host v0.2.1 answers a subscribe with ErrorPermissionDenied (103) unless the link already holds the bind for that exact channel ON THE SAME CONNECTION (Multiplexer.MaySubscribe), so a refusal here usually means the bind that preceded it went over a connection that has since been replaced.

View Source 
var ErrSubscribeUnsupported = errors.New("hostlink: link cannot carry a session's live tail")

ErrSubscribeUnsupported reports a link that cannot carry a session's live tail. Every link this package dials can; the error exists for a Link a test or a later transport supplies without the Subscriber capability, so a pool holding one fails a subscribe closed rather than pretending it is live.

View Source 
var ErrSubscriptionWithdrawn = errors.New("hostlink: the session subscription was withdrawn")

ErrSubscriptionWithdrawn reports a pending subscription its owner gave up before the Host answered.

View Source 
var ErrTargetMismatch = errors.New("hostlink: request names a different host")

ErrTargetMismatch reports a bind record naming a Host other than the one the link goes to.

View Source 
var ErrUnknownBinding = errors.New("hostlink: no binding for session")

ErrUnknownBinding reports work for a session this pool holds no route for.

View Source 
var ErrUnsupportedMethod = errors.New("hostlink: host did not advertise the requested method")

ErrUnsupportedMethod reports a reserved HostLink operation the connected Host did not advertise. It is a local refusal: no RPC is sent to a peer that did not promise to handle the method.

View Source 
var ErrUnsupportedProtocol = errors.New("hostlink: host does not speak this wire version")

ErrUnsupportedProtocol reports a Host that does not speak this build's sessionwire version.

It is TERMINAL for the connection. Reconnecting to the same Host would negotiate the same answer, and a link that stayed open speaking a version neither end agreed on would exchange control records whose meaning is not established.

Functions

func ConnectionGone added in v0.16.0 

func ConnectionGone(err error) bool

ConnectionGone reports an error from a link whose connection is gone and which sent nothing: the link is reconnecting (ErrLinkReconnecting), or it is terminal (terminalLinkError). Every one of those is returned BEFORE the request reaches the transport.

What it lets a caller conclude about an UNBIND is that the route it named no longer exists on the Host. A Host keeps routes per connection -- keyed by the connection's client id, minted per connection -- and drops all of them when the connection closes (host's Multiplexer.CloseLink, run from OnDisconnect in every released Host from v0.1.0); a reconnect is a new connection that inherits no route. A route this pool held was bound on the connection that has now gone, so the Host has dropped it or will on noticing the close.

It deliberately does NOT cover a cancelled or timed-out call, nor a failure returned by the transport itself: that request may have left, on a connection the Host still holds.

func GateResponseCapable added in v0.5.0 

func GateResponseCapable(reply sessionwire.VersionNegotiationResponse) bool

GateResponseCapable is THE ONE STATEMENT of "the Host at the other end of this link can apply a gate_response command", read from the Host's connect reply. Every gate on gate_response delivery in this module asks it through Pool.AcceptsGateResponses, and nothing else decides: admission's refusal before anything is written, placement's withheld wake, and the pooled placement filter that puts a session with a pending gate response only on a Host carrying the token.

The signal is Core's capability token HostLinkCapabilityGateResponse ("hostlink.command.gate_response", core v0.11.0), which a Host that applies gate_response through the disposition path (host >= v0.4.0) lists in its connect reply's hostlink_methods. It is a TOKEN, not a method: nothing is dispatched on it. The match is Supports' exact-name match, so a substring, a "hostlink.v1."-prefixed spelling or any other near miss is not the capability, and a Host that predates it (host v0.3.0 advertises only the five reserved methods) is refused, which keeps its behaviour unchanged.

It deliberately reads NOTHING else. In particular the catalog's LeaseEpoch is never read as a capability signal (owner ruling, 2026-09-19): a per-session, caller-asserted value cannot answer a per-process question.

The answer can be ONE REPLY STALE (quality gate F7, spec N3). It is read from the link's most recent verified reply, and a reply names no Host incarnation, so it is not fenced to the owner's HostGeneration: a Host restarted at another version, whose old connection this link has not yet noticed is dead, is answered from the previous reply until the reconnect lands. The same window exists for every reserved-method gate; the mixed-fleet rule (do not run Hosts that differ in this capability side by side for gating agents) is what bounds its consequence.

func PayloadReferenceCapable added in v0.14.0 

func PayloadReferenceCapable(reply sessionwire.VersionNegotiationResponse) bool

PayloadReferenceCapable reports the exact body dereference capability.

func PrincipalCapable added in v0.12.0 

func PrincipalCapable(reply sessionwire.VersionNegotiationResponse) bool

PrincipalCapable is the exact advertised support for reading principal and metadata on incoming commands, never inferred from lease or Host version.

Types

type CentrifugeDialer 

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

CentrifugeDialer opens real HostLink connections.

func NewCentrifugeDialer 

func NewCentrifugeDialer(cfg DialerConfig) (*CentrifugeDialer, error)

NewCentrifugeDialer validates a composition and returns it.

func (*CentrifugeDialer) Dial 

func (d *CentrifugeDialer) Dial(ctx context.Context, target Target, observer Observer) (Link, error)

Dial opens one authenticated, version-negotiated connection to a Host.

It returns only after the handshake has SETTLED -- connected and negotiated, or refused. A dialer that returned an unconnected client would hand the pool a link it would record as good and then discover was never usable, and the pool's ceiling would be spent on it.

type Config 

type Config struct {
	// Dialer opens one connection. Required.
	Dialer Dialer
	// Observer receives capacity and registry pushes. Optional; observations
	// are discarded when it is nil.
	Observer Observer
	// Limits bounds this replica's pool. A zero value takes DefaultLimits.
	Limits Limits
	// Now is the pool's clock, for the idle reaper. A nil value takes
	// time.Now. It is a seam so the reaper's boundary can be measured at an
	// exact instant rather than waited for.
	Now func() time.Time
	// GateResponses decides, from a Host's connect reply, whether that Host
	// can apply a gate_response command. A nil value takes
	// GateResponseCapable, which is what every production composition uses;
	// the field exists so the mechanism's ACCEPTING half can be driven before
	// any Host advertises the capability.
	GateResponses func(sessionwire.VersionNegotiationResponse) bool
}

Config composes the pool.

type Credential 

type Credential interface {
	// ServiceToken returns the credential to present, or an error if one
	// cannot be obtained. It is called on every dial and on every reconnect,
	// so a short-lived token is refreshed rather than replayed.
	ServiceToken(ctx context.Context) (string, error)
}

Credential supplies the SERVICE identity Factory presents to a Host.

It is deliberately not identity.Principal: a HostLink carries no end user. The token authenticates one Factory replica to one Host, and the Host's own authorization of a bind is made against the session's durable lease, not against a user this connection claims to act for.

type Dialer 

type Dialer interface {
	Dial(ctx context.Context, target Target, observer Observer) (Link, error)
}

Dialer opens one authenticated, multiplexed connection to a Host. The pool calls it; a test substitutes it.

The observer is handed over at DIAL time rather than registered afterwards, because a Host may push a capacity report the instant the connection is established. A link that had to be told where to send observations after it existed would have a window in which it could only drop them.

type DialerConfig 

type DialerConfig struct {
	// Credential supplies the service token. Required.
	Credential Credential
	// Version is the Factory build identity reported to a Host. Required.
	Version string
	// Limits supplies the dial timeout and the reconnect backoff. A zero value
	// takes DefaultLimits.
	Limits Limits
}

DialerConfig composes the real transport.

type EndpointError added in v0.5.0 

type EndpointError struct {
	Host   sessionwire.HostID
	Tenant sessionwire.TenantID
	Cause  *sessionwire.HostLinkEndpointError
}

EndpointError is ErrNoTenantEndpoint with its detail: which Host, which tenant, and Core's typed refusal, whose Code a caller may branch on and a log line should carry.

func (*EndpointError) Error added in v0.5.0 

func (e *EndpointError) Error() string

func (*EndpointError) Unwrap added in v0.5.0 

func (e *EndpointError) Unwrap() []error

Unwrap makes the error both ErrNoTenantEndpoint and Core's *HostLinkEndpointError, so a caller can test the class with errors.Is and read the code with errors.As without knowing this type.

type HostDisconnect 

type HostDisconnect struct {
	// Host is the Host that closed.
	Host sessionwire.HostID
	// Code is the close code, e.g. 3500 for an invalid service token.
	Code uint32
	// Reason is the Host's own short description, a diagnostic only.
	Reason string
}

HostDisconnect reports a Host closing a connection TERMINALLY.

The code is a structured field rather than a sentence, because a caller and a test both need to branch on it and neither may do that by matching text. centrifuge-go only surfaces a close it will not reconnect from -- codes 3500-3999 and 4500-4999 (centrifuge-go@v0.12.0/client.go:989) -- so every value that reaches here is one a reconnect would not fix.

func (*HostDisconnect) Error 

func (e *HostDisconnect) Error() string

func (*HostDisconnect) Unwrap 

func (e *HostDisconnect) Unwrap() error

Unwrap puts a terminal close in the dial-failure class, so a caller that only cares that the connection is unusable does not have to enumerate the codes.

type HostFailure added in v0.3.0 

type HostFailure struct {
	// Method is the RPC method the Host answered.
	Method string
	// Code and Message are the transport error the Host replied with. They are
	// diagnostics only.
	Code    uint32
	Message string
}

HostFailure is a Host's own ANSWER to an RPC that carried no Core record: a transport-level error reply, such as centrifuge's ErrorInternal (code 100).

It is kept apart from every other RPC failure because the Host ANSWERED: a cancelled context, a lost connection or an unanswered request is a plain error, and centrifuge-go builds *Error only from a reply the server sent. It says nothing about what the Host holds. host v0.2.1 answers a failed attach this way (hostlink/attach.go errAttachFailed) after a failed launch, after a rollback that did not complete, and after an attach whose observation it could not publish -- in which case the session IS resident. Host's own contract is "not a placement outcome, but a Factory may retry". A caller that moves on to another Host instead is safe only because of the session LEASE: while this Host holds it, any other candidate refuses epoch_mismatch, so no second residency can form. It is still not a *HostRefusal: it names no reason a caller may branch on.

func (*HostFailure) Error added in v0.3.0 

func (e *HostFailure) Error() string

func (*HostFailure) Unwrap added in v0.3.0 

func (e *HostFailure) Unwrap() error

Unwrap puts every HostFailure in the ErrHostFailed class.

type HostRefusal 

type HostRefusal struct {
	sessionwire.HostLinkError
}

HostRefusal is a Host's own answer to a control request.

It carries Core's typed record rather than a message, because a caller must branch on the reason: an epoch mismatch is a stale route to re-resolve and a runtime mismatch is a placement decision to remake, and telling them apart by matching text is how a caller silently starts doing the wrong one.

The record is EMBEDDED rather than copied field by field. Core's HostLinkError already states which detail belongs with which code -- an epoch only with a mismatch, a runtime id only with a runtime mismatch -- and a second flat copy of those three fields would be a second authority for that rule, which is the shape of defect this module has had to correct twice.

func (*HostRefusal) Error 

func (e *HostRefusal) Error() string

type Limits 

type Limits struct {
	// MaxLinks bounds concurrent HostLinks. One link multiplexes every session
	// binding of ONE TENANT to one Host, so this bounds (Host, tenant) pairs --
	// Hosts times the tenants this replica serves on each -- never sessions.
	MaxLinks int
	// DialTimeout bounds one dial.
	DialTimeout time.Duration
	// IdleTimeout is how long a link with no local demand is kept.
	IdleTimeout time.Duration
	// ReconnectMin and ReconnectMax bound the reconnect backoff.
	ReconnectMin time.Duration
	ReconnectMax time.Duration
}

Limits bounds one replica's HostLink pool.

It restates factory.HostLinkLimits for the package that consumes it, in the same way clientlink.Limits restates ClientLinkLimits. A9.1 converts; TestHostLinkLimitsAreCarriedWhole in the root package holds the two shapes in step by name and type, so a field added to one and not the other, or renamed on one side, fails there rather than configuring a zero.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits is what a pool composed with a zero Limits gets.

It restates factory.DefaultHostLinkLimits, held equal by TestTheTwoHostLinkDefaultsAreOneTable in the root package. A zero Limits is filled rather than rejected because the pool is composed from inside this module as well as from a deployer's options, and an in-module caller that had to restate five durations to open a connection would restate them wrongly.

func (Limits) Validate 

func (l Limits) Validate() error

Validate reports why these limits may not be used.

It is a SECOND check rather than a redundant one, for the reason clientlink.Limits.Validate is: the root package validates what a deployer supplied, and this validates what reached the engine, which is also what a direct in-module caller supplies. 

type Link interface {
	// Host is the Host this link was dialled for.
	Host() sessionwire.HostID
	// Bind establishes one Factory-local route to a Host-owned session. It is
	// a routing optimization and not a claim of ownership: the Host validates
	// the tuple against its own durable lease and may refuse.
	Bind(ctx context.Context, req sessionwire.HostLinkBindRequest) error
	// Unbind releases one route.
	Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
	// Attach asks the Host to make one session resident and returns the
	// registry observation it answered with, whose lease epoch is the one a
	// following Bind names. A Host that did not advertise hostlink.attach is
	// refused locally with *UnsupportedMethodError and is never sent one; a
	// Host's own refusal arrives as *HostRefusal.
	Attach(ctx context.Context, req sessionwire.HostLinkAttachRequest) (sessionwire.HostLinkRegistryObservation, error)
	// DeliverCommand hands an already committed inbox record's public command
	// id to the Host. The tenant and session travel beside the record because
	// Core's framing makes the RPC METHOD the session's channel,
	// sessionwire.HostLinkChannel(tenant, session); the body names no session.
	// See Pool.DeliverCommand for what a failure means.
	DeliverCommand(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, req sessionwire.HostLinkCommandDelivery) error
	// Close releases the connection.
	Close(ctx context.Context) error
}

Link is one live Factory-Host connection. Session bindings and RPCs are multiplexed over it, so the pool holds at most one per Host.

type Negotiator added in v0.5.0 

type Negotiator interface {
	// Negotiated returns the Host's most recent verified connect reply. It
	// fails with ErrLinkReconnecting between a dropped connection and the
	// next reply -- there is no reply to read then, and the previous one may
	// belong to a Host that has since restarted at another version -- and
	// with the link's terminal error once it will never answer again.
	Negotiated() (sessionwire.VersionNegotiationResponse, error)
}

Negotiator is the capability of a Link that can report the connect reply its Host last sent. It is discovered by assertion, like Subscriber, so a Link a test or a later transport supplies without it fails the capability question CLOSED rather than being asked something it cannot answer.

type Observer 

type Observer interface {
	// ObserveCapacity receives one Host-level advertisement.
	ObserveCapacity(report sessionwire.HostLinkCapacityReport)
	// ObserveRegistry receives one per-session ownership observation.
	ObserveRegistry(observation sessionwire.HostLinkRegistryObservation)
}

Observer receives what a Host pushes about itself and about the sessions it owns.

Both methods are called from the transport's read path, so an implementation that blocks stalls the connection. Neither may return an error: there is nothing a link could do with one, and a link that dropped a connection because a downstream consumer disliked an observation would make a reconciliation preference into an availability failure.

type Pool 

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

Pool holds at most one physical HostLink per (Host, tenant).

The invariant it exists for is in one sentence: a session binding never costs a connection, and a connection is never shared between Hosts OR BETWEEN TENANTS. Everything else here -- the ceiling, the idle window, the conflict refusal -- protects that sentence against the ways a caller can accidentally violate it.

The tenant half is v0.5.0's (Gap 1), and it is the Host's rule, not a preference: a Host serves each tenant's HostLink at its own derived address and authenticates the connection for the tenant that address names, so a link IS one tenant's. Keying by (Host, tenant) is what lets one pooled Host hold several tenants' sessions at once, and it is what keeps R-1 structural: every route names its tenant, a route's link is looked up under that tenant, so no path here can deliver, bind or subscribe one tenant's session over another tenant's connection. Every per-link structure -- the negotiated capability set, reconnect state, live-tail subscriptions and their regMu -- lives INSIDE the link, and is therefore per (Host, tenant) by construction.

Nothing in this type coordinates with another Factory replica. Runbook A7.1 step 2 is a decision, not an omission: several replicas each hold their own connection to one Host, and the Host reconciles what it is told against its own durable leases. There is no broker and no leader to lose.

func NewPool 

func NewPool(cfg Config) (*Pool, error)

NewPool validates a composition and returns it.

func (*Pool) AcceptsCommandPrincipal added in v0.12.0 

func (p *Pool) AcceptsCommandPrincipal(ctx context.Context, target Target, tenantID sessionwire.TenantID) (bool, error)

AcceptsCommandPrincipal asks the tenant's HostLink for its last verified attribution capability reply.

func (*Pool) AcceptsGateResponses added in v0.5.0 

func (p *Pool) AcceptsGateResponses(ctx context.Context, target Target, tenantID sessionwire.TenantID) (bool, error)

AcceptsGateResponses reports whether the Host at target can apply a gate_response command for tenantID's sessions, asked of THAT TENANT'S link -- the one the command would be read over -- and answered by the pool's gate-response predicate (GateResponseCapable unless the composition supplied another).

The link is acquired WITHOUT holding the pool's lock across a dial (spec gate N2): this is the one path a public HTTP request (a gate response) can reach that may open a link, and holding Pool.mu across a dial to an owner whose base black-holes would stall every bind, unbind and delivery on every Host for up to DialTimeout. See acquireUnlocked. The dial is bounded by the caller's context as well as the dialer's own timeout. A link that cannot report its reply answers false: an unknown capability is not one. A read that fails -- reconnecting, or a link that turned terminal -- is returned, and a terminal link is evicted as Attach evicts one, so the caller can tell "this Host cannot" (false, nil) from "this Host could not be asked" (false, err).

func (*Pool) AcceptsPayloadReferences added in v0.14.0 

func (p *Pool) AcceptsPayloadReferences(ctx context.Context, target Target, tenantID sessionwire.TenantID) (bool, error)

AcceptsPayloadReferences asks the tenant's HostLink whether the Host can dereference an inbox payload object before placement or delivery.

func (*Pool) Attach added in v0.3.0 

func (p *Pool) Attach(ctx context.Context, target Target, req sessionwire.HostLinkAttachRequest) (sessionwire.HostLinkRegistryObservation, error)

Attach asks target to make one session resident and returns the Host's observation of the residency it now holds.

The request must name the Host the link goes to, for Bind's reason, and it is validated before anything is dialled. The link is acquired under the pool's lock -- so a burst of attaches to one Host opens one connection, as a burst of binds does -- but the RPC is made OUTSIDE it, and that is the one way this differs from Bind. An attach can launch a runtime, which takes as long as a runtime takes to start; holding the pool across it would stall every bind, unbind and delivery on every Host for that long. Nothing here needs the ordering the lock gives Bind: an attach records no route, so there is no route table entry for a concurrent operation to race.

What that costs is stated: an attach records no binding, so the link it runs on can be collected by the reaper at ANY point from the acquisition to the end of the RPC -- before the request is sent, or while the Host is working on it. Either way the RPC fails as a transport error, which the caller sees as an attempt that did not complete and retries under the same idempotency key; a Host that had already attached keeps the residency, and the next pass finds the owner. Only the attach that DIALLED the link is safe from it at the start, because a link's idle window starts at the dial.

A link that turns out to be TERMINAL -- another wire version, or a close the transport will not reconnect from -- is evicted here, so the next attach to that Host dials afresh instead of being refused by a dead link forever.

An accepted observation is checked against the request before it is returned. The Host is required to answer for the session and incarnation it was asked about, and a caller is about to bind with the observation's epoch, so an answer about anything else is refused with ErrAttachMismatch rather than trusted.

func (*Pool) Bind 

func (p *Pool) Bind(ctx context.Context, target Target, req sessionwire.HostLinkBindRequest) error

Bind establishes one session route over the pooled link to target.

Both the target and the record are validated BEFORE anything is dialled, so a malformed control record costs neither a TCP connection nor a service handshake. The record must name the Host the link goes to: a bind carrying another Host's tuple would be refused by the receiving Host for a reason that reads as a lease problem rather than as a Factory routing bug.

Re-binding a session to the SAME Host is allowed and resends -- that is how a lease epoch is refreshed. Re-binding it to a DIFFERENT Host is refused rather than moved: a silent move leaves the first Host holding a route this pool no longer tracks and can therefore never unbind. The caller unbinds first.

func (*Pool) Bindings 

func (p *Pool) Bindings(host sessionwire.HostID) int

Bindings reports the session routes multiplexed over one Host's links, over every tenant.

func (*Pool) Close 

func (p *Pool) Close(ctx context.Context) error

Close closes every link and refuses further work.

Every link is closed even when one fails, and every failure is reported: one wedged connection must not strand the rest, and a Close that returned at the first error would leave the others open with no caller left to close them.

It is idempotent, and the idempotence comes from EMPTYING the table rather than from a closed check: a second call finds no links and closes nothing. An `if p.closed { return nil }` fast path was written here first and removed after a mutation showed it to be inert -- this lane does not keep a guard that guards nothing.

func (*Pool) DeliverCommand 

func (p *Pool) DeliverCommand(ctx context.Context, tenantID sessionwire.TenantID, sessionID sessionwire.SessionID, delivery sessionwire.HostLinkCommandDelivery) error

DeliverCommand hands one already committed inbox record to the Host that owns its session.

The reply is about DELIVERY, not about the command's outcome. A failure to hand it over is wrapped in ErrCommandUndelivered and leaves the inbox record pending; the Host's own refusal arrives as *HostRefusal and is returned unwrapped, so a caller can tell "never seen" from "answered". The binding survives either way: a transport failure is not evidence that the route is wrong, and this pool holds no store handle with which it could record anything about the record at all.

func (*Pool) Limits 

func (p *Pool) Limits() Limits

Limits returns the limits this pool was composed with, defaults included. 

func (p *Pool) Links() int

Links reports the physical connections this pool currently holds: one per (Host, tenant) pair.

func (*Pool) ReapIdle 

func (p *Pool) ReapIdle() int

ReapIdle closes every link that has had no binding for longer than IdleTimeout and reports how many it closed.

The cadence is the CALLER's, and the clock is read from the pool rather than passed in, so there is one authority for "now". A pool that ran its own timer would be a goroutine with a lifetime this type does not own, and A9.1 owns the start/stop ordering that would give it one.

func (*Pool) RouteFor 

func (p *Pool) RouteFor(tenantID sessionwire.TenantID, sessionID sessionwire.SessionID) (sessionwire.HostID, bool)

RouteFor reports which Host a tenant-scoped session is bound to.

It exists because the route table is the pool's actual routing authority and nothing else can observe it. Bindings counts entries in a DIFFERENT map, held per link, and the two are only kept in step by this file: a route released without its binding, or a binding released without its route, satisfies every count this type otherwise exposes. The second of those is the more expensive one -- an orphaned route names a Host the reaper is then free to collect, because the reaper's own test is the binding set -- so the table is made observable rather than inferred.

func (*Pool) Subscribe added in v0.4.0 

func (p *Pool) Subscribe(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, sink SessionSink) error

Subscribe starts a session's live tail on the link its route names.

The route is REQUIRED, and that is the Host's rule restated rather than a preference: a Host accepts a subscribe only from a link that already holds the bind for that channel on the same connection, so a subscribe with no route is one the Host will refuse. The pool's lock is released before the link is asked, because a subscribe waits for the Host's answer and holding the pool across it would stall every bind on every Host for that long.

func (*Pool) TenantBindings added in v0.5.0 

func (p *Pool) TenantBindings(host sessionwire.HostID, tenantID sessionwire.TenantID) int

TenantBindings reports the session routes multiplexed over one tenant's link to one Host. 

func (p *Pool) TenantLinks(host sessionwire.HostID) int

TenantLinks reports how many tenants' links this pool holds to one Host.

func (*Pool) Unbind 

func (p *Pool) Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error

Unbind releases one session route.

The local route is released even when the Host could not be told. A bind is Factory-local routing state and the Host validates ownership against its own durable lease, so a pool that kept a route it had failed to release would hold it forever: there is no retry, and the next Bind for that session would be refused as a conflict against a route nobody wants.

func (*Pool) Unsubscribe added in v0.4.0 

func (p *Pool) Unsubscribe(tenant sessionwire.TenantID, session sessionwire.SessionID)

Unsubscribe stops a session's live tail on EVERY link OF ITS TENANT that carries one. Another tenant's links cannot carry it -- a subscribe is made only over the route's own tenant's link -- so they are not asked.

It does not consult the route, deliberately: an unsubscribe is how the owner drops a tail whose route is being given up, and a route may already have been released -- or evicted with its link -- by the time the owner gets here. A tail left on a link because its route was gone first is the stale subscription Gap 3's ruling names: a Host never unsubscribes a Factory, so nothing else would ever stop it.

type SessionSink added in v0.4.0 

type SessionSink interface {
	// Subscribed reports the Host accepted the subscription. It is called
	// BEFORE Subscribe returns to its caller, so a sink can tell a
	// subscription started inside a known window from one started later.
	Subscribed()
	// Publication is one payload the Host published on the session channel,
	// exactly as it arrived. The slice is the sink's to keep.
	Publication(data []byte)
	// Ended reports a LIVE subscription that stopped without its owner asking:
	// the connection dropped or was closed, or the Host removed it. Every
	// publication from then until a new subscription is live is gone -- the
	// Host keeps no history -- and that is what a sink must repair.
	Ended()
	// Restored reports that the connection a subscription was lost to has
	// reconnected and negotiated again. Nothing is resubscribed: a Host
	// refuses a subscribe until the link re-binds on the NEW connection, so
	// re-binding and then subscribing is the sink owner's to sequence.
	Restored()
}

SessionSink receives one session channel's live tail.

EVERY METHOD IS CALLED ON THE TRANSPORT'S CALLBACK GOROUTINE AND MUST NOT BLOCK. centrifuge-go runs a publication handler synchronously on the goroutine that READS the connection (runHandlerSync), so a sink that waited for anything would stall every RPC reply on the link -- including, if the thing it waited for were a caller of Bind, the reply that caller is waiting for. A sink hands the event to its own goroutine and returns.

The order a sink sees is the order the transport delivered: Subscribed before the first Publication of that subscription, and nothing after Ended except, possibly, Restored.

type Subscriber added in v0.4.0 

type Subscriber interface {
	// Subscribe starts one session's live tail on this link and returns once
	// the Host has accepted it, refused it, or the context or the link's own
	// bound expired. A session already subscribed returns nil and keeps its
	// existing sink.
	Subscribe(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, sink SessionSink) error
	// Unsubscribe stops one session's live tail. It is the OWNER's call, so
	// its sink is not told Ended. It does not block on the Host.
	Unsubscribe(tenant sessionwire.TenantID, session sessionwire.SessionID)
}

Subscriber is the capability of a Link that carries a session's live tail.

It is separate from Link, and discovered by assertion in the pool, because a subscription is a transport operation rather than a HostLink method: Core names no reserved method for it, a Host advertises none, and the capability gate that admits bind and attach has nothing to say about it.

type Target 

type Target struct {
	Host     sessionwire.HostID
	Endpoint sessionwire.InternalEndpoint
	// Generation is the Host incarnation the endpoint was advertised under,
	// or zero when the caller does not know it. It is read by a path that
	// carries no request of its own (the gate-response capability read); a
	// bind and an attach take the generation their request is fenced to. The
	// pool uses it only to decide which of two advertisements of one HostID
	// is newer -- see Pool.acquireLocked -- and never dials it.
	Generation uint64
}

Target is one Host and the address to reach it at.

SINCE v0.5.0 THE ENDPOINT A CALLER HANDS THE POOL IS THE HOST'S ADVERTISED BASE, and the pool derives each tenant's address from it with Core's sessionwire.HostLinkEndpoint(base, tenant). A Host serves a separate HostLink per tenant, at HostLinkPathPrefix plus the tenant under its base (core v0.10.0, host v0.3.0), and a verbatim dial of the base is answered 404. The Dialer is handed the DERIVED address, so a Target reaching Dial names one tenant's link; nothing above the pool ever holds a derived one.

The two travel together because neither is usable alone: the identity is what the pool keys a connection by and what a bind record must name, and the endpoint is a durable routing observation that a registry entry or capacity report supplied. Core states what a usable endpoint is -- a bounded, credential-free WebSocket address -- and Validate defers to it rather than restating the rule.

func (Target) Validate 

func (t Target) Validate() error

Validate reports why this target may not be dialled.

type UnsupportedMethodError added in v0.2.0 

type UnsupportedMethodError struct {
	Method string
}

UnsupportedMethodError identifies the reserved operation refused locally because the negotiated HostLink capability set did not contain it.

func (*UnsupportedMethodError) Error added in v0.2.0 

func (e *UnsupportedMethodError) Error() string

func (*UnsupportedMethodError) Unwrap added in v0.2.0 

func (e *UnsupportedMethodError) Unwrap() error

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.