placement

type AttachRefusal ¶ added in v0.3.0

type AttachRefusal struct {
	sessionwire.HostLinkError
}

AttachRefusal is a Host's own coded answer to an attach. A HostLinks implementation returns it for a Host's HostLinkError.

It embeds Core's record for the reason hostlink.HostRefusal does: Core already states which detail travels with which code, and a flat copy would be a second authority for that rule.

func (e *AttachRefusal) Error() string

type CandidateRefusal ¶ added in v0.3.0

type CandidateRefusal struct {
	HostID         sessionwire.HostID
	HostGeneration uint64
	Code           sessionwire.HostLinkErrorCode
}

CandidateRefusal is one candidate a placement asked and was refused by.

type Catalog ¶

type Catalog interface {
	GetCatalogEntry(ctx context.Context, req sessionstore.GetCatalogEntryRequest) (sessionstore.CatalogEntry, error)
	UpdateCatalogDesiredState(ctx context.Context, req sessionstore.UpdateCatalogDesiredStateRequest) (sessionstore.CatalogEntry, error)
}

Catalog is the durable session record: the desired state a Factory authors and the record every decision here is made against.

type ClaimSweep ¶

type ClaimSweep interface {
	ReleaseReconciliationClaim(context.Context, sessionstore.ReleaseReconciliationClaimRequest) (sessionstore.ReconciliationClaimEntry, error)
}

ClaimSweep is the one durable write the claim half makes.

It is Release and nothing else. Acquire is deliberately absent: a sweeper that could acquire would be a second placement reconciler, and the work a claim suppresses is Reconciler's. Get is absent for a sharper reason — GetReconciliationClaim refuses a lapsed claim rather than returning it (sessionstore@v0.8.0/reconcile.go:450), so it cannot answer the question this sweep asks, and a caller that read it first would still have to attempt the release to learn anything.

type ClaimSweepResult ¶

type ClaimSweepResult struct {
	Examined int

	// Released counts the sessions whose claim is this replica's and is no
	// longer live afterwards. It deliberately does NOT separate "a write
	// happened" from "the record was already ours and lapsed": the store
	// answers both with success and no error, because a repeat release is the
	// ordinary case rather than a mistake, and a sweeper that reported them
	// differently would be reporting the reply it happened to race.
	Released int

	// Held counts the sessions another replica is reconciling right now.
	Held int

	// Lapsed counts another replica's claim that has run out. Nothing is
	// written for one and nothing needs to be: a lapsed claim is already
	// indistinguishable from no claim to every reader of this record.
	Lapsed int

	// Absent counts the sessions with no claim record at all.
	Absent int
}

ClaimSweepResult accounts for every session one SweepClaims call examined.

type Claims ¶

type Claims interface {
	AcquireReconciliationClaim(ctx context.Context, req sessionstore.AcquireReconciliationClaimRequest) (sessionstore.ReconciliationClaimEntry, error)
	ReleaseReconciliationClaim(ctx context.Context, req sessionstore.ReleaseReconciliationClaimRequest) (sessionstore.ReconciliationClaimEntry, error)
}

Claims is the short-lived reconciliation claim of specification section 10.

type Clock ¶

type Clock interface {
	Now() time.Time
}

Clock is the time seam. It declares only Now: this package sets a claim horizon and reads an owner's expiry, and schedules nothing.

type Config ¶

