diff --git a/internal/scheduler/retry_policy_test.go b/internal/scheduler/retry_policy_test.go
index 73c0bffd1f9..01b6b1746aa 100644
--- a/internal/scheduler/retry_policy_test.go
+++ b/internal/scheduler/retry_policy_test.go
@@ -108,6 +108,7 @@ func mkPolicy(t *testing.T, limit uint32, defaultAction api.RetryAction, rules .
 // leased/running run added last so it becomes the job's LatestRun.
 type jobRunOpts struct {
 	schedulingInfo   *schedulerobjects.JobSchedulingInfo
+	queue            string
 	leased           bool
 	running          bool
 	preemptRequested bool
@@ -122,8 +123,12 @@ type jobRunOpts struct {
 func makeRetryJob(t *testing.T, sched *Scheduler, opts jobRunOpts) *jobdb.Job {
 	t.Helper()
 	jobId := util.NewULID()
+	queue := opts.queue
+	if queue == "" {
+		queue = "testQueue"
+	}
 	job := testfixtures.NewJob(
-		jobId, "testJobset", "testQueue", uint32(10),
+		jobId, "testJobset", queue, uint32(10),
 		toInternalSchedulingInfo(opts.schedulingInfo),
 		false, 1, false, false, false, 1, true,
 	)
@@ -155,33 +160,36 @@ func makeRetryJob(t *testing.T, sched *Scheduler, opts jobRunOpts) *jobdb.Job {
 	return job
 }
 
-// runFailurePath drives generateUpdateMessagesFromJob for a single job through
-// the standard {"testQueue":"test-policy"} mapping and returns the emitted
-// events plus the open write txn (the caller must defer txn.Abort()).
+// runFailurePath drives generateUpdateMessages for a single job through the
+// standard {"testQueue":"test-policy"} mapping. It returns the emitted events
+// and the open write txn, which it aborts in t.Cleanup. It calls the
+// batch entry point, so the tests cover the planning pass and its probes.
 func runFailurePath(t *testing.T, sched *Scheduler, job *jobdb.Job, runErr *armadaevents.Error) (*armadaevents.EventSequence, *jobdb.Txn) {
 	t.Helper()
 	txn := sched.jobDb.WriteTxn()
+	t.Cleanup(txn.Abort)
 	require.NoError(t, txn.Upsert([]*jobdb.Job{job}))
 	jobErrors := map[string]*armadaevents.Error{job.LatestRun().Id(): runErr}
 	queueRetryPolicies := map[string]string{"testQueue": "test-policy"}
-	events, err := sched.generateUpdateMessagesFromJob(armadacontext.Background(), job, jobErrors, queueRetryPolicies, txn)
+	eventSequences, err := sched.generateUpdateMessages(armadacontext.Background(), txn, []*jobdb.Job{job}, jobErrors, queueRetryPolicies)
 	require.NoError(t, err)
-	require.NotNil(t, events)
-	return events, txn
+	require.Len(t, eventSequences, 1)
+	return eventSequences[0], txn
 }
 
 // runLeaseExpiryPath drives expireJobsIfNecessary for a single job whose
 // executor stopped heartbeating well past the 1h executorTimeout, and returns
-// the emitted event sequences plus the open write txn (the caller must defer
-// txn.Abort()).
+// the emitted event sequences plus the open write txn, which it aborts in
+// t.Cleanup.
 func runLeaseExpiryPath(t *testing.T, sched *Scheduler, job *jobdb.Job) ([]*armadaevents.EventSequence, *jobdb.Txn) {
 	t.Helper()
 	sched.executorRepository = &testExecutorRepository{
 		updateTimes: map[string]time.Time{"testExecutor": sched.clock.Now().Add(-2 * time.Hour)},
 	}
 	txn := sched.jobDb.WriteTxn()
+	t.Cleanup(txn.Abort)
 	require.NoError(t, txn.Upsert([]*jobdb.Job{job}))
-	eventSequences, err := sched.expireJobsIfNecessary(armadacontext.Background(), txn)
+	eventSequences, err := sched.expireJobsIfNecessary(armadacontext.Background(), txn, map[string]string{"testQueue": "test-policy"})
 	require.NoError(t, err)
 	return eventSequences, txn
 }
@@ -295,8 +303,7 @@ func TestRetryPolicy_FFOn_RetryDecision(t *testing.T) {
 	sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
 	job := makeFailedJobForRetry(t, sched)
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.True(t, hasRequeued(events.Events), "FF on with matching retry rule must emit JobRequeued")
 	assert.True(t, hasJobErrors(events.Events), "FF on with retry decision must emit a non-terminal JobErrors so the api event stream surfaces the retry")
@@ -319,8 +326,7 @@ func TestRetryPolicy_FFOn_PolicyLimitCapsRetries(t *testing.T) {
 	job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: schedulingInfo, failedRuns: 3})
 	require.Equal(t, uint32(3), job.FailureCount())
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.False(t, hasRequeued(events.Events), "engine at retry limit must not emit JobRequeued")
 	assert.Contains(t, terminalError(events.Events).GetMaxRunsExceeded().GetMessage(), "Retry policy:",
@@ -339,8 +345,7 @@ func TestRetryPolicy_FFOn_EngineRetryOverridesMaxAttemptedRuns(t *testing.T) {
 	job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: schedulingInfo, failedRuns: 3, runAttempted: true})
 	require.Greater(t, int(job.NumAttempts()), int(maxNumberOfAttempts))
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.True(t, hasRequeued(events.Events), "a decided engine retry must override the legacy attempt cap")
 }
@@ -364,8 +369,7 @@ func TestRetryPolicy_FFOn_TerminalFailPreservesOriginalError(t *testing.T) {
 			},
 		},
 	}
-	events, txn := runFailurePath(t, sched, job, runError)
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, runError)
 
 	terminal := terminalError(events.Events)
 	require.NotNil(t, terminal, "policy Fail must emit a terminal JobErrors event")
