fix: configure max pieces (#1566)

In lambda we only have 15 minutes max to execute. Without significant
changes to the way the library works there is only so many hashes we can
perform.

This PR adds a new configuration parameter `maxAggregatePieces` which
serves to restrict the maximum number of hashes the aggregator will have
to perform to build an aggregate. This will be tuned to be as high as
possible to allow execution within the time bounds. The PR also adds
logging in order to determine this limit.

packages/filecoin-api/package.json

@@ -189,6 +189,7 @@
       "project": "./tsconfig.json"
     },
     "env": {
+      "browser": true,
       "mocha": true
     },
     "ignorePatterns": [

packages/filecoin-api/src/aggregator/api.ts

@@ -328,6 +328,13 @@ export interface AggregateConfig {
   minUtilizationFactor: number
   prependBufferedPieces?: BufferedPiece[]
   hasher?: DataSegmentAPI.SyncMultihashHasher<DataSegmentAPI.SHA256_CODE>
+  /**
+   * The maximum number of pieces per aggregate. If set, it takes precedence
+   * over `minAggregateSize` because it is typically a hard limit related to
+   * the number of hashes that can be performed within the maximum lambda
+   * execution time limit.
+   */
+  maxAggregatePieces?: number
 }
 
 // Enums

packages/filecoin-api/src/aggregator/buffer-reducing.js

@@ -1,6 +1,5 @@
 import { Aggregate, Piece, NODE_SIZE, Index } from '@web3-storage/data-segment'
 import { CBOR } from '@ucanto/core'
-
 import { UnexpectedState } from '../errors.js'
 
 /**
@@ -192,22 +191,33 @@ export function aggregatePieces(bufferedPieces, config) {
     addedBufferedPieces.push(bufferedPiece)
   }
 
-  for (const bufferedPiece of bufferedPieces) {
+  for (const [i, bufferedPiece] of bufferedPieces.entries()) {
     const p = Piece.fromLink(bufferedPiece.piece)
     if (builder.estimate(p).error) {
       remainingBufferedPieces.push(bufferedPiece)
       continue
     }
+    if (addedBufferedPieces.length === config.maxAggregatePieces) {
+      remainingBufferedPieces.push(...bufferedPieces.slice(i))
+      break
+    }
     builder.write(p)
     addedBufferedPieces.push(bufferedPiece)
   }
   const totalUsedSpace =
     builder.offset * BigInt(NODE_SIZE) +
     BigInt(builder.limit) * BigInt(Index.EntrySize)
 
+  console.log(`Used ${totalUsedSpace} bytes in ${addedBufferedPieces.length} pieces (min ${config.minAggregateSize} bytes)`)
+
   // If not enough space return undefined
   if (totalUsedSpace < BigInt(config.minAggregateSize)) {
-    return
+    // ...but only if not exceeded max aggregate pieces
+    if (addedBufferedPieces.length === config.maxAggregatePieces) {
+      console.warn(`Building aggregate: max allowed pieces reached (${config.maxAggregatePieces})`)
+    } else {
+      return console.log('Not enough data for aggregate.')
+    }
   }
 
   const aggregate = builder.build()

packages/filecoin-api/src/aggregator/events.js

@@ -117,6 +117,7 @@ export const handleBufferQueueMessage = async (context, records) => {
     minUtilizationFactor: context.config.minUtilizationFactor,
     prependBufferedPieces: context.config.prependBufferedPieces,
     hasher: context.config.hasher,
+    maxAggregatePieces: context.config.maxAggregatePieces,
   })
 
   // Store buffered pieces if not enough to do aggregate and re-queue them

packages/filecoin-api/test/events/aggregator.js

@@ -546,6 +546,83 @@ export const test = {
         totalPieces
       )
     },
+  'handles buffer queue messages successfully when max aggregate pieces is exceeded':
+    async (assert, context) => {
+      const group = context.id.did()
+      const totalBuffers = 2
+      const piecesPerBuffer = 16
+      const pieceSize = 66_666
+      const totalPieces = totalBuffers * piecesPerBuffer
+
+      const { buffers, blocks } = await getBuffers(totalBuffers, group, {
+        length: piecesPerBuffer,
+        size: pieceSize,
+      })
+
+      // Store buffers
+      for (let i = 0; i < blocks.length; i++) {
+        const putBufferRes = await context.bufferStore.put({
+          buffer: buffers[i],
+          block: blocks[i].cid,
+        })
+        assert.ok(putBufferRes.ok)
+      }
+
+      const config = {
+        minAggregateSize: 2 ** 21, // 2,097,152
+        maxAggregateSize: 2 ** 22, // 4,194,304
+        // 15 (padded) pieces of 66,666 bytes results in an aggregate of size
+        // 1,968,128 which is below the `minAggregateSize` of 2,097,152 bytes.
+        // i.e. it wouldn't normally result in an aggregate.
+        maxAggregatePieces: 15,
+        minUtilizationFactor: 10,
+      }
+
+      const handledMessageRes = await AggregatorEvents.handleBufferQueueMessage(
+        { ...context, config },
+        blocks.map((b) => ({
+          pieces: b.cid,
+          group,
+        }))
+      )
+      assert.ok(handledMessageRes.ok)
+      assert.equal(handledMessageRes.ok?.aggregatedPieces, config.maxAggregatePieces)
+
+      await pWaitFor(
+        () =>
+          // there should still be an item of remaining pieces in the queue
+          context.queuedMessages.get('bufferQueue')?.length === 1 &&
+          // there should be 1 new aggregate in the queue
+          context.queuedMessages.get('aggregateOfferQueue')?.length === 1
+      )
+      /** @type {AggregateOfferMessage} */
+      // @ts-expect-error cannot infer buffer message
+      const aggregateOfferMessage = context.queuedMessages.get(
+        'aggregateOfferQueue'
+      )?.[0]
+      /** @type {BufferMessage} */
+      // @ts-expect-error cannot infer buffer message
+      const bufferMessage = context.queuedMessages.get('bufferQueue')?.[0]
+
+      const aggregateBufferGet = await context.bufferStore.get(
+        aggregateOfferMessage.buffer
+      )
+      assert.ok(aggregateBufferGet.ok)
+      assert.equal(
+        aggregateBufferGet.ok?.buffer.pieces.length,
+        handledMessageRes.ok?.aggregatedPieces
+      )
+
+      const remainingBufferGet = await context.bufferStore.get(
+        bufferMessage.pieces
+      )
+      assert.ok(remainingBufferGet.ok)
+      assert.equal(
+        (aggregateBufferGet.ok?.buffer.pieces.length || 0) +
+          (remainingBufferGet.ok?.buffer.pieces.length || 0),
+        totalPieces
+      )
+    },
   'handles buffer queue message errors when fails to access buffer store':
     wichMockableContext(
       async (assert, context) => {