Fix plugin RPC teardown race on shutdown (#37340)

* Fix plugin RPC teardown race in Shutdown and Deactivate

Environment.Shutdown() closed each plugin's RPC connection (via
supervisor.Shutdown) before removing it from registeredPlugins, leaving
a window where IsActive still returned true for a plugin whose transport
was already dead. Any concurrent RunMultiPluginHook* call in that window
would dispatch to the dead connection and log a spurious
"connection is shut down" error.

The fix: set PluginStateNotRunning immediately before supervisor.Shutdown
on every teardown path, so IsActive returns false before any dispatcher
can reach the closed transport. The state remains Running during
OnDeactivate so a plugin can still call back into the server (e.g.
CreatePost) and dispatch hooks to itself while its connection is live.

* fixup! Fix plugin RPC teardown race in Shutdown and Deactivate

* fixup! Fix plugin RPC teardown race in Shutdown and Deactivate

* Skip OnDeactivate teardown for already-inactive plugins

deactivateAndTeardown ran OnDeactivate and closed the RPC connection
guarding only on a nil supervisor. Shutdown calls it for every
registered plugin, so a plugin already deactivated (state NotRunning,
but still registered with a shut-down supervisor) had OnDeactivate
dispatched over its dead transport, logging a spurious 'connection is
shut down' error.

Fold the IsActive check into deactivateAndTeardown so both Shutdown and
Deactivate skip OnDeactivate and supervisor.Shutdown when the plugin
isn't running, only reconciling state to NotRunning.

* document and test intentional transition to PluginStatusNotRunning

server/public/plugin/environment.go

@@ -446,30 +446,55 @@ func (env *Environment) RemovePlugin(id string) {
 	}
 }
 
-// Deactivates the plugin with the given id.
+// deactivateAndTeardown runs OnDeactivate (with a 10-second timeout), then marks the plugin
+// as not running and closes its RPC connection. The plugin remains reachable via
+// RunMultiPluginHook* throughout OnDeactivate so it can dispatch hooks back to itself.
+//
+// If the plugin is not currently active, there's nothing to tear down: its state is reconciled
+// to not running and no OnDeactivate is dispatched, avoiding a spurious RPC call to an
+// already-closed connection.
+//
+// Deactivation always reconciles the plugin to PluginStateNotRunning, intentionally clearing any
+// error state (e.g. PluginStateFailedToStayRunning): once deactivated, the plugin is no longer
+// meant to be running, so the error no longer applies. Callers that want to record a more specific
+// state deactivate first and set it afterward, as the health check job does when a plugin exceeds
+// its restart limit.
+func (env *Environment) deactivateAndTeardown(rp registeredPlugin) bool {
+	id := rp.BundleInfo.Manifest.Id
+
+	if rp.supervisor == nil || !env.IsActive(id) {
+		env.setPluginState(id, model.PluginStateNotRunning)
+		return false
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		if err := rp.supervisor.Hooks().OnDeactivate(); err != nil {
+			env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", id), mlog.Err(err))
+		}
+	}()
+
+	select {
+	case <-time.After(10 * time.Second):
+		env.logger.Warn("Plugin OnDeactivate() failed to complete in 10 seconds", mlog.String("plugin_id", id))
+	case <-done:
+	}
+
+	env.setPluginState(id, model.PluginStateNotRunning)
+	rp.supervisor.Shutdown()
+
+	return true
+}
+
+// Deactivate the plugin with the given id.
 func (env *Environment) Deactivate(id string) bool {
 	p, ok := env.registeredPlugins.Load(id)
 	if !ok {
 		return false
 	}
 
-	isActive := env.IsActive(id)
-
-	env.setPluginState(id, model.PluginStateNotRunning)
-
-	if !isActive {
-		return false
-	}
-
-	rp := p.(registeredPlugin)
-	if rp.supervisor != nil {
-		if err := rp.supervisor.Hooks().OnDeactivate(); err != nil {
-			env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", rp.BundleInfo.Manifest.Id), mlog.Err(err))
-		}
-		rp.supervisor.Shutdown()
-	}
-
-	return true
+	return env.deactivateAndTeardown(p.(registeredPlugin))
 }
 
 // RestartPlugin deactivates, then activates the plugin with the given id.
