performance improvements

Author: ericallam

apps/webapp/app/entry.server.tsx

@@ -240,6 +240,7 @@ import { logger } from "./services/logger.server";
 import { Prisma } from "./db.server";
 import { registerRunEngineEventBusHandlers } from "./v3/runEngineHandlers.server";
 import { remoteBuildsEnabled } from "./v3/remoteImageBuilder.server";
+import { resourceMonitor } from "./services/resourceMonitor.server";
 
 if (env.EVENT_LOOP_MONITOR_ENABLED === "1") {
   eventLoopMonitor.enable();
@@ -250,3 +251,7 @@ if (remoteBuildsEnabled()) {
 } else {
   console.log("🏗️  Local builds enabled");
 }
+
+if (env.RESOURCE_MONITOR_ENABLED === "1") {
+  resourceMonitor.startMonitoring(1000);
+}

apps/webapp/app/env.server.ts

@@ -492,6 +492,7 @@ const EnvironmentSchema = z
     CENTS_PER_RUN: z.coerce.number().default(0),
 
     EVENT_LOOP_MONITOR_ENABLED: z.string().default("1"),
+    RESOURCE_MONITOR_ENABLED: z.string().default("0"),
     MAXIMUM_LIVE_RELOADING_EVENTS: z.coerce.number().int().default(1000),
     MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
     MAXIMUM_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(10_000),
@@ -1122,7 +1123,8 @@ const EnvironmentSchema = z
     EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().optional(),
     EVENT_REPOSITORY_DEFAULT_STORE: z.enum(["postgres", "clickhouse"]).default("postgres"),
-    EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(250_000),
+    EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
+    EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000),
 
     // Bootstrap
     TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),

apps/webapp/app/eventLoopMonitor.server.ts

@@ -65,6 +65,8 @@ function after(asyncId: number) {
       cached.parentCtx
     );
 
+    console.log("🥸  Event loop blocked", { asyncType: cached.type, time });
+
     newSpan.end();
   }
 }

apps/webapp/app/services/resourceMonitor.server.ts

@@ -0,0 +1,11 @@
+import { ResourceMonitor } from "@trigger.dev/core/v3/serverOnly";
+import { singleton } from "~/utils/singleton";
+
+export const resourceMonitor = singleton("resourceMonitor", initializeResourceMonitor);
+
+function initializeResourceMonitor() {
+  return new ResourceMonitor({
+    ctx: {},
+    verbose: false,
+  });
+}

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

@@ -1,4 +1,5 @@
 import { Logger } from "@trigger.dev/core/logger";
+import { tryCatch } from "@trigger.dev/core/utils";
 import { nanoid } from "nanoid";
 import pLimit from "p-limit";
 import { signalsEmitter } from "~/services/signals.server";
@@ -195,47 +196,62 @@ export class DynamicFlushScheduler<T> {
     // Schedule all batches for concurrent processing
     const flushPromises = batchesToFlush.map((batch) =>
       this.limiter(async () => {
-        const flushId = nanoid();
         const itemCount = batch.length;
 
-        try {
-          const startTime = Date.now();
-          await this.callback(flushId, batch);
-
-          const duration = Date.now() - startTime;
-          this.totalQueuedItems -= itemCount;
-          this.consecutiveFlushFailures = 0;
-          this.lastFlushTime = Date.now();
-          this.metrics.flushedBatches++;
-          this.metrics.totalItemsFlushed += itemCount;
-
-          this.logger.debug("Batch flushed successfully", {
-            flushId,
-            itemCount,
-            duration,
-            remainingQueueDepth: this.totalQueuedItems,
-            activeConcurrency: this.limiter.activeCount,
-            pendingConcurrency: this.limiter.pendingCount,
-          });
-        } catch (error) {
-          this.consecutiveFlushFailures++;
-          this.metrics.failedBatches++;
+        const self = this;
+
+        async function tryFlush(flushId: string, batchToFlush: T[], attempt: number = 1) {
+          try {
+            const startTime = Date.now();
+            await self.callback(flushId, batchToFlush);
+
+            const duration = Date.now() - startTime;
+            self.totalQueuedItems -= itemCount;
+            self.consecutiveFlushFailures = 0;
+            self.lastFlushTime = Date.now();
+            self.metrics.flushedBatches++;
+            self.metrics.totalItemsFlushed += itemCount;
+
+            self.logger.debug("Batch flushed successfully", {
+              flushId,
+              itemCount,
+              duration,
+              remainingQueueDepth: self.totalQueuedItems,
+              activeConcurrency: self.limiter.activeCount,
+              pendingConcurrency: self.limiter.pendingCount,
+            });
+          } catch (error) {
+            self.consecutiveFlushFailures++;
+            self.metrics.failedBatches++;
+
+            self.logger.error("Error attempting to flush batch", {
+              flushId,
+              itemCount,
+              error,
+              consecutiveFailures: self.consecutiveFlushFailures,
+              attempt,
+            });
+
+            // Back off on failures
+            if (self.consecutiveFlushFailures > 5) {
+              self.adjustConcurrency(true);
+            }
+
+            if (attempt <= 3) {
+              await new Promise((resolve) => setTimeout(resolve, 500));
+              return await tryFlush(flushId, batchToFlush, attempt + 1);
+            } else {
+              throw error;
+            }
+          }
+        }
+
+        const [flushError] = await tryCatch(tryFlush(nanoid(), batch));
 
+        if (flushError) {
           this.logger.error("Error flushing batch", {
-            flushId,
-            itemCount,
-            error,
-            consecutiveFailures: this.consecutiveFlushFailures,
+            error: flushError,
           });
-
-          // Re-queue the batch at the front if it fails
-          this.batchQueue.unshift(batch);
-          this.totalQueuedItems += itemCount;
-
-          // Back off on failures
-          if (this.consecutiveFlushFailures > 3) {
-            this.adjustConcurrency(true);
-          }
         }
       })
     );