From 9b8ca33279129c8c265081296c1280f1efa00fe5 Mon Sep 17 00:00:00 2001 From: truewhile <62226914+truewhile@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:23:58 +0800 Subject: [PATCH] =?UTF-8?q?bug=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/handler/reader_audio_range_test.go | 406 +++++++++++++++++++ internal/handler/reader_routes.go | 387 ++++++++++++++++-- internal/service/reader/audio_media_cache.go | 81 ++++ 3 files changed, 848 insertions(+), 26 deletions(-) create mode 100644 internal/handler/reader_audio_range_test.go diff --git a/internal/handler/reader_audio_range_test.go b/internal/handler/reader_audio_range_test.go new file mode 100644 index 0000000..b86f3f0 --- /dev/null +++ b/internal/handler/reader_audio_range_test.go @@ -0,0 +1,406 @@ +package handler + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/service" + "github.com/truewhile/MeBox/internal/service/reader" +) + +// 音轨「区间直通」的行为契约。 +// +// 背景(真实故障):播放器(media3)取一条 m4a 会开两次请求——先整段打开嗅探 +// 容器,再带 Range 从目标位置重开。旧实现只允许「无 Range + 无分片」的请求 +// 边下边发,于是带 Range 的那次(真正的起播请求)必须把整章下完再按区间下发。 +// 线上日志:每章固定 200(探针,被播放器掐断)+ 206(整章下完才吐第一个字节), +// 42 章里「媒体已装载 → 实际起播」平均 4.79 秒,最坏 35 秒。 +// +// 现在的契约: +// - 带 Range 的冷缓存请求 → 上游一到字节就往下发,不等整章; +// - 分片已有的前缀从本地发,上游只续下剩余字节; +// - 整章落满时把分片提升为正式缓存,之后走本地; +// - 上游不认 Range 时也要给出正确的 206(跳过前缀),且不写带洞的分片; +// - 拿不到写锁时降级成纯透传,绝不排队等前一个下载。 + +// rangeUpstream 支持 Range 的上游:/content 给出音轨地址,/media 回音频本体。 +type rangeUpstream struct { + *httptest.Server + payload []byte + honorRange bool + // firstChunk > 0 且 release 非 nil 时:先写这么多字节再等放行, + // 用来证明「客户端不必等整章下完」。 + firstChunk int + release chan struct{} + + mu sync.Mutex + ranges []string + hits int +} + +func newRangeUpstream(t *testing.T, payload []byte, honorRange bool) *rangeUpstream { + t.Helper() + u := &rangeUpstream{payload: payload, honorRange: honorRange} + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasPrefix(r.URL.Path, "/book/1"): + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write([]byte(`

区间之书

