fix(qq): add auto-reconnect with exponential backoff and model_not_found test coverage
QQ channel now automatically reconnects on WebSocket disconnects using exponential backoff (5s to 5min). Each reconnect fetches a fresh WS endpoint and creates a new SessionManager to avoid stale session/auth errors. Also adds 12 unit tests for model_not_found error classification patterns and transient status code override behavior.
This commit is contained in:
kent.liu
parent
dc109b6816
commit
d87ba9ab98
2 changed files with 171 additions and 25 deletions
|
|
@ -35,12 +35,15 @@ import (
|
|||
)
|
||||
|
||||
const (
|
||||
dedupTTL = 5 * time.Minute
|
||||
dedupInterval = 60 * time.Second
|
||||
dedupMaxSize = 10000 // hard cap on dedup map entries
|
||||
typingResend = 8 * time.Second
|
||||
typingSeconds = 10
|
||||
bytesPerMiB = 1024 * 1024
|
||||
dedupTTL = 5 * time.Minute
|
||||
dedupInterval = 60 * time.Second
|
||||
dedupMaxSize = 10000 // hard cap on dedup map entries
|
||||
typingResend = 8 * time.Second
|
||||
typingSeconds = 10
|
||||
bytesPerMiB = 1024 * 1024
|
||||
reconnectInitial = 5 * time.Second
|
||||
reconnectMax = 5 * time.Minute
|
||||
reconnectMultiplier = 2.0
|
||||
)
|
||||
|
||||
type qqAPI interface {
|
||||
|
|
@ -127,6 +130,28 @@ func (c *QQChannel) Start(ctx context.Context) error {
|
|||
// initialize OpenAPI client
|
||||
c.api = botgo.NewOpenAPI(c.config.AppID, c.tokenSource).WithTimeout(5 * time.Second)
|
||||
|
||||
// start dedup janitor goroutine
|
||||
go c.dedupJanitor()
|
||||
|
||||
// Pre-register reasoning_channel_id as group chat if configured,
|
||||
// so outbound-only destinations are routed correctly.
|
||||
if c.config.ReasoningChannelID != "" {
|
||||
c.chatType.Store(c.config.ReasoningChannelID, "group")
|
||||
}
|
||||
|
||||
// Start the reconnect loop. It handles both the initial connection and
|
||||
// subsequent reconnection attempts with exponential backoff.
|
||||
go c.reconnectLoop()
|
||||
|
||||
c.SetRunning(true)
|
||||
logger.InfoC("qq", "QQ bot started successfully")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// startSession fetches a fresh WebSocket endpoint and starts the session manager.
|
||||
// It blocks until the session goroutine exits (i.e. the WebSocket connection drops).
|
||||
func (c *QQChannel) startSession() error {
|
||||
// register event handlers
|
||||
intent := event.RegisterHandlers(
|
||||
c.handleC2CMessage(),
|
||||
|
|
@ -146,29 +171,59 @@ func (c *QQChannel) Start(ctx context.Context) error {
|
|||
// create and save sessionManager
|
||||
c.sessionManager = botgo.NewSessionManager()
|
||||
|
||||
// start WebSocket connection in goroutine to avoid blocking
|
||||
go func() {
|
||||
if err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent); err != nil {
|
||||
logger.ErrorCF("qq", "WebSocket session error", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
c.SetRunning(false)
|
||||
return c.sessionManager.Start(wsInfo, c.tokenSource, &intent)
|
||||
}
|
||||
|
||||
// reconnectLoop manages the QQ WebSocket session lifecycle with exponential backoff.
|
||||
// On initial start or after a disconnect, it re-acquires the WebSocket endpoint
|
||||
// and re-initializes the session manager.
|
||||
func (c *QQChannel) reconnectLoop() {
|
||||
backoff := reconnectInitial
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-c.done:
|
||||
return
|
||||
default:
|
||||
}
|
||||
}()
|
||||
|
||||
// start dedup janitor goroutine
|
||||
go c.dedupJanitor()
|
||||
// Check if channel was explicitly stopped.
|
||||
if !c.IsRunning() {
|
||||
return
|
||||
}
|
||||
|
||||
// Pre-register reasoning_channel_id as group chat if configured,
|
||||
// so outbound-only destinations are routed correctly.
|
||||
if c.config.ReasoningChannelID != "" {
|
||||
c.chatType.Store(c.config.ReasoningChannelID, "group")
|
||||
logger.InfoCF("qq", "Starting QQ session", map[string]any{
|
||||
"backoff": backoff.String(),
|
||||
})
|
||||
|
||||
err := c.startSession()
|
||||
if err == nil {
|
||||
// Session exited cleanly (e.g. via Stop/cancel).
|
||||
return
|
||||
}
|
||||
|
||||
// Session failed or disconnected -- log and schedule retry.
|
||||
logger.WarnCF("qq", "QQ session ended, reconnecting", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-c.done:
|
||||
return
|
||||
case <-time.After(backoff):
|
||||
if backoff < reconnectMax {
|
||||
next := time.Duration(float64(backoff) * reconnectMultiplier)
|
||||
if next > reconnectMax {
|
||||
next = reconnectMax
|
||||
}
|
||||
backoff = next
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
c.SetRunning(true)
|
||||
logger.InfoC("qq", "QQ bot started successfully")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *QQChannel) Stop(ctx context.Context) error {
|
||||
|
|
|
|||
|
|
@ -264,6 +264,97 @@ func TestClassifyError_UnknownError(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestClassifyError_ModelNotFoundPatterns(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
msg string
|
||||
want FailoverReason
|
||||
wantOK bool
|
||||
}{
|
||||
// Exact pattern: model_not_found (zhipu actual format)
|
||||
{name: "model_not_found", msg: "model_not_found", want: FailoverFormat, wantOK: true},
|
||||
// Space-separated
|
||||
{name: "model not found", msg: "model not found", want: FailoverFormat, wantOK: true},
|
||||
// invalid model
|
||||
{name: "invalid model", msg: "invalid model", want: FailoverFormat, wantOK: true},
|
||||
// model does not exist
|
||||
{name: "model does not exist", msg: "model does not exist", want: FailoverFormat, wantOK: true},
|
||||
// Case insensitive
|
||||
{name: "Model_Not_Found upper", msg: "Model_Not_Found", want: FailoverFormat, wantOK: true},
|
||||
// model not supported
|
||||
{name: "model not supported", msg: "model not supported", want: FailoverFormat, wantOK: true},
|
||||
// model not available
|
||||
{name: "model not available", msg: "model not available", want: FailoverFormat, wantOK: true},
|
||||
// unknown model
|
||||
{name: "unknown model", msg: "unknown model", want: FailoverFormat, wantOK: true},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := errors.New(tt.msg)
|
||||
result := ClassifyError(err, "zhipu", "glm-4")
|
||||
if tt.wantOK {
|
||||
if result == nil {
|
||||
t.Fatalf("expected non-nil for %q", tt.msg)
|
||||
}
|
||||
if result.Reason != tt.want {
|
||||
t.Errorf("reason = %q, want %q", result.Reason, tt.want)
|
||||
}
|
||||
} else {
|
||||
if result != nil {
|
||||
t.Errorf("expected nil for %q, got %+v", tt.msg, result)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassifyError_TransientStatusOverride(t *testing.T) {
|
||||
// When a transient 5xx status (e.g. 503) is accompanied by a
|
||||
// model_not_found message body, the message pattern should win over
|
||||
// the status code classification. This verifies the isTransientStatus
|
||||
// override logic in ClassifyError.
|
||||
tests := []struct {
|
||||
name string
|
||||
errMsg string
|
||||
want FailoverReason
|
||||
}{
|
||||
{
|
||||
name: "503 with model_not_found should be format not timeout",
|
||||
errMsg: "API error: status: 503 model_not_found",
|
||||
want: FailoverFormat,
|
||||
},
|
||||
{
|
||||
name: "500 with model not found should be format not timeout",
|
||||
errMsg: "API error: status: 500 model not found",
|
||||
want: FailoverFormat,
|
||||
},
|
||||
{
|
||||
name: "502 with invalid model should be format not timeout",
|
||||
errMsg: "status 502 - invalid model",
|
||||
want: FailoverFormat,
|
||||
},
|
||||
{
|
||||
name: "503 without model pattern should remain timeout",
|
||||
errMsg: "API error: status: 503 service unavailable",
|
||||
want: FailoverTimeout,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := errors.New(tt.errMsg)
|
||||
result := ClassifyError(err, "zhipu", "glm-4")
|
||||
if result == nil {
|
||||
t.Fatalf("expected non-nil for %q", tt.errMsg)
|
||||
}
|
||||
if result.Reason != tt.want {
|
||||
t.Errorf("reason = %q, want %q", result.Reason, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassifyError_ProviderModelPropagation(t *testing.T) {
|
||||
err := errors.New("rate limit exceeded")
|
||||
result := ClassifyError(err, "my-provider", "my-model")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue