feat: unify native API invocation with async callbacks (#1476)

pkg/rest/invoke.go

@@ -13,10 +13,13 @@ const (
 	TaskEventError = uint64(goapi.TaskEventError)
 )
 
-// Invoke dispatches a Core API request without using an HTTP listener. It is
-// the shared entry point for Desktop FFI and gomobile bindings.
-func Invoke(method, path, rawQuery, body string) string {
+// Dispatch executes one in-process API request. Native scheduling and callback
+// behavior belong to bind/native; this function only bridges into the shared
+// API Service without opening an HTTP listener.
+func Dispatch(method, path, rawQuery, body string) string {
+	runtimeMu.RLock()
 	service := APIService
+	runtimeMu.RUnlock()
 	if service == nil {
 		return marshalInvokeResult(model.NewErrorResult("service not started"))
 	}
@@ -29,6 +32,8 @@ func Invoke(method, path, rawQuery, body string) string {
 }
 
 func SubscribeTaskEvents(mask uint64, listener func(payload string)) {
+	runtimeMu.RLock()
+	defer runtimeMu.RUnlock()
 	if APIService == nil {
 		return
 	}

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"
@@ -27,7 +30,7 @@ func TestNativeInvokeWithoutRESTListener(t *testing.T) {
 	}
 
 	var result model.Result[json.RawMessage]
-	if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
+	if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
 		t.Fatal(err)
 	}
 	if result.Code != model.CodeOk {
@@ -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 {
@@ -106,7 +178,7 @@ func TestAPIServerLifecycleUsesPersistedConfigWithoutRestartingCore(t *testing.T
 	}
 
 	var result model.Result[json.RawMessage]
-	if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
+	if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
 		t.Fatal(err)
 	}
 	if result.Code != model.CodeOk {
@@ -153,7 +225,7 @@ func TestRestartAPIServerFailsFastAndLeavesListenerStopped(t *testing.T) {
 	}
 
 	var result model.Result[json.RawMessage]
-	if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
+	if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
 		t.Fatal(err)
 	}
 	if result.Code != model.CodeOk {
@@ -229,7 +301,7 @@ func TestNativeAPIServerAutoStartFailureKeepsCoreAvailable(t *testing.T) {
 	}
 
 	var result model.Result[json.RawMessage]
-	if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
+	if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil {
 		t.Fatal(err)
 	}
 	if result.Code != model.CodeOk {

pkg/rest/server.go

@@ -29,7 +29,7 @@ import (
 )
 
 var (
-	runtimeMu sync.Mutex
+	runtimeMu sync.RWMutex
 
 	Downloader *download.Downloader
 	APIService *goapi.Service