type Config struct {
	Directory Directory
	Catalog   Catalog
	Claims    Claims

	// Workloads is OPTIONAL, and its absence is a supported deployment rather
	// than a degraded one: H5's cmd/factory composes exactly this. See
	// ErrNoWorkloadController.
	Workloads WorkloadController

	Clock Clock

	// HolderID names this replica in a claim. It is not authority -- the claim
	// licenses nothing -- but it must be stable for a process, because
	// extending one's own claim and taking over a crashed replica's lapsed one
	// are the same store write and the holder is what tells them apart.
	HolderID string

	// ClaimTTL is how long this replica claims a session's scaling work for.
	ClaimTTL time.Duration

	// CandidateLimit bounds one capacity page.
	CandidateLimit int

	// Links is OPTIONAL. With it, a pooled session with no live owner is
	// ATTACHED to the first ranked admissible candidate that accepts it and
	// bound with the epoch that Host answered (B5). Without it, the reconciler
	// only NAMES the candidate in Result.Decision, which is what it did before
	// any caller existed; a composition with no HostLink pool uses that.
	Links HostLinks

	// ActorID is the requesting SERVICE identity an attach carries. Required
	// when Links is set. It names who asked for residency, never whose
	// authority a later command carries: placement runs from a sweeper with no
	// live user behind it.
	ActorID string

	// ReplaceAttempts bounds how many times one call re-runs placement after an
	// epoch_mismatch, counting the first attempt. Zero takes a small default.
	ReplaceAttempts int

	// ReplaceBackoff is the wait before the first re-placement, doubled for
	// each further one. Zero takes a small default.
	ReplaceBackoff time.Duration

	// Wait sleeps for a backoff or until the context ends. Nil uses a timer. It
	// is a seam so a re-placement's backoff is measured rather than waited for.
	Wait func(context.Context, time.Duration) error

	// Logger receives the one diagnostic this package emits: a pooled
	// candidate excluded because it does not advertise hostlink.attach, named
	// by Host, so a mixed fleet is diagnosable. Nil discards.
	Logger *slog.Logger
}

Config is one replica's placement configuration.

type Decision ¶

type Decision struct {
	Outcome Outcome
	Owner   sessionwire.HostLinkRegistryObservation
	Target  sessionwire.HostLinkCapacityReport
}

Decision is one placement answer together with exactly the record the outcome licenses acting on.

Owner is populated only for OutcomeReuseOwner and Target only for OutcomeAttachPooled; the other member stays zero. That is deliberate rather than tidy: a decision carrying both would let a caller act on the one its switch reached rather than the one the policy chose.

func Decide(
	record sessionstore.CatalogRecord,
	owner sessionwire.HostLinkRegistryObservation,
	ownerObserved bool,
	candidates []sessionwire.HostLinkCapacityReport,
	now time.Time,
) Decision

type Desired ¶

type Desired struct {
	Placement              sessionwire.HostPlacement
	RuntimeCompatibilityID string
	Workload               sessionstore.DesiredWorkload
}

Desired is the desired state a caller wants stored for a session.

It is an INPUT rather than something this package derives, and that boundary is where A4.2 stops. What a dedicated session's workload payload should contain is a launch-template question owned by composition (A9.1); this package's job is that whatever the caller names is stored idempotently and handed to the controller with a generation.

type Directory ¶

type Directory interface {
	Owner(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID) (sessionwire.HostLinkRegistryObservation, bool, error)
	Candidates(ctx context.Context, req sessionstore.ListCompatibleHostsRequest) (sessionstore.HostTargetPage, error)
}

Directory is the observed target directory: which Host currently owns a session, and which Hosts could accept one.

type HostLinks ¶ added in v0.3.0

type HostLinks interface {
	Attach(ctx context.Context, endpoint sessionwire.InternalEndpoint, req sessionwire.HostLinkAttachRequest) (sessionwire.HostLinkRegistryObservation, error)
	Bind(ctx context.Context, endpoint sessionwire.InternalEndpoint, req sessionwire.HostLinkBindRequest) error
	Unbind(ctx context.Context, req sessionwire.HostLinkUnbindRequest) error
	DeliverCommand(ctx context.Context, tenant sessionwire.TenantID, session sessionwire.SessionID, delivery sessionwire.HostLinkCommandDelivery) error
	// RouteFor reports whether this replica already routes the session, and
	// to which Host. Placement leaves a route it did not create in place.
	RouteFor(tenant sessionwire.TenantID, session sessionwire.SessionID) (sessionwire.HostID, bool)
}

