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, command delivery, and the capacity and registry observations a Host pushes. It does NOT carry session event data. The per-binding queues and the backpressure repair now live ABOVE this package, in internal/realtime/delivery and internal/routing's Relay; what is still absent here is the LIVE TAIL itself. Core v0.9.1 names the session channel a Host publishes on (sessionwire.HostLinkChannel), so the framing gap this package used to record is closed for the control plane, but no subscription to that channel exists here yet. Nothing here may be read as having solved any of the three. It also does not decide WHICH Host a session belongs to; that is A7.2's demand-driven binding, which calls Bind and Unbind.
Index ¶
- Constants
- Variables
- type CentrifugeDialer
- type Config
- type Credential
- type Dialer
- type DialerConfig
- type HostDisconnect
- type HostFailure
- type HostRefusal
- type Limits
- type Link
- type Observer
- type Pool
- func (p *Pool) Attach(ctx context.Context, target Target, req sessionwire.HostLinkAttachRequest) (sessionwire.HostLinkRegistryObservation, error)
- func (p *Pool) Bind(ctx context.Context, target Target, req sessionwire.HostLinkBindRequest) error
- func (p *Pool) Bindings(host sessionwire.HostID) int
- func (p *Pool) Close(ctx context.Context) error
- func (p *Pool) DeliverCommand(ctx context.Context, tenantID sessionwire.TenantID, ...) error
- func (p *Pool) Limits() Limits
- func (p *Pool) Links() int
- func (p *Pool) ReapIdle() int
- func (p *Pool) RouteFor(tenantID sessionwire.TenantID, sessionID sessionwire.SessionID) (sessionwire.HostID, bool)
- func (p *Pool) Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
- type Target
- type UnsupportedMethodError
Constants ¶
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.
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 ¶
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.
var ErrBindingConflict = errors.New("hostlink: session is bound to another host")
ErrBindingConflict reports a session already bound to a different Host.
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.
var ErrDialFailed = errors.New("hostlink: dial failed")
ErrDialFailed is the class of every dial rejection that is not the dialer's own composition error.
var ErrHostFailed = errors.New("hostlink: the host answered with a failure carrying no Core code")
ErrHostFailed is the class of every HostFailure.
var ErrInvalidConfig = errors.New("hostlink: invalid configuration")
ErrInvalidConfig is the class of every NewPool rejection.
var ErrLinkLimit = errors.New("hostlink: link limit reached")
ErrLinkLimit reports a bind that would open a link past MaxLinks.
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.
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.
var ErrPoolClosed = errors.New("hostlink: pool is closed")
ErrPoolClosed reports work asked of a pool that has been shut down.
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.
var ErrUnknownBinding = errors.New("hostlink: no binding for session")
ErrUnknownBinding reports work for a session this pool holds no route for.
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.
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 ¶
This section is empty.
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
}
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 ¶
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 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". 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 to one Host, so this bounds Hosts, not 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 ¶
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 ¶
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 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.
The invariant it exists for is in one sentence: a session binding never costs a connection, and a connection is never shared between Hosts. Everything else here -- the ceiling, the idle window, the conflict refusal -- protects that sentence against the ways a caller can accidentally violate it.
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 (*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 link.
func (*Pool) Close ¶
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) ReapIdle ¶
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) 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.
type Target ¶
type Target struct {
Host sessionwire.HostID
Endpoint sessionwire.InternalEndpoint
}
Target is one Host and the address to reach it at.
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.
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