PDP pull fixups: addPieces nonce/payload/metadata + shovel null blocks

cmd/deal/send-manual-pdp.go

@@ -44,6 +44,11 @@ into the existing one). Useful for e2e/diagnostic testing of the FWSS pull path.
 			Usage:    "Padded piece size in bytes",
 			Required: true,
 		},
+		&cli.Int64Flag{
+			Name:     "payload-size",
+			Usage:    "Real CAR file size in bytes (the data the SP will fetch). The CommPv2 piece CID encodes this size; SP zero-fills locally up to PieceSize when computing CommP.",
+			Required: true,
+		},
 		&cli.StringFlag{
 			Name:     "eth-rpc",
 			Usage:    "FEVM JSON-RPC endpoint",
@@ -108,6 +113,7 @@ into the existing one). Useful for e2e/diagnostic testing of the FWSS pull path.
 			return errors.Wrap(err, "invalid piece CID")
 		}
 		pieceSize := c.Int64("piece-size")
+		payloadSize := c.Int64("payload-size")
 
 		cfg := dealpusher.PDPSchedulingConfig{
 			BatchSize:            1,
@@ -120,7 +126,7 @@ into the existing one). Useful for e2e/diagnostic testing of the FWSS pull path.
 			c.Context,
 			evmSigner,
 			c.String("provider"),
-			[]dealpusher.PDPPieceInput{{PieceCID: pieceCID, PieceSize: pieceSize}},
+			[]dealpusher.PDPPieceInput{{PieceCID: pieceCID, PieceSize: pieceSize, PayloadSize: payloadSize}},
 			cfg,
 		)
 		if err != nil {
@@ -135,6 +141,8 @@ into the existing one). Useful for e2e/diagnostic testing of the FWSS pull path.
 			Provider:   c.String("provider"),
 			PieceCID:   model.CID(pieceCID),
 			PieceSize:  pieceSize,
+			StartEpoch: dealpusher.PDPDealEpochSentinel,
+			EndEpoch:   dealpusher.PDPDealEpochSentinel,
 			WalletID:   &walletObj.ID,
 			ProofSetID: &dataSetIDCopy,
 		}

go.mod

