amadan

carlos / platform Public

host: bound a run's staging by its deadline; no result for a run Prepare found not ours (review round 1)

Prepare runs under the activator's hold, and a stalled download held it
(and every later run on the instance) forever: it now ends at the run's
deadline from the record, as a failed start with a result and the done
CAS. Prepare records its own not-ours verdict, so a stop overlapping it
cannot make the box write a result for another boot's run. Pickup logs a
refused request.json once per run. The staging tests move to a linux test
file; the never-starts tests fail at once.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b225ba1b498d736ef67b8d2cbe2b7f22906c6bac parent 1b41f04
4 files changed, +169 −16
  • internal/host/runexec.go +23 −4
  • internal/host/{runexec_test.go => runexec_linux_test.go} +90 −10
  • internal/host/runs.go +24 −2
  • internal/host/runs_test.go +32 −0

CI passed — run details

diff --git a/internal/host/runexec.go b/internal/host/runexec.go
index 6e7cb5604..d0c83a5f5 100644
--- a/internal/host/runexec.go
+++ b/internal/host/runexec.go
@@ -138,6 +138,10 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
}
res.LastSeq, res.Truncated = last, truncated
}
+ // notOurs is prepare's own verdict: RunCmd reports a stop that
+ // overlapped Prepare in place of Prepare's error, and a run the record
+ // no longer gives this box must get no result whatever RunCmd says.
+ notOurs := false
spec := activator.CmdSpec{
RunID: req.ID,
Args: req.Argv,
@@ -145,14 +149,16 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
Stdout: chunks.Stream(runs.StreamOut),
Stderr: chunks.Stream(runs.StreamErr),
Prepare: func(ctx context.Context) error {
- return w.prepare(ctx, instDir, req, h)
+ err := w.prepare(ctx, instDir, req, h, launchDeadline(rec, req, w.now()))
+ notOurs = errors.Is(err, errRunNotOurs)
+ return err
},
}
handle, err := w.launch.RunCmd(ctx, rt, spec)
if err != nil {
bound := time.AfterFunc(runOutputGrace, endOutput)
defer bound.Stop()
- if errors.Is(err, errRunNotOurs) {
+ if notOurs || errors.Is(err, errRunNotOurs) {
// The record gives the run to no one, or to another boot:
// whoever holds it writes its result.
chunks.Close()
@@ -200,13 +206,26 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
w.finish(ctx, rec, req, res, handle)
}
+// launchDeadline is the run's deadline from the record (RunPick set it to
+// the pickup plus the timeout), or now plus the timeout without one.
+func launchDeadline(rec provision.Record, req runs.Request, now time.Time) time.Time {
+ if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() {
+ return r.Deadline
+ }
+ return now.Add(req.Timeout())
+}
+
// 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 {
+// that lands while the files download still starts no unit. All of it ends
+// at the run's deadline: a stalled download must not keep the hold (and
+// with it every later run on the instance) forever.
+func (w *runsWorker) prepare(ctx context.Context, instDir string, req runs.Request, h *runHandle, deadline time.Time) error {
+ ctx, cancel := context.WithDeadline(ctx, deadline)
+ defer cancel()
if err := w.launchable(ctx, req, h); err != nil {
return err
}
diff --git a/internal/host/runexec_test.go b/internal/host/runexec_linux_test.go
similarity index 82%
rename from internal/host/runexec_test.go
rename to internal/host/runexec_linux_test.go
index 431ae87c0..43d965c58 100644
--- a/internal/host/runexec_test.go
+++ b/internal/host/runexec_linux_test.go
@@ -57,15 +57,21 @@ func (h *fakeHandle) Stop(context.Context) error {
}
func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }
+// fakeLauncher counts the units it starts. With noUnit set it starts none:
+// a test that must start no unit fails at once instead of following one.
type fakeLauncher struct {
mu sync.Mutex
calls int
+ units int
+ noUnit bool
err error
handle *fakeHandle
out string
errOut string
}
+func (l *fakeLauncher) started() int { l.mu.Lock(); defer l.mu.Unlock(); return l.units }
+
func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
l.mu.Lock()
l.calls++
@@ -78,6 +84,12 @@ func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activ
return nil, err
}
}
+ l.mu.Lock()
+ l.units++
+ l.mu.Unlock()
+ if l.noUnit {
+ return nil, errors.New("fake: a unit was started")
+ }
io.WriteString(spec.Stdout, l.out)
io.WriteString(spec.Stderr, l.errOut)
return l.handle, nil
@@ -248,10 +260,10 @@ func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
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)
+ l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
+ res := runWith(t, s, l, rec, req, newRunHandle())
+ if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
+ t.Fatalf("result %+v, units started %d", res, l.started())
}
if r := currentRun(t, s); r.State != provision.RunDone {
t.Fatalf("run %+v, want done", r)
@@ -284,11 +296,10 @@ func TestRunCancelledWhileFilesDownloadNeverStarts(t *testing.T) {
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}
+ l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
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)
+ if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
+ t.Fatalf("result %+v, units started %d: the record is read again after the download", res, l.started())
}
}
@@ -411,11 +422,17 @@ func TestRunSeesACancelByRereadingTheRecord(t *testing.T) {
s := release.NewDirStore(t.TempDir())
rec, req := pickedRun(t, s, 60)
h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ errc := make(chan error, 1)
go func() {
time.Sleep(50 * time.Millisecond)
- provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID))
+ _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID))
+ errc <- err
}()
- if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); !res.Cancelled {
+ res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
+ if err := <-errc; err != nil {
+ t.Fatal(err)
+ }
+ if !res.Cancelled {
t.Fatalf("result %+v", res)
}
}
@@ -508,6 +525,69 @@ func TestRunFinishesOnlyItsOwnPick(t *testing.T) {
}
}
+// stalledInput blocks every read of a run input until its context ends.
+type stalledInput struct{ release.Store }
+
+func (s stalledInput) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "/in/") {
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+ }
+ return s.Store.Get(ctx, key)
+}
+
+// A download that stalls ends at the run's deadline: the run ends as a
+// failed start, with its result and the done CAS, and does not keep the
+// activator's hold (and every later run on the instance) forever.
+func TestRunStagingEndsAtTheRunsDeadline(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
+ req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Files = []runs.File{{Name: "in.csv", Size: 4, SHA256: strings.Repeat("0", 64)}}
+ got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
+ start := time.Now()
+ res := runWith(t, stalledInput{dir}, l, *got, *req, newRunHandle())
+ if res.Interrupted != runs.InterruptedStartFailed || l.started() != 0 || time.Since(start) > 5*time.Second {
+ t.Fatalf("result %+v after %s, units %d", res, time.Since(start), l.started())
+ }
+ if r := currentRun(t, dir); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+// stoppedDuringPrepare runs Prepare and then reports an instance stop, as
+// RunCmd does when a stop overlapped Prepare: the stop wins over Prepare's
+// own error.
+type stoppedDuringPrepare struct{}
+
+func (stoppedDuringPrepare) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
+ _ = spec.Prepare(ctx)
+ return nil, &activator.CmdStoppedError{Reason: activator.StopRestart}
+}
+
+// A run Prepare found is no longer this box's gets no result, even when a
+// stop overlapping Prepare is what RunCmd reports.
+func TestRunNotOursGetsNoResultWhenAStopOverlapsPrepare(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60) // picked under boot-1
+ w := execWorker(t, s, stoppedDuringPrepare{}, "boot-2")
+ rt, _ := runOwnedRoute(runHost)(runHost)
+ w.execute(context.Background(), rec, req, *rt, newRunHandle())
+ 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 a run the record gives another boot", res, err)
+ }
+ if r := currentRun(t, s); r.State != provision.RunPicked || r.BootID != "boot-1" {
+ t.Fatalf("run %+v", r)
+ }
+}
+
func TestExitOfReadsNumbersAndWords(t *testing.T) {
for _, c := range []struct {
code, status string
diff --git a/internal/host/runs.go b/internal/host/runs.go
index 9e2fabd12..c906e77ce 100644
--- a/internal/host/runs.go
+++ b/internal/host/runs.go
@@ -55,11 +55,12 @@ type runsWorker struct {
mu sync.Mutex
noted map[string]provision.Record // host -> the record reconcile read
active map[string]*runHandle // run id -> the run this agent is executing
+ unfit map[string]bool // run ids whose request.json was refused, logged once
}
func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker {
w := &runsWorker{store: store, routes: routes, launch: launch, dataDir: dataDir, box: box, boot: boot,
- now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{},
+ now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{}, unfit: map[string]bool{},
start: func(f func()) { go f() }}
w.exec = w.execute
return w
@@ -171,7 +172,9 @@ func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt regist
return
}
if err := requestFits(rec, id, *req); err != nil {
- log.Printf("agent: runs: %s %s: not picked: %v", rec.Host, id, err)
+ if w.firstUnfit(id) {
+ log.Printf("agent: runs: %s %s: not picked: %v", rec.Host, id, err)
+ }
return
}
h := newRunHandle()
@@ -198,6 +201,25 @@ func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt regist
w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })
}
+// unfitLogMax caps the ids firstUnfit remembers; past it the set starts
+// again, and a refusal still queued is logged once more.
+const unfitLogMax = 256
+
+// firstUnfit reports whether this is the first refusal of id's request,
+// so a run left for the expiry is logged once, not on every pass.
+func (w *runsWorker) firstUnfit(id string) bool {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.unfit[id] {
+ return false
+ }
+ if len(w.unfit) >= unfitLogMax {
+ clear(w.unfit)
+ }
+ w.unfit[id] = true
+ return true
+}
+
// 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 {
diff --git a/internal/host/runs_test.go b/internal/host/runs_test.go
index 50fe2162c..d6b0e8c9f 100644
--- a/internal/host/runs_test.go
+++ b/internal/host/runs_test.go
@@ -1,9 +1,13 @@
package host
import (
+ "bytes"
"context"
"encoding/json"
"errors"
+ "log"
+ "os"
+ "strings"
"sync"
"testing"
"time"
@@ -334,6 +338,34 @@ func TestRunsWorkerLeavesARequestThatDoesNotMatchItsRecord(t *testing.T) {
}
}
+// A refused request.json is logged once, not on every pass that sees the
+// run still queued.
+func TestRunsWorkerLogsAnUnfitRequestOnce(t *testing.T) {
+ 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)
+ }
+ req.TimeoutS = 0
+ 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 buf bytes.Buffer
+ log.SetOutput(&buf)
+ defer log.SetOutput(os.Stderr)
+ var l execLog
+ w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ for range 3 {
+ w.consider(context.Background(), rec)
+ }
+ if n := strings.Count(buf.String(), "not picked"); n != 1 {
+ t.Fatalf("logged %d times, want once:\n%s", n, buf.String())
+ }
+}
+
// 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 {