Documentation
¶
Overview ¶
Package task is keelson's bus-protocol primitive for long-running, cancellable, observable work — the M1 surface of ADR-0038.
An app spawns a task via Spawn(parent, bus, opts). The returned HandleI is the producer-side API:
h, err := task.Spawn(ctx, bus, task.SpawnOpts{
Kind: "ch.export",
Title: "Export rows to parquet",
Cancellable: true,
EstimatedMs: 30_000,
})
for row := range rows {
if h.Cancelled() {
break
}
h.Report(task.ProgressReport{Current: n, Total: nTotal, Unit: task.UnitItems})
}
_ = h.Done(nil)
The handle owns:
- A context.Context derived from the caller's parent, cancelled when either parent cancels OR a task.<id>.cancel message arrives OR the handle reaches a terminal state (Done/Error).
- An emission gate: Report is throttled by the humanized form of the progress (e.g., "47% · 2m12s left") — a publish happens only when the visible string would change, plus a 1 Hz heartbeat for indeterminate-mode tasks.
- A cancel subscription on task.<id>.cancel that cancels the internal context.
Observers watch via WatchAll(bus, observer). The wire is canonical CBOR through runtime/buscodec; subjects are flat: task.<id>.<verb> with verbs created | progress | cancel | done | error.
See ADR-0038 (background task primitive) for the rationale and EXPLANATION.md for the cancellation semantics and the emission gate design. The optional supervisor lands in M3 as a separate package.
Index ¶
- Constants
- Variables
- func CancelerCaps() (caps []app.SubjectFilter)
- func MarshalInflightSnapshotReply(r InflightSnapshotReply) (b []byte, err error)
- func MarshalTaskCancel(c taskcancel.TaskCancel) (b []byte, err error)
- func MarshalTaskCreated(c taskcreated.TaskCreated) (b []byte, err error)
- func MarshalTaskDone(d taskdone.TaskDone) (b []byte, err error)
- func MarshalTaskError(e taskerror.TaskError) (b []byte, err error)
- func MarshalTaskProgress(p taskprogress.TaskProgress) (b []byte, err error)
- func ObserverCaps() (caps []app.SubjectFilter)
- func ProducerCaps() (caps []app.SubjectFilter)
- func RequestCancel(bus app.BusI, id TaskIdT, reason string) (err error)
- func SubjectCancel(id TaskIdT) (subject string)
- func SubjectCreated(id TaskIdT) (subject string)
- func SubjectDone(id TaskIdT) (subject string)
- func SubjectError(id TaskIdT) (subject string)
- func SubjectProgress(id TaskIdT) (subject string)
- func UnmarshalTaskCancel(b []byte) (c taskcancel.TaskCancel, err error)
- func UnmarshalTaskCreated(b []byte) (c taskcreated.TaskCreated, err error)
- func UnmarshalTaskDone(b []byte) (d taskdone.TaskDone, err error)
- func UnmarshalTaskError(b []byte) (e taskerror.TaskError, err error)
- func UnmarshalTaskProgress(b []byte) (p taskprogress.TaskProgress, err error)
- func WatchAll(bus app.BusI, obs ObserverI) (unsubscribe func(), err error)
- type ApiConfig
- type BusApi
- func (inst *BusApi) AppId() (id app.AppIdT)
- func (inst *BusApi) InstanceKey() (key uint64)
- func (inst *BusApi) ListInflight() (entries []InflightSnapshotEntry, err error)
- func (inst *BusApi) RequestCancel(id TaskIdT, reason string) (err error)
- func (inst *BusApi) RunId() (id string)
- func (inst *BusApi) Spawn(callerCtx context.Context, opts SpawnOpts) (h HandleI, err error)
- func (inst *BusApi) WatchAll(obs ObserverI) (unsubscribe func(), err error)
- type Handle
- func (inst *Handle) Cancelled() (b bool)
- func (inst *Handle) Ctx() (ctx context.Context)
- func (inst *Handle) Done(result []byte) (err error)
- func (inst *Handle) Error(taskErr error, reason string) (rerr error)
- func (inst *Handle) Id() (id TaskIdT)
- func (inst *Handle) Note(note string)
- func (inst *Handle) Report(p ProgressReport)
- type HandleI
- type InflightSnapshotEntry
- type InflightSnapshotReply
- type NoopTaskApi
- func (inst *NoopTaskApi) AppId() (id app.AppIdT)
- func (inst *NoopTaskApi) InstanceKey() (key uint64)
- func (inst *NoopTaskApi) ListInflight() (entries []InflightSnapshotEntry, err error)
- func (inst *NoopTaskApi) RequestCancel(id TaskIdT, _ string) (err error)
- func (inst *NoopTaskApi) RunId() (id string)
- func (inst *NoopTaskApi) Spawn(_ context.Context, opts SpawnOpts) (h HandleI, err error)
- func (inst *NoopTaskApi) WatchAll(_ ObserverI) (unsubscribe func(), err error)
- type ObserverI
- type ProgressReport
- type SpawnOpts
- type TaskApiI
- type TaskIdT
- type UnitE
Constants ¶
const ( SubjectPrefix = "task." VerbCreated = "created" VerbProgress = "progress" VerbCancel = "cancel" VerbDone = "done" VerbError = "error" // PatternAll matches every task verb across every task id. Used by // WatchAll observers and by the M3 supervisor. PatternAll = "task.>" // PatternCancelAll matches the cancel verb across every task id. Used // by callers that want to issue cancels but never produce or observe // progress. PatternCancelAll = "task.*.cancel" // SubjectListInflight is the request/reply subject the M3 supervisor // serves. A consumer publishes a Request on this subject (empty // payload) and receives a marshalled InflightSnapshotReply on its // reply inbox. The constant lives on the task package — not on the // supervisor — so consumers can query the snapshot without taking // a build-time dependency on the supervisor package. SubjectListInflight = "task.list.inflight" )
Subject taxonomy (ADR-0038). Flat per-task layout:
task.<id>.<verb>
Producers publish created/progress/done/error; consumers publish cancel; observers subscribe task.> with one wildcard.
const DefaultIndeterminateHeartbeatMs int64 = 1_000
DefaultIndeterminateHeartbeatMs is the minimum emit cadence for indeterminate-mode tasks (Total=0). A no-change Report call still publishes at this cadence so observers know the task is alive.
const DefaultListTimeoutMs int64 = 2_000
DefaultListTimeoutMs bounds ListInflight's bus.Request wait. Chosen to comfortably exceed the supervisor's snapshot-build latency for any realistic in-flight count while still failing fast when no supervisor is wired.
Variables ¶
var AllUnits = []UnitE{ UnitItems, UnitBytes, UnitSteps, }
var PackageProps = packageprops.Props{ WASMWASI: packageprops.WASMBlocked, WASMJS: packageprops.WASMBlocked, WASMFreestanding: packageprops.WASMBlocked, }
PackageProps records this package's curated properties (ADR-0080). Seeded by `wasmsurvey props generate`; curate by hand, then `wasmsurvey props verify`.
Functions ¶
func CancelerCaps ¶
func CancelerCaps() (caps []app.SubjectFilter)
CancelerCaps returns the SubjectFilter set a consumer needs to request cancellation of any task — narrower than ObserverCaps for apps that issue cancels but never read progress.
func MarshalInflightSnapshotReply ¶
func MarshalInflightSnapshotReply(r InflightSnapshotReply) (b []byte, err error)
MarshalInflightSnapshotReply serialises a reply via the canonical bus codec. The codec wire form (inflightsnapshotreply.InflightSnapshotReply) flattens Entries into parallel `[]T` columns — one column per entry-field. This helper does the fan-out so callers keep using the broker's native shape.
func MarshalTaskCancel ¶
func MarshalTaskCancel(c taskcancel.TaskCancel) (b []byte, err error)
MarshalTaskCancel serialises a taskcancel.TaskCancel. The payload type lives in keelson/runtime/codec/taskcancel; this helper keeps the legacy task.MarshalTaskCancel entry point so existing callers remain valid after the type move.
func MarshalTaskCreated ¶
func MarshalTaskCreated(c taskcreated.TaskCreated) (b []byte, err error)
MarshalTaskCreated serialises a taskcreated.TaskCreated. The payload type lives in keelson/runtime/codec/taskcreated; this helper keeps the legacy task.MarshalTaskCreated entry point so existing callers remain valid after the type move.
func MarshalTaskDone ¶
MarshalTaskDone serialises a TaskDone.
func MarshalTaskError ¶
MarshalTaskError serialises a taskerror.TaskError. The payload type lives in keelson/runtime/codec/taskerror; this helper keeps the legacy task.MarshalTaskError entry point so existing callers remain valid after the type move.
func MarshalTaskProgress ¶
func MarshalTaskProgress(p taskprogress.TaskProgress) (b []byte, err error)
MarshalTaskProgress serialises a taskprogress.TaskProgress. The payload type lives in keelson/runtime/codec/taskprogress (the first broker DTO migrated onto the ADR-0042 leeway codec); this helper keeps the legacy task.MarshalTaskProgress entry point so existing callers remain valid after the type move.
func ObserverCaps ¶
func ObserverCaps() (caps []app.SubjectFilter)
ObserverCaps returns the SubjectFilter set a passive observer needs to watch every task on the bus (e.g., a status panel or audit supervisor).
func ProducerCaps ¶
func ProducerCaps() (caps []app.SubjectFilter)
ProducerCaps returns the SubjectFilter set an app needs to spawn tasks. task.> with Both direction covers publishing created/progress/done/error and subscribing to the per-task cancel inbox. Add to Manifest.Caps.
func RequestCancel ¶
RequestCancel publishes a TaskCancel on task.<id>.cancel. Used by UI cancel buttons and by supervisor abandoned-task cleanup. Returns the underlying publish error verbatim; callers that lack the CancelerCaps receive ErrPermissionViolation.
func SubjectCancel ¶
SubjectCancel returns the subject a consumer publishes to request cancellation. The handle subscribes to this on Spawn.
func SubjectCreated ¶
SubjectCreated returns the subject a producer publishes once per task on spawn.
func SubjectDone ¶
SubjectDone returns the terminal-success subject. Emitted once at most.
func SubjectError ¶
SubjectError returns the terminal-failure subject. Emitted once at most.
func SubjectProgress ¶
SubjectProgress returns the subject for periodic progress emissions.
func UnmarshalTaskCancel ¶
func UnmarshalTaskCancel(b []byte) (c taskcancel.TaskCancel, err error)
UnmarshalTaskCancel is the inverse of MarshalTaskCancel. Empty payload yields a zero TaskCancel without error — apps that publish "just cancel the task" with nil payload remain interoperable.
func UnmarshalTaskCreated ¶
func UnmarshalTaskCreated(b []byte) (c taskcreated.TaskCreated, err error)
UnmarshalTaskCreated is the inverse of MarshalTaskCreated.
func UnmarshalTaskDone ¶
UnmarshalTaskDone is the inverse of MarshalTaskDone.
func UnmarshalTaskError ¶
UnmarshalTaskError is the inverse of MarshalTaskError.
func UnmarshalTaskProgress ¶
func UnmarshalTaskProgress(b []byte) (p taskprogress.TaskProgress, err error)
UnmarshalTaskProgress is the inverse of MarshalTaskProgress.
func WatchAll ¶
WatchAll subscribes to PatternAll on bus and demuxes by verb suffix into the observer's per-verb methods. Returns an unsubscribe function; callers MUST defer it (or call it on Mount/Unmount) to avoid a subscription leak when the observer goes out of scope.
Decode errors are swallowed silently — an observer that wants visibility into malformed payloads should wrap WatchAll with its own bus subscription. The current consumers (UI status panel, M3 supervisor) treat undecodable frames as bus noise and do not need a recovery path.
Types ¶
type ApiConfig ¶
type ApiConfig struct {
// Bus is the host's app.BusI. nil substitutes a NoopBus that
// errors on every call.
Bus app.BusI
// AppId is the owner identity injected into SpawnOpts.OwnerAppId.
AppId app.AppIdT
// InstanceKey is the host-minted per-window instance id injected
// into SpawnOpts.OwnerTileKey.
InstanceKey uint64
// RunId is the process-wide run id injected into
// SpawnOpts.OwnerRunId.
RunId string
// Logger is the base producer-side logger. The API adds task_id
// before passing to each handle. Zero value writes nowhere.
Logger zerolog.Logger
// MountCancel is the app's mount-cancel channel. When non-nil, the
// API composes it into every spawned task's parent context so
// window close cascades into worker cancellation.
MountCancel <-chan struct{}
// ListTimeoutMs caps the bus.Request timeout for ListInflight.
// Defaults to DefaultListTimeoutMs (2000ms) when zero.
ListTimeoutMs int64
}
ApiConfig is the construction parameter set for NewBusApi. All fields are optional; the resulting API is usable with any subset (an empty config produces an API that delegates to a NoopBus and stamps zero values into every TaskCreated — fine for early tests).
type BusApi ¶
type BusApi struct {
// contains filtered or unexported fields
}
BusApi is the concrete TaskApiI handed out by the host. Exported so hosts can hold a typed reference for diagnostic helpers; consumers program against TaskApiI.
func NewBusApi ¶
NewBusApi constructs a TaskApiI bound to the supplied identity. Defensively substitutes a NoopBus when cfg.Bus is nil so callers can safely invoke Spawn without panicking — every operation returns the NoopBus error.
func (*BusApi) InstanceKey ¶
func (*BusApi) ListInflight ¶
func (inst *BusApi) ListInflight() (entries []InflightSnapshotEntry, err error)
ListInflight performs the supervisor request/reply. The bus' Request timeout (configured on inprocbus.Inst, typically 5s) is the authoritative bound; ListTimeoutMs in ApiConfig is reserved for a future API revision that uses a per-call context (the current app.BusI does not accept one).
func (*BusApi) RequestCancel ¶
RequestCancel publishes a cancel for the named task. Uses the host bus; the API itself does not own a cap check (the bus client does).
type Handle ¶
type Handle struct {
// contains filtered or unexported fields
}
Handle is the concrete HandleI returned by Spawn. Exported because tests poke at internal state via package-private helpers; consumers should program against HandleI.
func (*Handle) Report ¶
func (inst *Handle) Report(p ProgressReport)
type HandleI ¶
type HandleI interface {
// Id returns the task identifier — same value as appears in subject
// paths task.<id>.<verb>.
Id() (id TaskIdT)
// Ctx returns a context.Context that cancels when:
// - the parent context passed to Spawn cancels;
// - a task.<id>.cancel message arrives on the bus;
// - the handle reaches a terminal state (Done/Error).
// Worker loops poll Ctx().Done() (or call Cancelled() as a
// convenience) and exit promptly.
Ctx() (ctx context.Context)
// Cancelled is a shorthand for Ctx().Err() != nil.
Cancelled() (b bool)
// Report submits a progress sample. Publication is gated by the
// estimator's humanized-change rule plus a 1 Hz heartbeat for
// indeterminate tasks; callers should not rate-limit upstream.
Report(p ProgressReport)
// Note submits a text-only progress update. Same emission gate as
// Report; useful for tasks whose progress is qualitative
// ("connecting…", "negotiating tls…").
Note(note string)
// Done publishes the terminal-success message with an opaque result.
// Pass nil result for tasks whose outcome is the side effect itself
// (a file was written, a row was inserted). Idempotent.
Done(result []byte) (err error)
// Error publishes the terminal-failure message. err is encoded via
// the boxer eh.MarshalError chain so errorview renders it directly;
// reason is a short human label that surfaces in observer lists.
// Idempotent.
Error(err error, reason string) (rerr error)
}
HandleI is the producer-side contract returned by Spawn. Implementations are safe for concurrent use — a worker goroutine calling Report and a UI goroutine calling Note is the canonical pattern.
Lifecycle: Created at Spawn → Report/Note may run any number of times → exactly one of Done/Error is called → handle is terminal. After Done/Error, further Report/Note/Done/Error calls return without publishing (idempotent), and Ctx().Done() is closed.
func Spawn ¶
Spawn registers a new task and returns a HandleI for the producer side. The supplied bus is used to:
- publish task.<id>.created once now;
- publish task.<id>.progress as the handle reports;
- publish task.<id>.done or task.<id>.error on terminal;
- subscribe to task.<id>.cancel for the lifetime of the handle.
The handle's Ctx() is derived from parent and cancels on parent-cancel, bus-cancel, or Done/Error.
type InflightSnapshotEntry ¶
type InflightSnapshotEntry struct {
Id TaskIdT `json:"id"`
Kind string `json:"kind"`
Title string `json:"title,omitempty"`
OwnerAppId app.AppIdT `json:"ownerAppId,omitempty"`
State string `json:"state"`
CreatedAtMs int64 `json:"createdAtMs"`
LastEmitMs int64 `json:"lastEmitMs"`
Current uint64 `json:"current,omitempty"`
Total uint64 `json:"total,omitempty"`
Unit string `json:"unit,omitempty"`
EtaMs int64 `json:"etaMs,omitempty"`
}
InflightSnapshotEntry is one row in the snapshot. Optional fields (Total, Unit, EtaMs, Current) are zero until the supervisor has observed at least one TaskProgress for the task. State is a plain string ("running" | "cancelling" | "abandoned") so the wire stays stable when the supervisor's internal enum evolves.
type InflightSnapshotReply ¶
type InflightSnapshotReply struct {
Entries []InflightSnapshotEntry `json:"entries"`
AtMs int64 `json:"atMs"`
}
InflightSnapshotReply is the wire payload an M3 supervisor publishes on a list-inflight reply inbox in response to a Request on SubjectListInflight. Entries reflects the supervisor's in-memory map at the moment AtMs was sampled; order is stable but unspecified.
Defined on the task package (not the supervisor) so consumers can decode the reply without importing the supervisor and acquiring its factsstore dependency.
func UnmarshalInflightSnapshotReply ¶
func UnmarshalInflightSnapshotReply(b []byte) (r InflightSnapshotReply, err error)
UnmarshalInflightSnapshotReply is the inverse of MarshalInflightSnapshotReply. Reconstructs `[]InflightSnapshotEntry` from the parallel `[]T` columns. The parallel-array contract assumes each column carries exactly the same N entries in slice order; MarshalInflightSnapshotReply guarantees this on the writer side and the leeway codec preserves slice ordering through the wire.
type NoopTaskApi ¶
type NoopTaskApi struct{}
NoopTaskApi is the fallback API for hosts that have not wired a real task surface yet (M1 bootstrap, isolated tests). Every operation returns a structured error naming the operation, matching the NoopBus / NoopStorage shape so apps detect the missing-host case cleanly.
func (*NoopTaskApi) AppId ¶
func (inst *NoopTaskApi) AppId() (id app.AppIdT)
func (*NoopTaskApi) InstanceKey ¶
func (inst *NoopTaskApi) InstanceKey() (key uint64)
func (*NoopTaskApi) ListInflight ¶
func (inst *NoopTaskApi) ListInflight() (entries []InflightSnapshotEntry, err error)
func (*NoopTaskApi) RequestCancel ¶
func (inst *NoopTaskApi) RequestCancel(id TaskIdT, _ string) (err error)
func (*NoopTaskApi) RunId ¶
func (inst *NoopTaskApi) RunId() (id string)
func (*NoopTaskApi) WatchAll ¶
func (inst *NoopTaskApi) WatchAll(_ ObserverI) (unsubscribe func(), err error)
type ObserverI ¶
type ObserverI interface {
OnCreated(c taskcreated.TaskCreated)
OnProgress(p taskprogress.TaskProgress)
OnDone(d taskdone.TaskDone)
OnError(e taskerror.TaskError)
OnCancel(c taskcancel.TaskCancel)
}
ObserverI is the consumer-side contract: a visitor that receives one callback per verb. WatchAll fans out a single task.> subscription into these per-verb calls. Implementations need not be goroutine-safe — the in-proc bus delivers messages on the publisher's goroutine, one at a time per subscription. M4 NATS-backed delivery uses one goroutine per subscription, so ordering across verbs of a single task is preserved even there.
type ProgressReport ¶
ProgressReport is the producer-side input to HandleI.Report. The estimator computes throughput + ETA from a sliding window of these. Total=0 marks an indeterminate task (count visible, end unknown).
type SpawnOpts ¶
type SpawnOpts struct {
// Id sets a deterministic task id. Empty means "generate one via
// nanoid".
Id TaskIdT
// Kind groups tasks for observer dispatch ("ch.export", "fs.scan",
// "kafka.catchup"). Required. Conventional strings, not a registered
// enum.
Kind string
// Title is the human-readable label observers display. Defaults to
// Kind when empty.
Title string
// OwnerAppId attributes the task to the spawning app. Carried in the
// TaskCreated payload for audit and display; not validated. Empty is
// allowed for runtime-side tasks that don't correspond to a user app.
// Auto-filled by task.ForApp(ctx) — direct callers of Spawn
// supply it themselves.
OwnerAppId app.AppIdT
// OwnerTileKey is the host-minted per-window instance id. Auto-filled
// by task.ForApp(ctx) so audit rows can join back to
// AppLifecycleRow.TileKey. Direct callers may leave it zero.
OwnerTileKey uint64
// OwnerRunId is the process-wide run id. Auto-filled by
// task.ForApp(ctx) so audit rows can join back to
// RuntimeStartRow.RunId. Direct callers may leave it empty.
OwnerRunId string
// Cancellable surfaces a hint to consumers (UI: should we show a
// cancel button?). The handle's cancel subscription is always
// active; this field controls observer affordances, not engine
// behaviour.
Cancellable bool
// EstimatedMs gives an initial duration guess for observers (e.g.,
// to show a progress bar before the first Report). Zero means
// "unknown".
EstimatedMs int64
// HeartbeatMs overrides DefaultIndeterminateHeartbeatMs for tasks
// whose total is unknown. Zero ⇒ default.
HeartbeatMs int64
// Logger is the producer-side diagnostic logger. task.ForApp(ctx)
// pre-contextualises it with run_id / app_id / instance_id; the
// handle adds task_id internally. nil ⇒ handle uses a no-op logger
// (zero-value zerolog.Logger writes nowhere). Pointer because
// zerolog.Logger contains a []byte and cannot be compared with ==.
Logger *zerolog.Logger
}
SpawnOpts configures a new task. All fields are optional except Kind, which is the schema-key observers dispatch on. Title defaults to Kind. Id is auto-generated as a nanoid when empty.
type TaskApiI ¶
type TaskApiI interface {
// Spawn creates a new task. callerCtx is composed with the host's
// mount-cancel channel so the returned handle's ctx cancels on
// either signal (or on terminal Done/Error). Pass
// context.Background() to opt out of caller-side scoping.
Spawn(callerCtx context.Context, opts SpawnOpts) (h HandleI, err error)
// WatchAll attaches an observer to task.> on the host bus.
// Identical to task.WatchAll(bus, obs) — exposed here so apps
// don't need to thread the bus.
WatchAll(obs ObserverI) (unsubscribe func(), err error)
// RequestCancel publishes a cancel for the given task id.
RequestCancel(id TaskIdT, reason string) (err error)
// ListInflight queries the M3 supervisor's snapshot via the
// SubjectListInflight request/reply. Times out at ListTimeoutMs
// from ApiConfig. Returns an empty slice (not an error) when no
// supervisor is running and the bus' request timeout fires —
// callers distinguish "no supervisor" from "no in-flight tasks"
// by the err: a nil err with empty entries means the supervisor
// returned an empty snapshot; an error means the supervisor was
// unreachable.
ListInflight() (entries []InflightSnapshotEntry, err error)
// AppId returns the owner identity the API will stamp on spawned
// tasks. Useful for diagnostic logging at the app boundary.
AppId() (id app.AppIdT)
// InstanceKey returns the host-minted per-window instance id. Zero
// when the API was constructed without one (tests, standalone CLI).
InstanceKey() (key uint64)
// RunId returns the process-wide run id. Empty when the API was
// constructed without one.
RunId() (id string)
}
TaskApiI is the high-level, identity-aware surface task.ForApp builds from a MountContextI. The interface auto-injects OwnerAppId, OwnerTileKey, and OwnerRunId from the host into every spawned task, composes the caller's context with the app's mount-cancel channel so tasks auto-terminate on window close, and tags the producer-side logger with task_id (on top of the app's logger context which already carries run_id / app_id / instance_id).
Apps inside the keelson runtime use this surface; library code or runtime services that operate outside the AppI lifecycle call the bare task.Spawn / task.WatchAll / task.RequestCancel functions directly and supply their own identity.
func ForApp ¶
func ForApp(ctx app.MountContextI) (api TaskApiI)
ForApp is the idiomatic constructor for app code: pulls identity, bus, logger, mount-cancel from a MountContextI to produce a fully wired TaskApiI. Apps capture this in Mount and reuse across Frame passes:
func (inst *App) Mount(ctx app.MountContextI) (err error) {
inst.tasks = task.ForApp(ctx)
return
}
The TaskApiI is independent of the MountContextI — it captures the pieces it needs by value, so it remains usable after the MountContextI itself goes out of scope (within the goroutine that holds the reference). The MountCancel channel is the one host-level signal the API will observe to cascade-cancel running tasks on window close.
Hosts whose MountContextI carries a NoopBus return an API whose every operation surfaces the NoopBus error. This means callers don't need to defensively branch on "do I have a real bus"; they can call Spawn unconditionally and propagate any returned err.
type TaskIdT ¶
type TaskIdT string
TaskIdT is a per-task identifier. Generated as a nanoid by Spawn when SpawnOpts.Id is empty; callers may also supply a deterministic id (test fixtures, replay tools). Carried verbatim in subject paths as one NATS token — the nanoid default alphabet (A-Z, a-z, 0-9, _, -) is NATS-token-safe.
func ParseSubject ¶
ParseSubject splits a task.<id>.<verb> subject into its id and verb components. Returns ok=false for subjects that do not match the taxonomy. Used by WatchAll to demux a single task.> subscription into per-verb observer callbacks.
type UnitE ¶
type UnitE uint8
UnitE classifies the magnitude carried by ProgressReport.Current / .Total. The estimator uses it to humanize throughput ("18 MB/s" vs "240 items/s" vs "step 3 of 5").
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package estimator is the throughput + humanized-form computation that gates emission inside a task.Handle.
|
Package estimator is the throughput + humanized-form computation that gates emission inside a task.Handle. |
|
Package supervisor is the ADR-0038 M3 audit + heartbeat layer for the task primitive.
|
Package supervisor is the ADR-0038 M3 audit + heartbeat layer for the task primitive. |