@@ -441,3 +441,6 @@ require (
 replace github.com/filecoin-project/lassie => github.com/parkan/lassie v0.0.0-20251028120409-065e9fd563ad
 
 replace github.com/rclone/rclone => github.com/parkan/rclone v0.0.0-20260317152005-3777b647816d
+
+// fork tolerates null blocks in shovel's indexer; revert when upstreamed
+replace github.com/indexsupply/shovel => github.com/parkan/shovel v0.1.9-0.20260501183017-e5c6add716cf

go.sum

@@ -673,8 +673,6 @@ github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
 github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
-github.com/indexsupply/shovel v0.1.9-0.20260111041930-aea8d42c335c h1:w/28kFk3BSpsdRbTyAWJAnCt224qASF+FK92nlq/OBY=
-github.com/indexsupply/shovel v0.1.9-0.20260111041930-aea8d42c335c/go.mod h1:2fbvQP5CghUhrRn/XEQtwHXt51oaefJc0xqbAH+TTBE=
 github.com/internxt/rclone-adapter v0.0.0-20260220172730-613f4cc8b8fd h1:dSIuz2mpJAPQfhHYtG57D0qwSkgC/vQ69gHfeyQ4kxA=
 github.com/internxt/rclone-adapter v0.0.0-20260220172730-613f4cc8b8fd/go.mod h1:vdPya4AIcDjvng4ViaAzqjegJf0VHYpYHQguFx5xBp0=
 github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs=
@@ -1081,6 +1079,8 @@ github.com/parkan/lassie v0.0.0-20251028120409-065e9fd563ad h1:q56OIYelfaF5vvhyj
 github.com/parkan/lassie v0.0.0-20251028120409-065e9fd563ad/go.mod h1:P+HSZ4olI9imIXijFMLATQxfxu6erhWTtihKarK5tbs=
 github.com/parkan/rclone v0.0.0-20260317152005-3777b647816d h1:/PsMYjkePqFFCS8itIvzEMzC2MhTSssz9x+X8LJd6rc=
 github.com/parkan/rclone v0.0.0-20260317152005-3777b647816d/go.mod h1:0yvNwdX/c+ZOHm275UV7IpknJiWe/E/5U0uChfkXMGs=
+github.com/parkan/shovel v0.1.9-0.20260501183017-e5c6add716cf h1:ovBnvAdiiACV1EVRUrntNb4BxFiFOcOcYIIkYeyEJ3w=
+github.com/parkan/shovel v0.1.9-0.20260501183017-e5c6add716cf/go.mod h1:2fbvQP5CghUhrRn/XEQtwHXt51oaefJc0xqbAH+TTBE=
 github.com/parnurzeal/gorequest v0.2.16 h1:T/5x+/4BT+nj+3eSknXmCTnEVGSzFzPGdpqmUVVZXHQ=
 github.com/parnurzeal/gorequest v0.2.16/go.mod h1:3Kh2QUMJoqw3icWAecsyzkpY7UzRfDhbRdTjtNwNiUE=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=

service/dealpusher/pdp_api.go

@@ -39,10 +39,9 @@ func (c PDPSchedulingConfig) Validate() error {
 // PDPPieceInput names a piece the scheduler wants pushed to the SP. The
 // implementation constructs the SP-side source URL.
 type PDPPieceInput struct {
-	PieceCID cid.Cid
-	// PieceSize is the padded piece size, needed for CommPv2 conversion
-	// before signing.
-	PieceSize int64
+	PieceCID    cid.Cid
+	PieceSize   int64 // padded
+	PayloadSize int64 // real CAR bytes; encoded into the CommPv2 CID, fetched over HTTP
 }
 
 // PDPPullResult reports the outcome of a /pdp/piece/pull batch.

service/dealpusher/pdp_pull.go

@@ -163,21 +163,9 @@ func (o *OnChainPDP) PullPiecesToFWSS(
 		return PDPPullResult{}, errors.Wrap(err, "SP service-URL lookup")
 	}
 
-	// Convert v1 piece CIDs to CommPv2 (what FWSS / Curio expect).
-	pieceCIDsV2 := make([]cid.Cid, len(pieces))
-	pullInputs := make([]pdp.PullPieceInput, len(pieces))
-	for i, p := range pieces {
-		// FR32 padding: padded = raw * 128/127, raw = padded * 127/128.
-		payloadSize := uint64(p.PieceSize) * 127 / 128
-		v2, err := commcid.PieceCidV2FromV1(p.PieceCID, payloadSize)
-		if err != nil {
-			return PDPPullResult{}, errors.Wrapf(err, "convert piece %s to CommPv2", p.PieceCID)
-		}
-		pieceCIDsV2[i] = v2
-		pullInputs[i] = pdp.PullPieceInput{
-			PieceCID:  v2.String(),
-			SourceURL: o.sourceURLBase + "/piece/" + v2.String(),
-		}
+	pieceCIDsV2, pullInputs, err := buildPullInputs(pieces, o.sourceURLBase)
+	if err != nil {
+		return PDPPullResult{}, err
 	}
 
 	existing, err := o.findProofSetWithRoom(ctx, clientAddrStr, provider, cfg.MaxPiecesPerProofSet)
@@ -339,6 +327,26 @@ func waitForPullComplete(ctx context.Context, pdpServer *pdp.Server, opts pdp.Pu
 	}
 }
 
+func buildPullInputs(pieces []PDPPieceInput, sourceURLBase string) ([]cid.Cid, []pdp.PullPieceInput, error) {
+	cidsV2 := make([]cid.Cid, len(pieces))
+	inputs := make([]pdp.PullPieceInput, len(pieces))
+	for i, p := range pieces {
+		if p.PayloadSize <= 0 {
+			return nil, nil, fmt.Errorf("piece %s missing PayloadSize", p.PieceCID)
+		}
+		v2, err := commcid.PieceCidV2FromV1(p.PieceCID, uint64(p.PayloadSize))
+		if err != nil {
+			return nil, nil, errors.Wrapf(err, "convert piece %s to CommPv2", p.PieceCID)
+		}
+		cidsV2[i] = v2
+		inputs[i] = pdp.PullPieceInput{
+			PieceCID:  v2.String(),
+			SourceURL: sourceURLBase + "/piece/" + v2.String(),
+		}
+	}
+	return cidsV2, inputs, nil
+}
+
 func signCreateDataSetExtra(authHelper *pdp.AuthHelper, payer, payee common.Address, clientDataSetID *big.Int) (string, error) {
 	sig, err := authHelper.SignCreateDataSet(clientDataSetID, payee, nil)
 	if err != nil {
@@ -348,14 +356,17 @@ func signCreateDataSetExtra(authHelper *pdp.AuthHelper, payer, payee common.Addr
 }
 
 func signAddPiecesExtra(authHelper *pdp.AuthHelper, clientDataSetID *big.Int, pieceCIDsV2 []cid.Cid) (string, error) {
-	// FWSS uses (payer, clientDataSetId) as the cross-tx replay key, not
-	// the addPieces in-extraData nonce; zero is fine here.
-	nonce := big.NewInt(0)
-	sig, err := authHelper.SignAddPieces(clientDataSetID, nonce, pieceCIDsV2, nil)
+	// fresh random; FWSS uses clientNonces[payer][nonce] as addPieces replay key
+	nonce := randomClientDataSetID()
+	// FWSS validates metadataKeys.length == pieces.length and rejects with
+	// MetadataArrayCountMismatch otherwise. Pass an empty inner slice per
+	// piece (no metadata key/value pairs, but the outer dimension must match).
+	metadata := make([][]pdp.MetadataEntry, len(pieceCIDsV2))
+	sig, err := authHelper.SignAddPieces(clientDataSetID, nonce, pieceCIDsV2, metadata)
 	if err != nil {
 		return "", errors.Wrap(err, "sign AddPieces")
 	}
-	return pdp.EncodeAddPiecesExtraData(nonce, nil, sig.Signature)
+	return pdp.EncodeAddPiecesExtraData(nonce, metadata, sig.Signature)
 }
 
 func (o *OnChainPDP) findProofSetWithRoom(ctx context.Context, clientAddress, provider string, maxPieces int) (*model.PDPProofSet, error) {

service/dealpusher/pdp_pull_test.go

@@ -0,0 +1,80 @@
+package dealpusher
+
+import (
+	"testing"
+
+	commcid "github.com/filecoin-project/go-fil-commcid"
+	"github.com/ipfs/go-cid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildPullInputs_PayloadSizeEncodedNotPaddedSize(t *testing.T) {
+	t.Parallel()
+
+	v1 := cid.Cid(calculateCommp(t, generateRandomBytes(1000), 4194304))
+	const carSize = int64(2_097_436)
+	const pieceSize = int64(4_194_304)
+
+	cidsV2, inputs, err := buildPullInputs(
+		[]PDPPieceInput{{PieceCID: v1, PieceSize: pieceSize, PayloadSize: carSize}},
+		"https://example.test",
+	)
+	require.NoError(t, err)
+	require.Len(t, cidsV2, 1)
+	require.Len(t, inputs, 1)
+
+	_, decoded, err := commcid.PieceCidV2ToDataCommitment(cidsV2[0])
+	require.NoError(t, err)
+	require.Equal(t, uint64(carSize), decoded, "V2 CID must encode PayloadSize, not PieceSize*127/128")
+	require.NotEqual(t, uint64(pieceSize)*127/128, decoded)
+
+	require.Equal(t, "https://example.test/piece/"+cidsV2[0].String(), inputs[0].SourceURL)
+	require.Equal(t, cidsV2[0].String(), inputs[0].PieceCID)
+}
+
+func TestBuildPullInputs_MultiPiecePerIndexEncoding(t *testing.T) {
+	t.Parallel()
+
+	v1a := cid.Cid(calculateCommp(t, generateRandomBytes(1000), 4194304))
+	v1b := cid.Cid(calculateCommp(t, generateRandomBytes(2000), 4194304))
+	v1c := cid.Cid(calculateCommp(t, generateRandomBytes(3000), 8388608))
+	require.NotEqual(t, v1a, v1b)
+	require.NotEqual(t, v1b, v1c)
+
+	in := []PDPPieceInput{
+		{PieceCID: v1a, PieceSize: 4194304, PayloadSize: 1_000_000},
+		{PieceCID: v1b, PieceSize: 4194304, PayloadSize: 2_000_000},
+		{PieceCID: v1c, PieceSize: 8388608, PayloadSize: 5_000_000},
+	}
+	cidsV2, inputs, err := buildPullInputs(in, "https://example.test")
+	require.NoError(t, err)
+	require.Len(t, cidsV2, len(in))
+	require.Len(t, inputs, len(in))
+
+	for i, p := range in {
+		v1Decoded, payload, err := commcid.PieceCidV1FromV2(cidsV2[i])
+		require.NoError(t, err)
+		require.Equal(t, uint64(p.PayloadSize), payload, "piece %d payloadSize", i)
+		require.Equal(t, p.PieceCID, v1Decoded, "piece %d v1 CID", i)
+		require.Equal(t, cidsV2[i].String(), inputs[i].PieceCID)
+		require.Equal(t, "https://example.test/piece/"+cidsV2[i].String(), inputs[i].SourceURL)
+	}
+
+	require.NotEqual(t, cidsV2[0], cidsV2[1])
+	require.NotEqual(t, cidsV2[1], cidsV2[2])
+}
+
+func TestBuildPullInputs_RejectsMissingPayloadSize(t *testing.T) {
+	t.Parallel()
+
+	v1 := cid.Cid(calculateCommp(t, generateRandomBytes(1000), 4194304))
+
+	for _, payload := range []int64{0, -1} {
+		_, _, err := buildPullInputs(
+			[]PDPPieceInput{{PieceCID: v1, PieceSize: 4194304, PayloadSize: payload}},
+			"https://example.test",
+		)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "missing PayloadSize")
+	}
+}

service/dealpusher/pdp_schedule.go

@@ -16,7 +16,7 @@ import (
 	"gorm.io/gorm"
 )
 
-const pdpDealEpochSentinel = int32(math.MaxInt32)
+const PDPDealEpochSentinel = int32(math.MaxInt32)
 
 func defaultPDPSchedulingConfig() PDPSchedulingConfig {
 	return PDPSchedulingConfig{
@@ -193,8 +193,9 @@ func (d *DealPusher) runPDPSchedule(ctx context.Context, schedule *model.Schedul
 		pieceInputs := make([]PDPPieceInput, len(cars))
 		for i, car := range cars {
 			pieceInputs[i] = PDPPieceInput{
-				PieceCID:  cid.Cid(car.PieceCID),
-				PieceSize: car.PieceSize,
+				PieceCID:    cid.Cid(car.PieceCID),
+				PieceSize:   car.PieceSize,
+				PayloadSize: car.FileSize,
 			}
 		}
 
@@ -212,8 +213,8 @@ func (d *DealPusher) runPDPSchedule(ctx context.Context, schedule *model.Schedul
 				Provider:   schedule.Provider,
 				PieceCID:   car.PieceCID,
 				PieceSize:  car.PieceSize,
-				StartEpoch: pdpDealEpochSentinel,
-				EndEpoch:   pdpDealEpochSentinel,
+				StartEpoch: PDPDealEpochSentinel,
+				EndEpoch:   PDPDealEpochSentinel,
 				Verified:   schedule.Verified,
 				ScheduleID: &schedule.ID,
 				ClientID:   clientID,

service/dealpusher/pdp_wiring_test.go

@@ -115,6 +115,7 @@ func TestDealPusher_RunSchedule_PDPPushesBatchAndCreatesDeals(t *testing.T) {
 			PreparationID: &prep.ID,
 			PieceCID:      pieceCID,
 			PieceSize:     1024,
+			FileSize:      900,
 			StoragePath:   "car-1",
 		}
 		require.NoError(t, db.Create(&car).Error)
@@ -146,14 +147,15 @@ func TestDealPusher_RunSchedule_PDPPushesBatchAndCreatesDeals(t *testing.T) {
 		require.Len(t, psm.calls[0].pieces, 1)
 		require.Equal(t, cid.Cid(pieceCID), psm.calls[0].pieces[0].PieceCID)
 		require.Equal(t, int64(1024), psm.calls[0].pieces[0].PieceSize)
+		require.Equal(t, int64(900), psm.calls[0].pieces[0].PayloadSize)
 
 		var deals []model.Deal
 		require.NoError(t, db.Where("schedule_id = ?", schedule.ID).Find(&deals).Error)
 		require.Len(t, deals, 1)
 		require.Equal(t, model.DealTypePDP, deals[0].DealType)
 		require.Equal(t, model.DealProposed, deals[0].State)
-		require.Equal(t, pdpDealEpochSentinel, deals[0].StartEpoch)
-		require.Equal(t, pdpDealEpochSentinel, deals[0].EndEpoch)
+		require.Equal(t, PDPDealEpochSentinel, deals[0].StartEpoch)
+		require.Equal(t, PDPDealEpochSentinel, deals[0].EndEpoch)
 		require.NotNil(t, deals[0].ProofSetID)
 		require.Equal(t, uint64(42), *deals[0].ProofSetID)
 		require.NotNil(t, deals[0].WalletID)