services

package
v5.0.1 Latest Latest
Warning

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

 Go to latest Published: Jul 28, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Overview

Package services is the payload-agnostic transport substrate for v5's off-thread work (plan item P3.3).

runtime2 already solved transport selection and degradation properly: pick the best tier the browser supports, fall back cleanly when an encode fails, and never silently ship a worse payload than the caller thinks. That logic is correct and worth keeping — but in runtime2 it is welded to DOM-patch types (SnapshotEnvelope, PatchStreamRaw), so a domain service moving rows or projections cannot use any of it.

This extracts the decision-making and leaves encoding to the payload, which is the only part that was ever DOM-specific. runtime2's own types satisfy the interface unchanged; see transport_runtime2_conformance_test.go.

Deliberately NOT extracted here: the scheduler, recovery coordinator, and idempotency ledger. The plan's P3.3 contingency puts a decision point after the transports precisely so their extraction can be judged on measurements from this one rather than committed to up front.

Index

Constants

View Source 
const DefaultLedgerRetention = 1024

DefaultLedgerRetention is how many recent versions per stream keep their identity recorded.

The source tracker kept every version's identity for the life of an epoch, which is unbounded: a long-lived stream at a high message rate grows a map entry per message forever. Acceptable when an epoch is short and a region is one of a few dozen; not acceptable for domain streams that may run for a session. See the retention argument on recordVersionLocked for why bounding this cannot change any apply/skip outcome.

Variables

View Source 
var ErrNoReadyWorker = errors.New("services: no ready worker is available")

ErrNoReadyWorker is returned when no worker can take a unit.

View Source 
var ErrNoSupportedTier = errors.New("services: no supported transport tier is available")

ErrNoSupportedTier is returned when the environment supports no tier at all.

View Source 
var ErrQueueFull = errors.New("services: dispatcher queue is at its limit")

ErrQueueFull is returned when backpressure rejects new work.

Functions

func Decode 

func Decode[T any](parseBytes []byte, parseTier Tier, parseCodec Codec[T]) (T, error)

Decode decodes a payload that was encoded at a known tier.

The tier is carried explicitly rather than sniffed from the bytes. Guessing would make a downgraded payload indistinguishable from a corrupt one, and the receiver would decode the wrong way with no error.

func IsKnownFailureClass 

func IsKnownFailureClass(parseClass FailureClass) bool

IsKnownFailureClass reports whether a class is one the arbiter understands.

Unknown classes are rejected rather than absorbed: a typo'd class that silently entered fallback would look identical to a real failure in diagnostics, and the unit would never be handed back.

func IsValidWorkerHealth 

func IsValidWorkerHealth(parseHealth WorkerHealth) bool

IsValidWorkerHealth reports whether a health value is one a caller may set.

Types

type Arbiter 

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

Arbiter tracks ownership and remote-output admissibility across named units.

Not safe for concurrent use; it belongs to one coordinating thread.

func NewArbiter 

func NewArbiter() *Arbiter

NewArbiter creates an arbiter with every unit remotely owned by default.

func (*Arbiter) AdmitRemote 

func (parseArbiter *Arbiter) AdmitRemote(parseUnitID string, parseEpoch uint64, parseVersion uint64) RemoteVerdict

AdmitRemote decides whether arriving remote output should be used.

This is the call that must not be simplified to an ownership check. Ownership covers the during-fallback case; the epoch floor covers output that was in flight when a worker was replaced; the version check covers output computed before the local side advanced the unit. All three are the same hazard — older state overwriting newer — arriving by different routes.

func (*Arbiter) Fallback 

func (parseArbiter *Arbiter) Fallback(parseUnitID string, parseClass FailureClass, parseReason string, parseEpoch uint64, parseVersion uint64) error

Fallback moves a unit to local ownership.

Idempotent for repeated failures on an already-local unit: the class and reason update to the most recent cause, and the version high-water mark only ever rises. Losing the original cause is the right trade — the most recent failure is what a caller diagnosing a stuck unit needs.

func (*Arbiter) Forget 

