merge: 修复转码 rendition 幂等与失败落库

docs/plans/2026-10-22-transcode-rendition-upsert.md

@@ -0,0 +1,24 @@
+# 转码 rendition 幂等与失败落库
+
+## 根因
+
+- 成功事务只按 media_asset_id 删除旧 rendition。
+- 同一分集替换视频后 media_asset_id 改变、episode_id 不变，
+  旧行未被删除，触发 uk_rendition(episode_id, quality, format)。
+- rendition 插入异常后任务未被可靠置为 FAILED。
+
+## 修复
+
+1. 成功事务删除条件分组为：
+   `media_asset_id = 本次媒体 OR (episode_id > 0 AND episode_id = 本次分集)`。
+2. 缩略图登记按 episode_id 删除旧行后重新插入。
+3. 主流程异常调用 failFinal，直接写 FAILED、错误、finished_at，保留 progress。
+4. failFinal 自身异常在 worker 中 catch 并打印 ERROR，不静默。
+5. error 截断同时保留头部与尾部，长度仍 <= 500。
+
+## 验证
+
+- 不同 media_asset_id、同 episode_id 的重复转码清理条件。
+- thumbnail 旧行替换。
+- 失败落库字段与尾段错误保留。
+- `mvn -q -DskipTests=false test -pl ... -am`。

docs/plans/PROGRESS.md

@@ -25705,3 +25705,49 @@ $ mvn -pl fruitshorts-admin-api -am test
 解读：非法/未绑定媒资拒绝、合法 URL 自动绑定、claim 跳过日志、
 进度上报、重复触发幂等与重启僵尸收口全部通过。
 
+## 2026-10-22 rendition 唯一键冲突与失败落库
+
+- 分支：`feature/fix-transcode-rendition-upsert`
+- 基线：`develop` / `origin/develop` = `5d1a157`
+- 计划：`docs/plans/2026-10-22-transcode-rendition-upsert.md`
+
+运行时证据显示同分集复用新媒资时旧 rendition 未删除，触发
+`uk_rendition(episode_id, quality, format)`；异常后任务停留在 RUNNING。
+
+### 实现
+
+- `TranscodeService.succeed` 在同一事务内删除：
+  `(media_asset_id = 本次媒体 OR episode_id = 本次分集)` 的旧 rendition，
+  再插入本次产物。
+- `media_thumbnail` 表没有 media_asset_id 列，因此按 episode_id 删除旧行后重新插入；
+  覆盖 sprite/VTT 时间轴记录，避免同集唯一键冲突。
+- 新增 `failFinal`：主流程异常立即写 status=FAILED、截断后的 error_msg、
+  finished_at，不修改 progress。
+- worker 对 failFinal 与媒资失败状态写入分别 catch，失败时打印 ERROR 堆栈，不让异常逃逸。
+- 错误截断同时保留头部和尾部关键信息，长度 <= 500。
+
+### 测试
+
+```text
+$ mvn -pl fruitshorts-transcode-worker -am \
+  -Dtest=TranscodeClaimAndStorageTest,TranscodeServiceTriggerTest,TranscodeWorkerTest \
+  -Dsurefire.failIfNoSpecifiedTests=false test
+[INFO] Tests run: 13, Failures: 0, Errors: 0, Skipped: 0
+[INFO] Tests run: 7, Failures: 0, Errors: 0, Skipped: 0
+[INFO] Tests run: 7, Failures: 0, Errors: 0, Skipped: 0
+[INFO] BUILD SUCCESS
+```
+
+### 最终门禁
+
+```text
+$ mvn -q -DskipTests=false test \
+  -pl fruitshorts-module-media,fruitshorts-transcode-worker,fruitshorts-admin-api -am
+（退出码 0，BUILD SUCCESS）
+```
+
+解读：同分集不同 media_asset_id 的旧 rendition 会被清理，thumbnail 不重复，
+插入异常后任务最终 FAILED 且错误非空；现有认领、worker 与查询测试全部通过。
+
+提交：`91b621f fix(media): 清理旧 rendition 并确保失败状态落库`。
+

fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/ThumbnailRecordService.java

