fix(telegram): prevent typing indicator goroutine leaks

This fix addresses memory/goroutine leaks in the Telegram channel's typing
indicator implementation:

- Add typingStop map to track active typing goroutines per chatID
- Stop existing typing goroutine before starting new one (prevents duplicates)
- Add 5-minute hard timeout to prevent indefinite goroutine leaks
- Cleanup all typing goroutines on channel Stop()
- Add idempotent stopTyping() helper method

The implementation now follows the same proven pattern used in Discord channel,
ensuring consistency across the codebase.

Fixes: constant typing indicator when bot is idle
This commit is contained in:
Vishnuvardhan Reddy 2026-03-01 08:39:08 +00:00
parent 373be004e3
commit 9bfca13abd

View file

@ -9,6 +9,7 @@ import (
"regexp" "regexp"
"strconv" "strconv"
"strings" "strings"
"sync"
"time" "time"
"github.com/mymmrac/telego" "github.com/mymmrac/telego"
@ -48,6 +49,8 @@ type TelegramChannel struct {
usernameCache map[string]int64 // Cache username -> chat ID resolution usernameCache map[string]int64 // Cache username -> chat ID resolution
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
typingMu sync.Mutex
typingStop map[string]chan struct{} // chatID -> stop signal for typing goroutines
} }
func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChannel, error) { func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChannel, error) {
@ -95,6 +98,7 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann
config: cfg, config: cfg,
chatIDs: make(map[string]int64), chatIDs: make(map[string]int64),
usernameCache: make(map[string]int64), usernameCache: make(map[string]int64),
typingStop: make(map[string]chan struct{}),
}, nil }, nil
} }
@ -152,6 +156,14 @@ func (c *TelegramChannel) Stop(ctx context.Context) error {
logger.InfoC("telegram", "Stopping Telegram bot...") logger.InfoC("telegram", "Stopping Telegram bot...")
c.SetRunning(false) c.SetRunning(false)
// Stop all active typing goroutines
c.typingMu.Lock()
for chatID, stop := range c.typingStop {
close(stop)
delete(c.typingStop, chatID)
}
c.typingMu.Unlock()
// Stop the bot handler // Stop the bot handler
if c.bh != nil { if c.bh != nil {
c.bh.Stop() c.bh.Stop()
@ -198,30 +210,61 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
// It sends ChatAction(typing) immediately and then repeats every 4 seconds // It sends ChatAction(typing) immediately and then repeats every 4 seconds
// (Telegram's typing indicator expires after ~5s) in a background goroutine. // (Telegram's typing indicator expires after ~5s) in a background goroutine.
// The returned stop function is idempotent and cancels the goroutine. // The returned stop function is idempotent and cancels the goroutine.
// If a typing indicator already exists for this chatID, it is stopped first.
func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
cid, err := parseChatID(chatID) cid, err := parseChatID(chatID)
if err != nil { if err != nil {
return func() {}, err return func() {}, err
} }
c.typingMu.Lock()
// Stop existing typing loop for this chatID if any
if stop, ok := c.typingStop[chatID]; ok {
close(stop)
}
stop := make(chan struct{})
c.typingStop[chatID] = stop
c.typingMu.Unlock()
// Send the first typing action immediately // Send the first typing action immediately
_ = c.bot.SendChatAction(ctx, tu.ChatAction(cid, telego.ChatActionTyping)) _ = c.bot.SendChatAction(ctx, tu.ChatAction(cid, telego.ChatActionTyping))
typingCtx, cancel := context.WithCancel(ctx)
go func() { go func() {
ticker := time.NewTicker(4 * time.Second) ticker := time.NewTicker(4 * time.Second)
defer ticker.Stop() defer ticker.Stop()
// Hard timeout to prevent goroutine leaks (matching Discord's approach)
timeout := time.After(5 * time.Minute)
for { for {
select { select {
case <-typingCtx.Done(): case <-stop:
return
case <-timeout:
c.typingMu.Lock()
delete(c.typingStop, chatID)
c.typingMu.Unlock()
return
case <-c.ctx.Done():
c.typingMu.Lock()
delete(c.typingStop, chatID)
c.typingMu.Unlock()
return return
case <-ticker.C: case <-ticker.C:
_ = c.bot.SendChatAction(typingCtx, tu.ChatAction(cid, telego.ChatActionTyping)) _ = c.bot.SendChatAction(ctx, tu.ChatAction(cid, telego.ChatActionTyping))
} }
} }
}() }()
return cancel, nil return func() { c.stopTyping(chatID) }, nil
}
// stopTyping stops the typing indicator loop for the given chatID.
func (c *TelegramChannel) stopTyping(chatID string) {
c.typingMu.Lock()
defer c.typingMu.Unlock()
if stop, ok := c.typingStop[chatID]; ok {
close(stop)
delete(c.typingStop, chatID)
}
} }
// EditMessage implements channels.MessageEditor. // EditMessage implements channels.MessageEditor.