func (parseArbiter *Arbiter) Forget(parseUnitID string)

Forget drops a unit's arbitration state entirely, for units that genuinely end. It clears the local version and epoch floor along with ownership, so a unit that later resumes would readmit stale output — use Restore to hand a live unit back.

func (*Arbiter) HandleRemoteDeath 

func (parseArbiter *Arbiter) HandleRemoteDeath(parseUnitID string, hasReassignSupport bool, parseReassignEpoch uint64, parseVersion uint64) (DeathOutcome, error)

HandleRemoteDeath applies the reassign-or-fall-back policy for a dead remote.

The two "no reassignment happened" cases stay distinct rather than collapsing into one, because they mean opposite things about the system. A caller that does not support reassignment falling back is the design working; a caller that claims support and then supplies no epoch is a bug, and it still falls back — losing availability quietly would be worse — but it also returns an error so the bug is visible rather than absorbed as a normal recovery.

func (*Arbiter) LocalOwnedUnits 

func (parseArbiter *Arbiter) LocalOwnedUnits() []string

LocalOwnedUnits lists every unit currently in local fallback.

The source had no way to enumerate these, which made "why is the app slow" unanswerable when units had quietly fallen back one at a time. Order is unspecified; callers that display it should sort.

func (*Arbiter) LocalVersion 

func (parseArbiter *Arbiter) LocalVersion(parseUnitID string) uint64

LocalVersion reports the highest locally authoritative version for a unit.

func (*Arbiter) Owner 

func (parseArbiter *Arbiter) Owner(parseUnitID string) Owner

Owner reports which side is authoritative for a unit. Unknown units are remotely owned: nothing has failed.

func (*Arbiter) RecordLocalProgress 

func (parseArbiter *Arbiter) RecordLocalProgress(parseUnitID string, parseVersion uint64)

RecordLocalProgress records that the local side computed a unit up to a version while owning it. The high-water mark only rises.

func (*Arbiter) Restore 

func (parseArbiter *Arbiter) Restore(parseUnitID string, parseEpoch uint64) error

Restore hands a unit back to remote ownership at a fresh epoch.

The epoch must be strictly newer than anything seen, and that requirement is the whole mechanism: raising the floor invalidates the previous worker's in-flight output without the arbiter needing to know anything about it. Restoring at a reused epoch would readmit exactly the output the fallback existed to exclude, so it is refused rather than accepted with a warning.

func (*Arbiter) State 

func (parseArbiter *Arbiter) State(parseUnitID string) (FallbackState, bool)

State reports a unit's full ownership record, and whether one exists.

type Capabilities 

type Capabilities struct {
	HasBinary          bool
	HasStructuredClone bool
	HasShared          bool
}

Capabilities reports what the current environment supports.

type Codec 

type Codec[T any] interface {
	EncodeBinary(parsePayload T) ([]byte, error)
	EncodeJSON(parsePayload T) ([]byte, error)
	DecodeBinary(parseBytes []byte) (T, error)
	DecodeJSON(parseBytes []byte) (T, error)
}

Codec encodes and decodes one payload type for each tier.

Generic over the payload so the substrate never learns what it is carrying. A codec may report an error from any encode; that is the trigger for the documented downgrade rather than a fatal condition.

type DeathOutcome 

type DeathOutcome struct {
	// Reassigned means the unit is remotely owned again at RestoredEpoch.
	Reassigned    bool
	RestoredEpoch uint64
	// FellBack means the unit moved to local ownership instead.
	FellBack bool
}

DeathOutcome reports what happened to a unit when its remote side died.

type Decision 

type Decision uint8

Decision is what a ledger says to do with an arriving message. 

const (
	// DecisionApply means the message is new and in order.
	DecisionApply Decision = iota
	// DecisionSkipDuplicate means this exact (version, identity) already applied.
	// Skipping is the correct outcome, not an error — duplicate delivery is
	// normal across a worker bridge.
	DecisionSkipDuplicate
	// DecisionSkipStale means the version is below what has already been applied
	// in this epoch. Its payload describes older state, so applying it would
	// regress the receiver.
	DecisionSkipStale
	// DecisionSkipStaleEpoch means the message belongs to an epoch the stream has
	// already moved past.
	DecisionSkipStaleEpoch
)

