feat: aggregator keeping oldest piece ts (#1188)

Conveniently store the timestamp of the oldest piece in table, so that
when monitoring aggregates time to deal for alerts, we do not need to
read buffer, decode it, and find out it every time a CRON runs.

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

@@ -187,6 +187,10 @@ export interface AggregateRecord {
    * Insertion date ISO string.
    */
   insertedAt: string
+  /**
+   * ISO string date of oldest piece in the pipeline included into the aggregate.
+   */
+  minPieceInsertedAt: string
 }
 
 // TODO: probably group should also be key!
@@ -314,6 +318,10 @@ export interface AggregateOfferMessage {
    * Grouping information for submitted piece.
    */
   group: string
+  /**
+   * ISO string date of oldest piece in the pipeline included into the aggregate.
+   */
+  minPieceInsertedAt: string
 }
 
 export interface AggregateConfig {

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

@@ -50,6 +50,15 @@ export async function handleBufferReducingWithAggregate({
   )
   const aggregateBlock = await CBOR.write(aggregateReducedBuffer)
 
+  // Get timestamp of oldest piece in the pipeline included in the aggregate
+  const minPieceInsertedAtDate = new Date(
+    Math.min(
+      ...aggregateInfo.addedBufferedPieces.map((bf) =>
+        new Date(bf.insertedAt).getTime()
+      )
+    )
+  )
+
   // Store buffered pieces for aggregate
   const bufferStoreAggregatePut = await bufferStore.put({
     buffer: aggregateReducedBuffer,
@@ -65,6 +74,7 @@ export async function handleBufferReducingWithAggregate({
     buffer: aggregateBlock.cid,
     pieces: piecesBlock.cid,
     group,
+    minPieceInsertedAt: minPieceInsertedAtDate.toISOString(),
   })
   if (aggregateOfferQueueAdd.error) {
     return aggregateOfferQueueAdd

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

@@ -168,14 +168,15 @@ export const handleBufferQueueMessage = async (context, records) => {
  * @param {import('./api.js').AggregateOfferMessage} message
  */
 export const handleAggregateOfferMessage = async (context, message) => {
-  const { pieces, aggregate, buffer, group } = message
+  const { pieces, aggregate, buffer, group, minPieceInsertedAt } = 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,
+    minPieceInsertedAt,
     insertedAt: new Date().toISOString(),
   })
 

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

@@ -352,13 +352,26 @@ export const test = {
     /** @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)
+    // Validate min piece date
+    assert.ok(message.minPieceInsertedAt)
+
+    const minPieceInsertedAtDate = new Date(
+      Math.min(
+        ...(bufferGet.ok?.buffer.pieces?.map((bf) =>
+          new Date(bf.insertedAt).getTime()
+        ) || [])
+      )
+    )
+    assert.equal(
+      minPieceInsertedAtDate.toISOString(),
+      message.minPieceInsertedAt
+    )
   },
   'handles buffer queue messages successfully to queue aggregate and remaining buffer':
     async (assert, context) => {
@@ -562,6 +575,7 @@ export const test = {
       pieces: piecesBlock.cid,
       buffer: block.cid,
       group,
+      minPieceInsertedAt: new Date().toISOString(),
     }
 
     // Handle message
@@ -604,6 +618,7 @@ export const test = {
           buffer: block.cid,
           pieces: piecesBlock.cid,
           group,
+          minPieceInsertedAt: new Date().toISOString(),
         }
 
         // Handle message
@@ -652,6 +667,7 @@ export const test = {
       aggregate: aggregate.link,
       group,
       insertedAt: new Date().toISOString(),
+      minPieceInsertedAt: new Date().toISOString(),
     }
     const putAggregateRes = await context.aggregateStore.put(aggregateRecord)
     assert.ok(putAggregateRes.ok)
@@ -719,9 +735,10 @@ export const test = {
           pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
+          minPieceInsertedAt: new Date().toISOString(),
         }
         const putAggregateRes = await context.aggregateStore.put(
-          aggregateRecord
+          aggregateRecord,
         )
         assert.ok(putAggregateRes.ok)
 
@@ -776,6 +793,7 @@ export const test = {
           pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
+          minPieceInsertedAt: new Date().toISOString(),
         }
         const putAggregateRes = await context.aggregateStore.put(
           aggregateRecord
@@ -1149,6 +1167,7 @@ export const test = {
       pieces: blockPieces.cid,
       group,
       insertedAt: new Date().toISOString(),
+      minPieceInsertedAt: new Date().toISOString(),
     }
     const putAggregateRes = await context.aggregateStore.put(aggregateRecord)
     assert.ok(putAggregateRes.ok)
@@ -1203,6 +1222,7 @@ export const test = {
           pieces: piecesBlock.cid,
           group,
           insertedAt: new Date().toISOString(),
+          minPieceInsertedAt: new Date().toISOString(),
         }
         const putAggregateRes = await context.aggregateStore.put(
           aggregateRecord

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

@@ -255,6 +255,7 @@ export const test = {
       buffer: block.cid,
       group,
       insertedAt: new Date().toISOString(),
+      minPieceInsertedAt: new Date().toISOString(),
     })
     assert.ok(aggregatePutRes.ok)