amadan

carlos / platform Public

cmd/moveprobe: /report answers 500 on any failed query, never a partial 200

Both count queries run through one helper that checks rows.Err after the
loop, so a read that fails part way is an error rather than a shorter
list. Tested with a database/sql driver whose rows fail after one good row,
for each query, and with a closed database. Review fix round 1, item 3.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 51316d0c344429363b38641184230b039ee8d772 parent 6e3f525
2 files changed, +147 −29
  • cmd/moveprobe/main.go +49 −29
  • cmd/moveprobe/main_test.go +98 −0
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go
index 04a96b6e6..0e4f9e625 100644
--- a/cmd/moveprobe/main.go
+++ b/cmd/moveprobe/main.go
@@ -836,10 +836,55 @@ type report struct {
WorkerLastError string `json:"worker_last_error"`
}
+// counts reads the row counts and the timer fires. Any failed query, scan
+// or read part way through is an error, never a shorter list.
+func (a *app) counts(ctx context.Context) ([]rowCount, []firedCount, error) {
+ rowsOut := []rowCount{}
+ err := a.query(ctx, `SELECT kind, region, count(*), min(seq), max(seq) FROM rows GROUP BY kind, region ORDER BY kind, region`,
+ func(rows *sql.Rows) error {
+ var c rowCount
+ if err := rows.Scan(&c.Kind, &c.Region, &c.Count, &c.MinSeq, &c.MaxSeq); err != nil {
+ return err
+ }
+ rowsOut = append(rowsOut, c)
+ return nil
+ })
+ if err != nil {
+ return nil, nil, err
+ }
+ fired := []firedCount{}
+ err = a.query(ctx, `SELECT kind, name, due_at, count(*) FROM rows WHERE kind IN ('tick', 'once') GROUP BY kind, name, due_at ORDER BY kind, name, due_at`,
+ func(rows *sql.Rows) error {
+ var c firedCount
+ if err := rows.Scan(&c.Kind, &c.Name, &c.DueAt, &c.Count); err != nil {
+ return err
+ }
+ fired = append(fired, c)
+ return nil
+ })
+ if err != nil {
+ return nil, nil, err
+ }
+ return rowsOut, fired, nil
+}
+
+func (a *app) query(ctx context.Context, q string, each func(*sql.Rows) error) error {
+ rows, err := a.db.QueryContext(ctx, q)
+ if err != nil {
+ return err
+ }
+ defer rows.Close()
+ for rows.Next() {
+ if err := each(rows); err != nil {
+ return err
+ }
+ }
+ return rows.Err()
+}
+
func (a *app) handleReport(w http.ResponseWriter, r *http.Request) {
rep := report{
Host: a.host, Region: a.region, RegionFrom: a.regionFrom, Mode: a.mode, BootID: a.bootID,
- Rows: []rowCount{}, Fired: []firedCount{},
PlatformKeySHA256: a.platformFP,
HasSecret: os.Getenv("MOVEPROBE_SECRET") != "",
}
@@ -851,36 +896,11 @@ func (a *app) handleReport(w http.ResponseWriter, r *http.Request) {
rep.FileKeySHA256 = fingerprint(k)
}
- rows, err := a.db.QueryContext(r.Context(), `SELECT kind, region, count(*), min(seq), max(seq) FROM rows GROUP BY kind, region ORDER BY kind, region`)
- if err != nil {
- http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
+ var err error
+ if rep.Rows, rep.Fired, err = a.counts(r.Context()); err != nil {
+ http.Error(w, "report: "+err.Error(), http.StatusInternalServerError)
return
}
- for rows.Next() {
- var c rowCount
- if err := rows.Scan(&c.Kind, &c.Region, &c.Count, &c.MinSeq, &c.MaxSeq); err != nil {
- rows.Close()
- http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
- return
- }
- rep.Rows = append(rep.Rows, c)
- }
- rows.Close()
- rows, err = a.db.QueryContext(r.Context(), `SELECT kind, name, due_at, count(*) FROM rows WHERE kind IN ('tick', 'once') GROUP BY kind, name, due_at ORDER BY kind, name, due_at`)
- if err != nil {
- http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
- return
- }
- for rows.Next() {
- var c firedCount
- if err := rows.Scan(&c.Kind, &c.Name, &c.DueAt, &c.Count); err != nil {
- rows.Close()
- http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
- return
- }
- rep.Fired = append(rep.Fired, c)
- }
- rows.Close()
rep.Object.Key = a.objKey
if a.storeErr != nil {
diff --git a/cmd/moveprobe/main_test.go b/cmd/moveprobe/main_test.go
index 74d365063..3100e2ebb 100644
--- a/cmd/moveprobe/main_test.go
+++ b/cmd/moveprobe/main_test.go
@@ -14,9 +14,11 @@ import (
"crypto/x509"
"crypto/x509/pkix"
"database/sql"
+ "database/sql/driver"
"encoding/hex"
"encoding/json"
"encoding/pem"
+ "errors"
"fmt"
"io"
"io/fs"
@@ -1104,3 +1106,99 @@ func TestAfterWriteExitsWhenItsRowFails(t *testing.T) {
t.Errorf("%d bg rows in ie", n)
}
}
+
+// ---- report query failures ------------------------------------------------
+
+// failDriver is a database/sql driver whose query matching the DSN (a
+// substring) yields one good row and then fails, as a read that breaks part
+// way through would. Every other query yields no rows.
+type failDriver struct{}
+
+func (failDriver) Open(failOn string) (driver.Conn, error) { return failConn{failOn}, nil }
+
+type failConn struct{ failOn string }
+
+func (c failConn) Prepare(q string) (driver.Stmt, error) { return failStmt{c.failOn, q}, nil }
+func (failConn) Close() error { return nil }
+func (failConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") }
+
+type failStmt struct{ failOn, q string }
+
+func (failStmt) Close() error { return nil }
+func (failStmt) NumInput() int { return -1 }
+func (failStmt) Exec([]driver.Value) (driver.Result, error) {
+ return nil, errors.New("no writes")
+}
+func (s failStmt) Query([]driver.Value) (driver.Rows, error) {
+ cols := []string{"kind", "region", "count", "min", "max"}
+ row := []driver.Value{"bg", "uk", int64(3), int64(1), int64(3)}
+ if strings.Contains(s.q, "IN ('tick', 'once')") {
+ cols = []string{"kind", "name", "due_at", "count"}
+ row = []driver.Value{"once", "once-1", int64(1_700_000_000), int64(1)}
+ }
+ return &failRows{cols: cols, row: row, fail: strings.Contains(s.q, s.failOn)}, nil
+}
+
+type failRows struct {
+ cols []string
+ row []driver.Value
+ fail bool
+ n int
+}
+
+func (r *failRows) Columns() []string { return r.cols }
+func (r *failRows) Close() error { return nil }
+func (r *failRows) Next(dest []driver.Value) error {
+ r.n++
+ switch {
+ case !r.fail:
+ return io.EOF
+ case r.n == 1:
+ copy(dest, r.row)
+ return nil
+ default:
+ return errors.New("forced read failure")
+ }
+}
+
+func init() { sql.Register("moveprobe-fail", failDriver{}) }
+
+func TestReportNeverAnswersPartial(t *testing.T) {
+ report := func(t *testing.T, db *sql.DB) (int, string) {
+ t.Helper()
+ a := &app{db: db, filesDir: t.TempDir(), region: "uk", mode: modeNormal, storeErr: errors.New("store: not under test")}
+ rec := httptest.NewRecorder()
+ a.handleReport(rec, httptest.NewRequest(http.MethodGet, "/report", nil))
+ return rec.Code, rec.Body.String()
+ }
+ for _, failOn := range []string{"GROUP BY kind, region", "IN ('tick', 'once')"} {
+ t.Run("rows fail part way: "+failOn, func(t *testing.T) {
+ db, err := sql.Open("moveprobe-fail", failOn)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ code, body := report(t, db)
+ if code != http.StatusInternalServerError || !strings.Contains(body, "forced read failure") {
+ t.Errorf("/report: %d %q, want 500 naming the read failure", code, body)
+ }
+ })
+ }
+ t.Run("a closed database", func(t *testing.T) {
+ db := testDB(t, filepath.Join(t.TempDir(), "closed.db"))
+ db.Close()
+ if code, body := report(t, db); code != http.StatusInternalServerError {
+ t.Errorf("/report on a closed database: %d %q, want 500", code, body)
+ }
+ })
+ t.Run("nothing fails", func(t *testing.T) {
+ db, err := sql.Open("moveprobe-fail", "matches no query")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ if code, body := report(t, db); code != http.StatusOK {
+ t.Errorf("/report: %d %q, want 200", code, body)
+ }
+ })
+}