test: enforce async extension API contract (#1452)

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()
 	}
 }

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()

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)
 	}

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) => ({

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));

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();
 });

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();
 });

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;
   }
 });

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 = {

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();
 });

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');
 });

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")
     }
 

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)
 	}