refactor: background work reads app state in its own loop

go/avatars/appstate.go

@@ -0,0 +1,64 @@
+package avatars
+
+import (
+	"sync"
+
+	"github.com/keybase/client/go/libkb"
+	"github.com/keybase/client/go/protocol/keybase1"
+)
+
+// backgroundFlusher runs flush each time the app enters BACKGROUND, from
+// start until stop.
+type backgroundFlusher struct {
+	mu     sync.Mutex
+	stopCh chan struct{}
+	doneCh chan struct{}
+	// flushes counts flushes; tests use it.
+	flushes int
+}
+
+func (f *backgroundFlusher) start(m libkb.MetaContext, flush func(libkb.MetaContext)) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.stopCh != nil {
+		return
+	}
+	f.stopCh = make(chan struct{})
+	f.doneCh = make(chan struct{})
+	stopCh, doneCh := f.stopCh, f.doneCh
+	// Armed here, not in the goroutine, so a change made before the goroutine
+	// first runs still wakes it.
+	state := m.G().MobileAppState.State()
+	changed := m.G().MobileAppState.NextUpdate(state)
+	go func() {
+		defer close(doneCh)
+		for {
+			select {
+			case <-changed:
+			case <-stopCh:
+				return
+			}
+			state = m.G().MobileAppState.State()
+			changed = m.G().MobileAppState.NextUpdate(state)
+			if state == keybase1.MobileAppState_BACKGROUND {
+				flush(m)
+				f.mu.Lock()
+				f.flushes++
+				f.mu.Unlock()
+			}
+		}
+	}()
+}
+
+// stop ends the watcher goroutine and waits for it to exit.
+func (f *backgroundFlusher) stop() {
+	f.mu.Lock()
+	stopCh, doneCh := f.stopCh, f.doneCh
+	f.stopCh, f.doneCh = nil, nil
+	f.mu.Unlock()
+	if stopCh == nil {
+		return
+	}
+	close(stopCh)
+	<-doneCh
+}

go/avatars/appstate_test.go

@@ -0,0 +1,121 @@
+package avatars
+
+import (
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/keybase/client/go/libkb"
+	"github.com/keybase/client/go/protocol/keybase1"
+	"github.com/stretchr/testify/require"
+)
+
+// waitFlushes waits until f has flushed at least want times.
+func waitFlushes(t *testing.T, f *backgroundFlusher, want int) {
+	t.Helper()
+	require.Eventually(t, func() bool {
+		return flushes(f) >= want
+	}, 10*time.Second, time.Millisecond, "did not reach %d flushes", want)
+}
+
+func flushes(f *backgroundFlusher) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.flushes
+}
+
+type bgSource interface {
+	libkb.AvatarLoaderSource
+	flusher() *backgroundFlusher
+}
+
+func (c *FullCachingSource) flusher() *backgroundFlusher { return &c.bgFlusher }
+func (c *URLCachingSource) flusher() *backgroundFlusher  { return &c.bgFlusher }
+
+func forEachSource(t *testing.T, f func(t *testing.T, tc libkb.TestContext, s bgSource)) {
+	sources := map[string]func(t *testing.T, g *libkb.GlobalContext) bgSource{
+		"full": func(t *testing.T, g *libkb.GlobalContext) bgSource {
+			s := NewFullCachingSource(g, time.Hour, 10)
+			s.tempDir = t.TempDir()
+			return s
+		},
+		"url": func(_ *testing.T, _ *libkb.GlobalContext) bgSource {
+			return NewURLCachingSource(time.Hour, 10)
+		},
+	}
+	for name, mk := range sources {
+		t.Run(name, func(t *testing.T) {
+			tc := libkb.SetupTest(t, "avatars", 1)
+			defer tc.Cleanup()
+			f(t, tc, mk(t, tc.G))
+		})
+	}
+}
+
+func TestAvatarsFlushSeedsFromState(t *testing.T) {
+	forEachSource(t, func(t *testing.T, tc libkb.TestContext, s bgSource) {
+		m := libkb.NewMetaContextForTest(tc)
+		tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND)
+		s.StartBackgroundTasks(m)
+		defer s.StopBackgroundTasks(m)
+
+		for _, next := range []keybase1.MobileAppState{
+			keybase1.MobileAppState_FOREGROUND,
+			keybase1.MobileAppState_INACTIVE,
+			keybase1.MobileAppState_BACKGROUND,
+		} {
+			tc.G.MobileAppState.Update(next)
+		}
+		waitFlushes(t, s.flusher(), 1)
+		require.Equal(t, 1, flushes(s.flusher()),
+			"flushed on start already being in BACKGROUND, or flushed more than once for one transition into it")
+	})
+}
+
+func TestAvatarsMonitorExitsOnStop(t *testing.T) {
+	forEachSource(t, func(t *testing.T, tc libkb.TestContext, s bgSource) {
+		m := libkb.NewMetaContextForTest(tc)
+		// Warm up lazily started goroutines before taking the baseline.
+		s.StartBackgroundTasks(m)
+		s.StopBackgroundTasks(m)
+		baseline := runtime.NumGoroutine()
+
+		const cycles = 50
+		for range cycles {
+			s.StartBackgroundTasks(m)
+			s.StopBackgroundTasks(m)
+		}
+		require.Eventually(t, func() bool {
+			return runtime.NumGoroutine() < baseline+cycles/2
+		}, 10*time.Second, 10*time.Millisecond, "goroutines leaked across Start/Stop")
+	})
+}
+
+// Start/Stop racing app-state changes neither deadlocks nor leaks.
+func TestAvatarsMonitorStress(t *testing.T) {
+	forEachSource(t, func(t *testing.T, tc libkb.TestContext, s bgSource) {
+		m := libkb.NewMetaContextForTest(tc)
+		baseline := runtime.NumGoroutine()
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			states := []keybase1.MobileAppState{
+				keybase1.MobileAppState_FOREGROUND,
+				keybase1.MobileAppState_INACTIVE,
+				keybase1.MobileAppState_BACKGROUND,
+				keybase1.MobileAppState_BACKGROUNDACTIVE,
+			}
+			for i := range 400 {
+				tc.G.MobileAppState.Update(states[i%len(states)])
+			}
+		}()
+		for range 100 {
+			s.StartBackgroundTasks(m)
+			s.StopBackgroundTasks(m)
+		}
+		<-done
+		require.Eventually(t, func() bool {
+			return runtime.NumGoroutine() < baseline+10
+		}, 10*time.Second, 10*time.Millisecond, "goroutines leaked")
+	})
+}

