弹幕写盘出错不再被静默吞掉

审查发现：录制时每写一条弹幕都调用 os.File.Write / fmt.Fprintf，
但返回值一个都没接。磁盘写满、外接盘或网络盘掉线时，写入会失败而程序
毫无察觉 —— 照样打印「录制完成」、照样自动合并，最后拿到的成片缺了一大截
弹幕，界面上看不出是「缺了」还是「本来就没弹幕」。

1. 写入器记住写盘错误
   danmuWriter 增加一个 err 字段：每条弹幕的 ass / nick.ass / jsonl 写入，
   以及文件头的写入，都检查返回值。只保留第一个错误 —— 磁盘满了之后
   每写一条都会失败，全记下来只会把日志刷爆，而第一个就足以说明原因。

2. close() 负责把错误交出来
   三个文件的 Close 错误也要记：os.File 在部分系统上（NFS、某些网络盘）
   是到 Close 才报出「写不进去」，只盯每次 Write 的返回值会漏掉这种情况。
   getDanmu 里拿到错误就明确告警，并把这条记录标记成「弹幕文件写坏了」。

3. 写坏的本场跳过自动合并
   光打一条日志不够 —— 照样会合成一个缺弹幕的成片。liveInfo 增加
   danmuWriteFailed，finishPart 据此跳过合并，录播和弹幕文件都留着，
   用户检查磁盘后还能手动合并。

已验证：
- 新增 4 个用例：
  TestDanmuWriteErrorSurfacedOnClose（写入失败后 failed() 为真、
  close() 必须把错误交出来、错误里能看出是哪个文件）、
  TestDanmuWriteKeepsOnlyFirstError（只记第一个）、
  TestDanmuWriteNoFalseAlarm（正常录制不能误判，这是加检查最容易踩的反向坑）、
  TestFinishPartSkipsMergeWhenDanmuBroken（先跑一个「条件都满足应当合并」
  的对照组确认测试本身有效，再验证标了写坏之后不合并）。
- 反向验证：把 jsonl 的两处 Write 错误改回丢弃，
  TestDanmuWriteErrorSurfacedOnClose 立刻报「failed() 应为 true」。
- 测试踩到的两个坑（都记进注释）：替换 w.jsonl 前要先关掉原句柄否则
  Windows 上临时目录删不掉；测试里没跑 initialize()，lInfoMap.info 是 nil，
  直接赋值会 panic。
- gofmt / go vet / go build / GOOS=linux go build / go test / go test -race 全过，
  全量测试连跑两次稳定。

automerge.go

@@ -29,10 +29,15 @@ func finishPart(liveID string, danmu bool) {
 		info.isRecording = false
 		info.recordDone = true
 	}
-	shouldMerge := config.AutoMerge && mainCtx.Err() == nil &&
+	// 弹幕文件写坏过就别合并了：成片会缺一大截弹幕，界面上看不出是缺还是本来就没弹幕
+	broke := info.danmuWriteFailed
+	shouldMerge := config.AutoMerge && mainCtx.Err() == nil && !broke &&
 		!info.danmuStopped && !info.recordStopped &&
 		info.danmuDone && info.recordDone && !info.mergeStarted &&
 		info.assFile != "" && info.firstRecordFile != ""
+	if broke && config.AutoMerge && info.danmuDone && info.recordDone && !info.mergeStarted {
+		lPrintWarn("本场弹幕文件在录制时写盘出错，跳过自动合并（录播和弹幕文件都留着）")
+	}
 	var segs []recordSegment
 	if shouldMerge {
 		info.mergeStarted = true
@@ -242,9 +247,15 @@ func remapJsonlTimes(srcFile, dstFile string, starts []int64, durs []float64, da
 	return nil
 }
 
-// 自动合并入口：把结果写进任务进度；日志措辞仍由内部负责
+// 自动合并入口：排队跑完一场，把结果写进任务进度；日志措辞仍由内部负责
 func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int64) {
-	err := autoMergeJob(uid, segs, jsonlFile, danmuStartNs)
+	// 多个主播可能同时下播，而全局只有一份任务状态：先排队，跑到了再登记进度
+	if jobActive() {
+		lPrintf("已有合并任务在跑（%s），本场先排队等待", jobBusyName())
+	}
+	err := runMergeExclusive(func() error {
+		return autoMergeJob(uid, segs, jsonlFile, danmuStartNs)
+	})
 	if err == errJobStopped {
 		// 停止不是故障，内部不打印，这里补一句
 		lPrintWarn("自动合并已被手动停止" + longID(uid))
@@ -300,7 +311,10 @@ func autoMergeJob(uid int, segs []recordSegment, jsonlFile string, danmuStartNs
 			inputBytes += fileSizeOrZero(p)
 		}
 	}
-	jobStart("自动合并", filepath.Base(outFile), inputBytes, 0)
+	if !jobStart("自动合并", filepath.Base(outFile), inputBytes, 0) {
+		// 走到这里说明没先拿 mergeSlot，与其把别人的进度冲掉不如直接放弃这一场
+		return errors.New("开始任务失败：还有别的合并任务没结束，本场合并已跳过")
+	}
 
 	video, jsonl, offset, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonlFile, danmuStartNs)
 	defer cleanup()

avatar_test.go

@@ -10,9 +10,14 @@ import (
 	"image/gif"
 	"image/jpeg"
 	"image/png"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 )
 
 // 2x2 的静态 webp（纯红），用来确认 webp 解码器确实注册上了
@@ -167,3 +172,97 @@ func TestEnsureAvatarMigratesLegacyCache(t *testing.T) {
 		t.Errorf("命中缓存应返回 %q，实际 %q", want, got)
 	}
 }
