fix: pieces can be removed from a data-set

package.json

@@ -38,6 +38,10 @@
       "types": "./dist/core/payments/index.d.ts",
       "default": "./dist/core/payments/index.js"
     },
+    "./core/piece": {
+      "types": "./dist/core/piece/index.d.ts",
+      "default": "./dist/core/piece/index.js"
+    },
     "./core/synapse": {
       "types": "./dist/core/synapse/index.d.ts",
       "default": "./dist/core/synapse/index.js"

src/cli.ts

@@ -6,6 +6,7 @@ import { addCommand } from './commands/add.js'
 import { dataSetCommand } from './commands/data-set.js'
 import { importCommand } from './commands/import.js'
 import { paymentsCommand } from './commands/payments.js'
+import { rmCommand } from './commands/rm.js'
 import { serverCommand } from './commands/server.js'
 import { checkForUpdate, type UpdateCheckStatus } from './common/version-check.js'
 import { version as packageVersion } from './core/utils/version.js'
@@ -24,6 +25,7 @@ program.addCommand(paymentsCommand)
 program.addCommand(dataSetCommand)
 program.addCommand(importCommand)
 program.addCommand(addCommand)
+program.addCommand(rmCommand)
 
 // Default action - show help if no command specified
 program.action(() => {

src/commands/rm.ts

@@ -0,0 +1,19 @@
+import { Command } from 'commander'
+import { runRmPiece } from '../rm/index.js'
+import { addAuthOptions } from '../utils/cli-options.js'
+
+export const rmCommand = new Command('rm')
+  .description('Remove a Piece from a DataSet')
+  .option('--piece <cid>', 'Piece CID to remove')
+  .option('--data-set <id>', 'DataSet ID to remove the piece from')
+  .option('--wait-for-confirmation', 'Wait for transaction confirmation before exiting')
+  .action(async (options) => {
+    try {
+      await runRmPiece(options)
+    } catch {
+      // Error already displayed by clack UI in runRmPiece
+      process.exit(1)
+    }
+  })
+
+addAuthOptions(rmCommand)

src/core/data-set/get-data-set-pieces.ts

@@ -7,7 +7,15 @@
  */
 
 import { getSizeFromPieceCID } from '@filoz/synapse-core/piece'
-import { METADATA_KEYS, type StorageContext, type Synapse, WarmStorageService } from '@filoz/synapse-sdk'
+import {
+  type DataSetPieceData,
+  METADATA_KEYS,
+  PDPServer,
+  PDPVerifier,
+  type StorageContext,
+  type Synapse,
+  WarmStorageService,
+} from '@filoz/synapse-sdk'
 import { isStorageContextWithDataSetId } from './type-guards.js'
 import type {
   DataSetPiecesResult,
@@ -16,6 +24,7 @@
   PieceInfo,
   StorageContextWithDataSetId,
 } from './types.js'
+import { PieceStatus } from './types.js'
 
 /**
  * Get all pieces for a dataset from a StorageContext
@@ -59,13 +68,43 @@
   const pieces: PieceInfo[] = []
   const warnings: DataSetWarning[] = []
 
+  // call PDPVerifier.getScheduledRemovals to get the list of pieces that are scheduled for removal
+  let scheduledRemovals: number[] = []
+  let pdpServerPieces: DataSetPieceData[] = []
+  try {
+    const warmStorage = await WarmStorageService.create(synapse.getProvider(), synapse.getWarmStorageAddress())
+    const pdpVerifier = new PDPVerifier(synapse.getProvider(), warmStorage.getPDPVerifierAddress())
+    scheduledRemovals = await pdpVerifier.getScheduledRemovals(storageContext.dataSetId)
+    const providerInfo = await synapse.getProviderInfo(storageContext.provider.serviceProvider)
+    const pdpServer = new PDPServer(null, providerInfo.products?.PDP?.data?.serviceURL ?? '')
+    const dataSet = await pdpServer.getDataSet(storageContext.dataSetId)
+    pdpServerPieces = dataSet.pieces
+  } catch (error) {
+    logger?.warn({ error }, 'Failed to create WarmStorageService or PDPVerifier for scheduled removals')
+  }
+
   // Use the async generator to fetch all pieces
   try {
     const getPiecesOptions = { ...(signal && { signal }) }
+    const providerPiecesById = new Map(pdpServerPieces.map((piece) => [piece.pieceId, piece]))
     for await (const piece of storageContext.getPieces(getPiecesOptions)) {
       const pieceId = piece.pieceId
       const pieceCid = piece.pieceCid
-      const pieceInfo: PieceInfo = { pieceId, pieceCid: pieceCid.toString() }
+      const status = getPieceStatus(pieceId, scheduledRemovals, providerPiecesById)
+      const pieceInfo: PieceInfo = {
+        pieceId,
+        pieceCid: pieceCid.toString(),
+        status,
+      }
+      if (status === PieceStatus.ONCHAIN_ORPHANED) {
+        warnings.push({
+          code: 'ONCHAIN_ORPHANED',
+          message: 'Piece is on-chain but the provider does not report it',
+          context: { pieceId, pieceCid },
+        })
+      } else if (status === PieceStatus.ACTIVE) {
+        providerPiecesById.delete(pieceId)
+      }
 
       // Calculate piece size from CID
       try {
@@ -79,6 +118,20 @@
 
       pieces.push(pieceInfo)
     }
+    for (const piece of providerPiecesById.values()) {
+      // add the rest of the pieces to the pieces list
+      pieces.push({
+        pieceId: piece.pieceId,
+        pieceCid: piece.pieceCid.toString(),
+        status: PieceStatus.OFFCHAIN_ORPHANED,
+      })
+      warnings.push({
+        code: 'OFFCHAIN_ORPHANED',
+        message: 'Piece is reported by provider but not on-chain',
+        context: { pieceId: piece.pieceId, pieceCid: piece.pieceCid.toString() },
+      })
+    }
+    pieces.sort((a, b) => a.pieceId - b.pieceId)
   } catch (error) {
     // If getPieces fails completely, throw - this is a critical error
     logger?.error({ dataSetId: storageContext.dataSetId, error }, 'Failed to retrieve pieces from dataset')
@@ -106,6 +159,20 @@
   return result
 }
 
+function getPieceStatus(
+  pieceId: number,
+  scheduledRemovals: number[],
+  providerPiecesById: Map<DataSetPieceData['pieceId'], DataSetPieceData>
+): PieceStatus {
+  if (scheduledRemovals.includes(pieceId)) {
+    return PieceStatus.PENDING_REMOVAL
+  }
+  if (providerPiecesById.has(pieceId)) {
+    return PieceStatus.ACTIVE
+  }
+  return PieceStatus.ONCHAIN_ORPHANED
+}
+
 /**
  * Internal helper: Enrich pieces with metadata from WarmStorage
  *

src/core/data-set/get-detailed-data-set.ts

@@ -23,6 +23,7 @@ export async function getDetailedDataSet(
 
   const storageContext = await synapse.storage.createContext({
     dataSetId: dataSet.dataSetId,
+    providerId: dataSet.providerId,
   })
 
   const piecesResult = await getDataSetPieces(synapse, storageContext, {

src/core/data-set/types.ts

@@ -11,6 +11,23 @@
 import type { EnhancedDataSetInfo, ProviderInfo, StorageContext } from '@filoz/synapse-sdk'
 import type { Logger } from 'pino'
 
+/**
+ * Status of the piece, e.g. "pending removal", "active", "orphaned"
+ *
+ * - "pending-removal": the piece is scheduled for deletion, but still showing on chain
+ * - "active": the piece is active, onchain and known by the provider
+ * - "onchain-orphaned": the piece is not known by the provider, but still on chain
+ * - "offchain-orphaned": the piece is known by the provider, but not on chain
+ *
+ * The orphaned states should not happen, but have been observed and should be logged and displayed to the user.
+ */
+export enum PieceStatus {
+  ACTIVE = 'ACTIVE',
+  PENDING_REMOVAL = 'PENDING_REMOVAL',
+  ONCHAIN_ORPHANED = 'ONCHAIN_ORPHANED',
+  OFFCHAIN_ORPHANED = 'OFFCHAIN_ORPHANED',
+}
+
 /**
  * Information about a single piece in a dataset
  */
@@ -19,6 +36,7 @@ export interface PieceInfo {
   pieceId: number
   /** Piece Commitment (CommP) as string */
   pieceCid: string
+  status: PieceStatus
   /** Root IPFS CID (from metadata, if available) */
   rootIpfsCid?: string
   /** Piece size in bytes (if available) */

src/core/piece/index.ts

@@ -0,0 +1 @@
+export * from './remove-piece.js'