fix: release queued tests on cancellation

README.md

@@ -9,6 +9,8 @@ NetStable 是一个独立的 Go Web 测速项目，用于测试“打开网页
 - 启动时可设置程序理论最高带宽，例如 `-bandwidth-limit-mbps 30` 将测试流量限制到 30 Mbps。
 - 页面展示当前设置的理论最高限值。
 - 一次测试按阶段执行：先连续下载指定时长，再连续上传同样时长。页面固定提供 `15s` 和 `30s` 两档，默认 `30s`。
+- 手动停止、刷新页面或 CLI 中断时会取消当前会话并释放队列，后续用户会自动前移。
+- 下载/上传峰值使用连续样本的持续吞吐口径，避免单个短请求计时误差造成虚高峰值。
 - 保存每次测试的完整样本曲线、摘要、脱敏后的用户 IP、地区和运营商信息；ISP 缺失时显示“未知运营商”。
 - 所有历史测试用户数据通过网页和 `/api/records` 提供。
 - 历史时序图展示不同时段的平均下载速度变化。

cmd/netstable/main.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"os/signal"
 	"strings"
 	"time"
 
@@ -150,7 +151,9 @@ func runClient(args []string) {
 			printSample(os.Stderr, sample)
 		}
 	}
-	summary, err := appclient.Run(context.Background(), options)
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer stop()
+	summary, err := appclient.Run(ctx, options)
 	if err != nil {
 		_, _ = fmt.Fprintf(os.Stderr, "client failed: %v\n", err)
 		os.Exit(1)
@@ -179,7 +182,9 @@ func runIperf3Client(serverURL string, target string, durationSeconds int, timeo
 			printSample(os.Stderr, sample)
 		}
 	}
