fix: 处理第三轮 code review 意见(并发安全、路径语义一致性、后处理等待)

Gemini 高危:
- FFmpegAsyncInit 增加 ffmpegInitRunning CAS 守卫,并发调用(如 e2e
  多次 reinit)时只有第一个生效,避免并发 tool.Install()
- setFFmpegStatus 用 ffmpegStatusMu 串行化存储、变化通知与回调执行,
  保证订阅者看到的状态顺序与实际更新顺序一致
- ./ffmpeg 兜底改为 LookPath 任何失败都尝试:Unix 下当前目录不在 PATH,
  返回 ErrNotFound 而非 ErrDot,原来只处理 ErrDot 在 Unix 上不生效;
  该逻辑收敛到 utils.LookupSystemFFmpeg 供状态检测与实际查找共用

Gemini 中危:
- tool.Install() 前检查 ctx.Err(),程序退出中不再发起耗时下载

Codex P2:
- 配置了无效 ffmpeg_path 时不再走 remotetools 下载(下载了实际录制也
  不会用),直接置 not_found 并提示修正配置,状态与实际查找语义一致;
  e2e 相应改为 ffmpeg_path 留空 + BILILIVE_IGNORE_SYSTEM_FFMPEG=1
  环境变量屏蔽系统 FFmpeg(CI 预装 ffmpeg,无法用真实 PATH 模拟)
- 等待逻辑上移为 tools.WaitFFmpegAsyncInitDone 供多处复用:ffmpeg
  parser、pipeline 后处理阶段(convert_mp4 / burn_subtitles /
  custom_command)及 recorder 的 legacy custom commandline 在获取
  FFmpeg 路径前等待后台初始化结束,短直播后处理不再因下载未完成而失败

Copilot:
- WaitForToolsInit 轮询改用 time.NewTicker 复用计时器
- FFmpegBanner 初始状态查询失败时打印 console.warn,不再静默吞掉
- mock server 检查 w.Write 错误,客户端断开后立即停止发送

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
kira1928andClaude Fable 5 committed 2026-07-08 15:16:30 +09:00
1 parent 6096171663
commit d5d8b797e4
11 files changed
+128 -61

No files matched your search

