From 720f1cd62d945e2054291c791a654076847f9358 Mon Sep 17 00:00:00 2001 From: Alex Silva Medeiros Date: Mon, 27 Jul 2026 13:20:58 -0300 Subject: [PATCH] feat: add adaptive pacing and resolver selection --- README.md | 28 +- cmd/dnsbench/probe.go | 73 ++++- cmd/dnsbench/run.go | 8 +- cmd/dnsbench/run_test.go | 25 ++ cmd/dnsbench/selection_test.go | 241 ++++++++++++++++ docs/ARCHITECTURE.md | 10 +- docs/METHODOLOGY.md | 13 +- internal/bench/engine.go | 72 ++++- internal/bench/engine_test.go | 112 +++++++- internal/bench/pacer.go | 302 ++++++++++++++++---- internal/bench/pacer_test.go | 393 +++++++++++++++++++++++++-- internal/bench/schedule.go | 41 +-- internal/bench/triage.go | 23 +- internal/model/config_test.go | 7 + internal/model/model.go | 43 ++- internal/report/report_test.go | 39 +++ internal/serverlist/registry.go | 29 ++ internal/serverlist/registry_test.go | 22 ++ internal/ui/live_test.go | 2 +- 19 files changed, 1340 insertions(+), 143 deletions(-) diff --git a/README.md b/README.md index eaae082..3b80b2c 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,11 @@ it onto your `PATH`, drop the `./` prefix; on Windows use `.\dnsbench.exe`. # Benchmark two specific servers, show the detailed table and export reports ./dnsbench run --only 1.1.1.1,9.9.9.9 --details --export json,txt --out reports +# Benchmark one-off encrypted endpoints without saving them to the user list +./dnsbench run \ + --server https://dns.example/dns-query@192.0.2.1 \ + --server quic://dns.example@192.0.2.1 + # Benchmark and open a shareable HTML report in the browser when the run finishes ./dnsbench run --builtin --open @@ -145,8 +150,29 @@ dnsbench speaks six transports: plaintext `udp` and `tcp` on port 53, `dot` (DNS # Compare just the QUIC-based transports ./dnsbench run --protocols doh3,doq + +# Compare those built-in transports with the DNS currently used by the system +./dnsbench run --system --builtin --protocols doh3,doq ``` +`--protocols` normally filters every selected resolver, which excludes the +system DNS because operating systems expose it to dnsbench as UDP/53. Explicitly +selecting `--system` keeps the discovered system resolver(s) in the run +regardless of that protocol filter. Because source flags define which sources +participate, add `--builtin`, `--user` or one-off `--server` endpoints for the +encrypted side of the comparison. `--no-ipv6` and `--only` still apply. + +Use repeatable `--server` flags for endpoints that should participate in only +the current run. The syntax is the same as a text server list: a bare IP address +for UDP, `tls://hostname@address` for DoT, an HTTPS URL for DoH, +`h3://hostname/path@address` for DoH/3, or `quic://hostname@address` for DoQ. +An optional `:port` can follow the address. When at least one `--server` is +provided without `--system`, `--builtin` or `--user`, the automatic default that +includes all three sources is disabled, so only the one-off endpoints (and any +explicit `--servers-file`) enter the run. Pass a source flag to combine that +source with the one-off endpoints. These entries are never written to the user +configuration. + This matters if your router intercepts port 53: every plaintext query is then answered by the router rather than by the resolver you meant to measure, and the ranking compares your router against itself. Selecting encrypted transports does not change how a hostname-only custom endpoint is bootstrapped: the system resolver still has to resolve that hostname before the encrypted connection exists. Every encrypted entry in the built-in list avoids that lookup by pinning the resolver's IP address in the `address` field and dialing it directly, while keeping the hostname for SNI and certificate validation. Add an `address` to a custom entry when you want the same behavior. @@ -171,7 +197,7 @@ For a full, shareable report use `--export html`, or `--open`, which writes the ## Measurement pacing -During triage and benchmarking, dnsbench spaces query launches globally (`--pace`, default 20ms) and reuses one connected socket per server, so a run does not flood the local Wi-Fi link, NAT router or uplink and then misread the resulting drops as server loss. `--pace` is where the benchmark starts, not a bound: it shortens on clean networks to as little as a quarter of the configured value and, when timeouts appear across several servers at once (the signature of local congestion rather than of any one server), eases off to as much as 8× it and notes the adjustment in the live view. The preceding characterization phase has no launch pacer; its small qualitative workload reuses one persistent session per server and is bounded by `--concurrency`. See [docs/METHODOLOGY.md](docs/METHODOLOGY.md) for details. +During triage and benchmarking, dnsbench spaces query launches globally and reuses one connected socket per server, so a run does not flood the local Wi-Fi link, NAT router or uplink and then misread the resulting drops as server loss. When `--pace` is omitted, its 20ms default is adaptive but never becomes faster. A burst of launch-correlated timeouts across at least three independent, recently healthy resolver groups signals **possible shared-path congestion** and moves the interval from 20ms to 40ms and then at most 80ms; after the cooldown, clean traffic recovers gradually to 20ms. Endpoints from the same known operator count as one group, as do different protocols on the same address when no operator metadata exists. Passing `--pace DURATION` makes the interval fixed and disables both adaptation and pacing jitter. `--concurrency` remains a separate hard limit on simultaneous attempts and does not set QPS. Adjustment evidence is retained in JSON exports. See [docs/METHODOLOGY.md](docs/METHODOLOGY.md) for details. While a run is in progress dnsbench also asks the operating system not to enter idle sleep, so leaving the machine unattended (or letting the display turn off) does not suspend the process mid-run and poison the timings. It uses the native mechanism of each platform — `caffeinate` on macOS, a systemd-logind idle inhibitor lock held over D-Bus on Linux, and `SetThreadExecutionState` on Windows. diff --git a/cmd/dnsbench/probe.go b/cmd/dnsbench/probe.go index 42f9d2b..a14f779 100644 --- a/cmd/dnsbench/probe.go +++ b/cmd/dnsbench/probe.go @@ -25,6 +25,7 @@ type selectionFlags struct { system bool builtin bool user bool + servers []string only []string serversFile string noIPv6 bool @@ -32,9 +33,10 @@ type selectionFlags struct { } func registerSelectionFlags(cmd *cobra.Command, sel *selectionFlags) { - cmd.Flags().BoolVar(&sel.system, "system", false, "include the system-configured DNS resolvers") + cmd.Flags().BoolVar(&sel.system, "system", false, "include the system-configured DNS resolvers, even when excluded by --protocols") cmd.Flags().BoolVar(&sel.builtin, "builtin", false, "include the built-in public DNS resolvers") cmd.Flags().BoolVar(&sel.user, "user", false, "include the user resolver list") + cmd.Flags().StringArrayVar(&sel.servers, "server", nil, "one-off resolver endpoint (IP, tls://, https://, h3:// or quic://); repeat to add more") cmd.Flags().StringSliceVar(&sel.only, "only", nil, "restrict to resolvers matching these IDs or addresses") cmd.Flags().StringVar(&sel.serversFile, "servers-file", "", "extra resolver list file to include (.json, .csv or .txt)") cmd.Flags().BoolVar(&sel.noIPv6, "no-ipv6", false, "exclude resolvers with IPv6 addresses") @@ -50,13 +52,28 @@ type selectedServers struct { } func selectServers(ctx context.Context, sel selectionFlags) (selectedServers, error) { + return selectServersWithDiscovery(ctx, sel, sysdns.Discover) +} + +func selectServersWithDiscovery( + ctx context.Context, + sel selectionFlags, + discover func(context.Context) ([]model.Server, []string, []string, error), +) (selectedServers, error) { var out selectedServers includeSystem, includeBuiltin, includeUser := sel.system, sel.builtin, sel.user - if !includeSystem && !includeBuiltin && !includeUser { + if !includeSystem && !includeBuiltin && !includeUser && len(sel.servers) == 0 { includeSystem, includeBuiltin, includeUser = true, true, true } var lists [][]model.Server - system, ifaces, warnings, discoverErr := sysdns.Discover(ctx) + if len(sel.servers) > 0 { + inline, err := parseInlineServers(sel.servers) + if err != nil { + return out, err + } + lists = append(lists, inline) + } + system, ifaces, warnings, discoverErr := discover(ctx) if discoverErr != nil { out.warnings = append(out.warnings, "system DNS discovery failed: "+discoverErr.Error()) } else { @@ -89,10 +106,7 @@ func selectServers(ctx context.Context, sel selectionFlags) (selectedServers, er if err != nil { return out, err } - servers = serverlist.Filter(servers, serverlist.FilterOptions{ - Protocols: protos, - IPv4Only: sel.noIPv6, - }) + servers = filterSelectedServers(servers, out.systemServers, protos, sel.noIPv6, sel.system) if len(sel.only) > 0 { servers = filterOnly(servers, sel.only) } @@ -101,6 +115,51 @@ func selectServers(ctx context.Context, sel selectionFlags) (selectedServers, er return out, nil } +func filterSelectedServers(servers, system []model.Server, protocols []model.Protocol, ipv4Only, keepSystem bool) []model.Server { + systemKeys := make(map[string]bool, len(system)) + if keepSystem { + for _, server := range system { + systemKeys[server.Key()] = true + } + } + out := make([]model.Server, 0, len(servers)) + for _, server := range servers { + if ipv4Only && server.IsIPv6() { + continue + } + protocolMatches := len(protocols) == 0 || slices.Contains(protocols, server.Protocol) + if !protocolMatches && !systemKeys[server.Key()] { + continue + } + out = append(out, server) + } + return out +} + +func parseInlineServers(entries []string) ([]model.Server, error) { + servers := make([]model.Server, 0, len(entries)) + for _, raw := range entries { + entry := strings.TrimSpace(raw) + if entry == "" { + return nil, usageErrorf("--server requires a non-empty endpoint") + } + decoded, err := serverlist.DecodeText([]byte(entry)) + if err != nil { + return nil, usageErrorf("invalid --server value %q: %v", raw, err) + } + if len(decoded) != 1 { + return nil, usageErrorf("--server value %q must contain exactly one endpoint", raw) + } + server := decoded[0] + server.Source = model.SourceUser + if err := serverlist.ValidateAndPrepare(&server); err != nil { + return nil, usageErrorf("invalid --server value %q: %v", raw, err) + } + servers = append(servers, server) + } + return servers, nil +} + func matchSystemServerIDs(selected, system []model.Server) []string { systemKeys := make(map[string]bool, len(system)) for _, server := range system { diff --git a/cmd/dnsbench/run.go b/cmd/dnsbench/run.go index e4c911f..fff56bd 100644 --- a/cmd/dnsbench/run.go +++ b/cmd/dnsbench/run.go @@ -41,6 +41,7 @@ type runFlags struct { retryInterval time.Duration concurrency int pace time.Duration + paceExplicit bool gap time.Duration session string seed int64 @@ -74,6 +75,7 @@ time of day.`, Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { f.rounds = mustInt(cmd, "rounds") + f.paceExplicit = cmd.Flags().Changed("pace") return executeRun(cmd, &f) }, } @@ -90,8 +92,8 @@ time of day.`, cmd.Flags().DurationVar(&f.timeout, "timeout", base.Timeout, "timeout per query") cmd.Flags().IntVar(&f.retries, "retries", base.Retries, "retries per failed query") cmd.Flags().DurationVar(&f.retryInterval, "retry-interval", base.RetryInterval, "wait between retries") - cmd.Flags().IntVar(&f.concurrency, "concurrency", base.Concurrency, "maximum queries in flight at once") - cmd.Flags().DurationVar(&f.pace, "pace", base.PaceInterval, "starting spacing between any two query launches, across all resolvers; adapts between a quarter of it and 8x it (0 disables pacing)") + cmd.Flags().IntVar(&f.concurrency, "concurrency", base.Concurrency, "hard limit on simultaneous DNS queries; does not control QPS") + cmd.Flags().DurationVar(&f.pace, "pace", base.PaceInterval, "fixed minimum interval between query starts across all resolvers; when omitted, the 20ms default adapts up to 4x on possible shared-path congestion (0 disables pacing)") cmd.Flags().DurationVar(&f.gap, "gap", base.PerServerGap, "gap between consecutive queries to the same resolver") cmd.Flags().StringVar(&f.session, "session", string(base.Session), "connection reuse: cold or persistent") cmd.Flags().Int64Var(&f.seed, "seed", 0, "random seed (0 picks a random seed)") @@ -240,6 +242,7 @@ func executeRun(cmd *cobra.Command, f *runFlags) error { Stats: statsMap, Scores: scores, Samples: samples, + PaceAdjustments: engine.PaceAdjustments(), } res.Comparisons = buildComparisons(res, rankingMode) exportPaths, exportErr := writeExportFiles(res, f, rankingMode) @@ -389,6 +392,7 @@ func buildBenchConfig(f *runFlags, mode model.Mode, session model.SessionMode) ( cfg.RetryInterval = f.retryInterval cfg.Concurrency = f.concurrency cfg.PaceInterval = f.pace + cfg.PaceAdaptive = !f.paceExplicit && f.pace > 0 cfg.PerServerGap = f.gap cfg.Session = session cfg.Seed = f.seed diff --git a/cmd/dnsbench/run_test.go b/cmd/dnsbench/run_test.go index ce12ce1..aa3197c 100644 --- a/cmd/dnsbench/run_test.go +++ b/cmd/dnsbench/run_test.go @@ -22,6 +22,31 @@ func TestRunCommandModernDefaults(t *testing.T) { if got := cmd.Flags().Lookup("triage-threshold").DefValue; got != "200ms" { t.Fatalf("--triage-threshold default = %q, want 200ms", got) } + if got := cmd.Flags().Lookup("concurrency").Usage; !strings.Contains(got, "hard limit") || !strings.Contains(got, "does not control QPS") { + t.Fatalf("--concurrency help does not separate in-flight queries from QPS: %q", got) + } + if got := cmd.Flags().Lookup("pace").Usage; !strings.Contains(got, "fixed minimum interval") || !strings.Contains(got, "when omitted") { + t.Fatalf("--pace help does not distinguish fixed and adaptive behavior: %q", got) + } +} + +func TestBuildBenchConfigMakesExplicitPaceFixed(t *testing.T) { + base := model.DefaultBenchConfig(model.ModeStandard) + flags := runFlags{ + pace: 30 * time.Millisecond, + paceExplicit: true, + } + cfg, _ := buildBenchConfig(&flags, model.ModeStandard, model.SessionPersistent) + if cfg.PaceInterval != 30*time.Millisecond || cfg.PaceAdaptive { + t.Fatalf("explicit pacing = %s adaptive=%t, want fixed 30ms", cfg.PaceInterval, cfg.PaceAdaptive) + } + + flags.pace = base.PaceInterval + flags.paceExplicit = false + cfg, _ = buildBenchConfig(&flags, model.ModeStandard, model.SessionPersistent) + if cfg.PaceInterval != base.PaceInterval || !cfg.PaceAdaptive { + t.Fatalf("default pacing = %s adaptive=%t, want adaptive %s", cfg.PaceInterval, cfg.PaceAdaptive, base.PaceInterval) + } } func TestProbeCommandHasNoPacingFlag(t *testing.T) { diff --git a/cmd/dnsbench/selection_test.go b/cmd/dnsbench/selection_test.go index 707a5b9..22cd5a9 100644 --- a/cmd/dnsbench/selection_test.go +++ b/cmd/dnsbench/selection_test.go @@ -1,9 +1,13 @@ package main import ( + "context" "slices" + "strings" "testing" + "github.com/spf13/cobra" + "dnsbench/internal/model" ) @@ -52,3 +56,240 @@ func TestSelectionCommandsExposeProtocolsWithoutEncryptedOnly(t *testing.T) { t.Error("probe command unexpectedly exposes --encrypted-only") } } + +func TestSelectionCommandsExplainExplicitSystemProtocolOverride(t *testing.T) { + for name, cmd := range map[string]*cobra.Command{ + "run": newRunCmd(), + "probe": newProbeCmd(), + } { + flag := cmd.Flags().Lookup("system") + if flag == nil { + t.Errorf("%s command is missing --system", name) + continue + } + if !strings.Contains(flag.Usage, "--protocols") { + t.Errorf("%s --system help does not explain its precedence over --protocols: %q", name, flag.Usage) + } + } +} + +func TestExplicitSystemSourceBypassesOnlyTheProtocolFilter(t *testing.T) { + system := []model.Server{ + { + ID: "system-router", + Name: "System router", + Address: "192.168.1.1", + Port: 53, + Protocol: model.ProtoUDP, + Source: model.SourceSystem, + Enabled: true, + }, + } + discover := func(context.Context) ([]model.Server, []string, []string, error) { + return system, []string{"en0"}, nil, nil + } + selected, err := selectServersWithDiscovery(context.Background(), selectionFlags{ + system: true, + builtin: true, + protocols: []string{"doh3", "doq"}, + }, discover) + if err != nil { + t.Fatalf("selectServersWithDiscovery: %v", err) + } + if !slices.ContainsFunc(selected.servers, func(server model.Server) bool { + return server.ID == "system-router" + }) { + t.Fatal("the system DNS resolver was removed by --protocols") + } + if !slices.Equal(selected.systemIDs, []string{"system-router"}) { + t.Fatalf("system IDs = %q, want [system-router]", selected.systemIDs) + } + + encrypted := 0 + for _, server := range selected.servers { + if server.ID == "system-router" { + continue + } + if server.Protocol != model.ProtoDoH3 && server.Protocol != model.ProtoDoQ { + t.Errorf("non-system %s endpoint survived the protocol filter: %+v", server.Protocol, server) + } + encrypted++ + } + if encrypted == 0 { + t.Fatal("no DoH/3 or DoQ resolver remained alongside the system baseline") + } +} + +func TestExplicitSystemSourceStillHonorsNoIPv6(t *testing.T) { + system := []model.Server{ + {ID: "system-v4", Address: "192.168.1.1", Port: 53, Protocol: model.ProtoUDP, Source: model.SourceSystem, Enabled: true}, + {ID: "system-v6", Address: "2001:db8::53", Port: 53, Protocol: model.ProtoUDP, Source: model.SourceSystem, Enabled: true}, + } + discover := func(context.Context) ([]model.Server, []string, []string, error) { + return system, nil, nil, nil + } + selected, err := selectServersWithDiscovery(context.Background(), selectionFlags{ + system: true, + builtin: true, + protocols: []string{"doh3", "doq"}, + noIPv6: true, + }, discover) + if err != nil { + t.Fatalf("selectServersWithDiscovery: %v", err) + } + if slices.ContainsFunc(selected.servers, func(server model.Server) bool { + return server.IsIPv6() + }) { + t.Fatalf("IPv6 resolver survived --no-ipv6: %+v", selected.servers) + } + if !slices.Contains(selected.systemIDs, "system-v4") || slices.Contains(selected.systemIDs, "system-v6") { + t.Fatalf("system IDs = %q, want only the IPv4 system resolver", selected.systemIDs) + } +} + +func TestSelectionCommandsExposeRepeatableServerFlag(t *testing.T) { + commands := map[string]*cobra.Command{ + "run": newRunCmd(), + "probe": newProbeCmd(), + } + for name, cmd := range commands { + flag := cmd.Flag("server") + if flag == nil { + t.Errorf("%s command is missing --server", name) + continue + } + if got := flag.Value.Type(); got != "stringArray" { + t.Errorf("%s --server type = %q, want stringArray", name, got) + } + } + + run := newRunCmd() + if err := run.ParseFlags([]string{ + "--server", "https://dns.example/dns-query@192.0.2.1", + "--server", "quic://dns.example@192.0.2.1", + }); err != nil { + t.Fatalf("parse repeated --server flags: %v", err) + } + got, err := run.Flags().GetStringArray("server") + if err != nil { + t.Fatalf("get --server values: %v", err) + } + want := []string{ + "https://dns.example/dns-query@192.0.2.1", + "quic://dns.example@192.0.2.1", + } + if !slices.Equal(got, want) { + t.Errorf("--server values = %q, want %q", got, want) + } +} + +func TestParseInlineServersSupportsEncryptedEndpoints(t *testing.T) { + servers, err := parseInlineServers([]string{ + "https://dns.example/dns-query@192.0.2.1", + "quic://dns.example@192.0.2.1", + }) + if err != nil { + t.Fatalf("parseInlineServers: %v", err) + } + if len(servers) != 2 { + t.Fatalf("servers = %d, want 2", len(servers)) + } + + doh := servers[0] + if doh.Protocol != model.ProtoDoH || + doh.DoHURL != "https://dns.example/dns-query" || + doh.Address != "192.0.2.1" { + t.Errorf("DoH endpoint parsed as %+v", doh) + } + doq := servers[1] + if doq.Protocol != model.ProtoDoQ || + doq.TLSHostname != "dns.example" || + doq.Address != "192.0.2.1" { + t.Errorf("DoQ endpoint parsed as %+v", doq) + } + for i, server := range servers { + if server.Source != model.SourceUser || !server.Enabled || server.ID == "" { + t.Errorf("server %d was not prepared as a temporary user endpoint: %+v", i, server) + } + } +} + +func TestDistinctInlineEndpointsHaveUniqueSelectedIDs(t *testing.T) { + discover := func(context.Context) ([]model.Server, []string, []string, error) { + return nil, nil, nil, nil + } + selected, err := selectServersWithDiscovery(context.Background(), selectionFlags{ + servers: []string{ + "1.1.1.1", + "1.1.1.1:5353", + "https://dns.example/dns-query@192.0.2.1", + "https://dns.example/dns-query@192.0.2.2", + }, + }, discover) + if err != nil { + t.Fatalf("selectServersWithDiscovery: %v", err) + } + if len(selected.servers) != 4 { + t.Fatalf("selected %d servers, want 4", len(selected.servers)) + } + ids := make(map[string]bool, len(selected.servers)) + for _, server := range selected.servers { + if ids[server.ID] { + t.Errorf("duplicate selected server ID %q", server.ID) + } + ids[server.ID] = true + } +} + +func TestParseInlineServersRejectsInvalidOrMultipleEntries(t *testing.T) { + for _, entries := range [][]string{ + {""}, + {"not-an-endpoint"}, + {"8.8.8.8\n1.1.1.1"}, + } { + if _, err := parseInlineServers(entries); err == nil { + t.Errorf("parseInlineServers(%q): expected an error", entries) + } + } +} + +func TestInlineServersDisableImplicitSourceDefaults(t *testing.T) { + selected, err := selectServers(context.Background(), selectionFlags{ + servers: []string{ + "https://dns.example/dns-query@192.0.2.1", + "quic://dns.example@192.0.2.1", + }, + }) + if err != nil { + t.Fatalf("selectServers: %v", err) + } + if len(selected.servers) != 2 { + t.Fatalf("selected %d servers, want only the 2 inline endpoints", len(selected.servers)) + } + for _, server := range selected.servers { + if server.Source != model.SourceUser || server.Address != "192.0.2.1" { + t.Errorf("implicit source unexpectedly entered selection: %+v", server) + } + } +} + +func TestInlineServersCombineWithExplicitSources(t *testing.T) { + selected, err := selectServers(context.Background(), selectionFlags{ + builtin: true, + servers: []string{"https://dns.example/dns-query@192.0.2.1"}, + }) + if err != nil { + t.Fatalf("selectServers: %v", err) + } + if len(selected.servers) <= 1 { + t.Fatalf("selected %d servers, want inline endpoint plus built-ins", len(selected.servers)) + } + if selected.servers[0].DoHURL != "https://dns.example/dns-query" { + t.Fatalf("first selected server = %+v, want inline endpoint", selected.servers[0]) + } + if !slices.ContainsFunc(selected.servers[1:], func(server model.Server) bool { + return server.Source == model.SourceBuiltin + }) { + t.Error("explicit --builtin source was not combined with inline endpoint") + } +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e9c43c3..0ce3329 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -51,7 +51,7 @@ Supporting types: `TriageResult` (attempts, valid responses, best RTT, resulting ## Full benchmark flow (`dnsbench run`) -1. **Discovery and selection** — `sysdns.Discover` always identifies the system resolvers. Source flags only control which system, user and built-in endpoints (plus `--servers-file`) enter the benchmark; selected endpoints are merged by key and filtered (`--protocols`, `--no-ipv6`, `--only`). If no source flag is given, all three sources are included. +1. **Discovery and selection** — `sysdns.Discover` always identifies the system resolvers. Source flags only control which system, user and built-in endpoints (plus `--servers-file` and repeatable one-off `--server` entries) enter the benchmark; selected endpoints are merged by key and filtered (`--protocols`, `--no-ipv6`, `--only`). If no source flag or one-off endpoint is given, all three sources are included. Supplying `--server` disables that automatic three-source default while preserving any explicitly selected sources and server file. An explicitly selected `--system` source is exempt from `--protocols`, while `--no-ipv6` and `--only` remain in force. 2. **Registration** — selected servers carry stable IDs. Discovered system resolvers are retained separately, and any selected endpoint with the same transport/address/port key is marked as current DNS even if its selected record came from the built-in or user list. 3. **Characterization (probe)** — unless `--skip-probe`, every server is probed concurrently: reachability, baseline (A/AAAA, EDNS0), DNSSEC matrix, NXDOMAIN checks, reverse PTR, and extended checks with `--extended`. One persistent `Querier` is reused per server, and `--concurrency` bounds simultaneous server checks. Characterization has no launch pacer. 4. **Triage** — unless `--no-triage`, each server answers up to `--triage-attempts` (default 10) cached-domain queries, launched through the same global pacer and in-flight semaphore as benchmark queries. Only valid A/CNAME answers count. A built-in server whose best RTT exceeds `--triage-threshold` (default 200ms) becomes `benched` ("out of contention") unless `--force-all`; custom, user and system resolvers are checked only for availability and ignore the latency cutoff. A server with no valid answers becomes `offline`. Only `active` servers proceed. For the default `--session persistent`, one session per server is opened now; servers whose session fails to open become `error`. @@ -67,17 +67,17 @@ Interruption: the first Ctrl+C cancels the context, the engine stops cleanly and ## Concurrency strategy -- **Global pacer** — a single seeded clock spaces all query launches at least `--pace` apart (default 20ms plus jitter; 0 disables). Rounds, retries and triage probes all pass through it, so the worst-case instantaneous burst is one packet on any hardware. This is the primary protection against self-induced congestion — without it, launch bursts inflate tail latencies and produce timeout loss that gets misattributed to the servers under test. The interval is adaptive (AIMD): every completed query feeds `pacer.observe` — 50 answered queries shorten the interval by `--pace`/8 (floor `--pace`/4); timeouts on ≥3 distinct previously-answering servers within a 2s sliding window double it (ceiling 8× `--pace`), emit an `EvPaceAdjust` event and start a one-timeout cooldown. Timeouts from servers that never answered are excluded from the signature. -- **Semaphore** — a channel of capacity `--concurrency` bounds how many queries are in flight at once (shared by triage and rounds); it acts as backpressure when many servers stop answering. +- **Global pacer** — a single clock spaces rounds, retries and triage query launches, so the worst-case instantaneous burst is one packet per tick on any hardware (`--pace 0` disables it). When the flag is omitted, the default is an adaptive 20ms floor with seeded jitter. The pacer owns the congestion heuristic behind one observation interface: it groups endpoints by known operator (falling back to dial address, then hostname), remembers per-endpoint answers for 30 seconds, and correlates timeout attempt start times. Timeouts from ≥3 independent, recently healthy groups in a 2s launch window move pacing through 40ms and 80ms, emit a structured `EvPaceAdjust`, record exportable evidence and start a cooldown of at least 3 seconds or one query timeout. After the cooldown, every 50 answered queries subtract 10ms until 20ms is restored. Passing `--pace DURATION` constructs a fixed pacer with no observation-driven changes and no jitter. `BenchConfig.PaceAdaptive` records which behavior is active. +- **Semaphore** — a channel of capacity `--concurrency` bounds simultaneous query attempts (shared by triage and rounds). The engine acquires a permit after pause/per-server waits and immediately before the global pace reservation, holds it through network I/O, and releases it after every attempt, including retries. Category work, per-server gaps and retry delays therefore cannot reserve query capacity. - **Per-round barrier** — a `sync.WaitGroup` closes every round: no server starts round N+1 until every server has finished round N, keeping rounds time-aligned across servers. -- **Seeded shuffling** — all randomness derives from one run seed (`--seed`, random when 0, recorded in exports). The per-round server order uses `PCG(seed, seq<<1|1)`; each server's per-round RNG uses `PCG(seed, seq<<24 | stableIdx<<1)` for category order and random labels. The same seed reproduces the same schedule and query names. +- **Seeded shuffling** — all randomness derives from one run seed (`--seed`, random when 0, recorded in exports). The per-round server order uses `PCG(seed, seq<<1|1)`; each server's per-round RNG uses `PCG(seed, seq<<24 | stableIdx<<1)` for category order and random labels. The same seed reproduces those shuffled positions, category orders and query names; wall-clock admission order remains response- and scheduler-dependent. - **Cold vs persistent sessions** — persistent (default): one `Querier` per server is opened before warmup and reused (TCP/DoT keep the connection; DoH keeps HTTP keep-alives; UDP keeps one connected socket per server, so a whole run consumes one NAT/conntrack entry per server instead of one per query — small router state tables are a real-world loss source). A persistent UDP socket survives timeouts and drains stale late replies by DNS message ID before accepting an answer. DoQ likewise cancels a timed-out transaction at stream scope and preserves a healthy QUIC connection; framing violations remain connection-fatal. Cold mode explicitly builds and tears down a querier per query. The two modes are never mixed within a run. - **Per-server pacing** — the engine records each server's last query completion under a mutex. Before every subsequent query it waits until `lastQueryEnd + PerServerGap`, so round barriers cannot accidentally bypass pacing. - **Context cancellation** — one `context.Context` flows from the signal handler through triage, rounds, queries and sleeps; canceled queries are marked `canceled` and not recorded as samples. - **Pause/resume** — the engine has a pause gate (`Engine.Pause`/`Engine.Resume`); workers block at round and query boundaries while paused. In the current CLI the gate is driven by the connectivity watchdog. - **Connectivity watchdog with canary** — enabled by default (`ConnectivityWatch`). If two consecutive rounds end with 100% of their queries timing out, the engine emits `conn-lost`, pauses the gate, and sends a canary query (first cached domain to the first active server) every 5 seconds. On the first answered canary it emits `conn-restored` and resumes. This prevents an internet outage from being recorded as server loss. -The engine communicates with the UI through a buffered event channel (`triage`, `sample`, `round-done`, `state-change`, `warn`, `conn-lost`, `conn-restored`, `done`). +The engine communicates with the UI through a buffered event channel (`triage`, `sample`, `round-done`, `state-change`, `warn`, `pace-adjust`, `conn-lost`, `conn-restored`, `done`). Backoff events carry both a human-readable notice and structured `PaceAdjustment` evidence; clean-answer recovery adjustments are retained for JSON export without producing repetitive live notices. For an interactive TTY, the live UI is a Bubble Tea `Model` driven by benchmark messages through `Update` and rendered from `View`. Bubble Tea owns the alternate screen, coalesces state changes at a bounded frame rate, repaints in place, and reports `WindowSizeMsg` resize events. Each visible resolver is a multi-line block with one shared-axis latency bar per enabled category. Category names and markers live only in the ordered legend; resolver rows spend that space on the bars. Adjacent categories alternate lower and upper half-height blocks so the default pair touches at the line boundary. The shared linear scale is the P90 of the displayed statistic across visible active resolvers, and values above it carry an overflow marker. The statistic follows `--sort` for median, mean and p95; loss/name sorting keeps median bars, while `cost` (alias `score`) uses median until the final ranking exists. The reliability cell labels unanswered-query loss with its numerator/denominator and reports semantically invalid answers separately. diff --git a/docs/METHODOLOGY.md b/docs/METHODOLOGY.md index 34fe240..a70b8bc 100644 --- a/docs/METHODOLOGY.md +++ b/docs/METHODOLOGY.md @@ -52,11 +52,14 @@ Instead of testing server A to completion and then server B (which would let net - The **server order is reshuffled every round** with a seeded PCG generator, so no server systematically benefits from always being measured first or last. - Each server runs its **categories in per-server shuffled order**. A `--gap` pause (default 40ms) is enforced from the completion of one query to the start of the next query to that server, including across category and round boundaries. -- Every benchmark query launch — across all servers, including triage probes and retries — passes through a **global pacer**: launches are spaced at least `--pace` apart (default 20ms, plus a small seeded jitter; `--pace 0` disables). Pacing bounds the instantaneous packet burst to one query per tick regardless of hardware, so the benchmark cannot congest the local Wi-Fi link, NAT router or uplink and then misattribute the resulting drops and queueing delay as server loss and latency. -- The pace **adapts AIMD-style** while the run progresses. Speed-up: every 50 answered queries with no congestion signal shorten the interval by an eighth of `--pace`, down to a floor of a quarter of `--pace`, so clean networks earn back wall-clock time. Back-off: when timeouts land on **3 or more distinct servers that had previously answered** within a 2-second window — the signature of local congestion, not of any one server — the interval doubles, up to a ceiling of 8× `--pace`, and further backoffs are suppressed for one query-timeout period so the tail of the same event is not double-counted. Servers that never answered do not count toward the signature, so dead or blackholed endpoints (e.g. unroutable IPv6) cannot slow the run. Each backoff is reported live as a pacing notice. -- Cross-server parallelism is additionally capped by the `--concurrency` in-flight limit (default 8), which acts as a safety valve: when many servers stop answering, at most that many queries wait on timeouts at once, which naturally throttles the stream. +- Every benchmark query launch — across all servers, including triage probes and retries — passes through a **global pacer** (`--pace 0` disables it). Pacing bounds the instantaneous packet burst to one query per tick regardless of hardware, so the benchmark cannot congest the local Wi-Fi link, NAT router or uplink and then misattribute the resulting drops and queueing delay as server loss and latency. +- When `--pace` is omitted, the 20ms default uses a small seeded jitter and adaptive safety backoff, but **never launches faster than 20ms**. A backoff requires timeouts from **3 or more independent resolver failure domains** whose attempts started within the same 2-second window and whose individual endpoints answered within the preceding 30 seconds. Known operators define failure domains; without operator metadata, the dial address defines the domain even when protocols differ, with a DoH hostname used only when no address is pinned. This prevents several Google, Cloudflare or same-address transports from impersonating independent evidence. The 30-second freshness rule also excludes endpoints that answered only much earlier in the run. +- The signal is deliberately reported as **possible shared-path congestion**, not proof of local congestion: an ISP, route, authoritative target or other shared dependency can produce the same pattern. The live notice stays compact, while its structured event and JSON `pace_adjustments` evidence include the affected endpoint IDs, failure-domain labels, query categories and protocols, making category- or protocol-concentrated incidents visible instead of silently overstating confidence. The correlation window uses attempt launch times rather than timeout completion times, so transports with different timeout behavior can still be compared correctly. +- A qualifying burst moves the interval first to 40ms and then at most 80ms. Further backoffs are suppressed for at least 3 seconds or one query timeout, whichever is longer, so the tail of the same event is not double-counted. After that cooldown, every 50 answered queries recover by 10ms until the baseline is restored: `80ms → 70ms → 60ms → … → 20ms`. Runs with fewer than three independent, recently healthy domains retain baseline pacing because DNS observations alone cannot separate one resolver's instability from path trouble. +- Passing `--pace DURATION` selects **fixed pacing**: there is no adaptation and no jitter. The pacer uses `DURATION` for every reservation and never changes that value; observed launch timing can still vary because of occupied query slots and OS scheduling. This distinction is retained in exports as `pace_adaptive`. +- Cross-server parallelism is additionally capped by the `--concurrency` limit (default 8). A permit is acquired for each actual query attempt, immediately before its global pacing reservation, and released when that attempt finishes; per-server gaps, category processing and retry delays do not retain it. This acts as a safety valve when many servers stop answering while allowing another resolver to use capacity during those waits. -All randomness (orders, labels) derives from a single seed. `--seed 0` (the default) picks a random seed, which is recorded in exports; re-running with `--seed N` reproduces the exact schedule and query names. +All randomness (orders, labels) derives from a single seed. `--seed 0` (the default) picks a random seed, which is recorded in exports; re-running with `--seed N` reproduces shuffled positions, per-server category order and query names. Wall-clock admission order can still vary with goroutine and network timing. ## Triage @@ -93,7 +96,7 @@ An encrypted endpoint identified by hostname has a chicken-and-egg problem: the dnsbench avoids this by pinning: every encrypted endpoint in the built-in list carries the resolver's IP address, dials it directly, and uses the hostname only for SNI and certificate validation. No name resolution happens before an encrypted query. -Use `--protocols dot,doh,doh3,doq` to select only encrypted transports. Protocol selection does not exclude hostname-only custom entries, so it is a transport filter rather than a guarantee about their bootstrap lookup. +Use `--protocols dot,doh,doh3,doq` to select only encrypted transports. Explicitly select `--system` together with the desired encrypted sources when the system-configured UDP resolver should remain in the same run as a comparison baseline despite that filter. Protocol selection does not exclude hostname-only custom entries, so it is a transport filter rather than a guarantee about their bootstrap lookup. When an explicitly configured DoH/3 endpoint is not pinned, its hostname is resolved by the HTTP/3 transport with the query context. The same cancellation or deadline that bounds the request therefore also bounds bootstrap resolution. diff --git a/internal/bench/engine.go b/internal/bench/engine.go index 546ee87..101eb6a 100644 --- a/internal/bench/engine.go +++ b/internal/bench/engine.go @@ -6,6 +6,7 @@ import ( "encoding/binary" "errors" "fmt" + "slices" "sync" "time" @@ -29,6 +30,7 @@ type Engine struct { mu sync.Mutex samples []model.Sample + paceChanges []model.PaceAdjustment triage map[string]*model.TriageResult states map[string]model.ServerState paceMu sync.Mutex @@ -48,7 +50,7 @@ func NewEngine(servers []model.Server, cfg model.BenchConfig, factory transport. seed: seed, canaryInterval: 5 * time.Second, gate: &pauseGate{}, - pace: newPacer(cfg.PaceInterval, seed, cfg.Timeout), + pace: newPacer(cfg.PaceInterval, cfg.PaceAdaptive, seed, cfg.Timeout), slots: make(chan struct{}, max(1, cfg.Concurrency)), triage: make(map[string]*model.TriageResult), states: make(map[string]model.ServerState), @@ -109,6 +111,21 @@ func (e *Engine) Result() ([]model.Sample, map[string]*model.TriageResult, map[s return samples, triage, states, e.seed } +func (e *Engine) PaceAdjustments() []model.PaceAdjustment { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]model.PaceAdjustment, len(e.paceChanges)) + for i, adjustment := range e.paceChanges { + out[i] = adjustment + out[i].ServerIDs = append([]string(nil), adjustment.ServerIDs...) + out[i].FailureDomains = append([]string(nil), adjustment.FailureDomains...) + out[i].Categories = append([]model.Category(nil), adjustment.Categories...) + out[i].Protocols = append([]model.Protocol(nil), adjustment.Protocols...) + } + slices.SortFunc(out, func(a, b model.PaceAdjustment) int { return a.At.Compare(b.At) }) + return out +} + func (e *Engine) Run(ctx context.Context) error { defer close(e.events) active, err := e.prepare(ctx) @@ -206,19 +223,58 @@ func (e *Engine) attemptQuery(ctx context.Context, s model.Server, name string) return e.singleQuery(ctx, s, name) } -func (e *Engine) observePace(ctx context.Context, serverID string, res model.QueryResult) { - timedOut := res.Err.IsTimeout() - if !timedOut && !res.Answered() { +func (e *Engine) executeAttempt(ctx context.Context, attempt func() model.QueryResult) (model.QueryResult, bool) { + if e.gate.wait(ctx) != nil || !e.acquireSlot(ctx) { + return model.QueryResult{}, false + } + defer e.releaseSlot() + if !e.pace.wait(ctx) { + return model.QueryResult{}, false + } + return attempt(), true +} + +func (e *Engine) observePace( + ctx context.Context, + server model.Server, + category model.Category, + startedAt time.Time, + res model.QueryResult, +) { + adj, changed := e.pace.observeAttempt(server, category, startedAt, res) + if !changed { return } - adj, changed := e.pace.observe(serverID, timedOut) - if !changed || adj.faster { + reason := model.PaceSharedTimeoutBurst + if adj.recovering { + reason = model.PaceCleanAnswerRecovery + } + change := model.PaceAdjustment{ + At: adj.at, + Reason: reason, + FromInterval: adj.from, + ToInterval: adj.to, + EvidenceStartedAt: adj.evidenceStartedAt, + EvidenceEndedAt: adj.evidenceEndedAt, + Window: adj.window, + Timeouts: adj.timeouts, + ServerIDs: append([]string(nil), adj.serverIDs...), + FailureDomains: append([]string(nil), adj.failureDomains...), + Categories: append([]model.Category(nil), adj.categories...), + Protocols: append([]model.Protocol(nil), adj.protocols...), + CleanAnswers: adj.cleanAnswers, + } + e.mu.Lock() + e.paceChanges = append(e.paceChanges, change) + e.mu.Unlock() + if adj.recovering { return } e.emit(ctx, model.Event{ Type: model.EvPaceAdjust, - Msg: fmt.Sprintf("local congestion — pacing eased %s → %s (%d timeouts on %d servers)", - adj.from, adj.to, adj.timeouts, adj.servers), + Pace: &change, + Msg: fmt.Sprintf("possible shared-path congestion — pace %s → %s (%d resolver groups timed out in %s)", + adj.from, adj.to, len(adj.failureDomains), adj.window), }) } diff --git a/internal/bench/engine_test.go b/internal/bench/engine_test.go index 24fae62..dc05fc3 100644 --- a/internal/bench/engine_test.go +++ b/internal/bench/engine_test.go @@ -284,7 +284,7 @@ func TestTriageRejectsSemanticallyInvalidDNSResponses(t *testing.T) { } } -func TestDeterministicShufflingWithFixedSeed(t *testing.T) { +func TestDeterministicQueryPlanWithFixedSeed(t *testing.T) { runOnce := func() []string { f := newFakeFactory() cfg := testConfig() @@ -307,6 +307,7 @@ func TestDeterministicShufflingWithFixedSeed(t *testing.T) { s := ev.Sample seq = append(seq, fmt.Sprintf("%d|%t|%s|%s|%s", s.Round, s.Warmup, s.ServerID, s.Category, s.QName)) } + slices.Sort(seq) return seq } first := runOnce() @@ -483,6 +484,78 @@ func TestPersistentSessionCreatesOneQuerierPerServer(t *testing.T) { } } +func TestConcurrencyCapsActualQueryAttempts(t *testing.T) { + f := newFakeFactory() + var inFlight atomic.Int32 + var peak atomic.Int32 + track := func(_ context.Context, _ int, _ transport.Question) model.QueryResult { + current := inFlight.Add(1) + for { + seen := peak.Load() + if current <= seen || peak.CompareAndSwap(seen, current) { + break + } + } + time.Sleep(25 * time.Millisecond) + inFlight.Add(-1) + return okResult(time.Millisecond) + } + ids := []string{"s1", "s2", "s3", "s4", "s5", "s6"} + for _, id := range ids { + f.script(id, track) + } + cfg := testConfig() + cfg.TriageEnabled = false + cfg.WarmupRounds = 0 + cfg.Rounds = 1 + cfg.Categories = []model.Category{model.CatCached} + cfg.Concurrency = 2 + cfg.PaceInterval = 0 + cfg.PerServerGap = 0 + e := NewEngine(testServers(ids...), cfg, f.factory) + if _, err := runEngine(t, e, context.Background()); err != nil { + t.Fatalf("Run failed: %v", err) + } + if got := peak.Load(); got != int32(cfg.Concurrency) { + t.Fatalf("peak in-flight queries = %d, want %d", got, cfg.Concurrency) + } +} + +func TestConcurrencySlotIsReleasedDuringPerServerGap(t *testing.T) { + f := newFakeFactory() + var mu sync.Mutex + var order []string + for _, id := range []string{"s1", "s2"} { + id := id + f.script(id, func(_ context.Context, _ int, _ transport.Question) model.QueryResult { + mu.Lock() + order = append(order, id) + mu.Unlock() + return okResult(time.Millisecond) + }) + } + cfg := testConfig() + cfg.TriageEnabled = false + cfg.WarmupRounds = 0 + cfg.Rounds = 1 + cfg.Categories = []model.Category{model.CatCached, model.CatTLD} + cfg.Concurrency = 1 + cfg.PaceInterval = 0 + cfg.PerServerGap = 50 * time.Millisecond + e := NewEngine(testServers("s1", "s2"), cfg, f.factory) + if _, err := runEngine(t, e, context.Background()); err != nil { + t.Fatalf("Run failed: %v", err) + } + mu.Lock() + defer mu.Unlock() + if len(order) != 4 { + t.Fatalf("query order = %v, want four attempts", order) + } + if order[0] == order[1] { + t.Fatalf("query order = %v; concurrency slot stayed with one server during its gap", order) + } +} + func TestColdSessionCreatesOneQuerierPerQuery(t *testing.T) { f := newFakeFactory() cfg := testConfig() @@ -534,6 +607,43 @@ func TestMeasureRecordsEndToEndRetryCost(t *testing.T) { } } +func TestCancellationDuringRetryAdmissionDoesNotRecordAnAnsweredSample(t *testing.T) { + f := newFakeFactory() + firstFailure := make(chan struct{}) + var e *Engine + f.script("s1", func(_ context.Context, n int, _ transport.Question) model.QueryResult { + if n == 0 { + e.Pause() + close(firstFailure) + } + return networkResult("temporary failure") + }) + cfg := testConfig() + cfg.TriageEnabled = false + cfg.WarmupRounds = 0 + cfg.Rounds = 1 + cfg.Categories = []model.Category{model.CatCached} + cfg.Retries = 1 + cfg.RetryInterval = 0 + cfg.PaceInterval = 0 + cfg.PerServerGap = 0 + e = NewEngine(testServers("s1"), cfg, f.factory) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + <-firstFailure + time.Sleep(50 * time.Millisecond) + cancel() + }() + if _, err := runEngine(t, e, ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Run error = %v, want context.Canceled", err) + } + samples, _, _, _ := e.Result() + if len(samples) != 0 { + t.Fatalf("samples = %+v, want no fabricated partial sample", samples) + } +} + func TestPerServerGapAppliesAcrossRoundBoundaries(t *testing.T) { f := newFakeFactory() var mu sync.Mutex diff --git a/internal/bench/pacer.go b/internal/bench/pacer.go index 6f546b3..14f570b 100644 --- a/internal/bench/pacer.go +++ b/internal/bench/pacer.go @@ -3,43 +3,68 @@ package bench import ( "context" "math/rand/v2" + "net/url" + "slices" + "strings" "sync" "time" + + "dnsbench/internal/model" +) + +const ( + minCongestedFailureDomains = 3 + recentAnswerWindow = 30 * time.Second ) type paceAdjust struct { - from time.Duration - to time.Duration - window time.Duration - timeouts int - servers int - faster bool + at time.Time + from time.Duration + to time.Duration + window time.Duration + timeouts int + servers int + cleanAnswers int + evidenceStartedAt time.Time + evidenceEndedAt time.Time + serverIDs []string + failureDomains []string + categories []model.Category + protocols []model.Protocol + recovering bool } type paceTimeout struct { - server string - at time.Time + serverID string + domainKey string + domainName string + category model.Category + protocol model.Protocol + startedAt time.Time } type pacer struct { mu sync.Mutex interval time.Duration + adaptive bool rng *rand.Rand next time.Time - minInterval time.Duration + baseInterval time.Duration maxInterval time.Duration - step time.Duration + recoveryStep time.Duration window time.Duration + healthyFor time.Duration + retainFor time.Duration cooldown time.Duration cleanTarget int cleanCount int suppressUntil time.Time timeouts []paceTimeout - answered map[string]bool + answeredAt map[string]time.Time } -func newPacer(interval time.Duration, seed int64, queryTimeout time.Duration) *pacer { +func newPacer(interval time.Duration, adaptive bool, seed int64, queryTimeout time.Duration) *pacer { if interval <= 0 { return nil } @@ -47,16 +72,20 @@ func newPacer(interval time.Duration, seed int64, queryTimeout time.Duration) *p if queryTimeout > cooldown { cooldown = queryTimeout } + retainFor := max(queryTimeout, 0) + 2*time.Second return &pacer{ - interval: interval, - rng: rand.New(rand.NewPCG(uint64(seed), 0x9e3779b97f4a7c15)), - minInterval: max(interval/4, 1), - maxInterval: interval * 8, - step: max(interval/8, 1), - window: 2 * time.Second, - cooldown: cooldown, - cleanTarget: 50, - answered: make(map[string]bool), + interval: interval, + adaptive: adaptive, + rng: rand.New(rand.NewPCG(uint64(seed), 0x9e3779b97f4a7c15)), + baseInterval: interval, + maxInterval: interval * 4, + recoveryStep: max(interval/2, 1), + window: 2 * time.Second, + healthyFor: recentAnswerWindow, + retainFor: retainFor, + cooldown: cooldown, + cleanTarget: 50, + answeredAt: make(map[string]time.Time), } } @@ -69,27 +98,52 @@ func (p *pacer) wait(ctx context.Context) bool { if now := time.Now(); at.Before(now) { at = now } - jitter := time.Duration(p.rng.Int64N(int64(p.interval)/4 + 1)) + var jitter time.Duration + if p.adaptive { + jitter = time.Duration(p.rng.Int64N(int64(p.interval)/4 + 1)) + } p.next = at.Add(p.interval + jitter) p.mu.Unlock() return sleepCtx(ctx, time.Until(at)) } -func (p *pacer) observe(serverID string, timedOut bool) (paceAdjust, bool) { - if p == nil { +func (p *pacer) observeAttempt( + server model.Server, + category model.Category, + startedAt time.Time, + res model.QueryResult, +) (paceAdjust, bool) { + return p.observeAttemptAt(server, category, startedAt, time.Now(), res) +} + +func (p *pacer) observeAttemptAt( + server model.Server, + category model.Category, + startedAt time.Time, + observedAt time.Time, + res model.QueryResult, +) (paceAdjust, bool) { + if p == nil || !p.adaptive { return paceAdjust{}, false } + timedOut := res.Err.IsTimeout() + if !timedOut && !res.Answered() { + return paceAdjust{}, false + } + if startedAt.IsZero() { + startedAt = observedAt + } p.mu.Lock() defer p.mu.Unlock() if timedOut { - return p.backOffLocked(serverID, time.Now()) + return p.backOffLocked(server, category, startedAt, observedAt) } - p.answered[serverID] = true - return p.speedUpLocked() + p.answeredAt[serverHealthKey(server)] = observedAt + return p.recoverLocked(observedAt) } -func (p *pacer) speedUpLocked() (paceAdjust, bool) { - if p.interval <= p.minInterval { +func (p *pacer) recoverLocked(now time.Time) (paceAdjust, bool) { + if p.interval <= p.baseInterval || now.Before(p.suppressUntil) { return paceAdjust{}, false } p.cleanCount++ @@ -98,43 +152,193 @@ func (p *pacer) speedUpLocked() (paceAdjust, bool) { } p.cleanCount = 0 from := p.interval - p.interval = max(p.interval-p.step, p.minInterval) - return paceAdjust{from: from, to: p.interval, faster: true}, true + p.interval = max(p.interval-p.recoveryStep, p.baseInterval) + return paceAdjust{ + at: now, + from: from, + to: p.interval, + cleanAnswers: p.cleanTarget, + recovering: true, + }, true } -func (p *pacer) backOffLocked(serverID string, now time.Time) (paceAdjust, bool) { - if !p.answered[serverID] || now.Before(p.suppressUntil) { +func (p *pacer) backOffLocked( + server model.Server, + category model.Category, + startedAt time.Time, + observedAt time.Time, +) (paceAdjust, bool) { + if observedAt.Before(p.suppressUntil) { return paceAdjust{}, false } - cutoff := now.Add(-p.window) + answeredAt, answered := p.answeredAt[serverHealthKey(server)] + if !answered || answeredAt.After(startedAt) || startedAt.Sub(answeredAt) > p.healthyFor { + return paceAdjust{}, false + } + p.cleanCount = 0 + cutoff := observedAt.Add(-p.retainFor) kept := p.timeouts[:0] for _, t := range p.timeouts { - if t.at.After(cutoff) { + if !t.startedAt.Before(cutoff) { kept = append(kept, t) } } - p.timeouts = append(kept, paceTimeout{server: serverID, at: now}) - distinct := make(map[string]struct{}, len(p.timeouts)) - for _, t := range p.timeouts { - distinct[t.server] = struct{}{} - } - if len(distinct) < 3 { + domainKey, domainName := serverFailureDomain(server) + p.timeouts = append(kept, paceTimeout{ + serverID: serverHealthKey(server), + domainKey: domainKey, + domainName: domainName, + category: category, + protocol: server.Protocol, + startedAt: startedAt, + }) + cluster := p.congestedClusterLocked() + if len(cluster) == 0 { return paceAdjust{}, false } - count := len(p.timeouts) p.timeouts = p.timeouts[:0] p.cleanCount = 0 - p.suppressUntil = now.Add(p.cooldown) + p.suppressUntil = observedAt.Add(p.cooldown) from := p.interval - p.interval = min(p.interval*2, p.maxInterval) + firstBackoff := min(p.baseInterval*2, p.maxInterval) + if p.interval < firstBackoff { + p.interval = firstBackoff + } else { + p.interval = p.maxInterval + } if p.interval == from { return paceAdjust{}, false } + serverIDs, failureDomains, categories, protocols, firstStart, lastStart := paceEvidence(cluster) return paceAdjust{ - from: from, - to: p.interval, - window: p.window, - timeouts: count, - servers: len(distinct), + at: observedAt, + from: from, + to: p.interval, + window: p.window, + timeouts: len(cluster), + servers: len(serverIDs), + evidenceStartedAt: firstStart, + evidenceEndedAt: lastStart, + serverIDs: serverIDs, + failureDomains: failureDomains, + categories: categories, + protocols: protocols, }, true } + +func (p *pacer) congestedClusterLocked() []paceTimeout { + slices.SortFunc(p.timeouts, func(a, b paceTimeout) int { + return a.startedAt.Compare(b.startedAt) + }) + var best []paceTimeout + bestDomains := 0 + for left := range p.timeouts { + domains := make(map[string]struct{}) + for right := left; right < len(p.timeouts); right++ { + if p.timeouts[right].startedAt.Sub(p.timeouts[left].startedAt) > p.window { + break + } + domains[p.timeouts[right].domainKey] = struct{}{} + if len(domains) > bestDomains || + (len(domains) == bestDomains && right-left+1 > len(best)) { + bestDomains = len(domains) + best = p.timeouts[left : right+1] + } + } + } + if bestDomains < minCongestedFailureDomains { + return nil + } + return append([]paceTimeout(nil), best...) +} + +func paceEvidence(cluster []paceTimeout) ( + serverIDs []string, + failureDomains []string, + categories []model.Category, + protocols []model.Protocol, + firstStart time.Time, + lastStart time.Time, +) { + servers := make(map[string]struct{}, len(cluster)) + domains := make(map[string]string, len(cluster)) + cats := make(map[model.Category]struct{}, len(cluster)) + protos := make(map[model.Protocol]struct{}, len(cluster)) + for i, t := range cluster { + servers[t.serverID] = struct{}{} + domains[t.domainKey] = t.domainName + if t.category != "" { + cats[t.category] = struct{}{} + } + if t.protocol != "" { + protos[t.protocol] = struct{}{} + } + if i == 0 || t.startedAt.Before(firstStart) { + firstStart = t.startedAt + } + if i == 0 || t.startedAt.After(lastStart) { + lastStart = t.startedAt + } + } + for id := range servers { + serverIDs = append(serverIDs, id) + } + for _, name := range domains { + failureDomains = append(failureDomains, name) + } + for category := range cats { + categories = append(categories, category) + } + for protocol := range protos { + protocols = append(protocols, protocol) + } + slices.Sort(serverIDs) + slices.Sort(failureDomains) + slices.Sort(categories) + slices.Sort(protocols) + return serverIDs, failureDomains, categories, protocols, firstStart, lastStart +} + +func serverHealthKey(server model.Server) string { + if server.ID != "" { + return server.ID + } + key, _ := serverFailureDomain(server) + return key +} + +func serverFailureDomain(server model.Server) (string, string) { + if operator := normalizedLabel(server.Operator); operator != "" { + return "operator:" + strings.ToLower(operator), operator + } + if server.Address != "" { + address := strings.ToLower(strings.TrimSpace(server.Address)) + if ip, ok := server.IP(); ok { + address = ip.Unmap().String() + } + return "address:" + address, address + } + if server.DoHURL != "" { + if parsed, err := url.Parse(server.DoHURL); err == nil { + if host := strings.ToLower(parsed.Hostname()); host != "" { + return "host:" + host, host + } + } + } + if host := strings.ToLower(strings.TrimSpace(server.TLSHostname)); host != "" { + return "host:" + host, host + } + label := normalizedLabel(server.DisplayName()) + if label == "" { + label = "unknown resolver" + } + key := normalizedLabel(server.ID) + if key == "" { + key = label + } + return "server:" + strings.ToLower(key), label +} + +func normalizedLabel(label string) string { + return strings.Join(strings.Fields(label), " ") +} diff --git a/internal/bench/pacer_test.go b/internal/bench/pacer_test.go index 254cca1..300c979 100644 --- a/internal/bench/pacer_test.go +++ b/internal/bench/pacer_test.go @@ -2,6 +2,7 @@ package bench import ( "context" + "slices" "sort" "strings" "sync" @@ -13,7 +14,7 @@ import ( ) func TestPacerSpacesSequentialWaits(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) + p := newPacer(20*time.Millisecond, true, 1, 0) start := time.Now() for i := 0; i < 5; i++ { if !p.wait(context.Background()) { @@ -41,7 +42,7 @@ func TestNilPacerDoesNotBlock(t *testing.T) { func TestPacerStopsOnCanceledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - if newPacer(time.Hour, 1, 0).wait(ctx) { + if newPacer(time.Hour, true, 1, 0).wait(ctx) { t.Error("live pacer wait = true on canceled context") } var p *pacer @@ -50,23 +51,60 @@ func TestPacerStopsOnCanceledContext(t *testing.T) { } } +func TestFixedPacerKeepsTheExplicitInterval(t *testing.T) { + p := newPacer(20*time.Millisecond, false, 1, 0) + p.next = time.Now().Add(time.Hour) + before := p.next + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if p.wait(ctx) { + t.Fatal("fixed pacer wait returned true on canceled context") + } + if step := p.next.Sub(before); step != 20*time.Millisecond { + t.Fatalf("fixed pacer reservation step = %s, want exactly 20ms without jitter", step) + } + for i := 0; i < 100; i++ { + if adj, changed := observeTestPacer(p, "s1", i%2 == 0); changed { + t.Fatalf("fixed pacer adapted after an observation: %+v", adj) + } + } + if p.interval != 20*time.Millisecond { + t.Fatalf("fixed interval = %s, want 20ms", p.interval) + } +} + +func observeTestPacer(p *pacer, serverID string, timedOut bool) (paceAdjust, bool) { + now := time.Now() + result := okResult(time.Millisecond) + if timedOut { + result = timeoutResult() + } + return p.observeAttemptAt( + model.Server{ID: serverID, Name: serverID}, + "", + now, + now, + result, + ) +} + func markAnswered(p *pacer, ids ...string) { for _, id := range ids { - p.observe(id, false) + observeTestPacer(p, id, false) } } func TestPacerBacksOffOnCrossServerTimeouts(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) + p := newPacer(20*time.Millisecond, true, 1, 0) markAnswered(p, "s1", "s2", "s3") - if _, changed := p.observe("s1", true); changed { + if _, changed := observeTestPacer(p, "s1", true); changed { t.Fatal("backed off after one server timed out") } - if _, changed := p.observe("s2", true); changed { + if _, changed := observeTestPacer(p, "s2", true); changed { t.Fatal("backed off after two servers timed out") } - adj, changed := p.observe("s3", true) - if !changed || adj.faster { + adj, changed := observeTestPacer(p, "s3", true) + if !changed || adj.recovering { t.Fatalf("adjust = %+v, changed = %t, want a backoff on the third distinct server", adj, changed) } if adj.from != 20*time.Millisecond || adj.to != 40*time.Millisecond { @@ -77,11 +115,182 @@ func TestPacerBacksOffOnCrossServerTimeouts(t *testing.T) { } } +func TestPacerCountsIndependentFailureDomainsInsteadOfEndpoints(t *testing.T) { + tests := []struct { + name string + servers []model.Server + }{ + { + name: "same operator", + servers: []model.Server{ + {ID: "provider-udp", Operator: "Example DNS", Address: "192.0.2.1", Protocol: model.ProtoUDP}, + {ID: "provider-dot", Operator: " example dns ", Address: "192.0.2.1", Protocol: model.ProtoDoT}, + {ID: "provider-doh", Operator: "EXAMPLE DNS", Address: "192.0.2.2", Protocol: model.ProtoDoH}, + {ID: "provider-doq", Operator: "Example DNS", Address: "192.0.2.2", Protocol: model.ProtoDoQ}, + }, + }, + { + name: "same address without operator metadata", + servers: []model.Server{ + {ID: "address-udp", Address: "192.0.2.10", Protocol: model.ProtoUDP}, + {ID: "address-dot", Address: "192.0.2.10", Protocol: model.ProtoDoT}, + {ID: "address-doh", Address: "192.0.2.10", Protocol: model.ProtoDoH}, + {ID: "address-doq", Address: "192.0.2.10", Protocol: model.ProtoDoQ}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 3*time.Second) + base := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + for _, server := range tt.servers { + p.observeAttemptAt(server, model.CatCached, base, base, okResult(time.Millisecond)) + } + for i, server := range tt.servers { + startedAt := base.Add(time.Second + time.Duration(i)*100*time.Millisecond) + observedAt := startedAt.Add(3 * time.Second) + if adj, changed := p.observeAttemptAt( + server, model.CatCached, startedAt, observedAt, timeoutResult(), + ); changed { + t.Fatalf("backed off for correlated endpoints in one failure domain: %+v", adj) + } + } + if p.interval != 20*time.Millisecond { + t.Fatalf("interval = %s, want unchanged 20ms", p.interval) + } + }) + } +} + +func TestPacerBacksOffOnThreeRecentlyHealthyFailureDomains(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 3*time.Second) + base := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + servers := []model.Server{ + {ID: "alpha-udp", Operator: "Alpha DNS", Address: "192.0.2.1", Protocol: model.ProtoUDP}, + {ID: "beta-doh", Operator: "Beta DNS", Address: "192.0.2.2", Protocol: model.ProtoDoH}, + {ID: "gamma-dot", Operator: "Gamma DNS", Address: "192.0.2.3", Protocol: model.ProtoDoT}, + } + for _, server := range servers { + p.observeAttemptAt(server, model.CatCached, base, base, okResult(time.Millisecond)) + } + categories := []model.Category{model.CatCached, model.CatTLD, model.CatTLD} + var adj paceAdjust + var changed bool + for i, server := range servers { + startedAt := base.Add(time.Second + time.Duration(i)*500*time.Millisecond) + observedAt := startedAt.Add(3 * time.Second) + adj, changed = p.observeAttemptAt(server, categories[i], startedAt, observedAt, timeoutResult()) + } + if !changed { + t.Fatal("three independent, recently healthy failure domains did not trigger a backoff") + } + if adj.timeouts != 3 || adj.servers != 3 || len(adj.failureDomains) != 3 { + t.Fatalf("adjustment evidence = %+v, want 3 timeouts, servers and failure domains", adj) + } + if got, want := adj.failureDomains, []string{"Alpha DNS", "Beta DNS", "Gamma DNS"}; !slices.Equal(got, want) { + t.Fatalf("failure domains = %v, want %v", got, want) + } + if got, want := adj.categories, []model.Category{model.CatCached, model.CatTLD}; !slices.Equal(got, want) { + t.Fatalf("categories = %v, want %v", got, want) + } + if got, want := adj.protocols, []model.Protocol{model.ProtoDoH, model.ProtoDoT, model.ProtoUDP}; !slices.Equal(got, want) { + t.Fatalf("protocols = %v, want %v", got, want) + } + if span := adj.evidenceEndedAt.Sub(adj.evidenceStartedAt); span != time.Second { + t.Fatalf("launch evidence span = %s, want 1s", span) + } +} + +func TestPacerIgnoresServersWithoutARecentAnswer(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 3*time.Second) + base := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + servers := []model.Server{ + {ID: "alpha", Operator: "Alpha DNS"}, + {ID: "beta", Operator: "Beta DNS"}, + {ID: "gamma", Operator: "Gamma DNS"}, + } + for _, server := range servers { + p.observeAttemptAt(server, model.CatCached, base, base, okResult(time.Millisecond)) + } + for i, server := range servers { + startedAt := base.Add(recentAnswerWindow + time.Second + time.Duration(i)*100*time.Millisecond) + if adj, changed := p.observeAttemptAt( + server, model.CatCached, startedAt, startedAt.Add(3*time.Second), timeoutResult(), + ); changed { + t.Fatalf("backed off using stale health evidence: %+v", adj) + } + } + if p.interval != 20*time.Millisecond { + t.Fatalf("interval = %s, want unchanged 20ms", p.interval) + } +} + +func TestPacerCorrelatesTimeoutsByAttemptStartTime(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 10*time.Second) + base := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + servers := []model.Server{ + {ID: "alpha", Operator: "Alpha DNS"}, + {ID: "beta", Operator: "Beta DNS"}, + {ID: "gamma", Operator: "Gamma DNS"}, + } + for _, server := range servers { + p.observeAttemptAt(server, model.CatCached, base.Add(-time.Second), base, okResult(time.Millisecond)) + } + started := []time.Time{ + base.Add(time.Second), + base.Add(1500 * time.Millisecond), + base.Add(2 * time.Second), + } + observed := []time.Time{ + base.Add(4 * time.Second), + base.Add(7 * time.Second), + base.Add(10 * time.Second), + } + var adj paceAdjust + var changed bool + for i, server := range servers { + adj, changed = p.observeAttemptAt( + server, model.CatCached, started[i], observed[i], timeoutResult(), + ) + } + if !changed { + t.Fatal("launch-correlated timeouts did not trigger when their completions were far apart") + } + if adj.evidenceEndedAt.Sub(adj.evidenceStartedAt) != time.Second { + t.Fatalf("adjustment used completion times instead of the 1s launch window: %+v", adj) + } +} + +func TestPacerDoesNotCorrelateOnlyCompletionTimes(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 10*time.Second) + base := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + servers := []model.Server{ + {ID: "alpha", Operator: "Alpha DNS"}, + {ID: "beta", Operator: "Beta DNS"}, + {ID: "gamma", Operator: "Gamma DNS"}, + } + for _, server := range servers { + p.observeAttemptAt(server, model.CatCached, base.Add(-time.Second), base, okResult(time.Millisecond)) + } + for i, server := range servers { + startedAt := base.Add(time.Second + time.Duration(i)*3*time.Second) + observedAt := base.Add(10*time.Second + time.Duration(i)*10*time.Millisecond) + if adj, changed := p.observeAttemptAt( + server, model.CatCached, startedAt, observedAt, timeoutResult(), + ); changed { + t.Fatalf("backed off for completion-correlated but launch-distant timeouts: %+v", adj) + } + } + if p.interval != 20*time.Millisecond { + t.Fatalf("interval = %s, want unchanged 20ms", p.interval) + } +} + func TestPacerIgnoresSingleServerTimeouts(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) + p := newPacer(20*time.Millisecond, true, 1, 0) markAnswered(p, "s1") for i := 0; i < 10; i++ { - if _, changed := p.observe("s1", true); changed { + if _, changed := observeTestPacer(p, "s1", true); changed { t.Fatal("backed off for timeouts concentrated on a single server") } } @@ -91,9 +300,9 @@ func TestPacerIgnoresSingleServerTimeouts(t *testing.T) { } func TestPacerIgnoresNeverAnsweredServers(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) + p := newPacer(20*time.Millisecond, true, 1, 0) for _, id := range []string{"dead1", "dead2", "dead3", "dead4"} { - if _, changed := p.observe(id, true); changed { + if _, changed := observeTestPacer(p, id, true); changed { t.Fatal("backed off for servers that never answered") } } @@ -103,12 +312,12 @@ func TestPacerIgnoresNeverAnsweredServers(t *testing.T) { } func TestPacerCooldownAndCeiling(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) + p := newPacer(20*time.Millisecond, true, 1, 0) markAnswered(p, "s1", "s2", "s3") trigger := func() bool { var changed bool for _, id := range []string{"s1", "s2", "s3"} { - _, c := p.observe(id, true) + _, c := observeTestPacer(p, id, true) changed = changed || c } return changed @@ -125,28 +334,95 @@ func TestPacerCooldownAndCeiling(t *testing.T) { p.mu.Unlock() trigger() } - if p.interval != 160*time.Millisecond { - t.Fatalf("interval = %s, want the 160ms ceiling", p.interval) + if p.interval != 80*time.Millisecond { + t.Fatalf("interval = %s, want the 80ms ceiling", p.interval) } } -func TestPacerSpeedsUpAfterCleanTrafficDownToFloor(t *testing.T) { - p := newPacer(20*time.Millisecond, 1, 0) - sawFaster := false +func TestPacerNeverRunsFasterThanConfigured(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 0) for i := 0; i < 20*p.cleanTarget; i++ { - adj, changed := p.observe("s1", false) - if changed { - if !adj.faster || adj.to >= adj.from { - t.Fatalf("adjust = %+v, want a speed-up", adj) + if adj, changed := observeTestPacer(p, "s1", false); changed { + t.Fatalf("clean traffic at the configured pace adjusted it: %+v", adj) + } + } + if p.interval != 20*time.Millisecond { + t.Fatalf("interval = %s, want configured floor 20ms", p.interval) + } +} + +func TestPacerRecoversGraduallyToConfiguredInterval(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 0) + p.interval = 80 * time.Millisecond + for _, want := range []time.Duration{ + 70 * time.Millisecond, + 60 * time.Millisecond, + 50 * time.Millisecond, + 40 * time.Millisecond, + 30 * time.Millisecond, + 20 * time.Millisecond, + } { + for i := 1; i < p.cleanTarget; i++ { + if adj, changed := observeTestPacer(p, "s1", false); changed { + t.Fatalf("recovered before %d clean answers: %+v", p.cleanTarget, adj) } - sawFaster = true + } + adj, changed := observeTestPacer(p, "s1", false) + if !changed || !adj.recovering || adj.to != want || adj.to >= adj.from { + t.Fatalf("adjust = %+v, changed = %t, want gradual recovery to %s", adj, changed, want) } } - if !sawFaster { - t.Fatal("clean traffic never sped the pacer up") + for i := 0; i < 2*p.cleanTarget; i++ { + if adj, changed := observeTestPacer(p, "s1", false); changed { + t.Fatalf("recovered below the configured pace: %+v", adj) + } } - if p.interval != 5*time.Millisecond { - t.Fatalf("interval = %s, want the 5ms floor", p.interval) + if p.interval != 20*time.Millisecond { + t.Fatalf("interval = %s, want configured floor 20ms", p.interval) + } +} + +func TestPacerRecoveryWaitsForCooldown(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 0) + p.interval = 40 * time.Millisecond + p.suppressUntil = time.Now().Add(time.Hour) + for i := 0; i < 2*p.cleanTarget; i++ { + if adj, changed := observeTestPacer(p, "s1", false); changed { + t.Fatalf("recovered during cooldown: %+v", adj) + } + } + if p.interval != 40*time.Millisecond { + t.Fatalf("interval = %s during cooldown, want 40ms", p.interval) + } + p.suppressUntil = time.Now().Add(-time.Second) + for i := 0; i < p.cleanTarget; i++ { + observeTestPacer(p, "s1", false) + } + if p.interval != 30*time.Millisecond { + t.Fatalf("interval = %s after cooldown, want first recovery step 30ms", p.interval) + } +} + +func TestPacerBackoffReturnsToNextSafetyLevel(t *testing.T) { + p := newPacer(20*time.Millisecond, true, 1, 0) + markAnswered(p, "s1", "s2", "s3") + trigger := func() paceAdjust { + var got paceAdjust + for _, id := range []string{"s1", "s2", "s3"} { + if adj, changed := observeTestPacer(p, id, true); changed { + got = adj + } + } + return got + } + p.interval = 30 * time.Millisecond + if adj := trigger(); adj.from != 30*time.Millisecond || adj.to != 40*time.Millisecond { + t.Fatalf("partial-recovery backoff = %+v, want 30ms -> 40ms", adj) + } + p.suppressUntil = time.Now().Add(-time.Second) + p.interval = 70 * time.Millisecond + if adj := trigger(); adj.from != 70*time.Millisecond || adj.to != 80*time.Millisecond { + t.Fatalf("partial-recovery backoff = %+v, want 70ms -> 80ms", adj) } } @@ -167,8 +443,14 @@ func TestEngineEmitsPaceAdjustOnCongestion(t *testing.T) { cfg.Rounds = 4 cfg.Categories = []model.Category{model.CatCached} cfg.PaceInterval = time.Millisecond + cfg.PaceAdaptive = true cfg.PerServerGap = 0 - e := NewEngine(testServers("s1", "s2", "s3", "s4"), cfg, f.factory) + servers := testServers("s1", "s2", "s3", "s4") + for i := range servers { + servers[i].Operator = "operator-" + servers[i].ID + servers[i].Address = "" + } + e := NewEngine(servers, cfg, f.factory) events, err := runEngine(t, e, context.Background()) if err != nil { t.Fatalf("Run failed: %v", err) @@ -179,10 +461,61 @@ func TestEngineEmitsPaceAdjustOnCongestion(t *testing.T) { if e.pace.interval != 2*time.Millisecond { t.Fatalf("interval = %s, want 2ms after one backoff", e.pace.interval) } + adjustments := e.PaceAdjustments() + if len(adjustments) != 1 { + t.Fatalf("recorded pace adjustments = %d, want 1", len(adjustments)) + } + if got := adjustments[0]; got.Reason != model.PaceSharedTimeoutBurst || + len(got.FailureDomains) != 3 || + !slices.Equal(got.Categories, []model.Category{model.CatCached}) || + !slices.Equal(got.Protocols, []model.Protocol{model.ProtoUDP}) { + t.Fatalf("recorded pace evidence = %+v", got) + } for _, ev := range events { - if ev.Type == model.EvPaceAdjust && !strings.Contains(ev.Msg, "local congestion") { + if ev.Type != model.EvPaceAdjust { + continue + } + if !strings.Contains(ev.Msg, "possible shared-path congestion") || + !strings.Contains(ev.Msg, "3 resolver groups timed out in 2s") { t.Fatalf("pace event message = %q", ev.Msg) } + if ev.Pace == nil || ev.Pace.Reason != model.PaceSharedTimeoutBurst { + t.Fatalf("pace event lacks structured evidence: %+v", ev) + } + } +} + +func TestEngineRecordsPaceRecoveryWithoutEmittingAWarning(t *testing.T) { + server := model.Server{ + ID: "alpha", + Operator: "Alpha DNS", + Address: "192.0.2.1", + Protocol: model.ProtoUDP, + } + cfg := testConfig() + cfg.PaceInterval = 20 * time.Millisecond + cfg.PaceAdaptive = true + e := NewEngine([]model.Server{server}, cfg, nil) + e.pace.interval = 40 * time.Millisecond + e.pace.suppressUntil = time.Now().Add(-time.Second) + for range e.pace.cleanTarget { + e.observePace(context.Background(), server, model.CatCached, time.Now(), okResult(time.Millisecond)) + } + adjustments := e.PaceAdjustments() + if len(adjustments) != 1 { + t.Fatalf("recorded pace adjustments = %d, want one recovery", len(adjustments)) + } + got := adjustments[0] + if got.Reason != model.PaceCleanAnswerRecovery || + got.FromInterval != 40*time.Millisecond || + got.ToInterval != 30*time.Millisecond || + got.CleanAnswers != e.pace.cleanTarget { + t.Fatalf("recorded recovery = %+v", got) + } + select { + case ev := <-e.Events(): + t.Fatalf("recovery emitted a live warning event: %+v", ev) + default: } } diff --git a/internal/bench/schedule.go b/internal/bench/schedule.go index 42611ea..ed72325 100644 --- a/internal/bench/schedule.go +++ b/internal/bench/schedule.go @@ -64,13 +64,12 @@ func (e *Engine) runRound(ctx context.Context, active []model.Server, round int, var mu sync.Mutex total, timeouts := 0, 0 for pos, idx := range order { - if !e.acquireSlot(ctx) { + if ctx.Err() != nil { break } wg.Add(1) go func(pos, idx int) { defer wg.Done() - defer e.releaseSlot() t, to := e.runServerRound(ctx, active[idx], idx, pos, round, warmup, seq) mu.Lock() total += t @@ -104,14 +103,10 @@ func (e *Engine) runServerRound(ctx context.Context, s model.Server, stableIdx, if !e.waitServerGap(ctx, s.ID) { return total, timeouts } - if e.gate.wait(ctx) != nil { + sample, started := e.measure(ctx, s, cat, round, warmup, qname) + if !started { return total, timeouts } - if !e.pace.wait(ctx) { - return total, timeouts - } - e.emit(ctx, model.Event{Type: model.EvQueryStart, ServerID: s.ID, Round: round}) - sample := e.measure(ctx, s, cat, round, warmup, qname) e.markServerQueryDone(s.ID) if sample.Result.Err != nil && (sample.Result.Err.Kind == model.ErrCanceled || ctx.Err() != nil) { return total, timeouts @@ -154,8 +149,8 @@ func randomLabel(rng *rand.Rand) string { return string(b) } -func (e *Engine) measure(ctx context.Context, s model.Server, cat model.Category, round int, warmup bool, qname string) model.Sample { - startedAt := time.Now() +func (e *Engine) measure(ctx context.Context, s model.Server, cat model.Category, round int, warmup bool, qname string) (model.Sample, bool) { + var startedAt time.Time sample := model.Sample{ ServerID: s.ID, Category: cat, @@ -163,7 +158,6 @@ func (e *Engine) measure(ctx context.Context, s model.Server, cat model.Category Warmup: warmup, QName: qname, QType: "A", - At: startedAt, } retries := e.cfg.Retries if retries < 0 { @@ -175,12 +169,24 @@ func (e *Engine) measure(ctx context.Context, s model.Server, cat model.Category if !sleepCtx(ctx, e.cfg.RetryInterval) { break } - if !e.pace.wait(ctx) { - break + } + var attemptStartedAt time.Time + attemptResult, executed := e.executeAttempt(ctx, func() model.QueryResult { + if startedAt.IsZero() { + e.emit(ctx, model.Event{Type: model.EvQueryStart, ServerID: s.ID, Round: round}) + attemptStartedAt = time.Now() + startedAt = attemptStartedAt + sample.At = startedAt + } else { + attemptStartedAt = time.Now() } + return e.attemptQuery(ctx, s, qname) + }) + if !executed { + break } - res = e.attemptQuery(ctx, s, qname) - e.observePace(ctx, s.ID, res) + res = attemptResult + e.observePace(ctx, s, cat, attemptStartedAt, res) sample.Attempts++ if !res.Answered() { sample.FailedAttempts++ @@ -192,7 +198,10 @@ func (e *Engine) measure(ctx context.Context, s model.Server, cat model.Category break } } + if startedAt.IsZero() { + return sample, false + } sample.Elapsed = time.Since(startedAt) sample.Result = res - return sample + return sample, true } diff --git a/internal/bench/triage.go b/internal/bench/triage.go index 9e7e120..dee4f33 100644 --- a/internal/bench/triage.go +++ b/internal/bench/triage.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "sync" + "time" "dnsbench/internal/model" ) @@ -40,9 +41,6 @@ func (e *Engine) runTriage(ctx context.Context) ([]model.Server, error) { wg.Add(1) go func(i int) { defer wg.Done() - if e.acquireSlot(ctx) { - defer e.releaseSlot() - } results[i] = e.triageServer(ctx, e.servers[i]) }(i) } @@ -73,16 +71,21 @@ func (e *Engine) triageServer(ctx context.Context, s model.Server) *model.Triage if attempts < 1 { attempts = 1 } - e.emit(ctx, model.Event{Type: model.EvQueryStart, ServerID: s.ID}) + started := false for i := 0; i < attempts; i++ { - if e.gate.wait(ctx) != nil { - break - } - if !e.pace.wait(ctx) { + var attemptStartedAt time.Time + res, executed := e.executeAttempt(ctx, func() model.QueryResult { + if !started { + e.emit(ctx, model.Event{Type: model.EvQueryStart, ServerID: s.ID}) + started = true + } + attemptStartedAt = time.Now() + return e.singleQuery(ctx, s, domain) + }) + if !executed { break } - res := e.singleQuery(ctx, s, domain) - e.observePace(ctx, s.ID, res) + e.observePace(ctx, s, model.CatCached, attemptStartedAt, res) if res.Err != nil && res.Err.Kind == model.ErrCanceled { break } diff --git a/internal/model/config_test.go b/internal/model/config_test.go index 298405d..367b4de 100644 --- a/internal/model/config_test.go +++ b/internal/model/config_test.go @@ -20,6 +20,13 @@ func TestDefaultBenchConfigUses200MillisecondTriageThreshold(t *testing.T) { } } +func TestDefaultBenchConfigUsesAdaptivePacing(t *testing.T) { + cfg := DefaultBenchConfig(ModeStandard) + if !cfg.PaceAdaptive || cfg.PaceInterval != 20*time.Millisecond { + t.Fatalf("default pacing = %s adaptive=%t, want 20ms adaptive", cfg.PaceInterval, cfg.PaceAdaptive) + } +} + func TestDefaultCachedDomainsAreCuratedTopDomains(t *testing.T) { domains := DefaultCachedDomains() if got := len(domains); got < 50 { diff --git a/internal/model/model.go b/internal/model/model.go index e7e5242..9527c79 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -10,7 +10,7 @@ import ( const ( AppName = "dnsbench" - AppVersion = "0.6.0" + AppVersion = "0.7.0" ) type Protocol string @@ -630,6 +630,7 @@ type BenchConfig struct { RetryInterval time.Duration `json:"retry_interval_ns"` Concurrency int `json:"concurrency"` PaceInterval time.Duration `json:"pace_interval_ns"` + PaceAdaptive bool `json:"pace_adaptive"` PerServerGap time.Duration `json:"per_server_gap_ns"` Session SessionMode `json:"session"` Seed int64 `json:"seed"` @@ -657,6 +658,7 @@ func DefaultBenchConfig(m Mode) BenchConfig { RetryInterval: 200 * time.Millisecond, Concurrency: 8, PaceInterval: 20 * time.Millisecond, + PaceAdaptive: true, PerServerGap: 40 * time.Millisecond, Session: SessionPersistent, TriageEnabled: true, @@ -682,13 +684,37 @@ const ( ) type Event struct { - Type EventType `json:"type"` - ServerID string `json:"server_id,omitempty"` - Round int `json:"round,omitempty"` - Sample *Sample `json:"sample,omitempty"` - Triage *TriageResult `json:"triage,omitempty"` - State ServerState `json:"state,omitempty"` - Msg string `json:"msg,omitempty"` + Type EventType `json:"type"` + ServerID string `json:"server_id,omitempty"` + Round int `json:"round,omitempty"` + Sample *Sample `json:"sample,omitempty"` + Triage *TriageResult `json:"triage,omitempty"` + Pace *PaceAdjustment `json:"pace,omitempty"` + State ServerState `json:"state,omitempty"` + Msg string `json:"msg,omitempty"` +} + +type PaceAdjustmentReason string + +const ( + PaceSharedTimeoutBurst PaceAdjustmentReason = "shared_timeout_burst" + PaceCleanAnswerRecovery PaceAdjustmentReason = "clean_answer_recovery" +) + +type PaceAdjustment struct { + At time.Time `json:"at"` + Reason PaceAdjustmentReason `json:"reason"` + FromInterval time.Duration `json:"from_interval_ns"` + ToInterval time.Duration `json:"to_interval_ns"` + EvidenceStartedAt time.Time `json:"evidence_started_at,omitzero"` + EvidenceEndedAt time.Time `json:"evidence_ended_at,omitzero"` + Window time.Duration `json:"window_ns,omitempty"` + Timeouts int `json:"timeouts,omitempty"` + ServerIDs []string `json:"server_ids,omitempty"` + FailureDomains []string `json:"failure_domains,omitempty"` + Categories []Category `json:"categories,omitempty"` + Protocols []Protocol `json:"protocols,omitempty"` + CleanAnswers int `json:"clean_answers,omitempty"` } type RunInfo struct { @@ -713,6 +739,7 @@ type RunResult struct { Stats map[string]*ServerStats `json:"stats"` Scores map[RankMode][]Score `json:"scores"` Comparisons []Comparison `json:"comparisons,omitempty"` + PaceAdjustments []PaceAdjustment `json:"pace_adjustments,omitempty"` Samples []Sample `json:"samples,omitempty"` } diff --git a/internal/report/report_test.go b/internal/report/report_test.go index e229ba7..036de3d 100644 --- a/internal/report/report_test.go +++ b/internal/report/report_test.go @@ -248,6 +248,29 @@ func TestBuildConclusionsCoverageAndComparisons(t *testing.T) { func TestExportJSONStripsRawWithoutMutating(t *testing.T) { res := testResult() + res.PaceAdjustments = []model.PaceAdjustment{ + { + At: res.Info.StartedAt.Add(time.Second), + Reason: model.PaceSharedTimeoutBurst, + FromInterval: 20 * time.Millisecond, + ToInterval: 40 * time.Millisecond, + EvidenceStartedAt: res.Info.StartedAt.Add(-time.Second), + EvidenceEndedAt: res.Info.StartedAt, + Window: 2 * time.Second, + Timeouts: 3, + ServerIDs: []string{"alpha", "beta", "gamma"}, + FailureDomains: []string{"Alpha DNS", "Beta DNS", "Gamma DNS"}, + Categories: []model.Category{model.CatCached, model.CatTLD}, + Protocols: []model.Protocol{model.ProtoUDP, model.ProtoDoH}, + }, + { + At: res.Info.StartedAt.Add(10 * time.Second), + Reason: model.PaceCleanAnswerRecovery, + FromInterval: 40 * time.Millisecond, + ToInterval: 30 * time.Millisecond, + CleanAnswers: 50, + }, + } var buf bytes.Buffer if err := ExportJSON(&buf, res, false); err != nil { t.Fatalf("ExportJSON: %v", err) @@ -259,6 +282,22 @@ func TestExportJSONStripsRawWithoutMutating(t *testing.T) { if strings.Contains(out, `"samples"`) { t.Fatalf("stripped JSON must not contain samples key") } + for _, want := range []string{ + `"pace_adjustments"`, + `"reason": "shared_timeout_burst"`, + `"failure_domains"`, + `"Alpha DNS"`, + `"protocols"`, + `"reason": "clean_answer_recovery"`, + `"clean_answers": 50`, + } { + if !strings.Contains(out, want) { + t.Fatalf("stripped JSON lost pacing evidence %q:\n%s", want, out) + } + } + if count := strings.Count(out, `"evidence_started_at"`); count != 1 { + t.Fatalf("recovery should omit its zero-value evidence timestamp, got %d keys:\n%s", count, out) + } if res.Samples == nil || len(res.Samples) != 1 { t.Fatalf("original Samples was mutated") } diff --git a/internal/serverlist/registry.go b/internal/serverlist/registry.go index 1ffdefa..8b5bd4d 100644 --- a/internal/serverlist/registry.go +++ b/internal/serverlist/registry.go @@ -23,9 +23,38 @@ func Merge(lists ...[]model.Server) []model.Server { merged = append(merged, s) } } + ensureUniqueIDs(merged) return merged } +func ensureUniqueIDs(servers []model.Server) { + reserved := make(map[string]bool, len(servers)) + for _, server := range servers { + reserved[server.ID] = true + } + used := make(map[string]bool, len(servers)) + nextSuffix := make(map[string]int) + for i := range servers { + base := servers[i].ID + if !used[base] { + used[base] = true + continue + } + suffix := max(2, nextSuffix[base]) + for { + candidate := fmt.Sprintf("%s-%d", base, suffix) + suffix++ + if used[candidate] || reserved[candidate] { + continue + } + servers[i].ID = candidate + used[candidate] = true + nextSuffix[base] = suffix + break + } + } +} + type FilterOptions struct { Protocols []model.Protocol Operators []string diff --git a/internal/serverlist/registry_test.go b/internal/serverlist/registry_test.go index 807113f..ce8a235 100644 --- a/internal/serverlist/registry_test.go +++ b/internal/serverlist/registry_test.go @@ -1,6 +1,7 @@ package serverlist import ( + "slices" "testing" "dnsbench/internal/model" @@ -40,6 +41,27 @@ func TestMergeDistinguishesProtocolAndPort(t *testing.T) { } } +func TestMergeAssignsUniqueIDsToDistinctEndpoints(t *testing.T) { + merged := Merge([]model.Server{ + udpServer("resolver", "1.1.1.1"), + {ID: "resolver", Protocol: model.ProtoUDP, Address: "1.1.1.1", Port: 5353, Enabled: true}, + udpServer("resolver-2", "9.9.9.9"), + {ID: "doh", Protocol: model.ProtoDoH, DoHURL: "https://dns.example/dns-query", Address: "192.0.2.1", Enabled: true}, + {ID: "doh", Protocol: model.ProtoDoH, DoHURL: "https://dns.example/dns-query", Address: "192.0.2.2", Enabled: true}, + }) + if len(merged) != 5 { + t.Fatalf("got %d servers, want 5", len(merged)) + } + got := make([]string, len(merged)) + for i, server := range merged { + got[i] = server.ID + } + want := []string{"resolver", "resolver-3", "resolver-2", "doh", "doh-2"} + if !slices.Equal(got, want) { + t.Fatalf("IDs = %q, want %q", got, want) + } +} + func TestFilter(t *testing.T) { servers := []model.Server{ {ID: "g4", Protocol: model.ProtoUDP, Address: "8.8.8.8", Operator: "Google Public DNS", Source: model.SourceBuiltin}, diff --git a/internal/ui/live_test.go b/internal/ui/live_test.go index 75f4626..baca408 100644 --- a/internal/ui/live_test.go +++ b/internal/ui/live_test.go @@ -451,7 +451,7 @@ func TestLivePaceNoticeWrapsWithIcon(t *testing.T) { l := NewLive(liveServers(), cfg, &bytes.Buffer{}, true, "median") l.width = 40 l.height = 16 - msg := "local congestion — pacing eased 5ms → 10ms (3 timeouts on 3 servers)" + msg := "possible shared-path congestion — pace 20ms → 40ms (3 resolver groups timed out in 2s)" l.handle(model.Event{Type: model.EvPaceAdjust, Msg: msg}) frame := l.renderFrame()