diff --git a/internal/pluginhost/host.go b/internal/pluginhost/host.go index 0fc56cf1..dd6cfd1a 100644 --- a/internal/pluginhost/host.go +++ b/internal/pluginhost/host.go @@ -1,7 +1,9 @@ package pluginhost import ( + "bytes" "context" + "errors" "fmt" "path/filepath" "sort" @@ -23,6 +25,8 @@ type loadedPlugin struct { path string version string name string + configYAML []byte + plugin pluginapi.Plugin registered bool client pluginClient } @@ -276,15 +280,40 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) { } h.loading[file.ID] = request h.mu.Unlock() + + if replaced != nil { + h.callQuiesce(ctx, replaced) + if errContext := ctx.Err(); errContext != nil { + h.clearLoadingRequest(file.ID, request) + _, _, _ = h.rollbackReplacement(replaced, item) + return + } + } h.startPluginLoad(ctx, file, item, request) - loadResult, completed := h.waitForPluginLoad(ctx, file.ID, request) + loadResult, completed := h.waitForPluginLoad(ctx, request) if !completed { + if replaced == nil { + h.cleanupCanceledPluginLoad(file.ID, request) + } else { + h.cleanupCanceledPluginLoadAndWait(file.ID, request) + _, _, _ = h.rollbackReplacement(replaced, item) + } return } - if loadResult.err != nil { - h.cleanupPluginLoad(file.ID, request, loadResult.loaded) - log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, loadResult.err) + if loadResult.err != nil || (replaced != nil && !loadResult.initialized) { + if replaced == nil { + h.cleanupPluginLoad(file.ID, request, loadResult.loaded) + } else { + h.cleanupPluginLoadAndWait(file.ID, request, loadResult.loaded) + if rollbackRecord, rollbackFile, okRollback := h.rollbackReplacement(replaced, item); okRollback { + records = append(records, rollbackRecord) + loadedFiles = append(loadedFiles, rollbackFile) + } + } + if loadResult.err != nil { + log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, loadResult.err) + } continue } @@ -292,11 +321,19 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) { if h.loading[file.ID] != request { h.mu.Unlock() h.discardLoadedPlugin(loadResult.loaded) + if replaced != nil { + _, _, _ = h.rollbackReplacement(replaced, item) + } return } if errContext := ctx.Err(); errContext != nil { h.mu.Unlock() - h.cleanupPluginLoad(file.ID, request, loadResult.loaded) + if replaced == nil { + h.cleanupPluginLoad(file.ID, request, loadResult.loaded) + } else { + h.cleanupPluginLoadAndWait(file.ID, request, loadResult.loaded) + _, _, _ = h.rollbackReplacement(replaced, item) + } return } delete(h.loading, file.ID) @@ -399,7 +436,7 @@ func (h *Host) startPluginLoad(ctx context.Context, file pluginFile, item runtim }() } -func (h *Host) waitForPluginLoad(ctx context.Context, id string, request *pluginLoadRequest) (pluginLoadResult, bool) { +func (h *Host) waitForPluginLoad(ctx context.Context, request *pluginLoadRequest) (pluginLoadResult, bool) { if h == nil || request == nil || request.result == nil { return pluginLoadResult{}, false } @@ -410,7 +447,6 @@ func (h *Host) waitForPluginLoad(ctx context.Context, id string, request *plugin case result := <-request.result: return result, true case <-ctx.Done(): - h.cleanupCanceledPluginLoad(id, request) return pluginLoadResult{}, false } } @@ -433,6 +469,23 @@ func (h *Host) cleanupCanceledPluginLoad(id string, request *pluginLoadRequest) }() } +func (h *Host) cleanupCanceledPluginLoadAndWait(id string, request *pluginLoadRequest) { + if h == nil || request == nil || request.result == nil { + return + } + h.mu.Lock() + if h.loading[id] != request || request.cleanupStarted { + h.mu.Unlock() + return + } + request.cleanupStarted = true + h.mu.Unlock() + + result := <-request.result + h.discardLoadedPlugin(result.loaded) + h.clearLoadingRequest(id, request) +} + // cleanupPluginLoad retains the matching load token until the client has physically // shut down, preventing a replacement ApplyConfig from opening a second client. func (h *Host) cleanupPluginLoad(id string, request *pluginLoadRequest, loaded *loadedPlugin) { @@ -450,6 +503,22 @@ func (h *Host) cleanupPluginLoad(id string, request *pluginLoadRequest, loaded * h.finishPluginLoadCleanup(id, request, loaded) } +func (h *Host) cleanupPluginLoadAndWait(id string, request *pluginLoadRequest, loaded *loadedPlugin) { + if h == nil || request == nil { + return + } + h.mu.Lock() + if h.loading[id] != request || request.cleanupStarted { + h.mu.Unlock() + return + } + request.cleanupStarted = true + h.mu.Unlock() + + h.discardLoadedPlugin(loaded) + h.clearLoadingRequest(id, request) +} + func (h *Host) finishPluginLoadCleanup(id string, request *pluginLoadRequest, loaded *loadedPlugin) { go func() { h.discardLoadedPlugin(loaded) @@ -779,6 +848,99 @@ func (h *Host) rebuildActivePluginMapsLocked(records []capabilityRecord) { } } +func (h *Host) callQuiesce(ctx context.Context, lp *loadedPlugin) bool { + if h == nil || lp == nil || lp.client == nil { + return false + } + errQuiesce, okCall := h.safePluginAction(ctx, lp.id, pluginabi.MethodPluginQuiesce, func() error { + _, errCall := callPlugin[rpcEmptyResponse](ctx, lp.client, pluginabi.MethodPluginQuiesce, rpcEmptyResponse{}) + return errCall + }) + if errQuiesce != nil { + logQuiesceError(lp.id, errQuiesce) + } + return okCall && errQuiesce == nil +} + +func logQuiesceError(id string, errQuiesce error) { + entry := log.WithError(errQuiesce).WithField("plugin_id", id) + switch { + case quiesceUnsupported(errQuiesce): + entry.Debug("pluginhost: plugin quiesce unsupported") + case errors.Is(errQuiesce, context.Canceled), errors.Is(errQuiesce, context.DeadlineExceeded): + entry.Debug("pluginhost: plugin quiesce canceled") + default: + entry.Warn("pluginhost: plugin quiesce failed") + } +} + +func quiesceUnsupported(errQuiesce error) bool { + if errQuiesce == nil { + return false + } + var errRPC rpcError + if errors.As(errQuiesce, &errRPC) { + switch strings.ToLower(strings.TrimSpace(errRPC.Code)) { + case "unknown_method", "method_not_found", "unsupported_method": + return true + } + } + if errors.Is(errQuiesce, errors.ErrUnsupported) { + return true + } + message := strings.ToLower(strings.TrimSpace(errQuiesce.Error())) + for _, indication := range []string{ + "unknown method", + "method not found", + "unsupported method", + "method unsupported", + "method is not supported", + "method not supported", + } { + if strings.Contains(message, indication) { + return true + } + } + return false +} + +func (h *Host) rollbackReplacement(lp *loadedPlugin, item runtimeItemConfig) (capabilityRecord, pluginFile, bool) { + if h == nil || lp == nil { + return capabilityRecord{}, pluginFile{}, false + } + h.mu.Lock() + configYAML := bytes.Clone(lp.configYAML) + previousPlugin := lp.plugin + h.mu.Unlock() + if len(configYAML) > 0 { + item.ConfigYAML = configYAML + } + + plugin, okCall := h.callRegister(context.Background(), lp, item) + if okCall { + h.mu.Lock() + delete(h.fused, lp.id) + h.mu.Unlock() + } else { + plugin = previousPlugin + } + if !validPlugin(plugin) { + return capabilityRecord{}, pluginFile{}, false + } + return capabilityRecord{ + id: lp.id, + path: lp.path, + version: lp.version, + priority: item.Priority, + meta: plugin.Metadata, + plugin: plugin, + }, pluginFile{ + ID: lp.id, + Path: lp.path, + Version: lp.version, + }, true +} + func (h *Host) callRegister(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) { if lp == nil { return pluginapi.Plugin{}, false @@ -810,9 +972,37 @@ func (h *Host) callRegister(ctx context.Context, lp *loadedPlugin, item runtimeI log.Warnf("pluginhost: plugin %s returned invalid metadata or no capabilities", lp.id) return pluginapi.Plugin{}, false } + plugin.Metadata = clonePluginMetadata(plugin.Metadata) + h.mu.Lock() + lp.name = strings.TrimSpace(plugin.Metadata.Name) + if strings.TrimSpace(lp.version) == "" { + lp.version = strings.TrimSpace(plugin.Metadata.Version) + } + lp.configYAML = bytes.Clone(item.ConfigYAML) + lp.plugin = plugin + h.mu.Unlock() return plugin, true } +func (h *Host) safePluginAction(ctx context.Context, id, method string, fn func() error) (err error, ok bool) { + defer func() { + if recovered := recover(); recovered != nil { + h.fusePlugin(id, method, recovered) + err = nil + ok = false + } + }() + + if ctx != nil { + select { + case <-ctx.Done(): + return ctx.Err(), false + default: + } + } + return fn(), true +} + 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 { diff --git a/internal/pluginhost/host_test.go b/internal/pluginhost/host_test.go index 788a2855..16ae7eab 100644 --- a/internal/pluginhost/host_test.go +++ b/internal/pluginhost/host_test.go @@ -7,6 +7,7 @@ import ( "fmt" "net/http" "path/filepath" + "slices" "strings" "sync" "sync/atomic" @@ -751,6 +752,451 @@ func TestHostApplyConfigLogsHotReloadActiveAndRetiredVersions(t *testing.T) { } } +func TestHostApplyConfigQuiescesBeforeHotReloadRegistration(t *testing.T) { + events := &lifecycleEventRecorder{} + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + return marshalRPCResult(rpcEmptyResponse{}) + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + replacementClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("replacement." + method) + if method != pluginabi.MethodPluginRegister { + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + return lifecycleRegistrationResult(validTestPlugin("alpha")) + }} + loader := &sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}} + h := NewForTest(loader) + t.Cleanup(h.ShutdownAll) + pluginsDir, paths := makeVersionedPluginDir(t, "alpha", "1.0.0") + + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + paths["2.0.0"] = writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "2.0.0")) + + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } + if !h.pluginIdentityCurrent("alpha", paths["2.0.0"], "2.0.0") { + t.Fatal("replacement plugin did not become active") + } + h.mu.Lock() + retiredCount := len(h.retired["alpha"]) + h.mu.Unlock() + if retiredCount != 1 { + t.Fatalf("retired plugin count = %d, want 1", retiredCount) + } +} + +func TestHostApplyConfigRollsBackQuiescedPluginWhenReplacementFails(t *testing.T) { + events := &lifecycleEventRecorder{} + var configMu sync.Mutex + var registeredConfig []byte + var rollbackConfig []byte + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, request []byte) ([]byte, error) { + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister, pluginabi.MethodPluginReconfigure: + var lifecycleRequest rpcLifecycleRequest + if errUnmarshal := json.Unmarshal(request, &lifecycleRequest); errUnmarshal != nil { + return nil, errUnmarshal + } + configMu.Lock() + if method == pluginabi.MethodPluginRegister { + registeredConfig = bytes.Clone(lifecycleRequest.ConfigYAML) + } else { + rollbackConfig = bytes.Clone(lifecycleRequest.ConfigYAML) + } + configMu.Unlock() + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + return marshalRPCResult(rpcEmptyResponse{}) + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + replacementClient := &lifecycleTestClient{ + call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("replacement." + method) + if method != pluginabi.MethodPluginRegister { + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + return nil, fmt.Errorf("replacement registration failed") + }, + shutdown: func() { events.add("replacement.shutdown") }, + } + loader := &sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}} + h := NewForTest(loader) + t.Cleanup(h.ShutdownAll) + pluginsDir, paths := makeVersionedPluginDir(t, "alpha", "1.0.0") + + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "2.0.0")) + + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + "replacement.shutdown", + "old." + pluginabi.MethodPluginReconfigure, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } + configMu.Lock() + gotRegisteredConfig := bytes.Clone(registeredConfig) + gotRollbackConfig := bytes.Clone(rollbackConfig) + configMu.Unlock() + if !bytes.Equal(gotRollbackConfig, gotRegisteredConfig) { + t.Fatalf("rollback config = %q, want original config %q", gotRollbackConfig, gotRegisteredConfig) + } + if replacementClient.shutdownCalls.Load() != 1 { + t.Fatalf("replacement shutdown calls = %d, want 1", replacementClient.shutdownCalls.Load()) + } + if !h.pluginIdentityCurrent("alpha", paths["1.0.0"], "1.0.0") { + t.Fatal("old plugin did not remain active after rollback") + } + h.mu.Lock() + retiredCount := len(h.retired["alpha"]) + h.mu.Unlock() + if retiredCount != 0 { + t.Fatalf("retired plugin count = %d, want 0", retiredCount) + } +} + +func TestHostApplyConfigFallsBackWhenQuiesceUnsupported(t *testing.T) { + events := &lifecycleEventRecorder{} + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + return marshalRPCError("unknown_method", "unknown method plugin.quiesce"), nil + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + replacementClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("replacement." + method) + if method != pluginabi.MethodPluginRegister { + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + return lifecycleRegistrationResult(validTestPlugin("alpha")) + }} + h := NewForTest(&sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}}) + t.Cleanup(h.ShutdownAll) + pluginsDir, paths := makeVersionedPluginDir(t, "alpha", "1.0.0") + + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + paths["2.0.0"] = writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "2.0.0")) + + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } + if !h.pluginIdentityCurrent("alpha", paths["2.0.0"], "2.0.0") { + t.Fatal("unsupported quiesce prevented standard hot reload") + } +} + +func TestHostCallQuiesceClassifiesErrors(t *testing.T) { + var out bytes.Buffer + originalOut := log.StandardLogger().Out + originalFormatter := log.StandardLogger().Formatter + originalLevel := log.GetLevel() + log.SetOutput(&out) + log.SetFormatter(&log.TextFormatter{ + DisableColors: true, + DisableTimestamp: true, + }) + log.SetLevel(log.DebugLevel) + t.Cleanup(func() { + log.SetOutput(originalOut) + log.SetFormatter(originalFormatter) + log.SetLevel(originalLevel) + }) + + tests := []struct { + name string + response []byte + errCall error + wantMessage string + wantLevel string + }{ + { + name: "unknown method RPC error", + response: marshalRPCError("unknown_method", "plugin.quiesce is unavailable"), + wantMessage: "pluginhost: plugin quiesce unsupported", + wantLevel: "level=debug", + }, + { + name: "standard unsupported indication", + errCall: fmt.Errorf("method not found: plugin.quiesce"), + wantMessage: "pluginhost: plugin quiesce unsupported", + wantLevel: "level=debug", + }, + { + name: "runtime failure", + errCall: fmt.Errorf("quiesce runtime failure"), + wantMessage: "pluginhost: plugin quiesce failed", + wantLevel: "level=warning", + }, + { + name: "context cancellation", + errCall: context.Canceled, + wantMessage: "pluginhost: plugin quiesce canceled", + wantLevel: "level=debug", + }, + } + + h := New() + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + out.Reset() + client := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + if method != pluginabi.MethodPluginQuiesce { + return nil, fmt.Errorf("unexpected plugin method %s", method) + } + return tt.response, tt.errCall + }} + if h.callQuiesce(context.Background(), &loadedPlugin{id: "alpha", client: client}) { + t.Fatal("callQuiesce() = true, want false") + } + logs := out.String() + if !strings.Contains(logs, tt.wantMessage) || !strings.Contains(logs, tt.wantLevel) { + t.Fatalf("quiesce log = %q, want %q at %s", logs, tt.wantMessage, tt.wantLevel) + } + }) + } +} + +func TestHostApplyConfigSerializesLifecycleDuringQuiesce(t *testing.T) { + events := &lifecycleEventRecorder{} + quiesceStarted := make(chan struct{}) + releaseQuiesce := make(chan struct{}) + replacementRegisterStarted := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseQuiesce) }) } + t.Cleanup(release) + + var activeLifecycleCalls atomic.Int32 + var concurrentLifecycleCalls atomic.Bool + enterLifecycle := func() func() { + if activeLifecycleCalls.Add(1) > 1 { + concurrentLifecycleCalls.Store(true) + } + return func() { activeLifecycleCalls.Add(-1) } + } + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + exitLifecycle := enterLifecycle() + defer exitLifecycle() + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + close(quiesceStarted) + <-releaseQuiesce + return marshalRPCResult(rpcEmptyResponse{}) + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + var replacementRegisterOnce sync.Once + replacementClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + exitLifecycle := enterLifecycle() + defer exitLifecycle() + events.add("replacement." + method) + switch method { + case pluginabi.MethodPluginRegister: + replacementRegisterOnce.Do(func() { close(replacementRegisterStarted) }) + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginReconfigure: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + default: + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + }} + loader := &sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}} + h := NewForTest(loader) + t.Cleanup(h.ShutdownAll) + pluginsDir, _ := makeVersionedPluginDir(t, "alpha", "1.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + cfg := versionedPluginHostConfig(t, pluginsDir, "2.0.0") + + firstDone := make(chan struct{}) + go func() { + h.ApplyConfig(context.Background(), cfg) + close(firstDone) + }() + waitForHostTestSignal(t, quiesceStarted, "plugin quiesce") + + probeDone := make(chan struct{}) + go func() { + _ = h.currentModelExecutor() + close(probeDone) + }() + waitForHostTestSignal(t, probeDone, "Host.mu probe during quiesce") + + secondDone := make(chan struct{}) + go func() { + h.ApplyConfig(context.Background(), cfg) + close(secondDone) + }() + select { + case <-replacementRegisterStarted: + t.Fatal("replacement registration started before quiesce completed") + case <-secondDone: + t.Fatal("second ApplyConfig completed while quiesce was blocked") + case <-time.After(200 * time.Millisecond): + } + + release() + waitForHostTestSignal(t, firstDone, "first hot reload") + waitForHostTestSignal(t, secondDone, "serialized reconfigure") + if concurrentLifecycleCalls.Load() { + t.Fatal("plugin lifecycle calls ran concurrently during quiesce") + } + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + "replacement." + pluginabi.MethodPluginReconfigure, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } +} + +func TestHostApplyConfigRollsBackQuiescedPluginWhenContextCanceled(t *testing.T) { + events := &lifecycleEventRecorder{} + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister, pluginabi.MethodPluginReconfigure: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + return marshalRPCResult(rpcEmptyResponse{}) + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + replacementRegisterStarted := make(chan struct{}) + replacementClient := &lifecycleTestClient{ + call: func(ctx context.Context, method string, _ []byte) ([]byte, error) { + events.add("replacement." + method) + if method != pluginabi.MethodPluginRegister { + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + close(replacementRegisterStarted) + <-ctx.Done() + return nil, ctx.Err() + }, + shutdown: func() { events.add("replacement.shutdown") }, + } + h := NewForTest(&sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}}) + t.Cleanup(h.ShutdownAll) + pluginsDir, paths := makeVersionedPluginDir(t, "alpha", "1.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + + cfg := versionedPluginHostConfig(t, pluginsDir, "2.0.0") + ctx, cancel := context.WithCancel(context.Background()) + applyDone := make(chan struct{}) + go func() { + h.ApplyConfig(ctx, cfg) + close(applyDone) + }() + waitForHostTestSignal(t, replacementRegisterStarted, "replacement registration") + cancel() + waitForHostTestSignal(t, applyDone, "canceled hot reload") + + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + "replacement.shutdown", + "old." + pluginabi.MethodPluginReconfigure, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } + if !h.pluginIdentityCurrent("alpha", paths["1.0.0"], "1.0.0") { + t.Fatal("old plugin did not remain active after canceled hot reload") + } + if replacementClient.shutdownCalls.Load() != 1 { + t.Fatalf("replacement shutdown calls = %d, want 1", replacementClient.shutdownCalls.Load()) + } +} + +func TestHostApplyConfigShutsDownSuccessfulReplacementBeforeCanceledRollback(t *testing.T) { + events := &lifecycleEventRecorder{} + oldClient := &lifecycleTestClient{call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("old." + method) + switch method { + case pluginabi.MethodPluginRegister, pluginabi.MethodPluginReconfigure: + return lifecycleRegistrationResult(validTestPlugin("alpha")) + case pluginabi.MethodPluginQuiesce: + return marshalRPCResult(rpcEmptyResponse{}) + default: + return nil, fmt.Errorf("unexpected old plugin method %s", method) + } + }} + replacementClient := &lifecycleTestClient{ + call: func(_ context.Context, method string, _ []byte) ([]byte, error) { + events.add("replacement." + method) + if method != pluginabi.MethodPluginRegister { + return nil, fmt.Errorf("unexpected replacement plugin method %s", method) + } + return lifecycleRegistrationResult(validTestPlugin("alpha")) + }, + shutdown: func() { events.add("replacement.shutdown") }, + } + h := NewForTest(&sequencePluginLoader{clients: []pluginClient{oldClient, replacementClient}}) + t.Cleanup(h.ShutdownAll) + pluginsDir, paths := makeVersionedPluginDir(t, "alpha", "1.0.0") + h.ApplyConfig(context.Background(), versionedPluginHostConfig(t, pluginsDir, "1.0.0")) + writeVersionedPluginFile(t, pluginsDir, "alpha", "2.0.0") + + ctx := &cancelOnErrContext{ + Context: context.Background(), + done: make(chan struct{}), + cancelAt: 3, + } + h.ApplyConfig(ctx, versionedPluginHostConfig(t, pluginsDir, "2.0.0")) + + if got, want := events.snapshot(), []string{ + "old." + pluginabi.MethodPluginRegister, + "old." + pluginabi.MethodPluginQuiesce, + "replacement." + pluginabi.MethodPluginRegister, + "replacement.shutdown", + "old." + pluginabi.MethodPluginReconfigure, + }; !slices.Equal(got, want) { + t.Fatalf("lifecycle events = %v, want %v", got, want) + } + if replacementClient.shutdownCalls.Load() != 1 { + t.Fatalf("replacement shutdown calls = %d, want 1", replacementClient.shutdownCalls.Load()) + } + if !h.pluginIdentityCurrent("alpha", paths["1.0.0"], "1.0.0") { + t.Fatal("old plugin did not remain active after canceled replacement") + } +} + func TestHostApplyConfigKeepsLoadedVersionWhenPinnedVersionMissing(t *testing.T) { loader := newTestSymbolLoader() plugin := &testPlugin{ @@ -1428,6 +1874,99 @@ func TestSortRecordsPriorityDescendingAndIDTieBreak(t *testing.T) { } } +type lifecycleEventRecorder struct { + mu sync.Mutex + events []string +} + +func (r *lifecycleEventRecorder) add(event string) { + r.mu.Lock() + r.events = append(r.events, event) + r.mu.Unlock() +} + +func (r *lifecycleEventRecorder) snapshot() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.events...) +} + +type lifecycleTestClient struct { + call func(context.Context, string, []byte) ([]byte, error) + shutdown func() + shutdownCalls atomic.Int32 +} + +func (c *lifecycleTestClient) Call(ctx context.Context, method string, request []byte) ([]byte, error) { + if c == nil || c.call == nil { + return nil, fmt.Errorf("lifecycle test client has no call handler") + } + return c.call(ctx, method, request) +} + +func (c *lifecycleTestClient) Shutdown() { + c.shutdownCalls.Add(1) + if c.shutdown != nil { + c.shutdown() + } +} + +type cancelOnErrContext struct { + context.Context + done chan struct{} + cancelAt int32 + errCalls atomic.Int32 + cancel sync.Once +} + +func (c *cancelOnErrContext) Done() <-chan struct{} { + return c.done +} + +func (c *cancelOnErrContext) Err() error { + if c.errCalls.Add(1) < c.cancelAt { + return nil + } + c.cancel.Do(func() { close(c.done) }) + return context.Canceled +} + +type sequencePluginLoader struct { + mu sync.Mutex + clients []pluginClient + calls int +} + +func (l *sequencePluginLoader) Open(file pluginFile, _ *Host) (pluginClient, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.calls >= len(l.clients) { + return nil, fmt.Errorf("missing test client for %s", file.Path) + } + client := l.clients[l.calls] + l.calls++ + return client, nil +} + +func lifecycleRegistrationResult(plugin pluginapi.Plugin) ([]byte, error) { + return marshalRPCResult(rpcRegistration{ + SchemaVersion: pluginabi.SchemaVersion, + Metadata: plugin.Metadata, + Capabilities: rpcCapabilitiesFromPlugin(plugin), + }) +} + +func versionedPluginHostConfig(t *testing.T, pluginsDir string, version string) *config.Config { + t.Helper() + return &config.Config{Plugins: config.PluginsConfig{ + Enabled: true, + Dir: pluginsDir, + Configs: map[string]config.PluginInstanceConfig{ + "alpha": enabledPluginConfigWithStoreVersion(t, version), + }, + }} +} + type capturePluginClient struct { requests map[string][]byte } diff --git a/internal/pluginhost/rpc_client.go b/internal/pluginhost/rpc_client.go index 881f232c..1425d475 100644 --- a/internal/pluginhost/rpc_client.go +++ b/internal/pluginhost/rpc_client.go @@ -35,16 +35,17 @@ type rpcThinkingApplier struct { *rpcPluginAdapter } -type rpcPluginError struct { +type rpcError struct { + Code string message string statusCode int } -func (e rpcPluginError) Error() string { +func (e rpcError) Error() string { return e.message } -func (e rpcPluginError) StatusCode() int { +func (e rpcError) StatusCode() int { return e.statusCode } @@ -307,10 +308,11 @@ func decodeEnvelopeResult[T any](envelope pluginabi.Envelope) (T, error) { if message == "" { message = "plugin call failed" } - if envelope.Error.HTTPStatus > 0 { - return zero, rpcPluginError{message: message, statusCode: envelope.Error.HTTPStatus} + return zero, rpcError{ + Code: strings.TrimSpace(envelope.Error.Code), + message: message, + statusCode: envelope.Error.HTTPStatus, } - return zero, fmt.Errorf("%s", message) } return zero, fmt.Errorf("plugin call failed") } diff --git a/sdk/pluginabi/types.go b/sdk/pluginabi/types.go index 97c41a13..5089b132 100644 --- a/sdk/pluginabi/types.go +++ b/sdk/pluginabi/types.go @@ -18,6 +18,7 @@ const ( const ( MethodPluginRegister = "plugin.register" + MethodPluginQuiesce = "plugin.quiesce" MethodPluginReconfigure = "plugin.reconfigure" MethodPluginShutdown = "plugin.shutdown" diff --git a/sdk/pluginabi/types_test.go b/sdk/pluginabi/types_test.go index 8fa63542..45ac19f4 100644 --- a/sdk/pluginabi/types_test.go +++ b/sdk/pluginabi/types_test.go @@ -36,6 +36,9 @@ func TestMethodNamesAreStable(t *testing.T) { if MethodPluginRegister != "plugin.register" { t.Fatalf("MethodPluginRegister = %q", MethodPluginRegister) } + if MethodPluginQuiesce != "plugin.quiesce" { + t.Fatalf("MethodPluginQuiesce = %q", MethodPluginQuiesce) + } if MethodRequestInterceptBefore != "request.intercept_before" { t.Fatalf("MethodRequestInterceptBefore = %q", MethodRequestInterceptBefore) }