mirror of
https://github.com/chenhg5/cc-connect.git
synced 2026-10-26 01:50:40 +08:00
fix(weixin): treat sendmessage ret=-2 as a per-account send budget and fail fast (#1643)
* fix(weixin): treat sendmessage ret=-2 as a burst throttle, not a token issue Live testing showed ilink accepts any (or no) context_token on sendmessage; ret=-2 "prepare failed" is a bot-wide burst throttle that lasts on the order of an hour and is not cleared by restarting the daemon or re-binding getupdates. Rapid 500ms-interval retries during the throttle only add more refused sends (and may extend the penalty window). Replace the token-refresh retry logic with throttle-aware handling: - detect ret=-2 as a throttle, back off before a single retry, then fail with a clear error so the caller can retry the whole message later - skip the "incomplete delivery notice" send when the failure is a throttle - fix misleading "refresh the session token" log/error wording Applies to both the text (sendChunkWithRetry) and media (sendSingleItemWithRetry) send paths. Adds unit tests simulating throttled-then-success and persistent throttle. Closes #1640 * docs(config): document weixin outgoing rate limit to avoid ilink burst throttle ilink sendmessage throttles the bot after a short burst of rapid sends (ret=-2 "prepare failed"). Add the weixin per-platform override example so deployments can pace sends and avoid triggering the throttle. * fix(weixin): fail fast on sendmessage throttle instead of backing off Live testing showed the ilink sendmessage penalty (ret=-2 "prepare failed") is escalated by every send attempt made while it is active: the original incident recovered in ~1h with no attempts, while probing attempts kept the bot throttled for 7h+. Retrying (even with a backoff) only prolongs the outage. Change the text and media send paths to fail fast on ret=-2 with a clear throttled error, issuing a single sendmessage call and skipping the extra "incomplete delivery notice" send. Remove the now-unused retry constants. * feat(weixin): add send-volume window quota to avoid ilink burst throttle Live testing showed ilink's sendmessage throttle (ret=-2 "prepare failed") is triggered by roughly 5-6 separate messages within a short window (not by chunk count: multi-chunk sends are fine), and that rate-based pacing alone does not prevent it. Add a per-platform sliding-window quota (burst_limit / burst_window_secs, default 4 per 120s) that paces separate messages so the bot stays below the trigger. fail-fast on ret=-2 already prevents the penalty from escalating. * docs(config): document weixin burst quota options (burst_limit/burst_window_secs) * fix(weixin): make send quota a 24h budget that fails fast instead of pacing Live testing showed ilink's sendmessage throttle is a per-account message budget (~5-6 sends per 24h window), not a rate/burst limit: sends spread over hours still trip it at ~5-6 total, and pacing (1/s) did not prevent it. Change the quota from a short-window pacing mechanism to a 24h budget that fails fast (no waiting — waiting up to a day is useless) once exhausted, so the bot never hammers a throttled API. Configurable via burst_limit / burst_window_secs (default 4 per 86400s).
This commit is contained in:
1 parent
a79e981670
commit
9fe3ea261e
5 files changed
+304
-133
No files matched your search
@@ -291,6 +291,11 @@ level = "info" # debug, info, warn, error
|
||||
# max_per_second = 25
|
||||
# [outgoing_rate_limit.platforms.feishu]
|
||||
# max_per_second = 5
|
||||
# Weixin (ilink) sendmessage throttles the bot after a short burst (ret=-2
|
||||
# "prepare failed", penalty on the order of an hour). Pace sends to avoid it:
|
||||
# [outgoing_rate_limit.platforms.weixin]
|
||||
# max_per_second = 0.5
|
||||
# burst = 1
|
||||
|
||||
# =============================================================================
|
||||
# Relay Settings / Relay 设置
|
||||
@@ -1468,6 +1473,17 @@ app_secret = "your-feishu-app-secret"
|
||||
# proxy = "" # optional HTTP proxy / 可选代理
|
||||
# proxy_username = ""
|
||||
# proxy_password = ""
|
||||
#
|
||||
# Send-volume quota vs ilink's burst throttle (sendMessage ret=-2 "prepare failed"):
|
||||
# the gateway throttles the bot after roughly 5-6 separate messages within a short
|
||||
# window (multi-chunk sends are fine), and the penalty escalates with every attempt
|
||||
# made while active. The quota paces separate messages so the bot stays below the
|
||||
# trigger. 0 disables the quota.
|
||||
# 发送量配额(防 ilink 突发节流 ret=-2):网关在短窗口内约 5-6 条独立消息即节流
|
||||
# (多切块发送不计数),且节流期内每次尝试都会升级惩罚。配额对独立消息限速,
|
||||
# 使其低于触发阈值。设为 0 关闭配额。
|
||||
# burst_limit = 4 # max separate messages per window / 每窗口最多独立消息数
|
||||
# burst_window_secs = 86400 # window length (default 24h) / 窗口时长(秒,默认24小时)
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Tencent Yuanbao (yuanbao.tencent.com bot platform)
|
||||
|
||||
+1
-1
@@ -11718,7 +11718,7 @@ func (e *Engine) sendAlreadyRenderedWithError(p Platform, replyCtx any, content
|
||||
"platform", p.Name(),
|
||||
"error", err,
|
||||
"content_len", len(content),
|
||||
"hint", "user needs to send a new message to refresh context_token")
|
||||
"hint", "user needs to send a message to the bot first so a context_token can be captured")
|
||||
} else {
|
||||
slog.Error("platform send failed", "platform", p.Name(), "error", err, "content_len", len(content))
|
||||
}
|
||||
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/chenhg5/cc-connect/core"
|
||||
)
|
||||
@@ -109,8 +108,34 @@ func (p *Platform) uploadToWeixinCDN(ctx context.Context, to string, plaintext [
|
||||
}, nil
|
||||
}
|
||||
|
||||
// sendSingleItem sends a media item. If ilink throttles the send (ret=-2
|
||||
// "prepare failed"), it fails fast instead of retrying: the penalty is escalated
|
||||
// by every send attempt made while it is active, so retrying only prolongs the
|
||||
// outage.
|
||||
func (p *Platform) sendSingleItem(ctx context.Context, rc *replyContext, item messageItem) error {
|
||||
return p.sendSingleItemWithRetry(ctx, rc, item)
|
||||
if err := p.checkSendQuota(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
msg := sendMessageReq{
|
||||
Msg: weixinOutboundMsg{
|
||||
FromUserID: "",
|
||||
ToUserID: rc.peerUserID,
|
||||
ClientID: "cc-" + randomHex(8),
|
||||
MessageType: messageTypeBot,
|
||||
MessageState: messageStateFinish,
|
||||
ItemList: []messageItem{item},
|
||||
ContextToken: rc.contextToken,
|
||||
},
|
||||
}
|
||||
err := p.api.sendMessage(ctx, &msg)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if isSendThrottled(err) {
|
||||
return fmt.Errorf("weixin: sendMessage throttled by ilink (ret=-2); "+
|
||||
"the bot is rate-limited and sending during the penalty escalates it, retry the message later: %w", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func mediaFromUploadRef(ref *cdnUploadedRef) *cdnMedia {
|
||||
@@ -131,54 +156,6 @@ func buildVideoMessageItem(ref *cdnUploadedRef) messageItem {
|
||||
}
|
||||
}
|
||||
|
||||
// sendSingleItemWithRetry sends a media item with retry mechanism for ret=-2 errors.
|
||||
func (p *Platform) sendSingleItemWithRetry(ctx context.Context, rc *replyContext, item messageItem) error {
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < weixinSendMaxRetries; attempt++ {
|
||||
msg := sendMessageReq{
|
||||
Msg: weixinOutboundMsg{
|
||||
FromUserID: "",
|
||||
ToUserID: rc.peerUserID,
|
||||
ClientID: "cc-" + randomHex(8),
|
||||
MessageType: messageTypeBot,
|
||||
MessageState: messageStateFinish,
|
||||
ItemList: []messageItem{item},
|
||||
ContextToken: rc.contextToken,
|
||||
},
|
||||
}
|
||||
err := p.api.sendMessage(ctx, &msg)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
lastErr = err
|
||||
// Check if error is ret=-2 (API declined) - attempt token refresh
|
||||
if strings.Contains(err.Error(), "ret=-2") {
|
||||
freshToken := p.getContextToken(rc.peerUserID)
|
||||
if freshToken == "" || freshToken == rc.contextToken {
|
||||
slog.Warn("weixin: sendMessage ret=-2 for media, no fresh context_token — "+
|
||||
"user must send a new message to refresh session token",
|
||||
"attempt", attempt+1, "peer", rc.peerUserID)
|
||||
return fmt.Errorf("weixin: sendMessage ret=-2 (expired context_token); "+
|
||||
"user must send a new message to peer %q to refresh the session token: %w",
|
||||
rc.peerUserID, lastErr)
|
||||
}
|
||||
slog.Warn("weixin: sendMessage ret=-2 for media, retrying with fresh context_token",
|
||||
"attempt", attempt+1, "peer", rc.peerUserID)
|
||||
rc.contextToken = freshToken
|
||||
slog.Debug("weixin: using refreshed context_token for media retry", "peer", rc.peerUserID)
|
||||
// Brief delay before retry
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(weixinSendRetryDelay):
|
||||
}
|
||||
continue
|
||||
}
|
||||
// For other errors, don't retry
|
||||
return err
|
||||
}
|
||||
return lastErr
|
||||
}
|
||||
|
||||
// SendImage implements core.ImageSender.
|
||||
func (p *Platform) SendImage(ctx context.Context, replyCtx any, img core.ImageAttachment) error {
|
||||
|
||||
+119
-82
@@ -27,12 +27,18 @@ const (
|
||||
sessionKeyPrefix = "weixin:dm:"
|
||||
maxWeixinChunk = 3800 // stay under typical IM limits
|
||||
|
||||
// weixinSendMaxRetries is the maximum number of retries for sendMessage when API returns ret=-2.
|
||||
weixinSendMaxRetries = 3
|
||||
// weixinSendRetryDelay is the delay between retries when sendMessage fails.
|
||||
weixinSendRetryDelay = 500 * time.Millisecond
|
||||
// weixinChunkSendDelay is the delay between sending message chunks to avoid rate limiting.
|
||||
weixinChunkSendDelay = 100 * time.Millisecond
|
||||
|
||||
// Send-volume quota that keeps the bot under ilink's burst throttle
|
||||
// (sendMessage ret=-2 "prepare failed"). Live testing showed the gateway
|
||||
// throttles the bot after roughly 5-6 separate messages within a short window,
|
||||
// and that the penalty is escalated by every send attempt made while it is
|
||||
// active. We pace separate messages (not chunks: multi-chunk sends are fine)
|
||||
// to stay well below the trigger. Configurable via burst_limit /
|
||||
// burst_window_secs platform options.
|
||||
defaultBurstLimit = 4 // max separate messages per window
|
||||
defaultBurstWindowSecs = 86400 // window length (24h: ilink budgets ~5-6 sends/day)
|
||||
// typingTicketTTL is how long a cached typing ticket remains valid.
|
||||
typingTicketTTL = 10 * time.Minute
|
||||
// typingRepeatInterval is how often to resend the typing status to keep it alive.
|
||||
@@ -87,6 +93,12 @@ type Platform struct {
|
||||
|
||||
typingMu sync.RWMutex
|
||||
typingTickets map[string]typingTicketEntry // peerUserID → cached ticket
|
||||
|
||||
// Send-volume quota guarding against ilink's burst throttle (see constants).
|
||||
sendQuotaMu sync.Mutex
|
||||
sendQuotaTimes []time.Time
|
||||
sendQuotaLimit int
|
||||
sendQuotaWindow time.Duration
|
||||
}
|
||||
|
||||
type typingTicketEntry struct {
|
||||
@@ -135,6 +147,16 @@ func New(opts map[string]any) (core.Platform, error) {
|
||||
}
|
||||
lp := pickInt(opts["long_poll_timeout_ms"])
|
||||
|
||||
// Send-volume quota (see defaultBurstLimit constants). 0 disables the quota.
|
||||
burstLimit := pickInt(opts["burst_limit"])
|
||||
if burstLimit < 0 {
|
||||
burstLimit = 0
|
||||
}
|
||||
burstWindow := pickInt(opts["burst_window_secs"])
|
||||
if burstWindow < 0 {
|
||||
burstWindow = 0
|
||||
}
|
||||
|
||||
dataDir, _ := opts["cc_data_dir"].(string)
|
||||
project, _ := opts["cc_project"].(string)
|
||||
stateDir := ""
|
||||
@@ -167,20 +189,29 @@ func New(opts map[string]any) (core.Platform, error) {
|
||||
Transport: &http.Transport{Proxy: nil},
|
||||
}
|
||||
|
||||
if burstLimit <= 0 {
|
||||
burstLimit = defaultBurstLimit
|
||||
}
|
||||
if burstWindow <= 0 {
|
||||
burstWindow = defaultBurstWindowSecs
|
||||
}
|
||||
|
||||
p := &Platform{
|
||||
token: token,
|
||||
baseURL: baseURL,
|
||||
cdnBaseURL: cdnBaseURL,
|
||||
allowFrom: allowFrom,
|
||||
routeTag: routeTag,
|
||||
stateDir: stateDir,
|
||||
longPollMS: lp,
|
||||
accountLabel: accountLabel,
|
||||
httpClient: httpClient,
|
||||
cdnHttpClient: cdnHttpClient,
|
||||
tokens: make(map[string]string),
|
||||
dedup: make(map[string]time.Time),
|
||||
typingTickets: make(map[string]typingTicketEntry),
|
||||
token: token,
|
||||
baseURL: baseURL,
|
||||
cdnBaseURL: cdnBaseURL,
|
||||
allowFrom: allowFrom,
|
||||
routeTag: routeTag,
|
||||
stateDir: stateDir,
|
||||
longPollMS: lp,
|
||||
accountLabel: accountLabel,
|
||||
httpClient: httpClient,
|
||||
cdnHttpClient: cdnHttpClient,
|
||||
tokens: make(map[string]string),
|
||||
dedup: make(map[string]time.Time),
|
||||
typingTickets: make(map[string]typingTicketEntry),
|
||||
sendQuotaLimit: burstLimit,
|
||||
sendQuotaWindow: time.Duration(burstWindow) * time.Second,
|
||||
}
|
||||
p.api = newAPIClient(baseURL, token, routeTag, httpClient)
|
||||
|
||||
@@ -639,11 +670,49 @@ func (p *Platform) refreshTypingTicket(ctx context.Context, peerID, contextToken
|
||||
}()
|
||||
}
|
||||
|
||||
// checkSendQuota enforces the bot's separate-message budget so ilink's
|
||||
// sendmessage throttle (ret=-2 "prepare failed") is not triggered. Live testing
|
||||
// showed the gateway throttles a bot after roughly 5-6 separate messages per
|
||||
// long window (about a day; matches the 24h context TTL), regardless of pacing,
|
||||
// and that attempts made during the penalty escalate it. Multi-chunk sends do
|
||||
// not count (a chunked message is one logical message). This quota counts
|
||||
// logical messages in a sliding window and FAILS FAST once the budget is
|
||||
// exhausted — waiting for the window to slide (up to a day) is useless, and the
|
||||
// fail-fast philosophy (see sendChunk) applies: do not keep hammering a
|
||||
// throttled bot. Configure via burst_limit / burst_window_secs platform
|
||||
// options. A limit of 0 disables the quota.
|
||||
func (p *Platform) checkSendQuota(ctx context.Context) error {
|
||||
if p.sendQuotaLimit <= 0 || p.sendQuotaWindow <= 0 {
|
||||
return nil
|
||||
}
|
||||
p.sendQuotaMu.Lock()
|
||||
now := time.Now()
|
||||
cutoff := now.Add(-p.sendQuotaWindow)
|
||||
kept := p.sendQuotaTimes[:0]
|
||||
for _, t := range p.sendQuotaTimes {
|
||||
if t.After(cutoff) {
|
||||
kept = append(kept, t)
|
||||
}
|
||||
}
|
||||
p.sendQuotaTimes = kept
|
||||
if len(p.sendQuotaTimes) >= p.sendQuotaLimit {
|
||||
p.sendQuotaMu.Unlock()
|
||||
return fmt.Errorf("weixin: send budget exhausted (%d messages in the last %s); "+
|
||||
"ilink throttles the bot after roughly 5-6 sends per window — reduce messages or re-login later", p.sendQuotaLimit, p.sendQuotaWindow)
|
||||
}
|
||||
p.sendQuotaTimes = append(p.sendQuotaTimes, now)
|
||||
p.sendQuotaMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Platform) sendChunks(ctx context.Context, replyCtx any, content string) error {
|
||||
rc, ok := replyCtx.(*replyContext)
|
||||
if !ok || rc == nil {
|
||||
return fmt.Errorf("weixin: invalid reply context")
|
||||
}
|
||||
if err := p.checkSendQuota(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if strings.TrimSpace(rc.contextToken) == "" {
|
||||
rc.contextToken = p.getContextToken(rc.peerUserID)
|
||||
}
|
||||
@@ -651,8 +720,8 @@ func (p *Platform) sendChunks(ctx context.Context, replyCtx any, content string)
|
||||
slog.Error("weixin: cannot send message - missing context_token",
|
||||
"peer", rc.peerUserID,
|
||||
"content_preview", truncatePreview(content, 100),
|
||||
"hint", "user needs to send a new message to refresh context_token")
|
||||
return fmt.Errorf("weixin: missing context_token for peer %q - user must send a new message first", rc.peerUserID)
|
||||
"hint", "user needs to send a message to the bot first so a context_token can be captured")
|
||||
return fmt.Errorf("weixin: missing context_token for peer %q - user must send a message to the bot first", rc.peerUserID)
|
||||
}
|
||||
if strings.TrimSpace(content) == "" {
|
||||
return nil
|
||||
@@ -668,19 +737,21 @@ func (p *Platform) sendChunks(ctx context.Context, replyCtx any, content string)
|
||||
case <-time.After(weixinChunkSendDelay):
|
||||
}
|
||||
}
|
||||
// Retry sendText with context_token refresh on failure
|
||||
err := p.sendChunkWithRetry(ctx, rc, chunk, i+1, total)
|
||||
err := p.sendChunk(ctx, rc, chunk)
|
||||
if err != nil {
|
||||
slog.Error("weixin: chunk send failed, message incomplete",
|
||||
"peer", rc.peerUserID,
|
||||
"failed_chunk", fmt.Sprintf("%d/%d", i+1, total),
|
||||
"error", err)
|
||||
// Notify user that message delivery was incomplete.
|
||||
// Use a short message that is unlikely to fail itself.
|
||||
notice := "⚠️ 消息发送不完整,请在终端查看完整结果。"
|
||||
noticeID := "cc-" + randomHex(6)
|
||||
if nerr := p.api.sendText(ctx, rc.peerUserID, notice, rc.contextToken, noticeID); nerr != nil {
|
||||
slog.Warn("weixin: failed to send incomplete-delivery notice", "peer", rc.peerUserID, "error", nerr)
|
||||
// Notify user that message delivery was incomplete, unless the failure
|
||||
// is the ilink throttle: the notice send would be refused too, only
|
||||
// adding another throttled request.
|
||||
if !isSendThrottled(err) {
|
||||
notice := "⚠️ 消息发送不完整,请在终端查看完整结果。"
|
||||
noticeID := "cc-" + randomHex(6)
|
||||
if nerr := p.api.sendText(ctx, rc.peerUserID, notice, rc.contextToken, noticeID); nerr != nil {
|
||||
slog.Warn("weixin: failed to send incomplete-delivery notice", "peer", rc.peerUserID, "error", nerr)
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("weixin: send chunk %d/%d: %w", i+1, total, err)
|
||||
}
|
||||
@@ -688,62 +759,28 @@ func (p *Platform) sendChunks(ctx context.Context, replyCtx any, content string)
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendChunkWithRetry sends a single chunk with retry mechanism.
|
||||
// When sendMessage returns ret=-2, it tries to refresh the context_token from
|
||||
// storage (which is updated by every inbound message) before retrying.
|
||||
// If the stored token is the same as the current one (no refresh possible),
|
||||
// it fails fast rather than burning retries on a stale token.
|
||||
// chunkIdx and totalChunks are 1-based indices used for logging context.
|
||||
func (p *Platform) sendChunkWithRetry(ctx context.Context, rc *replyContext, chunk string, chunkIdx, totalChunks int) error {
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < weixinSendMaxRetries; attempt++ {
|
||||
clientID := "cc-" + randomHex(6)
|
||||
err := p.api.sendText(ctx, rc.peerUserID, chunk, rc.contextToken, clientID)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
lastErr = err
|
||||
// Check if error is ret=-2 (API declined) - attempt token refresh
|
||||
if strings.Contains(err.Error(), "ret=-2") {
|
||||
preview := []rune(chunk)
|
||||
if len(preview) > 50 {
|
||||
preview = preview[:50]
|
||||
}
|
||||
// Refresh context_token from stored tokens (may have been updated by a
|
||||
// concurrent inbound message while we were waiting).
|
||||
freshToken := p.getContextToken(rc.peerUserID)
|
||||
if freshToken == "" || freshToken == rc.contextToken {
|
||||
// No fresh token available — further retries would use the same stale
|
||||
// token and all fail. Fail fast with an actionable error.
|
||||
slog.Warn("weixin: sendMessage ret=-2, no fresh context_token available — "+
|
||||
"user must send a new message to refresh the session token",
|
||||
"attempt", attempt+1, "peer", rc.peerUserID,
|
||||
"chunk", fmt.Sprintf("%d/%d", chunkIdx, totalChunks),
|
||||
"chunk_runes", utf8.RuneCountInString(chunk),
|
||||
"preview", string(preview))
|
||||
return fmt.Errorf("weixin: sendMessage ret=-2 (expired context_token); "+
|
||||
"user must send a new message to peer %q to refresh the session token: %w",
|
||||
rc.peerUserID, lastErr)
|
||||
}
|
||||
slog.Warn("weixin: sendMessage ret=-2, retrying with fresh context_token",
|
||||
"attempt", attempt+1, "peer", rc.peerUserID,
|
||||
"chunk", fmt.Sprintf("%d/%d", chunkIdx, totalChunks),
|
||||
"chunk_runes", utf8.RuneCountInString(chunk),
|
||||
"preview", string(preview))
|
||||
rc.contextToken = freshToken
|
||||
slog.Debug("weixin: using refreshed context_token for retry", "peer", rc.peerUserID)
|
||||
// Brief delay before retry
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(weixinSendRetryDelay):
|
||||
}
|
||||
continue
|
||||
}
|
||||
// For other errors, don't retry
|
||||
return err
|
||||
// isSendThrottled reports whether err is ilink sendmessage's burst-throttle
|
||||
// response (ret=-2 "prepare failed"). This is a bot-wide rate-limit penalty, not a
|
||||
// context_token problem: the gateway accepts any (or no) context_token on sends.
|
||||
func isSendThrottled(err error) bool {
|
||||
return err != nil && strings.Contains(err.Error(), "ret=-2")
|
||||
}
|
||||
|
||||
// sendChunk sends a single chunk. If ilink throttles the send (ret=-2
|
||||
// "prepare failed"), it fails fast instead of retrying: live testing showed the
|
||||
// penalty is escalated by every send attempt made while it is active, so retrying
|
||||
// (e.g. the old 3×500ms loop plus the extra notice send) only prolongs the outage.
|
||||
func (p *Platform) sendChunk(ctx context.Context, rc *replyContext, chunk string) error {
|
||||
clientID := "cc-" + randomHex(6)
|
||||
err := p.api.sendText(ctx, rc.peerUserID, chunk, rc.contextToken, clientID)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
return lastErr
|
||||
if isSendThrottled(err) {
|
||||
return fmt.Errorf("weixin: sendMessage throttled by ilink (ret=-2); "+
|
||||
"the bot is rate-limited and sending during the penalty escalates it, retry the message later: %w", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func truncatePreview(s string, max int) string {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
@@ -458,3 +459,143 @@ func TestReconstructReplyCtx_MissingToken(t *testing.T) {
|
||||
t.Errorf("error = %q, want it to mention 'no stored context_token'", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// TestIsSendThrottled verifies ret=-2 (ilink sendmessage burst throttle) is
|
||||
// recognized as a throttle and other errors are not.
|
||||
func TestIsSendThrottled(t *testing.T) {
|
||||
if !isSendThrottled(fmt.Errorf("weixin: sendMessage: ret=-2 errcode=0 errmsg=prepare failed")) {
|
||||
t.Fatal("ret=-2 should be recognized as a throttle")
|
||||
}
|
||||
if isSendThrottled(fmt.Errorf("weixin: sendMessage: connection reset by peer")) {
|
||||
t.Fatal("non-ret=-2 error should not be treated as a throttle")
|
||||
}
|
||||
if isSendThrottled(nil) {
|
||||
t.Fatal("nil error should not be treated as a throttle")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSendChunk_FailsFastOnThrottle verifies a throttled send fails immediately
|
||||
// with a single sendmessage call — no retries, because every attempt made during
|
||||
// the ilink penalty window escalates it.
|
||||
func TestSendChunk_FailsFastOnThrottle(t *testing.T) {
|
||||
var sendCalls atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
sendCalls.Add(1)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"ret":-2,"errmsg":"prepare failed"}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
p := &Platform{httpClient: &http.Client{}}
|
||||
p.api = newAPIClient(srv.URL, "tok", "", p.httpClient)
|
||||
rc := &replyContext{peerUserID: "peer-1", contextToken: "tok-1"}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
err := p.sendChunk(ctx, rc, "hello")
|
||||
if err == nil {
|
||||
t.Fatal("expected a throttled error, got nil")
|
||||
}
|
||||
if !isSendThrottled(err) {
|
||||
t.Fatalf("error should be recognized as a throttle, got: %v", err)
|
||||
}
|
||||
if got := sendCalls.Load(); got != 1 {
|
||||
t.Fatalf("sendmessage calls = %d, want 1 (fail fast, no retries)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSendChunk_Succeeds verifies a normal send returns nil and issues exactly
|
||||
// one sendmessage call.
|
||||
func TestSendChunk_Succeeds(t *testing.T) {
|
||||
var sendCalls atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
sendCalls.Add(1)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"message_id":123}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
p := &Platform{httpClient: &http.Client{}}
|
||||
p.api = newAPIClient(srv.URL, "tok", "", p.httpClient)
|
||||
rc := &replyContext{peerUserID: "peer-1", contextToken: "tok-1"}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := p.sendChunk(ctx, rc, "hello"); err != nil {
|
||||
t.Fatalf("sendChunk failed: %v", err)
|
||||
}
|
||||
if got := sendCalls.Load(); got != 1 {
|
||||
t.Fatalf("sendmessage calls = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckSendQuota_AllowsUnderLimit verifies sends under the window limit pass
|
||||
// through without error.
|
||||
func TestCheckSendQuota_AllowsUnderLimit(t *testing.T) {
|
||||
p := &Platform{sendQuotaLimit: 4, sendQuotaWindow: time.Hour}
|
||||
ctx := context.Background()
|
||||
for i := 0; i < 4; i++ {
|
||||
if err := p.checkSendQuota(ctx); err != nil {
|
||||
t.Fatalf("checkSendQuota(%d) unexpectedly failed: %v", i, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckSendQuota_FailsWhenOverLimit verifies the quota fails fast (no
|
||||
// waiting) once the window budget is exhausted.
|
||||
func TestCheckSendQuota_FailsWhenOverLimit(t *testing.T) {
|
||||
p := &Platform{sendQuotaLimit: 1, sendQuotaWindow: time.Hour}
|
||||
ctx := context.Background()
|
||||
if err := p.checkSendQuota(ctx); err != nil {
|
||||
t.Fatalf("first send should pass: %v", err)
|
||||
}
|
||||
start := time.Now()
|
||||
if err := p.checkSendQuota(ctx); err == nil {
|
||||
t.Fatal("second send should fail (budget exhausted)")
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
|
||||
t.Fatalf("over-budget send should fail fast, took %v", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckSendQuota_DisabledWhenZero verifies limit 0 disables the quota entirely.
|
||||
func TestCheckSendQuota_DisabledWhenZero(t *testing.T) {
|
||||
p := &Platform{sendQuotaLimit: 0, sendQuotaWindow: time.Hour}
|
||||
ctx := context.Background()
|
||||
for i := 0; i < 10; i++ {
|
||||
if err := p.checkSendQuota(ctx); err != nil {
|
||||
t.Fatalf("disabled quota should never fail: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestSendChunks_AppliesQuota verifies the budget is enforced end-to-end through
|
||||
// sendChunks (httptest server): under budget sends succeed, over budget fails.
|
||||
func TestSendChunks_AppliesQuota(t *testing.T) {
|
||||
var sendCalls atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
sendCalls.Add(1)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"message_id":123}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
p := &Platform{httpClient: &http.Client{}, sendQuotaLimit: 1, sendQuotaWindow: time.Hour}
|
||||
p.api = newAPIClient(srv.URL, "tok", "", p.httpClient)
|
||||
rc := &replyContext{peerUserID: "peer-1", contextToken: "tok-1"}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := p.sendChunks(ctx, rc, "first"); err != nil {
|
||||
t.Fatalf("first send failed: %v", err)
|
||||
}
|
||||
if err := p.sendChunks(ctx, rc, "second"); err == nil {
|
||||
t.Fatal("second send should fail (budget exhausted)")
|
||||
}
|
||||
if got := sendCalls.Load(); got != 1 {
|
||||
t.Fatalf("sendmessage calls = %d, want 1 (over-budget send not attempted)", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user