v3: better handle null unicode characters when creating task events and completing a failed attempt

.changeset/shaggy-weeks-live.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+v3: sanitize errors with null unicode characters in some places

apps/webapp/app/entry.server.tsx

@@ -179,6 +179,30 @@ function logError(error: unknown, request?: Request) {
   }
 }
 
+process.on("uncaughtException", (error, origin) => {
+  if (
+    error instanceof Prisma.PrismaClientKnownRequestError ||
+    error instanceof Prisma.PrismaClientUnknownRequestError
+  ) {
+    // Don't exit the process if the error is a Prisma error
+    logger.error("uncaughtException prisma error", {
+      error,
+      prismaMessage: error.message,
+      code: "code" in error ? error.code : undefined,
+      meta: "meta" in error ? error.meta : undefined,
+      stack: error.stack,
+      origin,
+    });
+  } else {
+    logger.error("uncaughtException", {
+      error: { name: error.name, message: error.message, stack: error.stack },
+      origin,
+    });
+  }
+
+  process.exit(1);
+});
+
 const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);
 
 export { apiRateLimiter } from "./services/apiRateLimit.server";
@@ -188,6 +212,8 @@ export { registryProxy } from "./v3/registryProxy.server";
 export { runWithHttpContext } from "./services/httpAsyncStorage.server";
 import { eventLoopMonitor } from "./eventLoopMonitor.server";
 import { env } from "./env.server";
