fix: survive a cid causing an error during gc (#38)

Pass an error event to the on progress handler if an error occurs during
gc.

packages/helia/package.json

@@ -142,6 +142,7 @@
     "@ipld/dag-pb": "^4.0.2",
     "@libp2p/interface-libp2p": "^1.1.0",
     "@libp2p/interfaces": "^3.3.1",
+    "@libp2p/logger": "^2.0.5",
     "blockstore-core": "^3.0.0",
     "cborg": "^1.10.0",
     "datastore-core": "^8.0.4",

packages/helia/src/helia.ts

@@ -1,6 +1,7 @@
 import type { GCOptions, Helia } from '@helia/interface'
 import type { Libp2p } from '@libp2p/interface-libp2p'
 import type { Datastore } from 'interface-datastore'
+import type { CID } from 'multiformats/cid'
 import { identity } from 'multiformats/hashes/identity'
 import { sha256, sha512 } from 'multiformats/hashes/sha2'
 import type { MultihashHasher } from 'multiformats/hashes/interface'
@@ -14,6 +15,9 @@ import drain from 'it-drain'
 import { CustomProgressEvent } from 'progress-events'
 import { MemoryDatastore } from 'datastore-core'
 import { MemoryBlockstore } from 'blockstore-core'
+import { logger } from '@libp2p/logger'
+
+const log = logger('helia')
 
 export class HeliaImpl implements Helia {
   public libp2p: Libp2p
@@ -99,19 +103,28 @@ export class HeliaImpl implements Helia {
       const helia = this
       const blockstore = this.blockstore.unwrap()
 
+      log('gc start')
+
       await drain(blockstore.deleteMany((async function * () {
         for await (const cid of blockstore.queryKeys({})) {
-          if (await helia.pins.isPinned(cid, options)) {
-            continue
-          }
+          try {
+            if (await helia.pins.isPinned(cid, options)) {
+              continue
+            }
 
-          yield cid
+            yield cid
 
-          options.onProgress?.(new CustomProgressEvent('helia:gc:deleted', cid))
+            options.onProgress?.(new CustomProgressEvent<CID>('helia:gc:deleted', cid))
+          } catch (err) {
+            log.error('Error during gc', err)
+            options.onProgress?.(new CustomProgressEvent<Error>('helia:gc:error', err))
+          }
         }
       }())))
     } finally {
       releaseLock()
     }
+
+    log('gc finished')
   }
 }

packages/helia/test/gc.spec.ts

@@ -7,7 +7,7 @@ import { webSockets } from '@libp2p/websockets'
 import { noise } from '@chainsafe/libp2p-noise'
 import { yamux } from '@chainsafe/libp2p-yamux'
 import { createHelia } from '../src/index.js'
-import type { Helia } from '@helia/interface'
+import type { GcEvents, Helia } from '@helia/interface'
 import * as raw from 'multiformats/codecs/raw'
 import { createBlock } from './fixtures/create-block.js'
 import * as dagPb from '@ipld/dag-pb'
@@ -168,4 +168,29 @@ describe('gc', () => {
     await expect(helia.blockstore.has(cid)).to.eventually.be.true()
     await expect(helia.blockstore.has(doomed)).to.eventually.be.false()
   })
+
+  it('can garbage collect around a CID that causes an error', async () => {
+    const cid = await createBlock(0x10, Uint8Array.from([0, 1, 2, 3]), helia.blockstore)
+
+    await expect(helia.blockstore.has(cid)).to.eventually.be.true('did not have cid')
+
+    const events: GcEvents[] = []
+
+    // make the datastore break in some way
+    helia.datastore.has = async () => {
+      throw new Error('Urk!')
+    }
+
+    await helia.gc({
+      onProgress: (evt) => {
+        events.push(evt)
+      }
+    })
+
+    await expect(helia.blockstore.has(cid)).to.eventually.be.true('did not keep cid')
+
+    const errorEvents = events.filter(e => e.type === 'helia:gc:error')
+    expect(errorEvents).to.have.lengthOf(1)
+    expect(errorEvents[0].detail.toString()).to.include('Urk!')
+  })
 })

packages/interface/src/index.ts

@@ -64,7 +64,8 @@ export interface Helia {
 }
 
 export type GcEvents =
-  ProgressEvent<'helia:gc:deleted', CID>
+  ProgressEvent<'helia:gc:deleted', CID> |
+  ProgressEvent<'helia:gc:error', Error>
 
 export interface GCOptions extends AbortOptions, ProgressOptions<GcEvents> {