go/avatars/fullcaching.go

@@ -91,6 +91,7 @@ type FullCachingSource struct {
 	started              bool
 	diskLRU              *lru.DiskLRU
 	diskLRUCleanerCancel context.CancelFunc
+	bgFlusher            backgroundFlusher
 	staleThreshold       time.Duration
 	simpleSource         libkb.AvatarLoaderSource
 
@@ -212,10 +213,15 @@ func (c *FullCachingSource) StartBackgroundTasks(mctx libkb.MetaContext) {
 		return
 	}
 	c.started = true
-	go c.monitorAppState(mctx)
+	c.bgFlusher.start(mctx, func(m libkb.MetaContext) {
+		c.debug(m, "backgroundFlusher: flushing diskLRU")
+		if err := c.diskLRU.Flush(m.Ctx(), m.G()); err != nil {
+			c.debug(m, "backgroundFlusher: unable to flush diskLRU %v", err)
+		}
+	})
 	c.populateCacheCh = make(chan populateArg, 100)
 	for range 10 {
-		go c.populateCacheWorker(mctx)
+		go c.populateCacheWorker(mctx, c.populateCacheCh)
 	}
 	mctx, cancel := mctx.WithContextCancel()
 	c.diskLRUCleanerCancel = cancel
@@ -230,6 +236,7 @@ func (c *FullCachingSource) StopBackgroundTasks(mctx libkb.MetaContext) {
 		return
 	}
 	c.started = false
+	c.bgFlusher.stop()
 	close(c.populateCacheCh)
 	if c.diskLRUCleanerCancel != nil {
 		c.diskLRUCleanerCancel()
@@ -251,21 +258,6 @@ func (c *FullCachingSource) isStale(m libkb.MetaContext, item lru.DiskLRUEntry)
 	return m.G().GetClock().Now().Sub(item.Ctime) > c.staleThreshold
 }
 
-func (c *FullCachingSource) monitorAppState(m libkb.MetaContext) {
-	c.debug(m, "monitorAppState: starting up")
-	state := keybase1.MobileAppState_FOREGROUND
-	for {
-		<-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 {
-				c.debug(m, "monitorAppState: unable to flush diskLRU %v", err)
-			}
-		}
-	}
-}
-
 func (c *FullCachingSource) processLRUHit(entry lru.DiskLRUEntry) (res lruEntry) {
 	var ok bool
 	if _, ok = entry.Value.(map[string]any); ok {
@@ -392,8 +384,8 @@ func (c *FullCachingSource) removeFile(m libkb.MetaContext, ent *lru.DiskLRUEntr
 	}
 }
 
-func (c *FullCachingSource) populateCacheWorker(m libkb.MetaContext) {
-	for arg := range c.populateCacheCh {
+func (c *FullCachingSource) populateCacheWorker(m libkb.MetaContext, populateCacheCh <-chan populateArg) {
+	for arg := range populateCacheCh {
 		err := c.populateCacheJob(m, arg)
 		if err != nil {
 			c.debug(m, "populateCacheWorker: %s", err)

go/avatars/urlcaching.go

@@ -14,6 +14,7 @@ type URLCachingSource struct {
 	diskLRU        *lru.DiskLRU
 	staleThreshold time.Duration
 	simpleSource   *SimpleSource
+	bgFlusher      backgroundFlusher
 
 	// testing only
 	staleFetchCh chan struct{}
@@ -30,10 +31,14 @@ func NewURLCachingSource(staleThreshold time.Duration, size int) *URLCachingSour
 }
 
 func (c *URLCachingSource) StartBackgroundTasks(m libkb.MetaContext) {
-	go c.monitorAppState(m)
+	c.bgFlusher.start(m, func(m libkb.MetaContext) {
+		c.debug(m, "backgroundFlusher: flushing diskLRU")
+		c.diskLRU.Flush(m.Ctx(), m.G())
+	})
 }
 
 func (c *URLCachingSource) StopBackgroundTasks(m libkb.MetaContext) {
+	c.bgFlusher.stop()
 	c.diskLRU.Flush(m.Ctx(), m.G())
 }
 
@@ -49,19 +54,6 @@ func (c *URLCachingSource) isStale(m libkb.MetaContext, item lru.DiskLRUEntry) b
 	return m.G().GetClock().Now().Sub(item.Ctime) > c.staleThreshold
 }
 
-func (c *URLCachingSource) monitorAppState(m libkb.MetaContext) {
-	c.debug(m, "monitorAppState: starting up")
-	state := keybase1.MobileAppState_FOREGROUND
-	for {
-		<-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())
-		}
-	}
-}
-
 func (c *URLCachingSource) specLoad(m libkb.MetaContext, names []string, formats []keybase1.AvatarFormat) (res avatarLoadSpec, err error) {
 	for _, name := range names {
 		for _, format := range formats {