diff --git a/internal/handler/reader_media_test.go b/internal/handler/reader_media_test.go new file mode 100644 index 0000000..8fa8fc6 --- /dev/null +++ b/internal/handler/reader_media_test.go @@ -0,0 +1,445 @@ +package handler + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/gin-gonic/gin" + + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/service" + "github.com/truewhile/MeBox/internal/service/reader" +) + +// 媒体代理在「上游签名票被拒」时的行为契约。 +// +// 背景(真实故障):聆阅·懒人听书源的音频直链是第三方中转现签的腾讯 CDN 票, +// 票被 CDN 判掉时上游回的是**空的 401**(无 Content-Type、无 body)。旧行为把它 +// 原样透传给播放器,播放器重试的还是同一个 URL,于是 App 日志里连续 4~8 个 +// ERROR_CODE_IO_BAD_HTTP_STATUS (2004),表现为「这本书怎么都播不了」。 +// +// 现在的契约: +// - 能换票 → 用新票重拉,直接返回媒体本体(第一个请求就被治好); +// - 换不到 → 502 + 明确文案,而不是把 401 丢给播放器; +// - 非音轨地址(漫画图片防盗链等)→ 维持原样透传,不改变既有行为。 + +func TestReaderMediaPayloadRejected(t *testing.T) { + cases := []struct { + name string + head string + contentType string + want bool + }{ + {"HTML 错误页", "403", "text/html; charset=utf-8", true}, + {"JSON 报错", `{"code":403,"message":"expired"}`, "application/json", true}, + {"数组形态报错", `[{"error":"bad ticket"}]`, "application/json; charset=utf-8", true}, + {"纯文本报错", "ticket expired", "text/plain", false}, + {"正常 m4a", "\x00\x00\x00\x20ftypmp42", "audio/mp4", false}, + {"正常 mp3", "ID3\x04\x00", "audio/mpeg", false}, + {"正常图片", "\xff\xd8\xff\xe0", "image/jpeg", false}, + {"播放列表", "#EXTM3U\n#EXTINF:10,", "application/vnd.apple.mpegurl", false}, + {"空体", "", "text/html", false}, + {"HTML 类型但二进制体", "\x00\x00\x00\x20ftypmp42", "text/html", false}, + } + for _, c := range cases { + got := readerMediaPayloadRejected([]byte(c.head), c.contentType) + if got != c.want { + t.Errorf("%s: readerMediaPayloadRejected(%q, %q) = %v,期望 %v", + c.name, c.head, c.contentType, got, c.want) + } + } +} + +// TestReaderMediaProxyHealsStaleTicket 票被上游拒绝时代理自己换票重拉, +// 第一个请求就返回媒体本体(不必等播放器重试)。 +func TestReaderMediaProxyHealsStaleTicket(t *testing.T) { + svc, srv, book := prepareMediaTicketBook(t) + router := mediaRouterForTest(svc) + + signed := primeTicketBook(t, svc, book.ID) + // 让中转再签一张新票:章节缓存里那张就此变成「旧票」。 + srv.rotate() + + code, body := getReaderMedia(t, router, signed) + if code != http.StatusOK { + t.Fatalf("换票重试后 status=%d body=%s", code, body) + } + if !strings.Contains(body, "fake-audio-bytes") { + t.Fatalf("响应体不是媒体: %q", body) + } + if got := srv.contentHits(); got < 2 { + t.Fatalf("应当回源重取章节,实际请求 %d 次", got) + } +} + +// TestReaderMediaProxyFailsClearlyWhenRefreshImpossible 换不到票时给 502 + 明确文案, +// 而不是把 401 透传给播放器(那样播放器只会对着必然失败的地址反复重试)。 +func TestReaderMediaProxyFailsClearlyWhenRefreshImpossible(t *testing.T) { + svc, srv, book := prepareMediaTicketBook(t) + router := mediaRouterForTest(svc) + + signed := primeTicketBook(t, svc, book.ID) + srv.rotate() + srv.stopIssuing() + + code, body := getReaderMedia(t, router, signed) + if code != http.StatusBadGateway { + t.Fatalf("换票失败应当返回 502,实际 %d body=%s", code, body) + } + if !strings.Contains(body, "签名失效") { + t.Fatalf("错误文案应说明是音源签名失效: %s", body) + } +} + +// TestReaderMediaProxyPassesThroughNonTrackError 漫画图片这类不可换票的地址, +// 上游 4xx 仍按原状态透传(既有行为不变)。 +func TestReaderMediaProxyPassesThroughNonTrackError(t *testing.T) { + svc, srv, book := prepareMediaTicketBook(t) + router := mediaRouterForTest(svc) + + // 书源域名下的图片地址:不在任何音轨缓存里。 + imgURL := srv.URL + "/img/forbidden.jpg" + signed := svc.Reader.ProxyURL(book.ID, imgURL) + + code, body := getReaderMedia(t, router, signed) + if code == http.StatusBadGateway || code == http.StatusUnauthorized { + t.Fatalf("非音轨地址不该走换票分支,实际 %d body=%s", code, body) + } + if code != http.StatusForbidden { + t.Fatalf("图片 403 应原样透传,实际 %d body=%s", code, body) + } + if got := srv.contentHits(); got != 0 { + t.Fatalf("图片失败不该触发回源取章节,实际 %d 次", got) + } +} + +// TestReaderMediaProxyRejectsNonMediaPayload 上游用 200 回一段 HTML/JSON 错误页时, +// 代理直接判失败,不把网页喂给播放器(否则报「格式不支持」)。 +func TestReaderMediaProxyRejectsNonMediaPayload(t *testing.T) { + svc, srv, book := prepareMediaTicketBook(t) + router := mediaRouterForTest(svc) + + signed := primeTicketBook(t, svc, book.ID) + srv.serveErrorPage() + + code, body := getReaderMedia(t, router, signed) + if code != http.StatusBadGateway { + t.Fatalf("非媒体内容应当返回 502,实际 %d body=%.160s", code, body) + } + if !strings.Contains(body, "非媒体内容") { + t.Fatalf("错误文案应说明上游返回了非媒体内容: %s", body) + } +} + +// ─── 上游抖动:连接被 reset 时的重试与截断可见化 ───────────────────────── + +// TestReaderMediaProbeRetriesConnectionReset 上游在响应体中途 RST 连接时, +// 代理要退避重试,而不是第一次读失败就回 502。 +// +// 真实故障:腾讯 CDN 的跨境边缘在并发/长传输时会直接 RST,同一章重试一次 +// 往往就好。以前这里读 body 失败即判死,把一次可恢复的抖动变成「这本书播不了」。 +func TestReaderMediaProbeRetriesConnectionReset(t *testing.T) { + var attempts int32 + const bodyBytes = 200 << 10 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := atomic.AddInt32(&attempts, 1) + if n == 1 { + // 第一次:声明长度,写一小段后直接断开连接(模拟 RST/提前 EOF)。 + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Content-Length", strconv.Itoa(bodyBytes)) + _, _ = w.Write([]byte("\x00\x00\x00\x20ftypmp42")) + w.(http.Flusher).Flush() + if hj, ok := w.(http.Hijacker); ok { + if conn, _, err := hj.Hijack(); err == nil { + _ = conn.Close() + } + } + return + } + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Content-Length", strconv.Itoa(bodyBytes)) + payload := make([]byte, 8<<10) + for written := 0; written < bodyBytes; written += len(payload) { + if _, err := w.Write(payload); err != nil { + return + } + } + })) + 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("连接被 reset 后应当重试成功,实际 status=%d body=%.200s", code, body) + } + if int32(len(body)) != bodyBytes { + t.Fatalf("交付字节 = %d,期望 %d", len(body), bodyBytes) + } + if got := atomic.LoadInt32(&attempts); got < 2 { + t.Fatalf("应当重试过,实际上游请求 %d 次", got) + } +} + +// TestReaderMediaAbortsWhenUpstreamTruncates 上游中途断开且重试都失败时, +// 代理必须让客户端看到「传输不完整」,不能返回一个长度正确却内容缺失的响应。 +// +// 以前这里是 `_, _ = io.Copy(...)`,错误被丢掉,客户端拿到一个正常结束的 +// 半截文件 —— 播放器只能报「格式不支持」(3003),根因被彻底掩盖。 +func TestReaderMediaAbortsWhenUpstreamTruncates(t *testing.T) { + const declared = 400 << 10 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Content-Length", strconv.Itoa(declared)) + // 先写够探测用的前缀,再断开:探测能过,透传阶段才断。 + payload := make([]byte, 8<<10) + for written := 0; written < 128<<10; written += len(payload) { + if _, err := w.Write(payload); err != nil { + return + } + } + w.(http.Flusher).Flush() + 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") + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, signed, nil) + // panic(http.ErrAbortHandler) 由 net/http 捕获;httptest 下会向上冒泡, + // 这里断言它确实发生了(即「没有静默地把半截文件当成功」)。 + defer func() { + recovered := recover() + if recovered == nil { + t.Fatalf("上游截断时必须中止响应(panic(http.ErrAbortHandler)),实际正常返回了 %d 字节", + w.Body.Len()) + } + if recovered != http.ErrAbortHandler { + t.Fatalf("中止响应时应使用 http.ErrAbortHandler,实际 %v", recovered) + } + }() + router.ServeHTTP(w, req) + if w.Body.Len() >= declared { + t.Fatalf("不应交付完整长度的响应体(上游只写了 128KB)") + } +} + +func mediaRouterForTest(svc *service.Container) *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + // /api/reader/media 挂在公共 api 组(签名鉴权,不需要 JWT),单独注册即可。 + r.GET("/api/reader/media", readerMediaProxyHandler(svc)) + return r +} + +func getReaderMedia(t *testing.T, router *gin.Engine, signedPath string) (int, string) { + t.Helper() + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, signedPath, nil) + router.ServeHTTP(w, req) + return w.Code, w.Body.String() +} + +// primeTicketBook 先读一次章节(签出第一张票并落缓存),返回该音轨的签名代理路径。 +func primeTicketBook(t *testing.T, svc *service.Container, bookID string) string { + t.Helper() + out, err := svc.Reader.GetContentForBook(t.Context(), "u1", bookID, 0) + if err != nil { + t.Fatalf("首次取章节失败: %v", err) + } + if len(out.Tracks) != 1 { + t.Fatalf("音轨数 = %d,期望 1", len(out.Tracks)) + } + return out.Tracks[0] +} + +// ─── 模拟「中转现签 CDN 直链」的上游 ─────────────────────────────────────── + +// mediaTicketServer: +// - /book/1 详情页 +// - /content 章节正文,每次请求签一张新票 +// - /media 媒体本体,只认最新签出的票;旧票回**空的 401**(与腾讯 CDN 一致) +// - /img/* 图片资源,固定 403(模拟防盗链) +type mediaTicketServer struct { + *httptest.Server + mu sync.Mutex + issued []string + hits int + noToken bool + errorPage bool +} + +func newMediaTicketServer(t *testing.T) *mediaTicketServer { + t.Helper() + s := &mediaTicketServer{} + s.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": + s.mu.Lock() + s.hits++ + n := s.hits + noToken := s.noToken + s.mu.Unlock() + w.Header().Set("Content-Type", "application/json; charset=utf-8") + if noToken { + _, _ = w.Write([]byte(`{"data":{"url":""}}`)) + return + } + token := s.newTokenLocked(fmt.Sprintf("ticket-%d", n)) + _, _ = w.Write([]byte(fmt.Sprintf(`{"data":{"url":"%s/media?t=%s"}}`, s.URL, token))) + case r.URL.Path == "/media": + if s.errorPageNow() { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write([]byte(`ticket expired`)) + return + } + token := r.URL.Query().Get("t") + s.mu.Lock() + valid := len(s.issued) > 0 && s.issued[len(s.issued)-1] == token + s.mu.Unlock() + if !valid { + // 与上游一致:空 body、无 Content-Type。 + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "audio/mp4") + _, _ = w.Write([]byte("\x00\x00\x00\x20ftypmp42fake-audio-bytes")) + case strings.HasPrefix(r.URL.Path, "/img/"): + w.WriteHeader(http.StatusForbidden) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(s.Close) + return s +} + +// newTokenLocked 签一张新票(调用方需持有或随后获取锁;这里内部加锁)。 +func (s *mediaTicketServer) newTokenLocked(token string) string { + s.mu.Lock() + defer s.mu.Unlock() + s.issued = append(s.issued, token) + return token +} + +// rotate 模拟「中转又签了一张新票」:章节缓存里那张就此失效。 +func (s *mediaTicketServer) rotate() { + s.mu.Lock() + s.issued = append(s.issued, fmt.Sprintf("rotated-%d", len(s.issued))) + s.mu.Unlock() +} + +// stopIssuing 让书源此后给不出播放地址(模拟中转本身挂了)。 +func (s *mediaTicketServer) stopIssuing() { + s.mu.Lock() + s.noToken = true + s.mu.Unlock() +} + +// serveErrorPage 让媒体端点改用 200 + HTML 错误页(直链失效的另一种形态)。 +func (s *mediaTicketServer) serveErrorPage() { + s.mu.Lock() + s.errorPage = true + s.mu.Unlock() +} + +func (s *mediaTicketServer) errorPageNow() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.errorPage +} + +func (s *mediaTicketServer) contentHits() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.hits +} + +// prepareMediaTicketBookWithUpstream 用指定的上游地址建一本书(上游由调用方提供)。 +func prepareMediaTicketBookWithUpstream(t *testing.T, upstream string) (*service.Container, *mediaTicketServer, *model.ReaderBook) { + t.Helper() + svc := newReaderHandlerContainer(t) + svc.Cfg.Cache.CacheDir = t.TempDir() + // 占位 ticket server:只用来满足清理钩子,不参与请求。 + srv := newMediaTicketServer(t) + 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) + } + return svc, srv, book +} + +func prepareMediaTicketBook(t *testing.T) (*service.Container, *mediaTicketServer, *model.ReaderBook) { + t.Helper() + svc := newReaderHandlerContainer(t) + // 章节缓存必须落在磁盘上:换票要凭落盘载荷找回「这个地址属于哪一章」。 + // newReaderHandlerContainer 只给空 cfg,这里补一个临时缓存目录。 + svc.Cfg.Cache.CacheDir = t.TempDir() + srv := newMediaTicketServer(t) + srcJSON := fmt.Sprintf(`{ + "bookSourceName": "票源测试", + "bookSourceType": 1, + "bookSourceUrl": %q, + "ruleContent": {"content": "$.data.url"} +}`, srv.URL) + 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: srv.URL, + OriginName: "票源测试", + OriginType: 1, + BookURL: srv.URL + "/book/1", + }, "票源之书", "某作者", "") + if err != nil { + t.Fatalf("加入书架失败: %v", err) + } + if err := svc.Reader.SaveChapters(t.Context(), book.ID, []reader.ChapterInput{ + {Index: 0, Title: "第一章", URL: srv.URL + "/content"}, + }); err != nil { + t.Fatalf("保存目录失败: %v", err) + } + return svc, srv, book +} diff --git a/internal/handler/reader_routes.go b/internal/handler/reader_routes.go index baf3277..af16ce1 100644 --- a/internal/handler/reader_routes.go +++ b/internal/handler/reader_routes.go @@ -2,11 +2,16 @@ package handler import ( + "bytes" + "context" + "errors" + "fmt" "io" "net/http" "os" "strconv" "strings" + "time" "github.com/gin-gonic/gin" @@ -869,12 +874,18 @@ func readerMediaProxyHandler(svc *service.Container) gin.HandlerFunc { // 图片地址可自带 ",{headers:{...}}" 选项段:拆分后 headers 逐图应用, // 代理地址里只保留不带选项的地址(见 ProxyURL)。 rawURL, mediaOptions, _ := readerrule.ParseMediaOptions(rawURL) - resp, err := svc.Reader.FetchMediaWithOptions(c.Request.Context(), book, rawURL, c.GetHeader("Range"), mediaOptions.Headers) + // 只有 200 的 GET 需要嗅探响应体(判播放列表 / 判错误页); + // 206 的体是二进制分片、HEAD 没有体,读了反而要额外拼回去。 + probeBody := c.Request.Method == http.MethodGet + media, err := fetchReaderMedia(c.Request.Context(), svc, book, rawURL, c.GetHeader("Range"), mediaOptions.Headers, probeBody) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": "媒体拉取失败: " + err.Error()}) return } + resp := media.resp defer resp.Body.Close() + // 换过票时用新地址:m3u8 里的相对分片要按新地址解析。 + rawURL = media.url ct := resp.Header.Get("Content-Type") // 书源声明了 imageDecode 时,图片字节需要服务端二次解密。 @@ -885,6 +896,7 @@ func readerMediaProxyHandler(svc *service.Container) gin.HandlerFunc { c.JSON(http.StatusBadGateway, gin.H{"error": "图片过大或读取失败"}) return } + sniffed = append(append([]byte(nil), media.head...), sniffed...) decoded, decErr := svc.Reader.DecodeImageBytes(c.Request.Context(), book, rawURL, false, sniffed) if decErr != nil { c.JSON(http.StatusBadGateway, gin.H{"error": "图片解密失败: " + decErr.Error()}) @@ -897,20 +909,21 @@ func readerMediaProxyHandler(svc *service.Container) gin.HandlerFunc { c.Data(resp.StatusCode, contentType, decoded) return } - // m3u8 判定不能只看 Content-Type 或 URL 后缀:上游经常把播放列表标成 - // application/octet-stream,或者地址是 /index.m3u8?token=...(后缀判断不命中)。 - // 先读一小段用 #EXTM3U 标记确认;不是播放列表就把这段拼回响应体继续流式透传。 + // 200 的 GET:media.head 已经读过一小段,先用它判播放列表与错误页。 // - // 只嗅探 200 的 GET:206 的体是二进制分片、HEAD 没有体,都不需要判断。 - if c.Request.Method == http.MethodGet && resp.StatusCode == http.StatusOK { - head, readErr := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) - if readErr == nil && reader.BodyIsPlaylist(head, ct) { + // m3u8 判定不能只看 Content-Type 或 URL 后缀:上游经常把播放列表标成 + // application/octet-stream,或者地址是 /index.m3u8?token=...(后缀判断不命中), + // 所以以 #EXTM3U 标记为准。 + if probeBody && resp.StatusCode == http.StatusOK { + head := media.head + if reader.BodyIsPlaylist(head, ct) { rest, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) rewritten := svc.Reader.RewritePlaylist(bookID, rawURL, string(head)+string(rest)) c.Data(http.StatusOK, "application/vnd.apple.mpegurl", []byte(rewritten)) return } - // 不是播放列表:透传时 Content-Length 可能因为已经读过一部分而失配, + // 走到这里说明 head 不是错误页(fetchReaderMedia 已经把错误页换票重试过了), + // 原样拼回去继续流式透传。Content-Length 因为已读过一部分会失配, // 直接省略交给 net/http 按 chunked 处理,避免少写/多写导致截断。 for _, h := range []string{"Content-Type", "Content-Range", "Accept-Ranges"} { if v := resp.Header.Get(h); v != "" { @@ -921,7 +934,7 @@ func readerMediaProxyHandler(svc *service.Container) gin.HandlerFunc { if len(head) > 0 { _, _ = c.Writer.Write(head) } - _, _ = io.Copy(c.Writer, resp.Body) + streamReaderMediaBody(c.Writer, resp.Body) return } // 流式透传(含 206 Partial Content,支持音频拖动进度) @@ -931,13 +944,229 @@ func readerMediaProxyHandler(svc *service.Container) gin.HandlerFunc { } } c.Status(resp.StatusCode) - _, _ = io.Copy(c.Writer, resp.Body) + streamReaderMediaBody(c.Writer, resp.Body) + } +} + +// streamReaderMediaBody 把上游响应体写到客户端,并在上游中途断开时**主动掐断**连接。 +// +// 这是修复「播放器拿到半首歌」的关键:以前这里 `_, _ = io.Copy(...)` 把错误直接 +// 丢掉,上游(腾讯 CDN 边缘)在跨境链路上 RST 连接时,客户端收到的是一个正常 +// 结束、但字节数不足的响应 —— 播放器无从判断,只能报 +// ERROR_CODE_PARSING_CONTAINER_UNSUPPORTED(看起来像「格式不支持」,实际是被截断)。 +// +// 现在改成:上游没读完就中止这个响应,客户端因此拿到「传输不完整」而不是「正常结束」, +// 播放器可以按常规的重试/重连逻辑处理,而不是误判成格式问题。 +func streamReaderMediaBody(w io.Writer, body io.Reader) { + if _, err := io.Copy(w, body); err != nil { + // panic(http.ErrAbortHandler) 是 net/http 约定的「异常中止」:不再补写 + // chunked 结尾,直接关闭连接,客户端据此判定响应不完整。 + panic(http.ErrAbortHandler) } } // readerImageDecodeMaxBytes 需要服务端解密的单张图片上限。 const readerImageDecodeMaxBytes = 32 << 20 +// readerMediaProbeBytes 媒体代理嗅探上游响应体用的前缀长度。 +// +// 这个前缀同时用于三件事:判定 HLS 播放列表(#EXTM3U)、判定「其实是错误页」、 +// 以及把已读出的字节原样拼回响应体继续流式透传。 +const readerMediaProbeBytes = 64 << 10 + +// readerMediaResult 一次媒体拉取的结果。 +type readerMediaResult struct { + resp *http.Response + // url 是本次真正请求的媒体地址(换过票时是新票),m3u8 重写要用它做 base。 + url string + // head 是已从响应体读出的前缀(仅 200 的 GET 才有),调用方写响应时要先写它。 + head []byte + // rejected 表示 200 的响应体其实是错误页(HTML/JSON 报错),不能喂给播放器。 + rejected bool +} + +// fetchReaderMedia 拉取媒体,并在「签名票失效」时换票重试一次。 +// +// 听书源的播放地址是第三方中转现签的 CDN 直链(vkey/uid 带时效),票被上游判掉时 +// 有两种表现,这里都要兜住: +// +// ① 空的 401:没有 Content-Type、没有 body(腾讯 CDN 对无效 vkey/uid 的固定回答); +// ② 200 + HTML/JSON 错误页:中转自己出错时把它的报错页当成媒体回给我们。 +// +// 旧行为是两种都原样透传给播放器:① 变成连续多个 401(播放器重试的是同一个 +// 必然失败的地址),② 变成 ERROR_CODE_PARSING_CONTAINER_UNSUPPORTED(19 个 +// extractor 一个都不认)。两者的根因都是「这张票已经不能用了」,所以统一处理成 +// 回源换一张新票再拉: +// +// - 换票成功 → 用新地址重拉,新票同时写回章节缓存,后续请求自然有效; +// - 地址不属于可换票的音轨(漫画图片防盗链的 403 等)→ 原样透传上游状态, +// 不改变原有行为; +// - 该换票但换不到、或换票后依然不可用 → 返回错误,由调用方给出明确文案, +// 而不是让播放器对着必然失败的地址反复重试。 +func fetchReaderMedia( + ctx context.Context, + svc *service.Container, + book *model.ReaderBook, + rawURL, rangeHeader string, + options map[string]string, + probeBody bool, +) (*readerMediaResult, error) { + // 上游 CDN 在并发/跨境链路上偶发 reset 连接(实测同一章重试即可成功), + // 探测阶段读到一半断开不能直接判死,先按退避重试几次。 + out, refreshable, err := fetchReaderMediaWithRetry(ctx, svc, book, rawURL, rangeHeader, options, probeBody) + if err != nil || !refreshable { + return out, err + } + fresh, canRefresh, refreshErr := svc.Reader.RefreshMediaURL(ctx, book, rawURL) + if !canRefresh { + // 换不动(漫画图片防盗链的 403 等):原样透传上游状态,保持既有行为。 + // 但有两类失败例外——它们透传给播放器毫无意义(播放器只会一直重试 + // 同一个必然失败的地址,或者把网页当成音频),改成明确错误让前端提示。 + switch { + case out.rejected: + _ = out.resp.Body.Close() + return nil, fmt.Errorf("音源返回了非媒体内容(%s),请稍后重试或换源", + strings.TrimSpace(out.resp.Header.Get("Content-Type"))) + 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 out, nil + } + // 换得动就必须先释放这张失败响应的连接,再发新请求。 + _, _ = io.Copy(io.Discard, io.LimitReader(out.resp.Body, 1<<16)) + _ = out.resp.Body.Close() + if refreshErr != nil { + return nil, fmt.Errorf("音源签名失效且重新取票失败: %w", refreshErr) + } + retried, _, retryErr := fetchReaderMediaOnce(ctx, svc, book, fresh, rangeHeader, options, probeBody) + if retryErr != nil { + return nil, retryErr + } + if retried.rejected { + _ = retried.resp.Body.Close() + return nil, fmt.Errorf("音源换票后仍返回非媒体内容,请稍后重试或换源") + } + if reader.MediaNeedsRefresh(retried.resp.StatusCode) { + return nil, fmt.Errorf("音源换票后仍被拒绝(上游 %d),请稍后重试或换源", retried.resp.StatusCode) + } + return retried, nil +} + +// readerMediaProbeAttempts 探测阶段遇到连接中断(reset/EOF)的最大尝试次数。 +// +// 上游 CDN(腾讯 CDN 的跨境边缘)在并发或长传输时会直接 RST 连接,同一章 +// 重试一次往往就好。以前这里读 body 失败就直接回 502,把一次可恢复的抖动 +// 变成了「这本书播不了」。 +const readerMediaProbeAttempts = 3 + +// readerMediaProbeRetryDelay 探测重试的退避基数。 +const readerMediaProbeRetryDelay = 300 * time.Millisecond + +// readerMediaProbeError 探测阶段的可重试错误(连接被 reset / 提前 EOF)。 +type readerMediaProbeError struct{ err error } + +func (e *readerMediaProbeError) Error() string { return e.err.Error() } +func (e *readerMediaProbeError) Unwrap() error { return e.err } + +// fetchReaderMediaWithRetry 探测阶段按退避重试,只对「连接中断」类错误重试: +// 上游明确回了 4xx 时重试同一张票没有意义(那是换票流程的事)。 +func fetchReaderMediaWithRetry( + ctx context.Context, + svc *service.Container, + book *model.ReaderBook, + rawURL, rangeHeader string, + options map[string]string, + probeBody bool, +) (*readerMediaResult, bool, error) { + var lastErr error + for attempt := 1; attempt <= readerMediaProbeAttempts; attempt++ { + if attempt > 1 { + select { + case <-ctx.Done(): + return nil, false, ctx.Err() + case <-time.After(readerMediaProbeRetryDelay * time.Duration(attempt-1)): + } + } + out, refreshable, err := fetchReaderMediaOnce(ctx, svc, book, rawURL, rangeHeader, options, probeBody) + if err == nil { + return out, refreshable, nil + } + var probeErr *readerMediaProbeError + if !errors.As(err, &probeErr) { + return nil, false, err + } + lastErr = err + } + return nil, false, lastErr +} + +// fetchReaderMediaOnce 发一次媒体请求并做可用性判定。 +// +// 返回的 refreshable 表示「这次失败值得换一张签名票再试」: +// 上游 401/403/410,或 200 但响应体其实是错误页(详见 fetchReaderMedia 的说明)。 +// rejected 表示「200 但内容不是媒体」——调用方据此在无法换票时报明确错误, +// 而不是把网页喂给播放器。 +func fetchReaderMediaOnce( + ctx context.Context, + svc *service.Container, + book *model.ReaderBook, + rawURL, rangeHeader string, + options map[string]string, + probeBody bool, +) (*readerMediaResult, bool, error) { + resp, err := svc.Reader.FetchMediaWithOptions(ctx, book, rawURL, rangeHeader, options) + if err != nil { + return nil, false, err + } + if reader.MediaNeedsRefresh(resp.StatusCode) { + return &readerMediaResult{resp: resp, url: rawURL}, true, nil + } + if !probeBody || resp.StatusCode != http.StatusOK { + return &readerMediaResult{resp: resp, url: rawURL}, false, nil + } + // 先读一小段判断是不是错误页/播放列表;不是就原样拼回去继续流式透传。 + head, readErr := io.ReadAll(io.LimitReader(resp.Body, readerMediaProbeBytes)) + if readErr != nil { + _ = resp.Body.Close() + // 连接被 reset / 提前 EOF 属于上游抖动,交给上层退避重试。 + return nil, false, &readerMediaProbeError{err: fmt.Errorf("读取媒体响应失败: %w", readErr)} + } + rejected := readerMediaPayloadRejected(head, resp.Header.Get("Content-Type")) + return &readerMediaResult{resp: resp, url: rawURL, head: head, rejected: rejected}, rejected, nil +} + +// readerMediaPayloadRejected 判断上游 200 响应体其实是错误页而不是媒体。 +// +// 直链失效的另一个变种:CDN/中转不返回 4xx,而是回一段 HTML 错误页或 JSON +// 报错(Content-Type 是 text/html、application/json)。这类字节喂给播放器, +// ExoPlayer 的 19 个 extractor 全都不认,报出来的是 +// ERROR_CODE_PARSING_CONTAINER_UNSUPPORTED —— 看起来像「格式不支持」, +// 实际是「拿到了一张网页」。这里按 Content-Type + 首字符识别出来,直接判失败。 +func readerMediaPayloadRejected(head []byte, contentType string) bool { + if len(head) == 0 { + return false + } + lower := strings.ToLower(contentType) + errish := strings.Contains(lower, "html") || + strings.Contains(lower, "json") || + strings.HasPrefix(lower, "text/") + if !errish { + return false + } + trimmed := bytes.TrimSpace(head) + if len(trimmed) == 0 { + return false + } + switch trimmed[0] { + case '<', '{', '[': + return true + default: + return false + } +} + // isMediaPlaylistRequest 判断响应是否可能已是 HLS 播放列表。 // 播放列表太小、不该走图片解密分支(书源的 imageDecode 只会用于图片)。 func isMediaPlaylistRequest(resp *http.Response) bool { diff --git a/internal/helper/http.go b/internal/helper/http.go index be09ffb..cdc1461 100644 --- a/internal/helper/http.go +++ b/internal/helper/http.go @@ -7,6 +7,7 @@ import ( "compress/flate" "compress/gzip" "compress/zlib" + "crypto/tls" "io" "net/http" "net/url" @@ -88,10 +89,51 @@ func StripAcceptEncoding(h http.Header) { // // When useProxy is false, the client is created without proxy plumbing so // the request goes out direct, matching the user's checkbox intent. +// +// 注意 http.Client.Timeout 是「整个请求」的上限,包含读完响应体。书源的 +// 搜索/目录/正文都是小响应,30 秒足够;但同一个 client 也被用来流转发 +// 音频/视频(见 NewMediaHTTPClient),那里必须换一个没有整体超时的 client。 func NewSiteHTTPClient(timeoutSeconds int, useProxy bool) *http.Client { if timeoutSeconds <= 0 { timeoutSeconds = 15 } + return &http.Client{ + Timeout: time.Duration(timeoutSeconds) * time.Second, + Transport: newSiteTransport(useProxy), + } +} + +// NewMediaHTTPClient 拉取并转发媒体流(音频/视频/大图)的客户端。 +// +// 为什么不能用 NewSiteHTTPClient:http.Client.Timeout 会覆盖「读完 body」这个 +// 阶段。一个 4MB 的 m4a 在跨境链路上很容易超过 30 秒,一旦超时,net/http 会 +// 直接掐断响应体,而调用方的 io.Copy 只是正常返回 —— 表现为**静默截断** +// (播放器拿到半首歌,或判「格式不支持」),或者上游慢到 Do 本身报错 → 502。 +// +// 为什么强制 HTTP/1.1:实测腾讯 CDN(vb.stream.tencentmusic.com)的 h2 边缘在 +// 长连接大文件上不按预期回 WINDOW_UPDATE,Go 的 http2 Transport 初始连接窗口 +// 是 1MiB,于是下载**稳定卡在 1MiB/2MiB/3MiB 整数倍**处断开;同一 URL 用 +// HTTP/1.1(wget、curl --http1.1)从头到尾都完整。因此媒体通路固定走 h1.1: +// 音视频本来就没有「多路复用省请求」的收益,反而是流控 bug 的受害者。 +func NewMediaHTTPClient(useProxy bool) *http.Client { + tr := newSiteTransport(useProxy) + // 等响应头最多 90 秒:跨境 CDN 首包偶发要几十秒,但绝不能无上限挂着。 + tr.ResponseHeaderTimeout = 90 * time.Second + // 媒体是长连接大流量,连接池开大一些,避免并发取图/拖动进度时互相排队。 + tr.MaxIdleConns = 32 + tr.MaxIdleConnsPerHost = 16 + // 关闭 HTTP/2:见上面的流控说明。 + tr.ForceAttemptHTTP2 = false + tr.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{} + // 不复用连接:实测腾讯 CDN 的边缘在「同一条 keep-alive 连接上连续取多个 + // 大对象」时会直接 RST 第二个请求的响应体(容器内 wget/curl 每次新建连接 + // 就完全正常)。音视频请求本来就稀疏,为省一次 TLS 握手去冒截断风险不划算。 + tr.DisableKeepAlives = true + return &http.Client{Transport: tr} +} + +// newSiteTransport 书源请求共用的 Transport(媒体与规则请求配置一致的部分)。 +func newSiteTransport(useProxy bool) *http.Transport { tr := &http.Transport{ MaxIdleConns: 16, MaxIdleConnsPerHost: 4, @@ -104,10 +146,7 @@ func NewSiteHTTPClient(timeoutSeconds int, useProxy bool) *http.Client { return ProxyFromEnvironmentOrSystem(r) } } - return &http.Client{ - Timeout: time.Duration(timeoutSeconds) * time.Second, - Transport: tr, - } + return tr } // HTTPHeaderPresets returns a map of realistic browser HTTP headers. diff --git a/internal/middleware/gzip.go b/internal/middleware/gzip.go index c8b4f1e..34d6b14 100644 --- a/internal/middleware/gzip.go +++ b/internal/middleware/gzip.go @@ -17,12 +17,24 @@ var gzipExcludedExtensions = []string{ // apiGzipExcludedPrefixes 大文件流式传输(Range 语义)与 WS/SSE 长连接 // 不参与 gzip:压缩会破坏 Range / 逐块推送语义。 +// +// 阅读媒体代理(/api/reader/media、/api/reader/local/audio、 +// /api/reader/audio/transcode)同样必须排除,原因有两条: +// - 它们透传上游的 206 Partial Content。gin-contrib/gzip 在压缩时会删掉 +// Content-Length,却把 Content-Range 原样留下,客户端看到的是「声明 +// N 字节、实际收到 M 字节的 gzip 流」,自己发 Accept-Encoding 的播放器 +// (ExoPlayer 不透明处理)直接判 ERROR_CODE_PARSING_CONTAINER_UNSUPPORTED。 +// - 音视频本来就是压缩格式,再 gzip 一遍只烧 CPU(3.5MB 的 m4a 压完还是 +// 3.5MB 量级),对带宽没有收益。 var apiGzipExcludedPrefixes = []string{ "/api/stream/", "/api/hls/", "/api/img", "/api/subtitles/", "/api/strm/play/", + "/api/reader/media", + "/api/reader/local/audio", + "/api/reader/audio/transcode", "/api/ws", "/api/events", } diff --git a/internal/middleware/gzip_test.go b/internal/middleware/gzip_test.go index fca56dc..37ed2f4 100644 --- a/internal/middleware/gzip_test.go +++ b/internal/middleware/gzip_test.go @@ -70,6 +70,52 @@ func TestGzipAPIExcludesStreamPath(t *testing.T) { } } +// TestGzipAPIExcludesReaderMedia 阅读媒体代理必须排除在 gzip 之外。 +// +// 它会透传 CDN 的 206 Partial Content。gin-contrib/gzip 压缩时会删掉 +// Content-Length 却保留 Content-Range,客户端看到「声明 N 字节、实收 M 字节」, +// 自己发 Accept-Encoding 的播放器(ExoPlayer)直接判 +// ERROR_CODE_PARSING_CONTAINER_UNSUPPORTED,表现为「音频格式不支持」。 +func TestGzipAPIExcludesReaderMedia(t *testing.T) { + gin.SetMode(gin.TestMode) + router := gin.New() + api := router.Group("/api") + api.Use(GzipAPI()) + // 用 206 + Content-Range 模拟真实的音频分段响应。 + segment := func(c *gin.Context) { + c.Header("Content-Range", "bytes 0-99/100") + c.Data(http.StatusPartialContent, "audio/mp4", []byte(strings.Repeat("x", 4096))) + } + for _, path := range []string{ + "/reader/media", + "/reader/local/audio", + "/reader/audio/transcode", + } { + api.GET(path, segment) + } + + for _, path := range []string{ + "/reader/media", + "/reader/local/audio", + "/reader/audio/transcode", + } { + req := httptest.NewRequest(http.MethodGet, "/api"+path+"?b=book&u=x&s=y", nil) + req.Header.Set("Accept-Encoding", "gzip") + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if got := w.Header().Get("Content-Encoding"); got == "gzip" { + t.Errorf("%s 不该被 gzip(Range 语义会被破坏)", path) + } + if got := w.Header().Get("Content-Range"); got != "bytes 0-99/100" { + t.Errorf("%s 的 Content-Range 被破坏: %q", path, got) + } + if w.Body.Len() != 4096 { + t.Errorf("%s 响应体长度 = %d,期望 4096", path, w.Body.Len()) + } + } +} + func TestGzipStaticCompressesAssets(t *testing.T) { router := newGzipTestRouter() diff --git a/internal/service/reader/media_refresh.go b/internal/service/reader/media_refresh.go new file mode 100644 index 0000000..2caf453 --- /dev/null +++ b/internal/service/reader/media_refresh.go @@ -0,0 +1,190 @@ +package reader + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "sync" + "time" + + "github.com/truewhile/MeBox/internal/model" +) + +// 本文件处理「音频直链的签名票过期」这一类上游故障。 +// +// 背景:听书源(聆阅·懒人听书这类)的播放地址是第三方中转现签的 CDN 直链, +// 形如 https://vb.stream.tencentmusic.com/long/xxx.m4a?...&vkey=...&ttl=86400000。 +// 票的有效性不只取决于时间:中转背后是多个账号 token 轮换,某个 uid/vkey 组合 +// 会被 CDN 判掉,此时上游返回**空的 401**(无 Content-Type、无 body),而同一章 +// 隔几百毫秒重新签一张票就又能取到。 +// +// 旧行为:媒体代理把这个 401 原样透传给播放器,播放器重试的又是同一个 URL, +// 于是 App 日志里出现连续 4~8 个 401;同时音频清单还有 30 分钟缓存,这 30 分钟 +// 内 /content 一直下发同一张死票,表现为「这本书怎么都播不了,换本书才行」。 +// +// 现在的行为:代理遇到 401/403/410 时先把该章的缓存作废、重新走一次书源取正文, +// 拿到新票再拉一次;新票同时写回缓存,后续请求自然就是有效的。 + +// mediaRefreshTTL 新票的短缓存:播放器一次失败会立刻用同一 URL 重试多次, +// 这里让并发的重试共用一次回源,避免把书源打成限流。 +const mediaRefreshTTL = 2 * time.Minute + +// mediaRefreshEntry 一次成功换票的结果。 +type mediaRefreshEntry struct { + url string + exp time.Time +} + +// mediaRefreshState 保护「旧直链 → 新直链」的短缓存。 +type mediaRefreshState struct { + mu sync.Mutex + entries map[string]mediaRefreshEntry +} + +// lookupMediaRefresh 查换票短缓存(过期条目顺手清掉)。 +func (s *ReaderService) lookupMediaRefresh(staleURL string) (string, bool) { + s.mediaRefresh.mu.Lock() + defer s.mediaRefresh.mu.Unlock() + entry, ok := s.mediaRefresh.entries[staleURL] + if !ok { + return "", false + } + if time.Now().After(entry.exp) { + delete(s.mediaRefresh.entries, staleURL) + return "", false + } + return entry.url, true +} + +// rememberMediaRefresh 记住换票结果,并顺手清理过期条目(表很小,遍历即可)。 +func (s *ReaderService) rememberMediaRefresh(staleURL, freshURL string) { + s.mediaRefresh.mu.Lock() + defer s.mediaRefresh.mu.Unlock() + if s.mediaRefresh.entries == nil { + s.mediaRefresh.entries = map[string]mediaRefreshEntry{} + } + now := time.Now() + for k, v := range s.mediaRefresh.entries { + if now.After(v.exp) { + delete(s.mediaRefresh.entries, k) + } + } + s.mediaRefresh.entries[staleURL] = mediaRefreshEntry{url: freshURL, exp: now.Add(mediaRefreshTTL)} +} + +// RefreshMediaURL 在直链签名失效后重新换一张票。 +// +// staleURL 是书源侧产出的原始媒体地址(媒体代理已还原、未加签名)。返回三个值: +// +// refreshable=false 这个地址不是「可换票的音轨」(漫画图片、封面、或并非来自 +// 章节缓存的地址),调用方应原样透传上游状态,保持旧行为; +// refreshable=true 且 err!=nil 确实是音轨但换票失败,调用方应给出明确错误, +// 别让播放器对着必然失败的地址反复重试; +// refreshable=true 且 err==nil 拿到了新票。 +// +// 换票流程:在书架书籍的章节缓存里找到它所属的那一章 → 作废该章缓存(否则 +// /content 会继续下发同一张死票)→ 重新走书源取正文 → 按同一个音轨序号取新地址。 +func (s *ReaderService) RefreshMediaURL(ctx context.Context, book *model.ReaderBook, staleURL string) (string, bool, error) { + if s == nil || s.repo == nil || book == nil { + return "", false, nil + } + staleURL = strings.TrimSpace(staleURL) + if staleURL == "" { + return "", false, nil + } + if fresh, ok := s.lookupMediaRefresh(staleURL); ok { + return fresh, true, nil + } + sourceID := s.contentSourceID(ctx, nil, book) + if sourceID == "" { + return "", false, nil + } + bookKey := contentBookKey(sourceID, book.BookURL) + rows, err := s.repo.ListContentCacheByBook(ctx, bookKey) + if err != nil || len(rows) == 0 { + return "", false, nil + } + for i := range rows { + row := rows[i] + payload, ok := s.readContentPayload(&row) + if !ok || payload.ContentType != "audio" { + continue + } + trackIndex := indexOfString(payload.Tracks, staleURL) + if trackIndex < 0 { + continue + } + ch, ok := s.chapterForCacheRow(ctx, book, row) + if !ok { + return "", true, fmt.Errorf("章节已不在目录中") + } + // 先作废:即使后面重抓失败,也不会继续把死票发给播放器。 + s.dropContentCacheRow(ctx, &row) + out, fetchErr := s.fetchChapterContentShared(ctx, book, ch) + if fetchErr != nil { + return "", true, fmt.Errorf("重新取章节失败: %w", fetchErr) + } + if out == nil || trackIndex >= len(out.Tracks) { + return "", true, fmt.Errorf("新内容没有对应的音轨") + } + fresh := strings.TrimSpace(out.Tracks[trackIndex]) + if strings.HasPrefix(fresh, "/api/") { + // fetchChapterContentShared 返回缓存原样地址;兜底解析一次代理形态。 + return "", true, fmt.Errorf("书源返回了代理地址") + } + if fresh == "" || fresh == staleURL { + return "", true, fmt.Errorf("书源返回了相同地址") + } + s.rememberMediaRefresh(staleURL, fresh) + return fresh, true, nil + } + return "", false, nil +} + +// readContentPayload 读取一条章节缓存的落盘载荷。 +func (s *ReaderService) readContentPayload(row *model.ReaderContentCache) (cachedChapterContent, bool) { + var payload cachedChapterContent + path := s.contentPathFromRow(row) + if path == "" { + return payload, false + } + raw, err := os.ReadFile(path) // #nosec G304 -- 路径由服务端生成 + if err != nil { + return payload, false + } + if err := json.Unmarshal(raw, &payload); err != nil { + return payload, false + } + return payload, true +} + +// chapterForCacheRow 按缓存行的章节键找回目录里的章节(remap 后键仍一致)。 +func (s *ReaderService) chapterForCacheRow(ctx context.Context, book *model.ReaderBook, row model.ReaderContentCache) (model.ReaderChapter, bool) { + chapters, err := s.repo.ListChapters(ctx, book.ID) + if err != nil { + return model.ReaderChapter{}, false + } + for _, ch := range chapters { + if contentChapterKey(book, ch) == row.ChapterKey { + return ch, true + } + } + // 退一步按序号取:目录里同一章可能因为标题变化换了身份键。 + if row.ChapterIndex >= 0 && row.ChapterIndex < len(chapters) && !chapters[row.ChapterIndex].IsVolume { + return chapters[row.ChapterIndex], true + } + return model.ReaderChapter{}, false +} + +// indexOfString 返回目标字符串在列表中的下标,找不到返回 -1。 +func indexOfString(list []string, target string) int { + target = strings.TrimSpace(target) + for i, item := range list { + if strings.TrimSpace(item) == target { + return i + } + } + return -1 +} diff --git a/internal/service/reader/media_refresh_test.go b/internal/service/reader/media_refresh_test.go new file mode 100644 index 0000000..fe32039 --- /dev/null +++ b/internal/service/reader/media_refresh_test.go @@ -0,0 +1,263 @@ +package reader + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/truewhile/MeBox/internal/model" +) + +// 本文件:「音频直链签名票被 CDN 拒掉」后的换票回归测试。 +// +// 背景:听书源(聆阅·懒人听书)的播放地址是第三方中转现签的 CDN 直链,票会被 +// 上游判掉,此时 CDN 回的是**空的 401**(无 Content-Type、无 body)。旧行为把它 +// 原样透传给播放器,播放器重试的又是同一个 URL,于是连续 4~8 个 401, +// App 日志表现为 ERROR_CODE_IO_BAD_HTTP_STATUS (2004)「怎么都播不了」。 +// +// 现在的约定(RefreshMediaURL): +// - 能定位到归属章节 → 换票(先作废缓存,再重新走书源取正文); +// - 地址不是音轨(漫画图片、封面等)→ refreshable=false,调用方原样透传; +// - 是音轨但换不到 → refreshable=true + err,调用方给出明确错误不再让播放器重试。 + +// ticketServer 模拟「中转现签 CDN 直链」的上游: +// - /book/1 详情页(含目录链接) +// - /toc 目录 +// - /content 章节正文,每次请求签一张新票 +// - /media 媒体本体,只认最新签出的那张票,其余回空的 401 +type ticketServer struct { + *httptest.Server + mu sync.Mutex + issued []string + hits int + noToken bool // true 时正文里不带票(模拟书源不再给新地址) +} + +func newTicketServer() *ticketServer { + s := &ticketServer{} + s.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": + s.mu.Lock() + s.hits++ + n := s.hits + noToken := s.noToken + s.mu.Unlock() + w.Header().Set("Content-Type", "application/json; charset=utf-8") + if noToken { + _, _ = w.Write([]byte(`{"data":{"url":""}}`)) + return + } + token := fmt.Sprintf("ticket-%d", n) + s.mu.Lock() + s.issued = append(s.issued, token) + s.mu.Unlock() + _, _ = w.Write([]byte(fmt.Sprintf(`{"data":{"url":"%s/media?t=%s"}}`, s.URL, token))) + case r.URL.Path == "/media": + // 只认最近签出的那张票;旧票回空的 401(与腾讯 CDN 行为一致)。 + token := r.URL.Query().Get("t") + s.mu.Lock() + valid := len(s.issued) > 0 && s.issued[len(s.issued)-1] == token + s.mu.Unlock() + if !valid { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "audio/mp4") + _, _ = w.Write([]byte("\x00\x00\x00\x20ftypmp42fake-audio-bytes")) + default: + http.NotFound(w, r) + } + })) + return s +} + +func (s *ticketServer) contentHits() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.hits +} + +// ticketSourceJSON 一个最小听书源:正文规则取 JSON 里的播放直链。 +func ticketSourceJSON(t *testing.T, base string) string { + t.Helper() + return `{ + "bookSourceName": "票源测试", + "bookSourceType": 1, + "bookSourceUrl": "` + base + `", + "ruleContent": {"content": "$.data.url"} +}` +} + +// prepareTicketBook 建好「听书源 + 书架书籍 + 一章目录」,并开独立缓存目录。 +func prepareTicketBook(t *testing.T) (*ReaderService, *ticketServer, *model.ReaderBook) { + t.Helper() + srv := newTicketServer() + t.Cleanup(srv.Close) + + svc, _ := newLoginTestService(t) + svc.cfg.Cache.CacheDir = t.TempDir() + ctx := t.Context() + sourceID := prepareLoginSource(t, svc, ticketSourceJSON(t, srv.URL)) + book, err := svc.AddBook(ctx, readerTestUserID, SearchOrigin{ + SourceID: sourceID, + Origin: srv.URL, + OriginName: "票源测试", + OriginType: 1, + BookURL: srv.URL + "/book/1", + }, "票源之书", "某作者", "") + if err != nil { + t.Fatalf("加入书架失败: %v", err) + } + if err := svc.SaveChapters(ctx, book.ID, []ChapterInput{ + {Index: 0, Title: "第一章", URL: srv.URL + "/content"}, + }); err != nil { + t.Fatalf("保存目录失败: %v", err) + } + return svc, srv, book +} + +// TestRefreshMediaURLIssuesNewTicket 票被拒后换票:拿到新地址、缓存被刷新。 +func TestRefreshMediaURLIssuesNewTicket(t *testing.T) { + svc, srv, book := prepareTicketBook(t) + ctx := t.Context() + + // 首读:拿到第一张票并落缓存 + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + if err != nil { + t.Fatalf("首次读取失败: %v", err) + } + if len(out.Tracks) != 1 { + t.Fatalf("音轨数 = %d", len(out.Tracks)) + } + stale, err := svc.VerifyProxyURL(book.ID, trackQueryValue(t, out.Tracks[0], "u"), trackQueryValue(t, out.Tracks[0], "s")) + if err != nil { + t.Fatalf("还原直链失败: %v", err) + } + + fresh, refreshable, err := svc.RefreshMediaURL(ctx, book, stale) + if err != nil { + t.Fatalf("换票失败: %v", err) + } + if !refreshable { + t.Fatal("音轨地址应当可换票") + } + if fresh == stale { + t.Fatalf("换票后地址没变: %q", fresh) + } + if got := srv.contentHits(); got != 2 { + t.Fatalf("换票应回源重取一次章节,实际请求 %d 次", got) + } + + // 新票必须能取到媒体本体(旧票是空的 401) + resp, err := http.Get(fresh) + if err != nil { + t.Fatalf("新票请求失败: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("新票状态码 = %d,期望 200", resp.StatusCode) + } + + // 缓存要被刷新成新票:再次读章节不应再回源 + out2, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + if err != nil { + t.Fatalf("换票后读章节失败: %v", err) + } + refreshed, err := svc.VerifyProxyURL(book.ID, trackQueryValue(t, out2.Tracks[0], "u"), trackQueryValue(t, out2.Tracks[0], "s")) + if err != nil { + t.Fatalf("还原刷新后的直链失败: %v", err) + } + if refreshed != fresh { + t.Fatalf("章节缓存应刷新为新票 %q,实际 %q", fresh, refreshed) + } + if got := srv.contentHits(); got != 2 { + t.Fatalf("缓存已刷新,不应再回源,实际请求 %d 次", got) + } +} + +// TestRefreshMediaURLMemoryReusesTicket 同一张旧票的多次换票只回源一次: +// 播放器一次失败会立刻用同一 URL 重试多次,重复回源会把书源打成限流。 +func TestRefreshMediaURLMemoryReusesTicket(t *testing.T) { + svc, srv, book := prepareTicketBook(t) + ctx := t.Context() + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + if err != nil { + t.Fatalf("首次读取失败: %v", err) + } + stale, _ := svc.VerifyProxyURL(book.ID, trackQueryValue(t, out.Tracks[0], "u"), trackQueryValue(t, out.Tracks[0], "s")) + + first, _, err := svc.RefreshMediaURL(ctx, book, stale) + if err != nil { + t.Fatalf("首次换票失败: %v", err) + } + for i := 0; i < 3; i++ { + again, refreshable, err := svc.RefreshMediaURL(ctx, book, stale) + if err != nil || !refreshable { + t.Fatalf("第 %d 次换票失败: refreshable=%v err=%v", i+2, refreshable, err) + } + if again != first { + t.Fatalf("同一旧票应复用同一个新票:%q vs %q", again, first) + } + } + if got := srv.contentHits(); got != 2 { + t.Fatalf("重复换票只该回源一次,实际 %d 次", got) + } +} + +// TestRefreshMediaURLSkipsNonTrackURL 漫画图片/防盗链图片这类地址不属于音轨, +// 调用方应原样透传上游状态(不能被换票逻辑吃掉响应体)。 +func TestRefreshMediaURLSkipsNonTrackURL(t *testing.T) { + svc, _, book := prepareTicketBook(t) + ctx := t.Context() + + if _, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0); err != nil { + t.Fatalf("首次读取失败: %v", err) + } + _, refreshable, err := svc.RefreshMediaURL(ctx, book, "https://img.example.com/cover.jpg") + if err != nil { + t.Fatalf("非音轨地址不该报错: %v", err) + } + if refreshable { + t.Fatal("非音轨地址不该被判为可换票") + } +} + +// TestRefreshMediaURLErrorsWhenSourceRepeatsTicket 书源重新签出同一张票时, +// 说明中转本身没换票,应当明确失败而不是让播放器继续对着死票重试。 +func TestRefreshMediaURLErrorsWhenSourceRepeatsTicket(t *testing.T) { + svc, srv, book := prepareTicketBook(t) + ctx := t.Context() + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + if err != nil { + t.Fatalf("首次读取失败: %v", err) + } + stale, _ := svc.VerifyProxyURL(book.ID, trackQueryValue(t, out.Tracks[0], "u"), trackQueryValue(t, out.Tracks[0], "s")) + + // 让书源此后只回空地址:换票必然失败。 + srv.mu.Lock() + srv.noToken = true + srv.mu.Unlock() + + _, refreshable, err := svc.RefreshMediaURL(ctx, book, stale) + if !refreshable { + t.Fatal("音轨地址应当被判为可换票") + } + if err == nil { + t.Fatal("书源给不出新票时应当报错") + } + // 作废过的缓存不该把死票留在库里 + if _, ok := svc.loadCachedContent(ctx, nil, nil, book, model.ReaderChapter{ + Index: 0, Title: "第一章", URL: srv.URL + "/content", + }, "audio"); ok { + t.Fatal("换票失败后旧缓存应已被作废,不应继续下发死票") + } +} diff --git a/internal/service/reader/proxy_test.go b/internal/service/reader/proxy_test.go index bc1e892..f252f00 100644 --- a/internal/service/reader/proxy_test.go +++ b/internal/service/reader/proxy_test.go @@ -5,11 +5,13 @@ import ( "net/http" "net/http/httptest" "net/url" + "strconv" "sync/atomic" "testing" "time" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" ) @@ -132,3 +134,71 @@ func TestFetchMediaDoesNotRetryNotFound(t *testing.T) { t.Fatalf("上游请求次数 = %d,期望 1(404 不该重试)", got) } } + +// TestMediaHTTPClientHasNoOverallTimeout 媒体客户端不能用「整体超时」。 +// +// 这是本次故障的第二个根因:http.Client.Timeout 覆盖「读完 body」这一阶段。 +// 一个 4MB 的 m4a 在跨境链路上很容易超过 30 秒,超时后 net/http 直接掐断 +// 响应体,而调用方的 io.Copy 只是正常返回——表现为**静默截断**(播放器拿到 +// 半首歌,判「格式不支持」)或 502。所以媒体客户端必须没有整体超时。 +func TestMediaHTTPClientHasNoOverallTimeout(t *testing.T) { + client := helper.NewMediaHTTPClient(false) + if client.Timeout != 0 { + t.Fatalf("媒体 client 的整体 Timeout = %v,必须为 0(大文件下载会被静默截断)", client.Timeout) + } + tr, ok := client.Transport.(*http.Transport) + if !ok { + t.Fatalf("Transport 类型 = %T", client.Transport) + } + if tr.ResponseHeaderTimeout <= 0 { + t.Fatal("媒体 client 应设置 ResponseHeaderTimeout,避免上游不给首字节时无限挂起") + } + // 必须走 HTTP/1.1:腾讯 CDN 的 h2 边缘在长连接大文件上不回 WINDOW_UPDATE, + // Go 的 h2 初始连接窗口是 1MiB,下载会稳定断在 1MiB/2MiB/3MiB 整数倍处。 + if tr.ForceAttemptHTTP2 { + t.Error("媒体 client 必须关闭 HTTP/2(1MiB 窗口截断)") + } + if tr.TLSNextProto == nil || len(tr.TLSNextProto) != 0 { + t.Error("媒体 client 需要清空 TLSNextProto,才能确保不用 h2") + } +} + +// TestFetchMediaStreamsLargeBodyWithoutTruncation 大响应体必须完整送达。 +// +// 用「慢速分块发送」模拟跨境链路上的大音频:总耗时超过规则请求的 30 秒超时, +// 但媒体通路必须把每一个字节都交出来,绝不静默截断。 +func TestFetchMediaStreamsLargeBodyWithoutTruncation(t *testing.T) { + const totalBytes = 24 << 20 // 24MB + const chunk = 512 << 10 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "audio/mp4") + w.Header().Set("Content-Length", strconv.Itoa(totalBytes)) + payload := make([]byte, chunk) + for off := 0; off < totalBytes; off += chunk { + if _, err := w.Write(payload); err != nil { + return + } + w.(http.Flusher).Flush() + time.Sleep(5 * time.Millisecond) + } + })) + defer srv.Close() + + svc, _ := newLoginTestService(t) + book := &model.ReaderBook{Origin: srv.URL} + + resp, err := svc.FetchMedia(t.Context(), book, srv.URL+"/big.m4a", "") + if err != nil { + t.Fatalf("FetchMedia 失败: %v", err) + } + defer resp.Body.Close() + got, readErr := io.Copy(io.Discard, resp.Body) + if readErr != nil { + t.Fatalf("读取响应体失败: %v", readErr) + } + if got != totalBytes { + t.Fatalf("交付字节 = %d,期望 %d(少了 %d 字节 → 播放器会判格式不支持)", + got, totalBytes, int64(totalBytes)-got) + } +} diff --git a/internal/service/reader/rate_limit.go b/internal/service/reader/rate_limit.go index 5aca6b2..f0585f9 100644 --- a/internal/service/reader/rate_limit.go +++ b/internal/service/reader/rate_limit.go @@ -86,6 +86,9 @@ const ( // retryableStatus 该状态码是否值得重试:限流(403/429)与上游抖动(5xx)可重试, // 404/400 这类确定性错误不重试。规则请求与媒体代理共用。 +// +// 注意 401 不在这里:书源的规则请求返回 401 通常是「登录态失效」,反复重试 +// 只会白等退避时间。媒体直链另有语义,见 mediaRetryableStatus。 func retryableStatus(code int) bool { switch code { case 403, 408, 425, 429, 500, 502, 503, 504: @@ -95,6 +98,32 @@ func retryableStatus(code int) bool { } } +// mediaRetryableStatus 媒体拉取专用的可重试判定。 +// +// 与规则请求不同,CDN 直链的 410 表示「对象已下线」,与 403/429 一样属于 +// 上游抖动,同一张票稍后重试可能就恢复;而 401 是「签名票被拒」,重试同一张票 +// 一定还是 401,必须由调用方回源换票(见 MediaNeedsRefresh)。 +// 所以这里只补 410,不把 401 加进来。 +func mediaRetryableStatus(code int) bool { + if retryableStatus(code) { + return true + } + return code == 410 +} + +// MediaNeedsRefresh 判断媒体上游状态是否值得「换一张签名票再试」。 +// +// 401 是 CDN 对无效 vkey/uid 的固定回答(空 body、无 Content-Type),410 是 +// 对象已下线,403 常见于票与资源不匹配。这三类都值得回到书源重新签票。 +func MediaNeedsRefresh(code int) bool { + switch code { + case 401, 403, 410: + return true + default: + return false + } +} + // sleepWithContext 可被 ctx 取消的等待;返回 false 表示上下文已结束。 func sleepWithContext(ctx context.Context, d time.Duration) bool { if d <= 0 { diff --git a/internal/service/reader/rate_limit_test.go b/internal/service/reader/rate_limit_test.go index f697f8c..d2c1d8d 100644 --- a/internal/service/reader/rate_limit_test.go +++ b/internal/service/reader/rate_limit_test.go @@ -71,3 +71,36 @@ func TestRetryableStatus(t *testing.T) { } } } + +// TestMediaRetryableStatus 媒体直链的 410(对象已下线)与 403/429 一样值得原地重试, +// 但 401「签名票被拒」原地重试一定是同样的结果,只能靠回源换票。 +func TestMediaRetryableStatus(t *testing.T) { + for _, code := range []int{403, 410, 429, 500, 503} { + if !mediaRetryableStatus(code) { + t.Errorf("媒体状态码 %d 应当可重试", code) + } + } + for _, code := range []int{200, 206, 301, 400, 401, 404} { + if mediaRetryableStatus(code) { + t.Errorf("媒体状态码 %d 不该原地重试", code) + } + } + // 规则请求的判定不能被放宽:登录态失效的 401 反复重试只会白等。 + if retryableStatus(401) || retryableStatus(410) { + t.Error("retryableStatus 不应包含 401/410") + } +} + +// TestMediaNeedsRefresh 只有「票被拒」这一类状态才值得回源换票。 +func TestMediaNeedsRefresh(t *testing.T) { + for _, code := range []int{401, 403, 410} { + if !MediaNeedsRefresh(code) { + t.Errorf("状态码 %d 应当换票重试", code) + } + } + for _, code := range []int{200, 206, 304, 400, 404, 429, 500} { + if MediaNeedsRefresh(code) { + t.Errorf("状态码 %d 不该换票", code) + } + } +} diff --git a/internal/service/reader/reader.go b/internal/service/reader/reader.go index ba8767f..a8f15f3 100644 --- a/internal/service/reader/reader.go +++ b/internal/service/reader/reader.go @@ -43,7 +43,14 @@ type ReaderService struct { log *zap.Logger repo *repository.ReaderRepository http *http.Client - crypto *helper.SecretCipher + // mediaHTTP 用于转发媒体流(音频/视频/漫画大图)。 + // + // 必须与 http 分开:http 设了 30 秒的「整体」超时(含读完 body),而一个 + // 4MB 的音频在跨境链路上很容易超过它,超时后 net/http 直接掐断 body, + // 调用方的 io.Copy 却只是正常返回 —— 表现为静默截断(播放器拿到半首歌) + // 或 502。见 helper.NewMediaHTTPClient 的说明。 + mediaHTTP *http.Client + crypto *helper.SecretCipher // browserMu / browserPending 保护「待用户完成的页面」表 // (java.startBrowser / startBrowserAwait,见 browser_panel.go)。 @@ -71,6 +78,10 @@ type ReaderService struct { // 阅读页与批量预取会并发请求同一章(见 fetchChapterContentShared)。 contentFlightsMu sync.Mutex contentFlights map[string]*contentFlight + + // mediaRefresh 记录「已失效直链 → 重新签到的新直链」的短缓存: + // CDN 判掉某张签名票后,播放器的多次重试共用一次回源(见 media_refresh.go)。 + mediaRefresh mediaRefreshState } // NewReaderService 创建服务。 @@ -83,12 +94,21 @@ func NewReaderService(cfg *config.Config, log *zap.Logger, repos *repository.Con log: log, repo: repos.Reader, http: client, + mediaHTTP: helper.NewMediaHTTPClient(true), crypto: helper.NewSecretCipher(firstNonEmpty(cfg.Secrets.EncryptionKey, cfg.Secrets.JWTSecret)), limiter: newSourceRateLimiter(), coverDecodeOrigins: map[string]bool{}, } } +// mediaClient 返回转发媒体流用的客户端(未注入时退回规则请求的 client)。 +func (s *ReaderService) mediaClient() *http.Client { + if s != nil && s.mediaHTTP != nil { + return s.mediaHTTP + } + return s.http +} + // markSourceCoverDecode 记录某书源是否声明了 coverDecodeJs(导入/更新书源时调用)。 func (s *ReaderService) markSourceCoverDecode(sourceURL string, enabled bool) { if s == nil || strings.TrimSpace(sourceURL) == "" { @@ -1942,12 +1962,14 @@ func (s *ReaderService) FetchMediaWithOptions(ctx context.Context, book *model.R if err != nil { return nil, err } - resp, err := s.http.Do(httpReq) + // 媒体用无整体超时的 client:4MB 音频在跨境链路上常超过 30 秒, + // 用带整体超时的 client 会在读完 body 前被掐断(静默截断)。 + resp, err := s.mediaClient().Do(httpReq) if err != nil { lastErr = err continue } - if attempt == mediaFetchAttempts || !retryableStatus(resp.StatusCode) { + if attempt == mediaFetchAttempts || !mediaRetryableStatus(resp.StatusCode) { return resp, nil } // 可重试的状态码:读完并关闭响应体再试,避免连接泄漏。 @@ -2511,7 +2533,7 @@ func (s *ReaderService) fetchSourceResource(ctx context.Context, sourceURL, rawU httpReq.Header.Set("Referer", referer) } } - return s.http.Do(httpReq) + return s.mediaClient().Do(httpReq) } // ─── 书架 ──────────────────────────────────────────────────────────────────