feat: add support to prepend pieces while buffering to aggregate

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

@@ -328,6 +328,7 @@ export interface AggregateConfig {
   maxAggregateSize: number
   minAggregateSize: number
   minUtilizationFactor: number
+  prependBufferedPieces?: BufferedPiece[]
 }
 
 // Enums

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

@@ -152,12 +152,13 @@ export async function handleBufferReducingWithoutAggregate({
  * Attempt to build an aggregate with buffered pieces within ranges.
  *
  * @param {BufferedPiece[]} bufferedPieces
- * @param {object} sizes
- * @param {number} sizes.maxAggregateSize
- * @param {number} sizes.minAggregateSize
- * @param {number} sizes.minUtilizationFactor
+ * @param {object} config
+ * @param {number} config.maxAggregateSize
+ * @param {number} config.minAggregateSize
+ * @param {number} config.minUtilizationFactor
+ * @param {BufferedPiece[]} [config.prependBufferedPieces]
  */
-export function aggregatePieces(bufferedPieces, sizes) {
+export function aggregatePieces(bufferedPieces, config) {
   // Guarantee buffered pieces total size is bigger than the minimum utilization
   const bufferUtilizationSize = bufferedPieces.reduce((total, p) => {
     const piece = Piece.fromLink(p.piece)
@@ -166,14 +167,14 @@ export function aggregatePieces(bufferedPieces, sizes) {
   }, 0n)
   if (
     bufferUtilizationSize <
-    sizes.maxAggregateSize / sizes.minUtilizationFactor
+    config.maxAggregateSize / config.minUtilizationFactor
   ) {
     return
   }
 
   // Create builder with maximum size and try to fill it up
   const builder = Aggregate.createBuilder({
-    size: Aggregate.Size.from(sizes.maxAggregateSize),
+    size: Aggregate.Size.from(config.maxAggregateSize),
   })
 
   // add pieces to an aggregate until there is no more space, or no more pieces
@@ -182,6 +183,16 @@ export function aggregatePieces(bufferedPieces, sizes) {
   /** @type {BufferedPiece[]} */
   const remainingBufferedPieces = []
 
+  // start by adding prepend buffered pieces if available
+  for (const bufferedPiece of (config.prependBufferedPieces || [])) {
+    const p = Piece.fromLink(bufferedPiece.piece)
+    if (builder.estimate(p).error) {
+      throw new Error('aggregate builder is not able to create aggregates with only prepend buffered pieces')
+    }
+    builder.write(p)
+    addedBufferedPieces.push(bufferedPiece)
+  }
+
   for (const bufferedPiece of bufferedPieces) {
     const p = Piece.fromLink(bufferedPiece.piece)
     if (builder.estimate(p).error) {
@@ -196,7 +207,7 @@ export function aggregatePieces(bufferedPieces, sizes) {
     BigInt(builder.limit) * BigInt(Index.EntrySize)
 
   // If not enough space return undefined
-  if (totalUsedSpace < BigInt(sizes.minAggregateSize)) {
+  if (totalUsedSpace < BigInt(config.minAggregateSize)) {
     return
   }
 

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

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

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

@@ -397,6 +397,77 @@ export const test = {
       message.minPieceInsertedAt
     )
   },
+  'handles buffer queue messages successfully to queue aggregate prepended with a buffer piece': async (
+    assert,
+    context
+  ) => {
+    const group = context.id.did()
+    const { buffers, blocks } = await getBuffers(2, group, {
+      length: 100,
+      size: 128,
+    })
+
+    const [cargo] = await randomCargo(1, 128)
+    /** @type {import('../../src/aggregator/api.js').BufferedPiece} */
+    const bufferedPiece = {
+      piece: cargo.link.link(),
+      policy: 0,
+      insertedAt: (new Date()).toISOString()
+    }
+
+    const totalPieces = buffers.reduce((acc, v) => {
+      acc += v.pieces.length
+      return acc
+    }, 0)
+
+    // 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)
+    }
+
+    // Handle messages
+    const handledMessageRes = await AggregatorEvents.handleBufferQueueMessage(
+      {
+        ...context,
+        config: {
+          minAggregateSize: 2 ** 19,
+          minUtilizationFactor: 10e5,
+          maxAggregateSize: 2 ** 35,
+          prependBufferedPieces: [bufferedPiece]
+        },
+      },
+      blocks.map((b) => ({
+        pieces: b.cid,
+        group,
+      }))
+    )
+    assert.ok(handledMessageRes.ok)
+    assert.equal(handledMessageRes.ok?.aggregatedPieces, totalPieces + 1)
+
+    // Validate queue and store
+    await pWaitFor(
+      () =>
+        context.queuedMessages.get('aggregateOfferQueue')?.length === 1
+    )
+
+    /** @type {AggregateOfferMessage} */
+    // @ts-expect-error cannot infer buffer message
+    const message = context.queuedMessages.get('aggregateOfferQueue')?.[0]
+    const bufferGet = await context.bufferStore.get(message.buffer)
+    assert.ok(bufferGet.ok)
+    assert.ok(bufferGet.ok?.block.equals(message.buffer))
+    assert.equal(bufferGet.ok?.buffer.group, group)
+    assert.ok(message.aggregate.equals(bufferGet.ok?.buffer.aggregate))
+    assert.equal(bufferGet.ok?.buffer.pieces.length, totalPieces + 1)
+
+    // prepended piece
+    assert.ok(bufferGet.ok?.buffer.pieces.find(p => p.piece.link().equals(bufferedPiece.piece.link())))
+    assert.ok(bufferGet.ok?.buffer.pieces[0].piece.link().equals(bufferedPiece.piece.link()))
+  },
   'handles buffer queue messages successfully to queue aggregate and remaining buffer':
     async (assert, context) => {
       const group = context.id.did()