From 462040385b6d4f20604f012c3d96a1328cca1061 Mon Sep 17 00:00:00 2001 From: Arkadiy Kukarkin Date: Fri, 1 May 2026 19:45:29 +0100 Subject: [PATCH 1/4] fix start epoch on manual sends (PDP) --- cmd/deal/send-manual-pdp.go | 10 +++++++++- service/dealpusher/pdp_schedule.go | 11 ++++++----- service/dealpusher/pdp_wiring_test.go | 6 ++++-- 3 files changed, 19 insertions(+), 8 deletions(-) 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/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) From 81f0adf928c2becbdc7a035b890cf1fa9cc33392 Mon Sep 17 00:00:00 2001 From: Arkadiy Kukarkin Date: Fri, 1 May 2026 19:50:08 +0100 Subject: [PATCH 2/4] fix PDP pull addPieces nonce + payload size; add tests --- service/dealpusher/pdp_api.go | 7 ++- service/dealpusher/pdp_pull.go | 51 ++++++++++-------- service/dealpusher/pdp_pull_test.go | 80 +++++++++++++++++++++++++++++ 3 files changed, 114 insertions(+), 24 deletions(-) create mode 100644 service/dealpusher/pdp_pull_test.go 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") + } +} From 7fa51be49531d9da1563bca6e96765d7c97674f3 Mon Sep 17 00:00:00 2001 From: Arkadiy Kukarkin Date: Thu, 6 Aug 2026 15:13:24 +0200 Subject: [PATCH 3/4] shovel: replace with fork tolerating null blocks (revert when upstreamed) --- go.mod | 3 +++ go.sum | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) 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= From acb6cd2e350f80adfeee15da370dc81524250e15 Mon Sep 17 00:00:00 2001 From: Arkadiy Kukarkin Date: Thu, 6 Aug 2026 16:05:00 +0200 Subject: [PATCH 4/4] docs: regenerate cli reference for --payload-size flag --- docs/en/cli-reference/deal/send-manual-pdp.md | 1 + 1 file changed, 1 insertion(+) 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]