amadan

carlos / platform Public

cmd/moveprobe: /ready needs a worker row since start and a good last write

Replaces the five-in-a-row threshold: /ready answers 500 until at least one
worker row has committed since this start, and whenever the worker's latest
write failed; the next good write clears it. Before-write mode has no
worker and keeps its own answer. /report keeps the counters and adds
worker_writes_committed. Review fix round 2, item 1.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9d5949fcc6594ff6ab23a8552e233e2bcad0edd6 parent 066e838
3 files changed, +77 −39
  • cmd/moveprobe/README.md +11 −4
  • cmd/moveprobe/main.go +22 −13
  • cmd/moveprobe/main_test.go +44 −22
diff --git a/cmd/moveprobe/README.md b/cmd/moveprobe/README.md
index 8eaed9b22..710a428e3 100644
--- a/cmd/moveprobe/README.md
+++ b/cmd/moveprobe/README.md
@@ -110,8 +110,10 @@ instance started. Once the app has gone, the worker keeps writing for 60
seconds and then exits by itself.
The worker tells the app whether each write landed. `/report` shows how many
-failed, how many of the latest failed in a row, and the last error. After
-five failures in a row, `/ready` fails until a write lands again.
+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.
## Strings
@@ -146,8 +148,13 @@ On `/` and `/ready`:
(500).
- `store: <key> reads back different bytes`: the object came back changed
(500).
-- `worker: <n> writes failed in a row: <error>`: the worker is alive but
- its last five or more writes failed (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).
+- `worker: no row committed since start; last error: <error>`: as above,
+ and its writes are failing (500).
+- `worker: the last write failed: <error>`: the worker's latest write
+ failed (500).
On `/report`:
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go
index 262301f24..e8d8e7380 100644
--- a/cmd/moveprobe/main.go
+++ b/cmd/moveprobe/main.go
@@ -73,11 +73,7 @@ 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
- // workerFailLimit is how many of the worker's writes may fail in a row
- // before /ready fails: five seconds of a worker that is alive but not
- // writing. Any failure at all shows in /report.
- workerFailLimit = 5
- storeWait = 10 * time.Second
+ storeWait = 10 * time.Second
modeNormal = "normal"
modeBeforeWrite = "before-write"
@@ -219,11 +215,13 @@ type app struct {
mu sync.Mutex
objPut bool
workerPID int
- // The worker reports every write over a pipe: how many failed, how many
- // of the latest failed in a row, and the last error.
- workerFailed int64
- workerInARow int64
- workerLastErr string
+ // The worker reports every write over a pipe: how many committed and
+ // how many failed since this start, how many of the latest failed in a
+ // row, and the last error.
+ workerCommitted int64
+ workerFailed int64
+ workerInARow int64
+ workerLastErr string
// control is the one client for the instance control socket, nil when
// the agent delivered none.
@@ -609,17 +607,26 @@ func (a *app) readWorkerStatus(r *os.File) {
a.workerInARow++
a.workerLastErr = msg
} else {
+ a.workerCommitted++
a.workerInARow = 0
}
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.
func (a *app) checkWorker() error {
a.mu.Lock()
defer a.mu.Unlock()
- if a.workerInARow >= workerFailLimit {
- return fmt.Errorf("worker: %d writes failed in a row: %s", a.workerInARow, a.workerLastErr)
+ switch {
+ 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:
+ return errors.New("worker: no row committed since start")
+ case a.workerInARow > 0:
+ return fmt.Errorf("worker: the last write failed: %s", a.workerLastErr)
}
return nil
}
@@ -840,6 +847,7 @@ type report struct {
Object objectReport `json:"object"`
HasSecret bool `json:"has_secret"`
WorkerPID int `json:"worker_pid"`
+ WorkerCommitted int64 `json:"worker_writes_committed"`
WorkerFailed int64 `json:"worker_writes_failed"`
WorkerInARow int64 `json:"worker_failed_in_a_row"`
WorkerLastError string `json:"worker_last_error"`
@@ -899,7 +907,8 @@ func (a *app) handleReport(w http.ResponseWriter, r *http.Request) {
}
a.mu.Lock()
rep.WorkerPID = a.workerPID
- rep.WorkerFailed, rep.WorkerInARow, rep.WorkerLastError = a.workerFailed, a.workerInARow, a.workerLastErr
+ rep.WorkerCommitted, rep.WorkerFailed = a.workerCommitted, a.workerFailed
+ rep.WorkerInARow, rep.WorkerLastError = a.workerInARow, a.workerLastErr
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 8d28ee449..35ec45757 100644
--- a/cmd/moveprobe/main_test.go
+++ b/cmd/moveprobe/main_test.go
@@ -370,11 +370,12 @@ type reportDoc struct {
Body string `json:"body"`
Error string `json:"error"`
} `json:"object"`
- HasSecret bool `json:"has_secret"`
- WorkerPID int `json:"worker_pid"`
- WorkerWritesFailed int64 `json:"worker_writes_failed"`
- WorkerFailedInARow int64 `json:"worker_failed_in_a_row"`
- WorkerLastError string `json:"worker_last_error"`
+ HasSecret bool `json:"has_secret"`
+ WorkerPID int `json:"worker_pid"`
+ WorkerWritesFailed int64 `json:"worker_writes_failed"`
+ WorkerWritesCommitted int64 `json:"worker_writes_committed"`
+ WorkerFailedInARow int64 `json:"worker_failed_in_a_row"`
+ WorkerLastError string `json:"worker_last_error"`
}
func (r reportDoc) count(kind, region string) int64 {
@@ -1052,35 +1053,56 @@ func allowInserts(t *testing.T, db, kind string) {
}
}
-func TestWorkerWriteFailuresFailReady(t *testing.T) {
+// TestReadyFollowsTheWorkersLastWrite: /ready is 200 only once a worker row
+// has committed since this start and the worker's latest write succeeded.
+func TestReadyFollowsTheWorkersLastWrite(t *testing.T) {
f := newFixture(t)
+ seed := f.start(f.env)
+ seed.wait("/ready", http.StatusOK)
+ seed.stop()
+ killWorkers(f.db)
+
+ // Worker rows from the earlier boot are in the database; none can
+ // commit in this one.
+ failInserts(t, f.db, "worker")
p := f.start(f.env)
+ body := p.wait("/ready", http.StatusInternalServerError)
+ if !strings.Contains(body, "worker: no row committed since start") {
+ t.Errorf("/ready before the first worker row: %q", body)
+ }
+ time.Sleep(1500 * time.Millisecond)
+ if code, body, _ := p.do(http.MethodGet, "/ready", "", nil); code != http.StatusInternalServerError {
+ t.Errorf("/ready with no worker row yet: %d %q, want 500", code, body)
+ }
+ r := p.report()
+ if r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted != 0 || !strings.Contains(r.WorkerLastError, "forced worker failure") {
+ t.Errorf("report: committed %d, failed %d, last %q", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerLastError)
+ }
+
+ // The first good write makes it ready.
+ allowInserts(t, f.db, "worker")
p.wait("/ready", http.StatusOK)
+ // One failed write is enough to fail it.
failInserts(t, f.db, "worker")
- deadline := time.Now().Add(10 * time.Second)
- var r reportDoc
- for time.Now().Before(deadline) {
- if r = p.report(); r.WorkerWritesFailed > 0 {
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ r = p.report()
+ if r.WorkerFailedInARow > 0 || time.Now().After(deadline) {
break
}
- time.Sleep(200 * time.Millisecond)
+ time.Sleep(20 * time.Millisecond)
}
- if r.WorkerWritesFailed == 0 || r.WorkerFailedInARow == 0 || !strings.Contains(r.WorkerLastError, "forced worker failure") {
- t.Fatalf("the worker's failed writes are not in the report: failed %d, in a row %d, last %q",
- r.WorkerWritesFailed, r.WorkerFailedInARow, r.WorkerLastError)
- }
- body := p.wait("/ready", http.StatusInternalServerError)
- if !strings.Contains(body, "worker") || !strings.Contains(body, "in a row") {
- t.Errorf("/ready body %q does not say the worker's writes are failing", body)
+ code, body, _ := p.do(http.MethodGet, "/ready", "", nil)
+ if r.WorkerFailedInARow != 1 || code != http.StatusInternalServerError || !strings.Contains(body, "worker: the last write failed") {
+ t.Errorf("after one failed write (in a row %d): /ready %d %q, want 500", r.WorkerFailedInARow, code, body)
}
- // Once the worker's writes land again, readiness comes back; the total
- // stays in the report.
+ // The next good write makes it ready again; the counters stay.
allowInserts(t, f.db, "worker")
p.wait("/ready", http.StatusOK)
- if r := p.report(); r.WorkerFailedInARow != 0 || r.WorkerWritesFailed == 0 {
- t.Errorf("after recovery: failed %d, in a row %d", r.WorkerWritesFailed, r.WorkerFailedInARow)
+ if r := p.report(); r.WorkerFailedInARow != 0 || r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted == 0 {
+ t.Errorf("after recovery: committed %d, failed %d, in a row %d", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerFailedInARow)
}
p.stop()
}