fix(server): 上传失败后重建 segment 通道,避免后续段被静默丢弃 (#1623)

`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,恢复管道。
This commit is contained in:
Micuks authored and GitHub committed 2026-05-09 04:03:50 +00:00
1 parent 28b0bd7105
commit e70679dbd7
1 file changed
+13
@@ -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