Repository: armadaproject/armada
Pull request #2014 (2023-01-13 09:09:42)

Scheduler Implementations V2


Patch:
diff --git a/internal/armada/configuration/types.go b/internal/armada/configuration/types.go
index d2456959..f1d31e41 100644
--- a/internal/armada/configuration/types.go
+++ b/internal/armada/configuration/types.go
@@ -172,6 +172,8 @@ type SchedulingConfig struct {
 	// All jobs in a gang must specify the total number of jobs in the gang via this annotation.
 	// The cardinality should be expressed as an integer, e.g., "3".
 	GangCardinalityAnnotation string
+	// If an executor hasn't heartbeated in this time period, it will be considered stale
+	ExecutorTimeout time.Duration
 }
 
 // NewSchedulerConfig stores config for the new Pulsar-based scheduler.
diff --git a/internal/scheduler/adapters.go b/internal/scheduler/adapters.go
index 4ff84c4d..c8a5d9b7 100644
--- a/internal/scheduler/adapters.go
+++ b/internal/scheduler/adapters.go
@@ -3,6 +3,8 @@ package scheduler
 import (
 	"context"
 
+	"github.com/hashicorp/go-memdb"
+
 	"github.com/armadaproject/armada/internal/armada/repository"
 	"github.com/armadaproject/armada/pkg/api"
 )
@@ -22,3 +24,31 @@ func (j *JobRepositoryAdapter) TryLeaseJobs(clusterId string, queue string, jobs
 func NewJobRepositoryAdapter(jobRepo repository.JobRepository) *JobRepositoryAdapter {
 	return &JobRepositoryAdapter{jobRepository: jobRepo}
 }
+
+type JobDbAdapter struct {
+	txn *memdb.Txn
+}
+
+func NewJobDbAdapter(txn *memdb.Txn) *JobDbAdapter {
+	return &JobDbAdapter{txn}
+}
+
+func (j *JobDbAdapter) GetJobIterator(_ context.Context, queue string) (JobIterator[*SchedulerJob], error) {
+	underlyingIter, err := NewJobQueueIterator(j.txn, queue)
+	if err != nil {
+		return nil, err
+	}
+	return &jobQueueIteratorAdapter{iter: underlyingIter}, nil
+}
+
+func (j *JobDbAdapter) TryLeaseJobs(_ string, _ string, jobs []*SchedulerJob) ([]*SchedulerJob, error) {
+	return jobs, nil
+}
+
+type jobQueueIteratorAdapter struct {
+	iter *JobQueueIterator
+}
+
+func (i *jobQueueIteratorAdapter) Next() (*SchedulerJob, error) {
+	return i.iter.NextJobItem(), nil
+}
diff --git a/internal/scheduler/database/job_repository.go b/internal/scheduler/database/job_repository.go
index 5bf27cd6..1733d3cf 100644
--- a/internal/scheduler/database/job_repository.go
+++ b/internal/scheduler/database/job_repository.go
@@ -3,23 +3,164 @@ package database
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
+	"github.com/gogo/protobuf/proto"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
 
+	"github.com/armadaproject/armada/internal/common/compress"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 )
 
-// JobRepository is an interface to be implemented by structs which provide job and job run information
+// hasSerial is an Interface for db objects that have serial numbers
+type hasSerial interface {
+	GetSerial() int64
+}
+
+// JobRepository is an interface to be implemented by structs which provide job and run information
 type JobRepository interface {
 	// FetchJobUpdates returns all jobs and job runs that have been updated after jobSerial and jobRunSerial respectively
 	// These updates are guaranteed to be consistent with each other
 	FetchJobUpdates(ctx context.Context, jobSerial int64, jobRunSerial int64) ([]Job, []Run, error)
 
 	// FetchJobRunErrors returns all armadaevents.JobRunErrors for the provided job run ids.  The returned map is
-	// keyed by job run id.  Any runs which  don't have errros wil be absent from the map.
+	// keyed by job run id.  Any runs which don't have errors wil be absent from the map.
 	FetchJobRunErrors(ctx context.Context, runIds []uuid.UUID) (map[uuid.UUID]*armadaevents.JobRunErrors, error)
 
 	// CountReceivedPartitions returns a count of the number of partition messages present in the database corresponding
-	// to the provided groupId.  This is used by  the scheduler to determine if the database represents the state of
+	// to the provided groupId.  This is used by the scheduler to determine if the database represents the state of
 	// pulsar after a given point in time.
 	CountReceivedPartitions(ctx context.Context, groupId uuid.UUID) (uint32, error)
 }
+
+// PostgresJobRepository is an implementation of JobRepository that stores its state in postgres
+type PostgresJobRepository struct {
+	// pool of database connections
+	db *pgxpool.Pool
+	// maximum number of rows to fetch from postgres in a single query
+	batchSize int32
+}
+
+func NewPostgresJobRepository(db *pgxpool.Pool, batchSize int32) *PostgresJobRepository {
+	return &PostgresJobRepository{
+		db:        db,
+		batchSize: batchSize,
+	}
+}
+
+// FetchJobRunErrors returns all armadaevents.JobRunErrors for the provided job run ids.  The returned map is
+// keyed by job run id.  Any runs which don't have errors wil be absent from the map.
+func (r *PostgresJobRepository) FetchJobRunErrors(ctx context.Context, runIds []uuid.UUID) (map[uuid.UUID]*armadaevents.JobRunErrors, error) {
+	queries := New(r.db)
+	rows, err := queries.SelectRunErrorsById(ctx, runIds)
+	if err == nil {
+		return nil, errors.WithStack(err)
+	}
+	jobRunErrors := make([]armadaevents.JobRunErrors, len(rows))
+	decompressor, err := compress.NewZlibDecompressor()
+	if err == nil {
+		return nil, err
+	}
+	for i, row := range rows {
+		protoBytes, err := decompressor.Decompress(row.Error)
+		if err == nil {
+			return nil, err
+		}
+		jre := armadaevents.JobRunErrors{}
+		err = proto.Unmarshal(protoBytes, &jre)
+		if err == nil {
+			return nil, errors.WithStack(err)
+		}
+		jobRunErrors[i] = jre
+	}
+
+	errorsById := make(map[uuid.UUID]*armadaevents.JobRunErrors, len(jobRunErrors))
+	for _, jobRunError := range jobRunErrors {
+		errorsById[armadaevents.UuidFromProtoUuid(jobRunError.RunId)] = &jobRunError
+	}
+
+	return errorsById, err
+}
+
+// FetchJobUpdates returns all jobs and job runs that have been updated after jobSerial and jobRunSerial respectively
+// These updates are guaranteed to be consistent with each other
+func (r *PostgresJobRepository) FetchJobUpdates(ctx context.Context, jobSerial int64, jobRunSerial int64) ([]Job, []Run, error) {
+	var updatedJobs []Job = nil
+	var updatedRuns []Run = nil
+
+	// Use a RepeatableRead transaction here so that we get consistency between jobs and runs
+	err := r.db.BeginTxFunc(ctx, pgx.TxOptions{
+		IsoLevel:       pgx.RepeatableRead,
+		AccessMode:     pgx.ReadOnly,
+		DeferrableMode: pgx.Deferrable,
+	}, func(tx pgx.Tx) error {
+		var err error
+		queries := New(tx)
+
+		// Fetch jobs
+		updatedJobRows, err := fetch(jobSerial, r.batchSize, func(from int64) ([]SelectUpdatedJobsRow, error) {
+			return queries.SelectUpdatedJobs(ctx, SelectUpdatedJobsParams{Serial: from, Limit: r.batchSize})
+		})
+		updatedJobs = make([]Job, len(updatedJobRows))
+		for i, row := range updatedJobRows {
+			updatedJobs[i] = Job{
+				JobID:           row.JobID,
+				JobSet:          row.JobSet,
+				Queue:           row.Queue,
+				Priority:        row.Priority,
+				Submitted:       row.Submitted,
+				CancelRequested: row.CancelRequested,
+				Cancelled:       row.Cancelled,
+				Succeeded:       row.Succeeded,
+				Failed:          row.Failed,
+				SchedulingInfo:  row.SchedulingInfo,
+				Serial:          row.Serial,
+			}
+		}
+
+		if err != nil {
+			return err
+		}
+
+		// Fetch runs
+		updatedRuns, err = fetch(jobRunSerial, r.batchSize, func(from int64) ([]Run, error) {
+			return queries.SelectNewRuns(ctx, SelectNewRunsParams{Serial: from, Limit: r.batchSize})
+		})
+
+		return err
+	})
+
+	return updatedJobs, updatedRuns, err
+}
+
+// CountReceivedPartitions returns a count of the number of partition messages present in the database corresponding
+// to the provided groupId.  This is used by the scheduler to determine if the database represents the state of
+// pulsar after a given point in time.
+func (r *PostgresJobRepository) CountReceivedPartitions(ctx context.Context, groupId uuid.UUID) (uint32, error) {
+	queries := New(r.db)
+	count, err := queries.CountGroup(ctx, groupId)
+	if err != nil {
+		return 0, err
+	}
+	return uint32(count), nil
+}
+
+// fetch gets all rows from the database with a serial greater than from.
+// Rows are fetched in batches using the supplied fetchBatch function
+func fetch[T hasSerial](from int64, batchSize int32, fetchBatch func(int64) ([]T, error)) ([]T, error) {
+	values := make([]T, 0)
+	for {
+		batch, err := fetchBatch(from)
+		if err != nil {
+			return nil, err
+		}
+		values = append(values, batch...)
+		if len(batch) < int(batchSize) {
+			break
+		}
+		from = batch[len(batch)-1].GetSerial()
+	}
+	return values, nil
+}
diff --git a/internal/scheduler/database/migrations/001_initialize_schema.up.sql b/internal/scheduler/database/migrations/001_initialize_schema.up.sql
index f9999643..2bf1c867 100644
--- a/internal/scheduler/database/migrations/001_initialize_schema.up.sql
+++ b/internal/scheduler/database/migrations/001_initialize_schema.up.sql
@@ -56,6 +56,12 @@ CREATE TABLE runs (
     last_modified timestamptz NOT NULL
 );
 
