fix: prevent native requests from blocking shutdown

bind/desktop/main.go

@@ -48,8 +48,7 @@ func Start(cfg *C.char) (int, *C.char) {
 
 //export Stop
 func Stop() {
-	nativebridge.PauseInvokesAndWait()
+	nativebridge.Stop()
-	rest.Stop()
 }
 
 //export GetAPIServerState

bind/mobile/main.go

@@ -27,8 +27,7 @@ func Start(cfg string) (int, error) {
 }
 
 func Stop() {
-	nativebridge.PauseInvokesAndWait()
+	nativebridge.Stop()
-	rest.Stop()
 }
 
 func GetAPIServerState() string {

bind/native/invoke.go

@@ -3,17 +3,18 @@
 package nativebridge
 
 import (
+	"errors"
 	"fmt"
-	"runtime"
 	"sync"
+	"time"
 
 	"github.com/GopeedLab/gopeed/pkg/rest"
 )
 
-const (
+// StopTimeout bounds native shutdown even when an in-process request never returns.
-	minInvokeConcurrency = 4
+const StopTimeout = 3 * time.Second
-	maxInvokeConcurrency = 32
+
-)
+var errNativeBridgeStopping = errors.New("native bridge is stopping")
 
 type invokeRequest struct {
 	method string
@@ -23,11 +24,19 @@ type invokeRequest struct {
 }
 
 type invokeTask struct {
+	id       uint64
 	request  invokeRequest
 	complete func(result string, err error)
+	once     sync.Once
 }
 
-// InvokeAsync queues an in-process API request and completes it through the
+func (t *invokeTask) finish(result string, err error) {
+	t.once.Do(func() {
+		completeCallback(t.complete, result, err)
+	})
+}
+
+// InvokeAsync starts an in-process API request and completes it through the
 // supplied callback. Desktop and mobile adapters share this executor.
 func InvokeAsync(method, path, query, body string, callback func(result string, err error)) {
 	if callback == nil {
@@ -43,7 +52,7 @@ func InvokeAsync(method, path, query, body string, callback func(result string,
 		complete: callback,
 	}
 	if !getInvokeExecutor().submit(task) {
-		completeCallback(callback, "", fmt.Errorf("native bridge is stopping"))
+		completeCallback(callback, "", errNativeBridgeStopping)
 	}
 }
 
@@ -52,10 +61,41 @@ func ResumeInvokes() {
 	getInvokeExecutor().resume()
 }
 
-// PauseInvokesAndWait prevents new submissions and waits for every queued or
+// PauseInvokesAndWait prevents new submissions and gives running requests a
-// running request to finish before a native runtime is stopped.
+// bounded opportunity to finish before a native runtime is stopped.
-func PauseInvokesAndWait() {
+// Requests still pending after StopTimeout are failed and detached so shutdown
-	getInvokeExecutor().pauseAndWait()
+// can continue without invoking their callbacks a second time.
+func PauseInvokesAndWait() bool {
+	return getInvokeExecutor().pauseAndWait(StopTimeout)
+}
+
+// Stop pauses native requests and shuts down the shared runtime, but never
+// keeps the host application waiting longer than StopTimeout. Cleanup that
+// cannot finish within the deadline is allowed to continue in the background.
+func Stop() {
+	stopWithin(getInvokeExecutor(), rest.Stop, StopTimeout)
+}
+
+func stopWithin(executor *invokeExecutor, cleanup func(), timeout time.Duration) {
+	deadline := time.Now().Add(timeout)
+	executor.pauseAndWait(time.Until(deadline))
+
+	stopped := make(chan struct{})
+	go func() {
+		defer close(stopped)
+		cleanup()
+	}()
+
+	remaining := time.Until(deadline)
+	if remaining <= 0 {
+		return
+	}
+	timer := time.NewTimer(remaining)
+	defer timer.Stop()
+	select {
+	case <-stopped:
+	case <-timer.C:
+	}
 }
 
 var (
@@ -65,64 +105,56 @@ var (
 
 func getInvokeExecutor() *invokeExecutor {
 	invokeExecutorOnce.Do(func() {
-		sharedInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string {
+		sharedInvoker = newInvokeExecutor(func(request invokeRequest) string {
 			return rest.Dispatch(request.method, request.path, request.query, request.body)
 		})
 	})
 	return sharedInvoker
 }
 
-// invokeExecutor keeps native API work concurrent without rejecting requests
+// invokeExecutor starts every native API request in its own goroutine, matching
-// when every worker is busy. Additional requests wait in FIFO order and are
+// net/http's request concurrency. It only tracks them to support bounded drain
-// completed through their callback.
+// during shutdown; one stuck request must never prevent another from starting.
 type invokeExecutor struct {
 	mu        sync.Mutex
-	ready     *sync.Cond
-	queue     []invokeTask
 	closed    bool
 	accepting bool
-	pending   int
+	nextID    uint64
-	workers   sync.WaitGroup
+	pending   map[uint64]*invokeTask
+	drained   chan struct{}
+	running   sync.WaitGroup
 	handler   func(invokeRequest) string
 }
 
-func invokeConcurrency(cpuCount int) int {
+func newInvokeExecutor(handler func(invokeRequest) string) *invokeExecutor {
-	concurrency := cpuCount * 2
+	drained := make(chan struct{})
-	if concurrency < minInvokeConcurrency {
+	close(drained)
-		return minInvokeConcurrency
+	return &invokeExecutor{
+		handler:   handler,
+		accepting: true,
+		pending:   make(map[uint64]*invokeTask),
+		drained:   drained,
 	}
-	if concurrency > maxInvokeConcurrency {
-		return maxInvokeConcurrency
-	}
-	return concurrency
-}
-
-func newInvokeExecutor(workerCount int, handler func(invokeRequest) string) *invokeExecutor {
-	if workerCount < 1 {
-		workerCount = 1
-	}
-	executor := &invokeExecutor{handler: handler, accepting: true}
-	executor.ready = sync.NewCond(&executor.mu)
-	executor.workers.Add(workerCount)
-	for range workerCount {
-		go executor.run()
-	}
-	return executor
-}
-
-func newDefaultInvokeExecutor(handler func(invokeRequest) string) *invokeExecutor {
-	return newInvokeExecutor(invokeConcurrency(runtime.GOMAXPROCS(0)), handler)
 }
 
 func (e *invokeExecutor) submit(task invokeTask) bool {
 	e.mu.Lock()
-	defer e.mu.Unlock()
 	if e.closed || !e.accepting {
+		e.mu.Unlock()
 		return false
 	}
-	e.queue = append(e.queue, task)
+	if len(e.pending) == 0 {
-	e.pending++
+		e.drained = make(chan struct{})
-	e.ready.Signal()
+	}
+	taskID := e.nextID
+	e.nextID++
+	task.id = taskID
+	taskPtr := &task
+	e.pending[taskID] = taskPtr
+	e.running.Add(1)
+	e.mu.Unlock()
+
+	go e.run(taskPtr)
 	return true
 }
 
@@ -132,57 +164,63 @@ func (e *invokeExecutor) resume() {
 	e.mu.Unlock()
 }
 
-func (e *invokeExecutor) pauseAndWait() {
+func (e *invokeExecutor) pauseAndWait(timeout time.Duration) bool {
 	e.mu.Lock()
 	e.accepting = false
-	for e.pending > 0 {
+	if len(e.pending) == 0 {
-		e.ready.Wait()
+		e.mu.Unlock()
+		return true
 	}
+	drained := e.drained
 	e.mu.Unlock()
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case <-drained:
+		return true
+	case <-timer.C:
+	}
+
+	e.mu.Lock()
+	if len(e.pending) == 0 {
+		e.mu.Unlock()
+		return true
+	}
+	pending := make([]*invokeTask, 0, len(e.pending))
+	for _, task := range e.pending {
+		pending = append(pending, task)
+	}
+	e.pending = make(map[uint64]*invokeTask)
+	close(e.drained)
+	e.mu.Unlock()
+
+	for _, task := range pending {
+		task.finish("", errNativeBridgeStopping)
+	}
+	return false
 }
 
 func (e *invokeExecutor) close() {
 	e.mu.Lock()
 	e.closed = true
-	e.ready.Broadcast()
+	e.accepting = false
 	e.mu.Unlock()
-	e.workers.Wait()
+	e.running.Wait()
 }
 
-func (e *invokeExecutor) run() {
+func (e *invokeExecutor) run(task *invokeTask) {
-	defer e.workers.Done()
+	defer e.running.Done()
-	for {
+	result, err := e.execute(task.request)
-		task, ok := e.take()
+	task.finish(result, err)
-		if !ok {
-			return
-		}
-		result, err := e.execute(task.request)
-		completeCallback(task.complete, result, err)
-		e.mu.Lock()
-		e.pending--
-		if e.pending == 0 {
-			e.ready.Broadcast()
-		}
-		e.mu.Unlock()
-	}
-}
-
-func (e *invokeExecutor) take() (invokeTask, bool) {
 	e.mu.Lock()
-	defer e.mu.Unlock()
+	if _, exists := e.pending[task.id]; exists {
-	for len(e.queue) == 0 && !e.closed {
+		delete(e.pending, task.id)
-		e.ready.Wait()
+		if len(e.pending) == 0 {
+			close(e.drained)
+		}
 	}
-	if len(e.queue) == 0 {
+	e.mu.Unlock()
-		return invokeTask{}, false
-	}
-	task := e.queue[0]
-	e.queue[0] = invokeTask{}
-	e.queue = e.queue[1:]
-	if len(e.queue) == 0 {
-		e.queue = nil
-	}
-	return task, true
 }
 
 func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) {

bind/native/invoke_test.go

@@ -4,52 +4,17 @@ import (
 	"runtime"
 	"strconv"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 )
 
-func TestInvokeConcurrency(t *testing.T) {
+func TestInvokeExecutorStartsEveryRequestIndependently(t *testing.T) {
-	tests := []struct {
+	const taskCount = 64
-		cpus int
-		want int
-	}{
-		{cpus: 1, want: 4},
-		{cpus: 2, want: 4},
-		{cpus: 4, want: 8},
-		{cpus: 8, want: 16},
-		{cpus: 16, want: 32},
-		{cpus: 64, want: 32},
-	}
-	for _, test := range tests {
-		t.Run(strconv.Itoa(test.cpus), func(t *testing.T) {
-			if got := invokeConcurrency(test.cpus); got != test.want {
-				t.Fatalf("invokeConcurrency(%d) = %d, want %d", test.cpus, got, test.want)
-			}
-		})
-	}
-}
-
-func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) {
-	const (
-		workerCount = 2
-		taskCount   = 6
-	)
 	release := make(chan struct{})
 	started := make(chan struct{}, taskCount)
-	var active atomic.Int32
+	executor := newInvokeExecutor(func(request invokeRequest) string {
-	var maxActive atomic.Int32
-	executor := newInvokeExecutor(workerCount, func(request invokeRequest) string {
-		current := active.Add(1)
-		for {
-			maximum := maxActive.Load()
-			if current <= maximum || maxActive.CompareAndSwap(maximum, current) {
-				break
-			}
-		}
 		started <- struct{}{}
 		<-release
-		active.Add(-1)
 		return request.path
 	})
 
@@ -63,20 +28,19 @@ func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) {
 			},
 		})
 		if !accepted {
-			t.Fatalf("task %d was rejected while all workers were busy", i)
+			t.Fatalf("request %d was rejected", i)
 		}
 	}
 
-	for range workerCount {
+	// Every handler must start before any of them is released. A worker pool
+	// would stall here once all of its workers were occupied.
+	for range taskCount {
 		select {
 		case <-started:
 		case <-time.After(time.Second):
-			t.Fatal("workers did not start queued tasks")
+			t.Fatal("a request waited behind other blocked requests")
 		}
 	}
-	if got := maxActive.Load(); got != workerCount {
-		t.Fatalf("maximum active tasks = %d, want %d", got, workerCount)
-	}
 
 	close(release)
 	completed.Wait()
@@ -84,7 +48,7 @@ func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) {
 }
 
 func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) {
-	executor := newInvokeExecutor(1, func(invokeRequest) string {
+	executor := newInvokeExecutor(func(invokeRequest) string {
 		panic("boom")
 	})
 	completed := make(chan error, 1)
@@ -103,7 +67,7 @@ func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) {
 }
 
 func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) {
-	executor := newInvokeExecutor(1, func(request invokeRequest) string {
+	executor := newInvokeExecutor(func(request invokeRequest) string {
 		return request.path
 	})
 	executor.submit(invokeTask{complete: func(string, error) {
@@ -131,7 +95,7 @@ func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) {
 func TestInvokeExecutorPausesAndDrains(t *testing.T) {
 	release := make(chan struct{})
 	started := make(chan struct{})
-	executor := newInvokeExecutor(1, func(request invokeRequest) string {
+	executor := newInvokeExecutor(func(request invokeRequest) string {
 		if request.path == "running" {
 			close(started)
 			<-release
@@ -151,7 +115,7 @@ func TestInvokeExecutorPausesAndDrains(t *testing.T) {
 
 	drained := make(chan struct{})
 	go func() {
-		executor.pauseAndWait()
+		executor.pauseAndWait(time.Second)
 		close(drained)
 	}()
 	deadline := time.Now().Add(time.Second)
@@ -194,3 +158,101 @@ func TestInvokeExecutorPausesAndDrains(t *testing.T) {
 	<-resumed
 	executor.close()
 }
+
+func TestInvokeExecutorPauseTimesOutAndFailsPendingRequests(t *testing.T) {
+	release := make(chan struct{})
+	started := make(chan struct{})
+	executor := newInvokeExecutor(func(request invokeRequest) string {
+		started <- struct{}{}
+		<-release
+		return request.path
+	})
+
+	errors := make(chan error, 2)
+	for _, path := range []string{"first", "second"} {
+		if !executor.submit(invokeTask{
+			request: invokeRequest{path: path},
+			complete: func(_ string, err error) {
+				errors <- err
+			},
+		}) {
+			t.Fatalf("request %q was rejected", path)
+		}
+	}
+	for range 2 {
+		<-started
+	}
+
+	const timeout = 30 * time.Millisecond
+	startedAt := time.Now()
+	if executor.pauseAndWait(timeout) {
+		t.Fatal("pause unexpectedly reported a complete drain")
+	}
+	if elapsed := time.Since(startedAt); elapsed < timeout || elapsed > 10*timeout {
+		t.Fatalf("pause returned after %s, want close to %s", elapsed, timeout)
+	}
+	for range 2 {
+		select {
+		case err := <-errors:
+			if err == nil || err.Error() != errNativeBridgeStopping.Error() {
+				t.Fatalf("unexpected timeout error: %v", err)
+			}
+		case <-time.After(time.Second):
+			t.Fatal("pending request was not failed after the timeout")
+		}
+	}
+
+	close(release)
+	executor.close()
+	select {
+	case err := <-errors:
+		t.Fatalf("request callback ran more than once: %v", err)
+	default:
+	}
+}
+
+func TestStopWithinUsesOneDeadlineForRequestsAndCleanup(t *testing.T) {
+	releaseRequest := make(chan struct{})
+	requestStarted := make(chan struct{})
+	executor := newInvokeExecutor(func(invokeRequest) string {
+		close(requestStarted)
+		<-releaseRequest
+		return "done"
+	})
+	requestResult := make(chan error, 1)
+	executor.submit(invokeTask{complete: func(_ string, err error) {
+		requestResult <- err
+	}})
+	<-requestStarted
+
+	releaseCleanup := make(chan struct{})
+	cleanupStarted := make(chan struct{})
+	cleanup := func() {
+		close(cleanupStarted)
+		<-releaseCleanup
+	}
+
+	const timeout = 40 * time.Millisecond
+	startedAt := time.Now()
+	stopWithin(executor, cleanup, timeout)
+	if elapsed := time.Since(startedAt); elapsed < timeout || elapsed > 10*timeout {
+		t.Fatalf("stop returned after %s, want close to %s", elapsed, timeout)
+	}
+	select {
+	case <-cleanupStarted:
+	case <-time.After(time.Second):
+		t.Fatal("cleanup was not started after the request deadline")
+	}
+	select {
+	case err := <-requestResult:
+		if err == nil || err.Error() != errNativeBridgeStopping.Error() {
+			t.Fatalf("unexpected request result: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed-out request was not failed")
+	}
+
+	close(releaseRequest)
+	close(releaseCleanup)
+	executor.close()
+}

pkg/rest/invoke.go

@@ -18,8 +18,8 @@ const (
 // API Service without opening an HTTP listener.
 func Dispatch(method, path, rawQuery, body string) string {
 	runtimeMu.RLock()
-	defer runtimeMu.RUnlock()
 	service := APIService
+	runtimeMu.RUnlock()
 	if service == nil {
 		return marshalInvokeResult(model.NewErrorResult("service not started"))
 	}

pkg/rest/invoke_test.go

@@ -4,9 +4,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/http/httptest"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/GopeedLab/gopeed/pkg/base"
 	"github.com/GopeedLab/gopeed/pkg/rest/model"
@@ -35,6 +38,75 @@ func TestNativeInvokeWithoutRESTListener(t *testing.T) {
 	}
 }
 
+func TestStartAPIServerIsNotBlockedByPendingDispatch(t *testing.T) {
+	requestStarted := make(chan struct{})
+	releaseRequest := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseRequest) }) }
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		close(requestStarted)
+		<-releaseRequest
+		w.Header().Set("Content-Length", "0")
+	}))
+	t.Cleanup(func() {
+		release()
+		upstream.Close()
+		Stop()
+	})
+
+	if _, err := Start(&model.StartConfig{NativeMode: true, Storage: model.StorageMem}); err != nil {
+		t.Fatal(err)
+	}
+	requestBody, err := json.Marshal(&model.ResolveTask{Req: &base.Request{URL: upstream.URL}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	dispatchDone := make(chan struct{})
+	go func() {
+		defer close(dispatchDone)
+		Dispatch(http.MethodPost, "/api/v1/resolve", "", string(requestBody))
+	}()
+	select {
+	case <-requestStarted:
+	case <-time.After(time.Second):
+		t.Fatal("resolve request did not start")
+	}
+
+	config, err := Downloader.GetConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	config.API = &base.APIServerConfig{Enable: true, Network: "tcp", Address: "127.0.0.1:0"}
+	if err := Downloader.PutConfig(config); err != nil {
+		t.Fatal(err)
+	}
+	started := make(chan *model.APIServerState, 1)
+	startErrors := make(chan error, 1)
+	go func() {
+		state, err := StartAPIServer()
+		started <- state
+		startErrors <- err
+	}()
+	select {
+	case state := <-started:
+		if err := <-startErrors; err != nil {
+			t.Fatalf("start API server failed: %v", err)
+		}
+		if !state.Running || state.RunningPort == 0 {
+			t.Fatalf("API server did not start: %#v", state)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("pending Resolve blocked API server startup")
+	}
+
+	release()
+	select {
+	case <-dispatchDone:
+	case <-time.After(time.Second):
+		t.Fatal("resolve request did not finish after release")
+	}
+}
+
 func TestAPIServerLifecycleUsesPersistedConfigWithoutRestartingCore(t *testing.T) {
 	port, err := Start(&model.StartConfig{NativeMode: true, Storage: model.StorageMem})
 	if err != nil {

ui/flutter/lib/core/ffi/libgopeed_worker.dart

@@ -26,10 +26,8 @@ class LibgopeedWorker {
   final ReceivePort _errorPort = ReceivePort();
   final Map<int, Completer<Object?>> _pending = {};
   final StreamController<Map<String, dynamic>> _taskEvents = StreamController<Map<String, dynamic>>.broadcast();
-  final Set<int> _invokeRequests = {};
   late final Future<Isolate> _isolateReady;
   final Completer<SendPort> _sendPort = Completer<SendPort>();
-  Completer<void>? _invokeDrain;
   Future<void>? _stopFuture;
   int _requestID = 0;
   bool _stopping = false;
@@ -43,19 +41,22 @@ class LibgopeedWorker {
 
   Future<void> _stop() async {
     _stopping = true;
-    if (_invokeRequests.isNotEmpty) {
+    // The native bridge owns the bounded drain. Sending Stop immediately is
-      _invokeDrain ??= Completer<void>();
+    // important because an InvokeAsync handler such as Resolve may never return.
-      await _invokeDrain!.future;
-    }
     await _request<Object?>('stop', const [], allowWhileStopping: true);
     _stopped = true;
+    final stoppingError = StateError('libgopeed worker stopped before the request completed');
+    for (final completer in _pending.values) {
+      completer.completeError(stoppingError);
+    }
+    _pending.clear();
     await _taskEvents.close();
     _receivePort.close();
     _errorPort.close();
   }
 
   Future<Object?> invoke(String method, String path, String query, String body) {
-    return _request<Object?>('invoke', [method, path, query, body], trackInvoke: true);
+    return _request<Object?>('invoke', [method, path, query, body]);
   }
 
   Future<String> apiServer(String operation) => _request<String>('apiServer', [operation]);
@@ -64,12 +65,7 @@ class LibgopeedWorker {
     await _request<Object?>('subscribeTaskEvents', [mask]);
   }
 
-  Future<T> _request<T>(
+  Future<T> _request<T>(String type, List<Object?> arguments, {bool allowWhileStopping = false}) async {
-    String type,
-    List<Object?> arguments, {
-    bool trackInvoke = false,
-    bool allowWhileStopping = false,
-  }) async {
     if (_stopped || (_stopping && !allowWhileStopping)) {
       throw StateError('libgopeed worker is stopping');
     }
@@ -78,7 +74,6 @@ class LibgopeedWorker {
     final requestID = _requestID++;
     final completer = Completer<Object?>();
     _pending[requestID] = completer;
-    if (trackInvoke) _invokeRequests.add(requestID);
     sendPort.send([type, requestID, ...arguments]);
     return (await completer.future) as T;
   }
@@ -94,10 +89,6 @@ class LibgopeedWorker {
         final requestID = response[1] as int;
         final completer = _pending.remove(requestID);
         if (completer == null) return;
-        if (_invokeRequests.remove(requestID) && _invokeRequests.isEmpty) {
-          _invokeDrain?.complete();
-          _invokeDrain = null;
-        }
         if (response[2] as bool) {
           completer.complete(response[3]);
         } else {
@@ -116,9 +107,6 @@ class LibgopeedWorker {
       completer.completeError(error);
     }
     _pending.clear();
-    _invokeRequests.clear();
-    _invokeDrain?.completeError(error);
-    _invokeDrain = null;
     _taskEvents.addError(error);
   }