Getting the queues indexing to work by passing them to the create background worker request bodies

Author: ericallam

packages/cli-v3/src/dev/devSupervisor.ts

@@ -113,6 +113,7 @@ class DevSupervisor implements WorkerRuntime {
 
   async initializeWorker(manifest: BuildManifest, stop: () => void): Promise<void> {
     if (this.lastManifest && this.lastManifest.contentHash === manifest.contentHash) {
+      logger.debug("worker skipped", { lastManifestContentHash: this.lastManifest?.contentHash });
       eventBus.emit("workerSkipped");
       stop();
       return;
@@ -126,6 +127,8 @@ class DevSupervisor implements WorkerRuntime {
       stop,
     });
 
+    logger.debug("initializing background worker", { manifest });
+
     await backgroundWorker.initialize();
 
     if (!backgroundWorker.manifest) {
@@ -153,6 +156,7 @@ class DevSupervisor implements WorkerRuntime {
         packageVersion: manifest.packageVersion,
         cliPackageVersion: manifest.cliPackageVersion,
         tasks: backgroundWorker.manifest.tasks,
+        queues: backgroundWorker.manifest.queues,
         contentHash: manifest.contentHash,
         sourceFiles,
       },

packages/cli-v3/src/entryPoints/managed-index-controller.ts

@@ -95,6 +95,7 @@ async function indexDeployment({
         packageVersion: buildManifest.packageVersion,
         cliPackageVersion: buildManifest.cliPackageVersion,
         tasks: workerManifest.tasks,
+        queues: workerManifest.queues,
         sourceFiles,
       },
       engine: "V2",

packages/core/src/v3/types/tasks.ts

@@ -193,7 +193,10 @@ type CommonTaskOptions<
     });
    * ```
    */
-  queue?: Queue;
+  queue?: {
+    name?: string;
+    concurrencyLimit?: number;
+  };
   /** Configure the spec of the [machine](https://trigger.dev/docs/machines) you want your task to run on.
    *
    * @example

packages/trigger-sdk/src/v3/shared.ts

@@ -219,8 +219,13 @@ export function createTask<
     },
   });
 
-  if (params.queue) {
-    resourceCatalog.registerQueueMetadata(params.queue);
+  const queue = params.queue;
+
+  if (queue && typeof queue.name === "string") {
+    resourceCatalog.registerQueueMetadata({
+      name: queue.name,
+      concurrencyLimit: queue.concurrencyLimit,
+    });
   }
 
   // @ts-expect-error
@@ -363,6 +368,15 @@ export function createSchemaTask<
     },
   });
 
+  const queue = params.queue;
+
+  if (queue && typeof queue.name === "string") {
+    resourceCatalog.registerQueueMetadata({
+      name: queue.name,
+      concurrencyLimit: queue.concurrencyLimit,
+    });
+  }
+
   // @ts-expect-error
   task[Symbol.for("trigger.dev/task")] = true;
 

references/hello-world/src/trigger/priority.ts

@@ -3,7 +3,7 @@ import { logger, task } from "@trigger.dev/sdk/v3";
 export const priorityParent = task({
   id: "priority-parent",
   run: async (payload: any, { ctx }) => {
-    logger.log("Hello, world from the parent", { payload });
+    logger.log("Hello, world from the parents", { payload });
 
     const batch1 = await priorityChild.batchTriggerAndWait([
       {

references/hello-world/src/trigger/queues.ts

@@ -10,7 +10,7 @@ export const queuesTester = task({
     }
 
     const retrievedFromId = await queues.retrieve(ctx.queue.id);
-    logger.log("Retrieved from ID", { retrievedFromId });
+    logger.log("Retrieved from IDs", { retrievedFromId });
 
     const retrievedFromCtxName = await queues.retrieve({
       type: "task",