more wip

Author: ericallam

internal-packages/redis-worker/package.json

@@ -6,10 +6,9 @@
   "types": "./src/index.ts",
   "type": "module",
   "dependencies": {
-    "@opentelemetry/api": "^1.9.0",
+    "@internal/tracing": "workspace:*",
     "@internal/redis": "workspace:*",
     "@trigger.dev/core": "workspace:*",
-    "ioredis": "^5.3.2",
     "lodash.omit": "^4.5.0",
     "nanoid": "^5.0.7",
     "p-limit": "^6.2.0",
@@ -24,4 +23,4 @@
     "typecheck": "tsc --noEmit",
     "test": "vitest --no-file-parallelism"
   }
-}
+}

internal-packages/redis-worker/src/index.ts

@@ -1,2 +1,2 @@
-export * from "./queue";
-export * from "./worker";
+export * from "./queue.js";
+export * from "./worker.js";

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

@@ -1,6 +1,11 @@
-import { createRedisClient } from "@internal/redis";
+import {
+  createRedisClient,
+  type Redis,
+  type Callback,
+  type RedisOptions,
+  type Result,
+} from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
-import Redis, { type Callback, type RedisOptions, type Result } from "ioredis";
 import { nanoid } from "nanoid";
 import { z } from "zod";
 
@@ -436,7 +441,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
   }
 }
 
