fix: pieces can be removed from a data-set (#253)

* fix: pieces can be removed from a data-set

* fix: handle pieces pending removal

* fix: display piece onchain vs pdpserver status

* fix: rm cmd requires --piece and --data-set

* fix: fix default to orphaned piece status

* deps: upgrade to latest synapse

* fix: use new createStorageContext signature in rm

* test: fix mock merge issue

* chore: remove unnecessary as string casting

* fix: logger is no longer required for createStorageContext

* test: remove-piece is tested

* test: assert runRmPiece happy path

* chore: remove outdated comment

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

* Update src/rm/remove-piece.ts

* Update src/rm/remove-piece.ts

* Update src/rm/remove-piece.ts

* chore: update jsdoc comments in runRmPiece

* fix: removePiece requires storageContext

* fix: document more about piece orphans

* fix: export Warning type

---------

Co-authored-by: Copilot <[email protected]>

Author: SgtPooki

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"
@@ -109,8 +113,8 @@
   "homepage": "https://github.com/filecoin-project/filecoin-pin#readme",
   "dependencies": {
     "@clack/prompts": "^0.11.0",
-    "@filoz/synapse-core": "^0.1.3",
-    "@filoz/synapse-sdk": "^0.36.0",
+    "@filoz/synapse-core": "^0.1.4",
+    "@filoz/synapse-sdk": "^0.36.1",
     "@helia/unixfs": "^6.0.1",
     "@ipld/car": "^5.4.2",
     "commander": "^14.0.1",

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')
+  .requiredOption('--piece <cid>', 'Piece CID to remove')
+  .requiredOption('--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/calculate-actual-storage.ts

@@ -2,9 +2,9 @@ import type { Synapse } from '@filoz/synapse-sdk'
 import PQueue from 'p-queue'
 import type { Logger } from 'pino'
 import { createStorageContextFromDataSetId } from '../synapse/storage-context-helper.js'
-import type { ProgressEvent, ProgressEventHandler } from '../utils/types.js'
+import type { ProgressEvent, ProgressEventHandler, Warning } from '../utils/types.js'
 import { getDataSetPieces } from './get-data-set-pieces.js'
-import type { DataSetSummary, DataSetWarning } from './types.js'
+import type { DataSetSummary } from './types.js'
 
 export interface ActualStorageResult {
   /** Total storage in bytes across all active data sets */
@@ -18,7 +18,7 @@ export interface ActualStorageResult {
   /** Whether the calculation timed out */
   timedOut?: boolean
   /** Non-fatal warnings encountered during calculation */
-  warnings: DataSetWarning[]
+  warnings: Warning[]
 }
 
 export type ActualStorageProgressEvents = ProgressEvent<
@@ -101,7 +101,7 @@ export async function calculateActualStorage(
   const maxParallelPerProvider = Math.max(1, options?.maxParallelPerProvider ?? 10)
   const onProgress = options?.onProgress
 
-  const warnings: DataSetWarning[] = []
+  const warnings: Warning[] = []
   let totalBytes = 0n
   let pieceCount = 0
   let dataSetsProcessed = 0

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

@@ -7,15 +7,20 @@
  */
 
 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 { reconcilePieceStatus } from '../piece/piece-status.js'
+import type { Warning } from '../utils/types.js'
 import { isStorageContextWithDataSetId } from './type-guards.js'
-import type {
-  DataSetPiecesResult,
-  DataSetWarning,
-  GetDataSetPiecesOptions,
-  PieceInfo,
-  StorageContextWithDataSetId,
-} from './types.js'
+import type { DataSetPiecesResult, GetDataSetPiecesOptions, PieceInfo, StorageContextWithDataSetId } from './types.js'
+import { PieceStatus } from './types.js'
 
 /**
  * Get all pieces for a dataset from a StorageContext
@@ -57,15 +62,58 @@ export async function getDataSetPieces(
   }
 
   const pieces: PieceInfo[] = []
-  const warnings: DataSetWarning[] = []
+  const warnings: Warning[] = []
+
+  // call PDPVerifier.getScheduledRemovals to get the list of pieces that are scheduled for removal
+  let scheduledRemovals: number[] = []
+  let pdpServerPieces: DataSetPieceData[] | null = null
+  try {
+    const warmStorage = await WarmStorageService.create(synapse.getProvider(), synapse.getWarmStorageAddress())
+    const pdpVerifier = new PDPVerifier(synapse.getProvider(), warmStorage.getPDPVerifierAddress())
+    scheduledRemovals = await pdpVerifier.getScheduledRemovals(storageContext.dataSetId)
+    try {
+      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 fetch provider data for scheduled removals and orphan detection')
+      warnings.push({
+        code: 'PROVIDER_DATA_UNAVAILABLE',
+        message: 'Failed to fetch provider data; orphan detection disabled',
+        context: { dataSetId: storageContext.dataSetId, error: String(error) },
+      })
+    }
+  } catch (error) {
+    logger?.warn({ error }, 'Failed to get scheduled removals')
+    warnings.push({
+      code: 'SCHEDULED_REMOVALS_UNAVAILABLE',
+      message: 'Failed to get scheduled removals',
+      context: { dataSetId: storageContext.dataSetId, error: String(error) },
+    })
+  }
 
   // Use the async generator to fetch all pieces
   try {
     const getPiecesOptions = { ...(signal && { signal }) }
+    const providerPiecesById = pdpServerPieces ? new Map(pdpServerPieces.map((piece) => [piece.pieceId, piece])) : null
     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, warning } = reconcilePieceStatus({
+        pieceId,
+        pieceCid,
+        scheduledRemovals,
+        providerPiecesById,
+      })
+      const pieceInfo: PieceInfo = {
+        pieceId,
+        pieceCid: pieceCid.toString(),
+        status,
+      }
+      if (warning) {
+        warnings.push(warning)
+      }
 
       // Calculate piece size from CID
       try {
@@ -79,6 +127,24 @@ export async function getDataSetPieces(
 
       pieces.push(pieceInfo)
     }
+    if (providerPiecesById !== null) {
+      // reconcilePieceStatus removes provider matches as we stream on-chain pieces.
+      // Remaining entries are only reported by the provider, which are off-chain orphans.
+      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 (error instanceof Error && error.name === 'AbortError') {
       throw error
@@ -122,7 +188,7 @@ async function enrichPiecesWithMetadata(
   synapse: Synapse,
   storageContext: StorageContextWithDataSetId,
   pieces: PieceInfo[],
-  warnings: DataSetWarning[],
+  warnings: Warning[],
   logger?: GetDataSetPiecesOptions['logger']
 ): Promise<void> {
   const dataSetId = storageContext.dataSetId

src/core/data-set/types.ts

@@ -10,6 +10,24 @@
 
 import type { EnhancedDataSetInfo, ProviderInfo, StorageContext } from '@filoz/synapse-sdk'
 import type { Logger } from 'pino'
+import type { Warning } from '../utils/types.js'
+
+/**
+ * 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 +37,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) */
@@ -38,19 +57,7 @@ export interface DataSetPiecesResult {
   /** Total size of all pieces in bytes (sum of individual piece sizes) */
   totalSizeBytes?: bigint
   /** Non-fatal warnings encountered during retrieval */
-  warnings?: DataSetWarning[]
-}
-
-/**
- * Structured warning for non-fatal issues
- */
-export interface DataSetWarning {
-  /** Machine-readable warning code (e.g., 'METADATA_FETCH_FAILED') */
-  code: string
-  /** Human-readable warning message */
-  message: string
-  /** Additional context data (e.g., { pieceId: 123, dataSetId: 456 }) */
-  context?: Record<string, unknown>
+  warnings?: Warning[]
 }
 
 /**

src/core/piece/index.ts

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

src/core/piece/piece-status.ts

@@ -0,0 +1,71 @@
+import type { DataSetPieceData } from '@filoz/synapse-sdk'
+import { PieceStatus } from '../data-set/types.js'
+import type { Warning } from '../utils/types.js'
+
+interface PieceStatusContext {
+  pieceId: number
+  pieceCid: unknown
+  /**
+   * List of pieceIds that are scheduled for removal.
+   *
+   * This list is obtained from the PDPVerifier.getScheduledRemovals() method.
+   */
+  scheduledRemovals: number[]
+  /**
+   * Map of provider-reported pieces keyed by pieceId.
+   *
+   * This map is mutated: when we confirm a piece is both on-chain and reported
+   * by the provider, we delete it so leftovers represent provider-only pieces.
+   */
+  providerPiecesById: Map<DataSetPieceData['pieceId'], DataSetPieceData> | null
+}
+
+interface PieceStatusResult {
+  status: PieceStatus
+  warning?: Warning
+}
+
+/**
+ * Reconcile a piece's status across the two data sources we have:
+ *
+ * - On-chain: StorageContext.getPieces() (source of truth for what the PDP verifier knows)
+ * - Provider-reported: PDPServer.getDataSet() (what the storage provider says it stores)
+ *
+ * https://github.com/filecoin-project/curio/issues/815 showed these can drift. This helper documents the rules we apply
+ * to flag mismatches without blocking the listing flow:
+ *
+ * 1. If PDPVerifier marked the piece for removal, treat as PENDING_REMOVAL.
+ * 2. If provider data is unavailable, assume ACTIVE (best effort).
+ * 3. If provider reports the piece, treat as ACTIVE and remove it from the map so
+ *    any leftover entries become OFFCHAIN_ORPHANED later.
+ * 4. Otherwise, the piece is on-chain but missing from the provider => ONCHAIN_ORPHANED.
+ *
+ * The optional warning conveys orphan cases to callers for user-facing messaging.
+ */
+export function reconcilePieceStatus(context: PieceStatusContext): PieceStatusResult {
+  const { pieceId, pieceCid, scheduledRemovals, providerPiecesById } = context
+
+  if (scheduledRemovals.includes(pieceId)) {
+    return { status: PieceStatus.PENDING_REMOVAL }
+  }
+
+  if (providerPiecesById === null) {
+    // No provider data to compare against; assume the on-chain view is accurate.
+    return { status: PieceStatus.ACTIVE }
+  }
+
+  if (providerPiecesById.has(pieceId)) {
+    // Provider matches on-chain; remove so leftovers can be flagged as off-chain orphans.
+    providerPiecesById.delete(pieceId)
+    return { status: PieceStatus.ACTIVE }
+  }
+
+  return {
+    status: PieceStatus.ONCHAIN_ORPHANED,
+    warning: {
+      code: 'ONCHAIN_ORPHANED',
+      message: 'Piece is on-chain but the provider does not report it',
+      context: { pieceId, pieceCid },
+    },
+  }
+}