func (Decision) ShouldApply 

func (parseDecision Decision) ShouldApply() bool

ShouldApply reports whether a decision means "apply this message".

Callers should branch on this rather than comparing against DecisionApply, so a future skip reason does not silently become an apply at existing call sites.

func (Decision) String 

func (parseDecision Decision) String() string

type Dispatcher 

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

Dispatcher routes units of work across a worker pool.

Not safe for concurrent use; it belongs to one coordinating thread.

func NewDispatcher 

func NewDispatcher(parseWorkerIDs []WorkerID) *Dispatcher

NewDispatcher creates a dispatcher over a worker pool with no queue limit.

func NewDispatcherWithQueueLimit 

func NewDispatcherWithQueueLimit(parseWorkerIDs []WorkerID, parseQueueLimit int) *Dispatcher

NewDispatcherWithQueueLimit creates a dispatcher with queue backpressure.

A non-positive limit means unbounded, which is the source's default and is appropriate only when the producer is already rate-limited by something else.

func (*Dispatcher) Assignment 

func (parseDispatcher *Dispatcher) Assignment(parseUnitID string) (WorkerID, bool)

Assignment reports a unit's current worker, if it has one.

func (*Dispatcher) Cancel 

func (parseDispatcher *Dispatcher) Cancel(parseUnitID string) bool

Cancel supersedes all queued and in-flight work for a unit.

Advancing the generation rather than hunting down in-flight work is what makes this correct: a worker already running the old job cannot be recalled, so its result is rejected at commit time instead. Queued work is dropped eagerly because it is still reachable.

func (*Dispatcher) CommitAllowed 

func (parseDispatcher *Dispatcher) CommitAllowed(parseUnitID string, parseGeneration uint64) bool

CommitAllowed reports whether a completed job's result may still be applied.

Three ways to lose the right to commit, all checked here: the unit was disposed, it fell back to local ownership, or it was cancelled and this result predates the cancel.

func (*Dispatcher) Degraded 

func (parseDispatcher *Dispatcher) Degraded() bool

Degraded reports whether any worker is not ready, or a repair failed.

func (*Dispatcher) Dispose 

func (parseDispatcher *Dispatcher) Dispose(parseUnitID string) bool

Dispose removes a unit entirely and reports whether anything was removed.

func (*Dispatcher) DrainQueue 

func (parseDispatcher *Dispatcher) DrainQueue() []Job

DrainQueue removes and returns every queued job, dropping ones a cancel has superseded. Draining is how a caller actually dispatches: the queue is a staging buffer, not a permanent record.

func (*Dispatcher) Fallback 

func (parseDispatcher *Dispatcher) Fallback(parseUnitID string) bool

Fallback marks a started unit as locally owned, which blocks its commits.

func (*Dispatcher) InFallback 

func (parseDispatcher *Dispatcher) InFallback(parseUnitID string) bool

InFallback reports whether a unit is locally owned.

func (*Dispatcher) IsStale 

func (parseDispatcher *Dispatcher) IsStale(parseJob Job) bool

IsStale reports whether a job has been superseded by a cancel.

func (*Dispatcher) QueueDepth 

func (parseDispatcher *Dispatcher) QueueDepth() int

QueueDepth reports how many jobs are queued.

func (*Dispatcher) ReadyWorkers 

func (parseDispatcher *Dispatcher) ReadyWorkers() []WorkerID

ReadyWorkers lists workers that can take work, in canonical order.

func (*Dispatcher) RecordHeartbeat 

func (parseDispatcher *Dispatcher) RecordHeartbeat(parseWorkerID WorkerID, parseSequence uint64) error

RecordHeartbeat records a heartbeat and restores the worker to ready.

Sequence numbers must not go backwards: an out-of-order heartbeat from before a failure would otherwise mark a dead worker ready again.

func (*Dispatcher) RecordHeartbeatTimeout 

