test: speed up rds agent tests (#1618)

spinifex/agents/rds/agent/agent_test.go

@@ -459,6 +459,7 @@ func TestAcknowledgeBootstrap_WaitsForTheReceiptThenRetiresIt(t *testing.T) {
 // instance's volume carries the source instance's receipt. Acknowledging on one
 // would destroy the ciphertext this instance still needs.
 func TestAcknowledgeBootstrap_IgnoresAForeignReceipt(t *testing.T) {
+	t.Parallel()
 	tests := map[string]struct{ payloadID, dbID string }{
 		"another payload":  {payloadID: "bp-somebody-else", dbID: "db-resolved"},
 		"another instance": {payloadID: testPayloadID, dbID: "db-the-snapshot-source"},
@@ -466,6 +467,7 @@ func TestAcknowledgeBootstrap_IgnoresAForeignReceipt(t *testing.T) {
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
+			t.Parallel()
 			cp := newFakeControlPlane()
 			cfg := testConfig(t)
 			cfg.DataMount = t.TempDir()
@@ -568,6 +570,7 @@ func TestRun_SendsConfiguredIdentifier(t *testing.T) {
 }
 
 func TestRun_WaitsForDataMountBeforePollingCommands(t *testing.T) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cp.bootstrapOut = bootstrapOutput(nil)
 	cfg := testConfig(t)
@@ -854,6 +857,7 @@ func TestAcknowledgeBootstrap_StopsOnATerminalDenial(t *testing.T) {
 // window. Reporting that would put a failure on every healthy create's beat, and
 // then into the reason a create that timed out for some other cause records.
 func TestAcknowledgeBootstrap_DoesNotReportAnAbsentReceiptAsAFailure(t *testing.T) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cfg := testConfig(t)
 	cfg.DataMount = t.TempDir()

spinifex/agents/rds/agent/command.go

@@ -126,6 +126,8 @@ type commander struct {
 	id       identity
 	registry commandRegistry
 	wait     time.Duration
+	// The wait after a failed poll, pollErrorBackoff unless a test shortens it.
+	pollBackoff 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.
@@ -152,12 +154,13 @@ func newCommander(cp controlPlane, registry commandRegistry, wait time.Duration)
 		wait = defaultPollWait
 	}
 	return &commander{
-		cp:       cp,
-		registry: registry,
-		wait:     wait,
-		budget:   controlPlaneBudget,
-		serial:   make(chan queuedCommand, serialQueueDepth),
-		inFlight: make(map[string]struct{}),
+		cp:          cp,
+		registry:    registry,
+		wait:        wait,
+		pollBackoff: pollErrorBackoff,
+		budget:      controlPlaneBudget,
+		serial:      make(chan queuedCommand, serialQueueDepth),
+		inFlight:    make(map[string]struct{}),
 	}
 }
 
@@ -187,7 +190,7 @@ func (c *commander) Run(ctx context.Context) {
 			select {
 			case <-ctx.Done():
 				return
-			case <-time.After(pollErrorBackoff):
+			case <-time.After(c.pollBackoff):
 			}
 			continue
 		}

spinifex/agents/rds/agent/command_test.go

@@ -158,19 +158,30 @@ func TestCommander_RetainsAndRetriesTheCompletePendingBatch(t *testing.T) {
 // A poll error must not spin the loop: without a backoff a gateway that is down
 // would be re-polled at line rate for as long as it stayed down.
 func TestCommander_BacksOffAfterAFailedPoll(t *testing.T) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cp.pollErr = errors.New("gateway unreachable")
 
 	c := newCommander(cp, newCommandRegistry(&fakeEngine{}, &fakeStorage{}), 10*time.Millisecond)
-	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	// Outlasts the test, so any second poll inside the window is a missing backoff.
+	c.pollBackoff = time.Hour
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	done := make(chan struct{})
 	go func() { c.Run(ctx); close(done) }()
+
+	select {
+	case <-cp.polled:
+	case <-ctx.Done():
+		t.Fatal("the commander never polled")
+	}
+	// Without the backoff the loop re-polls within microseconds of the failure.
+	time.Sleep(50 * time.Millisecond)
+	cancel()
 	<-done
 
-	// pollErrorBackoff is 5s, so the window admits the first poll and no more.
-	if polls := len(cp.snapshotReplies()); polls > 1 {
-		t.Errorf("polled %d times in 200ms against a broken channel, want 1", polls)
+	if polls := len(cp.snapshotReplies()); polls != 1 {
+		t.Errorf("polled %d times against a broken channel inside the backoff, want 1", polls)
 	}
 }
 
@@ -306,6 +317,7 @@ func TestCommander_ACopyArrivingBeforeTheReplyIsCarriedDoesNotRunAgain(t *testin
 // 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) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cp.pollQueue = [][]rdsinstance.Command{
 		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},
@@ -351,6 +363,7 @@ func TestCommander_AStaleQuiesceNeverTakesAHold(t *testing.T) {
 // 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) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cp.pollQueue = [][]rdsinstance.Command{
 		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},
@@ -414,6 +427,7 @@ func (c *commander) pendingReplies() []rdsinstance.CommandReply {
 // 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) {
+	t.Parallel()
 	cp := newFakeControlPlane()
 	cp.pollQueue = [][]rdsinstance.Command{
 		{{CommandID: "cmd-apply", Type: rdsinstance.CommandApplyParams}},

spinifex/agents/rds/agent/config_test.go

@@ -16,8 +16,16 @@ func testLoadConfig(t *testing.T, engine string) Config {
 	if err := os.WriteFile(stamp, []byte(engine+"\n"), 0o444); err != nil {
 		t.Fatalf("write the engine stamp: %v", err)
 	}
-	t.Setenv("RDS_ENGINE_FILE", stamp)
-	cfg := LoadConfig(filepath.Join(t.TempDir(), "absent.env"))
+	// Delivered through the env file rather than t.Setenv, so callers can run in
+	// parallel.
+	envFile := filepath.Join(t.TempDir(), "agent.env")
+	if err := os.WriteFile(envFile, []byte("RDS_ENGINE_FILE="+stamp+"\n"), 0o600); err != nil {
+		t.Fatalf("write the env file: %v", err)
+	}
+	cfg := LoadConfig(envFile)
+	if cfg.EngineFile != stamp {
+		t.Fatalf("engine file = %q, want the stamp %q; is RDS_ENGINE_FILE set in the environment?", cfg.EngineFile, stamp)
+	}
 	cfg.EngineCatalog = testEngineCatalogLookup
 	return cfg
 }

spinifex/agents/rds/agent/engine_mariadb_test.go

@@ -883,6 +883,7 @@ func TestMariaDBProbe_SeparatesRecoveringFromAbsent(t *testing.T) {
 // mariadbd will not accept makes it exit during startup, so the pidfile names a
 // process that is gone and the deadline runs.
 func TestMariaDBProbe_BoundsAStalledClient(t *testing.T) {
+	t.Parallel()
 	cfg := mariadbTestProbeConfig(t)
 	cfg.EngineProbeTimeout = 20 * time.Millisecond
 	writePid(t, cfg.EnginePidFile, 4242)
@@ -946,6 +947,7 @@ func TestMariaDBProbe_MapsToTheHealthTheControlPlaneReads(t *testing.T) {
 // the server mid-recovery — a rollback that exists to break a boot loop would
 // create one, on an instance whose parameters were never at fault.
 func TestMariaDBProbe_LiveButUnreachableEngineResetsTheRollbackDeadline(t *testing.T) {
+	t.Parallel()
 	cfg := mariadbTestProbeConfig(t)
 	writePid(t, cfg.EnginePidFile, 4242)
 	probe := newEngineProbe(cfg.EnginePort, mariadbProbeState(cfg,
@@ -1152,12 +1154,18 @@ func TestMariaDBEngine_EveryResolvedDynamicValueRenders(t *testing.T) {
 	}
 }
 
+// Far below the production interval, so a hold test can wait out several polls
+// cheaply.
+const testHoldPoll = 5 * time.Millisecond
+
 // 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) {
+	t.Parallel()
 	runner := &recordingRunner{}
 	engine := newTestMariaDBEngine(t, runner.run)
+	engine.holdPoll = testHoldPoll
 	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)
@@ -1168,7 +1176,7 @@ func TestMariaDBEngine_SetPasswordWaitsForTheBackupHoldToEnd(t *testing.T) {
 	select {
 	case err := <-rotated:
 		t.Fatalf("SetPassword returned %v while the engine was held for a backup", err)
-	case <-time.After(3 * holdPollInterval):
+	case <-time.After(10 * testHoldPoll):
 	}
 
 	if err := engine.Unquiesce(context.Background()); err != nil {
@@ -1190,9 +1198,11 @@ func TestMariaDBEngine_SetPasswordWaitsForTheBackupHoldToEnd(t *testing.T) {
 // 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) {
+	t.Parallel()
 	runner := &recordingRunner{}
 	engine := newTestMariaDBEngine(t, runner.run)
-	engine.holdWait = 3 * holdPollInterval
+	engine.holdPoll = testHoldPoll
+	engine.holdWait = 3 * testHoldPoll
 	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)
@@ -1246,7 +1256,9 @@ func TestMariaDBEngine_SetPasswordBoundsTheStatementByTheRemainingBudget(t *test
 // 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) {
+	t.Parallel()
 	engine := newTestMariaDBEngine(t, (&recordingRunner{}).run)
+	engine.holdPoll = testHoldPoll
 	started := 0
 	engine.startSess = func(context.Context, command) (engineSession, error) {
 		started++
@@ -1255,7 +1267,7 @@ func TestMariaDBEngine_QuiesceGivesUpWaitingForTheBookkeepingWithItsContext(t *t
 	engine.mu.Lock()
 	defer engine.mu.Unlock()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 3*holdPollInterval)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*testHoldPoll)
 	defer cancel()
 	done := make(chan error, 1)
 	go func() { done <- engine.Quiesce(ctx, "orders-db-automated", time.Minute) }()

spinifex/agents/rds/agent/health_test.go

@@ -83,6 +83,7 @@ func TestEngineProbe_ProbesTheAssignedPort(t *testing.T) {
 }
 
 func TestExecProbeRunner_PreservesTheProbeDeadline(t *testing.T) {
+	t.Parallel()
 	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
 	defer cancel()
 

spinifex/agents/rds/agent/paramrecovery_test.go

@@ -385,6 +385,7 @@ func newTestGuard(engine parameterRecovery, probe *engineProbe) *paramGuard {
 // pg_isready exit 2 is an engine that is not there at all, which after a
 // parameter change is the boot-loop shape this breaks.
 func TestParamGuard_RollsBackReportsAndRestartsWhenTheEngineNeverComesUp(t *testing.T) {
+	t.Parallel()
 	engine := &fakeRecovery{restored: true}
 	cp := newFakeControlPlane()
 	guard := newParamGuard(engine, stubProbe(t, 2, nil), cp)
@@ -408,6 +409,7 @@ func TestParamGuard_RollsBackReportsAndRestartsWhenTheEngineNeverComesUp(t *test
 // the file can be replaced and the call still fail. The rollback has happened
 // either way, and the worse outcome is an instance left down and unreported.
 func TestParamGuard_ReportsAndRestartsWhenOnlyTheEnforcementFailed(t *testing.T) {
+	t.Parallel()
 	engine := &fakeRecovery{restored: true, restoreErr: errors.New("write the TLS enforcement rule: read-only file system")}
 	cp := newFakeControlPlane()
 	guard := newParamGuard(engine, stubProbe(t, 2, nil), cp)
@@ -431,6 +433,7 @@ func TestParamGuard_ReportsAndRestartsWhenOnlyTheEnforcementFailed(t *testing.T)
 // the instance on the set it is already failing under, so restarting it would
 // only churn a cluster that is down for a reason the rollback cannot fix.
 func TestParamGuard_DoesNotRestartWhenTheRestoreReplacedNothing(t *testing.T) {
+	t.Parallel()
 	engine := &fakeRecovery{restored: false, restoreErr: errors.New("read the last known good parameters: input/output error")}
 	cp := newFakeControlPlane()
 	guard := newParamGuard(engine, stubProbe(t, 2, nil), cp)
@@ -457,6 +460,7 @@ func TestParamGuard_DoesNothingWhenTheEngineIsServing(t *testing.T) {
 // pg_isready exit 1 is a postmaster that is up and replaying WAL. It is making
 // progress on its own, and restarting it would throw the recovery away.
 func TestParamGuard_LeavesARecoveringEngineAlone(t *testing.T) {
+	t.Parallel()
 	engine := &fakeRecovery{restored: true}
 	guard := newTestGuard(engine, stubProbe(t, 1, nil))
 	// Bounded so the deadline-resetting loop cannot run forever if it stops
@@ -473,6 +477,7 @@ func TestParamGuard_LeavesARecoveringEngineAlone(t *testing.T) {
 // The parameters are only the cause if they differ from the set the engine last
 // accepted; restarting on an identical set would churn for nothing.
 func TestParamGuard_DoesNotRestartWhenTheParametersAreAlreadyTheAcceptedSet(t *testing.T) {
+	t.Parallel()
 	engine := &fakeRecovery{restored: false}
 	newTestGuard(engine, stubProbe(t, 2, nil)).Run(t.Context())
 

spinifex/agents/rds/agent/quiesce.go

@@ -50,6 +50,8 @@ type quiesceState struct {
 	// Guarded because the expiry timer releases the hold on its own goroutine.
 	mu   sync.Mutex
 	held *quiesceHold
+	// How often a waiter looks again; zero means defaultHoldPollInterval.
+	holdPoll time.Duration
 }
 
 // The two things every quiesce needs, checked before a session is opened so a
@@ -121,14 +123,21 @@ func (q *quiesceState) lockWithin(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return fmt.Errorf("wait for the backup bookkeeping: %w", ctx.Err())
-		case <-time.After(holdPollInterval):
+		case <-time.After(q.holdPollInterval()):
 		}
 	}
 	return nil
 }
 
 // How often a statement waiting for a backup hold to end looks again.
-const holdPollInterval = 100 * time.Millisecond
+const defaultHoldPollInterval = 100 * time.Millisecond
+
+func (q *quiesceState) holdPollInterval() time.Duration {
+	if q.holdPoll > 0 {
+		return q.holdPoll
+	}
+	return defaultHoldPollInterval
+}
 
 // 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
@@ -151,7 +160,7 @@ func (q *quiesceState) outsideHold(ctx context.Context, wait time.Duration, fn f
 			return ctx.Err()
 		case <-deadline.C:
 			return fmt.Errorf("the engine is still held for a backup after %s", wait)
-		case <-time.After(holdPollInterval):
+		case <-time.After(q.holdPollInterval()):
 		}
 	}
 }