kanshan/speedtest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: persist partial speed tests

Browse Source
This commit is contained in:
Test User
2026-06-10 17:52:55 +08:00
parent 1938c4b0fc
commit 4df04d1605
4 changed files with 137 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/client/client.go
+18 -4
View File
@@ -69,6 +69,8 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
if err != nil {
return probe.Summary{}, err
}
var samples []probe.Sample
var startedAt time.Time
complete := false
defer func() {
if !complete {
@@ -77,12 +79,24 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
_ = cancelSession(cancelCtx, &httpClient, server, session.CancelURL)
}
}()
completePartial := func(original error) (probe.Summary, error) {
if len(samples) == 0 || startedAt.IsZero() {
return probe.Summary{}, original
}
finishedAt := time.Now().UTC()
completeCtx, cancel := context.WithTimeout(context.Background(), cancelRequestTimeout(options.Timeout))
defer cancel()
if _, completeErr := completeSession(completeCtx, &httpClient, server, session.CompleteURL, startedAt, finishedAt, samples); completeErr != nil {
return probe.Summary{}, fmt.Errorf("%w; also failed to complete partial test session: %v", original, completeErr)
}
complete = true
return probe.Summary{}, original
}
if err := waitForQueue(ctx, &httpClient, server, session, options); err != nil {
return probe.Summary{}, err
}
var samples []probe.Sample
record := func(sample probe.Sample) {
samples = append(samples, sample)
if options.OnSample != nil {
@@ -90,18 +104,18 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
}
}
startedAt := time.Now().UTC()
startedAt = time.Now().UTC()
record(measureLatency(ctx, &httpClient, server))
if err := runPhase(ctx, options.Duration, func() probe.Sample {
return measureDownload(ctx, &httpClient, server, session.DownloadURL, options.DownloadBytes)
}, record); err != nil {
return probe.Summary{}, err
return completePartial(err)
}
record(measureLatency(ctx, &httpClient, server))
if err := runPhase(ctx, options.Duration, func() probe.Sample {
return measureUpload(ctx, &httpClient, server, session.UploadURL, options.UploadBytes)
}, record); err != nil {
return probe.Summary{}, err
return completePartial(err)
}
record(measureLatency(ctx, &httpClient, server))
finishedAt := time.Now().UTC()
internal/client/client_test.go
+38
View File
@@ -131,6 +131,44 @@ func TestRunCancelsQueuedSessionWhenContextStops(t *testing.T) {
}
}
func TestRunPersistsPartialRecordWhenCanceledAfterSamples(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
canceled := false
_, err := Run(ctx, Options{
ServerURL: httpServer.URL,
Duration: time.Second,
Timeout: time.Second,
DownloadBytes: 1024,
UploadBytes: 512,
OnSample: func(sample probe.Sample) {
if sample.Kind == "download" && !canceled {
canceled = true
cancel()
}
},
})
if !errors.Is(err, context.Canceled) {
t.Fatalf("Run error = %v, want context canceled", err)
}
records := store.snapshot()
if len(records) != 1 {
t.Fatalf("records = %d, want one partial record", len(records))
}
if !hasKind(records[0].Samples, "download") {
t.Fatalf("samples = %#v, want partial download sample persisted", records[0].Samples)
}
if records[0].Summary.Samples != len(records[0].Samples) {
t.Fatalf("summary samples = %d, want %d", records[0].Summary.Samples, len(records[0].Samples))
}
}
func TestRunIperf3CompletesOneOffServerSession(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})
web/assets_test.go
+13
View File
@@ -104,6 +104,19 @@ func TestStaticAssetsIncludeManualCancelBehavior(t *testing.T) {
}
}
func TestStaticAssetsPersistPartialBrowserStop(t *testing.T) {
app, err := readAsset("app.js")
if err != nil {
t.Fatalf("read app.js: %v", err)
}
if !contains(app, "completeCurrentSession") ||
!contains(app, "session.completeUrl") ||
!contains(app, "state.samples.length") ||
!contains(app, "已保存已完成样本") {
t.Fatalf("app.js missing partial browser stop persistence behavior")
}
}
func TestStaticAssetsLabelTransferSamplesAsInstantaneous(t *testing.T) {
app, err := readAsset("app.js")
if err != nil {
web/static/app.js
+68 -23
View File
@@ -6,6 +6,8 @@ const state = {
currentSession: null,
abortController: null,
cancelRequested: false,
startedAt: null,
partialSaved: false,
};
const els = {
@@ -46,7 +48,9 @@ window.addEventListener("resize", () => {
});
window.addEventListener("beforeunload", () => {
state.cancelRequested = true;
cancelCurrentSession({ beacon: true });
if (!completeCurrentSession({ beacon: true, partial: true })) {
cancelCurrentSession({ beacon: true });
}
});
initializeDefaultTarget();
@@ -106,10 +110,13 @@ async function stopCurrentTest() {
if (state.abortController) {
state.abortController.abort();
}
await cancelCurrentSession();
const saved = await completeCurrentSession({ partial: true });
if (!saved) {
await cancelCurrentSession();
}
closeSource();
setServerState("idle", "空闲");
setNotice("测试已停止，队列已释放。");
setNotice(saved ? "测试已停止，已保存已完成样本。" : "测试已停止，队列已释放。");
setTestingControls(false);
}
@@ -135,6 +142,56 @@ async function cancelCurrentSession(options = {}) {
}
}
function completeCurrentSession(options = {}) {
const session = state.currentSession;
if (!session || !session.completeUrl || !state.samples.length) {
return false;
}
const startedAt = state.startedAt || new Date();
const body = JSON.stringify({
startedAt: startedAt.toISOString(),
finishedAt: new Date().toISOString(),
samples: state.samples,
});
if (options.beacon && navigator.sendBeacon) {
const sent = navigator.sendBeacon(session.completeUrl, new Blob([body], { type: "application/json" }));
if (sent) {
state.currentSession = null;
state.partialSaved = Boolean(options.partial);
}
return sent;
}
return completeCurrentSessionWithFetch(session, body, options);
}
async function completeCurrentSessionWithFetch(session, body, options = {}) {
try {
const fetchOptions = {
method: "POST",
headers: { "Content-Type": "application/json" },
body,
cache: "no-store",
};
if (options.keepalive) {
fetchOptions.keepalive = true;
}
const response = await fetch(session.completeUrl, fetchOptions);
const payload = await response.json();
if (!response.ok) {
throw new Error(payload.error || "保存测试记录失败");
}
state.currentSession = null;
state.partialSaved = Boolean(options.partial);
updateSummary(payload.summary);
await loadRecords();
return true;
} catch (error) {
return false;
}
}
async function startTest(event) {
event.preventDefault();
closeSource();
@@ -142,6 +199,8 @@ async function startTest(event) {
state.currentSession = null;
state.cancelRequested = false;
state.abortController = new AbortController();
state.startedAt = null;
state.partialSaved = false;
els.eventLog.textContent = "";
setNotice("");
setServerState("running", "运行中");
@@ -183,7 +242,7 @@ async function startTest(event) {
}
if (state.cancelRequested || error.name === "AbortError") {
setServerState("idle", "空闲");
setNotice("测试已停止，队列已释放。");
setNotice(state.partialSaved ? "测试已停止，已保存已完成样本。" : "测试已停止，队列已释放。");
} else {
setServerState("error", "错误");
setNotice(error.message, true);
@@ -191,6 +250,8 @@ async function startTest(event) {
} finally {
state.abortController = null;
state.currentSession = null;
state.startedAt = null;
state.partialSaved = false;
setTestingControls(false);
}
}
@@ -227,7 +288,7 @@ async function waitForBrowserQueue(session) {
}
async function runBrowserSpeedTest(session, request) {
const startedAt = new Date();
state.startedAt = new Date();
const durationMs = Math.max(1000, (request.durationSeconds || 30) * 1000);
throwIfCanceled();
@@ -242,27 +303,11 @@ async function runBrowserSpeedTest(session, request) {
await measureLatency(request.timeoutMillis);
throwIfCanceled();
const finishedAt = new Date();
const response = await fetch(session.completeUrl, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
startedAt: startedAt.toISOString(),
finishedAt: finishedAt.toISOString(),
samples: state.samples,
}),
signal: state.abortController && state.abortController.signal,
});
const body = await response.json();
if (!response.ok) {
throw new Error(body.error || "保存测试记录失败");
if (!await completeCurrentSession()) {
throw new Error("保存测试记录失败");
}
updateSummary(body.summary);
state.currentSession = null;
setServerState("idle", "空闲");
setNotice("浏览器上传/下载测速完成，记录已保存。");
await loadRecords();
}
async function runTransferPhase(durationMs, measure) {