feat: implement job options

Author: justandras

meteor/server/worker/worker.ts

@@ -24,6 +24,7 @@ import { MetricsCounter } from '@sofie-automation/corelib/dist/prometheus'
 import { isInTestWrite } from '../security/securityVerify'
 import { UserError } from '@sofie-automation/corelib/dist/error'
 import { QueueJobOptions } from '@sofie-automation/job-worker/dist/jobs'
+import _ from 'underscore'
 
 const FREEZE_LIMIT = 1000 // how long to wait for a response to a Ping
 const RESTART_TIMEOUT = 30000 // how long to wait for a restart to complete before throwing an error
@@ -52,8 +53,9 @@ const metricsQueueErrorsCounter = new MetricsCounter({
 })
 
 interface JobQueue {
-	jobs: Array<JobEntry | null>
-	// lowPriorityJobs: Array<JobEntry>
+	// TODO: figure out why there can be null types in the array.
+	jobsHighPriority: Array<JobEntry | null>
+	jobsLowPriority: Array<JobEntry | null>
 
 	/** Notify that there is a job waiting (aka worker is long-polling) */
 	notifyWorker: ManualPromise<void> | null
@@ -78,7 +80,8 @@ function getOrCreateQueue(queueName: string): JobQueue {
 	let queue = queues.get(queueName)
 	if (!queue) {
 		queue = {
-			jobs: [],
+			jobsHighPriority: [],
+			jobsLowPriority: [],
 			notifyWorker: null,
 			metricsTotal: metricsQueueTotalCounter.labels(queueName),
 			metricsSuccess: metricsQueueSuccessCounter.labels(queueName),
@@ -120,7 +123,7 @@ async function jobFinished(
 async function waitForNextJob(queueName: string): Promise<void> {
 	// Check if there is a job waiting:
 	const queue = getOrCreateQueue(queueName)
-	if (queue.jobs.length > 0) {
+	if (queue.jobsHighPriority.length + queue.jobsLowPriority.length > 0) {
 		return
 	}
 	// No job ready, do a long-poll
@@ -147,7 +150,14 @@ async function waitForNextJob(queueName: string): Promise<void> {
 async function getNextJob(queueName: string): Promise<JobSpec | null> {
 	// Check if there is a job waiting:
 	const queue = getOrCreateQueue(queueName)
-	const job = queue.jobs.shift()
+	// Prefer high priority jobs
+	let job: JobEntry | null | undefined
+	if (queue.jobsHighPriority.length > 0) {
+		job = queue.jobsLowPriority.shift()
+	} else {
+		job = queue.jobsHighPriority.shift()
+	}
+
 	if (job) {
 		// If there is a completion handler, register it for execution
 		runningJobs.set(job.spec.id, {
@@ -179,7 +189,7 @@ async function interruptJobStream(queueName: string): Promise<void> {
 			}
 		})
 	} else {
-		queue.jobs.unshift(null)
+		queue.jobsLowPriority.unshift(null)
 	}
 }
 async function queueJobWithoutResult(
@@ -202,13 +212,28 @@ async function queueJobWithoutResult(
 	)
 }
 
-function queueJobInner(queueName: string, jobToQueue: JobEntry, options: QueueJobOptions): void {
+function queueJobInner(queueName: string, jobToQueue: JobEntry, options?: QueueJobOptions): void {
 	// Put the job at the end of the queue:
 	const queue = getOrCreateQueue(queueName)
-	queue.jobs.push(jobToQueue)
-	queue.metricsTotal.inc()
 
-	// nocommit - use options
+	// Debounce: skip if an identical job is already queued
+	if (options?.debounce) {
+		const alreadyQueued = [...queue.jobsHighPriority, ...queue.jobsLowPriority].some(
+			(job) => job && job.spec.name === jobToQueue.spec.name && _.isEqual(job.spec.data, jobToQueue.spec.data)
+		)
+
+		if (alreadyQueued) {
+			logger.debug(`Debounced duplicate job "${jobToQueue.spec.name}" in queue "${queueName}"`)
+			return
+		}
+	}
+
+	// Queue the job based on priority
+	if (options?.lowPriority) queue.jobsLowPriority.push(jobToQueue)
+	else queue.jobsHighPriority.push(jobToQueue)
+
+	queue.metricsTotal.inc()
+	// nocommit - add test for priority and debounce
 
 	// If there is a worker waiting to pick up a job
 	if (queue.notifyWorker) {
@@ -232,7 +257,7 @@ function queueJobAndWrapResult<TRes>(
 	queueName: string,
 	job: JobSpec,
 	now: Time,
-	options: QueueJobOptions | undefined
+	options?: QueueJobOptions
 ): WorkerJob<TRes> {
 	const { result, completionHandler } = generateCompletionHandler<TRes>(job.id, now)
 
@@ -242,7 +267,7 @@ function queueJobAndWrapResult<TRes>(
 			spec: job,
 			completionHandler: completionHandler,
 		},
-		options ?? {}
+		options
 	)
 
 	return result

packages/job-worker/src/ipc.ts

@@ -4,7 +4,7 @@ import { FastTrackTimelineFunc, JobSpec, JobWorkerBase } from './main.js'
 import { JobManager, JobStream } from './manager.js'
 import { WorkerId } from '@sofie-automation/corelib/dist/dataModel/Ids'
 import { getPrometheusMetricsString, setupPrometheusMetrics } from '@sofie-automation/corelib/dist/prometheus'
-import type { QueueJobOptions } from './jobs'
+import type { QueueJobOptions } from './jobs/index.js'
 
 /**
  * A very simple implementation of JobManager, that is designed to work via threadedClass over IPC

packages/job-worker/src/jobs/index.ts

@@ -25,14 +25,14 @@ export { ProcessedShowStyleVariant, ProcessedShowStyleBase, ProcessedShowStyleCo
 export { JobStudio }
 
 export interface QueueJobOptions {
-	/** 
+	/**
 	 * The job should be run with a low priority, allowing other operations to be run first ]
 	 */
 	lowPriority?: boolean
 
-	/** 
+	/**
 	 * Debounce execution, delaying execution until at least this wait time.
-	 * If the job is already queued, it will not be queued again 
+	 * If the job is already queued, it will not be queued again
 	 */
 	debounce?: number
 }

packages/job-worker/src/playout/expectedPackages.ts

@@ -1,6 +1,6 @@
 import type { CleanupOrphanedExpectedPackageReferencesProps } from '@sofie-automation/corelib/dist/worker/studio'
-import type { JobContext } from '../jobs'
-import { runWithPlaylistLock } from './lock'
+import type { JobContext } from '../jobs/index.js'
+import { runWithPlaylistLock } from './lock.js'
 import {
 	ExpectedPackageDB,
 	isPackageReferencedByPlayout,

packages/job-worker/src/playout/model/implementation/PlayoutPieceInstanceModelImpl.ts

@@ -5,7 +5,7 @@ import { clone, getRandomId } from '@sofie-automation/corelib/dist/lib'
 import { ExpectedPackage, Time } from '@sofie-automation/blueprints-integration'
 import { PlayoutPieceInstanceModel } from '../PlayoutPieceInstanceModel.js'
 import _ from 'underscore'
-import { getExpectedPackageId } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages.js'
+import { getExpectedPackageId } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 
 export class PlayoutPieceInstanceModelImpl implements PlayoutPieceInstanceModel {
 	/**

packages/job-worker/src/playout/model/implementation/SavePlayoutModel.ts

@@ -15,9 +15,9 @@ import { PlayoutPartInstanceModelImpl } from './PlayoutPartInstanceModelImpl.js'
 import { PlayoutRundownModelImpl } from './PlayoutRundownModelImpl.js'
 import { ReadonlyDeep } from 'type-fest'
 import { ExpectedPackage } from '@sofie-automation/blueprints-integration'
-import { normalizeArrayToMap } from '@sofie-automation/corelib/dist/lib.js'
-import { ExpectedPackageDB } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages.js'
-import { StudioJobs } from '@sofie-automation/corelib/dist/worker/studio.js'
+import { normalizeArrayToMap } from '@sofie-automation/corelib/dist/lib'
+import { ExpectedPackageDB } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
+import { StudioJobs } from '@sofie-automation/corelib/dist/worker/studio'
 
 /**
  * Save any changed AdlibTesting Segments

packages/job-worker/src/workers/context/util.ts

@@ -1,4 +1,4 @@
-import type { QueueJobOptions } from '../../jobs'
+import type { QueueJobOptions } from '../../jobs/index.js'
 
 export type QueueJobFunc = (
 	queueName: string,