some fixes

Author: ericallam

apps/webapp/app/runEngine/services/batchTriggerV2.server.ts

@@ -122,7 +122,7 @@ export class RunEngineBatchTriggerServiceV2 extends WithRunEngine {
           const batchItems: BatchItem[] = body.items.map((item) => ({
             task: item.task,
             payload: item.payload,
-            payloadType: (item as { payloadType?: string }).payloadType,
+            payloadType: item.options?.payloadType ?? "application/json",
             options: item.options as Record<string, unknown> | undefined,
           }));
 

apps/webapp/app/v3/runEngine.server.ts

@@ -160,23 +160,26 @@ function createRunEngine() {
     },
     // BatchQueue with DRR scheduling for fair batch processing
     // Consumers are controlled by options.worker.disabled (same as main worker)
-    batchQueue: env.BATCH_TRIGGER_WORKER_ENABLED === "true" ? {
-      redis: {
-        keyPrefix: "engine:",
-        port: env.BATCH_TRIGGER_WORKER_REDIS_PORT ?? undefined,
-        host: env.BATCH_TRIGGER_WORKER_REDIS_HOST ?? undefined,
-        username: env.BATCH_TRIGGER_WORKER_REDIS_USERNAME ?? undefined,
-        password: env.BATCH_TRIGGER_WORKER_REDIS_PASSWORD ?? undefined,
-        enableAutoPipelining: true,
-        ...(env.BATCH_TRIGGER_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
-      },
-      drr: {
-        quantum: env.BATCH_QUEUE_DRR_QUANTUM,
-        maxDeficit: env.BATCH_QUEUE_MAX_DEFICIT,
-      },
-      consumerCount: env.BATCH_QUEUE_CONSUMER_COUNT,
-      consumerIntervalMs: env.BATCH_QUEUE_CONSUMER_INTERVAL_MS,
-    } : undefined,
+    batchQueue:
+      env.BATCH_TRIGGER_WORKER_ENABLED === "true"
+        ? {
+            redis: {
+              keyPrefix: "engine:",
+              port: env.BATCH_TRIGGER_WORKER_REDIS_PORT ?? undefined,
+              host: env.BATCH_TRIGGER_WORKER_REDIS_HOST ?? undefined,
+              username: env.BATCH_TRIGGER_WORKER_REDIS_USERNAME ?? undefined,
+              password: env.BATCH_TRIGGER_WORKER_REDIS_PASSWORD ?? undefined,
+              enableAutoPipelining: true,
+              ...(env.BATCH_TRIGGER_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+            },
+            drr: {
+              quantum: env.BATCH_QUEUE_DRR_QUANTUM,
+              maxDeficit: env.BATCH_QUEUE_MAX_DEFICIT,
+            },
+            consumerCount: env.BATCH_QUEUE_CONSUMER_COUNT,
+            consumerIntervalMs: env.BATCH_QUEUE_CONSUMER_INTERVAL_MS,
+          }
+        : undefined,
   });
 
   // Set up BatchQueue callbacks if enabled
@@ -187,6 +190,31 @@ function createRunEngine() {
   return engine;
 }
 
+/**
+ * Normalize the payload from BatchQueue.
+ * The payload might be a JSON string if the SDK sent it pre-serialized.
+ * If it's a JSON string and payloadType is "application/json", parse it
+ * to avoid double-stringification in DefaultPayloadProcessor.
+ */
+function normalizePayload(payload: unknown, payloadType?: string): unknown {
+  // Only normalize for JSON payloads
+  if (payloadType !== "application/json" && payloadType !== undefined) {
+    return payload;
+  }
+
+  // If payload is a string, try to parse it as JSON
+  if (typeof payload === "string") {
+    try {
+      return JSON.parse(payload);
+    } catch {
+      // If it's not valid JSON, return as-is
+      return payload;
+    }
+  }
+
+  return payload;
+}
+
 /**
  * Set up the BatchQueue processing callbacks.
  * These handle creating runs from batch items and completing batches.
@@ -197,6 +225,9 @@ function setupBatchQueueCallbacks(engine: RunEngine) {
     try {
       const triggerTaskService = new TriggerTaskService();
 
+      // Normalize payload to avoid double-stringification
+      const payload = normalizePayload(item.payload, item.payloadType);
+
       const result = await triggerTaskService.call(
         item.task,
         {
@@ -208,7 +239,7 @@ function setupBatchQueueCallbacks(engine: RunEngine) {
           project: { id: meta.projectId },
         } as AuthenticatedEnvironment,
         {
-          payload: item.payload,
+          payload,
           options: {
             ...(item.options as Record<string, unknown>),
             payloadType: item.payloadType,

internal-packages/run-engine/src/batch-queue/drrScheduler.ts

@@ -174,15 +174,30 @@ export class DRRScheduler {
   }
 
   /**
-   * Record a successful run for a batch.
+   * Record a successful run for a batch and increment processed count.
+   * Returns the new processed count.
    */
-  async recordSuccess(batchId: string, runId: string): Promise<void> {
+  async recordSuccess(batchId: string, runId: string): Promise<number> {
     const runsKey = this.keys.batchRunsKey(batchId);
-    await this.redis.rpush(runsKey, runId);
+    const processedKey = this.keys.batchProcessedCountKey(batchId);
+
+    // Use a pipeline to atomically record and increment
+    const pipeline = this.redis.pipeline();
+    pipeline.rpush(runsKey, runId);
+    pipeline.incr(processedKey);
+
+    const results = await pipeline.exec();
+    // incr result is the second command, returns [error, value]
+    const incrResult = results?.[1];
+    if (incrResult?.[0]) {
+      throw incrResult[0];
+    }
+    return incrResult?.[1] as number;
   }
 
   /**
-   * Record a failure for a batch item.
+   * Record a failure for a batch item and increment processed count.
+   * Returns the new processed count.
    */
   async recordFailure(
     batchId: string,
@@ -194,13 +209,26 @@ export class DRRScheduler {
       error: string;
       errorCode?: string;
     }
-  ): Promise<void> {
+  ): Promise<number> {
     const failuresKey = this.keys.batchFailuresKey(batchId);
+    const processedKey = this.keys.batchProcessedCountKey(batchId);
     const failureRecord = {
       ...failure,
       timestamp: Date.now(),
     };
-    await this.redis.rpush(failuresKey, JSON.stringify(failureRecord));
+
+    // Use a pipeline to atomically record and increment
+    const pipeline = this.redis.pipeline();
+    pipeline.rpush(failuresKey, JSON.stringify(failureRecord));
+    pipeline.incr(processedKey);
+
+    const results = await pipeline.exec();
+    // incr result is the second command, returns [error, value]
+    const incrResult = results?.[1];
+    if (incrResult?.[0]) {
+      throw incrResult[0];
+    }
+    return incrResult?.[1] as number;
   }
 
   /**
@@ -240,6 +268,7 @@ export class DRRScheduler {
       this.keys.batchMetaKey(batchId),
       this.keys.batchRunsKey(batchId),
       this.keys.batchFailuresKey(batchId),
+      this.keys.batchProcessedCountKey(batchId),
     ];
     await this.redis.del(...keys);
   }

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

@@ -278,6 +278,8 @@ export class BatchQueue {
 
   /**
    * Process a single dequeued item.
+   * Uses processed count (not isBatchComplete from dequeue) to determine when to finalize.
+   * This prevents race conditions when multiple consumers process items concurrently.
    */
   async #processItem(dequeued: {
     envId: string;
@@ -288,13 +290,15 @@ export class BatchQueue {
     isBatchComplete: boolean;
     envHasMoreBatches: boolean;
   }): Promise<void> {
-    const { batchId, itemIndex, item, meta, isBatchComplete } = dequeued;
+    const { batchId, itemIndex, item, meta } = dequeued;
 
     if (!this.processItemCallback) {
       this.logger.error("No process item callback set", { batchId, itemIndex });
       return;
     }
 
+    let processedCount: number;
+
     try {
       const result = await this.processItemCallback({
         batchId,
@@ -305,16 +309,18 @@ export class BatchQueue {
       });
 
       if (result.success) {
-        await this.scheduler.recordSuccess(batchId, result.runId);
+        processedCount = await this.scheduler.recordSuccess(batchId, result.runId);
         this.logger.debug("Batch item processed successfully", {
           batchId,
           itemIndex,
           runId: result.runId,
+          processedCount,
+          expectedCount: meta.runCount,
         });
       } else {
         const payloadStr =
           typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload);
-        await this.scheduler.recordFailure(batchId, {
+        processedCount = await this.scheduler.recordFailure(batchId, {
           index: itemIndex,
           taskIdentifier: item.task,
           payload: payloadStr?.substring(0, 1000), // Truncate large payloads
@@ -326,13 +332,15 @@ export class BatchQueue {
           batchId,
           itemIndex,
           error: result.error,
+          processedCount,
+          expectedCount: meta.runCount,
         });
       }
     } catch (error) {
       // Unexpected error during processing
       const payloadStr =
         typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload);
-      await this.scheduler.recordFailure(batchId, {
+      processedCount = await this.scheduler.recordFailure(batchId, {
         index: itemIndex,
         taskIdentifier: item.task,
         payload: payloadStr?.substring(0, 1000),
@@ -344,11 +352,19 @@ export class BatchQueue {
         batchId,
         itemIndex,
         error: error instanceof Error ? error.message : String(error),
+        processedCount,
+        expectedCount: meta.runCount,
       });
     }
 
-    // If batch is complete, finalize it
-    if (isBatchComplete) {
+    // Check if all items have been processed using atomic counter
+    // This is safe even with multiple concurrent consumers
+    if (processedCount === meta.runCount) {
+      this.logger.debug("All items processed, finalizing batch", {
+        batchId,
+        processedCount,
+        expectedCount: meta.runCount,
+      });
       await this.#finalizeBatch(batchId, meta);
     }
   }

internal-packages/run-engine/src/batch-queue/keyProducer.ts

@@ -9,6 +9,7 @@ const constants = {
   QUEUE_PART: "queue",
   RUNS_PART: "runs",
   FAILURES_PART: "failures",
+  PROCESSED_COUNT_PART: "processed",
 } as const;
 
 /**
@@ -22,6 +23,7 @@ const constants = {
  * - batch:{batchId}:meta - Per-batch metadata
  * - batch:{batchId}:runs - Per-batch successful runs (list of runIds)
  * - batch:{batchId}:failures - Per-batch failures (list of failure JSON)
+ * - batch:{batchId}:processed - Per-batch processed count (atomic counter)
  */
 export class BatchQueueFullKeyProducer implements BatchQueueKeyProducer {
   /**
@@ -80,6 +82,14 @@ export class BatchQueueFullKeyProducer implements BatchQueueKeyProducer {
     return [constants.BATCH_PART, batchId, constants.FAILURES_PART].join(":");
   }
 
+  /**
+   * Key for a batch's processed count.
+   * Atomic counter tracking how many items have been processed (success or failure).
+   */
+  batchProcessedCountKey(batchId: string): string {
+    return [constants.BATCH_PART, batchId, constants.PROCESSED_COUNT_PART].join(":");
+  }
+
   /**
    * Create a master queue member value from envId and batchId.
    * Format: "{envId}:{batchId}"

internal-packages/run-engine/src/batch-queue/types.ts

@@ -152,6 +152,8 @@ export interface BatchQueueKeyProducer {
   batchRunsKey(batchId: string): string;
   /** Key for a batch's failure list */
   batchFailuresKey(batchId: string): string;
+  /** Key for a batch's processed count (atomic counter) */
+  batchProcessedCountKey(batchId: string): string;
 
   // Master queue member utilities
   /** Create a master queue member value: "{envId}:{batchId}" */