Repository: armadaproject/armada
Pull request #3180 (2023-12-07 11:14:58)

Scheduler reconciliation fixes and testing


Patch:
diff --git a/cmd/scheduler/cmd/migrate_database.go b/cmd/scheduler/cmd/migrate_database.go
index 22d6dc12..9e48f5fb 100644
--- a/cmd/scheduler/cmd/migrate_database.go
+++ b/cmd/scheduler/cmd/migrate_database.go
@@ -21,7 +21,8 @@ func migrateDbCmd() *cobra.Command {
 	cmd.Flags().Duration(
 		"timeout",
 		5*time.Minute,
-		"Duration after which the migration will fail if it has not been created")
+		"Duration after which the migration will fail if it has not been created",
+	)
 
 	return cmd
 }
@@ -40,7 +41,7 @@ func migrateDatabase(cmd *cobra.Command, _ []string) error {
 	log.Info("Beginning scheduler database migration")
 	db, err := database.OpenPgxConn(config.Postgres)
 	if err != nil {
-		return errors.WithMessagef(err, "Failed to connect to database")
+		return errors.WithMessagef(err, "failed to connect to database")
 	}
 
 	ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), timeout)
diff --git a/cmd/scheduler/main.go b/cmd/scheduler/main.go
index feda278e..f895b330 100644
--- a/cmd/scheduler/main.go
+++ b/cmd/scheduler/main.go
@@ -11,8 +11,7 @@ import (
 func main() {
 	common.ConfigureLogging()
 	common.BindCommandlineArguments()
-	err := cmd.RootCmd().Execute()
-	if err != nil {
+	if err := cmd.RootCmd().Execute(); err != nil {
 		os.Exit(1)
 	}
 }
diff --git a/cmd/scheduleringester/main.go b/cmd/scheduleringester/main.go
index 5d20f3f3..9aa5207a 100644
--- a/cmd/scheduleringester/main.go
+++ b/cmd/scheduleringester/main.go
@@ -1,6 +1,9 @@
 package main
 
 import (
+	"fmt"
+	"os"
+
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 
@@ -29,5 +32,8 @@ func main() {
 
 	common.LoadConfig(&config, "./config/scheduleringester", userSpecifiedConfigs)
 
-	scheduleringester.Run(config)
+	if err := scheduleringester.Run(config); err != nil {
+		fmt.Println(err)
+		os.Exit(-1)
+	}
 }
diff --git a/internal/common/database/migrations.go b/internal/common/database/migrations.go
index b515c94f..b8f285d0 100644
--- a/internal/common/database/migrations.go
+++ b/internal/common/database/migrations.go
@@ -9,7 +9,6 @@ import (
 	"strings"
 
 	stakikfs "github.com/rakyll/statik/fs"
-	log "github.com/sirupsen/logrus"
 
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 )
@@ -30,16 +29,17 @@ func NewMigration(id int, name string, sql string) Migration {
 }
 
 func UpdateDatabase(ctx *armadacontext.Context, db Querier, migrations []Migration) error {
-	log.Info("Updating postgres...")
+	ctx.Info("Preparing to apply postgres migrations.")
 	version, err := readVersion(ctx, db)
 	if err != nil {
 		return err
 	}
-	log.Infof("Current version %v", version)
+	ctx.Infof("Current version: %d", version)
 
+	originalVersion := version
 	for _, m := range migrations {
 		if m.id > version {
-			log.Debugf("Executing %s", m.name)
+			ctx.Infof("Applying %s", m.name)
 			_, err := db.Exec(ctx, m.sql)
 			if err != nil {
 				return err
@@ -50,9 +50,16 @@ func UpdateDatabase(ctx *armadacontext.Context, db Querier, migrations []Migrati
 			if err != nil {
 				return err
 			}
+		} else {
+			ctx.Infof("Not applying %s as migration id %d is <= postgres version %d", m.name, m.id, version)
 		}
 	}
-	log.Info("Database updated.")
+
+	if version == originalVersion {
+		ctx.Info("Postgres was already the up-to-date")
+	} else {
+		ctx.Infof("Postgres updates from version %d to %d", originalVersion, version)
+	}
 	return nil
 }
 
diff --git a/internal/scheduler/api.go b/internal/scheduler/api.go
index 533abc4b..985ba876 100644
--- a/internal/scheduler/api.go
+++ b/internal/scheduler/api.go
@@ -213,7 +213,7 @@ func setPriorityClassName(podSpec *armadaevents.PodSpecWithAvoidList, priorityCl
 	podSpec.PodSpec.PriorityClassName = priorityClassName
 }
 
-// ReportEvents publishes all events to Pulsar. The events are compacted for more efficient publishing.
+// ReportEvents publishes all eventSequences to Pulsar. The eventSequences are compacted for more efficient publishing.
 func (srv *ExecutorApi) ReportEvents(grpcCtx context.Context, list *executorapi.EventList) (*types.Empty, error) {
 	ctx := armadacontext.FromGrpcCtx(grpcCtx)
 	err := pulsarutils.CompactAndPublishSequences(ctx, list.Events, srv.producer, srv.maxPulsarMessageSizeBytes, schedulers.Pulsar)
diff --git a/internal/scheduler/database/job_repository.go b/internal/scheduler/database/job_repository.go
index 684ad47a..ed83eda5 100644
--- a/internal/scheduler/database/job_repository.go
+++ b/internal/scheduler/database/job_repository.go
@@ -132,7 +132,10 @@ func (r *PostgresJobRepository) FetchJobUpdates(ctx *armadacontext.Context, jobS
 
 	start := time.Now()
 	defer func() {
-		ctx.Infof("received %d updated jobs and %d updated job runs from postgres in %s", len(updatedJobs), len(updatedRuns), time.Since(start))
+		ctx.Infof(
+			"received %d updated jobs and %d updated job runs from postgres in %s",
+			len(updatedJobs), len(updatedRuns), time.Since(start),
+		)
 	}()
 
 	// Use a RepeatableRead transaction here so that we get consistency between jobs and dbRuns
diff --git a/internal/scheduler/jobdb/job.go b/internal/scheduler/jobdb/job.go
index fc4bf4c4..84615080 100644
--- a/internal/scheduler/jobdb/job.go
+++ b/internal/scheduler/jobdb/job.go
@@ -1,10 +1,13 @@
 package jobdb
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/google/uuid"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
 	"golang.org/x/exp/maps"
 	v1 "k8s.io/api/core/v1"
 
@@ -15,6 +18,9 @@ import (
 
 // Job is the scheduler-internal representation of a job.
 type Job struct {
+	// Pointer to the jobDb used to create this job.
+	// Not included in equals comparison.
+	jobDb *JobDb
 	// String representation of the job id.
 	id string
 	// Name of the queue this job belongs to.
@@ -63,6 +69,189 @@ func EmptyJob(id string) *Job {
 	return &Job{id: id, runsById: map[uuid.UUID]*JobRun{}}
 }
 
+func (job *Job) String() string {
+	// Include field names in string representation by default.
+	return fmt.Sprintf("%#v", job)
+}
+
+// Assert makes the assertions outlined below and returns
+// - nil if the job is valid and
+// - an error explaining why not otherwise.
+//
+// Assertions:
+// Required fields must be set.
+//
+// All associated runs are valid. The state of LatestRun is compatible with that of the job.
+//
+// The states {Queued, Running, Cancelled, Failed, Succeeded} are mutually exclusive.
+//
+// Only these job state transitions are valid:
+// - Queued -> {Running, Cancelled, Failed}
+// - Running -> {Queued, Cancelled, Failed, Succeeded}
+// - Succeeded, Cancelled, Failed -> {}
+//
+// QueuedVersion is 0 initially and is incremented by 1 every time the queued status of the job changes. Hence:
+// - If the job is queued, the queuedVersion must be 2x the number of runs. E.g.:
+//   - queued (qv = 0)
+//   - queued -> running -> queued: qv = 2
+//
+// - If the job is running or succeeded, the queuedVersion must be one less than 2x number of runs. E.g.:
+//   - queued -> running -> succeeded: qv = 1
+//   - queued -> running -> queued -> running -> succeeded: qv = 3
+//
+// - If the job is failed or cancelled, the queuedVersion may be either equal to or 1 less than 2x the number of runs. E.g.:
+//   - queued -> failed/cancelled: qv = 0
+//   - queued -> running -> failed/cancelled: qv = 1
+//   - queued -> running -> queued -> failed/cancelled: qv = 2
+func (job *Job) Assert() error {
+	if job == nil {
+		return errors.Errorf("job is nil")
+	}
+	var result *multierror.Error
+
+	// Assert that required fields are set.
+	if job.Id() == "" {
+		result = multierror.Append(result, errors.New("job has an empty id"))
+	}
+	if job.Queue() == "" {
+		result = multierror.Append(result, errors.New("job has an empty queue"))
+	}
+	if job.Jobset() == "" {
+		result = multierror.Append(result, errors.New("job has an empty jobSet"))
+	}
+
+	// Assert that runs are valid.
+	latestRun := job.LatestRun()
+	for runId, run := range job.runsById {
+		if runId != run.Id() {
+			result = multierror.Append(result, errors.Errorf("run %s is keyed by %s", run.Id(), runId))
+		}
+		if run.Created() > latestRun.Created() {
+			result = multierror.Append(result, errors.Errorf("run %s is more recent than latestRun", run.Id()))
+		}
+		if run.JobId() != job.Id() {
+			result = multierror.Append(result, errors.Errorf("run %s has a conflicting jobId", run.Id()))
+		}
+		if err := run.Assert(); err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+
+	// Assertions specific to the state of the job.
+	if job.Queued() {
+		if job.Succeeded() {
+			result = multierror.Append(result, errors.New("job is marked as both queued and succeeded"))
+		}
+		if job.Cancelled() {
+			result = multierror.Append(result, errors.New("job is marked as both queued and cancelled"))
+		}
+		if job.Failed() {
+			result = multierror.Append(result, errors.New("job is marked as both queued and failed"))
+		}
+		if job.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is marked as queued and terminated"))
+		}
+
+		if 2*int32(len(job.runsById)) != job.QueuedVersion() {
+			result = multierror.Append(result, errors.Errorf("job is queued but queuedVersion (%d) is not 2x the number of runs (%d)", job.queuedVersion, len(job.runsById)))
+		}
+
+		if run := job.LatestRun(); run != nil {
+			if run.Running() {
+				result = multierror.Append(result, errors.New("job is queued but latestRun is running"))
+			} else if run.Succeeded() {
+				result = multierror.Append(result, errors.New("job is queued but latestRun is succeeded"))
+			} else if run.Cancelled() {
+				result = multierror.Append(result, errors.New("job is queued but latestRun is cancelled"))
+			}
+		}
+	} else if job.Succeeded() {
+		if job.Cancelled() {
+			result = multierror.Append(result, errors.New("job is marked as both succeeded and cancelled"))
+		}
+		if job.Failed() {
+			result = multierror.Append(result, errors.New("job is marked as both succeeded and failed"))
+		}
+		if !job.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is marked as succeeded but not terminated"))
+		}
+
+		if 2*int32(len(job.runsById))-1 != job.QueuedVersion() {
+			result = multierror.Append(result, errors.Errorf("job is succeeded but queuedVersion (%d) is not one less than 2x the number of runs (%d)", job.queuedVersion, len(job.runsById)))
+		}
+
+		if run := job.LatestRun(); run != nil {
+			if !run.Succeeded() {
+				// A job succeeds iff a job succeeds.
+				result = multierror.Append(result, errors.New("job is succeeded but latestRun is not"))
+			}
+		} else {
+			result = multierror.Append(result, errors.New("job is succeeded but has no associated runs"))
+		}
+	} else if job.Cancelled() {
+		if !job.CancelRequested() && !job.CancelByJobsetRequested() {
+			result = multierror.Append(result, errors.New("job is cancelled but not marked as cancel requested"))
+		}
+
+		if run := job.LatestRun(); run != nil && !job.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is cancelled but has an active run"))
+		}
+		if job.Failed() {
+			result = multierror.Append(result, errors.New("job is marked as both cancelled and failed"))
+		}
+		if !job.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is marked as cancelled but not terminated"))
+		}
+
+		if 2*int32(len(job.runsById)) != job.QueuedVersion() && 2*int32(len(job.runsById))-1 != job.QueuedVersion() {
+			result = multierror.Append(result, errors.Errorf("job is cancelled but queuedVersion (%d) is not either equal to or one less than 2x the number of runs (%d)", job.queuedVersion, len(job.runsById)))
+		}
+
+		// A job may be cancelled regardless of whether it has associated runs or not,
+		// and whether those runs succeeded or not.
+		if run := job.LatestRun(); run != nil && !run.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is cancelled but has an active run"))
+		}
+	} else if job.Failed() {
+		if !job.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is marked as failed but not terminated"))
+		}
+
+		if 2*int32(len(job.runsById)) != job.QueuedVersion() && 2*int32(len(job.runsById))-1 != job.QueuedVersion() {
+			result = multierror.Append(result, errors.Errorf("job is failed but queuedVersion (%d) is not either equal to or one less than 2x the number of runs (%d)", job.queuedVersion, len(job.runsById)))
+		}
+
+		if run := job.LatestRun(); run != nil && !run.InTerminalState() {
+			result = multierror.Append(result, errors.New("job is failed but has an active run"))
+		}
+	} else {
+		// Job must be running if it's not in any of the other states.
+		if 2*int32(len(job.runsById))-1 != job.QueuedVersion() {
+			result = multierror.Append(result, errors.Errorf("job is running but queuedVersion (%d) is not one less than 2x the number of runs (%d)", job.queuedVersion, len(job.runsById)))
+		}
+
+		if run := job.LatestRun(); run != nil {
+			if run.Cancelled() {
+				result = multierror.Append(result, errors.New("job is running but latestRun is cancelled"))
+			} else if run.Failed() {
+				result = multierror.Append(result, errors.New("job is running but latestRun is failed"))
+			} else if run.Succeeded() {
+				result = multierror.Append(result, errors.New("job is running but latestRun is succeeded"))
+			} else if run.Returned() {
+				result = multierror.Append(result, errors.New("job is running but latestRun is returned"))
+			}
+		} else {
+			result = multierror.Append(result, errors.New("job is running but has no associated run"))
+		}
+	}
+
+	if err := result.ErrorOrNil(); err != nil {
+		// Avoid allocating the message "... invalid state" string if there were no errors.
+		return errors.WithMessagef(err, "invalid job: %s", job)
+	}
+	return nil
+}
+
 func (job *Job) ensureJobSchedulingInfoFieldsInitialised() {
 	// Initialise the annotation and nodeSelector maps if nil.
 	// Since those need to be mutated in-place.
@@ -111,7 +300,7 @@ func (job *Job) Equal(other *Job) bool {
 		return false
 	}
 	if job.schedulingKey != other.schedulingKey {
-		// We assume jobSchedulingInfo is equal if schedulingKey is equal.
+		// Assume jobSchedulingInfo is equal if schedulingKey is equal.
 		return false
 	}
 	if job.queued != other.queued {
@@ -120,7 +309,7 @@ func (job *Job) Equal(other *Job) bool {
 	if job.queuedVersion != other.queuedVersion {
 		return false
 	}
-	if job.priorityClass.Equal(other.priorityClass) {
+	if !job.priorityClass.Equal(other.priorityClass) {
 		return false
 	}
 	if job.queued != other.queued {
@@ -144,15 +333,15 @@ func (job *Job) Equal(other *Job) bool {
 	if job.succeeded != other.succeeded {
 		return false
 	}
-	if !armadamaps.DeepEqual(job.runsById, other.runsById) {
-		return false
-	}
 	if !job.activeRun.Equal(other.activeRun) {
 		return false
 	}
 	if job.activeRunTimestamp != other.activeRunTimestamp {
 		return false
 	}
+	if !armadamaps.DeepEqual(job.runsById, other.runsById) {
+		return false
+	}
 	return true
 }
 
@@ -226,6 +415,13 @@ func (job *Job) WithPriority(priority uint32) *Job {
 	return j
 }
 
+// WithSubmittedTime returns a copy of the job with submittedTime updated.
+func (job *Job) WithSubmittedTime(submittedTime int64) *Job {
+	j := copyJob(*job)
+	j.submittedTime = submittedTime
+	return j
+}
+
 // WithRequestedPriority returns a copy of the job with the priority updated.
 func (job *Job) WithRequestedPriority(priority uint32) *Job {
 	j := copyJob(*job)
@@ -402,6 +598,13 @@ func (job *Job) Created() int64 {
 	return job.submittedTime
 }
 
+// ActiveRunTimestamp returns the creation time of the most recent run associated with this job.
+func (job *Job) ActiveRunTimestamp() int64 {
+	return job.activeRunTimestamp
+}
+
+// The timestamp of the currently active run.
+
 // InTerminalState returns true if the job  is in a terminal state
 func (job *Job) InTerminalState() bool {
 	return job.succeeded || job.cancelled || job.failed
@@ -415,30 +618,36 @@ func (job *Job) HasRuns() bool {
 
 // WithNewRun creates a copy of the job with a new run on the given executor.
 func (job *Job) WithNewRun(executor string, nodeId, nodeName string, scheduledAtPriority int32) *Job {
-	run := &JobRun{
-		id:                  uuid.New(),
-		jobId:               job.id,
-		created:             time.Now().UnixNano(),
-		executor:            executor,
-		nodeId:              nodeId,
-		nodeName:            nodeName,
-		scheduledAtPriority: &scheduledAtPriority,
-	}
-	return job.WithUpdatedRun(run)
-}
-
-// WithUpdatedRun creates a copy of the job with run details updated.
+	return job.WithUpdatedRun(job.jobDb.CreateRun(
+		job.jobDb.uuidProvider.New(),
+		job.Id(),
+		job.jobDb.clock.Now().UnixNano(),
+		executor,
+		nodeId,
+		nodeName,
+		&scheduledAtPriority,
+		false,
+		false,
+		false,
+		false,
+		false,
+		false,
+	))
+}
+
+// WithUpdatedRun returns a copy of the job with the provided run upserted.
 func (job *Job) WithUpdatedRun(run *JobRun) *Job {
 	j := copyJob(*job)
-	j.runsById = maps.Clone(j.runsById)
-	if run.created >= j.activeRunTimestamp {
+	if j.activeRun == nil || run.created >= j.activeRunTimestamp {
 		j.activeRunTimestamp = run.created
 		j.activeRun = run
 	}
-	if j.runsById == nil {
-		j.runsById = make(map[uuid.UUID]*JobRun)
+	if j.runsById != nil {
+		j.runsById = maps.Clone(j.runsById)
+		j.runsById[run.id] = run
+	} else {
+		j.runsById = map[uuid.UUID]*JobRun{run.id: run}
 	}
-	j.runsById[run.id] = run
 	return j
 }
 
@@ -454,7 +663,7 @@ func (job *Job) NumReturned() uint {
 	return returned
 }
 
-// NumAttempts returns the number of times the executors tried to run this job
+// NumAttempts returns the number of times the executors tried to run this job.
 // Note that this is O(N) on Runs, but this should be fine as the number of runs should be small.
 func (job *Job) NumAttempts() uint {
 	attempts := uint(0)
diff --git a/internal/scheduler/jobdb/job_run.go b/internal/scheduler/jobdb/job_run.go
index bbe7b363..c7b00315 100644
--- a/internal/scheduler/jobdb/job_run.go
+++ b/internal/scheduler/jobdb/job_run.go
@@ -1,7 +1,11 @@
 package jobdb
 
 import (
+	"fmt"
+
 	"github.com/google/uuid"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
 )
 
 // JobRun is the scheduler-internal representation of a job run.
@@ -36,6 +40,87 @@ type JobRun struct {
 	runAttempted bool
 }
 
+func (run *JobRun) String() string {
+	// Include field names in string representation by default.
+	return fmt.Sprintf("%#v", run)
+}
+
+// Assert makes the assertions outlined below and returns
+// - nil if the run is valid and
+// - an error explaining why not otherwise.
+//
+// Assertions:
+// Required fields must be set.
+//
+// The states {Running, Cancelled, Failed, Succeeded, Returned} are mutually exclusive.
+func (run *JobRun) Assert() error {
+	if run == nil {
+		return errors.Errorf("run is nil")
+	}
+	var result *multierror.Error
+
+	// Assert that required fields are set.
+	var emptyUuid uuid.UUID
+	if run.Id() == emptyUuid {
+		result = multierror.Append(result, errors.New("run has an empty id"))
+	}
+	if run.JobId() == "" {
+		result = multierror.Append(result, errors.New("run has an empty jobId"))
+	}
+	if run.Executor() == "" {
+		result = multierror.Append(result, errors.New("run has an empty executor"))
+	}
+	if run.NodeId() == "" {
+		result = multierror.Append(result, errors.New("run has an empty nodeId"))
+	}
+	if run.NodeName() == "" {
+		result = multierror.Append(result, errors.New("run has an empty nodeName"))
+	}
+
+	// Assertions specific to the state of the run.
+	if run.Running() {
+		if run.Succeeded() {
+			result = multierror.Append(result, errors.New("run is marked as both running and succeeded"))
+		}
+		if run.Failed() {
+			result = multierror.Append(result, errors.New("run is marked as both running and failed"))
+		}
+		if run.Cancelled() {
+			result = multierror.Append(result, errors.New("run is marked as both running and cancelled"))
+		}
+		if run.Returned() {
+			result = multierror.Append(result, errors.New("run is marked as both running and returned"))
+		}
+	} else if run.Succeeded() {
+		if run.Failed() {
+			result = multierror.Append(result, errors.New("run is marked as both succeeded and failed"))
+		}
+		if run.Cancelled() {
+			result = multierror.Append(result, errors.New("run is marked as both succeeded and cancelled"))
+		}
+		if run.Returned() {
+			result = multierror.Append(result, errors.New("run is marked as both succeeded and returned"))
+		}
+	} else if run.Failed() {
+		if run.Cancelled() {
+			result = multierror.Append(result, errors.New("run is marked as both failed and cancelled"))
+		}
+	} else if run.Cancelled() {
+		if run.Returned() {
+			result = multierror.Append(result, errors.New("run is marked as both cancelled and returned"))
+		}
+	} else if run.Returned() {
+		// Nothing to do.
+	} else {
+		// Run is leased or pending; nothing to do.
+	}
+	if err := result.ErrorOrNil(); err != nil {
+		// Avoid allocating the message "... invalid state" string if there were no errors.
+		return errors.WithMessagef(err, "invalid run: %s", run)
+	}
+	return nil
+}
+
 func (run *JobRun) Equal(other *JobRun) bool {
 	if run == other {
 		return true
@@ -46,7 +131,52 @@ func (run *JobRun) Equal(other *JobRun) bool {
 	if run != nil && other == nil {
 		return false
 	}
-	return *run == *other
+	if run.id != other.id {
+		return false
+	}
+	if run.jobId != other.jobId {
+		return false
+	}
+	if run.created != other.created {
+		return false
+	}
+	if run.executor != other.executor {
+		return false
+	}
+	if run.nodeId != other.nodeId {
+		return false
+	}
+	if run.nodeName != other.nodeName {
+		return false
+	}
+	if run.scheduledAtPriority != other.scheduledAtPriority {
+		if run.scheduledAtPriority != nil && other.scheduledAtPriority != nil {
+			if *run.scheduledAtPriority != *other.scheduledAtPriority {
+				return false
+			}
+		} else {
+			return false
+		}
+	}
+	if run.running != other.running {
+		return false
+	}
+	if run.succeeded != other.succeeded {
+		return false
+	}
+	if run.failed != other.failed {
+		return false
+	}
+	if run.cancelled != other.cancelled {
+		return false
+	}
+	if run.returned != other.returned {
+		return false
+	}
+	if run.runAttempted != other.runAttempted {
+		return false
+	}
+	return true
 }
 
 func MinimalRun(id uuid.UUID, creationTime int64) *JobRun {
diff --git a/internal/scheduler/jobdb/jobdb.go b/internal/scheduler/jobdb/jobdb.go
index 1ab68a59..47e04527 100644
--- a/internal/scheduler/jobdb/jobdb.go
+++ b/internal/scheduler/jobdb/jobdb.go
@@ -6,8 +6,10 @@ import (
 
 	"github.com/benbjohnson/immutable"
 	"github.com/google/uuid"
+	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"golang.org/x/exp/maps"
+	"k8s.io/apimachinery/pkg/util/clock"
 
 	"github.com/armadaproject/armada/internal/common/stringinterner"
 	"github.com/armadaproject/armada/internal/common/types"
@@ -32,8 +34,26 @@ type JobDb struct {
 	schedulingKeyGenerator *schedulerobjects.SchedulingKeyGenerator
 	// We intern strings to save memory.
 	stringInterner *stringinterner.StringInterner
-	copyMutex      sync.Mutex
-	writerMutex    sync.Mutex
+	// Mutexes protecting the jobDb.
+	copyMutex   sync.Mutex
+	writerMutex sync.Mutex
+	// Clock used when assigning timestamps to created job runs.
+	// Set here so that it can be mocked.
+	clock clock.PassiveClock
+	// Used for generating job run ids.
+	uuidProvider UUIDProvider
+}
+
+// UUIDProvider is an interface used to mock UUID generation for tests.
+type UUIDProvider interface {
+	New() uuid.UUID
+}
+
+// RealUUIDProvider calls uuid.New.
+type RealUUIDProvider struct{}
+
+func (_ RealUUIDProvider) New() uuid.UUID {
+	return uuid.New()
 }
 
 func NewJobDb(priorityClasses map[string]types.PriorityClass, defaultPriorityClassName string, stringInternerCacheSize uint32) *JobDb {
@@ -65,6 +85,30 @@ func NewJobDbWithSchedulingKeyGenerator(
 		defaultPriorityClass:   defaultPriorityClass,
 		schedulingKeyGenerator: skg,
 		stringInterner:         stringinterner.New(stringInternerCacheSize),
+		clock:                  clock.RealClock{},
+		uuidProvider:           RealUUIDProvider{},
+	}
+}
+
+func (jobDb *JobDb) SetClock(clock clock.PassiveClock) {
+	jobDb.clock = clock
+}
+
+func (jobDb *JobDb) SetUUIDProvider(uuidProvider UUIDProvider) {
+	jobDb.uuidProvider = uuidProvider
+}
+
+// Clone returns a copy of the jobDb.
+func (jobDb *JobDb) Clone() *JobDb {
+	return &JobDb{
+		jobsById:               jobDb.jobsById,
+		jobsByRunId:            jobDb.jobsByRunId,
+		jobsByQueue:            maps.Clone(jobDb.jobsByQueue),
+		queuedJobsByTtl:        jobDb.queuedJobsByTtl,
+		priorityClasses:        jobDb.priorityClasses,
+		defaultPriorityClass:   jobDb.defaultPriorityClass,
+		schedulingKeyGenerator: jobDb.schedulingKeyGenerator,
+		stringInterner:         jobDb.stringInterner,
 	}
 }
 
@@ -88,6 +132,7 @@ func (jobDb *JobDb) NewJob(
 		priorityClass = jobDb.defaultPriorityClass
 	}
 	job := &Job{
+		jobDb:                   jobDb,
 		id:                      jobId,
 		queue:                   jobDb.stringInterner.Intern(queue),
 		jobSet:                  jobDb.stringInterner.Intern(jobSet),
@@ -174,8 +219,10 @@ type Txn struct {
 	// Queued jobs for each queue ordered by remaining time-to-live.
 	// TODO: The ordering is wrong. Since we call time.Now() in the compare function.
 	queuedJobsByTtl *immutable.SortedSet[*Job]
-	jobDb           *JobDb
-	active          bool
+	// The jobDb from which this transaction was created.
+	jobDb *JobDb
+	// Set to false when this transaction is either committed or aborted.
+	active bool
 }
 
 func (txn *Txn) Commit() {
@@ -192,6 +239,92 @@ func (txn *Txn) Commit() {
 	txn.active = false
 }
 
+// Assert returns an error if the jobDb, or any job stored in the jobDb, is in an invalid state.
+// If assertOnlyActiveJobs is true, it also asserts that all jobs in the jobDb are active.
+func (txn *Txn) Assert(assertOnlyActiveJobs bool) error {
+	it := txn.jobsById.Iterator()
+	for {
+		jobId, job, ok := it.Next()
+		if !ok {
+			break
+		}
+		if job.Id() != jobId {
+			return errors.Errorf("jobDb contains a job with misaligned jobId: %s != %s", job.Id(), jobId)
+		}
+		if err := job.Assert(); err != nil {
+			return errors.WithMessage(err, "jobDb is invalid")
+		}
+		if assertOnlyActiveJobs && job.InTerminalState() {
+			return errors.Errorf("jobDb contains an inactive job %s", job)
+		}
+		if job.Queued() {
+			if queue, ok := txn.jobsByQueue[job.queue]; !ok {
+				return errors.Errorf("jobDb contains queued job %s but there is no sorted set for this queue", job)
+			} else if !queue.Has(job) {
+				return errors.Errorf("jobDb contains queued job %s but this job is not in the queue sorted set", job)
+			}
+		}
+		for runId := range job.runsById {
+			if otherJobId, ok := txn.jobsByRunId.Get(runId); !ok {
+				return errors.Errorf("jobDb contains job %s but there is no mapping from runId %s to this job", job, runId)
+			} else if jobId != otherJobId {
+				return errors.Errorf("jobDb contains job %s but runId %s does not map to this job", job, runId)
+			}
+		}
+	}
+	for queue, queueIt := range txn.jobsByQueue {
+		it := queueIt.Iterator()
+		for {
+			job, ok := it.Next()
+			if !ok {
+				break
+			}
+			if job.queue != queue {
+				return errors.Errorf("jobDb queue %s contains job %s but this job is in queue %s", queue, job, job.queue)
+			} else if other, ok := txn.jobsById.Get(job.id); !ok {
+				return errors.Errorf("jobDb queue %s contains job %s but this job is not in the jobDb", queue, job)
+			} else if !job.Equal(other) {
+				return errors.Errorf("jobDb queue %s contains job %s but this job differs from that stored in the jobDb %s", queue, job, other)
+			}
+		}
+	}
+	return nil
+}
+
+func (txn *Txn) AssertEqual(otherTxn *Txn) error {
+	var result *multierror.Error
+	it := txn.jobsById.Iterator()
+	for {
+		jobId, job, ok := it.Next()
+		if !ok {
+			break
+		}
+		otherJob := otherTxn.GetById(jobId)
+		if job != nil && otherJob == nil {
+			result = multierror.Append(result, errors.Errorf("job %s is not in otherTxn", jobId))
+		} else if job == nil && otherTxn != nil {
+			result = multierror.Append(result, errors.Errorf("job %s is not in txn", jobId))
+		} else if !job.Equal(otherJob) {
+			result = multierror.Append(result, errors.Errorf("job %s differs between txn and otherTxn: %s is not equal to %s", jobId, job, otherJob))
+		}
+	}
+	it = otherTxn.jobsById.Iterator()
+	for {
+		jobId, _, ok := it.Next()
+		if !ok {
+			break
+		}
+		job := otherTxn.GetById(jobId)
+		if job == nil && otherTxn != nil {
+			result = multierror.Append(result, errors.Errorf("job %s is not in txn", jobId))
+		}
+	}
+	if err := result.ErrorOrNil(); err != nil {
+		return errors.Wrap(err, "jobDb transactions are not equal")
+	}
+	return nil
+}
+
 func (txn *Txn) Abort() {
 	if txn.readOnly || !txn.active {
 		return
@@ -343,31 +476,37 @@ func (txn *Txn) GetAll() []*Job {
 
 // BatchDelete deletes the jobs with the given ids from the database.
 // Any ids not in the database are ignored.
-func (txn *Txn) BatchDelete(ids []string) error {
+func (txn *Txn) BatchDelete(jobIds []string) error {
 	if err := txn.checkWritableTransaction(); err != nil {
 		return err
 	}
-	for _, id := range ids {
-		job, present := txn.jobsById.Get(id)
-		if present {
-			txn.jobsById = txn.jobsById.Delete(id)
-			for _, run := range job.runsById {
-				txn.jobsByRunId = txn.jobsByRunId.Delete(run.id)
-			}
-			queue, ok := txn.jobsByQueue[job.queue]
-			if ok {
-				newQueue := queue.Delete(job)
-				txn.jobsByQueue[job.queue] = newQueue
-			}
+	for _, id := range jobIds {
+		txn.delete(id)
+	}
+	return nil
+}
 
-			// We only add these jobs into the collection if it has a queueTtl set, hence only remove if this is set.
-			if job.HasQueueTtlSet() {
-				newQueuedJobsByExpiry := txn.queuedJobsByTtl.Delete(job)
-				txn.queuedJobsByTtl = &newQueuedJobsByExpiry
-			}
+// delete a job from the txn.
+// The caller is responsible for checking that this is a writable txn by calling checkWritableTransaction.
+func (txn *Txn) delete(jobId string) {
+	job, present := txn.jobsById.Get(jobId)
+	if present {
+		txn.jobsById = txn.jobsById.Delete(jobId)
+		for _, run := range job.runsById {
+			txn.jobsByRunId = txn.jobsByRunId.Delete(run.id)
+		}
+		queue, ok := txn.jobsByQueue[job.queue]
+		if ok {
+			newQueue := queue.Delete(job)
+			txn.jobsByQueue[job.queue] = newQueue
+		}
+
+		// We only add these jobs into the collection if it has a queueTtl set, hence only remove if this is set.
+		if job.HasQueueTtlSet() {
+			newQueuedJobsByExpiry := txn.queuedJobsByTtl.Delete(job)
+			txn.queuedJobsByTtl = &newQueuedJobsByExpiry
 		}
 	}
-	return nil
 }
 
 func (txn *Txn) checkWritableTransaction() error {
diff --git a/internal/scheduler/jobdb/reconciliation.go b/internal/scheduler/jobdb/reconciliation.go
index 084a3843..505f28d7 100644
--- a/internal/scheduler/jobdb/reconciliation.go
+++ b/internal/scheduler/jobdb/reconciliation.go
@@ -202,7 +202,7 @@ func (jobDb *JobDb) schedulerJobFromDatabaseJob(dbJob *database.Job) (*Job, erro
 	if err := proto.Unmarshal(dbJob.SchedulingInfo, schedulingInfo); err != nil {
 		return nil, errors.Wrapf(err, "error unmarshalling scheduling info for job %s", dbJob.JobID)
 	}
-	return jobDb.NewJob(
+	job := jobDb.NewJob(
 		dbJob.JobID,
 		dbJob.JobSet,
 		dbJob.Queue,
@@ -214,7 +214,20 @@ func (jobDb *JobDb) schedulerJobFromDatabaseJob(dbJob *database.Job) (*Job, erro
 		dbJob.CancelByJobsetRequested,
 		dbJob.Cancelled,
 		dbJob.Submitted,
-	), nil
+	)
+	if dbJob.Failed {
+		// TODO(albin): Let's make this an argument to NewJob. Even better: have the state as an enum argument.
+		job = job.WithFailed(dbJob.Failed)
+	}
+	if dbJob.Succeeded {
+		// TODO(albin): Same comment as the above.
+		job = job.WithSucceeded(dbJob.Succeeded)
+	}
+	if uint32(dbJob.Priority) != job.RequestedPriority() {
+		// TODO(albin): Same comment as the above.
+		job = job.WithRequestedPriority(uint32(dbJob.Priority))
+	}
+	return job, nil
 }
 
 // schedulerRunFromDatabaseRun creates a new scheduler job run from a database job run
diff --git a/internal/scheduler/leader.go b/internal/scheduler/leader.go
index 714cf243..813b8f72 100644
--- a/internal/scheduler/leader.go
+++ b/internal/scheduler/leader.go
@@ -67,6 +67,10 @@ func NewStandaloneLeaderController() *StandaloneLeaderController {
 	}
 }
 
+func (lc *StandaloneLeaderController) SetToken(token LeaderToken) {
+	lc.token = token
+}
+
 func (lc *StandaloneLeaderController) GetToken() LeaderToken {
 	return lc.token
 }
@@ -89,7 +93,7 @@ func (lc *StandaloneLeaderController) Run(ctx *armadacontext.Context) error {
 	return nil
 }
 
-// LeaseListener allows clients to listen for lease events.
+// LeaseListener allows clients to listen for lease eventSequences.
 type LeaseListener interface {
 	// Called when the client has started leading.
 	onStartedLeading(*armadacontext.Context)
diff --git a/internal/scheduler/metrics.go b/internal/scheduler/metrics.go
index 3af04064..0d1ac99b 100644
--- a/internal/scheduler/metrics.go
+++ b/internal/scheduler/metrics.go
@@ -174,16 +174,29 @@ func (c *MetricsCollector) updateQueueMetrics(ctx *armadacontext.Context) ([]pro
 
 		var recorder *commonmetrics.JobMetricsRecorder
 		var timeInState time.Duration
-		if job.Queued() {
+		if job.InTerminalState() {
+			// Jobs in a terminal state should have been removed from the jobDb.
+			ctx.Warnf("job %s in jobDb is in a terminal state: %s", job.Id(), job)
+			continue
+		} else if job.Queued() {
+			if run := job.LatestRun(); run != nil && !run.InTerminalState() {
+				ctx.Warnf("job %s is marked as queued but has active runs: %s", job.Id(), job)
+				continue
+			}
 			recorder = qs.queuedJobRecorder
 			timeInState = currentTime.Sub(time.Unix(0, job.Created()))
 			queuedJobsCount[job.Queue()]++
-		} else if job.HasRuns() {
+		} else {
 			run := job.LatestRun()
-			timeInState = currentTime.Sub(time.Unix(0, run.Created()))
+			if run == nil {
+				ctx.Warnf("job %s is active and not marked as queued, but has no runs associated with it: %s", job.Id(), job)
+				continue
+			} else if run.InTerminalState() {
+				ctx.Warnf("job %s is active and not marked as queued, but its most recent run is in a terminal state: %s", job.Id(), job)
+				continue
+			}
 			recorder = qs.runningJobRecorder
-		} else {
-			ctx.Warnf("Job %s is marked as leased but has no runs", job.Id())
+			timeInState = currentTime.Sub(time.Unix(0, run.Created()))
 		}
 		recorder.RecordJobRuntime(pool, priorityClass, timeInState)
 		recorder.RecordResources(pool, priorityClass, jobResources)
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 2a14c296..7750a7a1 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -14,7 +14,6 @@ import (
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	"github.com/armadaproject/armada/internal/common/logging"
-	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/kubernetesobjects/affinity"
@@ -27,10 +26,10 @@ import (
 // It periodically performs the following cycle:
 // 1. Update state from postgres (via the jobRepository).
 // 2. Determine if leader and exit if not.
-// 3. Generate any necessary events resulting from the state update.
+// 3. Generate any necessary eventSequences resulting from the state update.
 // 4. Expire any jobs assigned to clusters that have timed out.
 // 5. Schedule jobs.
-// 6. Publish any Armada events resulting from the scheduling cycle.
+// 6. Publish any Armada eventSequences resulting from the scheduling cycle.
 type Scheduler struct {
 	// Provides job updates from Postgres.
 	jobRepository database.JobRepository
@@ -74,6 +73,9 @@ type Scheduler struct {
 	metrics *SchedulerMetrics
 	// New scheduler metrics due to replace the above.
 	schedulerMetrics *metrics.Metrics
+	// If true, enable scheduler assertions.
+	// In particular, assert that the jobDb is in a valid state at the end of each cycle.
+	enableAssertions bool
 }
 
 func NewScheduler(
@@ -114,6 +116,10 @@ func NewScheduler(
 	}, nil
 }
 
+func (s *Scheduler) EnableAssertions() {
+	s.enableAssertions = true
+}
+
 // Run enters the scheduling loop, which will continue until ctx is cancelled.
 func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 	ctx.Infof("starting scheduler with cycle time %s", s.cyclePeriod)
@@ -161,7 +167,6 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 			// and we must invalidate the held leader token to trigger flushing Pulsar at the next cycle.
 			//
 			// TODO: Once the Pulsar client supports transactions, we can guarantee consistency even in case of errors.
-
 			shouldSchedule := s.clock.Now().Sub(s.previousSchedulingRoundEnd) > s.schedulePeriod
 
 			result, err := s.cycle(ctx, fullUpdate, leaderToken, shouldSchedule)
@@ -192,15 +197,37 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 	}
 }
 
-// cycle is a single iteration of the main scheduling loop.
-// If updateAll is true, we generate events from all jobs in the jobDb.
-// Otherwise, we only generate events from jobs updated since the last cycle.
+// cycle runs one iteration of the scheduling loop.
+//
+// Each iteration of the scheduler is functionally equivalent to the following:
+//  1. Load the entire scheduler state, consisting of jobs, runs, and errors, from the persistent schedulerDb.
+//  2. Generate job state transitions, which we can be seen as performing a mapping
+//     (job_0, runs_0) -> (job_0', runs_0')
+//     ...
+//     (job_n, runs_n) -> (job_n', runs_n'),
+//     where runs_i is the set of jobs associated with job_i, and n is the number of jobs.
+//     These mappings come in two categories:
+//     - Triggered by an external event, e.g., jobs with a successful run should be marked as successful.
+//     - Triggered by a scheduling decision, e.g., a scheduled job should have an additional run associated with it.
+//  3. Generate eventSequences encoding these state transition and publish these to Pulsar.
+//  4. Wait for these Pulsar eventSequences to be persisted to the schedulerDb.
+//
+// For performance reasons, the actual cycle differs from the above in three ways:
+//   - In each cycle, we only load jobs, runs, and errors that have changed since the last cycle.
+//     For unchanged jobs, runs, and errors, we rely on an in-memory cache maintained between cycles, i.e., the jobDb.
+//   - Similarly, we only perform job state transitions for jobs with changed jobs and runs.
+//     This is unless updateAll is true, in which case we perform state transitions for all jobs.
+//     This is necessary for the first cycle after a leader failOver.
+//   - Instead of waiting for eventSequences to be persisted at the end of each cycle, the jobDb is updated directly.
+//     This means we can start the next cycle immediately after one cycle finishes.
+//     As state transitions are persisted and read back from the schedulerDb over later cycles,
+//     there is no change to the jobDb, since the correct changes have already been made.
 func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToken LeaderToken, shouldSchedule bool) (SchedulerResult, error) {
 	// TODO: Consider returning a slice of these instead.
 	overallSchedulerResult := SchedulerResult{}
 
 	// Update job state.
-	updatedJobs, jsts, jobRepoRunErrorsByRunId, err := s.syncState(ctx)
+	updatedJobs, jsts, err := s.syncState(ctx)
 	if err != nil {
 		return overallSchedulerResult, err
 	}
@@ -214,11 +241,6 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 		s.schedulerMetrics.Enable()
 	}
 
-	// Update metrics.
-	if err := s.schedulerMetrics.UpdateMany(ctx, jsts, jobRepoRunErrorsByRunId); err != nil {
-		return overallSchedulerResult, err
-	}
-
 	// If we've been asked to generate messages for all jobs, do so.
 	// Otherwise, generate messages only for jobs updated this cycle.
 	txn := s.jobDb.WriteTxn()
@@ -227,7 +249,28 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 		updatedJobs = txn.GetAll()
 	}
 
-	// Generate any events that came out of synchronising the db state.
+	// Load error messages for any failed runs.
+	// TODO(albin): An unbounded number of job runs may fail between subsequent cycles.
+	//              E.g., if 1M runs fail and each one generates a 1Mb error message, we'd need to load 1Tb of errors.
+	//              If so, the scheduler would not be able to progress until a human manually deleted those errors.
+	failedRunIds := make([]uuid.UUID, 0, len(updatedJobs))
+	for _, job := range updatedJobs {
+		run := job.LatestRun()
+		if run != nil && run.Failed() {
+			failedRunIds = append(failedRunIds, run.Id())
+		}
+	}
+	jobRepoRunErrorsByRunId, err := s.jobRepository.FetchJobRunErrors(ctx, failedRunIds)
+	if err != nil {
+		return overallSchedulerResult, err
+	}
+
+	// Update metrics.
+	if err := s.schedulerMetrics.UpdateMany(ctx, jsts, jobRepoRunErrorsByRunId); err != nil {
+		return overallSchedulerResult, err
+	}
+
+	// Generate any eventSequences that came out of synchronising the db state.
 	events, err := s.generateUpdateMessages(ctx, txn, updatedJobs, jobRepoRunErrorsByRunId)
 	if err != nil {
 		return overallSchedulerResult, err
@@ -274,7 +317,14 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 	if err = s.publisher.PublishMessages(ctx, events, isLeader); err != nil {
 		return overallSchedulerResult, err
 	}
-	ctx.Infof("published %d events to pulsar in %s", len(events), s.clock.Since(start))
+	ctx.Infof("published %d eventSequences to pulsar in %s", len(events), s.clock.Since(start))
+
+	// Optionally assert that the jobDb is in a valid state and then commit.
+	if s.enableAssertions {
+		if err := txn.Assert(false); err != nil {
+			return overallSchedulerResult, err
+		}
+	}
 	txn.Commit()
 
 	// Update metrics based on overallSchedulerResult.
@@ -312,27 +362,20 @@ func (s *Scheduler) updateMetricsFromSchedulerResult(ctx *armadacontext.Context,
 }
 
 // syncState updates jobs in jobDb to match state in postgres and returns all updated jobs.
-func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb.JobStateTransitions, map[uuid.UUID]*armadaevents.Error, error) {
+func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb.JobStateTransitions, error) {
 	txn := s.jobDb.WriteTxn()
 	defer txn.Abort()
 
 	// Load new and updated jobs from the jobRepo.
 	updatedJobs, updatedRuns, err := s.jobRepository.FetchJobUpdates(ctx, s.jobsSerial, s.runsSerial)
 	if err != nil {
-		return nil, nil, nil, err
-	}
-
-	// Load any error associated with updated runs.
-	jobRunIds := util.Map(updatedRuns, func(jobRepoRun database.Run) uuid.UUID { return jobRepoRun.RunID })
-	jobRepoRunErrorsByRunId, err := s.jobRepository.FetchJobRunErrors(ctx, jobRunIds)
-	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 
 	// Reconcile any differences between the updated jobs and runs.
 	jsts, err := s.jobDb.ReconcileDifferences(txn, updatedJobs, updatedRuns)
 	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 
 	// Upsert updated jobs (including associated runs).
@@ -345,7 +388,7 @@ func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb
 		}
 	}
 	if err := txn.Upsert(jobDbJobs); err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 
 	// Delete jobs in a terminal state.
@@ -356,7 +399,7 @@ func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb
 		}
 	}
 	if err := txn.BatchDelete(idsOfJobsToDelete); err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 
 	txn.Commit()
@@ -369,7 +412,7 @@ func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb
 		s.runsSerial = updatedRuns[len(updatedRuns)-1].Serial
 	}
 
-	return jobDbJobs, jsts, jobRepoRunErrorsByRunId, nil
+	return jobDbJobs, jsts, nil
 }
 
 func (s *Scheduler) createSchedulingInfoWithNodeAntiAffinityForAttemptedRuns(job *jobdb.Job) (*schedulerobjects.JobSchedulingInfo, error) {
@@ -418,7 +461,7 @@ func EventsFromSchedulerResult(result *SchedulerResult, time time.Time) ([]*arma
 	if err != nil {
 		return nil, err
 	}
-	eventSequences, err = AppendEventSequencesFromScheduledJobs(eventSequences, ScheduledJobsFromSchedulerResult[*jobdb.Job](result), result.AdditionalAnnotationsByJobId, time)
+	eventSequences, err = AppendEventSequencesFromScheduledJobs(eventSequences, ScheduledJobsFromSchedulerResult[*jobdb.Job](result), result.AdditionalAnnotationsByJobId)
 	if err != nil {
 		return nil, err
 	}
@@ -437,7 +480,7 @@ func AppendEventSequencesFromPreemptedJobs(eventSequences []*armadaevents.EventS
 		}
 		run := job.LatestRun()
 		if run == nil {
-			return nil, errors.Errorf("attempting to generate preempted events for job %s with no associated runs", job.Id())
+			return nil, errors.Errorf("attempting to generate preempted eventSequences for job %s with no associated runs", job.Id())
 		}
 		eventSequences = append(eventSequences, &armadaevents.EventSequence{
 			Queue:      job.Queue(),
@@ -491,7 +534,7 @@ func AppendEventSequencesFromPreemptedJobs(eventSequences []*armadaevents.EventS
 	return eventSequences, nil
 }
 
-func AppendEventSequencesFromScheduledJobs(eventSequences []*armadaevents.EventSequence, jobs []*jobdb.Job, additionalAnnotationsByJobId map[string]map[string]string, time time.Time) ([]*armadaevents.EventSequence, error) {
+func AppendEventSequencesFromScheduledJobs(eventSequences []*armadaevents.EventSequence, jobs []*jobdb.Job, additionalAnnotationsByJobId map[string]map[string]string) ([]*armadaevents.EventSequence, error) {
 	for _, job := range jobs {
 		jobId, err := armadaevents.ProtoUuidFromUlidString(job.Id())
 		if err != nil {
@@ -503,15 +546,16 @@ func AppendEventSequencesFromScheduledJobs(eventSequences []*armadaevents.EventS
 		}
 		run := job.LatestRun()
 		if run == nil {
-			return nil, errors.Errorf("attempting to generate lease events for job %s with no associated runs", job.Id())
+			return nil, errors.Errorf("attempting to generate lease eventSequences for job %s with no associated runs", job.Id())
 		}
+		runCreationTime := time.Unix(0, job.ActiveRunTimestamp())
 		scheduledAtPriority, hasScheduledAtPriority := job.GetScheduledAtPriority()
 		eventSequences = append(eventSequences, &armadaevents.EventSequence{
 			Queue:      job.Queue(),
 			JobSetName: job.Jobset(), // TODO: Rename to JobSet.
 			Events: []*armadaevents.EventSequence_Event{
 				{
-					Created: &time,
+					Created: &runCreationTime,
 					Event: &armadaevents.EventSequence_Event_JobRunLeased{
 						JobRunLeased: &armadaevents.JobRunLeased{
 							RunId:      armadaevents.ProtoUuidFromUuid(run.Id()),
@@ -559,10 +603,10 @@ func AppendEventSequencesFromUnschedulableJobs(eventSequences []*armadaevents.Ev
 	return eventSequences, nil
 }
 
-// generateUpdateMessages generates EventSequences representing the state changes on updated jobs
-// If there are no state changes then an empty slice will be returned
-func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, txn *jobdb.Txn, updatedJobs []*jobdb.Job, jobRunErrors map[uuid.UUID]*armadaevents.Error) ([]*armadaevents.EventSequence, error) {
-	// Generate any events that came out of synchronising the db state.
+// generateUpdateMessages generates EventSequences representing the state changes on updated jobs.
+// If there are no state changes then an empty slice will be returned.
+func (s *Scheduler) generateUpdateMessages(_ *armadacontext.Context, txn *jobdb.Txn, updatedJobs []*jobdb.Job, jobRunErrors map[uuid.UUID]*armadaevents.Error) ([]*armadaevents.EventSequence, error) {
+	// Generate any eventSequences that came out of synchronising the db state.
 	var events []*armadaevents.EventSequence
 	for _, job := range updatedJobs {
 		jobEvents, err := s.generateUpdateMessagesFromJob(job, jobRunErrors, txn)
@@ -576,8 +620,8 @@ func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, txn *jobd
 	return events, nil
 }
 
-// generateUpdateMessages generates EventSequence representing the state change on a single jobs
-// If there are no state changes then nil will be returned
+// generateUpdateMessages generates an EventSequence representing the state changes for a single job.
+// If there are no state changes it returns nil.
 func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors map[uuid.UUID]*armadaevents.Error, txn *jobdb.Txn) (*armadaevents.EventSequence, error) {
 	var events []*armadaevents.EventSequence_Event
 
@@ -594,9 +638,9 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors m
 	// Has the job been requested cancelled. If so, cancel the job
 	if job.CancelRequested() {
 		for _, run := range job.AllRuns() {
-			job = job.WithUpdatedRun(run.WithCancelled(true))
+			job = job.WithUpdatedRun(run.WithRunning(false).WithCancelled(true))
 		}
-		job = job.WithCancelled(true).WithQueued(false)
+		job = job.WithQueued(false).WithCancelled(true)
 		cancel := &armadaevents.EventSequence_Event{
 			Created: s.now(),
 			Event: &armadaevents.EventSequence_Event_CancelledJob{
@@ -606,9 +650,9 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors m
 		events = append(events, cancel)
 	} else if job.CancelByJobsetRequested() {
 		for _, run := range job.AllRuns() {
-			job = job.WithUpdatedRun(run.WithCancelled(true))
+			job = job.WithUpdatedRun(run.WithRunning(false).WithCancelled(true))
 		}
-		job = job.WithCancelled(true).WithQueued(false)
+		job = job.WithQueued(false).WithCancelled(true)
 		cancelRequest := &armadaevents.EventSequence_Event{
 			Created: s.now(),
 			Event: &armadaevents.EventSequence_Event_CancelJob{
@@ -672,6 +716,13 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors m
 				events = append(events, requeueJobEvent)
 			} else {
 				runError := jobRunErrors[lastRun.Id()]
+				if runError == nil {
+					return nil, errors.Errorf(
+						"no run error found for run %s (job id = %s), this must mean we're out of sync with the database",
+						lastRun.Id().String(), job.Id(),
+					)
+				}
+
 				job = job.WithFailed(true).WithQueued(false)
 				if lastRun.Returned() {
 					errorMessage := fmt.Sprintf("Maximum number of attempts (%d) reached - this job will no longer be retried", s.maxAttemptedRuns)
@@ -696,12 +747,6 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors m
 						},
 					}
 				}
-				if runError == nil {
-					panic(
-						fmt.Sprintf("No run error found for run %s (job id = %s), this must mean we're out of sync with the database",
-							lastRun.Id().String(), job.Id()),
-					)
-				}
 				jobErrors := &armadaevents.EventSequence_Event{
 					Created: s.now(),
 					Event: &armadaevents.EventSequence_Event_JobErrors{
@@ -898,7 +943,8 @@ func (s *Scheduler) initialise(ctx *armadacontext.Context) error {
 		case <-ctx.Done():
 			return nil
 		default:
-			if _, _, _, err := s.syncState(ctx); err != nil {
+			// TODO(albin): This doesn't need to be separate; we'd anyway load everything in the first scheduling cycle.
+			if _, _, err := s.syncState(ctx); err != nil {
 				logging.WithStacktrace(ctx, err).Error("failed to initialise; trying again in 1 second")
 				time.Sleep(1 * time.Second)
 			} else {
diff --git a/internal/scheduler/schedulerapp.go b/internal/scheduler/schedulerapp.go
index 77fa1445..3d55980a 100644
--- a/internal/scheduler/schedulerapp.go
+++ b/internal/scheduler/schedulerapp.go
@@ -230,6 +230,9 @@ func Run(config schedulerconfig.Configuration) error {
 	if err != nil {
 		return errors.WithMessage(err, "error creating scheduler")
 	}
+	if config.Scheduling.EnableAssertions {
+		scheduler.EnableAssertions()
+	}
 	services = append(services, func() error { return scheduler.Run(ctx) })
 
 	// ////////////////////////////////////////////////////////////////////////
diff --git a/internal/scheduler/scheduling_algo.go b/internal/scheduler/scheduling_algo.go
index 536f7b6c..53a36670 100644
--- a/internal/scheduler/scheduling_algo.go
+++ b/internal/scheduler/scheduling_algo.go
@@ -34,7 +34,7 @@ import (
 type SchedulingAlgo interface {
 	// Schedule should assign jobs to nodes.
 	// Any jobs that are scheduled should be marked as such in the JobDb using the transaction provided.
-	Schedule(ctx *armadacontext.Context, txn *jobdb.Txn) (*SchedulerResult, error)
+	Schedule(*armadacontext.Context, *jobdb.Txn) (*SchedulerResult, error)
 }
 
 // FairSchedulingAlgo is a SchedulingAlgo based on PreemptingQueueScheduler.
diff --git a/internal/scheduler/simulator/runner.go b/internal/scheduler/simulator/runner.go
index 3f8b145e..d6988489 100644
--- a/internal/scheduler/simulator/runner.go
+++ b/internal/scheduler/simulator/runner.go
@@ -10,40 +10,9 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
-	"github.com/armadaproject/armada/internal/common/armadacontext"
 	commonconfig "github.com/armadaproject/armada/internal/common/config"
 )
 
-func Simulate(ctx *armadacontext.Context, clusterSpecsPattern, workloadSpecsPattern, schedulingConfigsPattern string) error {
-	clusterSpecs, err := ClusterSpecsFromPattern(clusterSpecsPattern)
-	if err != nil {
-		return err
-	}
-	workloadSpecs, err := WorkloadsFromPattern(workloadSpecsPattern)
-	if err != nil {
-		return err
-	}
-	schedulingConfigs, err := SchedulingConfigsFromPattern(schedulingConfigsPattern)
-	if err != nil {
-		return err
-	}
-	g, ctx := armadacontext.ErrGroup(ctx)
-	for _, clusterSpec := range clusterSpecs {
-		for _, workloadSpec := range workloadSpecs {
-			for _, schedulingConfig := range schedulingConfigs {
-				s, err := NewSimulator(clusterSpec, workloadSpec, schedulingConfig)
-				if err != nil {
-					return err
-				}
-				g.Go(func() error {
-					return s.Run(ctx)
-				})
-			}
-		}
-	}
-	return g.Wait()
-}
-
 func SchedulingConfigsByFilePathFromPattern(pattern string) (map[string]configuration.SchedulingConfig, error) {
 	filePaths, err := zglob.Glob(pattern)
 	if err != nil {
diff --git a/internal/scheduler/simulator/simulator.go b/internal/scheduler/simulator/simulator.go
index 8dad4954..1ada3591 100644
--- a/internal/scheduler/simulator/simulator.go
+++ b/internal/scheduler/simulator/simulator.go
@@ -128,6 +128,7 @@ func NewSimulator(clusterSpec *ClusterSpec, workloadSpec *WorkloadSpec, scheduli
 		limiterByQueue: make(map[string]*rate.Limiter),
 		rand:           rand.New(rand.NewSource(randomSeed)),
 	}
+	jobDb.SetClock(s)
 	s.limiter.SetBurstAt(s.time, schedulingConfig.MaximumSchedulingBurst)
 	if err := s.setupClusters(); err != nil {
 		return nil, err
@@ -138,6 +139,14 @@ func NewSimulator(clusterSpec *ClusterSpec, workloadSpec *WorkloadSpec, scheduli
 	return s, nil
 }
 
+func (s *Simulator) Now() time.Time {
+	return s.time
+}
+
+func (s *Simulator) Since(t time.Time) time.Duration {
+	return s.Now().Sub(t)
+}
+
 // Run runs the scheduler until all jobs have finished successfully.
 func (s *Simulator) Run(ctx *armadacontext.Context) error {
 	defer func() {
@@ -556,8 +565,7 @@ func (s *Simulator) handleScheduleEvent(ctx *armadacontext.Context) error {
 			if err != nil {
 				return err
 			}
-
-			eventSequences, err = scheduler.AppendEventSequencesFromScheduledJobs(eventSequences, scheduledJobs, make(map[string]map[string]string), s.time)
+			eventSequences, err = scheduler.AppendEventSequencesFromScheduledJobs(eventSequences, scheduledJobs, make(map[string]map[string]string))
 			if err != nil {
 				return err
 			}
@@ -566,6 +574,14 @@ func (s *Simulator) handleScheduleEvent(ctx *armadacontext.Context) error {
 				return err
 			}
 
+			// Update event timestamps to be consistent with simulated time.
+			t := s.time
+			for _, eventSequence := range eventSequences {
+				for _, event := range eventSequence.Events {
+					event.Created = &t
+				}
+			}
+
 			// If nothing changed, we're in steady state and can safely skip scheduling until something external has changed.
 			// Do this only if a non-zero amount of time has passed.
 			if !s.time.Equal(time.Time{}) && len(result.ScheduledJobs) == 0 && len(result.PreemptedJobs) == 0 && len(result.FailedJobs) == 0 {
diff --git a/internal/scheduleringester/dbops.go b/internal/scheduleringester/dbops.go
index 1310b66c..236b6b76 100644
--- a/internal/scheduleringester/dbops.go
+++ b/internal/scheduleringester/dbops.go
@@ -43,8 +43,8 @@ type JobSetKey struct {
 }
 
 type JobRunDetails struct {
-	queue string
-	dbRun *schedulerdb.Run
+	Queue string
+	DbRun *schedulerdb.Run
 }
 
 type JobQueuedStateUpdate struct {
@@ -115,9 +115,8 @@ func discardNilOps(ops []DbOperation) []DbOperation {
 	return rv
 }
 
-type InsertJobs map[string]*schedulerdb.Job
-
 type (
+	InsertJobs                 map[string]*schedulerdb.Job
 	InsertRuns                 map[uuid.UUID]*JobRunDetails
 	UpdateJobSetPriorities     map[JobSetKey]int64
 	MarkJobSetsCancelRequested map[JobSetKey]*JobSetCancelAction
@@ -292,13 +291,13 @@ func (a InsertRuns) CanBeAppliedBefore(b DbOperation) bool {
 	switch op := b.(type) {
 	case JobSetOperation:
 		for _, run := range a {
-			if op.AffectsJobSet(run.queue, run.dbRun.JobSet) {
+			if op.AffectsJobSet(run.Queue, run.DbRun.JobSet) {
 				return false
 			}
 		}
 	case InsertJobs:
 		for _, run := range a {
-			if _, ok := op[run.dbRun.JobID]; ok {
+			if _, ok := op[run.DbRun.JobID]; ok {
 				return false
 			}
 		}
@@ -393,7 +392,7 @@ func definesJobInSet[M ~map[JobSetKey]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[JobSetKey{queue: run.queue, jobSet: run.dbRun.JobSet}]; ok {
+			if _, ok := a[JobSetKey{queue: run.Queue, jobSet: run.DbRun.JobSet}]; ok {
 				return true
 			}
 		}
@@ -419,7 +418,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.dbRun.RunID]; ok {
+			if _, ok := a[run.DbRun.RunID]; ok {
 				return true
 			}
 		}
@@ -432,7 +431,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.dbRun.JobID]; ok {
+			if _, ok := a[run.DbRun.JobID]; ok {
 				return true
 			}
 		}
diff --git a/internal/scheduleringester/ingester.go b/internal/scheduleringester/ingester.go
index 3560787d..cc2fbe66 100644
--- a/internal/scheduleringester/ingester.go
+++ b/internal/scheduleringester/ingester.go
@@ -9,7 +9,6 @@ import (
 
 	"github.com/armadaproject/armada/internal/common/app"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
-	"github.com/armadaproject/armada/internal/common/compress"
 	"github.com/armadaproject/armada/internal/common/database"
 	"github.com/armadaproject/armada/internal/common/ingest"
 	"github.com/armadaproject/armada/internal/common/ingest/metrics"
@@ -19,23 +18,22 @@ import (
 	"github.com/armadaproject/armada/internal/common/serve"
 )
 
-// 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) {
+// Run will create a pipeline that will take Armada event messages from Pulsar and update the schedulerDb.
+// This pipeline will run until a SIGTERM is received.
+func Run(config Configuration) error {
 	svcMetrics := metrics.NewMetrics(metrics.ArmadaEventIngesterMetricsPrefix + "armada_scheduler_ingester_")
 
-	log.Infof("Opening connection pool to postgres")
+	log.Infof("opening connection pool to postgres")
 	db, err := database.OpenPgxPool(config.Postgres)
 	if err != nil {
 		panic(errors.WithMessage(err, "Error opening connection to postgres"))
 	}
 	schedulerDb := NewSchedulerDb(db, svcMetrics, 100*time.Millisecond, 60*time.Second, 5*time.Second)
 
-	compressor, err := compress.NewZlibCompressor(1024)
+	converter, err := NewInstructionConverter(svcMetrics, config.PriorityClasses)
 	if err != nil {
-		panic(errors.WithMessage(err, "Error creating  compressor"))
+		return err
 	}
-	converter := NewInstructionConverter(svcMetrics, config.PriorityClasses, compressor)
 
 	// Expose profiling endpoints if enabled.
 	pprofServer := profiling.SetupPprofHttpServer(config.PprofPort)
@@ -46,7 +44,7 @@ func Run(config Configuration) {
 		}
 	}()
 
-	ingester := ingest.NewFilteredMsgIngestionPipeline(
+	ingester := ingest.NewFilteredMsgIngestionPipeline[*DbOperationsWithMessageIds](
 		config.Pulsar,
 		config.SubscriptionName,
 		config.BatchSize,
@@ -59,6 +57,7 @@ func Run(config Configuration) {
 		svcMetrics,
 	)
 	if err := ingester.Run(app.CreateContextWithShutdown()); err != nil {
-		panic(errors.WithMessage(err, "Error running ingestion pipeline"))
+		return errors.WithMessage(err, "error running ingestion pipeline")
 	}
+	return nil
 }
diff --git a/internal/scheduleringester/instructions.go b/internal/scheduleringester/instructions.go
index f2d751c2..e10789dc 100644
--- a/internal/scheduleringester/instructions.go
+++ b/internal/scheduleringester/instructions.go
@@ -37,13 +37,16 @@ type InstructionConverter struct {
 func NewInstructionConverter(
 	metrics *metrics.Metrics,
 	priorityClasses map[string]types.PriorityClass,
-	compressor compress.Compressor,
-) ingest.InstructionConverter[*DbOperationsWithMessageIds] {
+) (*InstructionConverter, error) {
+	compressor, err := compress.NewZlibCompressor(1024)
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed to create compressor")
+	}
 	return &InstructionConverter{
 		metrics:         metrics,
 		priorityClasses: priorityClasses,
 		compressor:      compressor,
-	}
+	}, nil
 }
 
 func (c *InstructionConverter) Convert(_ *armadacontext.Context, sequencesWithIds *ingest.EventSequencesWithIds) *DbOperationsWithMessageIds {
@@ -102,7 +105,7 @@ func (c *InstructionConverter) dbOperationsFromEventSequence(es *armadaevents.Ev
 		case *armadaevents.EventSequence_Event_JobRequeued:
 			operationsFromEvent, err = c.handleJobRequeued(event.GetJobRequeued())
 		case *armadaevents.EventSequence_Event_PartitionMarker:
-			operationsFromEvent, err = c.handlePartitionMarker(event.GetPartitionMarker(), *event.Created)
+			operationsFromEvent, err = c.handlePartitionMarker(event.GetPartitionMarker(), eventTime)
 		case *armadaevents.EventSequence_Event_JobRunPreempted:
 			operationsFromEvent, err = c.handleJobRunPreempted(event.GetJobRunPreempted(), eventTime)
 		case *armadaevents.EventSequence_Event_JobRunAssigned:
@@ -179,7 +182,7 @@ func (c *InstructionConverter) handleSubmitJob(job *armadaevents.SubmitJob, subm
 	}}}, nil
 }
 
-func (c *InstructionConverter) handleJobRunLeased(jobRunLeased *armadaevents.JobRunLeased, leaseTime time.Time, meta eventSequenceCommon) ([]DbOperation, error) {
+func (c *InstructionConverter) handleJobRunLeased(jobRunLeased *armadaevents.JobRunLeased, eventTime time.Time, meta eventSequenceCommon) ([]DbOperation, error) {
 	runId := armadaevents.UuidFromProtoUuid(jobRunLeased.GetRunId())
 	jobId, err := armadaevents.UlidStringFromProtoUuid(jobRunLeased.GetJobId())
 	if err != nil {
@@ -191,15 +194,16 @@ func (c *InstructionConverter) handleJobRunLeased(jobRunLeased *armadaevents.Job
 	}
 	return []DbOperation{
 		InsertRuns{runId: &JobRunDetails{
-			queue: meta.queue,
-			dbRun: &schedulerdb.Run{
+			Queue: meta.queue,
+			DbRun: &schedulerdb.Run{
 				RunID:               runId,
 				JobID:               jobId,
+				Created:             eventTime.UnixNano(),
 				JobSet:              meta.jobset,
 				Executor:            jobRunLeased.GetExecutorId(),
 				Node:                jobRunLeased.GetNodeId(),
 				ScheduledAtPriority: scheduledAtPriority,
-				LeasedTimestamp:     &leaseTime,
+				LeasedTimestamp:     &eventTime,
 			},
 		}},
 		UpdateJobQueuedState{jobId: &JobQueuedStateUpdate{
@@ -259,7 +263,7 @@ func (c *InstructionConverter) handleJobRunErrors(jobRunErrors *armadaevents.Job
 	insertJobRunErrors := make(InsertJobRunErrors)
 	markRunsFailed := make(MarkRunsFailed)
 	for _, runError := range jobRunErrors.GetErrors() {
-		// There should only be one terminal error
+		// There should only be one terminal error.
 		if runError.GetTerminal() {
 			bytes, err := protoutil.MarshallAndCompress(runError, c.compressor)
 			if err != nil {
diff --git a/internal/scheduleringester/schedulerdb.go b/internal/scheduleringester/schedulerdb.go
index 6d4d5854..70c09833 100644
--- a/internal/scheduleringester/schedulerdb.go
+++ b/internal/scheduleringester/schedulerdb.go
@@ -33,7 +33,7 @@ func NewSchedulerDb(
 	initialBackOff time.Duration,
 	maxBackOff time.Duration,
 	lockTimeout time.Duration,
-) ingest.Sink[*DbOperationsWithMessageIds] {
+) *SchedulerDb {
 	return &SchedulerDb{
 		db:             db,
 		metrics:        metrics,
@@ -105,7 +105,7 @@ func (s *SchedulerDb) WriteDbOp(ctx *armadacontext.Context, tx pgx.Tx, op DbOper
 		records := make([]any, len(o))
 		i := 0
 		for _, v := range o {
-			records[i] = *v.dbRun
+			records[i] = *v.DbRun
 			i++
 		}
 		err := database.Upsert(ctx, tx, "runs", records)
diff --git a/pkg/armadaevents/uuid.go b/pkg/armadaevents/uuid.go
index 28c21c86..edf10ec9 100644
--- a/pkg/armadaevents/uuid.go
+++ b/pkg/armadaevents/uuid.go
@@ -72,6 +72,16 @@ func UuidStringFromProtoUuid(id *Uuid) (string, error) {
 	return strings.ToLower(UuidFromProtoUuid(id).String()), nil
 }
 
+// MustProtoUuidFromUlidString parses a ULID string into a proto UUID and returns it.
+func MustProtoUuidFromUlidString(ulidString string) *Uuid {
+	id, err := ulid.Parse(ulidString)
+	if err != nil {
+		err = errors.WithStack(err)
+		panic(err)
+	}
+	return ProtoUuidFromUlid(id)
+}
+
 // ProtoUuidFromUlidString parses a ULID string into a proto UUID and returns it.
 func ProtoUuidFromUlidString(ulidString string) (*Uuid, error) {
 	id, err := ulid.Parse(ulidString)


Test patch:
diff --git a/internal/common/database/db_testutil.go b/internal/common/database/db_testutil.go
index 416b348d..9b228cb1 100644
--- a/internal/common/database/db_testutil.go
+++ b/internal/common/database/db_testutil.go
@@ -19,7 +19,7 @@ import (
 func WithTestDb(migrations []Migration, action func(db *pgxpool.Pool) error) error {
 	ctx := armadacontext.Background()
 
-	// Connect and create a dedicated database for the test
+	// Connect and create a dedicated database for the test.
 	dbName := "test_" + util.NewULID()
 	connectionString := "host=localhost port=5432 user=postgres password=psw sslmode=disable"
 	db, err := pgx.Connect(ctx, connectionString)
@@ -33,32 +33,29 @@ func WithTestDb(migrations []Migration, action func(db *pgxpool.Pool) error) err
 		return errors.WithStack(err)
 	}
 
-	// Connect again: this time to the database we just created.  This is the databse we use for tests
-	testDbPool, err := pgxpool.New(ctx, connectionString+" dbname="+dbName)
-	if err != nil {
-		return errors.WithStack(err)
-	}
-
+	// Disconnect all db users and drop the database we created at test completion.
 	defer func() {
-		// disconnect all db user before cleanup
-		_, err = db.Exec(ctx,
+		if _, err := db.Exec(ctx,
 			`SELECT pg_terminate_backend(pg_stat_activity.pid)
-			 FROM pg_stat_activity WHERE pg_stat_activity.datname = '`+dbName+`';`)
-		if err != nil {
-			fmt.Println("Failed to disconnect users")
+			 FROM pg_stat_activity WHERE pg_stat_activity.datname = '`+dbName+`';`,
+		); err != nil {
+			fmt.Println("failed to disconnect users:", err)
 		}
-
-		_, err = db.Exec(ctx, "DROP DATABASE "+dbName)
-		if err != nil {
-			fmt.Println("Failed to drop database")
+		if _, err := db.Exec(ctx, "DROP DATABASE "+dbName); err != nil {
+			fmt.Println("failed to drop database:", err)
 		}
 	}()
 
-	err = UpdateDatabase(ctx, testDbPool, migrations)
+	// Connect again: this time to the database we just created. This is the database we use for tests.
+	testDbPool, err := pgxpool.New(ctx, connectionString+" dbname="+dbName)
 	if err != nil {
 		return errors.WithStack(err)
 	}
 
+	if err := UpdateDatabase(ctx, testDbPool, migrations); err != nil {
+		return errors.WithStack(err)
+	}
+
 	return action(testDbPool)
 }
 
diff --git a/internal/scheduler/jobdb/comparison_test.go b/internal/scheduler/jobdb/comparison_test.go
index 9b9cd81e..84735444 100644
--- a/internal/scheduler/jobdb/comparison_test.go
+++ b/internal/scheduler/jobdb/comparison_test.go
@@ -51,7 +51,7 @@ func TestJobPriorityComparer(t *testing.T) {
 		},
 		"Running jobs come before queued jobs": {
 			a:        &Job{id: "a", priority: 1},
-			b:        (&Job{id: "b", priority: 2}).WithNewRun("", "", "", 0),
+			b:        (&Job{id: "b", priority: 2, jobDb: NewJobDb(map[string]types.PriorityClass{"foo": {}}, "foo", 1)}).WithNewRun("", "", "", 0),
 			expected: 1,
 		},
 		"Running jobs are ordered third by runtime": {
diff --git a/internal/scheduler/jobdb/jobdb_test.go b/internal/scheduler/jobdb/jobdb_test.go
index a795ffc8..abd6497d 100644
--- a/internal/scheduler/jobdb/jobdb_test.go
+++ b/internal/scheduler/jobdb/jobdb_test.go
@@ -1251,6 +1251,7 @@ func TestJobDb_SchedulingKey(t *testing.T) {
 
 func newJob() *Job {
 	return &Job{
+		jobDb:             NewTestJobDb(),
 		id:                util.NewULID(),
 		queue:             "test-queue",
 		priority:          0,
diff --git a/internal/scheduler/preempting_queue_scheduler_test.go b/internal/scheduler/preempting_queue_scheduler_test.go
index 25012dc2..4152fa23 100644
--- a/internal/scheduler/preempting_queue_scheduler_test.go
+++ b/internal/scheduler/preempting_queue_scheduler_test.go
@@ -2041,8 +2041,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 					require.True(t, ok)
 					scheduledJobs = append(
 						scheduledJobs,
-						job.
-							WithQueuedVersion(job.QueuedVersion()+1).
+						job.WithQueuedVersion(job.QueuedVersion()+1).
 							WithQueued(false).
 							WithNewRun(node.Executor, node.Id, node.Name, priority),
 					)
diff --git a/internal/scheduler/publisher_test.go b/internal/scheduler/publisher_test.go
index 6ecb200d..6e4e693d 100644
--- a/internal/scheduler/publisher_test.go
+++ b/internal/scheduler/publisher_test.go
@@ -60,7 +60,7 @@ func TestPulsarPublisher_TestPublish(t *testing.T) {
 				},
 			},
 		},
-		"Return error if all events fail to publish": {
+		"Return error if all eventSequences fail to publish": {
 			amLeader:               true,
 			numSuccessfulPublishes: 0,
 			eventSequences: []*armadaevents.EventSequence{
@@ -71,7 +71,7 @@ func TestPulsarPublisher_TestPublish(t *testing.T) {
 			},
 			expectedError: true,
 		},
-		"Return error if some events fail to publish": {
+		"Return error if some eventSequences fail to publish": {
 			amLeader:               true,
 			numSuccessfulPublishes: 1,
 			eventSequences: []*armadaevents.EventSequence{
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 24733262..8b6856a2 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -7,23 +7,28 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/exp/slices"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
+	"github.com/armadaproject/armada/internal/common/ingest"
 	protoutil "github.com/armadaproject/armada/internal/common/proto"
 	"github.com/armadaproject/armada/internal/common/util"
 	schedulercontext "github.com/armadaproject/armada/internal/scheduler/context"
 	"github.com/armadaproject/armada/internal/scheduler/database"
+	schedulerdb "github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/interfaces"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/kubernetesobjects/affinity"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 	"github.com/armadaproject/armada/internal/scheduler/testfixtures"
+	"github.com/armadaproject/armada/internal/scheduleringester"
 	"github.com/armadaproject/armada/pkg/api"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 )
@@ -44,13 +49,15 @@ var (
 						Annotations: map[string]string{
 							configuration.FailFastAnnotation: "true",
 						},
+						Priority: int32(10),
 					},
 				},
 			},
 		},
 		Version: 1,
 	}
-	schedulingInfo = &schedulerobjects.JobSchedulingInfo{
+	failFastSchedulingInfoBytes = protoutil.MustMarshall(failFastSchedulingInfo)
+	schedulingInfo              = &schedulerobjects.JobSchedulingInfo{
 		AtMostOnce: true,
 		ObjectRequirements: []*schedulerobjects.ObjectRequirements{
 			{
@@ -77,8 +84,22 @@ var (
 		},
 		Version: 2,
 	}
-	updatedSchedulingInfoBytes = protoutil.MustMarshall(updatedSchedulingInfo)
-	schedulingInfoWithQueueTtl = &schedulerobjects.JobSchedulingInfo{
+	updatedSchedulingInfoBytes        = protoutil.MustMarshall(updatedSchedulingInfo)
+	schedulingInfoWithUpdatedPriority = &schedulerobjects.JobSchedulingInfo{
+		AtMostOnce: true,
+		ObjectRequirements: []*schedulerobjects.ObjectRequirements{
+			{
+				Requirements: &schedulerobjects.ObjectRequirements_PodRequirements{
+					PodRequirements: &schedulerobjects.PodRequirements{
+						Priority: int32(20),
+					},
+				},
+			},
+		},
+		Version: 2,
+	}
+	schedulingInfoWithUpdatedPriorityBytes = protoutil.MustMarshall(schedulingInfoWithUpdatedPriority)
+	schedulingInfoWithQueueTtl             = &schedulerobjects.JobSchedulingInfo{
 		AtMostOnce: true,
 		ObjectRequirements: []*schedulerobjects.ObjectRequirements{
 			{
@@ -93,7 +114,8 @@ var (
 		Version:         1,
 	}
 	schedulingInfoWithQueueTtlBytes = protoutil.MustMarshall(schedulingInfoWithQueueTtl)
-	schedulerMetrics                = NewSchedulerMetrics(configuration.SchedulerMetricsConfig{
+
+	schedulerMetrics = NewSchedulerMetrics(configuration.SchedulerMetricsConfig{
 		ScheduleCycleTimeHistogramSettings: configuration.HistogramConfig{
 			Start:  1,
 			Factor: 1.1,
@@ -114,11 +136,12 @@ var queuedJob = testfixtures.JobDb.NewJob(
 	uint32(10),
 	schedulingInfo,
 	true,
-	1,
+	0,
 	false,
 	false,
 	false,
-	1)
+	1,
+)
 
 var queuedJobWithExpiredTtl = testfixtures.JobDb.NewJob(
 	util.NewULID(),
@@ -127,11 +150,12 @@ var queuedJobWithExpiredTtl = testfixtures.JobDb.NewJob(
 	0,
 	schedulingInfoWithQueueTtl,
 	true,
-	1,
+	0,
 	false,
 	false,
 	false,
-	1)
+	1,
+)
 
 var leasedJob = testfixtures.JobDb.NewJob(
 	util.NewULID(),
@@ -140,14 +164,42 @@ var leasedJob = testfixtures.JobDb.NewJob(
 	uint32(10),
 	schedulingInfo,
 	false,
-	2,
+	1,
+	false,
+	false,
+	false,
+	1,
+).WithNewRun("testExecutor", "test-node", "node", 5)
+
+var returnedOnceLeasedJob = testfixtures.JobDb.NewJob(
+	"01h3w2wtdchtc80hgyp782shrv",
+	"testJobset",
+	"testQueue",
+	uint32(10),
+	schedulingInfo,
+	false,
+	3,
 	false,
 	false,
 	false,
 	1,
-).WithQueued(false).WithNewRun("testExecutor", "test-node", "node", 5)
+).WithUpdatedRun(testfixtures.JobDb.CreateRun(
+	uuid.New(),
+	"01h3w2wtdchtc80hgyp782shrv",
+	0,
+	"testExecutor",
+	"testNodeId",
+	"testNodeName",
+	&scheduledAtPriority,
+	false,
+	false,
+	true,
+	false,
+	true,
+	true,
+)).WithNewRun("testExecutor", "test-node", "node", 5)
 
-var defaultJobRunError = &armadaevents.Error{
+var defaultJobError = &armadaevents.Error{
 	Terminal: true,
 	Reason: &armadaevents.Error_PodError{
 		PodError: &armadaevents.PodError{
@@ -156,6 +208,21 @@ var defaultJobRunError = &armadaevents.Error{
 	},
 }
 
+func defaultJobRunError(jobId string, runId uuid.UUID) *armadaevents.JobRunErrors {
+	protoJobId, err := armadaevents.ProtoUuidFromUlidString(jobId)
+	if err != nil {
+		panic(err)
+	}
+	protoRunId := armadaevents.ProtoUuidFromUuid(runId)
+	return &armadaevents.JobRunErrors{
+		RunId: protoRunId,
+		JobId: protoJobId,
+		Errors: []*armadaevents.Error{
+			defaultJobError,
+		},
+	}
+}
+
 var leasedFailFastJob = testfixtures.JobDb.NewJob(
 	util.NewULID(),
 	"testJobset",
@@ -163,18 +230,20 @@ var leasedFailFastJob = testfixtures.JobDb.NewJob(
 	uint32(10),
 	failFastSchedulingInfo,
 	false,
-	2,
+	1,
 	false,
 	false,
 	false,
 	1,
-).WithQueued(false).WithNewRun("testExecutor", "test-node", "node", 5)
-
-var scheduledAtPriority = int32(5)
+).WithNewRun("testExecutor", "test-node", "node", 5)
 
 var (
-	requeuedJobId = util.NewULID()
-	requeuedJob   = testfixtures.JobDb.NewJob(
+	testExecutor        = "test-executor"
+	testNode            = "test-node"
+	testNodeId          = api.NodeIdFromExecutorAndNodeName(testExecutor, testNode)
+	scheduledAtPriority = int32(10)
+	requeuedJobId       = util.NewULID()
+	requeuedJob         = testfixtures.JobDb.NewJob(
 		requeuedJobId,
 		"testJobset",
 		"testQueue",
@@ -186,29 +255,27 @@ var (
 		false,
 		false,
 		1,
-	).WithUpdatedRun(
-		testfixtures.JobDb.CreateRun(
-			uuid.New(),
-			requeuedJobId,
-			time.Now().Unix(),
-			"testExecutor",
-			"test-node",
-			"node",
-			&scheduledAtPriority,
-			false,
-			false,
-			true,
-			false,
-			true,
-			true,
-		),
-	)
+	).WithUpdatedRun(testfixtures.JobDb.CreateRun(
+		uuid.New(),
+		requeuedJobId,
+		time.Now().Unix(),
+		"testExecutor",
+		"test-node",
+		"node",
+		&scheduledAtPriority,
+		false,
+		false,
+		true,
+		false,
+		true,
+		true,
+	))
 )
 
 // Test a single scheduler cycle
 func TestScheduler_TestCycle(t *testing.T) {
 	tests := map[string]struct {
-		initialJobs                      []*jobdb.Job                      // jobs in the jobdb at the start of the cycle
+		initialJobs                      []*jobdb.Job                      // jobs in the jobDb at the start of the cycle
 		jobUpdates                       []database.Job                    // job updates from the database
 		runUpdates                       []database.Run                    // run updates from the database
 		jobRunErrors                     map[uuid.UUID]*armadaevents.Error // job run errors in the database
@@ -239,7 +306,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 			initialJobs:           []*jobdb.Job{queuedJob},
 			expectedJobRunLeased:  []string{queuedJob.Id()},
 			expectedLeased:        []string{queuedJob.Id()},
-			expectedQueuedVersion: queuedJob.QueuedVersion() + 1,
+			expectedQueuedVersion: 1,
 		},
 		"Lease a single job from an update": {
 			jobUpdates: []database.Job{
@@ -248,7 +315,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 					JobSet:                "testJobSet",
 					Queue:                 "testQueue",
 					Queued:                true,
-					QueuedVersion:         1,
+					QueuedVersion:         0,
 					SchedulingInfo:        schedulingInfoBytes,
 					SchedulingInfoVersion: int32(schedulingInfo.Version),
 					Serial:                1,
@@ -256,7 +323,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 			},
 			expectedJobRunLeased:  []string{queuedJob.Id()},
 			expectedLeased:        []string{queuedJob.Id()},
-			expectedQueuedVersion: queuedJob.QueuedVersion() + 1,
+			expectedQueuedVersion: 1,
 		},
 		"Lease two jobs from an update": {
 			jobUpdates: []database.Job{
@@ -265,7 +332,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 					JobSet:                "testJobSet",
 					Queue:                 "testQueue",
 					Queued:                true,
-					QueuedVersion:         1,
+					QueuedVersion:         0,
 					SchedulingInfo:        schedulingInfoBytes,
 					SchedulingInfoVersion: int32(schedulingInfo.Version),
 					Serial:                1,
@@ -275,7 +342,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 					JobSet:                "testJobSet",
 					Queue:                 "testQueue",
 					Queued:                true,
-					QueuedVersion:         1,
+					QueuedVersion:         0,
 					SchedulingInfo:        schedulingInfoBytes,
 					SchedulingInfoVersion: int32(schedulingInfo.Version),
 					Serial:                1,
@@ -283,7 +350,40 @@ func TestScheduler_TestCycle(t *testing.T) {
 			},
 			expectedJobRunLeased:  []string{"01h3w2wtdchtc80hgyp782shrv", "01h434g4hxww2pknb2q1nfmfph"},
 			expectedLeased:        []string{"01h3w2wtdchtc80hgyp782shrv", "01h434g4hxww2pknb2q1nfmfph"},
-			expectedQueuedVersion: queuedJob.QueuedVersion() + 1,
+			expectedQueuedVersion: 1,
+		},
+		"New failed job with no runs": {
+			// This happens if the scheduler decides to fail a job, e.g., due to min-max gang scheduling.
+			jobUpdates: []database.Job{
+				{
+					JobID:                 queuedJob.Id(),
+					JobSet:                "testJobSet",
+					Queue:                 "testQueue",
+					Failed:                true,
+					QueuedVersion:         0,
+					SchedulingInfo:        schedulingInfoBytes,
+					SchedulingInfoVersion: int32(schedulingInfo.Version),
+					Serial:                1,
+				},
+			},
+			expectedQueuedVersion: 0,
+		},
+		"Queued job transitions straight to failed without running": {
+			// This happens if the scheduler decides to fail a job, e.g., due to min-max gang scheduling.
+			initialJobs: []*jobdb.Job{queuedJob},
+			jobUpdates: []database.Job{
+				{
+					JobID:                 queuedJob.Id(),
+					JobSet:                "testJobSet",
+					Queue:                 "testQueue",
+					Failed:                true,
+					QueuedVersion:         0,
+					SchedulingInfo:        schedulingInfoBytes,
+					SchedulingInfoVersion: int32(schedulingInfo.Version),
+					Serial:                1,
+				},
+			},
+			expectedQueuedVersion: 0,
 		},
 		"Nothing leased": {
 			initialJobs:           []*jobdb.Job{queuedJob},
@@ -325,6 +425,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 					JobID:        requeuedJob.Id(),
 					JobSet:       "testJobSet",
 					Executor:     "testExecutor",
+					Node:         "node",
 					Failed:       true,
 					Returned:     true,
 					RunAttempted: true,
@@ -374,7 +475,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 			expectedQueuedVersion: leasedJob.QueuedVersion() + 1,
 		},
 		// When a lease is returned and the run was attempted, a node anti affinity is added
-		// If this node anti-affinity makes the job unschedulable, it should be failed
+		// If this node anti-affinity makes the job unschedulable, it should be failed.
 		"Lease returned and failed": {
 			initialJobs: []*jobdb.Job{leasedJob},
 			runUpdates: []database.Run{
@@ -389,39 +490,35 @@ func TestScheduler_TestCycle(t *testing.T) {
 					Serial:       1,
 				},
 			},
+			jobRunErrors: map[uuid.UUID]*armadaevents.Error{
+				leasedJob.LatestRun().Id(): defaultJobError,
+			},
 			submitCheckerFailure:  true,
 			expectedJobErrors:     []string{leasedJob.Id()},
 			expectedTerminal:      []string{leasedJob.Id()},
 			expectedQueuedVersion: leasedJob.QueuedVersion(),
 		},
 		"Lease returned too many times": {
-			initialJobs: []*jobdb.Job{leasedJob},
-			// 2 failures here so the second one should trigger a run failure
+			initialJobs: []*jobdb.Job{returnedOnceLeasedJob},
 			runUpdates: []database.Run{
 				{
-					RunID:        leasedJob.LatestRun().Id(),
-					JobID:        leasedJob.Id(),
-					JobSet:       "testJobSet",
-					Executor:     "testExecutor",
-					Failed:       true,
-					Returned:     true,
-					RunAttempted: true,
-					Serial:       1,
-				},
-				{
-					RunID:        uuid.New(),
-					JobID:        leasedJob.Id(),
+					RunID:        returnedOnceLeasedJob.LatestRun().Id(),
+					JobID:        returnedOnceLeasedJob.Id(),
 					JobSet:       "testJobSet",
 					Executor:     "testExecutor",
+					Node:         "testNode",
 					Failed:       true,
 					Returned:     true,
 					RunAttempted: true,
 					Serial:       2,
 				},
 			},
-			expectedJobErrors:     []string{leasedJob.Id()},
-			expectedTerminal:      []string{leasedJob.Id()},
-			expectedQueuedVersion: leasedJob.QueuedVersion(),
+			jobRunErrors: map[uuid.UUID]*armadaevents.Error{
+				returnedOnceLeasedJob.LatestRun().Id(): defaultJobError,
+			},
+			expectedJobErrors:     []string{returnedOnceLeasedJob.Id()},
+			expectedTerminal:      []string{returnedOnceLeasedJob.Id()},
+			expectedQueuedVersion: 3,
 		},
 		"Lease returned for fail fast job": {
 			initialJobs: []*jobdb.Job{leasedFailFastJob},
@@ -438,6 +535,9 @@ func TestScheduler_TestCycle(t *testing.T) {
 					Serial:       1,
 				},
 			},
+			jobRunErrors: map[uuid.UUID]*armadaevents.Error{
+				leasedFailFastJob.LatestRun().Id(): defaultJobError,
+			},
 			expectedJobErrors:     []string{leasedFailFastJob.Id()},
 			expectedTerminal:      []string{leasedFailFastJob.Id()},
 			expectedQueuedVersion: leasedFailFastJob.QueuedVersion(),
@@ -570,7 +670,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				},
 			},
 			jobRunErrors: map[uuid.UUID]*armadaevents.Error{
-				leasedJob.LatestRun().Id(): defaultJobRunError,
+				leasedJob.LatestRun().Id(): defaultJobError,
 			},
 			expectedJobErrors:     []string{leasedJob.Id()},
 			expectedTerminal:      []string{leasedJob.Id()},
@@ -664,6 +764,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				nil,
 			)
 			require.NoError(t, err)
+			sched.EnableAssertions()
 
 			sched.clock = testClock
 
@@ -682,7 +783,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			// Assert that all expected events are generated and that all events are expected.
+			// Assert that all expected eventSequences are generated and that all eventSequences are expected.
 			outstandingEventsByType := map[string]map[string]bool{
 				fmt.Sprintf("%T", &armadaevents.EventSequence_Event_JobRunLeased{}):     stringSet(tc.expectedJobRunLeased),
 				fmt.Sprintf("%T", &armadaevents.EventSequence_Event_JobErrors{}):        stringSet(tc.expectedJobErrors),
@@ -694,10 +795,10 @@ func TestScheduler_TestCycle(t *testing.T) {
 				fmt.Sprintf("%T", &armadaevents.EventSequence_Event_JobRequeued{}):      stringSet(tc.expectedRequeued),
 				fmt.Sprintf("%T", &armadaevents.EventSequence_Event_CancelJob{}):        stringSet(tc.expectedJobRequestCancel),
 			}
-			err = subtractEventsFromOutstandingEventsByType(publisher.events, outstandingEventsByType)
+			err = subtractEventsFromOutstandingEventsByType(publisher.eventSequences, outstandingEventsByType)
 			require.NoError(t, err)
 			for eventType, m := range outstandingEventsByType {
-				assert.Empty(t, m, "%d outstanding events of type %s", len(m), eventType)
+				assert.Empty(t, m, "%d outstanding eventSequences of type %s", len(m), eventType)
 			}
 
 			// assert that the serials are where we expect them to be
@@ -751,11 +852,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				podRequirements := job.PodRequirements()
 				assert.NotNil(t, podRequirements)
 
-				expectedQueuedVersion := int32(1)
-				if tc.expectedQueuedVersion != 0 {
-					expectedQueuedVersion = tc.expectedQueuedVersion
-				}
-				assert.Equal(t, expectedQueuedVersion, job.QueuedVersion())
+				assert.Equal(t, tc.expectedQueuedVersion, job.QueuedVersion())
 				expectedSchedulingInfoVersion := 1
 				if tc.expectedJobSchedulingInfoVersion != 0 {
 					expectedSchedulingInfoVersion = tc.expectedJobSchedulingInfoVersion
@@ -828,6 +925,7 @@ func TestRun(t *testing.T) {
 		nil,
 	)
 	require.NoError(t, err)
+	sched.EnableAssertions()
 
 	sched.clock = testClock
 
@@ -853,19 +951,19 @@ func TestRun(t *testing.T) {
 
 	// fire a cycle and assert that we became leader and published
 	fireCycle()
-	assert.Equal(t, 1, len(publisher.events))
+	assert.Equal(t, 1, len(publisher.eventSequences))
 	assert.Equal(t, schedulingAlgo.numberOfScheduleCalls, 1)
 
 	// invalidate our leadership: we should not publish
 	leaderController.token = InvalidLeaderToken()
 	fireCycle()
-	assert.Equal(t, 0, len(publisher.events))
+	assert.Equal(t, 0, len(publisher.eventSequences))
 	assert.Equal(t, schedulingAlgo.numberOfScheduleCalls, 1)
 
 	// become master again: we should publish
 	leaderController.token = NewLeaderToken()
 	fireCycle()
-	assert.Equal(t, 1, len(publisher.events))
+	assert.Equal(t, 1, len(publisher.eventSequences))
 	assert.Equal(t, schedulingAlgo.numberOfScheduleCalls, 2)
 
 	cancel()
@@ -887,7 +985,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 					Queue:          queuedJob.Queue(),
 					Submitted:      queuedJob.Created(),
 					Queued:         true,
-					QueuedVersion:  1,
+					QueuedVersion:  0,
 					Priority:       int64(queuedJob.Priority()),
 					SchedulingInfo: schedulingInfoBytes,
 					Serial:         1,
@@ -949,7 +1047,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 						"test-executor",
 						"test-executor-test-node",
 						"test-node",
-						&scheduledAtPriority,
+						pointerFromValue(int32(5)),
 						false,
 						false,
 						false,
@@ -1044,6 +1142,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 				nil,
 			)
 			require.NoError(t, err)
+			sched.EnableAssertions()
 
 			// The SchedulingKeyGenerator embedded in the jobDb has some randomness,
 			// which must be consistent within tests.
@@ -1055,10 +1154,12 @@ func TestScheduler_TestSyncState(t *testing.T) {
 			require.NoError(t, err)
 			txn.Commit()
 
-			updatedJobs, _, _, err := sched.syncState(ctx)
+			updatedJobs, _, err := sched.syncState(ctx)
 			require.NoError(t, err)
 
-			assert.Equal(t, tc.expectedUpdatedJobs, updatedJobs)
+			expectedJobDb := testfixtures.NewJobDbWithJobs(tc.expectedUpdatedJobs)
+			actualJobDb := testfixtures.NewJobDbWithJobs(updatedJobs)
+			assert.NoError(t, expectedJobDb.ReadTxn().AssertEqual(actualJobDb.ReadTxn()))
 			allDbJobs := sched.jobDb.ReadTxn().GetAll()
 
 			expectedIds := stringSet(tc.expectedJobDbIds)
@@ -1157,13 +1258,20 @@ type testSchedulingAlgo struct {
 	jobsToSchedule        []string
 	jobsToFail            []string
 	shouldError           bool
+	// Set to true to indicate that preemption/scheduling/failure decisions have been persisted.
+	// Until persisted is set to true, the same jobs are preempted/scheduled/failed on every call.
+	persisted bool
 }
 
-func (t *testSchedulingAlgo) Schedule(ctx *armadacontext.Context, txn *jobdb.Txn) (*SchedulerResult, error) {
+func (t *testSchedulingAlgo) Schedule(_ *armadacontext.Context, txn *jobdb.Txn) (*SchedulerResult, error) {
 	t.numberOfScheduleCalls++
 	if t.shouldError {
 		return nil, errors.New("error scheduling jobs")
 	}
+	if t.persisted {
+		// Exit right away if decisions have already been persisted.
+		return &SchedulerResult{}, nil
+	}
 	preemptedJobs := make([]*jobdb.Job, 0, len(t.jobsToPreempt))
 	scheduledJobs := make([]*jobdb.Job, 0, len(t.jobsToSchedule))
 	failedJobs := make([]*jobdb.Job, 0, len(t.jobsToFail))
@@ -1189,13 +1297,18 @@ func (t *testSchedulingAlgo) Schedule(ctx *armadacontext.Context, txn *jobdb.Txn
 			return nil, errors.Errorf("was asked to lease %s but job does not exist", id)
 		}
 		if !job.Queued() {
-			return nil, errors.Errorf("was asked to lease %s but job was already leased", job.Id())
+			return nil, errors.Errorf("was asked to lease %s but job is not queued", job.Id())
 		}
 		priority := int32(0)
 		if req := job.PodRequirements(); req != nil {
 			priority = req.Priority
 		}
-		job = job.WithQueuedVersion(job.QueuedVersion()+1).WithQueued(false).WithNewRun("test-executor", "test-node", "node", priority)
+		job = job.WithQueuedVersion(job.QueuedVersion()+1).WithQueued(false).WithNewRun(
+			testExecutor,
+			testNodeId,
+			testNode,
+			priority,
+		)
 		scheduledJobs = append(scheduledJobs, job)
 	}
 	for _, id := range t.jobsToFail {
@@ -1204,7 +1317,7 @@ func (t *testSchedulingAlgo) Schedule(ctx *armadacontext.Context, txn *jobdb.Txn
 			return nil, errors.Errorf("was asked to lease %s but job does not exist", id)
 		}
 		if !job.Queued() {
-			return nil, errors.Errorf("was asked to lease %s but job was already leased", job.Id())
+			return nil, errors.Errorf("was asked to lease %s but job is not queued", job.Id())
 		}
 		job = job.WithQueued(false).WithFailed(true)
 		failedJobs = append(failedJobs, job)
@@ -1221,6 +1334,15 @@ func (t *testSchedulingAlgo) Schedule(ctx *armadacontext.Context, txn *jobdb.Txn
 	return NewSchedulerResultForTest(preemptedJobs, scheduledJobs, failedJobs, nil), nil
 }
 
+func (t *testSchedulingAlgo) Persist() {
+	if t.numberOfScheduleCalls == 0 {
+		// Nothing to persist if there have been no calls to schedule.
+		return
+	}
+	t.persisted = true
+	return
+}
+
 func NewSchedulerResultForTest[S ~[]T, T interfaces.LegacySchedulerJob](
 	preemptedJobs S,
 	scheduledJobs S,
@@ -1236,20 +1358,37 @@ func NewSchedulerResultForTest[S ~[]T, T interfaces.LegacySchedulerJob](
 }
 
 type testPublisher struct {
-	events      []*armadaevents.EventSequence
-	shouldError bool
+	i              int
+	eventSequences []*armadaevents.EventSequence
+	shouldError    bool
 }
 
-func (t *testPublisher) PublishMessages(ctx *armadacontext.Context, events []*armadaevents.EventSequence, _ func() bool) error {
-	t.events = events
+func newTestPublisher() *testPublisher {
+	// Initialise with an empty slice to make require.Equal comparison with an initialised slice work.
+	return &testPublisher{eventSequences: make([]*armadaevents.EventSequence, 0)}
+}
+
+func (t *testPublisher) PublishMessages(_ *armadacontext.Context, events []*armadaevents.EventSequence, shouldPublish func() bool) error {
 	if t.shouldError {
-		return errors.New("Error when publishing")
+		return errors.New("testPublisher error")
+	}
+	if shouldPublish() {
+		t.eventSequences = append(t.eventSequences, events...)
 	}
 	return nil
 }
 
+func (t *testPublisher) ReadAll() []*armadaevents.EventSequence {
+	if len(t.eventSequences) == 0 {
+		return nil
+	}
+	rv := t.eventSequences[t.i:len(t.eventSequences)]
+	t.i += len(rv)
+	return rv
+}
+
 func (t *testPublisher) Reset() {
-	t.events = nil
+	t.eventSequences = nil
 }
 
 func (t *testPublisher) PublishMarkers(ctx *armadacontext.Context, groupId uuid.UUID) (uint32, error) {
@@ -1263,3 +1402,1068 @@ func stringSet(src []string) map[string]bool {
 	}
 	return set
 }
+
+var (
+	queuedJobA = &database.Job{
+		JobID:                 util.NewULID(),
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	queuedJobWithTTLA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoWithQueueTtlBytes,
+		SchedulingInfoVersion: int32(schedulingInfoWithQueueTtl.Version),
+		Serial:                0,
+	}
+	queuedJobWithFailFastA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                true,
+		QueuedVersion:         0,
+		SchedulingInfo:        failFastSchedulingInfoBytes,
+		SchedulingInfoVersion: int32(failFastSchedulingInfo.Version),
+		Serial:                0,
+	}
+	queuedJobWithUpdatedPriorityA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoWithUpdatedPriorityBytes,
+		SchedulingInfoVersion: int32(schedulingInfoWithUpdatedPriority.Version),
+		Serial:                1,
+	}
+	runningJobA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		QueuedVersion:         1,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	runningJobWithUpdatedPriorityA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		QueuedVersion:         1,
+		SchedulingInfo:        schedulingInfoWithUpdatedPriorityBytes,
+		SchedulingInfoVersion: int32(schedulingInfoWithUpdatedPriority.Version),
+		Serial:                1,
+	}
+	failedJobA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Failed:                true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	cancelledJobA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Cancelled:             true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	cancelRequestedJobA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                true,
+		CancelRequested:       true,
+		QueuedVersion:         0,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	cancelByJobSetRequestedJobA = &database.Job{
+		JobID:                   queuedJobA.JobID,
+		JobSet:                  "testJobSet",
+		Queue:                   "testQueue",
+		Queued:                  true,
+		CancelByJobsetRequested: true,
+		QueuedVersion:           0,
+		SchedulingInfo:          schedulingInfoBytes,
+		SchedulingInfoVersion:   int32(schedulingInfo.Version),
+		Serial:                  0,
+	}
+	runningCancelRequestedJobA = &database.Job{
+		JobID:                 queuedJobA.JobID,
+		JobSet:                "testJobSet",
+		Queue:                 "testQueue",
+		Queued:                false,
+		CancelRequested:       true,
+		QueuedVersion:         1,
+		SchedulingInfo:        schedulingInfoBytes,
+		SchedulingInfoVersion: int32(schedulingInfo.Version),
+		Serial:                0,
+	}
+	runningCancelByJobSetRequestedJobA = &database.Job{
+		JobID:                   queuedJobA.JobID,
+		JobSet:                  "testJobSet",
+		Queue:                   "testQueue",
+		CancelByJobsetRequested: true,
+		QueuedVersion:           1,
+		SchedulingInfo:          schedulingInfoBytes,
+		SchedulingInfoVersion:   int32(schedulingInfo.Version),
+		Serial:                  0,
+	}
+	newRunA = &database.Run{
+		RunID:               testfixtures.UUIDFromInt(1),
+		JobID:               queuedJobA.JobID,
+		JobSet:              queuedJobA.JobSet,
+		Executor:            testExecutor,
+		Node:                testNode,
+		RunAttempted:        true,
+		Serial:              0,
+		ScheduledAtPriority: &scheduledAtPriority,
+	}
+	successfulRunA = &database.Run{
+		RunID:               testfixtures.UUIDFromInt(1),
+		JobID:               queuedJobA.JobID,
+		JobSet:              queuedJobA.JobSet,
+		Executor:            testExecutor,
+		Node:                testNode,
+		Succeeded:           true,
+		RunAttempted:        true,
+		Serial:              0,
+		ScheduledAtPriority: &scheduledAtPriority,
+	}
+	failedAttemptedRunA = &database.Run{
+		RunID:               testfixtures.UUIDFromInt(1),
+		JobID:               queuedJobA.JobID,
+		JobSet:              queuedJobA.JobSet,
+		Executor:            testExecutor,
+		Node:                testNode,
+		Failed:              true,
+		RunAttempted:        true,
+		Serial:              0,
+		ScheduledAtPriority: &scheduledAtPriority,
+	}
+	failedRunA = &database.Run{
+		RunID:               testfixtures.UUIDFromInt(1),
+		JobID:               queuedJobA.JobID,
+		JobSet:              queuedJobA.JobSet,
+		Executor:            testExecutor,
+		Node:                testNode,
+		Failed:              true,
+		Serial:              0,
+		ScheduledAtPriority: &scheduledAtPriority,
+	}
+)
+
+// TestCycleConsistency runs two replicas of the scheduler and asserts that their state remains consistent
+// under various permutations of making scheduling decisions and failing over.
+//
+// TODO(albin): Test lease expiry.
+func TestCycleConsistency(t *testing.T) {
+	type schedulerDbUpdate struct {
+		jobUpdates   []*database.Job              // Job updates from the database.
+		runUpdates   []*database.Run              // Run updates from the database.
+		jobRunErrors []*armadaevents.JobRunErrors // Job run errors from the database.
+	}
+	tests := map[string]struct {
+		// Each test case consists of two updates written to the scheduler postgres.
+		firstSchedulerDbUpdate  schedulerDbUpdate
+		secondSchedulerDbUpdate schedulerDbUpdate
+
+		// Inputs of the mocked scheduling algorithm.
+		// Controls which jobs the scheduler should schedule/preempt/fail.
+		idsOfJobsToSchedule []string
+		idsOfJobsToPreempt  []string
+		idsOfJobsToFail     []string
+
+		// Expected jobDbs for scenario 1, i.e., the baseline scenario.
+		// Only compared against if not nil.
+		expectedJobDbCycleOne   []*jobdb.Job
+		expectedJobDbCycleTwo   []*jobdb.Job
+		expectedJobDbCycleThree []*jobdb.Job
+
+		// Expected published events for scenario 1, i.e., the baseline scenario.
+		// Only compared against if not nil.
+		expectedEventSequencesCycleOne   []*armadaevents.EventSequence
+		expectedEventSequencesCycleTwo   []*armadaevents.EventSequence
+		expectedEventSequencesCycleThree []*armadaevents.EventSequence
+	}{
+		"Load a queued job": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			expectedEventSequencesCycleThree: make([]*armadaevents.EventSequence, 0),
+		},
+		"Load a failed job": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					failedJobA,
+				},
+			},
+		},
+		"Load a cancelled job": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					cancelledJobA,
+				},
+			},
+		},
+		"Load a queued job with cancel requested": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					cancelRequestedJobA,
+				},
+			},
+		},
+		"Load a queued job with cancel by job set requested": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					cancelByJobSetRequestedJobA,
+				},
+			},
+		},
+		"Load a running job with cancel requested": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningCancelRequestedJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+		},
+		"Load a running job with cancel by job set requested": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningCancelByJobSetRequestedJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+		},
+		"Schedule a new job": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			idsOfJobsToSchedule: []string{queuedJobA.JobID},
+			expectedJobDbCycleOne: []*jobdb.Job{
+				testfixtures.NewJobDb().NewJob(
+					queuedJobA.JobID,
+					queuedJobA.JobSet,
+					queuedJobA.Queue,
+					uint32(queuedJobA.Priority),
+					schedulingInfo,
+					false,
+					1,
+					queuedJobA.CancelRequested,
+					queuedJobA.CancelByJobsetRequested,
+					queuedJobA.Cancelled,
+					0,
+				).WithNewRun(testExecutor, testNodeId, testNode, 10),
+			},
+			expectedEventSequencesCycleThree: []*armadaevents.EventSequence{
+				{
+					Queue:      queuedJobA.Queue,
+					JobSetName: queuedJobA.JobSet,
+					Events: func() []*armadaevents.EventSequence_Event {
+						uuidProvider := testfixtures.MockUUIDProvider{}
+						runId := uuidProvider.New()
+						created := time.Unix(0, 0)
+						return []*armadaevents.EventSequence_Event{
+							{
+								Created: &created,
+								Event: &armadaevents.EventSequence_Event_JobRunLeased{
+									JobRunLeased: &armadaevents.JobRunLeased{
+										RunId:                  armadaevents.ProtoUuidFromUuid(runId),
+										JobId:                  armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+										ExecutorId:             testExecutor,
+										NodeId:                 testNode,
+										UpdateSequenceNumber:   1,
+										HasScheduledAtPriority: true,
+										ScheduledAtPriority:    10,
+										AdditionalAnnotations:  make(map[string]string),
+									},
+								},
+							},
+						}
+					}(),
+				},
+			},
+		},
+		"Fail a new job": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			idsOfJobsToFail: []string{queuedJobA.JobID},
+		},
+		"Schedule a job that then succeeds": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				runUpdates: []*database.Run{
+					successfulRunA,
+				},
+			},
+			idsOfJobsToSchedule: []string{queuedJobA.JobID},
+			expectedJobDbCycleOne: []*jobdb.Job{
+				testfixtures.NewJobDb().NewJob(
+					queuedJobA.JobID,
+					queuedJobA.JobSet,
+					queuedJobA.Queue,
+					uint32(queuedJobA.Priority),
+					schedulingInfo,
+					false,
+					1,
+					queuedJobA.CancelRequested,
+					queuedJobA.CancelByJobsetRequested,
+					queuedJobA.Cancelled,
+					0,
+				).WithNewRun(testExecutor, testNodeId, testNode, 10),
+			},
+			expectedJobDbCycleTwo: func() []*jobdb.Job {
+				job := testfixtures.NewJobDb().NewJob(
+					queuedJobA.JobID,
+					queuedJobA.JobSet,
+					queuedJobA.Queue,
+					uint32(queuedJobA.Priority),
+					schedulingInfo,
+					false,
+					1,
+					queuedJobA.CancelRequested,
+					queuedJobA.CancelByJobsetRequested,
+					queuedJobA.Cancelled,
+					0,
+				).WithNewRun(testExecutor, testNodeId, testNode, 10).WithSucceeded(true)
+				job = job.WithUpdatedRun(job.LatestRun().WithSucceeded(true).WithAttempted(true))
+				return []*jobdb.Job{job}
+			}(),
+			expectedJobDbCycleThree: make([]*jobdb.Job, 0),
+			expectedEventSequencesCycleThree: []*armadaevents.EventSequence{
+				{
+					Queue:      queuedJobA.Queue,
+					JobSetName: queuedJobA.JobSet,
+					Events: []*armadaevents.EventSequence_Event{
+						{
+							Created: pointerFromValue(time.Unix(0, 0)),
+							Event: &armadaevents.EventSequence_Event_JobRunLeased{
+								JobRunLeased: &armadaevents.JobRunLeased{
+									RunId:                  armadaevents.ProtoUuidFromUuid(testfixtures.UUIDFromInt(1)),
+									JobId:                  armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+									ExecutorId:             testExecutor,
+									NodeId:                 testNode,
+									UpdateSequenceNumber:   1,
+									HasScheduledAtPriority: true,
+									ScheduledAtPriority:    10,
+									AdditionalAnnotations:  make(map[string]string),
+								},
+							},
+						},
+					},
+				},
+				{
+					Queue:      queuedJobA.Queue,
+					JobSetName: queuedJobA.JobSet,
+					Events: []*armadaevents.EventSequence_Event{
+						{
+							Created: pointerFromValue(time.Unix(0, 0)),
+							Event: &armadaevents.EventSequence_Event_JobSucceeded{
+								JobSucceeded: &armadaevents.JobSucceeded{
+									JobId: armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		"Schedule a new job that then fails after starting to run": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				runUpdates: []*database.Run{
+					failedAttemptedRunA,
+				},
+				jobRunErrors: []*armadaevents.JobRunErrors{
+					defaultJobRunError(queuedJobA.JobID, failedAttemptedRunA.RunID),
+				},
+			},
+			idsOfJobsToSchedule: []string{queuedJobA.JobID},
+		},
+		// TODO(albin): We should assert that anti-affinities have been added.
+		"Schedule a new job that then fails to start": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				runUpdates: []*database.Run{
+					failedRunA,
+				},
+				jobRunErrors: []*armadaevents.JobRunErrors{
+					defaultJobRunError(queuedJobA.JobID, failedAttemptedRunA.RunID),
+				},
+			},
+			idsOfJobsToSchedule: []string{queuedJobA.JobID},
+		},
+		"Schedule a new job that then fails to start with fail-fast set to true": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobWithFailFastA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				runUpdates: []*database.Run{
+					failedRunA,
+				},
+				jobRunErrors: []*armadaevents.JobRunErrors{
+					defaultJobRunError(queuedJobA.JobID, failedAttemptedRunA.RunID),
+				},
+			},
+			idsOfJobsToSchedule: []string{queuedJobA.JobID},
+		},
+		// TODO(albin): Also test a job being attempted to many times.
+		"Running job is preempted": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+			idsOfJobsToPreempt:      []string{queuedJobA.JobID},
+			expectedJobDbCycleThree: make([]*jobdb.Job, 0),
+			expectedEventSequencesCycleThree: []*armadaevents.EventSequence{
+				{
+					Queue:      queuedJobA.Queue,
+					JobSetName: queuedJobA.JobSet,
+					Events: []*armadaevents.EventSequence_Event{
+						{
+							Created: pointerFromValue(time.Unix(0, 0)),
+							Event: &armadaevents.EventSequence_Event_JobRunPreempted{
+								JobRunPreempted: &armadaevents.JobRunPreempted{
+									PreemptedRunId: armadaevents.ProtoUuidFromUuid(testfixtures.UUIDFromInt(1)),
+									PreemptedJobId: armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+								},
+							},
+						},
+						{
+							Created: pointerFromValue(time.Unix(0, 0)),
+							Event: &armadaevents.EventSequence_Event_JobRunErrors{
+								JobRunErrors: &armadaevents.JobRunErrors{
+									JobId: armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+									RunId: armadaevents.ProtoUuidFromUuid(testfixtures.UUIDFromInt(1)),
+									Errors: []*armadaevents.Error{
+										{
+											Terminal: true,
+											Reason: &armadaevents.Error_JobRunPreemptedError{
+												JobRunPreemptedError: &armadaevents.JobRunPreemptedError{},
+											},
+										},
+									},
+								},
+							},
+						},
+						{
+							Created: pointerFromValue(time.Unix(0, 0)),
+							Event: &armadaevents.EventSequence_Event_JobErrors{
+								JobErrors: &armadaevents.JobErrors{
+									JobId: armadaevents.MustProtoUuidFromUlidString(queuedJobA.JobID),
+									Errors: []*armadaevents.Error{
+										{
+											Terminal: true,
+											Reason: &armadaevents.Error_JobRunPreemptedError{
+												JobRunPreemptedError: &armadaevents.JobRunPreemptedError{},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		"Queued job is cancelled": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					cancelRequestedJobA,
+				},
+			},
+		},
+		"Running job is cancelled": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningCancelRequestedJobA,
+				},
+			},
+		},
+		"Queued job is cancelled by job set": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					cancelByJobSetRequestedJobA,
+				},
+			},
+		},
+		"Running job is cancelled by job set": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningCancelByJobSetRequestedJobA,
+				},
+			},
+		},
+		"Queued job with expired ttl results in cancellation": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobWithTTLA,
+				},
+			},
+		},
+		"Queued job is re-prioritised": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					queuedJobWithUpdatedPriorityA,
+				},
+			},
+		},
+		"Running job is re-prioritised": {
+			firstSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningJobA,
+				},
+				runUpdates: []*database.Run{
+					newRunA,
+				},
+			},
+			secondSchedulerDbUpdate: schedulerDbUpdate{
+				jobUpdates: []*database.Job{
+					runningJobWithUpdatedPriorityA,
+				},
+			},
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			ctx := armadacontext.Background()
+			testClock := clock.NewFakeClock(time.Unix(0, 0))
+
+			// Setup necessary for creating schedulerDb operations.
+			queueByJobId := make(map[string]string)
+			jobSetByJobId := make(map[string]string)
+			for _, jobUpdate := range tc.firstSchedulerDbUpdate.jobUpdates {
+				queueByJobId[jobUpdate.JobID] = jobUpdate.Queue
+				jobSetByJobId[jobUpdate.JobID] = jobUpdate.JobSet
+			}
+			for _, jobUpdate := range tc.secondSchedulerDbUpdate.jobUpdates {
+				queueByJobId[jobUpdate.JobID] = jobUpdate.Queue
+				jobSetByJobId[jobUpdate.JobID] = jobUpdate.JobSet
+			}
+			instructionConverter, err := scheduleringester.NewInstructionConverter(
+				nil,
+				testfixtures.TestPriorityClasses,
+			)
+			require.NoError(t, err)
+
+			// Helper function for creating new schedulers for use in tests.
+			newScheduler := func(db *pgxpool.Pool) *Scheduler {
+				scheduler, err := NewScheduler(
+					testfixtures.NewJobDb(),
+					database.NewPostgresJobRepository(db, 1024),
+					&testExecutorRepository{
+						updateTimes: map[string]time.Time{"test-executor": testClock.Now()},
+					},
+					&testSchedulingAlgo{
+						jobsToSchedule: tc.idsOfJobsToSchedule,
+						jobsToPreempt:  tc.idsOfJobsToPreempt,
+						jobsToFail:     tc.idsOfJobsToFail,
+					},
+					NewStandaloneLeaderController(),
+					newTestPublisher(),
+					&testSubmitChecker{},
+					1*time.Second,
+					5*time.Second,
+					0,
+					maxNumberOfAttempts,
+					nodeIdLabel,
+					schedulerMetrics,
+					nil,
+				)
+				require.NoError(t, err)
+				scheduler.clock = testClock
+				scheduler.EnableAssertions()
+				return scheduler
+			}
+
+			withTestSetup := func(f func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error) error {
+				return schedulerdb.WithTestDb(func(_ *schedulerdb.Queries, db *pgxpool.Pool) error {
+					// Create a schedulerDb using this db connection and initialise its state.
+					schedulerDb := scheduleringester.NewSchedulerDb(
+						db,
+						nil,
+						time.Second,
+						time.Second,
+						10*time.Second,
+					)
+
+					// Create two scheduler using the same db connection.
+					a := newScheduler(db)
+					b := newScheduler(db)
+
+					// Share the schedulingAlgo to ensure both schedulers make the same scheduling decisions.
+					b.schedulingAlgo = a.schedulingAlgo
+
+					// Initially, "a" is leader and "b" follower.
+					(b.leaderController.(*StandaloneLeaderController)).SetToken(InvalidLeaderToken())
+
+					return f(a, b, schedulerDb)
+				})
+			}
+
+			// Helper function for {first,second}DbUpdate.
+			dbUpdate := func(
+				schedulerDb *scheduleringester.SchedulerDb,
+				jobUpdates []*database.Job,
+				runUpdates []*database.Run,
+				jobRunErrors []*armadaevents.JobRunErrors,
+			) error {
+				dbOps, err := dbOpsFromDbObjects(
+					ctx,
+					instructionConverter,
+					queueByJobId,
+					jobSetByJobId,
+					jobUpdates,
+					runUpdates,
+					jobRunErrors,
+				)
+				if err != nil {
+					return err
+				}
+				return schedulerDb.Store(ctx, &scheduleringester.DbOperationsWithMessageIds{Ops: dbOps})
+			}
+
+			// write the first schedulerDb update to postgres.
+			firstDbUpdate := func(schedulerDb *scheduleringester.SchedulerDb) error {
+				t.Log("writing first schedulerDb update")
+				return dbUpdate(
+					schedulerDb,
+					tc.firstSchedulerDbUpdate.jobUpdates,
+					tc.firstSchedulerDbUpdate.runUpdates,
+					tc.firstSchedulerDbUpdate.jobRunErrors,
+				)
+			}
+
+			// write the second schedulerDb update to postgres.
+			secondDbUpdate := func(schedulerDb *scheduleringester.SchedulerDb) error {
+				t.Log("writing second schedulerDb update")
+				return dbUpdate(
+					schedulerDb,
+					tc.secondSchedulerDbUpdate.jobUpdates,
+					tc.secondSchedulerDbUpdate.runUpdates,
+					tc.secondSchedulerDbUpdate.jobRunErrors,
+				)
+			}
+
+			// cycle runs one cycle.
+			cycle := func(s *Scheduler, updateAll, shouldSchedule bool) error {
+				t.Logf("cycle scheduler %p", s)
+				_, err := s.cycle(ctx, updateAll, s.leaderController.GetToken(), shouldSchedule)
+				return err
+			}
+
+			// persist persists to the schedulerDb any published eventSequences.
+			persist := func(s *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				publisher := s.publisher.(*testPublisher)
+				eventSequences := publisher.ReadAll()
+				dbOpsWithMessageIds := instructionConverter.Convert(ctx, &ingest.EventSequencesWithIds{
+					EventSequences: eventSequences,
+				})
+
+				// Mark scheduling decisions as persisted.
+				// If not persisted, the same jobs are scheduled again on subsequent calls to schedule.
+				s.schedulingAlgo.(*testSchedulingAlgo).Persist()
+
+				// Logging
+				numEvents := 0
+				for _, eventSequence := range eventSequences {
+					numEvents += len(eventSequence.Events)
+				}
+				t.Logf("persist scheduler %p; %d eventSequences, %d eventSequences, %d ops", s, len(eventSequences), numEvents, len(dbOpsWithMessageIds.Ops))
+				for _, eventSequence := range eventSequences {
+					for _, event := range eventSequence.Events {
+						t.Logf("\tevent %s, %s, %T", eventSequence.Queue, eventSequence.JobSetName, event.Event)
+					}
+				}
+				for _, op := range dbOpsWithMessageIds.Ops {
+					t.Logf("\toperation %T: %v", op, op)
+				}
+
+				return schedulerDb.Store(ctx, dbOpsWithMessageIds)
+			}
+
+			// failover swaps the leader tokens between a and b, thus swapping which scheduler is leader.
+			failover := func(a, b *Scheduler) error {
+				t.Logf("failover schedulers %p, %p", a, b)
+				lca := a.leaderController.(*StandaloneLeaderController)
+				lcb := b.leaderController.(*StandaloneLeaderController)
+				lta := lca.GetToken()
+				ltb := lcb.GetToken()
+				lca.SetToken(ltb)
+				lcb.SetToken(lta)
+				return nil
+			}
+
+			eventsFromTestPublisher := func(p Publisher) []*armadaevents.EventSequence {
+				return p.(*testPublisher).eventSequences
+			}
+
+			var (
+				jobDbCycleOne   *jobdb.JobDb
+				jobDbCycleTwo   *jobdb.JobDb
+				jobDbCycleThree *jobdb.JobDb
+
+				eventsCycleOne   []*armadaevents.EventSequence
+				eventsCycleTwo   []*armadaevents.EventSequence
+				eventsCycleThree []*armadaevents.EventSequence
+			)
+
+			// One scheduler performing three cycles.
+			// Used for absolute assertions and as a baseline for relative assertions.
+			t.Log("scenario 1")
+			require.NoError(t, withTestSetup(func(a, _ *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+				jobDbCycleOne = a.jobDb.Clone()
+				eventsCycleOne = slices.Clone(eventsFromTestPublisher(a.publisher))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+				jobDbCycleTwo = a.jobDb.Clone()
+				eventsCycleTwo = slices.Clone(eventsFromTestPublisher(a.publisher))
+
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+				jobDbCycleThree = a.jobDb.Clone()
+				eventsCycleThree = slices.Clone(eventsFromTestPublisher(a.publisher))
+
+				return nil
+			}))
+
+			// Absolute assertions.
+			if tc.expectedJobDbCycleOne != nil {
+				jobDb := testfixtures.NewJobDbWithJobs(tc.expectedJobDbCycleOne)
+				require.NoError(t, jobDb.ReadTxn().AssertEqual(jobDbCycleOne.ReadTxn()), "unexpected cycle one jobDb")
+			}
+			if tc.expectedJobDbCycleTwo != nil {
+				jobDb := testfixtures.NewJobDbWithJobs(tc.expectedJobDbCycleTwo)
+				require.NoError(t, jobDb.ReadTxn().AssertEqual(jobDbCycleTwo.ReadTxn()), "unexpected cycle two jobDb")
+			}
+			if tc.expectedJobDbCycleThree != nil {
+				jobDb := testfixtures.NewJobDbWithJobs(tc.expectedJobDbCycleThree)
+				require.NoError(t, jobDb.ReadTxn().AssertEqual(jobDbCycleThree.ReadTxn()), "unexpected cycle three jobDb")
+			}
+
+			if tc.expectedEventSequencesCycleOne != nil {
+				require.Equal(t, tc.expectedEventSequencesCycleOne, eventsCycleOne, "unexpected cycle one events")
+			}
+			if tc.expectedEventSequencesCycleTwo != nil {
+				require.Equal(t, tc.expectedEventSequencesCycleTwo, eventsCycleTwo, "unexpected cycle two events")
+			}
+			if tc.expectedEventSequencesCycleThree != nil {
+				require.Equal(t, tc.expectedEventSequencesCycleThree, eventsCycleThree, "unexpected cycle three events")
+			}
+
+			// Test that the follower stays in sync with the leader.
+			// TODO(albin): We need to cycle "a" again after each persist for now.
+			//              The jobDb of the leader is supposed to be updated immediately to reflect published events.
+			//              However, the jobDb is only updated immediately for schedule/preempt/fail decisions.
+			//              For external changes, e.g., job succeeded, the jobDb is only updated on the second cycle.
+			// TODO(albin): We actually need to fix this since we may preempt an already successful job as it is.
+			// TODO(albin): Can we become leader again after failing over to become a follower when leader?
+			//              Let's test multiple failovers where you go back and forth.
+			t.Log("scenario 2")
+			// TODO(albin): Compare a against baseline a.
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, jobDbCycleOne.ReadTxn().AssertEqual(a.jobDb.ReadTxn()))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+				require.Empty(t, eventsFromTestPublisher(b.publisher))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, jobDbCycleTwo.ReadTxn().AssertEqual(a.jobDb.ReadTxn()))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+
+			// Test that the follower stays in sync with the leader when the follower cycles once before "a" persists.
+			t.Log("scenario 3")
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, persist(a, schedulerDb))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, a.jobDb.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+
+			// Test follower failover.
+			t.Log("scenario 4")
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(b, false, false))
+
+				require.Empty(t, eventsFromTestPublisher(b.publisher))
+				require.NoError(t, failover(a, b))
+				require.NoError(t, cycle(b, true, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleOne.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, cycle(b, false, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleTwo.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(b, false, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleThree.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+
+			// Test follower failover after the first db update.
+			t.Log("scenario 5")
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, failover(a, b))
+				require.NoError(t, cycle(b, true, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleTwo.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(b, false, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleThree.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+
+			// Test follower failover after the first db update, when the follower has already cycled once.
+			t.Log("scenario 6")
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, failover(a, b))
+				require.NoError(t, cycle(b, true, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleTwo.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(b, false, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleThree.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+
+			// Test follower failover after the first db update, when the follower has already cycled once before the second update.
+			t.Log("scenario 7")
+			require.NoError(t, withTestSetup(func(a, b *Scheduler, schedulerDb *scheduleringester.SchedulerDb) error {
+				require.NoError(t, firstDbUpdate(schedulerDb))
+				require.NoError(t, cycle(a, false, true))
+				require.NoError(t, persist(a, schedulerDb))
+
+				require.NoError(t, cycle(b, false, false))
+				require.NoError(t, secondDbUpdate(schedulerDb))
+				require.NoError(t, failover(a, b))
+				require.NoError(t, cycle(b, true, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleTwo.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				require.NoError(t, cycle(b, false, true))
+				require.NoError(t, persist(b, schedulerDb))
+				require.NoError(t, jobDbCycleThree.ReadTxn().AssertEqual(b.jobDb.ReadTxn()))
+
+				return nil
+			}))
+		})
+	}
+}
+
+func dbOpsFromDbObjects(
+	ctx *armadacontext.Context,
+	instructionConverter *scheduleringester.InstructionConverter,
+	queueByJobId map[string]string,
+	jobSetByJobId map[string]string,
+	jobUpdates []*database.Job,
+	runUpdates []*database.Run,
+	jobRunErrors []*armadaevents.JobRunErrors,
+) ([]scheduleringester.DbOperation, error) {
+	dbOps := make([]scheduleringester.DbOperation, 0)
+
+	// jobUpdatesByJobId := make(map[string]*database.Job)
+	insertJobsDbOp := make(scheduleringester.InsertJobs, len(jobUpdates))
+	for _, dbJob := range jobUpdates {
+		insertJobsDbOp[dbJob.JobID] = dbJob
+		// jobUpdatesByJobId[dbJob.JobID] = dbJob
+	}
+	dbOps = scheduleringester.AppendDbOperation(dbOps, fixInsertJobsDbOp(insertJobsDbOp))
+
+	insertRunsDbOp := make(scheduleringester.InsertRuns, len(runUpdates))
+	for _, dbRun := range runUpdates {
+		queue, ok := queueByJobId[dbRun.JobID]
+		if !ok {
+			return nil, errors.Errorf("run %s is associated with non-existing job %s", dbRun.RunID, dbRun.JobID)
+		}
+		insertRunsDbOp[dbRun.RunID] = &scheduleringester.JobRunDetails{
+			Queue: queue,
+			DbRun: dbRun,
+		}
+	}
+	dbOps = scheduleringester.AppendDbOperation(dbOps, insertRunsDbOp)
+
+	jobRunErrorsEventSequences := make([]*armadaevents.EventSequence, 0, len(jobRunErrors))
+	for _, jobRunError := range jobRunErrors {
+		jobId, err := armadaevents.UlidStringFromProtoUuid(jobRunError.JobId)
+		if err != nil {
+			return nil, err
+		}
+		queue, ok := queueByJobId[jobId]
+		if !ok {
+			return nil, errors.Errorf("jobRunError is associated with non-existing job %s", jobId)
+		}
+		jobSet, ok := jobSetByJobId[jobId]
+		if !ok {
+			return nil, errors.Errorf("jobRunError is associated with non-existing job %s", jobId)
+		}
+		eventSequence := &armadaevents.EventSequence{
+			Queue:      queue,
+			JobSetName: jobSet,
+			Events: []*armadaevents.EventSequence_Event{
+				{
+					Event: &armadaevents.EventSequence_Event_JobRunErrors{JobRunErrors: jobRunError},
+				},
+			},
+		}
+		jobRunErrorsEventSequences = append(jobRunErrorsEventSequences, eventSequence)
+	}
+	insertJobRunErrorsDbOps := instructionConverter.Convert(
+		ctx,
+		&ingest.EventSequencesWithIds{
+			EventSequences: jobRunErrorsEventSequences,
+		},
+	)
+	for _, dbOp := range insertJobRunErrorsDbOps.Ops {
+		dbOps = scheduleringester.AppendDbOperation(dbOps, dbOp)
+	}
+
+	return dbOps, nil
+}
+
+func fixInsertJobsDbOp(dbOp scheduleringester.InsertJobs) scheduleringester.InsertJobs {
+	for _, job := range dbOp {
+		// This field must be non-null when written to postgres.
+		job.SubmitMessage = make([]byte, 0)
+	}
+	return dbOp
+}
+
+func pointerFromValue[T any](v T) *T {
+	return &v
+}
diff --git a/internal/scheduler/testfixtures/testfixtures.go b/internal/scheduler/testfixtures/testfixtures.go
index 294e6ba2..6700b612 100644
--- a/internal/scheduler/testfixtures/testfixtures.go
+++ b/internal/scheduler/testfixtures/testfixtures.go
@@ -2,8 +2,10 @@ package testfixtures
 
 // This file contains test fixtures to be used throughout the tests for this package.
 import (
+	"encoding/binary"
 	"fmt"
 	"math"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -79,14 +81,29 @@ var (
 	JobDb = NewJobDb()
 )
 
+func NewJobDbWithJobs(jobs []*jobdb.Job) *jobdb.JobDb {
+	jobDb := NewJobDb()
+	txn := jobDb.WriteTxn()
+	defer txn.Abort()
+	if err := txn.Upsert(jobs); err != nil {
+		panic(err)
+	}
+	txn.Commit()
+	return jobDb
+}
+
 // NewJobDb returns a new default jobDb with defaults to use in tests.
 func NewJobDb() *jobdb.JobDb {
-	return jobdb.NewJobDbWithSchedulingKeyGenerator(
+	jobDb := jobdb.NewJobDbWithSchedulingKeyGenerator(
 		TestPriorityClasses,
 		TestDefaultPriorityClass,
 		SchedulingKeyGenerator,
 		1024,
 	)
+	// Mock out the clock and uuid provider to ensure consistent ids and timestamps are generated.
+	jobDb.SetClock(NewMockPassiveClock())
+	jobDb.SetUUIDProvider(NewMockUUIDProvider())
+	return jobDb
 }
 
 func IntRange(a, b int) []int {
@@ -901,3 +918,47 @@ func TestExecutor(lastUpdateTime time.Time) *schedulerobjects.Executor {
 		Nodes:          TestCluster(),
 	}
 }
+
+type MockUUIDProvider struct {
+	i  uint64
+	mu sync.Mutex
+}
+
+func NewMockUUIDProvider() *MockUUIDProvider {
+	return &MockUUIDProvider{}
+}
+
+func (p *MockUUIDProvider) New() uuid.UUID {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.i += 1 // Increment before write to avoid using the all-zeros UUID.
+	return UUIDFromInt(p.i)
+}
+
+func UUIDFromInt(i uint64) uuid.UUID {
+	var rv uuid.UUID
+	binary.LittleEndian.PutUint64(rv[:], i)
+	return rv
+}
+
+type MockPassiveClock struct {
+	t  time.Time
+	d  time.Duration
+	mu sync.Mutex
+}
+
+func NewMockPassiveClock() *MockPassiveClock {
+	return &MockPassiveClock{t: time.Unix(0, 0), d: time.Second}
+}
+
+func (p *MockPassiveClock) Now() time.Time {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	rv := p.t
+	p.t = p.t.Add(p.d)
+	return rv
+}
+
+func (p *MockPassiveClock) Since(time.Time) time.Duration {
+	panic("Not implemented")
+}
diff --git a/internal/scheduleringester/dbops_test.go b/internal/scheduleringester/dbops_test.go
index 0dc98973..a3e7e807 100644
--- a/internal/scheduleringester/dbops_test.go
+++ b/internal/scheduleringester/dbops_test.go
@@ -117,11 +117,11 @@ func TestDbOperationOptimisation(t *testing.T) {
 		}},
 		"InsertJobs, InsertRuns": {N: 2, Ops: []DbOperation{
 			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
+			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
+			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
+			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], Queue: testQueueName, JobSet: "set1"}}, // 1
@@ -179,28 +179,28 @@ func TestDbOperationOptimisation(t *testing.T) {
 		}},
 		"MarkRunsSucceeded": {N: 3, Ops: []DbOperation{
 			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
+			InsertRuns{runIds[0]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
 			MarkRunsSucceeded{runIds[0]: time.Time{}},                                                                                // 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
+			InsertRuns{runIds[1]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
 			MarkRunsSucceeded{runIds[1]: time.Time{}},                                                                                // 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]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			InsertRuns{runIds[0]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
 			MarkRunsFailed{runIds[0]: &JobRunFailed{true, true, time.Time{}}},                                                        // 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
+			InsertRuns{runIds[1]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
 			MarkRunsFailed{runIds[1]: &JobRunFailed{true, true, time.Time{}}},                                                        // 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]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
+			InsertRuns{runIds[0]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[0]}}}, // 2
 			MarkRunsRunning{runIds[0]: time.Time{}},                                                                                  // 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
+			InsertRuns{runIds[1]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[0], RunID: runIds[1]}}}, // 3
 			MarkRunsRunning{runIds[1]: time.Time{}},                                                                                  // 3
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}},                                                                // 3
 		}},
@@ -361,7 +361,7 @@ func (db *mockDb) apply(op DbOperation) error {
 	case InsertRuns:
 		n := len(db.Runs)
 		for _, run := range o {
-			run := *run.dbRun // Copy primitive types
+			run := *run.DbRun // Copy primitive types
 			db.Runs[run.RunID] = &run
 		}
 		if len(db.Runs) != n+len(o) {
diff --git a/internal/scheduleringester/instructions_test.go b/internal/scheduleringester/instructions_test.go
index 3f85d96e..37097a0f 100644
--- a/internal/scheduleringester/instructions_test.go
+++ b/internal/scheduleringester/instructions_test.go
@@ -53,13 +53,14 @@ func TestConvertSequence(t *testing.T) {
 		"job run leased": {
 			events: []*armadaevents.EventSequence_Event{f.Leased},
 			expected: []DbOperation{
-				InsertRuns{f.RunIdUuid: &JobRunDetails{queue: f.Queue, dbRun: &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,
 					ScheduledAtPriority: &f.ScheduledAtPriority,
+					Created:             f.BaseTime.UnixNano(),
 					LeasedTimestamp:     &f.BaseTime,
 				}}},
 				UpdateJobQueuedState{f.JobIdString: &JobQueuedStateUpdate{
diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go
index e3709ea3..7be39edf 100644
--- a/internal/scheduleringester/schedulerdb_test.go
+++ b/internal/scheduleringester/schedulerdb_test.go
@@ -51,12 +51,12 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], Queue: testQueueName, JobSet: "set2"},
 			},
 			InsertRuns{
-				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[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]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2], ScheduledAtPriority: &scheduledAtPriorities[0]}},
-				runIds[3]: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3], ScheduledAtPriority: &scheduledAtPriorities[1]}},
+				runIds[2]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[2], RunID: runIds[2], ScheduledAtPriority: &scheduledAtPriorities[0]}},
+				runIds[3]: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobIds[3], RunID: runIds[3], ScheduledAtPriority: &scheduledAtPriorities[1]}},
 			},
 			UpdateJobQueuedState{
 				jobIds[0]: &JobQueuedStateUpdate{Queued: false, QueuedStateVersion: 1},
@@ -135,10 +135,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				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]}},
+				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]: testfixtures.BaseTime,
@@ -163,8 +163,8 @@ func TestWriteOps(t *testing.T) {
 				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]},
 			},
 			InsertRuns{
-				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[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]}},
 			},
 			MarkRunsPending{
 				runIds[0]: testfixtures.BaseTime,
@@ -177,8 +177,8 @@ func TestWriteOps(t *testing.T) {
 				jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]},
 			},
 			InsertRuns{
-				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[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]}},
 			},
 			MarkRunsPreempted{
 				runIds[0]: testfixtures.BaseTime,
@@ -204,10 +204,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				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]}},
+				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]: testfixtures.BaseTime,
@@ -252,10 +252,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				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]}},
+				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, FailureTime: testfixtures.BaseTime},
@@ -271,10 +271,10 @@ func TestWriteOps(t *testing.T) {
 				jobIds[3]: &schedulerdb.Job{JobID: jobIds[3], JobSet: "set2"},
 			},
 			InsertRuns{
-				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]}},
+				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]: testfixtures.BaseTime,
@@ -400,11 +400,11 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		actual := make(InsertRuns)
 		for _, run := range runs {
 			run := run
-			actual[run.RunID] = &JobRunDetails{queue: testQueueName, dbRun: &run}
+			actual[run.RunID] = &JobRunDetails{Queue: testQueueName, DbRun: &run}
 			serials["runs"] = max(serials["runs"], run.Serial)
 			if v, ok := expected[run.RunID]; ok {
-				v.dbRun.Serial = run.Serial
-				v.dbRun.LastModified = run.LastModified
+				v.DbRun.Serial = run.Serial
+				v.DbRun.LastModified = run.LastModified
 			}
 		}
 		assert.Equal(t, expected, actual)
@@ -727,7 +727,7 @@ func TestStore(t *testing.T) {
 			},
 		},
 		InsertRuns{
-			runId: &JobRunDetails{queue: testQueueName, dbRun: &schedulerdb.Run{JobID: jobId, RunID: runId}},
+			runId: &JobRunDetails{Queue: testQueueName, DbRun: &schedulerdb.Run{JobID: jobId, RunID: runId}},
 		},
 	}
 	ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Second)