Documentation
¶
Overview ¶
Package fleetnode implements the node side of fleet-dispatcher CONTRACT.md v2: health / ack-then-poll dispatch / job status, with measured VRAM footprints. This file is the job store — the ack-then-poll heart of the dispatch contract: a POST acks in milliseconds (202 accepted) and the render runs in a tracked goroutine; pollers read state until it turns terminal (done|error). Once acked, a job is OURS FOREVER (never re-dispatched), so shutdown marks survivors terminal error:"interrupted" rather than losing them — pollers always reach a terminal state.
Package fleetnode implements the node side of fleet-dispatcher CONTRACT.md v2: health / ack-then-poll dispatch / job status, with measured VRAM footprints. See docs/superpowers/specs/2026-07-17-fleet-node-server-design.md.
vram.go is the two-source VRAM sampler's portable half: the nvidia-smi global snapshot (feeds /fleet/health every 2s) and the pure helper ParseSmiMemory. The Windows-only PDH per-process source lives in vram_windows.go.
Index ¶
- func BuildRequest(cfg config.Config, taskType string, payload json.RawMessage) (core.Request, func(), error)
- func Families(cfg config.Config) []string
- func ParseSmiMemory(out string) (totalGiB, usedGiB float64, err error)
- func SupportedTasks(cfg config.Config) []string
- func TreeDedicatedGiB(rootPid int) (float64, error)
- type FootprintEntry
- type Footprints
- type JobState
- type JobView
- type Jobs
- type Options
- type Runner
- type Sampler
- type Server
- type Snapshot
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func BuildRequest ¶
func BuildRequest(cfg config.Config, taskType string, payload json.RawMessage) (core.Request, func(), error)
BuildRequest translates a fleet dispatch (task_type + opaque payload) into the core.Request the pipeline runs, mirroring the MCP handlers' param mapping exactly. cleanup is ALWAYS non-nil (defer it unconditionally); for run-graph it removes the materialized temp files. Errors are caller mistakes (the server's 400s): unknown/unconfigured task_type (listing this box's supported set), malformed payload JSON, or a missing/invalid required field.
func Families ¶
Families returns the loadable model families for the advertised tasks, deduplicated, in task order. nil when nothing is bound.
func ParseSmiMemory ¶
ParseSmiMemory parses `nvidia-smi --query-gpu=memory.total,memory.used --format=csv,noheader,nounits` output ("16384, 1234", MiB) into GiB values. Whitespace and CRLF are tolerated (nvidia-smi emits \r\n on Windows); on a multi-GPU box the first line (GPU 0) wins. A total <= 0 or a negative used is an error: the contract treats vram_total_gb <= 0 as a failed probe, so a zero-total parse must never become a publishable snapshot.
func SupportedTasks ¶
SupportedTasks returns the fleet task_types this machine's config actually serves, in stable order. nil when nothing is bound.
func TreeDedicatedGiB ¶
TreeDedicatedGiB is the non-Windows stub for the PDH per-process tree sampler (vram_windows.go): the "\GPU Process Memory" counter set is a WDDM facility, so off-Windows it always errors and callers fall back to the nvidia-smi global-delta source. Present so portable callers (the pipeline's sampler composition) compile everywhere; sampler mode "auto" never selects the tree source off-Windows anyway.
Types ¶
type FootprintEntry ¶
type FootprintEntry struct {
ModelFamily string `json:"model_family"`
Quant string `json:"quant,omitempty"`
TaskType string `json:"task_type,omitempty"`
VramPeakGiB float64 `json:"vram_peak_gb"`
}
FootprintEntry is the wire shape /fleet/health advertises (contract v2).
type Footprints ¶
type Footprints struct {
// contains filtered or unexported fields
}
Footprints is a concurrency-safe, file-backed footprint store. Every Record persists synchronously via an atomic temp+rename write; a corrupt or missing file opens as empty (logged once, never a crash) so a bad disk state can only cost history, never availability.
func OpenFootprints ¶
func OpenFootprints(path string) *Footprints
OpenFootprints loads the store at path (missing → empty; corrupt → empty with one log line).
func (*Footprints) Entries ¶
func (f *Footprints) Entries() []FootprintEntry
Entries returns the wire-shaped entries with vram_peak_gb > 0, sorted by (family, quant, task) for stable health output.
func (*Footprints) Path ¶
func (f *Footprints) Path() string
Path returns the store's backing file path (fleet-measure prints the on-disk records — vram_peak_gb plus the raw observed_peak_gb/samples state the Afterburner validation procedure compares against).
func (*Footprints) Record ¶
func (f *Footprints) Record(family, quant, task string, observedGiB float64)
Record folds one observed per-render peak (GiB) into the (family, quant, task) entry: max-keep on the observation, vram_peak_gb = max x 1.2 rounded 0.1, samples++, then persist. Non-positive observations are dropped — a zero/negative "peak" is a sampling failure, and the contract forbids advertising vram_peak_gb <= 0.
func (*Footprints) ReloadIfChanged ¶
func (f *Footprints) ReloadIfChanged()
ReloadIfChanged re-reads the backing file when its mtime moved past the last load, MAX-MERGING disk records into memory (higher observed peak wins per key) so a record written by ANOTHER process — fleet-measure priming the store while fleet-serve is up was the live-found case — becomes visible to this process's Entries() without a restart, and an in-memory observation is never regressed by an older on-disk one. Cheap (one stat) when nothing changed; safe to call from the health accessor (no locks beyond the store's own, no spawns).
type JobState ¶
type JobState string
JobState is the contract's job lifecycle: accepted → running → done|error.
type JobView ¶
type JobView struct {
ID string
State JobState
Data json.RawMessage
Error string
}
JobView is a read-only copy of a job's externally visible state (the wire's `state` / `data` / `error` fields are shaped by the server layer).
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs is the concurrency-safe ack-then-poll job store. Terminal results are retained for ttl (contract: ≥ a few minutes; we run 1h) and swept by a janitor goroutine. DrainAndStop is the shutdown path.
func NewJobs ¶
NewJobs builds a store whose terminal entries live for ttl, with the janitor sweeping every 5 minutes (the spec's cadence).
func (*Jobs) Accept ¶
Accept registers id and starts run in a tracked goroutine, returning immediately (the ack). Idempotent: an id already present — in ANY state — returns false and never starts a second run (the contract's duplicate-POST rule). During drain it refuses all new work (false; the server maps that to 503 via Draining()).
func (*Jobs) DrainAndStop ¶
DrainAndStop stops accepting, waits up to timeout for in-flight runs, marks every non-terminal survivor error:"interrupted" — the contract's shutdown obligation (once acked, a job is never re-dispatched, so pollers must still reach a terminal state) — then cancels their context and waits (bounded by killDeliveryGrace) for the released run goroutines to actually return, so the cancel-triggered kill of child process trees lands before the process exits. The mark happens BEFORE the cancel so a run released by the cancel can never race its late completion past the mark (finish never overwrites a terminal state). Total bound: timeout + killDeliveryGrace. Also stops the janitor. Safe to call once per store.
func (*Jobs) QueueDepth ¶
QueueDepth counts accepted+running jobs (the health field). Terminal entries are results awaiting pollers, not load.
type Options ¶
type Options struct {
NodeID string
Snapshot func() (Snapshot, bool)
Footprints func() []FootprintEntry
GpuVendor string
GpuArch string
Cfg config.Config
}
Options is the server's static identity + read-only data feeds. Snapshot and Footprints are functions (not values) so health always reads live state without the handler owning any sampling machinery.
type Runner ¶
Runner runs one translated request to completion. The real pipeline satisfies it; tests inject fakes.
type Sampler ¶
type Sampler struct {
// contains filtered or unexported fields
}
Sampler publishes the latest good Snapshot via an atomic.Value so the health handler never blocks on (or spawns) nvidia-smi.
func StartGlobalSampler ¶
func StartGlobalSampler(ctx context.Context, interval time.Duration, run func() (string, error)) *Sampler
StartGlobalSampler samples run() (an injected nvidia-smi invocation) once synchronously — so a successful probe is Load-able the moment this returns — then keeps refreshing every interval until ctx is done. Runner or parse failures leave the previous snapshot in place: the sampler degrades to stale data, never to zeros. Staleness is bounded downstream — the health handler refuses (503) any snapshot older than maxSnapshotAge, so a persistently failing nvidia-smi (driver reset) cannot serve hours-stale 200s.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is the fleet-node HTTP server: three handlers over a Runner + Jobs store. Task/family advertisements are derived once at construction (config is immutable while serving).
func New ¶
New builds a Server. The supported-task/family lists are computed here, not per health request — the config cannot change under a running server.
Source Files