func (parseDispatcher *Dispatcher) RecordHeartbeatTimeout(parseWorkerID WorkerID) (WorkerHealth, error)

RecordHeartbeatTimeout records a missed heartbeat window.

One miss degrades, two kills. The intermediate state exists because a single missed window is usually a long task or a scheduling hiccup, and replacing a worker over it would discard live state for no reason.

func (*Dispatcher) ReplaceWorker 

func (parseDispatcher *Dispatcher) ReplaceWorker(parseDeadWorkerID WorkerID, parseReplacementWorkerID WorkerID) (ReplaceOutcome, error)

ReplaceWorker swaps a dead worker for a replacement and repairs its units.

EVERY affected unit is repaired, including after one of them fails. The source returns on the first failure, which leaves the remaining units assigned to a worker it has already deleted from the pool — they then dispatch into nothing and never recover, and the symptom appears far from the cause. Collecting the failures and continuing costs nothing and makes the outcome describable.

func (*Dispatcher) SetWorkerHealth 

func (parseDispatcher *Dispatcher) SetWorkerHealth(parseWorkerID WorkerID, parseHealth WorkerHealth) error

SetWorkerHealth sets a worker's health directly, for state a heartbeat cannot express — a deliberate restart, or a death observed out of band.

func (*Dispatcher) Start 

func (parseDispatcher *Dispatcher) Start(parseUnitID string) (Job, error)

Start establishes a unit's assignment and enqueues its first job.

Re-starting an already-assigned unit keeps its worker when that worker is still ready. Stability matters more than balance here: moving a unit discards whatever state its worker had built for it.

func (*Dispatcher) Update 

func (parseDispatcher *Dispatcher) Update(parseUnitID string) (Job, error)

Update enqueues work on an established assignment, coalescing when possible.

Coalescing is the reason this returns the EXISTING job when one is already queued for the same unit and generation: a burst of updates collapses to one dispatch carrying the newest state, rather than a queue of superseded work the worker would grind through in order.

func (*Dispatcher) WorkerHealthOf 

func (parseDispatcher *Dispatcher) WorkerHealthOf(parseWorkerID WorkerID) WorkerHealth

WorkerHealthOf reports one worker's health.

type EncodeResult 

type EncodeResult struct {
	Tier  Tier
	Bytes []byte
	// Downgraded records that the preferred tier failed to encode and a lower
	// one was used. R4: a degradation is reported, never silent.
	Downgraded bool
	// DowngradeReason explains why, for diagnostics.
	DowngradeReason string
}

EncodeResult carries an encoded payload and how it was actually encoded.

Tier is what the payload IS, not what was asked for. A caller that logs the requested tier rather than this one will misreport every downgrade.

func Encode 

func Encode[T any](parsePayload T, parseCodec Codec[T], parseCapabilities Capabilities) (EncodeResult, error)

Encode encodes a payload at the best available tier, downgrading on failure.

Mirrors runtime2's BuildSnapshotTransportPayloadWithFallback: prefer binary, and if binary encoding fails, fall back to structured clone WHEN AVAILABLE. If it is not available the binary error is returned unchanged, because reporting a downgrade that did not happen would be worse than failing.

type FailureClass 

type FailureClass string

FailureClass classifies why ownership moved.

The class is kept separate from the free-text reason because recovery policy branches on it: a transport failure may be retryable where a panic is not. Callers define their own reasons; the classes are the closed set the arbiter understands. 

const (
	// FailureTransport is a payload that could not be decoded or delivered.
	FailureTransport FailureClass = "transport"
	// FailureCommit is a failure applying an otherwise valid payload.
	FailureCommit FailureClass = "commit"
	// FailureWorkerDeath is the remote side terminating.
	FailureWorkerDeath FailureClass = "worker-death"
	// FailureTimeout is a remote side that did not answer in time. It is
	// distinguished from death because the worker may still be alive and may
	// still deliver — which is precisely the stale-output hazard.
	FailureTimeout FailureClass = "timeout"
	// FailurePanic is an unrecoverable fault inside the remote handler.
	FailurePanic FailureClass = "panic"
)

