diff --git a/cmd/deal/send-manual-pdp.go b/cmd/deal/send-manual-pdp.go
index d6d95e3d..23e6421a 100644
--- a/cmd/deal/send-manual-pdp.go
+++ b/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,
}
diff --git a/docs/en/cli-reference/deal/send-manual-pdp.md b/docs/en/cli-reference/deal/send-manual-pdp.md
index a7b557a6..9cd792ba 100644
--- a/docs/en/cli-reference/deal/send-manual-pdp.md
+++ b/docs/en/cli-reference/deal/send-manual-pdp.md
@@ -19,6 +19,7 @@ OPTIONS:
--provider value Storage provider f4/t4 address
--piece-cid value Piece CID (commp v1)
--piece-size value Padded piece size in bytes (default: 0)
+ --payload-size value 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. (default: 0)
--eth-rpc value FEVM JSON-RPC endpoint [$ETH_RPC_URL]
--source-url-base value HTTPS base where Curio fetches the piece (sourceUrl = /piece/) [$PDP_SOURCE_URL_BASE]
--record-keeper value FWSS contract address. Defaults to network FWSS from go-synapse. [$PDP_RECORD_KEEPER]
diff --git a/go.mod b/go.mod
index 7fee8bc4..6d464ae4 100644
--- a/go.mod
+++ b/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
diff --git a/go.sum b/go.sum
index e8da4d33..617062d3 100644
--- a/go.sum
+++ b/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=
diff --git a/service/dealpusher/pdp_api.go b/service/dealpusher/pdp_api.go
index 9ea001bd..10d5a672 100644
--- a/service/dealpusher/pdp_api.go
+++ b/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.
diff --git a/service/dealpusher/pdp_pull.go b/service/dealpusher/pdp_pull.go
index 105ef9fe..8be96715 100644
--- a/service/dealpusher/pdp_pull.go
+++ b/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) {
diff --git a/service/dealpusher/pdp_pull_test.go b/service/dealpusher/pdp_pull_test.go
new file mode 100644
index 00000000..87288cf8
--- /dev/null
+++ b/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")
+ }
+}
diff --git a/service/dealpusher/pdp_schedule.go b/service/dealpusher/pdp_schedule.go
index 62c92443..529f095f 100644
--- a/service/dealpusher/pdp_schedule.go
+++ b/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,
diff --git a/service/dealpusher/pdp_wiring_test.go b/service/dealpusher/pdp_wiring_test.go
index e6724947..f49d6915 100644
--- a/service/dealpusher/pdp_wiring_test.go
+++ b/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)