From f38076ce3416994b4a36c984633ca2760cb99daa Mon Sep 17 00:00:00 2001 From: DiheChen Date: Wed, 4 Nov 2026 21:18:08 +0800 Subject: [PATCH] fix: keep prepared media when a job runs again --- internal/app/prepare/media.go | 13 +++++- internal/app/prepare/media_test.go | 72 ++++++++++++++++++++++++++++++ internal/app/prepare/prepare.go | 2 +- 3 files changed, 85 insertions(+), 2 deletions(-) diff --git a/internal/app/prepare/media.go b/internal/app/prepare/media.go index b729d58..d02d7ec 100644 --- a/internal/app/prepare/media.go +++ b/internal/app/prepare/media.go @@ -32,10 +32,14 @@ type MediaRunner struct { } func NewMediaRunner(links port.LinkDownloader, bilibili port.BilibiliSource, transcoder port.Transcoder, store port.MediaStore, tracks port.Tracks, logger *slog.Logger) *MediaRunner { - return &MediaRunner{links: links, bilibili: bilibili, transcoder: transcoder, store: store, tracks: tracks, logger: logger} + return &MediaRunner{links: links, bilibili: bilibili, transcoder: transcoder, store: store, tracks: media.NewTracks(tracks, store), logger: logger} } func (r *MediaRunner) Run(ctx context.Context, t track.Track) (err error) { + // 任务可能在写入结果之后、River 记录完成之前中断，再次执行时结果仍然有效：往下走的话，clear 会删掉房间正在使用的音频。 + if done, err := r.prepared(ctx, t); done || err != nil { + return err + } d, release, err := r.download(ctx, t) if err != nil { return err @@ -80,6 +84,13 @@ func (r *MediaRunner) Run(ctx context.Context, t track.Track) (err error) { return r.tracks.SavePreparedTrack(ctx, t, port.PreparedTrack{Meta: meta, Result: result, MediaKey: key, MediaSize: size}) } +func (r *MediaRunner) prepared(ctx context.Context, t track.Track) (bool, error) { + ctx, cancel := context.WithTimeout(ctx, storeTimeout) + defer cancel() + _, ok, err := r.tracks.PreparedTrack(ctx, t) + return ok, err +} + // 任务参数来自 River 的任务表，不一定经过网关的校验。 func (r *MediaRunner) download(ctx context.Context, t track.Track) (port.LinkDownload, func(), error) { switch { diff --git a/internal/app/prepare/media_test.go b/internal/app/prepare/media_test.go index 8333a89..9267309 100644 --- a/internal/app/prepare/media_test.go +++ b/internal/app/prepare/media_test.go @@ -332,6 +332,78 @@ func TestLinkRunnerClearsStaleObjects(t *testing.T) { } } +func preparedLink(t *testing.T, store *porttest.MediaStore) (*tracksTable, port.PreparedTrack) { + t.Helper() + key := media.Key(link.ID) + if err := store.Put(t.Context(), media.AudioObject(key), strings.NewReader("old"), "audio/mpeg"); err != nil { + t.Fatal(err) + } + row := port.PreparedTrack{Meta: track.Meta{Title: "晴天"}, Result: track.Prepared{Duration: time.Second, Gain: 1}, MediaKey: key, MediaSize: 3} + tracks := newTracks() + tracks.rows[link] = row + return tracks, row +} + +func TestLinkRunnerKeepsMediaAlreadyPrepared(t *testing.T) { + store := porttest.NewMediaStore() + tracks, row := preparedLink(t, store) + d := &downloader{audio: porttest.SilentWAV(time.Second)} + if err := newLinkRunner(d, store, tracks, discard()).Run(t.Context(), link); err != nil { + t.Fatal(err) + } + if len(d.links) != 0 { + t.Errorf("下载了 %q，结果和音频都在时不用再下载", d.links) + } + if got, ok := store.Object(media.AudioObject(row.MediaKey)); !ok || string(got.Data) != "old" { + t.Errorf("音频有 %d 字节, %v，期望保留房间正在使用的音频", len(got.Data), ok) + } + if got, ok := tracks.row(link); !ok || got != row { + t.Errorf("曲目表中的结果 = %+v, %v，期望不变", got, ok) + } +} + +func TestLinkRunnerPreparesAgainWhenAudioIsGone(t *testing.T) { + store := porttest.NewMediaStore() + tracks, row := preparedLink(t, store) + if err := store.Delete(t.Context(), media.AudioObject(row.MediaKey)); err != nil { + t.Fatal(err) + } + audio := porttest.SilentWAV(2 * time.Second) + d := &downloader{meta: track.Meta{Title: "晴天"}, audio: audio} + if err := newLinkRunner(d, store, tracks, discard()).Run(t.Context(), link); err != nil { + t.Fatal(err) + } + if got, ok := store.Object(media.AudioObject(row.MediaKey)); !ok || string(got.Data) != string(audio) { + t.Errorf("音频 = %d 字节, %v，期望重新下载", len(got.Data), ok) + } + if got, ok := tracks.row(link); !ok || got.MediaSize != int64(len(audio)) { + t.Errorf("曲目表中的结果 = %+v, %v，期望写入这一次的结果", got, ok) + } +} + +func TestLinkRunnerKeepsMediaWhenPreparedResultCannotBeRead(t *testing.T) { + for name, fail := range map[string]func(*tracksTable, *porttest.MediaStore){ + "database": func(tracks *tracksTable, _ *porttest.MediaStore) { tracks.failures = 1 }, + "object_storage": func(_ *tracksTable, store *porttest.MediaStore) { store.Fail(errors.New("连接被拒绝")) }, + } { + t.Run(name, func(t *testing.T) { + store := porttest.NewMediaStore() + tracks, row := preparedLink(t, store) + fail(tracks, store) + d := &downloader{audio: porttest.SilentWAV(time.Second)} + if err := newLinkRunner(d, store, tracks, discard()).Run(t.Context(), link); err == nil { + t.Fatal("读取结果失败时任务要失败，由 River 重试") + } + if len(d.links) != 0 { + t.Errorf("下载了 %q，读取结果失败时不应继续", d.links) + } + if got, ok := store.Object(media.AudioObject(row.MediaKey)); !ok || string(got.Data) != "old" { + t.Errorf("音频有 %d 字节, %v，期望保留", len(got.Data), ok) + } + }) + } +} + func TestLinkRunnerRemovesUploadedObjectsOnFailure(t *testing.T) { key := media.Key(link.ID) cover := media.CoverObject(key, "jpg") diff --git a/internal/app/prepare/prepare.go b/internal/app/prepare/prepare.go index f2b5408..363a6ee 100644 --- a/internal/app/prepare/prepare.go +++ b/internal/app/prepare/prepare.go @@ -88,7 +88,7 @@ func (p *Preparer) prepared(ctx context.Context, t track.Track) (port.PreparedTr } // 返回之后要跟进的任务，提交失败时为 0。任务已经完成时返回 again，由调用方重新读取曲目表：结果可能是任务在刚才读取曲目表之后才写入的， -// 若直接重新提交，链接曲目的任务会先删掉已存储的音频再重新下载。 +// 这时不用再提交任务。 func (p *Preparer) follow(ctx context.Context, t track.Track, job int64) (int64, string, bool) { if job != 0 { state, reason, err := p.status(ctx, job)