feat: find endpoints asynchronously, cache errors (#419)

Goals:

1. Make endpoint lookup parallel across all SPs
2. Don't error unless we don't have any endpoints—but log an error
3. Cache the error case too because it's not cheap and with a
misbehaving SP it'll slow down every call for a particular piece

api/api.go

@@ -125,12 +125,17 @@ func InitServer(ctx context.Context, params APIParams) (Server, error) {
 	if err != nil {
 		return Server{}, errors.Wrap(err, "failed to init lassie")
 	}
-	endpointFinder, err := endpointfinder.NewEndpointFinder(replication.MinerInfoFetcher{
+	infoFetcher := replication.MinerInfoFetcher{
 		Client: util.NewLotusClient(params.LotusAPI, params.LotusToken),
-	}, h, 128)
-	if err != nil {
-		return Server{}, errors.Wrap(err, "failed to init endpoint finder")
 	}
+	endpointFinder := endpointfinder.NewEndpointFinder(
+		infoFetcher,
+		h,
+		endpointfinder.WithLruSize(128),
+		endpointfinder.WithLruTimeout(time.Hour*2),
+		endpointfinder.WithErrorLruSize(128),
+		endpointfinder.WithErrorLruTimeout(time.Minute*5),
+	)
 	return Server{
 		db:          db,
 		host:        h,
@@ -300,7 +305,8 @@ func (s Server) setupRoutes(e *echo.Echo) {
 		db *gorm.DB,
 		storageType string,
 		provider string,
-		request storage.CreateRequest) (*model.Storage, error) {
+		request storage.CreateRequest,
+	) (*model.Storage, error) {
 		request.Provider = provider
 		return s.storageHandler.CreateStorageHandler(ctx, db, storageType, request)
 	}))

retriever/endpointfinder/endpointfinder.go

@@ -7,12 +7,16 @@ import (
 
 	"github.com/data-preservation-programs/singularity/replication"
 	"github.com/filecoin-shipyard/boostly"
-	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/hashicorp/golang-lru/v2/expirable"
+	"github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
+	"go.uber.org/multierr"
 )
 
+var logger = log.Logger("singularity/retriever/endpointfinder")
+
 // ErrHTTPNotSupported indicates we were able to look up the provider and contact them,
 // but they reported that they do not serve HTTP retrievals
 var ErrHTTPNotSupported = errors.New("provider does not support http")
@@ -29,23 +33,26 @@ type MinerInfoFetcher interface {
 type EndpointFinder struct {
 	minerInfoFetcher MinerInfoFetcher
 	h                host.Host
-	httpEndpoints    *lru.Cache[string, []peer.AddrInfo]
+	httpEndpoints    *expirable.LRU[string, []peer.AddrInfo]
+	endpointErrors   *expirable.LRU[string, error]
 }
 
 // NewEndpointFinder returns a new instance of an EndpointFinder
-func NewEndpointFinder(minerInfoFetcher MinerInfoFetcher, h host.Host, size int) (*EndpointFinder, error) {
-	httpEndpoints, err := lru.New[string, []peer.AddrInfo](size)
-	if err != nil {
-		return nil, err
-	}
+func NewEndpointFinder(minerInfoFetcher MinerInfoFetcher, h host.Host, opts ...Option) *EndpointFinder {
+	cfg := applyOptions(opts...)
+
+	httpEndpoints := expirable.NewLRU[string, []peer.AddrInfo](cfg.LruSize, func(key string, value []peer.AddrInfo) {}, cfg.LruTimeout)
+	endpointErrors := expirable.NewLRU[string, error](cfg.ErrorLruSize, func(key string, value error) {}, cfg.ErrorLruTimeout)
+
 	return &EndpointFinder{
 		minerInfoFetcher: minerInfoFetcher,
 		h:                h,
 		httpEndpoints:    httpEndpoints,
-	}, nil
+		endpointErrors:   endpointErrors,
+	}
 }
 
-func (ef *EndpointFinder) fetchHTTPEndpoint(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
+func (ef *EndpointFinder) findHTTPEndpointsForProvider(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
 	// lookup the provider on chain
 	minerInfo, err := ef.minerInfoFetcher.GetProviderInfo(ctx, provider)
 	if err != nil {
@@ -74,30 +81,58 @@ func (ef *EndpointFinder) fetchHTTPEndpoint(ctx context.Context, provider string
 	return nil, ErrHTTPNotSupported
 }
 
-// findOrFetchHTTPEndpoint attempts to load from cache before calling fetchHTTPEndpoint
-func (ef *EndpointFinder) findOrFetchHTTPEndpoint(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
-	addrInfos, has := ef.httpEndpoints.Get(provider)
-	if has {
-		return addrInfos, nil
-	}
-	addrInfos, err := ef.fetchHTTPEndpoint(ctx, provider)
-	if err != nil {
-		return nil, err
-	}
-	ef.httpEndpoints.Add(provider, addrInfos)
-	return addrInfos, nil
-}
-
 // FindHTTPEndpoints finds http endpoints for a given set of providers
 func (ef *EndpointFinder) FindHTTPEndpoints(ctx context.Context, sps []string) ([]peer.AddrInfo, error) {
 	addrInfos := make([]peer.AddrInfo, 0, len(sps))
-	for _, sp := range sps {
-		// TODO: should we ignore if some but not all providers are configured correctly?
-		nextAddrInfos, err := ef.findOrFetchHTTPEndpoint(ctx, sp)
-		if err != nil {
-			return nil, err
+	type findResult struct {
+		addrs []peer.AddrInfo
+		err   error
+	}
+	addrChan := make(chan findResult)
+	var toLookup int
+	var errsum error
+
+	for _, provider := range sps {
+		// first check our caches
+		if providerAddrs, has := ef.httpEndpoints.Get(provider); has {
+			addrInfos = append(addrInfos, providerAddrs...)
+		} else if err, has := ef.endpointErrors.Get(provider); has {
+			logger.Errorf("error looking up http endpoint for %s (cached): %s", provider, err)
+			errsum = multierr.Append(errsum, err)
+		} else {
+			// not in caches, perform full lookup of provider asynchronously
+			toLookup++
+			go func(provider string) {
+				providerAddrs, err := ef.findHTTPEndpointsForProvider(ctx, provider)
+				if err != nil {
+					ef.endpointErrors.Add(provider, err)
+					logger.Errorf("error looking up http endpoint for %s: %s", provider, err)
+				} else {
+					ef.httpEndpoints.Add(provider, providerAddrs)
+				}
+				select {
+				case addrChan <- findResult{addrs: providerAddrs, err: err}:
+				case <-ctx.Done():
+				}
+			}(provider)
+		}
+	}
+
+	for i := 0; i < toLookup; i++ {
+		select {
+		case providerAddrs := <-addrChan:
+			if providerAddrs.addrs != nil {
+				addrInfos = append(addrInfos, providerAddrs.addrs...)
+			} else if providerAddrs.err != nil {
+				errsum = multierr.Append(errsum, providerAddrs.err)
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
 		}
-		addrInfos = append(addrInfos, nextAddrInfos...)
+	}
+
+	if len(addrInfos) == 0 {
+		return nil, fmt.Errorf("no http endpoints found for providers %v: %w", sps, errsum)
 	}
 	return addrInfos, nil
 }

retriever/endpointfinder/endpointfinder_test.go

@@ -23,6 +23,7 @@ import (
 func TestEndpointFetcher(t *testing.T) {
 	testCases := []struct {
 		testName                 string
+		providers                int
 		minerInfoNotFindable     bool
 		notDialable              bool
 		notListeningOnTransports bool
@@ -33,27 +34,27 @@ func TestEndpointFetcher(t *testing.T) {
 		{
 			testName:             "unable to find miner on chain",
 			minerInfoNotFindable: true,
-			expectedErrString:    fmt.Errorf("looking up provider info: %w", errMinerNotFound).Error(),
+			expectedErrString:    fmt.Errorf("no http endpoints found for providers [%%s]: looking up provider info: %w", errMinerNotFound).Error(),
 		},
 		{
 			testName:          "unable to dial provider",
 			notDialable:       true,
-			expectedErrString: "querying transports: failed to dial: %s cannot connect to %s",
+			expectedErrString: "no http endpoints found for providers [%s]: querying transports: failed to dial: %s cannot connect to %s",
 		},
 		{
 			testName:                 "provider not listening on protocol",
 			notListeningOnTransports: true,
-			expectedErrString:        "querying transports: failed to negotiate protocol: protocols not supported: [/fil/retrieval/transports/1.0.0]",
+			expectedErrString:        "no http endpoints found for providers [%s]: querying transports: failed to negotiate protocol: protocols not supported: [/fil/retrieval/transports/1.0.0]",
 		},
 		{
 			testName:          "provider not serving http",
 			noHTTP:            true,
-			expectedErrString: endpointfinder.ErrHTTPNotSupported.Error(),
+			expectedErrString: fmt.Errorf("no http endpoints found for providers [%%s]: %w", endpointfinder.ErrHTTPNotSupported).Error(),
 		},
 	}
-	for _, testCase := range testCases {
+	for i, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			testProvider := "t01000"
+			testProvider := fmt.Sprintf("t01000%d", i)
 			mn := mocknet.New()
 			source, err := mn.GenPeer()
 			require.NoError(t, err)
@@ -75,7 +76,6 @@ func TestEndpointFetcher(t *testing.T) {
 			require.NoError(t, err)
 			response := boostly.TransportsQueryResponse{}
 			if !testCase.noHTTP {
-
 				response.Protocols = append(response.Protocols, struct {
 					Name      string                `json:"name,omitempty"`
 					Addresses []multiaddr.Multiaddr `json:"addresses,omitempty"`
@@ -90,8 +90,7 @@ func TestEndpointFetcher(t *testing.T) {
 				other.SetStreamHandler(boostly.FilRetrievalTransportsProtocol_1_0_0, handler)
 			}
 
-			endpointFinder, err := endpointfinder.NewEndpointFinder(minerInfoFetcher, source, 3)
-			require.NoError(t, err)
+			endpointFinder := endpointfinder.NewEndpointFinder(minerInfoFetcher, source, endpointfinder.WithErrorLruSize(3), endpointfinder.WithErrorLruSize(3))
 
 			addrInfos, err := endpointFinder.FindHTTPEndpoints(context.Background(), []string{testProvider})
 			if testCase.expectedErrString == "" {
@@ -111,15 +110,15 @@ func TestEndpointFetcher(t *testing.T) {
 				})
 				require.Equal(t, minerInfoFetcher.callCount, 1)
 			} else {
-				errMessage := fmt.Sprintf(testCase.expectedErrString, source.ID(), other.ID())
+				errMessage := fmt.Sprintf(testCase.expectedErrString, testProvider, source.ID(), other.ID())
 				errMessage = strings.Split(errMessage, "%!(EXTRA")[0]
 				require.EqualError(t, err, errMessage)
 				require.Nil(t, addrInfos)
-				// second call should not cache
+				// second call should cache error
 				addrInfos, err := endpointFinder.FindHTTPEndpoints(context.Background(), []string{testProvider})
 				require.EqualError(t, err, errMessage)
 				require.Nil(t, addrInfos)
-				require.Equal(t, minerInfoFetcher.callCount, 2)
+				require.Equal(t, minerInfoFetcher.callCount, 1)
 			}
 		})
 	}

retriever/endpointfinder/options.go

@@ -0,0 +1,55 @@
+package endpointfinder
+
+import "time"
+
+const (
+	defaultLruSize         = 128
+	defaultLruTimeout      = 2 * time.Hour
+	defaultErrorLruSize    = 128
+	defaultErrorLruTimeout = 5 * time.Minute
+)
+
+type config struct {
+	LruSize         int
+	LruTimeout      time.Duration
+	ErrorLruSize    int
+	ErrorLruTimeout time.Duration
+}
+
+func applyOptions(opts ...Option) *config {
+	cfg := &config{
+		LruSize:         defaultLruSize,
+		ErrorLruSize:    defaultErrorLruSize,
+		ErrorLruTimeout: defaultErrorLruTimeout,
+	}
+	for _, opt := range opts {
+		opt(cfg)
+	}
+	return cfg
+}
+
+type Option func(*config)
+
+func WithLruSize(size int) Option {
+	return func(cfg *config) {
+		cfg.LruSize = size
+	}
+}
+
+func WithLruTimeout(timeout time.Duration) Option {
+	return func(cfg *config) {
+		cfg.LruTimeout = timeout
+	}
+}
+
+func WithErrorLruSize(size int) Option {
+	return func(cfg *config) {
+		cfg.ErrorLruSize = size
+	}
+}
+
+func WithErrorLruTimeout(timeout time.Duration) Option {
+	return func(cfg *config) {
+		cfg.ErrorLruTimeout = timeout
+	}
+}