Repository: armadaproject/armada
Pull request #2642 (2023-07-05 15:48:47)

Unified scheduling across clusters
This PR adds a new scheduler flag, which if enabled causes the Pulsar-backed scheduler to schedule across all executors in the same pool in a unified manner. Specifically, all nodes managed by executors in the same pool are added to the same NodeDb before scheduling.

The tests have been reworked quite a bit to support this.

┆Issue is synchronized with this [Jira Task](https://gr-oss.atlassian.net/browse/BATCH-429) by [Unito](https://www.unito.io)


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/armadaproject/armada/pull/2642?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) Report
Patch coverage has no change and project coverage change: **`-0.01`** :warning:
> Comparison is base [(`f128481`)](https://app.codecov.io/gh/armadaproject/armada/commit/f1284816cc4a90bc1e03f5970485f6247f996053?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) 16.07% compared to head [(`c37cdf5`)](https://app.codecov.io/gh/armadaproject/armada/pull/2642?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) 16.07%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #2642      +/-   ##
==========================================
- Coverage   16.07%   16.07%   -0.01%     
==========================================
  Files         159      159              
  Lines       12490    12491       +1     
  Branches      470      470              
==========================================
  Hits         2008     2008              
- Misses      10320    10321       +1     
  Partials      162      162              
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `16.07% <0.00%> (-0.01%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://app.codecov.io/gh/armadaproject/armada/pull/2642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) | Coverage Δ | |
|---|---|---|
| [pkg/api/util.go](https://app.codecov.io/gh/armadaproject/armada/pull/2642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-cGtnL2FwaS91dGlsLmdv) | `22.92% <0.00%> (-0.08%)` | :arrow_down: |


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/armadaproject/armada/pull/2642?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).


Patch:
diff --git a/internal/armada/configuration/types.go b/internal/armada/configuration/types.go
index e9994f9f..aa53d2d0 100644
--- a/internal/armada/configuration/types.go
+++ b/internal/armada/configuration/types.go
@@ -91,7 +91,10 @@ type PulsarConfig struct {
 type SchedulingConfig struct {
 	// Set to true to enable scheduler assertions. This results in some performance loss.
 	EnableAssertions bool
-	Preemption       PreemptionConfig
+	// If true, schedule jobs across all executors in the same pool in a unified manner.
+	// Otherwise, schedule each executor separately.
+	UnifiedSchedulingByPool bool
+	Preemption              PreemptionConfig
 	// Number of jobs to load from the database at a time.
 	MaxQueueLookback uint
 	// In each invocation of the scheduler, no more jobs are scheduled once this limit has been exceeded.
diff --git a/internal/armada/server/lease.go b/internal/armada/server/lease.go
index 9981c971..4bee2145 100644
--- a/internal/armada/server/lease.go
+++ b/internal/armada/server/lease.go
@@ -693,8 +693,8 @@ func (q *AggregatedQueueServer) getJobs(ctx context.Context, req *api.StreamingL
 			v := node.Labels[q.schedulingConfig.Preemption.NodeIdLabel]
 			if v == "" {
 				log.Warnf(
-					"failed to set node id selector on job %s to target node %s: nodeIdLabel missing from %s",
-					apiJob.Id, node.Name, node.Labels,
+					"failed to set node id selector on job %s to target node %s (id %s): nodeIdLabel missing from %s",
+					apiJob.Id, node.Name, node.Id, node.Labels,
 				)
 				continue
 			}
diff --git a/internal/common/slices/slices.go b/internal/common/slices/slices.go
index 0f527736..db8301e3 100644
--- a/internal/common/slices/slices.go
+++ b/internal/common/slices/slices.go
@@ -97,6 +97,16 @@ func GroupByFunc[S ~[]E, E any, K comparable](s S, keyFunc func(E) K) map[K]S {
 	return rv
 }
 
+// GroupByFuncUnique returns a map keyFunc(e) to e for each element e in s.
+func GroupByFuncUnique[S ~[]E, E any, K comparable](s S, keyFunc func(E) K) map[K]E {
+	rv := make(map[K]E, len(s))
+	for _, e := range s {
+		k := keyFunc(e)
+		rv[k] = e
+	}
+	return rv
+}
+
 // MapAndGroupByFuncs groups the elements e_1, ..., e_n of s into separate slices by keyFunc(e)
 // and then maps those resulting elements by mapFunc(e).
 func MapAndGroupByFuncs[S ~[]E, E any, K comparable, V any](s S, keyFunc func(E) K, mapFunc func(E) V) map[K][]V {
@@ -108,6 +118,14 @@ func MapAndGroupByFuncs[S ~[]E, E any, K comparable, V any](s S, keyFunc func(E)
 	return rv
 }
 
+// Pop removes the last item from s and returns it.
+// Calling pop on an empty slice causes a panic.
+func Pop[S ~[]E, E any](s *S) E {
+	v := (*s)[len(*s)-1]
+	*s = (*s)[:len(*s)-1]
+	return v
+}
+
 func Subtract[T comparable](list []T, toRemove []T) []T {
 	if list == nil {
 		return nil
diff --git a/internal/scheduler/api.go b/internal/scheduler/api.go
index 2e869782..9e40fcb9 100644
--- a/internal/scheduler/api.go
+++ b/internal/scheduler/api.go
@@ -40,9 +40,9 @@ type ExecutorApi struct {
 	maxJobsPerCall uint
 	// Max size of Pulsar messages produced.
 	maxPulsarMessageSizeBytes uint
-	// See scheduling config.
+	// See scheduling schedulingConfig.
 	nodeIdLabel string
-	// See scheduling config.
+	// See scheduling schedulingConfig.
 	priorityClassNameOverride *string
 	clock                     clock.Clock
 }
diff --git a/internal/scheduler/jobdb/job.go b/internal/scheduler/jobdb/job.go
index 5bac48da..f1f9da6e 100644
--- a/internal/scheduler/jobdb/job.go
+++ b/internal/scheduler/jobdb/job.go
@@ -333,13 +333,13 @@ func (job *Job) HasRuns() bool {
 }
 
 // WithNewRun creates a copy of the job with a new run on the given executor.
-func (job *Job) WithNewRun(executor string, node string) *Job {
+func (job *Job) WithNewRun(executor string, nodeId string) *Job {
 	run := &JobRun{
 		id:       uuid.New(),
 		jobId:    job.id,
 		created:  time.Now().UnixNano(),
 		executor: executor,
-		node:     node,
+		nodeId:   nodeId,
 	}
 	return job.WithUpdatedRun(run)
 }
diff --git a/internal/scheduler/jobdb/job_run.go b/internal/scheduler/jobdb/job_run.go
index 6d53c0a5..37b8e9d2 100644
--- a/internal/scheduler/jobdb/job_run.go
+++ b/internal/scheduler/jobdb/job_run.go
@@ -14,8 +14,8 @@ type JobRun struct {
 	created int64
 	// The name of the executor this run has been leased to.
 	executor string
-	// The name of the nodes this run has been leased to.
-	node string
+	// The id of the node this run has been leased to.
+	nodeId string
 	// True if the job has been reported as running by the executor.
 	running bool
 	// True if the job has been reported as succeeded by the executor.
@@ -43,7 +43,7 @@ func CreateRun(
 	jobId string,
 	creationTime int64,
 	executor string,
-	node string,
+	nodeId string,
 	running bool,
 	succeeded bool,
 	failed bool,
@@ -56,7 +56,7 @@ func CreateRun(
 		jobId:        jobId,
 		created:      creationTime,
 		executor:     executor,
-		node:         node,
+		nodeId:       nodeId,
 		running:      running,
 		succeeded:    succeeded,
 		failed:       failed,
@@ -81,9 +81,9 @@ func (run *JobRun) Executor() string {
 	return run.executor
 }
 
-// Node returns the node to which the JobRun is assigned.
+// Node returns the nodeId to which the JobRun is assigned.
 func (run *JobRun) Node() string {
-	return run.node
+	return run.nodeId
 }
 
 // Succeeded Returns true if the executor has reported the job run as successful
diff --git a/internal/scheduler/nodedb/nodedb.go b/internal/scheduler/nodedb/nodedb.go
index 45ff4ead..87aab622 100644
--- a/internal/scheduler/nodedb/nodedb.go
+++ b/internal/scheduler/nodedb/nodedb.go
@@ -39,8 +39,9 @@ const (
 var empty struct{}
 
 type Node struct {
-	Id   string
-	Name string
+	Id       string
+	Name     string
+	Executor string
 
 	// We need to store taints and labels separately from the node type: the latter only includes
 	// indexed taints and labels, but we need all of them when checking pod requirements.
@@ -64,8 +65,9 @@ type Node struct {
 // shallow copies of fields that are not mutated by methods of NodeDb.
 func (node *Node) UnsafeCopy() *Node {
 	return &Node{
-		Id:   node.Id,
-		Name: node.Name,
+		Id:       node.Id,
+		Name:     node.Name,
+		Executor: node.Executor,
 
 		Taints: node.Taints,
 		Labels: node.Labels,
@@ -144,8 +146,9 @@ func (nodeDb *NodeDb) create(node *schedulerobjects.Node) (*Node, error) {
 	nodeDb.mu.Unlock()
 
 	entry := &Node{
-		Id:   node.Id,
-		Name: node.Name,
+		Id:       node.Id,
+		Name:     node.Name,
+		Executor: node.Executor,
 
 		Taints: taints,
 		Labels: labels,
diff --git a/internal/scheduler/schedulerobjects/node.go b/internal/scheduler/schedulerobjects/node.go
index 7dc79612..22ba0ec5 100644
--- a/internal/scheduler/schedulerobjects/node.go
+++ b/internal/scheduler/schedulerobjects/node.go
@@ -21,6 +21,7 @@ func (node *Node) DeepCopy() *Node {
 	return &Node{
 		Id:             node.Id,
 		Name:           node.Name,
+		Executor:       node.Executor,
 		NodeDbKeys:     slices.Clone(node.NodeDbKeys),
 		LastSeen:       node.LastSeen,
 		NodeType:       node.NodeType.DeepCopy(),
diff --git a/internal/scheduler/schedulerobjects/reporting.pb.go b/internal/scheduler/schedulerobjects/reporting.pb.go
index 2669b8dd..18239dca 100644
--- a/internal/scheduler/schedulerobjects/reporting.pb.go
+++ b/internal/scheduler/schedulerobjects/reporting.pb.go
@@ -11,7 +11,6 @@ import (
 	math_bits "math/bits"
 
 	proto "github.com/gogo/protobuf/proto"
-	_ "github.com/gogo/protobuf/types"
 	grpc "google.golang.org/grpc"
 	codes "google.golang.org/grpc/codes"
 	status "google.golang.org/grpc/status"
@@ -454,41 +453,39 @@ func init() {
 }
 
 var fileDescriptor_131a439a3ff6540b = []byte{
-	// 529 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x54, 0xcf, 0x6a, 0x13, 0x5f,
-	0x14, 0xce, 0xe4, 0x47, 0xc3, 0x2f, 0xa7, 0xa2, 0xc3, 0x8d, 0xd2, 0x90, 0xea, 0x24, 0x0c, 0x2e,
-	0xaa, 0x94, 0x0c, 0xb4, 0x28, 0x88, 0x50, 0x24, 0x82, 0xd1, 0xe0, 0x1f, 0x9c, 0xe2, 0x46, 0x90,
-	0x30, 0x37, 0x39, 0x49, 0x27, 0x64, 0xe6, 0xa4, 0x77, 0xee, 0x08, 0xc5, 0xa5, 0x2f, 0xe0, 0x33,
-	0xf8, 0x34, 0x2e, 0x5c, 0x74, 0xe9, 0x2a, 0x48, 0xb2, 0x9b, 0xa7, 0x90, 0xde, 0x49, 0x93, 0xf9,
-	0x53, 0xdb, 0xc6, 0xdd, 0xcc, 0x77, 0xcf, 0xfd, 0xce, 0xf9, 0xce, 0x77, 0xce, 0x85, 0x7d, 0xd7,
-	0x97, 0x28, 0x7c, 0x67, 0x6c, 0x05, 0xbd, 0x23, 0xec, 0x87, 0x63, 0x14, 0xab, 0x2f, 0xe2, 0x23,
-	0xec, 0xc9, 0xc0, 0x12, 0x38, 0x21, 0x21, 0x5d, 0x7f, 0xd8, 0x9c, 0x08, 0x92, 0xc4, 0xf4, 0x6c,
-	0x44, 0x6d, 0x7b, 0x48, 0x34, 0x1c, 0xa3, 0xa5, 0xce, 0x79, 0x38, 0xb0, 0xd0, 0x9b, 0xc8, 0x93,
-	0x38, 0xdc, 0x7c, 0x0d, 0xec, 0x0d, 0x05, 0xd2, 0xc6, 0x1e, 0xfa, 0xf2, 0x05, 0x89, 0xf7, 0x21,
-	0x86, 0xc8, 0x1e, 0x03, 0x1c, 0x9f, 0x7d, 0x74, 0x7d, 0xc7, 0xc3, 0xaa, 0xd6, 0xd0, 0x76, 0xca,
-	0xad, 0xad, 0x68, 0x5a, 0xaf, 0x28, 0xf4, 0xad, 0xe3, 0xe1, 0x2e, 0x79, 0xae, 0x54, 0x44, 0x76,
-	0x79, 0x09, 0x9a, 0x07, 0xa0, 0xa7, 0xd8, 0x3a, 0xc4, 0xd9, 0x43, 0x28, 0x8d, 0x88, 0x77, 0xdd,
-	0xfe, 0x82, 0xa7, 0x12, 0x4d, 0xeb, 0xb7, 0x46, 0xc4, 0x5f, 0xf5, 0x13, 0x1c, 0x1b, 0x0a, 0x30,
-	0x7f, 0x16, 0x61, 0xeb, 0x30, 0xae, 0xdf, 0xf5, 0x87, 0xb6, 0x92, 0x66, 0xe3, 0x71, 0x88, 0x81,
-	0x64, 0x5f, 0xe0, 0x8e, 0x47, 0x81, 0xec, 0x0a, 0x45, 0xde, 0x1d, 0x90, 0xe8, 0xaa, 0xc4, 0x8a,
-	0x76, 0x73, 0xef, 0x7e, 0x33, 0x2b, 0xbc, 0x99, 0x17, 0xd6, 0x6a, 0x44, 0xd3, 0xfa, 0x5d, 0x2f,
-	0x87, 0xaf, 0x2a, 0x79, 0x59, 0xb0, 0x59, 0xfe, 0x9c, 0x05, 0x50, 0xc9, 0x26, 0x1f, 0x11, 0xaf,
-	0x16, 0x55, 0x6a, 0xf3, 0x8a, 0xd4, 0x1d, 0xe2, 0x2d, 0x23, 0x9a, 0xd6, 0x6b, 0x5e, 0x06, 0x4d,
-	0xa5, 0xd5, 0xb3, 0xa7, 0xec, 0x11, 0x94, 0x3f, 0xa3, 0xe0, 0x14, 0xb8, 0xf2, 0xa4, 0xfa, 0x5f,
-	0x43, 0xdb, 0xd9, 0x88, 0x4d, 0x58, 0x82, 0x49, 0x13, 0x96, 0x60, 0xeb, 0x7f, 0x28, 0x0d, 0xdc,
-	0xb1, 0x44, 0x61, 0x3e, 0x03, 0x3d, 0xdb, 0x4d, 0xb6, 0x0b, 0xa5, 0x78, 0x64, 0x16, 0x76, 0xdc,
-	0x8e, 0xa6, 0x75, 0x3d, 0x46, 0x12, 0x74, 0x8b, 0x18, 0xf3, 0xab, 0x06, 0x4c, 0x75, 0x20, 0xed,
-	0xc5, 0x3f, 0xce, 0x47, 0x5a, 0x51, 0xf1, 0xba, 0x8a, 0xcc, 0xa7, 0xb0, 0x99, 0x28, 0x62, 0x4d,
-	0x09, 0x07, 0xa0, 0x77, 0x88, 0xa7, 0xeb, 0x5f, 0x67, 0x26, 0x9f, 0x40, 0x79, 0x79, 0x7f, 0xbd,
-	0xd4, 0x7b, 0xdf, 0x8b, 0xc0, 0x0e, 0xcf, 0x47, 0xc3, 0x3e, 0x5f, 0x54, 0xd6, 0x87, 0x4a, 0x1b,
-	0x65, 0xce, 0x99, 0x07, 0xf9, 0x31, 0xfa, 0xcb, 0x2e, 0xd4, 0xcc, 0xab, 0x43, 0xd9, 0x07, 0xb8,
-	0xd9, 0x46, 0x99, 0xec, 0xdb, 0x05, 0x2b, 0x92, 0xf7, 0xb6, 0x76, 0xef, 0xd2, 0x28, 0xf6, 0x0e,
-	0x6e, 0xb4, 0x51, 0xae, 0x3a, 0x72, 0x41, 0x29, 0xd9, 0x76, 0xd7, 0xb6, 0x2f, 0x89, 0x69, 0x7d,
-	0xfa, 0x31, 0x33, 0xb4, 0xd3, 0x99, 0xa1, 0xfd, 0x9e, 0x19, 0xda, 0xb7, 0xb9, 0x51, 0x38, 0x9d,
-	0x1b, 0x85, 0x5f, 0x73, 0xa3, 0xf0, 0xf1, 0xf9, 0xd0, 0x95, 0x47, 0x21, 0x6f, 0xf6, 0xc8, 0xb3,
-	0x1c, 0xe1, 0x39, 0x7d, 0x67, 0x22, 0xe8, 0xec, 0xfa, 0xe2, 0xcf, 0xba, 0xc6, 0xfb, 0xc8, 0x4b,
-	0xea, 0x9d, 0xdb, 0xff, 0x13, 0x00, 0x00, 0xff, 0xff, 0xf1, 0xb4, 0x16, 0xb6, 0x4d, 0x05, 0x00,
-	0x00,
+	// 512 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x54, 0x5f, 0x6b, 0xd3, 0x50,
+	0x14, 0x6f, 0x2a, 0x2b, 0xf6, 0x4c, 0x34, 0xdc, 0x2a, 0x1b, 0x55, 0x93, 0x12, 0x7c, 0x98, 0x32,
+	0x5a, 0xd8, 0x50, 0x10, 0x61, 0x48, 0x04, 0xab, 0xc5, 0x3f, 0x98, 0xe1, 0x8b, 0x20, 0x21, 0xb7,
+	0x39, 0xeb, 0x52, 0x9a, 0x9c, 0xee, 0xe6, 0x46, 0x18, 0x3e, 0xfa, 0x05, 0xfc, 0x0c, 0x7e, 0x1a,
+	0x1f, 0x7c, 0xd8, 0xa3, 0x4f, 0x41, 0xda, 0xb7, 0x7c, 0x0a, 0x59, 0xda, 0xb5, 0xf9, 0x33, 0xb7,
+	0xd5, 0xb7, 0xe4, 0x77, 0xcf, 0x3d, 0xbf, 0xdf, 0x39, 0xbf, 0x73, 0x2e, 0xec, 0x7a, 0x81, 0x44,
+	0x11, 0x38, 0xa3, 0x4e, 0xd8, 0x3f, 0x44, 0x37, 0x1a, 0xa1, 0x58, 0x7e, 0x11, 0x1f, 0x62, 0x5f,
+	0x86, 0x1d, 0x81, 0x63, 0x12, 0xd2, 0x0b, 0x06, 0xed, 0xb1, 0x20, 0x49, 0x4c, 0x2d, 0x46, 0x18,
+	0x6f, 0x80, 0xbd, 0xa5, 0x50, 0x5a, 0xd8, 0xc7, 0x40, 0xbe, 0x24, 0xf1, 0x21, 0xc2, 0x08, 0xd9,
+	0x13, 0x80, 0xa3, 0xd3, 0x0f, 0x3b, 0x70, 0x7c, 0xdc, 0x54, 0x5a, 0xca, 0x56, 0xdd, 0xdc, 0x48,
+	0x62, 0xbd, 0x91, 0xa2, 0xef, 0x1c, 0x1f, 0xb7, 0xc9, 0xf7, 0x24, 0xfa, 0x63, 0x79, 0x6c, 0xd5,
+	0x17, 0xa0, 0xb1, 0x07, 0x6a, 0x2e, 0x5b, 0x8f, 0x38, 0x7b, 0x04, 0xb5, 0x21, 0x71, 0xdb, 0x73,
+	0xe7, 0x79, 0x1a, 0x49, 0xac, 0xdf, 0x1a, 0x12, 0x7f, 0xed, 0x66, 0x72, 0xac, 0xa5, 0x80, 0xf1,
+	0xab, 0x0a, 0x1b, 0xfb, 0x33, 0x89, 0x5e, 0x30, 0xb0, 0x52, 0xf5, 0x16, 0x1e, 0x45, 0x18, 0x4a,
+	0xf6, 0x15, 0xee, 0xf8, 0x14, 0x4a, 0x5b, 0xa4, 0xc9, 0xed, 0x03, 0x12, 0x76, 0x4a, 0x9c, 0xa6,
+	0x5d, 0xdf, 0x79, 0xd0, 0x2e, 0xd6, 0xd6, 0x2e, 0x17, 0x66, 0xb6, 0x92, 0x58, 0xbf, 0xe7, 0x97,
+	0xf0, 0xa5, 0x92, 0x57, 0x15, 0x8b, 0x95, 0xcf, 0x59, 0x08, 0x8d, 0x22, 0xf9, 0x90, 0xf8, 0x66,
+	0x35, 0xa5, 0x36, 0x2e, 0xa1, 0xee, 0x11, 0x37, 0xb5, 0x24, 0xd6, 0x9b, 0x7e, 0x01, 0xcd, 0xd1,
+	0xaa, 0xc5, 0x53, 0xf6, 0x18, 0xea, 0x5f, 0x50, 0x70, 0x0a, 0x3d, 0x79, 0xbc, 0x79, 0xad, 0xa5,
+	0x6c, 0xad, 0xcd, 0x4c, 0x58, 0x80, 0x59, 0x13, 0x16, 0xa0, 0x79, 0x1d, 0x6a, 0x07, 0xde, 0x48,
+	0xa2, 0x30, 0x9e, 0x83, 0x5a, 0xec, 0x26, 0xdb, 0x86, 0xda, 0x6c, 0x2a, 0xe6, 0x76, 0xdc, 0x4e,
+	0x62, 0x5d, 0x9d, 0x21, 0x99, 0x74, 0xf3, 0x18, 0xe3, 0x9b, 0x02, 0x2c, 0xed, 0x40, 0xde, 0x8b,
+	0xff, 0x9c, 0x8f, 0x7c, 0x45, 0xd5, 0xab, 0x56, 0x64, 0x3c, 0x83, 0xf5, 0x8c, 0x88, 0x15, 0x4b,
+	0xd8, 0x03, 0xb5, 0x47, 0x3c, 0xaf, 0x7f, 0x95, 0x99, 0x7c, 0x0a, 0xf5, 0xc5, 0xfd, 0xd5, 0xa8,
+	0x77, 0x7e, 0x54, 0x81, 0xed, 0x9f, 0x8d, 0x86, 0x75, 0xb6, 0x8b, 0xcc, 0x85, 0x46, 0x17, 0x65,
+	0xc9, 0x99, 0x87, 0xe5, 0x31, 0xfa, 0xc7, 0x2e, 0x34, 0x8d, 0xcb, 0x43, 0xd9, 0x47, 0xb8, 0xd9,
+	0x45, 0x99, 0xed, 0xdb, 0x39, 0x2b, 0x52, 0xf6, 0xb6, 0x79, 0xff, 0xc2, 0x28, 0xf6, 0x1e, 0x6e,
+	0x74, 0x51, 0x2e, 0x3b, 0x72, 0x8e, 0x94, 0x62, 0xbb, 0x9b, 0x77, 0x2f, 0x88, 0x31, 0x3f, 0xff,
+	0x9c, 0x68, 0xca, 0xc9, 0x44, 0x53, 0xfe, 0x4c, 0x34, 0xe5, 0xfb, 0x54, 0xab, 0x9c, 0x4c, 0xb5,
+	0xca, 0xef, 0xa9, 0x56, 0xf9, 0xf4, 0x62, 0xe0, 0xc9, 0xc3, 0x88, 0xb7, 0xfb, 0xe4, 0x77, 0x1c,
+	0xe1, 0x3b, 0xae, 0x33, 0x16, 0x74, 0x7a, 0x7d, 0xfe, 0xd7, 0xb9, 0xc2, 0x13, 0xc8, 0x6b, 0xe9,
+	0xcb, 0xb7, 0xfb, 0x37, 0x00, 0x00, 0xff, 0xff, 0x09, 0x81, 0x96, 0x21, 0x30, 0x05, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
diff --git a/internal/scheduler/schedulerobjects/reporting.proto b/internal/scheduler/schedulerobjects/reporting.proto
index 62a67d4e..ed61242f 100644
--- a/internal/scheduler/schedulerobjects/reporting.proto
+++ b/internal/scheduler/schedulerobjects/reporting.proto
@@ -1,6 +1,5 @@
 syntax = 'proto3';
 package schedulerobjects;
-import "google/protobuf/empty.proto";
 option go_package = "github.com/armadaproject/armada/internal/scheduler/schedulerobjects";
 
 message MostRecentForQueue {
diff --git a/internal/scheduler/schedulerobjects/schedulerobjects.pb.go b/internal/scheduler/schedulerobjects/schedulerobjects.pb.go
index 18cc6ced..402a1e4f 100644
--- a/internal/scheduler/schedulerobjects/schedulerobjects.pb.go
+++ b/internal/scheduler/schedulerobjects/schedulerobjects.pb.go
@@ -161,6 +161,8 @@ type Node struct {
 	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
 	// Kubernetes node name. Not guaranteed to be unique across clusters.
 	Name string `protobuf:"bytes,12,opt,name=name,proto3" json:"name,omitempty"`
+	// Name of the executor to which this node belongs.
+	Executor string `protobuf:"bytes,19,opt,name=executor,proto3" json:"executor,omitempty"`
 	// Keys by which nodes are sorted in the nodeDb.
 	NodeDbKeys [][]byte `protobuf:"bytes,18,rep,name=nodeDbKeys,proto3" json:"nodeDbKeys,omitempty"`
 	// Time at which this node was last updated.
@@ -250,6 +252,13 @@ func (m *Node) GetName() string {
 	return ""
 }
 
+func (m *Node) GetExecutor() string {
+	if m != nil {
+		return m.Executor
+	}
+	return ""
+}
+
 func (m *Node) GetNodeDbKeys() [][]byte {
 	if m != nil {
 		return m.NodeDbKeys
@@ -1016,143 +1025,144 @@ func init() {
 }
 
 var fileDescriptor_97dadc5fbd620721 = []byte{
-	// 2173 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xc4, 0x19, 0x4d, 0x6f, 0x1b, 0xc7,
-	0x55, 0x2b, 0x52, 0x12, 0x39, 0x94, 0x25, 0x6a, 0x64, 0xd9, 0x2b, 0xc6, 0xe6, 0xd2, 0x8a, 0x1b,
-	0x28, 0x8d, 0x43, 0x36, 0x4e, 0x81, 0x1a, 0x6e, 0x2f, 0xa2, 0xa5, 0xd6, 0xf4, 0x07, 0x25, 0x2f,
-	0xa5, 0x16, 0x2d, 0xd0, 0x2c, 0x96, 0xdc, 0x11, 0xbd, 0xd1, 0x72, 0x86, 0xde, 0x9d, 0x55, 0xc3,
-	0x9c, 0xdb, 0x43, 0x11, 0x20, 0x0d, 0x8a, 0xb4, 0x0d, 0x50, 0xa0, 0x45, 0x6e, 0xfd, 0x05, 0xed,
-	0xa1, 0xb7, 0x9e, 0x7c, 0xcc, 0xb1, 0x27, 0xb6, 0xb0, 0x6f, 0x3c, 0xf7, 0x07, 0x14, 0x33, 0xb3,
-	0xcb, 0x1d, 0xee, 0x2e, 0x45, 0x39, 0xa9, 0xab, 0x13, 0x39, 0xef, 0x7b, 0xde, 0x7b, 0xf3, 0xe6,
-	0xbd, 0x59, 0x70, 0xd7, 0xc6, 0x14, 0xb9, 0xd8, 0x74, 0x6a, 0x5e, 0xe7, 0x29, 0xb2, 0x7c, 0x07,
-	0xb9, 0xd1, 0x3f, 0xd2, 0xfe, 0x10, 0x75, 0xa8, 0x97, 0x00, 0x54, 0xfb, 0x2e, 0xa1, 0x04, 0x16,
-	0xe3, 0xf0, 0x92, 0xd6, 0x25, 0xa4, 0xeb, 0xa0, 0x1a, 0xc7, 0xb7, 0xfd, 0xe3, 0x1a, 0xb5, 0x7b,
-	0xc8, 0xa3, 0x66, 0xaf, 0x2f, 0x58, 0x4a, 0x5b, 0x27, 0x77, 0xbc, 0xaa, 0x4d, 0x6a, 0x66, 0xdf,
-	0xae, 0x75, 0x88, 0x8b, 0x6a, 0xa7, 0xef, 0xd5, 0xba, 0x08, 0x23, 0xd7, 0xa4, 0xc8, 0x0a, 0x68,
-	0xbe, 0x1b, 0xd1, 0xf4, 0xcc, 0xce, 0x53, 0x1b, 0x23, 0x77, 0x50, 0xeb, 0x9f, 0x74, 0x39, 0x93,
-	0x8b, 0x3c, 0xe2, 0xbb, 0x1d, 0x94, 0xe0, 0x7a, 0xb7, 0x6b, 0xd3, 0xa7, 0x7e, 0xbb, 0xda, 0x21,
-	0xbd, 0x5a, 0x97, 0x74, 0x49, 0x64, 0x03, 0x5b, 0xf1, 0x05, 0xff, 0x27, 0xc8, 0xb7, 0xfe, 0x92,
-	0x01, 0xb9, 0xbd, 0x8f, 0x50, 0xc7, 0xa7, 0xc4, 0x85, 0x15, 0x30, 0x6f, 0x5b, 0xaa, 0x52, 0x51,
-	0xb6, 0xf3, 0xf5, 0xe2, 0x68, 0xa8, 0x2d, 0xdb, 0xd6, 0x2d, 0xd2, 0xb3, 0x29, 0xea, 0xf5, 0xe9,
-	0x40, 0x9f, 0xb7, 0x2d, 0xf8, 0x16, 0xc8, 0xf6, 0x09, 0x71, 0xd4, 0x79, 0x4e, 0x03, 0x47, 0x43,
-	0x6d, 0x85, 0xad, 0x25, 0x2a, 0x8e, 0x87, 0x3b, 0x60, 0x01, 0x13, 0x0b, 0x79, 0x6a, 0xa6, 0x92,
-	0xd9, 0x2e, 0xdc, 0xbe, 0x52, 0x4d, 0xb8, 0xae, 0x49, 0x2c, 0x54, 0x5f, 0x1f, 0x0d, 0xb5, 0x55,
-	0x4e, 0x28, 0x49, 0x10, 0x9c, 0xf0, 0x03, 0xb0, 0xd2, 0xb3, 0xb1, 0xdd, 0xf3, 0x7b, 0x0f, 0x48,
-	0xbb, 0x65, 0x7f, 0x8c, 0xd4, 0x6c, 0x45, 0xd9, 0x2e, 0xdc, 0x2e, 0x27, 0x65, 0xe9, 0x81, 0x33,
-	0x1e, 0xd9, 0x1e, 0xad, 0x5f, 0x79, 0x3e, 0xd4, 0xe6, 0x98, 0x61, 0x93, 0xdc, 0x7a, 0x6c, 0xcd,
-	0xe4, 0x3b, 0xa6, 0x47, 0x8f, 0xfa, 0x96, 0x49, 0xd1, 0xa1, 0xdd, 0x43, 0xea, 0x02, 0x97, 0x5f,
-	0xaa, 0x8a, 0xe0, 0x55, 0x43, 0xc7, 0x55, 0x0f, 0xc3, 0xe0, 0xd5, 0x4b, 0xa1, 0xec, 0x49, 0xce,
-	0xcf, 0xfe, 0xa5, 0x29, 0x7a, 0x0c, 0x06, 0xf7, 0xc1, 0xba, 0x8f, 0x4d, 0xcf, 0xb3, 0xbb, 0x18,
-	0x59, 0xc6, 0x87, 0xa4, 0x6d, 0xb8, 0x3e, 0xf6, 0xd4, 0x7c, 0x25, 0xb3, 0x9d, 0xaf, 0x6b, 0xa3,
-	0xa1, 0xf6, 0x46, 0x84, 0x7e, 0x40, 0xda, 0xba, 0x8f, 0x65, 0x27, 0xac, 0x25, 0x90, 0x5b, 0xff,
-	0xd9, 0x00, 0x59, 0xe6, 0xb5, 0xf3, 0x85, 0x09, 0x9b, 0x3d, 0xa4, 0x2e, 0x47, 0x61, 0x62, 0x6b,
-	0x39, 0x4c, 0x6c, 0x0d, 0xef, 0x00, 0xc0, 0x9c, 0xbd, 0xdb, 0x7e, 0x88, 0x06, 0x9e, 0x0a, 0x2b,
-	0x99, 0xed, 0xe5, 0xba, 0x3a, 0x1a, 0x6a, 0x97, 0x23, 0xa8, 0xc4, 0x23, 0xd1, 0xc2, 0xc7, 0x20,
-	0xcf, 0xf6, 0x6b, 0x78, 0x08, 0x61, 0x9e, 0x0d, 0x67, 0x3b, 0xee, 0x72, 0xe0, 0xb8, 0x1c, 0x63,
-	0x6a, 0x21, 0x84, 0xb9, 0xcb, 0xc6, 0x2b, 0xb8, 0x0f, 0xf2, 0x4c, 0xb8, 0x41, 0x07, 0x7d, 0xa4,
-	0x66, 0x02, 0x71, 0xa9, 0x39, 0x73, 0x38, 0xe8, 0xa3, 0xfa, 0x95, 0xd1, 0x50, 0x83, 0x38, 0x58,
-	0x49, 0x16, 0xe6, 0x42, 0x18, 0xbc, 0x0b, 0x96, 0xc7, 0x02, 0x0d, 0xdb, 0xe2, 0xb9, 0x93, 0x8d,
-	0xf6, 0xc6, 0x68, 0x1a, 0x56, 0x7c, 0x6f, 0x02, 0x0a, 0x77, 0xc0, 0x22, 0x35, 0x6d, 0x4c, 0x3d,
-	0x75, 0x81, 0x67, 0xef, 0x66, 0x55, 0x9c, 0xc4, 0xaa, 0xd9, 0xb7, 0xab, 0xec, 0xb4, 0x56, 0x4f,
-	0xdf, 0xab, 0x1e, 0x32, 0x8a, 0xfa, 0x4a, 0xb0, 0xaf, 0x80, 0x41, 0x0f, 0x7e, 0xe1, 0x01, 0x58,
-	0x74, 0xcc, 0x36, 0x72, 0x3c, 0x75, 0x91, 0x8b, 0xd8, 0x4a, 0xdf, 0x4c, 0xf5, 0x11, 0x27, 0xda,
-	0xc3, 0xd4, 0x1d, 0xd4, 0x2f, 0x8f, 0x86, 0x5a, 0x51, 0x70, 0x49, 0x86, 0x05, 0x72, 0xa0, 0x01,
-	0x56, 0x29, 0xa1, 0xa6, 0x63, 0x84, 0x27, 0xdf, 0x53, 0x97, 0x5e, 0xed, 0x3c, 0x70, 0xf6, 0x10,
-	0xe5, 0xe9, 0xb1, 0x35, 0xfc, 0xab, 0x02, 0x6e, 0x9a, 0x8e, 0x43, 0x3a, 0x26, 0x35, 0xdb, 0x0e,
-	0x32, 0xda, 0x03, 0xa3, 0xef, 0xda, 0xc4, 0xb5, 0xe9, 0xc0, 0x30, 0xb1, 0x35, 0xd6, 0xab, 0xe6,
-	0xf8, 0x8e, 0x7e, 0x30, 0x65, 0x47, 0x3b, 0x91, 0x88, 0xfa, 0xe0, 0x20, 0x10, 0xb0, 0x83, 0xad,
-	0x50, 0x91, 0xd8, 0xeb, 0x76, 0x60, 0x54, 0xc5, 0x9c, 0x41, 0xae, 0xcf, 0xa4, 0x80, 0x2e, 0x58,
-	0xf7, 0xa8, 0x49, 0xb9, 0xc5, 0xc1, 0x31, 0x63, 0x11, 0xcf, 0x73, 0x33, 0xdf, 0x99, 0x62, 0x66,
-	0x8b, 0x71, 0xd4, 0x07, 0xe2, 0x6c, 0x35, 0x2c, 0x61, 0xd5, 0xd5, 0xc0, 0xaa, 0x55, 0x6f, 0x12,
-	0xab, 0xc7, 0x01, 0xd0, 0x07, 0xeb, 0x81, 0x5d, 0xc8, 0x0a, 0xf5, 0xda, 0x96, 0x0a, 0xb8, 0xce,
-	0x5b, 0x67, 0xbb, 0x06, 0x59, 0x5c, 0x50, 0xa8, 0x54, 0x0d, 0x94, 0x16, 0xcd, 0x18, 0x5a, 0x4f,
-	0x40, 0x20, 0x05, 0x70, 0x42, 0xed, 0x33, 0x1f, 0xf9, 0x48, 0x2d, 0x9c, 0x57, 0xeb, 0x13, 0x46,
-	0x3e, 0x5d, 0x2b, 0x47, 0xeb, 0x09, 0x08, 0xdb, 0x2c, 0x3a, 0xb5, 0x3b, 0x34, 0x2a, 0x63, 0x86,
-	0x6d, 0x79, 0xea, 0xca, 0x99, 0x6a, 0xf7, 0x04, 0x47, 0xe8, 0x31, 0x2f, 0xa6, 0x16, 0xc5, 0xd0,
-	0x7a, 0x02, 0x02, 0xbf, 0x54, 0x40, 0x19, 0x13, 0x6c, 0x98, 0x6e, 0xcf, 0xb4, 0x4c, 0x23, 0xda,
-	0x78, 0x74, 0x02, 0x2e, 0x71, 0x13, 0xbe, 0x37, 0xc5, 0x84, 0x26, 0xc1, 0x3b, 0x9c, 0x77, 0xec,
-	0x82, 0x71, 0xb6, 0x0b, 0x6b, 0xde, 0x0c, 0xac, 0x79, 0x03, 0x4f, 0xa7, 0xd4, 0xcf, 0x42, 0xc2,
-	0x1d, 0x70, 0xc9, 0xc7, 0x81, 0x76, 0x96, 0xa1, 0xea, 0x6a, 0x45, 0xd9, 0xce, 0xd5, 0xdf, 0x18,
-	0x0d, 0xb5, 0xab, 0x13, 0x08, 0xe9, 0x44, 0x4f, 0x72, 0xc0, 0x4f, 0x14, 0x70, 0x35, 0xdc, 0x91,
-	0xe1, 0x7b, 0x66, 0x17, 0x45, 0x91, 0x2d, 0xf2, 0xfd, 0x7d, 0x67, 0xca, 0xfe, 0x42, 0x33, 0x8e,
-	0x18, 0xd3, 0x44, 0x74, 0xb7, 0x46, 0x43, 0xad, 0xec, 0xa6, 0xa0, 0x25, 0x33, 0x2e, 0xa7, 0xe1,
-	0xd9, 0xad, 0xe5, 0xa2, 0x3e, 0x71, 0xa9, 0x8d, 0xbb, 0x46, 0x54, 0x92, 0xd7, 0xf8, 0x45, 0xc2,
-	0x6f, 0xad, 0x31, 0xba, 0x99, 0xac, 0xbf, 0x6b, 0x09, 0x64, 0xc9, 0x04, 0x05, 0xa9, 0xc8, 0xc1,
-	0x37, 0x41, 0xe6, 0x04, 0x0d, 0x82, 0xcb, 0x6b, 0x6d, 0x34, 0xd4, 0x2e, 0x9d, 0xa0, 0x81, 0x24,
-	0x81, 0x61, 0xe1, 0xdb, 0x60, 0xe1, 0xd4, 0x74, 0x7c, 0x14, 0xb4, 0x19, 0xbc, 0x4b, 0xe0, 0x00,
-	0xb9, 0x4b, 0xe0, 0x80, 0xbb, 0xf3, 0x77, 0x94, 0xd2, 0x1f, 0x15, 0xf0, 0xad, 0x73, 0x95, 0x1d,
-	0x59, 0xfb, 0xc2, 0x54, 0xed, 0x0d, 0x59, 0xfb, 0xec, 0xfa, 0x3a, 0xcb, 0xba, 0x5f, 0x2b, 0xe0,
-	0x72, 0x5a, 0xb5, 0x39, 0x9f, 0x2b, 0xee, 0xcb, 0xc6, 0xac, 0xdc, 0xbe, 0x9e, 0x34, 0x46, 0x08,
-	0x15, 0x1a, 0x66, 0xd9, 0xf2, 0x89, 0x02, 0x36, 0x52, 0xab, 0xd0, 0xf9, 0x8c, 0xf9, 0x1f, 0x7b,
-	0x26, 0x66, 0x4d, 0x94, 0xbf, 0x17, 0x62, 0xcd, 0x09, 0xd8, 0x48, 0xad, 0x59, 0x5f, 0x23, 0x65,
-	0x73, 0x33, 0x95, 0xfd, 0x5e, 0x01, 0x95, 0x59, 0xe5, 0xe9, 0x42, 0xb2, 0xf5, 0x37, 0x0a, 0xd8,
-	0x9c, 0x5a, 0x57, 0x2e, 0x22, 0x2e, 0x5b, 0x7f, 0xca, 0x82, 0x5c, 0x58, 0x4d, 0x58, 0xeb, 0xdb,
-	0x10, 0xad, 0x6f, 0x56, 0xb4, 0xbe, 0x13, 0x4d, 0xdc, 0xfc, 0x44, 0xf3, 0x36, 0xff, 0x75, 0x9b,
-	0xb7, 0xc3, 0x71, 0xf3, 0x26, 0xa6, 0x97, 0xb7, 0xa6, 0x77, 0xa2, 0xaf, 0xd0, 0xc0, 0xfd, 0x52,
-	0x01, 0xd0, 0xc7, 0x1e, 0xa2, 0x0d, 0x6c, 0xa1, 0x8f, 0x90, 0x25, 0x38, 0xd5, 0x2c, 0x57, 0x71,
-	0xfb, 0x0c, 0x15, 0x47, 0x09, 0x26, 0xa1, 0xae, 0x32, 0x1a, 0x6a, 0xd7, 0x92, 0x12, 0x25, 0xd5,
-	0x29, 0xfa, 0xfe, 0x1f, 0xf5, 0xb8, 0x07, 0xae, 0x4e, 0xb1, 0xf9, 0x75, 0xa8, 0xdb, 0x7a, 0xbe,
-	0x08, 0x36, 0x79, 0x8e, 0xde, 0x73, 0x7c, 0x8f, 0x22, 0x77, 0x22, 0x7d, 0x61, 0x03, 0x2c, 0x75,
-	0x5c, 0xc4, 0x4e, 0x17, 0xd7, 0x7a, 0xf6, 0x98, 0xb2, 0x1e, 0x64, 0x44, 0xc8, 0xc2, 0xa7, 0x94,
-	0x70, 0xc1, 0xec, 0x12, 0xd7, 0xb2, 0x64, 0xd7, 0xb3, 0xd8, 0xad, 0x2a, 0x28, 0xd8, 0x60, 0x85,
-	0x82, 0xa9, 0xba, 0x61, 0xf1, 0x81, 0x26, 0x2f, 0x86, 0x8f, 0x08, 0x2a, 0x0f, 0x1f, 0x11, 0x14,
-	0xfe, 0x4e, 0x61, 0x37, 0x70, 0x50, 0x07, 0xa2, 0xab, 0x2c, 0xc8, 0x93, 0xdd, 0x64, 0x9e, 0x4c,
-	0xdd, 0xfa, 0xf8, 0x98, 0x49, 0x62, 0x44, 0xe6, 0x5c, 0x0f, 0xb6, 0x99, 0xaa, 0x48, 0xd1, 0xd3,
-	0xc0, 0xf0, 0x6f, 0x0a, 0xb8, 0x96, 0x02, 0xbf, 0xe7, 0x98, 0x9e, 0xd7, 0x34, 0xf9, 0xf4, 0xcc,
-	0x0c, 0x7c, 0xfc, 0x0d, 0x0d, 0x1c, 0xcb, 0x13, 0x96, 0xde, 0x0c, 0x2c, 0x3d, 0x53, 0xb5, 0x7e,
-	0x26, 0xb6, 0xf4, 0xa9, 0x02, 0xd4, 0x69, 0xae, 0xb8, 0x90, 0x1a, 0xfb, 0x07, 0x05, 0xdc, 0x98,
-	0xb9, 0xf5, 0x0b, 0xa9, 0xb5, 0x7f, 0xcf, 0x80, 0x52, 0x5a, 0xa4, 0x74, 0xde, 0xd6, 0x8d, 0x5f,
-	0x7f, 0x94, 0x19, 0xaf, 0x3f, 0xd2, 0x99, 0x9b, 0xff, 0x86, 0x67, 0xee, 0x53, 0x05, 0x14, 0xa5,
-	0xe8, 0xf2, 0x5c, 0x0a, 0xca, 0x72, 0x3d, 0xb9, 0xd9, 0xe9, 0xb6, 0xcb, 0xb9, 0x26, 0x35, 0xca,
-	0xe5, 0xd1, 0x50, 0x2b, 0xc5, 0xe5, 0x4b, 0xfb, 0x49, 0xe8, 0x2e, 0x7d, 0xa1, 0x80, 0x8d, 0x54,
-	0x59, 0xe7, 0x0b, 0xd8, 0x8f, 0x27, 0x03, 0xf6, 0xce, 0x2b, 0x1c, 0x97, 0x99, 0xd1, 0xfb, 0xd5,
-	0x3c, 0x58, 0x96, 0xc3, 0x0d, 0x3f, 0x00, 0xf9, 0x68, 0x56, 0x52, 0xb8, 0xd3, 0xde, 0x3d, 0x3b,
-	0x43, 0xaa, 0xb1, 0x09, 0x69, 0x2d, 0x08, 0x4e, 0x24, 0x47, 0x8f, 0xfe, 0x96, 0x3e, 0x57, 0xc0,
-	0xca, 0xf4, 0x9e, 0x65, 0xba, 0x13, 0x7e, 0x3a, 0xe9, 0x84, 0xaa, 0x74, 0x45, 0x8f, 0x5f, 0x3a,
-	0xab, 0xfd, 0x93, 0x2e, 0xbf, 0xb3, 0x43, 0x75, 0xd5, 0x27, 0xbe, 0x89, 0xa9, 0x4d, 0x07, 0x33,
-	0xfd, 0xf0, 0xf9, 0x02, 0x58, 0x7b, 0x40, 0xda, 0x2d, 0xb1, 0x51, 0x1b, 0x77, 0x1b, 0xf8, 0x98,
-	0xc0, 0xdb, 0x20, 0xe7, 0xd8, 0xc7, 0x88, 0xda, 0x3d, 0xc4, 0xcd, 0xbb, 0x24, 0x5e, 0x91, 0x42,
-	0x98, 0xfc, 0x8a, 0x14, 0xc2, 0xe0, 0x5d, 0xb0, 0x6c, 0x52, 0xa3, 0x47, 0x3c, 0x6a, 0x10, 0xdc,
-	0x09, 0x9b, 0x3b, 0x5e, 0xc8, 0x4d, 0xfa, 0x98, 0x78, 0x74, 0x1f, 0x77, 0x64, 0x4e, 0x10, 0x41,
-	0xe1, 0xf7, 0x41, 0xa1, 0xef, 0x22, 0x06, 0xb7, 0xd9, 0x60, 0x98, 0xe1, 0xac, 0x9b, 0xa3, 0xa1,
-	0xb6, 0x21, 0x81, 0x25, 0x5e, 0x99, 0x1a, 0xde, 0x07, 0xc5, 0x0e, 0xc1, 0x1d, 0xdf, 0x75, 0x11,
-	0xee, 0x0c, 0x0c, 0xcf, 0x3c, 0x16, 0xcf, 0x9f, 0xb9, 0xfa, 0xf5, 0xd1, 0x50, 0xdb, 0x94, 0x70,
-	0x2d, 0xf3, 0x58, 0x96, 0xb2, 0x1a, 0x43, 0xb1, 0x81, 0x6e, 0xfc, 0x8c, 0xd3, 0x61, 0x15, 0xc6,
-	0xe0, 0x2f, 0x83, 0x8b, 0xd1, 0x40, 0xd7, 0x8f, 0xd7, 0x1f, 0x79, 0xa0, 0x4b, 0x20, 0x61, 0x0b,
-	0x14, 0x3c, 0xbf, 0xdd, 0xb3, 0xa9, 0xc1, 0x5d, 0xb9, 0x34, 0xf3, 0x80, 0x87, 0x0f, 0x50, 0x40,
-	0xb0, 0x8d, 0x1f, 0x4c, 0xa5, 0x35, 0x0b, 0x4e, 0xa8, 0x49, 0xcd, 0x45, 0xc1, 0x09, 0x61, 0x72,
-	0x70, 0x42, 0x18, 0xfc, 0x05, 0x58, 0x17, 0x29, 0x6c, 0xb8, 0xe8, 0x99, 0x6f, 0xbb, 0xa8, 0x87,
-	0xa2, 0x37, 0xbb, 0x9b, 0xc9, 0x3c, 0xdf, 0xe7, 0xbf, 0xba, 0x44, 0x2b, 0x5a, 0x28, 0x92, 0x80,
-	0xcb, 0x2d, 0x54, 0x12, 0x0b, 0x6b, 0x60, 0xe9, 0x14, 0xb9, 0x9e, 0x4d, 0xb0, 0x9a, 0xe7, 0xb6,
-	0x6e, 0x8c, 0x86, 0xda, 0x5a, 0x00, 0x92, 0x78, 0x43, 0xaa, 0xbb, 0xd9, 0x2f, 0xbe, 0xd4, 0x94,
-	0xad, 0xdf, 0x2a, 0x00, 0x26, 0x6d, 0x80, 0x0e, 0x58, 0xed, 0x13, 0x4b, 0x06, 0x05, 0x8d, 0xca,
-	0x8d, 0xe4, 0x16, 0x0e, 0x26, 0x09, 0x45, 0x32, 0xc4, 0xb8, 0x23, 0x03, 0xee, 0xcf, 0xe9, 0x71,
-	0xd1, 0xf5, 0x15, 0xb0, 0x2c, 0x7b, 0x6b, 0xeb, 0x1f, 0x4b, 0x60, 0x35, 0x26, 0x15, 0x7a, 0xe2,
-	0xed, 0xb4, 0x85, 0x1c, 0xd4, 0xa1, 0xc4, 0x0d, 0x2a, 0xc7, 0xfb, 0x33, 0xcd, 0xe1, 0x2d, 0x6b,
-	0xc8, 0x25, 0xea, 0x47, 0x69, 0x34, 0xd4, 0xae, 0xc8, 0xc2, 0x24, 0xf7, 0x4c, 0x28, 0x81, 0x07,
-	0x20, 0x67, 0x1e, 0x1f, 0xdb, 0x98, 0x65, 0x80, 0x28, 0x0b, 0xd7, 0xd2, 0x3a, 0xf7, 0x9d, 0x80,
-	0x46, 0xe4, 0x47, 0xc8, 0x21, 0xe7, 0x47, 0x08, 0x83, 0x47, 0xa0, 0x40, 0x89, 0x83, 0x5c, 0x93,
-	0xda, 0x04, 0x87, 0xbd, 0x7c, 0x39, 0x75, 0x1c, 0x18, 0x93, 0x8d, 0x6f, 0x23, 0x99, 0x55, 0x97,
-	0x17, 0x90, 0x80, 0x82, 0x89, 0x31, 0xa1, 0x81, 0xd8, 0xa5, 0x69, 0xfd, 0x7b, 0xdc, 0x39, 0x3b,
-	0x11, 0x93, 0xf0, 0x0d, 0xaf, 0x05, 0x92, 0x28, 0xb9, 0x16, 0x48, 0xe0, 0x89, 0xb3, 0x91, 0xe5,
-	0x7d, 0xca, 0xec, 0xb3, 0xf1, 0x00, 0x14, 0xc3, 0x72, 0x42, 0xf0, 0x01, 0x71, 0xec, 0xce, 0x80,
-	0x7f, 0xde, 0xc8, 0x8b, 0x1b, 0x2f, 0x8e, 0x93, 0x6f, 0xbc, 0x38, 0x0e, 0x7e, 0x0c, 0xc6, 0x4f,
-	0x45, 0x13, 0x59, 0xba, 0xc8, 0xa3, 0xb4, 0x9d, 0xe6, 0x50, 0x3d, 0x85, 0xbe, 0x7e, 0x2d, 0x70,
-	0x6d, 0xaa, 0x34, 0x3d, 0x15, 0x0a, 0x5b, 0x60, 0xbd, 0x63, 0x32, 0xcf, 0x46, 0xc5, 0xfc, 0x21,
-	0x12, 0x25, 0x62, 0xb9, 0x7e, 0x63, 0x34, 0xd4, 0xae, 0xa7, 0xa0, 0xa5, 0xdd, 0xa4, 0x71, 0x97,
-	0xba, 0x60, 0x2d, 0x91, 0xa9, 0xaf, 0x65, 0x10, 0x3a, 0x06, 0xc5, 0x78, 0xd4, 0x5f, 0xcb, 0x04,
-	0xf4, 0x67, 0x05, 0x6c, 0x1e, 0xf8, 0x8e, 0x67, 0xba, 0xad, 0x30, 0x0b, 0x1f, 0x90, 0xf6, 0x2e,
-	0xa2, 0xa6, 0xed, 0x78, 0x4c, 0x18, 0x7f, 0xe8, 0x09, 0x74, 0x72, 0x61, 0x1c, 0x20, 0x0b, 0x13,
-	0xef, 0xcb, 0x6f, 0x83, 0x85, 0x27, 0xf1, 0x09, 0x27, 0xde, 0x12, 0x09, 0x0a, 0x78, 0x0b, 0x2c,
-	0xb2, 0x3b, 0x16, 0xd1, 0x60, 0xba, 0xe1, 0xc3, 0xaf, 0x80, 0xc8, 0xc3, 0xaf, 0x80, 0x7c, 0x7b,
-	0x1f, 0x14, 0xa4, 0x77, 0x2a, 0x58, 0x00, 0x4b, 0x47, 0xcd, 0x87, 0xcd, 0xfd, 0x9f, 0x34, 0x8b,
-	0x73, 0x6c, 0x71, 0xb0, 0xd7, 0xdc, 0x6d, 0x34, 0x7f, 0x54, 0x54, 0xd8, 0x42, 0x3f, 0x6a, 0x36,
-	0xd9, 0x62, 0x1e, 0x5e, 0x02, 0xf9, 0xd6, 0xd1, 0xbd, 0x7b, 0x7b, 0x7b, 0xbb, 0x7b, 0xbb, 0xc5,
-	0x0c, 0x04, 0x60, 0xf1, 0x87, 0x3b, 0x8d, 0x47, 0x7b, 0xbb, 0xc5, 0x6c, 0xfd, 0xe7, 0xcf, 0x5f,
-	0x94, 0x95, 0xaf, 0x5e, 0x94, 0x95, 0x7f, 0xbf, 0x28, 0x2b, 0x9f, 0xbd, 0x2c, 0xcf, 0x7d, 0xf5,
-	0xb2, 0x3c, 0xf7, 0xcf, 0x97, 0xe5, 0xb9, 0x9f, 0xdd, 0x93, 0x3e, 0x80, 0x8a, 0xa7, 0xe3, 0xbe,
-	0x4b, 0xd8, 0x91, 0x0c, 0x56, 0xb5, 0x73, 0x7c, 0xe9, 0x6d, 0x2f, 0xf2, 0x7b, 0xec, 0xfd, 0xff,
-	0x06, 0x00, 0x00, 0xff, 0xff, 0xff, 0x66, 0x0a, 0x62, 0x17, 0x1e, 0x00, 0x00,
+	// 2186 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xc4, 0x19, 0x4d, 0x73, 0xdb, 0xc6,
+	0x55, 0x10, 0x29, 0x89, 0x5c, 0xea, 0x83, 0x5a, 0x7d, 0x18, 0xa2, 0x6d, 0x82, 0x56, 0xdc, 0x8c,
+	0xd2, 0x38, 0x64, 0xa3, 0x74, 0xa6, 0x1e, 0xb5, 0x17, 0x51, 0x52, 0x6b, 0xca, 0x36, 0x25, 0x83,
+	0x52, 0x3b, 0xed, 0x4c, 0x83, 0x01, 0x89, 0x15, 0x8d, 0x08, 0xc4, 0xd2, 0xc0, 0x42, 0x0d, 0x73,
+	0x6e, 0x0f, 0x9d, 0xcc, 0xa4, 0x99, 0x4e, 0xda, 0x66, 0xa6, 0x33, 0xed, 0xe4, 0x96, 0x5f, 0xd0,
+	0x1e, 0x7a, 0xeb, 0xc9, 0xc7, 0x1c, 0x7b, 0x62, 0x3b, 0xf6, 0x8d, 0xbf, 0xa2, 0xb3, 0xbb, 0x00,
+	0xb1, 0x04, 0x40, 0x51, 0x8e, 0xeb, 0xea, 0x44, 0xee, 0xfb, 0xde, 0xf7, 0xde, 0xbe, 0x7d, 0x6f,
+	0x01, 0x76, 0x4c, 0x9b, 0x20, 0xc7, 0xd6, 0xad, 0x8a, 0xdb, 0x7a, 0x8a, 0x0c, 0xcf, 0x42, 0x4e,
+	0xf8, 0x0f, 0x37, 0x3f, 0x42, 0x2d, 0xe2, 0xc6, 0x00, 0xe5, 0xae, 0x83, 0x09, 0x86, 0xf9, 0x28,
+	0xbc, 0xa0, 0xb4, 0x31, 0x6e, 0x5b, 0xa8, 0xc2, 0xf0, 0x4d, 0xef, 0xac, 0x42, 0xcc, 0x0e, 0x72,
+	0x89, 0xde, 0xe9, 0x72, 0x96, 0xc2, 0xe6, 0xf9, 0x7d, 0xb7, 0x6c, 0xe2, 0x8a, 0xde, 0x35, 0x2b,
+	0x2d, 0xec, 0xa0, 0xca, 0xc5, 0xfb, 0x95, 0x36, 0xb2, 0x91, 0xa3, 0x13, 0x64, 0xf8, 0x34, 0xdf,
+	0x0f, 0x69, 0x3a, 0x7a, 0xeb, 0xa9, 0x69, 0x23, 0xa7, 0x57, 0xe9, 0x9e, 0xb7, 0x19, 0x93, 0x83,
+	0x5c, 0xec, 0x39, 0x2d, 0x14, 0xe3, 0x7a, 0xaf, 0x6d, 0x92, 0xa7, 0x5e, 0xb3, 0xdc, 0xc2, 0x9d,
+	0x4a, 0x1b, 0xb7, 0x71, 0x68, 0x03, 0x5d, 0xb1, 0x05, 0xfb, 0xc7, 0xc9, 0x37, 0xbf, 0x4e, 0x81,
+	0xcc, 0xc1, 0xc7, 0xa8, 0xe5, 0x11, 0xec, 0xc0, 0x12, 0x98, 0x36, 0x0d, 0x59, 0x2a, 0x49, 0x5b,
+	0xd9, 0x6a, 0x7e, 0xd0, 0x57, 0xe6, 0x4d, 0xe3, 0x1e, 0xee, 0x98, 0x04, 0x75, 0xba, 0xa4, 0xa7,
+	0x4e, 0x9b, 0x06, 0x7c, 0x1b, 0xa4, 0xbb, 0x18, 0x5b, 0xf2, 0x34, 0xa3, 0x81, 0x83, 0xbe, 0xb2,
+	0x48, 0xd7, 0x02, 0x15, 0xc3, 0xc3, 0x5d, 0x30, 0x63, 0x63, 0x03, 0xb9, 0x72, 0xaa, 0x94, 0xda,
+	0xca, 0x6d, 0xaf, 0x97, 0x63, 0xae, 0xab, 0x63, 0x03, 0x55, 0x57, 0x06, 0x7d, 0x65, 0x89, 0x11,
+	0x0a, 0x12, 0x38, 0x27, 0xfc, 0x10, 0x2c, 0x76, 0x4c, 0xdb, 0xec, 0x78, 0x9d, 0x43, 0xdc, 0x6c,
+	0x98, 0x9f, 0x20, 0x39, 0x5d, 0x92, 0xb6, 0x72, 0xdb, 0xc5, 0xb8, 0x2c, 0xd5, 0x77, 0xc6, 0x23,
+	0xd3, 0x25, 0xd5, 0xf5, 0xe7, 0x7d, 0x65, 0x8a, 0x1a, 0x36, 0xca, 0xad, 0x46, 0xd6, 0x54, 0xbe,
+	0xa5, 0xbb, 0xe4, 0xb4, 0x6b, 0xe8, 0x04, 0x9d, 0x98, 0x1d, 0x24, 0xcf, 0x30, 0xf9, 0x85, 0x32,
+	0x0f, 0x5e, 0x39, 0x70, 0x5c, 0xf9, 0x24, 0x08, 0x5e, 0xb5, 0x10, 0xc8, 0x1e, 0xe5, 0xfc, 0xfc,
+	0xdf, 0x8a, 0xa4, 0x46, 0x60, 0xf0, 0x08, 0xac, 0x78, 0xb6, 0xee, 0xba, 0x66, 0xdb, 0x46, 0x86,
+	0xf6, 0x11, 0x6e, 0x6a, 0x8e, 0x67, 0xbb, 0x72, 0xb6, 0x94, 0xda, 0xca, 0x56, 0x95, 0x41, 0x5f,
+	0xb9, 0x19, 0xa2, 0x0f, 0x71, 0x53, 0xf5, 0x6c, 0xd1, 0x09, 0xcb, 0x31, 0xe4, 0xe6, 0xd7, 0xeb,
+	0x20, 0x4d, 0xbd, 0x76, 0xb5, 0x30, 0xd9, 0x7a, 0x07, 0xc9, 0xf3, 0x61, 0x98, 0xe8, 0x5a, 0x0c,
+	0x13, 0x5d, 0xc3, 0x6d, 0x90, 0x41, 0x7e, 0xf0, 0xe5, 0x15, 0x46, 0xbb, 0x3e, 0xe8, 0x2b, 0x30,
+	0x80, 0x09, 0xf4, 0x43, 0x3a, 0x78, 0x1f, 0x00, 0x1a, 0xa0, 0xfd, 0xe6, 0x43, 0xd4, 0x73, 0x65,
+	0x58, 0x4a, 0x6d, 0xcd, 0x57, 0xe5, 0x41, 0x5f, 0x59, 0x0d, 0xa1, 0x02, 0x9f, 0x40, 0x0b, 0x1f,
+	0x83, 0x2c, 0xf5, 0x91, 0xe6, 0x22, 0x64, 0xb3, 0x0c, 0xba, 0xdc, 0xd9, 0xab, 0xbe, 0xb3, 0x33,
+	0x94, 0xa9, 0x81, 0x90, 0xcd, 0xdc, 0x3c, 0x5c, 0xc1, 0x23, 0x90, 0xa5, 0xc2, 0x35, 0xd2, 0xeb,
+	0x22, 0x39, 0xe5, 0x8b, 0x4b, 0xcc, 0xb3, 0x93, 0x5e, 0x17, 0xf1, 0x9d, 0xd9, 0xfe, 0x4a, 0xdc,
+	0x59, 0x00, 0x83, 0x3b, 0x60, 0x7e, 0x28, 0x50, 0x33, 0x0d, 0x96, 0x6f, 0xe9, 0x70, 0x6f, 0x94,
+	0xa6, 0x66, 0x44, 0xf7, 0xc6, 0xa1, 0x70, 0x17, 0xcc, 0x12, 0xdd, 0xb4, 0x89, 0x2b, 0xcf, 0xb0,
+	0x8c, 0xdf, 0x28, 0xf3, 0xd3, 0x5b, 0xd6, 0xbb, 0x66, 0x99, 0x9e, 0xf0, 0xf2, 0xc5, 0xfb, 0xe5,
+	0x13, 0x4a, 0x51, 0x5d, 0xf4, 0xf7, 0xe5, 0x33, 0xa8, 0xfe, 0x2f, 0x3c, 0x06, 0xb3, 0x96, 0xde,
+	0x44, 0x96, 0x2b, 0xcf, 0x32, 0x11, 0x9b, 0xc9, 0x9b, 0x29, 0x3f, 0x62, 0x44, 0x07, 0x36, 0x71,
+	0x7a, 0xd5, 0xd5, 0x41, 0x5f, 0xc9, 0x73, 0x2e, 0xc1, 0x30, 0x5f, 0x0e, 0xd4, 0xc0, 0x12, 0xc1,
+	0x44, 0xb7, 0xb4, 0xa0, 0x5a, 0xb8, 0xf2, 0xdc, 0xab, 0x9d, 0x21, 0xc6, 0x1e, 0xa0, 0x5c, 0x35,
+	0xb2, 0x86, 0x7f, 0x93, 0xc0, 0x5d, 0xdd, 0xb2, 0x70, 0x4b, 0x27, 0x7a, 0xd3, 0x42, 0x5a, 0xb3,
+	0xa7, 0x75, 0x1d, 0x13, 0x3b, 0x26, 0xe9, 0x69, 0xba, 0x6d, 0x0c, 0xf5, 0xca, 0x19, 0xb6, 0xa3,
+	0x1f, 0x8d, 0xd9, 0xd1, 0x6e, 0x28, 0xa2, 0xda, 0x3b, 0xf6, 0x05, 0xec, 0xda, 0x46, 0xa0, 0x88,
+	0xef, 0x75, 0xcb, 0x37, 0xaa, 0xa4, 0x4f, 0x20, 0x57, 0x27, 0x52, 0x40, 0x07, 0xac, 0xb8, 0x44,
+	0x27, 0xcc, 0x62, 0xff, 0x68, 0xd2, 0x88, 0x67, 0x99, 0x99, 0xef, 0x8e, 0x31, 0xb3, 0x41, 0x39,
+	0xaa, 0x3d, 0x7e, 0x1e, 0x6b, 0x06, 0xb7, 0xea, 0x86, 0x6f, 0xd5, 0x92, 0x3b, 0x8a, 0x55, 0xa3,
+	0x00, 0xe8, 0x81, 0x15, 0xdf, 0x2e, 0x64, 0x04, 0x7a, 0x4d, 0x43, 0x06, 0x4c, 0xe7, 0xbd, 0xcb,
+	0x5d, 0x83, 0x0c, 0x26, 0x28, 0x50, 0x2a, 0xfb, 0x4a, 0xf3, 0x7a, 0x04, 0xad, 0xc6, 0x20, 0x90,
+	0x00, 0x38, 0xa2, 0xf6, 0x99, 0x87, 0x3c, 0x24, 0xe7, 0xae, 0xaa, 0xf5, 0x09, 0x25, 0x1f, 0xaf,
+	0x95, 0xa1, 0xd5, 0x18, 0x84, 0x6e, 0x16, 0x5d, 0x98, 0x2d, 0x12, 0x96, 0x3e, 0xcd, 0x34, 0x5c,
+	0x79, 0xf1, 0x52, 0xb5, 0x07, 0x9c, 0x23, 0xf0, 0x98, 0x1b, 0x51, 0x8b, 0x22, 0x68, 0x35, 0x06,
+	0x81, 0x5f, 0x49, 0xa0, 0x68, 0x63, 0x5b, 0xd3, 0x9d, 0x8e, 0x6e, 0xe8, 0x5a, 0xb8, 0xf1, 0xf0,
+	0x04, 0x2c, 0x30, 0x13, 0x7e, 0x30, 0xc6, 0x84, 0x3a, 0xb6, 0x77, 0x19, 0xef, 0xd0, 0x05, 0xc3,
+	0x6c, 0xe7, 0xd6, 0xbc, 0xe5, 0x5b, 0x73, 0xd3, 0x1e, 0x4f, 0xa9, 0x5e, 0x86, 0x84, 0xbb, 0x60,
+	0xc1, 0xb3, 0x7d, 0xed, 0x34, 0x43, 0xe5, 0xa5, 0x92, 0xb4, 0x95, 0xa9, 0xde, 0x1c, 0xf4, 0x95,
+	0x1b, 0x23, 0x08, 0xe1, 0x44, 0x8f, 0x72, 0xc0, 0x4f, 0x25, 0x70, 0x23, 0xd8, 0x91, 0xe6, 0xb9,
+	0x7a, 0x1b, 0x85, 0x91, 0xcd, 0xb3, 0xfd, 0x7d, 0x6f, 0xcc, 0xfe, 0x02, 0x33, 0x4e, 0x29, 0xd3,
+	0x48, 0x74, 0x37, 0x07, 0x7d, 0xa5, 0xe8, 0x24, 0xa0, 0x05, 0x33, 0x56, 0x93, 0xf0, 0xf4, 0xa6,
+	0x73, 0x50, 0x17, 0x3b, 0xc4, 0xb4, 0xdb, 0x5a, 0x58, 0x92, 0x97, 0xd9, 0x85, 0xc2, 0x6e, 0xba,
+	0x21, 0xba, 0x1e, 0xaf, 0xbf, 0xcb, 0x31, 0x64, 0x41, 0x07, 0x39, 0xa1, 0xc8, 0xc1, 0xb7, 0x40,
+	0xea, 0x1c, 0xf5, 0xfc, 0x0b, 0x6f, 0x79, 0xd0, 0x57, 0x16, 0xce, 0x51, 0x4f, 0x90, 0x40, 0xb1,
+	0xf0, 0x1d, 0x30, 0x73, 0xa1, 0x5b, 0x1e, 0xf2, 0x5b, 0x13, 0xd6, 0x59, 0x30, 0x80, 0xd8, 0x59,
+	0x30, 0xc0, 0xce, 0xf4, 0x7d, 0xa9, 0xf0, 0x67, 0x09, 0x7c, 0xe7, 0x4a, 0x65, 0x47, 0xd4, 0x3e,
+	0x33, 0x56, 0x7b, 0x4d, 0xd4, 0x3e, 0xb9, 0xbe, 0x4e, 0xb2, 0xee, 0xb7, 0x12, 0x58, 0x4d, 0xaa,
+	0x36, 0x57, 0x73, 0xc5, 0x03, 0xd1, 0x98, 0xc5, 0xed, 0xdb, 0x71, 0x63, 0xb8, 0x50, 0xae, 0x61,
+	0x92, 0x2d, 0x9f, 0x4a, 0x60, 0x2d, 0xb1, 0x0a, 0x5d, 0xcd, 0x98, 0xff, 0xb1, 0x67, 0x22, 0xd6,
+	0x84, 0xf9, 0x7b, 0x2d, 0xd6, 0x9c, 0x83, 0xb5, 0xc4, 0x9a, 0xf5, 0x2d, 0x52, 0x36, 0x33, 0x51,
+	0xd9, 0x1f, 0x25, 0x50, 0x9a, 0x54, 0x9e, 0xae, 0x25, 0x5b, 0x7f, 0x27, 0x81, 0x8d, 0xb1, 0x75,
+	0xe5, 0x3a, 0xe2, 0xb2, 0xf9, 0x97, 0x34, 0xc8, 0x04, 0xd5, 0x84, 0xb6, 0xcb, 0x35, 0xde, 0x2e,
+	0xa7, 0x79, 0xbb, 0x3c, 0xd2, 0xc4, 0x4d, 0x8f, 0x34, 0x6f, 0xd3, 0xdf, 0xb6, 0x79, 0x3b, 0x19,
+	0x36, 0x6f, 0x7c, 0xe2, 0x79, 0x7b, 0x7c, 0x27, 0xfa, 0x0a, 0x0d, 0xdc, 0xaf, 0x25, 0x00, 0x3d,
+	0xdb, 0x45, 0xa4, 0x66, 0x1b, 0xe8, 0x63, 0x64, 0x70, 0x4e, 0x39, 0xcd, 0x54, 0x6c, 0x5f, 0xa2,
+	0xe2, 0x34, 0xc6, 0xc4, 0xd5, 0x95, 0x06, 0x7d, 0xe5, 0x56, 0x5c, 0xa2, 0xa0, 0x3a, 0x41, 0xdf,
+	0xff, 0xa3, 0x1e, 0x77, 0xc0, 0x8d, 0x31, 0x36, 0xbf, 0x09, 0x75, 0x9b, 0xcf, 0x67, 0xc1, 0x06,
+	0xcb, 0xd1, 0x3d, 0xcb, 0x73, 0x09, 0x72, 0x46, 0xd2, 0x17, 0xd6, 0xc0, 0x5c, 0xcb, 0x41, 0xf4,
+	0x74, 0x31, 0xad, 0x97, 0x8f, 0x29, 0x2b, 0x7e, 0x46, 0x04, 0x2c, 0x6c, 0x4a, 0x09, 0x16, 0xd4,
+	0x2e, 0x7e, 0x2d, 0x0b, 0x76, 0x3d, 0x8b, 0xdc, 0xaa, 0x9c, 0x82, 0x0e, 0x56, 0xc1, 0x90, 0x55,
+	0x33, 0xd8, 0x40, 0x93, 0xe5, 0xc3, 0x47, 0x08, 0x15, 0x87, 0x8f, 0x10, 0x0a, 0xff, 0x20, 0xd1,
+	0x1b, 0xd8, 0xaf, 0x03, 0xe1, 0x55, 0xe6, 0xe7, 0xc9, 0x7e, 0x3c, 0x4f, 0xc6, 0x6e, 0x7d, 0x78,
+	0xcc, 0x04, 0x31, 0x3c, 0x73, 0x6e, 0xfb, 0xdb, 0x4c, 0x54, 0x24, 0xa9, 0x49, 0x60, 0xf8, 0x77,
+	0x09, 0xdc, 0x4a, 0x80, 0xef, 0x59, 0xba, 0xeb, 0xd6, 0x75, 0x36, 0x71, 0x53, 0x03, 0x1f, 0xbf,
+	0xa6, 0x81, 0x43, 0x79, 0xdc, 0xd2, 0xbb, 0xbe, 0xa5, 0x97, 0xaa, 0x56, 0x2f, 0xc5, 0x16, 0x3e,
+	0x93, 0x80, 0x3c, 0xce, 0x15, 0xd7, 0x52, 0x63, 0xff, 0x24, 0x81, 0x3b, 0x13, 0xb7, 0x7e, 0x2d,
+	0xb5, 0xf6, 0x1f, 0x29, 0x50, 0x48, 0x8a, 0x94, 0xca, 0xda, 0xba, 0xe1, 0x8b, 0x91, 0x34, 0xe1,
+	0xc5, 0x48, 0x38, 0x73, 0xd3, 0xaf, 0x79, 0xe6, 0x3e, 0x93, 0x40, 0x5e, 0x88, 0x2e, 0xcb, 0x25,
+	0xbf, 0x2c, 0x57, 0xe3, 0x9b, 0x1d, 0x6f, 0xbb, 0x98, 0x6b, 0x42, 0xa3, 0x5c, 0x1c, 0xf4, 0x95,
+	0x42, 0x54, 0xbe, 0xb0, 0x9f, 0x98, 0xee, 0xc2, 0x97, 0x12, 0x58, 0x4b, 0x94, 0x75, 0xb5, 0x80,
+	0xfd, 0x74, 0x34, 0x60, 0xef, 0xbe, 0xc2, 0x71, 0x99, 0x18, 0xbd, 0xdf, 0x4c, 0x83, 0x79, 0x31,
+	0xdc, 0xf0, 0x43, 0x90, 0x0d, 0x67, 0x25, 0x89, 0x39, 0xed, 0xbd, 0xcb, 0x33, 0xa4, 0x1c, 0x99,
+	0x90, 0x96, 0xfd, 0xe0, 0x84, 0x72, 0xd4, 0xf0, 0x6f, 0xe1, 0x0b, 0x09, 0x2c, 0x8e, 0xef, 0x59,
+	0xc6, 0x3b, 0xe1, 0xe7, 0xa3, 0x4e, 0x28, 0x0b, 0x57, 0xf4, 0xf0, 0x75, 0xb4, 0xdc, 0x3d, 0x6f,
+	0xb3, 0x3b, 0x3b, 0x50, 0x57, 0x7e, 0xe2, 0xe9, 0x36, 0x31, 0x49, 0x6f, 0xa2, 0x1f, 0xbe, 0x98,
+	0x01, 0xcb, 0x87, 0xb8, 0xd9, 0xe0, 0x1b, 0x35, 0xed, 0x76, 0xcd, 0x3e, 0xc3, 0x70, 0x1b, 0x64,
+	0x2c, 0xf3, 0x0c, 0x11, 0xb3, 0x83, 0x98, 0x79, 0x0b, 0xfc, 0x15, 0x29, 0x80, 0x89, 0xaf, 0x48,
+	0x01, 0x0c, 0xee, 0x80, 0x79, 0x9d, 0x68, 0x1d, 0xec, 0x12, 0x0d, 0xdb, 0xad, 0xa0, 0xb9, 0x63,
+	0x85, 0x5c, 0x27, 0x8f, 0xb1, 0x4b, 0x8e, 0xec, 0x96, 0xc8, 0x09, 0x42, 0x28, 0xfc, 0x21, 0xc8,
+	0x75, 0x1d, 0x44, 0xe1, 0x26, 0x1d, 0x0c, 0x53, 0x8c, 0x75, 0x63, 0xd0, 0x57, 0xd6, 0x04, 0xb0,
+	0xc0, 0x2b, 0x52, 0xc3, 0x07, 0x20, 0xdf, 0xc2, 0x76, 0xcb, 0x73, 0x1c, 0x64, 0xb7, 0x7a, 0x9a,
+	0xab, 0x9f, 0xf1, 0x27, 0xd3, 0x4c, 0xf5, 0xf6, 0xa0, 0xaf, 0x6c, 0x08, 0xb8, 0x86, 0x7e, 0x26,
+	0x4a, 0x59, 0x8a, 0xa0, 0xe8, 0x40, 0x37, 0x7c, 0xc6, 0x69, 0xd1, 0x0a, 0xa3, 0xb1, 0xd7, 0xc4,
+	0xd9, 0x70, 0xa0, 0xeb, 0x46, 0xeb, 0x8f, 0x38, 0xd0, 0xc5, 0x90, 0xb0, 0x01, 0x72, 0xae, 0xd7,
+	0xec, 0x98, 0x44, 0x63, 0xae, 0x9c, 0x9b, 0x78, 0xc0, 0x83, 0x07, 0x28, 0xc0, 0xd9, 0x86, 0x8f,
+	0xac, 0xc2, 0x9a, 0x06, 0x27, 0xd0, 0x24, 0x67, 0xc2, 0xe0, 0x04, 0x30, 0x31, 0x38, 0x01, 0x0c,
+	0xfe, 0x0a, 0xac, 0xf0, 0x14, 0xd6, 0x1c, 0xf4, 0xcc, 0x33, 0x1d, 0xd4, 0x41, 0xe1, 0x9b, 0xdd,
+	0xdd, 0x78, 0x9e, 0x1f, 0xb1, 0x5f, 0x55, 0xa0, 0xe5, 0x2d, 0x14, 0x8e, 0xc1, 0xc5, 0x16, 0x2a,
+	0x8e, 0x85, 0x15, 0x30, 0x77, 0x81, 0x1c, 0xd7, 0xc4, 0xb6, 0x9c, 0x65, 0xb6, 0xae, 0x0d, 0xfa,
+	0xca, 0xb2, 0x0f, 0x12, 0x78, 0x03, 0xaa, 0x9d, 0xf4, 0x97, 0x5f, 0x29, 0xd2, 0xe6, 0xef, 0x25,
+	0x00, 0xe3, 0x36, 0x40, 0x0b, 0x2c, 0x75, 0xb1, 0x21, 0x82, 0xfc, 0x46, 0xe5, 0x4e, 0x7c, 0x0b,
+	0xc7, 0xa3, 0x84, 0x3c, 0x19, 0x22, 0xdc, 0xa1, 0x01, 0x0f, 0xa6, 0xd4, 0xa8, 0xe8, 0xea, 0x22,
+	0x98, 0x17, 0xbd, 0xb5, 0xf9, 0xcf, 0x39, 0xb0, 0x14, 0x91, 0x0a, 0x5d, 0xfe, 0x76, 0xda, 0x40,
+	0x16, 0x6a, 0x11, 0xec, 0xf8, 0x95, 0xe3, 0x83, 0x89, 0xe6, 0xb0, 0x96, 0x35, 0xe0, 0xe2, 0xf5,
+	0xa3, 0x30, 0xe8, 0x2b, 0xeb, 0xa2, 0x30, 0xc1, 0x3d, 0x23, 0x4a, 0xe0, 0x31, 0xc8, 0xe8, 0x67,
+	0x67, 0xa6, 0x4d, 0x33, 0x80, 0x97, 0x85, 0x5b, 0x49, 0x9d, 0xfb, 0xae, 0x4f, 0xc3, 0xf3, 0x23,
+	0xe0, 0x10, 0xf3, 0x23, 0x80, 0xc1, 0x53, 0x90, 0x23, 0xd8, 0x42, 0x8e, 0x4e, 0x4c, 0x6c, 0x07,
+	0xbd, 0x7c, 0x31, 0x71, 0x1c, 0x18, 0x92, 0x0d, 0x6f, 0x23, 0x91, 0x55, 0x15, 0x17, 0x10, 0x83,
+	0x9c, 0x6e, 0xdb, 0x98, 0xf8, 0x62, 0xe7, 0xc6, 0xf5, 0xef, 0x51, 0xe7, 0xec, 0x86, 0x4c, 0xdc,
+	0x37, 0xac, 0x16, 0x08, 0xa2, 0xc4, 0x5a, 0x20, 0x80, 0x47, 0xce, 0x46, 0x9a, 0xf5, 0x29, 0x93,
+	0xcf, 0xc6, 0x21, 0xc8, 0x07, 0xe5, 0x04, 0xdb, 0xc7, 0xd8, 0x32, 0x5b, 0x3d, 0xf6, 0x49, 0x24,
+	0xcb, 0x6f, 0xbc, 0x28, 0x4e, 0xbc, 0xf1, 0xa2, 0x38, 0xf8, 0x09, 0x18, 0x3e, 0x15, 0x8d, 0x64,
+	0xe9, 0x2c, 0x8b, 0xd2, 0x56, 0x92, 0x43, 0xd5, 0x04, 0xfa, 0xea, 0x2d, 0xdf, 0xb5, 0x89, 0xd2,
+	0xd4, 0x44, 0x28, 0x6c, 0x80, 0x95, 0x96, 0x4e, 0x3d, 0x1b, 0x16, 0xf3, 0x87, 0x88, 0x97, 0x88,
+	0xf9, 0xea, 0x9d, 0x41, 0x5f, 0xb9, 0x9d, 0x80, 0x16, 0x76, 0x93, 0xc4, 0x5d, 0x68, 0x83, 0xe5,
+	0x58, 0xa6, 0xbe, 0x91, 0x41, 0xe8, 0x0c, 0xe4, 0xa3, 0x51, 0x7f, 0x23, 0x13, 0xd0, 0x5f, 0x25,
+	0xb0, 0x71, 0xec, 0x59, 0xae, 0xee, 0x34, 0x82, 0x2c, 0x3c, 0xc4, 0xcd, 0x7d, 0x44, 0x74, 0xd3,
+	0x72, 0xa9, 0x30, 0xf6, 0xd0, 0xe3, 0xeb, 0x64, 0xc2, 0x18, 0x40, 0x14, 0xc6, 0xdf, 0x97, 0xdf,
+	0x01, 0x33, 0x4f, 0xa2, 0x13, 0x4e, 0xb4, 0x25, 0xe2, 0x14, 0xf0, 0x1e, 0x98, 0xa5, 0x77, 0x2c,
+	0x22, 0xfe, 0x74, 0xc3, 0x86, 0x5f, 0x0e, 0x11, 0x87, 0x5f, 0x0e, 0xf9, 0xee, 0x11, 0xc8, 0x09,
+	0xef, 0x54, 0x30, 0x07, 0xe6, 0x4e, 0xeb, 0x0f, 0xeb, 0x47, 0x3f, 0xab, 0xe7, 0xa7, 0xe8, 0xe2,
+	0xf8, 0xa0, 0xbe, 0x5f, 0xab, 0xff, 0x24, 0x2f, 0xd1, 0x85, 0x7a, 0x5a, 0xaf, 0xd3, 0xc5, 0x34,
+	0x5c, 0x00, 0xd9, 0xc6, 0xe9, 0xde, 0xde, 0xc1, 0xc1, 0xfe, 0xc1, 0x7e, 0x3e, 0x05, 0x01, 0x98,
+	0xfd, 0xf1, 0x6e, 0xed, 0xd1, 0xc1, 0x7e, 0x3e, 0x5d, 0xfd, 0xe5, 0xf3, 0x17, 0x45, 0xe9, 0x9b,
+	0x17, 0x45, 0xe9, 0x3f, 0x2f, 0x8a, 0xd2, 0xe7, 0x2f, 0x8b, 0x53, 0xdf, 0xbc, 0x2c, 0x4e, 0xfd,
+	0xeb, 0x65, 0x71, 0xea, 0x17, 0x7b, 0xc2, 0x47, 0x53, 0xfe, 0x74, 0xdc, 0x75, 0x30, 0x3d, 0x92,
+	0xfe, 0xaa, 0x72, 0x85, 0xaf, 0xc3, 0xcd, 0x59, 0x76, 0x8f, 0x7d, 0xf0, 0xdf, 0x00, 0x00, 0x00,
+	0xff, 0xff, 0x3a, 0x96, 0x2c, 0x42, 0x4b, 0x1e, 0x00, 0x00,
 }
 
 func (m *Executor) Marshal() (dAtA []byte, err error) {
@@ -1253,6 +1263,15 @@ func (m *Node) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if len(m.Executor) > 0 {
+		i -= len(m.Executor)
+		copy(dAtA[i:], m.Executor)
+		i = encodeVarintSchedulerobjects(dAtA, i, uint64(len(m.Executor)))
+		i--
+		dAtA[i] = 0x1
+		i--
+		dAtA[i] = 0x9a
+	}
 	if len(m.NodeDbKeys) > 0 {
 		for iNdEx := len(m.NodeDbKeys) - 1; iNdEx >= 0; iNdEx-- {
 			i -= len(m.NodeDbKeys[iNdEx])
@@ -2280,6 +2299,10 @@ func (m *Node) Size() (n int) {
 			n += 2 + l + sovSchedulerobjects(uint64(l))
 		}
 	}
+	l = len(m.Executor)
+	if l > 0 {
+		n += 2 + l + sovSchedulerobjects(uint64(l))
+	}
 	return n
 }
 
@@ -4088,6 +4111,38 @@ func (m *Node) Unmarshal(dAtA []byte) error {
 			m.NodeDbKeys = append(m.NodeDbKeys, make([]byte, postIndex-iNdEx))
 			copy(m.NodeDbKeys[len(m.NodeDbKeys)-1], dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 19:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Executor", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowSchedulerobjects
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthSchedulerobjects
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthSchedulerobjects
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Executor = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipSchedulerobjects(dAtA[iNdEx:])
diff --git a/internal/scheduler/schedulerobjects/schedulerobjects.proto b/internal/scheduler/schedulerobjects/schedulerobjects.proto
index f87f8827..fd189f08 100644
--- a/internal/scheduler/schedulerobjects/schedulerobjects.proto
+++ b/internal/scheduler/schedulerobjects/schedulerobjects.proto
@@ -31,6 +31,8 @@ message Node {
     string id = 1;
     // Kubernetes node name. Not guaranteed to be unique across clusters.
     string name = 12;
+    // Name of the executor to which this node belongs.
+    string executor = 19;
     // Keys by which nodes are sorted in the nodeDb.
     repeated bytes nodeDbKeys = 18;
     // Time at which this node was last updated.
diff --git a/internal/scheduler/scheduling_algo.go b/internal/scheduler/scheduling_algo.go
index 5a29d9bc..39a3668e 100644
--- a/internal/scheduler/scheduling_algo.go
+++ b/internal/scheduler/scheduling_algo.go
@@ -3,21 +3,20 @@ package scheduler
 import (
 	"context"
 	"math/rand"
-	"strings"
 	"time"
 
 	"github.com/benbjohnson/immutable"
 	"github.com/google/uuid"
 	"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus"
 	"golang.org/x/exp/maps"
 	"golang.org/x/exp/slices"
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/logging"
-	"github.com/armadaproject/armada/internal/common/types"
+	armadaslices "github.com/armadaproject/armada/internal/common/slices"
 	"github.com/armadaproject/armada/internal/common/util"
 	schedulerconstraints "github.com/armadaproject/armada/internal/scheduler/constraints"
 	schedulercontext "github.com/armadaproject/armada/internal/scheduler/context"
@@ -38,18 +37,19 @@ type SchedulingAlgo interface {
 
 // FairSchedulingAlgo is a SchedulingAlgo based on PreemptingQueueScheduler.
 type FairSchedulingAlgo struct {
-	config                      configuration.SchedulingConfig
+	schedulingConfig            configuration.SchedulingConfig
 	executorRepository          database.ExecutorRepository
 	queueRepository             database.QueueRepository
 	schedulingContextRepository *SchedulingContextRepository
-	priorityClasses             map[string]types.PriorityClass
-	indexedResources            []configuration.IndexedResource
-	rand                        *rand.Rand // injected here for repeatable testing
-	previousScheduleClusterId   string
 	maxSchedulingDuration       time.Duration
-	clock                       clock.Clock
-	// Function that is called every time a executor is scheduled. Useful for testing.
+	// Order in which to schedule executor groups.
+	// Executors are grouped by either id (i.e., individually) or by pool.
+	executorGroupsToSchedule []string
+	// Function that is called every time an executor is scheduled. Useful for testing.
 	onExecutorScheduled func(executor *schedulerobjects.Executor)
+	// rand and clock injected here for repeatable testing.
+	rand  *rand.Rand
+	clock clock.Clock
 }
 
 func NewFairSchedulingAlgo(
@@ -62,25 +62,21 @@ func NewFairSchedulingAlgo(
 	if _, ok := config.Preemption.PriorityClasses[config.Preemption.DefaultPriorityClass]; !ok {
 		return nil, errors.Errorf("default priority class %s is missing from priority class mapping %v", config.Preemption.DefaultPriorityClass, config.Preemption.PriorityClasses)
 	}
-	algo := &FairSchedulingAlgo{
-		config:                      config,
+	return &FairSchedulingAlgo{
+		schedulingConfig:            config,
 		executorRepository:          executorRepository,
 		queueRepository:             queueRepository,
 		schedulingContextRepository: schedulingContextRepository,
-		priorityClasses:             config.Preemption.PriorityClasses,
-		indexedResources:            config.IndexedResources,
 		maxSchedulingDuration:       maxSchedulingDuration,
 		rand:                        util.NewThreadsafeRand(time.Now().UnixNano()),
 		clock:                       clock.RealClock{},
 		onExecutorScheduled:         func(executor *schedulerobjects.Executor) {},
-	}
-
-	return algo, nil
+	}, nil
 }
 
 // Schedule assigns jobs to nodes in the same way as the old lease call.
-// It iterates over each executor in turn (using lexicographical order) and assigns the jobs using a LegacyScheduler, before moving onto the next executor
-// It maintains state of which executors it has considered already and may take multiple Schedule() calls to consider all of the executors if scheduling is slow
+// It iterates over each executor in turn (using lexicographical order) and assigns the jobs using a LegacyScheduler, before moving onto the next executor.
+// It maintains state of which executors it has considered already and may take multiple Schedule() calls to consider all executors if scheduling is slow.
 // Newly leased jobs are updated as such in the jobDb using the transaction provided and are also returned to the caller.
 func (l *FairSchedulingAlgo) Schedule(
 	ctx context.Context,
@@ -88,40 +84,61 @@ func (l *FairSchedulingAlgo) Schedule(
 	jobDb *jobdb.JobDb,
 ) (*SchedulerResult, error) {
 	log := ctxlogrus.Extract(ctx)
-	accounting, err := l.newFairSchedulingAlgoContext(ctx, txn, jobDb)
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, l.maxSchedulingDuration)
+	defer cancel()
+
+	fsctx, err := l.newFairSchedulingAlgoContext(ctx, txn, jobDb)
 	if err != nil {
 		return nil, err
 	}
 	overallSchedulerResult := &SchedulerResult{
 		NodeIdByJobId: make(map[string]string),
 	}
-
-	timeout, cancel := context.WithTimeout(ctx, l.maxSchedulingDuration)
-	defer cancel()
-
-	allExecutorsConsidered := false
-	executorsToSchedule := accounting.getExecutorsToSchedule(l.previousScheduleClusterId)
-	for i, executor := range executorsToSchedule {
-		if timeout.Err() != nil {
-			// We've reached the scheduling time limit, exit gracefully
-			log.Infof("ending scheduling round early as we have hit the maximum scheduling duration")
-			break
+	executorGroups := l.groupExecutors(fsctx.executors)
+	if len(l.executorGroupsToSchedule) == 0 {
+		// Cycle over groups in a consistent order.
+		l.executorGroupsToSchedule = maps.Keys(executorGroups)
+		slices.Sort(l.executorGroupsToSchedule)
+	}
+	for len(l.executorGroupsToSchedule) > 0 {
+		select {
+		case <-ctxWithTimeout.Done():
+			// We've reached the scheduling time limit; exit gracefully.
+			log.Info("ending scheduling round early as we have hit the maximum scheduling duration")
+			return overallSchedulerResult, nil
+		default:
+		}
+		executorGroupLabel := armadaslices.Pop(&l.executorGroupsToSchedule)
+		executorGroup := executorGroups[executorGroupLabel]
+		if len(executorGroup) == 0 {
+			continue
+		}
+		for _, executor := range executorGroup {
+			if executor == nil {
+				return nil, errors.Errorf("nil executor in group %s", executorGroup)
+			}
 		}
 
-		log.Infof("scheduling on %s", executor.Id)
-		schedulerResult, sctx, err := l.scheduleOnExecutor(
-			timeout,
-			accounting,
-			txn,
-			executor,
-			jobDb,
+		// Schedule across the executors in this group.
+		// Assume pool and minimumJobSize are consistent within the group.
+		pool := executorGroup[0].Pool
+		minimumJobSize := executorGroup[0].MinimumJobSize
+		log.Infof("scheduling on executor group %s", executorGroupLabel)
+		schedulerResult, sctx, err := l.scheduleOnExecutors(
+			ctxWithTimeout,
+			fsctx,
+			pool,
+			minimumJobSize,
+			executorGroup,
 		)
-		if err != nil {
-			if err == context.DeadlineExceeded {
-				log.Infof("stopped scheduling on %s early as we have hit the maximum scheduling duration", executor.Id)
-				// We've reached the scheduling time limit, exit gracefully
-				break
-			}
+		if err == context.DeadlineExceeded {
+			// We've reached the scheduling time limit;
+			// add the executorGroupLabel back to l.executorGroupsToSchedule such that we try it again next time,
+			// and exit gracefully.
+			l.executorGroupsToSchedule = append(l.executorGroupsToSchedule, executorGroupLabel)
+			log.Info("stopped scheduling early as we have hit the maximum scheduling duration")
+			break
+		} else if err != nil {
 			return nil, err
 		}
 		if l.schedulingContextRepository != nil {
@@ -134,7 +151,6 @@ func (l *FairSchedulingAlgo) Schedule(
 		preemptedJobs := PreemptedJobsFromSchedulerResult[*jobdb.Job](schedulerResult)
 		scheduledJobs := ScheduledJobsFromSchedulerResult[*jobdb.Job](schedulerResult)
 		if err := jobDb.Upsert(txn, preemptedJobs); err != nil {
-			// TODO: We need to do something here to mark the jobs as preempted in the jobDb.
 			return nil, err
 		}
 		if err := jobDb.Upsert(txn, scheduledJobs); err != nil {
@@ -146,23 +162,34 @@ func (l *FairSchedulingAlgo) Schedule(
 		overallSchedulerResult.ScheduledJobs = append(overallSchedulerResult.ScheduledJobs, schedulerResult.ScheduledJobs...)
 		maps.Copy(overallSchedulerResult.NodeIdByJobId, schedulerResult.NodeIdByJobId)
 
-		// Update accounting.
-		accounting.allocationByPoolAndQueueAndPriorityClass[executor.Pool] = sctx.AllocatedByQueueAndPriority()
+		// Update fsctx.
+		fsctx.allocationByPoolAndQueueAndPriorityClass[pool] = sctx.AllocatedByQueueAndPriority()
 
-		// Update result to mark this executor as scheduled
-		l.previousScheduleClusterId = executor.Id
-		l.onExecutorScheduled(executor)
-
-		if i+1 == len(executorsToSchedule) {
-			allExecutorsConsidered = true
+		for _, executor := range executorGroup {
+			l.onExecutorScheduled(executor)
 		}
 	}
-	if allExecutorsConsidered {
-		log.Infof("successfully scheduled on all executors")
-	}
 	return overallSchedulerResult, nil
 }
 
+func (l *FairSchedulingAlgo) groupExecutors(executors []*schedulerobjects.Executor) map[string][]*schedulerobjects.Executor {
+	if l.schedulingConfig.UnifiedSchedulingByPool {
+		return armadaslices.GroupByFunc(
+			executors,
+			func(executor *schedulerobjects.Executor) string {
+				return executor.Pool
+			},
+		)
+	} else {
+		return armadaslices.GroupByFunc(
+			executors,
+			func(executor *schedulerobjects.Executor) string {
+				return executor.Id
+			},
+		)
+	}
+}
+
 type JobQueueIteratorAdapter struct {
 	it *immutable.SortedSetIterator[*jobdb.Job]
 }
@@ -178,45 +205,15 @@ func (it *JobQueueIteratorAdapter) Next() (interfaces.LegacySchedulerJob, error)
 type fairSchedulingAlgoContext struct {
 	priorityFactorByQueue                    map[string]float64
 	isActiveByQueueName                      map[string]bool
-	totalCapacity                            schedulerobjects.ResourceList
+	totalCapacityByPool                      schedulerobjects.QuantityByTAndResourceType[string]
 	jobsByExecutorId                         map[string][]*jobdb.Job
 	nodeIdByJobId                            map[string]string
 	jobIdsByGangId                           map[string]map[string]bool
 	gangIdByJobId                            map[string]string
 	allocationByPoolAndQueueAndPriorityClass map[string]map[string]schedulerobjects.QuantityByTAndResourceType[string]
 	executors                                []*schedulerobjects.Executor
-}
-
-// This function will return executors in the order they should be scheduled in
-// The order is lexicographical on executor id and adjusted to start with the id next after previousScheduledExecutorId
-// Example executors ids C, A, D, B
-// If previousScheduledExecutorId is blank
-//   - return executors in order A, B, C, D
-//
-// If previousScheduledExecutorId is B
-//   - return executors in order C, D, A, B
-func (f fairSchedulingAlgoContext) getExecutorsToSchedule(previousScheduledExecutorId string) []*schedulerobjects.Executor {
-	sortedExecutors := f.executors
-	slices.SortStableFunc(sortedExecutors, func(a, b *schedulerobjects.Executor) bool {
-		return strings.Compare(a.Id, b.Id) < 1
-	})
-
-	executorsToSchedule := make([]*schedulerobjects.Executor, 0, len(sortedExecutors))
-	if previousScheduledExecutorId == "" {
-		executorsToSchedule = sortedExecutors
-	} else {
-		for i, executor := range sortedExecutors {
-			if executor.Id > previousScheduledExecutorId {
-				executorsToSchedule = append(executorsToSchedule, sortedExecutors[i:]...)
-				executorsToSchedule = append(executorsToSchedule, sortedExecutors[:i]...)
-				break
-			} else if i+1 == len(sortedExecutors) {
-				// This means all executors ids are less than previousScheduleClusterId
-				executorsToSchedule = sortedExecutors
-			}
-		}
-	}
-	return executorsToSchedule
+	txn                                      *jobdb.Txn
+	jobDb                                    *jobdb.JobDb
 }
 
 func (l *FairSchedulingAlgo) newFairSchedulingAlgoContext(ctx context.Context, txn *jobdb.Txn, jobDb *jobdb.JobDb) (*fairSchedulingAlgoContext, error) {
@@ -236,10 +233,10 @@ func (l *FairSchedulingAlgo) newFairSchedulingAlgoContext(ctx context.Context, t
 	}
 
 	// Get the total capacity available across executors.
-	totalCapacity := schedulerobjects.ResourceList{}
+	totalCapacityByPool := make(schedulerobjects.QuantityByTAndResourceType[string])
 	for _, executor := range executors {
 		for _, node := range executor.Nodes {
-			totalCapacity.Add(node.TotalResources)
+			totalCapacityByPool.AddResourceList(executor.Pool, node.TotalResources)
 		}
 	}
 
@@ -288,55 +285,71 @@ func (l *FairSchedulingAlgo) newFairSchedulingAlgoContext(ctx context.Context, t
 
 	// Filter out any executor that isn't acknowledging jobs in a timely fashion
 	// Note that we do this after aggregating allocation across clusters for fair share.
-	executors = l.filterLaggingExecutors(executors, jobsByExecutorId)
+	executors = l.filterLaggingExecutors(ctx, executors, jobsByExecutorId)
 
 	return &fairSchedulingAlgoContext{
 		priorityFactorByQueue:                    priorityFactorByQueue,
 		isActiveByQueueName:                      isActiveByQueueName,
-		totalCapacity:                            totalCapacity,
+		totalCapacityByPool:                      totalCapacityByPool,
 		jobsByExecutorId:                         jobsByExecutorId,
 		nodeIdByJobId:                            nodeIdByJobId,
 		jobIdsByGangId:                           jobIdsByGangId,
 		gangIdByJobId:                            gangIdByJobId,
 		allocationByPoolAndQueueAndPriorityClass: totalAllocationByPoolAndQueue,
 		executors:                                executors,
+		jobDb:                                    jobDb,
+		txn:                                      txn,
 	}, nil
 }
 
-// scheduleOnExecutor schedules jobs on a specified executor.
-func (l *FairSchedulingAlgo) scheduleOnExecutor(
+// scheduleOnExecutors schedules jobs on a specified set of executors.
+func (l *FairSchedulingAlgo) scheduleOnExecutors(
 	ctx context.Context,
-	accounting *fairSchedulingAlgoContext,
-	txn *jobdb.Txn,
-	executor *schedulerobjects.Executor,
-	db *jobdb.JobDb,
+	fsctx *fairSchedulingAlgoContext,
+	pool string,
+	minimumJobSize schedulerobjects.ResourceList,
+	executors []*schedulerobjects.Executor,
 ) (*SchedulerResult, *schedulercontext.SchedulingContext, error) {
-	nodeDb, err := l.constructNodeDb(
-		l.config.Preemption.PriorityClasses,
-		accounting.jobsByExecutorId[executor.Id],
-		executor.Nodes,
+	nodeDb, err := nodedb.NewNodeDb(
+		l.schedulingConfig.Preemption.PriorityClasses,
+		l.schedulingConfig.MaxExtraNodesToConsider,
+		l.schedulingConfig.IndexedResources,
+		l.schedulingConfig.IndexedTaints,
+		l.schedulingConfig.IndexedNodeLabels,
 	)
 	if err != nil {
 		return nil, nil, err
 	}
+	for _, executor := range executors {
+		if err := l.addExecutorToNodeDb(nodeDb, fsctx.jobsByExecutorId[executor.Id], executor.Nodes); err != nil {
+			return nil, nil, err
+		}
+	}
+
+	// If there are multiple executors, use pool name instead of executorId.
+	// ExecutorId is only used for reporting so this results in an aggregated report for the pool.
+	executorId := pool
+	if len(executors) == 1 {
+		executorId = executors[0].Id
+	}
 	sctx := schedulercontext.NewSchedulingContext(
-		executor.Id,
-		executor.Pool,
-		l.config.Preemption.PriorityClasses,
-		l.config.Preemption.DefaultPriorityClass,
-		l.config.ResourceScarcity,
-		accounting.totalCapacity,
+		executorId,
+		pool,
+		l.schedulingConfig.Preemption.PriorityClasses,
+		l.schedulingConfig.Preemption.DefaultPriorityClass,
+		l.schedulingConfig.ResourceScarcity,
+		fsctx.totalCapacityByPool[pool],
 	)
-	if l.config.FairnessModel == configuration.DominantResourceFairness {
-		sctx.EnableDominantResourceFairness(l.config.DominantResourceFairnessResourcesToConsider)
+	if l.schedulingConfig.FairnessModel == configuration.DominantResourceFairness {
+		sctx.EnableDominantResourceFairness(l.schedulingConfig.DominantResourceFairnessResourcesToConsider)
 	}
-	for queue, priorityFactor := range accounting.priorityFactorByQueue {
-		if !accounting.isActiveByQueueName[queue] {
+	for queue, priorityFactor := range fsctx.priorityFactorByQueue {
+		if !fsctx.isActiveByQueueName[queue] {
 			// To ensure fair share is computed only from active queues, i.e., queues with jobs queued or running.
 			continue
 		}
 		var allocatedByPriorityClass schedulerobjects.QuantityByTAndResourceType[string]
-		if allocatedByQueueAndPriorityClass := accounting.allocationByPoolAndQueueAndPriorityClass[executor.Pool]; allocatedByQueueAndPriorityClass != nil {
+		if allocatedByQueueAndPriorityClass := fsctx.allocationByPoolAndQueueAndPriorityClass[pool]; allocatedByQueueAndPriorityClass != nil {
 			allocatedByPriorityClass = allocatedByQueueAndPriorityClass[queue]
 		}
 		var weight float64 = 1
@@ -348,30 +361,30 @@ func (l *FairSchedulingAlgo) scheduleOnExecutor(
 		}
 	}
 	constraints := schedulerconstraints.SchedulingConstraintsFromSchedulingConfig(
-		executor.Pool,
-		accounting.totalCapacity,
-		executor.MinimumJobSize,
-		l.config,
+		pool,
+		fsctx.totalCapacityByPool[pool],
+		minimumJobSize,
+		l.schedulingConfig,
 	)
 	scheduler := NewPreemptingQueueScheduler(
 		sctx,
 		constraints,
-		l.config.Preemption.NodeEvictionProbability,
-		l.config.Preemption.NodeOversubscriptionEvictionProbability,
-		l.config.Preemption.ProtectedFractionOfFairShare,
+		l.schedulingConfig.Preemption.NodeEvictionProbability,
+		l.schedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
+		l.schedulingConfig.Preemption.ProtectedFractionOfFairShare,
 		&schedulerJobRepositoryAdapter{
-			txn: txn,
-			db:  db,
+			txn: fsctx.txn,
+			db:  fsctx.jobDb,
 		},
 		nodeDb,
-		accounting.nodeIdByJobId,
-		accounting.jobIdsByGangId,
-		accounting.gangIdByJobId,
+		fsctx.nodeIdByJobId,
+		fsctx.jobIdsByGangId,
+		fsctx.gangIdByJobId,
 	)
-	if l.config.AlwaysAttemptScheduling {
+	if l.schedulingConfig.AlwaysAttemptScheduling {
 		scheduler.SkipUnsuccessfulSchedulingKeyCheck()
 	}
-	if l.config.EnableAssertions {
+	if l.schedulingConfig.EnableAssertions {
 		scheduler.EnableAssertions()
 	}
 	result, err := scheduler.Schedule(ctx)
@@ -397,7 +410,7 @@ func (l *FairSchedulingAlgo) scheduleOnExecutor(
 		if node, err := nodeDb.GetNode(nodeId); err != nil {
 			return nil, nil, err
 		} else {
-			result.ScheduledJobs[i] = jobDbJob.WithQueued(false).WithNewRun(executor.Id, node.Name)
+			result.ScheduledJobs[i] = jobDbJob.WithQueued(false).WithNewRun(node.Executor, node.Id)
 		}
 	}
 	return result, sctx, nil
@@ -409,7 +422,7 @@ type schedulerJobRepositoryAdapter struct {
 	txn *jobdb.Txn
 }
 
-// GetQueueJobIds is Necessary to implement the JobRepository interface, which we need while transitioning from the old
+// GetQueueJobIds is necessary to implement the JobRepository interface, which we need while transitioning from the old
 // to new scheduler.
 func (repo *schedulerJobRepositoryAdapter) GetQueueJobIds(queue string) ([]string, error) {
 	rv := make([]string, 0)
@@ -432,79 +445,71 @@ func (repo *schedulerJobRepositoryAdapter) GetExistingJobsByIds(ids []string) ([
 	return rv, nil
 }
 
-// constructNodeDb constructs a node db with all jobs bound to it.
-func (l *FairSchedulingAlgo) constructNodeDb(priorityClasses map[string]types.PriorityClass, jobs []*jobdb.Job, nodes []*schedulerobjects.Node) (*nodedb.NodeDb, error) {
-	nodeDb, err := nodedb.NewNodeDb(
-		priorityClasses,
-		l.config.MaxExtraNodesToConsider,
-		l.indexedResources,
-		l.config.IndexedTaints,
-		l.config.IndexedNodeLabels,
-	)
-	if err != nil {
-		return nil, err
-	}
+// addExecutorToNodeDb adds all the nodes and jobs associated with a particular executor to the nodeDb.
+func (l *FairSchedulingAlgo) addExecutorToNodeDb(nodeDb *nodedb.NodeDb, jobs []*jobdb.Job, nodes []*schedulerobjects.Node) error {
 	txn := nodeDb.Txn(true)
 	defer txn.Abort()
-	nodesByName := make(map[string]*schedulerobjects.Node, len(nodes))
-	for _, node := range nodes {
-		nodesByName[node.Name] = node
-	}
-	jobsByNodeName := make(map[string][]*jobdb.Job)
+	nodesById := armadaslices.GroupByFuncUnique(
+		nodes,
+		func(node *schedulerobjects.Node) string { return node.Id },
+	)
+	jobsByNodeId := make(map[string][]*jobdb.Job, len(nodes))
 	for _, job := range jobs {
 		if job.InTerminalState() || !job.HasRuns() {
 			continue
 		}
-		nodeName := job.LatestRun().Node()
-		if _, ok := nodesByName[nodeName]; !ok {
-			log.Warnf(
+		nodeId := job.LatestRun().Node()
+		if _, ok := nodesById[nodeId]; !ok {
+			logrus.Errorf(
 				"job %s assigned to node %s on executor %s, but no such node found",
-				job.Id(), nodeName, job.LatestRun().Executor(),
+				job.Id(), nodeId, job.LatestRun().Executor(),
 			)
 			continue
 		}
-		jobsByNodeName[nodeName] = append(jobsByNodeName[nodeName], job)
+		jobsByNodeId[nodeId] = append(jobsByNodeId[nodeId], job)
 	}
 	for _, node := range nodes {
-		if err := nodeDb.CreateAndInsertWithJobDbJobsWithTxn(txn, jobsByNodeName[node.Name], node); err != nil {
-			return nil, err
+		if err := nodeDb.CreateAndInsertWithJobDbJobsWithTxn(txn, jobsByNodeId[node.Id], node); err != nil {
+			return err
 		}
 	}
 	txn.Commit()
-	return nodeDb, nil
+	return nil
 }
 
-// filterStaleExecutors returns all executors which have sent a lease request within the duration given by l.config.ExecutorTimeout.
+// filterStaleExecutors returns all executors which have sent a lease request within the duration given by l.schedulingConfig.ExecutorTimeout.
 // This ensures that we don't continue to assign jobs to executors that are no longer active.
 func (l *FairSchedulingAlgo) filterStaleExecutors(executors []*schedulerobjects.Executor) []*schedulerobjects.Executor {
 	activeExecutors := make([]*schedulerobjects.Executor, 0, len(executors))
-	cutoff := l.clock.Now().Add(-l.config.ExecutorTimeout)
+	cutoff := l.clock.Now().Add(-l.schedulingConfig.ExecutorTimeout)
 	for _, executor := range executors {
 		if executor.LastUpdateTime.After(cutoff) {
 			activeExecutors = append(activeExecutors, executor)
 		} else {
-			log.Debugf("Ignoring executor %s because it hasn't heartbeated since %s", executor.Id, executor.LastUpdateTime)
+			logrus.Debugf("Ignoring executor %s because it hasn't heartbeated since %s", executor.Id, executor.LastUpdateTime)
 		}
 	}
 	return activeExecutors
 }
 
-// filterLaggingExecutors returns all executors with <= l.config.MaxUnacknowledgedJobsPerExecutor unacknowledged jobs,
+// filterLaggingExecutors returns all executors with <= l.schedulingConfig.MaxUnacknowledgedJobsPerExecutor unacknowledged jobs,
 // where unacknowledged means the executor has not echoed the job since it was scheduled.
 //
 // Used to rate-limit scheduling onto executors that can't keep up.
 //
 // TODO: Let's also check that jobs are on the right nodes.
 func (l *FairSchedulingAlgo) filterLaggingExecutors(
+	ctx context.Context,
 	executors []*schedulerobjects.Executor,
 	leasedJobsByExecutor map[string][]*jobdb.Job,
 ) []*schedulerobjects.Executor {
+	log := ctxlogrus.Extract(ctx)
 	activeExecutors := make([]*schedulerobjects.Executor, 0, len(executors))
 	for _, executor := range executors {
 		leasedJobs := leasedJobsByExecutor[executor.Id]
 		executorRuns, err := executor.AllRuns()
 		if err != nil {
-			log.Errorf("failed to retrieve runs for executor %s; will not be considered for scheduling: %s", executor.Id, err.Error())
+			logging.WithStacktrace(log, err).Errorf("failed to retrieve runs for executor %s; will not be considered for scheduling", executor.Id)
 			continue
 		}
 		executorRunIds := make(map[uuid.UUID]bool, len(executorRuns))
@@ -512,7 +517,7 @@ func (l *FairSchedulingAlgo) filterLaggingExecutors(
 			executorRunIds[run] = true
 		}
 
-		numUnacknowledgedJobs := uint(0)
+		var numUnacknowledgedJobs uint
 		for _, leasedJob := range leasedJobs {
 			if leasedJob.HasRuns() && !leasedJob.InTerminalState() {
 				if !executorRunIds[leasedJob.LatestRun().Id()] {
@@ -520,20 +525,22 @@ func (l *FairSchedulingAlgo) filterLaggingExecutors(
 				}
 			}
 		}
-		if numUnacknowledgedJobs <= l.config.MaxUnacknowledgedJobsPerExecutor {
+		if numUnacknowledgedJobs <= l.schedulingConfig.MaxUnacknowledgedJobsPerExecutor {
 			activeExecutors = append(activeExecutors, executor)
 		} else {
 			log.Warnf(
-				"%d unacknowledged jobs on executor %s exceeds limit of %d; will not be considered for scheduling",
-				numUnacknowledgedJobs, executor.Id, l.config.MaxUnacknowledgedJobsPerExecutor,
+				"%d unacknowledged jobs on executor %s exceeds limit of %d; executor will not be considered for scheduling",
+				numUnacknowledgedJobs, executor.Id, l.schedulingConfig.MaxUnacknowledgedJobsPerExecutor,
 			)
 		}
-
 	}
 	return activeExecutors
 }
 
-func (l *FairSchedulingAlgo) aggregateAllocationByPoolAndQueueAndPriorityClass(executors []*schedulerobjects.Executor, jobsByExecutorId map[string][]*jobdb.Job) map[string]map[string]schedulerobjects.QuantityByTAndResourceType[string] {
+func (l *FairSchedulingAlgo) aggregateAllocationByPoolAndQueueAndPriorityClass(
+	executors []*schedulerobjects.Executor,
+	jobsByExecutorId map[string][]*jobdb.Job,
+) map[string]map[string]schedulerobjects.QuantityByTAndResourceType[string] {
 	rv := make(map[string]map[string]schedulerobjects.QuantityByTAndResourceType[string])
 	for _, executor := range executors {
 		allocationByQueue := rv[executor.Pool]
diff --git a/pkg/api/util.go b/pkg/api/util.go
index 16f4e9ab..65c9c1b5 100644
--- a/pkg/api/util.go
+++ b/pkg/api/util.go
@@ -71,6 +71,7 @@ func NewNodeFromNodeInfo(nodeInfo *NodeInfo, executor string, allowedPriorities
 	return &schedulerobjects.Node{
 		Id:                               fmt.Sprintf("%s-%s", executor, nodeInfo.Name),
 		Name:                             nodeInfo.Name,
+		Executor:                         executor,
 		LastSeen:                         lastSeen,
 		Taints:                           nodeInfo.GetTaints(),
 		Labels:                           nodeInfo.GetLabels(),


Test patch:
diff --git a/internal/common/slices/slices_test.go b/internal/common/slices/slices_test.go
index 3f1644d1..e293c663 100644
--- a/internal/common/slices/slices_test.go
+++ b/internal/common/slices/slices_test.go
@@ -241,3 +241,26 @@ func TestFilter_Nil(t *testing.T) {
 	output := Filter(input, includeAll)
 	assert.Equal(t, expectedOutput, output)
 }
+
+func TestPop(t *testing.T) {
+	s := []int{1, 2, 3}
+	expected := []int{3, 2, 1}
+	var actual []int
+	for len(s) > 0 {
+		actual = append(actual, Pop(&s))
+	}
+	assert.Equal(t, expected, actual)
+	assert.Panics(
+		t,
+		func() {
+			Pop(&s)
+		},
+	)
+	assert.Panics(
+		t,
+		func() {
+			var s []int
+			Pop(&s)
+		},
+	)
+}
diff --git a/internal/scheduler/api_test.go b/internal/scheduler/api_test.go
index 5587c8cf..2f9d529d 100644
--- a/internal/scheduler/api_test.go
+++ b/internal/scheduler/api_test.go
@@ -56,6 +56,7 @@ func TestExecutorApi_LeaseJobRuns(t *testing.T) {
 			{
 				Id:                          "test-executor-test-node",
 				Name:                        "test-node",
+				Executor:                    "test-executor",
 				TotalResources:              schedulerobjects.ResourceList{},
 				StateByJobRunId:             map[string]schedulerobjects.JobRunState{runId1.String(): schedulerobjects.JobRunState_RUNNING, runId2.String(): schedulerobjects.JobRunState_RUNNING},
 				NonArmadaAllocatedResources: map[int32]schedulerobjects.ResourceList{},
diff --git a/internal/scheduler/jobdb/job_run_test.go b/internal/scheduler/jobdb/job_run_test.go
index 0a0e0669..7dde2de3 100644
--- a/internal/scheduler/jobdb/job_run_test.go
+++ b/internal/scheduler/jobdb/job_run_test.go
@@ -12,7 +12,7 @@ var baseJobRun = CreateRun(
 	uuid.NewString(),
 	5,
 	"test-executor",
-	"test-node",
+	"test-nodeId",
 	false,
 	false,
 	false,
@@ -26,7 +26,7 @@ func TestJobRun_TestGetter(t *testing.T) {
 	assert.Equal(t, baseJobRun.id, baseJobRun.Id())
 	assert.Equal(t, baseJobRun.created, baseJobRun.Created())
 	assert.Equal(t, baseJobRun.executor, baseJobRun.Executor())
-	assert.Equal(t, baseJobRun.node, baseJobRun.Node())
+	assert.Equal(t, baseJobRun.nodeId, baseJobRun.Node())
 }
 
 func TestJobRun_TestRunning(t *testing.T) {
@@ -71,7 +71,7 @@ func TestDeepCopy(t *testing.T) {
 		"job id",
 		1,
 		"executor",
-		"node",
+		"nodeId",
 		true,
 		true,
 		true,
@@ -84,7 +84,7 @@ func TestDeepCopy(t *testing.T) {
 		"job id",
 		1,
 		"executor",
-		"node",
+		"nodeId",
 		true,
 		true,
 		true,
@@ -93,7 +93,7 @@ func TestDeepCopy(t *testing.T) {
 		true,
 	)
 	actual := run.DeepCopy()
-	run.node = "new node"
+	run.nodeId = "new nodeId"
 	run.executor = "new executor"
 	assert.Equal(t, expected, actual)
 }
diff --git a/internal/scheduler/jobdb/job_test.go b/internal/scheduler/jobdb/job_test.go
index f5121c10..0ef2acdd 100644
--- a/internal/scheduler/jobdb/job_test.go
+++ b/internal/scheduler/jobdb/job_test.go
@@ -123,11 +123,11 @@ func TestJob_TestInTerminalState(t *testing.T) {
 
 func TestJob_TestHasRuns(t *testing.T) {
 	assert.Equal(t, false, baseJob.HasRuns())
-	assert.Equal(t, true, baseJob.WithNewRun("test-executor", "test-node").HasRuns())
+	assert.Equal(t, true, baseJob.WithNewRun("test-executor", "test-nodeId").HasRuns())
 }
 
 func TestJob_TestWithNewRun(t *testing.T) {
-	jobWithRun := baseJob.WithNewRun("test-executor", "test-node")
+	jobWithRun := baseJob.WithNewRun("test-executor", "test-nodeId")
 	assert.Equal(t, true, jobWithRun.HasRuns())
 	run := jobWithRun.LatestRun()
 	assert.NotNil(t, run)
@@ -136,7 +136,7 @@ func TestJob_TestWithNewRun(t *testing.T) {
 		jobId:    "test-job",
 		created:  run.created,
 		executor: "test-executor",
-		node:     "test-node",
+		nodeId:   "test-nodeId",
 	}, run)
 }
 
diff --git a/internal/scheduler/jobdb/jobdb_test.go b/internal/scheduler/jobdb/jobdb_test.go
index 0f34f3a8..19ff2fb8 100644
--- a/internal/scheduler/jobdb/jobdb_test.go
+++ b/internal/scheduler/jobdb/jobdb_test.go
@@ -54,8 +54,8 @@ func TestJobDb_TestGetById(t *testing.T) {
 
 func TestJobDb_TestGetByRunId(t *testing.T) {
 	jobDb := NewJobDb()
-	job1 := newJob().WithNewRun("executor", "node")
-	job2 := newJob().WithNewRun("executor", "node")
+	job1 := newJob().WithNewRun("executor", "nodeId")
+	job2 := newJob().WithNewRun("executor", "nodeId")
 	txn := jobDb.WriteTxn()
 
 	err := jobDb.Upsert(txn, []*Job{job1, job2})
@@ -71,8 +71,8 @@ func TestJobDb_TestGetByRunId(t *testing.T) {
 
 func TestJobDb_TestHasQueuedJobs(t *testing.T) {
 	jobDb := NewJobDb()
-	job1 := newJob().WithNewRun("executor", "node")
-	job2 := newJob().WithNewRun("executor", "node")
+	job1 := newJob().WithNewRun("executor", "nodeId")
+	job2 := newJob().WithNewRun("executor", "nodeId")
 	txn := jobDb.WriteTxn()
 
 	err := jobDb.Upsert(txn, []*Job{job1, job2})
@@ -141,8 +141,8 @@ func TestJobDb_TestQueuedJobs(t *testing.T) {
 
 func TestJobDb_TestGetAll(t *testing.T) {
 	jobDb := NewJobDb()
-	job1 := newJob().WithNewRun("executor", "node")
-	job2 := newJob().WithNewRun("executor", "node")
+	job1 := newJob().WithNewRun("executor", "nodeId")
+	job2 := newJob().WithNewRun("executor", "nodeId")
 	txn := jobDb.WriteTxn()
 	assert.Equal(t, []*Job{}, jobDb.GetAll(txn))
 
@@ -180,8 +180,8 @@ func TestJobDb_TestTransactions(t *testing.T) {
 
 func TestJobDb_TestBatchDelete(t *testing.T) {
 	jobDb := NewJobDb()
-	job1 := newJob().WithQueued(true).WithNewRun("executor", "node")
-	job2 := newJob().WithQueued(true).WithNewRun("executor", "node")
+	job1 := newJob().WithQueued(true).WithNewRun("executor", "nodeId")
+	job2 := newJob().WithQueued(true).WithNewRun("executor", "nodeId")
 	txn := jobDb.WriteTxn()
 
 	// Insert Job
diff --git a/internal/scheduler/scheduling_algo_test.go b/internal/scheduler/scheduling_algo_test.go
index b27a4756..11b7585f 100644
--- a/internal/scheduler/scheduling_algo_test.go
+++ b/internal/scheduler/scheduling_algo_test.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"math/rand"
 	"testing"
 	"time"
 
@@ -15,169 +14,154 @@ import (
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
+	armadaslices "github.com/armadaproject/armada/internal/common/slices"
 	"github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	schedulermocks "github.com/armadaproject/armada/internal/scheduler/mocks"
+	"github.com/armadaproject/armada/internal/scheduler/nodedb"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 	"github.com/armadaproject/armada/internal/scheduler/testfixtures"
 )
 
-func TestLegacySchedulingAlgo_TestSchedule(t *testing.T) {
+func TestSchedule(t *testing.T) {
+	type scheduledJobs struct {
+		jobs         []*jobdb.Job
+		acknowledged bool
+	}
 	tests := map[string]struct {
 		schedulingConfig configuration.SchedulingConfig
 
-		executors []*schedulerobjects.Executor
-		queues    []*database.Queue
+		executors  []*schedulerobjects.Executor
+		queues     []*database.Queue
+		queuedJobs []*jobdb.Job
 
-		existingJobs []*jobdb.Job
-		// Map from (executor ID, node name) to indices of acknowledged jobs for that node.
-		existingRunningIndices map[string]map[string][]int
-		// Map from (executor ID, node name) to indices of unacknowledged jobs for that node.
-		existingUnacknowledgedIndices map[string]map[string][]int
+		// Already scheduled jobs. Specifically,
+		// [executorIndex][nodeIndex] = jobs scheduled onto this executor and node,
+		// where executorIndex refers to the index of executors, and nodeIndex the index of the node on that executor.
+		scheduledJobsByExecutorIndexAndNodeIndex map[int]map[int]scheduledJobs
 
-		queuedJobs []*jobdb.Job
+		// Indices of existing jobs expected to be preempted.
+		// Uses the same structure as scheduledJobsByExecutorIndexAndNodeIndex.
+		expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex map[int]map[int][]int
 
-		// Indices of jobs that we expect to be preempted.
-		expectedPreemptedIndices []int
-		// Map from executor ID to indices of jobs that we expect to be scheduled.
-		expectedScheduledIndices map[string][]int
+		// Indices of queued jobs expected to be scheduled.
+		expectedScheduledIndices []int
 	}{
-		"fill up both clusters": {
+		"scheduling": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
-				"executor2": {2, 3},
-			},
+			queues:                   []*database.Queue{testfixtures.TestDbQueue()},
+			queuedJobs:               testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
+			expectedScheduledIndices: []int{0, 1, 2, 3},
 		},
-		"one executor stale": {
+		"do not schedule onto stale executors": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.WithLastUpdateTimeExecutor(testfixtures.BaseTime.Add(-1*time.Hour), testfixtures.Test1Node32CoreExecutor("executor2")),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
-			},
+			queues:                   []*database.Queue{testfixtures.TestDbQueue()},
+			queuedJobs:               testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
+			expectedScheduledIndices: []int{0, 1},
 		},
-		"one executor exceeds unacknowledged": {
+		"do not schedule onto executors with too many unacknowledged jobs": {
+			// TODO: This test doesn't look right; we never set MaxUnacknowledgedJobsPerExecutor.
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
-			existingUnacknowledgedIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
-			},
-
+			queues:     []*database.Queue{testfixtures.TestDbQueue()},
 			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-
-			expectedScheduledIndices: map[string][]int{
-				"executor2": {0, 1},
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
+						acknowledged: false,
+					},
+				},
 			},
+			expectedScheduledIndices: []int{0, 1},
 		},
 		"one executor full": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
-			},
-
+			queues:     []*database.Queue{testfixtures.TestDbQueue()},
 			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-
-			expectedScheduledIndices: map[string][]int{
-				"executor2": {0, 1},
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
+						acknowledged: true,
+					},
+				},
 			},
+			expectedScheduledIndices: []int{0, 1},
 		},
-		"user is at usage cap before scheduling": {
+		"MaximumResourceFractionPerQueue hit before scheduling": {
 			schedulingConfig: testfixtures.WithPerPriorityLimitsConfig(
 				map[string]map[string]float64{
 					testfixtures.PriorityClass3: {"cpu": 0.5},
 				},
 				testfixtures.TestSchedulingConfig(),
 			),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
-			},
-
+			queues:     []*database.Queue{testfixtures.TestDbQueue()},
 			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-
-			expectedScheduledIndices: nil,
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 2),
+						acknowledged: true,
+					},
+				},
+			},
 		},
-		"user hits usage cap during scheduling": {
+		"MaximumResourceFractionPerQueue hit during scheduling": {
 			schedulingConfig: testfixtures.WithPerPriorityLimitsConfig(
 				map[string]map[string]float64{
 					testfixtures.PriorityClass3: {"cpu": 0.5},
 				},
 				testfixtures.TestSchedulingConfig(),
 			),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 1),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0}},
-			},
-
+			queues:     []*database.Queue{testfixtures.TestDbQueue()},
 			queuedJobs: testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0},
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 1),
+						acknowledged: true,
+					},
+				},
 			},
+			expectedScheduledIndices: []int{0},
 		},
-		"no queuedJobs to schedule": {
+		"no queued jobs": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
 			queues: []*database.Queue{testfixtures.TestDbQueue()},
-
-			expectedScheduledIndices: nil,
 		},
-		"no executor available": {
+		"no executors": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{},
-			queues:    []*database.Queue{testfixtures.TestDbQueue()},
-
-			queuedJobs:               testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
-			expectedScheduledIndices: nil,
+			executors:        []*schedulerobjects.Executor{},
+			queues:           []*database.Queue{testfixtures.TestDbQueue()},
+			queuedJobs:       testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
 		},
 		"computation of allocated resources does not confuse priority class with per-queue priority": {
 			schedulingConfig: testfixtures.WithPerPriorityLimitsConfig(
@@ -186,15 +170,8 @@ func TestLegacySchedulingAlgo_TestSchedule(t *testing.T) {
 				},
 				testfixtures.TestSchedulingConfig(),
 			),
-
 			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
 			queues:    []*database.Queue{testfixtures.TestDbQueue()},
-
-			existingJobs: []*jobdb.Job{testfixtures.Test16Cpu128GiJob(testfixtures.TestQueue, testfixtures.PriorityClass3).WithPriority(0)},
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0}},
-			},
-
 			queuedJobs: []*jobdb.Job{
 				// Submit the next job with a per-queue priority number (i.e., 1) that is larger
 				// than the per-queue priority of the already-running job (i.e., 0), but smaller
@@ -203,148 +180,172 @@ func TestLegacySchedulingAlgo_TestSchedule(t *testing.T) {
 				// accounting, then it would schedule this job.
 				testfixtures.Test16Cpu128GiJob(testfixtures.TestQueue, testfixtures.PriorityClass3).WithPriority(1),
 			},
-
-			expectedScheduledIndices: nil,
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         []*jobdb.Job{testfixtures.Test16Cpu128GiJob(testfixtures.TestQueue, testfixtures.PriorityClass3).WithPriority(0)},
+						acknowledged: true,
+					},
+				},
+			},
 		},
 		"urgency-based preemption within a single queue": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
-			queues:    []*database.Queue{{Name: "queue1", Weight: 100}},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
+			executors:        []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
+			queues:           []*database.Queue{{Name: "A"}},
+			queuedJobs:       testfixtures.N16Cpu128GiJobs("A", testfixtures.PriorityClass1, 2),
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs("A", testfixtures.PriorityClass0, 1),
+						acknowledged: true,
+					},
+				},
 			},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass1, 2),
-
-			expectedPreemptedIndices: []int{0, 1},
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
+			expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex: map[int]map[int][]int{
+				0: {
+					0: {0},
+				},
 			},
+			expectedScheduledIndices: []int{0, 1},
 		},
-		"urgency-based preemption across queues": {
+		"urgency-based preemption between queues": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
-			queues:    []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
+			executors:        []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
+			queues:           []*database.Queue{{Name: "A"}, {Name: "B"}},
+			queuedJobs:       testfixtures.N16Cpu128GiJobs("B", testfixtures.PriorityClass1, 2),
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs("B", testfixtures.PriorityClass0, 1),
+						acknowledged: true,
+					},
+				},
 			},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass1, 2),
-
-			expectedPreemptedIndices: []int{0, 1},
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
+			expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex: map[int]map[int][]int{
+				0: {
+					0: {0},
+				},
 			},
+			expectedScheduledIndices: []int{0, 1},
 		},
 		"preemption to fair share": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
-			queues:    []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
-
-			existingJobs: testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
+			executors:        []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
+			queues:           []*database.Queue{{Name: "A", Weight: 100}, {Name: "B", Weight: 100}},
+			queuedJobs:       testfixtures.N16Cpu128GiJobs("A", testfixtures.PriorityClass0, 2),
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.N16Cpu128GiJobs("B", testfixtures.PriorityClass0, 2),
+						acknowledged: true,
+					},
+				},
 			},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass0, 2),
-
-			expectedPreemptedIndices: []int{1},
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0},
+			expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex: map[int]map[int][]int{
+				0: {
+					0: {1},
+				},
 			},
+			expectedScheduledIndices: []int{0},
 		},
-		"scheduling a gang": {
-			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
-			queues:    []*database.Queue{{Name: "queue1", Weight: 100}},
-
-			queuedJobs: testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2)),
-
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
-			},
+		"gang scheduling": {
+			schedulingConfig:         testfixtures.TestSchedulingConfig(),
+			executors:                []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
+			queues:                   []*database.Queue{{Name: "A", Weight: 100}},
+			queuedJobs:               testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("A", testfixtures.PriorityClass0, 2)),
+			expectedScheduledIndices: []int{0, 1},
 		},
-		"not scheduling a gang that is too large for any of the executors": {
+		"not scheduling a gang that does not fit on any executor": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
 				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{{Name: "queue1", Weight: 100}},
-
+			queues:     []*database.Queue{{Name: "A", Weight: 100}},
 			queuedJobs: testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 3)),
-
-			expectedScheduledIndices: nil,
 		},
-		"urgency-based preemption evicting a gang": {
+		"urgency-based gang preemption": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
 			executors: []*schedulerobjects.Executor{
 				testfixtures.Test1Node32CoreExecutor("executor1"),
-				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
-			queues: []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
-
-			existingJobs: testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2)),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
+			queues:     []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
+			queuedJobs: testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass1, 1),
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2)),
+						acknowledged: true,
+					},
+				},
 			},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass1, 4),
-
-			expectedPreemptedIndices: []int{0, 1},
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0, 1},
-				"executor2": {2, 3},
+			expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex: map[int]map[int][]int{
+				0: {
+					0: {0, 1},
+				},
 			},
+			expectedScheduledIndices: []int{0},
 		},
 		"preemption to fair share evicting a gang": {
 			schedulingConfig: testfixtures.TestSchedulingConfig(),
-
-			executors: []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
-			queues:    []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
-
-			existingJobs: testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2)),
-			existingRunningIndices: map[string]map[string][]int{
-				"executor1": {"executor1-node": {0, 1}},
+			executors:        []*schedulerobjects.Executor{testfixtures.Test1Node32CoreExecutor("executor1")},
+			queues:           []*database.Queue{{Name: "queue1", Weight: 100}, {Name: "queue2", Weight: 100}},
+			queuedJobs:       testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass0, 1),
+			scheduledJobsByExecutorIndexAndNodeIndex: map[int]map[int]scheduledJobs{
+				0: {
+					0: scheduledJobs{
+						jobs:         testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs("queue1", testfixtures.PriorityClass0, 2)),
+						acknowledged: true,
+					},
+				},
 			},
-
-			queuedJobs: testfixtures.N16Cpu128GiJobs("queue2", testfixtures.PriorityClass0, 1),
-
-			expectedPreemptedIndices: []int{0, 1},
-			expectedScheduledIndices: map[string][]int{
-				"executor1": {0},
+			expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex: map[int]map[int][]int{
+				0: {
+					0: {0, 1},
+				},
+			},
+			expectedScheduledIndices: []int{0},
+		},
+		"UnifiedSchedulingByPool": {
+			schedulingConfig: testfixtures.WithUnifiedSchedulingByPoolConfig(testfixtures.TestSchedulingConfig()),
+			executors: []*schedulerobjects.Executor{
+				testfixtures.Test1Node32CoreExecutor("executor1"),
+				testfixtures.Test1Node32CoreExecutor("executor2"),
 			},
+			queues:                   []*database.Queue{testfixtures.TestDbQueue()},
+			queuedJobs:               testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass3, 10),
+			expectedScheduledIndices: []int{0, 1, 2, 3},
+		},
+		"UnifiedSchedulingByPool schedule gang job over multiple executors": {
+			schedulingConfig: testfixtures.WithUnifiedSchedulingByPoolConfig(testfixtures.TestSchedulingConfig()),
+			executors: []*schedulerobjects.Executor{
+				testfixtures.Test1Node32CoreExecutor("executor1"),
+				testfixtures.Test1Node32CoreExecutor("executor2"),
+			},
+			queues:                   []*database.Queue{testfixtures.TestDbQueue()},
+			queuedJobs:               testfixtures.WithGangAnnotationsJobs(testfixtures.N16Cpu128GiJobs(testfixtures.TestQueue, testfixtures.PriorityClass0, 4)),
+			expectedScheduledIndices: []int{0, 1, 2, 3},
 		},
 	}
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
 			ctx := testfixtures.ContextWithDefaultLogger(context.Background())
-			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			timeout := 5 * time.Second
+			ctx, cancel := context.WithTimeout(ctx, timeout)
 			defer cancel()
 
 			ctrl := gomock.NewController(t)
-
 			mockExecutorRepo := schedulermocks.NewMockExecutorRepository(ctrl)
 			mockExecutorRepo.EXPECT().GetExecutors(ctx).Return(tc.executors, nil).AnyTimes()
-
 			mockQueueRepo := schedulermocks.NewMockQueueRepository(ctrl)
 			mockQueueRepo.EXPECT().GetAllQueues().Return(tc.queues, nil).AnyTimes()
 
 			schedulingContextRepo, err := NewSchedulingContextRepository(1024)
 			require.NoError(t, err)
-			algo, err := NewFairSchedulingAlgo(
+			sch, err := NewFairSchedulingAlgo(
 				tc.schedulingConfig,
-				time.Second*5,
+				timeout,
 				mockExecutorRepo,
 				mockQueueRepo,
 				schedulingContextRepo,
@@ -352,323 +353,113 @@ func TestLegacySchedulingAlgo_TestSchedule(t *testing.T) {
 			require.NoError(t, err)
 
 			// Use a test clock so we can control time
-			algo.clock = clock.NewFakeClock(testfixtures.BaseTime)
-
-			nodes := make(map[string]map[string]*schedulerobjects.Node)
-			for _, executor := range tc.executors {
-				nodesByName := make(map[string]*schedulerobjects.Node)
-				for _, node := range executor.Nodes {
-					nodesByName[node.Name] = node
-				}
-				nodes[executor.Id] = nodesByName
-			}
+			sch.clock = clock.NewFakeClock(testfixtures.BaseTime)
 
+			// Add queued jobs to the jobDb.
 			jobsToUpsert := make([]*jobdb.Job, 0)
-
-			for executorId, jobsByNodeName := range tc.existingRunningIndices {
-				for nodeName, jobIndices := range jobsByNodeName {
-					node := nodes[executorId][nodeName]
-					for _, i := range jobIndices {
-						job := tc.existingJobs[i].WithQueued(false).WithNewRun(executorId, nodeName)
-						jobsToUpsert = append(jobsToUpsert, job)
-						run := job.LatestRun()
-						node.StateByJobRunId[run.Id().String()] = schedulerobjects.JobRunState_RUNNING
-					}
-				}
+			queueIndexByJobId := make(map[string]int)
+			for i, job := range tc.queuedJobs {
+				job = job.WithQueued(true)
+				jobsToUpsert = append(jobsToUpsert, job)
+				queueIndexByJobId[job.Id()] = i
 			}
 
-			for executorId, jobsByNodeName := range tc.existingUnacknowledgedIndices {
-				for nodeName, jobIndices := range jobsByNodeName {
-					for _, i := range jobIndices {
-						job := tc.existingJobs[i].WithQueued(false).WithNewRun(executorId, nodeName)
+			// Add scheduled jobs to the jobDb. Bind acknowledged jobs to nodes.
+			executorIndexByJobId := make(map[string]int)
+			executorNodeIndexByJobId := make(map[string]int)
+			jobIndexByJobId := make(map[string]int)
+			for executorIndex, existingJobsByExecutorNodeIndex := range tc.scheduledJobsByExecutorIndexAndNodeIndex {
+				executor := tc.executors[executorIndex]
+				for nodeIndex, existingJobs := range existingJobsByExecutorNodeIndex {
+					node := executor.Nodes[nodeIndex]
+					for jobIndex, job := range existingJobs.jobs {
+						job = job.WithQueued(false).WithNewRun(executor.Id, node.Id)
+						if existingJobs.acknowledged {
+							run := job.LatestRun()
+							node.StateByJobRunId[run.Id().String()] = schedulerobjects.JobRunState_RUNNING
+						}
 						jobsToUpsert = append(jobsToUpsert, job)
+						executorIndexByJobId[job.Id()] = executorIndex
+						executorNodeIndexByJobId[job.Id()] = nodeIndex
+						jobIndexByJobId[job.Id()] = jobIndex
 					}
 				}
 			}
 
-			for _, job := range tc.queuedJobs {
-				job = job.WithQueued(true)
-				jobsToUpsert = append(jobsToUpsert, job)
-			}
-
+			// Setup jobDb.
 			jobDb := jobdb.NewJobDb()
 			txn := jobDb.WriteTxn()
 			err = jobDb.Upsert(txn, jobsToUpsert)
 			require.NoError(t, err)
 
-			schedulerResult, err := algo.Schedule(ctx, txn, jobDb)
+			// Run a scheduling round.
+			schedulerResult, err := sch.Schedule(ctx, txn, jobDb)
 			require.NoError(t, err)
 
-			expectedScheduledJobs := make(map[string]string)
-			for executorId, jobIndices := range tc.expectedScheduledIndices {
-				for _, i := range jobIndices {
-					expectedScheduledJobs[tc.queuedJobs[i].Id()] = executorId
+			// Check that the expected preemptions took place.
+			preemptedJobs := PreemptedJobsFromSchedulerResult[*jobdb.Job](schedulerResult)
+			actualPreemptedJobsByExecutorIndexAndNodeIndex := make(map[int]map[int][]int)
+			for _, job := range preemptedJobs {
+				executorIndex := executorIndexByJobId[job.Id()]
+				nodeIndex := executorNodeIndexByJobId[job.Id()]
+				jobIndex := jobIndexByJobId[job.Id()]
+				m := actualPreemptedJobsByExecutorIndexAndNodeIndex[executorIndex]
+				if m == nil {
+					m = make(map[int][]int)
+					actualPreemptedJobsByExecutorIndexAndNodeIndex[executorIndex] = m
 				}
+				m[nodeIndex] = append(m[nodeIndex], jobIndex)
 			}
-
-			assert.Equal(t, len(expectedScheduledJobs), len(schedulerResult.ScheduledJobs))
-
-			schedulingContextByExecutor := schedulingContextRepo.GetMostRecentSchedulingContextByExecutor()
-
-			for executorId, jobIndices := range tc.expectedScheduledIndices {
-				sctx := schedulingContextByExecutor[executorId]
-				require.NotNil(t, sctx)
-
-				assert.Equal(t, len(jobIndices), sctx.NumScheduledJobs)
-
-				expectedScheduledResources := schedulerobjects.ResourceList{}
-				expectedScheduledResourcesByPriorityClass := make(schedulerobjects.QuantityByTAndResourceType[string])
-				for _, i := range jobIndices {
-					job := tc.queuedJobs[i]
-					expectedScheduledResources.AddV1ResourceList(job.GetResourceRequirements().Requests)
-					expectedScheduledResourcesByPriorityClass.AddV1ResourceList(job.GetPriorityClassName(), job.GetResourceRequirements().Requests)
+			for _, m := range actualPreemptedJobsByExecutorIndexAndNodeIndex {
+				for _, s := range m {
+					slices.Sort(s)
 				}
-				assert.True(t, expectedScheduledResources.Equal(sctx.ScheduledResources))
-				assert.True(t, expectedScheduledResourcesByPriorityClass.Equal(sctx.ScheduledResourcesByPriorityClass))
 			}
-
-			scheduledJobs := ScheduledJobsFromSchedulerResult[*jobdb.Job](schedulerResult)
-			for _, job := range scheduledJobs {
-				assert.Equal(t, false, job.Queued())
-				expectedExecutor, ok := expectedScheduledJobs[job.Id()]
-				require.True(t, ok)
-				run := job.LatestRun()
-				require.NotNil(t, run)
-				assert.Equal(t, expectedExecutor, run.Executor())
+			if len(tc.expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex) == 0 {
+				assert.Equal(t, 0, len(actualPreemptedJobsByExecutorIndexAndNodeIndex))
+			} else {
+				assert.Equal(t, tc.expectedPreemptedJobIndicesByExecutorIndexAndNodeIndex, actualPreemptedJobsByExecutorIndexAndNodeIndex)
 			}
 
-			// check all scheduled queuedJobs are up-to-date in db
+			// Check that jobs were scheduled as expected.
+			scheduledJobs := ScheduledJobsFromSchedulerResult[*jobdb.Job](schedulerResult)
+			actualScheduledIndices := make([]int, 0)
 			for _, job := range scheduledJobs {
-				dbJob := jobDb.GetById(txn, job.Id())
-				require.NoError(t, err)
-				assert.Equal(t, job, dbJob)
-			}
-
-			expectedPreemptedJobs := make([]string, 0)
-			for _, i := range tc.expectedPreemptedIndices {
-				expectedPreemptedJobs = append(expectedPreemptedJobs, tc.existingJobs[i].Id())
+				actualScheduledIndices = append(actualScheduledIndices, queueIndexByJobId[job.Id()])
 			}
-			slices.Sort(expectedPreemptedJobs)
-			preemptedJobs := make([]string, 0)
-			for _, job := range PreemptedJobsFromSchedulerResult[*jobdb.Job](schedulerResult) {
-				preemptedJobs = append(preemptedJobs, job.Id())
+			slices.Sort(actualScheduledIndices)
+			if len(tc.expectedScheduledIndices) == 0 {
+				assert.Equal(t, 0, len(actualScheduledIndices))
+			} else {
+				assert.Equal(t, tc.expectedScheduledIndices, actualScheduledIndices)
 			}
-			slices.Sort(preemptedJobs)
-			assert.Equal(t, expectedPreemptedJobs, preemptedJobs)
 
-			numPreemptedJobs := 0
-			for _, sctx := range schedulingContextByExecutor {
-				numPreemptedJobs += sctx.NumEvictedJobs
+			// Check that preempted jobs are marked as such consistently.
+			for _, job := range preemptedJobs {
+				dbJob := jobDb.GetById(txn, job.Id())
+				assert.True(t, dbJob.Failed())
+				assert.False(t, dbJob.Queued())
 			}
-			assert.Equal(t, len(tc.expectedPreemptedIndices), numPreemptedJobs)
 
-			expectedPreemptedResources := schedulerobjects.ResourceList{}
-			expectedPreemptedResourcesByPriorityClass := make(schedulerobjects.QuantityByTAndResourceType[string])
-			for _, i := range tc.expectedPreemptedIndices {
-				job := tc.existingJobs[i]
-				expectedPreemptedResources.AddV1ResourceList(job.GetResourceRequirements().Requests)
-				expectedPreemptedResourcesByPriorityClass.AddV1ResourceList(job.GetPriorityClassName(), job.GetResourceRequirements().Requests)
-			}
-			preemptedResources := schedulerobjects.ResourceList{}
-			preemptedResourcesByPriority := make(schedulerobjects.QuantityByTAndResourceType[string])
-			for _, sctx := range schedulingContextByExecutor {
-				for resourceType, quantity := range sctx.EvictedResources.Resources {
-					preemptedResources.AddQuantity(resourceType, quantity)
-				}
-				for p, rl := range sctx.EvictedResourcesByPriorityClass {
-					preemptedResourcesByPriority.AddResourceList(p, rl)
-				}
+			// Check that scheduled jobs are marked as such consistently.
+			for _, job := range scheduledJobs {
+				dbJob := jobDb.GetById(txn, job.Id())
+				assert.False(t, dbJob.Failed())
+				assert.False(t, dbJob.Queued())
+				dbRun := dbJob.LatestRun()
+				assert.False(t, dbRun.Failed())
+				assert.Equal(t, schedulerResult.NodeIdByJobId[dbJob.Id()], dbRun.Node())
 			}
-			assert.True(t, expectedPreemptedResources.Equal(preemptedResources))
-			assert.True(t, expectedPreemptedResourcesByPriorityClass.Equal(preemptedResourcesByPriority))
-		})
-	}
-}
-
-func TestGetExecutorsToSchedule(t *testing.T) {
-	executorA := testfixtures.Test1Node32CoreExecutor("a")
-	executorA1 := testfixtures.Test1Node32CoreExecutor("a1")
-	executorB := testfixtures.Test1Node32CoreExecutor("b")
-	executorC := testfixtures.Test1Node32CoreExecutor("c")
-
-	tests := map[string]struct {
-		executors          []*schedulerobjects.Executor
-		expectedExecutors  []*schedulerobjects.Executor
-		previousExecutorId string
-	}{
-		"sorts executors lexographically": {
-			executors: []*schedulerobjects.Executor{
-				executorB,
-				executorA,
-				executorA1,
-			},
-			expectedExecutors: []*schedulerobjects.Executor{
-				executorA,
-				executorA1,
-				executorB,
-			},
-			previousExecutorId: "",
-		},
-		"adjusts order based on previous executor id": {
-			executors: []*schedulerobjects.Executor{
-				executorC,
-				executorB,
-				executorA,
-			},
-			expectedExecutors: []*schedulerobjects.Executor{
-				executorB,
-				executorC,
-				executorA,
-			},
-			previousExecutorId: "a",
-		},
-		"previous executor id greater than any known executor": {
-			executors: []*schedulerobjects.Executor{
-				executorC,
-				executorA,
-				executorB,
-			},
-			expectedExecutors: []*schedulerobjects.Executor{
-				executorA,
-				executorB,
-				executorC,
-			},
-			previousExecutorId: "d",
-		},
-	}
-
-	for name, tc := range tests {
-		t.Run(name, func(t *testing.T) {
-			schedulingAlgoContext := fairSchedulingAlgoContext{executors: tc.executors}
-			result := schedulingAlgoContext.getExecutorsToSchedule(tc.previousExecutorId)
-			assert.Equal(t, tc.expectedExecutors, result)
-		})
-	}
-}
-
-type executorOrderingTest struct {
-	executors                           []*schedulerobjects.Executor
-	expectedExecutorsScheduled          []string
-	expectedPreviousScheduledExecutorId string
-}
-
-func TestLegacySchedulingAlgo_TestSchedule_ExecutorOrdering(t *testing.T) {
-	tests := map[string]struct {
-		onExecutorScheduled func(executor *schedulerobjects.Executor)
-		maxScheduleDuration time.Duration
-		rounds              []executorOrderingTest
-	}{
-		"considers all executors in order": {
-			onExecutorScheduled: func(executor *schedulerobjects.Executor) {},
-			maxScheduleDuration: time.Second * 1,
-			rounds: []executorOrderingTest{
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor2"),
-					},
-					expectedExecutorsScheduled:          []string{"executor1", "executor2"},
-					expectedPreviousScheduledExecutorId: "executor2",
-				},
-			},
-		},
-		"maintains state between schedule calls": {
-			onExecutorScheduled: func(executor *schedulerobjects.Executor) { time.Sleep(time.Millisecond * 200) },
-			maxScheduleDuration: time.Millisecond * 100,
-			rounds: []executorOrderingTest{
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor2"),
-					},
-					expectedExecutorsScheduled:          []string{"executor1"},
-					expectedPreviousScheduledExecutorId: "executor1",
-				},
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor2"),
-					},
-					expectedExecutorsScheduled:          []string{"executor2"},
-					expectedPreviousScheduledExecutorId: "executor2",
-				},
-			},
-		},
-		"handles executors changing between schedule calls": {
-			onExecutorScheduled: func(executor *schedulerobjects.Executor) { time.Sleep(time.Millisecond * 200) },
-			maxScheduleDuration: time.Millisecond * 100,
-			rounds: []executorOrderingTest{
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor3"),
-					},
-					expectedExecutorsScheduled:          []string{"executor1"},
-					expectedPreviousScheduledExecutorId: "executor1",
-				},
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor2"),
-						testfixtures.Test1Node32CoreExecutor("executor3"),
-					},
-					expectedExecutorsScheduled:          []string{"executor2"},
-					expectedPreviousScheduledExecutorId: "executor2",
-				},
-				{
-					executors: []*schedulerobjects.Executor{
-						testfixtures.Test1Node32CoreExecutor("executor1"),
-						testfixtures.Test1Node32CoreExecutor("executor2"),
-					},
-					expectedExecutorsScheduled:          []string{"executor1"},
-					expectedPreviousScheduledExecutorId: "executor1",
-				},
-			},
-		},
-	}
-	for name, tc := range tests {
-		t.Run(name, func(t *testing.T) {
-			ctx := testfixtures.ContextWithDefaultLogger(context.Background())
-			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-			defer cancel()
-			config := testfixtures.TestSchedulingConfig()
 
-			ctrl := gomock.NewController(t)
-			mockExecutorRepo := schedulermocks.NewMockExecutorRepository(ctrl)
-			mockQueueRepo := schedulermocks.NewMockQueueRepository(ctrl)
-			mockQueueRepo.EXPECT().GetAllQueues().Return([]*database.Queue{}, nil).AnyTimes()
-
-			algo, err := NewFairSchedulingAlgo(
-				config,
-				tc.maxScheduleDuration,
-				mockExecutorRepo,
-				mockQueueRepo,
-				nil,
-			)
-			require.NoError(t, err)
-			scheduledExecutorsIds := []string{}
-			// Use a test clock so we can control time
-			algo.clock = clock.NewFakeClock(testfixtures.BaseTime)
-			algo.onExecutorScheduled = func(executor *schedulerobjects.Executor) {
-				scheduledExecutorsIds = append(scheduledExecutorsIds, executor.Id)
-				tc.onExecutorScheduled(executor)
+			// Check that jobDb was updated correctly.
+			// TODO: Check that there are no unexpected jobs in the jobDb.
+			for _, job := range preemptedJobs {
+				dbJob := jobDb.GetById(txn, job.Id())
+				assert.Equal(t, job, dbJob)
 			}
-
-			// Set up JobDb
-			jobDb := jobdb.NewJobDb()
-
-			txn := jobDb.WriteTxn()
-			for _, round := range tc.rounds {
-				scheduledExecutorsIds = []string{}
-
-				roundMockExecutorRepo := schedulermocks.NewMockExecutorRepository(ctrl)
-				roundMockExecutorRepo.EXPECT().GetExecutors(ctx).Return(round.executors, nil).AnyTimes()
-				algo.executorRepository = roundMockExecutorRepo
-
-				_, err := algo.Schedule(ctx, txn, jobDb)
-				require.NoError(t, err)
-				assert.Equal(t, scheduledExecutorsIds, round.expectedExecutorsScheduled)
-				assert.Equal(t, round.expectedPreviousScheduledExecutorId, algo.previousScheduleClusterId)
+			for _, job := range scheduledJobs {
+				dbJob := jobDb.GetById(txn, job.Id())
+				assert.Equal(t, job, dbJob)
 			}
 		})
 	}
@@ -685,12 +476,13 @@ func BenchmarkNodeDbConstruction(b *testing.B) {
 					jobs[j] = jobs[j].WithNewRun("executor-01", node.Name)
 				}
 			}
-			rand.Shuffle(len(jobs), func(i, j int) { jobs[i], jobs[j] = jobs[j], jobs[i] })
+			armadaslices.Shuffle(jobs)
+			schedulingConfig := testfixtures.TestSchedulingConfig()
 			b.ResetTimer()
 			for n := 0; n < b.N; n++ {
 				b.StopTimer()
 				algo, err := NewFairSchedulingAlgo(
-					testfixtures.TestSchedulingConfig(),
+					schedulingConfig,
 					time.Second*5,
 					nil,
 					nil,
@@ -699,7 +491,15 @@ func BenchmarkNodeDbConstruction(b *testing.B) {
 				require.NoError(b, err)
 				b.StartTimer()
 
-				_, err = algo.constructNodeDb(testfixtures.TestPriorityClasses, jobs, nodes)
+				nodeDb, err := nodedb.NewNodeDb(
+					schedulingConfig.Preemption.PriorityClasses,
+					schedulingConfig.MaxExtraNodesToConsider,
+					schedulingConfig.IndexedResources,
+					schedulingConfig.IndexedTaints,
+					schedulingConfig.IndexedNodeLabels,
+				)
+				require.NoError(b, err)
+				err = algo.addExecutorToNodeDb(nodeDb, jobs, nodes)
 				require.NoError(b, err)
 			}
 		})
diff --git a/internal/scheduler/testfixtures/testfixtures.go b/internal/scheduler/testfixtures/testfixtures.go
index 06cae2cc..bbd91d2e 100644
--- a/internal/scheduler/testfixtures/testfixtures.go
+++ b/internal/scheduler/testfixtures/testfixtures.go
@@ -103,6 +103,11 @@ func TestSchedulingConfig() configuration.SchedulingConfig {
 	}
 }
 
+func WithUnifiedSchedulingByPoolConfig(config configuration.SchedulingConfig) configuration.SchedulingConfig {
+	config.UnifiedSchedulingByPool = true
+	return config
+}
+
 func WithMaxUnacknowledgedJobsPerExecutorConfig(v uint, config configuration.SchedulingConfig) configuration.SchedulingConfig {
 	config.MaxUnacknowledgedJobsPerExecutor = v
 	return config
@@ -656,6 +661,7 @@ func TestNode(priorities []int32, resources map[string]resource.Quantity) *sched
 	id := uuid.NewString()
 	return &schedulerobjects.Node{
 		Id:             id,
+		Name:           id,
 		TotalResources: schedulerobjects.ResourceList{Resources: resources},
 		AllocatableByPriorityAndResource: schedulerobjects.NewAllocatableByPriorityAndResourceType(
 			priorities,
@@ -709,11 +715,12 @@ func WithLastUpdateTimeExecutor(lastUpdateTime time.Time, executor *schedulerobj
 	return executor
 }
 
-func Test1Node32CoreExecutor(name string) *schedulerobjects.Executor {
+func Test1Node32CoreExecutor(executorId string) *schedulerobjects.Executor {
 	node := Test32CpuNode(TestPriorities)
-	node.Name = fmt.Sprintf("%s-node", name)
+	node.Name = fmt.Sprintf("%s-node", executorId)
+	node.Executor = executorId
 	return &schedulerobjects.Executor{
-		Id:             name,
+		Id:             executorId,
 		Pool:           TestPool,
 		Nodes:          []*schedulerobjects.Node{node},
 		LastUpdateTime: BaseTime,