MM-59933: Increment/Decrement the member count directly (#28196)

While using Redis, there is no need to invalidate
and then update the new count from the DB when
we know the exact number it is going to be incremented
or decremented by.

In that case, we can directly use Redis primitives
to update the cache and prevent yet another DB query.

To achieve this, we modify the LRU cache get/set
paths slightly to not marshal into byte slices
for *int64 values. This is needed for Redis to operate
the INCR/DECR commands.

https://mattermost.atlassian.net/browse/MM-59933

```release-note
NONE
```

Co-authored-by: Mattermost Build <build@mattermost.com>

server/channels/store/localcachelayer/channel_layer.go

@@ -441,8 +441,13 @@ func (s LocalCacheChannelStore) SaveMember(rctx request.CTX, member *model.Chann
 	if err != nil {
 		return nil, err
 	}
+
 	// For redis, directly increment member count.
+	if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok {
+		s.rootStore.doIncrementCache(externalCache, member.ChannelId, 1)
+	} else {
 		s.InvalidateMemberCount(member.ChannelId)
+	}
 	return member, nil
 }
 
@@ -452,8 +457,16 @@ func (s LocalCacheChannelStore) SaveMultipleMembers(members []*model.ChannelMemb
 		return nil, err
 	}
 	for _, member := range members {
+		// For redis, directly increment member count.
+		// It should be possible to group the members from the slice
+		// by channelID and increment it once per channel. But it depends
+		// on whether all members are part of the same channel or not.
+		if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok {
+			s.rootStore.doIncrementCache(externalCache, member.ChannelId, 1)
+		} else {
 			s.InvalidateMemberCount(member.ChannelId)
 		}
+	}
 	return members, nil
 }
 
@@ -522,8 +535,13 @@ func (s LocalCacheChannelStore) RemoveMember(rctx request.CTX, channelId, userId
 	if err != nil {
 		return err
 	}
-	// For redis, directly decrement member count
+
+	// For redis, directly decrement member count.
+	if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok {
+		s.rootStore.doDecrementCache(externalCache, channelId, 1)
+	} else {
 		s.InvalidateMemberCount(channelId)
+	}
 	return nil
 }
 
@@ -532,7 +550,11 @@ func (s LocalCacheChannelStore) RemoveMembers(rctx request.CTX, channelId string
 	if err != nil {
 		return err
 	}
-	// For redis, directly decrement member count
+	// For redis, directly decrement member count.
+	if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok {
+		s.rootStore.doDecrementCache(externalCache, channelId, len(userIds))
+	} else {
 		s.InvalidateMemberCount(channelId)
+	}
 	return nil
 }

server/channels/store/localcachelayer/layer.go

@@ -526,6 +526,20 @@ func (s *LocalCacheStore) doMultiReadCache(cache cache.Cache, keys []string, val
 	return errs
 }
 
