From f3b2240142e855ed0373a5fe03c85eb1eff2f95b Mon Sep 17 00:00:00 2001 From: Song Gao Date: Fri, 31 Jul 2026 13:19:56 -0700 Subject: [PATCH 1/4] refactor(libkb): fix appstate subscriber leak via shared broadcast channel MobileAppState.NextUpdate / MobileNetState.NextUpdate / DesktopAppState.NextSuspendUpdate previously allocated a fresh channel per call and appended it to a per-caller subscriber slice, only draining that slice when the state actually changed. On headless daemons the state never changes, so the slices grew unbounded - ~880 MB of live heap on kbpagesd before OOM, plus slower growth in keybase.service via chat/search/indexer, service/gregor, ephemeral/lib, etc. Rework the internal representation: each state now holds a single close-only chan struct{} that is closed and replaced when the state actually changes. NextUpdate returns that shared channel; if the caller's lastState is stale, a package-level pre-closed sentinel is returned so the caller wakes immediately. Callers get the current value via a separate State() / NetworkState() / Suspended() accessor. Semantic change: subscribers can no longer observe every intermediate state transition. If the state moves A -> B -> A between two subscriber observations, the subscriber sees no wake. Guarantee is only that when the returned channel fires, the caller's next State() call returns a value different from the one they last observed. Audited every caller (prefetcher, folder_block_manager, libhttpserver, kbfs/search/indexer, chat/search/indexer, chat/convloader, chat/archive, chat/ephemeral_purger, service/gregor, ephemeral/lib, avatars/{url,full}caching, kbhttp/manager, stellar/util, bind/keybase, libkb/leveldb_cleaner) - each one reacts to the current state rather than requiring every transition, so collapsing intermediates is safe. Also adds AppState / NetworkState methods to env.AppStateUpdater so KBFS-side callers can fetch state without reaching into libkb directly. Made-with: Claude --- go/avatars/fullcaching.go | 3 +- go/avatars/urlcaching.go | 3 +- go/bind/keybase.go | 6 +- go/chat/archive.go | 3 +- go/chat/convloader.go | 3 +- go/chat/ephemeral_purger.go | 3 +- go/chat/search/indexer.go | 6 +- go/ephemeral/lib.go | 3 +- go/kbfs/env/context.go | 59 +++++++--- go/kbfs/libhttpserver/server.go | 3 +- go/kbfs/libkbfs/folder_block_manager.go | 12 ++- go/kbfs/libkbfs/prefetcher.go | 14 +-- go/kbfs/libkbfs/prefetcher_test.go | 70 +++++++++--- go/kbfs/search/indexer.go | 6 +- go/kbhttp/manager/manager.go | 3 +- go/libkb/appstate.go | 137 +++++++++++++++--------- go/libkb/leveldb_cleaner.go | 3 +- go/service/gregor.go | 6 +- go/stellar/util.go | 6 +- 19 files changed, 240 insertions(+), 109 deletions(-) diff --git a/go/avatars/fullcaching.go b/go/avatars/fullcaching.go index 897460f8b0b2..07cedbe65478 100644 --- a/go/avatars/fullcaching.go +++ b/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 { diff --git a/go/avatars/urlcaching.go b/go/avatars/urlcaching.go index deab9994fcfd..96b543848302 100644 --- a/go/avatars/urlcaching.go +++ b/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()) diff --git a/go/bind/keybase.go b/go/bind/keybase.go index 84983ccf4a73..b5104564a96a 100644 --- a/go/bind/keybase.go +++ b/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") diff --git a/go/chat/archive.go b/go/chat/archive.go index c001bc50a40f..b9d984b93af4 100644 --- a/go/chat/archive.go +++ b/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: diff --git a/go/chat/convloader.go b/go/chat/convloader.go index 9eb78f451713..1523ac249411 100644 --- a/go/chat/convloader.go +++ b/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) diff --git a/go/chat/ephemeral_purger.go b/go/chat/ephemeral_purger.go index 6592d469257c..2b8dd61d3694 100644 --- a/go/chat/ephemeral_purger.go +++ b/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) diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index 9cfeb78bc3fb..d2ae31496a4d 100644 --- a/go/chat/search/indexer.go +++ b/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() diff --git a/go/ephemeral/lib.go b/go/ephemeral/lib.go index df832e8f3376..e17ac4b2696d 100644 --- a/go/ephemeral/lib.go +++ b/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 diff --git a/go/kbfs/env/context.go b/go/kbfs/env/context.go index 93d8181a9c83..b3a490eda7b9 100644 --- a/go/kbfs/env/context.go +++ b/go/kbfs/env/context.go @@ -24,14 +24,21 @@ 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 @@ -39,19 +46,29 @@ type AppStateUpdater interface { 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 { diff --git a/go/kbfs/libhttpserver/server.go b/go/kbfs/libhttpserver/server.go index 2e172864186c..0f152d1a569e 100644 --- a/go/kbfs/libhttpserver/server.go +++ b/go/kbfs/libhttpserver/server.go @@ -248,7 +248,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 diff --git a/go/kbfs/libkbfs/folder_block_manager.go b/go/kbfs/libkbfs/folder_block_manager.go index dca16a0088d9..67e04812e9df 100644 --- a/go/kbfs/libkbfs/folder_block_manager.go +++ b/go/kbfs/libkbfs/folder_block_manager.go @@ -1312,11 +1312,13 @@ func (fbm *folderBlockManager) reclaimQuotaInBackground() { select { case <-fbm.shutdownChan: return - case state = <-fbm.appStateUpdater.NextAppStateUpdate(&state): + case <-fbm.appStateUpdater.NextAppStateUpdate(state): + state = fbm.appStateUpdater.AppState() for state != keybase1.MobileAppState_FOREGROUND { fbm.log.CDebugf(context.Background(), "Pausing QR while not foregrounded: state=%s", state) - state = <-fbm.appStateUpdater.NextAppStateUpdate(&state) + <-fbm.appStateUpdater.NextAppStateUpdate(state) + state = fbm.appStateUpdater.AppState() } fbm.log.CDebugf( context.Background(), "Resuming QR while foregrounded") @@ -1585,12 +1587,14 @@ func (fbm *folderBlockManager) cleanDiskCachesInBackground() { case <-fbm.latestMergedChan: case <-fbm.shutdownChan: return - case state = <-fbm.appStateUpdater.NextAppStateUpdate(&state): + case <-fbm.appStateUpdater.NextAppStateUpdate(state): + state = fbm.appStateUpdater.AppState() for state != keybase1.MobileAppState_FOREGROUND { fbm.log.CDebugf(context.Background(), "Pausing sync-cache cleaning while not foregrounded: "+ "state=%s", state) - state = <-fbm.appStateUpdater.NextAppStateUpdate(&state) + <-fbm.appStateUpdater.NextAppStateUpdate(state) + state = fbm.appStateUpdater.AppState() } fbm.log.CDebugf(context.Background(), "Resuming sync-cache cleaning while foregrounded") diff --git a/go/kbfs/libkbfs/prefetcher.go b/go/kbfs/libkbfs/prefetcher.go index 7767dd88a9de..c18b7c95f71a 100644 --- a/go/kbfs/libkbfs/prefetcher.go +++ b/go/kbfs/libkbfs/prefetcher.go @@ -1352,8 +1352,8 @@ func (p *blockPrefetcher) handleAppStateChange( p.log.CDebugf( context.TODO(), "Pausing prefetcher while backgrounded") select { - case *appState = <-p.appStateUpdater.NextAppStateUpdate( - appState): + case <-p.appStateUpdater.NextAppStateUpdate(*appState): + *appState = p.appStateUpdater.AppState() case req := <-p.prefetchStatusCh.Out(): p.handleStatusRequest(req.(*prefetchStatusRequest)) continue @@ -1423,8 +1423,8 @@ func (p *blockPrefetcher) handleNetStateChange( p.log.CDebugf( context.TODO(), "Pausing prefetcher on cell network") select { - case *netState = <-p.appStateUpdater.NextNetworkStateUpdate( - netState): + case <-p.appStateUpdater.NextNetworkStateUpdate(*netState): + *netState = p.appStateUpdater.NetworkState() case <-subCh: p.log.CDebugf(context.TODO(), "Settings changed") case req := <-p.prefetchStatusCh.Out(): @@ -1545,9 +1545,11 @@ func (p *blockPrefetcher) run( p.log.Debug("shutting down, clearing in flight fetches") ch := chInterface.(<-chan error) <-ch - case appState = <-p.appStateUpdater.NextAppStateUpdate(&appState): + case <-p.appStateUpdater.NextAppStateUpdate(appState): + appState = p.appStateUpdater.AppState() p.handleAppStateChange(&appState) - case netState = <-p.appStateUpdater.NextNetworkStateUpdate(&netState): + case <-p.appStateUpdater.NextNetworkStateUpdate(netState): + netState = p.appStateUpdater.NetworkState() p.handleNetStateChange(&netState, subCh) case <-subCh: // Settings have changed, so recheck the network state. diff --git a/go/kbfs/libkbfs/prefetcher_test.go b/go/kbfs/libkbfs/prefetcher_test.go index 56af357d39c7..87ad58061c61 100644 --- a/go/kbfs/libkbfs/prefetcher_test.go +++ b/go/kbfs/libkbfs/prefetcher_test.go @@ -8,6 +8,7 @@ import ( "context" "math" "runtime" + "sync" "testing" "time" @@ -2322,36 +2323,77 @@ func TestPrefetcherCancelTlfPrefetches(t *testing.T) { } type testAppStateUpdater struct { - c <-chan keybase1.MobileNetworkState - calls chan<- keybase1.MobileNetworkState - nCalls int + lock sync.Mutex + netState keybase1.MobileNetworkState + changed chan struct{} + calls chan<- keybase1.MobileNetworkState + nCalls int +} + +func newTestAppStateUpdater( + calls chan<- keybase1.MobileNetworkState, nCalls int, +) *testAppStateUpdater { + return &testAppStateUpdater{ + netState: keybase1.MobileNetworkState_NONE, + changed: make(chan struct{}), + calls: calls, + nCalls: nCalls, + } } func (tasu *testAppStateUpdater) NextAppStateUpdate( - _ *keybase1.MobileAppState, -) <-chan keybase1.MobileAppState { + _ keybase1.MobileAppState, +) <-chan struct{} { // Receiving on a nil channel blocks forever. return nil } func (tasu *testAppStateUpdater) NextNetworkStateUpdate( - lastState *keybase1.MobileNetworkState, -) <-chan keybase1.MobileNetworkState { + lastState keybase1.MobileNetworkState, +) <-chan struct{} { + tasu.lock.Lock() + defer tasu.lock.Unlock() if tasu.nCalls > 0 { - tasu.calls <- *lastState + tasu.calls <- lastState tasu.nCalls-- } - return tasu.c + if lastState != tasu.netState { + ch := make(chan struct{}) + close(ch) + return ch + } + return tasu.changed +} + +func (tasu *testAppStateUpdater) AppState() keybase1.MobileAppState { + return keybase1.MobileAppState_FOREGROUND +} + +func (tasu *testAppStateUpdater) NetworkState() keybase1.MobileNetworkState { + tasu.lock.Lock() + defer tasu.lock.Unlock() + return tasu.netState +} + +func (tasu *testAppStateUpdater) setNetworkState( + state keybase1.MobileNetworkState, +) { + tasu.lock.Lock() + defer tasu.lock.Unlock() + if tasu.netState != state { + tasu.netState = state + close(tasu.changed) + tasu.changed = make(chan struct{}) + } } func TestPrefetcherCellularPause(t *testing.T) { t.Log("Test that a cell mobile network pauses prefetching.") bg := newFakeBlockGetter(false) config := newTestBlockRetrievalConfig(t, bg, nil) - stateCh := make(chan keybase1.MobileNetworkState) callCh := make(chan keybase1.MobileNetworkState) - q := newBlockRetrievalQueue( - 1, 1, 0, config, &testAppStateUpdater{stateCh, callCh, 4}) + updater := newTestAppStateUpdater(callCh, 4) + q := newBlockRetrievalQueue(1, 1, 0, config, updater) require.NotNil(t, q) <-callCh // Initial prefetcher, before sync ch is set. @@ -2364,7 +2406,7 @@ func TestPrefetcherCellularPause(t *testing.T) { require.Equal(t, keybase1.MobileNetworkState_NONE, last) t.Log("Switch to cell and make sure it pauses") - stateCh <- keybase1.MobileNetworkState_CELLULAR + updater.setNetworkState(keybase1.MobileNetworkState_CELLULAR) // Should be called again without a call to syncCh. last = <-callCh require.Equal(t, keybase1.MobileNetworkState_CELLULAR, last) @@ -2378,7 +2420,7 @@ func TestPrefetcherCellularPause(t *testing.T) { require.NoError(t, err) t.Log("Unpause it to make it notify again") - stateCh <- keybase1.MobileNetworkState_NONE + updater.setNetworkState(keybase1.MobileNetworkState_NONE) notifySyncCh(t, prefetchSyncCh) last = <-callCh require.Equal(t, keybase1.MobileNetworkState_NONE, last) diff --git a/go/kbfs/search/indexer.go b/go/kbfs/search/indexer.go index 1827124fab8a..d2b6a054a390 100644 --- a/go/kbfs/search/indexer.go +++ b/go/kbfs/search/indexer.go @@ -1401,7 +1401,8 @@ outerLoop: // Re-load the index on each login/logout event. i.log.CDebugf(ctx, "User changed") continue outerLoop - case state = <-kbCtx.NextAppStateUpdate(&state): + case <-kbCtx.NextAppStateUpdate(state): + state = kbCtx.AppState() // TODO(HOTPOT-1494): once we are doing actual // indexing in a separate goroutine, pause/unpause it // via a channel send from here. @@ -1409,7 +1410,8 @@ outerLoop: i.log.CDebugf(ctx, "Pausing indexing while not foregrounded: state=%s", state) - state = <-kbCtx.NextAppStateUpdate(&state) + <-kbCtx.NextAppStateUpdate(state) + state = kbCtx.AppState() } i.log.CDebugf(ctx, "Resuming indexing while foregrounded") continue diff --git a/go/kbhttp/manager/manager.go b/go/kbhttp/manager/manager.go index 39017e2e801c..21cc093b43dd 100644 --- a/go/kbhttp/manager/manager.go +++ b/go/kbhttp/manager/manager.go @@ -112,7 +112,8 @@ func (r *Srv) monitorAppState() { return } for { - state = <-r.G().MobileAppState.NextUpdate(&state) + <-r.G().MobileAppState.NextUpdate(state) + state = r.G().MobileAppState.State() switch state { case keybase1.MobileAppState_FOREGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE: r.startHTTPSrv() diff --git a/go/libkb/appstate.go b/go/libkb/appstate.go index 419f4459316c..8badeca74466 100644 --- a/go/libkb/appstate.go +++ b/go/libkb/appstate.go @@ -10,13 +10,27 @@ import ( "github.com/keybase/go-framed-msgpack-rpc/rpc" ) +// alreadyClosed is a shared close-only channel returned from NextUpdate / +// NextNetworkStateUpdate / NextSuspendUpdate when the caller's lastState is +// stale relative to the current state. Since receiving from a closed +// chan struct{} is safe and idempotent, a single sentinel serves any number of +// concurrent callers without allocation. +var alreadyClosed = func() chan struct{} { + ch := make(chan struct{}) + close(ch) + return ch +}() + // MobileAppState tracks the state of foreground/background status of the app // in which the service is running in. type MobileAppState struct { Contextified sync.Mutex - state keybase1.MobileAppState - updateChs []chan keybase1.MobileAppState + state keybase1.MobileAppState + // changed is closed and replaced whenever state actually changes. Any + // caller holding a reference to the previous channel is woken by the + // close; they then re-read State() to see the new value. + changed chan struct{} // mtime is the time at which the appstate first switched to the current state. // It is a monotonic timestamp and should only be used relatively. @@ -24,32 +38,34 @@ type MobileAppState struct { } func NewMobileAppState(g *GlobalContext) *MobileAppState { + state := keybase1.MobileAppState_FOREGROUND if runtime.GOOS == "android" { // we need this so cold notifications work on android - return &MobileAppState{ - Contextified: NewContextified(g), - state: keybase1.MobileAppState_BACKGROUNDACTIVE, - mtime: nil, - } + state = keybase1.MobileAppState_BACKGROUNDACTIVE } return &MobileAppState{ Contextified: NewContextified(g), - state: keybase1.MobileAppState_FOREGROUND, - mtime: nil, + state: state, + changed: make(chan struct{}), } } -// NextUpdate returns a channel that triggers when the app state changes -func (a *MobileAppState) NextUpdate(lastState *keybase1.MobileAppState) chan keybase1.MobileAppState { +// NextUpdate returns a channel that will be closed the next time the app +// state changes. If lastState does not match the current state, an +// already-closed channel is returned so the caller wakes immediately and can +// re-fetch via State(). +// +// Note: state transitions between two NextUpdate calls may be collapsed - a +// caller is guaranteed only that when the returned channel fires, State() +// returns a value different from the one they last observed; they are not +// guaranteed to observe every intermediate transition. +func (a *MobileAppState) NextUpdate(lastState keybase1.MobileAppState) <-chan struct{} { a.Lock() defer a.Unlock() - ch := make(chan keybase1.MobileAppState, 1) - if lastState != nil && *lastState != a.state { - ch <- a.state - } else { - a.updateChs = append(a.updateChs, ch) + if lastState != a.state { + return alreadyClosed } - return ch + return a.changed } func (a *MobileAppState) updateLocked(state keybase1.MobileAppState) { @@ -61,10 +77,8 @@ func (a *MobileAppState) updateLocked(state keybase1.MobileAppState) { a.state = state t := time.Now() a.mtime = &t // only update mtime if we're changing state - for _, ch := range a.updateChs { - ch <- state - } - a.updateChs = nil + close(a.changed) + a.changed = make(chan struct{}) // cancel RPCs if we go into the background switch a.state { @@ -120,28 +134,29 @@ func (a *MobileAppState) StateAndMtime() (keybase1.MobileAppState, *time.Time) { type MobileNetState struct { Contextified sync.Mutex - state keybase1.MobileNetworkState - updateChs []chan keybase1.MobileNetworkState + state keybase1.MobileNetworkState + // changed is closed and replaced whenever state actually changes. + changed chan struct{} } func NewMobileNetState(g *GlobalContext) *MobileNetState { return &MobileNetState{ Contextified: NewContextified(g), state: keybase1.MobileNetworkState_NOTAVAILABLE, + changed: make(chan struct{}), } } -// NextUpdate returns a channel that triggers when the network state changes -func (a *MobileNetState) NextUpdate(lastState *keybase1.MobileNetworkState) chan keybase1.MobileNetworkState { +// NextUpdate returns a channel that will be closed the next time the network +// state changes. If lastState does not match the current state, an +// already-closed channel is returned so the caller wakes immediately. +func (a *MobileNetState) NextUpdate(lastState keybase1.MobileNetworkState) <-chan struct{} { a.Lock() defer a.Unlock() - ch := make(chan keybase1.MobileNetworkState, 1) - if lastState != nil && *lastState != a.state { - ch <- a.state - } else { - a.updateChs = append(a.updateChs, ch) + if lastState != a.state { + return alreadyClosed } - return ch + return a.changed } // Update updates the current network state, and notifies any waiting calls @@ -154,10 +169,8 @@ func (a *MobileNetState) Update(state keybase1.MobileNetworkState) { a.G().Log.Debug("MobileNetState.Update: useful update: %v, we are currently in state: %v", state, a.state) a.state = state - for _, ch := range a.updateChs { - ch <- state - } - a.updateChs = nil + close(a.changed) + a.changed = make(chan struct{}) } else { a.G().Log.Debug("MobileNetState.Update: ignoring update: %v, we are currently in state: %v", state, a.state) @@ -187,12 +200,14 @@ const ( type DesktopAppState struct { Contextified sync.Mutex - provider rpc.Transporter - suspended bool - locked bool - updateSuspendChs []chan bool - wakeWatcherOnce sync.Once - wakeWatcherStop chan struct{} + provider rpc.Transporter + suspended bool + locked bool + // suspendChanged is closed and replaced whenever suspended actually + // changes; readers wake and re-read Suspended(). + suspendChanged chan struct{} + wakeWatcherOnce sync.Once + wakeWatcherStop chan struct{} // wokeAt is the last time the wake watcher saw the machine come back from // sleep without a corresponding power event (dark wake, lost "suspend" // event, or no GUI connected to send one). @@ -200,7 +215,11 @@ type DesktopAppState struct { } func NewDesktopAppState(g *GlobalContext) *DesktopAppState { - d := &DesktopAppState{Contextified: NewContextified(g), wakeWatcherStop: make(chan struct{})} + d := &DesktopAppState{ + Contextified: NewContextified(g), + suspendChanged: make(chan struct{}), + wakeWatcherStop: make(chan struct{}), + } g.PushShutdownHook(func(mctx MetaContext) error { d.Lock() defer d.Unlock() @@ -249,16 +268,24 @@ func (a *DesktopAppState) wakeWatchLoop() { } } -func (a *DesktopAppState) NextSuspendUpdate(lastState *bool) chan bool { +// NextSuspendUpdate returns a channel that will be closed the next time the +// suspend state changes. If lastState does not match the current suspend +// state, an already-closed channel is returned so the caller wakes +// immediately and can re-fetch via Suspended(). +func (a *DesktopAppState) NextSuspendUpdate(lastState bool) <-chan struct{} { a.Lock() defer a.Unlock() - ch := make(chan bool, 1) - if lastState != nil && *lastState != a.suspended { - ch <- a.suspended - } else { - a.updateSuspendChs = append(a.updateSuspendChs, ch) + if lastState != a.suspended { + return alreadyClosed } - return ch + return a.suspendChanged +} + +// Suspended returns the current suspended state. +func (a *DesktopAppState) Suspended() bool { + a.Lock() + defer a.Unlock() + return a.suspended } // event from power monitor @@ -268,6 +295,7 @@ func (a *DesktopAppState) Update(mctx MetaContext, event string, provider rpc.Tr a.Lock() defer a.Unlock() a.provider = provider + prevSuspended := a.suspended switch event { case "suspend": a.suspended = true @@ -284,10 +312,10 @@ func (a *DesktopAppState) Update(mctx MetaContext, event string, provider rpc.Tr a.locked = false a.wokeAt = time.Time{} } - for _, ch := range a.updateSuspendChs { - ch <- a.suspended + if a.suspended != prevSuspended { + close(a.suspendChanged) + a.suspendChanged = make(chan struct{}) } - a.updateSuspendChs = nil } func (a *DesktopAppState) Disconnected(provider rpc.Transporter) { @@ -320,6 +348,11 @@ func (a *DesktopAppState) AwakeAndUnlocked(mctx MetaContext) bool { } func (a *DesktopAppState) resetLocked() { + prevSuspended := a.suspended a.suspended = false a.locked = false + if prevSuspended { + close(a.suspendChanged) + a.suspendChanged = make(chan struct{}) + } } diff --git a/go/libkb/leveldb_cleaner.go b/go/libkb/leveldb_cleaner.go index 145ce3b50625..891329c23049 100644 --- a/go/libkb/leveldb_cleaner.go +++ b/go/libkb/leveldb_cleaner.go @@ -134,7 +134,8 @@ func (c *levelDbCleaner) monitorAppState(stopCh chan struct{}) { state := keybase1.MobileAppState_FOREGROUND for { select { - case state = <-c.G().MobileAppState.NextUpdate(&state): + case <-c.G().MobileAppState.NextUpdate(state): + state = c.G().MobileAppState.State() switch state { case keybase1.MobileAppState_BACKGROUNDACTIVE: default: diff --git a/go/service/gregor.go b/go/service/gregor.go index c3d2e77a89b9..b48065ec0a88 100644 --- a/go/service/gregor.go +++ b/go/service/gregor.go @@ -277,7 +277,8 @@ func (g *gregorHandler) monitorAppState() { for { monitorAction := monitorNoop select { - case state = <-g.G().MobileAppState.NextUpdate(&state): + case <-g.G().MobileAppState.NextUpdate(state): + state = g.G().MobileAppState.State() switch state { case keybase1.MobileAppState_FOREGROUND: g.forcePing(ctx) @@ -287,7 +288,8 @@ func (g *gregorHandler) monitorAppState() { case keybase1.MobileAppState_BACKGROUND, keybase1.MobileAppState_INACTIVE: monitorAction = monitorDisconnect } - case suspended = <-g.G().DesktopAppState.NextSuspendUpdate(&suspended): + case <-g.G().DesktopAppState.NextSuspendUpdate(suspended): + suspended = g.G().DesktopAppState.Suspended() if !suspended { monitorAction = monitorConnect g.chatLog.Debug(ctx, "resumed, connecting") diff --git a/go/stellar/util.go b/go/stellar/util.go index 7991fc0fbd1d..3a26ff5b9d2b 100644 --- a/go/stellar/util.go +++ b/go/stellar/util.go @@ -101,11 +101,11 @@ func EmptyAmountStack(mctx libkb.MetaContext) { func cancelOnMobileBackground(mctx libkb.MetaContext) (libkb.MetaContext, context.CancelFunc) { mctx, cancel := mctx.WithContextCancel() go func() { + const foreground = keybase1.MobileAppState_FOREGROUND for { - foreground := keybase1.MobileAppState_FOREGROUND select { - case state := <-mctx.G().MobileAppState.NextUpdate(&foreground): - if state != foreground { + case <-mctx.G().MobileAppState.NextUpdate(foreground): + if mctx.G().MobileAppState.State() != foreground { cancel() return } From 96c48056e724e29504a000df0b868b73b05a3327 Mon Sep 17 00:00:00 2001 From: Song Gao Date: Fri, 31 Jul 2026 22:31:44 -0700 Subject: [PATCH 2/4] test(libkbfs): release testAppStateUpdater lock around blocking record send The mock's NextNetworkStateUpdate held tasu.lock across the blocking tasu.calls <- lastState send. On the test's happy path this is harmless because the test always reads callCh before touching setNetworkState / NetworkState. But if any assertion earlier in TestPrefetcherCellularPause fails, t.Fatal fires Goexit; the deferred shutdownPrefetcherTest then waits on a prefetcher goroutine that is stuck in the mock holding the lock, and any shutdown path that acquires the lock deadlocks - a clean test failure turns into a 5-minute go-test timeout with a large goroutine dump. Snapshot state and decrement the call counter under the lock, then do the blocking send after unlocking. Made-with: Claude --- go/kbfs/libkbfs/prefetcher_test.go | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/go/kbfs/libkbfs/prefetcher_test.go b/go/kbfs/libkbfs/prefetcher_test.go index 87ad58061c61..c4691607bf33 100644 --- a/go/kbfs/libkbfs/prefetcher_test.go +++ b/go/kbfs/libkbfs/prefetcher_test.go @@ -2351,18 +2351,27 @@ func (tasu *testAppStateUpdater) NextAppStateUpdate( func (tasu *testAppStateUpdater) NextNetworkStateUpdate( lastState keybase1.MobileNetworkState, ) <-chan struct{} { + // Snapshot state and decrement nCalls under the lock, but do the blocking + // send on tasu.calls outside the lock so a stalled receiver can't deadlock + // concurrent setNetworkState / NetworkState calls. tasu.lock.Lock() - defer tasu.lock.Unlock() - if tasu.nCalls > 0 { - tasu.calls <- lastState + shouldRecord := tasu.nCalls > 0 + if shouldRecord { tasu.nCalls-- } - if lastState != tasu.netState { - ch := make(chan struct{}) - close(ch) - return ch + stale := lastState != tasu.netState + ch := tasu.changed + tasu.lock.Unlock() + + if shouldRecord { + tasu.calls <- lastState + } + if stale { + closedCh := make(chan struct{}) + close(closedCh) + return closedCh } - return tasu.changed + return ch } func (tasu *testAppStateUpdater) AppState() keybase1.MobileAppState { From f9c3934a5fd8b478650ae5dcd7f547898208d8ed Mon Sep 17 00:00:00 2001 From: Song Gao Date: Fri, 7 Aug 2026 10:57:25 -0700 Subject: [PATCH 3/4] fix(libfs): unsubscribe folder-branch observers registered by SubscribeToObsolete SubscribeToObsolete registered a folder-branch observer via Notifier().RegisterForChanges but never unregistered it. The observer kept a reference to a closure that closed the returned channel on the next TlfHandleChange - a rare event that typically never fires on long-lived server deployments. On kbpagesd each unique-TLF site-cache miss added one observer to that TLF's folderBranchOps.observers slice for the process lifetime; on libhttpserver the LRU used lru.New (no evict callback), so observers stayed attached to popular TLFs across cache churn. Return an idempotent unsubscribe from SubscribeToObsolete and wire it into both callers: * libpages/root.MakeFS: chain unsubscribe into the returned shutdown func, and call it from the error-defer so a partial construction doesn't leak either. * libhttpserver/server: switch s.fs to lru.NewWithEvict with an eviction callback that unsubscribes, store the unsubscribe on obsoleteTrackingFS, and Purge the LRU in Server.Shutdown so any entries still resident get their observers released. Made-with: Claude --- go/kbfs/libfs/fs.go | 31 ++++++++++++++++++++++--------- go/kbfs/libhttpserver/server.go | 21 ++++++++++++++++----- go/kbfs/libpages/root.go | 12 ++++++++++-- 3 files changed, 48 insertions(+), 16 deletions(-) diff --git a/go/kbfs/libfs/fs.go b/go/kbfs/libfs/fs.go index b32528269f8b..3167717b5f89 100644 --- a/go/kbfs/libfs/fs.go +++ b/go/kbfs/libfs/fs.go @@ -1154,22 +1154,35 @@ func (o folderHandleChangeObserver) TlfHandleChange( // 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 + var closeOnce sync.Once onHandleChange := folderHandleChangeObserver( - func() { once.Do(func() { close(c) }) }) + 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. diff --git a/go/kbfs/libhttpserver/server.go b/go/kbfs/libhttpserver/server.go index 0f152d1a569e..fb711fb8233a 100644 --- a/go/kbfs/libhttpserver/server.go +++ b/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 } @@ -280,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 { @@ -305,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() } diff --git a/go/kbfs/libpages/root.go b/go/kbfs/libpages/root.go index 80d85a9c1139..669c35ba2e86 100644 --- a/go/kbfs/libpages/root.go +++ b/go/kbfs/libpages/root.go @@ -138,6 +138,7 @@ func (r *Root) MakeFS( fs CacheableFS, tlfID tlf.ID, shutdown func(), err error, ) { fsCtx, cancel := context.WithCancel(context.Background()) + var unsubscribe func() defer func() { zapFields := []zapcore.Field{ zap.String("root_type", r.Type.String()), @@ -148,6 +149,9 @@ func (r *Root) MakeFS( if err == nil { log.Info("root.MakeFS", zapFields...) } else { + if unsubscribe != nil { + unsubscribe() + } cancel() log.Warn("root.MakeFS", append(zapFields, zap.Error(err))...) } @@ -173,10 +177,11 @@ func (r *Root) MakeFS( if err != nil { return CacheableFS{}, tlf.ID{}, nil, err } - obsoleteCh, err := tlfFS.SubscribeToObsolete() + obsoleteCh, unsub, err := tlfFS.SubscribeToObsolete() if err != nil { return CacheableFS{}, tlf.ID{}, nil, err } + unsubscribe = unsub cacheableFS := CacheableFS{ obsoleteTrackingCh: obsoleteCh, tlfFS: tlfFS, @@ -185,7 +190,10 @@ func (r *Root) MakeFS( if _, err = cacheableFS.Use(); err != nil { return CacheableFS{}, tlf.ID{}, nil, err } - return cacheableFS, tlfHandle.TlfID(), cancel, nil + return cacheableFS, tlfHandle.TlfID(), func() { + unsub() + cancel() + }, nil case GitRoot: tlfHandle, err := libkbfs.GetHandleFromFolderNameAndType( ctx, kbfsConfig.KBPKI(), kbfsConfig.MDOps(), kbfsConfig, From feb8bcf936f1a9da525adb05816c4f302b8bd004 Mon Sep 17 00:00:00 2001 From: Song Gao Date: Fri, 7 Aug 2026 14:11:56 -0700 Subject: [PATCH 4/4] test(libfs): smoke test for SubscribeToObsolete + make observer comparable Add TestSubscribeToObsolete: verifies the (ch, unsubscribe, err) return contract, that the channel isn't spuriously closed, and that unsubscribe is idempotent. Doesn't attempt to prove the leak is fixed; scope is regression coverage for the API shape. The test surfaced a latent bug: libkbfs.observerList.remove uses == to find the entry to drop, which panics when the Observer's concrete type is a func (as folderHandleChangeObserver was). No prior caller ever unregistered a folderHandleChangeObserver, so this was never hit before. Convert it to a struct with an onChange func field and register a *folderHandleChangeObserver so the observer list can compare by pointer. Made-with: Claude --- go/kbfs/libfs/fs.go | 20 +++++++++++++------- go/kbfs/libfs/fs_test.go | 24 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 7 deletions(-) diff --git a/go/kbfs/libfs/fs.go b/go/kbfs/libfs/fs.go index 3167717b5f89..f4b2cc71de1c 100644 --- a/go/kbfs/libfs/fs.go +++ b/go/kbfs/libfs/fs.go @@ -1136,20 +1136,25 @@ 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 @@ -1168,8 +1173,9 @@ func (fs *FS) SubscribeToObsolete() ( c := make(chan struct{}) var closeOnce sync.Once - onHandleChange := folderHandleChangeObserver( - func() { closeOnce.Do(func() { close(c) }) }) + onHandleChange := &folderHandleChangeObserver{ + onChange: func() { closeOnce.Do(func() { close(c) }) }, + } fb := fs.root.GetFolderBranch() if err := fs.config.Notifier().RegisterForChanges( []data.FolderBranch{fb}, onHandleChange); err != nil { diff --git a/go/kbfs/libfs/fs_test.go b/go/kbfs/libfs/fs_test.go index f13f23a33e43..365341f41662 100644 --- a/go/kbfs/libfs/fs_test.go +++ b/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() +}