feat(storage): prefer active data sets for explicit providers

Author: strahe

internal/adapters/pdpverifier.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"fmt"
 	"math/big"
-	"strings"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ipfs/go-cid"
 
@@ -57,7 +55,7 @@ func (a *pdpVerifierReader) FindPieceIdsByCid(ctx context.Context, dataSetID sdk
 		new(big.Int).SetUint64(limit),
 	)
 	if err != nil {
-		if isDataSetUnavailable(err) {
+		if pdpverifier.IsDataSetUnavailable(err) {
 			return []sdktypes.BigInt{}, nil
 		}
 		return nil, fmt.Errorf("adapters.pdpVerifierReader.FindPieceIdsByCid: %w", err)
@@ -72,7 +70,7 @@ func (a *pdpVerifierReader) FindPieceIdsByCid(ctx context.Context, dataSetID sdk
 func (a *pdpVerifierReader) GetScheduledRemovals(ctx context.Context, dataSetID sdktypes.BigInt) ([]sdktypes.BigInt, error) {
 	raw, err := a.caller.GetScheduledRemovals(&bind.CallOpts{Context: ctx}, dataSetID.Big())
 	if err != nil {
-		if isDataSetUnavailable(err) {
+		if pdpverifier.IsDataSetUnavailable(err) {
 			return []sdktypes.BigInt{}, nil
 		}
 		return nil, fmt.Errorf("adapters.pdpVerifierReader.GetScheduledRemovals: %w", err)
@@ -87,7 +85,7 @@ func (a *pdpVerifierReader) GetScheduledRemovals(ctx context.Context, dataSetID
 func (a *pdpVerifierReader) GetNextChallengeEpoch(ctx context.Context, dataSetID sdktypes.BigInt) (*big.Int, error) {
 	v, err := a.caller.GetNextChallengeEpoch(&bind.CallOpts{Context: ctx}, dataSetID.Big())
 	if err != nil {
-		if isDataSetUnavailable(err) {
+		if pdpverifier.IsDataSetUnavailable(err) {
 			return nil, nil
 		}
 		return nil, fmt.Errorf("adapters.pdpVerifierReader.GetNextChallengeEpoch: %w", err)
@@ -113,7 +111,7 @@ func (a *pdpVerifierReader) BlockNumber(ctx context.Context) (uint64, error) {
 func (a *pdpVerifierReader) GetDataSetSizeBytes(ctx context.Context, dataSetID sdktypes.BigInt) (*big.Int, error) {
 	leafCount, err := a.caller.GetDataSetLeafCount(&bind.CallOpts{Context: ctx}, dataSetID.Big())
 	if err != nil {
-		if isDataSetUnavailable(err) {
+		if pdpverifier.IsDataSetUnavailable(err) {
 			return new(big.Int), nil
 		}
 		return nil, fmt.Errorf("adapters.pdpVerifierReader.GetDataSetSizeBytes: %w", err)
@@ -124,36 +122,6 @@ func (a *pdpVerifierReader) GetDataSetSizeBytes(ctx context.Context, dataSetID s
 	return new(big.Int).Mul(leafCount, big.NewInt(32)), nil
 }
 
-var (
-	dataSetNotFoundSelector = errorSelector("DataSetNotFound()")
-	dataSetNotLiveSelector  = errorSelector("DataSetNotLive()")
-)
-
-// isDataSetUnavailable reports whether err is the PDPVerifier revert raised
-// for terminated, missing, or otherwise non-readable data sets.
-func isDataSetUnavailable(err error) bool {
-	if err == nil {
-		return false
-	}
-	msg := err.Error()
-	if strings.Contains(msg, "Data set not live") ||
-		strings.Contains(msg, "DataSetNotLive") ||
-		strings.Contains(msg, "DataSetNotFound") {
-		return true
-	}
-	data, ok := ethclient.RevertErrorData(err)
-	if !ok || len(data) < 4 {
-		return false
-	}
-	selector := [4]byte{data[0], data[1], data[2], data[3]}
-	return selector == dataSetNotFoundSelector || selector == dataSetNotLiveSelector
-}
-
-func errorSelector(signature string) [4]byte {
-	hash := crypto.Keccak256([]byte(signature))
-	return [4]byte{hash[0], hash[1], hash[2], hash[3]}
-}
-
 func dedupeBigInts(values []sdktypes.BigInt) []sdktypes.BigInt {
 	if len(values) == 0 {
 		return values

internal/contracts/pdpverifier/doc.go

@@ -1,5 +1,5 @@
-// Package pdpverifier contains the generated Go bindings for the
-// PDPVerifier smart contract.
+// Package pdpverifier contains generated Go bindings and internal error
+// helpers for the PDPVerifier smart contract.
 //
 // Regenerate with: go generate ./internal/contracts/...
 package pdpverifier

internal/contracts/pdpverifier/errors.go

@@ -0,0 +1,38 @@
+package pdpverifier
+
+import (
+	"strings"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+var (
+	dataSetNotFoundSelector = pdpErrorSelector("DataSetNotFound()")
+	dataSetNotLiveSelector  = pdpErrorSelector("DataSetNotLive()")
+)
+
+// IsDataSetUnavailable reports whether err is the PDPVerifier revert raised
+// for terminated, missing, or otherwise non-readable data sets.
+func IsDataSetUnavailable(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	if strings.Contains(msg, "Data set not live") ||
+		strings.Contains(msg, "DataSetNotLive") ||
+		strings.Contains(msg, "DataSetNotFound") {
+		return true
+	}
+	data, ok := ethclient.RevertErrorData(err)
+	if !ok || len(data) < 4 {
+		return false
+	}
+	selector := [4]byte{data[0], data[1], data[2], data[3]}
+	return selector == dataSetNotFoundSelector || selector == dataSetNotLiveSelector
+}
+
+func pdpErrorSelector(signature string) [4]byte {
+	hash := crypto.Keccak256([]byte(signature))
+	return [4]byte{hash[0], hash[1], hash[2], hash[3]}
+}

internal/contracts/pdpverifier/errors_test.go

@@ -0,0 +1,58 @@
+package pdpverifier
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/crypto"
+)
+
+type revertDataError struct {
+	data string
+}
+
+func (e revertDataError) Error() string {
+	return "execution reverted"
+}
+
+func (e revertDataError) ErrorCode() int {
+	return 3
+}
+
+func (e revertDataError) ErrorData() interface{} {
+	return e.data
+}
+
+func customError(name string) error {
+	hash := crypto.Keccak256([]byte(name + "()"))
+	return revertDataError{data: hexutil.Encode(hash[:4])}
+}
+
+func TestIsDataSetUnavailable(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{name: "nil", err: nil, want: false},
+		{name: "ordinary error", err: errors.New("connection reset"), want: false},
+		{name: "legacy not live string", err: errors.New("execution reverted: Data set not live"), want: true},
+		{name: "custom not live string", err: errors.New("execution reverted: DataSetNotLive()"), want: true},
+		{name: "custom not found string", err: errors.New("execution reverted: DataSetNotFound()"), want: true},
+		{name: "custom not live selector", err: customError("DataSetNotLive"), want: true},
+		{name: "custom not found selector", err: customError("DataSetNotFound"), want: true},
+		{name: "other selector", err: customError("CleanupDepositRequired"), want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			if got := IsDataSetUnavailable(tt.err); got != tt.want {
+				t.Fatalf("IsDataSetUnavailable() = %t, want %t", got, tt.want)
+			}
+		})
+	}
+}

internal/upstream/upstream.go

@@ -6,8 +6,8 @@ const (
 	TSSDKRepo = "FilOzone/synapse-sdk"
 	// TSSDKLocalDir is the local TypeScript SDK checkout path.
 	TSSDKLocalDir = "synapse-sdk"
-	// TSSDKRef is the pinned TypeScript SDK commit for synapse-sdk-v1.0.1.
-	TSSDKRef = "e044ddf62210f505d918e723a7f4cd62f4e49ab0"
+	// TSSDKRef is the pinned TypeScript SDK commit for synapse-sdk-v1.1.0.
+	TSSDKRef = "5c7ad6ccc0500d937435a2e0dd7d09549aeccbeb"
 	// FilecoinServicesRepo is the upstream contract ABI repository.
 	FilecoinServicesRepo = "FilOzone/filecoin-services"
 	// FilecoinServicesRef is the pinned contract ABI commit.

internal/upstream/upstream_test.go

@@ -48,8 +48,8 @@ func TestLocalTSSDKBaseline(t *testing.T) {
 	if sdkPackage.Name != "@filoz/synapse-sdk" {
 		t.Fatalf("synapse-sdk package name = %q, want @filoz/synapse-sdk", sdkPackage.Name)
 	}
-	if sdkPackage.Version != "1.0.1" {
-		t.Fatalf("synapse-sdk package version = %q, want 1.0.1", sdkPackage.Version)
+	if sdkPackage.Version != "1.1.0" {
+		t.Fatalf("synapse-sdk package version = %q, want 1.1.0", sdkPackage.Version)
 	}
 	coreDep := sdkPackage.Dependencies["@filoz/synapse-core"]
 	if coreDep == "" {

storage/doc.go

@@ -45,11 +45,15 @@
 // The Service handles orchestration of the full multi-copy flow, while
 // ServiceResolver implements both resolver contracts. It reuses provider-local
 // datasets only when metadata matches exactly and the warmstorage-approved
-// provider set intersects active PDP providers from spregistry. Automatically
-// selected providers must also pass a bounded PDP health check before a context
-// is created; explicitly selected provider and data set IDs are not probed. If
-// only some automatically selected providers are healthy, the resolver returns
-// the healthy subset so upload callers can surface partial-copy results.
+// provider set intersects active PDP providers from spregistry. For explicit
+// provider IDs, the default warmstorage service checks matching data sets with
+// bounded concurrency, preferring the oldest one with active pieces and falling
+// back to the oldest empty match. Custom catalogs without active-piece reads
+// retain the original oldest-match behavior. Automatically selected providers
+// must also pass a bounded PDP health check before a context is created;
+// explicitly selected provider and data set IDs are not probed. If only some
+// automatically selected providers are healthy, the resolver returns the
+// healthy subset so upload callers can surface partial-copy results.
 // Existing data sets that cannot accept writes surface typed errors such as
 // DataSetPDPPaymentTerminatedError. Use errors.AsType to access fields like
 // PDPEndEpoch.