diff --git a/go/chat/deliverer.go b/go/chat/deliverer.go index 29de3216a823..bb8db6e3536c 100644 --- a/go/chat/deliverer.go +++ b/go/chat/deliverer.go @@ -355,9 +355,7 @@ func (s *Deliverer) doNotRetryFailure(ctx context.Context, obr chat1.OutboxRecor return 0, err, false case net.Error: s.Debug(ctx, "doNotRetryFailure: generic net error, reconnecting to the server: %s(%T)", berr, berr) - if _, rerr := s.serverConn.Reconnect(ctx); rerr != nil { - s.Debug(ctx, "doNotRetryFailure: failed to reconnect: %s", rerr) - } + s.serverConn.Reconnect(ctx) return chat1.OutboxErrorType_OFFLINE, err, !berr.Temporary() //nolint } if errors.Is(err, ErrChatServerTimeout) || errors.Is(err, ErrDuplicateConnection) || diff --git a/go/chat/server.go b/go/chat/server.go index 9e3f9ac1c931..7b9c1863eb7e 100644 --- a/go/chat/server.go +++ b/go/chat/server.go @@ -133,11 +133,7 @@ func (h *Server) handleOfflineError(ctx context.Context, err error, case OfflineErrorKindOfflineReconnect: // Reconnect Gregor if we think we are offline (and told to reconnect) h.Debug(ctx, "handleOfflineError: reconnecting to gregor") - if _, err := h.serverConn.Reconnect(ctx); err != nil { - h.Debug(ctx, "handleOfflineError: error reconnecting: %s", err) - } else { - h.Debug(ctx, "handleOfflineError: success reconnecting") - } + h.serverConn.Reconnect(ctx) default: // Nothing to do for other errors. } diff --git a/go/chat/server_test.go b/go/chat/server_test.go index 4094c15b2efe..4235d62b66a4 100644 --- a/go/chat/server_test.go +++ b/go/chat/server_test.go @@ -106,9 +106,7 @@ func (g *gregorTestConnection) GetClient() chat1.RemoteInterface { return chat1.RemoteClient{Cli: g.cli} } -func (g *gregorTestConnection) Reconnect(ctx context.Context) (bool, error) { - return false, nil -} +func (g *gregorTestConnection) Reconnect(ctx context.Context) {} func (g *gregorTestConnection) OnConnect(ctx context.Context, _ *rpc.Connection, cli rpc.GenericClient, srv *rpc.Server, diff --git a/go/chat/sync.go b/go/chat/sync.go index 880486eab197..4439a14d9c4e 100644 --- a/go/chat/sync.go +++ b/go/chat/sync.go @@ -177,6 +177,13 @@ func (s *Syncer) Connected(ctx context.Context, cli chat1.RemoteInterface, uid g ctx = globals.CtxAddLogTags(ctx, s.G()) defer s.Trace(ctx, &err, "Connected")() s.Lock() + // The caller cancels ctx when the connection it was made for shuts + // down, before it calls Disconnected, so a Connected that sees the cancel + // here must not mark the syncer connected after that Disconnected. + if err := ctx.Err(); err != nil { + s.Unlock() + return err + } s.isConnected = true // Let the Offlinables know that we are back online for _, o := range s.offlinables { diff --git a/go/chat/sync_test.go b/go/chat/sync_test.go index 0475c81d28ad..4ac09af5cef6 100644 --- a/go/chat/sync_test.go +++ b/go/chat/sync_test.go @@ -9,6 +9,7 @@ import ( "github.com/keybase/client/go/chat/storage" "github.com/keybase/client/go/chat/types" "github.com/keybase/client/go/chat/utils" + "github.com/keybase/client/go/externalstest" "github.com/keybase/client/go/kbtest" "github.com/keybase/client/go/libkb" "github.com/keybase/client/go/protocol/chat1" @@ -438,6 +439,20 @@ func TestSyncerMembersTypeChanged(t *testing.T) { } } +// Connected with a ctx its connection's Shutdown has already cancelled must +// not mark the syncer connected: the Disconnected that follows the cancel may +// already have run. +func TestSyncerConnectedAfterCancelIsIgnored(t *testing.T) { + tc := externalstest.SetupTest(t, "syncer-connected-cancel", 0) + defer tc.Cleanup() + syncer := NewSyncer(globals.NewContext(tc.G, &globals.ChatContext{})) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := syncer.Connected(ctx, nil, gregor1.UID(make([]byte, 16)), &chat1.SyncChatRes{}) + require.False(t, syncer.IsConnected(context.Background())) + require.ErrorIs(t, err, context.Canceled) +} + func TestSyncerAppState(t *testing.T) { ctx, world, ri2, _, sender, list := setupTest(t, 1) defer world.Cleanup() diff --git a/go/chat/types/interfaces.go b/go/chat/types/interfaces.go index 5212f879fab5..278e483d92ac 100644 --- a/go/chat/types/interfaces.go +++ b/go/chat/types/interfaces.go @@ -697,7 +697,8 @@ type ( ) type ServerConnection interface { - Reconnect(context.Context) (bool, error) + // Reconnect reconnects to the server without waiting for it. + Reconnect(context.Context) GetClient() chat1.RemoteInterface } diff --git a/go/kbhttp/manager/manager.go b/go/kbhttp/manager/manager.go index 50b14ae21425..6a407daf09c9 100644 --- a/go/kbhttp/manager/manager.go +++ b/go/kbhttp/manager/manager.go @@ -118,10 +118,13 @@ func (r *Srv) monitorAppState() { for { <-r.G().MobileAppState.NextUpdate(state) state = r.G().MobileAppState.State() + // Only BACKGROUND stops the server. INACTIVE is transient (control + // center, the app switcher, an incoming call), as gregor also treats it. switch state { - case keybase1.MobileAppState_FOREGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE: + case keybase1.MobileAppState_FOREGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_INACTIVE: r.startHTTPSrv() - case keybase1.MobileAppState_BACKGROUND, keybase1.MobileAppState_INACTIVE: + case keybase1.MobileAppState_BACKGROUND: r.httpSrv.Stop() } } diff --git a/go/kbhttp/manager/manager_test.go b/go/kbhttp/manager/manager_test.go new file mode 100644 index 000000000000..9c4e9fb06c80 --- /dev/null +++ b/go/kbhttp/manager/manager_test.go @@ -0,0 +1,31 @@ +package manager + +import ( + "testing" + "time" + + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +// Only BACKGROUND stops the server; INACTIVE keeps it up, and starts it when +// coming back from BACKGROUND. +func TestSrvAppState(t *testing.T) { + tc := libkb.SetupTest(t, "httpsrv", 1) + defer tc.Cleanup() + srv := NewSrv(tc.G) + require.True(t, srv.Active()) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + require.Never(t, func() bool { return !srv.Active() }, 200*time.Millisecond, time.Millisecond, + "INACTIVE stopped the server") + + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + require.Eventually(t, func() bool { return !srv.Active() }, 10*time.Second, time.Millisecond, + "BACKGROUND did not stop the server") + + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + require.Eventually(t, srv.Active, 10*time.Second, time.Millisecond, + "INACTIVE after BACKGROUND did not start the server") +} diff --git a/go/kbtest/chat.go b/go/kbtest/chat.go index 734c915d9b02..d743bc691d69 100644 --- a/go/kbtest/chat.go +++ b/go/kbtest/chat.go @@ -398,9 +398,7 @@ func (m ChatRemoteMockServerConnection) GetClient() chat1.RemoteInterface { return m.mock } -func (m ChatRemoteMockServerConnection) Reconnect(ctx context.Context) (bool, error) { - return false, nil -} +func (m ChatRemoteMockServerConnection) Reconnect(ctx context.Context) {} type ChatRemoteMock struct { world *ChatMockWorld diff --git a/go/service/gregor.go b/go/service/gregor.go index 5fe2ba3edbc9..aea055614204 100644 --- a/go/service/gregor.go +++ b/go/service/gregor.go @@ -180,6 +180,7 @@ type replayThreadArg struct { cli gregor1.IncomingInterface t time.Time ctx context.Context + uid gregor1.UID } type gregorHandler struct { @@ -196,14 +197,24 @@ type gregorHandler struct { gregorCli *grclient.Client firehoseHandlers []libkb.GregorFirehoseHandler - badger *badges.Badger + badger gregorBadger reachability *reachability chatLog utils.DebugLabeler + // connGate decides when to connect and disconnect, and runs (runIfLive) + // the steps OnConnect applies after syncing that can't be undone (badge + // pushes), so none of them lands after a Shutdown for the connection it + // came from. + connGate *gregorConnGate + // This mutex protects the con object connMutex sync.Mutex conn *rpc.Connection - uri *rpc.FMPURI + // connCtx lives as long as conn: Shutdown cancels it under connMutex. + // OnConnect runs under a ctx derived from it, and the connection's ping + // loop and push state debouncer exit when it is done. + connCtx context.Context + connCancel context.CancelFunc // connectHappened will be closed after gregor connection established connectHappened chan struct{} @@ -221,17 +232,26 @@ type gregorHandler struct { // a pushState call to firehose handlers pushStateFilter func(m gregor.Message) bool - shutdownCh chan struct{} broadcastCh chan gregor1.Message replayCh chan replayThreadArg pushStateCh chan struct{} forcePingCh chan struct{} // Testing - testingEvents *testingEvents + testingEvents *testingEvents + // authParamsForTest, if set, replaces authParams in OnConnect. + authParamsForTest func(ctx context.Context) (gregor1.UID, gregor1.DeviceID, gregor1.SessionToken, *libkb.NIST, error) transportForTesting *connTransport } +// gregorBadger is the part of the badger gregor pushes to. +type gregorBadger interface { + PushState(ctx context.Context, state gregor.State) + PushChatFullUpdate(ctx context.Context, update chat1.UnreadUpdateFull) +} + +var _ gregorBadger = (*badges.Badger)(nil) + var ( _ libkb.GregorState = (*gregorHandler)(nil) _ libkb.GregorListener = (*gregorHandler)(nil) @@ -250,6 +270,8 @@ func newGregorHandler(g *globals.Context) *gregorHandler { pushStateCh: make(chan struct{}, 100), forcePingCh: make(chan struct{}, 5), } + eg := g.ExternalG() + gh.connGate = newGregorConnGate(eg.MobileAppState, eg.DesktopAppState, gh, gh.chatLog.Debug, gh.forcePing) return gh } @@ -258,66 +280,15 @@ func (g *gregorHandler) Init() { // Start broadcast handler goroutine go g.broadcastMessageHandler() // Start the app state monitor thread - go g.monitorAppState() + g.connGate.start() + g.G().PushShutdownHook(func(libkb.MetaContext) error { + g.connGate.stop() + return nil + }) // Start replay thread go g.syncReplayThread() } -const ( - monitorConnect int = iota - monitorDisconnect - monitorNoop -) - -func (g *gregorHandler) monitorAppState() { - ctx := libkb.WithLogTag(context.Background(), "GRGRMON") - // Wait for state updates and react accordingly - state := keybase1.MobileAppState_FOREGROUND - suspended := false - for { - monitorAction := monitorNoop - select { - case <-g.G().MobileAppState.NextUpdate(state): - state = g.G().MobileAppState.State() - switch state { - case keybase1.MobileAppState_FOREGROUND: - g.forcePing(ctx) - monitorAction = monitorConnect - case keybase1.MobileAppState_BACKGROUNDACTIVE: - monitorAction = monitorConnect - case keybase1.MobileAppState_BACKGROUND, keybase1.MobileAppState_INACTIVE: - monitorAction = monitorDisconnect - } - case <-g.G().DesktopAppState.NextSuspendUpdate(suspended): - suspended = g.G().DesktopAppState.Suspended() - if !suspended { - monitorAction = monitorConnect - g.chatLog.Debug(ctx, "resumed, connecting") - } else { - g.chatLog.Debug(ctx, "suspended, disconnecting") - monitorAction = monitorDisconnect - } - } - switch monitorAction { - case monitorConnect: - // Make sure the URI is set before attempting this (possible it isn't in a race) - if g.uri != nil { - g.chatLog.Debug(ctx, "foregrounded, reconnecting") - if err := g.Connect(g.uri); err != nil { - g.chatLog.Debug(ctx, "error reconnecting: %s", err) - } - } - case monitorDisconnect: - g.chatLog.Debug(ctx, "backgrounded, shutting down connection") - g.Shutdown(ctx) - } - } -} - -func (g *gregorHandler) GetURI() *rpc.FMPURI { - return g.uri -} - func (g *gregorHandler) GetIncomingClient() gregor1.IncomingInterface { cli := g.getRPCCli() if g.IsShutdown() || cli == nil { @@ -371,6 +342,11 @@ func (g *gregorHandler) shutdownGregorClient(ctx context.Context) { } } +// resetGregorClient installs a new client for uid unless ctx is cancelled. +// OnConnect passes the ctx it derives for its connection, which Shutdown +// cancels under connMutex; checking and installing under connMutex too means +// Reset, which drops the client after its Shutdown, never runs between the +// two. func (g *gregorHandler) resetGregorClient(ctx context.Context, uid gregor1.UID, deviceID gregor1.DeviceID) (gcli *grclient.Client, err error) { defer g.chatLog.Trace(ctx, &err, "resetGregorClient")() // Create client object if we are logged in @@ -386,6 +362,14 @@ func (g *gregorHandler) resetGregorClient(ctx context.Context, uid gregor1.UID, g.Debug(ctx, "restore local state failed: %s", err) } } + g.connMutex.Lock() + defer g.connMutex.Unlock() + if ctx.Err() != nil { + if gcli != nil { + gcli.Stop() + } + return nil, chat.ErrDuplicateConnection + } g.gregorCliMu.Lock() gcliOld := g.gregorCli g.gregorCli = gcli @@ -437,7 +421,14 @@ func (g *gregorHandler) setReachability(r *reachability) { g.reachability = r } -func (g *gregorHandler) Connect(uri *rpc.FMPURI) (err error) { +// ConnectFresh connects to uri, resetting an existing connection first so it +// authenticates again. In BACKGROUND or while the desktop is suspended, it +// connects once that ends. +func (g *gregorHandler) ConnectFresh(uri *rpc.FMPURI) error { + return g.connGate.connect(libkb.WithLogTag(context.Background(), "GRGRCONN"), uri, true) +} + +func (g *gregorHandler) connectNow(uri *rpc.FMPURI) (err error) { ctx := libkb.WithLogTag(context.Background(), "GRGRCONN") defer g.chatLog.Trace(ctx, &err, "Connect")() @@ -452,18 +443,35 @@ func (g *gregorHandler) Connect(uri *rpc.FMPURI) (err error) { g.connectHappened = make(chan struct{}) }() - // In case we need to interrupt auth'ing or the ping loop, - // set up this channel. - g.shutdownCh = make(chan struct{}) - g.uri = uri - go g.pushStateNewDataDebouncer(g.shutdownCh) + connCtx, connCancel := context.WithCancel(context.Background()) + handler := &gregorConnHandler{gregorHandler: g, connCtx: connCtx} + var conn *rpc.Connection if uri.UseTLS() { - err = g.connectTLS(ctx) + conn, err = g.connectTLS(ctx, uri, handler) + if err != nil { + connCancel() + return err + } } else { - err = g.connectNoTLS(ctx) + conn = g.connectNoTLS(ctx, uri, handler) } + g.conn = conn + g.connCtx, g.connCancel = connCtx, connCancel - return err + // The client we get here will reconnect to gregord on disconnect if necessary. + // We should grab it here instead of in OnConnect, since the connection is not + // fully established in OnConnect. Anything that wants to make calls outside + // of OnConnect should use g.cli, everything else should the client that is + // a parameter to OnConnect + g.cli = WrapGenericClientWithTimeout(conn.GetClient(), GregorRequestTimeout, + chat.ErrChatServerTimeout) + g.pingCli = conn.GetClient() // Don't want this to have a timeout from here + + // Start up ping loop to keep the connection to gregord alive, and to kick + // off the reconnect logic in the RPC library + go g.pingLoop(ctx, g.connCtx) + go g.pushStateNewDataDebouncer(g.connCtx.Done()) + return nil } func (g *gregorHandler) HandlerName() string { @@ -536,7 +544,7 @@ func (g *gregorHandler) iterateOverFirehoseHandlers(f func(h libkb.GregorFirehos g.firehoseHandlers = freshHandlers } -func (g *gregorHandler) pushStateNewDataDebouncer(shutdownCh chan struct{}) { +func (g *gregorHandler) pushStateNewDataDebouncer(done <-chan struct{}) { shouldSend := false var lastTime time.Time dur := time.Second @@ -556,7 +564,7 @@ func (g *gregorHandler) pushStateNewDataDebouncer(shutdownCh chan struct{}) { } case <-time.After(dur): trigger() - case <-shutdownCh: + case <-done: return } } @@ -655,6 +663,10 @@ func (g *gregorHandler) IsShutdown() bool { return g.conn == nil } +func (g *gregorHandler) connExists() bool { + return !g.IsShutdown() +} + func (g *gregorHandler) IsConnected() bool { g.connMutex.Lock() defer g.connMutex.Unlock() @@ -665,6 +677,16 @@ func (g *gregorHandler) syncReplayThread() { for rarg := range g.replayCh { var trr testingReplayRes now := time.Now() + // A logout can land after serverSync queued this, and the replay's + // ctx is not cancelled by it; don't run the handlers for a user who + // is no longer logged in. + if !rarg.uid.Eq(g.currentUID()) { + g.Debug(rarg.ctx, "serverSync: replayThread: skipping, %s is no longer logged in", rarg.uid) + if g.testingEvents != nil { + g.testingEvents.replayThreadCh <- trr + } + continue + } g.Debug(rarg.ctx, "serverSync: starting replay thread") replayedMsgs, err := g.replayInBandMessages(rarg.ctx, rarg.cli, rarg.t, nil) if err != nil { @@ -714,6 +736,7 @@ func (g *gregorHandler) serverSync(ctx context.Context, cli: cli, t: t, ctx: globals.BackgroundChatCtx(ctx, g.G()), + uid: gregor1.UID(gcli.User.Bytes()), } g.pushState(keybase1.PushReason_RECONNECTED) @@ -768,25 +791,42 @@ func (g *gregorHandler) notificationParams(ctx context.Context, gcli *grclient.C return t } +// onConnectCtx returns the ctx OnConnect runs under, or ErrDuplicateConnection +// if conn is not the current connection. The rpc library's own ctx is not +// enough: it cancels only the reconnect loop running when the connection is +// shut down, and any later call on that connection starts a new loop, and so +// a new OnConnect, under a ctx nothing cancels. The returned ctx is cancelled +// by conn's Shutdown, synchronously under connMutex, as well as by the rpc +// library. The rpc library's ctx carries no values, so none are lost. +func (g *gregorHandler) onConnectCtx(ctx context.Context, conn *rpc.Connection) (context.Context, context.CancelFunc, error) { + g.connMutex.Lock() + defer g.connMutex.Unlock() + if conn == nil || conn != g.conn { + return nil, nil, chat.ErrDuplicateConnection + } + res, cancel := context.WithCancel(g.connCtx) + stop := context.AfterFunc(ctx, cancel) + return res, func() { + stop() + cancel() + }, nil +} + // OnConnect is called by the rpc library to indicate we have connected to // gregord -func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, +func (g *gregorHandler) OnConnect(rpcCtx context.Context, conn *rpc.Connection, cli rpc.GenericClient, srv *rpc.Server, ) (err error) { + ctx, cancel, err := g.onConnectCtx(rpcCtx, conn) + if err != nil { + g.chatLog.Debug(libkb.WithLogTag(rpcCtx, "GRGRONCONN"), "aborting, not the current connection") + return err + } + defer cancel() ctx = libkb.WithLogTag(ctx, "GRGRONCONN") defer g.chatLog.Trace(ctx, &err, "OnConnect")() - // If we get a random OnConnect on some other connection that is not g.conn, then - // just reject it. - g.connMutex.Lock() - if conn != g.conn { - g.connMutex.Unlock() - g.chatLog.Debug(ctx, "aborting on dup connection") - return chat.ErrDuplicateConnection - } - g.connMutex.Unlock() - g.chatLog.Debug(ctx, "connected") timeoutCli := WrapGenericClientWithTimeout(cli, GregorRequestTimeout, chat.ErrChatServerTimeout) chatCli := chat1.RemoteClient{Cli: chat.NewRemoteClient(g.G(), cli)} @@ -794,13 +834,18 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, return fmt.Errorf("error registering protocol: %s", err) } - uid, deviceID, token, nist, err := g.authParams(ctx) + authParams := g.authParams + if g.authParamsForTest != nil { + authParams = g.authParamsForTest + } + uid, deviceID, token, nist, err := authParams(ctx) if err != nil { return err } gcli, err := g.resetGregorClient(ctx, uid, deviceID) if err != nil { - return fmt.Errorf("failed to get gregor client: %s", err) + // %w keeps ErrDuplicateConnection visible to ShouldRetryOnConnect. + return fmt.Errorf("failed to get gregor client: %w", err) } iboxVers := g.inboxParams(ctx, uid) latestCtime := g.notificationParams(ctx, gcli) @@ -810,6 +855,12 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, var identBreaks []keybase1.TLFIdentifyFailure ctx = globals.ChatCtx(ctx, g.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, &identBreaks, chat.NewCachingIdentifyNotifier(g.G())) + // Every connect sets the gate's uri before connecting and a logout cancels + // ctx as it clears it, so the uri is set while ctx is live. + var uri *rpc.FMPURI + if !g.connGate.runIfLive(ctx, func(u *rpc.FMPURI) { uri = u }) { + return chat.ErrDuplicateConnection + } g.chatLog.Debug(ctx, "OnConnect begin") syncAllRes, err := chatCli.SyncAll(ctx, chat1.SyncAllArg{ Uid: uid, @@ -819,7 +870,7 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, Ctime: latestCtime, Fresh: g.isFirstConnect(), ProtVers: chat1.SyncAllProtVers_V1, - HostName: g.GetURI().Host, + HostName: uri.Host, SummarizeMaxMsgs: true, ParticipantsMode: chat1.InboxParticipantsMode_SKIP_TEAMS, }) @@ -841,6 +892,16 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, return fmt.Errorf("error authenticating: %s", err) } + return g.onConnectSynced(ctx, chatCli, timeoutCli, uid, gcli, syncAllRes) +} + +// onConnectSynced applies a SyncAll result for OnConnect's connection. A +// logout or reconnect can shut the connection down at any point, so each +// step applies only while ctx is live, and OnConnect then fails with +// ErrDuplicateConnection. +func (g *gregorHandler) onConnectSynced(ctx context.Context, chatCli chat1.RemoteInterface, + timeoutCli rpc.GenericClient, uid gregor1.UID, gcli *grclient.Client, syncAllRes chat1.SyncAllResult, +) error { // Update badging for chat. // This happens before Syncer.Connected for a reason. // If the new inbox version (e.g. 8) were committed to disk and then the @@ -848,40 +909,56 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, // badging update (7->8) then on reconnect an incomplete chat badge update (8->9) // could be received. // See: https://github.com/keybase/client/pull/12651 - if g.badger != nil { - g.badger.PushChatFullUpdate(ctx, syncAllRes.Badge) + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { + if g.badger != nil { + g.badger.PushChatFullUpdate(ctx, syncAllRes.Badge) + } + }) { + return chat.ErrDuplicateConnection } // Sync chat data using a Syncer object // This commits the new inbox version to persistent storage. - if err := g.G().Syncer.Connected(ctx, chatCli, uid, &syncAllRes.Chat); err != nil { + // It runs outside the gate, which holds off every connect and disconnect, + // to keep the gate's holds short: it writes storage and notifies. The + // Syncer ignores a cancelled ctx, and Shutdown marks it disconnected after + // cancelling. + if err := g.G().Syncer.Connected(ctx, chatCli, uid, &syncAllRes.Chat); err != nil && ctx.Err() == nil { return fmt.Errorf("error running chat sync: %s", err) } // Sync down events since we have been dead + // TODO: unlike the badge steps around it, serverSync is check-then-act: the connection can + // shut down between this check and the sync. Gating it means running an RPC under the gate. + if ctx.Err() != nil { + return chat.ErrDuplicateConnection + } if _, err := g.serverSync(ctx, gregor1.IncomingClient{Cli: timeoutCli}, gcli, &syncAllRes.Notification); err != nil { g.chatLog.Debug(ctx, "serverSync: failure: %s", err) return fmt.Errorf("error running state sync: %s", err) } - // Update badging from gregor. - if g.badger != nil { - state, err := gcli.StateMachineState(ctx, nil, false) - if err != nil { - g.chatLog.Debug(ctx, "unable to get gregor state for badging: %v", err) - g.badger.PushState(ctx, gregor1.State{}) - } else { - g.badger.PushState(ctx, state) + // Update badging from gregor, and call out to reachability module if we + // have one. + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { + if g.badger != nil { + state, err := gcli.StateMachineState(ctx, nil, false) + if err != nil { + g.chatLog.Debug(ctx, "unable to get gregor state for badging: %v", err) + g.badger.PushState(ctx, gregor1.State{}) + } else { + g.badger.PushState(ctx, state) + } } - } - - // Call out to reachability module if we have one - if g.reachability != nil { - g.chatLog.Debug(ctx, "setting reachability") - g.reachability.setReachability(keybase1.Reachability{ - Reachable: keybase1.Reachable_YES, - }) + if g.reachability != nil { + g.chatLog.Debug(ctx, "setting reachability") + g.reachability.setReachability(keybase1.Reachability{ + Reachable: keybase1.Reachable_YES, + }) + } + }) { + return chat.ErrDuplicateConnection } // Broadcast reconnect oobm. Spawn this off into a goroutine so that we don't delay @@ -895,12 +972,15 @@ func (g *gregorHandler) OnConnect(ctx context.Context, conn *rpc.Connection, } }(g.makeReconnectOobm()) - // No longer first connect if we are now connected - g.chatLog.Debug(ctx, "setting first connect to false") - g.setFirstConnect(false) - g.setConnectedAt(time.Now()) + // No longer first connect if we are now connected. + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { + g.chatLog.Debug(ctx, "setting first connect to false") + g.setFirstConnect(false) + g.setConnectedAt(time.Now()) + }) { + return chat.ErrDuplicateConnection + } g.chatLog.Debug(ctx, "OnConnect complete") - return nil } @@ -916,34 +996,53 @@ func (g *gregorHandler) connectedSince() time.Time { return g.connectedAt } -func (g *gregorHandler) OnConnectError(err error, reconnectThrottleDuration time.Duration) { +// gregorConnHandler is the rpc.ConnectionHandler for one connection. The rpc +// library can keep calling a connection's handler after Shutdown: any call on +// a shut-down connection starts a new reconnect loop, which reports +// OnDisconnected. connCtx, which Shutdown cancels under connMutex, tells those +// apart from events for the live connection. +type gregorConnHandler struct { + *gregorHandler + connCtx context.Context +} + +var _ rpc.ConnectionHandler = (*gregorConnHandler)(nil) + +func (h *gregorConnHandler) OnConnectError(err error, reconnectThrottleDuration time.Duration) { ctx := libkb.WithLogTag(context.Background(), "GRGRONERR") - defer g.chatLog.Trace(ctx, nil, "OnConnectError")() - g.chatLog.Debug(ctx, "OnConnectError: err: %s, reconnect throttle duration: %s", err, + defer h.chatLog.Trace(ctx, nil, "OnConnectError")() + h.chatLog.Debug(ctx, "OnConnectError: err: %s, reconnect throttle duration: %s", err, reconnectThrottleDuration) // Check reachability here to see the nature of our offline status go func() { - if g.reachability != nil && !g.isReachable(ctx) { - g.reachability.setReachability(keybase1.Reachability{ + if h.reachability != nil && !h.isReachable(ctx, h.connCtx) { + h.reachability.setReachability(keybase1.Reachability{ Reachable: keybase1.Reachable_NO, }) } }() } -func (g *gregorHandler) OnDisconnected(ctx context.Context, status rpc.DisconnectStatus) { +func (h *gregorConnHandler) OnDisconnected(ctx context.Context, status rpc.DisconnectStatus) { ctx = libkb.WithLogTag(ctx, "GRGRONDISC") - g.chatLog.Debug(ctx, "disconnected: %v", status) - g.setConnectedAt(time.Time{}) - + // Under connMutex, so no newer connection can be up while this runs. + h.connMutex.Lock() + if h.connCtx.Err() != nil { + h.connMutex.Unlock() + h.chatLog.Debug(ctx, "disconnected: %v, ignoring, not the current connection", status) + return + } + h.chatLog.Debug(ctx, "disconnected: %v", status) + h.setConnectedAt(time.Time{}) // Alert chat syncer that we are now disconnected - g.G().Syncer.Disconnected(ctx) + h.G().Syncer.Disconnected(ctx) + h.connMutex.Unlock() // Call out to reachability module if we have one (and we are currently connected) go func() { - if g.reachability != nil && status != rpc.StartingFirstConnection && !g.isReachable(ctx) { - g.reachability.setReachability(keybase1.Reachability{ + if h.reachability != nil && status != rpc.StartingFirstConnection && !h.isReachable(ctx, h.connCtx) { + h.reachability.setReachability(keybase1.Reachability{ Reachable: keybase1.Reachable_NO, }) } @@ -1355,6 +1454,10 @@ func (g *gregorHandler) handleOutOfBandMessage(ctx context.Context, obm gregor.O } } +// Shutdown disconnects. In production it is only ever called under the +// connection gate, from reconcile, reconnect or Reset, which is what keeps it +// from interleaving with the steps OnConnect applies after syncing. Tests +// call it directly. func (g *gregorHandler) Shutdown(ctx context.Context) { defer g.chatLog.Trace(ctx, nil, "Shutdown")() g.connMutex.Lock() @@ -1364,16 +1467,24 @@ func (g *gregorHandler) Shutdown(ctx context.Context) { return } - // Alert chat syncer that we are now disconnected - g.G().Syncer.Disconnected(ctx) - - close(g.shutdownCh) + g.connCancel() g.conn.Shutdown() + // After connCancel, which cancels the ctx of an OnConnect in flight, so a + // Syncer.Connected from it either lands before this and is overwritten, + // or sees the cancel and is skipped. + g.G().Syncer.Disconnected(ctx) g.conn = nil g.cli = nil + g.pingCli = nil g.setConnectedAt(time.Time{}) } +// Disconnect resets the connection and keeps it down until the next Connect, +// whatever the app state does meanwhile. +func (g *gregorHandler) Disconnect() error { + return g.connGate.forget(libkb.WithLogTag(context.Background(), "GRGRCONN")) +} + func (g *gregorHandler) Reset() error { g.Shutdown(context.Background()) g.setFirstConnect(true) @@ -1390,12 +1501,12 @@ const ( ) func (g *gregorHandler) loggedIn(ctx context.Context) (uid keybase1.UID, did keybase1.DeviceID, token string, nist *libkb.NIST, res loggedInRes) { - // Check to see if we have been shut down, - select { - case <-g.shutdownCh: + // Check to see if we have been shut down. + g.connMutex.Lock() + connCtx := g.connCtx + g.connMutex.Unlock() + if connCtx != nil && connCtx.Err() != nil { return uid, did, token, nil, loggedInMaybe - default: - // if we were going to block, then that means we are still alive } var err error @@ -1447,7 +1558,9 @@ func (g *gregorHandler) auth(ctx context.Context, cli rpc.GenericClient, auth *g return nil } -func (g *gregorHandler) isReachable(ctx context.Context) bool { +// isReachable dials gregord, and on failure reconnects the connection whose +// ctx is connCtx. +func (g *gregorHandler) isReachable(ctx context.Context, connCtx context.Context) bool { timeout := g.G().Env.GetGregorPingTimeout() url, err := url.Parse(g.G().Env.GetGregorURI()) if err != nil { @@ -1463,26 +1576,25 @@ func (g *gregorHandler) isReachable(ctx context.Context) bool { } if err != nil { g.chatLog.Debug(ctx, "isReachable: error: terminating connection: %s", err.Error()) - if _, err := g.Reconnect(ctx); err != nil { - g.chatLog.Debug(ctx, "isReachable: error reconnecting: %s", err.Error()) - } + g.connGate.requestReconnect(ctx, connCtx) return false } return true } -func (g *gregorHandler) Reconnect(ctx context.Context) (didShutdown bool, err error) { - if g.IsConnected() { - didShutdown = true - g.chatLog.Debug(ctx, "Reconnect: reconnecting to server") - g.Shutdown(ctx) - return didShutdown, g.Connect(g.uri) - } +// Reconnect drops the current connection and connects again when the app +// state allows it, without waiting for either. It does nothing if that +// connection has been replaced by the time the reconnect runs. +func (g *gregorHandler) Reconnect(ctx context.Context) { + g.connGate.requestReconnect(ctx, g.currentConnCtx()) +} - didShutdown = false - g.chatLog.Debug(ctx, "Reconnect: skipping reconnect, already disconnected") - return didShutdown, nil +// currentConnCtx is the ctx of the current connection, nil if there is none. +func (g *gregorHandler) currentConnCtx() context.Context { + g.connMutex.Lock() + defer g.connMutex.Unlock() + return g.connCtx } func (g *gregorHandler) forcePing(ctx context.Context) { @@ -1493,11 +1605,20 @@ func (g *gregorHandler) forcePing(ctx context.Context) { } } -func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, shutdownCancel context.CancelFunc) { - var err error - doneCh := make(chan error) +func (g *gregorHandler) pingOnce(ctx context.Context, conn context.Context, id []byte) { + g.connMutex.Lock() + pingCli := g.pingCli + g.connMutex.Unlock() + if pingCli == nil { + g.chatLog.Debug(ctx, "ping loop: id: %x no connection, skipping ping", id) + return + } + // Buffered so the goroutine never blocks after pingOnce has returned on + // conn's shutdown. + doneCh := make(chan error, 1) timeout := g.G().Env.GetGregorPingTimeout() go func(ctx context.Context) { + var err error if g.IsConnected() { // If we are connected, subject the ping call to a fairly // aggressive timeout so our chat stuff can be responsive @@ -1505,52 +1626,38 @@ func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, shutdownCancel var timeoutCancel context.CancelFunc var timeoutCtx context.Context timeoutCtx, timeoutCancel = context.WithTimeout(ctx, timeout) - _, err = gregor1.IncomingClient{Cli: g.pingCli}.Ping(timeoutCtx) + _, err = gregor1.IncomingClient{Cli: pingCli}.Ping(timeoutCtx) timeoutCancel() } else { // If we are not connected, we don't want to timeout anything // Just hook into the normal reconnect chan stuff in the RPC // library g.chatLog.Debug(ctx, "ping loop: id: %x normal ping, not connected", id) - _, err = gregor1.IncomingClient{Cli: g.pingCli}.Ping(ctx) + _, err = gregor1.IncomingClient{Cli: pingCli}.Ping(ctx) g.chatLog.Debug(ctx, "ping loop: id: %x normal ping success", id) } - select { - case <-ctx.Done(): - g.chatLog.Debug(ctx, "ping loop: id: %x context cancelled, so not sending err", id) - default: - doneCh <- err - } + doneCh <- err }(ctx) + var err error select { case err = <-doneCh: - case <-g.shutdownCh: + case <-conn.Done(): g.chatLog.Debug(ctx, "ping loop: id: %x shutdown received", id) - shutdownCancel() return } if err != nil { g.Debug(ctx, "ping loop: id: %x error: %s", id, err) if errors.Is(err, context.DeadlineExceeded) { g.chatLog.Debug(ctx, "ping loop: timeout: terminating connection") - var didShutdown bool - var err error - if didShutdown, err = g.Reconnect(ctx); err != nil { - g.chatLog.Debug(ctx, "ping loop: id: %x error reconnecting: %s", id, err) - } - // It is possible that we have already reconnected by the time we call Reconnect - // above. If that is the case, we don't want to terminate the ping loop. Only - // if Reconnect has actually reset the connection do we stop this ping loop. - if didShutdown { - shutdownCancel() - return - } + g.connGate.requestReconnect(ctx, conn) } } } -func (g *gregorHandler) pingLoop(ctx context.Context) { +// pingLoop runs until conn, the ctx of the connection it was started for, is +// done. +func (g *gregorHandler) pingLoop(ctx context.Context, conn context.Context) { id, _ := libkb.RandBytes(4) duration := g.G().Env.GetGregorPingInterval() timeout := g.G().Env.GetGregorPingTimeout() @@ -1564,34 +1671,27 @@ func (g *gregorHandler) pingLoop(ctx context.Context) { defer g.chatLog.Debug(ctx, "ping loop: id: %x terminating", id) ticker := time.NewTicker(duration) for { - pingCtx, shutdownCancel := context.WithCancel(libkb.CopyTagsToBackground(ctx)) + pingCtx, pingCancel := context.WithCancel(libkb.CopyTagsToBackground(ctx)) select { case <-g.forcePingCh: g.chatLog.Debug(pingCtx, "ping loop: forced attempt") - g.pingOnce(pingCtx, id, shutdownCancel) + g.pingOnce(pingCtx, conn, id) case <-ticker.C: - g.pingOnce(pingCtx, id, shutdownCancel) - case <-g.shutdownCh: + g.pingOnce(pingCtx, conn, id) + case <-conn.Done(): g.chatLog.Debug(pingCtx, "ping loop: id: %x shutdown received", id) - shutdownCancel() + pingCancel() return } - shutdownCancel() + pingCancel() } } -// connMutex must be locked before calling this -func (g *gregorHandler) connectTLS(ctx context.Context) error { - if g.conn != nil { - g.chatLog.Debug(ctx, "skipping connect, conn is not nil") - return nil - } - - uri := g.uri +func (g *gregorHandler) connectTLS(ctx context.Context, uri *rpc.FMPURI, handler rpc.ConnectionHandler) (*rpc.Connection, error) { g.chatLog.Debug(ctx, "connecting to gregord via TLS at %s", uri) rawCA := g.G().Env.GetBundledCA(uri.Host) if len(rawCA) == 0 { - return fmt.Errorf("No bundled CA for %s", uri.Host) + return nil, fmt.Errorf("No bundled CA for %s", uri.Host) } g.chatLog.Debug(ctx, "Using CA for gregor: %s", libkb.ShortCA(rawCA)) // Let people know we are trying to sync @@ -1608,37 +1708,17 @@ func (g *gregorHandler) connectTLS(ctx context.Context) error { // We deliberately avoid ForceInitialBackoff here, because we don't // want to penalize mobile, which tears down its connection frequently. } - g.conn = rpc.NewTLSConnectionWithDialable(rpc.NewFixedRemote(uri.HostPort), + return rpc.NewTLSConnectionWithDialable(rpc.NewFixedRemote(uri.HostPort), []byte(rawCA), libkb.NewContextifiedErrorUnwrapper(g.G().ExternalG()), - g, libkb.NewRPCLogFactory(g.G().ExternalG()), + handler, libkb.NewRPCLogFactory(g.G().ExternalG()), g.G().ExternalG().RemoteNetworkInstrumenterStorage, logger.LogOutputWithDepthAdder{Logger: g.G().Log}, rpc.DefaultMaxFrameLength, opts, - libkb.NewProxyDialable(g.G().Env)) - - // The client we get here will reconnect to gregord on disconnect if necessary. - // We should grab it here instead of in OnConnect, since the connection is not - // fully established in OnConnect. Anything that wants to make calls outside - // of OnConnect should use g.cli, everything else should the client that is - // a parameter to OnConnect - g.cli = WrapGenericClientWithTimeout(g.conn.GetClient(), GregorRequestTimeout, - chat.ErrChatServerTimeout) - g.pingCli = g.conn.GetClient() // Don't want this to have a timeout from here - - // Start up ping loop to keep the connection to gregord alive, and to kick - // off the reconnect logic in the RPC library - go g.pingLoop(ctx) - - return nil + libkb.NewProxyDialable(g.G().Env)), nil } // connMutex must be locked before calling this -func (g *gregorHandler) connectNoTLS(ctx context.Context) error { - if g.conn != nil { - g.chatLog.Debug(ctx, "skipping connect, conn is not nil") - return nil - } - uri := g.uri +func (g *gregorHandler) connectNoTLS(ctx context.Context, uri *rpc.FMPURI, handler rpc.ConnectionHandler) *rpc.Connection { g.chatLog.Debug(ctx, "connecting to gregord without TLS at %s", uri) t := newConnTransport(g.G().ExternalG(), uri.HostPort) g.transportForTesting = t @@ -1650,19 +1730,9 @@ func (g *gregorHandler) connectNoTLS(ctx context.Context) error { return backoff.NewConstantBackOff(GregorConnectionRetryInterval) }, } - g.conn = rpc.NewConnectionWithTransport(g, t, + return rpc.NewConnectionWithTransport(handler, t, libkb.NewContextifiedErrorUnwrapper(g.G().ExternalG()), logger.LogOutputWithDepthAdder{Logger: g.G().Log}, opts) - - g.cli = WrapGenericClientWithTimeout(g.conn.GetClient(), GregorRequestTimeout, - chat.ErrChatServerTimeout) - g.pingCli = g.conn.GetClient() - - // Start up ping loop to keep the connection to gregord alive, and to kick - // off the reconnect logic in the RPC library - go g.pingLoop(ctx) - - return nil } func (g *gregorHandler) currentUID() gregor1.UID { diff --git a/go/service/gregor_conn.go b/go/service/gregor_conn.go new file mode 100644 index 000000000000..e9df7e074b6b --- /dev/null +++ b/go/service/gregor_conn.go @@ -0,0 +1,255 @@ +package service + +import ( + "context" + "sync" + + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/keybase/go-framed-msgpack-rpc/rpc" +) + +// gregorConnector is the connection gregorConnGate drives: the gregor +// handler, or a fake in tests. +type gregorConnector interface { + // connectNow connects to uri, doing nothing if already connected. + connectNow(uri *rpc.FMPURI) error + // Shutdown disconnects, doing nothing if not connected. + Shutdown(ctx context.Context) + Reset() error + IsConnected() bool + // connExists reports whether there is a connection, connected or not. + connExists() bool +} + +// gregorAppState is the mobile app state the gate follows, as an interface so +// it can be substituted: tests wrap the real one to act between a connect's +// state read and what the connect does with it. +type gregorAppState interface { + State() keybase1.MobileAppState + NextUpdate(lastState keybase1.MobileAppState) <-chan struct{} +} + +// gregorConnGate decides when gregor is connected. Only BACKGROUND, or a +// desktop suspend, takes the connection down; INACTIVE keeps it up. +// +// Every connect and the monitor read the app state and act on it under mu. +// A BACKGROUND that lands after a connect read the state wakes the monitor, +// which then waits for that connect before taking the connection down. mu +// also runs the steps OnConnect applies after syncing (runIfLive), so none of +// them interleaves with a disconnect, and guards the uri OnConnect reads. +// +// This is a mutex gate rather than a single owning goroutine like +// kbhttp/manager's Srv: connect and forget return errors their callers need, +// and the OnConnect steps must report "no longer current" back on the caller's +// goroutine so onConnectSynced can return ErrDuplicateConnection. A +// request-channel loop would need a reply channel per request -- more code and +// more states -- so do not harmonise the two shapes. Only reconnect, whose +// callers need no result, is a request the monitor runs. +type gregorConnGate struct { + mobile gregorAppState + desktop *libkb.DesktopAppState + conn gregorConnector + debug func(ctx context.Context, format string, args ...any) + onForeground func(ctx context.Context) + + mu sync.Mutex + // uri is the last URI a connect asked for. It is kept when the connect is + // held back by BACKGROUND or a desktop suspend, so the monitor connects + // once that ends. + uri *rpc.FMPURI + // The monitor's last seen states and the change channels it waits on for + // them; tests use them to wait until the monitor has caught up. + monitorState keybase1.MobileAppState + monitorSuspended bool + monitorWait <-chan struct{} + monitorSuspendWait <-chan struct{} + + // reconnectCh holds at most one wakeup for the monitor, so a burst of + // reconnect requests coalesces. reconnectFor is the ctx of the connection + // the pending request is for; the monitor skips it once that connection + // has been shut down, so a request never tears down a newer one. + reconnectCh chan struct{} + reconnectMu sync.Mutex + reconnectFor context.Context + + startOnce sync.Once + stopOnce sync.Once + stopCh chan struct{} + monitorDone chan struct{} +} + +func newGregorConnGate(mobile gregorAppState, desktop *libkb.DesktopAppState, conn gregorConnector, + debug func(ctx context.Context, format string, args ...any), onForeground func(ctx context.Context), +) *gregorConnGate { + return &gregorConnGate{ + mobile: mobile, + desktop: desktop, + conn: conn, + debug: debug, + onForeground: onForeground, + reconnectCh: make(chan struct{}, 1), + stopCh: make(chan struct{}), + monitorDone: make(chan struct{}), + } +} + +// start reconciles against the current state and starts the monitor. +func (c *gregorConnGate) start() { + c.startOnce.Do(func() { + ctx := libkb.WithLogTag(context.Background(), "GRGRMON") + state, suspended := c.mobile.State(), c.desktop.Suspended() + c.debug(ctx, "monitorAppState: starting up in %v (suspended: %v)", state, suspended) + c.reconcile(ctx) + go c.monitor(ctx, state, suspended) + }) +} + +// stop tells the monitor to exit, without waiting for it. It does not +// disconnect. +func (c *gregorConnGate) stop() { + c.stopOnce.Do(func() { close(c.stopCh) }) +} + +// connect connects to uri when reconcile allows it. With reset, an existing +// connection is reset first so it authenticates again; that includes one that +// is not connected, such as one whose auth failed while logged out, which +// would otherwise keep connectNow from dialing. With no connection there is +// nothing to reset, and resetting would drop the gregor client until the next +// connect, which BACKGROUND or a suspend can put off. +func (c *gregorConnGate) connect(ctx context.Context, uri *rpc.FMPURI, reset bool) error { + c.mu.Lock() + defer c.mu.Unlock() + c.uri = uri + if reset && c.conn.connExists() { + if err := c.conn.Reset(); err != nil { + return err + } + } + return c.reconcileLocked(ctx) +} + +// forget resets the connection and drops the uri, so nothing reconnects until +// the next connect. +func (c *gregorConnGate) forget(ctx context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + c.debug(ctx, "forget: resetting and forgetting the uri") + c.uri = nil + return c.conn.Reset() +} + +// runIfLive runs f with the gate's uri, under mu, if ctx, an OnConnect ctx, is +// still live, and reports whether it ran. Every Shutdown and Reset is made +// under mu too, and Shutdown cancels ctx, so a disconnect lands entirely +// before f, and f is then skipped, or entirely after it. f must not call back +// into the gate: mu is not reentrant. +func (c *gregorConnGate) runIfLive(ctx context.Context, f func(uri *rpc.FMPURI)) bool { + c.mu.Lock() + defer c.mu.Unlock() + if ctx.Err() != nil { + return false + } + f(c.uri) + return true +} + +// requestReconnect asks the monitor to reconnect the connection whose ctx is +// conn, and returns without waiting. A pending request for a connection that +// is still live is kept over a newer one, which can only be for that +// connection or for one already shut down. +func (c *gregorConnGate) requestReconnect(ctx context.Context, conn context.Context) { + c.reconnectMu.Lock() + if c.reconnectFor == nil || c.reconnectFor.Err() != nil { + c.reconnectFor = conn + } + c.reconnectMu.Unlock() + select { + case c.reconnectCh <- struct{}{}: + c.debug(ctx, "Reconnect: requested") + default: + c.debug(ctx, "Reconnect: one is already pending") + } +} + +func (c *gregorConnGate) takeReconnect() context.Context { + c.reconnectMu.Lock() + defer c.reconnectMu.Unlock() + conn := c.reconnectFor + c.reconnectFor = nil + return conn +} + +// reconnect drops the connection whose ctx is conn, if it is still the live +// one, and connects again when reconcile allows it. Shutdown cancels a +// connection's ctx under mu, so the check holds for the whole reconnect. +func (c *gregorConnGate) reconnect(ctx context.Context, conn context.Context) { + c.mu.Lock() + defer c.mu.Unlock() + if conn == nil || conn.Err() != nil { + c.debug(ctx, "Reconnect: skipping reconnect, the connection was already replaced") + return + } + if !c.conn.IsConnected() { + c.debug(ctx, "Reconnect: skipping reconnect, already disconnected") + return + } + c.debug(ctx, "Reconnect: reconnecting to server") + c.conn.Shutdown(ctx) + if err := c.reconcileLocked(ctx); err != nil { + c.debug(ctx, "Reconnect: error connecting: %s", err) + } +} + +func (c *gregorConnGate) reconcile(ctx context.Context) { + c.mu.Lock() + defer c.mu.Unlock() + if err := c.reconcileLocked(ctx); err != nil { + c.debug(ctx, "reconcile: error connecting: %s", err) + } +} + +// reconcileLocked is the only place that decides whether a connection may +// exist: none in BACKGROUND or while the desktop is suspended, otherwise one +// to the uri, if any. c.mu must be held. +func (c *gregorConnGate) reconcileLocked(ctx context.Context) error { + state, suspended := c.mobile.State(), c.desktop.Suspended() + if state == keybase1.MobileAppState_BACKGROUND || suspended { + c.debug(ctx, "reconcile: disconnecting in %v (suspended: %v)", state, suspended) + c.conn.Shutdown(ctx) + return nil + } + // Nothing asked to connect yet, for example before login. + if c.uri == nil { + return nil + } + c.debug(ctx, "reconcile: connecting in %v", state) + return c.conn.connectNow(c.uri) +} + +func (c *gregorConnGate) monitor(ctx context.Context, state keybase1.MobileAppState, suspended bool) { + defer close(c.monitorDone) + for { + next := c.mobile.NextUpdate(state) + nextSuspend := c.desktop.NextSuspendUpdate(suspended) + c.mu.Lock() + c.monitorState, c.monitorSuspended = state, suspended + c.monitorWait, c.monitorSuspendWait = next, nextSuspend + c.mu.Unlock() + select { + case <-next: + case <-nextSuspend: + case <-c.reconnectCh: + c.reconnect(ctx, c.takeReconnect()) + continue + case <-c.stopCh: + return + } + prev := state + state, suspended = c.mobile.State(), c.desktop.Suspended() + if state != prev && state == keybase1.MobileAppState_FOREGROUND { + c.onForeground(ctx) + } + c.reconcile(ctx) + } +} diff --git a/go/service/gregor_conn_test.go b/go/service/gregor_conn_test.go new file mode 100644 index 000000000000..bbae8ea8f69b --- /dev/null +++ b/go/service/gregor_conn_test.go @@ -0,0 +1,1396 @@ +package service + +import ( + "context" + "errors" + "fmt" + "math/rand" + "net" + "os" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/keybase/client/go/chat" + "github.com/keybase/client/go/chat/types" + "github.com/keybase/client/go/gregor" + grclient "github.com/keybase/client/go/gregor/client" + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/chat1" + "github.com/keybase/client/go/protocol/gregor1" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/keybase/go-framed-msgpack-rpc/rpc" + "github.com/stretchr/testify/require" +) + +// fakeGregorConn models the handler's connection: it can exist without being +// connected (stale), and connectNow does nothing while one exists. +type fakeGregorConn struct { + sync.Mutex + exists bool + up bool + uri *rpc.FMPURI + connects int + shutdowns int + resets int +} + +func (f *fakeGregorConn) connectNow(uri *rpc.FMPURI) error { + f.Lock() + defer f.Unlock() + if !f.exists { + f.exists, f.up = true, true + f.uri = uri + f.connects++ + } + return nil +} + +func (f *fakeGregorConn) Shutdown(context.Context) { + f.Lock() + defer f.Unlock() + if f.exists { + f.exists, f.up = false, false + f.shutdowns++ + } +} + +func (f *fakeGregorConn) Reset() error { + f.Shutdown(context.Background()) + f.Lock() + defer f.Unlock() + f.resets++ + return nil +} + +// goStale leaves the connection in place but not connected, as when its auth +// fails with an error the connection does not retry. +func (f *fakeGregorConn) goStale() { + f.Lock() + defer f.Unlock() + f.up = false +} + +func (f *fakeGregorConn) connExists() bool { + f.Lock() + defer f.Unlock() + return f.exists +} + +func (f *fakeGregorConn) IsConnected() bool { + f.Lock() + defer f.Unlock() + return f.up +} + +type fakeGregorCounts struct { + up bool + connects, shutdowns, resets int +} + +func (f *fakeGregorConn) counts() fakeGregorCounts { + f.Lock() + defer f.Unlock() + return fakeGregorCounts{up: f.up, connects: f.connects, shutdowns: f.shutdowns, resets: f.resets} +} + +func (f *fakeGregorConn) lastURI() *rpc.FMPURI { + f.Lock() + defer f.Unlock() + return f.uri +} + +// gregorTestAppState wraps the real app state so a test can act between a +// connect's state read and what the connect does with that read. +type gregorTestAppState struct { + *libkb.MobileAppState + mu sync.Mutex + // afterRead, if set, runs once after a State read, before the reader acts. + afterRead func() +} + +func (a *gregorTestAppState) State() keybase1.MobileAppState { + state := a.MobileAppState.State() + a.mu.Lock() + f := a.afterRead + a.afterRead = nil + a.mu.Unlock() + if f != nil { + f() + } + return state +} + +func (a *gregorTestAppState) setAfterRead(f func()) { + a.mu.Lock() + defer a.mu.Unlock() + a.afterRead = f +} + +type gregorConnTest struct { + tc libkb.TestContext + gate *gregorConnGate + mobile *gregorTestAppState + conn *fakeGregorConn + pings *atomic.Int64 +} + +func testGregorURI(t testing.TB, host string) *rpc.FMPURI { + uri, err := rpc.ParseFMPURI(fmt.Sprintf("fmprpc+tls://%s:443", host)) + require.NoError(t, err) + return uri +} + +// setupGregorConn builds a gate in state and starts it, as Init does before +// the service's first connect. +func setupGregorConn(t *testing.T, state keybase1.MobileAppState) *gregorConnTest { + tc := libkb.SetupTest(t, "gregorconn", 2) + t.Cleanup(tc.Cleanup) + tc.G.MobileAppState.Update(state) + conn := &fakeGregorConn{} + pings := &atomic.Int64{} + mobile := &gregorTestAppState{MobileAppState: tc.G.MobileAppState} + gate := newGregorConnGate(mobile, tc.G.DesktopAppState, conn, + func(ctx context.Context, format string, args ...any) { t.Logf(format, args...) }, + func(context.Context) { pings.Add(1) }) + gate.start() + t.Cleanup(func() { + gate.stop() + select { + case <-gate.monitorDone: + case <-time.After(10 * time.Second): + t.Error("monitor did not exit on stop") + } + }) + return &gregorConnTest{tc: tc, gate: gate, mobile: mobile, conn: conn, pings: pings} +} + +// waitMonitor waits until the monitor has acted on the current states and is +// waiting for the next change. +func (c *gregorConnTest) waitMonitor(t *testing.T) { + t.Helper() + g := c.tc.G + require.Eventually(t, func() bool { + c.gate.mu.Lock() + state, suspended := c.gate.monitorState, c.gate.monitorSuspended + wait, suspendWait := c.gate.monitorWait, c.gate.monitorSuspendWait + c.gate.mu.Unlock() + if wait == nil || wait != g.MobileAppState.NextUpdate(state) || + suspendWait != g.DesktopAppState.NextSuspendUpdate(suspended) { + return false + } + select { + case <-wait: + return false + case <-suspendWait: + return false + default: + return true + } + }, 10*time.Second, time.Millisecond, "monitor did not catch up") +} + +func (c *gregorConnTest) update(t *testing.T, state keybase1.MobileAppState) { + t.Helper() + c.tc.G.MobileAppState.Update(state) + c.waitMonitor(t) +} + +func (c *gregorConnTest) requireUp(t *testing.T, up bool, msg string) { + t.Helper() + require.Equal(t, up, c.conn.IsConnected(), msg) +} + +func TestGregorConnStartupInBackground(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_BACKGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + c.requireUp(t, false, "connected during a background launch") + require.Equal(t, 0, c.conn.counts().connects) + + c.update(t, keybase1.MobileAppState_BACKGROUNDACTIVE) + c.requireUp(t, true, "did not connect on leaving BACKGROUND") + require.Equal(t, uri, c.conn.lastURI()) + require.Equal(t, fakeGregorCounts{up: true, connects: 1}, c.conn.counts()) +} + +func TestGregorConnLoginInBackground(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + first := testGregorURI(t, "first.test") + require.NoError(t, c.gate.connect(context.Background(), first, true)) + require.Equal(t, fakeGregorCounts{up: true, connects: 1}, c.conn.counts(), "reset with no connection") + + c.update(t, keybase1.MobileAppState_BACKGROUND) + c.requireUp(t, false, "still connected in BACKGROUND") + + second := testGregorURI(t, "second.test") + require.NoError(t, c.gate.connect(context.Background(), second, true)) + c.requireUp(t, false, "login connected in BACKGROUND") + // No connection exists in BACKGROUND, so the login resets nothing and + // the gregor client survives until the foreground connect. + require.Equal(t, fakeGregorCounts{connects: 1, shutdowns: 1}, c.conn.counts(), "reset with no connection") + + c.update(t, keybase1.MobileAppState_FOREGROUND) + c.requireUp(t, true, "did not connect on foreground after a background login") + require.Equal(t, second, c.conn.lastURI()) + + // A login while connected resets the connection before connecting. + require.NoError(t, c.gate.connect(context.Background(), first, true)) + require.Equal(t, fakeGregorCounts{up: true, connects: 3, shutdowns: 2, resets: 1}, c.conn.counts()) + require.Equal(t, first, c.conn.lastURI()) +} + +func TestGregorConnInactiveStaysConnected(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + require.NoError(t, c.gate.connect(context.Background(), testGregorURI(t, "gregord.test"), false)) + for range 3 { + c.update(t, keybase1.MobileAppState_INACTIVE) + c.requireUp(t, true, "INACTIVE disconnected") + c.update(t, keybase1.MobileAppState_FOREGROUND) + c.requireUp(t, true, "FOREGROUND disconnected") + } + require.Equal(t, fakeGregorCounts{up: true, connects: 1}, c.conn.counts()) + require.EqualValues(t, 3, c.pings.Load()) +} + +func TestGregorConnDuplicateEvents(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_BACKGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + for range 3 { + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + } + for range 3 { + c.update(t, keybase1.MobileAppState_BACKGROUND) + } + require.Equal(t, fakeGregorCounts{}, c.conn.counts()) + + for round := 1; round <= 3; round++ { + for range 3 { + c.update(t, keybase1.MobileAppState_BACKGROUNDACTIVE) + c.requireUp(t, true, "down in BACKGROUNDACTIVE") + } + for range 3 { + c.update(t, keybase1.MobileAppState_FOREGROUND) + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + c.requireUp(t, true, "down in FOREGROUND") + } + for range 3 { + c.update(t, keybase1.MobileAppState_BACKGROUND) + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + c.requireUp(t, false, "up in BACKGROUND") + } + require.Equal(t, fakeGregorCounts{connects: round, shutdowns: round}, c.conn.counts()) + } + require.EqualValues(t, 3, c.pings.Load()) +} + +var allAppStates = []keybase1.MobileAppState{ + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, +} + +// requireStaysDown drives every transition and checks that nothing connects. +func (c *gregorConnTest) requireStaysDown(t *testing.T, why string) { + t.Helper() + connects := c.conn.counts().connects + for _, state := range allAppStates { + c.update(t, state) + c.gate.reconnect(context.Background(), context.Background()) + c.requireUp(t, false, fmt.Sprintf("connected in %v %s", state, why)) + } + require.Equal(t, connects, c.conn.counts().connects, "connect attempted "+why) +} + +func TestGregorConnLogoutStaysDown(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + require.NoError(t, c.gate.connect(context.Background(), uri, true)) + c.requireUp(t, true, "login did not connect") + + require.NoError(t, c.gate.forget(context.Background())) + c.requireUp(t, false, "logout left gregor connected") + c.requireStaysDown(t, "after logout") + + c.update(t, keybase1.MobileAppState_FOREGROUND) + require.NoError(t, c.gate.connect(context.Background(), uri, true)) + c.requireUp(t, true, "login after logout did not connect") +} + +func TestGregorConnLogoutInBackground(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + require.NoError(t, c.gate.connect(context.Background(), testGregorURI(t, "gregord.test"), true)) + c.update(t, keybase1.MobileAppState_BACKGROUND) + require.NoError(t, c.gate.forget(context.Background())) + c.update(t, keybase1.MobileAppState_BACKGROUNDACTIVE) + c.update(t, keybase1.MobileAppState_FOREGROUND) + c.requireUp(t, false, "foreground after a background logout connected") + require.Equal(t, 1, c.conn.counts().connects) +} + +// A connection whose auth failed while logged out stays in place without +// being connected; the next login must still connect. +func TestGregorConnLoginReplacesStaleConn(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + c.conn.goStale() + c.update(t, keybase1.MobileAppState_INACTIVE) + c.update(t, keybase1.MobileAppState_FOREGROUND) + require.NoError(t, c.gate.connect(context.Background(), uri, true)) + c.requireUp(t, true, "login left a stale connection in place") + require.Equal(t, fakeGregorCounts{up: true, connects: 2, shutdowns: 1, resets: 1}, c.conn.counts()) +} + +// A BACKGROUND applied while a connect is deciding must not leave gregor +// connected. +func TestGregorConnBackgroundRacingConnect(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + c.mobile.setAfterRead(func() { + // connect has read FOREGROUND. The monitor is idle, so mu is held + // here only if connect holds it; otherwise let the monitor fully + // apply BACKGROUND before connect acts on its stale read. + holdsMu := !c.gate.mu.TryLock() + if !holdsMu { + c.gate.mu.Unlock() + } + c.tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + if !holdsMu { + c.waitMonitor(t) + } + }) + require.NoError(t, c.gate.connect(context.Background(), testGregorURI(t, "gregord.test"), false)) + c.waitMonitor(t) + require.Equal(t, keybase1.MobileAppState_BACKGROUND, c.tc.G.MobileAppState.State()) + c.requireUp(t, false, "connected in BACKGROUND after racing a connect") +} + +func TestGregorConnReconnectInBackground(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + + c.gate.reconnect(context.Background(), context.Background()) + require.Equal(t, fakeGregorCounts{up: true, connects: 2, shutdowns: 1}, c.conn.counts()) + + // A connection left up while BACKGROUND lands, as when a ping times out + // before the monitor has acted. + c.gate.mu.Lock() + c.tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + c.gate.mu.Unlock() + c.waitMonitor(t) + require.NoError(t, c.conn.connectNow(uri)) + c.gate.reconnect(context.Background(), context.Background()) + c.requireUp(t, false, "reconnect connected in BACKGROUND") + require.Equal(t, fakeGregorCounts{connects: 3, shutdowns: 3}, c.conn.counts()) + + c.gate.reconnect(context.Background(), context.Background()) + c.requireUp(t, false, "reconnect connected while disconnected") + require.Equal(t, fakeGregorCounts{connects: 3, shutdowns: 3}, c.conn.counts()) +} + +// A reconnect request returns without waiting, even while the gate is held, +// and a burst of requests made before the monitor runs is one reconnect. +func TestGregorConnReconnectRequestsCoalesce(t *testing.T) { + tc := libkb.SetupTest(t, "gregorconn", 1) + defer tc.Cleanup() + conn := &fakeGregorConn{} + mobile := &gregorTestAppState{MobileAppState: tc.G.MobileAppState} + gate := newGregorConnGate(mobile, tc.G.DesktopAppState, conn, + func(ctx context.Context, format string, args ...any) { t.Logf(format, args...) }, + func(context.Context) {}) + c := &gregorConnTest{tc: tc, gate: gate, mobile: mobile, conn: conn} + require.NoError(t, gate.connect(context.Background(), testGregorURI(t, "gregord.test"), false)) + + gate.mu.Lock() + done := make(chan struct{}) + go func() { + defer close(done) + for range 10 { + gate.requestReconnect(context.Background(), context.Background()) + } + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("a reconnect request waited") + } + gate.mu.Unlock() + require.Equal(t, fakeGregorCounts{up: true, connects: 1}, conn.counts(), "reconnected without the monitor") + + gate.start() + defer func() { + gate.stop() + <-gate.monitorDone + }() + want := fakeGregorCounts{up: true, connects: 2, shutdowns: 1} + require.Eventually(t, func() bool { return conn.counts() == want }, 10*time.Second, time.Millisecond, + "did not reconnect exactly once") + c.waitMonitor(t) + require.Empty(t, gate.reconnectCh, "a request is still pending") + require.Equal(t, want, conn.counts(), "reconnected more than once") +} + +func TestGregorConnDesktopSuspend(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + require.NoError(t, c.gate.connect(context.Background(), testGregorURI(t, "gregord.test"), false)) + mctx := libkb.NewMetaContextForTest(c.tc) + c.tc.G.DesktopAppState.Update(mctx, "suspend", nil) + c.waitMonitor(t) + c.requireUp(t, false, "connected while suspended") + c.tc.G.DesktopAppState.Update(mctx, "resume", nil) + c.waitMonitor(t) + c.requireUp(t, true, "did not connect on resume") + require.Equal(t, fakeGregorCounts{up: true, connects: 2, shutdowns: 1}, c.conn.counts()) +} + +// A ping timeout that reconnects while the machine is suspended must not +// dial, and neither must a connect; resuming connects. +func TestGregorReconnectWhileSuspendedDoesNotConnect(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + uri := testGregorURI(t, "gregord.test") + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + + // A connection left up while the suspend lands, as when a ping times out + // before the monitor has acted. + mctx := libkb.NewMetaContextForTest(c.tc) + c.gate.mu.Lock() + c.tc.G.DesktopAppState.Update(mctx, "suspend", nil) + c.gate.mu.Unlock() + c.waitMonitor(t) + require.NoError(t, c.conn.connectNow(uri)) + c.gate.reconnect(context.Background(), context.Background()) + c.requireUp(t, false, "reconnect connected while suspended") + require.Equal(t, fakeGregorCounts{connects: 2, shutdowns: 2}, c.conn.counts()) + + require.NoError(t, c.gate.connect(context.Background(), uri, false)) + c.requireUp(t, false, "connect connected while suspended") + require.Equal(t, 2, c.conn.counts().connects) + + c.tc.G.DesktopAppState.Update(mctx, "resume", nil) + c.waitMonitor(t) + c.requireUp(t, true, "did not connect on resume") + require.Equal(t, 3, c.conn.counts().connects) +} + +func TestGregorConnStress(t *testing.T) { + tc := libkb.SetupTest(t, "gregorconn", 1) + defer tc.Cleanup() + baseline := runtime.NumGoroutine() + + conn := &fakeGregorConn{} + mobile := &gregorTestAppState{MobileAppState: tc.G.MobileAppState} + gate := newGregorConnGate(mobile, tc.G.DesktopAppState, conn, + func(context.Context, string, ...any) {}, func(context.Context) {}) + gate.start() + c := &gregorConnTest{tc: tc, gate: gate, mobile: mobile, conn: conn} + uri := testGregorURI(t, "gregord.test") + states := []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } + + stop := make(chan struct{}) + var workers, writers sync.WaitGroup + for w := range 4 { + workers.Add(1) + go func() { + defer workers.Done() + ctx := context.Background() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + switch (i + w) % 4 { + case 0: + _ = gate.connect(ctx, uri, false) + case 1: + _ = gate.connect(ctx, uri, true) + case 2: + _ = gate.forget(ctx) + default: + gate.requestReconnect(ctx, ctx) + } + runtime.Gosched() + } + }() + } + for w := range 4 { + writers.Add(1) + go func() { + defer writers.Done() + rng := rand.New(rand.NewSource(int64(w))) + for range 500 { + tc.G.MobileAppState.Update(states[rng.Intn(len(states))]) + if rng.Intn(4) == 0 { + time.Sleep(time.Duration(rng.Intn(200)) * time.Microsecond) + } + } + }() + } + + done := make(chan struct{}) + go func() { + writers.Wait() + close(stop) + workers.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(60 * time.Second): + // A deadlock leaves the workers holding the gate, so the cleanup that + // stops the monitor never returns. Nothing this test writes is + // flushed through that hung unwind, neither t.Fatal's message nor a + // panic's, so say it on stderr first; go test's own timeout then + // dumps every stack. + const msg = "deadlock: transitions and connects did not finish" + fmt.Fprintln(os.Stderr, msg) + t.Fatal(msg) + } + + require.NoError(t, gate.forget(context.Background())) + c.requireStaysDown(t, "after settling logged out") + require.NoError(t, gate.connect(context.Background(), uri, true)) + c.requireUp(t, true, "login did not connect after settling") + c.update(t, keybase1.MobileAppState_BACKGROUND) + c.requireUp(t, false, "up after settling in BACKGROUND") + c.update(t, keybase1.MobileAppState_BACKGROUNDACTIVE) + c.requireUp(t, true, "down after settling in BACKGROUNDACTIVE") + counts := conn.counts() + t.Logf("%d connects, %d shutdowns, %d resets", counts.connects, counts.shutdowns, counts.resets) + + gate.stop() + select { + case <-gate.monitorDone: + case <-time.After(10 * time.Second): + t.Fatal("monitor did not exit on stop") + } + deadline := time.Now().Add(10 * time.Second) + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + require.LessOrEqual(t, runtime.NumGoroutine(), baseline, "leaked goroutines") +} + +// Connects and shutdowns race a reader of the gate's URI and the transport's +// dial. Nothing listens on the port, so OnConnect never runs; this covers the +// gate under -race, not a live connection. +func TestGregorHandlerConnectRaces(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + + uri := closedPortURI(t) + h := newGregorHandler(g) + stop := make(chan struct{}) + readerDone := make(chan struct{}) + go func() { + defer close(readerDone) + for { + select { + case <-stop: + return + default: + } + _ = gateURI(h) + runtime.Gosched() + } + }() + for i := range 20 { + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + // Vary how far the dial gets before the shutdown. + time.Sleep(time.Duration(i%4) * time.Millisecond) + h.Shutdown(context.Background()) + } + close(stop) + <-readerDone + require.Equal(t, uri, gateURI(h)) +} + +func gateURI(h *gregorHandler) *rpc.FMPURI { + h.connGate.mu.Lock() + defer h.connGate.mu.Unlock() + return h.connGate.uri +} + +// A connect that fails before it creates a connection, as with no bundled CA +// for the host, leaves nothing running for it, however often it is retried. +func TestGregorHandlerFailedConnectLeavesNothingRunning(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + h := newGregorHandler(g) + uri, err := rpc.ParseFMPURI("fmprpc+tls://no-bundled-ca.test:443") + require.NoError(t, err) + + baseline := runtime.NumGoroutine() + for range 20 { + require.ErrorContains(t, h.connGate.connect(context.Background(), uri, false), "No bundled CA") + h.connGate.reconcile(context.Background()) + } + require.False(t, hasConn(h)) + deadline := time.Now().Add(10 * time.Second) + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + require.LessOrEqual(t, runtime.NumGoroutine(), baseline, "a failed connect leaked goroutines") +} + +// Everything a connection starts, its ping loop and push state debouncer +// included, exits when it is shut down. +func TestGregorHandlerShutdownStopsConnGoroutines(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + h := newGregorHandler(g) + uri := closedPortURI(t) + + baseline := runtime.NumGoroutine() + for range 10 { + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + require.True(t, hasConn(h)) + h.Shutdown(context.Background()) + } + deadline := time.Now().Add(10 * time.Second) + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + require.LessOrEqual(t, runtime.NumGoroutine(), baseline, "a shut down connection left goroutines running") +} + +// A shut down connection's auth reports loggedInMaybe instead of checking the +// login. +func TestGregorHandlerLoggedInAfterShutdown(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + h := newGregorHandler(g) + ctx := context.Background() + + _, _, _, _, res := h.loggedIn(ctx) + require.Equal(t, loggedInNo, res) + require.NoError(t, h.connGate.connect(context.Background(), closedPortURI(t), false)) + _, _, _, _, res = h.loggedIn(ctx) + require.Equal(t, loggedInNo, res) + h.Shutdown(ctx) + _, _, _, _, res = h.loggedIn(ctx) + require.Equal(t, loggedInMaybe, res) +} + +// closedPortURI points at a closed port, so a connection only retries until +// shut down. +func closedPortURI(t *testing.T) *rpc.FMPURI { + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := l.Addr().String() + require.NoError(t, l.Close()) + uri, err := rpc.ParseFMPURI("fmprpc://" + addr) + require.NoError(t, err) + return uri +} + +func hasConn(h *gregorHandler) bool { + h.connMutex.Lock() + defer h.connMutex.Unlock() + return h.conn != nil +} + +// The service's startup and login connects go through the handler's gate. +// Logout goes through the handler's gate, so no transition reconnects. +func TestGregorHandlerDisconnectStaysDown(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + + h := newGregorHandler(g) + require.NoError(t, h.ConnectFresh(closedPortURI(t))) + require.True(t, hasConn(h), "did not connect") + require.NoError(t, h.Disconnect()) + require.False(t, hasConn(h), "Disconnect left a connection") + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + h.connGate.reconcile(context.Background()) + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + h.connGate.reconcile(context.Background()) + require.False(t, hasConn(h), "reconnected after Disconnect") +} + +// The service skips Init when gregor is disabled or in Tor mode, so the +// gate's monitor never starts, but a logout still disconnects. It must +// return instead of waiting for anything. +func TestGregorHandlerDisconnectWithoutInit(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + + h := newGregorHandler(g) + done := make(chan error, 1) + go func() { done <- h.Disconnect() }() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(10 * time.Second): + require.Fail(t, "Disconnect blocked with no Init") + } +} + +func TestGregorHandlerConnectInBackground(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + + h := newGregorHandler(g) + uri := closedPortURI(t) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + require.False(t, hasConn(h), "a connect connected in BACKGROUND") + require.NoError(t, h.ConnectFresh(uri)) + require.False(t, hasConn(h), "ConnectFresh connected in BACKGROUND") + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + h.connGate.reconcile(context.Background()) + require.True(t, hasConn(h), "did not connect on leaving BACKGROUND") + h.Shutdown(context.Background()) +} + +// acceptingListener accepts and holds connections, counting them, so a +// connection dials successfully and then fails in OnConnect. +type acceptingListener struct { + net.Listener + accepts atomic.Int64 + mu sync.Mutex + conns []net.Conn +} + +func newAcceptingListener(t *testing.T) *acceptingListener { + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + a := &acceptingListener{Listener: l} + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + a.mu.Lock() + a.conns = append(a.conns, c) + a.mu.Unlock() + a.accepts.Add(1) + } + }() + t.Cleanup(func() { + _ = l.Close() + a.mu.Lock() + defer a.mu.Unlock() + for _, c := range a.conns { + _ = c.Close() + } + }) + return a +} + +func (a *acceptingListener) uri(t *testing.T) *rpc.FMPURI { + uri, err := rpc.ParseFMPURI("fmprpc://" + a.Addr().String()) + require.NoError(t, err) + return uri +} + +// requireStale waits until the handler holds a connection that is not +// connected: with nobody logged in, OnConnect fails with an auth error the +// connection does not retry on its own. +func requireStale(t *testing.T, h *gregorHandler, a *acceptingListener, accepts int64) { + t.Helper() + require.Eventually(t, func() bool { + return a.accepts.Load() >= accepts && hasConn(h) && !h.IsConnected() + }, 10*time.Second, time.Millisecond, "connection did not fail") +} + +// After a terminal connect failure, the ping loop's pings redial at the ping +// interval, without tearing the connection down and without spinning. +func TestGregorHandlerTerminalFailureRedialsOnPing(t *testing.T) { + t.Setenv("KEYBASE_PUSH_PING_INTERVAL", "100ms") + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + a := newAcceptingListener(t) + + h := newGregorHandler(g) + defer h.Shutdown(context.Background()) + require.NoError(t, h.connGate.connect(context.Background(), a.uri(t), false)) + requireStale(t, h, a, 1) + // Bounds on elapsed time rather than on a count in a fixed window, so a + // slow machine can only make this take longer. + start, began := a.accepts.Load(), time.Now() + require.Eventually(t, func() bool { return a.accepts.Load()-start >= 3 }, 10*time.Second, time.Millisecond, + "ping loop did not redial a failed connection") + elapsed := time.Since(began) + t.Logf("3 redials in %v", elapsed) + require.GreaterOrEqual(t, elapsed, 200*time.Millisecond, "redialing faster than the ping interval") + require.True(t, hasConn(h), "failed connection was torn down") +} + +// A transition to FOREGROUND redials a failed connection right away instead +// of waiting for the next ping. +func TestGregorHandlerTerminalFailureRedialsOnForeground(t *testing.T) { + t.Setenv("KEYBASE_PUSH_PING_INTERVAL", "1h") + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = chat.NewSyncer(g) + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + a := newAcceptingListener(t) + + h := newGregorHandler(g) + h.Init() + defer h.Shutdown(context.Background()) + require.NoError(t, h.connGate.connect(context.Background(), a.uri(t), false)) + requireStale(t, h, a, 1) + time.Sleep(200 * time.Millisecond) + require.EqualValues(t, 1, a.accepts.Load()) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + require.Eventually(t, func() bool { return a.accepts.Load() >= 2 }, 10*time.Second, time.Millisecond, + "FOREGROUND did not redial a failed connection") + require.True(t, hasConn(h), "failed connection was torn down") +} + +// fakeSyncer ignores a Connected whose ctx is cancelled, as chat.Syncer does. +type fakeSyncer struct { + types.Syncer + mu sync.Mutex + connected bool + connects int + // onConnected, if set, runs once inside Connected, after the syncer is + // marked connected, as a logout landing during the sync would. + onConnected func() + // onDisconnected, if set, runs once inside Disconnected, after the + // syncer is marked disconnected. + onDisconnected func() +} + +func (s *fakeSyncer) IsConnected(context.Context) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.connected +} + +func (s *fakeSyncer) Connected(ctx context.Context, _ chat1.RemoteInterface, _ gregor1.UID, _ *chat1.SyncChatRes) error { + s.mu.Lock() + if err := ctx.Err(); err != nil { + s.mu.Unlock() + return err + } + s.connected = true + s.connects++ + f := s.onConnected + s.onConnected = nil + s.mu.Unlock() + if f != nil { + f() + } + return nil +} + +func (s *fakeSyncer) Disconnected(context.Context) { + s.mu.Lock() + s.connected = false + f := s.onDisconnected + s.onDisconnected = nil + s.mu.Unlock() + if f != nil { + f() + } +} + +func (s *fakeSyncer) connectCalls() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.connects +} + +type fakeBadger struct { + mu sync.Mutex + pushes int + // onPush, if set, runs once inside a push. + onPush func() +} + +func (b *fakeBadger) push() { + b.mu.Lock() + b.pushes++ + f := b.onPush + b.onPush = nil + b.mu.Unlock() + if f != nil { + f() + } +} + +func (b *fakeBadger) PushState(context.Context, gregor.State) { b.push() } +func (b *fakeBadger) PushChatFullUpdate(context.Context, chat1.UnreadUpdateFull) { b.push() } + +func (b *fakeBadger) count() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.pushes +} + +func currentConn(h *gregorHandler) *rpc.Connection { + h.connMutex.Lock() + defer h.connMutex.Unlock() + return h.conn +} + +// onConnectCtx returns the ctx OnConnect derives for h's current connection. +func onConnectCtx(t *testing.T, h *gregorHandler) context.Context { + ctx, cancel, err := h.onConnectCtx(context.Background(), currentConn(h)) + require.NoError(t, err) + t.Cleanup(cancel) + return ctx +} + +type onConnectTailTest struct { + h *gregorHandler + ctx context.Context + gcli *grclient.Client + syncer *fakeSyncer + badger *fakeBadger + uid gregor1.UID + syncRes chat1.SyncAllResult +} + +// setupOnConnectTail builds a handler with a current connection and a +// gregor client, as OnConnect has them once SyncAll has returned. +func setupOnConnectTail(t *testing.T) *onConnectTailTest { + tc, g := setupGregorTest(t) + t.Cleanup(tc.Cleanup) + syncer := &fakeSyncer{} + g.Syncer = syncer + h := newGregorHandler(g) + badger := &fakeBadger{} + h.badger = badger + require.NoError(t, h.connGate.connect(context.Background(), closedPortURI(t), false)) + t.Cleanup(func() { h.Shutdown(context.Background()) }) + ctx := onConnectCtx(t, h) + uid := gregor1.UID(make([]byte, 16)) + gcli, err := h.resetGregorClient(ctx, uid, gregor1.DeviceID(make([]byte, 16))) + require.NoError(t, err) + return &onConnectTailTest{ + h: h, ctx: ctx, gcli: gcli, syncer: syncer, badger: badger, uid: uid, + syncRes: chat1.SyncAllResult{Notification: chat1.NewSyncAllNotificationResWithState(gregor1.State{})}, + } +} + +func (c *onConnectTailTest) run(ctx context.Context) error { + return c.h.onConnectSynced(ctx, chat1.RemoteClient{}, nil, c.uid, c.gcli, c.syncRes) +} + +func TestGregorOnConnectTailApplies(t *testing.T) { + c := setupOnConnectTail(t) + require.NoError(t, c.run(c.ctx)) + require.Equal(t, 2, c.badger.count()) + require.Len(t, c.h.replayCh, 1) + require.True(t, c.syncer.IsConnected(context.Background())) + require.False(t, c.h.isFirstConnect()) + require.False(t, c.h.connectedSince().IsZero()) +} + +// A tail whose connection a logout has shut down applies nothing. +func TestGregorOnConnectTailAfterLogout(t *testing.T) { + c := setupOnConnectTail(t) + require.NoError(t, c.h.Disconnect()) + require.ErrorIs(t, c.run(c.ctx), chat.ErrDuplicateConnection) + require.Zero(t, c.badger.count(), "badges pushed after logout") + require.Zero(t, c.syncer.connectCalls(), "chat sync ran after logout") + require.Empty(t, c.h.replayCh, "gregor state sync ran after logout") + require.True(t, c.h.isFirstConnect(), "first connect cleared after logout") + require.True(t, c.h.connectedSince().IsZero(), "connected time set after logout") +} + +// A logout during the chat sync leaves the syncer disconnected and stops the +// rest of the tail. +func TestGregorOnConnectLogoutDuringChatSync(t *testing.T) { + c := setupOnConnectTail(t) + c.syncer.onConnected = func() { require.NoError(t, c.h.Disconnect()) } + require.ErrorIs(t, c.run(c.ctx), chat.ErrDuplicateConnection) + require.False(t, c.syncer.IsConnected(context.Background()), "syncer left connected after logout") + require.Equal(t, 1, c.badger.count(), "badges pushed after logout") + require.Empty(t, c.h.replayCh, "gregor state sync ran after logout") + require.True(t, c.h.isFirstConnect(), "first connect cleared after logout") + require.True(t, c.h.connectedSince().IsZero(), "connected time set after logout") +} + +// A Shutdown that lands between the gregor badge push and the connected +// step leaves first connect and the connected time alone. A real Shutdown +// can't land during the push, which holds the gate, so the push cancels the +// connection's ctx as that Shutdown would. +func TestGregorOnConnectShutdownBeforeConnectedStep(t *testing.T) { + c := setupOnConnectTail(t) + c.badger.onPush = func() { + c.badger.mu.Lock() + defer c.badger.mu.Unlock() + c.badger.onPush = func() { + c.h.connMutex.Lock() + defer c.h.connMutex.Unlock() + c.h.connCancel() + } + } + require.ErrorIs(t, c.run(c.ctx), chat.ErrDuplicateConnection) + require.Equal(t, 2, c.badger.count()) + require.True(t, c.h.isFirstConnect(), "first connect cleared after shutdown") + require.True(t, c.h.connectedSince().IsZero(), "connected time set after shutdown") +} + +// Shutdown cancels OnConnect's ctx before it marks the syncer disconnected, +// so a Syncer.Connected from that OnConnect landing just after the mark is +// ignored rather than leaving the syncer connected. +func TestGregorShutdownCancelsBeforeSyncerDisconnected(t *testing.T) { + c := setupOnConnectTail(t) + c.syncer.onDisconnected = func() { + _ = c.syncer.Connected(c.ctx, chat1.RemoteClient{}, c.uid, &chat1.SyncChatRes{}) + } + require.NoError(t, c.h.Disconnect()) + require.False(t, c.syncer.IsConnected(context.Background()), "syncer connected after shutdown") +} + +// A logout can't finish while a badge push for the old connection is in +// progress, so the push can't land after the logout. +func TestGregorOnConnectBadgePushHoldsOffLogout(t *testing.T) { + c := setupOnConnectTail(t) + logoutDone := make(chan struct{}) + c.badger.onPush = func() { + go func() { + defer close(logoutDone) + _ = c.h.Disconnect() + }() + select { + case <-logoutDone: + t.Error("logout finished during a badge push") + case <-time.After(100 * time.Millisecond): + } + } + require.ErrorIs(t, c.run(c.ctx), chat.ErrDuplicateConnection) + <-logoutDone + require.Equal(t, 1, c.badger.count()) +} + +// reinstall makes conn the current connection again, as connectNow would, +// so the next tail run is not short-circuited by the logout before it. No +// dial is involved, so no connection callback races this. +func (c *onConnectTailTest) reinstall(conn *rpc.Connection) { + c.h.connMutex.Lock() + defer c.h.connMutex.Unlock() + c.h.conn = conn + c.h.connCtx, c.h.connCancel = context.WithCancel(context.Background()) +} + +// OnConnect's tail, a logout and app state transitions all run under the +// connection gate. Racing them must not deadlock, and a logout must still +// leave gregor down. +func TestGregorOnConnectTailStress(t *testing.T) { + c := setupOnConnectTail(t) + // Swap in a connection that never dials. This test puts the current + // connection back after each logout, and a dialing one would reconnect + // behind it and outlive the test. + require.NoError(t, c.h.Disconnect()) + conn := &rpc.Connection{} + c.reinstall(conn) + c.h.connGate.start() + t.Cleanup(c.h.connGate.stop) + + stop := make(chan struct{}) + var tails, writers sync.WaitGroup + for range 2 { + tails.Add(1) + go func() { + defer tails.Done() + for { + select { + case <-stop: + return + default: + } + c.reinstall(conn) + // Another tail's logout can land before the ctx is derived. + if ctx, cancel, err := c.h.onConnectCtx(context.Background(), conn); err == nil { + _ = c.run(ctx) + cancel() + } + // A run queues at most one replay, and Init's replay thread + // is not running here to take it off. + select { + case <-c.h.replayCh: + default: + } + _ = c.h.Disconnect() + runtime.Gosched() + } + }() + } + for w := range 2 { + writers.Add(1) + go func() { + defer writers.Done() + rng := rand.New(rand.NewSource(int64(w))) + for range 500 { + c.h.G().MobileAppState.Update(allAppStates[rng.Intn(len(allAppStates))]) + if rng.Intn(4) == 0 { + time.Sleep(time.Duration(rng.Intn(200)) * time.Microsecond) + } + } + }() + } + + done := make(chan struct{}) + go func() { + writers.Wait() + close(stop) + tails.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(60 * time.Second): + // A deadlock leaves the racers holding the handler's locks, so the + // cleanup that shuts the handler down never returns. Nothing this + // test writes is flushed through that hung unwind, neither t.Fatal's + // message nor a panic's, so say it on stderr first; go test's own + // timeout then dumps every stack. + const msg = "deadlock: the connect tail, logouts and transitions did not finish" + fmt.Fprintln(os.Stderr, msg) + t.Fatal(msg) + } + + require.NoError(t, c.h.Disconnect()) + require.False(t, hasConn(c.h), "logout left a connection after settling") + c.h.G().MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + c.h.connGate.reconcile(context.Background()) + require.False(t, hasConn(c.h), "reconnected after a logout") +} + +type failingRPCClient struct{} + +func (failingRPCClient) Call(context.Context, string, any, any, time.Duration) error { + return errors.New("no server") +} + +func (failingRPCClient) CallCompressed(context.Context, string, any, any, rpc.CompressionType, time.Duration) error { + return errors.New("no server") +} + +func (failingRPCClient) Notify(context.Context, string, any, time.Duration) error { + return errors.New("no server") +} + +// An OnConnect for a connection that is no longer current, because it shut +// down before OnConnect started or while it ran, or because a newer +// connection replaced it, installs no gregor client, leaves the chat syncer +// alone, and fails with an error the connection does not retry. The rpc +// library hands a replaced connection's OnConnect a live ctx. +func TestGregorOnConnectAfterShutdownInstallsNothing(t *testing.T) { + for _, tt := range []struct { + name string + // before runs before OnConnect, during inside it, after the + // connection check. + before, during func(t *testing.T, h *gregorHandler, uri *rpc.FMPURI) + }{ + {name: "before", before: func(t *testing.T, h *gregorHandler, _ *rpc.FMPURI) { + require.NoError(t, h.Disconnect()) + }}, + {name: "during", during: func(t *testing.T, h *gregorHandler, _ *rpc.FMPURI) { + require.NoError(t, h.Disconnect()) + }}, + {name: "replaced", before: func(t *testing.T, h *gregorHandler, uri *rpc.FMPURI) { + require.NoError(t, h.Disconnect()) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + }}, + } { + t.Run(tt.name, func(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + syncer := &fakeSyncer{} + g.Syncer = syncer + h := newGregorHandler(g) + uri := closedPortURI(t) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + defer h.Shutdown(context.Background()) + conn := currentConn(h) + + h.authParamsForTest = func(context.Context) (gregor1.UID, gregor1.DeviceID, gregor1.SessionToken, *libkb.NIST, error) { + if tt.during != nil { + tt.during(t, h, uri) + } + return gregor1.UID(make([]byte, 16)), gregor1.DeviceID(make([]byte, 16)), "", nil, nil + } + if tt.before != nil { + tt.before(t, h, uri) + } + + local, remote := net.Pipe() + defer remote.Close() + xp := rpc.NewTransport(local, libkb.NewRPCLogFactory(tc.G), tc.G.RemoteNetworkInstrumenterStorage, + libkb.MakeWrapError(tc.G), rpc.DefaultMaxFrameLength) + defer xp.Close() + srv := rpc.NewServer(xp, libkb.MakeWrapError(tc.G)) + + err := h.OnConnect(context.Background(), conn, failingRPCClient{}, srv) + require.ErrorIs(t, err, chat.ErrDuplicateConnection) + require.False(t, h.ShouldRetryOnConnect(err), "retrying a connection that is not current") + _, err = h.getGregorCli() + require.Error(t, err, "installed a client for a connection that is not current") + require.Zero(t, syncer.connectCalls(), "chat sync ran for a connection that is not current") + }) + } +} + +// syncAllRecorder fails every call, recording the host of each SyncAll. +type syncAllRecorder struct { + failingRPCClient + mu sync.Mutex + hosts []string +} + +func (r *syncAllRecorder) CallCompressed(_ context.Context, _ string, arg any, _ any, _ rpc.CompressionType, _ time.Duration) error { + if args, ok := arg.([]any); ok && len(args) == 1 { + if sa, ok := args[0].(chat1.SyncAllArg); ok { + r.mu.Lock() + r.hosts = append(r.hosts, sa.HostName) + r.mu.Unlock() + } + } + return errors.New("no server") +} + +// OnConnect sends the host of the uri the gate connected to. +func TestGregorOnConnectSyncAllHost(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + g.Syncer = &fakeSyncer{} + h := newGregorHandler(g) + uri := closedPortURI(t) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) + defer h.Shutdown(context.Background()) + h.authParamsForTest = func(context.Context) (gregor1.UID, gregor1.DeviceID, gregor1.SessionToken, *libkb.NIST, error) { + return gregor1.UID(make([]byte, 16)), gregor1.DeviceID(make([]byte, 16)), "", nil, nil + } + + local, remote := net.Pipe() + defer remote.Close() + xp := rpc.NewTransport(local, libkb.NewRPCLogFactory(tc.G), tc.G.RemoteNetworkInstrumenterStorage, + libkb.MakeWrapError(tc.G), rpc.DefaultMaxFrameLength) + defer xp.Close() + srv := rpc.NewServer(xp, libkb.MakeWrapError(tc.G)) + + rec := &syncAllRecorder{} + err := h.OnConnect(context.Background(), currentConn(h), rec, srv) + require.ErrorContains(t, err, "error running SyncAll") + rec.mu.Lock() + defer rec.mu.Unlock() + require.Equal(t, []string{uri.Host}, rec.hosts) +} + +// A reconnect for a connection that has since been shut down does nothing: +// run directly, or queued before the replacement and run by the monitor. +func TestGregorConnReconnectSkipsReplacedConn(t *testing.T) { + c := setupGregorConn(t, keybase1.MobileAppState_FOREGROUND) + c.waitMonitor(t) + require.NoError(t, c.gate.connect(context.Background(), testGregorURI(t, "gregord.test"), false)) + want := fakeGregorCounts{up: true, connects: 1} + + replaced, cancel := context.WithCancel(context.Background()) + cancel() + c.gate.reconnect(context.Background(), replaced) + c.gate.reconnect(context.Background(), nil) + require.Equal(t, want, c.conn.counts(), "reconnected a replaced connection") + + queued, cancelQueued := context.WithCancel(context.Background()) + c.gate.mu.Lock() + c.gate.requestReconnect(context.Background(), queued) + cancelQueued() + c.gate.mu.Unlock() + require.Eventually(t, func() bool { return len(c.gate.reconnectCh) == 0 }, 10*time.Second, time.Millisecond) + c.waitMonitor(t) + require.Equal(t, want, c.conn.counts(), "a queued reconnect tore down a newer connection") + + // A request for a replaced connection does not displace a pending one + // for the live connection. + c.gate.mu.Lock() + c.gate.requestReconnect(context.Background(), context.Background()) + c.gate.requestReconnect(context.Background(), replaced) + c.gate.mu.Unlock() + want = fakeGregorCounts{up: true, connects: 2, shutdowns: 1} + require.Eventually(t, func() bool { return c.conn.counts() == want }, 10*time.Second, time.Millisecond, + "the live connection's reconnect was dropped") +} + +// OnDisconnected from a connection that has been shut down, as a reconnect +// loop started on it after Shutdown reports, leaves the current connection's +// state alone. +func TestGregorOnDisconnectedIgnoresReplacedConn(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + syncer := &fakeSyncer{} + g.Syncer = syncer + h := newGregorHandler(g) + // No real connection, whose own reconnect loop would report too. + h.connCtx, h.connCancel = context.WithCancel(context.Background()) + defer h.connCancel() + markConnected := func() { + syncer.mu.Lock() + syncer.connected = true + syncer.mu.Unlock() + h.setConnectedAt(time.Now()) + } + + replaced, cancel := context.WithCancel(context.Background()) + cancel() + markConnected() + (&gregorConnHandler{gregorHandler: h, connCtx: replaced}).OnDisconnected(context.Background(), + rpc.StartingNonFirstConnection) + require.True(t, syncer.IsConnected(context.Background()), "a replaced connection marked the syncer offline") + require.False(t, h.connectedSince().IsZero(), "a replaced connection cleared connectedAt") + + (&gregorConnHandler{gregorHandler: h, connCtx: h.currentConnCtx()}).OnDisconnected(context.Background(), + rpc.StartingNonFirstConnection) + require.False(t, syncer.IsConnected(context.Background()), "the current connection did not mark the syncer offline") + require.True(t, h.connectedSince().IsZero(), "the current connection did not clear connectedAt") +} + +// A replay queued for a user who has since logged out does not run. +func TestGregorReplaySkipsLoggedOutUser(t *testing.T) { + tc, g := setupGregorTest(t) + defer tc.Cleanup() + h := newGregorHandler(g) + h.testingEvents = newTestingEvents() + go h.syncReplayThread() + defer close(h.replayCh) + + h.replayCh <- replayThreadArg{ctx: context.Background(), uid: gregor1.UID(make([]byte, 16))} + select { + case res := <-h.testingEvents.replayThreadCh: + require.NoError(t, res.err, "replayed for a logged-out user") + require.Empty(t, res.replayed) + case <-time.After(10 * time.Second): + t.Fatal("replay thread did not report") + } +} diff --git a/go/service/gregor_test.go b/go/service/gregor_test.go index f90d63e17956..de23fe4e4893 100644 --- a/go/service/gregor_test.go +++ b/go/service/gregor_test.go @@ -725,7 +725,8 @@ func TestGregorBadgesIBM(t *testing.T) { // Set up client and server h, server, uid := setupSyncTests(t, g) defer h.Shutdown(context.Background()) - h.badger = badges.NewBadger(tc.G) + badger := badges.NewBadger(tc.G) + h.badger = badger t.Logf("client setup complete") t.Logf("server message") @@ -745,7 +746,7 @@ func TestGregorBadgesIBM(t *testing.T) { ri := func() chat1.RemoteInterface { return dummyRemoteClient{RemoteClient: chat1.RemoteClient{Cli: h.cli}} } - badgerResync(context.TODO(), t, h.badger, ri, h.gregorCli) + badgerResync(context.TODO(), t, badger, ri, h.gregorCli) listener.getBadgeState(t) // skip one since resync sends 2 bs := listener.getBadgeState(t) @@ -760,7 +761,7 @@ func TestGregorBadgesIBM(t *testing.T) { require.NoError(t, err) t.Logf("client sync complete") - badgerResync(context.TODO(), t, h.badger, ri, h.gregorCli) + badgerResync(context.TODO(), t, badger, ri, h.gregorCli) bs = listener.getBadgeState(t) require.Equal(t, 1, bs.NewTlfs, "no more badges") @@ -776,7 +777,8 @@ func TestGregorTeamBadges(t *testing.T) { // Set up client and server h, server, uid := setupSyncTests(t, g) defer h.Shutdown(context.Background()) - h.badger = badges.NewBadger(tc.G) + badger := badges.NewBadger(tc.G) + h.badger = badger t.Logf("client setup complete") t.Logf("server message") @@ -798,7 +800,7 @@ func TestGregorTeamBadges(t *testing.T) { ri := func() chat1.RemoteInterface { return dummyRemoteClient{RemoteClient: chat1.RemoteClient{Cli: h.cli}} } - badgerResync(context.TODO(), t, h.badger, ri, h.gregorCli) + badgerResync(context.TODO(), t, badger, ri, h.gregorCli) listener.getBadgeState(t) // skip one since resync sends 2 bs := listener.getBadgeState(t) @@ -823,18 +825,19 @@ func TestGregorBadgesOOBM(t *testing.T) { // Set up client and server h, _, _ := setupSyncTests(t, g) defer h.Shutdown(context.Background()) - h.badger = badges.NewBadger(tc.G) + badger := badges.NewBadger(tc.G) + h.badger = badger t.Logf("client setup complete") t.Logf("sending first chat update") - h.badger.PushChatUpdate(context.TODO(), chat1.UnreadUpdate{ + badger.PushChatUpdate(context.TODO(), chat1.UnreadUpdate{ ConvID: chat1.ConversationID(`a`), UnreadMessages: 2, }, 0) _ = listener.getBadgeState(t) t.Logf("sending second chat update") - h.badger.PushChatUpdate(context.TODO(), chat1.UnreadUpdate{ + badger.PushChatUpdate(context.TODO(), chat1.UnreadUpdate{ ConvID: chat1.ConversationID(`b`), UnreadMessages: 2, }, 1) @@ -845,7 +848,7 @@ func TestGregorBadgesOOBM(t *testing.T) { t.Logf("resyncing") // Instead of calling badger.Resync, reach in and twiddle the knobs. - h.badger.State().UpdateWithChatFull(context.TODO(), chat1.UnreadUpdateFull{ + badger.State().UpdateWithChatFull(context.TODO(), chat1.UnreadUpdateFull{ InboxVers: chat1.InboxVers(4), Updates: []chat1.UnreadUpdate{ {ConvID: chat1.ConversationID(`b`), UnreadMessages: 0}, @@ -853,14 +856,14 @@ func TestGregorBadgesOOBM(t *testing.T) { }, InboxSyncStatus: chat1.SyncInboxResType_CLEAR, }, false) - err := h.badger.Send(context.TODO()) + err := badger.Send(context.TODO()) require.NoError(t, err) bs = listener.getBadgeState(t) require.Equal(t, 1, badgeStateStats(bs).UnreadChatConversations, "unread chat convs") require.Equal(t, 3, badgeStateStats(bs).UnreadChatMessages, "unread chat messages") t.Logf("clearing") - h.badger.Clear(context.TODO()) + badger.Clear(context.TODO()) bs = listener.getBadgeState(t) require.Equal(t, 0, badgeStateStats(bs).UnreadChatConversations, "unread chat convs") require.Equal(t, 0, badgeStateStats(bs).UnreadChatMessages, "unread chat messages") diff --git a/go/service/main.go b/go/service/main.go index e780fb65be58..8d64373ca539 100644 --- a/go/service/main.go +++ b/go/service/main.go @@ -1032,7 +1032,7 @@ func (d *Service) OnLogout(m libkb.MetaContext) (err error) { log("shutting down gregor") if d.gregor != nil { - _ = d.gregor.Reset() + _ = d.gregor.Disconnect() } log("shutting down rekeyMaster") @@ -1071,16 +1071,9 @@ func (d *Service) gregordConnect() (err error) { } d.G().Log.Debug("| gregor URI: %s", uri) - // If we are already connected, then shutdown and reset the gregor - // handler - if d.gregor.IsConnected() { - if err := d.gregor.Reset(); err != nil { - return err - } - } - - // Connect to gregord - return d.gregor.Connect(uri) + // Reset an existing connection so it authenticates again. Nothing + // connects while the app is in BACKGROUND or the desktop is suspended. + return d.gregor.ConnectFresh(uri) } // ReleaseLock releases the locking pidfile by closing, unlocking and diff --git a/go/service/reachability.go b/go/service/reachability.go index cd651df431d0..4679da325cd8 100644 --- a/go/service/reachability.go +++ b/go/service/reachability.go @@ -80,7 +80,7 @@ func (h *reachability) setReachability(r keybase1.Reachability) { } func (h *reachability) check(ctx context.Context) (k keybase1.Reachability) { - reachable := h.gh.isReachable(ctx) + reachable := h.gh.isReachable(ctx, h.gh.currentConnCtx()) if reachable { k.Reachable = keybase1.Reachable_YES } else { diff --git a/go/service/rpc.go b/go/service/rpc.go index 15c9994e64db..d0c9838d6c89 100644 --- a/go/service/rpc.go +++ b/go/service/rpc.go @@ -3,6 +3,7 @@ package service import ( "context" "net" + "sync" "github.com/keybase/client/go/libkb" "github.com/keybase/go-framed-msgpack-rpc/rpc" @@ -11,7 +12,11 @@ import ( // connTransport implements rpc.ConnectionTransport type connTransport struct { libkb.Contextified - host string + host string + + // mu guards the fields below: the connection dials on its own goroutine + // while Shutdown closes the transport. + mu sync.Mutex conn net.Conn transport rpc.Transporter stagedTransport rpc.Transporter @@ -27,44 +32,60 @@ func newConnTransport(g *libkb.GlobalContext, host string) *connTransport { } func (t *connTransport) Dial(context.Context) (rpc.Transporter, error) { - var err error - t.conn, err = libkb.ProxyDial(t.G().Env, "tcp", t.host) + conn, err := libkb.ProxyDial(t.G().Env, "tcp", t.host) if err != nil { return nil, err } - t.stagedTransport = rpc.NewTransport(t.conn, libkb.NewRPCLogFactory(t.G()), + transport := rpc.NewTransport(conn, libkb.NewRPCLogFactory(t.G()), t.G().RemoteNetworkInstrumenterStorage, libkb.MakeWrapError(t.G()), rpc.DefaultMaxFrameLength) - return t.stagedTransport, nil + t.mu.Lock() + defer t.mu.Unlock() + t.conn = conn + t.stagedTransport = transport + return transport, nil } func (t *connTransport) IsConnected() bool { - return t.transport != nil && t.transport.IsConnected() + t.mu.Lock() + transport := t.transport + t.mu.Unlock() + return transport != nil && transport.IsConnected() } +// Finalize and Close close transports outside mu, because closing blocks until +// the transport's loops stop and IsConnected should not wait on that. func (t *connTransport) Finalize() { - if t.transport != nil { - t.transport.Close() - } + t.mu.Lock() + old := t.transport t.transport = t.stagedTransport t.stagedTransport = nil + t.mu.Unlock() + if old != nil { + old.Close() + } } func (t *connTransport) Close() { - if t.conn != nil { - t.conn.Close() + t.mu.Lock() + conn, transport, staged := t.conn, t.transport, t.stagedTransport + t.transport = nil + t.stagedTransport = nil + t.mu.Unlock() + if conn != nil { + conn.Close() } - if t.transport != nil { - t.transport.Close() + if transport != nil { + transport.Close() } - t.transport = nil - if t.stagedTransport != nil { - t.stagedTransport.Close() + if staged != nil { + staged.Close() } - t.stagedTransport = nil } func (t *connTransport) Reset() { + t.mu.Lock() + defer t.mu.Unlock() t.transport = nil t.stagedTransport = nil }