refactor(notify): one ordered per-connection clientState stream

go/bind/keybase.go

@@ -432,9 +432,11 @@ func Init(homeDir, mobileSharedHome, logFile, runModeStr string,
 	kbSvc = service.NewService(kbCtx, false)
 	// LoginAttemptNone: the login attempt happens inside RunBackgroundOperations
 	// below, off the Init path. It can block for seconds (leveldb
-	// open/recovery, keychain reads) and Init runs on the native main thread;
-	// GetBootstrapStatus waits for the attempt so the GUI doesn't see a stale
-	// logged-out state.
+	// open/recovery, keychain reads) and Init runs on the native main thread.
+	// The loopback listener is therefore up while the attempt is still running,
+	// so a client can connect and subscribe before there is any session to
+	// report: its first clientState has no session at all in that window, and
+	// the attempt settling sends another that does.
 	phase := time.Now()
 	if err = kbSvc.StartLoopbackServer(libkb.LoginAttemptNone); err != nil {
 		log("failed to start loopback: %s", err)

go/engine/bootstrap.go

@@ -62,28 +62,41 @@ func (e *Bootstrap) lookupFullname(m libkb.MetaContext, uv keybase1.UserVersion)
 	e.status.Fullname = pkg.FullName.FullName
 }
 
+// SessionState reads the session fields that are available with nothing to wait
+// on: the active device. Bootstrap fills the same fields plus the slower derived
+// ones, so the two cannot drift. The returned UserVersion is the active device's,
+// empty when logged out.
+func SessionState(m libkb.MetaContext) (res keybase1.ClientSession, uv keybase1.UserVersion) {
+	// if any Login engine worked previously, then ActiveDevice will
+	// be valid; the only way for it to be valid is to be logged in
+	// (and provisioned)
+	res.LoggedIn = m.G().ActiveDevice.Valid()
+	if !res.LoggedIn {
+		return res, uv
+	}
+
+	uv, res.DeviceID, res.DeviceName, _, _ = m.G().ActiveDevice.AllFields()
+	res.Uid = uv.Uid
+	res.Username = m.G().ActiveDevice.Username(m).String()
+	return res, uv
+}
+
 // Run starts the engine.
 func (e *Bootstrap) Run(m libkb.MetaContext) (err error) {
 	defer m.Trace("Bootstrap.Run", &err)()
-	e.status.Registered = e.signedUp(m)
-
-	// if any Login engine worked previously, then ActiveDevice will
-	// be valid:
-	validActiveDevice := m.G().ActiveDevice.Valid()
+	session, uv := SessionState(m)
+	e.status.Registered = signedUp(m)
+	e.status.LoggedIn = session.LoggedIn
+	e.status.Uid = session.Uid
+	e.status.Username = session.Username
+	e.status.DeviceID = session.DeviceID
+	e.status.DeviceName = session.DeviceName
 
-	// the only way for ActiveDevice to be valid is to be logged in
-	// (and provisioned)
-	e.status.LoggedIn = validActiveDevice
 	if !e.status.LoggedIn {
 		m.Debug("Bootstrap: not logged in")
 		return nil
 	}
 	m.Debug("Bootstrap: logged in (valid active device)")
-
-	var uv keybase1.UserVersion
-	uv, e.status.DeviceID, e.status.DeviceName, _, _ = e.G().ActiveDevice.AllFields()
-	e.status.Uid = uv.Uid
-	e.status.Username = e.G().ActiveDevice.Username(m).String()
 	m.Debug("Bootstrap status: uid=%s, username=%s, deviceID=%s, deviceName=%s", e.status.Uid, e.status.Username, e.status.DeviceID, e.status.DeviceName)
 
 	if chatHelper := e.G().ChatHelper; chatHelper != nil {
@@ -96,7 +109,7 @@ func (e *Bootstrap) Run(m libkb.MetaContext) (err error) {
 }
 
 // signedUp is true if there's a uid in config.json.
-func (e *Bootstrap) signedUp(m libkb.MetaContext) bool {
+func signedUp(m libkb.MetaContext) bool {
 	cr := m.G().Env.GetConfig()
 	if cr == nil {
 		return false

go/libkb/appstate_test.go

@@ -76,3 +76,62 @@ func TestMobileAppStateBackgroundCancelsRPCsOnlyOnChange(t *testing.T) {
 	a.Update(keybase1.MobileAppState_BACKGROUND)
 	requireOpen(t, second.Done())
 }
+
+func appStateChanges(t *testing.T, rec *NotifyRecorder) []keybase1.MobileAppState {
+	t.Helper()
+	var ret []keybase1.MobileAppState
+	for _, m := range rec.Messages() {
+		if m.Method != "keybase.1.NotifyApp.mobileAppStateChanged" {
+			continue
+		}
+		var arg keybase1.MobileAppStateChangedArg
+		require.NoError(t, m.Decode(&arg))
+		ret = append(ret, arg.State)
+	}
+	return ret
+}
+
+// Clients are told from the one place the value changes, so no writer can add a
+// path that moves the state without announcing it.
+func TestMobileAppStateAnnouncesOnlyOnChange(t *testing.T) {
+	tc := SetupTest(t, "MobileAppStateAnnounce", 0)
+	defer tc.Cleanup()
+	tc.G.SetService()
+	a := NewMobileAppState(tc.G)
+	rec := NewNotifyRecorder(tc.G, keybase1.NotificationChannels{App: true})
+	defer rec.Close()
+
+	a.Update(keybase1.MobileAppState_BACKGROUND)
+	a.Update(keybase1.MobileAppState_BACKGROUND)
+	a.Update(keybase1.MobileAppState_FOREGROUND)
+	rec.Flush()
+	require.Equal(t, []keybase1.MobileAppState{
+		keybase1.MobileAppState_BACKGROUND,
+		keybase1.MobileAppState_FOREGROUND,
+	}, appStateChanges(t, rec), "one notification per change, none for a same-value update")
+}
+
+// The notification is queued in the same critical section that wrote the
+// state, so two concurrent Updates queue in the order they wrote rather than in
+// whatever order they reached the router. Checked white-box: holding the lock
+// across updateLocked is the only way to observe "has it been queued yet", and
+// the answer must be yes before the lock is released.
+func TestMobileAppStateQueuesUnderTheLock(t *testing.T) {
+	tc := SetupTest(t, "MobileAppStateQueue", 0)
+	defer tc.Cleanup()
+	tc.G.SetService()
+	a := NewMobileAppState(tc.G)
+	rec := NewNotifyRecorder(tc.G, keybase1.NotificationChannels{App: true})
+	defer rec.Close()
+
+	a.Lock()
+	a.updateLocked(keybase1.MobileAppState_BACKGROUND)
+	// nothing queued reads app state (there is no clientState reader here), so
+	// flushing under the lock cannot deadlock
+	rec.Flush()
+	queued := appStateChanges(t, rec)
+	a.Unlock()
+
+	require.Equal(t, []keybase1.MobileAppState{keybase1.MobileAppState_BACKGROUND}, queued,
+		"the change was queued before the lock that wrote it was released")
+}

go/libkb/connmgr.go

@@ -22,11 +22,6 @@ type ConnectionID int
 // true to keep going and false to stop.
 type ApplyFn func(i ConnectionID, xp rpc.Transporter) bool
 
-// ApplyDetailsFn can be applied to every connection. It is called with the
-// RPC transporter, and also the connectionID. It should return a bool
-// true to keep going and false to stop.
-type ApplyDetailsFn func(i ConnectionID, xp rpc.Transporter, details *keybase1.ClientDetails) bool
-
 // LabelCb is a callback to be run when a client connects and labels itself.
 type LabelCb func(typ keybase1.ClientType)
 
@@ -44,23 +39,14 @@ type ConnectionManager struct {
 	labelCbs []LabelCb
 }
 
-// AddConnection adds a new connection to the table of Connection object, with a
-// related closeListener. We'll listen for a close on that channel, and when one occurs,
-// we'll remove the connection from the pool.
-func (c *ConnectionManager) AddConnection(xp rpc.Transporter, closeListener chan error) ConnectionID {
+// AddConnection adds a new connection to the table of Connection objects.
+// NotifyRouter.AddConnection removes it when the connection closes.
+func (c *ConnectionManager) AddConnection(xp rpc.Transporter) ConnectionID {
 	c.Lock()
+	defer c.Unlock()
 	c.nxt++ // increment first, since 0 is reserved
 	id := c.nxt
 	c.lookup[id] = &rpcConnection{transporter: xp}
-	c.Unlock()
-
-	if closeListener != nil {
-		go func() {
-			<-closeListener
-			c.removeConnection(id)
-		}()
-	}
-
 	return id
 }
 
@@ -183,24 +169,6 @@ func (c *ConnectionManager) ApplyAll(f ApplyFn) {
 	}
 }
 
-// ApplyAllDetails applies the given function f to all connections in the table.
-// If you're going to do something blocking, please do it in a GoRoutine,
-// since we're holding the lock for all connections as we do this.
-func (c *ConnectionManager) ApplyAllDetails(f ApplyDetailsFn) {
-	c.Lock()
-	defer c.Unlock()
-	for k, v := range c.lookup {
-		status := v.details
-		var details *keybase1.ClientDetails
-		if status != nil {
-			details = &status.Details
-		}
-		if !f(k, v.transporter, details) {
-			break
-		}
-	}
-}
-
 // NewConnectionManager makes a new ConnectionManager.
 func NewConnectionManager() *ConnectionManager {
 	return &ConnectionManager{

go/libkb/context.go

@@ -366,7 +366,7 @@ func (m MetaContext) SwitchUserNewConfig(u keybase1.UID, n NormalizedUsername, s
 
 func (m MetaContext) switchUserNewConfig(u keybase1.UID, n NormalizedUsername, salt []byte, d keybase1.DeviceID, ad *ActiveDevice) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "switchUserNewConfig")()
+	defer g.lockSwitchUser(m, ad != nil, "switchUserNewConfig")()
 	cw := g.Env.GetConfigWriter()
 	if cw == nil {
 		return NoConfigWriterError{}
@@ -398,7 +398,7 @@ func (m MetaContext) SwitchUserNewConfigActiveDevice(uv keybase1.UserVersion, n
 // etc). It does this in a critical section, holding switchUserMu.
 func (m MetaContext) SwitchUserNukeConfig(n NormalizedUsername) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SwitchUserNukeConfig")()
+	defer g.lockSwitchUser(m, false, "SwitchUserNukeConfig")()
 	cw := g.Env.GetConfigWriter()
 	cr := g.Env.GetConfig()
 	if cw == nil {
@@ -435,7 +435,7 @@ func (m MetaContext) SwitchUserToActiveDevice(n NormalizedUsername, ad *ActiveDe
 	if !n.IsValid() {
 		return NewBadUsernameError(n.String())
 	}
-	defer g.switchUserMu.Acquire(m, "SwitchUserToActiveDevice %v", n)()
+	defer g.lockSwitchUser(m, false, "SwitchUserToActiveDevice %v", n)()
 	cw := g.Env.GetConfigWriter()
 	if cw == nil {
 		return NoConfigWriterError{}
@@ -459,7 +459,7 @@ func (m MetaContext) SwitchUserToActiveDevice(n NormalizedUsername, ad *ActiveDe
 
 func (m MetaContext) SwitchUserDeprovisionNukeConfig(username NormalizedUsername) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SwitchUserDeprovisionNukeConfig %v", username)()
+	defer g.lockSwitchUser(m, false, "SwitchUserDeprovisionNukeConfig %v", username)()
 
 	cw := g.Env.GetConfigWriter()
 	if cw == nil {
@@ -481,7 +481,7 @@ func (m MetaContext) SwitchUserToActiveOneshotDevice(uv keybase1.UserVersion, nu
 	defer m.Trace("MetaContext#SwitchUserToActiveOneshotDevice", &err)()
 
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SwitchUserToActiveOneshotDevice")()
+	defer g.lockSwitchUser(m, true, "SwitchUserToActiveOneshotDevice")()
 	cw := g.Env.GetConfigWriter()
 	if cw == nil {
 		return NoConfigWriterError{}
@@ -504,7 +504,7 @@ func (m MetaContext) SwitchUserToActiveOneshotDevice(uv keybase1.UserVersion, nu
 func (m MetaContext) SwitchUserLoggedOut() (err error) {
 	defer m.Trace("MetaContext#SwitchUserLoggedOut", &err)()
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SwitchUserLoggedOut")()
+	defer g.lockSwitchUser(m, false, "SwitchUserLoggedOut")()
 	cw := g.Env.GetConfigWriter()
 	if cw == nil {
 		return NoConfigWriterError{}
@@ -530,7 +530,7 @@ func (m MetaContext) SetActiveDevice(uv keybase1.UserVersion, deviceID keybase1.
 	sigKey, encKey GenericKey, deviceName string, keychainMode KeychainMode,
 ) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SetActiveDevice")()
+	defer g.lockSwitchUser(m, false, "SetActiveDevice")()
 	if !g.Env.GetUID().Equal(uv.Uid) {
 		return NewUIDMismatchError("UID switched out from underneath provisioning process")
 	}
@@ -539,13 +539,13 @@ func (m MetaContext) SetActiveDevice(uv keybase1.UserVersion, deviceID keybase1.
 
 func (m MetaContext) SetSigningKey(uv keybase1.UserVersion, deviceID keybase1.DeviceID, sigKey GenericKey, deviceName string) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SetSigningKey")()
+	defer g.lockSwitchUser(m, false, "SetSigningKey")()
 	return g.ActiveDevice.setSigningKey(g, uv, deviceID, sigKey, deviceName)
 }
 
 func (m MetaContext) SetEncryptionKey(uv keybase1.UserVersion, deviceID keybase1.DeviceID, encKey GenericKey) error {
 	g := m.G()
-	defer g.switchUserMu.Acquire(m, "SetEncryptionKey")()
+	defer g.lockSwitchUser(m, false, "SetEncryptionKey")()
 	return g.ActiveDevice.setEncryptionKey(uv, deviceID, encKey)
 }
 

go/libkb/globals.go

@@ -169,7 +169,8 @@ type GlobalContext struct {
 
 	// It is threadsafe to call methods on ActiveDevice which will always be non-nil.
 	// But don't access its members directly. If you're going to be changing out the
-	// user (and resetting the ActiveDevice), then you should hold the switchUserMu
+	// user (and resetting the ActiveDevice), then you should hold the switchUserMu,
+	// through lockSwitchUser
 	switchUserMu  *VerboseLock
 	ActiveDevice  *ActiveDevice
 	switchedUsers map[NormalizedUsername]bool // bookkeep users who have been switched over (and are still in secret store)
@@ -358,10 +359,42 @@ func (g *GlobalContext) SetAvatarLoader(a AvatarLoaderSource) {
 	g.avatarLoader = a
 }
 
+type sessionIdentity struct {
+	valid    bool
+	uid      keybase1.UID
+	deviceID keybase1.DeviceID
+}
+
+func (g *GlobalContext) sessionIdentity() sessionIdentity {
+	return sessionIdentity{valid: g.ActiveDevice.Valid(), uid: g.ActiveDevice.UID(), deviceID: g.ActiveDevice.DeviceID()}
+}
+
+// lockSwitchUser takes switchUserMu, which every session write (the active
+// device, the config's current user) is made under. A release that changed the
+// session queues a clientState to connected clients, after unlocking.
+//
+// promotion marks the write a provisioning, signup or oneshot flow makes before
+// it completes: if it leaves a valid session it queues nothing, so clients do
+// not see the login early -- the flow completes with SendLogin, which queues
+// one. A promotion that leaves no valid session is a clear and queues one like
+// any other. See connSender for why that is enough.
+func (g *GlobalContext) lockSwitchUser(mctx MetaContext, promotion bool, reasonFormat string, args ...any) (release func()) {
+	unlock := g.switchUserMu.Acquire(mctx, reasonFormat, args...)
+	before := g.sessionIdentity()
+	return func() {
+		after := g.sessionIdentity()
+		unlock()
+		earlyLogin := promotion && after.valid
+		if after != before && !earlyLogin {
+			g.NotifyRouter.AnnounceClientState(mctx.Ctx())
+		}
+	}
+}
+
 // simulateServiceRestart simulates what happens when a service restarts for the
 // purposes of testing.
 func (g *GlobalContext) simulateServiceRestart() {
-	defer g.switchUserMu.Acquire(NewMetaContext(context.TODO(), g), "simulateServiceRestart")()
+	defer g.lockSwitchUser(NewMetaContext(context.TODO(), g), false, "simulateServiceRestart")()
 	_ = g.ActiveDevice.Clear()
 }
 

go/libkb/logout.go

@@ -30,7 +30,7 @@ func (mctx MetaContext) LogoutUsernameWithOptions(username NormalizedUsername, o
 	defer mctx.Trace(fmt.Sprintf("MetaContext#LogoutWithOptions(%#v)", options), &err)()
 
 	g := mctx.G()
-	defer g.switchUserMu.Acquire(mctx, "Logout")()
+	defer g.lockSwitchUser(mctx, false, "Logout")()
 
 	mctx.Debug("MetaContext#logoutWithSecretKill: after switchUserMu acquisition (username: %s, options: %#v)",
 		username, options)