Documentation
¶
Overview ¶
Package domain is v5's off-thread command runtime (plan item P3.4).
It executes commands away from the render thread with three guarantees the plan names as its acceptance criteria: an atomic command that is replayed produces no duplicate effects, a bulk command that dies partway resumes and finishes exactly once, and a cancelled command does neither.
All three are ordering problems, which is why this sits directly on the services extracted in P3.3 rather than reinventing them: the ledger decides replay, the dispatcher's generation model decides cancellation.
The package is deliberately free of syscall/js so the guarantees can be tested natively. Nothing here knows it will run in a worker.
Index ¶
- Variables
- type BulkCommand
- type BulkResult
- type Checkpoint
- type CheckpointStore
- type CommandID
- type MemoryCheckpointStore
- func (parseStore *MemoryCheckpointStore) Clear(parseCommandID CommandID) error
- func (parseStore *MemoryCheckpointStore) Load(parseCommandID CommandID) (Checkpoint, bool, error)
- func (parseStore *MemoryCheckpointStore) Save(parseCheckpoint Checkpoint) error
- func (parseStore *MemoryCheckpointStore) Snapshot() map[CommandID]Checkpoint
- type MemoryTransactionalCheckpointStore
- type Outcome
- type Runtime
- func (parseRuntime *Runtime) Cancel(parseCommandID CommandID) error
- func (parseRuntime *Runtime) ExactlyOnce() bool
- func (parseRuntime *Runtime) Execute(parseCommandID CommandID, parseEffect func() error) (Outcome, error)
- func (parseRuntime *Runtime) ExecuteBulk(parseCommand BulkCommand, parseStep func(parseIndex int) error) (BulkResult, error)
- func (parseRuntime *Runtime) ExportState() RuntimeState
- func (parseRuntime *Runtime) Forget(parseCommandID CommandID) error
- func (parseRuntime *Runtime) IsCancelled(parseCommandID CommandID) bool
- func (parseRuntime *Runtime) SetYield(parseYield func())
- func (parseRuntime *Runtime) TrackedCommands() int
- type RuntimeState
- type TransactionalCheckpointStore
Constants ¶
This section is empty.
Variables ¶
var ErrAlreadyApplied = errors.New("domain: command already applied")
ErrAlreadyApplied reports that a command was already recorded as applied, so its effect did not run again.
A sentinel rather than a boolean return, because an implementation that discovers this mid-transaction usually has an error value in hand and nothing useful to say in a second return.
var ErrCommandCancelled = errors.New("domain: command was cancelled")
ErrCommandCancelled is returned when work stops because of a cancellation.
Functions ¶
This section is empty.
Types ¶
type BulkCommand ¶
type BulkCommand struct {
// ID is the replay and resume key. It must be the same across the original
// run and every resumption of it.
ID CommandID
// Total is the number of items.
Total int
// YieldEvery is how many items run between turns given back to the host
// event loop.
//
// Zero never yields, which is correct for a runtime with a thread to itself
// and wrong for the case bulk commands are actually for. In a single-threaded
// worker message loop, a run that never yields cannot be cancelled by a
// message: the cancel is queued behind the run it is meant to stop. The
// per-item flag check below is necessary but not sufficient, because with no
// yield nothing can ever set that flag.
//
// Setting it requires Runtime.SetYield; ExecuteBulk refuses a command that
// asks to yield with no yielder installed, rather than running to completion
// while looking cancellable.
//
// Separate from CheckpointEvery on purpose. Checkpoints default to every
// item, and a round trip through the event loop per item would cost far more
// than the work.
YieldEvery int
// CheckpointEvery is how many items complete between durable progress writes.
// Zero or one checkpoints every item, which is the only interval that gives
// exactly-once against a non-transactional store — see CheckpointStore.
// Larger intervals trade a bounded window of re-executed items for fewer
// writes, and the window is reported rather than hidden.
CheckpointEvery int
}
BulkCommand describes work over an indexed range of items.
The shape is deliberately narrow — an item count and a step function — because the resume guarantee depends on items being addressable by a stable index. A bulk command over a stream whose order can change between runs cannot be resumed correctly by any bookkeeping, and this makes that constraint explicit rather than implied.
type BulkResult ¶
type BulkResult struct {
Total int
// ResumedFrom is the index this run started at. Non-zero means it resumed.
ResumedFrom int
// Processed counts step invocations in this run.
Processed int
// MaxReplayWindow is the largest number of items this run MAY have
// re-executed.
//
// Deliberately a bound rather than a count. A run that ends gracefully — a
// step returning an error — records exactly how far it got, so nothing is
// re-executed and this is zero. A run killed outright records nothing after
// its last checkpoint, and no later run can discover how far it actually got:
// that information died with it. The bound is the checkpoint interval, and
// reporting the bound is honest where reporting a count would be invented.
//
// A caller whose steps are not idempotent needs this to be zero, which means
// either a transactional CheckpointStore or accepting the window.
MaxReplayWindow int
// NextIndex is the first item not known to be complete after this run.
NextIndex int
// Completed reports whether every item is done.
Completed bool
// Outcome describes the run in the same vocabulary as an atomic command.
Outcome Outcome
}
BulkResult describes one run of a bulk command.
"One run" is the important qualifier: after a crash and a resume there are two results, and Processed is per-run. Completed and NextIndex are the fields that describe the command as a whole.
type Checkpoint ¶
type Checkpoint struct {
CommandID CommandID
// NextIndex is the first item not yet known to be complete. Resuming starts
// here, so it is deliberately conservative: an item may be re-executed, but
// none is ever skipped.
NextIndex int
// Total is the item count the checkpoint was written against. A resume
// against a different total is refused rather than guessed at.
Total int
// Exact reports that NextIndex is precisely where the previous run stopped,
// rather than the last interval boundary it happened to cross.
//
// It is true when a run ended gracefully — a step returned an error — because
// the runtime then knows exactly what completed and writes it. It is false
// for interval checkpoints, where a run killed outright may have progressed
// further than the record shows. The distinction is the difference between
// resuming with no re-execution and resuming with a bounded window of it, so
// it is recorded rather than assumed either way.
Exact bool
}
Checkpoint is a bulk command's durable progress marker.
type CheckpointStore ¶
type CheckpointStore interface {
Load(parseCommandID CommandID) (Checkpoint, bool, error)
Save(parseCheckpoint Checkpoint) error
Clear(parseCommandID CommandID) error
}
CheckpointStore persists bulk progress across a crash.
This interface is the durability seam, and the strength of the exactly-once guarantee depends entirely on what implements it:
- Backed by the SAME transactional store as the command's effects (P3.5's SQLite), progress and effect commit together and exactly-once holds for any checkpoint interval.
- Backed by anything else, progress and effects can diverge by up to one checkpoint interval, so items between the last checkpoint and the crash are re-executed on resume. The runtime reports that count rather than hiding it; see BulkResult.ReplayedItems.
§11-Q7 — whether WAL-mode crash atomicity holds over OPFS sync access handles — is exactly the question of whether the first bullet is available in the browser. It is unresolved, which is why the second bullet is a supported mode and not a degraded one.
type CommandID ¶
type CommandID string
CommandID identifies one logical command.
It is the replay key, so it must be stable across retries of the SAME logical command and distinct between different ones. A caller that generates a fresh ID per attempt gets no replay protection — which is the single most likely way to misuse this package, hence the emphasis.
type MemoryCheckpointStore ¶
type MemoryCheckpointStore struct {
// contains filtered or unexported fields
}
MemoryCheckpointStore is an in-process CheckpointStore.
It survives a command failure but not a process restart, which makes it the right store for tests and for bulk work whose cost is cheaper to redo than to persist. It is NOT the browser durability story; that is P3.6.
func NewMemoryCheckpointStore ¶
func NewMemoryCheckpointStore() *MemoryCheckpointStore
NewMemoryCheckpointStore creates an empty in-process checkpoint store.
func (*MemoryCheckpointStore) Clear ¶
func (parseStore *MemoryCheckpointStore) Clear(parseCommandID CommandID) error
Clear removes a command's checkpoint.
func (*MemoryCheckpointStore) Load ¶
func (parseStore *MemoryCheckpointStore) Load(parseCommandID CommandID) (Checkpoint, bool, error)
Load reports a command's checkpoint, if one exists.
func (*MemoryCheckpointStore) Save ¶
func (parseStore *MemoryCheckpointStore) Save(parseCheckpoint Checkpoint) error
Save writes a command's checkpoint.
func (*MemoryCheckpointStore) Snapshot ¶
func (parseStore *MemoryCheckpointStore) Snapshot() map[CommandID]Checkpoint
Snapshot copies every stored checkpoint, for diagnostics and tests.
type MemoryTransactionalCheckpointStore ¶
type MemoryTransactionalCheckpointStore struct {
*MemoryCheckpointStore
// contains filtered or unexported fields
}
MemoryTransactionalCheckpointStore is an in-process store that applies an effect and its record together.
It exists so the transactional path is testable and so an application can see what the interface asks for. It is NOT durable — a process restart loses everything — so it demonstrates atomicity, not persistence. A real implementation is the SQLite store writing the applied-record inside the same transaction as the effect's own writes.
func NewMemoryTransactionalCheckpointStore ¶
func NewMemoryTransactionalCheckpointStore() *MemoryTransactionalCheckpointStore
NewMemoryTransactionalCheckpointStore creates an in-process transactional store.
func (*MemoryTransactionalCheckpointStore) AppliedCount ¶
func (parseStore *MemoryTransactionalCheckpointStore) AppliedCount() int
AppliedCount reports how many distinct commands are recorded, so a test can assert the record survived independently of the in-memory ledger.
func (*MemoryTransactionalCheckpointStore) ApplyExactlyOnce ¶
func (parseStore *MemoryTransactionalCheckpointStore) ApplyExactlyOnce(parseCommandID CommandID, parseEffect func() error) error
ApplyExactlyOnce runs the effect and records the command under one lock.
The lock is this store's stand-in for a transaction: nothing else can observe the state between the effect running and the record being written, which is the property a database transaction provides and the reason the window this closes exists at all.
type Outcome ¶
type Outcome uint8
Outcome is what happened to a command.
const ( // OutcomeApplied means the command's effects ran, for the first and only time. OutcomeApplied Outcome = iota // OutcomeReplayed means the command had already been applied, so its effects // were deliberately NOT run again. This is a success, not a failure. OutcomeReplayed // OutcomeCancelled means the command was cancelled before its effects ran. OutcomeCancelled // OutcomeFailed means the effects ran and returned an error. The command is // not recorded as applied, so a retry with the same id will run again. OutcomeFailed )
type Runtime ¶
type Runtime struct {
// contains filtered or unexported fields
}
Runtime executes commands with replay, resume, and cancellation guarantees.
Not safe for concurrent use. It is designed for one worker's message loop, which is how it will run; the mutex in MemoryCheckpointStore protects the store alone, not this.
func NewRuntime ¶
func NewRuntime(parseCheckpoints CheckpointStore) *Runtime
NewRuntime creates a command runtime over a checkpoint store.
A nil store is allowed and disables bulk resume: bulk commands still run and still refuse to duplicate their effects, they simply restart from zero after a failure. That is a legitimate configuration for cheap work, so it is a documented mode rather than an error at construction.
func RestoreRuntime ¶
func RestoreRuntime(parseCheckpoints CheckpointStore, parseState RuntimeState) (*Runtime, error)
RestoreRuntime rebuilds a command runtime from exported state.
The checkpoint store is supplied separately because it is durable and was never part of the export.
func (*Runtime) Cancel ¶
Cancel stops a command from running, resuming, or being retried.
Cancellation is permanent for the id, and it clears any bulk progress. Both follow from what a cancel means: the work is not wanted. Leaving the checkpoint would let a later call with the same id silently resume abandoned work, which is criterion (c) of P3.4 — a cancelled command neither replays nor resumes.
Cancelling a command that already applied does not undo it. Undo is a compensating command, not a cancellation, and conflating the two would promise a rollback this cannot deliver.
func (*Runtime) ExactlyOnce ¶
ExactlyOnce reports whether this runtime can guarantee exactly-once execution across a crash.
False does not mean commands are unsafe — replay protection still holds for every failure short of a crash in one specific window. It means the window between an effect succeeding and its record being written is open, and an application whose effects are not idempotent should either supply a transactional store or accept that bound. Reporting it is what lets that be a decision rather than a surprise.
func (*Runtime) Execute ¶
func (parseRuntime *Runtime) Execute(parseCommandID CommandID, parseEffect func() error) (Outcome, error)
Execute runs an atomic command exactly once.
The check/commit split is what makes this safe under failure. The command is recorded as applied only AFTER its effects succeed, so a command that fails partway can be retried under the same id — where recording it up front would make the retry look like a replay and silently drop the work.
The converse is the cost: if the process dies between the effects succeeding and the record being written, the retry re-runs the effects. Reversing the order moves that window rather than removing it — a crash after the record and before the effect would drop the work instead, which is worse.
It closes only when the effect and the record commit TOGETHER. A store that implements TransactionalCheckpointStore can arrange that, and Execute uses it when present; see transactional.go. Runtime.ExactlyOnce reports which guarantee an application is actually running under.
func (*Runtime) ExecuteBulk ¶
func (parseRuntime *Runtime) ExecuteBulk(parseCommand BulkCommand, parseStep func(parseIndex int) error) (BulkResult, error)
ExecuteBulk runs a bulk command, resuming from any recorded progress.
The three P3.4 criteria all land here:
- Replay: a command already completed does no work and reports OutcomeReplayed.
- Resume: a command that died at item N restarts from its last checkpoint and finishes the remaining items exactly once.
- Cancellation: a cancelled command neither replays nor resumes, because Cancel clears its progress and its id stays refused.
A step that fails or panics stops the run, preserves progress, and returns the error. Progress is preserved deliberately: the failure is usually about one item, and discarding 30,000 completed items to re-do them is the wrong response to a bad row.
func (*Runtime) ExportState ¶
func (parseRuntime *Runtime) ExportState() RuntimeState
ExportState captures the replay state a reload must preserve.
func (*Runtime) Forget ¶
Forget drops all record of a command.
Replay detection requires remembering, so records accumulate for the life of the runtime. A caller that generates unbounded command ids — one per user action over a long session — should forget ids it will never legitimately retry. Forgetting an id makes a later command with that id run again.
func (*Runtime) IsCancelled ¶
IsCancelled reports whether a command has been cancelled.
func (*Runtime) SetYield ¶
func (parseRuntime *Runtime) SetYield(parseYield func())
SetYield installs the function ExecuteBulk uses to give the host event loop a turn between chunks.
This exists because of where bulk commands actually run. A domain worker is a single-threaded message loop: it receives a command, runs it, and only then takes the next message. A 50,000-item bulk run that never returns to that loop cannot be cancelled by a message, because the cancel message is sitting in a queue the worker will not read until the run it is meant to stop has finished. Checking a flag on every item does not help when nothing can set the flag.
The yielder is the environment's, not the domain's: in a wasm worker it is a setTimeout(0) round trip; in a test it can be a no-op or a channel receive. The domain layer must not know which.
parseRuntime.SetYield(func() {
parseDone := make(chan struct{})
var parseCallback js.Func
parseCallback = js.FuncOf(func(js.Value, []js.Value) any {
parseCallback.Release()
close(parseDone)
return nil
})
js.Global().Call("setTimeout", parseCallback, 0)
<-parseDone
})
func (*Runtime) TrackedCommands ¶
TrackedCommands reports how many commands hold replay state, so a caller can see the growth described on Forget rather than discover it as a leak.
type RuntimeState ¶
type RuntimeState struct {
// Applied lists commands already applied, so a retry after the reload is
// still recognized as a replay rather than run a second time.
//
// This is the guarantee that would silently break: a reloaded worker with a
// fresh ledger treats every in-flight retry as new work, so a command that
// succeeded just before the edit runs again just after it.
Applied []CommandID `json:"applied,omitempty"`
// Cancelled lists commands that must stay refused.
Cancelled []CommandID `json:"cancelled,omitempty"`
}
RuntimeState is a command runtime's replay state, extracted for transfer across a worker reload (plan item P3.14).
Only what a reload must not lose. Checkpoints are excluded on purpose: they already live in the CheckpointStore, which is the durable side and survives the reload by construction. Carrying them here would create a second copy that can disagree with the first.
type TransactionalCheckpointStore ¶
type TransactionalCheckpointStore interface {
CheckpointStore
ApplyExactlyOnce(parseCommandID CommandID, parseEffect func() error) error
}
TransactionalCheckpointStore commits a command's effect and its applied-record in one transaction.
The contract is the whole point and is stricter than it looks:
- If ApplyExactlyOnce returns nil, the effect ran and the command is recorded as applied. Both, or neither — a partial outcome is a violation.
- If it returns ErrAlreadyApplied, the effect did NOT run in this call and the command was already recorded. That is a replay, not a failure.
- If it returns any other error, the effect did not commit and the command is not recorded, so a retry under the same id is safe.
The effect must do its work through the transaction the store provides. A store cannot make an effect atomic if the effect writes somewhere else, and implementing this interface while allowing that would promise a guarantee the implementation cannot keep.
Source Files