wip: NodeManager using Scheduler and Queue

src/PolykeyAgent.ts

@@ -336,7 +336,8 @@ class PolykeyAgent {
           keyManager,
           nodeGraph,
           nodeConnectionManager,
-          queue,
+          scheduler: taskScheduler,
+          taskQueue,
           logger: logger.getChild(NodeManager.name),
         });
       await nodeManager.start();

src/nodes/NodeManager.ts

@@ -1,7 +1,6 @@
 import type { DB } from '@matrixai/db';
 import type NodeConnectionManager from './NodeConnectionManager';
 import type NodeGraph from './NodeGraph';
-import type Queue from './Queue';
 import type KeyManager from '../keys/KeyManager';
 import type { PublicKeyPem } from '../keys/types';
 import type Sigchain from '../sigchain/Sigchain';
@@ -12,6 +11,9 @@ import type { Timer } from '../types';
 import type { PromiseDeconstructed } from '../types';
 import type { DBTransaction } from '@matrixai/db';
 import type { NodeId, NodeAddress } from '../nodes/types';
+import type Scheduler from '../tasks/Scheduler';
+import type TaskQueue from '../tasks/Queue';
+import type { TaskHandlerId } from '../tasks/types';
 import Logger from '@matrixai/logger';
 import { StartStop, ready } from '@matrixai/async-init/dist/StartStop';
 import * as nodesErrors from './errors';
@@ -33,7 +35,8 @@ class NodeManager {
   protected keyManager: KeyManager;
   protected nodeConnectionManager: NodeConnectionManager;
   protected nodeGraph: NodeGraph;
-  protected queue: Queue;
+  protected scheduler: Scheduler;
+  protected taskQueue: TaskQueue;
   // Refresh bucket timer
   protected refreshBucketDeadlineMap: Map<NodeBucketIndex, number> = new Map();
   protected refreshBucketTimer: NodeJS.Timer;
@@ -47,13 +50,16 @@ class NodeManager {
   protected refreshBucketQueuePause_: PromiseDeconstructed<void> = promise();
   protected refreshBucketQueueAbortController: AbortController;
 
+  public readonly setNodeHanderId = 'setNode' as TaskHandlerId;
+
   constructor({
     db,
     keyManager,
     sigchain,
     nodeConnectionManager,
     nodeGraph,
-    queue,
+    scheduler,
+    taskQueue,
     refreshBucketTimerDefault = 3600000, // 1 hour in milliseconds
     logger,
   }: {
@@ -62,7 +68,8 @@ class NodeManager {
     sigchain: Sigchain;
     nodeConnectionManager: NodeConnectionManager;
     nodeGraph: NodeGraph;
-    queue: Queue;
+    scheduler: Scheduler;
+    taskQueue: TaskQueue;
     refreshBucketTimerDefault?: number;
     logger?: Logger;
   }) {
@@ -72,21 +79,30 @@ class NodeManager {
     this.sigchain = sigchain;
     this.nodeConnectionManager = nodeConnectionManager;
     this.nodeGraph = nodeGraph;
-    this.queue = queue;
+    this.scheduler = scheduler;
+    this.taskQueue = taskQueue;
     this.refreshBucketTimerDefault = refreshBucketTimerDefault;
   }
 
   public async start() {
     this.logger.info(`Starting ${this.constructor.name}`);
     this.startRefreshBucketTimers();
     this.refreshBucketQueueRunner = this.startRefreshBucketQueue();
+    this.logger.info(`Registering handler for setNode`);
+    this.taskQueue.registerHandler(
+      this.setNodeHanderId,
+      async (nodeId, nodeAddress, timeout) =>
+        this.setNode(nodeId, nodeAddress, true, false, timeout),
+    );
     this.logger.info(`Started ${this.constructor.name}`);
   }
 
   public async stop() {
     this.logger.info(`Stopping ${this.constructor.name}`);
     await this.stopRefreshBucketTimers();
     await this.stopRefreshBucketQueue();
+    this.logger.info(`Unregistering handler for setNode`);
+    this.taskQueue.deregisterHandler(this.setNodeHanderId);
     this.logger.info(`Stopped ${this.constructor.name}`);
   }
 
@@ -478,9 +494,15 @@ class NodeManager {
             nodeId,
           )} to queue`,
         );
-        // Re-attempt this later asynchronously by adding the the queue
-        this.queue.push(() =>
-          this.setNode(nodeId, nodeAddress, true, false, timeout),
+        // Re-attempt this later asynchronously by adding to the scheduler
+        await this.scheduler.scheduleTask(
+          this.setNodeHanderId,
+          [nodeId, nodeAddress, timeout],
+          undefined,
+          undefined,
+          ['setNode'],
+          true,
+          tran,
         );
       }
     }
@@ -494,6 +516,8 @@ class NodeManager {
   ) {
     const oldestNodeIds = await this.nodeGraph.getOldestNode(bucketIndex, 3);
     // We want to concurrently ping the nodes
+    // Fixme, remove concurrency? we'd want to stick to 1 active connection per
+    //  background task
     const pingPromises = oldestNodeIds.map((nodeId) => {
       const doPing = async (): Promise<{
         nodeId: NodeId;