Merge remote-tracking branch 'origin/feat/eip-stopstart' into e2e/fixes-1102

The agent measured each command's deadline from receipt, so a first copy
delayed by a gateway or bus outage was granted a whole fresh budget when
its issuer had nearly given up; a late quiesce could then take a hold
nobody would release. The issuer now stamps every republished copy with
the budget it has left, a duration that needs no clock agreement, and the
agent uses the smaller of that and its own type budget.

.github/scripts/wait-for-memory.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# Usage: wait-for-memory.sh <MiB>
+#
+# Waits until the runner's MemAvailable reaches <MiB>, then returns. Runners on
+# one host share its RAM, and a neighbour's build holds several GiB for a few
+# minutes, so one sample taken during it is not a verdict on the host.
+# Gives up after MEM_WAIT_TIMEOUT_S and names the biggest processes.
+#
+# Env: MEM_WAIT_TIMEOUT_S (900), MEM_WAIT_POLL_S (30), MEMINFO (/proc/meminfo).
+set -uo pipefail
+
+want_mb="${1:?usage: $0 <MiB>}"
+timeout_s="${MEM_WAIT_TIMEOUT_S:-900}"
+poll_s="${MEM_WAIT_POLL_S:-30}"
+meminfo="${MEMINFO:-/proc/meminfo}"
+
+avail_mb() { awk '/^MemAvailable:/ {print int($2 / 1024)}' "$meminfo"; }
+
+start=$SECONDS
+while :; do
+  have=$(avail_mb)
+  if [ -n "$have" ] && [ "$have" -ge "$want_mb" ]; then
+    echo "memory: ${have} MiB available, need ${want_mb} MiB"
+    exit 0
+  fi
+  waited=$(( SECONDS - start ))
+  if [ "$waited" -ge "$timeout_s" ]; then
+    echo "::error::need ${want_mb} MiB available, have ${have:-unknown} MiB after waiting ${waited}s"
+    ps -eo rss,pid,args --sort=-rss | head -11 | cut -c1-200
+    exit 1
+  fi
+  echo "memory: ${have:-unknown} MiB available, need ${want_mb} MiB; waiting (${waited}s of ${timeout_s}s)"
+  sleep "$poll_s"
+done

.github/scripts/wait-for-memory_test.sh

@@ -0,0 +1,47 @@
+#!/bin/sh
+# wait-for-memory_test.sh — drives wait-for-memory.sh against a fake meminfo
+# that a background writer changes mid-wait, so each verdict is checked without
+# a loaded host.
+set -eu
+
+SCRIPT_DIR=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
+WAIT="$SCRIPT_DIR/wait-for-memory.sh"
+TMP=$(mktemp -d)
+trap 'rm -rf "$TMP"' EXIT
+
+fails=0
+pass() { echo "  ok   - $1"; }
+fail() { echo "  FAIL - $1"; fails=$((fails + 1)); }
+
+meminfo() { printf 'MemTotal:       65536000 kB\nMemAvailable:   %s kB\n' "$(( $1 * 1024 ))" > "$TMP/meminfo"; }
+MEMINFO="$TMP/meminfo"
+MEM_WAIT_POLL_S=1
+export MEMINFO MEM_WAIT_POLL_S
+
+meminfo 30000
+if MEM_WAIT_TIMEOUT_S=5 bash "$WAIT" 26624 > "$TMP/out" 2>&1; then
+  pass "enough memory returns at once"
+else
+  fail "enough memory returns at once: $(cat "$TMP/out")"
+fi
+
+meminfo 21577
+( sleep 2; meminfo 30000 ) &
+if MEM_WAIT_TIMEOUT_S=20 bash "$WAIT" 26624 > "$TMP/out" 2>&1 && grep -q waiting "$TMP/out"; then
+  pass "memory freed mid-wait is admitted"
+else
+  fail "memory freed mid-wait is admitted: $(cat "$TMP/out")"
+fi
+wait
+
+meminfo 21577
+if MEM_WAIT_TIMEOUT_S=2 bash "$WAIT" 26624 > "$TMP/out" 2>&1; then
+  fail "memory that never frees fails: exited 0"
+elif grep -q '::error::need 26624 MiB available, have 21577 MiB' "$TMP/out" && grep -q 'RSS' "$TMP/out"; then
+  pass "memory that never frees fails and names the biggest processes"
+else
+  fail "memory that never frees fails and names the biggest processes: $(cat "$TMP/out")"
+fi
+
+[ "$fails" -eq 0 ] || { echo "wait-for-memory: $fails failed"; exit 1; }
+echo "wait-for-memory: all passed"

.github/workflows/e2e-nightly.yml

@@ -1057,9 +1057,6 @@ jobs:
             { echo "::error::br-wan not permitted in /etc/qemu/bridge.conf"; exit 1; }
           command -v sshpass >/dev/null ||
             { echo "::error::sshpass not installed"; exit 1; }
-          free_mb=$(free -m | awk '/^Mem:/ {print $7}')
-          [ "$free_mb" -ge 8192 ] ||
-            { echo "::error::need 8 GiB available, have ${free_mb} MiB"; exit 1; }
           # A 40G image on tmpfs competes with the 8 GiB the VM itself needs.
           fstype=$(df --output=fstype "$TMPDIR" | tail -1 | tr -d ' ')
           [ "$fstype" != tmpfs ] ||
@@ -1090,6 +1087,10 @@ jobs:
               ssh-keygen -t ed25519 -N "" -f "$HOME/.ssh/$k" -C "$k-ci"
           done
 
+      # The node's 8 GiB, claimed when the VM boots rather than at preflight.
+      - name: Wait for memory
+        run: mulga/spinifex/.github/scripts/wait-for-memory.sh 8192
+
       # Boots the ISO, installs unattended, reads the lease off the serial
       # console, runs the in-guest smoke test, and installs the key for the
       # next step.
@@ -1301,15 +1302,10 @@ jobs:
           ref: ${{ steps.siblings.outputs.bluebottle }}
           path: mulga/bluebottle
 
-      # 3 x 8 GiB plus 9 sparse 40G disks, before the instances the cluster
-      # launches on top.
       - name: Preflight
         run: |
           set -euo pipefail
           mkdir -p "$TMPDIR"
-          free_mb=$(free -m | awk '/^Mem:/ {print $7}')
-          [ "$free_mb" -ge 26624 ] ||
-            { echo "::error::need 26 GiB available, have ${free_mb} MiB"; exit 1; }
           fstype=$(df --output=fstype "$TMPDIR" | tail -1 | tr -d ' ')
           [ "$fstype" != tmpfs ] ||
             { echo "::error::$TMPDIR is tmpfs — nine node disks would be in RAM"; exit 1; }
@@ -1332,6 +1328,11 @@ jobs:
       # formation port. Both land with mulga-yxtda and the bridged WAN mode;
       # until then this cell proves the cluster forms and survives, not that a
       # customer can use it.
+      # 3 x 8 GiB plus headroom for the instances the cluster launches on top,
+      # claimed when the nodes boot rather than at preflight.
+      - name: Wait for memory
+        run: mulga/spinifex/.github/scripts/wait-for-memory.sh 26624
+
       - name: Install, form and verify the cluster
         working-directory: mulga/scripts/iso-builder
         run: |

Makefile

@@ -255,7 +255,8 @@ test-build-scripts:
 		echo "-- $$t"; \
 		sh "$$t" || exit 1; \
 	done
-	shellcheck -S warning .github/scripts/hypervisor-disk-preflight.sh .github/scripts/hypervisor-disk-preflight_test.sh
+	shellcheck -S warning .github/scripts/hypervisor-disk-preflight.sh .github/scripts/hypervisor-disk-preflight_test.sh \
+		.github/scripts/wait-for-memory.sh .github/scripts/wait-for-memory_test.sh
 	@echo -e "\n....Running shellcheck over build/scripts/**/*.sh + the node scripts...."
 	shellcheck -S warning $$(find build/scripts -name '*.sh' | sort) $(NODE_SCRIPTS)
 	shellcheck -S warning -x scripts/build-oracle-appliance.sh $$(find scripts/appliance -name '*.sh' | sort)
@@ -326,7 +327,6 @@ test-cover:
 test-race:
 	@echo -e "\n....Running tests with race detector for $(GO_PROJECT_NAME)...."
 	$(_Q)LOG_IGNORE=1 SPINIFEX_SQS_R3=1 go test $(GOTESTFLAGS) -race -timeout 300s ./spinifex/... ./cmd/... ./internal/... ./contracts/... ./architecture/... $(_RACEQ)
-	$(_Q)LOG_IGNORE=1 go test $(GOTESTFLAGS) -race -tags=lambdatestlink -timeout 120s ./cmd/lambda-agent/ ./spinifex/agents/lambda/hostlink/ $(_RACEQ)
 
 # Unit tests for in-repo GitHub Actions (e.g. .github/actions/e2e-analyze).
 # Kept out of `test-cover` so coverage % isn't diluted by CI-only tooling.

cmd/lambda-agent/connect.go

@@ -1,48 +0,0 @@
-package main
-
-import (
-	"context"
-	"crypto/tls"
-	"errors"
-	"fmt"
-
-	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
-)
-
-// Exit statuses: the guest refused to start, the control connection failed,
-// or the host closed it.
-const (
-	exitRefused = 2
-	exitLink    = 1
-	exitClosed  = 0
-)
-
-type startup struct {
-	fwCfgDir  string
-	fwCfgRoot string
-	unload    func() error
-}
-
-func defaultStartup() startup {
-	return startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
-}
-
-// connect loads the identity, seals fw_cfg whether or not the load succeeded,
-// then dials the host. Under TLS 1.3 a host refusal is seen only on first read.
-func (s startup) connect(ctx context.Context) (*tls.Conn, int, error) {
-	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
-	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
-		return nil, exitRefused, errors.Join(loadErr, err)
-	}
-	if loadErr != nil {
-		return nil, exitRefused, loadErr
-	}
-	conn, err := hostlink.Dial(ctx, id)
-	if err != nil {
-		if ctx.Err() != nil {
-			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
-		}
-		return nil, exitLink, err
-	}
-	return conn, 0, nil
-}

cmd/lambda-agent/helpers_test.go

@@ -1,98 +0,0 @@
-package main
-
-import (
-	"context"
-	"crypto/tls"
-	"net"
-	"os"
-	"path/filepath"
-	"testing"
-	"time"
-
-	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
-	"github.com/stretchr/testify/require"
-)
-
-const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
-
-// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
-type guest struct {
-	root     string
-	unloads  int
-	sealedAt chan struct{}
-}
-
-func newGuest(t *testing.T, entries map[string][]byte) *guest {
-	t.Helper()
-	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
-	for name, b := range entries {
-		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
-		require.NoError(t, os.MkdirAll(dir, 0o700))
-		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
-	}
-	return g
-}
-
-func (g *guest) startup() startup {
-	return startup{
-		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
-		fwCfgRoot: g.root,
-		unload: func() error {
-			g.unloads++
-			g.sealedAt <- struct{}{}
-			return os.RemoveAll(g.root)
-		},
-	}
-}
-
-type host struct {
-	ln   *net.TCPListener
-	gen  *guestconn.Generation
-	conn chan accepted
-}
-
-func newHost(t *testing.T) *host {
-	t.Helper()
-	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
-	require.NoError(t, err)
-	t.Cleanup(func() { ln.Close() })
-	gen, err := guestconn.NewGeneration(scope, time.Hour)
-	require.NoError(t, err)
-	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
-}
-
-func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
-
-type accepted struct {
-	conn *tls.Conn
-	err  error
-}
-
-// accept runs the host end until ctx ends and reports what it accepted.
-func (h *host) accept(ctx context.Context) {
-	go func() {
-		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
-		h.conn <- accepted{c, err}
-	}()
-}
-
-// guest waits for the host to authenticate the guest.
-func (h *host) guest(t *testing.T) *tls.Conn {
-	t.Helper()
-	a := wait(t, h.conn, "host to authenticate the guest")
-	require.NoError(t, a.err)
-	return a.conn
-}
-
-// wait bounds every test wait so a stalled host or startup fails promptly.
-func wait[T any](t *testing.T, ch <-chan T, what string) T {
-	t.Helper()
-	select {
-	case v := <-ch:
-		return v
-	case <-time.After(10 * time.Second):
-		t.Fatalf("timed out waiting for %s", what)
-		panic("unreachable")
-	}
-}

