feat(synapse-core): add piece batcher with message-size addPieces limiter

packages/synapse-core/src/errors/pdp.ts

@@ -1,8 +1,10 @@
 import type { Hash } from 'viem'
+import type { PieceCID } from '../piece/piece-cid.ts'
 import type { AddPiecesRejected } from '../sp/add-pieces.ts'
 import type { CreateDataSetRejected } from '../sp/create-dataset.ts'
 import { SIZE_CONSTANTS } from '../utils/constants.ts'
 import { decodePDPError } from '../utils/decode-pdp-errors.ts'
+import type { MetadataObject } from '../utils/metadata.ts'
 import { isSynapseError, SynapseError } from './base.ts'
 
 export class LocationHeaderError extends SynapseError {
@@ -133,6 +135,44 @@ export class AddPiecesError extends SynapseError {
   }
 }
 
+export class AddPiecesBatchTooLargeError extends SynapseError {
+  override name: 'AddPiecesBatchTooLargeError' = 'AddPiecesBatchTooLargeError'
+  readonly pieceCount: number
+
+  constructor(pieceCount: number) {
+    super(`Piece batch of ${pieceCount} does not fit in a single Filecoin message. Split into smaller batches.`)
+    this.pieceCount = pieceCount
+  }
+
+  static override is(value: unknown): value is AddPiecesBatchTooLargeError {
+    return isSynapseError(value) && value.name === 'AddPiecesBatchTooLargeError'
+  }
+}
+
+export class AddPiecesFlushError extends SynapseError {
+  override name: 'AddPiecesFlushError' = 'AddPiecesFlushError'
+  readonly pieceCid: PieceCID
+  readonly metadata?: MetadataObject
+  /** The window that failed. Same extraData / same tx attempt. */
+  readonly pieces: Array<{ pieceCid: PieceCID; metadata?: MetadataObject }>
+
+  constructor(options: {
+    pieceCid: PieceCID
+    metadata?: MetadataObject
+    pieces: Array<{ pieceCid: PieceCID; metadata?: MetadataObject }>
+    cause: Error
+  }) {
+    super('Failed to add pieces.', { cause: options.cause })
+    this.pieceCid = options.pieceCid
+    this.metadata = options.metadata
+    this.pieces = options.pieces
+  }
+
+  static override is(value: unknown): value is AddPiecesFlushError {
+    return isSynapseError(value) && value.name === 'AddPiecesFlushError'
+  }
+}
+
 export class WaitForAddPiecesError extends SynapseError {
   override name: 'WaitForAddPiecesError' = 'WaitForAddPiecesError'
 

packages/synapse-core/src/sp/add-pieces-fits.ts

@@ -0,0 +1,139 @@
+import { encodeAbiParameters, encodeFunctionData, type Hex, size, toHex, zeroAddress } from 'viem'
+import { pdpVerifierAbi } from '../abis/generated.ts'
+import { AddPiecesBatchTooLargeError, InvalidUploadSizeError } from '../errors/pdp.ts'
+import { AtLeastOnePieceRequiredError } from '../errors/warm-storage.ts'
+import type { PieceCID } from '../piece/piece-cid.ts'
+import { signAddPiecesAbiParameters } from '../typed-data/sign-add-pieces.ts'
+import { signCreateDataSetAbiParameters } from '../typed-data/sign-create-dataset.ts'
+import { signcreateDataSetAndAddPiecesAbiParameters } from '../typed-data/sign-create-dataset-add-pieces.ts'
+import { SIZE_CONSTANTS } from '../utils/constants.ts'
+import { datasetMetadataObjectToEntry, type MetadataObject, pieceMetadataObjectToEntry } from '../utils/metadata.ts'
+import type { PdpDataSet } from '../warm-storage/types.ts'
+
+/** Dummy secp256k1 signature used only to size extraData. */
+const DUMMY_SIGNATURE = `0x${'00'.repeat(65)}` as Hex
+
+export type LimiterPiece = {
+  pieceCid: PieceCID
+  metadata?: MetadataObject
+}
+
+export type LimiterOptions =
+  | {
+      kind: 'addPieces'
+      dataSet?: PdpDataSet
+      pieces: LimiterPiece[]
+    }
+  | {
+      kind: 'createDataSetAndAddPieces'
+      metadata?: MetadataObject
+      cdn?: boolean
+      pieces: LimiterPiece[]
+    }
+
+/** `true` if `pieces` still fit in one addPieces / createAndAdd operation. */
+export type Limiter = (options: LimiterOptions) => boolean
+
+export namespace addPiecesFits {
+  export type OptionsType = LimiterOptions
+  export type OutputType = boolean
+}
+
+/**
+ * Whether a candidate piece list fits in one addPieces / createAndAdd message.
+ *
+ * Uses estimated encoded-params size (PieceCID bytes + dummy extraData) against
+ * {@link SIZE_CONSTANTS.MAX_ADD_PIECES_MESSAGE_SIZE} (64 KiB message cap minus
+ * overhead). Empty `pieces` does not fit.
+ *
+ * @param options - {@link addPiecesFits.OptionsType}
+ * @returns Whether the pieces fit {@link addPiecesFits.OutputType}
+ *
+ * @example
+ * ```ts
+ * import { addPiecesFits } from '@filoz/synapse-core/sp'
+ *
+ * const fits = addPiecesFits({
+ *   kind: 'addPieces',
+ *   dataSet,
+ *   pieces: [{ pieceCid }],
+ * })
+ * ```
+ */
+export function addPiecesFits(options: addPiecesFits.OptionsType): addPiecesFits.OutputType {
+  if (options.pieces.length < 1) {
+    return false
+  }
+  return estimateAddPiecesCalldataSize(options) <= SIZE_CONSTANTS.MAX_ADD_PIECES_MESSAGE_SIZE
+}
+
+/**
+ * Throw if a PieceCID's encoded raw size is outside Curio's upload bounds
+ * ({@link SIZE_CONSTANTS.MIN_UPLOAD_SIZE}–{@link SIZE_CONSTANTS.MAX_UPLOAD_SIZE}).
+ *
+ * @throws {@link InvalidUploadSizeError}
+ */
+export function assertPieceCidSize(pieceCid: PieceCID): void {
+  const pieceSize = pieceCid.size
+  if (pieceSize < SIZE_CONSTANTS.MIN_UPLOAD_SIZE || pieceSize > SIZE_CONSTANTS.MAX_UPLOAD_SIZE) {
+    throw new InvalidUploadSizeError(pieceSize)
+  }
+}
+
+/**
+ * Throw if `pieces` is empty, a PieceCID is outside Curio's size bounds, or the
+ * list does not fit in one addPieces / createAndAdd message.
+ *
+ * @param options - {@link LimiterOptions}
+ * @throws {@link AtLeastOnePieceRequiredError} when `pieces` is empty
+ * @throws {@link InvalidUploadSizeError} when a PieceCID size is below {@link SIZE_CONSTANTS.MIN_UPLOAD_SIZE} or above {@link SIZE_CONSTANTS.MAX_UPLOAD_SIZE}
+ * @throws {@link AddPiecesBatchTooLargeError} when the estimated message exceeds {@link SIZE_CONSTANTS.MAX_ADD_PIECES_MESSAGE_SIZE}
+ */
+export function assertAddPiecesFit(options: LimiterOptions): void {
+  if (options.pieces.length < 1) {
+    throw new AtLeastOnePieceRequiredError()
+  }
+  for (const piece of options.pieces) {
+    assertPieceCidSize(piece.pieceCid)
+  }
+  if (!addPiecesFits(options)) {
+    throw new AddPiecesBatchTooLargeError(options.pieces.length)
+  }
+}
+
+/**
+ * Estimated on-chain calldata size in bytes for the given piece list.
+ */
+export function estimateAddPiecesCalldataSize(options: LimiterOptions): number {
+  const extraData = dummyExtraData(options)
+  const pieceData = options.pieces.map((piece) => ({ data: toHex(piece.pieceCid.bytes) }))
+  const calldata = encodeFunctionData({
+    abi: pdpVerifierAbi,
+    functionName: 'addPieces',
+    args: [0n, zeroAddress, pieceData, extraData],
+  })
+  return size(calldata)
+}
+
+function dummyExtraData(options: LimiterOptions): Hex {
+  const addPiecesExtraData = dummyAddPiecesExtraData(options.pieces)
+  if (options.kind === 'addPieces') {
+    return addPiecesExtraData
+  }
+  const createEntries = datasetMetadataObjectToEntry(options.metadata, { cdn: options.cdn ?? false })
+  const createExtraData = encodeAbiParameters(signCreateDataSetAbiParameters, [
+    zeroAddress,
+    0n,
+    createEntries.map((entry) => entry.key),
+    createEntries.map((entry) => entry.value),
+    DUMMY_SIGNATURE,
+  ])
+  return encodeAbiParameters(signcreateDataSetAndAddPiecesAbiParameters, [createExtraData, addPiecesExtraData])
+}
+
+function dummyAddPiecesExtraData(pieces: LimiterPiece[]): Hex {
+  const metadataKV = pieces.map((piece) => pieceMetadataObjectToEntry(piece.metadata))
+  const keys = metadataKV.map((entries) => entries.map((entry) => entry.key))
+  const values = metadataKV.map((entries) => entries.map((entry) => entry.value))
+  return encodeAbiParameters(signAddPiecesAbiParameters, [0n, keys, values, DUMMY_SIGNATURE])
+}

packages/synapse-core/src/sp/add-pieces.ts

@@ -3,13 +3,15 @@ import type { ToString } from 'multiformats'
 import { type Account, type Chain, type Client, type Hex, isHex, type Transport } from 'viem'
 import * as z from 'zod'
 import { AddPiecesError, LocationHeaderError } from '../errors/index.ts'
+import type { AddPiecesBatchTooLargeError, InvalidUploadSizeError } from '../errors/pdp.ts'
 import { WaitForAddPiecesError, WaitForAddPiecesRejectedError } from '../errors/pdp.ts'
-import { AtLeastOnePieceRequiredError, TooManyPiecesError } from '../errors/warm-storage.ts'
+import type { AtLeastOnePieceRequiredError } from '../errors/warm-storage.ts'
 import type { PieceCID } from '../piece/piece-cid.ts'
 import { signAddPieces } from '../typed-data/sign-add-pieces.ts'
-import { RETRY_CONSTANTS, SIZE_CONSTANTS } from '../utils/constants.ts'
+import { RETRY_CONSTANTS } from '../utils/constants.ts'
 import { type MetadataObject, pieceMetadataObjectToEntry } from '../utils/metadata.ts'
 import { zHex, zNumberToBigInt } from '../utils/schemas.ts'
+import { assertAddPiecesFit } from './add-pieces-fits.ts'
 
 export namespace addPiecesApiRequest {
   export type OptionsType = {
@@ -114,24 +116,12 @@ export namespace addPieces {
   }
 
   export type OutputType = addPiecesApiRequest.OutputType
-  export type ErrorType = addPiecesApiRequest.ErrorType | signAddPieces.ErrorType
-}
-
-/**
- * Validate the piece count for an addPieces (or createDataSetAndAddPieces) batch,
- * failing early instead of reverting on-chain.
- *
- * @param pieceCount - Number of pieces in the batch
- * @throws AtLeastOnePieceRequiredError when not a positive integer
- * @throws TooManyPiecesError when above {@link SIZE_CONSTANTS.MAX_ADD_PIECES_BATCH_SIZE}
- */
-export function validateAddPiecesBatch(pieceCount: number): void {
-  if (!Number.isInteger(pieceCount) || pieceCount < 1) {
-    throw new AtLeastOnePieceRequiredError()
-  }
-  if (pieceCount > SIZE_CONSTANTS.MAX_ADD_PIECES_BATCH_SIZE) {
-    throw new TooManyPiecesError(pieceCount, SIZE_CONSTANTS.MAX_ADD_PIECES_BATCH_SIZE)
-  }
+  export type ErrorType =
+    | addPiecesApiRequest.ErrorType
+    | signAddPieces.ErrorType
+    | AtLeastOnePieceRequiredError
+    | AddPiecesBatchTooLargeError
+    | InvalidUploadSizeError
 }
 
 /**
@@ -148,7 +138,7 @@ export async function addPieces(
   client: Client<Transport, Chain, Account>,
   options: addPieces.OptionsType
 ): Promise<addPieces.OutputType> {
-  validateAddPiecesBatch(options.pieces.length)
+  assertAddPiecesFit({ kind: 'addPieces', pieces: options.pieces })
   const extraData =
     options.extraData ??
     (await signAddPieces(client, {

packages/synapse-core/src/sp/create-dataset-add-pieces.ts

@@ -4,16 +4,20 @@ import { type Account, type Address, type Chain, type Client, type Hex, isHex, t
 import { asChain } from '../chains.ts'
 import { CreateDataSetError, LocationHeaderError } from '../errors/index.ts'
 import type {
+  AddPiecesBatchTooLargeError,
+  InvalidUploadSizeError,
   WaitForAddPiecesError,
   WaitForAddPiecesRejectedError,
   WaitForCreateDataSetError,
   WaitForCreateDataSetRejectedError,
 } from '../errors/pdp.ts'
+import type { AtLeastOnePieceRequiredError } from '../errors/warm-storage.ts'
 import type { PieceCID } from '../piece/piece-cid.ts'
 import { signCreateDataSetAndAddPieces } from '../typed-data/sign-create-dataset-add-pieces.ts'
 import { RETRY_CONSTANTS } from '../utils/constants.ts'
 import { datasetMetadataObjectToEntry, type MetadataObject, pieceMetadataObjectToEntry } from '../utils/metadata.ts'
-import { validateAddPiecesBatch, waitForAddPieces } from './add-pieces.ts'
+import { waitForAddPieces } from './add-pieces.ts'
+import { assertAddPiecesFit } from './add-pieces-fits.ts'
 import { waitForCreateDataSet } from './create-dataset.ts'
 
 export namespace createDataSetAndAddPiecesApiRequest {
@@ -129,7 +133,12 @@ export type CreateDataSetAndAddPiecesOptions = {
 export namespace createDataSetAndAddPieces {
   export type OptionsType = CreateDataSetAndAddPiecesOptions
   export type ReturnType = createDataSetAndAddPiecesApiRequest.OutputType
-  export type ErrorType = createDataSetAndAddPiecesApiRequest.ErrorType | signCreateDataSetAndAddPieces.ErrorType
+  export type ErrorType =
+    | createDataSetAndAddPiecesApiRequest.ErrorType
+    | signCreateDataSetAndAddPieces.ErrorType
+    | AtLeastOnePieceRequiredError
+    | AddPiecesBatchTooLargeError
+    | InvalidUploadSizeError
 }
 
 /**
@@ -144,7 +153,12 @@ export async function createDataSetAndAddPieces(
   client: Client<Transport, Chain, Account>,
   options: CreateDataSetAndAddPiecesOptions
 ): Promise<createDataSetAndAddPieces.ReturnType> {
-  validateAddPiecesBatch(options.pieces.length)
+  assertAddPiecesFit({
+    kind: 'createDataSetAndAddPieces',
+    metadata: options.metadata,
+    cdn: options.cdn,
+    pieces: options.pieces,
+  })
   const chain = asChain(client.chain)
   const extraData =
     options.extraData ??