mirror of
https://github.com/hanzoai/team.git
synced 2026-09-30 01:57:35 +00:00
* MM-69792: Recover shared channel sync after a brief remote outage When a remote went offline for less than the 5-minute IsOnline() threshold, sync recovery never ran. IsOnline() never flipped, so no connection-state-change event fired and ForceSyncForRemote was never triggered. Posts created during the outage stayed unsynced until the next organic change in the channel. The root cause was that send failures are reported asynchronously through a callback that sendSyncMsgToRemote discarded. On a failed send syncForRemote still returned nil, so processTask saw success: the retry loop and the failure-notification hook were effectively dead code for delivery failures. Changes: - sendSyncMsgToRemote now propagates the delivery error reported via the callback instead of swallowing it. A failed send is retried rather than silently dropped. This applies to every sync type (posts, users, memberships, reactions, statuses); the success path is unchanged. - remotecluster tracks a per-remote "sync failed since last ping" flag (NotifySyncFailed). On the next successful ping, PingNow fires a connection-state-change event even when IsOnline() never flipped, driving a single ForceSyncForRemote on recovery. - Sync retries are now spaced by SyncRetryDelay (15s) instead of retrying immediately, giving a short outage time to recover and guaranteeing a failed ping before retries exhaust. Also adds a post/offline-recovery case to the sharedchannel-test integration tool (stop and restart Server B mid-run) and disables metrics on the two test servers so they do not collide on the metrics port when the loaded config has metrics enabled. * fix linter error * Fix Rabbit nit * fix CI * Delete pr37499-fix-findings-1-and-4.md * address review comments * MM-69792: Keep a per-remote retry task when a shared channel sync fans out A sync task with no specific remote (remoteID == "") fans out to every remote sharing the channel. On failure each remote's retry was a copy of the original task that still carried the original, remote-less task id. Because addTask merges on task id, all of those per-remote retries collapsed into a single entry, so only one failed remote was ever retried and the rest were silently dropped. Delivery failures were partly masked by the ping-driven ForceSyncForRemote recovery, but non-delivery errors for the other remotes had no path back. Give each per-remote retry a remote-specific id (task.id + remoteId) when the originating task had no remote, so every failed remote keeps its own retry task. Single-remote tasks already have a remote-specific id and are left untouched, since recomputing would grow the id on each retry. Also harden the tests: assert the retry is scheduled the full SyncRetryDelay out (no tolerance), assert the result callback is not invoked on delivery failures or unconfirmed responses (so the cursor stays un-advanced), and add a regression test covering fan-out to multiple failing remotes. * fix one more pre-existing edge case. * fix linter
306 lines
8.8 KiB
Go
306 lines
8.8 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/mattermost/mattermost/server/public/model"
|
|
"github.com/mattermost/mattermost/server/public/shared/mlog"
|
|
)
|
|
|
|
// TestResult tracks the outcome of a single test case.
|
|
type TestResult struct {
|
|
Name string
|
|
Passed bool
|
|
Detail string
|
|
}
|
|
|
|
// TestRunner orchestrates the full integration test sequence.
|
|
type TestRunner struct {
|
|
cfg Config
|
|
logger *mlog.Logger
|
|
mgr *ServerManager
|
|
clientA *model.Client4
|
|
clientB *model.Client4
|
|
teamA *model.Team
|
|
teamB *model.Team
|
|
remoteID string // single remote cluster connection, reused for all tests
|
|
suffix string // random suffix for unique names across repeated runs
|
|
results []TestResult
|
|
}
|
|
|
|
func NewTestRunner(cfg Config, logger *mlog.Logger) *TestRunner {
|
|
return &TestRunner{
|
|
cfg: cfg,
|
|
logger: logger,
|
|
suffix: model.NewId()[:6],
|
|
}
|
|
}
|
|
|
|
func (tr *TestRunner) pass(name string) {
|
|
tr.results = append(tr.results, TestResult{Name: name, Passed: true})
|
|
tr.logger.Info("PASS", mlog.String("test", name))
|
|
}
|
|
|
|
func (tr *TestRunner) fail(name, detail string) {
|
|
tr.results = append(tr.results, TestResult{Name: name, Passed: false, Detail: detail})
|
|
tr.logger.Error("FAIL", mlog.String("test", name), mlog.String("detail", detail))
|
|
}
|
|
|
|
// Run executes the full test suite.
|
|
func (tr *TestRunner) Run(ctx context.Context) error {
|
|
// Lifecycle management
|
|
if tr.cfg.Manage {
|
|
tr.mgr = NewServerManager(tr.cfg, tr.logger)
|
|
defer tr.mgr.Teardown()
|
|
if err := tr.mgr.Setup(ctx); err != nil {
|
|
return fmt.Errorf("server setup: %w", err)
|
|
}
|
|
}
|
|
|
|
// Provision
|
|
if err := tr.provision(ctx); err != nil {
|
|
return fmt.Errorf("provision: %w", err)
|
|
}
|
|
|
|
// Run test suites
|
|
if err := tr.runMembershipTests(ctx); err != nil {
|
|
tr.logger.Error("Membership tests had errors", mlog.Err(err))
|
|
}
|
|
if err := tr.runPostTests(ctx); err != nil {
|
|
tr.logger.Error("Post tests had errors", mlog.Err(err))
|
|
}
|
|
if err := tr.runReactionTests(ctx); err != nil {
|
|
tr.logger.Error("Reaction tests had errors", mlog.Err(err))
|
|
}
|
|
if err := tr.runOfflineRecoveryTests(ctx); err != nil {
|
|
tr.logger.Error("Offline recovery tests had errors", mlog.Err(err))
|
|
}
|
|
|
|
// Report
|
|
return tr.report()
|
|
}
|
|
|
|
func (tr *TestRunner) provision(ctx context.Context) error {
|
|
var err error
|
|
|
|
// Create admin users and login
|
|
tr.logger.Info("Provisioning admin users...")
|
|
tr.clientA, err = ProvisionAdmin(ctx, tr.cfg.ServerAURL, "mattermost_test",
|
|
tr.cfg.AdminUser, "[email protected]", tr.cfg.AdminPass)
|
|
if err != nil {
|
|
return fmt.Errorf("provision server A: %w", err)
|
|
}
|
|
|
|
tr.clientB, err = ProvisionAdmin(ctx, tr.cfg.ServerBURL, "mattermost_node_test",
|
|
tr.cfg.AdminUser, "[email protected]", tr.cfg.AdminPass)
|
|
if err != nil {
|
|
return fmt.Errorf("provision server B: %w", err)
|
|
}
|
|
|
|
// Upload license
|
|
tr.logger.Info("Uploading license...")
|
|
if err := UploadLicense(ctx, tr.clientA, tr.cfg.LicensePath); err != nil {
|
|
return fmt.Errorf("license server A: %w", err)
|
|
}
|
|
if err := UploadLicense(ctx, tr.clientB, tr.cfg.LicensePath); err != nil {
|
|
return fmt.Errorf("license server B: %w", err)
|
|
}
|
|
|
|
// Wait for remote cluster service to become available after license upload
|
|
if err := tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
_, _, err := tr.clientA.GetRemoteClusters(ctx, 0, 1, model.RemoteClusterQueryFilter{})
|
|
return err == nil
|
|
}); err != nil {
|
|
return fmt.Errorf("remote cluster service not ready on server A: %w", err)
|
|
}
|
|
|
|
// Create teams
|
|
tr.logger.Info("Creating teams...")
|
|
tr.teamA, _, err = tr.clientA.CreateTeam(ctx, &model.Team{
|
|
Name: "team-a-" + tr.suffix,
|
|
DisplayName: "Team A",
|
|
Type: model.TeamOpen,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("create team A: %w", err)
|
|
}
|
|
|
|
tr.teamB, _, err = tr.clientB.CreateTeam(ctx, &model.Team{
|
|
Name: "team-b-" + tr.suffix,
|
|
DisplayName: "Team B",
|
|
Type: model.TeamOpen,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("create team B: %w", err)
|
|
}
|
|
|
|
// Create a single remote cluster connection for all tests
|
|
tr.logger.Info("Setting up remote cluster connection...")
|
|
invitePassword := model.NewTestPassword()
|
|
invite, _, err := tr.clientA.CreateRemoteCluster(ctx, &model.RemoteClusterWithPassword{
|
|
RemoteCluster: &model.RemoteCluster{
|
|
Name: "server-b-" + tr.suffix,
|
|
DisplayName: "Server B",
|
|
DefaultTeamId: tr.teamA.Id,
|
|
},
|
|
Password: invitePassword,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("create remote cluster: %w", err)
|
|
}
|
|
tr.remoteID = invite.RemoteCluster.RemoteId
|
|
|
|
_, _, err = tr.clientB.RemoteClusterAcceptInvite(ctx, &model.RemoteClusterAcceptInvite{
|
|
Name: "server-a-" + tr.suffix,
|
|
DisplayName: "Server A",
|
|
DefaultTeamId: tr.teamB.Id,
|
|
Invite: invite.Invite,
|
|
Password: invitePassword,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("accept invite: %w", err)
|
|
}
|
|
|
|
if err := tr.waitForRemoteOnline(ctx, tr.remoteID); err != nil {
|
|
return fmt.Errorf("remote cluster not online: %w", err)
|
|
}
|
|
tr.logger.Info("Remote cluster connection established")
|
|
|
|
return nil
|
|
}
|
|
|
|
// setupSharedChannel creates a channel on Server A, shares it with Server B via
|
|
// the pre-established remote cluster, and returns both channel IDs.
|
|
func (tr *TestRunner) setupSharedChannel(ctx context.Context, channelName string) (channelA, channelB string, err error) {
|
|
uniqueName := channelName + "-" + tr.suffix
|
|
ch, _, err := tr.clientA.CreateChannel(ctx, &model.Channel{
|
|
TeamId: tr.teamA.Id,
|
|
Name: uniqueName,
|
|
DisplayName: "Shared: " + channelName,
|
|
Type: model.ChannelTypeOpen,
|
|
})
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("create channel: %w", err)
|
|
}
|
|
channelA = ch.Id
|
|
|
|
// Share the channel through the existing remote
|
|
_, err = tr.clientA.InviteRemoteClusterToChannel(ctx, tr.remoteID, channelA)
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("share channel: %w", err)
|
|
}
|
|
|
|
// Wait for channel to appear on B
|
|
tr.logger.Info("Waiting for channel to sync to Server B...", mlog.String("channel", channelName))
|
|
if err := tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
channels, _, err := tr.clientB.GetAllSharedChannels(ctx, tr.teamB.Id, 0, 100)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
for _, sc := range channels {
|
|
if sc.ShareName == uniqueName {
|
|
channelB = sc.ChannelId
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}); err != nil {
|
|
return "", "", fmt.Errorf("channel did not appear on server B: %w", err)
|
|
}
|
|
|
|
return channelA, channelB, nil
|
|
}
|
|
|
|
func (tr *TestRunner) waitForRemoteOnline(ctx context.Context, remoteID string) error {
|
|
return tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
rc, _, err := tr.clientA.GetRemoteCluster(ctx, remoteID)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return rc.LastPingAt > 0
|
|
})
|
|
}
|
|
|
|
func (tr *TestRunner) waitFor(ctx context.Context, timeout time.Duration, condition func() bool) error {
|
|
deadline := time.After(timeout)
|
|
ticker := time.NewTicker(2 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-deadline:
|
|
return fmt.Errorf("timed out after %s", timeout)
|
|
case <-ticker.C:
|
|
if condition() {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// createTestUser creates a user on Server A, adds them to the team, and returns their ID.
|
|
func (tr *TestRunner) createTestUser(ctx context.Context, username string) (string, error) {
|
|
uniqueName := username + "-" + tr.suffix
|
|
user, _, err := tr.clientA.CreateUser(ctx, &model.User{
|
|
Username: uniqueName,
|
|
Email: uniqueName + "@test.local",
|
|
Password: model.NewTestPassword(),
|
|
})
|
|
if err != nil {
|
|
return "", fmt.Errorf("create user %s: %w", username, err)
|
|
}
|
|
|
|
_, _, err = tr.clientA.AddTeamMember(ctx, tr.teamA.Id, user.Id)
|
|
if err != nil {
|
|
return "", fmt.Errorf("add user %s to team: %w", username, err)
|
|
}
|
|
|
|
return user.Id, nil
|
|
}
|
|
|
|
// verifyMemberOnB checks whether a user (by username prefix) is a member of a channel on Server B.
|
|
func (tr *TestRunner) verifyMemberOnB(ctx context.Context, channelB, usernamePrefix string) bool {
|
|
users, _, err := tr.clientB.GetUsersInChannel(ctx, channelB, 0, 200, "")
|
|
if err != nil {
|
|
return false
|
|
}
|
|
for _, u := range users {
|
|
if len(u.Username) >= len(usernamePrefix) && u.Username[:len(usernamePrefix)] == usernamePrefix {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// report prints the final summary. Errors go to stderr so that a clean
|
|
// stderr signals success in CI/scripted environments. Returns non-nil
|
|
// on any failure so the process exits with a non-zero code.
|
|
func (tr *TestRunner) report() error {
|
|
passed, failed := 0, 0
|
|
for _, r := range tr.results {
|
|
if r.Passed {
|
|
passed++
|
|
} else {
|
|
failed++
|
|
}
|
|
}
|
|
|
|
tr.logger.Info("=========================================")
|
|
tr.logger.Info(fmt.Sprintf("Results: %d passed, %d failed", passed, failed))
|
|
tr.logger.Info("=========================================")
|
|
|
|
if failed > 0 {
|
|
for _, r := range tr.results {
|
|
if !r.Passed {
|
|
tr.logger.Error("FAIL", mlog.String("test", r.Name), mlog.String("detail", r.Detail))
|
|
}
|
|
}
|
|
return fmt.Errorf("%d test(s) failed", failed)
|
|
}
|
|
return nil
|
|
}
|