xiaoski/PicoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(feishu): harden websocket event processing

Browse Source
This commit is contained in:
xiaoxixi 2026-07-19 15:47:27 +08:00
parent 5d081e2580
commit 0512d91729
2 changed files with 320 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/ARCHITECTURE.md
View File

@ -139,6 +139,7 @@ sequenceDiagram
- `LivePolicy::Snapshot` 按渠道间隔发送最新运行态；`FinalOnly` 忽略运行态，只处理终态。终态绕过节流并只对明确的瞬态错误重试。
- `TurnDeliveryService` 返回可等待的终态句柄；sink 生命周期启动不等于终态已送达。Session 在终态重试最终失败时通过普通出站路径兜底一次。
- `cli_chat` 将同一 `turn_updated` 快照发给 TUI 和 WebUI。客户端只保留当前 session 中 revision 更新的 `active_turn`，终态随后由持久化历史校准。
- 飞书对每个 DATA 帧先在 2 秒硬期限内 ACK，再进行有界分片重组，并把完整事件交给容量 32 的连接内处理队列；媒体下载和引用查询不占用正常的 WebSocket 读循环。队列饱和时当前事件在连接任务中同步处理而不丢弃。连接异常采用有上限的指数退避持续重连，不因累计故障永久停止。
- 飞书在协议解析阶段按 `allow_from` 拒绝未授权用户；群聊默认必须明确 @ 运行时解析出的机器人身份，身份解析失败时安全地忽略群消息。飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。
- DeliveryCoordinator 与 OutboundDispatcher 共享 `(channel, chat_id)` 写锁，避免活动 Turn 终态与独立消息并发写入同一目标。

381
src/channels/feishu.rs
View File

