fix(media): 清理旧 rendition 并确保失败状态落库
This commit is contained in:
1 parent
5d1a157085
commit
91b621fe75
9 files changed
+226
-35
No files matched your search
@@ -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`。
|
||||
@@ -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 与查询测试全部通过。
|
||||
|
||||
提交:`fix(media): 清理旧 rendition 并确保失败状态落库`(提交号见最终回执)。
|
||||
|
||||
+4
-10
@@ -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) {
|
||||
mediaThumbnailMapper.insert(thumbnail);
|
||||
return;
|
||||
}
|
||||
thumbnail.setId(existing.getId());
|
||||
mediaThumbnailMapper.updateById(thumbnail);
|
||||
thumbnail.setId(null);
|
||||
mediaThumbnailMapper.delete(Wrappers.<MediaThumbnail>lambdaQuery()
|
||||
.eq(MediaThumbnail::getEpisodeId, episodeId));
|
||||
mediaThumbnailMapper.insert(thumbnail);
|
||||
}
|
||||
}
|
||||
+6
-3
@@ -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());
|
||||
return message.substring(0, prefixLength) + suffix;
|
||||
int remain = Math.max(0, MAX_ERROR_CHARS - suffix.length());
|
||||
int prefixLength = remain / 2;
|
||||
int tailLength = remain - prefixLength;
|
||||
return message.substring(0, prefixLength)
|
||||
+ suffix + message.substring(message.length() - tailLength);
|
||||
}
|
||||
}
|
||||
+26
-2
@@ -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()
|
||||
.eq(MediaRendition::getMediaAssetId, task.getMediaAssetId()));
|
||||
var renditionDelete = Wrappers.<MediaRendition>lambdaQuery();
|
||||
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();
|
||||
|
||||
+16
-8
@@ -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;
|
||||
String status = transcodeService.fail(task.getId(), reason);
|
||||
if (TranscodeService.STATUS_FAILED.equals(status)
|
||||
&& task.getMediaAssetId() != null && task.getMediaAssetId() > 0) {
|
||||
mediaService.markFailed(task.getMediaAssetId(), reason);
|
||||
try {
|
||||
transcodeService.failFinal(task.getId(), reason);
|
||||
} catch (Exception persistError) {
|
||||
log.error("转码任务失败状态落库失败 task={} 原因={}",
|
||||
task.getId(), persistError.getMessage(), persistError);
|
||||
}
|
||||
log.error("转码失败 task={} 第 {}/{} 次尝试 耗时={}ms 结果={} 原因={}",
|
||||
task.getId(), attempt, TranscodeService.MAX_RETRY, System.currentTimeMillis() - startedAt,
|
||||
status, reason);
|
||||
if (task.getMediaAssetId() != null && task.getMediaAssetId() > 0) {
|
||||
try {
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+47
@@ -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"))
|
||||
|
||||
+30
@@ -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);
|
||||
|
||||
+27
-12
@@ -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"));
|
||||
// 还会重试,所以媒资不能置 failed(仍期待下一次成功)
|
||||
verify(mediaService, never()).markFailed(any(), anyString());
|
||||
verify(transcodeService).failFinal(eq(TASK_ID), contains("Invalid data found"));
|
||||
verify(mediaService).markFailed(eq(ASSET_ID), contains("Invalid data found"));
|
||||
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);
|
||||
runner.failure = new IllegalStateException("ffmpeg 未产出播放列表");
|
||||
doThrow(new IllegalStateException(
|
||||
"duplicate key uk_rendition"))
|
||||
.when(transcodeService).succeed(eq(TASK_ID), any(), any());
|
||||
|
||||
worker.processOne();
|
||||
|
||||
verify(transcodeService).fail(eq(TASK_ID), contains("未产出播放列表"));
|
||||
verify(mediaService).markFailed(eq(ASSET_ID), contains("未产出播放列表"));
|
||||
verify(transcodeService).failFinal(eq(TASK_ID), contains("uk_rendition"));
|
||||
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() {
|
||||
|
||||
Reference in new issue
Block a user