From 15d13b6cff422c822df2c4074f750458bb1c3d4c Mon Sep 17 00:00:00 2001 From: Eric Li Date: Mon, 2 Nov 2026 18:03:09 -0500 Subject: [PATCH] fix(distribution): time out config fetches and stop leaking connections Previously Nylon opened a new connection for every fetch and never disposed of it. Additionally, established but hung tcp connections are never timed out, also leaking over time. This fixes both issues. --- core/nylon_distribution.go | 8 ++- core/nylon_distribution_test.go | 89 +++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 1 deletion(-) create mode 100644 core/nylon_distribution_test.go diff --git a/core/nylon_distribution.go b/core/nylon_distribution.go index 7f2ef4e7..debc035d 100644 --- a/core/nylon_distribution.go +++ b/core/nylon_distribution.go @@ -8,11 +8,14 @@ import ( "net/url" "os" "slices" + "time" "github.com/encodeous/nylon/state" "github.com/goccy/go-yaml" ) +var fetchConfigTimeout = 30 * time.Second + // fetches and unbundles central config from url func FetchConfig(repoStr string, key state.NyPublicKey, maxSize int64) (*state.CentralCfg, error) { return fetchConfig(repoStr, key, maxSize, state.NewDNSResolver(nil)) @@ -32,9 +35,12 @@ func fetchConfig(repoStr string, key state.NyPublicKey, maxSize int64, resolver } cfgBody = file } else if repo.Scheme == "http" || repo.Scheme == "https" { + // fresh transport per fetch so dont keep connections alive client := &http.Client{ + Timeout: fetchConfigTimeout, Transport: &http.Transport{ - DialContext: resolver.DialContext, + DialContext: resolver.DialContext, + DisableKeepAlives: true, }, } res, err := client.Get(repo.String()) diff --git a/core/nylon_distribution_test.go b/core/nylon_distribution_test.go new file mode 100644 index 00000000..b3e7b551 --- /dev/null +++ b/core/nylon_distribution_test.go @@ -0,0 +1,89 @@ +package core + +import ( + "net" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/encodeous/nylon/state" + "github.com/goccy/go-yaml" + "github.com/stretchr/testify/require" +) + +func testBundle(t *testing.T) (string, state.NyPublicKey) { + t.Helper() + key := state.GenerateKey() + cfg := state.CentralCfg{ + Routers: []state.RouterCfg{{NodeCfg: state.NodeCfg{Id: "a", PubKey: state.GenerateKey().Pubkey()}}}, + } + raw, err := yaml.Marshal(cfg) + require.NoError(t, err) + bundle, err := state.BundleConfig(string(raw), key) + require.NoError(t, err) + return bundle, key.Pubkey() +} + +// polling a repo must not leave a keep-alive connection open on every fetch +func TestFetchConfigDoesNotLeakConnections(t *testing.T) { + bundle, pub := testBundle(t) + var mu sync.Mutex + open := make(map[net.Conn]struct{}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(bundle)) + })) + srv.Config.ConnState = func(c net.Conn, s http.ConnState) { + mu.Lock() + defer mu.Unlock() + switch s { + case http.StateNew: + open[c] = struct{}{} + case http.StateClosed, http.StateHijacked: + delete(open, c) + } + } + srv.Start() + defer srv.Close() + + resolver := state.NewDNSResolver(nil) + for range 20 { + _, err := fetchConfig(srv.URL, pub, 1<<20, resolver) + require.NoError(t, err) + } + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(open) <= 1 + }, time.Second, 10*time.Millisecond, "connections left open after 20 polls") +} + +// a repo that accepts the request but never answers must not block the poller forever +func TestFetchConfigTimesOutOnStalledRepo(t *testing.T) { + _, pub := testBundle(t) + old := fetchConfigTimeout + fetchConfigTimeout = 200 * time.Millisecond + t.Cleanup(func() { fetchConfigTimeout = old }) + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-release: + case <-r.Context().Done(): + } + })) + defer srv.Close() + defer close(release) + + done := make(chan error, 1) + go func() { + _, err := fetchConfig(srv.URL, pub, 1<<20, state.NewDNSResolver(nil)) + done <- err + }() + select { + case err := <-done: + require.Error(t, err) + case <-time.After(10 * time.Second): + t.Fatal("fetchConfig still blocked on a stalled repo") + } +}