diff --git a/server/channels/app/remote_cluster.go b/server/channels/app/remote_cluster.go index 4f7b144b2fe..b87ba525eaf 100644 --- a/server/channels/app/remote_cluster.go +++ b/server/channels/app/remote_cluster.go @@ -8,6 +8,7 @@ import ( "encoding/base64" "fmt" "net/http" + "time" "github.com/pkg/errors" @@ -19,6 +20,34 @@ import ( "github.com/mattermost/mattermost/server/public/shared/request" ) +// pluginRemoteInitialPingDelay is how long the framework waits after +// RegisterPluginForSharedChannels returns before firing the first ping to +// the newly created or restored plugin remote. The delay gives the +// calling plugin a chance to record the returned RemoteId in its own +// state, so the synchronous OnSharedChannelsPing hook the framework +// invokes can resolve the remote. Without the delay, the first ping for +// every freshly registered SiteURL fails and the remote stays offline +// until the periodic pingLoop refreshes it (up to PingFreq, default 1 +// minute). Declared as a var, not const, so tests can shorten it. +var pluginRemoteInitialPingDelay = 5 * time.Second + +// schedulePluginRemoteInitialPing schedules a single deferred ping for a +// freshly registered or restored plugin remote. The goroutine is launched +// via Server.Go so it cannot outlive the server. The remote is re-read +// before the ping fires because the plugin may have unregistered it +// inside the delay window; pinging a soft-deleted row is harmless but +// produces a stray "ping failed" warning. +func (a *App) schedulePluginRemoteInitialPing(rcService remotecluster.RemoteClusterServiceIFace, rc *model.RemoteCluster) { + a.Srv().Go(func() { + time.Sleep(pluginRemoteInitialPingDelay) + current, err := a.Srv().Store().RemoteCluster().Get(rc.RemoteId, true) + if err != nil || current.DeleteAt != 0 { + return + } + rcService.PingNow(current) + }) +} + func (a *App) RegisterPluginForSharedChannels(rctx request.CTX, opts model.RegisterPluginOpts) (remoteID string, err error) { // When SiteURL is empty, fall back to the legacy single-remote behavior // using the "plugin_" prefix. This preserves compatibility for older plugins @@ -59,6 +88,18 @@ func (a *App) RegisterPluginForSharedChannels(rctx request.CTX, opts model.Regis if _, err = a.Srv().Store().RemoteCluster().Update(rc); err != nil { return "", err } + + // Ping the restored plugin remote so its LastPingAt is refreshed + // before sync attempts. Deferred via a goroutine (see + // schedulePluginRemoteInitialPing) so the caller has a chance + // to record the returned RemoteId before the synchronous + // OnSharedChannelsPing hook fires. Without this the restored + // remote stays offline until the next pingLoop iteration (up to + // PingFreq), causing transient sync failures. + rcService, _ := a.GetRemoteClusterService() + if rcService != nil { + a.schedulePluginRemoteInitialPing(rcService, rc) + } return rc.RemoteId, nil } @@ -86,13 +127,19 @@ func (a *App) RegisterPluginForSharedChannels(rctx request.CTX, opts model.Regis mlog.String("site_url", opts.SiteURL), ) - // Ping the plugin remote immediately if the service is running. - // If the service is not available the ping will happen once the - // service starts. This is expected since plugins start before the - // service. + // Ping the new plugin remote, deferred via a goroutine so the + // calling plugin has a chance to record the returned RemoteId + // before the synchronous OnSharedChannelsPing hook fires for the + // first time. A synchronous ping here would invoke the hook + // before the caller's return statement, the plugin would fail to + // resolve the new RemoteId, the ping would be recorded as failed, + // and the remote would stay offline until the next pingLoop + // iteration (up to PingFreq, default 1 minute). If the service is + // not yet running the ping will fire from the periodic pingLoop + // once the service starts. rcService, _ := a.GetRemoteClusterService() if rcService != nil { - rcService.PingNow(rcSaved) + a.schedulePluginRemoteInitialPing(rcService, rcSaved) } return rcSaved.RemoteId, nil diff --git a/server/channels/app/remote_cluster_test.go b/server/channels/app/remote_cluster_test.go index f4b5ec604dd..4a5637d4a1d 100644 --- a/server/channels/app/remote_cluster_test.go +++ b/server/channels/app/remote_cluster_test.go @@ -6,13 +6,39 @@ package app import ( "strings" "testing" + "time" "github.com/stretchr/testify/require" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/v8/channels/testlib" + "github.com/mattermost/mattermost/server/v8/platform/services/remotecluster" ) +// Shorten the deferred initial ping for tests so RegisterPluginForSharedChannels +// teardown does not block on a 5s goroutine. No test in this package needs the +// production headroom. The value is large enough that even a slow runner where +// RegisterPluginForSharedChannels takes a couple hundred milliseconds still +// has comfortable margin before the deferred goroutine fires. +func init() { + pluginRemoteInitialPingDelay = 500 * time.Millisecond +} + +// pingTrackingRCService wraps a real RemoteClusterServiceIFace and records the +// time of every PingNow call without forwarding it. Embedding the interface +// satisfies the other methods by delegation. +type pingTrackingRCService struct { + remotecluster.RemoteClusterServiceIFace + pings chan time.Time +} + +func (p *pingTrackingRCService) PingNow(rc *model.RemoteCluster) { + select { + case p.pings <- time.Now(): + default: + } +} + func setupRemoteCluster(tb testing.TB) *TestHelper { return SetupConfig(tb, func(cfg *model.Config) { *cfg.ConnectedWorkspacesSettings.EnableRemoteClusterService = true @@ -221,6 +247,46 @@ func TestRegisterPluginForSharedChannels(t *testing.T) { require.Equal(t, id1, id2) }) + t.Run("re-registering a soft-deleted SiteURL restores the row and pings the remote (MM-68838)", func(t *testing.T) { + pluginID := "com.test.restore-" + model.NewId() + siteURL := "nats://restore-" + model.NewId() + + // 1. Initial registration. + id1, err := th.App.RegisterPluginForSharedChannels(th.Context, model.RegisterPluginOpts{ + Displayname: "restore test plugin", + PluginID: pluginID, + CreatorID: th.BasicUser.Id, + SiteURL: siteURL, + }) + require.NoError(t, err) + + // 2. Unregister soft-deletes the row. + require.NoError(t, th.App.UnregisterPluginRemoteForSharedChannels(pluginID, id1)) + + rcDeleted, err := th.App.Srv().Store().RemoteCluster().Get(id1, true) + require.NoError(t, err) + require.NotZero(t, rcDeleted.DeleteAt, "row should be soft-deleted after unregister") + + // 3. Re-register the same SiteURL. The restore path must run. + id2, err := th.App.RegisterPluginForSharedChannels(th.Context, model.RegisterPluginOpts{ + Displayname: "restore test plugin", + PluginID: pluginID, + CreatorID: th.BasicUser.Id, + SiteURL: siteURL, + }) + require.NoError(t, err) + require.Equal(t, id1, id2, "restore path must reuse the existing remoteID") + + // 4. The row must be restored (DeleteAt cleared). PingNow is called + // inside the restore branch; the actual ping fails in unit tests + // because no plugin process is loaded to answer OnSharedChannelsPing, + // so we cannot assert on LastPingAt here. The presence of the call + // is what fixes MM-68838 (offline-for-PingFreq window on restart). + rcRestored, err := th.App.Srv().Store().RemoteCluster().Get(id2, false) + require.NoError(t, err) + require.Zero(t, rcRestored.DeleteAt, "row should be restored after re-register") + }) + t.Run("multi-remote registration returns distinct remoteIDs", func(t *testing.T) { pluginID := "com.test.multi-" + model.NewId() @@ -322,3 +388,117 @@ func TestUnregisterPluginForSharedChannelsBulk(t *testing.T) { require.NoError(t, err) require.Empty(t, remotes) } + +// TestRegisterPluginForSharedChannelsPingIsDeferred guards the race fix. +// A synchronous PingNow inside RegisterPluginForSharedChannels invoked the +// plugin's OnSharedChannelsPing hook before the calling plugin could record +// the returned RemoteId, so the very first ping always failed and the remote +// stayed offline for ~PingFreq (1 minute). The fix is to defer the initial +// ping to a goroutine. Both the new-row branch and the soft-delete-restore +// branch must defer. +func TestRegisterPluginForSharedChannelsPingIsDeferred(t *testing.T) { + mainHelper.Parallel(t) + th := setupRemoteCluster(t).InitBasic(t) + + tracker := &pingTrackingRCService{ + RemoteClusterServiceIFace: th.Server.remoteClusterService, + pings: make(chan time.Time, 8), + } + original := th.Server.remoteClusterService + th.Server.remoteClusterService = tracker + t.Cleanup(func() { th.Server.remoteClusterService = original }) + + // Generous upper bound on real wall-time variance under load: the + // production delay is 5s; init() overrides to 100ms; we wait up to + // delay + 2s for the ping to actually arrive. + const arrivalGrace = 2 * time.Second + delay := pluginRemoteInitialPingDelay + + // drain consumes any pending ping timestamps so a later sub-case does + // not see a stale one from an earlier sub-case. + drain := func(ch <-chan time.Time) { + for { + select { + case <-ch: + default: + return + } + } + } + + assertDeferred := func(t *testing.T, registerStart time.Time) { + t.Helper() + // Phase 1: no ping in the first half of the delay (proves async). + var prematurePing bool + select { + case <-tracker.pings: + prematurePing = true + case <-time.After(delay / 2): + } + require.False(t, prematurePing, "PingNow fired synchronously inside RegisterPluginForSharedChannels; the deferral is broken") + // Phase 2: a ping arrives within delay + grace, and not before delay. + var pingAt time.Time + var pingArrived bool + select { + case pingAt = <-tracker.pings: + pingArrived = true + case <-time.After(delay + arrivalGrace): + } + require.True(t, pingArrived, "expected PingNow to fire within delay + grace, but it never did") + elapsed := pingAt.Sub(registerStart) + require.GreaterOrEqual(t, elapsed, delay, + "PingNow fired %s after Register returned, before the configured delay of %s", elapsed, delay) + } + + t.Run("new-row branch defers the initial ping", func(t *testing.T) { + drain(tracker.pings) + + start := time.Now() + _, err := th.App.RegisterPluginForSharedChannels(th.Context, model.RegisterPluginOpts{ + Displayname: "deferred ping plugin", + PluginID: "com.test.deferred-" + model.NewId(), + CreatorID: th.BasicUser.Id, + SiteURL: "nats://deferred-" + model.NewId(), + }) + require.NoError(t, err) + assertDeferred(t, start) + }) + + t.Run("soft-delete restore branch defers the ping (MM-68838)", func(t *testing.T) { + drain(tracker.pings) + + pluginID := "com.test.restore-defer-" + model.NewId() + siteURL := "nats://restore-defer-" + model.NewId() + + // Initial register to create the row; consume its deferred ping. + id1, err := th.App.RegisterPluginForSharedChannels(th.Context, model.RegisterPluginOpts{ + Displayname: "restore defer plugin", + PluginID: pluginID, + CreatorID: th.BasicUser.Id, + SiteURL: siteURL, + }) + require.NoError(t, err) + var initialPingArrived bool + select { + case <-tracker.pings: + initialPingArrived = true + case <-time.After(delay + arrivalGrace): + } + require.True(t, initialPingArrived, "initial register's deferred ping never arrived") + + // Unregister soft-deletes the row. + require.NoError(t, th.App.UnregisterPluginRemoteForSharedChannels(pluginID, id1)) + drain(tracker.pings) + + // Re-register: the restore branch must also defer. + start := time.Now() + _, err = th.App.RegisterPluginForSharedChannels(th.Context, model.RegisterPluginOpts{ + Displayname: "restore defer plugin", + PluginID: pluginID, + CreatorID: th.BasicUser.Id, + SiteURL: siteURL, + }) + require.NoError(t, err) + assertDeferred(t, start) + }) +} diff --git a/server/public/plugin/client_rpc.go b/server/public/plugin/client_rpc.go index c0c03f93df7..196b43ed2b3 100644 --- a/server/public/plugin/client_rpc.go +++ b/server/public/plugin/client_rpc.go @@ -1260,6 +1260,7 @@ func (s *apiRPCServer) ReceiveSharedChannelAttachmentSyncMsg(args *Z_ReceiveShar defer dataReader.Close() returns.A, returns.B = hook.ReceiveSharedChannelAttachmentSyncMsg(args.A, args.B, args.C, dataReader) + returns.B = encodableError(returns.B) return nil } diff --git a/server/public/plugin/client_rpc_test.go b/server/public/plugin/client_rpc_test.go new file mode 100644 index 00000000000..179b392e0f8 --- /dev/null +++ b/server/public/plugin/client_rpc_test.go @@ -0,0 +1,46 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package plugin + +import ( + "bytes" + "encoding/gob" + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestReceiveSharedChannelAttachmentSyncMsgReturns_GobRoundTrip pins the fix +// for the gob-encoding bug in apiRPCServer.ReceiveSharedChannelAttachmentSyncMsg. +// The hook may return errors wrapped with fmt.Errorf("...%w", err), producing +// values of the unexported type *fmt.wrapError that gob refuses to encode. +// The RPC server must run the error through encodableError before assigning +// it to the returns struct. Without that, the RPC connection breaks and +// every subsequent plugin to server call returns zero values. +func TestReceiveSharedChannelAttachmentSyncMsgReturns_GobRoundTrip(t *testing.T) { + wrapped := fmt.Errorf("attachment sync failed: %w", errors.New("upstream boom")) + + t.Run("raw wrapped error fails to gob-encode (reproduces the bug)", func(t *testing.T) { + returns := Z_ReceiveSharedChannelAttachmentSyncMsgReturns{B: wrapped} + + var buf bytes.Buffer + err := gob.NewEncoder(&buf).Encode(&returns) + require.Error(t, err, "raw *fmt.wrapError must not be gob-encodable; if this assertion ever fails the bug guarded by encodableError no longer exists") + require.Contains(t, err.Error(), "fmt.wrapError") + }) + + t.Run("encodableError-wrapped error round-trips through gob", func(t *testing.T) { + returns := Z_ReceiveSharedChannelAttachmentSyncMsgReturns{B: encodableError(wrapped)} + + var buf bytes.Buffer + require.NoError(t, gob.NewEncoder(&buf).Encode(&returns)) + + var decoded Z_ReceiveSharedChannelAttachmentSyncMsgReturns + require.NoError(t, gob.NewDecoder(&buf).Decode(&decoded)) + require.Error(t, decoded.B) + require.Equal(t, wrapped.Error(), decoded.B.Error()) + }) +}