MM-68838: Ping a restored plugin remote immediately on re-register (#36592)

* MM-68838: ping restored plugin remote immediately on re-register

  RegisterPluginForSharedChannels' restore branch updated the row but did
  not call PingNow, leaving the restored remote offline until the next
  pingLoop tick (up to PingFreq, default 1 minute). The new-connection
  branch already calls PingNow; the restore branch now mirrors it so
  sync attempts immediately after a plugin restart no longer fail with
  "offline remote cluster".

* MM-68838: gob-encode error returns in apiRPCServer.ReceiveSharedChannelAttachmentSyncMsg

  The apiRPCServer wrapper for ReceiveSharedChannelAttachmentSyncMsg
  assigned the hook's error return directly to the gob-encoded response
  struct. When the framework's App.ReceiveSharedChannelAttachmentSyncMsg
  returned an error wrapped with %w (*fmt.wrapError, an unexported type),
  gob refused to encode it and the RPC server broke the connection with
  "type not registered for interface: fmt.wrapError".

  Every subsequent plugin/server RPC call then returned the zero-value
  response struct, causing plugins that dereferenced the nil returns to
  crash.

  Apply the existing encodableError() helper so the returned error
  becomes a gob-safe ErrorString, matching every other apiRPCServer
  method in this file.

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

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)
+	})
+}

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
 }
 

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())
+	})
+}