From 141d697e40fa0d412b8bfbea2cccd6da5350db4a Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:47:40 +1100 Subject: [PATCH] test: speed up rds agent tests (#1618) --- spinifex/agents/rds/agent/agent_test.go | 4 ++++ spinifex/agents/rds/agent/command.go | 17 ++++++++------ spinifex/agents/rds/agent/command_test.go | 22 +++++++++++++++---- spinifex/agents/rds/agent/config_test.go | 12 ++++++++-- .../agents/rds/agent/engine_mariadb_test.go | 18 ++++++++++++--- spinifex/agents/rds/agent/health_test.go | 1 + .../agents/rds/agent/paramrecovery_test.go | 5 +++++ spinifex/agents/rds/agent/quiesce.go | 15 ++++++++++--- 8 files changed, 75 insertions(+), 19 deletions(-) diff --git a/spinifex/agents/rds/agent/agent_test.go b/spinifex/agents/rds/agent/agent_test.go index caed1993e..ca6a7f27b 100644 --- a/spinifex/agents/rds/agent/agent_test.go +++ b/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() diff --git a/spinifex/agents/rds/agent/command.go b/spinifex/agents/rds/agent/command.go index 8288ca060..1528153d4 100644 --- a/spinifex/agents/rds/agent/command.go +++ b/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 } diff --git a/spinifex/agents/rds/agent/command_test.go b/spinifex/agents/rds/agent/command_test.go index b9d0d7217..14ef29be2 100644 --- a/spinifex/agents/rds/agent/command_test.go +++ b/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}}, diff --git a/spinifex/agents/rds/agent/config_test.go b/spinifex/agents/rds/agent/config_test.go index 62bec708a..3a563cd9b 100644 --- a/spinifex/agents/rds/agent/config_test.go +++ b/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 } diff --git a/spinifex/agents/rds/agent/engine_mariadb_test.go b/spinifex/agents/rds/agent/engine_mariadb_test.go index 46b74f1e7..a8f02652a 100644 --- a/spinifex/agents/rds/agent/engine_mariadb_test.go +++ b/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) }() diff --git a/spinifex/agents/rds/agent/health_test.go b/spinifex/agents/rds/agent/health_test.go index 482cf538b..dca619aaf 100644 --- a/spinifex/agents/rds/agent/health_test.go +++ b/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() diff --git a/spinifex/agents/rds/agent/paramrecovery_test.go b/spinifex/agents/rds/agent/paramrecovery_test.go index e7fb85a7c..f2634c568 100644 --- a/spinifex/agents/rds/agent/paramrecovery_test.go +++ b/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()) diff --git a/spinifex/agents/rds/agent/quiesce.go b/spinifex/agents/rds/agent/quiesce.go index 1d1720a27..ecbf01d35 100644 --- a/spinifex/agents/rds/agent/quiesce.go +++ b/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()): } } }