amadan

carlos / platform Public

cmd/moveprobe: the rehearsal probe app (§14.4)

A small app shaped like a customer app, for the move rehearsals: rows from
a background loop, a forked worker and timer fires, each stamped with its
region; instance.key in the files directory checked against the database;
the platform's CARLOS_INSTANCE_KEY fingerprinted at first boot and held to
it; an object written to the MOVEPROBE store binding and read back on
readiness; the sealed MOVEPROBE_SECRET; once-timers armed through the
control socket. "/" answers what /ready answers.

Per the preflight rulings: the store client uses the binding's static keys
only, with the endpoint optional and path-style when set (N5); the app
declares no schedules, /arm-once PUTs once-<n> with at, and fires take
their name and due time from the X-Carlos-Schedule headers (N6); a missing
CARLOS_REGION is home (N7); the worker detaches its stdio and exits 60 s
after its parent has gone (N9); before-write mode writes nothing at all
(N11); /arm-once and /report need the MOVEPROBE_SECRET bearer (N23).
Setsid sits behind //go:build !windows. Nothing here runs against a
deployment.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9b07ae0a5bc302e89f78041e5cb3b5ebb897efb8 parent e00efe0
5 files changed, +1974 −0
  • cmd/moveprobe/README.md +110 −0
  • cmd/moveprobe/detach_unix.go +9 −0
  • cmd/moveprobe/detach_windows.go +9 −0
  • cmd/moveprobe/main.go +828 −0
  • cmd/moveprobe/main_test.go +1018 −0
