Return {@code true} when the underlying transport rejects multiple + * concurrent connections from the same credentials — e.g. a bot WebSocket + * gateway that enforces a per-app connection cap, or a long-polling + * endpoint where multiple consumers would steal updates from each other. + * The channel manager will gate {@link #start()} on a distributed lease + * so only one node connects at a time, and failover to another node when + * the lease holder dies. + * + *
Webhook-based channels (DingTalk, WeCom, Slack, …) should leave this + * at the default {@code false}: inbound HTTP traffic is fanned out by the + * load balancer, so every node may safely subscribe. + * + *
Scope: this hook is honored by the framework for DB-backed
+ * channels registered via {@code ChannelManager.startChannel}. For
+ * plugin-registered channels the framework can only gate the initial
+ * register attempt — there is no follower retry, no hot-swap, and no
+ * disable-detection (plugins have a register/unregister lifecycle, not
+ * a DB-driven one). Plugin authors needing full single-leader semantics
+ * should depend on {@code ChannelLeaderElection} directly.
+ */
+ default boolean requiresSingleLeader() {
+ return false;
+ }
+
/**
* RFC-024 Change 2：本 adapter 认为"多久没活动就视作 stale 需要重启"的阈值。
*
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
index 5130c159..ca3bfffa 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
@@ -10,6 +10,8 @@ import org.springframework.stereotype.Component;
import vip.mate.channel.dingtalk.DingTalkChannelAdapter;
import vip.mate.channel.discord.DiscordChannelAdapter;
import vip.mate.channel.feishu.FeishuChannelAdapter;
+import vip.mate.channel.leader.ChannelLeaderElection;
+import vip.mate.channel.leader.LeaderLease;
import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.qq.QQChannelAdapter;
import vip.mate.channel.service.ChannelService;
@@ -17,7 +19,9 @@ import vip.mate.channel.telegram.TelegramChannelAdapter;
import vip.mate.channel.web.WebChannelAdapter;
import vip.mate.channel.wecom.WeComChannelAdapter;
import vip.mate.channel.weixin.WeixinChannelAdapter;
+import vip.mate.exception.MateClawException;
+import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReadWriteLock;
@@ -71,12 +75,66 @@ public class ChannelManager {
*/
private final vip.mate.channel.wecom.WeComKeepaliveScheduler weComKeepaliveScheduler;
+ /**
+ * Distributed leader election. Channels whose adapter reports
+ * {@link ChannelAdapter#requiresSingleLeader()} are gated on a lease so
+ * only one node opens the upstream WebSocket / long-poll at a time.
+ */
+ private final ChannelLeaderElection leaderElection;
+
/** 运行中的渠道适配器：channelId -> adapter */
private final Map Caller must hold the adapter write lock.
+ */
+ private void attemptLeaderStart(ChannelEntity channel, ChannelAdapter adapter) {
+ String key = channel.getChannelType() + ":" + channel.getId();
+ Optional Package-private for unit testing — callers should rely on the
+ * heartbeat scheduler invoking this on its tick.
+ */
+ void reconcileChannel(Long channelId, String channelName) {
+ ChannelEntity current;
+ try {
+ current = channelService.getChannel(channelId);
+ } catch (MateClawException e) {
+ if (e.getMsgKey() != null && e.getMsgKey().startsWith("err.channel.not_found")) {
+ log.info("[reconcile] Channel id={} no longer exists — stopping local adapter and releasing lease",
+ channelId);
+ stopChannel(channelId);
+ } else {
+ log.debug("[reconcile] Channel lookup failed for id={}: {}", channelId, e.getMessage());
+ }
+ return;
+ } catch (Exception e) {
+ // Transient DB issue; skip this tick and try again on the next heartbeat.
+ log.debug("[reconcile] Channel lookup failed for id={}: {}", channelId, e.getMessage());
+ return;
+ }
+
+ if (!Boolean.TRUE.equals(current.getEnabled())) {
+ log.info("[reconcile] Channel {} (id={}) is now disabled — stopping local adapter",
+ channelName, channelId);
+ stopChannel(channelId);
+ return;
+ }
+
+ LocalDateTime previousSeen;
+ adapterLock.readLock().lock();
+ try {
+ previousSeen = lastSeenChannelUpdateTime.get(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ LocalDateTime currentUpdateTime = current.getUpdateTime();
+ if (currentUpdateTime != null && previousSeen != null
+ && currentUpdateTime.isAfter(previousSeen)) {
+ log.info("[reconcile] Channel {} (id={}) config changed ({} → {})",
+ channelName, channelId, previousSeen, currentUpdateTime);
+ applyConfigChange(channelId, current);
+ }
+ }
+
+ /**
+ * Apply a detected config change to a locally-running channel.
+ *
+ * The fast path is the in-place swap that preserves the lease —
+ * but it is only valid when we are the current leader (we already
+ * hold {@code activeLeases[channelId]}). Without that gate, a
+ * non-leader node observing a {@code webhook → websocket} flip
+ * would call {@code newAdapter.start()} directly inside the swap
+ * and open a duplicate upstream connection, defeating the leader
+ * election. Every other transition — including
+ * {@code non-leader → leader-required}, {@code leader-required →
+ * non-leader}, and plain non-leader config updates — must go
+ * through {@code stopChannel} + {@code startChannel} so the lease
+ * is correctly released or acquired and follower retry is
+ * scheduled when election is lost.
+ *
+ * Package-private for unit testing — see {@link #reconcileChannel}.
+ */
+ void applyConfigChange(Long channelId, ChannelEntity newChannel) {
+ ChannelAdapter probe = createAdapter(newChannel);
+ boolean newRequiresLeader = probe.requiresSingleLeader();
+ boolean weHaveLease;
+ adapterLock.readLock().lock();
+ try {
+ weHaveLease = activeLeases.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+
+ if (newRequiresLeader && weHaveLease) {
+ // Case A: same leader-required mode and we are the current leader
+ // — preserve the lease across the adapter swap.
+ swapAdapterPreservingLease(channelId, newChannel);
+ return;
+ }
+
+ // All other cases: tear down local state and route through
+ // startChannel so leader election runs, lease is released, or both.
+ log.info("[reconcile] Channel {} (id={}) config change (newRequiresLeader={}, weHaveLease={}) — stop+start",
+ newChannel.getName(), channelId, newRequiresLeader, weHaveLease);
+ stopChannel(channelId);
+ try {
+ startChannel(newChannel);
+ } catch (Exception e) {
+ log.error("[reconcile] Restart after config change failed for channel {} (id={}): {}",
+ newChannel.getName(), channelId, e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Swap to a freshly-built adapter using the new config, while keeping
+ * the leadership lease and heartbeat in place. The lease is only
+ * released if the new adapter fails to start, in which case we fall
+ * back to follower mode so another node can try.
+ */
+ private void swapAdapterPreservingLease(Long channelId, ChannelEntity newChannel) {
+ ChannelAdapter oldAdapter;
+ adapterLock.writeLock().lock();
+ try {
+ oldAdapter = activeAdapters.remove(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ if (oldAdapter != null) {
+ stopAdapterSafely(oldAdapter, "reconcile-swap");
+ }
+
+ ChannelAdapter newAdapter = createAdapter(newChannel);
+ boolean started = false;
+ Exception startError = null;
+ try {
+ newAdapter.start();
+ started = true;
+ } catch (Exception e) {
+ startError = e;
+ }
+
+ LeaderLease leaseToRelease = null;
+ adapterLock.writeLock().lock();
+ try {
+ if (started) {
+ activeAdapters.put(channelId, newAdapter);
+ lastSeenChannelUpdateTime.put(channelId, newChannel.getUpdateTime());
+ } else {
+ leaseToRelease = activeLeases.remove(channelId);
+ lastSeenChannelUpdateTime.remove(channelId);
+ cancelHeartbeatLocked(channelId);
+ scheduleFollowerRetryLocked(channelId);
+ }
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+
+ if (!started) {
+ log.error("[reconcile] New adapter start failed for channel {} (id={}): {} — released lease, entering follower mode",
+ newChannel.getName(), channelId,
+ startError != null ? startError.getMessage() : "unknown");
+ if (leaseToRelease != null) {
+ leaseToRelease.release();
+ }
+ }
+ }
+
+ /**
+ * Drop the local adapter (without releasing the already-lost lease)
+ * and start follower retry so we'll attempt to reclaim leadership
+ * once the current owner stops renewing.
+ */
+ private void handleLeadershipLoss(Long channelId) {
+ ChannelAdapter local;
+ adapterLock.writeLock().lock();
+ try {
+ local = activeAdapters.remove(channelId);
+ activeLeases.remove(channelId); // already lost; do not call release()
+ cancelHeartbeatLocked(channelId);
+ scheduleFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ if (local != null) {
+ stopAdapterSafely(local, "leadership-loss");
+ }
+ }
+
+ /**
+ * Schedule periodic follower retry. Caller must hold the adapter
+ * write lock.
+ */
+ private void scheduleFollowerRetryLocked(Long channelId) {
+ if (followerRetryFutures.containsKey(channelId)) {
+ return;
+ }
+ ScheduledFuture f = leaderScheduler.scheduleAtFixedRate(
+ () -> followerRetry(channelId),
+ FOLLOWER_RETRY_INTERVAL_SECONDS, FOLLOWER_RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
+ followerRetryFutures.put(channelId, f);
+ }
+
+ /**
+ * One follower-retry tick. Re-reads the channel from the DB (it may
+ * have been disabled or deleted) and attempts to start it. The retry
+ * cancels itself once we successfully become leader.
+ */
+ /** Package-private for unit testing — see {@link #reconcileChannel}. */
+ void followerRetry(Long channelId) {
+ ChannelEntity current;
+ try {
+ current = channelService.getChannel(channelId);
+ } catch (MateClawException e) {
+ // Channel was deleted on another node — cancel the retry so we
+ // don't leak a scheduled task forever. Other exception codes
+ // (e.g. transient DB errors) fall through to the generic catch
+ // and let the retry continue.
+ if (e.getMsgKey() != null && e.getMsgKey().startsWith("err.channel.not_found")) {
+ log.info("[leader] Follower retry: channel id={} no longer exists — cancelling retry", channelId);
+ adapterLock.writeLock().lock();
+ try {
+ cancelFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ return;
+ }
+ log.debug("[leader] Follower retry: lookup failed for channel id={}: {}", channelId, e.getMessage());
+ return;
+ } catch (Exception e) {
+ log.debug("[leader] Follower retry: lookup failed for channel id={}: {}", channelId, e.getMessage());
+ return;
+ }
+ if (!Boolean.TRUE.equals(current.getEnabled())) {
+ adapterLock.writeLock().lock();
+ try {
+ cancelFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ return;
+ }
+ try {
+ startChannel(current);
+ } catch (Exception e) {
+ log.debug("[leader] Follower retry: startChannel failed for id={}: {}", channelId, e.getMessage());
+ }
+ }
+
+ /** Package-private for unit testing. */
+ boolean hasFollowerRetry(Long channelId) {
+ adapterLock.readLock().lock();
+ try {
+ return followerRetryFutures.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ }
+
+ private void cancelHeartbeatLocked(Long channelId) {
+ ScheduledFuture f = heartbeatFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
+ }
+
+ private void cancelFollowerRetryLocked(Long channelId) {
+ ScheduledFuture f = followerRetryFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
+ }
+
+ /**
+ * Schedule the cross-node reconcile ticker for a non-leader active
+ * adapter. Caller must hold the adapter write lock.
+ */
+ private void scheduleReconcileLocked(Long channelId, String channelName) {
+ cancelReconcileLocked(channelId);
+ ScheduledFuture f = leaderScheduler.scheduleAtFixedRate(
+ () -> reconcileChannel(channelId, channelName),
+ FOLLOWER_RETRY_INTERVAL_SECONDS, FOLLOWER_RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
+ reconcileFutures.put(channelId, f);
+ }
+
+ private void cancelReconcileLocked(Long channelId) {
+ ScheduledFuture f = reconcileFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
}
/**
@@ -186,6 +657,50 @@ public class ChannelManager {
return;
}
+ // Leader-required channels: if we don't already own the local adapter
+ // (i.e. we are a follower, or this is a brand-new channel), there is
+ // nothing to hot-swap. Fall through to startChannel which handles
+ // lease acquisition and follower retry. Hot-swap (which briefly opens
+ // a second upstream connection) is also avoided here so we don't
+ // double-occupy the bot's connection quota during a restart.
+ boolean weHaveAdapter;
+ boolean weHaveLease;
+ adapterLock.readLock().lock();
+ try {
+ weHaveAdapter = activeAdapters.containsKey(channelId);
+ weHaveLease = activeLeases.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ ChannelAdapter probe = createAdapter(channel);
+ if (probe.requiresSingleLeader()) {
+ log.info("[hot-swap] Channel {} requires single-leader; stop+start instead of hot-swap (weHaveAdapter={}, weHaveLease={})",
+ channel.getName(), weHaveAdapter, weHaveLease);
+ stopChannel(channelId);
+ startChannel(channel);
+ return;
+ }
+ // Mode flip: we currently hold a lease but the new config is no
+ // longer leader-required (e.g. Feishu WS → webhook). The lease,
+ // heartbeat, and lastSeenUpdateTime must all be torn down before
+ // the new non-leader adapter starts — the in-place hot-swap path
+ // below would leave them behind until the next heartbeat tick
+ // noticed and re-restarted, causing a redundant restart and a
+ // window where this node is silently holding a lease nobody else
+ // can grab. Stop+start handles all the cleanup in one shot.
+ if (weHaveLease) {
+ log.info("[hot-swap] Channel {} flipping leader-required → non-leader; stop+start to release lease",
+ channel.getName());
+ stopChannel(channelId);
+ startChannel(channel);
+ return;
+ }
+ if (!weHaveAdapter) {
+ log.info("[hot-swap] No local adapter for channel {}, delegating to startChannel", channel.getName());
+ startChannel(channel);
+ return;
+ }
+
log.info("[hot-swap] Starting hot-swap for channel: {} (type={}, id={})",
channel.getName(), channel.getChannelType(), channelId);
@@ -207,6 +722,10 @@ public class ChannelManager {
adapterLock.writeLock().lock();
try {
oldAdapter = activeAdapters.put(channelId, newAdapter);
+ // Mark the version we've now applied so the reconcile ticker
+ // (running for non-leader adapters) doesn't immediately fire a
+ // redundant swap on its next tick.
+ lastSeenChannelUpdateTime.put(channelId, channel.getUpdateTime());
log.info("[hot-swap] Adapter reference swapped for channel: {} (old={})",
channel.getName(), oldAdapter != null ? "present" : "none");
} finally {
@@ -228,17 +747,65 @@ public class ChannelManager {
*/
public void stopAll() {
List If the adapter reports {@link ChannelAdapter#requiresSingleLeader()},
+ * the framework gates the local register on a distributed lease keyed by
+ * {@code plugin:{pluginName}}. When another node already owns the lease
+ * this node skips registration (its plugin instance is loaded but inert
+ * locally) — see the scope note on {@link ChannelAdapter#requiresSingleLeader()}.
+ *
* @param pluginName the plugin name (used as key for unregistration)
* @param adapter the channel adapter
*/
public void registerPluginChannel(String pluginName, ChannelAdapter adapter) {
- try {
- adapter.start();
- pluginChannels.put(pluginName, adapter);
- log.info("Plugin channel registered: {} (type={})", pluginName, adapter.getChannelType());
- } catch (Exception e) {
- log.error("Failed to start plugin channel {}: {}", pluginName, e.getMessage(), e);
+ synchronized (pluginLifecycleLock) {
+ LeaderLease lease = null;
+ if (adapter.requiresSingleLeader()) {
+ Optional Package-private + non-final so unit tests can substitute a stub
+ * adapter without spinning up real WebSocket / HTTP clients.
*/
- private ChannelAdapter createAdapter(ChannelEntity channel) {
+ ChannelAdapter createAdapter(ChannelEntity channel) {
String type = channel.getChannelType();
return switch (type) {
case "web" -> new WebChannelAdapter(channel, messageRouter, objectMapper);
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java
index 8fb17b6a..86658eea 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java
@@ -373,6 +373,18 @@ public class DiscordChannelAdapter extends AbstractChannelAdapter {
return CHANNEL_TYPE;
}
+ /**
+ * Discord enforces a single Gateway session per bot token (any duplicate
+ * {@code IDENTIFY} closes the previous shard) and there is no active
+ * webhook fallback in this adapter — the legacy webhook endpoint is
+ * a no-op. The leader gate ensures only one node holds the Gateway
+ * connection at a time.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return true;
+ }
+
// ==================== Webhook 兼容（保留接口，不再使用） ====================
/**
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
index 602251bf..992937ed 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
@@ -1435,4 +1435,19 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter {
public String getChannelType() {
return CHANNEL_TYPE;
}
+
+ /**
+ * WebSocket mode opens a long-lived connection to Lark's gateway, which
+ * caps concurrent connections per bot app (~2). In a multi-instance
+ * deployment every node would race for that quota and reconnect-loop on
+ * {@code 1000040350: the number of connections exceeded the limit}.
+ * The leader gate ensures only one node holds the connection at a time.
+ *
+ * Webhook mode is exempt: callbacks are HTTP-fanned by the load
+ * balancer, so all nodes can safely subscribe.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return "websocket".equals(getConfigString("connection_mode", "websocket"));
+ }
}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java b/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java
new file mode 100644
index 00000000..8951cb08
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java
@@ -0,0 +1,84 @@
+package vip.mate.channel.leader;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import net.javacrumbs.shedlock.core.LockConfiguration;
+import net.javacrumbs.shedlock.core.LockProvider;
+import net.javacrumbs.shedlock.core.SimpleLock;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+
+/**
+ * Distributed leader election for channel adapters whose upstream
+ * service rejects multiple concurrent connections from the same bot
+ * credentials.
+ *
+ * Typical examples are WebSocket-mode IM channels: Feishu's Lark
+ * SDK enforces a per-app connection cap (~2) and QQ's bot gateway
+ * rejects duplicate {@code IDENTIFY} sessions. Without coordination,
+ * every node of a multi-instance deployment trips that cap on startup
+ * and reconnects in a tight loop.
+ *
+ * Backed by ShedLock's {@link LockProvider} (already wired for cron
+ * coordination), so single-node deployments incur no extra
+ * infrastructure — the lock is acquired trivially on the only node.
+ *
+ * Semantics:
+ * Wraps a ShedLock {@link SimpleLock} so callers don't depend on the
+ * underlying lock provider. The lease must be periodically extended via
+ * {@link #extend(Duration)} or it expires automatically, at which point
+ * another node can claim leadership.
+ *
+ * Threading: a single lease instance is not safe for concurrent
+ * {@link #extend(Duration)} / {@link #release()} calls. The owning
+ * scheduler is expected to serialize them.
+ */
+@Slf4j
+public class LeaderLease {
+
+ private final String name;
+ private volatile SimpleLock current;
+ private volatile boolean released;
+
+ LeaderLease(String name, SimpleLock initial) {
+ this.name = name;
+ this.current = initial;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Try to extend this lease for another {@code lockAtMostFor} window.
+ *
+ * @return true if the lease is still ours; false if it has been lost
+ * (e.g. the previous window expired before extend ran and
+ * another node acquired the lock — the caller should treat
+ * this as a leadership loss and stop the protected resource).
+ */
+ public boolean extend(Duration lockAtMostFor) {
+ if (released) {
+ return false;
+ }
+ try {
+ Optional Package-private so {@link #requiresSingleLeader()} can mirror the
+ * same predicate — the two answers must stay in lockstep, otherwise a
+ * single change to mode detection here would silently mis-classify the
+ * channel for multi-node coordination.
*/
- private boolean resolveWebhookMode() {
+ boolean resolveWebhookMode() {
String connectionMode = getConfigString("connection_mode");
String webhookUrl = getConfigString("webhook_url");
boolean hasWebhookUrl = webhookUrl != null && !webhookUrl.isBlank();
@@ -794,6 +799,21 @@ public class TelegramChannelAdapter extends AbstractChannelAdapter {
return CHANNEL_TYPE;
}
+ /**
+ * Long-polling mode runs a {@code getUpdates(offset=…)} loop that
+ * acknowledges each delivered update; multiple nodes polling the same
+ * bot token would steal updates from each other (whichever node calls
+ * {@code getUpdates} next consumes the queue, and the others get
+ * nothing). The leader gate ensures only one node polls at a time.
+ *
+ * Webhook mode is exempt: Telegram POSTs to a public URL fanned
+ * by the load balancer, so every node may safely receive callbacks.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return !resolveWebhookMode();
+ }
+
/** RFC-025 Change 4 入站文本净化上限（防止 caption 含超长二进制撑爆 prompt）。 */
private static final int INBOUND_TEXT_MAX = 4096;
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
index 6c33e208..aa27f7f8 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
@@ -3300,6 +3300,20 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
return CHANNEL_TYPE;
}
+ /**
+ * WeCom's AI-bot transport is WebSocket-only ({@code wss://openws.work.weixin.qq.com}
+ * with {@code aibot_subscribe} authenticated by {@code bot_id + secret}) — there is
+ * no HTTP webhook fallback. Multiple nodes subscribing with the same
+ * credentials would each receive every {@code aibot_msg_callback} and race
+ * to send {@code aibot_respond_msg}, producing duplicate replies and
+ * eventual gateway rejection. The leader gate ensures only one node
+ * holds the subscription at a time.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return true;
+ }
+
// ==================== 工具方法 ====================
/**
+ *
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ChannelLeaderElection {
+
+ /**
+ * How long the lock is held without a renewal. A failed node's lease
+ * stays locked for this long before another node can take over —
+ * so longer values increase failover latency, shorter values increase
+ * the risk of false handover during a GC pause or DB hiccup.
+ */
+ public static final Duration LOCK_AT_MOST_FOR = Duration.ofSeconds(60);
+
+ private final LockProvider lockProvider;
+
+ /**
+ * Attempt to acquire leadership for the given key.
+ *
+ * @param key a stable identifier for the resource (e.g.
+ * {@code "feishu:42"}). Used verbatim as the underlying
+ * lock name (prefixed by this class to avoid collisions
+ * with other lock users).
+ * @return an empty optional if another node already holds the lease,
+ * otherwise a {@link LeaderLease} that the caller is
+ * responsible for periodically extending and finally
+ * releasing.
+ */
+ public Optional