prune processes if IPC is not responding (#35)

livekit · Sep 12, 2024 · 4467523 · 4467523
1 parent b112db3
commit 4467523
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 0 deletions.
diff --git a/agents/src/ipc/job_main.ts b/agents/src/ipc/job_main.ts
@@ -15,6 +15,8 @@ import { initializeLogger, log } from '../log.js';
 import { defaultInitializeProcessFunc } from '../worker.js';
 import type { IPCMessage } from './message.js';
 
+const ORPHANED_TIMEOUT = 15 * 1000;
+
 type StartArgs = {
   agentFile: string;
   // userArguments: unknown;
@@ -116,9 +118,16 @@ if (process.send) {
 
   let job: JobTask | undefined = undefined;
   const closeEvent = new EventEmitter();
+
+  const orphanedTimeout = setTimeout(() => {
+    logger.warn('process orphaned, shutting down');
+    process.exit();
+  }, ORPHANED_TIMEOUT);
+
   process.on('message', (msg: IPCMessage) => {
     switch (msg.case) {
       case 'pingRequest': {
+        orphanedTimeout.refresh();
         process.send!({
           case: 'pongResponse',
           value: { lastTimestamp: msg.value.timestamp, timestamp: Date.now() },

diff --git a/agents/src/ipc/proc_job_executor.ts b/agents/src/ipc/proc_job_executor.ts
@@ -64,6 +64,10 @@ export class ProcJobExecutor extends JobExecutor {
 
     this.#pongTimeout = setTimeout(() => {
       log().warn('job is unresponsive');
+      clearTimeout(this.#pongTimeout);
+      clearInterval(this.#pingInterval);
+      this.#proc!.kill();
+      this.#join.resolve();
     }, this.PING_TIMEOUT);
 
     const listener = (msg: IPCMessage) => {
@@ -88,6 +92,12 @@ export class ProcJobExecutor extends JobExecutor {
       }
     };
     this.#proc!.on('message', listener);
+    this.#proc!.on('error', () => {
+      log().warn('job process exited unexpectedly');
+      clearTimeout(this.#pongTimeout);
+      clearInterval(this.#pingInterval);
+      this.#join.resolve();
+    });
 
     await this.#join.await;
   }