feat!: streaming blockstores (#358)

Changes the blockstore interface to return (and accept) streams of
`Uint8Array`s instead of individual `Uint8Array`s.

This is to enable future work on large blocks (for streaming hashes)
and also to reduce memory usage since we can stream from the
filesystem in smaller chunks (typically 64KiB) rather than 256KiB at
a time.

BREAKING CHANGE: blockstore.get and similar now return streams of bytes

Author: achingbrain

packages/blockstore-core/package.json

@@ -187,15 +187,16 @@
     "@libp2p/logger": "^6.0.0",
     "interface-blockstore": "^5.0.0",
     "interface-store": "^6.0.0",
+    "it-all": "^3.0.9",
     "it-filter": "^3.1.3",
     "it-merge": "^3.0.11",
     "multiformats": "^13.3.6"
   },
   "devDependencies": {
     "aegir": "^47.0.16",
     "interface-blockstore-tests": "^7.0.0",
-    "it-all": "^3.0.8",
     "it-drain": "^3.0.9",
+    "it-to-buffer": "^4.0.10",
     "uint8arrays": "^5.1.0"
   }
 }

packages/blockstore-core/src/base.ts

@@ -1,32 +1,32 @@
-import type { Blockstore, Pair } from 'interface-blockstore'
-import type { AbortOptions, Await, AwaitIterable } from 'interface-store'
+import type { Blockstore, InputPair, Pair } from 'interface-blockstore'
+import type { AbortOptions, Await, AwaitGenerator, AwaitIterable } from 'interface-store'
 import type { CID } from 'multiformats/cid'
 
 export class BaseBlockstore implements Blockstore {
   has (key: CID, options?: AbortOptions): Await<boolean> {
     return Promise.reject(new Error('.has is not implemented'))
   }
 
-  put (key: CID, val: Uint8Array, options?: AbortOptions): Await<CID> {
+  put (key: CID, val: Uint8Array | AwaitIterable<Uint8Array>, options?: AbortOptions): Await<CID> {
     return Promise.reject(new Error('.put is not implemented'))
   }
 
-  async * putMany (source: AwaitIterable<Pair>, options?: AbortOptions): AwaitIterable<CID> {
-    for await (const { cid, block } of source) {
-      await this.put(cid, block, options)
+  async * putMany (source: AwaitIterable<InputPair>, options?: AbortOptions): AwaitGenerator<CID> {
+    for await (const { cid, bytes } of source) {
+      await this.put(cid, bytes, options)
       yield cid
     }
   }
 
-  get (key: CID, options?: AbortOptions): Await<Uint8Array> {
-    return Promise.reject(new Error('.get is not implemented'))
+  get (key: CID, options?: AbortOptions): AwaitGenerator<Uint8Array> {
+    throw new Error('.get is not implemented')
   }
 
-  async * getMany (source: AwaitIterable<CID>, options?: AbortOptions): AwaitIterable<Pair> {
+  async * getMany (source: AwaitIterable<CID>, options?: AbortOptions): AwaitGenerator<Pair> {
     for await (const key of source) {
       yield {
         cid: key,
-        block: await this.get(key, options)
+        bytes: this.get(key, options)
       }
     }
   }
@@ -35,7 +35,7 @@ export class BaseBlockstore implements Blockstore {
     return Promise.reject(new Error('.delete is not implemented'))
   }
 
-  async * deleteMany (source: AwaitIterable<CID>, options?: AbortOptions): AwaitIterable<CID> {
+  async * deleteMany (source: AwaitIterable<CID>, options?: AbortOptions): AwaitGenerator<CID> {
     for await (const key of source) {
       await this.delete(key, options)
       yield key
@@ -45,7 +45,7 @@ export class BaseBlockstore implements Blockstore {
   /**
    * Extending classes should override `query` or implement this method
    */
-  async * getAll (options?: AbortOptions): AwaitIterable<Pair> { // eslint-disable-line require-yield
+  async * getAll (options?: AbortOptions): AwaitGenerator<Pair> { // eslint-disable-line require-yield
     throw new Error('.getAll is not implemented')
   }
 }

packages/blockstore-core/src/black-hole.ts

@@ -1,16 +1,16 @@
 import { NotFoundError } from 'interface-store'
 import { BaseBlockstore } from './base.js'
 import type { Pair } from 'interface-blockstore'
-import type { AbortOptions, Await, AwaitIterable } from 'interface-store'
+import type { AbortOptions, Await, AwaitGenerator, AwaitIterable } from 'interface-store'
 import type { CID } from 'multiformats/cid'
 
 export class BlackHoleBlockstore extends BaseBlockstore {
-  put (key: CID, value: Uint8Array, options?: AbortOptions): Await<CID> {
+  put (key: CID, value: Uint8Array | AwaitIterable<Uint8Array>, options?: AbortOptions): Await<CID> {
     options?.signal?.throwIfAborted()
     return key
   }
 
-  get (key: CID, options?: AbortOptions): Await<Uint8Array> {
+  get (key: CID, options?: AbortOptions): AwaitGenerator<Uint8Array> {
     options?.signal?.throwIfAborted()
     throw new NotFoundError()
   }
@@ -25,7 +25,7 @@ export class BlackHoleBlockstore extends BaseBlockstore {
   }
 
   // eslint-disable-next-line require-yield
-  async * getAll (options?: AbortOptions): AwaitIterable<Pair> {
+  async * getAll (options?: AbortOptions): AwaitGenerator<Pair> {
     options?.signal?.throwIfAborted()
   }
 }

packages/blockstore-core/src/identity.ts

@@ -1,7 +1,7 @@
 import { NotFoundError } from 'interface-store'
 import { BaseBlockstore } from './base.js'
 import type { Blockstore, Pair } from 'interface-blockstore'
-import type { AbortOptions, Await, AwaitIterable } from 'interface-store'
+import type { AbortOptions, Await, AwaitGenerator, AwaitIterable } from 'interface-store'
 import type { CID } from 'multiformats/cid'
 
 // https://github.com/multiformats/multicodec/blob/d06fc6194710e8909bac64273c43f16b56ca4c34/table.csv#L2
@@ -16,7 +16,7 @@ export class IdentityBlockstore extends BaseBlockstore {
     this.child = child
   }
 
-  put (key: CID, block: Uint8Array, options?: AbortOptions): Await<CID> {
+  put (key: CID, block: Uint8Array | AwaitIterable<Uint8Array>, options?: AbortOptions): Await<CID> {
     if (key.multihash.code === IDENTITY_CODEC) {
       options?.signal?.throwIfAborted()
       return key
@@ -30,18 +30,19 @@ export class IdentityBlockstore extends BaseBlockstore {
     return this.child.put(key, block, options)
   }
 
-  get (key: CID, options?: AbortOptions): Await<Uint8Array> {
+  * get (key: CID, options?: AbortOptions): AwaitGenerator<Uint8Array> {
     if (key.multihash.code === IDENTITY_CODEC) {
       options?.signal?.throwIfAborted()
-      return key.multihash.digest
+      yield key.multihash.digest
+      return
     }
 
     if (this.child == null) {
       options?.signal?.throwIfAborted()
       throw new NotFoundError()
     }
 
-    return this.child.get(key, options)
+    yield * this.child.get(key, options)
   }
 
   has (key: CID, options?: AbortOptions): Await<boolean> {
@@ -69,12 +70,11 @@ export class IdentityBlockstore extends BaseBlockstore {
     }
   }
 
-  getAll (options?: AbortOptions): AwaitIterable<Pair> {
+  * getAll (options?: AbortOptions): AwaitGenerator<Pair> {
     if (this.child != null) {
-      return this.child.getAll(options)
+      yield * this.child.getAll(options)
     }
 
     options?.signal?.throwIfAborted()
-    return []
   }
 }

packages/blockstore-core/src/memory.ts

@@ -1,37 +1,65 @@
 import { NotFoundError } from 'interface-store'
+import all from 'it-all'
 import { base32 } from 'multiformats/bases/base32'
 import { CID } from 'multiformats/cid'
 import * as raw from 'multiformats/codecs/raw'
 import * as Digest from 'multiformats/hashes/digest'
 import { BaseBlockstore } from './base.js'
 import type { Pair } from 'interface-blockstore'
-import type { AbortOptions, Await, AwaitIterable } from 'interface-store'
+import type { AbortOptions, Await, AwaitGenerator, AwaitIterable } from 'interface-store'
+
+function isPromise <T> (p?: any): p is Promise<T> {
+  return typeof p?.then === 'function'
+}
 
 export class MemoryBlockstore extends BaseBlockstore {
-  private readonly data: Map<string, Uint8Array>
+  private readonly data: Map<string, Uint8Array[]>
 
   constructor () {
     super()
 
     this.data = new Map()
   }
 
-  put (key: CID, val: Uint8Array, options?: AbortOptions): Await<CID> {
+  put (key: CID, val: Uint8Array | AwaitIterable<Uint8Array>, options?: AbortOptions): Await<CID> {
+    options?.signal?.throwIfAborted()
+
+    let buf: Uint8Array[]
+
+    if (val instanceof Uint8Array) {
+      buf = [val]
+    } else {
+      const result = all(val)
+
+      if (isPromise<Uint8Array[]>(result)) {
+        return result.then(val => {
+          return this._put(key, val, options)
+        })
+      } else {
+        buf = result
+      }
+    }
+
+    return this._put(key, buf, options)
+  }
+
+  private _put (key: CID, val: Uint8Array[], options?: AbortOptions): Await<CID> {
     options?.signal?.throwIfAborted()
+
     this.data.set(base32.encode(key.multihash.bytes), val)
 
     return key
   }
 
-  get (key: CID, options?: AbortOptions): Await<Uint8Array> {
+  * get (key: CID, options?: AbortOptions): AwaitGenerator<Uint8Array> {
     options?.signal?.throwIfAborted()
     const buf = this.data.get(base32.encode(key.multihash.bytes))
 
     if (buf == null) {
       throw new NotFoundError()
     }
 
-    return buf
+    yield * buf
   }
 
   has (key: CID, options?: AbortOptions): Await<boolean> {
@@ -44,13 +72,15 @@ export class MemoryBlockstore extends BaseBlockstore {
     this.data.delete(base32.encode(key.multihash.bytes))
   }
 
-  async * getAll (options?: AbortOptions): AwaitIterable<Pair> {
+  * getAll (options?: AbortOptions): AwaitGenerator<Pair> {
     options?.signal?.throwIfAborted()
 
     for (const [key, value] of this.data.entries()) {
       yield {
         cid: CID.createV1(raw.code, Digest.decode(base32.decode(key))),
-        block: value
+        bytes: (async function * () {
+          yield * value
+        })()
       }
       options?.signal?.throwIfAborted()
     }

packages/blockstore-core/src/tiered.ts

@@ -3,8 +3,8 @@ import { NotFoundError } from 'interface-store'
 import filter from 'it-filter'
 import merge from 'it-merge'
 import { BaseBlockstore } from './base.js'
-import type { Blockstore, Pair } from 'interface-blockstore'
-import type { AbortOptions, AwaitIterable } from 'interface-store'
+import type { Blockstore, InputPair, Pair } from 'interface-blockstore'
+import type { AbortOptions, AwaitGenerator, AwaitIterable } from 'interface-store'
 import type { CID } from 'multiformats/cid'
 
 const log = logger('blockstore:core:tiered')
@@ -24,7 +24,7 @@ export class TieredBlockstore extends BaseBlockstore {
     this.stores = stores.slice()
   }
 
-  async put (key: CID, value: Uint8Array, options?: AbortOptions): Promise<CID> {
+  async put (key: CID, value: Uint8Array | AwaitIterable<Uint8Array>, options?: AbortOptions): Promise<CID> {
     await Promise.all(
       this.stores.map(async store => {
         await store.put(key, value, options)
@@ -34,16 +34,13 @@ export class TieredBlockstore extends BaseBlockstore {
     return key
   }
 
-  async get (key: CID, options?: AbortOptions): Promise<Uint8Array> {
+  async * get (key: CID, options?: AbortOptions): AwaitGenerator<Uint8Array> {
     let error: Error | undefined
 
     for (const store of this.stores) {
       try {
-        const res = await store.get(key, options)
-
-        if (res != null) {
-          return res
-        }
+        yield * store.get(key, options)
+        return
       } catch (err: any) {
         error = err
         log.error(err)
@@ -71,21 +68,21 @@ export class TieredBlockstore extends BaseBlockstore {
     )
   }
 
-  async * putMany (source: AwaitIterable<Pair>, options: AbortOptions = {}): AsyncIterable<CID> {
+  async * putMany (source: AwaitIterable<InputPair>, options: AbortOptions = {}): AwaitGenerator<CID> {
     for await (const pair of source) {
-      await this.put(pair.cid, pair.block, options)
+      await this.put(pair.cid, pair.bytes, options)
       yield pair.cid
     }
   }
 
-  async * deleteMany (source: AwaitIterable<CID>, options: AbortOptions = {}): AsyncIterable<CID> {
+  async * deleteMany (source: AwaitIterable<CID>, options: AbortOptions = {}): AwaitGenerator<CID> {
     for await (const cid of source) {
       await this.delete(cid, options)
       yield cid
     }
   }
 
-  async * getAll (options?: AbortOptions): AwaitIterable<Pair> {
+  async * getAll (options?: AbortOptions): AwaitGenerator<Pair> {
     // deduplicate yielded pairs
     const seen = new Set<string>()
 

packages/blockstore-core/test/identity.spec.ts

@@ -3,6 +3,7 @@
 import { expect } from 'aegir/chai'
 import all from 'it-all'
 import drain from 'it-drain'
+import toBuffer from 'it-to-buffer'
 import { CID } from 'multiformats/cid'
 import * as raw from 'multiformats/codecs/raw'
 import { identity } from 'multiformats/hashes/identity'
@@ -20,13 +21,13 @@ describe('identity', () => {
     child = new MemoryBlockstore()
   })
 
-  it('has an identity CID', () => {
+  it('has an identity CID', async () => {
     const block = Uint8Array.from([0, 1, 2, 3, 4])
     const multihash = identity.digest(block)
     const cid = CID.createV1(identity.code, multihash)
 
     expect(blockstore.has(cid)).to.be.true()
-    expect(blockstore.get(cid)).to.equalBytes(block)
+    expect(toBuffer(await all(blockstore.get(cid)))).to.equalBytes(block)
   })
 
   it('does not have a non-identity CID', async () => {
@@ -70,7 +71,7 @@ describe('identity', () => {
 
     await blockstore.put(cid, block)
     expect(child.has(cid)).to.be.true()
-    expect(child.get(cid)).to.equalBytes(block)
+    expect(toBuffer(await all(child.get(cid)))).to.equalBytes(block)
   })
 
   it('gets CIDs from child', async () => {
@@ -82,7 +83,7 @@ describe('identity', () => {
 
     blockstore = new IdentityBlockstore(child)
     expect(blockstore.has(cid)).to.be.true()
-    expect(blockstore.get(cid)).to.equalBytes(block)
+    expect(toBuffer(await all(blockstore.get(cid)))).to.equalBytes(block)
   })
 
   it('has CIDs from child', async () => {