fix: aggregate offer invocation cid wrong because not pieces cid bloc…

…k being used but buffer block

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

@@ -167,6 +167,10 @@ export interface AggregateRecord {
    * `bagy...aggregate` Piece CID of an aggregate
    */
   aggregate: PieceLink
+  /**
+   * `bafy...cbor` as CID of dag-cbor block with buffer of pieces in an aggregate.
+   */
+  buffer: Link
   /**
    * `bafy...cbor` as CID of dag-cbor block with list of pieces in an aggregate.
    */
@@ -298,6 +302,10 @@ export interface AggregateOfferMessage {
    * List of pieces in an aggregate.
    */
   pieces: Link
+  /**
+   * `bafy...cbor` as CID of dag-cbor block with buffer of pieces in an aggregate.
+   */
+  buffer: Link
   /**
    * Grouping information for submitted piece.
    */

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

@@ -45,6 +45,7 @@ export async function handleBufferReducingWithAggregate({
     pieces: aggregateInfo.addedBufferedPieces,
     group,
   }
+  const piecesBlock = await CBOR.write(aggregateInfo.addedBufferedPieces.map(bf => bf.piece))
   const aggregateBlock = await CBOR.write(aggregateReducedBuffer)
 
   // Store buffered pieces for aggregate
@@ -59,7 +60,8 @@ export async function handleBufferReducingWithAggregate({
   // Propagate message for aggregate offer queue
   const aggregateOfferQueueAdd = await aggregateOfferQueue.add({
     aggregate: aggregateInfo.aggregate.link,
-    pieces: aggregateBlock.cid,
+    buffer: aggregateBlock.cid,
+    pieces: piecesBlock.cid,
     group,
   })
   if (aggregateOfferQueueAdd.error) {

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

@@ -168,12 +168,13 @@ export const handleBufferQueueMessage = async (context, records) => {
  * @param {import('./api.js').AggregateOfferMessage} message
  */
 export const handleAggregateOfferMessage = async (context, message) => {
-  const { pieces, aggregate, group } = message
+  const { pieces, aggregate, buffer, group } = message
 
   // Store aggregate information into the store. Store events MAY be used to propagate aggregate over
   const putRes = await context.aggregateStore.put({
     pieces,
     aggregate,
+    buffer,
     group,
     insertedAt: new Date().toISOString(),
   })
@@ -196,7 +197,7 @@ export const handleAggregateInsertToPieceAcceptQueue = async (
   context,
   record
 ) => {
-  const bufferStoreRes = await context.bufferStore.get(record.pieces)
+  const bufferStoreRes = await context.bufferStore.get(record.buffer)
   if (bufferStoreRes.error) {
     return bufferStoreRes
   }
@@ -331,7 +332,7 @@ export const handleAggregateInsertToAggregateOffer = async (
   context,
   record
 ) => {
-  const bufferStoreRes = await context.bufferStore.get(record.pieces)
+  const bufferStoreRes = await context.bufferStore.get(record.buffer)
   if (bufferStoreRes.error) {
     return {
       error: bufferStoreRes.error,

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

@@ -77,6 +77,7 @@ export const pieceAccept = async ({ capability }, context) => {
     }
   }
 
+  // Get buffered pieces 
   const [{ aggregate, inclusion }] = getInclusionRes.ok
   const getAggregateRes = await context.aggregateStore.get({ aggregate })
   if (getAggregateRes.error) {

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

@@ -353,9 +353,9 @@ export const test = {
     // @ts-expect-error cannot infer buffer message
     const message = context.queuedMessages.get('aggregateOfferQueue')?.[0]
 
-    const bufferGet = await context.bufferStore.get(message.pieces)
+    const bufferGet = await context.bufferStore.get(message.buffer)
     assert.ok(bufferGet.ok)
-    assert.ok(bufferGet.ok?.block.equals(message.pieces))
+    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)
@@ -414,7 +414,7 @@ export const test = {
       const bufferMessage = context.queuedMessages.get('bufferQueue')?.[0]
 
       const aggregateBufferGet = await context.bufferStore.get(
-        aggregateOfferMessage.pieces
+        aggregateOfferMessage.buffer
       )
       assert.ok(aggregateBufferGet.ok)
       const remainingBufferGet = await context.bufferStore.get(
@@ -554,11 +554,13 @@ export const test = {
       group,
     }
     const block = await CBOR.write(buffer)
+    const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
 
     /** @type {AggregateOfferMessage} */
     const message = {
       aggregate: aggregate.link,
-      pieces: block.cid,
+      pieces: piecesBlock.cid,
+      buffer: block.cid,
       group,
     }
 
@@ -573,7 +575,8 @@ export const test = {
     })
     assert.ok(hasStoredAggregate.ok)
     assert.ok(hasStoredAggregate.ok?.aggregate.equals(aggregate.link))
-    assert.ok(hasStoredAggregate.ok?.pieces.equals(block.cid))
+    assert.ok(hasStoredAggregate.ok?.buffer.equals(block.cid))
+    assert.ok(hasStoredAggregate.ok?.pieces.equals(piecesBlock.cid))
     assert.equal(hasStoredAggregate.ok?.group, group)
     assert.ok(hasStoredAggregate.ok?.insertedAt)
   },
@@ -593,11 +596,13 @@ export const test = {
           group,
         }
         const block = await CBOR.write(buffer)
+        const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
 
         /** @type {AggregateOfferMessage} */
         const message = {
           aggregate: aggregate.link,
-          pieces: block.cid,
+          buffer: block.cid,
+          pieces: piecesBlock.cid,
           group,
         }
 
@@ -631,6 +636,7 @@ export const test = {
       group,
     }
     const block = await CBOR.write(buffer)
+    const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
 
     // Put buffer record
     const putBufferRes = await context.bufferStore.put({
@@ -641,7 +647,8 @@ export const test = {
 
     // Put aggregate record
     const aggregateRecord = {
-      pieces: block.cid,
+      buffer: block.cid,
+      pieces: piecesBlock.cid,
       aggregate: aggregate.link,
       group,
       insertedAt: new Date().toISOString(),
@@ -697,11 +704,13 @@ export const test = {
           group,
         }
         const block = await CBOR.write(buffer)
+        const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
 
         // Put aggregate record
         const aggregateRecord = {
-          pieces: block.cid,
+          buffer: block.cid,
           aggregate: aggregate.link,
+          pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
         }
@@ -745,6 +754,8 @@ export const test = {
           group,
         }
         const block = await CBOR.write(buffer)
+        const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
+
         // Put buffer record
         const putBufferRes = await context.bufferStore.put({
           buffer,
@@ -754,8 +765,9 @@ export const test = {
 
         // Put aggregate record
         const aggregateRecord = {
-          pieces: block.cid,
+          buffer: block.cid,
           aggregate: aggregate.link,
+          pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
         }
@@ -1114,8 +1126,9 @@ export const test = {
 
     // Put aggregate record
     const aggregateRecord = {
-      pieces: blockBuffer.cid,
+      buffer: blockBuffer.cid,
       aggregate: aggregate.link,
+      pieces: blockPieces.cid,
       group,
       insertedAt: new Date().toISOString(),
     }
@@ -1128,6 +1141,7 @@ export const test = {
         context,
         aggregateRecord
       )
+    console.log('xxx', handledAggregateInsertsRes.error)
     assert.ok(handledAggregateInsertsRes.ok)
 
     // Verify invocation
@@ -1163,11 +1177,13 @@ export const test = {
           group,
         }
         const blockBuffer = await CBOR.write(buffer)
+        const piecesBlock = await CBOR.write(pieces.map((p) => p.link))
 
         // Put aggregate record
         const aggregateRecord = {
-          pieces: blockBuffer.cid,
+          buffer: blockBuffer.cid,
           aggregate: aggregate.link,
+          pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
         }

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

@@ -20,6 +20,7 @@ import {
 /**
  * @typedef {import('@web3-storage/data-segment').PieceLink} PieceLink
  * @typedef {import('@ucanto/interface').Link} Link
+ * @typedef {import('../../src/aggregator/api.js').Buffer} Buffer
  * @typedef {import('../../src/aggregator/api.js').PieceRecord} PieceRecord
  * @typedef {import('../../src/aggregator/api.js').PieceRecordKey} PieceRecordKey
  * @typedef {import('../../src/aggregator/api.js').BufferRecord} BufferRecord
@@ -234,13 +235,24 @@ export const test = {
     const group = storefront.did()
     const { pieces, aggregate } = await randomAggregate(100, 128)
     const piece = pieces[0].link
+    /** @type {Buffer} */
+    const buffer = {
+      pieces: pieces.map((p) => ({
+        piece: p.link,
+        insertedAt: new Date().toISOString(),
+        policy: 0,
+      })),
+      group,
+    }
+    const block = await CBOR.write(buffer)
 
     // Store aggregate record into store
     const offer = pieces.map((p) => p.link)
     const piecesBlock = await CBOR.write(offer)
     const aggregatePutRes = await context.aggregateStore.put({
       aggregate: aggregate.link,
       pieces: piecesBlock.cid,
+      buffer: block.cid,
       group,
       insertedAt: new Date().toISOString(),
     })