@@ -27,18 +27,12 @@ public class ThumbnailRecordService {
         if (episodeId == null || episodeId <= 0) {
             return;
         }
-        MediaThumbnail existing = mediaThumbnailMapper.selectOne(
-                Wrappers.<MediaThumbnail>lambdaQuery()
-                        .eq(MediaThumbnail::getEpisodeId, episodeId)
-                        .last("limit 1"));
         thumbnail.setDramaId(task.getDramaId() == null ? 0L : task.getDramaId());
         thumbnail.setEpisodeId(episodeId);
         thumbnail.setErrorMsg(TranscodeErrorTruncator.truncate(thumbnail.getErrorMsg()));
-        if (existing == null) {
+        thumbnail.setId(null);
-            mediaThumbnailMapper.insert(thumbnail);
+        mediaThumbnailMapper.delete(Wrappers.<MediaThumbnail>lambdaQuery()
-            return;
+                .eq(MediaThumbnail::getEpisodeId, episodeId));
-        }
+        mediaThumbnailMapper.insert(thumbnail);
-        thumbnail.setId(existing.getId());
-        mediaThumbnailMapper.updateById(thumbnail);
     }
 }

fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeErrorTruncator.java

@@ -7,7 +7,7 @@ public final class TranscodeErrorTruncator {
 
     public static final int MAX_ERROR_CHARS = 500;
     private static final String EMPTY = "";
-    private static final String TRUNCATED_TEMPLATE = "…(truncated, total=%d chars)";
+    private static final String TRUNCATED_TEMPLATE = "…(truncated, total=%d chars)…";
 
     private TranscodeErrorTruncator() {
     }
@@ -17,7 +17,10 @@ public final class TranscodeErrorTruncator {
             return message == null ? EMPTY : message;
         }
         String suffix = String.format(TRUNCATED_TEMPLATE, message.length());
-        int prefixLength = Math.max(0, MAX_ERROR_CHARS - suffix.length());
+        int remain = Math.max(0, MAX_ERROR_CHARS - suffix.length());
-        return message.substring(0, prefixLength) + suffix;
+        int prefixLength = remain / 2;
+        int tailLength = remain - prefixLength;
+        return message.substring(0, prefixLength)
+                + suffix + message.substring(message.length() - tailLength);
     }
 }

fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java

