fix: support resumable pinning

Author: SgtPooki

packages/helia/src/pins.ts

@@ -35,6 +35,17 @@ interface DatastorePinnedBlock {
   pinnedBy: Uint8Array[]
 }
 
+/**
+ * Callback for updating a {@link DatastorePinnedBlock}'s properties when
+ * calling `#updatePinnedBlock`
+ *
+ * The callback should return `false` to prevent any pinning modifications to
+ * the block, and true in all other cases.
+ */
+interface WithPinnedBlockCallback {
+  (pinnedBlock: DatastorePinnedBlock): boolean
+}
+
 const DATASTORE_PIN_PREFIX = '/pin/'
 const DATASTORE_BLOCK_PREFIX = '/pinned-block/'
 const DATASTORE_ENCODING = base36
@@ -82,11 +93,12 @@ export class PinsImpl implements Pins {
         await this.#updatePinnedBlock(cid, (pinnedBlock: DatastorePinnedBlock) => {
           // do not update pinned block if this block is already pinned by this CID
           if (pinnedBlock.pinnedBy.find(c => uint8ArrayEquals(c, cid.bytes)) != null) {
-            return
+            return false
           }
 
           pinnedBlock.pinCount++
           pinnedBlock.pinnedBy.push(cid.bytes)
+          return true
         }, options)
 
         return cid
@@ -157,7 +169,7 @@ export class PinsImpl implements Pins {
   /**
    * Update the pin count for the CID
    */
-  async #updatePinnedBlock (cid: CID, withPinnedBlock: (pinnedBlock: DatastorePinnedBlock) => void, options: AddOptions): Promise<void> {
+  async #updatePinnedBlock (cid: CID, withPinnedBlock: WithPinnedBlockCallback, options: AddOptions): Promise<void> {
     const blockKey = new Key(`${DATASTORE_BLOCK_PREFIX}${DATASTORE_ENCODING.encode(cid.multihash.bytes)}`)
 
     let pinnedBlock: DatastorePinnedBlock = {
@@ -173,7 +185,10 @@ export class PinsImpl implements Pins {
       }
     }
 
-    withPinnedBlock(pinnedBlock)
+    const shouldContinue = withPinnedBlock(pinnedBlock)
+    if (!shouldContinue) {
+      return
+    }
 
     if (pinnedBlock.pinCount === 0) {
       if (await this.datastore.has(blockKey)) {
@@ -197,9 +212,10 @@ export class PinsImpl implements Pins {
       yield async () => {
         const cid = await promise()
 
-        await this.#updatePinnedBlock(cid, (pinnedBlock): void => {
+        await this.#updatePinnedBlock(cid, (pinnedBlock): boolean => {
           pinnedBlock.pinCount--
           pinnedBlock.pinnedBy = pinnedBlock.pinnedBy.filter(c => uint8ArrayEquals(c, cid.bytes))
+          return true
         }, {
           ...options,
           depth: pin.depth

packages/helia/test/pins.recursive.spec.ts

@@ -8,7 +8,7 @@ import { MemoryDatastore } from 'datastore-core'
 import all from 'it-all'
 import parallel from 'it-parallel'
 import { createLibp2p } from 'libp2p'
-import { createHelia } from '../src/index.js'
+import { type AddPinEvents, createHelia } from '../src/index.js'
 import { createDag, type DAGNode } from './fixtures/create-dag.js'
 import { dagWalker } from './fixtures/dag-walker.js'
 import type { Helia } from '@helia/interface'
@@ -87,4 +87,35 @@ describe('pins (recursive)', () => {
       }
     }
   })
+
+  it('can resume an interrupted pinning operation', async () => {
+    // dag has 13 nodes. We should abort after 5
+    const events: AddPinEvents[] = []
+    const getPinIterator = (): ReturnType<typeof helia.pins.add> => helia.pins.add(dag['level-0'].cid, {
+      onProgress: (evt) => {
+        if (evt.type === 'helia:pin:add') {
+          events.push(evt)
+        }
+      }
+    })
+    const pinIter = getPinIterator()
+    let output = await pinIter.next()
+    const firstTryPins = []
+
+    while (output.done === false && events.length < 5) {
+      firstTryPins.push(await output.value())
+      output = await pinIter.next()
+    }
+
+    expect(firstTryPins).to.have.lengthOf(5)
+    expect(events.length).to.eq(5)
+    expect(output.done).to.eq(false) // we're not actually done. We simulated a crash
+
+    // now resume, and consume the entire iterator to completion
+    const pin = await all(parallel(getPinIterator()))
+
+    expect(pin).to.have.lengthOf(13)
+    // we did not re-pin things we already pinned
+    expect(events.length).to.eq(13)
+  })
 })