fixed configuration

Author: ericallam

apps/webapp/app/env.server.ts

@@ -415,8 +415,9 @@ const EnvironmentSchema = z.object({
 
   MARQS_SHARED_WORKER_QUEUE_EAGER_DEQUEUE_ENABLED: z.string().default("0"),
   MARQS_WORKER_ENABLED: z.string().default("0"),
-  MARQS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
-  MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
+  MARQS_WORKER_COUNT: z.coerce.number().int().default(2),
+  MARQS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
+  MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(5),
   MARQS_WORKER_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
   MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
   MARQS_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),

apps/webapp/app/v3/marqs/index.server.ts

@@ -8,6 +8,7 @@ import {
   trace,
   Tracer,
 } from "@opentelemetry/api";
+import { type RedisOptions } from "@internal/redis";
 import {
   SEMATTRS_MESSAGE_ID,
   SEMATTRS_MESSAGING_SYSTEM,
@@ -84,6 +85,7 @@ export type MarQSOptions = {
     shutdownTimeoutMs?: number;
     concurrency?: WorkerConcurrencyOptions;
     enabled?: boolean;
+    redisOptions: RedisOptions;
   };
 };
 
@@ -116,10 +118,7 @@ export class MarQS {
 
     this.worker = new Worker({
       name: "marqs-worker",
-      redisOptions: {
-        ...options.redis.options,
-        keyPrefix: `${options.redis.options.keyPrefix}:worker`,
-      },
+      redisOptions: options.workerOptions.redisOptions,
       catalog: workerCatalog,
       concurrency: options.workerOptions?.concurrency,
       pollIntervalMs: options.workerOptions?.pollIntervalMs ?? 1000,
@@ -2621,10 +2620,19 @@ function getMarQSClient() {
       immediatePollIntervalMs: env.MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS,
       shutdownTimeoutMs: env.MARQS_WORKER_SHUTDOWN_TIMEOUT_MS,
       concurrency: {
-        workers: env.MARQS_WORKER_CONCURRENCY_LIMIT,
+        workers: env.MARQS_WORKER_COUNT,
         tasksPerWorker: env.MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER,
         limit: env.MARQS_WORKER_CONCURRENCY_LIMIT,
       },
+      redisOptions: {
+        keyPrefix: KEY_PREFIX,
+        port: env.REDIS_PORT ?? undefined,
+        host: env.REDIS_HOST ?? undefined,
+        username: env.REDIS_USERNAME ?? undefined,
+        password: env.REDIS_PASSWORD ?? undefined,
+        enableAutoPipelining: true,
+        ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+      },
     },
   });
 }