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
226 lines
7.3 KiB
Go
226 lines
7.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/mattermost/mattermost/server/public/model"
|
|
"github.com/mattermost/mattermost/server/public/shared/mlog"
|
|
)
|
|
|
|
func (tr *TestRunner) runPostTests(ctx context.Context) error {
|
|
tr.logger.Info("=== Post Sync Tests ===")
|
|
|
|
channelA, channelB, err := tr.setupSharedChannel(ctx, "post-test")
|
|
if err != nil {
|
|
tr.fail("post/setup", err.Error())
|
|
return err
|
|
}
|
|
|
|
// Create a user and add to channel
|
|
userID, err := tr.createTestUser(ctx, "postuser1")
|
|
if err != nil {
|
|
tr.fail("post/create-user", err.Error())
|
|
return err
|
|
}
|
|
_, _, err = tr.clientA.AddChannelMember(ctx, channelA, userID)
|
|
if err != nil {
|
|
tr.fail("post/add-user-to-channel", err.Error())
|
|
return err
|
|
}
|
|
|
|
// Wait for user to sync before posting
|
|
if err := tr.waitFor(ctx, 15*time.Second, func() bool {
|
|
return tr.verifyMemberOnB(ctx, channelB, "postuser1")
|
|
}); err != nil {
|
|
tr.fail("post/user-sync", "postuser1 did not sync to Server B before posting")
|
|
return err
|
|
}
|
|
|
|
// ── Test: Create post and verify sync ───────────────────
|
|
tr.logger.Info("Creating post on Server A...")
|
|
postMessage := "Hello from Server A! " + model.NewId()[:8]
|
|
post, _, err := tr.clientA.CreatePost(ctx, &model.Post{
|
|
ChannelId: channelA,
|
|
Message: postMessage,
|
|
})
|
|
if err != nil {
|
|
tr.fail("post/create", err.Error())
|
|
return err
|
|
}
|
|
|
|
testName := "post/sync-create"
|
|
err = tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
return tr.findPostOnB(ctx, channelB, postMessage) != ""
|
|
})
|
|
if err != nil {
|
|
tr.fail(testName, "post did not sync to Server B")
|
|
} else {
|
|
tr.pass(testName)
|
|
}
|
|
|
|
// ── Test: Edit post and verify sync ─────────────────────
|
|
tr.logger.Info("Editing post on Server A...")
|
|
editedMessage := "Edited: " + postMessage
|
|
post.Message = editedMessage
|
|
_, _, err = tr.clientA.UpdatePost(ctx, post.Id, post)
|
|
if err != nil {
|
|
tr.fail("post/edit", err.Error())
|
|
} else {
|
|
testName := "post/sync-edit"
|
|
err := tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
return tr.findPostOnB(ctx, channelB, editedMessage) != ""
|
|
})
|
|
if err != nil {
|
|
tr.fail(testName, "post edit did not sync to Server B")
|
|
} else {
|
|
tr.pass(testName)
|
|
}
|
|
}
|
|
|
|
// ── Test: Delete post and verify sync ───────────────────
|
|
tr.logger.Info("Deleting post on Server A...")
|
|
_, err = tr.clientA.DeletePost(ctx, post.Id)
|
|
if err != nil {
|
|
tr.fail("post/delete", err.Error())
|
|
} else {
|
|
testName := "post/sync-delete"
|
|
err := tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
return tr.findPostOnB(ctx, channelB, editedMessage) == ""
|
|
})
|
|
if err != nil {
|
|
tr.fail(testName, "post delete did not sync to Server B")
|
|
} else {
|
|
tr.pass(testName)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// runOfflineRecoveryTests verifies that messages posted while a remote is briefly
|
|
// offline (for less than the 5-minute IsOnline() threshold) are eventually synced
|
|
// once the remote comes back online — even though no new organic message triggers a
|
|
// sync and no connection-state flip occurs via the IsOnline() path.
|
|
//
|
|
// This exercises the recovery path added for MM-69792: a failed sync sets a flag in
|
|
// the remote cluster service, and the next successful ping fires a targeted resync.
|
|
//
|
|
// The test requires managing the Server B lifecycle, so it is skipped in unmanaged mode.
|
|
func (tr *TestRunner) runOfflineRecoveryTests(ctx context.Context) error {
|
|
tr.logger.Info("=== Offline Recovery Tests ===")
|
|
|
|
if !tr.cfg.Manage || tr.mgr == nil {
|
|
tr.logger.Info("Skipping offline-recovery test (requires managed mode to stop/restart Server B)")
|
|
return nil
|
|
}
|
|
|
|
channelA, channelB, err := tr.setupSharedChannel(ctx, "offline-recovery-test")
|
|
if err != nil {
|
|
tr.fail("post/offline-recovery-setup", err.Error())
|
|
return err
|
|
}
|
|
|
|
// Create a user and add to the channel so posts have an author that syncs.
|
|
userID, err := tr.createTestUser(ctx, "offlineuser1")
|
|
if err != nil {
|
|
tr.fail("post/offline-recovery-create-user", err.Error())
|
|
return err
|
|
}
|
|
if _, _, err = tr.clientA.AddChannelMember(ctx, channelA, userID); err != nil {
|
|
tr.fail("post/offline-recovery-add-user", err.Error())
|
|
return err
|
|
}
|
|
if err := tr.waitFor(ctx, 15*time.Second, func() bool {
|
|
return tr.verifyMemberOnB(ctx, channelB, "offlineuser1")
|
|
}); err != nil {
|
|
tr.fail("post/offline-recovery-user-sync", "offlineuser1 did not sync to Server B before posting")
|
|
return err
|
|
}
|
|
|
|
testName := "post/offline-recovery"
|
|
|
|
// Step 1: baseline sync works.
|
|
m1 := "M1 baseline " + model.NewId()[:8]
|
|
tr.logger.Info("Posting M1 (baseline) on Server A...")
|
|
if _, _, err := tr.clientA.CreatePost(ctx, &model.Post{ChannelId: channelA, Message: m1}); err != nil {
|
|
tr.fail(testName, "failed to create baseline post M1: "+err.Error())
|
|
return nil
|
|
}
|
|
if err := tr.waitFor(ctx, 30*time.Second, func() bool {
|
|
return tr.findPostOnB(ctx, channelB, m1) != ""
|
|
}); err != nil {
|
|
tr.fail(testName, "baseline post M1 did not sync to Server B")
|
|
return nil
|
|
}
|
|
tr.logger.Info("Baseline post M1 synced to Server B")
|
|
|
|
// Step 2: stop Server B.
|
|
tr.mgr.StopServerB()
|
|
|
|
// Step 3: post M2 and M3 while Server B is offline.
|
|
m2 := "M2 during-outage " + model.NewId()[:8]
|
|
m3 := "M3 during-outage " + model.NewId()[:8]
|
|
tr.logger.Info("Posting M2 and M3 on Server A while Server B is offline...")
|
|
for _, msg := range []string{m2, m3} {
|
|
if _, _, err := tr.clientA.CreatePost(ctx, &model.Post{ChannelId: channelA, Message: msg}); err != nil {
|
|
tr.fail(testName, "failed to create post during outage: "+err.Error())
|
|
// Best effort: bring Server B back before returning so later tests can run.
|
|
if startErr := tr.mgr.StartServerB(ctx); startErr != nil {
|
|
tr.logger.Error("Failed to restart Server B", mlog.Err(startErr))
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// Step 4: wait long enough for send retries to exhaust, but short enough that
|
|
// IsOnline() has not flipped (LastPingAt stays within the 5-minute window).
|
|
tr.logger.Info("Waiting 90s for send retries to exhaust while Server B stays offline...")
|
|
if err := interruptibleSleep(ctx, 90*time.Second); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Step 5: restart Server B.
|
|
if err := tr.mgr.StartServerB(ctx); err != nil {
|
|
tr.fail(testName, "failed to restart Server B: "+err.Error())
|
|
return err
|
|
}
|
|
|
|
// Step 6: M2 and M3 must eventually appear on Server B via the recovery path.
|
|
tr.logger.Info("Waiting up to 120s for M2 and M3 to sync to Server B after recovery...")
|
|
if err := tr.waitFor(ctx, 120*time.Second, func() bool {
|
|
return tr.findPostOnB(ctx, channelB, m2) != "" && tr.findPostOnB(ctx, channelB, m3) != ""
|
|
}); err != nil {
|
|
tr.fail(testName, "M2 and/or M3 did not sync to Server B after recovery")
|
|
return nil
|
|
}
|
|
|
|
tr.pass(testName)
|
|
return nil
|
|
}
|
|
|
|
// interruptibleSleep sleeps for d, returning early with an error if ctx is cancelled.
|
|
func interruptibleSleep(ctx context.Context, d time.Duration) error {
|
|
select {
|
|
case <-time.After(d):
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
// findPostOnB searches for a post with the given message in a channel on Server B.
|
|
// Returns the post ID if found, empty string otherwise.
|
|
func (tr *TestRunner) findPostOnB(ctx context.Context, channelB, message string) string {
|
|
postList, _, err := tr.clientB.GetPostsForChannel(ctx, channelB, 0, 100, "", false, false)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
for _, post := range postList.Posts {
|
|
if post.Message == message {
|
|
return post.Id
|
|
}
|
|
}
|
|
return ""
|
|
}
|