@@ -385,8 +389,7 @@ func TestRetryPolicy_FFOn_MissingPolicyFallsThrough(t *testing.T) {
 	sched := makeRetryTestScheduler(t, true, fakePolicyCache{}) // empty cache
 
 	job := makeFailedJobForRetry(t, sched)
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.True(t, hasJobErrors(events.Events), "missing policy must not crash; falls back to legacy terminal-failure path")
 	assert.False(t, hasRequeued(events.Events), "missing policy must not requeue; the legacy path terminally fails the job")
@@ -455,8 +458,7 @@ func TestRetryPolicy_FFOn_GlobalCapExcludesPreemptions(t *testing.T) {
 	require.Equal(t, uint32(1), job.FailureCount(), "fixture must have exactly one failed run")
 	require.Equal(t, 4, len(job.AllRuns()), "fixture must have four total runs")
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.True(t, hasRequeued(events.Events),
 		"preemptions must not consume the global cap: 3 preemptions + 1 failure with a cap of 2 must still retry")
@@ -472,7 +474,6 @@ func TestRetryPolicy_FFOn_GlobalMaxZeroDisablesRetries(t *testing.T) {
 	job := makeFailedJobForRetry(t, sched)
 
 	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
 
 	assert.False(t, hasRequeued(events.Events), "GlobalMaxRetries 0 must never retry")
 	updated := txn.GetById(job.Id())
@@ -499,7 +500,6 @@ func TestRetryPolicy_FFOn_LeaseExpiryRetriesWhenPolicyMatches(t *testing.T) {
 	job := makeRunningJobOnExecutor(t, sched)
 
 	eventSequences, txn := runLeaseExpiryPath(t, sched, job)
-	defer txn.Abort()
 	require.Len(t, eventSequences, 1)
 
 	evs := eventSequences[0].Events
@@ -551,7 +551,6 @@ func TestRetryPolicy_FFOn_LeaseExpiryTerminalWhenNoMatch(t *testing.T) {
 	job := makeRunningJobOnExecutor(t, sched)
 
 	eventSequences, txn := runLeaseExpiryPath(t, sched, job)
-	defer txn.Abort()
 	require.Len(t, eventSequences, 1)
 
 	// The policy was consulted and declined to retry, so the terminal event is a
@@ -642,8 +641,7 @@ func TestRetryPolicy_FFOn_MemoryBumpGrowsRequeuedJob(t *testing.T) {
 			sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
 			job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), failedRuns: 1, runAttempted: true})
 
-			events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-			defer txn.Abort()
+			events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 			si := requeuedSchedulingInfo(t, events.Events)
 			expected := resource.MustParse(tc.expectedMemory)
@@ -658,6 +656,151 @@ func TestRetryPolicy_FFOn_MemoryBumpGrowsRequeuedJob(t *testing.T) {
 	}
 }
 
+func TestRetryPolicy_FFOn_ProbesRunAsRoundBatches(t *testing.T) {
+	tests := map[string]struct {
+		mutate             *api.RetryMutation
+		checkSuccess       bool
+		expectedCheckCalls int
+		expectRequeued     bool
+	}{
+		"memory bump probes once for the whole round": {
+			mutate:             &api.RetryMutation{Resources: &api.RetryResourceMutation{Memory: &api.RetryResourceBump{Factor: 1.5}}},
+			checkSuccess:       true,
+			expectedCheckCalls: 1,
+			expectRequeued:     true,
+		},
+		"memory bump and avoidSameNode probe twice for the whole round": {
+			mutate: &api.RetryMutation{
+				Resources: &api.RetryResourceMutation{Memory: &api.RetryResourceBump{Factor: 1.5}},
+				Affinity:  &api.RetryAffinityMutation{AvoidSameNode: true},
+			},
+			checkSuccess:       true,
+			expectedCheckCalls: 2,
+			expectRequeued:     true,
+		},
+		"an unschedulable verdict reaches every job in the class": {
+			mutate:             &api.RetryMutation{Resources: &api.RetryResourceMutation{Memory: &api.RetryResourceBump{Factor: 1.5}}},
+			checkSuccess:       false,
+			expectedCheckCalls: 1,
+			expectRequeued:     false,
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			policy := mkPolicy(t, 3, api.RetryAction_RETRY_ACTION_FAIL, &api.RetryRule{
+				Action:     api.RetryAction_RETRY_ACTION_RETRY,
+				OnCategory: "app-error",
+				Mutate:     tc.mutate,
+			})
+			sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
+			checker := &testSubmitChecker{checkSuccess: tc.checkSuccess}
+			sched.submitChecker = checker
+
+			txn := sched.jobDb.WriteTxn()
+			t.Cleanup(txn.Abort)
+			jobs := make([]*jobdb.Job, 3)
+			jobErrors := map[string]*armadaevents.Error{}
+			for i := range jobs {
+				jobs[i] = makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), failedRuns: 1, runAttempted: true})
+				jobErrors[jobs[i].LatestRun().Id()] = categorizedError("app-error")
+			}
+			require.NoError(t, txn.Upsert(jobs))
+
+			eventSequences, err := sched.generateUpdateMessages(armadacontext.Background(), txn, jobs, jobErrors, map[string]string{"testQueue": "test-policy"})
+			require.NoError(t, err)
+			require.Len(t, eventSequences, 3)
+			for _, es := range eventSequences {
+				assert.Equal(t, tc.expectRequeued, hasRequeued(es.Events), "the class verdict must reach every job")
+			}
+			assert.Equal(t, tc.expectedCheckCalls, checker.checkCalls, "probes must run per round, not per job")
+			for _, jobCount := range checker.checkJobCounts {
+				assert.Equal(t, 1, jobCount, "same-shaped candidates must collapse to one probed representative")
+			}
+		})
+	}
+}
+
+func TestRetryPolicy_FFOn_ProbeSplitsClassesByQueue(t *testing.T) {
+	// The checker applies a per-queue resource limit, so same-shaped jobs in
+	// different queues can get different verdicts and must probe separately.
+	policy := mkPolicy(t, 3, api.RetryAction_RETRY_ACTION_FAIL, &api.RetryRule{
+		Action:     api.RetryAction_RETRY_ACTION_RETRY,
+		OnCategory: "app-error",
+		Mutate:     &api.RetryMutation{Resources: &api.RetryResourceMutation{Memory: &api.RetryResourceBump{Factor: 1.5}}},
+	})
+	sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
+	checker := &testSubmitChecker{checkSuccess: true}
+	sched.submitChecker = checker
+
+	txn := sched.jobDb.WriteTxn()
+	t.Cleanup(txn.Abort)
+	jobs := []*jobdb.Job{
+		makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), failedRuns: 1, runAttempted: true}),
+		makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), queue: "otherQueue", failedRuns: 1, runAttempted: true}),
+	}
+	jobErrors := map[string]*armadaevents.Error{}
+	for _, job := range jobs {
+		jobErrors[job.LatestRun().Id()] = categorizedError("app-error")
+	}
+	require.NoError(t, txn.Upsert(jobs))
+
+	queueRetryPolicies := map[string]string{"testQueue": "test-policy", "otherQueue": "test-policy"}
+	eventSequences, err := sched.generateUpdateMessages(armadacontext.Background(), txn, jobs, jobErrors, queueRetryPolicies)
+	require.NoError(t, err)
+	require.Len(t, eventSequences, 2)
+	for _, es := range eventSequences {
+		assert.True(t, hasRequeued(es.Events))
+	}
+	require.Len(t, checker.checkJobCounts, 1)
+	assert.Equal(t, 2, checker.checkJobCounts[0], "same-shaped jobs in different queues must probe separately")
+}
+
+func TestRetryPolicy_FFOn_ProbeAsksAgainForUnprobedJob(t *testing.T) {
+	tests := map[string]struct {
+		skippedChecks      int
+		expectedCheckCalls int
+		expectRequeued     bool
+	}{
+		"a later call answers and its verdict applies": {
+			skippedChecks:      1,
+			expectedCheckCalls: 2,
+			expectRequeued:     false,
+		},
+		"a job unprobed after every call keeps its retry": {
+			skippedChecks:      3,
+			expectedCheckCalls: 3,
+			expectRequeued:     true,
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			policy := mkPolicy(t, 3, api.RetryAction_RETRY_ACTION_FAIL, &api.RetryRule{
+				Action:     api.RetryAction_RETRY_ACTION_RETRY,
+				OnCategory: "app-error",
+				Mutate:     &api.RetryMutation{Resources: &api.RetryResourceMutation{Memory: &api.RetryResourceBump{Factor: 1.5}}},
+			})
+			sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
+			job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), failedRuns: 1, runAttempted: true})
+			checker := &testSubmitChecker{checkSuccess: false, skipJobChecks: map[string]int{job.Id(): tc.skippedChecks}}
+			sched.submitChecker = checker
+
+			events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
+
+			assert.Equal(t, tc.expectedCheckCalls, checker.checkCalls, "the probe must ask again with a fresh budget")
+			require.Equal(t, tc.expectRequeued, hasRequeued(events.Events))
+			if tc.expectRequeued {
+				si := requeuedSchedulingInfo(t, events.Events)
+				expected := resource.MustParse("1536Mi")
+				grown := si.GetPodRequirements().ResourceRequirements.Requests["memory"]
+				assert.Equal(t, expected.Value(), grown.Value(), "an unprobed job must keep its mutation")
+			} else {
+				assert.Contains(t, terminalError(events.Events).GetMaxRunsExceeded().GetMessage(), "fits no node",
+					"a late verdict must decide the retry")
+			}
+		})
+	}
+}
+
 func TestRetryPolicy_FFOn_MemoryBumpFailsWhenUnschedulable(t *testing.T) {
 	policy := mkPolicy(t, 3, api.RetryAction_RETRY_ACTION_FAIL, &api.RetryRule{
 		Action:     api.RetryAction_RETRY_ACTION_RETRY,
@@ -669,7 +812,6 @@ func TestRetryPolicy_FFOn_MemoryBumpFailsWhenUnschedulable(t *testing.T) {
 	job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: memorySchedulingInfoFixture(), failedRuns: 1, runAttempted: true})
 
 	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
 
 	assert.False(t, hasRequeued(events.Events), "a bumped job that fits no node must fail instead of requeueing")
 	updated := txn.GetById(job.Id())
@@ -719,8 +861,7 @@ func TestRetryPolicy_FFOn_EngineRetryOptInAddsNodeAntiAffinity(t *testing.T) {
 	sched := makeRetryTestScheduler(t, true, fakePolicyCache{"test-policy": policy})
 	job := makeAttemptedFailedJobForRetry(t, sched)
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	si := requeuedSchedulingInfo(t, events.Events)
 	assert.Equal(t, []string{"testNode"}, nodeAntiAffinityValues(si),
@@ -739,7 +880,6 @@ func TestRetryPolicy_FFOn_EngineRetryOptInFailsWhenUnschedulable(t *testing.T) {
 	job := makeAttemptedFailedJobForRetry(t, sched)
 
 	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
 
 	// Opt-in matches the legacy path: if the anti-affinity makes the job
 	// unschedulable, it is failed terminally rather than requeued.
@@ -764,8 +904,7 @@ func TestRetryPolicy_FFOn_EngineRetryWithoutOptInSkipsAntiAffinity(t *testing.T)
 	sched.submitChecker = &testSubmitChecker{checkSuccess: false}
 	job := makeAttemptedFailedJobForRetry(t, sched)
 
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.True(t, hasRequeued(events.Events), "an engine retry without opt-in must requeue without consulting the scheduling probe")
 	si := requeuedSchedulingInfo(t, events.Events)
@@ -782,13 +921,13 @@ func TestRetryPolicy_FFOff_FailedRunIdentity(t *testing.T) {
 	job := makeFailedJobForRetry(t, sched)
 
 	txn := sched.jobDb.WriteTxn()
-	defer txn.Abort()
+	t.Cleanup(txn.Abort)
 	require.NoError(t, txn.Upsert([]*jobdb.Job{job}))
 
 	runError := containerErrorWithExitCode(42)
 	jobErrors := map[string]*armadaevents.Error{job.LatestRun().Id(): runError}
 
-	events, err := sched.generateUpdateMessagesFromJob(armadacontext.Background(), job, jobErrors, nil, txn)
+	events, err := sched.generateUpdateMessagesFromJob(armadacontext.Background(), job, jobErrors, nil, nil, txn)
 	require.NoError(t, err)
 	require.NotNil(t, events)
 
@@ -823,10 +962,10 @@ func TestRetryPolicy_FFOff_ApiPreemptionIdentity(t *testing.T) {
 	})
 
 	txn := sched.jobDb.WriteTxn()
-	defer txn.Abort()
+	t.Cleanup(txn.Abort)
 	require.NoError(t, txn.Upsert([]*jobdb.Job{job}))
 
-	events, err := sched.generateUpdateMessagesFromJob(armadacontext.Background(), job, nil, nil, txn)
+	events, err := sched.generateUpdateMessagesFromJob(armadacontext.Background(), job, nil, nil, nil, txn)
 	require.NoError(t, err)
 	require.NotNil(t, events)
 
@@ -857,7 +996,6 @@ func TestRetryPolicy_FFOn_FailFastLeaseExpiryFailsTerminally(t *testing.T) {
 	require.False(t, job.IsInGang(), "fail-fast fixture must not be a gang, so the failFast guard is what excludes it")
 
 	_, txn := runLeaseExpiryPath(t, sched, job)
-	defer txn.Abort()
 
 	updated := txn.GetById(job.Id())
 	require.NotNil(t, updated)
@@ -875,7 +1013,6 @@ func TestRetryPolicy_FFOn_FailFastFailurePathFailsTerminally(t *testing.T) {
 	job := makeRetryJob(t, sched, jobRunOpts{schedulingInfo: failFastSchedulingInfo, failedRuns: 1})
 
 	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
 
 	assert.False(t, hasRequeued(events.Events), "fail-fast job must not be requeued even when a retry rule matches")
 
@@ -908,8 +1045,7 @@ func TestRetryPolicy_FFOn_GangSkipIncrementsMetricAndDoesNotRetry(t *testing.T)
 	require.True(t, job.IsInGang(), "gang fixture must be a gang for this test to exercise the gang-skip branch")
 
 	before := testutil.ToFloat64(retryPolicyGangSkippedCounter.WithLabelValues("test-policy"))
-	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
+	events, _ := runFailurePath(t, sched, job, categorizedError("app-error"))
 
 	assert.False(t, hasRequeued(events.Events), "a gang job must never be retried by the engine even when a rule matches")
 	after := testutil.ToFloat64(retryPolicyGangSkippedCounter.WithLabelValues("test-policy"))
@@ -921,7 +1057,6 @@ func TestRetryPolicy_FFOff_LeaseExpiryIdentity(t *testing.T) {
 	job := makeRunningJobOnExecutor(t, sched)
 
 	eventSequences, txn := runLeaseExpiryPath(t, sched, job)
-	defer txn.Abort()
 	require.Len(t, eventSequences, 1)
 
 	expected := createEventsForFailedJob(
@@ -957,7 +1092,6 @@ func TestRetryPolicy_FFOff_FailurePathIgnoresPopulatedPolicy(t *testing.T) {
 	job := makeFailedJobForRetry(t, sched)
 
 	events, txn := runFailurePath(t, sched, job, categorizedError("app-error"))
-	defer txn.Abort()
 
 	assert.False(t, hasRequeued(events.Events), "flag off must not consult the engine, so a cached Retry policy must not requeue the job")
 	updated := txn.GetById(job.Id())
@@ -978,7 +1112,6 @@ func TestRetryPolicy_FFOff_LeaseExpiryIgnoresPopulatedPolicy(t *testing.T) {
 	job := makeRunningJobOnExecutor(t, sched)
 
 	_, txn := runLeaseExpiryPath(t, sched, job)
-	defer txn.Abort()
 
 	updated := txn.GetById(job.Id())
 	require.NotNil(t, updated)
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 610508dfdcb..6d6faf2af29 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -379,7 +379,9 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 
 	// Generate any eventSequences that came out of synchronising the db state.
 	ctx.Info("Generating update messages based on reconciliation changes")
-	events, err := s.generateUpdateMessages(ctx, txn, updatedJobs, jobRepoRunErrorsByRunId)
+	// The update messages and the expiry sweep below share one queue-to-policy map.
+	queueRetryPolicies := s.buildQueueRetryPolicyMap(ctx)
+	events, err := s.generateUpdateMessages(ctx, txn, updatedJobs, jobRepoRunErrorsByRunId, queueRetryPolicies)
 	if err != nil {
 		return false, err
 	}
@@ -394,7 +396,7 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 
 	// Expire any jobs running on clusters that haven't heartbeated within the configured deadline.
 	ctx.Info("Looking for jobs to expire")
-	expirationEvents, err := s.expireJobsIfNecessary(ctx, txn)
+	expirationEvents, err := s.expireJobsIfNecessary(ctx, txn, queueRetryPolicies)
 	if err != nil {
 		return false, err
 	}
@@ -1119,13 +1121,16 @@ func runErrorDetail(runError *armadaevents.Error) string {
 
 // generateUpdateMessages generates EventSequences representing the state changes on updated jobs.
 // If there are no state changes then an empty slice will be returned.
-func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, txn *jobdb.Txn, updatedJobs []*jobdb.Job, jobRunErrors map[string]*armadaevents.Error) ([]*armadaevents.EventSequence, error) {
-	queueRetryPolicies := s.buildQueueRetryPolicyMap(ctx)
+func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, txn *jobdb.Txn, updatedJobs []*jobdb.Job, jobRunErrors map[string]*armadaevents.Error, queueRetryPolicies map[string]string) ([]*armadaevents.EventSequence, error) {
+	plans, err := s.planRetryDecisions(ctx, updatedJobs, jobRunErrors, queueRetryPolicies)
+	if err != nil {
+		return nil, err
+	}
 
 	// Generate any eventSequences that came out of synchronising the db state.
 	var events []*armadaevents.EventSequence
 	for _, job := range updatedJobs {
-		jobEvents, err := s.generateUpdateMessagesFromJob(ctx, job, jobRunErrors, queueRetryPolicies, txn)
+		jobEvents, err := s.generateUpdateMessagesFromJob(ctx, job, jobRunErrors, queueRetryPolicies, plans, txn)
 		if err != nil {
 			return nil, err
 		}
@@ -1164,9 +1169,238 @@ func (s *Scheduler) buildQueueRetryPolicyMap(ctx *armadacontext.Context) map[str
 	return m
 }
 
+// plannedRetry is the resolved retry decision for one failed run.
+// planRetryDecisions computes it before event generation, so the round makes at
+// most two schedulability probe calls.
+type plannedRetry struct {
+	engineResult     retry.Result
+	enginePolicyName string
+	engineDecided    bool
+	requeueJob       bool
+	// newSchedulingInfo carries the granted mutations (memory bump, node
+	// anti-affinity). It is nil when the retry changes nothing. The event
+	// generator applies it to the job it holds, so the job keeps the changes
+	// from the earlier branches, for example a reprioritisation.
+	newSchedulingInfo *internaltypes.JobSchedulingInfo
+}
+
+// probeCandidate pairs a mutated job with the plan that receives the
+// schedulability verdict.
+type probeCandidate struct {
+	plan *plannedRetry
+	job  *jobdb.Job
+	info *internaltypes.JobSchedulingInfo
+}
+
+// retryDecisionPending reports whether generateUpdateMessagesFromJob resolves
+// a retry decision for this job. The job needs a failed latest run that no
+// earlier branch (terminal state, cancellation, success) takes first.
+func retryDecisionPending(job *jobdb.Job) bool {
+	if job.InTerminalState() || job.CancelRequested() || job.CancelByJobsetRequested() || !job.HasRuns() {
+		return false
+	}
+	lastRun := job.LatestRun()
+	return !lastRun.Succeeded() && lastRun.Failed() && !job.Queued()
+}
+
+// planRetryDecisions resolves the retry decision for every failed run in the
+// batch. It evaluates the retry engine once per job. It then probes
+// schedulability with at most two batched SubmitChecker calls per round: one
+// for the memory-grown candidates and one for the node-anti-affinity
+// candidates. One call per job is expensive under mass failure.
+// generateUpdateMessagesFromJob consumes the plans.
+//
+// A decided engine verdict overrides the maxAttemptedRuns logic. The engine
+// evaluation applies the failFast and gang opt-outs internally, so the
+// gang-skip metric stays consistent across call sites.
+func (s *Scheduler) planRetryDecisions(ctx *armadacontext.Context, jobs []*jobdb.Job, jobRunErrors map[string]*armadaevents.Error, queueRetryPolicies map[string]string) (map[string]*plannedRetry, error) {
+	plans := map[string]*plannedRetry{}
+
+	var bumpCandidates []probeCandidate
+	for _, job := range jobs {
+		if !retryDecisionPending(job) {
+			continue
+		}
+		lastRun := job.LatestRun()
+		failFast := job.Annotations()[constants.FailFastAnnotation] == "true"
+		requeueJob := !failFast && lastRun.Returned() && job.NumAttempts() < s.maxAttemptedRuns
+		runError := jobRunErrors[lastRun.Id()]
+
+		plan := &plannedRetry{}
+		if s.retryPolicyConfig.Enabled {
+			plan.engineResult, plan.enginePolicyName, plan.engineDecided = s.evaluateRetryPolicy(ctx, job, runError, queueRetryPolicies)
+			if plan.engineDecided {
+				requeueJob = plan.engineResult.ShouldRetry
+			}
+		}
+		plan.requeueJob = requeueJob
+		plans[job.Id()] = plan
+
+		// A memory bump grows the job before it re-enters the queue: the
+		// scheduling info aggregate for placement and accounting, and the
+		// cumulative record the lease pipeline applies to the pod spec.
+		// Both change together or not at all.
+		memoryBump := plan.engineResult.Mutation.Resources.Memory
+		if requeueJob && !memoryBump.IsZero() {
+			newInfo, err := createSchedulingInfoWithMemoryBump(job, memoryBump)
+			if err != nil {
+				return nil, errors.Errorf("unable to apply memory bump for job %s because %s", job.Id(), err)
+			}
+			if newInfo == nil {
+				ctx.Warnf("skipping memory bump for job %s: the bump kind differs from the job's accumulated record", job.Id())
+			} else {
+				candidate, err := job.WithJobSchedulingInfo(newInfo)
+				if err != nil {
+					return nil, err
+				}
+				bumpCandidates = append(bumpCandidates, probeCandidate{plan: plan, job: candidate, info: newInfo})
+			}
+		}
+	}
+
+	// A grown job that fits no node fails terminally instead of queueing
+	// forever.
+	if err := s.resolveProbes(ctx, bumpCandidates, "retry granted, but the job grown by the rule's memory bump fits no node"); err != nil {
+		return nil, err
+	}
+
+	// Node anti-affinity steers a retry away from every node the job failed on.
+	// The lease-return retry path applies it to every attempted run. An engine
+	// retry applies it only when the matched rule opts in via
+	// mutate.affinity.avoidSameNode. An opted-in retry then behaves like a
+	// lease-return retry: the job fails if the anti-affinity makes it
+	// unschedulable.
+	var affinityCandidates []probeCandidate
+	for _, job := range jobs {
+		plan, ok := plans[job.Id()]
+		if !ok || !plan.requeueJob {
+			continue
+		}
+		lastRun := job.LatestRun()
+		if !lastRun.RunAttempted() || (plan.engineDecided && !plan.engineResult.Mutation.Affinity.AvoidSameNode) {
+			continue
+		}
+		base := job
+		if plan.newSchedulingInfo != nil {
+			var err error
+			base, err = job.WithJobSchedulingInfo(plan.newSchedulingInfo)
+			if err != nil {
+				return nil, err
+			}
+		}
+		newInfo, err := s.createSchedulingInfoWithNodeAntiAffinityForAttemptedRuns(base)
+		if err != nil {
+			return nil, errors.Errorf("unable to set node anti-affinity for job %s because %s", job.Id(), err)
+		}
+		candidate, err := base.WithJobSchedulingInfo(newInfo)
+		if err != nil {
+			return nil, err
+		}
+		affinityCandidates = append(affinityCandidates, probeCandidate{plan: plan, job: candidate, info: newInfo})
+	}
+
+	if err := s.resolveProbes(ctx, affinityCandidates, "retry granted, but no untried node fits the job"); err != nil {
+		return nil, err
+	}
+
+	return plans, nil
+}
+
+// resolveProbes checks all the candidates in one batch. A candidate that fits
+// a node keeps its mutation. A candidate that fits no node loses the retry,
+// and its plan records unschedulableReason.
+//
+// Candidates in the same queue with the same scheduling key get the same
+// verdict: the scheduling key covers the placement requirements, the checker
+// applies its per-queue resource limit, and the check is deterministic
+// against one snapshot. The probe therefore checks one representative per
+// queue and key and applies its verdict to the whole class. Gang members
+// reach this probe only on the lease-return path, where the probe judges
+// each member alone, as it always has.
+// A mass failure affects many jobs of few distinct shapes. The
+// probe cost therefore scales with the number of shapes, not with the number
+// of jobs.
+func (s *Scheduler) resolveProbes(ctx *armadacontext.Context, candidates []probeCandidate, unschedulableReason string) error {
+	if len(candidates) == 0 {
+		return nil
+	}
+	type probeClass struct {
+		queue string
+		key   internaltypes.SchedulingKey
+	}
+	classOf := func(job *jobdb.Job) probeClass {
+		return probeClass{queue: job.Queue(), key: job.SchedulingKey()}
+	}
+	representativeIdByClass := map[probeClass]string{}
+	var probedJobs []*jobdb.Job
+	for _, candidate := range candidates {
+		class := classOf(candidate.job)
+		if _, ok := representativeIdByClass[class]; ok {
+			continue
+		}
+		representativeIdByClass[class] = candidate.job.Id()
+		probedJobs = append(probedJobs, candidate.job)
+	}
+	// The checker applies its time limits per call, so a new call starts with
+	// a fresh budget. The probe therefore asks again for the representatives
+	// that an earlier call did not reach.
+	const maxProbeAttempts = 3
+	results := map[string]schedulingResult{}
+	unresolved := probedJobs
+	for attempt := 0; attempt < maxProbeAttempts && len(unresolved) > 0; attempt++ {
+		callResults, _, err := s.submitChecker.Check(ctx, unresolved)
+		if err != nil {
+			return err
+		}
+		var remaining []*jobdb.Job
+		for _, job := range unresolved {
+			if result, ok := callResults[job.Id()]; ok {
+				results[job.Id()] = result
+			} else {
+				remaining = append(remaining, job)
+			}
+		}
+		unresolved = remaining
+	}
+	unprobed := 0
+	for _, candidate := range candidates {
+		result, ok := results[representativeIdByClass[classOf(candidate.job)]]
+		if !ok {
+			// The checker returns partial results when it reaches its time
+			// limits. An absent result means "not probed", not
+			// "unschedulable". A job that fits no node then waits in the
+			// queue, and an operator can recover it. A terminal failure is
+			// not recoverable.
+			unprobed++
+			candidate.plan.newSchedulingInfo = candidate.info
+			continue
+		}
+		if result.isSchedulable {
+			candidate.plan.newSchedulingInfo = candidate.info
+			continue
+		}
+		candidate.plan.requeueJob = false
+		if candidate.plan.engineDecided {
+			candidate.plan.engineResult.Decision = retry.DecisionRetryUnschedulable
+			candidate.plan.engineResult.Reason = unschedulableReason
+		}
+	}
+	if unprobed > 0 {
+		ctx.Warnf("the schedulability probe returned no result for %d of %d retry candidates. Their retries proceed unprobed", unprobed, len(candidates))
+	}
+	return nil
+}
+
 // generateUpdateMessages generates an EventSequence representing the state changes for a single job.
 // If there are no state changes it returns nil.
-func (s *Scheduler) generateUpdateMessagesFromJob(ctx *armadacontext.Context, job *jobdb.Job, jobRunErrors map[string]*armadaevents.Error, queueRetryPolicies map[string]string, txn *jobdb.Txn) (*armadaevents.EventSequence, error) {
+func (s *Scheduler) generateUpdateMessagesFromJob(
+	ctx *armadacontext.Context,
+	job *jobdb.Job,
+	jobRunErrors map[string]*armadaevents.Error,
+	queueRetryPolicies map[string]string,
+	plans map[string]*plannedRetry,
+	txn *jobdb.Txn,
+) (*armadaevents.EventSequence, error) {
 	var events []*armadaevents.EventSequence_Event
 
 	// Is the job already in a terminal state? If so then don't send any more messages
@@ -1288,65 +1522,68 @@ func (s *Scheduler) generateUpdateMessagesFromJob(ctx *armadacontext.Context, jo
 			events = append(events, jobSucceeded)
 		} else if lastRun.Failed() && !job.Queued() {
 			failFast := job.Annotations()[constants.FailFastAnnotation] == "true"
-			requeueJob := !failFast && lastRun.Returned() && job.NumAttempts() < s.maxAttemptedRuns
 			runError := jobRunErrors[lastRun.Id()]
 
-			// A decided engine verdict overrides the maxAttemptedRuns logic
-			// and suppresses its "Maximum number of attempts ..." message.
-			// evaluateRetryPolicy applies the failFast and gang opt-outs
-			// internally, so both call sites reach it and the gang-skip
-			// metric stays consistent.
-			// enginePolicyName is recorded on the emitted event so the decision
-			// is attributable downstream.
+			// enginePolicyName is recorded on the emitted event so the
+			// decision is attributable downstream.
 			var engineResult retry.Result
 			var enginePolicyName string
 			var engineDecided bool
-			if s.retryPolicyConfig.Enabled {
-				engineResult, enginePolicyName, engineDecided = s.evaluateRetryPolicy(ctx, job, runError, queueRetryPolicies)
-				if engineDecided {
-					requeueJob = engineResult.ShouldRetry
-				}
-			}
-
-			// A memory bump grows the job before it re-enters the queue: the
-			// scheduling info aggregate for placement and accounting, and the
-			// cumulative record the lease pipeline applies to the pod spec.
-			// Both change together or not at all. If the bumped job fits no
-			// node, it fails terminally instead of queueing forever.
-			memoryBump := engineResult.Mutation.Resources.Memory
-			if requeueJob && !memoryBump.IsZero() {
-				bumpedJob, schedulable, err := s.applyMemoryBumpIfSchedulable(ctx, job, memoryBump)
-				if err != nil {
-					return nil, errors.Errorf("unable to apply memory bump for job %s because %s", job.Id(), err)
+			var requeueJob bool
+			if plan, ok := plans[job.Id()]; ok {
+				engineResult = plan.engineResult
+				enginePolicyName = plan.enginePolicyName
+				engineDecided = plan.engineDecided
+				requeueJob = plan.requeueJob
+				if requeueJob && plan.newSchedulingInfo != nil {
+					var err error
+					job, err = job.WithJobSchedulingInfo(plan.newSchedulingInfo)
+					if err != nil {
+						return nil, err
+					}
 				}
-				if schedulable {
-					job = bumpedJob
-				} else {
-					requeueJob = false
-					engineResult.Decision = retry.DecisionRetryUnschedulable
-					engineResult.Reason = "retry granted, but the job grown by the rule's memory bump fits no node"
+			} else {
+				// Fallback for a job the planning pass did not cover. It
+				// repeats the decision logic of planRetryDecisions and probes
+				// this job alone. The warning surfaces a drift between
+				// retryDecisionPending and this branch.
+				ctx.Warnf("job %s reached the failure branch without a retry plan; deciding with a per-job probe", job.Id())
+				requeueJob = !failFast && lastRun.Returned() && job.NumAttempts() < s.maxAttemptedRuns
+				if s.retryPolicyConfig.Enabled {
+					engineResult, enginePolicyName, engineDecided = s.evaluateRetryPolicy(ctx, job, runError, queueRetryPolicies)
+					if engineDecided {
+						requeueJob = engineResult.ShouldRetry
+					}
 				}
-			}
 
-			// Node anti-affinity steers a retry away from every node it failed
-			// on. The lease-return retry path always applies it. Engine retries
-			// apply it only when the matched rule opts in via
-			// mutate.affinity.avoidSameNode, because the schedulability probe
-			// costs a per-job SubmitChecker.Check, which is expensive under
-			// mass failure. An opted-in retry behaves like a lease-return
-			// retry: the job fails if the anti-affinity makes it unschedulable.
-			if requeueJob && lastRun.RunAttempted() && (!engineDecided || engineResult.Mutation.Affinity.AvoidSameNode) {
-				jobWithAntiAffinity, schedulable, err := s.addNodeAntiAffinitiesForAttemptedRunsIfSchedulable(ctx, job)
-				if err != nil {
-					return nil, errors.Errorf("unable to set node anti-affinity for job %s because %s", job.Id(), err)
-				}
-				if schedulable {
-					job = jobWithAntiAffinity
-				} else {
-					requeueJob = false
-					if engineDecided {
+				memoryBump := engineResult.Mutation.Resources.Memory
+				if requeueJob && !memoryBump.IsZero() {
+					bumpedJob, schedulable, err := s.applyMemoryBumpIfSchedulable(ctx, job, memoryBump)
+					if err != nil {
+						return nil, errors.Errorf("unable to apply memory bump for job %s because %s", job.Id(), err)
+					}
+					if schedulable {
+						job = bumpedJob
+					} else {
+						requeueJob = false
 						engineResult.Decision = retry.DecisionRetryUnschedulable
-						engineResult.Reason = "retry granted, but no untried node fits the job"
+						engineResult.Reason = "retry granted, but the job grown by the rule's memory bump fits no node"
+					}
+				}
+
+				if requeueJob && lastRun.RunAttempted() && (!engineDecided || engineResult.Mutation.Affinity.AvoidSameNode) {
+					jobWithAntiAffinity, schedulable, err := s.addNodeAntiAffinitiesForAttemptedRunsIfSchedulable(ctx, job)
+					if err != nil {
+						return nil, errors.Errorf("unable to set node anti-affinity for job %s because %s", job.Id(), err)
+					}
+					if schedulable {
+						job = jobWithAntiAffinity
+					} else {
+						requeueJob = false
+						if engineDecided {
+							engineResult.Decision = retry.DecisionRetryUnschedulable
+							engineResult.Reason = "retry granted, but no untried node fits the job"
+						}
 					}
 				}
 			}
@@ -1496,7 +1733,7 @@ func (s *Scheduler) generateUpdateMessagesFromJob(ctx *armadacontext.Context, jo
 // expireJobsIfNecessary removes any jobs from the JobDb which are running on stale executors.
 // It also generates an EventSequence for each job, indicating that both the run and the job has failed
 // Note that this is different behaviour from the old scheduler which would allow expired jobs to be rerun
-func (s *Scheduler) expireJobsIfNecessary(ctx *armadacontext.Context, txn *jobdb.Txn) ([]*armadaevents.EventSequence, error) {
+func (s *Scheduler) expireJobsIfNecessary(ctx *armadacontext.Context, txn *jobdb.Txn, queueRetryPolicies map[string]string) ([]*armadaevents.EventSequence, error) {
 	heartbeatTimes, err := s.executorRepository.GetLastUpdateTimes(ctx)
 	if err != nil {
 		return nil, err
@@ -1524,10 +1761,6 @@ func (s *Scheduler) expireJobsIfNecessary(ctx *armadacontext.Context, txn *jobdb
 
 	events := make([]*armadaevents.EventSequence, 0)
 
-	// Resolved once per expiry sweep. It is nil when the feature flag is off,
-	// in which case every expired job takes the terminal path below.
-	queueRetryPolicies := s.buildQueueRetryPolicyMap(ctx)
-
 	jobs := txn.GetAllLeasedJobs()
 
 	for _, job := range jobs {
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 91b6f8ab1bb..7e927ad1b22 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -466,6 +466,9 @@ func TestScheduler_TestCycle(t *testing.T) {
 		expectedQueuedVersion              int32                                 // expected queued version of jobs at the end of the cycle
 		cordonedQueues                     []string                              // queues that are cordoned
 		queueCacheError                    bool                                  // if true then the queue cache will throw an error
+		retryPolicyEnabled                 bool                                  // if true then the retry policy engine runs with a global cap of 10 retries
+		retryPolicies                      fakePolicyCache                       // policies served to the retry engine, keyed by name
+		queuesWithRetryPolicies            map[string][]string                   // queue name -> attached policy names, added to the queue cache
 		expectedPreemptReasons             map[string]string                     // map of job id to expected preempt reason on the latest run
 		expectedShortJobPenalties          map[string]internaltypes.ResourceList // map of queue to the short-job penalty resources expected for testPool
 	}{
@@ -646,6 +649,33 @@ func TestScheduler_TestCycle(t *testing.T) {
 			expectedJobSchedulingInfoVersion: 2,
 			expectedQueuedVersion:            leasedJob.QueuedVersion() + 1,
 		},
+		"Failed categorized run requeued by retry policy": {
+			initialJobs:        []*jobdb.Job{leasedJob},
+			retryPolicyEnabled: true,
+			retryPolicies: fakePolicyCache{"test-policy": mkPolicy(t, 3, api.RetryAction_RETRY_ACTION_FAIL, &api.RetryRule{
+				Action:     api.RetryAction_RETRY_ACTION_RETRY,
+				OnCategory: "app-error",
+			})},
+			queuesWithRetryPolicies: map[string][]string{"testQueue": {"test-policy"}},
+			runUpdates: []database.Run{
+				{
+					RunID:        leasedJob.LatestRun().Id(),
+					JobID:        leasedJob.Id(),
+					JobSet:       "testJobSet",
+					Executor:     "testExecutor",
+					Failed:       true,
+					RunAttempted: true,
+					Serial:       1,
+				},
+			},
+			jobRunErrors: map[string]*armadaevents.Error{
+				leasedJob.LatestRun().Id(): categorizedError("app-error"),
+			},
+			expectedJobErrors:     []string{leasedJob.Id()},
+			expectedQueued:        []string{leasedJob.Id()},
+			expectedRequeued:      []string{leasedJob.Id()},
+			expectedQueuedVersion: leasedJob.QueuedVersion() + 1,
+		},
 		"Lease returned and re-queued when run not attempted": {
 			initialJobs: []*jobdb.Job{leasedJob},
 			runUpdates: []database.Run{
@@ -1128,7 +1158,16 @@ func TestScheduler_TestCycle(t *testing.T) {
 			for _, name := range tc.cordonedQueues {
 				queues = append(queues, &api.Queue{Name: name, Cordoned: true})
 			}
+			for queueName, policyNames := range tc.queuesWithRetryPolicies {
+				queues = append(queues, &api.Queue{Name: queueName, RetryPolicies: policyNames})
+			}
 			queueCache := &testQueueCache{queues: queues, shouldError: tc.queueCacheError}
+			retryPolicyConfig := schedulerconfig.RetryPolicyConfig{}
+			var retryPolicyCache retry.PolicyCache = retry.NoopPolicyCache{}
+			if tc.retryPolicyEnabled {
+				retryPolicyConfig = schedulerconfig.RetryPolicyConfig{Enabled: true, GlobalMaxRetries: 10}
+				retryPolicyCache = tc.retryPolicies
+			}
 			shortJobPenalty := scheduling.NewShortJobPenalty(map[string]time.Duration{"pool": time.Minute})
 			shortJobPenalty.SetNow(shortJobRunningTime.Add(time.Second))
 			sched, err := NewScheduler(
@@ -1150,8 +1189,8 @@ func TestScheduler_TestCycle(t *testing.T) {
 				pricing.NoopBidPriceProvider{},
 				[]string{},
 				queueCache,
-				schedulerconfig.RetryPolicyConfig{},
-				retry.NoopPolicyCache{},
+				retryPolicyConfig,
+				retryPolicyCache,
 			)
 			require.NoError(t, err)
 			sched.EnableAssertions()
@@ -2408,12 +2447,24 @@ func (t *testGangValidator) Validate(txn *jobdb.Txn, jobs []*jobdb.Job) ([]*inva
 }
 
 type testSubmitChecker struct {
-	checkSuccess bool
+	checkSuccess   bool
+	checkCalls     int
+	checkJobCounts []int
+	// skipJobChecks simulates the checker's time limits. The real checker
+	// leaves a job it does not reach out of the result map. The value counts
+	// the calls that skip the job. The next call answers for it.
+	skipJobChecks map[string]int
 }
 
 func (t *testSubmitChecker) Check(_ *armadacontext.Context, jobs []*jobdb.Job) (map[string]schedulingResult, map[string]time.Duration, error) {
+	t.checkCalls++
+	t.checkJobCounts = append(t.checkJobCounts, len(jobs))
 	result := make(map[string]schedulingResult)
 	for _, job := range jobs {
+		if t.skipJobChecks[job.Id()] > 0 {
+			t.skipJobChecks[job.Id()]--
+			continue
+		}
 		if t.checkSuccess {
 			result[job.Id()] = schedulingResult{isSchedulable: true}
 		} else {