MM-10856: deduplicate posts with the same pending post id (#10006)

* MM-10856: deduplicate posts with the same pending post id Leverage a fixed size cache with a window of 30 seconds to deduplicate posts received by a single app server. Clients that duplicate the same pending post id will see one request potentially delayed until the first finishes, after which the same payload should be returned by both. Duplicate posts outside the 30 second window will not be de-duplicated. Note that the cache is not synchronized between app servers. In an HA cluster consisting of more than one app server, sticky load balancing (e.g. hashing by IP or Session ID) is required to route the users to the same app instance for multiple requests. Other options considered for this feature: * adding a column to the `Posts` table: rejected as being too heavyweight * maintaining a `PendingPostIds` table: similarly rejected for the database impact * using the pending post id as the post id and relying on the unique constraints on the Post table: rejected for being difficult to show that it's safe to use a client-provided value as the row identifier * utils/lru: simplify to ttl internally and for new methods * move seenPendingPostIdsCache to App.Server * just fail concurrent post requests, vs. trying to wait * add debug log when create post is deduplicated
2018-12-17 18:16:57 -05:00
--- a/app/plugin_hooks_test.go
+++ b/app/plugin_hooks_test.go
@@ -32,7 +32,7 @@ func compileGo(t *testing.T, sourceCode, outputPath string) {
 	cmd.Dir = dir
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
-	require.NoError(t, cmd.Run())
+	require.NoError(t, cmd.Run(), "failed to compile go")
 }

 func SetAppEnvironmentWithPlugins(t *testing.T, pluginCode []string, app *App, apiFunc func(*model.Manifest) plugin.API) (func(), []string, []error) {
--- a/app/post.go
+++ b/app/post.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"time"

 	"github.com/mattermost/mattermost-server/mlog"
 	"github.com/mattermost/mattermost-server/model"
@@ -19,6 +20,11 @@ import (
 	"github.com/mattermost/mattermost-server/utils"
 )

+const (
+	PENDING_POST_IDS_CACHE_SIZE = 25000
+	PENDING_POST_IDS_CACHE_TTL  = 30 * time.Second
+)
+
 func (a *App) CreatePostAsUser(post *model.Post, clearPushNotifications bool) (*model.Post, *model.AppError) {
 	// Check that channel has not been deleted
 	result := <-a.Srv.Store.Channel().Get(post.ChannelId, true)
@@ -89,7 +95,69 @@ func (a *App) CreatePostMissingChannel(post *model.Post, triggerWebhooks bool) (
 	return a.CreatePost(post, channel, triggerWebhooks)
 }

-func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhooks bool) (*model.Post, *model.AppError) {
+// deduplicateCreatePost attempts to make posting idempotent within a caching window.
+func (a *App) deduplicateCreatePost(post *model.Post) (foundPost *model.Post, err *model.AppError) {
+	// We rely on the client sending the pending post id across "duplicate" requests. If there
+	// isn't one, we can't deduplicate, so allow creation normally.
+	if post.PendingPostId == "" {
+		return nil, nil
+	}
+
+	const unknownPostId = ""
+
+	// Query the cache atomically for the given pending post id, saving a record if
+	// it hasn't previously been seen.
+	value, loaded := a.Srv.seenPendingPostIdsCache.GetOrAdd(post.PendingPostId, unknownPostId, PENDING_POST_IDS_CACHE_TTL)
+
+	// If we were the first thread to save this pending post id into the cache,
+	// proceed with create post normally.
+	if !loaded {
+		return nil, nil
+	}
+
+	postId := value.(string)
+
+	// If another thread saved the cache record, but hasn't yet updated it with the actual post
+	// id (because it's still saving), notify the client with an error. Ideally, we'd wait
+	// for the other thread, but coordinating that adds complexity to the happy path.
+	if postId == unknownPostId {
+		return nil, model.NewAppError("deduplicateCreatePost", "api.post.deduplicate_create_post.pending", nil, "", http.StatusInternalServerError)
+	}
+
+	// If the other thread finished creating the post, return the created post back to the
+	// client, making the API call feel idempotent.
+	actualPost, err := a.GetSinglePost(postId)
+	if err != nil {
+		return nil, model.NewAppError("deduplicateCreatePost", "api.post.deduplicate_create_post.failed_to_get", nil, err.Error(), http.StatusInternalServerError)
+	}
+
+	mlog.Debug("Deduplicated create post", mlog.String("post_id", actualPost.Id), mlog.String("pending_post_id", post.PendingPostId))
+
+	return actualPost, nil
+}
+
+func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhooks bool) (savedPost *model.Post, err *model.AppError) {
+	if foundPost, err := a.deduplicateCreatePost(post); err != nil {
+		return nil, err
+	} else if foundPost != nil {
+		return foundPost, nil
+	}
+
+	// If we get this far, we've recorded the client-provided pending post id to the cache.
+	// Remove it if we fail below, allowing a proper retry by the client.
+	defer func() {
+		if post.PendingPostId == "" {
+			return
+		}
+
+		if err != nil {
+			a.Srv.seenPendingPostIdsCache.Remove(post.PendingPostId)
+			return
+		}
+
+		a.Srv.seenPendingPostIdsCache.AddWithExpiresInSecs(post.PendingPostId, savedPost.Id, int64(PENDING_POST_IDS_CACHE_TTL.Seconds()))
+	}()
+
 	post.SanitizeProps()

 	var pchan store.StoreChannel
@@ -179,6 +247,10 @@ func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhoo
 	}
 	rpost := result.Data.(*model.Post)

+	// Update the mapping from pending post id to the actual post id, for any clients that
+	// might be duplicating requests.
+	a.Srv.seenPendingPostIdsCache.AddWithExpiresInSecs(post.PendingPostId, rpost.Id, int64(PENDING_POST_IDS_CACHE_TTL.Seconds()))
+
 	if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
 		a.Srv.Go(func() {
 			pluginContext := a.PluginContext()
@@ -220,7 +292,7 @@ func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhoo
 	rpost = a.PreparePostForClient(rpost)

 	if err := a.handlePostEvents(rpost, user, channel, triggerWebhooks, parentPostList); err != nil {
-		return nil, err
+		mlog.Error("Failed to handle post events", mlog.Err(err))
 	}

 	return rpost, nil
--- a/app/post_test.go
+++ b/app/post_test.go
@@ -6,6 +6,7 @@ package app
 import (
 	"fmt"
 	"net/http"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -18,6 +19,174 @@ import (
 	"github.com/mattermost/mattermost-server/store/storetest"
 )

+func TestCreatePostDeduplicate(t *testing.T) {
+	th := Setup().InitBasic()
+	defer th.TearDown()
+
+	t.Run("duplicate create post is idempotent", func(t *testing.T) {
+		pendingPostId := model.NewId()
+		post, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.Nil(t, err)
+		require.Equal(t, "message", post.Message)
+
+		duplicatePost, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.Nil(t, err)
+		require.Equal(t, post.Id, duplicatePost.Id, "should have returned previously created post id")
+		require.Equal(t, "message", duplicatePost.Message)
+	})
+
+	t.Run("post rejected by plugin leaves cache ready for non-deduplicated try", func(t *testing.T) {
+		setupPluginApiTest(t, `
+			package main
+
+			import (
+				"github.com/mattermost/mattermost-server/plugin"
+				"github.com/mattermost/mattermost-server/model"
+			)
+
+			type MyPlugin struct {
+				plugin.MattermostPlugin
+				allow bool
+			}
+
+			func (p *MyPlugin) MessageWillBePosted(c *plugin.Context, post *model.Post) (*model.Post, string) {
+				if !p.allow {
+					p.allow = true
+					return nil, "rejected"
+				}
+
+				return nil, ""
+			}
+
+			func main() {
+				plugin.ClientMain(&MyPlugin{})
+			}
+		`, `{"id": "testrejectfirstpost", "backend": {"executable": "backend.exe"}}`, "testrejectfirstpost", th.App)
+
+		pendingPostId := model.NewId()
+		post, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.NotNil(t, err)
+		require.Equal(t, "Post rejected by plugin. rejected", err.Id)
+		require.Nil(t, post)
+
+		duplicatePost, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.Nil(t, err)
+		require.Equal(t, "message", duplicatePost.Message)
+	})
+
+	t.Run("slow posting after cache entry blocks duplicate request", func(t *testing.T) {
+		setupPluginApiTest(t, `
+			package main
+
+			import (
+				"github.com/mattermost/mattermost-server/plugin"
+				"github.com/mattermost/mattermost-server/model"
+				"time"
+			)
+
+			type MyPlugin struct {
+				plugin.MattermostPlugin
+				instant bool
+			}
+
+			func (p *MyPlugin) MessageWillBePosted(c *plugin.Context, post *model.Post) (*model.Post, string) {
+				if !p.instant {
+					p.instant = true
+					time.Sleep(3 * time.Second)
+				}
+
+				return nil, ""
+			}
+
+			func main() {
+				plugin.ClientMain(&MyPlugin{})
+			}
+		`, `{"id": "testdelayfirstpost", "backend": {"executable": "backend.exe"}}`, "testdelayfirstpost", th.App)
+
+		var post *model.Post
+		pendingPostId := model.NewId()
+
+		wg := sync.WaitGroup{}
+
+		// Launch a goroutine to make the first CreatePost call that will get delayed
+		// by the plugin above.
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var err error
+			post, err = th.App.CreatePostAsUser(&model.Post{
+				UserId:        th.BasicUser.Id,
+				ChannelId:     th.BasicChannel.Id,
+				Message:       "plugin delayed",
+				PendingPostId: pendingPostId,
+			}, false)
+			require.Nil(t, err)
+			require.Equal(t, post.Message, "plugin delayed")
+		}()
+
+		// Give the goroutine above a chance to start and get delayed by the plugin.
+		time.Sleep(2 * time.Second)
+
+		// Try creating a duplicate post
+		duplicatePost, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "plugin delayed",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.NotNil(t, err)
+		require.Equal(t, "api.post.deduplicate_create_post.pending", err.Id)
+		require.Nil(t, duplicatePost)
+
+		// Wait for the first CreatePost to finish to ensure assertions are made.
+		wg.Wait()
+	})
+
+	t.Run("duplicate create post after cache expires is not idempotent", func(t *testing.T) {
+		pendingPostId := model.NewId()
+		post, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.Nil(t, err)
+		require.Equal(t, "message", post.Message)
+
+		time.Sleep(PENDING_POST_IDS_CACHE_TTL)
+
+		duplicatePost, err := th.App.CreatePostAsUser(&model.Post{
+			UserId:        th.BasicUser.Id,
+			ChannelId:     th.BasicChannel.Id,
+			Message:       "message",
+			PendingPostId: pendingPostId,
+		}, false)
+		require.Nil(t, err)
+		require.NotEqual(t, post.Id, duplicatePost.Id, "should have created new post id")
+		require.Equal(t, "message", duplicatePost.Message)
+	})
+}
+
 func TestUpdatePostEditAt(t *testing.T) {
 	th := Setup().InitBasic()
 	defer th.TearDown()
--- a/app/server.go
+++ b/app/server.go
@@ -91,6 +91,7 @@ type Server struct {

 	htmlTemplateWatcher     *utils.HTMLTemplateWatcher
 	sessionCache            *utils.Cache
+	seenPendingPostIdsCache *utils.Cache
 	configListenerId        string
 	licenseListenerId       string
 	logListenerId           string
@@ -128,13 +129,14 @@ func NewServer(options ...Option) (*Server, error) {
 	rootRouter := mux.NewRouter()

 	s := &Server{
-		goroutineExitSignal: make(chan struct{}, 1),
-		RootRouter:          rootRouter,
-		configFile:          "config.json",
-		configListeners:     make(map[string]func(*model.Config, *model.Config)),
-		licenseListeners:    map[string]func(){},
-		sessionCache:        utils.NewLru(model.SESSION_CACHE_SIZE),
-		clientConfig:        make(map[string]string),
+		goroutineExitSignal:     make(chan struct{}, 1),
+		RootRouter:              rootRouter,
+		configFile:              "config.json",
+		configListeners:         make(map[string]func(*model.Config, *model.Config)),
+		licenseListeners:        map[string]func(){},
+		sessionCache:            utils.NewLru(model.SESSION_CACHE_SIZE),
+		seenPendingPostIdsCache: utils.NewLru(PENDING_POST_IDS_CACHE_SIZE),
+		clientConfig:            make(map[string]string),
 	}
 	for _, option := range options {
 		option(s)