amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package main
4
5import (
6 "bufio"
7 "bytes"
8 "context"
9 "crypto/ecdsa"
10 "crypto/elliptic"
11 "crypto/rand"
12 "crypto/sha256"
13 "crypto/tls"
14 "crypto/x509"
15 "crypto/x509/pkix"
16 "database/sql"
17 "database/sql/driver"
18 "encoding/hex"
19 "encoding/json"
20 "encoding/pem"
21 "errors"
22 "fmt"
23 "io"
24 "io/fs"
25 "math/big"
26 "net"
27 "net/http"
28 "net/http/httptest"
29 "os"
30 "os/exec"
31 "path/filepath"
32 "sort"
33 "strconv"
34 "strings"
35 "sync"
36 "sync/atomic"
37 "syscall"
38 "testing"
39 "time"
40
41 "github.com/carlosframework/platform/internal/lease/leasetest"
42 "github.com/carlosframework/platform/internal/registry"
43 "github.com/carlosframework/platform/internal/schedule"
44
45 _ "modernc.org/sqlite"
46)
47
48const (
49 testHost = "h1.example.com"
50 adminToken = "admin-token-for-tests"
51 probeSecret = "probe-secret-for-tests"
52 platformKey = "platform-instance-key-for-tests"
53 accessKey = "AKIDMOVEPROBETEST"
54 bucketName = "probe-bucket"
55 storeRegion = "eu-west-1"
56)
57
58// binPath is the moveprobe binary the tests start, built once in TestMain,
59// so every test starts the app the way the exec backend does: a separate
60// process given --socket and --db.
61var binPath string
62
63func TestMain(m *testing.M) {
64 dir, err := os.MkdirTemp("", "mpbin")
65 if err != nil {
66 fmt.Fprintln(os.Stderr, err)
67 os.Exit(1)
68 }
69 binPath = filepath.Join(dir, "moveprobe")
70 if out, err := exec.Command("go", "build", "-o", binPath, ".").CombinedOutput(); err != nil {
71 fmt.Fprintf(os.Stderr, "build moveprobe: %v\n%s", err, out)
72 os.RemoveAll(dir)
73 os.Exit(1)
74 }
75 code := m.Run()
76 os.RemoveAll(dir)
77 os.Exit(code)
78}
79
80// shortTempDir keeps unix socket paths under the kernel's length limit
81// (as internal/activator/rolelaunch_root_linux_test.go does); t.TempDir's
82// names carry the test's name and can pass 107 bytes.
83func shortTempDir(t *testing.T) string {
84 t.Helper()
85 dir, err := os.MkdirTemp("", "mp")
86 if err != nil {
87 t.Fatal(err)
88 }
89 t.Cleanup(func() { os.RemoveAll(dir) })
90 return dir
91}
92
93// fixture is one instance's world: its data directory (the flat layout,
94// <dir>/h1.db and <dir>/h1/), a test S3 holding its store, and a control
95// socket served by the platform's own schedule verbs over a registry.
96// Each test makes its own, and shares nothing else but the read-only
97// binPath, so every test runs in parallel.
98type fixture struct {
99 t *testing.T
100 sockDir string
101 dataDir string
102 db string
103 bucket *leasetest.VersionedBucket
104 store http.Handler
105 storeURL string
106 reg *registry.Registry
107 control string
108 env map[string]string
109
110 mu sync.Mutex
111 auths []string
112 storeReq atomic.Int64
113 // storeDelay slows every store request, in milliseconds.
114 storeDelay atomic.Int64
115 // controlConns counts the connections the control socket accepted.
116 controlConns atomic.Int64
117 starts int
118}
119
120func newFixture(t *testing.T) *fixture {
121 t.Helper()
122 f := &fixture{t: t, sockDir: shortTempDir(t), dataDir: t.TempDir()}
123 f.db = filepath.Join(f.dataDir, "h1.db")
124
125 f.bucket = leasetest.NewVersionedBucket()
126 f.bucket.AddBucket(bucketName)
127 s3h, _ := leasetest.NewS3Server(f.bucket)
128 f.store = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
129 f.storeReq.Add(1)
130 time.Sleep(time.Duration(f.storeDelay.Load()) * time.Millisecond)
131 f.mu.Lock()
132 f.auths = append(f.auths, r.Header.Get("Authorization"))
133 f.mu.Unlock()
134 s3h.ServeHTTP(w, r)
135 })
136 srv := httptest.NewServer(f.store)
137 t.Cleanup(srv.Close)
138 f.storeURL = srv.URL
139
140 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
141 if err != nil {
142 t.Fatal(err)
143 }
144 t.Cleanup(func() { reg.Close() })
145 f.reg = reg
146 f.control = filepath.Join(f.sockDir, "c.sock")
147 ln, err := net.Listen("unix", f.control)
148 if err != nil {
149 t.Fatal(err)
150 }
151 mux := http.NewServeMux()
152 schedule.Mount(mux, testHost, reg, func() (string, error) { return adminToken, nil }, time.Now)
153 cs := &http.Server{Handler: mux, ConnState: func(_ net.Conn, s http.ConnState) {
154 if s == http.StateNew {
155 f.controlConns.Add(1)
156 }
157 }}
158 go cs.Serve(ln)
159 t.Cleanup(func() { cs.Close() })
160
161 // The environment agentOwnedVars composes over <host>.env: the control
162 // socket (when the listener bound), the purge channel, the region (when
163 // the box has one) and the host; then the env file's own variables.
164 f.env = map[string]string{
165 "CARLOS_CONTROL_SOCKET": f.control,
166 "CARLOS_PURGE": "header",
167 "CARLOS_REGION": "uk",
168 "CARLOS_HOST": testHost,
169 "CARLOS_ADMIN_TOKEN": adminToken,
170 "CARLOS_INSTANCE_KEY": platformKey,
171 "MOVEPROBE_BUCKET": bucketName,
172 "MOVEPROBE_REGION": storeRegion,
173 "MOVEPROBE_ENDPOINT": f.storeURL,
174 "MOVEPROBE_ACCESS_KEY": accessKey,
175 "MOVEPROBE_SECRET_KEY": "secret-key-for-tests",
176 "MOVEPROBE_SECRET": probeSecret,
177 }
178 // Registered first, so it runs last: every probe's main is gone by then.
179 t.Cleanup(func() { killWorkers(f.db) })
180 return f
181}
182
183func (f *fixture) seenAuth() []string {
184 f.mu.Lock()
185 defer f.mu.Unlock()
186 return append([]string(nil), f.auths...)
187}
188
189// envWith is the fixture's environment with changes applied; an empty
190// value removes the variable.
191func (f *fixture) envWith(changes map[string]string) map[string]string {
192 out := map[string]string{}
193 for k, v := range f.env {
194 out[k] = v
195 }
196 for k, v := range changes {
197 if v == "" {
198 delete(out, k)
199 } else {
200 out[k] = v
201 }
202 }
203 return out
204}
205
206// childEnv mirrors the backend's: the parent's environment without any
207// CARLOS_ variable, then the instance's variables. The test also drops the
208// probe's and AWS's own names and every proxy setting, so nothing from the
209// shell running the tests reaches the app.
210func childEnv(vars map[string]string) []string {
211 var out []string
212 for _, kv := range os.Environ() {
213 k, _, _ := strings.Cut(kv, "=")
214 switch {
215 case strings.HasPrefix(k, "CARLOS_"), strings.HasPrefix(k, "MOVEPROBE_"), strings.HasPrefix(k, "AWS_"):
216 continue
217 case strings.EqualFold(k, "HTTP_PROXY"), strings.EqualFold(k, "HTTPS_PROXY"), strings.EqualFold(k, "NO_PROXY"), k == "SSL_CERT_FILE", k == "SSL_CERT_DIR":
218 continue
219 }
220 out = append(out, kv)
221 }
222 for k, v := range vars {
223 out = append(out, k+"="+v)
224 }
225 return out
226}
227
228type lockedBuffer struct {
229 mu sync.Mutex
230 b bytes.Buffer
231}
232
233func (l *lockedBuffer) Write(p []byte) (int, error) {
234 l.mu.Lock()
235 defer l.mu.Unlock()
236 return l.b.Write(p)
237}
238
239func (l *lockedBuffer) String() string {
240 l.mu.Lock()
241 defer l.mu.Unlock()
242 return l.b.String()
243}
244
245type probe struct {
246 t *testing.T
247 cmd *exec.Cmd
248 sock string
249 out *lockedBuffer
250 done chan struct{}
251}
252
253// start runs the binary exactly as spawnChildFrom does:
254// exec.Command(bin, "--socket", sock, "--db", dbPath) with the composed env.
255// Its stdout and stderr are pipes, as the backend's tee makes them.
256func (f *fixture) start(env map[string]string) *probe {
257 f.t.Helper()
258 return f.startBin(binPath, env)
259}
260
261// startBin is start with another build of moveprobe.
262func (f *fixture) startBin(bin string, env map[string]string) *probe {
263 f.t.Helper()
264 f.starts++
265 p := &probe{t: f.t, sock: filepath.Join(f.sockDir, fmt.Sprintf("a%d.sock", f.starts)), out: &lockedBuffer{}, done: make(chan struct{})}
266 p.cmd = exec.Command(bin, "--socket", p.sock, "--db", f.db)
267 p.cmd.Env = childEnv(env)
268 p.cmd.Stdout = p.out
269 p.cmd.Stderr = p.out
270 if err := p.cmd.Start(); err != nil {
271 f.t.Fatal(err)
272 }
273 go func() { p.cmd.Wait(); close(p.done) }()
274 f.t.Cleanup(p.kill)
275 return p
276}
277
278func (p *probe) exited() bool {
279 select {
280 case <-p.done:
281 return true
282 default:
283 return false
284 }
285}
286
287func (p *probe) kill() {
288 if p.exited() {
289 return
290 }
291 p.cmd.Process.Kill()
292 <-p.done
293}
294
295// stop is the backend's Stop: SIGTERM to the leader only.
296func (p *probe) stop() {
297 p.t.Helper()
298 if p.exited() {
299 return
300 }
301 p.cmd.Process.Signal(syscall.SIGTERM)
302 select {
303 case <-p.done:
304 case <-time.After(10 * time.Second):
305 p.t.Errorf("moveprobe did not exit on SIGTERM within 10s; output:\n%s", p.out)
306 p.kill()
307 }
308}
309
310func unixClient(sock string) *http.Client {
311 return &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
312 DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
313 var d net.Dialer
314 return d.DialContext(ctx, "unix", sock)
315 },
316 }}
317}
318
319func (p *probe) do(method, path, bearer string, hdr map[string]string) (int, string, error) {
320 req, err := http.NewRequest(method, "http://"+testHost+path, nil)
321 if err != nil {
322 return 0, "", err
323 }
324 if bearer != "" {
325 req.Header.Set("Authorization", "Bearer "+bearer)
326 }
327 for k, v := range hdr {
328 req.Header.Set(k, v)
329 }
330 resp, err := unixClient(p.sock).Do(req)
331 if err != nil {
332 return 0, "", err
333 }
334 defer resp.Body.Close()
335 b, _ := io.ReadAll(resp.Body)
336 return resp.StatusCode, string(b), nil
337}
338
339// wait polls path until it answers want, and fails the test with the last
340// answer and the app's output otherwise.
341func (p *probe) wait(path string, want int) string {
342 p.t.Helper()
343 deadline := time.Now().Add(20 * time.Second)
344 var last string
345 for time.Now().Before(deadline) {
346 code, body, err := p.do(http.MethodGet, path, "", nil)
347 if err == nil && code == want {
348 return body
349 }
350 last = fmt.Sprintf("code %d body %q err %v", code, body, err)
351 if p.exited() {
352 break
353 }
354 time.Sleep(100 * time.Millisecond)
355 }
356 p.t.Fatalf("GET %s never answered %d; last: %s; output:\n%s", path, want, last, p.out)
357 return ""
358}
359
360type reportDoc struct {
361 Region string `json:"region"`
362 RegionFrom string `json:"region_from"`
363 Mode string `json:"mode"`
364 Rows []struct {
365 Kind string `json:"kind"`
366 Region string `json:"region"`
367 Count int64 `json:"count"`
368 MinSeq int64 `json:"min_seq"`
369 MaxSeq int64 `json:"max_seq"`
370 } `json:"rows"`
371 Fired []struct {
372 Kind string `json:"kind"`
373 Name string `json:"name"`
374 DueAt int64 `json:"due_at"`
375 Count int64 `json:"count"`
376 } `json:"fired"`
377 FileKeySHA256 string `json:"file_key_sha256"`
378 PlatformKeySHA256 string `json:"platform_key_sha256"`
379 Object struct {
380 Key string `json:"key"`
381 Body string `json:"body"`
382 Error string `json:"error"`
383 } `json:"object"`
384 HasSecret bool `json:"has_secret"`
385 WorkerPID int `json:"worker_pid"`
386 WorkerWritesFailed int64 `json:"worker_writes_failed"`
387 WorkerWritesCommitted int64 `json:"worker_writes_committed"`
388 WorkerFailedInARow int64 `json:"worker_failed_in_a_row"`
389 WorkerLastError string `json:"worker_last_error"`
390 WorkerExit string `json:"worker_exit"`
391 WorkerLastStatusAt int64 `json:"worker_last_status_at"`
392}
393
394func (r reportDoc) count(kind, region string) int64 {
395 for _, row := range r.Rows {
396 if row.Kind == kind && row.Region == region {
397 return row.Count
398 }
399 }
400 return 0
401}
402
403func (p *probe) report() reportDoc {
404 p.t.Helper()
405 code, body, err := p.do(http.MethodGet, "/report", probeSecret, nil)
406 if err != nil || code != http.StatusOK {
407 p.t.Fatalf("/report: code %d err %v body %q; output:\n%s", code, err, body, p.out)
408 }
409 var r reportDoc
410 if err := json.Unmarshal([]byte(body), &r); err != nil {
411 p.t.Fatalf("/report: %v in %q", err, body)
412 }
413 return r
414}
415
416// workerPIDs lists the pids whose command line starts with a moveprobe
417// build the tests made (binPath, or a test's own, all named moveprobe*) and
418// "worker", and names db.
419func workerPIDs(db string) []int {
420 var pids []int
421 ents, _ := os.ReadDir("/proc")
422 for _, e := range ents {
423 pid, err := strconv.Atoi(e.Name())
424 if err != nil {
425 continue
426 }
427 raw, err := os.ReadFile(filepath.Join("/proc", e.Name(), "cmdline"))
428 if err != nil {
429 continue
430 }
431 args := strings.Split(strings.TrimRight(string(raw), "\x00"), "\x00")
432 if len(args) < 2 || !strings.HasPrefix(filepath.Base(args[0]), "moveprobe") || args[1] != "worker" || !strings.Contains(string(raw), db) {
433 continue
434 }
435 if alive(pid) {
436 pids = append(pids, pid)
437 }
438 }
439 return pids
440}
441
442// alive is true while pid exists and is neither a zombie nor dead.
443func alive(pid int) bool {
444 raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
445 if err != nil {
446 return false
447 }
448 i := strings.LastIndexByte(string(raw), ')')
449 if i < 0 || i+2 >= len(raw) {
450 return false
451 }
452 state := raw[i+2]
453 return state != 'Z' && state != 'X'
454}
455
456func killWorkers(db string) {
457 for _, pid := range workerPIDs(db) {
458 syscall.Kill(pid, syscall.SIGKILL)
459 }
460 deadline := time.Now().Add(5 * time.Second)
461 for len(workerPIDs(db)) > 0 && time.Now().Before(deadline) {
462 time.Sleep(50 * time.Millisecond)
463 }
464}
465
466func testDB(t *testing.T, db string) *sql.DB {
467 t.Helper()
468 d, err := sql.Open("sqlite", "file:"+db+"?_pragma=busy_timeout(5000)")
469 if err != nil {
470 t.Fatal(err)
471 }
472 return d
473}
474
475func countRows(t *testing.T, db, kind, region string) int64 {
476 t.Helper()
477 d := testDB(t, db)
478 defer d.Close()
479 var n int64
480 if err := d.QueryRow(`SELECT count(*) FROM rows WHERE kind=? AND region=?`, kind, region).Scan(&n); err != nil {
481 t.Fatal(err)
482 }
483 return n
484}
485
486// checkpoint folds the WAL into the database file and closes, so the file
487// is what a move would install.
488func checkpoint(t *testing.T, db string) {
489 t.Helper()
490 d := testDB(t, db)
491 if _, err := d.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
492 t.Fatal(err)
493 }
494 d.Close()
495}
496
497// snapshot is every entry under dir with its size and SHA-256: the
498// database file's digest and the files directory's listing, and anything
499// else that appears beside them (a -wal or -shm file included).
500func snapshot(t *testing.T, dir string) string {
501 t.Helper()
502 var lines []string
503 err := filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
504 if err != nil {
505 return err
506 }
507 rel, _ := filepath.Rel(dir, p)
508 if d.IsDir() {
509 lines = append(lines, rel+"/")
510 return nil
511 }
512 b, err := os.ReadFile(p)
513 if err != nil {
514 return err
515 }
516 sum := sha256.Sum256(b)
517 lines = append(lines, fmt.Sprintf("%s %d %x", rel, len(b), sum))
518 return nil
519 })
520 if err != nil {
521 t.Fatal(err)
522 }
523 sort.Strings(lines)
524 return strings.Join(lines, "\n")
525}
526
527func sha(s string) string {
528 sum := sha256.Sum256([]byte(s))
529 return hex.EncodeToString(sum[:])
530}
531
532// fire delivers a timer the way the platform's runner does: the request
533// schedule.BuildRequest builds, sent over the instance's socket.
534func (p *probe) fire(tm registry.Timer, token string) (int, string) {
535 p.t.Helper()
536 req, err := schedule.BuildRequest(registry.Route{Host: testHost}, tm, token, "test")
537 if err != nil {
538 p.t.Fatal(err)
539 }
540 resp, err := unixClient(p.sock).Do(req)
541 if err != nil {
542 p.t.Fatal(err)
543 }
544 defer resp.Body.Close()
545 b, _ := io.ReadAll(resp.Body)
546 return resp.StatusCode, string(b)
547}
548
549// ---- the launch contract --------------------------------------------------
550
551func TestMoveprobeLaunchContract(t *testing.T) {
552 t.Parallel()
553 t.Run("the agent's variables and a store endpoint", func(t *testing.T) {
554 f := newFixture(t)
555 p := f.start(f.env)
556 if body := p.wait("/ready", http.StatusOK); !strings.Contains(body, "ready") {
557 t.Errorf("/ready body %q", body)
558 }
559 // The platform probes "/" when moves.ready_path is unset.
560 p.wait("/", http.StatusOK)
561
562 vs, err := f.bucket.ListVersions(context.Background(), bucketName, "probe/")
563 if err != nil || len(vs) != 1 {
564 t.Fatalf("probe objects: %v %v", vs, err)
565 }
566 for _, a := range f.seenAuth() {
567 if !strings.Contains(a, "Credential="+accessKey+"/") {
568 t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
569 }
570 }
571
572 // /arm-once and /report need the probe's bearer.
573 if code, _, _ := p.do(http.MethodGet, "/report", "", nil); code != http.StatusUnauthorized {
574 t.Errorf("GET /report without a bearer: %d, want 401", code)
575 }
576 if code, _, _ := p.do(http.MethodGet, "/report", "wrong", nil); code != http.StatusUnauthorized {
577 t.Errorf("GET /report with the wrong bearer: %d, want 401", code)
578 }
579 if code, _, _ := p.do(http.MethodPost, "/arm-once?in=600", "", nil); code != http.StatusUnauthorized {
580 t.Errorf("POST /arm-once without a bearer: %d, want 401", code)
581 }
582 before := time.Now().Unix()
583 code, body, err := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil)
584 if err != nil || code != http.StatusOK {
585 t.Fatalf("POST /arm-once: %d %q %v", code, body, err)
586 }
587 tm, err := f.reg.GetTimer(testHost, "once-1")
588 if err != nil {
589 t.Fatalf("once-1 not registered through the control socket: %v (arm answered %q)", err, body)
590 }
591 if tm.Kind != schedule.KindOnce || tm.Path != "/once" || tm.DueAt < before+600 || tm.DueAt > time.Now().Unix()+601 {
592 t.Errorf("once-1 = %+v, want a once-timer for /once due in 600s", tm)
593 }
594 if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
595 t.Errorf("the second arm: %d %q", code, body)
596 }
597 if _, err := f.reg.GetTimer(testHost, "once-2"); err != nil {
598 t.Errorf("the second arm is not once-2: %v", err)
599 }
600
601 r := p.report()
602 if r.Region != "uk" || r.RegionFrom != "CARLOS_REGION" {
603 t.Errorf("report region %q from %q, want uk from CARLOS_REGION", r.Region, r.RegionFrom)
604 }
605 if r.PlatformKeySHA256 != sha(platformKey) || r.FileKeySHA256 == "" || !r.HasSecret || r.Object.Body == "" || r.Object.Error != "" {
606 t.Errorf("report = %+v", r)
607 }
608 p.stop()
609 })
610
611 t.Run("no CARLOS_REGION and no control socket", func(t *testing.T) {
612 f := newFixture(t)
613 p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "CARLOS_CONTROL_SOCKET": ""}))
614 p.wait("/ready", http.StatusOK)
615 code, body, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil)
616 if code != http.StatusServiceUnavailable || !strings.Contains(body, "CARLOS_CONTROL_SOCKET") {
617 t.Errorf("/arm-once with no control socket: %d %q, want 503 naming CARLOS_CONTROL_SOCKET", code, body)
618 }
619 p.wait("/ready", http.StatusOK) // unaffected
620 r := p.report()
621 if r.Region != "home" || !strings.Contains(r.RegionFrom, "unset") {
622 t.Errorf("report region %q from %q, want home, with CARLOS_REGION unset", r.Region, r.RegionFrom)
623 }
624 time.Sleep(700 * time.Millisecond)
625 if n := countRows(t, f.db, "bg", "home"); n == 0 {
626 t.Errorf("no bg rows written as region home")
627 }
628 p.stop()
629 })
630
631 t.Run("no MOVEPROBE_ENDPOINT: AWS's own host, the delivered keys", func(t *testing.T) {
632 f := newFixture(t)
633 proxy, caFile, connects := awsLookalike(t, f.store)
634 p := f.start(f.envWith(map[string]string{
635 "MOVEPROBE_ENDPOINT": "",
636 "HTTPS_PROXY": proxy,
637 "SSL_CERT_FILE": caFile,
638 }))
639 p.wait("/ready", http.StatusOK)
640 want := bucketName + ".s3." + storeRegion + ".amazonaws.com:443"
641 got := connects()
642 if len(got) == 0 {
643 t.Fatalf("no store request went to AWS's host")
644 }
645 for _, h := range got {
646 if h != want {
647 t.Errorf("store request to %q, want %q (virtual-hosted, no endpoint)", h, want)
648 }
649 }
650 for _, a := range f.seenAuth() {
651 if !strings.Contains(a, "Credential="+accessKey+"/") {
652 t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
653 }
654 }
655 p.stop()
656 })
657
658 t.Run("the store keys come only from MOVEPROBE_", func(t *testing.T) {
659 f := newFixture(t)
660 p := f.start(f.envWith(map[string]string{
661 "MOVEPROBE_ACCESS_KEY": "",
662 "AWS_ACCESS_KEY_ID": accessKey,
663 "AWS_SECRET_ACCESS_KEY": "x",
664 "AWS_ENDPOINT_URL": f.storeURL,
665 "AWS_ENDPOINT_URL_S3": f.storeURL,
666 }))
667 body := p.wait("/ready", http.StatusInternalServerError)
668 if !strings.Contains(body, "MOVEPROBE_ACCESS_KEY") {
669 t.Errorf("/ready body %q does not name MOVEPROBE_ACCESS_KEY", body)
670 }
671 if n := f.storeReq.Load(); n != 0 {
672 t.Errorf("%d store requests made without the binding's keys", n)
673 }
674 p.stop()
675 })
676
677 t.Run("the sealed secret is required", func(t *testing.T) {
678 f := newFixture(t)
679 p := f.start(f.envWith(map[string]string{"MOVEPROBE_SECRET": ""}))
680 body := p.wait("/ready", http.StatusInternalServerError)
681 if !strings.Contains(body, "MOVEPROBE_SECRET") {
682 t.Errorf("/ready body %q does not name MOVEPROBE_SECRET", body)
683 }
684 p.stop()
685 })
686}
687
688// awsLookalike stands in for S3's regional host: an HTTPS proxy whose
689// CONNECT tunnels end at a TLS server holding a certificate for
690// *.s3.<region>.amazonaws.com, signed by a CA written to the returned file.
691// The tunnelled requests are virtual-hosted; they reach store path-style.
692func awsLookalike(t *testing.T, store http.Handler) (proxyURL, caFile string, connects func() []string) {
693 t.Helper()
694 key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
695 if err != nil {
696 t.Fatal(err)
697 }
698 tmpl := &x509.Certificate{
699 SerialNumber: big.NewInt(1),
700 Subject: pkix.Name{CommonName: "moveprobe test s3"},
701 NotBefore: time.Now().Add(-time.Hour),
702 NotAfter: time.Now().Add(time.Hour),
703 DNSNames: []string{"*.s3." + storeRegion + ".amazonaws.com", "s3." + storeRegion + ".amazonaws.com"},
704 KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
705 ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
706 BasicConstraintsValid: true,
707 IsCA: true,
708 }
709 der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
710 if err != nil {
711 t.Fatal(err)
712 }
713 caFile = filepath.Join(t.TempDir(), "ca.pem")
714 if err := os.WriteFile(caFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil {
715 t.Fatal(err)
716 }
717 tlsCfg := &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}}
718
719 vhost := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
720 bucket, _, _ := strings.Cut(r.Host, ".")
721 r.URL.Path = "/" + bucket + r.URL.Path
722 r.URL.RawPath = ""
723 store.ServeHTTP(w, r)
724 })
725 var mu sync.Mutex
726 var hosts []string
727 proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
728 if r.Method != http.MethodConnect {
729 http.Error(w, "CONNECT only", http.StatusMethodNotAllowed)
730 return
731 }
732 mu.Lock()
733 hosts = append(hosts, r.Host)
734 mu.Unlock()
735 conn, brw, err := w.(http.Hijacker).Hijack()
736 if err != nil {
737 return
738 }
739 brw.WriteString("HTTP/1.1 200 Connection established\r\n\r\n")
740 brw.Flush()
741 srv := &http.Server{Handler: vhost}
742 srv.Serve(&oneConn{c: tls.Server(&bufferedConn{Conn: conn, r: brw.Reader}, tlsCfg)})
743 }))
744 t.Cleanup(proxy.Close)
745 return proxy.URL, caFile, func() []string {
746 mu.Lock()
747 defer mu.Unlock()
748 return append([]string(nil), hosts...)
749 }
750}
751
752type bufferedConn struct {
753 net.Conn
754 r *bufio.Reader
755}
756
757func (b *bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) }
758
759// oneConn is a listener that yields one connection; Serve returns once it
760// has been handed out and the connection is served on its own goroutine.
761type oneConn struct {
762 c net.Conn
763 used bool
764}
765
766func (o *oneConn) Accept() (net.Conn, error) {
767 if o.used {
768 return nil, io.EOF
769 }
770 o.used = true
771 return o.c, nil
772}
773func (o *oneConn) Close() error { return nil }
774func (o *oneConn) Addr() net.Addr { return o.c.LocalAddr() }
775
776// ---- the forked worker ------------------------------------------------------
777
778// The worker's grace is a minute in every build but this test's: it links
779// its own moveprobe with an 8 s grace (workerGraceOverride) and watches
780// that build's worker outlive its main, keep writing, and leave on its own
781// at the grace, as the shipped build does at a minute.
782func TestMoveprobeWorkerOutlivesMain(t *testing.T) {
783 t.Parallel()
784 if workerGrace != 60*time.Second || workerGraceOverride != "" || orphanGrace() != workerGrace {
785 t.Fatalf("a plain build's worker grace is %s (const %s, override %q), want a minute", orphanGrace(), workerGrace, workerGraceOverride)
786 }
787 const grace = 8 * time.Second
788 bin := filepath.Join(t.TempDir(), "moveprobe-grace")
789 build := exec.Command("go", "build", "-ldflags", "-X main.workerGraceOverride="+grace.String(), "-o", bin, ".")
790 if out, err := build.CombinedOutput(); err != nil {
791 t.Fatalf("build moveprobe with an %s grace: %v\n%s", grace, err, out)
792 }
793 f := newFixture(t)
794 p := f.startBin(bin, f.env)
795 p.wait("/ready", http.StatusOK)
796 r := p.report()
797 if r.WorkerPID <= 0 || !alive(r.WorkerPID) {
798 t.Fatalf("no live worker in the report: %d", r.WorkerPID)
799 }
800 if got := workerPIDs(f.db); len(got) != 1 || got[0] != r.WorkerPID {
801 t.Fatalf("worker processes %v, report says %d", got, r.WorkerPID)
802 }
803
804 // The main dies outright. Its stdout and stderr are pipes; a worker
805 // still holding them would keep Wait from returning (the backend's tee
806 // waits up to its drain grace for exactly that).
807 killed := time.Now()
808 p.cmd.Process.Kill()
809 select {
810 case <-p.done:
811 case <-time.After(2 * time.Second):
812 t.Fatalf("the main's Wait did not return within 2s of its death: the worker holds its stdio")
813 }
814
815 n0 := countRows(t, f.db, "worker", "uk")
816 time.Sleep(3 * time.Second)
817 if !alive(r.WorkerPID) {
818 t.Fatalf("the worker died with the main")
819 }
820 if n1 := countRows(t, f.db, "worker", "uk"); n1 < n0+2 {
821 t.Fatalf("the worker stopped writing after the main died: %d then %d rows", n0, n1)
822 }
823
824 // It exits on its own about its grace after its parent has gone.
825 for alive(r.WorkerPID) && time.Since(killed) < grace+20*time.Second {
826 time.Sleep(50 * time.Millisecond)
827 }
828 gone := time.Since(killed)
829 if alive(r.WorkerPID) {
830 t.Fatalf("the worker is still alive %s after the main died", gone.Round(time.Second))
831 }
832 // The worker notices its parent has gone on its next 1 s tick and
833 // leaves on the first tick at or past the grace, so it exits between
834 // the grace and about 2 s after it. 4 s of slack still refuses a worker
835 // that waits twice its grace.
836 if gone < grace-time.Second || gone > grace+4*time.Second {
837 t.Errorf("the worker exited %s after the main died, want about %s", gone.Round(time.Second), grace)
838 }
839}
840
841// ---- the failure switch ---------------------------------------------------
842
843func TestReadyFailsWhereTold(t *testing.T) {
844 t.Parallel()
845 f := newFixture(t)
846
847 // The installed state: a normal boot in uk, stopped, its worker killed
848 // (the stop evidence does that on a real box) and the WAL folded in.
849 seed := f.start(f.env)
850 seed.wait("/ready", http.StatusOK)
851 time.Sleep(time.Second)
852 seed.stop()
853 killWorkers(f.db)
854 checkpoint(t, f.db)
855 installed := snapshot(t, f.dataDir)
856 if !strings.Contains(installed, "h1/instance.key") || !strings.Contains(installed, "h1.db ") {
857 t.Fatalf("the seed boot left no database or key file:\n%s", installed)
858 }
859 timers, _ := f.reg.TimersFor(testHost)
860
861 t.Run("before-write writes nothing", func(t *testing.T) {
862 p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie"}))
863 p.wait("/ready", http.StatusServiceUnavailable)
864 p.wait("/", http.StatusServiceUnavailable)
865 if code, _, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil); code != http.StatusServiceUnavailable {
866 t.Errorf("/arm-once in before-write: %d, want 503", code)
867 }
868 code, _ := p.fire(registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: time.Now().Unix()}, adminToken)
869 if code != http.StatusServiceUnavailable {
870 t.Errorf("/tick in before-write: %d, want 503", code)
871 }
872 code, _ = p.fire(registry.Timer{Host: testHost, Name: "once-9", Kind: schedule.KindOnce, Path: "/once", DueAt: time.Now().Unix()}, adminToken)
873 if code != http.StatusServiceUnavailable {
874 t.Errorf("/once in before-write: %d, want 503", code)
875 }
876 time.Sleep(2 * time.Second)
877 if w := workerPIDs(f.db); len(w) != 0 {
878 t.Errorf("before-write forked a worker: %v", w)
879 }
880 p.stop()
881 if got := snapshot(t, f.dataDir); got != installed {
882 t.Errorf("before-write changed the data directory\ninstalled:\n%s\nafter:\n%s", installed, got)
883 }
884 if got, _ := f.reg.TimersFor(testHost); len(got) != len(timers) {
885 t.Errorf("before-write changed the timers: %v then %v", timers, got)
886 }
887 })
888
889 t.Run("after-write writes one bg row, then fails", func(t *testing.T) {
890 p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie,after-write"}))
891 p.wait("/ready", http.StatusServiceUnavailable)
892 p.wait("/", http.StatusServiceUnavailable)
893 time.Sleep(2 * time.Second)
894 r := p.report()
895 if r.Mode != "after-write" {
896 t.Errorf("mode %q, want after-write", r.Mode)
897 }
898 if n := r.count("bg", "ie"); n != 1 {
899 t.Errorf("after-write wrote %d bg rows in ie, want exactly 1", n)
900 }
901 if n := r.count("worker", "ie"); n < 1 {
902 t.Errorf("after-write's worker wrote no rows in ie")
903 }
904 p.stop()
905 killWorkers(f.db)
906 })
907
908 t.Run("another region's switch does not apply", func(t *testing.T) {
909 p := f.start(f.envWith(map[string]string{"MOVEPROBE_FAIL_READY_IN": "ie"}))
910 p.wait("/ready", http.StatusOK)
911 p.stop()
912 killWorkers(f.db)
913 })
914
915 t.Run("a box with no region is home", func(t *testing.T) {
916 checkpoint(t, f.db)
917 before := snapshot(t, f.dataDir)
918 p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "MOVEPROBE_FAIL_READY_IN": "home"}))
919 p.wait("/ready", http.StatusServiceUnavailable)
920 time.Sleep(time.Second)
921 p.stop()
922 if got := snapshot(t, f.dataDir); got != before {
923 t.Errorf("before-write at home changed the data directory")
924 }
925 })
926}
927
928// ---- the report -----------------------------------------------------------
929
930func TestReportCountsRowsByRegion(t *testing.T) {
931 t.Parallel()
932 f := newFixture(t)
933 uk := f.start(f.env)
934 uk.wait("/ready", http.StatusOK)
935
936 code, body, _ := uk.do(http.MethodPost, "/arm-once?in=1", probeSecret, nil)
937 if code != http.StatusOK {
938 t.Fatalf("/arm-once: %d %q", code, body)
939 }
940 once, err := f.reg.GetTimer(testHost, "once-1")
941 if err != nil {
942 t.Fatal(err)
943 }
944 // A tick and the once-timer, delivered as the runner delivers them.
945 tick := registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: 1_700_000_040}
946 if code, body := uk.fire(tick, adminToken); code != http.StatusOK {
947 t.Fatalf("/tick: %d %q", code, body)
948 }
949 if code, body := uk.fire(*once, adminToken); code != http.StatusOK {
950 t.Fatalf("/once: %d %q", code, body)
951 }
952 // Neither fires without the platform's token.
953 if code, _ := uk.fire(tick, "wrong"); code != http.StatusUnauthorized {
954 t.Errorf("/tick with the wrong token: %d, want 401", code)
955 }
956 if code, _ := uk.fire(*once, ""); code != http.StatusUnauthorized {
957 t.Errorf("/once with no token: %d, want 401", code)
958 }
959 time.Sleep(1500 * time.Millisecond)
960 ukReport := uk.report()
961 uk.stop()
962 killWorkers(f.db)
963
964 // The same database, now in ie.
965 ie := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie"}))
966 ie.wait("/ready", http.StatusOK)
967 time.Sleep(1500 * time.Millisecond)
968 r := ie.report()
969 ie.stop()
970
971 if r.Region != "ie" {
972 t.Errorf("region %q, want ie", r.Region)
973 }
974 for _, c := range []struct {
975 kind, region string
976 min int64
977 }{{"bg", "uk", 2}, {"bg", "ie", 2}, {"worker", "uk", 1}, {"worker", "ie", 1}, {"tick", "uk", 1}, {"once", "uk", 1}} {
978 if n := r.count(c.kind, c.region); n < c.min {
979 t.Errorf("%s rows in %s: %d, want at least %d (rows %+v)", c.kind, c.region, n, c.min, r.Rows)
980 }
981 }
982 if n := r.count("tick", "ie") + r.count("once", "ie"); n != 0 {
983 t.Errorf("%d timer rows in ie, none fired there", n)
984 }
985 // No background row is missing: the sequence runs on across boots.
986 var bgCount, bgMax int64
987 for _, row := range r.Rows {
988 if row.Kind == "bg" {
989 bgCount += row.Count
990 bgMax = max(bgMax, row.MaxSeq)
991 }
992 }
993 if bgCount != bgMax {
994 t.Errorf("bg rows %d, highest seq %d: a row is missing or repeated", bgCount, bgMax)
995 }
996
997 var gotOnce, gotTick bool
998 for _, fr := range r.Fired {
999 switch {
1000 case fr.Kind == "once" && fr.Name == "once-1":
1001 gotOnce = fr.Count == 1 && fr.DueAt == once.DueAt
1002 case fr.Kind == "tick" && fr.Name == "every-minute":
1003 gotTick = fr.Count == 1 && fr.DueAt == 1_700_000_040
1004 }
1005 }
1006 if !gotOnce || !gotTick {
1007 t.Errorf("fired = %+v, want once-1 once at %d and every-minute once at 1700000040", r.Fired, once.DueAt)
1008 }
1009 if r.FileKeySHA256 == "" || r.FileKeySHA256 != ukReport.FileKeySHA256 {
1010 t.Errorf("file key fingerprint %q in ie, %q in uk", r.FileKeySHA256, ukReport.FileKeySHA256)
1011 }
1012 if r.PlatformKeySHA256 != sha(platformKey) {
1013 t.Errorf("platform key fingerprint %q, want %q", r.PlatformKeySHA256, sha(platformKey))
1014 }
1015 if !r.HasSecret || r.Object.Body == "" || !strings.HasPrefix(r.Object.Key, "probe/") {
1016 t.Errorf("report = %+v", r)
1017 }
1018}
1019
1020// ---- the instance key -----------------------------------------------------
1021
1022func TestReadyChecksBothKeys(t *testing.T) {
1023 t.Parallel()
1024 f := newFixture(t)
1025 p := f.start(f.env)
1026 p.wait("/ready", http.StatusOK)
1027 p.stop()
1028 killWorkers(f.db)
1029
1030 keyFile := filepath.Join(strings.TrimSuffix(f.db, ".db"), "instance.key")
1031 orig, err := os.ReadFile(keyFile)
1032 if err != nil {
1033 t.Fatalf("first boot wrote no key file beside the database: %v", err)
1034 }
1035 if b, err := hex.DecodeString(strings.TrimSpace(string(orig))); err != nil || len(b) != 32 {
1036 t.Errorf("instance.key is not 32 bytes of hex: %q", orig)
1037 }
1038
1039 check := func(name string, env map[string]string, want int, says string) {
1040 t.Helper()
1041 p := f.start(env)
1042 body := p.wait("/ready", want)
1043 if says != "" && !strings.Contains(body, says) {
1044 t.Errorf("%s: /ready body %q, want it to say %q", name, body, says)
1045 }
1046 p.stop()
1047 killWorkers(f.db)
1048 }
1049
1050 if err := os.WriteFile(keyFile, []byte(strings.Repeat("ab", 32)), 0o600); err != nil {
1051 t.Fatal(err)
1052 }
1053 check("changed file", f.env, http.StatusInternalServerError, "key mismatch")
1054 if err := os.Remove(keyFile); err != nil {
1055 t.Fatal(err)
1056 }
1057 check("missing file", f.env, http.StatusInternalServerError, "key mismatch")
1058 if err := os.WriteFile(keyFile, orig, 0o600); err != nil {
1059 t.Fatal(err)
1060 }
1061 check("restored file", f.env, http.StatusOK, "")
1062 check("platform key changed", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": "another-key"}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
1063 check("platform key absent", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": ""}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
1064 check("both restored", f.env, http.StatusOK, "")
1065}
1066
1067// ---- write failures -------------------------------------------------------
1068
1069// failInserts makes every insert of kind into rows abort, through a trigger
1070// the test adds to the instance's own database: a real write failure, with
1071// nothing in the app to switch it on.
1072func failInserts(t *testing.T, db, kind string) {
1073 t.Helper()
1074 d := testDB(t, db)
1075 defer d.Close()
1076 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)
1077 if _, err := d.Exec(q); err != nil {
1078 t.Fatal(err)
1079 }
1080}
1081
1082func allowInserts(t *testing.T, db, kind string) {
1083 t.Helper()
1084 d := testDB(t, db)
1085 defer d.Close()
1086 if _, err := d.Exec(`DROP TRIGGER fail_` + kind); err != nil {
1087 t.Fatal(err)
1088 }
1089}
1090
1091// TestReadyFollowsTheWorkersLastWrite: /ready is 200 only once a worker row
1092// has committed since this start and the worker's latest write succeeded.
1093func TestReadyFollowsTheWorkersLastWrite(t *testing.T) {
1094 t.Parallel()
1095 f := newFixture(t)
1096 seed := f.start(f.env)
1097 seed.wait("/ready", http.StatusOK)
1098 seed.stop()
1099 killWorkers(f.db)
1100
1101 // Worker rows from the earlier boot are in the database; none can
1102 // commit in this one.
1103 failInserts(t, f.db, "worker")
1104 p := f.start(f.env)
1105 body := p.wait("/ready", http.StatusInternalServerError)
1106 if !strings.Contains(body, "worker: no row committed since start") {
1107 t.Errorf("/ready before the first worker row: %q", body)
1108 }
1109 time.Sleep(1500 * time.Millisecond)
1110 if code, body, _ := p.do(http.MethodGet, "/ready", "", nil); code != http.StatusInternalServerError {
1111 t.Errorf("/ready with no worker row yet: %d %q, want 500", code, body)
1112 }
1113 r := p.report()
1114 if r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted != 0 || !strings.Contains(r.WorkerLastError, "forced worker failure") {
1115 t.Errorf("report: committed %d, failed %d, last %q", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerLastError)
1116 }
1117
1118 // The first good write makes it ready.
1119 allowInserts(t, f.db, "worker")
1120 p.wait("/ready", http.StatusOK)
1121
1122 // One failed write is enough to fail it.
1123 failInserts(t, f.db, "worker")
1124 deadline := time.Now().Add(5 * time.Second)
1125 for {
1126 r = p.report()
1127 if r.WorkerFailedInARow > 0 || time.Now().After(deadline) {
1128 break
1129 }
1130 time.Sleep(20 * time.Millisecond)
1131 }
1132 code, body, _ := p.do(http.MethodGet, "/ready", "", nil)
1133 if r.WorkerFailedInARow != 1 || code != http.StatusInternalServerError || !strings.Contains(body, "worker: the last write failed") {
1134 t.Errorf("after one failed write (in a row %d): /ready %d %q, want 500", r.WorkerFailedInARow, code, body)
1135 }
1136
1137 // The next good write makes it ready again; the counters stay.
1138 allowInserts(t, f.db, "worker")
1139 p.wait("/ready", http.StatusOK)
1140 if r := p.report(); r.WorkerFailedInARow != 0 || r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted == 0 {
1141 t.Errorf("after recovery: committed %d, failed %d, in a row %d", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerFailedInARow)
1142 }
1143 p.stop()
1144}
1145
1146func TestAfterWriteExitsWhenItsRowFails(t *testing.T) {
1147 t.Parallel()
1148 f := newFixture(t)
1149 seed := f.start(f.env)
1150 seed.wait("/ready", http.StatusOK)
1151 seed.stop()
1152 killWorkers(f.db)
1153
1154 failInserts(t, f.db, "bg")
1155 p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie,after-write"}))
1156 select {
1157 case <-p.done:
1158 case <-time.After(15 * time.Second):
1159 t.Fatalf("after-write kept running without its bg row; output:\n%s", p.out)
1160 }
1161 if code := p.cmd.ProcessState.ExitCode(); code == 0 {
1162 t.Errorf("after-write exited 0 without its bg row; output:\n%s", p.out)
1163 }
1164 if !strings.Contains(p.out.String(), "forced bg failure") {
1165 t.Errorf("the exit does not say why; output:\n%s", p.out)
1166 }
1167 if w := workerPIDs(f.db); len(w) != 0 {
1168 t.Errorf("after-write forked a worker before its row committed: %v", w)
1169 }
1170 if n := countRows(t, f.db, "bg", "ie"); n != 0 {
1171 t.Errorf("%d bg rows in ie", n)
1172 }
1173}
1174
1175// ---- report query failures ------------------------------------------------
1176
1177// failDriver is a database/sql driver whose query matching the DSN (a
1178// substring) yields one good row and then fails, as a read that breaks part
1179// way through would. Every other query yields no rows.
1180type failDriver struct{}
1181
1182func (failDriver) Open(failOn string) (driver.Conn, error) { return failConn{failOn}, nil }
1183
1184type failConn struct{ failOn string }
1185
1186func (c failConn) Prepare(q string) (driver.Stmt, error) { return failStmt{c.failOn, q}, nil }
1187func (failConn) Close() error { return nil }
1188func (failConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") }
1189
1190type failStmt struct{ failOn, q string }
1191
1192func (failStmt) Close() error { return nil }
1193func (failStmt) NumInput() int { return -1 }
1194func (failStmt) Exec([]driver.Value) (driver.Result, error) {
1195 return nil, errors.New("no writes")
1196}
1197func (s failStmt) Query([]driver.Value) (driver.Rows, error) {
1198 cols := []string{"kind", "region", "count", "min", "max"}
1199 row := []driver.Value{"bg", "uk", int64(3), int64(1), int64(3)}
1200 if strings.Contains(s.q, "IN ('tick', 'once')") {
1201 cols = []string{"kind", "name", "due_at", "count"}
1202 row = []driver.Value{"once", "once-1", int64(1_700_000_000), int64(1)}
1203 }
1204 return &failRows{cols: cols, row: row, fail: strings.Contains(s.q, s.failOn)}, nil
1205}
1206
1207type failRows struct {
1208 cols []string
1209 row []driver.Value
1210 fail bool
1211 n int
1212}
1213
1214func (r *failRows) Columns() []string { return r.cols }
1215func (r *failRows) Close() error { return nil }
1216func (r *failRows) Next(dest []driver.Value) error {
1217 r.n++
1218 switch {
1219 case !r.fail:
1220 return io.EOF
1221 case r.n == 1:
1222 copy(dest, r.row)
1223 return nil
1224 default:
1225 return errors.New("forced read failure")
1226 }
1227}
1228
1229func init() { sql.Register("moveprobe-fail", failDriver{}) }
1230
1231func TestReportNeverAnswersPartial(t *testing.T) {
1232 t.Parallel()
1233 report := func(t *testing.T, db *sql.DB) (int, string) {
1234 t.Helper()
1235 a := &app{db: db, filesDir: t.TempDir(), region: "uk", mode: modeNormal, storeErr: errors.New("store: not under test")}
1236 rec := httptest.NewRecorder()
1237 a.handleReport(rec, httptest.NewRequest(http.MethodGet, "/report", nil))
1238 return rec.Code, rec.Body.String()
1239 }
1240 for _, failOn := range []string{"GROUP BY kind, region", "IN ('tick', 'once')"} {
1241 t.Run("rows fail part way: "+failOn, func(t *testing.T) {
1242 db, err := sql.Open("moveprobe-fail", failOn)
1243 if err != nil {
1244 t.Fatal(err)
1245 }
1246 defer db.Close()
1247 code, body := report(t, db)
1248 if code != http.StatusInternalServerError || !strings.Contains(body, "forced read failure") {
1249 t.Errorf("/report: %d %q, want 500 naming the read failure", code, body)
1250 }
1251 })
1252 }
1253 t.Run("a closed database", func(t *testing.T) {
1254 db := testDB(t, filepath.Join(t.TempDir(), "closed.db"))
1255 db.Close()
1256 if code, body := report(t, db); code != http.StatusInternalServerError {
1257 t.Errorf("/report on a closed database: %d %q, want 500", code, body)
1258 }
1259 })
1260 t.Run("nothing fails", func(t *testing.T) {
1261 db, err := sql.Open("moveprobe-fail", "matches no query")
1262 if err != nil {
1263 t.Fatal(err)
1264 }
1265 defer db.Close()
1266 if code, body := report(t, db); code != http.StatusOK {
1267 t.Errorf("/report: %d %q, want 200", code, body)
1268 }
1269 })
1270}
1271
1272func TestArmOnceReusesOneControlConnection(t *testing.T) {
1273 t.Parallel()
1274 f := newFixture(t)
1275 p := f.start(f.env)
1276 p.wait("/ready", http.StatusOK)
1277 for i := 0; i < 3; i++ {
1278 if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
1279 t.Fatalf("arm %d: %d %q", i+1, code, body)
1280 }
1281 }
1282 if n := f.controlConns.Load(); n != 1 {
1283 t.Errorf("three arms opened %d connections to the control socket, want 1", n)
1284 }
1285 p.stop()
1286}
1287
1288// TestReadyFailsWhenTheWorkerStops: a worker that has died, or has sent no
1289// status for 3 s, fails readiness, and the report says so.
1290func TestReadyFailsWhenTheWorkerStops(t *testing.T) {
1291 t.Parallel()
1292 t.Run("the worker dies after its first good write", func(t *testing.T) {
1293 f := newFixture(t)
1294 p := f.start(f.env)
1295 p.wait("/ready", http.StatusOK)
1296 r := p.report()
1297 if r.WorkerWritesCommitted == 0 || r.WorkerExit != "" || r.WorkerLastStatusAt == 0 {
1298 t.Fatalf("before: committed %d, exit %q, last status %d", r.WorkerWritesCommitted, r.WorkerExit, r.WorkerLastStatusAt)
1299 }
1300 if err := syscall.Kill(r.WorkerPID, syscall.SIGKILL); err != nil {
1301 t.Fatal(err)
1302 }
1303 body := p.wait("/ready", http.StatusInternalServerError)
1304 if !strings.Contains(body, "worker: exited") {
1305 t.Errorf("/ready after the worker died: %q", body)
1306 }
1307 if r := p.report(); !strings.Contains(r.WorkerExit, "exited") {
1308 t.Errorf("report worker_exit %q after the worker died", r.WorkerExit)
1309 }
1310 p.stop()
1311 })
1312
1313 t.Run("the worker sends no status for 3 s", func(t *testing.T) {
1314 f := newFixture(t)
1315 p := f.start(f.env)
1316 p.wait("/ready", http.StatusOK)
1317 pid := p.report().WorkerPID
1318 if err := syscall.Kill(pid, syscall.SIGSTOP); err != nil {
1319 t.Fatal(err)
1320 }
1321 stopped := time.Now()
1322 body := p.wait("/ready", http.StatusInternalServerError)
1323 if waited := time.Since(stopped); waited < 2*time.Second {
1324 t.Errorf("/ready failed %s after the worker stopped, before 3 s without status", waited)
1325 }
1326 if !strings.Contains(body, "worker: no status for") {
1327 t.Errorf("/ready with a stalled worker: %q", body)
1328 }
1329 syscall.Kill(pid, syscall.SIGCONT)
1330 p.wait("/ready", http.StatusOK)
1331 p.stop()
1332 })
1333}
1334
1335// TestReadyCoalescesTheStoreCheck: anonymous /ready traffic costs at most one
1336// store check per 5 s, however many requests arrive, and requests that land
1337// while a check is running share its result.
1338func TestReadyCoalescesTheStoreCheck(t *testing.T) {
1339 t.Parallel()
1340 f := newFixture(t)
1341 p := f.start(f.env)
1342 p.wait("/ready", http.StatusOK)
1343
1344 burst := func() {
1345 t.Helper()
1346 var wg sync.WaitGroup
1347 codes := make(chan int, 50)
1348 for i := 0; i < 50; i++ {
1349 wg.Add(1)
1350 go func() {
1351 defer wg.Done()
1352 code, _, err := p.do(http.MethodGet, "/ready", "", nil)
1353 if err != nil {
1354 code = 0
1355 }
1356 codes <- code
1357 }()
1358 }
1359 wg.Wait()
1360 close(codes)
1361 for c := range codes {
1362 if c != http.StatusOK {
1363 t.Errorf("a /ready in the burst answered %d", c)
1364 }
1365 }
1366 }
1367
1368 // A slow store, so the burst lands while a check is in flight.
1369 f.storeDelay.Store(300)
1370 time.Sleep(5100 * time.Millisecond) // past any cached result
1371 n0 := f.storeReq.Load()
1372 burst()
1373 n1 := f.storeReq.Load()
1374 if n1-n0 > 1 {
1375 t.Errorf("50 concurrent /ready calls made %d store requests, want at most 1", n1-n0)
1376 }
1377 burst()
1378 if n2 := f.storeReq.Load(); n2 != n1 {
1379 t.Errorf("a second burst inside the 5 s window made %d store requests, want 0", n2-n1)
1380 }
1381 time.Sleep(5100 * time.Millisecond)
1382 n3 := f.storeReq.Load()
1383 burst()
1384 if n4 := f.storeReq.Load(); n4-n3 > 1 {
1385 t.Errorf("the next window's burst made %d store requests, want at most 1", n4-n3)
1386 }
1387 p.stop()
1388}
1389