HostLinks is the Host-facing half of pooled placement, stated in Core's vocabulary. The HostLink pool implements it through one adapter in the composition, which is also where transport errors are classified onto ErrAttachUnsupported, ErrHostUnreachable and *AttachRefusal: this package names no transport type, for the reason internal/routing's Binder gives.

type Outcome ¶

type Outcome uint8

Outcome is the closed set of answers Decide gives.

It is an enum rather than a pair of booleans so a caller's handling is exhaustive by construction, and Undecided is a member rather than a zero to fall through: a record naming a desired placement this build does not understand must reach a caller as "no decision", not as the first arm of a switch that happens to be pooled.

const (
	// OutcomeUndecided means no rule here applies to the record, which today
	// means its desired placement is neither of the two admission models.
	OutcomeUndecided Outcome = iota

	// OutcomeReuseOwner means the session already has a live owner and no
	// placement is needed. Specification section 15 step 1.
	OutcomeReuseOwner

	// OutcomeAttachPooled names one ranked pooled candidate the caller should
	// ask to take the session. It is not a reservation: the candidate may
	// refuse, and only its lease acquisition decides ownership.
	OutcomeAttachPooled

	// OutcomeReconcileDedicated means one workload should exist for this one
	// session. Specification section 13.
	OutcomeReconcileDedicated

	// OutcomeNoCapacity means the target has no admissible seat right now.
	// It is not an error: pooled capacity is scaled by an autoscaler outside
	// this module, and the caller retries.
	OutcomeNoCapacity
)

func (o Outcome) String() string

type PendingCommands ¶ added in v0.3.0

type PendingCommands interface {
	ControlShards() int
	ListDueDispositionCommands(context.Context, sessionstore.ListDueDispositionCommandsRequest) (sessionstore.DispositionDueCommandPage, error)
}

PendingCommands is the disposition inbox's service-plane due query.

type PendingSweepResult ¶ added in v0.3.0

type PendingSweepResult struct {
	Shard      int
	Pages      int
	Examined   int
	Unreadable int
	Truncated  bool
	// Resumed reports that this pass continued from the position an earlier
	// truncated pass over the same shard kept, rather than from the head.
	Resumed bool

	// Sessions is the number of distinct sessions this pass reconciled.
	Sessions int
	// Outcomes counts the decisions those reconciliations reported.
	Outcomes map[Outcome]int
	// Attached counts the sessions this pass attached to a Host.
	Attached int
	// Deferred counts the sessions another replica held the claim for.
	Deferred int
	// NoController counts dedicated sessions this replica cannot place because
	// it composes no workload controller (ErrNoWorkloadController).
	NoController int
	// Failures counts the sessions whose reconciliation returned an error. A
	// failure is per session and never stops the pass: one session whose
	// registry is stale must not keep every other session in the shard cold.
	Failures int
}

PendingSweepResult is what one pass over one shard did.

type PendingSweeper ¶ added in v0.3.0

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

PendingSweeper places the sessions with open commands, one control shard per pass, round-robin.

It REMEMBERS WHERE EACH SHARD'S PASS STOPPED, and without that it starves. The due view is deadline-ordered from the head, and the head is where rows this sweep skips pile up: an applying command past its deadline, a claimed one whose claim outlived it, and -- until the disposition deadline sweep retires them -- expired pending ones. A pass that always read from the head would, once a shard held more than MaxPages*PageLimit of those, never again reach a live create behind them (the B5 spec gate's F3: 2 expired rows ahead of a live create at 2 rows per pass, never attached across 5 passes). So a truncated pass keeps the store's continuation for its shard, the next pass over that shard resumes from it, and only a pass that reaches the end re-arms at the head against a fresh bound. Retirement keeps the head short; this is what makes the sweep correct even when it does not.

