Method sweepIdleStreams

coderd/x/chatd/chatd.go:4782–4808

sweepIdleStreams iterates chatStreams once and delegates each entry to cleanupStreamIfIdle. Range may skip entries that become reapable concurrently. Any such entry is reaped on the next tick.

()

// to cleanupStreamIfIdle. Range may skip entries that become reapable
// concurrently. Any such entry is reaped on the next tick.
func (p *Server) sweepIdleStreams() {
	var reaped atomic.Int64
	defer func() {
		if count := reaped.Load(); count > 0 {
			p.logger.Info(context.Background(), "reaped idle chat streams", slog.F("count", count))
		}
	}()
	p.chatStreams.Range(func(key, value any) bool {
		chatID, ok := key.(uuid.UUID)
		if !ok {
			return true
		}
		state, ok := value.(*chatStreamState)
		if !ok {
			return true
		}
		// guard against any panic from cleanupStreamIfIdle locking state.mu for all time
		func() {
			state.mu.Lock()
			defer state.mu.Unlock()
			if p.cleanupStreamIfIdle(chatID, state) {
				reaped.Add(1)
			}
		}()
		return true
	})
}

// registerHeartbeat enrolls a chat in the centralized batch
// heartbeat loop. Must be called after chatCtx is created.