From e70679dbd7cabb3592d313e96b778b7d2f0e173d Mon Sep 17 00:00:00 2001
From: Micuks <43397551+Micuks@users.noreply.github.com>
Date: Sat, 9 May 2026 12:03:50 +0800
Subject: [PATCH] =?UTF-8?q?fix(server):=20=E4=B8=8A=E4=BC=A0=E5=A4=B1?=
 =?UTF-8?q?=E8=B4=A5=E5=90=8E=E9=87=8D=E5=BB=BA=20segment=20=E9=80=9A?=
 =?UTF-8?q?=E9=81=93=EF=BC=8C=E9=81=BF=E5=85=8D=E5=90=8E=E7=BB=AD=E6=AE=B5?=
 =?UTF-8?q?=E8=A2=AB=E9=9D=99=E9=BB=98=E4=B8=A2=E5=BC=83=20(#1623)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`SegmentEventProcessor` 第一次为某个 streamer 处理事件时会创建一条
`async_channel::bounded(32)` 把段送给 `UActor`，并把 `tx` 缓存到
`self.channel`。一旦后续 `process_with_upload` 因为 `upload_single_file`
失败而 `?` 抛错，`UActor` 在 `handle_message` 走完 `Err` 分支就把 `rx`
drop 掉，但 `processor` 这边的 `tx` 仍然挂着。

之后只要 `execute()` 还在重试循环里没退出（streamer 在播 + 下载继续），
新段都会落到 `Some(tx)` 分支：`tx.force_send` 对已关闭通道返回 Err →
`Failed to send to buffer` → 段被静默丢弃。直到下载侧 `max_retries` 用
完、`processor` 才被 drop 重建。

实测：单台机器 24h 内出现 64 次 `Failed to send to buffer`，集中在两位
upload 失败过的 streamer 上，每 30min（segment_time）和每次 HLS
`#EXT-X-DISCONTINUITY` 各丢一段。

修复：进入 `match` 之前先用 `Sender::is_closed()` 探测，发现挂掉的 tx
就把 `self.channel` 置 None，复用下面的「首次创建」分支重新发 rx 给
UActor，恢复管道。
---
 crates/biliup-cli/src/server/common/download.rs | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs
index 84d788d..cd2e475 100644
--- a/crates/biliup-cli/src/server/common/download.rs
+++ b/crates/biliup-cli/src/server/common/download.rs
@@ -66,6 +66,19 @@ impl SegmentEventProcessor {
     pub fn process(&mut self, event: SegmentInfo) -> AppResult<()> {
         // 验证文件有效性
         self.file_validator.validate(&event.prev_file_path)?;
+
+        // 上一轮 process_with_upload 可能因上传失败提前返回，UActor 已 drop rx，
+        // 这里挂着的 tx 是死的；丢弃后下面会重建一条新的管道。
+        if let Some(tx) = &self.channel
+            && tx.is_closed()
+        {
+            warn!(
+                url = self.ctx.live_streamer().url,
+                "upload channel closed by uploader, reopening"
+            );
+            self.channel = None;
+        }
+
         match &self.channel {
             None => {
                 let (tx, rx) = async_channel::bounded(32); // Use tokio channel for async