indexer: reuse one discovery client per URL (#197)

FileLocationCache.getFileLocation built a ZgsClient for every candidate URL on every
lookup that reached discovery, inside a loop over locations and ports. Each client
carries its own HTTP transport, and Close does not release its connections - measured at
two descriptors per client, the same whether the client succeeded or failed and whether
or not it was closed. So the descriptor count grew with traffic.

Report finding 25 proposed closing the client on the shard-config error path. That fixes
nothing measurable: 50 successful, explicitly closed clients leak exactly as much as 50
unclosed failures. The error path was never the problem.

Key clients by URL and keep them, so the cost is proportional to the number of distinct
nodes seen rather than to the number of lookups. The network bounds the former. Deliberately
no eviction: an evicted client's connections would not be reclaimed either, so eviction
would restore the growth rather than cap it. Close now shuts down every cached client.

The defer inside the loop is gone too - it held every probed client until the enclosing
call returned, which for a multi-port sweep meant all of them at once.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: 0g-peterzhb

indexer/file_location_cache.go

@@ -35,6 +35,7 @@ type FileLocationCache struct {
 	latestFindFile    sync.Map // tx seq -> time.Time
 	latestFailedCall  sync.Map // url -> time.Time
 	latestSuccessCall sync.Map // url -> successCall
+	zgsClients        sync.Map // url -> *node.ZgsClient, kept for the process lifetime
 	discoverNode      *node.AdminClient
 	discoveryPorts    []int
 }
@@ -56,6 +57,40 @@ func (c *FileLocationCache) Close() {
 	if c.discoverNode != nil {
 		c.discoverNode.Close()
 	}
+
+	c.zgsClients.Range(func(_, value any) bool {
+		value.(*node.ZgsClient).Close()
+		return true
+	})
+}
+
+// zgsClient returns a client for url, creating it once and reusing it thereafter.
+//
+// Discovery probes a client per candidate URL on every lookup that reaches it, and each
+// client carries its own HTTP transport whose connections its Close does not release -
+// so constructing one per probe grew the process's descriptor count with traffic. Keying
+// them by URL makes that cost proportional to the number of distinct nodes seen instead,
+// which the network bounds.
+//
+// They are deliberately not evicted: an evicted client's connections would not be
+// reclaimed either, so eviction would restore the unbounded growth rather than cap it.
+func (c *FileLocationCache) zgsClient(url string) (*node.ZgsClient, error) {
+	if cached, ok := c.zgsClients.Load(url); ok {
+		return cached.(*node.ZgsClient), nil
+	}
+
+	client, err := node.NewZgsClient(url, nil, defaultZgsClientOpt)
+	if err != nil {
+		return nil, err
+	}
+
+	// Another goroutine may have won the race; keep whichever is stored and discard ours.
+	if actual, loaded := c.zgsClients.LoadOrStore(url, client); loaded {
+		client.Close()
+		return actual.(*node.ZgsClient), nil
+	}
+
+	return client, nil
 }
 
 func (c *FileLocationCache) GetFileLocations(ctx context.Context, txSeq uint64) ([]*shard.ShardedNode, error) {
@@ -143,11 +178,10 @@ func (c *FileLocationCache) getFileLocation(ctx context.Context, txSeq uint64, c
 						continue
 					}
 				}
-				zgsClient, err := node.NewZgsClient(url, nil, defaultZgsClientOpt)
+				zgsClient, err := c.zgsClient(url)
 				if err != nil {
 					continue
 				}
-				defer zgsClient.Close()
 				fileInfo, err := zgsClient.GetFileInfoByTxSeq(ctx, txSeq)
 				if err != nil {
 					c.latestFailedCall.Store(url, time.Now())

indexer/file_location_cache_test.go

@@ -0,0 +1,104 @@
+package indexer
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// countOpenFDs returns the number of descriptors currently open in this process.
+// Readdirnames is used rather than os.ReadDir because the latter lstats every entry, and
+// on Darwin the directory's own transient descriptor is already gone by then.
+func countOpenFDs(t *testing.T) int {
+	t.Helper()
+
+	for _, dir := range []string{"/proc/self/fd", "/dev/fd"} {
+		dirFile, err := os.Open(dir)
+		if err != nil {
+			continue
+		}
+		names, err := dirFile.Readdirnames(-1)
+		dirFile.Close()
+		if err == nil {
+			return len(names)
+		}
+	}
+
+	t.Skip("cannot enumerate open descriptors on this platform")
+	return 0
+}
+
+func shardConfigServer(t *testing.T) *httptest.Server {
+	t.Helper()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"shardId":0,"numShard":1}}`))
+	}))
+	t.Cleanup(server.Close)
+
+	return server
+}
+
+// Discovery probes a client per candidate URL on every lookup that reaches it. Each
+// client carries its own HTTP transport whose connections Close does not release, so
+// constructing one per probe grew the descriptor count with traffic rather than with the
+// number of distinct nodes.
+func TestFileLocationCache_ReusesOneClientPerURL(t *testing.T) {
+	server := shardConfigServer(t)
+
+	var cache FileLocationCache
+	defer cache.Close()
+
+	first, err := cache.zgsClient(server.URL)
+	require.NoError(t, err)
+	require.NotNil(t, first)
+
+	const iterations = 50
+	before := countOpenFDs(t)
+	for i := 0; i < iterations; i++ {
+		again, err := cache.zgsClient(server.URL)
+		require.NoError(t, err)
+		require.Same(t, first, again, "the same URL must yield the same client")
+	}
+	after := countOpenFDs(t)
+
+	assert.LessOrEqual(t, after, before+2,
+		"repeated lookups leaked descriptors: %d open before, %d after %d lookups", before, after, iterations)
+}
+
+// Distinct URLs get distinct clients — the cost scales with nodes, not with traffic.
+func TestFileLocationCache_DistinctURLsGetDistinctClients(t *testing.T) {
+	one, two := shardConfigServer(t), shardConfigServer(t)
+
+	var cache FileLocationCache
+	defer cache.Close()
+
+	a, err := cache.zgsClient(one.URL)
+	require.NoError(t, err)
+	b, err := cache.zgsClient(two.URL)
+	require.NoError(t, err)
+
+	assert.NotSame(t, a, b)
+}
+
+// A URL whose shard-config lookup fails is not cached, so a later attempt can retry it.
+func TestFileLocationCache_FailedClientIsNotCached(t *testing.T) {
+	failing := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "unavailable", http.StatusInternalServerError)
+	}))
+	defer failing.Close()
+
+	var cache FileLocationCache
+	defer cache.Close()
+
+	_, err := cache.zgsClient(failing.URL)
+	require.Error(t, err)
+
+	_, cached := cache.zgsClients.Load(failing.URL)
+	assert.False(t, cached, "a client that could not be initialized must not be stored")
+}