rename HeartbeatService to IntervalService

Author: nicktrn

packages/core/src/v3/runEngineWorker/supervisor/session.ts

@@ -8,7 +8,7 @@ import { VERSION } from "../../../version.js";
 import { io, Socket } from "socket.io-client";
 import { WorkerClientToServerEvents, WorkerServerToClientEvents } from "../types.js";
 import { getDefaultWorkerHeaders } from "./util.js";
-import { HeartbeatService } from "../../utils/interval.js";
+import { IntervalService } from "../../utils/interval.js";
 
 type SupervisorSessionOptions = SupervisorClientCommonOptions & {
   queueConsumerEnabled?: boolean;
@@ -29,7 +29,7 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
   private readonly queueConsumerEnabled: boolean;
   private readonly queueConsumer: RunQueueConsumer;
 
-  private readonly heartbeatService: HeartbeatService;
+  private readonly heartbeat: IntervalService;
   private readonly heartbeatIntervalSeconds: number;
 
   constructor(private opts: SupervisorSessionOptions) {
@@ -50,8 +50,8 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
 
     // TODO: This should be dynamic and set by (or at least overridden by) the platform
     this.heartbeatIntervalSeconds = opts.heartbeatIntervalSeconds || 30;
-    this.heartbeatService = new HeartbeatService({
-      heartbeat: async () => {
+    this.heartbeat = new IntervalService({
+      onInterval: async () => {
         console.debug("[SupervisorSession] Sending heartbeat");
 
         const body = this.getHeartbeatBody();
@@ -182,7 +182,7 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
     if (this.queueConsumerEnabled) {
       console.log("[SupervisorSession] Queue consumer enabled");
       this.queueConsumer.start();
-      this.heartbeatService.start();
+      this.heartbeat.start();
     } else {
       console.warn("[SupervisorSession] Queue consumer disabled");
     }
@@ -196,7 +196,7 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
   }
 
   async stop() {
-    this.heartbeatService.stop();
+    this.heartbeat.stop();
     this.runNotificationsSocket?.disconnect();
   }
 

packages/core/src/v3/utils/interval.ts

@@ -1,93 +1,95 @@
-type HeartbeatServiceOptions = {
-  heartbeat: () => Promise<void>;
+type IntervalServiceOptions = {
+  onInterval: () => Promise<void>;
+  onError?: (error: unknown) => Promise<void>;
   intervalMs?: number;
   leadingEdge?: boolean;
-  onError?: (error: unknown) => Promise<void>;
 };
 
-export class HeartbeatService {
-  private _heartbeat: () => Promise<void>;
+export class IntervalService {
+  private _onInterval: () => Promise<void>;
+  private _onError?: (error: unknown) => Promise<void>;
+
   private _intervalMs: number;
-  private _nextHeartbeat: NodeJS.Timeout | undefined;
+  private _nextInterval: NodeJS.Timeout | undefined;
   private _leadingEdge: boolean;
-  private _isHeartbeating: boolean;
-  private _onError?: (error: unknown) => Promise<void>;
+  private _isEnabled: boolean;
+
+  constructor(opts: IntervalServiceOptions) {
+    this._onInterval = opts.onInterval;
+    this._onError = opts.onError;
 
-  constructor(opts: HeartbeatServiceOptions) {
-    this._heartbeat = opts.heartbeat;
     this._intervalMs = opts.intervalMs ?? 45_000;
-    this._nextHeartbeat = undefined;
+    this._nextInterval = undefined;
     this._leadingEdge = opts.leadingEdge ?? false;
-    this._isHeartbeating = false;
-    this._onError = opts.onError;
+    this._isEnabled = false;
   }
 
   start() {
-    if (this._isHeartbeating) {
+    if (this._isEnabled) {
       return;
     }
 
-    this._isHeartbeating = true;
+    this._isEnabled = true;
 
     if (this._leadingEdge) {
-      this.#doHeartbeat();
+      this.#doInterval();
     } else {
-      this.#scheduleNextHeartbeat();
+      this.#scheduleNextInterval();
     }
   }
 
   stop() {
-    if (!this._isHeartbeating) {
+    if (!this._isEnabled) {
       return;
     }
 
-    this._isHeartbeating = false;
-    this.#clearNextHeartbeat();
+    this._isEnabled = false;
+    this.#clearNextInterval();
   }
 
   resetCurrentInterval() {
-    if (!this._isHeartbeating) {
+    if (!this._isEnabled) {
       return;
     }
 
-    this.#clearNextHeartbeat();
-    this.#scheduleNextHeartbeat();
+    this.#clearNextInterval();
+    this.#scheduleNextInterval();
   }
 
   updateInterval(intervalMs: number) {
     this._intervalMs = intervalMs;
     this.resetCurrentInterval();
   }
 
-  #doHeartbeat = async () => {
-    this.#clearNextHeartbeat();
+  #doInterval = async () => {
+    this.#clearNextInterval();
 
-    if (!this._isHeartbeating) {
+    if (!this._isEnabled) {
       return;
     }
 
     try {
-      await this._heartbeat();
+      await this._onInterval();
     } catch (error) {
       if (this._onError) {
         try {
           await this._onError(error);
         } catch (error) {
-          console.error("Error handling heartbeat error", error);
+          console.error("Error during interval error handler", error);
         }
       }
     }
 
-    this.#scheduleNextHeartbeat();
+    this.#scheduleNextInterval();
   };
 
-  #clearNextHeartbeat() {
-    if (this._nextHeartbeat) {
-      clearTimeout(this._nextHeartbeat);
+  #clearNextInterval() {
+    if (this._nextInterval) {
+      clearTimeout(this._nextInterval);
     }
   }
 
-  #scheduleNextHeartbeat() {
-    this._nextHeartbeat = setTimeout(this.#doHeartbeat, this._intervalMs);
+  #scheduleNextInterval() {
+    this._nextInterval = setTimeout(this.#doInterval, this._intervalMs);
   }
 }