fix(pluginhost): avoid holding host lock during plugin lifecycle

2026-11-08 01:16:47 +08:00 · 2026-06-17 01:05:21 +08:00
parent 30dc2e7f34
commit 13f51d96cb
2 changed files with 252 additions and 12 deletions
@@ -3,7 +3,6 @@ package pluginhost
 import (
 	"context"
 	"fmt"
-	"runtime/debug"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -36,6 +35,7 @@ type pluginUnloadTarget struct {
 }

 type Host struct {
+	applyMu                sync.Mutex
 	mu                     sync.Mutex
 	loader                 pluginLoader
 	loaded                 map[string]*loadedPlugin
@@ -141,12 +141,16 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
 	if h == nil {
 		return
 	}
+	h.applyMu.Lock()
+	defer h.applyMu.Unlock()

 	rc := runtimeConfigFromConfig(cfg)
 	h.mu.Lock()
 	h.runtimeConfig = cfg
+	h.mu.Unlock()

 	if !rc.Enabled {
+		h.mu.Lock()
 		h.managementRoutes = make(map[string]managementRouteRecord)
 		h.resourceRoutes = make(map[string]resourceRouteRecord)
 		h.snapshot.Store(emptySnapshot())
@@ -158,6 +162,7 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
 	files, errSelect := selectPluginFiles(rc.Dir)
 	if errSelect != nil {
 		log.Warnf("pluginhost: failed to select plugin files: %v", errSelect)
+		h.mu.Lock()
 		h.managementRoutes = make(map[string]managementRouteRecord)
 		h.resourceRoutes = make(map[string]resourceRouteRecord)
 		h.snapshot.Store(emptySnapshot())
@@ -175,26 +180,33 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
 		if !item.Enabled {
 			continue
 		}
-		if _, disabled := h.fused[file.ID]; disabled {
+		h.mu.Lock()
+		lp := h.loaded[file.ID]
+		_, disabled := h.fused[file.ID]
+		h.mu.Unlock()
+		if disabled {
 			continue
 		}

-		lp := h.loaded[file.ID]
 		if lp == nil {
-			loaded, errLoad := h.loadLocked(file)
+			loaded, errLoad := h.load(file)
 			if errLoad != nil {
 				log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, errLoad)
 				continue
 			}
+			h.mu.Lock()
+			// ApplyConfig, UnloadPlugin, and ShutdownAll are serialized by applyMu,
+			// so a nil read cannot race into a duplicate load.
 			lp = loaded
 			h.loaded[file.ID] = lp
+			h.mu.Unlock()
 			log.WithFields(log.Fields{
 				"plugin_id": file.ID,
 				"path":      file.Path,
 			}).Info("pluginhost: plugin loaded")
 		}

-		plugin, okCall := h.callRegisterLocked(ctx, lp, item)
+		plugin, okCall := h.callRegister(ctx, lp, item)
 		if !okCall {
 			continue
 		}
@@ -208,12 +220,13 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
 	}

 	sortRecords(records)
+	h.mu.Lock()
 	h.snapshot.Store(&Snapshot{enabled: true, records: records})
 	h.mu.Unlock()
 	h.refreshThinkingProviders(records)
 }

-func (h *Host) loadLocked(file pluginFile) (*loadedPlugin, error) {
+func (h *Host) load(file pluginFile) (*loadedPlugin, error) {
 	client, errOpen := h.loader.Open(file, h)
 	if errOpen != nil {
 		return nil, errOpen
@@ -236,6 +249,9 @@ func (h *Host) UnloadPlugin(id string) bool {
 		return false
 	}

+	h.applyMu.Lock()
+	defer h.applyMu.Unlock()
+
 	var target pluginUnloadTarget
 	h.mu.Lock()
 	lp := h.loaded[id]
@@ -269,6 +285,9 @@ func (h *Host) ShutdownAll() {
 		return
 	}

+	h.applyMu.Lock()
+	defer h.applyMu.Unlock()
+
 	targets := make([]pluginUnloadTarget, 0)
 	h.mu.Lock()
 	for _, lp := range h.loaded {
@@ -346,17 +365,20 @@ func (h *Host) removePluginRuntimeStateLocked(id string) {
 	delete(h.modelRegistrations, id)
 }

-func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) {
+func (h *Host) callRegister(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) {
 	if lp == nil {
 		return pluginapi.Plugin{}, false
 	}

 	method := pluginabi.MethodPluginRegister
-	if lp.registered {
+	h.mu.Lock()
+	registered := lp.registered
+	h.mu.Unlock()
+	if registered {
 		method = pluginabi.MethodPluginReconfigure
 	}

-	plugin, okCall := h.safePluginCallLocked(ctx, lp.id, method, func() pluginapi.Plugin {
+	plugin, okCall := h.safePluginCall(ctx, lp.id, method, func() pluginapi.Plugin {
 		plugin, errRegister := registerRPCPlugin(ctx, h, lp.id, lp.client, method, item.ConfigYAML)
 		if errRegister != nil {
 			log.Warnf("pluginhost: plugin %s %s failed: %v", lp.id, method, errRegister)
@@ -367,7 +389,9 @@ func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item ru
 	if !okCall {
 		return pluginapi.Plugin{}, false
 	}
+	h.mu.Lock()
 	lp.registered = true
+	h.mu.Unlock()
 	if !validPlugin(plugin) {
 		log.Warnf("pluginhost: plugin %s returned invalid metadata or no capabilities", lp.id)
 		return pluginapi.Plugin{}, false
@@ -375,11 +399,10 @@ func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item ru
 	return plugin, true
 }

-func (h *Host) safePluginCallLocked(ctx context.Context, id, method string, fn func() pluginapi.Plugin) (out pluginapi.Plugin, ok bool) {
+func (h *Host) safePluginCall(ctx context.Context, id, method string, fn func() pluginapi.Plugin) (out pluginapi.Plugin, ok bool) {
 	defer func() {
 		if recovered := recover(); recovered != nil {
-			h.fused[id] = fmt.Sprintf("%s panic: %v", method, recovered)
-			log.WithField("plugin_id", id).WithField("method", method).Errorf("pluginhost: plugin panic recovered: %v\n%s", recovered, debug.Stack())
+			h.fusePlugin(id, method, recovered)
 			out = pluginapi.Plugin{}
 			ok = false
 		}
@@ -4,7 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"

 	"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
 	"github.com/router-for-me/CLIProxyAPI/v7/internal/thinking"
@@ -605,6 +608,168 @@ func TestHostApplyConfig_PanicFusesPluginForProcessLifetime(t *testing.T) {
 	}
 }

+func TestHostApplyConfigDoesNotHoldHostMuDuringRegister(t *testing.T) {
+	h, cfg, registerStarted, releaseRegister := newBlockingRegisterHost(t)
+	applyDone := make(chan struct{})
+	go func() {
+		h.ApplyConfig(context.Background(), cfg)
+		close(applyDone)
+	}()
+
+	waitForHostTestSignal(t, registerStarted, "register start")
+	probeDone := make(chan struct{})
+	go func() {
+		_ = h.currentModelExecutor()
+		close(probeDone)
+	}()
+	waitForHostTestSignal(t, probeDone, "Host.mu probe")
+
+	releaseRegister()
+	waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
+
+	snap := h.Snapshot()
+	if !snap.enabled || len(snap.records) != 1 || snap.records[0].id != "alpha" {
+		t.Fatalf("Snapshot() = %+v, want alpha registered", snap)
+	}
+}
+
+func TestHostApplyConfigSerializesLifecycleCalls(t *testing.T) {
+	loader := newTestSymbolLoader()
+	started := make(chan struct{})
+	release := make(chan struct{})
+	secondEntered := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseFirst := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseFirst)
+
+	var startOnce sync.Once
+	var secondOnce sync.Once
+	var lifecycleCalls int32
+	var activeLifecycleCalls int32
+	var concurrentLifecycleCalls int32
+	lifecycle := func([]byte) pluginapi.Plugin {
+		if active := atomic.AddInt32(&activeLifecycleCalls, 1); active > 1 {
+			atomic.StoreInt32(&concurrentLifecycleCalls, 1)
+		}
+		call := atomic.AddInt32(&lifecycleCalls, 1)
+		if call == 1 {
+			startOnce.Do(func() { close(started) })
+			<-release
+		} else {
+			secondOnce.Do(func() { close(secondEntered) })
+		}
+		atomic.AddInt32(&activeLifecycleCalls, -1)
+		return validTestPlugin("alpha")
+	}
+	plugin := &testPlugin{
+		registerResult:    validTestPlugin("alpha"),
+		reconfigureResult: validTestPlugin("alpha"),
+	}
+	lookup := newTestSymbolLookup(plugin)
+	lookup.registerOverride = lifecycle
+	lookup.reconfigureOverride = lifecycle
+	loader.lookups["alpha"] = lookup
+	h := NewForTest(loader)
+	cfg := &config.Config{
+		Plugins: config.PluginsConfig{
+			Enabled: true,
+			Dir:     makePluginDir(t, "alpha"),
+		},
+	}
+
+	firstDone := make(chan struct{})
+	go func() {
+		h.ApplyConfig(context.Background(), cfg)
+		close(firstDone)
+	}()
+	waitForHostTestSignal(t, started, "first register start")
+
+	secondDone := make(chan struct{})
+	go func() {
+		h.ApplyConfig(context.Background(), cfg)
+		close(secondDone)
+	}()
+	select {
+	case <-secondEntered:
+		t.Fatal("second ApplyConfig entered plugin lifecycle before first ApplyConfig finished")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	releaseFirst()
+	waitForHostTestSignal(t, firstDone, "first ApplyConfig completion")
+	waitForHostTestSignal(t, secondDone, "second ApplyConfig completion")
+
+	if got := atomic.LoadInt32(&lifecycleCalls); got != 2 {
+		t.Fatalf("lifecycle calls = %d, want 2", got)
+	}
+	if atomic.LoadInt32(&concurrentLifecycleCalls) != 0 {
+		t.Fatal("plugin lifecycle calls ran concurrently")
+	}
+}
+
+func TestHostUnloadAndShutdownWaitForBlockingRegister(t *testing.T) {
+	tests := []struct {
+		name       string
+		action     func(*Host) bool
+		assertDone func(*testing.T, *Host)
+	}{
+		{
+			name: "unload",
+			action: func(h *Host) bool {
+				return h.UnloadPlugin("alpha")
+			},
+			assertDone: func(t *testing.T, h *Host) {
+				t.Helper()
+				if h.PluginLoaded("alpha") {
+					t.Fatal("PluginLoaded(alpha) = true, want false after unload")
+				}
+			},
+		},
+		{
+			name: "shutdown",
+			action: func(h *Host) bool {
+				h.ShutdownAll()
+				return true
+			},
+			assertDone: func(t *testing.T, h *Host) {
+				t.Helper()
+				if h.PluginLoaded("alpha") {
+					t.Fatal("PluginLoaded(alpha) = true, want false after shutdown")
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			h, cfg, registerStarted, releaseRegister := newBlockingRegisterHost(t)
+			applyDone := make(chan struct{})
+			go func() {
+				h.ApplyConfig(context.Background(), cfg)
+				close(applyDone)
+			}()
+			waitForHostTestSignal(t, registerStarted, "register start")
+
+			actionDone := make(chan bool)
+			go func() {
+				actionDone <- tt.action(h)
+			}()
+			select {
+			case <-actionDone:
+				t.Fatalf("%s completed while ApplyConfig was still registering", tt.name)
+			case <-time.After(200 * time.Millisecond):
+			}
+
+			releaseRegister()
+			waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
+			if ok := waitForHostTestBool(t, actionDone, tt.name+" completion"); !ok {
+				t.Fatalf("%s returned false, want true", tt.name)
+			}
+			tt.assertDone(t, h)
+		})
+	}
+}
+
 func TestSortRecordsPriorityDescendingAndIDTieBreak(t *testing.T) {
 	records := []capabilityRecord{
 		{id: "charlie", priority: 1},
@@ -635,3 +800,55 @@ func (c *capturePluginClient) Call(ctx context.Context, method string, request [
 }

 func (c *capturePluginClient) Shutdown() {}
+
+func newBlockingRegisterHost(t *testing.T) (*Host, *config.Config, <-chan struct{}, func()) {
+	t.Helper()
+
+	loader := newTestSymbolLoader()
+	registerStarted := make(chan struct{})
+	release := make(chan struct{})
+	var startOnce sync.Once
+	var releaseOnce sync.Once
+	releaseRegister := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseRegister)
+
+	plugin := &testPlugin{
+		registerResult:    validTestPlugin("alpha"),
+		reconfigureResult: validTestPlugin("alpha"),
+	}
+	lookup := newTestSymbolLookup(plugin)
+	lookup.registerOverride = func([]byte) pluginapi.Plugin {
+		startOnce.Do(func() { close(registerStarted) })
+		<-release
+		return validTestPlugin("alpha")
+	}
+	loader.lookups["alpha"] = lookup
+	h := NewForTest(loader)
+	cfg := &config.Config{
+		Plugins: config.PluginsConfig{
+			Enabled: true,
+			Dir:     makePluginDir(t, "alpha"),
+		},
+	}
+	return h, cfg, registerStarted, releaseRegister
+}
+
+func waitForHostTestSignal(t *testing.T, ch <-chan struct{}, name string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for %s", name)
+	}
+}
+
+func waitForHostTestBool(t *testing.T, ch <-chan bool, name string) bool {
+	t.Helper()
+	select {
+	case ok := <-ch:
+		return ok
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for %s", name)
+		return false
+	}
+}