From 7770b2d3436bbcfe972cefcf86d565e80d201b2f Mon Sep 17 00:00:00 2001 From: Joshua Blum Date: Thu, 17 Sep 2026 11:18:00 -0400 Subject: [PATCH] wait for Stop and clear uid when restarting reused modules --- go/chat/archive.go | 1 + go/chat/bots/commands.go | 1 + go/chat/convloader.go | 1 + go/chat/ephemeral_purger.go | 1 + go/chat/ephemeral_purger_test.go | 4 ++-- go/chat/inboxsource.go | 12 +++++++--- go/chat/localizer.go | 39 ++++++++++++++++++++++++-------- go/chat/maps/livelocation.go | 1 + go/chat/retry.go | 1 + go/chat/search/indexer.go | 21 +++++++++-------- go/chat/uiinboxloader.go | 26 +++++++++++++++------ 11 files changed, 77 insertions(+), 31 deletions(-) diff --git a/go/chat/archive.go b/go/chat/archive.go index 69af6e9cdfb1..12adeae4bb0d 100644 --- a/go/chat/archive.go +++ b/go/chat/archive.go @@ -283,6 +283,7 @@ func (r *ChatArchiveRegistry) Stop(ctx context.Context) chan struct{} { r.Debug(ctx, err.Error()) } r.started = false + r.uid = nil close(r.stopCh) go func() { r.Debug(context.Background(), "Stop: waiting for shutdown") diff --git a/go/chat/bots/commands.go b/go/chat/bots/commands.go index cd7545a787ad..9ee5af22bc2d 100644 --- a/go/chat/bots/commands.go +++ b/go/chat/bots/commands.go @@ -122,6 +122,7 @@ func (b *CachingBotCommandManager) Stop(ctx context.Context) chan struct{} { if b.started { close(b.stopCh) b.started = false + b.uid = nil go func() { err := b.eg.Wait() if err != nil { diff --git a/go/chat/convloader.go b/go/chat/convloader.go index 1523ac249411..916939c0972a 100644 --- a/go/chat/convloader.go +++ b/go/chat/convloader.go @@ -235,6 +235,7 @@ func (b *BackgroundConvLoader) Stop(ctx context.Context) chan struct{} { ch := make(chan struct{}) if b.started { b.started = false + b.uid = nil close(b.stopCh) b.stopCh = make(chan struct{}) go func() { diff --git a/go/chat/ephemeral_purger.go b/go/chat/ephemeral_purger.go index 2b8dd61d3694..02d1c5c643c5 100644 --- a/go/chat/ephemeral_purger.go +++ b/go/chat/ephemeral_purger.go @@ -173,6 +173,7 @@ func (b *BackgroundEphemeralPurger) Stop(ctx context.Context) (ch chan struct{}) if b.started { close(b.shutdownCh) b.started = false + b.uid = nil go func() { if err := b.eg.Wait(); err != nil { b.Debug(ctx, "error stopping background loop: %v", err) diff --git a/go/chat/ephemeral_purger_test.go b/go/chat/ephemeral_purger_test.go index 7dce4dc5e0a9..69ba6fe8bf51 100644 --- a/go/chat/ephemeral_purger_test.go +++ b/go/chat/ephemeral_purger_test.go @@ -378,11 +378,11 @@ func TestQueueState(t *testing.T) { purger := NewBackgroundEphemeralPurger(g) purger.SetClock(world.Fc) purger.Start(context.Background(), uid) - <-purger.Stop(context.Background()) + defer func() { <-purger.Stop(context.Background()) }() + require.Equal(t, 0, purger.Len()) pq := purger.pq require.NotNil(t, pq) - require.Zero(t, pq.Len()) require.Nil(t, pq.Peek()) now := world.Fc.Now() diff --git a/go/chat/inboxsource.go b/go/chat/inboxsource.go index ef75d92e7004..255df9e7a91f 100644 --- a/go/chat/inboxsource.go +++ b/go/chat/inboxsource.go @@ -665,12 +665,13 @@ func (s *HybridInboxSource) Connected(ctx context.Context) { func (s *HybridInboxSource) Start(ctx context.Context, uid gregor1.UID) { defer s.Trace(ctx, nil, "Start")() + s.Lock() + waitCh := s.doStopLocked() + s.Unlock() + <-waitCh s.baseInboxSource.Start(ctx, uid) s.Lock() defer s.Unlock() - if s.started { - return - } s.stopCh = make(chan struct{}) s.started = true s.uid = uid @@ -683,10 +684,15 @@ func (s *HybridInboxSource) Stop(ctx context.Context) chan struct{} { <-s.baseInboxSource.Stop(ctx) s.Lock() defer s.Unlock() + return s.doStopLocked() +} + +func (s *HybridInboxSource) doStopLocked() chan struct{} { ch := make(chan struct{}) if s.started { close(s.stopCh) s.started = false + s.uid = nil go func() { _ = s.eg.Wait() close(ch) diff --git a/go/chat/localizer.go b/go/chat/localizer.go index 6a1b73e48cd6..880c705632fb 100644 --- a/go/chat/localizer.go +++ b/go/chat/localizer.go @@ -286,6 +286,8 @@ type localizerPipeline struct { suspendCount int suspendWaiters []chan struct{} jobQueue chan *localizerPipelineJob + loopDone chan struct{} + jobWG sync.WaitGroup // testing useGateCh bool @@ -338,28 +340,44 @@ func (s *localizerPipeline) clearQueue() { func (s *localizerPipeline) start(ctx context.Context) { defer s.Trace(ctx, nil, "start")() s.Lock() + waitCh := s.doStopLocked() + s.Unlock() + <-waitCh + s.Lock() defer s.Unlock() - if s.started { - close(s.stopCh) - s.stopCh = make(chan struct{}) - } s.clearQueue() s.started = true + s.stopCh = make(chan struct{}) + s.loopDone = make(chan struct{}) stopCh := s.stopCh - go s.localizeLoop(stopCh) + loopDone := s.loopDone + go s.localizeLoop(stopCh, loopDone) } func (s *localizerPipeline) stop(ctx context.Context) chan struct{} { defer s.Trace(ctx, nil, "stop")() s.Lock() defer s.Unlock() + return s.doStopLocked() +} + +func (s *localizerPipeline) doStopLocked() chan struct{} { ch := make(chan struct{}) if s.started { close(s.stopCh) - s.stopCh = make(chan struct{}) s.started = false + loopDone := s.loopDone + go func() { + if loopDone != nil { + <-loopDone + } + s.jobWG.Wait() + close(ch) + }() + } else { + close(ch) } - close(ch) + s.clearQueue() return ch } @@ -494,13 +512,16 @@ func (s *localizerPipeline) localizeJobPulled(job *localizerPipelineJob, stopCh s.Debug(job.ctx, "localizeJobPulled[%s]: job pass complete", id) } -func (s *localizerPipeline) localizeLoop(stopCh chan struct{}) { +func (s *localizerPipeline) localizeLoop(stopCh chan struct{}, loopDone chan struct{}) { ctx := context.Background() s.Debug(ctx, "localizeLoop: starting up") + defer close(loopDone) for { select { case job := <-s.jobQueue: - go s.localizeJobPulled(job, stopCh) + s.jobWG.Go(func() { + s.localizeJobPulled(job, stopCh) + }) case <-stopCh: s.Debug(ctx, "localizeLoop: shutting down") return diff --git a/go/chat/maps/livelocation.go b/go/chat/maps/livelocation.go index 76e24d886e23..532c00670120 100644 --- a/go/chat/maps/livelocation.go +++ b/go/chat/maps/livelocation.go @@ -69,6 +69,7 @@ func (l *LiveLocationTracker) Stop(ctx context.Context) chan struct{} { for _, t := range l.trackers { t.Stop() } + l.uid = nil go func() { _ = l.eg.Wait() close(ch) diff --git a/go/chat/retry.go b/go/chat/retry.go index ac656fdbe55b..eb640b1cf54d 100644 --- a/go/chat/retry.go +++ b/go/chat/retry.go @@ -391,6 +391,7 @@ func (f *FetchRetrier) Stop(ctx context.Context) chan struct{} { for _, control := range f.retriers { control.Shutdown() } + f.retriers = make(map[string]*retrierControl) ch := make(chan struct{}) close(ch) return ch diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index ac373804c636..178438ac08ca 100644 --- a/go/chat/search/indexer.go +++ b/go/chat/search/indexer.go @@ -235,7 +235,7 @@ func (idx *Indexer) PokeSync(ctx context.Context) { } func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { - ctx := globals.ChatCtx(context.Background(), idx.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, nil) + ctx := context.Background() idx.Lock() suspendCh := idx.suspendCh idx.Unlock() @@ -256,21 +256,22 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { cancelFn = nil } } - attemptSync := func(ctx context.Context) { + attemptSync := func() { if netState.IsLimited() { return } + syncCtx := globals.ChatCtx(context.Background(), idx.G(), keybase1.TLFIdentifyBehavior_CHAT_GUI, nil, nil) l.Lock() defer l.Unlock() if cancelFn != nil { - idx.Debug(ctx, "SelectiveSync already running, skipping new sync attempt") + idx.Debug(syncCtx, "SelectiveSync already running, skipping new sync attempt") return } - ctx, cancelFn = context.WithCancel(ctx) + syncCtx, cancelFn = context.WithCancel(syncCtx) syncAttemptWG.Go(func() { - idx.Debug(ctx, "running SelectiveSync") - if err := idx.SelectiveSync(ctx); err != nil { - idx.Debug(ctx, "unable to complete SelectiveSync: %v", err) + idx.Debug(syncCtx, "running SelectiveSync") + if err := idx.SelectiveSync(syncCtx); err != nil { + idx.Debug(syncCtx, "unable to complete SelectiveSync: %v", err) if idx.syncLoopCh != nil { select { case idx.syncLoopCh <- struct{}{}: @@ -303,11 +304,11 @@ func (idx *Indexer) SyncLoop(stopCh chan struct{}) error { case <-idx.cancelSyncCh: cancelSync() case <-idx.pokeSyncCh: - attemptSync(ctx) + attemptSync() case <-after: - attemptSync(ctx) + attemptSync() case <-ticker.C: - attemptSync(ctx) + attemptSync() case <-idx.G().MobileAppState.NextUpdate(appState): appState = idx.G().MobileAppState.State() switch appState { diff --git a/go/chat/uiinboxloader.go b/go/chat/uiinboxloader.go index 0410141530c0..b0cb30c5ae81 100644 --- a/go/chat/uiinboxloader.go +++ b/go/chat/uiinboxloader.go @@ -75,29 +75,39 @@ func NewUIInboxLoader(g *globals.Context) *UIInboxLoader { func (h *UIInboxLoader) Start(ctx context.Context, uid gregor1.UID) { defer h.Trace(ctx, nil, "Start")() h.Lock() - defer h.Unlock() - if h.started { - return - } + waitCh := h.doStopLocked(ctx) + h.Unlock() + <-waitCh + h.Lock() h.transmitCh = make(chan any, 1000) h.layoutCh = make(chan chat1.InboxLayoutReselectMode, 1000) h.bigTeamUnboxCh = make(chan []chat1.ConversationID, 1000) h.stopCh = make(chan struct{}) h.started = true h.uid = uid - h.eg.Go(func() error { return h.transmitLoop(h.stopCh) }) - h.eg.Go(func() error { return h.layoutLoop(h.stopCh) }) - h.eg.Go(func() error { return h.bigTeamUnboxLoop(h.stopCh) }) + h.convTransmitBatch = make(map[chat1.ConvIDStr]chat1.ConversationLocal) + stopCh := h.stopCh + h.Unlock() + h.setLastLayout(nil) + h.eg.Go(func() error { return h.transmitLoop(stopCh) }) + h.eg.Go(func() error { return h.layoutLoop(stopCh) }) + h.eg.Go(func() error { return h.bigTeamUnboxLoop(stopCh) }) } func (h *UIInboxLoader) Stop(ctx context.Context) chan struct{} { defer h.Trace(ctx, nil, "Stop")() h.Lock() defer h.Unlock() + return h.doStopLocked(ctx) +} + +func (h *UIInboxLoader) doStopLocked(ctx context.Context) chan struct{} { ch := make(chan struct{}) if h.started { close(h.stopCh) h.started = false + h.uid = nil + h.convTransmitBatch = make(map[chat1.ConvIDStr]chat1.ConversationLocal) go func() { err := h.eg.Wait() if err != nil { @@ -108,6 +118,7 @@ func (h *UIInboxLoader) Stop(ctx context.Context) chan struct{} { } else { close(ch) } + h.setLastLayout(nil) return ch } @@ -658,6 +669,7 @@ func (h *UIInboxLoader) prepareShareConversations(ctx context.Context, widgetLis // OnLogout clears donated share intents on logout so the next user does not see the previous user's suggestions. func (h *UIInboxLoader) OnLogout(mctx libkb.MetaContext) error { + <-h.Stop(mctx.Ctx()) if h.G().ShareIntentDonator != nil { h.G().ShareIntentDonator.DeleteAllDonations() }