web/queue: make completedWorkers into set, require pipelineResults

web/src/lib/queen-bee/run-worker.ts

@@ -24,7 +24,7 @@ export const startWorker = async ({ worker, workerId, parentId, workerArgs }: Co
 
             if (files?.length === 0) {
                 const parent = get(queue)[parentId];
-                if (parent.state === "running" && parent.pipelineResults) {
+                if (parent.state === "running" && parent.pipelineResults.length) {
                     files = parent.pipelineResults;
                 }
             }

web/src/lib/queen-bee/scheduler.ts

@@ -29,15 +29,17 @@ export const checkTasks = () => {
         const task = queueItems[item];
 
         if (task.state === "running") {
-            // if the running worker isn't completed and wait to be called again
-            // (on worker completion)
-            if (!task.completedWorkers?.includes(task.runningWorker)) {
+            // if the running worker isn't completed, wait
+            // to be called again on worker completion
+            if (!task.completedWorkers.has(task.runningWorker)) {
                 break;
             }
 
-            // if all workers are completed, then return the final file and go to next task
-            if (task.completedWorkers.length === task.pipeline.length) {
-                const finalFile = task.pipelineResults?.pop();
+            // if all workers are completed, then return the
+            // the final file and go to the next task
+            if (task.completedWorkers.size === task.pipeline.length) {
+                const finalFile = task.pipelineResults.pop();
+
                 if (finalFile) {
                     itemDone(task.id, finalFile);
                     continue;
@@ -49,9 +51,9 @@ export const checkTasks = () => {
 
             // if current worker is completed, but there are more workers,
             // then start the next one and wait to be called again
-            for (let i = 0; i < task.pipeline.length; i++) {
-                if (!task.completedWorkers.includes(task.pipeline[i].workerId)) {
-                    startPipeline(task.pipeline[i]);
+            for (const worker of task.pipeline) {
+                if (!task.completedWorkers.has(worker.workerId)) {
+                    startPipeline(worker);
                     break;
                 }
             }

web/src/lib/state/queen-bee/queue.ts

@@ -5,7 +5,7 @@ import { clearFileStorage, removeFromFileStorage } from "$lib/storage";
 import { clearCurrentTasks, removeWorkerFromQueue } from "$lib/state/queen-bee/current-tasks";
 
 import type { CobaltFileReference } from "$lib/types/storage";
-import type { CobaltQueue, CobaltQueueItem } from "$lib/types/queue";
+import type { CobaltQueue, CobaltQueueItem, CobaltQueueItemRunning } from "$lib/types/queue";
 
 const clearPipelineCache = (queueItem: CobaltQueueItem) => {
     if (queueItem.state === "running" && queueItem.pipelineResults) {
@@ -74,10 +74,13 @@ export function itemDone(id: string, file: CobaltFileReference) {
 
 export function pipelineTaskDone(id: string, workerId: string, file: CobaltFileReference) {
     update(queueData => {
-        if (queueData[id] && queueData[id].state === "running") {
-            queueData[id].pipelineResults = [...queueData[id].pipelineResults || [], file];
-            queueData[id].completedWorkers = [...queueData[id].completedWorkers || [], workerId];
+        const item = queueData[id];
+
+        if (item && item.state === 'running') {
+            item.pipelineResults.push(file);
+            item.completedWorkers.add(workerId);
         }
+
         return queueData;
     });
 
@@ -87,13 +90,15 @@ export function pipelineTaskDone(id: string, workerId: string, file: CobaltFileR
 
 export function itemRunning(id: string, workerId: string) {
     update(queueData => {
-        if (queueData[id]) {
-            queueData[id] = {
-                ...queueData[id],
-                state: "running",
-                runningWorker: workerId,
-            }
+        const data = queueData[id] as CobaltQueueItemRunning;
+
+        if (data) {
+            data.state = 'running';
+            data.runningWorker = workerId;
+            data.completedWorkers ??= new Set();
+            data.pipelineResults ??= [];
         }
+
         return queueData;
     });
 

web/src/lib/types/queue.ts

@@ -23,8 +23,8 @@ export type CobaltQueueItemWaiting = CobaltQueueBaseItem & {
 export type CobaltQueueItemRunning = CobaltQueueBaseItem & {
     state: "running",
     runningWorker: string,
-    completedWorkers?: string[],
-    pipelineResults?: CobaltFileReference[],
+    completedWorkers: Set<string>,
+    pipelineResults: CobaltFileReference[],
 };
 
 export type CobaltQueueItemDone = CobaltQueueBaseItem & {