From 13f51d96cb2297c25c8227f932d9303daabef4d8 Mon Sep 17 00:00:00 2001 From: LTbinglingfeng Date: Wed, 17 Jun 2026 01:05:21 +0800 Subject: [PATCH] fix(pluginhost): avoid holding host lock during plugin lifecycle --- internal/pluginhost/host.go | 47 +++++-- internal/pluginhost/host_test.go | 217 +++++++++++++++++++++++++++++++ 2 files changed, 252 insertions(+), 12 deletions(-) diff --git a/internal/pluginhost/host.go b/internal/pluginhost/host.go index b9fc008a..83c82152 100644 --- a/internal/pluginhost/host.go +++ b/internal/pluginhost/host.go @@ -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 } diff --git a/internal/pluginhost/host_test.go b/internal/pluginhost/host_test.go index 2272da8e..df49bd86 100644 --- a/internal/pluginhost/host_test.go +++ b/internal/pluginhost/host_test.go @@ -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 + } +}