fix: try reducing batch size

Author: Alan Shaw

Diff for: filecoin/functions/metrics-aggregate-offer-and-accept-total.js

@@ -1,5 +1,4 @@
 import * as Sentry from '@sentry/serverless'
-import { toString } from 'uint8arrays/to-string'
 import { fromString } from 'uint8arrays/from-string'
 import * as DAGJson from '@ipld/dag-json'
 
@@ -64,5 +63,5 @@ export const consumer = Sentry.AWSLambda.wrapHandler(handler)
  */
 function parseKinesisEvent (event) {
   const batch = event.Records.map(r => fromString(r.kinesis.data, 'base64'))
-  return batch.map(b => DAGJson.parse(toString(b, 'utf8')))
+  return batch.map(b => DAGJson.decode(b))
 }

Diff for: filecoin/metrics.js

@@ -59,7 +59,6 @@ export async function updateAggregateOfferTotal (ucanInvocations, ctx) {
   console.log(`${workflowsWithAggregateOffers.size} aggregate offer workflows`)
 
   // From workflows that include aggregate offer receipts, try to get the block with Pieces included in Aggregate
-  /** @type {AggregateOfferGet[]} */
   const aggregateOfferGets = (await Promise.all(
     Array.from(workflowsWithAggregateOffers.entries()).map(async ([carCid, aggregateOfferInvocation]) => {
       console.log(`getting agent message for task: ${aggregateOfferInvocation.invocationCid}`)
@@ -75,8 +74,7 @@ export async function updateAggregateOfferTotal (ucanInvocations, ctx) {
       console.log('update aggregate/offer total for worflow', carCid, aggregateOfferInvocation.invocationCid, 'with pieces', aggregateOfferInvocation.capabilities.map(aggregateOfferCap => aggregateOfferCap.nb.pieces.toString()))
       return Promise.all(aggregateOfferInvocation.capabilities.map(aggregateOfferCap => getOfferInfoBlock(aggregateOfferCap, agentMessage.ok)))
     })
-  // @ts-expect-error error types
-  )).flatMap(get => get)
+  )).flat()
   const aggregateOfferGetError = aggregateOfferGets.find(get => get.error)
   if (aggregateOfferGetError) {
     throw aggregateOfferGetError.error

Diff for: stacks/filecoin-stack.js

@@ -328,6 +328,7 @@ export function FilecoinStack({ stack, app }) {
       cdk: {
         eventSource: {
           ...(getEventSourceConfig(stack)),
+          batchSize: 1,
           parallelizationFactor: 10
         }
       }