+
+// ==================== 头像下载的容量上限 ====================
+// CDN 的响应大小和图片尺寸都是外部输入，必须有上限。
+// 改之前 downloadAvatar 用的是裸 io.ReadAll：实测一个头像就能吃掉 512MB 内存。
+
+// 异常大的响应必须在读完之前就被截住，不能整份读进内存
+func TestDownloadAvatarCapsResponseBody(t *testing.T) {
+	var wanted int64 = avatarMaxBytes + 8<<20 // 比上限多 8MB
+	var served int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "image/png")
+		buf := make([]byte, 1<<20)
+		for atomic.AddInt64(&served, int64(len(buf))) < wanted {
+			if _, err := w.Write(buf); err != nil {
+				return
+			}
+		}
+	}))
+	defer srv.Close()
+
+	out := filepath.Join(t.TempDir(), "a.png")
+	done := make(chan error, 1)
+	go func() { done <- downloadAvatar(1, srv.URL, out) }()
+
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("超过体积上限的头像应当报错，实际下载成功了")
+		}
+		if !strings.Contains(err.Error(), "超过") {
+			t.Errorf("报错要说清是体积超限，实际：%v", err)
+		}
+		// 客户端只读到上限就撒手，服务端这边根本写不完整个响应。
+		// 反过来，如果 served 达到了 wanted，说明整份都被读进内存了 —— 限制没生效
+		if got := atomic.LoadInt64(&served); got >= wanted {
+			t.Errorf("整份响应都被读完了（%.1f MB），说明没有截断（上限 %d MB）",
+				float64(got)/1048576, avatarMaxBytes>>20)
+		}
+		t.Logf("  已接收 %.1f MB 时就被截住（上限 %d MB）",
+			float64(atomic.LoadInt64(&served))/1048576, avatarMaxBytes>>20)
+	case <-time.After(30 * time.Second):
+		t.Fatalf("超过 30 秒还没结束，限制没生效（已接收 %.1f MB）",
+			float64(atomic.LoadInt64(&served))/1048576)
+	}
+	if _, err := os.Stat(out); err == nil {
+		t.Error("下载失败不该留下文件")
+	}
+}
+
+// 正常的头像是能下下来的：上限不能误伤
+func TestDownloadAvatarAcceptsNormalSize(t *testing.T) {
+	var buf bytes.Buffer
+	if err := png.Encode(&buf, image.NewRGBA(image.Rect(0, 0, 96, 96))); err != nil {
+		t.Fatal(err)
+	}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "image/png")
+		_, _ = w.Write(buf.Bytes())
+	}))
+	defer srv.Close()
+
+	out := filepath.Join(t.TempDir(), "ok.png")
+	if err := downloadAvatar(1, srv.URL, out); err != nil {
+		t.Fatalf("正常大小的头像应当能下载，实际报错：%v", err)
+	}
+	if _, err := os.Stat(out); err != nil {
+		t.Errorf("下载成功却没生成文件：%v", err)
+	}
+}
+
+// 压缩炸弹：几十 KB 的图可以解压出上亿像素，解码后必须再检查一次尺寸
+func TestDecodeAvatarCapsPixels(t *testing.T) {
+	side := 3000 // 900 万像素，刚过 2048x2048 的上限
+	big := image.NewGray(image.Rect(0, 0, side, side))
+	var buf bytes.Buffer
+	if err := png.Encode(&buf, big); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := decodeAvatar(buf.Bytes(), "image/png"); err == nil {
+		t.Fatalf("%dx%d 的头像应当被尺寸上限拦住", side, side)
+	} else if !strings.Contains(err.Error(), "超出上限") {
+		t.Errorf("报错要说清是尺寸超限，实际：%v", err)
+	}
+
+	// 常见的头像尺寸不受影响
+	ok := image.NewRGBA(image.Rect(0, 0, 300, 300))
+	var b2 bytes.Buffer
+	if err := png.Encode(&b2, ok); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := decodeAvatar(b2.Bytes(), "image/png"); err != nil {
+		t.Errorf("300x300 的普通头像不该被拦：%v", err)
+	}
+}

danmu.go

@@ -165,6 +165,21 @@ type danmuWriter struct {
 	jsonl *os.File
 	lastO []laneState // 原格式的弹道状态
 	lastN []laneState // 昵称头像版的弹道状态
+	err   error       // 写盘出过的错。只记头一个，见 danmuWriter.fail
+}
+
+// 记一次写盘错误。只保留第一个：磁盘满了之后每写一条都会失败，
+// 全记下来只会把日志刷爆，而第一个就足以说明原因
+func (w *danmuWriter) fail(what string, err error) {
+	if w.err == nil {
+		w.err = fmt.Errorf("%s失败：%v", what, err)
+	}
+}
+
+// 本场弹幕有没有写坏。写坏的文件不能当正常的用 —— 否则合并出来的成片
+// 会缺一大截弹幕，而日志里什么都看不出来
+func (w *danmuWriter) failed() bool {
+	return w.err != nil
 }
 
 // 返回昵称头像版 ASS 和 jsonl 的文件路径
@@ -189,17 +204,33 @@ func newDanmuWriter(liveID, streamName string, cfg acfundanmu.SubConfig, assFile
 	w.jsonl, err = os.OpenFile(jsonlFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 	checkErr(err)
 
-	fmt.Fprintf(w.ass, scriptInfoTmpl, liveID, streamName, cfg.Title, cfg.PlayResX, cfg.PlayResY)
-	fmt.Fprintf(w.ass, stylesTmpl, cfg.FontSize)
-	fmt.Fprint(w.ass, eventsTmpl)
-	fmt.Fprintf(w.nick, scriptInfoTmpl, liveID, streamName, cfg.Title+" 昵称头像版", cfg.PlayResX, cfg.PlayResY)
-	fmt.Fprintf(w.nick, stylesTmpl, cfg.FontSize)
-	fmt.Fprint(w.nick, eventsTmpl)
+	if _, err := fmt.Fprintf(w.ass, scriptInfoTmpl, liveID, streamName, cfg.Title, cfg.PlayResX, cfg.PlayResY); err != nil {
+		w.fail("写原格式 ASS 文件头", err)
+	}
+	if _, err := fmt.Fprintf(w.ass, stylesTmpl, cfg.FontSize); err != nil {
+		w.fail("写原格式 ASS 文件头", err)
+	}
+	if _, err := fmt.Fprint(w.ass, eventsTmpl); err != nil {
+		w.fail("写原格式 ASS 文件头", err)
+	}
+	if _, err := fmt.Fprintf(w.nick, scriptInfoTmpl, liveID, streamName, cfg.Title+" 昵称头像版", cfg.PlayResX, cfg.PlayResY); err != nil {
+		w.fail("写昵称头像版 ASS 文件头", err)
+	}
+	if _, err := fmt.Fprintf(w.nick, stylesTmpl, cfg.FontSize); err != nil {
+		w.fail("写昵称头像版 ASS 文件头", err)
+	}
+	if _, err := fmt.Fprint(w.nick, eventsTmpl); err != nil {
+		w.fail("写昵称头像版 ASS 文件头", err)
+	}
 
 	meta, err := json.Marshal(danmuMeta{PlayResX: cfg.PlayResX, PlayResY: cfg.PlayResY, FontSize: cfg.FontSize, StartTime: cfg.StartTime})
 	checkErr(err)
-	w.jsonl.Write(meta)
-	w.jsonl.Write([]byte("\n"))
+	if _, err := w.jsonl.Write(meta); err != nil {
+		w.fail("写弹幕 jsonl 文件头", err)
+	}
+	if _, err := w.jsonl.Write([]byte("\n")); err != nil {
+		w.fail("写弹幕 jsonl 文件头", err)
+	}
 	return w
 }
 
@@ -218,10 +249,12 @@ func (w *danmuWriter) write(c *acfundanmu.Comment) {
 	}
 	leftO := rel + (int64(s.PlayResX)*danmuShowDur)/int64(s.PlayResX+length)
 	if i := placeLane(w.lastO, dtO, leftO); i >= 0 {
-		fmt.Fprintf(w.ass, dialogueTmpl,
+		if _, err := fmt.Fprintf(w.ass, dialogueTmpl,
 			assTime(dtO.appear), assTime(dtO.disappear),
 			strings.ReplaceAll(c.Nickname, ",", " "), c.UserID,
-			s.PlayResX+length/2, s.FontSize*(i+1), -length/2, s.FontSize*(i+1), c.Content)
+			s.PlayResX+length/2, s.FontSize*(i+1), -length/2, s.FontSize*(i+1), c.Content); err != nil {
+			w.fail("写原格式 ASS", err)
+		}
 	}
 
 	// 昵称头像版 ASS，\an7 锚点在文本左上角，头像由 ffmpeg 叠在文字左侧，不依赖字体渲染宽度
@@ -236,10 +269,12 @@ func (w *danmuWriter) write(c *acfundanmu.Comment) {
 	}
 	leftN := rel + (int64(s.PlayResX)*danmuShowDur)/int64(s.PlayResX+total)
 	if i := placeLane(w.lastN, dtN, leftN); i >= 0 {
-		fmt.Fprintf(w.nick, dialogueNickTmpl,
+		if _, err := fmt.Fprintf(w.nick, dialogueNickTmpl,
 			assTime(dtN.appear), assTime(dtN.disappear),
 			strings.ReplaceAll(c.Nickname, ",", " "), c.UserID,
-			s.PlayResX+avW+avGap, s.FontSize*i, -width, s.FontSize*i, text)
+			s.PlayResX+avW+avGap, s.FontSize*i, -width, s.FontSize*i, text); err != nil {
+			w.fail("写昵称头像版 ASS", err)
+		}
 		rec := danmuRecord{
 			Time:    float64(rel) / 1e9,
 			UID:     c.UserID,
@@ -251,8 +286,12 @@ func (w *danmuWriter) write(c *acfundanmu.Comment) {
 		}
 		b, err := json.Marshal(rec)
 		checkErr(err)
-		w.jsonl.Write(b)
-		w.jsonl.Write([]byte("\n"))
+		if _, err := w.jsonl.Write(b); err != nil {
+			w.fail("写弹幕 jsonl", err)
+		}
+		if _, err := w.jsonl.Write([]byte("\n")); err != nil {
+			w.fail("写弹幕 jsonl", err)
+		}
 	}
 }
 
@@ -276,11 +315,20 @@ func (w *danmuWriter) consume(ctx context.Context, ac *acfundanmu.AcFunLive) {
 	}
 }
 