@@ -487,31 +512,7 @@ func (env *Environment) Shutdown() {
 	env.registeredPlugins.Range(func(_, value any) bool {
 		rp := value.(registeredPlugin)
 
-		if rp.supervisor == nil || !env.IsActive(rp.BundleInfo.Manifest.Id) {
-			return true
-		}
-
-		wg.Add(1)
-
-		done := make(chan bool)
-		go func() {
-			defer close(done)
-			if err := rp.supervisor.Hooks().OnDeactivate(); err != nil {
-				env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", rp.BundleInfo.Manifest.Id), mlog.Err(err))
-			}
-		}()
-
-		go func() {
-			defer wg.Done()
-
-			select {
-			case <-time.After(10 * time.Second):
-				env.logger.Warn("Plugin OnDeactivate() failed to complete in 10 seconds", mlog.String("plugin_id", rp.BundleInfo.Manifest.Id))
-			case <-done:
-			}
-
-			rp.supervisor.Shutdown()
-		}()
+		wg.Go(func() { env.deactivateAndTeardown(rp) })
 
 		return true
 	})

server/public/plugin/environment_shutdown_state_test.go

@@ -0,0 +1,278 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package plugin
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/plugin/utils"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+)
+
+// TestPluginMarksNotRunningAfterOnDeactivate verifies the state transitions during plugin
+// teardown, for both Shutdown and Deactivate. While a plugin's OnDeactivate is still running,
+// IsActive must return true so that hook dispatches the plugin makes to itself (e.g. CreatePost)
+// are not rejected. Once OnDeactivate completes, IsActive must return false before the RPC
+// connection is torn down.
+func TestPluginMarksNotRunningAfterOnDeactivate(t *testing.T) {
+	testCases := []struct {
+		name     string
+		teardown func(env *Environment, pluginID string)
+	}{
+		{
+			name: "Shutdown",
+			teardown: func(env *Environment, _ string) {
+				env.Shutdown()
+			},
+		},
+		{
+			name: "Deactivate",
+			teardown: func(env *Environment, pluginID string) {
+				env.Deactivate(pluginID)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			pluginDir, err := os.MkdirTemp("", "mm-shutdown-state-plugin")
+			require.NoError(t, err)
+			t.Cleanup(func() { os.RemoveAll(pluginDir) })
+			webappPluginDir, err := os.MkdirTemp("", "mm-shutdown-state-webapp")
+			require.NoError(t, err)
+			t.Cleanup(func() { os.RemoveAll(webappPluginDir) })
+
+			pluginID := "test-shutdown-state-plugin"
+			require.NoError(t, os.MkdirAll(filepath.Join(pluginDir, pluginID), 0700))
+			backend := filepath.Join(pluginDir, pluginID, "backend.exe")
+
+			// OnDeactivate blocks until the test signals it via MessageWillBePosted.
+			utils.CompileGo(t, `
+				package main
+
+				import (
+					"sync"
+
+					"github.com/mattermost/mattermost/server/public/model"
+					"github.com/mattermost/mattermost/server/public/plugin"
+				)
+
+				type MyPlugin struct {
+					plugin.MattermostPlugin
+					once    sync.Once
+					proceed chan struct{}
+				}
+
+				func (p *MyPlugin) OnActivate() error {
+					p.proceed = make(chan struct{})
+					return nil
+				}
+
+				func (p *MyPlugin) OnDeactivate() error {
+					<-p.proceed
+					return nil
+				}
+
+				func (p *MyPlugin) MessageWillBePosted(_ *plugin.Context, _ *model.Post) (*model.Post, string) {
+					p.once.Do(func() { close(p.proceed) })
+					return nil, ""
+				}
+
+				func main() {
+					plugin.ClientMain(&MyPlugin{})
+				}
+			`, backend)
+
+			require.NoError(t, os.WriteFile(
+				filepath.Join(pluginDir, pluginID, "plugin.json"),
+				[]byte(`{"id":"`+pluginID+`","server":{"executable":"backend.exe"}}`),
+				0600,
+			))
+
+			logger := mlog.CreateConsoleTestLogger(t)
+			apiImpl := func(*model.Manifest) API { return nil }
+			env, err := NewEnvironment(apiImpl, nil, pluginDir, webappPluginDir, logger, nil)
+			require.NoError(t, err)
+
+			_, _, err = env.Activate(pluginID)
+			require.NoError(t, err)
+			require.True(t, env.IsActive(pluginID))
+
+			teardownDone := make(chan struct{})
+			go func() {
+				defer close(teardownDone)
+				tc.teardown(env, pluginID)
+			}()
+
+			// Plugin is blocked in OnDeactivate — state must still be Running so a plugin
+			// dispatching hooks back to itself from OnDeactivate isn't rejected.
+			require.True(t, env.IsActive(pluginID), "IsActive should be true while OnDeactivate is blocking")
+
+			// Signal the plugin to complete OnDeactivate.
+			env.RunMultiPluginHook(func(hooks Hooks, _ *model.Manifest) bool {
+				hooks.MessageWillBePosted(&Context{}, &model.Post{})
+				return true
+			}, MessageWillBePostedID)
+
+			select {
+			case <-teardownDone:
+			case <-time.After(2 * time.Second):
+				t.Fatalf("%s did not return", tc.name)
+			}
+
+			require.False(t, env.IsActive(pluginID))
+		})
+	}
+}
+
+// TestShutdownAfterDeactivateNoOnDeactivateRPC verifies that shutting down the environment after a
+// plugin has already been deactivated does not dispatch a second OnDeactivate over the plugin's
+// already-closed RPC connection. Previously Shutdown ran deactivateAndTeardown unconditionally for
+// every registered plugin, dispatching OnDeactivate to inactive plugins and logging a spurious
+// "connection is shut down" error.
+func TestShutdownAfterDeactivateNoOnDeactivateRPC(t *testing.T) {
+	pluginDir, err := os.MkdirTemp("", "mm-shutdown-after-deactivate-plugin")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(pluginDir) })
+	webappPluginDir, err := os.MkdirTemp("", "mm-shutdown-after-deactivate-webapp")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(webappPluginDir) })
+
+	pluginID := "test-shutdown-after-deactivate-plugin"
+	require.NoError(t, os.MkdirAll(filepath.Join(pluginDir, pluginID), 0700))
+	backend := filepath.Join(pluginDir, pluginID, "backend.exe")
+
+	utils.CompileGo(t, `
+		package main
+
+		import (
+			"github.com/mattermost/mattermost/server/public/plugin"
+		)
+
+		type MyPlugin struct {
+			plugin.MattermostPlugin
+		}
+
+		func (p *MyPlugin) OnDeactivate() error {
+			return nil
+		}
+
+		func main() {
+			plugin.ClientMain(&MyPlugin{})
+		}
+	`, backend)
+
+	require.NoError(t, os.WriteFile(
+		filepath.Join(pluginDir, pluginID, "plugin.json"),
+		[]byte(`{"id":"`+pluginID+`","server":{"executable":"backend.exe"}}`),
+		0600,
+	))
+
+	logger := mlog.CreateConsoleTestLogger(t)
+	var buf mlog.Buffer
+	require.NoError(t, mlog.AddWriterTarget(logger, &buf, true, mlog.LvlError))
+
+	apiImpl := func(*model.Manifest) API { return nil }
+	env, err := NewEnvironment(apiImpl, nil, pluginDir, webappPluginDir, logger, nil)
+	require.NoError(t, err)
+
+	_, _, err = env.Activate(pluginID)
+	require.NoError(t, err)
+	require.True(t, env.IsActive(pluginID))
+
+	// Deactivate tears down the RPC connection but leaves the plugin registered.
+	require.True(t, env.Deactivate(pluginID))
+	require.False(t, env.IsActive(pluginID))
+
+	// Shutdown must not dispatch OnDeactivate again to the now-inactive plugin.
+	env.Shutdown()
+
+	require.NoError(t, logger.Flush())
+	assert.NotContains(t, buf.String(), "OnDeactivate",
+		"Shutdown dispatched OnDeactivate to an already-deactivated plugin")
+}
+
+// TestDeactivateReconcilesPluginState verifies that deactivation always reconciles the plugin to
+// PluginStateNotRunning, clearing any prior error state, and that the health check job's ordering
+// (deactivate, then set state) still records PluginStateFailedToStayRunning.
+func TestDeactivateReconcilesPluginState(t *testing.T) {
+	pluginDir, err := os.MkdirTemp("", "mm-deactivate-state-plugin")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(pluginDir) })
+	webappPluginDir, err := os.MkdirTemp("", "mm-deactivate-state-webapp")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(webappPluginDir) })
+
+	pluginID := "test-deactivate-state-plugin"
+	require.NoError(t, os.MkdirAll(filepath.Join(pluginDir, pluginID), 0700))
+	backend := filepath.Join(pluginDir, pluginID, "backend.exe")
+
+	utils.CompileGo(t, `
+		package main
+
+		import (
+			"github.com/mattermost/mattermost/server/public/plugin"
+		)
+
+		type MyPlugin struct {
+			plugin.MattermostPlugin
+		}
+
+		func (p *MyPlugin) OnDeactivate() error {
+			return nil
+		}
+
+		func main() {
+			plugin.ClientMain(&MyPlugin{})
+		}
+	`, backend)
+
+	require.NoError(t, os.WriteFile(
+		filepath.Join(pluginDir, pluginID, "plugin.json"),
+		[]byte(`{"id":"`+pluginID+`","server":{"executable":"backend.exe"}}`),
+		0600,
+	))
+
+	logger := mlog.CreateConsoleTestLogger(t)
+	apiImpl := func(*model.Manifest) API { return nil }
+	env, err := NewEnvironment(apiImpl, nil, pluginDir, webappPluginDir, logger, nil)
+	require.NoError(t, err)
+	t.Cleanup(env.Shutdown)
+
+	t.Run("deactivating an inactive, failed plugin clears the error state", func(t *testing.T) {
+		_, _, err = env.Activate(pluginID)
+		require.NoError(t, err)
+		require.True(t, env.IsActive(pluginID))
+
+		// Tear the plugin down, then simulate the health check marking it as failed while it
+		// remains registered but inactive.
+		require.True(t, env.Deactivate(pluginID))
+		require.False(t, env.IsActive(pluginID))
+		env.setPluginState(pluginID, model.PluginStateFailedToStayRunning)
+
+		// Deactivating an already-inactive plugin (e.g. on Shutdown or an explicit disable) must
+		// reconcile the error state back to not running rather than leaving it as failed.
+		require.False(t, env.Deactivate(pluginID))
+		assert.Equal(t, model.PluginStateNotRunning, env.GetPluginState(pluginID))
+	})
+
+	t.Run("health check ordering records FailedToStayRunning", func(t *testing.T) {
+		_, _, err = env.Activate(pluginID)
+		require.NoError(t, err)
+		require.True(t, env.IsActive(pluginID))
+
+		// Mirror PluginHealthCheckJob.CheckPlugin: deactivate first, then record the failed state.
+		require.True(t, env.Deactivate(pluginID))
+		env.setPluginState(pluginID, model.PluginStateFailedToStayRunning)
+
+		assert.Equal(t, model.PluginStateFailedToStayRunning, env.GetPluginState(pluginID))
+	})
+}

