amadan

carlos / platform Public

cmd/moveprobe: the worker's write failures show in /report and fail /ready

The worker reports each write over a pipe of the main's (its fd 3): ok, or
fail with the error. /report carries the total failed, the failures in a
row and the last error; five failures in a row (five seconds of a worker
that is alive but not writing) fail /ready with a plain reason, which
clears once a write lands again. Review fix round 1, item 1.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 8471c59e7c28bc17fa9ebf3221dac164566860cb parent 9b07ae0
2 files changed, +138 −8
  • cmd/moveprobe/main.go +76 −6
  • cmd/moveprobe/main_test.go +62 −2
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go
index 30d2f23bf..f57355d77 100644
--- a/cmd/moveprobe/main.go
+++ b/cmd/moveprobe/main.go
@@ -32,6 +32,7 @@
package main
import (
+ "bufio"
"bytes"
"context"
"crypto/rand"
@@ -72,7 +73,11 @@ 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
- storeWait = 10 * 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
modeNormal = "normal"
modeBeforeWrite = "before-write"
@@ -211,6 +216,11 @@ 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
}
func openDB(path string) (*sql.DB, error) {
@@ -545,13 +555,23 @@ func (a *app) bgLoop(ctx context.Context) {
// ---- the worker ---------------------------------------------------------------
-// forkWorker starts the worker in its own session with its stdio on
+// forkWorker starts the worker in its own session, with its stdio on
// /dev/null, so it neither shares the main's signals nor holds the pipes the
-// backend reads the main's output from.
+// backend reads the main's output from. A pipe of the main's own (the
+// worker's fd 3) carries the outcome of each of the worker's writes.
func (a *app) forkWorker() {
- cmd := exec.Command(os.Args[0], "worker", "--db", a.dbPath, "--parent", strconv.Itoa(os.Getpid()))
+ status, statusW, err := os.Pipe()
+ if err != nil {
+ logf("worker: %v", err)
+ return
+ }
+ cmd := exec.Command(os.Args[0], "worker", "--db", a.dbPath, "--parent", strconv.Itoa(os.Getpid()), "--status-fd", "3")
cmd.SysProcAttr = detachAttr()
- if err := cmd.Start(); err != nil {
+ cmd.ExtraFiles = []*os.File{statusW}
+ err = cmd.Start()
+ statusW.Close()
+ if err != nil {
+ status.Close()
logf("worker: %v", err)
return
}
@@ -559,6 +579,34 @@ func (a *app) forkWorker() {
a.workerPID = cmd.Process.Pid
a.mu.Unlock()
go cmd.Wait()
+ go a.readWorkerStatus(status)
+}
+
+// readWorkerStatus reads the worker's lines: "ok", or "fail <error>".
+func (a *app) readWorkerStatus(r *os.File) {
+ defer r.Close()
+ sc := bufio.NewScanner(r)
+ for sc.Scan() {
+ line := sc.Text()
+ a.mu.Lock()
+ if msg, failed := strings.CutPrefix(line, "fail "); failed {
+ a.workerFailed++
+ a.workerInARow++
+ a.workerLastErr = msg
+ } else {
+ a.workerInARow = 0
+ }
+ a.mu.Unlock()
+ }
+}
+
+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)
+ }
+ return nil
}
// runWorker writes a worker row every second. Once its parent has gone it
@@ -567,12 +615,27 @@ func runWorker(args []string) int {
fl := flag.NewFlagSet("moveprobe worker", flag.ContinueOnError)
dbPath := fl.String("db", "", "the instance database")
parent := fl.Int("parent", 0, "the main's pid")
+ statusFD := fl.Int("status-fd", 0, "a pipe to report each write on (0: none)")
if err := fl.Parse(args); err != nil || *dbPath == "" || *parent <= 0 {
return 2
}
here, _ := region()
+ // Writes to the pipe fail once the main has gone; they are ignored.
+ report := func(err error) {}
+ if *statusFD > 0 {
+ status := os.NewFile(uintptr(*statusFD), "status")
+ defer status.Close()
+ report = func(err error) {
+ if err != nil {
+ fmt.Fprintf(status, "fail %s\n", strings.ReplaceAll(err.Error(), "\n", " "))
+ } else {
+ fmt.Fprintln(status, "ok")
+ }
+ }
+ }
db, err := openDB(*dbPath)
if err != nil {
+ report(err)
return 1
}
defer db.Close()
@@ -586,7 +649,7 @@ func runWorker(args []string) int {
if !orphaned.IsZero() && time.Since(orphaned) >= workerGrace {
return 0
}
- insertSeq(db, "worker", here)
+ report(insertSeq(db, "worker", here))
<-t.C
}
}
@@ -637,6 +700,9 @@ func (a *app) readiness(ctx context.Context) (int, string) {
if err := a.checkStore(ctx); err != nil {
fails = append(fails, err.Error())
}
+ if err := a.checkWorker(); err != nil {
+ fails = append(fails, err.Error())
+ }
if len(fails) > 0 {
return http.StatusInternalServerError, strings.Join(fails, "; ")
}
@@ -762,6 +828,9 @@ type report struct {
Object objectReport `json:"object"`
HasSecret bool `json:"has_secret"`
WorkerPID int `json:"worker_pid"`
+ WorkerFailed int64 `json:"worker_writes_failed"`
+ WorkerInARow int64 `json:"worker_failed_in_a_row"`
+ WorkerLastError string `json:"worker_last_error"`
}
func (a *app) handleReport(w http.ResponseWriter, r *http.Request) {
@@ -773,6 +842,7 @@ 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
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 a66e0e6d6..53193e14d 100644
--- a/cmd/moveprobe/main_test.go
+++ b/cmd/moveprobe/main_test.go
@@ -362,8 +362,11 @@ type reportDoc struct {
Body string `json:"body"`
Error string `json:"error"`
} `json:"object"`
- HasSecret bool `json:"has_secret"`
- WorkerPID int `json:"worker_pid"`
+ 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"`
}
func (r reportDoc) count(kind, region string) int64 {
@@ -1016,3 +1019,60 @@ func TestReadyChecksBothKeys(t *testing.T) {
check("platform key absent", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": ""}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
check("both restored", f.env, http.StatusOK, "")
}
+
+// ---- write failures -------------------------------------------------------
+
+// failInserts makes every insert of kind into rows abort, through a trigger
+// the test adds to the instance's own database: a real write failure, with
+// nothing in the app to switch it on.
+func failInserts(t *testing.T, db, kind string) {
+ t.Helper()
+ d := testDB(t, db)
+ defer d.Close()
+ q := fmt.Sprintf(`CREATE TRIGGER fail_%[1]s BEFORE INSERT ON rows WHEN NEW.kind = '%[1]s' BEGIN SELECT RAISE(ABORT, 'forced %[1]s failure'); END`, kind)
+ if _, err := d.Exec(q); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func allowInserts(t *testing.T, db, kind string) {
+ t.Helper()
+ d := testDB(t, db)
+ defer d.Close()
+ if _, err := d.Exec(`DROP TRIGGER fail_` + kind); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestWorkerWriteFailuresFailReady(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.env)
+ p.wait("/ready", http.StatusOK)
+
+ 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 {
+ break
+ }
+ time.Sleep(200 * 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)
+ }
+
+ // Once the worker's writes land again, readiness comes back; the total
+ // stays in the report.
+ 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)
+ }
+ p.stop()
+}