+func (s *LocalCacheStore) doIncrementCache(cache cache.ExternalCache, key string, val int) {
+	err := cache.Increment(key, val)
+	if err != nil {
+		s.logger.Warn("Error while incrementing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
+}
+
+func (s *LocalCacheStore) doDecrementCache(cache cache.ExternalCache, key string, val int) {
+	err := cache.Decrement(key, val)
+	if err != nil {
+		s.logger.Warn("Error while decrementing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
+}
+
 func (s *LocalCacheStore) doClearCacheCluster(cache cache.Cache) {
 	err := cache.Purge()
 	if err != nil {

server/platform/services/cache/cache.go

@@ -52,3 +52,17 @@ type Cache interface {
 	// Name returns the name of the cache
 	Name() string
 }
+
+// ExternalCache is a super-set of the Cache interface with
+// a couple of more methods that allows for more efficient cache updates.
+// This can be achieved because the cache is external and an update
+// is visible to all nodes.
+type ExternalCache interface {
+	Cache
+	// Increment will increment the
+	// number stored at that key by the value.
+	Increment(key string, val int) error
+	// Decrement will decrement the
+	// number stored at that key by the value.
+	Decrement(key string, val int) error
+}

server/platform/services/cache/mocks/ExternalCache.go

@@ -0,0 +1,249 @@
+// Code generated by mockery v2.42.2. DO NOT EDIT.
+
+// Regenerate this file using `make cache-mocks`.
+
+package mocks
+
+import (
+	time "time"
+
+	model "github.com/mattermost/mattermost/server/public/model"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// ExternalCache is an autogenerated mock type for the ExternalCache type
+type ExternalCache struct {
+	mock.Mock
+}
+
+// Decrement provides a mock function with given fields: key, val
+func (_m *ExternalCache) Decrement(key string, val int) error {
+	ret := _m.Called(key, val)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Decrement")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, int) error); ok {
+		r0 = rf(key, val)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Get provides a mock function with given fields: key, value
+func (_m *ExternalCache) Get(key string, value interface{}) error {
+	ret := _m.Called(key, value)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Get")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, interface{}) error); ok {
+		r0 = rf(key, value)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// GetInvalidateClusterEvent provides a mock function with given fields:
+func (_m *ExternalCache) GetInvalidateClusterEvent() model.ClusterEvent {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetInvalidateClusterEvent")
+	}
+
+	var r0 model.ClusterEvent
+	if rf, ok := ret.Get(0).(func() model.ClusterEvent); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(model.ClusterEvent)
+	}
+
+	return r0
+}
+
+// GetMulti provides a mock function with given fields: keys, values
+func (_m *ExternalCache) GetMulti(keys []string, values []interface{}) []error {
+	ret := _m.Called(keys, values)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetMulti")
+	}
+
+	var r0 []error
+	if rf, ok := ret.Get(0).(func([]string, []interface{}) []error); ok {
+		r0 = rf(keys, values)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]error)
+		}
+	}
+
+	return r0
+}
+
+// Increment provides a mock function with given fields: key, val
+func (_m *ExternalCache) Increment(key string, val int) error {
+	ret := _m.Called(key, val)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Increment")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, int) error); ok {
+		r0 = rf(key, val)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Name provides a mock function with given fields:
+func (_m *ExternalCache) Name() string {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for Name")
+	}
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func() string); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
+// Purge provides a mock function with given fields:
+func (_m *ExternalCache) Purge() error {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for Purge")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Remove provides a mock function with given fields: key
+func (_m *ExternalCache) Remove(key string) error {
+	ret := _m.Called(key)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Remove")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string) error); ok {
+		r0 = rf(key)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// RemoveMulti provides a mock function with given fields: keys
+func (_m *ExternalCache) RemoveMulti(keys []string) error {
+	ret := _m.Called(keys)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RemoveMulti")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func([]string) error); ok {
+		r0 = rf(keys)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Scan provides a mock function with given fields: f
+func (_m *ExternalCache) Scan(f func([]string) error) error {
+	ret := _m.Called(f)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Scan")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(func([]string) error) error); ok {
+		r0 = rf(f)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// SetWithDefaultExpiry provides a mock function with given fields: key, value
+func (_m *ExternalCache) SetWithDefaultExpiry(key string, value interface{}) error {
+	ret := _m.Called(key, value)
+
+	if len(ret) == 0 {
+		panic("no return value specified for SetWithDefaultExpiry")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, interface{}) error); ok {
+		r0 = rf(key, value)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// SetWithExpiry provides a mock function with given fields: key, value, ttl
+func (_m *ExternalCache) SetWithExpiry(key string, value interface{}, ttl time.Duration) error {
+	ret := _m.Called(key, value, ttl)
+
+	if len(ret) == 0 {
+		panic("no return value specified for SetWithExpiry")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, interface{}, time.Duration) error); ok {
+		r0 = rf(key, value, ttl)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// NewExternalCache creates a new instance of ExternalCache. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewExternalCache(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *ExternalCache {
+	mock := &ExternalCache{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

server/platform/services/cache/redis.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -58,6 +59,11 @@ func (r *Redis) SetWithExpiry(key string, value any, ttl time.Duration) error {
 			r.metrics.ObserveRedisEndpointDuration(r.name, "Set", elapsed)
 		}
 	}()
+
+	var valueString string
+	if intVal, ok := value.(int64); ok {
+		valueString = strconv.Itoa(int(intVal))
+	} else {
 		var buf []byte
 		var err error
 		// We use a fast path for hot structs.
@@ -70,16 +76,54 @@ func (r *Redis) SetWithExpiry(key string, value any, ttl time.Duration) error {
 		if err != nil {
 			return err
 		}
+		valueString = rueidis.BinaryString(buf)
+	}
 
 	return r.client.Do(context.Background(),
 		r.client.B().Set().
 			Key(r.name+":"+key).
-			Value(rueidis.BinaryString(buf)).
+			Value(valueString).
 			Ex(ttl).
 			Build(),
 	).Error()
 }
 
+// Increment increments the value of the key by the value.
+func (r *Redis) Increment(key string, val int) error {
+	now := time.Now()
+	defer func() {
+		if r.metrics != nil {
+			elapsed := time.Since(now).Seconds()
+			r.metrics.ObserveRedisEndpointDuration(r.name, "Incr", elapsed)
+		}
+	}()
+
+	return r.client.Do(context.Background(),
+		r.client.B().Incrby().
+			Key(r.name+":"+key).
+			Increment(int64(val)).
+			Build(),
+	).Error()
+}
+
+// Decrement decrements the value of the key by the value.
+func (r *Redis) Decrement(key string, val int) error {
+	now := time.Now()
+	defer func() {
+		if r.metrics != nil {
+			elapsed := time.Since(now).Seconds()
+			r.metrics.ObserveRedisEndpointDuration(r.name, "Decr", elapsed)
+		}
+	}()
+
+	return r.client.Do(context.Background(),
+		r.client.B().Decrby().
+			Key(r.name+":"+key).
+			Decrement(int64(val)).
+			Build(),
+	).Error()
+}
+
 // Get the content stored in the cache for the given key, and decode it into the value interface.
 // Return ErrKeyNotFound if the key is missing from the cache
 func (r *Redis) Get(key string, value any) error {
@@ -90,12 +134,23 @@ func (r *Redis) Get(key string, value any) error {
 			r.metrics.ObserveRedisEndpointDuration(r.name, "Get", elapsed)
 		}
 	}()
-	val, err := r.client.DoCache(context.Background(),
+
+	resp := r.client.DoCache(context.Background(),
 		r.client.B().Get().
 			Key(r.name+":"+key).
 			Cache(),
 		clientSideTTL,
-	).AsBytes()
+	)
+
+	var intVal int64
+	var bytesVal []byte
+	var err error
+	vPtr, ok := value.(*int64)
+	if ok {
+		intVal, err = resp.AsInt64()
+	} else {
+		bytesVal, err = resp.AsBytes()
+	}
 	if err != nil {
 		if rueidis.IsRedisNil(err) {
 			return ErrKeyNotFound
@@ -103,9 +158,14 @@ func (r *Redis) Get(key string, value any) error {
 		return err
 	}
 
+	if ok {
+		*vPtr = intVal
+		return nil
+	}
+
 	// We use a fast path for hot structs.
 	if msgpVal, ok := value.(msgp.Unmarshaler); ok {
-		_, err := msgpVal.UnmarshalMsg(val)
+		_, err := msgpVal.UnmarshalMsg(bytesVal)
 		return err
 	}
 
@@ -120,15 +180,16 @@ func (r *Redis) Get(key string, value any) error {
 	switch v := value.(type) {
 	case **model.User:
 		var u model.User
-		_, err := u.UnmarshalMsg(val)
+		_, err := u.UnmarshalMsg(bytesVal)
 		*v = &u
 		return err
 	}
 
 	// Slow path for other structs.
-	return msgpack.Unmarshal(val, value)
+	return msgpack.Unmarshal(bytesVal, value)
 }
 
+// GetMulti uses the MGET primitive to fetch multiple keys in a single operation.
 func (r *Redis) GetMulti(keys []string, values []any) []error {
 	now := time.Now()
 	defer func() {
@@ -162,21 +223,35 @@ func (r *Redis) GetMulti(keys []string, values []any) []error {
 		return errs
 	}
 
-	for i, val := range vals {
+	for i, resp := range vals {
-		if val.IsNil() {
+		if resp.IsNil() {
 			errs[i] = ErrKeyNotFound
 			continue
 		}
 
-		buf, err := val.AsBytes()
+		var intVal int64
+		var bytesVal []byte
+		var err error
+		vPtr, ok := values[i].(*int64)
+		if ok {
+			intVal, err = resp.AsInt64()
+		} else {
+			bytesVal, err = resp.AsBytes()
+		}
 		if err != nil {
 			errs[i] = err
 			continue
 		}
 
+		if ok {
+			*vPtr = intVal
+			errs[i] = nil
+			continue
+		}
+
 		// We use a fast path for hot structs.
 		if msgpVal, ok := values[i].(msgp.Unmarshaler); ok {
-			_, err := msgpVal.UnmarshalMsg(buf)
+			_, err := msgpVal.UnmarshalMsg(bytesVal)
 			errs[i] = err
 			continue
 		}
@@ -184,14 +259,14 @@ func (r *Redis) GetMulti(keys []string, values []any) []error {
 		switch v := values[i].(type) {
 		case **model.User:
 			var u model.User
-			_, err := u.UnmarshalMsg(buf)
+			_, err := u.UnmarshalMsg(bytesVal)
 			*v = &u
 			errs[i] = err
 			continue
 		}
 
 		// Slow path for other structs.
-		errs[i] = msgpack.Unmarshal(buf, values[i])
+		errs[i] = msgpack.Unmarshal(bytesVal, values[i])
 	}
 
 	return errs