From ecfc50523f4f73c834b6e1c06308337318c4551b Mon Sep 17 00:00:00 2001 From: Test User Date: Wed, 10 Jun 2026 16:51:18 +0800 Subject: [PATCH] fix: release queued tests on cancellation --- README.md | 2 + cmd/netstable/main.go | 11 +- internal/client/client.go | 53 +++++++++- internal/client/client_test.go | 48 +++++++++ internal/client/iperf3.go | 4 +- internal/web/server.go | 188 +++++++++++++++++++++++++++++---- internal/web/server_test.go | 187 ++++++++++++++++++++++++++++++++ web/assets_test.go | 31 ++++++ web/static/app.js | 153 ++++++++++++++++++++++++--- web/static/index.html | 5 +- web/static/styles.css | 16 +++ 11 files changed, 659 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index c62e125..18c1303 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,8 @@ NetStable 是一个独立的 Go Web 测速项目，用于测试“打开网页 - 启动时可设置程序理论最高带宽，例如 `-bandwidth-limit-mbps 30` 将测试流量限制到 30 Mbps。 - 页面展示当前设置的理论最高限值。 - 一次测试按阶段执行：先连续下载指定时长，再连续上传同样时长。页面固定提供 `15s` 和 `30s` 两档，默认 `30s`。 +- 手动停止、刷新页面或 CLI 中断时会取消当前会话并释放队列，后续用户会自动前移。 +- 下载/上传峰值使用连续样本的持续吞吐口径，避免单个短请求计时误差造成虚高峰值。 - 保存每次测试的完整样本曲线、摘要、脱敏后的用户 IP、地区和运营商信息；ISP 缺失时显示“未知运营商”。 - 所有历史测试用户数据通过网页和 `/api/records` 提供。 - 历史时序图展示不同时段的平均下载速度变化。 diff --git a/cmd/netstable/main.go b/cmd/netstable/main.go index 4cc1f1e..d6d5211 100644 --- a/cmd/netstable/main.go +++ b/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) } diff --git a/internal/client/client.go b/internal/client/client.go index 21fca65..afefce1 100644 --- a/internal/client/client.go +++ b/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()) } diff --git a/internal/client/client_test.go b/internal/client/client_test.go index a956e16..5a18fe6 100644 --- a/internal/client/client_test.go +++ b/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)}) diff --git a/internal/client/iperf3.go b/internal/client/iperf3.go index eb9bf73..9224d3f 100644 --- a/internal/client/iperf3.go +++ b/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 diff --git a/internal/web/server.go b/internal/web/server.go index ad01144..8d47bab 100644 --- a/internal/web/server.go +++ b/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 diff --git a/internal/web/server_test.go b/internal/web/server_test.go index 73c29ba..8603544 100644 --- a/internal/web/server_test.go +++ b/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}) diff --git a/web/assets_test.go b/web/assets_test.go index d2a0c3b..24d85bc 100644 --- a/web/assets_test.go +++ b/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) } diff --git a/web/static/app.js b/web/static/app.js index dff9203..892856e 100644 --- a/web/static/app.js +++ b/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); + } } } diff --git a/web/static/index.html b/web/static/index.html index 46b325b..c8c4567 100644 --- a/web/static/index.html +++ b/web/static/index.html @@ -46,7 +46,10 @@ - +
+ + +
diff --git a/web/static/styles.css b/web/static/styles.css index c3c77ce..45361db 100644 --- a/web/static/styles.css +++ b/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);