merge: 修复转码任务认领与媒资校验

This commit is contained in:
Zhou Weidong committed 2026-10-22 15:27:58 +08:00
commit 5d1a157085
12 files changed
+268 -150

No files matched your search

@@ -0,0 +1,24 @@
# 转码任务认领与媒资 ID 校验
## 现场结论
- `media_asset.id=21` 确实存在,source_url 对应分集 id=13。
- 分集 13 的 video_asset_id=0,旧手动触发只拿媒资,不知道分集绑定,
因而创建了 media_asset_id=21、episode_id=0 的 QUEUED 任务。
- claim 只筛选 episode_id>0,因此该任务永久跳过且没有原因日志。
## 修复
1. POST /admin/media/{id}/transcode 严格只按 media_asset.id 查询;
不再把候选 id 当分集 id 兜底或补登记。
2. 媒资存在时按 source_url 反查分集;找到后绑定 drama_id/episode_id。
3. 找不到分集时拒绝建任务,避免生成永远无法认领的 episode_id=0 记录。
4. claim 增加 media_asset_id/source_url 条件,并记录被跳过任务 id 与原因。
5. RUNNING 无更新阈值收敛到 5 分钟,覆盖 worker 重启后的孤儿任务。
6. 一次性 patch 将 media_asset_id 不存在且 episode_id=0 的 QUEUED/RUNNING 标记 FAILED。
## 验证
- 非法 media id 拒绝、未绑定分集拒绝、合法 URL 自动绑定。
- claim 跳过原因、进度更新、僵尸任务恢复。
- `mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test`。
+80
View File
@@ -25625,3 +25625,83 @@ $ mvn -pl fruitshorts-admin-api -am test
解读:真实 ffmpeg smoke 已有进度回调且达到 90% 阶段点;长错误截断、
进度更新、僵尸任务收口、重复触发复用均通过。
## 2026-10-22 转码任务认领与媒资 ID 校验
- 分支:`feature/transcode-claim-fix`
- 基线:`develop` / `origin/develop` = `08eddc9`
- 计划:`docs/plans/2026-10-22-transcode-claim-fix.md`
### 现场补充取证
```text
transcode_task id=2113167358829924354
media_asset_id=21, episode_id=0, status=QUEUED,
source_url=/files/episode/202610/f1b105e185f2405d838c1066313a7ff9.mp4
media_asset id=21 存在且 source_url 相同
drama_episode id=13 使用同一 video_url,但 video_asset_id=0
```
解读:21 不是不存在的媒资,而是未绑定分集的合法媒资;旧手动触发生成的
episode_id=0 任务被 claim 条件静默跳过。
### TDD 红测
```text
$ mvn -pl fruitshorts-transcode-worker -am \
-Dtest=MediaServiceTranscodePipelineTest,TranscodeClaimAndStorageTest \
-Dsurefire.failIfNoSpecifiedTests=false test
[ERROR] 找不到符号: resolveByVideoUrl(String)
[ERROR] trigger(Long,String,Long,Long) 不存在
[INFO] BUILD FAILURE
```
解读:红测先锁定“按视频 URL 绑定分集”与严格媒资 id 触发接口。
### 实现
- POST /admin/media/{id}/transcode 只按 `media_asset.id` 查询,不再把路径 id
当成分集 id 兜底,也不再自动生成替代媒资。
- 媒资存在后按 source_url 查找使用该视频的分集;未找到直接返回
`BAD_REQUEST: 媒资未绑定分集视频`,不创建 episode_id=0 任务。
- 找到分集后先回填 video_asset_id,再向 TranscodeService 传
mediaAssetId/dramaId/episodeId/sourceUrl。
- QUEUED/RUNNING 任务复用时会补齐 drama/episode 绑定,避免历史零绑任务继续卡住。
- claim 查询增加 media_asset_id>0、episode_id>0、source_url 非空条件;
对不可认领任务按 id 去重打印一次 WARN 和具体原因。
- RUNNING 无 updated_at 变化阈值从 10 分钟收紧到 5 分钟,覆盖 worker 重启遗留。
- 认领、每档完成、整体成功/失败均有明确日志;ffmpeg 进度仍每秒节流写库。
### 数据修复脚本
文件:`sql/patch-2026-10-22-transcode-task-id-fix.sql`
脚本将 QUEUED/RUNNING 且 `episode_id=0` 或 media_asset_id 不存在于
media_asset 的任务标记 FAILED 并写明原因,可重复执行。本任务未执行数据库变更。
### 定向测试
```text
$ mvn -pl fruitshorts-transcode-worker -am \
-Dtest=MediaServiceTranscodePipelineTest,TranscodeClaimAndStorageTest,TranscodeServiceTriggerTest \
-Dsurefire.failIfNoSpecifiedTests=false test
[INFO] Tests run: 22, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS
```
### 最终门禁
```text
$ mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test
[INFO] Tests run: 8, Failures: 0, Errors: 0, Skipped: 0
[INFO] Tests run: 39, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS
$ mvn -pl fruitshorts-admin-api -am test
[INFO] Tests run: 53, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS
```
解读:非法/未绑定媒资拒绝、合法 URL 自动绑定、claim 跳过日志、
进度上报、重复触发幂等与重启僵尸收口全部通过。
@@ -43,22 +43,22 @@ public class EpisodeService implements EpisodeTranscodeCompletionPort, EpisodeVi
private final EpisodeTranscodeBindingService transcodeBindingService;
@Override
public java.util.Optional<EpisodeVideoSource> resolve(Long candidateId) {
if (candidateId == null || candidateId <= 0) {
public java.util.Optional<EpisodeVideoSource> resolveByVideoUrl(String videoUrl) {
if (!StringUtils.hasText(videoUrl)) {
return java.util.Optional.empty();
}
DramaEpisode episode = dramaEpisodeMapper.selectById(candidateId);
if (episode == null) {
episode = dramaEpisodeMapper.selectOne(Wrappers.<DramaEpisode>lambdaQuery()
.eq(DramaEpisode::getVideoAssetId, candidateId)
.last("limit 1"));
}
if (episode == null) {
return java.util.Optional.empty();
}
return java.util.Optional.of(new EpisodeVideoSource(
DramaEpisode episode = dramaEpisodeMapper.selectOne(Wrappers.<DramaEpisode>lambdaQuery()
.eq(DramaEpisode::getVideoUrl, videoUrl)
.last("limit 1"));
return episode == null
? java.util.Optional.empty()
: java.util.Optional.of(toVideoSource(episode));
}
private EpisodeVideoSource toVideoSource(DramaEpisode episode) {
return new EpisodeVideoSource(
episode.getId(), episode.getDramaId(), episode.getVideoAssetId(),
episode.getVideoUrl(), episode.getDurationSec()));
episode.getVideoUrl(), episode.getDurationSec());
}
@Override
@@ -9,8 +9,8 @@ import java.util.Optional;
*/
public interface EpisodeVideoSourcePort {
/** 按候选 id 查找分集;候选值可能是分集 id 或 video_asset_id。 */
Optional<EpisodeVideoSource> resolve(Long candidateId);
/** 按视频播放地址查找分集,用于把未回填 video_asset_id 的媒资绑定到分集。 */
Optional<EpisodeVideoSource> resolveByVideoUrl(String videoUrl);
/** 回填分集的 video_asset_id。 */
void link(Long episodeId, Long mediaAssetId);
@@ -21,9 +21,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Set;
import java.util.UUID;
@@ -37,8 +34,6 @@ public class MediaService {
private static final String STATUS_FAILED = "failed";
private static final String TASK_UPLOADING = "uploading";
private static final String TASK_COMPLETED = "completed";
private static final String BIZ_TYPE_EPISODE = "EPISODE";
private static final String STORAGE_TYPE_LOCAL = "LOCAL";
private static final Set<String> IMAGE_EXTENSIONS =
Set.of(".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp", ".avif");
@@ -210,9 +205,6 @@ public class MediaService {
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO triggerTranscode(Long id) {
MediaAsset asset = lockAsset(id);
if (asset == null) {
asset = resolveLegacyAsset(id);
}
if (asset == null) {
throw new ServiceException(ErrorCode.DATA_NOT_FOUND, "媒资不存在");
}
@@ -226,74 +218,6 @@ public class MediaService {
.last("for update"));
}
private MediaAsset resolveLegacyAsset(Long candidateId) {
return episodeVideoSourcePort.resolve(candidateId)
.map(this::resolveOrRegisterAsset)
.orElse(null);
}
private MediaAsset resolveOrRegisterAsset(EpisodeVideoSourcePort.EpisodeVideoSource source) {
String videoUrl = source.videoUrl();
if (videoUrl == null || videoUrl.isBlank()) {
return null;
}
String objectKey = storageService.objectKeyOf(videoUrl);
if (objectKey == null || objectKey.isBlank()) {
return null;
}
MediaAsset asset = mediaAssetMapper.selectOne(Wrappers.<MediaAsset>lambdaQuery()
.and(wrapper -> wrapper.eq(MediaAsset::getSourceUrl, videoUrl)
.or().eq(MediaAsset::getObjectKey, objectKey))
.last("limit 1"));
if (asset == null) {
asset = registerLegacyAsset(source, objectKey);
}
if (asset != null && source.episodeId() != null) {
episodeVideoSourcePort.link(source.episodeId(), asset.getId());
}
return asset;
}
private MediaAsset registerLegacyAsset(
EpisodeVideoSourcePort.EpisodeVideoSource source, String objectKey) {
Path file;
try {
file = storageService.fileFor(objectKey);
} catch (ServiceException ex) {
return null;
}
if (!Files.isRegularFile(file)) {
return null;
}
MediaAsset asset = new MediaAsset();
asset.setBizType(BIZ_TYPE_EPISODE);
asset.setFileName(file.getFileName().toString());
asset.setObjectKey(objectKey);
asset.setSourceUrl(source.videoUrl());
asset.setSizeBytes(sizeOf(file));
asset.setMd5("");
asset.setDurationSec(source.durationSec() == null ? 0 : source.durationSec());
asset.setWidth(0);
asset.setHeight(0);
asset.setStatus(STATUS_UPLOADED);
asset.setHlsUrl("");
asset.setMp4Url(source.videoUrl());
asset.setPosterUrl("");
asset.setTranscodeStatus(TranscodeService.STATUS_QUEUED);
asset.setStorageType(STORAGE_TYPE_LOCAL);
asset.setCreateBy(0L);
mediaAssetMapper.insert(asset);
return asset;
}
private long sizeOf(Path file) {
try {
return Files.size(file);
} catch (IOException ex) {
return 0L;
}
}
private TranscodeTriggerRespVO triggerAsset(MediaAsset asset) {
if (isImage(asset.getFileName())) {
throw new ServiceException(ErrorCode.BAD_REQUEST, "图片媒资无需转码");
@@ -302,7 +226,13 @@ public class MediaService {
if (sourceUrl == null || sourceUrl.isBlank()) {
sourceUrl = asset.getMp4Url();
}
TranscodeTriggerRespVO result = transcodeService.trigger(asset.getId(), sourceUrl);
EpisodeVideoSourcePort.EpisodeVideoSource episode = episodeVideoSourcePort
.resolveByVideoUrl(sourceUrl)
.orElseThrow(() -> new ServiceException(
ErrorCode.BAD_REQUEST, "媒资未绑定分集视频"));
episodeVideoSourcePort.link(episode.episodeId(), asset.getId());
TranscodeTriggerRespVO result = transcodeService.trigger(
asset.getId(), sourceUrl, episode.dramaId(), episode.episodeId());
asset.setStatus(STATUS_TRANSCODING);
asset.setTranscodeStatus(result.getStatus());
mediaAssetMapper.updateById(asset);
@@ -23,6 +23,8 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* 转码任务编排。
@@ -42,15 +44,17 @@ public class TranscodeService {
public static final String STATUS_FAILED = "FAILED";
/** 失败重试上限(`backend-design-phase1.md` §7.2:失败重试 3 次并记录原因)。 */
public static final int MAX_RETRY = 3;
public static final int STALE_RUNNING_MINUTES = 10;
public static final int STALE_RUNNING_MINUTES = 5;
private static final int MAX_RUNNING_PROGRESS = 99;
private static final int CLAIM_SKIP_LOG_LIMIT = 20;
private static final String STALE_RUNNING_REASON =
"任务 RUNNING 超过 10 分钟无进度更新,已自动标记失败,请重新转码";
"任务 RUNNING 超过 5 分钟无进度更新,已自动标记失败,请重新转码";
private final TranscodeTaskMapper transcodeTaskMapper;
private final MediaRenditionMapper mediaRenditionMapper;
private final MediaThumbnailMapper mediaThumbnailMapper;
private final ThumbnailRecordService thumbnailRecordService;
private final Set<Long> loggedSkipTaskIds = ConcurrentHashMap.newKeySet();
/** 上传完成后入队;同一媒资重复入队时复用未完成的任务,避免重复转码。 */
@Transactional(rollbackFor = Exception.class)
@@ -82,20 +86,27 @@ public class TranscodeService {
*/
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO trigger(Long mediaAssetId, String sourceUrl) {
return trigger(mediaAssetId, sourceUrl, 0L, 0L);
}
/** 手动重转时同时补齐分集绑定,禁止创建 episode_id=0 的不可认领任务。 */
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO trigger(
Long mediaAssetId, String sourceUrl, Long dramaId, Long episodeId) {
TranscodeTask latest = findLatestTaskByMediaAssetId(mediaAssetId);
if (latest != null && isActive(latest.getStatus())) {
updateTaskBinding(latest, dramaId, episodeId, sourceUrl);
return response(latest.getId(), latest.getStatus());
}
if (latest != null && isRetryable(latest.getStatus())) {
resetTaskForRetry(latest, sourceUrl);
resetTaskForRetry(latest, dramaId, episodeId, sourceUrl);
return response(latest.getId(), STATUS_QUEUED);
}
Long dramaId = latest == null ? 0L : latest.getDramaId();
Long episodeId = latest == null ? 0L : latest.getEpisodeId();
String retrySource = latest == null || latest.getSourceUrl() == null
|| latest.getSourceUrl().isBlank() ? sourceUrl : latest.getSourceUrl();
Long taskId = enqueue(mediaAssetId, dramaId, episodeId, retrySource);
Long taskId = enqueue(mediaAssetId, nonPositiveToZero(dramaId),
nonPositiveToZero(episodeId), retrySource);
return response(taskId, STATUS_QUEUED);
}
@@ -139,9 +150,13 @@ public class TranscodeService {
/** worker 认领任务。并发下依赖状态条件更新,抢不到返回 null。 */
public TranscodeTask claimNext() {
logSkippedQueuedTasks();
TranscodeTask candidate = transcodeTaskMapper.selectOne(Wrappers.<TranscodeTask>lambdaQuery()
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.gt(TranscodeTask::getMediaAssetId, 0)
.gt(TranscodeTask::getEpisodeId, 0)
.isNotNull(TranscodeTask::getSourceUrl)
.ne(TranscodeTask::getSourceUrl, "")
.orderByAsc(TranscodeTask::getId)
.last("limit 1"));
if (candidate == null) {
@@ -150,12 +165,48 @@ public class TranscodeService {
int updated = transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
.eq(TranscodeTask::getId, candidate.getId())
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.gt(TranscodeTask::getMediaAssetId, 0)
.gt(TranscodeTask::getEpisodeId, 0)
.isNotNull(TranscodeTask::getSourceUrl)
.ne(TranscodeTask::getSourceUrl, "")
.set(TranscodeTask::getStatus, STATUS_RUNNING)
.set(TranscodeTask::getStartedAt, LocalDateTime.now()));
return updated == 1 ? candidate : null;
}
private void logSkippedQueuedTasks() {
List<TranscodeTask> skipped = transcodeTaskMapper.selectList(
Wrappers.<TranscodeTask>lambdaQuery()
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.and(wrapper -> wrapper
.isNull(TranscodeTask::getMediaAssetId)
.or().le(TranscodeTask::getMediaAssetId, 0)
.or().isNull(TranscodeTask::getEpisodeId)
.or().le(TranscodeTask::getEpisodeId, 0)
.or().isNull(TranscodeTask::getSourceUrl)
.or().eq(TranscodeTask::getSourceUrl, ""))
.orderByAsc(TranscodeTask::getId)
.last("limit " + CLAIM_SKIP_LOG_LIMIT));
for (TranscodeTask task : skipped) {
if (task.getId() != null && loggedSkipTaskIds.add(task.getId())) {
log.warn("跳过转码任务 {}:{}", task.getId(), claimSkipReason(task));
}
}
}
private String claimSkipReason(TranscodeTask task) {
if (task.getMediaAssetId() == null || task.getMediaAssetId() <= 0) {
return "media_asset_id 无效";
}
if (task.getEpisodeId() == null || task.getEpisodeId() <= 0) {
return "episode_id 未绑定";
}
if (task.getSourceUrl() == null || task.getSourceUrl().isBlank()) {
return "source_url 为空";
}
return "未知原因";
}
@Transactional(rollbackFor = Exception.class)
public void succeed(Long taskId, List<MediaRendition> renditions) {
succeed(taskId, renditions, null);
@@ -347,8 +398,11 @@ public class TranscodeService {
|| STATUS_FAILED.equals(status) || "CANCELED".equals(status);
}
private void resetTaskForRetry(TranscodeTask task, String sourceUrl) {
private void resetTaskForRetry(
TranscodeTask task, Long dramaId, Long episodeId, String sourceUrl) {
boolean hasSource = sourceUrl != null && !sourceUrl.isBlank();
boolean hasDrama = dramaId != null && dramaId > 0;
boolean hasEpisode = episodeId != null && episodeId > 0;
transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
.eq(TranscodeTask::getId, task.getId())
.in(TranscodeTask::getStatus,
@@ -359,9 +413,15 @@ public class TranscodeService {
.set(TranscodeTask::getErrorMsg, "")
.set(TranscodeTask::getStartedAt, null)
.set(TranscodeTask::getFinishedAt, null)
.set(hasDrama, TranscodeTask::getDramaId, dramaId)
.set(hasEpisode, TranscodeTask::getEpisodeId, episodeId)
.set(hasSource, TranscodeTask::getSourceUrl, sourceUrl));
}
private long nonPositiveToZero(Long value) {
return value == null || value <= 0 ? 0L : value;
}
private TranscodeTriggerRespVO response(Long taskId, String status) {
return new TranscodeTriggerRespVO(taskId, status);
}
@@ -60,8 +60,13 @@ public class FfmpegTranscodeRunner implements TranscodeRunner {
for (int index = 0; index < request.ladder().size(); index++) {
int start = index * TRANSCODE_PROGRESS_END / totalRungs;
int end = (index + 1) * TRANSCODE_PROGRESS_END / totalRungs;
renditions.add(transcodeRung(
request, request.ladder().get(index), source, durationSec, start, end));
LadderRung rung = request.ladder().get(index);
ProducedRendition produced = transcodeRung(
request, rung, source, durationSec, start, end);
renditions.add(produced);
log.info("档位完成 task={} quality={} 实际={}x{} 进度={}%",
request.taskId(), rung.quality(),
produced.width(), produced.height(), end);
}
reportProgress(request.progressListener(), POSTER_PROGRESS_START);
writeMasterPlaylist(request, renditions);
@@ -75,6 +75,8 @@ public class TranscodeWorker {
if (task == null) {
return false;
}
log.info("开始处理转码任务 task={} asset={} episode={}",
task.getId(), task.getMediaAssetId(), task.getEpisodeId());
process(task);
return true;
}
@@ -15,9 +15,7 @@ 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 org.junit.jupiter.api.io.TempDir;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Optional;
@@ -40,10 +38,6 @@ class MediaServiceTranscodePipelineTest {
private static final Long EPISODE_ID = 31L;
private static final Long DRAMA_ID = 7L;
private static final String SOURCE_URL = "/files/episode/202610/video.mp4";
private static final String OBJECT_KEY = "episode/202610/video.mp4";
@TempDir
Path tempDir;
private MediaAssetMapper mediaAssetMapper;
private MediaUploadTaskMapper uploadTaskMapper;
@@ -99,70 +93,46 @@ class MediaServiceTranscodePipelineTest {
@Test
void should_return_not_found_when_triggering_transcode_for_missing_media() {
given(mediaAssetMapper.selectOne(any())).willReturn(null);
given(episodeVideoSourcePort.resolve(404L)).willReturn(Optional.empty());
assertThatThrownBy(() -> mediaService.triggerTranscode(404L))
.isInstanceOf(com.fruitshorts.common.exception.ServiceException.class)
.hasMessage("媒资不存在");
verify(transcodeService, never()).trigger(any(), anyString());
verify(transcodeService, never()).trigger(
any(), anyString(), any(), any());
verify(episodeVideoSourcePort, never()).resolveByVideoUrl(anyString());
}
@Test
void should_trigger_transcode_and_mark_asset_queued() {
void should_bind_episode_by_video_url_before_triggering_transcode() {
MediaAsset asset = asset("video.mp4");
given(mediaAssetMapper.selectOne(any())).willReturn(asset);
given(transcodeService.trigger(ASSET_ID, SOURCE_URL))
given(episodeVideoSourcePort.resolveByVideoUrl(SOURCE_URL))
.willReturn(Optional.of(episodeSource()));
given(transcodeService.trigger(ASSET_ID, SOURCE_URL, DRAMA_ID, EPISODE_ID))
.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);
verify(episodeVideoSourcePort).link(EPISODE_ID, ASSET_ID);
assertThat(asset.getStatus()).isEqualTo("transcoding");
assertThat(asset.getTranscodeStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(mediaAssetMapper).updateById(asset);
}
@Test
void should_resolve_existing_media_asset_by_episode_video_url() {
void should_reject_existing_media_when_no_episode_uses_video_url() {
MediaAsset asset = asset("video.mp4");
given(mediaAssetMapper.selectOne(any())).willReturn(null).willReturn(asset);
given(storageService.objectKeyOf(SOURCE_URL)).willReturn(OBJECT_KEY);
given(episodeVideoSourcePort.resolve(EPISODE_ID))
.willReturn(Optional.of(episodeSource()));
given(transcodeService.trigger(ASSET_ID, SOURCE_URL))
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
given(mediaAssetMapper.selectOne(any())).willReturn(asset);
given(episodeVideoSourcePort.resolveByVideoUrl(SOURCE_URL))
.willReturn(Optional.empty());
TranscodeTriggerRespVO result = mediaService.triggerTranscode(EPISODE_ID);
assertThat(result.getTaskId()).isEqualTo(100L);
verify(episodeVideoSourcePort).link(EPISODE_ID, ASSET_ID);
verify(transcodeService).trigger(ASSET_ID, SOURCE_URL);
}
@Test
void should_register_legacy_media_asset_when_episode_file_exists() throws Exception {
Path file = tempDir.resolve("video.mp4");
Files.writeString(file, "legacy-video");
given(mediaAssetMapper.selectOne(any())).willReturn(null);
doAnswer(invocation -> {
MediaAsset inserted = invocation.getArgument(0);
inserted.setId(ASSET_ID);
return 1;
}).when(mediaAssetMapper).insert(any(MediaAsset.class));
given(storageService.objectKeyOf(SOURCE_URL)).willReturn(OBJECT_KEY);
given(storageService.fileFor(OBJECT_KEY)).willReturn(file);
given(episodeVideoSourcePort.resolve(EPISODE_ID))
.willReturn(Optional.of(episodeSource()));
given(transcodeService.trigger(ASSET_ID, SOURCE_URL))
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
TranscodeTriggerRespVO result = mediaService.triggerTranscode(EPISODE_ID);
assertThat(result.getTaskId()).isEqualTo(100L);
verify(episodeVideoSourcePort).link(EPISODE_ID, ASSET_ID);
verify(mediaAssetMapper).insert(any(MediaAsset.class));
verify(transcodeService).trigger(ASSET_ID, SOURCE_URL);
assertThatThrownBy(() -> mediaService.triggerTranscode(ASSET_ID))
.isInstanceOf(com.fruitshorts.common.exception.ServiceException.class)
.hasMessage("媒资未绑定分集视频");
verify(transcodeService, never()).trigger(
any(), anyString(), any(), any());
}
private void givenCompletedUpload(String fileName) {
@@ -20,7 +20,10 @@ 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.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.Path;
@@ -38,6 +41,7 @@ import static org.mockito.Mockito.verify;
* ① 认领任务靠状态条件更新,抢不到就是 null(多进程同时跑也只有一个能转);
* ② objectKey → 本地路径要挡住目录穿越。
*/
@ExtendWith(OutputCaptureExtension.class)
class TranscodeClaimAndStorageTest {
/**
@@ -98,6 +102,10 @@ class TranscodeClaimAndStorageTest {
@Test
void should_exclude_unbound_queued_task_from_claim_query() {
TranscodeTask unbound = task(3L);
unbound.setMediaAssetId(88L);
unbound.setEpisodeId(0L);
given(transcodeTaskMapper.selectList(any())).willReturn(java.util.List.of(unbound));
given(transcodeTaskMapper.selectOne(any())).willReturn(null);
assertThat(transcodeService.claimNext()).isNull();
@@ -105,7 +113,23 @@ class TranscodeClaimAndStorageTest {
@SuppressWarnings("unchecked")
ArgumentCaptor<Wrapper<TranscodeTask>> captor = ArgumentCaptor.forClass(Wrapper.class);
verify(transcodeTaskMapper).selectOne(captor.capture());
assertThat(captor.getValue().getSqlSegment()).contains("episode_id >");
assertThat(captor.getValue().getSqlSegment())
.contains("episode_id >")
.contains("media_asset_id >")
.contains("source_url");
}
@Test
void should_log_skip_reason_when_queued_task_has_no_episode(CapturedOutput output) {
TranscodeTask unbound = task(9L);
unbound.setMediaAssetId(88L);
unbound.setEpisodeId(0L);
given(transcodeTaskMapper.selectList(any())).willReturn(java.util.List.of(unbound));
given(transcodeTaskMapper.selectOne(any())).willReturn(null);
assertThat(transcodeService.claimNext()).isNull();
assertThat(output).contains("跳过转码任务 9").contains("episode_id");
}
@Test
@@ -21,6 +21,7 @@ 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.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -68,7 +69,7 @@ class TranscodeServiceTriggerTest {
assertThat(second.getTaskId()).isEqualTo(10L);
assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
verify(taskMapper, never()).insert(any(TranscodeTask.class));
verify(taskMapper).update(any(), any());
verify(taskMapper, atLeastOnce()).update(any(), any());
}
@Test
@@ -0,0 +1,22 @@
-- Author: fruit.com.cn
-- 收口未绑定分集或媒资不存在的僵尸转码任务(可重复执行)
BEGIN;
UPDATE transcode_task t
SET status = 'FAILED',
error_msg = '历史任务无效:episode_id=0 或 media_asset_id 指向不存在的媒资,请从有视频的分集重新转码',
finished_at = CURRENT_TIMESTAMP
WHERE t.deleted = 0
AND t.status IN ('QUEUED', 'RUNNING')
AND (
t.episode_id = 0
OR NOT EXISTS (
SELECT 1
FROM media_asset a
WHERE a.id = t.media_asset_id
AND a.deleted = 0
)
);
COMMIT;