Address code review comments

src/blob-store/index.js

@@ -137,13 +137,7 @@ export class BlobStore extends TypedEmitter {
     const entriesStream = createEntriesStream(this.#driveIndex, { live })
     if (!filter) return entriesStream
     const filterStream = new FilterEntriesStream(filter)
-    const result = pipeline(entriesStream, filterStream, noop)
-    // Destroying the pipeline should destroy the *first* stream, not the
-    // whole pipeline.
-    Object.defineProperty(result, 'destroy', {
-      value: entriesStream.destroy.bind(entriesStream),
-    })
-    return result
+    return pipeline(entriesStream, filterStream, noop)
   }
 
   /**

src/discovery/local-discovery.js

@@ -9,6 +9,7 @@ import StartStopStateMachine from 'start-stop-state-machine'
 import pTimeout from 'p-timeout'
 import { keyToPublicId } from '@mapeo/crypto'
 import { Logger } from '../logger.js'
+import { getErrorCode } from '../lib/error.js'
 /** @import { OpenedNoiseStream } from '../lib/noise-secret-stream-helpers.js' */
 
 /** @typedef {{ publicKey: Buffer, secretKey: Buffer }} Keypair */
@@ -117,7 +118,7 @@ export class LocalDiscovery extends TypedEmitter {
 
     /** @param {Error} e */
     function onSocketError(e) {
-      if ('code' in e && e.code === 'EPIPE') {
+      if (getErrorCode(e) === 'EPIPE') {
         socket.destroy()
         if (secretStream) {
           secretStream.destroy()

src/fastify-plugins/maps.js

@@ -5,6 +5,7 @@ import { ReaderWatch, Server as SMPServerPlugin } from 'styled-map-package'
 
 import { noop } from '../utils.js'
 import { NotFoundError, ENOENTError } from './utils.js'
+import { getErrorCode } from '../lib/error.js'
 
 /** @import { FastifyPluginAsync } from 'fastify' */
 /** @import { Stats } from 'node:fs' */
@@ -56,7 +57,7 @@ export async function plugin(fastify, opts) {
       try {
         stats = await fs.stat(customMapPath)
       } catch (err) {
-        if (err instanceof Error && 'code' in err && err.code === 'ENOENT') {
+        if (getErrorCode(err) === 'ENOENT') {
           throw new ENOENTError(customMapPath)
         }
 

src/lib/error.js

@@ -21,6 +21,30 @@ export class ErrorWithCode extends Error {
   }
 }
 
+/**
+ * If the argument is an `Error` instance, return its `code` property if it is a string.
+ * Otherwise, returns `undefined`.
+ *
+ * @param {unknown} maybeError
+ * @returns {undefined | string}
+ * @example
+ * try {
+ *   // do something
+ * } catch (err) {
+ *   console.error(getErrorCode(err))
+ * }
+ */
+export function getErrorCode(maybeError) {
+  if (
+    maybeError instanceof Error &&
+    'code' in maybeError &&
+    typeof maybeError.code === 'string'
+  ) {
+    return maybeError.code
+  }
+  return undefined
+}
+
 /**
  * Get the error message from an object if possible.
  * Otherwise, stringify the argument.

src/mapeo-project.js

@@ -60,6 +60,7 @@ import { IconApi } from './icon-api.js'
 import { readConfig } from './config-import.js'
 import TranslationApi from './translation-api.js'
 import { NotFoundError } from './errors.js'
+import { getErrorCode, getErrorMessage } from './lib/error.js'
 /** @import { ProjectSettingsValue } from '@comapeo/schema' */
 /** @import { CoreStorage, BlobFilter, BlobStoreEntriesStream, KeyPair, Namespace, ReplicationStream } from './types.js' */
 
@@ -435,6 +436,7 @@ export class MapeoProject extends TypedEmitter {
 
     /** @type {Map<string, BlobStoreEntriesStream>} */
     const entriesReadStreams = new Map()
+
     this.#coreManager.on('peer-download-intent', async (filter, peerId) => {
       entriesReadStreams.get(peerId)?.destroy()
 
@@ -444,16 +446,36 @@ export class MapeoProject extends TypedEmitter {
       })
       entriesReadStreams.set(peerId, entriesReadStream)
 
+      entriesReadStream.once('close', () => {
+        if (entriesReadStreams.get(peerId) === entriesReadStream) {
+          entriesReadStreams.delete(peerId)
+        }
+      })
+
       this.#syncApi[kClearBlobWantRanges](peerId)
 
-      for await (const entry of entriesReadStream) {
-        if (entriesReadStream.destroyed) break
-        const { blockOffset, blockLength } = entry.value.blob
-        this.#syncApi[kAddBlobWantRange](peerId, blockOffset, blockLength)
-        if (entriesReadStream.destroyed) break
+      try {
+        for await (const entry of entriesReadStream) {
+          const { blockOffset, blockLength } = entry.value.blob
+          this.#syncApi[kAddBlobWantRange](peerId, blockOffset, blockLength)
+        }
+      } catch (err) {
+        if (getErrorCode(err) === 'ERR_STREAM_PREMATURE_CLOSE') return
+        this.#l.log(
+          'Error getting blob entries stream for peer %h: %s',
+          peerId,
+          getErrorMessage(err)
+        )
       }
     })
 
+    this.#coreManager.creatorCore.on('peer-remove', (peer) => {
+      const peerKey = peer.protomux.stream.remotePublicKey
+      const peerId = peerKey.toString('hex')
+      entriesReadStreams.get(peerId)?.destroy()
+      entriesReadStreams.delete(peerId)
+    })
+
     this.#translationApi = new TranslationApi({
       dataType: this.#dataTypes.translation,
     })

test/lib/error.js

@@ -1,6 +1,10 @@
 import assert from 'node:assert/strict'
 import test, { describe } from 'node:test'
-import { ErrorWithCode, getErrorMessage } from '../../src/lib/error.js'
+import {
+  ErrorWithCode,
+  getErrorCode,
+  getErrorMessage,
+} from '../../src/lib/error.js'
 
 describe('ErrorWithCode', () => {
   test('ErrorWithCode with two arguments', () => {
@@ -22,6 +26,44 @@ describe('ErrorWithCode', () => {
   })
 })
 
+describe('getErrorCode', () => {
+  test('from values without a string code', () => {
+    class ErrorWithNumericCode extends Error {
+      code = 123
+    }
+
+    const testCases = [
+      undefined,
+      null,
+      'ignored',
+      { code: 'ignored' },
+      new Error('has no code'),
+      new ErrorWithNumericCode(),
+    ]
+
+    for (const testCase of testCases) {
+      assert.equal(getErrorCode(testCase), undefined)
+    }
+  })
+
+  test('from Errors with a string code', () => {
+    class ErrorWithInheritedCode extends Error {
+      get code() {
+        return 'foo'
+      }
+    }
+
+    const testCases = [
+      new ErrorWithCode('foo', 'message'),
+      new ErrorWithInheritedCode(),
+    ]
+
+    for (const testCase of testCases) {
+      assert.equal(getErrorCode(testCase), 'foo')
+    }
+  })
+})
+
 describe('getErrorMessage', () => {
   test('from objects without a string message', () => {
     const testCases = [