-// 关闭所有文件
-func (w *danmuWriter) close() {
-	w.ass.Close()
-	w.nick.Close()
-	w.jsonl.Close()
+// 关闭所有文件，并把整个录制过程中出过的写盘错误一并交出来。
+// Close 自己的错误也要记：os.File 在部分系统上（NFS、某些网络盘）是到 Close
+// 才报出「写不进去」的，只盯每次 Write 的返回值会漏掉这种情况
+func (w *danmuWriter) close() error {
+	if err := w.ass.Close(); err != nil {
+		w.fail("关闭原格式 ASS", err)
+	}
+	if err := w.nick.Close(); err != nil {
+		w.fail("关闭昵称头像版 ASS", err)
+	}
+	if err := w.jsonl.Close(); err != nil {
+		w.fail("关闭弹幕 jsonl", err)
+	}
+	return w.err
 }
 
 // ==================== 录制流程 ====================
@@ -305,7 +353,15 @@ func (s streamer) getDanmu(ctx context.Context, info liveInfo) {
 	_ = ac.StartDanmu(ctx, false)
 	streamInfo := ac.GetStreamInfo()
 	w := newDanmuWriter(streamInfo.LiveID, streamInfo.StreamName, info.cfg, info.assFile)
-	defer w.close()
+	// 写盘出错要到关文件时才收得全（有的系统到 Close 才报「写不进去」）。
+	// 必须说出来：否则本场照常自动合并，出来的成片缺一大截弹幕，日志里却一切正常
+	defer func() {
+		if err := w.close(); err != nil {
+			lPrintErrf("录制%s的直播弹幕时写盘出错，弹幕文件不完整：%v", s.longID(), err)
+			lPrintErrf("本场不做自动合并，弹幕文件保留在 %s，请检查磁盘空间后手动合并", info.assFile)
+			updateLiveInfo(info.LiveID, func(i *liveInfo) { i.danmuWriteFailed = true })
+		}
+	}()
 	w.consume(ctx, ac)
 
 	// 断流后先观察一个窗口：网络波动或主播短暂掉线（20 秒内）恢复时沿用同一个写入器接着写，

danmuburn.go

@@ -1100,14 +1100,24 @@ func ensureAvatar(uid int64, url, cacheDir string, legacyDirs ...string) string
 	return path
 }
 
+const (
+	// 头像下载的体积上限。真实头像最大也就几百 KB，这里给个大方的余量。
+	// 没有上限的话，CDN 一旦返回异常大的响应就能把内存吃光（实测单个头像吃掉 512MB）
+	avatarMaxBytes = 2 << 20
+	// 解码后允许的像素数上限（长 × 宽）。2048×2048 对头像来说是夸张地宽裕
+	avatarMaxPixels = 2048 * 2048
+)
+
+// 复用连接：一场直播要下几百个头像，每个都新建 client 就是几百次握手
+var avatarClient = &http.Client{Timeout: 30 * time.Second}
+
 func downloadAvatar(uid int64, url, path string) error {
-	client := &http.Client{Timeout: 30 * time.Second}
 	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
 		return err
 	}
 	req.Header.Set("User-Agent", "Mozilla/5.0")
-	resp, err := client.Do(req)
+	resp, err := avatarClient.Do(req)
 	if err != nil {
 		return err
 	}
@@ -1115,10 +1125,14 @@ func downloadAvatar(uid int64, url, path string) error {
 	if resp.StatusCode != http.StatusOK {
 		return fmt.Errorf("HTTP %d", resp.StatusCode)
 	}
-	data, err := io.ReadAll(resp.Body)
+	// 多读一个字节才能分辨「刚好到上限」和「超过了上限」
+	data, err := io.ReadAll(io.LimitReader(resp.Body, avatarMaxBytes+1))
 	if err != nil {
 		return err
 	}
+	if len(data) > avatarMaxBytes {
+		return fmt.Errorf("头像体积超过 %d MB，不下载", avatarMaxBytes>>20)
+	}
 	src, err := decodeAvatar(data, resp.Header.Get("Content-Type"))
 	if err != nil {
 		return err
@@ -1140,6 +1154,11 @@ func decodeAvatar(data []byte, contentType string) (image.Image, error) {
 	if err != nil {
 		return nil, fmt.Errorf("图片解析不了（Content-Type: %s）：%v", contentType, err)
 	}
+	// 防压缩炸弹：几十 KB 的 PNG/GIF 能解压出上亿像素，几 MB 的文件就能吃掉几 GB 内存。
+	// 头像最后只会被缩到几十像素（跟随字号），根本用不到这么大的原图
+	if b := src.Bounds(); b.Dx()*b.Dy() > avatarMaxPixels {
+		return nil, fmt.Errorf("头像尺寸 %dx%d 超出上限（最多 %d 像素）", b.Dx(), b.Dy(), avatarMaxPixels)
+	}
 	return src, nil
 }
 
@@ -1175,16 +1194,18 @@ func avatarURLByUID(uid int64) string {
 	return url
 }
 
+// 查头像链接用的 client，同样复用连接
+var avatarAPIClient = &http.Client{Timeout: 15 * time.Second}
+
 // 查 AcFun 的用户信息接口取头像地址
 func fetchAvatarURL(uid int64) string {
-	client := &http.Client{Timeout: 15 * time.Second}
 	addr := fmt.Sprintf("https://www.acfun.cn/rest/pc-direct/user/userInfo?userId=%d", uid)
 	req, err := http.NewRequest("GET", addr, nil)
 	if err != nil {
 		return ""
 	}
 	req.Header.Set("User-Agent", "Mozilla/5.0")
-	resp, err := client.Do(req)
+	resp, err := avatarAPIClient.Do(req)
 	if err != nil {
 		return ""
 	}

manualmerge.go

@@ -471,7 +471,9 @@ func runManualMerge(plan mergePlan) error {
 	for _, v := range plan.info.Videos {
 		inputBytes += v.Size
 	}
-	jobStart("手动合并", plan.info.OutName, inputBytes, plan.info.Duration)
+	if !jobStart("手动合并", plan.info.OutName, inputBytes, plan.info.Duration) {
+		return errors.New("已经有别的合并任务在跑，请先等它结束")
+	}
 
 	src, err := prepareDanmuSource(plan.danmu, true)
 	if err != nil {
@@ -596,12 +598,23 @@ func mergeHandler(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "已经有一个手动合并在进行（"+busyOut+"），等它结束再说", http.StatusConflict)
 		return
 	}
+	// 任务卡片只有一份：自动合并也可能占着它，那时手动的也得等
+	if !tryTakeMergeSlot() {
+		mergeRun.Unlock()
+		busy := jobBusyName()
+		if busy == "" {
+			busy = "另一个合并任务"
+		}
+		http.Error(w, "已经有一个合并在进行（"+busy+"），等它结束再说", http.StatusConflict)
+		return
+	}
 	mergeRun.busy = true
 	mergeRun.outName = plan.info.OutName
 	mergeRun.lastErr = ""
 	mergeRun.Unlock()
 
 	safeGo("手动合并", func() {
+		defer releaseMergeSlot()
 		defer func() {
 			mergeRun.Lock()
 			mergeRun.busy = false

manualmerge_test.go

@@ -63,6 +63,16 @@ Dialogue: 0,0:00:03.50,0:00:13.50,Danmu,丙(303),20,20,2,,{\move(400,24,-40,24)}
 
 // 测试用的 ffmpeg，找不到就跳过
 // 依次找 PATH 和源码目录（以及上一级目录）里的 ffmpeg，不写死某台机器上的路径
+// 按真实链路跑一次手动合并：独占任务位，跑完一定要收尾。
+// 不能直接调 runManualMerge —— 上一个用例如果没收尾，jobStart 会拒掉这一场
+// （正是它的职责），测试就会莫名其妙失败
+func execManualMerge(plan mergePlan) error {
+	jobFinish("")
+	err := runMergeExclusive(func() error { return runManualMerge(plan) })
+	jobFinish(jobErrText(err))
+	return err
+}
+
 func testFFmpeg(t *testing.T) string {
 	t.Helper()
 	cands := []string{"ffmpeg"}
@@ -641,7 +651,7 @@ Dialogue: 0,0:00:05.00,0:00:15.00,Danmu,甲(101),20,20,2,,{\move(400,72,-40,72)}
 	}
 
 	// 真正合并
-	if err := runManualMerge(plan); err != nil {
+	if err := execManualMerge(plan); err != nil {
 		t.Fatalf("手动合并失败：%v", err)
 	}
 	if _, err := os.Stat(plan.out); err != nil {
@@ -708,7 +718,7 @@ Dialogue: 0,0:00:05.00,0:00:15.00,Danmu,乙(202),0,0,0,,{\move(400,48,-40,48)}
 	if filepath.Base(plan.videos[0]) != filepath.Base(v1) {
 		t.Fatalf("分段顺序应纠正为第一段在前，实际 %q", filepath.Base(plan.videos[0]))
 	}
-	if err := runManualMerge(plan); err != nil {
+	if err := execManualMerge(plan); err != nil {
 		t.Fatalf("多段手动合并失败：%v", err)
 	}
 	d := probeDuration(ffmpeg, plan.out)

progress.go

@@ -1,10 +1,12 @@
 // 转换任务（合并 / 烧录弹幕）的实时进度与暂停、停止控制。
-// 进度只保留一份全局状态：同一时间只允许一个合并任务在跑（mergeRun 已经拦住了）。
+// 进度只保留一份全局状态：同一时间只允许一个合并任务在跑，串行靠下面的 mergeSlot 保证
+// （mergeRun.busy 只拦得住手动合并之间，拦不住自动合并，也拦不住手动与自动互相穿插）。
 package main
 
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"math"
 	"net/http"
 	"os/exec"
@@ -77,7 +79,7 @@ type jobInfo struct {
 
 // 全局唯一的任务状态
 type jobState struct {
-	sync.Mutex
+	sync.RWMutex
 	active   bool
 	kind     string
 	outName  string
@@ -115,10 +117,49 @@ type jobState struct {
 
 var job jobState
 
-// 开始一个新任务。kind 是任务类型，totalSec 是视频总时长（拿不到填 0）
-func jobStart(kind, outName string, inputBytes int64, totalSec float64) {
+// 合并任务的串行锁。全局 job 状态只有一份，两个任务同时跑的话后启动的会把前一个的
+// 进度和输出名清掉，「停止」也会杀掉别家任务的 ffmpeg，所以必须排队。
+// 手动合并用 TryLock（抢不到立刻告诉用户），自动合并用 Lock（后台任务，排队等就行）。
+var mergeSlot sync.Mutex
+
+// 尝试占用合并任务位，被占着返回 false。给手动合并用：不能让 HTTP 请求干等着
+func tryTakeMergeSlot() bool {
+	return mergeSlot.TryLock()
+}
+
+// 占用合并任务位，被占着就等到前一个跑完。给自动合并用
+func takeMergeSlot() {
+	mergeSlot.Lock()
+}
+
+func releaseMergeSlot() {
+	mergeSlot.Unlock()
+}
+
+// 排队跑完一整个合并任务：先独占任务位，跑完后一定释放，被调用方 panic 也不漏。
+// 内部出错会被翻成 error 而不是让上层去 recover —— 任务卡片只有走到 jobFinish 才会收尾，
+// 漏掉的话进度会永远停在界面上（之前 panic 后卡片一直转就是这么来的）
+func runMergeExclusive(f func() error) (err error) {
+	takeMergeSlot()
+	defer releaseMergeSlot()
+	defer func() {
+		if p := recover(); p != nil {
+			err = fmt.Errorf("合并任务内部出错：%v", p)
+			lPrintErrf("合并任务内部出错，已兜住：%v", p)
+		}
+	}()
+	return f()
+}
+
+// 开始一个新任务。kind 是任务类型，totalSec 是视频总时长（拿不到填 0）。
+// 已经有任务在跑时返回 false 并且什么都不动：调用方应该先拿到 mergeSlot，
+// 这里再做一次判断是为了让误用（忘记排队）只会失败而不会互相覆盖
+func jobStart(kind, outName string, inputBytes int64, totalSec float64) bool {
 	job.Lock()
 	defer job.Unlock()
+	if job.active {
+		return false
+	}
 	job.active = true
 	job.kind = kind
 	job.outName = outName
@@ -143,6 +184,24 @@ func jobStart(kind, outName string, inputBytes int64, totalSec float64) {
 	job.errText = ""
 	job.message = ""
 	job.doneAt = 0
+	return true
+}
+
+// 当前是否有任务在跑。给界面提示和「现在能不能接新任务」的判断用
+func jobActive() bool {
+	job.RLock()
+	defer job.RUnlock()
+	return job.active
+}
+
+// 当前任务的名字（没任务时为空），拒绝新任务时用它告诉用户在忙什么
+func jobBusyName() string {
+	job.RLock()
+	defer job.RUnlock()
+	if !job.active {
+		return ""
+	}
+	return job.outName
 }
 
 // 任务结束。errText 为空表示成功，errJobStopped.Error() 表示被手动停止
@@ -168,13 +227,6 @@ func jobFinish(errText string) {
 	}
 }
 
-// 任务是否正在跑
-func jobActive() bool {
-	job.Lock()
-	defer job.Unlock()
-	return job.active
-}
-
 // 是否已经被要求停止
 func jobStopRequested() bool {
 	job.Lock()

progress_test.go

@@ -3,10 +3,22 @@ package main
 import (
 	"errors"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
 
+// 测试用的「开一个新任务」。全局任务只有一份，用例之间必须清干净：
+// 上一个没收尾的话 jobStart 会拒绝覆盖（这是刻意的防串台），断言就会读到别人的数据
+func startJobForTest(t *testing.T, kind, outName string, inputBytes int64, totalSec float64) {
+	t.Helper()
+	jobFinish("")
+	if !jobStart(kind, outName, inputBytes, totalSec) {
+		t.Fatalf("jobStart 失败：还残留着未结束的任务 %q", jobBusyName())
+	}
+	t.Cleanup(func() { jobFinish("") })
+}
+
 // 阶段权重加起来必须正好是 1，否则总进度永远到不了 100%
 func TestJobPhaseWeightsSumToOne(t *testing.T) {
 	if len(jobPhaseNames) != len(jobPhaseWeights) {
@@ -23,8 +35,7 @@ func TestJobPhaseWeightsSumToOne(t *testing.T) {
 
 // 总进度：走过的阶段按权重满算，当前阶段按比例折算
 func TestJobPercentByPhase(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 100)
-	t.Cleanup(func() { jobFinish("") })
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 
 	if got := jobSnapshot().Percent; got != 0 {
 		t.Errorf("刚开始时总进度应为 0，实际 %v", got)
@@ -50,7 +61,7 @@ func TestJobPercentByPhase(t *testing.T) {
 
 // 阶段序号只往前走，重复设置同一阶段不会把进度打回 0
 func TestJobPhaseOnlyMovesForward(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 100)
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 	defer jobFinish("")
 	jobSetPhase(jobPhaseRender)
 	jobRender(60, 1, 25, 0)
@@ -66,7 +77,7 @@ func TestJobPhaseOnlyMovesForward(t *testing.T) {
 
 // 渲染阶段的 ETA = 剩余时长 ÷ 倍速
 func TestJobEtaFromSpeed(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 200)
+	startJobForTest(t, "测试", "x.mp4", 0, 200)
 	defer jobFinish("")
 	jobSetPhase(jobPhaseRender)
 	jobRender(50, 2.0, 25, 0)
@@ -78,7 +89,7 @@ func TestJobEtaFromSpeed(t *testing.T) {
 
 // 预计输出大小：先给按码率估的值，渲染过 5% 之后换成按已写出字节外推
 func TestJobEstOutSize(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 100)
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 	defer jobFinish("")
 	jobSetPhase(jobPhaseRender)
 
@@ -99,7 +110,7 @@ func TestJobEstOutSize(t *testing.T) {
 	}
 
 	// 比例太小的时候不急着外推，免得开头抖动
-	jobStart("测试", "x.mp4", 0, 100)
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 	jobSetPhase(jobPhaseRender)
 	jobPlanSize(5000)
 	jobRender(1, 1, 25, 100)
@@ -110,7 +121,7 @@ func TestJobEstOutSize(t *testing.T) {
 
 // 速度折线：最多保留固定点数，且按秒采样
 func TestJobSamples(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 100)
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 	defer jobFinish("")
 	jobSetPhase(jobPhaseRender)
 	// 同一秒内连着报多次只留一个点
@@ -136,7 +147,7 @@ func TestJobSamples(t *testing.T) {
 
 // 暂停 / 恢复 / 停止的状态流转；停止后闸门必须让上层拿到 errJobStopped
 func TestJobPauseStopGate(t *testing.T) {
-	jobStart("测试", "x.mp4", 0, 100)
+	startJobForTest(t, "测试", "x.mp4", 0, 100)
 	defer jobFinish("")
 
 	if err := jobPause(); err != nil {
@@ -281,7 +292,7 @@ func TestScanFFmpegProgressTolerant(t *testing.T) {
 
 // 截图式地看一眼任务快照的字段都填上了
 func TestJobSnapshotFields(t *testing.T) {
-	jobStart("手动合并", "成片.mp4", 123456, 3600)
+	startJobForTest(t, "手动合并", "成片.mp4", 123456, 3600)
 	defer jobFinish("")
 	jobSetPhase(jobPhaseAvatar)
 	jobAvatars(3, 10)
@@ -306,3 +317,99 @@ func TestJobSnapshotFields(t *testing.T) {
 		t.Error("任务在跑的时候应该可暂停")
 	}
 }
+
+// ==================== 任务串行化 ====================
+// 全局任务状态只有一份：两个合并同时跑的话，后启动的会把前一个的进度和输出名清掉，
+// 点「停止」还会杀掉别家任务的 ffmpeg。下面几条例外情况锁住改串行之后的语义。
+
+// jobStart 在已经有任务在跑时必须拒绝，并且一个字段都不能动
+func TestJobStartRejectsWhenBusy(t *testing.T) {
+	startJobForTest(t, "自动合并", "甲_弹幕合并.mp4", 1000, 100)
+	jobSetPhase(jobPhaseRender)
+	jobRender(30, 1.2, 25, 5000)
+
+	if ok := jobStart("手动合并", "乙_弹幕合并.mp4", 2000, 200); ok {
+		t.Fatal("已经有任务在跑，jobStart 应当返回 false")
+	}
+	info := jobSnapshot()
+	if info.Kind != "自动合并" || info.OutName != "甲_弹幕合并.mp4" {
+		t.Errorf("被拒之后不该动别人的任务信息：%+v", info)
+	}
+	if info.InputBytes != 1000 || info.TotalSec != 100 {
+		t.Errorf("被拒之后输入大小和总时长也不该变：%+v", info)
+	}
+	if info.Percent < 25 || info.Percent > 45 {
+		t.Errorf("被拒之后进度必须还是原任务自己的（25~45），实际 %v", info.Percent)
+	}
+	if !info.Active {
+		t.Error("拒绝新任务不代表把老任务结束掉")
+	}
+}
+
+// 任务位被占着时手动合并要立刻知道（不能让页面干等），抢不到只能有一个赢
+func TestMergeSlotTryLockExclusive(t *testing.T) {
+	if !tryTakeMergeSlot() {
+		t.Fatal("空闲时手动合并应该抢得到任务位")
+	}
+	if tryTakeMergeSlot() {
+		t.Fatal("任务位已经被占着，第二次 TryLock 不该成功")
+	}
+	releaseMergeSlot()
+	if !tryTakeMergeSlot() {
+		t.Fatal("释放之后应该又能抢到")
+	}
+	releaseMergeSlot()
+}
+
+// 自动合并（十几个主播同时下播很常见）必须排队而不是并发
+func TestMergeSlotSerializes(t *testing.T) {
+	if !tryTakeMergeSlot() {
+		t.Fatal("前置条件：任务位本该是空的")
+	}
+	const n = 4
+	order := make([]int, 0, n)
+	var mu sync.Mutex
+	done := make(chan struct{}, n)
+	for i := 0; i < n; i++ {
+		idx := i
+		go func() {
+			defer func() { done <- struct{}{} }()
+			// 和真实链路一致：自动合并会一直等到轮到自己
+			_ = runMergeExclusive(func() error {
+				mu.Lock()
+				order = append(order, idx)
+				mu.Unlock()
+				time.Sleep(20 * time.Millisecond)
+				return nil
+			})
+		}()
+	}
+	time.Sleep(60 * time.Millisecond) // 让它们先堆在门外
+	releaseMergeSlot()
+	for i := 0; i < n; i++ {
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("第 %d 个任务没跑完，队列卡住了", i)
+		}
+	}
+	if len(order) != n {
+		t.Fatalf("应当 %d 个任务都跑完，实际 %d 个", n, len(order))
+	}
+}
+
+// 任务内部 panic 时也必须释放任务位、并且能翻成错误
+// —— 否则卡片会永远停在「进行中」，队列也被锁死
+func TestMergeExclusiveReleasesOnPanic(t *testing.T) {
+	err := runMergeExclusive(func() error { panic("模拟内部出错") })
+	if err == nil {
+		t.Fatal("内部 panic 应当被翻成错误返回")
+	}
+	if !strings.Contains(err.Error(), "模拟内部出错") {
+		t.Errorf("错误信息里应带上原因，实际：%v", err)
+	}
+	if !tryTakeMergeSlot() {
+		t.Fatal("panic 之后任务位必须已经释放")
+	}
+	releaseMergeSlot()
+}

recordjsonl_test.go

@@ -2,10 +2,14 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
+	"errors"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/orzogc/acfundanmu"
 )
@@ -91,3 +95,149 @@ func countEmpty(recs []danmuRecord) int {
 	}
 	return n
 }
+
+// ==================== 写盘出错必须能被发现 ====================
+// 弹幕是边录边写的，写盘失败（磁盘满、外接盘/网络盘掉线）如果没人管，
+// 程序会照常宣告「录制完成」并自动合并 —— 出来的成片缺一大截弹幕，
+// 而日志里什么都看不出来。下面几条例外情况锁住「失败要浮出来」。
+
+// 写入失败必须被记下来，并且在 close() 时交出来
+func TestDanmuWriteErrorSurfacedOnClose(t *testing.T) {
+	dir := t.TempDir()
+	cfg := acfundanmu.SubConfig{Title: "测试直播", PlayResX: 1920, PlayResY: 1080,
+		FontSize: 48, StartTime: 1700000000000000000}
+	ass := filepath.Join(dir, "live.ass")
+	w := newDanmuWriter("liveid", "stream", cfg, ass)
+	if w.failed() {
+		t.Fatalf("刚建好的写入器不该有错误，实际：%v", w.err)
+	}
+
+	// 用「写到已关闭的句柄」来走同一条错误路径（和磁盘满、盘掉线一样，
+	// 都是 os.File.Write 返回 error）。换掉之前先把原来的句柄关掉，别泄漏
+	if err := w.jsonl.Close(); err != nil {
+		t.Fatal(err)
+	}
+	dead, err := os.Open(filepath.Join(dir, "live.jsonl"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := dead.Close(); err != nil {
+		t.Fatal(err)
+	}
+	w.jsonl = dead
+
+	c := &acfundanmu.Comment{Content: "这条应该写失败"}
+	c.SendTime = (cfg.StartTime + int64(500)*1e6) / 1e6
+	c.UserID = 7
+	c.Nickname = "观众"
+	w.write(c)
+
+	if !w.failed() {
+		t.Fatal("写入失败之后 failed() 应为 true（改之前错误是被丢掉的）")
+	}
+	err = w.close()
+	if err == nil {
+		t.Fatal("close() 必须把写盘错误交出来，实际返回 nil")
+	}
+	if !strings.Contains(err.Error(), "jsonl") {
+		t.Errorf("错误信息里要能看出是哪个文件出的问题，实际：%v", err)
+	}
+}
+
+// 只记第一个错：磁盘满了之后每写一条都会失败，全记下来会把日志刷爆
+func TestDanmuWriteKeepsOnlyFirstError(t *testing.T) {
+	w := &danmuWriter{}
+	w.fail("第一次", errors.New("磁盘满了"))
+	first := w.err
+	w.fail("第二次", errors.New("还是写不进去"))
+	w.fail("第三次", errors.New("依旧写不进去"))
+	if w.err != first {
+		t.Fatalf("错误应保持在第一个，实际被后来的覆盖了：%v", w.err)
+	}
+	if !strings.Contains(w.err.Error(), "磁盘满了") {
+		t.Errorf("应保留第一个错误的原因，实际：%v", w.err)
+	}
+}
+
+// 正常录制不能被误判成失败（这是「加检查」最容易踩的反向坑）
+func TestDanmuWriteNoFalseAlarm(t *testing.T) {
+	dir := t.TempDir()
+	cfg := acfundanmu.SubConfig{Title: "测试直播", PlayResX: 1920, PlayResY: 1080,
+		FontSize: 48, StartTime: 1700000000000000000}
+	w := newDanmuWriter("liveid", "stream", cfg, filepath.Join(dir, "ok.ass"))
+	for i := 0; i < 30; i++ {
+		c := &acfundanmu.Comment{Content: "正常弹幕"}
+		c.SendTime = (cfg.StartTime + int64(i)*220*1e6) / 1e6
+		c.UserID = int64(100 + i)
+		c.Nickname = "观众"
+		w.write(c)
+	}
+	if w.failed() {
+		t.Fatalf("正常写入不该判失败：%v", w.err)
+	}
+	if err := w.close(); err != nil {
+		t.Fatalf("正常写完后 close() 应返回 nil，实际：%v", err)
+	}
+}
+
+// 弹幕文件写坏之后必须跳过自动合并：只是打一条日志的话，
+// 照样会合成一个「缺了一大截弹幕」的成片，而且是静默的
+func TestFinishPartSkipsMergeWhenDanmuBroken(t *testing.T) {
+	oldAuto, oldCtx := config.AutoMerge, mainCtx
+	t.Cleanup(func() {
+		config.AutoMerge, mainCtx = oldAuto, oldCtx
+		// 等后台那两个（可能启动了的）合并跑完，别把任务状态留给后面的用例
+		takeMergeSlot()
+		releaseMergeSlot()
+	})
+	config.AutoMerge = true
+	mainCtx = context.Background()
+
+	dir := t.TempDir()
+	video := filepath.Join(dir, "v.mp4")
+	ass := filepath.Join(dir, "v.ass")
+	if err := os.WriteFile(video, []byte("fake"), 0644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(ass, []byte("[Script Info]"), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	// 测试里没跑 initialize()，这张表还没建起来
+	lInfoMap.Lock()
+	if lInfoMap.info == nil {
+		lInfoMap.info = make(map[string]liveInfo)
+	}
+	lInfoMap.Unlock()
+
+	// 对照组：一切都正常时应该触发合并
+	good := "good-live-id"
+	lInfoMap.Lock()
+	lInfoMap.info[good] = liveInfo{LiveID: good, uid: 1, assFile: ass, firstRecordFile: video,
+		recordStartNs: time.Now().UnixNano(), recordSegments: []recordSegment{{file: video, startNs: time.Now().UnixNano()}}}
+	lInfoMap.Unlock()
+	finishPart(good, true)
+	finishPart(good, false)
+	lInfoMap.RLock()
+	started := lInfoMap.info[good].mergeStarted
+	lInfoMap.RUnlock()
+	if !started {
+		t.Fatal("对照组：条件都满足时应当触发自动合并（否则这条测试本身没意义）")
+	}
+
+	// 实验组：同样条件，但弹幕写盘出过错
+	bad := "bad-live-id"
+	lInfoMap.Lock()
+	lInfoMap.info[bad] = liveInfo{LiveID: bad, uid: 2, assFile: ass, firstRecordFile: video,
+		danmuWriteFailed: true,
+		recordStartNs:    time.Now().UnixNano(), recordSegments: []recordSegment{{file: video, startNs: time.Now().UnixNano()}}}
+	lInfoMap.Unlock()
+	finishPart(bad, true)
+	finishPart(bad, false)
+	lInfoMap.RLock()
+	got := lInfoMap.info[bad].mergeStarted
+	lInfoMap.RUnlock()
+	if got {
+		t.Error("弹幕文件写坏过，不该触发自动合并（否则会合成一个缺弹幕的成片）")
+	}
+}

utils.go

@@ -67,8 +67,11 @@ type liveInfo struct {
 	danmuStopped    bool               // 弹幕是否被用户手动停止（手动停止不触发自动合并）
 	recordStopped   bool               // 视频是否被用户手动停止
 	mergeStarted    bool               // 自动合并是否已经启动
-	doneAt          int64              // 最近一次录制结束的时刻，纳秒 Unix（用于延迟清理这条记录）
-	resumeUntil     int64              // 断流后等待直播恢复的截止时刻，纳秒 Unix；0 表示没在等待
+	// 弹幕文件在写盘时出过错（磁盘满、网络盘断开等）。文件是不完整的，
+	// 拿它合并会得到一个缺弹幕的成片，所以本场跳过自动合并
+	danmuWriteFailed bool
+	doneAt           int64 // 最近一次录制结束的时刻，纳秒 Unix（用于延迟清理这条记录）
+	resumeUntil      int64 // 断流后等待直播恢复的截止时刻，纳秒 Unix；0 表示没在等待
 }
 
 // streamerInfo 的 map

webui.go

@@ -2460,7 +2460,8 @@ func checkAuthToken(r *http.Request) bool {
 // 口令错误次数的记录，给反复尝试的请求加一点延迟
 var authFail struct {
 	sync.Mutex
-	m map[string]authFailEntry
+	m          map[string]authFailEntry
+	sinceSweep int // 距上次扫表又记了多少条
 }
 
 type authFailEntry struct {
@@ -2471,6 +2472,62 @@ type authFailEntry struct {
 // 口令错误时最多延迟这么久，避免拖垮服务器
 const authMaxDelay = 2 * time.Second
 
+const (
+	// 表里最多留这么多条。IP 是外部给的，必须有个硬上限：只靠过期清理的话，
+	// 每换一个新来源就多一条，清理追不上写入，最后把内存写满
+	authFailMaxEntries = 10000
+	// 超过这么久没再失败，这条记录就当没人用了
+	authFailExpire = 10 * time.Minute
+	// 同一个来源最多记到这个次数：再往上涨延迟也是封顶的，没必要让数字一直加
+	authFailMaxCount = 1000
+	// 每记这么多条扫一次表。扫一次要遍历（必要时还要排序），不必每次都做
+	authFailSweepEvery = 256
+)
+
+// 扫一遍口令错误记录：先删过期项，还多就把「最久没动静」的淘汰掉。
+// 淘汰时留出一成余量，这样满了之后还能再收一千来个新来源才需要重排下一次全表，
+// 不至于每来一个就把整张表排一遍（那反倒成了另一条拖慢服务的路子）。
+func sweepAuthFail(now time.Time) {
+	for k, v := range authFail.m {
+		if now.Sub(v.last) > authFailExpire {
+			delete(authFail.m, k)
+		}
+	}
+	target := authFailMaxEntries * 9 / 10
+	over := len(authFail.m) - target
+	if over <= 0 {
+		return
+	}
+	type record struct {
+		ip   string
+		last time.Time
+	}
+	all := make([]record, 0, len(authFail.m))
+	for k, v := range authFail.m {
+		all = append(all, record{k, v.last})
+	}
+	sort.Slice(all, func(i, j int) bool { return all[i].last.Before(all[j].last) })
+	for i := 0; i < over && i < len(all); i++ {
+		delete(authFail.m, all[i].ip)
+	}
+}
+
+// 等待口令错误的惩罚延迟。客户端一断开就立刻返回，
+// 否则并发撞口令时一堆 goroutine 各占着一条连接干等两秒，很容易把服务拖住
+func awaitAuthDelay(ctx context.Context, d time.Duration) bool {
+	if d <= 0 {
+		return true
+	}
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
 // 取请求的来源 IP
 func clientIP(r *http.Request) string {
 	if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
@@ -2486,17 +2543,23 @@ func authFailDelay(ip string) time.Duration {
 	if authFail.m == nil {
 		authFail.m = make(map[string]authFailEntry)
 	}
-	// 记录太多时清掉过期的，避免大量不同来源的请求把内存撑满
-	if len(authFail.m) > 1000 {
-		for k, v := range authFail.m {
-			if time.Since(v.last) > 10*time.Minute {
-				delete(authFail.m, k)
-			}
-		}
+	now := time.Now()
+	// 到上限就立刻扫一次。只靠定期扫的话，两个周期之间涌入的新来源
+	// 会把表顶破上限（IP 是外部给的，写入速度不受我们控制）
+	if len(authFail.m) >= authFailMaxEntries {
+		authFail.sinceSweep = 0
+		sweepAuthFail(now)
+	}
+	authFail.sinceSweep++
+	if authFail.sinceSweep >= authFailSweepEvery {
+		authFail.sinceSweep = 0
+		sweepAuthFail(now)
 	}
 	e := authFail.m[ip]
-	e.count++
-	e.last = time.Now()
+	if e.count < authFailMaxCount {
+		e.count++
+	}
+	e.last = now
 	authFail.m[ip] = e
 	d := time.Duration(e.count) * 100 * time.Millisecond
 	if d > authMaxDelay {
@@ -2527,7 +2590,10 @@ func webAuth(next http.Handler) http.Handler {
 			return
 		}
 		if d := authFailDelay(ip); d > 0 {
-			time.Sleep(d)
+			// 客户端已经断开就不用再等了，直接收工
+			if !awaitAuthDelay(r.Context(), d) {
+				return
+			}
 		}
 		w.Header().Set("Content-Type", "application/json; charset=utf-8")
 		w.WriteHeader(http.StatusUnauthorized)

webui_test.go

@@ -2,10 +2,13 @@
 package main
 
 import (
+	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"testing"
+	"time"
 )
 
 // 测试期间换一个管理口令
@@ -155,3 +158,148 @@ func TestAuthTokenCandidatesOrder(t *testing.T) {
 		t.Fatalf("非 Bearer 的 Authorization 不该当候选：%q", got)
 	}
 }
+
+// ==================== 口令错误记录的容量 ====================
+// IP 是外部给的：只靠「过期清理」挡不住换着地址撞口令的情况，
+// 必须要有硬上限。这些用例锁住这一点。
+
+// 换着来源持续撞口令，表必须被压在上限以内，不能一直涨
+func TestAuthFailTableHasHardCap(t *testing.T) {
+	saved, savedSweep := authFail.m, authFail.sinceSweep
+	t.Cleanup(func() {
+		authFail.Lock()
+		authFail.m, authFail.sinceSweep = saved, savedSweep
+		authFail.Unlock()
+	})
+	authFail.Lock()
+	authFail.m, authFail.sinceSweep = nil, 0
+	authFail.Unlock()
+
+	// 每个都是新来源、且都是「刚刚」失败过 —— 过期清理一条都删不掉的情况
+	for i := 0; i < authFailMaxEntries*2; i++ {
+		authFailDelay(fmt.Sprintf("198.51.100.%d", i))
+	}
+	got := 0
+	authFail.Lock()
+	got = len(authFail.m)
+	authFail.Unlock()
+	if got > authFailMaxEntries {
+		t.Errorf("记录数应被压在上限 %d 以内，实际 %d（换 IP 撞口令会把内存写满）",
+			authFailMaxEntries, got)
+	}
+	t.Logf("  %d 个新来源撞完，表内留下 %d 条（上限 %d）", authFailMaxEntries*2, got, authFailMaxEntries)
+}
+
+// 超过过期时间的记录要被扫掉，不是在表里一直躺着
+func TestAuthFailExpiredEntriesGetSwept(t *testing.T) {
+	saved, savedSweep := authFail.m, authFail.sinceSweep
+	t.Cleanup(func() {
+		authFail.Lock()
+		authFail.m, authFail.sinceSweep = saved, savedSweep
+		authFail.Unlock()
+	})
+	authFail.Lock()
+	authFail.m, authFail.sinceSweep = nil, 0
+	authFail.Unlock()
+
+	for i := 0; i < authFailSweepEvery+5; i++ {
+		authFailDelay(fmt.Sprintf("203.0.113.%d", i))
+	}
+	authFail.Lock()
+	before := len(authFail.m)
+	// 把记录的最后时间推到过期之前
+	for k, v := range authFail.m {
+		authFail.m[k] = authFailEntry{count: v.count, last: time.Now().Add(-2 * authFailExpire)}
+	}
+	authFail.Unlock()
+
+	sweepAuthFail(time.Now())
+	authFail.Lock()
+	after := len(authFail.m)
+	authFail.Unlock()
+	if after != 0 {
+		t.Errorf("已经全部过期的记录应被扫干净，扫之前 %d 条，扫之后 %d 条", before, after)
+	}
+}
+
+// 扫表必须是自动发生的：不要求调用方记着手动清理，
+// 混在里面的过期项应该在若干次记录之后自己消失
+func TestAuthFailSweepsPeriodically(t *testing.T) {
+	saved, savedSweep := authFail.m, authFail.sinceSweep
+	t.Cleanup(func() {
+		authFail.Lock()
+		authFail.m, authFail.sinceSweep = saved, savedSweep
+		authFail.Unlock()
+	})
+	authFail.Lock()
+	authFail.m, authFail.sinceSweep = nil, 0
+	authFail.Unlock()
+
+	// 直接往里塞一条早就过期了的记录（绕过 authFailDelay，模拟「以前留下的」）
+	authFail.Lock()
+	if authFail.m == nil {
+		authFail.m = make(map[string]authFailEntry)
+	}
+	authFail.m["203.0.113.99"] = authFailEntry{count: 1, last: time.Now().Add(-2 * authFailExpire)}
+	authFail.Unlock()
+
+	// 之后只要还在有人失败，过一阵子它就应该被清掉
+	for i := 0; i < authFailSweepEvery+1; i++ {
+		authFailDelay(fmt.Sprintf("198.51.100.%d", i))
+	}
+
+	authFail.Lock()
+	_, still := authFail.m["203.0.113.99"]
+	authFail.Unlock()
+	if still {
+		t.Errorf("记了 %d 次之后，过期记录应当已经被自动扫掉", authFailSweepEvery+1)
+	}
+}
+
+// 同一个来源的计数不能无限涨（临界点是延迟早已封顶，再涨只是让数字溢出）
+func TestAuthFailCountCapped(t *testing.T) {
+	saved, savedSweep := authFail.m, authFail.sinceSweep
+	t.Cleanup(func() {
+		authFail.Lock()
+		authFail.m, authFail.sinceSweep = saved, savedSweep
+		authFail.Unlock()
+	})
+	authFail.Lock()
+	authFail.m, authFail.sinceSweep = nil, 0
+	authFail.Unlock()
+
+	for i := 0; i < authFailMaxCount+50; i++ {
+		authFailDelay("192.0.2.9")
+	}
+	authFail.Lock()
+	c := authFail.m["192.0.2.9"].count
+	authFail.Unlock()
+	if c > authFailMaxCount {
+		t.Errorf("同一来源的次数应封顶在 %d，实际 %d", authFailMaxCount, c)
+	}
+	if d := authFailDelay("192.0.2.9"); d != authMaxDelay {
+		t.Errorf("打满之后延迟应封顶在 %v，实际 %v", authMaxDelay, d)
+	}
+}
+
+// 客户端断开时延迟等待必须立刻返回，不能傻等满 2 秒
+func TestAwaitAuthDelayStopsOnCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		time.Sleep(80 * time.Millisecond)
+		cancel()
+	}()
+	start := time.Now()
+	waited := awaitAuthDelay(ctx, 2*time.Second)
+	used := time.Since(start)
+	if waited {
+		t.Error("上游取消时 awaitAuthDelay 应返回 false")
+	}
+	if used > time.Second {
+		t.Errorf("上游取消后应立刻返回，实际等了 %v", used)
+	}
+	// 正常等待不受影响
+	if !awaitAuthDelay(context.Background(), 10*time.Millisecond) {
+		t.Error("没被取消时应当正常等完再返回 true")
+	}
+}