feat: allow joining jobs in peer queues (#2316)

When we have a PeerJobQueue, the jobs are usually expensive but the
result is transferable - opening a connection, etc.

This PR adds a `joinJob` method to `PeerJobQueue` - if a job is queued
for the passed `PeerId`, a promise is returned that resolves/rejects
based on the outcome of the job.

packages/utils/package.json

@@ -112,6 +112,7 @@
   "dependencies": {
     "@chainsafe/is-ip": "^2.0.2",
     "@libp2p/interface": "^1.0.2",
+    "@libp2p/peer-collections": "^5.1.1",
     "@multiformats/multiaddr": "^12.1.10",
     "@multiformats/multiaddr-matcher": "^1.1.0",
     "get-iterator": "^2.0.1",

packages/utils/src/peer-job-queue.ts

@@ -1,8 +1,12 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 
 import { CodeError, ERR_INVALID_PARAMETERS } from '@libp2p/interface'
+import { PeerMap } from '@libp2p/peer-collections'
+import pDefer from 'p-defer'
 import PQueue from 'p-queue'
 import type { PeerId } from '@libp2p/interface'
+import type { AbortOptions } from 'it-pushable'
+import type { DeferredPromise } from 'p-defer'
 import type { QueueAddOptions, Options, Queue } from 'p-queue'
 
 // Port of lower_bound from https://en.cppreference.com/w/cpp/algorithm/lower_bound
@@ -26,7 +30,9 @@ function lowerBound<T> (array: readonly T[], value: T, comparator: (a: T, b: T)
   return first
 }
 
-interface RunFunction { (): Promise<unknown> }
+interface RunFunction<T> {
+  (options?: AbortOptions): Promise<T>
+}
 
 export interface PeerPriorityQueueOptions extends QueueAddOptions {
   peerId: PeerId
@@ -35,17 +41,17 @@ export interface PeerPriorityQueueOptions extends QueueAddOptions {
 interface PeerJob {
   priority: number
   peerId: PeerId
-  run: RunFunction
+  run: RunFunction<any>
 }
 
 /**
  * Port of https://github.com/sindresorhus/p-queue/blob/main/source/priority-queue.ts
  * that adds support for filtering jobs by peer id
  */
-class PeerPriorityQueue implements Queue<RunFunction, PeerPriorityQueueOptions> {
+class PeerPriorityQueue implements Queue<RunFunction<unknown>, PeerPriorityQueueOptions> {
   readonly #queue: PeerJob[] = []
 
-  enqueue (run: RunFunction, options?: Partial<PeerPriorityQueueOptions>): void {
+  enqueue (run: RunFunction<unknown>, options?: Partial<PeerPriorityQueueOptions>): void {
     const peerId = options?.peerId
     const priority = options?.priority ?? 0
 
@@ -71,23 +77,23 @@ class PeerPriorityQueue implements Queue<RunFunction, PeerPriorityQueueOptions>
     this.#queue.splice(index, 0, element)
   }
 
-  dequeue (): RunFunction | undefined {
+  dequeue (): RunFunction<unknown> | undefined {
     const item = this.#queue.shift()
     return item?.run
   }
 
-  filter (options: Readonly<Partial<PeerPriorityQueueOptions>>): RunFunction[] {
+  filter (options: Readonly<Partial<PeerPriorityQueueOptions>>): Array<RunFunction<unknown>> {
     if (options.peerId != null) {
       const peerId = options.peerId
 
       return this.#queue.filter(
         (element: Readonly<PeerPriorityQueueOptions>) => peerId.equals(element.peerId)
-      ).map((element: Readonly<{ run: RunFunction }>) => element.run)
+      ).map((element: Readonly<{ run: RunFunction<unknown> }>) => element.run)
     }
 
     return this.#queue.filter(
       (element: Readonly<PeerPriorityQueueOptions>) => element.priority === options.priority
-    ).map((element: Readonly<{ run: RunFunction }>) => element.run)
+    ).map((element: Readonly<{ run: RunFunction<unknown> }>) => element.run)
   }
 
   get size (): number {
@@ -99,20 +105,78 @@ class PeerPriorityQueue implements Queue<RunFunction, PeerPriorityQueueOptions>
  * Extends PQueue to add support for querying queued jobs by peer id
  */
 export class PeerJobQueue extends PQueue<PeerPriorityQueue, PeerPriorityQueueOptions> {
+  private readonly results: PeerMap<DeferredPromise<any> | true>
+
   constructor (options: Options<PeerPriorityQueue, PeerPriorityQueueOptions> = {}) {
     super({
       ...options,
       queueClass: PeerPriorityQueue
     })
+
+    this.results = new PeerMap()
   }
 
   /**
-   * Returns true if this queue has a job for the passed peer id that has not yet
-   * started to run
+   * Returns true if this queue has a job for the passed peer id that has not
+   * yet started to run
    */
   hasJob (peerId: PeerId): boolean {
     return this.sizeBy({
       peerId
     }) > 0
   }
+
+  /**
+   * Returns a promise for the result of the job in the queue for the passed
+   * peer id.
+   */
+  async joinJob <Result = void> (peerId: PeerId): Promise<Result> {
+    let deferred = this.results.get(peerId)
+
+    if (deferred == null) {
+      throw new CodeError('No job found for peer id', 'ERR_NO_JOB_FOR_PEER_ID')
+    }
+
+    if (deferred === true) {
+      // a job has been added but so far nothing has tried to join the job
+      deferred = pDefer<Result>()
+      this.results.set(peerId, deferred)
+    }
+
+    return deferred.promise
+  }
+
+  async add <T> (fn: RunFunction<T>, opts: PeerPriorityQueueOptions): Promise<T> {
+    const peerId = opts?.peerId
+
+    if (peerId == null) {
+      throw new CodeError('missing peer id', ERR_INVALID_PARAMETERS)
+    }
+
+    this.results.set(opts.peerId, true)
+
+    return super.add(async (opts?: AbortOptions) => {
+      try {
+        const value = await fn(opts)
+
+        const deferred = this.results.get(peerId)
+
+        if (deferred != null && deferred !== true) {
+          deferred.resolve(value)
+        }
+
+        return value
+      } catch (err) {
+        const deferred = this.results.get(peerId)
+
+        if (deferred != null && deferred !== true) {
+          deferred.reject(err)
+        }
+
+        throw err
+      } finally {
+        this.results.delete(peerId)
+      }
+    }, opts) as Promise<T>
+  }
 }

packages/utils/src/stream-to-ma-conn.ts

@@ -44,7 +44,7 @@ export function streamToMaConnection (props: StreamProperties): MultiaddrConnect
         // If the source errored the socket will already have been destroyed by
         // toIterable.duplex(). If the socket errored it will already be
         // destroyed. There's nothing to do here except log the error & return.
-        log(err)
+        log.error('%s error in sink', remoteAddr, err)
       }
     } finally {
       closedWrite = true

packages/utils/test/peer-job-queue.spec.ts

@@ -37,4 +37,60 @@ describe('peer job queue', () => {
 
     expect(queue.hasJob(peerIdA)).to.be.false()
   })
+
+  it('can join existing jobs', async () => {
+    const value = 'hello world'
+    const deferred = pDefer<string>()
+
+    const peerIdA = await createEd25519PeerId()
+    const queue = new PeerJobQueue({
+      concurrency: 1
+    })
+
+    expect(queue.hasJob(peerIdA)).to.be.false()
+
+    await expect(queue.joinJob(peerIdA)).to.eventually.rejected
+      .with.property('code', 'ERR_NO_JOB_FOR_PEER_ID')
+
+    void queue.add(async () => {
+      return deferred.promise
+    }, {
+      peerId: peerIdA
+    })
+
+    const join = queue.joinJob<string>(peerIdA)
+
+    deferred.resolve(value)
+
+    await expect(join).to.eventually.equal(value)
+
+    expect(queue.hasJob(peerIdA)).to.be.false()
+
+    await expect(queue.joinJob(peerIdA)).to.eventually.rejected
+      .with.property('code', 'ERR_NO_JOB_FOR_PEER_ID')
+  })
+
+  it('can join an existing job that fails', async () => {
+    const error = new Error('nope!')
+    const deferred = pDefer<string>()
+
+    const peerIdA = await createEd25519PeerId()
+    const queue = new PeerJobQueue({
+      concurrency: 1
+    })
+
+    void queue.add(async () => {
+      return deferred.promise
+    }, {
+      peerId: peerIdA
+    })
+      .catch(() => {})
+
+    const joinedJob = queue.joinJob(peerIdA)
+
+    deferred.reject(error)
+
+    await expect(joinedJob).to.eventually.rejected
+      .with.property('message', error.message)
+  })
 })