test(e2e): dump the IMDS datapath when a restarted guest never answers SSH

A stop/start that leaves the guest stuck in cloud-init-local waiting on
IMDS fails the restart SSH probe, but the diagnostics only captured the
EIP/OVN side, which was healthy. Capture br-imds flows, the endpoint's
link, addresses, sysctls, reply rule and route, the responder listener
and conntrack too, so the next occurrence names the broken hop.

spinifex/agents/rds/agent/agent_test.go

@@ -763,11 +763,10 @@ 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,
-			RemainingBudgetMs: 29500,
+			CommandID:  "cmd-1",
+			Type:       "reload-parameters",
+			Parameters: []rdsinstance.Parameter{{Name: "work_mem", Value: "8MB"}},
+			IssuedAt:   &issued,
 		}},
 	})
 	body, err := awsxml.Marshal(payload)
@@ -803,9 +802,6 @@ 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,13 +2,10 @@ package agent
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log/slog"
-	"slices"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	rdsinstance "github.com/mulgadc/spinifex/spinifex/domains/rds/instance"
@@ -93,96 +90,35 @@ 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 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.
+// whatever the previous poll delivered.
 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,
-		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
+	return &commander{cp: cp, registry: registry, wait: wait}
 }
 
 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, sent, err := c.poll(ctx)
+		commands, err := c.cp.PollCommands(ctx, c.id, c.pending, c.wait)
 		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():
@@ -191,122 +127,14 @@ func (c *commander) Run(ctx context.Context) {
 			}
 			continue
 		}
-		c.mu.Lock()
-		c.pending = c.pending[sent:]
-		c.mu.Unlock()
+		c.pending = nil
 
 		for _, cmd := range commands {
-			c.dispatch(ctx, cmd)
+			c.pending = append(c.pending, c.execute(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,9 +3,6 @@ package agent
 import (
 	"context"
 	"errors"
-	"fmt"
-	"slices"
-	"sync"
 	"testing"
 	"time"
 
@@ -73,9 +70,9 @@ func TestCommander_UnknownTypeRepliesFailedNotDropped(t *testing.T) {
 	}
 }
 
-// The reply for work the guest actually did rides a later poll. A poll that
+// The reply for work the guest actually did rides the next poll. A poll that
 // fails must not consume it, or the issuer never learns the outcome.
-func TestCommander_RunCarriesRepliesOnALaterPoll(t *testing.T) {
+func TestCommander_RunCarriesRepliesOnTheNextPoll(t *testing.T) {
 	cp := newFakeControlPlane()
 	cp.pollQueue = [][]rdsinstance.Command{
 		{{CommandID: "cmd-1", Type: "reload-parameters"}},
@@ -90,15 +87,19 @@ func TestCommander_RunCarriesRepliesOnALaterPoll(t *testing.T) {
 	done := make(chan struct{})
 	go func() { c.Run(ctx); close(done) }()
 
-	waitFor(t, func() bool { return len(repliesFor(cp, "cmd-1")) > 0 }, "the reply to ride a later poll")
+	waitFor(t, func() bool {
+		replies := cp.snapshotReplies()
+		return len(replies) >= 2 && len(replies[1]) == 1
+	}, "the reply to ride the second poll")
 	cancel()
 	<-done
 
-	if replies := cp.snapshotReplies(); len(replies[0]) != 0 {
+	replies := cp.snapshotReplies()
+	if len(replies[0]) != 0 {
 		t.Errorf("first poll carried replies %+v, want none", replies[0])
 	}
-	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)
+	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)
 	}
 }
 
@@ -140,19 +141,16 @@ func TestCommander_RetainsAndRetriesTheCompletePendingBatch(t *testing.T) {
 	done = make(chan struct{})
 	go func() { c.Run(ctx); close(done) }()
 	waitFor(t, func() bool {
-		return len(repliesFor(cp, "cmd-3")) > 0
-	}, "the retained batch to succeed and the new reply to reach a later poll")
+		return len(cp.snapshotReplies()) >= 3
+	}, "the retained batch to succeed and the new reply to reach the following poll")
 	cancel()
 	<-done
 
 	replies = cp.snapshotReplies()
 	assert.Equal(t, pending, replies[1], "the complete failed batch must be retried in order")
-	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")
-		}
-	}
+	require.Len(t, replies[2], 1)
+	assert.Equal(t, "cmd-3", replies[2][0].CommandID)
+	assert.Equal(t, rdsinstance.CommandStatusSucceeded, replies[2][0].Status)
 }
 
 // A poll error must not spin the loop: without a backoff a gateway that is down
@@ -173,340 +171,3 @@ 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,8 +94,6 @@ 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)
@@ -153,7 +151,6 @@ 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{
@@ -383,11 +380,6 @@ 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.
@@ -403,34 +395,16 @@ 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)
-	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
+	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 max(1, int(remaining.Seconds()))
+	return nil
 }
 
 // Installs the resolved set and applies the half a running server can adopt.
@@ -740,9 +714,7 @@ 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.
-	if err := e.lockWithin(ctx); err != nil {
-		return err
-	}
+	e.mu.Lock()
 	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,123 +1151,3 @@ 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,9 +516,7 @@ 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.
-	if err := e.lockWithin(ctx); err != nil {
-		return err
-	}
+	e.mu.Lock()
 	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,48 +114,6 @@ 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,9 +201,8 @@ type PollDBCommandsInput struct {
 	Replies              []rdsinstance.CommandReply `locationName:"Replies" locationNameList:"member"`
 }
 
-// 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.
+// 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.
 type PollDBCommandsOutput struct {
 	Commands []rdsinstance.Command `locationName:"Commands" locationNameList:"member"`
 }

spinifex/domains/rds/instance/agent.go

@@ -137,10 +137,6 @@ 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,24 +75,6 @@ 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.
@@ -170,6 +152,10 @@ 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.
@@ -187,15 +173,7 @@ 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,7 +2,6 @@ package instance
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -81,26 +80,3 @@ 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

@@ -1,258 +0,0 @@
-//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,29 +302,17 @@ 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),
-		PreferredBackupWindow:      aws.String(backup),
-		PreferredMaintenanceWindow: aws.String(maintenance),
+		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),
 	}
 }
 
-// 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,8 +31,6 @@ 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.
@@ -66,17 +64,14 @@ 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),
-		PreferredBackupWindow:      aws.String(backup),
-		PreferredMaintenanceWindow: aws.String(maintenance),
+		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),
 	}
 }
 
@@ -273,13 +268,6 @@ 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")