diff --git a/go.mod b/go.mod index 627a637..5675268 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/gorilla/websocket v1.5.3 github.com/hr3lxphr6j/requests v0.0.1 github.com/joho/godotenv v1.5.1 - github.com/kira1928/remotetools v0.3.5 + github.com/kira1928/remotetools v0.3.6 github.com/prometheus/client_golang v1.11.0 github.com/robertkrimen/otto v0.5.1 github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b diff --git a/go.sum b/go.sum index 81299ee..90c73eb 100644 --- a/go.sum +++ b/go.sum @@ -145,8 +145,8 @@ github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/ github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= -github.com/kira1928/remotetools v0.3.5 h1:FJa4j3qIVJBL9GkPdnwmLe4YEfUoqeM9tLK4mfjhSS4= -github.com/kira1928/remotetools v0.3.5/go.mod h1:2fqah92kvzy8eFn9BAyrbrANHH6auVE+JU0y9D1qh6w= +github.com/kira1928/remotetools v0.3.6 h1:tY1g0AsnZKp3WgaEw742RoKiOeYwseTh6hHg7qb/oEU= +github.com/kira1928/remotetools v0.3.6/go.mod h1:2fqah92kvzy8eFn9BAyrbrANHH6auVE+JU0y9D1qh6w= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= diff --git a/scripts/docker-prepare-tools.mjs b/scripts/docker-prepare-tools.mjs index fe89301..fa18e62 100644 --- a/scripts/docker-prepare-tools.mjs +++ b/scripts/docker-prepare-tools.mjs @@ -11,7 +11,8 @@ * 此脚本在 CI 的原生 x86/ARM runner 上运行，避免在 QEMU 中解压大文件 * * 版本号和下载 URL 自动从 src/tools/remote-tools-config.json 读取， - * 无需手动维护版本号的一致性。 + * 无需手动维护版本号的一致性。输出目录遵循 remotetools 的 + * //// 布局。 * * 重要: 解压逻辑必须与 remotetools 的 extractDownloadedFile() 行为一致： * 1. 先解压到临时目录 @@ -206,7 +207,7 @@ function downloadFile(url, destPath) { * 下载并解压工具到目标目录 * 模拟 remotetools 的 extractDownloadedFile() 行为 * @param {string} url - * @param {string} dest - 最终目标目录（例如 tools/ffmpeg/n8.1-latest） + * @param {string} dest - 最终目标目录（例如 tools/linux/amd64/ffmpeg/n8.1-latest） */ async function downloadAndExtract(url, dest) { const filename = basename(new URL(url).pathname); @@ -303,7 +304,12 @@ async function main() { process.exit(1); } - mkdirSync(outputDir, { recursive: true }); + // remotetools 会按 //// 查找工具。 + // Docker 将 outputDir 复制为只读工具根目录，因此这里必须保留平台目录层级。 + const platformOutputDir = join(outputDir, 'linux', arch); + if (!dryRun) { + mkdirSync(platformOutputDir, { recursive: true }); + } console.log(`=== 为 linux/${arch} 预下载 Docker 内置工具${dryRun ? ' (dry-run)' : ''} ===`); console.log(`配置来源: ${configPath}`); @@ -330,7 +336,7 @@ async function main() { console.log(` 版本: ${result.version}`); console.log(` URL: ${result.url}`); - const destDir = join(outputDir, toolName, result.version); + const destDir = join(platformOutputDir, toolName, result.version); console.log(` 目标: ${destDir}`); if (dryRun) { @@ -345,10 +351,10 @@ async function main() { console.log(`\n=== 所有工具下载完成 (下载: ${downloadCount}, 跳过: ${skipCount}) ===`); // 列出下载结果 - if (existsSync(outputDir)) { - const toolDirs = readdirSync(outputDir); + if (existsSync(platformOutputDir)) { + const toolDirs = readdirSync(platformOutputDir); for (const dir of toolDirs) { - const dirPath = join(outputDir, dir); + const dirPath = join(platformOutputDir, dir); if (statSync(dirPath).isDirectory()) { console.log(` ${dir}/`); } diff --git a/src/cmd/bililive/bililive.go b/src/cmd/bililive/bililive.go index ebe754b..23d83b8 100644 --- a/src/cmd/bililive/bililive.go +++ b/src/cmd/bililive/bililive.go @@ -33,7 +33,6 @@ import ( "github.com/bililive-go/bililive-go/src/pkg/memwatch" "github.com/bililive-go/bililive-go/src/pkg/metadata" "github.com/bililive-go/bililive-go/src/pkg/openlist" - "github.com/bililive-go/bililive-go/src/pkg/ratelimit" bilisentryPkg "github.com/bililive-go/bililive-go/src/pkg/sentry" "github.com/bililive-go/bililive-go/src/pkg/telemetry" "github.com/bililive-go/bililive-go/src/pkg/update" @@ -88,13 +87,13 @@ func getConfigBesidesExecutable() (*configs.Config, error) { fmt.Fprintf(os.Stderr, "[Config] 使用 Launcher 目录的配置文件: %s\n", launcherConfigPath) return config, nil } else { - fmt.Fprintf(os.Stderr, "[Config] Launcher 配置文件加载失败 (%s): %v，回退到 exe 目录\n", launcherConfigPath, loadErr) + fmt.Fprintf(os.Stderr, "[Config] Launcher 配置文件加载失败 (%s): %v，回退到程序所在目录\n", launcherConfigPath, loadErr) } } - // 回退：在当前 exe 旁边查找（用户直接双击运行的场景） + // 回退：在当前程序所在目录查找（用户直接双击运行的场景） configPath := filepath.Join(filepath.Dir(exePath), "config.yml") - fmt.Fprintf(os.Stderr, "[Config] 使用当前 exe 目录的配置文件: %s\n", configPath) + fmt.Fprintf(os.Stderr, "[Config] 使用当前程序所在目录的配置文件: %s\n", configPath) config, err := configs.NewConfigWithFile(configPath) if err != nil { return nil, err @@ -279,6 +278,10 @@ func main() { logger.Debugf("%+v", consts.GetAppInfo()) logger.Debugf("%+v", configs.GetCurrentConfig()) + // 提前声明信号通道，以便后续 OnShutdownRequest 闭包可以引用它 + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt, syscall.SIGTERM) + // 初始化更新管理器并尝试连接到启动器（如果由启动器启动） var updateManager *update.Manager if os.Getenv("BILILIVE_LAUNCHER") != "" { @@ -315,7 +318,14 @@ func main() { logger.Infof("收到启动器关闭请求，优雅期 %d 秒", gracePeriod) // 发送关闭确认 updateManager.AckShutdown() - // 触发主程序关闭 + // 触发完整关闭流程（包括 tools.Cleanup），而非仅 rootCancel。 + // 确保 bililive-tools 等子进程在 Launcher 超时重启时被正确清理， + // 避免端口占用导致下次启动失败（issue #1129）。 + select { + case c <- syscall.SIGTERM: + default: + } + // 立即取消 context，确保关闭请求即时生效（即使关闭 goroutine 尚未启动也能中断初始化流程） rootCancel() }) } @@ -437,6 +447,11 @@ func main() { // 异步检测并（按需）下载 FFmpeg，不阻塞启动流程 tools.FFmpegAsyncInit(ctx) + // 工具链是异步初始化的，WebUI 会先于工具就绪启动。 + // 注册就绪检查，让依赖外部工具的平台（抖音依赖 bililive-tools）在工具可用之前 + // 不去请求直播间信息，从而也不会启动录制任务。 + live.SetPlatformReadinessChecker(tools.PlatformDependenciesReady) + // 启动 manager if err = lm.Start(ctx); err != nil { logger.Fatalf("failed to init listener manager, error: %s", err) @@ -531,15 +546,6 @@ func main() { // 第一步：立即为所有配置的直播间创建 InitializingLive，让前端可以看到 cfg := configs.GetCurrentConfig() - // 确保所有平台都有最小访问限制（用于控制并行初始化时的请求速度） - for _, room := range cfg.LiveRooms { - platformKey := configs.GetPlatformKeyFromUrl(room.Url) - if platformKey != "" { - minInterval := cfg.GetPlatformMinAccessInterval(platformKey) - ratelimit.GetGlobalRateLimiter().SetPlatformLimit(platformKey, minInterval) - } - } - // 分两批处理：监听中的直播间和非监听的直播间 var listeningRooms []live.Live var nonListeningRooms []live.Live @@ -547,8 +553,11 @@ func main() { // 创建初始化完成的回调函数 // 当 InitializingLive.GetInfo() 成功获取真实信息时，会自动调用此回调 onInitFinished := func(initializingLive live.Live, originalLive live.Live, info *live.Info) { - // 触发 RoomInitializingFinished 事件，让 manager 处理后续逻辑 - ed.DispatchEvent(events.NewEvent(listeners.RoomInitializingFinished, live.InitializingFinishedParam{ + // 必须等待旧监听器替换完成后再让 InitializingLive.GetInfo 返回。 + // 否则旧监听器会先派发 LiveStart，而替换过程又会异步派发 ListenStop 和新的 + // LiveStart，事件执行顺序不确定，可能出现「新 LiveStart 判断录制器已存在， + // 随后旧 ListenStop 又删除录制器」并导致直播中但不再录制。 + ed.DispatchEventSync(events.NewEvent(listeners.RoomInitializingFinished, live.InitializingFinishedParam{ InitializingLive: initializingLive, Live: originalLive, Info: info, @@ -677,16 +686,18 @@ func main() { logger.Infof("Created %d live rooms (%d listening, %d not listening)", inst.Lives.Len(), len(listeningRooms), len(nonListeningRooms)) - c := make(chan os.Signal, 1) - // 使用 os.Interrupt 更跨平台，在 Windows 上 SIGHUP 可能不被支持 - signal.Notify(c, os.Interrupt, syscall.SIGTERM) msgChan := c // 注册关闭回调，供更新系统在热重启时触发优雅关闭 servers.SetShutdownFunc(func() { - c <- os.Interrupt + select { + case c <- os.Interrupt: + default: + } }) + shutdownComplete := make(chan struct{}) bilisentryPkg.Go(func() { + defer close(shutdownComplete) <-msgChan logger.Info("Received shutdown signal, closing...") // 取消根 context，这会导致所有派生的 context 被取消 @@ -719,10 +730,21 @@ func main() { } // 停止自动更新器 servers.StopAutoUpdater() + // 终止所有子进程（bililive-tools 等）并关闭 remotetools WebUI。 + // 在所有关闭路径下执行，确保端口被释放（issue #1129）。 + tools.Cleanup() logger.Info("Shutdown complete") }) inst.WaitGroup.Wait() + // WaitGroup 只覆盖 Server、ListenerManager 和 RecorderManager。 + // 收到关闭请求后还必须等待其余模块和 tools.Cleanup() 完成，避免主程序退出后 + // Launcher 立即启动新版本，与尚未退出的 bililive-tools 等子进程争用端口。 + select { + case <-rootCtx.Done(): + <-shutdownComplete + default: + } // 检查是否需要就地切换到 launcher 模式 // 如果用户在前端点击了"立即更新"，doApplyUpdate 会设置此标志并触发服务关闭 @@ -733,11 +755,6 @@ func main() { // 这样进程不会退出，Docker 容器不会重启 if servers.PendingLauncherTransition() { logger.Infof("====== 版本切换: 所有服务已关闭，正在进入 Launcher 模式 (AppDataPath=%s) ======", configs.GetCurrentConfig().AppDataPath) - // 取消根 context，确保日志文件句柄被关闭（避免新版本清理时文件冲突） - rootCancel() - // 在进入 launcher 模式前，终止所有子进程（btools、klive 等）并关闭 remotetools WebUI - // 确保端口被释放，否则新版本 bgo 启动时会遇到 EADDRINUSE - tools.Cleanup() // 如果当前进程是由父 Launcher 启动的（BILILIVE_LAUNCHER=1）， // 不能自己再变成 launcher——否则两个 launcher 会争抢同一个 Named Pipe。 diff --git a/src/configs/config.go b/src/configs/config.go index 4945cd1..4b0a8e5 100644 --- a/src/configs/config.go +++ b/src/configs/config.go @@ -1316,20 +1316,32 @@ func (c *Config) GetPlatformMinAccessInterval(platformName string) int { // syncPlatformRateLimits 同步平台访问频率限制到全局限制器 func (c *Config) syncPlatformRateLimits() { rateLimiter := ratelimit.GetGlobalRateLimiter() - - // 清除已有限制 currentLimits := rateLimiter.GetAllPlatformLimits() + desiredLimits := make(map[string]int) - // 设置新的平台限制 - for platformKey, platformConfig := range c.PlatformConfigs { - if platformConfig.MinAccessIntervalSec > 0 { - rateLimiter.SetPlatformLimit(platformKey, platformConfig.MinAccessIntervalSec) + // 显式平台配置即使当前没有房间也要保留，便于随后添加房间时立即生效。 + for platformKey := range c.PlatformConfigs { + if platformKey != "" { + desiredLimits[platformKey] = c.GetPlatformMinAccessInterval(platformKey) } - // 从当前限制列表中移除此平台（标记为已处理） + } + + // 所有已配置房间的平台都必须至少应用默认 1 秒限制。 + // SetLiveRoomId 等配置更新也会调用本方法，不能只同步 platform_configs：否则启动代码 + // 刚设置的默认限制会在第一个房间写入 LiveId 时被删除，异步初始化便会形成请求风暴。 + for _, room := range c.LiveRooms { + platformKey := GetPlatformKeyFromUrl(room.Url) + if platformKey != "" { + desiredLimits[platformKey] = c.GetPlatformMinAccessInterval(platformKey) + } + } + + for platformKey, intervalSec := range desiredLimits { + rateLimiter.SetPlatformLimit(platformKey, intervalSec) delete(currentLimits, platformKey) } - // 清除配置中不再存在的平台限制 + // 清除配置中已经不存在的平台限制。 for platformKey := range currentLimits { rateLimiter.RemovePlatformLimit(platformKey) } @@ -1391,6 +1403,10 @@ func (r *ResolvedConfig) applyOverrides(override *OverridableConfig) { } } +// PlatformKeyDouyin 抖音的平台键。抖音的直播间解析依赖本地 bililive-tools 服务， +// 需要在若干处按平台键做就绪判断，因此单独导出常量避免散落的字符串字面量。 +const PlatformKeyDouyin = "douyin" + // GetPlatformKeyFromUrl 从URL中提取平台键，用于配置查找 func GetPlatformKeyFromUrl(urlStr string) string { u, err := url.Parse(urlStr) diff --git a/src/configs/config_test.go b/src/configs/config_test.go index fa7c77e..eb857ec 100644 --- a/src/configs/config_test.go +++ b/src/configs/config_test.go @@ -4,6 +4,8 @@ import ( "os" "testing" + "github.com/bililive-go/bililive-go/src/pkg/ratelimit" + "github.com/bililive-go/bililive-go/src/types" "github.com/stretchr/testify/assert" ) @@ -95,6 +97,30 @@ func TestGetPlatformMinAccessInterval(t *testing.T) { assert.Equal(t, 1, interval) // 默认最小间隔为 1 秒，防止无限制高频访问 } +func TestDefaultPlatformRateLimitSurvivesTransientConfigUpdate(t *testing.T) { + const roomURL = "https://live.douyin.com/123456" + limiter := ratelimit.GetGlobalRateLimiter() + t.Cleanup(func() { + SetCurrentConfig(NewConfig()) + }) + + cfg := NewConfig() + cfg.LiveRooms = []LiveRoom{{Url: roomURL, IsListening: true}} + cfg.RefreshLiveRoomIndexCache() + SetCurrentConfig(cfg) + + if got := limiter.GetAllPlatformLimits()[PlatformKeyDouyin]; got != 1 { + t.Fatalf("初始默认平台限流 = %d 秒，期望 1 秒", got) + } + + if _, err := SetLiveRoomId(roomURL, types.LiveID("douyin-test")); err != nil { + t.Fatalf("写入临时 LiveId 失败: %v", err) + } + if got := limiter.GetAllPlatformLimits()[PlatformKeyDouyin]; got != 1 { + t.Fatalf("临时配置更新后的默认平台限流 = %d 秒，期望仍为 1 秒", got) + } +} + func TestBackwardsCompatibility(t *testing.T) { // Test that old config files still work oldConfigYaml := ` diff --git a/src/listeners/listener.go b/src/listeners/listener.go index 8570460..a7fd1fb 100644 --- a/src/listeners/listener.go +++ b/src/listeners/listener.go @@ -3,6 +3,7 @@ package listeners import ( "context" + "errors" "sync/atomic" "time" @@ -25,7 +26,9 @@ const ( type Listener interface { Start() error + StartWithInfo(info *live.Info) error Close() + CloseSync() } func NewListener(ctx context.Context, live live.Live) Listener { @@ -55,24 +58,70 @@ type listener struct { } func (l *listener) Start() error { + return l.start(nil) +} + +// StartWithInfo 使用已经获取到的直播间信息启动监听器，避免初始化交接后立即重复请求平台。 +func (l *listener) StartWithInfo(info *live.Info) error { + return l.start(info) +} + +func (l *listener) start(initialInfo *live.Info) error { if !atomic.CompareAndSwapUint32(&l.state, begin, pending) { return nil } defer atomic.CompareAndSwapUint32(&l.state, pending, running) l.ed.DispatchEvent(events.NewEvent(ListenStart, l.Live)) - l.refresh() - bilisentry.Go(func() { l.run() }) + + // 首次信息获取放到后台执行。调用方 manager.AddListener 全程持有管理器的全局锁， + // 而 refresh 是一次真实的网络请求，还要排队等待平台访问频率限制； + // 几百个直播间串行下来会把锁占用好几分钟，期间任何增删直播间的操作都会被卡住。 + bilisentry.Go(func() { + if !l.isStopped() { + if initialInfo != nil { + l.processInfo(initialInfo) + } else { + l.refresh() + } + } + l.run() + }) return nil } +// isStopped 返回 listener 是否已经被关闭 +func (l *listener) isStopped() bool { + select { + case <-l.stop: + return true + default: + return false + } +} + func (l *listener) Close() { + l.close(false) +} + +// CloseSync 同步完成 ListenStop 的所有处理器，仅用于初始化 listener 的交接路径。 +// 普通关闭仍保持异步，避免改变其他调用方的延迟语义。 +func (l *listener) CloseSync() { + l.close(true) +} + +func (l *listener) close(syncEvent bool) { if !atomic.CompareAndSwapUint32(&l.state, running, stopped) { return } - l.ed.DispatchEvent(events.NewEvent(ListenStop, l.Live)) - l.runCancel() // 取消 run 循环中的等待 + l.runCancel() // 先取消等待并标记停止，禁止并发请求结果继续发布事件 close(l.stop) + event := events.NewEvent(ListenStop, l.Live) + if syncEvent { + l.ed.DispatchEventSync(event) + } else { + l.ed.DispatchEvent(event) + } } // sendLiveNotification 发送直播状态变更通知 @@ -95,15 +144,26 @@ func (l *listener) sendLiveNotification(hostName, status string) { func (l *listener) refresh() { info, err := l.Live.GetInfo() if err != nil { - l.Live.GetLogger(). - WithError(err). - WithField("url", l.Live.GetRawUrl()). - Error("failed to load room info") + l.logGetInfoError(err) return } l.processInfo(info) } +// logGetInfoError 记录获取直播间信息失败的日志。 +// 依赖工具尚未就绪不是异常状况（工具还在下载/启动中，稍后调度器会自动恢复）， +// 几百个直播间同时报错只会淹没日志，因此降级为 debug。 +func (l *listener) logGetInfoError(err error) { + entry := l.Live.GetLogger(). + WithError(err). + WithField("url", l.Live.GetRawUrl()) + if errors.Is(err, live.ErrPlatformToolsNotReady) { + entry.Debug("skip loading room info") + return + } + entry.Error("failed to load room info") +} + func (l *listener) run() { // 使用 GetInfoWithInterval 来处理等待和请求 // 它会自动获取配置的间隔时间，并在尊重平台速率限制的前提下等待后发送请求 @@ -119,10 +179,7 @@ func (l *listener) run() { if l.runCtx.Err() != nil { return } - l.Live.GetLogger(). - WithError(err). - WithField("url", l.Live.GetRawUrl()). - Error("failed to load room info") + l.logGetInfoError(err) continue } l.processInfo(info) @@ -132,11 +189,17 @@ func (l *listener) run() { // processInfo 处理获取到的直播间信息，检测状态变化并触发事件 func (l *listener) processInfo(info *live.Info) { + // 初始化完成回调可能在 GetInfo 返回前同步替换并关闭当前 listener。 + // 已关闭的旧 listener 不得再发布 LiveStart/LiveEnd，否则会与新 listener 的事件竞态。 + if l.isStopped() { + return + } + // 尝试从缓存中获取主播姓名，以防API调用失败 hostName := info.HostName if hostName == "" { if wrappedLive, ok := l.Live.(*live.WrappedLive); ok { - if cachedInfo, get_err := wrappedLive.GetInfo(); get_err == nil && cachedInfo != nil { + if cachedInfo, found := wrappedLive.GetCachedInfo(); found { hostName = cachedInfo.HostName } } diff --git a/src/listeners/listener_test.go b/src/listeners/listener_test.go index 93df6c5..621bbc8 100644 --- a/src/listeners/listener_test.go +++ b/src/listeners/listener_test.go @@ -3,6 +3,7 @@ package listeners import ( "context" "errors" + "sync/atomic" "testing" "github.com/bluele/gcache" @@ -134,3 +135,35 @@ func TestListenerStartAndClose(t *testing.T) { l.Close() l.Close() } + +func TestClosedListenerDoesNotPublishInfo(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + ed := evtmock.NewMockDispatcher(ctrl) + ctx := context.WithValue(context.Background(), instance.Key, &instance.Instance{EventDispatcher: ed}) + live := livemock.NewMockLive(ctrl) + l := NewListener(ctx, live).(*listener) + atomic.StoreUint32(&l.state, running) + ed.EXPECT().DispatchEvent(events.NewEvent(ListenStop, live)) + l.Close() + + // 模拟网络请求在 listener 关闭后才返回开播信息；不得再发布 LiveStart。 + l.processInfo(&livepkg.Info{Status: true}) + assert.False(t, l.status.roomStatus) +} + +func TestListenerCloseSyncWaitsForStopHandlers(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + ed := evtmock.NewMockDispatcher(ctrl) + ctx := context.WithValue(context.Background(), instance.Key, &instance.Instance{EventDispatcher: ed}) + live := livemock.NewMockLive(ctrl) + l := NewListener(ctx, live).(*listener) + atomic.StoreUint32(&l.state, running) + ed.EXPECT().DispatchEventSync(events.NewEvent(ListenStop, live)) + + l.CloseSync() + assert.True(t, l.isStopped()) +} diff --git a/src/listeners/manager.go b/src/listeners/manager.go index 9f4d104..3b8de35 100644 --- a/src/listeners/manager.go +++ b/src/listeners/manager.go @@ -56,10 +56,10 @@ func (m *manager) registryListener(ctx context.Context, ed events.Dispatcher) { // 原子地替换 Lives map 中的条目（删除旧 InitializingLive，添加新 wrappedLive） inst.Lives.ReplaceKey(oldLiveId, wrappedLive.GetLiveId(), wrappedLive) - // 直接将已有的 info 存入缓存，避免再次调用 GetInfo() 触发 API 请求 - // info.Live 需要更新为新的 wrappedLive，确保后续 JSON 序列化正确 + // 将已有的 info 注入新的 WrappedLive，避免 listener 交接后立即重复请求平台， + // 同时让新调度器从本次成功请求开始计算下一轮间隔。 info.Live = wrappedLive - if err := inst.Cache.Set(wrappedLive, info); err != nil { + if err := wrappedLive.(*live.WrappedLive).SeedInfo(info); err != nil { logger.WithError(err).Warn("failed to cache info for new live") } @@ -73,7 +73,7 @@ func (m *manager) registryListener(ctx context.Context, ed events.Dispatcher) { } configs.SetLiveRoomId(wrappedLive.GetRawUrl(), wrappedLive.GetLiveId()) if room.IsListening { - if err := m.replaceListener(ctx, initializingLive, wrappedLive); err != nil { + if err := m.replaceListener(ctx, initializingLive, wrappedLive, info); err != nil { logger.WithFields(map[string]any{ "url": wrappedLive.GetRawUrl(), }).Error(err) @@ -126,7 +126,7 @@ func (m *manager) RemoveListener(ctx context.Context, liveId types.LiveID) error return nil } -func (m *manager) replaceListener(ctx context.Context, oldLive live.Live, newLive live.Live) error { +func (m *manager) replaceListener(ctx context.Context, oldLive live.Live, newLive live.Live, info *live.Info) error { m.lock.Lock() defer m.lock.Unlock() oldLiveId := oldLive.GetLiveId() @@ -134,7 +134,8 @@ func (m *manager) replaceListener(ctx context.Context, oldLive live.Live, newLiv if !ok { return ErrListenerNotExist } - oldListener.Close() + // 必须等旧 ListenStop 的录制器清理完成后，才能让新 listener 发布 LiveStart。 + oldListener.CloseSync() newListener := newListener(ctx, newLive) if oldLiveId == newLive.GetLiveId() { m.savers[oldLiveId] = newListener @@ -142,7 +143,7 @@ func (m *manager) replaceListener(ctx context.Context, oldLive live.Live, newLiv delete(m.savers, oldLiveId) m.savers[newLive.GetLiveId()] = newListener } - return newListener.Start() + return newListener.StartWithInfo(info) } func (m *manager) GetListener(ctx context.Context, liveId types.LiveID) (Listener, error) { diff --git a/src/listeners/manager_test.go b/src/listeners/manager_test.go index 7ecfd4e..5c4ef7d 100644 --- a/src/listeners/manager_test.go +++ b/src/listeners/manager_test.go @@ -74,3 +74,28 @@ func TestManagerStartAndClose(t *testing.T) { } m.Close(ctx) } + +func TestReplaceListenerUsesSynchronousHandoverAndInitialInfo(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + oldListener := NewMockListener(ctrl) + oldListener.EXPECT().CloseSync() + newListenerMock := NewMockListener(ctrl) + info := &live.Info{Status: true} + newListenerMock.EXPECT().StartWithInfo(info).Return(nil) + + oldLive := livemock.NewMockLive(ctrl) + oldLive.EXPECT().GetLiveId().Return(types.LiveID("old")) + newLive := livemock.NewMockLive(ctrl) + newLive.EXPECT().GetLiveId().Return(types.LiveID("new")).Times(2) + + m := &manager{savers: map[types.LiveID]Listener{"old": oldListener}} + backup := newListener + newListener = func(context.Context, live.Live) Listener { return newListenerMock } + defer func() { newListener = backup }() + + assert.NoError(t, m.replaceListener(context.Background(), oldLive, newLive, info)) + assert.Same(t, newListenerMock, m.savers["new"]) + assert.NotContains(t, m.savers, types.LiveID("old")) +} diff --git a/src/listeners/mock_test.go b/src/listeners/mock_test.go index f5fb795..ae33788 100644 --- a/src/listeners/mock_test.go +++ b/src/listeners/mock_test.go @@ -54,6 +54,18 @@ func (mr *MockListenerMockRecorder) Close() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockListener)(nil).Close)) } +// CloseSync mocks base method. +func (m *MockListener) CloseSync() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "CloseSync") +} + +// CloseSync indicates an expected call of CloseSync. +func (mr *MockListenerMockRecorder) CloseSync() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSync", reflect.TypeOf((*MockListener)(nil).CloseSync)) +} + // Start mocks base method. func (m *MockListener) Start() error { m.ctrl.T.Helper() @@ -68,6 +80,20 @@ func (mr *MockListenerMockRecorder) Start() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockListener)(nil).Start)) } +// StartWithInfo mocks base method. +func (m *MockListener) StartWithInfo(info *live.Info) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "StartWithInfo", info) + ret0, _ := ret[0].(error) + return ret0 +} + +// StartWithInfo indicates an expected call of StartWithInfo. +func (mr *MockListenerMockRecorder) StartWithInfo(info any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartWithInfo", reflect.TypeOf((*MockListener)(nil).StartWithInfo), info) +} + // MockManager is a mock of Manager interface. type MockManager struct { ctrl *gomock.Controller diff --git a/src/live/douyin/douyin_btools.go b/src/live/douyin/douyin_btools.go index 6c08792..5c0d8c3 100644 --- a/src/live/douyin/douyin_btools.go +++ b/src/live/douyin/douyin_btools.go @@ -3,18 +3,55 @@ package douyin import ( "encoding/json" "fmt" + "io" "net/http" "net/url" + "time" "github.com/bililive-go/bililive-go/src/live" + "github.com/bililive-go/bililive-go/src/tools" ) var btoolsConsts = struct { port int authToken string }{ - port: 18110, - authToken: "Basic YTph", + port: tools.BToolsPort, + authToken: tools.BToolsAuthToken, +} + +// btoolsClient 访问本地 bililive-tools 服务的专用客户端。 +// +// 不能用 http.DefaultClient：它没有超时，本地服务一旦卡住（而不是报错）， +// 调用方会永久阻塞，直播间轮询 goroutine 会不断堆积。 +// 这里的请求全部打到 127.0.0.1，正常应在毫秒级返回，给 15 秒已经非常宽松。 +var btoolsClient = &http.Client{ + Timeout: 15 * time.Second, +} + +// doBToolsRequest 向本地 bililive-tools 服务发起一次带鉴权的 GET 请求。 +// 返回的 body 已经读取完毕并关闭，调用方直接解析即可。 +func doBToolsRequest(endpoint string) ([]byte, error) { + req, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", btoolsConsts.authToken) + + resp, err := btoolsClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + // 读掉 body 才能让连接回到连接池复用，否则每次失败都会新建一条 TCP 连接， + // 大量直播间同时失败时会迅速耗尽本地端口 + _, _ = io.Copy(io.Discard, resp.Body) + return nil, fmt.Errorf("请求失败: %s", resp.Status) + } + + return io.ReadAll(resp.Body) } type ChannelInfo struct { @@ -68,30 +105,12 @@ func (l *btoolsLive) updateChannelInfo() (err error) { } func (l *btoolsLive) fetchChannelInfo() (channelInfo ChannelInfo, err error) { - // 使用自定义请求以便添加认证Header endpoint := fmt.Sprintf("http://127.0.0.1:%d/bgo/channel-info?url=%s", btoolsConsts.port, url.QueryEscape(l.Url.String())) - req, reqErr := http.NewRequest(http.MethodGet, endpoint, nil) - if reqErr != nil { - err = reqErr - return - } - req.Header.Set("Authorization", btoolsConsts.authToken) - - resp, doErr := http.DefaultClient.Do(req) - if doErr != nil { - err = doErr - return - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - err = fmt.Errorf("请求失败: %s", resp.Status) - return - } - - if err = json.NewDecoder(resp.Body).Decode(&channelInfo); err != nil { + body, err := doBToolsRequest(endpoint) + if err != nil { return } + err = json.Unmarshal(body, &channelInfo) return } @@ -104,21 +123,11 @@ func (l *btoolsLive) fetchLiveInfo() (liveInfo liveInfoResp, err error) { } endpoint := fmt.Sprintf("http://127.0.0.1:%d/bgo/live-info?platform=douyin&roomId=%s", btoolsConsts.port, url.QueryEscape(l.roomId)) - req, reqErr := http.NewRequest(http.MethodGet, endpoint, nil) - if reqErr != nil { - return liveInfo, reqErr + body, err := doBToolsRequest(endpoint) + if err != nil { + return liveInfo, err } - req.Header.Set("Authorization", btoolsConsts.authToken) - - resp, doErr := http.DefaultClient.Do(req) - if doErr != nil { - return liveInfo, doErr - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return liveInfo, fmt.Errorf("请求失败: %s", resp.Status) - } - if err = json.NewDecoder(resp.Body).Decode(&liveInfo); err != nil { + if err = json.Unmarshal(body, &liveInfo); err != nil { return liveInfo, err } return liveInfo, nil @@ -133,21 +142,11 @@ func (l *btoolsLive) fetchStreamInfo() (streamInfo streamInfoResp, err error) { } endpoint := fmt.Sprintf("http://127.0.0.1:%d/bgo/stream-info?platform=douyin&roomId=%s", btoolsConsts.port, url.QueryEscape(l.roomId)) - req, reqErr := http.NewRequest(http.MethodGet, endpoint, nil) - if reqErr != nil { - return streamInfo, reqErr + body, err := doBToolsRequest(endpoint) + if err != nil { + return streamInfo, err } - req.Header.Set("Authorization", btoolsConsts.authToken) - - resp, doErr := http.DefaultClient.Do(req) - if doErr != nil { - return streamInfo, doErr - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return streamInfo, fmt.Errorf("请求失败: %s", resp.Status) - } - if err = json.NewDecoder(resp.Body).Decode(&streamInfo); err != nil { + if err = json.Unmarshal(body, &streamInfo); err != nil { return streamInfo, err } return streamInfo, nil diff --git a/src/live/lives.go b/src/live/lives.go index b3d640f..92fcca9 100755 --- a/src/live/lives.go +++ b/src/live/lives.go @@ -4,6 +4,7 @@ package live import ( "context" "errors" + "fmt" "net/http" "net/http/cookiejar" "net/url" @@ -25,12 +26,19 @@ type SchedulerRefreshCallback func(live Live, status SchedulerStatus) // RequestStatusCallback 请求状态追踪的回调函数类型 type RequestStatusCallback func(liveID, platform string, success bool, errMsg string) +// PlatformReadinessChecker 检查某个平台依赖的外部工具是否已就绪。 +// 返回 false 时调度器不会发起请求，reason 用于日志说明原因。 +type PlatformReadinessChecker func(platformKey string) (ready bool, reason string) + // 全局调度器刷新回调（由外部包设置，避免循环依赖） var schedulerRefreshCallback SchedulerRefreshCallback // 全局请求状态追踪回调（由 iostats 包设置，避免循环依赖） var requestStatusCallback RequestStatusCallback +// 全局平台工具就绪检查回调（由 tools 包设置，避免循环依赖） +var platformReadinessChecker PlatformReadinessChecker + // SetSchedulerRefreshCallback 设置调度器刷新完成的回调函数 func SetSchedulerRefreshCallback(callback SchedulerRefreshCallback) { schedulerRefreshCallback = callback @@ -41,6 +49,12 @@ func SetRequestStatusCallback(callback RequestStatusCallback) { requestStatusCallback = callback } +// SetPlatformReadinessChecker 设置平台依赖工具的就绪检查函数。 +// 应在创建直播间之前调用；未设置时视为所有平台都已就绪。 +func SetPlatformReadinessChecker(checker PlatformReadinessChecker) { + platformReadinessChecker = checker +} + var ( m = make(map[string]Builder) InitializingLiveBuilderInstance InitializingLiveBuilder @@ -212,6 +226,21 @@ type Live interface { Close() } +// ErrPlatformToolsNotReady 表示直播间所属平台依赖的外部工具还没就绪， +// 此时不应该发起请求，也不应该按「获取失败」来对待（不是平台或网络的问题）。 +var ErrPlatformToolsNotReady = errors.New("平台依赖的工具尚未就绪") + +const ( + // defaultInterval 没有配置轮询间隔时使用的默认值 + defaultInterval = 30 * time.Second + // maxFailureBackoff 连续失败时退避间隔的上限。 + // 取值需要兼顾两点：故障时不能把请求打成风暴，恢复后也不能太久才发现开播。 + maxFailureBackoff = 5 * time.Minute + // toolReadinessPollInterval 依赖工具未就绪时重新检查的间隔。 + // 这里只是读一个内存中的状态，不产生任何网络请求，因此可以查得比较勤。 + toolReadinessPollInterval = 2 * time.Second +) + // infoResult 用于传递 GetInfo 的结果 type infoResult struct { info *Info @@ -252,16 +281,20 @@ type SchedulerStatusProvider interface { type WrappedLive struct { Live cache gcache.Cache + // requestMu 串行化同一直播间的直接刷新与调度刷新，并保护缓存的读改写顺序。 + requestMu sync.Mutex // 请求调度相关字段 - mu sync.Mutex - waiters []waiter // 等待下一次请求结果的调用方 - lastRequestAt time.Time // 上次发送请求的时间 - schedulerOnce sync.Once // 确保调度器只启动一次 - schedulerStarted bool // 调度器是否已启动 - schedulerStop chan struct{} // 停止调度器的信号 - schedulerCtx context.Context - schedulerCancel context.CancelFunc + mu sync.Mutex + waiters []waiter // 等待下一次请求结果的调用方 + lastRequestAt time.Time // 上次发送请求的时间（不论成功还是失败） + consecutiveFailures int // 连续失败次数，用于计算退避间隔 + waitingForTool bool // 是否正在等待平台依赖的外部工具就绪 + schedulerOnce sync.Once // 确保调度器只启动一次 + schedulerStarted bool // 调度器是否已启动 + schedulerStop chan struct{} // 停止调度器的信号 + schedulerCtx context.Context + schedulerCancel context.CancelFunc } // NewWrappedLive 创建一个带有缓存功能的 Live 包装器 @@ -302,21 +335,63 @@ func (w *WrappedLive) GetRoomID() string { return "" } +// GetCachedInfo 只读取最近一次缓存，不触发平台请求。 +func (w *WrappedLive) GetCachedInfo() (*Info, bool) { + if w.cache == nil { + return nil, false + } + cached, err := w.cache.Get(w) + if err != nil { + return nil, false + } + info, ok := cached.(*Info) + return info, ok && info != nil +} + func (w *WrappedLive) GetInfo() (*Info, error) { - // 在通用位置应用平台访问频率限制 - // 如果被取消则直接返回 - if !w.waitForPlatformRateLimit() { - return nil, w.schedulerCtx.Err() + // forceRefresh 与调度器可能同时进入 GetInfo。若允许它们并发，较早开始但较晚写缓存的 + // 失败请求可能覆盖较新的成功结果；整个请求链串行后，缓存提交顺序与请求顺序一致。 + w.requestMu.Lock() + defer w.requestMu.Unlock() + + // 依赖的外部工具还没就绪时不发请求（调度器之外还有 listener.refresh 等直接调用方）。 + // 正常情况下调度器会先拦住，这里主要覆盖直接调用以及「刚检查完就绪、随即工具挂掉」的竞态， + // 因此同样要通知等待者，避免它们一直挂着。 + if ready, reason := w.platformToolsReadyWithReason(); !ready { + err := fmt.Errorf("%w: %s", ErrPlatformToolsNotReady, reason) + w.notifyWaiters(nil, err) + return nil, err } + // 在通用位置获取平台请求许可。许可同时约束请求开始间隔和同平台在途数量， + // 防止启动或工具恢复时几百个房间一起进入平台实现。 + releasePlatform, ok := w.acquirePlatformRequest() + if !ok { + return nil, w.schedulerCtx.Err() + } + defer releasePlatform() + i, err := w.Live.GetInfo() + // 不论成功还是失败都要记录本次请求时间： + // 如果失败时不更新 lastRequestAt，调度器会一直认为「早就该请求了」， + // 从而退化成固定 3 秒一次的重试，反而把故障状态下的请求频率放大十倍， + // 把本地工具和平台接口一起打爆，永远无法自愈。 + requestFailed := w.recordRequestResult(i, err) + // 记录请求状态到 IO 统计（通过回调避免循环依赖） if requestStatusCallback != nil { liveID := string(w.GetLiveId()) platform := w.GetPlatformCNName() - if err != nil { - requestStatusCallback(liveID, platform, false, err.Error()) + if requestFailed { + errMessage := "" + if i != nil { + errMessage = i.LastError + } + if err != nil { + errMessage = err.Error() + } + requestStatusCallback(liveID, platform, false, errMessage) } else { requestStatusCallback(liveID, platform, true, "") } @@ -326,30 +401,87 @@ func (w *WrappedLive) GetInfo() (*Info, error) { w.notifyWaiters(i, err) if err != nil { - if info, err2 := w.cache.Get(w); err2 == nil { - // 将错误信息存到 LastError 而非 RoomName - // 避免错误文本出现在录制文件名中 - info.(*Info).LastError = err.Error() - } + // 将错误信息存到 LastError 而非 RoomName，避免错误文本出现在录制文件名中。 + // 注意这里不能就地修改缓存里的 *Info：它同时被 API/SSE 等读取方持有， + // 就地写字段是无锁的数据竞争。改为写入一份副本再替换缓存。 + w.cacheLastError(err) + // 失败同样要通知前端更新倒计时，否则退避期间界面会一直显示「即将刷新」 + w.dispatchSchedulerRefreshEvent() return nil, err } if w.cache != nil { - // 成功获取信息，清除之前的错误 - i.LastError = "" + // 初始化占位 Info 会通过 LastError 告诉调度器底层请求实际失败，既保留可展示的 + // 占位信息，也不能把这次请求误算成成功并清零退避。 + if !requestFailed { + i.LastError = "" + } w.cache.Set(w, i) } - // 更新最后请求时间 - w.mu.Lock() - w.lastRequestAt = time.Now() - w.mu.Unlock() - // 发送调度器刷新完成事件，通知前端更新倒计时 w.dispatchSchedulerRefreshEvent() return i, nil } +// recordRequestResult 更新调度器的请求时间和连续失败计数。 +// InitializingLive 为了向前端保留占位信息会返回 nil error，因此还要识别其 LastError。 +func (w *WrappedLive) recordRequestResult(info *Info, err error) bool { + requestFailed := err != nil || (info != nil && info.Initializing && info.LastError != "") + w.mu.Lock() + w.lastRequestAt = time.Now() + if requestFailed { + w.consecutiveFailures++ + } else { + w.consecutiveFailures = 0 + } + w.mu.Unlock() + return requestFailed +} + +// SeedInfo 把已经完成的初始化请求结果注入新的 WrappedLive。 +// 监听器交接时复用这份结果，可以避免新 listener 立刻重复请求一次，并让调度器从本次 +// 请求时间开始计算下一轮间隔。 +func (w *WrappedLive) SeedInfo(info *Info) error { + if info == nil { + return errors.New("不能注入空的直播间信息") + } + + w.requestMu.Lock() + defer w.requestMu.Unlock() + + w.mu.Lock() + w.lastRequestAt = time.Now() + w.consecutiveFailures = 0 + w.mu.Unlock() + + info.LastError = "" + if w.cache == nil { + return nil + } + return w.cache.Set(w, info) +} + +// cacheLastError 把最近一次的错误信息记录到缓存的直播间信息上。 +// 通过「读出来 → 复制 → 改副本 → 写回」的方式更新，避免修改其他 goroutine +// 正在读取的 *Info（gcache 只保证自身的并发安全，不保证值对象的）。 +func (w *WrappedLive) cacheLastError(err error) { + if w.cache == nil { + return + } + cached, cacheErr := w.cache.Get(w) + if cacheErr != nil { + return + } + info, ok := cached.(*Info) + if !ok || info == nil { + return + } + updated := *info + updated.LastError = err.Error() + w.cache.Set(w, &updated) +} + // dispatchSchedulerRefreshEvent 发送调度器刷新完成事件 func (w *WrappedLive) dispatchSchedulerRefreshEvent() { if schedulerRefreshCallback != nil { @@ -431,7 +563,7 @@ func (w *WrappedLive) GetSchedulerStatus() SchedulerStatus { now := time.Now() interval := w.getConfiguredInterval() - intervalDuration := time.Duration(interval) * time.Second + intervalDuration := w.nextRequestIntervalLocked() status := SchedulerStatus{ HasWaiters: len(w.waiters) > 0, @@ -484,11 +616,23 @@ func (w *WrappedLive) runScheduler() { } } - // 有等待者，计算需要等待的时间 - interval := time.Duration(w.getConfiguredInterval()) * time.Second + // 平台依赖的外部工具（如抖音需要的 bililive-tools）还没就绪时不发起请求。 + // 此时请求必然失败，只会把还在启动中的本地服务打爆，并让直播间被误判为未开播； + // 等待期间调用方继续阻塞在 GetInfoWithInterval 上，也就不会触发录制。 + if !w.platformToolsReady() { + select { + case <-w.schedulerStop: + return + case <-w.schedulerCtx.Done(): + return + case <-time.After(toolReadinessPollInterval): + continue + } + } + // 有等待者，计算需要等待的时间（连续失败时会自动退避） w.mu.Lock() - nextRequestAt := w.lastRequestAt.Add(interval) + nextRequestAt := w.lastRequestAt.Add(w.nextRequestIntervalLocked()) w.mu.Unlock() now := time.Now() @@ -529,11 +673,40 @@ func (w *WrappedLive) runScheduler() { } } +// nextRequestIntervalLocked 计算下一次请求前应该等待的间隔。 +// 正常情况下就是配置的轮询间隔；连续失败时按 2 的幂次退避，最长为 +// max(maxFailureBackoff, 配置间隔)，一次成功即恢复到配置间隔。 +// 调用方必须持有 w.mu。 +func (w *WrappedLive) nextRequestIntervalLocked() time.Duration { + interval := time.Duration(w.getConfiguredInterval()) * time.Second + if interval <= 0 { + interval = defaultInterval + } + return failureBackoffInterval(interval, w.consecutiveFailures) +} + +func failureBackoffInterval(interval time.Duration, consecutiveFailures int) time.Duration { + maxBackoff := maxFailureBackoff + if interval > maxBackoff { + maxBackoff = interval + } + + backoff := interval + for i := 0; i < consecutiveFailures && backoff < maxBackoff; i++ { + if backoff > maxBackoff/2 { + backoff = maxBackoff + } else { + backoff *= 2 + } + } + return backoff +} + // getConfiguredInterval 获取此直播间配置的访问间隔（秒） func (w *WrappedLive) getConfiguredInterval() int { cfg := configs.GetCurrentConfig() if cfg == nil { - return 30 // 默认30秒 + return int(defaultInterval.Seconds()) } room, err := cfg.GetLiveRoomByUrl(w.GetRawUrl()) @@ -541,7 +714,7 @@ func (w *WrappedLive) getConfiguredInterval() int { if cfg.Interval > 0 { return cfg.Interval } - return 30 + return int(defaultInterval.Seconds()) } platformKey := configs.GetPlatformKeyFromUrl(w.GetRawUrl()) @@ -555,14 +728,58 @@ func randomJitter() int64 { return (time.Now().UnixNano() % 6001) - 3000 } -// waitForPlatformRateLimit 在通用位置等待平台访问频率限制 -// 使用 scheduler 的 context，这样在关闭时可以被取消 -func (w *WrappedLive) waitForPlatformRateLimit() bool { +// platformToolsReady 判断本直播间所属平台依赖的外部工具是否已就绪。 +func (w *WrappedLive) platformToolsReady() bool { + ready, _ := w.platformToolsReadyWithReason() + return ready +} + +// platformToolsReadyWithReason 判断依赖的外部工具是否已就绪，并返回未就绪的原因。 +// 状态发生变化时输出一条日志，方便用户理解「为什么这个直播间还没开始轮询」。 +func (w *WrappedLive) platformToolsReadyWithReason() (bool, string) { + checker := platformReadinessChecker + if checker == nil { + return true, "" + } + + platformKey := configs.GetPlatformKeyFromUrl(w.GetRawUrl()) + if platformKey == "" { + return true, "" + } + + ready, reason := checker(platformKey) + + w.mu.Lock() + wasWaiting := w.waitingForTool + w.waitingForTool = !ready + w.mu.Unlock() + + switch { + case !ready && !wasWaiting: + w.GetLogger().Debugf("依赖的工具尚未就绪，暂不请求直播间信息：%s", reason) + case ready && wasWaiting: + w.GetLogger().Debug("依赖的工具已就绪，恢复请求直播间信息") + } + + return ready, reason +} + +// acquirePlatformRequest 在通用位置获取平台请求许可。 +// 使用 scheduler 的 context，这样在关闭时可以取消排队。 +func (w *WrappedLive) acquirePlatformRequest() (release func(), ok bool) { platformKey := configs.GetPlatformKeyFromUrl(w.GetRawUrl()) if platformKey != "" { - return ratelimit.GetGlobalRateLimiter().WaitForPlatformWithContext(w.schedulerCtx, platformKey) + minInterval := 1 + if cfg := configs.GetCurrentConfig(); cfg != nil { + minInterval = cfg.GetPlatformMinAccessInterval(platformKey) + } + rateLimiter := ratelimit.GetGlobalRateLimiter() + // 添加新平台的第一个房间时，Live 会先于配置持久化创建；在请求入口兜底补齐 + // 默认限制，避免该时间窗内首次请求直接放行。 + rateLimiter.EnsurePlatformLimit(platformKey, minInterval) + return rateLimiter.AcquirePlatformWithContext(w.schedulerCtx, platformKey) } - return true + return func() {}, true } func New(ctx context.Context, room *configs.LiveRoom, cache gcache.Cache) (live Live, err error) { diff --git a/src/live/lives_test.go b/src/live/lives_test.go new file mode 100644 index 0000000..c8c2a89 --- /dev/null +++ b/src/live/lives_test.go @@ -0,0 +1,170 @@ +package live + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/bililive-go/bililive-go/src/types" +) + +type blockingInfoLive struct { + Live + calls atomic.Int32 + firstStarted chan struct{} + releaseFirst chan struct{} +} + +func (l *blockingInfoLive) GetRawUrl() string { return "" } + +func (l *blockingInfoLive) GetInfo() (*Info, error) { + if l.calls.Add(1) == 1 { + close(l.firstStarted) + <-l.releaseFirst + } + return &Info{}, nil +} + +type errorInfoLive struct{ Live } + +func (l *errorInfoLive) GetRawUrl() string { return "" } +func (l *errorInfoLive) GetLiveId() types.LiveID { return "error-test" } +func (l *errorInfoLive) GetPlatformCNName() string { return "测试平台" } +func (l *errorInfoLive) GetInfo() (*Info, error) { return nil, errors.New("请求失败") } + +// 未加载全局配置时 getConfiguredInterval 会直接返回默认间隔， +// 因此可以直接构造一个空的 WrappedLive 来验证退避曲线。 +func TestNextRequestIntervalLocked(t *testing.T) { + tests := []struct { + name string + failures int + want time.Duration + }{ + {name: "无失败时使用配置间隔", failures: 0, want: defaultInterval}, + {name: "一次失败翻倍", failures: 1, want: 2 * defaultInterval}, + {name: "两次失败四倍", failures: 2, want: 4 * defaultInterval}, + {name: "多次失败后封顶", failures: 100, want: maxFailureBackoff}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := &WrappedLive{consecutiveFailures: tt.failures} + w.mu.Lock() + got := w.nextRequestIntervalLocked() + w.mu.Unlock() + if got != tt.want { + t.Errorf("consecutiveFailures=%d: got %v, want %v", tt.failures, got, tt.want) + } + }) + } +} + +// 退避间隔必须单调不减，且永远不会超过上限， +// 否则失败时的请求频率反而会高于正常轮询频率。 +func TestNextRequestIntervalNeverShrinksBelowConfigured(t *testing.T) { + prev := time.Duration(0) + for failures := 0; failures < 64; failures++ { + w := &WrappedLive{consecutiveFailures: failures} + w.mu.Lock() + got := w.nextRequestIntervalLocked() + w.mu.Unlock() + if got < defaultInterval { + t.Fatalf("consecutiveFailures=%d: 退避间隔 %v 小于配置间隔 %v", failures, got, defaultInterval) + } + if got > maxFailureBackoff { + t.Fatalf("consecutiveFailures=%d: 退避间隔 %v 超过上限 %v", failures, got, maxFailureBackoff) + } + if got < prev { + t.Fatalf("consecutiveFailures=%d: 退避间隔 %v 小于上一次的 %v", failures, got, prev) + } + prev = got + } +} + +func TestFailureBackoffDoesNotShortenLongConfiguredInterval(t *testing.T) { + configured := 10 * time.Minute + for _, failures := range []int{0, 1, 100} { + if got := failureBackoffInterval(configured, failures); got < configured { + t.Fatalf("consecutiveFailures=%d: 退避间隔 %v 小于配置间隔 %v", failures, got, configured) + } + } +} + +func TestInitializingFailureCountsTowardBackoff(t *testing.T) { + w := &WrappedLive{} + placeholder := &Info{Initializing: true, LastError: "平台请求失败"} + + if failed := w.recordRequestResult(placeholder, nil); !failed { + t.Fatal("初始化占位信息携带 LastError 时应计为失败") + } + w.mu.Lock() + failures := w.consecutiveFailures + w.mu.Unlock() + if failures != 1 { + t.Fatalf("连续失败次数 = %d，期望 1", failures) + } + + if failed := w.recordRequestResult(&Info{}, nil); failed { + t.Fatal("正常信息不应计为失败") + } + w.mu.Lock() + failures = w.consecutiveFailures + w.mu.Unlock() + if failures != 0 { + t.Fatalf("成功后连续失败次数 = %d，期望清零", failures) + } +} + +func TestWrappedLiveSerializesConcurrentGetInfo(t *testing.T) { + underlying := &blockingInfoLive{ + firstStarted: make(chan struct{}), + releaseFirst: make(chan struct{}), + } + w := NewWrappedLive(context.Background(), underlying, nil).(*WrappedLive) + + done := make(chan struct{}, 2) + go func() { + _, _ = w.GetInfo() + done <- struct{}{} + }() + <-underlying.firstStarted + go func() { + _, _ = w.GetInfo() + done <- struct{}{} + }() + + time.Sleep(50 * time.Millisecond) + if calls := underlying.calls.Load(); calls != 1 { + t.Fatalf("首个请求完成前底层 GetInfo 被调用 %d 次，期望 1 次", calls) + } + + close(underlying.releaseFirst) + for range 2 { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("并发 GetInfo 未完成") + } + } +} + +func TestRequestStatusCallbackHandlesNilInfoOnFailure(t *testing.T) { + callbackCalled := false + SetRequestStatusCallback(func(_, _ string, success bool, errMsg string) { + callbackCalled = true + if success || errMsg != "请求失败" { + t.Fatalf("错误的请求状态回调：success=%v, err=%q", success, errMsg) + } + }) + t.Cleanup(func() { SetRequestStatusCallback(nil) }) + + w := NewWrappedLive(context.Background(), &errorInfoLive{}, nil).(*WrappedLive) + if _, err := w.GetInfo(); err == nil { + t.Fatal("底层请求失败时应返回错误") + } + if !callbackCalled { + t.Fatal("请求失败时未调用状态回调") + } +} diff --git a/src/live/system/initializing_live.go b/src/live/system/initializing_live.go index f61aa32..1d6f72e 100644 --- a/src/live/system/initializing_live.go +++ b/src/live/system/initializing_live.go @@ -29,6 +29,8 @@ type InitializingLive struct { onFinished live.InitializingFinishedCallback // 用于标记初始化是否已完成（防止重复触发回调） finished bool + // 是否已经记录过初始化失败日志（后续重试降级为 debug，避免刷屏） + failureLogged bool // 用于保护并发访问 mu sync.Mutex @@ -86,6 +88,20 @@ func (l *InitializingLive) GetInfo() (info *live.Info, err error) { // 尝试获取真实信息 realInfo, err := l.OriginalLive.GetInfo() if err != nil { + // 这里向上返回 nil error（初始化中是一个有效状态）， + // 但错误本身不能完全吞掉：否则「直播间一直卡在初始化中」在日志里 + // 没有任何线索，排查时完全看不出根因。 + // 首次失败记 warn，后续重试降为 debug，避免大量直播间同时刷屏。 + l.mu.Lock() + firstFailure := !l.failureLogged + l.failureLogged = true + l.mu.Unlock() + if firstFailure { + l.GetLogger().WithError(err).Warn("初始化直播间信息失败，将继续重试") + } else { + l.GetLogger().WithError(err).Debug("初始化直播间信息失败，将继续重试") + } + // 获取失败，返回初始化状态信息（使用缓存信息，允许后续重试） roomName := l.GetRawUrl() if cachedRoom != "" { @@ -97,6 +113,9 @@ func (l *InitializingLive) GetInfo() (info *live.Info, err error) { RoomName: roomName, Status: false, Initializing: true, + // 上层仍收到可展示的初始化占位信息和 nil error，但通过 LastError 可以把 + // 真实的平台失败计入调度退避，避免故障期间始终按基础间隔重试。 + LastError: err.Error(), }, nil // 返回 nil 错误，让上层知道这是一个有效的（初始化中的）状态 } diff --git a/src/livestate/store.go b/src/livestate/store.go index fba4af0..3cbdeb7 100644 --- a/src/livestate/store.go +++ b/src/livestate/store.go @@ -77,6 +77,12 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { if err != nil { return nil, fmt.Errorf("打开数据库失败: %w", err) } + if _, err := db.Exec("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;"); err != nil { + db.Close() + return nil, fmt.Errorf("设置数据库 PRAGMA 失败: %w", err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) store := &SQLiteStore{ db: db, @@ -123,6 +129,12 @@ func (s *SQLiteStore) runMigrations() error { if err != nil { return fmt.Errorf("恢复后重新打开数据库失败: %w", err) } + if _, err := db.Exec("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;"); err != nil { + db.Close() + return fmt.Errorf("恢复后设置数据库 PRAGMA 失败: %w", err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) s.db = db config.DB = s.db migrator, err = migration.NewMigrator(config) diff --git a/src/metrics/metrics.go b/src/metrics/metrics.go index f329cee..a7c5517 100644 --- a/src/metrics/metrics.go +++ b/src/metrics/metrics.go @@ -80,9 +80,11 @@ func (c collector) Collect(ch chan<- prometheus.Metric) { if r, err := c.inst.RecorderManager.(recorders.Manager).GetRecorder(context.Background(), id); err == nil { if status, err := r.GetStatus(); err == nil { - if value, err := strconv.ParseFloat(status["total_size"].(string), 64); err == nil { - ch <- prometheus.MustNewConstMetric(recorderTotalBytes, prometheus.CounterValue, value, - string(id), l.GetRawUrl(), info.HostName, info.RoomName) + if totalSize, ok := status["total_size"].(string); ok { + if value, err := strconv.ParseFloat(totalSize, 64); err == nil { + ch <- prometheus.MustNewConstMetric(recorderTotalBytes, prometheus.CounterValue, value, + string(id), l.GetRawUrl(), info.HostName, info.RoomName) + } } } } diff --git a/src/pkg/events/dispatcher.go b/src/pkg/events/dispatcher.go index 1c1afa4..41f33c4 100644 --- a/src/pkg/events/dispatcher.go +++ b/src/pkg/events/dispatcher.go @@ -28,6 +28,9 @@ type Dispatcher interface { RemoveEventListener(eventType EventType, listener *EventListener) RemoveAllEventListener(eventType EventType) DispatchEvent(event *Event) + // DispatchEventSync 在当前 goroutine 中按注册顺序执行事件处理器。 + // 仅应用于调用方必须等待状态交接完成后才能继续的场景；普通事件仍应使用 DispatchEvent。 + DispatchEventSync(event *Event) } type dispatcher struct { @@ -77,24 +80,49 @@ func (e *dispatcher) RemoveAllEventListener(eventType EventType) { e.saver = make(map[EventType]*list.List) } -func (e *dispatcher) DispatchEvent(event *Event) { +func (e *dispatcher) snapshotListeners(event *Event) []*EventListener { if event == nil { - return + return nil } e.RLock() listeners, ok := e.saver[event.Type] if !ok || listeners == nil { e.RUnlock() - return + return nil } - hs := make([]*EventListener, 0) + hs := make([]*EventListener, 0, listeners.Len()) for e := listeners.Front(); e != nil; e = e.Next() { hs = append(hs, e.Value.(*EventListener)) } e.RUnlock() + return hs +} + +func dispatchToListeners(event *Event, listeners []*EventListener) { + for _, listener := range listeners { + listener.Handler(event) + } +} + +func (e *dispatcher) DispatchEvent(event *Event) { + hs := e.snapshotListeners(event) + if len(hs) == 0 { + return + } bilisentry.Go(func() { - for _, h := range hs { - h.Handler(event) - } + dispatchToListeners(event, hs) }) } + +func (e *dispatcher) DispatchEventSync(event *Event) { + hs := e.snapshotListeners(event) + if len(hs) == 0 { + return + } + + // 与异步派发保持相同的 panic 隔离语义，避免单个事件处理器打断调用方。 + func() { + defer bilisentry.Recover() + dispatchToListeners(event, hs) + }() +} diff --git a/src/pkg/events/dispatcher_test.go b/src/pkg/events/dispatcher_test.go index be25a80..d779266 100644 --- a/src/pkg/events/dispatcher_test.go +++ b/src/pkg/events/dispatcher_test.go @@ -42,3 +42,39 @@ func TestDispatchEvent(t *testing.T) { time.Sleep(time.Second) assert.Equal(t, []int{0, 1, 2, 3}, l) } + +func TestDispatchEventSyncWaitsForHandlers(t *testing.T) { + d := NewDispatcher(context.Background()).(*dispatcher) + handlerStarted := make(chan struct{}) + releaseHandler := make(chan struct{}) + dispatchDone := make(chan struct{}) + + d.AddEventListener("test", NewEventListener(func(event *Event) { + close(handlerStarted) + <-releaseHandler + })) + + go func() { + d.DispatchEventSync(NewEvent("test", nil)) + close(dispatchDone) + }() + + select { + case <-handlerStarted: + case <-time.After(time.Second): + t.Fatal("同步事件处理器未启动") + } + + select { + case <-dispatchDone: + t.Fatal("同步派发在事件处理器完成前返回") + default: + } + + close(releaseHandler) + select { + case <-dispatchDone: + case <-time.After(time.Second): + t.Fatal("同步派发未在事件处理器完成后返回") + } +} diff --git a/src/pkg/events/mock/mock.go b/src/pkg/events/mock/mock.go index 0947632..65de180 100644 --- a/src/pkg/events/mock/mock.go +++ b/src/pkg/events/mock/mock.go @@ -77,6 +77,18 @@ func (mr *MockDispatcherMockRecorder) DispatchEvent(event any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DispatchEvent", reflect.TypeOf((*MockDispatcher)(nil).DispatchEvent), event) } +// DispatchEventSync mocks base method. +func (m *MockDispatcher) DispatchEventSync(event *events.Event) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "DispatchEventSync", event) +} + +// DispatchEventSync indicates an expected call of DispatchEventSync. +func (mr *MockDispatcherMockRecorder) DispatchEventSync(event any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DispatchEventSync", reflect.TypeOf((*MockDispatcher)(nil).DispatchEventSync), event) +} + // RemoveAllEventListener mocks base method. func (m *MockDispatcher) RemoveAllEventListener(eventType events.EventType) { m.ctrl.T.Helper() diff --git a/src/pkg/iostats/module.go b/src/pkg/iostats/module.go index fe8f1f0..6976344 100644 --- a/src/pkg/iostats/module.go +++ b/src/pkg/iostats/module.go @@ -83,6 +83,9 @@ func (m *Module) Close(ctx context.Context) { // 等待清理任务结束 m.cleanupWg.Wait() + // 停止请求追踪器（会把缓冲区中剩余的记录写完），必须在关闭存储之前 + m.tracker.Stop() + // 关闭存储 if err := m.store.Close(); err != nil { logrus.WithError(err).Error("关闭 IO 统计存储失败") diff --git a/src/pkg/iostats/request_tracker.go b/src/pkg/iostats/request_tracker.go index c55372b..b1d50db 100644 --- a/src/pkg/iostats/request_tracker.go +++ b/src/pkg/iostats/request_tracker.go @@ -5,21 +5,54 @@ import ( "sync" "time" + bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry" "github.com/sirupsen/logrus" ) +const ( + // requestStatusBufferSize 待落库的请求状态缓冲区大小。 + // 按几百个直播间、每轮各产生一条记录估算，一次批量写入之间足够容纳数轮。 + requestStatusBufferSize = 4096 + // requestStatusFlushInterval 批量落库的最长等待时间 + requestStatusFlushInterval = 2 * time.Second + // requestStatusFlushBatch 单次批量落库的最大条数 + requestStatusFlushBatch = 256 +) + // RequestTracker 请求状态追踪器 // 用于记录直播间请求的成功/失败状态 +// +// 记录动作发生在直播间轮询的关键路径上，因此不能同步写库： +// 原先每条记录都在一把全局锁下同步执行一次 SQLite 事务， +// 直播间数量多、平台又在报错时，轮询会被磁盘 I/O 串行拖住。 +// 这里改为投递到缓冲区，由后台 goroutine 批量落库。 type RequestTracker struct { store Store - mu sync.Mutex + + queue chan *RequestStatus + stopOnce sync.Once + stopCh chan struct{} + doneCh chan struct{} + + // acceptMu 保证 Stop 标记停止与 record 投递原子化，避免后台退出后仍有数据进入队列。 + acceptMu sync.RWMutex + stopped bool + + // droppedCount 缓冲区满时丢弃的记录数，仅用于日志提示 + droppedMu sync.Mutex + droppedCount int } -// NewRequestTracker 创建请求追踪器 +// NewRequestTracker 创建请求追踪器并启动后台落库 goroutine func NewRequestTracker(store Store) *RequestTracker { - return &RequestTracker{ - store: store, + t := &RequestTracker{ + store: store, + queue: make(chan *RequestStatus, requestStatusBufferSize), + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), } + bilisentry.Go(t.run) + return t } // RecordSuccess 记录成功的请求 @@ -32,10 +65,13 @@ func (t *RequestTracker) RecordFailure(liveID, platform string, errMsg string) { t.record(liveID, platform, false, errMsg) } -// record 内部记录方法 +// record 内部记录方法，不会阻塞调用方 func (t *RequestTracker) record(liveID, platform string, success bool, errMsg string) { - t.mu.Lock() - defer t.mu.Unlock() + t.acceptMu.RLock() + defer t.acceptMu.RUnlock() + if t.stopped { + return + } status := &RequestStatus{ Timestamp: time.Now().UnixMilli(), @@ -45,15 +81,89 @@ func (t *RequestTracker) record(liveID, platform string, success bool, errMsg st ErrorMessage: errMsg, } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() + select { + case t.queue <- status: + default: + // 缓冲区已满，丢弃这条统计数据。 + // 统计数据的完整性远不如录制本身重要，绝不能因此阻塞轮询。 + t.droppedMu.Lock() + t.droppedCount++ + t.droppedMu.Unlock() + } +} - if err := t.store.SaveRequestStatus(ctx, status); err != nil { - logrus.WithError(err).WithFields(logrus.Fields{ - "live_id": liveID, - "platform": platform, - "success": success, - }).Error("保存请求状态失败") +// Stop 停止后台落库 goroutine，并把缓冲区中剩余的数据写完 +func (t *RequestTracker) Stop() { + t.stopOnce.Do(func() { + t.acceptMu.Lock() + t.stopped = true + close(t.stopCh) + t.acceptMu.Unlock() + <-t.doneCh + }) +} + +// run 后台批量落库循环 +func (t *RequestTracker) run() { + defer close(t.doneCh) + + ticker := time.NewTicker(requestStatusFlushInterval) + defer ticker.Stop() + + batch := make([]*RequestStatus, 0, requestStatusFlushBatch) + + flush := func() { + if len(batch) == 0 { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + if err := t.store.SaveRequestStatuses(ctx, batch); err != nil { + logrus.WithError(err).WithField("count", len(batch)).Error("保存请求状态失败") + } + cancel() + batch = batch[:0] + } + + for { + select { + case status := <-t.queue: + batch = append(batch, status) + if len(batch) >= requestStatusFlushBatch { + flush() + } + case <-ticker.C: + flush() + t.reportDropped() + case <-t.stopCh: + // 退出前把已经排队的数据尽量写完 + for { + select { + case status := <-t.queue: + batch = append(batch, status) + if len(batch) >= requestStatusFlushBatch { + flush() + } + continue + default: + } + break + } + flush() + t.reportDropped() + return + } + } +} + +// reportDropped 汇总输出被丢弃的统计数据条数 +func (t *RequestTracker) reportDropped() { + t.droppedMu.Lock() + dropped := t.droppedCount + t.droppedCount = 0 + t.droppedMu.Unlock() + + if dropped > 0 { + logrus.WithField("count", dropped).Warn("请求状态统计写入积压，已丢弃部分记录") } } diff --git a/src/pkg/iostats/request_tracker_test.go b/src/pkg/iostats/request_tracker_test.go new file mode 100644 index 0000000..18d8e50 --- /dev/null +++ b/src/pkg/iostats/request_tracker_test.go @@ -0,0 +1,97 @@ +package iostats + +import ( + "context" + "sync" + "testing" + "time" +) + +// fakeStore 只实现请求状态的批量写入，其余方法继承自嵌入的接口（测试中不会调用） +type fakeStore struct { + Store + + mu sync.Mutex + saved []*RequestStatus + batches int + blockFor time.Duration +} + +func (f *fakeStore) SaveRequestStatuses(ctx context.Context, statuses []*RequestStatus) error { + if f.blockFor > 0 { + time.Sleep(f.blockFor) + } + f.mu.Lock() + defer f.mu.Unlock() + f.saved = append(f.saved, statuses...) + f.batches++ + return nil +} + +func (f *fakeStore) snapshot() (count, batches int) { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.saved), f.batches +} + +// Stop 必须把缓冲区里尚未落库的记录写完 +func TestRequestTrackerFlushesOnStop(t *testing.T) { + store := &fakeStore{} + tracker := NewRequestTracker(store) + + const total = 50 + for i := 0; i < total; i++ { + tracker.RecordFailure("live-1", "抖音", "请求失败: 500 Internal Server Error") + } + tracker.Stop() + + count, batches := store.snapshot() + if count != total { + t.Fatalf("落库记录数 = %d，期望 %d", count, total) + } + if batches == 0 { + t.Fatal("没有发生任何批量写入") + } + if batches > total { + t.Fatalf("批量写入次数 = %d，没有起到合批效果", batches) + } +} + +// 记录动作在轮询关键路径上，即使存储很慢也不能阻塞调用方 +func TestRequestTrackerRecordDoesNotBlock(t *testing.T) { + store := &fakeStore{blockFor: 50 * time.Millisecond} + tracker := NewRequestTracker(store) + defer tracker.Stop() + + // 远超缓冲区容量，超出的部分应被丢弃而不是阻塞 + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < requestStatusBufferSize*2; i++ { + tracker.RecordSuccess("live-1", "抖音") + } + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("记录请求状态阻塞了调用方") + } +} + +func TestRequestTrackerIgnoresRecordsAfterStop(t *testing.T) { + store := &fakeStore{} + tracker := NewRequestTracker(store) + tracker.RecordSuccess("before-stop", "抖音") + tracker.Stop() + + before, _ := store.snapshot() + tracker.RecordFailure("after-stop", "抖音", "不应入队") + if queued := len(tracker.queue); queued != 0 { + t.Fatalf("停止后仍有 %d 条记录进入无人消费的队列", queued) + } + after, _ := store.snapshot() + if after != before { + t.Fatalf("停止后的记录被写入：停止前 %d 条，停止后 %d 条", before, after) + } +} diff --git a/src/pkg/iostats/store.go b/src/pkg/iostats/store.go index 693a640..fccac2b 100644 --- a/src/pkg/iostats/store.go +++ b/src/pkg/iostats/store.go @@ -30,8 +30,8 @@ type Store interface { // QueryIOStats 查询 IO 统计数据 QueryIOStats(ctx context.Context, query IOStatsQuery) ([]IOStat, error) - // SaveRequestStatus 保存请求状态 - SaveRequestStatus(ctx context.Context, status *RequestStatus) error + // SaveRequestStatuses 批量保存请求状态 + SaveRequestStatuses(ctx context.Context, statuses []*RequestStatus) error // QueryRequestStatus 查询请求状态 QueryRequestStatus(ctx context.Context, query RequestStatusQuery) ([]RequestStatus, error) // QueryRequestStatusSegments 查询请求状态时间段（用于横条图） @@ -80,6 +80,10 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { if err != nil { return nil, fmt.Errorf("failed to open database: %w", err) } + if _, err := db.Exec("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;"); err != nil { + db.Close() + return nil, fmt.Errorf("failed to set pragmas: %w", err) + } // 设置连接池参数 db.SetMaxOpenConns(1) // SQLite 单写入 @@ -123,6 +127,12 @@ func (s *SQLiteStore) runMigrations() error { if err != nil { return fmt.Errorf("恢复后重新打开数据库失败: %w", err) } + if _, err := db.Exec("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;"); err != nil { + db.Close() + return fmt.Errorf("恢复后设置数据库 PRAGMA 失败: %w", err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) s.db = db config.DB = s.db migrator, err = migration.NewMigrator(config) @@ -317,22 +327,43 @@ func (s *SQLiteStore) aggregateStats(stats []IOStat, aggregation string) []IOSta return result } -// SaveRequestStatus 保存请求状态 -func (s *SQLiteStore) SaveRequestStatus(ctx context.Context, status *RequestStatus) error { +// SaveRequestStatuses 批量保存请求状态。 +// 直播间轮询失败时会在短时间内产生大量记录，逐条写入会带来大量事务提交， +// 因此调用方应尽量成批提交。 +func (s *SQLiteStore) SaveRequestStatuses(ctx context.Context, statuses []*RequestStatus) error { + if len(statuses) == 0 { + return nil + } + s.mu.Lock() defer s.mu.Unlock() - successInt := 0 - if status.Success { - successInt = 1 + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err } + defer tx.Rollback() - _, err := s.db.ExecContext(ctx, + stmt, err := tx.PrepareContext(ctx, `INSERT INTO request_status (timestamp, live_id, platform, success, error_message) VALUES (?, ?, ?, ?, ?)`, - status.Timestamp, status.LiveID, status.Platform, successInt, status.ErrorMessage, ) - return err + if err != nil { + return err + } + defer stmt.Close() + + for _, status := range statuses { + successInt := 0 + if status.Success { + successInt = 1 + } + if _, err = stmt.ExecContext(ctx, status.Timestamp, status.LiveID, status.Platform, successInt, status.ErrorMessage); err != nil { + return err + } + } + + return tx.Commit() } // QueryRequestStatus 查询请求状态 diff --git a/src/pkg/launcher/launcher.go b/src/pkg/launcher/launcher.go index 9277537..17dfcb8 100644 --- a/src/pkg/launcher/launcher.go +++ b/src/pkg/launcher/launcher.go @@ -5,6 +5,7 @@ package launcher import ( "context" "encoding/json" + "errors" "fmt" "os" "os/exec" @@ -43,7 +44,7 @@ type State struct { // DefaultState 返回默认状态 func DefaultState() *State { return &State{ - StartupTimeout: 60, + StartupTimeout: 120, MaxRetries: 3, } } @@ -265,7 +266,7 @@ func (r *Runner) Run(ctx context.Context, args []string) error { // 等待主程序确认启动或退出 // 注意：必须有 startupCh case，否则即使子进程秒回 startup_success， - // 也要等 StartupTimeout（60秒）才能进入 waitForMainProgram + // 也要等 StartupTimeout 才能进入 waitForMainProgram r.startupCh = make(chan struct{}) startupTimer := time.NewTimer(time.Duration(r.state.StartupTimeout) * time.Second) @@ -281,6 +282,9 @@ func (r *Runner) Run(ctx context.Context, args []string) error { case <-r.processDone: // 子进程在发送 startup_success 之前就退出了（如 Fatal 错误） startupTimer.Stop() + // 主程序可能已经退出，但它启动的 bililive-tools 等后代进程仍然存活。 + // 重试前必须清理整个进程组，否则新进程会因端口占用再次启动失败。 + r.cleanupMainProcessGroup() if !r.startupOK { r.log("主程序在启动确认前退出") @@ -374,6 +378,7 @@ func (r *Runner) startMainProgram(ctx context.Context, args []string) error { fmt.Sprintf("BILILIVE_LAUNCHER_PID=%d", os.Getpid()), fmt.Sprintf("BILILIVE_LAUNCHER_EXE=%s", launcherExe), ) + setProcAttr(r.mainProcess) if err := r.mainProcess.Start(); err != nil { return fmt.Errorf("启动进程失败: %w", err) @@ -424,10 +429,31 @@ func (r *Runner) stopMainProgram() { select { case <-waitCh: + // 即使主程序已正常退出，其未被回收的后代进程仍可能继续占用端口。 + r.cleanupMainProcessGroup() r.log("主程序已正常退出") case <-time.After(35 * time.Second): r.log("主程序未响应，强制终止") - r.mainProcess.Process.Kill() + r.cleanupMainProcessGroup() + if err := r.mainProcess.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + r.log("强制终止主程序失败: %v", err) + } + + // 等待 Wait goroutine 完成，确保主进程资源在启动下一版本前已回收。 + select { + case <-waitCh: + r.log("主程序已被强制终止") + case <-time.After(5 * time.Second): + r.log("等待主程序强制退出超时") + } + } +} + +// cleanupMainProcessGroup 清理主程序所在进程组中仍存活的后代进程。 +// 主程序无论是正常退出、启动失败还是被强制终止，Launcher 在继续前都必须调用它。 +func (r *Runner) cleanupMainProcessGroup() { + if err := killProcessGroup(r.mainPID); err != nil { + r.log("清理主程序进程组失败（PID: %d）: %v", r.mainPID, err) } } @@ -442,6 +468,7 @@ func (r *Runner) waitForMainProgram() { } else { r.mainProcess.Wait() } + r.cleanupMainProcessGroup() r.log("主程序已退出") } diff --git a/src/pkg/launcher/launcher_unix.go b/src/pkg/launcher/launcher_unix.go new file mode 100644 index 0000000..165c73e --- /dev/null +++ b/src/pkg/launcher/launcher_unix.go @@ -0,0 +1,28 @@ +//go:build !windows + +package launcher + +import ( + "errors" + "os/exec" + "syscall" +) + +// setProcAttr 设置独立进程组，强制终止时可通过 kill(-pgid) 同时清理所有子进程 +func setProcAttr(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} +} + +// killProcessGroup 杀掉整个进程组（包含 bililive-tools 等子进程），避免端口残留 +func killProcessGroup(pid int) error { + if pid <= 0 { + return nil + } + + err := syscall.Kill(-pid, syscall.SIGKILL) + // 主程序及其所有后代均已退出时，进程组不存在是正常状态。 + if errors.Is(err, syscall.ESRCH) { + return nil + } + return err +} diff --git a/src/pkg/launcher/launcher_unix_test.go b/src/pkg/launcher/launcher_unix_test.go new file mode 100644 index 0000000..95e10ac --- /dev/null +++ b/src/pkg/launcher/launcher_unix_test.go @@ -0,0 +1,136 @@ +//go:build !windows + +package launcher + +import ( + "fmt" + "net" + "os" + "os/exec" + "strconv" + "strings" + "testing" + "time" +) + +const launcherProcessHelperMode = "BILILIVE_LAUNCHER_PROCESS_HELPER" + +// TestLauncherProcessHelper 由 TestWaitForMainProgramCleansDescendants 作为子进程调用。 +func TestLauncherProcessHelper(t *testing.T) { + mode := os.Getenv(launcherProcessHelperMode) + if mode == "" { + return + } + + statePath := os.Getenv("BILILIVE_LAUNCHER_HELPER_STATE") + switch mode { + case "leader": + cmd := exec.Command(os.Args[0], "-test.run=^TestLauncherProcessHelper$") + cmd.Env = append(os.Environ(), + launcherProcessHelperMode+"=listener", + "BILILIVE_LAUNCHER_HELPER_STATE="+statePath, + ) + if err := cmd.Start(); err != nil { + t.Fatalf("启动监听子进程失败: %v", err) + } + + // 确保后代进程已经监听端口，再让主程序退出。 + waitForHelperState(t, statePath) + return + case "listener": + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatalf("监听测试端口失败: %v", err) + } + defer listener.Close() + + state := fmt.Sprintf("%s|%d", listener.Addr().String(), os.Getpid()) + if err := os.WriteFile(statePath, []byte(state), 0o600); err != nil { + t.Fatalf("写入测试状态失败: %v", err) + } + + for { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + _ = conn.Close() + } + default: + t.Fatalf("未知辅助进程模式: %s", mode) + } +} + +func TestWaitForMainProgramCleansDescendants(t *testing.T) { + statePath := t.TempDir() + "/listener.state" + cmd := exec.Command(os.Args[0], "-test.run=^TestLauncherProcessHelper$") + cmd.Env = append(os.Environ(), + launcherProcessHelperMode+"=leader", + "BILILIVE_LAUNCHER_HELPER_STATE="+statePath, + ) + setProcAttr(cmd) + if err := cmd.Start(); err != nil { + t.Fatalf("启动主测试进程失败: %v", err) + } + + runner := &Runner{ + mainProcess: cmd, + mainPID: cmd.Process.Pid, + processDone: make(chan struct{}), + } + go func() { + _ = cmd.Wait() + close(runner.processDone) + }() + + state := waitForHelperState(t, statePath) + parts := strings.Split(state, "|") + if len(parts) != 2 { + t.Fatalf("无效的测试状态: %q", state) + } + address := parts[0] + childPID, err := strconv.Atoi(parts[1]) + if err != nil { + t.Fatalf("无效的子进程 PID: %q", parts[1]) + } + defer func() { + if process, findErr := os.FindProcess(childPID); findErr == nil { + _ = process.Kill() + } + }() + + conn, err := net.DialTimeout("tcp4", address, time.Second) + if err != nil { + t.Fatalf("后代进程未监听测试端口: %v", err) + } + _ = conn.Close() + + // 主程序已经退出，但监听端口的后代进程仍在同一进程组中。 + // waitForMainProgram 必须在返回前清理该进程组。 + runner.waitForMainProgram() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + conn, err = net.DialTimeout("tcp4", address, 100*time.Millisecond) + if err != nil { + return + } + _ = conn.Close() + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("waitForMainProgram 返回后，后代进程仍占用端口 %s", address) +} + +func waitForHelperState(t *testing.T, statePath string) string { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + data, err := os.ReadFile(statePath) + if err == nil && len(data) > 0 { + return string(data) + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("等待辅助进程状态超时: %s", statePath) + return "" +} diff --git a/src/pkg/launcher/launcher_windows.go b/src/pkg/launcher/launcher_windows.go new file mode 100644 index 0000000..3f6f43a --- /dev/null +++ b/src/pkg/launcher/launcher_windows.go @@ -0,0 +1,12 @@ +//go:build windows + +package launcher + +import "os/exec" + +// setProcAttr 在 Windows 上无需设置 Unix 进程组。 +func setProcAttr(cmd *exec.Cmd) {} + +// killProcessGroup 在 Windows 上无需处理 Unix 进程组。 +// bililive-tools 等子进程由主程序创建的 Job Object 负责随主程序退出。 +func killProcessGroup(pid int) error { return nil } diff --git a/src/pkg/ratelimit/ratelimit.go b/src/pkg/ratelimit/ratelimit.go index c2da2ab..fd2a73b 100644 --- a/src/pkg/ratelimit/ratelimit.go +++ b/src/pkg/ratelimit/ratelimit.go @@ -18,6 +18,7 @@ type PlatformLimiter struct { minInterval time.Duration // 最小访问间隔 lastAccess time.Time // 上次访问时间 mu sync.Mutex // 保护访问时间的互斥锁 + inFlight chan struct{} // 限制同一平台最多一个请求在途 } var globalRateLimiter = &PlatformRateLimiter{ @@ -54,10 +55,30 @@ func (prl *PlatformRateLimiter) SetPlatformLimit(platform string, intervalSec in prl.limiters[platform] = &PlatformLimiter{ minInterval: interval, lastAccess: time.Time{}, // 零值时间，首次访问不会被限制 + inFlight: make(chan struct{}, 1), } } } +// EnsurePlatformLimit 在平台尚无限制时设置默认值，不覆盖已经显式配置的间隔。 +// 用于房间对象早于配置持久化创建的路径，保证首次请求也不会绕过平台保护。 +func (prl *PlatformRateLimiter) EnsurePlatformLimit(platform string, intervalSec int) { + if platform == "" || intervalSec <= 0 { + return + } + + prl.mu.Lock() + defer prl.mu.Unlock() + if _, exists := prl.limiters[platform]; exists { + return + } + prl.limiters[platform] = &PlatformLimiter{ + minInterval: time.Duration(intervalSec) * time.Second, + lastAccess: time.Time{}, + inFlight: make(chan struct{}, 1), + } +} + // WaitForPlatform 等待直到允许访问指定平台 // 如果平台没有设置限制，立即返回 // 注意：此函数在等待期间不持有锁，以允许 ForceAccess 等操作可以随时执行 @@ -78,6 +99,41 @@ func (prl *PlatformRateLimiter) WaitForPlatformWithContext(ctx context.Context, return true } + return waitForLimiterWithContext(ctx, limiter) +} + +// AcquirePlatformWithContext 获取指定平台的请求许可，并保证同一平台最多只有一个请求在途。 +// 返回的 release 必须在请求完成后调用；平台未配置限制时返回空操作 release。 +// +// 仅限制请求开始间隔不足以保护启动阶段：当前一个请求耗时超过最小间隔时，后续请求仍会 +// 重叠执行。这里把并发槽位与开始间隔合并为一次许可，使大量直播间并发初始化时仍按平台串行。 +func (prl *PlatformRateLimiter) AcquirePlatformWithContext(ctx context.Context, platform string) (release func(), ok bool) { + prl.mu.RLock() + limiter, exists := prl.limiters[platform] + prl.mu.RUnlock() + if !exists { + return func() {}, true + } + + select { + case limiter.inFlight <- struct{}{}: + case <-ctx.Done(): + return nil, false + } + + releaseSlot := func() { <-limiter.inFlight } + if !waitForLimiterWithContext(ctx, limiter) { + releaseSlot() + return nil, false + } + + var once sync.Once + return func() { + once.Do(releaseSlot) + }, true +} + +func waitForLimiterWithContext(ctx context.Context, limiter *PlatformLimiter) bool { for { // 检查 context 是否已取消 select { diff --git a/src/pkg/ratelimit/ratelimit_test.go b/src/pkg/ratelimit/ratelimit_test.go index 0cb0c58..6df3932 100644 --- a/src/pkg/ratelimit/ratelimit_test.go +++ b/src/pkg/ratelimit/ratelimit_test.go @@ -1,6 +1,7 @@ package ratelimit import ( + "context" "testing" "time" ) @@ -71,3 +72,51 @@ func TestConfigSyncRateLimits(t *testing.T) { // 这个测试需要配置系统的支持，暂时跳过具体实现 t.Skip("Config sync test requires full config system") } + +func TestAcquirePlatformSerializesInFlightRequests(t *testing.T) { + limiter := &PlatformRateLimiter{ + limiters: map[string]*PlatformLimiter{ + "test": { + inFlight: make(chan struct{}, 1), + }, + }, + } + + releaseFirst, ok := limiter.AcquirePlatformWithContext(context.Background(), "test") + if !ok { + t.Fatal("首次请求未获取到平台许可") + } + + secondAcquired := make(chan func(), 1) + go func() { + release, acquired := limiter.AcquirePlatformWithContext(context.Background(), "test") + if acquired { + secondAcquired <- release + } + }() + + select { + case release := <-secondAcquired: + release() + t.Fatal("首个请求未释放时，第二个同平台请求不应进入") + case <-time.After(50 * time.Millisecond): + } + + releaseFirst() + select { + case release := <-secondAcquired: + release() + case <-time.After(time.Second): + t.Fatal("首个请求释放后，第二个请求未获取平台许可") + } +} + +func TestEnsurePlatformLimitDoesNotOverrideExplicitLimit(t *testing.T) { + limiter := &PlatformRateLimiter{limiters: make(map[string]*PlatformLimiter)} + limiter.SetPlatformLimit("test", 5) + limiter.EnsurePlatformLimit("test", 1) + + if got := limiter.GetAllPlatformLimits()["test"]; got != 5 { + t.Fatalf("兜底限制覆盖了显式配置：得到 %d 秒，期望 5 秒", got) + } +} diff --git a/src/pkg/update/manager.go b/src/pkg/update/manager.go index b373eeb..74821c0 100644 --- a/src/pkg/update/manager.go +++ b/src/pkg/update/manager.go @@ -376,7 +376,7 @@ func (m *Manager) ApplyUpdateSelfHosted(ctx context.Context) error { ActiveVersion: info.Version, ActiveBinaryPath: newBinaryPath, BackupVersion: m.currentVer, - StartupTimeout: 60, + StartupTimeout: 120, MaxRetries: 3, } diff --git a/src/pkg/utils/conn_counter.go b/src/pkg/utils/conn_counter.go index c60cd58..e7d0022 100644 --- a/src/pkg/utils/conn_counter.go +++ b/src/pkg/utils/conn_counter.go @@ -14,6 +14,12 @@ import ( "github.com/bililive-go/bililive-go/src/pkg/proxy" ) +// infoRequestTimeout 限制直播间信息及播放地址等元数据请求的总耗时。 +// http.Transport 的 ResponseHeaderTimeout 只覆盖响应头；如果服务端返回响应头后停止发送 +// body，请求仍会永久阻塞。平台级在途限制会在整个 GetInfo 期间持有唯一许可，因此必须用 +// http.Client.Timeout 同时覆盖响应体读取，确保故障请求最终返回并释放许可。 +const infoRequestTimeout = 30 * time.Second + type ByteCounter struct { ReadBytes int64 WriteBytes int64 @@ -263,7 +269,10 @@ func CreateDefaultClient() *http.Client { // 应用信息获取代理（这些客户端主要用于获取直播间信息等 API 请求） proxy.ApplyInfoProxyToTransport(transport) - return &http.Client{Transport: transport} + return &http.Client{ + Transport: transport, + Timeout: infoRequestTimeout, + } } func CreateDownloadClient() *http.Client { @@ -307,5 +316,8 @@ func CreateConnCounterClient() (*http.Client, error) { // 应用信息获取代理（这些客户端主要用于获取直播间信息等 API 请求） proxy.ApplyInfoProxyToTransport(transport) - return &http.Client{Transport: transport}, nil + return &http.Client{ + Transport: transport, + Timeout: infoRequestTimeout, + }, nil } diff --git a/src/pkg/utils/conn_counter_test.go b/src/pkg/utils/conn_counter_test.go index f771a5c..b53993e 100644 --- a/src/pkg/utils/conn_counter_test.go +++ b/src/pkg/utils/conn_counter_test.go @@ -1,11 +1,15 @@ package utils import ( + "context" "crypto/tls" "errors" + "io" "net/http" + "net/http/httptest" "sync" "testing" + "time" "github.com/stretchr/testify/assert" ) @@ -231,6 +235,7 @@ func TestCreateDefaultClient(t *testing.T) { assert.NotNil(t, client) assert.NotNil(t, client.Transport) + assert.Equal(t, infoRequestTimeout, client.Timeout) transport, ok := client.Transport.(*http.Transport) assert.True(t, ok, "Transport should be *http.Transport") @@ -250,6 +255,7 @@ func TestCreateDownloadClient(t *testing.T) { assert.NotNil(t, client) assert.NotNil(t, client.Transport) + assert.Zero(t, client.Timeout, "直播流下载不能使用信息请求的总超时") transport, ok := client.Transport.(*http.Transport) assert.True(t, ok, "Transport should be *http.Transport") @@ -269,6 +275,7 @@ func TestCreateConnCounterClient(t *testing.T) { assert.NoError(t, err) assert.NotNil(t, client) assert.NotNil(t, client.Transport) + assert.Equal(t, infoRequestTimeout, client.Timeout) transport, ok := client.Transport.(*http.Transport) assert.True(t, ok, "Transport should be *http.Transport") @@ -282,3 +289,35 @@ func TestCreateConnCounterClient(t *testing.T) { assert.Greater(t, transport.TLSHandshakeTimeout.Seconds(), 0.0) assert.Greater(t, transport.ResponseHeaderTimeout.Seconds(), 0.0) } + +func TestInfoClientTimeoutCoversResponseBody(t *testing.T) { + requestCanceled := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(requestCanceled) + w.WriteHeader(http.StatusOK) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + <-r.Context().Done() + })) + t.Cleanup(server.Close) + + client := CreateDefaultClient() + client.Timeout = 50 * time.Millisecond + resp, err := client.Get(server.URL) + if err != nil { + t.Fatalf("读取响应头失败: %v", err) + } + defer resp.Body.Close() + + _, err = io.ReadAll(resp.Body) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("响应体读取错误 = %v，期望请求总超时", err) + } + + select { + case <-requestCanceled: + case <-time.After(time.Second): + t.Fatal("请求总超时后服务端未收到 context 取消信号") + } +} diff --git a/src/recorders/recorder.go b/src/recorders/recorder.go index f539acc..97baa22 100644 --- a/src/recorders/recorder.go +++ b/src/recorders/recorder.go @@ -381,7 +381,8 @@ func (r *recorder) tryRecord(ctx context.Context) { buf := new(bytes.Buffer) if err = tmpl.Execute(buf, info); err != nil { - panic(fmt.Sprintf("failed to render filename, err: %v", err)) + r.getLogger().WithError(err).Error("failed to render filename, recording aborted") + return } // 使用层级配置的 OutPutPath fileName := filepath.Join(resolvedConfig.OutPutPath, buf.String()) diff --git a/src/recorders/recorder_test.go b/src/recorders/recorder_test.go new file mode 100644 index 0000000..f0899c9 --- /dev/null +++ b/src/recorders/recorder_test.go @@ -0,0 +1,52 @@ +package recorders + +import ( + "context" + "net/url" + "strings" + "testing" + + "github.com/bluele/gcache" + "github.com/sirupsen/logrus" + gomock "go.uber.org/mock/gomock" + + "github.com/bililive-go/bililive-go/src/configs" + "github.com/bililive-go/bililive-go/src/live" + livemock "github.com/bililive-go/bililive-go/src/live/mock" + "github.com/bililive-go/bililive-go/src/pkg/livelogger" +) + +func TestTryRecordStopsWithoutPanicWhenFilenameRenderFails(t *testing.T) { + previousConfig := configs.GetCurrentConfig() + cfg := configs.NewConfig() + cfg.OutputTmpl = `{{ fail "render failed" }}` + configs.SetCurrentConfig(cfg) + t.Cleanup(func() { + configs.SetCurrentConfig(previousConfig) + }) + + ctrl := gomock.NewController(t) + l := livemock.NewMockLive(ctrl) + logger := livelogger.New(0, logrus.Fields{"test": t.Name()}) + streamURL := &url.URL{Scheme: "https", Host: "example.com", Path: "/stream.flv"} + + l.EXPECT().GetRawUrl().Return("https://example.com/room").AnyTimes() + l.EXPECT().GetStreamInfos().Return([]*live.StreamUrlInfo{{Url: streamURL}}, nil) + l.EXPECT().GetLogger().Return(logger) + + cache := gcache.New(1).LRU().Build() + if err := cache.Set(l, &live.Info{Live: l}); err != nil { + t.Fatalf("写入直播信息缓存失败: %v", err) + } + r := &recorder{Live: l, cache: cache} + + r.tryRecord(context.Background()) + + logs := logger.GetLogs() + if !strings.Contains(logs, "failed to render filename, recording aborted") { + t.Fatalf("未记录文件名渲染失败日志: %s", logs) + } + if !strings.Contains(logs, "render failed") { + t.Fatalf("日志未保留原始模板错误: %s", logs) + } +} diff --git a/src/tools/tools.go b/src/tools/tools.go index edf1899..7d2d01d 100644 --- a/src/tools/tools.go +++ b/src/tools/tools.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net" + "net/http" "os" "os/exec" "path/filepath" @@ -306,13 +307,92 @@ const ( BToolsStatusFailed ) +const ( + // BToolsPort bililive-tools 本地 HTTP 服务监听的端口 + BToolsPort = 18110 + // BToolsAuthToken 访问 bililive-tools 本地 HTTP 服务所需的 Authorization 头 + BToolsAuthToken = "Basic YTph" + + // btoolsHealthCheckInterval 健康检查的轮询间隔 + btoolsHealthCheckInterval = time.Second + // btoolsHealthCheckTimeout bililive-tools 长时间未就绪时输出一次警告的间隔。 + // 首次启动需要 Node 加载依赖，慢盘上可能相当久，因此超时后继续探测而不是永久失败。 + btoolsHealthCheckTimeout = 3 * time.Minute +) + var currentBToolsStatus atomic.Int32 -// IsBToolsReady 检查 bililive-tools 是否已就绪 +// IsBToolsReady 检查 bililive-tools 是否已就绪。 +// 就绪的含义是「本地 HTTP 服务已经能够响应请求」，而不仅仅是进程已经拉起。 func IsBToolsReady() bool { return btoolsStatusValue(currentBToolsStatus.Load()) == BToolsStatusReady } +// GetBToolsStatus 返回 bililive-tools 当前状态 +func GetBToolsStatus() btoolsStatusValue { + return btoolsStatusValue(currentBToolsStatus.Load()) +} + +// waitBToolsHealthy 轮询 bililive-tools 的本地 HTTP 端口，直到它能够响应请求为止。 +// 只要拿到任意 HTTP 响应（哪怕是 404/500）就说明服务已经在监听并进入了请求处理流程。 +// +// 进程被拉起 ≠ 服务可用：Node 侧还需要若干秒完成初始化。如果在这段时间里就宣布就绪， +// 上层几百个直播间会在服务刚 listen 的瞬间一拥而上，把它直接打爆。 +func waitBToolsHealthy() { + client := &http.Client{Timeout: 5 * time.Second} + endpoint := fmt.Sprintf("http://127.0.0.1:%d/", BToolsPort) + deadline := time.Now().Add(btoolsHealthCheckTimeout) + + for { + // 进程已经退出或启动失败时不必再等 + if GetBToolsStatus() != BToolsStatusStarting { + return + } + + req, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err == nil { + req.Header.Set("Authorization", BToolsAuthToken) + if resp, doErr := client.Do(req); doErr == nil { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if currentBToolsStatus.CompareAndSwap(int32(BToolsStatusStarting), int32(BToolsStatusReady)) { + blog.GetLogger().Infoln("bililive-tools 服务已就绪") + } + return + } + } + + // 超时只表示启动异常缓慢，不能永久停止唯一的健康探测：子进程此时可能仍然 + // 存活并在稍后恢复响应。继续探测可以让平台依赖自动恢复，无需重启整个应用。 + if time.Now().After(deadline) { + blog.GetLogger().Warnf("bililive-tools 在 %s 内尚未就绪，将继续在后台探测", btoolsHealthCheckTimeout) + deadline = time.Now().Add(btoolsHealthCheckTimeout) + } + + time.Sleep(btoolsHealthCheckInterval) + } +} + +// PlatformDependenciesReady 返回指定平台（configs 中的 platformKey）依赖的外部工具是否已就绪， +// 以及未就绪时可供展示的原因。 +// +// 工具未就绪时不应该去请求直播间信息：请求必然失败，只会把本地工具和平台接口打爆， +// 还会让直播间被误判成未开播。等工具就绪后再开始轮询即可，WebUI 本身不受影响。 +func PlatformDependenciesReady(platformKey string) (bool, string) { + if platformKey != configs.PlatformKeyDouyin { + return true, "" + } + + switch GetBToolsStatus() { + case BToolsStatusReady: + return true, "" + case BToolsStatusFailed: + return false, "bililive-tools 启动失败，抖音直播间无法录制" + default: + return false, "bililive-tools 尚未就绪（正在下载或启动中）" + } +} + // IsBToolsStarting 检查 bililive-tools 是否正在启动 func IsBToolsStarting() bool { status := btoolsStatusValue(currentBToolsStatus.Load()) @@ -326,6 +406,9 @@ func IsBToolsStarting() bool { func Cleanup() { logger := blog.GetLogger() + // 先置位关闭标记，避免守护逻辑把刚被杀掉的子进程又拉起来 + btoolsShuttingDown.Store(true) + // 1. 终止所有已注册的子进程 KillAllProcesses() @@ -516,12 +599,7 @@ func Init() (err error) { } { AsyncDownloadIfNecessary(toolName) } - bilisentry.Go(func() { - err := startBTools() - if err != nil { - blog.GetLogger().WithError(err).Errorln("Failed to start bililive-tools") - } - }) + bilisentry.Go(superviseBTools) bilisentry.Go(func() { if cfg := configs.GetCurrentConfig(); cfg != nil && cfg.RPC.Enable { startScheduler() @@ -531,6 +609,51 @@ func Init() (err error) { return nil } +// btoolsShuttingDown 标记程序正在退出，避免守护逻辑在关闭过程中重启子进程 +var btoolsShuttingDown atomic.Bool + +// superviseBTools 启动 bililive-tools 并在它意外退出后自动重启。 +// +// 没有守护的话，Node 进程一旦崩溃（或被外部结束），抖音等依赖它的平台 +// 在本次运行的剩余时间里就再也无法恢复，只能靠用户重启整个程序。 +func superviseBTools() { + const ( + minRestartDelay = 5 * time.Second + maxRestartDelay = 5 * time.Minute + // 运行超过这个时长才认为「曾经正常工作过」，可以把退避重置 + healthyRunDuration = time.Minute + ) + + delay := minRestartDelay + for { + startedAt := time.Now() + if err := startBTools(); err != nil { + blog.GetLogger().WithError(err).Errorln("Failed to start bililive-tools") + } + + if btoolsShuttingDown.Load() { + return + } + + // 之前稳定运行过一段时间，说明不是启动就失败，重置退避 + if time.Since(startedAt) >= healthyRunDuration { + delay = minRestartDelay + } + + blog.GetLogger().Warnf("bililive-tools 将在 %s 后重启", delay) + time.Sleep(delay) + + if btoolsShuttingDown.Load() { + return + } + + delay *= 2 + if delay > maxRestartDelay { + delay = maxRestartDelay + } + } +} + func startBTools() error { // 设置状态为正在启动 currentBToolsStatus.Store(int32(BToolsStatusStarting)) @@ -598,16 +721,30 @@ func startBTools() error { blog.GetLogger().Infoln("Starting bililive-tools server…") - // 设置状态为已就绪（服务已启动） - currentBToolsStatus.Store(int32(BToolsStatusReady)) + // 状态保持 Starting，由健康检查在服务真正可以响应请求后置为 Ready， + // 避免进程刚拉起、HTTP 端口还没 listen 时上层就一拥而上。 + bilisentry.Go(waitBToolsHealthy) // 在 Windows 下使用 Job Object，确保主进程退出时子进程被一并终止 // 使用 runWithKillOnCloseAndGetPID 来获取进程 PID - return runWithKillOnCloseAndGetPID(cmd, func(pid int) { + // 该调用会阻塞到子进程退出为止 + err = runWithKillOnCloseAndGetPID(cmd, func(pid int) { // 使用通用的进程跟踪器注册子进程 RegisterProcess("bililive-tools", pid, ProcessCategoryBTools) blog.GetLogger().Infof("bililive-tools process started with PID: %d", pid) }) + + // 进程已经退出，及时反注册，避免 KillAllProcesses 去操作一个已经失效 + // （甚至可能被系统回收复用）的 PID + UnregisterProcess("bililive-tools") + + // 子进程退出后服务不再可用，必须把状态改回去， + // 否则依赖它的平台会继续认为服务健康并不断发出必然失败的请求 + if previous := GetBToolsStatus(); previous != BToolsStatusFailed { + currentBToolsStatus.Store(int32(BToolsStatusFailed)) + blog.GetLogger().WithError(err).Warnln("bililive-tools 进程已退出，依赖它的平台（如抖音）将暂停请求") + } + return err } func AsyncDownloadIfNecessary(toolName string) {