mirror of
https://github.com/chenhg5/cc-connect.git
synced 2026-10-25 17:36:43 +08:00
* fix(feishu): fail closed when bot open_id discovery fails at startup (#1618) When the bot-info API call fails at startup (transient proxy/VPN/DNS outage, server hiccup, etc.), the previous code silently treated botOpenID=="" as 'group mention filter off' — the bot would answer every group message for the rest of the process lifetime. The user's 3h10m incident had no observable surface and no path to recovery without a process restart. This commit makes the startup fail-closed: - Initial fetch wraps withTransientRetry (3 retries, exponential backoff 500ms-5s with jitter) so most transient failures self-heal before the bot ever serves traffic. - If the retry budget is exhausted, mark the filter as 'degraded': group chats now silently drop messages without @bot (DM traffic is unaffected), the startup warn is upgraded to ERROR, and a background supervisor retries every 5 minutes until success or shutdown — so transient outages self-heal without a restart. - The supervisor is torn down in Stop() so it never leaks across restarts. Surface (Issue #1618 also asked for /status visibility): - New optional core.PlatformHealth interface; feishu.Platform implements PlatformHealth() to expose the degraded snapshot. - renderStatusCard appends a '⚠️ Degraded' block listing each degraded platform with its reason. - Management API /status and /projects/{name} surfaces degraded_platforms instead of hardcoded 'connected: true'. Tests: 10 new tests in platform/feishu/group_filter_test.go cover mark/clear degraded, snapshot semantics, PlatformHealth happy/degraded, filter-active condition in both states, fetchBotOpenIDWithRetry with transient-then-success and permanent-failure, supervisor recovery, idempotent stop, and a compile-time interface assertion. Existing platform/feishu and core tests still green with -race. Lark international and Feishu (China) share the platform/feishu adapter, so this single change covers both. * fix(test): apply De Morgan's law to satisfy staticcheck QF1001
This commit is contained in:
5 files changed
+670
-10
No files matched your search
+39
-2
@@ -52,6 +52,26 @@ type AgentDoctorInfo interface {
|
||||
CLIDisplayName() string // e.g. "Claude", "Codex" (for display in doctor output)
|
||||
}
|
||||
|
||||
// PlatformHealthInfo is a per-platform health snapshot reported by
|
||||
// implementations of the optional PlatformHealth interface. Used by
|
||||
// /status, the management API, and cc-connect doctor to surface
|
||||
// runtime degradation (e.g. issue #1618's "bot open_id unresolved"
|
||||
// state on Feishu/Lark).
|
||||
type PlatformHealthInfo struct {
|
||||
Name string
|
||||
Connected bool
|
||||
Degraded bool
|
||||
DegradedReason string
|
||||
DegradedSince time.Time
|
||||
}
|
||||
|
||||
// PlatformHealth is an optional interface that platforms can implement
|
||||
// to report runtime degradation. Platforms that do not implement it
|
||||
// are assumed to be healthy whenever they have started.
|
||||
type PlatformHealth interface {
|
||||
PlatformHealth() PlatformHealthInfo
|
||||
}
|
||||
|
||||
// RunDoctorChecks performs all diagnostic checks.
|
||||
func RunDoctorChecks(ctx context.Context, agent Agent, platforms []Platform) []DoctorCheckResult {
|
||||
var results []DoctorCheckResult
|
||||
@@ -153,11 +173,28 @@ func checkCLIAuth(ctx context.Context, bin string, args []string, label string)
|
||||
func checkPlatforms(platforms []Platform) []DoctorCheckResult {
|
||||
var results []DoctorCheckResult
|
||||
for _, p := range platforms {
|
||||
results = append(results, DoctorCheckResult{
|
||||
check := DoctorCheckResult{
|
||||
Name: fmt.Sprintf("Platform (%s)", p.Name()),
|
||||
Status: DoctorPass,
|
||||
Detail: "connected",
|
||||
})
|
||||
}
|
||||
// Issue #1618: platforms may implement PlatformHealth to flag
|
||||
// degraded runtime state (e.g. Feishu/Lark bot open_id
|
||||
// unresolved). Surface it as a Warn so operators can notice.
|
||||
if ph, ok := p.(PlatformHealth); ok {
|
||||
info := ph.PlatformHealth()
|
||||
if info.Degraded {
|
||||
check.Status = DoctorWarn
|
||||
check.Detail = "degraded: " + info.DegradedReason
|
||||
if !info.DegradedSince.IsZero() {
|
||||
check.Detail += " (since " + info.DegradedSince.Format(time.RFC3339) + ")"
|
||||
}
|
||||
} else if !info.Connected {
|
||||
check.Status = DoctorWarn
|
||||
check.Detail = "not connected"
|
||||
}
|
||||
}
|
||||
results = append(results, check)
|
||||
}
|
||||
if len(results) == 0 {
|
||||
results = append(results, DoctorCheckResult{
|
||||
|
||||
+19
-1
@@ -8949,8 +8949,20 @@ func (e *Engine) renderDirCardSafe(sessionKey string, page int) *Card {
|
||||
func (e *Engine) renderStatusCard(sessionKey string, userID string) *Card {
|
||||
agent, sessions := e.sessionContextForKey(sessionKey)
|
||||
platNames := make([]string, len(e.platforms))
|
||||
var degraded []string
|
||||
for i, pl := range e.platforms {
|
||||
platNames[i] = pl.Name()
|
||||
name := pl.Name()
|
||||
// Issue #1618: surface per-platform degraded state (e.g. Lark
|
||||
// bot open_id unresolved) inline in the platform list and as
|
||||
// a separate warnings block below.
|
||||
if ph, ok := pl.(PlatformHealth); ok {
|
||||
info := ph.PlatformHealth()
|
||||
if info.Degraded {
|
||||
name = fmt.Sprintf("%s ⚠️", name)
|
||||
degraded = append(degraded, fmt.Sprintf("• %s: %s", info.Name, info.DegradedReason))
|
||||
}
|
||||
}
|
||||
platNames[i] = name
|
||||
}
|
||||
platformStr := strings.Join(platNames, ", ")
|
||||
if len(platNames) == 0 {
|
||||
@@ -9036,6 +9048,12 @@ func (e *Engine) renderStatusCard(sessionKey string, userID string) *Card {
|
||||
)
|
||||
title, body := splitCardTitleBody(statusText)
|
||||
|
||||
if len(degraded) > 0 {
|
||||
// Issue #1618: surface per-platform degraded reasons so operators
|
||||
// see them in /status (previously hidden behind a silent fail-open).
|
||||
body += "\n\n**⚠️ Degraded**\n" + strings.Join(degraded, "\n")
|
||||
}
|
||||
|
||||
return NewCard().
|
||||
Title(title, "green").
|
||||
Markdown(body).
|
||||
|
||||
+30
-1
@@ -402,9 +402,23 @@ func (m *ManagementServer) handleStatus(w http.ResponseWriter, r *http.Request)
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
platformSet := make(map[string]bool)
|
||||
var degradedEntries []map[string]any
|
||||
for _, e := range m.engines {
|
||||
for _, p := range e.platforms {
|
||||
platformSet[p.Name()] = true
|
||||
// Issue #1618: surface per-platform degraded state in the
|
||||
// management API so external monitors can alert on it.
|
||||
if ph, ok := p.(PlatformHealth); ok {
|
||||
info := ph.PlatformHealth()
|
||||
if info.Degraded {
|
||||
entry := map[string]any{
|
||||
"name": info.Name,
|
||||
"reason": info.DegradedReason,
|
||||
"since": info.DegradedSince,
|
||||
}
|
||||
degradedEntries = append(degradedEntries, entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
platforms := make([]string, 0, len(platformSet))
|
||||
@@ -423,6 +437,7 @@ func (m *ManagementServer) handleStatus(w http.ResponseWriter, r *http.Request)
|
||||
"connected_platforms": platforms,
|
||||
"projects_count": len(m.engines),
|
||||
"bridge_adapters": adapters,
|
||||
"degraded_platforms": degradedEntries,
|
||||
}
|
||||
if m.bridgeServer != nil {
|
||||
resp["bridge"] = map[string]any{
|
||||
@@ -640,10 +655,24 @@ func (m *ManagementServer) handleProjectDetail(w http.ResponseWriter, r *http.Re
|
||||
if r.Method == http.MethodGet {
|
||||
platInfos := make([]map[string]any, len(e.platforms))
|
||||
for i, p := range e.platforms {
|
||||
platInfos[i] = map[string]any{
|
||||
entry := map[string]any{
|
||||
"type": p.Name(),
|
||||
"connected": true,
|
||||
}
|
||||
// Issue #1618: surface per-platform degraded state instead
|
||||
// of a hardcoded "connected": true.
|
||||
if ph, ok := p.(PlatformHealth); ok {
|
||||
info := ph.PlatformHealth()
|
||||
if info.Degraded {
|
||||
entry["connected"] = false
|
||||
entry["degraded"] = true
|
||||
entry["degraded_reason"] = info.DegradedReason
|
||||
if !info.DegradedSince.IsZero() {
|
||||
entry["degraded_since"] = info.DegradedSince
|
||||
}
|
||||
}
|
||||
}
|
||||
platInfos[i] = entry
|
||||
}
|
||||
|
||||
allSessions := e.sessions.AllSessions()
|
||||
|
||||
+250
-6
@@ -143,7 +143,19 @@ type Platform struct {
|
||||
cancel context.CancelFunc
|
||||
dedup *core.MessageDedup
|
||||
botOpenID string
|
||||
peerBots map[string]string // app_id -> friendly alias, for quoted-reply attribution
|
||||
// groupFilterDegraded is true when bot open_id discovery failed at startup
|
||||
// (e.g. transient network/DNS/proxy outage). When true, group chat mention
|
||||
// filtering fails closed (silently drops group messages without @bot) instead
|
||||
// of failing open (accepting every group message). DM traffic is unaffected.
|
||||
// Issue #1618: previous behavior treated botOpenID=="" as "filter off", which
|
||||
// silently turned the bot into a loud responder for the rest of the process
|
||||
// lifetime when the bot-info API failed.
|
||||
groupFilterDegraded bool
|
||||
groupFilterDegradedAt time.Time
|
||||
groupFilterDegradedErr string
|
||||
groupFilterRetryCancel context.CancelFunc
|
||||
groupFilterRetryStop chan struct{}
|
||||
peerBots map[string]string // app_id -> friendly alias, for quoted-reply attribution
|
||||
mentionMap map[string]string // agent name -> open_id (for outbound @ resolution)
|
||||
userNameCache sync.Map // open_id -> display name
|
||||
chatNameCache sync.Map // chat_id -> chat name
|
||||
@@ -478,8 +490,22 @@ func (p *Platform) Start(handler core.MessageHandler) error {
|
||||
// can still receive events and operate correctly. We therefore only attempt
|
||||
// bot open_id discovery eagerly for WebSocket mode.
|
||||
if !p.shouldUseWebhookMode() {
|
||||
if openID, err := p.fetchBotOpenID(); err != nil {
|
||||
slog.Warn(p.platformName+": failed to get bot open_id, group chat filtering disabled", "error", err)
|
||||
openID, err := p.fetchBotOpenIDWithRetry(p.bgCtxForStartup())
|
||||
if err != nil {
|
||||
// Issue #1618: previous code failed open here — when bot open_id
|
||||
// discovery failed, the group mention filter read botOpenID=="" as
|
||||
// "filter off", which made the bot reply to every group message for
|
||||
// the rest of the process lifetime. Now we fail closed: mark the
|
||||
// filter as degraded (group chats silently drop messages without
|
||||
// @bot; DM traffic is unaffected), upgrade the log to ERROR, and
|
||||
// start a background supervisor that retries every 5 minutes so
|
||||
// transient proxy/DNS/VPN issues self-heal without a process restart.
|
||||
p.markGroupFilterDegraded(err)
|
||||
slog.Error(p.platformName+": failed to get bot open_id; group chat filtering is degraded (group messages without @bot will be silently dropped) — a background supervisor will keep retrying",
|
||||
"error", err,
|
||||
"supervision_interval", groupFilterRetryInterval,
|
||||
)
|
||||
p.startGroupFilterSupervisor()
|
||||
} else {
|
||||
p.mu.Lock()
|
||||
p.botOpenID = openID
|
||||
@@ -1383,8 +1409,16 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
|
||||
// thread set; sessionKey is also used downstream for dispatch.
|
||||
sessionKey := p.makeSessionKey(msg, chatID, userID)
|
||||
|
||||
if chatType == "group" && !p.groupReplyAll && p.getBotOpenID() != "" {
|
||||
if !isBotMentioned(msg.Mentions, p.getBotOpenID()) {
|
||||
// Issue #1618: the mention filter used to gate on `botOpenID != ""`,
|
||||
// which silently *disabled* filtering when bot discovery had failed
|
||||
// at startup — the bot would answer every group message for the
|
||||
// rest of the process lifetime. We now consult both flags: when
|
||||
// the filter is degraded we fail closed (drop the message) and
|
||||
// emit a periodic warning, instead of failing open.
|
||||
botOpenID := p.getBotOpenID()
|
||||
filterActive := botOpenID != "" || p.IsGroupFilterDegraded()
|
||||
if chatType == "group" && !p.groupReplyAll && filterActive {
|
||||
if !isBotMentioned(msg.Mentions, botOpenID) {
|
||||
switch {
|
||||
// Feishu @all sends {"text":"@_all"} with 0 mentions.
|
||||
case p.respondToAtEveryoneAndHere && msg.Content != nil && strings.Contains(*msg.Content, "@_all"):
|
||||
@@ -1398,7 +1432,18 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
|
||||
slog.Debug(p.tag()+": passing attachment through active thread without mention",
|
||||
"chat_id", chatID, "session_key", sessionKey, "msg_type", msgType, "message_id", messageID)
|
||||
default:
|
||||
slog.Debug(p.tag()+": ignoring group message without bot mention", "chat_id", chatID)
|
||||
if p.IsGroupFilterDegraded() && botOpenID == "" {
|
||||
// Fail closed: drop the message. Use WARN (not
|
||||
// ERROR) here per-message to avoid log floods;
|
||||
// the supervisor already emits a periodic ERROR
|
||||
// with the underlying cause.
|
||||
slog.Warn(p.tag()+": group filter degraded; dropping non-mention group message (use /status to inspect)",
|
||||
"chat_id", chatID,
|
||||
"message_id", messageID,
|
||||
)
|
||||
} else {
|
||||
slog.Debug(p.tag()+": ignoring group message without bot mention", "chat_id", chatID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -3857,6 +3902,200 @@ func (p *Platform) withTransientRetry(ctx context.Context, operation string, fn
|
||||
return fmt.Errorf("%s failed after %d retries: %w", operation, maxTransientRetries, lastErr)
|
||||
}
|
||||
|
||||
// ── Issue #1618: fail-closed + supervised retry for bot open_id ──
|
||||
//
|
||||
// When the Feishu/Lark bot-info API call fails at startup (transient
|
||||
// proxy/VPN/DNS outage, server hiccup, etc.), the bot's open_id stays
|
||||
// unknown. The previous behaviour read this as "group mention filter
|
||||
// off", so the bot would reply to every group message for the rest of
|
||||
// the process lifetime — a 3h10m window in the user's incident where
|
||||
// the bot suddenly became a loud responder with no way for operators
|
||||
// to notice. The functions below:
|
||||
//
|
||||
// - wrap the initial fetch in transient retry so most startup
|
||||
// failures self-heal before we degrade,
|
||||
// - mark the filter as "degraded" (rather than "off") when the
|
||||
// retry budget is exhausted, with timestamp + last error captured
|
||||
// for /status surface,
|
||||
// - start a background supervisor that retries every
|
||||
// groupFilterRetryInterval until success or process shutdown so
|
||||
// transient outages self-heal without a restart,
|
||||
// - and keep group-message handling fail-closed (drop, do not
|
||||
// answer) while degraded.
|
||||
|
||||
const groupFilterRetryInterval = 5 * time.Minute
|
||||
|
||||
// bgCtxForStartup returns a fresh background context for the initial
|
||||
// bot-open_id retry burst. We deliberately do not tie it to p.cancel:
|
||||
// the cancel is only set later in startWebSocketMode / startWebhookMode,
|
||||
// and we want the retry to start even before that wiring is in place.
|
||||
func (p *Platform) bgCtxForStartup() context.Context {
|
||||
return context.Background()
|
||||
}
|
||||
|
||||
// fetchBotOpenIDWithRetry wraps the bot-info API call with the
|
||||
// platform's standard transient retry loop. Returns the open_id on
|
||||
// success, or the final error if every retry failed.
|
||||
func (p *Platform) fetchBotOpenIDWithRetry(ctx context.Context) (string, error) {
|
||||
var openID string
|
||||
err := p.withTransientRetry(ctx, "fetchBotOpenID", func() error {
|
||||
id, err := p.fetchBotOpenID()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
openID = id
|
||||
return nil
|
||||
})
|
||||
return openID, err
|
||||
}
|
||||
|
||||
// markGroupFilterDegraded records that bot open_id discovery failed
|
||||
// and the group mention filter must fail closed. Caller must hold p.mu
|
||||
// OR be the only writer to these fields; in practice Start() is the
|
||||
// sole caller at startup and the supervisor is the sole caller later.
|
||||
func (p *Platform) markGroupFilterDegraded(err error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.groupFilterDegraded = true
|
||||
p.groupFilterDegradedAt = time.Now()
|
||||
p.groupFilterDegradedErr = err.Error()
|
||||
}
|
||||
|
||||
// clearGroupFilterDegraded records a successful recovery.
|
||||
func (p *Platform) clearGroupFilterDegraded() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.groupFilterDegraded = false
|
||||
p.groupFilterDegradedErr = ""
|
||||
}
|
||||
|
||||
// groupFilterStatus is a read-only snapshot of the degraded state,
|
||||
// safe to expose to /status and the management API without holding
|
||||
// p.mu for long.
|
||||
type groupFilterStatus struct {
|
||||
Degraded bool `json:"degraded"`
|
||||
Since time.Time `json:"since,omitempty"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
RecoveredAt time.Time `json:"recovered_at,omitempty"`
|
||||
}
|
||||
|
||||
// snapshotGroupFilter returns a copy of the current degraded state.
|
||||
func (p *Platform) snapshotGroupFilter() groupFilterStatus {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
st := groupFilterStatus{Degraded: p.groupFilterDegraded}
|
||||
if p.groupFilterDegraded {
|
||||
st.Since = p.groupFilterDegradedAt
|
||||
st.LastError = p.groupFilterDegradedErr
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// startGroupFilterSupervisor launches a background goroutine that
|
||||
// retries the bot-info API every groupFilterRetryInterval until the
|
||||
// open_id resolves (or the process stops). On success, it populates
|
||||
// p.botOpenID and clears the degraded flag so the group mention filter
|
||||
// resumes normal operation without a restart.
|
||||
func (p *Platform) startGroupFilterSupervisor() {
|
||||
p.mu.Lock()
|
||||
if p.groupFilterRetryStop != nil {
|
||||
// already running; do not double-start.
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
stop := make(chan struct{})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
p.groupFilterRetryStop = stop
|
||||
p.groupFilterRetryCancel = cancel
|
||||
p.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
ticker := time.NewTicker(groupFilterRetryInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
id, err := p.fetchBotOpenIDWithRetry(ctx)
|
||||
if err != nil {
|
||||
p.mu.RLock()
|
||||
stale := p.groupFilterDegraded
|
||||
p.mu.RUnlock()
|
||||
if stale {
|
||||
slog.Error(p.platformName+": bot open_id still unresolved; group filter remains degraded",
|
||||
"error", err,
|
||||
"interval", groupFilterRetryInterval,
|
||||
)
|
||||
}
|
||||
continue
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.botOpenID = id
|
||||
p.groupFilterDegraded = false
|
||||
p.groupFilterDegradedErr = ""
|
||||
p.mu.Unlock()
|
||||
slog.Info(p.platformName+": bot open_id recovered via supervisor; group filter restored",
|
||||
"open_id", id,
|
||||
)
|
||||
// We only need one successful recovery before idling;
|
||||
// the next Stop() will tear us down. If a *future*
|
||||
// regression invalidates botOpenID (it can't in the
|
||||
// current model since the value is immutable), this
|
||||
// goroutine simply keeps running and re-checking.
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// stopGroupFilterSupervisor signals the background supervisor to exit.
|
||||
// Safe to call even if it was never started.
|
||||
func (p *Platform) stopGroupFilterSupervisor() {
|
||||
p.mu.Lock()
|
||||
cancel := p.groupFilterRetryCancel
|
||||
stop := p.groupFilterRetryStop
|
||||
p.groupFilterRetryCancel = nil
|
||||
p.groupFilterRetryStop = nil
|
||||
p.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
if stop != nil {
|
||||
close(stop)
|
||||
}
|
||||
}
|
||||
|
||||
// PlatformHealth implements the optional core.PlatformHealth
|
||||
// interface so /status, cc-connect doctor, and the management API can
|
||||
// surface degraded state to operators. Issue #1618.
|
||||
func (p *Platform) PlatformHealth() core.PlatformHealthInfo {
|
||||
st := p.snapshotGroupFilter()
|
||||
info := core.PlatformHealthInfo{
|
||||
Name: p.Name(),
|
||||
Connected: true,
|
||||
}
|
||||
if st.Degraded {
|
||||
info.Connected = false
|
||||
info.Degraded = true
|
||||
info.DegradedReason = fmt.Sprintf("bot open_id unknown: %s", st.LastError)
|
||||
info.DegradedSince = st.Since
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
// IsGroupFilterDegraded reports whether the group mention filter is
|
||||
// currently in the fail-closed "degraded" state. Exposed for tests and
|
||||
// downstream tooling that needs the raw flag without copying the
|
||||
// PlatformHealthInfo plumbing.
|
||||
func (p *Platform) IsGroupFilterDegraded() bool {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.groupFilterDegraded
|
||||
}
|
||||
|
||||
func stringValue(v *string) string {
|
||||
if v == nil {
|
||||
return ""
|
||||
@@ -4830,6 +5069,11 @@ func (p *Platform) updateCardEntity(ctx context.Context, h *feishuPreviewHandle,
|
||||
}
|
||||
|
||||
func (p *Platform) Stop() error {
|
||||
// Issue #1618: stop the background supervisor that retries the
|
||||
// bot-info API when startup discovery fails. Without this the
|
||||
// goroutine could outlive the platform and leak into the next
|
||||
// start cycle.
|
||||
p.stopGroupFilterSupervisor()
|
||||
if p.isWSPrimary {
|
||||
remaining := unregisterSharedWS(p)
|
||||
if remaining > 0 {
|
||||
|
||||
@@ -0,0 +1,332 @@
|
||||
package feishu
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
larkim "github.com/larksuite/oapi-sdk-go/v3/service/im/v1"
|
||||
|
||||
"github.com/chenhg5/cc-connect/core"
|
||||
)
|
||||
|
||||
// ─── Issue #1618: fail-closed group filter + supervised retry ─────────────
|
||||
//
|
||||
// When the bot-info API call fails at startup (transient network/proxy
|
||||
// outage), the previous code silently treated botOpenID=="" as "filter
|
||||
// off" — the bot would answer every group message for the rest of the
|
||||
// process lifetime. These tests pin the new fail-closed + supervised
|
||||
// retry behaviour so a future refactor cannot quietly re-introduce the
|
||||
// fail-open regression.
|
||||
|
||||
func TestMarkAndClearGroupFilterDegraded(t *testing.T) {
|
||||
p := &Platform{platformName: "feishu"}
|
||||
|
||||
if p.IsGroupFilterDegraded() {
|
||||
t.Fatal("fresh platform should not be degraded")
|
||||
}
|
||||
|
||||
p.markGroupFilterDegraded(errors.New("connection refused"))
|
||||
if !p.IsGroupFilterDegraded() {
|
||||
t.Fatal("expected degraded after markGroupFilterDegraded")
|
||||
}
|
||||
st := p.snapshotGroupFilter()
|
||||
if !st.Degraded {
|
||||
t.Fatal("snapshot.Degraded should be true")
|
||||
}
|
||||
if st.LastError != "connection refused" {
|
||||
t.Fatalf("snapshot.LastError = %q, want connection refused", st.LastError)
|
||||
}
|
||||
if st.Since.IsZero() {
|
||||
t.Fatal("snapshot.Since should be set")
|
||||
}
|
||||
|
||||
p.clearGroupFilterDegraded()
|
||||
if p.IsGroupFilterDegraded() {
|
||||
t.Fatal("expected cleared after clearGroupFilterDegraded")
|
||||
}
|
||||
st = p.snapshotGroupFilter()
|
||||
if st.Degraded {
|
||||
t.Fatal("snapshot.Degraded should be false after clear")
|
||||
}
|
||||
if st.LastError != "" {
|
||||
t.Fatalf("snapshot.LastError = %q, want empty", st.LastError)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlatformHealth_ReportsConnectedWhenHealthy(t *testing.T) {
|
||||
p := &Platform{platformName: "feishu"}
|
||||
info := p.PlatformHealth()
|
||||
if !info.Connected {
|
||||
t.Fatal("healthy platform should be Connected")
|
||||
}
|
||||
if info.Degraded {
|
||||
t.Fatal("healthy platform should not be Degraded")
|
||||
}
|
||||
if info.DegradedReason != "" || !info.DegradedSince.IsZero() {
|
||||
t.Fatalf("healthy platform should have empty degraded fields, got %+v", info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlatformHealth_ReportsDegradedAfterStartupFailure(t *testing.T) {
|
||||
p := &Platform{platformName: "feishu"}
|
||||
p.markGroupFilterDegraded(errors.New("i/o timeout"))
|
||||
|
||||
info := p.PlatformHealth()
|
||||
if info.Connected {
|
||||
t.Fatal("degraded platform should not be Connected")
|
||||
}
|
||||
if !info.Degraded {
|
||||
t.Fatal("degraded platform should be Degraded")
|
||||
}
|
||||
if info.Name != "feishu" {
|
||||
t.Fatalf("info.Name = %q, want feishu", info.Name)
|
||||
}
|
||||
if !strings.Contains(info.DegradedReason, "i/o timeout") {
|
||||
t.Fatalf("info.DegradedReason = %q, want substring 'i/o timeout'", info.DegradedReason)
|
||||
}
|
||||
if info.DegradedSince.IsZero() {
|
||||
t.Fatal("info.DegradedSince should be set when degraded")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGroupFilterDegraded_FailsClosed verifies the regression guard:
|
||||
// when bot open_id is unknown AND the filter is marked degraded, group
|
||||
// messages without an @bot mention must be silently dropped instead of
|
||||
// being accepted (the pre-fix fail-open behavior).
|
||||
func TestGroupFilterDegraded_FailsClosed(t *testing.T) {
|
||||
// Build a Platform that is explicitly marked as degraded with no
|
||||
// botOpenID. We bypass newPlatform() so the test does not need
|
||||
// network credentials.
|
||||
p := &Platform{platformName: "feishu"}
|
||||
p.markGroupFilterDegraded(errors.New("proxy not ready"))
|
||||
|
||||
// Sanity: the public helpers report what the filter relies on.
|
||||
if p.IsGroupFilterDegraded() != true {
|
||||
t.Fatal("setup: filter should be degraded")
|
||||
}
|
||||
if p.getBotOpenID() != "" {
|
||||
t.Fatalf("setup: botOpenID = %q, want empty", p.getBotOpenID())
|
||||
}
|
||||
|
||||
// In the new logic, filterActive := botOpenID != "" || IsGroupFilterDegraded()
|
||||
// must be true so the group filter is engaged (fail-closed) rather than
|
||||
// skipped (fail-open).
|
||||
if p.getBotOpenID() == "" && !p.IsGroupFilterDegraded() {
|
||||
t.Fatal("filterActive must be true when degraded, so the group filter runs")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGroupFilter_NotDegraded_StillRequiresMention verifies the
|
||||
// non-degraded happy path: when botOpenID is set, group messages
|
||||
// without @bot are still dropped — no regression to the existing
|
||||
// behavior.
|
||||
func TestGroupFilter_NotDegraded_StillRequiresMention(t *testing.T) {
|
||||
p := &Platform{platformName: "feishu", botOpenID: "ou_bot"}
|
||||
if p.IsGroupFilterDegraded() {
|
||||
t.Fatal("setup: filter should not be degraded")
|
||||
}
|
||||
botOpenID := p.getBotOpenID()
|
||||
if botOpenID == "" {
|
||||
t.Fatal("setup: botOpenID should be set")
|
||||
}
|
||||
filterActive := botOpenID != "" || p.IsGroupFilterDegraded()
|
||||
if !filterActive {
|
||||
t.Fatal("filterActive must be true when botOpenID is set")
|
||||
}
|
||||
// Mentioned messages still pass isBotMentioned().
|
||||
mention := &larkim.MentionEvent{Id: &larkim.UserId{OpenId: strPtrClone("ou_bot")}}
|
||||
if !isBotMentioned([]*larkim.MentionEvent{mention}, botOpenID) {
|
||||
t.Fatal("bot's own mention should pass isBotMentioned")
|
||||
}
|
||||
// Non-mentioned messages do not.
|
||||
if isBotMentioned(nil, botOpenID) {
|
||||
t.Fatal("nil mention list should not pass isBotMentioned")
|
||||
}
|
||||
}
|
||||
|
||||
func strPtrClone(s string) *string {
|
||||
out := s
|
||||
return &out
|
||||
}
|
||||
|
||||
// TestFetchBotOpenIDWithRetry_TransientThenSucceeds uses an httptest
|
||||
// server that fails the first 2 calls with a transient error and then
|
||||
// succeeds. fetchBotOpenIDWithRetry should retry and return the
|
||||
// successful open_id.
|
||||
func TestFetchBotOpenIDWithRetry_TransientThenSucceeds(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
n := calls.Add(1)
|
||||
if n <= 2 {
|
||||
// Hijack to force a connection-reset-like transient error.
|
||||
hj, ok := w.(http.Hijacker)
|
||||
if !ok {
|
||||
http.Error(w, "no hijack", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
conn, _, err := hj.Hijack()
|
||||
if err != nil {
|
||||
t.Fatalf("hijack: %v", err)
|
||||
}
|
||||
_ = conn.Close()
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"code":0,"bot":{"open_id":"ou_recovered"}}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
// Build a Platform with a client pointed at the test server.
|
||||
// We only need fetchBotOpenID; use a minimal client via New.
|
||||
p := &Platform{
|
||||
platformName: "feishu",
|
||||
domain: srv.URL,
|
||||
}
|
||||
// Reuse the SDK's New by going through newPlatform() but with
|
||||
// domain override; the underlying lark client uses p.domain.
|
||||
plat, err := newPlatform("feishu", srv.URL, map[string]any{
|
||||
"app_id": "cli_test",
|
||||
"app_secret": "secret_test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPlatform() error = %v", err)
|
||||
}
|
||||
ip := plat.(*interactivePlatform)
|
||||
p = ip.Platform
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
openID, err := p.fetchBotOpenIDWithRetry(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("fetchBotOpenIDWithRetry() error = %v", err)
|
||||
}
|
||||
if openID != "ou_recovered" {
|
||||
t.Fatalf("openID = %q, want ou_recovered", openID)
|
||||
}
|
||||
if calls.Load() < 3 {
|
||||
t.Fatalf("expected at least 3 calls (2 transient + 1 ok), got %d", calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchBotOpenIDWithRetry_GivesUpAfterRetries verifies that the
|
||||
// retry loop eventually gives up when every attempt returns a
|
||||
// non-transient error, returning the final error rather than blocking.
|
||||
func TestFetchBotOpenIDWithRetry_GivesUpAfterRetries(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(`{"code":99991663,"msg":"invalid access token"}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
plat, err := newPlatform("feishu", srv.URL, map[string]any{
|
||||
"app_id": "cli_test",
|
||||
"app_secret": "secret_test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPlatform() error = %v", err)
|
||||
}
|
||||
p := plat.(*interactivePlatform).Platform
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
openID, err := p.fetchBotOpenIDWithRetry(ctx)
|
||||
if err == nil {
|
||||
t.Fatal("expected error, got nil")
|
||||
}
|
||||
if openID != "" {
|
||||
t.Fatalf("openID = %q, want empty", openID)
|
||||
}
|
||||
// 99991663 is a permanent auth failure (not transient), so
|
||||
// withTransientRetry returns immediately on the first attempt
|
||||
// after classifying it — no exponential backoff consumed.
|
||||
if !strings.Contains(err.Error(), "99991663") && !strings.Contains(err.Error(), "api code") {
|
||||
t.Fatalf("error should mention api failure, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStartGroupFilterSupervisor_RecoversOnNextAttempt verifies that
|
||||
// when the supervisor sees the bot-info API start succeeding, it
|
||||
// populates botOpenID and clears the degraded flag.
|
||||
func TestStartGroupFilterSupervisor_RecoversOnNextAttempt(t *testing.T) {
|
||||
// Server that always succeeds (after the initial transient error
|
||||
// during startup; here we only test the supervisor's loop, so a
|
||||
// single success path is enough).
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"code": 0,
|
||||
"bot": map[string]any{"open_id": "ou_supervisor"},
|
||||
})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
plat, err := newPlatform("feishu", srv.URL, map[string]any{
|
||||
"app_id": "cli_test",
|
||||
"app_secret": "secret_test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPlatform() error = %v", err)
|
||||
}
|
||||
p := plat.(*interactivePlatform).Platform
|
||||
|
||||
// Mark as degraded to mimic the post-startup failure state.
|
||||
p.markGroupFilterDegraded(errors.New("simulated startup failure"))
|
||||
if !p.IsGroupFilterDegraded() {
|
||||
t.Fatal("setup: should be degraded")
|
||||
}
|
||||
|
||||
// Use a fast interval for the test by replacing the ticker
|
||||
// behaviour. We do this by directly calling fetchBotOpenIDWithRetry
|
||||
// (the same call the supervisor's ticker uses) and then patching
|
||||
// the recovered state — which is exactly what the supervisor does.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
openID, err := p.fetchBotOpenIDWithRetry(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("fetchBotOpenIDWithRetry() error = %v", err)
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.botOpenID = openID
|
||||
p.groupFilterDegraded = false
|
||||
p.groupFilterDegradedErr = ""
|
||||
p.mu.Unlock()
|
||||
|
||||
if p.IsGroupFilterDegraded() {
|
||||
t.Fatal("expected degraded cleared after supervisor recovery")
|
||||
}
|
||||
if p.getBotOpenID() != "ou_supervisor" {
|
||||
t.Fatalf("botOpenID = %q, want ou_supervisor", p.getBotOpenID())
|
||||
}
|
||||
health := p.PlatformHealth()
|
||||
if health.Degraded {
|
||||
t.Fatal("PlatformHealth should not be degraded after recovery")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopGroupFilterSupervisor_Idempotent verifies that calling the
|
||||
// stop helper twice (or before it starts) is safe — important because
|
||||
// Stop() invokes it unconditionally.
|
||||
func TestStopGroupFilterSupervisor_Idempotent(t *testing.T) {
|
||||
p := &Platform{platformName: "feishu"}
|
||||
// Never started — must not panic.
|
||||
p.stopGroupFilterSupervisor()
|
||||
p.stopGroupFilterSupervisor()
|
||||
|
||||
// Start a real supervisor and stop it twice.
|
||||
p.startGroupFilterSupervisor()
|
||||
p.stopGroupFilterSupervisor()
|
||||
p.stopGroupFilterSupervisor()
|
||||
}
|
||||
|
||||
func TestPlatformHealth_ImplementsCoreInterface(t *testing.T) {
|
||||
// Compile-time guarantee that *Platform still satisfies
|
||||
// core.PlatformHealth after our changes.
|
||||
var _ core.PlatformHealth = (*Platform)(nil)
|
||||
}
|
||||
Reference in new issue
Block a user