Merge remote-tracking branch 'origin/main' into increase-v4-worker-visibility-timeouts

Author: matt-aitken

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx

@@ -285,7 +285,7 @@ export default function Page() {
           {success ? (
             <div
               className={cn(
-                "grid h-fit max-h-full min-h-full grid-rows-[1fr] overflow-x-auto",
+                "grid max-h-full min-h-full grid-rows-[1fr] overflow-x-auto",
                 pagination.totalPages > 1 && "grid-rows-[1fr_auto]"
               )}
             >

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules/route.tsx

@@ -229,7 +229,7 @@ export default function Page() {
 
                   <div
                     className={cn(
-                      "grid h-fit max-h-full min-h-full overflow-x-auto",
+                      "grid max-h-full min-h-full overflow-x-auto",
                       totalPages > 1 ? "grid-rows-[1fr_auto]" : "grid-rows-[1fr]"
                     )}
                   >

docs/upgrade-to-v4.mdx

@@ -508,7 +508,7 @@ pnpm dlx trigger.dev@v4-beta dev
 
 During the beta we will be tracking issues and releasing regular fixes.
 
-There are no known issues at the moment.
+**ISSUE:** Runs not continuing after a child run(s) have completed.
 
 ## Deprecations
 
@@ -703,7 +703,8 @@ import { task } from "@trigger.dev/sdk";
 export const myTask = task({
   id: "my-task",
   onStart: ({ payload, ctx }) => {},
-  run: async ({ payload, ctx }) => {},
+  // The run function still uses separate parameters
+  run: async ( payload, { ctx }) => {},
 });
 ```
 
@@ -721,7 +722,7 @@ export const myTask = task({
   onResume: ({ payload, ctx, task, wait }) => {},
   onComplete: ({ payload, ctx, task, result }) => {},
   catchError: ({ payload, ctx, task, error, retry, retryAt, retryDelayInMs }) => {},
-  run: async ({ payload, ctx }) => {},
+  run: async (payload, { ctx }) => {},
 });
 ```
 
@@ -731,3 +732,32 @@ We've made a few small changes to the `ctx` object:
 
 - `ctx.attempt.id` and `ctx.attempt.status` have been removed. `ctx.attempt.number` is still available.
 - `ctx.task.exportName` has been removed (since we no longer require tasks to be exported to be triggered).
+
+### BatchTrigger changes
+
+The `batchTrigger` function no longer returns a `runs` list directly. In v3, you could access the runs directly from the batch handle:
+
+```ts
+// In v3
+const batchHandle = await tasks.batchTrigger([
+  [myTask, { foo: "bar" }],
+  [myOtherTask, { baz: "qux" }],
+]);
+
+// You could access runs directly
+console.log(batchHandle.runs);
+```
+
+In v4, you now need to use the `runs.list()` method to get the list of runs:
+
+```ts
+// In v4
+const batchHandle = await tasks.batchTrigger([
+  [myTask, { foo: "bar" }],
+  [myOtherTask, { baz: "qux" }],
+]);
+
+// Now you need to call runs.list()
+const runs = await batchHandle.runs.list();
+console.log(runs);
+```

packages/redis-worker/src/queue.test.ts

@@ -4,6 +4,7 @@ import { expect } from "vitest";
 import { z } from "zod";
 import { SimpleQueue } from "./queue.js";
 import { Logger } from "@trigger.dev/core/logger";
+import { createRedisClient } from "@internal/redis";
 
 describe("SimpleQueue", () => {
   redisTest("enqueue/dequeue", { timeout: 20_000 }, async ({ redisContainer }) => {
@@ -209,6 +210,10 @@ describe("SimpleQueue", () => {
           timestamp: expect.any(Date),
         })
       );
+
+      // Acknowledge the item and verify it's removed
+      await queue.ack(second!.id);
+      expect(await queue.size({ includeFuture: true })).toBe(0);
     } finally {
       await queue.close();
     }
@@ -328,6 +333,7 @@ describe("SimpleQueue", () => {
 
       // Redrive item from DLQ
       await queue.redriveFromDeadLetterQueue("1");
+      await new Promise((resolve) => setTimeout(resolve, 200));
       expect(await queue.size()).toBe(1);
       expect(await queue.sizeOfDeadLetterQueue()).toBe(0);
 
@@ -357,4 +363,64 @@ describe("SimpleQueue", () => {
       await queue.close();
     }
   });
+
+  redisTest("cleanup orphaned queue entries", { timeout: 20_000 }, async ({ redisContainer }) => {
+    const queue = new SimpleQueue({
+      name: "test-orphaned",
+      schema: {
+        test: z.object({
+          value: z.number(),
+        }),
+      },
+      redisOptions: {
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      },
+      logger: new Logger("test", "log"),
+    });
+
+    try {
+      // First, add a normal item
+      await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
+
+      const redisClient = createRedisClient({
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      });
+
+      // Manually add an orphaned item to the queue (without corresponding hash entry)
+      await redisClient.zadd(`{queue:test-orphaned:}queue`, Date.now(), "orphaned-id");
+
+      // Verify both items are in the queue
+      expect(await queue.size()).toBe(2);
+
+      // Dequeue should process both items, but only return the valid one
+      // and clean up the orphaned entry
+      const dequeued = await queue.dequeue(2);
+
+      // Should only get the valid item
+      expect(dequeued).toHaveLength(1);
+      expect(dequeued[0]).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
+
+      // The orphaned item should have been removed
+      expect(await queue.size({ includeFuture: true })).toBe(1);
+
+      // Verify the orphaned ID is no longer in the queue
+      const orphanedScore = await redisClient.zscore(`{queue:test-orphaned:}queue`, "orphaned-id");
+      expect(orphanedScore).toBeNull();
+    } finally {
+      await queue.close();
+    }
+  });
 });

packages/redis-worker/src/queue.ts

@@ -130,6 +130,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
       throw e;
     }
   }
+
   async dequeue(count: number = 1): Promise<Array<QueueItem<TMessageCatalog>>> {
     const now = Date.now();
 
@@ -179,9 +180,6 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
         }
 
         const visibilityTimeoutMs = parsedItem.visibilityTimeoutMs as number;
-        const invisibleUntil = now + visibilityTimeoutMs;
-
-        await this.redis.zadd(`queue`, invisibleUntil, id);
 
         dequeuedItems.push({
           id,
@@ -374,6 +372,9 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
 
             redis.call('ZADD', queue, invisibleUntil, id)
             table.insert(dequeued, {id, serializedItem, score})
+          else
+            -- Remove the orphaned queue entry if no corresponding item exists
+            redis.call('ZREM', queue, id)
           end
         end