-declare module "ioredis" {
+declare module "@internal/redis" {
   interface RedisCommander<Context> {
     enqueueItem(
       //keys

internal-packages/redis-worker/src/telemetry.ts

@@ -4,7 +4,6 @@ import { describe } from "node:test";
 import { expect } from "vitest";
 import { z } from "zod";
 import { Worker } from "./worker.js";
-import Redis from "ioredis";
 import { createRedisClient } from "@internal/redis";
 
 describe("Worker", () => {

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

@@ -1,13 +1,11 @@
-import { SpanKind, trace, Tracer } from "@opentelemetry/api";
+import { SpanKind, startSpan, trace, Tracer } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { calculateNextRetryDelay } from "@trigger.dev/core/v3";
 import { type RetryOptions } from "@trigger.dev/core/v3/schemas";
-import { type RedisOptions } from "ioredis";
+import { Redis, type RedisOptions } from "@internal/redis";
 import { z } from "zod";
 import { AnyQueueItem, SimpleQueue } from "./queue.js";
-import Redis from "ioredis";
 import { nanoid } from "nanoid";
-import { startSpan } from "./telemetry.js";
 import pLimit from "p-limit";
 import { createRedisClient } from "@internal/redis";
 

internal-packages/redis-worker/src/worker.ts

@@ -2,8 +2,8 @@
   "compilerOptions": {
     "target": "ES2019",
     "lib": ["ES2019", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
-    "module": "CommonJS",
-    "moduleResolution": "Node",
+    "module": "Node16",
+    "moduleResolution": "Node16",
     "moduleDetection": "force",
     "verbatimModuleSyntax": false,
     "types": ["vitest/globals"],

internal-packages/redis-worker/tsconfig.json

@@ -1,7 +1,7 @@
 import { Redis, RedisOptions } from "ioredis";
 import { Logger } from "@trigger.dev/core/logger";
 
-export { Redis, type Callback, type RedisOptions, type Result } from "ioredis";
+export { Redis, type Callback, type RedisOptions, type Result, type RedisCommander } from "ioredis";
 
 const defaultOptions: Partial<RedisOptions> = {
   retryStrategy: (times: number) => {

internal-packages/redis/src/index.ts

@@ -12,7 +12,6 @@
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/database": "workspace:*",
     "assert-never": "^1.2.1",
-    "ioredis": "^5.3.2",
     "nanoid": "^3.3.4",
     "redlock": "5.0.0-beta.2",
     "zod": "3.23.8",

internal-packages/run-engine/package.json

@@ -1,5 +1,5 @@
 import { TaskRunExecutionStatus, TaskRunStatus } from "@trigger.dev/database";
-import { AuthenticatedEnvironment } from "../shared";
+import { AuthenticatedEnvironment } from "../shared/index.js";
 import { FlushedRunMetadata, TaskRunError } from "@trigger.dev/core/v3";
 
 export type EventBusEvents = {

internal-packages/run-engine/src/engine/eventBus.ts

@@ -1,6 +1,6 @@
-import { createRedisClient } from "@internal/redis";
+import { createRedisClient, Redis } from "@internal/redis";
 import { Worker } from "@internal/redis-worker";
-import { Attributes, Span, SpanKind, trace, Tracer } from "@opentelemetry/api";
+import { Attributes, Span, SpanKind, trace, Tracer } from "@internal/tracing";
 import { assertExhaustive } from "@trigger.dev/core";
 import { Logger } from "@trigger.dev/core/logger";
 import {
@@ -48,29 +48,29 @@ import {
   TaskRunStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import assertNever from "assert-never";
-import { Redis } from "ioredis";
+import { assertNever } from "assert-never";
 import { nanoid } from "nanoid";
 import { EventEmitter } from "node:events";
 import { z } from "zod";
-import { RunQueue } from "../run-queue";
-import { SimpleWeightedChoiceStrategy } from "../run-queue/simpleWeightedPriorityStrategy";
-import { MinimalAuthenticatedEnvironment } from "../shared";
-import { MAX_TASK_RUN_ATTEMPTS } from "./consts";
-import { getRunWithBackgroundWorkerTasks } from "./db/worker";
-import { runStatusFromError } from "./errors";
-import { EventBusEvents } from "./eventBus";
-import { executionResultFromSnapshot, getLatestExecutionSnapshot } from "./executionSnapshots";
-import { RunLocker } from "./locking";
-import { getMachinePreset } from "./machinePresets";
+import { RunQueue } from "../run-queue/index.js";
+import { FairDequeuingStrategy } from "../run-queue/fairDequeuingStrategy.js";
+import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
+import { MAX_TASK_RUN_ATTEMPTS } from "./consts.js";
+import { getRunWithBackgroundWorkerTasks } from "./db/worker.js";
+import { runStatusFromError } from "./errors.js";
+import { EventBusEvents } from "./eventBus.js";
+import { executionResultFromSnapshot, getLatestExecutionSnapshot } from "./executionSnapshots.js";
+import { RunLocker } from "./locking.js";
+import { getMachinePreset } from "./machinePresets.js";
 import {
   isCheckpointable,
   isDequeueableExecutionStatus,
   isExecuting,
   isFinalRunStatus,
   isPendingExecuting,
-} from "./statuses";
-import { HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types";
+} from "./statuses.js";
+import { HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js";
+import { RunQueueShortKeyProducer } from "../run-queue/keyProducer.js";
 import { retryOutcomeFromCompletion } from "./retrying";
 
 const workerCatalog = {
@@ -153,11 +153,16 @@ export class RunEngine {
     );
     this.runLock = new RunLocker({ redis: this.runLockRedis });
 
+    const keys = new RunQueueShortKeyProducer("rq:");
+
     this.runQueue = new RunQueue({
       name: "rq",
       tracer: trace.getTracer("rq"),
-      queuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 36 }),
-      envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
+      keys,
+      queuePriorityStrategy: new FairDequeuingStrategy({
+        keys,
+        redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
+      }),
       defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10,
       logger: new Logger("RunQueue", "debug"),
       redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },

internal-packages/run-engine/src/engine/index.ts

@@ -1,6 +1,6 @@
-import Redis from "ioredis";
 import Redlock, { RedlockAbortSignal } from "redlock";
 import { AsyncLocalStorage } from "async_hooks";
+import { Redis } from "@internal/redis";
 
 interface LockContext {
   resources: string;

internal-packages/run-engine/src/engine/locking.ts

@@ -3,7 +3,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { generateFriendlyId } from "@trigger.dev/core/v3/apps";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";

internal-packages/run-engine/src/engine/tests/batchTrigger.test.ts

@@ -4,7 +4,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect, describe } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "node:timers/promises";

internal-packages/run-engine/src/engine/tests/batchTriggerAndWait.test.ts

@@ -4,7 +4,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/cancelling.test.ts

@@ -5,7 +5,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/checkpoints.test.ts

@@ -4,7 +4,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/delays.test.ts

@@ -3,7 +3,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { generateFriendlyId } from "@trigger.dev/core/v3/apps";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";

internal-packages/run-engine/src/engine/tests/dequeuing.test.ts

@@ -4,7 +4,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect, describe } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/heartbeats.test.ts

@@ -4,7 +4,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/notDeployed.test.ts

@@ -3,7 +3,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { generateFriendlyId } from "@trigger.dev/core/v3/apps";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";

internal-packages/run-engine/src/engine/tests/priority.test.ts

@@ -4,7 +4,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { EventBusEventArgs } from "../eventBus.js";
 import { RunEngine } from "../index.js";

internal-packages/run-engine/src/engine/tests/trigger.test.ts

@@ -4,7 +4,7 @@ import {
   setupAuthenticatedEnvironment,
   setupBackgroundWorker,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "node:timers/promises";

internal-packages/run-engine/src/engine/tests/triggerAndWait.test.ts

@@ -4,7 +4,7 @@ import {
   setupBackgroundWorker,
   assertNonNullable,
 } from "@internal/testcontainers";
-import { trace } from "@opentelemetry/api";
+import { trace } from "@internal/tracing";
 import { expect } from "vitest";
 import { RunEngine } from "../index.js";
 import { setTimeout } from "timers/promises";

internal-packages/run-engine/src/engine/tests/ttl.test.ts

@@ -1,9 +1,9 @@
 import { type WorkerConcurrencyOptions } from "@internal/redis-worker";
-import { Tracer } from "@opentelemetry/api";
+import { Tracer } from "@internal/tracing";
 import { MachinePreset, MachinePresetName, QueueOptions, RetryOptions } from "@trigger.dev/core/v3";
 import { PrismaClient } from "@trigger.dev/database";
-import { type RedisOptions } from "ioredis";
-import { MinimalAuthenticatedEnvironment } from "../shared";
+import { type RedisOptions } from "@internal/redis";
+import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
 
 export type RunEngineOptions = {
   prisma: PrismaClient;