fix: keep prepared media when a job runs again

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 {

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")

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)