MM-11431: handle plugin deadlocks (#9167)

* ensure plugin is always shutdown

Once we call `.client.Client()` the plugin has started, and must be shut
down. `newSupervisor` sometimes returned with an error (and without a
reference to the supervisor), leaving the client running indefinitely.

* Clarify the documentation to explain that plugin hooks will not trigger until `OnActivate` returns successfully, and will stop triggering just before `OnDeactivate` is called.

* test for plugin deadlock

* plugin/environment.go: switch to sync.Map

From: https://golang.org/pkg/sync/#Map

> If a goroutine holds a RWMutex for reading and another goroutine might call Lock, no goroutine should expect to be able to acquire a read lock until the initial read lock is released. In particular, this prohibits recursive read locking. This is to ensure that the lock eventually becomes available; a blocked Lock call excludes new readers from acquiring the lock.

The previous `RWMutex` was not safe given that we effectively acquired read locks recursively (hook -> api -> hook). This worked up until we activated or deactivated plugins, tried to acquire a write lock, and the plugin used the API to effectively trigger another hook.

Switching to sync.Map avoids this by divesting the need to lock at all, avoiding the potential for a recursive lock in the first place.

app/plugin_deadlock_test.go

@@ -0,0 +1,211 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package app
+
+import (
+	"os"
+	"strings"
+	"testing"
+	"text/template"
+	"time"
+
+	"github.com/mattermost/mattermost-server/model"
+)
+
+func TestPluginDeadlock(t *testing.T) {
+	t.Run("Single Plugin", func(t *testing.T) {
+		th := Setup().InitBasic()
+		defer th.TearDown()
+
+		pluginPostOnActivate := template.Must(template.New("pluginPostOnActivate").Parse(`
+			package main
+
+			import (
+				"github.com/mattermost/mattermost-server/plugin"
+				"github.com/mattermost/mattermost-server/model"
+			)
+
+			type MyPlugin struct {
+				plugin.MattermostPlugin
+			}
+
+			func (p *MyPlugin) OnActivate() error {
+				_, err := p.API.CreatePost(&model.Post{
+					UserId: "{{.User.Id}}",
+					ChannelId: "{{.Channel.Id}}",
+					Message:   "message",
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+
+				return nil
+			}
+
+			func (p *MyPlugin) MessageWillBePosted(c *plugin.Context, post *model.Post) (*model.Post, string) {
+				if _, from_plugin := post.Props["from_plugin"]; from_plugin {
+					return post, ""
+				}
+
+				p.API.CreatePost(&model.Post{
+					UserId: "{{.User.Id}}",
+					ChannelId: "{{.Channel.Id}}",
+					Message:   "message",
+					Props: map[string]interface{}{
+						"from_plugin": true,
+					},
+				})
+
+				return post, ""
+			}
+
+			func main() {
+				plugin.ClientMain(&MyPlugin{})
+			}
+`,
+		))
+
+		templateData := struct {
+			User    *model.User
+			Channel *model.Channel
+		}{
+			th.BasicUser,
+			th.BasicChannel,
+		}
+
+		plugins := []string{}
+		pluginTemplates := []*template.Template{
+			pluginPostOnActivate,
+		}
+		for _, pluginTemplate := range pluginTemplates {
+			b := &strings.Builder{}
+			pluginTemplate.Execute(b, templateData)
+
+			plugins = append(plugins, b.String())
+		}
+
+		done := make(chan bool)
+		go func() {
+			SetAppEnvironmentWithPlugins(t, plugins, th.App, th.App.NewPluginAPI)
+			close(done)
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(30 * time.Second):
+			t.Fatal("plugin failed to activate: likely deadlocked")
+			go func() {
+				time.Sleep(5 * time.Second)
+				os.Exit(1)
+			}()
+		}
+	})
+
+	t.Run("Multiple Plugins", func(t *testing.T) {
+		th := Setup().InitBasic()
+		defer th.TearDown()
+
+		pluginPostOnHasBeenPosted := template.Must(template.New("pluginPostOnHasBeenPosted").Parse(`
+			package main
+
+			import (
+				"github.com/mattermost/mattermost-server/plugin"
+				"github.com/mattermost/mattermost-server/model"
+			)
+
+			type MyPlugin struct {
+				plugin.MattermostPlugin
+			}
+
+			func (p *MyPlugin) MessageWillBePosted(c *plugin.Context, post *model.Post) (*model.Post, string) {
+				if _, from_plugin := post.Props["from_plugin"]; from_plugin {
+					return post, ""
+				}
+
+				p.API.CreatePost(&model.Post{
+					UserId: "{{.User.Id}}",
+					ChannelId: "{{.Channel.Id}}",
+					Message:   "message",
+					Props: map[string]interface{}{
+						"from_plugin": true,
+					},
+				})
+
+				return post, ""
+			}
+
+			func main() {
+				plugin.ClientMain(&MyPlugin{})
+			}
+`,
+		))
+
+		pluginPostOnActivate := template.Must(template.New("pluginPostOnActivate").Parse(`
+			package main
+
+			import (
+				"github.com/mattermost/mattermost-server/plugin"
+				"github.com/mattermost/mattermost-server/model"
+			)
+
+			type MyPlugin struct {
+				plugin.MattermostPlugin
+			}
+
+			func (p *MyPlugin) OnActivate() error {
+				_, err := p.API.CreatePost(&model.Post{
+					UserId: "{{.User.Id}}",
+					ChannelId: "{{.Channel.Id}}",
+					Message:   "message",
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+
+				return nil
+			}
+
+			func main() {
+				plugin.ClientMain(&MyPlugin{})
+			}
+`,
+		))
+
+		templateData := struct {
+			User    *model.User
+			Channel *model.Channel
+		}{
+			th.BasicUser,
+			th.BasicChannel,
+		}
+
+		plugins := []string{}
+		pluginTemplates := []*template.Template{
+			pluginPostOnHasBeenPosted,
+			pluginPostOnActivate,
+		}
+		for _, pluginTemplate := range pluginTemplates {
+			b := &strings.Builder{}
+			pluginTemplate.Execute(b, templateData)
+
+			plugins = append(plugins, b.String())
+		}
+
+		done := make(chan bool)
+		go func() {
+			SetAppEnvironmentWithPlugins(t, plugins, th.App, th.App.NewPluginAPI)
+			close(done)
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(30 * time.Second):
+			t.Fatal("plugin failed to activate: likely deadlocked")
+			go func() {
+				time.Sleep(5 * time.Second)
+				os.Exit(1)
+			}()
+		}
+	})
+}

app/plugin_hooks_test.go

@@ -43,6 +43,7 @@ func SetAppEnvironmentWithPlugins(t *testing.T, pluginCode []string, app *App, a
 	env, err := plugin.NewEnvironment(apiFunc, pluginDir, webappPluginDir, app.Log)
 	require.NoError(t, err)
 
+	app.Plugins = env
 	for _, code := range pluginCode {
 		pluginId := model.NewId()
 		backend := filepath.Join(pluginDir, pluginId, "backend.exe")
@@ -51,8 +52,6 @@ func SetAppEnvironmentWithPlugins(t *testing.T, pluginCode []string, app *App, a
 		ioutil.WriteFile(filepath.Join(pluginDir, pluginId, "plugin.json"), []byte(`{"id": "`+pluginId+`", "backend": {"executable": "backend.exe"}}`), 0600)
 		env.Activate(pluginId)
 	}
-
-	app.Plugins = env
 }
 
 func TestHookMessageWillBePosted(t *testing.T) {

plugin/environment.go

@@ -34,8 +34,7 @@ type activePlugin struct {
 // It is meant for use by the Mattermost server to manipulate, interact with and report on the set
 // of active plugins.
 type Environment struct {
-	activePlugins   map[string]activePlugin
-	mutex           sync.RWMutex
+	activePlugins   sync.Map
 	logger          *mlog.Logger
 	newAPIImpl      apiImplCreatorFunc
 	pluginDir       string
@@ -44,7 +43,6 @@ type Environment struct {
 
 func NewEnvironment(newAPIImpl apiImplCreatorFunc, pluginDir string, webappPluginDir string, logger *mlog.Logger) (*Environment, error) {
 	return &Environment{
-		activePlugins:   make(map[string]activePlugin),
 		logger:          logger,
 		newAPIImpl:      newAPIImpl,
 		pluginDir:       pluginDir,
@@ -83,28 +81,24 @@ func (env *Environment) Available() ([]*model.BundleInfo, error) {
 
 // Returns a list of all currently active plugins within the environment.
 func (env *Environment) Active() []*model.BundleInfo {
-	env.mutex.RLock()
-	defer env.mutex.RUnlock()
-
 	activePlugins := []*model.BundleInfo{}
-	for _, p := range env.activePlugins {
-		activePlugins = append(activePlugins, p.BundleInfo)
-	}
+	env.activePlugins.Range(func(key, value interface{}) bool {
+		activePlugins = append(activePlugins, value.(activePlugin).BundleInfo)
+
+		return true
+	})
 
 	return activePlugins
 }
 
 // IsActive returns true if the plugin with the given id is active.
 func (env *Environment) IsActive(id string) bool {
-	_, ok := env.activePlugins[id]
+	_, ok := env.activePlugins.Load(id)
 	return ok
 }
 
 // Statuses returns a list of plugin statuses representing the state of every plugin
 func (env *Environment) Statuses() (model.PluginStatuses, error) {
-	env.mutex.RLock()
-	defer env.mutex.RUnlock()
-
 	plugins, err := env.Available()
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to get plugin statuses")
@@ -118,8 +112,8 @@ func (env *Environment) Statuses() (model.PluginStatuses, error) {
 		}
 
 		pluginState := model.PluginStateNotRunning
-		if plugin, ok := env.activePlugins[plugin.Manifest.Id]; ok {
-			pluginState = plugin.State
+		if plugin, ok := env.activePlugins.Load(plugin.Manifest.Id); ok {
+			pluginState = plugin.(activePlugin).State
 		}
 
 		status := &model.PluginStatus{
@@ -139,11 +133,9 @@ func (env *Environment) Statuses() (model.PluginStatuses, error) {
 
 // Activate activates the plugin with the given id.
 func (env *Environment) Activate(id string) (reterr error) {
-	env.mutex.Lock()
-	defer env.mutex.Unlock()
 
 	// Check if we are already active
-	if _, ok := env.activePlugins[id]; ok {
+	if _, ok := env.activePlugins.Load(id); ok {
 		return nil
 	}
 
@@ -171,7 +163,7 @@ func (env *Environment) Activate(id string) (reterr error) {
 		} else {
 			activePlugin.State = model.PluginStateFailedToStart
 		}
-		env.activePlugins[pluginInfo.Manifest.Id] = activePlugin
+		env.activePlugins.Store(pluginInfo.Manifest.Id, activePlugin)
 	}()
 
 	if pluginInfo.Manifest.Webapp != nil {
@@ -205,48 +197,49 @@ func (env *Environment) Activate(id string) (reterr error) {
 
 // Deactivates the plugin with the given id.
 func (env *Environment) Deactivate(id string) {
-	env.mutex.Lock()
-	defer env.mutex.Unlock()
-
-	if activePlugin, ok := env.activePlugins[id]; !ok {
+	p, ok := env.activePlugins.Load(id)
+	if !ok {
 		return
-	} else {
-		delete(env.activePlugins, id)
-		if activePlugin.supervisor != nil {
-			if err := activePlugin.supervisor.Hooks().OnDeactivate(); err != nil {
-				env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", activePlugin.BundleInfo.Manifest.Id), mlog.Err(err))
-			}
-			activePlugin.supervisor.Shutdown()
+	}
+
+	env.activePlugins.Delete(id)
+
+	activePlugin := p.(activePlugin)
+	if activePlugin.supervisor != nil {
+		if err := activePlugin.supervisor.Hooks().OnDeactivate(); err != nil {
+			env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", activePlugin.BundleInfo.Manifest.Id), mlog.Err(err))
 		}
+		activePlugin.supervisor.Shutdown()
 	}
 }
 
 // Shutdown deactivates all plugins and gracefully shuts down the environment.
 func (env *Environment) Shutdown() {
-	env.mutex.Lock()
-	defer env.mutex.Unlock()
+	env.activePlugins.Range(func(key, value interface{}) bool {
+		activePlugin := value.(activePlugin)
 
-	for _, activePlugin := range env.activePlugins {
 		if activePlugin.supervisor != nil {
 			if err := activePlugin.supervisor.Hooks().OnDeactivate(); err != nil {
 				env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", activePlugin.BundleInfo.Manifest.Id), mlog.Err(err))
 			}
 			activePlugin.supervisor.Shutdown()
 		}
-	}
-	env.activePlugins = make(map[string]activePlugin)
-	return
+
+		env.activePlugins.Delete(key)
+
+		return true
+	})
 }
 
 // HooksForPlugin returns the hooks API for the plugin with the given id.
 //
 // Consider using RunMultiPluginHook instead.
 func (env *Environment) HooksForPlugin(id string) (Hooks, error) {
-	env.mutex.RLock()
-	defer env.mutex.RUnlock()
-
-	if plug, ok := env.activePlugins[id]; ok && plug.supervisor != nil {
-		return plug.supervisor.Hooks(), nil
+	if p, ok := env.activePlugins.Load(id); ok {
+		activePlugin := p.(activePlugin)
+		if activePlugin.supervisor != nil {
+			return activePlugin.supervisor.Hooks(), nil
+		}
 	}
 
 	return nil, fmt.Errorf("plugin not found: %v", id)
@@ -257,15 +250,16 @@ func (env *Environment) HooksForPlugin(id string) (Hooks, error) {
 // If hookRunnerFunc returns false, iteration will not continue. The iteration order among active
 // plugins is not specified.
 func (env *Environment) RunMultiPluginHook(hookRunnerFunc multiPluginHookRunnerFunc, hookId int) {
-	env.mutex.RLock()
-	defer env.mutex.RUnlock()
+	env.activePlugins.Range(func(key, value interface{}) bool {
+		activePlugin := value.(activePlugin)
 
-	for _, activePlugin := range env.activePlugins {
 		if activePlugin.supervisor == nil || !activePlugin.supervisor.Implements(hookId) {
-			continue
+			return true
 		}
 		if !hookRunnerFunc(activePlugin.supervisor.Hooks()) {
-			break
+			return false
 		}
-	}
+
+		return true
+	})
 }

plugin/hooks.go

@@ -39,7 +39,9 @@ const (
 // A plugin only need implement the hooks it cares about. The MattermostPlugin provides some
 // default implementations for convenience but may be overridden.
 type Hooks interface {
-	// OnActivate is invoked when the plugin is activated.
+	// OnActivate is invoked when the plugin is activated. If an error is returned, the plugin
+	// will be terminated. The plugin will not receive hooks until after OnActivate returns
+	// without error.
 	OnActivate() error
 
 	// Implemented returns a list of hooks that are implemented by the plugin.
@@ -47,7 +49,8 @@ type Hooks interface {
 	Implemented() ([]string, error)
 
 	// OnDeactivate is invoked when the plugin is deactivated. This is the plugin's last chance to
-	// use the API, and the plugin will be terminated shortly after this invocation.
+	// use the API, and the plugin will be terminated shortly after this invocation. The plugin
+	// will stop receiving hooks just prior to this method being called.
 	OnDeactivate() error
 
 	// OnConfigurationChange is invoked when configuration changes may have been made.

plugin/supervisor.go

@@ -23,8 +23,13 @@ type supervisor struct {
 	implemented [TotalHooksId]bool
 }
 
-func newSupervisor(pluginInfo *model.BundleInfo, parentLogger *mlog.Logger, apiImpl API) (*supervisor, error) {
+func newSupervisor(pluginInfo *model.BundleInfo, parentLogger *mlog.Logger, apiImpl API) (retSupervisor *supervisor, retErr error) {
 	supervisor := supervisor{}
+	defer func() {
+		if retErr != nil {
+			supervisor.Shutdown()
+		}
+	}()
 
 	wrappedLogger := pluginInfo.WrapLogger(parentLogger)
 
@@ -90,7 +95,9 @@ func newSupervisor(pluginInfo *model.BundleInfo, parentLogger *mlog.Logger, apiI
 }
 
 func (sup *supervisor) Shutdown() {
-	sup.client.Kill()
+	if sup.client != nil {
+		sup.client.Kill()
+	}
 }
 
 func (sup *supervisor) Hooks() Hooks {