perf(webapp,run-engine): skip dev worker notifications for non-dev runs

Run state notifications were broadcast to both the deployed worker and
dev worker socket.io namespaces for every run, regardless of
environment. The dev worker namespace only accepts connections from
development environments, so that second broadcast was discarded for
every non-development run.

Notifications now reach the dev worker namespace only for runs in a
development environment. This removes the redundant broadcast and
roughly halves the Redis pub/sub traffic generated by the run
notification path.

Mono-RevId: 84cb5aead09067daeb979d793e3974c2a1da57c8

Author: matt-aitken

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

@@ -690,9 +690,11 @@ export function registerRunEngineEventBusHandlers() {
         .emit("run:notify", { version: "1", run: { friendlyId: runFriendlyId } });
 
       //send the notification to connected dev workers
-      socketIo.devWorkerNamespace
-        .to(room)
-        .emit("run:notify", { version: "1", run: { friendlyId: runFriendlyId } });
+      if (snapshot.environmentType === "DEVELOPMENT") {
+        socketIo.devWorkerNamespace
+          .to(room)
+          .emit("run:notify", { version: "1", run: { friendlyId: runFriendlyId } });
+      }
 
       if (!env.RUN_ENGINE_DEBUG_WORKER_NOTIFICATIONS) {
         return;

internal-packages/run-engine/src/engine/eventBus.ts

@@ -1,5 +1,9 @@
 import type { FlushedRunMetadata, TaskRunError } from "@trigger.dev/core/v3";
-import type { TaskRunExecutionStatus, TaskRunStatus } from "@trigger.dev/database";
+import type {
+  RuntimeEnvironmentType,
+  TaskRunExecutionStatus,
+  TaskRunStatus,
+} from "@trigger.dev/database";
 import type { EventEmitter } from "events";
 import type { AuthenticatedEnvironment } from "../shared/index.js";
 
@@ -324,6 +328,7 @@ export type EventBusEvents = {
       snapshot: {
         id: string;
         executionStatus: TaskRunExecutionStatus;
+        environmentType: RuntimeEnvironmentType;
       };
     },
   ];
@@ -383,6 +388,7 @@ export async function sendNotificationToWorker({
   snapshot: {
     id: string;
     executionStatus: TaskRunExecutionStatus;
+    environmentType: RuntimeEnvironmentType;
   };
   eventBus: EventBus;
 }) {
@@ -394,6 +400,7 @@ export async function sendNotificationToWorker({
     snapshot: {
       id: snapshot.id,
       executionStatus: snapshot.executionStatus,
+      environmentType: snapshot.environmentType,
     },
   });
 }

internal-packages/run-engine/src/engine/tests/waitpoints.test.ts

@@ -393,6 +393,7 @@ describe("RunEngine Waitpoints", () => {
         assertNonNullable(event);
         const notificationEvent = event as EventBusEventArgs<"workerNotification">[0];
         expect(notificationEvent.run.id).toBe(run.id);
+        expect(notificationEvent.snapshot.environmentType).toBe("PRODUCTION");
 
         const executionData2 = await engine.getRunExecutionData({ runId: run.id });
         expect(executionData2?.snapshot.executionStatus).toBe("EXECUTING");