type FallbackState 

type FallbackState struct {
	Owner Owner
	Class FailureClass
	// Reason is the caller's free-text detail, for diagnostics only.
	Reason string
	// Epoch is the remote epoch in effect when ownership moved.
	Epoch uint64
	// Version is the highest locally authoritative version at that moment.
	Version uint64
}

FallbackState is one unit's ownership record.

type Job 

type Job struct {
	Kind     JobKind
	UnitID   string
	WorkerID WorkerID
	// CancelGeneration stamps the job with the unit's generation at enqueue time.
	// A job whose generation is behind the unit's current one has been superseded
	// and must not be committed, even if a worker already finished it.
	CancelGeneration uint64
}

Job is one queued unit of work.

type JobKind 

type JobKind string

JobKind distinguishes the first job for a unit from subsequent ones.

The distinction is load-bearing for coalescing: a start cannot be coalesced away because the worker has no prior state for the unit, while updates can because a later one supersedes an earlier one entirely. 

const (
	// JobStart is the first job for a unit, establishing its assignment.
	JobStart JobKind = "start"
	// JobUpdate is subsequent work on an established assignment.
	JobUpdate JobKind = "update"
)

type Ledger 

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

Ledger tracks duplicate, stale, and conflicting messages across named streams.

Not safe for concurrent use. A ledger belongs to one receiving thread, which is the arrangement everywhere it is used: one worker, one message loop.

func NewLedger 

func NewLedger() *Ledger

NewLedger creates a ledger with the default retention window.

func NewLedgerWithRetention 

func NewLedgerWithRetention(parseRetention int) *Ledger

NewLedgerWithRetention creates a ledger retaining the given number of recent versions per stream. A non-positive retention uses the default.

func (*Ledger) Apply 

func (parseLedger *Ledger) Apply(parseStreamID string, parseEpoch uint64, parseVersion uint64, parseIdentity string) (Decision, error)

Apply is the check-and-commit convenience for callers with no separate body validation step. Callers that validate a body must use Check and Commit instead; see the note on Check.

func (*Ledger) Check 

func (parseLedger *Ledger) Check(parseStreamID string, parseEpoch uint64, parseVersion uint64, parseIdentity string) (Decision, error)

Check reports what to do with a message WITHOUT mutating ledger state.

