Added a mock blockstore to compute CID of large data streams with no/tiny memory usage (#677)

Another production-readiness fix. This is particularly useful when
dealing with large files and a necessity in a large-scale production
service environment.

Prior to this PR, we used the In-Memory Blockstore to compute the CID,
which would temporarily store the entire data in memory until the the
CID is computed. (e.g. 10 people uploading 1GB at the same time would
require 10GB of memory)

Did a sanity test with a 5 GB data stream: original implementation would
consume gigabytes of memory during the computation of the CID; the new
implementation consumes zero/little additional memory (minus the process
itself).

Author: thehenrytsai

src/store/blockstore-mock.ts

@@ -0,0 +1,80 @@
+import { CID } from 'multiformats';
+import type { AbortOptions, AwaitIterable } from 'interface-store';
+import type { Blockstore, Pair } from 'interface-blockstore';
+
+/**
+ * Mock implementation for the Blockstore interface.
+ *
+ * WARNING!!! Purely to be used with `ipfs-unixfs-importer` to compute CID without needing consume any memory.
+ * This is particularly useful when dealing with large files and a necessity in a large-scale production service environment.
+ */
+export class BlockstoreMock implements Blockstore {
+
+  async open(): Promise<void> {
+  }
+
+  async close(): Promise<void> {
+  }
+
+  async put(key: CID, _val: Uint8Array, _options?: AbortOptions): Promise<CID> {
+    return key;
+  }
+
+  async get(_key: CID, _options?: AbortOptions): Promise<Uint8Array> {
+    return new Uint8Array();
+  }
+
+  async has(_key: CID, _options?: AbortOptions): Promise<boolean> {
+    return false;
+  }
+
+  async delete(_key: CID, _options?: AbortOptions): Promise<void> {
+  }
+
+  async isEmpty(_options?: AbortOptions): Promise<boolean> {
+    return true;
+  }
+
+  async * putMany(source: AwaitIterable<Pair>, options?: AbortOptions): AsyncIterable<CID> {
+    for await (const entry of source) {
+      await this.put(entry.cid, entry.block, options);
+
+      yield entry.cid;
+    }
+  }
+
+  async * getMany(source: AwaitIterable<CID>, options?: AbortOptions): AsyncIterable<Pair> {
+    for await (const key of source) {
+      yield {
+        cid   : key,
+        block : await this.get(key, options)
+      };
+    }
+  }
+
+  async * getAll(options?: AbortOptions): AsyncIterable<Pair> {
+    // @ts-expect-error keyEncoding is 'buffer' but types for db.iterator always return the key type as 'string'
+    const li: AsyncGenerator<[Uint8Array, Uint8Array]> = this.db.iterator({
+      keys        : true,
+      keyEncoding : 'buffer'
+    }, options);
+
+    for await (const [key, value] of li) {
+      yield { cid: CID.decode(key), block: value };
+    }
+  }
+
+  async * deleteMany(source: AwaitIterable<CID>, options?: AbortOptions): AsyncIterable<CID> {
+    for await (const key of source) {
+      await this.delete(key, options);
+
+      yield key;
+    }
+  }
+
+  /**
+   * deletes all entries
+   */
+  async clear(): Promise<void> {
+  }
+}

src/utils/cid.ts

@@ -2,9 +2,9 @@ import * as cbor from '@ipld/dag-cbor';
 
 import type { Readable } from 'readable-stream';
 
+import { BlockstoreMock } from '../store/blockstore-mock.js';
 import { CID } from 'multiformats/cid';
 import { importer } from 'ipfs-unixfs-importer';
-import { MemoryBlockstore } from 'blockstore-core';
 import { sha256 } from 'multiformats/hashes/sha2';
 import { DwnError, DwnErrorCode } from '../core/dwn-error.js';
 
@@ -77,7 +77,7 @@ export class Cid {
    * @returns V1 CID of the DAG comprised by chunking data into unixfs DAG-PB encoded blocks
    */
   public static async computeDagPbCidFromBytes(content: Uint8Array): Promise<string> {
-    const asyncDataBlocks = importer([{ content }], new MemoryBlockstore(), { cidVersion: 1 });
+    const asyncDataBlocks = importer([{ content }], new BlockstoreMock(), { cidVersion: 1 });
 
     // NOTE: the last block contains the root CID
     let block;
@@ -90,7 +90,7 @@ export class Cid {
    * @returns V1 CID of the DAG comprised by chunking data into unixfs DAG-PB encoded blocks
    */
   public static async computeDagPbCidFromStream(dataStream: Readable): Promise<string> {
-    const asyncDataBlocks = importer([{ content: dataStream }], new MemoryBlockstore(), { cidVersion: 1 });
+    const asyncDataBlocks = importer([{ content: dataStream }], new BlockstoreMock(), { cidVersion: 1 });
 
     // NOTE: the last block contains the root CID
     let block;

tests/store/blockstore-mock.spec.ts

@@ -0,0 +1,43 @@
+import chaiAsPromised from 'chai-as-promised';
+import chai, { expect } from 'chai';
+
+import { BlockstoreMock } from '../../src/store/blockstore-mock.js';
+import { DataStream } from '../../src/index.js';
+import { importer } from 'ipfs-unixfs-importer';
+import { MemoryBlockstore } from 'blockstore-core';
+import { TestDataGenerator } from '../utils/test-data-generator.js';
+
+chai.use(chaiAsPromised);
+
+describe('BlockstoreMock', () => {
+  it('should facilitate the same CID computation as other implementations', async () => {
+
+    let dataSizeInBytes = 10;
+
+    // iterate through order of magnitude in size until hitting 10MB
+    // to ensure that the same CID is computed for the same data with the MockBlockstore as with the MemoryBlockstore
+    while (dataSizeInBytes <= 10_000_000) {
+      const dataBytes = TestDataGenerator.randomBytes(dataSizeInBytes);
+      const dataStreamForMemoryBlockstore = DataStream.fromBytes(dataBytes);
+      const dataStreamForMockBlockstore = DataStream.fromBytes(dataBytes);
+
+      const asyncDataBlocksByMemoryBlockstore = importer([{ content: dataStreamForMemoryBlockstore }], new MemoryBlockstore(), { cidVersion: 1 });
+      const asyncDataBlocksByMockBlockstore = importer([{ content: dataStreamForMockBlockstore }], new BlockstoreMock(), { cidVersion: 1 });
+
+      // NOTE: the last block contains the root CID
+      let blockByMemoryBlockstore;
+      for await (blockByMemoryBlockstore of asyncDataBlocksByMemoryBlockstore) { ; }
+      const dataCidByMemoryBlockstore = blockByMemoryBlockstore ? blockByMemoryBlockstore.cid.toString() : '';
+
+      let blockByMockBlockstore;
+      for await (blockByMockBlockstore of asyncDataBlocksByMockBlockstore) { ; }
+      const dataCidByMockBlockstore = blockByMockBlockstore ? blockByMockBlockstore.cid.toString() : '';
+
+      expect(dataCidByMockBlockstore).to.exist;
+      expect(dataCidByMockBlockstore.length).to.be.greaterThan(0);
+      expect(dataCidByMockBlockstore).to.be.equal(dataCidByMemoryBlockstore);
+
+      dataSizeInBytes *= 10;
+    }
+  });
+});