[GH-13099] Migrate lastPostTimeCache cache from store/sqlstore/post_store.go to the new store/localcachelayer (#13134)

Automatic Merge

model/cluster_message.go

@@ -32,6 +32,7 @@ const (
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_PINNEDPOSTS_COUNTS   = "inv_channel_pinnedposts_counts"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_MEMBER_COUNTS        = "inv_channel_member_counts"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS                   = "inv_last_posts"
+	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POST_TIME               = "inv_last_post_time"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_TEAMS                        = "inv_teams"
 	CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS                 = "inv_all_user_sessions"
 	CLUSTER_EVENT_INSTALL_PLUGIN                                    = "install_plugin"

store/localcachelayer/layer.go

@@ -38,6 +38,9 @@ const (
 	LAST_POSTS_CACHE_SIZE = 20000
 	LAST_POSTS_CACHE_SEC  = 30 * 60
 
+	LAST_POST_TIME_CACHE_SIZE = 25000
+	LAST_POST_TIME_CACHE_SEC  = 15 * 60
+
 	USER_PROFILE_BY_ID_CACHE_SIZE = 20000
 	USER_PROFILE_BY_ID_SEC        = 30 * 60
 
@@ -68,6 +71,7 @@ type LocalCacheStore struct {
 	webhookCache                 *utils.Cache
 	post                         LocalCachePostStore
 	postLastPostsCache           *utils.Cache
+	lastPostTimeCache            *utils.Cache
 	user                         LocalCacheUserStore
 	userProfileByIdsCache        *utils.Cache
 	team                         LocalCacheTeamStore
@@ -95,6 +99,7 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 	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.channelGuestCountCache = utils.NewLruWithParams(CHANNEL_GUEST_COUNT_CACHE_SIZE, "ChannelGuestsCount", CHANNEL_GUEST_COUNT_CACHE_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_GUEST_COUNT)
 	localCacheStore.channel = LocalCacheChannelStore{ChannelStore: baseStore.Channel(), rootStore: &localCacheStore}
+	localCacheStore.lastPostTimeCache = utils.NewLruWithParams(LAST_POST_TIME_CACHE_SIZE, "LastPostTime", LAST_POST_TIME_CACHE_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POST_TIME)
 	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}
 	localCacheStore.userProfileByIdsCache = utils.NewLruWithParams(USER_PROFILE_BY_ID_CACHE_SIZE, "UserProfileByIds", USER_PROFILE_BY_ID_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_PROFILE_BY_IDS)
@@ -106,6 +111,7 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_REACTIONS, localCacheStore.reaction.handleClusterInvalidateReaction)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_ROLES, localCacheStore.role.handleClusterInvalidateRole)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_SCHEMES, localCacheStore.scheme.handleClusterInvalidateScheme)
+		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POST_TIME, localCacheStore.post.handleClusterInvalidateLastPostTime)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_WEBHOOKS, localCacheStore.webhook.handleClusterInvalidateWebhook)
 		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)
@@ -212,6 +218,7 @@ func (s *LocalCacheStore) Invalidate() {
 	s.doClearCacheCluster(s.channelPinnedPostCountsCache)
 	s.doClearCacheCluster(s.channelGuestCountCache)
 	s.doClearCacheCluster(s.postLastPostsCache)
+	s.doClearCacheCluster(s.lastPostTimeCache)
 	s.doClearCacheCluster(s.userProfileByIdsCache)
 	s.doClearCacheCluster(s.teamAllTeamIdsForUserCache)
 }

store/localcachelayer/main_test.go