The cost is stated: a resumed pass carries the bound its cycle STARTED with, which is the store's rule for a continuation, so a command admitted after a long cycle began is met when the next cycle starts rather than immediately. The position is advice: a restarted replica starts every shard at the head.

func NewPendingSweeper(cfg PendingSweeperConfig) (*PendingSweeper, error)

func (s *PendingSweeper) Horizon() time.Duration

func (s *PendingSweeper) Logger() *slog.Logger

func (s *PendingSweeper) Sweep(ctx context.Context, principal identity.Principal) (PendingSweepResult, error)

type PendingSweeperConfig ¶ added in v0.3.0

type PendingSweeperConfig struct {
	Authorizer SweepAuthorizer
	Pending    PendingCommands
	Placer     Placer
	Clock      Clock

	// Horizon is how far past now the due read reaches. It is the apply
	// deadline admission gives every command, so every command accepted up to
	// now is inside it.
	Horizon time.Duration

	// PageLimit bounds one due page and MaxPages one pass over one shard, with
	// admission.Reconciler's meaning: a shard with more open commands than
	// PageLimit*MaxPages is reported Truncated and finished by the next pass.
	PageLimit int
	MaxPages  int

	// Logger receives one line per session whose placement failed. Nil
	// discards.
	Logger *slog.Logger
}

PendingSweeperConfig is one replica's pending-work sweep.

type Placer ¶ added in v0.3.0

type Placer interface {
	Reconcile(ctx context.Context, req Request) (Result, error)
}

Placer is the per-session reconciliation a sweep drives. *Reconciler implements it.

type Reconciler ¶

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

Reconciler places one session at a time.

It holds no per-session state. Everything it decides from is read within the call: the catalog record, the registry observation and one capacity page. A replica that restarts mid-placement leaves only a claim, which lapses.

func NewReconciler(cfg Config) (*Reconciler, error)

func (r *Reconciler) ActorID() string

func (r *Reconciler) Logger() *slog.Logger

func (r *Reconciler) Reconcile(ctx context.Context, req Request) (Result, error)

type RecordSweeper ¶

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

RecordSweeper sweeps the two CONTROL RECORDS placement leaves behind: a crashed Host's target advertisement and this replica's own placement claim.

IT IS NOT Reconciler AND DOES NOT EXTEND IT, which is worth stating because both live in this package and both are called "reconciliation" upstream. Reconciler's subject is a SESSION: it reads a catalog record, takes that session's claim and drives a workload controller. This one's subject is a RECORD, it names no session of its own in the target half, and it makes no placement decision at all. They were kept separate rather than fused because a fused type would have to hold both a per-session request and a directory-wide continuation, and the continuation is the one piece of state this module carries between calls — putting it on the type that is invoked once per session would make its lifetime a function of traffic.

func NewRecordSweeper(cfg SweeperConfig) (*RecordSweeper, error)

func (s *RecordSweeper) SweepClaims(ctx context.Context, refs []SessionRef) (ClaimSweepResult, error)

func (s *RecordSweeper) SweepTargets(ctx context.Context) (TargetSweepResult, error)

type Request ¶

type Request struct {
	TenantID  sessionwire.TenantID
	SessionID sessionwire.SessionID
	Desired   *Desired

	// Wake is the already accepted commands to deliver once the session has a
	// route: after an attach this call made, or to a live owner it found. It
	// is a HINT -- delivery wakes a Host's consumption of its durable inbox and
	// carries only the retry-stable public CommandID -- so an empty Wake still
	// places, and a failed delivery is counted, not returned.
	Wake []sessionwire.CommandID
}

Request is one reconciliation.

Desired is optional. Nil means "reconcile what the record already says", which is the ordinary path for a session whose desired state was written at creation and has not changed since; no desired-state write is attempted at all in that case, so the generation cannot move.

type Result ¶

