Files
mostlymatter/server/channels/app/busy_test.go
Agniva De Sarker cb75a20c54 MM-61904: Make reliable websockets work in HA (#29489)
We do a cluster request to get the active and dead queues
from other nodes in the cluster to sync any missing
information.

We check the dead queue in the other nodes to see
if there's been any message loss or not. Accordingly,
we send just the active queue or both active and dead queues.

There's still an edge case that is left out where
a client could have potentially connected and reconnected
to multiple nodes leaving multiple active queues
in multiple nodes. We don't handle this scenario
because then potentially we need to create
a slice of sendQueueSize * number_of_nodes. And then
this can happen again, leading to an infinite increase
in sendQueueSize.

We leave this edge-case to Redis, acknowledging
a limitation in our architecture.

In this PR, when there's no message loss, we just
take the active queue from the last node it connected
to.

And if there's message loss where the client's
seqNum is within the last node's dead queue, we also
handle that.

But if there's severe message loss where the client's
seqNum falls within the dead queue of another node, then
we just send the data from that node to reconstruct the
data as much as possible. It could be possible to set
a new connection ID in this case, but this involves
more data transfer always from all nodes and recomputing
the state in the requestor node.

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

```release-note
NONE
```

Co-authored-by: Mattermost Build <build@mattermost.com>
2025-01-17 11:11:32 +05:30

168 строки
5.4 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package app
import (
"encoding/json"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/request"
"github.com/mattermost/mattermost/server/v8/einterfaces"
)
func TestBusySet(t *testing.T) {
cluster := &ClusterMock{Busy: &Busy{}, t: t}
busy := NewBusy(cluster)
isNotBusy := func() bool {
return !busy.IsBusy()
}
require.False(t, busy.IsBusy())
busy.Set(time.Millisecond * 500)
require.True(t, busy.IsBusy())
require.True(t, compareBusyState(t, busy, cluster.Busy))
// should automatically expire after 500ms.
require.Eventually(t, isNotBusy, time.Second*15, time.Millisecond*100)
// allow a moment for cluster to sync.
require.Eventually(t, func() bool { return compareBusyState(t, busy, cluster.Busy) }, time.Second*15, time.Millisecond*20)
// test set after auto expiry.
busy.Set(time.Second * 30)
require.True(t, busy.IsBusy())
require.True(t, compareBusyState(t, busy, cluster.Busy))
expire := busy.Expires()
require.Greater(t, expire.Unix(), time.Now().Add(time.Second*10).Unix())
// test extending existing expiry
busy.Set(time.Minute * 5)
require.True(t, busy.IsBusy())
require.True(t, compareBusyState(t, busy, cluster.Busy))
expire = busy.Expires()
require.Greater(t, expire.Unix(), time.Now().Add(time.Minute*2).Unix())
busy.Clear()
require.False(t, busy.IsBusy())
require.True(t, compareBusyState(t, busy, cluster.Busy))
}
func TestBusyExpires(t *testing.T) {
cluster := &ClusterMock{Busy: &Busy{}, t: t}
busy := NewBusy(cluster)
isNotBusy := func() bool {
return !busy.IsBusy()
}
// get expiry before it is set
expire := busy.Expires()
// should be time.Time zero value
require.Equal(t, time.Time{}.Unix(), expire.Unix())
// get expiry after it is set
busy.Set(time.Minute * 5)
expire = busy.Expires()
require.Greater(t, expire.Unix(), time.Now().Add(time.Minute*2).Unix())
require.True(t, compareBusyState(t, busy, cluster.Busy))
// get expiry after clear
busy.Clear()
expire = busy.Expires()
// should be time.Time zero value
require.Equal(t, time.Time{}.Unix(), expire.Unix())
require.True(t, compareBusyState(t, busy, cluster.Busy))
// get expiry after auto-expire
busy.Set(time.Millisecond * 100)
require.Eventually(t, isNotBusy, time.Second*5, time.Millisecond*20)
expire = busy.Expires()
// should be time.Time zero value
require.Equal(t, time.Time{}.Unix(), expire.Unix())
// allow a moment for cluster to sync
require.Eventually(t, func() bool { return compareBusyState(t, busy, cluster.Busy) }, time.Second*15, time.Millisecond*20)
}
func TestBusyRace(t *testing.T) {
cluster := &ClusterMock{Busy: &Busy{}, t: t}
busy := NewBusy(cluster)
busy.Set(500 * time.Millisecond)
// We are sleeping in order to let the race trigger.
time.Sleep(time.Second)
}
func compareBusyState(t *testing.T, busy1 *Busy, busy2 *Busy) bool {
t.Helper()
if busy1.IsBusy() != busy2.IsBusy() {
busy1JSON, _ := busy1.ToJSON()
busy2JSON, _ := busy2.ToJSON()
t.Logf("busy1:%s; busy2:%s\n", busy1JSON, busy2JSON)
return false
}
if busy1.Expires().Unix() != busy2.Expires().Unix() {
busy1JSON, _ := busy1.ToJSON()
busy2JSON, _ := busy2.ToJSON()
t.Logf("busy1:%s; busy2:%s\n", busy1JSON, busy2JSON)
return false
}
return true
}
// ClusterMock simulates the busy state of a cluster.
type ClusterMock struct {
Busy *Busy
t *testing.T
}
func (c *ClusterMock) SendClusterMessage(msg *model.ClusterMessage) {
var sbs model.ServerBusyState
err := json.Unmarshal(msg.Data, &sbs)
require.NoError(c.t, err)
c.Busy.ClusterEventChanged(&sbs)
}
func (c *ClusterMock) SendClusterMessageToNode(nodeID string, msg *model.ClusterMessage) error {
return nil
}
func (c *ClusterMock) StartInterNodeCommunication() {}
func (c *ClusterMock) StopInterNodeCommunication() {}
func (c *ClusterMock) RegisterClusterMessageHandler(event model.ClusterEvent, crm einterfaces.ClusterMessageHandler) {
}
func (c *ClusterMock) GetClusterId() string { return "cluster_mock" }
func (c *ClusterMock) IsLeader() bool { return false }
func (c *ClusterMock) GetMyClusterInfo() *model.ClusterInfo { return nil }
func (c *ClusterMock) GetClusterInfos() []*model.ClusterInfo { return nil }
func (c *ClusterMock) NotifyMsg(buf []byte) {}
func (c *ClusterMock) GetClusterStats(rctx request.CTX) ([]*model.ClusterStats, *model.AppError) {
return nil, nil
}
func (c *ClusterMock) GetLogs(rctx request.CTX, page, perPage int) ([]string, *model.AppError) {
return nil, nil
}
func (c *ClusterMock) QueryLogs(rctx request.CTX, page, perPage int) (map[string][]string, *model.AppError) {
return nil, nil
}
func (c *ClusterMock) GenerateSupportPacket(rctx request.CTX, options *model.SupportPacketOptions) (map[string][]model.FileData, error) {
return nil, nil
}
func (c *ClusterMock) GetPluginStatuses() (model.PluginStatuses, *model.AppError) { return nil, nil }
func (c *ClusterMock) ConfigChanged(previousConfig *model.Config, newConfig *model.Config, sendToOtherServer bool) *model.AppError {
return nil
}
func (c *ClusterMock) HealthScore() int { return 0 }
func (c *ClusterMock) WebConnCountForUser(userID string) (int, *model.AppError) {
return 0, nil
}
func (c *ClusterMock) GetWSQueues(userID, connectionID string, seqNum int64) (map[string]*model.WSQueues, error) {
return nil, nil
}