`)) + case r.URL.Path == "/content": + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _, _ = w.Write([]byte(fmt.Sprintf(`{"data":{"url":"%s/media"}}`, u.URL))) + case r.URL.Path == "/media": + u.serveMedia(w, r) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(u.Close) + return u +} + +func (u *rangeUpstream) serveMedia(w http.ResponseWriter, r *http.Request) { + u.mu.Lock() + u.hits++ + u.ranges = append(u.ranges, r.Header.Get("Range")) + u.mu.Unlock() + + total := len(u.payload) + start, end := 0, total-1 + partial := false + if u.honorRange { + if s, e, ok := parseReaderAudioRange(r.Header.Get("Range")); ok { + start = int(s) + if e >= 0 && int(e) < end { + end = int(e) + } + partial = true + } + } + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Accept-Ranges", "bytes") + if start >= total { + w.WriteHeader(http.StatusRequestedRangeNotSatisfiable) + return + } + body := u.payload[start : end+1] + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + if partial { + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, total)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.WriteHeader(http.StatusOK) + } + if u.release != nil && u.firstChunk > 0 && len(body) > u.firstChunk { + _, _ = w.Write(body[:u.firstChunk]) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + select { + case <-u.release: + case <-r.Context().Done(): + return + } + body = body[u.firstChunk:] + } + _, _ = w.Write(body) +} + +func (u *rangeUpstream) rangeHeaders() []string { + u.mu.Lock() + defer u.mu.Unlock() + return append([]string(nil), u.ranges...) +} + +func (u *rangeUpstream) requestCount() int { + u.mu.Lock() + defer u.mu.Unlock() + return u.hits +} + +// prepareRangeBook 建一本「书源直连上游」的音频书,并取回该音轨的签名代理路径。 +func prepareRangeBook(t *testing.T, upstream string) (*service.Container, *model.ReaderBook, string) { + t.Helper() + svc := newReaderHandlerContainer(t) + svc.Cfg.Cache.CacheDir = t.TempDir() + srcJSON := fmt.Sprintf(`{ + "bookSourceName": "区间直通测试", + "bookSourceType": 1, + "bookSourceUrl": %q, + "ruleContent": {"content": "$.data.url"} +}`, upstream) + if _, err := svc.Reader.ImportSources(t.Context(), "u1", srcJSON); err != nil { + t.Fatalf("导入书源失败: %v", err) + } + srcs, err := svc.Reader.ListSources(t.Context(), "u1") + if err != nil || len(srcs) == 0 { + t.Fatalf("读不到书源: %v", err) + } + book, err := svc.Reader.AddBook(t.Context(), "u1", reader.SearchOrigin{ + SourceID: srcs[0].ID, + Origin: upstream, + OriginName: "区间直通测试", + OriginType: 1, + BookURL: upstream + "/book/1", + }, "区间之书", "某作者", "") + if err != nil { + t.Fatalf("加入书架失败: %v", err) + } + if err := svc.Reader.SaveChapters(t.Context(), book.ID, []reader.ChapterInput{ + {Index: 0, Title: "第一章", URL: upstream + "/content"}, + }); err != nil { + t.Fatalf("保存目录失败: %v", err) + } + out, err := svc.Reader.GetContentForBook(t.Context(), "u1", book.ID, 0, reader.ContentOptions{}) + if err != nil { + t.Fatalf("取章节失败: %v", err) + } + if len(out.Tracks) != 1 { + t.Fatalf("音轨数 = %d,期望 1", len(out.Tracks)) + } + return svc, book, out.Tracks[0] +} + +func audioPayload(n int) []byte { + payload := make([]byte, n) + for i := range payload { + payload[i] = byte(i % 251) + } + return payload +} + +// waitCached 等待分片被提升为正式缓存(提升发生在响应收尾,与客户端读完成之间有微小竞态)。 +func waitCached(t *testing.T, svc *service.Container, bookID, mediaURL string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if svc.Reader.HasCachedRemoteAudio(bookID, mediaURL) { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("整章下完后应提升为正式缓存") +} + +// TestReaderAudioRangeStreamsWithoutWaitingForWholeChapter 冷缓存 + Range: +// 上游只吐了第一段,客户端就必须已经拿到字节。 +func TestReaderAudioRangeStreamsWithoutWaitingForWholeChapter(t *testing.T) { + payload := audioPayload(512 << 10) + upstream := newRangeUpstream(t, payload, true) + upstream.firstChunk = 64 << 10 + upstream.release = make(chan struct{}) + + svc, book, track := prepareRangeBook(t, upstream.URL) + mediaURL := upstream.URL + "/media" + srv := httptest.NewServer(mediaRouterForTest(svc)) + defer srv.Close() + + req, err := http.NewRequest(http.MethodGet, srv.URL+track, nil) + if err != nil { + t.Fatalf("构造请求失败: %v", err) + } + req.Header.Set("Range", "bytes=0-") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("请求失败: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusPartialContent { + t.Fatalf("状态码 = %d,期望 206", resp.StatusCode) + } + if got := resp.Header.Get("Content-Range"); got != fmt.Sprintf("bytes 0-%d/%d", len(payload)-1, len(payload)) { + t.Fatalf("Content-Range = %q", got) + } + if got := resp.Header.Get("Content-Length"); got != strconv.Itoa(len(payload)) { + t.Fatalf("Content-Length = %q,期望 %d", got, len(payload)) + } + + // 上游此时还被 release 卡着:能在超时内读到第一段就说明没有等整章。 + head := make([]byte, upstream.firstChunk) + readDone := make(chan error, 1) + go func() { + _, readErr := io.ReadFull(resp.Body, head) + readDone <- readErr + }() + select { + case readErr := <-readDone: + if readErr != nil { + t.Fatalf("读首段失败: %v", readErr) + } + case <-time.After(3 * time.Second): + t.Fatalf("上游还没给完就卡住了:客户端没能在整章下完前拿到第一个字节") + } + if !bytes.Equal(head, payload[:upstream.firstChunk]) { + t.Fatalf("首段字节不一致") + } + + close(upstream.release) + rest, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("读剩余字节失败: %v", err) + } + if got := append(head, rest...); !bytes.Equal(got, payload) { + t.Fatalf("整体字节不一致: got %d bytes", len(got)) + } + if got := upstream.rangeHeaders(); len(got) != 1 || got[0] != "bytes=0-" { + t.Fatalf("上游收到的 Range = %v,期望且仅一次 bytes=0-", got) + } + waitCached(t, svc, book.ID, mediaURL) +} + +// TestReaderAudioRangeResumesFromPartialFragment 分片里已有前缀时,前缀从本地发, +// 上游只续下剩余字节;整章落满后提升为正式缓存。 +func TestReaderAudioRangeResumesFromPartialFragment(t *testing.T) { + payload := audioPayload(256 << 10) + upstream := newRangeUpstream(t, payload, true) + svc, book, track := prepareRangeBook(t, upstream.URL) + mediaURL := upstream.URL + "/media" + + // 模拟「播放器嗅探完就断开」留下的分片:前 96KB 已经在盘上。 + const partLen = 96 << 10 + cachePath := svc.Reader.RemoteAudioCachePath(book.ID, mediaURL) + if cachePath == "" { + t.Fatalf("缓存路径为空") + } + if mkErr := os.MkdirAll(cachePath[:strings.LastIndex(cachePath, string(os.PathSeparator))], 0o750); mkErr != nil { + t.Fatalf("建缓存目录失败: %v", mkErr) + } + if writeErr := os.WriteFile(cachePath+".part", payload[:partLen], 0o640); writeErr != nil { + t.Fatalf("写分片失败: %v", writeErr) + } + + code, body := getReaderMediaWithRange(t, mediaRouterForTest(svc), track, "bytes=0-") + if code != http.StatusPartialContent { + t.Fatalf("状态码 = %d,期望 206 body=%.120s", code, body) + } + if !bytes.Equal([]byte(body), payload) { + t.Fatalf("拼接后的字节不一致: got %d bytes want %d", len(body), len(payload)) + } + headers := upstream.rangeHeaders() + if len(headers) != 1 || headers[0] != fmt.Sprintf("bytes=%d-", partLen) { + t.Fatalf("上游收到的 Range = %v,期望从分片长度续下 bytes=%d-", headers, partLen) + } + waitCached(t, svc, book.ID, mediaURL) + + // 提升之后同一首歌不再回源。 + before := upstream.requestCount() + code, body = getReaderMedia(t, mediaRouterForTest(svc), track) + if code != http.StatusOK || !bytes.Equal([]byte(body), payload) { + t.Fatalf("命中缓存失败 status=%d len=%d", code, len(body)) + } + if after := upstream.requestCount(); after != before { + t.Fatalf("命中缓存后不应再回源,多了 %d 次", after-before) + } +} + +// TestReaderAudioRangeHandlesUpstreamThatIgnoresRange 上游不认 Range(回 200 整段)时, +// 仍要按客户端要的起点给出正确的 206,并且不写带洞的分片。 +func TestReaderAudioRangeHandlesUpstreamThatIgnoresRange(t *testing.T) { + payload := audioPayload(300 << 10) + upstream := newRangeUpstream(t, payload, false) + svc, book, track := prepareRangeBook(t, upstream.URL) + mediaURL := upstream.URL + "/media" + + const start = 1000 + code, body := getReaderMediaWithRange(t, mediaRouterForTest(svc), track, fmt.Sprintf("bytes=%d-", start)) + if code != http.StatusPartialContent { + t.Fatalf("状态码 = %d,期望 206 body=%.120s", code, body) + } + if !bytes.Equal([]byte(body), payload[start:]) { + t.Fatalf("区间字节不一致: got %d want %d", len(body), len(payload)-start) + } + // 起点不在分片内 → 不能落盘(否则分片开头会是空洞)。 + if svc.Reader.HasPartialRemoteAudio(book.ID, mediaURL) { + t.Fatalf("起点靠后的区间请求不应生成分片") + } + if svc.Reader.HasCachedRemoteAudio(book.ID, mediaURL) { + t.Fatalf("只下了一段不应生成完整缓存") + } +} + +// TestReaderAudioRangeFallsBackToPassthroughWhenWriterBusy 拿不到写锁时不排队: +// 仍然立刻把区间发出去,只是这一轮不落盘。 +func TestReaderAudioRangeFallsBackToPassthroughWhenWriterBusy(t *testing.T) { + payload := audioPayload(128 << 10) + upstream := newRangeUpstream(t, payload, true) + svc, book, track := prepareRangeBook(t, upstream.URL) + mediaURL := upstream.URL + "/media" + + cacheKey := svc.Reader.RemoteAudioCachePath(book.ID, mediaURL) + if cacheKey == "" { + t.Fatalf("缓存路径为空") + } + unlock := readerAudioStreamLocks.Lock(cacheKey) + defer unlock() + + done := make(chan struct{}) + var code int + var body string + go func() { + code, body = getReaderMediaWithRange(t, mediaRouterForTest(svc), track, "bytes=0-") + close(done) + }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatalf("写锁被占用时不该让请求排队等待") + } + if code != http.StatusPartialContent { + t.Fatalf("状态码 = %d,期望 206 body=%.120s", code, body) + } + if !bytes.Equal([]byte(body), payload) { + t.Fatalf("区间字节不一致: got %d want %d", len(body), len(payload)) + } + if svc.Reader.HasCachedRemoteAudio(book.ID, mediaURL) { + t.Fatalf("降级透传不应写缓存") + } +} + +func TestParseReaderAudioRangeCases(t *testing.T) { + cases := []struct { + in string + start int64 + end int64 + want bool + }{ + {"bytes=0-", 0, -1, true}, + {"bytes=100-199", 100, 199, true}, + {"Bytes=1024-", 1024, -1, true}, + {" bytes=5-5 ", 5, 5, true}, + {"bytes=-500", 0, 0, false}, + {"bytes=0-99,200-299", 0, 0, false}, + {"items=0-", 0, 0, false}, + {"bytes=abc-", 0, 0, false}, + {"bytes=200-100", 0, 0, false}, + {"", 0, 0, false}, + } + for _, c := range cases { + start, end, ok := parseReaderAudioRange(c.in) + if ok != c.want || start != c.start || end != c.end { + t.Errorf("parseReaderAudioRange(%q) = (%d,%d,%v),期望 (%d,%d,%v)", + c.in, start, end, ok, c.start, c.end, c.want) + } + } +} + +// getReaderMediaWithRange 发一次带 Range 头的媒体代理请求。 +func getReaderMediaWithRange(t *testing.T, router *gin.Engine, signedPath, rangeHeader string) (int, string) { + t.Helper() + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, signedPath, nil) + req.Header.Set("Range", rangeHeader) + router.ServeHTTP(w, req) + return w.Code, w.Body.String() +} diff --git a/internal/handler/reader_routes.go b/internal/handler/reader_routes.go index d8ab5b6..3e4dd39 100644 --- a/internal/handler/reader_routes.go +++ b/internal/handler/reader_routes.go @@ -1062,15 +1062,77 @@ func (k *keyedMutex) Lock(key string) func() { } } +// keyedMutexPollInterval 等锁的重试间隔。只用于「已有写入者时短暂等待」, +// 不值得为它引入条件变量。 +const keyedMutexPollInterval = 20 * time.Millisecond + +// LockWait 最多等 timeout 就放弃。 +// +// 音频缓存的写入者可能是「首播直通」,它会一直持锁到客户端读完整章;让后来的 +// 请求排队等它,播放器的重试就会干等几十秒(用户看到的就是「一直转圈」)。 +// 等不到锁的请求改走纯透传,字节照样能出声,只是这次不落盘。 +func (k *keyedMutex) LockWait(key string, timeout time.Duration) (func(), bool) { + deadline := time.Now().Add(timeout) + for { + if unlock, ok := k.acquire(key); ok { + return unlock, true + } + if time.Now().After(deadline) { + return nil, false + } + time.Sleep(keyedMutexPollInterval) + } +} + +// acquire 取一次锁,取不到立刻返回。引用计数保证条目在最后一个持有者归还后回收。 +func (k *keyedMutex) acquire(key string) (func(), bool) { + k.mu.Lock() + entry := k.locks[key] + if entry == nil { + entry = &keyedMutexEntry{} + k.locks[key] = entry + } + entry.refs++ + k.mu.Unlock() + + if !entry.mu.TryLock() { + k.release(key, entry) + return nil, false + } + return func() { + entry.mu.Unlock() + k.release(key, entry) + }, true +} + +// release 归还一次引用,无人持有时回收条目。 +func (k *keyedMutex) release(key string, entry *keyedMutexEntry) { + k.mu.Lock() + entry.refs-- + if entry.refs == 0 { + delete(k.locks, key) + } + k.mu.Unlock() +} + +// readerAudioLockWait 单条音轨的写锁最多等多久;等不到就降级成纯透传。 +const readerAudioLockWait = 500 * time.Millisecond + // handleCachedReaderAudio 音频直链的统一入口。 // -// 三条路径,按代价从低到高: +// 四条路径,按代价从低到高: // 1. 命中完整缓存 → http.ServeContent 按 Range 下发(拖动、续播都读本地); -// 2. 冷启动首播 → 边下边发给客户端,同时落盘(DownloadRemoteAudioStreaming)。 -// 旧实现是「整段下完再下发」,慢链路上用户要对着转圈等整章 3–5MB, -// 把「偶发失败」换成了「必然慢」;直通后几百毫秒就能出声,落盘照旧; -// 3. 已有 .part 分片 → 走整段下载:分片意味着上次已经下到一半,同步续传 -// 剩余字节比重新直通更省流量,下完再一次性下发。 +// 2. 带 Range 的冷缓存 → 区间直通(serveReaderAudioRange):客户端要哪一段就 +// 从上游边下边发哪一段,同时补全分片。**播放器嗅探完容器一定会带 Range 重开 +// 一次**(media3 的固有行为),所以这条才是真正的起播路径。旧实现要求整章 +// 下完才能按区间下发,每章起播都要等 1–4MB 跨境回源,实测 1.5–35 秒; +// 3. 冷启动整段请求 → 边下边发给客户端,同时落盘(streamReaderAudioInto): +// 播放器嗅探容器的那次 200 请求,几百毫秒就能出声,落盘照旧; +// 4. 已有 .part 分片的整段请求 → 走整段下载:分片意味着上次已经下到一半, +// 同步续传剩余字节比重新直通更省流量,下完再一次性下发。 +// +// 同一条音轨同时只允许一个写入者。等不到写锁的请求降级成纯透传(不排队、不 +// 落盘):首播直通会一直持锁到客户端读完整章,让它排队等于让播放器的重试干等。 func handleCachedReaderAudio( c *gin.Context, svc *service.Container, @@ -1083,15 +1145,36 @@ func handleCachedReaderAudio( serveCachedReaderAudio(c, f, info, rawURL) return } + cacheKey := svc.Reader.RemoteAudioCachePath(bookID, rawURL) + if cacheKey == "" { + c.String(http.StatusBadGateway, "音频缓存不可用") + return + } + rangeHeader := strings.TrimSpace(c.GetHeader("Range")) + unlock, locked := readerAudioStreamLocks.LockWait(cacheKey, readerAudioLockWait) + if locked { + defer unlock() + // 排队期间前一个请求可能已经把整章下完:直接下发本地文件,不重复回源。 + if f, info, ok := svc.Reader.OpenCachedRemoteAudio(bookID, rawURL); ok { + defer f.Close() + serveCachedReaderAudio(c, f, info, rawURL) + return + } + } + if rangeHeader != "" && serveReaderAudioRange(c, svc, book, bookID, rawURL, headers, rangeHeader, locked) { + return + } + if !locked { + proxyReaderAudio(c, svc, book, bookID, rawURL, headers, rangeHeader) + return + } // 播放器拖进度时会带 Range。已缓存的文件才谈得上 Range;未缓存时直通 // 只能从 0 发,否则客户端拿到的区间与它请求的对不上。 - rangeHeader := strings.TrimSpace(c.GetHeader("Range")) canStream := c.Request.Method == http.MethodGet && rangeHeader == "" && !svc.Reader.HasPartialRemoteAudio(bookID, rawURL) - if canStream { - handleStreamingReaderAudio(c, svc, book, bookID, rawURL, headers) + streamReaderAudioInto(c, svc, book, bookID, rawURL, headers) return } @@ -1117,33 +1200,285 @@ func handleCachedReaderAudio( serveCachedReaderAudio(c, f, info, rawURL) } -// handleStreamingReaderAudio 首播直通:一边落盘一边把字节写给客户端。 +// serveReaderAudioRange 区间直通:把客户端请求的那一段从上游边下边发。 +// +// persist 表示调用方已拿到写锁,可以顺手把字节补进分片(落盘后下次拖动/重播 +// 直接读本地)。返回 false 表示这次没法按区间处理(多区间、后缀区间、上游不认 +// Range 且总长未知等),调用方应退回整段路径。 +func serveReaderAudioRange( + c *gin.Context, + svc *service.Container, + book *model.ReaderBook, + bookID, rawURL string, + headers map[string]string, + rangeHeader string, + persist bool, +) bool { + start, end, ok := parseReaderAudioRange(rangeHeader) + if !ok { + return false + } + + // 分片里已经躺着 [0, partLen) 的字节:起点在分片内就先把这段发出去,再从 + // partLen 处回源续下。起点比分片还靠后说明中间有空洞,拼不上,只能从起点 + // 重新取,并且不落盘(否则分片会被写成带洞的文件)。 + var partFile *os.File + partLen := int64(0) + upstreamStart := start + if f, size, okPart := svc.Reader.PartialRemoteAudio(bookID, rawURL); okPart { + partFile = f + partLen = size + if partLen > start { + upstreamStart = partLen + } + } + // 收尾时关掉分片句柄:Windows 下「文件还开着就 rename」会被共享冲突挡住, + // 提升分片为正式缓存之前必须先关。 + defer func() { + if partFile != nil { + _ = partFile.Close() + } + }() + + out, err := fetchReaderMediaResilient(c.Request.Context(), svc, book, rawURL, + fmt.Sprintf("bytes=%d-", upstreamStart), headers, false) + if err != nil || out == nil || out.resp == nil { + if svc.Log != nil && err != nil { + svc.Log.Warn("reader 音频区间直通失败", zap.String("book", bookID), zap.Error(err)) + } + return false + } + resp := out.resp + + total := int64(-1) + switch resp.StatusCode { + case http.StatusPartialContent: + crStart, crTotal, crOK := reader.ParseContentRange(resp.Header.Get("Content-Range")) + if !crOK || crTotal <= 0 || crStart != upstreamStart { + drainReaderMediaBody(resp) + return false + } + total = crTotal + case http.StatusOK: + // 上游忽略了 Range,整段从头给:把前缀读掉丢弃,后面的字节与分片仍然 + // 对得上(前缀正好也是 [0, upstreamStart)),所以落盘可以照旧。 + if resp.ContentLength <= 0 { + drainReaderMediaBody(resp) + return false + } + if upstreamStart > 0 { + if _, copyErr := io.CopyN(io.Discard, resp.Body, upstreamStart); copyErr != nil { + drainReaderMediaBody(resp) + return false + } + } + total = resp.ContentLength + case http.StatusRequestedRangeNotSatisfiable: + // 分片比对象还长(对象换过):丢掉分片走整段路径重来。 + drainReaderMediaBody(resp) + return false + default: + drainReaderMediaBody(resp) + return false + } + if total <= 0 || start >= total { + drainReaderMediaBody(resp) + if total > 0 { + c.Header("Content-Range", fmt.Sprintf("bytes */%d", total)) + c.Status(http.StatusRequestedRangeNotSatisfiable) + return true + } + return false + } + + serveEnd := end + if serveEnd < 0 || serveEnd >= total { + serveEnd = total - 1 + } + wantLen := serveEnd - start + 1 + prefixLen := int64(0) + if partFile != nil && partLen > start { + prefixLen = partLen - start + if prefixLen > wantLen { + prefixLen = wantLen + } + } + + // 响应头必须一次写全:播放器据此决定后面要不要等更多字节。 + contentType := readerAudioContentType(rawURL) + if upstreamType := strings.TrimSpace(resp.Header.Get("Content-Type")); strings.HasPrefix(strings.ToLower(upstreamType), "audio/") { + contentType = upstreamType + } + writer := c.Writer + writer.Header().Set("Content-Type", contentType) + writer.Header().Set("Accept-Ranges", "bytes") + writer.Header().Set("Cache-Control", "private, max-age=604800") + writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, serveEnd, total)) + writer.Header().Set("Content-Length", strconv.FormatInt(wantLen, 10)) + writer.WriteHeader(http.StatusPartialContent) + writer.Flush() + + written := int64(0) + if prefixLen > 0 { + n, copyErr := io.CopyN(writer, io.NewSectionReader(partFile, start, prefixLen), prefixLen) + written += n + if copyErr != nil { + drainReaderMediaBody(resp) + if svc.Log != nil { + svc.Log.Warn("reader 音频区间下发失败", zap.String("book", bookID), zap.Error(copyErr)) + } + panic(http.ErrAbortHandler) + } + } + // 前缀发完就不需要读句柄了。必须先关掉:Windows 下文件还开着时 rename + // 会被共享冲突挡住,下面的「提升分片为正式缓存」就会静默失败。 + _ = partFile.Close() + partFile = nil + if written >= wantLen { + // 客户端要的区间完全落在本地分片里:这一轮不必再回源。 + drainReaderMediaBody(resp) + return true + } + + // 剩余字节边下边发;拿得到写锁时同时补进分片,落满后转成完整缓存。 + var partWriter *os.File + if persist && partLen >= start { + if f, size, appendErr := svc.Reader.AppendRemoteAudioPartial(bookID, rawURL); appendErr == nil { + if size == partLen { + partWriter = f + } else { + _ = f.Close() + } + } + } + sink := io.Writer(writer) + if partWriter != nil { + sink = io.MultiWriter(writer, partWriter) + } + copied, copyErr := io.CopyBuffer(sink, io.LimitReader(resp.Body, wantLen-written), make([]byte, 32<<10)) + written += copied + _ = resp.Body.Close() + if partWriter != nil { + _ = partWriter.Close() + if copyErr == nil { + // 分片正好落满整章时提升为正式缓存(长度对不上时内部会拒绝)。 + svc.Reader.PromoteRemoteAudioPartial(bookID, rawURL, total) + } + } + if copyErr == nil && written >= wantLen { + return true + } + if svc.Log != nil { + svc.Log.Warn("reader 音频区间中断", + zap.String("book", bookID), + zap.Int64("written", written), + zap.Int64("want", wantLen), + zap.Error(copyErr)) + } + // 已经发出去的字节收不回来:只能中止响应让播放器判「不完整」,分片留给下次续传。 + panic(http.ErrAbortHandler) +} + +// proxyReaderAudio 纯透传:不落盘、不排队,把上游字节原样转给播放器。 +// +// 用在「同一条音轨已经有别的请求在写分片」时:旧实现让第二个请求排队等前一个 +// 下载结束,而首播直通会一直持锁到客户端读完整章,播放器重试因此可能干等几十秒。 +func proxyReaderAudio( + c *gin.Context, + svc *service.Container, + book *model.ReaderBook, + bookID, rawURL string, + headers map[string]string, + rangeHeader string, +) { + out, err := fetchReaderMediaResilient(c.Request.Context(), svc, book, rawURL, + rangeHeader, headers, rangeHeader == "") + if err != nil || out == nil || out.resp == nil { + if svc.Log != nil && err != nil { + svc.Log.Warn("reader 音频透传失败", zap.String("book", bookID), zap.Error(err)) + } + c.String(http.StatusBadGateway, "音频拉取失败") + return + } + resp := out.resp + defer func() { _ = resp.Body.Close() }() + contentType := readerAudioContentType(rawURL) + if upstreamType := strings.TrimSpace(resp.Header.Get("Content-Type")); strings.HasPrefix(strings.ToLower(upstreamType), "audio/") { + contentType = upstreamType + } + writer := c.Writer + writer.Header().Set("Content-Type", contentType) + writer.Header().Set("Accept-Ranges", "bytes") + writer.Header().Set("Cache-Control", "private, max-age=604800") + if v := resp.Header.Get("Content-Range"); v != "" { + writer.Header().Set("Content-Range", v) + } + if resp.ContentLength > 0 { + writer.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10)) + } + writer.WriteHeader(resp.StatusCode) + if len(out.head) > 0 { + if _, writeErr := writer.Write(out.head); writeErr != nil { + panic(http.ErrAbortHandler) + } + } + streamReaderMediaBody(writer, resp.Body) +} + +// drainReaderMediaBody 丢掉剩余响应体并关闭连接(让上游把连接收回去)。 +func drainReaderMediaBody(resp *http.Response) { + if resp == nil || resp.Body == nil { + return + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) + _ = resp.Body.Close() +} + +// parseReaderAudioRange 解析客户端的单区间 Range 头("bytes=0-" / "bytes=100-199")。 +// +// 多区间与后缀区间("bytes=-500" 要总长才能换算)不在这里处理:播放器不会这么 +// 发,真遇到就退回磁盘路径,由 http.ServeContent 负责。 +func parseReaderAudioRange(header string) (start, end int64, ok bool) { + header = strings.TrimSpace(header) + const prefix = "bytes=" + if len(header) <= len(prefix) || !strings.EqualFold(header[:len(prefix)], prefix) { + return 0, 0, false + } + spec := strings.TrimSpace(header[len(prefix):]) + if spec == "" || strings.Contains(spec, ",") { + return 0, 0, false + } + dash := strings.IndexByte(spec, '-') + if dash <= 0 { + return 0, 0, false + } + start, err := strconv.ParseInt(strings.TrimSpace(spec[:dash]), 10, 64) + if err != nil || start < 0 { + return 0, 0, false + } + endPart := strings.TrimSpace(spec[dash+1:]) + if endPart == "" { + return start, -1, true + } + end, err = strconv.ParseInt(endPart, 10, 64) + if err != nil || end < start { + return 0, 0, false + } + return start, end, true +} + +// streamReaderAudioInto 首播直通:一边落盘一边把字节写给客户端(调用方已持写锁)。 // // 直接写 c.Writer 而不是先落盘再 ServeContent,是为了让第一个字节尽早到达。 // 下载失败时客户端已经收到部分字节,只能中止响应(见下面的 panic), // 不能改成 502 —— 状态码早就发出去了。 -func handleStreamingReaderAudio( +func streamReaderAudioInto( c *gin.Context, svc *service.Container, book *model.ReaderBook, bookID, rawURL string, headers map[string]string, ) { - key := svc.Reader.RemoteAudioCachePath(bookID, rawURL) - if key == "" { - c.String(http.StatusBadGateway, "音频缓存不可用") - return - } - unlock := readerAudioStreamLocks.Lock(key) - defer unlock() - - // 排队期间前一个请求可能已经把整章下完:直接下发本地文件,不重复回源。 - if f, info, ok := svc.Reader.OpenCachedRemoteAudio(bookID, rawURL); ok { - defer f.Close() - serveCachedReaderAudio(c, f, info, rawURL) - return - } - writer := c.Writer // started 表示响应头是否已发出。没发出时失败还能改成 502;已经发出就只能 // 中止连接——状态码收不回来了。 diff --git a/internal/service/reader/audio_media_cache.go b/internal/service/reader/audio_media_cache.go index c0c8d1e..617bb80 100644 --- a/internal/service/reader/audio_media_cache.go +++ b/internal/service/reader/audio_media_cache.go @@ -647,3 +647,84 @@ func (s *ReaderService) pruneAudioMediaCache() { } } } + +// ─── 区间直通:首播不必等整章下完 ─────────────────────────────────────────── +// +// 播放器(media3)取一条 m4a 时会开两次请求:先整段打开嗅探容器,再带 +// Range 从目标位置重开。过去「带 Range 的冷缓存请求」只能整章下完再按区间 +// 下发,于是每一章的起播都要等 1–4MB 的跨境回源(实测 1.5–35 秒)。下面这 +// 三个方法让 handler 能把客户端要的那一段边下边发,同时把字节补进分片。 + +// PartialRemoteAudio 打开未完成分片(只读),返回文件句柄与当前长度。 +// +// 播放器重开同一条音轨时,分片里已经躺着 [0, size) 的字节:区间直通先把这段 +// 发出去,再从 size 处回源续下,不必为了一个字节重下整章。 +func (s *ReaderService) PartialRemoteAudio(bookID, mediaURL string) (*os.File, int64, bool) { + path := s.RemoteAudioCachePath(bookID, mediaURL) + if path == "" { + return nil, 0, false + } + f, err := os.Open(path + audioPartialSuffix) + if err != nil { + return nil, 0, false + } + info, err := f.Stat() + if err != nil || !info.Mode().IsRegular() || info.Size() <= 0 || info.Size() > audioMediaObjectMaxBytes { + _ = f.Close() + return nil, 0, false + } + return f, info.Size(), true +} + +// AppendRemoteAudioPartial 以追加方式打开分片,返回打开那一刻的长度。 +// +// 调用方必须核对「返回长度 == 它算出的断点」:对不上说明分片已被别的写入者 +// 推进过,此时继续写会把字节落到错误位置,宁可放弃落盘只做纯透传。 +func (s *ReaderService) AppendRemoteAudioPartial(bookID, mediaURL string) (*os.File, int64, error) { + path := s.RemoteAudioCachePath(bookID, mediaURL) + if path == "" { + return nil, 0, errors.New("音频缓存目录未配置") + } + if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { + return nil, 0, fmt.Errorf("创建音频缓存目录失败: %w", err) + } + f, err := os.OpenFile(path+audioPartialSuffix, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o640) + if err != nil { + return nil, 0, err + } + info, err := f.Stat() + if err != nil { + _ = f.Close() + return nil, 0, err + } + return f, info.Size(), nil +} + +// PromoteRemoteAudioPartial 把已经写满的分片转成正式缓存文件。 +// +// 只认「分片长度 == 上游声明的总长」。区间直通是按播放器节奏下发的,中途断开 +// 时字节数必然不足;把这种分片当完整缓存下发,播放器会表现为「时长正确但后 +// 半段没声音」。 +func (s *ReaderService) PromoteRemoteAudioPartial(bookID, mediaURL string, total int64) bool { + path := s.RemoteAudioCachePath(bookID, mediaURL) + if path == "" || total <= 0 { + return false + } + partPath := path + audioPartialSuffix + info, err := os.Stat(partPath) + if err != nil || !info.Mode().IsRegular() || info.Size() != total { + return false + } + if err := os.Rename(partPath, path); err != nil { + return false + } + s.pruneAudioMediaCache() + return true +} + +// ParseContentRange 解析 "bytes 100-199/1234" 的起点与总长度。 +// +// 导出给 handler 用:区间直通要按上游声明的总长自己组装 206 响应头。 +func ParseContentRange(v string) (start, total int64, ok bool) { + return parseContentRange(v) +}