cmd/lambda-agent/main.go

@@ -6,8 +6,8 @@
 // never starts the function's bootstrap: the message protocol and grant
 // fencing do not exist yet. The agent never listens for the host.
 //
-// A lambdatestlink build opens the same connection, then relays one invocation
-// over a test carrier for local proofs; the carrier is not the guest protocol.
+// A lambdatestlink build instead relays invocations over an unauthenticated
+// test carrier for local proofs; it is not the guest protocol.
 package main
 
 import (

cmd/lambda-agent/start_tcp.go

@@ -9,28 +9,56 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+)
+
+// Exit statuses: the guest refused to start, the control connection failed,
+// or the host closed it.
+const (
+	exitRefused = 2
+	exitLink    = 1
+	exitClosed  = 0
 )
 
 // errNoProtocol is returned when the host sends anything: this build defines no
 // message protocol, so it accepts no work.
 var errNoProtocol = errors.New("host sent data, but this build accepts no invocations")
 
+type startup struct {
+	fwCfgDir  string
+	fwCfgRoot string
+	unload    func() error
+}
+
 func start(ctx context.Context) int {
-	code, err := defaultStartup().run(ctx)
+	s := startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
+	code, err := s.run(ctx)
 	if err != nil {
 		slog.Error("lambda-agent stopping", "exit", code, "err", err)
 	}
 	return code
 }
 
-// run authenticates to the host and holds the connection until it closes.
+// run loads the identity, seals fw_cfg whether or not the load succeeded,
+// then authenticates to the host and holds the connection until it closes.
 func (s startup) run(ctx context.Context) (int, error) {
-	conn, code, err := s.connect(ctx)
+	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
+	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
+		return exitRefused, errors.Join(loadErr, err)
+	}
+	if loadErr != nil {
+		return exitRefused, loadErr
+	}
+	conn, err := hostlink.Dial(ctx, id)
 	if err != nil {
-		return code, err
+		if ctx.Err() != nil {
+			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
+		}
+		return exitLink, err
 	}
 	defer conn.Close()
-	slog.Info("control connection open; invocations are not enabled in this build", "host", conn.RemoteAddr().String())
+	slog.Info("control connection open; invocations are not enabled in this build", "host", id.HostAddr)
 	if err := awaitHostClose(ctx, conn); err != nil {
 		return exitLink, err
 	}

cmd/lambda-agent/start_tcp_test.go

@@ -4,9 +4,11 @@ package main
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
 	"net"
 	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -16,6 +18,89 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
