Migrates the lastPostsCache from PostStore into cache layer (#13141)

* Some advances

* Partial advances

* Cache moved

* Tests finished

* Removed all test for PostStore (store/sqlstore) related with Cache. This is tested in the cache layer now

model/cluster_message.go

@@ -27,6 +27,7 @@ const (
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_BY_ID                 = "inv_emojis_by_id"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_ID_BY_NAME            = "inv_emojis_id_by_name"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_MEMBER_COUNTS        = "inv_channel_member_counts"
+	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS                   = "inv_last_posts"
 	CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS                 = "inv_all_user_sessions"
 	CLUSTER_EVENT_INSTALL_PLUGIN                                    = "install_plugin"
 	CLUSTER_EVENT_REMOVE_PLUGIN                                     = "remove_plugin"

store/localcachelayer/layer.go

@@ -26,6 +26,9 @@ const (
 	CHANNEL_MEMBERS_COUNTS_CACHE_SIZE = model.CHANNEL_CACHE_SIZE
 	CHANNEL_MEMBERS_COUNTS_CACHE_SEC  = 30 * 60
 
+	LAST_POSTS_CACHE_SIZE = 20000
+	LAST_POSTS_CACHE_SEC  = 30 * 60
+
 	CLEAR_CACHE_MESSAGE_DATA = ""
 )
 
@@ -44,6 +47,8 @@ type LocalCacheStore struct {
 	emojiIdCacheByName       *utils.Cache
 	channel                  LocalCacheChannelStore
 	channelMemberCountsCache *utils.Cache
+	post                     LocalCachePostStore
+	postLastPostsCache       *utils.Cache
 }
 
 func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterface, cluster einterfaces.ClusterInterface) LocalCacheStore {
@@ -63,6 +68,8 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 	localCacheStore.emoji = LocalCacheEmojiStore{EmojiStore: baseStore.Emoji(), rootStore: &localCacheStore}
 	localCacheStore.channelMemberCountsCache = utils.NewLruWithParams(CHANNEL_MEMBERS_COUNTS_CACHE_SIZE, "ChannelMemberCounts", CHANNEL_MEMBERS_COUNTS_CACHE_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_MEMBER_COUNTS)
 	localCacheStore.channel = LocalCacheChannelStore{ChannelStore: baseStore.Channel(), rootStore: &localCacheStore}
+	localCacheStore.postLastPostsCache = utils.NewLruWithParams(LAST_POSTS_CACHE_SIZE, "LastPost", LAST_POSTS_CACHE_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS)
+	localCacheStore.post = LocalCachePostStore{PostStore: baseStore.Post(), rootStore: &localCacheStore}
 
 	if cluster != nil {
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_REACTIONS, localCacheStore.reaction.handleClusterInvalidateReaction)
@@ -71,6 +78,7 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_BY_ID, localCacheStore.emoji.handleClusterInvalidateEmojiById)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_ID_BY_NAME, localCacheStore.emoji.handleClusterInvalidateEmojiIdByName)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_MEMBER_COUNTS, localCacheStore.channel.handleClusterInvalidateChannelMemberCounts)
+		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS, localCacheStore.post.handleClusterInvalidateLastPosts)
 	}
 	return localCacheStore
 }
@@ -95,6 +103,10 @@ func (s LocalCacheStore) Channel() store.ChannelStore {
 	return s.channel
 }
 
+func (s LocalCacheStore) Post() store.PostStore {
+	return s.post
+}
+
 func (s LocalCacheStore) DropAllTables() {
 	s.Invalidate()
 	s.Store.DropAllTables()
@@ -148,4 +160,5 @@ func (s *LocalCacheStore) Invalidate() {
 	s.doClearCacheCluster(s.emojiCacheById)
 	s.doClearCacheCluster(s.emojiIdCacheByName)
 	s.doClearCacheCluster(s.channelMemberCountsCache)
+	s.doClearCacheCluster(s.postLastPostsCache)
 }

store/localcachelayer/layer_test.go

@@ -35,6 +35,19 @@ func StoreTest(t *testing.T, f func(*testing.T, store.Store)) {
 	}
 }
 
