fix(storage): restore identity, resume, and context validation

Author: strahe

docs/GETTING_STARTED.md

@@ -275,7 +275,7 @@ target. Its zero value is invalid; construct it explicitly and use accessors to
 read IDs.
 
 ```go
-ref, err := storage.NewDataSetRef(providerID, dataSetID, clientDataSetID)
+ref, err := storage.NewDataSetRef(providerID, dataSetID, dataSetCtx.ClientDataSetID())
 if err != nil {
     return err
 }

storage/context.go

@@ -673,11 +673,17 @@ func (c *DataSetContext) PieceURL(pieceCID cid.Cid) string {
 
 // ProviderID returns the provider's numeric ID.
 func (c *ProviderContext) ProviderID() types.BigInt {
+	if c == nil || c.core == nil {
+		return types.BigInt{}
+	}
 	return copyBigInt(c.core.provider.ID)
 }
 
 // ProviderID returns the provider's numeric ID.
 func (c *DataSetContext) ProviderID() types.BigInt {
+	if c == nil || c.core == nil {
+		return types.BigInt{}
+	}
 	return copyBigInt(c.core.provider.ID)
 }
 

storage/context_dataset.go

@@ -144,6 +144,9 @@ func (c *ProviderContext) waitForDataSetCreated(ctx context.Context, op string,
 
 func validateCreateDataSetSubmission(op string, providerID types.BigInt, submission CreateDataSetSubmission) (CreateDataSetSubmission, error) {
 	submission = copyCreateDataSetSubmission(submission)
+	if submission.ProviderID.IsZero() {
+		submission.ProviderID = copyBigInt(providerID)
+	}
 	if submission.ProviderID.IsZero() {
 		return CreateDataSetSubmission{}, fmt.Errorf("%s: %w: zero providerID", op, ErrInvalidArgument)
 	}

storage/context_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/strahe/synapse-go/piece"
 	"github.com/strahe/synapse-go/signer"
 	"github.com/strahe/synapse-go/types"
+	"github.com/strahe/synapse-go/warmstorage"
 )
 
 func mustPieceInfo(t *testing.T) piece.PieceInfo {
@@ -576,6 +577,148 @@ func TestProviderContextWaitForDataSetCreatedRejectsWrongProvider(t *testing.T)
 	}
 }
 
+func TestProviderContextWaitForDataSetCreatedAcceptsZeroProviderID(t *testing.T) {
+	txHash := common.HexToHash("0x1234")
+	dataSetID := types.NewBigInt(77)
+	clientID := types.NewBigInt(7)
+	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
+		},
+	}
+	c := mustWritableProviderContext(t, client)
+	result, err := c.WaitForDataSetCreated(context.Background(), CreateDataSetSubmission{
+		TransactionID:   txHash.Hex(),
+		StatusURL:       "https://sp.example.com/status",
+		ClientDataSetID: &clientID,
+	})
+	if err != nil {
+		t.Fatalf("WaitForDataSetCreated: %v", err)
+	}
+	if waitCalls != 1 {
+		t.Fatalf("waitCalls=%d want 1", waitCalls)
+	}
+	if !result.DataSet.ProviderID().Equal(testProvider().ID) ||
+		!result.DataSet.DataSetID().Equal(dataSetID) ||
+		!result.DataSet.ClientDataSetID().Equal(clientID) {
+		t.Fatalf("result=%+v", result.DataSet)
+	}
+}
+
+func TestProviderContextWaitForDataSetCreatedRejectsInvalidSubmission(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},
+	}
+	for name, submission := range tests {
+		t.Run(name, func(t *testing.T) {
+			waitCalls := 0
+			client := &fakePDPProviderClient{
+				waitForCreatedFn: func(context.Context, string, time.Duration) (*pdp.CreateDataSetStatus, error) {
+					waitCalls++
+					return nil, errors.New("wait should not be called")
+				},
+			}
+			c := mustWritableProviderContext(t, client)
+			_, err := c.WaitForDataSetCreated(context.Background(), submission)
+			if !errors.Is(err, ErrInvalidArgument) {
+				t.Fatalf("WaitForDataSetCreated error=%v want ErrInvalidArgument", err)
+			}
+			if waitCalls != 0 {
+				t.Fatalf("waitCalls=%d want 0", waitCalls)
+			}
+		})
+	}
+}
+
+func TestDataSetContextCommitRejectsEndedAndValidatorFailuresBeforeAddPieces(t *testing.T) {
+	info := mustPieceInfo(t)
+	dataSetID := types.NewBigInt(42)
+	want := errors.New("not live")
+
+	t.Run("ended rail", func(t *testing.T) {
+		addCalls := 0
+		reader := &fakeFWSSDataSetReader{
+			info: &warmstorage.DataSetInfo{DataSetID: dataSetID, PDPEndEpoch: 3778900},
+		}
+		client := &fakePDPProviderClient{
+			addPiecesFn: func(context.Context, types.BigInt, []pdp.AddPieceInput, []byte) (*pdp.AddPiecesResult, error) {
+				addCalls++
+				t.Fatal("AddPieces must not run for an ended data set")
+				return nil, nil
+			},
+		}
+		c := mustWritableDataSetContext(t, client, testDataSetRef(dataSetID, types.NewBigInt(7)), WithFWSSDataSetReader(reader))
+		_, err := c.Commit(context.Background(), CommitRequest{Pieces: []PieceInput{{PieceCID: info.CIDv2}}})
+		requireDataSetPDPPaymentTerminated(t, err, dataSetID, 3778900)
+		if addCalls != 0 {
+			t.Fatalf("addCalls=%d want 0", addCalls)
+		}
+		if reader.calls != 1 || !reader.gotID.Equal(dataSetID) {
+			t.Fatalf("reader calls=%d gotID=%s", reader.calls, reader.gotID.String())
+		}
+	})
+
+	t.Run("validator failure", func(t *testing.T) {
+		addCalls := 0
+		validator := &fakeDataSetValidator{err: want}
+		client := &fakePDPProviderClient{
+			addPiecesFn: func(context.Context, types.BigInt, []pdp.AddPieceInput, []byte) (*pdp.AddPiecesResult, error) {
+				addCalls++
+				t.Fatal("AddPieces must not run after validator failure")
+				return nil, nil
+			},
+		}
+		c := mustWritableDataSetContext(t, client, testDataSetRef(dataSetID, types.NewBigInt(7)), WithDataSetValidator(validator))
+		_, err := c.Commit(context.Background(), CommitRequest{Pieces: []PieceInput{{PieceCID: info.CIDv2}}})
+		if !errors.Is(err, want) {
+			t.Fatalf("Commit error=%v want %v", err, want)
+		}
+		if addCalls != 0 {
+			t.Fatalf("addCalls=%d want 0", addCalls)
+		}
+		if len(validator.calls) != 1 || !validator.calls[0].Equal(dataSetID) {
+			t.Fatalf("validator calls=%v", validator.calls)
+		}
+	})
+}
+
+func TestDataSetContextUploadRejectsEndedExistingDataSetBeforeStore(t *testing.T) {
+	dataSetID := types.NewBigInt(13269)
+	storeCalled := false
+	reader := &fakeFWSSDataSetReader{
+		info: &warmstorage.DataSetInfo{DataSetID: dataSetID, PDPEndEpoch: 3778900},
+	}
+	client := &fakePDPProviderClient{
+		uploadStreamingFn: func(context.Context, io.Reader, pdp.UploadPieceStreamingOptions) (*pdp.UploadStreamingResult, error) {
+			storeCalled = true
+			t.Fatal("Store must not run when the existing data set cannot accept uploads")
+			return nil, nil
+		},
+	}
+	c := mustWritableDataSetContext(t, client, testDataSetRef(dataSetID, types.NewBigInt(99)), WithFWSSDataSetReader(reader))
+	_, err := c.Upload(context.Background(), bytes.NewReader([]byte("payload")), nil)
+	requireDataSetPDPPaymentTerminated(t, err, dataSetID, 3778900)
+	if storeCalled {
+		t.Fatal("Store was called")
+	}
+	if reader.calls != 1 || !reader.gotID.Equal(dataSetID) {
+		t.Fatalf("reader calls=%d gotID=%s", reader.calls, reader.gotID.String())
+	}
+}
+
 func TestContextPullRoutesByConcreteType(t *testing.T) {
 	info := mustPieceInfo(t)
 	dataSetID := types.NewBigInt(42)

storage/create.go

@@ -275,10 +275,11 @@ func (s *Service) validateContextIdentity(op string, storageCtx StorageContext)
 	if isNilStorageContext(storageCtx) {
 		return fmt.Errorf("%s: %w: nil storage context", op, ErrInvalidArgument)
 	}
+	if s.signerAddr == (common.Address{}) || !s.chainID.IsValid() || s.recordKeeper == (common.Address{}) {
+		return fmt.Errorf("%s: %w: service identity is incomplete (payer, chain, and record keeper)", op, ErrInvalidArgument)
+	}
 	identity := storageCtx.ContextIdentity()
-	if identity.Payer == s.signerAddr && identity.Payer != (common.Address{}) &&
-		identity.ChainID == s.chainID && identity.ChainID.IsValid() &&
-		identity.RecordKeeper == s.recordKeeper && identity.RecordKeeper != (common.Address{}) {
+	if identity.Payer == s.signerAddr && identity.ChainID == s.chainID && identity.RecordKeeper == s.recordKeeper {
 		return nil
 	}
 	return fmt.Errorf("%s: %w: context identity does not match service payer, chain, and record keeper", op, ErrInvalidArgument)
@@ -342,6 +343,12 @@ func isNilStorageContext(storageCtx StorageContext) bool {
 	if storageCtx == nil {
 		return true
 	}
+	switch c := storageCtx.(type) {
+	case *ProviderContext:
+		return c == nil || c.core == nil
+	case *DataSetContext:
+		return c == nil || c.core == nil
+	}
 	value := reflect.ValueOf(storageCtx)
 	switch value.Kind() {
 	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
@@ -352,7 +359,7 @@ func isNilStorageContext(storageCtx StorageContext) bool {
 }
 
 func storageCtxID(storageCtx *ProviderContext) types.BigInt {
-	if storageCtx == nil {
+	if storageCtx == nil || storageCtx.core == nil {
 		return types.BigInt{}
 	}
 	return storageCtx.ProviderID()

storage/create_test.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -231,12 +232,39 @@ func TestServiceSelectUploadContextsDiscardsHardErrorResult(t *testing.T) {
 	}
 }
 
+func TestServiceSelectProviderContextRejectsNilCore(t *testing.T) {
+	svc := newTestService()
+	svc.contextSelector = &fakeContextSelector{providerFn: func(context.Context, SelectProviderContextOptions) (*ProviderContext, error) {
+		return &ProviderContext{}, nil
+	}}
+	result, err := svc.SelectProviderContext(context.Background(), SelectProviderContextOptions{})
+	if result != nil || !errors.Is(err, ErrInvalidArgument) {
+		t.Fatalf("result=%v error=%v want ErrInvalidArgument", result, err)
+	}
+}
+
+func TestServiceUploadToContextsIncompleteServiceIdentity(t *testing.T) {
+	svc, err := New(Options{SignerAddress: testPayer()})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	reader := &readCountingReader{}
+	result, err := svc.UploadToContexts(context.Background(), reader, []StorageContext{&fakeUploadContext{id: types.NewBigInt(1)}}, nil)
+	if result != nil || !errors.Is(err, ErrInvalidArgument) || !strings.Contains(err.Error(), "service identity") || strings.Contains(err.Error(), "does not match") {
+		t.Fatalf("result=%v error=%v want ErrInvalidArgument service identity", result, err)
+	}
+	if reader.reads != 0 {
+		t.Fatalf("reader reads=%d want 0", reader.reads)
+	}
+}
+
 func TestServiceSelectUploadContextsRejectsTypedNilAndIdentityMismatch(t *testing.T) {
 	for _, tt := range []struct {
 		name string
 		ctx  StorageContext
 	}{
 		{name: "typed nil", ctx: (*ProviderContext)(nil)},
+		{name: "nil core", ctx: &ProviderContext{}},
 		{name: "wrong payer", ctx: testProviderContextWithID(t, types.NewBigInt(1), ContextIdentity{
 			Payer: common.HexToAddress("0x9999"), ChainID: types.ChainID(314159), RecordKeeper: testRecordKeeper(),
 		})},

storage/service.go

@@ -767,16 +767,16 @@ func validateExplicitUploadOptions(opts *UploadOptions) error {
 		return nil
 	}
 	if opts.Copies != 0 {
-		return fmt.Errorf("%w: Copies is not supported by UploadToContexts", ErrInvalidArgument)
+		return fmt.Errorf("%w: Copies is not supported for explicit-context uploads", ErrInvalidArgument)
 	}
 	if len(opts.ExcludeProviderIDs) != 0 {
-		return fmt.Errorf("%w: ExcludeProviderIDs is not supported by UploadToContexts", ErrInvalidArgument)
+		return fmt.Errorf("%w: ExcludeProviderIDs is not supported for explicit-context uploads; pass it to SelectUploadContexts instead", ErrInvalidArgument)
 	}
 	if len(opts.DataSetMetadata) != 0 {
-		return fmt.Errorf("%w: DataSetMetadata is not supported by UploadToContexts", ErrInvalidArgument)
+		return fmt.Errorf("%w: DataSetMetadata is not supported for explicit-context uploads; pass it to SelectUploadContexts or the context constructor instead", ErrInvalidArgument)
 	}
 	if opts.WithCDN != nil {
-		return fmt.Errorf("%w: WithCDN is not supported by UploadToContexts", ErrInvalidArgument)
+		return fmt.Errorf("%w: WithCDN is not supported for explicit-context uploads; pass it to SelectUploadContexts or the context constructor instead", ErrInvalidArgument)
 	}
 	return nil
 }

storage/service_test.go

@@ -942,6 +942,8 @@ func TestServiceUploadToContextsValidatesBeforeReading(t *testing.T) {
 	tests := map[string][]StorageContext{
 		"empty":              nil,
 		"typed nil":          {(*fakeUploadContext)(nil)},
+		"nil-core provider":  {&ProviderContext{}},
+		"nil-core data set":  {&DataSetContext{}},
 		"duplicate provider": {valid, valid},
 		"identity mismatch":  {wrong},
 	}
@@ -976,6 +978,9 @@ func TestServiceUploadToContextsRejectsSelectionOptionsBeforeReading(t *testing.
 			if result != nil || !errors.Is(err, ErrInvalidArgument) {
 				t.Fatalf("result=%v error=%v want ErrInvalidArgument", result, err)
 			}
+			if !strings.Contains(err.Error(), "explicit-context uploads") {
+				t.Fatalf("error=%q want explicit-context uploads", err)
+			}
 			if reader.reads != 0 {
 				t.Fatalf("reader reads=%d want 0", reader.reads)
 			}

storage/types.go

@@ -139,6 +139,7 @@ type CreateDataSetOptions struct {
 
 // CreateDataSetSubmission identifies a submitted create-dataset transaction.
 // Persist and restore all fields together; incomplete submissions are rejected.
+// A zero ProviderID is filled from the ProviderContext used to wait.
 type CreateDataSetSubmission struct {
 	ProviderID    types.BigInt
 	TransactionID string