fix(rds): cap an agent command by the budget its issuer has left

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.

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/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.