Repository: armadaproject/armada
Pull request #2060 (2023-01-25 14:58:02)

Bring Scheduler Ingester Up To Date
In the course of the writing the scheduler, some features have been implemented which needed work to be done on the scheduler ingester.  This PR address that. Specific features are:

* Scheduler ingester now persists terminal job run errors.
* Scheduler ingester no longer now handles job cancel requested messages
* Scheduler ingester no longer cancels job runs when on receipt of a job cancel message
* When a job run error indicates that the lease was returned, the ingester marks it as such in the db.

I've also added some indexes to the scheduler database which we will need for efficient access.  The only one that worries me is the indexes on `serial` which we have added to the `jobs` and `runs` table.  That's because we update the `serial` on *every* row insert so it's likely that we'll be constantly rebuilding this index.  Whether or not this is is an issue in practise we'll have to test.

codecov-commenter: # [Codecov](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) Report
Base: **93.83**% // Head: **91.87**% // Decreases project coverage by **`-1.96%`** :warning:
> Coverage data is based on head [(`921678e`)](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) compared to base [(`221d03b`)](https://codecov.io/gh/armadaproject/armada/commit/221d03b69919e30cd3f38f7450577ec2626b244b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).
> Patch has no changes to coverable lines.

> :exclamation: Current head 921678e differs from pull request most recent head b23b7d5. Consider uploading reports for the commit b23b7d5 to get more accurate results

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject)

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


```diff
@@            Coverage Diff             @@
##           master    #2060      +/-   ##
==========================================
- Coverage   93.83%   91.87%   -1.96%     
==========================================
  Files           9       18       +9     
  Lines         389      665     +276     
==========================================
+ Hits          365      611     +246     
- Misses         24       54      +30     
```

| Flag | Coverage Δ | |
|---|---|---|
| airflow-operator | `89.13% <ø> (?)` | |
| python-client | `93.83% <ø> (ø)` | |

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://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) | Coverage Δ | |
|---|---|---|
| [third\_party/airflow/armada/operators/armada.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy9hcm1hZGEvb3BlcmF0b3JzL2FybWFkYS5weQ==) | `70.17% <0.00%> (ø)` | |
| [...ird\_party/airflow/tests/unit/test\_airflow\_error.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L3Rlc3RfYWlyZmxvd19lcnJvci5weQ==) | `100.00% <0.00%> (ø)` | |
| [...airflow/tests/unit/test\_search\_for\_job\_complete.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L3Rlc3Rfc2VhcmNoX2Zvcl9qb2JfY29tcGxldGUucHk=) | `100.00% <0.00%> (ø)` | |
| [...ird\_party/airflow/tests/unit/armada\_client\_mock.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L2FybWFkYV9jbGllbnRfbW9jay5weQ==) | `65.21% <0.00%> (ø)` | |
| [third\_party/airflow/armada/operators/jobservice.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy9hcm1hZGEvb3BlcmF0b3JzL2pvYnNlcnZpY2UucHk=) | `100.00% <0.00%> (ø)` | |
| [...y/airflow/tests/unit/test\_airflow\_operator\_mock.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L3Rlc3RfYWlyZmxvd19vcGVyYXRvcl9tb2NrLnB5) | `100.00% <0.00%> (ø)` | |
| [...d\_party/airflow/tests/unit/test\_armada\_operator.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L3Rlc3RfYXJtYWRhX29wZXJhdG9yLnB5) | `100.00% <0.00%> (ø)` | |
| [third\_party/airflow/armada/operators/utils.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy9hcm1hZGEvb3BlcmF0b3JzL3V0aWxzLnB5) | `91.07% <0.00%> (ø)` | |
| [third\_party/airflow/tests/unit/job\_service\_mock.py](https://codecov.io/gh/armadaproject/armada/pull/2060?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-dGhpcmRfcGFydHkvYWlyZmxvdy90ZXN0cy91bml0L2pvYl9zZXJ2aWNlX21vY2sucHk=) | `100.00% <0.00%> (ø)` | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/armadaproject/armada/pull/2060?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/cmd/scheduleringester/main.go b/cmd/scheduleringester/main.go
index 27ef02b6..5d20f3f3 100644
--- a/cmd/scheduleringester/main.go
+++ b/cmd/scheduleringester/main.go
@@ -29,5 +29,5 @@ func main() {
 
 	common.LoadConfig(&config, "./config/scheduleringester", userSpecifiedConfigs)
 
-	scheduleringester.Run(&config)
+	scheduleringester.Run(config)
 }
diff --git a/internal/common/compress/list.go b/internal/common/compress/list.go
index 9aa26674..72f34f7f 100644
--- a/internal/common/compress/list.go
+++ b/internal/common/compress/list.go
@@ -20,6 +20,14 @@ func CompressStringArray(input []string, compressor Compressor) ([]byte, error)
 	return compressor.Compress(bs)
 }
 
