From 52cbddf61639f12c8528c5a0384dcdba0c22593d Mon Sep 17 00:00:00 2001
From: Brian Botha <brian.botha@matrix.ai>
Date: Fri, 6 Dec 2024 17:44:50 +1100
Subject: [PATCH] fix: cleaning up nodes tasks during startup that didn't clean
 up properly during ungraceful shutdown

[ci skip]
---
 src/nodes/NodeManager.ts | 27 ++++++++++++++++++---------
 1 file changed, 18 insertions(+), 9 deletions(-)

diff --git a/src/nodes/NodeManager.ts b/src/nodes/NodeManager.ts
index 6ab4b5b8e..8d7e72063 100644
--- a/src/nodes/NodeManager.ts
+++ b/src/nodes/NodeManager.ts
@@ -403,6 +403,8 @@ class NodeManager {
       this.syncNodeGraphHandlerId,
       this.syncNodeGraphHandler,
     );
+    // This will clean up tasks that were not properly cleaned up during an ungracefully shutdown of the process
+    await this.stopTasks();
     await this.setupRefreshBucketTasks();
     // Can be disabled with 0 delay, only use for testing
     if (this.retryConnectionsDelayTime > 0) {
@@ -445,15 +447,7 @@ class NodeManager {
       this.handleEventNodeConnectionManagerConnection,
     );
     await this.mdns?.stop();
-    // Cancels all NodeManager tasks
-    const taskPs: Array<Promise<any>> = [];
-    for await (const task of this.taskManager.getTasks(undefined, false, [
-      this.tasksPath,
-    ])) {
-      taskPs.push(task.promise());
-      task.cancel(abortEphemeralTaskReason);
-    }
-    await Promise.allSettled(taskPs);
+    await this.stopTasks();
     this.taskManager.deregisterHandler(this.refreshBucketHandlerId);
     this.taskManager.deregisterHandler(this.gcBucketHandlerId);
     this.taskManager.deregisterHandler(this.checkConnectionsHandlerId);
@@ -2089,6 +2083,21 @@ class NodeManager {
     this.logger.info('Set up refreshBucket tasks');
   }
 
+  /**
+   * Cancels all NodeManager tasks.
+   * These are ephemeral and will be recreated next time we start.
+   */
+  protected async stopTasks() {
+    const taskPs: Array<Promise<any>> = [];
+    for await (const task of this.taskManager.getTasks(undefined, false, [
+      this.tasksPath,
+    ])) {
+      taskPs.push(task.promise());
+      task.cancel(abortEphemeralTaskReason);
+    }
+    await Promise.allSettled(taskPs);
+  }
+
   @ready(new nodesErrors.ErrorNodeManagerNotRunning(), true, ['stopping'])
   public async updateRefreshBucketDelay(
     bucketIndex: number,