+
+// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
+type guest struct {
+	root     string
+	unloads  int
+	sealedAt chan struct{}
+}
+
+func newGuest(t *testing.T, entries map[string][]byte) *guest {
+	t.Helper()
+	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
+	for name, b := range entries {
+		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
+		require.NoError(t, os.MkdirAll(dir, 0o700))
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
+	}
+	return g
+}
+
+func (g *guest) startup() startup {
+	return startup{
+		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
+		fwCfgRoot: g.root,
+		unload: func() error {
+			g.unloads++
+			g.sealedAt <- struct{}{}
+			return os.RemoveAll(g.root)
+		},
+	}
+}
+
+type host struct {
+	ln   *net.TCPListener
+	gen  *guestconn.Generation
+	conn chan accepted
+}
+
+func newHost(t *testing.T) *host {
+	t.Helper()
+	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	gen, err := guestconn.NewGeneration(scope, time.Hour)
+	require.NoError(t, err)
+	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
+}
+
+func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
+
+type accepted struct {
+	conn *tls.Conn
+	err  error
+}
+
+// accept runs the host end until ctx ends and reports what it accepted.
+func (h *host) accept(ctx context.Context) {
+	go func() {
+		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
+		h.conn <- accepted{c, err}
+	}()
+}
+
+// guest waits for the host to authenticate the guest.
+func (h *host) guest(t *testing.T) *tls.Conn {
+	t.Helper()
+	a := wait(t, h.conn, "host to authenticate the guest")
+	require.NoError(t, a.err)
+	return a.conn
+}
+
+// wait bounds every test wait so a stalled host or startup fails promptly.
+func wait[T any](t *testing.T, ch <-chan T, what string) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(10 * time.Second):
+		t.Fatalf("timed out waiting for %s", what)
+		panic("unreachable")
+	}
+}
+
 type result struct {
 	code int
 	err  error

cmd/lambda-agent/start_testlink.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"os/exec"
+	"syscall"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
@@ -24,56 +25,34 @@ const (
 	taskRoot    = "/var/task"
 )
 
-var handler = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
+var (
+	linkPath = flag.String("link", os.Getenv("LAMBDA_AGENT_LINK"), "test carrier device path")
+	handler  = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
+)
 
-// testlink relays one invocation over the test carrier on the authenticated
-// connection; the carrier is not the guest protocol.
-type testlink struct {
-	startup
-	runtimeAddr    string
-	startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
-}
-
-func start(ctx context.Context) int {
-	t := testlink{
-		startup:     defaultStartup(),
-		runtimeAddr: runtimeAddr,
-		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
-			return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
-		},
+func start(parent context.Context) int {
+	if *linkPath == "" {
+		slog.Error("no test carrier configured; refusing to start")
+		return 2
 	}
-	code, err := t.run(ctx)
+	// The bootstrap is tenant code; fw_cfg must be unreadable before it runs.
+	if err := hostlink.SealFwCfg(hostlink.FwCfgRoot, hostlink.UnloadFwCfg); err != nil {
+		slog.Error("seal fw_cfg; refusing to start", "err", err)
+		return 2
+	}
+	dev, err := os.OpenFile(*linkPath, os.O_RDWR, 0)
 	if err != nil {
-		slog.Error("lambda-agent stopping", "exit", code, "err", err)
-	}
-	return code
-}
-
-func (t testlink) run(parent context.Context) (int, error) {
-	conn, code, err := t.connect(parent)
-	if err != nil {
-		return code, err
-	}
-	defer conn.Close()
-	link := hostlink.NewTestStream(conn)
-
-	// The handshake can return before the host refuses the guest, so nothing
-	// tenant-facing starts until the host's first valid frame arrives.
-	first, err := link.Next(parent)
-	if errors.Is(err, hostlink.ErrShutdown) {
-		return exitClosed, nil
-	}
-	if err != nil {
-		return exitLink, fmt.Errorf("first frame from host: %w", err)
-	}
-	if parent.Err() != nil {
-		return exitLink, context.Cause(parent)
+		slog.Error("open test carrier; refusing to start", "err", err)
+		return 2
 	}
+	defer dev.Close()
+	link := hostlink.NewTestStream(dev)
 
 	srv := runtimeapi.New()
-	ln, err := net.Listen("tcp", t.runtimeAddr)
+	ln, err := net.Listen("tcp", runtimeAddr)
 	if err != nil {
-		return exitLink, fmt.Errorf("listen runtime api: %w", err)
+		slog.Error("listen runtime api", "err", err)
+		return 1
 	}
 	hs := &http.Server{Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}
 	go func() { _ = hs.Serve(ln) }()
@@ -81,14 +60,20 @@ func (t testlink) run(parent context.Context) (int, error) {
 
 	ctx, cancel := context.WithCancelCause(parent)
 	defer cancel(nil)
-	cmd, err := t.startBootstrap(ctx, []string{
-		"AWS_LAMBDA_RUNTIME_API=" + ln.Addr().String(),
+
+	cmd := exec.CommandContext(ctx, taskRoot+"/bootstrap")
+	cmd.Env = []string{
+		"AWS_LAMBDA_RUNTIME_API=" + runtimeAddr,
 		"_HANDLER=" + *handler,
 		"LAMBDA_TASK_ROOT=" + taskRoot,
 		"PATH=/bin:/sbin:/usr/bin",
-	})
-	if err != nil {
-		return exitLink, fmt.Errorf("start bootstrap: %w", err)
+	}
+	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
+	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
+	cmd.WaitDelay = 5 * time.Second
+	if err := cmd.Start(); err != nil {
+		slog.Error("start bootstrap", "err", err)
+		return 1
 	}
 	exited := make(chan struct{})
 	go func() {
@@ -97,32 +82,12 @@ func (t testlink) run(parent context.Context) (int, error) {
 		cancel(fmt.Errorf("bootstrap exited: %w", err))
 	}()
 
-	err = agent.Run(ctx, &oneInvocation{Link: link, first: &first}, srv)
+	err = agent.Run(ctx, link, srv)
 	cancel(errors.New("agent stopping"))
 	<-exited
 	if err != nil {
-		return exitLink, err
+		slog.Error("agent stopped", "err", err)
+		return 1
 	}
-	return exitClosed, nil
-}
-
-// oneInvocation serves the host's first request and then stops the agent:
-// each test VM runs a single invocation.
-type oneInvocation struct {
-	hostlink.Link
-	first *hostlink.Request
-}
-
-var _ hostlink.Link = (*oneInvocation)(nil)
-
-func (o *oneInvocation) Next(ctx context.Context) (hostlink.Request, error) {
-	if err := ctx.Err(); err != nil {
-		return hostlink.Request{}, err
-	}
-	if o.first == nil {
-		return hostlink.Request{}, hostlink.ErrShutdown
-	}
-	req := *o.first
-	o.first = nil
-	return req, nil
+	return 0
 }

cmd/lambda-agent/start_testlink_test.go

@@ -1,179 +0,0 @@
-//go:build lambdatestlink
-
-package main
-
-import (
-	"bytes"
-	"context"
-	"io"
-	"net/http"
-	"os"
-	"os/exec"
-	"testing"
-	"time"
-
-	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
-	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-const helperEnv = "LAMBDA_AGENT_TEST_BOOTSTRAP"
-
-// TestHelperBootstrap is the fake function bootstrap, run as a subprocess: it
-// echoes one invocation through the Runtime API, then waits for the next.
-func TestHelperBootstrap(t *testing.T) {
-	if os.Getenv(helperEnv) != "1" {
-		return
-	}
-	base := "http://" + os.Getenv("AWS_LAMBDA_RUNTIME_API") + "/2018-06-01/runtime/invocation/"
-	resp, err := http.Get(base + "next")
-	require.NoError(t, err)
-	body, err := io.ReadAll(resp.Body)
-	require.NoError(t, err)
-	resp.Body.Close()
-	id := resp.Header.Get("Lambda-Runtime-Aws-Request-Id")
-	post, err := http.Post(base+id+"/response", "application/json", bytes.NewReader(body))
-	require.NoError(t, err)
-	post.Body.Close()
-	_, _ = http.Get(base + "next")
-}
-
-type bootstrapStarts struct {
-	n       int
-	started chan struct{}
-}
-
-func (b *bootstrapStarts) testlink(g *guest) testlink {
-	b.started = make(chan struct{}, 1)
-	return testlink{
-		startup:     g.startup(),
-		runtimeAddr: "127.0.0.1:0",
-		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
-			b.n++
-			b.started <- struct{}{}
-			cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestHelperBootstrap$")
-			cmd.Env = append(env, helperEnv+"=1")
-			return cmd, cmd.Start()
-		},
-	}
-}
-
-func runTestlinkAsync(ctx context.Context, tl testlink) <-chan result {
-	out := make(chan result, 1)
-	go func() {
-		code, err := tl.run(ctx)
-		out <- result{code, err}
-	}()
-	return out
-}
-
-type result struct {
-	code int
-	err  error
-}
-
-func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	h := newHost(t)
-	h.accept(ctx)
-	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
-
-	conn := h.guest(t)
-	defer conn.Close()
-	select {
-	case <-b.started:
-		t.Fatal("bootstrap started before the host sent a frame")
-	case <-time.After(100 * time.Millisecond):
-	}
-	hs := hostlink.NewTestStream(conn)
-	req := hostlink.Request{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{"a":40,"b":2}`)}
-	require.NoError(t, hs.HostSend(req))
-	got, err := hs.HostResult(ctx)
-	require.NoError(t, err)
-	assert.Equal(t, hostlink.Response{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", Payload: req.Payload}, got)
-
-	// A second invocation on the same VM is never served. The agent may
-	// already have closed the connection, so a failed send is a refusal too.
-	second := req
-	second.RequestID = "r-2"
-	sendErr := hs.HostSend(second)
-	res := wait(t, done, "agent to stop after one invocation")
-	assert.Equal(t, exitClosed, res.code)
-	assert.NoError(t, res.err)
-	if sendErr == nil {
-		_, err = hs.HostResult(ctx)
-		assert.Error(t, err, "agent answered a second invocation")
-	}
-	assert.Equal(t, 1, b.n)
-}
-
-func TestTestlinkShutdownBeforeInvocationNeverStartsBootstrap(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	h := newHost(t)
-	h.accept(ctx)
-	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
-
-	conn := h.guest(t)
-	defer conn.Close()
-	require.NoError(t, hostlink.NewTestStream(conn).HostShutdown())
-	res := wait(t, done, "agent to stop")
-	assert.Equal(t, exitClosed, res.code)
-	assert.NoError(t, res.err)
-	assert.Zero(t, b.n)
-}
-
-// The guest's handshake completes, but the host pins another generation's
-// guest; the refusal arrives on first read and the bootstrap never starts.
-func TestTestlinkHostRefusalNeverStartsBootstrap(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	h := newHost(t)
-	h.accept(ctx)
-	other, err := guestconn.NewGeneration(scope, time.Hour)
-	require.NoError(t, err)
-	entries := h.entries()
-	wrong := other.GuestEntries(h.ln.Addr().String())
-	entries[lambdav1.EntryGuestKey], entries[lambdav1.EntryGuestCert] = wrong[lambdav1.EntryGuestKey], wrong[lambdav1.EntryGuestCert]
-	var b bootstrapStarts
-
-	res := wait(t, runTestlinkAsync(ctx, b.testlink(newGuest(t, entries))), "agent to stop")
-	assert.Equal(t, exitLink, res.code)
-	require.Error(t, res.err)
-	assert.Contains(t, res.err.Error(), "bad certificate")
-	assert.Zero(t, b.n)
-}
-
-func TestTestlinkHostCloseBeforeFrameNeverStartsBootstrap(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	h := newHost(t)
-	h.accept(ctx)
-	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
-
-	require.NoError(t, h.guest(t).Close())
-	res := wait(t, done, "agent to stop")
-	assert.Equal(t, exitLink, res.code)
-	assert.ErrorContains(t, res.err, "first frame from host")
-	assert.Zero(t, b.n)
-}
-
-// A cancelled context is reported before the buffered request is delivered,
-// and the request stays undelivered.
-func TestOneInvocationHonoursCancelledContext(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	o := &oneInvocation{first: &hostlink.Request{RequestID: "r-1"}}
-	_, err := o.Next(ctx)
-	require.ErrorIs(t, err, context.Canceled)
-
-	req, err := o.Next(context.Background())
-	require.NoError(t, err)
-	assert.Equal(t, "r-1", req.RequestID)
-}

spinifex/agents/lambda/agent/bootstrap.go

@@ -1,54 +0,0 @@
-//go:build linux
-
-package agent
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"os/exec"
-	goruntime "runtime"
-	"syscall"
-	"time"
-
-	"golang.org/x/sys/unix"
-)
-
-// sandboxID is the uid and gid the function's bootstrap runs as.
-const sandboxID = 993
-
-// StartBootstrap starts the function's bootstrap at path as the sandbox uid and
-// gid with no supplementary groups, with no_new_privs, and with exactly env.
-// The caller waits on the returned command.
-func StartBootstrap(ctx context.Context, path string, env []string, stdout, stderr io.Writer) (*exec.Cmd, error) {
-	cred := &syscall.Credential{Uid: sandboxID, Gid: sandboxID, Groups: []uint32{}}
-	return startBootstrap(ctx, path, env, stdout, stderr, cred)
-}
-
-func startBootstrap(ctx context.Context, path string, env []string, stdout, stderr io.Writer, cred *syscall.Credential) (*exec.Cmd, error) {
-	cmd := exec.CommandContext(ctx, path)
-	// A nil Env inherits the agent's environment; the bootstrap gets only env.
-	cmd.Env = append([]string{}, env...)
-	cmd.Stdout, cmd.Stderr = stdout, stderr
-	if cred != nil {
-		cmd.SysProcAttr = &syscall.SysProcAttr{Credential: cred}
-	}
-	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
-	cmd.WaitDelay = 5 * time.Second
-
-	started := make(chan error, 1)
-	go func() {
-		// no_new_privs is per thread and the child inherits it from this one. The
-		// thread is never unlocked, so Go ends it when this goroutine returns.
-		goruntime.LockOSThread()
-		if err := unix.Prctl(unix.PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0); err != nil {
-			started <- fmt.Errorf("set no_new_privs: %w", err)
-			return
-		}
-		started <- cmd.Start()
-	}()
-	if err := <-started; err != nil {
-		return nil, err
-	}
-	return cmd, nil
-}

spinifex/agents/lambda/agent/bootstrap_test.go

@@ -1,171 +0,0 @@
-//go:build linux
-
-package agent
-
-import (
-	"bytes"
-	"context"
-	"io"
-	"os"
-	"os/exec"
-	"path/filepath"
-	goruntime "runtime"
-	"strings"
-	"syscall"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// statusProbe writes an executable that prints its own credential lines.
-func statusProbe(t *testing.T) string {
-	t.Helper()
-	// The sandbox uid must be able to reach the probe through the test's dirs.
-	dir := t.TempDir()
-	require.NoError(t, os.Chmod(filepath.Dir(dir), 0o755))
-	require.NoError(t, os.Chmod(dir, 0o755))
-	path := filepath.Join(dir, "bootstrap")
-	script := "#!/bin/sh\ngrep -E '^(Uid|Gid|Groups|NoNewPrivs):' /proc/self/status\n"
-	require.NoError(t, os.WriteFile(path, []byte(script), 0o755))
-	return path
-}
-
-func statusLines(t *testing.T, out []byte) map[string]string {
-	t.Helper()
-	lines := map[string]string{}
-	for l := range strings.SplitSeq(strings.TrimSpace(string(out)), "\n") {
-		k, v, ok := strings.Cut(l, ":")
-		require.True(t, ok, "status line %q", l)
-		lines[k] = strings.Join(strings.Fields(v), " ")
-	}
-	return lines
-}
-
-// run starts a bootstrap with the given credential and returns its stdout.
-func run(t *testing.T, path string, env []string, sandbox bool) []byte {
-	t.Helper()
-	var out bytes.Buffer
-	start := startBootstrap
-	if sandbox {
-		start = func(ctx context.Context, p string, e []string, o, s io.Writer, _ *syscall.Credential) (*exec.Cmd, error) {
-			return StartBootstrap(ctx, p, e, o, s)
-		}
-	}
-	cmd, err := start(context.Background(), path, env, &out, os.Stderr, nil)
-	require.NoError(t, err)
-	require.NoError(t, cmd.Wait())
-	return out.Bytes()
-}
-
-// threadNoNewPrivs reads no_new_privs for the calling OS thread.
-func threadNoNewPrivs(t *testing.T) string {
-	t.Helper()
-	b, err := os.ReadFile("/proc/thread-self/status")
-	require.NoError(t, err)
-	return statusLines(t, grepLines(b, "NoNewPrivs:"))["NoNewPrivs"]
-}
-
-func grepLines(b []byte, prefix string) []byte {
-	var keep []string
-	for l := range strings.SplitSeq(string(b), "\n") {
-		if strings.HasPrefix(l, prefix) {
-			keep = append(keep, l)
-		}
-	}
-	return []byte(strings.Join(keep, "\n"))
-}
-
-func TestStartBootstrapSetsNoNewPrivsWithoutChangingTheCaller(t *testing.T) {
-	goruntime.LockOSThread()
-	defer goruntime.UnlockOSThread()
-	if threadNoNewPrivs(t) != "0" {
-		t.Skip("test thread already has no_new_privs; the child would inherit it")
-	}
-
-	out := run(t, statusProbe(t), nil, false)
-	assert.Equal(t, "1", statusLines(t, out)["NoNewPrivs"])
-	assert.Equal(t, "0", threadNoNewPrivs(t), "the calling thread must keep no_new_privs unset")
-}
-
-func TestStartBootstrapDropsToSandboxIdentity(t *testing.T) {
-	if os.Geteuid() != 0 {
-		t.Skip("dropping to the sandbox uid needs root")
-	}
-	lines := statusLines(t, run(t, statusProbe(t), nil, true))
-	assert.Equal(t, "993 993 993 993", lines["Uid"])
-	assert.Equal(t, "993 993 993 993", lines["Gid"])
-	assert.Empty(t, lines["Groups"])
-	assert.Equal(t, "1", lines["NoNewPrivs"])
-}
-
-// A tenant LD_PRELOAD constructor runs before the bootstrap's main; it must
-// already see no_new_privs, so no loader runs before the flag is set.
-func TestStartBootstrapPreloadConstructorSeesNoNewPrivs(t *testing.T) {
-	goruntime.LockOSThread()
-	defer goruntime.UnlockOSThread()
-	if threadNoNewPrivs(t) != "0" {
-		t.Skip("test thread already has no_new_privs; the check would pass vacuously")
-	}
-	cc, err := exec.LookPath("cc")
-	if err != nil {
-		t.Skip("no C compiler for the preload library")
-	}
-	bootstrap, err := exec.LookPath("true")
-	require.NoError(t, err)
-	dir := t.TempDir()
-	src := filepath.Join(dir, "preload.c")
-	lib := filepath.Join(dir, "preload.so")
-	require.NoError(t, os.WriteFile(src, []byte(preloadSource), 0o644))
-	build := exec.Command(cc, "-shared", "-fPIC", "-o", lib, src)
-	build.Stderr = os.Stderr
-	require.NoError(t, build.Run())
-
-	out := run(t, bootstrap, []string{"LD_PRELOAD=" + lib}, false)
-	assert.Equal(t, "preload NoNewPrivs: 1", strings.Join(strings.Fields(string(out)), " "))
-}
-
-const preloadSource = `#include <stdio.h>
-#include <string.h>
-
-__attribute__((constructor)) static void report(void) {
-	char line[256];
-	FILE *f = fopen("/proc/self/status", "r");
-	if (!f)
-		return;
-	while (fgets(line, sizeof line, f))
-		if (strncmp(line, "NoNewPrivs:", 11) == 0)
-			printf("preload %s", line);
-	fclose(f);
-	fflush(stdout);
-}
-`
-
-func TestStartBootstrapDoesNotInheritAgentEnvironment(t *testing.T) {
-	t.Setenv("ZZ_AGENT_SECRET", "must-not-leak")
-	probe, err := exec.LookPath("env")
-	require.NoError(t, err)
-
-	for name, env := range map[string][]string{"nil": nil, "empty": {}, "given": {"ONLY=this"}} {
-		out := run(t, probe, env, false)
-		// Never print the captured environment: an inherited one may hold secrets.
-		if strings.TrimSuffix(string(out), "\n") != strings.Join(env, "\n") {
-			t.Errorf("%s: bootstrap environment is not exactly the one given", name)
-		}
-	}
-}
-
-// Without root the sandbox credential cannot be applied, so the exported entry
-// point must refuse rather than run the bootstrap as the caller.
-func TestStartBootstrapAsNonRootRefusesInsteadOfKeepingCallerIdentity(t *testing.T) {
-	if os.Geteuid() == 0 {
-		t.Skip("root can apply the sandbox credential")
-	}
-	_, err := StartBootstrap(context.Background(), statusProbe(t), nil, nil, nil)
-	assert.ErrorIs(t, err, syscall.EPERM)
-}
-
-func TestStartBootstrapMissingExecutableFails(t *testing.T) {
-	_, err := startBootstrap(context.Background(), filepath.Join(t.TempDir(), "bootstrap"), nil, nil, nil, nil)
-	assert.Error(t, err)
-}

spinifex/agents/rds/agent/agent_test.go

@@ -763,10 +763,11 @@ func TestPollDBCommandsOutput_MatchesGatewayShape(t *testing.T) {
 	// either side's tags fails here.
 	payload := awsxml.QueryResponsePayload("PollDBCommands", &rdsawsapi.PollDBCommandsOutput{
 		Commands: []rdsinstance.Command{{
-			CommandID:  "cmd-1",
-			Type:       "reload-parameters",
-			Parameters: []rdsinstance.Parameter{{Name: "work_mem", Value: "8MB"}},
-			IssuedAt:   &issued,
+			CommandID:         "cmd-1",
+			Type:              "reload-parameters",
+			Parameters:        []rdsinstance.Parameter{{Name: "work_mem", Value: "8MB"}},
+			IssuedAt:          &issued,
+			RemainingBudgetMs: 29500,
 		}},
 	})
 	body, err := awsxml.Marshal(payload)
@@ -802,6 +803,9 @@ func TestPollDBCommandsOutput_MatchesGatewayShape(t *testing.T) {
 	if cmd.IssuedAt == nil || !cmd.IssuedAt.Equal(issued) {
 		t.Errorf("IssuedAt = %v, want %v", cmd.IssuedAt, issued)
 	}
+	if cmd.RemainingBudgetMs != 29500 {
+		t.Errorf("RemainingBudgetMs = %d, want 29500: a copy that loses it is granted a fresh budget", cmd.RemainingBudgetMs)
+	}
 }
 
 func readFile(t *testing.T, path string) string {

spinifex/agents/rds/agent/command.go

@@ -2,10 +2,13 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	rdsinstance "github.com/mulgadc/spinifex/spinifex/domains/rds/instance"
@@ -90,35 +93,96 @@ func commandParams(cmd rdsinstance.Command) map[string]string {
 // line rate for as long as it stays broken.
 const pollErrorBackoff = 5 * time.Second
 
+// How many finished command IDs are remembered, so a copy the issuer re-published
+// while the command ran is not executed a second time.
+const completedCommandMemory = 64
+
+// How many engine commands may wait behind the one running. Past this a command
+// is refused at once rather than queued behind work its issuer will not outlive.
+const serialQueueDepth = 16
+
+// The budget for a type the control plane does not issue. Such a command is
+// answered as unsupported before it could use any of it.
+const fallbackCommandBudget = 30 * time.Second
+
+// The only command that runs beside the serial lane. It releases the backup hold
+// an engine command can be blocked on, so it must never queue behind one.
+func outOfBand(commandType string) bool {
+	return commandType == rdsinstance.CommandUnquiesce
+}
+
+// A command waiting its turn on the serial lane, with the moment its issuer
+// stops waiting for it.
+type queuedCommand struct {
+	cmd      rdsinstance.Command
+	deadline time.Time
+}
+
 // The agent's command channel: a long poll that carries back the replies for
-// whatever the previous poll delivered.
+// whatever earlier polls delivered. Engine commands run one at a time on a single
+// lane; the release of a backup hold runs beside it, so it cannot be kept out.
 type commander struct {
 	cp       controlPlane
 	id       identity
 	registry commandRegistry
 	wait     time.Duration
+	// The control plane's budget per type, capped by the remaining budget the
+	// issuer stamps on each copy and measured from receipt. IssuedAt is not
+	// compared: the guest clock is not trusted against the host's.
+	budget func(commandType string) time.Duration
+	serial chan queuedCommand
+
+	mu sync.Mutex
 	// Cleared only once a poll has carried them, so a failed poll re-delivers
 	// rather than dropping the result of work the guest actually did.
 	pending []rdsinstance.CommandReply
+	// Queued or running, so a re-published copy is not queued a second time.
+	inFlight map[string]struct{}
+	// Oldest first, bounded by completedCommandMemory.
+	completed []string
+	// Ends the poll in progress, so a reply that finished during it rides the
+	// next one rather than waiting out the window. Nil between polls.
+	interruptPoll func()
+	interrupted   bool
+	running       sync.WaitGroup
 }
 
 func newCommander(cp controlPlane, registry commandRegistry, wait time.Duration) *commander {
 	if wait <= 0 {
 		wait = defaultPollWait
 	}
-	return &commander{cp: cp, registry: registry, wait: wait}
+	return &commander{
+		cp:       cp,
+		registry: registry,
+		wait:     wait,
+		budget:   controlPlaneBudget,
+		serial:   make(chan queuedCommand, serialQueueDepth),
+		inFlight: make(map[string]struct{}),
+	}
+}
+
+func controlPlaneBudget(commandType string) time.Duration {
+	if budget, ok := rdsinstance.CommandBudget(commandType); ok {
+		return budget
+	}
+	return fallbackCommandBudget
 }
 
 func (c *commander) Run(ctx context.Context) {
+	defer c.running.Wait()
+	c.running.Go(func() { c.serveSerial(ctx) })
 	for {
 		if ctx.Err() != nil {
 			return
 		}
-		commands, err := c.cp.PollCommands(ctx, c.id, c.pending, c.wait)
+		commands, sent, err := c.poll(ctx)
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
+			if errors.Is(err, errReplyReady) {
+				continue
+			}
 			slog.Warn("rds-agent: command poll failed", "err", err)
 			select {
 			case <-ctx.Done():
@@ -127,14 +191,122 @@ func (c *commander) Run(ctx context.Context) {
 			}
 			continue
 		}
-		c.pending = nil
+		c.mu.Lock()
+		c.pending = c.pending[sent:]
+		c.mu.Unlock()
 
 		for _, cmd := range commands {
-			c.pending = append(c.pending, c.execute(ctx, cmd))
+			c.dispatch(ctx, cmd)
 		}
 	}
 }
 
+// Reports that a poll was ended early so a finished reply can ride the next one.
+var errReplyReady = errors.New("a command reply is ready")
+
+// One long poll carrying every reply ready when it starts. It is cut short when
+// another reply becomes ready, so a finished command is never answered a whole
+// poll window late. Returns how many pending replies the poll carried.
+func (c *commander) poll(ctx context.Context) ([]rdsinstance.Command, int, error) {
+	pollCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	c.mu.Lock()
+	replies := slices.Clone(c.pending)
+	c.interruptPoll, c.interrupted = cancel, false
+	c.mu.Unlock()
+
+	commands, err := c.cp.PollCommands(pollCtx, c.id, replies, c.wait)
+
+	c.mu.Lock()
+	interrupted := c.interrupted
+	c.interruptPoll = nil
+	c.mu.Unlock()
+	if err != nil {
+		if interrupted && ctx.Err() == nil {
+			return nil, 0, errReplyReady
+		}
+		return nil, 0, err
+	}
+	return commands, len(replies), nil
+}
+
+// Routes cmd to its lane unless it is already queued, running or answered: the
+// issuer re-publishes until it has a reply, so the same ID arrives again.
+func (c *commander) dispatch(ctx context.Context, cmd rdsinstance.Command) {
+	c.mu.Lock()
+	_, queued := c.inFlight[cmd.CommandID]
+	if queued || slices.Contains(c.completed, cmd.CommandID) {
+		c.mu.Unlock()
+		slog.Debug("rds-agent: duplicate command ignored", "commandID", cmd.CommandID, "type", cmd.Type)
+		return
+	}
+	c.inFlight[cmd.CommandID] = struct{}{}
+	c.mu.Unlock()
+
+	q := queuedCommand{cmd: cmd, deadline: time.Now().Add(c.budgetFor(cmd))}
+	if outOfBand(cmd.Type) {
+		c.running.Go(func() { c.runQueued(ctx, q) })
+		return
+	}
+	select {
+	case c.serial <- q:
+	default:
+		c.finish(cmd, failedReply(cmd, "the agent is busy; too many commands are waiting"))
+	}
+}
+
+// The type's budget, or what the issuer had left when it published this copy if
+// that is less: a copy delayed in transit must not be granted a fresh budget its
+// issuer will not wait out.
+func (c *commander) budgetFor(cmd rdsinstance.Command) time.Duration {
+	budget := c.budget(cmd.Type)
+	if cmd.RemainingBudgetMs > 0 {
+		budget = min(budget, time.Duration(cmd.RemainingBudgetMs)*time.Millisecond)
+	}
+	return budget
+}
+
+func (c *commander) serveSerial(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case q := <-c.serial:
+			c.runQueued(ctx, q)
+		}
+	}
+}
+
+// Runs a command inside what is left of its issuer's budget. One whose budget
+// ran out while it waited is answered as expired and never touches the engine,
+// since its issuer has already reported it failed.
+func (c *commander) runQueued(ctx context.Context, q queuedCommand) {
+	if !time.Now().Before(q.deadline) {
+		slog.Warn("rds-agent: command expired before it ran", "commandID", q.cmd.CommandID, "type", q.cmd.Type)
+		c.finish(q.cmd, failedReply(q.cmd, "the command expired before the agent could run it"))
+		return
+	}
+	cmdCtx, cancel := context.WithDeadline(ctx, q.deadline)
+	defer cancel()
+	c.finish(q.cmd, c.execute(cmdCtx, q.cmd))
+}
+
+func (c *commander) finish(cmd rdsinstance.Command, reply rdsinstance.CommandReply) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.inFlight, cmd.CommandID)
+	c.completed = append(c.completed, cmd.CommandID)
+	if len(c.completed) > completedCommandMemory {
+		c.completed = c.completed[len(c.completed)-completedCommandMemory:]
+	}
+	c.pending = append(c.pending, reply)
+	if c.interruptPoll != nil {
+		c.interruptPoll()
+		c.interrupted = true
+	}
+}
+
 // Every command gets a reply: the issuer is blocked on this command ID, so a
 // silent drop costs it a full timeout.
 func (c *commander) execute(ctx context.Context, cmd rdsinstance.Command) rdsinstance.CommandReply {

spinifex/agents/rds/agent/command_test.go

@@ -3,6 +3,9 @@ package agent
 import (
 	"context"
 	"errors"
+	"fmt"
+	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -70,9 +73,9 @@ func TestCommander_UnknownTypeRepliesFailedNotDropped(t *testing.T) {
 	}
 }
 
-// The reply for work the guest actually did rides the next poll. A poll that
+// The reply for work the guest actually did rides a later poll. A poll that
 // fails must not consume it, or the issuer never learns the outcome.
-func TestCommander_RunCarriesRepliesOnTheNextPoll(t *testing.T) {
+func TestCommander_RunCarriesRepliesOnALaterPoll(t *testing.T) {
 	cp := newFakeControlPlane()
 	cp.pollQueue = [][]rdsinstance.Command{
 		{{CommandID: "cmd-1", Type: "reload-parameters"}},
@@ -87,19 +90,15 @@ func TestCommander_RunCarriesRepliesOnTheNextPoll(t *testing.T) {
 	done := make(chan struct{})
 	go func() { c.Run(ctx); close(done) }()
 
-	waitFor(t, func() bool {
-		replies := cp.snapshotReplies()
-		return len(replies) >= 2 && len(replies[1]) == 1
-	}, "the reply to ride the second poll")
+	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-1")) > 0 }, "the reply to ride a later poll")
 	cancel()
 	<-done
 
-	replies := cp.snapshotReplies()
-	if len(replies[0]) != 0 {
+	if replies := cp.snapshotReplies(); len(replies[0]) != 0 {
 		t.Errorf("first poll carried replies %+v, want none", replies[0])
 	}
-	if got := replies[1][0]; got.CommandID != "cmd-1" || got.Status != rdsinstance.CommandStatusSucceeded {
-		t.Errorf("second poll carried %+v, want the succeeded reply for cmd-1", got)
+	if got := repliesFor(cp, "cmd-1")[0]; got.Status != rdsinstance.CommandStatusSucceeded {
+		t.Errorf("the poll carried %+v, want the succeeded reply for cmd-1", got)
 	}
 }
 
@@ -141,16 +140,19 @@ func TestCommander_RetainsAndRetriesTheCompletePendingBatch(t *testing.T) {
 	done = make(chan struct{})
 	go func() { c.Run(ctx); close(done) }()
 	waitFor(t, func() bool {
-		return len(cp.snapshotReplies()) >= 3
-	}, "the retained batch to succeed and the new reply to reach the following poll")
+		return len(repliesFor(cp, "cmd-3")) > 0
+	}, "the retained batch to succeed and the new reply to reach a later poll")
 	cancel()
 	<-done
 
 	replies = cp.snapshotReplies()
 	assert.Equal(t, pending, replies[1], "the complete failed batch must be retried in order")
-	require.Len(t, replies[2], 1)
-	assert.Equal(t, "cmd-3", replies[2][0].CommandID)
-	assert.Equal(t, rdsinstance.CommandStatusSucceeded, replies[2][0].Status)
+	assert.Equal(t, rdsinstance.CommandStatusSucceeded, repliesFor(cp, "cmd-3")[0].Status)
+	for _, later := range replies[2:] {
+		for _, reply := range later {
+			assert.Equal(t, "cmd-3", reply.CommandID, "a batch a poll carried must not be sent again")
+		}
+	}
 }
 
 // A poll error must not spin the loop: without a backoff a gateway that is down
@@ -171,3 +173,340 @@ func TestCommander_BacksOffAfterAFailedPoll(t *testing.T) {
 		t.Errorf("polled %d times in 200ms against a broken channel, want 1", polls)
 	}
 }
+
+// A MariaDB statement that commits waits out a held BACKUP STAGE, and the
+// release is itself a command. Run one at a time, the release would queue behind
+// the statement it is meant to unblock until the hold's own deadline.
+func TestCommander_ABlockedCommandDoesNotKeepOutTheRelease(t *testing.T) {
+	cp := newFakeControlPlane()
+	cp.pollQueue = [][]rdsinstance.Command{
+		{{CommandID: "cmd-password", Type: rdsinstance.CommandSetPassword}},
+		{{CommandID: "cmd-release", Type: rdsinstance.CommandUnquiesce}},
+	}
+	released := make(chan struct{})
+	registry := commandRegistry{
+		rdsinstance.CommandSetPassword: func(ctx context.Context, _ rdsinstance.Command) (string, error) {
+			select {
+			case <-released:
+				return "", nil
+			case <-ctx.Done():
+				return "", ctx.Err()
+			}
+		},
+		rdsinstance.CommandUnquiesce: func(context.Context, rdsinstance.Command) (string, error) {
+			close(released)
+			return "", nil
+		},
+	}
+
+	c := newCommander(cp, registry, 10*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { c.Run(ctx); close(done) }()
+
+	waitFor(t, func() bool {
+		return len(repliesFor(cp, "cmd-password")) > 0 && len(repliesFor(cp, "cmd-release")) > 0
+	}, "both the release and the command it unblocked to be answered")
+	cancel()
+	<-done
+
+	for _, id := range []string{"cmd-password", "cmd-release"} {
+		assert.Equal(t, rdsinstance.CommandStatusSucceeded, repliesFor(cp, id)[0].Status, id)
+	}
+}
+
+// The issuer re-publishes until it has a reply, so a command still running is
+// delivered again. Running the copy would rotate a password twice or open a
+// second backup alongside the first.
+func TestCommander_ARepublishedCommandRunsOnce(t *testing.T) {
+	cp := newFakeControlPlane()
+	cmd := rdsinstance.Command{CommandID: "cmd-1", Type: "reload-parameters"}
+	cp.pollQueue = [][]rdsinstance.Command{{cmd}, {cmd}}
+	var mu sync.Mutex
+	runs := 0
+	secondDelivered := make(chan struct{})
+	registry := commandRegistry{
+		"reload-parameters": func(ctx context.Context, _ rdsinstance.Command) (string, error) {
+			mu.Lock()
+			runs++
+			mu.Unlock()
+			select {
+			case <-secondDelivered:
+			case <-ctx.Done():
+			}
+			return "done", nil
+		},
+	}
+
+	c := newCommander(cp, registry, 10*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { c.Run(ctx); close(done) }()
+
+	// The third poll means both copies have been handed to the commander.
+	waitFor(t, func() bool { return len(cp.snapshotReplies()) >= 3 }, "both copies to be delivered")
+	close(secondDelivered)
+	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-1")) > 0 }, "the reply")
+	cancel()
+	<-done
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, 1, runs, "a re-published copy of a running command must not run again")
+}
+
+// Every reply to id carried on any poll so far.
+func repliesFor(cp *fakeControlPlane, id string) []rdsinstance.CommandReply {
+	var out []rdsinstance.CommandReply
+	for _, batch := range cp.snapshotReplies() {
+		for _, reply := range batch {
+			if reply.CommandID == id {
+				out = append(out, reply)
+			}
+		}
+	}
+	return out
+}
+
+// A copy can arrive after the handler has returned but before a poll has
+// carried its reply. The in-flight set has already let go of it then, so only
+// the completed set stops it running a second time.
+func TestCommander_ACopyArrivingBeforeTheReplyIsCarriedDoesNotRunAgain(t *testing.T) {
+	var mu sync.Mutex
+	runs := map[string]int{}
+	handler := func(_ context.Context, cmd rdsinstance.Command) (string, error) {
+		mu.Lock()
+		runs[cmd.CommandID]++
+		mu.Unlock()
+		return "", nil
+	}
+	c := newCommander(nil, commandRegistry{"reload-parameters": handler}, time.Second)
+	ctx := t.Context()
+	go c.serveSerial(ctx)
+
+	first := rdsinstance.Command{CommandID: "cmd-1", Type: "reload-parameters"}
+	c.dispatch(ctx, first)
+	waitFor(t, func() bool { return len(c.pendingReplies()) == 1 }, "the first run's reply to be pending")
+
+	c.dispatch(ctx, first)
+	// The lane is FIFO, so once this one has answered a queued copy would have run.
+	c.dispatch(ctx, rdsinstance.Command{CommandID: "cmd-2", Type: "reload-parameters"})
+	waitFor(t, func() bool {
+		return slices.ContainsFunc(c.pendingReplies(), func(r rdsinstance.CommandReply) bool { return r.CommandID == "cmd-2" })
+	}, "the following command's reply")
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, 1, runs["cmd-1"], "a copy arriving before the reply was carried must not run")
+	assert.Len(t, c.pendingReplies(), 2, "the copy must not be answered a second time either")
+}
+
+// A quiesce that waited out its issuer's budget behind another command would
+// put the engine in backup mode for a snapshot nobody is taking.
+func TestCommander_AStaleQuiesceNeverTakesAHold(t *testing.T) {
+	cp := newFakeControlPlane()
+	cp.pollQueue = [][]rdsinstance.Command{
+		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},
+		{{CommandID: "cmd-quiesce", Type: rdsinstance.CommandQuiesce, Parameters: []rdsinstance.Parameter{
+			{Name: rdsinstance.CommandParamQuiesceLabel, Value: "orders-db-automated"},
+			{Name: rdsinstance.CommandParamQuiesceDeadlineSeconds, Value: "360"},
+		}}},
+	}
+	engine := &fakeEngine{}
+	registry := newCommandRegistry(engine, &fakeStorage{})
+	applying, release := make(chan struct{}), make(chan struct{})
+	registry[rdsinstance.CommandApplyParams] = func(ctx context.Context, _ rdsinstance.Command) (string, error) {
+		close(applying)
+		<-release
+		return "", nil
+	}
+	c := newCommander(cp, registry, 10*time.Millisecond)
+	c.budget = func(commandType string) time.Duration {
+		if commandType == rdsinstance.CommandQuiesce {
+			return 50 * time.Millisecond
+		}
+		return controlPlaneBudget(commandType)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { c.Run(ctx); close(done) }()
+
+	<-applying
+	waitFor(t, func() bool { return len(cp.snapshotReplies()) >= 3 }, "the quiesce to be delivered behind the apply")
+	time.Sleep(100 * time.Millisecond)
+	close(release)
+	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-quiesce")) > 0 }, "the quiesce to be answered")
+	cancel()
+	<-done
+
+	reply := repliesFor(cp, "cmd-quiesce")[0]
+	assert.Equal(t, rdsinstance.CommandStatusFailed, reply.Status)
+	assert.Contains(t, reply.Message, "expired")
+	assert.Empty(t, engine.label, "an expired quiesce must never reach the engine")
+}
+
+// Engine commands still run one at a time: a parameter apply rewrites the
+// option file a graceful stop would otherwise race.
+func TestCommander_AParameterApplyNeverRunsAlongsideAnEngineStop(t *testing.T) {
+	cp := newFakeControlPlane()
+	cp.pollQueue = [][]rdsinstance.Command{
+		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},
+		{{CommandID: "cmd-stop", Type: rdsinstance.CommandStopEngine}},
+	}
+	var mu sync.Mutex
+	active, most := 0, 0
+	release := make(chan struct{})
+	// Each holds the lane until released, so two running at once are both seen
+	// whichever is scheduled first.
+	engineWork := func(context.Context, rdsinstance.Command) (string, error) {
+		mu.Lock()
+		active++
+		most = max(most, active)
+		mu.Unlock()
+		<-release
+		mu.Lock()
+		active--
+		mu.Unlock()
+		return "", nil
+	}
+	registry := commandRegistry{
+		rdsinstance.CommandApplyParams: engineWork,
+		rdsinstance.CommandStopEngine:  engineWork,
+	}
+	c := newCommander(cp, registry, 10*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { c.Run(ctx); close(done) }()
+
+	waitFor(t, func() bool { return len(cp.snapshotReplies()) >= 3 }, "the stop to be delivered behind the apply")
+	time.Sleep(50 * time.Millisecond)
+	close(release)
+	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-stop")) > 0 }, "the stop to be answered")
+	cancel()
+	<-done
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, 1, most, "the engine stop ran while the parameter apply was still running")
+}
+
+// The agent bounds each command by its issuer's budget, so a type the control
+// plane has no budget for would run on a guess.
+func TestCommandRegistry_EveryTypeHasAControlPlaneBudget(t *testing.T) {
+	for commandType := range newCommandRegistry(&fakeEngine{}, &fakeStorage{}) {
+		if _, ok := rdsinstance.CommandBudget(commandType); !ok {
+			t.Errorf("%s has no control-plane budget", commandType)
+		}
+	}
+}
+
+func (c *commander) pendingReplies() []rdsinstance.CommandReply {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return slices.Clone(c.pending)
+}
+
+// The first copy can reach the agent late, after a gateway or bus outage, with
+// its issuer about to give up. A quiesce granted a fresh budget then would take
+// a hold nobody is left to release.
+func TestCommander_AQuiesceStampedNearlyOutOfBudgetNeverTakesAHold(t *testing.T) {
+	cp := newFakeControlPlane()
+	cp.pollQueue = [][]rdsinstance.Command{
+		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},
+		{{CommandID: "cmd-quiesce", Type: rdsinstance.CommandQuiesce, RemainingBudgetMs: 50, Parameters: []rdsinstance.Parameter{
+			{Name: rdsinstance.CommandParamQuiesceLabel, Value: "orders-db-automated"},
+			{Name: rdsinstance.CommandParamQuiesceDeadlineSeconds, Value: "360"},
+		}}},
+	}
+	engine := &fakeEngine{}
+	registry := newCommandRegistry(engine, &fakeStorage{})
+	applying, release := make(chan struct{}), make(chan struct{})
+	registry[rdsinstance.CommandApplyParams] = func(context.Context, rdsinstance.Command) (string, error) {
+		close(applying)
+		<-release
+		return "", nil
+	}
+	c := newCommander(cp, registry, 10*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { c.Run(ctx); close(done) }()
+
+	<-applying
+	waitFor(t, func() bool { return len(cp.snapshotReplies()) >= 3 }, "the quiesce to be delivered behind the apply")
+	// The quiesce was received before the third poll began, so its stamped budget
+	// has run out 50ms after this point.
+	queuedAt := time.Now()
+	waitFor(t, func() bool { return time.Since(queuedAt) > 50*time.Millisecond }, "the stamped budget to pass")
+	close(release)
+	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-quiesce")) > 0 }, "the quiesce to be answered")
+	cancel()
+	<-done
+
+	reply := repliesFor(cp, "cmd-quiesce")[0]
+	assert.Equal(t, rdsinstance.CommandStatusFailed, reply.Status)
+	assert.Contains(t, reply.Message, "expired")
+	assert.Empty(t, engine.label, "a quiesce whose issuer had all but given up must never reach the engine")
+}
+
+// A full lane refuses at once rather than queueing work its issuer will not
+// outlive, and the refusal is the command's answer: a re-published copy must
+// not slip in once the lane has room.
+func TestCommander_AFullLaneRefusesAtOnceAndNeverRunsTheCommand(t *testing.T) {
+	var mu sync.Mutex
+	runs := map[string]int{}
+	running, release := make(chan struct{}, 1), make(chan struct{})
+	handler := func(_ context.Context, cmd rdsinstance.Command) (string, error) {
+		mu.Lock()
+		runs[cmd.CommandID]++
+		mu.Unlock()
+		if cmd.CommandID == "cmd-blocker" {
+			running <- struct{}{}
+			<-release
+		}
+		return "", nil
+	}
+	c := newCommander(nil, commandRegistry{"reload-parameters": handler}, time.Second)
+	ctx := t.Context()
+	go c.serveSerial(ctx)
+
+	c.dispatch(ctx, rdsinstance.Command{CommandID: "cmd-blocker", Type: "reload-parameters"})
+	<-running
+	for i := range serialQueueDepth {
+		c.dispatch(ctx, rdsinstance.Command{CommandID: fmt.Sprintf("cmd-queued-%d", i), Type: "reload-parameters"})
+	}
+
+	overflow := rdsinstance.Command{CommandID: "cmd-overflow", Type: "reload-parameters"}
+	c.dispatch(ctx, overflow)
+	refused := slices.IndexFunc(c.pendingReplies(), func(r rdsinstance.CommandReply) bool { return r.CommandID == overflow.CommandID })
+	require.GreaterOrEqual(t, refused, 0, "a full lane must answer at once, before anything ahead of it finishes")
+	assert.Equal(t, rdsinstance.CommandStatusFailed, c.pendingReplies()[refused].Status)
+	assert.Contains(t, c.pendingReplies()[refused].Message, "busy")
+
+	close(release)
+	last := fmt.Sprintf("cmd-queued-%d", serialQueueDepth-1)
+	waitFor(t, func() bool {
+		return slices.ContainsFunc(c.pendingReplies(), func(r rdsinstance.CommandReply) bool { return r.CommandID == last })
+	}, "the queue to drain")
+	c.dispatch(ctx, overflow)
+	// FIFO again: once this answers, a copy let onto the lane would have run.
+	c.dispatch(ctx, rdsinstance.Command{CommandID: "cmd-after", Type: "reload-parameters"})
+	waitFor(t, func() bool {
+		return slices.ContainsFunc(c.pendingReplies(), func(r rdsinstance.CommandReply) bool { return r.CommandID == "cmd-after" })
+	}, "the command after the re-published copy")
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Zero(t, runs[overflow.CommandID], "a refused command must never run")
+	answers := 0
+	for _, r := range c.pendingReplies() {
+		if r.CommandID == overflow.CommandID {
+			answers++
+		}
+	}
+	assert.Equal(t, 1, answers, "a re-published copy of a refused command must be ignored, not answered again")
+}

spinifex/agents/rds/agent/engine_mariadb.go

@@ -94,6 +94,8 @@ type mariadbEngine struct {
 	socket    string
 	rcService string
 	service   string
+	// How long a password rotation waits for a backup hold to end.
+	holdWait time.Duration
 }
 
 var _ engine = (*mariadbEngine)(nil)
@@ -151,6 +153,7 @@ func newMariaDBEngine(cfg Config, rules controlPlaneRules, run commandRunner, st
 		socket:    mariadbSocketPath(cfg),
 		rcService: cfg.RCService,
 		service:   cfg.EngineService,
+		holdWait:  mariadbPasswordHoldWait,
 		parameterManager: parameterManager{
 			probe: probe,
 			params: parameterStore{
@@ -380,6 +383,11 @@ SET SESSION slow_query_log = 0;
 SET SESSION sql_mode = 'NO_ENGINE_SUBSTITUTION';
 `
 
+// ALTER USER can take effect in memory and then wait at its commit for a held
+// BACKUP STAGE to end. Waiting for the agent's own hold before the statement, for
+// less than the control plane's budget, keeps that hold from half-applying it.
+const mariadbPasswordHoldWait = 20 * time.Second
+
 // Rotates the master's password. The statement is built here rather than by the
 // client, which has neither psql's identifier interpolation nor its parameter
 // quoting, and it rides stdin so it is never visible in the process table.
@@ -395,16 +403,34 @@ func (e *mariadbEngine) SetPassword(ctx context.Context, username, password stri
 		return fmt.Errorf("refusing to set the password of role %q: %w", username, err)
 	}
 
+	// Only the agent's own backup hold is waited out before the statement. A lock
+	// someone else holds, such as a customer FTWRL, is met inside it and ended by
+	// lock_wait_timeout, by which point the password may already have changed.
 	escaped := sqlLiteral(password)
-	sql := fmt.Sprintf("%sALTER USER '%s'@'%s' IDENTIFIED BY '%s';\n",
-		mariadbSessionGuard, username, mariadbMasterHost, escaped)
-	if _, err := e.clientRun(ctx, sql); err != nil {
-		// The client echoes the failing statement, which carries the escaped form
-		// of the password rather than the raw one — so redacting only the raw one
-		// would leak exactly the passwords that needed escaping.
-		return fmt.Errorf("apply the master password: %s", redact(redact(err.Error(), password), escaped))
+	return e.outsideHold(ctx, e.holdWait, func() error {
+		// Computed after the hold wait, so the statement gets only what is left.
+		sql := fmt.Sprintf("%sSET SESSION lock_wait_timeout = %d;\nALTER USER '%s'@'%s' IDENTIFIED BY '%s';\n",
+			mariadbSessionGuard, statementLockWait(ctx), username, mariadbMasterHost, escaped)
+		if _, err := e.clientRun(ctx, sql); err != nil {
+			// The client echoes the failing statement, which carries the escaped form
+			// of the password rather than the raw one — so redacting only the raw one
+			// would leak exactly the passwords that needed escaping.
+			return fmt.Errorf("apply the master password: %s", redact(redact(err.Error(), password), escaped))
+		}
+		return nil
+	})
+}
+
+// The server-side lock wait for a statement, in whole seconds: what remains of
+// ctx less a margin for the reply, so the server gives up its wait before the
+// client is killed. A ctx with no deadline gets the full password budget.
+func statementLockWait(ctx context.Context) int {
+	const replyMargin = 2 * time.Second
+	remaining := mariadbPasswordHoldWait
+	if deadline, ok := ctx.Deadline(); ok {
+		remaining = time.Until(deadline) - replyMargin
 	}
-	return nil
+	return max(1, int(remaining.Seconds()))
 }
 
 // Installs the resolved set and applies the half a running server can adopt.
@@ -714,7 +740,9 @@ func (e *mariadbEngine) Quiesce(ctx context.Context, label string, hold time.Dur
 
 	// Held across the whole start, so a second quiesce waits and then finds the
 	// first one's hold rather than opening a concurrent backup alongside it.
-	e.mu.Lock()
+	if err := e.lockWithin(ctx); err != nil {
+		return err
+	}
 	defer e.mu.Unlock()
 	if e.held != nil {
 		return fmt.Errorf("the engine is already quiesced for backup %s", e.held.label)

spinifex/agents/rds/agent/engine_mariadb_test.go

@@ -1151,3 +1151,123 @@ func TestMariaDBEngine_EveryResolvedDynamicValueRenders(t *testing.T) {
 		}
 	}
 }
+
+// ALTER USER under a held BACKUP STAGE swaps the password in memory and then
+// waits at its commit, so a rotation issued mid-backup must not reach the
+// engine until the hold is released.
+func TestMariaDBEngine_SetPasswordWaitsForTheBackupHoldToEnd(t *testing.T) {
+	runner := &recordingRunner{}
+	engine := newTestMariaDBEngine(t, runner.run)
+	engine.startSess = func(context.Context, command) (engineSession, error) { return &fakeSession{}, nil }
+	if err := engine.Quiesce(context.Background(), "orders-db-automated", time.Minute); err != nil {
+		t.Fatalf("Quiesce: %v", err)
+	}
+
+	rotated := make(chan error, 1)
+	go func() { rotated <- engine.SetPassword(context.Background(), "mulgamaster", "n3w-pw") }()
+	select {
+	case err := <-rotated:
+		t.Fatalf("SetPassword returned %v while the engine was held for a backup", err)
+	case <-time.After(3 * holdPollInterval):
+	}
+
+	if err := engine.Unquiesce(context.Background()); err != nil {
+		t.Fatalf("Unquiesce: %v", err)
+	}
+	select {
+	case err := <-rotated:
+		if err != nil {
+			t.Fatalf("SetPassword after the release: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("SetPassword did not run once the hold was released")
+	}
+	if len(runner.calls) != 1 || !strings.Contains(runner.calls[0].Stdin, "ALTER USER") {
+		t.Errorf("ran %d commands, want the one rotation after the release", len(runner.calls))
+	}
+}
+
+// A hold that outlasts the wait fails the rotation before anything is sent, so
+// the control plane's error is the truth rather than half of it.
+func TestMariaDBEngine_SetPasswordFailsUntouchedWhenTheHoldOutlastsItsWait(t *testing.T) {
+	runner := &recordingRunner{}
+	engine := newTestMariaDBEngine(t, runner.run)
+	engine.holdWait = 3 * holdPollInterval
+	engine.startSess = func(context.Context, command) (engineSession, error) { return &fakeSession{}, nil }
+	if err := engine.Quiesce(context.Background(), "orders-db-automated", time.Minute); err != nil {
+		t.Fatalf("Quiesce: %v", err)
+	}
+
+	if err := engine.SetPassword(context.Background(), "mulgamaster", "n3w-pw"); err == nil {
+		t.Fatal("SetPassword succeeded while the engine stayed held for a backup")
+	}
+	if len(runner.calls) != 0 {
+		t.Errorf("ran %d commands against a held engine, want none", len(runner.calls))
+	}
+}
+
+// The wait has to end before the control plane gives up on the command, or the
+// rotation can still land after the caller was told it failed.
+func TestMariaDBEngine_PasswordHoldWaitFitsInsideTheControlPlaneTimeout(t *testing.T) {
+	budget, ok := rdsinstance.CommandBudget(rdsinstance.CommandSetPassword)
+	if !ok {
+		t.Fatal("the control plane has no budget for set-password")
+	}
+	if mariadbPasswordHoldWait >= budget-5*time.Second {
+		t.Errorf("a %s hold wait leaves under 5s of the control plane's %s set-password budget for the statement", mariadbPasswordHoldWait, budget)
+	}
+}
+
+// The statement is bounded by what is left of the command, so the server gives
+// up a lock wait before the client is killed under it.
+func TestMariaDBEngine_SetPasswordBoundsTheStatementByTheRemainingBudget(t *testing.T) {
+	runner := &recordingRunner{}
+	engine := newTestMariaDBEngine(t, runner.run)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	if err := engine.SetPassword(ctx, "mulgamaster", "n3w-pw"); err != nil {
+		t.Fatalf("SetPassword: %v", err)
+	}
+	sql := runner.calls[0].Stdin
+	at := strings.Index(sql, "SET SESSION lock_wait_timeout = ")
+	if at < 0 || at > strings.Index(sql, "ALTER USER") {
+		t.Fatalf("SQL %q does not bound the statement's lock wait before it runs", sql)
+	}
+	var seconds int
+	if _, err := fmt.Sscanf(sql[at:], "SET SESSION lock_wait_timeout = %d;", &seconds); err != nil {
+		t.Fatalf("parse the lock wait: %v", err)
+	}
+	if seconds < 1 || seconds > 8 {
+		t.Errorf("lock wait = %ds with 10s of budget left, want it inside the budget less the reply margin", seconds)
+	}
+}
+
+// A quiesce that cannot get the hold bookkeeping must give up with its command,
+// not wait on the mutex after its issuer has moved on.
+func TestMariaDBEngine_QuiesceGivesUpWaitingForTheBookkeepingWithItsContext(t *testing.T) {
+	engine := newTestMariaDBEngine(t, (&recordingRunner{}).run)
+	started := 0
+	engine.startSess = func(context.Context, command) (engineSession, error) {
+		started++
+		return &fakeSession{}, nil
+	}
+	engine.mu.Lock()
+	defer engine.mu.Unlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*holdPollInterval)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- engine.Quiesce(ctx, "orders-db-automated", time.Minute) }()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("Quiesce succeeded without the bookkeeping lock")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Quiesce waited on the bookkeeping past its context")
+	}
+	if started != 0 {
+		t.Errorf("opened %d backup sessions, want none", started)
+	}
+}

spinifex/agents/rds/agent/engine_postgres.go

@@ -516,7 +516,9 @@ func (e *postgresEngine) Quiesce(ctx context.Context, label string, hold time.Du
 
 	// Held across the whole start, so a second quiesce waits and then finds the
 	// first one's hold rather than opening a concurrent backup alongside it.
-	e.mu.Lock()
+	if err := e.lockWithin(ctx); err != nil {
+		return err
+	}
 	defer e.mu.Unlock()
 	if e.held != nil {
 		return fmt.Errorf("the engine is already quiesced for backup %s", e.held.label)

spinifex/agents/rds/agent/quiesce.go

@@ -114,6 +114,48 @@ func (q *quiesceState) releaseHold(ctx context.Context, releaseSQL string) error
 	return nil
 }
 
+// Takes mu, giving up when ctx ends. A quiesce that cannot get the bookkeeping
+// inside its budget must fail rather than start a backup its issuer abandoned.
+func (q *quiesceState) lockWithin(ctx context.Context) error {
+	for !q.mu.TryLock() {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("wait for the backup bookkeeping: %w", ctx.Err())
+		case <-time.After(holdPollInterval):
+		}
+	}
+	return nil
+}
+
+// How often a statement waiting for a backup hold to end looks again.
+const holdPollInterval = 100 * time.Millisecond
+
+// Runs fn while no backup hold is on the engine, and keeps a new one from
+// starting until it returns. A hold that outlasts wait fails the call before fn
+// has touched the engine.
+func (q *quiesceState) outsideHold(ctx context.Context, wait time.Duration, fn func() error) error {
+	deadline := time.NewTimer(wait)
+	defer deadline.Stop()
+	for {
+		// Not Lock: a quiesce holds mu while its stages wait on their own locks,
+		// and this wait has to stay bounded by ctx and wait.
+		if q.mu.TryLock() {
+			if q.held == nil {
+				defer q.mu.Unlock()
+				return fn()
+			}
+			q.mu.Unlock()
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-deadline.C:
+			return fmt.Errorf("the engine is still held for a backup after %s", wait)
+		case <-time.After(holdPollInterval):
+		}
+	}
+}
+
 // The deadline. Ending the session is enough — both engines release the hold
 // with it — so this never has to reach the engine itself.
 func (q *quiesceState) expire(label string) {

spinifex/domains/rds/awsapi/agent.go

@@ -201,8 +201,9 @@ type PollDBCommandsInput struct {
 	Replies              []rdsinstance.CommandReply `locationName:"Replies" locationNameList:"member"`
 }
 
-// PollDBCommandsOutput holds at most one command per poll: each is a discrete guest operation the
-// agent must complete and report before the next is safe to issue.
+// PollDBCommandsOutput holds at most one command per poll. The agent runs engine commands one at a
+// time in arrival order and answers on a later poll; only unquiesce runs beside them, so it can release
+// a backup hold a queued command is waiting on. A command that outlives its type's budget is not run.
 type PollDBCommandsOutput struct {
 	Commands []rdsinstance.Command `locationName:"Commands" locationNameList:"member"`
 }

spinifex/domains/rds/instance/agent.go

@@ -137,6 +137,10 @@ type Command struct {
 	Type       string      `json:"type" locationName:"Type"`
 	Parameters []Parameter `json:"parameters,omitempty" locationName:"Parameters" locationNameList:"member" xml:"Parameters>member"`
 	IssuedAt   *time.Time  `json:"issuedAt,omitempty" locationName:"IssuedAt" type:"timestamp"`
+	// How much of the issuer's budget was left when this copy was published. A
+	// duration rather than a deadline, so the guest needs no clock agreement with
+	// the host; zero means the issuer predates the field.
+	RemainingBudgetMs int64 `json:"remainingBudgetMs,omitempty" locationName:"RemainingBudgetMs"`
 }
 
 // CommandReply is the agent's result for a command from an earlier poll, carried on the next poll

spinifex/domains/rds/instance/commands.go

@@ -75,6 +75,24 @@ const (
 	commandRepublishEvery = 2 * time.Second
 )
 
+// How long the control plane waits for each command type by default. The agent
+// reads the same table, so it never starts work its issuer has stopped waiting for.
+var commandBudgets = map[string]time.Duration{
+	CommandSetPassword:    setPasswordTimeout,
+	CommandApplyParams:    defaultApplyParamsTimeout,
+	CommandStopEngine:     stopEngineTimeout,
+	CommandGrowFilesystem: growFilesystemTimeout,
+	CommandQuiesce:        quiesceTimeout,
+	CommandUnquiesce:      unquiesceTimeout,
+}
+
+// CommandBudget returns how long the control plane waits for a reply to a
+// command of this type, and false for a type it never issues.
+func CommandBudget(commandType string) (time.Duration, bool) {
+	budget, ok := commandBudgets[commandType]
+	return budget, ok
+}
+
 // ErrCommandUnreachable means the agent was unreachable or missed the command's
 // budget. Callers that can degrade — the graceful engine stop — check for this;
 // callers that must not, like a password rotation, surface it as retryable.
@@ -152,10 +170,6 @@ func (s *Service) issueCommand(ctx context.Context, accountID, dbInstanceIdentif
 		Parameters: params,
 		IssuedAt:   &issuedAt,
 	}
-	data, err := json.Marshal(cmd)
-	if err != nil {
-		return nil, err
-	}
 
 	// Subscribed before the first publish, so a reply cannot land in the gap
 	// between the command going out and the issuer starting to listen.
@@ -173,7 +187,15 @@ func (s *Service) issueCommand(ctx context.Context, accountID, dbInstanceIdentif
 	defer cancel()
 	subject := BusCommandSubject(accountID, dbInstanceIdentifier)
 
+	deadline, _ := deadlineCtx.Deadline()
 	for {
+		// Stamped per copy, so a copy that reaches the agent late carries only what
+		// its issuer has left rather than a whole fresh budget.
+		cmd.RemainingBudgetMs = max(time.Until(deadline).Milliseconds(), 1)
+		data, err := json.Marshal(cmd)
+		if err != nil {
+			return nil, err
+		}
 		if err := s.nc.Publish(subject, data); err != nil {
 			return nil, fmt.Errorf("rds: publish %s command to %s: %w", commandType, dbInstanceIdentifier, err)
 		}

spinifex/domains/rds/instance/commands_test.go

@@ -2,6 +2,7 @@ package instance
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -80,3 +81,26 @@ func TestIssueCommand_CorrelatesTheReplyByCommandID(t *testing.T) {
 	require.NotEmpty(t, issued)
 	assert.Equal(t, issued[0].CommandID, reply.CommandID)
 }
+
+// Each copy carries what its issuer has left, so a copy that reaches the agent
+// late cannot be granted a whole budget its issuer is about to give up on.
+func TestIssueCommand_StampsEachCopyWithTheBudgetLeft(t *testing.T) {
+	t.Parallel()
+	h := newLifecycleHarness(t, false)
+	h.agent.silenceType(CommandQuiesce)
+
+	const budget = 4500 * time.Millisecond
+	_, err := h.svc.issueCommand(t.Context(), testAccountID, testDBID, CommandQuiesce, budget, nil)
+	require.ErrorIs(t, err, ErrCommandUnreachable)
+
+	issued := h.agent.received()
+	require.GreaterOrEqual(t, len(issued), 2, "the issuer must have re-published at least once")
+	for i, copied := range issued {
+		assert.Positive(t, copied.RemainingBudgetMs, "copy %d carries no remaining budget", i)
+		assert.LessOrEqual(t, copied.RemainingBudgetMs, budget.Milliseconds(), "copy %d", i)
+		if i > 0 {
+			assert.LessOrEqual(t, copied.RemainingBudgetMs, issued[i-1].RemainingBudgetMs-commandRepublishEvery.Milliseconds()/2,
+				"copy %d was stamped with as much budget as the copy before it", i)
+		}
+	}
+}

tests/e2e/harness/imds_diag.go

@@ -51,6 +51,23 @@ func DumpIMDSDatapathDiagnostics(t *testing.T, subnetID, guestIP, artifactDir st
 				"list", "Port_Binding"},
 			grepFor: []string{guestIP},
 		},
+		{
+			filename: "imds-endpoint-links.txt",
+			label:    "ip -br link (endpoint up, MAC matching the demux flow's mod_dl_dst?)",
+			argv:     []string{"ip", "-br", "link", "show"},
+			grepFor:  []string{"ime-"},
+		},
+		{
+			filename: "imds-endpoint-addrs.txt",
+			label:    "ip -br addr (endpoint owns 169.254.169.254/.253?)",
+			argv:     []string{"ip", "-br", "addr", "show"},
+			grepFor:  []string{"ime-"},
+		},
+		{
+			filename: "imds-endpoint-sysctls.txt",
+			label:    "endpoint rp_filter/accept_local (a strict rp_filter drops guest requests)",
+			argv:     []string{"sysctl", "-a", "-r", `conf\.(all|default|ime-.*)\.(rp_filter|accept_local)$`},
+		},
 		{
 			filename: "imds-reply-rules.txt",
 			label:    "ip rule (per-tap 'oif <endpoint> lookup <table>' reply rule present?)",

tests/e2e/rds/backup_overlap_test.go

@@ -0,0 +1,258 @@
+//go:build e2e
+
+package rds
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/rds"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	// The status an instance holds for the length of a snapshot, from the quiesce
+	// to the release.
+	dbInstanceBackingUp = "backing-up"
+
+	// How often the watcher samples the status. The hold lasts about a second, so
+	// the harness's 10s status interval would step straight over it.
+	backingUpSampleEvery = 100 * time.Millisecond
+
+	// One window: the reconciler's pass that fires it, the backup itself, and the
+	// minute boundary a later attempt has to wait for.
+	backupOverlapAttemptTimeout = 5 * time.Minute
+
+	// The hold is short and the rotations are discrete, so one backup can finish
+	// between two of them. Each further attempt opens a fresh window.
+	maxBackupOverlapAttempts = 3
+
+	// A probe table only this check writes to, so its DDL and commit say nothing
+	// about any row the rest of the test reads.
+	backupProbeTable = "e2e_backup_probe"
+)
+
+// Rotates the master password back and forth through an automated backup until
+// one rotation is seen in flight while the instance reports backing-up. conn is
+// kept on whichever password is live, so later subtests connect with it.
+func rotateDuringAutomatedBackup(t *testing.T, f *Fixture, client harness.SSHTarget, id string, conn *harness.MariaDBConn) {
+	t.Helper()
+	// Put back once done, so no later subtest can be interrupted by a retry of
+	// a backup that failed inside the window this opened.
+	t.Cleanup(func() { pinBackupWindowClearOfTheRun(t, f, id) })
+
+	var lastOpened time.Time
+	for attempt := 1; attempt <= maxBackupOverlapAttempts; attempt++ {
+		before, err := automatedSnapshots(f, id)
+		require.NoError(t, err)
+		beforeIDs := automatedSnapshotIDs(before)
+
+		lastOpened = openBackupWindowNow(t, f, id, lastOpened)
+		watch := watchStatus(f, id)
+		overlapped := rotateUntilTheBackupLands(t, f, client, id, conn, beforeIDs, watch)
+		watch.stop()
+		if err := watch.err(); err != nil {
+			t.Logf("the status watcher missed samples; the last describe failed with: %v", err)
+		}
+
+		harness.WaitForDBInstanceAvailable(t, f.AWS, id, harness.WithPoll(2*time.Second),
+			harness.WithTimeout(2*time.Minute))
+		assertTheEngineIsReleased(t, client, *conn)
+
+		if overlapped {
+			return
+		}
+		t.Logf("attempt %d: the backup of %s finished between two rotations; opening another window", attempt, id)
+	}
+	t.Fatalf("no rotation overlapped an automated backup of %s in %d windows", id, maxBackupOverlapAttempts)
+}
+
+// Opens the backup window on the current minute, or on the next one when the
+// previous attempt's window already opened on this one: a window fires once,
+// and only for a backup older than its opening.
+func openBackupWindowNow(t *testing.T, f *Fixture, id string, previous time.Time) time.Time {
+	t.Helper()
+	opens := time.Now().UTC().Truncate(time.Minute)
+	if !opens.After(previous) {
+		opens = previous.Add(time.Minute)
+	}
+	window := dailyWindowAt(opens, 30*time.Minute)
+	harness.Step(t, "Opening the backup window of %q at %s", id, window)
+	_, err := f.AWS.RDS.ModifyDBInstance(&rds.ModifyDBInstanceInput{
+		DBInstanceIdentifier:  aws.String(id),
+		PreferredBackupWindow: aws.String(window),
+		ApplyImmediately:      aws.Bool(true),
+	})
+	require.NoError(t, err, "modify-db-instance backup window")
+	return opens
+}
+
+// Rotates until the backup has produced its snapshot and the instance has left
+// backing-up, and reports whether any rotation was in flight while it was in it.
+func rotateUntilTheBackupLands(t *testing.T, f *Fixture, client harness.SSHTarget, id string,
+	conn *harness.MariaDBConn, beforeIDs []string, watch *statusWatch) bool {
+	t.Helper()
+	deadline := time.Now().Add(backupOverlapAttemptTimeout)
+	overlapped := false
+	for {
+		require.True(t, time.Now().Before(deadline),
+			"no automated backup of %s finished within %s of its window opening", id, backupOverlapAttemptTimeout)
+
+		next := mariaRotatedPassword
+		if conn.Password == mariaRotatedPassword {
+			next = mariaBackupRotationPassword
+		}
+		issued := time.Now()
+		_, err := f.AWS.RDS.ModifyDBInstance(&rds.ModifyDBInstanceInput{
+			DBInstanceIdentifier: aws.String(id),
+			MasterUserPassword:   aws.String(next),
+			ApplyImmediately:     aws.Bool(true),
+		})
+		answered := time.Now()
+		if watch.sawBetween(dbInstanceBackingUp, issued, answered) {
+			overlapped = true
+			t.Logf("rotation in flight %s-%s overlapped the backup (err=%v)",
+				issued.Format(time.RFC3339Nano), answered.Format(time.RFC3339Nano), err)
+		}
+		assertRotationIsWhole(t, client, conn, next, err)
+
+		if watch.latest() != dbInstanceBackingUp && newAutomatedSnapshot(t, f, id, beforeIDs) {
+			return overlapped
+		}
+	}
+}
+
+// An accepted rotation leaves only the new password working; a refused one
+// leaves only the old. Anything else is a rotation the caller cannot reason about.
+func assertRotationIsWhole(t *testing.T, client harness.SSHTarget, conn *harness.MariaDBConn, next string, err error) {
+	t.Helper()
+	rotated := *conn
+	rotated.Password = next
+	live, retired := rotated, *conn
+	if err != nil {
+		harness.AssertAWSError(t, err, "InvalidDBInstanceState")
+		live, retired = *conn, rotated
+	}
+
+	assert.Equal(t, mariaNote, mariaValue(t, client, live,
+		fmt.Sprintf("SELECT note FROM %s WHERE id = 1;", mariaTable)),
+		"the password the API's answer implies must be the one that works (modify err=%v)", err)
+	out, loginErr := harness.TryMariaDB(client, retired, "SELECT 1;")
+	require.Error(t, loginErr, "the password the API's answer rules out still connects (modify err=%v): %s", err, out)
+	require.Contains(t, out, "Access denied", "the other password must be refused by the engine, not the network")
+
+	*conn = live
+}
+
+// A held BACKUP STAGE blocks DDL and commits, so a statement doing both
+// completing is what shows the hold was released rather than left to expire.
+func assertTheEngineIsReleased(t *testing.T, client harness.SSHTarget, conn harness.MariaDBConn) {
+	t.Helper()
+	harness.MariaDB(t, client, conn, fmt.Sprintf(
+		"CREATE TABLE IF NOT EXISTS %s (id int primary key); REPLACE INTO %s VALUES (1);",
+		backupProbeTable, backupProbeTable))
+}
+
+func newAutomatedSnapshot(t *testing.T, f *Fixture, id string, beforeIDs []string) bool {
+	t.Helper()
+	snapshots, err := automatedSnapshots(f, id)
+	require.NoError(t, err)
+	for _, snapshot := range snapshots {
+		if !slices.Contains(beforeIDs, aws.StringValue(snapshot.DBSnapshotIdentifier)) &&
+			aws.StringValue(snapshot.Status) == harness.DBSnapshotAvailable {
+			return true
+		}
+	}
+	return false
+}
+
+func pinBackupWindowClearOfTheRun(t *testing.T, f *Fixture, id string) {
+	t.Helper()
+	backup, _ := windowsClearOfTheRun()
+	_, err := f.AWS.RDS.ModifyDBInstance(&rds.ModifyDBInstanceInput{
+		DBInstanceIdentifier:  aws.String(id),
+		PreferredBackupWindow: aws.String(backup),
+		ApplyImmediately:      aws.Bool(true),
+	})
+	assert.NoError(t, err, "move the backup window of %s clear of the run again", id)
+}
+
+// Samples an instance's status on its own goroutine, fast enough to see a hold
+// that lasts about a second.
+type statusWatch struct {
+	cancel context.CancelFunc
+	done   chan struct{}
+
+	mu      sync.Mutex
+	samples []statusSample
+	lastErr error
+}
+
+type statusSample struct {
+	at     time.Time
+	status string
+}
+
+func watchStatus(f *Fixture, id string) *statusWatch {
+	ctx, cancel := context.WithCancel(context.Background())
+	w := &statusWatch{cancel: cancel, done: make(chan struct{})}
+	go func() {
+		defer close(w.done)
+		ticker := time.NewTicker(backingUpSampleEvery)
+		defer ticker.Stop()
+		for {
+			instance, err := harness.DescribeDBInstance(f.AWS, id)
+			w.mu.Lock()
+			if err != nil {
+				w.lastErr = err
+			} else {
+				w.samples = append(w.samples, statusSample{at: time.Now(), status: aws.StringValue(instance.DBInstanceStatus)})
+			}
+			w.mu.Unlock()
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+	return w
+}
+
+func (w *statusWatch) stop() {
+	w.cancel()
+	<-w.done
+}
+
+func (w *statusWatch) err() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.lastErr
+}
+
+func (w *statusWatch) latest() string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if len(w.samples) == 0 {
+		return ""
+	}
+	return w.samples[len(w.samples)-1].status
+}
+
+func (w *statusWatch) sawBetween(status string, from, to time.Time) bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	for _, s := range w.samples {
+		if s.status == status && !s.at.Before(from) && !s.at.After(to) {
+			return true
+		}
+	}
+	return false
+}

tests/e2e/rds/main_test.go

@@ -302,17 +302,29 @@ func (f *Fixture) dbDiag(t *testing.T) harness.DBDiag {
 // The suite's own create request: valid as it stands, so a caller mutates only
 // the field it cares about.
 func validCreateInput(id string) *rds.CreateDBInstanceInput {
+	backup, maintenance := windowsClearOfTheRun()
 	return &rds.CreateDBInstanceInput{
-		DBInstanceIdentifier: aws.String(id),
-		Engine:               aws.String(dbEngine),
-		DBInstanceClass:      aws.String(dbClass),
-		AllocatedStorage:     aws.Int64(dbStorageGiB),
-		DBName:               aws.String(dbName),
-		MasterUsername:       aws.String(dbMasterUser),
-		MasterUserPassword:   aws.String(dbMasterPassword),
+		DBInstanceIdentifier:       aws.String(id),
+		Engine:                     aws.String(dbEngine),
+		DBInstanceClass:            aws.String(dbClass),
+		AllocatedStorage:           aws.Int64(dbStorageGiB),
+		DBName:                     aws.String(dbName),
+		MasterUsername:             aws.String(dbMasterUser),
+		MasterUserPassword:         aws.String(dbMasterPassword),
+		PreferredBackupWindow:      aws.String(backup),
+		PreferredMaintenanceWindow: aws.String(maintenance),
 	}
 }
 
+// Windows that cannot open while the suite runs. An assigned window lands
+// wherever the identifier hashes, so without these whether a backup or a
+// maintenance pass interrupts a subtest depends on the hour the cell runs.
+func windowsClearOfTheRun() (backup, maintenance string) {
+	now := time.Now().UTC()
+	return dailyWindowAt(now.Add(12*time.Hour), 30*time.Minute),
+		weeklyWindowAt(now.Add(18*time.Hour), 30*time.Minute)
+}
+
 // createDBInstance creates the suite's standard instance and returns the create
 // response's own view of it — status creating, no endpoint yet.
 func createDBInstance(t *testing.T, f *Fixture, id string, opts ...func(*rds.CreateDBInstanceInput)) *rds.DBInstance {

tests/e2e/rds/mariadb_test.go

@@ -31,6 +31,8 @@ const (
 	// and no MariaDB client can quote on our behalf, so rds-init's escaping and
 	// the agent's rotation are under test rather than assumed.
 	mariaRotatedPassword = "e2eR0tated\\Maria'2"
+	// The other half of the pair the backup-overlap rotations alternate between.
+	mariaBackupRotationPassword = "e2eDur1ng\\Backup'3"
 
 	// The table every MariaDB assertion is made against: a stop, a start or a VM
 	// replacement that loses it has not preserved the datadir.
@@ -64,14 +66,17 @@ var mariaUnpinnedVersions = []string{"11.4", "10.11", "11.8.8"}
 // field it cares about. Same class and storage as the PostgreSQL spec: the
 // engine is what is under test, not the sizing.
 func validMariaDBCreateInput(id string) *rds.CreateDBInstanceInput {
+	backup, maintenance := windowsClearOfTheRun()
 	return &rds.CreateDBInstanceInput{
-		DBInstanceIdentifier: aws.String(id),
-		Engine:               aws.String(mariaEngine),
-		DBInstanceClass:      aws.String(dbClass),
-		AllocatedStorage:     aws.Int64(dbStorageGiB),
-		DBName:               aws.String(mariaDBName),
-		MasterUsername:       aws.String(mariaMasterUser),
-		MasterUserPassword:   aws.String(mariaMasterPassword),
+		DBInstanceIdentifier:       aws.String(id),
+		Engine:                     aws.String(mariaEngine),
+		DBInstanceClass:            aws.String(dbClass),
+		AllocatedStorage:           aws.Int64(dbStorageGiB),
+		DBName:                     aws.String(mariaDBName),
+		MasterUsername:             aws.String(mariaMasterUser),
+		MasterUserPassword:         aws.String(mariaMasterPassword),
+		PreferredBackupWindow:      aws.String(backup),
+		PreferredMaintenanceWindow: aws.String(maintenance),
 	}
 }
 
@@ -268,6 +273,13 @@ func TestMariaDBInstance(t *testing.T) {
 		conn = rotated
 	})
 
+	// A rotation that lands while an automated backup holds the engine. The window
+	// is opened on purpose, so this runs every time rather than only when one
+	// happens to open mid-test.
+	t.Run("TheMasterPasswordRotatesDuringAnAutomatedBackup", func(t *testing.T) {
+		rotateDuringAutomatedBackup(t, f, client, id, &conn)
+	})
+
 	// The family check is one funnel every binding path goes through, and modify
 	// is the one a customer reaches by hand. Non-disruptive: it is refused before
 	// anything is written, so the instance is left where it was found.

tests/e2e/single/vpcegress_test.go

@@ -735,6 +735,8 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 					LogicalIP:   pubPrivIP,
 					ArtifactDir: fix.ArtifactDir(t),
 				})
+			// A restarted guest that never reaches IMDS stalls in cloud-init before sshd starts.
+			harness.DumpIMDSDatapathDiagnostics(t, pubSubnetID, pubPrivIP, fix.ArtifactDir(t))
 			t.Fatalf("guest unreachable via %s within %s after restart (see diagnostics above)", newIP, harness.SSHReadyBudget)
 		}
 		out := harness.RunSSH(t, harness.SSHTarget{User: "ubuntu", Host: newIP, Port: 22, KeyPath: keyPath}, "id")
@@ -859,6 +861,8 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 						LogicalIP:   aws.StringValue(started.PrivateIpAddress),
 						ArtifactDir: fix.ArtifactDir(t),
 					})
+				// A restarted guest that never reaches IMDS stalls in cloud-init before sshd starts.
+				harness.DumpIMDSDatapathDiagnostics(t, pubSubnetID, aws.StringValue(started.PrivateIpAddress), fix.ArtifactDir(t))
 				t.Fatalf("guest unreachable via EIP %s within %s after restart (see diagnostics above)", eipIP, harness.SSHReadyBudget)
 			}
 			out := harness.RunSSH(t, harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}, "id")