From 868f19534ecdaf329a681fe7760bdec04303d0fc Mon Sep 17 00:00:00 2001 From: Lee <7932644+strahe@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:50:19 +0800 Subject: [PATCH 1/3] refactor(storage): minimize recovery state --- docs/GETTING_STARTED.md | 73 +++--- internal/apiaudit/apiaudit_test.go | 12 +- pdp/client_test.go | 22 +- pdp/coverage_test.go | 36 +-- pdp/dataset.go | 11 +- pdp/doc.go | 8 +- pdp/errors.go | 4 + pdp/pieces.go | 11 +- pdp/status.go | 49 +++- pdp/status_test.go | 106 ++++++-- storage/context.go | 4 +- storage/context_commit.go | 238 +++++++++++------- storage/context_commit_test.go | 165 ++++-------- storage/context_dataset.go | 97 +++---- storage/context_integration_test.go | 6 +- storage/context_test.go | 70 ++---- storage/context_upload.go | 7 +- storage/doc.go | 37 +-- storage/errors.go | 12 +- storage/lifecycle_json.go | 42 ---- storage/lifecycle_json_test.go | 131 ---------- storage/types.go | 30 ++- storage/upload_batcher_test.go | 3 +- .../delegated_storage_signer_test.go | 12 +- 24 files changed, 549 insertions(+), 637 deletions(-) diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index fb17175..13dc9eb 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -471,15 +471,19 @@ if err != nil { fmt.Println("dataset:", ref.DataSetID()) ``` -To create an empty dataset first, persist the submission if the process may -restart before confirmation. Creation is available only on `ProviderContext`. +To create an empty dataset first, save the status URL and client dataset ID if +the process may restart before confirmation. Creation is available only on +`ProviderContext`. ```go -var submitted storage.CreateDataSetSubmission +var statusURL string +var clientDataSetID types.BigInt created, err := providerCtx.CreateDataSet(ctx, &storage.CreateDataSetOptions{ OnSubmitted: func(s storage.CreateDataSetSubmission) { - submitted = s + statusURL = s.StatusURL + clientDataSetID = s.ClientDataSetID + // Save both values before this callback returns. }, }) if err != nil { @@ -493,7 +497,7 @@ same provider, then convert the returned reference without mutating that context: ```go -created, err := providerCtx.WaitForDataSetCreated(ctx, submitted) +created, err := providerCtx.WaitForDataSetCreated(ctx, statusURL, clientDataSetID) if err != nil { return err } @@ -518,22 +522,29 @@ them automatically. ### Recovering create-and-add and add-pieces submissions `CreateAndAddRequest.OnSubmitted` and `CommitRequest.OnSubmitted` receive an -independent copy of the complete, JSON-serializable `CommitSubmission` after -the provider handle has been validated and before confirmation begins. A -single-step call can therefore preserve its handle even when the later wait -fails: +independent `CommitSubmission` after the provider accepts the request and +before confirmation begins. This value contains runtime and diagnostic data; +it is not a persistence schema. Save only the recovery fields needed by the +operation. + +Create-and-add requires the status URL and client dataset ID: ```go -var submitted storage.CommitSubmission +var statusURL string +var clientDataSetID types.BigInt +var providerID types.BigInt result, err := providerCtx.CreateAndAdd(ctx, storage.CreateAndAddRequest{ Pieces: pieces, OnSubmitted: func(s storage.CommitSubmission) { - submitted = s // persist all fields here + statusURL = s.StatusURL + clientDataSetID = *s.ClientDataSetID + providerID = s.ProviderID + // Save these values before this callback returns. }, }) if err != nil { - if submitted.TransactionID == "" { + if statusURL == "" { return err } recoveryCtx, cancel := context.WithTimeout(context.Background(), 3 * time.Minute) @@ -541,13 +552,17 @@ if err != nil { fresh, openErr := client.Storage().NewProviderContext( recoveryCtx, - submitted.ProviderID, + providerID, storage.NewProviderContextOptions{}, ) if openErr != nil { return openErr } - result, err = fresh.WaitForCreateAndAdd(recoveryCtx, submitted) + result, err = fresh.WaitForCreateAndAdd( + recoveryCtx, + statusURL, + clientDataSetID, + ) } if err != nil { return err @@ -565,27 +580,25 @@ submitted, err := providerCtx.SubmitCreateAndAdd(ctx, storage.CreateAndAddReques if err != nil { return err } -// Persist submitted before waiting. -result, err := providerCtx.WaitForCreateAndAdd(ctx, *submitted) +// Persist submitted.StatusURL and *submitted.ClientDataSetID before waiting. +result, err := providerCtx.WaitForCreateAndAdd( + ctx, + submitted.StatusURL, + *submitted.ClientDataSetID, +) ``` For an existing dataset, use `DataSetContext.SubmitCommit`, -`GetCommitStatus`, and `WaitForCommit` in the same pattern. Use +`GetCommitStatus`, and `WaitForCommit` in the same pattern, persisting the +`DataSetRef` and `submitted.StatusURL`. Use `ProviderContext.GetCreateAndAddStatus` and `WaitForCreateAndAdd` for a new dataset. High-level upload recovery continues to use -`FailedAttempt.Submission`; `OnPiecesAdded` remains a transaction progress -event and still receives a transaction hash rather than a recovery handle. - -Migration from the previous context API: - -| Previous API | Updated API | -|---|---| -| `provider.Commit(req)` | `provider.CreateAndAdd(storage.CreateAndAddRequest{...})` | -| `provider.SubmitCommit(req)` | `provider.SubmitCreateAndAdd(storage.CreateAndAddRequest{...})` | -| `provider.GetCommitStatus(submission)` | `provider.GetCreateAndAddStatus(submission)` | -| `provider.WaitForCommit(submission)` | `provider.WaitForCreateAndAdd(submission)` | -| `CommitRequest.ClientDataSetID` | `CreateAndAddRequest.ClientDataSetID` | -| `CommitRequest.OnSubmitted: func(txHash string)` | `CreateAndAddRequest.OnSubmitted` and `CommitRequest.OnSubmitted`: `func(submission storage.CommitSubmission)` | +`FailedAttempt.Submission`: extract the same minimal fields rather than storing +the complete value. `OnPiecesAdded` remains a transaction progress event and +still receives a transaction hash rather than a recovery handle. + +Applications that need to map returned piece IDs to CIDs must persist that +business mapping with their original request. ## Discovery And Lifecycle diff --git a/internal/apiaudit/apiaudit_test.go b/internal/apiaudit/apiaudit_test.go index 213b49b..b67623b 100644 --- a/internal/apiaudit/apiaudit_test.go +++ b/internal/apiaudit/apiaudit_test.go @@ -254,17 +254,21 @@ var ( _ func(*storage.ProviderContext, context.Context, storage.CreateAndAddRequest) (*storage.CommitResult, error) = (*storage.ProviderContext).CreateAndAdd _ func(*storage.ProviderContext, context.Context, storage.CreateAndAddRequest) (*storage.CommitSubmission, error) = (*storage.ProviderContext).SubmitCreateAndAdd - _ func(*storage.ProviderContext, context.Context, storage.CommitSubmission) (*storage.CommitStatus, error) = (*storage.ProviderContext).GetCreateAndAddStatus - _ func(*storage.ProviderContext, context.Context, storage.CommitSubmission) (*storage.CommitResult, error) = (*storage.ProviderContext).WaitForCreateAndAdd + _ func(*storage.ProviderContext, context.Context, string, types.BigInt) (*storage.CreateDataSetResult, error) = (*storage.ProviderContext).WaitForDataSetCreated + _ func(*storage.ProviderContext, context.Context, string, types.BigInt) (*storage.CommitStatus, error) = (*storage.ProviderContext).GetCreateAndAddStatus + _ func(*storage.ProviderContext, context.Context, string, types.BigInt) (*storage.CommitResult, error) = (*storage.ProviderContext).WaitForCreateAndAdd _ func(*storage.DataSetContext, context.Context, storage.CommitRequest) (*storage.CommitResult, error) = (*storage.DataSetContext).Commit _ func(*storage.DataSetContext, context.Context, storage.CommitRequest) (*storage.CommitSubmission, error) = (*storage.DataSetContext).SubmitCommit - _ func(*storage.DataSetContext, context.Context, storage.CommitSubmission) (*storage.CommitStatus, error) = (*storage.DataSetContext).GetCommitStatus - _ func(*storage.DataSetContext, context.Context, storage.CommitSubmission) (*storage.CommitResult, error) = (*storage.DataSetContext).WaitForCommit + _ func(*storage.DataSetContext, context.Context, string) (*storage.CommitStatus, error) = (*storage.DataSetContext).GetCommitStatus + _ func(*storage.DataSetContext, context.Context, string) (*storage.CommitResult, error) = (*storage.DataSetContext).WaitForCommit _ func(*storage.ProviderContext, context.Context, io.Reader, *storage.ContextUploadOptions) (*storage.UploadResult, error) = (*storage.ProviderContext).Upload _ func(*storage.DataSetContext, context.Context, io.Reader, *storage.ContextUploadOptions) (*storage.UploadResult, error) = (*storage.DataSetContext).Upload _ storage.StorageContext = (*storage.ProviderContext)(nil) _ storage.StorageContext = (*storage.DataSetContext)(nil) + _ types.BigInt = storage.CreateDataSetSubmission{}.ClientDataSetID + _ types.BigInt = storage.CommitRejectedError{}.ProviderID + _ storage.CommitStatus = storage.CommitRejectedError{}.Status ) func keepStorageContext(ctx storage.StorageContext) storage.StorageContext { return ctx } diff --git a/pdp/client_test.go b/pdp/client_test.go index 7c28bc0..506267e 100644 --- a/pdp/client_test.go +++ b/pdp/client_test.go @@ -784,7 +784,7 @@ func TestWaitForDataSetCreated(t *testing.T) { } _, _ = fmt.Fprint(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000001","service":"svc","txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":42}`) })) - status, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/0x1", 10*time.Millisecond) + status, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/"+testTxOne, 10*time.Millisecond) if err != nil { t.Fatal(err) } @@ -799,7 +799,7 @@ func TestGetDataSetCreationStatus_Accepts202(t *testing.T) { w.WriteHeader(http.StatusAccepted) _, _ = fmt.Fprint(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000001","service":"svc","txStatus":"pending","dataSetCreated":false,"ok":null}`) })) - status, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/0x1") + status, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/"+testTxOne) if err != nil { t.Fatal(err) } @@ -819,7 +819,7 @@ func TestWaitForDataSetCreated_ConfirmedWithoutResultStillPending(t *testing.T) } _, _ = fmt.Fprint(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000001","service":"svc","txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":42}`) })) - status, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/0x1", time.Millisecond) + status, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/"+testTxOne, time.Millisecond) if err != nil { t.Fatal(err) } @@ -833,7 +833,7 @@ func TestWaitForDataSetCreated_Rejected(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000001","service":"svc","txStatus":"rejected","dataSetCreated":false,"ok":false}`) })) - _, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/0x1", 10*time.Millisecond) + _, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/"+testTxOne, 10*time.Millisecond) if !errors.Is(err, ErrTxRejected) { t.Fatalf("want ErrTxRejected, got %v", err) } @@ -843,7 +843,7 @@ func TestWaitForDataSetCreated_404ReturnsHTTPError(t *testing.T) { c, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "", http.StatusNotFound) })) - _, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/0x1", 10*time.Millisecond) + _, err := c.WaitForDataSetCreated(context.Background(), c.BaseURL().String()+"pdp/data-sets/created/"+testTxOne, 10*time.Millisecond) he, ok := errors.AsType[*HTTPError](err) if !ok { t.Fatalf("want HTTPError, got %T (%v)", err, err) @@ -1054,9 +1054,9 @@ func TestWaitForPiecesAdded(t *testing.T) { _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"pending","dataSetId":5,"pieceCount":1,"addMessageOk":null,"piecesAdded":false}`) return } - _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10,11]}`) + _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":5,"pieceCount":2,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10,11]}`) })) - status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status", 10*time.Millisecond) + status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status/"+testTxOne, 10*time.Millisecond) if err != nil { t.Fatal(err) } @@ -1071,7 +1071,7 @@ func TestGetAddPiecesStatus_Accepts202(t *testing.T) { w.WriteHeader(http.StatusAccepted) _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"pending","dataSetId":5,"pieceCount":1,"addMessageOk":null,"piecesAdded":false}`) })) - status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err != nil { t.Fatal(err) } @@ -1091,7 +1091,7 @@ func TestWaitForPiecesAdded_ConfirmedWithoutResultStillPending(t *testing.T) { } _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10]}`) })) - status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status", time.Millisecond) + status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status/"+testTxOne, time.Millisecond) if err != nil { t.Fatal(err) } @@ -1104,7 +1104,7 @@ func TestWaitForPiecesAdded_404ReturnsHTTPError(t *testing.T) { c, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "", http.StatusNotFound) })) - _, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status", 10*time.Millisecond) + _, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status/"+testTxOne, 10*time.Millisecond) he, ok := errors.AsType[*HTTPError](err) if !ok { t.Fatalf("want HTTPError, got %T (%v)", err, err) @@ -1119,7 +1119,7 @@ func TestGetAddPiecesStatus_LargeUint64DataSetID(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":9223372036854775808,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10]}`) })) - status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err != nil { t.Fatal(err) } diff --git a/pdp/coverage_test.go b/pdp/coverage_test.go index 0a32fec..4fd5ec0 100644 --- a/pdp/coverage_test.go +++ b/pdp/coverage_test.go @@ -292,7 +292,7 @@ func TestWaitForCreateDataSetAndAddPieces_MissingDataSetID(t *testing.T) { // dataSetCreated=true but no dataSetId _, _ = fmt.Fprintf(w, `{"createMessageHash":%q,"service":"svc","txStatus":"confirmed","dataSetCreated":true,"ok":true}`, txHash) })) - _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status", 5*time.Millisecond) + _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status/"+txHash, 5*time.Millisecond) if !errors.Is(err, ErrInvalidStatus) { t.Errorf("want ErrInvalidStatus, got %v", err) } @@ -303,7 +303,7 @@ func TestWaitForCreateDataSetAndAddPieces_ZeroCreateMessageHash(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprintf(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000000","service":"svc","txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":42}`) })) - _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status", 5*time.Millisecond) + _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status/"+testOriginalTx, 5*time.Millisecond) if !errors.Is(err, ErrInvalidStatus) { t.Errorf("want ErrInvalidStatus, got %v", err) } @@ -313,7 +313,7 @@ func TestWaitForCreateDataSetAndAddPieces_WaitError(t *testing.T) { c, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "boom", http.StatusInternalServerError) })) - _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status", 5*time.Millisecond) + _, err := c.WaitForCreateDataSetAndAddPieces(context.Background(), c.BaseURL().String()+"status/"+testOriginalTx, 5*time.Millisecond) if err == nil { t.Error("expected error from WaitForDataSetCreated propagation") } @@ -398,7 +398,7 @@ func TestWaitForPiecesAdded_Rejected(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"rejected","dataSetId":5,"pieceCount":1,"addMessageOk":false,"piecesAdded":false}`) })) - _, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status", 10*time.Millisecond) + _, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status/"+testTxOne, 10*time.Millisecond) if !errors.Is(err, ErrTxRejected) { t.Fatalf("want ErrTxRejected, got %v", err) } @@ -411,7 +411,7 @@ func TestWaitForPiecesAdded_ContextCancelled(t *testing.T) { })) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - _, err := c.WaitForPiecesAdded(ctx, c.BaseURL().String()+"status", 50*time.Millisecond) + _, err := c.WaitForPiecesAdded(ctx, c.BaseURL().String()+"status/"+testTxOne, 50*time.Millisecond) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("want DeadlineExceeded, got %v", err) } @@ -540,8 +540,8 @@ func TestGetAddPiecesStatus_EmptyURL(t *testing.T) { t.Fatal("should not be called") })) _, err := c.GetAddPiecesStatus(context.Background(), "") - if err == nil || !strings.Contains(err.Error(), "empty statusURL") { - t.Errorf("want empty statusURL error, got %v", err) + if !errors.Is(err, ErrInvalidStatusURL) { + t.Errorf("want ErrInvalidStatusURL, got %v", err) } } @@ -550,7 +550,7 @@ func TestGetAddPiecesStatus_BadJSON(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{invalid json`) })) - _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err == nil { t.Error("expected JSON decode error") } @@ -561,7 +561,7 @@ func TestGetAddPiecesStatus_BadDataSetID(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":1.5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true}`) })) - _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err == nil || !strings.Contains(err.Error(), "bad dataSetId") { t.Errorf("want bad dataSetId error, got %v", err) } @@ -572,7 +572,7 @@ func TestGetAddPiecesStatus_BadConfirmedPieceID(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[1.5]}`) })) - _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + _, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err == nil || !strings.Contains(err.Error(), "bad confirmedPieceId") { t.Errorf("want bad confirmedPieceId error, got %v", err) } @@ -585,8 +585,8 @@ func TestGetDataSetCreationStatus_EmptyURL(t *testing.T) { t.Fatal("should not be called") })) _, err := c.GetDataSetCreationStatus(context.Background(), "") - if err == nil || !strings.Contains(err.Error(), "empty statusURL") { - t.Errorf("want empty statusURL error, got %v", err) + if !errors.Is(err, ErrInvalidStatusURL) { + t.Errorf("want ErrInvalidStatusURL, got %v", err) } } @@ -595,7 +595,7 @@ func TestGetDataSetCreationStatus_BadJSON(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `not json`) })) - _, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"status") + _, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err == nil { t.Error("expected JSON decode error") } @@ -606,7 +606,7 @@ func TestGetDataSetCreationStatus_BadDataSetID(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"createMessageHash":"0x0000000000000000000000000000000000000000000000000000000000000001","service":"svc","txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":1.5}`) })) - _, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"status") + _, err := c.GetDataSetCreationStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err == nil || !strings.Contains(err.Error(), "bad dataSetId") { t.Errorf("want bad dataSetId error, got %v", err) } @@ -621,7 +621,7 @@ func TestWaitForDataSetCreated_ContextCancelled(t *testing.T) { })) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - _, err := c.WaitForDataSetCreated(ctx, c.BaseURL().String()+"status", 50*time.Millisecond) + _, err := c.WaitForDataSetCreated(ctx, c.BaseURL().String()+"status/"+testTxOne, 50*time.Millisecond) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("want DeadlineExceeded, got %v", err) } @@ -1010,7 +1010,7 @@ func TestWaitForPiecesAdded_ZeroPollIntervalUsesDefault(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10]}`) })) - status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status", 0) + status, err := c.WaitForPiecesAdded(context.Background(), c.BaseURL().String()+"status/"+testTxOne, 0) if err != nil { t.Fatal(err) } @@ -1024,9 +1024,9 @@ func TestWaitForPiecesAdded_ZeroPollIntervalUsesDefault(t *testing.T) { func TestGetAddPiecesStatus_FullWidthDataSetID(t *testing.T) { c, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":99999999999999999999999999999999,"pieceCount":1,"addMessageOk":true,"piecesAdded":true}`) + _, _ = fmt.Fprint(w, `{"txHash":"0x0000000000000000000000000000000000000000000000000000000000000001","txStatus":"confirmed","dataSetId":99999999999999999999999999999999,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[10]}`) })) - status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status") + status, err := c.GetAddPiecesStatus(context.Background(), c.BaseURL().String()+"status/"+testTxOne) if err != nil { t.Fatal(err) } diff --git a/pdp/dataset.go b/pdp/dataset.go index 2a0d315..cf49142 100644 --- a/pdp/dataset.go +++ b/pdp/dataset.go @@ -99,10 +99,12 @@ type rawCreateDataSetStatus struct { // GetDataSetCreationStatus polls the status URL once. Providers may return // either HTTP 200 or 202 with the same JSON body shape. func (c *Client) GetDataSetCreationStatus(ctx context.Context, statusURL string) (*CreateDataSetStatus, error) { - if statusURL == "" { - return nil, fmt.Errorf("pdp.GetDataSetCreationStatus: %w: empty statusURL", ErrStatusURLOrigin) + const op = "pdp.GetDataSetCreationStatus" + expectedHash, err := c.statusURLTransactionHash(op, statusURL) + if err != nil { + return nil, err } - body, err := c.getStatusBody(ctx, "pdp.GetDataSetCreationStatus", statusURL, http.StatusOK, http.StatusAccepted) + body, err := c.getStatusBody(ctx, op, statusURL, http.StatusOK, http.StatusAccepted) if err != nil { return nil, err } @@ -114,6 +116,9 @@ func (c *Client) GetDataSetCreationStatus(ctx context.Context, statusURL string) if err != nil { return nil, err } + if createMessageHash != expectedHash { + return nil, invalidStatusf(op, "createMessageHash does not match status URL") + } confirmedTxHash, err := parseOptionalStatusHash("pdp.GetDataSetCreationStatus", "confirmedTxHash", raw.ConfirmedTxHash) if err != nil { return nil, err diff --git a/pdp/doc.go b/pdp/doc.go index 8a3a1da..e58d441 100644 --- a/pdp/doc.go +++ b/pdp/doc.go @@ -46,9 +46,11 @@ // # Status URLs // // Transaction status URLs must be absolute HTTP(S) URLs with the same scheme, -// hostname, and effective port as the client's service URL. The same check is -// applied to every redirect. This pins recovery handles to their provider -// origin; it does not block private addresses or DNS rebinding. +// hostname, and effective port as the client's service URL, and their final +// path segment must be a nonzero transaction hash. Status responses must report +// the same hash. The origin check is applied to every redirect. This pins +// recovery handles to their provider origin and transaction; it does not block +// private addresses or DNS rebinding. // // # Response size cap // diff --git a/pdp/errors.go b/pdp/errors.go index db8f2a4..44174da 100644 --- a/pdp/errors.go +++ b/pdp/errors.go @@ -131,6 +131,10 @@ var ErrTxRejected = errors.New("pdp: transaction rejected") // incomplete, or internally inconsistent. var ErrInvalidStatus = errors.New("pdp: invalid status response") +// ErrInvalidStatusURL is returned when a provider status URL is malformed or +// does not identify a transaction status resource. +var ErrInvalidStatusURL = errors.New("pdp: invalid status URL") + // ErrPingResponseMismatch is returned when /pdp/ping responds successfully // but its bounded response body is not the Curio PDP identity token. var ErrPingResponseMismatch = errors.New("pdp: ping response mismatch") diff --git a/pdp/pieces.go b/pdp/pieces.go index 6e63b21..3253383 100644 --- a/pdp/pieces.go +++ b/pdp/pieces.go @@ -146,10 +146,12 @@ type rawAddPiecesStatus struct { // GetAddPiecesStatus polls the status URL once. Providers may return either // HTTP 200 or 202 with the same JSON body shape. func (c *Client) GetAddPiecesStatus(ctx context.Context, statusURL string) (*AddPiecesStatus, error) { - if statusURL == "" { - return nil, fmt.Errorf("pdp.GetAddPiecesStatus: %w: empty statusURL", ErrStatusURLOrigin) + const op = "pdp.GetAddPiecesStatus" + expectedHash, err := c.statusURLTransactionHash(op, statusURL) + if err != nil { + return nil, err } - body, err := c.getStatusBody(ctx, "pdp.GetAddPiecesStatus", statusURL, http.StatusOK, http.StatusAccepted) + body, err := c.getStatusBody(ctx, op, statusURL, http.StatusOK, http.StatusAccepted) if err != nil { return nil, err } @@ -161,6 +163,9 @@ func (c *Client) GetAddPiecesStatus(ctx context.Context, statusURL string) (*Add if err != nil { return nil, err } + if txHash != expectedHash { + return nil, invalidStatusf(op, "txHash does not match status URL") + } confirmedTxHash, err := parseOptionalStatusHash("pdp.GetAddPiecesStatus", "confirmedTxHash", raw.ConfirmedTxHash) if err != nil { return nil, err diff --git a/pdp/status.go b/pdp/status.go index 871d688..b5896aa 100644 --- a/pdp/status.go +++ b/pdp/status.go @@ -39,6 +39,34 @@ func parseOptionalStatusHash(op, field, value string) (common.Hash, error) { return parseRequiredStatusHash(op, field, value) } +func (c *Client) statusURLTransactionHash(op, statusURL string) (common.Hash, error) { + if statusURL == "" { + return common.Hash{}, fmt.Errorf("%s: %w: empty status URL", op, ErrInvalidStatusURL) + } + u, err := url.Parse(statusURL) + if err != nil { + return common.Hash{}, fmt.Errorf("%s: %w", op, ErrInvalidStatusURL) + } + if err := c.validateStatusURL(u); err != nil { + return common.Hash{}, fmt.Errorf("%s: %w", op, err) + } + hashHex := lastPathSegment(statusURL) + if hashHex == "" { + return common.Hash{}, fmt.Errorf("%s: %w: missing transaction hash", op, ErrInvalidStatusURL) + } + if !strings.HasPrefix(hashHex, "0x") { + hashHex = "0x" + hashHex + } + if !common.IsHexHash(hashHex) { + return common.Hash{}, fmt.Errorf("%s: %w: invalid transaction hash", op, ErrInvalidStatusURL) + } + hash := common.HexToHash(hashHex) + if hash == (common.Hash{}) { + return common.Hash{}, fmt.Errorf("%s: %w: zero transaction hash", op, ErrInvalidStatusURL) + } + return hash, nil +} + func invalidStatusf(op, format string, args ...any) error { return fmt.Errorf("%s: %w: %s", op, ErrInvalidStatus, fmt.Sprintf(format, args...)) } @@ -108,6 +136,19 @@ func classifyAddPiecesStatus(op string, status *AddPiecesStatus) (transactionSta if len(status.ConfirmedPieceIDs) > 0 && !status.PiecesAdded { return transactionPending, invalidStatusf(op, "confirmedPieceIds without piecesAdded") } + if status.PiecesAdded { + if status.PieceCount == 0 { + return transactionPending, invalidStatusf(op, "piecesAdded with zero pieceCount") + } + if status.PieceCount != len(status.ConfirmedPieceIDs) { + return transactionPending, invalidStatusf( + op, + "pieceCount %d does not match confirmedPieceIds count %d", + status.PieceCount, + len(status.ConfirmedPieceIDs), + ) + } + } switch status.TxStatus { case "pending": @@ -146,7 +187,7 @@ func classifyAddPiecesStatus(op string, status *AddPiecesStatus) (transactionSta func (c *Client) resolveStatusURL(ref string) (string, error) { u, err := c.resolve(ref) if err != nil { - return "", fmt.Errorf("%w: invalid status URL", ErrStatusURLOrigin) + return "", fmt.Errorf("%w: invalid status URL", ErrInvalidStatusURL) } if err := c.validateStatusURL(u); err != nil { return "", err @@ -157,7 +198,7 @@ func (c *Client) resolveStatusURL(ref string) (string, error) { func (c *Client) getStatusBody(ctx context.Context, op, statusURL string, expectStatuses ...int) ([]byte, error) { u, err := url.Parse(statusURL) if err != nil { - return nil, fmt.Errorf("%s: %w: invalid status URL", op, ErrStatusURLOrigin) + return nil, fmt.Errorf("%s: %w", op, ErrInvalidStatusURL) } if err := c.validateStatusURL(u); err != nil { return nil, fmt.Errorf("%s: %w", op, err) @@ -180,10 +221,10 @@ func (c *Client) getStatusBody(ctx context.Context, op, statusURL string, expect func (c *Client) validateStatusURL(u *url.URL) error { if u == nil || !u.IsAbs() || (u.Scheme != "http" && u.Scheme != "https") || u.Hostname() == "" { - return fmt.Errorf("%w: %s", ErrStatusURLOrigin, redact.URL(u)) + return fmt.Errorf("%w: %w: %s", ErrInvalidStatusURL, ErrStatusURLOrigin, redact.URL(u)) } if !sameOrigin(c.baseURL, u) { - return fmt.Errorf("%w: %s", ErrStatusURLOrigin, redact.URL(u)) + return fmt.Errorf("%w: %w: %s", ErrInvalidStatusURL, ErrStatusURLOrigin, redact.URL(u)) } return nil } diff --git a/pdp/status_test.go b/pdp/status_test.go index bb14f3f..839756e 100644 --- a/pdp/status_test.go +++ b/pdp/status_test.go @@ -20,6 +20,7 @@ import ( const ( testOriginalTx = "0x0000000000000000000000000000000000000000000000000000000000000011" testConfirmedTx = "0x0000000000000000000000000000000000000000000000000000000000000022" + testTxOne = "0x0000000000000000000000000000000000000000000000000000000000000001" ) func TestGetAddPiecesStatusNormalizesWireStates(t *testing.T) { @@ -99,6 +100,16 @@ func TestGetAddPiecesStatusNormalizesWireStates(t *testing.T) { body: fmt.Sprintf(`{"txHash":%q,"txStatus":"failed","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":false}`, testOriginalTx), wantError: ErrInvalidStatus, }, + { + name: "confirmed with zero pieces", + body: fmt.Sprintf(`{"txHash":%q,"txStatus":"confirmed","dataSetId":5,"pieceCount":0,"addMessageOk":true,"piecesAdded":true}`, testOriginalTx), + wantError: ErrInvalidStatus, + }, + { + name: "confirmed piece count mismatch", + body: fmt.Sprintf(`{"txHash":%q,"txStatus":"confirmed","dataSetId":5,"pieceCount":2,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[7]}`, testOriginalTx), + wantError: ErrInvalidStatus, + }, } for _, test := range tests { @@ -109,7 +120,7 @@ func TestGetAddPiecesStatusNormalizesWireStates(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, test.body) })) - status, err := client.GetAddPiecesStatus(context.Background(), client.BaseURL().String()+"status") + status, err := client.GetAddPiecesStatus(context.Background(), client.BaseURL().String()+"status/"+testOriginalTx) if !errors.Is(err, test.wantError) { t.Fatalf("error=%v want %v", err, test.wantError) } @@ -196,7 +207,7 @@ func TestGetDataSetCreationStatusNormalizesWireStates(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprint(w, test.body) })) - status, err := client.GetDataSetCreationStatus(context.Background(), client.BaseURL().String()+"status") + status, err := client.GetDataSetCreationStatus(context.Background(), client.BaseURL().String()+"status/"+testOriginalTx) if !errors.Is(err, test.wantError) { t.Fatalf("error=%v want %v", err, test.wantError) } @@ -207,6 +218,67 @@ func TestGetDataSetCreationStatusNormalizesWireStates(t *testing.T) { } } +func TestStatusResponsesMustMatchTransactionHashInURL(t *testing.T) { + tests := []struct { + name string + call func(context.Context, *Client, string) error + body string + }{ + { + name: "add pieces", + call: func(ctx context.Context, client *Client, statusURL string) error { + _, err := client.GetAddPiecesStatus(ctx, statusURL) + return err + }, + body: fmt.Sprintf(`{"txHash":%q,"txStatus":"pending","dataSetId":5,"pieceCount":0,"piecesAdded":false}`, testOriginalTx), + }, + { + name: "create data set", + call: func(ctx context.Context, client *Client, statusURL string) error { + _, err := client.GetDataSetCreationStatus(ctx, statusURL) + return err + }, + body: fmt.Sprintf(`{"createMessageHash":%q,"txStatus":"pending","dataSetCreated":false}`, testOriginalTx), + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + client, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprint(w, test.body) + })) + err := test.call(context.Background(), client, client.BaseURL().String()+"status/"+testConfirmedTx) + if !errors.Is(err, ErrInvalidStatus) || errors.Is(err, ErrInvalidStatusURL) { + t.Fatalf("error=%v want ErrInvalidStatus only", err) + } + }) + } +} + +func TestStatusMethodsRejectURLWithoutTransactionHashBeforeRequest(t *testing.T) { + requests := 0 + client, _ := newTestClient(t, http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + requests++ + })) + for _, call := range []func() error{ + func() error { + _, err := client.GetAddPiecesStatus(context.Background(), client.BaseURL().String()+"status/not-a-hash") + return err + }, + func() error { + _, err := client.GetDataSetCreationStatus(context.Background(), client.BaseURL().String()+"status/not-a-hash") + return err + }, + } { + if err := call(); !errors.Is(err, ErrInvalidStatusURL) || errors.Is(err, ErrStatusURLOrigin) { + t.Fatalf("error=%v want ErrInvalidStatusURL only", err) + } + } + if requests != 0 { + t.Fatalf("requests=%d want 0", requests) + } +} + func TestGetAddPiecesStatusRetriesTransportStatusButNotPending(t *testing.T) { requests := 0 client, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -218,7 +290,7 @@ func TestGetAddPiecesStatusRetriesTransportStatusButNotPending(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprintf(w, `{"txHash":%q,"txStatus":"pending","dataSetId":5,"pieceCount":0,"piecesAdded":false}`, testOriginalTx) })) - status, err := client.GetAddPiecesStatus(context.Background(), client.BaseURL().String()+"status") + status, err := client.GetAddPiecesStatus(context.Background(), client.BaseURL().String()+"status/"+testOriginalTx) if err != nil { t.Fatal(err) } @@ -252,7 +324,7 @@ func TestGetAddPiecesStatusRetriesTransportTimeout(t *testing.T) { t.Fatal(err) } client.retryDelayFn = noRetryDelay - status, err := client.GetAddPiecesStatus(context.Background(), "https://provider.example/status") + status, err := client.GetAddPiecesStatus(context.Background(), "https://provider.example/status/"+testOriginalTx) if err != nil { t.Fatal(err) } @@ -273,7 +345,7 @@ func TestStatusTimeoutPreservesClassificationAndRedactsURL(t *testing.T) { if err != nil { t.Fatal(err) } - _, err = client.GetAddPiecesStatus(context.Background(), "https://provider.example/status?token=secret") + _, err = client.GetAddPiecesStatus(context.Background(), "https://provider.example/status/"+testOriginalTx+"?token=secret") if err == nil { t.Fatal("expected timeout") } @@ -304,7 +376,7 @@ func TestGetAddPiecesStatusDoesNotRetryCallerCancellation(t *testing.T) { t.Fatal(err) } client.retryDelayFn = noRetryDelay - _, err = client.GetAddPiecesStatus(ctx, "https://provider.example/status") + _, err = client.GetAddPiecesStatus(ctx, "https://provider.example/status/"+testOriginalTx) if !errors.Is(err, context.Canceled) { t.Fatalf("error=%v want context.Canceled", err) } @@ -341,7 +413,7 @@ func TestStatusURLOriginValidation(t *testing.T) { if parseErr != nil { t.Fatal(parseErr) } - if err := client.validateStatusURL(u); !errors.Is(err, ErrStatusURLOrigin) { + if err := client.validateStatusURL(u); !errors.Is(err, ErrStatusURLOrigin) || !errors.Is(err, ErrInvalidStatusURL) { t.Fatalf("validate %q: %v", raw, err) } } @@ -368,8 +440,8 @@ func TestGetStatusRejectsCrossOriginRedirectBeforeFollowing(t *testing.T) { if err != nil { t.Fatal(err) } - _, err = client.GetAddPiecesStatus(context.Background(), source.URL+"/start") - if !errors.Is(err, ErrStatusURLOrigin) { + _, err = client.GetAddPiecesStatus(context.Background(), source.URL+"/start/"+testOriginalTx) + if !errors.Is(err, ErrStatusURLOrigin) || !errors.Is(err, ErrInvalidStatusURL) { t.Fatalf("error=%v want ErrStatusURLOrigin", err) } if targetRequests != 0 || customRedirectCalls != 0 { @@ -383,8 +455,8 @@ func TestGetStatusRejectsCrossOriginRedirectBeforeFollowing(t *testing.T) { func TestGetStatusPreservesCustomRedirectPolicy(t *testing.T) { want := errors.New("redirect denied") source := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/start" { - http.Redirect(w, r, "/status", http.StatusFound) + if r.URL.Path == "/start/"+testOriginalTx { + http.Redirect(w, r, "/status/"+testOriginalTx, http.StatusFound) return } t.Fatal("custom redirect policy should stop the redirect") @@ -396,7 +468,7 @@ func TestGetStatusPreservesCustomRedirectPolicy(t *testing.T) { if err != nil { t.Fatal(err) } - _, err = client.GetAddPiecesStatus(context.Background(), source.URL+"/start") + _, err = client.GetAddPiecesStatus(context.Background(), source.URL+"/start/"+testOriginalTx) if !errors.Is(err, want) { t.Fatalf("error=%v want custom redirect error", err) } @@ -444,7 +516,7 @@ func TestSubmissionsRejectCrossOriginStatusLocations(t *testing.T) { w.Header().Set("Location", location) w.WriteHeader(http.StatusCreated) })) - if err := test.call(context.Background(), client); !errors.Is(err, ErrStatusURLOrigin) { + if err := test.call(context.Background(), client); !errors.Is(err, ErrStatusURLOrigin) || !errors.Is(err, ErrInvalidStatusURL) { t.Fatalf("error=%v want ErrStatusURLOrigin", err) } else if strings.Contains(err.Error(), "secret") { t.Fatalf("error leaked Location query value: %v", err) @@ -457,7 +529,7 @@ func TestGetCreateDataSetAndAddPiecesStatusRejectsCrossStageMismatch(t *testing. client, server := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.URL.Path { - case "/create": + case "/create/" + testOriginalTx: _, _ = fmt.Fprintf(w, `{"createMessageHash":%q,"txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":42}`, testOriginalTx) case "/pdp/data-sets/42/pieces/added/" + testOriginalTx: _, _ = fmt.Fprintf(w, `{"txHash":%q,"txStatus":"confirmed","dataSetId":99,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[7]}`, testOriginalTx) @@ -465,7 +537,7 @@ func TestGetCreateDataSetAndAddPiecesStatusRejectsCrossStageMismatch(t *testing. t.Fatalf("unexpected path %s", r.URL.Path) } })) - status, err := client.GetCreateDataSetAndAddPiecesStatus(context.Background(), server.URL+"/create") + status, err := client.GetCreateDataSetAndAddPiecesStatus(context.Background(), server.URL+"/create/"+testOriginalTx) if !errors.Is(err, ErrInvalidStatus) { t.Fatalf("error=%v want ErrInvalidStatus", err) } @@ -480,7 +552,7 @@ func TestGetCreateDataSetAndAddPiecesStatusRejectsConfirmedHashConflict(t *testi client, server := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.URL.Path { - case "/create": + case "/create/" + testOriginalTx: _, _ = fmt.Fprintf(w, `{"createMessageHash":%q,"confirmedTxHash":%q,"txStatus":"confirmed","dataSetCreated":true,"ok":true,"dataSetId":42}`, testOriginalTx, createConfirmed) case "/pdp/data-sets/42/pieces/added/" + testOriginalTx: _, _ = fmt.Fprintf(w, `{"txHash":%q,"confirmedTxHash":%q,"txStatus":"confirmed","dataSetId":42,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[7]}`, testOriginalTx, addConfirmed) @@ -488,7 +560,7 @@ func TestGetCreateDataSetAndAddPiecesStatusRejectsConfirmedHashConflict(t *testi t.Fatalf("unexpected path %s", r.URL.Path) } })) - _, err := client.GetCreateDataSetAndAddPiecesStatus(context.Background(), server.URL+"/create") + _, err := client.GetCreateDataSetAndAddPiecesStatus(context.Background(), server.URL+"/create/"+testOriginalTx) if !errors.Is(err, ErrInvalidStatus) { t.Fatalf("error=%v want ErrInvalidStatus", err) } diff --git a/storage/context.go b/storage/context.go index 030f387..f4f2e75 100644 --- a/storage/context.go +++ b/storage/context.go @@ -652,7 +652,7 @@ func (c *ProviderContext) CreateAndAdd(ctx context.Context, req CreateAndAddRequ if err != nil { return nil, err } - return c.WaitForCreateAndAdd(ctx, *submission) + return c.WaitForCreateAndAdd(ctx, submission.StatusURL, *submission.ClientDataSetID) } // Commit adds pieces to the bound data set and waits for confirmation. @@ -661,7 +661,7 @@ func (c *DataSetContext) Commit(ctx context.Context, req CommitRequest) (*Commit if err != nil { return nil, err } - return c.WaitForCommit(ctx, *submission) + return c.WaitForCommit(ctx, submission.StatusURL) } func (c *contextCore) validateWritableDataSet(ctx context.Context, op string, ref *DataSetRef) error { diff --git a/storage/context_commit.go b/storage/context_commit.go index 91db1d2..1abeb68 100644 --- a/storage/context_commit.go +++ b/storage/context_commit.go @@ -49,28 +49,62 @@ func (c *ProviderContext) submitCommit(ctx context.Context, req commitRequest) ( return c.core.submitCommit(ctx, "storage.ProviderContext.SubmitCreateAndAdd", nil, req) } -// GetCreateAndAddStatus checks a create-and-add submission once. A -// caller-supplied status URL outside the provider origin returns an error -// matching both [ErrInvalidArgument] and [pdp.ErrStatusURLOrigin]. -func (c *ProviderContext) GetCreateAndAddStatus(ctx context.Context, submission CommitSubmission) (*CommitStatus, error) { +// GetCreateAndAddStatus checks a create-and-add status URL once. An invalid +// status URL returns an error matching [ErrInvalidArgument] and +// [pdp.ErrInvalidStatusURL]. An origin mismatch also matches +// [pdp.ErrStatusURLOrigin]. ClientDataSetID may be zero. +func (c *ProviderContext) GetCreateAndAddStatus( + ctx context.Context, + statusURL string, + clientDataSetID types.BigInt, +) (*CommitStatus, error) { if c == nil || c.core == nil { return nil, fmt.Errorf("storage.ProviderContext.GetCreateAndAddStatus: %w: nil context", ErrInvalidArgument) } - return c.core.getCommitStatus(ctx, "storage.ProviderContext.GetCreateAndAddStatus", nil, submission) + return c.core.getCommitStatus( + ctx, + "storage.ProviderContext.GetCreateAndAddStatus", + nil, + statusURL, + clientDataSetID, + ) } -// WaitForCreateAndAdd waits for a create-and-add submission to confirm or -// reject. A caller-supplied status URL outside the provider origin returns an -// error matching both [ErrInvalidArgument] and [pdp.ErrStatusURLOrigin]. -func (c *ProviderContext) WaitForCreateAndAdd(ctx context.Context, submission CommitSubmission) (*CommitResult, error) { - return c.waitForCommit(ctx, submission) +// WaitForCreateAndAdd waits for a create-and-add status URL to confirm or +// reject. An invalid status URL returns an error matching [ErrInvalidArgument] +// and [pdp.ErrInvalidStatusURL]. An origin mismatch also matches +// [pdp.ErrStatusURLOrigin]. ClientDataSetID may be zero. +func (c *ProviderContext) WaitForCreateAndAdd( + ctx context.Context, + statusURL string, + clientDataSetID types.BigInt, +) (*CommitResult, error) { + if c == nil || c.core == nil { + return nil, fmt.Errorf("storage.ProviderContext.WaitForCreateAndAdd: %w: nil context", ErrInvalidArgument) + } + return c.core.waitForCommit( + ctx, + "storage.ProviderContext.WaitForCreateAndAdd", + nil, + statusURL, + clientDataSetID, + ) } func (c *ProviderContext) waitForCommit(ctx context.Context, submission CommitSubmission) (*CommitResult, error) { if c == nil || c.core == nil { return nil, fmt.Errorf("storage.ProviderContext.WaitForCreateAndAdd: %w: nil context", ErrInvalidArgument) } - return c.core.waitForCommit(ctx, "storage.ProviderContext.WaitForCreateAndAdd", nil, submission) + if submission.ClientDataSetID == nil { + return nil, errors.New("storage.ProviderContext.WaitForCreateAndAdd: invalid provider submission: missing clientDataSetID") + } + return c.core.waitForCommit( + ctx, + "storage.ProviderContext.WaitForCreateAndAdd", + nil, + submission.StatusURL, + *submission.ClientDataSetID, + ) } // SubmitCommit submits one add-pieces transaction and returns without waiting @@ -86,28 +120,51 @@ func (c *DataSetContext) submitCommit(ctx context.Context, req commitRequest) (* return c.core.submitCommit(ctx, "storage.DataSetContext.SubmitCommit", &c.ref, req) } -// GetCommitStatus checks an add-pieces submission once. A caller-supplied -// status URL outside the provider origin returns an error matching both -// [ErrInvalidArgument] and [pdp.ErrStatusURLOrigin]. -func (c *DataSetContext) GetCommitStatus(ctx context.Context, submission CommitSubmission) (*CommitStatus, error) { +// GetCommitStatus checks an add-pieces status URL once. A caller-supplied +// invalid status URL returns an error matching [ErrInvalidArgument] and +// [pdp.ErrInvalidStatusURL]. An origin mismatch also matches +// [pdp.ErrStatusURLOrigin]. +func (c *DataSetContext) GetCommitStatus(ctx context.Context, statusURL string) (*CommitStatus, error) { if c == nil || c.core == nil { return nil, fmt.Errorf("storage.DataSetContext.GetCommitStatus: %w: nil context", ErrInvalidArgument) } - return c.core.getCommitStatus(ctx, "storage.DataSetContext.GetCommitStatus", &c.ref, submission) + return c.core.getCommitStatus( + ctx, + "storage.DataSetContext.GetCommitStatus", + &c.ref, + statusURL, + types.BigInt{}, + ) } -// WaitForCommit waits for an add-pieces submission to confirm or reject. A -// caller-supplied status URL outside the provider origin returns an error -// matching both [ErrInvalidArgument] and [pdp.ErrStatusURLOrigin]. -func (c *DataSetContext) WaitForCommit(ctx context.Context, submission CommitSubmission) (*CommitResult, error) { - return c.waitForCommit(ctx, submission) +// WaitForCommit waits for an add-pieces status URL to confirm or reject. A +// caller-supplied invalid status URL returns an error matching +// [ErrInvalidArgument] and [pdp.ErrInvalidStatusURL]. An origin mismatch also +// matches [pdp.ErrStatusURLOrigin]. +func (c *DataSetContext) WaitForCommit(ctx context.Context, statusURL string) (*CommitResult, error) { + if c == nil || c.core == nil { + return nil, fmt.Errorf("storage.DataSetContext.WaitForCommit: %w: nil context", ErrInvalidArgument) + } + return c.core.waitForCommit( + ctx, + "storage.DataSetContext.WaitForCommit", + &c.ref, + statusURL, + types.BigInt{}, + ) } func (c *DataSetContext) waitForCommit(ctx context.Context, submission CommitSubmission) (*CommitResult, error) { if c == nil || c.core == nil { return nil, fmt.Errorf("storage.DataSetContext.WaitForCommit: %w: nil context", ErrInvalidArgument) } - return c.core.waitForCommit(ctx, "storage.DataSetContext.WaitForCommit", &c.ref, submission) + return c.core.waitForCommit( + ctx, + "storage.DataSetContext.WaitForCommit", + &c.ref, + submission.StatusURL, + types.BigInt{}, + ) } func (c *contextCore) submitCommit( @@ -195,7 +252,7 @@ func (c *contextCore) submitCommit( submission.StatusURL = created.StatusURL } - validated, err := c.validateCommitSubmission(op, ref, submission, false) + validated, err := c.validateCommitSubmission(op, ref, submission) if err != nil { return nil, err } @@ -299,42 +356,50 @@ func (c *contextCore) getCommitStatus( ctx context.Context, op string, ref *DataSetRef, - submission CommitSubmission, + statusURL string, + clientDataSetID types.BigInt, ) (*CommitStatus, error) { - validated, err := c.validateCommitSubmission(op, ref, submission, true) - if err != nil { + if err := validateRecoveryStatusURL(op, c.provider.ServiceURL, statusURL); err != nil { return nil, err } - return c.getValidatedCommitStatus(ctx, op, ref, validated) + return c.getValidatedCommitStatus(ctx, op, ref, statusURL, clientDataSetID) } func (c *contextCore) getValidatedCommitStatus( ctx context.Context, op string, ref *DataSetRef, - submission CommitSubmission, + statusURL string, + clientDataSetID types.BigInt, ) (*CommitStatus, error) { if ref != nil { - return c.getAddPiecesCommitStatus(ctx, op, submission) + return c.getAddPiecesCommitStatus(ctx, op, *ref, statusURL) } - return c.getCreateAndAddCommitStatus(ctx, op, submission) + return c.getCreateAndAddCommitStatus(ctx, op, statusURL, clientDataSetID) } func (c *contextCore) getAddPiecesCommitStatus( ctx context.Context, op string, - submission CommitSubmission, + ref DataSetRef, + statusURL string, ) (*CommitStatus, error) { - snapshot, err := c.client.GetAddPiecesStatus(ctx, submission.StatusURL) + snapshot, err := c.client.GetAddPiecesStatus(ctx, statusURL) rejected := errors.Is(err, pdp.ErrTxRejected) if err != nil && !rejected { - return nil, fmt.Errorf("%s: get add-pieces status: %w", op, err) + return nil, wrapRecoveryStatusError(op, "get add-pieces status", err) } if snapshot == nil { return nil, invalidCommitStatusf(op, "nil add-pieces status") } - if err := validateAddCommitSnapshot(op, submission, snapshot, rejected); err != nil { - return nil, err + if !snapshot.DataSetID.Equal(ref.DataSetID()) { + return nil, invalidCommitStatusf(op, "dataSetID does not match context") + } + if snapshot.TxHash == (common.Hash{}) { + return nil, invalidCommitStatusf(op, "zero transactionID") + } + if snapshot.PiecesAdded && (snapshot.PieceCount <= 0 || snapshot.PieceCount != len(snapshot.ConfirmedPieceIDs)) { + return nil, invalidCommitStatusf(op, "confirmed piece counts differ") } state := CommitStatePending @@ -344,11 +409,11 @@ func (c *contextCore) getAddPiecesCommitStatus( state = CommitStateConfirmed } status := &CommitStatus{ - Kind: submission.Kind, + Kind: CommitKindAddPieces, State: state, - TransactionID: submission.TransactionID, + TransactionID: snapshot.TxHash.Hex(), ConfirmedTransactionID: optionalHashString(snapshot.ConfirmedTxHash), - DataSet: copyDataSetRefPtr(submission.DataSet), + DataSet: copyDataSetRefPtr(&ref), } if state == CommitStateConfirmed { status.PieceIDs = copyBigInts(snapshot.ConfirmedPieceIDs) @@ -359,24 +424,25 @@ func (c *contextCore) getAddPiecesCommitStatus( func (c *contextCore) getCreateAndAddCommitStatus( ctx context.Context, op string, - submission CommitSubmission, + statusURL string, + clientDataSetID types.BigInt, ) (*CommitStatus, error) { - snapshot, err := c.client.GetCreateDataSetAndAddPiecesStatus(ctx, submission.StatusURL) + snapshot, err := c.client.GetCreateDataSetAndAddPiecesStatus(ctx, statusURL) rejected := errors.Is(err, pdp.ErrTxRejected) if err != nil && !rejected { - return nil, fmt.Errorf("%s: get create-and-add status: %w", op, err) + return nil, wrapRecoveryStatusError(op, "get create-and-add status", err) } if snapshot == nil || snapshot.Create == nil { return nil, invalidCommitStatusf(op, "nil create status") } - if snapshot.Create.CreateMessageHash != common.HexToHash(submission.TransactionID) { - return nil, invalidCommitStatusf(op, "create transactionID does not match submission") + if snapshot.Create.CreateMessageHash == (common.Hash{}) { + return nil, invalidCommitStatusf(op, "zero transactionID") } confirmedHash := snapshot.Create.ConfirmedTxHash if snapshot.Add != nil { - if snapshot.Add.TxHash != common.HexToHash(submission.TransactionID) { - return nil, invalidCommitStatusf(op, "add transactionID does not match submission") + if snapshot.Add.TxHash != snapshot.Create.CreateMessageHash { + return nil, invalidCommitStatusf(op, "create and add transactionIDs differ") } if snapshot.Create.DataSetID == nil || !snapshot.Add.DataSetID.Equal(*snapshot.Create.DataSetID) { return nil, invalidCommitStatusf(op, "create and add dataSetIds differ") @@ -387,8 +453,9 @@ func (c *contextCore) getCreateAndAddCommitStatus( if snapshot.Add.ConfirmedTxHash != (common.Hash{}) { confirmedHash = snapshot.Add.ConfirmedTxHash } - if err := validateCommitPieceCount(op, snapshot.Add, len(submission.PieceCIDs), rejected); err != nil { - return nil, err + if snapshot.Add.PiecesAdded && + (snapshot.Add.PieceCount <= 0 || snapshot.Add.PieceCount != len(snapshot.Add.ConfirmedPieceIDs)) { + return nil, invalidCommitStatusf(op, "confirmed piece counts differ") } } @@ -399,16 +466,16 @@ func (c *contextCore) getCreateAndAddCommitStatus( state = CommitStateConfirmed } status := &CommitStatus{ - Kind: submission.Kind, + Kind: CommitKindCreateAndAdd, State: state, - TransactionID: submission.TransactionID, + TransactionID: snapshot.Create.CreateMessageHash.Hex(), ConfirmedTransactionID: optionalHashString(confirmedHash), } if state == CommitStateConfirmed { dataSet, err := NewDataSetRef( - submission.ProviderID, + c.provider.ID, *snapshot.Create.DataSetID, - *submission.ClientDataSetID, + clientDataSetID, ) if err != nil { return nil, invalidCommitStatusf(op, "invalid confirmed data-set identity") @@ -419,46 +486,18 @@ func (c *contextCore) getCreateAndAddCommitStatus( return status, nil } -func validateAddCommitSnapshot(op string, submission CommitSubmission, snapshot *pdp.AddPiecesStatus, rejected bool) error { - if snapshot.TxHash != common.HexToHash(submission.TransactionID) { - return invalidCommitStatusf(op, "transactionID does not match submission") - } - if !snapshot.DataSetID.Equal(submission.DataSet.DataSetID()) { - return invalidCommitStatusf(op, "dataSetID does not match submission") - } - return validateCommitPieceCount(op, snapshot, len(submission.PieceCIDs), rejected) -} - -func validateCommitPieceCount(op string, snapshot *pdp.AddPiecesStatus, expected int, rejected bool) error { - if snapshot.PiecesAdded { - if snapshot.PieceCount != expected { - return invalidCommitStatusf(op, "pieceCount %d does not match submission count %d", snapshot.PieceCount, expected) - } - } else if snapshot.PieceCount != 0 && snapshot.PieceCount != expected { - state := "pending" - if rejected { - state = "rejected" - } - return invalidCommitStatusf(op, "%s pieceCount %d does not match submission count %d", state, snapshot.PieceCount, expected) - } - if snapshot.PiecesAdded && len(snapshot.ConfirmedPieceIDs) != expected { - return invalidCommitStatusf(op, "confirmed piece ID count %d does not match submission count %d", len(snapshot.ConfirmedPieceIDs), expected) - } - return nil -} - func (c *contextCore) waitForCommit( ctx context.Context, op string, ref *DataSetRef, - submission CommitSubmission, + statusURL string, + clientDataSetID types.BigInt, ) (*CommitResult, error) { - validated, err := c.validateCommitSubmission(op, ref, submission, true) - if err != nil { + if err := validateRecoveryStatusURL(op, c.provider.ServiceURL, statusURL); err != nil { return nil, err } for { - status, err := c.getValidatedCommitStatus(ctx, op, ref, validated) + status, err := c.getValidatedCommitStatus(ctx, op, ref, statusURL, clientDataSetID) if err != nil { return nil, err } @@ -475,7 +514,7 @@ func (c *contextCore) waitForCommit( IsNewDataSet: status.Kind == CommitKindCreateAndAdd, }, nil case CommitStateRejected: - return nil, newCommitRejectedError(validated, *status) + return nil, newCommitRejectedError(c.provider.ID, *status) case CommitStatePending: select { case <-ctx.Done(): @@ -492,14 +531,10 @@ func (c *contextCore) validateCommitSubmission( op string, ref *DataSetRef, submission CommitSubmission, - callerSupplied bool, ) (CommitSubmission, error) { submission = copyCommitSubmission(submission) invalid := func(format string, args ...any) (CommitSubmission, error) { message := fmt.Sprintf(format, args...) - if callerSupplied { - return CommitSubmission{}, fmt.Errorf("%s: %w: %s", op, ErrInvalidArgument, message) - } return CommitSubmission{}, fmt.Errorf("%s: invalid provider submission: %s", op, message) } @@ -518,9 +553,6 @@ func (c *contextCore) validateCommitSubmission( } submission.TransactionID = common.HexToHash(submission.TransactionID).Hex() if err := validateProviderStatusURL(c.provider.ServiceURL, submission.StatusURL); err != nil { - if callerSupplied { - return CommitSubmission{}, fmt.Errorf("%s: %w: %w", op, ErrInvalidArgument, err) - } return CommitSubmission{}, fmt.Errorf("%s: %w", op, err) } if len(submission.PieceCIDs) == 0 || len(submission.PieceCIDs) > pdp.MaxAddPiecesBatchSize { @@ -578,11 +610,29 @@ func validateProviderStatusURL(serviceURL, statusURL string) error { !strings.EqualFold(base.Scheme, status.Scheme) || !strings.EqualFold(base.Hostname(), status.Hostname()) || effectiveStatusPort(base) != effectiveStatusPort(status) { - return fmt.Errorf("%w: provider status URL origin does not match service URL", pdp.ErrStatusURLOrigin) + return fmt.Errorf( + "%w: %w: provider status URL origin does not match service URL", + pdp.ErrInvalidStatusURL, + pdp.ErrStatusURLOrigin, + ) } return nil } +func validateRecoveryStatusURL(op, serviceURL, statusURL string) error { + if err := validateProviderStatusURL(serviceURL, statusURL); err != nil { + return fmt.Errorf("%s: %w: %w", op, ErrInvalidArgument, err) + } + return nil +} + +func wrapRecoveryStatusError(op, action string, err error) error { + if errors.Is(err, pdp.ErrInvalidStatusURL) { + return fmt.Errorf("%s: %w: %s: %w", op, ErrInvalidArgument, action, err) + } + return fmt.Errorf("%s: %s: %w", op, action, err) +} + func validStatusScheme(scheme string) bool { return strings.EqualFold(scheme, "http") || strings.EqualFold(scheme, "https") } @@ -646,9 +696,9 @@ func copyBigInts(in []types.BigInt) []types.BigInt { return out } -func newCommitRejectedError(submission CommitSubmission, status CommitStatus) *CommitRejectedError { +func newCommitRejectedError(providerID types.BigInt, status CommitStatus) *CommitRejectedError { return &CommitRejectedError{ - Submission: copyCommitSubmission(submission), + ProviderID: copyBigInt(providerID), Status: copyCommitStatus(status), } } diff --git a/storage/context_commit_test.go b/storage/context_commit_test.go index 1db42c6..873a895 100644 --- a/storage/context_commit_test.go +++ b/storage/context_commit_test.go @@ -3,7 +3,6 @@ package storage import ( "bytes" "context" - "encoding/json" "errors" "io" "math/big" @@ -21,7 +20,7 @@ import ( "github.com/strahe/synapse-go/types" ) -func TestAddCommitLifecycleCanResumeFromJSON(t *testing.T) { +func TestAddCommitLifecycleCanResumeFromStatusURL(t *testing.T) { pieceCID := mustPieceInfo(t).CIDv2 secondInfo, err := piece.CalculateFromBytes(bytes.Repeat([]byte("rho"), 128)) if err != nil { @@ -111,35 +110,19 @@ func TestAddCommitLifecycleCanResumeFromJSON(t *testing.T) { if submission.DataSet == nil || !submission.DataSet.Equal(ref) { t.Fatalf("callback mutated returned data set: %+v", submission.DataSet) } - callbackJSON, err := json.Marshal(callbackSubmission) - if err != nil { - t.Fatal(err) - } - var callbackRestored CommitSubmission - if err := json.Unmarshal(callbackJSON, &callbackRestored); err != nil { - t.Fatal(err) - } - if callbackRestored.TransactionID != submission.TransactionID || callbackRestored.DataSet == nil || !callbackRestored.DataSet.Equal(ref) { - t.Fatalf("callback JSON round trip=%+v", callbackRestored) + if callbackSubmission.TransactionID != submission.TransactionID || callbackSubmission.DataSet == nil || !callbackSubmission.DataSet.Equal(ref) { + t.Fatalf("callback submission=%+v", callbackSubmission) } if submission.Identity != (ContextIdentity{}) { t.Fatalf("identity=%+v want exact zero receiver identity", submission.Identity) } - persisted, err := json.Marshal(submission) - if err != nil { - t.Fatal(err) - } - var restored CommitSubmission - if err := json.Unmarshal(persisted, &restored); err != nil { - t.Fatal(err) - } validator := &fakeDataSetValidator{} fresh, err := NewDataSetContext(testProvider(), client, nil, ref, WithDataSetValidator(validator)) if err != nil { t.Fatal(err) } - pending, err := fresh.GetCommitStatus(context.Background(), restored) + pending, err := fresh.GetCommitStatus(context.Background(), submission.StatusURL) if err != nil { t.Fatal(err) } @@ -150,14 +133,14 @@ func TestAddCommitLifecycleCanResumeFromJSON(t *testing.T) { t.Fatalf("one-shot status calls=%d want 1", statusCalls) } - confirmed, err := fresh.GetCommitStatus(context.Background(), restored) + confirmed, err := fresh.GetCommitStatus(context.Background(), submission.StatusURL) if err != nil { t.Fatal(err) } if confirmed.State != CommitStateConfirmed || confirmed.ConfirmedTransactionID != confirmedTx.Hex() { t.Fatalf("confirmed=%+v", confirmed) } - result, err := fresh.WaitForCommit(context.Background(), restored) + result, err := fresh.WaitForCommit(context.Background(), submission.StatusURL) if err != nil { t.Fatal(err) } @@ -276,36 +259,6 @@ func TestSubmitCommitRejectsDuplicatePieceCIDBeforeDependencies(t *testing.T) { } } -func TestGetCommitStatusRejectsPersistedDuplicateCIDBeforeNetwork(t *testing.T) { - pieceCID := mustPieceInfo(t).CIDv2 - ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) - statusCalls := 0 - client := &fakePDPProviderClient{ - getAddedFn: func(context.Context, string) (*pdp.AddPiecesStatus, error) { - statusCalls++ - return nil, errors.New("unexpected status request") - }, - } - ctx, err := NewDataSetContext(testProvider(), client, nil, ref) - if err != nil { - t.Fatal(err) - } - submission := CommitSubmission{ - Kind: CommitKindAddPieces, - TransactionID: common.HexToHash("0x11").Hex(), - StatusURL: "https://sp.example.com/status/add", - ProviderID: testProvider().ID, - DataSet: &ref, - PieceCIDs: []cid.Cid{pieceCID, pieceCID}, - } - if _, err := ctx.GetCommitStatus(context.Background(), submission); !errors.Is(err, ErrInvalidArgument) || !strings.Contains(err.Error(), "duplicate pieceCID") { - t.Fatalf("error=%v want duplicate ErrInvalidArgument", err) - } - if statusCalls != 0 { - t.Fatalf("statusCalls=%d want 0", statusCalls) - } -} - func TestCreateAndAddExternalExtraDataExtractsIdentityBeforeNetwork(t *testing.T) { pieceCID := mustPieceInfo(t).CIDv2 clientDataSetID := types.NewBigInt(0) @@ -466,14 +419,14 @@ func TestCreateAndAddStatusHidesDataSetUntilOverallConfirmation(t *testing.T) { if err != nil { t.Fatal(err) } - pending, err := ctx.GetCreateAndAddStatus(context.Background(), *submission) + pending, err := ctx.GetCreateAndAddStatus(context.Background(), submission.StatusURL, *submission.ClientDataSetID) if err != nil { t.Fatal(err) } if pending.State != CommitStatePending || pending.DataSet != nil { t.Fatalf("pending exposed data set: %+v", pending) } - result, err := ctx.WaitForCreateAndAdd(context.Background(), *submission) + result, err := ctx.WaitForCreateAndAdd(context.Background(), submission.StatusURL, *submission.ClientDataSetID) if err != nil { t.Fatal(err) } @@ -516,11 +469,11 @@ func TestWaitForCommitReturnsTypedRejectionWithDefensiveSnapshots(t *testing.T) if err != nil { t.Fatal(err) } - status, err := ctx.GetCommitStatus(context.Background(), *submission) + status, err := ctx.GetCommitStatus(context.Background(), submission.StatusURL) if err != nil || status.State != CommitStateRejected || status.DataSet == nil || !status.DataSet.Equal(ref) { t.Fatalf("status=%+v error=%v", status, err) } - _, err = ctx.WaitForCommit(context.Background(), *submission) + _, err = ctx.WaitForCommit(context.Background(), submission.StatusURL) if !errors.Is(err, pdp.ErrTxRejected) { t.Fatalf("error=%v want ErrTxRejected", err) } @@ -531,14 +484,13 @@ func TestWaitForCommitReturnsTypedRejectionWithDefensiveSnapshots(t *testing.T) if strings.Contains(rejected.Error(), submission.StatusURL) || strings.Contains(rejected.Error(), "secret") { t.Fatalf("rejection error leaked status URL: %v", rejected) } - submission.PieceCIDs[0] = cid.Undef - if rejected.Submission.PieceCIDs[0] != pieceCID || rejected.Status.DataSet == nil || !rejected.Status.DataSet.Equal(ref) { + *status.DataSet = testDataSetRef(types.NewBigInt(99), types.NewBigInt(100)) + if !rejected.ProviderID.Equal(testProvider().ID) || rejected.Status.DataSet == nil || !rejected.Status.DataSet.Equal(ref) { t.Fatalf("rejection snapshots were not copied: %+v", rejected) } } -func TestRejectedCommitRejectsNonzeroPieceCountMismatch(t *testing.T) { - pieceCID := mustPieceInfo(t).CIDv2 +func TestRejectedCommitDoesNotRequireOriginalPieceCount(t *testing.T) { ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) originalTx := common.HexToHash("0x58") client := &fakePDPProviderClient{ @@ -556,21 +508,13 @@ func TestRejectedCommitRejectsNonzeroPieceCountMismatch(t *testing.T) { if err != nil { t.Fatal(err) } - submission := CommitSubmission{ - Kind: CommitKindAddPieces, - TransactionID: originalTx.Hex(), - StatusURL: "https://sp.example.com/status/rejected", - ProviderID: testProvider().ID, - DataSet: &ref, - PieceCIDs: []cid.Cid{pieceCID}, - } - if _, err := ctx.GetCommitStatus(context.Background(), submission); !errors.Is(err, pdp.ErrInvalidStatus) { - t.Fatalf("error=%v want ErrInvalidStatus", err) + status, err := ctx.GetCommitStatus(context.Background(), "https://sp.example.com/status/rejected") + if err != nil || status.State != CommitStateRejected { + t.Fatalf("status=%+v error=%v", status, err) } } func TestWaitForCommitTreatsReorgedSuccessSnapshotAsRejected(t *testing.T) { - pieceCID := mustPieceInfo(t).CIDv2 ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) originalTx := common.HexToHash("0x59") client := &fakePDPProviderClient{ @@ -590,19 +534,12 @@ func TestWaitForCommitTreatsReorgedSuccessSnapshotAsRejected(t *testing.T) { if err != nil { t.Fatal(err) } - submission := CommitSubmission{ - Kind: CommitKindAddPieces, - TransactionID: originalTx.Hex(), - StatusURL: "https://sp.example.com/status/reorged", - ProviderID: testProvider().ID, - DataSet: &ref, - PieceCIDs: []cid.Cid{pieceCID}, - } - status, err := ctx.GetCommitStatus(context.Background(), submission) + statusURL := "https://sp.example.com/status/reorged" + status, err := ctx.GetCommitStatus(context.Background(), statusURL) if err != nil || status.State != CommitStateRejected || len(status.PieceIDs) != 0 { t.Fatalf("status=%+v error=%v", status, err) } - _, err = ctx.WaitForCommit(context.Background(), submission) + _, err = ctx.WaitForCommit(context.Background(), statusURL) var rejected *CommitRejectedError if !errors.As(err, &rejected) || !errors.Is(err, pdp.ErrTxRejected) { t.Fatalf("error=%v want CommitRejectedError", err) @@ -610,7 +547,6 @@ func TestWaitForCommitTreatsReorgedSuccessSnapshotAsRejected(t *testing.T) { } func TestCreateStageRejectionDoesNotRequireAddSnapshot(t *testing.T) { - pieceCID := mustPieceInfo(t).CIDv2 originalTx := common.HexToHash("0x56") clientDataSetID := types.NewBigInt(0) client := &fakePDPProviderClient{ @@ -625,20 +561,12 @@ func TestCreateStageRejectionDoesNotRequireAddSnapshot(t *testing.T) { }, } ctx := mustWritableProviderContext(t, client) - submission := CommitSubmission{ - Kind: CommitKindCreateAndAdd, - TransactionID: originalTx.Hex(), - StatusURL: "https://sp.example.com/status/create", - ProviderID: testProvider().ID, - Identity: ctx.ContextIdentity(), - ClientDataSetID: &clientDataSetID, - PieceCIDs: []cid.Cid{pieceCID}, - } - status, err := ctx.GetCreateAndAddStatus(context.Background(), submission) + statusURL := "https://sp.example.com/status/create" + status, err := ctx.GetCreateAndAddStatus(context.Background(), statusURL, clientDataSetID) if err != nil || status.State != CommitStateRejected || status.DataSet != nil { t.Fatalf("status=%+v error=%v", status, err) } - _, err = ctx.WaitForCreateAndAdd(context.Background(), submission) + _, err = ctx.WaitForCreateAndAdd(context.Background(), statusURL, clientDataSetID) var rejected *CommitRejectedError if !errors.As(err, &rejected) || !errors.Is(err, pdp.ErrTxRejected) { t.Fatalf("error=%v", err) @@ -683,11 +611,11 @@ func TestWaitForCommitCanResumeAfterCancellationWithoutResubmitting(t *testing.T } cancelled, cancel := context.WithCancel(context.Background()) cancel() - if _, err := ctx.WaitForCommit(cancelled, *submission); !errors.Is(err, context.Canceled) { + if _, err := ctx.WaitForCommit(cancelled, submission.StatusURL); !errors.Is(err, context.Canceled) { t.Fatalf("cancelled wait error=%v", err) } confirmed = true - result, err := ctx.WaitForCommit(context.Background(), *submission) + result, err := ctx.WaitForCommit(context.Background(), submission.StatusURL) if err != nil { t.Fatal(err) } @@ -706,8 +634,11 @@ func TestCommitStatusSeparatesCallerAndProviderValidationErrors(t *testing.T) { addPiecesFn: func(context.Context, types.BigInt, []pdp.AddPieceInput, []byte) (*pdp.AddPiecesResult, error) { return &pdp.AddPiecesResult{TxHash: originalTx, StatusURL: "https://sp.example.com/status/add"}, nil }, - getAddedFn: func(context.Context, string) (*pdp.AddPiecesStatus, error) { + getAddedFn: func(_ context.Context, statusURL string) (*pdp.AddPiecesStatus, error) { statusCalls++ + if strings.HasSuffix(statusURL, "/not-a-hash") { + return nil, pdp.ErrInvalidStatusURL + } dataSetID := ref.DataSetID() pieceCount := 0 if wrongDataSet { @@ -736,28 +667,24 @@ func TestCommitStatusSeparatesCallerAndProviderValidationErrors(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := ctx.GetCommitStatus(context.Background(), *submission); !errors.Is(err, pdp.ErrInvalidStatus) || errors.Is(err, ErrInvalidArgument) { + if _, err := ctx.GetCommitStatus(context.Background(), submission.StatusURL); !errors.Is(err, pdp.ErrInvalidStatus) || errors.Is(err, ErrInvalidArgument) { t.Fatalf("provider mismatch error=%v", err) } wrongDataSet = false - if _, err := ctx.GetCommitStatus(context.Background(), *submission); !errors.Is(err, pdp.ErrInvalidStatus) || errors.Is(err, ErrInvalidArgument) { + if _, err := ctx.GetCommitStatus(context.Background(), submission.StatusURL); !errors.Is(err, pdp.ErrInvalidStatus) || errors.Is(err, ErrInvalidArgument) { t.Fatalf("missing confirmed pieceCount error=%v", err) } - invalid := copyCommitSubmission(*submission) - invalid.StatusURL = "https://other.example/status?token=secret" - if _, err := ctx.GetCommitStatus(context.Background(), invalid); !errors.Is(err, ErrInvalidArgument) || !errors.Is(err, pdp.ErrStatusURLOrigin) { + if _, err := ctx.GetCommitStatus(context.Background(), "https://other.example/status?token=secret"); !errors.Is(err, ErrInvalidArgument) || !errors.Is(err, pdp.ErrStatusURLOrigin) || !errors.Is(err, pdp.ErrInvalidStatusURL) { t.Fatalf("caller URL error=%v", err) } if statusCalls != 2 { - t.Fatalf("invalid caller submission reached network: statusCalls=%d", statusCalls) + t.Fatalf("invalid caller URL reached network: statusCalls=%d", statusCalls) } - invalid = copyCommitSubmission(*submission) - invalid.Identity.Payer = common.HexToAddress("0x9999") - if _, err := ctx.GetCommitStatus(context.Background(), invalid); !errors.Is(err, ErrInvalidArgument) { - t.Fatalf("caller identity error=%v", err) + if _, err := ctx.GetCommitStatus(context.Background(), "https://sp.example.com/status/not-a-hash"); !errors.Is(err, ErrInvalidArgument) || !errors.Is(err, pdp.ErrInvalidStatusURL) || errors.Is(err, pdp.ErrStatusURLOrigin) { + t.Fatalf("malformed caller URL error=%v", err) } - if statusCalls != 2 { - t.Fatalf("identity mismatch reached network: statusCalls=%d", statusCalls) + if statusCalls != 3 { + t.Fatalf("malformed caller URL statusCalls=%d want 3", statusCalls) } } @@ -915,7 +842,11 @@ func TestCreateAndAddCallbackHandleResumesAfterWaitFailure(t *testing.T) { } statusAvailable = true - result, err := mustWritableProviderContext(t, client).WaitForCreateAndAdd(context.Background(), saved) + result, err := mustWritableProviderContext(t, client).WaitForCreateAndAdd( + context.Background(), + saved.StatusURL, + *saved.ClientDataSetID, + ) if err != nil { t.Fatal(err) } @@ -943,14 +874,14 @@ func TestCommitLifecycleRejectsNilConcreteContexts(t *testing.T) { { name: "provider get", call: func() error { - _, err := providerContext.GetCreateAndAddStatus(ctx, CommitSubmission{}) + _, err := providerContext.GetCreateAndAddStatus(ctx, "", types.BigInt{}) return err }, }, { name: "provider wait", call: func() error { - _, err := providerContext.WaitForCreateAndAdd(ctx, CommitSubmission{}) + _, err := providerContext.WaitForCreateAndAdd(ctx, "", types.BigInt{}) return err }, }, @@ -964,14 +895,14 @@ func TestCommitLifecycleRejectsNilConcreteContexts(t *testing.T) { { name: "data set get", call: func() error { - _, err := dataSetContext.GetCommitStatus(ctx, CommitSubmission{}) + _, err := dataSetContext.GetCommitStatus(ctx, "") return err }, }, { name: "data set wait", call: func() error { - _, err := dataSetContext.WaitForCommit(ctx, CommitSubmission{}) + _, err := dataSetContext.WaitForCommit(ctx, "") return err }, }, @@ -1036,7 +967,11 @@ func TestProviderContextUploadCommitFailureKeepsSubmission(t *testing.T) { // A fresh context for the same provider resumes the kept submission. statusAvailable.Store(true) - result, err := mustWritableProviderContext(t, client).WaitForCreateAndAdd(context.Background(), *attempt.Submission) + result, err := mustWritableProviderContext(t, client).WaitForCreateAndAdd( + context.Background(), + attempt.Submission.StatusURL, + *attempt.Submission.ClientDataSetID, + ) if err != nil { t.Fatalf("WaitForCreateAndAdd: %v", err) } diff --git a/storage/context_dataset.go b/storage/context_dataset.go index e262c87..5f91bbb 100644 --- a/storage/context_dataset.go +++ b/storage/context_dataset.go @@ -28,13 +28,30 @@ func (c *ProviderContext) CreateDataSet(ctx context.Context, opts *CreateDataSet if opts != nil && opts.OnSubmitted != nil { opts.OnSubmitted(copyCreateDataSetSubmission(submission)) } - return c.waitForDataSetCreated(ctx, "storage.ProviderContext.CreateDataSet", submission) + return c.waitForDataSetCreated( + ctx, + "storage.ProviderContext.CreateDataSet", + submission.StatusURL, + submission.ClientDataSetID, + ) } // WaitForDataSetCreated waits for a previously submitted create-dataset -// transaction. The receiver remains unbound. -func (c *ProviderContext) WaitForDataSetCreated(ctx context.Context, submission CreateDataSetSubmission) (*CreateDataSetResult, error) { - return c.waitForDataSetCreated(ctx, "storage.ProviderContext.WaitForDataSetCreated", submission) +// status URL. ClientDataSetID may be zero. The receiver remains unbound. +func (c *ProviderContext) WaitForDataSetCreated( + ctx context.Context, + statusURL string, + clientDataSetID types.BigInt, +) (*CreateDataSetResult, error) { + if c == nil || c.core == nil { + return nil, fmt.Errorf("storage.ProviderContext.WaitForDataSetCreated: %w: nil context", ErrInvalidArgument) + } + return c.waitForDataSetCreated( + ctx, + "storage.ProviderContext.WaitForDataSetCreated", + statusURL, + clientDataSetID, + ) } // FindDataSetByClientDataSetID checks whether a data set created with the @@ -168,7 +185,7 @@ func (c *ProviderContext) submitCreateDataSet(ctx context.Context, requestedClie ProviderID: copyBigInt(c.core.provider.ID), TransactionID: created.TxHash.Hex(), StatusURL: created.StatusURL, - ClientDataSetID: copyClientDataSetIDPtr(clientDataSetID), + ClientDataSetID: copyBigInt(clientDataSetID), }, nil } @@ -215,15 +232,19 @@ func (c *ProviderContext) signCreateDataSet(ctx context.Context, op string, requ return extraData, clientDataSetID, nil } -func (c *ProviderContext) waitForDataSetCreated(ctx context.Context, op string, submission CreateDataSetSubmission) (*CreateDataSetResult, error) { - submission, err := validateCreateDataSetSubmission(op, c.core.provider, submission) - if err != nil { +func (c *ProviderContext) waitForDataSetCreated( + ctx context.Context, + op string, + statusURL string, + clientDataSetID types.BigInt, +) (*CreateDataSetResult, error) { + if err := validateRecoveryStatusURL(op, c.core.provider.ServiceURL, statusURL); err != nil { return nil, err } - status, err := c.core.client.WaitForDataSetCreated(ctx, submission.StatusURL, 0) + status, err := c.core.client.WaitForDataSetCreated(ctx, statusURL, 0) if err != nil { - return nil, fmt.Errorf("%s: wait dataset created: %w", op, err) + return nil, wrapRecoveryStatusError(op, "wait dataset created", err) } if status == nil { return nil, errors.New(op + ": wait dataset created returned nil status") @@ -231,70 +252,24 @@ func (c *ProviderContext) waitForDataSetCreated(ctx context.Context, op string, if status.DataSetID == nil || status.DataSetID.IsZero() { return nil, errors.New(op + ": server returned zero dataSetID") } - wantTransactionID := common.HexToHash(submission.TransactionID) - if got := status.CreateMessageHash; got != wantTransactionID { - return nil, fmt.Errorf( - "%s: %w: server returned mismatched transactionID: got %s want %s", - op, - pdp.ErrInvalidStatus, - got.Hex(), - wantTransactionID.Hex(), - ) + if status.CreateMessageHash == (common.Hash{}) { + return nil, fmt.Errorf("%s: %w: server returned zero transactionID", op, pdp.ErrInvalidStatus) } - ref, err := NewDataSetRef(c.core.provider.ID, *status.DataSetID, *submission.ClientDataSetID) + ref, err := NewDataSetRef(c.core.provider.ID, *status.DataSetID, clientDataSetID) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } return &CreateDataSetResult{ - TransactionID: submission.TransactionID, + TransactionID: status.CreateMessageHash.Hex(), ConfirmedTransactionID: optionalHashString(status.ConfirmedTxHash), DataSet: ref, }, nil } -func validateCreateDataSetSubmission(op string, provider Provider, submission CreateDataSetSubmission) (CreateDataSetSubmission, error) { - submission = copyCreateDataSetSubmission(submission) - if submission.ProviderID.IsZero() { - submission.ProviderID = copyBigInt(provider.ID) - } - if submission.ProviderID.IsZero() { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: zero providerID", op, ErrInvalidArgument) - } - if !submission.ProviderID.Equal(provider.ID) { - return CreateDataSetSubmission{}, fmt.Errorf( - "%s: %w: submission providerID %s does not match context providerID %s", - op, - ErrInvalidArgument, - submission.ProviderID.String(), - provider.ID.String(), - ) - } - if submission.TransactionID == "" { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: empty transactionID", op, ErrInvalidArgument) - } - if !common.IsHexHash(submission.TransactionID) { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: invalid transactionID %q", op, ErrInvalidArgument, submission.TransactionID) - } - wantTransactionID := common.HexToHash(submission.TransactionID) - if wantTransactionID == (common.Hash{}) { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: invalid transactionID %q", op, ErrInvalidArgument, submission.TransactionID) - } - if submission.StatusURL == "" { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: empty statusURL", op, ErrInvalidArgument) - } - if err := validateProviderStatusURL(provider.ServiceURL, submission.StatusURL); err != nil { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: %w", op, ErrInvalidArgument, err) - } - if submission.ClientDataSetID == nil { - return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: missing clientDataSetID", op, ErrInvalidArgument) - } - return submission, nil -} - func copyCreateDataSetSubmission(in CreateDataSetSubmission) CreateDataSetSubmission { out := in out.ProviderID = copyBigInt(in.ProviderID) - out.ClientDataSetID = copyBigIntPtr(in.ClientDataSetID) + out.ClientDataSetID = copyBigInt(in.ClientDataSetID) return out } diff --git a/storage/context_integration_test.go b/storage/context_integration_test.go index f4cb10d..bac3dcb 100644 --- a/storage/context_integration_test.go +++ b/storage/context_integration_test.go @@ -206,13 +206,13 @@ func TestIntegration_ContextCreateDataSetStagedFlow(t *testing.T) { if submission.StatusURL == "" { t.Fatal("CreateDataSet submission missing StatusURL") } - if submission.ClientDataSetID == nil || submission.ClientDataSetID.IsZero() { + if submission.ClientDataSetID.IsZero() { t.Fatal("CreateDataSet submission missing ClientDataSetID") } start = time.Now() t.Log("start storage staged WaitForDataSetCreated") - created, err := secondary.WaitForDataSetCreated(ctx, submission) + created, err := secondary.WaitForDataSetCreated(ctx, submission.StatusURL, submission.ClientDataSetID) t.Logf("done storage staged WaitForDataSetCreated elapsed=%s", time.Since(start).Round(time.Second)) if err != nil { t.Fatalf("WaitForDataSetCreated: %v", err) @@ -220,7 +220,7 @@ func TestIntegration_ContextCreateDataSetStagedFlow(t *testing.T) { if created.DataSet.DataSetID().IsZero() { t.Fatal("WaitForDataSetCreated returned zero DataSetID") } - if !created.DataSet.ClientDataSetID().Equal(*submission.ClientDataSetID) { + if !created.DataSet.ClientDataSetID().Equal(submission.ClientDataSetID) { t.Fatalf("ClientDataSetID mismatch: got %v want %v", created.DataSet.ClientDataSetID(), submission.ClientDataSetID) } recovered, err := secondary.ForDataSet(created.DataSet) diff --git a/storage/context_test.go b/storage/context_test.go index 812f080..6e4cd85 100644 --- a/storage/context_test.go +++ b/storage/context_test.go @@ -685,12 +685,12 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test if err != nil { t.Fatalf("CreateDataSet: %v", err) } - if !submission.ProviderID.Equal(testProvider().ID) || submission.ClientDataSetID == nil { + if !submission.ProviderID.Equal(testProvider().ID) { t.Fatalf("submission=%+v", submission) } if !result.DataSet.ProviderID().Equal(testProvider().ID) || !result.DataSet.DataSetID().Equal(dataSetID) || - !result.DataSet.ClientDataSetID().Equal(*submission.ClientDataSetID) || + !result.DataSet.ClientDataSetID().Equal(submission.ClientDataSetID) || result.ConfirmedTransactionID != confirmedTxHash.Hex() { t.Fatalf("result=%+v submission=%+v", result, submission) } @@ -699,7 +699,11 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test } fresh := mustWritableProviderContext(t, client) - recovered, err := fresh.WaitForDataSetCreated(context.Background(), submission) + recovered, err := fresh.WaitForDataSetCreated( + context.Background(), + submission.StatusURL, + submission.ClientDataSetID, + ) if err != nil { t.Fatalf("WaitForDataSetCreated: %v", err) } @@ -723,34 +727,10 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test } } -func TestProviderContextWaitForDataSetCreatedRejectsWrongProvider(t *testing.T) { - waitCalls := 0 - client := &fakePDPProviderClient{ - waitForCreatedFn: func(context.Context, string, time.Duration) (*pdp.CreateDataSetStatus, error) { - waitCalls++ - return nil, nil - }, - } - c := mustWritableProviderContext(t, client) - clientID := types.NewBigInt(7) - _, err := c.WaitForDataSetCreated(context.Background(), CreateDataSetSubmission{ - ProviderID: types.NewBigInt(2), - TransactionID: common.HexToHash("0x1234").Hex(), - StatusURL: "https://sp.example.com/status", - ClientDataSetID: &clientID, - }) - if !errors.Is(err, ErrInvalidArgument) || !strings.Contains(err.Error(), "providerID") { - t.Fatalf("WaitForDataSetCreated error=%v", err) - } - if waitCalls != 0 { - t.Fatalf("waitCalls=%d want 0", waitCalls) - } -} - -func TestProviderContextWaitForDataSetCreatedAcceptsZeroProviderID(t *testing.T) { +func TestProviderContextWaitForDataSetCreatedAcceptsZeroClientDataSetID(t *testing.T) { txHash := common.HexToHash("0x1234") dataSetID := types.NewBigInt(77) - clientID := types.NewBigInt(7) + clientID := types.NewBigInt(0) waitCalls := 0 client := &fakePDPProviderClient{ waitForCreatedFn: func(context.Context, string, time.Duration) (*pdp.CreateDataSetStatus, error) { @@ -760,11 +740,7 @@ func TestProviderContextWaitForDataSetCreatedAcceptsZeroProviderID(t *testing.T) }, } c := mustWritableProviderContext(t, client) - result, err := c.WaitForDataSetCreated(context.Background(), CreateDataSetSubmission{ - TransactionID: txHash.Hex(), - StatusURL: "https://sp.example.com/status", - ClientDataSetID: &clientID, - }) + result, err := c.WaitForDataSetCreated(context.Background(), "https://sp.example.com/status", clientID) if err != nil { t.Fatalf("WaitForDataSetCreated: %v", err) } @@ -778,26 +754,14 @@ func TestProviderContextWaitForDataSetCreatedAcceptsZeroProviderID(t *testing.T) } } -func TestProviderContextWaitForDataSetCreatedRejectsInvalidSubmission(t *testing.T) { +func TestProviderContextWaitForDataSetCreatedRejectsInvalidStatusURL(t *testing.T) { clientID := types.NewBigInt(7) - valid := CreateDataSetSubmission{ - TransactionID: common.HexToHash("0x1234").Hex(), - StatusURL: "https://sp.example.com/status", - ClientDataSetID: &clientID, - } - tests := map[string]CreateDataSetSubmission{ - "empty transaction": {TransactionID: "", StatusURL: valid.StatusURL, ClientDataSetID: valid.ClientDataSetID}, - "short transaction": {TransactionID: "0xbeef", StatusURL: valid.StatusURL, ClientDataSetID: valid.ClientDataSetID}, - "zero transaction": {TransactionID: common.Hash{}.Hex(), StatusURL: valid.StatusURL, ClientDataSetID: valid.ClientDataSetID}, - "empty status URL": {TransactionID: valid.TransactionID, StatusURL: "", ClientDataSetID: valid.ClientDataSetID}, - "missing client ID": {TransactionID: valid.TransactionID, StatusURL: valid.StatusURL}, - "cross-origin status URL": { - TransactionID: valid.TransactionID, - StatusURL: "https://other.example/status", - ClientDataSetID: valid.ClientDataSetID, - }, + tests := map[string]string{ + "empty status URL": "", + "relative status URL": "/status", + "cross-origin status URL": "https://other.example/status", } - for name, submission := range tests { + for name, statusURL := range tests { t.Run(name, func(t *testing.T) { waitCalls := 0 client := &fakePDPProviderClient{ @@ -807,7 +771,7 @@ func TestProviderContextWaitForDataSetCreatedRejectsInvalidSubmission(t *testing }, } c := mustWritableProviderContext(t, client) - _, err := c.WaitForDataSetCreated(context.Background(), submission) + _, err := c.WaitForDataSetCreated(context.Background(), statusURL, clientID) if !errors.Is(err, ErrInvalidArgument) { t.Fatalf("WaitForDataSetCreated error=%v want ErrInvalidArgument", err) } diff --git a/storage/context_upload.go b/storage/context_upload.go index 5f566ed..430b1d0 100644 --- a/storage/context_upload.go +++ b/storage/context_upload.go @@ -6,6 +6,7 @@ import ( "io" "github.com/strahe/synapse-go/internal/redact" + "github.com/strahe/synapse-go/types" ) // Upload stores a single copy and commits it to a new data set. opts may be @@ -118,7 +119,11 @@ func (c *contextCore) upload(ctx context.Context, op string, target StorageConte OnSubmitted: onCommitSubmitted, }}) if err == nil { - commit, err = c.waitForCommit(ctx, op, ref, *submission) + var clientDataSetID types.BigInt + if submission.ClientDataSetID != nil { + clientDataSetID = *submission.ClientDataSetID + } + commit, err = c.waitForCommit(ctx, op, ref, submission.StatusURL, clientDataSetID) } } else { task, enqueueErr := c.uploadBatcher.enqueue(ctx, reservation.seq, target, pieceInputs[0], transfer) diff --git a/storage/doc.go b/storage/doc.go index 9258866..92f7d4c 100644 --- a/storage/doc.go +++ b/storage/doc.go @@ -30,10 +30,10 @@ // its ordinary zero value, including nil pointer and map fields. Check // DataSetID and ProviderID when a complete identity is required. // -// Persistent data-set references, context identities, and create/commit -// lifecycle values use strict lowerCamel JSON field names. Alternate +// Persistent data-set references, context identities, and commit status and +// result values use strict lowerCamel JSON field names. Alternate // capitalization, unknown or duplicate fields, and incomplete objects are -// rejected. +// rejected. Submission values are runtime results, not persistence schemas. // // # Upload batching // @@ -212,22 +212,23 @@ // The ID is a correlation key, not an HTTP idempotency key; create POSTs are // sent once and are not automatically retried. // -// [ProviderContext.CreateDataSet] leaves its receiver unbound. Persist the -// [CreateDataSetSubmission] received through [CreateDataSetOptions.OnSubmitted] -// when confirmation must survive a restart. A fresh ProviderContext for the -// same provider can resume with [ProviderContext.WaitForDataSetCreated]. Pass -// the returned DataSetRef to [ProviderContext.ForDataSet] to obtain a -// DataSetContext. +// [ProviderContext.CreateDataSet] leaves its receiver unbound. When +// confirmation must survive a restart, persist StatusURL and ClientDataSetID +// from the [CreateDataSetSubmission] received through +// [CreateDataSetOptions.OnSubmitted]. A fresh ProviderContext for the same +// provider can pass those values to [ProviderContext.WaitForDataSetCreated]. +// Pass the returned DataSetRef to [ProviderContext.ForDataSet] to obtain a +// DataSetContext. Waiting does not require an FWSS reader. // // [ProviderContext.CreateAndAdd] and [DataSetContext.Commit] are convenience // methods that submit once and wait for confirmation. Their OnSubmitted // callback receives an independent, complete [CommitSubmission] after the -// provider handle is validated and before confirmation starts. Persisting that -// value allows a failed wait to resume on a fresh context for the same target. +// provider handle is validated and before confirmation starts. For +// create-and-add, persist StatusURL and ClientDataSetID. For add-pieces, +// persist StatusURL with the target DataSetRef. // For explicit recovery control, prefer SubmitCreateAndAdd followed by // WaitForCreateAndAdd on ProviderContext, or SubmitCommit followed by -// WaitForCommit on DataSetContext, and persist the returned submission between -// those calls. +// WaitForCommit on DataSetContext. // // GetCreateAndAddStatus and GetCommitStatus perform one logical status check // and return [CommitStatePending], [CommitStateConfirmed], or @@ -240,10 +241,12 @@ // [FailedAttempt] in [UploadResult.FailedAttempts] or // [CommitError.FailedAttempts] carries it as Submission. Resume a // create-and-add submission on [Service.NewProviderContext] for its ProviderID -// and call WaitForCreateAndAdd, or open an add-pieces submission's DataSet with -// [Service.NewDataSetContext] and call WaitForCommit. A batched submission can -// include other uploads' pieces; the result's PieceIDs follow the order of -// Submission.PieceCIDs. +// and call WaitForCreateAndAdd with its StatusURL and ClientDataSetID. For +// add-pieces, open its DataSet with [Service.NewDataSetContext] and call +// WaitForCommit with StatusURL. A batched submission can include other uploads' +// pieces. Applications that need a durable CID-to-piece-ID mapping must retain +// their original request order; generic recovery validates only that the +// provider's confirmed count matches its returned piece IDs. // // # Service termination // diff --git a/storage/errors.go b/storage/errors.go index 47f2432..4dff8f6 100644 --- a/storage/errors.go +++ b/storage/errors.go @@ -170,10 +170,10 @@ type CommitError struct { FailedAttempts []FailedAttempt } -// CommitRejectedError reports a valid terminal rejection while preserving the -// submission and final status needed for diagnostics. +// CommitRejectedError reports a valid terminal rejection with the provider and +// final status needed for diagnostics. type CommitRejectedError struct { - Submission CommitSubmission + ProviderID types.BigInt Status CommitStatus } @@ -183,9 +183,9 @@ func (e *CommitRejectedError) Error() string { } return fmt.Sprintf( "storage: %s commit %s for provider %s was rejected", - e.Submission.Kind, - e.Submission.TransactionID, - e.Submission.ProviderID.String(), + e.Status.Kind, + e.Status.TransactionID, + e.ProviderID.String(), ) } diff --git a/storage/lifecycle_json.go b/storage/lifecycle_json.go index 209728e..fe78302 100644 --- a/storage/lifecycle_json.go +++ b/storage/lifecycle_json.go @@ -95,28 +95,6 @@ func (i *ContextIdentity) UnmarshalJSON(data []byte) error { return nil } -func (s *CommitSubmission) UnmarshalJSON(data []byte) error { - if s == nil { - return fmt.Errorf("storage.CommitSubmission.UnmarshalJSON: %w: nil receiver", ErrInvalidArgument) - } - type wireSubmission CommitSubmission - var wire wireSubmission - if err := decodeLifecycleJSON(data, "storage.CommitSubmission.UnmarshalJSON", &wire, []lifecycleJSONField{ - {name: "kind"}, - {name: "transactionId"}, - {name: "statusUrl"}, - {name: "providerId"}, - {name: "identity"}, - {name: "dataSet", nullable: true}, - {name: "clientDataSetId", nullable: true}, - {name: "pieceCids"}, - }); err != nil { - return err - } - *s = copyCommitSubmission(CommitSubmission(wire)) - return nil -} - func (s *CommitStatus) UnmarshalJSON(data []byte) error { if s == nil { return fmt.Errorf("storage.CommitStatus.UnmarshalJSON: %w: nil receiver", ErrInvalidArgument) @@ -158,26 +136,6 @@ func (r *CommitResult) UnmarshalJSON(data []byte) error { return nil } -func (s *CreateDataSetSubmission) UnmarshalJSON(data []byte) error { - if s == nil { - return fmt.Errorf("storage.CreateDataSetSubmission.UnmarshalJSON: %w: nil receiver", ErrInvalidArgument) - } - type wireSubmission CreateDataSetSubmission - var wire wireSubmission - if err := decodeLifecycleJSON(data, "storage.CreateDataSetSubmission.UnmarshalJSON", &wire, []lifecycleJSONField{ - {name: "providerId"}, - {name: "transactionId"}, - {name: "statusUrl"}, - {name: "clientDataSetId"}, - }); err != nil { - return err - } - *s = CreateDataSetSubmission(wire) - s.ProviderID = copyBigInt(s.ProviderID) - s.ClientDataSetID = copyBigIntPtr(s.ClientDataSetID) - return nil -} - func (r *CreateDataSetResult) UnmarshalJSON(data []byte) error { if r == nil { return fmt.Errorf("storage.CreateDataSetResult.UnmarshalJSON: %w: nil receiver", ErrInvalidArgument) diff --git a/storage/lifecycle_json_test.go b/storage/lifecycle_json_test.go index 06861b0..1df9dc4 100644 --- a/storage/lifecycle_json_test.go +++ b/storage/lifecycle_json_test.go @@ -8,8 +8,6 @@ import ( "testing" "github.com/ethereum/go-ethereum/common" - "github.com/ipfs/go-cid" - "github.com/multiformats/go-multihash" "github.com/strahe/synapse-go/types" ) @@ -28,17 +26,6 @@ func TestLifecycleJSONUsesStrictLowerCamelFields(t *testing.T) { ChainID: types.ChainID(314159), RecordKeeper: common.HexToAddress("0x5678"), } - pieceCID := mustPieceInfo(t).CIDv2 - clientDataSetID := types.BigInt{} - commitSubmission := CommitSubmission{ - Kind: CommitKindCreateAndAdd, - TransactionID: common.HexToHash("0x11").Hex(), - StatusURL: "https://provider.example/status", - ProviderID: providerID, - Identity: identity, - ClientDataSetID: &clientDataSetID, - PieceCIDs: []cid.Cid{pieceCID}, - } commitStatus := CommitStatus{ Kind: CommitKindAddPieces, State: CommitStatePending, @@ -50,12 +37,6 @@ func TestLifecycleJSONUsesStrictLowerCamelFields(t *testing.T) { DataSet: ref, PieceIDs: []types.BigInt{types.NewBigInt(7)}, } - createSubmission := CreateDataSetSubmission{ - ProviderID: providerID, - TransactionID: common.HexToHash("0x22").Hex(), - StatusURL: "https://provider.example/create", - ClientDataSetID: &clientDataSetID, - } createResult := CreateDataSetResult{ TransactionID: common.HexToHash("0x22").Hex(), DataSet: ref, @@ -69,10 +50,8 @@ func TestLifecycleJSONUsesStrictLowerCamelFields(t *testing.T) { }{ {name: "data-set ref", value: ref, newTarget: func() any { return new(DataSetRef) }, pascalField: "providerId"}, {name: "context identity", value: identity, newTarget: func() any { return new(ContextIdentity) }, pascalField: "payer"}, - {name: "commit submission", value: commitSubmission, newTarget: func() any { return new(CommitSubmission) }, pascalField: "kind"}, {name: "commit status", value: commitStatus, newTarget: func() any { return new(CommitStatus) }, pascalField: "kind"}, {name: "commit result", value: commitResult, newTarget: func() any { return new(CommitResult) }, pascalField: "transactionId"}, - {name: "create submission", value: createSubmission, newTarget: func() any { return new(CreateDataSetSubmission) }, pascalField: "providerId"}, {name: "create result", value: createResult, newTarget: func() any { return new(CreateDataSetResult) }, pascalField: "transactionId"}, } for _, test := range tests { @@ -93,101 +72,6 @@ func TestLifecycleJSONUsesStrictLowerCamelFields(t *testing.T) { } } -func TestCommitSubmissionJSONContract(t *testing.T) { - pieceCID := mustPieceInfo(t).CIDv2 - v0Hash, err := multihash.Sum([]byte("cidv0"), multihash.SHA2_256, -1) - if err != nil { - t.Fatal(err) - } - pieceCIDv0 := cid.NewCidV0(v0Hash) - clientDataSetID := types.BigInt{} - submission := CommitSubmission{ - Kind: CommitKindCreateAndAdd, - TransactionID: common.HexToHash("0x11").Hex(), - StatusURL: "https://provider.example/status", - ProviderID: types.NewBigInt(7), - Identity: ContextIdentity{ - Payer: common.HexToAddress("0x1234"), - ChainID: types.ChainID(314159), - RecordKeeper: common.HexToAddress("0x5678"), - }, - ClientDataSetID: &clientDataSetID, - PieceCIDs: []cid.Cid{pieceCIDv0, pieceCID}, - } - encoded, err := json.Marshal(submission) - if err != nil { - t.Fatal(err) - } - var object map[string]json.RawMessage - if err := json.Unmarshal(encoded, &object); err != nil { - t.Fatal(err) - } - for _, field := range []string{ - "kind", "transactionId", "statusUrl", "providerId", "identity", - "dataSet", "clientDataSetId", "pieceCids", - } { - if _, ok := object[field]; !ok { - t.Fatalf("missing field %q in %s", field, encoded) - } - } - if len(object) != 8 { - t.Fatalf("fields=%v", object) - } - if string(object["dataSet"]) != "null" || string(object["clientDataSetId"]) != `"0"` { - t.Fatalf("nullable/zero fields dataSet=%s clientDataSetId=%s", object["dataSet"], object["clientDataSetId"]) - } - - var restored CommitSubmission - if err := json.Unmarshal(encoded, &restored); err != nil { - t.Fatal(err) - } - if restored.ClientDataSetID == nil || !restored.ClientDataSetID.IsZero() || - len(restored.PieceCIDs) != 2 || - !restored.PieceCIDs[0].Equals(pieceCIDv0) || restored.PieceCIDs[0].Version() != 0 || - !restored.PieceCIDs[1].Equals(pieceCID) || restored.PieceCIDs[1].Version() != pieceCID.Version() { - t.Fatalf("restored=%+v", restored) - } - - for name, mutate := range map[string]func(map[string]json.RawMessage){ - "missing": func(raw map[string]json.RawMessage) { delete(raw, "statusUrl") }, - "unknown": func(raw map[string]json.RawMessage) { raw["other"] = json.RawMessage(`1`) }, - "non-nullable null": func(raw map[string]json.RawMessage) { - raw["pieceCids"] = json.RawMessage(`null`) - }, - } { - t.Run(name, func(t *testing.T) { - var raw map[string]json.RawMessage - if err := json.Unmarshal(encoded, &raw); err != nil { - t.Fatal(err) - } - mutate(raw) - invalid, err := json.Marshal(raw) - if err != nil { - t.Fatal(err) - } - before := restored - if err := json.Unmarshal(invalid, &restored); !errors.Is(err, ErrInvalidArgument) { - t.Fatalf("error=%v want ErrInvalidArgument", err) - } - if !restored.ProviderID.Equal(before.ProviderID) || len(restored.PieceCIDs) != len(before.PieceCIDs) { - t.Fatal("failed decode mutated receiver") - } - }) - } - if err := json.Unmarshal(append(encoded, []byte(` {}`)...), &restored); err == nil { - t.Fatal("trailing JSON should be rejected") - } - duplicate := bytes.Replace( - encoded, - []byte(`"transactionId":`), - []byte(`"transactionId":"duplicate","transactionId":`), - 1, - ) - if err := json.Unmarshal(duplicate, &restored); !errors.Is(err, ErrInvalidArgument) { - t.Fatalf("duplicate field error=%v want ErrInvalidArgument", err) - } -} - func TestLifecycleJSONRejectsNilReceivers(t *testing.T) { tests := []struct { name string @@ -195,10 +79,8 @@ func TestLifecycleJSONRejectsNilReceivers(t *testing.T) { }{ {name: "data-set ref", call: func() error { return (*DataSetRef)(nil).UnmarshalJSON([]byte(`{}`)) }}, {name: "context identity", call: func() error { return (*ContextIdentity)(nil).UnmarshalJSON([]byte(`{}`)) }}, - {name: "commit submission", call: func() error { return (*CommitSubmission)(nil).UnmarshalJSON([]byte(`{}`)) }}, {name: "commit status", call: func() error { return (*CommitStatus)(nil).UnmarshalJSON([]byte(`{}`)) }}, {name: "commit result", call: func() error { return (*CommitResult)(nil).UnmarshalJSON([]byte(`{}`)) }}, - {name: "create submission", call: func() error { return (*CreateDataSetSubmission)(nil).UnmarshalJSON([]byte(`{}`)) }}, {name: "create result", call: func() error { return (*CreateDataSetResult)(nil).UnmarshalJSON([]byte(`{}`)) }}, } for _, test := range tests { @@ -209,16 +91,3 @@ func TestLifecycleJSONRejectsNilReceivers(t *testing.T) { }) } } - -func TestCommitSubmissionJSONRejectsMalformedTopLevelAndNestedValues(t *testing.T) { - var submission CommitSubmission - for _, data := range [][]byte{ - []byte(`{`), - []byte(`null`), - []byte(`{"kind":"addPieces","transactionId":"tx","statusUrl":"https://provider.example/status","providerId":"7","identity":"invalid","dataSet":null,"clientDataSetId":null,"pieceCids":[]}`), - } { - if err := submission.UnmarshalJSON(data); !errors.Is(err, ErrInvalidArgument) { - t.Fatalf("JSON=%s error=%v want ErrInvalidArgument", data, err) - } - } -} diff --git a/storage/types.go b/storage/types.go index e183e80..180f8a2 100644 --- a/storage/types.go +++ b/storage/types.go @@ -174,10 +174,10 @@ const ( CommitStateRejected CommitState = "rejected" ) -// CommitSubmission is a persistable handle returned after one successful -// provider submission. Persist all fields together and resume it with the same -// concrete context type. Its JSON form uses strict lowerCamel field names and -// rejects alternate capitalization. +// CommitSubmission describes one successful provider submission. It is a +// runtime result for callbacks and diagnostics, not a durable recovery record. +// Persist StatusURL to resume add-pieces. Create-and-add recovery also requires +// ClientDataSetID. type CommitSubmission struct { Kind CommitKind `json:"kind"` TransactionID string `json:"transactionId"` @@ -220,16 +220,14 @@ type CreateDataSetOptions struct { OnSubmitted func(CreateDataSetSubmission) } -// CreateDataSetSubmission identifies a submitted create-dataset transaction. -// Persist and restore all fields together using the strict lowerCamel JSON -// form; incomplete submissions and alternate capitalization are rejected. A -// zero ProviderID is filled from the ProviderContext used to wait. +// CreateDataSetSubmission describes a submitted create-dataset transaction. It +// is a runtime result for callbacks and diagnostics, not a durable recovery +// record. Persist StatusURL and ClientDataSetID to resume waiting. type CreateDataSetSubmission struct { - ProviderID types.BigInt `json:"providerId"` - TransactionID string `json:"transactionId"` - StatusURL string `json:"statusUrl"` - // ClientDataSetID must be non-nil when resuming a submitted create. - ClientDataSetID *types.BigInt `json:"clientDataSetId"` + ProviderID types.BigInt `json:"providerId"` + TransactionID string `json:"transactionId"` + StatusURL string `json:"statusUrl"` + ClientDataSetID types.BigInt `json:"clientDataSetId"` } // CreateDataSetResult is returned after standalone dataset creation confirms. @@ -257,9 +255,9 @@ type FailedAttempt struct { Err error Explicit bool // true when the provider was caller-specified (no auto-retry) // Submission is set when the provider accepted a commit submission for - // this attempt, even if confirmation later failed or timed out. Resume it - // with ProviderContext.WaitForCreateAndAdd or DataSetContext.WaitForCommit, - // according to its Kind; see Submission recovery in the package documentation. + // this attempt, even if confirmation later failed or timed out. Use its + // StatusURL, and ClientDataSetID for create-and-add, to resume waiting; see + // Submission recovery in the package documentation. Submission *CommitSubmission } diff --git a/storage/upload_batcher_test.go b/storage/upload_batcher_test.go index 9f97539..733ec7e 100644 --- a/storage/upload_batcher_test.go +++ b/storage/upload_batcher_test.go @@ -1606,7 +1606,8 @@ func TestServiceUploadPreservesConfirmedCopyWhenContextCancelsDuringSecondaryPul func TestUploadBatchContextErrorPreservesPublishedFailure(t *testing.T) { rejected := &CommitRejectedError{ - Submission: CommitSubmission{TransactionID: "0xrej", ProviderID: types.NewBigInt(1)}, + ProviderID: types.NewBigInt(1), + Status: CommitStatus{TransactionID: "0xrej"}, } canceled, cancel := context.WithCancel(context.Background()) cancel() diff --git a/tests/integration/delegated_storage_signer_test.go b/tests/integration/delegated_storage_signer_test.go index ae4514a..3b311af 100644 --- a/tests/integration/delegated_storage_signer_test.go +++ b/tests/integration/delegated_storage_signer_test.go @@ -331,7 +331,11 @@ func TestIntegration_DelegatedStorageSigner(t *testing.T) { cleanupDataSetID := dataSetID if cleanupDataSetID.IsZero() { waitCtx, waitCancel := context.WithTimeout(context.Background(), 3*time.Minute) - commitResult, waitErr := providerCtx.WaitForCreateAndAdd(waitCtx, *commitSubmission) + commitResult, waitErr := providerCtx.WaitForCreateAndAdd( + waitCtx, + commitSubmission.StatusURL, + *commitSubmission.ClientDataSetID, + ) waitCancel() if waitErr != nil { t.Errorf("cleanup WaitForCreateAndAdd(submission=%+v session=%s): %v", *commitSubmission, delegatedAddress, waitErr) @@ -367,7 +371,11 @@ func TestIntegration_DelegatedStorageSigner(t *testing.T) { } }) - commitResult, err := providerCtx.WaitForCreateAndAdd(ctx, *commitSubmission) + commitResult, err := providerCtx.WaitForCreateAndAdd( + ctx, + commitSubmission.StatusURL, + *commitSubmission.ClientDataSetID, + ) if err != nil { t.Fatalf("WaitForCreateAndAdd: %v", err) } From 8b9823a8a2a818db26dcbd744500d79d68f060cc Mon Sep 17 00:00:00 2001 From: Lee <7932644+strahe@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:33:26 +0800 Subject: [PATCH 2/3] fix(pdp/storage): enforce commit recovery invariants --- pdp/status.go | 9 +- pdp/status_test.go | 2 +- storage/callbacks_contract_test.go | 53 ++++++++ storage/context.go | 4 +- storage/context_commit.go | 52 ++++++-- storage/context_commit_test.go | 188 +++++++++++++++++++++++++++-- storage/context_upload.go | 7 +- 7 files changed, 283 insertions(+), 32 deletions(-) diff --git a/pdp/status.go b/pdp/status.go index b5896aa..dc22dba 100644 --- a/pdp/status.go +++ b/pdp/status.go @@ -130,6 +130,11 @@ func classifyAddPiecesStatus(op string, status *AddPiecesStatus) (transactionSta if status.PieceCount < 0 { return transactionPending, invalidStatusf(op, "negative pieceCount") } + if status.TxStatus == "reorged" { + // The transaction is no longer canonical. A provider can retain stage-local + // success fields without retaining the confirmed piece IDs. + return transactionRejected, nil + } if status.PiecesAdded && (status.AddMessageOK == nil || !*status.AddMessageOK) { return transactionPending, invalidStatusf(op, "piecesAdded without successful add message") } @@ -156,10 +161,6 @@ func classifyAddPiecesStatus(op string, status *AddPiecesStatus) (transactionSta return transactionPending, invalidStatusf(op, "pending response contains terminal fields") } return transactionPending, nil - case "reorged": - // A reorged transaction is no longer canonical. Stage-local success - // fields may still describe its pre-reorg result. - return transactionRejected, nil case "failed", "rejected": if (status.AddMessageOK != nil && *status.AddMessageOK) || status.PiecesAdded || len(status.ConfirmedPieceIDs) > 0 { return transactionPending, invalidStatusf(op, "%s response contains successful fields", status.TxStatus) diff --git a/pdp/status_test.go b/pdp/status_test.go index 839756e..65c0e49 100644 --- a/pdp/status_test.go +++ b/pdp/status_test.go @@ -50,7 +50,7 @@ func TestGetAddPiecesStatusNormalizesWireStates(t *testing.T) { }, { name: "reorged after success", - body: fmt.Sprintf(`{"txHash":%q,"txStatus":"reorged","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true,"confirmedPieceIds":[7]}`, testOriginalTx), + body: fmt.Sprintf(`{"txHash":%q,"txStatus":"reorged","dataSetId":5,"pieceCount":1,"addMessageOk":true,"piecesAdded":true}`, testOriginalTx), wantError: ErrTxRejected, }, { diff --git a/storage/callbacks_contract_test.go b/storage/callbacks_contract_test.go index 2ccb7ee..dbb8ec5 100644 --- a/storage/callbacks_contract_test.go +++ b/storage/callbacks_contract_test.go @@ -600,6 +600,59 @@ func TestContextUpload_Callbacks(t *testing.T) { } } +func TestContextUploadRejectsUnexpectedConfirmedPieceCount(t *testing.T) { + data := bytes.Repeat([]byte("count"), 64) + info, err := piece.CalculateFromBytes(data) + if err != nil { + t.Fatal(err) + } + client := &fakePDPProviderClient{ + uploadStreamingFn: func(_ context.Context, r io.Reader, _ pdp.UploadPieceStreamingOptions) (*pdp.UploadStreamingResult, error) { + _, _ = io.Copy(io.Discard, r) + return &pdp.UploadStreamingResult{PieceCID: info.CIDv2, Size: int64(len(data))}, nil + }, + waitForPieceFn: func(context.Context, cid.Cid, time.Duration) error { return nil }, + createAndAddFn: func(context.Context, common.Address, []pdp.AddPieceInput, []byte) (*pdp.CreateDataSetResult, error) { + return &pdp.CreateDataSetResult{ + TxHash: common.HexToHash("0xabc"), + StatusURL: "https://sp.example.com/status", + }, nil + }, + waitForCreateAndAddFn: func(context.Context, string, time.Duration) (*pdp.AddPiecesStatus, error) { + return &pdp.AddPiecesStatus{ + TxHash: common.HexToHash("0xabc"), + DataSetID: types.NewBigInt(55), + PiecesAdded: true, + ConfirmedPieceIDs: []types.BigInt{types.NewBigInt(77), types.NewBigInt(78)}, + }, nil + }, + } + ctx, err := NewProviderContext( + testProvider(), + client, + mustTestSigner(t), + WithPayer(testPayer()), + WithRecordKeeper(testRecordKeeper()), + WithChainID(types.ChainID(314159)), + ) + if err != nil { + t.Fatal(err) + } + confirmedCalls := 0 + result, err := ctx.Upload(context.Background(), bytes.NewReader(data), &ContextUploadOptions{ + OnPiecesConfirmed: func(types.BigInt, types.BigInt, []ConfirmedPiece) { + confirmedCalls++ + }, + }) + var commitErr *CommitError + if result != nil || !errors.As(err, &commitErr) || !errors.Is(err, pdp.ErrInvalidStatus) { + t.Fatalf("result=%+v error=%v want CommitError wrapping ErrInvalidStatus", result, err) + } + if confirmedCalls != 0 { + t.Fatalf("OnPiecesConfirmed calls=%d want 0", confirmedCalls) + } +} + func TestContextUpload_CallbacksAllowZeroPieceID(t *testing.T) { data := bytes.Repeat([]byte("cz"), 128) info, err := piece.CalculateFromBytes(data) diff --git a/storage/context.go b/storage/context.go index f4f2e75..9c01247 100644 --- a/storage/context.go +++ b/storage/context.go @@ -652,7 +652,7 @@ func (c *ProviderContext) CreateAndAdd(ctx context.Context, req CreateAndAddRequ if err != nil { return nil, err } - return c.WaitForCreateAndAdd(ctx, submission.StatusURL, *submission.ClientDataSetID) + return c.waitForCommit(ctx, *submission) } // Commit adds pieces to the bound data set and waits for confirmation. @@ -661,7 +661,7 @@ func (c *DataSetContext) Commit(ctx context.Context, req CommitRequest) (*Commit if err != nil { return nil, err } - return c.WaitForCommit(ctx, submission.StatusURL) + return c.waitForCommit(ctx, *submission) } func (c *contextCore) validateWritableDataSet(ctx context.Context, op string, ref *DataSetRef) error { diff --git a/storage/context_commit.go b/storage/context_commit.go index 1abeb68..ae3209c 100644 --- a/storage/context_commit.go +++ b/storage/context_commit.go @@ -95,15 +95,11 @@ func (c *ProviderContext) waitForCommit(ctx context.Context, submission CommitSu if c == nil || c.core == nil { return nil, fmt.Errorf("storage.ProviderContext.WaitForCreateAndAdd: %w: nil context", ErrInvalidArgument) } - if submission.ClientDataSetID == nil { - return nil, errors.New("storage.ProviderContext.WaitForCreateAndAdd: invalid provider submission: missing clientDataSetID") - } - return c.core.waitForCommit( + return c.core.waitForSubmission( ctx, "storage.ProviderContext.WaitForCreateAndAdd", nil, - submission.StatusURL, - *submission.ClientDataSetID, + submission, ) } @@ -158,12 +154,11 @@ func (c *DataSetContext) waitForCommit(ctx context.Context, submission CommitSub if c == nil || c.core == nil { return nil, fmt.Errorf("storage.DataSetContext.WaitForCommit: %w: nil context", ErrInvalidArgument) } - return c.core.waitForCommit( + return c.core.waitForSubmission( ctx, "storage.DataSetContext.WaitForCommit", &c.ref, - submission.StatusURL, - types.BigInt{}, + submission, ) } @@ -398,7 +393,7 @@ func (c *contextCore) getAddPiecesCommitStatus( if snapshot.TxHash == (common.Hash{}) { return nil, invalidCommitStatusf(op, "zero transactionID") } - if snapshot.PiecesAdded && (snapshot.PieceCount <= 0 || snapshot.PieceCount != len(snapshot.ConfirmedPieceIDs)) { + if !rejected && snapshot.PiecesAdded && (snapshot.PieceCount <= 0 || snapshot.PieceCount != len(snapshot.ConfirmedPieceIDs)) { return nil, invalidCommitStatusf(op, "confirmed piece counts differ") } @@ -453,7 +448,7 @@ func (c *contextCore) getCreateAndAddCommitStatus( if snapshot.Add.ConfirmedTxHash != (common.Hash{}) { confirmedHash = snapshot.Add.ConfirmedTxHash } - if snapshot.Add.PiecesAdded && + if !rejected && snapshot.Add.PiecesAdded && (snapshot.Add.PieceCount <= 0 || snapshot.Add.PieceCount != len(snapshot.Add.ConfirmedPieceIDs)) { return nil, invalidCommitStatusf(op, "confirmed piece counts differ") } @@ -527,6 +522,35 @@ func (c *contextCore) waitForCommit( } } +func (c *contextCore) waitForSubmission( + ctx context.Context, + op string, + ref *DataSetRef, + submission CommitSubmission, +) (*CommitResult, error) { + validated, err := c.validateCommitSubmission(op, ref, submission) + if err != nil { + return nil, err + } + var clientDataSetID types.BigInt + if validated.ClientDataSetID != nil { + clientDataSetID = *validated.ClientDataSetID + } + result, err := c.waitForCommit(ctx, op, ref, validated.StatusURL, clientDataSetID) + if err != nil { + return nil, err + } + if len(result.PieceIDs) != len(validated.PieceCIDs) { + return nil, invalidCommitStatusf( + op, + "confirmed piece ID count %d does not match submission count %d", + len(result.PieceIDs), + len(validated.PieceCIDs), + ) + } + return result, nil +} + func (c *contextCore) validateCommitSubmission( op string, ref *DataSetRef, @@ -627,7 +651,11 @@ func validateRecoveryStatusURL(op, serviceURL, statusURL string) error { } func wrapRecoveryStatusError(op, action string, err error) error { - if errors.Is(err, pdp.ErrInvalidStatusURL) { + // The PDP client validates the original URL before issuing HTTP. An origin + // error inside url.Error therefore comes from redirect handling. + _, requestErr := errors.AsType[*url.Error](err) + providerRedirect := requestErr && errors.Is(err, pdp.ErrStatusURLOrigin) + if errors.Is(err, pdp.ErrInvalidStatusURL) && !providerRedirect { return fmt.Errorf("%s: %w: %s: %w", op, ErrInvalidArgument, action, err) } return fmt.Errorf("%s: %s: %w", op, action, err) diff --git a/storage/context_commit_test.go b/storage/context_commit_test.go index 873a895..aa0146e 100644 --- a/storage/context_commit_test.go +++ b/storage/context_commit_test.go @@ -6,6 +6,8 @@ import ( "errors" "io" "math/big" + "net/http" + "net/http/httptest" "strings" "sync/atomic" "testing" @@ -159,6 +161,105 @@ func TestAddCommitLifecycleCanResumeFromStatusURL(t *testing.T) { } } +func TestSynchronousCommitRejectsConfirmedPieceCountMismatch(t *testing.T) { + firstPieceCID := mustPieceInfo(t).CIDv2 + secondInfo, err := piece.CalculateFromBytes(bytes.Repeat([]byte("count-mismatch"), 32)) + if err != nil { + t.Fatal(err) + } + pieces := []PieceInput{{PieceCID: firstPieceCID}, {PieceCID: secondInfo.CIDv2}} + originalTx := common.HexToHash("0x31") + dataSetID := types.NewBigInt(42) + confirmedPieceIDs := []types.BigInt{types.NewBigInt(8)} + + t.Run("add pieces", func(t *testing.T) { + ref := testDataSetRef(dataSetID, types.NewBigInt(7)) + client := &fakePDPProviderClient{ + addPiecesFn: func(context.Context, types.BigInt, []pdp.AddPieceInput, []byte) (*pdp.AddPiecesResult, error) { + return &pdp.AddPiecesResult{TxHash: originalTx, StatusURL: "https://sp.example.com/status/add"}, nil + }, + getAddedFn: func(context.Context, string) (*pdp.AddPiecesStatus, error) { + return &pdp.AddPiecesStatus{ + TxHash: originalTx, + TxStatus: "confirmed", + DataSetID: dataSetID, + PieceCount: 1, + AddMessageOK: new(true), + PiecesAdded: true, + ConfirmedPieceIDs: confirmedPieceIDs, + }, nil + }, + } + ctx, err := NewDataSetContext(testProvider(), client, nil, ref) + if err != nil { + t.Fatal(err) + } + result, err := ctx.Commit(context.Background(), CommitRequest{Pieces: pieces, ExtraData: []byte{1}}) + if result != nil || !errors.Is(err, pdp.ErrInvalidStatus) { + t.Fatalf("result=%+v error=%v want ErrInvalidStatus", result, err) + } + }) + + t.Run("create and add", func(t *testing.T) { + client := &fakePDPProviderClient{ + createAndAddFn: func(context.Context, common.Address, []pdp.AddPieceInput, []byte) (*pdp.CreateDataSetResult, error) { + return &pdp.CreateDataSetResult{TxHash: originalTx, StatusURL: "https://sp.example.com/status/create"}, nil + }, + getCreateAndAddFn: func(context.Context, string) (*pdp.CreateAndAddPiecesStatus, error) { + createdDataSetID := dataSetID.Copy() + return &pdp.CreateAndAddPiecesStatus{ + Create: &pdp.CreateDataSetStatus{ + CreateMessageHash: originalTx, + TxStatus: "confirmed", + DataSetCreated: true, + OK: new(true), + DataSetID: &createdDataSetID, + }, + Add: &pdp.AddPiecesStatus{ + TxHash: originalTx, + TxStatus: "confirmed", + DataSetID: dataSetID, + PieceCount: 1, + AddMessageOK: new(true), + PiecesAdded: true, + ConfirmedPieceIDs: confirmedPieceIDs, + }, + }, nil + }, + } + result, err := mustWritableProviderContext(t, client).CreateAndAdd(context.Background(), CreateAndAddRequest{Pieces: pieces}) + if result != nil || !errors.Is(err, pdp.ErrInvalidStatus) { + t.Fatalf("result=%+v error=%v want ErrInvalidStatus", result, err) + } + }) +} + +func TestURLOnlyCommitRecoveryAcceptsSelfConsistentPieceCount(t *testing.T) { + ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) + originalTx := common.HexToHash("0x32") + client := &fakePDPProviderClient{ + getAddedFn: func(context.Context, string) (*pdp.AddPiecesStatus, error) { + return &pdp.AddPiecesStatus{ + TxHash: originalTx, + TxStatus: "confirmed", + DataSetID: ref.DataSetID(), + PieceCount: 1, + AddMessageOK: new(true), + PiecesAdded: true, + ConfirmedPieceIDs: []types.BigInt{types.NewBigInt(8)}, + }, nil + }, + } + ctx, err := NewDataSetContext(testProvider(), client, nil, ref) + if err != nil { + t.Fatal(err) + } + result, err := ctx.WaitForCommit(context.Background(), "https://sp.example.com/status/add") + if err != nil || len(result.PieceIDs) != 1 { + t.Fatalf("result=%+v error=%v", result, err) + } +} + func TestContextRejectsOversizedExternalPayloadBeforeProviderCall(t *testing.T) { pieceCID := mustPieceInfo(t).CIDv2 providerCalls := 0 @@ -520,13 +621,12 @@ func TestWaitForCommitTreatsReorgedSuccessSnapshotAsRejected(t *testing.T) { client := &fakePDPProviderClient{ getAddedFn: func(context.Context, string) (*pdp.AddPiecesStatus, error) { return &pdp.AddPiecesStatus{ - TxHash: originalTx, - TxStatus: "reorged", - DataSetID: ref.DataSetID(), - PieceCount: 1, - AddMessageOK: new(true), - PiecesAdded: true, - ConfirmedPieceIDs: []types.BigInt{types.NewBigInt(8)}, + TxHash: originalTx, + TxStatus: "reorged", + DataSetID: ref.DataSetID(), + PieceCount: 1, + AddMessageOK: new(true), + PiecesAdded: true, }, pdp.ErrTxRejected }, } @@ -546,6 +646,45 @@ func TestWaitForCommitTreatsReorgedSuccessSnapshotAsRejected(t *testing.T) { } } +func TestWaitForCreateAndAddTreatsAddReorgedSuccessSnapshotAsRejected(t *testing.T) { + originalTx := common.HexToHash("0x5a") + dataSetID := types.NewBigInt(42) + clientDataSetID := types.NewBigInt(7) + client := &fakePDPProviderClient{ + getCreateAndAddFn: func(context.Context, string) (*pdp.CreateAndAddPiecesStatus, error) { + createdDataSetID := dataSetID.Copy() + return &pdp.CreateAndAddPiecesStatus{ + Create: &pdp.CreateDataSetStatus{ + CreateMessageHash: originalTx, + TxStatus: "confirmed", + DataSetCreated: true, + OK: new(true), + DataSetID: &createdDataSetID, + }, + Add: &pdp.AddPiecesStatus{ + TxHash: originalTx, + TxStatus: "reorged", + DataSetID: dataSetID, + PieceCount: 1, + AddMessageOK: new(true), + PiecesAdded: true, + }, + }, pdp.ErrTxRejected + }, + } + ctx := mustWritableProviderContext(t, client) + statusURL := "https://sp.example.com/status/create" + status, err := ctx.GetCreateAndAddStatus(context.Background(), statusURL, clientDataSetID) + if err != nil || status.State != CommitStateRejected || len(status.PieceIDs) != 0 { + t.Fatalf("status=%+v error=%v", status, err) + } + _, err = ctx.WaitForCreateAndAdd(context.Background(), statusURL, clientDataSetID) + var rejected *CommitRejectedError + if !errors.As(err, &rejected) || !errors.Is(err, pdp.ErrTxRejected) { + t.Fatalf("error=%v want CommitRejectedError", err) + } +} + func TestCreateStageRejectionDoesNotRequireAddSnapshot(t *testing.T) { originalTx := common.HexToHash("0x56") clientDataSetID := types.NewBigInt(0) @@ -688,6 +827,41 @@ func TestCommitStatusSeparatesCallerAndProviderValidationErrors(t *testing.T) { } } +func TestCommitStatusDoesNotClassifyProviderRedirectAsInvalidArgument(t *testing.T) { + targetRequests := 0 + target := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + targetRequests++ + })) + t.Cleanup(target.Close) + + source := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL+r.URL.Path+"?token=secret", http.StatusFound) + })) + t.Cleanup(source.Close) + client, err := pdp.New(source.URL, pdp.WithHTTPClient(source.Client())) + if err != nil { + t.Fatal(err) + } + provider := testProvider() + provider.ServiceURL = source.URL + ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) + ctx, err := NewDataSetContext(provider, client, nil, ref) + if err != nil { + t.Fatal(err) + } + statusURL := source.URL + "/status/" + common.HexToHash("0x78").Hex() + status, err := ctx.GetCommitStatus(context.Background(), statusURL) + if status != nil || !errors.Is(err, pdp.ErrInvalidStatusURL) || !errors.Is(err, pdp.ErrStatusURLOrigin) { + t.Fatalf("status=%+v error=%v want PDP status URL errors", status, err) + } + if errors.Is(err, ErrInvalidArgument) { + t.Fatalf("provider redirect error matched ErrInvalidArgument: %v", err) + } + if targetRequests != 0 { + t.Fatalf("targetRequests=%d want 0", targetRequests) + } +} + func TestSubmitCommitDoesNotInvokeCallbackWithoutValidProviderHandle(t *testing.T) { pieceCID := mustPieceInfo(t).CIDv2 ref := testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)) diff --git a/storage/context_upload.go b/storage/context_upload.go index 430b1d0..9777460 100644 --- a/storage/context_upload.go +++ b/storage/context_upload.go @@ -6,7 +6,6 @@ import ( "io" "github.com/strahe/synapse-go/internal/redact" - "github.com/strahe/synapse-go/types" ) // Upload stores a single copy and commits it to a new data set. opts may be @@ -119,11 +118,7 @@ func (c *contextCore) upload(ctx context.Context, op string, target StorageConte OnSubmitted: onCommitSubmitted, }}) if err == nil { - var clientDataSetID types.BigInt - if submission.ClientDataSetID != nil { - clientDataSetID = *submission.ClientDataSetID - } - commit, err = c.waitForCommit(ctx, op, ref, submission.StatusURL, clientDataSetID) + commit, err = c.waitForSubmission(ctx, op, ref, *submission) } } else { task, enqueueErr := c.uploadBatcher.enqueue(ctx, reservation.seq, target, pieceInputs[0], transfer) From 0a72e02364941ea7636c24a763763d016a31a6b4 Mon Sep 17 00:00:00 2001 From: Lee <7932644+strahe@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:12:54 +0800 Subject: [PATCH 3/3] fix(storage): clarify recovery state contract --- docs/GETTING_STARTED.md | 17 +++++---- storage/context_commit.go | 6 ++-- storage/context_dataset.go | 5 +-- storage/context_test.go | 73 ++++++++++++++++++++++++-------------- storage/doc.go | 10 +++--- storage/types.go | 20 ++++++----- 6 files changed, 81 insertions(+), 50 deletions(-) diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 13dc9eb..23dcb5a 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -471,9 +471,9 @@ if err != nil { fmt.Println("dataset:", ref.DataSetID()) ``` -To create an empty dataset first, save the status URL and client dataset ID if -the process may restart before confirmation. Creation is available only on -`ProviderContext`. +To create an empty dataset first, save the status URL and original client +dataset ID if the process may restart before confirmation. Creation is +available only on `ProviderContext`. ```go var statusURL string @@ -494,7 +494,9 @@ fmt.Println("dataset:", created.DataSet.DataSetID()) Resume a submitted create transaction with any fresh `ProviderContext` for the same provider, then convert the returned reference without mutating that -context: +context. Pass the exact client dataset ID used for the original submission; +zero is valid only if that original ID was zero. The status URL alone cannot +recover a lost client dataset ID: ```go created, err := providerCtx.WaitForDataSetCreated(ctx, statusURL, clientDataSetID) @@ -527,7 +529,9 @@ before confirmation begins. This value contains runtime and diagnostic data; it is not a persistence schema. Save only the recovery fields needed by the operation. -Create-and-add requires the status URL and client dataset ID: +Create-and-add requires the status URL and original client dataset ID. Pass +zero only if the original submission used zero; the status URL alone cannot +recover a lost client dataset ID: ```go var statusURL string @@ -580,7 +584,8 @@ submitted, err := providerCtx.SubmitCreateAndAdd(ctx, storage.CreateAndAddReques if err != nil { return err } -// Persist submitted.StatusURL and *submitted.ClientDataSetID before waiting. +// Persist submitted.StatusURL and the original +// *submitted.ClientDataSetID before waiting. result, err := providerCtx.WaitForCreateAndAdd( ctx, submitted.StatusURL, diff --git a/storage/context_commit.go b/storage/context_commit.go index ae3209c..7369a06 100644 --- a/storage/context_commit.go +++ b/storage/context_commit.go @@ -52,7 +52,8 @@ func (c *ProviderContext) submitCommit(ctx context.Context, req commitRequest) ( // GetCreateAndAddStatus checks a create-and-add status URL once. An invalid // status URL returns an error matching [ErrInvalidArgument] and // [pdp.ErrInvalidStatusURL]. An origin mismatch also matches -// [pdp.ErrStatusURLOrigin]. ClientDataSetID may be zero. +// [pdp.ErrStatusURLOrigin]. ClientDataSetID must be the value used for the +// original submission; zero is valid only when that original value was zero. func (c *ProviderContext) GetCreateAndAddStatus( ctx context.Context, statusURL string, @@ -73,7 +74,8 @@ func (c *ProviderContext) GetCreateAndAddStatus( // WaitForCreateAndAdd waits for a create-and-add status URL to confirm or // reject. An invalid status URL returns an error matching [ErrInvalidArgument] // and [pdp.ErrInvalidStatusURL]. An origin mismatch also matches -// [pdp.ErrStatusURLOrigin]. ClientDataSetID may be zero. +// [pdp.ErrStatusURLOrigin]. ClientDataSetID must be the value used for the +// original submission; zero is valid only when that original value was zero. func (c *ProviderContext) WaitForCreateAndAdd( ctx context.Context, statusURL string, diff --git a/storage/context_dataset.go b/storage/context_dataset.go index 5f91bbb..3f74ac7 100644 --- a/storage/context_dataset.go +++ b/storage/context_dataset.go @@ -36,8 +36,9 @@ func (c *ProviderContext) CreateDataSet(ctx context.Context, opts *CreateDataSet ) } -// WaitForDataSetCreated waits for a previously submitted create-dataset -// status URL. ClientDataSetID may be zero. The receiver remains unbound. +// WaitForDataSetCreated waits for a previously submitted create-dataset status +// URL. ClientDataSetID must be the value used for the original submission; zero +// is valid only when that original value was zero. The receiver remains unbound. func (c *ProviderContext) WaitForDataSetCreated( ctx context.Context, statusURL string, diff --git a/storage/context_test.go b/storage/context_test.go index 6e4cd85..c59febb 100644 --- a/storage/context_test.go +++ b/storage/context_test.go @@ -661,15 +661,24 @@ func TestProviderContextConcurrentCommitsCreateIndependently(t *testing.T) { } } -func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *testing.T) { +func TestProviderContextCreateDataSetRecoversOriginalZeroClientDataSetIDForSigning(t *testing.T) { txHash := common.HexToHash("0x1234") confirmedTxHash := common.HexToHash("0x5678") dataSetID := types.NewBigInt(77) + originalClientDataSetID := types.NewBigInt(0) + storageSigner := mustTestSigner(t) client := &fakePDPProviderClient{ createDataSetFn: func(_ context.Context, recordKeeper common.Address, extraData []byte) (*pdp.CreateDataSetResult, error) { if recordKeeper != testRecordKeeper() || len(extraData) == 0 { t.Fatalf("CreateDataSet recordKeeper=%s extraData=%x", recordKeeper, extraData) } + values, err := createDataSetArgs.Unpack(extraData) + if err != nil { + t.Fatalf("unpack create payload: %v", err) + } + if values[1].(*big.Int).Sign() != 0 { + t.Fatalf("submitted clientDataSetID=%s want explicit zero", values[1].(*big.Int)) + } return &pdp.CreateDataSetResult{TxHash: txHash, StatusURL: "https://sp.example.com/create/1234"}, nil }, waitForCreatedFn: func(context.Context, string, time.Duration) (*pdp.CreateDataSetStatus, error) { @@ -677,15 +686,30 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test return &pdp.CreateDataSetStatus{CreateMessageHash: txHash, ConfirmedTxHash: confirmedTxHash, DataSetID: &id}, nil }, } - providerCtx := mustWritableProviderContext(t, client) + newContext := func() *ProviderContext { + providerContext, err := NewProviderContext( + testProvider(), + client, + storageSigner, + WithPayer(testPayer()), + WithRecordKeeper(testRecordKeeper()), + WithChainID(types.ChainID(314159)), + ) + if err != nil { + t.Fatalf("NewProviderContext: %v", err) + } + return providerContext + } + providerCtx := newContext() var submission CreateDataSetSubmission result, err := providerCtx.CreateDataSet(context.Background(), &CreateDataSetOptions{ - OnSubmitted: func(got CreateDataSetSubmission) { submission = got }, + ClientDataSetID: &originalClientDataSetID, + OnSubmitted: func(got CreateDataSetSubmission) { submission = got }, }) if err != nil { t.Fatalf("CreateDataSet: %v", err) } - if !submission.ProviderID.Equal(testProvider().ID) { + if !submission.ProviderID.Equal(testProvider().ID) || !submission.ClientDataSetID.IsZero() { t.Fatalf("submission=%+v", submission) } if !result.DataSet.ProviderID().Equal(testProvider().ID) || @@ -698,7 +722,7 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test t.Fatal("CreateDataSet mutated ProviderContext") } - fresh := mustWritableProviderContext(t, client) + fresh := newContext() recovered, err := fresh.WaitForDataSetCreated( context.Background(), submission.StatusURL, @@ -725,32 +749,27 @@ func TestProviderContextCreateDataSetReturnsRecoverableRefWithoutBinding(t *test !ref.ClientDataSetID().Equal(recovered.DataSet.ClientDataSetID()) { t.Fatalf("bound ref=(%+v, %t) want %+v", ref, ok, recovered.DataSet) } -} - -func TestProviderContextWaitForDataSetCreatedAcceptsZeroClientDataSetID(t *testing.T) { - txHash := common.HexToHash("0x1234") - dataSetID := types.NewBigInt(77) - clientID := types.NewBigInt(0) - waitCalls := 0 - client := &fakePDPProviderClient{ - waitForCreatedFn: func(context.Context, string, time.Duration) (*pdp.CreateDataSetStatus, error) { - waitCalls++ - id := copyBigInt(dataSetID) - return &pdp.CreateDataSetStatus{CreateMessageHash: txHash, DataSetID: &id}, nil - }, + pieceInfo := mustPieceInfo(t) + payload, err := bound.PresignForCommit(context.Background(), []PieceInput{{PieceCID: pieceInfo.CIDv2}}) + if err != nil { + t.Fatalf("PresignForCommit: %v", err) } - c := mustWritableProviderContext(t, client) - result, err := c.WaitForDataSetCreated(context.Background(), "https://sp.example.com/status", clientID) + values, err := addPiecesArgs.Unpack(payload) if err != nil { - t.Fatalf("WaitForDataSetCreated: %v", err) + t.Fatalf("unpack add-pieces payload: %v", err) } - if waitCalls != 1 { - t.Fatalf("waitCalls=%d want 1", waitCalls) + message, err := ityped.AddPiecesMessage( + originalClientDataSetID.Big(), + values[0].(*big.Int), + []cid.Cid{pieceInfo.CIDv2}, + nil, + ) + if err != nil { + t.Fatalf("build AddPieces message: %v", err) } - if !result.DataSet.ProviderID().Equal(testProvider().ID) || - !result.DataSet.DataSetID().Equal(dataSetID) || - !result.DataSet.ClientDataSetID().Equal(clientID) { - t.Fatalf("result=%+v", result.DataSet) + domain := ityped.NewDomain(big.NewInt(314159), testRecordKeeper()) + if recovered := recoverRawTypedDataSigner(t, domain, "AddPieces", message, values[3].([]byte)); recovered != storageSigner.EVMAddress() { + t.Fatalf("AddPieces signer=%s want %s", recovered, storageSigner.EVMAddress()) } } diff --git a/storage/doc.go b/storage/doc.go index 92f7d4c..188b314 100644 --- a/storage/doc.go +++ b/storage/doc.go @@ -222,10 +222,10 @@ // // [ProviderContext.CreateAndAdd] and [DataSetContext.Commit] are convenience // methods that submit once and wait for confirmation. Their OnSubmitted -// callback receives an independent, complete [CommitSubmission] after the +// callback receives an independent runtime [CommitSubmission] after the // provider handle is validated and before confirmation starts. For -// create-and-add, persist StatusURL and ClientDataSetID. For add-pieces, -// persist StatusURL with the target DataSetRef. +// create-and-add, persist StatusURL and the original ClientDataSetID. For +// add-pieces, persist StatusURL with the target DataSetRef. // For explicit recovery control, prefer SubmitCreateAndAdd followed by // WaitForCreateAndAdd on ProviderContext, or SubmitCommit followed by // WaitForCommit on DataSetContext. @@ -241,8 +241,8 @@ // [FailedAttempt] in [UploadResult.FailedAttempts] or // [CommitError.FailedAttempts] carries it as Submission. Resume a // create-and-add submission on [Service.NewProviderContext] for its ProviderID -// and call WaitForCreateAndAdd with its StatusURL and ClientDataSetID. For -// add-pieces, open its DataSet with [Service.NewDataSetContext] and call +// and call WaitForCreateAndAdd with its StatusURL and original ClientDataSetID. +// For add-pieces, open its DataSet with [Service.NewDataSetContext] and call // WaitForCommit with StatusURL. A batched submission can include other uploads' // pieces. Applications that need a durable CID-to-piece-ID mapping must retain // their original request order; generic recovery validates only that the diff --git a/storage/types.go b/storage/types.go index 180f8a2..e7616c0 100644 --- a/storage/types.go +++ b/storage/types.go @@ -132,9 +132,11 @@ type CreateAndAddRequest struct { // set. Nil generates a random ID. When ExtraData is set, the value must // match the ID embedded in its create payload. ClientDataSetID *types.BigInt - // OnSubmitted is invoked with a persistable copy of the complete submission - // immediately after the provider returns a valid handle, before confirmation. - // It may be nil. Direct calls do not recover callback panics. + // OnSubmitted is invoked with an independent runtime snapshot immediately + // after the provider returns a valid handle, before confirmation. For restart + // recovery, persist StatusURL and the original ClientDataSetID; do not treat + // the complete CommitSubmission as a persistence schema. It may be nil. + // Direct calls do not recover callback panics. OnSubmitted func(CommitSubmission) } @@ -145,9 +147,11 @@ type CommitRequest struct { // DataSetContext sign the request. Payloads whose encoded add-pieces calldata // exceeds pdp.MaxAddPiecesMessageSize are rejected. ExtraData []byte - // OnSubmitted is invoked with a persistable copy of the complete submission - // immediately after the provider returns a valid handle, before confirmation. - // It may be nil. Direct calls do not recover callback panics. + // OnSubmitted is invoked with an independent runtime snapshot immediately + // after the provider returns a valid handle, before confirmation. For restart + // recovery, persist StatusURL together with the target DataSetRef; do not treat + // the complete CommitSubmission as a persistence schema. It may be nil. + // Direct calls do not recover callback panics. OnSubmitted func(CommitSubmission) } @@ -177,7 +181,7 @@ const ( // CommitSubmission describes one successful provider submission. It is a // runtime result for callbacks and diagnostics, not a durable recovery record. // Persist StatusURL to resume add-pieces. Create-and-add recovery also requires -// ClientDataSetID. +// the original ClientDataSetID. type CommitSubmission struct { Kind CommitKind `json:"kind"` TransactionID string `json:"transactionId"` @@ -222,7 +226,7 @@ type CreateDataSetOptions struct { // CreateDataSetSubmission describes a submitted create-dataset transaction. It // is a runtime result for callbacks and diagnostics, not a durable recovery -// record. Persist StatusURL and ClientDataSetID to resume waiting. +// record. Persist StatusURL and the original ClientDataSetID to resume waiting. type CreateDataSetSubmission struct { ProviderID types.BigInt `json:"providerId"` TransactionID string `json:"transactionId"`