From 01819b5bc22e47448dc23e6dc2f462cf93e2b5fc Mon Sep 17 00:00:00 2001 From: wizardchen Date: Mon, 22 Jun 2026 15:00:16 +0800 Subject: [PATCH] fix(im): prevent DingTalk stream connection from going stale DingTalk websocket (Stream) channels stopped delivering @-mentions after a day or two and only recovered when the IM channel was toggled off/on. The factory fired the SDK's Start once and relied entirely on the SDK's internal auto-reconnect, with no liveness supervision on WeKnora's side. On long-running connections the SDK can enter a zombie state where the connection object stays alive but no longer receives messages, leaving no way to recover automatically. The cancel func was also ineffective because the SDK ignores the context after the initial dial, so stopping a channel leaked the old connection. Add a reusable RunSupervised helper that keeps a long connection alive and proactively recycles it every 6h on top of the SDK's own reconnect, which bounds the worst-case outage and clears stuck connections. Wire DingTalk to it, add LongConnClient.Close (disables AutoReconnect before closing so the SDK does not resurrect the connection), and make the cancel func actually tear the connection down. Drop the now-unused stream client reference from the adapter, which only sends replies via sessionWebhook or OpenAPI. --- internal/im/dingtalk/adapter.go | 12 ++-- internal/im/dingtalk/factory.go | 21 ++++--- internal/im/dingtalk/longconn.go | 24 ++++---- internal/im/supervisor.go | 97 ++++++++++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 28 deletions(-) create mode 100644 internal/im/supervisor.go diff --git a/internal/im/dingtalk/adapter.go b/internal/im/dingtalk/adapter.go index 80844825..3540de09 100644 --- a/internal/im/dingtalk/adapter.go +++ b/internal/im/dingtalk/adapter.go @@ -41,8 +41,7 @@ var ( type Adapter struct { clientID string clientSecret string - cardTemplateID string // optional: enables AI card streaming when set - client *LongConnClient // nil for webhook mode + cardTemplateID string // optional: enables AI card streaming when set // accessToken cache tokenMu sync.RWMutex @@ -60,14 +59,15 @@ func NewWebhookAdapter(clientID, clientSecret, cardTemplateID string) *Adapter { } } -// NewAdapter creates a DingTalk adapter backed by a stream client. -func NewAdapter(client *LongConnClient, clientID, clientSecret, cardTemplateID string) *Adapter { +// NewAdapter creates a DingTalk adapter for stream (websocket) mode. +// The stream connection itself is managed separately by the supervisor; the +// adapter only sends replies (via sessionWebhook or OpenAPI). +func NewAdapter(clientID, clientSecret, cardTemplateID string) *Adapter { startStreamReaper() return &Adapter{ clientID: clientID, clientSecret: clientSecret, cardTemplateID: cardTemplateID, - client: client, } } @@ -457,7 +457,7 @@ const ( var ( streamsMu sync.Mutex - dStreams = map[string]*streamState{} + dStreams = map[string]*streamState{} startReaperOnce sync.Once reaperStopCh = make(chan struct{}) ) diff --git a/internal/im/dingtalk/factory.go b/internal/im/dingtalk/factory.go index e911aea4..f36bb4d3 100644 --- a/internal/im/dingtalk/factory.go +++ b/internal/im/dingtalk/factory.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/Tencent/WeKnora/internal/im" - "github.com/Tencent/WeKnora/internal/logger" ) // NewFactory returns an im.AdapterFactory for DingTalk channels. @@ -29,16 +28,20 @@ func NewFactory() im.AdapterFactory { return adapter, nil, nil case "websocket": - client := NewLongConnClient(clientID, clientSecret, msgHandler) - wsCtx, wsCancel := context.WithCancel(context.Background()) - go func() { - if err := client.Start(wsCtx); err != nil && wsCtx.Err() == nil { - logger.Errorf(context.Background(), "[IM] DingTalk stream stopped for channel %s: %v", channel.ID, err) - } - }() + go im.RunSupervised(wsCtx, im.SupervisorConfig{ + Name: fmt.Sprintf("DingTalk channel %s", channel.ID), + Connect: func(ctx context.Context) (func(), error) { + client := NewLongConnClient(clientID, clientSecret, msgHandler) + if err := client.Start(ctx); err != nil { + client.Close() + return nil, err + } + return client.Close, nil + }, + }) - adapter := NewAdapter(client, clientID, clientSecret, cardTemplateID) + adapter := NewAdapter(clientID, clientSecret, cardTemplateID) return adapter, wsCancel, nil default: diff --git a/internal/im/dingtalk/longconn.go b/internal/im/dingtalk/longconn.go index 0ef103a1..0931b912 100644 --- a/internal/im/dingtalk/longconn.go +++ b/internal/im/dingtalk/longconn.go @@ -4,8 +4,8 @@ import ( "context" "strings" - dtsdk "github.com/open-dingtalk/dingtalk-stream-sdk-go/client" "github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot" + dtsdk "github.com/open-dingtalk/dingtalk-stream-sdk-go/client" "github.com/Tencent/WeKnora/internal/im" "github.com/Tencent/WeKnora/internal/logger" @@ -43,21 +43,19 @@ func NewLongConnClient(clientID, clientSecret string, handler MessageHandler) *L return c } -// Start begins the stream connection. It blocks until ctx is cancelled. +// Start establishes the stream connection. The underlying SDK's Start is +// non-blocking: it dials the websocket, spawns its internal read/reconnect +// loops, and returns once the connection is established (or the attempt fails). func (c *LongConnClient) Start(ctx context.Context) error { logger.Infof(ctx, "[IM] DingTalk Stream connecting...") + return c.streamClient.Start(ctx) +} - errCh := make(chan error, 1) - go func() { - errCh <- c.streamClient.Start(ctx) - }() - - select { - case <-ctx.Done(): - return ctx.Err() - case err := <-errCh: - return err - } +// Close tears down the stream connection. AutoReconnect is disabled first so +// that closing the connection does not trigger the SDK's internal reconnect. +func (c *LongConnClient) Close() { + c.streamClient.AutoReconnect = false + c.streamClient.Close() } func (c *LongConnClient) onChatBotMessage(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) { diff --git a/internal/im/supervisor.go b/internal/im/supervisor.go new file mode 100644 index 00000000..4c920b60 --- /dev/null +++ b/internal/im/supervisor.go @@ -0,0 +1,97 @@ +package im + +import ( + "context" + "time" + + "github.com/Tencent/WeKnora/internal/logger" +) + +// defaultRecycleInterval is how often a supervised long connection is proactively +// torn down and rebuilt. Some IM SDKs (e.g. DingTalk, Feishu) delegate reconnection +// to internal logic that can silently enter a "zombie" state on long-running +// connections — the connection object stays alive but no longer receives messages. +// Periodically recreating the connection bounds the worst-case outage to this +// interval, on top of whatever auto-reconnect the SDK already provides. +const defaultRecycleInterval = 6 * time.Hour + +// defaultSupervisorRetryDelay is the backoff applied after a failed connect attempt. +const defaultSupervisorRetryDelay = 5 * time.Second + +// SupervisorConfig configures RunSupervised. +type SupervisorConfig struct { + // Name identifies the supervised connection in logs (e.g. "DingTalk channel xxx"). + Name string + + // MaxConnAge is the interval between proactive reconnections. + // Defaults to defaultRecycleInterval when <= 0. + MaxConnAge time.Duration + + // RetryDelay is the backoff after a failed connect attempt. + // Defaults to defaultSupervisorRetryDelay when <= 0. + RetryDelay time.Duration + + // Connect establishes a fresh connection and returns a stop function that + // tears it down cleanly (preventing any further internal reconnects). + // It may wrap either a blocking or a non-blocking SDK Start: the only + // contract is that it returns once the connection has been established + // (or the attempt has failed), along with a stop function. + Connect func(ctx context.Context) (stop func(), err error) +} + +// RunSupervised manages the lifecycle of a long-lived IM connection. +// +// It keeps a connection alive by (re)establishing it via cfg.Connect, then +// proactively recycling it every cfg.MaxConnAge. The underlying SDK is still +// expected to handle transient drops via its own auto-reconnect; the periodic +// recycle is a safety net that eliminates stuck/zombie connections that the SDK +// fails to recover on its own. +// +// RunSupervised blocks until ctx is cancelled. On cancellation it tears down the +// active connection via the stop function before returning, so callers can rely +// on cancelling ctx to fully stop the connection. +func RunSupervised(ctx context.Context, cfg SupervisorConfig) { + maxAge := cfg.MaxConnAge + if maxAge <= 0 { + maxAge = defaultRecycleInterval + } + retryDelay := cfg.RetryDelay + if retryDelay <= 0 { + retryDelay = defaultSupervisorRetryDelay + } + + for { + if ctx.Err() != nil { + return + } + + stop, err := cfg.Connect(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + logger.Warnf(ctx, "[IM] %s connect failed: %v, retrying in %v", cfg.Name, err, retryDelay) + select { + case <-time.After(retryDelay): + continue + case <-ctx.Done(): + return + } + } + + logger.Infof(ctx, "[IM] %s connection established (recycle in %v)", cfg.Name, maxAge) + + select { + case <-time.After(maxAge): + logger.Infof(ctx, "[IM] %s periodic reconnect to refresh connection", cfg.Name) + if stop != nil { + stop() + } + case <-ctx.Done(): + if stop != nil { + stop() + } + return + } + } +}