diff --git a/.claude/skills/verify-wsstat/SKILL.md b/.claude/skills/verify-wsstat/SKILL.md index 5dc7355..b6c2a8a 100644 --- a/.claude/skills/verify-wsstat/SKILL.md +++ b/.claude/skills/verify-wsstat/SKILL.md @@ -17,6 +17,15 @@ make build # -> ./bin/wsstat ./dev/run.sh up # Dockerized mock on ws://localhost:17080 and wss://localhost:17443 ``` +`run.sh up` runs in the **foreground** and blocks until Ctrl+C, which fires its +teardown trap (`docker compose down`). To drive the binary from the same session +you must background it — and then **you** own the teardown (see below): + +```bash +./dev/run.sh up & # background so you can drive; you must tear it down when done +until curl -fsS http://localhost:17080/healthz >/dev/null 2>&1; do sleep 0.5; done +``` + No Docker? Run the mock on the host instead: ```bash @@ -34,6 +43,24 @@ Pick the endpoint that isolates the changed feature (`/echo`, `/jsonrpc`, ./bin/wsstat stream -c 2 -o json -t '{"method":"subscribe","subscription":{}}' ws://localhost:17080/subscriptions ``` +## Tear down + +Always stop the stack you started — a backgrounded `run.sh up` keeps blocking and +holds ports 17080/17443 otherwise. Ctrl+C (SIGINT) is the clean teardown: it fires +run.sh's trap, which runs `docker compose down`. From a non-interactive session, +send it that signal — do **not** just `docker compose down`, which leaves the +`run.sh` process orphaned: + +```bash +pkill -INT -f 'dev/run.sh up' # Ctrl+C the mock; its trap tears the stack down +# host-run mock instead? kill by port -- `go run .` compiles to /tmp/go-build…/exe/mock-server, +# so a pkill on 'dev/mock-server' misses it: +fuser -k 17080/tcp 17443/tcp +``` + +Or skip the manual lifecycle entirely: `make smoke` / `make soak` build, run, and +tear the stack down in one shot — prefer these unless you need ad-hoc drive commands. + ## Make it stick A one-off invocation proves the change today; the suites keep proving it: diff --git a/AGENTS.md b/AGENTS.md index d1b597f..5b07e1e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ go test ./ -run 'TestWSStat/Basic' -v # focused subtest Module `github.com/jkbrsn/wsstat/v3`. Three layers: 1. **CLI** (`cmd/wsstat/`) - parses flags, validates URLs (auto-adds `wss://`), builds config, delegates to the app layer. -2. **App** (`internal/app/`) - `Client` orchestrates measurement/subscription flows and output formatting. +2. **App** (`internal/app/`) - `Client` orchestrates measurement/subscription/ping flows and output formatting. 3. **Core** (root pkg: `wsstat.go`, `measure.go`, `result.go`, `subscription.go`) - public API; wraps coder/websocket with timing instrumentation. `WSStat` produces a `Result` with DNS/TCP/TLS/WS/RTT timings; the one-shot `Measure*` free functions live in `measure.go`. All layers use the functional options pattern: `New(opts ...Option)` with `WithTimeout()`, `WithTLSConfig()`, `WithBufferSize()`, etc. diff --git a/CHANGELOG.md b/CHANGELOG.md index 62350a1..85de5c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **Repeatable `-t` in `stream` mode.** `wsstat stream` now accepts `-t/--text` multiple times and sends each message in argv order on the same connection, enabling multi-frame conversations (e.g. subscribe then unsubscribe) against a single server-side session. Sends are spaced by the new `--send-delay ` flag (default `1s`) so responses interleave predictably under `-o json`. The first message remains the subscribe payload; receiving (`-c`, `--once`, `--timeout`) is unchanged and a single `-t` behaves exactly as before; if the receive limit is reached before all messages are sent, the remaining sends are skipped. `measure` mode rejects repeated `-t`. +- **`ping` subcommand.** `wsstat ping ` dials once and sends a WebSocket ping frame every `-i/--interval`, printing a per-ping RTT line and a `ping(8)`-style `STATS` summary; a missed pong is a survivable `timeout` and the run continues, exit 1 covers total loss and dial failure. See the README's "Ping Mode" section and `wsstat ping -h` for the full flag and output contract. +- **(lib) `WithUnboundedReads` option.** Drops the read pump's per-read timeout so long-lived sessions carried only by control frames (e.g. a ping/pong monitor) are not torn down as idle. +- **(lib) `WithDiscardReads` option.** Makes the read pump drop data frames not claimed by a subscription instead of queueing them, so a session that never reads keeps processing pongs against a chatty peer. +- **(lib) `PingPongContext` method.** `PingPong` additionally bounded by a caller context, so cancellation interrupts a ping blocked on an unresponsive peer instead of waiting out the full read timeout. +- **Repeatable `-t` in `stream` mode.** `wsstat stream` now accepts `-t/--text` multiple times, sending each message in argv order on the same connection, spaced by the new `--send-delay` flag (default `1s`); a single `-t` behaves as before and `measure` rejects repeats. See the README's "Stream Mode" section for details. ## [3.1.1] - 2026-07-13 diff --git a/README.md b/README.md index b8382cf..df2a0a7 100644 --- a/README.md +++ b/README.md @@ -165,8 +165,9 @@ For a full list of the available options, run `wsstat -h`, `wsstat measure -h`, ### Modes `wsstat ` (or `wsstat measure `) measures connection latency. `wsstat -stream ` keeps the socket open for long-lived feeds. A host literally named -`measure`/`stream` must be spelled with a scheme (`wsstat wss://stream`). +stream ` keeps the socket open for long-lived feeds. `wsstat ping ` +watches ping/pong latency over time. A host literally named +`measure`/`stream`/`ping` must be spelled with a scheme (`wsstat wss://ping`). In measure mode, `-c N` aggregates timing across all interactions; the response printed is the first one (measure does not concatenate responses). @@ -214,6 +215,39 @@ wsstat stream -c 4 -o json \ wss://example.org/ws ``` +### Ping Mode + +`wsstat ping ` dials once and sends a WebSocket ping frame every +`-i/--interval` (default `1s`) on that connection, printing a per-ping RTT line +as each pong arrives and a summary at the end: + +```sh +wsstat ping -c 5 wss://echo.example.com +``` + +```text +PING wss://echo.example.com (dns 5ms, tcp 10ms, tls 12ms, ws 7ms) +pong: seq=1 rtt=12.3ms +pong: seq=2 rtt=11.8ms +timeout: seq=3 (5s) +pong: seq=4 rtt=12.1ms +... +STATS wss://echo.example.com (5 sent, 4 received, 20.0% loss) +rtt: min=11.8ms avg=12.1ms max=12.3ms stddev=0.2ms +``` + +With no `-c` the run continues until you interrupt it (`Ctrl-C`) or the optional +`-w/--deadline` elapses, like `ping(8)`. `-o json` emits one `ping_reply` record +per ping and a final `ping_summary` (which stays the last record even on total +loss); `-q` prints the summary block only. + +A missed pong (no reply within `--timeout`, default `5s`) is reported as a +`timeout` line and the run **continues**, exactly like `ping(8)` — so a transient +drop shows up as loss in the summary without ending the run. The run ends only at +`--count`, on `Ctrl-C`/`--deadline`, or when the connection actually closes. The +exit code is 0 when at least one pong was received and 1 on total loss or a dial +failure, so `wsstat ping -c 3 ` doubles as a liveness gate. + ### Output Output is split across three orthogonal axes: @@ -257,6 +291,10 @@ rejected under `-o json|raw`. | 2 | Usage error (bad flag or argument) | | 130 | Interrupted; a second `Ctrl-C` forces teardown | +In `ping` mode exit 1 also covers total loss (zero pongs received), while partial +loss with at least one pong still exits 0, so `wsstat ping -c N ` works as a +liveness gate. + Usage errors (exit 2) always print plain text to stderr. Under `-o json`, a runtime failure (exit 1) prints a `{"type":"error"}` envelope to stdout so a `wsstat ... -o json | jq` pipeline stays parseable on the failure path: diff --git a/cmd/wsstat/config.go b/cmd/wsstat/config.go index da1b54c..8bca116 100644 --- a/cmd/wsstat/config.go +++ b/cmd/wsstat/config.go @@ -58,23 +58,44 @@ type commonFlags struct { version bool } -// registerCommon registers the shared flags onto fs, binding them to c. +// newCommonFlags returns commonFlags with the resolved defaults pre-set, so a subcommand +// that skips a registration group still passes validation with the same defaults the +// flags would have carried. Registration binds flag defaults from these fields. +func newCommonFlags() commonFlags { + return commonFlags{output: "text", body: "auto", color: "auto", rpcVersion: "2.0"} +} + +// registerCommon registers all shared flag groups onto fs, binding them to c. Subcommands +// that support only a subset (ping) call the group functions individually instead. func registerCommon(fs *flag.FlagSet, c *commonFlags) { - fs.Var(&c.headers, "H", "HTTP header to include with the request (repeatable; format: \"Key: Value\")") - fs.Var(&c.headers, "header", "HTTP header to include with the request (repeatable; format: \"Key: Value\")") - fs.Var(&c.resolves, "resolve", "resolve host:port to address (repeatable; format: \"HOST:PORT:ADDRESS\")") + registerMessagingFlags(fs, c) + registerResponseFlags(fs, c) + registerOutputFlags(fs, c) + registerConnectionFlags(fs, c) + registerDiagnosticFlags(fs, c) +} +// registerMessagingFlags registers the outbound-payload flags (what gets sent). +func registerMessagingFlags(fs *flag.FlagSet, c *commonFlags) { fs.StringVar(&c.rpcMethod, "rpc-method", "", "JSON-RPC method name to send (id=1, jsonrpc=2.0)") - fs.StringVar(&c.rpcVersion, "rpc-version", "2.0", "JSON-RPC version for --rpc-method: 2.0 or 1.0") + fs.StringVar(&c.rpcVersion, "rpc-version", c.rpcVersion, "JSON-RPC version for --rpc-method: 2.0 or 1.0") fs.Var(&c.text, "t", "text message to send (repeatable in stream mode; @file or @- reads payload from a file or stdin)") fs.Var(&c.text, "text", "text message to send (repeatable in stream mode; @file or @- reads payload from a file or stdin)") +} - fs.StringVar(&c.output, "o", "text", "output contract: text, json, or raw") - fs.StringVar(&c.output, "output", "text", "output contract: text, json, or raw") +// registerResponseFlags registers the response-payload flags (recording and rendering of +// what comes back). Ping mode has no response payloads and skips this group. +func registerResponseFlags(fs *flag.FlagSet, c *commonFlags) { fs.StringVar(&c.file, "file", "", "record response payloads to PATH as NDJSON, one per line (fails if PATH exists)") - fs.StringVar(&c.body, "body", "auto", "body rendering for text output: auto or compact") + fs.StringVar(&c.body, "body", c.body, "body rendering for text output: auto or compact") fs.BoolVar(&c.clip, "clip", false, "clip each rendered line to terminal width (text output, TTY only)") +} + +// registerOutputFlags registers the output-contract and verbosity flags. +func registerOutputFlags(fs *flag.FlagSet, c *commonFlags) { + fs.StringVar(&c.output, "o", c.output, "output contract: text, json, or raw") + fs.StringVar(&c.output, "output", c.output, "output contract: text, json, or raw") fs.BoolVar(&c.showSecrets, "show-secrets", false, "show sensitive header values in -vv output instead of masking them") fs.BoolVar(&c.quiet, "q", false, "suppress all output except the response") @@ -82,10 +103,16 @@ func registerCommon(fs *flag.FlagSet, c *commonFlags) { fs.BoolVar(&c.v1, "v", false, "increase verbosity (level 1)") fs.BoolVar(&c.v1, "verbose", false, "increase verbosity (level 1)") fs.BoolVar(&c.v2, "vv", false, "increase verbosity (level 2)") + fs.StringVar(&c.color, "color", c.color, "color output: auto, always, or never") +} +// registerConnectionFlags registers the dial- and transport-level flags. +func registerConnectionFlags(fs *flag.FlagSet, c *commonFlags) { + fs.Var(&c.headers, "H", "HTTP header to include with the request (repeatable; format: \"Key: Value\")") + fs.Var(&c.headers, "header", "HTTP header to include with the request (repeatable; format: \"Key: Value\")") + fs.Var(&c.resolves, "resolve", "resolve host:port to address (repeatable; format: \"HOST:PORT:ADDRESS\")") fs.BoolVar(&c.insecure, "k", false, "skip TLS certificate verification") fs.BoolVar(&c.insecure, "insecure", false, "skip TLS certificate verification") - fs.StringVar(&c.color, "color", "auto", "color output: auto, always, or never") fs.DurationVar(&c.timeout, "timeout", 0, "read/dial timeout (e.g., 30s, 1m); 0 uses default (5s)") fs.DurationVar(&c.closeTimeout, "close-timeout", 0, "max wait for the peer's close echo before forcing teardown; 0 uses default (3s); capped at 5s") @@ -95,6 +122,10 @@ func registerCommon(fs *flag.FlagSet, c *commonFlags) { "WebSocket subprotocol(s) to negotiate, in preference order (comma-separated)") fs.BoolVar(&c.validateUTF8, "validate-utf8", false, "validate UTF-8 on inbound text frames and warn on violations (coder/websocket skips this)") +} + +// registerDiagnosticFlags registers the diagnostics flags. +func registerDiagnosticFlags(fs *flag.FlagSet, c *commonFlags) { fs.BoolVar(&c.debug, "debug", false, "emit core debug logs to stderr (independent of -v/-vv output verbosity)") fs.BoolVar(&c.version, "version", false, "print program version and exit") diff --git a/cmd/wsstat/main.go b/cmd/wsstat/main.go index e5e1efc..cc90f0d 100644 --- a/cmd/wsstat/main.go +++ b/cmd/wsstat/main.go @@ -19,6 +19,14 @@ // wsstat stream -t '{"method":"subscribe"}' wss://stream.example.com // wsstat stream --once -t '{"method":"ticker"}' wss://api.example.com // +// # Ping Mode +// +// To watch ping/pong latency over time, use the ping subcommand, which sends a WebSocket +// ping frame every interval on a single connection and prints per-ping RTT plus a summary: +// +// wsstat ping -c 5 wss://echo.example.com +// wsstat ping -i 500ms -w 30s wss://echo.example.com +// // # Architecture // // The package is organized into: @@ -63,6 +71,12 @@ var errUsageShown = errors.New("usage shown") // exits 0 with no further output. var errVersionShown = errors.New("version shown") +// errPingTotalLoss signals ping-mode total loss (zero pongs received). RunPing has already +// printed the per-ping lines and the summary, so main exits with the runtime code without +// printing an additional error line or JSON envelope, keeping ping_summary the final record. +// A dial failure (nothing printed yet) still flows through runtimeErr instead. +var errPingTotalLoss = errors.New("no pongs received") + // responseFilePerm is the mode for the --file response sink (owner read/write, group/other read). const responseFilePerm = 0o644 @@ -123,6 +137,8 @@ func main() { err = runStream(args[1:]) case args[0] == "measure": err = runMeasure(args[1:]) + case args[0] == "ping": + err = runPing(args[1:]) case args[0] == "--version" || args[0] == "-version": fmt.Printf("wsstat %s\n", version) return @@ -138,6 +154,9 @@ func main() { return case errors.Is(err, errUsageShown): os.Exit(exitUsage) + case errors.Is(err, errPingTotalLoss): + // The summary already reported the loss; exit non-zero without extra output. + os.Exit(exitRuntime) default: fail(err) } @@ -232,7 +251,7 @@ func interruptContext() (context.Context, context.CancelFunc) { // buildMeasure parses measure-mode args and returns a validated client and target. func buildMeasure(args []string) (*app.Client, *url.URL, error) { fs := flag.NewFlagSet("measure", flag.ContinueOnError) - var cf commonFlags + cf := newCommonFlags() registerCommon(fs, &cf) registerRemoved(fs) count := fs.Int("c", 1, "number of interactions to perform (>= 1)") @@ -270,7 +289,7 @@ func buildMeasure(args []string) (*app.Client, *url.URL, error) { // Whether --once was requested is available via client.Once(). func buildStream(args []string) (*app.Client, *url.URL, error) { fs := flag.NewFlagSet("stream", flag.ContinueOnError) - var cf commonFlags + cf := newCommonFlags() registerCommon(fs, &cf) registerRemoved(fs) count := fs.Int("c", 0, "number of events to receive; 0 = unlimited") @@ -334,6 +353,123 @@ func buildStream(args []string) (*app.Client, *url.URL, error) { return client, target, nil } +// unsupportedFlag describes a flag a subcommand rejects: the reason shown in the error, +// and whether the flag takes a value (its arity), so the stub registration consumes any +// value token: -t @- must not read stdin, and a following value must not be misread as +// the positional URL. +type unsupportedFlag struct { + reason string + valued bool +} + +// pingUnsupported maps the flags the ping subcommand rejects (internal flag name, +// without dashes) to their rejection details. +var pingUnsupported = map[string]unsupportedFlag{ + "t": {"ping sends ping frames, not messages", true}, + "text": {"ping sends ping frames, not messages", true}, + "rpc-method": {"ping sends ping frames, not messages", true}, + "rpc-version": {"ping sends ping frames, not messages", true}, + "file": {"ping has no response payloads to record", true}, + "body": {"ping renders no response bodies", true}, + "clip": {"ping renders no response bodies", false}, +} + +// registerUnsupported registers a subcommand's unsupported flags as inert stubs on fs, +// matching each flag's arity. Whether one was actually used is reported by +// unsupportedFlagError after Parse, keyed on the flag being set rather than its resolved +// value, so an explicitly empty value (-t ”, --file ”) is rejected too. Mirrors the +// app-layer validatePing rejections for the direct-API path. +func registerUnsupported(fs *flag.FlagSet, flags map[string]unsupportedFlag) { + for name, f := range flags { + if f.valued { + fs.String(name, "", "not supported by this subcommand") + } else { + fs.Bool(name, false, "not supported by this subcommand") + } + } +} + +// unsupportedFlagError returns a targeted error if any of the subcommand's unsupported +// flags was explicitly set on fs. Detection runs after Parse, so it sees only genuine +// flag tokens, never values that merely look like one. +func unsupportedFlagError(fs *flag.FlagSet, cmd string, flags map[string]unsupportedFlag) error { + var err error + fs.Visit(func(f *flag.Flag) { + if err != nil { + return + } + if u, ok := flags[f.Name]; ok { + dashes := "--" + if len(f.Name) == 1 { + dashes = "-" + } + err = fmt.Errorf("%s%s is not supported in %s mode: %s", dashes, f.Name, cmd, u.reason) + } + }) + return err +} + +// pingConfig bundles the validated ping-mode settings. The deadline is CLI-layer-only (it +// never reaches app.Client), so it rides alongside the client rather than inside it. +type pingConfig struct { + client *app.Client + target *url.URL + deadline time.Duration +} + +// buildPing parses ping-mode args and returns the validated ping configuration. +func buildPing(args []string) (pingConfig, error) { + fs := flag.NewFlagSet("ping", flag.ContinueOnError) + cf := newCommonFlags() + registerOutputFlags(fs, &cf) + registerConnectionFlags(fs, &cf) + registerDiagnosticFlags(fs, &cf) + registerUnsupported(fs, pingUnsupported) + registerRemoved(fs) + count := fs.Int("c", 0, "number of pings to send; 0 = until interrupted") + fs.IntVar(count, "count", 0, "number of pings to send; 0 = until interrupted") + interval := fs.Duration("i", time.Second, "delay between pings (e.g., 500ms, 2s)") + fs.DurationVar(interval, "interval", time.Second, "delay between pings (e.g., 500ms, 2s)") + deadline := fs.Duration("w", 0, "max total run time (e.g., 10s); 0 = no deadline") + fs.DurationVar(deadline, "deadline", 0, "max total run time (e.g., 10s); 0 = no deadline") + fs.Usage = func() {} // parseErr owns usage printing (stdout for -h, stderr otherwise) + + if err := fs.Parse(args); err != nil { + return pingConfig{}, parseErr(err, printPingUsage) + } + if cf.version { + fmt.Printf("wsstat %s\n", version) + return pingConfig{}, errVersionShown + } + if err := removedFlagError(fs); err != nil { + return pingConfig{}, err + } + + if err := unsupportedFlagError(fs, "ping", pingUnsupported); err != nil { + return pingConfig{}, err + } + + set := setFlagNames(fs) + opts, target, err := resolveCommon(fs, &cf, app.ModePing) + if err != nil { + return pingConfig{}, err + } + if *count < 0 { + return pingConfig{}, errors.New("count must be zero or greater") + } + // --deadline is validated here (not in Validate) since it never reaches app.Client. + if (set["w"] || set["deadline"]) && *deadline <= 0 { + return pingConfig{}, errors.New("--deadline must be greater than zero") + } + opts = append(opts, app.WithCount(*count), app.WithInterval(*interval)) + + client := app.NewClient(opts...) + if err := client.Validate(); err != nil { + return pingConfig{}, fmt.Errorf("invalid settings: %w", err) + } + return pingConfig{client: client, target: target, deadline: *deadline}, nil +} + // parseErr maps a FlagSet parse error to the appropriate sentinel and prints the // command usage: to stdout for explicitly requested help (GNU convention, and // matching top-level -h), to stderr after a genuine parse error (whose message @@ -445,3 +581,34 @@ func runStream(args []string) error { } return runtimeErr(out, client.StreamSubscription(ctx, target)) } + +// runPing runs ping mode. Unlike runMeasure/runStream it does not funnel every returned error +// through runtimeErr: RunPing swallows context cancellation (Ctrl-C, --deadline) and connection +// loss, returning the report with a nil error, so those never become exit 1. runPing derives +// the exit code from the report instead: zero pongs received (total loss) is exit 1, making +// `wsstat ping -c N ` a usable liveness gate; any pong received is exit 0. Only dial and +// output-write failures flow through the error return. +func runPing(args []string) error { + cfg, err := buildPing(args) + if err != nil { + return usageErr(err) + } + + ctx, cancel := interruptContext() + defer cancel() + if cfg.deadline > 0 { + var deadlineCancel context.CancelFunc + ctx, deadlineCancel = context.WithTimeout(ctx, cfg.deadline) + defer deadlineCancel() + } + + out := cfg.client.Output() + report, err := cfg.client.RunPing(ctx, cfg.target) + if err != nil { + return runtimeErr(out, err) + } + if report.Received == 0 { + return errPingTotalLoss + } + return nil +} diff --git a/cmd/wsstat/main_test.go b/cmd/wsstat/main_test.go index 9f5bd4e..ba8abae 100644 --- a/cmd/wsstat/main_test.go +++ b/cmd/wsstat/main_test.go @@ -3,8 +3,15 @@ package main import ( "errors" "flag" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" "testing" + "time" + "github.com/coder/websocket" "github.com/jkbrsn/wsstat/v3/internal/app" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -63,6 +70,20 @@ func TestRemovedFlagsRejected(t *testing.T) { } } +// TestMeasureStreamRejectPingFlags pins that ping-only flags stay unknown outside ping +// mode: a refactor that re-broadens flag registration must not let -w/--deadline parse in +// measure or stream. +func TestMeasureStreamRejectPingFlags(t *testing.T) { + t.Parallel() + + for _, flagArg := range []string{"-w", "--deadline"} { + _, _, err := buildMeasure([]string{flagArg, "10s", "wss://example.com"}) + assert.ErrorIs(t, err, errUsageShown, "measure must reject %s", flagArg) + _, _, err = buildStream([]string{flagArg, "10s", "wss://example.com"}) + assert.ErrorIs(t, err, errUsageShown, "stream must reject %s", flagArg) + } +} + // TestErrorClassification verifies the exit-code contract: flag-parse sentinels pass // through untouched, post-parse validation maps to exit 2, and runtime failures map to // exit 1 carrying the output contract for the JSON envelope. @@ -127,3 +148,153 @@ func TestDispatchRouting(t *testing.T) { assert.True(t, client.Once()) }) } + +// pingTestServer starts a WebSocket server and returns its ws:// URL. When answer is true it +// reads in the background, so coder auto-answers pings; when false it never reads, so pings go +// unanswered and time out client-side. Torn down via t.Cleanup. +// +//revive:disable-next-line:flag-parameter test-server behavior toggle +func pingTestServer(t *testing.T, answer bool) string { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true}) + if err != nil { + return + } + defer func() { _ = conn.CloseNow() }() + if answer { + ctx := conn.CloseRead(r.Context()) + <-ctx.Done() + return + } + <-r.Context().Done() + })) + t.Cleanup(srv.Close) + return "ws" + strings.TrimPrefix(srv.URL, "http") +} + +// runPingExitCode runs runPing with stdout suppressed and maps its error to a process exit code. +func runPingExitCode(t *testing.T, args []string) int { + t.Helper() + old := os.Stdout + r, w, err := os.Pipe() + require.NoError(t, err) + os.Stdout = w + runErr := runPing(args) + require.NoError(t, w.Close()) + os.Stdout = old + _, _ = io.Copy(io.Discard, r) + _ = r.Close() + + if runErr == nil { + return 0 + } + if errors.Is(runErr, errPingTotalLoss) { + return exitRuntime // main exits exitRuntime on this sentinel without extra output + } + var ce *cliError + if errors.As(runErr, &ce) { + return ce.code + } + return exitRuntime +} + +// TestRunPingExitCodes exercises the runtime exit-code contract end-to-end against a live +// WebSocket server: a pong yields exit 0, total loss yields exit 1, and a deadline terminates +// the run on its own with exit 0. +func TestRunPingExitCodes(t *testing.T) { + t.Run("count reached exits 0", func(t *testing.T) { + url := pingTestServer(t, true) + assert.Equal(t, 0, runPingExitCode(t, []string{"-c", "2", "-i", "20ms", url})) + }) + + t.Run("total loss exits 1", func(t *testing.T) { + url := pingTestServer(t, false) + code := runPingExitCode(t, []string{ + "-c", "2", "-i", "20ms", "--timeout", "150ms", "--close-timeout", "150ms", url, + }) + assert.Equal(t, exitRuntime, code) + }) + + t.Run("deadline self-terminates exit 0", func(t *testing.T) { + url := pingTestServer(t, true) + assert.Equal(t, 0, runPingExitCode(t, []string{"-w", "300ms", "-i", "100ms", url})) + }) +} + +// TestBuildPingUsageErrors verifies ping-mode usage rejections map to exit 2 and a valid +// invocation parses cleanly. +func TestBuildPingUsageErrors(t *testing.T) { + t.Parallel() + + usageCases := []struct { + name string + args []string + }{ + {"text rejected", []string{"-t", "hi", "wss://example.com"}}, + {"empty text rejected", []string{"-t", "", "wss://example.com"}}, + {"rpc-method rejected", []string{"--rpc-method", "eth_x", "wss://example.com"}}, + {"empty rpc-method rejected", []string{"--rpc-method", "", "wss://example.com"}}, + {"rpc-version rejected", []string{"--rpc-version", "1.0", "wss://example.com"}}, + {"file rejected", []string{"--file", "cap.ndjson", "wss://example.com"}}, + {"body rejected", []string{"--body", "compact", "wss://example.com"}}, + {"clip rejected", []string{"--clip", "wss://example.com"}}, + {"raw output rejected", []string{"-o", "raw", "wss://example.com"}}, + {"zero deadline rejected", []string{"-w", "0s", "wss://example.com"}}, + {"interval below floor rejected", []string{"-i", "1ms", "wss://example.com"}}, + } + for _, tc := range usageCases { + t.Run(tc.name, func(t *testing.T) { + _, err := buildPing(tc.args) + var ce *cliError + require.ErrorAs(t, usageErr(err), &ce) + assert.Equal(t, exitUsage, ce.code) + }) + } + + t.Run("valid ping parses", func(t *testing.T) { + cfg, err := buildPing([]string{"-c", "3", "-i", "500ms", "wss://example.com"}) + require.NoError(t, err) + assert.Equal(t, "wss://example.com", cfg.target.String()) + assert.Equal(t, 3, cfg.client.Count()) + assert.Equal(t, 500*time.Millisecond, cfg.client.Interval()) + assert.Equal(t, time.Duration(0), cfg.deadline) + }) + + t.Run("unsupported flag error names the mode", func(t *testing.T) { + _, err := buildPing([]string{"-t", "hi", "wss://example.com"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "-t is not supported in ping mode") + }) + + t.Run("unsupported stub consumes its value", func(t *testing.T) { + // The value token must be swallowed by the stub, not misread as the URL: + // the error is the targeted rejection, not a positional-argument error. + _, err := buildPing([]string{"--file", "wss://other.example.com", "wss://example.com"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "--file is not supported in ping mode") + }) +} + +// TestPingUsageOmitsUnsupportedFlags pins the registration/usage invariant: every flag +// ping rejects is both stub-registered (so buildPing errors) and absent from its help. +func TestPingUsageOmitsUnsupportedFlags(t *testing.T) { + t.Parallel() + + var buf strings.Builder + printPingUsage(&buf) + help := buf.String() + + for name, u := range pingUnsupported { + if len(name) > 1 { + assert.NotContains(t, help, "--"+name, "ping help must not advertise --%s", name) + } + args := []string{"-" + name, "wss://example.com"} + if u.valued { + args = []string{"-" + name, "x", "wss://example.com"} + } + _, err := buildPing(args) + require.Error(t, err, "-%s must be rejected in ping mode", name) + assert.Contains(t, err.Error(), "not supported in ping mode") + } +} diff --git a/cmd/wsstat/usage.go b/cmd/wsstat/usage.go index 6ddc62a..50a44e5 100644 --- a/cmd/wsstat/usage.go +++ b/cmd/wsstat/usage.go @@ -18,6 +18,9 @@ func printHelpFor(rest []string, w io.Writer) { case "stream": printStreamUsage(w) return + case "ping": + printPingUsage(w) + return } } printTopUsage(w) @@ -32,10 +35,12 @@ func printTopUsage(w io.Writer) { fmt.Fprintln(w, " wsstat measure connection latency (bare form)") fmt.Fprintln(w, " wsstat measure [options] measure connection latency") fmt.Fprintln(w, " wsstat stream [options] stream subscription events") + fmt.Fprintln(w, " wsstat ping [options] ping/pong latency over time") fmt.Fprintln(w) fmt.Fprintln(w, "COMMANDS:") fmt.Fprintln(w, " measure send ping/text/JSON-RPC and report timing (DNS, TCP, TLS, WS, RTT)") fmt.Fprintln(w, " stream keep the connection open and forward incoming frames") + fmt.Fprintln(w, " ping send a ping frame every interval and report per-ping RTT + a summary") fmt.Fprintln(w) fmt.Fprintln(w, " --version print program version and exit") fmt.Fprintln(w, " -h, --help show this help") @@ -48,7 +53,7 @@ func printTopUsage(w io.Writer) { fmt.Fprintln(w) fmt.Fprintln(w, " Under -o json a runtime failure (exit 1) prints a {\"type\":\"error\"} envelope to stdout.") fmt.Fprintln(w) - fmt.Fprintln(w, "Run 'wsstat measure -h' or 'wsstat stream -h' for command-specific flags.") + fmt.Fprintln(w, "Run 'wsstat measure -h', 'wsstat stream -h', or 'wsstat ping -h' for command-specific flags.") fmt.Fprintln(w) fmt.Fprintln(w, "Examples:") fmt.Fprintln(w, " wsstat wss://echo.example.com") @@ -56,32 +61,73 @@ func printTopUsage(w io.Writer) { fmt.Fprintln(w, " wsstat measure --rpc-method eth_blockNumber wss://rpc.example.com/ws") fmt.Fprintln(w, " wsstat stream --summary-interval 5s wss://stream.example.com/feed") fmt.Fprintln(w, " wsstat stream --once -o json wss://api.example.com/ws") + fmt.Fprintln(w, " wsstat ping -c 5 wss://echo.example.com") } -// printCommonFlags prints the flags shared by every subcommand. -func printCommonFlags(w io.Writer) { +// The -t usage line differs per mode: stream sends each -t in order on the open +// connection; measure accepts exactly one. +const ( + textDescSingle = "text message to send (@file or @- reads file/stdin)" + textDescRepeatable = "text message to send (repeatable; @file or @- reads file/stdin)" +) + +// printInputFlags prints the outbound-payload section with the mode's -t description. +func printInputFlags(w io.Writer, textDesc string) { fmt.Fprintln(w, "Input (choose one):") fmt.Fprintln(w, " --rpc-method JSON-RPC method name to send (id=1, jsonrpc=2.0)") fmt.Fprintln(w, " --rpc-version JSON-RPC version for --rpc-method: 2.0 or 1.0 [default: 2.0]") - fmt.Fprintln(w, " -t, --text text message to send (@file or @- reads file/stdin; repeatable in stream mode)") + fmt.Fprintln(w, " -t, --text "+textDesc) fmt.Fprintln(w) +} + +// printOutputFlags prints the output section for mode ("measure", "stream", "ping"). +// Ping has no response payloads, so --file/--body/--clip and their notes are omitted; +// the stream-frame and rpc-decode raw notes appear only where they apply. +func printOutputFlags(w io.Writer, mode string) { + ping := mode == "ping" + fmt.Fprintln(w, "Output:") - fmt.Fprintln(w, " -o, --output output contract: text, json, raw [default: text]") - fmt.Fprintln(w, " --file also record response payloads to PATH as NDJSON (fails if PATH exists)") - fmt.Fprintln(w, " --body text body rendering: auto, compact [default: auto]") - fmt.Fprintln(w, " --clip clip each rendered line to terminal width (TTY only)") - fmt.Fprintln(w, " -q, --quiet suppress all output except the response") + if ping { + // Ping has no response payloads, so raw is rejected by validation and hidden here. + fmt.Fprintln(w, " -o, --output output contract: text or json [default: text]") + } else { + fmt.Fprintln(w, " -o, --output output contract: text, json, raw [default: text]") + } + if !ping { + fmt.Fprintln(w, " --file also record response payloads to PATH as NDJSON (fails if PATH exists)") + fmt.Fprintln(w, " --body text body rendering: auto, compact [default: auto]") + fmt.Fprintln(w, " --clip clip each rendered line to terminal width (TTY only)") + } + if ping { + fmt.Fprintln(w, " -q, --quiet suppress per-ping lines; print only the summary") + } else { + fmt.Fprintln(w, " -q, --quiet suppress all output except the response") + } fmt.Fprintln(w, " -v, --verbose increase verbosity (level 1)") fmt.Fprintln(w, " -vv increase verbosity (level 2)") fmt.Fprintln(w, " --show-secrets show sensitive header values in -vv (masked by default)") fmt.Fprintln(w, " --color color output: auto, always, never [default: auto]") fmt.Fprintln(w) - fmt.Fprintln(w, " Note: --body, --clip, --show-secrets, -q, -v, -vv apply only to -o text; -o json is schema-stable.") + if ping { + fmt.Fprintln(w, " Note: --show-secrets, -q, -v, -vv apply only to -o text; -o json is schema-stable.") + } else { + fmt.Fprintln(w, " Note: --body, --clip, --show-secrets, -q, -v, -vv apply only to -o text; -o json is schema-stable.") + } fmt.Fprintln(w, " NO_COLOR (any value) in the environment forces color off under --color auto.") - fmt.Fprintln(w, " -o raw with --rpc-method emits compact JSON (the frame is decoded before output).") - fmt.Fprintln(w, " -o raw concatenates stream frames undelimited (binary-safe); use -o json for delimited streaming.") - fmt.Fprintln(w, " --file is additive and orthogonal to -o: it records response bodies only; summaries and chrome still go to stdout.") + if !ping { + fmt.Fprintln(w, " -o raw with --rpc-method emits compact JSON (the frame is decoded before output).") + } + if mode == "stream" { + fmt.Fprintln(w, " -o raw concatenates stream frames undelimited (binary-safe); use -o json for delimited streaming.") + } + if !ping { + fmt.Fprintln(w, " --file is additive and orthogonal to -o: it records response bodies only; summaries and chrome still go to stdout.") + } fmt.Fprintln(w) +} + +// printConnectionFlags prints the dial- and transport-level section (identical everywhere). +func printConnectionFlags(w io.Writer) { fmt.Fprintln(w, "Connection:") fmt.Fprintln(w, " -H, --header HTTP header to include (repeatable; \"Key: Value\")") fmt.Fprintln(w, " --resolve resolve host:port to address (repeatable; \"HOST:PORT:ADDRESS\")") @@ -92,6 +138,10 @@ func printCommonFlags(w io.Writer) { fmt.Fprintln(w, " --subprotocol WebSocket subprotocol(s) to negotiate (comma-separated)") fmt.Fprintln(w, " --validate-utf8 validate UTF-8 on inbound text frames; warn on violations") fmt.Fprintln(w) +} + +// printDiagnosticFlags prints the diagnostics section (identical everywhere). +func printDiagnosticFlags(w io.Writer) { fmt.Fprintln(w, "Diagnostics:") fmt.Fprintln(w, " --debug emit core debug logs to stderr (independent of -v/-vv)") fmt.Fprintln(w, " --version print program version and exit") @@ -109,8 +159,12 @@ func printMeasureUsage(w io.Writer) { fmt.Fprintln(w, " -c, --count number of interactions to perform [default: 1; >= 1]") fmt.Fprintln(w) fmt.Fprintln(w, " Note: timing is aggregated across all interactions; the response shown is the first.") + fmt.Fprintln(w, " Repeated -t requires the stream subcommand; measure sends a single message.") fmt.Fprintln(w) - printCommonFlags(w) + printInputFlags(w, textDescSingle) + printOutputFlags(w, "measure") + printConnectionFlags(w) + printDiagnosticFlags(w) fmt.Fprintln(w) fmt.Fprintln(w, "Verbosity Levels (text output):") fmt.Fprintln(w, " (default) minimal request info with summary timings") @@ -136,5 +190,34 @@ func printStreamUsage(w io.Writer) { fmt.Fprintln(w, " Note: -t may be repeated; each message is sent in order on the same connection, --send-delay apart.") fmt.Fprintln(w, " If the receive limit (-c, --once) is reached first, remaining sends are skipped.") fmt.Fprintln(w) - printCommonFlags(w) + printInputFlags(w, textDescRepeatable) + printOutputFlags(w, "stream") + printConnectionFlags(w) + printDiagnosticFlags(w) +} + +// printPingUsage prints usage for the ping subcommand. +func printPingUsage(w io.Writer) { + fmt.Fprintln(w, "wsstat ping - continuous WebSocket ping/pong latency monitoring") + fmt.Fprintln(w) + fmt.Fprintln(w, "USAGE:") + fmt.Fprintln(w, " wsstat ping [options] ") + fmt.Fprintln(w) + fmt.Fprintln(w, "Ping:") + fmt.Fprintln(w, " -c, --count number of pings to send [default: 0 = until interrupted]") + fmt.Fprintln(w, " -i, --interval delay between pings, e.g. 500ms, 2s [default: 1s; min 10ms]") + fmt.Fprintln(w, " -w, --deadline max total run time, e.g. 10s [default: 0 = none]") + fmt.Fprintln(w) + fmt.Fprintln(w, " Dials once and sends a ping frame every interval on that connection, printing a") + fmt.Fprintln(w, " per-ping RTT line and, at the end, a summary (sent/received/loss and rtt") + fmt.Fprintln(w, " min/avg/max/stddev). A missed pong (no reply within --timeout, default 5s) is") + fmt.Fprintln(w, " reported and the run continues, like ping(8); the run ends at --count, on Ctrl-C or") + fmt.Fprintln(w, " --deadline, or when the connection closes.") + fmt.Fprintln(w) + fmt.Fprintln(w, " Exit 0 if at least one pong was received (partial loss included); exit 1 on total") + fmt.Fprintln(w, " loss or dial failure, so `wsstat ping -c 3 ` works as a liveness gate.") + fmt.Fprintln(w) + printOutputFlags(w, "ping") + printConnectionFlags(w) + printDiagnosticFlags(w) } diff --git a/dev/smoke-test.sh b/dev/smoke-test.sh index 2e41a05..44b030a 100755 --- a/dev/smoke-test.sh +++ b/dev/smoke-test.sh @@ -53,6 +53,26 @@ check_teardown_bound() { [[ "$ms" -lt 2500 ]] } +# check_ping_json asserts `ping -c 2 -o json` emits exactly two ping_reply records and one +# ping_summary record. coder answers ping frames below the handler, so both pings pong. +check_ping_json() { + "$WSSTAT" ping -c 2 -i 50ms -o json "$WS_URL/echo" 2>/dev/null | jq -es ' + ([.[] | select(.type == "ping_reply")] | length == 2) and + ([.[] | select(.type == "ping_summary")] | length == 1) + ' >/dev/null +} + +# check_ping_total_loss drives ping against /push, a write-only peer that never reads and so +# never answers a ping. The connection stays fed by its pushed frames, so the ping times out +# on its own deadline (a real no-response), the run ends with zero pongs, and the exit code is +# exactly 1 -- the CLI-surface liveness-gate contract. --close-timeout bounds the teardown +# against the non-echoing peer so the case stays fast. +check_ping_total_loss() { + "$WSSTAT" ping -c 1 -i 300ms --timeout 500ms --close-timeout 300ms \ + "$WS_URL/push?rate=20" >/dev/null 2>&1 + [[ $? -eq 1 ]] +} + # check_multi_send drives a two-frame conversation on one connection: subscribe, # then a duplicate subscribe --send-delay later. The second reply must be the # server's "Already subscribed" rejection, which is only reachable when both @@ -153,6 +173,19 @@ check "stream raw clean" check_stream_raw_clean check "summary-interval" check_summary_interval check "repeated -t conversation" check_multi_send +# --- Ping subcommand -------------------------------------------------------- +# /echo pongs every ping (coder answers below the handler); /push never reads, so +# it never pongs and drives the total-loss (exit 1) path end to end. +check "ping bounded" "$WSSTAT" ping -c 3 -i 100ms "$WS_URL/echo" +if [[ $HAVE_JQ -eq 1 ]]; then + check "ping json" check_ping_json +else + skip "ping json" "jq not installed" +fi +check "ping total loss" check_ping_total_loss +# Usage rejection: ping mode has no payload, so -t is an argument error (exit 2). +check "ping rejects -t" bash -c "! $WSSTAT ping -t hi $WS_URL/echo" + # --- Failure & edge paths --------------------------------------------------- check "timeout trips" bash -c "! $WSSTAT -timeout 1s -t hi $WS_URL/slow" check "large frame" bash -c "$WSSTAT -o raw --rpc-method ws_large $WS_URL/large | wc -c | awk '{exit (\$1 > 32768) ? 0 : 1}'" diff --git a/dev/soak-test.sh b/dev/soak-test.sh index dcf06ed..8e734fe 100755 --- a/dev/soak-test.sh +++ b/dev/soak-test.sh @@ -110,6 +110,24 @@ multi_send_ordered() { sed -n 3p "$OUTF" | grep -q '"method":"unsubscribe"' } json_stream_method() { "$B" stream -o json -t s -c 3 "$WS_URL/stream?rate=10" 2>/dev/null | jq -es 'all(.[]; .schema_version != null and .type != null) and any(.[]; .type == "subscription_message" and .payload.method == "subscription")' >/dev/null; } +recv_limit_skips_sends() { + # Receive limit reached before all sends: the remaining -t frames are skipped + # and the run still exits 0 with exactly one reply received (the stream ends + # with a subscription_summary record either way, so count message records). + "$B" stream -c 1 -o json --send-delay 200ms \ + -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" >"$OUTF" 2>/dev/null || return 1 + [[ $(grep -c '"type":"subscription_message"' "$OUTF") -eq 1 ]] && + ! grep -q '"method":"unsubscribe"' "$OUTF" +} +send_delay_staggers() { + # --send-delay's only failure mode is being ignored (an undelayed conversation + # still arrives in order): two sends 600ms apart put a floor on the wall clock. + local t0 t1; t0=$(date +%s%N) + "$B" stream -c 2 --send-delay 600ms \ + -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" >/dev/null 2>&1 || return 1 + t1=$(date +%s%N) + [[ $(( (t1 - t0) / 1000000 )) -ge 550 ]] +} # _pty_maxwidth COLS -- CMD... -> widest output line in columns (ANSI/CR stripped) _pty_maxwidth() { @@ -190,6 +208,36 @@ file_rejects_existing() { grep -Eq "response file|file exists" "$ERRF" } +# --- ping-mode predicates ---------------------------------------------------- +ping_json_shape() { + # Exactly one ping_reply per ping, and ping_summary is the final record. + "$B" ping -c 2 -i 50ms -o json "$WS_URL/echo" 2>/dev/null | jq -es ' + ([.[] | select(.type == "ping_reply")] | length == 2) and + (last | .type == "ping_summary" and .loss_pct == 0)' >/dev/null +} +ping_deadline_ends() { + # No -c: only --deadline bounds the run, so this is the row that catches a + # broken -w. timeout(1) is the hang backstop that keeps the harness moving. + timeout 5 "$B" ping -w 500ms -i 100ms "$WS_URL/echo" >"$OUTF" 2>/dev/null || return 1 + grep -q "STATS" "$OUTF" +} +ping_timeout_survivable() { + # A missed pong is survivable: the run outlives the first timeout and keeps + # pinging until --count, so both pings time out instead of the first one + # killing the connection. + timeout 15 "$B" ping -c 2 -i 100ms --timeout 400ms --close-timeout 300ms \ + "$WS_URL/push?rate=20" >"$OUTF" 2>/dev/null && return 1 + [[ $(grep -c "timeout: seq=" "$OUTF") -eq 2 ]] && grep -q "2 sent, 0 received" "$OUTF" +} +ping_total_loss() { + # /push never reads, so it never pongs: the single ping times out on its own + # deadline, the run exits 1 (the liveness-gate contract), and the summary + # reports total loss. --close-timeout bounds the non-echoing teardown. + timeout 10 "$B" ping -c 1 -i 100ms --timeout 500ms --close-timeout 300ms \ + "$WS_URL/push?rate=20" >"$OUTF" 2>/dev/null && return 1 + grep -q "1 sent, 0 received, 100.0% loss" "$OUTF" +} + echo "wsstat soak (structured matrix) against $WS_URL" echo "binary: $B" @@ -250,6 +298,18 @@ ok "stream repeated -t" -- "$B" stream -c 2 --send-delay 100ms -t "$S ok "stream repeated --text (alias)" -- "$B" stream -c 2 --send-delay 100ms --text "$SUB_BTC" --text "$UNSUB_BTC" "$WS_URL/subscriptions" ok "stream repeated -t default delay" -- "$B" stream -c 2 -t "$SUB_BTC" -t "$UNSUB_BTC" "$WS_URL/subscriptions" +# =========================================================================== +section "POSITIVE: ping-mode flags (each flag, both aliases)" +ok "ping -c bounded" -- "$B" ping -c 3 -i 50ms "$WS_URL/echo" +ok "ping --count (alias)" -- "$B" ping --count 3 -i 50ms "$WS_URL/echo" +ok "ping --interval (alias)" -- "$B" ping -c 2 --interval 50ms "$WS_URL/echo" +ok "ping -w deadline" -- timeout 5 "$B" ping -w 400ms -i 100ms "$WS_URL/echo" +ok "ping --deadline (alias)" -- timeout 5 "$B" ping --deadline 400ms -i 100ms "$WS_URL/echo" +ok "ping -o json" -- "$B" ping -c 2 -i 50ms -o json "$WS_URL/echo" +ok "ping -q" -- "$B" ping -c 2 -i 50ms -q "$WS_URL/echo" +ok "ping -v" -- "$B" ping -c 2 -i 50ms -v "$WS_URL/echo" +ok "ping wss + -k" -- "$B" ping -c 2 -i 50ms -k "$WSS_URL/echo" + # =========================================================================== section "REJECT: mutually exclusive / invalid-value rules" reject "-q + -v" "cannot be combined" -- "$B" -q -v -t hi "$WS_URL/echo" @@ -284,6 +344,26 @@ reject "stream --send-delay no -t" "no effect without repeated -t" -- "$B" s reject "stream --send-delay negative" "zero or greater" -- "$B" stream --send-delay -1s -t a -t b "$WS_URL/stream" reject "measure repeated -t" "stream subcommand" -- "$B" -t a -t b "$WS_URL/echo" +section "REJECT: ping-mode rules" +# Flags without meaning in ping mode are stub-registered and rejected outright; +# a row per flag guards against a regression back to silent accept. +reject "ping + -t" "not supported in ping mode" -- "$B" ping -t hi "$WS_URL/echo" +reject "ping + --text" "not supported in ping mode" -- "$B" ping --text hi "$WS_URL/echo" +reject "ping + --rpc-method" "not supported in ping mode" -- "$B" ping --rpc-method m "$WS_URL/echo" +reject "ping + --rpc-version" "not supported in ping mode" -- "$B" ping --rpc-version 1.0 "$WS_URL/echo" +reject "ping + --file" "not supported in ping mode" -- "$B" ping --file rec.ndjson "$WS_URL/echo" +reject "ping + --body" "not supported in ping mode" -- "$B" ping --body compact "$WS_URL/echo" +reject "ping + --clip" "not supported in ping mode" -- "$B" ping --clip "$WS_URL/echo" +reject "ping + -o raw" "no meaning in ping mode" -- "$B" ping -o raw "$WS_URL/echo" +reject "ping -c negative" "zero or greater" -- "$B" ping -c -1 "$WS_URL/echo" +reject "ping -w zero" "greater than zero" -- "$B" ping -w 0s "$WS_URL/echo" +reject "ping -i below floor" "at least 10ms" -- "$B" ping -i 1ms "$WS_URL/echo" +# Stream-only flags under ping: the flag package must reject them outright. +reject "ping + --once" "not defined" -- "$B" ping --once "$WS_URL/echo" +reject "ping + --buffer" "not defined" -- "$B" ping --buffer 8 "$WS_URL/echo" +reject "ping + --summary-interval" "not defined" -- "$B" ping --summary-interval 1s "$WS_URL/echo" +reject "ping + --send-delay" "not defined" -- "$B" ping --send-delay 1s "$WS_URL/echo" + section "REJECT: v2 flags removed in v3" reject "-subscribe" "removed in v3" -- "$B" -subscribe -t hi "$WS_URL/stream" reject "-s" "removed in v3" -- "$B" -s -t hi "$WS_URL/stream" @@ -324,6 +404,10 @@ pred "stream -o raw is header/summary-free" stream_raw_clean pred "stream --summary-interval fires periodically" summary_interval_fires # Repeated -t: three staggered frames on one connection, replies in send order. pred "stream repeated -t conversation is ordered" multi_send_ordered +# Repeated -t edge: the receive limit ends the run before the second send fires. +pred "stream -c 1 skips the remaining sends" recv_limit_skips_sends +# --send-delay observably staggers the sends rather than being dropped. +pred "stream --send-delay staggers sends" send_delay_staggers section "EFFECT: -o json is valid, schema-stable JSON" if [[ $HAVE_JQ -eq 1 ]]; then @@ -333,6 +417,22 @@ else S "-o json structural checks" "jq not installed" fi +# =========================================================================== +section "EFFECT: ping output" +outhas "ping prints pong lines" 'pong: seq=1 rtt=' -- "$B" ping -c 2 -i 50ms "$WS_URL/echo" +outhas "ping STATS summary" 'STATS .*2 sent, 2 received, 0.0% loss' -- "$B" ping -c 2 -i 50ms "$WS_URL/echo" +outhas "ping rtt stats line" 'rtt: min=.*ms avg=.*ms max=.*ms stddev=.*ms' -- "$B" ping -c 2 -i 50ms "$WS_URL/echo" +outlacks "ping -q drops pong lines" 'pong: seq=' -- "$B" ping -c 2 -i 50ms -q "$WS_URL/echo" +outhas "ping -q keeps the summary" 'STATS ' -- "$B" ping -c 2 -i 50ms -q "$WS_URL/echo" +pred "ping -w ends an unbounded run" ping_deadline_ends +pred "ping missed pong is survivable (2 timeouts)" ping_timeout_survivable +pred "ping total loss: exit 1 and 100% loss summary" ping_total_loss +if [[ $HAVE_JQ -eq 1 ]]; then + pred "ping -o json shape (replies + final summary)" ping_json_shape +else + S "ping -o json shape" "jq not installed" +fi + # =========================================================================== section "EFFECT: --file records response payloads as NDJSON" # --file is additive and orthogonal to -o; it has no validation rules beyond diff --git a/docs/schema/wsstat-output-v1.schema.json b/docs/schema/wsstat-output-v1.schema.json index 1aca8c7..621810f 100644 --- a/docs/schema/wsstat-output-v1.schema.json +++ b/docs/schema/wsstat-output-v1.schema.json @@ -9,6 +9,8 @@ { "$ref": "#/$defs/response" }, { "$ref": "#/$defs/subscription_summary" }, { "$ref": "#/$defs/subscription_message" }, + { "$ref": "#/$defs/ping_reply" }, + { "$ref": "#/$defs/ping_summary" }, { "$ref": "#/$defs/error" } ], "$defs": { @@ -141,6 +143,36 @@ } } }, + "ping_reply": { + "type": "object", + "description": "One record per ping in ping mode. A pong carries rtt_ms; a lost ping sets lost=true and error (rtt_ms omitted). A timeout loss is survivable and the run continues; only a connection loss ends the run.", + "required": ["schema_version", "type", "seq"], + "properties": { + "schema_version": { "$ref": "#/$defs/schemaVersion" }, + "type": { "const": "ping_reply" }, + "seq": { "type": "integer", "minimum": 1 }, + "rtt_ms": { "$ref": "#/$defs/ms" }, + "lost": { "type": "boolean" }, + "error": { "type": "string" } + } + }, + "ping_summary": { + "type": "object", + "description": "Final record in ping mode: counts, loss, and RTT aggregates over received pongs. The min/avg/max/stddev fields are omitted when no pong was received.", + "required": ["schema_version", "type", "url", "sent", "received", "loss_pct"], + "properties": { + "schema_version": { "$ref": "#/$defs/schemaVersion" }, + "type": { "const": "ping_summary" }, + "url": { "type": "string" }, + "sent": { "type": "integer", "minimum": 0 }, + "received": { "type": "integer", "minimum": 0 }, + "loss_pct": { "type": "number", "minimum": 0, "maximum": 100 }, + "min_ms": { "$ref": "#/$defs/ms" }, + "avg_ms": { "$ref": "#/$defs/ms" }, + "max_ms": { "$ref": "#/$defs/ms" }, + "stddev_ms": { "$ref": "#/$defs/ms" } + } + }, "error": { "type": "object", "required": ["schema_version", "type", "error"], diff --git a/internal/app/client.go b/internal/app/client.go index fe33d3a..c78029d 100644 --- a/internal/app/client.go +++ b/internal/app/client.go @@ -84,11 +84,12 @@ type Client struct { verbosityLevel int // 0 = summary, 1 = extended, >=2 = full detail // Mode - mode Mode // measure or stream + mode Mode // measure, stream, or ping once bool // stream: exit after the first event buffer int summaryInterval time.Duration sendDelay time.Duration // stream: delay between successive text message sends + interval time.Duration // ping: delay between successive ping frames // TLS configuration insecure bool // skip TLS certificate verification @@ -240,6 +241,12 @@ func WithSendDelay(d time.Duration) Option { return func(c *Client) { c.sendDelay = d } } +// WithInterval sets the ping-mode delay between successive ping frames. Zero applies the +// default (1s) in Validate; values below 10ms are rejected there to keep the tool polite. +func WithInterval(d time.Duration) Option { + return func(c *Client) { c.interval = d } +} + // WithInsecure configures whether to skip TLS certificate verification. func WithInsecure(insecure bool) Option { return func(c *Client) { c.insecure = insecure } @@ -324,6 +331,9 @@ func (c *Client) TextMessages() []string { return c.textMessages } // SendDelay returns the stream-mode delay between successive text message sends. func (c *Client) SendDelay() time.Duration { return c.sendDelay } +// Interval returns the ping-mode delay between successive ping frames. +func (c *Client) Interval() time.Duration { return c.interval } + // wsstatOptions builds wsstat options based on client configuration. func (c *Client) wsstatOptions() []wsstat.Option { var opts []wsstat.Option @@ -463,6 +473,10 @@ func (c *Client) Validate() error { return errors.New("send-delay must be zero or greater") } + if c.mode == ModePing { + return c.validatePing() + } + if c.mode == ModeStream { if c.once && c.count > 1 { return errors.New("count must be 0 or 1 when --once is set") diff --git a/internal/app/client_subscription_test.go b/internal/app/client_subscription_test.go index 1f9a413..ba7f53c 100644 --- a/internal/app/client_subscription_test.go +++ b/internal/app/client_subscription_test.go @@ -97,10 +97,10 @@ func TestStreamSubscriptionRespectsCount(t *testing.T) { defer server.cleanup() c := &Client{ - count: 2, - mode: ModeStream, + count: 2, + mode: ModeStream, textMessages: []string{"start"}, - quiet: true, + quiet: true, } require.NoError(t, c.Validate()) ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) @@ -167,9 +167,9 @@ func TestStreamSubscriptionUnlimitedRequiresCancel(t *testing.T) { defer server.cleanup() c := &Client{ - mode: ModeStream, + mode: ModeStream, textMessages: []string{"start"}, - quiet: true, + quiet: true, } require.NoError(t, c.Validate()) ctx, cancel := context.WithCancel(context.Background()) @@ -261,9 +261,9 @@ func TestStreamSubscriptionRawIsByteClean(t *testing.T) { defer server.cleanup() c := &Client{ - count: 2, - mode: ModeStream, - output: OutputRaw, + count: 2, + mode: ModeStream, + output: OutputRaw, textMessages: []string{"start"}, } require.NoError(t, c.Validate()) diff --git a/internal/app/client_validation_test.go b/internal/app/client_validation_test.go index e0cfe17..56fea7b 100644 --- a/internal/app/client_validation_test.go +++ b/internal/app/client_validation_test.go @@ -2,6 +2,7 @@ package app import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -109,6 +110,55 @@ func TestClientValidate(t *testing.T) { }) } +// TestValidatePing exercises the ModePing branch: the interval default/floor and the rejection +// of measure/stream-only knobs. +func TestValidatePing(t *testing.T) { + t.Parallel() + + t.Run("defaults interval to 1s", func(t *testing.T) { + c := &Client{mode: ModePing} + require.NoError(t, c.Validate()) + assert.Equal(t, time.Second, c.Interval()) + }) + + t.Run("unlimited count allowed", func(t *testing.T) { + c := &Client{mode: ModePing, count: 0} + require.NoError(t, c.Validate()) + assert.Equal(t, 0, c.Count()) + }) + + t.Run("interval below floor rejected", func(t *testing.T) { + c := &Client{mode: ModePing, interval: time.Millisecond} + assert.ErrorContains(t, c.Validate(), "at least") + }) + + t.Run("interval above timeout allowed", func(t *testing.T) { + // Unbounded reads keep the connection alive between pings, so a long interval no + // longer needs to stay below --timeout. + c := &Client{mode: ModePing, interval: 30 * time.Second, timeout: 5 * time.Second} + require.NoError(t, c.Validate()) + }) + + rejections := []struct { + name string + client Client + }{ + {"text", Client{mode: ModePing, textMessages: []string{"hi"}}}, + {"rpc method", Client{mode: ModePing, rpcMethod: "eth_x"}}, + {"once", Client{mode: ModePing, once: true}}, + {"buffer", Client{mode: ModePing, buffer: 8}}, + {"summary interval", Client{mode: ModePing, summaryInterval: time.Second}}, + {"send delay", Client{mode: ModePing, sendDelay: time.Second}}, + {"raw output", Client{mode: ModePing, output: "raw"}}, + {"file sink", Client{mode: ModePing, responseFilePath: "cap.ndjson"}}, + } + for _, tc := range rejections { + t.Run("rejects "+tc.name, func(t *testing.T) { + assert.Error(t, tc.client.Validate()) + }) + } +} + //revive:disable:function-length comprehensive table-driven test func TestValidateComplexCombinations(t *testing.T) { tests := []struct { diff --git a/internal/app/color/color.go b/internal/app/color/color.go index c1ddbf9..502132c 100644 --- a/internal/app/color/color.go +++ b/internal/app/color/color.go @@ -12,6 +12,7 @@ type RGB struct { var ( WSOrange = RGB{255, 102, 0} // WebSocket orange (#ff6600) TeaGreen = RGB{211, 249, 181} // Tea green (#d3f9b5) + Red = RGB{255, 85, 85} // Soft red (#ff5555), used for failure lines ) // Sprint returns the text with ANSI color codes applied diff --git a/internal/app/format.go b/internal/app/format.go index 890d2cd..14d0a44 100644 --- a/internal/app/format.go +++ b/internal/app/format.go @@ -37,6 +37,9 @@ const ( ModeMeasure Mode = iota // ModeStream streams subscription events. ModeStream + // ModePing sends periodic WebSocket ping frames on one connection and reports + // per-ping RTT plus a ping(8)-style summary. + ModePing ) // ParseOutput normalizes and validates an output contract string. An empty diff --git a/internal/app/output.go b/internal/app/output.go index dba12fd..7e0dc70 100644 --- a/internal/app/output.go +++ b/internal/app/output.go @@ -95,6 +95,15 @@ func (c *Client) colorizeGreen(text string) string { return color.TeaGreen.Sprint(text) } +// colorizeRed returns the text with red color applied if color output is enabled. Used for +// ping connection-death lines. +func (c *Client) colorizeRed(text string) string { + if !c.colorEnabled() { + return text + } + return color.Red.Sprint(text) +} + // printJSONLine prints a JSON line. func (*Client) printJSONLine(payload any) error { data, err := json.Marshal(payload) @@ -292,6 +301,98 @@ func (c *Client) printSubscriptionSummary(target *url.URL, result *wsstat.Result } } +// dialTimingSummary renders the one-line dial-phase breakdown for the ping header. The tls +// segment is omitted for ws:// targets (no TLS handshake). +func dialTimingSummary(result *wsstat.Result) string { + segs := []string{ + "dns " + formatDuration(result.DNSLookup), + "tcp " + formatDuration(result.TCPConnection), + } + if result.TLSHandshake > 0 { + segs = append(segs, "tls "+formatDuration(result.TLSHandshake)) + } + segs = append(segs, "ws "+formatDuration(result.WSHandshake)) + return strings.Join(segs, ", ") +} + +// printPingHeader prints the "PING (dns .. tcp .. ws ..)" line from the dial timings. +// JSON/raw output and -q suppress it (the dial breakdown is measure mode's job and stays out +// of the JSON contract in v1); -v adds the target/TLS request-detail block. +func (c *Client) printPingHeader(target *url.URL, result *wsstat.Result) error { + if c.output != OutputText || c.quiet { + return nil + } + fmt.Printf("%s %s (%s)\n", + c.colorizeOrange("PING"), target.String(), dialTimingSummary(result)) + if c.verbosityLevel >= 1 { + return c.PrintRequestDetails(&MeasurementResult{Result: result}) + } + return nil +} + +// printPingReply prints a single ping outcome. JSON emits one ping_reply record; text prints a +// colored line (green pong, orange timeout, red connection loss). -q suppresses text reply lines +// (the summary block is still printed), matching `ping -q`. +func (c *Client) printPingReply( + seq int, rtt time.Duration, outcome pingOutcome, reason string, +) error { + if c.output == OutputJSON { + return c.printJSONLine(c.pingReplyJSONFor(seq, rtt, outcome, reason)) + } + if c.output == OutputRaw || c.quiet { + return nil + } + switch outcome { + case pingPong: + fmt.Printf("%s seq=%d rtt=%s\n", c.colorizeGreen("pong:"), seq, formatDuration(rtt)) + case pingTimeout: + fmt.Printf("%s seq=%d (%s)\n", c.colorizeOrange("timeout:"), seq, c.pingTimeout()) + case pingDead: + fmt.Printf("%s seq=%d %s\n", c.colorizeRed("lost:"), seq, reason) + default: + // pingCanceled is handled before printing; nothing to render. + } + return nil +} + +// colorizeLossPct colors a loss percentage by severity: green at 0%, red at total loss, +// orange in between, mirroring the pong/timeout/lost reply-line colors. +func (c *Client) colorizeLossPct(pct float64) string { + s := fmt.Sprintf("%.1f%%", pct) + switch { + case pct <= 0: + return c.colorizeGreen(s) + case pct >= pctScale: + return c.colorizeRed(s) + default: + return c.colorizeOrange(s) + } +} + +// printPingSummary prints the closing statistics block. JSON emits one ping_summary record; +// text prints a "STATS (...)" line mirroring the PING header, omitting the rtt line +// when no pong was received. Printed on every termination path, including under -q. +func (c *Client) printPingSummary(report *PingReport) error { + if c.output == OutputJSON { + return c.printJSONLine(c.pingSummaryJSONFor(report)) + } + if c.output == OutputRaw { + return nil + } + fmt.Println() + fmt.Printf("%s %s (%d sent, %d received, %s loss)\n", + c.colorizeOrange("STATS"), report.Target.String(), + report.Sent, report.Received, c.colorizeLossPct(report.LossPct())) + if report.Received > 0 { + fmt.Printf("rtt: min=%s avg=%s max=%s stddev=%s\n", + c.colorizeGreen(msString(report.Min)+"ms"), + c.colorizeGreen(msString(report.Avg)+"ms"), + c.colorizeGreen(msString(report.Max)+"ms"), + c.colorizeGreen(msString(report.Stddev)+"ms")) + } + return nil +} + // printTimingResultsBasic prints a concise timing summary used for verbosity level 0. func (c *Client) printTimingResultsBasic(result *wsstat.Result, count int) { const printValueTemp = "%s: %s\n" diff --git a/internal/app/ping.go b/internal/app/ping.go new file mode 100644 index 0000000..6a3f0ac --- /dev/null +++ b/internal/app/ping.go @@ -0,0 +1,314 @@ +package app + +import ( + "context" + "errors" + "fmt" + "math" + "net" + "net/url" + "time" + + "github.com/jkbrsn/wsstat/v3" +) + +const ( + // defaultPingInterval is the delay between successive pings when --interval is unset. + defaultPingInterval = time.Second + // minPingInterval is the floor on --interval; a smaller value is rejected so a typo + // cannot flood a production endpoint. + minPingInterval = 10 * time.Millisecond + // defaultReadTimeout mirrors the core's read/dial timeout default (wsstat.defaultTimeout), + // used to render loss reasons when --timeout is unset. + defaultReadTimeout = 5 * time.Second + // pingCloseGrace bounds the closing handshake so teardown cannot blow past --deadline: + // the deferred Close runs after the deadline fires, and the default 3s grace would let a + // non-echoing peer hold the process well beyond the advertised max run time. An echoing + // peer completes the handshake in one RTT, far under this cap. + pingCloseGrace = 500 * time.Millisecond + // pctScale converts a fraction to a percentage. + pctScale = 100 +) + +// pingOutcome classifies a single ping's result. The connection dials with WithUnboundedReads, +// so a missed pong no longer tears the socket down: a timeout is a survivable loss and the run +// continues, exactly like ping(8). Only a real connection close (or a context canceled mid-ping) +// ends the run. +type pingOutcome int + +const ( + // pingPong is a successful round-trip (a pong was received). + pingPong pingOutcome = iota + // pingTimeout is a missed pong within the per-ping timeout; the connection survives, so + // the run continues. + pingTimeout + // pingDead is a closed connection or other transport error; it ends the run. + pingDead + // pingCanceled marks a ping interrupted mid-flight by context cancellation (Ctrl-C or + // --deadline). It counts as neither sent nor lost, prints no reply line, and ends the run. + pingCanceled +) + +// pingStats accumulates per-ping RTTs across a run: sent/received counts, min/max/sum, and a +// sum-of-squares for population stddev (what ping(8) labels mdev). Only received pongs feed the +// RTT aggregates; a lost ping counts toward sent only. +type pingStats struct { + sent int + received int + min time.Duration + max time.Duration + sum time.Duration + sumSq float64 // sum of squared RTTs in ns^2 (float64 to avoid int64 overflow) +} + +// observe records a received pong's round-trip time. +func (s *pingStats) observe(rtt time.Duration) { + s.received++ + if s.received == 1 || rtt < s.min { + s.min = rtt + } + if rtt > s.max { + s.max = rtt + } + s.sum += rtt + ns := float64(rtt) + s.sumSq += ns * ns +} + +// report snapshots the accumulator into a PingReport. RTT aggregates are left zero when no +// pong was received (the output layer omits them). +func (s *pingStats) report(target *url.URL) *PingReport { + r := &PingReport{Target: target, Sent: s.sent, Received: s.received} + if s.received == 0 { + return r + } + n := float64(s.received) + r.Min = s.min + r.Max = s.max + r.Avg = s.sum / time.Duration(s.received) + mean := float64(s.sum) / n + variance := s.sumSq/n - mean*mean + if variance < 0 { + // Floating-point cancellation can push an all-equal sample slightly negative. + variance = 0 + } + r.Stddev = time.Duration(math.Sqrt(variance)) + return r +} + +// PingReport is the outcome of a ping run, returned so the caller can decide the exit code +// (zero pongs received == total loss). Per-ping lines are printed live inside RunPing. +type PingReport struct { + Target *url.URL + Sent int + Received int + Min, Avg, Max, Stddev time.Duration +} + +// LossPct returns the packet-loss percentage over the run (0 when nothing was sent). +func (r *PingReport) LossPct() float64 { + if r.Sent == 0 { + return 0 + } + return float64(r.Sent-r.Received) / float64(r.Sent) * pctScale +} + +// validatePing checks ping-mode configuration and applies the interval default. Ping dials once +// and sends bare ping frames, so it rejects every measure/stream-only knob that implies a +// payload, a second message, or a summary cadence. +func (c *Client) validatePing() error { + switch { + case c.output == OutputRaw: + return errors.New("-o raw has no meaning in ping mode (no response payloads)") + case c.responseFilePath != "": + return errors.New("--file has no meaning in ping mode (no response payloads)") + case len(c.textMessages) > 0: + return errors.New("-t/--text is not supported in ping mode") + case c.rpcMethod != "": + return errors.New("--rpc-method is not supported in ping mode") + case c.once: + return errors.New("--once is not supported in ping mode") + case c.buffer > 0: + return errors.New("-b/--buffer is not supported in ping mode") + case c.summaryInterval > 0: + return errors.New("--summary-interval is not supported in ping mode") + case c.sendDelay > 0: + return errors.New("--send-delay is not supported in ping mode") + } + if c.interval == 0 { + c.interval = defaultPingInterval + } + if c.interval < minPingInterval { + return fmt.Errorf("interval must be at least %s", minPingInterval) + } + return nil +} + +// pingTimeout returns the effective per-ping read timeout, used to bound the pong wait and to +// render loss reasons. +func (c *Client) pingTimeout() time.Duration { + if c.timeout > 0 { + return c.timeout + } + return defaultReadTimeout +} + +// classifyPing maps a non-nil PingPong error to an outcome and a human reason. With unbounded +// reads a missed pong surfaces as a clean context.DeadlineExceeded and leaves the connection +// alive, so it is a survivable timeout; anything else (a peer close, a transport error) means +// the connection is gone. Only stdlib and wsstat sentinels are consulted (no coder/websocket +// import). +func (c *Client) classifyPing(err error) (pingOutcome, string) { + if errors.Is(err, context.DeadlineExceeded) { + return pingTimeout, fmt.Sprintf("no response within %s", c.pingTimeout()) + } + return pingDead, deadReason(err) +} + +// deadReason renders the human reason for a terminal ping loss. +func deadReason(err error) string { + if errors.Is(err, net.ErrClosed) || errors.Is(err, wsstat.ErrClosed) { + return "connection closed" + } + return err.Error() +} + +// RunPing dials the target once (with unbounded reads so an idle connection is not torn down +// between pings), then sends a WebSocket ping frame every --interval on that connection, printing +// a per-ping RTT line live and a ping(8)-style summary at the end. A missed pong is reported and +// the run continues; the run ends only when the count is reached, the context is canceled (Ctrl-C +// or --deadline), or the connection closes. All paths print the summary. +// +// The error return is reserved for runtime failures the caller must surface as a non-zero exit +// (bad header, dial failure, output-write failure). Context cancellation and connection loss are +// swallowed: the summary is printed and a nil error is returned, so the caller derives the exit +// code from the report (zero pongs received == total loss) rather than from the error. +func (c *Client) RunPing(ctx context.Context, target *url.URL) (*PingReport, error) { + header, err := parseHeaders(c.headers) + if err != nil { + return nil, err + } + + // Unbounded reads keep the connection alive while only ping/pong traffic flows: pongs are + // control frames the read pump never sees as reads, so the default per-read timeout would + // otherwise close the socket one --timeout after dial. Discarded reads keep the pump + // running against a chatty peer (welcome messages, heartbeats, feed data): nothing here + // consumes data frames, and a stalled pump would starve pong processing and misreport a + // healthy endpoint as total loss. + ws := wsstat.New(append( + c.wsstatOptions(), wsstat.WithUnboundedReads(), wsstat.WithDiscardReads(), + wsstat.WithCloseGrace(pingCloseGrace))...) + if err := ws.DialContext(ctx, target, header); err != nil { + ws.Close() + return nil, handleConnectionError(err, target.String()) + } + defer ws.Close() + + if err := c.printPingHeader(target, ws.ExtractResult()); err != nil { + return nil, err + } + + interval := c.interval + if interval <= 0 { + interval = defaultPingInterval + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + stats := &pingStats{} +loop: + for seq := 1; c.count == 0 || seq <= c.count; seq++ { + if ctx.Err() != nil { + break + } + outcome, err := c.pingOnce(ctx, ws, stats, seq) + if err != nil { + return nil, err + } + if outcome == pingDead || outcome == pingCanceled { + break + } + if c.count != 0 && seq == c.count { + break + } + + select { + case <-ctx.Done(): + break loop // labeled: a bare break would exit only the select + case <-ticker.C: + } + } + + report := stats.report(target) + if err := c.printPingSummary(report); err != nil { + return nil, err + } + return report, nil +} + +// pingOnce sends one ping, records it, and prints the reply line, returning the outcome. A pong +// or a survivable timeout lets the run continue; pingDead (a closed connection) and pingCanceled +// (a context canceled mid-ping) end it. A pong that arrived in the same instant the context was +// canceled is still a received pong (the run then ends via the loop's context check), so a +// --deadline that fires just after the reply cannot flip a live host to total loss. A canceled +// ping with no pong counts as neither sent nor lost: its wait was cut short of the full timeout +// window, so letting it into the stats would report phantom loss on every Ctrl-C. sent otherwise +// counts attempts, so a write that failed on an already-dead connection is included (unlike +// ping(8), which counts only transmitted probes). The error return is an output-write error. +func (c *Client) pingOnce( + ctx context.Context, ws *wsstat.WSStat, stats *pingStats, seq int, +) (pingOutcome, error) { + start := time.Now() + pingErr := ws.PingPongContext(ctx) + rtt := time.Since(start) + + if pingErr == nil { + stats.sent++ + stats.observe(rtt) + return pingPong, c.printPingReply(seq, rtt, pingPong, "") + } + if ctx.Err() != nil { + return pingCanceled, nil + } + stats.sent++ + outcome, reason := c.classifyPing(pingErr) + return outcome, c.printPingReply(seq, rtt, outcome, reason) +} + +// pingReplyJSONFor builds the NDJSON envelope for a single ping reply. +func (*Client) pingReplyJSONFor( + seq int, rtt time.Duration, outcome pingOutcome, reason string, +) pingReplyJSON { + rec := pingReplyJSON{Schema: JSONSchemaVersion, Type: "ping_reply", Seq: seq} + if outcome == pingPong { + // Pointer (not omitempty float) so a sub-microsecond RTT that rounds to 0.0 still + // serializes; a lost reply leaves it nil. + ms := msFloat(rtt) + rec.RTTMs = &ms + } else { + // pingTimeout and pingDead both record a lost ping with a reason. + rec.Lost = true + rec.Error = reason + } + return rec +} + +// pingSummaryJSONFor builds the NDJSON envelope for the run summary. +func (*Client) pingSummaryJSONFor(report *PingReport) pingSummaryJSON { + rec := pingSummaryJSON{ + Schema: JSONSchemaVersion, + Type: "ping_summary", + URL: report.Target.String(), + Sent: report.Sent, + Received: report.Received, + LossPct: report.LossPct(), + } + if report.Received > 0 { + // Pointers (not omitempty floats) so a legitimately-zero aggregate — e.g. a + // single-sample stddev — stays present whenever a pong was received. + minMs, avgMs := msFloat(report.Min), msFloat(report.Avg) + maxMs, stddevMs := msFloat(report.Max), msFloat(report.Stddev) + rec.MinMs, rec.AvgMs, rec.MaxMs, rec.StddevMs = &minMs, &avgMs, &maxMs, &stddevMs + } + return rec +} diff --git a/internal/app/ping_test.go b/internal/app/ping_test.go new file mode 100644 index 0000000..d69fb24 --- /dev/null +++ b/internal/app/ping_test.go @@ -0,0 +1,545 @@ +package app + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// pingServerMode selects how the ping test server answers (or ignores) client pings. +type pingServerMode int + +const ( + // pingServerAnswer auto-answers every ping (coder responds below the app once reading). + pingServerAnswer pingServerMode = iota + // pingServerNoRead never reads, so pings go unanswered and time out client-side. + pingServerNoRead + // pingServerCloseAfter answers for closeAfter, then closes the connection normally. + pingServerCloseAfter + // pingServerDelayRead ignores pings for delayRead (they time out), then answers. + pingServerDelayRead + // pingServerChatty answers pings while continuously pushing unsolicited text frames. + pingServerChatty +) + +type pingServerOpts struct { + mode pingServerMode + closeAfter time.Duration + delayRead time.Duration +} + +// newPingServer starts an httptest WebSocket server whose ping handling follows opts, and +// returns its ws:// URL. The server is torn down via t.Cleanup. +func newPingServer(t *testing.T, opts pingServerOpts) *url.URL { + t.Helper() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true}) + if err != nil { + return + } + defer func() { _ = conn.CloseNow() }() + ctx := r.Context() + + switch opts.mode { + case pingServerNoRead: + <-ctx.Done() + case pingServerDelayRead: + select { + case <-time.After(opts.delayRead): + case <-ctx.Done(): + return + } + readCtx := conn.CloseRead(ctx) + <-readCtx.Done() + case pingServerChatty: + pushChatter(conn.CloseRead(ctx), conn) + case pingServerCloseAfter: + readCtx := conn.CloseRead(ctx) + select { + case <-time.After(opts.closeAfter): + _ = conn.Close(websocket.StatusNormalClosure, "bye") + case <-readCtx.Done(): + } + default: // pingServerAnswer + readCtx := conn.CloseRead(ctx) + <-readCtx.Done() + } + })) + t.Cleanup(server.Close) + + u, err := url.Parse("ws" + strings.TrimPrefix(server.URL, "http")) + require.NoError(t, err) + return u +} + +// pushChatter writes unsolicited text frames every 2ms until the connection or context ends, +// simulating a feed endpoint that talks without being asked. +func pushChatter(ctx context.Context, conn *websocket.Conn) { + for { + if err := conn.Write(ctx, websocket.MessageText, []byte("chatter")); err != nil { + return + } + select { + case <-time.After(2 * time.Millisecond): + case <-ctx.Done(): + return + } + } +} + +func TestPingStats(t *testing.T) { + t.Parallel() + + t.Run("empty yields no rtt aggregates", func(t *testing.T) { + s := &pingStats{} + s.sent = 3 + r := s.report(nil) + assert.Equal(t, 3, r.Sent) + assert.Equal(t, 0, r.Received) + assert.Zero(t, r.Min) + assert.Zero(t, r.Avg) + assert.Zero(t, r.Max) + assert.Zero(t, r.Stddev) + assert.InDelta(t, 100.0, r.LossPct(), 0.001) + }) + + t.Run("single sample has zero stddev", func(t *testing.T) { + s := &pingStats{} + s.sent = 1 + s.observe(10 * time.Millisecond) + r := s.report(nil) + assert.Equal(t, 1, r.Received) + assert.Equal(t, 10*time.Millisecond, r.Min) + assert.Equal(t, 10*time.Millisecond, r.Avg) + assert.Equal(t, 10*time.Millisecond, r.Max) + assert.Zero(t, r.Stddev) + assert.InDelta(t, 0.0, r.LossPct(), 0.001) + }) + + t.Run("known fixture stddev", func(t *testing.T) { + // Samples 10/20/30/40ms: mean 25ms, population variance 125ms^2, stddev sqrt(125) + // = 11.1803ms. + s := &pingStats{} + s.sent = 4 + for _, ms := range []int{10, 20, 30, 40} { + s.observe(time.Duration(ms) * time.Millisecond) + } + r := s.report(nil) + assert.Equal(t, 4, r.Received) + assert.Equal(t, 10*time.Millisecond, r.Min) + assert.Equal(t, 40*time.Millisecond, r.Max) + assert.Equal(t, 25*time.Millisecond, r.Avg) + assert.InDelta(t, 11.1803, float64(r.Stddev)/float64(time.Millisecond), 0.001) + assert.InDelta(t, 0.0, r.LossPct(), 0.001) + }) +} + +func TestRunPingCountReached(t *testing.T) { + // No t.Parallel: captureStdoutFrom swaps the global os.Stdout. + u := newPingServer(t, pingServerOpts{mode: pingServerAnswer}) + c := &Client{count: 3, mode: ModePing, interval: 10 * time.Millisecond} + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + require.NotNil(t, report) + assert.Equal(t, 3, report.Sent) + assert.Equal(t, 3, report.Received) + // Sequence numbers are 1-based and monotonic. + assert.Less(t, strings.Index(out, "seq=1"), strings.Index(out, "seq=2")) + assert.Less(t, strings.Index(out, "seq=2"), strings.Index(out, "seq=3")) + assert.Equal(t, 3, strings.Count(out, "pong: seq=")) + assert.Contains(t, out, "3 sent, 3 received, 0.0% loss") + assert.Contains(t, out, "STATS ") +} + +func TestRunPingContextCancelPrintsSummary(t *testing.T) { + u := newPingServer(t, pingServerOpts{mode: pingServerAnswer}) + c := &Client{count: 0, mode: ModePing, interval: 15 * time.Millisecond} + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(120 * time.Millisecond) + cancel() + }() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err // nil: ctx cancellation is swallowed + }) + + require.NotNil(t, report) + assert.GreaterOrEqual(t, report.Received, 1, "at least one pong before cancel") + assert.Equal(t, report.Received, report.Sent, + "a canceled in-flight ping must not count as loss") + assert.Contains(t, out, "STATS ") +} + +func TestRunPingSurvivesChattyPeer(t *testing.T) { + // A peer that pushes unsolicited data frames must not starve pong processing: without + // discarded reads the pump stalls once readChan's 8-slot buffer fills, coder/websocket + // stops reading, and every subsequent ping times out against a healthy endpoint. + u := newPingServer(t, pingServerOpts{mode: pingServerChatty}) + c := &Client{ + count: 5, mode: ModePing, interval: 20 * time.Millisecond, timeout: 2 * time.Second, + } + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + require.NotNil(t, report) + assert.Equal(t, 5, report.Sent) + assert.Equal(t, 5, report.Received, "pongs must get through despite unsolicited data frames") + assert.Equal(t, 5, strings.Count(out, "pong: seq=")) +} + +func TestRunPingCancelInterruptsBlockedPing(t *testing.T) { + // Cancellation (Ctrl-C, --deadline) must cut short a ping blocked on an unresponsive + // peer instead of waiting out the full --timeout: PingPongContext bounds the pong wait + // by the caller context too. + u := newPingServer(t, pingServerOpts{mode: pingServerNoRead}) + c := &Client{ + count: 0, + mode: ModePing, + interval: 10 * time.Millisecond, + timeout: 30 * time.Second, + closeGrace: 100 * time.Millisecond, + } + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(80 * time.Millisecond) + cancel() + }() + + start := time.Now() + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + assert.Less(t, time.Since(start), 5*time.Second, "cancel must not wait out --timeout") + require.NotNil(t, report) + assert.Zero(t, report.Sent, "a canceled in-flight ping is not counted") + assert.Zero(t, report.Received) + assert.Contains(t, out, "STATS ") +} + +func TestRunPingConnectionDeath(t *testing.T) { + u := newPingServer(t, pingServerOpts{ + mode: pingServerCloseAfter, closeAfter: 60 * time.Millisecond, + }) + c := &Client{count: 0, mode: ModePing, interval: 15 * time.Millisecond, timeout: time.Second} + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err // nil: connection death is reported via the summary, not the error + }) + + require.NotNil(t, report) + assert.GreaterOrEqual(t, report.Received, 1, "some pongs before the server closed") + assert.Equal(t, report.Received+1, report.Sent, "the failing ping counts as sent, not received") + assert.Contains(t, out, "lost: seq=") + assert.Contains(t, out, "STATS ") +} + +func TestRunPingZeroPongs(t *testing.T) { + // A server that never answers times out every ping. With unbounded reads a timeout is + // survivable, so the run does not end early: all -c pings fire, each a timeout, and the + // summary reports total loss. + u := newPingServer(t, pingServerOpts{mode: pingServerNoRead}) + c := &Client{ + count: 3, + mode: ModePing, + interval: 10 * time.Millisecond, + timeout: 120 * time.Millisecond, + closeGrace: 100 * time.Millisecond, + } + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + require.NotNil(t, report) + assert.Equal(t, 3, report.Sent, "all pings fire; a timeout is survivable") + assert.Equal(t, 0, report.Received) + assert.InDelta(t, 100.0, report.LossPct(), 0.001) + assert.Equal(t, 3, strings.Count(out, "timeout: seq=")) + assert.Contains(t, out, "3 sent, 0 received, 100.0% loss") + assert.NotContains(t, out, "rtt:") +} + +func TestRunPingTimeoutIsSurvivable(t *testing.T) { + // The server ignores the first ping (it times out) then starts answering. With unbounded + // reads the connection survives the timeout, so later pings pong: a run continues through + // a transient loss rather than ending on it. This is the core of the fix. + // Margins are deliberately wide (delayRead is 2.5x the ping timeout) so scheduling + // jitter under -race and CI repetition cannot let the server answer before the first + // ping's timeout fires. + u := newPingServer(t, pingServerOpts{ + mode: pingServerDelayRead, delayRead: 150 * time.Millisecond, + }) + c := &Client{ + count: 8, + mode: ModePing, + interval: 50 * time.Millisecond, + timeout: 60 * time.Millisecond, + } + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + require.NotNil(t, report) + assert.Equal(t, 8, report.Sent, "all pings fire; the run is not cut short by the timeout") + assert.GreaterOrEqual(t, report.Received, 1, "connection survives the timeout and later pongs") + assert.Contains(t, out, "timeout: seq=1", "the first ping times out") + assert.Contains(t, out, "pong: seq=", "a later ping succeeds on the same connection") +} + +func TestRunPingTinyIntervalStaysSequential(t *testing.T) { + // A small interval must not pile up pings: PingPong is synchronous, so the run fires + // exactly count sequential pings with monotonic seq (missed ticks are dropped). + u := newPingServer(t, pingServerOpts{mode: pingServerAnswer}) + c := &Client{count: 5, mode: ModePing, interval: minPingInterval} + require.NoError(t, c.Validate()) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var report *PingReport + out := captureStdoutFrom(t, func() error { + var err error + report, err = c.RunPing(ctx, u) + return err + }) + + require.NotNil(t, report) + assert.Equal(t, 5, report.Sent) + assert.Equal(t, 5, report.Received) + assert.Equal(t, 5, strings.Count(out, "pong: seq=")) +} + +func TestPingHeaderOutput(t *testing.T) { + res := sampleTimingResult(t) // wss:// with dns/tcp/tls/ws timings + u := res.URL + + t.Run("text includes dial phases", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { return c.printPingHeader(u, res) }) + assert.Contains(t, out, "PING "+u.String()) + assert.Contains(t, out, "dns 10ms") + assert.Contains(t, out, "tcp 20ms") + assert.Contains(t, out, "tls 30ms") + assert.Contains(t, out, "ws 40ms") + }) + + t.Run("ws scheme omits tls segment", func(t *testing.T) { + wsRes := sampleTimingResult(t) + wsURL, err := url.Parse("ws://plain.test/ws") + require.NoError(t, err) + wsRes.URL = wsURL + wsRes.TLSHandshake = 0 + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { return c.printPingHeader(wsURL, wsRes) }) + assert.NotContains(t, out, "tls ") + assert.Contains(t, out, "ws 40ms") + }) + + t.Run("quiet suppresses header", func(t *testing.T) { + c := &Client{output: OutputText, quiet: true} + out := captureStdoutFrom(t, func() error { return c.printPingHeader(u, res) }) + assert.Empty(t, out) + }) + + t.Run("json suppresses header", func(t *testing.T) { + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { return c.printPingHeader(u, res) }) + assert.Empty(t, out) + }) +} + +func TestPingReplyOutput(t *testing.T) { + t.Run("text pong", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(2, 12300*time.Microsecond, pingPong, "") + }) + assert.Equal(t, "pong: seq=2 rtt=12.3ms\n", out) + }) + + t.Run("text timeout", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(3, 0, pingTimeout, "no response within 5s") + }) + assert.Equal(t, "timeout: seq=3 (5s)\n", out) + }) + + t.Run("text connection loss", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(4, 0, pingDead, "connection closed") + }) + assert.Equal(t, "lost: seq=4 connection closed\n", out) + }) + + t.Run("quiet suppresses text reply", func(t *testing.T) { + c := &Client{output: OutputText, quiet: true} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(1, time.Millisecond, pingPong, "") + }) + assert.Empty(t, out) + }) + + t.Run("json pong record", func(t *testing.T) { + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(4, 12300*time.Microsecond, pingPong, "") + }) + p := decodeJSONLine(t, out) + assert.Equal(t, "ping_reply", p["type"]) + assert.EqualValues(t, 4, p["seq"]) + assert.InDelta(t, 12.3, p["rtt_ms"], 0.001) + _, hasLost := p["lost"] + assert.False(t, hasLost, "pong record omits lost") + }) + + t.Run("json lost record", func(t *testing.T) { + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { + return c.printPingReply(5, 0, pingTimeout, "no response within 5s") + }) + p := decodeJSONLine(t, out) + assert.Equal(t, "ping_reply", p["type"]) + assert.EqualValues(t, 5, p["seq"]) + assert.Equal(t, true, p["lost"]) + assert.Equal(t, "no response within 5s", p["error"]) + _, hasRTT := p["rtt_ms"] + assert.False(t, hasRTT, "lost record omits rtt_ms") + }) +} + +func TestPingSummaryOutput(t *testing.T) { + u, err := url.Parse("wss://echo.test/ws") + require.NoError(t, err) + + full := func() *PingReport { + return &PingReport{ + Target: u, Sent: 4, Received: 3, + Min: 11800 * time.Microsecond, + Avg: 12100 * time.Microsecond, + Max: 12300 * time.Microsecond, + Stddev: 200 * time.Microsecond, + } + } + + t.Run("text with rtt line", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { return c.printPingSummary(full()) }) + assert.Contains(t, out, "STATS "+u.String()+" (4 sent, 3 received, 25.0% loss)") + assert.Contains(t, out, "rtt: min=11.8ms avg=12.1ms max=12.3ms stddev=0.2ms") + }) + + t.Run("text omits rtt when zero received", func(t *testing.T) { + c := &Client{output: OutputText, colorMode: "never"} + out := captureStdoutFrom(t, func() error { + return c.printPingSummary(&PingReport{Target: u, Sent: 2, Received: 0}) + }) + assert.Contains(t, out, "2 sent, 0 received, 100.0% loss") + assert.NotContains(t, out, "rtt:") + }) + + t.Run("json record", func(t *testing.T) { + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { return c.printPingSummary(full()) }) + p := decodeJSONLine(t, out) + assert.Equal(t, "ping_summary", p["type"]) + assert.Equal(t, u.String(), p["url"]) + assert.EqualValues(t, 4, p["sent"]) + assert.EqualValues(t, 3, p["received"]) + assert.InDelta(t, 25.0, p["loss_pct"], 0.001) + assert.InDelta(t, 11.8, p["min_ms"], 0.001) + assert.InDelta(t, 0.2, p["stddev_ms"], 0.001) + }) + + t.Run("json omits rtt when zero received", func(t *testing.T) { + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { + return c.printPingSummary(&PingReport{Target: u, Sent: 3, Received: 0}) + }) + p := decodeJSONLine(t, out) + assert.EqualValues(t, 0, p["received"]) + assert.InDelta(t, 100.0, p["loss_pct"], 0.001) + _, hasMin := p["min_ms"] + assert.False(t, hasMin, "zero-received summary omits min_ms") + }) + + t.Run("json single sample keeps zero stddev", func(t *testing.T) { + // A single pong has a legitimately-zero stddev; it must stay present (not dropped by + // omitempty) so a received>0 summary always carries all four aggregates. + report := &PingReport{ + Target: u, Sent: 1, Received: 1, + Min: 5 * time.Millisecond, Avg: 5 * time.Millisecond, + Max: 5 * time.Millisecond, Stddev: 0, + } + c := &Client{output: OutputJSON} + out := captureStdoutFrom(t, func() error { return c.printPingSummary(report) }) + p := decodeJSONLine(t, out) + v, ok := p["stddev_ms"] + require.True(t, ok, "stddev_ms present for a received>0 summary even when zero") + assert.InDelta(t, 0.0, v, 0.0001) + assert.InDelta(t, 5.0, p["min_ms"], 0.001) + }) +} diff --git a/internal/app/schema_doc_test.go b/internal/app/schema_doc_test.go index 141efac..83939c8 100644 --- a/internal/app/schema_doc_test.go +++ b/internal/app/schema_doc_test.go @@ -40,7 +40,8 @@ func TestSchemaDocDrift(t *testing.T) { got := recordTypeConsts(defs) slices.Sort(got) want := []string{ - "error", "response", "subscription_message", "subscription_summary", "timing", + "error", "ping_reply", "ping_summary", "response", + "subscription_message", "subscription_summary", "timing", } assert.Equal(t, want, got, "schema doc record types out of sync with emitted records") } diff --git a/internal/app/sink_test.go b/internal/app/sink_test.go index 135e533..1303d08 100644 --- a/internal/app/sink_test.go +++ b/internal/app/sink_test.go @@ -86,10 +86,10 @@ func TestStreamSubscriptionRecordsToSink(t *testing.T) { var buf bytes.Buffer c := &Client{ - count: 2, - mode: ModeStream, + count: 2, + mode: ModeStream, textMessages: []string{"start"}, - quiet: true, + quiet: true, } require.NoError(t, c.Validate()) c.SetResponseSink(&buf) diff --git a/internal/app/types.go b/internal/app/types.go index e8e3867..e7b430e 100644 --- a/internal/app/types.go +++ b/internal/app/types.go @@ -95,6 +95,35 @@ type subscriptionEntryJSON struct { Error string `json:"error,omitempty"` } +// pingReplyJSON is one NDJSON record per ping in ping mode. On a pong rtt_ms is set; on a +// timeout or connection death lost is true and error carries the detail (rtt_ms omitted). +// A pong sets RTTMs (a non-nil pointer, so a sub-microsecond RTT that rounds to 0.0 still +// serializes); a lost reply leaves it nil and sets Lost + Error. +type pingReplyJSON struct { + Schema string `json:"schema_version"` + Type string `json:"type"` // "ping_reply" + Seq int `json:"seq"` + RTTMs *float64 `json:"rtt_ms,omitempty"` // nil (omitted) when lost + Lost bool `json:"lost,omitempty"` + Error string `json:"error,omitempty"` // timeout / close detail +} + +// pingSummaryJSON is the final NDJSON record in ping mode: counts, loss, and RTT aggregates. +// The RTT fields are pointers so they stay present (and zero-valued, e.g. a single-sample +// stddev) whenever a pong was received, and are omitted only when received == 0. +type pingSummaryJSON struct { + Schema string `json:"schema_version"` + Type string `json:"type"` // "ping_summary" + URL string `json:"url"` + Sent int `json:"sent"` + Received int `json:"received"` + LossPct float64 `json:"loss_pct"` + MinMs *float64 `json:"min_ms,omitempty"` // nil (omitted) when received == 0 + AvgMs *float64 `json:"avg_ms,omitempty"` + MaxMs *float64 `json:"max_ms,omitempty"` + StddevMs *float64 `json:"stddev_ms,omitempty"` +} + // errorOutputJSON is the schema-stable error envelope emitted under -o json when a runtime // failure occurs, so a `wsstat ... -o json | jq` pipeline always receives a parseable record // on the failure path instead of falling back to plain stderr text. diff --git a/wsstat.go b/wsstat.go index 6441b22..1e74579 100644 --- a/wsstat.go +++ b/wsstat.go @@ -127,16 +127,18 @@ type WSStat struct { wgPumps sync.WaitGroup // instance configuration - timeout time.Duration - closeGrace time.Duration - tlsConf *tls.Config - resolves map[string]string // DNS resolution overrides: "host:port" → "address" - readLimit int64 // max inbound message size; -1 disables the limit - subprotocols []string // WebSocket subprotocols to negotiate - headers http.Header // headers merged into every handshake - compress bool // negotiate permessage-deflate - validateUTF8 bool // validate UTF-8 on inbound text frames - invalidUTF8 atomic.Int64 // count of text frames that failed UTF-8 validation + timeout time.Duration + closeGrace time.Duration + tlsConf *tls.Config + resolves map[string]string // DNS resolution overrides: "host:port" → "address" + readLimit int64 // max inbound message size; -1 disables the limit + subprotocols []string // WebSocket subprotocols to negotiate + headers http.Header // headers merged into every handshake + compress bool // negotiate permessage-deflate + validateUTF8 bool // validate UTF-8 on inbound text frames + unboundedReads bool // drop the read pump's per-read timeout (long-lived sessions) + discardReads bool // drop unclaimed inbound data frames instead of queueing them + invalidUTF8 atomic.Int64 // count of text frames that failed UTF-8 validation // Close-handshake frame, settable via CloseWith before teardown. closeStatus atomic.Int64 // handshake close code (default StatusNormalClosure) @@ -183,6 +185,8 @@ func New(opts ...Option) *WSStat { headers: cfg.headers, compress: cfg.compress, validateUTF8: cfg.validateUTF8, + unboundedReads: cfg.unboundedReads, + discardReads: cfg.discardReads, subscriptions: make(map[string]*subscriptionState), subscriptionArchive: make(map[string]SubscriptionStats), defaultSubscriptionBuffer: defaultSubscriptionBufferSize, @@ -353,29 +357,43 @@ func (ws *WSStat) readPump() { ws.log.Warn().Int("bytes", len(read.data)).Msg("received text frame with invalid UTF-8") } - // Message read successfully, dispatch if not handled by subscription - if ws.dispatchIncoming(read) { - continue - } - - select { - case ws.readChan <- read: - case <-ws.ctx.Done(): - ws.log.Debug().Msg("Context done, dropping read message") + if !ws.deliverRead(read) { return } } } -// readFrame reads one frame, bounding the read with the dial/read timeout only when -// no subscription is active. Subscriptions are long-lived and idle by nature, so a -// per-read deadline would tear them down after a quiet interval; while one is active -// the read blocks until ws.ctx is canceled (Close). deadlineHit reports that the bound -// fired (a one-shot timeout) rather than a real transport error or context cancel. +// deliverRead routes a successfully-read frame: subscription dispatch first; in discard mode +// unclaimed frames are dropped so the pump keeps reading (no consumer drains readChan, and a +// blocked pump would also starve pong control-frame handling — coder/websocket processes +// control frames only inside Read); otherwise the frame is queued for the read methods. +// Reports false when the pump should exit. +func (ws *WSStat) deliverRead(read *wsRead) bool { + if ws.dispatchIncoming(read) { + return true + } + if ws.discardReads { + return true + } + select { + case ws.readChan <- read: + return true + case <-ws.ctx.Done(): + ws.log.Debug().Msg("Context done, dropping read message") + return false + } +} + +// readFrame reads one frame, bounding the read with the dial/read timeout only when no +// subscription is active and WithUnboundedReads was not set. Subscriptions (and unbounded-read +// sessions such as a ping/pong monitor) are long-lived and idle by nature, so a per-read +// deadline would tear them down after a quiet interval; in those modes the read blocks until +// ws.ctx is canceled (Close). deadlineHit reports that the bound fired (a one-shot timeout) +// rather than a real transport error or context cancel. func (ws *WSStat) readFrame(conn *websocket.Conn) (*wsRead, bool) { readCtx := ws.ctx var cancel context.CancelFunc - if !ws.hasActiveSubscriptions() { + if !ws.unboundedReads && !ws.hasActiveSubscriptions() { readCtx, cancel = context.WithTimeout(ws.ctx, ws.timeout) } coderType, p, err := conn.Read(readCtx) @@ -601,6 +619,13 @@ func (ws *WSStat) WriteMessageJSON(v any) { // are recorded around the single call. // Sets result times: MessageReads, MessageWrites func (ws *WSStat) PingPong() error { + return ws.PingPongContext(context.Background()) +} + +// PingPongContext is PingPong with the pong wait additionally bounded by ctx: the round-trip +// ends at the earliest of ctx's cancellation, the connection's context, and the read timeout. +// It lets a caller-side deadline or interrupt cut short a ping blocked on an unresponsive peer. +func (ws *WSStat) PingPongContext(ctx context.Context) error { if ws.closed.Load() { return ErrClosed } @@ -613,8 +638,10 @@ func (ws *WSStat) PingPong() error { ws.timings.messageWrites = append(ws.timings.messageWrites, time.Now()) ws.timings.mu.Unlock() - pingCtx, cancel := context.WithTimeout(ws.ctx, ws.timeout) + pingCtx, cancel := context.WithTimeout(ctx, ws.timeout) defer cancel() + stop := context.AfterFunc(ws.ctx, cancel) + defer stop() if err := conn.Ping(pingCtx); err != nil { return err } @@ -1061,6 +1088,13 @@ type options struct { headers http.Header // headers merged into every handshake compress bool // negotiate permessage-deflate validateUTF8 bool // validate UTF-8 on inbound text frames + // unboundedReads drops the per-read timeout on the read pump (like an active + // subscription), for long-lived sessions where control-frame traffic (ping/pong) + // carries the connection but never surfaces as a read. + unboundedReads bool + // discardReads drops inbound data frames not claimed by a subscription instead of + // queueing them on the read channel, for sessions that never call ReadMessage. + discardReads bool } // WithBufferSize sets the buffer size for read/write/pong channels. @@ -1072,6 +1106,22 @@ func WithLogger(logger zerolog.Logger) Option { return func(o *options) { o.logg // WithTimeout sets the timeout used for dialing and read deadlines. func WithTimeout(d time.Duration) Option { return func(o *options) { o.timeout = d } } +// WithUnboundedReads drops the read pump's per-read timeout so an idle connection is not torn +// down after the read/dial timeout elapses with no inbound data frame. It matches how reads +// behave while a subscription is active. Use it for long-lived sessions driven by control +// frames the read pump never sees as reads: a ping/pong monitor sends ping frames and receives +// pongs (both handled below Read), so without this the connection would be closed one timeout +// after the last data frame. PingPong keeps its own per-ping timeout, so a lost pong is still +// detected; it just no longer kills the connection. +func WithUnboundedReads() Option { return func(o *options) { o.unboundedReads = true } } + +// WithDiscardReads drops inbound data frames not claimed by a subscription instead of +// queueing them on the read channel. Without it, a session that never calls ReadMessage +// (such as a ping/pong monitor) fills the read channel after bufferSize unsolicited frames +// from a chatty peer; the blocked read pump then stops calling Read, which also starves +// pong control-frame processing. Error reads are still delivered to the read channel. +func WithDiscardReads() Option { return func(o *options) { o.discardReads = true } } + // WithCloseGrace bounds how long Close waits for the peer's closing-handshake echo // before forcing the connection shut. Zero or negative fires the teardown immediately // rather than granting a grace window; a peer that echoes in that instant may still