From ff069425b4c77b989faf98aaeee6f1414e55df81 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Mon, 21 Sep 2026 10:03:22 -0400 Subject: [PATCH 1/5] refactor(gregor): connection identity, own ctx, and app-state gating The gregor handler had no way to tell one connection from the next. A late callback from a torn-down connection -- OnConnect finishing its tail, a Syncer.Connected landing after Disconnected, a reconnect fired by a connection that had already shut down -- would act on whatever connection happened to be current. Shutdown used the caller's ctx, so work started by a connection outlived it. Every connection now has an identity and a ctx of its own. OnConnect's post-sync steps run only while their connection is still the current one, and report "no longer current" back to the caller so onConnectSynced can return ErrDuplicateConnection instead of installing state for a dead connection. Shutdown cancels that connection's ctx, so its in-flight work stops with it rather than with whoever called Shutdown. Syncer.Connected bails when its ctx is already cancelled, so it cannot mark the syncer connected after the matching Disconnected. Connecting is gated on app state in one place, gregorConnGate: only BACKGROUND, or a desktop suspend, takes the connection down; INACTIVE keeps it up. A BACKGROUND that lands after a connect has read the state wakes the monitor, which waits for that connect to finish before taking the connection down, so a connect and a state change cannot cross. The service asks for ConnectFresh(uri) rather than doing its own IsConnected()/Reset() dance, and OnLogout disconnects instead of resetting. ServerConnection.Reconnect no longer returns anything: callers never waited for the result, and the value it returned described a connection that could already be gone. connTransport guards conn/transport/stagedTransport with a mutex -- the connection dials on its own goroutine while Shutdown closes the transport. --- go/chat/deliverer.go | 4 +- go/chat/server.go | 6 +- go/chat/server_test.go | 4 +- go/chat/sync.go | 7 + go/chat/sync_test.go | 15 + go/chat/types/interfaces.go | 3 +- go/kbtest/chat.go | 4 +- go/service/gregor.go | 441 ++++++----- go/service/gregor_conn.go | 212 ++++++ go/service/gregor_conn_test.go | 1300 ++++++++++++++++++++++++++++++++ go/service/gregor_test.go | 25 +- go/service/main.go | 15 +- go/service/rpc.go | 55 +- 13 files changed, 1838 insertions(+), 253 deletions(-) create mode 100644 go/service/gregor_conn.go create mode 100644 go/service/gregor_conn_test.go 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/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..7484ee6836db 100644 --- a/go/service/gregor.go +++ b/go/service/gregor.go @@ -196,14 +196,23 @@ 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 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 +230,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 +268,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 +278,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 +340,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 +360,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 +419,19 @@ func (g *gregorHandler) setReachability(r *reachability) { g.reachability = r } -func (g *gregorHandler) Connect(uri *rpc.FMPURI) (err error) { +// Connect connects to uri unless the app is in BACKGROUND or the desktop is +// suspended, in which case it connects once that ends. +func (g *gregorHandler) Connect(uri *rpc.FMPURI) error { + return g.connGate.connect(libkb.WithLogTag(context.Background(), "GRGRCONN"), uri, false) +} + +// ConnectFresh is Connect, resetting any existing connection first so it +// authenticates again. +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 +446,32 @@ 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) + var conn *rpc.Connection if uri.UseTLS() { - err = g.connectTLS(ctx) + conn, err = g.connectTLS(ctx, uri) + if err != nil { + return err + } } else { - err = g.connectNoTLS(ctx) + conn = g.connectNoTLS(ctx, uri) } + g.conn = conn + g.connCtx, g.connCancel = context.WithCancel(context.Background()) - 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.Done()) + 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 } } @@ -768,25 +776,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 +819,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 +840,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.onGateIfCurrent(ctx, func() { uri = g.connGate.uri }) { + return chat.ErrDuplicateConnection + } g.chatLog.Debug(ctx, "OnConnect begin") syncAllRes, err := chatCli.SyncAll(ctx, chat1.SyncAllArg{ Uid: uid, @@ -819,7 +855,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 +877,31 @@ 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) +} + +// onGateIfCurrent runs f under the connection gate if OnConnect's ctx is +// still live, and reports whether it ran. Every Shutdown and Reset is made +// under the gate 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: its mutex is not reentrant. +func (g *gregorHandler) onGateIfCurrent(ctx context.Context, f func()) bool { + g.connGate.mu.Lock() + defer g.connGate.mu.Unlock() + if ctx.Err() != nil { + return false + } + f() + return true +} + +// 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.onGateIfCurrent(ctx, func() { + 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.onGateIfCurrent(ctx, func() { + 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.onGateIfCurrent(ctx, func() { + 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 } @@ -1355,6 +1435,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 +1448,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 +1482,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 @@ -1463,26 +1555,17 @@ 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.Reconnect(ctx) 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) - } - - didShutdown = false - g.chatLog.Debug(ctx, "Reconnect: skipping reconnect, already disconnected") - return didShutdown, nil +// Reconnect drops a live connection and connects again when the app state +// allows it, without waiting for either. +func (g *gregorHandler) Reconnect(ctx context.Context) { + g.connGate.requestReconnect(ctx) } func (g *gregorHandler) forcePing(ctx context.Context) { @@ -1493,7 +1576,14 @@ func (g *gregorHandler) forcePing(ctx context.Context) { } } -func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, shutdownCancel context.CancelFunc) { +func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, done <-chan struct{}) { + 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 + } var err error doneCh := make(chan error) timeout := g.G().Env.GetGregorPingTimeout() @@ -1505,14 +1595,14 @@ 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 { @@ -1525,32 +1615,22 @@ func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, shutdownCancel select { case err = <-doneCh: - case <-g.shutdownCh: + case <-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.Reconnect(ctx) } } } -func (g *gregorHandler) pingLoop(ctx context.Context) { +// pingLoop runs until done, the Done channel of the ctx of the connection it +// was started for, closes. +func (g *gregorHandler) pingLoop(ctx context.Context, done <-chan struct{}) { id, _ := libkb.RandBytes(4) duration := g.G().Env.GetGregorPingInterval() timeout := g.G().Env.GetGregorPingTimeout() @@ -1564,34 +1644,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, id, done) case <-ticker.C: - g.pingOnce(pingCtx, id, shutdownCancel) - case <-g.shutdownCh: + g.pingOnce(pingCtx, id, done) + case <-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) (*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 +1681,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()), 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) *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 +1703,9 @@ func (g *gregorHandler) connectNoTLS(ctx context.Context) error { return backoff.NewConstantBackOff(GregorConnectionRetryInterval) }, } - g.conn = rpc.NewConnectionWithTransport(g, t, + return rpc.NewConnectionWithTransport(g, 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..2b0fe6b89b27 --- /dev/null +++ b/go/service/gregor_conn.go @@ -0,0 +1,212 @@ +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 +} + +// 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 (the handler takes it in +// onGateIfCurrent), 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 reconnect request for the monitor, so a + // burst of requests coalesces. + reconnectCh chan struct{} + + 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, any 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. +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 { + 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() +} + +// requestReconnect asks the monitor to reconnect and returns without waiting. +func (c *gregorConnGate) requestReconnect(ctx context.Context) { + select { + case c.reconnectCh <- struct{}{}: + c.debug(ctx, "Reconnect: requested") + default: + c.debug(ctx, "Reconnect: one is already pending") + } +} + +// reconnect drops a live connection and connects again when reconcile allows +// it. +func (c *gregorConnGate) reconnect(ctx context.Context) { + c.mu.Lock() + defer c.mu.Unlock() + 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) + 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..10c9445c9dc3 --- /dev/null +++ b/go/service/gregor_conn_test.go @@ -0,0 +1,1300 @@ +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) 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, resets: 1}, c.conn.counts()) + + 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") + require.Equal(t, fakeGregorCounts{connects: 1, shutdowns: 1, resets: 2}, c.conn.counts()) + + 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: 3}, 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()) + 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()) + 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()) + c.requireUp(t, false, "reconnect connected in BACKGROUND") + require.Equal(t, fakeGregorCounts{connects: 3, shutdowns: 3}, c.conn.counts()) + + c.gate.reconnect(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()) + } + }() + 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()) + 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) + } + 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.Connect(uri)) + // 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.Connect(uri), "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.Connect(uri)) + 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.Connect(closedPortURI(t))) + _, _, _, _, 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.Connect(uri)) + require.False(t, hasConn(h), "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.Connect(a.uri(t))) + requireStale(t, h, a, 1) + start := a.accepts.Load() + time.Sleep(time.Second) + redials := a.accepts.Load() - start + t.Logf("%d redials in 1s", redials) + require.GreaterOrEqual(t, redials, int64(3), "ping loop did not redial a failed connection") + require.LessOrEqual(t, redials, int64(13), "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.Connect(a.uri(t))) + 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.Connect(closedPortURI(t))) + 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.Connect(uri)) + }}, + } { + 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.Connect(uri)) + 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.Connect(uri)) + 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) +} 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..b2b30309ba7b 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 a live 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/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 } From 2d59e6fa2d44bbcd5070c63cedc0a1818cfd840a Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Tue, 22 Sep 2026 15:00:34 -0400 Subject: [PATCH 2/5] fix(gregor): scope disconnects and reconnects to their connection OnDisconnected and OnConnectError now go through a per-connection handler that ignores events from a connection already shut down. Reconnect requests carry the ctx of the connection they are for and are skipped once it is replaced. pingOnce no longer leaks its goroutine on shutdown, a queued replay is skipped once its user logs out, and the ping redial test bounds elapsed time instead of counting in a fixed window. --- go/service/gregor.go | 126 ++++++++++++++++++++++----------- go/service/gregor_conn.go | 43 ++++++++--- go/service/gregor_conn_test.go | 114 +++++++++++++++++++++++++---- go/service/reachability.go | 2 +- 4 files changed, 220 insertions(+), 65 deletions(-) diff --git a/go/service/gregor.go b/go/service/gregor.go index 7484ee6836db..c25b66dcf85a 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 { @@ -446,17 +447,20 @@ func (g *gregorHandler) connectNow(uri *rpc.FMPURI) (err error) { g.connectHappened = make(chan struct{}) }() + connCtx, connCancel := context.WithCancel(context.Background()) + handler := &gregorConnHandler{gregorHandler: g, connCtx: connCtx} var conn *rpc.Connection if uri.UseTLS() { - conn, err = g.connectTLS(ctx, uri) + conn, err = g.connectTLS(ctx, uri, handler) if err != nil { + connCancel() return err } } else { - conn = g.connectNoTLS(ctx, uri) + conn = g.connectNoTLS(ctx, uri, handler) } g.conn = conn - g.connCtx, g.connCancel = context.WithCancel(context.Background()) + g.connCtx, g.connCancel = connCtx, connCancel // 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 @@ -469,7 +473,7 @@ func (g *gregorHandler) connectNow(uri *rpc.FMPURI) (err error) { // 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.Done()) + go g.pingLoop(ctx, g.connCtx) go g.pushStateNewDataDebouncer(g.connCtx.Done()) return nil } @@ -673,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 { @@ -722,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) @@ -996,34 +1011,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, }) } @@ -1539,7 +1573,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 { @@ -1555,17 +1591,25 @@ func (g *gregorHandler) isReachable(ctx context.Context) bool { } if err != nil { g.chatLog.Debug(ctx, "isReachable: error: terminating connection: %s", err.Error()) - g.Reconnect(ctx) + g.connGate.requestReconnect(ctx, connCtx) return false } return true } -// Reconnect drops a live connection and connects again when the app state -// allows it, without waiting for either. +// 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.connGate.requestReconnect(ctx, g.currentConnCtx()) +} + +// 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) { @@ -1576,7 +1620,7 @@ func (g *gregorHandler) forcePing(ctx context.Context) { } } -func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, done <-chan struct{}) { +func (g *gregorHandler) pingOnce(ctx context.Context, conn context.Context, id []byte) { g.connMutex.Lock() pingCli := g.pingCli g.connMutex.Unlock() @@ -1584,10 +1628,12 @@ func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, done <-chan str g.chatLog.Debug(ctx, "ping loop: id: %x no connection, skipping ping", id) return } - var err error - doneCh := make(chan error) + // 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 @@ -1605,17 +1651,13 @@ func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, done <-chan str _, 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 <-done: + case <-conn.Done(): g.chatLog.Debug(ctx, "ping loop: id: %x shutdown received", id) return } @@ -1623,14 +1665,14 @@ func (g *gregorHandler) pingOnce(ctx context.Context, id []byte, done <-chan str 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") - g.Reconnect(ctx) + g.connGate.requestReconnect(ctx, conn) } } } -// pingLoop runs until done, the Done channel of the ctx of the connection it -// was started for, closes. -func (g *gregorHandler) pingLoop(ctx context.Context, done <-chan struct{}) { +// 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() @@ -1648,10 +1690,10 @@ func (g *gregorHandler) pingLoop(ctx context.Context, done <-chan struct{}) { select { case <-g.forcePingCh: g.chatLog.Debug(pingCtx, "ping loop: forced attempt") - g.pingOnce(pingCtx, id, done) + g.pingOnce(pingCtx, conn, id) case <-ticker.C: - g.pingOnce(pingCtx, id, done) - case <-done: + g.pingOnce(pingCtx, conn, id) + case <-conn.Done(): g.chatLog.Debug(pingCtx, "ping loop: id: %x shutdown received", id) pingCancel() return @@ -1660,7 +1702,7 @@ func (g *gregorHandler) pingLoop(ctx context.Context, done <-chan struct{}) { } } -func (g *gregorHandler) connectTLS(ctx context.Context, uri *rpc.FMPURI) (*rpc.Connection, error) { +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 { @@ -1683,7 +1725,7 @@ func (g *gregorHandler) connectTLS(ctx context.Context, uri *rpc.FMPURI) (*rpc.C } 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, @@ -1691,7 +1733,7 @@ func (g *gregorHandler) connectTLS(ctx context.Context, uri *rpc.FMPURI) (*rpc.C } // connMutex must be locked before calling this -func (g *gregorHandler) connectNoTLS(ctx context.Context, uri *rpc.FMPURI) *rpc.Connection { +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 @@ -1703,7 +1745,7 @@ func (g *gregorHandler) connectNoTLS(ctx context.Context, uri *rpc.FMPURI) *rpc. return backoff.NewConstantBackOff(GregorConnectionRetryInterval) }, } - return rpc.NewConnectionWithTransport(g, t, + return rpc.NewConnectionWithTransport(handler, t, libkb.NewContextifiedErrorUnwrapper(g.G().ExternalG()), logger.LogOutputWithDepthAdder{Logger: g.G().Log}, opts) } diff --git a/go/service/gregor_conn.go b/go/service/gregor_conn.go index 2b0fe6b89b27..9979e424709e 100644 --- a/go/service/gregor_conn.go +++ b/go/service/gregor_conn.go @@ -64,9 +64,13 @@ type gregorConnGate struct { monitorWait <-chan struct{} monitorSuspendWait <-chan struct{} - // reconnectCh holds at most one reconnect request for the monitor, so a - // burst of requests coalesces. - reconnectCh 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 @@ -132,8 +136,16 @@ func (c *gregorConnGate) forget(ctx context.Context) error { return c.conn.Reset() } -// requestReconnect asks the monitor to reconnect and returns without waiting. -func (c *gregorConnGate) requestReconnect(ctx context.Context) { +// 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") @@ -142,11 +154,24 @@ func (c *gregorConnGate) requestReconnect(ctx context.Context) { } } -// reconnect drops a live connection and connects again when reconcile allows -// it. -func (c *gregorConnGate) reconnect(ctx context.Context) { +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 @@ -197,7 +222,7 @@ func (c *gregorConnGate) monitor(ctx context.Context, state keybase1.MobileAppSt case <-next: case <-nextSuspend: case <-c.reconnectCh: - c.reconnect(ctx) + c.reconnect(ctx, c.takeReconnect()) continue case <-c.stopCh: return diff --git a/go/service/gregor_conn_test.go b/go/service/gregor_conn_test.go index 10c9445c9dc3..7d1396f2b8c4 100644 --- a/go/service/gregor_conn_test.go +++ b/go/service/gregor_conn_test.go @@ -298,7 +298,7 @@ func (c *gregorConnTest) requireStaysDown(t *testing.T, why string) { connects := c.conn.counts().connects for _, state := range allAppStates { c.update(t, state) - c.gate.reconnect(context.Background()) + 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) @@ -377,7 +377,7 @@ func TestGregorConnReconnectInBackground(t *testing.T) { uri := testGregorURI(t, "gregord.test") require.NoError(t, c.gate.connect(context.Background(), uri, false)) - c.gate.reconnect(context.Background()) + 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 @@ -387,11 +387,11 @@ func TestGregorConnReconnectInBackground(t *testing.T) { c.gate.mu.Unlock() c.waitMonitor(t) require.NoError(t, c.conn.connectNow(uri)) - c.gate.reconnect(context.Background()) + 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()) + 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()) } @@ -414,7 +414,7 @@ func TestGregorConnReconnectRequestsCoalesce(t *testing.T) { go func() { defer close(done) for range 10 { - gate.requestReconnect(context.Background()) + gate.requestReconnect(context.Background(), context.Background()) } }() select { @@ -468,7 +468,7 @@ func TestGregorReconnectWhileSuspendedDoesNotConnect(t *testing.T) { c.gate.mu.Unlock() c.waitMonitor(t) require.NoError(t, c.conn.connectNow(uri)) - c.gate.reconnect(context.Background()) + 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()) @@ -522,7 +522,7 @@ func TestGregorConnStress(t *testing.T) { case 2: _ = gate.forget(ctx) default: - gate.requestReconnect(ctx) + gate.requestReconnect(ctx, ctx) } runtime.Gosched() } @@ -831,12 +831,14 @@ func TestGregorHandlerTerminalFailureRedialsOnPing(t *testing.T) { defer h.Shutdown(context.Background()) require.NoError(t, h.Connect(a.uri(t))) requireStale(t, h, a, 1) - start := a.accepts.Load() - time.Sleep(time.Second) - redials := a.accepts.Load() - start - t.Logf("%d redials in 1s", redials) - require.GreaterOrEqual(t, redials, int64(3), "ping loop did not redial a failed connection") - require.LessOrEqual(t, redials, int64(13), "redialing faster than the ping interval") + // 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") } @@ -1298,3 +1300,89 @@ func TestGregorOnConnectSyncAllHost(t *testing.T) { 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/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 { From 86d662268c07017bad5028c2b78005cae2c1ba3b Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Tue, 22 Sep 2026 15:02:08 -0400 Subject: [PATCH 3/5] fix(gregor): login resets only an existing connection With no connection, as in BACKGROUND, a reset only dropped the gregor client until the next connect. --- go/service/gregor.go | 6 +++++- go/service/gregor_conn.go | 10 +++++++--- go/service/gregor_conn_test.go | 14 +++++++++++--- go/service/main.go | 4 ++-- 4 files changed, 25 insertions(+), 9 deletions(-) diff --git a/go/service/gregor.go b/go/service/gregor.go index c25b66dcf85a..9f52963fb1ae 100644 --- a/go/service/gregor.go +++ b/go/service/gregor.go @@ -426,7 +426,7 @@ func (g *gregorHandler) Connect(uri *rpc.FMPURI) error { return g.connGate.connect(libkb.WithLogTag(context.Background(), "GRGRCONN"), uri, false) } -// ConnectFresh is Connect, resetting any existing connection first so it +// ConnectFresh is Connect, resetting an existing connection first so it // authenticates again. func (g *gregorHandler) ConnectFresh(uri *rpc.FMPURI) error { return g.connGate.connect(libkb.WithLogTag(context.Background(), "GRGRCONN"), uri, true) @@ -667,6 +667,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() diff --git a/go/service/gregor_conn.go b/go/service/gregor_conn.go index 9979e424709e..71a9a3e0c7be 100644 --- a/go/service/gregor_conn.go +++ b/go/service/gregor_conn.go @@ -18,6 +18,8 @@ type gregorConnector interface { 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 @@ -110,15 +112,17 @@ func (c *gregorConnGate) stop() { c.stopOnce.Do(func() { close(c.stopCh) }) } -// connect connects to uri when reconcile allows it. With reset, any existing +// 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. +// 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 { + if reset && c.conn.connExists() { if err := c.conn.Reset(); err != nil { return err } diff --git a/go/service/gregor_conn_test.go b/go/service/gregor_conn_test.go index 7d1396f2b8c4..458f5a66284b 100644 --- a/go/service/gregor_conn_test.go +++ b/go/service/gregor_conn_test.go @@ -73,6 +73,12 @@ func (f *fakeGregorConn) goStale() { 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() @@ -216,7 +222,7 @@ func TestGregorConnLoginInBackground(t *testing.T) { 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, resets: 1}, c.conn.counts()) + 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") @@ -224,7 +230,9 @@ func TestGregorConnLoginInBackground(t *testing.T) { second := testGregorURI(t, "second.test") require.NoError(t, c.gate.connect(context.Background(), second, true)) c.requireUp(t, false, "login connected in BACKGROUND") - require.Equal(t, fakeGregorCounts{connects: 1, shutdowns: 1, resets: 2}, c.conn.counts()) + // 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") @@ -232,7 +240,7 @@ func TestGregorConnLoginInBackground(t *testing.T) { // 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: 3}, c.conn.counts()) + require.Equal(t, fakeGregorCounts{up: true, connects: 3, shutdowns: 2, resets: 1}, c.conn.counts()) require.Equal(t, first, c.conn.lastURI()) } diff --git a/go/service/main.go b/go/service/main.go index b2b30309ba7b..8d64373ca539 100644 --- a/go/service/main.go +++ b/go/service/main.go @@ -1071,8 +1071,8 @@ func (d *Service) gregordConnect() (err error) { } d.G().Log.Debug("| gregor URI: %s", uri) - // Reset a live connection so it authenticates again. Nothing connects - // while the app is in BACKGROUND or the desktop is suspended. + // 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) } From 3b9e572601a0b1fd1d60f053aad6472273dbb034 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Tue, 22 Sep 2026 15:06:09 -0400 Subject: [PATCH 4/5] fix(httpsrv): keep the attachment server up on INACTIVE Only BACKGROUND stops it, matching gregor. INACTIVE is transient on iOS (control center, the app switcher, an incoming call). --- go/kbhttp/manager/manager.go | 7 +++++-- go/kbhttp/manager/manager_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) create mode 100644 go/kbhttp/manager/manager_test.go 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") +} From 893de52ce8febf2e931e9def9aa92a4cff978dbb Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Tue, 22 Sep 2026 15:06:09 -0400 Subject: [PATCH 5/5] refactor(gregor): keep the gate's lock inside the gate OnConnect's gated steps go through connGate.runIfLive, which also hands over the uri. Drop the non-resetting Connect, which only tests used. --- go/service/gregor.go | 41 +++++++++------------------------- go/service/gregor_conn.go | 20 ++++++++++++++--- go/service/gregor_conn_test.go | 24 ++++++++++---------- 3 files changed, 40 insertions(+), 45 deletions(-) diff --git a/go/service/gregor.go b/go/service/gregor.go index 9f52963fb1ae..aea055614204 100644 --- a/go/service/gregor.go +++ b/go/service/gregor.go @@ -201,9 +201,10 @@ type gregorHandler struct { reachability *reachability chatLog utils.DebugLabeler - // connGate decides when to connect and disconnect, and runs 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 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 @@ -420,14 +421,9 @@ func (g *gregorHandler) setReachability(r *reachability) { g.reachability = r } -// Connect connects to uri unless the app is in BACKGROUND or the desktop is -// suspended, in which case it connects once that ends. -func (g *gregorHandler) Connect(uri *rpc.FMPURI) error { - return g.connGate.connect(libkb.WithLogTag(context.Background(), "GRGRCONN"), uri, false) -} - -// ConnectFresh is Connect, resetting an existing connection first so it -// authenticates again. +// 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) } @@ -862,7 +858,7 @@ func (g *gregorHandler) OnConnect(rpcCtx context.Context, conn *rpc.Connection, // 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.onGateIfCurrent(ctx, func() { uri = g.connGate.uri }) { + if !g.connGate.runIfLive(ctx, func(u *rpc.FMPURI) { uri = u }) { return chat.ErrDuplicateConnection } g.chatLog.Debug(ctx, "OnConnect begin") @@ -899,21 +895,6 @@ func (g *gregorHandler) OnConnect(rpcCtx context.Context, conn *rpc.Connection, return g.onConnectSynced(ctx, chatCli, timeoutCli, uid, gcli, syncAllRes) } -// onGateIfCurrent runs f under the connection gate if OnConnect's ctx is -// still live, and reports whether it ran. Every Shutdown and Reset is made -// under the gate 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: its mutex is not reentrant. -func (g *gregorHandler) onGateIfCurrent(ctx context.Context, f func()) bool { - g.connGate.mu.Lock() - defer g.connGate.mu.Unlock() - if ctx.Err() != nil { - return false - } - f() - return true -} - // 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 @@ -928,7 +909,7 @@ func (g *gregorHandler) onConnectSynced(ctx context.Context, chatCli chat1.Remot // 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.onGateIfCurrent(ctx, func() { + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { if g.badger != nil { g.badger.PushChatFullUpdate(ctx, syncAllRes.Badge) } @@ -960,7 +941,7 @@ func (g *gregorHandler) onConnectSynced(ctx context.Context, chatCli chat1.Remot // Update badging from gregor, and call out to reachability module if we // have one. - if !g.onGateIfCurrent(ctx, func() { + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { if g.badger != nil { state, err := gcli.StateMachineState(ctx, nil, false) if err != nil { @@ -992,7 +973,7 @@ func (g *gregorHandler) onConnectSynced(ctx context.Context, chatCli chat1.Remot }(g.makeReconnectOobm()) // No longer first connect if we are now connected. - if !g.onGateIfCurrent(ctx, func() { + if !g.connGate.runIfLive(ctx, func(*rpc.FMPURI) { g.chatLog.Debug(ctx, "setting first connect to false") g.setFirstConnect(false) g.setConnectedAt(time.Now()) diff --git a/go/service/gregor_conn.go b/go/service/gregor_conn.go index 71a9a3e0c7be..e9df7e074b6b 100644 --- a/go/service/gregor_conn.go +++ b/go/service/gregor_conn.go @@ -36,9 +36,8 @@ type gregorAppState interface { // 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 (the handler takes it in -// onGateIfCurrent), so none of them interleaves with a disconnect, and guards -// the uri OnConnect reads. +// 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, @@ -140,6 +139,21 @@ func (c *gregorConnGate) forget(ctx context.Context) error { 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 diff --git a/go/service/gregor_conn_test.go b/go/service/gregor_conn_test.go index 458f5a66284b..bbae8ea8f69b 100644 --- a/go/service/gregor_conn_test.go +++ b/go/service/gregor_conn_test.go @@ -619,7 +619,7 @@ func TestGregorHandlerConnectRaces(t *testing.T) { } }() for i := range 20 { - require.NoError(t, h.Connect(uri)) + 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()) @@ -647,7 +647,7 @@ func TestGregorHandlerFailedConnectLeavesNothingRunning(t *testing.T) { baseline := runtime.NumGoroutine() for range 20 { - require.ErrorContains(t, h.Connect(uri), "No bundled CA") + require.ErrorContains(t, h.connGate.connect(context.Background(), uri, false), "No bundled CA") h.connGate.reconcile(context.Background()) } require.False(t, hasConn(h)) @@ -669,7 +669,7 @@ func TestGregorHandlerShutdownStopsConnGoroutines(t *testing.T) { baseline := runtime.NumGoroutine() for range 10 { - require.NoError(t, h.Connect(uri)) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) require.True(t, hasConn(h)) h.Shutdown(context.Background()) } @@ -691,7 +691,7 @@ func TestGregorHandlerLoggedInAfterShutdown(t *testing.T) { _, _, _, _, res := h.loggedIn(ctx) require.Equal(t, loggedInNo, res) - require.NoError(t, h.Connect(closedPortURI(t))) + require.NoError(t, h.connGate.connect(context.Background(), closedPortURI(t), false)) _, _, _, _, res = h.loggedIn(ctx) require.Equal(t, loggedInNo, res) h.Shutdown(ctx) @@ -763,8 +763,8 @@ func TestGregorHandlerConnectInBackground(t *testing.T) { h := newGregorHandler(g) uri := closedPortURI(t) - require.NoError(t, h.Connect(uri)) - require.False(t, hasConn(h), "Connect connected in BACKGROUND") + 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") @@ -837,7 +837,7 @@ func TestGregorHandlerTerminalFailureRedialsOnPing(t *testing.T) { h := newGregorHandler(g) defer h.Shutdown(context.Background()) - require.NoError(t, h.Connect(a.uri(t))) + 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. @@ -863,7 +863,7 @@ func TestGregorHandlerTerminalFailureRedialsOnForeground(t *testing.T) { h := newGregorHandler(g) h.Init() defer h.Shutdown(context.Background()) - require.NoError(t, h.Connect(a.uri(t))) + 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()) @@ -989,7 +989,7 @@ func setupOnConnectTail(t *testing.T) *onConnectTailTest { h := newGregorHandler(g) badger := &fakeBadger{} h.badger = badger - require.NoError(t, h.Connect(closedPortURI(t))) + 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)) @@ -1222,7 +1222,7 @@ func TestGregorOnConnectAfterShutdownInstallsNothing(t *testing.T) { }}, {name: "replaced", before: func(t *testing.T, h *gregorHandler, uri *rpc.FMPURI) { require.NoError(t, h.Disconnect()) - require.NoError(t, h.Connect(uri)) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) }}, } { t.Run(tt.name, func(t *testing.T) { @@ -1232,7 +1232,7 @@ func TestGregorOnConnectAfterShutdownInstallsNothing(t *testing.T) { g.Syncer = syncer h := newGregorHandler(g) uri := closedPortURI(t) - require.NoError(t, h.Connect(uri)) + require.NoError(t, h.connGate.connect(context.Background(), uri, false)) defer h.Shutdown(context.Background()) conn := currentConn(h) @@ -1288,7 +1288,7 @@ func TestGregorOnConnectSyncAllHost(t *testing.T) { g.Syncer = &fakeSyncer{} h := newGregorHandler(g) uri := closedPortURI(t) - require.NoError(t, h.Connect(uri)) + 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