wait for Stop and clear uid when restarting reused modules

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")

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 {

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() {

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)

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()

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)

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

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)

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

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 {

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()
 	}