+9 -2
View File
@@ -41,18 +41,25 @@ if (!process.env.TEST_WORKER_INDEX) {
}
}
// 构建启动 bgo 的命令(需设置 REMOTETOOLS_CONFIG 环境变量)
// 构建启动 bgo 的命令。
// REMOTETOOLS_CONFIG:将 remotetools 下载源重定向到本地 mock server;
// BILILIVE_IGNORE_SYSTEM_FFMPEG:跳过系统 PATH 中的 FFmpeg 查找(CI 机器预装了 ffmpeg),
// 强制走 remotetools 下载流程。
function makeBgoCommand(): string {
const configArg = `test-output/ffmpeg-test-config.yml`;
if (process.platform === 'win32') {
// PowerShell 语法设置环境变量
return (
`powershell -Command "$env:REMOTETOOLS_CONFIG='${toolsConfigPath}'; ` +
`$env:BILILIVE_IGNORE_SYSTEM_FFMPEG='1'; ` +
`go run -tags dev ./src/cmd/bililive --config ${configArg}"`
);
}
// bash/sh 语法
return `REMOTETOOLS_CONFIG='${toolsConfigPath}' go run -tags dev ./src/cmd/bililive --config ${configArg}`;
return (
`REMOTETOOLS_CONFIG='${toolsConfigPath}' BILILIVE_IGNORE_SYSTEM_FFMPEG=1 ` +
`go run -tags dev ./src/cmd/bililive --config ${configArg}`
);
}
export default defineConfig({
+4
View File
@@ -15,6 +15,7 @@ import (
"github.com/bililive-go/bililive-go/src/pipeline"
bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry"
"github.com/bililive-go/bililive-go/src/pkg/utils"
"github.com/bililive-go/bililive-go/src/tools"
"github.com/sirupsen/logrus"
)
@@ -59,6 +60,9 @@ func (s *BurnSubtitlesStage) Execute(ctx *pipeline.PipelineContext, input []pipe
ffmpegPath := ctx.FFmpegPath
if ffmpegPath == "" {
// FFmpeg 可能仍在后台异步下载(首次启动场景),等待其就绪后再查找,
// 避免短直播录制结束时后处理任务因 FFmpeg 未下载完成而失败
_ = tools.WaitFFmpegAsyncInitDone(ctx.Ctx, nil)
var err error
ffmpegPath, err = utils.GetFFmpegPath(ctx.Ctx)
if err != nil {
+4
View File
@@ -15,6 +15,7 @@ import (
"github.com/bililive-go/bililive-go/src/pipeline"
bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry"
"github.com/bililive-go/bililive-go/src/pkg/utils"
"github.com/bililive-go/bililive-go/src/tools"
"github.com/sirupsen/logrus"
)
@@ -47,6 +48,9 @@ func (s *ConvertMp4Stage) Execute(ctx *pipeline.PipelineContext, input []pipelin
ffmpegPath := ctx.FFmpegPath
if ffmpegPath == "" {
// FFmpeg 可能仍在后台异步下载(首次启动场景),等待其就绪后再查找,
// 避免短直播录制结束时后处理任务因 FFmpeg 未下载完成而失败
_ = tools.WaitFFmpegAsyncInitDone(ctx.Ctx, nil)
var err error
ffmpegPath, err = utils.GetFFmpegPath(ctx.Ctx)
if err != nil {
+3 -1
View File
@@ -12,6 +12,7 @@ import (
"github.com/bililive-go/bililive-go/src/pipeline"
"github.com/bililive-go/bililive-go/src/pkg/utils"
"github.com/bililive-go/bililive-go/src/tools"
)
// CustomCommandStage 自定义命令阶段
@@ -99,8 +100,9 @@ func (s *CustomCommandStage) renderCommand(ctx *pipeline.PipelineContext, file p
FFmpeg: ctx.FFmpegPath,
}
// 如果 FFmpeg 路径为空,尝试获取
// 如果 FFmpeg 路径为空,尝试获取(FFmpeg 可能仍在后台异步下载,先等待其就绪)
if data.FFmpeg == "" {
_ = tools.WaitFFmpegAsyncInitDone(ctx.Ctx, nil)
if ffmpegPath, err := utils.GetFFmpegPath(ctx.Ctx); err == nil {
data.FFmpeg = ffmpegPath
}
+1 -27
View File
@@ -173,7 +173,7 @@ func (p *Parser) ParseLiveStream(ctx context.Context, streamUrlInfo *live.Stream
// 找不到 FFmpeg 且后台异步初始化(tools.FFmpegAsyncInit)仍在检测/下载时,
// 挂起等待其进入终态后重查一次,避免开播瞬间陷入"找不到 FFmpeg → 5 秒重试"
// 的失败循环;下载一完成录制立即开始。等待可被 ctx 取消或 Stop() 中断。
if waitErr := p.waitFFmpegAsyncInit(ctx); waitErr != nil {
if waitErr := tools.WaitFFmpegAsyncInitDone(ctx, p.stopped); waitErr != nil {
return waitErr
}
if ffmpegPath, err = utils.GetFFmpegPathForLive(ctx, live); err != nil {
@@ -319,32 +319,6 @@ func (p *Parser) ParseLiveStream(ctx context.Context, streamUrlInfo *live.Stream
return nil
}
// waitFFmpegAsyncInit 在 FFmpeg 异步初始化处于 checking/downloading 状态时阻塞,
// 直到其进入终态(ready/not_found/error)、ctx 取消或 Stop() 被调用。
// 终态时返回 nil,由调用方重新查找 FFmpeg 路径并给出与原有逻辑一致的错误。
func (p *Parser) waitFFmpegAsyncInit(ctx context.Context) error {
logged := false
for {
// 先取变化通知 channel 再读状态,避免漏掉两步之间的状态变化
changed := tools.FFmpegStatusChanged()
state := tools.GetFFmpegStatus().State
if state != "checking" && state != "downloading" {
return nil
}
if !logged {
p.logger.Infof("FFmpeg 尚未就绪(%s),等待后台下载完成后开始录制...", state)
logged = true
}
select {
case <-changed:
case <-ctx.Done():
return ctx.Err()
case <-p.stopped:
return fmt.Errorf("录制在等待 FFmpeg 就绪时被停止")
}
}
}
// isFlvStream 判断 URL 是否指向 FLV 流
func (p *Parser) isFlvStream(u *url.URL) bool {
path := strings.ToLower(u.Path)
+17 -2
View File
@@ -46,9 +46,24 @@ func GetFFmpegPath(ctx context.Context) (string, error) {
}
}
return LookupSystemFFmpeg()
}
// EnvIgnoreSystemFFmpeg 设置该环境变量后跳过系统 PATH 中的 FFmpeg 查找,
// 强制走 remotetools 下载流程。仅用于 e2e 测试模拟"无 FFmpeg"环境
// (CI 机器普遍预装 ffmpeg,无法用真实 PATH 构造该场景)。
const EnvIgnoreSystemFFmpeg = "BILILIVE_IGNORE_SYSTEM_FFMPEG"
// LookupSystemFFmpeg 在系统 PATH 中查找 ffmpeg,找不到时回退查找工作目录下的 ./ffmpeg
// (允许把 ffmpeg 和主程序放在同一目录)。
// 注意不能只处理 exec.ErrDot:Unix 下当前目录不在 PATH 中,
// LookPath 返回 ErrNotFound 而非 ErrDot,因此任何失败都应尝试 ./ffmpeg。
func LookupSystemFFmpeg() (string, error) {
if os.Getenv(EnvIgnoreSystemFFmpeg) != "" {
return "", errors.New("system ffmpeg lookup disabled by " + EnvIgnoreSystemFFmpeg)
}
path, err := exec.LookPath("ffmpeg")
if errors.Is(err, exec.ErrDot) {
// put ffmpeg.exe and binary like bililive-windows-amd64.exe to the same folder is allowed
if err != nil {
path, err = exec.LookPath("./ffmpeg")
}
return path, err
+4
View File
@@ -39,6 +39,7 @@ import (
"github.com/bililive-go/bililive-go/src/pkg/streamprobe"
"github.com/bililive-go/bililive-go/src/pkg/utils"
"github.com/bililive-go/bililive-go/src/recorders/danmaku"
"github.com/bililive-go/bililive-go/src/tools"
)
const (
@@ -626,6 +627,9 @@ func (r *recorder) tryRecord(ctx context.Context) {
// 累积录制文件信息(legacy 路径),待录制结束后统一推送摘要
r.accumulateRecordedFiles(fileName)
// FFmpeg 可能仍在后台异步下载(如 native 下载器录制的短直播刚结束时),
// 等待其就绪后再查找,避免后处理命令因 FFmpeg 未下载完成而失败
_ = tools.WaitFFmpegAsyncInitDone(ctx, nil)
ffmpegPath, ffmpegErr := utils.GetFFmpegPathForLive(ctx, r.Live)
if ffmpegErr != nil {
r.getLogger().WithError(ffmpegErr).Error("failed to find ffmpeg")
+72 -24
View File
@@ -68,11 +68,13 @@ func WaitForToolsInit(ctx context.Context) error {
// 进入 initializing),此时轮询等待本轮结束,确保读到的是最新一轮的结果。
// Init() 的 defer 中先调用 signalToolsReady 更新 toolsInitError、再更新
// currentToolStatus,因此观察到状态离开 initializing 时错误值一定已是本轮的。
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
for toolStatusValue(currentToolStatus.Load()) == toolStatusValueInitializing {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(50 * time.Millisecond):
case <-ticker.C:
}
}
toolsInitMu.Lock()
@@ -90,12 +92,19 @@ type FFmpegStatus struct {
var (
ffmpegStatusVal atomic.Value // stores FFmpegStatus
// ffmpegStatusMu 串行化 setFFmpegStatus 的整个流程(存储、变化通知、回调执行),
// 保证并发更新时订阅者(如 SSE 广播)看到的状态顺序与实际更新顺序一致
ffmpegStatusMu sync.Mutex
ffmpegCbMu sync.RWMutex
ffmpegCallbacks []func(FFmpegStatus)
// ffmpegStatusChangedCh 在每次状态变化时被关闭并重建,供等待方感知状态变化
ffmpegStatusChangedMu sync.Mutex
ffmpegStatusChangedCh = make(chan struct{})
// ffmpegInitRunning 确保同一时间只有一个 FFmpegAsyncInit 流程在运行,
// 避免并发调用(如 e2e 测试多次触发 reinit)导致并发 tool.Install()
ffmpegInitRunning atomic.Bool
)
func init() {
@@ -103,6 +112,9 @@ func init() {
}
func setFFmpegStatus(s FFmpegStatus) {
ffmpegStatusMu.Lock()
defer ffmpegStatusMu.Unlock()
ffmpegStatusVal.Store(s)
ffmpegStatusChangedMu.Lock()
close(ffmpegStatusChangedCh)
@@ -139,6 +151,35 @@ func FFmpegStatusChanged() <-chan struct{} {
return ffmpegStatusChangedCh
}
// WaitFFmpegAsyncInitDone 在 FFmpeg 异步初始化仍在进行(checking/downloading)时阻塞,
// 直到其进入终态(ready/not_found/error)。用于录制、后处理等依赖 FFmpeg 的路径
// 在后台下载完成后立即继续,而非陷入失败重试循环。
// stop 可为 nil,用于额外的中断信号(如 parser 的 Stop())。
// 到达终态返回 nil(是否可用由调用方随后的路径查找决定,错误语义与原有逻辑一致);
// 因 ctx 取消或 stop 关闭而中断时返回相应错误。
func WaitFFmpegAsyncInitDone(ctx context.Context, stop <-chan struct{}) error {
logged := false
for {
// 先取变化通知 channel 再读状态,避免漏掉两步之间的状态变化
changed := FFmpegStatusChanged()
state := GetFFmpegStatus().State
if state != "checking" && state != "downloading" {
return nil
}
if !logged {
blog.GetLogger().Infof("FFmpeg 尚未就绪(%s),等待后台初始化完成...", state)
logged = true
}
select {
case <-changed:
case <-ctx.Done():
return ctx.Err()
case <-stop: // stop 为 nil 时此分支永远不会触发
return errors.New("等待 FFmpeg 就绪时被中断")
}
}
}
// OnFFmpegStatusChange 注册 FFmpeg 状态变化回调,在状态改变时被调用
func OnFFmpegStatusChange(fn func(FFmpegStatus)) {
ffmpegCbMu.Lock()
@@ -151,33 +192,37 @@ func ForceFFmpegStatus(s FFmpegStatus) {
setFFmpegStatus(s)
}
// isFFmpegAvailableWithoutRemotetools 仅检查配置路径和系统 PATH,
// 不访问 remotetools(避免在 Init() 完成前触及未配置好的 remotetools 状态)。
// 语义与 utils.GetFFmpegPath 保持一致:配置了 ffmpeg_path 时以该路径为准,
// 路径无效时不回退到系统 PATH(应走 remotetools 下载流程)。
func isFFmpegAvailableWithoutRemotetools() bool {
if cfg := configs.GetCurrentConfig(); cfg != nil {
if path := strings.TrimSpace(cfg.FfmpegPath); path != "" {
_, err := os.Stat(path)
return err == nil
}
}
_, err := exec.LookPath("ffmpeg")
if errors.Is(err, exec.ErrDot) {
// 允许把 ffmpeg 和主程序放在同一目录(与 utils.GetFFmpegPath 行为一致)
_, err = exec.LookPath("./ffmpeg")
}
return err == nil
}
// FFmpegAsyncInit 异步检测并(按需)下载 FFmpeg,期间通过状态回调通知进度。
// 可在 WebUI 启动后立即调用,不会阻塞主流程。
// 并发调用时只有第一个调用生效,其余直接返回(避免并发 Install)。
func FFmpegAsyncInit(ctx context.Context) {
if !ffmpegInitRunning.CompareAndSwap(false, true) {
return
}
setFFmpegStatus(FFmpegStatus{State: "checking"})
bilisentry.GoWithContext(ctx, func(ctx context.Context) {
// 先仅检查配置路径与系统 PATH,不触碰 remotetools(此时 Init() 尚未完成)
if isFFmpegAvailableWithoutRemotetools() {
blog.GetLogger().Info("FFmpeg found in system PATH or config")
defer ffmpegInitRunning.Store(false)
// 配置显式指定了 ffmpeg_path 时以其为准:实际录制查找(utils.GetFFmpegPath)
// 对非空配置路径只做存在性检查、不回退 remotetools/PATH,
// 因此路径无效时下载也无济于事,直接提示用户修正配置
if cfg := configs.GetCurrentConfig(); cfg != nil && cfg.FfmpegPath != "" {
if _, err := os.Stat(cfg.FfmpegPath); err == nil {
blog.GetLogger().Infof("FFmpeg found at configured path: %s", cfg.FfmpegPath)
setFFmpegStatus(FFmpegStatus{State: "ready", Source: "system"})
} else {
blog.GetLogger().Warnf("configured ffmpeg_path does not exist: %s", cfg.FfmpegPath)
setFFmpegStatus(FFmpegStatus{
State: "not_found",
Message: "配置的 ffmpeg_path 不存在: " + cfg.FfmpegPath,
})
}
return
}
// 检查系统 PATH(含 ./ffmpeg 兜底),不触碰 remotetools(此时 Init() 尚未完成)
if _, err := utils.LookupSystemFFmpeg(); err == nil {
blog.GetLogger().Info("FFmpeg found in system PATH")
setFFmpegStatus(FFmpegStatus{State: "ready", Source: "system"})
return
}
@@ -218,7 +263,10 @@ func FFmpegAsyncInit(ctx context.Context) {
return
}
// 需要下载
// 需要下载;程序若已在退出流程中则不再发起耗时下载
if ctx.Err() != nil {
return
}
source := tool.GetInstallSource()
blog.GetLogger().Infof("FFmpeg not found locally, downloading from %s...", source)
setFFmpegStatus(FFmpegStatus{
@@ -39,7 +39,10 @@ const FFmpegBanner: React.FC = () => {
if (active && !sseReceived && res?.state) {
setStatus(res as FFmpegStatus);
}
}).catch(() => {});
}).catch((err) => {
// 初始查询失败时依赖 SSE 推送兜底;打印日志便于定位"横幅一直不出现"类问题
console.warn('[FFmpegBanner] 获取 FFmpeg 状态失败:', err);
});
return () => {
active = false;
+6 -2
View File
@@ -130,7 +130,8 @@ func handleDownload(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Length", strconv.Itoa(len(zipData)))
w.WriteHeader(http.StatusOK)
// 限速发送(speed <= 0 时一次性发送剩余数据)
// 限速发送(speed <= 0 时一次性发送剩余数据)。
// 写入失败说明客户端已断开,立即返回以免继续空转 sleep
const chunkSize = 4096
data := zipData
for len(data) > 0 {
@@ -143,7 +144,10 @@ func handleDownload(w http.ResponseWriter, r *http.Request) {
if n > len(data) {
n = len(data)
}
w.Write(data[:n])
if _, err := w.Write(data[:n]); err != nil {
log.Printf("[ffmpeg-mock] 客户端中断下载: %v", err)
return
}
data = data[n:]
if flusher, ok2 := w.(http.Flusher); ok2 {
flusher.Flush()
@@ -1,12 +1,14 @@
# bililive-go 配置 — FFmpeg e2e 测试专用
# ffmpeg_path 指向一个不存在的路径,以确保程序走 remotetools 下载路径
# ffmpeg_path 留空 + 环境变量 BILILIVE_IGNORE_SYSTEM_FFMPEG=1(由 playwright 配置注入),
# 确保程序走 remotetools 下载路径(CI 机器普遍预装 ffmpeg,无法用真实 PATH 模拟无 FFmpeg 环境;
# 配置无效路径也不行——该场景下程序会直接报 not_found 提示修正配置,不会去下载)
rpc:
enable: true
bind: 127.0.0.1:8085
debug: false
interval: 5
out_put_path: ./test-output
ffmpeg_path: "./test-output/nonexistent-ffmpeg-for-e2e-test"
ffmpeg_path: ""
log:
out_put_folder: ./
save_last_log: true