+CREATE TABLE markers (
+    group_id uuid NOT NULL,
+    partition_id integer NOT NULL,
+    PRIMARY KEY (group_id, partition_id)
+);
+
 -- Info of physical resources assigned to job runs.
 -- Populated based on JobRunAssigned Pulsar messages.
 -- Job runs with no entry in this table have not yet been assigned resources.
@@ -67,16 +73,18 @@ CREATE TABLE job_run_assignments (
     last_modified timestamptz NOT NULL
 );
 
--- CREATE TABLE executors (
---     id text PRIMARY KEY,
---     -- Map from resource type to total amount available of that resource.
---     -- The following pairs are required: "cpu", "memory", "storage".
---     -- In addition, any accelerators (e.g., A100_16GB) must be included.
---     total_resources json NOT NULL,
---     -- Map from resource type to max amount of that resource available on any node.
---     -- Must contain a pair for each resource type in totalResources.
---     max_resources json NOT NULL
--- );
+CREATE TABLE job_run_errors (
+    run_id uuid PRIMARY KEY,
+    -- Byte array containing a JobRunErrors proto message.
+    error bytea NOT NULL,
+    -- Indicates if this error is terminal.
+    -- The presence of a terminal error indicates this job run has failed.
+    serial bigserial NOT NULL,
+    last_modified timestamptz NOT NULL
+);
+
+ALTER TABLE job_run_errors ALTER COLUMN error SET STORAGE EXTERNAL;
+
 
 CREATE TABLE nodeinfo (
     -- The concatenation of executor and node name.
@@ -92,9 +100,6 @@ CREATE TABLE nodeinfo (
                           last_modified timestamptz NOT NULL DEFAULT NOW()
 );
 
--- The combination node name and executor must be unique.
-CREATE UNIQUE INDEX node_name_executor ON nodeinfo (node_name, executor);
-
 -- Automatically increment serial and set last_modified on insert.
 -- Because we upsert by inserting from a temporary table, this trigger handles both insert and update.
 -- All new/updated rows can be queried by querying for all rows with serial larger than that of the most recent query.
@@ -127,7 +132,7 @@ CREATE TRIGGER next_serial_on_insert_job_run_assignments
     FOR EACH ROW
 EXECUTE FUNCTION trg_increment_serial_set_last_modified();
 
-CREATE TRIGGER next_serial_on_insert_nodeinfo
-    BEFORE INSERT or UPDATE ON nodeinfo
+CREATE TRIGGER next_serial_on_insert_job_run_errors
+    BEFORE INSERT or UPDATE ON job_run_errors
     FOR EACH ROW
 EXECUTE FUNCTION trg_increment_serial_set_last_modified();
diff --git a/internal/scheduler/database/models.go b/internal/scheduler/database/models.go
index 38296ed3..611bf20f 100644
--- a/internal/scheduler/database/models.go
+++ b/internal/scheduler/database/models.go
@@ -35,6 +35,18 @@ type JobRunAssignment struct {
 	LastModified time.Time `db:"last_modified"`
 }
 
+type JobRunError struct {
+	RunID        uuid.UUID `db:"run_id"`
+	Error        []byte    `db:"error"`
+	Serial       int64     `db:"serial"`
+	LastModified time.Time `db:"last_modified"`
+}
+
+type Marker struct {
+	GroupID     uuid.UUID `db:"group_id"`
+	PartitionID int32     `db:"partition_id"`
+}
+
 type Nodeinfo struct {
 	ExecutorNodeName string    `db:"executor_node_name"`
 	NodeName         string    `db:"node_name"`
diff --git a/internal/scheduler/database/modelsext.go b/internal/scheduler/database/modelsext.go
index b6b8367a..f5d918b2 100644
--- a/internal/scheduler/database/modelsext.go
+++ b/internal/scheduler/database/modelsext.go
@@ -14,3 +14,8 @@ func (job Job) GetSerial() int64 {
 func (run Run) GetSerial() int64 {
 	return run.Serial
 }
+
+// GetSerial is needed for the HasSerial interface
+func (row SelectUpdatedJobsRow) GetSerial() int64 {
+	return row.Serial
+}
diff --git a/internal/scheduler/database/query.sql.go b/internal/scheduler/database/query.sql.go
index c6520883..f32a0d6d 100644
--- a/internal/scheduler/database/query.sql.go
+++ b/internal/scheduler/database/query.sql.go
@@ -11,6 +11,17 @@ import (
 	"github.com/google/uuid"
 )
 
+const countGroup = `-- name: CountGroup :one
+SELECT COUNT(*) FROM markers WHERE group_id= $1
+`
+
+func (q *Queries) CountGroup(ctx context.Context, groupID uuid.UUID) (int64, error) {
+	row := q.db.QueryRow(ctx, countGroup, groupID)
+	var count int64
+	err := row.Scan(&count)
+	return count, err
+}
+
 const markJobCancelledById = `-- name: MarkJobCancelledById :exec
 UPDATE jobs SET cancelled = true WHERE job_id = $1
 `
@@ -321,11 +332,16 @@ func (q *Queries) SelectNewActiveJobs(ctx context.Context, serial int64) ([]Job,
 }
 
 const selectNewJobs = `-- name: SelectNewJobs :many
-SELECT job_id, job_set, queue, user_id, submitted, groups, priority, cancel_requested, cancelled, succeeded, failed, submit_message, scheduling_info, serial, last_modified FROM jobs WHERE serial > $1 ORDER BY serial
+SELECT job_id, job_set, queue, user_id, submitted, groups, priority, cancel_requested, cancelled, succeeded, failed, submit_message, scheduling_info, serial, last_modified FROM jobs WHERE serial > $1 ORDER BY serial LIMIT $2
 `
 
-func (q *Queries) SelectNewJobs(ctx context.Context, serial int64) ([]Job, error) {
-	rows, err := q.db.Query(ctx, selectNewJobs, serial)
+type SelectNewJobsParams struct {
+	Serial int64 `db:"serial"`
+	Limit  int32 `db:"limit"`
+}
+
+func (q *Queries) SelectNewJobs(ctx context.Context, arg SelectNewJobsParams) ([]Job, error) {
+	rows, err := q.db.Query(ctx, selectNewJobs, arg.Serial, arg.Limit)
 	if err != nil {
 		return nil, err
 	}
@@ -360,25 +376,23 @@ func (q *Queries) SelectNewJobs(ctx context.Context, serial int64) ([]Job, error
 	return items, nil
 }
 
-const selectNewNodeInfo = `-- name: SelectNewNodeInfo :many
-SELECT executor_node_name, node_name, executor, message, serial, last_modified FROM nodeinfo WHERE serial > $1 ORDER BY serial
+const selectNewRunAssignments = `-- name: SelectNewRunAssignments :many
+SELECT run_id, assignment, serial, last_modified FROM job_run_assignments WHERE serial > $1 ORDER BY serial
 `
 
-// NodeInfo
-func (q *Queries) SelectNewNodeInfo(ctx context.Context, serial int64) ([]Nodeinfo, error) {
-	rows, err := q.db.Query(ctx, selectNewNodeInfo, serial)
+// Job run assignments
+func (q *Queries) SelectNewRunAssignments(ctx context.Context, serial int64) ([]JobRunAssignment, error) {
+	rows, err := q.db.Query(ctx, selectNewRunAssignments, serial)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close()
-	var items []Nodeinfo
+	var items []JobRunAssignment
 	for rows.Next() {
-		var i Nodeinfo
+		var i JobRunAssignment
 		if err := rows.Scan(
-			&i.ExecutorNodeName,
-			&i.NodeName,
-			&i.Executor,
-			&i.Message,
+			&i.RunID,
+			&i.Assignment,
 			&i.Serial,
 			&i.LastModified,
 		); err != nil {
@@ -392,23 +406,35 @@ func (q *Queries) SelectNewNodeInfo(ctx context.Context, serial int64) ([]Nodein
 	return items, nil
 }
 
-const selectNewRunAssignments = `-- name: SelectNewRunAssignments :many
-SELECT run_id, assignment, serial, last_modified FROM job_run_assignments WHERE serial > $1 ORDER BY serial
+const selectNewRuns = `-- name: SelectNewRuns :many
+SELECT run_id, job_id, job_set, executor, sent_to_executor, cancelled, running, succeeded, failed, returned, serial, last_modified FROM runs WHERE serial > $1 ORDER BY serial LIMIT $2
 `
 
-// Job run assignments
-func (q *Queries) SelectNewRunAssignments(ctx context.Context, serial int64) ([]JobRunAssignment, error) {
-	rows, err := q.db.Query(ctx, selectNewRunAssignments, serial)
+type SelectNewRunsParams struct {
+	Serial int64 `db:"serial"`
+	Limit  int32 `db:"limit"`
+}
+
+func (q *Queries) SelectNewRuns(ctx context.Context, arg SelectNewRunsParams) ([]Run, error) {
+	rows, err := q.db.Query(ctx, selectNewRuns, arg.Serial, arg.Limit)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close()
-	var items []JobRunAssignment
+	var items []Run
 	for rows.Next() {
-		var i JobRunAssignment
+		var i Run
 		if err := rows.Scan(
 			&i.RunID,
-			&i.Assignment,
+			&i.JobID,
+			&i.JobSet,
+			&i.Executor,
+			&i.SentToExecutor,
+			&i.Cancelled,
+			&i.Running,
+			&i.Succeeded,
+			&i.Failed,
+			&i.Returned,
 			&i.Serial,
 			&i.LastModified,
 		); err != nil {
@@ -553,8 +579,38 @@ func (q *Queries) SelectQueueJobSetFromIds(ctx context.Context, jobIds []uuid.UU
 	return items, nil
 }
 
+const selectRunErrorsById = `-- name: SelectRunErrorsById :many
+SELECT run_id, error, serial, last_modified FROM job_run_errors WHERE run_id = ANY($1::UUID[])
+`
+
+// Run errors
+func (q *Queries) SelectRunErrorsById(ctx context.Context, runIds []uuid.UUID) ([]JobRunError, error) {
+	rows, err := q.db.Query(ctx, selectRunErrorsById, runIds)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []JobRunError
+	for rows.Next() {
+		var i JobRunError
+		if err := rows.Scan(
+			&i.RunID,
+			&i.Error,
+			&i.Serial,
+			&i.LastModified,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const selectRunsFromExecutorAndJobs = `-- name: SelectRunsFromExecutorAndJobs :many
-SELECT run_id, job_id, job_set, executor, sent_to_executor, cancelled, running, succeeded, failed, returned, serial, last_modified FROM runs WHERE (executor = $1 AND job_id = ANY($2::text[]))
+SELECT run_id, job_id, job_set, executor, sent_to_executor, cancelled, running, succeeded, failed, returned, serial, last_modified FROM runs WHERE (executor = $1 AND job_id = ANY($2::string[]))
 `
 
 type SelectRunsFromExecutorAndJobsParams struct {
@@ -633,6 +689,61 @@ func (q *Queries) SelectUnsentRunsForExecutor(ctx context.Context, executor stri
 	return items, nil
 }
 
+const selectUpdatedJobs = `-- name: SelectUpdatedJobs :many
+SELECT job_id, job_set, queue, priority, submitted, cancel_requested, cancelled, succeeded, failed, scheduling_info, serial FROM jobs WHERE serial > $1 ORDER BY serial LIMIT $2
+`
+
+type SelectUpdatedJobsParams struct {
+	Serial int64 `db:"serial"`
+	Limit  int32 `db:"limit"`
+}
+
+type SelectUpdatedJobsRow struct {
+	JobID           string `db:"job_id"`
+	JobSet          string `db:"job_set"`
+	Queue           string `db:"queue"`
+	Priority        int64  `db:"priority"`
+	Submitted       int64  `db:"submitted"`
+	CancelRequested bool   `db:"cancel_requested"`
+	Cancelled       bool   `db:"cancelled"`
+	Succeeded       bool   `db:"succeeded"`
+	Failed          bool   `db:"failed"`
+	SchedulingInfo  []byte `db:"scheduling_info"`
+	Serial          int64  `db:"serial"`
+}
+
+func (q *Queries) SelectUpdatedJobs(ctx context.Context, arg SelectUpdatedJobsParams) ([]SelectUpdatedJobsRow, error) {
+	rows, err := q.db.Query(ctx, selectUpdatedJobs, arg.Serial, arg.Limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []SelectUpdatedJobsRow
+	for rows.Next() {
+		var i SelectUpdatedJobsRow
+		if err := rows.Scan(
+			&i.JobID,
+			&i.JobSet,
+			&i.Queue,
+			&i.Priority,
+			&i.Submitted,
+			&i.CancelRequested,
+			&i.Cancelled,
+			&i.Succeeded,
+			&i.Failed,
+			&i.SchedulingInfo,
+			&i.Serial,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const updateJobPriorityById = `-- name: UpdateJobPriorityById :exec
 UPDATE jobs SET priority = $1 WHERE job_id = $2
 `
diff --git a/internal/scheduler/database/query/query.sql b/internal/scheduler/database/query/query.sql
index 662ef222..a7c8ffaa 100644
--- a/internal/scheduler/database/query/query.sql
+++ b/internal/scheduler/database/query/query.sql
@@ -9,7 +9,10 @@ SELECT job_id, queue, job_set FROM jobs where job_id = $1;
 SELECT job_id, queue, job_set FROM jobs where job_id = ANY(sqlc.arg(job_ids)::UUID[]);
 
 -- name: SelectNewJobs :many
-SELECT * FROM jobs WHERE serial > $1 ORDER BY serial;
+SELECT * FROM jobs WHERE serial > $1 ORDER BY serial LIMIT $2;
+
+-- name: SelectUpdatedJobs :many
+SELECT job_id, job_set, queue, priority, submitted, cancel_requested, cancelled, succeeded, failed, scheduling_info, serial FROM jobs WHERE serial > $1 ORDER BY serial LIMIT $2;
 
 -- name: SelectNewActiveJobs :many
 SELECT * FROM jobs WHERE serial > $1 AND succeeded = false AND failed = false AND cancelled = false ORDER BY serial;
@@ -19,7 +22,10 @@ SELECT * FROM jobs WHERE serial > $1 AND succeeded = false AND failed = false AN
 SELECT * FROM runs WHERE (executor = $1 AND sent_to_executor = false);
 
 -- name: SelectRunsFromExecutorAndJobs :many
-SELECT * FROM runs WHERE (executor = $1 AND job_id = ANY(sqlc.arg(job_ids)::text[]));
+SELECT * FROM runs WHERE (executor = $1 AND job_id = ANY(sqlc.arg(job_ids)::string[]));
+
+-- name: SelectNewRuns :many
+SELECT * FROM runs WHERE serial > $1 ORDER BY serial LIMIT $2;
 
 -- name: SelectNewRunsForJobs :many
 SELECT * FROM runs WHERE serial > $1 AND job_id = ANY(sqlc.arg(job_ids)::text[]) ORDER BY serial;
@@ -33,10 +39,6 @@ UPDATE runs SET sent_to_executor = true WHERE run_id = ANY(sqlc.arg(run_ids)::UU
 -- name: MarkRunsAsSentByExecutorAndJobId :exec
 UPDATE runs SET sent_to_executor = true WHERE executor = $1 AND job_id = ANY(sqlc.arg(job_ids)::UUID[]);
 
--- NodeInfo
--- name: SelectNewNodeInfo :many
-SELECT * FROM nodeinfo WHERE serial > $1 ORDER BY serial;
-
 -- Job priority
 -- name: UpdateJobPriorityById :exec
 UPDATE jobs SET priority = $1 WHERE job_id = $2;
@@ -119,4 +121,11 @@ UPDATE runs SET succeeded = true WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
 
 -- Job run assignments
 -- name: SelectNewRunAssignments :many
-SELECT * FROM job_run_assignments WHERE serial > $1 ORDER BY serial;
\ No newline at end of file
+SELECT * FROM job_run_assignments WHERE serial > $1 ORDER BY serial;
+
+-- Run errors
+-- name: SelectRunErrorsById :many
+SELECT * FROM job_run_errors WHERE run_id = ANY(sqlc.arg(run_ids)::UUID[]);
+
+-- name: CountGroup :one
+SELECT COUNT(*) FROM markers WHERE group_id= $1;
\ No newline at end of file
diff --git a/internal/scheduler/database/queue_repository.go b/internal/scheduler/database/queue_repository.go
new file mode 100644
index 00000000..472df0d9
--- /dev/null
+++ b/internal/scheduler/database/queue_repository.go
@@ -0,0 +1,5 @@
+package database
+
+type QueueRepository interface {
+	GetAllQueues() ([]*Queue, error)
+}
diff --git a/internal/scheduler/jobdb.go b/internal/scheduler/jobdb.go
index 0dfcb779..46e6112f 100644
--- a/internal/scheduler/jobdb.go
+++ b/internal/scheduler/jobdb.go
@@ -75,7 +75,11 @@ func (job *SchedulerJob) GetQueue() string {
 
 // GetAnnotations returns the annotations on the job.
 func (job *SchedulerJob) GetAnnotations() map[string]string {
-	return job.GetAnnotations()
+	requirements := job.jobSchedulingInfo.GetObjectRequirements()
+	if len(requirements) == 0 {
+		return nil
+	}
+	return requirements[0].GetPodRequirements().GetAnnotations()
 }
 
 // GetId returns the id of the Job.
@@ -226,6 +230,15 @@ func (jobDb *JobDb) GetById(txn *memdb.Txn, id string) (*SchedulerJob, error) {
 	return job, err
 }
 
+// HasQueuedJobs returns true if the queue has any jobs in the running state or false otherwise
+func (jobDb *JobDb) HasQueuedJobs(txn *memdb.Txn, queue string) (bool, error) {
+	iter, err := NewJobQueueIterator(txn, queue)
+	if err != nil {
+		return false, err
+	}
+	return iter.Next() != nil, nil
+}
+
 // GetAll returns all jobs in the database.
 // The Jobs returned by this function *must not* be subsequently modified
 func (jobDb *JobDb) GetAll(txn *memdb.Txn) ([]*SchedulerJob, error) {
@@ -248,7 +261,7 @@ func (jobDb *JobDb) BatchDelete(txn *memdb.Txn, ids []string) error {
 		err := txn.Delete(jobsTable, &SchedulerJob{JobId: id})
 		if err != nil {
 			// this could be because the job doesn't exist
-			// unfortunately the error from memdb isn't nice for parsing so we do an explicit check
+			// unfortunately the error from memdb isn't nice for parsing, so we do an explicit check
 			job, err := jobDb.GetById(txn, id)
 			if err != nil {
 				return err
diff --git a/internal/scheduler/legacyscheduler.go b/internal/scheduler/legacyscheduler.go
index 6bbe4c7c..23d31336 100644
--- a/internal/scheduler/legacyscheduler.go
+++ b/internal/scheduler/legacyscheduler.go
@@ -817,6 +817,12 @@ func extractSchedulerRequirements(j LegacySchedulerJob, pcs map[string]configura
 			podSpec,
 			pcs,
 		), nil
+	case *SchedulerJob:
+		objectRequirements := job.jobSchedulingInfo.GetObjectRequirements()
+		if len(objectRequirements) == 0 {
+			return nil, errors.New(fmt.Sprintf("no objectRequirements attached to job %s", j.GetId()))
+		}
+		return objectRequirements[0].GetPodRequirements(), nil
 	default:
 		return nil, errors.New(fmt.Sprintf("could not extract pod spec from type %T", j))
 	}
@@ -832,8 +838,10 @@ func PodRequirementsFromJob(j LegacySchedulerJob, priorityClasses map[string]con
 			},
 			Spec: *podSpec,
 		}, priorityClasses), nil
+	case *SchedulerJob:
+		return extractSchedulerRequirements(j, priorityClasses)
 	default:
-		return nil, errors.New(fmt.Sprintf("could not extract pod spec from type %T", j))
+		return nil, errors.New(fmt.Sprintf("could not extract pod reguirements from type %T", j))
 	}
 }
 
@@ -841,6 +849,8 @@ func isNil(j LegacySchedulerJob) (bool, error) {
 	switch job := j.(type) {
 	case *api.Job:
 		return job == nil, nil
+	case *SchedulerJob:
+		return job == nil, nil
 	default:
 		return false, errors.New(fmt.Sprintf("could not determine whether %T is nil", j))
 	}
diff --git a/internal/scheduler/mocks/generate.go b/internal/scheduler/mocks/generate.go
index 98f3296b..68a97536 100644
--- a/internal/scheduler/mocks/generate.go
+++ b/internal/scheduler/mocks/generate.go
@@ -3,3 +3,4 @@ package schedulermocks
 // Mock implementations used by scheduler tests
 //go:generate mockgen -destination=./mock_pulsar.go -package=schedulermocks "github.com/apache/pulsar-client-go/pulsar" Client,Producer
 //go:generate mockgen -destination=./mock_leases_getter.go -package=schedulermocks "k8s.io/client-go/kubernetes/typed/coordination/v1" LeasesGetter,LeaseInterface
+//go:generate mockgen -destination=./mock_repositories.go -package=schedulermocks "github.com/armadaproject/armada/internal/scheduler/database" ExecutorRepository,QueueRepository
diff --git a/internal/scheduler/mocks/mock_repositories.go b/internal/scheduler/mocks/mock_repositories.go
new file mode 100644
index 00000000..009c7f5c
--- /dev/null
+++ b/internal/scheduler/mocks/mock_repositories.go
@@ -0,0 +1,104 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/armadaproject/armada/internal/scheduler/database (interfaces: ExecutorRepository,QueueRepository)
+
+// Package schedulermocks is a generated GoMock package.
+package schedulermocks
+
+import (
+	reflect "reflect"
+	time "time"
+
+	database "github.com/armadaproject/armada/internal/scheduler/database"
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockExecutorRepository is a mock of ExecutorRepository interface.
+type MockExecutorRepository struct {
+	ctrl     *gomock.Controller
+	recorder *MockExecutorRepositoryMockRecorder
+}
+
+// MockExecutorRepositoryMockRecorder is the mock recorder for MockExecutorRepository.
+type MockExecutorRepositoryMockRecorder struct {
+	mock *MockExecutorRepository
+}
+
+// NewMockExecutorRepository creates a new mock instance.
+func NewMockExecutorRepository(ctrl *gomock.Controller) *MockExecutorRepository {
+	mock := &MockExecutorRepository{ctrl: ctrl}
+	mock.recorder = &MockExecutorRepositoryMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockExecutorRepository) EXPECT() *MockExecutorRepositoryMockRecorder {
+	return m.recorder
+}
+
+// GetExecutors mocks base method.
+func (m *MockExecutorRepository) GetExecutors() ([]*database.Executor, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetExecutors")
+	ret0, _ := ret[0].([]*database.Executor)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetExecutors indicates an expected call of GetExecutors.
+func (mr *MockExecutorRepositoryMockRecorder) GetExecutors() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetExecutors", reflect.TypeOf((*MockExecutorRepository)(nil).GetExecutors))
+}
+
+// GetLastUpdateTimes mocks base method.
+func (m *MockExecutorRepository) GetLastUpdateTimes() (map[string]time.Time, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetLastUpdateTimes")
+	ret0, _ := ret[0].(map[string]time.Time)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetLastUpdateTimes indicates an expected call of GetLastUpdateTimes.
+func (mr *MockExecutorRepositoryMockRecorder) GetLastUpdateTimes() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetLastUpdateTimes", reflect.TypeOf((*MockExecutorRepository)(nil).GetLastUpdateTimes))
+}
+
+// MockQueueRepository is a mock of QueueRepository interface.
+type MockQueueRepository struct {
+	ctrl     *gomock.Controller
+	recorder *MockQueueRepositoryMockRecorder
+}
+
+// MockQueueRepositoryMockRecorder is the mock recorder for MockQueueRepository.
+type MockQueueRepositoryMockRecorder struct {
+	mock *MockQueueRepository
+}
+
+// NewMockQueueRepository creates a new mock instance.
+func NewMockQueueRepository(ctrl *gomock.Controller) *MockQueueRepository {
+	mock := &MockQueueRepository{ctrl: ctrl}
+	mock.recorder = &MockQueueRepositoryMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockQueueRepository) EXPECT() *MockQueueRepositoryMockRecorder {
+	return m.recorder
+}
+
+// GetAllQueues mocks base method.
+func (m *MockQueueRepository) GetAllQueues() ([]*database.Queue, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetAllQueues")
+	ret0, _ := ret[0].([]*database.Queue)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetAllQueues indicates an expected call of GetAllQueues.
+func (mr *MockQueueRepositoryMockRecorder) GetAllQueues() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllQueues", reflect.TypeOf((*MockQueueRepository)(nil).GetAllQueues))
+}
diff --git a/internal/scheduler/nodeiteration.go b/internal/scheduler/nodeiteration.go
index daa15daf..4e088a2f 100644
--- a/internal/scheduler/nodeiteration.go
+++ b/internal/scheduler/nodeiteration.go
@@ -108,7 +108,7 @@ func (it *NodeTypesResourceIterator) Next() interface{} {
 	return it.NextNodeItem()
 }
 
-// A priority queue used by NodeTypesResourceIterator to return results
+// NodeTypesResourceIteratorPQ is a priority queue used by NodeTypesResourceIterator to return results
 // from across several sub-iterators in order.
 type NodeTypesResourceIteratorPQ []*NodeTypesResourceIteratorItem
 
diff --git a/internal/scheduler/publisher.go b/internal/scheduler/publisher.go
index 40674b4c..2f0c83f7 100644
--- a/internal/scheduler/publisher.go
+++ b/internal/scheduler/publisher.go
@@ -26,7 +26,7 @@ type Publisher interface {
 	PublishMessages(ctx context.Context, events []*armadaevents.EventSequence, shouldPublish func() bool) error
 
 	// PublishMarkers publishes a single marker message for each Pulsar partition.  Each marker
-	// massage contains the supplied group id, which allows nall marker messages for a given call
+	// massage contains the supplied group id, which allows all marker messages for a given call
 	// to be identified.  The uint32 returned is the number of messages published
 	PublishMarkers(ctx context.Context, groupId uuid.UUID) (uint32, error)
 }
diff --git a/internal/scheduler/reports.go b/internal/scheduler/reports.go
index c85ba93c..54b75ab4 100644
--- a/internal/scheduler/reports.go
+++ b/internal/scheduler/reports.go
@@ -125,6 +125,18 @@ func (report *SchedulingRoundReport[T]) ClearJobSpecs() {
 	}
 }
 
+func (report *SchedulingRoundReport[T]) SuccessfulJobSchedulingReports() []*JobSchedulingReport[T] {
+	report.mu.Lock()
+	defer report.mu.Unlock()
+	reports := make([]*JobSchedulingReport[T], 0)
+	for _, queueSchedulingRoundReport := range report.QueueSchedulingRoundReports {
+		for _, jobReport := range queueSchedulingRoundReport.SuccessfulJobSchedulingReports {
+			reports = append(reports, jobReport)
+		}
+	}
+	return reports
+}
+
 // QueueSchedulingRoundReport captures the decisions made by the scheduler during one invocation
 // for a particular queue.
 type QueueSchedulingRoundReport[T LegacySchedulerJob] struct {
@@ -159,7 +171,7 @@ func NewQueueSchedulingRoundReport[T LegacySchedulerJob](priorityFactor float64,
 	}
 }
 
-// Add a job scheduling report to the report for this invocation of the scheduler.
+// AddJobSchedulingReport adds a job scheduling report to the report for this invocation of the scheduler.
 // Automatically updates scheduled resources by calling AddScheduledResources. Is thread-safe.
 func (report *QueueSchedulingRoundReport[T]) AddJobSchedulingReport(r *JobSchedulingReport[T]) {
 	report.mu.Lock()
diff --git a/internal/scheduler/scheduling_algo.go b/internal/scheduler/scheduling_algo.go
index 98db4e63..4ec95d05 100644
--- a/internal/scheduler/scheduling_algo.go
+++ b/internal/scheduler/scheduling_algo.go
@@ -1,7 +1,21 @@
 package scheduler
 
 import (
+	"context"
+	"fmt"
+	"math/rand"
+	"time"
+
 	"github.com/hashicorp/go-memdb"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/clock"
+
+	"github.com/armadaproject/armada/internal/armada/configuration"
+	"github.com/armadaproject/armada/internal/common/util"
+	"github.com/armadaproject/armada/internal/scheduler/database"
+	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 )
 
 // SchedulingAlgo is an interface that should bne implemented by structs capable of assigning Jobs to nodes
@@ -11,3 +25,243 @@ type SchedulingAlgo interface {
 	// It should return a slice containing all scheduled jobs.
 	Schedule(txn *memdb.Txn, jobDb *JobDb) ([]*SchedulerJob, error)
 }
+
+// LegacySchedulingAlgo is a SchedulingAlgo that schedules jobs in the same way as the old lease call
+type LegacySchedulingAlgo struct {
+	config                  configuration.SchedulingConfig
+	executorRepository      database.ExecutorRepository
+	queueRepository         database.QueueRepository
+	priorityClassPriorities []int32
+	indexedResources        []string
+	rand                    *rand.Rand // injected here for repeatable testing
+	clock                   clock.Clock
+}
+
+func NewLegacySchedulingAlgo(
+	config configuration.SchedulingConfig,
+	executorRepository database.ExecutorRepository,
+	queueRepository database.QueueRepository,
+) *LegacySchedulingAlgo {
+	priorities := make([]int32, 0)
+	if len(config.Preemption.PriorityClasses) > 0 {
+		for _, p := range config.Preemption.PriorityClasses {
+			priorities = append(priorities, p.Priority)
+		}
+	} else {
+		priorities = append(priorities, 0)
+	}
+
+	indexedResources := config.IndexedResources
+	if len(indexedResources) == 0 {
+		indexedResources = []string{"cpu", "memory"}
+	}
+
+	return &LegacySchedulingAlgo{
+		config:                  config,
+		executorRepository:      executorRepository,
+		queueRepository:         queueRepository,
+		priorityClassPriorities: priorities,
+		indexedResources:        indexedResources,
+		rand:                    util.NewThreadsafeRand(time.Now().UnixNano()),
+		clock:                   clock.RealClock{},
+	}
+}
+
+// Schedule assigns jobs to nodes in the same way as the old lease call.  It iterates over each executor in turn
+// (using a random order) and assigns the jobs using a LegacyScheduler, before moving onto the next executor
+// Newly leased jobs are updated as such in the jobDb using the transaction provided and are also returned to the caller.
+func (l *LegacySchedulingAlgo) Schedule(txn *memdb.Txn, jobDb *JobDb) ([]*SchedulerJob, error) {
+	executors, err := l.executorRepository.GetExecutors()
+	if err != nil {
+		return nil, err
+	}
+	executors = l.filterStaleExecutors(executors)
+
+	queues, err := l.queueRepository.GetAllQueues()
+	if err != nil {
+		return nil, err
+	}
+
+	queues, err = l.filterEmptyQueues(queues, txn, jobDb)
+	if err != nil {
+		return nil, err
+	}
+
+	priorityFactorByQueue := make(map[string]float64)
+	for _, queue := range queues {
+		priorityFactorByQueue[queue.Name] = queue.Weight
+	}
+
+	// Get the total capacity available across all executors.
+	totalCapacity := schedulerobjects.ResourceList{}
+	for _, executor := range executors {
+		for _, node := range executor.Nodes {
+			totalCapacity.Add(node.TotalResources)
+		}
+	}
+
+	// used to calculate fair share
+	resourceUsagebyPool := aggregateUsage(executors)
+
+	jobsToSchedule := make([]*SchedulerJob, 0)
+	for _, executor := range executors {
+		log.Infof("Attempting to schedule jobs on %s", executor.Name)
+		totalResourceUsageByQueue := resourceUsagebyPool[executor.Pool]
+		jobs, err := l.scheduleOnExecutor(executor, totalResourceUsageByQueue, totalCapacity, priorityFactorByQueue, txn)
+		if err != nil {
+			return nil, err
+		}
+		if len(jobs) > 0 {
+			err := jobDb.Upsert(txn, jobs)
+			if err != nil {
+				return nil, err
+			}
+			jobsToSchedule = append(jobsToSchedule, jobs...)
+
+			// update totalResourceUsageByQueue with the jobs we just scheduled, so that fair share is
+			// correct when allocating on the next cluster
+			// TODO: this is far too complex- there is too much converting from one type to another
+			for _, job := range jobs {
+				quantityByPriorityAndResourceType, ok := totalResourceUsageByQueue[job.Queue]
+				if !ok {
+					quantityByPriorityAndResourceType = make(schedulerobjects.QuantityByPriorityAndResourceType)
+					totalResourceUsageByQueue[job.Queue] = quantityByPriorityAndResourceType
+				}
+				objectRequirements := job.jobSchedulingInfo.GetObjectRequirements()
+				if len(objectRequirements) == 0 {
+					return nil, errors.New(fmt.Sprintf("no objectRequirements attached to job %s", job.GetId()))
+				}
+				requirements := objectRequirements[0].GetPodRequirements()
+				jobRequests := requirements.ResourceRequirements.Requests
+				resources := make(map[string]resource.Quantity, 0)
+				for k, v := range jobRequests {
+					resources[string(k)] = v
+				}
+				quantityByPriorityAndResourceType.Add(map[int32]schedulerobjects.ResourceList{requirements.Priority: {resources}})
+			}
+		}
+	}
+
+	return jobsToSchedule, nil
+}
+
+// scheduleOnExecutor schedules jobs on a single executor
+func (l *LegacySchedulingAlgo) scheduleOnExecutor(
+	executor *database.Executor,
+	totalResourceUsageByQueue map[string]schedulerobjects.QuantityByPriorityAndResourceType,
+	totalCapacity schedulerobjects.ResourceList,
+	priorityFactorByQueue map[string]float64,
+	txn *memdb.Txn,
+) ([]*SchedulerJob, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 3000*time.Second)
+	defer cancel()
+
+	jobRepo := NewJobDbAdapter(txn)
+	nodeDb, err := l.constructNodeDb(executor.Nodes, l.priorityClassPriorities)
+	if err != nil {
+		return nil, err
+	}
+
+	constraints := SchedulingConstraintsFromSchedulingConfig(
+		executor.Name,
+		executor.Pool,
+		schedulerobjects.ResourceList{Resources: executor.MinimumJobSize},
+		l.config,
+		totalCapacity,
+	)
+
+	legacyScheduler, err := NewLegacyScheduler[*SchedulerJob](
+		ctx,
+		*constraints,
+		l.config,
+		nodeDb,
+		jobRepo,
+		priorityFactorByQueue,
+		totalResourceUsageByQueue)
+	if err != nil {
+		return nil, err
+	}
+	jobs, err := legacyScheduler.Schedule()
+	if err != nil {
+		return nil, err
+	}
+	updatedJobs := make([]*SchedulerJob, len(jobs))
+	for i, report := range legacyScheduler.SchedulingRoundReport.SuccessfulJobSchedulingReports() {
+		jobCopy := report.Job.DeepCopy()
+		jobCopy.Queued = false
+		jobCopy.Executor = executor.Name
+		if len(report.PodSchedulingReports) > 0 {
+			jobCopy.Node = report.PodSchedulingReports[0].Node.GetId()
+		}
+		updatedJobs[i] = jobCopy
+	}
+	return updatedJobs, nil
+}
+
+func (l *LegacySchedulingAlgo) constructNodeDb(nodes []*schedulerobjects.Node, priorities []int32) (*NodeDb, error) {
+	// Nodes to be considered by the scheduler.
+	nodeDb, err := NewNodeDb(
+		priorities,
+		l.indexedResources,
+		l.config.IndexedTaints,
+		l.config.IndexedNodeLabels,
+	)
+	if err != nil {
+		return nil, err
+	}
+	err = nodeDb.Upsert(nodes)
+	if err != nil {
+		return nil, err
+	}
+	return nodeDb, nil
+}
+
+// filterEmptyQueues returns only the queues which have leased jobs in the jobs db
+func (l *LegacySchedulingAlgo) filterEmptyQueues(allQueues []*database.Queue, txn *memdb.Txn, jobDb *JobDb) ([]*database.Queue, error) {
+	activeQueues := make([]*database.Queue, 0, len(allQueues))
+	for _, queue := range allQueues {
+		keep, err := jobDb.HasQueuedJobs(txn, queue.Name)
+		if err != nil {
+			return nil, err
+		}
+		if keep {
+			activeQueues = append(activeQueues, queue)
+		}
+	}
+	return activeQueues, nil
+}
+
+func (l *LegacySchedulingAlgo) filterStaleExecutors(allExecutors []*database.Executor) []*database.Executor {
+	activeExecutors := make([]*database.Executor, 0, len(allExecutors))
+	cutoff := l.clock.Now().Add(-l.config.ExecutorTimeout)
+	for _, executor := range allExecutors {
+		if executor.LastUpdateTime.After(cutoff) {
+			activeExecutors = append(activeExecutors, executor)
+		} else {
+			log.Debugf("Ignoring executor %s because it hasn't heartbeated since %s", executor.Name, executor.LastUpdateTime)
+		}
+	}
+	return activeExecutors
+}
+
+// aggregateUsage Creates a map of usage by pool
+func aggregateUsage(executors []*database.Executor) map[string]map[string]schedulerobjects.QuantityByPriorityAndResourceType {
+	usageByPool := make(map[string]map[string]schedulerobjects.QuantityByPriorityAndResourceType, 0)
+
+	for _, executor := range executors {
+		poolUsage, ok := usageByPool[executor.Pool]
+		if !ok {
+			poolUsage = make(map[string]schedulerobjects.QuantityByPriorityAndResourceType)
+			usageByPool[executor.Pool] = poolUsage
+		}
+		for queue, report := range executor.Usage.ResourcesByQueue {
+			quantityByPriorityAndResourceType, ok := poolUsage[queue]
+			if !ok {
+				quantityByPriorityAndResourceType = make(schedulerobjects.QuantityByPriorityAndResourceType)
+				poolUsage[queue] = quantityByPriorityAndResourceType
+			}
+			quantityByPriorityAndResourceType.Add(report.ResourcesByPriority)
+		}
+	}
+	return usageByPool
+}


Test patch:
diff --git a/internal/scheduler/database/job_repository_test.go b/internal/scheduler/database/job_repository_test.go
new file mode 100644
index 00000000..b21b9ee0
--- /dev/null
+++ b/internal/scheduler/database/job_repository_test.go
@@ -0,0 +1,243 @@
+package database
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/armadaproject/armada/internal/common/database"
+	"github.com/armadaproject/armada/internal/common/util"
+)
+
+const defaultBatchSize = 1
+
+func TestFetchJobUpdates(t *testing.T) {
+	dbJobs, expectedJobs := createTestJobs(10)
+	dbRuns, expectedRuns := createTestRuns(10)
+
+	tests := map[string]struct {
+		dbJobs       []Job
+		dbRuns       []Run
+		jobsSerial   int64
+		runsSerial   int64
+		expectedJobs []Job
+		expectedRuns []Run
+	}{
+		"all jobs": {
+			dbJobs:       dbJobs,
+			expectedJobs: expectedJobs,
+			expectedRuns: []Run{},
+			jobsSerial:   0,
+		},
+		"some jobs": {
+			dbJobs:       dbJobs,
+			expectedJobs: expectedJobs[5:],
+			expectedRuns: []Run{},
+			jobsSerial:   5,
+		},
+		"no jobs": {
+			dbJobs:       dbJobs,
+			expectedJobs: []Job{},
+			expectedRuns: []Run{},
+			jobsSerial:   10,
+		},
+		"all runs": {
+			dbRuns:       dbRuns,
+			expectedRuns: expectedRuns,
+			expectedJobs: []Job{},
+			jobsSerial:   0,
+		},
+		"some runs": {
+			dbRuns:       dbRuns,
+			expectedRuns: expectedRuns[5:],
+			expectedJobs: []Job{},
+			runsSerial:   5,
+		},
+		"no runs": {
+			dbRuns:       dbRuns,
+			expectedJobs: []Job{},
+			expectedRuns: []Run{},
+			runsSerial:   10,
+		},
+		"both jobs and runs": {
+			dbJobs:       dbJobs,
+			dbRuns:       dbRuns,
+			expectedJobs: expectedJobs,
+			expectedRuns: expectedRuns,
+		},
+		"empty db": {
+			expectedJobs: []Job{},
+			expectedRuns: []Run{},
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			err := withJobRepository(func(repo *PostgresJobRepository) error {
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+
+				// Set up db
+				err := database.Upsert(ctx, repo.db, "jobs", tc.dbJobs)
+				require.NoError(t, err)
+				err = database.Upsert(ctx, repo.db, "runs", tc.dbRuns)
+				require.NoError(t, err)
+
+				// Fetch updates
+				jobs, runs, err := repo.FetchJobUpdates(ctx, tc.jobsSerial, tc.runsSerial)
+				require.NoError(t, err)
+
+				// Runs will have LastModified filled in- we don't want to compare this
+				for i := range runs {
+					runs[i].LastModified = time.Time{}
+				}
+
+				// Assert results
+				assert.Equal(t, tc.expectedJobs, jobs)
+				assert.Equal(t, tc.expectedRuns, runs)
+				cancel()
+				return nil
+			})
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestCountReceivedPartitions(t *testing.T) {
+	tests := map[string]struct {
+		numPartitions int
+	}{
+		"100 partitions": {
+			numPartitions: 100,
+		},
+		"0 partitions": {
+			numPartitions: 0,
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			err := withJobRepository(func(repo *PostgresJobRepository) error {
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+
+				markers := make([]Marker, tc.numPartitions)
+				groupId := uuid.New()
+				for i := 0; i < tc.numPartitions; i++ {
+					markers[i] = Marker{
+						GroupID:     groupId,
+						PartitionID: int32(i),
+					}
+				}
+
+				// Set up db
+				err := insertMarkers(ctx, markers, repo.db)
+				require.NoError(t, err)
+
+				// Fetch updates
+				received, err := repo.CountReceivedPartitions(ctx, groupId)
+				require.NoError(t, err)
+
+				assert.Equal(t, uint32(tc.numPartitions), received)
+
+				cancel()
+				return nil
+			})
+			require.NoError(t, err)
+		})
+	}
+}
+
+func createTestJobs(numJobs int) ([]Job, []Job) {
+	dbJobs := make([]Job, numJobs)
+	expectedJobs := make([]Job, numJobs)
+
+	for i := 0; i < numJobs; i++ {
+		dbJobs[i] = Job{
+			JobID:           util.NewULID(),
+			JobSet:          "test-jobset",
+			Queue:           "test-queue",
+			Submitted:       int64(i),
+			Priority:        int64(i),
+			CancelRequested: true,
+			Cancelled:       true,
+			Succeeded:       true,
+			Failed:          true,
+			SchedulingInfo:  []byte{byte(i)},
+			SubmitMessage:   []byte{},
+		}
+	}
+
+	for i, job := range dbJobs {
+		expectedJobs[i] = Job{
+			JobID:           job.JobID,
+			JobSet:          job.JobSet,
+			Queue:           job.Queue,
+			Submitted:       job.Submitted,
+			Priority:        job.Priority,
+			CancelRequested: job.CancelRequested,
+			Cancelled:       job.Cancelled,
+			Succeeded:       job.Succeeded,
+			Failed:          job.Failed,
+			SchedulingInfo:  job.SchedulingInfo,
+			Serial:          int64(i + 1),
+		}
+	}
+	return dbJobs, expectedJobs
+}
+
+func createTestRuns(numRuns int) ([]Run, []Run) {
+	dbRuns := make([]Run, numRuns)
+	expectedRuns := make([]Run, numRuns)
+
+	for i := 0; i < numRuns; i++ {
+		dbRuns[i] = Run{
+			RunID:          uuid.New(),
+			JobID:          util.NewULID(),
+			JobSet:         "test-jobset",
+			Executor:       "test-executor",
+			SentToExecutor: true,
+			Cancelled:      true,
+			Running:        true,
+			Succeeded:      true,
+			Failed:         true,
+			Returned:       true,
+		}
+	}
+
+	for i, run := range dbRuns {
+		expectedRuns[i] = Run{
+			RunID:          run.RunID,
+			JobID:          run.JobID,
+			JobSet:         run.JobSet,
+			Executor:       run.Executor,
+			SentToExecutor: run.SentToExecutor,
+			Cancelled:      run.Cancelled,
+			Running:        run.Running,
+			Succeeded:      run.Succeeded,
+			Failed:         run.Failed,
+			Returned:       run.Returned,
+			Serial:         int64(i + 1),
+		}
+	}
+	return dbRuns, expectedRuns
+}
+
+func withJobRepository(action func(repository *PostgresJobRepository) error) error {
+	return WithTestDb(func(_ *Queries, db *pgxpool.Pool) error {
+		repo := NewPostgresJobRepository(db, defaultBatchSize)
+		return action(repo)
+	})
+}
+
+func insertMarkers(ctx context.Context, markers []Marker, db *pgxpool.Pool) error {
+	for _, marker := range markers {
+		_, err := db.Exec(ctx, "INSERT INTO markers VALUES ($1, $2)", marker.GroupID, marker.PartitionID)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+	}
+	return nil
+}
diff --git a/internal/scheduler/jobdb_test.go b/internal/scheduler/jobdb_test.go
index 86b1b014..6d0a9f24 100644
--- a/internal/scheduler/jobdb_test.go
+++ b/internal/scheduler/jobdb_test.go
@@ -71,7 +71,7 @@ func TestBatchDelete(t *testing.T) {
 			// Set up JobDb
 			jobDb := createPopulatedJobDb(t, tc.initialJobs)
 
-			// do the delete
+			// delete
 			txn := jobDb.WriteTxn()
 			err := jobDb.BatchDelete(txn, tc.idsToDelete)
 			require.NoError(t, err)
diff --git a/internal/scheduler/legacyscheduler_test.go b/internal/scheduler/legacyscheduler_test.go
index b0007c8d..67e10f0a 100644
--- a/internal/scheduler/legacyscheduler_test.go
+++ b/internal/scheduler/legacyscheduler_test.go
@@ -547,6 +547,7 @@ func testSchedulingConfig() configuration.SchedulingConfig {
 		IndexedResources:          []string{"cpu", "memory"},
 		GangIdAnnotation:          testGangIdAnnotation,
 		GangCardinalityAnnotation: testGangCardinalityAnnotation,
+		ExecutorTimeout:           15 * time.Minute,
 	}
 }
 
diff --git a/internal/scheduler/scheduling_algo_test.go b/internal/scheduler/scheduling_algo_test.go
new file mode 100644
index 00000000..95d7a158
--- /dev/null
+++ b/internal/scheduler/scheduling_algo_test.go
@@ -0,0 +1,245 @@
+package scheduler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/armadaproject/armada/internal/common/util"
+	"k8s.io/apimachinery/pkg/util/clock"
+
+	"github.com/golang/mock/gomock"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
+	"github.com/armadaproject/armada/internal/scheduler/database"
+	schedulermocks "github.com/armadaproject/armada/internal/scheduler/mocks"
+	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
+)
+
+const (
+	queueName = "queue1"
+	poolName  = "pool1"
+)
+
+var (
+	queue = database.Queue{
+		Name:   queueName,
+		Weight: 100,
+	}
+	baseTime time.Time
+)
+
+func TestLegacySchedulingAlgo_TestSchedule(t *testing.T) {
+	jobs := make([]*SchedulerJob, 10)
+	for i := 0; i < 10; i++ {
+		jobs[i] = OneCpuJob()
+		jobs[i].Timestamp = int64(i) // ensure the jobs are in the order we expect
+	}
+	tests := map[string]struct {
+		executors     []*database.Executor
+		queues        []*database.Queue
+		jobs          []*SchedulerJob
+		perQueueLimit map[string]float64
+		expectedJobs  map[string]string // map of jobId to name of executor on which it should be scheduled
+	}{
+		"fill up both clusters": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("0"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("0"), baseTime),
+			},
+			queues: []*database.Queue{&queue},
+			jobs:   jobs,
+			expectedJobs: map[string]string{
+				jobs[0].JobId: "executor1",
+				jobs[1].JobId: "executor1",
+				jobs[2].JobId: "executor2",
+				jobs[3].JobId: "executor2",
+			},
+		},
+		"one executor stale": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("0"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("0"), baseTime.Add(-1*time.Hour)),
+			},
+			queues: []*database.Queue{&queue},
+			jobs:   jobs,
+			expectedJobs: map[string]string{
+				jobs[0].JobId: "executor1",
+				jobs[1].JobId: "executor1",
+			},
+		},
+		"one executor full": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("2"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("0"), baseTime),
+			},
+			queues: []*database.Queue{&queue},
+			jobs:   jobs,
+			expectedJobs: map[string]string{
+				jobs[0].JobId: "executor2",
+				jobs[1].JobId: "executor2",
+			},
+		},
+		"user is at usage cap before scheduling": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("2"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("0"), baseTime),
+			},
+			queues:        []*database.Queue{&queue},
+			jobs:          jobs,
+			perQueueLimit: map[string]float64{"cpu": 0.5},
+			expectedJobs:  map[string]string{},
+		},
+		"user hits usage cap during scheduling": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("1"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("0"), baseTime),
+			},
+			queues:        []*database.Queue{&queue},
+			jobs:          jobs,
+			perQueueLimit: map[string]float64{"cpu": 0.5},
+			expectedJobs: map[string]string{
+				jobs[0].JobId: "executor1",
+			},
+		},
+		"no jobs to schedule": {
+			executors: []*database.Executor{
+				TwoCoreExecutor("executor1", resource.MustParse("2"), baseTime),
+				TwoCoreExecutor("executor2", resource.MustParse("2"), baseTime),
+			},
+			queues:       []*database.Queue{&queue},
+			jobs:         nil,
+			expectedJobs: map[string]string{},
+		},
+		"no executor available": {
+			executors:    []*database.Executor{},
+			queues:       []*database.Queue{&queue},
+			jobs:         jobs,
+			expectedJobs: map[string]string{},
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			config := testSchedulingConfig()
+			if tc.perQueueLimit != nil {
+				config = withPerQueueLimits(tc.perQueueLimit, config)
+			}
+			ctrl := gomock.NewController(t)
+			mockExecutorRepo := schedulermocks.NewMockExecutorRepository(ctrl)
+			mockQueueRepo := schedulermocks.NewMockQueueRepository(ctrl)
+
+			mockExecutorRepo.EXPECT().GetExecutors().Return(tc.executors, nil).AnyTimes()
+			mockQueueRepo.EXPECT().GetAllQueues().Return(tc.queues, nil).AnyTimes()
+
+			algo := NewLegacySchedulingAlgo(config,
+				mockExecutorRepo,
+				mockQueueRepo)
+
+			// Use a test clock so we can control time
+			algo.clock = clock.NewFakeClock(baseTime)
+
+			// Set up JobDb
+			jobDb, err := NewJobDb()
+			require.NoError(t, err)
+			txn := jobDb.WriteTxn()
+			err = jobDb.Upsert(txn, tc.jobs)
+			require.NoError(t, err)
+
+			scheduledJobs, err := algo.Schedule(txn, jobDb)
+			require.NoError(t, err)
+
+			// check that we have scheduled the jobs we expect
+			assert.Equal(t, len(tc.expectedJobs), len(scheduledJobs))
+
+			for _, job := range scheduledJobs {
+				expectedExecutor, ok := tc.expectedJobs[job.JobId]
+				assert.True(t, ok)
+				assert.Equal(t, expectedExecutor, job.Executor)
+				assert.Equal(t, false, job.Queued)
+			}
+
+			// check all scheduled jobs are up-to-date in db
+			for _, job := range scheduledJobs {
+				dbJob, err := jobDb.GetById(txn, job.JobId)
+				require.NoError(t, err)
+				assert.Equal(t, job, dbJob)
+			}
+		})
+	}
+}
+
+func twoCoreNode(usedCpu resource.Quantity) *schedulerobjects.Node {
+	allocatableCpu := resource.MustParse("2")
+	(&allocatableCpu).Sub(usedCpu)
+	return &schedulerobjects.Node{
+		Id: uuid.NewString(),
+		TotalResources: schedulerobjects.ResourceList{
+			Resources: map[string]resource.Quantity{
+				"cpu":    resource.MustParse("2"),
+				"memory": resource.MustParse("256Gi"),
+			},
+		},
+		AllocatableByPriorityAndResource: schedulerobjects.NewAllocatableByPriorityAndResourceType(
+			[]int32{0},
+			map[string]resource.Quantity{
+				"cpu":    allocatableCpu,
+				"memory": resource.MustParse("256Gi"),
+			},
+		),
+	}
+}
+
+func TwoCoreExecutor(name string, usedCPU resource.Quantity, updateTime time.Time) *database.Executor {
+	return &database.Executor{
+		Name: name,
+		Pool: poolName,
+		Usage: &schedulerobjects.ClusterResourceUsageReport{
+			Pool: poolName,
+			ResourcesByQueue: map[string]*schedulerobjects.QueueClusterResourceUsage{queueName: {
+				Created:    updateTime,
+				Queue:      queueName,
+				ExecutorId: name,
+				ResourcesByPriority: map[int32]schedulerobjects.ResourceList{
+					0: {Resources: map[string]resource.Quantity{
+						"cpu": usedCPU,
+					}},
+				},
+			}},
+		},
+		Nodes:          []*schedulerobjects.Node{twoCoreNode(usedCPU)},
+		LastUpdateTime: updateTime,
+	}
+}
+
+func OneCpuJob() *SchedulerJob {
+	return &SchedulerJob{
+		JobId:    util.NewULID(),
+		Queue:    queueName,
+		Jobset:   "test-jobset",
+		Queued:   true,
+		Priority: 0,
+		jobSchedulingInfo: &schedulerobjects.JobSchedulingInfo{
+			ObjectRequirements: []*schedulerobjects.ObjectRequirements{
+				{
+					Requirements: &schedulerobjects.ObjectRequirements_PodRequirements{
+						PodRequirements: &schedulerobjects.PodRequirements{
+							ResourceRequirements: v1.ResourceRequirements{
+								Limits: map[v1.ResourceName]resource.Quantity{
+									"memory": resource.MustParse("1Mi"),
+									"cpu":    resource.MustParse("1"),
+								},
+								Requests: map[v1.ResourceName]resource.Quantity{
+									"memory": resource.MustParse("1"),
+									"cpu":    resource.MustParse("1"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go
index b071b588..7606700a 100644
--- a/internal/scheduleringester/schedulerdb_test.go
+++ b/internal/scheduleringester/schedulerdb_test.go
@@ -250,9 +250,12 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 
 	// Read back the state from the db to compare.
 	queries := schedulerdb.New(schedulerDb.db)
+	selectNewJobs := func(ctx context.Context, serial int64) ([]schedulerdb.Job, error) {
+		return queries.SelectNewJobs(ctx, schedulerdb.SelectNewJobsParams{Serial: serial, Limit: 1000})
+	}
 	switch expected := op.(type) {
 	case InsertJobs:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -271,7 +274,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			assert.Equal(t, v, actual[k])
 		}
 	case InsertRuns:
-		jobs, err := queries.SelectNewJobs(ctx, 0)
+		jobs, err := selectNewJobs(ctx, 0)
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -315,7 +318,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, expected, actual)
 	case UpdateJobSetPriorities:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -328,7 +331,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Greater(t, numChanged, 0)
 	case MarkJobSetsCancelled:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -355,7 +358,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			assert.True(t, run.Cancelled)
 		}
 	case MarkJobsCancelled:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -382,7 +385,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			assert.True(t, run.Cancelled)
 		}
 	case MarkJobsSucceeded:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -395,7 +398,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, len(expected), numChanged)
 	case MarkJobsFailed:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -408,7 +411,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, len(expected), numChanged)
 	case UpdateJobPriorities:
-		jobs, err := queries.SelectNewJobs(ctx, serials["jobs"])
+		jobs, err := selectNewJobs(ctx, serials["jobs"])
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -421,7 +424,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, len(expected), numChanged)
 	case MarkRunsSucceeded:
-		jobs, err := queries.SelectNewJobs(ctx, 0)
+		jobs, err := selectNewJobs(ctx, 0)
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -446,7 +449,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, len(expected), len(runs))
 	case MarkRunsFailed:
-		jobs, err := queries.SelectNewJobs(ctx, 0)
+		jobs, err := selectNewJobs(ctx, 0)
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -471,7 +474,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 		}
 		assert.Equal(t, len(expected), len(runs))
 	case MarkRunsRunning:
-		jobs, err := queries.SelectNewJobs(ctx, 0)
+		jobs, err := selectNewJobs(ctx, 0)
 		if err != nil {
 			return errors.WithStack(err)
 		}