diff --git a/CHANGELOG.md b/CHANGELOG.md index 957c7ea..3711bd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,24 @@ All notable changes to this project are documented here. The format follows ## [Unreleased] +### Added +- `Options.IdleTimeout` and `nq --idle-timeout` bound the entire idle probing + phase (default 10 s). A stalled idle response now times out, preserves any + completed samples, warns, and allows the selected load phases to run (#39). + This flat cap also applies to healthy slow paths and existing callers with + large `IdleProbes` values; raise `IdleTimeout` to collect more samples and + reach higher percentile thresholds. + ### Fixed +- Idle-timeout warnings retain the preceding probe error and are omitted + when all requested idle samples have already completed (#39). +- Caller cancellation retains completed idle samples in partial results; + caller deadlines during load report `cancelled` rather than the phase's + `duration_cap`. Discovery, idle, and load phase budgets are documented + together with their teardown and caller-code limitations (#39). +- Cancellation closes an upload's response body and joins its cleanup, so + an HTTP/2 upload peer that responds early and then stops reading cannot + leave the client waiting on flow control after its deadline (#39). - 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`. diff --git a/README.md b/README.md index 7e66a57..74ece59 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ res, err := netquality.Run(ctx, netquality.Cloudflare, netquality.Options{ MaxDuration: 10 * time.Second, // per direction; the budget MaxBytes: 100 << 20, // optional: only on metered links MaxFlows: 8, + IdleTimeout: 5 * time.Second, // budget for all idle probes together }) if err != nil { /* discovery failed, or ctx cancelled (res is then partial) */ } fmt.Println(res.Download.RPM, res.Download.ThroughputBPS, res.Download.Truncated) @@ -102,6 +103,7 @@ nq --well-known nq.example.com:8443 [--insecure] nq --config-url https://host/path/config nq --download-only | --upload-only nq --max-duration 8s --max-flows 8 # time is the budget +nq --idle-timeout 5s # cap the whole idle measurement phase nq --max-bytes 100MB # metered link: add a byte cap nq --json # Result as one JSON document on stdout nq --events # JSON-lines progress on stderr @@ -215,20 +217,40 @@ compromise) and transparent TCP-level proxies that pass TLS through untouched | Limit | Default | Effect | |---|---|---| +| `ConfigTimeout` | 10 s | bounds discovery; failure returns no result | +| `IdleTimeout` | 10 s | bounds all idle probes together; keeps completed samples, warns, then proceeds to load | | `MaxDuration` | 12 s per direction | the budget: phase ends; if not yet stable → `truncated`, `reason=duration_cap`. Cost ≤ rate × 12 s | | `MaxBytes` | **none** (opt-in) | set on metered links; phase ends → `reason=bytes_cap` | | `MaxFlows` | 16 | never more concurrent load connections | | `ctx` cancellation | – | all flows stop within ~200 ms; partial result, `cancelled=true` | -There are no retries, no background goroutines after `Run` returns, and no -telemetry. +The combined phase budget is `ConfigTimeout + IdleTimeout + N × MaxDuration`, +where N is the selected direction count; omit `IdleTimeout` when idle is +skipped. Defaults total 44 s for both directions, 32 s for one, or 34 s for +both with idle skipped. An earlier caller deadline stops the run and retains +completed idle samples and other partial results. Zero or negative duration +options select defaults, so they cannot disable the bounds. +`IdleTimeout` does not grow with `IdleProbes`. Larger sample sets and healthy +slow paths may need a larger timeout to complete all probes and obtain the +requested percentiles. An idle-timeout warning means the measurement budget +ran out; it does not by itself diagnose a faulty connection. + +Return time also includes local orchestration and prompt teardown. Supplied +transports, dialers, body closers, event sinks, and log handlers must honor +cancellation where applicable and return promptly; the library cannot +forcibly stop caller code. Runner goroutines join and owned sockets close +before return. A custom TLS dialer still executing at teardown has its +connection closed when it hands it over. There are no retries or telemetry. + +The client's `--idle-timeout` measures the whole idle probing phase; the +server's flag of the same name limits quiet connections between requests. ## Deviations from the draft | Item | Draft | Here | Why | |---|---|---|---| | Interval (ID) | 5 s | **1 s** | 4 intervals must complete before stability can be declared; with the 12 s per-direction budget a 5 s interval could never stabilise. Earlier drafts and shipping tools use 1 s. Configurable via `Stability.Interval`. | -| Time budget | "implementations may" limit | mandatory `MaxDuration`; `MaxBytes` opt-in | Runs on other people's machines and networks; a time bound makes cost proportional to the link instead of unbounded. | +| Time budget | "implementations may" limit | mandatory discovery, idle, and per-direction time caps; `MaxBytes` opt-in | Runs on other people's machines and networks; a time bound makes cost proportional to the link instead of unbounded. | | Byte cap default | (handoff spec: 250 MB) | none | A fixed byte cap starves fast links of the intervals a confident RPM needs (≈ 8 × rate); the caller knows which networks are metered, the library cannot. | | Flow error | abort the test | abort the **phase**, report `reason=flow_error`, keep other results | Partial data with a flag beats none. | | Self probes on HTTP/1.1 | use TCP RTT estimate | omitted; RPM from foreign probes only, warning recorded | TCP_INFO is not portable in pure Go. | @@ -242,7 +264,7 @@ telemetry. | Config field names | `*_download_url`, `upload_url` | also accepts Apple/Cloudflare `*_https_*` names, preferring them | Interop with deployed servers. | | Cloudflare target | `mach` hardcodes `h3.speed.cloudflare.com` URLs | uses `aim.cloudflare.com/responsiveness/api/v1/config`, which returns the same URLs | Keeps discovery uniform. | -Other constants: `IdleProbes=5` (enough for a median, cheap), `ConfigTimeout=10s`, +Other constants: `IdleProbes=5` (enough for a median, cheap), `ConfigTimeout=10s`, `IdleTimeout=10s`, in-flight probe cap 64 (bounds goroutines on high-RTT links), TLS handshake normalised to 1 RTT for TLS 1.3 and 2 for TLS 1.2. diff --git a/clock_test.go b/clock_test.go index 29ff1f6..7012b19 100644 --- a/clock_test.go +++ b/clock_test.go @@ -1,6 +1,9 @@ package netquality -import "time" +import ( + "context" + "time" +) // fakeClock drives the interval ticker manually; Now, Mono and After use real // time so probes and network I/O keep flowing. coarse makes it claim the @@ -17,7 +20,10 @@ func (f *fakeClock) Now() time.Time { return time.Now() func (f *fakeClock) Mono() instant { return monoNow() } func (f *fakeClock) HighResolution() bool { return !f.coarse } func (f *fakeClock) After(d time.Duration) <-chan time.Time { return time.After(d) } -func (f *fakeClock) NewTicker(time.Duration) ticker { return f } -func (f *fakeClock) C() <-chan time.Time { return f.ch } -func (f *fakeClock) Stop() {} -func (f *fakeClock) tick() { f.ch <- time.Now() } +func (f *fakeClock) WithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, d) +} +func (f *fakeClock) NewTicker(time.Duration) ticker { return f } +func (f *fakeClock) C() <-chan time.Time { return f.ch } +func (f *fakeClock) Stop() {} +func (f *fakeClock) tick() { f.ch <- time.Now() } diff --git a/cmd/nq/idle_timeout_test.go b/cmd/nq/idle_timeout_test.go new file mode 100644 index 0000000..dd4c2c8 --- /dev/null +++ b/cmd/nq/idle_timeout_test.go @@ -0,0 +1,103 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/korya/netquality" + "github.com/korya/netquality/server" +) + +func idleStallServer(t *testing.T, h2, body bool) string { + t.Helper() + var loading atomic.Bool + var small atomic.Int64 + h := server.Handler(server.Options{MaxClientBytes: -1}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.LargePath { + loading.Store(true) + } + if r.URL.Path == server.SmallPath && !loading.Load() && small.Add(1) > 1 { + if body { + w.Header().Set("Content-Length", "1") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + } + <-r.Context().Done() + return + } + h.ServeHTTP(w, r) + })) + srv.EnableHTTP2 = h2 + srv.StartTLS() + t.Cleanup(srv.Close) + return srv.URL + server.ConfigPath +} + +func checkIdleTimeoutOutput(t *testing.T, out []byte) { + t.Helper() + var res netquality.Result + if err := json.Unmarshal(out, &res); err != nil { + t.Fatalf("invalid JSON: %v\n%s", err, out) + } + if res.Idle == nil || res.Idle.Samples != 1 || res.Cancelled || res.Download == nil || res.Upload != nil { + t.Fatalf("idle samples and selected load must survive expiry: %+v", res) + } + var warnings int + for _, w := range res.Warnings { + if strings.Contains(w, "idle timeout (500ms; 1/1000 probes completed)") { + warnings++ + } + } + if warnings != 1 { + t.Errorf("timeout warning missing or repeated: %v", res.Warnings) + } +} + +func TestIdleTimeoutOutput(t *testing.T) { + for _, h2 := range []bool{false, true} { + for _, body := range []bool{false, true} { + t.Run(fmt.Sprintf("h2=%v/body=%v", h2, body), func(t *testing.T) { + var out, errb bytes.Buffer + args := base(idleStallServer(t, h2, body), "--json", "--download-only", "--max-bytes", "1MB", "--idle-probes", "1000", "--idle-timeout", "500ms") + if code := run(args, &out, &errb); code != exitOK { + t.Fatalf("idle expiry must not fail the run: exit=%d stderr=%s", code, errb.String()) + } + if errb.Len() != 0 { + t.Errorf("JSON run must keep stderr quiet: %s", errb.String()) + } + checkIdleTimeoutOutput(t, out.Bytes()) + }) + } + } +} + +func TestIdleTimeoutBinary(t *testing.T) { + bin := filepath.Join(t.TempDir(), "nq.exe") + buildCtx, stopBuild := context.WithTimeout(context.Background(), time.Minute) + defer stopBuild() + if out, err := exec.CommandContext(buildCtx, "go", "build", "-o", bin, ".").CombinedOutput(); err != nil { + t.Fatalf("build: %v\n%s", err, out) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + args := base(idleStallServer(t, true, true), "--json", "--download-only", "--max-bytes", "1MB", "--idle-probes", "1000", "--idle-timeout", "500ms") + cmd := exec.CommandContext(ctx, bin, args...) + var errb bytes.Buffer + cmd.Stderr = &errb + out, err := cmd.Output() + if err != nil { + t.Fatalf("binary did not finish successfully: %v; watchdog=%v stderr=%s", err, ctx.Err(), errb.String()) + } + checkIdleTimeoutOutput(t, out) +} diff --git a/cmd/nq/main.go b/cmd/nq/main.go index efbb93a..ab19c74 100644 --- a/cmd/nq/main.go +++ b/cmd/nq/main.go @@ -48,6 +48,7 @@ func run(args []string, stdout, stderr io.Writer) int { maxBytes = fs.String("max-bytes", "", "per-direction byte cap for metered links (e.g. 100MB, 1GB); omitted = no cap, time bounds the run") maxFlows = fs.Int("max-flows", netquality.DefaultMaxFlows, "maximum concurrent load connections") idleProbes = fs.Int("idle-probes", netquality.DefaultIdleProbes, "number of idle latency probes") + idleTimeout = fs.Duration("idle-timeout", netquality.DefaultIdleTimeout, "time cap for all idle probes; increase for more probes or slow paths") interval = fs.Duration("interval", 0, "stability interval (default 1s; draft says 5s)") insecure = fs.Bool("insecure", false, "skip TLS certificate verification (self-hosted dev servers)") authToken = fs.String("auth-token", os.Getenv("NQ_AUTH_TOKEN"), "bearer token for a protected server (env NQ_AUTH_TOKEN)") @@ -103,6 +104,7 @@ func run(args []string, stdout, stderr io.Writer) int { MaxBytes: mb, MaxFlows: *maxFlows, IdleProbes: *idleProbes, + IdleTimeout: *idleTimeout, } opts.Stability.Interval = *interval if *authToken != "" { diff --git a/docs/architecture.md b/docs/architecture.md index ac38908..b77aa5e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,8 +47,8 @@ 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. | | 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). | +| 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 and discovery/idle/load timeout creation in `run.go` go through the injectable clock; production deadlines use `context.WithTimeout`. | +| Cancellation via context only | Flows read bodies until the context ends; the upload body reader stops on context. Runner goroutines join and owned sockets close before return. Caller code must return promptly; a custom TLS dialer still executing at teardown has its connection closed on handover (INV-4, LIM-9). | | 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`. | | Interval default 1 s, not the draft's 5 s | A 12 s budget cannot fit four 5 s intervals; see README "Deviations" (INV-6). | diff --git a/docs/product-specs/cli.md b/docs/product-specs/cli.md index 7f2996a..f596cff 100644 --- a/docs/product-specs/cli.md +++ b/docs/product-specs/cli.md @@ -13,8 +13,10 @@ An unknown target name is a usage error. ### CLI-2: Limits and parameters `--max-duration`, `--max-bytes` (accepts `250MB`, `1GB`, `100MiB`, plain -bytes), `--max-flows`, `--idle-probes`, `--interval` map onto the library +bytes), `--max-flows`, `--idle-probes`, `--idle-timeout`, `--interval` map onto the library options; invalid sizes are usage errors. +`--idle-timeout` (default 10 s) caps the whole idle measurement phase, not +each probe; a non-positive value selects the default (LIM-8). ### CLI-3: Direction flags `--download-only` and `--upload-only` select one phase; passing both is a diff --git a/docs/product-specs/invariants.md b/docs/product-specs/invariants.md index f4c1acd..b3972eb 100644 --- a/docs/product-specs/invariants.md +++ b/docs/product-specs/invariants.md @@ -22,9 +22,13 @@ A result never implies "not measured" with a zero. Absent phases are absent levels accompany converged values. ### INV-4: No work outlives `Run` -When `Run` returns, all goroutines, connections, and timers it created are -gone. Cancellation propagates to every flow and probe. This covers the -client's own sockets. Abandoning a load flow closes its socket with data still +When `Run` returns, its flow/probe goroutines have joined, its context timers +are canceled, and its owned connections are closed. Cancellation propagates +to every flow and probe. Caller-supplied code must cooperate and return +promptly (LIM-9). A custom TLS dialer may still be inside a transport-owned +dial goroutine with a socket it has not returned; if it returns after +teardown, that connection is closed on handover. This covers the +client's owned sockets. Abandoning a load flow closes its socket with data still unread, which TCP turns into an abortive close, so how long the *server's* socket survives is the server's business, not something the client can promise. diff --git a/docs/product-specs/latency.md b/docs/product-specs/latency.md index cefa8f4..3f70d4f 100644 --- a/docs/product-specs/latency.md +++ b/docs/product-specs/latency.md @@ -12,6 +12,9 @@ Before any load, `IdleProbes` (default 5) sequential GETs of the small resource run, each on a brand-new connection. A negative `IdleProbes` skips the phase and `Result.Idle` is absent. Failed probes are dropped; if none succeed the phase yields no result and a warning. +The whole phase is bounded by `IdleTimeout` (LIM-8). Completed samples are +reported even when the phase times out or the caller cancels before all +requested probes complete; interrupted probes never become latency samples. ### LAT-2: Per-stage timings Every fresh-connection sample records DNS, TCP connect, TLS handshake, @@ -43,8 +46,12 @@ defined as the mean absolute deviation from the mean. Percentiles use the nearest-rank method and appear only when the sample count makes them distinct from the maximum: `p80` from 5 samples, `p90` from 10, `p95` from 20, `p99` from 100. A percentile field never holds a lower percentile than its name; -an absent field means too few samples. With the default 5 idle probes the -idle set reports `p80`; loaded sets usually report all four. +an absent field means too few samples. When all 5 default idle probes +complete, the idle set reports `p80`; loaded sets usually report all four. +For idle percentiles, requesting more probes does not increase the phase +budget. Increase `IdleTimeout` as needed along with `IdleProbes`, especially +on slow paths: only samples completed within that budget count towards +these thresholds, even when every response would eventually succeed. ### LAT-8: Combined loaded latency `loaded.combined` merges foreign and self samples using each probe's diff --git a/docs/product-specs/limits.md b/docs/product-specs/limits.md index 29b360f..52789fa 100644 --- a/docs/product-specs/limits.md +++ b/docs/product-specs/limits.md @@ -26,6 +26,10 @@ per probe (5000 B foreign, 1000 B self); reaching it ends the phase with Cancelling the context stops all flows and probes within about 200 ms. `Run` returns the partial `Result` with `cancelled=true`, the current direction marked `reason=cancelled`, together with the context error. +An earlier caller deadline has the same effect; `duration_cap` identifies +the direction's own time cap. An independently recorded stop (such as a +byte cap or flow error) retains its reason. Completed idle samples survive +cancellation, and directions that have not started remain absent. ### LIM-5: Cancelled results keep the network identity `target.resolved_ips` and `target.local_ips` are populated on cancelled and @@ -38,3 +42,32 @@ amount of traffic. ### LIM-7: Zero and negative options Zero values in `Options` select the defaults; a negative `IdleProbes` skips idle probing. Zero `StabilityParams` fields select the draft defaults. +Non-positive `ConfigTimeout`, `IdleTimeout`, and `MaxDuration` select their +defaults; they never disable the time bounds. + +### LIM-8: Idle phase cap +`IdleTimeout` (default 10 s) bounds the entire idle measurement phase, +regardless of `IdleProbes`. On expiry, completed samples are retained and a +warning names the timeout and successful/requested probe counts. An idle +timeout alone does not cancel the run: selected load phases still execute +with their own budgets. An earlier caller cancellation or deadline stops +the run instead (LIM-4). Skipping idle measurement creates no idle deadline. +The cap does not scale with `IdleProbes`. A healthy slow path or a larger +probe count may exhaust it; callers needing all requested samples must +budget enough time for their sequential fresh-connection probes by raising +`IdleTimeout`. Ten seconds is a configurable product default, not a latency +threshold that distinguishes a healthy path from a faulty one. +A timeout warning retains the last preceding probe error when available. +If every requested sample completed, a deadline observed afterwards does +not produce an idle-timeout warning; caller cancellation still wins (LIM-4). + +### LIM-9: Combined phase budget +Before network work begins, the combined phase budget is `ConfigTimeout` +plus `IdleTimeout` when idle is enabled, plus `MaxDuration` for each selected +direction. Defaults total 44 s for both directions, 32 s for one, or 34 s +for both with idle skipped. An earlier parent deadline takes precedence. +Return time also includes local orchestration and prompt teardown: this is +not an unconditional wall-clock deadline for a descheduled process or +blocking caller code. Supplied transports, dialers, body closers, event +sinks and log handlers must cooperate with cancellation and return promptly; +they cannot be forcibly stopped by the library (INV-4). diff --git a/docs/product-specs/result.md b/docs/product-specs/result.md index 1bb2fe9..e875377 100644 --- a/docs/product-specs/result.md +++ b/docs/product-specs/result.md @@ -55,3 +55,5 @@ one `warning` event per warning. Every event is timestamped. ### RES-8: Sink contract The sink is called synchronously from test goroutines; concurrent calls are possible and the sink must be safe for that. +It must return promptly; a blocked sink cannot be interrupted by a context +deadline and delays phase transitions or teardown (LIM-9). diff --git a/docs/test-matrix.md b/docs/test-matrix.md index 99e0e49..91ca533 100644 --- a/docs/test-matrix.md +++ b/docs/test-matrix.md @@ -29,7 +29,12 @@ directory relative to the repo root; `.` is the library. |---|---|---|---| | Fresh-connection probes | Per-stage medians reported | . | TestRunLoopback | | Skipping | `IdleProbes < 0` skips the phase | . | TestRunDirections | -| Cancellation | Context cancelled during idle probes returns partial result | . | TestCancelDuringIdle | +| Cancellation | Parent cancellation/deadline retains completed idle samples and addresses, omits load, and wins over the idle cap (LIM-4) | . | TestCancelDuringIdle | +| Idle cap | HTTP/1.1 and HTTP/2 stalls before headers or in the body end the whole idle phase, retain samples, warn once, close owned sockets, and continue load (LIM-8) | . | TestIdleTimeout | +| Phase budgets | Injected discovery/idle/load deadlines are canceled on return; skipped idle has none; elapsed return fits budgets plus a scheduling/teardown allowance; early HTTP/2 upload responses with stalled request reads cannot block teardown (LIM-9) | . | TestPhaseTimeoutBudgets | +| Idle timeout diagnostics | Retain the preceding HTTP failure when a later probe times out; a deadline after the final completed sample does not warn (LIM-8) | . | TestIdleTimeoutDiagnostics | +| Cancellation at phase event | A synchronous sink can cancel between the outer guard and load admission; no direction result or load request is created (LIM-4) | . | TestCancelAtLoadPhaseEvent | +| Caller deadline | Parent deadline during load reports cancelled, retaining the current direction and omitting later load (LIM-4) | . | TestCallerDeadlineDuringLoad | | TLS normalisation | TLS 1.2 handshake counted as 2 RTTs, TLS 1.3 as 1 | . | TestTLS12Normalisation | | Statistics | min/median/mean/max/jitter; no percentile at 4 samples | internal/engine | TestStatsOf | | Statistics | Percentile presence thresholds 5/10/20/100 and values; never equal to the max; highest-present helper | internal/engine | TestPercentilePresenceThresholds | @@ -124,6 +129,8 @@ directory relative to the repo root; `.` is the library. | Feature | Scenario | Package | Test | |---|---|---|---| | `--json` | stdout is a `Result`, stderr quiet | cmd/nq | TestJSONOutput | +| `--idle-timeout` | HTTP/1.1 and HTTP/2 header/body stalls retain idle samples and a JSON warning, then complete load with exit 0 (LIM-8, CLI-2) | cmd/nq | TestIdleTimeoutOutput | +| Binary idle cap | Built CLI terminates a stalled idle body within its watchdog and prints partial idle statistics (LIM-8) | cmd/nq | TestIdleTimeoutBinary | | Human output | Table, progress on stderr, `not run` directions | cmd/nq | TestHumanOutput | | `--events` | JSON lines on stderr | cmd/nq | TestEventsOutput | | Exit codes | 0 ok, 1 failed, 2 usage | cmd/nq | TestExitCodes | diff --git a/e2e_test.go b/e2e_test.go index 6ec74ba..d65c518 100644 --- a/e2e_test.go +++ b/e2e_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" @@ -254,19 +255,40 @@ func TestConfigTimeoutAndStatus(t *testing.T) { } func TestCancelDuringIdle(t *testing.T) { - srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == server.SmallPath { - time.Sleep(30 * time.Millisecond) + for _, mode := range []string{"cancel", "deadline"} { + t.Run(mode, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + var attempts atomic.Int64 + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.SmallPath && attempts.Add(1) == 2 { + if mode == "cancel" { + cancel() + } + <-r.Context().Done() + return + } + h.ServeHTTP(w, r) + }) + }, nil, true) + res, err := Run(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: 1000, IdleTimeout: 10 * time.Second, + }) + wantErr := context.Canceled + if mode == "deadline" { + wantErr = context.DeadlineExceeded + } + if !errors.Is(err, wantErr) || res == nil || !res.Cancelled { + t.Fatalf("err=%v res=%+v", err, res) + } + if res.Download != nil || res.Upload != nil || res.Idle == nil || res.Idle.Samples != 1 || attempts.Load() != 2 { + t.Errorf("completed idle samples must survive caller cancellation: %+v", res) + } + if len(res.Target.LocalIPs) == 0 || len(res.Target.ResolvedIPs) == 0 || hasWarning(res, "idle timeout") { + t.Errorf("network identity or parent priority lost: %+v", res) } - h.ServeHTTP(w, r) }) - }, nil, true) - ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) - defer cancel() - res, err := Run(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{HTTPClient: insecureClient(), IdleProbes: 1000}) - if !errors.Is(err, context.DeadlineExceeded) || res == nil || !res.Cancelled || res.Download != nil || res.Idle != nil { - t.Errorf("err=%v res=%+v", err, res) } } @@ -321,7 +343,7 @@ func TestEventStream(t *testing.T) { func TestMaxFlowsDefaultAndOptionsDefaults(t *testing.T) { o := Options{}.withDefaults() if o.MaxDuration != DefaultMaxDuration || o.MaxBytes != 0 || DefaultMaxBytes != 0 || o.MaxFlows != DefaultMaxFlows || - o.IdleProbes != DefaultIdleProbes || o.ConfigTimeout != DefaultConfigTimeout || o.HTTPClient == nil || o.Logger == nil || o.clock == nil { + o.IdleProbes != DefaultIdleProbes || o.IdleTimeout != DefaultIdleTimeout || o.ConfigTimeout != DefaultConfigTimeout || o.HTTPClient == nil || o.Logger == nil || o.clock == nil { t.Errorf("%+v", o) } if (Options{IdleProbes: -1}).withDefaults().IdleProbes != -1 { @@ -330,6 +352,15 @@ func TestMaxFlowsDefaultAndOptionsDefaults(t *testing.T) { if (Options{MaxBytes: -5}).withDefaults().MaxBytes != 0 { t.Error("negative MaxBytes must mean unlimited") } + for _, d := range []time.Duration{0, -time.Second, 123 * time.Millisecond} { + want := d + if d <= 0 { + want = 10 * time.Second + } + if got := (Options{IdleTimeout: d}).withDefaults().IdleTimeout; got != want { + t.Errorf("IdleTimeout=%s: got %s, want %s", d, got, want) + } + } for _, tc := range []struct { d Directions want string diff --git a/idle_timeout_test.go b/idle_timeout_test.go new file mode 100644 index 0000000..d581541 --- /dev/null +++ b/idle_timeout_test.go @@ -0,0 +1,325 @@ +package netquality + +import ( + "context" + "fmt" + "net/http" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/korya/netquality/server" +) + +// A background-context run must survive either kind of idle stall on both +// supported HTTP versions, retaining only the probes that actually finished. +func TestIdleTimeout(t *testing.T) { + for _, h2 := range []bool{false, true} { + for _, body := range []bool{false, true} { + for _, successful := range []int64{0, 1} { + t.Run(fmt.Sprintf("h2=%v/body=%v/samples=%d", h2, body, successful), func(t *testing.T) { + var loading atomic.Bool + var attempts, warningEvents atomic.Int64 + stalled, released := make(chan struct{}), make(chan struct{}) + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.SmallPath && !loading.Load() && attempts.Add(1) > successful { + close(stalled) // idle requests are sequential; only this one can stall + if body { + w.Header().Set("Content-Length", "1") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + } + <-r.Context().Done() + close(released) + return + } + h.ServeHTTP(w, r) + }) + }, nil, h2) + client, sockets := countingClient() + if h2 { + client, sockets = countingTLSClient() + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watchdog := time.AfterFunc(5*time.Second, cancel) + defer watchdog.Stop() + res, err := RunWithEvents(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: client, IdleProbes: 1000, IdleTimeout: 500 * time.Millisecond, + Directions: Download, MaxDuration: 100 * time.Millisecond, MaxBytes: 1 << 20, + }, func(e Event) { + if e.Kind == EventPhase && e.Phase == "download" { + // A custom TLS dial can still be handing its socket + // over after cancellation; match TestNoLeaksAcrossRuns. + closed := sockets.open.Load() == 0 + if h2 { + closed = eventually(time.Second, func() bool { return sockets.open.Load() == 0 }) + } + if !closed { + t.Errorf("idle left %d owned sockets open before load", sockets.open.Load()) + } + loading.Store(true) + } + if e.Kind == EventWarning && strings.Contains(e.Message, "idle timeout") { + warningEvents.Add(1) + } + }) + if err != nil || res == nil { + t.Fatalf("idle timeout must be nonfatal: result=%+v err=%v", res, err) + } + select { + case <-stalled: + default: + t.Fatal("test never reached the stalled idle request") + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("server did not observe idle cancellation") + } + if res.Cancelled || res.Download == nil || res.Upload != nil || !loading.Load() { + t.Fatalf("selected load must run after idle expiry: %+v", res) + } + if successful == 0 && res.Idle != nil || successful > 0 && (res.Idle == nil || res.Idle.Samples != int(successful)) { + t.Errorf("completed samples=%d, idle=%+v", successful, res.Idle) + } + want := fmt.Sprintf("%d/1000 probes completed", successful) + if !hasWarning(res, want) || warningEvents.Load() != 1 || attempts.Load() != successful+1 { + t.Errorf("warnings=%v events=%d attempts=%d", res.Warnings, warningEvents.Load(), attempts.Load()) + } + }) + } + } + } +} + +// Record real timeout contexts rather than substituting the transport. This +// verifies phase budgets and their cancellation without tightly timing I/O. +type budgetClock struct { + realClock + mu sync.Mutex + durations []time.Duration + contexts []context.Context +} + +func (c *budgetClock) WithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + child, cancel := c.realClock.WithTimeout(ctx, d) + c.mu.Lock() + c.durations = append(c.durations, d) + c.contexts = append(c.contexts, child) + c.mu.Unlock() + return child, cancel +} + +func TestPhaseTimeoutBudgets(t *testing.T) { + for _, tc := range []struct { + name string + dir Directions + idle int + }{ + {"both", Both, 1000}, {"download", Download, 1000}, + {"upload", Upload, 1000}, {"skip-idle", Both, -1}, + } { + t.Run(tc.name, func(t *testing.T) { + var loading atomic.Bool + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case server.ConfigPath: + select { + case <-time.After(20 * time.Millisecond): + case <-r.Context().Done(): + return + } + case server.LargePath, server.UploadPath: + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + return + case server.SmallPath: + if !loading.Load() { + <-r.Context().Done() + return + } + } + h.ServeHTTP(w, r) + }) + }, nil, true) + clock := &budgetClock{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watchdog := time.AfterFunc(5*time.Second, cancel) + defer watchdog.Stop() + started := time.Now() + res, err := RunWithEvents(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), ConfigTimeout: time.Second, IdleTimeout: 200 * time.Millisecond, + IdleProbes: tc.idle, Directions: tc.dir, MaxDuration: 100 * time.Millisecond, clock: clock, + }, func(e Event) { + if e.Kind == EventPhase && (e.Phase == "download" || e.Phase == "upload") { + loading.Store(true) + } + }) + elapsed := time.Since(started) + if err != nil || res == nil || res.Cancelled { + t.Fatalf("phase budget did not bound run: res=%+v err=%v", res, err) + } + want := []time.Duration{time.Second} + if tc.idle > 0 { + want = append(want, 200*time.Millisecond) + } else if res.Idle != nil || hasWarning(res, "idle timeout") { + t.Errorf("skipped idle was measured or timed out: %+v", res) + } + for _, dir := range []struct { + selected bool + result *DirectionResult + }{{tc.dir != Upload, res.Download}, {tc.dir != Download, res.Upload}} { + if !dir.selected { + if dir.result != nil { + t.Error("unselected direction ran") + } + continue + } + want = append(want, 100*time.Millisecond) + if dir.result == nil || dir.result.Reason != ReasonDurationCap { + t.Errorf("own direction cap: %+v", dir.result) + } + } + clock.mu.Lock() + defer clock.mu.Unlock() + if !reflect.DeepEqual(clock.durations, want) { + t.Errorf("timeout creation: got %v, want %v", clock.durations, want) + } + var budget time.Duration + for _, d := range want { + budget += d + } + // Include generous scheduling/teardown allowance for CI. Exact + // configured budgets are asserted above; this catches slow returns. + if elapsed > budget+2*time.Second { + t.Errorf("Run took %s for phase budget %s plus 2s allowance", elapsed, budget) + } + for i, child := range clock.contexts { + if child.Err() == nil { + t.Errorf("phase %d context/timer survived Run", i) + } + } + }) + } +} + +func TestCallerDeadlineDuringLoad(t *testing.T) { + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.LargePath { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + return + } + h.ServeHTTP(w, r) + }) + }, nil, true) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + res, err := Run(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: -1, MaxDuration: 10 * time.Second, + }) + if err != context.DeadlineExceeded || res == nil || !res.Cancelled || res.Download == nil { + t.Fatalf("caller deadline: res=%+v err=%v", res, err) + } + if res.Download.Reason != ReasonCancelled || res.Upload != nil { + t.Errorf("parent deadline must cancel the current phase and omit later phases: %+v", res) + } +} + +func TestIdleTimeoutDiagnostics(t *testing.T) { + for _, completed := range []bool{false, true} { + t.Run(fmt.Sprintf("completed=%v", completed), func(t *testing.T) { + var attempts atomic.Int64 + var loading atomic.Bool + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.SmallPath && !loading.Load() && !completed { + if attempts.Add(1) == 1 { + w.WriteHeader(http.StatusForbidden) + return + } + <-r.Context().Done() + return + } + h.ServeHTTP(w, r) + }) + }, nil, true) + probes := 2 + if completed { + probes = 1 + } + clock := &budgetClock{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watchdog := time.AfterFunc(5*time.Second, cancel) + defer watchdog.Stop() + res, err := RunWithEvents(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: probes, IdleTimeout: 500 * time.Millisecond, + Directions: Download, MaxDuration: 100 * time.Millisecond, MaxBytes: 1 << 20, clock: clock, + }, func(e Event) { + if completed && e.Kind == EventProbe && e.Phase == "idle" { + // Force the deadline to land after the final successful sample, + // before idle examines its context. Production sinks return promptly. + clock.mu.Lock() + idleContext := clock.contexts[1] // discovery then idle + clock.mu.Unlock() + <-idleContext.Done() + } + if e.Kind == EventPhase && e.Phase == "download" { + loading.Store(true) + } + }) + if err != nil || res == nil || res.Cancelled || res.Download == nil { + t.Fatalf("idle outcome must allow load: result=%+v err=%v", res, err) + } + if completed { + if res.Idle == nil || res.Idle.Samples != probes || hasWarning(res, "idle timeout") { + t.Errorf("completed sample set must not warn about idle timeout: %+v", res) + } + } else if res.Idle != nil || !hasWarning(res, "idle timeout") || !hasWarning(res, "last probe error: unexpected status 403 Forbidden") || attempts.Load() != 2 { + t.Errorf("preceding probe failure lost: result=%+v attempts=%d", res, attempts.Load()) + } + }) + } +} + +func TestCancelAtLoadPhaseEvent(t *testing.T) { + for _, dir := range []Directions{Download, Upload} { + t.Run(dir.String(), func(t *testing.T) { + var requests atomic.Int64 + srv := startServer(t, server.Options{}, func(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == server.LargePath || r.URL.Path == server.UploadPath { + requests.Add(1) + } + h.ServeHTTP(w, r) + }) + }, nil, true) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + res, err := RunWithEvents(ctx, Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: -1, Directions: dir, + }, func(e Event) { + if e.Kind == EventPhase && e.Phase == dir.String() { + cancel() // runs between the outer guard and loadPhase's guard + } + }) + if err != context.Canceled || res == nil || !res.Cancelled { + t.Fatalf("phase event cancellation: result=%+v err=%v", res, err) + } + if res.Download != nil || res.Upload != nil || requests.Load() != 0 { + t.Errorf("cancellation must prevent load admission: result=%+v requests=%d", res, requests.Load()) + } + }) + } +} diff --git a/load.go b/load.go index 5702bb1..a8df62a 100644 --- a/load.go +++ b/load.go @@ -150,7 +150,27 @@ func oneRequest(ctx context.Context, f *flow, dir Directions, url string, c *byt if err != nil { return err } - defer resp.Body.Close() + if dir == Upload { + // An HTTP/2 peer can respond before consuming the upload and then + // stop reading. Once RoundTrip returns headers, cancellation alone + // may not wake its writer from flow control. Closing the response + // body aborts the stream and wakes both the writer and our reader. + // Join the callback and close exactly once, even on normal completion. + closed := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + _ = resp.Body.Close() + close(closed) + }) + defer func() { + if stop() { + _ = resp.Body.Close() + } else { + <-closed + } + }() + } else { + defer resp.Body.Close() + } proto := resp.Proto f.proto.Store(&proto) if err := checkStatus(resp); err != nil { diff --git a/options.go b/options.go index 998fa95..3303655 100644 --- a/options.go +++ b/options.go @@ -1,6 +1,7 @@ package netquality import ( + "context" "github.com/korya/netquality/internal/engine" "log/slog" "net/http" @@ -56,6 +57,9 @@ const ( // DefaultConfigTimeout bounds config discovery, which happens before the // per-direction budgets apply. DefaultConfigTimeout = 10 * time.Second + // DefaultIdleTimeout bounds the whole idle measurement phase, regardless + // of how many probes were requested. + DefaultIdleTimeout = 10 * time.Second ) // Options configures a test run. The zero value is valid and uses the defaults. @@ -65,8 +69,8 @@ type Options struct { MaxDuration time.Duration // MaxBytes, if > 0, bounds the bytes moved by each direction's load // phase, probes included; hitting it truncates the phase with - // reason=bytes_cap. 0 (the default) means no byte cap: MaxDuration alone - // bounds the run. Set it on metered links. + // reason=bytes_cap. 0 (the default) means no byte cap: MaxDuration + // bounds each load phase. Set it on metered links. MaxBytes int64 // MaxFlows caps the number of concurrent load-generating connections // (default 16, draft MNP). @@ -76,6 +80,13 @@ type Options struct { // IdleProbes is the number of fresh-connection probes for idle latency // (default 5). 0 uses the default; negative skips idle measurement. IdleProbes int + // IdleTimeout bounds the whole idle measurement phase (default 10s), not + // each probe. On expiry, successful samples are kept, a warning is recorded, + // and the selected load phases still run. Non-positive values use the + // default. An earlier caller deadline or cancellation stops the whole run. + // A healthy slow path or a larger IdleProbes count may require a larger + // IdleTimeout to collect all requested samples and their percentiles. + IdleTimeout time.Duration // Stability holds the draft's algorithm parameters; zero fields use // defaults. SendBufferBytes applies to upload phases only and defaults to // DefaultUploadSendBuffer there; set it negative to disable. @@ -83,7 +94,9 @@ type Options struct { // HTTPClient supplies the base transport (proxy, TLS config, dialer). Only // its Transport is used; each load flow gets its own clone so flows do not // share a connection. If the Transport is not an *http.Transport it is used - // as-is and flows may share connections (a warning is recorded). + // as-is and flows may share connections (a warning is recorded). Custom + // transports and dialers must honor cancellation and close promptly; + // Run cannot forcibly stop caller-supplied code. HTTPClient.Timeout is unused. HTTPClient *http.Client // Logger receives debug logs; nil discards them. Logger *slog.Logger @@ -116,6 +129,9 @@ func (o Options) withDefaults() Options { if o.ConfigTimeout <= 0 { o.ConfigTimeout = DefaultConfigTimeout } + if o.IdleTimeout <= 0 { + o.IdleTimeout = DefaultIdleTimeout + } o.Stability = o.Stability.WithDefaults() if o.HTTPClient == nil { o.HTTPClient = &http.Client{} @@ -141,6 +157,7 @@ type clock interface { HighResolution() bool NewTicker(d time.Duration) ticker After(d time.Duration) <-chan time.Time + WithTimeout(context.Context, time.Duration) (context.Context, context.CancelFunc) } type ticker interface { @@ -155,6 +172,9 @@ func (realClock) Mono() instant { return monoNow() } func (realClock) HighResolution() bool { return monoHighResolution() } func (realClock) NewTicker(d time.Duration) ticker { return realTicker{time.NewTicker(d)} } func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) } +func (realClock) WithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, d) +} type realTicker struct{ t *time.Ticker } diff --git a/run.go b/run.go index 0eba2f9..76c3faa 100644 --- a/run.go +++ b/run.go @@ -133,6 +133,7 @@ func (r *runner) run(ctx context.Context, t Target) (*Result, error) { if r.opts.IdleProbes > 0 { r.emit(Event{Kind: EventPhase, Phase: "idle"}) idle, err := r.idle(ctx) + r.res.Idle = idle // retain completed samples even when the phase ends early if ctx.Err() != nil { r.res.Cancelled = true finish() @@ -140,8 +141,6 @@ func (r *runner) run(ctx context.Context, t Target) (*Result, error) { } if err != nil { r.warn("idle latency: %v", err) - } else { - r.res.Idle = idle } } @@ -153,6 +152,11 @@ func (r *runner) run(ctx context.Context, t Target) (*Result, error) { dirs = []Directions{Upload} } for _, d := range dirs { + if ctx.Err() != nil { + r.res.Cancelled = true + finish() + return r.res, ctx.Err() + } r.emit(Event{Kind: EventPhase, Phase: d.String(), Direction: d.String()}) dr, err := r.loadPhase(ctx, d) if d == Download { @@ -186,7 +190,7 @@ func (r *runner) discover(ctx context.Context, t Target) (*ServerConfig, error) if t.ConfigURL == "" { return nil, errors.New("netquality: empty config URL") } - cctx, cancel := context.WithTimeout(ctx, r.opts.ConfigTimeout) + cctx, cancel := r.opts.clock.WithTimeout(ctx, r.opts.ConfigTimeout) defer cancel() req, err := http.NewRequestWithContext(cctx, http.MethodGet, t.ConfigURL, nil) if err != nil { @@ -228,6 +232,8 @@ func (r *runner) discover(ctx context.Context, t Target) (*ServerConfig, error) // idle measures idle latency with sequential fresh-connection probes. func (r *runner) idle(ctx context.Context) (*LatencyStats, error) { + ctx, cancel := r.opts.clock.WithTimeout(ctx, r.opts.IdleTimeout) + defer cancel() rt := r.factory.newTransport(false) defer closeIdle(rt) var samples []LatencySample @@ -238,20 +244,34 @@ func (r *runner) idle(ctx context.Context) (*LatencyStats, error) { } s, err := foreignProbe(ctx, rt, r.cfg.SmallDownloadURL, r.opts.Header, r.opts.clock.Mono, r.observeTLS) if err != nil { + if ctx.Err() != nil && errors.Is(err, ctx.Err()) { + break // retain the preceding probe failure as timeout context + } lastErr = err continue } samples = append(samples, s) r.emit(Event{Kind: EventProbe, Phase: "idle", ProbeKind: "idle", Latency: s.Total}) } - if len(samples) == 0 { + var st *LatencyStats + if len(samples) > 0 { + stats := engine.ComputeLatencyStats(samples) + st = &stats + } + if ctx.Err() != nil && len(samples) < r.opts.IdleProbes { + err := fmt.Errorf("idle timeout (%s; %d/%d probes completed): %w", r.opts.IdleTimeout, len(samples), r.opts.IdleProbes, ctx.Err()) + if lastErr != nil { + err = fmt.Errorf("%w; last probe error: %v", err, lastErr) + } + return st, err + } + if st == nil { if lastErr == nil { lastErr = errors.New("no samples") } return nil, lastErr } - st := engine.ComputeLatencyStats(samples) - return &st, nil + return st, nil } // phaseState is the mutable state of one load phase. @@ -364,13 +384,16 @@ func (p *phaseState) proto() string { // loadPhase runs one direction: ramp flows, probe, evaluate stability, stop on // stability or a limit. func (r *runner) loadPhase(ctx context.Context, dir Directions) (*DirectionResult, error) { + if err := ctx.Err(); err != nil { + return nil, err + } sp := r.opts.Stability if dir == Upload && sp.SendBufferBytes == 0 { // Upload bytes are counted when the transport takes them, ahead of // the wire by up to the HTTP/2 stream window per flow. sp.SendBufferBytes = DefaultUploadSendBuffer } - pctx, cancel := context.WithTimeout(ctx, r.opts.MaxDuration) + pctx, cancel := r.opts.clock.WithTimeout(ctx, r.opts.MaxDuration) defer cancel() p := &phaseState{dir: dir, cancel: cancel} @@ -449,7 +472,9 @@ loop: // Determine why we stopped, then tear everything down. stop() is // once-guarded, so a flow or probe goroutine that already named a reason // wins; reading p.reason here to pre-empt it would race with them. - if pctx.Err() != nil { + if ctx.Err() != nil { + p.stop(ReasonCancelled) + } else if pctx.Err() != nil { p.stop(ctxReason(pctx)) } cancel()