test(media): 覆盖转码入队与重转幂等

This commit is contained in:
Zhou Weidong committed 2026-10-22 14:01:26 +08:00
1 parent cd51aabe7e
commit 7e6075a2e9
5 files changed
+435

No files matched your search

@@ -0,0 +1,37 @@
# 后台转码链路打通
## 现状取证
- 上传完成时 `MediaService.complete` 已为视频调用 `TranscodeService.enqueue`。
- 分集创建/替换视频地址时 `EpisodeService` 已调用
`EpisodeTranscodeBindingService.bind`,会复用并绑定同一媒体的未完成任务。
- 转码失败原因已写入 `transcode_task.error_msg`,最终状态写入 `status`。
- 缺口是后台手动触发/重转接口,以及对应幂等和链路测试。
## 实现
1. 保留上传完成和改片自动入队逻辑,补测试锁定行为。
2. `TranscodeService` 增加手动触发:
- QUEUED/RUNNING 直接返回现有任务;
- FAILED 重置原任务为 QUEUED;
- SUCCESS 才创建新任务,紧接着重复触发会复用新 QUEUED 任务。
3. 新增 `POST /admin/media/{id}/transcode`,校验媒资存在,复用 `episode:upload`。
4. 返回任务 id 与状态,便于后台提示。
## TDD 测试矩阵
| 场景 | 断言 |
| --- | --- |
| 视频上传完成 | `MediaService.complete` 调用一次 `enqueue(assetId, 0, 0, sourceUrl)` |
| 图片上传完成 | 不创建转码任务 |
| 手动触发成功媒资 | 新建一个 QUEUED 任务,重复调用复用同一任务 |
| 手动触发进行中媒资 | 直接返回 QUEUED/RUNNING 任务,不新增记录 |
| 手动触发失败媒资 | 原任务重置为 QUEUED,保留任务 id |
| 媒资不存在 | 返回既有 `DATA_NOT_FOUND`,控制器不产生 500 |
| worker 最终失败 | 任务状态 FAILED,`error_msg` 保留 ffmpeg 原因 |
## 验证
- 先提交失败测试,再提交实现。
- `mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test`
- 不启停任何服务,不执行数据库变更,不改 Web/Flutter/specs/API 文档。
+51
View File
@@ -25205,3 +25205,54 @@ dramaTitle/episodeNumber/totalEpisodes 非空、越界页空 list 且 total 保
`limit` 旧 List 兼容均通过;服务层断言 `selectPage`、稳定排序和两批 IN 查询。
未启停服务、未操作真机、未修改客户端。
## 2026-10-22 后台转码链路打通 P1
- 分支:`feature/media-transcode-pipeline`
- 基线:`develop` / `origin/develop` = `cd51aabe`
- 隔离工作树:`/private/tmp/fruitserver-media-pipeline`
- 计划:`docs/plans/2026-10-22-media-transcode-pipeline.md`
### 现状取证
- `/admin/media/upload/complete`:视频完成合并后会调用
`transcodeService.enqueue(assetId, 0L, 0L, sourceUrl)`;图片直接置 ready,不入队。
- 分集创建/更新并保存 `videoUrl`:`EpisodeService` 调用
`EpisodeTranscodeBindingService.bind`,按 objectKey 找到 `media_asset`,
再把 `dramaId/episodeId/sourceUrl` 绑定到任务;未完成任务复用,不重复入队。
- worker:`claimNext` 只认领 `episode_id > 0`;成功后回填媒资 HLS、分集播放地址与时长;
失败超过 3 次写 `FAILED + error_msg`。这些既有链路已由新增测试锁定。
- 实际缺口:后台没有手动触发/重转接口,也没有其幂等测试。
### TDD 红测
先只提交测试,不写生产实现。定向命令:
```text
$ mvn -pl fruitshorts-transcode-worker -am \
-Dtest=MediaServiceTranscodePipelineTest,TranscodeServiceTriggerTest \
-Dsurefire.failIfNoSpecifiedTests=false test
[ERROR] ... MediaServiceTranscodePipelineTest.java:[13,32] 找不到符号
符号: 类 TranscodeTriggerRespVO
[ERROR] ... TranscodeServiceTriggerTest.java:[15,32] 找不到符号
符号: 类 TranscodeTriggerRespVO
[INFO] fruitshorts-transcode-worker ....................... FAILURE
[INFO] BUILD FAILURE
```
解读:失败原因为预期的生产接口尚不存在,不是测试环境或既有代码编译问题。
下一步再实现最小生产代码令红测转绿。
控制器契约红测同样先失败在缺失的响应类型:
```text
$ mvn -pl fruitshorts-admin-api -am \
-Dtest=MediaTranscodeTriggerContractTest \
-Dsurefire.failIfNoSpecifiedTests=false test
[ERROR] MediaTranscodeTriggerContractTest.java:[11,32] 找不到符号
符号: 类 TranscodeTriggerRespVO
[INFO] fruitshorts-admin-api .............................. FAILURE
[INFO] BUILD FAILURE
```
解读:控制器测试已经先于实现落地,现有上传/分集链路未被破坏。
@@ -0,0 +1,63 @@
// Author: fruit.com.cn
package com.fruitshorts.admin.controller;
import com.fruitshorts.admin.support.ControllerContractSupport;
import com.fruitshorts.common.exception.ErrorCode;
import com.fruitshorts.common.exception.ServiceException;
import com.fruitshorts.media.service.MediaService;
import com.fruitshorts.media.service.MediaTranscodeProfileService;
import com.fruitshorts.media.service.TranscodeService;
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.test.web.servlet.MockMvc;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
/**
* 后台手动触发转码的最小契约。
*/
class MediaTranscodeTriggerContractTest {
private MediaService mediaService;
private MockMvc mockMvc;
@BeforeEach
void setUp() {
mediaService = mock(MediaService.class);
MediaController controller = new MediaController(
mediaService, mock(MediaTranscodeProfileService.class));
mockMvc = ControllerContractSupport.mockMvcOf(controller);
}
@Test
void should_return_task_id_and_status_when_triggering_transcode() throws Exception {
given(mediaService.triggerTranscode(88L))
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
mockMvc.perform(post("/admin/media/88/transcode"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(0))
.andExpect(jsonPath("$.data.taskId").value("100"))
.andExpect(jsonPath("$.data.status").value(TranscodeService.STATUS_QUEUED));
verify(mediaService).triggerTranscode(88L);
}
@Test
void should_keep_not_found_semantics_when_media_does_not_exist() throws Exception {
given(mediaService.triggerTranscode(404L))
.willThrow(new ServiceException(ErrorCode.DATA_NOT_FOUND, "媒资不存在"));
mockMvc.perform(post("/admin/media/404/transcode"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(ErrorCode.DATA_NOT_FOUND.getCode()))
.andExpect(jsonPath("$.message").value("媒资不存在"));
}
}
@@ -0,0 +1,146 @@
// Author: fruit.com.cn
package com.fruitshorts.worker;
import com.fruitshorts.media.config.MediaProperties;
import com.fruitshorts.media.dal.entity.MediaAsset;
import com.fruitshorts.media.dal.entity.MediaUploadTask;
import com.fruitshorts.media.dal.mapper.MediaAssetMapper;
import com.fruitshorts.media.dal.mapper.MediaUploadTaskMapper;
import com.fruitshorts.media.service.MediaService;
import com.fruitshorts.media.service.MediaStorageService;
import com.fruitshorts.media.service.TranscodeService;
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
import com.fruitshorts.media.vo.UploadCompleteReqVO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.nio.file.Path;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
/**
* 媒资完成上传与后台手动重转的入队语义。
*/
class MediaServiceTranscodePipelineTest {
private static final Long ASSET_ID = 88L;
private static final String SOURCE_URL = "/files/episode/202610/video.mp4";
private MediaAssetMapper mediaAssetMapper;
private MediaUploadTaskMapper uploadTaskMapper;
private MediaStorageService storageService;
private TranscodeService transcodeService;
private MediaService mediaService;
@BeforeEach
void setUp() {
mediaAssetMapper = mock(MediaAssetMapper.class);
uploadTaskMapper = mock(MediaUploadTaskMapper.class);
storageService = mock(MediaStorageService.class);
transcodeService = mock(TranscodeService.class);
mediaService = new MediaService(
new MediaProperties(), storageService, mediaAssetMapper,
uploadTaskMapper, transcodeService);
}
@Test
void should_enqueue_video_transcode_when_upload_completes() {
givenCompletedUpload("video.mp4");
given(storageService.countChunks("upload-1")).willReturn(1);
given(storageService.merge("upload-1", "episode/202610/video.mp4", 1))
.willReturn(new MediaStorageService.MergeResult(
Path.of("/tmp/video.mp4"), 1024L, "md5"));
given(storageService.toPublicUrl("episode/202610/video.mp4"))
.willReturn(SOURCE_URL);
given(mediaAssetMapper.selectById(ASSET_ID)).willReturn(asset("video.mp4"));
mediaService.complete(completeReq(), 10L);
verify(transcodeService).enqueue(ASSET_ID, 0L, 0L, SOURCE_URL);
}
@Test
void should_not_enqueue_image_when_upload_completes() {
givenCompletedUpload("cover.jpg");
given(storageService.countChunks("upload-1")).willReturn(1);
given(storageService.merge("upload-1", "episode/202610/cover.jpg", 1))
.willReturn(new MediaStorageService.MergeResult(
Path.of("/tmp/cover.jpg"), 512L, "md5"));
given(storageService.toPublicUrl("episode/202610/cover.jpg"))
.willReturn("/files/episode/202610/cover.jpg");
given(mediaAssetMapper.selectById(ASSET_ID)).willReturn(asset("cover.jpg"));
mediaService.complete(completeReq(), 10L);
verify(transcodeService, never()).enqueue(any(), any(), any(), anyString());
}
@Test
void should_return_not_found_when_triggering_transcode_for_missing_media() {
given(mediaAssetMapper.selectOne(any())).willReturn(null);
assertThatThrownBy(() -> mediaService.triggerTranscode(404L))
.isInstanceOf(com.fruitshorts.common.exception.ServiceException.class)
.hasMessage("媒资不存在");
verify(transcodeService, never()).trigger(any(), anyString());
}
@Test
void should_trigger_transcode_and_mark_asset_queued() {
MediaAsset asset = asset("video.mp4");
given(mediaAssetMapper.selectOne(any())).willReturn(asset);
given(transcodeService.trigger(ASSET_ID, SOURCE_URL))
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
TranscodeTriggerRespVO result = mediaService.triggerTranscode(ASSET_ID);
assertThat(result.getTaskId()).isEqualTo(100L);
assertThat(result.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
assertThat(asset.getStatus()).isEqualTo("transcoding");
assertThat(asset.getTranscodeStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(mediaAssetMapper).updateById(asset);
}
private void givenCompletedUpload(String fileName) {
String objectKey = "episode/202610/" + fileName;
MediaUploadTask task = new MediaUploadTask();
task.setUploadId("upload-1");
task.setFileName(fileName);
task.setObjectKey(objectKey);
task.setTotalChunks(1);
task.setStatus("uploading");
task.setCreateBy(10L);
given(uploadTaskMapper.selectOne(any())).willReturn(task);
doAnswer(invocation -> {
MediaAsset inserted = invocation.getArgument(0);
inserted.setId(ASSET_ID);
return 1;
}).when(mediaAssetMapper).insert(any(MediaAsset.class));
}
private UploadCompleteReqVO completeReq() {
UploadCompleteReqVO req = new UploadCompleteReqVO();
req.setUploadId("upload-1");
req.setDurationSec(120);
return req;
}
private MediaAsset asset(String fileName) {
MediaAsset asset = new MediaAsset();
asset.setId(ASSET_ID);
asset.setFileName(fileName);
asset.setObjectKey("episode/202610/" + fileName);
asset.setSourceUrl(SOURCE_URL);
asset.setStatus("uploaded");
return asset;
}
}
@@ -0,0 +1,138 @@
// Author: fruit.com.cn
package com.fruitshorts.worker;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fruitshorts.media.dal.entity.MediaRendition;
import com.fruitshorts.media.dal.entity.MediaThumbnail;
import com.fruitshorts.media.dal.entity.TranscodeTask;
import com.fruitshorts.media.dal.mapper.MediaRenditionMapper;
import com.fruitshorts.media.dal.mapper.MediaThumbnailMapper;
import com.fruitshorts.media.dal.mapper.TranscodeTaskMapper;
import com.fruitshorts.media.service.ThumbnailRecordService;
import com.fruitshorts.media.service.TranscodeService;
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
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 static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
/**
* 后台重转的幂等规则:进行中复用、失败重置、成功才新建。
*/
class TranscodeServiceTriggerTest {
private static final Long ASSET_ID = 88L;
private TranscodeTaskMapper taskMapper;
private TranscodeService transcodeService;
@BeforeAll
static void initTableInfoCache() {
MapperBuilderAssistant assistant =
new MapperBuilderAssistant(new MybatisConfiguration(), "");
TableInfoHelper.initTableInfo(assistant, TranscodeTask.class);
TableInfoHelper.initTableInfo(assistant, MediaRendition.class);
TableInfoHelper.initTableInfo(assistant, MediaThumbnail.class);
}
@BeforeEach
void setUp() {
taskMapper = mock(TranscodeTaskMapper.class);
transcodeService = new TranscodeService(
taskMapper, mock(MediaRenditionMapper.class),
mock(MediaThumbnailMapper.class), mock(ThumbnailRecordService.class));
}
@Test
void should_create_once_and_reuse_queued_task_when_retrying_successful_asset() {
TranscodeTask success = task(10L, TranscodeService.STATUS_SUCCESS);
TranscodeTask queued = task(100L, TranscodeService.STATUS_QUEUED);
given(taskMapper.selectOne(any())).willReturn(success, null, queued);
doAnswer(invocation -> {
TranscodeTask inserted = invocation.getArgument(0);
inserted.setId(100L);
return 1;
}).when(taskMapper).insert(any(TranscodeTask.class));
TranscodeTriggerRespVO first = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
TranscodeTriggerRespVO second = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
assertThat(first.getTaskId()).isEqualTo(100L);
assertThat(first.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
assertThat(second.getTaskId()).isEqualTo(100L);
assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(taskMapper).insert(any(TranscodeTask.class));
}
@Test
void should_reuse_queued_task_for_repeated_trigger() {
TranscodeTask queued = task(21L, TranscodeService.STATUS_QUEUED);
given(taskMapper.selectOne(any())).willReturn(queued);
TranscodeTriggerRespVO first = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
TranscodeTriggerRespVO second = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
assertThat(first.getTaskId()).isEqualTo(21L);
assertThat(second.getTaskId()).isEqualTo(21L);
assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(taskMapper, never()).insert(any(TranscodeTask.class));
verify(taskMapper, never()).updateById(any(TranscodeTask.class));
}
@Test
void should_reset_the_same_failed_task_for_manual_retry() {
TranscodeTask failed = task(31L, TranscodeService.STATUS_FAILED);
failed.setRetryCount(3);
failed.setErrorMsg("ffmpeg failed");
given(taskMapper.selectOne(any())).willReturn(failed);
given(taskMapper.update(any(), any())).willReturn(1);
TranscodeTriggerRespVO result =
transcodeService.trigger(ASSET_ID, "/files/video.mp4");
assertThat(result.getTaskId()).isEqualTo(31L);
assertThat(result.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(taskMapper).update(any(), any());
verify(taskMapper, never()).insert(any(TranscodeTask.class));
}
@Test
void should_persist_failure_reason_and_status_after_max_retry() {
TranscodeTask running = task(41L, TranscodeService.STATUS_RUNNING);
running.setRetryCount(2);
given(taskMapper.selectById(41L)).willReturn(running);
String status = transcodeService.fail(
41L, "ffmpeg 退出码 1:Invalid data found");
assertThat(status).isEqualTo(TranscodeService.STATUS_FAILED);
assertThat(running.getStatus()).isEqualTo(TranscodeService.STATUS_FAILED);
assertThat(running.getRetryCount()).isEqualTo(3);
assertThat(running.getErrorMsg()).contains("Invalid data found");
verify(taskMapper).updateById(running);
}
private TranscodeTask task(Long id, String status) {
TranscodeTask task = new TranscodeTask();
task.setId(id);
task.setMediaAssetId(ASSET_ID);
task.setDramaId(7L);
task.setEpisodeId(31L);
task.setSourceUrl("/files/episode/202610/video.mp4");
task.setStatus(status);
task.setProgress(100);
task.setRetryCount(0);
task.setErrorMsg("");
return task;
}
}