amadan

carlos / platform Public

host: one run's life: launch, follow, cancel, timeout, result after the last chunk (run spec §5.2 to §5.7, codex 3, 4, 5, R2-2, R2-6)

Prepare checks the record before and after staging the files, so a cancel
during the download starts no unit. The chunker's context outlives the run
and is bounded once the command ends. The done CAS only lands for a run the
record names as this box's under this boot, and only after result.json.
Pickup (task 4 review): a request.json that is not the record's run or has
an out-of-range timeout is never picked; a pickup CAS that answered an
error is read back and launched if it landed.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1b41f045602b6f4153c73d3226d77a6e094f6923 parent d963bc0
4 files changed, +1027 −10
  • internal/host/runexec.go +370 −0
  • internal/host/runexec_test.go +524 −0
  • internal/host/runs.go +54 −10
  • internal/host/runs_test.go +79 −0
diff --git a/internal/host/runexec.go b/internal/host/runexec.go
new file mode 100644
index 000000000..6e7cb5604
--- /dev/null
+++ b/internal/host/runexec.go
@@ -0,0 +1,370 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+// runPollRetry paces a failed long-poll; runRecheckEvery re-reads the
+// record while a run executes, a backstop for a cancel whose serial bump
+// this box missed.
+var (
+ runPollRetry = time.Second
+ runRecheckEvery = 30 * time.Second
+)
+
+// runOutputGrace bounds the output once the command has ended: the unit's
+// stop (whose streams drain into the chunker) and the last chunk. The
+// chunker's context outlives the command, so a cancel or a timeout never
+// gives up the last chunk; this bound is what keeps a store that refuses
+// every Put from retrying under the chunker's lock forever. It is longer
+// than runStopBound.
+var runOutputGrace = 5 * time.Minute
+
+// runFinishBound bounds the retries of result.json and of the done CAS.
+var runFinishBound = 2 * time.Minute
+
+const (
+ // runStopBound bounds CmdHandle.Stop: above the activator's own stop
+ // bound, which covers the start, the stop, the cgroup and both streams.
+ runStopBound = 3 * time.Minute
+ // runReadBound bounds one read of the record.
+ runReadBound = 30 * time.Second
+)
+
+var (
+ errRunCancelledBeforeLaunch = errors.New("host: the run was cancelled before launch")
+ errRunImportAtLaunch = errors.New("host: the instance is an import that is not committed")
+ errRunIncarnation = errors.New("host: the record is another incarnation")
+ errRunNotOurs = errors.New("host: the record no longer names this run as this box's")
+)
+
+// exitOf reads systemd's ExecMainCode and ExecMainStatus. `systemctl show`
+// prints the code as a number (1 exited, 2 killed, 3 dumped); the words
+// are accepted too. A signal leaves exit 0 and names the signal.
+func exitOf(st instancectl.CmdStatus) (int, string) {
+ n, _ := strconv.Atoi(st.Status)
+ switch st.Code {
+ case "2", "3", "killed", "dumped":
+ return 0, strconv.Itoa(n)
+ }
+ return n, ""
+}
+
+// runResultForLaunch maps RunCmd's refusals, and Prepare's, to the result
+// the member reads. A route that is not an instance's (a worker) is
+// refused as a flat box is: runs need an instance on the nested layout.
+// Anything else (a busy hold, a failed download, a failed start) is the
+// box failing to start the command.
+func runResultForLaunch(err error) runs.Result {
+ var we *activator.CmdWakeError
+ var se *activator.CmdStoppedError
+ switch {
+ case errors.Is(err, activator.ErrCmdClaim):
+ return runs.Result{Refused: runs.RefusedClaim}
+ case errors.Is(err, activator.ErrCmdNotNested), errors.Is(err, activator.ErrCmdNotInstance):
+ return runs.Result{Refused: runs.RefusedFlat}
+ case errors.As(err, &we):
+ return runs.Result{Refused: runs.RefusedWake, Detail: we.Reason}
+ case errors.As(err, &se):
+ return runs.Result{Interrupted: se.Reason}
+ case errors.Is(err, errRunCancelledBeforeLaunch):
+ return runs.Result{Cancelled: true}
+ case errors.Is(err, errRunImportAtLaunch):
+ return runs.Result{Refused: runs.RefusedImport}
+ case errors.Is(err, errRunIncarnation):
+ return runs.Result{Refused: runs.RefusedIncarnation}
+ }
+ return runs.Result{Interrupted: runs.InterruptedStartFailed}
+}
+
+// recordIncarnation is the incarnation request.json must carry for rec;
+// "" when it cannot be computed, which matches no request.
+func recordIncarnation(rec provision.Record) string {
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ return ""
+ }
+ return inc
+}
+
+// runFinishOurs is RunFinish for a run the record names as picked by box
+// under boot. RunFinish itself does not check who picked the run.
+func runFinishOurs(id, box, boot string) func(*provision.Record) error {
+ finish := provision.RunFinish(id)
+ return func(rec *provision.Record) error {
+ if r := rec.ActiveRun; r == nil || r.ID != id || r.Box != box || r.BootID != boot {
+ return fmt.Errorf("%w: the record does not name run %s as this box's", provision.ErrRunConflict, id)
+ }
+ return finish(rec)
+ }
+}
+
+// execute is one picked run's life (run spec §5.2 to §5.7): the
+// incarnation, the launch (Prepare checks the record again and stages the
+// files), the poll, a cancel or a timeout, then the unit stopped, every
+// chunk written, result.json, the done CAS, the run directory deleted and
+// the hold released, in that order.
+func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
+ defer w.forget(req.ID)
+ instDir := provision.InstanceDir(w.dataDir, rec.Host)
+ if req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host || req.Incarnation == "" || req.Incarnation != recordIncarnation(rec) {
+ w.finish(ctx, rec, req, runs.Result{Refused: runs.RefusedIncarnation}, nil)
+ return
+ }
+ // The chunker's context is not the run's: a cancel or a timeout must
+ // not give up the last chunk. endOutput is armed once the command has
+ // ended (runOutputGrace).
+ octx, endOutput := context.WithCancel(context.WithoutCancel(ctx))
+ defer endOutput()
+ chunks := runs.NewChunker(octx, w.store, rec.Account, rec.App, req.ID)
+ closeOutput := func(res *runs.Result) {
+ last, truncated, err := chunks.Close()
+ if err != nil {
+ log.Printf("agent: runs: %s %s: output: %v", rec.Host, req.ID, err)
+ }
+ res.LastSeq, res.Truncated = last, truncated
+ }
+ spec := activator.CmdSpec{
+ RunID: req.ID,
+ Args: req.Argv,
+ RunDir: filepath.Join(instDir, runs.DirName, req.ID),
+ Stdout: chunks.Stream(runs.StreamOut),
+ Stderr: chunks.Stream(runs.StreamErr),
+ Prepare: func(ctx context.Context) error {
+ return w.prepare(ctx, instDir, req, h)
+ },
+ }
+ handle, err := w.launch.RunCmd(ctx, rt, spec)
+ if err != nil {
+ bound := time.AfterFunc(runOutputGrace, endOutput)
+ defer bound.Stop()
+ if errors.Is(err, errRunNotOurs) {
+ // The record gives the run to no one, or to another boot:
+ // whoever holds it writes its result.
+ chunks.Close()
+ w.cleanUp(rec, req, nil)
+ log.Printf("agent: runs: %s %s: not launched: %v", rec.Host, req.ID, err)
+ return
+ }
+ res := runResultForLaunch(err)
+ if res.Interrupted == runs.InterruptedStartFailed {
+ log.Printf("agent: runs: %s %s: start: %v", rec.Host, req.ID, err)
+ }
+ closeOutput(&res)
+ w.finish(ctx, rec, req, res, nil)
+ return
+ }
+ log.Printf("agent: runs: %s: run %s started (%s)", rec.Host, req.ID, handle.Invocation())
+ res := runs.Result{Started: w.now().UTC()}
+ status := w.follow(ctx, rec, req, handle, h, &res)
+ bound := time.AfterFunc(runOutputGrace, endOutput)
+ defer bound.Stop()
+ // The unit is stopped and settled after every ending, done included
+ // (run spec §5.4), so its streams have reached their end before the
+ // chunker closes and the result names the last chunk.
+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runStopBound)
+ if err := handle.Stop(sctx); err != nil {
+ log.Printf("agent: runs: %s %s: stop: %v", rec.Host, req.ID, err)
+ }
+ cancel()
+ if status == nil && (res.Cancelled || res.TimedOut) {
+ pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineCmdPoll)
+ if st, err := handle.Poll(pctx); err == nil && st.State == instancectl.CmdDone {
+ status = &st
+ }
+ pcancel()
+ }
+ if status != nil && status.State == instancectl.CmdDone {
+ res.Exit, res.Signal = exitOf(*status)
+ }
+ if reason := handle.Interrupted(); reason != "" {
+ res.Interrupted = reason
+ } else if status != nil && status.State == instancectl.CmdGone && !res.Cancelled && !res.TimedOut {
+ res.Interrupted = runs.InterruptedGone
+ }
+ closeOutput(&res)
+ w.finish(ctx, rec, req, res, handle)
+}
+
+// prepare is the run's CmdSpec.Prepare, run under the activator's hold
+// right before the unit starts (run spec §5.2): the record must still name
+// the run as this box's under this boot, not cancelled, not an uncommitted
+// import, the same incarnation. Then the files are staged, a cancel cutting
+// the download short, and the record is checked once more, so a cancel
+// that lands while the files download still starts no unit.
+func (w *runsWorker) prepare(ctx context.Context, instDir string, req runs.Request, h *runHandle) error {
+ if err := w.launchable(ctx, req, h); err != nil {
+ return err
+ }
+ sctx, stop := context.WithCancel(ctx)
+ defer stop()
+ go func() {
+ select {
+ case <-h.cancel:
+ stop()
+ case <-sctx.Done():
+ }
+ }()
+ if _, err := runs.StageFiles(sctx, w.store, instDir, req); err != nil {
+ if cancelRequested(h) {
+ return errRunCancelledBeforeLaunch
+ }
+ return fmt.Errorf("host: stage the run's files: %w", err)
+ }
+ return w.launchable(ctx, req, h)
+}
+
+// launchable reads the record and says why the run must not start, or nil.
+func (w *runsWorker) launchable(ctx context.Context, req runs.Request, h *runHandle) error {
+ if cancelRequested(h) {
+ return errRunCancelledBeforeLaunch
+ }
+ rctx, cancel := context.WithTimeout(ctx, runReadBound)
+ defer cancel()
+ cur, err := provision.GetRecord(rctx, w.store, req.Account, req.App, req.Host)
+ if errors.Is(err, os.ErrNotExist) {
+ return errRunNotOurs
+ }
+ if err != nil {
+ return fmt.Errorf("host: read the record before launch: %w", err)
+ }
+ r := cur.ActiveRun
+ switch {
+ case r == nil || r.ID != req.ID || r.Box != w.box || r.BootID != w.boot || (r.State != provision.RunPicked && r.State != provision.RunCancel):
+ return errRunNotOurs
+ case r.State == provision.RunCancel:
+ return errRunCancelledBeforeLaunch
+ case cur.ImportUncommitted():
+ return errRunImportAtLaunch
+ case recordIncarnation(*cur) != req.Incarnation:
+ return errRunIncarnation
+ }
+ return nil
+}
+
+func cancelRequested(h *runHandle) bool {
+ select {
+ case <-h.cancel:
+ return true
+ default:
+ return false
+ }
+}
+
+// follow long-polls the unit until it is done or gone, the run is
+// cancelled, or its timeout passes. It returns the final status when the
+// unit reported one, and marks a cancel or a timeout on res.
+func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) *instancectl.CmdStatus {
+ pctx, stopPolling := context.WithCancel(ctx)
+ defer stopPolling()
+ statuses := make(chan instancectl.CmdStatus, 1)
+ go func() {
+ for {
+ st, err := handle.Poll(pctx)
+ if pctx.Err() != nil {
+ return
+ }
+ if err != nil {
+ select {
+ case <-pctx.Done():
+ return
+ case <-time.After(runPollRetry):
+ }
+ continue
+ }
+ if st.State != instancectl.CmdRunning {
+ statuses <- st
+ return
+ }
+ }
+ }()
+ timeout := time.NewTimer(req.Timeout())
+ defer timeout.Stop()
+ recheck := time.NewTicker(runRecheckEvery)
+ defer recheck.Stop()
+ for {
+ select {
+ case st := <-statuses:
+ return &st
+ case <-h.cancel:
+ res.Cancelled = true
+ return nil
+ case <-timeout.C:
+ res.TimedOut = true
+ return nil
+ case <-recheck.C:
+ rctx, cancel := context.WithTimeout(ctx, runReadBound)
+ cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host)
+ cancel()
+ if err == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == req.ID && cur.ActiveRun.State == provision.RunCancel {
+ h.requestCancel()
+ }
+ }
+ }
+}
+
+// finish writes the run's result, then CASes the run to done only if the
+// result landed and the record names the run as this box's under this
+// boot, then cleans up. A result that cannot be written leaves the run
+// picked: the sweep ends it as lost with a result of its own, so a run is
+// never done without one.
+func (w *runsWorker) finish(ctx context.Context, rec provision.Record, req runs.Request, res runs.Result, handle activator.CmdHandle) {
+ res.Box, res.Ended = w.box, w.now().UTC()
+ fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runFinishBound)
+ defer cancel()
+ err := untilLanded(fctx, func(ctx context.Context) error {
+ return runs.WriteResult(ctx, w.store, rec.Account, rec.App, req.ID, res)
+ })
+ if err != nil {
+ log.Printf("agent: runs: %s %s: result: %v", rec.Host, req.ID, err)
+ } else if err := untilLanded(fctx, func(ctx context.Context) error {
+ _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, runFinishOurs(req.ID, w.box, w.boot))
+ return err
+ }); err != nil && !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: done: %v", rec.Host, req.ID, err)
+ }
+ w.cleanUp(rec, req, handle)
+}
+
+// cleanUp deletes the run's directory (an error is logged, never acted
+// on) and releases the activator's hold.
+func (w *runsWorker) cleanUp(rec provision.Record, req runs.Request, handle activator.CmdHandle) {
+ if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), req.ID); err != nil && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, req.ID, err)
+ }
+ if handle != nil {
+ handle.Release()
+ }
+}
+
+// untilLanded retries f until it succeeds, answers a conflict or a missing
+// object (neither changes on a retry), or ctx ends.
+func untilLanded(ctx context.Context, f func(context.Context) error) error {
+ backoff := 250 * time.Millisecond
+ for {
+ err := f(ctx)
+ if err == nil || errors.Is(err, provision.ErrRunConflict) || errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ select {
+ case <-ctx.Done():
+ return err
+ case <-time.After(backoff):
+ }
+ backoff = min(backoff*2, 5*time.Second)
+ }
+}
diff --git a/internal/host/runexec_test.go b/internal/host/runexec_test.go
new file mode 100644
index 000000000..431ae87c0
--- /dev/null
+++ b/internal/host/runexec_test.go
@@ -0,0 +1,524 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "io"
+ "os"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+type fakeHandle struct {
+ mu sync.Mutex
+ polls chan instancectl.CmdStatus
+ stops int
+ released bool
+ interrupted string
+ afterStop *instancectl.CmdStatus
+}
+
+func (h *fakeHandle) Invocation() string { return "inv1" }
+func (h *fakeHandle) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
+ h.mu.Lock()
+ if h.stops > 0 {
+ st := instancectl.CmdStatus{State: instancectl.CmdGone}
+ if h.afterStop != nil {
+ st = *h.afterStop
+ }
+ h.mu.Unlock()
+ return st, nil
+ }
+ h.mu.Unlock()
+ select {
+ case st := <-h.polls:
+ return st, nil
+ case <-ctx.Done():
+ return instancectl.CmdStatus{}, ctx.Err()
+ }
+}
+func (h *fakeHandle) Interrupted() string { h.mu.Lock(); defer h.mu.Unlock(); return h.interrupted }
+func (h *fakeHandle) Stop(context.Context) error {
+ h.mu.Lock()
+ h.stops++
+ h.mu.Unlock()
+ return nil
+}
+func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }
+
+type fakeLauncher struct {
+ mu sync.Mutex
+ calls int
+ err error
+ handle *fakeHandle
+ out string
+ errOut string
+}
+
+func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
+ l.mu.Lock()
+ l.calls++
+ l.mu.Unlock()
+ if l.err != nil {
+ return nil, l.err
+ }
+ if spec.Prepare != nil {
+ if err := spec.Prepare(ctx); err != nil {
+ return nil, err
+ }
+ }
+ io.WriteString(spec.Stdout, l.out)
+ io.WriteString(spec.Stderr, l.errOut)
+ return l.handle, nil
+}
+
+// pickedRun queues and picks a run for box-1/boot-1 and returns what
+// execute is called with.
+func pickedRun(t *testing.T, s release.Store, timeoutS int64) (provision.Record, runs.Request) {
+ t.Helper()
+ rec, id := queueRun(t, s, time.Now().Add(time.Minute))
+ req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.TimeoutS = timeoutS
+ got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return *got, *req
+}
+
+// execWorker is a box-1/boot-1 worker over a data dir that holds the
+// instance's directory, as a real box does.
+func execWorker(t *testing.T, s release.Store, l runLauncher, boot string) *runsWorker {
+ t.Helper()
+ dataDir := t.TempDir()
+ if err := os.MkdirAll(provision.InstanceDir(dataDir, runHost), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ return newRunsWorker(s, runOwnedRoute(runHost), l, dataDir, "box-1", boot)
+}
+
+func runWith(t *testing.T, s release.Store, l *fakeLauncher, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
+ t.Helper()
+ return runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
+}
+
+func runOn(t *testing.T, w *runsWorker, s release.Store, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
+ t.Helper()
+ rt, _ := runOwnedRoute(runHost)(runHost)
+ w.active[req.ID] = h
+ w.execute(context.Background(), rec, req, *rt, h)
+ res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID)
+ if err != nil || res == nil {
+ t.Fatalf("no result: %v", err)
+ }
+ if w.active[req.ID] != nil {
+ t.Error("the run was not forgotten")
+ }
+ if _, err := os.Stat(runDirOf(w, req)); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("the run directory is still there: %v", err)
+ }
+ return res
+}
+
+func runDirOf(w *runsWorker, req runs.Request) string {
+ return provision.InstanceDir(w.dataDir, runHost) + "/" + runs.DirName + "/" + req.ID
+}
+
+func TestRunExecutesAndWritesTheResultAfterTheLastChunk(t *testing.T) {
+ s := &putOrder{Store: release.NewDirStore(t.TempDir())}
+ rec, req := pickedRun(t, s, 60)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
+ h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "3", Code: "1"}
+ l := &fakeLauncher{handle: h, out: "hello\n", errOut: "oops\n"}
+ res := runWith(t, s, l, rec, req, newRunHandle())
+ if res.Exit != 3 || res.Signal != "" || res.LastSeq != 2 || res.Box != "box-1" || res.Started.IsZero() || res.Ended.IsZero() {
+ t.Fatalf("result %+v", res)
+ }
+ if h.stops == 0 || !h.released {
+ t.Fatal("the unit was not stopped after done, or the hold not released")
+ }
+ if r := currentRun(t, s.Store); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+ if !s.resultAfterChunks() {
+ t.Fatalf("puts %q: result.json must come after every chunk", s.keys)
+ }
+}
+
+// putOrder records the order of Puts.
+type putOrder struct {
+ release.Store
+ mu sync.Mutex
+ keys []string
+}
+
+func (p *putOrder) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ p.mu.Lock()
+ p.keys = append(p.keys, key)
+ p.mu.Unlock()
+ return p.Store.Put(ctx, key, body, ifMatch)
+}
+
+func (p *putOrder) resultAfterChunks() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ result := -1
+ lastChunk := -1
+ for i, k := range p.keys {
+ if strings.HasSuffix(k, "/result.json") {
+ result = i
+ }
+ if strings.Contains(k, "/out/") {
+ lastChunk = i
+ }
+ }
+ return result > lastChunk
+}
+
+func TestRunKilledBySignal(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
+ h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "signal", Status: "9", Code: "2"}
+ if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Signal != "9" || res.Exit != 0 {
+ t.Fatalf("result %+v", res)
+ }
+}
+
+// ctxStore refuses a Put whose context has ended, as S3 does; DirStore
+// ignores it.
+type ctxStore struct{ release.Store }
+
+func (c ctxStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if err := ctx.Err(); err != nil {
+ return "", err
+ }
+ return c.Store.Put(ctx, key, body, ifMatch)
+}
+
+func TestRunCancelledAfterLaunch(t *testing.T) {
+ s := ctxStore{release.NewDirStore(t.TempDir())}
+ rec, req := pickedRun(t, s, 60)
+ done := instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "143", Code: "1"}
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus), afterStop: &done}
+ rh := newRunHandle()
+ go func() { time.Sleep(50 * time.Millisecond); rh.requestCancel() }()
+ res := runWith(t, s, &fakeLauncher{handle: h, out: "partial\n"}, rec, req, rh)
+ if !res.Cancelled || res.Exit != 143 || res.LastSeq != 1 {
+ t.Fatalf("result %+v (the last chunk lands after a cancel)", res)
+ }
+ if r := currentRun(t, s); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+func TestRunTimesOut(t *testing.T) {
+ s := ctxStore{release.NewDirStore(t.TempDir())}
+ rec, req := pickedRun(t, s, 1)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ start := time.Now()
+ res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, rec, req, newRunHandle())
+ if !res.TimedOut || time.Since(start) < time.Second || h.stops == 0 {
+ t.Fatalf("result %+v after %s, stops %d", res, time.Since(start), h.stops)
+ }
+ if res.LastSeq != 1 || res.Interrupted != "" {
+ t.Fatalf("result %+v: the last chunk lands after a timeout, and a stopped unit is not interrupted", res)
+ }
+}
+
+// Review Focus 2: a cancel after the pickup CAS and before the unit starts
+// ends the run cancelled with no unit started.
+func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ if _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID)); err != nil {
+ t.Fatal(err)
+ }
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
+ if !res.Cancelled || h.stops != 0 || !res.Started.IsZero() {
+ t.Fatalf("result %+v, unit stops %d", res, h.stops)
+ }
+ if r := currentRun(t, s); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+// cancelOnInput cancels the run the moment its input is read: a cancel that
+// lands while the files download.
+type cancelOnInput struct {
+ release.Store
+ acct, app, host, id string
+}
+
+func (c *cancelOnInput) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "/in/") {
+ if _, err := provision.SetRecordRun(ctx, c.Store, c.acct, c.app, c.host, provision.RunCancelTransition(c.id)); err != nil {
+ return nil, "", err
+ }
+ }
+ return c.Store.Get(ctx, key)
+}
+
+func TestRunCancelledWhileFilesDownloadNeverStarts(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, dir, 60)
+ body := []byte("a,b\n")
+ sum := sha256.Sum256(body)
+ req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
+ if _, err := dir.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ s := &cancelOnInput{Store: dir, acct: rec.Account, app: rec.App, host: rec.Host, id: req.ID}
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ l := &fakeLauncher{handle: h}
+ res := runWith(t, s, l, rec, req, newRunHandle())
+ if !res.Cancelled || h.stops != 0 || !res.Started.IsZero() {
+ t.Fatalf("result %+v, unit stops %d: the record is read again after the download", res, h.stops)
+ }
+}
+
+func TestRunStagesItsFiles(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ body := []byte("a,b\n")
+ sum := sha256.Sum256(body)
+ req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
+ if _, err := s.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ w := execWorker(t, s, nil, "boot-1")
+ var staged []byte
+ l := &stagedReader{read: func(spec activator.CmdSpec) {
+ staged, _ = os.ReadFile(spec.RunDir + "/in.csv")
+ if spec.RunDir != runDirOf(w, req) {
+ t.Errorf("RunDir %s, want %s", spec.RunDir, runDirOf(w, req))
+ }
+ }}
+ w.launch = l
+ runOn(t, w, s, rec, req, newRunHandle())
+ if string(staged) != string(body) {
+ t.Fatalf("staged %q, want %q", staged, body)
+ }
+}
+
+// stagedReader prepares, reads what was staged, and refuses to start.
+type stagedReader struct{ read func(activator.CmdSpec) }
+
+func (l *stagedReader) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
+ if err := spec.Prepare(ctx); err != nil {
+ return nil, err
+ }
+ l.read(spec)
+ return nil, errors.New("start cmd unit: 502")
+}
+
+func TestRunRefusesAnIncarnationThatDoesNotMatch(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ req.Incarnation = strings.Repeat("0", 64)
+ l := &fakeLauncher{handle: &fakeHandle{}}
+ if res := runWith(t, s, l, rec, req, newRunHandle()); res.Refused != runs.RefusedIncarnation || l.calls != 0 {
+ t.Fatalf("result %+v, launches %d", res, l.calls)
+ }
+ if r := currentRun(t, s); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+func TestRunRefusesAnImportUncommittedAtLaunch(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ cur, etag, err := provision.GetRecordETag(context.Background(), s, rec.Account, rec.App, rec.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ cur.Seed = &provision.SeedSource{Kind: provision.SeedKindImport, State: "placed"}
+ if err := provision.UpdateRecord(context.Background(), s, *cur, etag); err != nil {
+ t.Fatal(err)
+ }
+ if res := runWith(t, s, &fakeLauncher{handle: &fakeHandle{}}, rec, req, newRunHandle()); res.Refused != runs.RefusedImport {
+ t.Fatalf("result %+v (codex R2-2: checked again at launch)", res)
+ }
+}
+
+func TestRunLaunchErrorsBecomeResults(t *testing.T) {
+ cases := []struct {
+ err error
+ want func(runs.Result) bool
+ }{
+ {activator.ErrCmdClaim, func(r runs.Result) bool { return r.Refused == runs.RefusedClaim }},
+ {activator.ErrCmdNotNested, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
+ {activator.ErrCmdNotInstance, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
+ {&activator.CmdWakeError{Reason: "wake failed: restore", Err: errors.New("x")}, func(r runs.Result) bool {
+ return r.Refused == runs.RefusedWake && r.Detail == "wake failed: restore"
+ }},
+ {&activator.CmdStoppedError{Reason: activator.StopRestart}, func(r runs.Result) bool { return r.Interrupted == activator.StopRestart }},
+ {activator.ErrCmdBusy, func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
+ {errors.New("start cmd unit: 502"), func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
+ }
+ for _, c := range cases {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ if res := runWith(t, s, &fakeLauncher{err: c.err}, rec, req, newRunHandle()); !c.want(*res) {
+ t.Errorf("%v: result %+v", c.err, res)
+ }
+ if r := currentRun(t, s); r.State != provision.RunDone {
+ t.Errorf("%v: run %+v, want done", c.err, r)
+ }
+ }
+}
+
+func TestRunInterruptedByAnInstanceStop(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart}
+ h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "143", Code: "1"}
+ if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != activator.StopRestart {
+ t.Fatalf("result %+v", res)
+ }
+}
+
+func TestRunWhoseUnitWentAway(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
+ h.polls <- instancectl.CmdStatus{State: instancectl.CmdGone, InvocationID: "inv2"}
+ if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != runs.InterruptedGone {
+ t.Fatalf("result %+v", res)
+ }
+}
+
+// The record-read backstop: a cancel whose serial bump this box missed is
+// still seen while the run executes.
+func TestRunSeesACancelByRereadingTheRecord(t *testing.T) {
+ defer func(d time.Duration) { runRecheckEvery = d }(runRecheckEvery)
+ runRecheckEvery = 20 * time.Millisecond
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ go func() {
+ time.Sleep(50 * time.Millisecond)
+ provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID))
+ }()
+ if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); !res.Cancelled {
+ t.Fatalf("result %+v", res)
+ }
+}
+
+// failingPuts fails every Put whose key holds part.
+type failingPuts struct {
+ release.Store
+ part string
+}
+
+func (f *failingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if strings.Contains(key, f.part) {
+ return "", errors.New("AccessDenied")
+ }
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+// A store that refuses every chunk cannot hold the run open: the output's
+// bound ends the chunker, and the result says truncated.
+func TestRunOutputIsBoundedWhenEveryChunkFails(t *testing.T) {
+ defer func(d time.Duration) { runOutputGrace = d }(runOutputGrace)
+ runOutputGrace = 300 * time.Millisecond
+ dir := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, dir, 60)
+ s := &failingPuts{Store: dir, part: "/out/"}
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
+ h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"}
+ start := time.Now()
+ res := runWith(t, s, &fakeLauncher{handle: h, out: "lost\n"}, rec, req, newRunHandle())
+ if !res.Truncated || res.LastSeq != 0 || time.Since(start) > 10*time.Second {
+ t.Fatalf("result %+v after %s", res, time.Since(start))
+ }
+ if r := currentRun(t, dir); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+// A result that cannot be written leaves the run picked, for the sweep to
+// end as lost: done is never set without a result.
+func TestRunWithoutAResultIsNotDone(t *testing.T) {
+ defer func(d time.Duration) { runFinishBound = d }(runFinishBound)
+ runFinishBound = 300 * time.Millisecond
+ dir := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, dir, 60)
+ s := &failingPuts{Store: dir, part: "/result.json"}
+ w := execWorker(t, s, &fakeLauncher{err: activator.ErrCmdClaim}, "boot-1")
+ rt, _ := runOwnedRoute(runHost)(runHost)
+ w.execute(context.Background(), rec, req, *rt, newRunHandle())
+ if r := currentRun(t, dir); r.State != provision.RunPicked {
+ t.Fatalf("run %+v, want still picked", r)
+ }
+}
+
+// The done CAS is only for a run the record names as this box's under this
+// boot (RunFinish does not check), and a box writes no result for a run the
+// record gives to another: whoever holds it ends it.
+func TestRunFinishesOnlyItsOwnPick(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60) // picked under boot-1
+ h := &fakeHandle{}
+ w := execWorker(t, s, &fakeLauncher{handle: h}, "boot-2")
+ rt, _ := runOwnedRoute(runHost)(runHost)
+ w.active[req.ID] = newRunHandle()
+ w.execute(context.Background(), rec, req, *rt, w.active[req.ID])
+ if res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID); err != nil || res != nil {
+ t.Fatalf("result %+v, %v: a box wrote a result for another boot's run", res, err)
+ }
+ if r := currentRun(t, s); h.stops != 0 || r.State != provision.RunPicked || r.BootID != "boot-1" {
+ t.Fatalf("run %+v, stops %d: another boot's run was launched or finished", r, h.stops)
+ }
+ if w.active[req.ID] != nil {
+ t.Error("the run was not forgotten")
+ }
+ for _, mut := range []func(*provision.RunState){
+ func(r *provision.RunState) { r.Box = "box-2" },
+ func(r *provision.RunState) { r.BootID = "boot-2" },
+ func(r *provision.RunState) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
+ } {
+ cur := rec
+ r := *rec.ActiveRun
+ mut(&r)
+ cur.ActiveRun = &r
+ if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); !errors.Is(err, provision.ErrRunConflict) {
+ t.Errorf("run %+v: err %v, want a conflict", r, err)
+ }
+ }
+ cur := rec
+ if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); err != nil || cur.ActiveRun.State != provision.RunDone {
+ t.Fatalf("own run: %v, %+v", err, cur.ActiveRun)
+ }
+}
+
+func TestExitOfReadsNumbersAndWords(t *testing.T) {
+ for _, c := range []struct {
+ code, status string
+ exit int
+ signal string
+ }{
+ {"1", "3", 3, ""}, {"exited", "3", 3, ""}, {"2", "9", 0, "9"}, {"killed", "15", 0, "15"}, {"3", "6", 0, "6"}, {"dumped", "6", 0, "6"},
+ } {
+ exit, signal := exitOf(instancectl.CmdStatus{Code: c.code, Status: c.status})
+ if exit != c.exit || signal != c.signal {
+ t.Errorf("%s/%s: %d %q", c.code, c.status, exit, signal)
+ }
+ }
+}
diff --git a/internal/host/runs.go b/internal/host/runs.go
index 418af4382..9e2fabd12 100644
--- a/internal/host/runs.go
+++ b/internal/host/runs.go
@@ -5,6 +5,7 @@ import (
"crypto/rand"
"encoding/hex"
"errors"
+ "fmt"
"log"
"os"
"sync"
@@ -157,7 +158,9 @@ func (w *runsWorker) consider(ctx context.Context, rec provision.Record) {
// pickup takes a queued run with one CAS from queued to picked. Only the
// box whose CAS lands launches; any other outcome (another box, a cancel,
-// an expiry) leaves the run alone.
+// an expiry) leaves the run alone. A request.json that is not the run the
+// record names, or whose timeout is out of range, is never picked: the
+// expiry ends it.
func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt registry.Route) {
id := rec.ActiveRun.ID
req, err := runs.ReadRequest(ctx, w.store, rec.Account, rec.App, id)
@@ -167,6 +170,10 @@ func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt regist
log.Printf("agent: runs: %s %s: request: %v", rec.Host, id, err)
return
}
+ if err := requestFits(rec, id, *req); err != nil {
+ log.Printf("agent: runs: %s %s: not picked: %v", rec.Host, id, err)
+ return
+ }
h := newRunHandle()
w.mu.Lock()
if w.active[id] != nil {
@@ -178,16 +185,58 @@ func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt regist
got, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host,
provision.RunPick(id, w.box, w.boot, w.now().Add(req.Timeout())))
if err != nil {
- w.forget(id)
- if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
- log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err)
+ if got = w.pickedAnyway(ctx, rec, id); got == nil {
+ w.forget(id)
+ if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err)
+ }
+ return
}
- return
+ log.Printf("agent: runs: %s %s: pickup answered %v, but the record names the run as this box's", rec.Host, id, err)
}
log.Printf("agent: runs: %s: picked run %s", rec.Host, id)
w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })
}
+// requestFits says why request.json is not the run rec names, with a
+// timeout the console would have accepted, or nil.
+func requestFits(rec provision.Record, id string, req runs.Request) error {
+ if req.ID != id || req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host {
+ return fmt.Errorf("request.json names %s/%s/%s %s, not the record's run", req.Account, req.App, req.Host, req.ID)
+ }
+ return runs.CheckTimeout(req.Timeout())
+}
+
+// pickedAnyway reads the record back after a pickup CAS that answered an
+// error: a write whose answer was lost, or a conditional write that landed
+// and then answered a conflict, may have picked the run all the same. The
+// record is what the box launches from when it names the run as picked by
+// this box under this boot; nil otherwise.
+func (w *runsWorker) pickedAnyway(ctx context.Context, rec provision.Record, id string) *provision.Record {
+ for range 3 {
+ rctx, cancel := context.WithTimeout(ctx, runReadBound)
+ cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host)
+ cancel()
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err == nil {
+ if r := cur.ActiveRun; r != nil && r.ID == id && r.Box == w.box && r.BootID == w.boot &&
+ (r.State == provision.RunPicked || r.State == provision.RunCancel) {
+ return cur
+ }
+ return nil
+ }
+ log.Printf("agent: runs: %s %s: read back the pickup: %v", rec.Host, id, err)
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-time.After(runPollRetry):
+ }
+ }
+ return nil
+}
+
func (w *runsWorker) forget(id string) {
w.mu.Lock()
delete(w.active, id)
@@ -218,8 +267,3 @@ func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record)
}
log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id)
}
-
-// execute runs one picked run; Task 5 gives it its body.
-func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
- defer w.forget(req.ID)
-}
diff --git a/internal/host/runs_test.go b/internal/host/runs_test.go
index 2cb483c26..50fe2162c 100644
--- a/internal/host/runs_test.go
+++ b/internal/host/runs_test.go
@@ -3,6 +3,7 @@ package host
import (
"context"
"encoding/json"
+ "errors"
"sync"
"testing"
"time"
@@ -300,3 +301,81 @@ func TestRunsWorkerPicksOnlyForActivatorManagedRowsOfTheRecord(t *testing.T) {
}
}
}
+
+// A request.json that is not the run the record names, or whose timeout is
+// outside 1 s to 6 h, is never picked: the run is left for the expiry.
+func TestRunsWorkerLeavesARequestThatDoesNotMatchItsRecord(t *testing.T) {
+ for name, mut := range map[string]func(*runs.Request){
+ "no timeout": func(r *runs.Request) { r.TimeoutS = 0 },
+ "timeout over 6h": func(r *runs.Request) { r.TimeoutS = 6*3600 + 1 },
+ "another id": func(r *runs.Request) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
+ "another account": func(r *runs.Request) { r.Account = "other" },
+ "another app": func(r *runs.Request) { r.App = "other" },
+ "another host": func(r *runs.Request) { r.Host = "other.go.tito.io" },
+ } {
+ s := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, s, time.Now().Add(time.Minute))
+ req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ mut(req)
+ key := runs.RequestKey(rec.Account, rec.App, id)
+ if err := s.Delete(context.Background(), key); err != nil {
+ t.Fatal(err)
+ }
+ putJSONObject(t, s, key, req)
+ var l execLog
+ w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.consider(context.Background(), rec)
+ if l.count() != 0 || currentRun(t, s).State != provision.RunQueued || len(w.active) != 0 {
+ t.Errorf("%s: picked (execs %d, run %+v)", name, l.count(), currentRun(t, s))
+ }
+ }
+}
+
+// lostAnswer answers every Put of the record with an error; when land is
+// set the write lands first, as a request whose answer was lost does.
+type lostAnswer struct {
+ release.Store
+ land bool
+}
+
+func (l *lostAnswer) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key != provision.RecordKey("bab", "titogo", runHost) {
+ return l.Store.Put(ctx, key, body, ifMatch)
+ }
+ if l.land {
+ if _, err := l.Store.Put(ctx, key, body, ifMatch); err != nil {
+ return "", err
+ }
+ }
+ return "", errors.New("connection reset by peer")
+}
+
+// A pickup CAS whose answer was lost is read back: if the record names
+// this run as this box's under this boot, the box launches it, so the run
+// is not stranded until the sweep calls it lost.
+func TestRunsWorkerLaunchesAPickupWhoseAnswerWasLost(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(&lostAnswer{Store: dir, land: true}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.consider(context.Background(), rec)
+ if r := currentRun(t, dir); l.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.ID != id {
+ t.Fatalf("execs %d, run %+v", l.count(), r)
+ }
+}
+
+// A pickup CAS that failed and did not land is forgotten: the run stays
+// queued and nothing is launched.
+func TestRunsWorkerForgetsAPickupThatDidNotLand(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(&lostAnswer{Store: dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.consider(context.Background(), rec)
+ if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued || len(w.active) != 0 {
+ t.Fatalf("execs %d, run %+v, active %d", l.count(), currentRun(t, dir), len(w.active))
+ }
+}