server/public/plugin/environment_with_rpcerr_test.go

@@ -6,11 +6,13 @@ package plugin
 import (
 	"errors"
 	"io"
+	"net/rpc"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/mattermost/mattermost/server/public/model"
@@ -131,6 +133,150 @@ func TestRunMultiPluginHookWithRPCErr(t *testing.T) {
 	})
 }
 
+// TestShutdownNoRPCErrorsDuringConcurrentHookDispatch verifies that concurrent
+// RunMultiPluginHookWithRPCErr calls during Shutdown do not observe "connection
+// is shut down" errors.
+//
+// The race: Shutdown closes each plugin's RPC connection (supervisor.Shutdown)
+// before removing it from registeredPlugins, leaving a window where the plugin
+// still appears active but its transport is dead.  The fix sets the plugin state
+// to NotRunning before closing the RPC, so concurrent callers skip it cleanly.
+func TestShutdownNoRPCErrorsDuringConcurrentHookDispatch(t *testing.T) {
+	pluginDir, err := os.MkdirTemp("", "mm-shutdown-race-plugin")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(pluginDir) })
+	webappPluginDir, err := os.MkdirTemp("", "mm-shutdown-race-webapp")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(webappPluginDir) })
+
+	fastID := "test-shutdown-race-fast"
+	slowID := "test-shutdown-race-slow"
+	require.NoError(t, os.MkdirAll(filepath.Join(pluginDir, fastID), 0700))
+	require.NoError(t, os.MkdirAll(filepath.Join(pluginDir, slowID), 0700))
+
+	fastBackend := filepath.Join(pluginDir, fastID, "backend.exe")
+	slowBackend := filepath.Join(pluginDir, slowID, "backend.exe")
+
+	// fast plugin: instant OnDeactivate
+	utils.CompileGo(t, `
+		package main
+
+		import (
+			"github.com/mattermost/mattermost/server/public/model"
+			"github.com/mattermost/mattermost/server/public/plugin"
+		)
+
+		type FastPlugin struct{ plugin.MattermostPlugin }
+
+		func (p *FastPlugin) MessageHasBeenPosted(_ *plugin.Context, _ *model.Post) {}
+
+		func main() { plugin.ClientMain(&FastPlugin{}) }
+	`, fastBackend)
+
+	// slow plugin: OnDeactivate blocks until the test signals it via MessageWillBePosted,
+	// keeping Shutdown's wg.Wait alive so the fast plugin's entry lingers in registeredPlugins.
+	utils.CompileGo(t, `
+		package main
+
+		import (
+			"sync"
+
+			"github.com/mattermost/mattermost/server/public/model"
+			"github.com/mattermost/mattermost/server/public/plugin"
+		)
+
+		type SlowPlugin struct {
+			plugin.MattermostPlugin
+			once    sync.Once
+			proceed chan struct{}
+		}
+
+		func (p *SlowPlugin) OnActivate() error {
+			p.proceed = make(chan struct{})
+			return nil
+		}
+
+		func (p *SlowPlugin) OnDeactivate() error {
+			<-p.proceed
+			return nil
+		}
+
+		func (p *SlowPlugin) MessageWillBePosted(_ *plugin.Context, _ *model.Post) (*model.Post, string) {
+			p.once.Do(func() { close(p.proceed) })
+			return nil, ""
+		}
+
+		func (p *SlowPlugin) MessageHasBeenPosted(_ *plugin.Context, _ *model.Post) {}
+
+		func main() { plugin.ClientMain(&SlowPlugin{}) }
+	`, slowBackend)
+
+	require.NoError(t, os.WriteFile(
+		filepath.Join(pluginDir, fastID, "plugin.json"),
+		[]byte(`{"id":"`+fastID+`","server":{"executable":"backend.exe"}}`),
+		0600,
+	))
+	require.NoError(t, os.WriteFile(
+		filepath.Join(pluginDir, slowID, "plugin.json"),
+		[]byte(`{"id":"`+slowID+`","server":{"executable":"backend.exe"}}`),
+		0600,
+	))
+
+	logger := mlog.CreateConsoleTestLogger(t)
+	apiImpl := func(*model.Manifest) API { return nil }
+	env, err := NewEnvironment(apiImpl, nil, pluginDir, webappPluginDir, logger, nil)
+	require.NoError(t, err)
+
+	_, _, err = env.Activate(fastID)
+	require.NoError(t, err)
+	_, _, err = env.Activate(slowID)
+	require.NoError(t, err)
+
+	// Race window: Shutdown is running concurrently.  After the fast plugin's RPC
+	// is closed (it finishes OnDeactivate first) but before registeredPlugins is
+	// cleaned up (gated on the slow plugin finishing), concurrent hook dispatches
+	// must not observe net/rpc.ErrShutdown.
+	//
+	// The slow plugin blocks in OnDeactivate until we call MessageWillBePosted,
+	// giving us a controlled window to dispatch hooks while the fast plugin's RPC
+	// is already closed.  Each dispatch involves IPC (a Unix socket round-trip),
+	// which yields to the scheduler and lets the fast plugin's teardown goroutine run.
+	shutdownDone := make(chan struct{})
+	go func() {
+		defer close(shutdownDone)
+		env.Shutdown()
+	}()
+
+	var rpcErrs []error
+	for range 200 {
+		_ = env.RunMultiPluginHookWithRPCErr(func(hooks HooksWithRPCErr, _ *model.Manifest) (bool, error) {
+			if rpcErr := hooks.MessageHasBeenPostedWithRPCErr(&Context{}, &model.Post{}); rpcErr != nil {
+				rpcErrs = append(rpcErrs, rpcErr)
+			}
+			return true, nil
+		}, MessageHasBeenPostedID)
+	}
+
+	// Signal the slow plugin to finish OnDeactivate, unblocking Shutdown.
+	env.RunMultiPluginHook(func(hooks Hooks, _ *model.Manifest) bool {
+		hooks.MessageWillBePosted(&Context{}, &model.Post{})
+		return true
+	}, MessageWillBePostedID)
+
+	<-shutdownDone
+
+	// Filter to only the canonical shutdown error so the test isn't brittle
+	// against other transient RPC errors (e.g. EOF on race-y reads).
+	var shutdownErrs []error
+	for _, e := range rpcErrs {
+		if errors.Is(e, rpc.ErrShutdown) {
+			shutdownErrs = append(shutdownErrs, e)
+		}
+	}
+	assert.Empty(t, shutdownErrs,
+		"RunMultiPluginHookWithRPCErr dispatched to a plugin whose RPC connection was already closed during Shutdown")
+}
+
 func copyExecutable(t *testing.T, src, dst string) {
 	t.Helper()
 	in, err := os.Open(src)