type Result struct {
	// Decision is the placement answer. It is the zero Decision when Deferred.
	Decision Decision

	// Deferred reports that another replica holds this session's claim and
	// this one did nothing. It is not an error: it is the claim working.
	Deferred bool

	// ClaimExpiresAt is the holder's horizon, populated only when Deferred. It
	// is how long a caller can wait before the work is anyone's again.
	ClaimExpiresAt time.Time

	// DesiredWrites counts the desired-state write calls this reconciliation
	// made: zero when the record already carried the intent, one for an
	// ordinary application, and two when a racer's different intent forced one
	// re-read and retry.
	//
	// It counts CALLS rather than reporting "this replica applied it", and the
	// difference is not a hedge. Two replicas deriving one intent derive one
	// idempotency key, so the loser's write is absorbed as a replay of the
	// winner's and the two are indistinguishable in the record they leave --
	// which is the whole point of the derived key. What is observable, and what
	// the caller can act on, is how much work this call did.
	DesiredWrites int

	// Intent is the stored desired state handed to the controller, populated
	// only for OutcomeReconcileDedicated.
	Intent sessionstore.PlacementIntent

	// Attached is the Host's observation of the residency an attach made by
	// THIS call produced. Its lease epoch is the one the bind named. Zero when
	// this call attached nothing.
	Attached sessionwire.HostLinkRegistryObservation

	// Bound reports that this call bound a route: to the Host it attached, or
	// to a live owner it woke.
	Bound bool

	// Delivered and DeliveryFailures count the Wake deliveries.
	Delivered        int
	DeliveryFailures int

	// Excluded names the admissible candidates skipped because they do not
	// advertise hostlink.attach; Unreachable those this replica could not ask;
	// Refused those that answered with a HostLinkError; Failed those that
	// answered with a failure carrying no code (ErrAttachFailed). Each is in
	// the order asked.
	Excluded    []sessionwire.HostID
	Unreachable []sessionwire.HostID
	Refused     []CandidateRefusal
	Failed      []sessionwire.HostID

	// Replacements counts the times placement re-ran after an epoch_mismatch.
	Replacements int
}

Result is what one reconciliation did.

type SessionRef ¶

type SessionRef struct {
	TenantID  sessionwire.TenantID
	SessionID sessionwire.SessionID
}

SessionRef names one session for the claim half.

type SweepAuthorizer ¶ added in v0.3.0

type SweepAuthorizer interface {
	AuthorizeServiceSweep(ctx context.Context, principal identity.Principal) error
}

SweepAuthorizer decides the sweep, as it does for every other cross-tenant sweep this module runs.

type SweeperConfig ¶

type SweeperConfig struct {
	Targets TargetSweep
	Claims  ClaimSweep

	// HolderID names this replica in a claim release. It must be the SAME
	// string this replica's Reconciler puts in Config.HolderID, because a
	// release is refused for any other holder and the whole reach of this
	// sweep is the claims this replica itself left behind.
	HolderID string

	// PageLimit bounds one due page.
	PageLimit int

	// MaxPages bounds one SweepTargets call.
	MaxPages int
}

SweeperConfig is one replica's record-sweep configuration.

type TargetSweep ¶

type TargetSweep interface {
	ReconcileHostTargets(context.Context, sessionstore.ReconcileHostTargetsRequest) (sessionstore.HostTargetReconcileResult, error)
}

TargetSweep is the service-owned directory sweep.

It is ONE method and it is the store's, not this package's, which is the whole structural claim of the target half. SessionStore revalidates every due row against that row's OWN STORED EXPIRY at a fresh clock reading and closes the write with a compare-and-swap onto the revision the due page reported (sessionstore@v0.8.0/host_targets.go:1631 reconcileHostTargetRow).