@ -33,6 +33,12 @@ const TOKEN_REFRESH_SKEW: Duration = Duration::from_secs(120);
const DEFAULT_TOKEN_TTL: Duration = Duration::from_secs(7200);
/// Dedup cache TTL (30 minutes).
const DEDUP_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
const WS_EVENT_QUEUE_CAPACITY: usize = 32;
const WS_ACK_TIMEOUT: Duration = Duration::from_secs(2);
const WS_FRAGMENT_TTL: Duration = Duration::from_secs(5 * 60);
const MAX_WS_FRAGMENTS: usize = 256;
const MAX_WS_EVENT_BYTES: usize = 8 * 1024 * 1024;
const STABLE_CONNECTION_WINDOW: Duration = Duration::from_secs(30);
// ─────────────────────────────────────────────────────────────────────────────
// Protobuf types for Feishu WebSocket protocol (pbbp2.proto)
@ -64,6 +70,21 @@ struct PbFrame {
pub payload: Option<Vec<u8>>,
}
impl PbFrame {
fn header_value(&self, key: &str) -> Option<&str> {
self.headers
.iter()
.find(|header| header.key == key)
.map(|header| header.value.as_str())
}
}
struct FragmentAssembly {
parts: Vec<Option<Vec<u8>>>,
total_bytes: usize,
created_at: Instant,
}
/// POST /callback/ws/endpoint response
#[derive(Deserialize)]
struct WsEndpointResp {
@ -974,7 +995,7 @@ impl FeishuChannel {
.unwrap_or(0)
}
/// Handle incoming binary PbFrame - returns Some(ParsedMessage) if we need to ack
/// Parse one complete event frame. Transport ACK and fragment handling happen earlier.
async fn handle_frame(&self, frame: &PbFrame) -> Result<Option<ParsedMessage>, ChannelError> {
// method 0 = CONTROL (ping/pong)
if frame.method == 0 {
@ -1202,6 +1223,54 @@ impl FeishuChannel {
Ok((text, media))
}
async fn process_event_frame(&self, bus: &Arc<MessageBus>, frame: &PbFrame) {
let parsed = match self.handle_frame(frame).await {
Ok(Some(parsed)) => parsed,
Ok(None) => return,
Err(error) => {
tracing::warn!(error = %error, "Failed to parse Feishu frame");
return;
}
};
let message_id = parsed.message_id.clone();
let reaction_id = match self.add_reaction(&message_id).await {
Ok(Some(reaction_id)) => Some(reaction_id),
Ok(None) => None,
Err(error) => {
tracing::debug!(error = %error, message_id = %message_id, "Failed to add reaction");
None
}
};
let mut private_context = HashMap::new();
private_context.insert("feishu.message_id".to_string(), message_id);
if let Some(reaction_id) = reaction_id {
private_context.insert("feishu.reaction_id".to_string(), reaction_id);
}
let msg = crate::bus::InboundMessage {
channel: "feishu".to_string(),
sender_id: parsed.open_id.clone(),
chat_id: parsed.chat_id.clone(),
content: parsed.content,
received_at: crate::bus::message::current_timestamp(),
media: parsed.media,
channel_context: crate::bus::ChannelContext {
reply_to: parsed.parent_id,
private: private_context,
},
};
if let Err(error) = self.handle_and_publish(bus, &msg).await {
tracing::error!(
error = %error,
open_id = %parsed.open_id,
chat_id = %parsed.chat_id,
"Failed to publish Feishu message to bus"
);
}
}
/// Send acknowledgment for a message
async fn send_ack(
frame: &PbFrame,
@ -1231,21 +1300,22 @@ impl FeishuChannel {
&self,
bus: Arc<MessageBus>,
shutdown: CancellationToken,
) -> Result<(), ChannelError> {
) -> Result<Duration, ChannelError> {
let (wss_url, client_config) = tokio::select! {
result = self.get_ws_endpoint(&self.http_client) => result?,
_ = shutdown.cancelled() => return Ok(()),
_ = shutdown.cancelled() => return Ok(Duration::ZERO),
};
let service_id = Self::extract_service_id(&wss_url);
tracing::info!(url = %wss_url, "Connecting to Feishu WebSocket");
tracing::info!(service_id, "Connecting to Feishu WebSocket");
let (ws_stream, _) = tokio::select! {
result = tokio_tungstenite::connect_async(&wss_url) => result.map_err(|e| {
ChannelError::ConnectionError(format!("WebSocket connection failed: {}", e))
})?,
_ = shutdown.cancelled() => return Ok(()),
_ = shutdown.cancelled() => return Ok(Duration::ZERO),
};
let connected_at = Instant::now();
*self.connected.write().await = true;
tracing::info!("Feishu WebSocket connected");
@ -1270,15 +1340,27 @@ impl FeishuChannel {
)) => result.map_err(|e| {
ChannelError::ConnectionError(format!("Failed to send initial ping: {}", e))
})?,
_ = shutdown.cancelled() => return Ok(()),
_ = shutdown.cancelled() => return Ok(connected_at.elapsed()),
};
let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(WS_EVENT_QUEUE_CAPACITY);
let worker_channel = self.clone();
let worker_bus = bus.clone();
let mut event_worker = tokio::spawn(async move {
while let Some(frame) = event_rx.recv().await {
worker_channel
.process_event_frame(&worker_bus, &frame)
.await;
}
});
let ping_interval = client_config.ping_interval.unwrap_or(120).max(10);
let mut ping_interval_tok =
tokio::time::interval(tokio::time::Duration::from_secs(ping_interval));
let mut timeout_check = tokio::time::interval(tokio::time::Duration::from_secs(10));
let mut seq: u64 = 1;
let mut last_recv = Instant::now();
let mut fragment_cache: HashMap<String, FragmentAssembly> = HashMap::new();
// Consume the immediate tick
ping_interval_tok.tick().await;
@ -1292,54 +1374,46 @@ impl FeishuChannel {
last_recv = Instant::now();
let bytes: Bytes = data;
if let Ok(frame) = PbFrame::decode(bytes.as_ref()) {
match self.handle_frame(&frame).await {
Ok(Some(parsed)) => {
// Send ACK immediately (Feishu requires within 3 s)
if let Err(e) = Self::send_ack(&frame, &mut write).await {
tracing::error!(error = %e, "Failed to send ACK to Feishu");
}
if frame.method != 1 {
continue;
}
match tokio::time::timeout(
WS_ACK_TIMEOUT,
Self::send_ack(&frame, &mut write),
)
.await
{
Ok(Ok(())) => {}
Ok(Err(error)) => {
tracing::warn!(error = %error, "Failed to ACK Feishu DATA frame");
break;
}
Err(_) => {
tracing::warn!("Timed out ACKing Feishu DATA frame");
break;
}
}
// Add reaction emoji (await so we get the reaction_id for later removal)
let message_id = parsed.message_id.clone();
let reaction_id = match self.add_reaction(&message_id).await {
Ok(Some(rid)) => Some(rid),
Ok(None) => None,
Err(e) => {
tracing::debug!(error = %e, message_id = %message_id, "Failed to add reaction");
None
match reassemble_data_frame(&frame, &mut fragment_cache) {
Ok(Some(complete_frame)) => {
match event_tx.try_send(complete_frame) {
Ok(()) => {}
Err(tokio::sync::mpsc::error::TrySendError::Full(frame)) => {
tracing::warn!(
capacity = WS_EVENT_QUEUE_CAPACITY,
"Feishu event queue is full; processing one event on the connection task"
);
self.process_event_frame(&bus, &frame).await;
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
tracing::warn!("Feishu event worker stopped unexpectedly");
break;
}
};
let mut private_context = std::collections::HashMap::new();
private_context.insert("feishu.message_id".to_string(), message_id.clone());
if let Some(ref rid) = reaction_id {
private_context.insert("feishu.reaction_id".to_string(), rid.clone());
}
#[cfg(debug_assertions)]
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, content_len = %parsed.content.len(), media_count = %parsed.media.len(), "Publishing message to bus");
let msg = crate::bus::InboundMessage {
channel: "feishu".to_string(),
sender_id: parsed.open_id.clone(),
chat_id: parsed.chat_id.clone(),
content: parsed.content.clone(),
received_at: crate::bus::message::current_timestamp(),
media: parsed.media.clone(),
channel_context: crate::bus::ChannelContext {
reply_to: parsed.parent_id.clone(),
private: private_context,
},
};
if let Err(e) = self.handle_and_publish(&bus, &msg).await {
tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus");
} else {
#[cfg(debug_assertions)]
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Message published to bus successfully");
}
}
Ok(None) => {}
Err(e) => {
tracing::warn!(error = %e, "Failed to parse Feishu frame");
Err(error) => {
tracing::warn!(error = %error, "Rejected invalid Feishu fragment sequence");
}
}
}
@ -1401,6 +1475,9 @@ impl FeishuChannel {
let now = Instant::now();
let mut seen = self.seen_message_ids.write().await;
seen.retain(|_, ts| now.duration_since(*ts) < DEDUP_CACHE_TTL);
fragment_cache.retain(|_, assembly| {
now.duration_since(assembly.created_at) < WS_FRAGMENT_TTL
});
}
_ = shutdown.cancelled() => {
tracing::info!("Feishu channel shutdown signal received");
@ -1410,10 +1487,106 @@ impl FeishuChannel {
}
*self.connected.write().await = false;
Ok(())
drop(event_tx);
if tokio::time::timeout(CHANNEL_STOP_GRACE, &mut event_worker)
.await
.is_err()
{
event_worker.abort();
let _ = event_worker.await;
}
Ok(connected_at.elapsed())
}
}
fn reassemble_data_frame(
frame: &PbFrame,
cache: &mut HashMap<String, FragmentAssembly>,
) -> Result<Option<PbFrame>, ChannelError> {
let sum = frame
.header_value("sum")
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or(1)
.max(1);
if sum == 1 {
if frame
.payload
.as_ref()
.is_some_and(|payload| payload.len() > MAX_WS_EVENT_BYTES)
{
return Err(ChannelError::Other(format!(
"Feishu event exceeds {MAX_WS_EVENT_BYTES} byte limit"
)));
}
return Ok(Some(frame.clone()));
}
if sum > MAX_WS_FRAGMENTS {
return Err(ChannelError::Other(format!(
"Feishu event declares too many fragments: {sum}"
)));
}
let message_id = frame.header_value("message_id").unwrap_or_default();
if message_id.is_empty() {
return Err(ChannelError::Other(
"Fragmented Feishu event has no message_id".to_string(),
));
}
let seq = frame
.header_value("seq")
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or(0);
if seq >= sum {
return Err(ChannelError::Other(format!(
"Feishu fragment index {seq} is outside declared count {sum}"
)));
}
let assembly = cache
.entry(message_id.to_string())
.or_insert_with(|| FragmentAssembly {
parts: vec![None; sum],
total_bytes: 0,
created_at: Instant::now(),
});
if assembly.parts.len() != sum {
*assembly = FragmentAssembly {
parts: vec![None; sum],
total_bytes: 0,
created_at: Instant::now(),
};
}
let payload = frame.payload.clone().unwrap_or_default();
let previous_len = assembly.parts[seq].as_ref().map_or(0, Vec::len);
assembly.total_bytes = assembly.total_bytes - previous_len + payload.len();
if assembly.total_bytes > MAX_WS_EVENT_BYTES {
cache.remove(message_id);
return Err(ChannelError::Other(format!(
"Feishu fragmented event exceeds {MAX_WS_EVENT_BYTES} byte limit"
)));
}
assembly.parts[seq] = Some(payload);
if assembly.parts.iter().any(Option::is_none) {
return Ok(None);
}
let complete_payload = assembly
.parts
.iter()
.flat_map(|part| part.as_deref().unwrap_or_default())
.copied()
.collect();
cache.remove(message_id);
let mut complete = frame.clone();
complete.payload = Some(complete_payload);
Ok(Some(complete))
}
fn reconnect_delay(attempt: u32) -> Duration {
let exponent = attempt.saturating_sub(1).min(6);
Duration::from_secs(1_u64 << exponent)
}
fn parse_post_content(content: &str) -> String {
/// Extract text from a single post element (text, link, at-mention).
fn extract_element(el: &serde_json::Value, out: &mut Vec<String>) {
@ -2317,8 +2490,7 @@ impl Channel for FeishuChannel {
let channel = self.clone();
let bus = bus.clone();
*run_task = Some(tokio::spawn(async move {
let mut consecutive_failures = 0;
let max_failures = 3;
let mut retry_attempt = 0_u32;
loop {
if !*channel.running.read().await {
@ -2326,16 +2498,20 @@ impl Channel for FeishuChannel {
}
match channel.run_ws_loop(bus.clone(), shutdown.clone()).await {
Ok(_) => {
tracing::info!("Feishu WebSocket disconnected");
Ok(connected_for) => {
if connected_for >= STABLE_CONNECTION_WINDOW {
retry_attempt = 0;
} else {
retry_attempt = retry_attempt.saturating_add(1).min(7);
}
tracing::info!(
connected_secs = connected_for.as_secs(),
"Feishu WebSocket disconnected"
);
}
Err(e) => {
consecutive_failures += 1;
tracing::error!(attempt = consecutive_failures, error = %e, "Feishu WebSocket error");
if consecutive_failures >= max_failures {
tracing::error!("Feishu channel: max failures reached, stopping");
break;
}
retry_attempt = retry_attempt.saturating_add(1).min(6);
tracing::error!(attempt = retry_attempt, error = %e, "Feishu WebSocket error");
}
}
@ -2343,9 +2519,13 @@ impl Channel for FeishuChannel {
break;
}
tracing::info!("Feishu channel retrying in 5s...");
let retry_delay = reconnect_delay(retry_attempt);
tracing::info!(
delay_secs = retry_delay.as_secs(),
"Feishu channel reconnect scheduled"
);
tokio::select! {
_ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {}
_ = tokio::time::sleep(retry_delay) => {}
_ = shutdown.cancelled() => break,
}
}
@ -2799,6 +2979,83 @@ mod tests {
assert!(!message_mentions_bot(&message, Some("ou_other")));
}
fn fragment_frame(message_id: &str, seq: usize, sum: usize, payload: &[u8]) -> PbFrame {
PbFrame {
seq_id: seq as u64,
log_id: 1,
service: 1,
method: 1,
headers: vec![
PbHeader {
key: "message_id".to_string(),
value: message_id.to_string(),
},
PbHeader {
key: "seq".to_string(),
value: seq.to_string(),
},
PbHeader {
key: "sum".to_string(),
value: sum.to_string(),
},
],
payload: Some(payload.to_vec()),
}
}
#[test]
fn fragmented_frames_are_reassembled_in_sequence_order() {
let mut cache = HashMap::new();
assert!(
reassemble_data_frame(&fragment_frame("event", 1, 3, b"world"), &mut cache)
.unwrap()
.is_none()
);
assert!(
reassemble_data_frame(&fragment_frame("event", 0, 3, b"hello "), &mut cache)
.unwrap()
.is_none()
);
let complete = reassemble_data_frame(&fragment_frame("event", 2, 3, b"!"), &mut cache)
.unwrap()
.expect("last fragment should complete the event");
assert_eq!(
complete.payload.as_deref(),
Some(b"hello world!".as_slice())
);
assert!(cache.is_empty());
}
#[test]
fn fragmented_frames_reject_invalid_shape_and_bound_total_size() {
let mut cache = HashMap::new();
assert!(
reassemble_data_frame(
&fragment_frame("event", MAX_WS_FRAGMENTS, MAX_WS_FRAGMENTS, b"x"),
&mut cache,
)
.is_err()
);
assert!(
reassemble_data_frame(
&fragment_frame("event", 0, 2, &vec![0; MAX_WS_EVENT_BYTES + 1]),
&mut cache,
)
.is_err()
);
assert!(cache.is_empty());
}
#[test]
fn reconnect_backoff_is_bounded_and_resets_after_success() {
assert_eq!(reconnect_delay(0), Duration::from_secs(1));
assert_eq!(reconnect_delay(1), Duration::from_secs(1));
assert_eq!(reconnect_delay(2), Duration::from_secs(2));
assert_eq!(reconnect_delay(7), Duration::from_secs(64));
assert_eq!(reconnect_delay(100), Duration::from_secs(64));
}
#[tokio::test]
async fn turn_sink_creates_once_updates_same_card_and_cleans_up_at_finish() {
let state = Arc::new(Mutex::new(MockTurnState::default()));