amadan
cmd/moveprobe: a dead or silent worker fails /ready
The main now records the worker's exit (from Wait, or the status pipe closing first) and the time of its latest status line. /ready answers 500 once the worker has exited or did not start, and when it has sent no status for 3 s; /report carries worker_exit and worker_last_status_at. The README's Strings section lists the new bodies and fields. Final review I2.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+104
−8
cmd/moveprobe/README.md+17 −4cmd/moveprobe/main.go+39 −4cmd/moveprobe/main_test.go+48 −0
diff --git a/cmd/moveprobe/README.md b/cmd/moveprobe/README.md| index a64800515..e729077aa 100644 |
| --- a/cmd/moveprobe/README.md |
| +++ b/cmd/moveprobe/README.md |
| @@ -111,9 +111,11 @@ seconds and then exits by itself. |
| The worker tells the app whether each write landed. `/report` shows how many |
| committed and how many failed since the app started, how many of the latest |
| -failed in a row, and the last error. `/ready` fails until a worker row has |
| -committed since the app started, and fails again whenever the worker's |
| -latest write failed. |
| +failed in a row, the last error, when it last heard from the worker, and |
| +why the worker is gone if it is. `/ready` fails until a worker row has |
| +committed since the app started, whenever the worker's latest write failed, |
| +if the worker has sent nothing for 3 seconds, and once the worker has |
| +exited. |
| ## Strings |
| @@ -148,6 +150,13 @@ On `/` and `/ready`: |
| (500). |
| - `store: <key> reads back different bytes`: the object came back changed |
| (500). |
| +- `worker: exited (<how>)`: the worker has exited. `<how>` is how it |
| + ended, such as `exit status 1` or `signal: killed`, or `its status pipe |
| + closed` when the app saw it go before learning how (500). |
| +- `worker: did not start: <error>`: the app could not start the worker |
| + (500). |
| +- `worker: no status for <n>s`: the worker has not reported a write for 3 |
| + seconds or more, for example because it is stuck (500). |
| - `worker: no row committed since start`: the worker has not written a row |
| since the app started, for example in its first moments or if it did not |
| start (500). |
| @@ -164,7 +173,8 @@ On `/report`: |
| `count`), `file_key_sha256`, `platform_key_sha256`, `object` (with |
| `key`, `body` and, when it could not be read, `error`), `has_secret`, |
| `worker_pid`, `worker_writes_committed`, `worker_writes_failed`, |
| - `worker_failed_in_a_row` and `worker_last_error`. |
| + `worker_failed_in_a_row`, `worker_last_error`, `worker_exit` and |
| + `worker_last_status_at` (unix seconds, 0 before the first). |
| - `report: <error>`: a database read failed (500). |
| The values the app writes into the report itself: |
| @@ -186,6 +196,9 @@ The values the app writes into the report itself: |
| missing, or `get <key>: <error>` when the object cannot be read back. |
| - `worker_last_error`: the database's own error text for the worker's |
| latest failed write. The app adds no words of its own. |
| +- `worker_exit`: empty while the worker runs; otherwise |
| + `exited (<how>)` or `did not start: <error>`, as on `/ready` without the |
| + `worker: ` prefix. |
| On `/arm-once`: |
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go| index e8d8e7380..a548d257c 100644 |
| --- a/cmd/moveprobe/main.go |
| +++ b/cmd/moveprobe/main.go |
| @@ -73,6 +73,9 @@ const ( |
| // evidence (seconds); short enough that an ordinary sleep, which stops |
| // only the main, does not leave it writing for ever. |
| workerGrace = 60 * time.Second |
| + // workerStale is how long the main waits for the worker's next status |
| + // line (one a second) before readiness fails. |
| + workerStale = 3 * time.Second |
| storeWait = 10 * time.Second |
| modeNormal = "normal" |
| @@ -222,6 +225,8 @@ type app struct { |
| workerFailed int64 |
| workerInARow int64 |
| workerLastErr string |
| + workerLastAt time.Time // the latest status line |
| + workerExit string // why the worker is gone; empty while it runs |
| // control is the one client for the instance control socket, nil when |
| // the agent delivered none. |
| @@ -586,12 +591,24 @@ func (a *app) forkWorker() { |
| if err != nil { |
| status.Close() |
| logf("worker: %v", err) |
| + a.mu.Lock() |
| + a.workerExit = "did not start: " + err.Error() |
| + a.mu.Unlock() |
| return |
| } |
| a.mu.Lock() |
| a.workerPID = cmd.Process.Pid |
| a.mu.Unlock() |
| - go cmd.Wait() |
| + go func() { |
| + err := cmd.Wait() |
| + how := "exit status 0" |
| + if err != nil { |
| + how = err.Error() |
| + } |
| + a.mu.Lock() |
| + a.workerExit = "exited (" + how + ")" |
| + a.mu.Unlock() |
| + }() |
| go a.readWorkerStatus(status) |
| } |
| @@ -602,6 +619,7 @@ func (a *app) readWorkerStatus(r *os.File) { |
| for sc.Scan() { |
| line := sc.Text() |
| a.mu.Lock() |
| + a.workerLastAt = time.Now() |
| if msg, failed := strings.CutPrefix(line, "fail "); failed { |
| a.workerFailed++ |
| a.workerInARow++ |
| @@ -612,15 +630,26 @@ func (a *app) readWorkerStatus(r *os.File) { |
| } |
| a.mu.Unlock() |
| } |
| + // The pipe closes when the worker exits; Wait's own account, when it |
| + // comes, replaces this one. |
| + a.mu.Lock() |
| + if a.workerExit == "" { |
| + a.workerExit = "exited (its status pipe closed)" |
| + } |
| + a.mu.Unlock() |
| } |
| -// checkWorker holds readiness to the worker: at least one of its rows has |
| -// committed since this start, and its latest write succeeded. A worker that |
| -// never started has committed nothing, so it fails too. |
| +// checkWorker holds readiness to the worker: it is still running and has |
| +// sent a status line within workerStale, at least one of its rows has |
| +// committed since this start, and its latest write succeeded. |
| func (a *app) checkWorker() error { |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| switch { |
| + case a.workerExit != "": |
| + return errors.New("worker: " + a.workerExit) |
| + case !a.workerLastAt.IsZero() && time.Since(a.workerLastAt) > workerStale: |
| + return fmt.Errorf("worker: no status for %ds", int(time.Since(a.workerLastAt).Seconds())) |
| case a.workerCommitted == 0 && a.workerInARow > 0: |
| return fmt.Errorf("worker: no row committed since start; last error: %s", a.workerLastErr) |
| case a.workerCommitted == 0: |
| @@ -851,6 +880,8 @@ type report struct { |
| WorkerFailed int64 `json:"worker_writes_failed"` |
| WorkerInARow int64 `json:"worker_failed_in_a_row"` |
| WorkerLastError string `json:"worker_last_error"` |
| + WorkerExit string `json:"worker_exit"` |
| + WorkerLastStatus int64 `json:"worker_last_status_at"` |
| } |
| // counts reads the row counts and the timer fires. Any failed query, scan |
| @@ -909,6 +940,10 @@ func (a *app) handleReport(w http.ResponseWriter, r *http.Request) { |
| rep.WorkerPID = a.workerPID |
| rep.WorkerCommitted, rep.WorkerFailed = a.workerCommitted, a.workerFailed |
| rep.WorkerInARow, rep.WorkerLastError = a.workerInARow, a.workerLastErr |
| + rep.WorkerExit = a.workerExit |
| + if !a.workerLastAt.IsZero() { |
| + rep.WorkerLastStatus = a.workerLastAt.Unix() |
| + } |
| a.mu.Unlock() |
| if k, err := a.fileKeyNow(); err == nil { |
| rep.FileKeySHA256 = fingerprint(k) |
diff --git a/cmd/moveprobe/main_test.go b/cmd/moveprobe/main_test.go| index 35ec45757..ff7624a30 100644 |
| --- a/cmd/moveprobe/main_test.go |
| +++ b/cmd/moveprobe/main_test.go |
| @@ -376,6 +376,8 @@ type reportDoc struct { |
| WorkerWritesCommitted int64 `json:"worker_writes_committed"` |
| WorkerFailedInARow int64 `json:"worker_failed_in_a_row"` |
| WorkerLastError string `json:"worker_last_error"` |
| + WorkerExit string `json:"worker_exit"` |
| + WorkerLastStatusAt int64 `json:"worker_last_status_at"` |
| } |
| func (r reportDoc) count(kind, region string) int64 { |
| @@ -1245,3 +1247,49 @@ func TestArmOnceReusesOneControlConnection(t *testing.T) { |
| } |
| p.stop() |
| } |
| + |
| +// TestReadyFailsWhenTheWorkerStops: a worker that has died, or has sent no |
| +// status for 3 s, fails readiness, and the report says so. |
| +func TestReadyFailsWhenTheWorkerStops(t *testing.T) { |
| + t.Run("the worker dies after its first good write", func(t *testing.T) { |
| + f := newFixture(t) |
| + p := f.start(f.env) |
| + p.wait("/ready", http.StatusOK) |
| + r := p.report() |
| + if r.WorkerWritesCommitted == 0 || r.WorkerExit != "" || r.WorkerLastStatusAt == 0 { |
| + t.Fatalf("before: committed %d, exit %q, last status %d", r.WorkerWritesCommitted, r.WorkerExit, r.WorkerLastStatusAt) |
| + } |
| + if err := syscall.Kill(r.WorkerPID, syscall.SIGKILL); err != nil { |
| + t.Fatal(err) |
| + } |
| + body := p.wait("/ready", http.StatusInternalServerError) |
| + if !strings.Contains(body, "worker: exited") { |
| + t.Errorf("/ready after the worker died: %q", body) |
| + } |
| + if r := p.report(); !strings.Contains(r.WorkerExit, "exited") { |
| + t.Errorf("report worker_exit %q after the worker died", r.WorkerExit) |
| + } |
| + p.stop() |
| + }) |
| + |
| + t.Run("the worker sends no status for 3 s", func(t *testing.T) { |
| + f := newFixture(t) |
| + p := f.start(f.env) |
| + p.wait("/ready", http.StatusOK) |
| + pid := p.report().WorkerPID |
| + if err := syscall.Kill(pid, syscall.SIGSTOP); err != nil { |
| + t.Fatal(err) |
| + } |
| + stopped := time.Now() |
| + body := p.wait("/ready", http.StatusInternalServerError) |
| + if waited := time.Since(stopped); waited < 2*time.Second { |
| + t.Errorf("/ready failed %s after the worker stopped, before 3 s without status", waited) |
| + } |
| + if !strings.Contains(body, "worker: no status for") { |
| + t.Errorf("/ready with a stalled worker: %q", body) |
| + } |
| + syscall.Kill(pid, syscall.SIGCONT) |
| + p.wait("/ready", http.StatusOK) |
| + p.stop() |
| + }) |
| +} |