diff --git a/free/ci/bench/perf-budget.json b/free/ci/bench/perf-budget.json index 67620e48e..afaeb8886 100644 --- a/free/ci/bench/perf-budget.json +++ b/free/ci/bench/perf-budget.json @@ -8,7 +8,7 @@ "ci.e2e_tiny.v15": 2541, "ci.exec_overhead": 59, "ci.plan": 5, - "ci.startup": 14, + "ci.startup": 34, "ci.store_open": 58 } } diff --git a/free/ci/cmd/nodes_pools.go b/free/ci/cmd/nodes_pools.go index 3f2a5624c..71dbe5ca0 100644 --- a/free/ci/cmd/nodes_pools.go +++ b/free/ci/cmd/nodes_pools.go @@ -253,7 +253,7 @@ func resolvePoolsPolicy(ctx context.Context, st *store.Store, dir string) (strin return "", trust.Effective{}, errors.New(plainText(dir) + " is not a directory") } run := policy.Runner(report.Capture) - project := policy.ProjectIDGit(ctx, run, root) + project := policy.ProjectIDGitCached(ctx, run, root) // the run's start path asks again (policy hook, dispatch) res := policy.Resolver{Store: st, Run: run} got, err := res.Resolve(ctx, policy.Repo{Root: root, Project: project, Visibility: "unknown", Class: "owner"}, policy.Revision{Dir: root}) if err != nil { diff --git a/free/ci/cmd/v15.go b/free/ci/cmd/v15.go index fda77934e..e4da6e242 100644 --- a/free/ci/cmd/v15.go +++ b/free/ci/cmd/v15.go @@ -160,6 +160,9 @@ func parseBare(args []string) (RunOptions, error) { } return RunOptions{}, errFlagParse } + if bare.help { // help prints and exits: the option hooks open the store and run git (P7-CI-46) + return RunOptions{}, nil + } if dashes && len(argv) == 0 { return RunOptions{}, adhoc.UsageError("`nself ci --` needs a command after the dashes") } @@ -256,12 +259,14 @@ func runBare(args []string) int { } return code } + outside := make(chan bool, 1) // two git reads, independent of the plan: they run beside it (P7-CI-46) + go func() { outside <- configOutside(ctx, o.Root) }() p, digest, subst, err := preparePlan(ctx, o) if err != nil { return exitForError(err) } attachPlacement(ctx, o) - c := &cirun.Coordinator{Opts: o, Plan: p, ConfigDigest: digest, ConfigOutside: configOutside(ctx, o.Root), Substantive: subst, Out: os.Stdout} + c := &cirun.Coordinator{Opts: o, Plan: p, ConfigDigest: digest, ConfigOutside: <-outside, Substantive: subst, Out: os.Stdout} res, err := c.Run(ctx) if err != nil { return exitForError(err) diff --git a/free/ci/internal/dispatch/loop.go b/free/ci/internal/dispatch/loop.go index a626333a7..41d34a4bb 100644 --- a/free/ci/internal/dispatch/loop.go +++ b/free/ci/internal/dispatch/loop.go @@ -3,7 +3,8 @@ package dispatch // Purpose: the dispatcher (Epic D7): a queue of attempts waiting for a // placement and the tick that places and admits them. A tick runs when a job // is enqueued, when a waiter's poll fires (attempt end and lease expiry free -// capacity without a signal), and every Poll interval (one second in TEAM mode). +// capacity without a signal unless the caller calls Wake), and every Poll +// interval (one second in TEAM mode). // Inputs: Config (store, world builder, orderer, sink, transport, clock) and // Submit calls, one per attempt, each blocking until its attempt is admitted. // Outputs: Admitted attempts; E618 for a foreground job no runner can take; @@ -98,6 +99,7 @@ type Dispatcher struct { seq uint64 ticked uint64 lastTick time.Time + wake chan struct{} // closed by Wake: the waiters tick now } var ( @@ -187,6 +189,7 @@ func (d *Dispatcher) Submit(ctx context.Context, j Job) (Admitted, error) { return Admitted{}, err } for { + wake := d.wakeCh() // taken before due(): a Wake between the two is not lost if d.due() { _ = d.Tick(ctx) // a failed tick leaves the job queued; foreground world errors finish it } @@ -197,6 +200,7 @@ func (d *Dispatcher) Submit(ctx context.Context, j Job) (Admitted, error) { return d.drop(e, ctx.Err()) case <-d.life.Done(): return d.drop(e, errClosed) + case <-wake: case <-time.After(d.cfg.Poll): } } diff --git a/free/ci/internal/dispatch/queue.go b/free/ci/internal/dispatch/queue.go index 904ac8809..ae8e3f8fe 100644 --- a/free/ci/internal/dispatch/queue.go +++ b/free/ci/internal/dispatch/queue.go @@ -72,6 +72,30 @@ func (d *Dispatcher) push(j Job, max int64) (*entry, int64, error) { return e, depth, nil } +// Wake makes the waiters tick now instead of at their next poll: the caller +// freed capacity (an attempt ended) that a queued job may be waiting for. On a +// tiny run the poll was a measured share of every hand-off (P7-CI-46). +func (d *Dispatcher) Wake() { + d.mu.Lock() + d.lastTick = time.Time{} + ch := d.wake + d.wake = nil + d.mu.Unlock() + if ch != nil { + close(ch) + } +} + +// wakeCh is the channel the next Wake closes. +func (d *Dispatcher) wakeCh() <-chan struct{} { + d.mu.Lock() + defer d.mu.Unlock() + if d.wake == nil { + d.wake = make(chan struct{}) + } + return d.wake +} + // due reports whether a waiter should tick: something was enqueued since the // last tick, or Poll has passed. func (d *Dispatcher) due() bool { diff --git a/free/ci/internal/evidence/binding.go b/free/ci/internal/evidence/binding.go index 3ea406ed5..be6fd75a7 100644 --- a/free/ci/internal/evidence/binding.go +++ b/free/ci/internal/evidence/binding.go @@ -25,9 +25,13 @@ type Worktree struct { } // ReadWorktree reads the state of root. +// The status and the hidden-edit listing are independent reads: they run side +// by side (P7-CI-46), so a clean tree costs one funnel spawn of wall time. func ReadWorktree(ctx context.Context, c Capture, root string) Worktree { + hidden := make(chan bool, 1) + go func() { hidden <- hiddenEdits(ctx, c, root) }() status, ok := git(ctx, c, root, "status", "--porcelain", "--untracked-files=normal") - if ok && strings.TrimSpace(status) == "" && !hiddenEdits(ctx, c, root) { + if ok && strings.TrimSpace(status) == "" && !<-hidden { return Worktree{} } w := Worktree{Dirty: true} diff --git a/free/ci/internal/exec/capacity.go b/free/ci/internal/exec/capacity.go index 316c7b763..600b172f7 100644 --- a/free/ci/internal/exec/capacity.go +++ b/free/ci/internal/exec/capacity.go @@ -4,7 +4,6 @@ import ( "bufio" "bytes" "os" - osexec "os/exec" "runtime" "strconv" "strings" @@ -120,46 +119,3 @@ func probeLinux(c *Capacity) { } } } - -func probeDarwin(c *Capacity) { - if out, err := osexec.Command("sysctl", "-n", "hw.memsize").Output(); err == nil { - n, _ := strconv.ParseInt(strings.TrimSpace(string(out)), 10, 64) - c.MemoryMB = int(n / (1024 * 1024)) - } - if out, err := osexec.Command("vm_stat").Output(); err == nil { - pageSize := int64(4096) - var pages int64 - for _, line := range strings.Split(string(out), "\n") { - if strings.Contains(line, "page size of") { - f := strings.Fields(line) - for i, v := range f { - if v == "of" && i+1 < len(f) { - pageSize, _ = strconv.ParseInt(f[i+1], 10, 64) - } - } - } - if strings.HasPrefix(line, "Pages free:") || strings.HasPrefix(line, "Pages inactive:") || strings.HasPrefix(line, "Pages speculative:") { - f := strings.Fields(line) - if len(f) > 0 { - n, _ := strconv.ParseInt(strings.TrimSuffix(f[len(f)-1], "."), 10, 64) - pages += n - } - } - } - c.FreeMemoryMB = int(pages * pageSize / (1024 * 1024)) - } - if out, err := osexec.Command("pmset", "-g", "batt").Output(); err == nil { - s := string(out) - c.PluggedIn = strings.Contains(s, "AC Power") - for _, field := range strings.Fields(s) { - if strings.HasSuffix(field, "%;") || strings.HasSuffix(field, "%") { - n, _ := strconv.Atoi(strings.TrimSuffix(strings.TrimSuffix(field, ";"), "%")) - c.BatteryPercent = n - break - } - } - } - if out, err := osexec.Command("pmset", "-g", "custom").Output(); err == nil { - c.LowPower = strings.Contains(string(out), "lowpowermode 1") - } -} diff --git a/free/ci/internal/exec/capacity_darwin.go b/free/ci/internal/exec/capacity_darwin.go new file mode 100644 index 000000000..f8b878e55 --- /dev/null +++ b/free/ci/internal/exec/capacity_darwin.go @@ -0,0 +1,132 @@ +//go:build darwin + +package exec + +// Purpose: the macOS half of the capacity probe (P7-CI-32): total memory, free +// memory and power state. +// Inputs: the hw.memsize sysctl, vm_stat and pmset. +// Outputs: the memory and power fields of a Capacity. +// Constraints: a run probes once per admission and once per dispatcher tick +// (P7-CI-46 measured four child processes, about 17 ms, every time), so the +// total memory is read once with a sysctl (no fork), free memory is cached for +// freeMemoryTTL and the power state for powerTTL, and the two pmset calls and +// vm_stat run side by side when they are due. Neither value moves at the +// rate a run asks for it; the store's reservation ledger, not this snapshot, +// is what keeps two admissions from taking the same memory. + +import ( + osexec "os/exec" + "strconv" + "strings" + "sync" + "time" + + "golang.org/x/sys/unix" +) + +const ( + freeMemoryTTL = 250 * time.Millisecond + powerTTL = 30 * time.Second +) + +// darwinPower is the parsed pmset state; battery is -1 when pmset names none. +type darwinPower struct { + battery int + plugged, low bool +} + +var darwinProbe struct { + mu sync.Mutex + memMB int + memRead bool + freeMB int + freeAt time.Time + power darwinPower + powerAt time.Time +} + +func probeDarwin(c *Capacity) { + p := &darwinProbe + p.mu.Lock() + defer p.mu.Unlock() + now := time.Now() + if !p.memRead { + if n, err := unix.SysctlUint64("hw.memsize"); err == nil { + p.memMB = int(n / (1024 * 1024)) + } + p.memRead = true + } + var wg sync.WaitGroup + if p.freeAt.IsZero() || now.Sub(p.freeAt) >= freeMemoryTTL { + wg.Add(1) + go func() { + defer wg.Done() + p.freeMB = 0 + if out, err := osexec.Command("vm_stat").Output(); err == nil { + p.freeMB = parseVMStat(string(out)) + } + }() + p.freeAt = now + } + if p.powerAt.IsZero() || now.Sub(p.powerAt) >= powerTTL { + wg.Add(1) + go func() { + defer wg.Done() + p.power = readPower() + }() + p.powerAt = now + } + wg.Wait() + c.MemoryMB, c.FreeMemoryMB = p.memMB, p.freeMB + c.BatteryPercent, c.PluggedIn, c.LowPower = p.power.battery, p.power.plugged, p.power.low +} + +// parseVMStat is free + inactive + speculative pages, in MB. +func parseVMStat(out string) int { + pageSize := int64(4096) + var pages int64 + for _, line := range strings.Split(out, "\n") { + if strings.Contains(line, "page size of") { + f := strings.Fields(line) + for i, v := range f { + if v == "of" && i+1 < len(f) { + pageSize, _ = strconv.ParseInt(f[i+1], 10, 64) + } + } + } + if strings.HasPrefix(line, "Pages free:") || strings.HasPrefix(line, "Pages inactive:") || strings.HasPrefix(line, "Pages speculative:") { + f := strings.Fields(line) + if len(f) > 0 { + n, _ := strconv.ParseInt(strings.TrimSuffix(f[len(f)-1], "."), 10, 64) + pages += n + } + } + } + return int(pages * pageSize / (1024 * 1024)) +} + +// readPower runs the two pmset queries side by side. +func readPower() darwinPower { + pw := darwinPower{battery: -1} + var batt, custom []byte + var battErr, customErr error + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); batt, battErr = osexec.Command("pmset", "-g", "batt").Output() }() + go func() { defer wg.Done(); custom, customErr = osexec.Command("pmset", "-g", "custom").Output() }() + wg.Wait() + if battErr == nil { + s := string(batt) + pw.plugged = strings.Contains(s, "AC Power") + for _, field := range strings.Fields(s) { + if strings.HasSuffix(field, "%;") || strings.HasSuffix(field, "%") { + pw.battery, _ = strconv.Atoi(strings.TrimSuffix(strings.TrimSuffix(field, ";"), "%")) + break + } + } + } + if customErr == nil { + pw.low = strings.Contains(string(custom), "lowpowermode 1") + } + return pw +} diff --git a/free/ci/internal/exec/capacity_other.go b/free/ci/internal/exec/capacity_other.go new file mode 100644 index 000000000..54e8414e4 --- /dev/null +++ b/free/ci/internal/exec/capacity_other.go @@ -0,0 +1,6 @@ +//go:build !darwin + +package exec + +// probeDarwin is the macOS probe (capacity_darwin.go); Probe calls it only on darwin. +func probeDarwin(*Capacity) {} diff --git a/free/ci/internal/run/cache_hook.go b/free/ci/internal/run/cache_hook.go index a0adc4d89..765d95a5e 100644 --- a/free/ci/internal/run/cache_hook.go +++ b/free/ci/internal/run/cache_hook.go @@ -172,7 +172,9 @@ func (h *cacheHook) begin(ctx context.Context, a AttemptInfo, p store.PipelineRo es := entries.New(a.Store, blobs, entries.Options{Home: home, Project: func(string) authz.ProjectFacts { return authz.ProjectFacts{NoPrivateDeps: !priv} }}) in := specs.BeginInput{Engine: &specs.Engine{Entries: es, Blobs: blobs, Scanner: h.scanner(), Backend: a.Store}, Attempt: a.AttemptID, Epoch: a.Epoch, Facts: facts, Root: filepath.Join(home, "cache-work", a.AttemptID), Detected: detected, Scope: scope, ExtraKeys: keys, Container: container, - Platform: runtime.GOOS + "/" + runtime.GOARCH, Isolation: "process", JobEnv: a.Job.Env, Environ: flatEnv(env), Probe: func(ctx context.Context, argv []string) (string, error) { return h.probe(ctx, a.Job, p.Project, argv) }, Home: h.home(), Getenv: func(k string) string { return env[k] }, Warn: h.warn} + Platform: runtime.GOOS + "/" + runtime.GOARCH, Isolation: "process", JobEnv: a.Job.Env, Environ: flatEnv(env), Probe: func(ctx context.Context, argv []string) (string, error) { + return h.probeOnce(ctx, a.RunID, a.Job, p.Project, argv) + }, Home: h.home(), Getenv: func(k string) string { return env[k] }, Warn: h.warn} if container { in.Platform, in.Isolation, in.Image = "linux/"+runtime.GOARCH, "container", h.imageDigest(ctx, a) } diff --git a/free/ci/internal/run/cache_hook_probe.go b/free/ci/internal/run/cache_hook_probe.go new file mode 100644 index 000000000..834a92a6b --- /dev/null +++ b/free/ci/internal/run/cache_hook_probe.go @@ -0,0 +1,85 @@ +package run + +// Purpose: run each cache tool probe (`go version` and the like) once per run. +// P7-CI-46 measured the probe at 40 to 70 ms on macOS (it runs through the +// exec funnel in a copy of the tree), paid again by every job of the run. +// Inputs: the run id, the job (workdir, env), the project root and the argv. +// Outputs: the probe's output, shared by every job that asks the same question. +// Constraints: the key holds everything the answer depends on (run, root, +// workdir, the job's env, argv), so two jobs with a different toolchain pin +// still probe separately; a failed probe is not remembered; entries older than +// probeMemoTTL are dropped, so a long-lived coordinator does not grow. + +import ( + "context" + "sort" + "strings" + "sync" + "time" + + "github.com/nself-org/plugins/free/ci/internal/model" +) + +const probeMemoTTL = 10 * time.Minute + +type probeResult struct { + done chan struct{} + out string + err error + at time.Time +} + +var probeMemo struct { + mu sync.Mutex + m map[string]*probeResult +} + +func probeKey(runID, root string, job model.Job, argv []string) string { + env := make([]string, 0, len(job.Env)) + for k, v := range job.Env { + env = append(env, k+"="+v) + } + sort.Strings(env) + return strings.Join([]string{runID, root, job.Workdir, strings.Join(env, "\x01"), strings.Join(argv, "\x01")}, "\x00") +} + +// probeOnce answers argv for the job from the run's memo, probing on a miss. +// Concurrent callers with the same key wait for the one probe in flight. +func (h *cacheHook) probeOnce(ctx context.Context, runID string, job model.Job, root string, argv []string) (string, error) { + key := probeKey(runID, root, job, argv) + now := time.Now() + probeMemo.mu.Lock() + if probeMemo.m == nil { + probeMemo.m = map[string]*probeResult{} + } + if r, ok := probeMemo.m[key]; ok { + probeMemo.mu.Unlock() + select { + case <-r.done: + if r.err == nil { + return r.out, nil + } + case <-ctx.Done(): + return "", ctx.Err() + } + return h.probe(ctx, job, root, argv) // the shared probe failed: ask again + } + for k, r := range probeMemo.m { + if now.Sub(r.at) > probeMemoTTL { + delete(probeMemo.m, k) + } + } + r := &probeResult{done: make(chan struct{}), at: now} + probeMemo.m[key] = r + probeMemo.mu.Unlock() + r.out, r.err = h.probe(ctx, job, root, argv) + if r.err != nil { + probeMemo.mu.Lock() + if probeMemo.m[key] == r { + delete(probeMemo.m, key) + } + probeMemo.mu.Unlock() + } + close(r.done) + return r.out, r.err +} diff --git a/free/ci/internal/run/cache_hook_session.go b/free/ci/internal/run/cache_hook_session.go index db3a3ff5f..19f0cf319 100644 --- a/free/ci/internal/run/cache_hook_session.go +++ b/free/ci/internal/run/cache_hook_session.go @@ -3,6 +3,7 @@ package run import ( "context" "os" + "sync" "time" "github.com/nself-org/plugins/free/ci/internal/model" @@ -41,25 +42,56 @@ func (h *cacheHook) Wants(a AttemptInfo) bool { return ok && e.(cacheEntry).s.Wants() } -// After removes the attempt's cache directories; it saves nothing (closeSession -// also drops the entry of an attempt that did not settle passed). +// After removes the attempt's cache directories; it saves nothing (it also +// drops the entry of an attempt that did not settle passed). The retraction +// reads the store now; the removal (a restored build cache is tens of MB of +// small files, P7-CI-46 measured 60 to 95 ms) runs in the background, and the +// run waits for it before Coordinator.Run returns (waitCacheCleanups). func (h *cacheHook) After(ctx context.Context, a AttemptInfo, _ Outcome) error { if e, ok := h.sessions.LoadAndDelete(a.AttemptID); ok { - closeSession(ctx, a.Store, e.(cacheEntry)) + ce := e.(cacheEntry) + retractUnpassed(ctx, a.Store, ce) + wg := cleanupGroup(a.RunID) + wg.Add(1) + go func() { + defer wg.Done() + ce.s.Cleanup() + }() } return nil } +// cacheCleanups holds, per run id, the directory removals After started. +var cacheCleanups sync.Map // run id -> *sync.WaitGroup + +func cleanupGroup(runID string) *sync.WaitGroup { + wg, _ := cacheCleanups.LoadOrStore(runID, &sync.WaitGroup{}) + return wg.(*sync.WaitGroup) +} + +// waitCacheCleanups waits for the removals the run's attempts started. Every +// After of the run happened before its jobs finished, so before this call. +func waitCacheCleanups(runID string) { + if wg, ok := cacheCleanups.LoadAndDelete(runID); ok { + wg.(*sync.WaitGroup).Wait() + } +} + // closeSession retracts the entries of an attempt that did not settle passed -// (a cancel won the finalizing -> passed move after the commit) and removes the -// attempt's cache directories. After and eviction both end a session this way. +// and removes the attempt's cache directories (eviction ends a session this way). func closeSession(ctx context.Context, st *store.Store, e cacheEntry) { + retractUnpassed(ctx, st, e) + e.s.Cleanup() +} + +// retractUnpassed retracts the entries of an attempt that did not settle passed +// (a cancel won the finalizing -> passed move after the commit). +func retractUnpassed(ctx context.Context, st *store.Store, e cacheEntry) { if e.s.Saved() { if state, ok := settledState(ctx, e, st); ok && state != "passed" { // only a confirmed non-pass retracts e.s.Retract(ctx) } } - e.s.Cleanup() } // settledState reads the attempt's state from the store it ran in, falling back diff --git a/free/ci/internal/run/coordinator.go b/free/ci/internal/run/coordinator.go index df19d313b..6bca88aa4 100644 --- a/free/ci/internal/run/coordinator.go +++ b/free/ci/internal/run/coordinator.go @@ -185,6 +185,9 @@ func (c *Coordinator) Run(ctx context.Context) (Result, error) { } defer release() rev.Dirty = rev.Dirty || c.ConfigOutside // a snapshot revision never hides a config that lives outside the tree + // the evidence's start reads git only: it runs beside admission, policy and the store (P7-CI-46) + begun := make(chan any, 1) + go func() { begun <- c.beginEvidence(ctx, h, rev) }() admitted, source, err := c.classify(ctx, h, rev) if err != nil { return Result{}, err @@ -208,7 +211,8 @@ func (c *Coordinator) Run(ctx context.Context) (Result, error) { post := c.prepareStatus(ctx, st, rev) _ = st.Heartbeat(ctx, coord.ID) runID := newRunID() - defer dispatchRelease(runID) // also closes the dispatcher of a run that ends before AfterRun (D-1044) + defer waitCacheCleanups(runID) // the attempts' cache directories are gone when the run returns + defer dispatchRelease(runID) // also closes the dispatcher of a run that ends before AfterRun (D-1044) sess := newSession() mp := c.Plan.ModelPipeline() rows, err := c.persist(ctx, st, runID, rev, source, pol, sess, mp) @@ -237,7 +241,7 @@ func (c *Coordinator) Run(ctx context.Context) (Result, error) { if subst == nil { subst = func(plan.PlannedJob) bool { return true } } - rs := &runState{source: model.TrustClass(source), c: c, sess: sess, hooks: h, pol: pol, pl: c.Plan, mp: mp, rev: rev, rows: rows, subst: subst, render: stampRun(journal(ctx, st, runID, NewRenderer(out, c.Opts)), runID), subset: keep, begin: c.beginEvidence(ctx, h, rev), + rs := &runState{source: model.TrustClass(source), c: c, sess: sess, hooks: h, pol: pol, pl: c.Plan, mp: mp, rev: rev, rows: rows, subst: subst, render: stampRun(journal(ctx, st, runID, NewRenderer(out, c.Opts)), runID), subset: keep, begin: <-begun, out: map[string]*JobOutcome{}, done: map[string]chan struct{}{}, marks: map[string]model.Reason{}, cancels: map[string]context.CancelFunc{}, attempts: map[string]string{}} sess.Dirty, sess.ConfigOutside, sess.Subset = rev.Dirty, c.ConfigOutside, keep != nil sess.AdmitTrust = model.TrustClass(admitted) diff --git a/free/ci/internal/run/dispatch_hook.go b/free/ci/internal/run/dispatch_hook.go index 6f36a3ab6..b4aec4e56 100644 --- a/free/ci/internal/run/dispatch_hook.go +++ b/free/ci/internal/run/dispatch_hook.go @@ -75,6 +75,14 @@ func dispatchRelease(runID string) { } } +// dispatchWake lets the run's queued jobs tick now: an attempt ended and its +// host reservation is free. +func dispatchWake(runID string) { + if v, ok := runDispatchers.Load(runID); ok { + v.(*runDispatch).d.Wake() + } +} + func dispatchHook(ctx context.Context, s AttemptSpec) (Admitted, error) { if !compat.V15() { return LocalAdmit(ctx, s) diff --git a/free/ci/internal/run/jobrun.go b/free/ci/internal/run/jobrun.go index e703cb996..2700a7f33 100644 --- a/free/ci/internal/run/jobrun.go +++ b/free/ci/internal/run/jobrun.go @@ -51,7 +51,13 @@ func (rs *runState) trans(id string, from, to model.JobState, class model.Failur if state != from { return fail(CodeRun, "attempt %s is %s, not %s", id, state, from) } - return rs.sess.Store.Transition(bg, id, from, to, store.TransitionDetail{Epoch: epoch, FailureClass: class}) + if err = rs.sess.Store.Transition(bg, id, from, to, store.TransitionDetail{Epoch: epoch, FailureClass: class}); err != nil { + return err + } + if to == "passed" || to == "failed" || to == "cancelled" || to == "lost" { // the store freed the host reservation + dispatchWake(rs.sess.RunID) + } + return nil } // settle ends an attempt that is in state `from`. A committed cancel intent