feat: spool media inputs to disk and report producer progress (#1520)

internal/protocol/http/fetcher.go

@@ -512,17 +512,9 @@ func (f *Fetcher) asyncPrefetch() {
 		return
 	}
 
-	// Create temporary file for prefetch data
-	tmpFile, err := os.CreateTemp("", "gopeed-prefetch-*")
-	if err != nil {
-		f.prefetchErr = err
-		return
-	}
-	f.prefetchFile = tmpFile
-	f.prefetchFilePath = tmpFile.Name()
-
 	defer func() {
-		// Close response body when prefetch stops
+		// Close the response even if temporary-file setup fails or shutdown
+		// rejects a late prefetch. Otherwise the server can remain blocked writing.
 		f.resolveRespLock.Lock()
 		if f.resolveResp != nil {
 			f.resolveResp.Body.Close()
@@ -531,8 +523,33 @@ func (f *Fetcher) asyncPrefetch() {
 		f.resolveRespLock.Unlock()
 	}()
 
+	// Create temporary file for prefetch data
+	tempDir := ""
+	if f.ctl != nil && f.ctl.TempDir != "" {
+		tempDir = f.ctl.TempDir
+		if err := os.MkdirAll(tempDir, 0700); err != nil {
+			f.prefetchErr = err
+			return
+		}
+	}
+	tmpFile, err := os.CreateTemp(tempDir, "gopeed-prefetch-*")
+	if err != nil {
+		f.prefetchErr = err
+		return
+	}
+	if f.ctl != nil {
+		if err := f.ctl.TempFiles.Track(tmpFile.Name()); err != nil {
+			tmpFile.Close()
+			os.Remove(tmpFile.Name())
+			f.prefetchErr = err
+			return
+		}
+	}
+	f.prefetchFile = tmpFile
+	f.prefetchFilePath = tmpFile.Name()
+
 	buf := make([]byte, 32*1024) // 32KB buffer
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 
 	for {
 		select {
@@ -617,7 +634,11 @@ func (f *Fetcher) cleanupPrefetchFile() {
 		f.prefetchFile = nil
 	}
 	if f.prefetchFilePath != "" {
-		os.Remove(f.prefetchFilePath)
+		if f.ctl != nil {
+			f.ctl.TempFiles.Remove(f.prefetchFilePath)
+		} else {
+			os.Remove(f.prefetchFilePath)
+		}
 		f.prefetchFilePath = ""
 	}
 	// The byte count is only reusable while the backing prefetch file exists.
@@ -1329,7 +1350,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
 		f.slowStart.onConnectSuccess()
 	}
 
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 	var responseBytesRead int64
 	for {
 		if conn.ctx.Err() != nil {
@@ -1711,7 +1732,7 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
 	}
 
 	// Download data from resolve response
-	reader := NewTimeoutReader(resp.Body, readTimeout)
+	reader := f.responseReader(resp)
 	for {
 		if conn.ctx.Err() != nil {
 			return
@@ -1828,7 +1849,7 @@ func (f *Fetcher) runConnectionFallback(conn *connection) {
 				f.slowStart.onConnectSuccess()
 			}
 
-			reader := NewTimeoutReader(resp.Body, readTimeout)
+			reader := f.responseReader(resp)
 			for {
 				if conn.ctx.Err() != nil {
 					return conn.ctx.Err()
@@ -2451,3 +2472,12 @@ func (f *Fetcher) Progress() fetcher.Progress {
 func (f *Fetcher) Wait() error {
 	return <-f.doneCh
 }
+
+// Generated output can pause while its producer waits for processing capacity.
+// Its producer owns upstream deadlines; cancellation still closes the response.
+func (f *Fetcher) responseReader(resp *http.Response) io.Reader {
+	if f.ctl != nil && f.ctl.ManagedProduction != nil && resp.Request != nil && f.ctl.ManagedProduction(resp.Request.URL.String()) {
+		return resp.Body
+	}
+	return NewTimeoutReader(resp.Body, readTimeout)
+}

internal/protocol/http/prefetch_setup_test.go

@@ -0,0 +1,74 @@
+package http
+
+import (
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/internal/controller"
+	"github.com/GopeedLab/gopeed/internal/tempfiles"
+)
+
+type prefetchSetupBody struct {
+	io.Reader
+	closed bool
+}
+
+func (b *prefetchSetupBody) Close() error {
+	b.closed = true
+	return nil
+}
+
+func TestPrefetchSetupFailureClosesResponse(t *testing.T) {
+	for _, closedScope := range []bool{false, true} {
+		name := "invalid_temp_directory"
+		if closedScope {
+			name = "downloader_already_closed"
+		}
+		t.Run(name, func(t *testing.T) {
+			root := t.TempDir()
+			scope := &tempfiles.Scope{}
+			dir := root
+			if closedScope {
+				if err := scope.Close(); err != nil {
+					t.Fatal(err)
+				}
+			} else {
+				dir = filepath.Join(root, "file")
+				if err := os.WriteFile(dir, nil, 0600); err != nil {
+					t.Fatal(err)
+				}
+			}
+			body := &prefetchSetupBody{Reader: strings.NewReader("unused response")}
+			f := &Fetcher{
+				ctl:         &controller.Controller{TempDir: dir, TempFiles: scope},
+				resolveResp: &http.Response{Body: body},
+			}
+			f.asyncPrefetch()
+			if f.prefetchErr == nil {
+				t.Fatal("expected temporary-file initialization to fail")
+			}
+			if !body.closed {
+				t.Error("prefetch setup failure left the response body open")
+			}
+			if f.resolveResp != nil {
+				t.Error("prefetch retained its failed resolve response")
+			}
+			if !f.prefetchDone.Load() {
+				t.Error("prefetch did not report completion")
+			}
+			if closedScope {
+				entries, err := os.ReadDir(root)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(entries) != 0 {
+					t.Errorf("late prefetch left temporary files: %v", entries)
+				}
+			}
+		})
+	}
+}