diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 25827b37..e5604a04 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -612,14 +612,18 @@ func (f *Fetcher) doStart() error { return nil } - // If retrying after error, reset connection states for retry - if state == stateError { - // Drain any pending error from doneCh before retry + // Each Start/Wait cycle must observe only its own completion result. + // A paused or failed cycle may have completed concurrently with Pause and + // left a result buffered before its download loop fully stopped. + if state == statePaused || state == stateError { select { case <-f.doneCh: default: } + } + // If retrying after error, reset connection states for retry + if state == stateError { f.connMu.Lock() for _, conn := range f.connections { // Reset connections that can be retried @@ -695,6 +699,8 @@ func (f *Fetcher) doStart() error { } func (f *Fetcher) downloadLoop() { + ctx := f.ctx + defer func() { // Update file last modified time before closing if f.config.UseServerCtime && f.meta.Res.Files[0].Ctime != nil { @@ -713,24 +719,30 @@ func (f *Fetcher) downloadLoop() { if !isResume { // Fresh start: begin with resolve connection f.startResolveDownload() + // Non-range downloads wait for their only connection in + // startResolveDownload. Falling through would consume the connection's + // expansion signal and publish the same completion a second time. + if !f.meta.Res.Range || f.meta.Res.Size == 0 || f.getState() == stateDone { + return + } } else { // Resume: restart existing connections f.resumeConnections() - f.waitForCompletion() + f.waitForCompletion(ctx) return } // Slow start loop for { select { - case <-f.ctx.Done(): + case <-ctx.Done(): // Paused or cancelled return case <-f.slowStart.expansionCh: // Batch completed, try to expand if f.checkCompletion() { // All work is done, wait for connections to finish - f.waitForCompletion() + f.waitForCompletion(ctx) return } f.expandConnections() @@ -738,7 +750,7 @@ func (f *Fetcher) downloadLoop() { // Check if we've reached steady state (max connections) if f.getState() == stateSteady { // Wait for all connections to complete - f.waitForCompletion() + f.waitForCompletion(ctx) return } } @@ -764,7 +776,7 @@ func (f *Fetcher) startResolveDownload() { // For non-range downloads, wait for completion directly in this goroutine // Don't create another goroutine to avoid WaitGroup reuse issues - f.waitForCompletion() + f.waitForCompletion(f.ctx) return } @@ -881,7 +893,6 @@ func (f *Fetcher) expandConnections() { if len(newConns) == 0 { // No new connections could be created, stop expansion f.setState(stateSteady) - go f.waitForCompletion() return } @@ -1181,9 +1192,13 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) { n, err := reader.Read(buf) if n > 0 { + f.connMu.Lock() + writeOffset := conn.Chunk.Downloaded + f.connMu.Unlock() + f.fileMu.Lock() if f.file != nil { - _, writeErr := f.file.WriteAt(buf[:n], conn.Chunk.Downloaded) + _, writeErr := f.file.WriteAt(buf[:n], writeOffset) if writeErr != nil { f.fileMu.Unlock() f.connMu.Lock() @@ -1198,8 +1213,10 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) { } f.fileMu.Unlock() + f.connMu.Lock() conn.Chunk.Downloaded += int64(n) conn.Downloaded += int64(n) + f.connMu.Unlock() } if err != nil { if err == io.EOF { @@ -1283,9 +1300,13 @@ func (f *Fetcher) runConnectionFallback(conn *connection) { n, err := reader.Read(buf) if n > 0 { + f.connMu.Lock() + writeOffset := conn.Chunk.Downloaded + f.connMu.Unlock() + f.fileMu.Lock() if f.file != nil { - _, writeErr := f.file.WriteAt(buf[:n], conn.Chunk.Downloaded) + _, writeErr := f.file.WriteAt(buf[:n], writeOffset) if writeErr != nil { f.fileMu.Unlock() return writeErr @@ -1293,8 +1314,10 @@ func (f *Fetcher) runConnectionFallback(conn *connection) { } f.fileMu.Unlock() + f.connMu.Lock() conn.Chunk.Downloaded += int64(n) conn.Downloaded += int64(n) + f.connMu.Unlock() } if err != nil { if err == io.EOF { @@ -1480,10 +1503,10 @@ func (f *Fetcher) resumeConnections() { } } -func (f *Fetcher) waitForCompletion() { +func (f *Fetcher) waitForCompletion(ctx context.Context) { f.wg.Wait() - // Only trigger completion if not cancelled/paused - if f.ctx != nil && f.ctx.Err() == nil { + // Only trigger completion for the Start cycle that launched this waiter. + if ctx != nil && ctx.Err() == nil { f.onDownloadComplete() } } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index fb1386e0..b5ad1659 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -422,6 +422,25 @@ func TestFetcher_DownloadContinue_NoRangeRestart(t *testing.T) { } } +func TestFetcher_DownloadNoRangeLeavesNoStaleCompletion(t *testing.T) { + listener := test.StartTestNoRangeSlowServer(0) + defer listener.Close() + + f := downloadReady(listener, 1, t).(*Fetcher) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := f.Wait(); err != nil { + t.Fatal(err) + } + + select { + case err := <-f.doneCh: + t.Fatalf("stale completion signal: %v", err) + case <-time.After(100 * time.Millisecond): + } +} + func TestFetcher_DownloadChunked(t *testing.T) { listener := test.StartTestCustomServer() defer listener.Close() diff --git a/pkg/download/extension_test.go b/pkg/download/extension_test.go index fd63d9a5..b78380d5 100644 --- a/pkg/download/extension_test.go +++ b/pkg/download/extension_test.go @@ -838,7 +838,9 @@ func TestDownloader_Extension_BlobPauseContinueKeepsSource(t *testing.T) { for _, tc := range cases { tc := tc t.Run(tc.name, func(t *testing.T) { - downloader := NewDownloader(nil) + downloader := NewDownloader(&DownloaderConfig{ + StorageDir: t.TempDir(), + }) downloader.cfg.RefreshInterval = 50 if err := downloader.Setup(); err != nil { t.Fatal(err) @@ -1479,7 +1481,7 @@ func TestDownloader_ExtensionRuntimeWebViewInjected(t *testing.T) { } defer runtime.Close() - value, err := runtime.Eval(`({ + value, err := runtime.Eval(`(async () => ({ hasRuntime: !!gopeed.runtime, hasBlob: !!(gopeed.runtime && gopeed.runtime.blob), hasCreateObjectURL: typeof gopeed.runtime.blob.createObjectURL, @@ -1487,8 +1489,8 @@ func TestDownloader_ExtensionRuntimeWebViewInjected(t *testing.T) { hasWebView: !!(gopeed.runtime && gopeed.runtime.webview), hasOpen: typeof gopeed.runtime.webview.open, hasWebViewIsAvailable: typeof gopeed.runtime.webview.isAvailable, - webViewAvailable: gopeed.runtime.webview.isAvailable() - })`) + webViewAvailable: await gopeed.runtime.webview.isAvailable() + }))()`) if err != nil { t.Fatal(err) } @@ -1530,9 +1532,9 @@ func TestDownloader_ExtensionRuntimeBlobRejectsReadableStreamSource(t *testing.T } defer runtime.Close() - value, err := runtime.Eval(`(() => { + value, err := runtime.Eval(`(async () => { try { - gopeed.runtime.blob.createObjectURL(new ReadableStream({ + await gopeed.runtime.blob.createObjectURL(new ReadableStream({ start(controller) { controller.close(); }, @@ -1573,7 +1575,7 @@ func TestDownloader_ExtensionRuntimeWebViewAvailabilityFromProvider(t *testing.T } defer runtime.Close() - value, err := runtime.Eval(`gopeed.runtime.webview.isAvailable()`) + value, err := runtime.Eval(`(async () => await gopeed.runtime.webview.isAvailable())()`) if err != nil { t.Fatal(err) } diff --git a/pkg/download/testdata/extensions/basic/index.js b/pkg/download/testdata/extensions/basic/index.js index c4751dbb..cfbde77b 100644 --- a/pkg/download/testdata/extensions/basic/index.js +++ b/pkg/download/testdata/extensions/basic/index.js @@ -1,7 +1,7 @@ gopeed.events.onResolve(async function (ctx) { - ctx.req.setLabels({ replaced: "true", removed: "true" }); - ctx.req.putLabel("modified", "true"); - ctx.req.delLabel("removed"); + await ctx.req.setLabels({ replaced: "true", removed: "true" }); + await ctx.req.putLabel("modified", "true"); + await ctx.req.delLabel("removed"); ctx.res = { name: "test", files: Array(2).fill(true).map((_, i) => ({ diff --git a/pkg/download/testdata/extensions/blob/index.js b/pkg/download/testdata/extensions/blob/index.js index b4d88a15..4002549c 100644 --- a/pkg/download/testdata/extensions/blob/index.js +++ b/pkg/download/testdata/extensions/blob/index.js @@ -3,7 +3,7 @@ gopeed.events.onResolve(async function (ctx) { if (ctx.req.url.endsWith("/blob")) { const blob = new Blob(["hello world"], {type: "text/plain"}); - const url = gopeed.runtime.blob.createObjectURL(blob); + const url = await gopeed.runtime.blob.createObjectURL(blob); ctx.res = { name: "blob-blob", files: [ @@ -42,7 +42,7 @@ gopeed.events.onResolve(async function (ctx) { } }); - const url = gopeed.runtime.blob.createObjectURL(open, { size: payload.byteLength, range: true }); + const url = await gopeed.runtime.blob.createObjectURL(open, { size: payload.byteLength, range: true }); ctx.res = { name: "blob-opener-range", @@ -61,7 +61,7 @@ gopeed.events.onResolve(async function (ctx) { } if (ctx.req.url.endsWith("/opener-unknown")) { - const url = gopeed.runtime.blob.createObjectURL(async ({ offset = 0 }) => new ReadableStream({ + const url = await gopeed.runtime.blob.createObjectURL(async ({ offset = 0 }) => new ReadableStream({ async start(controller) { if (offset > 0) { controller.close(); @@ -88,7 +88,7 @@ gopeed.events.onResolve(async function (ctx) { return; } - const url = gopeed.runtime.blob.createObjectURL(async ({ offset = 0, end = -1 }) => new ReadableStream({ + const url = await gopeed.runtime.blob.createObjectURL(async ({ offset = 0, end = -1 }) => new ReadableStream({ start(controller) { const payload = encoder.encode("line 1\nline 2\n"); controller.enqueue(payload.slice(offset, end >= 0 ? end + 1 : undefined)); diff --git a/pkg/download/testdata/extensions/blob_pause_rebuild/index.js b/pkg/download/testdata/extensions/blob_pause_rebuild/index.js index 93263b0c..4b3fbabc 100644 --- a/pkg/download/testdata/extensions/blob_pause_rebuild/index.js +++ b/pkg/download/testdata/extensions/blob_pause_rebuild/index.js @@ -39,7 +39,7 @@ function fastPayloadStream(offset = 0, end = -1) { }); } -function createBlobSourceURL(delayed) { +async function createBlobSourceURL(delayed) { const blob = new Blob([payloadText], { type: 'application/octet-stream' }); const originalSlice = blob.slice.bind(blob); blob.slice = function (offset = 0, end) { @@ -53,20 +53,20 @@ function createBlobSourceURL(delayed) { }, }; }; - return gopeed.runtime.blob.createObjectURL(blob); + return await gopeed.runtime.blob.createObjectURL(blob); } -function createOpenerSourceURL(delayed) { +async function createOpenerSourceURL(delayed) { const open = async () => delayed ? delayedPayloadStream(0, -1) : fastPayloadStream(0, -1); - return gopeed.runtime.blob.createObjectURL(open, { + return await gopeed.runtime.blob.createObjectURL(open, { size: payloadSize, contentType: 'application/octet-stream', }); } -function createRangeSourceURL(delayed) { +async function createRangeSourceURL(delayed) { const open = async ({ offset = 0, end = -1 }) => delayed ? delayedPayloadStream(offset, end) : fastPayloadStream(offset, end); - return gopeed.runtime.blob.createObjectURL(open, { + return await gopeed.runtime.blob.createObjectURL(open, { size: payloadSize, range: true, contentType: 'application/octet-stream', @@ -86,14 +86,14 @@ function modeFromURL(url) { return ''; } -function createPayloadURL(mode, delayed) { +async function createPayloadURL(mode, delayed) { switch (mode) { case 'blob': - return createBlobSourceURL(delayed); + return await createBlobSourceURL(delayed); case 'opener': - return createOpenerSourceURL(delayed); + return await createOpenerSourceURL(delayed); case 'range': - return createRangeSourceURL(delayed); + return await createRangeSourceURL(delayed); default: throw new Error('unsupported pause rebuild mode: ' + mode); } @@ -113,7 +113,7 @@ gopeed.events.onResolve(async function (ctx) { name: 'pause-rebuild-' + mode + '.bin', size: payloadSize, req: { - url: createPayloadURL(mode, true), + url: await createPayloadURL(mode, true), rawUrl: ctx.req.rawUrl || ctx.req.url, labels: { mode: 'pause-rebuild', @@ -132,8 +132,8 @@ gopeed.events.onError(async function (ctx) { return; } - ctx.task.setUrl(createPayloadURL(source, false)); - req.putLabel('started', 'true'); - req.putLabel('rebuilt', 'true'); - ctx.task.continue(); + await ctx.task.setUrl(await createPayloadURL(source, false)); + await req.putLabel('started', 'true'); + await req.putLabel('rebuilt', 'true'); + await ctx.task.continue(); }); diff --git a/pkg/download/testdata/extensions/blob_recover/index.js b/pkg/download/testdata/extensions/blob_recover/index.js index 911ae199..7a6c4926 100644 --- a/pkg/download/testdata/extensions/blob_recover/index.js +++ b/pkg/download/testdata/extensions/blob_recover/index.js @@ -4,7 +4,7 @@ gopeed.events.onResolve(async function (ctx) { return; } - const url = gopeed.runtime.blob.createObjectURL(async ({ offset = 0 }) => new ReadableStream({ + const url = await gopeed.runtime.blob.createObjectURL(async ({ offset = 0 }) => new ReadableStream({ async start(controller) { if (offset > 0) { controller.close(); @@ -40,7 +40,7 @@ gopeed.events.onError(async function (ctx) { return; } - req.putLabel("recovered", "true"); - ctx.task.setUrl(gopeed.runtime.blob.createObjectURL(new Blob(["ok\n"], { type: "text/plain" }))); - ctx.task.continue(); + await req.putLabel("recovered", "true"); + await ctx.task.setUrl(await gopeed.runtime.blob.createObjectURL(new Blob(["ok\n"], { type: "text/plain" }))); + await ctx.task.continue(); }); diff --git a/pkg/download/testdata/extensions/blob_restart/index.js b/pkg/download/testdata/extensions/blob_restart/index.js index 4a7eb240..b640f674 100644 --- a/pkg/download/testdata/extensions/blob_restart/index.js +++ b/pkg/download/testdata/extensions/blob_restart/index.js @@ -1,5 +1,5 @@ -function createRangeBlobUrl(target) { - return gopeed.runtime.blob.createObjectURL(async ({ offset = 0, end = -1 }) => { +async function createRangeBlobUrl(target) { + return await gopeed.runtime.blob.createObjectURL(async ({ offset = 0, end = -1 }) => { const headers = {}; if (offset > 0 || end >= 0) { headers.Range = `bytes=${offset}-${end >= 0 ? end : ''}`; @@ -25,7 +25,7 @@ gopeed.events.onResolve(async function (ctx) { name: 'restart.bin', size: 262144, req: { - url: createRangeBlobUrl(ctx.req.rawUrl || ctx.req.url), + url: await createRangeBlobUrl(ctx.req.rawUrl || ctx.req.url), rawUrl: ctx.req.rawUrl || ctx.req.url, labels: { mode: 'restart', @@ -46,12 +46,12 @@ gopeed.events.onError(async function (ctx) { } try { - ctx.task.setUrl(createRangeBlobUrl(req.rawUrl)); - req.putLabel('started', 'true'); - req.putLabel('rebuilt', 'true'); - ctx.task.continue(); + await ctx.task.setUrl(await createRangeBlobUrl(req.rawUrl)); + await req.putLabel('started', 'true'); + await req.putLabel('rebuilt', 'true'); + await ctx.task.continue(); } catch (error) { - req.putLabel('rebuildError', String(error)); + await req.putLabel('rebuildError', String(error)); throw error; } }); diff --git a/pkg/download/testdata/extensions/blob_stream_bridge/index.js b/pkg/download/testdata/extensions/blob_stream_bridge/index.js index f3c7e616..04e0d284 100644 --- a/pkg/download/testdata/extensions/blob_stream_bridge/index.js +++ b/pkg/download/testdata/extensions/blob_stream_bridge/index.js @@ -124,7 +124,7 @@ function createMultiplexStreams(url) { gopeed.events.onResolve(async function (ctx) { if (ctx.req.url.includes("/single-fetch-stream")) { const payloadSize = Number(new URL(ctx.req.url).searchParams.get("size")) || 0; - const url = gopeed.runtime.blob.createObjectURL(async () => createFetchDrainedStream(sourceURL(ctx.req.url)), { + const url = await gopeed.runtime.blob.createObjectURL(async () => createFetchDrainedStream(sourceURL(ctx.req.url)), { size: payloadSize, }); ctx.res = { @@ -142,7 +142,7 @@ gopeed.events.onResolve(async function (ctx) { if (ctx.req.url.includes("/range-fetch-stream")) { const payloadSize = Number(new URL(ctx.req.url).searchParams.get("size")) || 0; - const url = gopeed.runtime.blob.createObjectURL( + const url = await gopeed.runtime.blob.createObjectURL( async ({ offset = 0, end = -1 }) => createFetchDrainedRangeStream(sourceURL(ctx.req.url), offset, end), { size: payloadSize, @@ -173,10 +173,10 @@ gopeed.events.onResolve(async function (ctx) { } return streams; }; - const videoUrl = gopeed.runtime.blob.createObjectURL(async () => openStreams().video, { + const videoUrl = await gopeed.runtime.blob.createObjectURL(async () => openStreams().video, { size: videoChunks.reduce((sum, chunk) => sum + byteLength(chunk), 0), }); - const audioUrl = gopeed.runtime.blob.createObjectURL(async () => openStreams().audio, { + const audioUrl = await gopeed.runtime.blob.createObjectURL(async () => openStreams().audio, { size: audioChunks.reduce((sum, chunk) => sum + byteLength(chunk), 0), }); ctx.res = { diff --git a/pkg/download/testdata/extensions/on_error/index.js b/pkg/download/testdata/extensions/on_error/index.js index c8fa26b2..e5fa2955 100644 --- a/pkg/download/testdata/extensions/on_error/index.js +++ b/pkg/download/testdata/extensions/on_error/index.js @@ -1,6 +1,6 @@ gopeed.events.onError(async function (ctx) { gopeed.logger.info("url", ctx.task.meta.req.url); gopeed.logger.info("error", ctx.error); - ctx.task.setUrl("https://github.com"); - ctx.task.continue(); + await ctx.task.setUrl("https://github.com"); + await ctx.task.continue(); }); diff --git a/pkg/download/testdata/extensions/on_start/index.js b/pkg/download/testdata/extensions/on_start/index.js index 671c99e9..59370be3 100644 --- a/pkg/download/testdata/extensions/on_start/index.js +++ b/pkg/download/testdata/extensions/on_start/index.js @@ -1,5 +1,5 @@ gopeed.events.onStart(async function (ctx) { gopeed.logger.info("url", ctx.task.meta.req.url); - ctx.task.setUrl("https://github.com"); - ctx.task.meta.req.putLabel('modified', 'true'); + await ctx.task.setUrl("https://github.com"); + await ctx.task.meta.req.putLabel('modified', 'true'); }); diff --git a/pkg/download/testdata/extensions/storage/index.js b/pkg/download/testdata/extensions/storage/index.js index 2d7b3d1f..99937290 100644 --- a/pkg/download/testdata/extensions/storage/index.js +++ b/pkg/download/testdata/extensions/storage/index.js @@ -2,36 +2,36 @@ gopeed.events.onResolve(async function (ctx) { const key = "key" const value1 = "value1", value2 = JSON.stringify({a: 1, b: "2"}) - if (gopeed.storage.get(key) !== null) { + if ((await gopeed.storage.get(key)) !== null) { throw new Error("storage get null error") } - gopeed.storage.remove(key) - if(gopeed.storage.keys().length !== 0) { + await gopeed.storage.remove(key) + if((await gopeed.storage.keys()).length !== 0) { throw new Error("storage keys null error") } - gopeed.storage.set(key, value1) - if (gopeed.storage.get(key) !== value1) { + await gopeed.storage.set(key, value1) + if ((await gopeed.storage.get(key)) !== value1) { throw new Error("storage put1 error") } - gopeed.storage.set(key, value2) - if (gopeed.storage.get(key) !== value2) { + await gopeed.storage.set(key, value2) + if ((await gopeed.storage.get(key)) !== value2) { throw new Error("storage put2 error") } - if(gopeed.storage.keys().length !== 1) { + if((await gopeed.storage.keys()).length !== 1) { throw new Error("storage keys error") } - gopeed.storage.remove(key) - if (gopeed.storage.get(key) !== null) { + await gopeed.storage.remove(key) + if ((await gopeed.storage.get(key)) !== null) { throw new Error("storage delete error") } - gopeed.storage.set(key, value1) - gopeed.storage.clear() - if (gopeed.storage.get(key) !== null) { + await gopeed.storage.set(key, value1) + await gopeed.storage.clear() + if ((await gopeed.storage.get(key)) !== null) { throw new Error("storage clear error") } diff --git a/pkg/rest/server_test.go b/pkg/rest/server_test.go index 6ff1fb7d..a61933b6 100644 --- a/pkg/rest/server_test.go +++ b/pkg/rest/server_test.go @@ -924,7 +924,7 @@ func TestBuildServerPropagatesWebViewProvider(t *testing.T) { } defer runtime.Close() - value, err := runtime.Eval("gopeed.runtime.webview.isAvailable()") + value, err := runtime.Eval("(async () => await gopeed.runtime.webview.isAvailable())()") if err != nil { t.Fatal(err) }