Some tweaks

Author: ericallam

apps/webapp/app/env.server.ts

@@ -255,9 +255,9 @@ const EnvironmentSchema = z.object({
   EVENTS_MIN_CONCURRENCY: z.coerce.number().int().default(1),
   EVENTS_MAX_CONCURRENCY: z.coerce.number().int().default(10),
   EVENTS_MAX_BATCH_SIZE: z.coerce.number().int().default(500),
-  EVENTS_MEMORY_PRESSURE_THRESHOLD: z.coerce.number().int().default(2000),
-  EVENTS_LOAD_SHEDDING_THRESHOLD: z.coerce.number().int().default(5000),
-  EVENTS_LOAD_SHEDDING_ENABLED: z.coerce.boolean().default(true),
+  EVENTS_MEMORY_PRESSURE_THRESHOLD: z.coerce.number().int().default(5000),
+  EVENTS_LOAD_SHEDDING_THRESHOLD: z.coerce.number().int().default(100000),
+  EVENTS_LOAD_SHEDDING_ENABLED: z.string().default("1"),
   SHARED_QUEUE_CONSUMER_POOL_SIZE: z.coerce.number().int().default(10),
   SHARED_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(100),
   SHARED_QUEUE_CONSUMER_NEXT_TICK_INTERVAL_MS: z.coerce.number().int().default(100),

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

@@ -1,7 +1,6 @@
+import { Logger } from "@trigger.dev/core/logger";
 import { nanoid } from "nanoid";
 import pLimit from "p-limit";
-import { logger } from "~/services/logger.server";
-import { TaskEventKind } from "@trigger.dev/database";
 
 export type DynamicFlushSchedulerConfig<T> = {
   batchSize: number;
@@ -49,6 +48,8 @@ export class DynamicFlushScheduler<T> {
   private readonly isDroppableEvent?: (item: T) => boolean;
   private isLoadShedding: boolean = false;
 
+  private readonly logger: Logger = new Logger("EventRepo.DynamicFlushScheduler", "debug");
+
   constructor(config: DynamicFlushSchedulerConfig<T>) {
     this.batchQueue = [];
     this.currentBatch = [];
@@ -98,16 +99,17 @@ export class DynamicFlushScheduler<T> {
 
         if (!this.isLoadShedding) {
           this.isLoadShedding = true;
-          logger.warn("Load shedding activated", {
-            totalQueuedItems: this.totalQueuedItems,
-            threshold: this.loadSheddingThreshold,
-            droppedCount: dropped.length,
-          });
         }
+
+        this.logger.warn("Load shedding", {
+          totalQueuedItems: this.totalQueuedItems,
+          threshold: this.loadSheddingThreshold,
+          droppedCount: dropped.length,
+        });
       }
     } else if (this.isLoadShedding && this.totalQueuedItems < this.loadSheddingThreshold * 0.8) {
       this.isLoadShedding = false;
-      logger.info("Load shedding deactivated", {
+      this.logger.info("Load shedding deactivated", {
         totalQueuedItems: this.totalQueuedItems,
         threshold: this.loadSheddingThreshold,
         totalDropped: this.metrics.droppedEvents,
@@ -183,7 +185,7 @@ export class DynamicFlushScheduler<T> {
           this.metrics.flushedBatches++;
           this.metrics.totalItemsFlushed += itemCount;
 
-          logger.debug("Batch flushed successfully", {
+          this.logger.debug("Batch flushed successfully", {
             flushId,
             itemCount,
             duration,
@@ -195,7 +197,7 @@ export class DynamicFlushScheduler<T> {
           this.consecutiveFlushFailures++;
           this.metrics.failedBatches++;
 
-          logger.error("Error flushing batch", {
+          this.logger.error("Error flushing batch", {
             flushId,
             itemCount,
             error,
@@ -223,13 +225,21 @@ export class DynamicFlushScheduler<T> {
     });
   }
 
+  private lastConcurrencyAdjustment: number = Date.now();
+  
   private adjustConcurrency(backOff: boolean = false): void {
     const currentConcurrency = this.limiter.concurrency;
     let newConcurrency = currentConcurrency;
-    
+
     // Calculate pressure metrics - moved outside the if/else block
     const queuePressure = this.totalQueuedItems / this.memoryPressureThreshold;
     const timeSinceLastFlush = Date.now() - this.lastFlushTime;
+    const timeSinceLastAdjustment = Date.now() - this.lastConcurrencyAdjustment;
+
+    // Don't adjust too frequently (except for backoff)
+    if (!backOff && timeSinceLastAdjustment < 1000) {
+      return;
+    }
 
     if (backOff) {
       // Reduce concurrency on failures
@@ -258,12 +268,13 @@ export class DynamicFlushScheduler<T> {
     if (newConcurrency !== currentConcurrency) {
       this.limiter = pLimit(newConcurrency);
 
-      logger.info("Adjusted flush concurrency", {
+      this.logger.info("Adjusted flush concurrency", {
         previousConcurrency: currentConcurrency,
         newConcurrency,
         queuePressure,
         totalQueuedItems: this.totalQueuedItems,
         currentBatchSize: this.currentBatchSize,
+        memoryPressureThreshold: this.memoryPressureThreshold,
       });
     }
   }
@@ -276,7 +287,7 @@ export class DynamicFlushScheduler<T> {
         droppedByKind[kind] = count;
       });
 
-      logger.info("DynamicFlushScheduler metrics", {
+      this.logger.info("DynamicFlushScheduler metrics", {
         totalQueuedItems: this.totalQueuedItems,
         batchQueueLength: this.batchQueue.length,
         currentBatchLength: this.currentBatch.length,

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

@@ -1349,7 +1349,7 @@ function initializeEventRepo() {
     maxBatchSize: env.EVENTS_MAX_BATCH_SIZE,
     memoryPressureThreshold: env.EVENTS_MEMORY_PRESSURE_THRESHOLD,
     loadSheddingThreshold: env.EVENTS_LOAD_SHEDDING_THRESHOLD,
-    loadSheddingEnabled: env.EVENTS_LOAD_SHEDDING_ENABLED,
+    loadSheddingEnabled: env.EVENTS_LOAD_SHEDDING_ENABLED === "1",
     redis: {
       port: env.PUBSUB_REDIS_PORT,
       host: env.PUBSUB_REDIS_HOST,

references/hello-world/src/trigger/example.ts

@@ -373,3 +373,68 @@ export const largeAttributesTask = task({
     });
   },
 });
+
+export const lotsOfLogsParentTask = task({
+  id: "lots-of-logs-parent",
+  run: async (payload: { count: number }, { ctx }) => {
+    logger.info("Hello, world from the lots of logs parent task", { count: payload.count });
+    await lotsOfLogsTask.batchTriggerAndWait(
+      Array.from({ length: 20 }, (_, i) => ({
+        payload: { count: payload.count },
+      }))
+    );
+  },
+});
+
+export const lotsOfLogsTask = task({
+  id: "lots-of-logs",
+  run: async (payload: { count: number }, { ctx }) => {
+    logger.info("Hello, world from the lots of logs task", { count: payload.count });
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+
+    await setTimeout(1000);
+
+    for (let i = 0; i < payload.count; i++) {
+      logger.info("Hello, world from the lots of logs task", { count: i });
+    }
+  },
+});