fix: an observer told a tunnel closed could still see its session open

README.md

@@ -96,7 +96,7 @@ The access token and the address it grants come from the authorization server; `
 
 ## Benchmarks
 
-[`bench`](bench) is a separate module. It compares `CONNECT` over HTTP/1.1 and HTTP/2 with a WebSocket tunnel: throughput through one and 64 tunnels, and the latency from dialing to the first relayed byte. CI runs it on the base branch and the change on the same runner. It fails when a result is significantly worse (benchstat statistics, p < 0.05) by more than 10%, or when `CONNECT` over HTTP/1.1 moves fewer bytes per second than the WebSocket tunnel. The WebSocket code lives only in that module, which never runs in FIPS 140-3 only mode: the WebSocket handshake needs SHA-1.
+[`bench`](bench) is a separate module. It compares `CONNECT` over HTTP/1.1 and HTTP/2 with a WebSocket tunnel: throughput through one and 64 tunnels, and the latency from dialing to the first relayed byte. CI runs it on the base branch and the change on the same runner. It fails when a result is significantly worse by more than 10%, or when `CONNECT` over HTTP/1.1 moves fewer bytes per second than the WebSocket tunnel. Significance uses benchstat's Mann-Whitney U test with Holm's correction across all comparisons, so noise alone fails a run at most 5% of the time. The WebSocket code lives only in that module, which never runs in FIPS 140-3 only mode: the WebSocket handshake needs SHA-1.
 
 ## Conformance
 

bench/cmd/gate/main.go

@@ -1,6 +1,8 @@
 // Command gate compares benchmark results and fails on a significant regression. It fails when a
 // benchmark is significantly worse than on the base branch by more than -max-regression, and when
-// CONNECT over HTTP/1.1 moves fewer bytes per second than the WebSocket baseline.
+// CONNECT over HTTP/1.1 moves fewer bytes per second than the WebSocket baseline. Significance is
+// judged across each family of comparisons with Holm's step-down procedure, so the chance that any
+// comparison fails by noise alone stays at alpha however many benchmarks there are.
 package main
 
 import (
@@ -108,27 +110,37 @@ func names[V any](m map[string]V) []string {
 	return keys
 }
 
+type measure struct {
+	name, unit string
+	lower      bool
+	cmp        comparison
+}
+
 func regressions(base, head samples, maxRegression float64) (report, failures []string) {
+	var ms []measure
 	for _, name := range names(head) {
 		for _, unit := range names(head[name]) {
 			lower, checked := lowerIsBetter[unit]
 			prior := base[name][unit]
 			if !checked || len(prior) == 0 {
 				continue
 			}
-			cmp := compare(prior, head[name][unit])
-			worse := cmp.change
-			if !lower {
-				worse = -worse
-			}
-			verdict := "~"
-			if cmp.significant {
-				verdict = fmt.Sprintf("%+.1f%%", cmp.change*100)
-			}
-			report = append(report, fmt.Sprintf("%-60s %-10s %12.4g -> %12.4g  %s (p=%.3f)", name, unit, cmp.base, cmp.head, verdict, cmp.p))
-			if cmp.significant && (worse > maxRegression || (cmp.base == 0 && cmp.head > 0 && lower)) {
-				failures = append(failures, fmt.Sprintf("%s %s regressed %+.1f%%", name, unit, cmp.change*100))
-			}
+			ms = append(ms, measure{name: name, unit: unit, lower: lower, cmp: compare(prior, head[name][unit])})
+		}
+	}
+	judge(ms)
+	for _, m := range ms {
+		worse := m.cmp.change
+		if !m.lower {
+			worse = -worse
+		}
+		verdict := "~"
+		if m.cmp.significant {
+			verdict = fmt.Sprintf("%+.1f%%", m.cmp.change*100)
+		}
+		report = append(report, fmt.Sprintf("%-60s %-10s %12.4g -> %12.4g  %s (p=%.3f)", m.name, m.unit, m.cmp.base, m.cmp.head, verdict, m.cmp.p))
+		if m.cmp.significant && (worse > maxRegression || (m.cmp.base == 0 && m.cmp.head > 0 && m.lower)) {
+			failures = append(failures, fmt.Sprintf("%s %s regressed %+.1f%%", m.name, m.unit, m.cmp.change*100))
 		}
 	}
 	return report, failures
@@ -141,6 +153,24 @@ type comparison struct {
 	significant bool
 }
 
+// judge marks which comparisons are significant at alpha by Holm's step-down procedure: in order of
+// increasing p, the k-th smallest of m p-values must be at most alpha/(m-k+1), and testing stops at
+// the first that is not. Equal centers are never significant.
+func judge(ms []measure) {
+	order := make([]int, len(ms))
+	for i := range order {
+		order[i] = i
+	}
+	sort.SliceStable(order, func(a, b int) bool { return ms[order[a]].cmp.p < ms[order[b]].cmp.p })
+	for rank, i := range order {
+		c := &ms[i].cmp
+		if c.p > alpha/float64(len(ms)-rank) {
+			return
+		}
+		c.significant = c.base != c.head
+	}
+}
+
 func compare(base, head []float64) comparison {
 	b := benchmath.NewSample(base, &benchmath.DefaultThresholds)
 	h := benchmath.NewSample(head, &benchmath.DefaultThresholds)
@@ -152,13 +182,13 @@ func compare(base, head []float64) comparison {
 	if c.base != 0 {
 		c.change = (c.head - c.base) / c.base
 	}
-	c.significant = c.p < alpha && c.base != c.head
 	return c
 }
 
 // belowBaseline fails when CONNECT over HTTP/1.1 is significantly slower than the WebSocket tunnel
 // at the same number of tunnels.
 func belowBaseline(head samples) (report, failures []string) {
+	var ms []measure
 	for _, name := range names(head) {
 		if !strings.Contains(name, "transport=websocket") {
 			continue
@@ -168,10 +198,13 @@ func belowBaseline(head samples) (report, failures []string) {
 		if len(ws) == 0 || len(connect) == 0 {
 			continue
 		}
-		cmp := compare(ws, connect)
-		report = append(report, fmt.Sprintf("%-60s CONNECT/WebSocket throughput %.2fx (p=%.3f)", name, ratio(cmp), cmp.p))
-		if cmp.significant && cmp.head < cmp.base {
-			failures = append(failures, name+": CONNECT is slower than the WebSocket baseline")
+		ms = append(ms, measure{name: name, cmp: compare(ws, connect)})
+	}
+	judge(ms)
+	for _, m := range ms {
+		report = append(report, fmt.Sprintf("%-60s CONNECT/WebSocket throughput %.2fx (p=%.3f)", m.name, ratio(m.cmp), m.cmp.p))
+		if m.cmp.significant && m.cmp.head < m.cmp.base {
+			failures = append(failures, m.name+": CONNECT is slower than the WebSocket baseline")
 		}
 	}
 	return report, failures

bench/cmd/gate/main_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"
 
@@ -133,3 +134,25 @@ func TestGateInputErrors(t *testing.T) {
 		}
 	}
 }
+
+func TestJudgeControlsTheFamilyWiseErrorRate(t *testing.T) {
+	t.Parallel()
+	ms := []measure{
+		{name: "a", cmp: comparison{base: 1, head: 2, p: 0.04}},
+		{name: "b", cmp: comparison{base: 1, head: 2, p: 0.001}},
+		{name: "c", cmp: comparison{base: 1, head: 2, p: 0.03}},
+		{name: "d", cmp: comparison{base: 1, head: 1, p: 0.0001}},
+		{name: "e", cmp: comparison{base: 1, head: 2, p: 0.5}},
+	}
+	judge(ms)
+	var got []string
+	for _, m := range ms {
+		if m.cmp.significant {
+			got = append(got, m.name)
+		}
+	}
+	if !slices.Equal(got, []string{"b"}) {
+		t.Fatalf("significant %v, want [b]: 0.0001 passes 0.05/5 but its centers are equal, 0.001 passes 0.05/4, and 0.03 fails 0.05/3", got)
+	}
+	judge(nil)
+}

release-please-config.json

@@ -4,6 +4,7 @@
   "include-component-in-tag": false,
   "include-v-in-tag": true,
   "bump-minor-pre-major": true,
+  "initial-version": "0.1.0",
   "packages": {
     ".": {
       "extra-files": [

server/handler.go

@@ -244,6 +244,7 @@ func (s *Server) relay(ctx context.Context, w http.ResponseWriter, r *http.Reque
 			s.cfg.Logger.LogAttrs(ctx, slog.LevelError, "tunnel.panic", slog.Uint64("tunnel", t.ID), slog.Any("recovered", recovered))
 		},
 	})
+	stop()
 	c := Closed{
 		Tunnel:     t,
 		Reason:     closeReason(err),

server/lifecycle_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"io"
 	"net/http"
+	"slices"
 	"sync"
 	"testing"
 	"time"
@@ -20,6 +21,9 @@ type recorder struct {
 	closed   []server.Closed
 	rejected []server.Rejection
 	done     chan server.Closed
+	// sessions, when set, is sampled as each tunnel is reported closed.
+	sessions      func() int
+	openAtClosure []int
 }
 
 func newRecorder() *recorder { return &recorder{done: make(chan server.Closed, 16)} }
@@ -33,6 +37,9 @@ func (r *recorder) TunnelOpened(_ context.Context, t *server.Tunnel) {
 func (r *recorder) TunnelClosed(_ context.Context, c *server.Closed) {
 	r.mu.Lock()
 	r.closed = append(r.closed, *c)
+	if r.sessions != nil {
+		r.openAtClosure = append(r.openAtClosure, r.sessions())
+	}
 	r.mu.Unlock()
 	r.done <- *c
 }
@@ -59,6 +66,7 @@ func TestRevocationClosesTunnel(t *testing.T) {
 	protocols(t, func(t *testing.T, http2 bool) {
 		rec := newRecorder()
 		e := relaytest.Start(t, &relaytest.Options{Observer: rec, SessionInterval: 20 * time.Millisecond})
+		rec.sessions = e.Sessions.Open
 		g := e.Grant(e.Echo)
 		resp, conn := e.Connect(t.Context(), t, e.Echo.String(), e.AS.Issue(g), http2)
 		if resp.StatusCode != http.StatusOK {
@@ -80,8 +88,10 @@ func TestRevocationClosesTunnel(t *testing.T) {
 		if _, err := io.Copy(io.Discard, conn); err == nil && http2 {
 			t.Fatal("HTTP/2 stream ended cleanly after revocation")
 		}
-		if e.Sessions.Open() != 0 {
-			t.Fatal("session still registered")
+		rec.mu.Lock()
+		defer rec.mu.Unlock()
+		if !slices.Equal(rec.openAtClosure, []int{0}) {
+			t.Fatalf("sessions open when the tunnel was reported closed: %v", rec.openAtClosure)
 		}
 	})
 }