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, because Core v0.7.0 defines the session-channel record bodies but no HostLink transport framing to carry them -- the same gap centrifuge.go records about the method names. 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 HostRefusal
- type Limits
- type Link
- type Observer
- type Pool
- 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
Constants ¶
const ( // MethodBind establishes one Factory-local route to a Host-owned session. MethodBind = "hostlink.bind" // MethodUnbind releases one. MethodUnbind = "hostlink.unbind" // MethodCommand hands over one committed inbox record's public command id. MethodCommand = "hostlink.command" )
The HostLink control vocabulary.
These names, and the push envelope below, are FACTORY'S HALF of a protocol whose Host half does not exist in this repository. Core v0.7.0 defines the record bodies -- HostLinkBindRequest, HostLinkCapacityReport and the rest -- and their strict JSON, but it defines no transport framing for them: no method names, no push discriminator, no channel vocabulary. That gap is real and is recorded here rather than papered over. The Host writer must implement the mirror of exactly this, or one of the two must move.
Where these strings should eventually live is UNRESOLVED, and internal/ is not it: Host cannot import a Factory-internal package, so one half of a two-repo contract currently sits somewhere the other half cannot see. Core's sessionwire/v1 is deliberately NOT the answer either. A method name and a push discriminator are transport-SHAPED -- they exist because the transport is centrifuge, and they would read differently under gRPC or raw WebSocket frames -- while sessionwire/v1 is today transport-neutral: it says what a record is and nothing about how it travels. Putting hostlink.bind into a tier-0 module would leak the transport choice downward and make a future transport change a tier-0 breaking release. What this needs is a shared, explicitly transport-scoped home; booking one is not this package's to decide, and nothing here should be read as that decision having been taken.
Because the strings are the artifact Host mirrors, they are pinned as absolute literals by TestTheWireVocabularyIsPinnedToItsLiterals rather than through the constants themselves. A fixture built from the constant under test pins nothing: a rename would move both sides together and the suite would stay green while Factory stopped speaking the protocol Host implements.
const ( // PushTypeCapacity carries a HostLinkCapacityReport. PushTypeCapacity = "host.capacity" // PushTypeRegistry carries a HostLinkRegistryObservation. PushTypeRegistry = "host.registry" )
The asynchronous messages a Host pushes.
They are async messages rather than channel publications on purpose. A channel would need a namespace both ends agree on and 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 there is no agreed channel namespace to take it with -- that is the same gap as the method names -- 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 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 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 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 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 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
// DeliverCommand hands an already committed inbox record's public command
// id to the Host. See Pool.DeliverCommand for what a failure means.
DeliverCommand(ctx context.Context, 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) 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.
Source Files