tc-mccarthy · tc-mccarthy · Jun 27, 2025 · Jun 27, 2025 · Jun 27, 2025 · Jun 27, 2025
diff --git a/transcode/eslint.config.mjs b/transcode/eslint.config.mjs
@@ -159,6 +159,7 @@ export default [{
         "new-parens": "error",
         "no-array-constructor": "error",
         "no-async-promise-executor": 0,
+        "no-await-in-loop": 0,
         "no-caller": "error",
         "no-case-declarations": "error",
         "no-class-assign": "error",

diff --git a/transcode/index.js b/transcode/index.js
@@ -3,7 +3,6 @@ import dayjs from 'dayjs';
 import mongo_connect from './lib/mongo_connection';
 import update_active from './lib/update_active';
 import update_queue from './lib/update_queue';
-import transcode_loop from './lib/transcode_loop';
 import fs_monitor from './lib/fs_monitor';
 import redisClient from './lib/redis';
 import logger from './lib/logger';
@@ -12,7 +11,8 @@ import pre_sanitize from './lib/pre_sanitize';
 import { create_scratch_disks } from './lib/fs';
 import config from './lib/config';
 import generate_filelist from './lib/generate_filelist';
-import integrity_loop from './lib/integrity_check_loop';
+import IntegrityQueue from './lib/integrityQueue';
+import TranscodeQueue from './lib/transcodeQueue';
 
 const {
   concurrent_transcodes,
@@ -55,38 +55,31 @@ async function run () {
     // update the transcode queue
     update_queue();
 
-    // start the transcode loops
-    logger.info(`Starting ${concurrent_transcodes} transcode loops...`);
+    const transcodeQueue = new TranscodeQueue({ maxScore: concurrent_transcodes, pollDelay: 10000 });
+    transcodeQueue.start();
 
-    Array.from({ length: concurrent_transcodes }).forEach((val, idx) => {
-      transcode_loop(idx);
-    });
+    const integrityQueue = new IntegrityQueue({ maxScore: concurrent_integrity_checks });

     const currentHourLocalTime = dayjs().tz(process.env.TZ).hour();
     logger.info(
       `Current local time is ${currentHourLocalTime}`
     );
     if (currentHourLocalTime >= 0 && currentHourLocalTime < 9) {
-      logger.info(
-        'Starting integrity check loop immediately since it is before 9 AM'
-      );
-      integrity_loop();
+      integrityQueue.start();
     }
 
-    // start the integrity check loops every day at midnight
+    // start the integrity check queue every day at midnight
     cron.schedule('0 0 * * *', () => {
-      logger.info(
-        `Starting ${concurrent_integrity_checks} integrity check loops...`
-      );
-      Array.from({ length: concurrent_integrity_checks }).forEach(
-        (val, idx) => {
-          integrity_loop(idx);
-        }
-      );
+      integrityQueue.start();
+    });
+
+    // pause the integrity check queue every day at 9am
+    cron.schedule('0 9 * * *', () => {
+      integrityQueue.stop();
     });
 
     // generate the filelist every 10 minutes
-    cron.schedule('*/5 * * * *', async () => {
+    cron.schedule('*/2 * * * *', async () => {
       generate_filelist({ limit: 1000, writeToFile: true });
     });
 

diff --git a/transcode/lib/ffprobe.js b/transcode/lib/ffprobe.js
@@ -44,11 +44,21 @@ export default async function ffprobe_func (file) {
       video.aspect = aspect_round(video.width / video.height);
     }
 
+    // mark the video as unsupported if it has a dv_profile value and that value is less than 8
+    if (video.side_data_list?.dv_profile && video.side_data_list?.dv_profile < 8) {
+      throw new Error(`Video not supported: Dolby Vision profile version is less than 8`);
+    }
+
     return data;
   } catch (e) {
     if (/command\s+failed/gi.test(e.message)) {
       trash(file);
     }
+
+    if (/video\s+not\s+supported/gi.test(e.message)) {
+      trash(file);
+    }
+
     logger.error('FFPROBE FAILED', e);
     return false;
   }

diff --git a/transcode/lib/generate_filelist.js b/transcode/lib/generate_filelist.js
@@ -35,7 +35,8 @@ export default async function generate_filelist ({ limit = 1, writeToFile = fals
           codec: `${
           f.probe.streams.find((v) => v.codec_type === 'video')?.codec_name
         }/${f.probe.streams.find((v) => v.codec_type === 'audio')?.codec_name}`,
-          encode_version: f.encode_version
+          encode_version: f.encode_version,
+          computeScore: f.computeScore
         }))
       )
     );

diff --git a/transcode/lib/integrityCheck.js b/transcode/lib/integrityCheck.js
@@ -35,19 +35,11 @@ export default function integrityCheck (file) {
   return new Promise(async (resolve, reject) => {
     try {
       // mongo record of the video
-      logger.info(file, { label: 'INTEGRITY CHECKING FILE' });
+      logger.debug(file, { label: 'INTEGRITY CHECKING FILE' });
 
       const video_record = file;
       file = file.path;
 
-      // if the file is locked, short circuit
-      if (await video_record.hasLock()) {
-        logger.info(
-          `File is locked. Skipping integrity check: ${file} - ${video_record._id}`
-        );
-        return resolve();
-      }
-
       await video_record.setLock('integrity');
 
       const exists = fs.existsSync(file);

diff --git a/transcode/lib/integrityQueue.js b/transcode/lib/integrityQueue.js
@@ -0,0 +1,95 @@
+import { setTimeout as delay } from 'timers/promises';
+import integrityCheck from './integrityCheck';
+import logger from './logger';
+import generate_integrity_filelist from './generate_integrity_filelist';
+
+export default class IntegrityQueue {
+  constructor ({ maxScore = 4, pollDelay = 2000 }) {
+    // start the integrity check loops
+    logger.info(`Initiating integrity check queue for a max compute of ${maxScore}...`);
+    this.maxScore = maxScore; // Max compute units allowed simultaneously
+    this.pollDelay = pollDelay; // Delay between scheduling attempts (ms)
+    this.runningJobs = []; // In-memory list of currently active jobs
+    this._isRunning = false; // Flag for controlling the loop
+  }
+
+  // Starts the recursive scheduling loop
+  async start () {
+    if (this._isRunning) return;
+    this._isRunning = true;
+    logger.info('Integrity check queue started.');
+    await this.loop();
+  }
+
+  // Stops the queue
+  stop () {
+    this._isRunning = false;
+    console.log('Integrity queue stopped.');
+  }
+
+  // Returns total compute in use
+  getUsedCompute () {
+    return this.runningJobs.reduce((sum, job) => sum + job.computeScore, 0);
+  }
+
+  // Returns available compute capacity
+  getAvailableCompute () {
+    return this.maxScore - this.getUsedCompute();
+  }
+
+  // Main loop: tries to schedule jobs and waits before the next run
+  async loop () {
+    while (this._isRunning) {
+      await this.scheduleJobs();
+      await delay(this.pollDelay); // Wait before checking again
+    }
+  }
+
+  // Attempts to find and run a job that fits within available compute
+  async scheduleJobs () {
+    const availableCompute = this.getAvailableCompute();
+    logger.info(`Available integrity check compute: ${availableCompute}.`);
+
+    if (availableCompute <= 0) return;
+
+    logger.info('Checking for new jobs to run...');
+    const jobs = await generate_integrity_filelist({ limit: 50 });
+
+    // Are there any jobs being blocked due to lack of compute?
+    const blockedHighPriorityJob = jobs.find((job) => {
+      const alreadyRunning = this.runningJobs.some((j) => j._id.toString() === job._id.toString());
+      return !alreadyRunning && job.computeScore > availableCompute;
+    });
+
+    // Now let's find the next job that will fit within available compute
+    const nextJob = jobs.find((job) => {
+      const alreadyRunning = this.runningJobs.some((j) => j._id.toString() === job._id.toString()); // skip any already running jobs
+      if (alreadyRunning || job.computeScore > availableCompute) return false; // discount any jobs that are already running or exceed available compute
+
+      // If a higher-priority job is blocked, don't schedule lower-priority jobs
+      if (blockedHighPriorityJob && job.sortFields.priority > blockedHighPriorityJob.sortFields.priority) return false; // if a higher-priority job is blocked, don't schedule lower-priority jobs, let the queue open up to process the higher-priority job
+
+      // If we reach here, the job is eligible to run
+      return true;
+    });
+
+    if (nextJob) {
+      this.runJob(nextJob);
+    }
+  }
+
+  // Handles job execution and cleanup
+      async runJob (job) {
+    try {
+      this.runningJobs.push(job);
+      await integrityCheck(job); // Await external ffmpeg logic
+    } catch (err) {
+      console.error(`Integrity check failed for ${job.inputPath}: ${err.message}`);
+    } finally {
+      // Always clean up the memory queue
+      this.runningJobs = this.runningJobs.filter(
+        (j) => j._id.toString() !== job._id.toString()
+      );
+    }
+  }
+}
diff --git a/transcode/lib/integrity_check_loop.js b/transcode/lib/integrity_check_loop.js
diff --git a/transcode/lib/round-to-nearest-quarter.js b/transcode/lib/round-to-nearest-quarter.js
@@ -0,0 +1,3 @@
+export default function roundToNearestQuarter (number) {
+  return Math.round(number * 4) / 4;
+}
diff --git a/transcode/lib/transcode.js b/transcode/lib/transcode.js
@@ -24,14 +24,6 @@ export default function transcode (file) {
         throw new Error(`Video record not found for file: ${file}`);
       }
 
-      // if the file is locked, short circuit
-      if (await video_record.hasLock('transcode')) {
-        logger.info(
-          `File is locked. Skipping transcode: ${file} - ${video_record._id}`
-        );
-        return resolve({ locked: true });
-      }
-
       const { profiles } = config;
       const exists = fs.existsSync(file);
 
@@ -169,7 +161,7 @@ export default function transcode (file) {
         ffprobe_data.format.size <= 350000
       ) {
         logger.debug(
-          'Video stream codec is HEVC and size is less than 1GB. Not transcoding'
+          'Video stream codec is h264 and size is less than 350mb. Not transcoding'
         );
         transcode_video = false;
       }
@@ -202,6 +194,14 @@ export default function transcode (file) {
         await integrityCheck(video_record);
       }
 
+      if (!transcode_video) {
+        video_record.computeScore = 0.2; // set the compute score to 0.2 because we're not transcoding
+      }
+
+      if (!transcode_audio) {
+        video_record.computeScore -= 0.1; // reduce 0.1 from the compute score because we're not transcoding audio
+      }
+
       let cmd = ffmpeg(file);
 
       cmd = cmd
@@ -293,6 +293,8 @@ export default function transcode (file) {
               pct_remaining,
               time_remaining,
               est_completed_seconds,
+              computeScore: video_record.computeScore,
+              priority: video_record.sortFields.priority,
               size: {
                 progress: {
                   kb: progress.targetSize,