+import { logger } from "./services/logger.server";
+import { Prisma } from "./db.server";
 
 if (env.EVENT_LOOP_MONITOR_ENABLED === "1") {
   eventLoopMonitor.enable();

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

@@ -33,6 +33,8 @@ import { singleton } from "~/utils/singleton";
 import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server";
 import { startActiveSpan } from "./tracer.server";
 
+const MAX_FLUSH_DEPTH = 5;
+
 export type CreatableEvent = Omit<
   Prisma.TaskEventCreateInput,
   "id" | "createdAt" | "properties" | "metadata" | "style" | "output" | "payload"
@@ -1009,11 +1011,79 @@ export class EventRepository {
   async #flushBatch(batch: CreatableEvent[]) {
     const events = excludePartialEventsWithCorrespondingFullEvent(batch);
 
-    await this.db.taskEvent.createMany({
-      data: events as Prisma.TaskEventCreateManyInput[],
-    });
+    const flushedEvents = await this.#doFlushBatch(events);
 
-    this.#publishToRedis(events);
+    if (flushedEvents.length !== events.length) {
+      logger.debug("[EventRepository][flushBatch] Failed to insert all events", {
+        attemptCount: events.length,
+        successCount: flushedEvents.length,
+      });
+    }
+
+    this.#publishToRedis(flushedEvents);
+  }
+
+  async #doFlushBatch(events: CreatableEvent[], depth: number = 1): Promise<CreatableEvent[]> {
+    try {
+      await this.db.taskEvent.createMany({
+        data: events as Prisma.TaskEventCreateManyInput[],
+      });
+
+      return events;
+    } catch (error) {
+      if (error instanceof Prisma.PrismaClientUnknownRequestError) {
+        logger.error("Failed to insert events, most likely because of null characters", {
+          error: {
+            name: error.name,
+            message: error.message,
+            stack: error.stack,
+            clientVersion: error.clientVersion,
+          },
+        });
+
+        if (events.length === 1) {
+          logger.debug("Attempting to insert event individually and it failed", {
+            event: events[0],
+            error: {
+              name: error.name,
+              message: error.message,
+              stack: error.stack,
+              clientVersion: error.clientVersion,
+            },
+          });
+
+          return [];
+        }
+
+        if (depth > MAX_FLUSH_DEPTH) {
+          logger.error("Failed to insert events, reached maximum depth", {
+            error: {
+              name: error.name,
+              message: error.message,
+              stack: error.stack,
+              clientVersion: error.clientVersion,
+            },
+            depth,
+            eventsCount: events.length,
+          });
+
+          return [];
+        }
+
+        // Split the events into two batches, and recursively try to insert them.
+        const middle = Math.floor(events.length / 2);
+        const [firstHalf, secondHalf] = [events.slice(0, middle), events.slice(middle)];
+
+        const [firstHalfEvents, secondHalfEvents] = await Promise.all([
+          this.#doFlushBatch(firstHalf, depth + 1),
+          this.#doFlushBatch(secondHalf, depth + 1),
+        ]);
+
+        return firstHalfEvents.concat(secondHalfEvents);
+      }
+
+      throw error;
+    }
   }
 
   async #publishToRedis(events: CreatableEvent[]) {

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -6,6 +6,7 @@ import {
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
   flattenAttributes,
+  sanitizeError,
 } from "@trigger.dev/core/v3";
 import { PrismaClientOrTransaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -166,12 +167,14 @@ export class CompleteAttemptService extends BaseService {
       return "COMPLETED";
     }
 
+    const sanitizedError = sanitizeError(completion.error);
+
     await this._prisma.taskRunAttempt.update({
       where: { id: taskRunAttempt.id },
       data: {
         status: "FAILED",
         completedAt: new Date(),
-        error: completion.error,
+        error: sanitizedError,
         usageDurationMs: completion.usage?.durationMs,
       },
     });
@@ -289,15 +292,15 @@ export class CompleteAttemptService extends BaseService {
             name: "exception",
             time: new Date(),
             properties: {
-              exception: createExceptionPropertiesFromError(completion.error),
+              exception: createExceptionPropertiesFromError(sanitizedError),
             },
           },
         ],
       });
 
       if (
-        completion.error.type === "INTERNAL_ERROR" &&
-        completion.error.code === "GRACEFUL_EXIT_TIMEOUT"
+        sanitizedError.type === "INTERNAL_ERROR" &&
+        sanitizedError.code === "GRACEFUL_EXIT_TIMEOUT"
       ) {
         // We need to fail all incomplete spans
         const inProgressEvents = await eventRepository.queryIncompleteEvents({
@@ -310,7 +313,7 @@ export class CompleteAttemptService extends BaseService {
 
         const exception = {
           type: "Graceful exit timeout",
-          message: completion.error.message,
+          message: sanitizedError.message,
         };
 
         await Promise.all(

packages/core/src/v3/errors.ts

@@ -95,6 +95,40 @@ export function createJsonErrorObject(error: TaskRunError): SerializedError {
   }
 }
 
+// Removes any null characters from the error message
+export function sanitizeError(error: TaskRunError): TaskRunError {
+  switch (error.type) {
+    case "BUILT_IN_ERROR": {
+      return {
+        type: "BUILT_IN_ERROR",
+        message: error.message?.replace(/\0/g, ""),
+        name: error.name?.replace(/\0/g, ""),
+        stackTrace: error.stackTrace?.replace(/\0/g, ""),
+      };
+    }
+    case "STRING_ERROR": {
+      return {
+        type: "STRING_ERROR",
+        raw: error.raw.replace(/\0/g, ""),
+      };
+    }
+    case "CUSTOM_ERROR": {
+      return {
+        type: "CUSTOM_ERROR",
+        raw: error.raw.replace(/\0/g, ""),
+      };
+    }
+    case "INTERNAL_ERROR": {
+      return {
+        type: "INTERNAL_ERROR",
+        code: error.code,
+        message: error.message?.replace(/\0/g, ""),
+        stackTrace: error.stackTrace?.replace(/\0/g, ""),
+      };
+    }
+  }
+}
+
 export function correctErrorStackTrace(
   stackTrace: string,
   projectDir?: string,

packages/core/src/v3/otel/index.ts

@@ -4,12 +4,21 @@ export { TracingSDK, type TracingSDKConfig, type TracingDiagnosticLogLevel } fro
 
 export function recordSpanException(span: Span, error: unknown) {
   if (error instanceof Error) {
-    span.recordException(error);
+    span.recordException(sanitizeSpanError(error));
   } else if (typeof error === "string") {
-    span.recordException(new Error(error));
+    span.recordException(error.replace(/\0/g, ""));
   } else {
-    span.recordException(new Error(JSON.stringify(error)));
+    span.recordException(JSON.stringify(error).replace(/\0/g, ""));
   }
 
   span.setStatus({ code: SpanStatusCode.ERROR });
 }
+
+function sanitizeSpanError(error: Error) {
+  // Create a new error object with the same name, message and stack trace
+  const sanitizedError = new Error(error.message.replace(/\0/g, ""));
+  sanitizedError.name = error.name.replace(/\0/g, "");
+  sanitizedError.stack = error.stack?.replace(/\0/g, "");
+
+  return sanitizedError;
+}

packages/core/src/v3/workers/taskExecutor.ts

@@ -1,6 +1,6 @@
 import { SpanKind } from "@opentelemetry/api";
 import { ConsoleInterceptor } from "../consoleInterceptor";
-import { parseError } from "../errors";
+import { parseError, sanitizeError } from "../errors";
 import { TracingSDK, recordSpanException } from "../otel";
 import {
   BackgroundWorkerProperties,
@@ -169,9 +169,11 @@ export class TaskExecutor {
               return {
                 id: execution.run.id,
                 ok: false,
-                error: handleErrorResult.error
-                  ? parseError(handleErrorResult.error)
-                  : parseError(runError),
+                error: sanitizeError(
+                  handleErrorResult.error
+                    ? parseError(handleErrorResult.error)
+                    : parseError(runError)
+                ),
                 retry: handleErrorResult.status === "retry" ? handleErrorResult.retry : undefined,
                 skippedRetrying: handleErrorResult.status === "skipped",
               } satisfies TaskRunExecutionResult;

references/v3-catalog/src/trigger/other.ts

@@ -107,3 +107,18 @@ export const oomTask = task({
     }
   },
 });
+
+export const returnZeroCharacters = task({
+  id: "return-zero-characters",
+  run: async (payload: { forceError?: boolean }) => {
+    if (payload.forceError) {
+      throw new Error("All zeros: \u0000\x00\0");
+    }
+
+    return {
+      unicode: "\u0000",
+      hex: "\x00",
+      octal: "\0",
+    };
+  },
+});

references/v3-catalog/trigger.config.ts

@@ -61,7 +61,5 @@ export const config: TriggerConfig = {
   },
   onFailure: async (payload, error, { ctx }) => {
     console.log(`Task ${ctx.task.id} failed ${ctx.run.id}`);
-
-    throw error;
   },
 };