core: terminalservice fork across cluster (#1721)

* core: terminalservice fork across cluster * exit cluster fork on completion * force terminate on errors * make isClusterFork internal to prevent callers from killing core plugin * implement forkInterface and share forks * use correct native id * use correct native id in primary device construction
koush · Feb 2, 2025 · 590f955 · 590f955
1 parent 7df4bf2
commit 590f955
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 6 deletions.
diff --git a/plugins/core/src/main.ts b/plugins/core/src/main.ts
@@ -13,7 +13,7 @@ import { MediaCore } from './media-core';
 import { checkLegacyLxc, checkLxc } from './platform/lxc';
 import { ConsoleServiceNativeId, PluginSocketService, ReplServiceNativeId } from './plugin-socket-service';
 import { ScriptCore, ScriptCoreNativeId, newScript } from './script-core';
-import { TerminalService, TerminalServiceNativeId } from './terminal-service';
+import { TerminalService, TerminalServiceNativeId, newTerminalService } from './terminal-service';
 import { UsersCore, UsersNativeId } from './user';
 import { ClusterCore, ClusterCoreNativeId } from './cluster';
 
@@ -140,7 +140,7 @@ class ScryptedCore extends ScryptedDeviceBase implements HttpRequestHandler, Dev
                 {
                     name: 'Terminal Service',
                     nativeId: TerminalServiceNativeId,
-                    interfaces: [ScryptedInterface.StreamService, ScryptedInterface.TTY],
+                    interfaces: [ScryptedInterface.StreamService, ScryptedInterface.TTY, ScryptedInterface.ClusterForkInterface],
                     type: ScryptedDeviceType.Builtin,
                 },
             );
@@ -242,7 +242,7 @@ class ScryptedCore extends ScryptedDeviceBase implements HttpRequestHandler, Dev
         if (nativeId === UsersNativeId)
             return this.users ||= new UsersCore();
         if (nativeId === TerminalServiceNativeId)
-            return this.terminalService ||= new TerminalService();
+            return this.terminalService ||= new TerminalService(TerminalServiceNativeId, false);
         if (nativeId === ReplServiceNativeId)
             return this.replService ||= new PluginSocketService(ReplServiceNativeId, 'repl');
         if (nativeId === ConsoleServiceNativeId)
@@ -331,5 +331,6 @@ export async function fork() {
     return {
         tsCompile,
         newScript,
+        newTerminalService,
     }
 }
diff --git a/plugins/core/src/terminal-service.ts b/plugins/core/src/terminal-service.ts
@@ -1,4 +1,4 @@
-import sdk, { ScryptedDeviceBase, ScryptedInterface, ScryptedNativeId, StreamService, TTYSettings } from "@scrypted/sdk";
+import sdk, { ClusterForkInterface, ClusterForkInterfaceOptions, ScryptedDeviceBase, ScryptedInterface, ScryptedNativeId, StreamService, TTYSettings } from "@scrypted/sdk";
 import type { IPty, spawn as ptySpawn } from 'node-pty';
 import { createAsyncQueue } from '@scrypted/common/src/async-queue'
 import { ChildProcess, spawn as childSpawn } from "child_process";
@@ -111,8 +111,11 @@ class NoninteractiveTerminal {
 }
 
 
-export class TerminalService extends ScryptedDeviceBase implements StreamService<Buffer | string, Buffer> {
-    constructor(nativeId?: ScryptedNativeId) {
+export class TerminalService extends ScryptedDeviceBase implements StreamService<Buffer | string, Buffer>, ClusterForkInterface {
+    private forks: { [clusterWorkerId: string]: TerminalService } = {};
+    private forkClients: 0;
+
+    constructor(nativeId?: ScryptedNativeId, private isFork: boolean = false) {
         super(nativeId);
     }
 
@@ -134,6 +137,42 @@ export class TerminalService extends ScryptedDeviceBase implements StreamService
         return extraPaths;
     }
 
+    async forkInterface<StreamService>(forkInterface: ScryptedInterface, options?: ClusterForkInterfaceOptions): Promise<StreamService> {
+        if (forkInterface !== ScryptedInterface.StreamService) {
+            throw new Error('can only fork StreamService');
+        }
+
+        if (!options?.clusterWorkerId) {
+            throw new Error('clusterWorkerId required');
+        }
+
+        if (this.isFork) {
+            throw new Error('cannot fork a fork');
+        }
+
+        const clusterWorkerId = options.clusterWorkerId;
+        if (this.forks[clusterWorkerId]) {
+            return this.forks[clusterWorkerId] as StreamService;
+        }
+
+        const fork = sdk.fork<{
+            newTerminalService: typeof newTerminalService,
+        }>({ clusterWorkerId });
+        try {
+            const result = await fork.result;
+            const terminalService = await result.newTerminalService();
+            this.forks[clusterWorkerId] = terminalService;
+            fork.worker.on('exit', () => {
+                delete this.forks[clusterWorkerId];
+            });
+            return terminalService as StreamService;
+        }
+        catch (e) {
+            fork.worker.terminate();
+            throw e;
+        }
+    }
+
     /*
      * The input to this stream can send buffers for normal terminal data and strings
      * for control messages. Control messages are JSON-formatted.
@@ -149,6 +188,19 @@ export class TerminalService extends ScryptedDeviceBase implements StreamService
         const queue = createAsyncQueue<Buffer>();
         const extraPaths = await this.getExtraPaths();
 
+        if (this.isFork) {
+            this.forkClients++;
+        }
+
+        queue.endPromise.then(() => {
+            if (this.isFork) {
+                this.forkClients--;
+                if (this.forkClients === 0) {
+                    process.exit();
+                }
+            }
+        });
+
         function registerChildListeners() {
             cp.onExit(() => queue.end());
 
@@ -232,4 +284,8 @@ export class TerminalService extends ScryptedDeviceBase implements StreamService
 
         return generator();
     }
+}
+
+export async function newTerminalService(): Promise<TerminalService> {
+    return new TerminalService(TerminalServiceNativeId, true);
 }