diff --git a/cmd/moveprobe/README.md b/cmd/moveprobe/README.md
new file mode 100644
index 000000000..927beaa65
--- /dev/null
+++ b/cmd/moveprobe/README.md
@@ -0,0 +1,110 @@
+# moveprobe
+
+moveprobe is a small app for rehearsing instance moves. It keeps state in
+every place a real app can, so that after a move you can check that each kind
+of state arrived. It is not part of the platform release. You build it and
+ship it like any other app.
+
+## What it keeps
+
+- **Database rows.** A background loop writes a row every half second. A
+ separate worker process writes one every second. Each timer fire writes one.
+ Every row records the region it was written in.
+- **A key file.** On first boot it writes `instance.key` into the instance's
+ files directory and records the same key in the database. Every boot checks
+ that the two still match.
+- **The platform's instance key.** On first boot it records a fingerprint of
+ `CARLOS_INSTANCE_KEY`. Every boot checks that the key is still set and
+ unchanged.
+- **An object in its store.** On every boot it writes `probe/<boot id>.txt`
+ to the store bound with the prefix `MOVEPROBE`, and reads it back on every
+ readiness check.
+- **A sealed secret.** It needs `MOVEPROBE_SECRET` to be set.
+- **Timers.** A once-timer you arm through the app, and a schedule you set
+ with `carlos schedule set`.
+
+## Build
+
+The rehearsal boxes are arm64:
+
+```sh
+CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o moveprobe ./cmd/moveprobe
+```
+
+## Deploy
+
+Run these signed in to the ops account, unless the rehearsal names another.
+Use the deployment's instance domain where these say `example.com`, and pick
+a host name no earlier rehearsal has used.
+
+```sh
+carlos ship --app moveprobe --label "moveprobe" ./moveprobe
+carlos instances enable --app moveprobe
+carlos store create --app moveprobe --env-prefix MOVEPROBE
+carlos secrets set --app moveprobe MOVEPROBE_SECRET=<a long random value>
+carlos schedule set --app moveprobe --name every-minute --every 1m --path /tick
+carlos instances create --app moveprobe --host mp1.example.com --region ie
+```
+
+Leave private storage off when you enable instances, and create the store
+without `--environment`. A move only carries the app's own store. It refuses
+an instance whose store belongs to an environment or to the instance itself.
+The account must be approved for object storage before `store create`
+provisions anything.
+
+Keep the value of `MOVEPROBE_SECRET` in your shell. It is also the password
+for `/arm-once` and `/report`.
+
+## Use it
+
+```sh
+curl https://mp1.example.com/ready
+curl -X POST -H "Authorization: Bearer $MOVEPROBE_SECRET" "https://mp1.example.com/arm-once?in=600"
+curl -H "Authorization: Bearer $MOVEPROBE_SECRET" https://mp1.example.com/report
+```
+
+| path | who can call it | what it does |
+|---|---|---|
+| `/` and `/ready` | anyone | 200 when the key file, the platform key, the stored object and the secret all check out. Otherwise 500 with every failed check. A move probes `/`. |
+| `GET /report` | `MOVEPROBE_SECRET` | JSON: row counts by kind and region with their lowest and highest sequence numbers, the timer fires by name and due time, both key fingerprints, the stored object, whether the secret is set, the worker's pid, and the region the app saw. |
+| `POST /arm-once?in=<seconds>` | `MOVEPROBE_SECRET` | Arms a once-timer named `once-<n>` that calls `/once` after that many seconds. Answers 503 when the instance has no control socket. |
+| `POST /tick`, `POST /once` | the platform | Timer fires. Each one writes a row with the timer's name and due time. |
+
+The background rows are numbered in one sequence that carries on across
+boots and regions. If no row is missing, the count of `bg` rows equals the
+highest `bg` sequence number.
+
+The region comes from `CARLOS_REGION`. On a box that has no region, the app
+uses `home`, and `/report` says so.
+
+## Make readiness fail on purpose
+
+`MOVEPROBE_FAIL_READY_IN=<region>` makes the app fail its readiness check in
+that region. The app reads it when it starts, so set it before you request
+the move, and remove it once the run's checks are recorded:
+
+```sh
+carlos env set --app moveprobe MOVEPROBE_FAIL_READY_IN=ie
+carlos env unset --app moveprobe MOVEPROBE_FAIL_READY_IN
+```
+
+There are two modes:
+
+- `MOVEPROBE_FAIL_READY_IN=ie` writes nothing at all in `ie`. It does not
+ open the database, touch the files directory, start the worker or call the
+ control socket. Every path answers 503. A rollback from here should be
+ clean.
+- `MOVEPROBE_FAIL_READY_IN=ie,after-write` writes one background row in
+ `ie`, starts the worker, and then answers 503 on `/` and `/ready`. A
+ rollback from here is dirty, and the source should end up holding that
+ row.
+
+In any other region the variable has no effect.
+
+## The worker
+
+The app starts a worker process in its own session, with no connection to
+the app's output. Stopping the app does not stop the worker. That is the
+point: it shows whether a move's stop really stops every process the
+instance started. Once the app has gone, the worker keeps writing for 60
+seconds and then exits by itself.
diff --git a/cmd/moveprobe/detach_unix.go b/cmd/moveprobe/detach_unix.go
new file mode 100644
index 000000000..596f3cc16
--- /dev/null
+++ b/cmd/moveprobe/detach_unix.go
@@ -0,0 +1,9 @@
+//go:build !windows
+
+package main
+
+import "syscall"
+
+// detachAttr puts the worker in its own session, so a signal to the main's
+// process group never reaches it.
+func detachAttr() *syscall.SysProcAttr { return &syscall.SysProcAttr{Setsid: true} }
diff --git a/cmd/moveprobe/detach_windows.go b/cmd/moveprobe/detach_windows.go
new file mode 100644
index 000000000..4aaa9a771
--- /dev/null
+++ b/cmd/moveprobe/detach_windows.go
@@ -0,0 +1,9 @@
+//go:build windows
+
+package main
+
+import "syscall"
+
+// detachAttr is nil on Windows, which has no sessions to leave; moveprobe
+// only runs on the platform's Linux boxes, and builds here so the tree does.
+func detachAttr() *syscall.SysProcAttr { return nil }
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go
new file mode 100644
index 000000000..30d2f23bf
--- /dev/null
+++ b/cmd/moveprobe/main.go
@@ -0,0 +1,828 @@
+// Command moveprobe is the moves rehearsal probe (moves spec §14.4): a small
+// app shaped like a customer app, deployed by the rehearsal runbook so that a
+// move can be shown to carry every kind of state an instance has.
+//
+// It holds state in each place an instance can:
+//
+// - the database: a background loop writes a row every 500 ms, a forked
+// worker writes one every second, and timer fires write one each;
+// - the files directory: instance.key, written at first boot and checked
+// against the database at every boot;
+// - the platform's instance key (CARLOS_INSTANCE_KEY), fingerprinted at
+// first boot and checked at every boot;
+// - its store binding (MOVEPROBE_*): an object written at boot and read
+// back on every readiness check;
+// - a sealed secret (MOVEPROBE_SECRET);
+// - timers: once-timers armed through the instance control socket, and a
+// declared schedule set by the operator with `carlos schedule set`.
+//
+// The exec backend starts it as `moveprobe --socket <path> --db <path>`
+// (internal/activator/backend_exec.go, spawnChildFrom). The files directory
+// is the database path without ".db" (the flat layout). "/" answers what
+// "/ready" answers, because a move probes "/" when moves.ready_path is unset.
+//
+// MOVEPROBE_FAIL_READY_IN=<region>[,after-write], read at startup, makes the
+// probe fail its readiness in the named region. With no suffix (before-write)
+// it writes nothing at all: no database open, no file, no worker, no socket
+// call. With ",after-write" it writes one background row, lets its worker
+// write, and then fails readiness. A box with no CARLOS_REGION is region
+// "home".
+//
+// The README beside this file has the build and deploy lines.
+package main
+
+import (
+ "bytes"
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/subtle"
+ "database/sql"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "io/fs"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "os/exec"
+ "os/signal"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/service/s3"
+
+ _ "modernc.org/sqlite"
+)
+
+const (
+ bgEvery = 500 * time.Millisecond
+ workerEvery = time.Second
+ // workerGrace is how long the worker keeps writing after its parent has
+ // gone. Long enough to outlive the main through a cut's stop and stop
+ // 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
+
+ modeNormal = "normal"
+ modeBeforeWrite = "before-write"
+ modeAfterWrite = "after-write"
+)
+
+func main() {
+ if len(os.Args) > 1 && os.Args[1] == "worker" {
+ os.Exit(runWorker(os.Args[2:]))
+ }
+ os.Exit(run(os.Args[1:]))
+}
+
+func logf(format string, args ...any) {
+ fmt.Fprintf(os.Stderr, "moveprobe: "+format+"\n", args...)
+}
+
+// region is CARLOS_REGION, or "home" when the box has no placement code and
+// the agent leaves the variable out.
+func region() (name, from string) {
+ if r := strings.TrimSpace(os.Getenv("CARLOS_REGION")); r != "" {
+ return r, "CARLOS_REGION"
+ }
+ return "home", "unset, treated as home"
+}
+
+// failMode reads MOVEPROBE_FAIL_READY_IN for this region. A suffix other
+// than after-write is read as before-write, the mode that writes nothing.
+func failMode(spec, here string) string {
+ spec = strings.TrimSpace(spec)
+ if spec == "" {
+ return modeNormal
+ }
+ where, suffix, hasSuffix := strings.Cut(spec, ",")
+ if strings.TrimSpace(where) != here {
+ return modeNormal
+ }
+ if hasSuffix && strings.TrimSpace(suffix) == modeAfterWrite {
+ return modeAfterWrite
+ }
+ if hasSuffix {
+ logf("MOVEPROBE_FAIL_READY_IN suffix %q is not after-write; using before-write", suffix)
+ }
+ return modeBeforeWrite
+}
+
+func run(args []string) int {
+ fl := flag.NewFlagSet("moveprobe", flag.ContinueOnError)
+ sock := fl.String("socket", "", "unix socket to listen on")
+ dbPath := fl.String("db", "", "the instance database")
+ if err := fl.Parse(args); err != nil {
+ return 2
+ }
+ if *sock == "" || *dbPath == "" {
+ logf("--socket and --db are required")
+ return 2
+ }
+ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+ defer stop()
+
+ here, from := region()
+ mode := failMode(os.Getenv("MOVEPROBE_FAIL_READY_IN"), here)
+ if mode == modeBeforeWrite {
+ logf("before-write mode in region %s: nothing is written", here)
+ return serve(ctx, *sock, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ http.Error(w, "before-write mode: this box writes nothing", http.StatusServiceUnavailable)
+ }))
+ }
+
+ a, err := boot(ctx, *dbPath, here, from, mode)
+ if err != nil {
+ logf("%v", err)
+ return 1
+ }
+ defer a.db.Close()
+
+ loopCtx, stopLoop := context.WithCancel(ctx)
+ defer stopLoop()
+ if mode == modeAfterWrite {
+ if err := a.insertSeq("bg"); err != nil {
+ logf("after-write: the bg row: %v", err)
+ }
+ } else {
+ go a.bgLoop(loopCtx)
+ }
+ a.forkWorker()
+ return serve(ctx, *sock, a.routes())
+}
+
+// serve listens on sock until ctx ends, then shuts down.
+func serve(ctx context.Context, sock string, h http.Handler) int {
+ if fi, err := os.Lstat(sock); err == nil && fi.Mode()&fs.ModeSocket != 0 {
+ os.Remove(sock)
+ }
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ logf("listen: %v", err)
+ return 1
+ }
+ srv := &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second}
+ errc := make(chan error, 1)
+ go func() { errc <- srv.Serve(ln) }()
+ select {
+ case err := <-errc:
+ logf("serve: %v", err)
+ return 1
+ case <-ctx.Done():
+ }
+ shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ srv.Shutdown(shutCtx)
+ return 0
+}
+
+type app struct {
+ db *sql.DB
+ dbPath string
+ filesDir string
+ host string
+ region string
+ regionFrom string
+ mode string
+ bootID string
+
+ fileKey string // meta instance_key; empty when the boot could not settle it
+ fileKeyErr error // set at boot when the file and the database disagree
+ platformFP string // SHA-256 of CARLOS_INSTANCE_KEY
+ platErr error
+
+ store *s3.Client
+ storeErr error
+ bucket string
+ objKey string
+ objBody string
+
+ mu sync.Mutex
+ objPut bool
+ workerPID int
+}
+
+func openDB(path string) (*sql.DB, error) {
+ db, err := sql.Open("sqlite", "file:"+url.PathEscape(path)+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)")
+ if err != nil {
+ return nil, err
+ }
+ db.SetMaxOpenConns(1)
+ return db, nil
+}
+
+func boot(ctx context.Context, dbPath, here, from, mode string) (*app, error) {
+ id := make([]byte, 8)
+ if _, err := rand.Read(id); err != nil {
+ return nil, err
+ }
+ a := &app{
+ dbPath: dbPath,
+ filesDir: strings.TrimSuffix(dbPath, ".db"),
+ host: os.Getenv("CARLOS_HOST"),
+ region: here,
+ regionFrom: from,
+ mode: mode,
+ bootID: hex.EncodeToString(id),
+ }
+ db, err := openDB(dbPath)
+ if err != nil {
+ return nil, fmt.Errorf("open %s: %w", dbPath, err)
+ }
+ a.db = db
+ for _, q := range []string{
+ `CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)`,
+ `CREATE TABLE IF NOT EXISTS rows (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ kind TEXT NOT NULL,
+ name TEXT NOT NULL DEFAULT '',
+ seq INTEGER NOT NULL DEFAULT 0,
+ due_at INTEGER NOT NULL DEFAULT 0,
+ fired_at INTEGER NOT NULL,
+ region TEXT NOT NULL)`,
+ `CREATE INDEX IF NOT EXISTS rows_kind_seq ON rows (kind, seq)`,
+ } {
+ if _, err := db.Exec(q); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("schema: %w", err)
+ }
+ }
+ if err := os.MkdirAll(a.filesDir, 0o700); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("files directory: %w", err)
+ }
+ if err := a.settleFileKey(); err != nil {
+ db.Close()
+ return nil, err
+ }
+ if err := a.settlePlatformKey(); err != nil {
+ db.Close()
+ return nil, err
+ }
+ a.setupStore()
+ if a.storeErr == nil {
+ putCtx, cancel := context.WithTimeout(ctx, storeWait)
+ if err := a.putObject(putCtx); err != nil {
+ logf("store: %v", err)
+ }
+ cancel()
+ }
+ return a, nil
+}
+
+func (a *app) meta(key string) (string, bool, error) {
+ var v string
+ err := a.db.QueryRow(`SELECT value FROM meta WHERE key = ?`, key).Scan(&v)
+ if errors.Is(err, sql.ErrNoRows) {
+ return "", false, nil
+ }
+ return v, err == nil, err
+}
+
+func (a *app) keyPath() string { return filepath.Join(a.filesDir, "instance.key") }
+
+// settleFileKey writes instance.key at first boot and records it in meta;
+// on a later boot it remembers meta's value for the readiness check. A file
+// with no database key is a mismatch, never a reason to adopt the file.
+func (a *app) settleFileKey() error {
+ v, found, err := a.meta("instance_key")
+ if err != nil {
+ return fmt.Errorf("meta: %w", err)
+ }
+ if found {
+ a.fileKey = v
+ return nil
+ }
+ if _, err := os.Lstat(a.keyPath()); err == nil {
+ a.fileKeyErr = errors.New("key mismatch: instance.key exists but the database has no key")
+ return nil
+ }
+ raw := make([]byte, 32)
+ if _, err := rand.Read(raw); err != nil {
+ return err
+ }
+ key := hex.EncodeToString(raw)
+ f, err := os.OpenFile(a.keyPath(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+ if err != nil {
+ return fmt.Errorf("instance.key: %w", err)
+ }
+ if _, err := f.WriteString(key + "\n"); err != nil {
+ f.Close()
+ return fmt.Errorf("instance.key: %w", err)
+ }
+ if err := f.Sync(); err != nil {
+ f.Close()
+ return fmt.Errorf("instance.key: %w", err)
+ }
+ if err := f.Close(); err != nil {
+ return fmt.Errorf("instance.key: %w", err)
+ }
+ if _, err := a.db.Exec(`INSERT INTO meta (key, value) VALUES ('instance_key', ?)`, key); err != nil {
+ return fmt.Errorf("meta: %w", err)
+ }
+ a.fileKey = key
+ return nil
+}
+
+// settlePlatformKey records CARLOS_INSTANCE_KEY's fingerprint the first time
+// it is seen and holds every later boot to it.
+func (a *app) settlePlatformKey() error {
+ k := os.Getenv("CARLOS_INSTANCE_KEY")
+ if k == "" {
+ a.platErr = errors.New("CARLOS_INSTANCE_KEY is not set")
+ return nil
+ }
+ a.platformFP = fingerprint(k)
+ v, found, err := a.meta("platform_key_sha256")
+ if err != nil {
+ return fmt.Errorf("meta: %w", err)
+ }
+ if !found {
+ if _, err := a.db.Exec(`INSERT INTO meta (key, value) VALUES ('platform_key_sha256', ?)`, a.platformFP); err != nil {
+ return fmt.Errorf("meta: %w", err)
+ }
+ return nil
+ }
+ if v != a.platformFP {
+ a.platErr = errors.New("CARLOS_INSTANCE_KEY differs from the one seen at first boot")
+ }
+ return nil
+}
+
+func fingerprint(s string) string {
+ sum := sha256.Sum256([]byte(s))
+ return hex.EncodeToString(sum[:])
+}
+
+// fileKeyNow is instance.key as it is on disk now, trimmed.
+func (a *app) fileKeyNow() (string, error) {
+ b, err := os.ReadFile(a.keyPath())
+ if err != nil {
+ return "", err
+ }
+ return strings.TrimSpace(string(b)), nil
+}
+
+func (a *app) checkFileKey() error {
+ if a.fileKeyErr != nil {
+ return a.fileKeyErr
+ }
+ got, err := a.fileKeyNow()
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ return errors.New("key mismatch: instance.key is missing")
+ case err != nil:
+ return fmt.Errorf("key mismatch: instance.key: %v", err)
+ case subtle.ConstantTimeCompare([]byte(got), []byte(a.fileKey)) != 1:
+ return errors.New("key mismatch: instance.key differs from the database")
+ }
+ return nil
+}
+
+// ---- the store ----------------------------------------------------------------
+
+type storeConfig struct {
+ bucket, region, endpoint, accessKey, secretKey string
+}
+
+// storeFromEnv reads the MOVEPROBE binding. The endpoint is optional: the
+// platform delivers it only when the deployment has one, and never on AWS.
+func storeFromEnv() (storeConfig, error) {
+ c := storeConfig{
+ bucket: os.Getenv("MOVEPROBE_BUCKET"),
+ region: os.Getenv("MOVEPROBE_REGION"),
+ endpoint: os.Getenv("MOVEPROBE_ENDPOINT"),
+ accessKey: os.Getenv("MOVEPROBE_ACCESS_KEY"),
+ secretKey: os.Getenv("MOVEPROBE_SECRET_KEY"),
+ }
+ var missing []string
+ for _, v := range []struct{ name, val string }{
+ {"MOVEPROBE_BUCKET", c.bucket},
+ {"MOVEPROBE_REGION", c.region},
+ {"MOVEPROBE_ACCESS_KEY", c.accessKey},
+ {"MOVEPROBE_SECRET_KEY", c.secretKey},
+ } {
+ if v.val == "" {
+ missing = append(missing, v.name)
+ }
+ }
+ if len(missing) > 0 {
+ return c, fmt.Errorf("store: %s not set", strings.Join(missing, ", "))
+ }
+ return c, nil
+}
+
+// newStoreClient builds the S3 client from the binding's keys alone. It
+// never loads the default credential chain: the app shares the agent's uid
+// on a flat box, so the chain could find the box's own role and read the
+// object back without the binding. Path-style only with an endpoint, as
+// objstore.S3ClientFor does.
+func newStoreClient(c storeConfig) *s3.Client {
+ return s3.New(s3.Options{
+ Region: c.region,
+ Credentials: aws.CredentialsProviderFunc(func(context.Context) (aws.Credentials, error) {
+ return aws.Credentials{AccessKeyID: c.accessKey, SecretAccessKey: c.secretKey, Source: "MOVEPROBE"}, nil
+ }),
+ BaseEndpoint: func() *string {
+ if c.endpoint == "" {
+ return nil
+ }
+ return aws.String(c.endpoint)
+ }(),
+ UsePathStyle: c.endpoint != "",
+ RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+ ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
+ })
+}
+
+func (a *app) setupStore() {
+ c, err := storeFromEnv()
+ if err != nil {
+ a.storeErr = err
+ return
+ }
+ a.store = newStoreClient(c)
+ a.bucket = c.bucket
+ a.objKey = "probe/" + a.bootID + ".txt"
+ a.objBody = fmt.Sprintf("moveprobe boot %s host %s region %s at %s\n", a.bootID, a.host, a.region, time.Now().UTC().Format(time.RFC3339))
+}
+
+func (a *app) putObject(ctx context.Context) error {
+ _, err := a.store.PutObject(ctx, &s3.PutObjectInput{
+ Bucket: aws.String(a.bucket),
+ Key: aws.String(a.objKey),
+ Body: bytes.NewReader([]byte(a.objBody)),
+ })
+ if err != nil {
+ return fmt.Errorf("put %s: %w", a.objKey, err)
+ }
+ a.mu.Lock()
+ a.objPut = true
+ a.mu.Unlock()
+ return nil
+}
+
+func (a *app) getObject(ctx context.Context) (string, error) {
+ out, err := a.store.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(a.bucket), Key: aws.String(a.objKey)})
+ if err != nil {
+ return "", fmt.Errorf("get %s: %w", a.objKey, err)
+ }
+ defer out.Body.Close()
+ b, err := io.ReadAll(io.LimitReader(out.Body, 1<<16))
+ if err != nil {
+ return "", fmt.Errorf("get %s: %w", a.objKey, err)
+ }
+ return string(b), nil
+}
+
+// checkStore puts the boot's object if the boot could not, then reads it
+// back.
+func (a *app) checkStore(ctx context.Context) error {
+ if a.storeErr != nil {
+ return a.storeErr
+ }
+ ctx, cancel := context.WithTimeout(ctx, storeWait)
+ defer cancel()
+ a.mu.Lock()
+ put := a.objPut
+ a.mu.Unlock()
+ if !put {
+ if err := a.putObject(ctx); err != nil {
+ return fmt.Errorf("store: %v", err)
+ }
+ }
+ got, err := a.getObject(ctx)
+ if err != nil {
+ return fmt.Errorf("store: %v", err)
+ }
+ if got != a.objBody {
+ return fmt.Errorf("store: %s reads back different bytes", a.objKey)
+ }
+ return nil
+}
+
+// ---- rows ---------------------------------------------------------------------
+
+// insertSeq writes the next row of a sequenced kind. The sequence is taken
+// inside the insert, so two writers (a main and a worker from an earlier
+// boot, or two generations) never take the same number.
+func (a *app) insertSeq(kind string) error {
+ return insertSeq(a.db, kind, a.region)
+}
+
+func insertSeq(db *sql.DB, kind, region string) error {
+ _, err := db.Exec(`INSERT INTO rows (kind, seq, fired_at, region)
+ SELECT ?, coalesce(max(seq), 0) + 1, ?, ? FROM rows WHERE kind = ?`,
+ kind, time.Now().Unix(), region, kind)
+ return err
+}
+
+func (a *app) bgLoop(ctx context.Context) {
+ t := time.NewTicker(bgEvery)
+ defer t.Stop()
+ for {
+ if err := a.insertSeq("bg"); err != nil {
+ logf("bg: %v", err)
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ }
+ }
+}
+
+// ---- the worker ---------------------------------------------------------------
+
+// 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.
+func (a *app) forkWorker() {
+ cmd := exec.Command(os.Args[0], "worker", "--db", a.dbPath, "--parent", strconv.Itoa(os.Getpid()))
+ cmd.SysProcAttr = detachAttr()
+ if err := cmd.Start(); err != nil {
+ logf("worker: %v", err)
+ return
+ }
+ a.mu.Lock()
+ a.workerPID = cmd.Process.Pid
+ a.mu.Unlock()
+ go cmd.Wait()
+}
+
+// runWorker writes a worker row every second. Once its parent has gone it
+// keeps writing for workerGrace and then exits.
+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")
+ if err := fl.Parse(args); err != nil || *dbPath == "" || *parent <= 0 {
+ return 2
+ }
+ here, _ := region()
+ db, err := openDB(*dbPath)
+ if err != nil {
+ return 1
+ }
+ defer db.Close()
+ var orphaned time.Time
+ t := time.NewTicker(workerEvery)
+ defer t.Stop()
+ for {
+ if orphaned.IsZero() && os.Getppid() != *parent {
+ orphaned = time.Now()
+ }
+ if !orphaned.IsZero() && time.Since(orphaned) >= workerGrace {
+ return 0
+ }
+ insertSeq(db, "worker", here)
+ <-t.C
+ }
+}
+
+// ---- HTTP ---------------------------------------------------------------------
+
+func (a *app) routes() http.Handler {
+ mux := http.NewServeMux()
+ mux.HandleFunc("/{$}", a.handleReady)
+ mux.HandleFunc("/ready", a.handleReady)
+ mux.HandleFunc("GET /report", a.bearer(os.Getenv("MOVEPROBE_SECRET"), a.handleReport))
+ mux.HandleFunc("POST /arm-once", a.bearer(os.Getenv("MOVEPROBE_SECRET"), a.handleArmOnce))
+ // The platform's runner signs each fire with the instance's admin token.
+ mux.HandleFunc("POST /tick", a.bearer(os.Getenv("CARLOS_ADMIN_TOKEN"), a.fired("tick")))
+ mux.HandleFunc("POST /once", a.bearer(os.Getenv("CARLOS_ADMIN_TOKEN"), a.fired("once")))
+ return mux
+}
+
+// bearer admits a request carrying "Authorization: Bearer <want>". An empty
+// want admits nothing.
+func (a *app) bearer(want string, next http.HandlerFunc) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ got := sha256.Sum256([]byte(r.Header.Get("Authorization")))
+ exp := sha256.Sum256([]byte("Bearer " + want))
+ if want == "" || subtle.ConstantTimeCompare(got[:], exp[:]) != 1 {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ next(w, r)
+ }
+}
+
+// readiness is /ready's answer: every check that fails, or ready.
+func (a *app) readiness(ctx context.Context) (int, string) {
+ if a.mode == modeAfterWrite {
+ return http.StatusServiceUnavailable, "after-write mode: not ready here"
+ }
+ var fails []string
+ if err := a.checkFileKey(); err != nil {
+ fails = append(fails, err.Error())
+ }
+ if a.platErr != nil {
+ fails = append(fails, a.platErr.Error())
+ }
+ if os.Getenv("MOVEPROBE_SECRET") == "" {
+ fails = append(fails, "MOVEPROBE_SECRET is not set")
+ }
+ if err := a.checkStore(ctx); err != nil {
+ fails = append(fails, err.Error())
+ }
+ if len(fails) > 0 {
+ return http.StatusInternalServerError, strings.Join(fails, "; ")
+ }
+ return http.StatusOK, "ready"
+}
+
+func (a *app) handleReady(w http.ResponseWriter, r *http.Request) {
+ code, msg := a.readiness(r.Context())
+ if code != http.StatusOK {
+ http.Error(w, msg, code)
+ return
+ }
+ fmt.Fprintln(w, msg)
+}
+
+func (a *app) fired(kind string) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ name := r.Header.Get("X-Carlos-Schedule")
+ due, err := strconv.ParseInt(r.Header.Get("X-Carlos-Schedule-At"), 10, 64)
+ if name == "" || err != nil {
+ http.Error(w, "want X-Carlos-Schedule and X-Carlos-Schedule-At", http.StatusBadRequest)
+ return
+ }
+ if _, err := a.db.Exec(`INSERT INTO rows (kind, name, due_at, fired_at, region) VALUES (?, ?, ?, ?, ?)`,
+ kind, name, due, time.Now().Unix(), a.region); err != nil {
+ http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ fmt.Fprintln(w, "ok")
+ }
+}
+
+// handleArmOnce registers once-<n> for /once, due in the given seconds,
+// through the instance control socket.
+func (a *app) handleArmOnce(w http.ResponseWriter, r *http.Request) {
+ in, err := strconv.ParseInt(r.URL.Query().Get("in"), 10, 64)
+ if err != nil || in < 0 || in > 400*24*3600 {
+ http.Error(w, "in: want whole seconds, at most 400 days", http.StatusBadRequest)
+ return
+ }
+ control := os.Getenv("CARLOS_CONTROL_SOCKET")
+ if control == "" {
+ http.Error(w, "no control socket: CARLOS_CONTROL_SOCKET is not set", http.StatusServiceUnavailable)
+ return
+ }
+ token := os.Getenv("CARLOS_ADMIN_TOKEN")
+ if token == "" {
+ http.Error(w, "CARLOS_ADMIN_TOKEN is not set", http.StatusServiceUnavailable)
+ return
+ }
+ var n int64
+ if err := a.db.QueryRow(`INSERT INTO meta (key, value) VALUES ('once_seq', '1')
+ ON CONFLICT (key) DO UPDATE SET value = CAST(value AS INTEGER) + 1
+ RETURNING CAST(value AS INTEGER)`).Scan(&n); err != nil {
+ http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ name := fmt.Sprintf("once-%d", n)
+ at := time.Now().Add(time.Duration(in) * time.Second).UTC().Truncate(time.Second)
+ body, _ := json.Marshal(map[string]string{"at": at.Format(time.RFC3339), "path": "/once"})
+
+ ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
+ defer cancel()
+ req, err := http.NewRequestWithContext(ctx, http.MethodPut, "http://carlos/schedule/"+name, bytes.NewReader(body))
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ req.Header.Set("Authorization", "Bearer "+token)
+ req.Header.Set("Content-Type", "application/json")
+ client := &http.Client{Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ var d net.Dialer
+ return d.DialContext(ctx, "unix", control)
+ },
+ }}
+ resp, err := client.Do(req)
+ if err != nil {
+ http.Error(w, "control socket: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ defer resp.Body.Close()
+ reply, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+ if resp.StatusCode/100 != 2 {
+ http.Error(w, fmt.Sprintf("control socket: %d %s", resp.StatusCode, strings.TrimSpace(string(reply))), http.StatusBadGateway)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(map[string]any{"name": name, "at": at.Format(time.RFC3339), "due_at": at.Unix()})
+}
+
+type rowCount struct {
+ Kind string `json:"kind"`
+ Region string `json:"region"`
+ Count int64 `json:"count"`
+ MinSeq int64 `json:"min_seq"`
+ MaxSeq int64 `json:"max_seq"`
+}
+
+type firedCount struct {
+ Kind string `json:"kind"`
+ Name string `json:"name"`
+ DueAt int64 `json:"due_at"`
+ Count int64 `json:"count"`
+}
+
+type objectReport struct {
+ Key string `json:"key"`
+ Body string `json:"body"`
+ Error string `json:"error,omitempty"`
+}
+
+type report struct {
+ Host string `json:"host"`
+ Region string `json:"region"`
+ RegionFrom string `json:"region_from"`
+ Mode string `json:"mode"`
+ BootID string `json:"boot_id"`
+ Rows []rowCount `json:"rows"`
+ Fired []firedCount `json:"fired"`
+ FileKeySHA256 string `json:"file_key_sha256"`
+ PlatformKeySHA256 string `json:"platform_key_sha256"`
+ Object objectReport `json:"object"`
+ HasSecret bool `json:"has_secret"`
+ WorkerPID int `json:"worker_pid"`
+}
+
+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") != "",
+ }
+ a.mu.Lock()
+ rep.WorkerPID = a.workerPID
+ a.mu.Unlock()
+ if k, err := a.fileKeyNow(); err == nil {
+ 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)
+ 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 {
+ rep.Object.Error = a.storeErr.Error()
+ } else {
+ ctx, cancel := context.WithTimeout(r.Context(), storeWait)
+ body, err := a.getObject(ctx)
+ cancel()
+ rep.Object.Body = body
+ if err != nil {
+ rep.Object.Error = err.Error()
+ }
+ }
+ w.Header().Set("Content-Type", "application/json")
+ enc := json.NewEncoder(w)
+ enc.SetIndent("", " ")
+ enc.Encode(rep)
+}
diff --git a/cmd/moveprobe/main_test.go b/cmd/moveprobe/main_test.go
new file mode 100644
index 000000000..a66e0e6d6
--- /dev/null
+++ b/cmd/moveprobe/main_test.go
@@ -0,0 +1,1018 @@
+//go:build linux
+
+package main
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "database/sql"
+ "encoding/hex"
+ "encoding/json"
+ "encoding/pem"
+ "fmt"
+ "io"
+ "io/fs"
+ "math/big"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/schedule"
+
+ _ "modernc.org/sqlite"
+)
+
+const (
+ testHost = "h1.example.com"
+ adminToken = "admin-token-for-tests"
+ probeSecret = "probe-secret-for-tests"
+ platformKey = "platform-instance-key-for-tests"
+ accessKey = "AKIDMOVEPROBETEST"
+ bucketName = "probe-bucket"
+ storeRegion = "eu-west-1"
+)
+
+// binPath is the moveprobe binary the tests start, built once in TestMain,
+// so every test starts the app the way the exec backend does: a separate
+// process given --socket and --db.
+var binPath string
+
+func TestMain(m *testing.M) {
+ dir, err := os.MkdirTemp("", "mpbin")
+ if err != nil {
+ fmt.Fprintln(os.Stderr, err)
+ os.Exit(1)
+ }
+ binPath = filepath.Join(dir, "moveprobe")
+ if out, err := exec.Command("go", "build", "-o", binPath, ".").CombinedOutput(); err != nil {
+ fmt.Fprintf(os.Stderr, "build moveprobe: %v\n%s", err, out)
+ os.RemoveAll(dir)
+ os.Exit(1)
+ }
+ code := m.Run()
+ os.RemoveAll(dir)
+ os.Exit(code)
+}
+
+// shortTempDir keeps unix socket paths under the kernel's length limit
+// (as internal/activator/rolelaunch_root_linux_test.go does); t.TempDir's
+// names carry the test's name and can pass 107 bytes.
+func shortTempDir(t *testing.T) string {
+ t.Helper()
+ dir, err := os.MkdirTemp("", "mp")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+}
+
+// fixture is one instance's world: its data directory (the flat layout,
+// <dir>/h1.db and <dir>/h1/), a test S3 holding its store, and a control
+// socket served by the platform's own schedule verbs over a registry.
+type fixture struct {
+ t *testing.T
+ sockDir string
+ dataDir string
+ db string
+ bucket *leasetest.VersionedBucket
+ store http.Handler
+ storeURL string
+ reg *registry.Registry
+ control string
+ env map[string]string
+
+ mu sync.Mutex
+ auths []string
+ storeReq atomic.Int64
+ starts int
+}
+
+func newFixture(t *testing.T) *fixture {
+ t.Helper()
+ f := &fixture{t: t, sockDir: shortTempDir(t), dataDir: t.TempDir()}
+ f.db = filepath.Join(f.dataDir, "h1.db")
+
+ f.bucket = leasetest.NewVersionedBucket()
+ f.bucket.AddBucket(bucketName)
+ s3h, _ := leasetest.NewS3Server(f.bucket)
+ f.store = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ f.storeReq.Add(1)
+ f.mu.Lock()
+ f.auths = append(f.auths, r.Header.Get("Authorization"))
+ f.mu.Unlock()
+ s3h.ServeHTTP(w, r)
+ })
+ srv := httptest.NewServer(f.store)
+ t.Cleanup(srv.Close)
+ f.storeURL = srv.URL
+
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ f.reg = reg
+ f.control = filepath.Join(f.sockDir, "c.sock")
+ ln, err := net.Listen("unix", f.control)
+ if err != nil {
+ t.Fatal(err)
+ }
+ mux := http.NewServeMux()
+ schedule.Mount(mux, testHost, reg, func() (string, error) { return adminToken, nil }, time.Now)
+ cs := &http.Server{Handler: mux}
+ go cs.Serve(ln)
+ t.Cleanup(func() { cs.Close() })
+
+ // The environment agentOwnedVars composes over <host>.env: the control
+ // socket (when the listener bound), the purge channel, the region (when
+ // the box has one) and the host; then the env file's own variables.
+ f.env = map[string]string{
+ "CARLOS_CONTROL_SOCKET": f.control,
+ "CARLOS_PURGE": "header",
+ "CARLOS_REGION": "uk",
+ "CARLOS_HOST": testHost,
+ "CARLOS_ADMIN_TOKEN": adminToken,
+ "CARLOS_INSTANCE_KEY": platformKey,
+ "MOVEPROBE_BUCKET": bucketName,
+ "MOVEPROBE_REGION": storeRegion,
+ "MOVEPROBE_ENDPOINT": f.storeURL,
+ "MOVEPROBE_ACCESS_KEY": accessKey,
+ "MOVEPROBE_SECRET_KEY": "secret-key-for-tests",
+ "MOVEPROBE_SECRET": probeSecret,
+ }
+ // Registered first, so it runs last: every probe's main is gone by then.
+ t.Cleanup(func() { killWorkers(f.db) })
+ return f
+}
+
+func (f *fixture) seenAuth() []string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]string(nil), f.auths...)
+}
+
+// envWith is the fixture's environment with changes applied; an empty
+// value removes the variable.
+func (f *fixture) envWith(changes map[string]string) map[string]string {
+ out := map[string]string{}
+ for k, v := range f.env {
+ out[k] = v
+ }
+ for k, v := range changes {
+ if v == "" {
+ delete(out, k)
+ } else {
+ out[k] = v
+ }
+ }
+ return out
+}
+
+// childEnv mirrors the backend's: the parent's environment without any
+// CARLOS_ variable, then the instance's variables. The test also drops the
+// probe's and AWS's own names and every proxy setting, so nothing from the
+// shell running the tests reaches the app.
+func childEnv(vars map[string]string) []string {
+ var out []string
+ for _, kv := range os.Environ() {
+ k, _, _ := strings.Cut(kv, "=")
+ switch {
+ case strings.HasPrefix(k, "CARLOS_"), strings.HasPrefix(k, "MOVEPROBE_"), strings.HasPrefix(k, "AWS_"):
+ continue
+ case strings.EqualFold(k, "HTTP_PROXY"), strings.EqualFold(k, "HTTPS_PROXY"), strings.EqualFold(k, "NO_PROXY"), k == "SSL_CERT_FILE", k == "SSL_CERT_DIR":
+ continue
+ }
+ out = append(out, kv)
+ }
+ for k, v := range vars {
+ out = append(out, k+"="+v)
+ }
+ return out
+}
+
+type lockedBuffer struct {
+ mu sync.Mutex
+ b bytes.Buffer
+}
+
+func (l *lockedBuffer) Write(p []byte) (int, error) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.b.Write(p)
+}
+
+func (l *lockedBuffer) String() string {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.b.String()
+}
+
+type probe struct {
+ t *testing.T
+ cmd *exec.Cmd
+ sock string
+ out *lockedBuffer
+ done chan struct{}
+}
+
+// start runs the binary exactly as spawnChildFrom does:
+// exec.Command(bin, "--socket", sock, "--db", dbPath) with the composed env.
+// Its stdout and stderr are pipes, as the backend's tee makes them.
+func (f *fixture) start(env map[string]string) *probe {
+ f.t.Helper()
+ f.starts++
+ p := &probe{t: f.t, sock: filepath.Join(f.sockDir, fmt.Sprintf("a%d.sock", f.starts)), out: &lockedBuffer{}, done: make(chan struct{})}
+ p.cmd = exec.Command(binPath, "--socket", p.sock, "--db", f.db)
+ p.cmd.Env = childEnv(env)
+ p.cmd.Stdout = p.out
+ p.cmd.Stderr = p.out
+ if err := p.cmd.Start(); err != nil {
+ f.t.Fatal(err)
+ }
+ go func() { p.cmd.Wait(); close(p.done) }()
+ f.t.Cleanup(p.kill)
+ return p
+}
+
+func (p *probe) exited() bool {
+ select {
+ case <-p.done:
+ return true
+ default:
+ return false
+ }
+}
+
+func (p *probe) kill() {
+ if p.exited() {
+ return
+ }
+ p.cmd.Process.Kill()
+ <-p.done
+}
+
+// stop is the backend's Stop: SIGTERM to the leader only.
+func (p *probe) stop() {
+ p.t.Helper()
+ if p.exited() {
+ return
+ }
+ p.cmd.Process.Signal(syscall.SIGTERM)
+ select {
+ case <-p.done:
+ case <-time.After(10 * time.Second):
+ p.t.Errorf("moveprobe did not exit on SIGTERM within 10s; output:\n%s", p.out)
+ p.kill()
+ }
+}
+
+func unixClient(sock string) *http.Client {
+ return &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ var d net.Dialer
+ return d.DialContext(ctx, "unix", sock)
+ },
+ }}
+}
+
+func (p *probe) do(method, path, bearer string, hdr map[string]string) (int, string, error) {
+ req, err := http.NewRequest(method, "http://"+testHost+path, nil)
+ if err != nil {
+ return 0, "", err
+ }
+ if bearer != "" {
+ req.Header.Set("Authorization", "Bearer "+bearer)
+ }
+ for k, v := range hdr {
+ req.Header.Set(k, v)
+ }
+ resp, err := unixClient(p.sock).Do(req)
+ if err != nil {
+ return 0, "", err
+ }
+ defer resp.Body.Close()
+ b, _ := io.ReadAll(resp.Body)
+ return resp.StatusCode, string(b), nil
+}
+
+// wait polls path until it answers want, and fails the test with the last
+// answer and the app's output otherwise.
+func (p *probe) wait(path string, want int) string {
+ p.t.Helper()
+ deadline := time.Now().Add(20 * time.Second)
+ var last string
+ for time.Now().Before(deadline) {
+ code, body, err := p.do(http.MethodGet, path, "", nil)
+ if err == nil && code == want {
+ return body
+ }
+ last = fmt.Sprintf("code %d body %q err %v", code, body, err)
+ if p.exited() {
+ break
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ p.t.Fatalf("GET %s never answered %d; last: %s; output:\n%s", path, want, last, p.out)
+ return ""
+}
+
+type reportDoc struct {
+ Region string `json:"region"`
+ RegionFrom string `json:"region_from"`
+ Mode string `json:"mode"`
+ Rows []struct {
+ Kind string `json:"kind"`
+ Region string `json:"region"`
+ Count int64 `json:"count"`
+ MinSeq int64 `json:"min_seq"`
+ MaxSeq int64 `json:"max_seq"`
+ } `json:"rows"`
+ Fired []struct {
+ Kind string `json:"kind"`
+ Name string `json:"name"`
+ DueAt int64 `json:"due_at"`
+ Count int64 `json:"count"`
+ } `json:"fired"`
+ FileKeySHA256 string `json:"file_key_sha256"`
+ PlatformKeySHA256 string `json:"platform_key_sha256"`
+ Object struct {
+ Key string `json:"key"`
+ Body string `json:"body"`
+ Error string `json:"error"`
+ } `json:"object"`
+ HasSecret bool `json:"has_secret"`
+ WorkerPID int `json:"worker_pid"`
+}
+
+func (r reportDoc) count(kind, region string) int64 {
+ for _, row := range r.Rows {
+ if row.Kind == kind && row.Region == region {
+ return row.Count
+ }
+ }
+ return 0
+}
+
+func (p *probe) report() reportDoc {
+ p.t.Helper()
+ code, body, err := p.do(http.MethodGet, "/report", probeSecret, nil)
+ if err != nil || code != http.StatusOK {
+ p.t.Fatalf("/report: code %d err %v body %q; output:\n%s", code, err, body, p.out)
+ }
+ var r reportDoc
+ if err := json.Unmarshal([]byte(body), &r); err != nil {
+ p.t.Fatalf("/report: %v in %q", err, body)
+ }
+ return r
+}
+
+// workerPIDs lists the pids whose command line starts with the test binary
+// and "worker" and names db.
+func workerPIDs(db string) []int {
+ var pids []int
+ ents, _ := os.ReadDir("/proc")
+ for _, e := range ents {
+ pid, err := strconv.Atoi(e.Name())
+ if err != nil {
+ continue
+ }
+ raw, err := os.ReadFile(filepath.Join("/proc", e.Name(), "cmdline"))
+ if err != nil {
+ continue
+ }
+ args := strings.Split(strings.TrimRight(string(raw), "\x00"), "\x00")
+ if len(args) < 2 || args[0] != binPath || args[1] != "worker" || !strings.Contains(string(raw), db) {
+ continue
+ }
+ if alive(pid) {
+ pids = append(pids, pid)
+ }
+ }
+ return pids
+}
+
+// alive is true while pid exists and is neither a zombie nor dead.
+func alive(pid int) bool {
+ raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
+ if err != nil {
+ return false
+ }
+ i := strings.LastIndexByte(string(raw), ')')
+ if i < 0 || i+2 >= len(raw) {
+ return false
+ }
+ state := raw[i+2]
+ return state != 'Z' && state != 'X'
+}
+
+func killWorkers(db string) {
+ for _, pid := range workerPIDs(db) {
+ syscall.Kill(pid, syscall.SIGKILL)
+ }
+ deadline := time.Now().Add(5 * time.Second)
+ for len(workerPIDs(db)) > 0 && time.Now().Before(deadline) {
+ time.Sleep(50 * time.Millisecond)
+ }
+}
+
+func testDB(t *testing.T, db string) *sql.DB {
+ t.Helper()
+ d, err := sql.Open("sqlite", "file:"+db+"?_pragma=busy_timeout(5000)")
+ if err != nil {
+ t.Fatal(err)
+ }
+ return d
+}
+
+func countRows(t *testing.T, db, kind, region string) int64 {
+ t.Helper()
+ d := testDB(t, db)
+ defer d.Close()
+ var n int64
+ if err := d.QueryRow(`SELECT count(*) FROM rows WHERE kind=? AND region=?`, kind, region).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+
+// checkpoint folds the WAL into the database file and closes, so the file
+// is what a move would install.
+func checkpoint(t *testing.T, db string) {
+ t.Helper()
+ d := testDB(t, db)
+ if _, err := d.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
+ t.Fatal(err)
+ }
+ d.Close()
+}
+
+// snapshot is every entry under dir with its size and SHA-256: the
+// database file's digest and the files directory's listing, and anything
+// else that appears beside them (a -wal or -shm file included).
+func snapshot(t *testing.T, dir string) string {
+ t.Helper()
+ var lines []string
+ err := filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ rel, _ := filepath.Rel(dir, p)
+ if d.IsDir() {
+ lines = append(lines, rel+"/")
+ return nil
+ }
+ b, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ sum := sha256.Sum256(b)
+ lines = append(lines, fmt.Sprintf("%s %d %x", rel, len(b), sum))
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ sort.Strings(lines)
+ return strings.Join(lines, "\n")
+}
+
+func sha(s string) string {
+ sum := sha256.Sum256([]byte(s))
+ return hex.EncodeToString(sum[:])
+}
+
+// fire delivers a timer the way the platform's runner does: the request
+// schedule.BuildRequest builds, sent over the instance's socket.
+func (p *probe) fire(tm registry.Timer, token string) (int, string) {
+ p.t.Helper()
+ req, err := schedule.BuildRequest(registry.Route{Host: testHost}, tm, token, "test")
+ if err != nil {
+ p.t.Fatal(err)
+ }
+ resp, err := unixClient(p.sock).Do(req)
+ if err != nil {
+ p.t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ b, _ := io.ReadAll(resp.Body)
+ return resp.StatusCode, string(b)
+}
+
+// ---- the launch contract --------------------------------------------------
+
+func TestMoveprobeLaunchContract(t *testing.T) {
+ t.Run("the agent's variables and a store endpoint", func(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.env)
+ if body := p.wait("/ready", http.StatusOK); !strings.Contains(body, "ready") {
+ t.Errorf("/ready body %q", body)
+ }
+ // The platform probes "/" when moves.ready_path is unset.
+ p.wait("/", http.StatusOK)
+
+ vs, err := f.bucket.ListVersions(context.Background(), bucketName, "probe/")
+ if err != nil || len(vs) != 1 {
+ t.Fatalf("probe objects: %v %v", vs, err)
+ }
+ for _, a := range f.seenAuth() {
+ if !strings.Contains(a, "Credential="+accessKey+"/") {
+ t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
+ }
+ }
+
+ // /arm-once and /report need the probe's bearer.
+ if code, _, _ := p.do(http.MethodGet, "/report", "", nil); code != http.StatusUnauthorized {
+ t.Errorf("GET /report without a bearer: %d, want 401", code)
+ }
+ if code, _, _ := p.do(http.MethodGet, "/report", "wrong", nil); code != http.StatusUnauthorized {
+ t.Errorf("GET /report with the wrong bearer: %d, want 401", code)
+ }
+ if code, _, _ := p.do(http.MethodPost, "/arm-once?in=600", "", nil); code != http.StatusUnauthorized {
+ t.Errorf("POST /arm-once without a bearer: %d, want 401", code)
+ }
+ before := time.Now().Unix()
+ code, body, err := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil)
+ if err != nil || code != http.StatusOK {
+ t.Fatalf("POST /arm-once: %d %q %v", code, body, err)
+ }
+ tm, err := f.reg.GetTimer(testHost, "once-1")
+ if err != nil {
+ t.Fatalf("once-1 not registered through the control socket: %v (arm answered %q)", err, body)
+ }
+ if tm.Kind != schedule.KindOnce || tm.Path != "/once" || tm.DueAt < before+600 || tm.DueAt > time.Now().Unix()+601 {
+ t.Errorf("once-1 = %+v, want a once-timer for /once due in 600s", tm)
+ }
+ if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
+ t.Errorf("the second arm: %d %q", code, body)
+ }
+ if _, err := f.reg.GetTimer(testHost, "once-2"); err != nil {
+ t.Errorf("the second arm is not once-2: %v", err)
+ }
+
+ r := p.report()
+ if r.Region != "uk" || r.RegionFrom != "CARLOS_REGION" {
+ t.Errorf("report region %q from %q, want uk from CARLOS_REGION", r.Region, r.RegionFrom)
+ }
+ if r.PlatformKeySHA256 != sha(platformKey) || r.FileKeySHA256 == "" || !r.HasSecret || r.Object.Body == "" || r.Object.Error != "" {
+ t.Errorf("report = %+v", r)
+ }
+ p.stop()
+ })
+
+ t.Run("no CARLOS_REGION and no control socket", func(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "CARLOS_CONTROL_SOCKET": ""}))
+ p.wait("/ready", http.StatusOK)
+ code, body, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil)
+ if code != http.StatusServiceUnavailable || !strings.Contains(body, "CARLOS_CONTROL_SOCKET") {
+ t.Errorf("/arm-once with no control socket: %d %q, want 503 naming CARLOS_CONTROL_SOCKET", code, body)
+ }
+ p.wait("/ready", http.StatusOK) // unaffected
+ r := p.report()
+ if r.Region != "home" || !strings.Contains(r.RegionFrom, "unset") {
+ t.Errorf("report region %q from %q, want home, with CARLOS_REGION unset", r.Region, r.RegionFrom)
+ }
+ time.Sleep(700 * time.Millisecond)
+ if n := countRows(t, f.db, "bg", "home"); n == 0 {
+ t.Errorf("no bg rows written as region home")
+ }
+ p.stop()
+ })
+
+ t.Run("no MOVEPROBE_ENDPOINT: AWS's own host, the delivered keys", func(t *testing.T) {
+ f := newFixture(t)
+ proxy, caFile, connects := awsLookalike(t, f.store)
+ p := f.start(f.envWith(map[string]string{
+ "MOVEPROBE_ENDPOINT": "",
+ "HTTPS_PROXY": proxy,
+ "SSL_CERT_FILE": caFile,
+ }))
+ p.wait("/ready", http.StatusOK)
+ want := bucketName + ".s3." + storeRegion + ".amazonaws.com:443"
+ got := connects()
+ if len(got) == 0 {
+ t.Fatalf("no store request went to AWS's host")
+ }
+ for _, h := range got {
+ if h != want {
+ t.Errorf("store request to %q, want %q (virtual-hosted, no endpoint)", h, want)
+ }
+ }
+ for _, a := range f.seenAuth() {
+ if !strings.Contains(a, "Credential="+accessKey+"/") {
+ t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
+ }
+ }
+ p.stop()
+ })
+
+ t.Run("the store keys come only from MOVEPROBE_", func(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.envWith(map[string]string{
+ "MOVEPROBE_ACCESS_KEY": "",
+ "AWS_ACCESS_KEY_ID": accessKey,
+ "AWS_SECRET_ACCESS_KEY": "x",
+ "AWS_ENDPOINT_URL": f.storeURL,
+ "AWS_ENDPOINT_URL_S3": f.storeURL,
+ }))
+ body := p.wait("/ready", http.StatusInternalServerError)
+ if !strings.Contains(body, "MOVEPROBE_ACCESS_KEY") {
+ t.Errorf("/ready body %q does not name MOVEPROBE_ACCESS_KEY", body)
+ }
+ if n := f.storeReq.Load(); n != 0 {
+ t.Errorf("%d store requests made without the binding's keys", n)
+ }
+ p.stop()
+ })
+
+ t.Run("the sealed secret is required", func(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.envWith(map[string]string{"MOVEPROBE_SECRET": ""}))
+ body := p.wait("/ready", http.StatusInternalServerError)
+ if !strings.Contains(body, "MOVEPROBE_SECRET") {
+ t.Errorf("/ready body %q does not name MOVEPROBE_SECRET", body)
+ }
+ p.stop()
+ })
+}
+
+// awsLookalike stands in for S3's regional host: an HTTPS proxy whose
+// CONNECT tunnels end at a TLS server holding a certificate for
+// *.s3.<region>.amazonaws.com, signed by a CA written to the returned file.
+// The tunnelled requests are virtual-hosted; they reach store path-style.
+func awsLookalike(t *testing.T, store http.Handler) (proxyURL, caFile string, connects func() []string) {
+ t.Helper()
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(1),
+ Subject: pkix.Name{CommonName: "moveprobe test s3"},
+ NotBefore: time.Now().Add(-time.Hour),
+ NotAfter: time.Now().Add(time.Hour),
+ DNSNames: []string{"*.s3." + storeRegion + ".amazonaws.com", "s3." + storeRegion + ".amazonaws.com"},
+ KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+ BasicConstraintsValid: true,
+ IsCA: true,
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ caFile = filepath.Join(t.TempDir(), "ca.pem")
+ if err := os.WriteFile(caFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ tlsCfg := &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}}
+
+ vhost := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ bucket, _, _ := strings.Cut(r.Host, ".")
+ r.URL.Path = "/" + bucket + r.URL.Path
+ r.URL.RawPath = ""
+ store.ServeHTTP(w, r)
+ })
+ var mu sync.Mutex
+ var hosts []string
+ proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodConnect {
+ http.Error(w, "CONNECT only", http.StatusMethodNotAllowed)
+ return
+ }
+ mu.Lock()
+ hosts = append(hosts, r.Host)
+ mu.Unlock()
+ conn, brw, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ return
+ }
+ brw.WriteString("HTTP/1.1 200 Connection established\r\n\r\n")
+ brw.Flush()
+ srv := &http.Server{Handler: vhost}
+ srv.Serve(&oneConn{c: tls.Server(&bufferedConn{Conn: conn, r: brw.Reader}, tlsCfg)})
+ }))
+ t.Cleanup(proxy.Close)
+ return proxy.URL, caFile, func() []string {
+ mu.Lock()
+ defer mu.Unlock()
+ return append([]string(nil), hosts...)
+ }
+}
+
+type bufferedConn struct {
+ net.Conn
+ r *bufio.Reader
+}
+
+func (b *bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) }
+
+// oneConn is a listener that yields one connection; Serve returns once it
+// has been handed out and the connection is served on its own goroutine.
+type oneConn struct {
+ c net.Conn
+ used bool
+}
+
+func (o *oneConn) Accept() (net.Conn, error) {
+ if o.used {
+ return nil, io.EOF
+ }
+ o.used = true
+ return o.c, nil
+}
+func (o *oneConn) Close() error { return nil }
+func (o *oneConn) Addr() net.Addr { return o.c.LocalAddr() }
+
+// ---- the forked worker ------------------------------------------------------
+
+func TestMoveprobeWorkerOutlivesMain(t *testing.T) {
+ t.Parallel() // a minute of waiting for the worker's own exit
+ f := newFixture(t)
+ p := f.start(f.env)
+ p.wait("/ready", http.StatusOK)
+ r := p.report()
+ if r.WorkerPID <= 0 || !alive(r.WorkerPID) {
+ t.Fatalf("no live worker in the report: %d", r.WorkerPID)
+ }
+ if got := workerPIDs(f.db); len(got) != 1 || got[0] != r.WorkerPID {
+ t.Fatalf("worker processes %v, report says %d", got, r.WorkerPID)
+ }
+
+ // The main dies outright. Its stdout and stderr are pipes; a worker
+ // still holding them would keep Wait from returning (the backend's tee
+ // waits up to its drain grace for exactly that).
+ killed := time.Now()
+ p.cmd.Process.Kill()
+ select {
+ case <-p.done:
+ case <-time.After(2 * time.Second):
+ t.Fatalf("the main's Wait did not return within 2s of its death: the worker holds its stdio")
+ }
+
+ n0 := countRows(t, f.db, "worker", "uk")
+ time.Sleep(3 * time.Second)
+ if !alive(r.WorkerPID) {
+ t.Fatalf("the worker died with the main")
+ }
+ if n1 := countRows(t, f.db, "worker", "uk"); n1 < n0+2 {
+ t.Fatalf("the worker stopped writing after the main died: %d then %d rows", n0, n1)
+ }
+
+ // It exits on its own about a minute after its parent has gone.
+ for alive(r.WorkerPID) && time.Since(killed) < 80*time.Second {
+ time.Sleep(250 * time.Millisecond)
+ }
+ gone := time.Since(killed)
+ if alive(r.WorkerPID) {
+ t.Fatalf("the worker is still alive %s after the main died", gone.Round(time.Second))
+ }
+ if gone < 59*time.Second || gone > 70*time.Second {
+ t.Errorf("the worker exited %s after the main died, want about 60s", gone.Round(time.Second))
+ }
+}
+
+// ---- the failure switch ---------------------------------------------------
+
+func TestReadyFailsWhereTold(t *testing.T) {
+ f := newFixture(t)
+
+ // The installed state: a normal boot in uk, stopped, its worker killed
+ // (the stop evidence does that on a real box) and the WAL folded in.
+ seed := f.start(f.env)
+ seed.wait("/ready", http.StatusOK)
+ time.Sleep(time.Second)
+ seed.stop()
+ killWorkers(f.db)
+ checkpoint(t, f.db)
+ installed := snapshot(t, f.dataDir)
+ if !strings.Contains(installed, "h1/instance.key") || !strings.Contains(installed, "h1.db ") {
+ t.Fatalf("the seed boot left no database or key file:\n%s", installed)
+ }
+ timers, _ := f.reg.TimersFor(testHost)
+
+ t.Run("before-write writes nothing", func(t *testing.T) {
+ p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie"}))
+ p.wait("/ready", http.StatusServiceUnavailable)
+ p.wait("/", http.StatusServiceUnavailable)
+ if code, _, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil); code != http.StatusServiceUnavailable {
+ t.Errorf("/arm-once in before-write: %d, want 503", code)
+ }
+ code, _ := p.fire(registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: time.Now().Unix()}, adminToken)
+ if code != http.StatusServiceUnavailable {
+ t.Errorf("/tick in before-write: %d, want 503", code)
+ }
+ code, _ = p.fire(registry.Timer{Host: testHost, Name: "once-9", Kind: schedule.KindOnce, Path: "/once", DueAt: time.Now().Unix()}, adminToken)
+ if code != http.StatusServiceUnavailable {
+ t.Errorf("/once in before-write: %d, want 503", code)
+ }
+ time.Sleep(2 * time.Second)
+ if w := workerPIDs(f.db); len(w) != 0 {
+ t.Errorf("before-write forked a worker: %v", w)
+ }
+ p.stop()
+ if got := snapshot(t, f.dataDir); got != installed {
+ t.Errorf("before-write changed the data directory\ninstalled:\n%s\nafter:\n%s", installed, got)
+ }
+ if got, _ := f.reg.TimersFor(testHost); len(got) != len(timers) {
+ t.Errorf("before-write changed the timers: %v then %v", timers, got)
+ }
+ })
+
+ t.Run("after-write writes one bg row, then fails", func(t *testing.T) {
+ p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie,after-write"}))
+ p.wait("/ready", http.StatusServiceUnavailable)
+ p.wait("/", http.StatusServiceUnavailable)
+ time.Sleep(2 * time.Second)
+ r := p.report()
+ if r.Mode != "after-write" {
+ t.Errorf("mode %q, want after-write", r.Mode)
+ }
+ if n := r.count("bg", "ie"); n != 1 {
+ t.Errorf("after-write wrote %d bg rows in ie, want exactly 1", n)
+ }
+ if n := r.count("worker", "ie"); n < 1 {
+ t.Errorf("after-write's worker wrote no rows in ie")
+ }
+ p.stop()
+ killWorkers(f.db)
+ })
+
+ t.Run("another region's switch does not apply", func(t *testing.T) {
+ p := f.start(f.envWith(map[string]string{"MOVEPROBE_FAIL_READY_IN": "ie"}))
+ p.wait("/ready", http.StatusOK)
+ p.stop()
+ killWorkers(f.db)
+ })
+
+ t.Run("a box with no region is home", func(t *testing.T) {
+ checkpoint(t, f.db)
+ before := snapshot(t, f.dataDir)
+ p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "MOVEPROBE_FAIL_READY_IN": "home"}))
+ p.wait("/ready", http.StatusServiceUnavailable)
+ time.Sleep(time.Second)
+ p.stop()
+ if got := snapshot(t, f.dataDir); got != before {
+ t.Errorf("before-write at home changed the data directory")
+ }
+ })
+}
+
+// ---- the report -----------------------------------------------------------
+
+func TestReportCountsRowsByRegion(t *testing.T) {
+ f := newFixture(t)
+ uk := f.start(f.env)
+ uk.wait("/ready", http.StatusOK)
+
+ code, body, _ := uk.do(http.MethodPost, "/arm-once?in=1", probeSecret, nil)
+ if code != http.StatusOK {
+ t.Fatalf("/arm-once: %d %q", code, body)
+ }
+ once, err := f.reg.GetTimer(testHost, "once-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ // A tick and the once-timer, delivered as the runner delivers them.
+ tick := registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: 1_700_000_040}
+ if code, body := uk.fire(tick, adminToken); code != http.StatusOK {
+ t.Fatalf("/tick: %d %q", code, body)
+ }
+ if code, body := uk.fire(*once, adminToken); code != http.StatusOK {
+ t.Fatalf("/once: %d %q", code, body)
+ }
+ // Neither fires without the platform's token.
+ if code, _ := uk.fire(tick, "wrong"); code != http.StatusUnauthorized {
+ t.Errorf("/tick with the wrong token: %d, want 401", code)
+ }
+ if code, _ := uk.fire(*once, ""); code != http.StatusUnauthorized {
+ t.Errorf("/once with no token: %d, want 401", code)
+ }
+ time.Sleep(1500 * time.Millisecond)
+ ukReport := uk.report()
+ uk.stop()
+ killWorkers(f.db)
+
+ // The same database, now in ie.
+ ie := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie"}))
+ ie.wait("/ready", http.StatusOK)
+ time.Sleep(1500 * time.Millisecond)
+ r := ie.report()
+ ie.stop()
+
+ if r.Region != "ie" {
+ t.Errorf("region %q, want ie", r.Region)
+ }
+ for _, c := range []struct {
+ kind, region string
+ min int64
+ }{{"bg", "uk", 2}, {"bg", "ie", 2}, {"worker", "uk", 1}, {"worker", "ie", 1}, {"tick", "uk", 1}, {"once", "uk", 1}} {
+ if n := r.count(c.kind, c.region); n < c.min {
+ t.Errorf("%s rows in %s: %d, want at least %d (rows %+v)", c.kind, c.region, n, c.min, r.Rows)
+ }
+ }
+ if n := r.count("tick", "ie") + r.count("once", "ie"); n != 0 {
+ t.Errorf("%d timer rows in ie, none fired there", n)
+ }
+ // No background row is missing: the sequence runs on across boots.
+ var bgCount, bgMax int64
+ for _, row := range r.Rows {
+ if row.Kind == "bg" {
+ bgCount += row.Count
+ bgMax = max(bgMax, row.MaxSeq)
+ }
+ }
+ if bgCount != bgMax {
+ t.Errorf("bg rows %d, highest seq %d: a row is missing or repeated", bgCount, bgMax)
+ }
+
+ var gotOnce, gotTick bool
+ for _, fr := range r.Fired {
+ switch {
+ case fr.Kind == "once" && fr.Name == "once-1":
+ gotOnce = fr.Count == 1 && fr.DueAt == once.DueAt
+ case fr.Kind == "tick" && fr.Name == "every-minute":
+ gotTick = fr.Count == 1 && fr.DueAt == 1_700_000_040
+ }
+ }
+ if !gotOnce || !gotTick {
+ t.Errorf("fired = %+v, want once-1 once at %d and every-minute once at 1700000040", r.Fired, once.DueAt)
+ }
+ if r.FileKeySHA256 == "" || r.FileKeySHA256 != ukReport.FileKeySHA256 {
+ t.Errorf("file key fingerprint %q in ie, %q in uk", r.FileKeySHA256, ukReport.FileKeySHA256)
+ }
+ if r.PlatformKeySHA256 != sha(platformKey) {
+ t.Errorf("platform key fingerprint %q, want %q", r.PlatformKeySHA256, sha(platformKey))
+ }
+ if !r.HasSecret || r.Object.Body == "" || !strings.HasPrefix(r.Object.Key, "probe/") {
+ t.Errorf("report = %+v", r)
+ }
+}
+
+// ---- the instance key -----------------------------------------------------
+
+func TestReadyChecksBothKeys(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.env)
+ p.wait("/ready", http.StatusOK)
+ p.stop()
+ killWorkers(f.db)
+
+ keyFile := filepath.Join(strings.TrimSuffix(f.db, ".db"), "instance.key")
+ orig, err := os.ReadFile(keyFile)
+ if err != nil {
+ t.Fatalf("first boot wrote no key file beside the database: %v", err)
+ }
+ if b, err := hex.DecodeString(strings.TrimSpace(string(orig))); err != nil || len(b) != 32 {
+ t.Errorf("instance.key is not 32 bytes of hex: %q", orig)
+ }
+
+ check := func(name string, env map[string]string, want int, says string) {
+ t.Helper()
+ p := f.start(env)
+ body := p.wait("/ready", want)
+ if says != "" && !strings.Contains(body, says) {
+ t.Errorf("%s: /ready body %q, want it to say %q", name, body, says)
+ }
+ p.stop()
+ killWorkers(f.db)
+ }
+
+ if err := os.WriteFile(keyFile, []byte(strings.Repeat("ab", 32)), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ check("changed file", f.env, http.StatusInternalServerError, "key mismatch")
+ if err := os.Remove(keyFile); err != nil {
+ t.Fatal(err)
+ }
+ check("missing file", f.env, http.StatusInternalServerError, "key mismatch")
+ if err := os.WriteFile(keyFile, orig, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ check("restored file", f.env, http.StatusOK, "")
+ check("platform key changed", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": "another-key"}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
+ check("platform key absent", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": ""}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
+ check("both restored", f.env, http.StatusOK, "")
+}