fix(dispatcher): guard eventEmitters.emit() against a torn-down project (aicshud/WHEEL#1027)

server/app/core/dispatcher.js

@@ -697,7 +697,9 @@ class Dispatcher extends EventEmitter {
     this.runningTasks.push(component);
     this.dispatchedTasks.add(component);
     const ee = eventEmitters.get(this.projectRootDir);
-    ee.emit("taskDispatched", component);
+    if (ee) {
+      ee.emit("taskDispatched", component);
+    }
     await writeComponentJson(this.projectRootDir, component.workingDir, component, true);
     await this._addNextComponent(component);
   }
@@ -724,7 +726,9 @@ class Dispatcher extends EventEmitter {
       component.state = "running";
       await fs.writeJson(path.resolve(childDir, componentJsonFilename), component);
       const ee = eventEmitters.get(this.projectRootDir);
-      ee.emit("componentStateChanged", component);
+      if (ee) {
+        ee.emit("componentStateChanged", component);
+      }
     }
     const ancestorsType = typeof this.ancestorsType === "string" ? `${this.ancestorsType}/${component.type}` : component.type;
     const childEnv = Object.assign({}, this.env, component.env);
@@ -746,7 +750,9 @@ class Dispatcher extends EventEmitter {
       //so, it is no need to emit "componentStateChanged" here.
       if (component.type === "workflow" || component.type === "stepjob") {
         const ee = eventEmitters.get(this.projectRootDir);
-        ee.emit("componentStateChanged", component);
+        if (ee) {
+          ee.emit("componentStateChanged", component);
+        }
       }
     } finally {
       await this._addNextComponent(component);
@@ -1030,7 +1036,9 @@ class Dispatcher extends EventEmitter {
 
     const updateComponentJson = debounce(async ()=>{
       const ee = eventEmitters.get(this.projectRootDir);
-      ee.emit("componentStateChanged", component);
+      if (ee) {
+        ee.emit("componentStateChanged", component);
+      }
       return writeComponentJson(this.projectRootDir, templateRoot, component, true);
     });
     //templateRoot's descendants (e.g. the task shown when navigating into this PS component)
@@ -1218,7 +1226,9 @@ class Dispatcher extends EventEmitter {
     });
     await writeJsonWrapper(filename, filesJson);
     const ee = eventEmitters.get(this.projectRootDir);
-    ee.emit("resultFilesReady", dir);
+    if (ee) {
+      ee.emit("resultFilesReady", dir);
+    }
     await this._setComponentState(component, "finished");
   }
 

server/app/core/transferrer.js

@@ -192,6 +192,23 @@ export async function stageOut(task) {
   await _internal.setTaskState(task, taskState);
 }
 
+/**
+ * Get the distinct remotehostIDs that a project has pending deferred cleanups for, without
+ * consuming/clearing the registry (aicshud/WHEEL#1023). A caller that needs its SSH
+ * connections re-established before runDeferredCleanups() can run (e.g. onCleanProject(),
+ * invoked after stopProject() has already unconditionally disconnected everything) uses this
+ * to know which remotehosts to reconnect to first.
+ * @param {string} projectRootDir - project's root path
+ * @returns {string[]} - distinct remotehostIDs referenced by this project's registered entries
+ */
+export function getDeferredCleanupRemotehostIDs(projectRootDir) {
+  const entries = deferredCleanupRegistry.get(projectRootDir) || [];
+  const remotehostIDs = entries.map((entry)=>{
+    return entry.remotehostID;
+  });
+  return [...new Set(remotehostIDs)];
+}
+
 /**
  * Run all deferred cleanup operations for a project.
  * Deletes remote-symlink output files that were preserved during per-component cleanup,

server/app/handlers/projectController.js

@@ -22,6 +22,7 @@ import { getProjectJson, getProjectState, setProjectState, updateProjectDescript
 import { createSsh, removeSsh, askPassword } from "../core/sshManager.js";
 import { setJWTServerPassphrase, removeAllJWTServerPassphrase } from "../core/jwtServerPassphraseManager.js";
 import { runProject, cleanProject, stopProject, releaseRuntimeResources } from "../core/projectController.js";
+import { runDeferredCleanups, getDeferredCleanupRemotehostIDs } from "../core/transferrer.js";
 import { isValidOutputFilename } from "../lib/utility.js";
 import { checkWritePermissions, parentDirs, eventEmitters } from "../core/global.js";
 import { sendWorkflow, sendProjectJson, sendTaskStateList, sendResultsFileDir, sendComponentTree } from "./senders.js";
@@ -50,7 +51,12 @@ const _internal = {
   setProjectState,
   selectRunHandler,
   unlockIfFinished,
-  rootDispatchers: new Map()
+  rootDispatchers: new Map(),
+  remoteHost,
+  createSsh,
+  removeSsh,
+  runDeferredCleanups,
+  getDeferredCleanupRemotehostIDs
 };
 
 async function updateProjectState(projectRootDir, state, force) {
@@ -529,6 +535,39 @@ export async function onCleanComponent(clientID, projectRootDir, targetComponent
   ]);
 }
 
+/**
+ * reconnect SSH to every remotehost this project still has pending deferred cleanups for,
+ * run them, then disconnect again (aicshud/WHEEL#1023).
+ * stopProject() (aicshud/WHEEL#1020/#1021) unconditionally disconnects every SSH connection
+ * on stop, but deliberately leaves deferredCleanupRegistry populated - a not-yet-executed
+ * downstream task may still need the preserved remote-symlink target file if the project is
+ * resumed. If the user cleans the project instead of resuming it, that entry never gets a
+ * chance to run naturally (only runProject()'s own natural-completion path calls
+ * runDeferredCleanups()), permanently leaking the preserved remote files. Since cleanProject
+ * is only ever triggered by an explicit user action (the "clean" button), it is fine for this
+ * to need a fresh password/passphrase prompt (via createSsh()'s askPassword callback) even
+ * though the project is not running.
+ * @param {string} clientID - socket's ID, used if createSsh needs to ask for a password
+ * @param {string} projectRootDir - project's root path
+ */
+async function reconnectAndRunDeferredCleanups(clientID, projectRootDir) {
+  const remotehostIDs = _internal.getDeferredCleanupRemotehostIDs(projectRootDir);
+  if (remotehostIDs.length === 0) {
+    return;
+  }
+  for (const id of remotehostIDs) {
+    const hostinfo = _internal.remoteHost.get(id);
+    if (!hostinfo) {
+      getLogger(projectRootDir).warn(`remotehost ${id} is no longer defined; skipping its deferred cleanup`);
+      continue;
+    }
+    await _internal.createSsh(projectRootDir, hostinfo.name, hostinfo, clientID, false);
+  }
+  await _internal.runDeferredCleanups(projectRootDir);
+  _internal.removeSsh(projectRootDir);
+}
+_internal.reconnectAndRunDeferredCleanups = reconnectAndRunDeferredCleanups;
+
 async function onCleanProject(clientID, projectRootDir) {
   try {
     await askUnsavedFiles(clientID, projectRootDir);
@@ -540,6 +579,7 @@ async function onCleanProject(clientID, projectRootDir) {
   }
   try {
     await clearProjectEdits(projectRootDir);
+    await _internal.reconnectAndRunDeferredCleanups(clientID, projectRootDir);
     await Promise.all([
       cleanProject(projectRootDir),
       removeTempd(projectRootDir, "viewer"),

server/test/app/core/dispatcher.js

@@ -35,7 +35,7 @@ import { removeTransferrers } from "../../../app/core/transferManager.js";
 import { addInputFile, addOutputFile, renameOutputFile, toggleInputFileMandatory } from "../../../app/core/componentFiles.js";
 import { addLink, addFileLink } from "../../../app/core/componentLinks.js";
 import { validateComponents } from "../../../app/core/validateComponents.js";
-import { scriptName, pwdCmd, scriptHeader } from "../../testScript.js";
+import { scriptName, pwdCmd, scriptHeader, exit } from "../../testScript.js";
 const scriptPwd = `${scriptHeader}\n${pwdCmd}`;
 const wait = ()=>{
   return new Promise((resolve)=>{
@@ -1595,6 +1595,60 @@ describe("UT for Dispatcher class", function () {
     });
   });
 
+  //reproduction for aicshud/WHEEL#1027: same class of bug as #1019, found by auditing the rest
+  //of this file after confirming a live "TypeError: Cannot read properties of undefined
+  //(reading 'emit')" crash from _dispatchTask on real Fugaku hardware during aicshud/WHEEL#1024's
+  //investigation (a late-resolving task dispatch racing project teardown). _delegate and
+  //_viewerHandler have the identical unguarded pattern.
+  describe("#_dispatchTask (aicshud/WHEEL#1027)", ()=>{
+    let task;
+    beforeEach(async ()=>{
+      task = await createNewComponent(projectRootDir, projectRootDir, "task", { x: 10, y: 10 });
+      await updateComponentProperty(projectRootDir, task.ID, "script", scriptName);
+      await fs.outputFile(path.resolve(projectRootDir, task.name, scriptName), `${scriptHeader}\n${pwdCmd}\n${exit(0)}`);
+    });
+
+    it("should not throw when eventEmitters has no entry for the project (project already torn down)", async ()=>{
+      const updatedTask = await fs.readJson(path.resolve(projectRootDir, task.name, componentJsonFilename));
+      const projJson = await fs.readJson(path.resolve(projectRootDir, projectJsonFilename));
+      const DP = new Dispatcher(projectRootDir, rootWF.ID, projectRootDir, "dummy start time", projJson.componentPath, {}, "");
+      await DP._asyncInit();
+      eventEmitters.delete(projectRootDir); //simulate: project already torn down
+
+      await expect(DP._dispatchTask(updatedTask)).to.not.be.rejected;
+    });
+  });
+
+  describe("#_delegate (aicshud/WHEEL#1027)", ()=>{
+    let subWorkflow;
+    beforeEach(async ()=>{
+      subWorkflow = await createNewComponent(projectRootDir, projectRootDir, "workflow", { x: 10, y: 10 });
+    });
+
+    it("should not throw when eventEmitters has no entry for the project (project already torn down)", async ()=>{
+      const updatedSubWorkflow = await fs.readJson(path.resolve(projectRootDir, subWorkflow.name, componentJsonFilename));
+      const projJson = await fs.readJson(path.resolve(projectRootDir, projectJsonFilename));
+      const DP = new Dispatcher(projectRootDir, rootWF.ID, projectRootDir, "dummy start time", projJson.componentPath, {}, "");
+      eventEmitters.delete(projectRootDir); //simulate: project already torn down
+
+      await expect(DP._delegate(updatedSubWorkflow, false)).to.not.be.rejected;
+      expect(updatedSubWorkflow.state).to.equal("finished");
+    });
+  });
+
+  describe("#_viewerHandler (aicshud/WHEEL#1027)", ()=>{
+    it("should not throw when eventEmitters has no entry for the project (project already torn down)", async ()=>{
+      const viewer = await createNewComponent(projectRootDir, projectRootDir, "viewer", { x: 10, y: 10 });
+      const updatedViewer = await fs.readJson(path.resolve(projectRootDir, viewer.name, componentJsonFilename));
+      updatedViewer.files = [];
+      const projJson = await fs.readJson(path.resolve(projectRootDir, projectJsonFilename));
+      const DP = new Dispatcher(projectRootDir, rootWF.ID, projectRootDir, "dummy start time", projJson.componentPath, {}, "");
+      eventEmitters.delete(projectRootDir); //simulate: project already torn down
+
+      await expect(DP._viewerHandler(updatedViewer)).to.not.be.rejected;
+    });
+  });
+
   //reproduction for aicshud/WHEEL#1020: start()'s promise must record that it settled via an
   //external stop (pause()/remove(), e.g. from stopProject()) rather than natural completion,
   //so callers (runProject()) can tell not to also run their own teardown - see #1020's

server/test/app/core/transferrer.js

@@ -9,7 +9,7 @@ const expect = chai.expect;
 import chaiAsPromised from "chai-as-promised";
 chai.use(chaiAsPromised);
 import sinon from "sinon";
-import { stageIn, stageOut, runDeferredCleanups, clearDeferredCleanups, _internal } from "../../../app/core/transferrer.js";
+import { stageIn, stageOut, runDeferredCleanups, clearDeferredCleanups, getDeferredCleanupRemotehostIDs, _internal } from "../../../app/core/transferrer.js";
 describe("#stageIn", ()=>{
   let setTaskStateStub;
   let getSshHostinfoStub;
@@ -454,6 +454,44 @@ describe("#runDeferredCleanups", ()=>{
   });
 });
 
+describe("[reproduction] issue aicshud/WHEEL#1023 - getDeferredCleanupRemotehostIDs must expose which remotehosts need reconnecting", ()=>{
+  afterEach(()=>{
+    sinon.restore();
+  });
+
+  it("should return an empty array when no deferred cleanups are registered for the project", ()=>{
+    expect(getDeferredCleanupRemotehostIDs("/proj/not-registered")).to.deep.equal([]);
+  });
+
+  it("should return the distinct remotehostIDs of all registered entries, without mutating the registry", ()=>{
+    _internal.addDeferredCleanup("/proj/multi", {
+      remoteWorkingDir: "/remote/task1",
+      remotehostID: "hostX",
+      symlinkTargetNames: ["file1.dat"]
+    });
+    _internal.addDeferredCleanup("/proj/multi", {
+      remoteWorkingDir: "/remote/task2",
+      remotehostID: "hostY",
+      symlinkTargetNames: ["file2.dat"]
+    });
+    _internal.addDeferredCleanup("/proj/multi", {
+      remoteWorkingDir: "/remote/task3",
+      remotehostID: "hostX",
+      symlinkTargetNames: ["file3.dat"]
+    });
+
+    expect(getDeferredCleanupRemotehostIDs("/proj/multi").sort()).to.deep.equal(["hostX", "hostY"]);
+
+    //must be a non-destructive peek - runDeferredCleanups must still see all 3 entries
+    const sshExecStub = sinon.stub().resolves(0);
+    sinon.stub(_internal, "getSsh").returns({ exec: sshExecStub });
+    sinon.stub(_internal, "getLogger").returns({ debug: sinon.stub(), warn: sinon.stub() });
+    return runDeferredCleanups("/proj/multi").then(()=>{
+      expect(sshExecStub.callCount).to.equal(6); //1 file + 1 dir for each of the 3 entries
+    });
+  });
+});
+
 describe("#clearDeferredCleanups", ()=>{
   afterEach(()=>{
     sinon.restore();

server/test/app/handlers/projectController.js

@@ -500,4 +500,44 @@ describe("project Controller handler UT", function () {
       sinon.assert.calledOnceWithExactly(removeTransferrersStub, projectRootDir);
     });
   });
+
+  describe("[reproduction] issue aicshud/WHEEL#1023 - onCleanProject must reconnect SSH before running deferred cleanups", ()=>{
+    it("should reconnect ssh for every remotehost with a pending deferred cleanup, run them, then disconnect again", async ()=>{
+      //stopProject() (aicshud/WHEEL#1020/#1021) leaves the deferredCleanupRegistry populated
+      //but has already unconditionally disconnected every SSH connection - so by the time the
+      //user clicks "clean", runDeferredCleanups() can not just be called directly (it would
+      //throw "ssh instance is not registerd for the project"). onCleanProject must reconnect
+      //first, using the same remotehost.json entry runDispatcher() itself would have used.
+      const clientID = "test-client-id";
+      const hostinfoA = { id: "hostA", name: "hostA-name" };
+      sinon.stub(_internal, "getDeferredCleanupRemotehostIDs").returns(["hostA"]);
+      sinon.stub(_internal, "remoteHost").value({
+        get: sinon.stub().withArgs("hostA")
+          .returns(hostinfoA)
+      });
+      const createSshStub = sinon.stub(_internal, "createSsh").resolves({});
+      const runDeferredCleanupsStub = sinon.stub(_internal, "runDeferredCleanups").resolves();
+      const removeSshStub = sinon.stub(_internal, "removeSsh");
+
+      await _internal.onCleanProject(clientID, projectRootDir);
+
+      sinon.assert.calledOnceWithExactly(createSshStub, projectRootDir, "hostA-name", hostinfoA, clientID, false);
+      sinon.assert.calledOnceWithExactly(runDeferredCleanupsStub, projectRootDir);
+      sinon.assert.calledOnceWithExactly(removeSshStub, projectRootDir);
+      sinon.assert.callOrder(createSshStub, runDeferredCleanupsStub, removeSshStub);
+    });
+
+    it("should not attempt any SSH reconnection when there are no pending deferred cleanups", async ()=>{
+      sinon.stub(_internal, "getDeferredCleanupRemotehostIDs").returns([]);
+      const createSshStub = sinon.stub(_internal, "createSsh");
+      const runDeferredCleanupsStub = sinon.stub(_internal, "runDeferredCleanups");
+      const removeSshStub = sinon.stub(_internal, "removeSsh");
+
+      await _internal.onCleanProject("test-client-id", projectRootDir);
+
+      expect(createSshStub.called).to.be.false;
+      expect(runDeferredCleanupsStub.called).to.be.false;
+      expect(removeSshStub.called).to.be.false;
+    });
+  });
 });