Skip to content

refactor(storage): minimize recovery state #34

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
strahe merged 3 commits into from
Sep 21, 2026
Merged

refactor(storage): minimize recovery state #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
868f195
refactor(storage): minimize recovery state
strahe Sep 21, 2026
8b9823a
fix(pdp/storage): enforce commit recovery invariants
strahe Sep 21, 2026
0a72e02
fix(storage): clarify recovery state contract
strahe Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 49 additions & 31 deletions docs/GETTING_STARTED.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 original 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 {
Expand All @@ -490,10 +494,12 @@ 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, submitted)
created, err := providerCtx.WaitForDataSetCreated(ctx, statusURL, clientDataSetID)
if err != nil {
return err
}
Expand All @@ -518,36 +524,49 @@ 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 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 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)
defer cancel()

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
Expand All @@ -565,27 +584,26 @@ 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 the original
// *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

Expand Down
12 changes: 8 additions & 4 deletions internal/apiaudit/apiaudit_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
22 changes: 11 additions & 11 deletions pdp/client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
Expand All @@ -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)
}
Expand Down
Loading
Loading