feat: add batch piece deletion (#898)

Author: hugomrdias

examples/cli/src/commands/pieces-removal.ts

@@ -1,5 +1,5 @@
 import * as p from '@clack/prompts'
-import { schedulePieceDeletion } from '@filoz/synapse-core/sp'
+import { schedulePieceDeletions } from '@filoz/synapse-core/sp'
 import { getPdpDataSet } from '@filoz/synapse-core/warm-storage'
 import { type Command, command } from 'cleye'
 import { waitForTransactionReceipt } from 'viem/actions'
@@ -42,10 +42,10 @@ export const piecesRemoval: Command = command(
         : await selectPiece(client, dataSet, argv.flags)
 
       p.log.info(`Removing piece ${pieceId} from data set ${dataSetId}...`)
-      const result = await schedulePieceDeletion(client, {
+      const result = await schedulePieceDeletions(client, {
         dataSetId,
         clientDataSetId: dataSet.clientDataSetId,
-        pieceId,
+        pieceIds: [pieceId],
         serviceURL: dataSet.provider.pdp.serviceURL,
       })
 

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

@@ -179,6 +179,20 @@ export class DeletePieceError extends SynapseError {
   }
 }
 
+export class TooManyPiecesQueuedError extends SynapseError {
+  override name: 'TooManyPiecesQueuedError' = 'TooManyPiecesQueuedError'
+
+  constructor() {
+    super(`Too many pieces queued.`, {
+      details: `The data set already has ${SIZE_CONSTANTS.MAX_DELETE_PIECES_BATCH_SIZE} or more scheduled removals queued on-chain; retry after the next proving period flushes the queue.`,
+    })
+  }
+
+  static override is(value: unknown): value is TooManyPiecesQueuedError {
+    return isSynapseError(value) && value.name === 'TooManyPiecesQueuedError'
+  }
+}
+
 export class TerminateServiceError extends SynapseError {
   override name: 'TerminateServiceError' = 'TerminateServiceError'
 

packages/synapse-core/src/sp/schedule-piece-deletion.ts

@@ -1,14 +1,17 @@
-import { HttpError, type RequestJsonErrors, request } from 'iso-web/http'
+import { HttpError, type RequestErrors, request } from 'iso-web/http'
 import type { Account, Chain, Client, Hex, Transport } from 'viem'
-import { DeletePieceError } from '../errors/pdp.ts'
+import { DeletePieceError, TooManyPiecesQueuedError } from '../errors/pdp.ts'
+import { AtLeastOnePieceRequiredError, TooManyPiecesError } from '../errors/warm-storage.ts'
 import { signSchedulePieceRemovals } from '../typed-data/sign-schedule-piece-removals.ts'
-import { RETRY_CONSTANTS } from '../utils/constants.ts'
+import { RETRY_CONSTANTS, SIZE_CONSTANTS } from '../utils/constants.ts'
 
-export namespace deletePiece {
+const MAX_CURIO_PIECE_ID = (1n << 63n) - 1n
+
+export namespace deletePieces {
   export type OptionsType = {
     serviceURL: string
     dataSetId: bigint
-    pieceId: bigint
+    pieceIds: bigint[]
     extraData: Hex
     /** The number of retries. Defaults to 2. */
     retryCount?: number
@@ -18,47 +21,86 @@ export namespace deletePiece {
   export type OutputType = {
     hash: Hex
   }
-  export type ErrorType = DeletePieceError | RequestJsonErrors
+  export type ErrorType =
+    | AtLeastOnePieceRequiredError
+    | TooManyPiecesError
+    | RangeError
+    | DeletePieceError
+    | RequestErrors
 }
 
 /**
- * Delete a piece from a data set on the PDP API.
+ * Delete pieces from a data set on the PDP API in one transaction.
  *
  * DELETE /pdp/data-sets/{dataSetId}/pieces/{pieceId}
  *
- * @param options - {@link deletePiece.OptionsType}
- * @returns Hash of the delete operation {@link deletePiece.OutputType}
- * @throws Errors {@link deletePiece.ErrorType}
+ * Curio uses the first piece ID in the URL for backwards-compatible routing and
+ * the pieceIds request field as the authoritative list when it is non-empty.
+ *
+ * @param options - {@link deletePieces.OptionsType}
+ * @returns Hash of the delete operation {@link deletePieces.OutputType}
+ * @throws Errors {@link deletePieces.ErrorType}
  */
-export async function deletePiece(options: deletePiece.OptionsType): Promise<deletePiece.OutputType> {
-  const { serviceURL, dataSetId, pieceId, extraData } = options
-  const response = await request.json.delete<{ txHash: Hex }>(
-    new URL(`pdp/data-sets/${dataSetId}/pieces/${pieceId}`, serviceURL),
-    {
-      body: { extraData },
-      timeout: RETRY_CONSTANTS.TIMEOUT,
-      retry: {
-        retries: options.retryCount,
-        minTimeout: options.retryDelay ?? RETRY_CONSTANTS.RETRY_DELAY,
-        shouldRetry: (ctx) => HttpError.is(ctx.error) && ctx.error.code === 429,
-      },
-    }
-  )
+export async function deletePieces(options: deletePieces.OptionsType): Promise<deletePieces.OutputType> {
+  const { serviceURL, dataSetId, extraData } = options
+  const pieceIds = normalizeDeletePieceIds(options.pieceIds)
+
+  // Curio accepts uint64 JSON numbers. Construct the array from bigint decimal
+  // strings so IDs above Number.MAX_SAFE_INTEGER are not rounded by JSON.stringify.
+  const body = `{"extraData":${JSON.stringify(extraData)},"pieceIds":[${pieceIds.join(',')}]}`
+  const response = await request.delete(new URL(`pdp/data-sets/${dataSetId}/pieces/${pieceIds[0]}`, serviceURL), {
+    body,
+    headers: { 'content-type': 'application/json' },
+    timeout: RETRY_CONSTANTS.TIMEOUT,
+    retry: {
+      retries: options.retryCount,
+      minTimeout: options.retryDelay ?? RETRY_CONSTANTS.RETRY_DELAY,
+    },
+  })
 
   if (response.error) {
     if (HttpError.is(response.error)) {
+      if (response.error.code === 429) {
+        throw new TooManyPiecesQueuedError()
+      }
       throw new DeletePieceError(await response.error.response.text())
     }
     throw response.error
   }
 
-  return { hash: response.result.txHash }
+  const result = (await response.result.json()) as { txHash: Hex }
+  return { hash: result.txHash }
 }
 
-export namespace schedulePieceDeletion {
+/**
+ * Validate a delete-pieces batch before signing or sending it.
+ */
+export function validateDeletePiecesBatch(pieceCount: number): void {
+  if (!Number.isInteger(pieceCount) || pieceCount < 1) {
+    throw new AtLeastOnePieceRequiredError()
+  }
+  if (pieceCount > SIZE_CONSTANTS.MAX_DELETE_PIECES_BATCH_SIZE) {
+    throw new TooManyPiecesError(pieceCount, SIZE_CONSTANTS.MAX_DELETE_PIECES_BATCH_SIZE)
+  }
+}
+
+function normalizeDeletePieceIds(pieceIds: bigint[]): bigint[] {
+  const normalized = [...new Set(pieceIds)]
+  validateDeletePiecesBatch(normalized.length)
+
+  for (const pieceId of normalized) {
+    if (pieceId < 0n || pieceId > MAX_CURIO_PIECE_ID) {
+      throw new RangeError(`Piece ID ${pieceId} is outside Curio's supported range of 0 to ${MAX_CURIO_PIECE_ID}`)
+    }
+  }
+
+  return normalized
+}
+
+export namespace schedulePieceDeletions {
   export type OptionsType = {
-    /** The piece ID to delete. */
-    pieceId: bigint
+    /** The piece IDs to delete. Duplicate IDs are removed before signing. */
+    pieceIds: bigint[]
     /** The data set ID to delete the piece from. */
     dataSetId: bigint
     /** The client data set id (nonce) to use for the signature. Must be unique for each data set. */
@@ -70,23 +112,23 @@ export namespace schedulePieceDeletion {
     /** The delay with exponential backoff between retries in milliseconds. Defaults to {@link RETRY_CONSTANTS.RETRY_DELAY}. */
     retryDelay?: number
   }
-  export type OutputType = deletePiece.OutputType
-  export type ErrorType = deletePiece.ErrorType
+  export type OutputType = deletePieces.OutputType
+  export type ErrorType = deletePieces.ErrorType
 }
 
 /**
- * Schedule a piece deletion
+ * Schedule piece deletions in one transaction.
  *
  * Call the Service Provider API to schedule the piece deletion.
  *
  * @param client - The client to use to schedule the piece deletion.
- * @param options - {@link schedulePieceDeletion.OptionsType}
- * @returns schedule piece deletion operation hash {@link schedulePieceDeletion.OutputType}
- * @throws Errors {@link schedulePieceDeletion.ErrorType}
+ * @param options - {@link schedulePieceDeletions.OptionsType}
+ * @returns Schedule piece deletions operation hash {@link schedulePieceDeletions.OutputType}
+ * @throws Errors {@link schedulePieceDeletions.ErrorType}
  *
  * @example
  * ```ts
- * import { schedulePieceDeletion } from '@filoz/synapse-core/sp'
+ * import { schedulePieceDeletions } from '@filoz/synapse-core/sp'
  * import { createWalletClient, http } from 'viem'
  * import { privateKeyToAccount } from 'viem/accounts'
  * import { calibration } from '@filoz/synapse-core/chains'
@@ -98,8 +140,8 @@ export namespace schedulePieceDeletion {
  *   transport: http(),
  * })
  *
- * const result = await schedulePieceDeletion(client, {
- *   pieceId: 1n,
+ * const result = await schedulePieceDeletions(client, {
+ *   pieceIds: [1n, 2n],
  *   dataSetId: 1n,
  *   clientDataSetId: 1n,
  *   serviceURL: 'https://pdp.example.com',
@@ -108,19 +150,56 @@ export namespace schedulePieceDeletion {
  * console.log(result.hash)
  * ```
  */
-export async function schedulePieceDeletion(
+export async function schedulePieceDeletions(
   client: Client<Transport, Chain, Account>,
-  options: schedulePieceDeletion.OptionsType
-): Promise<schedulePieceDeletion.OutputType> {
-  return deletePiece({
+  options: schedulePieceDeletions.OptionsType
+): Promise<schedulePieceDeletions.OutputType> {
+  const pieceIds = normalizeDeletePieceIds(options.pieceIds)
+
+  return deletePieces({
     serviceURL: options.serviceURL,
     dataSetId: options.dataSetId,
-    pieceId: options.pieceId,
+    pieceIds,
     extraData: await signSchedulePieceRemovals(client, {
       clientDataSetId: options.clientDataSetId,
-      pieceIds: [options.pieceId],
+      pieceIds,
     }),
     retryCount: options.retryCount,
     retryDelay: options.retryDelay,
   })
 }
+
+export namespace deletePiece {
+  export type OptionsType = Omit<deletePieces.OptionsType, 'pieceIds'> & { pieceId: bigint }
+  export type OutputType = deletePieces.OutputType
+  export type ErrorType = deletePieces.ErrorType
+}
+
+/**
+ * Delete one piece from a data set on the PDP API.
+ *
+ * @deprecated Use {@link deletePieces} instead.
+ */
+export function deletePiece(options: deletePiece.OptionsType): Promise<deletePiece.OutputType> {
+  const { pieceId, ...rest } = options
+  return deletePieces({ ...rest, pieceIds: [pieceId] })
+}
+
+export namespace schedulePieceDeletion {
+  export type OptionsType = Omit<schedulePieceDeletions.OptionsType, 'pieceIds'> & { pieceId: bigint }
+  export type OutputType = schedulePieceDeletions.OutputType
+  export type ErrorType = schedulePieceDeletions.ErrorType
+}
+
+/**
+ * Schedule one piece deletion.
+ *
+ * @deprecated Use {@link schedulePieceDeletions} instead.
+ */
+export function schedulePieceDeletion(
+  client: Client<Transport, Chain, Account>,
+  options: schedulePieceDeletion.OptionsType
+): Promise<schedulePieceDeletion.OutputType> {
+  const { pieceId, ...rest } = options
+  return schedulePieceDeletions(client, { ...rest, pieceIds: [pieceId] })
+}

packages/synapse-core/src/utils/constants.ts

@@ -99,6 +99,14 @@ export const SIZE_CONSTANTS = {
    */
   MAX_ADD_PIECES_BATCH_SIZE: 40,
 
+  /**
+   * Maximum pieces per schedulePieceDeletions call accepted by the Curio PDP API.
+   *
+   * Curio also rejects requests (429) when the data set already has 200 or more
+   * removals queued on-chain; the queue only drains at the next proving period.
+   */
+  MAX_DELETE_PIECES_BATCH_SIZE: 35,
+
   /**
    * Bytes per leaf in the PDP merkle tree.
    * The FWSS contract converts leaf counts to bytes via `totalBytes = leafCount * BYTES_PER_LEAF`.