The split between Check and Commit is the fix for a real bug in the source (runtime2 #72): committing version state before the message body validated let a message that then failed validation record its version, so a later corrected message at that same version was rejected as a conflict and the stream wedged permanently. Commit only after the body is known good.

An error means a genuine conflict — two different payloads claiming one version — which is a producer bug the caller should surface. Every ordinary duplicate or reorder returns a skip decision and a nil error.

func (*Ledger) Commit 

func (parseLedger *Ledger) Commit(parseStreamID string, parseEpoch uint64, parseVersion uint64, parseIdentity string) error

Commit records an applied message. Call it only after Check returned an apply decision AND the message body validated. Repeating it for the same (version, identity) is harmless.

func (*Ledger) Forget 

func (parseLedger *Ledger) Forget(parseStreamID string)

Forget drops a stream's state entirely.

The source had no way to do this: a region's state lived as long as the tracker. Streams that genuinely end — a closed subscription, a completed bulk command — should release their state rather than accumulate for the session. Forgetting a stream that later resumes is safe but not free: it loses the high-water mark, so an in-flight stale delivery would be applied.

func (*Ledger) Streams 

func (parseLedger *Ledger) Streams() []string

Streams lists every tracked stream id.

Exists for P3.14: a worker reload must carry its replay state forward, and that means enumerating what has been recorded. Order is unspecified — map iteration — because the ledger's decisions never depend on it.

func (*Ledger) TrackedStreams 

func (parseLedger *Ledger) TrackedStreams() int

TrackedStreams reports how many streams hold state, for diagnostics and for tests that assert the ledger is not growing without bound.

func (*Ledger) TrackedVersions 

func (parseLedger *Ledger) TrackedVersions(parseStreamID string) int

TrackedVersions reports how many versions one stream currently retains, which is what the retention window bounds.

type Owner 

type Owner uint8

Owner names which side is authoritative for a unit of work. 

const (
	// OwnerRemote is the normal state: the worker owns the unit.
	OwnerRemote Owner = iota
	// OwnerLocal means a failure moved authority to the local side. Remote
	// output is ignored until ownership is explicitly restored.
	OwnerLocal
)

func (Owner) String 

func (parseOwner Owner) String() string

type RemoteVerdict 

type RemoteVerdict uint8

RemoteVerdict is what an arbiter says about arriving remote output. 

const (
	// VerdictAccept means the output is current and authoritative.
	VerdictAccept RemoteVerdict = iota
	// VerdictRejectLocalOwned means the local side currently owns the unit.
	VerdictRejectLocalOwned
	// VerdictRejectStaleEpoch means the output predates the current remote
	// assignment — typically a worker that died and was replaced.
	VerdictRejectStaleEpoch
	// VerdictRejectStaleVersion means the output is behind what the local side
	// already computed authoritatively while it owned the unit.
	VerdictRejectStaleVersion
)

func (RemoteVerdict) Accepted 

func (parseVerdict RemoteVerdict) Accepted() bool

Accepted reports whether a verdict means "use this output".

Callers should branch on this rather than comparing against VerdictAccept, so a future rejection reason cannot silently become an acceptance.

func (RemoteVerdict) String 

func (parseVerdict RemoteVerdict) String() string

type ReplaceOutcome 

type ReplaceOutcome struct {
	// Reassigned lists units moved onto a healthy worker.
	Reassigned []string
	// FellBack lists units that had no healthy worker to move to.
	FellBack []string
}

ReplaceOutcome reports what happened to each unit during a replacement.

type Tier 

type Tier string

Tier names one wire encoding, most preferred first. 

const (
	// TierBinary is a compact byte encoding. Preferred: smallest payload and no
	// JSON parse on the receiving side.
	TierBinary Tier = "binary"
	// TierStructuredClone is a JSON encoding carried by postMessage's own
	// structured clone. Universally available and the fallback for everything.
	TierStructuredClone Tier = "structured-clone"
	// TierShared is a SharedArrayBuffer page. Fastest where cross-origin
	// isolation is available, which most deployments do not have — which is why
	// v5 made it opt-in and never required (decision D5).
	TierShared Tier = "shared"
)

func RoundTrip 

func RoundTrip[T any](parsePayload T, parseCodec Codec[T], parseCapabilities Capabilities) (T, Tier, error)

RoundTrip encodes then decodes, returning the tier actually used.

Exists for conformance testing: a codec whose decode cannot read its own encode is broken in a way neither operation reveals alone.

func SelectTier 

func SelectTier(parseCapabilities Capabilities) (Tier, error)

SelectTier picks the best available tier.

Shared memory is NOT preferred automatically even when available. It requires cross-origin isolation, its ordering guarantees differ from message passing, and D5 made it opt-in — so a caller asks for it explicitly via SelectTierPreferring rather than silently receiving it.

func SelectTierPreferring 

func SelectTierPreferring(parseCapabilities Capabilities, parsePreferred Tier) (Tier, error)

SelectTierPreferring honors an explicit tier request, falling back through the normal order when it is unavailable.

type WorkerHealth 

type WorkerHealth uint8

WorkerHealth is a worker's dispatch-visible state. 

const (

	// HealthReady means the worker can accept work.
	HealthReady WorkerHealth
	// HealthDegraded means one heartbeat window was missed. Work is not routed to
	// it, but it is not yet replaced — a degraded worker usually recovers.
	HealthDegraded
	// HealthRestarting means the worker is being brought back deliberately.
	HealthRestarting
	// HealthDead means the worker is gone and must be replaced.
	HealthDead
)

func (WorkerHealth) String 

func (parseHealth WorkerHealth) String() string

type WorkerID 

type WorkerID string

WorkerID identifies one worker in the pool.

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.