Repository: armadaproject/armada
Pull request #2477 (2023-05-17 15:16:30)

[scheduler-ingester] Improvements to jobset operations

 - Now CancelJobSet and ReprioritiseJobSet correctly only operate on a specific queue. They were incorrectly updating all jobsets with a given name, ignoring queue
 - Made CancelJobSet so it handles the JobState filter
   - If Queued -  we cancel all queued
   - If Pending or Running - we cancel all running
   - No states means cancel the whole jobset
   - This matches the existing functionality. In future we may want better support for Pending vs Running - or change the API to only support Queued vs Leased

┆Issue is synchronized with this [Jira Task](https://gr-oss.atlassian.net/browse/BATCH-291) by [Unito](https://www.unito.io)


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/armadaproject/armada/pull/2477?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) Report
Patch coverage: **`83.33`**% and project coverage change: **`+0.11`** :tada:
> Comparison is base [(`8011b0a`)](https://app.codecov.io/gh/armadaproject/armada/commit/8011b0ac07d99827d7b6208ea97111c169d22d2f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) 58.37% compared to head [(`21ea773`)](https://app.codecov.io/gh/armadaproject/armada/pull/2477?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) 58.49%.

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


```diff
@@            Coverage Diff             @@
##           master    #2477      +/-   ##
==========================================
+ Coverage   58.37%   58.49%   +0.11%     
==========================================
  Files         235      235              
  Lines       29537    29572      +35     
==========================================
+ Hits        17243    17297      +54     
+ Misses      10976    10952      -24     
- Partials     1318     1323       +5     
```

| Flag | Coverage Δ | |
|---|---|---|
| armada-server | `58.49% <83.33%> (+0.11%)` | :arrow_up: |

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/2477?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) | Coverage Δ | |
|---|---|---|
| [internal/scheduleringester/schedulerdb.go](https://app.codecov.io/gh/armadaproject/armada/pull/2477?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvc2NoZWR1bGVyaW5nZXN0ZXIvc2NoZWR1bGVyZGIuZ28=) | `68.12% <74.19%> (+1.13%)` | :arrow_up: |
| [internal/scheduleringester/dbops.go](https://app.codecov.io/gh/armadaproject/armada/pull/2477?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvc2NoZWR1bGVyaW5nZXN0ZXIvZGJvcHMuZ28=) | `83.42% <76.92%> (ø)` | |
| [internal/scheduleringester/instructions.go](https://app.codecov.io/gh/armadaproject/armada/pull/2477?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvc2NoZWR1bGVyaW5nZXN0ZXIvaW5zdHJ1Y3Rpb25zLmdv) | `74.31% <100.00%> (+1.10%)` | :arrow_up: |

... and [1 file with indirect coverage changes](https://app.codecov.io/gh/armadaproject/armada/pull/2477/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/armadaproject/armada/pull/2477?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/scheduler/database/query.sql.go b/internal/scheduler/database/query.sql.go
index 9b58689e..29b9ed59 100644
--- a/internal/scheduler/database/query.sql.go
+++ b/internal/scheduler/database/query.sql.go
@@ -126,12 +126,32 @@ func (q *Queries) MarkJobsCancelRequestedById(ctx context.Context, jobIds []stri
 	return err
 }
 
-const markJobsCancelRequestedBySets = `-- name: MarkJobsCancelRequestedBySets :exec
-UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set = ANY($1::text[])
+const markJobsCancelRequestedBySet = `-- name: MarkJobsCancelRequestedBySet :exec
+UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set =$1 and queue = $2
 `
 
-func (q *Queries) MarkJobsCancelRequestedBySets(ctx context.Context, jobSets []string) error {
-	_, err := q.db.Exec(ctx, markJobsCancelRequestedBySets, jobSets)
+type MarkJobsCancelRequestedBySetParams struct {
+	JobSet string `db:"job_set"`
+	Queue  string `db:"queue"`
+}
+
+func (q *Queries) MarkJobsCancelRequestedBySet(ctx context.Context, arg MarkJobsCancelRequestedBySetParams) error {
+	_, err := q.db.Exec(ctx, markJobsCancelRequestedBySet, arg.JobSet, arg.Queue)
+	return err
+}
+
+const markJobsCancelRequestedBySetAndState = `-- name: MarkJobsCancelRequestedBySetAndState :exec
+UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set = $1 and queue = $2 and queued = $3
+`
+
+type MarkJobsCancelRequestedBySetAndStateParams struct {
+	JobSet string `db:"job_set"`
+	Queue  string `db:"queue"`
+	Queued bool   `db:"queued"`
+}
+
+func (q *Queries) MarkJobsCancelRequestedBySetAndState(ctx context.Context, arg MarkJobsCancelRequestedBySetAndStateParams) error {
+	_, err := q.db.Exec(ctx, markJobsCancelRequestedBySetAndState, arg.JobSet, arg.Queue, arg.Queued)
 	return err
 }
 
@@ -611,46 +631,17 @@ func (q *Queries) UpdateJobPriorityById(ctx context.Context, arg UpdateJobPriori
 }
 
 const updateJobPriorityByJobSet = `-- name: UpdateJobPriorityByJobSet :exec
-UPDATE jobs SET priority = $1 WHERE job_set = $2
+UPDATE jobs SET priority = $1 WHERE job_set = $2 and queue = $3
 `
 
 type UpdateJobPriorityByJobSetParams struct {
 	Priority int64  `db:"priority"`
 	JobSet   string `db:"job_set"`
+	Queue    string `db:"queue"`
 }
 
 func (q *Queries) UpdateJobPriorityByJobSet(ctx context.Context, arg UpdateJobPriorityByJobSetParams) error {
-	_, err := q.db.Exec(ctx, updateJobPriorityByJobSet, arg.Priority, arg.JobSet)
-	return err
-}
-
-const updateJobQueued = `-- name: UpdateJobQueued :exec
-UPDATE jobs SET (queued, queued_version) = ($1, $2) WHERE job_id = $3 AND queued_version < $2
-`
-
-type UpdateJobQueuedParams struct {
-	Queued        bool   `db:"queued"`
-	QueuedVersion int32  `db:"queued_version"`
-	JobID         string `db:"job_id"`
-}
-
-func (q *Queries) UpdateJobQueued(ctx context.Context, arg UpdateJobQueuedParams) error {
-	_, err := q.db.Exec(ctx, updateJobQueued, arg.Queued, arg.QueuedVersion, arg.JobID)
-	return err
-}
-
-const updateJobSchedulingInfo = `-- name: UpdateJobSchedulingInfo :exec
-UPDATE jobs SET (scheduling_info, scheduling_info_version) = ($1, $2) WHERE job_id = $3 AND scheduling_info_version < $2
-`
-
-type UpdateJobSchedulingInfoParams struct {
-	SchedulingInfo        []byte `db:"scheduling_info"`
-	SchedulingInfoVersion int32  `db:"scheduling_info_version"`
-	JobID                 string `db:"job_id"`
-}
-
-func (q *Queries) UpdateJobSchedulingInfo(ctx context.Context, arg UpdateJobSchedulingInfoParams) error {
-	_, err := q.db.Exec(ctx, updateJobSchedulingInfo, arg.SchedulingInfo, arg.SchedulingInfoVersion, arg.JobID)
+	_, err := q.db.Exec(ctx, updateJobPriorityByJobSet, arg.Priority, arg.JobSet, arg.Queue)
 	return err
 }
 
diff --git a/internal/scheduler/database/query/query.sql b/internal/scheduler/database/query/query.sql
index 7dbb60ff..4e34af03 100644
--- a/internal/scheduler/database/query/query.sql
+++ b/internal/scheduler/database/query/query.sql
@@ -8,10 +8,13 @@ SELECT job_id FROM jobs;
 SELECT job_id, job_set, queue, priority, submitted, queued, queued_version, cancel_requested, cancel_by_jobset_requested, cancelled, succeeded, failed, scheduling_info, scheduling_info_version, serial FROM jobs WHERE serial > $1 ORDER BY serial LIMIT $2;
 
 -- name: UpdateJobPriorityByJobSet :exec
-UPDATE jobs SET priority = $1 WHERE job_set = $2;
+UPDATE jobs SET priority = $1 WHERE job_set = $2 and queue = $3;
 
--- name: MarkJobsCancelRequestedBySets :exec
-UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set = ANY(sqlc.arg(job_sets)::text[]);
+-- name: MarkJobsCancelRequestedBySet :exec
+UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set =sqlc.arg(job_set) and queue = sqlc.arg(queue);
+
+-- name: MarkJobsCancelRequestedBySetAndState :exec
+UPDATE jobs SET cancel_by_jobset_requested = true WHERE job_set = sqlc.arg(job_set) and queue = sqlc.arg(queue) and queued = sqlc.arg(queued);
 
 -- name: MarkJobsSucceededById :exec
 UPDATE jobs SET succeeded = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
@@ -25,12 +28,6 @@ UPDATE jobs SET cancelled = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
 -- name: MarkJobsFailedById :exec
 UPDATE jobs SET failed = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
 
--- name: UpdateJobQueued :exec
-UPDATE jobs SET (queued, queued_version) = ($1, $2) WHERE job_id = $3 AND queued_version < $2;
-
--- name: UpdateJobSchedulingInfo :exec
-UPDATE jobs SET (scheduling_info, scheduling_info_version) = ($1, $2) WHERE job_id = $3 AND scheduling_info_version < $2;
-
 -- name: UpdateJobPriorityById :exec
 UPDATE jobs SET priority = $1 WHERE job_id = $2;
 
diff --git a/internal/scheduleringester/dbops.go b/internal/scheduleringester/dbops.go
index 51dbcde2..f4b448de 100644
--- a/internal/scheduleringester/dbops.go
+++ b/internal/scheduleringester/dbops.go
@@ -29,6 +29,21 @@ type JobSchedulingInfoUpdate struct {
 	JobSchedulingInfoVersion int32
 }
 
+type JobSetCancelAction struct {
+	cancelQueued bool
+	cancelLeased bool
+}
+
+type JobSetKey struct {
+	queue  string
+	jobSet string
+}
+
+type JobRunDetails struct {
+	queue string
+	dbRun *schedulerdb.Run
+}
+
 type JobQueuedStateUpdate struct {
 	Queued             bool
 	QueuedStateVersion int32
@@ -100,9 +115,9 @@ func discardNilOps(ops []DbOperation) []DbOperation {
 type InsertJobs map[string]*schedulerdb.Job
 
 type (
-	InsertRuns                 map[uuid.UUID]*schedulerdb.Run
-	UpdateJobSetPriorities     map[string]int64
-	MarkJobSetsCancelRequested map[string]bool
+	InsertRuns                 map[uuid.UUID]*JobRunDetails
+	UpdateJobSetPriorities     map[JobSetKey]int64
+	MarkJobSetsCancelRequested map[JobSetKey]*JobSetCancelAction
 	MarkJobsCancelRequested    map[string]bool
 	MarkJobsCancelled          map[string]bool
 	MarkJobsSucceeded          map[string]bool
@@ -120,16 +135,16 @@ type (
 )
 
 type JobSetOperation interface {
-	AffectsJobSet(string) bool
+	AffectsJobSet(queue string, jobSet string) bool
 }
 
-func (a UpdateJobSetPriorities) AffectsJobSet(jobSet string) bool {
-	_, ok := a[jobSet]
+func (a UpdateJobSetPriorities) AffectsJobSet(queue string, jobSet string) bool {
+	_, ok := a[JobSetKey{queue: queue, jobSet: jobSet}]
 	return ok
 }
 
-func (a MarkJobSetsCancelRequested) AffectsJobSet(jobSet string) bool {
-	_, ok := a[jobSet]
+func (a MarkJobSetsCancelRequested) AffectsJobSet(queue string, jobSet string) bool {
+	_, ok := a[JobSetKey{queue: queue, jobSet: jobSet}]
 	return ok
 }
 
@@ -250,7 +265,7 @@ func (a InsertJobs) CanBeAppliedBefore(b DbOperation) bool {
 	switch op := b.(type) {
 	case JobSetOperation:
 		for _, job := range a {
-			if op.AffectsJobSet(job.JobSet) {
+			if op.AffectsJobSet(job.Queue, job.JobSet) {
 				return false
 			}
 		}
@@ -264,13 +279,13 @@ func (a InsertRuns) CanBeAppliedBefore(b DbOperation) bool {
 	switch op := b.(type) {
 	case JobSetOperation:
 		for _, run := range a {
-			if op.AffectsJobSet(run.JobSet) {
+			if op.AffectsJobSet(run.queue, run.dbRun.JobSet) {
 				return false
 			}
 		}
 	case InsertJobs:
 		for _, run := range a {
-			if _, ok := op[run.JobID]; ok {
+			if _, ok := op[run.dbRun.JobID]; ok {
 				return false
 			}
 		}
@@ -342,10 +357,10 @@ func (a InsertJobRunErrors) CanBeAppliedBefore(_ DbOperation) bool {
 // definesJobInSet returns true if b is an InsertJobs operation
 // that inserts at least one job in any of the job sets that make
 // up the keys of a.
-func definesJobInSet[M ~map[string]V, V any](a M, b DbOperation) bool {
+func definesJobInSet[M ~map[JobSetKey]V, V any](a M, b DbOperation) bool {
 	if op, ok := b.(InsertJobs); ok {
 		for _, job := range op {
-			if _, ok := a[job.JobSet]; ok {
+			if _, ok := a[JobSetKey{queue: job.Queue, jobSet: job.JobSet}]; ok {
 				return true
 			}
 		}
@@ -354,10 +369,10 @@ func definesJobInSet[M ~map[string]V, V any](a M, b DbOperation) bool {
 }
 
 // Like definesJobInSet, but checks if b defines a run.
-func definesRunInSet[M ~map[string]V, V any](a M, b DbOperation) bool {
+func definesRunInSet[M ~map[JobSetKey]V, V any](a M, b DbOperation) bool {
 	if op, ok := b.(InsertRuns); ok {
 		for _, run := range op {
-			if _, ok := a[run.JobSet]; ok {
+			if _, ok := a[JobSetKey{queue: run.queue, jobSet: run.dbRun.JobSet}]; ok {
 				return true
 			}
 		}
@@ -383,7 +398,7 @@ func definesJob[M ~map[string]V, V any](a M, b DbOperation) bool {
 func definesRun[M ~map[uuid.UUID]V, V any](a M, b DbOperation) bool {
 	if op, ok := b.(InsertRuns); ok {
 		for _, run := range op {
-			if _, ok := a[run.RunID]; ok {
+			if _, ok := a[run.dbRun.RunID]; ok {
 				return true
 			}
 		}
@@ -396,7 +411,7 @@ func definesRun[M ~map[uuid.UUID]V, V any](a M, b DbOperation) bool {
 func definesRunForJob[M ~map[string]V, V any](a M, b DbOperation) bool {
 	if op, ok := b.(InsertRuns); ok {
 		for _, run := range op {
-			if _, ok := a[run.JobID]; ok {
+			if _, ok := a[run.dbRun.JobID]; ok {
 				return true
 			}
 		}
diff --git a/internal/scheduleringester/instructions.go b/internal/scheduleringester/instructions.go
index 6dd89c44..2166c842 100644
--- a/internal/scheduleringester/instructions.go
+++ b/internal/scheduleringester/instructions.go
@@ -7,6 +7,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/exp/slices"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/compress"
@@ -95,7 +96,7 @@ func (c *InstructionConverter) convertSequence(es *armadaevents.EventSequence) [
 		case *armadaevents.EventSequence_Event_CancelJob:
 			operationsFromEvent, err = c.handleCancelJob(event.GetCancelJob())
 		case *armadaevents.EventSequence_Event_CancelJobSet:
-			operationsFromEvent, err = c.handleCancelJobSet(meta.jobset)
+			operationsFromEvent, err = c.handleCancelJobSet(event.GetCancelJobSet(), meta)
 		case *armadaevents.EventSequence_Event_CancelledJob:
 			operationsFromEvent, err = c.handleCancelledJob(event.GetCancelledJob())
 		case *armadaevents.EventSequence_Event_JobRequeued:
@@ -183,12 +184,15 @@ func (c *InstructionConverter) handleJobRunLeased(jobRunLeased *armadaevents.Job
 		return nil, err
 	}
 	return []DbOperation{
-		InsertRuns{runId: &schedulerdb.Run{
-			RunID:    runId,
-			JobID:    jobId,
-			JobSet:   meta.jobset,
-			Executor: jobRunLeased.GetExecutorId(),
-			Node:     jobRunLeased.GetNodeId(),
+		InsertRuns{runId: &JobRunDetails{
+			queue: meta.queue,
+			dbRun: &schedulerdb.Run{
+				RunID:    runId,
+				JobID:    jobId,
+				JobSet:   meta.jobset,
+				Executor: jobRunLeased.GetExecutorId(),
+				Node:     jobRunLeased.GetNodeId(),
+			},
 		}},
 		UpdateJobQueuedState{jobId: &JobQueuedStateUpdate{
 			Queued:             false,
@@ -300,7 +304,7 @@ func (c *InstructionConverter) handleReprioritiseJob(reprioritiseJob *armadaeven
 
 func (c *InstructionConverter) handleReprioritiseJobSet(reprioritiseJobSet *armadaevents.ReprioritiseJobSet, meta eventSequenceCommon) ([]DbOperation, error) {
 	return []DbOperation{UpdateJobSetPriorities{
-		meta.jobset: int64(reprioritiseJobSet.Priority),
+		JobSetKey{queue: meta.queue, jobSet: meta.jobset}: int64(reprioritiseJobSet.Priority),
 	}}, nil
 }
 
@@ -314,9 +318,18 @@ func (c *InstructionConverter) handleCancelJob(cancelJob *armadaevents.CancelJob
 	}}, nil
 }
 
-func (c *InstructionConverter) handleCancelJobSet(jobset string) ([]DbOperation, error) {
+func (c *InstructionConverter) handleCancelJobSet(cancelJobSet *armadaevents.CancelJobSet, meta eventSequenceCommon) ([]DbOperation, error) {
+	cancelQueued := len(cancelJobSet.States) == 0 || slices.Contains(cancelJobSet.States, armadaevents.JobState_QUEUED)
+	cancelLeased := len(cancelJobSet.States) == 0 || slices.Contains(cancelJobSet.States, armadaevents.JobState_PENDING) || slices.Contains(cancelJobSet.States, armadaevents.JobState_RUNNING)
+
 	return []DbOperation{MarkJobSetsCancelRequested{
-		jobset: true,
+		JobSetKey{
+			queue:  meta.queue,
+			jobSet: meta.jobset,
+		}: &JobSetCancelAction{
+			cancelQueued: cancelQueued,
+			cancelLeased: cancelLeased,
+		},
 	}}, nil
 }
 
diff --git a/internal/scheduleringester/schedulerdb.go b/internal/scheduleringester/schedulerdb.go
index 027c90f7..6f7007e6 100644
--- a/internal/scheduleringester/schedulerdb.go
+++ b/internal/scheduleringester/schedulerdb.go
@@ -103,7 +103,7 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, tx pgx.Tx, op DbOperation)
 		records := make([]any, len(o))
 		i := 0
 		for _, v := range o {
-			records[i] = *v
+			records[i] = *v.dbRun
 			i++
 		}
 		err := database.Upsert(ctx, tx, "runs", records)
@@ -111,11 +111,12 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, tx pgx.Tx, op DbOperation)
 			return err
 		}
 	case UpdateJobSetPriorities:
-		for jobSet, priority := range o {
+		for jobSetInfo, priority := range o {
 			err := queries.UpdateJobPriorityByJobSet(
 				ctx,
 				schedulerdb.UpdateJobPriorityByJobSetParams{
-					JobSet:   jobSet,
+					JobSet:   jobSetInfo.jobSet,
+					Queue:    jobSetInfo.queue,
 					Priority: priority,
 				},
 			)
@@ -148,10 +149,36 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, tx pgx.Tx, op DbOperation)
 			return errors.WithStack(err)
 		}
 	case MarkJobSetsCancelRequested:
-		jobSets := maps.Keys(o)
-		err := queries.MarkJobsCancelRequestedBySets(ctx, jobSets)
-		if err != nil {
-			return errors.WithStack(err)
+		for jobSetInfo, cancelDetails := range o {
+			if !cancelDetails.cancelQueued && !cancelDetails.cancelLeased {
+				continue
+			}
+			if cancelDetails.cancelQueued && cancelDetails.cancelLeased {
+				err := queries.MarkJobsCancelRequestedBySet(
+					ctx,
+					schedulerdb.MarkJobsCancelRequestedBySetParams{
+						Queue:  jobSetInfo.queue,
+						JobSet: jobSetInfo.jobSet,
+					},
+				)
+				if err != nil {
+					return errors.WithStack(err)
+				}
+			} else {
+				queuedStateToCancel := cancelDetails.cancelQueued
+				err := queries.MarkJobsCancelRequestedBySetAndState(
+					ctx,
+					schedulerdb.MarkJobsCancelRequestedBySetAndStateParams{
+						Queue:  jobSetInfo.queue,
+						JobSet: jobSetInfo.jobSet,
+						Queued: queuedStateToCancel,
+					},
+				)
+				if err != nil {
+					return errors.WithStack(err)
+				}
+			}
+
 		}
 	case MarkJobsCancelRequested:
 		jobIds := maps.Keys(o)


Test patch:
diff --git a/internal/common/ingest/testfixtures/event.go b/internal/common/ingest/testfixtures/event.go
index d919e5d2..561e2673 100644
--- a/internal/common/ingest/testfixtures/event.go
+++ b/internal/common/ingest/testfixtures/event.go
@@ -500,6 +500,17 @@ var JobSucceeded = &armadaevents.EventSequence_Event{
 	},
 }
 
+func JobSetCancelRequestedWithStateFilter(states ...armadaevents.JobState) *armadaevents.EventSequence_Event {
+	return &armadaevents.EventSequence_Event{
+		Created: &testfixtures.BaseTime,
+		Event: &armadaevents.EventSequence_Event_CancelJobSet{
+			CancelJobSet: &armadaevents.CancelJobSet{
+				States: states,
+			},
+		},
+	}
+}
+
 func DeepCopy(events *armadaevents.EventSequence_Event) (*armadaevents.EventSequence_Event, error) {
 	bytes, err := proto.Marshal(events)
 	if err != nil {
diff --git a/internal/scheduleringester/dbops_test.go b/internal/scheduleringester/dbops_test.go
index 211b5046..385ec158 100644
--- a/internal/scheduleringester/dbops_test.go
+++ b/internal/scheduleringester/dbops_test.go
@@ -12,6 +12,8 @@ import (
 	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
 )
 
+const testQueueName = "test"
+
 func TestMerge(t *testing.T) {
 	jobId1 := util.NewULID()
 	jobId2 := util.NewULID()
@@ -107,51 +109,51 @@ func TestDbOperationOptimisation(t *testing.T) {
 		Ops []DbOperation // Ops sequence to optimise.
 	}{
 		"InsertJobs": {N: 1, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set2"}}, // 1
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 1
-			InsertJobs{jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"}}, // 1
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"}}, // 1
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2"}}, // 1
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"}}, // 1
+			InsertJobs{jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2"}}, // 1
 		}},
 		"InsertJobs, InsertRuns": {N: 2, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                   // 1
-			InsertRuns{runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}, // 2
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                   // 2
-			InsertRuns{runIds[1]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}, // 2
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                   // 2
-			InsertRuns{runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}}, // 2
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                                                                // 1
+			InsertRuns{runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                                                                // 2
+			InsertRuns{runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 2
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                                                                // 2
+			InsertRuns{runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}}}, // 2
 		}},
 		"UpdateJobSetPriorities": {N: 3, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			UpdateJobSetPriorities{"set1": 1},                                         // 2
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set1"}}, // 3
-			UpdateJobSetPriorities{"set2": 2},                                         // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 3
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"}}, // 1
+			UpdateJobSetPriorities{JobSetKey{queue: testQueueName, jobSet: "set1"}: 1},                      // 2
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set1"}}, // 3
+			UpdateJobSetPriorities{JobSetKey{queue: testQueueName, jobSet: "set2"}: 2},                      // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"}}, // 3
 		}},
 		"UpdateJobSetPriorities, UpdateJobPriorities": {N: 4, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set1"}}, // 1
-			UpdateJobPriorities{jobIds[0]: 1},                                         // 2
-			UpdateJobSetPriorities{"set1": 2},                                         // 3
-			UpdateJobPriorities{jobIds[1]: 3},                                         // 4
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set2"}}, // 4
-			UpdateJobPriorities{jobIds[1]: 4},                                         // 4
-			UpdateJobPriorities{jobIds[2]: 5},                                         // 4
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"}}, // 1
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set1"}}, // 1
+			UpdateJobPriorities{jobIds[0]: 1},                                                               // 2
+			UpdateJobSetPriorities{JobSetKey{queue: testQueueName, jobSet: "set1"}: 2},                      // 3
+			UpdateJobPriorities{jobIds[1]: 3},                                                               // 4
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set2"}}, // 4
+			UpdateJobPriorities{jobIds[1]: 4},                                                               // 4
+			UpdateJobPriorities{jobIds[2]: 5},                                                               // 4
 		}},
 		"MarkJobSetsCancelRequested": {N: 3, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			MarkJobSetsCancelRequested{"set1": true},                                  // 2
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set1"}}, // 3
-			MarkJobSetsCancelRequested{"set2": true},                                  // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 3
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"}},                                          // 1
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}}, // 2
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set1"}},                                          // 3
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set2"}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}}, // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"}},                                          // 3
 		}},
 		"MarkJobSetsCancelRequested, MarkJobsCancelRequested": {N: 4, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set1"}}, // 1
-			MarkJobsCancelRequested{jobIds[0]: true},                                  // 2
-			MarkJobSetsCancelRequested{"set1": true},                                  // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 4
-			MarkJobsCancelRequested{jobIds[1]: true},                                  // 4
-			MarkJobsCancelRequested{jobIds[2]: true},                                  // 4
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"}},                                          // 1
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set1"}},                                          // 1
+			MarkJobsCancelRequested{jobIds[0]: true},                                                                                                 // 2
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}}, // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"}},                                          // 4
+			MarkJobsCancelRequested{jobIds[1]: true},                                                                                                 // 4
+			MarkJobsCancelRequested{jobIds[2]: true},                                                                                                 // 4
 		}},
 		"MarkJobsSucceeded": {N: 2, Ops: []DbOperation{
 			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}}, // 1
@@ -175,31 +177,31 @@ func TestDbOperationOptimisation(t *testing.T) {
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}}, // 2
 		}},
 		"MarkRunsSucceeded": {N: 3, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                   // 1
-			InsertRuns{runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}, // 2
-			MarkRunsSucceeded{runIds[0]: true},                                          // 3
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                   // 3
-			InsertRuns{runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}}, // 3
-			MarkRunsSucceeded{runIds[1]: true},                                          // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                   // 3
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                                                                // 1
+			InsertRuns{runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			MarkRunsSucceeded{runIds[0]: true},                                                                                       // 3
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                                                                // 3
+			InsertRuns{runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
+			MarkRunsSucceeded{runIds[1]: true},                                                                                       // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                                                                // 3
 		}},
 		"MarkRunsFailed": {N: 3, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                   // 1
-			InsertRuns{runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}, // 2
-			MarkRunsFailed{runIds[0]: &JobRunFailed{true, true}},                        // 3
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                   // 3
-			InsertRuns{runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}}, // 3
-			MarkRunsFailed{runIds[1]: &JobRunFailed{true, true}},                        // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                   // 3
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                                                                // 1
+			InsertRuns{runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			MarkRunsFailed{runIds[0]: &JobRunFailed{true, true}},                                                                     // 3
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                                                                // 3
+			InsertRuns{runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
+			MarkRunsFailed{runIds[1]: &JobRunFailed{true, true}},                                                                     // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                                                                // 3
 		}},
 		"MarkRunsRunning": {N: 3, Ops: []DbOperation{
-			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                   // 1
-			InsertRuns{runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}, // 2
-			MarkRunsRunning{runIds[0]: true},                                            // 3
-			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}},                   // 3
-			InsertRuns{runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}}, // 3
-			MarkRunsRunning{runIds[1]: true},                                            // 3
-			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                   // 3
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}},                                                                // 1
+			InsertRuns{runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			MarkRunsRunning{runIds[0]: true},                          // 3
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}}, // 3
+			InsertRuns{runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
+			MarkRunsRunning{runIds[1]: true},                          // 3
+			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}}, // 3
 		}},
 		"InsertPartitionMarker": {N: 2, Ops: []DbOperation{
 			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}}, // 1
@@ -262,25 +264,25 @@ func TestInsertJobRequestCancel(t *testing.T) {
 	var ops []DbOperation
 	expectedCancelledIds := make(map[string]bool)
 	for i := 0; i < 2; i++ {
-		job := &schedulerdb.Job{JobID: util.NewULID(), JobSet: "set1"}
+		job := &schedulerdb.Job{JobID: util.NewULID(), Queue: testQueueName, JobSet: "set1"}
 		expectedCancelledIds[job.JobID] = true
 		ops = append(ops, InsertJobs{job.JobID: job})
 	}
 	for i := 0; i < 2; i++ {
-		job := &schedulerdb.Job{JobID: util.NewULID(), JobSet: "set2"}
+		job := &schedulerdb.Job{JobID: util.NewULID(), Queue: testQueueName, JobSet: "set2"}
 		ops = append(ops, InsertJobs{job.JobID: job})
 	}
 
 	// Cancel one job set.
-	ops = append(ops, MarkJobSetsCancelRequested{"set1": true})
+	ops = append(ops, MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}})
 
 	// Submit some more jobs to both job sets.
 	for i := 0; i < 2; i++ {
-		job := &schedulerdb.Job{JobID: util.NewULID(), JobSet: "set2"}
+		job := &schedulerdb.Job{JobID: util.NewULID(), Queue: testQueueName, JobSet: "set2"}
 		ops = append(ops, InsertJobs{job.JobID: job})
 	}
 	for i := 0; i < 2; i++ {
-		job := &schedulerdb.Job{JobID: util.NewULID(), JobSet: "set1"}
+		job := &schedulerdb.Job{JobID: util.NewULID(), Queue: testQueueName, JobSet: "set1"}
 		ops = append(ops, InsertJobs{job.JobID: job})
 	}
 
@@ -358,24 +360,24 @@ func (db *mockDb) apply(op DbOperation) error {
 	case InsertRuns:
 		n := len(db.Runs)
 		for _, run := range o {
-			run := *run // Copy primitive types
+			run := *run.dbRun // Copy primitive types
 			db.Runs[run.RunID] = &run
 		}
 		if len(db.Runs) != n+len(o) {
 			return errors.New("duplicate run id")
 		}
 	case UpdateJobSetPriorities:
-		for jobSet, priority := range o {
+		for jobSetKey, priority := range o {
 			for _, job := range db.Jobs {
-				if job.JobSet == jobSet {
+				if job.JobSet == jobSetKey.jobSet && job.Queue == jobSetKey.queue {
 					job.Priority = priority
 				}
 			}
 		}
 	case MarkJobSetsCancelRequested:
-		for jobSet := range o {
+		for jobSetKey := range o {
 			for _, job := range db.Jobs {
-				if job.JobSet == jobSet {
+				if job.JobSet == jobSetKey.jobSet && job.Queue == jobSetKey.queue {
 					job.CancelRequested = true
 				}
 			}
diff --git a/internal/scheduleringester/instructions_test.go b/internal/scheduleringester/instructions_test.go
index a0815080..fc95ec09 100644
--- a/internal/scheduleringester/instructions_test.go
+++ b/internal/scheduleringester/instructions_test.go
@@ -52,13 +52,13 @@ func TestConvertSequence(t *testing.T) {
 		"job run leased": {
 			events: []*armadaevents.EventSequence_Event{f.Leased},
 			expected: []DbOperation{
-				InsertRuns{f.RunIdUuid: &schedulerdb.Run{
+				InsertRuns{f.RunIdUuid: &JobRunDetails{queue: f.Queue, dbRun: &schedulerdb.Run{
 					RunID:    f.RunIdUuid,
 					JobID:    f.JobIdString,
 					JobSet:   f.JobSetName,
 					Executor: f.ExecutorId,
 					Node:     f.NodeName,
-				}},
+				}}},
 				UpdateJobQueuedState{f.JobIdString: &JobQueuedStateUpdate{
 					Queued:             false,
 					QueuedStateVersion: 1,
@@ -116,7 +116,7 @@ func TestConvertSequence(t *testing.T) {
 		"reprioritise jobset": {
 			events: []*armadaevents.EventSequence_Event{f.JobSetReprioritiseRequested},
 			expected: []DbOperation{
-				UpdateJobSetPriorities{f.JobSetName: f.NewPriority},
+				UpdateJobSetPriorities{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: f.NewPriority},
 			},
 		},
 		"JobCancelRequested": {
@@ -128,7 +128,25 @@ func TestConvertSequence(t *testing.T) {
 		"JobSetCancelRequested": {
 			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequested},
 			expected: []DbOperation{
-				MarkJobSetsCancelRequested{f.JobSetName: true},
+				MarkJobSetsCancelRequested{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}},
+			},
+		},
+		"JobSetCancelRequested - Queued only": {
+			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequestedWithStateFilter(armadaevents.JobState_QUEUED)},
+			expected: []DbOperation{
+				MarkJobSetsCancelRequested{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: &JobSetCancelAction{cancelQueued: true, cancelLeased: false}},
+			},
+		},
+		"JobSetCancelRequested - Pending only": {
+			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequestedWithStateFilter(armadaevents.JobState_PENDING)},
+			expected: []DbOperation{
+				MarkJobSetsCancelRequested{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: &JobSetCancelAction{cancelQueued: false, cancelLeased: true}},
+			},
+		},
+		"JobSetCancelRequested - Running only": {
+			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequestedWithStateFilter(armadaevents.JobState_RUNNING)},
+			expected: []DbOperation{
+				MarkJobSetsCancelRequested{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: &JobSetCancelAction{cancelQueued: false, cancelLeased: true}},
 			},
 		},
 		"JobCancelled": {
@@ -165,7 +183,7 @@ func TestConvertSequence(t *testing.T) {
 		"multiple events": {
 			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequested, f.Running, f.JobSucceeded},
 			expected: []DbOperation{
-				MarkJobSetsCancelRequested{f.JobSetName: true},
+				MarkJobSetsCancelRequested{JobSetKey{queue: f.Queue, jobSet: f.JobSetName}: &JobSetCancelAction{cancelQueued: true, cancelLeased: true}},
 				MarkRunsRunning{f.RunIdUuid: true},
 				MarkJobsSucceeded{f.JobIdString: true},
 			},
diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go
index b810308d..e390950a 100644
--- a/internal/scheduleringester/schedulerdb_test.go
+++ b/internal/scheduleringester/schedulerdb_test.go
@@ -43,18 +43,18 @@ func TestWriteOps(t *testing.T) {
 		}},
 		"InsertRuns": {Ops: []DbOperation{
 			InsertJobs{
-				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"},
-				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set2"},
-				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"},
-				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
+				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"},
+				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2"},
+				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"},
+				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2"},
 			},
 			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]},
+				runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}},
+				runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}},
 			},
 			InsertRuns{
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
+				runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}},
+				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]}},
 			},
 			UpdateJobQueuedState{
 				jobIds[0]: &JobQueuedStateUpdate{Queued: false, QueuedStateVersion: 1},
@@ -65,12 +65,13 @@ func TestWriteOps(t *testing.T) {
 		}},
 		"UpdateJobSetPriorities": {Ops: []DbOperation{
 			InsertJobs{
-				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"},
-				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set2"},
-				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"},
-				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
+				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"},
+				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2"},
+				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"},
+				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2"},
+				jobIds[4]: &schedulerdb.Job{JobID: jobIds[4], Queue: "queue-2", JobSet: "set1"},
 			},
-			UpdateJobSetPriorities{"set1": 1},
+			UpdateJobSetPriorities{JobSetKey{queue: testQueueName, jobSet: "set1"}: 1},
 		}},
 		"UpdateJobPriorities": {Ops: []DbOperation{
 			InsertJobs{
@@ -86,12 +87,31 @@ func TestWriteOps(t *testing.T) {
 		}},
 		"MarkJobSetsCancelRequested": {Ops: []DbOperation{
 			InsertJobs{
-				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"},
-				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set2"},
-				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"},
-				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
+				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1"},
+				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2"},
+				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1"},
+				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2"},
+				jobIds[4]: &schedulerdb.Job{JobID: jobIds[4], Queue: "queue-2", JobSet: "set1"},
+			},
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelLeased: true, cancelQueued: true}},
+		}},
+		"MarkJobSetsCancelRequested - Queued only": {Ops: []DbOperation{
+			InsertJobs{
+				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1", Queued: true},
+				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2", Queued: true},
+				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1", Queued: false},
+				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2", Queued: false},
+			},
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelLeased: false, cancelQueued: true}},
+		}},
+		"MarkJobSetsCancelRequested - Leased only": {Ops: []DbOperation{
+			InsertJobs{
+				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], Queue: testQueueName, JobSet: "set1", Queued: true},
+				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], Queue: testQueueName, JobSet: "set2", Queued: true},
+				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], Queue: testQueueName, JobSet: "set1", Queued: false},
+				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2", Queued: false},
 			},
-			MarkJobSetsCancelRequested{"set1": true},
+			MarkJobSetsCancelRequested{JobSetKey{queue: testQueueName, jobSet: "set1"}: &JobSetCancelAction{cancelLeased: true, cancelQueued: false}},
 		}},
 		"MarkJobsCancelRequested": {Ops: []DbOperation{
 			InsertJobs{
@@ -113,10 +133,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]},
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
+				runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}},
+				runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}},
+				runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}},
+				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]}},
 			},
 			MarkJobsCancelled{
 				jobIds[0]: true,
@@ -155,10 +175,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]},
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
+				runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}},
+				runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}},
+				runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}},
+				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]}},
 			},
 			MarkRunsSucceeded{
 				runIds[0]: true,
@@ -203,10 +223,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]},
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
+				runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}},
+				runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}},
+				runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}},
+				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]}},
 			},
 			MarkRunsFailed{
 				runIds[0]: &JobRunFailed{LeaseReturned: true},
@@ -222,10 +242,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]},
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
+				runIds[0]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}},
+				runIds[1]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1]}},
+				runIds[2]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2]}},
+				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]}},
 			},
 			MarkRunsRunning{
 				runIds[0]: true,
@@ -351,11 +371,11 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		actual := make(InsertRuns)
 		for _, run := range runs {
 			run := run
-			actual[run.RunID] = &run
+			actual[run.RunID] = &JobRunDetails{queue: testQueueName, dbRun: &run}
 			serials["runs"] = max(serials["runs"], run.Serial)
 			if v, ok := expected[run.RunID]; ok {
-				v.Serial = run.Serial
-				v.LastModified = run.LastModified
+				v.dbRun.Serial = run.Serial
+				v.dbRun.LastModified = run.LastModified
 			}
 		}
 		assert.Equal(t, expected, actual)
@@ -394,7 +414,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		numChanged := 0
 		for _, job := range jobs {
-			if e, ok := expected[job.JobSet]; ok {
+			if e, ok := expected[JobSetKey{queue: job.Queue, jobSet: job.JobSet}]; ok {
 				assert.Equal(t, e, job.Priority)
 				numChanged++
 			}
@@ -408,7 +428,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		numChanged := 0
 		jobIds := make([]string, 0)
 		for _, job := range jobs {
-			if _, ok := expected[job.JobSet]; ok {
+			if _, ok := expected[JobSetKey{queue: job.Queue, jobSet: job.JobSet}]; ok {
 				assert.True(t, job.CancelByJobsetRequested)
 				numChanged++
 				jobIds = append(jobIds, job.JobID)
@@ -622,7 +642,7 @@ func TestStore(t *testing.T) {
 			},
 		},
 		InsertRuns{
-			runId: &schedulerdb.Run{JobID: jobId, RunID: runId},
+			runId: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobId, RunID: runId}},
 		},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)