From 73875d2c3fe44edcd20620b5db0ded1156a66fe7 Mon Sep 17 00:00:00 2001 From: korya <148461+korya@users.noreply.github.com> Date: Thu, 10 Sep 2026 00:15:10 -0400 Subject: [PATCH 1/3] fix(probes): Bound responses with a ten-byte compatibility ceiling Reject empty and oversized probe responses before they can distort latency or drain arbitrary payloads against a fixed accounting estimate. Accept complete 1-10 byte bodies to preserve Apple and Cloudflare compatibility. Retain valid idle samples, report bounded size warnings per phase and probe kind, and preserve shared HTTP/2 load streams on rejection. Document the read limit separately from estimated costs and transport buffering. Add real HTTP/1.1 and HTTP/2 boundary, cancellation, isolation, accounting, warning and CLI coverage, with matching specs and matrix rows. Closes #40 Co-Authored-By: GPT-6 --- CHANGELOG.md | 7 + README.md | 13 +- cmd/nq/probe_size_test.go | 90 +++++++ docs/architecture.md | 2 +- docs/product-specs/latency.md | 11 + docs/product-specs/limits.md | 12 + docs/product-specs/result.md | 6 + docs/test-matrix.md | 6 + options.go | 2 + probe.go | 25 +- probe_size_test.go | 434 ++++++++++++++++++++++++++++++++++ run.go | 21 ++ 12 files changed, 624 insertions(+), 5 deletions(-) create mode 100644 cmd/nq/probe_size_test.go create mode 100644 probe_size_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3711bd4..3ac4771 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ All notable changes to this project are documented here. The format follows reach higher percentile thresholds. ### Fixed +- Small probe responses are limited to complete, nonempty bodies of 1–10 + bytes, preserving Apple and Cloudflare compatibility. Declared oversized + bodies are rejected before reading, and streamed bodies after at most 11 + bytes. Invalid sizes no longer become latency samples; bounded warnings + explain rejection while valid samples and load measurements survive (#40). + Previously accepted empty or larger custom-server responses are now invalid. + Probe cost remains estimated; transport buffering can exceed body-read limits. - 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; diff --git a/README.md b/README.md index 74ece59..2c6b16a 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,8 @@ Cost 370.5 MB moved in 21.0s 1. **Discovery** – fetches the server's JSON config (`/.well-known/nq` or a vendor path) to learn the small-download, large-download and upload URLs. -2. **Idle latency** – `IdleProbes` sequential GETs of the 1-byte resource, +2. **Idle latency** – `IdleProbes` sequential GETs of the small resource + (1 byte in the draft; up to 10 accepted for Cloudflare compatibility), each on a **fresh connection**, so a sample includes DNS + TCP + TLS + HTTP. Per-stage medians are reported via `net/http/httptrace`. 3. **Download under load**, then **upload under load** (sequentially, so the @@ -222,6 +223,7 @@ compromise) and transparent TCP-level proxies that pass TLS through untouched | `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 | +| Small response body | 1–10 bytes | reject empty/oversized responses; read at most 11 bytes to detect overflow; discard invalid samples and warn | | `ctx` cancellation | – | all flows stop within ~200 ms; partial result, `cancelled=true` | The combined phase budget is `ConfigTimeout + IdleTimeout + N × MaxDuration`, @@ -245,6 +247,14 @@ 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. +Probe cost is estimated, including failed loaded attempts: 5000 bytes for a +foreign probe and 1000 for a self probe. Idle and discovery are outside the +per-direction byte totals. Body-read limits do not prevent transport/socket +read-ahead, and headers, TLS, in-flight work, and cancellation can exceed the +accounted budget. `MaxBytes` is not an exact wire-byte limit. Invalid-size +warnings are limited to one per phase/probe kind; load continues with only +valid latency samples. + ## Deviations from the draft | Item | Draft | Here | Why | @@ -260,6 +270,7 @@ server's flag of the same name limits quiet connections between requests. | Capacity change | – | a > 25 % goodput drop restarts stability tracking | The draft averages across the change. | | Responsiveness window | last MAD intervals | every sample since throughput became stable (`loaded_window`) | Foreign probes are sparse (a TLS handshake each); a fixed 4-tick window could hold self samples and no foreign ones, which read as "no fresh connection ever succeeded". Stability is still judged on the draft's window. | | Probe byte accounting | – | foreign 5000 B, self 1000 B (draft's estimates) | Counted against `MaxBytes` and the 5 % capacity rule. | +| Small response size | 1 byte | complete nonempty bodies up to 10 bytes accepted | Preserve Cloudflare's advertised ten-byte probe; reject empty and larger bodies before they become latency samples. | | Config `version` | must be `1` | `1` or `"1"` accepted | Lenient on the wire, strict on everything else (duplicates, hosts, scheme). | | 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. | diff --git a/cmd/nq/probe_size_test.go b/cmd/nq/probe_size_test.go new file mode 100644 index 0000000..f568d58 --- /dev/null +++ b/cmd/nq/probe_size_test.go @@ -0,0 +1,90 @@ +package main + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/korya/netquality" + "github.com/korya/netquality/server" +) + +func TestInvalidProbeOutput(t *testing.T) { + for _, mode := range []string{"json", "human", "events"} { + t.Run(mode, func(t *testing.T) { + 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.SmallPath && small.Add(1) > 1 { + _, _ = io.WriteString(w, "01234567890") + return + } + h.ServeHTTP(w, r) + })) + srv.EnableHTTP2 = true + srv.StartTLS() + defer srv.Close() + args := base(srv.URL+server.ConfigPath, "--download-only", "--idle-probes", "3", "--max-bytes", "1MB") + if mode != "human" { + args = append(args, "--json") + } + if mode == "events" { + args = append(args, "--events") + } + var out, errb bytes.Buffer + if code := run(args, &out, &errb); code != exitOK { + t.Fatalf("invalid probe must not fail load: exit=%d stderr=%s", code, errb.String()) + } + const diagnostic = "invalid probe response size" + if mode == "human" { + if !strings.Contains(out.String(), "Warning idle latency: "+diagnostic) || !strings.Contains(lineWith(out.String(), "Idle"), "(1 probes)") { + t.Errorf("partial idle/warning missing:\n%s", out.String()) + } + return + } + var res netquality.Result + if err := json.Unmarshal(out.Bytes(), &res); err != nil { + t.Fatal(err) + } + if res.Idle == nil || res.Idle.Samples != 1 || res.Download == nil || res.Download.Bytes == 0 || res.Cancelled { + t.Fatalf("partial idle and download missing: %+v", res) + } + var warnings int + for _, w := range res.Warnings { + if strings.HasPrefix(w, "idle latency: "+diagnostic) { + warnings++ + } + } + if warnings != 1 { + t.Errorf("idle warning missing/repeated: %v", res.Warnings) + } + if mode == "json" { + if errb.Len() != 0 { + t.Errorf("JSON stderr must be quiet: %s", errb.String()) + } + return + } + var samples, warningEvents int + for _, line := range strings.Split(strings.TrimSpace(errb.String()), "\n") { + var e netquality.Event + if err := json.Unmarshal([]byte(line), &e); err != nil { + t.Fatalf("bad event %q: %v", line, err) + } + if e.Kind == netquality.EventProbe { + samples++ + } + if e.Kind == netquality.EventWarning && strings.HasPrefix(e.Message, "idle latency: "+diagnostic) { + warningEvents++ + } + } + if samples != 1 || warningEvents != 1 { + t.Errorf("invalid successes or repeated warnings: samples=%d warnings=%d", samples, warningEvents) + } + }) + } +} diff --git a/docs/architecture.md b/docs/architecture.md index b77aa5e..9d4a660 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -49,7 +49,7 @@ real protocol in-process over TLS + HTTP/2 without a network. | 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 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`. | +| Byte accounting is client-side | Upload bytes are counted when handed to the transport. Loaded probes charge fixed estimates even on failure; idle/discovery are excluded. Probe body reads are bounded independently, but HTTP/TLS overhead, transport buffering and cancellation can consume traffic beyond `MaxBytes`; it is not a wire-byte meter. | | 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). | ## Technology choices diff --git a/docs/product-specs/latency.md b/docs/product-specs/latency.md index 3f70d4f..17e5906 100644 --- a/docs/product-specs/latency.md +++ b/docs/product-specs/latency.md @@ -74,3 +74,14 @@ mean to zero. Windows therefore reads `QueryPerformanceCounter` instead. Where no such clock can be reached the run falls back to `time.Now` for every sample — never mixing clocks within one sample — reports its numbers, and records a warning saying they are quantised (RES-6, INV-3). + +### LAT-11: Small response validation +Idle, foreign, and self probes accept only complete, nonempty response bodies +of 1–10 bytes. The draft's one-byte response and Cloudflare's ten-byte +response both fit this compatibility ceiling. A declared length above ten +is rejected before reading payload; otherwise at most eleven body bytes are +consumed to detect overflow. Receipt of ten bytes without response completion +does not establish success: the existing phase/caller deadline still applies. +Empty, oversized, and incomplete responses produce no latency sample or +successful-probe event. Rejection closes the response body, preserving other +streams on a shared HTTP/2 load connection. Invalid-size warnings follow RES-6. diff --git a/docs/product-specs/limits.md b/docs/product-specs/limits.md index 52789fa..19948cc 100644 --- a/docs/product-specs/limits.md +++ b/docs/product-specs/limits.md @@ -18,6 +18,10 @@ any speed. A caller on a metered link sets `MaxBytes` (> 0), which then bounds bytes moved per direction, counting load payload plus a fixed estimate per probe (5000 B foreign, 1000 B self); reaching it ends the phase with `reason=bytes_cap` and a warning. `MaxBytes` ≤ 0 means unlimited. +Failed loaded probe attempts retain their fixed charge. Idle and discovery +traffic are outside these per-direction totals. The accounting is an estimate, +not a wire-byte meter: headers, TLS, transport read-ahead, in-flight requests, +and cancellation can consume more traffic than the reported budget. ### LIM-3: Flow cap `MaxFlows` (default 16) is never exceeded in any direction. @@ -71,3 +75,11 @@ 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). + +### LIM-10: Small response consumption +The small-response ceiling (LAT-11) applies independently of the optional +byte budget. At most eleven bytes of each probe response body are read by +the client, including the byte used to detect overflow; a declared oversize +body is rejected without reading payload. Transports and sockets may buffer +additional data before rejection. This consumption bound neither changes the +fixed probe estimates nor promises an exact limit on transferred wire bytes. diff --git a/docs/product-specs/result.md b/docs/product-specs/result.md index e875377..5583524 100644 --- a/docs/product-specs/result.md +++ b/docs/product-specs/result.md @@ -44,6 +44,12 @@ RPM is round trips per minute; bytes are bytes. ### RES-6: Warnings Every cap hit, fallback, proxy finding, flow error, and idle failure appends a human-readable string to `warnings`, and is logged at warning level. +Invalid probe sizes produce at most one warning per phase and probe kind: +one for idle, and one each for foreign/self in each selected load direction. +Idle retains valid samples even when another probe has an invalid size, and +combines the size diagnostic with any idle timeout or all-failed warning. +Caller cancellation retains its precedence. Loaded size failures discard +samples without aborting load. Warning messages contain no response body. ### RES-7: Events `RunWithEvents` delivers `phase` events for discover/idle/download/upload/done, diff --git a/docs/test-matrix.md b/docs/test-matrix.md index 91ca533..0bd254f 100644 --- a/docs/test-matrix.md +++ b/docs/test-matrix.md @@ -90,6 +90,11 @@ directory relative to the repo root; `.` is the library. | Feature | Scenario | Package | Test | |---|---|---|---| +| Probe body cap | HTTP/1.1 and HTTP/2: complete 1–10 bytes accepted, empty/oversized rejected, application reads bounded to 11 bytes, truncation and deadlines preserved (LAT-11, LIM-10) | . | TestProbeResponseSize | +| Probe isolation | Rejected or timed-out self probe shares an active HTTP/2 connection; load continues on that connection (LAT-11, INV-4) | . | TestProbeRejectionPreservesLoad | +| Rejected probe cost | Known and unknown oversized probes still hit MaxBytes with their fixed estimates, producing no goodput or latency (LIM-2, LIM-10) | . | TestRejectedProbesHitByteCap | +| Invalid probes | Partial idle retained; both loaded series reject invalid samples, keep estimated charges out of goodput, dedupe warning events per direction/kind and reset between runs (LAT-11, LIM-2, RES-6) | . | TestInvalidProbeResponses | +| Invalid idle | All-invalid, mixed failures, timeout and parent cancellation retain samples and diagnostic precedence with one idle warning (LAT-11, LIM-8, RES-6) | . | TestInvalidIdleDiagnostics | | MaxBytes | Download truncated with `bytes_cap`, warning, mean throughput fallback | . | TestBytesCap | | MaxBytes | Upload truncated with `bytes_cap` | . | TestUploadBytesCap | | MaxBytes | Counter: limit 0 never trips, positive limit trips exactly once | . | TestByteCounterLimits | @@ -129,6 +134,7 @@ directory relative to the repo root; `.` is the library. | Feature | Scenario | Package | Test | |---|---|---|---| | `--json` | stdout is a `Result`, stderr quiet | cmd/nq | TestJSONOutput | +| Invalid probe output | JSON, human and event output retain partial idle statistics and a single size warning while load succeeds (LAT-11, RES-6) | cmd/nq | TestInvalidProbeOutput | | `--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 | diff --git a/options.go b/options.go index 3303655..0123a20 100644 --- a/options.go +++ b/options.go @@ -71,6 +71,8 @@ type Options struct { // phase, probes included; hitting it truncates the phase with // reason=bytes_cap. 0 (the default) means no byte cap: MaxDuration // bounds each load phase. Set it on metered links. + // Probe attempts use fixed cost estimates; idle/discovery are excluded. + // Transport buffering and cancellation can exceed this accounting budget. MaxBytes int64 // MaxFlows caps the number of concurrent load-generating connections // (default 16, draft MNP). diff --git a/probe.go b/probe.go index dd9df67..92893f0 100644 --- a/probe.go +++ b/probe.go @@ -3,6 +3,8 @@ package netquality import ( "context" "crypto/tls" + "errors" + "fmt" "io" "net/http" "net/http/httptrace" @@ -24,8 +26,12 @@ type probeTimes struct { const ( foreignProbeBytes = 5000 selfProbeBytes = 1000 + // The draft uses one byte; Cloudflare's supported endpoint returns ten. + maxProbeBodyBytes = 10 ) +var errInvalidProbeSize = errors.New("invalid probe response size") + // foreignProbe performs a GET of the small URL on a brand-new connection and // records per-stage timings. rt must not reuse connections. // observe, if non-nil, receives the TLS state of every successful handshake. @@ -100,7 +106,8 @@ func foreignProbe(ctx context.Context, rt http.RoundTripper, url string, extra h return s, nil } -// doProbe issues the GET and drains the 1-byte body. +// doProbe requires a complete, nonempty response within the compatibility +// ceiling. The extra byte detects overflow without draining an arbitrary body. func doProbe(ctx context.Context, rt http.RoundTripper, url string, extra http.Header) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { @@ -115,8 +122,20 @@ func doProbe(ctx context.Context, rt http.RoundTripper, url string, extra http.H if err := checkStatus(resp); err != nil { return err } - _, err = io.Copy(io.Discard, resp.Body) - return err + if resp.ContentLength > maxProbeBodyBytes { + return fmt.Errorf("%w: declared %d bytes exceeds %d-byte limit", errInvalidProbeSize, resp.ContentLength, maxProbeBodyBytes) + } + n, err := io.Copy(io.Discard, io.LimitReader(resp.Body, maxProbeBodyBytes+1)) + if n > maxProbeBodyBytes { + return fmt.Errorf("%w: body exceeds %d-byte limit", errInvalidProbeSize, maxProbeBodyBytes) + } + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("%w: empty body (expected 1-%d bytes)", errInvalidProbeSize, maxProbeBodyBytes) + } + return nil } // selfProbe performs a GET of the small URL on an existing (load) transport. diff --git a/probe_size_test.go b/probe_size_test.go new file mode 100644 index 0000000..7f99a8e --- /dev/null +++ b/probe_size_test.go @@ -0,0 +1,434 @@ +package netquality + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/korya/netquality/server" +) + +// Instrument the real transport's response body, including unsuccessful reads. +type probeReadTracker struct { + http.RoundTripper + read atomic.Int64 + closed atomic.Bool +} + +type trackedProbeBody struct { + io.ReadCloser + tracker *probeReadTracker +} + +func (rt *probeReadTracker) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := rt.RoundTripper.RoundTrip(req) + if err == nil { + resp.Body = &trackedProbeBody{ReadCloser: resp.Body, tracker: rt} + } + return resp, err +} + +func (b *trackedProbeBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + b.tracker.read.Add(int64(n)) + return n, err +} + +func (b *trackedProbeBody) Close() error { + err := b.ReadCloser.Close() + b.tracker.closed.Store(true) + return err +} + +func TestProbeResponseSize(t *testing.T) { + for _, h2 := range []bool{false, true} { + for _, tc := range []struct { + name string + declared string + body string + stall bool + status int + wantRead int64 + wantErr string + }{ + {"one", "1", "x", false, 200, 1, ""}, + {"two", "2", "xx", false, 200, 2, ""}, + {"nine", "9", "123456789", false, 200, 9, ""}, + {"ten", "10", "0123456789", false, 200, 10, ""}, + {"unknown-one", "", "x", false, 200, 1, ""}, + {"unknown-ten", "", "0123456789", false, 200, 10, ""}, + {"empty", "0", "", false, 200, 0, "size"}, + {"unknown-empty", "", "", false, 200, 0, "size"}, + {"declared-eleven", "11", "", true, 200, 0, "size"}, + {"declared-megabyte", "1048576", "", true, 200, 0, "size"}, + {"unknown-eleven", "", "01234567890", true, 200, 11, "size"}, + {"unknown-megabyte", "", strings.Repeat("x", 1<<20), false, 200, 11, "size"}, + {"ten-stall", "", "0123456789", true, 200, 10, "deadline"}, + {"body-stall", "1", "", true, 200, 0, "deadline"}, + {"truncated", "5", "x", false, 200, 1, "read"}, + {"status-precedence", "1048576", "", true, 403, 0, "status"}, + } { + t.Run(fmt.Sprintf("h2=%v/%s", h2, tc.name), func(t *testing.T) { + exited := make(chan struct{}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(exited) + if tc.declared != "" { + w.Header().Set("Content-Length", tc.declared) + } + w.WriteHeader(tc.status) + w.(http.Flusher).Flush() // prevent automatic Content-Length for unknown cases + _, _ = io.WriteString(w, tc.body) + if tc.stall { + w.(http.Flusher).Flush() + <-r.Context().Done() + } + })) + srv.EnableHTTP2 = h2 + srv.StartTLS() + defer srv.Close() + client := srv.Client() + defer client.CloseIdleConnections() + rt := &probeReadTracker{RoundTripper: client.Transport} + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + sample, err := foreignProbe(ctx, rt, srv.URL, nil, monoNow, nil) + if (err == nil) != (tc.wantErr == "") { + t.Fatalf("sample=%+v err=%v; want %q", sample, err, tc.wantErr) + } + if err != nil && sample != (LatencySample{}) { + t.Errorf("failed probe produced a sample: %+v", sample) + } + switch tc.wantErr { + case "size": + if !errors.Is(err, errInvalidProbeSize) { + t.Errorf("want size error: %v", err) + } + case "deadline": + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("incomplete body must wait for deadline: %v", err) + } + case "status": + if errors.Is(err, errInvalidProbeSize) || !strings.Contains(err.Error(), "403") { + t.Errorf("status must win: %v", err) + } + case "read": + if errors.Is(err, errInvalidProbeSize) || !errors.Is(err, io.ErrUnexpectedEOF) { + t.Errorf("truncated body must retain read error: %v", err) + } + } + if rt.read.Load() != tc.wantRead || !rt.closed.Load() { + t.Errorf("read=%d want=%d closed=%v", rt.read.Load(), tc.wantRead, rt.closed.Load()) + } + select { + case <-exited: + case <-time.After(time.Second): + t.Fatal("response close did not release handler") + } + }) + } + } +} + +func TestProbeRejectionPreservesLoad(t *testing.T) { + for _, mode := range []string{"declared", "streamed", "deadline"} { + t.Run(mode, func(t *testing.T) { + more := make(chan struct{}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/load" { + _, _ = io.WriteString(w, "LOAD") + w.(http.Flusher).Flush() + select { + case <-more: + case <-r.Context().Done(): + return + } + _, _ = io.WriteString(w, "MORE") + w.(http.Flusher).Flush() + } else { + if mode == "declared" { + w.Header().Set("Content-Length", "1048576") + } + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + if mode == "streamed" { + _, _ = io.WriteString(w, "01234567890") + w.(http.Flusher).Flush() + } + } + <-r.Context().Done() + })) + srv.EnableHTTP2 = true + srv.StartTLS() + defer srv.Close() + client := srv.Client() + defer client.CloseIdleConnections() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + var mu sync.Mutex + var connections []httptrace.GotConnInfo + ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{GotConn: func(i httptrace.GotConnInfo) { + mu.Lock() + defer mu.Unlock() + connections = append(connections, i) + }}) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/load", nil) + if err != nil { + t.Fatal(err) + } + load, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + defer load.Body.Close() + buf := make([]byte, 4) + if _, err := io.ReadFull(load.Body, buf); err != nil { + t.Fatal(err) + } + probeCtx, stopProbe := context.WithTimeout(ctx, 500*time.Millisecond) + defer stopProbe() + sample, err := selfProbe(probeCtx, client.Transport, srv.URL+"/small", nil, monoNow) + want := errInvalidProbeSize + if mode == "deadline" { + want = context.DeadlineExceeded + } + if !errors.Is(err, want) || sample != (LatencySample{}) { + t.Fatalf("sample=%+v error=%v; want %v", sample, err, want) + } + mu.Lock() + shared := len(connections) == 2 && connections[0].Conn == connections[1].Conn && connections[1].Reused + mu.Unlock() + if !shared || load.ProtoMajor != 2 { + t.Fatal("probe did not share the active HTTP/2 load connection") + } + close(more) + if _, err := io.ReadFull(load.Body, buf); err != nil || string(buf) != "MORE" { + t.Fatalf("load broken after rejected probe: %q %v", buf, err) + } + }) + } +} + +func TestInvalidProbeResponses(t *testing.T) { + for _, body := range []string{"", "01234567890"} { + t.Run(fmt.Sprintf("bytes=%d", len(body)), func(t *testing.T) { + var idleCalls 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 { + if !loading.Load() && idleCalls.Add(1) != 2 { + _, _ = io.WriteString(w, "x") + } else { + _, _ = io.WriteString(w, body) + } + return + } + if r.URL.Path == server.LargePath { + // No load payload: every accounted byte must be a probe estimate. + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + return + } + h.ServeHTTP(w, r) + }) + }, nil, true) + for range 2 { // warning guards must reset between runs as well as directions + idleCalls.Store(0) + loading.Store(false) + var mu sync.Mutex + var events []Event + res, err := RunWithEvents(context.Background(), Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: 3, MaxFlows: 1, + MaxDuration: 500 * time.Millisecond, MaxBytes: 1 << 40, + }, func(e Event) { + if e.Kind == EventPhase && e.Phase == "download" { + loading.Store(true) + } + mu.Lock() + defer mu.Unlock() + events = append(events, e) + }) + if err != nil || res.Idle == nil || res.Idle.Samples != 2 || res.Download == nil || res.Upload == nil { + t.Fatalf("partial idle and both load results must survive: %+v err=%v", res, err) + } + for _, d := range []*DirectionResult{res.Download, res.Upload} { + if d.Loaded.Foreign != nil || d.Loaded.Self != nil || d.RPM != 0 || d.FlowErrors != 0 || d.Reason != ReasonDurationCap { + t.Errorf("invalid probes polluted load result: %+v", d) + } + } + if d := res.Download; d.Bytes < foreignProbeBytes+selfProbeBytes || d.Bytes%1000 != 0 || d.ThroughputBPS != 0 { + t.Errorf("rejected probes must cost estimates, never goodput: %+v", d) + } + wantWarnings := []string{"idle latency:", "download foreign probe:", "download self probe:", "upload foreign probe:", "upload self probe:"} + for _, prefix := range wantWarnings { + var results, emitted int + for _, w := range res.Warnings { + if strings.HasPrefix(w, prefix) && strings.Contains(w, errInvalidProbeSize.Error()) { + results++ + } + } + mu.Lock() + for _, e := range events { + if e.Kind == EventWarning && strings.HasPrefix(e.Message, prefix) && strings.Contains(e.Message, errInvalidProbeSize.Error()) { + emitted++ + } + } + mu.Unlock() + if results != 1 || emitted != 1 { + t.Errorf("%s warnings: result=%d events=%d; %v", prefix, results, emitted, res.Warnings) + } + } + mu.Lock() + var samples int + for _, e := range events { + if e.Kind == EventProbe { + samples++ + if e.ProbeKind != "idle" { + t.Errorf("invalid loaded probe emitted success: %+v", e) + } + } + } + mu.Unlock() + if samples != 2 { + t.Errorf("successful probe events=%d; want 2", samples) + } + } + }) + } +} + +func TestInvalidIdleDiagnostics(t *testing.T) { + for _, mode := range []string{"all-invalid", "last-status", "timeout", "cancel"} { + t.Run(mode, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + 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() { + h.ServeHTTP(w, r) + return + } + n := attempts.Add(1) + switch { + case mode == "all-invalid" || n == 2: + w.Header().Set("Content-Length", "11") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + case n == 1: + h.ServeHTTP(w, r) + case n == 3: + w.WriteHeader(http.StatusForbidden) + default: + if mode == "cancel" { + cancel() + } + <-r.Context().Done() + } + }) + }, nil, true) + probes := 4 + if mode == "last-status" { + probes = 3 + } + 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, + }, func(e Event) { + if e.Kind == EventPhase && e.Phase == "download" { + loading.Store(true) + } + }) + if res == nil { + t.Fatalf("missing partial result: %v", err) + } + if mode == "cancel" { + if !errors.Is(err, context.Canceled) || !res.Cancelled || res.Download != nil || hasWarning(res, "idle latency:") { + t.Fatalf("parent cancellation must take precedence: %+v err=%v", res, err) + } + } else { + if err != nil || res.Download == nil || res.Cancelled { + t.Fatalf("idle rejection must allow load: %+v err=%v", res, err) + } + var warnings []string + for _, w := range res.Warnings { + if strings.HasPrefix(w, "idle latency:") { + warnings = append(warnings, w) + } + } + if len(warnings) != 1 || !strings.Contains(warnings[0], errInvalidProbeSize.Error()) { + t.Fatalf("expected one size warning: %v", res.Warnings) + } + if mode != "all-invalid" && !strings.Contains(warnings[0], "403") { + t.Errorf("last failure lost: %v", warnings) + } + if strings.Contains(warnings[0], "idle timeout") != (mode == "timeout") { + t.Errorf("timeout diagnosis: %v", warnings) + } + } + if mode == "all-invalid" { + if res.Idle != nil { + t.Errorf("invalid responses produced idle statistics: %+v", res.Idle) + } + } else if res.Idle == nil || res.Idle.Samples != 1 { + t.Errorf("valid sample lost: %+v", res.Idle) + } + }) + } +} + +func TestRejectedProbesHitByteCap(t *testing.T) { + for _, known := range []bool{false, true} { + t.Run(fmt.Sprintf("known=%v", known), func(t *testing.T) { + 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.SmallPath: + if known { + w.Header().Set("Content-Length", "1048576") + } + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + if !known { + _, _ = io.WriteString(w, "01234567890") + w.(http.Flusher).Flush() + } + <-r.Context().Done() + case server.LargePath: + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + default: + h.ServeHTTP(w, r) + } + }) + }, nil, true) + res, err := Run(context.Background(), Target{ConfigURL: srv.URL + server.ConfigPath}, Options{ + HTTPClient: insecureClient(), IdleProbes: -1, Directions: Download, + MaxBytes: 10_000, MaxDuration: 2 * time.Second, MaxFlows: 1, + }) + if err != nil || res == nil || res.Download == nil { + t.Fatalf("result=%+v err=%v", res, err) + } + d := res.Download + if d.Reason != ReasonBytesCap || d.Bytes < 10_000 || d.Bytes > 15_000 || d.Bytes%1000 != 0 { + t.Errorf("failed estimates must trip cap: %+v", d) + } + if d.ThroughputBPS != 0 || d.RPM != 0 || d.Loaded.Foreign != nil || d.Loaded.Self != nil { + t.Errorf("rejected probes became measurements: %+v", d) + } + }) + } +} diff --git a/run.go b/run.go index 76c3faa..9024f1b 100644 --- a/run.go +++ b/run.go @@ -238,6 +238,7 @@ func (r *runner) idle(ctx context.Context) (*LatencyStats, error) { defer closeIdle(rt) var samples []LatencySample var lastErr error + var invalidSize error for i := 0; i < r.opts.IdleProbes; i++ { if ctx.Err() != nil { break @@ -248,6 +249,9 @@ func (r *runner) idle(ctx context.Context) (*LatencyStats, error) { break // retain the preceding probe failure as timeout context } lastErr = err + if invalidSize == nil && errors.Is(err, errInvalidProbeSize) { + invalidSize = err + } continue } samples = append(samples, s) @@ -263,8 +267,17 @@ func (r *runner) idle(ctx context.Context) (*LatencyStats, error) { if lastErr != nil { err = fmt.Errorf("%w; last probe error: %v", err, lastErr) } + if invalidSize != nil && !errors.Is(lastErr, errInvalidProbeSize) { + err = fmt.Errorf("%w; %v", err, invalidSize) + } return st, err } + if invalidSize != nil { + if lastErr != nil && !errors.Is(lastErr, errInvalidProbeSize) { + return st, fmt.Errorf("%w; last probe error: %v", invalidSize, lastErr) + } + return st, invalidSize + } if st == nil { if lastErr == nil { lastErr = errors.New("no samples") @@ -581,6 +594,7 @@ func (r *runner) probeLoop(ctx context.Context, p *phaseState) { sem := make(chan struct{}, maxInFlight) var wg sync.WaitGroup defer wg.Wait() + var foreignWarning, selfWarning sync.Once launch := func(self bool) { select { @@ -613,6 +627,13 @@ func (r *runner) probeLoop(ctx context.Context, p *phaseState) { if err != nil { if ctx.Err() == nil { r.opts.Logger.Debug("probe failed", "kind", kind, "err", err) + if errors.Is(err, errInvalidProbeSize) { + warning := &foreignWarning + if self { + warning = &selfWarning + } + warning.Do(func() { r.warn("%s %s probe: %v; invalid samples discarded", p.dir, kind, err) }) + } } return } From c85e3ce3b8b43a4df8d14e6bc4e4ef1e9f0aa0fb Mon Sep 17 00:00:00 2001 From: korya <148461+korya@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:57:07 -0400 Subject: [PATCH 2/3] docs(probes): Clarify the fixed ceiling and record vendor checks State that the ten-byte response ceiling has no caller override and that load measurements continue when probes fail. Keep the approved runtime policy. Separate the compatibility change from the bounded-read fix in release notes. Record dated config and small-response observations for Apple and Cloudflare, with commands to repeat the checks without generating load traffic. Co-Authored-By: GPT-6 --- CHANGELOG.md | 16 +++++++++------ README.md | 10 +++++++++- testdata/config/README.md | 41 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 7 deletions(-) create mode 100644 testdata/config/README.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ac4771..28ab6d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,13 +13,17 @@ All notable changes to this project are documented here. The format follows large `IdleProbes` values; raise `IdleTimeout` to collect more samples and reach higher percentile thresholds. +### Changed +- Small probe responses must now be complete, nonempty bodies of 1–10 bytes. + Previously accepted empty or larger custom-server responses are invalid. + The ten-byte ceiling is fixed, with no caller override; it preserves the + observed Apple and Cloudflare responses (#40). + ### Fixed -- Small probe responses are limited to complete, nonempty bodies of 1–10 - bytes, preserving Apple and Cloudflare compatibility. Declared oversized - bodies are rejected before reading, and streamed bodies after at most 11 - bytes. Invalid sizes no longer become latency samples; bounded warnings - explain rejection while valid samples and load measurements survive (#40). - Previously accepted empty or larger custom-server responses are now invalid. +- Declared oversized probe bodies are rejected before reading, and streamed + bodies after at most 11 bytes. Invalid sizes no longer become latency + samples; bounded warnings explain rejection while valid samples and load + measurements survive (#40). Probe cost remains estimated; transport buffering can exceed body-read limits. - Idle-timeout warnings retain the preceding probe error and are omitted when all requested idle samples have already completed (#39). diff --git a/README.md b/README.md index 2c6b16a..f345321 100644 --- a/README.md +++ b/README.md @@ -255,6 +255,14 @@ accounted budget. `MaxBytes` is not an exact wire-byte limit. Invalid-size warnings are limited to one per phase/probe kind; load continues with only valid latency samples. +The ten-byte response ceiling is fixed; there is no caller override. A server +that changes its small response above ten bytes becomes incompatible with +latency probing. Load continues to collect capacity measurements within the +configured budgets, even if every probe fails; loaded latency is then absent +and RPM is zero. Set `MaxBytes` and `MaxDuration` to bound that cost. +[Dated compatibility checks](testdata/config/README.md) record the observed +Apple and Cloudflare response sizes and how to repeat the small GETs. + ## Deviations from the draft | Item | Draft | Here | Why | @@ -270,7 +278,7 @@ valid latency samples. | Capacity change | – | a > 25 % goodput drop restarts stability tracking | The draft averages across the change. | | Responsiveness window | last MAD intervals | every sample since throughput became stable (`loaded_window`) | Foreign probes are sparse (a TLS handshake each); a fixed 4-tick window could hold self samples and no foreign ones, which read as "no fresh connection ever succeeded". Stability is still judged on the draft's window. | | Probe byte accounting | – | foreign 5000 B, self 1000 B (draft's estimates) | Counted against `MaxBytes` and the 5 % capacity rule. | -| Small response size | 1 byte | complete nonempty bodies up to 10 bytes accepted | Preserve Cloudflare's advertised ten-byte probe; reject empty and larger bodies before they become latency samples. | +| Small response size | 1 byte | complete nonempty bodies up to 10 bytes accepted; fixed ceiling, no override | Preserve Cloudflare's advertised ten-byte probe; reject empty and larger bodies before they become latency samples. | | Config `version` | must be `1` | `1` or `"1"` accepted | Lenient on the wire, strict on everything else (duplicates, hosts, scheme). | | 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. | diff --git a/testdata/config/README.md b/testdata/config/README.md new file mode 100644 index 0000000..abe31cd --- /dev/null +++ b/testdata/config/README.md @@ -0,0 +1,41 @@ +# Public target compatibility evidence + +The JSON files in this directory exercise config parsing; a URL in a config +fixture does not establish the size of the response it serves. + +On **2026-09-10**, fresh config fetches named the small URLs below. GETs with +`Accept-Encoding: identity` returned HTTP/2 200, declared the listed +`Content-Length`, and completed with the same number of body bytes: + +| Target | Config | Small response URL | Declared / received bytes | +|---|---|---|---| +| Apple | [Config](https://mensura.cdn-apple.com/api/v1/gm/config) | [Small](https://mensura.cdn-apple.com/api/v1/gm/small) | 1 / 1 | +| Cloudflare | [Config](https://aim.cloudflare.com/responsiveness/api/v1/config) | [Small](https://h3.speed.cloudflare.com/__down?bytes=10) | 10 / 10 | + +These are dated observations, not vendor guarantees. They support the fixed +ten-byte compatibility ceiling in LAT-11. Apple's observed one-byte body +agrees with the draft and reference server; Cloudflare's observed ten-byte +body requires the documented deviation. A future config or body change may +require revisiting compatibility. + +To repeat just the response checks without starting any load flows: + +```sh +curl --fail --silent --show-error --max-time 10 --max-filesize 32 \ + --header 'Accept-Encoding: identity' --output /dev/null \ + --write-out 'status=%{http_code} bytes=%{size_download}\n' \ + 'https://mensura.cdn-apple.com/api/v1/gm/small' +curl --fail --silent --show-error --max-time 10 --max-filesize 32 \ + --header 'Accept-Encoding: identity' --output /dev/null \ + --write-out 'status=%{http_code} bytes=%{size_download}\n' \ + 'https://h3.speed.cloudflare.com/__down?bytes=10' +``` + +In native Windows shells, use `curl.exe`, put each command on one line, and +use `NUL` in place of `/dev/null`. Fetch each target's current config first +if its advertised URL may have changed. The curl limits bound +response consumption and waiting, not HTTP/TLS or socket buffering. + +`TestProbeResponseSize` exercises one- and ten-byte bodies offline over +HTTP/1.1 and HTTP/2. `TestLive` exercises both full public targets in the +opt-in/nightly suite. Full live measurements were not run for this check. From 697a9b2c05eda3fd6c600931e21101cb86814b56 Mon Sep 17 00:00:00 2001 From: korya <148461+korya@users.noreply.github.com> Date: Thu, 10 Sep 2026 08:23:07 -0400 Subject: [PATCH 3/3] test(cli): Decouple byte-cap reporting from loopback throughput The truncation-output test required a 1 MB transfer within a 300 ms deadline. On a contended Windows race runner, the duration cap won and the correct CLI output failed the byte-cap assertion. Use a one-byte accounting cap with a five-second watchdog so the assertion checks truncation reporting without a minimum transfer-rate requirement. Keep the exact byte-cap reason and warning checks. Co-Authored-By: GPT-6 --- cmd/nq/e2e_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmd/nq/e2e_test.go b/cmd/nq/e2e_test.go index c8480f5..1857c82 100644 --- a/cmd/nq/e2e_test.go +++ b/cmd/nq/e2e_test.go @@ -104,7 +104,9 @@ func TestCancelledExitCode(t *testing.T) { func TestTruncatedStillExitsZero(t *testing.T) { var out, errb bytes.Buffer - if c := run(base(startServer(t), "--download-only", "--max-bytes", "1MB"), &out, &errb); c != exitOK { + // Exercise byte-cap reporting without requiring a minimum loopback rate. + // One accounted byte trips the cap; duration is only a generous watchdog. + if c := run(base(startServer(t), "--download-only", "--max-bytes", "1", "--max-duration", "5s"), &out, &errb); c != exitOK { t.Fatalf("exit %d: %s", c, errb.String()) } if !strings.Contains(out.String(), "TRUNCATED: bytes_cap") || !strings.Contains(out.String(), "Warning download: byte cap") {