-	summary, err := appclient.RunIperf3(context.Background(), options)
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer stop()
+	summary, err := appclient.RunIperf3(ctx, options)
 	if err != nil {
 		_, _ = fmt.Fprintf(os.Stderr, "iperf3 client failed: %v\n", err)
 		os.Exit(1)
@@ -200,7 +205,7 @@ func printSample(out *os.File, sample probe.Sample) {
 	case "latency":
 		_, _ = fmt.Fprintf(out, "%s %s %.1f ms %s\n", sample.Kind, status, sample.LatencyMS, sample.Error)
 	case "download", "upload":
-		_, _ = fmt.Fprintf(out, "%s %s %.2f Mbps %d bytes %s\n", sample.Kind, status, sample.Mbps, sample.Bytes, sample.Error)
+		_, _ = fmt.Fprintf(out, "%s %s 瞬时 %.2f Mbps %d bytes %s\n", sample.Kind, status, sample.Mbps, sample.Bytes, sample.Error)
 	default:
 		_, _ = fmt.Fprintf(out, "%s %s %s\n", sample.Kind, status, sample.Error)
 	}

internal/client/client.go

@@ -41,6 +41,7 @@ type createResponse struct {
 	DownloadURL   string `json:"downloadUrl"`
 	UploadURL     string `json:"uploadUrl"`
 	CompleteURL   string `json:"completeUrl"`
+	CancelURL     string `json:"cancelUrl"`
 }
 
 type completeResponse struct {
@@ -68,6 +69,15 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
 	if err != nil {
 		return probe.Summary{}, err
 	}
+	complete := false
+	defer func() {
+		if !complete {
+			cancelCtx, cancel := context.WithTimeout(context.Background(), cancelRequestTimeout(options.Timeout))
+			defer cancel()
+			_ = cancelSession(cancelCtx, &httpClient, server, session.CancelURL)
+		}
+	}()
+
 	if err := waitForQueue(ctx, &httpClient, server, session, options); err != nil {
 		return probe.Summary{}, err
 	}
@@ -96,7 +106,12 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
 	record(measureLatency(ctx, &httpClient, server))
 	finishedAt := time.Now().UTC()
 
-	return completeSession(ctx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, samples)
+	summary, err := completeSession(ctx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, samples)
+	if err != nil {
+		return probe.Summary{}, err
+	}
+	complete = true
+	return summary, nil
 }
 
 func normalizeOptions(options Options) (Options, *url.URL, error) {
@@ -345,6 +360,32 @@ func completeSession(ctx context.Context, httpClient *http.Client, server *url.U
 	return completed.Summary, nil
 }
 
+func cancelSession(ctx context.Context, httpClient *http.Client, server *url.URL, path string) error {
+	if strings.TrimSpace(path) == "" {
+		return nil
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, resolve(server, path), nil)
+	if err != nil {
+		return err
+	}
+	response, err := httpClient.Do(request)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNotFound {
+		return nil
+	}
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		message := decodeError(response.Body)
+		if message == "" {
+			message = response.Status
+		}
+		return fmt.Errorf("cancel test session: %s", message)
+	}
+	return nil
+}
+
 func resolve(server *url.URL, path string) string {
 	relative, err := url.Parse(path)
 	if err != nil {
@@ -400,6 +441,16 @@ func transferTimeout(timeout time.Duration) time.Duration {
 	return timeout
 }
 
+func cancelRequestTimeout(timeout time.Duration) time.Duration {
+	if timeout <= 0 {
+		return 3 * time.Second
+	}
+	if timeout > 5*time.Second {
+		return 5 * time.Second
+	}
+	return timeout
+}
+
 func cacheToken() string {
 	return fmt.Sprint(time.Now().UnixNano())
 }

internal/client/client_test.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/url"
 	"os"
@@ -83,6 +84,53 @@ func TestRunWaitsInQueueUntilBusySessionCompletes(t *testing.T) {
 	}
 }
 
+func TestRunCancelsQueuedSessionWhenContextStops(t *testing.T) {
+	store := &memoryStore{}
+	server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
+	httpServer := httptest.NewServer(server.Handler())
+	defer httpServer.Close()
+
+	busy := createBusySession(t, httpServer.URL)
+	ctx, cancel := context.WithCancel(context.Background())
+	_, err := Run(ctx, Options{
+		ServerURL:     httpServer.URL,
+		Duration:      20 * time.Millisecond,
+		Timeout:       time.Second,
+		DownloadBytes: 1024,
+		UploadBytes:   512,
+		QueueInterval: time.Hour,
+		OnQueueWaiting: func(message string) {
+			cancel()
+		},
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run error = %v, want context canceled", err)
+	}
+
+	completeBusySession(t, httpServer.URL, busy)
+
+	serverURL, err := url.Parse(httpServer.URL)
+	if err != nil {
+		t.Fatalf("parse server URL: %v", err)
+	}
+	session, err := createSession(context.Background(), &http.Client{Timeout: time.Second}, serverURL, Options{
+		ServerURL:      httpServer.URL,
+		Target:         serverURL.Host,
+		Duration:       time.Second,
+		Timeout:        time.Second,
+		DownloadBytes:  1024,
+		UploadBytes:    512,
+		QueueInterval:  time.Millisecond,
+		OnQueueWaiting: nil,
+	})
+	if err != nil {
+		t.Fatalf("create next session: %v", err)
+	}
+	if session.Status != "ready" || session.QueuePosition != 0 {
+		t.Fatalf("next session = %#v, want ready after canceled queued session was removed", session)
+	}
+}
+
 func TestRunIperf3CompletesOneOffServerSession(t *testing.T) {
 	store := &memoryStore{}
 	server := appweb.New(appweb.Options{Store: store, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})

internal/client/iperf3.go

@@ -69,7 +69,9 @@ func RunIperf3(ctx context.Context, options Iperf3Options) (probe.Summary, error
 			Success: false,
 			Error:   err.Error(),
 		}
-		if _, completeErr := completeIperf3Session(ctx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, []probe.Sample{failedSample}); completeErr != nil {
+		completeCtx, cancel := context.WithTimeout(context.Background(), cancelRequestTimeout(options.Timeout))
+		defer cancel()
+		if _, completeErr := completeIperf3Session(completeCtx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, []probe.Sample{failedSample}); completeErr != nil {
 			return probe.Summary{}, fmt.Errorf("%w; also failed to complete iperf3 session: %v", err, completeErr)
 		}
 		return probe.Summary{}, err

internal/web/server.go

@@ -16,6 +16,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"runtime"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -134,6 +135,7 @@ type createTestResponse struct {
 	DownloadURL   string `json:"downloadUrl"`
 	UploadURL     string `json:"uploadUrl"`
 	CompleteURL   string `json:"completeUrl"`
+	CancelURL     string `json:"cancelUrl"`
 }
 
 type queueStatusResponse struct {
@@ -697,6 +699,7 @@ func (s *Server) handleCreateTest(w http.ResponseWriter, r *http.Request) {
 		DownloadURL:   "/api/download?testId=" + request.ID,
 		UploadURL:     "/api/upload?testId=" + request.ID,
 		CompleteURL:   "/api/tests/" + request.ID + "/complete",
+		CancelURL:     "/api/tests/" + request.ID + "/cancel",
 	})
 }
 
@@ -708,6 +711,8 @@ func (s *Server) handleTestRoute(w http.ResponseWriter, r *http.Request) {
 		s.handleTestEvents(w, r)
 	case strings.HasSuffix(r.URL.Path, "/complete"):
 		s.handleCompleteTest(w, r)
+	case strings.HasSuffix(r.URL.Path, "/cancel"):
+		s.handleCancelTest(w, r)
 	default:
 		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
 	}
@@ -777,6 +782,23 @@ func (s *Server) handleCompleteTest(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, completeTestResponse{Summary: summary})
 }
 
+func (s *Server) handleCancelTest(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
+		return
+	}
+	id, ok := cancelSessionID(r.URL.Path)
+	if !ok {
+		writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
+		return
+	}
+	if _, exists := s.takeSession(id); !exists {
+		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"status": "canceled"})
+}
+
 func (s *Server) handleTestEvents(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet {
 		writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
@@ -1171,6 +1193,15 @@ func completeSessionID(path string) (string, bool) {
 	return id, id != ""
 }
 
+func cancelSessionID(path string) (string, bool) {
+	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/cancel") {
+		return "", false
+	}
+	id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/cancel")
+	id = strings.Trim(id, "/")
+	return id, id != ""
+}
+
 func queueSessionID(path string) (string, bool) {
 	if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/queue") {
 		return "", false
@@ -1277,10 +1308,6 @@ func summarizeBrowserSamples(request TestRequest, raw completeTestRequest) probe
 	var lastLatency float64
 	var jitterSum float64
 	var jitterPairs int
-	var downloadSum float64
-	var downloadCount int
-	var uploadSum float64
-	var uploadCount int
 
 	for _, sample := range raw.Samples {
 		if !sample.Success {
@@ -1304,18 +1331,6 @@ func summarizeBrowserSamples(request TestRequest, raw completeTestRequest) probe
 				jitterPairs++
 			}
 			lastLatency = latency
-		case "download":
-			downloadSum += sample.Mbps
-			downloadCount++
-			if sample.Mbps > summary.MaxDownloadMbps {
-				summary.MaxDownloadMbps = sample.Mbps
-			}
-		case "upload":
-			uploadSum += sample.Mbps
-			uploadCount++
-			if sample.Mbps > summary.MaxUploadMbps {
-				summary.MaxUploadMbps = sample.Mbps
-			}
 		}
 	}
 
@@ -1330,16 +1345,153 @@ func summarizeBrowserSamples(request TestRequest, raw completeTestRequest) probe
 	if jitterPairs > 0 {
 		summary.JitterMS = round2(jitterSum / float64(jitterPairs))
 	}
+	avgDownload, maxDownload, downloadCount := transferStats(raw.Samples, "download")
 	if downloadCount > 0 {
-		summary.AvgDownloadMbps = round2(downloadSum / float64(downloadCount))
+		summary.AvgDownloadMbps = round2(avgDownload)
+		summary.MaxDownloadMbps = round2(maxDownload)
 	}
+	avgUpload, maxUpload, uploadCount := transferStats(raw.Samples, "upload")
 	if uploadCount > 0 {
-		summary.AvgUploadMbps = round2(uploadSum / float64(uploadCount))
+		summary.AvgUploadMbps = round2(avgUpload)
+		summary.MaxUploadMbps = round2(maxUpload)
 	}
 	summary.Score = browserScore(summary)
 	return summary
 }
 
+type transferPoint struct {
+	at        time.Time
+	bytes     int64
+	mbps      float64
+	latencyMS float64
+}
+
+type transferSegment struct {
+	bits    float64
+	seconds float64
+}
+
+const sustainedPeakWindowSeconds = 5
+
+func transferStats(samples []probe.Sample, kind string) (avgMbps float64, maxMbps float64, count int) {
+	var points []transferPoint
+	for _, sample := range samples {
+		if !sample.Success || sample.Kind != kind {
+			continue
+		}
+		count++
+		if sample.Bytes <= 0 && sample.Mbps <= 0 {
+			continue
+		}
+		points = append(points, transferPoint{
+			at:        sample.At,
+			bytes:     sample.Bytes,
+			mbps:      sample.Mbps,
+			latencyMS: sample.LatencyMS,
+		})
+	}
+	if len(points) == 0 {
+		return 0, 0, count
+	}
+	sort.Slice(points, func(i, j int) bool {
+		return points[i].at.Before(points[j].at)
+	})
+
+	var totalBits float64
+	var totalSeconds float64
+	var fallbackSum float64
+	var fallbackCount int
+	var segments []transferSegment
+	for index, point := range points {
+		seconds := transferSampleSeconds(point)
+		if seconds <= 0 {
+			seconds = transferIntervalSeconds(points, index)
+		}
+
+		rate := point.mbps
+		if point.bytes > 0 && seconds > 0 {
+			bits := float64(point.bytes) * 8
+			rate = bits / seconds / 1_000_000
+			totalBits += bits
+			totalSeconds += seconds
+			segments = append(segments, transferSegment{bits: bits, seconds: seconds})
+		} else if rate > 0 {
+			fallbackSum += rate
+			fallbackCount++
+		}
+	}
+
+	if totalSeconds > 0 {
+		avgMbps = totalBits / totalSeconds / 1_000_000
+	} else if fallbackCount > 0 {
+		avgMbps = fallbackSum / float64(fallbackCount)
+	}
+	maxMbps = sustainedPeakMbps(segments, avgMbps)
+	if maxMbps == 0 && fallbackCount > 0 {
+		maxMbps = avgMbps
+	}
+	return avgMbps, maxMbps, count
+}
+
+func sustainedPeakMbps(segments []transferSegment, fallbackAvg float64) float64 {
+	var totalSeconds float64
+	for _, segment := range segments {
+		totalSeconds += segment.seconds
+	}
+	if totalSeconds <= 0 {
+		return 0
+	}
+	if totalSeconds < sustainedPeakWindowSeconds {
+		return fallbackAvg
+	}
+
+	var maxMbps float64
+	for start := range segments {
+		var bits float64
+		var seconds float64
+		for end := start; end < len(segments); end++ {
+			bits += segments[end].bits
+			seconds += segments[end].seconds
+			if seconds >= sustainedPeakWindowSeconds {
+				mbps := bits / seconds / 1_000_000
+				if mbps > maxMbps {
+					maxMbps = mbps
+				}
+			}
+		}
+	}
+	if maxMbps == 0 {
+		return fallbackAvg
+	}
+	return maxMbps
+}
+
+func transferIntervalSeconds(points []transferPoint, index int) float64 {
+	if len(points) < 2 || points[index].at.IsZero() {
+		return 0
+	}
+	if index == 0 {
+		if points[1].at.IsZero() {
+			return 0
+		}
+		return points[1].at.Sub(points[0].at).Seconds()
+	}
+	if points[index-1].at.IsZero() {
+		return 0
+	}
+	return points[index].at.Sub(points[index-1].at).Seconds()
+}
+
+func transferSampleSeconds(point transferPoint) float64 {
+	if point.latencyMS > 0 {
+		return point.latencyMS / 1000
+	}
+	if point.bytes > 0 && point.mbps > 0 {
+		return float64(point.bytes) * 8 / (point.mbps * 1_000_000)
+	}
+	return 0
+}
+
 func browserScore(summary probe.Summary) float64 {
 	if summary.Samples == 0 {
 		return 0

internal/web/server_test.go

@@ -294,6 +294,9 @@ func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
 	if payload.CompleteURL != "/api/tests/"+payload.ID+"/complete" {
 		t.Fatalf("CompleteURL = %q, want /api/tests/%s/complete", payload.CompleteURL, payload.ID)
 	}
+	if payload.CancelURL != "/api/tests/"+payload.ID+"/cancel" {
+		t.Fatalf("CancelURL = %q, want /api/tests/%s/cancel", payload.CancelURL, payload.ID)
+	}
 }
 
 func TestCreateTestQueuesWhenAnotherSessionIsActive(t *testing.T) {
@@ -392,6 +395,120 @@ func TestQueuedTestPromotesAfterActiveSessionCompletes(t *testing.T) {
 	}
 }
 
+func TestCancelActiveTestPromotesQueuedSession(t *testing.T) {
+	server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
+
+	firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
+	firstRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(firstRes, firstReq)
+	if firstRes.Code != http.StatusAccepted {
+		t.Fatalf("first status = %d, want 202; body=%s", firstRes.Code, firstRes.Body.String())
+	}
+	var first createTestResponse
+	if err := json.Unmarshal(firstRes.Body.Bytes(), &first); err != nil {
+		t.Fatalf("decode first response: %v", err)
+	}
+
+	secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
+	secondRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(secondRes, secondReq)
+	if secondRes.Code != http.StatusAccepted {
+		t.Fatalf("second status = %d, want 202; body=%s", secondRes.Code, secondRes.Body.String())
+	}
+	var second createTestResponse
+	if err := json.Unmarshal(secondRes.Body.Bytes(), &second); err != nil {
+		t.Fatalf("decode second response: %v", err)
+	}
+	if second.Status != "queued" {
+		t.Fatalf("second status = %q, want queued", second.Status)
+	}
+
+	cancelReq := httptest.NewRequest(http.MethodPost, first.CancelURL, nil)
+	cancelRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(cancelRes, cancelReq)
+	if cancelRes.Code != http.StatusOK {
+		t.Fatalf("cancel status = %d, want 200; body=%s", cancelRes.Code, cancelRes.Body.String())
+	}
+
+	queueReq := httptest.NewRequest(http.MethodGet, second.QueueURL, nil)
+	queueRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(queueRes, queueReq)
+	if queueRes.Code != http.StatusOK {
+		t.Fatalf("queue status = %d, want 200; body=%s", queueRes.Code, queueRes.Body.String())
+	}
+	var ready queueStatusResponse
+	if err := json.Unmarshal(queueRes.Body.Bytes(), &ready); err != nil {
+		t.Fatalf("decode queue status: %v", err)
+	}
+	if ready.Status != "ready" || ready.QueuePosition != 0 {
+		t.Fatalf("queue status = %#v, want promoted ready session", ready)
+	}
+}
+
+func TestCancelQueuedTestRemovesItAndUpdatesPositions(t *testing.T) {
+	server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
+
+	firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
+	firstRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(firstRes, firstReq)
+	if firstRes.Code != http.StatusAccepted {
+		t.Fatalf("first status = %d, want 202; body=%s", firstRes.Code, firstRes.Body.String())
+	}
+
+	secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
+	secondRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(secondRes, secondReq)
+	if secondRes.Code != http.StatusAccepted {
+		t.Fatalf("second status = %d, want 202; body=%s", secondRes.Code, secondRes.Body.String())
+	}
+	var second createTestResponse
+	if err := json.Unmarshal(secondRes.Body.Bytes(), &second); err != nil {
+		t.Fatalf("decode second response: %v", err)
+	}
+
+	thirdReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.net","durationSeconds":1}`))
+	thirdRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(thirdRes, thirdReq)
+	if thirdRes.Code != http.StatusAccepted {
+		t.Fatalf("third status = %d, want 202; body=%s", thirdRes.Code, thirdRes.Body.String())
+	}
+	var third createTestResponse
+	if err := json.Unmarshal(thirdRes.Body.Bytes(), &third); err != nil {
+		t.Fatalf("decode third response: %v", err)
+	}
+	if third.QueuePosition != 2 {
+		t.Fatalf("third position = %d, want 2 before cancel", third.QueuePosition)
+	}
+
+	cancelReq := httptest.NewRequest(http.MethodPost, second.CancelURL, nil)
+	cancelRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(cancelRes, cancelReq)
+	if cancelRes.Code != http.StatusOK {
+		t.Fatalf("cancel status = %d, want 200; body=%s", cancelRes.Code, cancelRes.Body.String())
+	}
+
+	secondQueueReq := httptest.NewRequest(http.MethodGet, second.QueueURL, nil)
+	secondQueueRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(secondQueueRes, secondQueueReq)
+	if secondQueueRes.Code != http.StatusNotFound {
+		t.Fatalf("canceled queue status = %d, want 404; body=%s", secondQueueRes.Code, secondQueueRes.Body.String())
+	}
+
+	thirdQueueReq := httptest.NewRequest(http.MethodGet, third.QueueURL, nil)
+	thirdQueueRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(thirdQueueRes, thirdQueueReq)
+	if thirdQueueRes.Code != http.StatusOK {
+		t.Fatalf("third queue status = %d, want 200; body=%s", thirdQueueRes.Code, thirdQueueRes.Body.String())
+	}
+	var queued queueStatusResponse
+	if err := json.Unmarshal(thirdQueueRes.Body.Bytes(), &queued); err != nil {
+		t.Fatalf("decode third queue status: %v", err)
+	}
+	if queued.Status != "queued" || queued.QueuePosition != 1 {
+		t.Fatalf("third queue status = %#v, want queued position 1 after cancel", queued)
+	}
+}
+
 func TestCreateIperf3SessionStartsOneOffServerAndUsesGlobalLock(t *testing.T) {
 	server := New(Options{Store: &memoryStore{}, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})
 
@@ -662,6 +779,76 @@ func TestCompleteEndpointPersistsBrowserSpeedRecordAndReleasesLock(t *testing.T)
 	}
 }
 
+func TestBrowserSummaryUsesSustainedTransferRatesForMax(t *testing.T) {
+	start := time.Date(2026, 6, 10, 8, 0, 0, 0, time.UTC)
+	request := TestRequest{
+		ID:        "test-1",
+		Target:    probe.Target{Address: "example.com:443"},
+		CreatedAt: start,
+	}
+	raw := completeTestRequest{
+		StartedAt:  start,
+		FinishedAt: start.Add(5 * time.Second),
+		Samples: []probe.Sample{
+			{At: start.Add(1 * time.Second), Kind: "download", Success: true, LatencyMS: 1000, Bytes: 5_000_000, Mbps: 40},
+			{At: start.Add(2 * time.Second), Kind: "download", Success: true, LatencyMS: 1000, Bytes: 5_000_000, Mbps: 1000},
+			{At: start.Add(3 * time.Second), Kind: "download", Success: true, LatencyMS: 1000, Bytes: 5_000_000, Mbps: 40},
+			{At: start.Add(4 * time.Second), Kind: "download", Success: true, LatencyMS: 1000, Bytes: 5_000_000, Mbps: 40},
+		},
+	}
+
+	summary := summarizeBrowserSamples(request, raw)
+
+	if summary.AvgDownloadMbps != 40 {
+		t.Fatalf("AvgDownloadMbps = %.2f, want sustained 40 Mbps", summary.AvgDownloadMbps)
+	}
+	if summary.MaxDownloadMbps != 40 {
+		t.Fatalf("MaxDownloadMbps = %.2f, want sustained 40 Mbps instead of single-sample spike", summary.MaxDownloadMbps)
+	}
+}
+
+func TestBrowserSummaryDoesNotTreatShortBurstsAsSustainedMax(t *testing.T) {
+	start := time.Date(2026, 6, 10, 8, 0, 0, 0, time.UTC)
+	var samples []probe.Sample
+	elapsed := time.Duration(0)
+	for i := 0; i < 10; i++ {
+		latency := 50 * time.Millisecond
+		mbps := 671.09
+		if i%2 == 1 {
+			latency = 1750 * time.Millisecond
+			mbps = 19.17
+		}
+		elapsed += latency
+		samples = append(samples, probe.Sample{
+			At:        start.Add(elapsed),
+			Kind:      "download",
+			Success:   true,
+			LatencyMS: float64(latency.Milliseconds()),
+			Bytes:     4 * 1024 * 1024,
+			Mbps:      mbps,
+		})
+	}
+	request := TestRequest{
+		ID:        "test-1",
+		Target:    probe.Target{Address: "example.com:443"},
+		CreatedAt: start,
+	}
+	raw := completeTestRequest{
+		StartedAt:  start,
+		FinishedAt: start.Add(elapsed),
+		Samples:    samples,
+	}
+
+	summary := summarizeBrowserSamples(request, raw)
+
+	if summary.MaxDownloadMbps > 60 {
+		t.Fatalf("MaxDownloadMbps = %.2f, want sustained max below short burst rates", summary.MaxDownloadMbps)
+	}
+	if summary.AvgDownloadMbps < 35 || summary.AvgDownloadMbps > 39 {
+		t.Fatalf("AvgDownloadMbps = %.2f, want long-run average around 37 Mbps", summary.AvgDownloadMbps)
+	}
+}
+
 func TestSSEUnknownSessionReturnsNotFound(t *testing.T) {
 	server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
 

web/assets_test.go

@@ -83,6 +83,37 @@ func TestStaticAssetsIncludeQueueWaitingBehavior(t *testing.T) {
 	}
 }
 
+func TestStaticAssetsIncludeManualCancelBehavior(t *testing.T) {
+	html, err := readAsset("index.html")
+	if err != nil {
+		t.Fatalf("read index.html: %v", err)
+	}
+	if !contains(html, `id="stopButton"`) {
+		t.Fatalf("index.html missing manual stop control")
+	}
+
+	app, err := readAsset("app.js")
+	if err != nil {
+		t.Fatalf("read app.js: %v", err)
+	}
+	if !contains(app, "cancelUrl") ||
+		!contains(app, "cancelCurrentSession") ||
+		!contains(app, "beforeunload") ||
+		!contains(app, "navigator.sendBeacon") {
+		t.Fatalf("app.js missing manual/unload cancel behavior")
+	}
+}
+
+func TestStaticAssetsLabelTransferSamplesAsInstantaneous(t *testing.T) {
+	app, err := readAsset("app.js")
+	if err != nil {
+		t.Fatalf("read app.js: %v", err)
+	}
+	if !contains(app, "瞬时") {
+		t.Fatalf("app.js missing instantaneous transfer sample label")
+	}
+}
+
 func contains(text, pattern string) bool {
 	return strings.Contains(text, pattern)
 }

web/static/app.js

@@ -3,11 +3,15 @@ const state = {
   samples: [],
   records: [],
   clientCommand: "",
+  currentSession: null,
+  abortController: null,
+  cancelRequested: false,
 };
 
 const els = {
   form: document.getElementById("testForm"),
   startButton: document.getElementById("startButton"),
+  stopButton: document.getElementById("stopButton"),
   copyCommandButton: document.getElementById("copyCommandButton"),
   refreshButton: document.getElementById("refreshButton"),
   notice: document.getElementById("notice"),
@@ -30,6 +34,7 @@ const DOWNLOAD_BYTES = 4 * 1024 * 1024;
 const UPLOAD_BYTES = 2 * 1024 * 1024;
 
 els.form.addEventListener("submit", startTest);
+els.stopButton.addEventListener("click", stopCurrentTest);
 els.copyCommandButton.addEventListener("click", copyClientCommand);
 document.querySelectorAll('input[name="durationSeconds"]').forEach(input => {
   input.addEventListener("change", updateClientCommand);
@@ -39,6 +44,10 @@ window.addEventListener("resize", () => {
   drawLiveChart();
   drawHistoryChart();
 });
+window.addEventListener("beforeunload", () => {
+  state.cancelRequested = true;
+  cancelCurrentSession({ beacon: true });
+});
 
 initializeDefaultTarget();
 updateClientCommand();
@@ -92,14 +101,51 @@ async function copyClientCommand() {
   }
 }
 
+async function stopCurrentTest() {
+  state.cancelRequested = true;
+  if (state.abortController) {
+    state.abortController.abort();
+  }
+  await cancelCurrentSession();
+  closeSource();
+  setServerState("idle", "空闲");
+  setNotice("测试已停止，队列已释放。");
+  setTestingControls(false);
+}
+
+async function cancelCurrentSession(options = {}) {
+  const session = state.currentSession;
+  if (!session || !session.cancelUrl) {
+    return false;
+  }
+  state.currentSession = null;
+  const url = `${session.cancelUrl}?cache=${Date.now()}`;
+  if (options.beacon && navigator.sendBeacon) {
+    return navigator.sendBeacon(url, new Blob([], { type: "text/plain" }));
+  }
+  try {
+    const response = await fetch(url, {
+      method: "POST",
+      cache: "no-store",
+      keepalive: true,
+    });
+    return response.ok || response.status === 404;
+  } catch (error) {
+    return false;
+  }
+}
+
 async function startTest(event) {
   event.preventDefault();
   closeSource();
   state.samples = [];
+  state.currentSession = null;
+  state.cancelRequested = false;
+  state.abortController = new AbortController();
   els.eventLog.textContent = "";
   setNotice("");
   setServerState("running", "运行中");
-  els.startButton.disabled = true;
+  setTestingControls(true);
   updateLiveMetrics();
   drawLiveChart();
 
@@ -115,23 +161,37 @@ async function startTest(event) {
       method: "POST",
       headers: { "Content-Type": "application/json" },
       body: JSON.stringify(payload),
+      signal: state.abortController.signal,
     });
     const body = await response.json();
     if (response.status === 409) {
       setServerState("busy", "忙碌");
       setNotice(body.error || "有其他用户正在测速，请稍等。", true);
-      els.startButton.disabled = false;
+      setTestingControls(false);
       return;
     }
     if (!response.ok) {
       throw new Error(body.error || "创建测试失败");
     }
+    state.currentSession = body;
     await waitForBrowserQueue(body);
+    throwIfCanceled();
     await runBrowserSpeedTest(body, payload);
   } catch (error) {
-    setServerState("error", "错误");
-    setNotice(error.message, true);
-    els.startButton.disabled = false;
+    if (state.currentSession && !state.cancelRequested) {
+      await cancelCurrentSession();
+    }
+    if (state.cancelRequested || error.name === "AbortError") {
+      setServerState("idle", "空闲");
+      setNotice("测试已停止，队列已释放。");
+    } else {
+      setServerState("error", "错误");
+      setNotice(error.message, true);
+    }
+  } finally {
+    state.abortController = null;
+    state.currentSession = null;
+    setTestingControls(false);
   }
 }
 
@@ -146,9 +206,14 @@ async function waitForBrowserQueue(session) {
 
   setServerState("busy", "排队中");
   while (true) {
+    throwIfCanceled();
     setNotice(queueNotice(position));
-    await sleep(5000);
-    const response = await fetch(`${session.queueUrl}?cache=${Date.now()}`, { cache: "no-store" });
+    await sleep(5000, state.abortController && state.abortController.signal);
+    throwIfCanceled();
+    const response = await fetch(`${session.queueUrl}?cache=${Date.now()}`, {
+      cache: "no-store",
+      signal: state.abortController && state.abortController.signal,
+    });
     const body = await response.json();
     if (!response.ok) {
       throw new Error(body.error || "读取队列状态失败");
@@ -165,14 +230,17 @@ async function runBrowserSpeedTest(session, request) {
   const startedAt = new Date();
   const durationMs = Math.max(1000, (request.durationSeconds || 30) * 1000);
 
+  throwIfCanceled();
   setNotice("下载测速中...");
   await measureLatency(request.timeoutMillis);
   await runTransferPhase(durationMs, () => measureDownload(session.downloadUrl, request.timeoutMillis));
 
+  throwIfCanceled();
   setNotice("上传测速中...");
   await measureLatency(request.timeoutMillis);
   await runTransferPhase(durationMs, () => measureUpload(session.uploadUrl, request.timeoutMillis));
   await measureLatency(request.timeoutMillis);
+  throwIfCanceled();
 
   const finishedAt = new Date();
   const response = await fetch(session.completeUrl, {
@@ -183,6 +251,7 @@ async function runBrowserSpeedTest(session, request) {
       finishedAt: finishedAt.toISOString(),
       samples: state.samples,
     }),
+    signal: state.abortController && state.abortController.signal,
   });
   const body = await response.json();
   if (!response.ok) {
@@ -190,9 +259,9 @@ async function runBrowserSpeedTest(session, request) {
   }
 
   updateSummary(body.summary);
+  state.currentSession = null;
   setServerState("idle", "空闲");
   setNotice("浏览器上传/下载测速完成，记录已保存。");
-  els.startButton.disabled = false;
   await loadRecords();
 }
 
@@ -200,7 +269,9 @@ async function runTransferPhase(durationMs, measure) {
   const deadline = performance.now() + durationMs;
   let samples = 0;
   while (performance.now() < deadline || samples === 0) {
+    throwIfCanceled();
     await measure();
+    throwIfCanceled();
     samples++;
   }
 }
@@ -208,7 +279,12 @@ async function runTransferPhase(durationMs, measure) {
 async function measureLatency(timeoutMillis) {
   const started = performance.now();
   try {
-    const response = await fetchWithTimeout(`/api/config?cache=${Date.now()}`, { cache: "no-store" }, timeoutMillis || 3000);
+    const response = await fetchWithTimeout(
+      `/api/config?cache=${Date.now()}`,
+      { cache: "no-store" },
+      timeoutMillis || 3000,
+      state.abortController && state.abortController.signal,
+    );
     await response.arrayBuffer();
     recordSample({
       at: new Date().toISOString(),
@@ -232,7 +308,12 @@ async function measureDownload(downloadUrl, timeoutMillis) {
   const url = `${downloadUrl}&bytes=${DOWNLOAD_BYTES}&cache=${Date.now()}`;
   const started = performance.now();
   try {
-    const response = await fetchWithTimeout(url, { cache: "no-store" }, transferTimeoutMillis(timeoutMillis));
+    const response = await fetchWithTimeout(
+      url,
+      { cache: "no-store" },
+      transferTimeoutMillis(timeoutMillis),
+      state.abortController && state.abortController.signal,
+    );
     const data = await response.arrayBuffer();
     const elapsed = performance.now() - started;
     recordSample({
@@ -264,7 +345,7 @@ async function measureUpload(uploadUrl, timeoutMillis) {
     const response = await fetchWithTimeout(`${uploadUrl}&cache=${Date.now()}`, {
       method: "POST",
       body: payload,
-    }, transferTimeoutMillis(timeoutMillis));
+    }, transferTimeoutMillis(timeoutMillis), state.abortController && state.abortController.signal);
     await response.arrayBuffer();
     const elapsed = performance.now() - started;
     recordSample({
@@ -547,7 +628,7 @@ function drawPeriodBands(ctx, pad, plotWidth, plotHeight) {
 }
 
 function appendLog(sample) {
-  const value = sample.kind === "latency" ? `${formatNumber(sample.latencyMs)}ms` : `${formatNumber(sample.mbps)}Mbps`;
+  const value = sample.kind === "latency" ? `${formatNumber(sample.latencyMs)}ms` : `瞬时 ${formatNumber(sample.mbps)}Mbps`;
   const line = `${formatTime(sample.at)} ${sample.kind} ${sample.success ? "OK" : "FAIL"} ${value} ${sample.error || ""}`;
   const div = document.createElement("div");
   div.textContent = line;
@@ -574,8 +655,39 @@ function queueNotice(position) {
   return "正在排队，等待其他用户完成...";
 }
 
-function sleep(ms) {
-  return new Promise(resolve => window.setTimeout(resolve, ms));
+function setTestingControls(running) {
+  els.startButton.disabled = running;
+  els.stopButton.hidden = !running;
+  els.stopButton.disabled = !running;
+}
+
+function throwIfCanceled() {
+  if (state.cancelRequested || (state.abortController && state.abortController.signal.aborted)) {
+    throw cancellationError();
+  }
+}
+
+function cancellationError() {
+  const error = new Error("测试已停止");
+  error.name = "AbortError";
+  return error;
+}
+
+function sleep(ms, signal) {
+  return new Promise((resolve, reject) => {
+    if (signal && signal.aborted) {
+      reject(cancellationError());
+      return;
+    }
+    const timer = window.setTimeout(resolve, ms);
+    const abort = () => {
+      window.clearTimeout(timer);
+      reject(cancellationError());
+    };
+    if (signal) {
+      signal.addEventListener("abort", abort, { once: true });
+    }
+  });
 }
 
 function closeSource() {
@@ -623,13 +735,24 @@ function transferTimeoutMillis(timeoutMillis) {
   return Math.max(timeoutMillis || 3000, 30000);
 }
 
-async function fetchWithTimeout(url, options, timeoutMillis) {
+async function fetchWithTimeout(url, options, timeoutMillis, signal) {
   const controller = new AbortController();
   const timeoutID = window.setTimeout(() => controller.abort(), timeoutMillis);
+  const abort = () => controller.abort();
+  if (signal) {
+    if (signal.aborted) {
+      controller.abort();
+    } else {
+      signal.addEventListener("abort", abort, { once: true });
+    }
+  }
   try {
     return await fetch(url, { ...options, signal: controller.signal });
   } finally {
     window.clearTimeout(timeoutID);
+    if (signal) {
+      signal.removeEventListener("abort", abort);
+    }
   }
 }
 

web/static/index.html

@@ -46,7 +46,10 @@
             <input id="timeoutMillis" name="timeoutMillis" type="number" min="100" max="30000" value="3000">
           </label>
         </div>
-        <button id="startButton" type="submit">开始实时测试</button>
+        <div class="action-row">
+          <button id="startButton" type="submit">开始实时测试</button>
+          <button id="stopButton" class="stop-button" type="button" hidden>停止测试</button>
+        </div>
       </form>
       <div class="command-panel" aria-label="CLI 一键测试命令">
         <div class="command-title">

web/static/styles.css

@@ -215,6 +215,22 @@ button:disabled {
   opacity: 0.6;
 }
 
+.action-row {
+  display: grid;
+  grid-template-columns: minmax(0, 1fr) auto;
+  gap: 10px;
+}
+
+.stop-button {
+  color: var(--red);
+  background: #fff1f1;
+  border-color: #f5bcbc;
+}
+
+[hidden] {
+  display: none !important;
+}
+
 #refreshButton {
   min-height: 34px;
   color: var(--blue);