refactor(libkb): fix appstate subscriber leak via shared broadcast channel

go/avatars/fullcaching.go

@@ -255,7 +255,8 @@ func (c *FullCachingSource) monitorAppState(m libkb.MetaContext) {
 	c.debug(m, "monitorAppState: starting up")
 	state := keybase1.MobileAppState_FOREGROUND
 	for {
-		state = <-m.G().MobileAppState.NextUpdate(&state)
+		<-m.G().MobileAppState.NextUpdate(state)
+		state = m.G().MobileAppState.State()
 		if state == keybase1.MobileAppState_BACKGROUND {
 			c.debug(m, "monitorAppState: backgrounded")
 			if err := c.diskLRU.Flush(m.Ctx(), m.G()); err != nil {

go/avatars/urlcaching.go

@@ -53,7 +53,8 @@ func (c *URLCachingSource) monitorAppState(m libkb.MetaContext) {
 	c.debug(m, "monitorAppState: starting up")
 	state := keybase1.MobileAppState_FOREGROUND
 	for {
-		state = <-m.G().MobileAppState.NextUpdate(&state)
+		<-m.G().MobileAppState.NextUpdate(state)
+		state = m.G().MobileAppState.State()
 		if state == keybase1.MobileAppState_BACKGROUND {
 			c.debug(m, "monitorAppState: backgrounded")
 			c.diskLRU.Flush(m.Ctx(), m.G())

go/bind/keybase.go

@@ -854,8 +854,9 @@ func BackgroundSync() string {
 		return s == keybase1.MobileAppState_BACKGROUND
 	})
 	select {
-	case state := <-kbCtx.MobileAppState.NextUpdate(&nextState):
+	case <-kbCtx.MobileAppState.NextUpdate(nextState):
 		// if literally anything happens, let's get out of here
+		state := kbCtx.MobileAppState.State()
 		msg := fmt.Sprintf("bailing out early, appstate change: %v", state)
 		kbCtx.Log.Debug("BackgroundSync: %s", msg)
 		return msg
@@ -970,7 +971,8 @@ func AppBeginBackgroundTask(pusher PushNotifier) {
 	g, ctx = errgroup.WithContext(ctx)
 	g.Go(func() error {
 		select {
-		case appState = <-kbCtx.MobileAppState.NextUpdate(&appState):
+		case <-kbCtx.MobileAppState.NextUpdate(appState):
+			appState = kbCtx.MobileAppState.State()
 			kbCtx.Log.Debug(
 				"AppBeginBackgroundTask: app state change, aborting with no task shutdown: %v", appState)
 			return errors.New("app state change")

go/chat/archive.go

@@ -196,7 +196,8 @@ func (r *ChatArchiveRegistry) monitorAppState(stopCh chan struct{}) error {
 		case <-stopCh:
 			cancel()
 			return nil
-		case appState = <-r.G().MobileAppState.NextUpdate(&appState):
+		case <-r.G().MobileAppState.NextUpdate(appState):
+			appState = r.G().MobileAppState.State()
 			r.Debug(ctx, "monitorAppState: next state -> %v", appState)
 			switch appState {
 			case keybase1.MobileAppState_FOREGROUND:

go/chat/convloader.go

@@ -178,7 +178,8 @@ func (b *BackgroundConvLoader) monitorAppState(stopCh chan struct{}) error {
 	state := keybase1.MobileAppState_FOREGROUND
 	for {
 		select {
-		case state = <-b.G().MobileAppState.NextUpdate(&state):
+		case <-b.G().MobileAppState.NextUpdate(state):
+			state = b.G().MobileAppState.State()
 			switch state {
 			case keybase1.MobileAppState_FOREGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE:
 				b.Debug(ctx, "monitorAppState: active state: %v", state)

go/chat/ephemeral_purger.go

@@ -282,7 +282,8 @@ func (b *BackgroundEphemeralPurger) loop(shutdownCh chan struct{}) error {
 		case <-b.purgeTimer.C:
 			b.Debug(bgctx, "loop: timer fired %s", b.uid)
 			b.queuePurges(bgctx)
-		case suspended = <-b.G().DesktopAppState.NextSuspendUpdate(&suspended):
+		case <-b.G().DesktopAppState.NextSuspendUpdate(suspended):
+			suspended = b.G().DesktopAppState.Suspended()
 			if !suspended {
 				b.Debug(bgctx, "loop: queuing purges on resume %s", b.uid)
 				b.queuePurges(bgctx)

go/chat/search/indexer.go

@@ -262,14 +262,16 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error {
 			attemptSync(ctx)
 		case <-ticker.C:
 			attemptSync(ctx)
-		case appState = <-idx.G().MobileAppState.NextUpdate(&appState):
+		case <-idx.G().MobileAppState.NextUpdate(appState):
+			appState = idx.G().MobileAppState.State()
 			switch appState {
 			case keybase1.MobileAppState_FOREGROUND:
 			// if we enter any state besides foreground cancel any running syncs
 			default:
 				cancelSync()
 			}
-		case netState = <-idx.G().MobileNetState.NextUpdate(&netState):
+		case <-idx.G().MobileNetState.NextUpdate(netState):
+			netState = idx.G().MobileNetState.State()
 			if netState.IsLimited() {
 				// if we switch off of wifi cancel any running syncs
 				cancelSync()

go/ephemeral/lib.go

@@ -125,7 +125,8 @@ func (e *EKLib) backgroundKeygen(mctx libkb.MetaContext, stopCh <-chan struct{})
 		select {
 		case <-ticker.C:
 			runIfNeeded(false /* force */)
-		case state = <-mctx.G().MobileAppState.NextUpdate(&state):
+		case <-mctx.G().MobileAppState.NextUpdate(state):
+			state = mctx.G().MobileAppState.State()
 			if state == keybase1.MobileAppState_BACKGROUNDACTIVE {
 				// Before running we pause briefly so we don't stampede for
 				// resources with other background tasks. libkb.BgTicker

go/kbfs/env/context.go

@@ -24,34 +24,51 @@ const (
 )
 
 // AppStateUpdater is an interface for things that need to listen to
-// app state changes.
+// app state changes. Callers subscribe by calling NextAppStateUpdate /
+// NextNetworkStateUpdate with their last-observed state; when the returned
+// channel is closed, the caller re-fetches via AppState() / NetworkState().
 type AppStateUpdater interface {
-	// NextAppStateUpdate returns a channel that app state changes
-	// are sent to.
-	NextAppStateUpdate(lastState *keybase1.MobileAppState) <-chan keybase1.MobileAppState
-	// NextNetworkStateUpdate returns a channel that mobile network
-	// state changes are sent to.
-	NextNetworkStateUpdate(lastState *keybase1.MobileNetworkState) <-chan keybase1.MobileNetworkState
+	// NextAppStateUpdate returns a channel that will be closed the next time
+	// the app state changes. If lastState is stale, an already-closed channel
+	// is returned so the caller wakes immediately.
+	NextAppStateUpdate(lastState keybase1.MobileAppState) <-chan struct{}
+	// NextNetworkStateUpdate returns a channel that will be closed the next
+	// time the network state changes.
+	NextNetworkStateUpdate(lastState keybase1.MobileNetworkState) <-chan struct{}
+	// AppState returns the current app state.
+	AppState() keybase1.MobileAppState
+	// NetworkState returns the current network state.
+	NetworkState() keybase1.MobileNetworkState
 }
 
 // EmptyAppStateUpdater is an implementation of AppStateUpdater that
 // never returns an update, for testing.
 type EmptyAppStateUpdater struct{}
 
 // NextAppStateUpdate implements AppStateUpdater.
-func (easu EmptyAppStateUpdater) NextAppStateUpdate(lastState *keybase1.MobileAppState) <-chan keybase1.MobileAppState {
+func (easu EmptyAppStateUpdater) NextAppStateUpdate(lastState keybase1.MobileAppState) <-chan struct{} {
 	// Receiving on a nil channel blocks forever.
 	return nil
 }
 
 // NextNetworkStateUpdate implements AppStateUpdater.
 func (easu EmptyAppStateUpdater) NextNetworkStateUpdate(
-	lastState *keybase1.MobileNetworkState,
-) <-chan keybase1.MobileNetworkState {
+	lastState keybase1.MobileNetworkState,
+) <-chan struct{} {
 	// Receiving on a nil channel blocks forever.
 	return nil
 }
 
+// AppState implements AppStateUpdater.
+func (easu EmptyAppStateUpdater) AppState() keybase1.MobileAppState {
+	return keybase1.MobileAppState_FOREGROUND
+}
+
+// NetworkState implements AppStateUpdater.
+func (easu EmptyAppStateUpdater) NetworkState() keybase1.MobileNetworkState {
+	return keybase1.MobileNetworkState_NOTAVAILABLE
+}
+
 // Context defines the environment for this package
 type Context interface {
 	AppStateUpdater
@@ -171,7 +188,7 @@ func (c *KBFSContext) GetPerfLog() logger.Logger {
 }
 
 // NextAppStateUpdate implements AppStateUpdater.
-func (c *KBFSContext) NextAppStateUpdate(lastState *keybase1.MobileAppState) <-chan keybase1.MobileAppState {
+func (c *KBFSContext) NextAppStateUpdate(lastState keybase1.MobileAppState) <-chan struct{} {
 	if c.g.MobileAppState == nil {
 		return nil
 	}
@@ -180,14 +197,30 @@ func (c *KBFSContext) NextAppStateUpdate(lastState *keybase1.MobileAppState) <-c
 
 // NextNetworkStateUpdate implements AppStateUpdater.
 func (c *KBFSContext) NextNetworkStateUpdate(
-	lastState *keybase1.MobileNetworkState,
-) <-chan keybase1.MobileNetworkState {
+	lastState keybase1.MobileNetworkState,
+) <-chan struct{} {
 	if c.g.MobileNetState == nil {
 		return nil
 	}
 	return c.g.MobileNetState.NextUpdate(lastState)
 }
 
+// AppState implements AppStateUpdater.
+func (c *KBFSContext) AppState() keybase1.MobileAppState {
+	if c.g.MobileAppState == nil {
+		return keybase1.MobileAppState_FOREGROUND
+	}
+	return c.g.MobileAppState.State()
+}
+
+// NetworkState implements AppStateUpdater.
+func (c *KBFSContext) NetworkState() keybase1.MobileNetworkState {
+	if c.g.MobileNetState == nil {
+		return keybase1.MobileNetworkState_NOTAVAILABLE
+	}
+	return c.g.MobileNetState.State()
+}
+
 // CheckService checks if the service is running and returns nil if
 // so, and an error otherwise.
 func (c *KBFSContext) CheckService() error {

go/kbfs/libfs/fs.go

@@ -1136,40 +1136,59 @@ func (fs *FS) Handle() *tlfhandle.Handle {
 	return fs.h
 }
 
-type folderHandleChangeObserver func()
+// folderHandleChangeObserver is a struct (rather than a bare func()) so it's
+// comparable by pointer identity - libkbfs.observerList.remove uses == to find
+// the entry to drop, which panics on func types.
+type folderHandleChangeObserver struct {
+	onChange func()
+}
 
-func (folderHandleChangeObserver) LocalChange(
+func (*folderHandleChangeObserver) LocalChange(
 	context.Context, libkbfs.Node, libkbfs.WriteRange) {
 }
 
-func (folderHandleChangeObserver) BatchChanges(
+func (*folderHandleChangeObserver) BatchChanges(
 	context.Context, []libkbfs.NodeChange, []libkbfs.NodeID) {
 }
 
-func (o folderHandleChangeObserver) TlfHandleChange(
+func (o *folderHandleChangeObserver) TlfHandleChange(
 	context.Context, *tlfhandle.Handle,
 ) {
-	o()
+	o.onChange()
 }
 
 // SubscribeToObsolete returns a channel that will be closed when this *FS
 // reaches obsolescence, meaning if user of this object caches it for long term
-// use, it should invalide this entry and create a new one using NewFS.
-func (fs *FS) SubscribeToObsolete() (<-chan struct{}, error) {
+// use, it should invalidate this entry and create a new one using NewFS. The
+// returned unsubscribe function must be called when the caller is done with
+// the subscription so the underlying folder-branch observer can be removed;
+// otherwise the observer leaks on that TLF's folderBranchOps for the process
+// lifetime. Calling unsubscribe more than once is safe.
+func (fs *FS) SubscribeToObsolete() (
+	obsoleteCh <-chan struct{}, unsubscribe func(), err error,
+) {
 	if err := fs.chooseErrorIfEmpty(onFsEmptyErrNotSupported); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	c := make(chan struct{})
-	var once sync.Once
-	onHandleChange := folderHandleChangeObserver(
-		func() { once.Do(func() { close(c) }) })
+	var closeOnce sync.Once
+	onHandleChange := &folderHandleChangeObserver{
+		onChange: func() { closeOnce.Do(func() { close(c) }) },
+	}
+	fb := fs.root.GetFolderBranch()
 	if err := fs.config.Notifier().RegisterForChanges(
-		[]data.FolderBranch{fs.root.GetFolderBranch()},
-		onHandleChange); err != nil {
-		return nil, err
+		[]data.FolderBranch{fb}, onHandleChange); err != nil {
+		return nil, nil, err
+	}
+	var unsubOnce sync.Once
+	unsubscribe = func() {
+		unsubOnce.Do(func() {
+			_ = fs.config.Notifier().UnregisterFromChanges(
+				[]data.FolderBranch{fb}, onHandleChange)
+		})
 	}
-	return c, nil
+	return c, unsubscribe, nil
 }
 
 // IsEmpty returns true if this is a faked-out empty TLF.

go/kbfs/libfs/fs_test.go

@@ -737,3 +737,27 @@ func TestEmptyFS(t *testing.T) {
 	err = fs.MkdirAll("a", 0o777)
 	require.Error(t, err)
 }
+
+// TestSubscribeToObsolete is a smoke test for the (ch, unsubscribe, err)
+// contract: subscribe returns a live channel and a callable unsubscribe, the
+// channel is not closed while the TLF handle is unchanged, and unsubscribe is
+// safe to call multiple times.
+func TestSubscribeToObsolete(t *testing.T) {
+	ctx, _, fs := makeFS(t, "")
+	defer libkbfs.CheckConfigAndShutdown(ctx, t, fs.config)
+
+	obsoleteCh, unsubscribe, err := fs.SubscribeToObsolete()
+	require.NoError(t, err)
+	require.NotNil(t, obsoleteCh)
+	require.NotNil(t, unsubscribe)
+
+	select {
+	case <-obsoleteCh:
+		t.Fatal("obsoleteCh unexpectedly closed before any handle change")
+	default:
+	}
+
+	unsubscribe()
+	// Idempotent: second call must not panic.
+	unsubscribe()
+}

go/kbfs/libhttpserver/server.go

@@ -107,8 +107,9 @@ func (s *Server) handleInternalServerError(w http.ResponseWriter) {
 }
 
 type obsoleteTrackingFS struct {
-	fs *libfs.FS
-	ch <-chan struct{}
+	fs          *libfs.FS
+	ch          <-chan struct{}
+	unsubscribe func()
 }
 
 func (e obsoleteTrackingFS) isObsolete() bool {
@@ -156,12 +157,14 @@ func (s *Server) getHTTPFileSystem(ctx context.Context, requestPath string) (
 		return "", nil, err
 	}
 
-	fsLifeCh, err := tlfFS.SubscribeToObsolete()
+	fsLifeCh, unsubscribe, err := tlfFS.SubscribeToObsolete()
 	if err != nil {
 		return "", nil, err
 	}
 
-	s.fs.Add(toStrip, obsoleteTrackingFS{fs: tlfFS, ch: fsLifeCh})
+	s.fs.Add(toStrip, obsoleteTrackingFS{
+		fs: tlfFS, ch: fsLifeCh, unsubscribe: unsubscribe,
+	})
 
 	return toStrip, tlfFS.ToHTTPFileSystem(ctx), nil
 }
@@ -248,7 +251,8 @@ func (s *Server) monitorAppState(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case state = <-s.appStateUpdater.NextAppStateUpdate(&state):
+		case <-s.appStateUpdater.NextAppStateUpdate(state):
+			state = s.appStateUpdater.AppState()
 			// Due to the way NextUpdate is designed, it's possible we miss an
 			// update if processing the last update takes too long. So it's
 			// possible to get consecutive FOREGROUND updates even if there are
@@ -279,7 +283,12 @@ func New(appStateUpdater env.AppStateUpdater, config libkbfs.Config) (
 		logger:          logger,
 		vlog:            config.MakeVLogger(logger),
 	}
-	if s.fs, err = lru.New(fsCacheSize); err != nil {
+	s.fs, err = lru.NewWithEvict(fsCacheSize, func(_ any, value any) {
+		if e, ok := value.(obsoleteTrackingFS); ok && e.unsubscribe != nil {
+			e.unsubscribe()
+		}
+	})
+	if err != nil {
 		return nil, err
 	}
 	if err = s.restart(); err != nil {
@@ -304,5 +313,8 @@ func (s *Server) Shutdown() {
 	s.serverLock.Lock()
 	defer s.serverLock.Unlock()
 	s.server.Stop()
+	// Purge the LRU so its evict callback runs and unsubscribes any
+	// folder-branch observers still held by cached entries.
+	s.fs.Purge()
 	s.cancel()
 }