fix(storage): harden provider health checks (#1)

* fix(storage): harden provider health checks

* fix(storage): stop probes beyond selection frontier

Author: strahe

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.1.0.
-	TSSDKRef = "5c7ad6ccc0500d937435a2e0dd7d09549aeccbeb"
+	// TSSDKRef is the pinned TypeScript SDK commit for synapse-sdk-v1.1.1.
+	TSSDKRef = "a1d44296ad27b4a2631cb744de95a6a94c8097a7"
 	// 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.1.0" {
-		t.Fatalf("synapse-sdk package version = %q, want 1.1.0", sdkPackage.Version)
+	if sdkPackage.Version != "1.1.1" {
+		t.Fatalf("synapse-sdk package version = %q, want 1.1.1", sdkPackage.Version)
 	}
 	coreDep := sdkPackage.Dependencies["@filoz/synapse-core"]
 	if coreDep == "" {
@@ -63,8 +63,8 @@ func TestLocalTSSDKBaseline(t *testing.T) {
 	if corePackage.Name != "@filoz/synapse-core" {
 		t.Fatalf("synapse-core package name = %q, want @filoz/synapse-core", corePackage.Name)
 	}
-	if corePackage.Version != "0.7.0" {
-		t.Fatalf("synapse-core package version = %q, want 0.7.0", corePackage.Version)
+	if corePackage.Version != "0.7.1" {
+		t.Fatalf("synapse-core package version = %q, want 0.7.1", corePackage.Version)
 	}
 	t.Logf("local TS baseline packages: %s@%s depends on %s@%s via %s", sdkPackage.Name, sdkPackage.Version, corePackage.Name, corePackage.Version, coreDep)
 

payments/coverage_test.go

@@ -245,6 +245,29 @@ func TestGetSettlementAmounts_DecodesTuple(t *testing.T) {
 	}
 }
 
+func TestReadsDoNotUseSignerAddress(t *testing.T) {
+	s, mb := newTestService(t)
+	if s.Account() == (common.Address{}) {
+		t.Fatal("test service has no signer account")
+	}
+	mb.rejectNonZeroCallFrom = true
+	mb.setFilPayReply(t, filPayAddr, "accounts",
+		big.NewInt(1000), big.NewInt(200), big.NewInt(5), big.NewInt(42))
+	mb.setFilPayReply(t, filPayAddr, "getAccountInfoIfSettled",
+		big.NewInt(0), big.NewInt(1000), big.NewInt(123), big.NewInt(5))
+	mb.setFilPayReply(t, filPayAddr, "settleRail",
+		big.NewInt(100), big.NewInt(90), big.NewInt(5), big.NewInt(5),
+		big.NewInt(50), "ok",
+	)
+
+	if _, err := s.AccountInfo(context.Background(), tokenAddr, s.Account()); err != nil {
+		t.Fatalf("AccountInfo: %v", err)
+	}
+	if _, err := s.GetSettlementAmounts(context.Background(), sdktypes.NewBigInt(3), big.NewInt(50)); err != nil {
+		t.Fatalf("GetSettlementAmounts: %v", err)
+	}
+}
+
 func TestGetSettlementAmounts_InvalidRailID(t *testing.T) {
 	s, _ := newTestService(t)
 	if _, err := s.GetSettlementAmounts(context.Background(), sdktypes.NewBigInt(0), nil); !errors.Is(err, ErrInvalidArgument) {

payments/service_test.go

@@ -36,7 +36,8 @@ type mockBackend struct {
 	// errors keyed the same way
 	errs map[string]error
 	// callReplyFn, when set, can override a contract call reply.
-	callReplyFn func(contractHex, method string, data []byte) ([]byte, bool, error)
+	callReplyFn           func(contractHex, method string, data []byte) ([]byte, bool, error)
+	rejectNonZeroCallFrom bool
 
 	// per-account balances; nil -> 0
 	balances  map[common.Address]*big.Int
@@ -91,6 +92,9 @@ func (m *mockBackend) CodeAt(_ context.Context, _ common.Address, _ *big.Int) ([
 func (m *mockBackend) CallContract(_ context.Context, call ethereum.CallMsg, blockNumber *big.Int) ([]byte, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	if m.rejectNonZeroCallFrom && call.From != (common.Address{}) {
+		return nil, errors.New("non-zero eth_call sender")
+	}
 	if len(call.Data) < 4 || call.To == nil {
 		return nil, errors.New("calldata too short")
 	}

services.go

@@ -126,16 +126,10 @@ func (c *Client) initServices() error {
 	}
 
 	resolver, err := storage.NewServiceResolver(storage.ServiceResolverOptions{
-		Payer:       c.evmSigner.EVMAddress(),
-		SPRegistry:  spReg,
-		WarmStorage: ws,
-		ProviderPing: func(ctx context.Context, serviceURL string) error {
-			pdpClient, err := c.newPDPClient(serviceURL, pdp.WithMaxRetries(0))
-			if err != nil {
-				return err
-			}
-			return pdpClient.Ping(ctx)
-		},
+		Payer:        c.evmSigner.EVMAddress(),
+		SPRegistry:   spReg,
+		WarmStorage:  ws,
+		ProviderPing: c.pingProvider,
 		NewContext: func(sel storage.ResolvedUploadContext, opts *storage.UploadOptions) (*storage.Context, error) {
 			pdpClient, err := c.newPDPClient(sel.Provider.ServiceURL)
 			if err != nil {
@@ -224,6 +218,14 @@ func (c *Client) newPDPClient(serviceURL string, opts ...pdp.Option) (*pdp.Clien
 	return pdp.New(serviceURL, pdpOpts...)
 }
 
+func (c *Client) pingProvider(ctx context.Context, serviceURL string) error {
+	pdpClient, err := c.newPDPClient(serviceURL, pdp.WithMaxRetries(2))
+	if err != nil {
+		return err
+	}
+	return pdpClient.Ping(ctx)
+}
+
 // WarmStorage returns the [warmstorage.Service].
 func (c *Client) WarmStorage() *warmstorage.Service {
 	return c.warmStorage

storage/selector.go

@@ -27,7 +27,9 @@ const selectorListPageSize = 100
 
 const selectorRetryInitialDelay = 200 * time.Millisecond
 
-const providerPingTimeout = 2 * time.Second
+const providerPingTimeout = 8 * time.Second
+
+const providerPingConcurrency = 16
 
 const resolveConcurrency = 10
 
@@ -43,6 +45,19 @@ type autoSelectCandidate struct {
 	metadata        map[string]string
 }
 
+type providerProbeState uint8
+
+const (
+	providerProbePending providerProbeState = iota
+	providerProbeHealthy
+	providerProbeUnhealthy
+)
+
+type providerProbeResult struct {
+	index int
+	err   error
+}
+
 // PDPProviderSource is the subset of spregistry.Service used by ServiceResolver.
 type PDPProviderSource interface {
 	GetPDPProvider(context.Context, types.BigInt) (*spregistry.PDPProvider, error)
@@ -84,9 +99,12 @@ type ServiceResolverOptions struct {
 	DataSetValidator DataSetValidator
 	DataSetDetails   DataSetDetailsCatalog
 	// ProviderPing checks one automatically selected provider endpoint. The
-	// resolver passes a context with a two-second deadline; the implementation
-	// is responsible for honoring it. nil uses an unretried [pdp.Client.Ping]
-	// request.
+	// resolver may call it concurrently for up to 16 candidates per selection;
+	// concurrent selections have independent limits. Each call receives a
+	// context whose deadline is no later than eight seconds; a shorter parent
+	// deadline takes precedence. Implementations must be safe for concurrent use
+	// and honor the context. nil uses [pdp.Client.Ping] with up to two retries for
+	// transient failures.
 	ProviderPing func(context.Context, string) error
 	NewContext   ContextFactory // called per-provider to construct a managed Context
 }
@@ -122,8 +140,8 @@ var (
 // NewServiceResolver constructs a ServiceResolver. Payer, SPRegistry,
 // WarmStorage, and NewContext are required. DataSetValidator, DataSetDetails,
 // and active-piece reads are optional and auto-detected from WarmStorage when
-// available and configured. ProviderPing is optional and defaults to an
-// unretried PDP ping request.
+// available and configured. ProviderPing is optional and defaults to a PDP
+// ping with up to two retries for transient failures.
 func NewServiceResolver(opts ServiceResolverOptions) (*ServiceResolver, error) {
 	if opts.Payer == (common.Address{}) {
 		return nil, fmt.Errorf("storage.NewServiceResolver: %w: zero payer", ErrInvalidArgument)
@@ -169,7 +187,7 @@ func NewServiceResolver(opts ServiceResolverOptions) (*ServiceResolver, error) {
 }
 
 func defaultProviderPing(ctx context.Context, serviceURL string) error {
-	client, err := pdp.New(serviceURL, pdp.WithMaxRetries(0))
+	client, err := pdp.New(serviceURL, pdp.WithMaxRetries(2))
 	if err != nil {
 		return err
 	}
@@ -441,29 +459,7 @@ func (r *ServiceResolver) autoSelect(ctx context.Context, opts *UploadOptions, e
 		providersWithDetails = detailedCandidateProviders(detailedDataSets, selectableProviders)
 	}
 	requestedMetadata := dataSetMetadataFromOptions(opts)
-	selected := make([]ResolvedUploadContext, 0, min(count, len(providers)))
-	failedProviderIDs := make([]types.BigInt, 0, min(count, len(providers)))
-	probeCandidate := func(candidate autoSelectCandidate) (bool, error) {
-		pingCtx, cancel := context.WithTimeout(ctx, providerPingTimeout)
-		pingErr := r.providerPing(pingCtx, candidate.provider.Offering.ServiceURL)
-		cancel()
-		if ctxErr := ctx.Err(); ctxErr != nil {
-			return false, fmt.Errorf("storage.ServiceResolver.ResolveUploadContexts: health-check provider %s: %w", candidate.provider.Info.ID.String(), ctxErr)
-		}
-		if pingErr != nil {
-			failedProviderIDs = append(failedProviderIDs, candidate.provider.Info.ID)
-			return false, nil
-		}
-		selected = append(selected, buildResolvedUploadContext(
-			*candidate.provider,
-			candidate.dataSetID,
-			candidate.clientDataSetID,
-			candidate.metadata,
-		))
-		return len(selected) == count, nil
-	}
-
-	deferWithoutDataSet := len(providersWithDetails) > 0
+	withDataSet := make([]autoSelectCandidate, 0, min(count, len(providers)))
 	withoutDataSet := make([]autoSelectCandidate, 0, min(count, len(providers)))
 	seenProviderIDs := make(map[string]struct{}, min(count, len(providers)))
 	for i := range providers {
@@ -481,49 +477,117 @@ func (r *ServiceResolver) autoSelect(ctx context.Context, opts *UploadOptions, e
 		var metadata map[string]string
 		if providerDataSets, ok := providersWithDetails[providerKey]; ok {
 			dataSetID, clientDataSetID, metadata = selectMatchingDetailedDataSet(provider.Info.ID, providerDataSets, requestedMetadata)
-			delete(providersWithDetails, providerKey)
 		}
 		candidate := autoSelectCandidate{
 			provider:        provider,
 			dataSetID:       dataSetID,
 			clientDataSetID: clientDataSetID,
 			metadata:        metadata,
 		}
-		switch {
-		case dataSetID != nil:
-			done, err := probeCandidate(candidate)
-			if err != nil {
-				return nil, err
-			}
-			if done {
-				return selected, nil
-			}
-		case deferWithoutDataSet:
+		if dataSetID != nil {
+			withDataSet = append(withDataSet, candidate)
+		} else {
 			candidate.metadata = requestedMetadata
 			withoutDataSet = append(withoutDataSet, candidate)
-		default:
-			candidate.metadata = requestedMetadata
-			done, err := probeCandidate(candidate)
-			if err != nil {
-				return nil, err
-			}
-			if done {
-				return selected, nil
+		}
+	}
+	candidates := slices.Concat(withDataSet, withoutDataSet)
+	return r.selectHealthyCandidates(ctx, candidates, count)
+}
+
+func (r *ServiceResolver) selectHealthyCandidates(ctx context.Context, candidates []autoSelectCandidate, count int) ([]ResolvedUploadContext, error) {
+	if len(candidates) == 0 {
+		return nil, errors.New("storage.ServiceResolver.ResolveUploadContexts: no remaining providers")
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("storage.ServiceResolver.ResolveUploadContexts: health-check providers: %w", err)
+	}
+
+	probeCtx, cancelProbes := context.WithCancel(ctx)
+	defer cancelProbes()
+	states := make([]providerProbeState, len(candidates))
+	results := make(chan providerProbeResult, min(providerPingConcurrency, len(candidates)))
+	probeCancels := make([]context.CancelFunc, len(candidates))
+	nextIndex := 0
+	inFlight := 0
+	probeLimit := len(candidates)
+	stopping := false
+	selectionDetermined := false
+
+	startProbes := func() {
+		for !stopping && inFlight < providerPingConcurrency && nextIndex < probeLimit {
+			index := nextIndex
+			nextIndex++
+			inFlight++
+			pingCtx, cancel := context.WithTimeout(probeCtx, providerPingTimeout)
+			probeCancels[index] = cancel
+			go func() {
+				err := r.providerPing(pingCtx, candidates[index].provider.Offering.ServiceURL)
+				cancel()
+				results <- providerProbeResult{index: index, err: err}
+			}()
+		}
+	}
+	cancelProbesFrom := func(index int) {
+		for i := index; i < nextIndex; i++ {
+			if probeCancels[i] != nil {
+				probeCancels[i]()
 			}
 		}
+	}
 
-		if deferWithoutDataSet && len(providersWithDetails) == 0 {
-			deferWithoutDataSet = false
-			for _, deferred := range withoutDataSet {
-				done, err := probeCandidate(deferred)
-				if err != nil {
-					return nil, err
+	startProbes()
+	for inFlight > 0 {
+		result := <-results
+		inFlight--
+		probeCancels[result.index] = nil
+		if !stopping {
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				stopping = true
+				cancelProbes()
+			} else if result.index < probeLimit {
+				if result.err == nil {
+					states[result.index] = providerProbeHealthy
+				} else {
+					states[result.index] = providerProbeUnhealthy
+				}
+				frontier, ready := providerSelectionProgress(states, count)
+				if frontier >= 0 && frontier+1 < probeLimit {
+					// Later candidates cannot enter the stable ranking once this
+					// prefix already contains the requested healthy copies.
+					probeLimit = frontier + 1
+					cancelProbesFrom(probeLimit)
 				}
-				if done {
-					return selected, nil
+				if ready {
+					selectionDetermined = true
+					stopping = true
+					cancelProbes()
 				}
 			}
-			withoutDataSet = nil
+		}
+		startProbes()
+	}
+
+	if err := ctx.Err(); err != nil && !selectionDetermined {
+		return nil, fmt.Errorf("storage.ServiceResolver.ResolveUploadContexts: health-check providers: %w", err)
+	}
+	selected := make([]ResolvedUploadContext, 0, min(count, len(candidates)))
+	failedProviderIDs := make([]types.BigInt, 0, min(count, len(candidates)))
+	for i, state := range states {
+		switch state {
+		case providerProbeHealthy:
+			candidate := candidates[i]
+			selected = append(selected, buildResolvedUploadContext(
+				*candidate.provider,
+				candidate.dataSetID,
+				candidate.clientDataSetID,
+				candidate.metadata,
+			))
+			if len(selected) == count {
+				return selected, nil
+			}
+		case providerProbeUnhealthy:
+			failedProviderIDs = append(failedProviderIDs, candidates[i].provider.Info.ID)
 		}
 	}
 
@@ -540,6 +604,23 @@ func (r *ServiceResolver) autoSelect(ctx context.Context, opts *UploadOptions, e
 	return selected, nil
 }
 
+func providerSelectionProgress(states []providerProbeState, count int) (frontier int, ready bool) {
+	healthy := 0
+	pending := false
+	for i, state := range states {
+		if state == providerProbePending {
+			pending = true
+		}
+		if state == providerProbeHealthy {
+			healthy++
+			if healthy == count {
+				return i, !pending
+			}
+		}
+	}
+	return -1, false
+}
+
 func formatProviderIDs(ids []types.BigInt) string {
 	values := make([]string, len(ids))
 	for i, id := range ids {