fix: SSH-race and remote-resource-leak fixes around stopProject/runProject (aicshud/WHEEL#1020, #1021, #1022)

server/app/core/dispatcher.js

@@ -518,6 +518,16 @@ class Dispatcher extends EventEmitter {
         //start()'s promise with the current outcome instead of leaving it pending forever -
         //otherwise the caller (runProject()) hangs indefinitely and never reaches its own
         //state update/cleanup, which in turn leaves the project stuck instead of concluding.
+        //
+        //record that this was an external stop (aicshud/WHEEL#1020): pause()/remove() (which
+        //triggered this) is still busy recursively canceling nested job tasks - using the
+        //project's SSH connections - well after this resolves, since pause() emits "stop"
+        //synchronously before awaiting that cancellation. The caller (runProject()) must not
+        //also run its own SSH/executer/transferrer teardown once this resolves - that races
+        //the still-in-flight cancellation and can disconnect SSH out from under it. Whoever
+        //called pause()/remove() (stopProject()) already owns that teardown once its own
+        //await on remove() finishes.
+        this.stoppedExternally = true;
         resolve(this._getState());
       };
       this.once("done", this.onDone);

server/app/core/projectController.js

@@ -9,7 +9,7 @@ import { gitResetHEAD, gitClean } from "../core/gitOperator2.js";
 import { removeSsh } from "./sshManager.js";
 import { removeExecuters } from "./executerManager.js";
 import { removeTransferrers } from "./transferManager.js";
-import { runDeferredCleanups, clearDeferredCleanups } from "./transferrer.js";
+import { runDeferredCleanups } from "./transferrer.js";
 import { defaultCleanupRemoteRoot, projectJsonFilename, componentJsonFilename } from "../db/db.js";
 import { setProjectState } from "../core/projectJsonFileOperator.js";
 import { writeComponentJson } from "./componentJsonIO.js";
@@ -23,9 +23,30 @@ const _internal = {
   eventEmitters,
   gitClean,
   gitResetHEAD,
-  setProjectState
+  setProjectState,
+  removeSsh,
+  removeExecuters,
+  removeTransferrers,
+  runDeferredCleanups
 };
 
+/**
+ * release the runtime resources (SSH connections, executers, transferrers) a project run
+ * accumulated - the 3-point cleanup set shared by every path that tears down a run
+ * (natural completion, an external stop, or a fatal dispatch error), consolidated here so
+ * every call site releases the same three things instead of each hand-rolling its own
+ * subset (aicshud/WHEEL#1022 - removeExecuters/removeTransferrers used to be missing from
+ * runDispatcher()'s fatal-error path, leaking stale executer/transferrer map entries that a
+ * subsequent run attempt could otherwise reuse).
+ * @param {string} projectRootDir - project's root path
+ */
+function releaseRuntimeResources(projectRootDir) {
+  _internal.removeExecuters(projectRootDir);
+  _internal.removeTransferrers(projectRootDir);
+  _internal.removeSsh(projectRootDir);
+}
+_internal.releaseRuntimeResources = releaseRuntimeResources;
+
 /**
  * @event projectStateChanged
  * @type {object} - updated projectJson
@@ -85,10 +106,12 @@ async function stopProject(projectRootDir, reasonState) {
     await rootDispatcher.remove();
     _internal.rootDispatchers.delete(projectRootDir);
   }
-  clearDeferredCleanups(projectRootDir);
-  removeExecuters(projectRootDir);
-  removeTransferrers(projectRootDir);
-  removeSsh(projectRootDir);
+  //deliberately do NOT discard pending deferred remote cleanups here (aicshud/WHEEL#1021):
+  //an entry means a not-yet-executed downstream task may still need the preserved
+  //remote-symlink target file on resume, and there is no way to safely tell from here
+  //whether that consumer has already run. Leave the registry as-is so a later natural
+  //completion (after 0+ resumes) can still process it via runDeferredCleanups().
+  _internal.releaseRuntimeResources(projectRootDir);
   //project state must be updated by onStopProject()
 }
 
@@ -123,11 +146,16 @@ async function runProject(projectRootDir) {
   await updateProjectState(projectRootDir, rootWF.state, projectJson);
   await writeComponentJson(projectRootDir, projectRootDir, rootWF, true);
   _internal.rootDispatchers.delete(projectRootDir);
-  await runDeferredCleanups(projectRootDir);
-  removeExecuters(projectRootDir);
-  removeTransferrers(projectRootDir);
-  removeSsh(projectRootDir);
+  //if the dispatcher settled via an external stop (stopProject() -> Dispatcher.remove() ->
+  //pause()), that caller already owns this teardown once its own await on remove() finishes -
+  //doing it again here races it, since start() now resolves as soon as pause() emits "stop",
+  //well before its still-in-flight nested job cancellation is done using these same SSH
+  //connections (aicshud/WHEEL#1020).
+  if (!rootDispatcher.stoppedExternally) {
+    await _internal.runDeferredCleanups(projectRootDir);
+    _internal.releaseRuntimeResources(projectRootDir);
+  }
   return rootWF.state;
 }
 
-export { cleanProject, runProject, stopProject, updateProjectState, _internal };
+export { cleanProject, runProject, stopProject, updateProjectState, releaseRuntimeResources, _internal };

server/app/handlers/projectController.js

@@ -21,7 +21,7 @@ import { checkRemoteStoragePathWritePermission } from "../core/checkRemoteStorag
 import { getProjectJson, getProjectState, setProjectState, updateProjectDescription, updateProjectROStatus } from "../core/projectJsonFileOperator.js";
 import { createSsh, removeSsh, askPassword } from "../core/sshManager.js";
 import { setJWTServerPassphrase, removeAllJWTServerPassphrase } from "../core/jwtServerPassphraseManager.js";
-import { runProject, cleanProject, stopProject } from "../core/projectController.js";
+import { runProject, cleanProject, stopProject, releaseRuntimeResources } from "../core/projectController.js";
 import { isValidOutputFilename } from "../lib/utility.js";
 import { checkWritePermissions, parentDirs, eventEmitters } from "../core/global.js";
 import { sendWorkflow, sendProjectJson, sendTaskStateList, sendResultsFileDir, sendComponentTree } from "./senders.js";
@@ -393,6 +393,17 @@ async function runDispatcher(clientID, projectRootDir, ack) {
   } catch (err) {
     notifyUser(projectRootDir, "fatal error occurred while parsing workflow:", err);
     await updateProjectState(projectRootDir, "failed");
+    //runProject() (core/projectController.js) owns releaseRuntimeResources() (removeSsh/
+    //removeExecuters/removeTransferrers) on every non-throwing exit (natural completion or
+    //external stop, aicshud/WHEEL#1020) - only do it here as a safety net for the case where
+    //runProject() threw before ever reaching its own cleanup (e.g. rootDispatcher.start()
+    //itself rejected). Doing this unconditionally in a shared `finally` below used to always
+    //fire on every exit path, including the external-stop one - racing stopProject()'s own
+    //still-in-flight teardown of the very same SSH connections. This used to call removeSsh()
+    //alone, leaking stale removeExecuters()/removeTransferrers() entries on this path
+    //(aicshud/WHEEL#1022) - use the same 3-point cleanup every other exit path uses instead.
+    releaseRuntimeResources(projectRootDir);
+    removeAllJWTServerPassphrase(projectRootDir);
     ack(err);
   } finally {
     //make sure any in-flight abort-on-failure work has fully settled before this handler
@@ -401,8 +412,6 @@ async function runDispatcher(clientID, projectRootDir, ack) {
     emitAll(projectRootDir, "projectJson", await getProjectJson(projectRootDir));
     await _internal.sendWorkflow(ack, projectRootDir);
     _internal.eventEmitters.delete(projectRootDir);
-    removeSsh(projectRootDir);
-    removeAllJWTServerPassphrase(projectRootDir);
   }
   return;
 }

server/test/app/core/dispatcher.js

@@ -1595,6 +1595,30 @@ describe("UT for Dispatcher class", function () {
     });
   });
 
+  //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
+  //investigation for why racing that teardown disconnects SSH out from under stopProject()'s
+  //still-in-flight nested job cancellation.
+  describe("#start stoppedExternally flag (aicshud/WHEEL#1020)", ()=>{
+    it("should set stoppedExternally and resolve when stopped via pause() before any component is dispatched", async ()=>{
+      const projectJson = await fs.readJson(path.resolve(projectRootDir, projectJsonFilename));
+      const DP = new Dispatcher(projectRootDir, rootWF.ID, projectRootDir, "dummy start time", projectJson.componentPath, {}, "");
+      const startPromise = DP.start();
+      await DP.pause();
+      const state = await startPromise;
+      expect(DP.stoppedExternally).to.be.true;
+      expect(state).to.equal("finished");
+    });
+
+    it("should NOT set stoppedExternally when the dispatcher finishes naturally", async ()=>{
+      const projectJson = await fs.readJson(path.resolve(projectRootDir, projectJsonFilename));
+      const DP = new Dispatcher(projectRootDir, rootWF.ID, projectRootDir, "dummy start time", projectJson.componentPath, {}, "");
+      await DP.start();
+      expect(DP.stoppedExternally).to.not.be.true;
+    });
+  });
+
   describe("#_checkMandatoryInputFilesExist", ()=>{
     let task;
     beforeEach(async ()=>{

server/test/app/core/projectController.js

@@ -20,6 +20,8 @@ import chaiAsPromised from "chai-as-promised";
 chai.use(chaiAsPromised);
 
 import { _internal, runProject, stopProject, cleanProject, updateProjectState } from "../../../app/core/projectController.js";
+import Dispatcher from "../../../app/core/dispatcher.js";
+import { runDeferredCleanups, _internal as transferrerInternal } from "../../../app/core/transferrer.js";
 
 //test data
 const testDirRoot = "WHEEL_TEST_TMP";
@@ -1142,6 +1144,53 @@ describe("project Controller UT", function () {
       expect(result).to.be.an("error");
       expect(result.message).to.include("project is already running");
     });
+
+    //reproduction for aicshud/WHEEL#1020: when a Dispatcher settles because it was stopped
+    //externally (stopProject() -> Dispatcher.remove() -> pause(), which resolves start()'s
+    //promise as soon as it emits "stop" - well before its own, still-in-flight nested
+    //cancellation of remote job tasks is done using the project's SSH connections),
+    //runProject() must not also tear down removeSsh/removeExecuters/removeTransferrers here:
+    //stopProject() already owns that once its own await on rootDispatcher.remove() finishes,
+    //and racing it by doing it again here can disconnect SSH out from under that still-running
+    //cancellation (observed as "ssh instance is not registerd for the project").
+    describe("when the dispatcher was stopped externally", ()=>{
+      let removeSshStub, removeExecutersStub, removeTransferrersStub, runDeferredCleanupsStub;
+      beforeEach(()=>{
+        sinon.stub(Dispatcher.prototype, "start").callsFake(function () {
+          this.stoppedExternally = true;
+          return Promise.resolve("stopped");
+        });
+        removeSshStub = sinon.stub(_internal, "removeSsh");
+        removeExecutersStub = sinon.stub(_internal, "removeExecuters");
+        removeTransferrersStub = sinon.stub(_internal, "removeTransferrers");
+        runDeferredCleanupsStub = sinon.stub(_internal, "runDeferredCleanups").resolves();
+      });
+      it("should not call removeSsh/removeExecuters/removeTransferrers/runDeferredCleanups", async ()=>{
+        await runProject(projectRootDir);
+        sinon.assert.notCalled(removeSshStub);
+        sinon.assert.notCalled(removeExecutersStub);
+        sinon.assert.notCalled(removeTransferrersStub);
+        sinon.assert.notCalled(runDeferredCleanupsStub);
+      });
+    });
+
+    describe("when the dispatcher finishes naturally (not stopped externally)", ()=>{
+      let removeSshStub, removeExecutersStub, removeTransferrersStub, runDeferredCleanupsStub;
+      beforeEach(()=>{
+        sinon.stub(Dispatcher.prototype, "start").resolves("finished");
+        removeSshStub = sinon.stub(_internal, "removeSsh");
+        removeExecutersStub = sinon.stub(_internal, "removeExecuters");
+        removeTransferrersStub = sinon.stub(_internal, "removeTransferrers");
+        runDeferredCleanupsStub = sinon.stub(_internal, "runDeferredCleanups").resolves();
+      });
+      it("should still call removeSsh/removeExecuters/removeTransferrers/runDeferredCleanups", async ()=>{
+        await runProject(projectRootDir);
+        sinon.assert.calledOnceWithExactly(removeSshStub, projectRootDir);
+        sinon.assert.calledOnceWithExactly(removeExecutersStub, projectRootDir);
+        sinon.assert.calledOnceWithExactly(removeTransferrersStub, projectRootDir);
+        sinon.assert.calledOnceWithExactly(runDeferredCleanupsStub, projectRootDir);
+      });
+    });
   });
   describe("#stopProject", ()=>{
     const projectRootDir = "/test/project";
@@ -1159,6 +1208,29 @@ describe("project Controller UT", function () {
       sinon.assert.calledOnce(mockDispatcher.remove);
       expect(_internal.rootDispatchers.has(projectRootDir)).to.be.false;
     });
+    //reproduction for aicshud/WHEEL#1021: stopProject() must not discard pending deferred
+    //remote cleanups (files preserved as remote-symlink targets for a not-yet-executed
+    //downstream task) - doing so leaks them permanently, since a resumed run never
+    //re-registers them (finished components are skipped) and cleanProject only touches
+    //local git state, never the remote host.
+    it("should NOT discard entries registered in the deferred-cleanup registry", async ()=>{
+      const getSshStub = sinon.stub(transferrerInternal, "getSsh").returns({ exec: sinon.stub().resolves() });
+      transferrerInternal.addDeferredCleanup(projectRootDir, {
+        remoteWorkingDir: "/remote/work/dir",
+        remotehostID: "dummyHostID",
+        symlinkTargetNames: ["result.txt"]
+      });
+
+      try {
+        await stopProject(projectRootDir);
+        //if the entry survived stopProject(), a later runDeferredCleanups() (e.g. after a
+        //resume eventually finishes naturally) must still find and process it.
+        await runDeferredCleanups(projectRootDir);
+        sinon.assert.calledOnce(getSshStub);
+      } finally {
+        getSshStub.restore();
+      }
+    });
     it("should handle the case where the dispatcher does not exist", async ()=>{
       _internal.rootDispatchers.delete(projectRootDir);
       await stopProject(projectRootDir);

server/test/app/handlers/projectController.js

@@ -22,6 +22,7 @@ import allowedOperations from "../../../../common/allowedOperations.js";
 
 //helper functions
 import { _internal as coreProjectControllerInternal } from "../../../app/core/projectController.js";
+import Dispatcher from "../../../app/core/dispatcher.js";
 import { createNewProject } from "../../../app/core/projectOperations.js";
 import { createNewComponent } from "../../../app/core/componentOperations.js";
 import { addLink } from "../../../app/core/componentLinks.js";
@@ -39,16 +40,18 @@ const scriptPwd = `${scriptHeader}\n${pwdCmd}`;
 /**
  * Wait until a project's dispatch has fully wound down after its ack has already
  * fired. onRunProject/onContinueProject fire-and-forget the actual dispatch
- * (runDispatcher), whose finally block calls ack (via sendWorkflow) *before* its
- * own remaining cleanup (eventEmitters.delete/removeSsh/removeAllJWTServerPassphrase)
- * runs, and dispatcher itself never awaits individual task execution (by design,
- * so a canceled task isn't left blocked forever inside it) - so a test that only
- * awaits the ack can still race the next test's beforeEach (which removes/
- * recreates projectRootDir) against trailing dispatcher/task cleanup work,
- * intermittently crashing with ENOENT on stale .git/objects files. Poll for the
- * eventEmitter runDispatcher registers for the run (and clears right before its
- * own trailing cleanup) as a best-effort signal, then add a short fixed grace
- * period on top for any other fire-and-forgotten work outside that scope.
+ * (runDispatcher), whose finally block calls ack (via sendWorkflow) *before*
+ * eventEmitters.delete runs (removeSsh/removeAllJWTServerPassphrase only run there
+ * too if runProject() itself threw before reaching its own cleanup - see
+ * aicshud/WHEEL#1020 - otherwise runProject() already handled them), and dispatcher
+ * itself never awaits individual task execution (by design, so a canceled task
+ * isn't left blocked forever inside it) - so a test that only awaits the ack can
+ * still race the next test's beforeEach (which removes/recreates projectRootDir)
+ * against trailing dispatcher/task cleanup work, intermittently crashing with
+ * ENOENT on stale .git/objects files. Poll for the eventEmitter runDispatcher
+ * registers for the run (and clears right before its own trailing cleanup) as a
+ * best-effort signal, then add a short fixed grace period on top for any other
+ * fire-and-forgotten work outside that scope.
  * @param {string} projectRootDir - project's root path
  * @param {number} timeoutMs - give up polling and move on to the grace period after this long
  */
@@ -475,4 +478,26 @@ describe("project Controller handler UT", function () {
       await drainProjectDispatch(projectRootDir);
     });
   });