+func StoreTestWithSqlSupplier(t *testing.T, f func(*testing.T, store.Store, storetest.SqlSupplier)) {
+	defer func() {
+		if err := recover(); err != nil {
+			tearDownStores()
+			panic(err)
+		}
+	}()
+	for _, st := range storeTypes {
+		st := st
+		t.Run(st.Name, func(t *testing.T) { f(t, st.Store, st.SqlSupplier) })
+	}
+}
+
 func initStores() {
 	storeTypes = append(storeTypes, &storeType{
 		Name:        "LocalCache+MySQL",

store/localcachelayer/main_test.go

@@ -57,6 +57,14 @@ func getMockStore() *mocks.Store {
 	mockChannelStore.On("GetMemberCount", "id", false).Return(mockCount, nil)
 	mockStore.On("Channel").Return(&mockChannelStore)
 
+	fakePosts := &model.PostList{}
+	fakeOptions := model.GetPostsOptions{ChannelId: "123", PerPage: 30}
+	mockPostStore := mocks.PostStore{}
+	mockPostStore.On("GetPosts", fakeOptions, true).Return(fakePosts, nil)
+	mockPostStore.On("GetPosts", fakeOptions, false).Return(fakePosts, nil)
+	mockPostStore.On("InvalidateLastPostTimeCache", "12360")
+	mockStore.On("Post").Return(&mockPostStore)
+
 	return &mockStore
 }
 

store/localcachelayer/post_layer.go

@@ -0,0 +1,70 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package localcachelayer
+
+import (
+	"fmt"
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store"
+)
+
+type LocalCachePostStore struct {
+	store.PostStore
+	rootStore *LocalCacheStore
+}
+
+func (s *LocalCachePostStore) handleClusterInvalidateLastPosts(msg *model.ClusterMessage) {
+	if msg.Data == CLEAR_CACHE_MESSAGE_DATA {
+		s.rootStore.postLastPostsCache.Purge()
+	} else {
+		s.rootStore.postLastPostsCache.Remove(msg.Data)
+	}
+}
+
+func (s LocalCachePostStore) ClearCaches() {
+	s.PostStore.ClearCaches()
+
+	s.rootStore.postLastPostsCache.Purge()
+	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Purge")
+	}
+}
+
+func (s LocalCachePostStore) InvalidateLastPostTimeCache(channelId string) {
+	s.PostStore.InvalidateLastPostTimeCache(channelId)
+
+	// Keys are "{channelid}{limit}" and caching only occurs on limits of 30 and 60
+	s.rootStore.doInvalidateCacheCluster(s.rootStore.postLastPostsCache, channelId+"30")
+	s.rootStore.doInvalidateCacheCluster(s.rootStore.postLastPostsCache, channelId+"60")
+
+	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Remove by Channel Id")
+	}
+}
+
+func (s LocalCachePostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
+	if !allowFromCache {
+		return s.PostStore.GetPosts(options, allowFromCache)
+	}
+
+	offset := options.PerPage * options.Page
+	// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
+	if offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
+		if cacheItem := s.rootStore.doStandardReadCache(s.rootStore.postLastPostsCache, fmt.Sprintf("%s%v", options.ChannelId, options.PerPage)); cacheItem != nil {
+			return cacheItem.(*model.PostList), nil
+		}
+	}
+
+	list, err := s.PostStore.GetPosts(options, false)
+	if err != nil {
+		return nil, err
+	}
+
+	// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
+	if offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
+		s.rootStore.doStandardAddToCache(s.rootStore.postLastPostsCache, fmt.Sprintf("%s%v", options.ChannelId, options.PerPage), list)
+	}
+
+	return list, err
+}

store/localcachelayer/post_layer_test.go

@@ -0,0 +1,65 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package localcachelayer
+
+import (
+	"testing"
+
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store/storetest"
+	"github.com/mattermost/mattermost-server/store/storetest/mocks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPostStore(t *testing.T) {
+	StoreTestWithSqlSupplier(t, storetest.TestPostStore)
+}
+
+func TestPostStoreCache(t *testing.T) {
+	fakePosts := &model.PostList{}
+	fakeOptions := model.GetPostsOptions{ChannelId: "123", PerPage: 30}
+
+	t.Run("first call not cached, second cached and returning same data", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotPosts, err := cachedStore.Post().GetPosts(fakeOptions, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakePosts, gotPosts)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 1)
+
+		_, _ = cachedStore.Post().GetPosts(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 1)
+	})
+
+	t.Run("first call not cached, second force no cached", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotPosts, err := cachedStore.Post().GetPosts(fakeOptions, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakePosts, gotPosts)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 1)
+
+		_, _ = cachedStore.Post().GetPosts(fakeOptions, false)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 2)
+	})
+
+	t.Run("first call not cached, invalidate, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotPosts, err := cachedStore.Post().GetPosts(fakeOptions, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakePosts, gotPosts)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 1)
+
+		cachedStore.Post().InvalidateLastPostTimeCache("12360")
+
+		_, _ = cachedStore.Post().GetPosts(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPosts", 1)
+
+	})
+}

store/sqlstore/post_store.go

@@ -24,7 +24,6 @@ type SqlPostStore struct {
 	SqlStore
 	metrics           einterfaces.MetricsInterface
 	lastPostTimeCache *utils.Cache
-	lastPostsCache    *utils.Cache
 	maxPostSizeOnce   sync.Once
 	maxPostSizeCached int
 }
