internal/agent/reconcile/reconcile.go
Ref: Size: 44.0 KiB History
// Package reconcile implements the agent's level-triggered reconcile loop.
//
// Definitions (normative, from the spec):
//
// lost = boot ID changed OR process died without a recorded stop request;
// a deliberately stopped VM is stopped, NOT lost
//
// Shape: Step is a router, not a worker. It fences stale snapshots, hands each
// VM its slice of desired state to that VM's own long-lived goroutine, and
// aggregates every worker's last-published result into the report — never
// waiting on a worker, so slow VM work cannot delay the host heartbeat. One
// goroutine per VM is also the serialization primitive: a single VM's
// operations are serial by construction. See worker.go.
//
// Belt-and-suspenders: PrepareRootDisk and seed.Build both
// write via a temp file + rename, so a killed create can never leave a torn artifact.
package reconcile
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"slices"
"strings"
"sync"
"time"
"google.golang.org/protobuf/proto"
"github.com/a73x/eitri/internal/agent/seed"
"github.com/a73x/eitri/internal/agent/state"
"github.com/a73x/eitri/internal/pb"
"github.com/a73x/eitri/internal/version"
)
// Provisioner is one VM's whole lifecycle on this host, networking included.
// Implemented once per platform: cloudhv.Provisioner drives cloud-hypervisor on
// Linux, vfkit.Provisioner drives Apple's Virtualization.framework on macOS.
//
// Networking is part of this seam rather than beside it because a VM's network
// attachment is not separately lifecycled from the VM: cloud-hypervisor takes
// its tap as a launch argument and vfkit takes --device virtio-net, so on both
// backends "attach the NIC" is a step inside "start the VM". The verbs that
// would make it a seam of its own — create a tap, reserve an address — are host
// mechanism and do not generalize. What generalizes is the data: the
// deterministic MAC, the host's subnet, and a VM's current address.
type Provisioner interface {
// Preflight reports whether this backend can run a guest on this host at
// all.
Preflight(ctx context.Context) error
// PrepareRootDisk materialises the VM's root disk from a base image
// (clone + grow). It is root-disk-only by contract: reconcile's rebuild
// path calls create again, so routing a user volume through it would
// destroy data that is meant to outlive the VM.
PrepareRootDisk(ctx context.Context, spec state.VMSpec, basePath string) error
// Boot starts the VM and attaches it to the host network. The address the
// guest ends up with is the backend's business — state.MAC is the
// stickiness key on a backend that allocates addresses itself, and the host
// OS decides on one that does not — so Boot is told nothing about it and
// reports it through Address rather than returning it.
Boot(ctx context.Context, vmID string, spec state.VMSpec) error
Shutdown(ctx context.Context, vmID string) error
// Destroy stops the VM and releases every host resource it holds — its
// process and its network attachment, address reservation included.
Destroy(ctx context.Context, vmID string) error
Running(vmID string) bool
// FailureReason returns what the hypervisor said before it stopped running,
// or "" when the backend has nothing to add. It is asked when a guest that
// should be running is not, to give that report a cause: the process table
// can say a guest is gone but never why, and every backend already keeps its
// hypervisor's own output on disk.
FailureReason(vmID string) string
// Address returns the VM's current guest address, or "" when the backend
// does not know one (never booted, or gone). It is POLLED rather than
// returned by Boot: where the host OS's own DHCP server hands out the
// address, it does not exist until the guest has booted and asked for one,
// and Boot must not block inside the create slot waiting for that.
Address(vmID string) string
// NetworkAddress returns the address the VM's SECOND NIC — the one on the
// named host network its spec asked for — was granted, or "" when it has no
// such NIC or the backend has not learned one yet. A backend that serves no
// named networks answers "" always, which is every Mac and every Linux host
// with no --host-network flag.
//
// Separate from Address rather than replacing it because the two are
// different facts with different timings: Address is allocated by the host
// and known at boot, this one is granted by someone else's DHCP server and
// discovered afterwards.
NetworkAddress(vmID string) string
}
// Engine is the reconcile loop. All fields must be set before calling Step.
type Engine struct {
St *state.Store
Prov Provisioner
// Images resolves an image URL+sha256 to a local base-image path, fetching
// if necessary. Returns the path to the raw base image.
//
// progress is called as the download advances, and is the one thing on this
// seam that exists for the operator rather than for the VM: an image is
// gigabytes and a create is otherwise one word for the whole time it takes.
// An implementation that cannot say (a cache hit, a fetch collapsed onto
// another VM's) simply never calls it.
Images func(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error)
// Seed builds the cloud-init NoCloud seed ISO at outPath.
Seed func(outPath string, p seed.Params) error
// HostKey returns the guest host key stored at path, generating and
// persisting one when there is none. Both halves come back because both are
// needed here and neither travels together: the public line is reported
// upward for certification, the private PEM goes straight into the seed.
//
// It is load-or-create, not create: a VM waiting for its certificate must
// present the same key after an agent restart as before one.
HostKey func(path string) (state.HostKey, error)
// AgentVersion is this binary's stamped release, ordered against
// Snapshot.min_agent_version. "" or "dev" is unordered and never floored.
AgentVersion string
// BootID returns the current host boot identifier (e.g. /proc/sys/kernel/random/boot_id).
// Changes on reboot, enabling lost-VM detection.
BootID func() string
// Now returns the current time. Injectable for deterministic tests.
Now func() time.Time
// mu guards committed. It serializes admission so the concurrent per-VM
// workers cannot oversubscribe a cap or double-count. It is the one
// cross-VM invariant the worker layer relies on (see admit).
mu sync.Mutex
// committed is the in-memory admission ledger: vm_id -> the spec whose
// compute counts against the host caps. A VM is committed at create,
// released at quarantine (compute frees while the guest is stopped), and
// re-noted while it exists. Rebuilt from persisted records via SeedLedger.
committed map[string]state.VMSpec
// mgrOnce/mgr hold the per-VM worker manager, built on first use. It is not
// a constructor argument so that Step stays the single entry point and the
// agent needs no extra wiring beyond Stop at shutdown.
mgrOnce sync.Once
mgr *manager
// TombstoneGrace is the quarantine period for tombstoned VMs before destroy.
TombstoneGrace time.Duration
// VanishGrace is the quarantine period for VMs that vanished without a tombstone.
VanishGrace time.Duration
// MaxCreateAttempts is the maximum number of create attempts before terminal failed.
MaxCreateAttempts int
// MaxConcurrentCreates caps how many VMs on this host may be inside the
// I/O-heavy part of create at once (image fetch, disk materialisation).
// Per-VM workers made creates concurrent; this bounds
// how much of that concurrency reaches the disk. See acquireCreateSlot.
MaxConcurrentCreates int
// createSlotsOnce/createSlots hold the create throttle, built on first use
// so a zero-value Engine needs no constructor.
createSlotsOnce sync.Once
createSlots chan struct{}
// MaxVCPUs, MaxMemMB, MaxDiskGB cap the host resources this agent will
// commit to live VMs (0 = unlimited). This is the
// enforced half of agent-side quotas; syncclient advertises the same caps.
MaxVCPUs int64
MaxMemMB int64
MaxDiskGB int64
// VMTimeout bounds ONE VM's reconcile pass — every operation for that VM in
// that pass, not each operation and not the whole host tick. Convergence across ticks is guaranteed
// because completed image downloads are durably cached per-sha. Set
// comfortably above the longest legitimate operation (imagecache's HTTP
// client allows 10m for a first-time image download).
VMTimeout time.Duration
}
// assignment is one VM's slice of a desired-state snapshot. desired is nil when
// the VM is absent from desired state entirely (vanished); tombstoned marks a
// desired entry flagged for deletion. Both mean "reap", and the pair stays
// explicit because the two carry different grace periods (VanishGrace vs the
// shorter TombstoneGrace).
//
// LIFETIME: desired points into the caller's Snapshot, and a worker holds that
// pointer well past the Step that delivered it — for as long as its pass runs,
// up to VMTimeout. A snapshot handed to Step must therefore be treated as
// immutable while any pass may still be running.
type assignment struct {
desired *pb.VMSpec
tombstoned bool
}
// assignments slices a snapshot into one assignment per VM, over the UNION of
// desired state and local records: a desired-only id is a create, a record-only
// id is a vanished VM to reap, and an id in both converges. The union is also
// exactly the set of VMs that need a reconcile pass this tick.
func assignments(snap *pb.Snapshot, recs map[string]state.Record) map[string]assignment {
out := make(map[string]assignment, len(snap.Vms)+len(recs))
for _, d := range snap.Vms {
out[d.VmId] = assignment{desired: d, tombstoned: d.Tombstoned}
}
for id := range recs {
if _, ok := out[id]; !ok {
out[id] = assignment{} // absent from desired: vanished
}
}
return out
}
// tombstonedSet returns the ids the control plane has flagged for deletion.
// The destroy ack is level-triggered from it (see ackDestroyed).
func tombstonedSet(snap *pb.Snapshot) map[string]bool {
out := make(map[string]bool, len(snap.Vms))
for _, d := range snap.Vms {
if d.Tombstoned {
out[d.VmId] = true
}
}
return out
}
// Step routes one desired-state snapshot to the per-VM workers and returns the
// host's Report. It never waits for a worker: the report carries each VM's
// LAST-PUBLISHED state, so a VM busy in a multi-second operation cannot delay
// the heartbeat. A VM that has not published yet simply has no row.
//
// The algorithm is level-triggered: every call re-examines full state and drives
// toward desired. Idempotent under repeated identical snapshots.
//
// ctx is accepted for signature stability and bounds nothing here — the work
// happens in workers, where VMTimeout bounds each VM's pass.
func (e *Engine) Step(ctx context.Context, snap *pb.Snapshot) *pb.Report {
// ── 1. Epoch fence ───────────────────────────────────────────────────────
// CRITICAL: the fence path must CHANGE nothing: no SaveEpoch, no create,
// destroy or power call, no dispatch, no state mutations. It reads freely —
// fenceReport asks Prov.Running for every non-quarantined record — because it
// returns the current actual state so the control plane can observe what the
// agent actually has. A pass dispatched by an EARLIER, accepted snapshot may
// still be running: the fence refuses the stale snapshot, it does not freeze
// the host.
currentEpoch := e.St.Epoch()
if snap.Epoch < currentEpoch {
return e.fenceReport(currentEpoch)
}
// Advance epoch (equal is fine — same snapshot repeated).
_ = e.St.SaveEpoch(snap.Epoch)
// ── 2. Version floor ─────────────────────────────────────────────────────
// A snapshot this agent cannot fully read is not acted on: every VM in it
// fails legibly and nothing is dispatched. Checked AFTER the epoch is
// accepted so the report is not fenced and the control plane sees the
// refusal rather than a silent stall.
if floor := snap.GetMinAgentVersion(); floor != "" && version.Less(e.AgentVersion, floor) {
return e.refuseSnapshot(snap, floor)
}
// ── 3. Volumes ───────────────────────────────────────────────────────────
// Synchronously, and BEFORE dispatch: a VM's volumes must be files by the
// time its backend is handed them, so a VM and its volumes arriving in the
// same snapshot boot on the first try rather than the second. Cheap enough
// to stay on the heartbeat's path — a sparse truncate, no copy.
volumes := e.reconcileVolumes(snap)
// One record scan serves the whole tick: dispatch slices it into assignments
// and aggregate acks destroys against it. Reading it twice cost a second
// full state-dir scan on the one blocking path in Step, microseconds after
// the first, and bought no freshness worth having (see ackDestroyed).
recs, _ := e.St.LoadVMs()
// Deliberately NOT threaded with ctx: dispatch hands work to long-lived
// per-VM workers whose passes outlive this Step by design, so inheriting
// ctx would cancel them the instant the report went out. VMTimeout bounds
// each pass instead (see reconcileOne).
e.dispatch(snap, recs) //nolint:contextcheck // a pass deliberately outlives its Step
return e.aggregate(snap.Epoch, recs, volumes)
}
// dispatch hands every VM its slice of this snapshot and reaps the workers for
// VMs that are gone from both desired state and local records. recs is the
// caller's record view; the assignment set is the union of it and the snapshot.
//
// There is deliberately NO ordering between the resulting passes: a VM whose
// reap frees compute may run after a sibling's admission, so the sibling is
// quota-refused and boots on a later tick. The refusal is non-terminal and the
// loop is level-triggered, so the guarantee is eventual rather than same-tick.
func (e *Engine) dispatch(snap *pb.Snapshot, recs map[string]state.Record) {
live := assignments(snap, recs)
m := e.manager()
m.setTombstoned(tombstonedSet(snap))
for id, a := range live {
m.deliver(id, a)
}
m.reapAbsent(live)
}
// aggregate builds the host report from each worker's last-published result
// plus the level-triggered destroy ack. It is a pure read of published state:
// safe to call at any time, and it blocks on nothing.
//
// epoch is the ACCEPTED snapshot's epoch, threaded in from the caller and
// deliberately NOT re-read from disk: Store.Epoch fails open to 0, so one
// transient read error would report LastSeenEpoch 0 on a snapshot this host has
// just accepted and acted on — telling the control plane the agent is arbitrarily
// far behind. The caller knows the epoch it accepted; that is the true answer.
//
// recs is the record view the destroy ack is computed against, with the caller
// choosing how fresh it is (see ackDestroyed).
//
// volumes is this tick's volume convergence, threaded in for the same reason
// epoch is: the work was done in Step, and re-deriving it here would converge
// the volumes a second time per report. Nil is a caller that converged none —
// the paths that report on state without acting on it.
func (e *Engine) aggregate(epoch uint64, recs map[string]state.Record, volumes []*pb.VolumeStatus) *pb.Report {
rep := &pb.Report{LastSeenEpoch: epoch, Volumes: volumes}
m := e.manager()
m.collect(rep)
e.ackDestroyed(rep, m.tombstones(), recs)
return rep
}
// refuseSnapshot reports every VM in snap as failed because this agent is
// below the snapshot's floor. No record is written and no worker is poked.
//
// The power state is deliberately EMPTY rather than "stopped". This path
// inspects nothing — it is refusing before it reads any record — and the
// control plane persists what a report claims: saying "stopped" about guests
// that may well be running would turn a version refusal into a fleet-wide lie
// about power, and an operator would see healthy VMs go dark. Empty is the
// wire's "no observation", which is exactly the truth here.
//
// Tombstoned VMs are skipped: they are already deleted, nothing will act on
// them again, and hanging an upgrade-the-agent error on a VM on its way out
// would leave a spurious failure the operator cannot clear.
//
// VOLUMES: this report deliberately carries none, and that is safe only
// because of who reaches it. The control plane reaps a tombstoned volume when
// a VOLUMES-CAPABLE agent's report omits it, and an agent refusing a snapshot
// is by definition below that snapshot's floor — which, today, any snapshot
// carrying volumes sets to the release that introduced them. If a later
// feature ever raises a floor above a version that already reports volumes,
// this refusal must sweep and report them (see Engine.reportVolumes) or the
// refusal will read upward as "every tombstoned volume on this host is gone".
func (e *Engine) refuseSnapshot(snap *pb.Snapshot, floor string) *pb.Report {
rep := &pb.Report{LastSeenEpoch: snap.Epoch}
reason := fmt.Sprintf("agent %s is below this snapshot's floor %s; upgrade the agent", e.AgentVersion, floor)
for _, d := range snap.GetVms() {
if d.GetTombstoned() {
continue
}
rep.Vms = append(rep.Vms, newVMStatus(d.GetVmId(), addrs{}, "", "failed", reason))
}
return rep
}
// fenceReport is the read-only report returned for a stale snapshot: current
// actual state, derived entirely from persisted records and a stat of the
// volumes directory, with no mutation and no dispatch.
//
// The volume sweep is a report, not a convergence: no file is made, no
// tombstone marker written, nothing reclaimed. It is here because a report
// that named no volumes would be read upward as this host having none — and
// the control plane reaps a tombstoned volume on exactly that silence. The
// server refuses to reap off a fenced report as well; the two halves are
// independent, and each is worth having on its own.
func (e *Engine) fenceReport(currentEpoch uint64) *pb.Report {
rep := &pb.Report{FenceViolation: true, LastSeenEpoch: currentEpoch, Volumes: e.reportVolumes(nil)}
recs, err := e.St.LoadVMs()
if err != nil {
return rep
}
for _, rec := range recs {
var res vmResult
res.hostPubKey = rec.HostPubKey
if rec.QuarantinedAt != nil {
res.quarantined = quarantinedEntry(rec, e.graceFor(rec))
res.merge(rep)
continue
}
powerState := "stopped"
if e.Prov.Running(rec.Spec.VMID) {
powerState = "running"
}
phase := "ready"
if rec.LastError != "" {
phase = "failed"
}
res.report(rec.Spec.VMID, recAddrs(rec), powerState, phase, rec.LastError)
res.merge(rep)
}
return rep
}
// publisher hands a VM's row upward mid-pass, before the pass that owns it has
// finished. A pass publishes its result when it ENDS, and a first create ends
// minutes after it starts — image, disk, seed, boot — so the slowest and most
// opaque part of a VM's life was the part the host report carried no row for at
// all, and everything above it could say was "creating".
//
// Nil is a caller with nothing collecting rows (the fence path, a test calling
// a pass directly), and narration is then skipped rather than being an error:
// what it produces is commentary, and no decision anywhere rests on it.
type publisher func(vmResult)
// reconcileOne is ONE VM's complete reconcile pass: bounded by VMTimeout, it
// loads that VM's own record and either drives it toward desired or reaps it.
// It is the entire unit of work a per-VM worker runs, and it touches no other
// VM's record. The only shared state reconcile itself owns is the compute
// ledger, serialized under Engine.mu (see admit). The pass also reaches shared
// subsystems it does NOT own — the image cache (Images), and the host backend
// (Prov.Boot/Destroy), which owns both the guest's network attachment and its
// serial pump — each of which carries its own locking.
func (e *Engine) reconcileOne(ctx context.Context, id string, a assignment, pub publisher) (vmResult, bool) {
if e.VMTimeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, e.VMTimeout)
defer cancel()
}
var res vmResult
rec, ok, err := e.St.Get(id)
if err != nil {
// Record present but unreadable: skip this VM's pass entirely. While
// this repeats, the VM's reported row stays frozen at its last known
// state, so this log is the ONLY signal that the agent has stopped
// reconciling it — a stale row is otherwise indistinguishable from a
// healthy one. Not a per-tick flood for a normal condition: a VM with
// no record yet returns ok=false and a nil error from Store.Get and
// never reaches here.
slog.Warn("reconcile: skipping pass, VM record unreadable", "vm_id", id, "err", err)
return res, false
}
// Whatever this pass goes on to do, the VM's public host key rides its row.
// create overwrites this the moment it generates one.
res.hostPubKey = rec.HostPubKey
if a.desired != nil && !a.tombstoned {
if ok && rec.QuarantinedAt != nil {
rec.QuarantinedAt = nil
rec.QuarantineTombstoned = false
_ = e.St.SaveVM(rec)
}
e.reconcileVM(ctx, a.desired, rec, ok, &res, pub)
return res, true
}
// Absent from desired, or tombstoned: reap it. A tombstoned VM this host
// has no record of needs no pass at all — the destroy ack covers it.
if ok {
e.reapVM(ctx, id, rec, a.tombstoned, &res)
}
return res, true
}
// Step passes the view it dispatched with, read before this tick's destroying
// passes ran, so the ack lands one or more ticks AFTER the pass that removed the
// record, not in the same tick. Being level-triggered is what makes that fine.
func (e *Engine) ackDestroyed(rep *pb.Report, tombstoned map[string]bool, recs map[string]state.Record) {
for id := range tombstoned {
if _, hasRecord := recs[id]; !hasRecord {
rep.Destroyed = append(rep.Destroyed, id)
}
}
}
// reapVM drives ONE local record that is absent from desired or tombstoned:
// quarantine it (recording the stop intent before shutting down), then destroy
// it once its grace expires. Appends to rep as needed.
//
// This is the teardown branch of a VM's reconcile pass (see reconcileOne),
// split out so a single VM's teardown is expressible on its own.
func (e *Engine) reapVM(ctx context.Context, id string, rec state.Record, isTombstoned bool, res *vmResult) {
e.releaseCompute(id)
now := e.Now()
if rec.QuarantinedAt == nil {
// First time we see this VM needs reaping: enter quarantine.
t := now
rec.QuarantinedAt = &t
rec.QuarantineTombstoned = isTombstoned
rec.StopRequested = true // record BEFORE side effects
// Fix 5: only Shutdown after the stop intent is durably persisted.
// If SaveVM fails, skip Shutdown this cycle — the next reconcile
// will retry. This upholds "record stop BEFORE stopping".
if err := e.St.SaveVM(rec); err == nil {
_ = e.Prov.Shutdown(ctx, id)
}
} else if isTombstoned && !rec.QuarantineTombstoned {
// Upgrade: vanished quarantine → tombstoned quarantine (shorter grace).
rec.QuarantineTombstoned = true
_ = e.St.SaveVM(rec)
}
grace := e.graceFor(rec)
if now.Sub(*rec.QuarantinedAt) >= grace {
// NOTE: this may run with an expired pass ctx (watchdog). Safe today
// because the cloud-hypervisor backend's kill ignores ctx (SIGKILL via
// pidfile) — a backend that honors ctx throughout would skip the
// destroy until a later tick, which the level-triggered loop tolerates
// but delays.
if err := e.Prov.Destroy(ctx, id); err != nil {
return
}
_ = e.St.DeleteVM(id)
// Do NOT record a quarantined entry — VM is gone.
return
}
res.quarantined = quarantinedEntry(rec, grace)
}
// reconcileVM drives ONE active (non-tombstoned) desired VM toward its desired
// state: create it when it does not yet exist, else converge it. rec is this
// VM's own persisted record and ok reports whether one exists.
//
// exists = record present AND create completed. rec.BootID is the sole
// completion witness: create() writes it only after every side effect (image,
// disk, seed, boot) has succeeded. Disk presence is NOT a witness —
// create() writes disk.raw mid-sequence, so a create that fails after
// PrepareRootDisk but before boot leaves a disk on a still-empty BootID. Treating
// that disk as "exists" would divert the retry to converge(), which never
// rebuilds the disk or seed, and the VM would never recover.
func (e *Engine) reconcileVM(ctx context.Context, d *pb.VMSpec, rec state.Record, ok bool, res *vmResult, pub publisher) {
if !ok || rec.BootID == "" {
e.create(ctx, d, rec, ok, res, pub)
return
}
e.converge(ctx, d, rec, res)
}
func (e *Engine) graceFor(rec state.Record) time.Duration {
if rec.QuarantineTombstoned {
return e.TombstoneGrace
}
return e.VanishGrace
}
func (e *Engine) admit(vmID string, spec state.VMSpec) string {
e.mu.Lock()
defer e.mu.Unlock()
if e.committed == nil {
e.committed = map[string]state.VMSpec{}
}
if msg := e.quotaCheckLocked(vmID, spec); msg != "" {
return msg
}
e.committed[vmID] = spec
return ""
}
// acquireCreateSlot takes one of the host's create slots, returning the release
// func. It blocks until a slot frees or ctx expires — never longer, so a queued
// VM cannot outlive its own pass.
//
// This is deliberately NOT the admission gate: quota decides whether a VM may
// exist on this host at all and is recorded durably, whereas a create slot is
// transient scheduling that shapes how fast the host does the work.
func (e *Engine) acquireCreateSlot(ctx context.Context) (func(), error) {
if e.MaxConcurrentCreates <= 0 {
return func() {}, nil
}
e.createSlotsOnce.Do(func() {
e.createSlots = make(chan struct{}, e.MaxConcurrentCreates)
})
select {
case e.createSlots <- struct{}{}:
return func() { <-e.createSlots }, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
// note commits spec's compute for an existing VM without a quota check. Used for
// VMs already created (converge, un-delete re-adoption, startup rebuild) so the
// ledger always reflects every live VM regardless of how it got there.
func (e *Engine) note(vmID string, spec state.VMSpec) {
e.mu.Lock()
defer e.mu.Unlock()
if e.committed == nil {
e.committed = map[string]state.VMSpec{}
}
e.committed[vmID] = spec
}
func (e *Engine) releaseCompute(vmID string) {
e.mu.Lock()
defer e.mu.Unlock()
delete(e.committed, vmID)
}
// noteAddress folds the backend's current answers for this VM's addresses into
// rec, reporting whether either changed.
//
// The named
// NIC's address is the one that genuinely arrives late — the host discovers it
// by watching the guest's DHCP exchange — while rec.IP is known before the
// guest boots.
func (e *Engine) noteAddress(rec *state.Record) bool {
changed := false
if ip := e.Prov.Address(rec.Spec.VMID); ip != "" && ip != rec.IP {
rec.IP = ip
changed = true
}
if netIP := e.Prov.NetworkAddress(rec.Spec.VMID); netIP != "" && netIP != rec.NetworkIP {
rec.NetworkIP = netIP
changed = true
}
return changed
}
// SeedLedger rebuilds the compute ledger from persisted records at startup so
// the first step accounts for every surviving VM. Quarantined records are
// excluded (their guests are stopped; their compute is free).
func (e *Engine) SeedLedger(recs map[string]state.Record) {
e.mu.Lock()
defer e.mu.Unlock()
if e.committed == nil {
e.committed = map[string]state.VMSpec{}
}
for id, rec := range recs {
if rec.QuarantinedAt == nil {
e.committed[id] = rec.Spec
}
}
}
// quotaCheckLocked returns a non-empty reason when booting spec would exceed a
// configured host cap, else "". Caller holds e.mu. It sums the committed specs
// (excluding vmID itself, so a retry does not double-count) and adds spec's
// request.
func (e *Engine) quotaCheckLocked(vmID string, spec state.VMSpec) string {
vcpus, mem, disk := spec.VCPUs, spec.MemMB, spec.DiskGB
for id, s := range e.committed {
if id == vmID {
continue
}
vcpus += s.VCPUs
mem += s.MemMB
disk += s.DiskGB
}
switch {
case e.MaxVCPUs > 0 && vcpus > e.MaxVCPUs:
return fmt.Sprintf("host capacity limit reached: needs %d vcpus, host cap %d", vcpus, e.MaxVCPUs)
case e.MaxMemMB > 0 && mem > e.MaxMemMB:
return fmt.Sprintf("host capacity limit reached: needs %d MB memory, host cap %d", mem, e.MaxMemMB)
case e.MaxDiskGB > 0 && disk > e.MaxDiskGB:
return fmt.Sprintf("host capacity limit reached: needs %d GB disk, host cap %d", disk, e.MaxDiskGB)
}
return ""
}
// create attempts to create a new VM from desired state d. rec is this VM's own
// prior record (retry budget, last known address) and ok reports whether one
// exists. Quota comes from the serialized admission ledger (see admit); the
// address comes from the backend, at boot.
func (e *Engine) create(ctx context.Context, d *pb.VMSpec, rec state.Record, ok bool, res *vmResult, pub publisher) {
// Defensive: never start an attempt (which would burn retry budget) on a
// context that is already dead. UNREACHABLE today — a pass context is a
// fresh context.Background plus VMTimeout (see worker.run), so it cannot
// arrive here already expired, and Step's own ctx is threaded nowhere. Kept
// as cheap insurance for a caller that later threads a cancellable context
// into a pass. Report and let the next tick do the work.
if err := ctx.Err(); err != nil {
res.report(d.VmId, recAddrs(rec), "stopped", "creating", "reconcile aborted: "+err.Error())
return
}
spec := specFromWire(d)
// VMSpec stopped being ==-comparable when it grew a volume list; Equal is
// what == was, order of the volumes included.
if ok && !spec.Equal(rec.Spec) {
rec.CreateAttempts = 0
rec.LastError = ""
}
if ok && rec.CreateAttempts >= e.MaxCreateAttempts {
res.report(d.VmId, recAddrs(rec), "stopped", "failed", rec.LastError)
return
}
var hostKey state.HostKey
if d.HostCertRequired {
rec.Spec = spec // SaveVM keys off the record's own spec
var err error
if hostKey, err = e.HostKey(e.St.HostKeyPath(d.VmId)); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
if rec.HostPubKey != hostKey.PublicLine {
rec.HostPubKey = hostKey.PublicLine
if err := e.St.SaveVM(rec); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
}
res.hostPubKey = rec.HostPubKey
if d.SshHostCert == "" {
res.report(d.VmId, recAddrs(rec), "stopped", "creating", "awaiting host certificate")
return
}
}
rec.Spec = spec
if quotaMsg := e.admit(d.VmId, spec); quotaMsg != "" {
res.report(d.VmId, recAddrs(rec), "stopped", "failed", quotaMsg)
return
}
rec.CreateAttempts++
rec.CreatedAt = e.Now()
rec.LastError = "" // clear for this attempt
// Record BEFORE side effects so a crash is recoverable.
if err := e.St.SaveVM(rec); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
// say publishes what this create is doing right now, on the row this VM
// would report anyway: still creating, still stopped, no error. It carries
// rec's address and this VM's public host key because BOTH are
// level-triggered — a row that dropped the key would ask the control plane
// to forget a guest's certificate for a tick — and rec is read at each call,
// so a step after Boot narrates with the address Boot found.
//
// A dead pass says nothing. say runs on whatever goroutine calls it, and the
// image cache's progress callback runs on a shared fetch goroutine that
// keeps draining after this pass abandons the download at VMTimeout — this
// stops that goroutine calling into a finished pass at all. It is the polite
// half of the guarantee, not the load-bearing one: ctx expiry and the pass
// ending are not the same instant, and a pass with no VMTimeout has a
// context that never expires. What actually makes a stale row unlandable is
// the publisher's generation check (see worker.publish).
say := func(detail string) {
if pub == nil || ctx.Err() != nil {
return
}
var interim vmResult
interim.hostPubKey = res.hostPubKey
interim.report(d.VmId, recAddrs(rec), "stopped", "creating", "")
interim.vm.StatusDetail = detail
pub(interim)
}
if err := e.Prov.Preflight(ctx); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
// Per-VM workers made these concurrent — N simultaneous
// creates mean N image downloads, N image decodes and N multi-GB disk
// copies against one device, which can starve the state dir that Step scans
// every tick (the heartbeat's one remaining blocking path) and can exhaust
// disk on the in-flight temporaries alone. The wait costs the VM nothing but
// time: it publishes the line below and then stays on it, so a queued VM
// reads as queued rather than as stalled. A wait that outlives VMTimeout
// fails the attempt, and failCreate REFUNDS a ctx-expiry failure, so a queued
// VM never burns retry budget for waiting.
say("waiting for a create slot on this host")
release, err := e.acquireCreateSlot(ctx)
if err != nil {
e.failCreate(ctx, rec, err, res)
return
}
defer release()
say("downloading image")
basePath, err := e.Images(ctx, d.ImageUrl, d.ImageSha256, func(done, total int64) {
say(downloadDetail(done, total))
})
if err != nil {
e.failCreate(ctx, rec, err, res)
return
}
say("preparing root disk")
if err := e.Prov.PrepareRootDisk(ctx, rec.Spec, basePath); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
// Build cloud-init seed ISO. The guest gets its address from the host
// network, so no IP or gateway is baked into the seed.
say("building the cloud-init seed")
if err := e.Seed(e.St.SeedPath(d.VmId), seed.Params{
Hostname: d.Name,
InstanceID: d.VmId,
SSHAuthorizedKey: d.SshAuthorizedKey,
UserData: d.CloudInit,
SSHUserCAAuthorizedKey: joinCALines(d.GetSshUserCaAuthorizedKeys()),
SSHHostKeyPEM: hostKey.PrivatePEM,
SSHHostCert: d.SshHostCert,
MAC: state.MAC(d.VmId),
NetworkMAC: netMACIfNetworked(rec.Spec),
}); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
// Boot if desired running.
if d.PowerState == "running" {
if err := e.volumesMaterialized(rec.Spec); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
say("booting")
if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {
e.failCreate(ctx, rec, err, res)
return
}
e.noteAddress(&rec)
}
// Success: record completion.
rec.BootID = e.BootID()
rec.StopRequested = d.PowerState != "running"
rec.LastError = ""
_ = e.St.SaveVM(rec)
powerState := "running"
if d.PowerState != "running" {
powerState = "stopped"
}
res.report(d.VmId, recAddrs(rec), powerState, "ready", "")
}
// isPermanent reports whether err (anywhere in its chain) carries the
// consumer-owned permanence marker — the provisioner's way of saying no
// retry can ever succeed (e.g. the disk-shrink guard). Consumer-side
// interface per the R5 convention: reconcile declares it; the producers
// (cloudhv, vfkit, imagecache, netenv) mint one via internal/agent/permanent,
// as does this package's own volumesMaterialized, which knows a missing volume
// is not something this host can heal.
func isPermanent(err error) bool {
var p interface{ Permanent() bool }
return errors.As(err, &p) && p.Permanent()
}
// failCreate records a failed create attempt and appends a report row. Ordering: the ctx
// refund wins over permanence — a permanent error surfacing under an expired
// ctx is refunded this tick and, being deterministic, terminal-fails on the
// next tick's fresh ctx. Keeps the watchdog invariant unconditional.
func (e *Engine) failCreate(ctx context.Context, rec state.Record, err error, res *vmResult) {
if ctx.Err() != nil {
rec.CreateAttempts-- // refund: this VM's pass died mid-operation
} else if isPermanent(err) {
rec.CreateAttempts = e.MaxCreateAttempts // terminal now; retry cannot succeed
}
rec.LastError = err.Error()
_ = e.St.SaveVM(rec)
phase := "creating"
if rec.CreateAttempts >= e.MaxCreateAttempts {
phase = "failed"
}
res.report(rec.Spec.VMID, recAddrs(rec), "stopped", phase, rec.LastError)
}
func (e *Engine) failConverge(rec state.Record, err error, res *vmResult) {
rec.LastError = err.Error()
if why := e.Prov.FailureReason(rec.Spec.VMID); why != "" {
rec.LastError += ": " + why
}
_ = e.St.SaveVM(rec)
res.report(rec.Spec.VMID, recAddrs(rec), "stopped", "failed", rec.LastError)
}
// converge drives an existing VM toward its desired power state,
// handling lost detection and restart logic.
func (e *Engine) converge(ctx context.Context, d *pb.VMSpec, rec state.Record, res *vmResult) {
// Keep the ledger reflecting this live VM (covers the un-delete case, where a
// quarantined VM returns to desired after its compute was released).
e.note(d.VmId, rec.Spec)
running := e.Prov.Running(d.VmId)
bootID := e.BootID()
// The address is a polled fact — a backend whose host OS hands it out only
// learns it once the guest has asked. On a
// backend that allocates before boot this never fires after the first pass.
if e.noteAddress(&rec) {
_ = e.St.SaveVM(rec)
}
// lost = boot ID changed OR process died without a recorded stop request.
// A deliberately stopped VM has StopRequested=true, so !running && StopRequested is NOT lost.
lost := rec.BootID != bootID || (!running && !rec.StopRequested)
if lost {
// Every VM is persistent — a lost guest is booted again, and there is no
// longer a policy to consult before doing it.
if d.PowerState == "running" {
// Restart: the backend re-attaches the VM to the host network as
// part of Boot, which is what rebuilds a tap that did not survive
// the host reboot. Its volumes have to survive too.
if err := e.volumesMaterialized(rec.Spec); err != nil {
e.failConverge(rec, err, res)
return
}
if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {
e.failConverge(rec, err, res)
return
}
e.noteAddress(&rec)
rec.BootID = bootID
rec.StopRequested = false
rec.LastError = "" // Fix 3: clear stale error on successful restart
_ = e.St.SaveVM(rec)
res.report(d.VmId, recAddrs(rec), "running", "ready", "")
} else {
// Lost + desired stopped: update boot ID, mark stop recorded.
rec.BootID = bootID
rec.StopRequested = true
_ = e.St.SaveVM(rec)
res.report(d.VmId, recAddrs(rec), "stopped", "ready", "")
}
return
}
// Not lost: drive power state.
if d.PowerState == "running" && !running {
// Start the VM. Boot re-attaches the network, idempotently.
if err := e.volumesMaterialized(rec.Spec); err != nil {
e.failConverge(rec, err, res)
return
}
if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {
e.failConverge(rec, err, res)
return
}
e.noteAddress(&rec)
rec.StopRequested = false
rec.LastError = "" // Fix 3: clear stale error on successful boot
_ = e.St.SaveVM(rec)
res.report(d.VmId, recAddrs(rec), "running", "ready", "")
} else if d.PowerState == "stopped" && running {
// Stop the VM.
// Fix 5: if SaveVM fails, skip Shutdown — the durability guarantee
// (record stop BEFORE stopping) must hold; stopping without a durable
// record would cause the VM to be treated as lost after a crash.
rec.StopRequested = true
if err := e.St.SaveVM(rec); err != nil {
// Cannot durably record the stop intent; skip Shutdown this cycle.
// The next reconcile will retry once the store recovers.
res.report(d.VmId, recAddrs(rec), "running", "failed", err.Error())
return
}
_ = e.Prov.Shutdown(ctx, d.VmId)
res.report(d.VmId, recAddrs(rec), "stopped", "ready", "")
} else {
// Already at desired state.
powerState := "stopped"
if running {
powerState = "running"
}
// Preserve last error in the report field but phase stays ready
// (the VM is converged; the error is informational history).
res.report(d.VmId, recAddrs(rec), powerState, "ready", rec.LastError)
}
}
func quarantinedEntry(rec state.Record, grace time.Duration) *pb.QuarantinedVM {
specJSON, _ := json.Marshal(rec.Spec)
destroyAt := rec.QuarantinedAt.Add(grace).Unix()
return &pb.QuarantinedVM{
VmId: rec.Spec.VMID,
Name: rec.Spec.Name,
VmspecJson: specJSON,
DestroyAtUnix: destroyAt,
}
}
// vmResult is ONE VM's contribution to the host report: at most one actual-VM
// row, or one quarantined entry. Per-VM reconcile writes here instead of
// appending straight into the shared report, so a single VM's output stands on
// its own — which is what lets a per-VM worker publish it independently.
type vmResult struct {
vm *pb.VMStatus
quarantined *pb.QuarantinedVM
hostPubKey string
}
// addrs is where a VM is: its address on its host's NAT underlay, which every
// guest has from boot, and — for a guest with a second NIC on a named host
// network — the address the site's own DHCP server granted that one.
type addrs struct{ ip, networkIP string }
// recAddrs reads a record's pair. The record is the only place both are known
// at once: the backend answers them one question at a time.
func recAddrs(rec state.Record) addrs { return addrs{ip: rec.IP, networkIP: rec.NetworkIP} }
// report records this VM's actual row. A VM contributes at most one row, so a
// later call in the same reconcile replaces an earlier one.
func (r *vmResult) report(vmID string, at addrs, powerState, phase, lastError string) {
r.vm = newVMStatus(vmID, at, powerState, phase, lastError)
}
// clone returns a deep copy, so the caller's report owns its rows outright.
// A worker publishes a result once and then keeps serving it until its next
// pass ends, so without this every report taken in between aliases the same
// protos — see collect.
func (r vmResult) clone() vmResult {
out := r
if r.vm != nil {
out.vm = proto.Clone(r.vm).(*pb.VMStatus)
}
if r.quarantined != nil {
out.quarantined = proto.Clone(r.quarantined).(*pb.QuarantinedVM)
}
return out
}
// merge folds this VM's result into the host report.
func (r *vmResult) merge(rep *pb.Report) {
if r.vm != nil {
r.vm.SshHostPubkey = r.hostPubKey
rep.Vms = append(rep.Vms, r.vm)
}
if r.quarantined != nil {
rep.Quarantined = append(rep.Quarantined, r.quarantined)
}
}
// downloadDetail words how far an image download has come.
func downloadDetail(done, total int64) string {
scale, unit := byteScale(max(done, total))
if total <= 0 {
return fmt.Sprintf("downloading image %.1f %s", float64(done)/scale, unit)
}
return fmt.Sprintf("downloading image %.1f/%.1f %s", float64(done)/scale, float64(total)/scale, unit)
}
// byteScale picks the unit a size reads best in: the largest one it is at least
// one of, floored at MiB because an image smaller than that does not exist and
// a download reported in kilobytes would look like a failure.
func byteScale(n int64) (float64, string) {
if n >= 1<<30 {
return 1 << 30, "GiB"
}
return 1 << 20, "MiB"
}
// newVMStatus builds one VMStatus row from what a reconcile pass observed.
// The row's remaining field, ssh_host_pubkey, is stamped by merge — see
// vmResult.hostPubKey. Unset values are the proto zero-value "".
func newVMStatus(vmID string, at addrs, powerState, phase, lastError string) *pb.VMStatus {
return &pb.VMStatus{
VmId: vmID,
Ip: at.ip,
NetworkIp: at.networkIP,
PowerState: powerState,
Phase: phase,
LastError: lastError,
}
}
// joinCALines renders the tenant user-CA set into the multi-line content of the
// guest's TrustedUserCAKeys file — one canonical CA per line, each newline-
// terminated. Empty set ⇒ "" ⇒ the seed writes no drop-in (gate off / no CA).
func joinCALines(lines []string) string {
var b strings.Builder
for _, l := range lines {
b.WriteString(strings.TrimRight(l, "\r\n") + "\n")
}
return b.String()
}
func netMACIfNetworked(spec state.VMSpec) string {
if spec.Network == "" {
return ""
}
return state.NetMAC(spec.VMID)
}
// specFromWire maps a pb.VMSpec to state.VMSpec.
func specFromWire(d *pb.VMSpec) state.VMSpec {
return state.VMSpec{
VMID: d.VmId,
Name: d.Name,
ImageURL: d.ImageUrl,
ImageSHA256: d.ImageSha256,
CloudInit: d.CloudInit,
SSHAuthorizedKey: d.SshAuthorizedKey,
Network: d.Network,
// Cloned: the record outlives the snapshot it came from, and a worker
// holds its assignment well past the Step that delivered it.
VolumeIDs: slices.Clone(d.VolumeIds),
VCPUs: d.Vcpus,
MemMB: d.MemMb,
DiskGB: d.DiskGb,
}
}