+
+  describe("[reproduction] issue aicshud/WHEEL#1022 - removeExecuters/removeTransferrers are not released on runDispatcher's fatal-error path", ()=>{
+    it("should call removeExecuters/removeTransferrers as well as removeSsh when runProject() throws before ever reaching its own cleanup", async ()=>{
+      //simulate rootDispatcher.start() itself rejecting (e.g. a bug in the dispatch loop
+      //itself, not a task failure) - this is the one path where runProject() throws before
+      //ever reaching its own removeSsh/removeExecuters/removeTransferrers teardown, so
+      //runDispatcher()'s own catch block is the last chance to release these resources.
+      sinon.stub(Dispatcher.prototype, "start").rejects(new Error("dummy dispatch failure"));
+      const removeExecutersStub = sinon.stub(coreProjectControllerInternal, "removeExecuters");
+      const removeTransferrersStub = sinon.stub(coreProjectControllerInternal, "removeTransferrers");
+      const removeSshStub = sinon.stub(coreProjectControllerInternal, "removeSsh");
+
+      await new Promise((resolve)=>{
+        _internal.onRunProject("test-client-id", projectRootDir, resolve);
+      });
+      await drainProjectDispatch(projectRootDir);
+
+      sinon.assert.calledOnceWithExactly(removeSshStub, projectRootDir);
+      sinon.assert.calledOnceWithExactly(removeExecutersStub, projectRootDir);
+      sinon.assert.calledOnceWithExactly(removeTransferrersStub, projectRootDir);
+    });
+  });
 });