THE TWO MECHANISMS CATCH DIFFERENT HEARTBEATS, and the distinction was stated wrongly here before. reconcileHostTargetRow takes the due page's FROZEN BYTES (stored storage.OrderedRecord) and NEVER RE-READS THE ROW; only the clock it compares against is fresh. So a heartbeat that landed AFTER the page was read cannot present an unlapsed expiry to the revalidation at all -- its only catcher is the compare-and-swap, which loses against the revision that heartbeat advanced. The expiry revalidation catches the EARLIER landing: one already carried in the page's bytes when they were captured, which a weakly consistent due view can still name as due.

The conclusion is unchanged and rests on the pair: Factory must not re-derive that judgement, because a due observation this side of the seam is a hint about which rows to LOOK at and is never evidence that a Host is gone.

type TargetSweepResult ¶

type TargetSweepResult struct {
	// Scanned is every row the store looked at. It is the sum of the five
	// outcomes below, which the store asserts for itself.
	Scanned int

	// Unranked is the store's Withdrawn: rows this sweep removed from the
	// placement rank.
	Unranked int

	// Retained is the store's StillLive: rows the revalidation refused to
	// withdraw. A nonzero value is the heartbeat race being won by the Host,
	// which is the correct outcome and not a failure.
	Retained int

	Contended  int
	Unreadable int
	Unverified int

	// Exhausted reports that the sweep reached the end of the deadline view.
	Exhausted bool

	// Resumed reports that this call presented a continuation retained from an
	// earlier one. It is observable state rather than a log detail: a sweeper
	// that never resumes makes no progress past its first page budget, and
	// nothing else about the result would say so.
	Resumed bool
}

TargetSweepResult is what one SweepTargets call did.

Unranked and Retained are RENAMED rather than passed through, and the names are the point of the type. The store's Withdrawn is the count of rows whose own stored expiry had lapsed at the revalidation instant and whose compare-and-swap then won -- those, and only those, left the placement rank. StillLive is the count the revalidation SAVED: rows a due page named and a heartbeat rescued. Anything that reported the due page's own count as "removed" would be claiming a Host is gone on the strength of an observation that is explicitly weakly consistent.

type WorkloadController ¶

type WorkloadController interface {
	EnsureWorkload(context.Context, sessionstore.PlacementIntent) error
	ObserveWorkload(context.Context, sessionstore.PlacementIntent) (sessionwire.HostLinkRegistryObservation, bool, error)
	RequestDrain(context.Context, sessionstore.PlacementIntent) (sessionwire.HostLinkDrainObservation, error)
	DeleteWorkload(context.Context, sessionstore.PlacementIntent) error
}

WorkloadController is the consumer-owned domain seam for a dedicated workload's lifecycle.

Every argument and result is a Core or SessionStore record. In particular, the platform adapter receives the Factory-authored PlacementIntent and returns the Host-owned registry/drain observations; no Kubernetes, Nomad or other platform type can cross this boundary. The adapter may use the intent to derive its own platform object name, but the placement package never sees that object.

EnsureWorkload is idempotent for an unchanged intent. ObserveWorkload reports whether a Host registry observation exists for that intent. RequestDrain is the only lifecycle operation that initiates Host draining, and DeleteWorkload is called only after the caller has observed a completed drain. Keeping the ordering in the caller makes a platform adapter unable to treat a transport close or a platform deletion as graceful release.

type WorkloadIdentity ¶ added in v0.2.0

type WorkloadIdentity struct {
	Name                   string
	TenantID               sessionwire.TenantID
	SessionID              sessionwire.SessionID
	AgentID                sessionwire.AgentID
	RuntimeCompatibilityID string
	Generation             uint64
}

WorkloadIdentity is the stable identity of one desired workload revision.

Name is deliberately a digest rather than a concatenation of user-controlled identifiers. TenantID and SessionID may contain characters that are legal in Core but unsafe in platform names, and leaking either into a workload name also exposes tenant data to platform operators. The source fields remain available to a domain caller for correlation; only Name is intended for use as a platform object name.

func DeriveWorkloadIdentity(intent sessionstore.PlacementIntent) WorkloadIdentity