@@ -4,6 +4,7 @@
 package localcachelayer
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/mattermost/mattermost-server/v5/model"
@@ -77,6 +78,20 @@ func getMockStore() *mocks.Store {
 	mockPostStore.On("GetPosts", fakeOptions, true).Return(fakePosts, nil)
 	mockPostStore.On("GetPosts", fakeOptions, false).Return(fakePosts, nil)
 	mockPostStore.On("InvalidateLastPostTimeCache", "12360")
+
+	mockPostStoreOptions := model.GetPostsSinceOptions{
+		ChannelId:        "channelId",
+		Time:             1,
+		SkipFetchThreads: false,
+	}
+
+	mockPostStoreEtagResult := fmt.Sprintf("%v.%v", model.CurrentVersion, 1)
+	mockPostStore.On("ClearCaches")
+	mockPostStore.On("InvalidateLastPostTimeCache", "channelId")
+	mockPostStore.On("GetEtag", "channelId", true).Return(mockPostStoreEtagResult)
+	mockPostStore.On("GetEtag", "channelId", false).Return(mockPostStoreEtagResult)
+	mockPostStore.On("GetPostsSince", mockPostStoreOptions, true).Return(model.NewPostList(), nil)
+	mockPostStore.On("GetPostsSince", mockPostStoreOptions, false).Return(model.NewPostList(), nil)
 	mockStore.On("Post").Return(&mockPostStore)
 
 	fakeUser := []*model.User{{Id: "123"}}

store/localcachelayer/post_layer.go

@@ -4,9 +4,13 @@
 package localcachelayer
 
 import (
-	"fmt"
+	"strconv"
+	"strings"
+
 	"github.com/mattermost/mattermost-server/v5/model"
 	"github.com/mattermost/mattermost-server/v5/store"
+
+	"fmt"
 )
 
 type LocalCachePostStore struct {
@@ -14,6 +18,14 @@ type LocalCachePostStore struct {
 	rootStore *LocalCacheStore
 }
 
+func (s *LocalCachePostStore) handleClusterInvalidateLastPostTime(msg *model.ClusterMessage) {
+	if msg.Data == CLEAR_CACHE_MESSAGE_DATA {
+		s.rootStore.lastPostTimeCache.Purge()
+	} else {
+		s.rootStore.lastPostTimeCache.Remove(msg.Data)
+	}
+}
+
 func (s *LocalCachePostStore) handleClusterInvalidateLastPosts(msg *model.ClusterMessage) {
 	if msg.Data == CLEAR_CACHE_MESSAGE_DATA {
 		s.rootStore.postLastPostsCache.Purge()
@@ -23,26 +35,74 @@ func (s *LocalCachePostStore) handleClusterInvalidateLastPosts(msg *model.Cluste
 }
 
 func (s LocalCachePostStore) ClearCaches() {
+	s.rootStore.doClearCacheCluster(s.rootStore.lastPostTimeCache)
+	s.rootStore.doClearCacheCluster(s.rootStore.postLastPostsCache)
 	s.PostStore.ClearCaches()
 
-	s.rootStore.postLastPostsCache.Purge()
 	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Purge")
 		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Purge")
 	}
 }
 
 func (s LocalCachePostStore) InvalidateLastPostTimeCache(channelId string) {
-	s.PostStore.InvalidateLastPostTimeCache(channelId)
+	s.rootStore.doInvalidateCacheCluster(s.rootStore.lastPostTimeCache, 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")
 
+	s.PostStore.InvalidateLastPostTimeCache(channelId)
+
 	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Remove by Channel Id")
 		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Last Posts Cache - Remove by Channel Id")
 	}
 }
 
+func (s LocalCachePostStore) GetEtag(channelId string, allowFromCache bool) string {
+	if allowFromCache {
+		if lastTime := s.rootStore.doStandardReadCache(s.rootStore.lastPostTimeCache, channelId); lastTime != nil {
+			return fmt.Sprintf("%v.%v", model.CurrentVersion, lastTime.(int64))
+		}
+	}
+
+	result := s.PostStore.GetEtag(channelId, allowFromCache)
+
+	splittedResult := strings.Split(result, ".")
+
+	lastTime, _ := strconv.ParseInt((splittedResult[len(splittedResult)-1]), 10, 64)
+
+	s.rootStore.doStandardAddToCache(s.rootStore.lastPostTimeCache, channelId, lastTime)
+
+	return result
+}
+
+func (s LocalCachePostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
+	if allowFromCache {
+		// If the last post in the channel's time is less than or equal to the time we are getting posts since,
+		// we can safely return no posts.
+		if lastTime := s.rootStore.doStandardReadCache(s.rootStore.lastPostTimeCache, options.ChannelId); lastTime != nil && lastTime.(int64) <= options.Time {
+			list := model.NewPostList()
+			return list, nil
+		}
+	}
+
+	list, err := s.PostStore.GetPostsSince(options, allowFromCache)
+
+	latestUpdate := options.Time
+	if err == nil {
+		for _, p := range list.ToSlice() {
+			if latestUpdate < p.UpdateAt {
+				latestUpdate = p.UpdateAt
+			}
+		}
+		s.rootStore.doStandardAddToCache(s.rootStore.lastPostTimeCache, options.ChannelId, latestUpdate)
+	}
+
+	return list, err
+}
+
 func (s LocalCachePostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
 	if !allowFromCache {
 		return s.PostStore.GetPosts(options, allowFromCache)

store/localcachelayer/post_layer_test.go

@@ -4,6 +4,7 @@
 package localcachelayer
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/mattermost/mattermost-server/v5/model"
@@ -17,6 +18,112 @@ func TestPostStore(t *testing.T) {
 	StoreTestWithSqlSupplier(t, storetest.TestPostStore)
 }
 
+func TestPostStoreLastPostTimeCache(t *testing.T) {
+	var fakeLastTime int64 = 1
+	channelId := "channelId"
+	fakeOptions := model.GetPostsSinceOptions{
+		ChannelId:        channelId,
+		Time:             fakeLastTime,
+		SkipFetchThreads: false,
+	}
+
+	t.Run("GetEtag: first call not cached, second cached and returning same data", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		expectedResult := fmt.Sprintf("%v.%v", model.CurrentVersion, fakeLastTime)
+
+		etag := cachedStore.Post().GetEtag(channelId, true)
+		assert.Equal(t, etag, expectedResult)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 1)
+
+		etag = cachedStore.Post().GetEtag(channelId, true)
+		assert.Equal(t, etag, expectedResult)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 1)
+	})
+
+	t.Run("GetEtag: first call not cached, second force no cached", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetEtag(channelId, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 1)
+		cachedStore.Post().GetEtag(channelId, false)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 2)
+	})
+
+	t.Run("GetEtag: first call not cached, invalidate, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetEtag(channelId, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 1)
+		cachedStore.Post().InvalidateLastPostTimeCache(channelId)
+		cachedStore.Post().GetEtag(channelId, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 2)
+	})
+
+	t.Run("GetEtag: first call not cached, clear caches, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetEtag(channelId, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 1)
+		cachedStore.Post().ClearCaches()
+		cachedStore.Post().GetEtag(channelId, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetEtag", 2)
+	})
+
+	t.Run("GetPostsSince: first call not cached, second cached and returning same data", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		expectedResult := model.NewPostList()
+
+		list, err := cachedStore.Post().GetPostsSince(fakeOptions, true)
+		require.Nil(t, err)
+		assert.Equal(t, list, expectedResult)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 1)
+
+		list, err = cachedStore.Post().GetPostsSince(fakeOptions, true)
+		require.Nil(t, err)
+		assert.Equal(t, list, expectedResult)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 1)
+	})
+
+	t.Run("GetPostsSince: first call not cached, second force no cached", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetPostsSince(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 1)
+		cachedStore.Post().GetPostsSince(fakeOptions, false)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 2)
+	})
+
+	t.Run("GetPostsSince: first call not cached, invalidate, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetPostsSince(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 1)
+		cachedStore.Post().InvalidateLastPostTimeCache(channelId)
+		cachedStore.Post().GetPostsSince(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 2)
+	})
+
+	t.Run("GetPostsSince: first call not cached, clear caches, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		cachedStore.Post().GetPostsSince(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 1)
+		cachedStore.Post().ClearCaches()
+		cachedStore.Post().GetPostsSince(fakeOptions, true)
+		mockStore.Post().(*mocks.PostStore).AssertNumberOfCalls(t, "GetPostsSince", 2)
+	})
+}
+
 func TestPostStoreCache(t *testing.T) {
 	fakePosts := &model.PostList{}
 	fakeOptions := model.GetPostsOptions{ChannelId: "123", PerPage: 30}

store/sqlstore/post_store.go

@@ -23,29 +23,17 @@ import (
 type SqlPostStore struct {
 	SqlStore
 	metrics           einterfaces.MetricsInterface
-	lastPostTimeCache *utils.Cache
 	maxPostSizeOnce   sync.Once
 	maxPostSizeCached int
 }
 
-const (
-	LAST_POST_TIME_CACHE_SIZE = 25000
-	LAST_POST_TIME_CACHE_SEC  = 900 // 15 minutes
-)
-
 func (s *SqlPostStore) ClearCaches() {
-	s.lastPostTimeCache.Purge()
-
-	if s.metrics != nil {
-		s.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Purge")
-	}
 }
 
 func NewSqlPostStore(sqlStore SqlStore, metrics einterfaces.MetricsInterface) store.PostStore {
 	s := &SqlPostStore{
 		SqlStore:          sqlStore,
 		metrics:           metrics,
-		lastPostTimeCache: utils.NewLru(LAST_POST_TIME_CACHE_SIZE),
 		maxPostSizeCached: model.POST_MESSAGE_MAX_RUNES_V1,
 	}
 
@@ -316,31 +304,9 @@ type etagPosts struct {
 }
 
 func (s *SqlPostStore) InvalidateLastPostTimeCache(channelId string) {
-	s.lastPostTimeCache.Remove(channelId)
-
-	if s.metrics != nil {
-		s.metrics.IncrementMemCacheInvalidationCounter("Last Post Time - Remove by Channel Id")
-	}
 }
 
 func (s *SqlPostStore) GetEtag(channelId string, allowFromCache bool) string {
-	if allowFromCache {
-		if cacheItem, ok := s.lastPostTimeCache.Get(channelId); ok {
-			if s.metrics != nil {
-				s.metrics.IncrementMemCacheHitCounter("Last Post Time")
-			}
-			return fmt.Sprintf("%v.%v", model.CurrentVersion, cacheItem.(int64))
-		} else {
-			if s.metrics != nil {
-				s.metrics.IncrementMemCacheMissCounter("Last Post Time")
-			}
-		}
-	} else {
-		if s.metrics != nil {
-			s.metrics.IncrementMemCacheMissCounter("Last Post Time")
-		}
-	}
-
 	var et etagPosts
 	err := s.GetReplica().SelectOne(&et, "SELECT Id, UpdateAt FROM Posts WHERE ChannelId = :ChannelId ORDER BY UpdateAt DESC LIMIT 1", map[string]interface{}{"ChannelId": channelId})
 	var result string
@@ -350,7 +316,6 @@ func (s *SqlPostStore) GetEtag(channelId string, allowFromCache bool) string {
 		result = fmt.Sprintf("%v.%v", model.CurrentVersion, et.UpdateAt)
 	}
 
-	s.lastPostTimeCache.AddWithExpiresInSecs(channelId, et.UpdateAt, LAST_POST_TIME_CACHE_SEC)
 	return result
 }
 
@@ -485,22 +450,6 @@ func (s *SqlPostStore) GetPosts(options model.GetPostsOptions, _ bool) (*model.P
 }
 
 func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
-	if allowFromCache {
-		// If the last post in the channel's time is less than or equal to the time we are getting posts since,
-		// we can safely return no posts.
-		if cacheItem, ok := s.lastPostTimeCache.Get(options.ChannelId); ok && cacheItem.(int64) <= options.Time {
-			if s.metrics != nil {
-				s.metrics.IncrementMemCacheHitCounter("Last Post Time")
-			}
-			list := model.NewPostList()
-			return list, nil
-		}
-	}
-
-	if s.metrics != nil {
-		s.metrics.IncrementMemCacheMissCounter("Last Post Time")
-	}
-
 	var posts []*model.Post
 
 	replyCountQuery1 := ""
@@ -544,20 +493,13 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr
 
 	list := model.NewPostList()
 
-	latestUpdate := options.Time
-
 	for _, p := range posts {
 		list.AddPost(p)
 		if p.UpdateAt > options.Time {
 			list.AddOrder(p.Id)
 		}
-		if latestUpdate < p.UpdateAt {
-			latestUpdate = p.UpdateAt
-		}
 	}
 
-	s.lastPostTimeCache.AddWithExpiresInSecs(options.ChannelId, latestUpdate, LAST_POST_TIME_CACHE_SEC)
-
 	return list, nil
 }