use the new interval service everywhere

Author: nicktrn

apps/supervisor/src/services/podCleaner.ts

@@ -1,7 +1,7 @@
 import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 import { K8sApi } from "../clients/kubernetes.js";
 import { createK8sApi } from "../clients/kubernetes.js";
-import { HeartbeatService } from "@trigger.dev/core/v3";
+import { IntervalService } from "@trigger.dev/core/v3";
 import { Counter, Gauge, Registry } from "prom-client";
 import { register } from "../metrics.js";
 
@@ -19,7 +19,7 @@ export class PodCleaner {
   private readonly namespace: string;
 
   private readonly batchSize: number;
-  private readonly deletionHeartbeat: HeartbeatService;
+  private readonly deletionInterval: IntervalService;
 
   // Metrics
   private readonly register: Registry;
@@ -32,10 +32,10 @@ export class PodCleaner {
     this.namespace = opts.namespace;
     this.batchSize = opts.batchSize ?? 500;
 
-    this.deletionHeartbeat = new HeartbeatService({
+    this.deletionInterval = new IntervalService({
       intervalMs: opts.intervalMs ?? 10000,
       leadingEdge: true,
-      heartbeat: this.deleteCompletedPods.bind(this),
+      onInterval: this.deleteCompletedPods.bind(this),
     });
 
     // Initialize metrics
@@ -57,11 +57,11 @@ export class PodCleaner {
   }
 
   async start() {
-    this.deletionHeartbeat.start();
+    this.deletionInterval.start();
   }
 
   async stop() {
-    this.deletionHeartbeat.stop();
+    this.deletionInterval.stop();
   }
 
   private async deleteCompletedPods() {

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

@@ -1,6 +1,6 @@
 import {
   clientWebsocketMessages,
-  HeartbeatService,
+  IntervalService,
   serverWebsocketMessages,
 } from "@trigger.dev/core/v3";
 import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
@@ -19,7 +19,7 @@ export class AuthenticatedSocketConnection {
   private _sender: ZodMessageSender<typeof serverWebsocketMessages>;
   private _consumer: DevQueueConsumer;
   private _messageHandler: ZodMessageHandler<typeof clientWebsocketMessages>;
-  private _pingService: HeartbeatService;
+  private _pingService: IntervalService;
 
   constructor(
     public ws: WebSocket,
@@ -75,8 +75,8 @@ export class AuthenticatedSocketConnection {
       // });
     });
 
-    this._pingService = new HeartbeatService({
-      heartbeat: async () => {
+    this._pingService = new IntervalService({
+      onInterval: async () => {
         if (ws.readyState !== WebSocket.OPEN) {
           logger.debug("[AuthenticatedSocketConnection] Websocket not open, skipping ping");
           return;

packages/cli-v3/src/entryPoints/dev-run-controller.ts

@@ -1,7 +1,7 @@
 import {
   CompleteRunAttemptResult,
   DequeuedMessage,
-  HeartbeatService,
+  IntervalService,
   LogLevel,
   RunExecutionData,
   TaskRunExecution,
@@ -44,9 +44,9 @@ export class DevRunController {
   private taskRunProcess?: TaskRunProcess;
   private readonly worker: BackgroundWorker;
   private readonly httpClient: CliApiClient;
-  private readonly runHeartbeat: HeartbeatService;
+  private readonly runHeartbeat: IntervalService;
   private readonly heartbeatIntervalSeconds: number;
-  private readonly snapshotPoller: HeartbeatService;
+  private readonly snapshotPoller: IntervalService;
   private readonly snapshotPollIntervalSeconds: number;
 
   private state:
@@ -78,8 +78,8 @@ export class DevRunController {
 
     this.httpClient = opts.httpClient;
 
-    this.snapshotPoller = new HeartbeatService({
-      heartbeat: async () => {
+    this.snapshotPoller = new IntervalService({
+      onInterval: async () => {
         if (!this.runFriendlyId) {
           logger.debug("[DevRunController] Skipping snapshot poll, no run ID");
           return;
@@ -121,8 +121,8 @@ export class DevRunController {
       },
     });
 
-    this.runHeartbeat = new HeartbeatService({
-      heartbeat: async () => {
+    this.runHeartbeat = new IntervalService({
+      onInterval: async () => {
         if (!this.runFriendlyId || !this.snapshotFriendlyId) {
           logger.debug("[DevRunController] Skipping heartbeat, no run ID or snapshot ID");
           return;

packages/cli-v3/src/entryPoints/managed/heartbeat.ts

@@ -1,4 +1,4 @@
-import { HeartbeatService, RunExecutionData } from "@trigger.dev/core/v3";
+import { IntervalService } from "@trigger.dev/core/v3";
 import { WorkloadHttpClient } from "@trigger.dev/core/v3/runEngineWorker";
 import { RunLogger } from "./logger.js";
 
@@ -17,7 +17,7 @@ export class RunExecutionHeartbeat {
   private readonly httpClient: WorkloadHttpClient;
   private readonly logger: RunLogger;
   private readonly heartbeatIntervalMs: number;
-  private readonly heartbeat: HeartbeatService;
+  private readonly heartbeat: IntervalService;
 
   constructor(opts: RunExecutionHeartbeatOptions) {
     this.runFriendlyId = opts.runFriendlyId;
@@ -36,8 +36,8 @@ export class RunExecutionHeartbeat {
       },
     });
 
-    this.heartbeat = new HeartbeatService({
-      heartbeat: async () => {
+    this.heartbeat = new IntervalService({
+      onInterval: async () => {
         this.logger.sendDebugLog({
           runId: this.runFriendlyId,
           message: "heartbeat: started",
@@ -90,108 +90,3 @@ export class RunExecutionHeartbeat {
     this.heartbeat.stop();
   }
 }
-
-type RunExecutionSnapshotPollerOptions = {
-  runFriendlyId: string;
-  snapshotFriendlyId: string;
-  httpClient: WorkloadHttpClient;
-  logger: RunLogger;
-  snapshotPollIntervalSeconds: number;
-  handleSnapshotChange: (execution: RunExecutionData) => Promise<void>;
-};
-
-class RunExecutionSnapshotPoller {
-  private readonly logger: RunLogger;
-  private readonly poller: HeartbeatService;
-  private readonly httpClient: WorkloadHttpClient;
-
-  private readonly runFriendlyId: string;
-  private readonly snapshotFriendlyId: string;
-
-  private readonly handleSnapshotChange: (execution: RunExecutionData) => Promise<void>;
-
-  constructor(opts: RunExecutionSnapshotPollerOptions) {
-    this.logger = opts.logger;
-    this.httpClient = opts.httpClient;
-
-    this.runFriendlyId = opts.runFriendlyId;
-    this.snapshotFriendlyId = opts.snapshotFriendlyId;
-
-    this.handleSnapshotChange = opts.handleSnapshotChange;
-
-    this.poller = new HeartbeatService({
-      heartbeat: async () => {
-        if (!this.runFriendlyId) {
-          this.logger.sendDebugLog({
-            runId: this.runFriendlyId,
-            message: "Skipping snapshot poll, no run ID",
-          });
-          return;
-        }
-
-        this.logger.sendDebugLog({
-          runId: this.runFriendlyId,
-          message: "Polling for latest snapshot",
-        });
-
-        this.logger.sendDebugLog({
-          runId: this.runFriendlyId,
-          message: `snapshot poll: started`,
-          properties: {
-            snapshotId: this.snapshotFriendlyId,
-          },
-        });
-
-        const response = await this.httpClient.getRunExecutionData(this.runFriendlyId);
-
-        if (!response.success) {
-          this.logger.sendDebugLog({
-            runId: this.runFriendlyId,
-            message: "Snapshot poll failed",
-            properties: {
-              error: response.error,
-            },
-          });
-
-          this.logger.sendDebugLog({
-            runId: this.runFriendlyId,
-            message: `snapshot poll: failed`,
-            properties: {
-              snapshotId: this.snapshotFriendlyId,
-              error: response.error,
-            },
-          });
-
-          return;
-        }
-
-        await this.handleSnapshotChange(response.data.execution);
-      },
-      intervalMs: opts.snapshotPollIntervalSeconds * 1000,
-      leadingEdge: false,
-      onError: async (error) => {
-        this.logger.sendDebugLog({
-          runId: this.runFriendlyId,
-          message: "Failed to poll for snapshot",
-          properties: { error: error instanceof Error ? error.message : String(error) },
-        });
-      },
-    });
-  }
-
-  resetCurrentInterval() {
-    this.poller.resetCurrentInterval();
-  }
-
-  updateInterval(intervalMs: number) {
-    this.poller.updateInterval(intervalMs);
-  }
-
-  start() {
-    this.poller.start();
-  }
-
-  stop() {
-    this.poller.stop();
-  }
-}

packages/cli-v3/src/entryPoints/managed/poller.ts

@@ -1,6 +1,6 @@
 import { WorkloadHttpClient } from "@trigger.dev/core/v3/runEngineWorker";
 import { RunLogger } from "./logger.js";
-import { HeartbeatService, RunExecutionData } from "@trigger.dev/core/v3";
+import { IntervalService, RunExecutionData } from "@trigger.dev/core/v3";
 
 export type RunExecutionSnapshotPollerOptions = {
   runFriendlyId: string;
@@ -19,7 +19,7 @@ export class RunExecutionSnapshotPoller {
   private readonly logger: RunLogger;
   private readonly snapshotPollIntervalMs: number;
   private readonly handleSnapshotChange: (runData: RunExecutionData) => Promise<void>;
-  private readonly poller: HeartbeatService;
+  private readonly poller: IntervalService;
 
   constructor(opts: RunExecutionSnapshotPollerOptions) {
     this.runFriendlyId = opts.runFriendlyId;
@@ -39,8 +39,8 @@ export class RunExecutionSnapshotPoller {
       },
     });
 
-    this.poller = new HeartbeatService({
-      heartbeat: async () => {
+    this.poller = new IntervalService({
+      onInterval: async () => {
         if (!this.runFriendlyId) {
           this.logger.sendDebugLog({
             runId: this.runFriendlyId,