diff --git a/go/avatars/appstate.go b/go/avatars/appstate.go new file mode 100644 index 000000000000..39e15f0d4827 --- /dev/null +++ b/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 +} diff --git a/go/avatars/appstate_test.go b/go/avatars/appstate_test.go new file mode 100644 index 000000000000..5b03630d6478 --- /dev/null +++ b/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") + }) +} diff --git a/go/avatars/fullcaching.go b/go/avatars/fullcaching.go index e934f896d2d0..cea758e3abfa 100644 --- a/go/avatars/fullcaching.go +++ b/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) diff --git a/go/avatars/urlcaching.go b/go/avatars/urlcaching.go index 96b543848302..0adf0b7accc9 100644 --- a/go/avatars/urlcaching.go +++ b/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 { diff --git a/go/chat/archive.go b/go/chat/archive.go index 69af6e9cdfb1..8d991f73c1f9 100644 --- a/go/chat/archive.go +++ b/go/chat/archive.go @@ -42,11 +42,23 @@ type ChatArchiveRegistry struct { flushDelay time.Duration stopCh chan struct{} clock clockwork.Clock - eg errgroup.Group + // eg holds the current run's loop. Each run gets its own, since Stop + // waits on it from a goroutine and a Group cannot be added to while + // somebody waits on it. + eg *errgroup.Group // Changes to flush to disk? dirty bool remoteClient func() chat1.RemoteInterface runningJobs map[chat1.ArchiveJobID]types.PauseArchiveFn + // launching holds jobs started by a resume that have not registered as + // running yet, so an overlapping resume does not start them again. Each + // entry is its launch's number, so a launch that ends clears only its + // own entry and never that of a later launch of the same job. + launching map[chat1.ArchiveJobID]uint64 + lastLaunch uint64 + // runJob, if set, runs a launched job in place of a ChatArchiver. Tests + // only. + runJob func(ctx context.Context, uid gregor1.UID, req chat1.ArchiveChatJobRequest) error edb *encrypteddb.EncryptedDB jobHistory chat1.ArchiveChatHistory @@ -66,6 +78,15 @@ func NewArchiveJobNotFoundError(jobID chat1.ArchiveJobID) ArchiveJobNotFoundErro var _ error = ArchiveJobNotFoundError{} +// errArchiveJobBackgroundPaused is returned by Set when a job registers as +// running while the app is out of the foreground, so the job stops at once. +var errArchiveJobBackgroundPaused = errors.New("archive job paused: app not in foreground") + +// errArchiveJobOtherUser is returned by Set for a job of a user other than the +// one the registry runs for: one launched before a logout, still running into +// the next user's run. +var errArchiveJobOtherUser = errors.New("archive job belongs to another user") + func NewChatArchiveRegistry(g *globals.Context, remoteClient func() chat1.RemoteInterface) *ChatArchiveRegistry { keyFn := func(ctx context.Context) ([32]byte, error) { return storage.GetSecretBoxKey(ctx, g.ExternalG()) @@ -80,6 +101,7 @@ func NewChatArchiveRegistry(g *globals.Context, remoteClient func() chat1.Remote clock: clockwork.NewRealClock(), flushDelay: 15 * time.Second, runningJobs: make(map[chat1.ArchiveJobID]types.PauseArchiveFn), + launching: make(map[chat1.ArchiveJobID]uint64), jobHistory: chat1.ArchiveChatHistory{JobHistory: make(map[chat1.ArchiveJobID]chat1.ArchiveChatJob)}, edb: encrypteddb.New(g.ExternalG(), dbFn, keyFn), } @@ -135,94 +157,142 @@ func (r *ChatArchiveRegistry) flushLocked(ctx context.Context) error { return nil } -func (r *ChatArchiveRegistry) flushLoop(stopCh chan struct{}) error { +// archiveRunsIn is whether jobs run in state; they pause in every other. +func archiveRunsIn(state keybase1.MobileAppState) bool { + return state == keybase1.MobileAppState_FOREGROUND +} + +// runEnded reports whether the run stopCh belongs to is over. Stop closes +// stopCh under r's lock, so under that lock a closed channel means the run is +// over, whether or not a later Start (possibly for another user) has since +// replaced r.stopCh. +func runEnded(stopCh chan struct{}) bool { + select { + case <-stopCh: + return true + default: + return false + } +} + +func (r *ChatArchiveRegistry) flush(ctx context.Context, stopCh chan struct{}) { + var err error + defer r.Trace(ctx, &err, "flush")() + r.Lock() + defer r.Unlock() + if runEnded(stopCh) { + return + } + err = r.flushLocked(ctx) +} + +func (r *ChatArchiveRegistry) bgPauseAllJobs(ctx context.Context, stopCh chan struct{}) { + r.Lock() + defer r.Unlock() + if runEnded(stopCh) { + return + } + _ = r.bgPauseAllJobsLocked(ctx) +} + +// loop runs one run of the registry until stopCh closes: it flushes on a +// timer, pauses running jobs whenever the app leaves the foreground, and +// resumes paused jobs once the app has been in the foreground for +// resumeJobsDelay. +func (r *ChatArchiveRegistry) loop(stopCh chan struct{}, state keybase1.MobileAppState) error { ctx := context.Background() - r.Debug(ctx, "flushLoop: starting") + r.Debug(ctx, "loop: starting in %v", state) + defer r.Debug(ctx, "loop: shutting down") + flushCh := r.clock.After(r.flushDelay) + resume := time.NewTimer(r.resumeJobsDelay) + if !archiveRunsIn(state) { + resume.Stop() + } + // changed is refreshed only when the loop reads a new state: a resume + // can skip on a state the loop has not seen yet, and a fresh NextUpdate + // taken after the state came back would miss that change. + changed := r.G().MobileAppState.NextUpdate(state) for { select { case <-stopCh: - r.Debug(ctx, "flushLoop: shutting down") return nil - case <-r.clock.After(r.flushDelay): - func() { - var err error - defer r.Trace(ctx, &err, "flushLoop")() - r.Lock() - defer r.Unlock() - err = r.flushLocked(ctx) - if err != nil { - r.Debug(ctx, "flushLoop: failed to flush: %s", err) - } - }() + case <-flushCh: + r.flush(ctx, stopCh) + flushCh = r.clock.After(r.flushDelay) + case <-changed: + state = r.G().MobileAppState.State() + changed = r.G().MobileAppState.NextUpdate(state) + r.Debug(ctx, "loop: next state -> %v", state) + if archiveRunsIn(state) { + resume.Reset(r.resumeJobsDelay) + } else { + resume.Stop() + r.bgPauseAllJobs(ctx, stopCh) + } + case <-resume.C: + if err := r.resumeAllBgJobs(ctx, stopCh); err != nil { + r.Debug(ctx, err.Error()) + } } } } func (r *ChatArchiveRegistry) resumeAllBgJobs(ctx context.Context, stopCh chan struct{}) (err error) { defer r.Trace(ctx, &err, "resumeAllBgJobs")() - select { - case <-stopCh: - return nil - case <-ctx.Done(): - return ctx.Err() - case <-time.After(r.resumeJobsDelay): - } r.Lock() defer r.Unlock() + if runEnded(stopCh) { + return nil + } + if state := r.G().MobileAppState.State(); !archiveRunsIn(state) { + r.Debug(ctx, "resumeAllBgJobs: not resuming in %v", state) + return nil + } err = r.initLocked(ctx) if err != nil { return err } for _, job := range r.jobHistory.JobHistory { if job.Status == chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED { - go func(job chat1.ArchiveChatJob) { - ctx := globals.ChatCtx(context.Background(), r.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, NewSimpleIdentifyNotifier(r.G())) - _, err := NewChatArchiver(r.G(), r.uid, r.remoteClient).ArchiveChat(ctx, job.Request) - if err != nil { - r.Debug(ctx, err.Error()) - } - }(job) + r.launchLocked(ctx, job.Request) } } return nil } -func (r *ChatArchiveRegistry) monitorAppState(stopCh chan struct{}) error { - appState := keybase1.MobileAppState_FOREGROUND - ctx, cancel := context.WithCancel(context.Background()) - for { - select { - case <-stopCh: - cancel() - return nil - 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 func() { - ierr := r.resumeAllBgJobs(ctx, stopCh) - if ierr != nil { - r.Debug(ctx, ierr.Error()) - } - }() - default: - cancel() - ctx, cancel = context.WithCancel(context.Background()) - - func() { - var err error - defer r.Trace(ctx, &err, "monitorAppState")() - r.Lock() - defer r.Unlock() - err = r.bgPauseAllJobsLocked(ctx) - }() - } - } +// launchLocked runs a job in the background unless an earlier launch of it +// has not registered yet. The job registers itself as running through Set. +func (r *ChatArchiveRegistry) launchLocked(ctx context.Context, req chat1.ArchiveChatJobRequest) { + jobID := req.JobID + if _, ok := r.launching[jobID]; ok { + r.Debug(ctx, "launch: %v is already starting", jobID) + return } + r.lastLaunch++ + launch := r.lastLaunch + r.launching[jobID] = launch + uid, runJob := r.uid, r.runJob + go func() { + ctx := globals.ChatCtx(context.Background(), r.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, NewSimpleIdentifyNotifier(r.G())) + var err error + if runJob != nil { + err = runJob(ctx, uid, req) + } else { + _, err = NewChatArchiver(r.G(), uid, r.remoteClient).ArchiveChat(ctx, req) + } + if err != nil { + r.Debug(ctx, err.Error()) + } + r.Lock() + defer r.Unlock() + if r.launching[jobID] == launch { + delete(r.launching, jobID) + } + }() } -// Resumes previously BACKGROUND_PAUSED jobs, after a delay. +// Resumes previously BACKGROUND_PAUSED jobs, after a delay, if the app is in +// the foreground. func (r *ChatArchiveRegistry) Start(ctx context.Context, uid gregor1.UID) { defer r.Trace(ctx, nil, "Start")() r.Lock() @@ -233,15 +303,11 @@ func (r *ChatArchiveRegistry) Start(ctx context.Context, uid gregor1.UID) { r.uid = uid r.started = true r.stopCh = make(chan struct{}) + r.eg = new(errgroup.Group) stopCh := r.stopCh + state := r.G().MobileAppState.State() r.eg.Go(func() error { - return r.flushLoop(stopCh) - }) - r.eg.Go(func() error { - return r.resumeAllBgJobs(context.Background(), stopCh) - }) - r.eg.Go(func() error { - return r.monitorAppState(stopCh) + return r.loop(stopCh, state) }) } @@ -283,10 +349,16 @@ func (r *ChatArchiveRegistry) Stop(ctx context.Context) chan struct{} { r.Debug(ctx, err.Error()) } r.started = false + // The history belongs to this run's user, and the pause above flushed + // it. The next Start may be for another user, so it reads its own. + r.inited = false + r.dirty = false + r.jobHistory = chat1.ArchiveChatHistory{JobHistory: make(map[chat1.ArchiveJobID]chat1.ArchiveChatJob)} close(r.stopCh) + eg := r.eg go func() { r.Debug(context.Background(), "Stop: waiting for shutdown") - _ = r.eg.Wait() + _ = eg.Wait() r.Debug(context.Background(), "Stop: shutdown complete") close(ch) }() @@ -381,28 +453,47 @@ func (r *ChatArchiveRegistry) Delete(ctx context.Context, jobID chat1.ArchiveJob return nil } -func (r *ChatArchiveRegistry) Set(ctx context.Context, cancel types.PauseArchiveFn, job chat1.ArchiveChatJob) (err error) { +func (r *ChatArchiveRegistry) Set(ctx context.Context, uid gregor1.UID, cancel types.PauseArchiveFn, job chat1.ArchiveChatJob) (err error) { defer r.Trace(ctx, &err, "Set(%v) -> %v", job.Request.JobID, job.Status)() r.Lock() defer r.Unlock() + if !r.uid.Eq(uid) { + if cancel != nil { + cancel() + } + return errArchiveJobOtherUser + } err = r.initLocked(ctx) if err != nil { return err } jobID := job.Request.JobID + var pausedErr error switch job.Status { case chat1.ArchiveChatJobStatus_COMPLETE, chat1.ArchiveChatJobStatus_ERROR: delete(r.runningJobs, jobID) case chat1.ArchiveChatJobStatus_RUNNING: - if cancel != nil { - r.runningJobs[jobID] = cancel + if cancel == nil { + break } + delete(r.launching, jobID) + // The loop pauses running jobs under this lock when the app leaves + // the foreground. A job registering while the app is out of it came + // after that pause, so it is paused here. + if state := r.G().MobileAppState.State(); !archiveRunsIn(state) { + r.Debug(ctx, "Set: pausing %v in %v", jobID, state) + cancel() + job.Status = chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED + pausedErr = errArchiveJobBackgroundPaused + break + } + r.runningJobs[jobID] = cancel } r.jobHistory.JobHistory[jobID] = job.DeepCopy() r.dirty = true - return nil + return pausedErr } func (r *ChatArchiveRegistry) Pause(ctx context.Context, jobID chat1.ArchiveJobID) (err error) { @@ -463,14 +554,7 @@ func (r *ChatArchiveRegistry) Resume(ctx context.Context, jobID chat1.ArchiveJob return fmt.Errorf("Cannot resume a non-paused job. Found status %v", job.Status) } - // Resume the job in the background, the job will register itself as running - go func() { - ctx := globals.ChatCtx(context.Background(), r.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, NewSimpleIdentifyNotifier(r.G())) - _, err := NewChatArchiver(r.G(), r.uid, r.remoteClient).ArchiveChat(ctx, job.Request) - if err != nil { - r.Debug(ctx, err.Error()) - } - }() + r.launchLocked(ctx, job.Request) return nil } @@ -556,7 +640,7 @@ func (c *ChatArchiver) checkpointConv(ctx context.Context, f *os.File, checkpoin // Add this conv's individual progress. job.Checkpoints[convID.DbShortFormString()] = checkpoint - err = c.G().ArchiveRegistry.Set(ctx, nil, *job) + err = c.G().ArchiveRegistry.Set(ctx, c.uid, nil, *job) return job.MessagesComplete, job.MessagesTotal, err } @@ -728,7 +812,7 @@ func (c *ChatArchiver) ArchiveChat(ctx context.Context, arg chat1.ArchiveChatJob } // Write even if our context was canceled - ierr := c.G().ArchiveRegistry.Set(context.TODO(), nil, jobInfo) + ierr := c.G().ArchiveRegistry.Set(context.TODO(), c.uid, nil, jobInfo) if ierr != nil { c.Debug(ctx, "ArchiveChat.cleanup %v", ierr) } @@ -743,7 +827,7 @@ func (c *ChatArchiver) ArchiveChat(ctx context.Context, arg chat1.ArchiveChatJob jobInfo.Err = "" // Update the store ASAP, we will update it again once we resolve the inbox query but that may take some time. - err = c.G().ArchiveRegistry.Set(ctx, pause, jobInfo) + err = c.G().ArchiveRegistry.Set(ctx, c.uid, pause, jobInfo) if err != nil { return "", err } @@ -778,7 +862,7 @@ func (c *ChatArchiver) ArchiveChat(ctx context.Context, arg chat1.ArchiveChatJob jobInfo.MessagesTotal = totalMsgs jobInfo.MatchingConvs = utils.PresentConversationLocals(ctx, c.G(), c.uid, convs, utils.PresentParticipantsModeSkip) - err = c.G().ArchiveRegistry.Set(ctx, nil, jobInfo) + err = c.G().ArchiveRegistry.Set(ctx, c.uid, nil, jobInfo) if err != nil { return "", err } diff --git a/go/chat/archive_appstate_test.go b/go/chat/archive_appstate_test.go new file mode 100644 index 000000000000..a8f43c58410d --- /dev/null +++ b/go/chat/archive_appstate_test.go @@ -0,0 +1,594 @@ +package chat + +import ( + "context" + "fmt" + "math/rand" + "runtime" + "sync" + "testing" + "time" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/encrypteddb" + "github.com/keybase/client/go/externalstest" + "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/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// archiveJobRunner stands in for ChatArchiver: a launched job waits for +// release, registers as running through Set, and runs until paused. +type archiveJobRunner struct { + r *ChatArchiveRegistry + release chan struct{} + + mu sync.Mutex + launches map[chat1.ArchiveJobID]int + active int + launched chan chat1.ArchiveJobID +} + +func (a *archiveJobRunner) run(ctx context.Context, uid gregor1.UID, req chat1.ArchiveChatJobRequest) error { + a.mu.Lock() + a.launches[req.JobID]++ + a.active++ + a.mu.Unlock() + defer func() { + a.mu.Lock() + a.active-- + a.mu.Unlock() + }() + select { + case a.launched <- req.JobID: + default: + } + <-a.release + pauseCh := make(chan struct{}) + var once sync.Once + pause := func() { once.Do(func() { close(pauseCh) }) } + job := chat1.ArchiveChatJob{Request: req, Status: chat1.ArchiveChatJobStatus_RUNNING} + if err := a.r.Set(ctx, uid, pause, job); err != nil { + return err + } + <-pauseCh + return nil +} + +func (a *archiveJobRunner) counts() (launches map[chat1.ArchiveJobID]int, active int) { + a.mu.Lock() + defer a.mu.Unlock() + launches = make(map[chat1.ArchiveJobID]int, len(a.launches)) + for id, n := range a.launches { + launches[id] = n + } + return launches, a.active +} + +var archiveTestJobIDs = []chat1.ArchiveJobID{"job-a", "job-b", "job-c"} + +// setupAppStateArchive returns a registry whose history holds paused jobs +// and is treated as already read from disk. +func setupAppStateArchive(t *testing.T, released bool) (*ChatArchiveRegistry, *archiveJobRunner, libkb.TestContext) { + tc := externalstest.SetupTest(t, "archive-appstate", 0) + t.Cleanup(tc.Cleanup) + g := globals.NewContext(tc.G, &globals.ChatContext{CtxFactory: appStateCtxFactory{}}) + r := NewChatArchiveRegistry(g, nil) + r.resumeJobsDelay = 0 + // The real key needs a logged-in user. + r.edb = encrypteddb.New(tc.G, func(g *libkb.GlobalContext) *libkb.JSONLocalDb { return g.LocalChatDb }, + func(context.Context) ([32]byte, error) { return [32]byte{}, nil }) + r.inited = true + for _, id := range archiveTestJobIDs { + r.jobHistory.JobHistory[id] = chat1.ArchiveChatJob{ + Request: chat1.ArchiveChatJobRequest{JobID: id}, + Status: chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED, + } + } + runner := &archiveJobRunner{ + r: r, + release: make(chan struct{}), + launches: make(map[chat1.ArchiveJobID]int), + launched: make(chan chat1.ArchiveJobID, 100), + } + if released { + close(runner.release) + } + r.runJob = runner.run + return r, runner, tc +} + +func archiveStatuses(r *ChatArchiveRegistry) (statuses map[chat1.ArchiveJobID]chat1.ArchiveChatJobStatus, running int) { + r.Lock() + defer r.Unlock() + statuses = make(map[chat1.ArchiveJobID]chat1.ArchiveChatJobStatus) + for id, job := range r.jobHistory.JobHistory { + statuses[id] = job.Status + } + return statuses, len(r.runningJobs) +} + +func requireArchiveStopped(t *testing.T, r *ChatArchiveRegistry) { + t.Helper() + select { + case <-r.Stop(context.TODO()): + case <-time.After(10 * time.Second): + require.FailNow(t, "Stop did not finish") + } +} + +func requireArchiveJobsRunning(t *testing.T, r *ChatArchiveRegistry) { + t.Helper() + require.Eventually(t, func() bool { + statuses, running := archiveStatuses(r) + for _, status := range statuses { + if status != chat1.ArchiveChatJobStatus_RUNNING { + return false + } + } + // Empty until a resume has read the history. + return len(statuses) > 0 && running == len(statuses) + }, 10*time.Second, time.Millisecond, "jobs did not resume") +} + +func requireArchiveJobsPaused(t *testing.T, r *ChatArchiveRegistry, runner *archiveJobRunner) { + t.Helper() + require.Eventually(t, func() bool { + statuses, running := archiveStatuses(r) + for _, status := range statuses { + if status != chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED { + return false + } + } + _, active := runner.counts() + return running == 0 && active == 0 + }, 10*time.Second, time.Millisecond, "jobs did not pause") +} + +func TestArchiveConcurrentResumesLaunchOnce(t *testing.T) { + r, runner, _ := setupAppStateArchive(t, false) + stopCh := make(chan struct{}) + r.Lock() + r.started = true + r.stopCh = stopCh + r.Unlock() + defer close(stopCh) + + var wg sync.WaitGroup + for range 20 { + wg.Go(func() { + assert.NoError(t, r.resumeAllBgJobs(context.Background(), stopCh)) + }) + } + wg.Wait() + for range archiveTestJobIDs { + select { + case <-runner.launched: + case <-time.After(10 * time.Second): + require.FailNow(t, "jobs did not launch") + } + } + launches, _ := runner.counts() + for _, id := range archiveTestJobIDs { + require.Equal(t, 1, launches[id], "launches of %v before it registered", id) + } + + close(runner.release) + requireArchiveJobsRunning(t, r) + for range 5 { + require.NoError(t, r.resumeAllBgJobs(context.Background(), stopCh)) + } + launches, _ = runner.counts() + for _, id := range archiveTestJobIDs { + require.Equal(t, 1, launches[id], "launches of %v after it registered", id) + } + r.Lock() + require.NoError(t, r.bgPauseAllJobsLocked(context.Background())) + r.Unlock() + requireArchiveJobsPaused(t, r, runner) +} + +// A run's loop that is still going after Stop neither pauses the jobs nor +// flushes the history of whatever run comes next. +func TestArchiveEndedRunLoopTouchesNothing(t *testing.T) { + r, _, _ := setupAppStateArchive(t, true) + ctx := context.Background() + r.Start(ctx, gregor1.UID([]byte{1, 2, 3, 4})) + defer requireArchiveStopped(t, r) + requireArchiveJobsRunning(t, r) + + ended := make(chan struct{}) + close(ended) + r.bgPauseAllJobs(ctx, ended) + statuses, running := archiveStatuses(r) + require.Equal(t, len(archiveTestJobIDs), running, "an ended run paused jobs") + for id, status := range statuses { + require.Equal(t, chat1.ArchiveChatJobStatus_RUNNING, status, "%v", id) + } + + r.Lock() + r.dirty = true + r.Unlock() + r.flush(ctx, ended) + r.Lock() + defer r.Unlock() + require.True(t, r.dirty, "an ended run flushed") +} + +// A job launched by one resume, passed over by a pause because it had not +// registered yet, and skipped by the next resume because it was still +// launching, runs once it registers in the foreground. +func TestArchiveRelaunchAfterPauseWhileLaunching(t *testing.T) { + r, runner, tc := setupAppStateArchive(t, false) + stopCh := make(chan struct{}) + r.Lock() + r.started = true + r.stopCh = stopCh + r.Unlock() + defer close(stopCh) + ctx := context.Background() + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + require.NoError(t, r.resumeAllBgJobs(ctx, stopCh)) + for range archiveTestJobIDs { + select { + case <-runner.launched: + case <-time.After(10 * time.Second): + require.FailNow(t, "jobs did not launch") + } + } + + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + r.Lock() + require.NoError(t, r.bgPauseAllJobsLocked(ctx)) + r.Unlock() + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + require.NoError(t, r.resumeAllBgJobs(ctx, stopCh)) + launches, _ := runner.counts() + for _, id := range archiveTestJobIDs { + require.Equal(t, 1, launches[id], "launches of %v", id) + } + + close(runner.release) + requireArchiveJobsRunning(t, r) + r.Lock() + require.NoError(t, r.bgPauseAllJobsLocked(ctx)) + r.Unlock() + requireArchiveJobsPaused(t, r, runner) +} + +// A job that registers as running while the app is not in the foreground is +// paused at once, and resumes on the next FOREGROUND. +func TestArchiveSetWhileInactivePauses(t *testing.T) { + r, _, tc := setupAppStateArchive(t, true) + ctx := context.Background() + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + uid := gregor1.UID([]byte{1, 2, 3, 4}) + r.Start(ctx, uid) + defer requireArchiveStopped(t, r) + + jobID := chat1.ArchiveJobID("job-manual") + paused := make(chan struct{}) + var once sync.Once + job := chat1.ArchiveChatJob{ + Request: chat1.ArchiveChatJobRequest{JobID: jobID}, + Status: chat1.ArchiveChatJobStatus_RUNNING, + } + require.ErrorIs(t, r.Set(ctx, uid, func() { once.Do(func() { close(paused) }) }, job), errArchiveJobBackgroundPaused) + select { + case <-paused: + default: + require.FailNow(t, "Set did not pause the job") + } + statuses, running := archiveStatuses(r) + require.Equal(t, chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED, statuses[jobID]) + require.Zero(t, running) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + requireArchiveJobsRunning(t, r) +} + +// Stop drops the run's history, so a Start for another user reads that user's +// own and resumes none of the previous user's paused jobs. The previous +// user's jobs come back when that user starts again. +func TestArchiveStartForAnotherUserReadsItsOwnHistory(t *testing.T) { + r, runner, _ := setupAppStateArchive(t, true) + ctx := context.TODO() + uidA := gregor1.UID([]byte{1, 2, 3, 4}) + uidB := gregor1.UID([]byte{5, 6, 7, 8}) + requireLaunches := func(want int) { + t.Helper() + require.Eventually(t, func() bool { + launches, _ := runner.counts() + for _, id := range archiveTestJobIDs { + if launches[id] != want { + return false + } + } + return len(launches) == len(archiveTestJobIDs) + }, 10*time.Second, time.Millisecond, "jobs not launched %d times", want) + } + + r.Start(ctx, uidA) + requireLaunches(1) + requireArchiveJobsRunning(t, r) + requireArchiveStopped(t, r) + + r.Start(ctx, uidB) + require.Eventually(t, func() bool { + r.Lock() + defer r.Unlock() + return r.inited + }, 10*time.Second, time.Millisecond, "resume did not read the history") + statuses, _ := archiveStatuses(r) + require.Empty(t, statuses, "another user's jobs carried over") + requireLaunches(1) + requireArchiveStopped(t, r) + + r.Start(ctx, uidA) + defer requireArchiveStopped(t, r) + requireLaunches(2) + requireArchiveJobsRunning(t, r) +} + +// A job launched in a previous user's run that registers or reports progress +// after the next user's Start is refused and stopped, and leaves the next +// user's history alone. +func TestArchiveSetFromPreviousUserRefused(t *testing.T) { + r, _, _ := setupAppStateArchive(t, true) + ctx := context.TODO() + uidA := gregor1.UID([]byte{1, 2, 3, 4}) + uidB := gregor1.UID([]byte{5, 6, 7, 8}) + r.resumeJobsDelay = time.Hour + r.Start(ctx, uidA) + requireArchiveStopped(t, r) + r.Start(ctx, uidB) + defer requireArchiveStopped(t, r) + + job := chat1.ArchiveChatJob{ + Request: chat1.ArchiveChatJobRequest{JobID: "job-late"}, + Status: chat1.ArchiveChatJobStatus_RUNNING, + } + paused := false + require.ErrorIs(t, r.Set(ctx, uidA, func() { paused = true }, job), errArchiveJobOtherUser) + require.True(t, paused, "refused job kept running") + require.ErrorIs(t, r.Set(ctx, uidA, nil, job), errArchiveJobOtherUser) + statuses, running := archiveStatuses(r) + require.NotContains(t, statuses, job.Request.JobID) + require.Zero(t, running) + + require.NoError(t, r.Set(ctx, uidB, nil, job)) + statuses, _ = archiveStatuses(r) + require.Contains(t, statuses, job.Request.JobID) +} + +// A pause that lands while launched jobs have not registered yet leaves them +// paused once they do, and the next FOREGROUND resumes them. +func TestArchivePauseBeforeRegistration(t *testing.T) { + r, runner, tc := setupAppStateArchive(t, false) + r.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireArchiveStopped(t, r) + for range archiveTestJobIDs { + select { + case <-runner.launched: + case <-time.After(10 * time.Second): + require.FailNow(t, "jobs did not launch") + } + } + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + close(runner.release) + requireArchiveJobsPaused(t, r, runner) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + requireArchiveJobsRunning(t, r) +} + +// A resume whose timer fired as its run stopped must not launch jobs in the +// run, possibly another user's, that started next; that run resumes on its +// own schedule. +func TestArchiveStaleResumeAfterRestart(t *testing.T) { + r, _, _ := setupAppStateArchive(t, true) + r.resumeJobsDelay = time.Hour + r.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + r.Lock() + oldStopCh := r.stopCh + r.Unlock() + requireArchiveStopped(t, r) + r.Start(context.TODO(), gregor1.UID([]byte{5, 6, 7, 8})) + defer requireArchiveStopped(t, r) + + require.NoError(t, r.resumeAllBgJobs(context.Background(), oldStopCh)) + r.Lock() + defer r.Unlock() + require.Empty(t, r.launching, "stale resume launched jobs") +} + +// A resume whose timer fired just as a plain Stop, with no Start following +// it, took the lock must not launch jobs: there is no live run left to +// launch them into. +func TestArchiveResumeAfterPlainStopLaunchesNothing(t *testing.T) { + r, _, _ := setupAppStateArchive(t, true) + r.resumeJobsDelay = time.Hour + r.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + r.Lock() + stopCh := r.stopCh + r.Unlock() + requireArchiveStopped(t, r) + + require.NoError(t, r.resumeAllBgJobs(context.Background(), stopCh)) + r.Lock() + defer r.Unlock() + require.Empty(t, r.launching, "resumed after a plain Stop") +} + +// A launch that ends after a later launch of the same job started must not +// clear the later one's entry, or the next resume starts the job again +// before the later launch registers. +func TestArchiveEndedLaunchKeepsLaterLaunch(t *testing.T) { + r, _, _ := setupAppStateArchive(t, true) + jobID := archiveTestJobIDs[0] + r.jobHistory.JobHistory = map[chat1.ArchiveJobID]chat1.ArchiveChatJob{jobID: { + Request: chat1.ArchiveChatJobRequest{JobID: jobID}, + Status: chat1.ArchiveChatJobStatus_BACKGROUND_PAUSED, + }} + var mu sync.Mutex + launches := 0 + firstExit := make(chan struct{}) + secondLaunched := make(chan struct{}) + secondRelease := make(chan struct{}) + r.runJob = func(ctx context.Context, uid gregor1.UID, req chat1.ArchiveChatJobRequest) error { + mu.Lock() + launches++ + n := launches + mu.Unlock() + switch n { + case 1: + pauseCh := make(chan struct{}) + job := chat1.ArchiveChatJob{Request: req, Status: chat1.ArchiveChatJobStatus_RUNNING} + if err := r.Set(ctx, uid, func() { close(pauseCh) }, job); err != nil { + return err + } + <-pauseCh + <-firstExit + case 2: + close(secondLaunched) + <-secondRelease + } + return nil + } + launchCount := func() int { + mu.Lock() + defer mu.Unlock() + return launches + } + stopCh := make(chan struct{}) + r.Lock() + r.started = true + r.stopCh = stopCh + r.Unlock() + defer close(stopCh) + ctx := context.Background() + + require.NoError(t, r.resumeAllBgJobs(ctx, stopCh)) + require.Eventually(t, func() bool { + _, running := archiveStatuses(r) + return running == 1 + }, 10*time.Second, time.Millisecond, "first launch did not register") + r.Lock() + require.NoError(t, r.bgPauseAllJobsLocked(ctx)) + r.Unlock() + require.NoError(t, r.resumeAllBgJobs(ctx, stopCh)) + select { + case <-secondLaunched: + case <-time.After(10 * time.Second): + require.FailNow(t, "second launch did not start") + } + + close(firstExit) + require.Never(t, func() bool { + if err := r.resumeAllBgJobs(ctx, stopCh); err != nil { + return true + } + return launchCount() > 2 + }, 300*time.Millisecond, 10*time.Millisecond, "job launched again before its launch registered") + close(secondRelease) +} + +func TestArchiveStartInBackgroundDoesNotResume(t *testing.T) { + r, runner, tc := setupAppStateArchive(t, true) + for _, state := range []keybase1.MobileAppState{ + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } { + tc.G.MobileAppState.Update(state) + r.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + select { + case id := <-runner.launched: + require.FailNow(t, fmt.Sprintf("resumed %v at a Start in %v", id, state)) + case <-time.After(200 * time.Millisecond): + } + requireArchiveStopped(t, r) + } + + r.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireArchiveStopped(t, r) + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + requireArchiveJobsRunning(t, r) + launches, _ := runner.counts() + for _, id := range archiveTestJobIDs { + require.Equal(t, 1, launches[id]) + } +} + +// Rapid transitions race resumes against pauses, and Starts and Stops against +// the loop. +func TestArchiveAppStateStress(t *testing.T) { + r, runner, tc := setupAppStateArchive(t, true) + // Pauses flush, and the first flush opens the local db and its goroutines. + r.Lock() + r.dirty = true + require.NoError(t, r.flushLocked(context.Background())) + r.Unlock() + baseline := runtime.NumGoroutine() + uid := gregor1.UID([]byte{1, 2, 3, 4}) + states := []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } + r.Start(context.TODO(), uid) + done := make(chan struct{}) + go func() { + defer close(done) + var wg sync.WaitGroup + for w := range 4 { + wg.Go(func() { + rng := rand.New(rand.NewSource(int64(w))) + for range 300 { + tc.G.MobileAppState.Update(states[rng.Intn(len(states))]) + } + }) + } + for w := range 2 { + wg.Go(func() { + rng := rand.New(rand.NewSource(int64(99 + w))) + for range 40 { + switch rng.Intn(3) { + case 0: + r.Start(context.TODO(), uid) + case 1: + <-r.Stop(context.TODO()) + default: + // Start again without waiting for the old run. + r.Stop(context.TODO()) + } + } + }) + } + wg.Wait() + }() + select { + case <-done: + case <-time.After(60 * time.Second): + require.FailNow(t, "deadlock") + } + + r.Start(context.TODO(), uid) + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + requireArchiveJobsPaused(t, r, runner) + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + requireArchiveJobsRunning(t, r) + for id, n := range func() map[chat1.ArchiveJobID]int { l, _ := runner.counts(); return l }() { + require.Positive(t, n, "%v", id) + } + _, active := runner.counts() + require.Equal(t, len(archiveTestJobIDs), active, "one live run per job") + requireArchiveStopped(t, r) + requireArchiveJobsPaused(t, r, runner) + requireNoGoroutineLeak(t, baseline) +} diff --git a/go/chat/convloader.go b/go/chat/convloader.go index 1523ac249411..665a40c7880e 100644 --- a/go/chat/convloader.go +++ b/go/chat/convloader.go @@ -115,15 +115,23 @@ type BackgroundConvLoader struct { utils.DebugLabeler sync.Mutex - uid gregor1.UID - started bool - queue *jobQueue - stopCh chan struct{} - suspendCh chan chan struct{} + uid gregor1.UID + started bool + // gen counts Start and Stop calls, so a Start that waited for the + // previous run can tell whether a later call overtook it. + gen uint64 + queue *jobQueue + stopCh chan struct{} + // suspendCh wakes the loop when Suspend takes hold; the loop reads the + // suspension itself. + suspendCh chan struct{} resumeCh chan struct{} loadCh chan *clTask identNotifier types.IdentifyNotifier - eg errgroup.Group + // eg holds the last run's goroutines, which may still be exiting. Each + // run gets its own, since a Group cannot be added to while somebody + // waits on it. + eg *errgroup.Group clock clockwork.Clock resumeWait time.Duration @@ -135,7 +143,6 @@ type BackgroundConvLoader struct { // for testing, make this and can check conv load successes loads chan chat1.ConversationID testingNameInfoSource types.NameInfoSource - appStateCh chan struct{} } var _ types.ConvLoader = (*BackgroundConvLoader)(nil) @@ -145,7 +152,8 @@ func NewBackgroundConvLoader(g *globals.Context) *BackgroundConvLoader { Contextified: globals.NewContextified(g), DebugLabeler: utils.NewDebugLabeler(g.ExternalG(), "BackgroundConvLoader", false), stopCh: make(chan struct{}), - suspendCh: make(chan chan struct{}, 10), + suspendCh: make(chan struct{}, 1), + eg: new(errgroup.Group), identNotifier: NewCachingIdentifyNotifier(g), clock: clockwork.NewRealClock(), resumeWait: time.Second, @@ -154,9 +162,6 @@ func NewBackgroundConvLoader(g *globals.Context) *BackgroundConvLoader { } b.identNotifier.ResetOnGUIConnect() b.newQueue() - stopCh := b.stopCh - go func() { _ = b.monitorAppState(stopCh) }() - return b } @@ -170,80 +175,74 @@ func (b *BackgroundConvLoader) removeActiveLoadLocked(key string) { delete(b.activeLoads, key) } -func (b *BackgroundConvLoader) monitorAppState(stopCh chan struct{}) error { - ctx := context.Background() - b.Debug(ctx, "monitorAppState: starting up") - - suspended := false - state := keybase1.MobileAppState_FOREGROUND - for { - select { - 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) - // Only resume if we had suspended earlier (frontend can spam us with these) - if suspended { - b.Debug(ctx, "monitorAppState: resuming load thread") - b.Resume(ctx) - suspended = false - } - case keybase1.MobileAppState_BACKGROUND: - b.Debug(ctx, "monitorAppState: backgrounded, suspending load thread") - if !suspended { - b.Suspend(ctx) - suspended = true - } - } - if b.appStateCh != nil { - b.appStateCh <- struct{}{} - } - case <-stopCh: - b.Debug(ctx, "monitorAppState: shutting down") - return nil - } - } +// suspendInAppState is whether background loads pause in state. INACTIVE +// (Control Center, system alerts) keeps loading, as does BACKGROUNDACTIVE. +func suspendInAppState(state keybase1.MobileAppState) bool { + return state == keybase1.MobileAppState_BACKGROUND } +// Start replaces any current run with one for uid, once the previous run's +// goroutines have exited. The last Start or Stop wins: a Start that a later +// Start or Stop overtook while it waited returns without starting a run. func (b *BackgroundConvLoader) Start(ctx context.Context, uid gregor1.UID) { - b.Lock() - defer b.Unlock() - if b.G().GetEnv().GetDisableBgConvLoader() { b.Debug(ctx, "BackgroundConvLoader disabled, aborting Start") return } b.Debug(ctx, "Start") - if b.started { - close(b.stopCh) - b.stopCh = make(chan struct{}) + b.Lock() + b.gen++ + gen := b.gen + prevRun := b.endRunLocked() + b.Unlock() + + // The previous run's goroutines take b's lock, so wait for them outside it. + _ = prevRun.Wait() + + b.Lock() + defer b.Unlock() + if b.gen != gen { + b.Debug(ctx, "Start: overtaken by a later Start or Stop") + return + } + // A wake-up the previous run never read would park this run's loop; a + // suspension still in force parks it anyway, through suspendCount. + select { + case <-b.suspendCh: + default: } b.newQueue() b.started = true b.uid = uid - stopCh := b.stopCh - b.eg.Go(func() error { return b.loop(uid, stopCh) }) - b.eg.Go(func() error { return b.loadLoop(uid, stopCh) }) + b.eg = new(errgroup.Group) + stopCh, eg, queue, loadCh := b.stopCh, b.eg, b.queue, b.loadCh + eg.Go(func() error { return b.loop(uid, stopCh, queue, loadCh) }) + eg.Go(func() error { return b.loadLoop(uid, stopCh, queue, loadCh) }) +} + +// endRunLocked ends the current run, if there is one, and returns the group of +// the last run's goroutines. +func (b *BackgroundConvLoader) endRunLocked() *errgroup.Group { + if b.started { + b.started = false + b.cancelActiveLoadsLocked() + close(b.stopCh) + b.stopCh = make(chan struct{}) + } + return b.eg } func (b *BackgroundConvLoader) Stop(ctx context.Context) chan struct{} { b.Lock() defer b.Unlock() b.Debug(ctx, "Stop") - b.cancelActiveLoadsLocked() + b.gen++ + eg := b.endRunLocked() ch := make(chan struct{}) - if b.started { - b.started = false - close(b.stopCh) - b.stopCh = make(chan struct{}) - go func() { - _ = b.eg.Wait() - close(ch) - }() - } else { + go func() { + _ = eg.Wait() close(ch) - } + }() return ch } @@ -284,12 +283,11 @@ func (b *BackgroundConvLoader) Suspend(ctx context.Context) (canceled bool) { return false } if b.suspendCount == 0 { - b.Debug(ctx, "Suspend: sending on suspendCh") + b.Debug(ctx, "Suspend: waking loop") b.resumeCh = make(chan struct{}) select { - case b.suspendCh <- b.resumeCh: + case b.suspendCh <- struct{}{}: default: - b.Debug(ctx, "Suspend: failed to suspend loop") } } b.suspendCount++ @@ -300,21 +298,20 @@ func (b *BackgroundConvLoader) Resume(ctx context.Context) bool { defer b.Trace(ctx, nil, "Resume")() b.Lock() defer b.Unlock() + if b.suspendCount == 0 { + return false + } + b.suspendCount-- if b.suspendCount > 0 { - b.suspendCount-- - if b.suspendCount == 0 && b.resumeCh != nil { - b.Debug(ctx, "Resume: closing resumeCh") - close(b.resumeCh) - return true - } + return false } - return false + b.Debug(ctx, "Resume: closing resumeCh") + close(b.resumeCh) + return true } -func (b *BackgroundConvLoader) isSuspended() bool { - b.Lock() - defer b.Unlock() - return b.suspendCount > 0 +func (b *BackgroundConvLoader) suspendedLocked() bool { + return b.suspendCount > 0 || suspendInAppState(b.G().MobileAppState.State()) } func (b *BackgroundConvLoader) isRunning() bool { @@ -326,8 +323,20 @@ func (b *BackgroundConvLoader) isRunning() bool { func (b *BackgroundConvLoader) enqueue(ctx context.Context, task clTask) error { b.Lock() defer b.Unlock() + return b.push(ctx, b.queue, task) +} + +// requeue puts a task back on the queue of the run that loaded it. Once that +// run has stopped, nobody reads its queue. +func (b *BackgroundConvLoader) requeue(ctx context.Context, queue *jobQueue, task clTask) { + if err := b.push(ctx, queue, task); err != nil { + b.Debug(ctx, "enqueue error %s", err) + } +} + +func (b *BackgroundConvLoader) push(ctx context.Context, queue *jobQueue, task clTask) error { b.Debug(ctx, "enqueue: adding task: %s", task.job) - queued, err := b.queue.Push(task) + queued, err := queue.Push(task) if err != nil { return err } @@ -337,35 +346,79 @@ func (b *BackgroundConvLoader) enqueue(ctx context.Context, task clTask) error { return nil } -func (b *BackgroundConvLoader) loop(uid gregor1.UID, stopCh chan struct{}) error { +func (b *BackgroundConvLoader) loop(uid gregor1.UID, stopCh chan struct{}, queue *jobQueue, + loadCh chan *clTask, +) error { bgctx := context.Background() b.Debug(bgctx, "loop: starting conv loader loop for %s", uid) - - // waitForResume is called on suspend. It will wait for a resume event, and then pause - // for b.resumeWait amount of time. Returns false if the outer loop should shutdown. - waitForResume := func(ch chan struct{}) bool { - b.Debug(bgctx, "waitForResume: suspending loop") - select { - case <-ch: - case <-stopCh: + appState := b.G().MobileAppState + state := appState.State() + + // appStateChanged reads the new app state and reports whether it suspends + // the loop. Nothing else watches the app state, so going to BACKGROUND + // cancels active loads here, at once. + appStateChanged := func() (suspended bool) { + state = appState.State() + if !suspendInAppState(state) { return false } - b.clock.Sleep(libkb.RandomJitter(b.resumeWait)) - b.Debug(bgctx, "waitForResume: resuming loop") + b.Debug(bgctx, "loop: suspending in %v", state) + b.Lock() + b.cancelActiveLoadsLocked() + b.Unlock() return true } - // On mobile fresh start, apply the foreground wait - if b.G().IsMobileAppType() { - b.Debug(bgctx, "loop: delaying startup since on mobile") - b.clock.Sleep(libkb.RandomJitter(b.resumeWait)) + // suspension reports whether the loop is held, with the channel Resume + // closes when a Suspend holds it. + suspension := func() (held bool, resumeCh chan struct{}) { + b.Lock() + defer b.Unlock() + if b.suspendCount > 0 { + return true, b.resumeCh + } + return suspendInAppState(state), nil + } + // waitForResume parks the loop until neither Suspend nor the app state + // holds it, then waits for b.resumeWait with jitter. Returns false if the + // run stopped. + waitForResume := func() bool { + b.Debug(bgctx, "waitForResume: suspending loop") + var resumeDelay <-chan time.Time + for { + held, resumeCh := suspension() + switch { + case held: + resumeDelay = nil + case resumeDelay == nil: + resumeDelay = b.clock.After(libkb.RandomJitter(b.resumeWait)) + } + select { + case <-resumeCh: + case <-b.suspendCh: + case <-resumeDelay: + b.Debug(bgctx, "waitForResume: resuming loop") + return true + case <-appState.NextUpdate(state): + appStateChanged() + case <-stopCh: + return false + } + } + } + // Park if already suspended, and on a mobile fresh start apply the + // foreground wait. + if held, _ := suspension(); held || b.G().IsMobileAppType() { + if !waitForResume() { + return nil + } } // Main loop for { b.Debug(bgctx, "loop: waiting for job") select { - case <-b.queue.Wait(): - task, ok := b.queue.PopFront() + case <-queue.Wait(): + task, ok := queue.PopFront() if !ok { continue } @@ -380,24 +433,46 @@ func (b *BackgroundConvLoader) loop(uid gregor1.UID, stopCh chan struct{}) error duration = max(bgLoaderErrDelay-time.Since(task.lastAttemptAt), bgLoaderInitDelay) } // Make sure we aren't suspended (also make sure we don't get shutdown). Charge through if - // neither have any data on them. - select { - case <-b.clock.After(duration): - case ch := <-b.suspendCh: - b.Debug(bgctx, "loop: pulled queue task, but suspended, so waiting") - if !waitForResume(ch) { + // neither have any data on them. An app-state change that doesn't suspend keeps waiting + // out the delay, so a retry still gets its full backoff. + delay := b.clock.After(duration) + waitDelay: + for { + select { + case <-delay: + break waitDelay + case <-b.suspendCh: + b.Debug(bgctx, "loop: pulled queue task, but suspended, so waiting") + if !waitForResume() { + return nil + } + break waitDelay + case <-appState.NextUpdate(state): + if !appStateChanged() { + continue + } + if !waitForResume() { + return nil + } + break waitDelay + case <-stopCh: + b.Debug(bgctx, "loop: shutting down for %s", uid) return nil } } b.Debug(bgctx, "loop: pulled queued task: %s", task.job) select { - case b.loadCh <- &task: + case loadCh <- &task: default: b.Debug(bgctx, "loop: failed to dispatch load, queue full") } - case ch := <-b.suspendCh: + case <-b.suspendCh: b.Debug(bgctx, "loop: received suspend") - if !waitForResume(ch) { + if !waitForResume() { + return nil + } + case <-appState.NextUpdate(state): + if appStateChanged() && !waitForResume() { return nil } case <-stopCh: @@ -407,31 +482,23 @@ func (b *BackgroundConvLoader) loop(uid gregor1.UID, stopCh chan struct{}) error } } -func (b *BackgroundConvLoader) loadLoop(uid gregor1.UID, stopCh chan struct{}) error { +func (b *BackgroundConvLoader) loadLoop(uid gregor1.UID, stopCh chan struct{}, queue *jobQueue, + loadCh chan *clTask, +) error { bgctx := context.Background() b.Debug(bgctx, "loadLoop: starting for uid: %s", uid) for { select { - case task := <-b.loadCh: - switch { - case !b.isRunning(): + case task := <-loadCh: + if nextTask := b.load(bgctx, stopCh, *task, uid); nextTask != nil { + b.requeue(bgctx, queue, *nextTask) + } + select { + case <-b.clock.After(b.loadWait): + case <-stopCh: b.Debug(bgctx, "loadLoop: shutting down for %s", uid) return nil - case b.isSuspended(): - b.Debug(bgctx, "loadLoop: suspended, re-enqueueing task: %s", task.job) - if err := b.enqueue(bgctx, *task); err != nil { - b.Debug(bgctx, "enqueue error %s", err) - } - default: - b.Debug(bgctx, "loadLoop: running task: %s", task.job) - nextTask := b.load(bgctx, *task, uid) - if nextTask != nil { - if err := b.enqueue(bgctx, *nextTask); err != nil { - b.Debug(bgctx, "enqueue error %s", err) - } - } } - b.clock.Sleep(b.loadWait) case <-stopCh: b.Debug(bgctx, "loadLoop: shutting down for %s", uid) return nil @@ -465,10 +532,28 @@ func (b *BackgroundConvLoader) IsBackgroundActive() bool { return len(b.activeLoads) > 0 } -func (b *BackgroundConvLoader) load(ictx context.Context, task clTask, uid gregor1.UID) *clTask { +// load runs task unless its run has stopped, and returns a task to requeue: +// task itself while suspended, or its retry. +func (b *BackgroundConvLoader) load(ictx context.Context, stopCh chan struct{}, task clTask, + uid gregor1.UID, +) *clTask { + b.Lock() + // Checked under the lock that cancels active loads, so a load either sees + // the stop or the suspension here, or is registered in time to be canceled. + select { + case <-stopCh: + b.Unlock() + b.Debug(ictx, "load: run stopped, dropping task: %s", task.job) + return nil + default: + } + if b.suspendedLocked() { + b.Unlock() + b.Debug(ictx, "load: suspended, re-enqueueing task: %s", task.job) + return &task + } defer b.Trace(ictx, nil, "load: %s", task.job)() defer b.PerfTrace(ictx, nil, "load: %s", task.job)() - b.Lock() var al activeLoad al.Ctx, al.CancelFn = context.WithCancel( globals.ChatCtx(utils.MakeConvLoaderContext(ictx), b.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, diff --git a/go/chat/convloader_appstate_test.go b/go/chat/convloader_appstate_test.go new file mode 100644 index 000000000000..79fe358e6c5c --- /dev/null +++ b/go/chat/convloader_appstate_test.go @@ -0,0 +1,628 @@ +package chat + +import ( + "context" + "math/rand" + "runtime" + "sync" + "testing" + "time" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/chat/types" + "github.com/keybase/client/go/externalstest" + "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/clockwork" + "github.com/stretchr/testify/require" +) + +type appStateCtxFactory struct{} + +func (appStateCtxFactory) NewKeyFinder() types.KeyFinder { return nil } +func (appStateCtxFactory) NewUPAKFinder() types.UPAKFinder { return nil } + +// pullRecorder is the only part of ConversationSource a background load +// reaches; anything else panics on the nil embedded interface. +type pullRecorder struct { + types.ConversationSource + pulls chan chat1.ConversationID +} + +func (p *pullRecorder) Pull(ctx context.Context, convID chat1.ConversationID, uid gregor1.UID, + reason chat1.GetThreadReason, customRi func() chat1.RemoteInterface, query *chat1.GetThreadQuery, + pagination *chat1.Pagination, +) (chat1.ThreadView, error) { + select { + case p.pulls <- convID: + default: + } + return chat1.ThreadView{}, nil +} + +func setupAppStateConvLoader(t *testing.T) (*BackgroundConvLoader, *pullRecorder, libkb.TestContext) { + tc := externalstest.SetupTest(t, "convloader-appstate", 0) + t.Cleanup(tc.Cleanup) + tc.G.ConnectionManager = libkb.NewConnectionManager() + pulls := &pullRecorder{pulls: make(chan chat1.ConversationID, 100)} + g := globals.NewContext(tc.G, &globals.ChatContext{ + CtxFactory: appStateCtxFactory{}, + ConvSource: pulls, + }) + b := NewBackgroundConvLoader(g) + b.resumeWait = time.Millisecond + b.loadWait = time.Millisecond + return b, pulls, tc +} + +func requireConvLoaderStopped(t *testing.T, b *BackgroundConvLoader) { + t.Helper() + select { + case <-b.Stop(context.TODO()): + case <-time.After(10 * time.Second): + require.FailNow(t, "Stop did not finish") + } +} + +var convLoaderTestConvID = chat1.ConversationID([]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}) + +func convLoaderTestJob() types.ConvLoaderJob { + return types.NewConvLoaderJob(convLoaderTestConvID, &chat1.Pagination{Num: 1}, + types.ConvLoaderPriorityHigh, types.ConvLoaderGeneric, nil) +} + +// A load checks for a stop and a suspension under the lock that cancels +// active loads, so it never starts after either. +func TestConvLoaderLoadChecksStopAndSuspension(t *testing.T) { + b, pulls, tc := setupAppStateConvLoader(t) + uid := gregor1.UID([]byte{1, 2, 3, 4}) + task := clTask{job: convLoaderTestJob()} + + stopped := make(chan struct{}) + close(stopped) + require.Nil(t, b.load(context.TODO(), stopped, task, uid)) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + next := b.load(context.TODO(), make(chan struct{}), task, uid) + require.NotNil(t, next) + require.Equal(t, task.job.ConvID, next.job.ConvID) + require.Zero(t, next.attempt) + + select { + case <-pulls.pulls: + require.FailNow(t, "loaded after a stop or in BACKGROUND") + default: + } +} + +// Stop does not wait for the loop's delay before dispatching a job. +func TestConvLoaderStopDuringLoadDelay(t *testing.T) { + b, _, _ := setupAppStateConvLoader(t) + clock := clockwork.NewFakeClock() + b.clock = clock + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + clock.BlockUntil(1) + requireConvLoaderStopped(t, b) +} + +// The loop also watches the app state while it waits out the delay before +// dispatching the next job, with the previous job still loading. +func TestConvLoaderBackgroundCancelsDuringLoadDelay(t *testing.T) { + b, _, tc := setupAppStateConvLoader(t) + pulls := newCtxPuller(false) + b.G().ConvSource = pulls + clock := clockwork.NewFakeClock() + b.clock = clock + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireConvLoaderStopped(t, b) + defer close(pulls.release) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + clock.BlockUntil(1) + clock.Advance(bgLoaderInitDelay) + load := requirePull(t, pulls) + + otherConvID := chat1.ConversationID([]byte{16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1}) + require.NoError(t, b.Queue(context.TODO(), types.NewConvLoaderJob(otherConvID, &chat1.Pagination{Num: 1}, + types.ConvLoaderPriorityHigh, types.ConvLoaderGeneric, nil))) + // the loop has pulled the second job and waits out its delay + clock.BlockUntil(1) + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + select { + case <-load.ctx.Done(): + case <-time.After(100 * time.Millisecond): + require.FailNow(t, "active load not canceled on BACKGROUND") + } +} + +// An app-state change that doesn't suspend the loop, like FOREGROUND -> +// INACTIVE, keeps the loop waiting out the load delay instead of dispatching +// the job early. +func TestConvLoaderNonSuspendingStateKeepsLoadDelay(t *testing.T) { + b, _, tc := setupAppStateConvLoader(t) + pulls := newCtxPuller(false) + b.G().ConvSource = pulls + clock := clockwork.NewFakeClock() + b.clock = clock + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireConvLoaderStopped(t, b) + defer close(pulls.release) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + clock.BlockUntil(1) + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + select { + case <-pulls.calls: + require.FailNow(t, "dispatched before the load delay ran out") + case <-time.After(100 * time.Millisecond): + } + clock.Advance(bgLoaderInitDelay) + requirePull(t, pulls) +} + +// Each run's loop watches the app state: BACKGROUND cancels its active load +// and parks it, and leaving BACKGROUND loads the retry. +func TestConvLoaderAppStateAcrossRuns(t *testing.T) { + b, _, tc := setupAppStateConvLoader(t) + pulls := newCtxPuller(false) + b.G().ConvSource = pulls + defer close(pulls.release) + uid := gregor1.UID([]byte{1, 2, 3, 4}) + appState := tc.G.MobileAppState + requireCanceled := func(i int, load pullCall) { + t.Helper() + select { + case <-load.ctx.Done(): + case <-time.After(10 * time.Second): + require.FailNow(t, "load not canceled in BACKGROUND", "run %d", i) + } + } + for i := range 3 { + appState.Update(keybase1.MobileAppState_FOREGROUND) + b.Start(context.TODO(), uid) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + load := requirePull(t, pulls) + + appState.Update(keybase1.MobileAppState_BACKGROUND) + requireCanceled(i, load) + + appState.Update(keybase1.MobileAppState_INACTIVE) + load = requirePull(t, pulls) + + appState.Update(keybase1.MobileAppState_BACKGROUND) + requireCanceled(i, load) + requireConvLoaderStopped(t, b) + + // A run started in BACKGROUND loads nothing until the app leaves it. + b.Start(context.TODO(), uid) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case <-pulls.calls: + require.FailNow(t, "loaded in BACKGROUND", "run %d", i) + case <-time.After(300 * time.Millisecond): + } + appState.Update(keybase1.MobileAppState_BACKGROUNDACTIVE) + load = requirePull(t, pulls) + appState.Update(keybase1.MobileAppState_BACKGROUND) + requireCanceled(i, load) + requireConvLoaderStopped(t, b) + } +} + +func TestConvLoaderBackgroundLaunchStaysSuspended(t *testing.T) { + b, pulls, tc := setupAppStateConvLoader(t) + uid := gregor1.UID([]byte{1, 2, 3, 4}) + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + require.False(t, b.Suspend(context.TODO()), "Suspend before Start") + require.False(t, b.Resume(context.TODO())) + b.Start(context.TODO(), uid) + defer requireConvLoaderStopped(t, b) + + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case <-pulls.pulls: + require.FailNow(t, "loaded in BACKGROUND") + case <-time.After(300 * time.Millisecond): + } + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + select { + case convID := <-pulls.pulls: + require.Equal(t, convLoaderTestConvID, convID) + case <-time.After(10 * time.Second): + require.FailNow(t, "no load after FOREGROUND") + } +} + +// An unbalanced Resume, or a Suspend and Resume pair, must not release the +// app-state suspension. +func TestConvLoaderResumeKeepsAppStateSuspension(t *testing.T) { + b, pulls, tc := setupAppStateConvLoader(t) + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireConvLoaderStopped(t, b) + require.False(t, b.Resume(context.TODO())) + b.Suspend(context.TODO()) + require.True(t, b.Resume(context.TODO())) + + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case <-pulls.pulls: + require.FailNow(t, "loaded in BACKGROUND") + case <-time.After(300 * time.Millisecond): + } + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + select { + case convID := <-pulls.pulls: + require.Equal(t, convLoaderTestConvID, convID) + case <-time.After(10 * time.Second): + require.FailNow(t, "no load after FOREGROUND") + } +} + +// A Suspend's wake-up that the previous run never read doesn't park the next +// run's loop. +func TestConvLoaderStartDropsStaleSuspendWake(t *testing.T) { + b, pulls, _ := setupAppStateConvLoader(t) + b.resumeWait = time.Hour + b.suspendCh <- struct{}{} + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireConvLoaderStopped(t, b) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case convID := <-pulls.pulls: + require.Equal(t, convLoaderTestConvID, convID) + case <-time.After(10 * time.Second): + require.FailNow(t, "no load: the stale wake-up parked the loop") + } +} + +// A Stop that comes while a Start waits for the previous run wins: it waits +// for that run too, and the Start does not start a new one. +func TestConvLoaderStopOvertakesWaitingStart(t *testing.T) { + b, _, _ := setupAppStateConvLoader(t) + pulls := newCtxPuller(true) + b.G().ConvSource = pulls + uid := gregor1.UID([]byte{1, 2, 3, 4}) + b.Start(context.TODO(), uid) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + requirePull(t, pulls) + + started := make(chan struct{}) + go func() { + b.Start(context.TODO(), uid) + close(started) + }() + // the waiting Start has ended the previous run + require.Eventually(t, func() bool { return !b.isRunning() }, 10*time.Second, time.Millisecond) + stopped := b.Stop(context.TODO()) + select { + case <-stopped: + require.FailNow(t, "Stop finished while the previous run was still running") + case <-time.After(200 * time.Millisecond): + } + close(pulls.release) + for _, ch := range []chan struct{}{started, stopped} { + select { + case <-ch: + case <-time.After(10 * time.Second): + require.FailNow(t, "Start or Stop did not return") + } + } + require.False(t, b.isRunning(), "an overtaken Start started a run") +} + +// Of two Starts waiting for the previous run, the later one's run is the one +// that starts. +func TestConvLoaderLastWaitingStartWins(t *testing.T) { + b, _, _ := setupAppStateConvLoader(t) + pulls := newCtxPuller(true) + b.G().ConvSource = pulls + baseline := runtime.NumGoroutine() + oldUID := gregor1.UID([]byte{1, 2, 3, 4}) + uids := []gregor1.UID{{5, 6, 7, 8}, {9, 10, 11, 12}} + b.Start(context.TODO(), oldUID) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + requirePull(t, pulls) + + var wg sync.WaitGroup + for i, uid := range uids { + wg.Go(func() { b.Start(context.TODO(), uid) }) + require.Eventually(t, func() bool { + b.Lock() + defer b.Unlock() + return b.gen == uint64(i+2) + }, 10*time.Second, time.Millisecond, "Start %d did not begin waiting", i) + } + close(pulls.release) + wg.Wait() + require.True(t, b.isRunning()) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + require.Equal(t, uids[1], requirePull(t, pulls).uid) + // the overtaken Start left no run of its own behind + requireConvLoaderStopped(t, b) + requireNoGoroutineLeak(t, baseline) +} + +// A suspension that outlives a run parks the next run's loop before it +// takes anything off the queue. +func TestConvLoaderSuspensionCarriesIntoNextRun(t *testing.T) { + uid := gregor1.UID([]byte{1, 2, 3, 4}) + for _, tt := range []struct { + name string + suspend func(*BackgroundConvLoader, libkb.TestContext) + }{ + {"background", func(_ *BackgroundConvLoader, tc libkb.TestContext) { + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + }}, + {"suspend", func(b *BackgroundConvLoader, _ libkb.TestContext) { + require.False(t, b.Suspend(context.TODO())) + }}, + } { + t.Run(tt.name, func(t *testing.T) { + b, _, tc := setupAppStateConvLoader(t) + clock := clockwork.NewFakeClock() + b.clock = clock + b.Start(context.TODO(), uid) + tt.suspend(b, tc) + requireConvLoaderStopped(t, b) + b.Start(context.TODO(), uid) + defer requireConvLoaderStopped(t, b) + + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + // A loop that pulls the job waits out its delay on the clock. + blocked := make(chan struct{}) + go func() { + clock.BlockUntil(1) + close(blocked) + }() + defer clock.After(time.Hour) + select { + case <-blocked: + require.FailNow(t, "loop pulled a job while suspended") + case <-time.After(300 * time.Millisecond): + } + b.Lock() + queued := b.queue.queue.Len() + b.Unlock() + require.Equal(t, 1, queued, "queue drained while suspended") + }) + } +} + +// pullBlocker fails the first load of the old user's conversation once +// released, so the old run asks to retry it. +type pullBlocker struct { + types.ConversationSource + oldUID gregor1.UID + started chan struct{} + release chan struct{} + + mu sync.Mutex + uids []gregor1.UID +} + +func (p *pullBlocker) Pull(ctx context.Context, convID chat1.ConversationID, uid gregor1.UID, + reason chat1.GetThreadReason, customRi func() chat1.RemoteInterface, query *chat1.GetThreadQuery, + pagination *chat1.Pagination, +) (chat1.ThreadView, error) { + p.mu.Lock() + p.uids = append(p.uids, uid) + p.mu.Unlock() + if uid.Eq(p.oldUID) { + close(p.started) + <-p.release + return chat1.ThreadView{}, context.Canceled + } + return chat1.ThreadView{}, nil +} + +func TestConvLoaderReplacedRunRetryStaysInItsRun(t *testing.T) { + b, _, _ := setupAppStateConvLoader(t) + oldUID, newUID := gregor1.UID([]byte{1, 2, 3, 4}), gregor1.UID([]byte{5, 6, 7, 8}) + pulls := &pullBlocker{oldUID: oldUID, started: make(chan struct{}), release: make(chan struct{})} + b.G().ConvSource = pulls + b.Start(context.TODO(), oldUID) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case <-pulls.started: + case <-time.After(10 * time.Second): + require.FailNow(t, "old run did not load") + } + started := make(chan struct{}) + go func() { + b.Start(context.TODO(), newUID) + close(started) + }() + defer requireConvLoaderStopped(t, b) + close(pulls.release) + select { + case <-started: + case <-time.After(10 * time.Second): + require.FailNow(t, "Start did not return") + } + + // past the retry delay and the new run's load delays + time.Sleep(time.Second) + b.Lock() + queued := b.queue.queue.Len() + b.Unlock() + require.Zero(t, queued, "old run's retry reached the new queue") + pulls.mu.Lock() + defer pulls.mu.Unlock() + require.Equal(t, []gregor1.UID{oldUID}, pulls.uids) +} + +func TestConvLoaderAppStateStress(t *testing.T) { + b, pulls, tc := setupAppStateConvLoader(t) + baseline := runtime.NumGoroutine() + uid := gregor1.UID([]byte{1, 2, 3, 4}) + states := []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } + done := make(chan struct{}) + go func() { + defer close(done) + var wg sync.WaitGroup + for w := range 4 { + wg.Go(func() { + rng := rand.New(rand.NewSource(int64(w))) + for range 300 { + tc.G.MobileAppState.Update(states[rng.Intn(len(states))]) + } + }) + } + for w := range 4 { + wg.Go(func() { + rng := rand.New(rand.NewSource(int64(100 + w))) + for range 150 { + switch rng.Intn(4) { + case 0: + b.Start(context.TODO(), uid) + case 1: + <-b.Stop(context.TODO()) + case 2: + _ = b.Queue(context.TODO(), convLoaderTestJob()) + default: + b.Suspend(context.TODO()) + b.Resume(context.TODO()) + } + } + }) + } + wg.Wait() + }() + select { + case <-done: + case <-time.After(60 * time.Second): + require.FailNow(t, "deadlock") + } + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + b.Start(context.TODO(), uid) + // the loader still loads once the churn is over + for len(pulls.pulls) > 0 { + <-pulls.pulls + } + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + select { + case <-pulls.pulls: + case <-time.After(10 * time.Second): + require.FailNow(t, "no load after the churn") + } + requireConvLoaderStopped(t, b) + requireNoGoroutineLeak(t, baseline) +} + +// requireNoGoroutineLeak polls without require.Eventually, whose own +// goroutines would count against the baseline. +func requireNoGoroutineLeak(t *testing.T, baseline int) { + t.Helper() + 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") +} + +type pullCall struct { + ctx context.Context + uid gregor1.UID +} + +// ctxPuller hands each load to the test and holds it until release closes, +// or until its ctx is canceled unless ignoreCancel is set. +type ctxPuller struct { + types.ConversationSource + calls chan pullCall + release chan struct{} + ignoreCancel bool +} + +func newCtxPuller(ignoreCancel bool) *ctxPuller { + return &ctxPuller{ + calls: make(chan pullCall, 100), + release: make(chan struct{}), + ignoreCancel: ignoreCancel, + } +} + +func (p *ctxPuller) Pull(ctx context.Context, convID chat1.ConversationID, uid gregor1.UID, + reason chat1.GetThreadReason, customRi func() chat1.RemoteInterface, query *chat1.GetThreadQuery, + pagination *chat1.Pagination, +) (chat1.ThreadView, error) { + p.calls <- pullCall{ctx: ctx, uid: uid} + done := ctx.Done() + if p.ignoreCancel { + done = nil + } + select { + case <-p.release: + case <-done: + } + return chat1.ThreadView{}, ctx.Err() +} + +func requirePull(t *testing.T, p *ctxPuller) pullCall { + t.Helper() + select { + case call := <-p.calls: + return call + case <-time.After(10 * time.Second): + require.FailNow(t, "no load") + return pullCall{} + } +} + +func TestConvLoaderStartWaitsForPreviousRun(t *testing.T) { + b, _, _ := setupAppStateConvLoader(t) + pulls := newCtxPuller(true) + b.G().ConvSource = pulls + uid := gregor1.UID([]byte{1, 2, 3, 4}) + b.Start(context.TODO(), uid) + defer requireConvLoaderStopped(t, b) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + load := requirePull(t, pulls) + + started := make(chan struct{}) + go func() { + b.Start(context.TODO(), uid) + close(started) + }() + select { + case <-started: + require.FailNow(t, "Start returned while the previous run's load was still running") + case <-time.After(200 * time.Millisecond): + } + require.Error(t, load.ctx.Err(), "Start did not cancel the previous run's load") + close(pulls.release) + select { + case <-started: + case <-time.After(10 * time.Second): + require.FailNow(t, "Start did not return after the previous run exited") + } + require.True(t, b.isRunning()) +} + +func TestConvLoaderBackgroundCancelsActiveLoadImmediately(t *testing.T) { + b, _, tc := setupAppStateConvLoader(t) + pulls := newCtxPuller(false) + b.G().ConvSource = pulls + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + b.Start(context.TODO(), gregor1.UID([]byte{1, 2, 3, 4})) + defer requireConvLoaderStopped(t, b) + defer close(pulls.release) + require.NoError(t, b.Queue(context.TODO(), convLoaderTestJob())) + load := requirePull(t, pulls) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + select { + case <-load.ctx.Done(): + case <-time.After(100 * time.Millisecond): + require.FailNow(t, "active load not canceled on BACKGROUND") + } +} diff --git a/go/chat/convloader_test.go b/go/chat/convloader_test.go index 46ea15ade033..e152b79e84b5 100644 --- a/go/chat/convloader_test.go +++ b/go/chat/convloader_test.go @@ -134,15 +134,17 @@ func TestConvLoaderAppState(t *testing.T) { defer world.Cleanup() clock := clockwork.NewFakeClock() - appStateCh := make(chan struct{}) - tc.ChatG.ConvLoader.(*BackgroundConvLoader).loadWait = 0 - tc.ChatG.ConvLoader.(*BackgroundConvLoader).clock = clock - tc.ChatG.ConvLoader.(*BackgroundConvLoader).appStateCh = appStateCh + uid := gregor1.UID(tc.G.Env.GetUID().ToBytes()) + // The loops read these, so set them while the loader is stopped. + loader := tc.ChatG.ConvLoader.(*BackgroundConvLoader) + <-loader.Stop(context.TODO()) + loader.loadWait = 0 + loader.clock = clock + loader.Start(context.TODO(), uid) ri := tc.ChatG.ConvSource.(*HybridConversationSource).ri _ = ri slowRi := makeSlowestRemote() failDuration := 2 * time.Second - uid := gregor1.UID(tc.G.Env.GetUID().ToBytes()) // Test that a foreground with no background doesnt do anything tc.ChatG.ConvSource.(*HybridConversationSource).ri = func() chat1.RemoteInterface { return slowRi @@ -160,11 +162,6 @@ func TestConvLoaderAppState(t *testing.T) { require.True(t, tc.Context().ConvLoader.Suspend(context.TODO())) tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) select { - case <-appStateCh: - require.Fail(t, "no app state") - default: - } - select { case <-listener.bgConvLoads: require.Fail(t, "no load yet") default: @@ -199,18 +196,10 @@ func TestConvLoaderAppState(t *testing.T) { require.Fail(t, "no remote call") } tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) - select { - case <-appStateCh: - case <-time.After(failDuration): - require.Fail(t, "no app state") - } + // the loop cancels the active load + require.Eventually(t, func() bool { return !loader.IsBackgroundActive() }, failDuration, time.Millisecond) tc.ChatG.ConvSource.(*HybridConversationSource).ri = ri tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) - select { - case <-appStateCh: - case <-time.After(failDuration): - require.Fail(t, "no app state") - } // Need to advance clock select { case <-listener.bgConvLoads: diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index ac373804c636..63d796a52610 100644 --- a/go/chat/search/indexer.go +++ b/go/chat/search/indexer.go @@ -81,6 +81,11 @@ type Indexer struct { consumeCh chan chat1.ConversationID reindexCh chan chat1.ConversationID syncLoopCh, cancelSyncCh, pokeSyncCh chan struct{} + // selectiveSync, if set, runs in place of SelectiveSync. Tests only. + selectiveSync func(ctx context.Context) error + // beforeSyncStateCheck and afterSyncStart, if set, run in attemptSync + // around its app-state check and sync start. Tests only. + beforeSyncStateCheck, afterSyncStart func() } var _ types.Indexer = (*Indexer)(nil) @@ -243,11 +248,14 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { ticker := libkb.NewBgTicker(idx.syncInterval) after := time.After(idx.startSyncDelay) - appState := keybase1.MobileAppState_FOREGROUND + appState := idx.G().MobileAppState.State() netState := keybase1.MobileNetworkState_WIFI var cancelFn context.CancelFunc var l sync.Mutex var syncAttemptWG sync.WaitGroup + // syncDeferred is set when an attempt is skipped outside FOREGROUND, so the + // next change into FOREGROUND runs it instead of dropping it. + syncDeferred := false cancelSync := func() { l.Lock() defer l.Unlock() @@ -260,6 +268,22 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { if netState.IsLimited() { return } + if idx.beforeSyncStateCheck != nil { + idx.beforeSyncStateCheck() + } + if state := idx.G().MobileAppState.State(); state != keybase1.MobileAppState_FOREGROUND { + idx.Debug(ctx, "not running SelectiveSync in %v", state) + syncDeferred = true + return + } + syncDeferred = false + // The loop may not have woken for the change into FOREGROUND yet. Wait + // on changes from FOREGROUND from here on, so leaving it after this + // read wakes the loop, which cancels the sync. + appState = keybase1.MobileAppState_FOREGROUND + if idx.afterSyncStart != nil { + defer idx.afterSyncStart() + } l.Lock() defer l.Unlock() if cancelFn != nil { @@ -267,9 +291,13 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { return } ctx, cancelFn = context.WithCancel(ctx) + selectiveSync := idx.SelectiveSync + if idx.selectiveSync != nil { + selectiveSync = idx.selectiveSync + } syncAttemptWG.Go(func() { idx.Debug(ctx, "running SelectiveSync") - if err := idx.SelectiveSync(ctx); err != nil { + if err := selectiveSync(ctx); err != nil { idx.Debug(ctx, "unable to complete SelectiveSync: %v", err) if idx.syncLoopCh != nil { select { @@ -312,6 +340,9 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { appState = idx.G().MobileAppState.State() switch appState { case keybase1.MobileAppState_FOREGROUND: + if syncDeferred { + attemptSync(ctx) + } // if we enter any state besides foreground cancel any running syncs default: cancelSync() diff --git a/go/chat/search/indexer_appstate_test.go b/go/chat/search/indexer_appstate_test.go new file mode 100644 index 000000000000..59be838ea582 --- /dev/null +++ b/go/chat/search/indexer_appstate_test.go @@ -0,0 +1,230 @@ +package search + +import ( + "context" + "math/rand" + "runtime" + "sync" + "testing" + "time" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/externalstest" + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +// syncRecorder stands in for SelectiveSync: each sync runs until canceled. +type syncRecorder struct { + mu sync.Mutex + starts int + active int +} + +func (s *syncRecorder) sync(ctx context.Context) error { + s.mu.Lock() + s.starts++ + s.active++ + s.mu.Unlock() + <-ctx.Done() + s.mu.Lock() + s.active-- + s.mu.Unlock() + return ctx.Err() +} + +func (s *syncRecorder) counts() (starts, active int) { + s.mu.Lock() + defer s.mu.Unlock() + return s.starts, s.active +} + +type syncLoopTest struct { + t *testing.T + tc libkb.TestContext + idx *Indexer + syncs *syncRecorder + stopCh chan struct{} + loopDone chan error +} + +// The loop's ticker is left at an hour: a BgTicker cannot tick faster than its +// 5s resume wait. The start delay and pokes reach the same attemptSync. +func newAppStateSyncLoop(t *testing.T, state keybase1.MobileAppState) *syncLoopTest { + tc := externalstest.SetupTest(t, "indexer-appstate", 0) + t.Cleanup(tc.Cleanup) + tc.G.MobileAppState.Update(state) + g := globals.NewContext(tc.G, &globals.ChatContext{CtxFactory: stubCtxFactory{}}) + idx := NewIndexer(g) + idx.SetStartSyncDelay(0) + idx.syncInterval = time.Hour + s := &syncLoopTest{ + t: t, + tc: tc, + idx: idx, + syncs: &syncRecorder{}, + stopCh: make(chan struct{}), + loopDone: make(chan error, 1), + } + idx.selectiveSync = s.syncs.sync + return s +} + +func startAppStateSyncLoop(t *testing.T, state keybase1.MobileAppState) *syncLoopTest { + s := newAppStateSyncLoop(t, state) + s.start() + return s +} + +func (s *syncLoopTest) start() { + go func() { s.loopDone <- s.idx.SyncLoop(s.stopCh) }() +} + +func (s *syncLoopTest) stop() { + close(s.stopCh) + select { + case err := <-s.loopDone: + require.NoError(s.t, err) + case <-time.After(10 * time.Second): + require.FailNow(s.t, "SyncLoop did not stop") + } +} + +// poke sends a poke and returns once the loop has finished handling it: the +// loop takes one message at a time, so taking a second poke means it is done +// with the first. +func (s *syncLoopTest) poke() { + for range 2 { + s.idx.PokeSync(context.Background()) + require.Eventually(s.t, func() bool { return len(s.idx.pokeSyncCh) == 0 }, + 10*time.Second, time.Millisecond, "poke not taken") + } +} + +func (s *syncLoopTest) requireActive(active int, msg string) { + s.t.Helper() + require.Eventually(s.t, func() bool { + _, got := s.syncs.counts() + return got == active + }, 10*time.Second, time.Millisecond, msg) +} + +func TestSyncLoopDoesNotSyncOutsideForeground(t *testing.T) { + for _, state := range []keybase1.MobileAppState{ + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } { + t.Run(state.String(), func(t *testing.T) { + s := startAppStateSyncLoop(t, state) + defer s.stop() + for range 5 { + s.poke() + } + time.Sleep(100 * time.Millisecond) + starts, _ := s.syncs.counts() + require.Zero(t, starts, "synced in %v", state) + + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + s.poke() + s.requireActive(1, "no sync after FOREGROUND") + }) + } +} + +// A sync skipped outside FOREGROUND, like the one-shot start sync, runs once +// the app comes to FOREGROUND, with no poke or tick to trigger it. +func TestSyncLoopRunsDeferredSyncOnForeground(t *testing.T) { + s := startAppStateSyncLoop(t, keybase1.MobileAppState_BACKGROUNDACTIVE) + defer s.stop() + // The start delay is 0, so the start sync has been skipped once the loop + // takes a poke. + s.poke() + starts, _ := s.syncs.counts() + require.Zero(t, starts) + + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + s.requireActive(1, "deferred sync did not run on FOREGROUND") + starts, _ = s.syncs.counts() + require.Equal(t, 1, starts) +} + +func TestSyncLoopBackgroundCancelsSync(t *testing.T) { + s := startAppStateSyncLoop(t, keybase1.MobileAppState_FOREGROUND) + defer s.stop() + s.poke() + s.requireActive(1, "no sync in FOREGROUND") + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + s.requireActive(0, "sync not canceled by BACKGROUND") + s.poke() + time.Sleep(50 * time.Millisecond) + starts, active := s.syncs.counts() + require.Equal(t, 1, starts) + require.Zero(t, active) +} + +// The loop can start a sync on a poke before it wakes for the change into +// FOREGROUND. A BACKGROUND that lands before it returns to its select must +// still cancel that sync. +func TestSyncLoopBackgroundAfterUnobservedForeground(t *testing.T) { + s := newAppStateSyncLoop(t, keybase1.MobileAppState_BACKGROUND) + var beforeOnce, afterOnce sync.Once + s.idx.beforeSyncStateCheck = func() { + beforeOnce.Do(func() { s.tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) }) + } + s.idx.afterSyncStart = func() { + afterOnce.Do(func() { + for { + if _, active := s.syncs.counts(); active == 1 { + break + } + time.Sleep(time.Millisecond) + } + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + }) + } + s.start() + defer s.stop() + s.poke() + s.requireActive(0, "sync kept running in BACKGROUND") + starts, _ := s.syncs.counts() + require.Equal(t, 1, starts) +} + +func TestSyncLoopAppStateStress(t *testing.T) { + s := newAppStateSyncLoop(t, keybase1.MobileAppState_FOREGROUND) + baseline := runtime.NumGoroutine() + s.start() + states := []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } + var wg sync.WaitGroup + for w := range 4 { + wg.Go(func() { + rng := rand.New(rand.NewSource(int64(w))) + for range 500 { + s.tc.G.MobileAppState.Update(states[rng.Intn(len(states))]) + if rng.Intn(4) == 0 { + s.idx.PokeSync(context.Background()) + } + } + }) + } + wg.Wait() + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + s.requireActive(0, "sync running in BACKGROUND") + s.tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + s.poke() + s.requireActive(1, "no sync in FOREGROUND") + s.stop() + s.requireActive(0, "sync outlived the loop") + 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") +} diff --git a/go/chat/types/interfaces.go b/go/chat/types/interfaces.go index 5212f879fab5..4e9bb7ddc473 100644 --- a/go/chat/types/interfaces.go +++ b/go/chat/types/interfaces.go @@ -687,7 +687,8 @@ type ( Delete(ctx context.Context, jobID chat1.ArchiveJobID, deleteOutputPath bool) (err error) // Sets (possibly updating) the job to the given state. // cancel stops a running job by cancelling it's context and returns it's current state - Set(ctx context.Context, cancel PauseArchiveFn, job chat1.ArchiveChatJob) (err error) + // uid is the user the job runs as; a job for any user but the current one is refused. + Set(ctx context.Context, uid gregor1.UID, cancel PauseArchiveFn, job chat1.ArchiveChatJob) (err error) // Stop a running job Pause(ctx context.Context, jobID chat1.ArchiveJobID) (err error) // Resume a paused job diff --git a/go/ephemeral/keygen_loop_test.go b/go/ephemeral/keygen_loop_test.go new file mode 100644 index 000000000000..a457f4d354ca --- /dev/null +++ b/go/ephemeral/keygen_loop_test.go @@ -0,0 +1,93 @@ +package ephemeral + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +func TestKeygenLoopSeedsFromState(t *testing.T) { + tc := libkb.SetupTest(t, "ephemeral", 2) + defer tc.Cleanup() + mctx := libkb.NewMetaContextForTest(tc) + appState := tc.G.MobileAppState + appState.Update(keybase1.MobileAppState_BACKGROUNDACTIVE) + + var runs atomic.Int32 + waiting := make(chan keybase1.MobileAppState, 10) + stopCh := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + (&EKLib{}).keygenLoop(mctx, stopCh, nil, + func() time.Duration { return 0 }, + func() { runs.Add(1) }, + func(state keybase1.MobileAppState) { waiting <- state }) + }() + + next := func(want keybase1.MobileAppState) { + t.Helper() + select { + case got := <-waiting: + require.Equal(t, want, got) + case <-time.After(10 * time.Second): + t.Fatal("keygen loop did not wait") + } + } + + // A background-active launch is not a transition into BACKGROUNDACTIVE. + next(keybase1.MobileAppState_BACKGROUNDACTIVE) + require.Zero(t, runs.Load()) + + appState.Update(keybase1.MobileAppState_FOREGROUND) + next(keybase1.MobileAppState_FOREGROUND) + require.Zero(t, runs.Load()) + + appState.Update(keybase1.MobileAppState_BACKGROUNDACTIVE) + next(keybase1.MobileAppState_BACKGROUNDACTIVE) + require.EqualValues(t, 1, runs.Load()) + + close(stopCh) + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("keygen loop did not stop") + } +} + +// Keygen runs when work wakes the app in the background and when the app +// leaves BACKGROUND, but not when the UI merely stops being active. +func TestKeygenOnTransition(t *testing.T) { + const ( + fg = keybase1.MobileAppState_FOREGROUND + bg = keybase1.MobileAppState_BACKGROUND + ina = keybase1.MobileAppState_INACTIVE + bga = keybase1.MobileAppState_BACKGROUNDACTIVE + ) + cases := []struct { + prev, state keybase1.MobileAppState + run bool + }{ + {bg, bga, true}, + {fg, bga, true}, + {ina, bga, true}, + {bg, ina, true}, + // NextUpdate collapses willEnterForeground's INACTIVE and + // didBecomeActive's FOREGROUND when they land together. + {bg, fg, true}, + {ina, fg, false}, + {fg, ina, false}, + {bga, ina, false}, + {bga, fg, false}, + {ina, bg, false}, + {bga, bg, false}, + {fg, bg, false}, + } + for _, tc := range cases { + require.Equal(t, tc.run, keygenOnTransition(tc.prev, tc.state), "%v -> %v", tc.prev, tc.state) + } +} diff --git a/go/ephemeral/lib.go b/go/ephemeral/lib.go index e17ac4b2696d..66ec64c9cb4e 100644 --- a/go/ephemeral/lib.go +++ b/go/ephemeral/lib.go @@ -118,35 +118,52 @@ func (e *EKLib) backgroundKeygen(mctx libkb.MetaContext, stopCh <-chan struct{}) runIfNeeded(true /* force */) ticker := libkb.NewBgTicker(keygenInterval) - state := keybase1.MobileAppState_FOREGROUND - // Run every hour but also check if enough wall clock time has elapsed when - // we are in a BACKGROUNDACTIVE state. + defer ticker.Stop() + e.keygenLoop(mctx, stopCh, ticker.C, func() time.Duration { return libkb.RandomJitter(time.Second) }, + func() { runIfNeeded(false /* force */) }, nil) +} + +// keygenLoop runs run on every tick, and also when the app enters +// BACKGROUNDACTIVE or leaves BACKGROUND, after a jittered pause so it doesn't +// stampede for resources with other background tasks (libkb.BgTicker handles +// this internally for ticks). waiting, if set, is told the state before each +// wait. +func (e *EKLib) keygenLoop(mctx libkb.MetaContext, stopCh <-chan struct{}, tick <-chan time.Time, + jitter func() time.Duration, run func(), waiting func(keybase1.MobileAppState), +) { + state := mctx.G().MobileAppState.State() for { + if waiting != nil { + waiting(state) + } select { - case <-ticker.C: - runIfNeeded(false /* force */) + case <-tick: + run() case <-mctx.G().MobileAppState.NextUpdate(state): + prev := 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 - // handles this internally, so we only need to throttle on - // MobileAppState change. + if keygenOnTransition(prev, state) { select { - case <-time.After(libkb.RandomJitter(time.Second)): - runIfNeeded(false /* force */) + case <-time.After(jitter()): + run() case <-stopCh: - ticker.Stop() return } } case <-stopCh: - ticker.Stop() return } } } +// keygenOnTransition: work woke the app in the background, or the app left +// BACKGROUND. NextUpdate collapses changes, so a return to the foreground can +// arrive as BACKGROUND to FOREGROUND without the INACTIVE in between. +func keygenOnTransition(prev, state keybase1.MobileAppState) bool { + return state == keybase1.MobileAppState_BACKGROUNDACTIVE || + (prev == keybase1.MobileAppState_BACKGROUND && state != keybase1.MobileAppState_BACKGROUND) +} + func (e *EKLib) SetClock(clock clockwork.Clock) { e.clock = clock } diff --git a/go/kbfs/libkbfs/app_state_test.go b/go/kbfs/libkbfs/app_state_test.go new file mode 100644 index 000000000000..376d5de3b409 --- /dev/null +++ b/go/kbfs/libkbfs/app_state_test.go @@ -0,0 +1,259 @@ +// Copyright 2026 Keybase Inc. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package libkbfs + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/keybase/client/go/logger" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +// fakeAppState is a settable env.AppStateUpdater. appWaits and netWaits +// receive the state passed to every NextAppStateUpdate and +// NextNetworkStateUpdate call, while they have room. +type fakeAppState struct { + lock sync.Mutex + appState keybase1.MobileAppState + netState keybase1.MobileNetworkState + appChanged chan struct{} + netChanged chan struct{} + appWaits chan keybase1.MobileAppState + netWaits chan keybase1.MobileNetworkState +} + +func newFakeAppState( + appState keybase1.MobileAppState, netState keybase1.MobileNetworkState, +) *fakeAppState { + return &fakeAppState{ + appState: appState, + netState: netState, + appChanged: make(chan struct{}), + netChanged: make(chan struct{}), + appWaits: make(chan keybase1.MobileAppState, 1000), + netWaits: make(chan keybase1.MobileNetworkState, 1000), + } +} + +var closedAppStateCh = func() chan struct{} { + ch := make(chan struct{}) + close(ch) + return ch +}() + +func (f *fakeAppState) NextAppStateUpdate( + lastState keybase1.MobileAppState, +) <-chan struct{} { + f.lock.Lock() + defer f.lock.Unlock() + select { + case f.appWaits <- lastState: + default: + } + if lastState != f.appState { + return closedAppStateCh + } + return f.appChanged +} + +func (f *fakeAppState) NextNetworkStateUpdate( + lastState keybase1.MobileNetworkState, +) <-chan struct{} { + f.lock.Lock() + defer f.lock.Unlock() + select { + case f.netWaits <- lastState: + default: + } + if lastState != f.netState { + return closedAppStateCh + } + return f.netChanged +} + +func (f *fakeAppState) AppState() keybase1.MobileAppState { + f.lock.Lock() + defer f.lock.Unlock() + return f.appState +} + +func (f *fakeAppState) NetworkState() keybase1.MobileNetworkState { + f.lock.Lock() + defer f.lock.Unlock() + return f.netState +} + +// setAppState changes the app state and waits until someone waits for the +// next change from it. +func (f *fakeAppState) setAppState(t *testing.T, state keybase1.MobileAppState) { + t.Helper() + f.drain() + f.setAppStateNoWait(state) + waitFor(t, f.appWaits, state) +} + +// setNetworkState changes the network state and waits until someone waits +// for the next change from it. +func (f *fakeAppState) setNetworkState(t *testing.T, state keybase1.MobileNetworkState) { + t.Helper() + f.drain() + f.setNetworkStateNoWait(state) + waitFor(t, f.netWaits, state) +} + +func (f *fakeAppState) drain() { + for { + select { + case <-f.appWaits: + case <-f.netWaits: + default: + return + } + } +} + +func waitFor[T comparable](t *testing.T, waits <-chan T, want T) { + t.Helper() + timeout := time.After(10 * time.Second) + for { + select { + case got := <-waits: + if got == want { + return + } + case <-timeout: + t.Fatalf("nothing waited for a change from %v", want) + } + } +} + +func (f *fakeAppState) setAppStateNoWait(state keybase1.MobileAppState) { + f.lock.Lock() + defer f.lock.Unlock() + if f.appState != state { + f.appState = state + close(f.appChanged) + f.appChanged = make(chan struct{}) + } +} + +func (f *fakeAppState) setNetworkStateNoWait(state keybase1.MobileNetworkState) { + f.lock.Lock() + defer f.lock.Unlock() + if f.netState != state { + f.netState = state + close(f.netChanged) + f.netChanged = make(chan struct{}) + } +} + +type fbmNoTimedQRConfig struct { + Config +} + +func (c fbmNoTimedQRConfig) Mode() InitMode { + return modeTestWithNoTimedQR{modeTest{NewInitModeFromType(InitDefault)}} +} + +// The folder block manager's app-state waits end on shutdown while the app +// is backgrounded. +func TestFolderBlockManagerPausedLoopsExitOnShutdown(t *testing.T) { + loops := map[string]func(fbm *folderBlockManager){ + "reclaimQuota": (*folderBlockManager).reclaimQuotaInBackground, + "cleanDiskCaches": (*folderBlockManager).cleanDiskCachesInBackground, + } + for name, loop := range loops { + t.Run(name, func(t *testing.T) { + appState := newFakeAppState( + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileNetworkState_WIFI) + fbm := &folderBlockManager{ + appStateUpdater: appState, + config: fbmNoTimedQRConfig{}, + log: logger.NewTestLogger(t), + shutdownChan: make(chan struct{}), + forceReclamationChan: make(chan struct{}, 1), + latestMergedChan: make(chan struct{}, 1), + } + done := make(chan struct{}) + go func() { + defer close(done) + loop(fbm) + }() + + waitFor(t, appState.appWaits, keybase1.MobileAppState_BACKGROUND) + fbm.shutdown() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("paused loop did not exit on shutdown") + } + }) + } +} + +// requirePaused checks the prefetcher's pause after the state changes that +// setAppState/setNetworkState waited for. +func requirePaused(t *testing.T, q *blockRetrievalQueue, want bool, msg string) { + t.Helper() + paused, _ := q.Prefetcher().(*blockPrefetcher).getPaused() + require.Equal(t, want, paused, msg) +} + +// Neither pause reason ends the other's pause. +func TestPrefetcherPauseReasonsDoNotUndoEachOther(t *testing.T) { + for _, appFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("appFirst=%t", appFirst), func(t *testing.T) { + bg := newFakeBlockGetter(false) + config := newTestBlockRetrievalConfig(t, bg, nil) + appState := newFakeAppState( + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileNetworkState_WIFI) + q := newBlockRetrievalQueue(1, 1, 0, config, appState) + require.NotNil(t, q) + prefetchSyncCh := make(chan struct{}) + defer shutdownPrefetcherTest(t, q, prefetchSyncCh) + <-q.TogglePrefetcher(true, prefetchSyncCh, nil) + // The first iteration reads the network state; the second waits + // for a change from it. + notifySyncCh(t, prefetchSyncCh) + notifySyncCh(t, prefetchSyncCh) + waitFor(t, appState.netWaits, keybase1.MobileNetworkState_WIFI) + requirePaused(t, q, false, "paused in the foreground on wifi") + + if appFirst { + appState.setAppState(t, keybase1.MobileAppState_BACKGROUND) + requirePaused(t, q, true, "not paused in the background") + appState.setNetworkState(t, keybase1.MobileNetworkState_CELLULAR) + requirePaused(t, q, true, "not paused in the background on cellular") + } else { + appState.setNetworkState(t, keybase1.MobileNetworkState_CELLULAR) + requirePaused(t, q, true, "not paused on cellular") + appState.setAppState(t, keybase1.MobileAppState_BACKGROUND) + requirePaused(t, q, true, "not paused in the background on cellular") + } + + appState.setAppState(t, keybase1.MobileAppState_INACTIVE) + requirePaused(t, q, true, "an app-state change undid the cellular pause") + appState.setAppState(t, keybase1.MobileAppState_FOREGROUND) + requirePaused(t, q, true, "foregrounding undid the cellular pause") + + appState.setAppState(t, keybase1.MobileAppState_BACKGROUND) + requirePaused(t, q, true, "not paused in the background on cellular") + appState.setNetworkState(t, keybase1.MobileNetworkState_WIFI) + requirePaused(t, q, true, "leaving cellular undid the background pause") + + appState.setAppStateNoWait(keybase1.MobileAppState_FOREGROUND) + require.Eventually(t, func() bool { + paused, _ := q.Prefetcher().(*blockPrefetcher).getPaused() + return !paused + }, 10*time.Second, time.Millisecond, "still paused in the foreground on wifi") + }) + } +} diff --git a/go/kbfs/libkbfs/folder_block_manager.go b/go/kbfs/libkbfs/folder_block_manager.go index 54206e62f5d6..0b1de568ca5b 100644 --- a/go/kbfs/libkbfs/folder_block_manager.go +++ b/go/kbfs/libkbfs/folder_block_manager.go @@ -1284,6 +1284,21 @@ func isPermanentQRError(err error) bool { } } +// WaitForeground blocks until u reports the app state as FOREGROUND, and +// reports false if stop closes first. +func WaitForeground(u env.AppStateUpdater, stop <-chan struct{}) bool { + state := u.AppState() + for state != keybase1.MobileAppState_FOREGROUND { + select { + case <-u.NextAppStateUpdate(state): + case <-stop: + return false + } + state = u.AppState() + } + return true +} + func (fbm *folderBlockManager) reclaimQuotaInBackground() { autoQR := true timer := time.NewTimer(fbm.config.Mode().QuotaReclamationPeriod()) @@ -1314,15 +1329,18 @@ func (fbm *folderBlockManager) reclaimQuotaInBackground() { case <-fbm.shutdownChan: return case <-fbm.appStateUpdater.NextAppStateUpdate(state): - state = fbm.appStateUpdater.AppState() - for state != keybase1.MobileAppState_FOREGROUND { + if s := fbm.appStateUpdater.AppState(); s != keybase1.MobileAppState_FOREGROUND { fbm.log.CDebugf(context.Background(), - "Pausing QR while not foregrounded: state=%s", state) - <-fbm.appStateUpdater.NextAppStateUpdate(state) - state = fbm.appStateUpdater.AppState() + "Pausing QR while not foregrounded: state=%s", s) + // forceReclamationChan is not read while paused, so a forced + // reclamation blocks its sender until the app is + // foregrounded again or shuts down. + if !WaitForeground(fbm.appStateUpdater, fbm.shutdownChan) { + return + } + fbm.log.CDebugf( + context.Background(), "Resuming QR while foregrounded") } - fbm.log.CDebugf( - context.Background(), "Resuming QR while foregrounded") continue case <-timerChan: fbm.reclamationGroup.Add(1) @@ -1589,16 +1607,15 @@ func (fbm *folderBlockManager) cleanDiskCachesInBackground() { case <-fbm.shutdownChan: return case <-fbm.appStateUpdater.NextAppStateUpdate(state): - state = fbm.appStateUpdater.AppState() - for state != keybase1.MobileAppState_FOREGROUND { + if s := fbm.appStateUpdater.AppState(); s != keybase1.MobileAppState_FOREGROUND { fbm.log.CDebugf(context.Background(), - "Pausing sync-cache cleaning while not foregrounded: "+ - "state=%s", state) - <-fbm.appStateUpdater.NextAppStateUpdate(state) - state = fbm.appStateUpdater.AppState() + "Pausing sync-cache cleaning while not foregrounded: state=%s", s) + if !WaitForeground(fbm.appStateUpdater, fbm.shutdownChan) { + return + } + fbm.log.CDebugf(context.Background(), + "Resuming sync-cache cleaning while foregrounded") } - fbm.log.CDebugf(context.Background(), - "Resuming sync-cache cleaning while foregrounded") continue } diff --git a/go/kbfs/libkbfs/prefetcher.go b/go/kbfs/libkbfs/prefetcher.go index c18b7c95f71a..ec17020b85f8 100644 --- a/go/kbfs/libkbfs/prefetcher.go +++ b/go/kbfs/libkbfs/prefetcher.go @@ -1339,30 +1339,6 @@ func (p *blockPrefetcher) getPaused() (paused bool, ch <-chan struct{}) { return p.paused, p.pausedCh } -func (p *blockPrefetcher) handleAppStateChange( - appState *keybase1.MobileAppState, -) { - defer func() { - p.setPaused(false) - }() - - // Pause the prefetcher when backgrounded. - for *appState != keybase1.MobileAppState_FOREGROUND { - p.setPaused(true) - p.log.CDebugf( - context.TODO(), "Pausing prefetcher while backgrounded") - select { - case <-p.appStateUpdater.NextAppStateUpdate(*appState): - *appState = p.appStateUpdater.AppState() - case req := <-p.prefetchStatusCh.Out(): - p.handleStatusRequest(req.(*prefetchStatusRequest)) - continue - case <-p.almostDoneCh: - return - } - } -} - type prefetcherSubscriber struct { ch chan<- struct{} clientID SubscriptionManagerClientID @@ -1392,44 +1368,51 @@ func (ps prefetcherSubscriber) OnNonPathChange( } } -func (p *blockPrefetcher) handleNetStateChange( - netState *keybase1.MobileNetworkState, subCh <-chan struct{}, -) { - for *netState != keybase1.MobileNetworkState_CELLULAR { - return +func (p *blockPrefetcher) syncOnCellular() bool { + // Default to not syncing while on a cell network. + db := p.config.GetSettingsDB() + if db == nil { + return false } + s, err := db.Settings(context.TODO()) + return err == nil && s.SyncOnCellular +} - defer func() { - p.setPaused(false) - }() - - for *netState == keybase1.MobileNetworkState_CELLULAR { - // Default to not syncing while on a cell network. - syncOnCellular := false - db := p.config.GetSettingsDB() - if db != nil { - s, err := db.Settings(context.TODO()) - if err == nil { - syncOnCellular = s.SyncOnCellular - } - } - - if syncOnCellular { - // Can ignore this network change. - break +// waitWhilePaused pauses the prefetcher while the app is not in the +// foreground, or while on a cell network without syncing on cellular, and +// returns once neither holds or the prefetcher is shutting down. It watches +// both states whichever one paused it, so the end of one reason never +// unpauses while the other still holds. +func (p *blockPrefetcher) waitWhilePaused( + appState *keybase1.MobileAppState, netState *keybase1.MobileNetworkState, + subCh <-chan struct{}, +) { + defer p.setPaused(false) + for { + appPaused := *appState != keybase1.MobileAppState_FOREGROUND + netPaused := *netState == keybase1.MobileNetworkState_CELLULAR && + !p.syncOnCellular() + if !appPaused && !netPaused { + return } - p.setPaused(true) - p.log.CDebugf( - context.TODO(), "Pausing prefetcher on cell network") + if appPaused { + p.log.CDebugf( + context.TODO(), "Pausing prefetcher while backgrounded") + } + if netPaused { + p.log.CDebugf( + context.TODO(), "Pausing prefetcher on cell network") + } select { + case <-p.appStateUpdater.NextAppStateUpdate(*appState): + *appState = p.appStateUpdater.AppState() case <-p.appStateUpdater.NextNetworkStateUpdate(*netState): *netState = p.appStateUpdater.NetworkState() case <-subCh: p.log.CDebugf(context.TODO(), "Settings changed") case req := <-p.prefetchStatusCh.Out(): p.handleStatusRequest(req.(*prefetchStatusRequest)) - continue case <-p.almostDoneCh: return } @@ -1547,10 +1530,10 @@ func (p *blockPrefetcher) run( <-ch case <-p.appStateUpdater.NextAppStateUpdate(appState): appState = p.appStateUpdater.AppState() - p.handleAppStateChange(&appState) + p.waitWhilePaused(&appState, &netState, subCh) case <-p.appStateUpdater.NextNetworkStateUpdate(netState): netState = p.appStateUpdater.NetworkState() - p.handleNetStateChange(&netState, subCh) + p.waitWhilePaused(&appState, &netState, subCh) case <-subCh: // Settings have changed, so recheck the network state. netState = keybase1.MobileNetworkState_NONE diff --git a/go/kbfs/search/indexer.go b/go/kbfs/search/indexer.go index d2b6a054a390..0ef94a097385 100644 --- a/go/kbfs/search/indexer.go +++ b/go/kbfs/search/indexer.go @@ -1386,6 +1386,17 @@ func (i *Indexer) loop(ctx context.Context) { ctx, "Couldn't register for synced TLF updates: %+v", err) } + // stopped closes when either ctx or i.shutdownCh ends the loop, so the + // foreground wait below can watch both through one channel. + stopped := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + case <-i.shutdownCh: + } + close(stopped) + }() + outerLoop: for { err := i.loadIndex(ctx) @@ -1402,18 +1413,21 @@ outerLoop: i.log.CDebugf(ctx, "User changed") continue outerLoop 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. - for state != keybase1.MobileAppState_FOREGROUND { + if s := kbCtx.AppState(); s != keybase1.MobileAppState_FOREGROUND { i.log.CDebugf(ctx, - "Pausing indexing while not foregrounded: state=%s", - state) - <-kbCtx.NextAppStateUpdate(state) - state = kbCtx.AppState() + "Pausing indexing while not foregrounded: state=%s", s) + if !libkbfs.WaitForeground(kbCtx, stopped) { + if ctx.Err() == nil { + i.cancelLoop() + } + return + } + i.log.CDebugf(ctx, "Resuming indexing while foregrounded") } - i.log.CDebugf(ctx, "Resuming indexing while foregrounded") + state = keybase1.MobileAppState_FOREGROUND continue case m := <-i.tlfCh: ctx := i.makeContext(ctx) diff --git a/go/kbfs/search/indexer_app_state_test.go b/go/kbfs/search/indexer_app_state_test.go new file mode 100644 index 000000000000..34931df18f23 --- /dev/null +++ b/go/kbfs/search/indexer_app_state_test.go @@ -0,0 +1,93 @@ +// Copyright 2026 Keybase Inc. All rights reserved. +// Use of this source code is governed by a BSD +// license that can be found in the LICENSE file. + +package search + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/keybase/client/go/kbfs/env" + "github.com/keybase/client/go/kbfs/idutil" + "github.com/keybase/client/go/kbfs/libcontext" + "github.com/keybase/client/go/kbfs/libkbfs" + "github.com/keybase/client/go/logger" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +// backgroundKbCtx reports a BACKGROUND app state that never changes, and +// sends every state it is asked to wait on to waits. +type backgroundKbCtx struct { + env.Context + waits chan keybase1.MobileAppState +} + +func (c backgroundKbCtx) NextAppStateUpdate( + lastState keybase1.MobileAppState, +) <-chan struct{} { + select { + case c.waits <- lastState: + default: + } + if lastState != keybase1.MobileAppState_BACKGROUND { + ch := make(chan struct{}) + close(ch) + return ch + } + return nil +} + +func (c backgroundKbCtx) AppState() keybase1.MobileAppState { + return keybase1.MobileAppState_BACKGROUND +} + +type backgroundConfig struct { + libkbfs.Config + kbCtx backgroundKbCtx +} + +func (c backgroundConfig) KbContext() libkbfs.Context { + return c.kbCtx +} + +func TestIndexerPausedLoopExitsOnShutdown(t *testing.T) { + ctx := libcontext.BackgroundContextWithCancellationDelayer() + ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + config := libkbfs.MakeTestConfigOrBust(t, "user1") + defer libkbfs.CheckConfigAndShutdown(ctx, t, config) + + bgConfig := backgroundConfig{ + Config: config, + kbCtx: backgroundKbCtx{ + Context: config.KbContext(), + waits: make(chan keybase1.MobileAppState, 100), + }, + } + noIndex := func( + context.Context, libkbfs.Config, idutil.SessionInfo, logger.Logger, + ) (context.Context, libkbfs.Config, func(context.Context) error, error) { + return nil, nil, nil, errors.New("no index in this test") + } + i, err := newIndexerWithConfigInit( + bgConfig, noIndex, testKVStoreName("TestIndexerPausedLoopExitsOnShutdown")) + require.NoError(t, err) + + timeout := time.After(30 * time.Second) + for paused := false; !paused; { + select { + case state := <-bgConfig.kbCtx.waits: + paused = state == keybase1.MobileAppState_BACKGROUND + case <-timeout: + t.Fatal("indexer loop did not pause") + } + } + + shutdownCtx, shutdownCancel := context.WithTimeout(ctx, 10*time.Second) + defer shutdownCancel() + require.NoError(t, i.Shutdown(shutdownCtx), "paused indexer loop did not exit on shutdown") +} diff --git a/go/libkb/bgticker.go b/go/libkb/bgticker.go index 48d7b4d6eb65..d8eda492f1e5 100644 --- a/go/libkb/bgticker.go +++ b/go/libkb/bgticker.go @@ -1,6 +1,7 @@ package libkb import ( + "sync" "time" ) @@ -11,6 +12,8 @@ type BgTicker struct { c chan time.Time ticker *time.Ticker resumeWait time.Duration + done chan struct{} + stopOnce sync.Once } // This ticker wrap's Go's time.Ticker to wait a given time.Duration before @@ -30,18 +33,38 @@ func NewBgTickerWithWait(duration time.Duration, wait time.Duration) *BgTicker { c: c, ticker: time.NewTicker(duration - wait), resumeWait: wait, + done: make(chan struct{}), } go t.tick() return t } +// tick ends on Stop: a stopped time.Ticker never closes its channel, and +// nobody may be left to read C. func (t *BgTicker) tick() { - for c := range t.ticker.C { - time.Sleep(RandomJitter(t.resumeWait)) - t.c <- c + for { + var c time.Time + select { + case c = <-t.ticker.C: + case <-t.done: + return + } + wait := time.NewTimer(RandomJitter(t.resumeWait)) + select { + case <-wait.C: + case <-t.done: + wait.Stop() + return + } + select { + case t.c <- c: + case <-t.done: + return + } } } func (t *BgTicker) Stop() { t.ticker.Stop() + t.stopOnce.Do(func() { close(t.done) }) } diff --git a/go/libkb/bgticker_test.go b/go/libkb/bgticker_test.go index 8cecf7f8c6d1..dd5d2e7dd9e7 100644 --- a/go/libkb/bgticker_test.go +++ b/go/libkb/bgticker_test.go @@ -1,6 +1,7 @@ package libkb import ( + "runtime" "testing" "time" @@ -27,3 +28,33 @@ func TestBgTicker(t *testing.T) { } } } + +// Stop ends the tick goroutine whether it waits for a tick, waits out the +// resume wait, or is blocked handing a tick to a reader that went away. +func TestBgTickerStopEndsGoroutine(t *testing.T) { + baseline := runtime.NumGoroutine() + var tickers []*BgTicker + for i := range 30 { + switch i % 3 { + case 0: + tickers = append(tickers, NewBgTickerWithWait(time.Hour, time.Millisecond)) + case 1: + tickers = append(tickers, NewBgTickerWithWait(time.Hour+time.Millisecond, time.Hour)) + default: + ticker := NewBgTickerWithWait(2*time.Millisecond, time.Millisecond) + // fill C, so the next tick blocks on the send + <-ticker.C + tickers = append(tickers, ticker) + } + } + time.Sleep(50 * time.Millisecond) + for _, ticker := range tickers { + ticker.Stop() + ticker.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 tick goroutines") +}