From 4a30614cd062930d9c641ba1907e818c08b28b27 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Thu, 19 Mar 2026 20:11:21 +0900 Subject: [PATCH 1/2] fix: add HTTP timeout to Telegram long polling and reconnection indicator The Telegram HTTP client had no Timeout set, causing requests to hang indefinitely when TCP connections died silently. This adds a 65s timeout (30s long poll + 35s margin) to all client configurations. Also adds: - resilientTransport: detects polling failures and sends a recovery notice to the last active chat - MessageDeduplicator: moved from wecom to shared channels package, now also used by Telegram to guard against update redelivery Co-Authored-By: Claude Opus 4.6 (1M context) --- pkg/channels/{wecom => }/dedupe.go | 6 +- pkg/channels/{wecom => }/dedupe_test.go | 6 +- pkg/channels/telegram/telegram.go | 86 +++++++++++++++---- .../telegram/telegram_dispatch_test.go | 1 + .../telegram_group_command_filter_test.go | 1 + pkg/channels/telegram/telegram_test.go | 3 + pkg/channels/telegram/transport.go | 27 ++++++ pkg/channels/wecom/app.go | 4 +- pkg/channels/wecom/bot.go | 4 +- 9 files changed, 111 insertions(+), 27 deletions(-) rename pkg/channels/{wecom => }/dedupe.go (92%) rename pkg/channels/{wecom => }/dedupe_test.go (93%) create mode 100644 pkg/channels/telegram/transport.go diff --git a/pkg/channels/wecom/dedupe.go b/pkg/channels/dedupe.go similarity index 92% rename from pkg/channels/wecom/dedupe.go rename to pkg/channels/dedupe.go index 865be668e..fd4e53237 100644 --- a/pkg/channels/wecom/dedupe.go +++ b/pkg/channels/dedupe.go @@ -1,8 +1,8 @@ -package wecom +package channels import "sync" -const wecomMaxProcessedMessages = 1000 +const defaultMaxProcessedMessages = 1000 // MessageDeduplicator provides thread-safe message deduplication using a circular queue (ring buffer) // combined with a hash map. This ensures fast O(1) lookups while naturally evicting the oldest @@ -18,7 +18,7 @@ type MessageDeduplicator struct { // NewMessageDeduplicator creates a new deduplicator with the specified capacity. func NewMessageDeduplicator(maxEntries int) *MessageDeduplicator { if maxEntries <= 0 { - maxEntries = wecomMaxProcessedMessages + maxEntries = defaultMaxProcessedMessages } return &MessageDeduplicator{ msgs: make(map[string]bool, maxEntries), diff --git a/pkg/channels/wecom/dedupe_test.go b/pkg/channels/dedupe_test.go similarity index 93% rename from pkg/channels/wecom/dedupe_test.go rename to pkg/channels/dedupe_test.go index 10dff4cfe..e69e7e0ba 100644 --- a/pkg/channels/wecom/dedupe_test.go +++ b/pkg/channels/dedupe_test.go @@ -1,4 +1,4 @@ -package wecom +package channels import ( "sync" @@ -6,7 +6,7 @@ import ( ) func TestMessageDeduplicator_DuplicateDetection(t *testing.T) { - d := NewMessageDeduplicator(wecomMaxProcessedMessages) + d := NewMessageDeduplicator(defaultMaxProcessedMessages) if ok := d.MarkMessageProcessed("msg-1"); !ok { t.Fatalf("first message should be accepted") @@ -18,7 +18,7 @@ func TestMessageDeduplicator_DuplicateDetection(t *testing.T) { } func TestMessageDeduplicator_ConcurrentSameMessage(t *testing.T) { - d := NewMessageDeduplicator(wecomMaxProcessedMessages) + d := NewMessageDeduplicator(defaultMaxProcessedMessages) const goroutines = 64 var wg sync.WaitGroup diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 6ce7e4a97..e7f3e2603 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -26,15 +26,23 @@ import ( "github.com/sipeed/picoclaw/pkg/utils" ) +// telegramHTTPTimeout is the HTTP client timeout for Telegram API requests. +// Long polling uses Timeout=30s on the API side; the HTTP client timeout +// must be longer to avoid cancelling valid long-poll responses. +const telegramHTTPTimeout = 65 * time.Second + type TelegramChannel struct { *channels.BaseChannel bot *telego.Bot bh *th.BotHandler config *config.Config chatIDs map[string]int64 + dedupe *channels.MessageDeduplicator ctx context.Context cancel context.CancelFunc + lastActiveChatID string // composite chat ID of last received message + registerFunc func(context.Context, []commands.Definition) error commandRegCancel context.CancelFunc } @@ -43,25 +51,47 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann var opts []telego.BotOption telegramCfg := cfg.Channels.Telegram + // Build the base transport (with optional proxy) and wrap it with + // resilientTransport for connection-failure detection and recovery logging. + var baseTransport http.RoundTripper if telegramCfg.Proxy != "" { proxyURL, parseErr := url.Parse(telegramCfg.Proxy) if parseErr != nil { return nil, fmt.Errorf("invalid proxy URL %q: %w", telegramCfg.Proxy, parseErr) } - opts = append(opts, telego.WithHTTPClient(&http.Client{ - Transport: &http.Transport{ - Proxy: http.ProxyURL(proxyURL), - }, - })) + baseTransport = &http.Transport{Proxy: http.ProxyURL(proxyURL)} } else if os.Getenv("HTTP_PROXY") != "" || os.Getenv("HTTPS_PROXY") != "" { - // Use environment proxy if configured - opts = append(opts, telego.WithHTTPClient(&http.Client{ - Transport: &http.Transport{ - Proxy: http.ProxyFromEnvironment, - }, - })) + baseTransport = &http.Transport{Proxy: http.ProxyFromEnvironment} + } else { + baseTransport = http.DefaultTransport } + ch := &TelegramChannel{ + config: cfg, + chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), + } + + transport := &resilientTransport{ + base: baseTransport, + onFailure: func() { + logger.WarnC("telegram", "Polling connection lost, retrying...") + }, + onRecover: func() { + logger.InfoC("telegram", "Polling connection recovered") + if chatID := ch.lastActiveChatID; chatID != "" { + go func() { + _ = ch.sendReconnectNotice(chatID) + }() + } + }, + } + + opts = append(opts, telego.WithHTTPClient(&http.Client{ + Transport: transport, + Timeout: telegramHTTPTimeout, + })) + if baseURL := strings.TrimRight(strings.TrimSpace(telegramCfg.BaseURL), "/"); baseURL != "" { opts = append(opts, telego.WithAPIServer(baseURL)) } @@ -82,12 +112,10 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann channels.WithReasoningChannelID(telegramCfg.ReasoningChannelID), ) - return &TelegramChannel{ - BaseChannel: base, - bot: bot, - config: cfg, - chatIDs: make(map[string]int64), - }, nil + ch.BaseChannel = base + ch.bot = bot + + return ch, nil } func (c *TelegramChannel) Start(ctx context.Context) error { @@ -505,6 +533,15 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes return nil } + // Deduplicate: Telegram may redeliver the same update after a timeout. + msgID := fmt.Sprintf("%d", message.MessageID) + if !c.dedupe.MarkMessageProcessed(msgID) { + logger.DebugCF("telegram", "Skipping duplicate message", map[string]any{ + "message_id": msgID, + }) + return nil + } + chatID := message.Chat.ID c.chatIDs[platformID] = chatID @@ -617,6 +654,8 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) } + c.lastActiveChatID = compositeChatID + logger.DebugCF("telegram", "Received message", map[string]any{ "sender_id": sender.CanonicalID, "chat_id": compositeChatID, @@ -818,6 +857,19 @@ func isBotCommandEntityForThisBot(entityText, botUsername string) bool { return strings.EqualFold(mentionUsername, botUsername) } +// sendReconnectNotice sends a short notification to the given chat after +// the polling connection recovers from a failure. +func (c *TelegramChannel) sendReconnectNotice(compositeChatID string) error { + cid, threadID, err := parseTelegramChatID(compositeChatID) + if err != nil { + return err + } + msg := tu.Message(tu.ID(cid), "[system] Connection recovered — messages during the outage may have been delayed.") + msg.MessageThreadID = threadID + _, err = c.bot.SendMessage(c.ctx, msg) + return err +} + // stripBotMention removes the @bot mention from the content. func (c *TelegramChannel) stripBotMention(content string) string { botUsername := c.bot.Username() diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 0eb1de5ea..9b35cdc3a 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -15,6 +15,7 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { ch := &TelegramChannel{ BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), ctx: context.Background(), } diff --git a/pkg/channels/telegram/telegram_group_command_filter_test.go b/pkg/channels/telegram/telegram_group_command_filter_test.go index 614b2ca7f..ff107d474 100644 --- a/pkg/channels/telegram/telegram_group_command_filter_test.go +++ b/pkg/channels/telegram/telegram_group_command_filter_test.go @@ -51,6 +51,7 @@ func newGroupMentionOnlyChannel(t *testing.T, botUsername string) (*TelegramChan ), bot: newTestTelegramBot(t, botUsername), chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), ctx: context.Background(), } return ch, messageBus diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 4b596f45f..b8b7f20dc 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -524,6 +524,7 @@ func TestHandleMessage_ForumTopic_SetsMetadata(t *testing.T) { ch := &TelegramChannel{ BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), ctx: context.Background(), } @@ -565,6 +566,7 @@ func TestHandleMessage_NoForum_NoThreadMetadata(t *testing.T) { ch := &TelegramChannel{ BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), ctx: context.Background(), } @@ -604,6 +606,7 @@ func TestHandleMessage_ReplyThread_NonForum_NoIsolation(t *testing.T) { ch := &TelegramChannel{ BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), chatIDs: make(map[string]int64), + dedupe: channels.NewMessageDeduplicator(1000), ctx: context.Background(), } diff --git a/pkg/channels/telegram/transport.go b/pkg/channels/telegram/transport.go new file mode 100644 index 000000000..39206ed25 --- /dev/null +++ b/pkg/channels/telegram/transport.go @@ -0,0 +1,27 @@ +package telegram + +import ( + "net/http" + "sync/atomic" +) + +// resilientTransport wraps http.RoundTripper to detect polling failures +// and notify on recovery. +type resilientTransport struct { + base http.RoundTripper + onFailure func() // called once when first failure detected + onRecover func() // called once when connection recovers after failure + failed atomic.Bool +} + +func (t *resilientTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil { + if !t.failed.Swap(true) && t.onFailure != nil { + t.onFailure() + } + } else if t.failed.Swap(false) && t.onRecover != nil { + t.onRecover() + } + return resp, err +} diff --git a/pkg/channels/wecom/app.go b/pkg/channels/wecom/app.go index 2098fcd4e..61c94259f 100644 --- a/pkg/channels/wecom/app.go +++ b/pkg/channels/wecom/app.go @@ -38,7 +38,7 @@ type WeComAppChannel struct { tokenMu sync.RWMutex ctx context.Context cancel context.CancelFunc - processedMsgs *MessageDeduplicator + processedMsgs *channels.MessageDeduplicator } // WeComXMLMessage represents the XML message structure from WeCom @@ -143,7 +143,7 @@ func NewWeComAppChannel(cfg config.WeComAppConfig, messageBus *bus.MessageBus) ( client: &http.Client{Timeout: clientTimeout}, ctx: ctx, cancel: cancel, - processedMsgs: NewMessageDeduplicator(wecomMaxProcessedMessages), + processedMsgs: channels.NewMessageDeduplicator(1000), }, nil } diff --git a/pkg/channels/wecom/bot.go b/pkg/channels/wecom/bot.go index 96d5a961f..77f0a11ce 100644 --- a/pkg/channels/wecom/bot.go +++ b/pkg/channels/wecom/bot.go @@ -27,7 +27,7 @@ type WeComBotChannel struct { client *http.Client ctx context.Context cancel context.CancelFunc - processedMsgs *MessageDeduplicator + processedMsgs *channels.MessageDeduplicator } // WeComBotMessage represents the JSON message structure from WeCom Bot (AIBOT) @@ -106,7 +106,7 @@ func NewWeComBotChannel(cfg config.WeComConfig, messageBus *bus.MessageBus) (*We client: &http.Client{Timeout: clientTimeout}, ctx: ctx, cancel: cancel, - processedMsgs: NewMessageDeduplicator(wecomMaxProcessedMessages), + processedMsgs: channels.NewMessageDeduplicator(1000), }, nil } From b0e737e9834f8692ae559473851bd461cad6c99b Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Thu, 19 Mar 2026 20:14:39 +0900 Subject: [PATCH 2/2] fix: correct misspelling flagged by linter Co-Authored-By: Claude Opus 4.6 (1M context) --- pkg/channels/telegram/telegram.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index e7f3e2603..d666b4b03 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -28,7 +28,7 @@ import ( // telegramHTTPTimeout is the HTTP client timeout for Telegram API requests. // Long polling uses Timeout=30s on the API side; the HTTP client timeout -// must be longer to avoid cancelling valid long-poll responses. +// must be longer to avoid canceling valid long-poll responses. const telegramHTTPTimeout = 65 * time.Second type TelegramChannel struct {