Documentation
¶
Overview ¶
Package supervisor is the ADR-0038 M3 audit + heartbeat layer for the task primitive. Apps using the M1 / M2 surface need no supervisor; hosts that want a durable record of started / finished / cancelled / abandoned tasks instantiate one through New() and call Start().
What the supervisor does:
- Subscribes task.> via task.WatchAll and writes one factsstore LogRow per terminal verb (created / done / error / cancel) plus one row per abandoned-by-heartbeat detection. Progress events are observed but not persisted (too high-frequency for an audit trail).
- Maintains an in-memory snapshot of currently in-flight tasks and serves it on a request/reply subject (default "task.list.inflight"). UI status panels and supervisors of supervisors (M4 NATS cluster bridges) query this without reconstructing from history.
- Runs a heartbeat watchdog: an in-flight task whose last observed emission is older than HeartbeatThresholdMs is promoted to InflightStateAbandoned. The state surfaces in the snapshot and a final audit row is written; no synthetic bus event is emitted (the abandoned producer may yet recover).
The supervisor does NOT own bus client construction — the host passes an app.BusI permissioned according to Caps(). This keeps the supervisor unit-testable over any BusI implementation and matches the M1 / M2 idiom.
Index ¶
- Constants
- Variables
- func Caps() (caps []app.SubjectFilter)
- func RequesterCaps() (caps []app.SubjectFilter)
- type InflightEntry
- type InflightStateE
- type Opts
- type Supervisor
- func (inst *Supervisor) InflightSnapshot() (entries []InflightEntry)
- func (inst *Supervisor) OnCancel(c taskcancel.TaskCancel)
- func (inst *Supervisor) OnCreated(c taskcreated.TaskCreated)
- func (inst *Supervisor) OnDone(d taskdone.TaskDone)
- func (inst *Supervisor) OnError(e taskerror.TaskError)
- func (inst *Supervisor) OnProgress(p taskprogress.TaskProgress)
- func (inst *Supervisor) PersistedCount() (n uint64)
- func (inst *Supervisor) Start() (err error)
- func (inst *Supervisor) Stop() (err error)
Constants ¶
const AppId = "runtime.task.supervisor"
AppId is the conventional bus identity the host registers for the supervisor's bus client. Used in audit Sender fields.
const DefaultHeartbeatThresholdMs int64 = 30_000
DefaultHeartbeatThresholdMs is the no-emission gap after which an in-flight task is promoted to InflightStateAbandoned.
const DefaultHeartbeatTickMs int64 = 5_000
DefaultHeartbeatTickMs is the watchdog's scan cadence. Coarser than the threshold by ~6x so a barely-late producer doesn't get marked abandoned in the first window.
const LogService = "runtime.task.supervisor"
LogService is the Service tag the supervisor writes on every audited LogRow so factsstore queries can scope by service = LogService.
Variables ¶
var PackageProps = packageprops.Props{ WASMWASI: packageprops.WASMBlocked, WASMJS: packageprops.WASMBlocked, WASMFreestanding: packageprops.WASMBlocked, }
PackageProps records this package's curated properties (ADR-0080). Seeded by `boxer code analysis golang wasmsurvey props generate`; curate by hand. The same group's `props verify` reconciles it.
Functions ¶
func Caps ¶
func Caps() (caps []app.SubjectFilter)
Caps returns the SubjectFilter set the supervisor's bus client needs:
- task.> Sub — observe lifecycle for audit + inflight
- task.list.inflight Sub — receive list-inflight requests
- _INBOX.> Pub — reply on consumer inboxes
Same shape as persist/fsbroker — services need _INBOX.> publish to reply, and Sub on their own subject space.
func RequesterCaps ¶
func RequesterCaps() (caps []app.SubjectFilter)
RequesterCaps returns the SubjectFilter set a consumer needs to query the supervisor's in-flight snapshot via Request. The reply inbox subscribe is bypassed by the inprocbus client (see permission.go) so only the publish cap is required.
Types ¶
type InflightEntry ¶
type InflightEntry struct {
Created taskcreated.TaskCreated
Progress taskprogress.TaskProgress // zero value when no progress observed yet
State InflightStateE
LastEmitMs int64
}
InflightEntry is the Go-facing in-memory row exposed by InflightSnapshot. Carries the raw wire payloads so the caller can reach for any field the projected wire shape (InflightSnapshotEntry) omits.
type InflightStateE ¶
type InflightStateE uint8
InflightStateE classifies an in-flight row. Promoted to abandoned by the heartbeat watchdog when the producer goes silent past HeartbeatThresholdMs; promoted to cancelling on OnCancel before the producer's terminal verb arrives.
The wire shape (task.InflightSnapshotEntry.State) is a string — "running" | "cancelling" | "abandoned" — so consumers do not need to import this package to decode a snapshot reply.
const ( InflightStateUnspecified InflightStateE = 0 InflightStateRunning InflightStateE = 1 InflightStateCancelling InflightStateE = 2 InflightStateAbandoned InflightStateE = 3 )
func (InflightStateE) String ¶
func (inst InflightStateE) String() (s string)
type Opts ¶
type Opts struct {
HeartbeatThresholdMs int64
HeartbeatTickMs int64
ListSubject string
NowFn func() time.Time
}
Opts configures the supervisor at construction. All fields have sensible defaults; Opts{} is a valid zero value.
type Supervisor ¶
type Supervisor struct {
// contains filtered or unexported fields
}
Supervisor is the audit + watchdog hub. Construct with New, drive with Start / Stop; the embedded task.ObserverI methods are wired by Start via task.WatchAll and should not be invoked by callers directly (they are exported only because the interface is).
func New ¶
func New(bus app.BusI, facts factsstore.FactsStoreI, log zerolog.Logger, opts Opts) (inst *Supervisor)
New constructs a Supervisor against the provided bus + facts store. A nil facts store is permitted — the supervisor still maintains the in-flight map and serves list-inflight requests; audit rows are dropped. Useful for tests that want lifecycle visibility without asserting on persisted rows.
func (*Supervisor) InflightSnapshot ¶
func (inst *Supervisor) InflightSnapshot() (entries []InflightEntry)
InflightSnapshot returns a copy of the current in-flight map. Each entry's payloads are by value so the caller may retain them; the snapshot itself is not live (subsequent supervisor mutations are not reflected). Order is unspecified but stable per call.
func (*Supervisor) OnCancel ¶
func (inst *Supervisor) OnCancel(c taskcancel.TaskCancel)
func (*Supervisor) OnCreated ¶
func (inst *Supervisor) OnCreated(c taskcreated.TaskCreated)
func (*Supervisor) OnDone ¶
func (inst *Supervisor) OnDone(d taskdone.TaskDone)
func (*Supervisor) OnError ¶
func (inst *Supervisor) OnError(e taskerror.TaskError)
func (*Supervisor) OnProgress ¶
func (inst *Supervisor) OnProgress(p taskprogress.TaskProgress)
func (*Supervisor) PersistedCount ¶
func (inst *Supervisor) PersistedCount() (n uint64)
PersistedCount returns the number of audit rows written since construction. Includes abandoned-promotion rows. Atomic — safe to call from any goroutine.
func (*Supervisor) Start ¶
func (inst *Supervisor) Start() (err error)
Start subscribes to task.> and listSubject, then launches the heartbeat watchdog goroutine. Idempotent: a second call returns an error without restarting. Bus errors during subscribe roll back any partial state.
func (*Supervisor) Stop ¶
func (inst *Supervisor) Stop() (err error)
Stop tears down subscriptions and waits for the heartbeat goroutine to exit. Idempotent: calling on a non-started supervisor returns nil. Returns after the heartbeat goroutine has stopped so callers can rely on no further audit writes happening after Stop returns.
Source Files