diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a2f14d..957c7ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ All notable changes to this project are documented here. The format follows ## [Unreleased] +### Fixed +- Connections opened through a caller-supplied `DialTLSContext` (or `DialTLS`) + were not tracked, so an HTTP/2 connection still winding down when `Run` + returned outlived it — for ever, on a transport without `IdleConnTimeout`. + They are now torn down with the run (INV-4). `test_endpoint` cannot be + honoured through a custom TLS dialer; the run now warns instead of silently + ignoring it (DISC-9). + ## [0.4.0] - 2026-08-29 The measurement algorithm reaches a confident answer on links the previous diff --git a/docs/architecture.md b/docs/architecture.md index ce0be8f..ac38908 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -46,7 +46,7 @@ real protocol in-process over TLS + HTTP/2 without a network. |---|---| | One `*http.Transport` per load flow | Each flow is its own TCP/TLS connection (draft requirement); self probes multiplex onto it via HTTP/2. A user-supplied non-`*http.Transport` cannot be cloned, so flows may share connections and the library warns. | | Fresh connection per foreign/idle probe | `DisableKeepAlives` on a dedicated transport; `httptrace` supplies DNS/connect/TLS/TTFB stages. | -| `DialContext` wrapper sees every connection | It implements `test_endpoint`, records remote/local IPs, and is bypassed by user `DialTLSContext` or custom `RoundTripper`s (documented limitation). | +| The dial wrappers see every connection | `DialContext` implements `test_endpoint` and records remote/local IPs. A user `DialTLSContext`/`DialTLS` bypasses it for https, so it is wrapped too: its connections are tracked for teardown and recorded, but not redirected to `test_endpoint` (DISC-9). A custom `RoundTripper` is opaque (documented limitation). | | The engine is pure | `internal/engine` sees only `Observation`s (elapsed, bytes, flows, probe samples) and returns `Decision`s; it holds no clock, goroutine, or socket, so the same code runs against real transports and, in tests, against recorded series or a simulator. Public latency/confidence types are aliases of engine types. The interval loop in `run.go` is driven by an injectable clock. | | Cancellation via context only | Flows read bodies until the context ends; the upload body reader stops on context; no goroutine outlives `Run` (INV-4). | | Byte accounting is client-side | Upload bytes are counted when handed to the transport, so a few MB of HTTP/2 flow-control window may be in flight beyond `MaxBytes`. | diff --git a/docs/product-specs/discovery.md b/docs/product-specs/discovery.md index 35c21ac..e249327 100644 --- a/docs/product-specs/discovery.md +++ b/docs/product-specs/discovery.md @@ -47,3 +47,10 @@ rejects them fails the run at discovery, before any load traffic. ### DISC-7: `test_endpoint` under a proxy When an explicit proxy is in use the override cannot be honoured; the proxy dials the origin, and a warning names the ignored endpoint. + +### DISC-9: `test_endpoint` with a custom TLS dialer +When the caller's transport sets `DialTLSContext` (or `DialTLS`) the override +cannot be honoured either: the caller's dialer would verify the certificate +against the rewritten address. The dialer is called with the URLs' host, its +connections are still tracked and torn down with the run (INV-4), and a +warning names the ignored endpoint. diff --git a/docs/test-matrix.md b/docs/test-matrix.md index fa1b5ec..99e0e49 100644 --- a/docs/test-matrix.md +++ b/docs/test-matrix.md @@ -21,6 +21,7 @@ directory relative to the repo root; `.` is the library. | Discovery over HTTPS | `ConfigTimeout` bounds a slow server; 429 is reported | . | TestConfigTimeoutAndStatus | | test_endpoint | Dial override honoured (URLs name an unresolvable host) | . | TestTestEndpointHonoured | | test_endpoint | Ignored under an explicit proxy, with warning | . | TestExplicitProxyDetected | +| test_endpoint | Ignored with a custom TLS dialer, with warning; HTTP/2 kept | . | TestTestEndpointCustomTLSDialer | ## Idle latency @@ -182,7 +183,8 @@ than describe a feature. Goldens are regenerated deliberately with | Guard | Scenario | Package | Test | |---|---|---|---| -| INV-4 | Eight mixed runs (download, upload, cancelled) leave no goroutine and no open client socket behind, checked the instant `Run` returns | . | TestNoLeaksAcrossRuns | +| INV-4 | Eight mixed runs (download, upload, cancelled) leave no goroutine and no open client socket behind, checked the instant `Run` returns; with `DialContext` and with `DialTLSContext` | . | TestNoLeaksAcrossRuns | +| INV-4 | 40-run soak per scenario (completed, cancelled, custom TLS dialer without idle timeout): goroutine count and post-GC `HeapInuse` plateau; skipped under `-short` | . | TestSoak | | Wire contract | Identity encoding on every request, octet-stream POST uploads, GET elsewhere, fresh connection per idle/foreign probe, self probes on load connections | . | TestWireContract | | No global state | Differently configured runs in one process do not influence each other; each result's warnings name its own budget and no other run's | . | TestRepeatedRunsAreIndependent | | INV-7 | Every JSON path and kind of `Result` pinned in `testdata/result_schema.txt`; snake_case enforced | . | TestResultSchemaGolden | diff --git a/e2e_test.go b/e2e_test.go index f5b7a03..6ec74ba 100644 --- a/e2e_test.go +++ b/e2e_test.go @@ -351,3 +351,23 @@ func hasWarning(res *Result, substr string) bool { } return false } + +// TestTestEndpointCustomTLSDialer (DISC-9): a custom TLS dialer cannot honour +// test_endpoint, and the run says so instead of silently ignoring it. +func TestTestEndpointCustomTLSDialer(t *testing.T) { + srv := startServer(t, server.Options{TestEndpoint: "192.0.2.1"}, nil, nil, true) + client, _ := countingTLSClient() + res, err := Run(context.Background(), Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: client, Directions: Download, IdleProbes: 1, + MaxDuration: 300 * time.Millisecond, MaxBytes: 1 << 40, Stability: fastStability(), + }) + if err != nil { + t.Fatal(err) + } + if !hasWarning(res, "custom TLS dialer") { + t.Errorf("warnings = %v", res.Warnings) + } + if res.Download.HTTPVersion != "HTTP/2.0" { + t.Errorf("tracking must not cost HTTP/2: got %q", res.Download.HTTPVersion) + } +} diff --git a/regression_test.go b/regression_test.go index 8d306a2..a8aa227 100644 --- a/regression_test.go +++ b/regression_test.go @@ -113,13 +113,50 @@ func eventually(d time.Duration, cond func() bool) bool { return cond() } +// countingTLSClient is countingClient with the sockets opened through +// DialTLSContext, which net/http uses instead of DialContext for https. +func countingTLSClient() (*http.Client, *countingDialer) { + d := &countingDialer{inner: net.Dialer{Timeout: 10 * time.Second}} + tr := &http.Transport{ForceAttemptHTTP2: true} // IdleConnTimeout 0: idle conns never expire on their own + tr.DialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + raw, err := d.DialContext(ctx, network, addr) + if err != nil { + return nil, err + } + tc := tls.Client(raw, &tls.Config{InsecureSkipVerify: true, NextProtos: []string{"h2", "http/1.1"}}) //nolint:gosec // test server + if err := tc.HandshakeContext(ctx); err != nil { + _ = raw.Close() + return nil, err + } + return tc, nil + } + return &http.Client{Transport: tr}, d +} + // TestNoLeaksAcrossRuns guards INV-4: after Run returns, no goroutine it // started survives and every connection it opened is closed — measured over // repeated runs in one process, the way an agent uses the library. func TestNoLeaksAcrossRuns(t *testing.T) { + for _, tc := range []struct { + name string + client func() (*http.Client, *countingDialer) + settle bool + }{{"DialContext", countingClient, false}, {"DialTLSContext", countingTLSClient, true}} { + t.Run(tc.name, func(t *testing.T) { testNoLeaksAcrossRuns(t, tc.client, tc.settle) }) + } +} + +// settle allows the sockets to reach zero shortly after Run returns instead of +// at the instant it does. It is set for a caller-supplied TLS dialer: net/http +// dials in a goroutine that outlives the cancelled request that started it, so +// when a phase ends, that goroutine can still be inside the caller's dialer, +// handshaking a socket the library has not been given and cannot close. What +// the library owes is that such a connection dies the moment it is handed over +// (ownedTransport.closed), never that it was never opened. +func testNoLeaksAcrossRuns(t *testing.T, newClient func() (*http.Client, *countingDialer), settle bool) { srv, _ := startCountingServer(t, nil) target := Target{ConfigURL: srv.URL + server.ConfigPath} - client, dialer := countingClient() + client, dialer := newClient() opts := Options{HTTPClient: client, IdleProbes: 2, MaxFlows: 6, MaxDuration: 300 * time.Millisecond, MaxBytes: 1 << 40, Stability: fastStability()} @@ -149,8 +186,12 @@ func TestNoLeaksAcrossRuns(t *testing.T) { t.Fatalf("run %d: %v", i, err) } // INV-4 is a promise about the moment Run returns, so check it then — - // not after a grace period. - if n := dialer.open.Load(); n != 0 { + // not after a grace period (see settle for the one exception). + if settle { + if !eventually(3*time.Second, func() bool { return dialer.open.Load() == 0 }) { + t.Fatalf("run %d: %d sockets still open after Run returned", i, dialer.open.Load()) + } + } else if n := dialer.open.Load(); n != 0 { t.Fatalf("run %d: %d sockets still open when Run returned", i, n) } } diff --git a/run.go b/run.go index 19b8505..0eba2f9 100644 --- a/run.go +++ b/run.go @@ -126,6 +126,9 @@ func (r *runner) run(ctx context.Context, t Target) (*Result, error) { r.warn("test_endpoint %q is ignored because a proxy dials the origin", cfg.TestEndpoint) } } + if r.factory.customTLS && r.factory.testEndpoint != "" { + r.warn("test_endpoint %q is ignored because the transport has a custom TLS dialer", cfg.TestEndpoint) + } if r.opts.IdleProbes > 0 { r.emit(Event{Kind: EventPhase, Phase: "idle"}) diff --git a/soak_test.go b/soak_test.go new file mode 100644 index 0000000..bf1a0e5 --- /dev/null +++ b/soak_test.go @@ -0,0 +1,101 @@ +package netquality + +import ( + "context" + "crypto/tls" + "fmt" + "net" + "net/http" + "runtime" + "testing" + "time" + + "github.com/korya/netquality/server" +) + +// TestSoak loops Run and checks that goroutines and heap plateau. The server +// runs in-process, so a leaked client connection shows up as server +// goroutines too. +func TestSoak(t *testing.T) { + if testing.Short() { + t.Skip("soak") + } + const warm, iters = 5, 40 + scenarios := []struct { + name string + client func() *http.Client + cancel time.Duration // cancel ctx this long into the run; 0 = let it finish + }{ + {"default", insecureClient, 0}, + {"default-cancel", insecureClient, 250 * time.Millisecond}, + {"dialTLSContext-noIdleTimeout", func() *http.Client { + tr := &http.Transport{ + DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + d := tls.Dialer{Config: &tls.Config{InsecureSkipVerify: true, NextProtos: []string{"h2", "http/1.1"}}} //nolint:gosec + return d.DialContext(ctx, network, addr) + }, + ForceAttemptHTTP2: true, + IdleConnTimeout: 0, + } + return &http.Client{Transport: tr} + }, 0}, + } + for _, sc := range scenarios { + t.Run(sc.name, func(t *testing.T) { + srv := startServer(t, server.Options{}, nil, nil, true) + client := sc.client() + sample := func() (int, uint64) { + runtime.GC() + time.Sleep(50 * time.Millisecond) // let closed conns' goroutines exit + runtime.GC() + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + return runtime.NumGoroutine(), ms.HeapInuse + } + one := func() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if sc.cancel > 0 { + time.AfterFunc(sc.cancel, cancel) + } + _, err := Run(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: client, IdleProbes: 2, + MaxDuration: 400 * time.Millisecond, MaxBytes: 1 << 40, Stability: fastStability(), + }) + if err != nil && ctx.Err() == nil { + t.Fatal(err) + } + } + for i := 0; i < warm; i++ { + one() + } + g0, h0 := sample() + for i := 0; i < iters; i++ { + one() + if i%10 == 9 { + g, h := sample() + fmt.Printf("%-30s iter %2d goroutines %3d (%+d) heapInuse %6.2fMiB (%+.2f)\n", sc.name, i+1, g, g-g0, float64(h)/1048576, float64(int64(h)-int64(h0))/1048576) + } + } + // Poll for the plateau rather than demanding it at one instant. The + // counts include the in-process server's per-connection goroutines, + // and the client aborts its flows, so the server side unwinds on the + // kernel's schedule, not ours (INV-4 covers the client's sockets + // only). The leak this guards against is ~3 goroutines and ~150KiB + // per run, so a settled bound of +2 and +8MiB still catches it many + // times over. + var g1 int + var h1 uint64 + ok := eventually(15*time.Second, func() bool { + g1, h1 = sample() + return g1-g0 <= 2 && h1 <= h0+8<<20 + }) + if !ok { + buf := make([]byte, 1<<16) + n := runtime.Stack(buf, true) + t.Errorf("no plateau over %d runs: goroutines %d -> %d, heapInuse %d -> %d\n%s", + iters, g0, g1, h0, h1, buf[:n]) + } + }) + } +} diff --git a/transport.go b/transport.go index d9bb98e..390ba0c 100644 --- a/transport.go +++ b/transport.go @@ -23,6 +23,7 @@ type transportFactory struct { urlHost string // host[:port] from the config URLs urlHostPort string // urlHost with the scheme's default port filled in dialTimeout time.Duration + customTLS bool // the base transport has DialTLSContext/DialTLS remote addrSet // server IPs the flows connected to local addrSet // source IPs the flows went out on } @@ -80,6 +81,7 @@ func newTransportFactory(client *http.Client, cfg *ServerConfig, u *url.URL) (*t } if t, ok := rt.(*http.Transport); ok { f.base = t + f.customTLS = t.DialTLSContext != nil || t.DialTLS != nil //nolint:staticcheck // DialTLS is deprecated but still honoured by net/http } else { f.custom = rt warnings = append(warnings, "custom RoundTripper in use: load flows may share connections, probes may reuse connections (no per-stage timings), test_endpoint is ignored") @@ -103,16 +105,44 @@ type ownedTransport struct { *http.Transport mu sync.Mutex conns map[net.Conn]struct{} + // closed latches at teardown. net/http dials in a goroutine that outlives + // the cancelled request that started it, so a dial can still complete after + // closeAll has taken its snapshot; that connection would otherwise be filed + // in a map nobody reads again and stay open for ever. Closing it on arrival + // costs six lines. In practice the abandoned dial is cancelled with the + // phase and never gets this far — this was not observed firing. + closed bool } func (o *ownedTransport) track(c net.Conn) net.Conn { tc := &trackedConn{Conn: c, owner: o} o.mu.Lock() + if o.closed { + o.mu.Unlock() + _ = tc.Close() + return tc + } o.conns[tc] = struct{}{} o.mu.Unlock() return tc } +// trackRaw registers c without wrapping it. net/http only upgrades to HTTP/2 +// when the dialled value is exactly a *tls.Conn, so connections from a +// caller's DialTLSContext cannot be wrapped; they stay in the map until +// closeAll, which is bounded by the dials of one run. +func (o *ownedTransport) trackRaw(c net.Conn) net.Conn { + o.mu.Lock() + if o.closed { + o.mu.Unlock() + _ = c.Close() + return c + } + o.conns[c] = struct{}{} + o.mu.Unlock() + return c +} + func (o *ownedTransport) forget(c net.Conn) { o.mu.Lock() delete(o.conns, c) @@ -124,6 +154,7 @@ func (o *ownedTransport) forget(c net.Conn) { func (o *ownedTransport) closeAll() { o.CloseIdleConnections() o.mu.Lock() + o.closed = true conns := make([]net.Conn, 0, len(o.conns)) for c := range o.conns { conns = append(conns, c) @@ -190,6 +221,30 @@ func (f *transportFactory) newTransport(keepAlive bool) http.RoundTripper { } return owned.track(c), nil } + if f.customTLS { + // A custom TLS dialer bypasses DialContext for https, so track its + // connections here. The address is not rewritten for test_endpoint: + // the caller's dialer would verify the certificate against it. + innerTLS := t.DialTLSContext + if innerTLS == nil { + legacy := t.DialTLS //nolint:staticcheck // deprecated but still honoured by net/http + innerTLS = func(_ context.Context, network, addr string) (net.Conn, error) { return legacy(network, addr) } + } + t.DialTLS = nil //nolint:staticcheck // DialTLSContext takes precedence; make that explicit + t.DialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + c, err := innerTLS(ctx, network, addr) + if err != nil { + return nil, err + } + if ra := c.RemoteAddr(); ra != nil { + f.remote.add(hostOnly(ra.String())) + } + if la := c.LocalAddr(); la != nil { + f.local.add(hostOnly(la.String())) + } + return owned.trackRaw(c), nil + } + } return owned }