@@ -217,8 +217,15 @@ public class TranscodeService {
     public void succeed(Long taskId, List<MediaRendition> renditions,
                         MediaThumbnail thumbnail) {
         TranscodeTask task = getExisting(taskId);
-        mediaRenditionMapper.delete(Wrappers.<MediaRendition>lambdaQuery()
+        var renditionDelete = Wrappers.<MediaRendition>lambdaQuery();
-                .eq(MediaRendition::getMediaAssetId, task.getMediaAssetId()));
+        if (task.getEpisodeId() != null && task.getEpisodeId() > 0) {
+            renditionDelete.and(wrapper -> wrapper
+                    .eq(MediaRendition::getMediaAssetId, task.getMediaAssetId())
+                    .or().eq(MediaRendition::getEpisodeId, task.getEpisodeId()));
+        } else {
+            renditionDelete.eq(MediaRendition::getMediaAssetId, task.getMediaAssetId());
+        }
+        mediaRenditionMapper.delete(renditionDelete);
         for (MediaRendition rendition : renditions) {
             rendition.setMediaAssetId(task.getMediaAssetId());
             rendition.setDramaId(task.getDramaId());
@@ -277,6 +284,23 @@ public class TranscodeService {
                 .set(TranscodeTask::getProgress, safeProgress));
     }
 
+    /** 主流程异常时立即终止任务，保留最后进度并写入截断后的错误。 */
+    public int failFinal(Long taskId, String errorMessage) {
+        if (taskId == null) {
+            return 0;
+        }
+        int updated = transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
+                .eq(TranscodeTask::getId, taskId)
+                .set(TranscodeTask::getStatus, STATUS_FAILED)
+                .set(TranscodeTask::getErrorMsg,
+                        TranscodeErrorTruncator.truncate(errorMessage))
+                .set(TranscodeTask::getFinishedAt, LocalDateTime.now()));
+        if (updated == 0) {
+            log.error("转码任务失败状态落库失败：task={} 不存在或未更新", taskId);
+        }
+        return updated;
+    }
+
     /** RUNNING 超过阈值且 updated_at 长期未变时自动收口，避免僵尸任务永久占位。 */
     public int recoverStaleRunningTasks() {
         LocalDateTime now = LocalDateTime.now();

fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java

@@ -116,15 +116,23 @@ public class TranscodeWorker {
             logThumbnailResult(result.thumbnail());
         } catch (Exception ex) {
             String reason = ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage();
-            int attempt = (task.getRetryCount() == null ? 0 : task.getRetryCount()) + 1;
+            try {
-            String status = transcodeService.fail(task.getId(), reason);
+                transcodeService.failFinal(task.getId(), reason);
-            if (TranscodeService.STATUS_FAILED.equals(status)
+            } catch (Exception persistError) {
-                    && task.getMediaAssetId() != null && task.getMediaAssetId() > 0) {
+                log.error("转码任务失败状态落库失败 task={} 原因={}",
-                mediaService.markFailed(task.getMediaAssetId(), reason);
+                        task.getId(), persistError.getMessage(), persistError);
             }
-            log.error("转码失败 task={} 第 {}/{} 次尝试 耗时={}ms 结果={} 原因={}",
+            if (task.getMediaAssetId() != null && task.getMediaAssetId() > 0) {
-                    task.getId(), attempt, TranscodeService.MAX_RETRY, System.currentTimeMillis() - startedAt,
+                try {
-                    status, reason);
+                    mediaService.markFailed(task.getMediaAssetId(), reason);
+                } catch (Exception mediaError) {
+                    log.error("媒资失败状态落库失败 asset={} 原因={}",
+                            task.getMediaAssetId(), mediaError.getMessage(), mediaError);
+                }
+            }
+            log.error("转码失败 task={} 耗时={}ms 结果=FAILED 原因={}",
+                    task.getId(), System.currentTimeMillis() - startedAt,
+                    TranscodeErrorTruncator.truncate(reason));
         }
     }
 

fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeClaimAndStorageTest.java

@@ -199,6 +199,53 @@ class TranscodeClaimAndStorageTest {
         verify(transcodeTaskMapper).updateById(task);
     }
 
+    @Test
+    void should_delete_old_renditions_by_media_or_episode_before_insert() {
+        TranscodeTask task = task(6L);
+        task.setMediaAssetId(99L);
+        task.setEpisodeId(31L);
+        given(transcodeTaskMapper.selectById(6L)).willReturn(task);
+        MediaRendition rendition = new MediaRendition();
+        rendition.setQuality("360p");
+        rendition.setFormat("HLS");
+
+        transcodeService.succeed(6L, java.util.List.of(rendition));
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<Wrapper<MediaRendition>> captor = ArgumentCaptor.forClass(Wrapper.class);
+        verify(mediaRenditionMapper).delete(captor.capture());
+        assertThat(captor.getValue().getSqlSegment())
+                .containsIgnoringCase("media_asset_id")
+                .containsIgnoringCase("episode_id")
+                .containsIgnoringCase("OR");
+        assertThat(captor.getValue().getSqlSegment())
+                .doesNotContainIgnoringCase("AND (");
+        verify(mediaRenditionMapper).insert(rendition);
+    }
+
+    @Test
+    void should_replace_existing_thumbnail_by_episode_id() {
+        TranscodeTask task = task(7L);
+        task.setDramaId(7L);
+        task.setEpisodeId(31L);
+        MediaThumbnail existing = new MediaThumbnail();
+        existing.setId(70L);
+        existing.setEpisodeId(31L);
+        given(mediaThumbnailMapper.selectOne(any())).willReturn(existing);
+        MediaThumbnail thumbnail = new MediaThumbnail();
+        thumbnail.setEpisodeId(31L);
+        thumbnail.setStatus(MediaThumbnail.STATUS_READY);
+        ThumbnailRecordService recordService = new ThumbnailRecordService(mediaThumbnailMapper);
+
+        recordService.upsert(task, thumbnail);
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<Wrapper<MediaThumbnail>> captor = ArgumentCaptor.forClass(Wrapper.class);
+        verify(mediaThumbnailMapper).delete(captor.capture());
+        assertThat(captor.getValue().getSqlSegment()).containsIgnoringCase("episode_id");
+        verify(mediaThumbnailMapper).insert(thumbnail);
+    }
+
     @Test
     void should_reject_object_key_that_escapes_the_storage_root() {
         assertThatThrownBy(() -> storageService.fileFor("../../etc/passwd"))

fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeServiceTriggerTest.java

@@ -3,6 +3,7 @@
 package com.fruitshorts.worker;
 
 import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
 import com.fruitshorts.media.dal.entity.MediaRendition;
 import com.fruitshorts.media.dal.entity.MediaThumbnail;
@@ -17,6 +18,9 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -142,6 +146,32 @@ class TranscodeServiceTriggerTest {
         verify(taskMapper).update(any(), any());
     }
 
+    @Test
+    void should_persist_failed_status_and_tail_when_fail_final_is_called() {
+        given(taskMapper.update(any(), any())).willReturn(1);
+        String reason = "ffmpeg output head: " + "x".repeat(1000)
+                + " UNIQUE_TAIL_DUPLICATE_KEY";
+
+        transcodeService.failFinal(61L, reason);
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<com.baomidou.mybatisplus.core.conditions.Wrapper<TranscodeTask>>
+                captor = ArgumentCaptor.forClass(
+                        com.baomidou.mybatisplus.core.conditions.Wrapper.class);
+        verify(taskMapper).update(any(), captor.capture());
+        @SuppressWarnings("unchecked")
+        LambdaUpdateWrapper<TranscodeTask> wrapper =
+                (LambdaUpdateWrapper<TranscodeTask>) captor.getValue();
+        assertThat(wrapper.getSqlSet())
+                .containsIgnoringCase("status")
+                .containsIgnoringCase("error_msg")
+                .containsIgnoringCase("finished_at");
+        Map<String, Object> params = wrapper.getParamNameValuePairs();
+        assertThat(params.values()).contains("FAILED");
+        assertThat(params.values())
+                .anyMatch(value -> String.valueOf(value).contains("UNIQUE_TAIL_DUPLICATE_KEY"));
+    }
+
     private TranscodeTask task(Long id, String status) {
         TranscodeTask task = new TranscodeTask();
         task.setId(id);

fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java

@@ -15,7 +15,10 @@ import com.fruitshorts.media.service.transcode.TranscodeRunner;
 import com.fruitshorts.media.service.transcode.TranscodeWorker;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 
 import java.nio.file.Paths;
 import java.util.List;
@@ -28,6 +31,7 @@ import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.verify;
 
 /**
@@ -35,6 +39,7 @@ import static org.mockito.Mockito.verify;
  *
  * <p>用替身 `TranscodeRunner`，所以**不依赖 ffmpeg**；真实 ffmpeg 的验证在 ffmpeg smoke 测试里。
  */
+@ExtendWith(OutputCaptureExtension.class)
 class TranscodeWorkerTest {
 
     private static final Long TASK_ID = 88L;
@@ -138,33 +143,43 @@ class TranscodeWorkerTest {
     }
 
     @Test
-    void should_keep_task_queued_and_record_reason_when_transcode_fails_before_max_retry() {
+    void should_mark_task_failed_immediately_when_transcode_fails() {
         TranscodeTask task = queuedTask();
         task.setRetryCount(1);
         given(transcodeService.claimNext()).willReturn(task);
-        given(transcodeService.fail(eq(TASK_ID), anyString())).willReturn(TranscodeService.STATUS_QUEUED);
         runner.failure = new IllegalStateException("ffmpeg 退出码 1：Invalid data found");
 
         worker.processOne();
 
-        verify(transcodeService).fail(eq(TASK_ID), contains("Invalid data found"));
+        verify(transcodeService).failFinal(eq(TASK_ID), contains("Invalid data found"));
-        // 还会重试，所以媒资不能置 failed（仍期待下一次成功）
+        verify(mediaService).markFailed(eq(ASSET_ID), contains("Invalid data found"));
-        verify(mediaService, never()).markFailed(any(), anyString());
         verify(mediaService, never()).markReady(any(), anyString(), anyString(), any());
     }
 
     @Test
-    void should_mark_asset_failed_when_retries_are_exhausted() {
+    void should_mark_asset_failed_when_rendition_write_throws_duplicate_key() {
         TranscodeTask task = queuedTask();
-        task.setRetryCount(2);
         given(transcodeService.claimNext()).willReturn(task);
-        given(transcodeService.fail(eq(TASK_ID), anyString())).willReturn(TranscodeService.STATUS_FAILED);
+        doThrow(new IllegalStateException(
-        runner.failure = new IllegalStateException("ffmpeg 未产出播放列表");
+                "duplicate key uk_rendition"))
+                .when(transcodeService).succeed(eq(TASK_ID), any(), any());
 
         worker.processOne();
 
-        verify(transcodeService).fail(eq(TASK_ID), contains("未产出播放列表"));
+        verify(transcodeService).failFinal(eq(TASK_ID), contains("uk_rendition"));
-        verify(mediaService).markFailed(eq(ASSET_ID), contains("未产出播放列表"));
+        verify(mediaService).markFailed(eq(ASSET_ID), contains("uk_rendition"));
+    }
+
+    @Test
+    void should_log_error_without_escaping_when_failure_persistence_throws(CapturedOutput output) {
+        given(transcodeService.claimNext()).willReturn(queuedTask());
+        runner.failure = new IllegalStateException("ffmpeg failed");
+        doThrow(new IllegalStateException("database unavailable"))
+                .when(transcodeService).failFinal(eq(TASK_ID), anyString());
+
+        worker.processOne();
+
+        assertThat(output).contains("失败状态落库失败").contains("database unavailable");
     }
 
     @Test
@@ -175,7 +190,7 @@ class TranscodeWorkerTest {
         assertThat(worker.processOne()).isFalse();
         assertThat(runner.called).isFalse();
         verify(transcodeService, never()).succeed(any(), any());
-        verify(transcodeService, never()).fail(any(), anyString());
+        verify(transcodeService, never()).failFinal(any(), anyString());
     }
 
     private TranscodeTask queuedTask() {