diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 20b4fa2..d6dc115 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -45,7 +45,7 @@ func setDefaults(v *viper.Viper) { v.SetDefault("logging.max_backups", 10) v.SetDefault("cache.cache_dir", "./cache") - v.SetDefault("cache.images_max_size_mb", 2000) + v.SetDefault("cache.images_max_size_mb", 2048) v.SetDefault("cache.images_originals_max_size_mb", 256) v.SetDefault("cache.memory_max_size_mb", DefaultCacheMemoryMaxSizeMB) v.SetDefault("cache.redis_url", "") diff --git a/internal/config/types.go b/internal/config/types.go index fd0881d..252703c 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -120,8 +120,12 @@ type LoggingConfig struct { // CacheConfig 控制磁盘转写缓存和进程内热缓存。 type CacheConfig struct { - CacheDir string `mapstructure:"cache_dir"` - ImagesMaxSizeMB int `mapstructure:"images_max_size_mb"` + CacheDir string `mapstructure:"cache_dir"` + // ImagesMaxSizeMB 是整个磁盘缓存的总容量上限（MB），不是 images 子目录 + // 的配额。参与统计的池为 hls、images、library-covers；超限时按该顺序淘汰 + // 最旧文件。reader-audio 由自身 1GB 配额在写入时独立淘汰，不参与总容量。 + // 0 表示不限制。 + ImagesMaxSizeMB int `mapstructure:"images_max_size_mb"` // ImagesOriginalsMaxSizeMB 限制 cache/images 下原图子配额（MB）。原图只是 // 生成缩略图的原料，丢了可以重新回源；0 表示不单独限制（只受总量约束）。 ImagesOriginalsMaxSizeMB int `mapstructure:"images_originals_max_size_mb"` diff --git a/internal/handler/admin_settings.go b/internal/handler/admin_settings.go index 6665665..f00f717 100644 --- a/internal/handler/admin_settings.go +++ b/internal/handler/admin_settings.go @@ -89,7 +89,7 @@ func updateSettingHandler(svc *service.Container) gin.HandlerFunc { } if req.Key == "cache.images_max_size_mb" || req.Key == "cache.images_originals_max_size_mb" { if svc.Scheduler != nil { - _ = svc.Scheduler.RunNowAsync(c.Request.Context(), "image_cache_cleanup") + _ = svc.Scheduler.RunNowAsync(c.Request.Context(), "cache_cleanup") } } if req.Key == "cache.memory_max_size_mb" && svc.Cache != nil { diff --git a/internal/handler/reader_media_test.go b/internal/handler/reader_media_test.go index 5c12d24..29b0ad1 100644 --- a/internal/handler/reader_media_test.go +++ b/internal/handler/reader_media_test.go @@ -1,6 +1,7 @@ package handler import ( + "bytes" "fmt" "net/http" "net/http/httptest" @@ -97,6 +98,99 @@ func TestReaderMediaProxyFailsClearlyWhenRefreshImpossible(t *testing.T) { } } +// TestReaderMediaProxyStopsResendingRejectedTicket 上游成批拒绝时（连新签的票 +// 也是空 401），不能在播放器那一次请求里把同一张被拒的票打上十几遍：连续两轮 +// 换票都是同一个状态码就立刻失败，把明确文案交出去。 +func TestReaderMediaProxyStopsResendingRejectedTicket(t *testing.T) { + svc, srv, book := prepareMediaTicketBook(t) + router := mediaRouterForTest(svc) + + signed := primeTicketBook(t, svc, book.ID) + srv.denyAlways() + + code, body := getReaderMedia(t, router, signed) + if code != http.StatusBadGateway { + t.Fatalf("上游持续拒绝应返回 502，实际 %d body=%s", code, body) + } + if !strings.Contains(body, "401") { + t.Fatalf("错误文案应带上上游状态码: %s", body) + } + // 每轮 fetchReaderMedia = 首拉(401) + 换票后重拉(401)，两轮即收手。 + if got := srv.mediaRequests(); got > 4 { + t.Fatalf("无进展的拒绝应当两轮就收手，实际上游请求 %d 次", got) + } +} + +// TestReaderAudioCacheResumesTruncatedUpstream 上游中途 RST 后用 Range 续传： +// 播放器最终拿到的仍是完整的一章，而不是从头重下。 +func TestReaderAudioCacheResumesTruncatedUpstream(t *testing.T) { + const bodyBytes = 300 << 10 + payload := bytes.Repeat([]byte("m4a!"), bodyBytes/4) + var ranges []string + var mu sync.Mutex + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + ranges = append(ranges, r.Header.Get("Range")) + count := len(ranges) + mu.Unlock() + + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Accept-Ranges", "bytes") + start := 0 + if count > 1 { + // 第二次起按 Range 续传。 + start = bodyBytes / 2 + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, bodyBytes-1, bodyBytes)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.Header().Set("Content-Length", strconv.Itoa(bodyBytes)) + } + chunk := payload[start:] + if count == 1 { + // 第一次只写一半就断开。 + chunk = payload[:bodyBytes/2] + } + for written := 0; written < len(chunk); written += 8 << 10 { + end := written + 8<<10 + if end > len(chunk) { + end = len(chunk) + } + if _, err := w.Write(chunk[written:end]); err != nil { + return + } + } + w.(http.Flusher).Flush() + if count == 1 { + if hj, ok := w.(http.Hijacker); ok { + if conn, _, err := hj.Hijack(); err == nil { + _ = conn.Close() + } + } + } + })) + defer srv.Close() + + svc, _, book := prepareMediaTicketBookWithUpstream(t, srv.URL) + router := mediaRouterForTest(svc) + signed := svc.Reader.ProxyURL(book.ID, srv.URL+"/media") + + code, body := getReaderMedia(t, router, signed) + if code != http.StatusOK { + t.Fatalf("续传后应当成功，实际 status=%d body=%.120s", code, body) + } + if len(body) != bodyBytes { + t.Fatalf("交付字节 = %d，期望 %d", len(body), bodyBytes) + } + mu.Lock() + defer mu.Unlock() + if len(ranges) < 2 || ranges[0] != "" { + t.Fatalf("请求序列 = %q，期望先整段再续传", ranges) + } + if !strings.HasPrefix(ranges[1], "bytes=") { + t.Fatalf("第二轮应当带 Range，实际 %q", ranges[1]) + } +} + // TestReaderMediaProxyPassesThroughNonTrackError 漫画图片这类不可换票的地址， // 上游 4xx 仍按原状态透传（既有行为不变）。 func TestReaderMediaProxyPassesThroughNonTrackError(t *testing.T) { @@ -347,12 +441,13 @@ func primeTicketBook(t *testing.T, svc *service.Container, bookID string) string // - /img/* 图片资源，固定 403（模拟防盗链） type mediaTicketServer struct { *httptest.Server - mu sync.Mutex - issued []string - hits int - mediaHits int - noToken bool - errorPage bool + mu sync.Mutex + issued []string + hits int + mediaHits int + noToken bool + errorPage bool + alwaysDeny bool } func newMediaTicketServer(t *testing.T) *mediaTicketServer { @@ -379,7 +474,13 @@ func newMediaTicketServer(t *testing.T) *mediaTicketServer { case r.URL.Path == "/media": s.mu.Lock() s.mediaHits++ + deny := s.alwaysDeny s.mu.Unlock() + if deny { + // 与腾讯 CDN 一致：连新签的票也直接空 401（成批拒绝的那几分钟）。 + w.WriteHeader(http.StatusUnauthorized) + return + } if s.errorPageNow() { w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write([]byte(`ticket expired`)) @@ -435,6 +536,13 @@ func (s *mediaTicketServer) serveErrorPage() { s.mu.Unlock() } +// denyAlways 让媒体端点连新签的票也一律空 401（模拟上游成批拒绝的那几分钟）。 +func (s *mediaTicketServer) denyAlways() { + s.mu.Lock() + s.alwaysDeny = true + s.mu.Unlock() +} + func (s *mediaTicketServer) errorPageNow() bool { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/handler/reader_routes.go b/internal/handler/reader_routes.go index 32fab4e..d5800c9 100644 --- a/internal/handler/reader_routes.go +++ b/internal/handler/reader_routes.go @@ -993,12 +993,8 @@ func streamReaderMediaBody(w io.Writer, body io.Reader) { const ( // readerAudioCacheAttempts 首次落盘的最多尝试次数（每次内部都会走「失败换票」）。 + // 续传只补没下完的部分，所以这里的次数比过去更可用：每次失败都能推进进度。 readerAudioCacheAttempts = 4 - // readerAudioCacheRetryDelay 重试退避基数，实际等待为 (attempt-1) 倍。 - readerAudioCacheRetryDelay = 500 * time.Millisecond - // readerAudioCacheMaxBytes 单个音频的落盘上限。有声书一章约 3–5MB， - // 给到 64MB 足够容纳长章节，也能挡住误当成音频的超大对象。 - readerAudioCacheMaxBytes = 64 << 20 // readerMediaReissueAttempts 换票重试次数：跨境 CDN 的「空 401」是概率性的， // 一张票不行往往换一张就好；旧实现只换一次，第二次失败就把 502 交给播放器。 readerMediaReissueAttempts = 3 @@ -1082,8 +1078,9 @@ func ensureCachedRemoteAudio( return path, nil } -// downloadReaderAudioToCache 整段拉取并落盘：上游断流/换票失败时重试， -// 落盘后按 Content-Length 校验长度，防止把截断的音频缓存下来。 +// downloadReaderAudioToCache 整段拉取并落盘，支持断点续传： +// 上游 RST/超时后用 Range 从已写入的位置接着下，不再整段重下； +// 只有字节数与上游声明的总长一致时才会生成正式缓存文件。 func downloadReaderAudioToCache( ctx context.Context, svc *service.Container, @@ -1091,76 +1088,30 @@ func downloadReaderAudioToCache( bookID, rawURL string, headers map[string]string, ) error { - var lastErr error - for attempt := 1; attempt <= readerAudioCacheAttempts; attempt++ { - if attempt > 1 { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(readerAudioCacheRetryDelay * time.Duration(attempt-1)): - } - } - out, err := fetchReaderMediaResilient(ctx, svc, book, rawURL, "", headers, true) - if err != nil { - lastErr = err - continue - } - if out == nil || out.resp == nil { - lastErr = fmt.Errorf("音源响应为空") - continue - } - resp := out.resp - status := resp.StatusCode - if status != http.StatusOK && status != http.StatusPartialContent { - _ = resp.Body.Close() - lastErr = fmt.Errorf("音源返回 %d", status) - continue - } - expected := resp.ContentLength - path, storeErr := svc.Reader.StoreCachedRemoteAudio(bookID, rawURL, func(w io.Writer) error { - if len(out.head) > 0 { - if _, err := w.Write(out.head); err != nil { - return err + _, err := svc.Reader.DownloadRemoteAudio(ctx, bookID, rawURL, readerAudioCacheAttempts, + func(rangeHeader string) (*http.Response, []byte, error) { + // 只有整段请求才需要嗅探响应体（判 HLS/错误页）；续传拿到的是 206 + // 二进制分片，读了还得再补写一次，没必要。 + out, fetchErr := fetchReaderMediaResilient(ctx, svc, book, rawURL, rangeHeader, headers, rangeHeader == "") + if fetchErr != nil { + if isReaderMediaNoRetry(fetchErr) { + return nil, nil, &reader.NoRetryError{Err: fetchErr} } + return nil, nil, fetchErr } - return copyLimited(w, resp.Body, readerAudioCacheMaxBytes) + return out.resp, out.head, nil }) - _ = resp.Body.Close() - if storeErr != nil { - lastErr = storeErr - continue - } - if expected > 0 { - info, statErr := os.Stat(path) - if statErr != nil { - _ = os.Remove(path) - lastErr = fmt.Errorf("音频缓存写入后不可读: %w", statErr) - continue - } - if info.Size() != expected { - _ = os.Remove(path) - lastErr = fmt.Errorf("音频写入不完整（%d/%d 字节）", info.Size(), expected) - continue - } - } - return nil - } - if lastErr == nil { - lastErr = fmt.Errorf("音源拉取失败") - } - return fmt.Errorf("已重试 %d 次: %w", readerAudioCacheAttempts, lastErr) + return err } -// copyLimited 复制响应体，超过上限时报错（避免把超大对象写进缓存）。 -func copyLimited(w io.Writer, r io.Reader, limit int64) error { - n, err := io.Copy(w, io.LimitReader(r, limit+1)) - if err != nil { - return err +// isReaderMediaNoRetry 判断这次失败是否「换个花样重试也没用」。 +// 用于让外层下载循环立刻收手，而不是把同一张被拒的票再打四遍。 +func isReaderMediaNoRetry(err error) bool { + if errors.Is(err, errReaderMediaStaleTicket) { + return true } - if n > limit { - return fmt.Errorf("音源超过 %d 字节上限", limit) - } - return nil + var rejected *readerMediaRejectedError + return errors.As(err, &rejected) } // readerAudioExt 取音频地址的后缀（用于推断 Content-Type）。 @@ -1230,6 +1181,33 @@ type readerMediaResult struct { // 不改变原有行为； // - 该换票但换不到、或换票后依然不可用 → 返回错误，由调用方给出明确文案， // 而不是让播放器对着必然失败的地址反复重试。 +// +// readerMediaRejectedError 上游拒绝了这张票，继续走同一条重试路径不会有进展。 +// +// status —— 上游状态码（401/403/410） +// refreshed —— 这次拒绝发生在「已经换过一张新票」之后 +// message —— 对外文案（留空时按状态码生成） +// +// 上游的拒绝常常成批出现（几分钟一波，随后自己恢复）：换票拿到的是新票，但 +// 同一时刻仍被同一个状态码挡回。连续两轮都这样时立刻停止重试——继续重试只会 +// 让播放器多等十几秒，结果并不会更好。 +type readerMediaRejectedError struct { + status int + refreshed bool + message string +} + +func (e *readerMediaRejectedError) Error() string { + if e.message != "" { + return e.message + } + return fmt.Sprintf("音源暂时被上游拒绝（HTTP %d），请稍后重试或换源", e.status) +} + +// errReaderMediaStaleTicket 换票没有拿到新票（书源还是给回同一张、或干脆给不出）， +// 再重试同一个书源不会有不同结果。具体原因由外层用 %w 串上。 +var errReaderMediaStaleTicket = errors.New("音源签名失效且重新取票失败") + func fetchReaderMedia( ctx context.Context, svc *service.Container, @@ -1257,7 +1235,10 @@ func fetchReaderMedia( case out.resp.StatusCode == http.StatusUnauthorized: _, _ = io.Copy(io.Discard, io.LimitReader(out.resp.Body, 1<<16)) _ = out.resp.Body.Close() - return nil, fmt.Errorf("音源签名失效（上游 401 拒绝），请稍后重试或换源") + return nil, &readerMediaRejectedError{ + status: http.StatusUnauthorized, + message: "音源签名失效（上游 401 拒绝），请稍后重试或换源", + } } return out, nil } @@ -1265,7 +1246,7 @@ func fetchReaderMedia( _, _ = io.Copy(io.Discard, io.LimitReader(out.resp.Body, 1<<16)) _ = out.resp.Body.Close() if refreshErr != nil { - return nil, fmt.Errorf("音源签名失效且重新取票失败: %w", refreshErr) + return nil, fmt.Errorf("%w: %v", errReaderMediaStaleTicket, refreshErr) } retried, _, retryErr := fetchReaderMediaOnce(ctx, svc, book, fresh, rangeHeader, options, probeBody) if retryErr != nil { @@ -1273,10 +1254,16 @@ func fetchReaderMedia( } if retried.rejected { _ = retried.resp.Body.Close() - return nil, fmt.Errorf("音源换票后仍返回非媒体内容，请稍后重试或换源") + return nil, &readerMediaRejectedError{ + status: retried.resp.StatusCode, + refreshed: true, + message: "音源换票后仍返回非媒体内容，请稍后重试或换源", + } } if reader.MediaNeedsRefresh(retried.resp.StatusCode) { - return nil, fmt.Errorf("音源换票后仍被拒绝（上游 %d），请稍后重试或换源", retried.resp.StatusCode) + status := retried.resp.StatusCode + _ = retried.resp.Body.Close() + return nil, &readerMediaRejectedError{status: status, refreshed: true} } return retried, nil } @@ -1285,8 +1272,11 @@ func fetchReaderMedia( // // 跨境 CDN 判掉签名票是概率性的：换一张票往往就能取到，而旧实现只换一次， // 第二次失败直接把 502 交给播放器。这里最多再重来 readerMediaReissueAttempts 轮， -// 每轮都会作废章节缓存、重新回源签票。不可恢复的 4xx（如 404）原样返回， -// 不做无意义的等待。 +// 每轮都会作废章节缓存、重新回源签票。 +// +// 但"没有进展"的失败会提前收手，否则用户要对着转圈干等三十多秒： +// - 换票拿不到新票（errReaderMediaStaleTicket）→ 立即结束； +// - 换过票仍是同一个拒绝状态码 → 两轮即结束（见 readerMediaRejectedError）。 func fetchReaderMediaResilient( ctx context.Context, svc *service.Container, @@ -1296,6 +1286,7 @@ func fetchReaderMediaResilient( probeBody bool, ) (*readerMediaResult, error) { var lastErr error + lastRejected := 0 for attempt := 1; attempt <= readerMediaReissueAttempts; attempt++ { if attempt > 1 { select { @@ -1313,13 +1304,29 @@ func fetchReaderMediaResilient( if !out.rejected && !reader.MediaNeedsRefresh(out.resp.StatusCode) { return out, nil } - lastErr = fmt.Errorf("音源换票后仍不可用（上游 %d）", out.resp.StatusCode) + status := out.resp.StatusCode _ = out.resp.Body.Close() - continue + lastErr = &readerMediaRejectedError{status: status, refreshed: true} + } else { + lastErr = err + if ctx.Err() != nil { + return nil, err + } + if errors.Is(err, errReaderMediaStaleTicket) { + return nil, err + } } - lastErr = err - if ctx.Err() != nil { - return nil, err + var rejected *readerMediaRejectedError + if errors.As(lastErr, &rejected) { + // 没有换票空间的拒绝：原样重试同一张票没有意义。 + if !rejected.refreshed { + return nil, lastErr + } + // 换过票还是同一个状态码：上游这几分钟就是不放行。 + if lastRejected == rejected.status { + return nil, lastErr + } + lastRejected = rejected.status } } return nil, lastErr diff --git a/internal/service/cache_cleanup.go b/internal/service/cache_cleanup.go index 8769e14..e2cfa75 100644 --- a/internal/service/cache_cleanup.go +++ b/internal/service/cache_cleanup.go @@ -364,3 +364,197 @@ func ImageCachePools(imagesDir string, originalsMaxBytes int64) []ImageCachePool } return []ImageCachePool{originals, derived} } + +// CacheCleanupResult 是全缓存容量淘汰的统计。 +type CacheCleanupResult struct { + TotalFilesBefore int + TotalBytesBefore int64 + DeletedFiles int + FreedBytes int64 + RemainingBytes int64 + // FreedByPool 记录每个池释放的字节数，方便日志定位到底是哪类缓存占满了。 + FreedByPool map[string]int64 +} + +// cacheEvictionEntry 是全局容量淘汰里的一个候选文件。 +type cacheEvictionEntry struct { + path string + pool string + size int64 + modTime time.Time +} + +// cacheEvictionPoolOrder 是全局容量清理的固定优先级。 +// 顺序即“先清谁”：超限时前面的池会被整池清空，才会轮到后面的池。 +// reader-audio 不在其中：它由自身 1GB 配额在写入时独立淘汰。 +var cacheEvictionPoolOrder = []string{ + "hls", + "images", + "library-covers", +} + +// cacheCleanupKeepPercent 是触发清理后回落到的水位（占上限的百分比）。 +// 留出余量，避免刚清完又被下一次写入立刻顶到上限。 +const cacheCleanupKeepPercent = 80 + +// tempCacheFile 判断文件名是否是缓存写入中的临时产物。这些文件正在被并发 +// 写入，容量统计和清理都应跳过。 +func tempCacheFile(name string) bool { + if tempImageCacheFile(name) { + return true + } + if strings.HasSuffix(name, ".part") { + return true + } + // 各缓存目录的原子写入临时文件前缀：.content- / .cover- / .audio- / .js-cache-。 + return strings.HasPrefix(name, ".content-") || + strings.HasPrefix(name, ".cover-") || + strings.HasPrefix(name, ".audio-") || + strings.HasPrefix(name, ".js-cache-") +} + +// collectCachePoolFiles 递归收集一个普通缓存池里的候选文件，按 mtime 从旧到新。 +func collectCachePoolFiles(root, pool string) []cacheEvictionEntry { + if strings.TrimSpace(root) == "" { + return nil + } + var out []cacheEvictionEntry + _ = filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil || info == nil || info.IsDir() { + return nil + } + if !info.Mode().IsRegular() { + return nil + } + name := info.Name() + if tempCacheFile(name) || strings.HasSuffix(name, ".fail") { + return nil + } + out = append(out, cacheEvictionEntry{ + path: path, + pool: pool, + size: info.Size(), + modTime: info.ModTime(), + }) + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].modTime.Before(out[j].modTime) }) + return out +} + +// collectImageCacheEntries 收集图片缓存候选文件。返回顺序沿用图片池的优先级： +// 先原图、后派生成品，池内按 mtime 从旧到新。 +func collectImageCacheEntries(imagesDir string) []cacheEvictionEntry { + if strings.TrimSpace(imagesDir) == "" { + return nil + } + if _, err := os.Stat(imagesDir); err != nil { + return nil + } + pools := ImageCachePools(imagesDir, 0) + var entries []imageCacheFileEntry + var failMarkers []imageCacheFileEntry + for i, pool := range pools { + collectImageCachePoolFiles(pool, i, &entries, &failMarkers) + } + ordered := evictionOrder(entries, map[string]bool{}) + out := make([]cacheEvictionEntry, 0, len(ordered)) + for _, e := range ordered { + out = append(out, cacheEvictionEntry{ + path: e.path, + pool: "images", + size: e.size, + modTime: e.modTime, + }) + } + return out +} + +// removeEmptyDirsUnder 删除 root 下淘汰后留下的空子目录（不含 root 本身）。 +func removeEmptyDirsUnder(root string) { + if strings.TrimSpace(root) == "" { + return + } + if _, err := os.Stat(root); err != nil { + return + } + dirs := []string{} + _ = filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil || info == nil { + return nil + } + if info.IsDir() && path != root { + dirs = append(dirs, path) + } + return nil + }) + for i := len(dirs) - 1; i >= 0; i-- { + _ = os.Remove(dirs[i]) + } +} + +// PruneCache 按全局容量池清理磁盘缓存。 +// +// 总量口径是 hls、images、library-covers 三个池之和；reader-audio 及正文缓存、 +// 书源文件缓存都不参与这里的总量计算——reader-audio 由自身 1GB 配额在写入时 +// 淘汰，正文缓存和书源文件缓存另有自己的 TTL / 配额。超过 totalBytes 时严格 +// 按 hls → images → library-covers 的顺序淘汰最旧文件，直到回落到上限的 80%。 +// originalsMaxBytes 是 images 原图池的独立子配额，在总量淘汰之后单独执行 +// （即使总量没超也会把原图压回配额）。 +func PruneCache(cacheDir string, totalBytes, originalsMaxBytes int64) (CacheCleanupResult, error) { + result := CacheCleanupResult{FreedByPool: map[string]int64{}} + if strings.TrimSpace(cacheDir) == "" { + return result, nil + } + + imagesDir := filepath.Join(cacheDir, "images") + ordered := make([]cacheEvictionEntry, 0, 1024) + for _, name := range cacheEvictionPoolOrder { + if name == "images" { + ordered = append(ordered, collectImageCacheEntries(imagesDir)...) + continue + } + ordered = append(ordered, collectCachePoolFiles(filepath.Join(cacheDir, name), name)...) + } + + result.TotalFilesBefore = len(ordered) + for _, entry := range ordered { + result.TotalBytesBefore += entry.size + } + result.RemainingBytes = result.TotalBytesBefore + + // 1. 全局总量：严格按池优先级淘汰，直到回落到水位。 + if totalBytes > 0 && result.RemainingBytes > totalBytes { + target := totalBytes * cacheCleanupKeepPercent / 100 + for _, entry := range ordered { + if result.RemainingBytes <= target { + break + } + if err := os.Remove(entry.path); err != nil { + continue + } + result.DeletedFiles++ + result.FreedBytes += entry.size + result.RemainingBytes -= entry.size + result.FreedByPool[entry.pool] += entry.size + _ = os.Remove(entry.path + ".fail") + } + } + + // 2. 图片原图子配额独立于总量：即使总量没超，也要把原图压回自己的配额， + // 同时清掉旧命名变体和孤儿失败标记。 + if _, err := os.Stat(imagesDir); err == nil { + pools := ImageCachePools(imagesDir, originalsMaxBytes) + if sub, err := PruneImageCachePools(pools, 0); err == nil { + result.DeletedFiles += sub.DeletedFiles + result.FreedBytes += sub.FreedBytes + result.RemainingBytes -= sub.FreedBytes + result.FreedByPool["images"] += sub.FreedBytes + } + } + + for _, name := range cacheEvictionPoolOrder { + removeEmptyDirsUnder(filepath.Join(cacheDir, name)) + } + return result, nil +} diff --git a/internal/service/cache_cleanup_test.go b/internal/service/cache_cleanup_test.go index bcc5b2d..51bf607 100644 --- a/internal/service/cache_cleanup_test.go +++ b/internal/service/cache_cleanup_test.go @@ -134,7 +134,7 @@ func TestPruneImageCache_ZeroOrNegativeLimit(t *testing.T) { } } -func TestSchedulerJobCleanImageCache(t *testing.T) { +func TestSchedulerJobCleanCache(t *testing.T) { cacheRoot := t.TempDir() imagesDir := filepath.Join(cacheRoot, "images") if err := os.MkdirAll(imagesDir, 0o750); err != nil { @@ -148,15 +148,117 @@ func TestSchedulerJobCleanImageCache(t *testing.T) { scheduler := NewSchedulerService(zap.NewNop(), nil, nil, nil, nil, nil, cacheRoot) // Set limit to 1MB; our file is 2MB -> should be pruned - scheduler.SetImageCachePolicyProvider(func() ImageCachePolicy { - return ImageCachePolicy{TotalBytes: 1 * 1024 * 1024} + scheduler.SetCacheCleanupPolicyProvider(func() CacheCleanupPolicy { + return CacheCleanupPolicy{TotalBytes: 1 * 1024 * 1024} }) - if err := scheduler.jobCleanImageCache(context.Background()); err != nil { - t.Fatalf("jobCleanImageCache failed: %v", err) + if err := scheduler.jobCleanCache(context.Background()); err != nil { + t.Fatalf("jobCleanCache failed: %v", err) } if _, err := os.Stat(f); !os.IsNotExist(err) { t.Fatalf("expected file to be pruned, got err=%v", err) } } + +// TestPruneCacheEvictionOrder 验证总量超限时严格按 +// hls → images → library-covers 的顺序淘汰，且 reader-audio 不参与。 +func TestPruneCacheEvictionOrder(t *testing.T) { + cacheRoot := t.TempDir() + now := time.Now() + + audio := filepath.Join(cacheRoot, "reader-audio", "old.mp3") + hls := filepath.Join(cacheRoot, "hls", "job", "seg.ts") + image := filepath.Join(cacheRoot, "images", "originals", "poster.jpg") + cover := filepath.Join(cacheRoot, "library-covers", "cover.jpg") + for _, p := range []string{audio, hls, image, cover} { + writePoolFile(t, p, 1024*1024, now.Add(-time.Hour)) + } + + // 计入总量的只有 hls+images+cover = 3MB，上限 2MB -> 回落到 1.6MB： + // 按顺序删 hls、images 两个文件即可；reader-audio 与封面都应保留。 + res, err := PruneCache(cacheRoot, 2*1024*1024, 0) + if err != nil { + t.Fatalf("prune failed: %v", err) + } + if res.DeletedFiles != 2 { + t.Fatalf("deleted files = %d, want 2", res.DeletedFiles) + } + mustExist(t, audio) + mustNotExist(t, hls) + mustNotExist(t, image) + mustExist(t, cover) + if _, ok := res.FreedByPool["reader-audio"]; ok { + t.Fatalf("reader-audio must not be part of the global pool, got %+v", res.FreedByPool) + } + if res.FreedByPool["hls"] != 1024*1024 || res.FreedByPool["images"] != 1024*1024 { + t.Fatalf("unexpected per-pool stats: %+v", res.FreedByPool) + } +} + +// TestPruneCacheLeavesReaderAudioAlone reader-audio 有独立的 1GB 写入时淘汰， +// 不参与总容量统计：即使它单独就超过全局上限，全局任务也不应删它。 +func TestPruneCacheLeavesReaderAudioAlone(t *testing.T) { + cacheRoot := t.TempDir() + audio := filepath.Join(cacheRoot, "reader-audio", "book", "chapter.mp3") + writePoolFile(t, audio, 4*1024*1024, time.Now().Add(-time.Hour)) + + res, err := PruneCache(cacheRoot, 1*1024*1024, 0) + if err != nil { + t.Fatalf("prune failed: %v", err) + } + if res.DeletedFiles != 0 { + t.Fatalf("deleted files = %d, want 0", res.DeletedFiles) + } + if res.TotalBytesBefore != 0 { + t.Fatalf("reader-audio should not count toward total, got %d bytes", res.TotalBytesBefore) + } + mustExist(t, audio) +} + +// TestPruneCacheAppliesOriginalsSubQuota 验证总量没超时，图片原图子配额仍然生效。 +func TestPruneCacheAppliesOriginalsSubQuota(t *testing.T) { + cacheRoot := t.TempDir() + now := time.Now() + + oldOriginal := filepath.Join(cacheRoot, "images", "originals", "old.jpg") + newOriginal := filepath.Join(cacheRoot, "images", "originals", "new.jpg") + derived := currentVariantPath(filepath.Join(cacheRoot, "images", "resize"), "d", 480, 600, 80) + writePoolFile(t, oldOriginal, 800, now.Add(-2*time.Hour)) + writePoolFile(t, newOriginal, 800, now.Add(-time.Hour)) + writePoolFile(t, derived, 800, now.Add(-time.Hour)) + + // 总量上限很大（不触发全局淘汰），原图子配额 1000B -> 删到 800B。 + res, err := PruneCache(cacheRoot, 100*1024*1024, 1000) + if err != nil { + t.Fatalf("prune failed: %v", err) + } + if res.DeletedFiles != 1 { + t.Fatalf("deleted files = %d, want 1", res.DeletedFiles) + } + mustNotExist(t, oldOriginal) + mustExist(t, newOriginal) + mustExist(t, derived) +} + +// TestPruneCacheOldestFirstWithinPool 验证同一池内先删最旧的文件。 +func TestPruneCacheOldestFirstWithinPool(t *testing.T) { + cacheRoot := t.TempDir() + now := time.Now() + + oldHLS := filepath.Join(cacheRoot, "hls", "job-a", "seg1.ts") + newHLS := filepath.Join(cacheRoot, "hls", "job-b", "seg1.ts") + writePoolFile(t, oldHLS, 1024*1024, now.Add(-2*time.Hour)) + writePoolFile(t, newHLS, 1024*1024, now.Add(-time.Minute)) + + // 上限 1.25MB -> 回落水位 1MB：删掉最旧的 1MB 后正好达标。 + res, err := PruneCache(cacheRoot, 1280*1024, 0) + if err != nil { + t.Fatalf("prune failed: %v", err) + } + if res.DeletedFiles != 1 { + t.Fatalf("deleted files = %d, want 1", res.DeletedFiles) + } + mustNotExist(t, oldHLS) + mustExist(t, newHLS) +} diff --git a/internal/service/image_proxy.go b/internal/service/image_proxy.go index c457957..128dc7d 100644 --- a/internal/service/image_proxy.go +++ b/internal/service/image_proxy.go @@ -199,17 +199,19 @@ func imageResizeConcurrency() int { return n } -// Prune removes oldest cached images until disk usage is within the configured limit. -// 原图池按独立配额优先淘汰，客户端热路径读取的成品图最后淘汰。 +// Prune 只执行图片原图子配额与旧缓存清理；整个磁盘缓存的总量淘汰由调度器的 +// cache_cleanup 任务（PruneCache）负责。 func (p *ImageProxy) Prune() (PruneImageCacheResult, error) { - if p.cfg == nil || p.cfg.Cache.ImagesMaxSizeMB <= 0 { + if p.cfg == nil { return PruneImageCacheResult{}, nil } var originalsMaxBytes int64 if p.cfg.Cache.ImagesOriginalsMaxSizeMB > 0 { originalsMaxBytes = int64(p.cfg.Cache.ImagesOriginalsMaxSizeMB) * 1024 * 1024 } - totalBytes := int64(p.cfg.Cache.ImagesMaxSizeMB) * 1024 * 1024 + if originalsMaxBytes <= 0 { + return PruneImageCacheResult{}, nil + } pools := ImageCachePools(p.cacheDir, originalsMaxBytes) - return PruneImageCachePools(pools, totalBytes) + return PruneImageCachePools(pools, 0) } diff --git a/internal/service/reader/audio_media_cache.go b/internal/service/reader/audio_media_cache.go index 4e1b578..7c2110b 100644 --- a/internal/service/reader/audio_media_cache.go +++ b/internal/service/reader/audio_media_cache.go @@ -1,15 +1,18 @@ package reader import ( + "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "io" + "net/http" "net/url" "os" "path/filepath" "sort" + "strconv" "strings" "sync" "time" @@ -29,6 +32,18 @@ const ( audioMediaMaxBytes = 2 << 30 // 2GiB：超出后按 mtime 淘汰最旧的 audioMediaPrunePeriod = 10 * time.Minute audioMediaPruneKeepPct = 90 + + // audioPartialSuffix 未下完的音轨分片后缀。 + // + // 分片跨请求保留：首播失败后用户再点一次、或播放器重连，都能用 Range 接着下， + // 不必从第 0 字节重来。分片名带 .part 后缀，不会被 OpenCachedRemoteAudio 当成 + // 完整音轨下发。 + audioPartialSuffix = ".part" + // audioMediaObjectMaxBytes 单个音频对象的落盘上限。有声书一章约 3–5MB， + // 给到 64MB 足够容纳长章节，也能挡住误当成音频的超大对象。 + audioMediaObjectMaxBytes = 64 << 20 + // audioDownloadRetryDelay 断点续传重试的退避基数，实际等待为 (attempt-1) 倍。 + audioDownloadRetryDelay = 500 * time.Millisecond ) var ( @@ -132,41 +147,253 @@ func (s *ReaderService) OpenCachedRemoteAudio(bookID, mediaURL string) (*os.File return f, info, true } -// StoreCachedRemoteAudio 把一段音源原子写入缓存，fill 负责把内容写进 writer。 -// 返回最终文件路径；内容为空或写入失败时不留下半成品。 -func (s *ReaderService) StoreCachedRemoteAudio(bookID, mediaURL string, fill func(io.Writer) error) (string, error) { +// errAudioObjectTooLarge 上游对象超过单章缓存上限（不是可重试的错误）。 +var errAudioObjectTooLarge = errors.New("音源超过缓存上限") + +// NoRetryError 标记「换着花样再试也不会有进展」的失败（例如上游正成批拒绝 +// 签名票、书源根本给不出新票）。带断点续传的下载器收到它会立刻停止，并把 +// Err 原样交给调用方——继续重试只会让播放器多等十几秒，还会让对方更坚定地 +// 拒绝我们。 +type NoRetryError struct{ Err error } + +func (e *NoRetryError) Error() string { return e.Err.Error() } +func (e *NoRetryError) Unwrap() error { return e.Err } + +// AudioRangeFetcher 取一次上游音轨响应。 +// +// rangeHeader 为空表示整段请求；否则形如 "bytes=N-"，用于从断点续传。 +// head 是调用方已经嗅探读过、必须补写回文件的前缀（只有整段请求才有）。 +// 调用方负责在内部完成换票；这里只看状态码和字节。 +type AudioRangeFetcher func(rangeHeader string) (resp *http.Response, head []byte, err error) + +// DownloadRemoteAudio 带断点续传地把整段音轨写进缓存，返回最终文件路径。 +// +// 为什么需要续传：腾讯 CDN 的跨境边缘常在传输中途 RST，旧实现每次重试都从 +// 第 0 字节重下，一章 3MB 白下几次就把首播拖到几十秒。这里把已写入的字节留在 +// .part 里，下一次用 Range: bytes=N- 续上；分片跨请求保留，用户点重试 +// 或播放器重连都能接着下。 +// +// 上游忽略 Range 直接回 200、或分片区间对不上时，丢弃分片从头重来。 +// 下载过程中的失败不会留下「看起来完整」的缓存文件：只有字节数与上游声明的 +// 总长一致才会原子改名到正式路径。 +func (s *ReaderService) DownloadRemoteAudio( + ctx context.Context, + bookID, mediaURL string, + attempts int, + fetch AudioRangeFetcher, +) (string, error) { path := s.RemoteAudioCachePath(bookID, mediaURL) if path == "" { return "", errors.New("音频缓存目录未配置") } + if fetch == nil { + return "", errors.New("音频下载器未配置") + } + if attempts <= 0 { + attempts = 1 + } if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { return "", fmt.Errorf("创建音频缓存目录失败: %w", err) } - tmp, err := os.CreateTemp(filepath.Dir(path), ".audio-*") + partPath := path + audioPartialSuffix + var offset int64 + if info, statErr := os.Stat(partPath); statErr == nil && info.Mode().IsRegular() { + offset = info.Size() + // 分片太旧或超过上限：续传没有意义，直接丢掉重下。 + if offset > audioMediaObjectMaxBytes || info.ModTime().Add(audioMediaTTL).Before(time.Now()) { + _ = os.Remove(partPath) + offset = 0 + } + } + file, err := os.OpenFile(partPath, os.O_CREATE|os.O_WRONLY, 0o640) if err != nil { return "", fmt.Errorf("创建音频缓存文件失败: %w", err) } - tmpName := tmp.Name() - fillErr := fill(tmp) - closeErr := tmp.Close() - if fillErr != nil || closeErr != nil { - _ = os.Remove(tmpName) - if fillErr != nil { - return "", fillErr + + var lastErr error + for attempt := 1; attempt <= attempts; attempt++ { + if attempt > 1 { + select { + case <-ctx.Done(): + _ = file.Close() + return "", ctx.Err() + case <-time.After(audioDownloadRetryDelay * time.Duration(attempt-1)): + } } - return "", closeErr + + rangeHeader := "" + if offset > 0 { + rangeHeader = fmt.Sprintf("bytes=%d-", offset) + } + resp, head, fetchErr := fetch(rangeHeader) + if fetchErr != nil { + var noRetry *NoRetryError + if errors.As(fetchErr, &noRetry) { + _ = file.Close() + // 已经写下的分片留着：上游恢复后用户再点一次能接着续。 + if offset == 0 { + _ = os.Remove(partPath) + } + return "", noRetry.Err + } + lastErr = fetchErr + continue + } + if resp == nil { + lastErr = errors.New("音源响应为空") + continue + } + + expected := int64(-1) + switch resp.StatusCode { + case http.StatusPartialContent: + start, total, ok := parseContentRange(resp.Header.Get("Content-Range")) + if !ok || offset == 0 || start != offset { + drainAndClose(resp.Body) + offset = 0 + _ = file.Truncate(0) + lastErr = fmt.Errorf("音源返回的分片区间不可用（%q）", resp.Header.Get("Content-Range")) + continue + } + expected = total + case http.StatusOK: + // 上游忽略了 Range（或者本来就是整段请求）：只能从第 0 字节写。 + if offset > 0 { + offset = 0 + if truncErr := file.Truncate(0); truncErr != nil { + drainAndClose(resp.Body) + lastErr = fmt.Errorf("重置音频缓存失败: %w", truncErr) + continue + } + } + expected = resp.ContentLength + case http.StatusRequestedRangeNotSatisfiable: + // 分片比上游对象还长（对象换过）：丢掉分片重来。 + drainAndClose(resp.Body) + offset = 0 + _ = file.Truncate(0) + lastErr = fmt.Errorf("音源分片已失效（%d）", resp.StatusCode) + continue + default: + drainAndClose(resp.Body) + lastErr = fmt.Errorf("音源返回 %d", resp.StatusCode) + continue + } + + if _, seekErr := file.Seek(offset, io.SeekStart); seekErr != nil { + drainAndClose(resp.Body) + lastErr = fmt.Errorf("音频缓存定位失败: %w", seekErr) + continue + } + written, writeErr := writeAudioChunk(file, head, resp.Body, audioMediaObjectMaxBytes-offset) + _ = resp.Body.Close() + offset += written + if errors.Is(writeErr, errAudioObjectTooLarge) { + _ = file.Close() + _ = os.Remove(partPath) + return "", fmt.Errorf("已重试 %d 次: %w", attempts, writeErr) + } + if writeErr != nil { + // 已写入的字节保留在分片里，下一轮从新的断点续传。 + lastErr = writeErr + continue + } + if expected >= 0 && offset != expected { + lastErr = fmt.Errorf("音频写入不完整（%d/%d 字节）", offset, expected) + continue + } + if offset == 0 { + lastErr = errors.New("音源内容为空") + continue + } + if truncErr := file.Truncate(offset); truncErr != nil { + _ = file.Close() + return "", fmt.Errorf("写入音频缓存失败: %w", truncErr) + } + if closeErr := file.Close(); closeErr != nil { + return "", fmt.Errorf("写入音频缓存失败: %w", closeErr) + } + if renameErr := os.Rename(partPath, path); renameErr != nil { + return "", fmt.Errorf("写入音频缓存失败: %w", renameErr) + } + s.pruneAudioMediaCache() + return path, nil } - info, statErr := os.Stat(tmpName) - if statErr != nil || info.Size() == 0 { - _ = os.Remove(tmpName) - return "", errors.New("音源内容为空") + + _ = file.Close() + if lastErr == nil { + lastErr = errors.New("音源拉取失败") } - if err := os.Rename(tmpName, path); err != nil { - _ = os.Remove(tmpName) - return "", fmt.Errorf("写入音频缓存失败: %w", err) + if offset > 0 { + return "", fmt.Errorf("已重试 %d 次: %w（已缓存 %d 字节，重试可续传）", attempts, lastErr, offset) } - s.pruneAudioMediaCache() - return path, nil + return "", fmt.Errorf("已重试 %d 次: %w", attempts, lastErr) +} + +// writeAudioChunk 先把已读前缀补写回文件，再把响应体写到 limit 字节为止。 +// 返回本次实际写入的字节数（出错时也包含已写成功的部分）。 +func writeAudioChunk(w io.Writer, head []byte, body io.Reader, limit int64) (int64, error) { + var total int64 + if len(head) > 0 { + n, err := w.Write(head) + total += int64(n) + if err != nil { + return total, err + } + } + if limit <= 0 { + return total, errAudioObjectTooLarge + } + n, err := io.Copy(w, io.LimitReader(body, limit-total+1)) + total += n + if err != nil { + return total, err + } + if total > limit { + return total, errAudioObjectTooLarge + } + return total, nil +} + +// parseContentRange 解析 "bytes 100-199/1234" 的起点与总长度。 +// 总长度未知（"*"）时 total 为 -1。 +func parseContentRange(v string) (start, total int64, ok bool) { + v = strings.TrimSpace(v) + const prefix = "bytes " + if len(v) <= len(prefix) || !strings.EqualFold(v[:len(prefix)], prefix) { + return 0, -1, false + } + v = strings.TrimSpace(v[len(prefix):]) + slash := strings.IndexByte(v, '/') + if slash < 0 { + return 0, -1, false + } + span, size := v[:slash], strings.TrimSpace(v[slash+1:]) + dash := strings.IndexByte(span, '-') + if dash <= 0 { + return 0, -1, false + } + start, err := strconv.ParseInt(strings.TrimSpace(span[:dash]), 10, 64) + if err != nil || start < 0 { + return 0, -1, false + } + if size == "" || size == "*" { + return start, -1, true + } + total, err = strconv.ParseInt(size, 10, 64) + if err != nil || total <= 0 { + return start, -1, true + } + return start, total, true +} + +// drainAndClose 丢弃剩余响应体并关闭连接（尽量让上游把连接收回去）。 +func drainAndClose(body io.ReadCloser) { + if body == nil { + return + } + _, _ = io.Copy(io.Discard, io.LimitReader(body, 64<<10)) + _ = body.Close() } // pruneAudioMediaCache 低频清理：删过期文件，超容量时按 mtime 淘汰最旧的。 diff --git a/internal/service/reader/audio_media_cache_test.go b/internal/service/reader/audio_media_cache_test.go index a420a9f..1d21682 100644 --- a/internal/service/reader/audio_media_cache_test.go +++ b/internal/service/reader/audio_media_cache_test.go @@ -1,6 +1,17 @@ package reader -import "testing" +import ( + "bytes" + "errors" + "fmt" + "io" + "net/http" + "os" + "strings" + "testing" + + "github.com/truewhile/MeBox/internal/config" +) func TestIsAudioMediaURL(t *testing.T) { cases := []struct { @@ -39,3 +50,220 @@ func TestAudioMediaCacheKeyIgnoresQuery(t *testing.T) { t.Fatalf("不同对象的缓存键不应相同") } } + +// ─── 断点续传下载器 ────────────────────────────────────────────────────── + +func TestParseContentRange(t *testing.T) { + cases := []struct { + in string + start int64 + total int64 + wantOK bool + }{ + {"bytes 100-199/1234", 100, 1234, true}, + {"bytes 0-99/*", 0, -1, true}, + {"BYTES 500-599/600", 500, 600, true}, + {"", 0, -1, false}, + {"bytes */1234", 0, -1, false}, + {"bytes x-1/2", 0, -1, false}, + } + for _, c := range cases { + start, total, ok := parseContentRange(c.in) + if ok != c.wantOK || start != c.start || total != c.total { + t.Errorf("parseContentRange(%q) = (%d,%d,%v)，期望 (%d,%d,%v)", + c.in, start, total, ok, c.start, c.total, c.wantOK) + } + } +} + +// errAfterReader 读到 data 末尾后返回 err，模拟上游中途 RST/提前 EOF。 +type errAfterReader struct { + data []byte + pos int + err error +} + +func (r *errAfterReader) Read(p []byte) (int, error) { + if r.pos >= len(r.data) { + return 0, r.err + } + n := copy(p, r.data[r.pos:]) + r.pos += n + return n, nil +} + +func (r *errAfterReader) Close() error { return nil } + +func newAudioDownloadTestService(t *testing.T) *ReaderService { + t.Helper() + return &ReaderService{cfg: &config.Config{Cache: config.CacheConfig{CacheDir: t.TempDir()}}} +} + +func okResponse(body []byte) *http.Response { + return &http.Response{ + StatusCode: http.StatusOK, + ContentLength: int64(len(body)), + Header: http.Header{"Content-Type": {"audio/mp4"}}, + Body: io.NopCloser(bytes.NewReader(body)), + } +} + +func rangeResponse(header string, body []byte) *http.Response { + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: http.Header{"Content-Range": {header}}, + Body: io.NopCloser(bytes.NewReader(body)), + } +} + +// TestDownloadRemoteAudioResumesAfterReset 上游写到一半断开后，第二轮必须用 +// Range 从断点续传，最终文件与上游内容逐字节一致。 +func TestDownloadRemoteAudioResumesAfterReset(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/a.m4a?vkey=1" + payload := bytes.Repeat([]byte("0123456789"), 500) // 5000 字节 + const firstChunk = 2000 + + var ranges []string + attempt := 0 + path, err := svc.DownloadRemoteAudio(t.Context(), "book-1", mediaURL, 3, + func(rangeHeader string) (*http.Response, []byte, error) { + ranges = append(ranges, rangeHeader) + attempt++ + if attempt == 1 { + // 声明全量长度，只写 2000 字节就断开。 + return &http.Response{ + StatusCode: http.StatusOK, + ContentLength: int64(len(payload)), + Header: http.Header{"Content-Type": {"audio/mp4"}}, + Body: &errAfterReader{ + data: payload[:firstChunk], + err: io.ErrUnexpectedEOF, + }, + }, nil, nil + } + if rangeHeader != fmt.Sprintf("bytes=%d-", firstChunk) { + t.Fatalf("第二轮应请求断点续传，实际 Range=%q", rangeHeader) + } + return rangeResponse( + fmt.Sprintf("bytes %d-%d/%d", firstChunk, len(payload)-1, len(payload)), + payload[firstChunk:], + ), nil, nil + }) + if err != nil { + t.Fatalf("续传后应当成功: %v", err) + } + got, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("读缓存文件失败: %v", readErr) + } + if !bytes.Equal(got, payload) { + t.Fatalf("续传后的内容不完整: %d 字节，期望 %d", len(got), len(payload)) + } + if len(ranges) != 2 || ranges[0] != "" { + t.Fatalf("请求序列 = %q，期望首次整段、第二次续传", ranges) + } +} + +// TestDownloadRemoteAudioResumesAcrossCalls 上一次请求失败留下的分片，下一次 +// （用户点重试）要能直接接着下，而不是从第 0 字节重来。 +func TestDownloadRemoteAudioResumesAcrossCalls(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/b.m4a" + payload := bytes.Repeat([]byte("abcdefgh"), 1000) + const firstChunk = 3000 + + // 第一次调用：上游只给 3000 字节就断，重试也没用（每次都 200 + 截断）。 + _, err := svc.DownloadRemoteAudio(t.Context(), "book-2", mediaURL, 1, + func(rangeHeader string) (*http.Response, []byte, error) { + return &http.Response{ + StatusCode: http.StatusOK, + ContentLength: int64(len(payload)), + Header: http.Header{"Content-Type": {"audio/mp4"}}, + Body: &errAfterReader{ + data: payload[:firstChunk], + err: io.ErrUnexpectedEOF, + }, + }, nil, nil + }) + if err == nil { + t.Fatalf("上游截断时应当失败") + } + + // 第二次调用：直接请求 bytes=3000-，续完。 + var seenRange string + path, err := svc.DownloadRemoteAudio(t.Context(), "book-2", mediaURL, 1, + func(rangeHeader string) (*http.Response, []byte, error) { + seenRange = rangeHeader + return rangeResponse( + fmt.Sprintf("bytes %d-%d/%d", firstChunk, len(payload)-1, len(payload)), + payload[firstChunk:], + ), nil, nil + }) + if err != nil { + t.Fatalf("跨请求续传应当成功: %v", err) + } + if seenRange != fmt.Sprintf("bytes=%d-", firstChunk) { + t.Fatalf("应当从上次的断点续传，实际 Range=%q", seenRange) + } + got, _ := os.ReadFile(path) + if !bytes.Equal(got, payload) { + t.Fatalf("跨请求续传后的内容不完整: %d 字节", len(got)) + } +} + +// TestDownloadRemoteAudioRestartsWhenRangeIgnored 上游忽略 Range 仍回 200 时， +// 必须丢掉分片从头写，不能把两段字节拼在一起。 +func TestDownloadRemoteAudioRestartsWhenRangeIgnored(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/c.m4a" + payload := bytes.Repeat([]byte("xyz9"), 500) + + var ranges []string + attempt := 0 + path, err := svc.DownloadRemoteAudio(t.Context(), "book-3", mediaURL, 3, + func(rangeHeader string) (*http.Response, []byte, error) { + ranges = append(ranges, rangeHeader) + attempt++ + if attempt == 1 { + return &http.Response{ + StatusCode: http.StatusOK, + ContentLength: int64(len(payload)), + Header: http.Header{"Content-Type": {"audio/mp4"}}, + Body: &errAfterReader{data: payload[:700], err: io.ErrUnexpectedEOF}, + }, nil, nil + } + // 无视 Range，整段回 200。 + return okResponse(payload), nil, nil + }) + if err != nil { + t.Fatalf("应当成功: %v", err) + } + if ranges[1] == "" { + t.Fatalf("第二轮应当带 Range") + } + got, _ := os.ReadFile(path) + if !bytes.Equal(got, payload) { + t.Fatalf("忽略 Range 后必须从头重写，实际 %d 字节（期望 %d）", len(got), len(payload)) + } +} + +// TestDownloadRemoteAudioStopsOnNoRetry 上游成批拒绝时（NoRetryError）， +// 下载器要立刻收手，不能把同一张被拒的票再打几遍。 +func TestDownloadRemoteAudioStopsOnNoRetry(t *testing.T) { + svc := newAudioDownloadTestService(t) + denied := errors.New("音源暂时被上游拒绝（HTTP 401）") + + calls := 0 + _, err := svc.DownloadRemoteAudio(t.Context(), "book-4", "https://cdn.example.com/long/d.m4a", 4, + func(rangeHeader string) (*http.Response, []byte, error) { + calls++ + return nil, nil, &NoRetryError{Err: denied} + }) + if err == nil || !strings.Contains(err.Error(), "401") { + t.Fatalf("应原样返回上游拒绝原因，实际 %v", err) + } + if calls != 1 { + t.Fatalf("不可重试的失败只应请求 1 次，实际 %d 次", calls) + } +} diff --git a/internal/service/scheduler.go b/internal/service/scheduler.go index 5248162..0615a5b 100644 --- a/internal/service/scheduler.go +++ b/internal/service/scheduler.go @@ -8,6 +8,9 @@ // organize_source opt-in — organize the configured staging folder. // transcode_cleanup every 24 h — purge HLS transcode artefacts // older than 24 h. +// cache_cleanup every 1 h — enforce the global disk-cache capacity +// (hls → images → library-covers; reader-audio +// has its own 1 GB eviction on write). // segment_prewarm every 6 h — fill IntroDB skip segments for queryable media. // // Each job runs at most once at a time (an in-flight run blocks the @@ -41,7 +44,7 @@ type SchedulerService struct { cacheDir string now func() time.Time - imagesPolicyProvider func() ImageCachePolicy + cachePolicyProvider func() CacheCleanupPolicy // readerContentCleaner 阅读正文缓存的清理钩子（由阅读模块注入）。 readerContentCleaner func(context.Context) @@ -79,22 +82,24 @@ func (s *SchedulerService) SetExpiryWatcher(watcher *TelegramExpiryWatcher) { s.expiryWatcher = watcher } -// ImageCachePolicy 是一次图片缓存清理要用的策略（全部为 0 表示不做任何清理）。 +// CacheCleanupPolicy 是一次全缓存容量清理要用的策略（全部为 0 表示不做任何清理）。 +// TotalBytes 是 hls / images / library-covers 三个池的总容量上限；OriginalsBytes +// 是 images 原图池的独立子配额。reader-audio 由自身 1GB 配额管理，不在此列。 // 只有容量配额，没有保留时长：淘汰只看占用，不按文件年龄删除。 -type ImageCachePolicy struct { +type CacheCleanupPolicy struct { TotalBytes int64 OriginalsBytes int64 } -func (s *SchedulerService) SetImageCachePolicyProvider(fn func() ImageCachePolicy) { - s.imagesPolicyProvider = fn +func (s *SchedulerService) SetCacheCleanupPolicyProvider(fn func() CacheCleanupPolicy) { + s.cachePolicyProvider = fn } -func (s *SchedulerService) imageCachePolicy() ImageCachePolicy { - if s.imagesPolicyProvider != nil { - return s.imagesPolicyProvider() +func (s *SchedulerService) cacheCleanupPolicy() CacheCleanupPolicy { + if s.cachePolicyProvider != nil { + return s.cachePolicyProvider() } - return ImageCachePolicy{} + return CacheCleanupPolicy{} } // SetReaderContentCleaner 注入阅读正文缓存的清理钩子（TTL 过期 + 容量 LRU）。 @@ -166,9 +171,9 @@ func (s *SchedulerService) Start(ctx context.Context) { run: s.jobCleanTranscodeCache, }, { - name: "image_cache_cleanup", + name: "cache_cleanup", interval: 1 * time.Hour, - run: s.jobCleanImageCache, + run: s.jobCleanCache, }, } // 阅读正文缓存清理只在注入钩子后注册（见 SetReaderContentCleaner）。 diff --git a/internal/service/scheduler_local_jobs.go b/internal/service/scheduler_local_jobs.go index d0fcee9..466e57c 100644 --- a/internal/service/scheduler_local_jobs.go +++ b/internal/service/scheduler_local_jobs.go @@ -299,30 +299,33 @@ func (s *SchedulerService) jobCleanTranscodeCache(ctx context.Context) error { return walkAndPrune(s.cacheDir+"/hls", cutoff) } -// jobCleanImageCache prunes the image proxy cache: 原图按独立配额优先淘汰， -// 总量超限时再淘汰派生成品。全程只看占用，不按文件年龄删除。 -func (s *SchedulerService) jobCleanImageCache(ctx context.Context) error { +// jobCleanCache enforces the global disk-cache capacity: when the total size of +// hls + images + library-covers exceeds the configured limit, the pools are +// evicted strictly in that order (oldest file first within each pool) until +// usage falls back to 80% of the limit. reader-audio is not part of the total; +// it has its own 1 GB eviction on write. The images original sub-quota is +// enforced afterwards as an independent constraint. +func (s *SchedulerService) jobCleanCache(ctx context.Context) error { if s.cacheDir == "" { return nil } - policy := s.imageCachePolicy() + policy := s.cacheCleanupPolicy() if policy.TotalBytes <= 0 && policy.OriginalsBytes <= 0 { return nil } - imagesDir := filepath.Join(s.cacheDir, "images") - pools := ImageCachePools(imagesDir, policy.OriginalsBytes) - res, err := PruneImageCachePools(pools, policy.TotalBytes) + res, err := PruneCache(s.cacheDir, policy.TotalBytes, policy.OriginalsBytes) if err != nil { if s.log != nil { - s.log.Warn("scheduled image cache cleanup failed", zap.Error(err)) + s.log.Warn("scheduled cache cleanup failed", zap.Error(err)) } return err } if res.DeletedFiles > 0 && s.log != nil { - s.log.Info("scheduled image cache cleanup completed", + s.log.Info("scheduled cache cleanup completed", zap.Int("deleted_files", res.DeletedFiles), zap.Int64("freed_bytes", res.FreedBytes), zap.Int64("remaining_bytes", res.RemainingBytes), + zap.Any("freed_by_pool", res.FreedByPool), ) } return nil diff --git a/internal/service/service_builder.go b/internal/service/service_builder.go index af8c9f7..a09d901 100644 --- a/internal/service/service_builder.go +++ b/internal/service/service_builder.go @@ -217,11 +217,11 @@ func (b *serviceContainerBuilder) initAccessAndStorageServices() { b.c.Scheduler.SetBackupService(b.c.Backup) b.c.Scheduler.SetOrganizePipeline(b.c.OrganizePipeline) b.c.Scheduler.SetSegments(b.c.Segments) - b.c.Scheduler.SetImageCachePolicyProvider(func() ImageCachePolicy { + b.c.Scheduler.SetCacheCleanupPolicyProvider(func() CacheCleanupPolicy { if b.cfg == nil { - return ImageCachePolicy{} + return CacheCleanupPolicy{} } - policy := ImageCachePolicy{ + policy := CacheCleanupPolicy{ TotalBytes: int64(b.cfg.Cache.ImagesMaxSizeMB) * 1024 * 1024, } if b.cfg.Cache.ImagesOriginalsMaxSizeMB > 0 { diff --git a/web/src/pages/reader/ReaderAudioPanel.tsx b/web/src/pages/reader/ReaderAudioPanel.tsx index 836e82f..39d5861 100644 --- a/web/src/pages/reader/ReaderAudioPanel.tsx +++ b/web/src/pages/reader/ReaderAudioPanel.tsx @@ -1,5 +1,5 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' -import { ChevronLeft, ChevronRight, Gauge, ListMusic, Loader2, Pause, Play, Scissors, Timer, X } from 'lucide-react' +import { ChevronLeft, ChevronRight, Gauge, ListMusic, Loader2, Pause, Play, RefreshCw, Scissors, Timer, X } from 'lucide-react' import { Virtuoso, type VirtuosoHandle } from 'react-virtuoso' import { useReaderAudioStore } from '../../stores/readerAudio' @@ -196,9 +196,21 @@ export function ReaderAudioPanel({ theme, onToggleMenu }: ReaderAudioPanelProps)

{audioError ? ( - - {audioError} - + // 失败态下这行就是重试入口：服务端这一轮要十几秒才判定失败， + // 与其让用户对着转圈，不如把「为什么」和「再试一次」一起给他。 + ) : ( <> {loading ? (transcoding ? '服务端转码中，请稍候…' : '加载中…') : playing ? '播放中' : '已暂停'} · diff --git a/web/src/pages/reader/readerAudioEngine.ts b/web/src/pages/reader/readerAudioEngine.ts index adcf1cf..53d6967 100644 --- a/web/src/pages/reader/readerAudioEngine.ts +++ b/web/src/pages/reader/readerAudioEngine.ts @@ -22,8 +22,15 @@ import { readerCoverSrc } from '../../utils/readerCover' const PROGRESS_SAVE_INTERVAL_MS = 5_000 /** -15s / +15s 步长（对齐面板与 legado SEEK_STEP）。 */ const SEEK_STEP = 15 -/** 音源加载失败后自动重试（重新换取签名地址）的次数上限。 */ -const AUDIO_RETRY_LIMIT = 2 +/** + * 音源加载失败后自动重试的退避间隔（毫秒），数组长度即自动重试次数。 + * + * 别调得太短：服务端媒体代理要回源换票、拉整章音频，失败一次本身就要十几秒， + * 立刻重试只会让用户对着转圈更久。第一次 3 秒、第二次 9 秒，总等待可控。 + */ +const AUDIO_RETRY_DELAYS_MS = [3_000, 9_000] +/** 自动重试次数上限（与退避表的长度一致）。 */ +const AUDIO_RETRY_LIMIT = AUDIO_RETRY_DELAYS_MS.length export interface LoadChapterInput { bookId: string @@ -183,10 +190,23 @@ class ReaderAudioEngine { } toggle(): void { + // 失败态下播放键就是「重试」：重新解析本章音轨（换一张签名票）再装源， + // 而不是对着一份已经出错的 src 再调 play()。 + if (useReaderAudioStore.getState().status === 'error') { + this.retry() + return + } if (this.audio.paused) this.play() else this.pause() } + /** 用户手动重试：清掉自动重试计数与待执行定时器，立刻重新解析本章音轨。 */ + retry(): void { + this.resetRetry() + useReaderAudioStore.setState({ status: 'loading', error: '' }) + void this.reloadCurrentChapter() + } + seekBy(delta: number): void { const total = Number.isFinite(this.audio.duration) ? this.audio.duration @@ -482,13 +502,23 @@ class ReaderAudioEngine { } } - /** 退避后再重新解析本章音轨；期间保持 loading 态，不给用户闪错误。 */ + /** + * 退避后再重新解析本章音轨。 + * + * 期间刻意停留在 error 态而不是 loading：服务端这次请求本来就要十几秒才判定 + * 失败，再挂一个转圈只会让人以为卡死。明确告诉他「几秒后自动重试」，并且这 + * 行提示本身可以点，等于把等待权交回用户。 + */ private scheduleRetry(): void { if (this.retryTimer !== null) window.clearTimeout(this.retryTimer) - useReaderAudioStore.setState({ status: 'loading', error: '' }) - const delay = 800 * this.retryCount + const delay = AUDIO_RETRY_DELAYS_MS[Math.min(this.retryCount - 1, AUDIO_RETRY_LIMIT - 1)] + useReaderAudioStore.setState({ + status: 'error', + error: `音源暂时取不到，${Math.round(delay / 1000)} 秒后自动重试`, + }) this.retryTimer = window.setTimeout(() => { this.retryTimer = null + useReaderAudioStore.setState({ status: 'loading', error: '' }) void this.reloadCurrentChapter() }, delay) } @@ -615,7 +645,7 @@ class ReaderAudioEngine { audio.addEventListener('error', () => { const code = audio.error?.code - // 服务端代理拉的是腾讯 CDN 的直链，跨境链路会偶发 502 / 中途断流。 + // 服务端代理拉的是第三方 CDN 的直链，跨境链路会偶发被上游拒绝 / 中途断流。 // 浏览器对这两种情况都报 SRC_NOT_SUPPORTED（看起来像「格式不支持」）， // 实际上重新取一次章节（换一张签名票）通常就能播，所以先自动重试。 const retryable = @@ -625,11 +655,12 @@ class ReaderAudioEngine { this.scheduleRetry() return } + // 重试用完：文案要能解释「为什么播不了」，并且面板上这行本身就是重试按钮。 this.fail( code === MediaError.MEDIA_ERR_SRC_NOT_SUPPORTED - ? '音频加载失败：服务端多次重试后仍取不到音源，请稍后再试' + ? '音源暂时被上游拒绝' : code === MediaError.MEDIA_ERR_NETWORK - ? '音频加载失败，请检查网络后重试' + ? '网络异常，音源没取到' : '音频无法播放', ) }) diff --git a/web/src/pages/settingsGroupGeneral.ts b/web/src/pages/settingsGroupGeneral.ts index 738ea93..a5cc810 100644 --- a/web/src/pages/settingsGroupGeneral.ts +++ b/web/src/pages/settingsGroupGeneral.ts @@ -129,17 +129,17 @@ export const generalSettingsGroup: SettingGroup = { }, { key: 'cache.images_max_size_mb', - label: '图片缓存上限 (MB)', + label: '磁盘缓存总容量上限 (MB)', type: 'number', - hint: '海报、剧照等图片代理缓存目录 (cache/images) 的总占用上限。超过上限时先淘汰原图、再按修改时间清理最旧的缩略图；上限过小会导致旧图被淘汰后又要重新从上游/挂载的 Emby 下载。设为 0 表示不限制，默认 2000MB', - defaultValue: '2000', - placeholder: '2000', + hint: '磁盘缓存（hls、images、library-covers）的总占用上限。超过上限时严格按「转码 HLS → 图片 → 媒体库封面」的顺序淘汰最旧的文件，直到回落到上限的 80%。有声书转码缓存（reader-audio）另有 1GB 写入时淘汰，不计入这里。设为 0 表示不限制，默认 2048MB（2GB）', + defaultValue: '2048', + placeholder: '2048', }, { key: 'cache.images_originals_max_size_mb', - label: '原图缓存上限 (MB)', + label: '图片原图缓存上限 (MB)', type: 'number', - hint: '原图只是生成缩略图的原料，丢了可以重新回源，因此单独给一个小配额（默认 256MB）。总量超限时原图也优先被淘汰，保证客户端实际读取的缩略图能留住。设为 0 表示不单独限制', + hint: 'originals 子目录的独立配额（默认 256MB）。原图只是生成缩略图的原料，丢了可以重新回源，因此单独给小配额；图片池被淘汰时优先删原图，保证客户端实际读取的缩略图能留住。设为 0 表示不单独限制', defaultValue: '256', placeholder: '256', },