+func MustCompressStringArray(input []string, compressor Compressor) []byte {
+	b, err := CompressStringArray(input, compressor)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
 func DecompressStringArray(input []byte, decompressor Decompressor) ([]string, error) {
 	if len(input) <= 0 {
 		return []string{}, nil
@@ -36,3 +44,11 @@ func DecompressStringArray(input []byte, decompressor Decompressor) ([]string, e
 	err = dec.Decode(&data)
 	return data, err
 }
+
+func MustDecompressStringArray(input []byte, compressor Decompressor) []string {
+	data, err := DecompressStringArray(input, compressor)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
diff --git a/internal/common/proto/protoutil.go b/internal/common/proto/protoutil.go
new file mode 100644
index 00000000..a222966c
--- /dev/null
+++ b/internal/common/proto/protoutil.go
@@ -0,0 +1,74 @@
+package protoutil
+
+import (
+	"github.com/gogo/protobuf/proto"
+	"github.com/pkg/errors"
+
+	"github.com/armadaproject/armada/internal/common/compress"
+)
+
+// Unmarshall unmarshalls a proto message in a type-safe way.
+// Hopefully go-generics will one day be powerful enough that we won't need the msg param
+func Unmarshall[T proto.Message](buf []byte, msg T) (T, error) {
+	err := proto.Unmarshal(buf, msg)
+	return msg, err
+}
+
+// DecompressAndUnmarshall first decompressed the message and then unmarshalls
+func DecompressAndUnmarshall[T proto.Message](buf []byte, msg T, decompressor compress.Decompressor) (T, error) {
+	decompressed, err := decompressor.Decompress(buf)
+	if err != nil {
+		return msg, err
+	}
+	return Unmarshall[T](decompressed, msg)
+}
+
+// MustUnmarshall unmarshalls a proto message and panics if the unmarshall fails.  The main use case here is for unit tests.
+// Think carefully if you intend to use this elsewhere
+func MustUnmarshall[T proto.Message](buf []byte, msg T) T {
+	msg, err := Unmarshall[T](buf, msg)
+	if err != nil {
+		panic(errors.Wrap(err, "Error unmarshalling object"))
+	}
+	return msg
+}
+
+// MustDecompressAndUnmarshall first decompressed the message and then unmarshalls.  If either of these steps fail then it will panic.
+// The main use case here is for unit tests. Think carefully if you intend to use this elsewhere
+func MustDecompressAndUnmarshall[T proto.Message](buf []byte, msg T, decompressor compress.Decompressor) T {
+	msg, err := DecompressAndUnmarshall[T](buf, msg, decompressor)
+	if err != nil {
+		panic(errors.Wrap(err, "Error unmarshalling object"))
+	}
+	return msg
+}
+
+// MarshallAndCompress first marshalls the supplied proto message and then compresses it.
+func MarshallAndCompress(msg proto.Message, compressor compress.Compressor) ([]byte, error) {
+	b, err := proto.Marshal(msg)
+	if err != nil {
+		return nil, errors.Wrap(err, "Error marshalling object")
+	}
+	return compressor.Compress(b)
+}
+
+// MustMarshall marshalls a proto message and panics if the unmarshall fails.  The main use case here is for unit tests.
+// Think carefully if you intend to use this elsewhere
+func MustMarshall(msg proto.Message) []byte {
+	b, err := proto.Marshal(msg)
+	if err != nil {
+		panic(errors.Wrap(err, "Error marshalling object"))
+	}
+	return b
+}
+
+// MustMarshallAndCompress first marshalls the supplied proto message and then compresses it.
+// If either of these steps fail then it will panic. The main use case here is for unit tests.
+// Think carefully if you intend to use this elsewhere
+func MustMarshallAndCompress(msg proto.Message, compressor compress.Compressor) []byte {
+	b, err := MarshallAndCompress(msg, compressor)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
diff --git a/internal/scheduler/database/migrations/001_initialize_schema.up.sql b/internal/scheduler/database/migrations/001_initialize_schema.up.sql
index e9fe25c3..2438c044 100644
--- a/internal/scheduler/database/migrations/001_initialize_schema.up.sql
+++ b/internal/scheduler/database/migrations/001_initialize_schema.up.sql
@@ -27,7 +27,7 @@ CREATE TABLE jobs (
     serial bigserial NOT NULL,
     last_modified timestamptz NOT NULL
 );
-
+CREATE UNIQUE INDEX idx_jobs_serial ON jobs (serial);
 ALTER TABLE jobs ALTER COLUMN groups SET STORAGE EXTERNAL;
 ALTER TABLE jobs ALTER COLUMN submit_message SET STORAGE EXTERNAL;
 
@@ -52,10 +52,14 @@ CREATE TABLE runs (
     serial bigserial NOT NULL,
     last_modified timestamptz NOT NULL
 );
+CREATE UNIQUE INDEX idx_runs_serial ON runs (serial);
+CREATE INDEX idx_runs_job_id ON runs (job_id);
+CREATE INDEX idx_runs_job_set ON runs (job_set);
 
 CREATE TABLE markers (
     group_id uuid NOT NULL,
     partition_id integer NOT NULL,
+    created timestamptz default now(),
     PRIMARY KEY (group_id, partition_id)
 );
 
@@ -68,15 +72,15 @@ CREATE TABLE executors (
     last_updated timestamptz NOT NULL
 );
 
+-- stores terminal job run errors
+-- we need this because the job error generated by the scheduler needs to contain the run error
 CREATE TABLE job_run_errors (
     run_id uuid PRIMARY KEY,
-    -- Byte array containing a JobRunErrors proto message.
-    error bytea NOT NULL,
-    -- Indicates if this error is terminal.
-    -- The presence of a terminal error indicates this job run has failed.
-    serial bigserial NOT NULL,
-    last_modified timestamptz NOT NULL
+    job_id text NOT NULL,
+    -- Byte array containing a compressed JobRunErrors proto message.
+    error bytea NOT NULL
 );
+CREATE INDEX idx_job_run_errors_job_id ON job_run_errors (job_id);
 
 ALTER TABLE job_run_errors ALTER COLUMN error SET STORAGE EXTERNAL;
 
@@ -107,8 +111,3 @@ CREATE TRIGGER next_serial_on_insert_runs
     BEFORE INSERT or UPDATE ON runs
     FOR EACH ROW
 EXECUTE FUNCTION trg_increment_serial_set_last_modified();
-
-CREATE TRIGGER next_serial_on_insert_job_run_errors
-    BEFORE INSERT or UPDATE ON job_run_errors
-    FOR EACH ROW
-EXECUTE FUNCTION trg_increment_serial_set_last_modified();
diff --git a/internal/scheduler/database/models.go b/internal/scheduler/database/models.go
index 449b747d..d438cbd1 100644
--- a/internal/scheduler/database/models.go
+++ b/internal/scheduler/database/models.go
@@ -5,6 +5,7 @@
 package database
 
 import (
+	"database/sql"
 	"time"
 
 	"github.com/google/uuid"
@@ -35,15 +36,15 @@ type Job struct {
 }
 
 type JobRunError struct {
-	RunID        uuid.UUID `db:"run_id"`
-	Error        []byte    `db:"error"`
-	Serial       int64     `db:"serial"`
-	LastModified time.Time `db:"last_modified"`
+	RunID uuid.UUID `db:"run_id"`
+	JobID string    `db:"job_id"`
+	Error []byte    `db:"error"`
 }
 
 type Marker struct {
-	GroupID     uuid.UUID `db:"group_id"`
-	PartitionID int32     `db:"partition_id"`
+	GroupID     uuid.UUID    `db:"group_id"`
+	PartitionID int32        `db:"partition_id"`
+	Created     sql.NullTime `db:"created"`
 }
 
 type Queue struct {
diff --git a/internal/scheduler/database/query.sql.go b/internal/scheduler/database/query.sql.go
index 1beccd85..e6302787 100644
--- a/internal/scheduler/database/query.sql.go
+++ b/internal/scheduler/database/query.sql.go
@@ -48,30 +48,21 @@ func (q *Queries) FindActiveRuns(ctx context.Context, runIds []uuid.UUID) ([]uui
 	return items, nil
 }
 
-const markJobRunsCancelledByJobId = `-- name: MarkJobRunsCancelledByJobId :exec
-UPDATE runs SET cancelled = true WHERE job_id = ANY($1::text[])
-`
-
-func (q *Queries) MarkJobRunsCancelledByJobId(ctx context.Context, jobIds []string) error {
-	_, err := q.db.Exec(ctx, markJobRunsCancelledByJobId, jobIds)
-	return err
-}
-
-const markJobRunsCancelledBySets = `-- name: MarkJobRunsCancelledBySets :exec
-UPDATE runs SET cancelled = true WHERE job_set = ANY($1::text[])
+const markJobRunsFailedById = `-- name: MarkJobRunsFailedById :exec
+UPDATE runs SET failed = true WHERE run_id = ANY($1::UUID[])
 `
 
-func (q *Queries) MarkJobRunsCancelledBySets(ctx context.Context, jobSets []string) error {
-	_, err := q.db.Exec(ctx, markJobRunsCancelledBySets, jobSets)
+func (q *Queries) MarkJobRunsFailedById(ctx context.Context, runIds []uuid.UUID) error {
+	_, err := q.db.Exec(ctx, markJobRunsFailedById, runIds)
 	return err
 }
 
-const markJobRunsFailedById = `-- name: MarkJobRunsFailedById :exec
-UPDATE runs SET failed = true WHERE run_id = ANY($1::UUID[])
+const markJobRunsReturnedById = `-- name: MarkJobRunsReturnedById :exec
+UPDATE runs SET returned = true WHERE run_id = ANY($1::UUID[])
 `
 
-func (q *Queries) MarkJobRunsFailedById(ctx context.Context, runIds []uuid.UUID) error {
-	_, err := q.db.Exec(ctx, markJobRunsFailedById, runIds)
+func (q *Queries) MarkJobRunsReturnedById(ctx context.Context, runIds []uuid.UUID) error {
+	_, err := q.db.Exec(ctx, markJobRunsReturnedById, runIds)
 	return err
 }
 
@@ -93,21 +84,30 @@ func (q *Queries) MarkJobRunsSucceededById(ctx context.Context, runIds []uuid.UU
 	return err
 }
 
-const markJobsCancelledById = `-- name: MarkJobsCancelledById :exec
-UPDATE jobs SET cancelled = true WHERE job_id = ANY($1::text[])
+const markJobsCancelRequestedById = `-- name: MarkJobsCancelRequestedById :exec
+UPDATE jobs SET cancel_requested = true WHERE job_id = ANY($1::text[])
 `
 
-func (q *Queries) MarkJobsCancelledById(ctx context.Context, jobIds []string) error {
-	_, err := q.db.Exec(ctx, markJobsCancelledById, jobIds)
+func (q *Queries) MarkJobsCancelRequestedById(ctx context.Context, jobIds []string) error {
+	_, err := q.db.Exec(ctx, markJobsCancelRequestedById, jobIds)
 	return err
 }
 
-const markJobsCancelledBySets = `-- name: MarkJobsCancelledBySets :exec
-UPDATE jobs SET cancelled = true WHERE job_set = ANY($1::text[])
+const markJobsCancelRequestedBySets = `-- name: MarkJobsCancelRequestedBySets :exec
+UPDATE jobs SET cancel_requested = true WHERE job_set = ANY($1::text[])
 `
 
-func (q *Queries) MarkJobsCancelledBySets(ctx context.Context, jobSets []string) error {
-	_, err := q.db.Exec(ctx, markJobsCancelledBySets, jobSets)
+func (q *Queries) MarkJobsCancelRequestedBySets(ctx context.Context, jobSets []string) error {
+	_, err := q.db.Exec(ctx, markJobsCancelRequestedBySets, jobSets)
+	return err
+}
+
+const markJobsCancelledById = `-- name: MarkJobsCancelledById :exec
+UPDATE jobs SET cancelled = true WHERE job_id = ANY($1::text[])
+`
+
+func (q *Queries) MarkJobsCancelledById(ctx context.Context, jobIds []string) error {
+	_, err := q.db.Exec(ctx, markJobsCancelledById, jobIds)
 	return err
 }
 
@@ -361,7 +361,7 @@ func (q *Queries) SelectNewRunsForJobs(ctx context.Context, arg SelectNewRunsFor
 }
 
 const selectRunErrorsById = `-- name: SelectRunErrorsById :many
-SELECT run_id, error, serial, last_modified FROM job_run_errors WHERE run_id = ANY($1::UUID[])
+SELECT run_id, job_id, error FROM job_run_errors WHERE run_id = ANY($1::UUID[])
 `
 
 // Run errors
@@ -374,12 +374,7 @@ func (q *Queries) SelectRunErrorsById(ctx context.Context, runIds []uuid.UUID) (
 	var items []JobRunError
 	for rows.Next() {
 		var i JobRunError
-		if err := rows.Scan(
-			&i.RunID,
-			&i.Error,
-			&i.Serial,
-			&i.LastModified,
-		); err != nil {
+		if err := rows.Scan(&i.RunID, &i.JobID, &i.Error); err != nil {
 			return nil, err
 		}
 		items = append(items, i)
diff --git a/internal/scheduler/database/query/query.sql b/internal/scheduler/database/query/query.sql
index f21875cd..48e4e935 100644
--- a/internal/scheduler/database/query/query.sql
+++ b/internal/scheduler/database/query/query.sql
@@ -7,12 +7,15 @@ SELECT job_id, job_set, queue, priority, submitted, cancel_requested, cancelled,
 -- name: UpdateJobPriorityByJobSet :exec
 UPDATE jobs SET priority = $1 WHERE job_set = $2;
 
--- name: MarkJobsCancelledBySets :exec
-UPDATE jobs SET cancelled = true WHERE job_set = ANY(sqlc.arg(job_sets)::text[]);
+-- name: MarkJobsCancelRequestedBySets :exec
+UPDATE jobs SET cancel_requested = true WHERE job_set = ANY(sqlc.arg(job_sets)::text[]);
 
 -- name: MarkJobsSucceededById :exec
 UPDATE jobs SET succeeded = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
 
+-- name: MarkJobsCancelRequestedById :exec
+UPDATE jobs SET cancel_requested = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
+
 -- name: MarkJobsCancelledById :exec
 UPDATE jobs SET cancelled = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
 
@@ -28,18 +31,15 @@ SELECT * FROM runs WHERE serial > $1 ORDER BY serial LIMIT $2;
 -- name: SelectNewRunsForJobs :many
 SELECT * FROM runs WHERE serial > $1 AND job_id = ANY(sqlc.arg(job_ids)::text[]) ORDER BY serial;
 
--- name: MarkJobRunsCancelledBySets :exec
-UPDATE runs SET cancelled = true WHERE job_set = ANY(sqlc.arg(job_sets)::text[]);
-
--- name: MarkJobRunsCancelledByJobId :exec
-UPDATE runs SET cancelled = true WHERE job_id = ANY(sqlc.arg(job_ids)::text[]);
-
 -- name: MarkJobRunsSucceededById :exec
 UPDATE runs SET succeeded = true WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
 
 -- name: MarkJobRunsFailedById :exec
 UPDATE runs SET failed = true WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
 
+-- name: MarkJobRunsReturnedById :exec
+UPDATE runs SET returned = true WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
+
 -- name: MarkJobRunsRunningById :exec
 UPDATE runs SET running = true WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
 
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 8e1e85f3..439ed48c 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -404,7 +404,7 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *SchedulerJob, jobRunError
 		return nil, err
 	}
 
-	// Has the job been requested cancelled.  If so, cancel the job
+	// Has the job been requested cancelled. If so, cancel the job
 	if job.CancelRequested {
 		job.Cancelled = true
 		cancel := &armadaevents.EventSequence_Event{
diff --git a/internal/scheduleringester/config.go b/internal/scheduleringester/config.go
index e158ed1a..a8e05ce9 100644
--- a/internal/scheduleringester/config.go
+++ b/internal/scheduleringester/config.go
@@ -13,6 +13,8 @@ type Configuration struct {
 	Metrics configuration.MetricsConfig
 	// General Pulsar configuration
 	Pulsar configuration.PulsarConfig
+	// Map of allowed priority classes by name
+	PriorityClasses map[string]configuration.PriorityClass
 	// Pulsar subscription name
 	SubscriptionName string
 	// Number of messages that will be batched together before being inserted into the database
diff --git a/internal/scheduleringester/dbops.go b/internal/scheduleringester/dbops.go
index 28282fb0..b6f68e8f 100644
--- a/internal/scheduleringester/dbops.go
+++ b/internal/scheduleringester/dbops.go
@@ -19,6 +19,10 @@ func (d *DbOperationsWithMessageIds) GetMessageIDs() []pulsar.MessageID {
 	return d.MessageIds
 }
 
+type JobRunFailed struct {
+	LeaseReturned bool
+}
+
 // DbOperation captures a generic batch database operation.
 //
 // There are 5 types of operations:
@@ -85,16 +89,18 @@ func discardNilOps(ops []DbOperation) []DbOperation {
 type InsertJobs map[string]*schedulerdb.Job
 
 type (
-	InsertRuns             map[uuid.UUID]*schedulerdb.Run
-	UpdateJobSetPriorities map[string]int64
-	MarkJobSetsCancelled   map[string]bool
-	MarkJobsCancelled      map[string]bool
-	MarkJobsSucceeded      map[string]bool
-	MarkJobsFailed         map[string]bool
-	UpdateJobPriorities    map[string]int64
-	MarkRunsSucceeded      map[uuid.UUID]bool
-	MarkRunsFailed         map[uuid.UUID]bool
-	MarkRunsRunning        map[uuid.UUID]bool
+	InsertRuns                 map[uuid.UUID]*schedulerdb.Run
+	UpdateJobSetPriorities     map[string]int64
+	MarkJobSetsCancelRequested map[string]bool
+	MarkJobsCancelRequested    map[string]bool
+	MarkJobsCancelled          map[string]bool
+	MarkJobsSucceeded          map[string]bool
+	MarkJobsFailed             map[string]bool
+	UpdateJobPriorities        map[string]int64
+	MarkRunsSucceeded          map[uuid.UUID]bool
+	MarkRunsFailed             map[uuid.UUID]*JobRunFailed
+	MarkRunsRunning            map[uuid.UUID]bool
+	InsertJobRunErrors         map[uuid.UUID]*schedulerdb.JobRunError
 )
 
 type JobSetOperation interface {
@@ -106,7 +112,7 @@ func (a UpdateJobSetPriorities) AffectsJobSet(jobSet string) bool {
 	return ok
 }
 
-func (a MarkJobSetsCancelled) AffectsJobSet(jobSet string) bool {
+func (a MarkJobSetsCancelRequested) AffectsJobSet(jobSet string) bool {
 	_, ok := a[jobSet]
 	return ok
 }
@@ -123,7 +129,11 @@ func (a UpdateJobSetPriorities) Merge(b DbOperation) bool {
 	return mergeInMap(a, b)
 }
 
-func (a MarkJobSetsCancelled) Merge(b DbOperation) bool {
+func (a MarkJobSetsCancelRequested) Merge(b DbOperation) bool {
+	return mergeInMap(a, b)
+}
+
+func (a MarkJobsCancelRequested) Merge(b DbOperation) bool {
 	return mergeInMap(a, b)
 }
 
@@ -155,8 +165,12 @@ func (a MarkRunsRunning) Merge(b DbOperation) bool {
 	return mergeInMap(a, b)
 }
 
+func (a InsertJobRunErrors) Merge(b DbOperation) bool {
+	return mergeInMap(a, b)
+}
+
 // mergeInMap merges an op b into a, provided that b is of the same type as a.
-// For example, if a is of type MarkJobsCancelled, b is only merged if also of type MarkJobsCancelled.
+// For example, if a is of type MarkJobSetsCancelRequested, b is only merged if also of type MarkJobsCancelRequested.
 // Returns true if the ops were merged and false otherwise.
 func mergeInMap[M ~map[K]V, K comparable, V any](a M, b DbOperation) bool {
 	// Using a type switch here, since using a type assertion
@@ -208,11 +222,11 @@ func (a UpdateJobSetPriorities) CanBeAppliedBefore(b DbOperation) bool {
 	return !isUpdateJobPriorities && !definesJobInSet(a, b)
 }
 
-func (a MarkJobSetsCancelled) CanBeAppliedBefore(b DbOperation) bool {
+func (a MarkJobSetsCancelRequested) CanBeAppliedBefore(b DbOperation) bool {
 	return !definesJobInSet(a, b) && !definesRunInSet(a, b)
 }
 
-func (a MarkJobsCancelled) CanBeAppliedBefore(b DbOperation) bool {
+func (a MarkJobsCancelRequested) CanBeAppliedBefore(b DbOperation) bool {
 	return !definesJob(a, b) && !definesRunForJob(a, b)
 }
 
@@ -224,6 +238,10 @@ func (a MarkJobsFailed) CanBeAppliedBefore(b DbOperation) bool {
 	return !definesJob(a, b)
 }
 
+func (a MarkJobsCancelled) CanBeAppliedBefore(b DbOperation) bool {
+	return !definesJob(a, b)
+}
+
 func (a UpdateJobPriorities) CanBeAppliedBefore(b DbOperation) bool {
 	_, isUpdateJobSetPriorities := b.(UpdateJobSetPriorities)
 	return !isUpdateJobSetPriorities && !definesJob(a, b)
@@ -241,6 +259,12 @@ func (a MarkRunsRunning) CanBeAppliedBefore(b DbOperation) bool {
 	return !definesRun(a, b)
 }
 
+func (a InsertJobRunErrors) CanBeAppliedBefore(_ DbOperation) bool {
+	// Inserting errors before a run has been marked as failed is ok.
+	// We only require that errors are written to the schedulerdb before the run is marked as failed.
+	return true
+}
+
 // 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.
diff --git a/internal/scheduleringester/ingester.go b/internal/scheduleringester/ingester.go
index 9c5ba41c..a4eb5b26 100644
--- a/internal/scheduleringester/ingester.go
+++ b/internal/scheduleringester/ingester.go
@@ -19,7 +19,7 @@ import (
 
 // Run will create a pipeline that will take Armada event messages from Pulsar and update the
 // Scheduler database accordingly.  This pipeline will run until a SIGTERM is received
-func Run(config *Configuration) {
+func Run(config Configuration) {
 	metrics := metrics.NewMetrics(metrics.ArmadaEventIngesterMetricsPrefix + "armada_scheduler_ingester_")
 
 	log.Infof("Opening connection pool to postgres")
@@ -42,7 +42,7 @@ func Run(config *Configuration) {
 	if err != nil {
 		panic(errors.WithMessage(err, "Error creating  compressor"))
 	}
-	converter := NewInstructionConverter(metrics, submitJobFilter, compressor)
+	converter := NewInstructionConverter(metrics, submitJobFilter, config.PriorityClasses, compressor)
 
 	ingester := ingest.NewIngestionPipeline(
 		config.Pulsar,
diff --git a/internal/scheduleringester/instructions.go b/internal/scheduleringester/instructions.go
index 67edecc3..11abfa77 100644
--- a/internal/scheduleringester/instructions.go
+++ b/internal/scheduleringester/instructions.go
@@ -7,9 +7,11 @@ import (
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/compress"
 	"github.com/armadaproject/armada/internal/common/ingest"
 	"github.com/armadaproject/armada/internal/common/ingest/metrics"
+	protoutil "github.com/armadaproject/armada/internal/common/proto"
 	"github.com/armadaproject/armada/internal/scheduler/adapters"
 	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
@@ -24,18 +26,23 @@ type eventSequenceCommon struct {
 }
 
 type InstructionConverter struct {
-	metrics     *metrics.Metrics
-	eventFilter func(event *armadaevents.EventSequence_Event) bool
-	compressor  compress.Compressor
+	metrics         *metrics.Metrics
+	eventFilter     func(event *armadaevents.EventSequence_Event) bool
+	priorityClasses map[string]configuration.PriorityClass
+	compressor      compress.Compressor
 }
 
-func NewInstructionConverter(metrics *metrics.Metrics,
-	filter func(event *armadaevents.EventSequence_Event) bool, compressor compress.Compressor,
+func NewInstructionConverter(
+	metrics *metrics.Metrics,
+	filter func(event *armadaevents.EventSequence_Event) bool,
+	priorityClasses map[string]configuration.PriorityClass,
+	compressor compress.Compressor,
 ) ingest.InstructionConverter[*DbOperationsWithMessageIds] {
 	return &InstructionConverter{
-		metrics:     metrics,
-		eventFilter: filter,
-		compressor:  compressor,
+		metrics:         metrics,
+		eventFilter:     filter,
+		priorityClasses: priorityClasses,
+		compressor:      compressor,
 	}
 }
 
@@ -88,8 +95,9 @@ func (c *InstructionConverter) convertSequence(es *armadaevents.EventSequence) [
 				operationsFromEvent, err = c.handleCancelJob(event.GetCancelJob())
 			case *armadaevents.EventSequence_Event_CancelJobSet:
 				operationsFromEvent, err = c.handleCancelJobSet(meta.jobset)
-			case *armadaevents.EventSequence_Event_CancelledJob,
-				*armadaevents.EventSequence_Event_ReprioritisedJob,
+			case *armadaevents.EventSequence_Event_CancelledJob:
+				operationsFromEvent, err = c.handleCancelledJob(event.GetCancelledJob())
+			case *armadaevents.EventSequence_Event_ReprioritisedJob,
 				*armadaevents.EventSequence_Event_JobDuplicateDetected,
 				*armadaevents.EventSequence_Event_ResourceUtilisation,
 				*armadaevents.EventSequence_Event_StandaloneIngressInfo,
@@ -125,7 +133,7 @@ func (c *InstructionConverter) handleSubmitJob(job *armadaevents.SubmitJob, meta
 
 	// Produce a minimal representation of the job for the scheduler.
 	// To avoid the scheduler needing to load the entire job spec.
-	schedulingInfo, err := schedulingInfoFromSubmitJob(job)
+	schedulingInfo, err := c.schedulingInfoFromSubmitJob(job)
 	if err != nil {
 		return nil, err
 	}
@@ -179,13 +187,27 @@ func (c *InstructionConverter) handleJobRunSucceeded(jobRunSucceeded *armadaeven
 }
 
 func (c *InstructionConverter) handleJobRunErrors(jobRunErrors *armadaevents.JobRunErrors) ([]DbOperation, error) {
-	runId := jobRunErrors.GetRunId()
+	runId := armadaevents.UuidFromProtoUuid(jobRunErrors.GetRunId())
+	jobId, err := armadaevents.UlidStringFromProtoUuid(jobRunErrors.JobId)
+	if err != nil {
+		return nil, err
+	}
+	insertJobRunErrors := make(InsertJobRunErrors)
+	markRunsFailed := make(MarkRunsFailed)
 	for _, runError := range jobRunErrors.GetErrors() {
-		// For terminal errors, we also need to mark the run as failed.
+		// There should only be one terminal error
 		if runError.GetTerminal() {
-			markRunsFailed := make(MarkRunsFailed)
-			markRunsFailed[armadaevents.UuidFromProtoUuid(runId)] = true
-			return []DbOperation{markRunsFailed}, nil
+			bytes, err := protoutil.MarshallAndCompress(runError, c.compressor)
+			if err != nil {
+				return nil, errors.Wrap(err, "failed to marshal RunError")
+			}
+			insertJobRunErrors[runId] = &schedulerdb.JobRunError{
+				RunID: runId,
+				JobID: jobId,
+				Error: bytes,
+			}
+			markRunsFailed[runId] = &JobRunFailed{LeaseReturned: runError.GetPodLeaseReturned() != nil}
+			return []DbOperation{insertJobRunErrors, markRunsFailed}, nil
 		}
 	}
 	return nil, nil
@@ -238,20 +260,30 @@ func (c *InstructionConverter) handleCancelJob(cancelJob *armadaevents.CancelJob
 	if err != nil {
 		return nil, err
 	}
-	return []DbOperation{MarkJobsCancelled{
+	return []DbOperation{MarkJobsCancelRequested{
 		jobId: true,
 	}}, nil
 }
 
 func (c *InstructionConverter) handleCancelJobSet(jobset string) ([]DbOperation, error) {
-	return []DbOperation{MarkJobSetsCancelled{
+	return []DbOperation{MarkJobSetsCancelRequested{
 		jobset: true,
 	}}, nil
 }
 
+func (c *InstructionConverter) handleCancelledJob(cancelledJob *armadaevents.CancelledJob) ([]DbOperation, error) {
+	jobId, err := armadaevents.UlidStringFromProtoUuid(cancelledJob.GetJobId())
+	if err != nil {
+		return nil, err
+	}
+	return []DbOperation{MarkJobsCancelled{
+		jobId: true,
+	}}, nil
+}
+
 // schedulingInfoFromSubmitJob returns a minimal representation of a job
 // containing only the info needed by the scheduler.
-func schedulingInfoFromSubmitJob(submitJob *armadaevents.SubmitJob) (*schedulerobjects.JobSchedulingInfo, error) {
+func (c *InstructionConverter) schedulingInfoFromSubmitJob(submitJob *armadaevents.SubmitJob) (*schedulerobjects.JobSchedulingInfo, error) {
 	// Component common to all jobs.
 	schedulingInfo := &schedulerobjects.JobSchedulingInfo{
 		Lifetime:        submitJob.Lifetime,
@@ -265,8 +297,7 @@ func schedulingInfoFromSubmitJob(submitJob *armadaevents.SubmitJob) (*schedulero
 	case *armadaevents.KubernetesMainObject_PodSpec:
 		podSpec := object.PodSpec.PodSpec
 		requirements := &schedulerobjects.ObjectRequirements_PodRequirements{
-			// TODO: We should not pass in nil here. Priority will not be set correctly.
-			PodRequirements: adapters.PodRequirementsFromPodSpec(podSpec, nil),
+			PodRequirements: adapters.PodRequirementsFromPodSpec(podSpec, c.priorityClasses),
 		}
 		schedulingInfo.ObjectRequirements = append(
 			schedulingInfo.ObjectRequirements,
diff --git a/internal/scheduleringester/schedulerdb.go b/internal/scheduleringester/schedulerdb.go
index 59df9deb..a3f8c72a 100644
--- a/internal/scheduleringester/schedulerdb.go
+++ b/internal/scheduleringester/schedulerdb.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/hashicorp/go-multierror"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/pkg/errors"
@@ -80,13 +81,15 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, op DbOperation) error {
 				return errors.WithStack(err)
 			}
 		}
-	case MarkJobSetsCancelled:
+	case MarkJobSetsCancelRequested:
 		jobSets := maps.Keys(o)
-		err := queries.MarkJobsCancelledBySets(ctx, jobSets)
+		err := queries.MarkJobsCancelRequestedBySets(ctx, jobSets)
 		if err != nil {
 			return errors.WithStack(err)
 		}
-		err = queries.MarkJobRunsCancelledBySets(ctx, jobSets)
+	case MarkJobsCancelRequested:
+		jobIds := maps.Keys(o)
+		err := queries.MarkJobsCancelRequestedById(ctx, jobIds)
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -96,10 +99,6 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, op DbOperation) error {
 		if err != nil {
 			return errors.WithStack(err)
 		}
-		err = queries.MarkJobRunsCancelledByJobId(ctx, jobIds)
-		if err != nil {
-			return errors.WithStack(err)
-		}
 	case MarkJobsSucceeded:
 		jobIds := maps.Keys(o)
 		err := queries.MarkJobsSucceededById(ctx, jobIds)
@@ -132,16 +131,37 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, op DbOperation) error {
 		}
 	case MarkRunsFailed:
 		runIds := maps.Keys(o)
+		returned := make([]uuid.UUID, 0, len(runIds))
+		for k, v := range o {
+			if v.LeaseReturned {
+				returned = append(returned, k)
+			}
+		}
 		err := queries.MarkJobRunsFailedById(ctx, runIds)
 		if err != nil {
 			return errors.WithStack(err)
 		}
+		err = queries.MarkJobRunsReturnedById(ctx, returned)
+		if err != nil {
+			return errors.WithStack(err)
+		}
 	case MarkRunsRunning:
 		runIds := maps.Keys(o)
 		err := queries.MarkJobRunsRunningById(ctx, runIds)
 		if err != nil {
 			return errors.WithStack(err)
 		}
+	case InsertJobRunErrors:
+		records := make([]any, len(o))
+		i := 0
+		for _, v := range o {
+			records[i] = *v
+			i++
+		}
+		err := database.Upsert(ctx, s.db, "job_run_errors", records)
+		if err != nil {
+			return err
+		}
 	default:
 		return errors.Errorf("received unexpected op %+v", op)
 	}


Test patch:
diff --git a/internal/common/compress/list_test.go b/internal/common/compress/list_test.go
index 3a022179..c8144890 100644
--- a/internal/common/compress/list_test.go
+++ b/internal/common/compress/list_test.go
@@ -19,3 +19,16 @@ func TestCompressAndDecompressGiveOriginalValue(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, input, decompressedOutput)
 }
+
+func TestMustCompressAndDecompressGiveOriginalValue(t *testing.T) {
+	compressor, err := NewZlibCompressor(0)
+	assert.NoError(t, err)
+	decompressor := NewZlibDecompressor()
+
+	input := []string{"test", "array", "values"}
+
+	compressedInput := MustCompressStringArray(input, compressor)
+	decompressedOutput := MustDecompressStringArray(compressedInput, decompressor)
+	assert.NoError(t, err)
+	assert.Equal(t, input, decompressedOutput)
+}
diff --git a/internal/common/ingest/testfixtures/event.go b/internal/common/ingest/testfixtures/event.go
index 359712eb..6d50bcec 100644
--- a/internal/common/ingest/testfixtures/event.go
+++ b/internal/common/ingest/testfixtures/event.go
@@ -8,6 +8,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
+	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/eventutil"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 )
@@ -19,13 +20,16 @@ const (
 )
 
 var (
-	JobIdProto, _ = armadaevents.ProtoUuidFromUlidString(JobIdString)
-	RunIdProto    = armadaevents.ProtoUuidFromUuid(uuid.MustParse(RunIdString))
-	JobIdUuid     = armadaevents.UuidFromProtoUuid(JobIdProto)
-	RunIdUuid     = armadaevents.UuidFromProtoUuid(RunIdProto)
-	Groups        = []string{"group1", "group2"}
-	NodeSelector  = map[string]string{"foo": "bar"}
-	Tolerations   = []v1.Toleration{{
+	JobIdProto, _      = armadaevents.ProtoUuidFromUlidString(JobIdString)
+	RunIdProto         = armadaevents.ProtoUuidFromUuid(uuid.MustParse(RunIdString))
+	JobIdUuid          = armadaevents.UuidFromProtoUuid(JobIdProto)
+	RunIdUuid          = armadaevents.UuidFromProtoUuid(RunIdProto)
+	PriorityClassName  = "test-priority"
+	PriorityClassValue = int32(100)
+	PriorityClasses    = map[string]configuration.PriorityClass{PriorityClassName: {Priority: PriorityClassValue}}
+	Groups             = []string{"group1", "group2"}
+	NodeSelector       = map[string]string{"foo": "bar"}
+	Tolerations        = []v1.Toleration{{
 		Key:      "fish",
 		Operator: "exists",
 	}}
@@ -67,8 +71,9 @@ var Submit = &armadaevents.EventSequence_Event{
 				Object: &armadaevents.KubernetesMainObject_PodSpec{
 					PodSpec: &armadaevents.PodSpecWithAvoidList{
 						PodSpec: &v1.PodSpec{
-							NodeSelector: NodeSelector,
-							Tolerations:  Tolerations,
+							NodeSelector:      NodeSelector,
+							Tolerations:       Tolerations,
+							PriorityClassName: PriorityClassName,
 							Containers: []v1.Container{
 								{
 									Name:    "container1",
diff --git a/internal/common/proto/protoutil_test.go b/internal/common/proto/protoutil_test.go
new file mode 100644
index 00000000..3d717e9e
--- /dev/null
+++ b/internal/common/proto/protoutil_test.go
@@ -0,0 +1,65 @@
+package protoutil
+
+import (
+	"testing"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/armadaproject/armada/internal/common/compress"
+	"github.com/armadaproject/armada/pkg/armadaevents"
+)
+
+var (
+	msg              = &armadaevents.CancelJob{JobId: armadaevents.ProtoUuidFromUuid(uuid.New())}
+	compressor       = compress.NewThreadSafeZlibCompressor(1024)
+	decompressor     = compress.NewThreadSafeZlibDecompressor()
+	marshalledMsg, _ = proto.Marshal(msg)
+	compressedMsg, _ = compressor.Compress(marshalledMsg)
+	invalidMsg       = []byte{0x3}
+)
+
+func TestUnmarshall_Valid(t *testing.T) {
+	unmarshalled, err := Unmarshall(marshalledMsg, &armadaevents.CancelJob{})
+	require.NoError(t, err)
+	assert.Equal(t, msg, unmarshalled)
+}
+
+func TestUnmarshall_Invalid(t *testing.T) {
+	_, err := Unmarshall(invalidMsg, &armadaevents.CancelJob{})
+	require.Error(t, err)
+}
+
+func TestMustUnmarshall(t *testing.T) {
+	unmarshalled := MustUnmarshall(marshalledMsg, &armadaevents.CancelJob{})
+	assert.Equal(t, msg, unmarshalled)
+}
+
+func TestDecompressAndUnmarshall_Valid(t *testing.T) {
+	unmarshalled, err := DecompressAndUnmarshall(compressedMsg, &armadaevents.CancelJob{}, decompressor)
+	require.NoError(t, err)
+	assert.Equal(t, msg, unmarshalled)
+}
+
+func TestDecompressAndUnmarshall_Invalid(t *testing.T) {
+	_, err := DecompressAndUnmarshall(invalidMsg, &armadaevents.CancelJob{}, decompressor)
+	require.Error(t, err)
+}
+
+func TestMustDecompressAndUnmarshall(t *testing.T) {
+	unmarshalled := MustDecompressAndUnmarshall(compressedMsg, &armadaevents.CancelJob{}, decompressor)
+	assert.Equal(t, msg, unmarshalled)
+}
+
+func TestMarshallAndCompress(t *testing.T) {
+	bytes, err := MarshallAndCompress(msg, compressor)
+	require.NoError(t, err)
+	assert.Equal(t, compressedMsg, bytes)
+}
+
+func TestMustMarshallAndCompress(t *testing.T) {
+	bytes := MustMarshallAndCompress(msg, compressor)
+	assert.Equal(t, compressedMsg, bytes)
+}
diff --git a/internal/scheduleringester/dbops_test.go b/internal/scheduleringester/dbops_test.go
index 08e00b25..c97aa5e7 100644
--- a/internal/scheduleringester/dbops_test.go
+++ b/internal/scheduleringester/dbops_test.go
@@ -16,17 +16,17 @@ func TestMerge(t *testing.T) {
 	jobId1 := util.NewULID()
 	jobId2 := util.NewULID()
 	jobId3 := util.NewULID()
-	markJobsCancelled1 := MarkJobsCancelled{jobId1: false, jobId2: false}
-	markJobsCancelled2 := MarkJobsCancelled{jobId2: true, jobId3: true}
+	markJobsCancelled1 := MarkJobsCancelRequested{jobId1: false, jobId2: false}
+	markJobsCancelled2 := MarkJobsCancelRequested{jobId2: true, jobId3: true}
 	ok := markJobsCancelled1.Merge(markJobsCancelled2)
 	assert.True(t, ok)
-	assert.Equal(t, MarkJobsCancelled{jobId1: false, jobId2: true, jobId3: true}, markJobsCancelled1)
+	assert.Equal(t, MarkJobsCancelRequested{jobId1: false, jobId2: true, jobId3: true}, markJobsCancelled1)
 
 	jobId4 := util.NewULID()
 	markJobsSucceeded1 := MarkJobsSucceeded{jobId1: true, jobId4: true}
 	ok = markJobsCancelled1.Merge(markJobsSucceeded1)
 	assert.False(t, ok)
-	assert.Equal(t, MarkJobsCancelled{jobId1: false, jobId2: true, jobId3: true}, markJobsCancelled1)
+	assert.Equal(t, MarkJobsCancelRequested{jobId1: false, jobId2: true, jobId3: true}, markJobsCancelled1)
 }
 
 // Test that db op optimisation
@@ -76,21 +76,21 @@ func TestDbOperationOptimisation(t *testing.T) {
 			UpdateJobPriorities{jobIds[1]: 4},                                         // 4
 			UpdateJobPriorities{jobIds[2]: 5},                                         // 4
 		}},
-		"MarkJobSetsCancelled": {N: 3, Ops: []DbOperation{
+		"MarkJobSetsCancelRequested": {N: 3, Ops: []DbOperation{
 			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"}}, // 1
-			MarkJobSetsCancelled{"set1": true},                                        // 2
+			MarkJobSetsCancelRequested{"set1": true},                                  // 2
 			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1], JobSet: "set1"}}, // 3
-			MarkJobSetsCancelled{"set2": true},                                        // 3
+			MarkJobSetsCancelRequested{"set2": true},                                  // 3
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 3
 		}},
-		"MarkJobSetsCancelled, MarkJobsCancelled": {N: 4, Ops: []DbOperation{
+		"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
-			MarkJobsCancelled{jobIds[0]: true},                                        // 2
-			MarkJobSetsCancelled{"set1": true},                                        // 3
+			MarkJobsCancelRequested{jobIds[0]: true},                                  // 2
+			MarkJobSetsCancelRequested{"set1": true},                                  // 3
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"}}, // 4
-			MarkJobsCancelled{jobIds[1]: true},                                        // 4
-			MarkJobsCancelled{jobIds[2]: true},                                        // 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
@@ -106,6 +106,13 @@ func TestDbOperationOptimisation(t *testing.T) {
 			MarkJobsFailed{jobIds[1]: true},                           // 2
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}}, // 2
 		}},
+		"MarkJobsCancelled": {N: 2, Ops: []DbOperation{
+			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}}, // 1
+			MarkJobsCancelled{jobIds[0]: true},                        // 2
+			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}}, // 2
+			MarkJobsCancelled{jobIds[1]: true},                        // 2
+			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
@@ -118,10 +125,10 @@ func TestDbOperationOptimisation(t *testing.T) {
 		"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]: true},                                             // 3
+			MarkRunsFailed{runIds[0]: &JobRunFailed{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]: true},                                             // 3
+			MarkRunsFailed{runIds[1]: &JobRunFailed{true}},                              // 3
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                   // 3
 		}},
 		"MarkRunsRunning": {N: 3, Ops: []DbOperation{
@@ -169,7 +176,7 @@ func TestDbOperationOptimisation(t *testing.T) {
 	}
 }
 
-func TestInsertJobCancel(t *testing.T) {
+func TestInsertJobRequestCancel(t *testing.T) {
 	// Submit jobs to two different job sets.
 	var ops []DbOperation
 	expectedCancelledIds := make(map[string]bool)
@@ -184,7 +191,7 @@ func TestInsertJobCancel(t *testing.T) {
 	}
 
 	// Cancel one job set.
-	ops = append(ops, MarkJobSetsCancelled{"set1": true})
+	ops = append(ops, MarkJobSetsCancelRequested{"set1": true})
 
 	// Submit some more jobs to both job sets.
 	for i := 0; i < 2; i++ {
@@ -206,7 +213,7 @@ func TestInsertJobCancel(t *testing.T) {
 	// Check that the mockDb did the right thing.
 	for jobId, job := range expectedDb.Jobs {
 		_, ok := expectedCancelledIds[jobId]
-		if job.Cancelled {
+		if job.CancelRequested {
 			assert.True(t, ok)
 		} else {
 			assert.False(t, ok)
@@ -284,18 +291,18 @@ func (db *mockDb) apply(op DbOperation) error {
 				}
 			}
 		}
-	case MarkJobSetsCancelled:
+	case MarkJobSetsCancelRequested:
 		for jobSet := range o {
 			for _, job := range db.Jobs {
 				if job.JobSet == jobSet {
-					job.Cancelled = true
+					job.CancelRequested = true
 				}
 			}
 		}
-	case MarkJobsCancelled:
+	case MarkJobsCancelRequested:
 		for jobId := range o {
 			if job, ok := db.Jobs[jobId]; ok {
-				job.Cancelled = true
+				job.CancelRequested = true
 			} else {
 				return errors.Errorf("job %s not in db", jobId)
 			}
diff --git a/internal/scheduleringester/instructions_test.go b/internal/scheduleringester/instructions_test.go
index 0019bc13..07b8bc49 100644
--- a/internal/scheduleringester/instructions_test.go
+++ b/internal/scheduleringester/instructions_test.go
@@ -4,23 +4,24 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/armadaproject/armada/internal/common/compress"
-
-	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
+	"github.com/armadaproject/armada/internal/common/compress"
 	"github.com/armadaproject/armada/internal/common/ingest/metrics"
 	f "github.com/armadaproject/armada/internal/common/ingest/testfixtures"
+	protoutil "github.com/armadaproject/armada/internal/common/proto"
 	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 )
 
 var (
-	m                   = metrics.NewMetrics(metrics.ArmadaEventIngesterMetricsPrefix + "test_")
-	compressedGroups, _ = compress.CompressStringArray(f.Groups, &compress.NoOpCompressor{})
+	compressor   = compress.NewThreadSafeZlibCompressor(1024)
+	decompressor = compress.NewThreadSafeZlibDecompressor()
+	m            = metrics.NewMetrics(metrics.ArmadaEventIngesterMetricsPrefix + "test_")
 )
 
 func TestConvertSequence(t *testing.T) {
@@ -35,11 +36,11 @@ func TestConvertSequence(t *testing.T) {
 				JobID:         f.JobIdString,
 				JobSet:        f.JobSetName,
 				UserID:        f.UserId,
-				Groups:        compressedGroups,
+				Groups:        compress.MustCompressStringArray(f.Groups, compressor),
 				Queue:         f.Queue,
 				Priority:      int64(f.Priority),
-				SubmitMessage: mustMarshall(f.Submit.GetSubmitJob()),
-				SchedulingInfo: mustMarshall(&schedulerobjects.JobSchedulingInfo{
+				SubmitMessage: protoutil.MustMarshallAndCompress(f.Submit.GetSubmitJob(), compressor),
+				SchedulingInfo: protoutil.MustMarshall(&schedulerobjects.JobSchedulingInfo{
 					Lifetime:        0,
 					AtMostOnce:      true,
 					Preemptible:     true,
@@ -51,6 +52,7 @@ func TestConvertSequence(t *testing.T) {
 									NodeSelector:     f.NodeSelector,
 									Tolerations:      f.Tolerations,
 									PreemptionPolicy: "PreemptLowerPriority",
+									Priority:         f.PriorityClassValue,
 									ResourceRequirements: v1.ResourceRequirements{
 										Limits: map[v1.ResourceName]resource.Quantity{
 											"memory": resource.MustParse("64Mi"),
@@ -85,10 +87,26 @@ func TestConvertSequence(t *testing.T) {
 			events:   []*armadaevents.EventSequence_Event{f.JobRunSucceeded},
 			expected: []DbOperation{MarkRunsSucceeded{f.RunIdUuid: true}},
 		},
-		"job run errors terminal": {
+		"lease returned": {
 			events: []*armadaevents.EventSequence_Event{f.LeaseReturned},
 			expected: []DbOperation{
-				MarkRunsFailed{f.RunIdUuid: true},
+				InsertJobRunErrors{f.RunIdUuid: &schedulerdb.JobRunError{
+					RunID: f.RunIdUuid,
+					JobID: f.JobIdString,
+					Error: protoutil.MustMarshallAndCompress(f.LeaseReturned.GetJobRunErrors().Errors[0], compressor),
+				}},
+				MarkRunsFailed{f.RunIdUuid: &JobRunFailed{LeaseReturned: true}},
+			},
+		},
+		"job failed": {
+			events: []*armadaevents.EventSequence_Event{f.JobRunFailed},
+			expected: []DbOperation{
+				InsertJobRunErrors{f.RunIdUuid: &schedulerdb.JobRunError{
+					RunID: f.RunIdUuid,
+					JobID: f.JobIdString,
+					Error: protoutil.MustMarshallAndCompress(f.JobRunFailed.GetJobRunErrors().Errors[0], compressor),
+				}},
+				MarkRunsFailed{f.RunIdUuid: &JobRunFailed{LeaseReturned: false}},
 			},
 		},
 		"job errors terminal": {
@@ -115,22 +133,28 @@ func TestConvertSequence(t *testing.T) {
 				UpdateJobSetPriorities{f.JobSetName: f.NewPriority},
 			},
 		},
-		"cancel job": {
+		"JobCancelRequested": {
 			events: []*armadaevents.EventSequence_Event{f.JobCancelRequested},
 			expected: []DbOperation{
-				MarkJobsCancelled{f.JobIdString: true},
+				MarkJobsCancelRequested{f.JobIdString: true},
 			},
 		},
-		"cancel jobSet": {
+		"JobSetCancelRequested": {
 			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequested},
 			expected: []DbOperation{
-				MarkJobSetsCancelled{f.JobSetName: true},
+				MarkJobSetsCancelRequested{f.JobSetName: true},
+			},
+		},
+		"JobCancelled": {
+			events: []*armadaevents.EventSequence_Event{f.JobCancelled},
+			expected: []DbOperation{
+				MarkJobsCancelled{f.JobIdString: true},
 			},
 		},
 		"multiple events": {
 			events: []*armadaevents.EventSequence_Event{f.JobSetCancelRequested, f.Running, f.JobSucceeded},
 			expected: []DbOperation{
-				MarkJobSetsCancelled{f.JobSetName: true},
+				MarkJobSetsCancelRequested{f.JobSetName: true},
 				MarkRunsRunning{f.RunIdUuid: true},
 				MarkJobsSucceeded{f.JobIdString: true},
 			},
@@ -143,7 +167,7 @@ func TestConvertSequence(t *testing.T) {
 			expected: []DbOperation{MarkRunsRunning{f.RunIdUuid: true}},
 		},
 		"ignored events": {
-			events: []*armadaevents.EventSequence_Event{f.Running, f.JobCancelled, f.JobSucceeded},
+			events: []*armadaevents.EventSequence_Event{f.Running, f.JobPreempted, f.JobSucceeded},
 			expected: []DbOperation{
 				MarkRunsRunning{f.RunIdUuid: true},
 				MarkJobsSucceeded{f.JobIdString: true},
@@ -158,7 +182,7 @@ func TestConvertSequence(t *testing.T) {
 					return true
 				}
 			}
-			converter := InstructionConverter{m, tc.filter, &compress.NoOpCompressor{}}
+			converter := InstructionConverter{m, tc.filter, f.PriorityClasses, compressor}
 			es := f.NewEventSequence(tc.events...)
 			results := converter.convertSequence(es)
 			assertOperationsEqual(t, tc.expected, results)
@@ -168,7 +192,7 @@ func TestConvertSequence(t *testing.T) {
 
 func assertOperationsEqual(t *testing.T, expectedOps []DbOperation, actualOps []DbOperation) {
 	t.Helper()
-	assert.Equal(t, len(expectedOps), len(actualOps), "operations arrays are not the same length")
+	require.Equal(t, len(expectedOps), len(actualOps), "operations arrays are not the same length")
 	for i := 0; i < len(expectedOps); i++ {
 		expectedOp := expectedOps[i]
 		actualOp := actualOps[i]
@@ -188,6 +212,17 @@ func assertOperationsEqual(t *testing.T, expectedOps []DbOperation, actualOps []
 				expectedSubmit.SubmitMessage = nil
 				assert.Equal(t, expectedSubmit, actualSubmit)
 			}
+		case InsertJobRunErrors:
+			actualErrors := actualOp.(InsertJobRunErrors)
+			for k, expectedError := range expectedOp.(InsertJobRunErrors) {
+				actualError, ok := actualErrors[k]
+				assert.True(t, ok)
+				assertErrorMessagesEqual(t, expectedError.Error, actualError.Error)
+				// nil out the byte arrays
+				actualError.Error = nil
+				expectedError.Error = nil
+				assert.Equal(t, expectedError, actualError)
+			}
 		default:
 			assert.Equal(t, expectedOp, actualOp)
 		}
@@ -195,37 +230,25 @@ func assertOperationsEqual(t *testing.T, expectedOps []DbOperation, actualOps []
 }
 
 func assertSchedulingInfoEqual(t *testing.T, expectedBytes []byte, actualBytes []byte) {
-	actualSchedInfo, err := unmarshalSchedulingInfo(actualBytes)
+	actualSchedInfo, err := protoutil.Unmarshall(actualBytes, &schedulerobjects.JobSchedulingInfo{})
 	assert.NoError(t, err)
-	expectedSchedInfo, err := unmarshalSchedulingInfo(expectedBytes)
+	expectedSchedInfo, err := protoutil.Unmarshall(expectedBytes, &schedulerobjects.JobSchedulingInfo{})
 	assert.NoError(t, err)
 	assert.Equal(t, expectedSchedInfo, actualSchedInfo)
 }
 
 func assertSubmitMessagesEqual(t *testing.T, expectedBytes []byte, actualBytes []byte) {
-	actualSubmitMessage, err := unmarshalSubmitMsg(actualBytes)
+	actualSubmitMessage, err := protoutil.DecompressAndUnmarshall(actualBytes, &armadaevents.SubmitJob{}, decompressor)
 	assert.NoError(t, err)
-	expectedSubmitMessage, err := unmarshalSubmitMsg(expectedBytes)
+	expectedSubmitMessage, err := protoutil.DecompressAndUnmarshall(expectedBytes, &armadaevents.SubmitJob{}, decompressor)
 	assert.NoError(t, err)
 	assert.Equal(t, expectedSubmitMessage, actualSubmitMessage)
 }
 
-func unmarshalSubmitMsg(b []byte) (*armadaevents.SubmitJob, error) {
-	sm := &armadaevents.SubmitJob{}
-	err := proto.Unmarshal(b, sm)
-	return sm, err
-}
-
-func unmarshalSchedulingInfo(b []byte) (*schedulerobjects.JobSchedulingInfo, error) {
-	sm := &schedulerobjects.JobSchedulingInfo{}
-	err := proto.Unmarshal(b, sm)
-	return sm, err
-}
-
-func mustMarshall(msg proto.Message) []byte {
-	b, err := proto.Marshal(msg)
-	if err != nil {
-		panic(err)
-	}
-	return b
+func assertErrorMessagesEqual(t *testing.T, expectedBytes []byte, actualBytes []byte) {
+	actualError, err := protoutil.DecompressAndUnmarshall(actualBytes, &armadaevents.Error{}, decompressor)
+	assert.NoError(t, err)
+	expectedError, err := protoutil.DecompressAndUnmarshall(expectedBytes, &armadaevents.Error{}, decompressor)
+	assert.NoError(t, err)
+	assert.Equal(t, expectedError, actualError)
 }
diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go
index fa75316b..dc2682f7 100644
--- a/internal/scheduleringester/schedulerdb_test.go
+++ b/internal/scheduleringester/schedulerdb_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/exp/constraints"
+	"golang.org/x/exp/maps"
 
 	"github.com/armadaproject/armada/internal/common/util"
 	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
@@ -74,20 +75,26 @@ func TestWriteOps(t *testing.T) {
 				jobIds[1]: 3,
 			},
 		}},
-		"MarkJobSetsCancelled": {Ops: []DbOperation{
+		"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"},
 			},
-			InsertRuns{
-				runIds[0]: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0], JobSet: "set1"},
-				runIds[1]: &schedulerdb.Run{JobID: jobIds[1], RunID: runIds[1], JobSet: "set2"},
-				runIds[2]: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2], JobSet: "set1"},
-				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3], JobSet: "set2"},
+			MarkJobSetsCancelRequested{"set1": true},
+		}},
+		"MarkJobsCancelRequested": {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"},
+			},
+			MarkJobsCancelRequested{
+				jobIds[0]: true,
+				jobIds[1]: true,
 			},
-			MarkJobSetsCancelled{"set1": true},
 		}},
 		"MarkJobsCancelled": {Ops: []DbOperation{
 			InsertJobs{
@@ -96,12 +103,6 @@ func TestWriteOps(t *testing.T) {
 				jobIds[2]: &schedulerdb.Job{JobID: jobIds[2], JobSet: "set1"},
 				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]},
-			},
 			MarkJobsCancelled{
 				jobIds[0]: true,
 				jobIds[1]: true,
@@ -149,6 +150,20 @@ func TestWriteOps(t *testing.T) {
 				runIds[1]: true,
 			},
 		}},
+		"Insert JobRunErrors": {Ops: []DbOperation{
+			InsertJobRunErrors{
+				runIds[0]: &schedulerdb.JobRunError{
+					RunID: runIds[0],
+					JobID: jobIds[0],
+					Error: []byte{0x1},
+				},
+				runIds[1]: &schedulerdb.JobRunError{
+					RunID: runIds[1],
+					JobID: jobIds[1],
+					Error: []byte{0x2},
+				},
+			},
+		}},
 		"MarkRunsFailed": {Ops: []DbOperation{
 			InsertJobs{
 				jobIds[0]: &schedulerdb.Job{JobID: jobIds[0], JobSet: "set1"},
@@ -163,8 +178,8 @@ func TestWriteOps(t *testing.T) {
 				runIds[3]: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3]},
 			},
 			MarkRunsFailed{
-				runIds[0]: true,
-				runIds[1]: true,
+				runIds[0]: &JobRunFailed{LeaseReturned: true},
+				runIds[1]: &JobRunFailed{LeaseReturned: false},
 			},
 		}},
 		"MarkRunsRunning": {Ops: []DbOperation{
@@ -214,8 +229,8 @@ func addDefaultValues(op DbOperation) DbOperation {
 		}
 	case InsertRuns:
 	case UpdateJobSetPriorities:
-	case MarkJobSetsCancelled:
-	case MarkJobsCancelled:
+	case MarkJobSetsCancelRequested:
+	case MarkJobsCancelRequested:
 	case MarkJobsSucceeded:
 	case MarkJobsFailed:
 	case UpdateJobPriorities:
@@ -302,7 +317,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			}
 		}
 		assert.Greater(t, numChanged, 0)
-	case MarkJobSetsCancelled:
+	case MarkJobSetsCancelRequested:
 		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
@@ -311,24 +326,27 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		jobIds := make([]string, 0)
 		for _, job := range jobs {
 			if _, ok := expected[job.JobSet]; ok {
-				assert.True(t, job.Cancelled)
+				assert.True(t, job.CancelRequested)
 				numChanged++
 				jobIds = append(jobIds, job.JobID)
 			}
 		}
 		assert.Greater(t, numChanged, 0)
-
-		runs, err := queries.SelectNewRunsForJobs(ctx, schedulerdb.SelectNewRunsForJobsParams{
-			Serial: serials["runs"],
-			JobIds: jobIds,
-		})
+	case MarkJobsCancelRequested:
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
-		assert.Greater(t, len(runs), 0)
-		for _, run := range runs {
-			assert.True(t, run.Cancelled)
+		numChanged := 0
+		jobIds := make([]string, 0)
+		for _, job := range jobs {
+			if _, ok := expected[job.JobID]; ok {
+				assert.True(t, job.CancelRequested)
+				numChanged++
+				jobIds = append(jobIds, job.JobID)
+			}
 		}
+		assert.Equal(t, len(expected), numChanged)
 	case MarkJobsCancelled:
 		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
@@ -343,19 +361,6 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 				jobIds = append(jobIds, job.JobID)
 			}
 		}
-		assert.Equal(t, len(expected), numChanged)
-
-		runs, err := queries.SelectNewRunsForJobs(ctx, schedulerdb.SelectNewRunsForJobsParams{
-			Serial: serials["runs"],
-			JobIds: jobIds,
-		})
-		if err != nil {
-			return errors.WithStack(err)
-		}
-		assert.Greater(t, len(runs), 0)
-		for _, run := range runs {
-			assert.True(t, run.Cancelled)
-		}
 	case MarkJobsSucceeded:
 		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
@@ -439,8 +444,9 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		numChanged := 0
 		for _, run := range runs {
-			if _, ok := expected[run.RunID]; ok {
+			if expectedRun, ok := expected[run.RunID]; ok {
 				assert.True(t, run.Failed)
+				assert.Equal(t, expectedRun.LeaseReturned, run.Returned)
 				numChanged++
 			}
 		}
@@ -470,6 +476,21 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			}
 		}
 		assert.Equal(t, len(expected), len(runs))
+	case InsertJobRunErrors:
+		expectedIds := maps.Keys(expected)
+		as, err := queries.SelectRunErrorsById(ctx, expectedIds)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		actual := make(InsertJobRunErrors, len(as))
+		for _, a := range as {
+			actual[a.RunID] = &schedulerdb.JobRunError{
+				RunID: a.RunID,
+				JobID: a.JobID,
+				Error: a.Error,
+			}
+		}
+		assert.Equal(t, expected, actual)
 	default:
 		return errors.Errorf("received unexpected op %+v", op)
 	}