From 7751aafca1bc240523b55c229bba8e94071a61c0 Mon Sep 17 00:00:00 2001 From: Arkadiy Kukarkin Date: Wed, 29 Apr 2026 14:34:30 +0200 Subject: [PATCH] pdp: add /pdp/data-sets/create-and-add client Server.CreateDataSetAndAddPieces wraps Curio's atomic create+add endpoint: the SP submits a single addPieces(dataSetId=0, recordKeeper, ...) on chain from its own wallet, creating a new FWSS-listened data set and adding the supplied pieces in one tx. This is the on-chain phase of the FWSS-pull flow -- /pdp/piece/pull only does the SP-side piece transfer, the on-chain commit goes through this endpoint with the full abi.encode(bytes,bytes) of the EIP-712-signed CreateDataSet+AddPieces extraData. Mirrors synapse-sdk's createDataSetAndAddPiecesApiRequest. Status (and the resulting dataSetId, once mined) is observed via the existing GetDataSetCreationStatus / WaitForDataSetCreation pollers. --- pdp/server.go | 67 ++++++++++++++++++++++++++++++++ pdp/server_test.go | 97 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+) diff --git a/pdp/server.go b/pdp/server.go index d243d0e..a47074e 100644 --- a/pdp/server.go +++ b/pdp/server.go @@ -106,6 +106,73 @@ func (s *Server) CreateDataSet(ctx context.Context, recordKeeper string, extraDa }, nil } +// CreateDataSetAndAddPieces issues POST /pdp/data-sets/create-and-add. The +// SP submits a single addPieces(dataSetId=0, recordKeeper, ...) on chain +// from its own wallet, atomically creating a new data set with the given +// recordKeeper as listener and adding the supplied pieces in one tx. The +// pieces must already be local on the SP (e.g. via prior /pdp/piece/pull +// or /pdp/piece/uploads). extraData is the abi.encode(bytes,bytes) of the +// EIP-712-signed CreateDataSet and AddPieces blobs (see +// EncodeCreateDataSetAndAddPiecesExtraData). Status (and the eventual +// dataSetId) is observed through GetDataSetCreationStatus on the returned +// txHash. +func (s *Server) CreateDataSetAndAddPieces(ctx context.Context, recordKeeper string, pieceCIDs []cid.Cid, extraData string) (*CreateDataSetResponse, error) { + pieces := make([]PieceData, len(pieceCIDs)) + for i, c := range pieceCIDs { + cidStr := c.String() + pieces[i] = PieceData{ + PieceCID: cidStr, + SubPieces: []SubPieceData{ + {SubPieceCID: cidStr}, + }, + } + } + + reqBody := CreateAndAddRequest{ + RecordKeeper: recordKeeper, + Pieces: pieces, + ExtraData: extraData, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("failed to marshal create-and-add request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", s.baseURL+"/pdp/data-sets/create-and-add", bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("failed to create create-and-add request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := s.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("create-and-add request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusAccepted { + respBody, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(respBody)) + } + + location := resp.Header.Get("Location") + if location == "" { + return nil, fmt.Errorf("missing Location header") + } + + parts := strings.Split(location, "/") + txHash := parts[len(parts)-1] + if !strings.HasPrefix(txHash, "0x") { + return nil, fmt.Errorf("invalid txHash in Location header: %s", txHash) + } + + return &CreateDataSetResponse{ + TxHash: txHash, + StatusURL: s.baseURL + location, + }, nil +} + func (s *Server) GetDataSetCreationStatus(ctx context.Context, txHash string) (*DataSetCreationStatus, error) { req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/pdp/data-sets/created/"+txHash, nil) if err != nil { diff --git a/pdp/server_test.go b/pdp/server_test.go index 831afe3..cb13617 100644 --- a/pdp/server_test.go +++ b/pdp/server_test.go @@ -14,6 +14,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ipfs/go-cid" ) func testAuthHelper(t *testing.T) *AuthHelper { @@ -165,6 +166,102 @@ func TestServer_CreateDataSet(t *testing.T) { }) } +func TestServer_CreateDataSetAndAddPieces(t *testing.T) { + pieceCID := mustCID(t, "baga6ea4seaqao7s73y24kcutaosvacpdjgfe5pw76ooefnyqw4ynr3d2y6x2mpq") + recordKeeper := "0x02925630df557F957f70E112bA06e50965417CA0" + extraData := "0xdeadbeef" + + t.Run("successful creation", func(t *testing.T) { + expectedTxHash := "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef" + var seen CreateAndAddRequest + + server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("Expected POST, got %s", r.Method) + } + if r.URL.Path != "/pdp/data-sets/create-and-add" { + t.Errorf("Expected path /pdp/data-sets/create-and-add, got %s", r.URL.Path) + } + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &seen); err != nil { + t.Fatalf("decode body: %v", err) + } + w.Header().Set("Location", "/pdp/data-sets/created/"+expectedTxHash) + w.WriteHeader(http.StatusCreated) + })) + + result, err := server.CreateDataSetAndAddPieces( + context.Background(), + recordKeeper, + []cid.Cid{pieceCID}, + extraData, + ) + if err != nil { + t.Fatalf("CreateDataSetAndAddPieces failed: %v", err) + } + if result.TxHash != expectedTxHash { + t.Errorf("TxHash = %s, want %s", result.TxHash, expectedTxHash) + } + if seen.RecordKeeper != recordKeeper { + t.Errorf("RecordKeeper = %s, want %s", seen.RecordKeeper, recordKeeper) + } + if seen.ExtraData != extraData { + t.Errorf("ExtraData = %s, want %s", seen.ExtraData, extraData) + } + if len(seen.Pieces) != 1 { + t.Fatalf("len(Pieces) = %d, want 1", len(seen.Pieces)) + } + if seen.Pieces[0].PieceCID != pieceCID.String() { + t.Errorf("Pieces[0].PieceCID = %s, want %s", seen.Pieces[0].PieceCID, pieceCID.String()) + } + if len(seen.Pieces[0].SubPieces) != 1 || seen.Pieces[0].SubPieces[0].SubPieceCID != pieceCID.String() { + t.Errorf("Pieces[0].SubPieces shape mismatch: %+v", seen.Pieces[0].SubPieces) + } + }) + + t.Run("missing Location header", func(t *testing.T) { + server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusCreated) + })) + + _, err := server.CreateDataSetAndAddPieces( + context.Background(), + recordKeeper, + []cid.Cid{pieceCID}, + extraData, + ) + if err == nil { + t.Error("Expected error for missing Location header, got nil") + } + }) + + t.Run("server error is surfaced", func(t *testing.T) { + server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte("extraData validation failed")) + })) + + _, err := server.CreateDataSetAndAddPieces( + context.Background(), + recordKeeper, + []cid.Cid{pieceCID}, + extraData, + ) + if err == nil { + t.Error("Expected error for 400 response, got nil") + } + }) +} + +func mustCID(t *testing.T, s string) cid.Cid { + t.Helper() + c, err := cid.Decode(s) + if err != nil { + t.Fatalf("decode CID %q: %v", s, err) + } + return c +} + func TestServer_GetDataSetCreationStatus(t *testing.T) { t.Run("successful status check", func(t *testing.T) { txHash := "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"