@@ -32,18 +31,13 @@ type SqlPostStore struct {
 const (
 	LAST_POST_TIME_CACHE_SIZE = 25000
 	LAST_POST_TIME_CACHE_SEC  = 900 // 15 minutes
-
-	LAST_POSTS_CACHE_SIZE = 1000
-	LAST_POSTS_CACHE_SEC  = 900 // 15 minutes
 )
 
 func (s *SqlPostStore) ClearCaches() {
 	s.lastPostTimeCache.Purge()
-	s.lastPostsCache.Purge()
 
 	if s.metrics != nil {
 		s.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Purge")
-		s.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Purge")
 	}
 }
 
@@ -52,7 +46,6 @@ func NewSqlPostStore(sqlStore SqlStore, metrics einterfaces.MetricsInterface) st
 		SqlStore:          sqlStore,
 		metrics:           metrics,
 		lastPostTimeCache: utils.NewLru(LAST_POST_TIME_CACHE_SIZE),
-		lastPostsCache:    utils.NewLru(LAST_POSTS_CACHE_SIZE),
 		maxPostSizeCached: model.POST_MESSAGE_MAX_RUNES_V1,
 	}
 
@@ -329,13 +322,8 @@ type etagPosts struct {
 func (s *SqlPostStore) InvalidateLastPostTimeCache(channelId string) {
 	s.lastPostTimeCache.Remove(channelId)
 
-	// Keys are "{channelid}{limit}" and caching only occurs on limits of 30 and 60
-	s.lastPostsCache.Remove(channelId + "30")
-	s.lastPostsCache.Remove(channelId + "60")
-
 	if s.metrics != nil {
 		s.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Remove by Channel Id")
-		s.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Remove by Channel Id")
 	}
 }
 
@@ -451,24 +439,11 @@ func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) *model.AppErro
 	return nil
 }
 
-func (s *SqlPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
+func (s *SqlPostStore) GetPosts(options model.GetPostsOptions, _ bool) (*model.PostList, *model.AppError) {
 	if options.PerPage > 1000 {
 		return nil, model.NewAppError("SqlPostStore.GetLinearPosts", "store.sql_post.get_posts.app_error", nil, "channelId="+options.ChannelId, http.StatusBadRequest)
 	}
 	offset := options.PerPage * options.Page
-	// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
-	if allowFromCache && offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
-		if cacheItem, ok := s.lastPostsCache.Get(fmt.Sprintf("%s%v", options.ChannelId, options.PerPage)); ok {
-			if s.metrics != nil {
-				s.metrics.IncrementMemCacheHitCounter("Last Posts Cache")
-			}
-			return cacheItem.(*model.PostList), nil
-		}
-	}
-
-	if s.metrics != nil {
-		s.metrics.IncrementMemCacheMissCounter("Last Posts Cache")
-	}
 
 	rpc := make(chan store.StoreResult, 1)
 	go func() {
@@ -510,11 +485,6 @@ func (s *SqlPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bo
 
 	list.MakeNonNil()
 
-	// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
-	if offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
-		s.lastPostsCache.AddWithExpiresInSecs(fmt.Sprintf("%s%v", options.ChannelId, options.PerPage), list, LAST_POSTS_CACHE_SEC)
-	}
-
 	return list, err
 }
 

store/storetest/post_store.go

@@ -697,7 +697,7 @@ func testPostStoreGetPostsWithDetails(t *testing.T, ss store.Store) {
 		t.Fatal("Missing parent")
 	}
 
-	r2, err := ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 4}, true)
+	r2, err := ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 4}, false)
 	require.Nil(t, err)
 
 	if r2.Order[0] != o5.Id {
@@ -725,7 +725,7 @@ func testPostStoreGetPostsWithDetails(t *testing.T, ss store.Store) {
 	}
 
 	// Run once to fill cache
-	_, err = ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 30}, true)
+	_, err = ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 30}, false)
 	require.Nil(t, err)
 
 	o6 := &model.Post{}
@@ -735,17 +735,9 @@ func testPostStoreGetPostsWithDetails(t *testing.T, ss store.Store) {
 	_, err = ss.Post().Save(o6)
 	require.Nil(t, err)
 
-	// Should only be 6 since we hit the cache
+	r3, err := ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 30}, false)
-	r3, err := ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 30}, true)
 	require.Nil(t, err)
-	assert.Equal(t, 6, len(r3.Order))
+	assert.Equal(t, 7, len(r3.Order))
-
-	ss.Post().InvalidateLastPostTimeCache(o1.ChannelId)
-
-	// Cache was invalidated, we should get all the posts
-	r4, err := ss.Post().GetPosts(model.GetPostsOptions{ChannelId: o1.ChannelId, Page: 0, PerPage: 30}, true)
-	require.Nil(t, err)
-	assert.Equal(t, 7, len(r4.Order))
 }
 
 func testPostStoreGetPostsBeforeAfter(t *testing.T, ss store.Store) {