From 5552caa6c17893f93d8c935a0e3bf8464e5897d9 Mon Sep 17 00:00:00 2001 From: Zhou Weidong <502333560@qq.com> Date: Thu, 22 Oct 2026 15:27:51 +0800 Subject: [PATCH] =?UTF-8?q?fix(media):=20=E4=BF=AE=E6=AD=A3=E8=BD=AC?= =?UTF-8?q?=E7=A0=81=E4=BB=BB=E5=8A=A1=E8=AE=A4=E9=A2=86=E4=B8=8E=E5=AA=92?= =?UTF-8?q?=E8=B5=84=20id=20=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/plans/PROGRESS.md | 80 ++++++++++++++++++ .../content/service/EpisodeService.java | 26 +++--- .../media/service/EpisodeVideoSourcePort.java | 4 +- .../media/service/MediaService.java | 84 ++----------------- .../media/service/TranscodeService.java | 74 ++++++++++++++-- .../transcode/FfmpegTranscodeRunner.java | 9 +- .../service/transcode/TranscodeWorker.java | 2 + ...patch-2026-10-22-transcode-task-id-fix.sql | 22 +++++ 8 files changed, 200 insertions(+), 101 deletions(-) create mode 100644 sql/patch-2026-10-22-transcode-task-id-fix.sql diff --git a/docs/plans/PROGRESS.md b/docs/plans/PROGRESS.md index daefa51..b5d5917 100644 --- a/docs/plans/PROGRESS.md +++ b/docs/plans/PROGRESS.md @@ -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 跳过日志、 +进度上报、重复触发幂等与重启僵尸收口全部通过。 + diff --git a/fruitshorts-module-content/src/main/java/com/fruitshorts/content/service/EpisodeService.java b/fruitshorts-module-content/src/main/java/com/fruitshorts/content/service/EpisodeService.java index 7c73325..fa5655f 100644 --- a/fruitshorts-module-content/src/main/java/com/fruitshorts/content/service/EpisodeService.java +++ b/fruitshorts-module-content/src/main/java/com/fruitshorts/content/service/EpisodeService.java @@ -43,22 +43,22 @@ public class EpisodeService implements EpisodeTranscodeCompletionPort, EpisodeVi private final EpisodeTranscodeBindingService transcodeBindingService; @Override - public java.util.Optional resolve(Long candidateId) { - if (candidateId == null || candidateId <= 0) { + public java.util.Optional resolveByVideoUrl(String videoUrl) { + if (!StringUtils.hasText(videoUrl)) { return java.util.Optional.empty(); } - DramaEpisode episode = dramaEpisodeMapper.selectById(candidateId); - if (episode == null) { - episode = dramaEpisodeMapper.selectOne(Wrappers.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.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 diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/EpisodeVideoSourcePort.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/EpisodeVideoSourcePort.java index d3d57e3..563d1b1 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/EpisodeVideoSourcePort.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/EpisodeVideoSourcePort.java @@ -9,8 +9,8 @@ import java.util.Optional; */ public interface EpisodeVideoSourcePort { - /** 按候选 id 查找分集；候选值可能是分集 id 或 video_asset_id。 */ - Optional resolve(Long candidateId); + /** 按视频播放地址查找分集，用于把未回填 video_asset_id 的媒资绑定到分集。 */ + Optional resolveByVideoUrl(String videoUrl); /** 回填分集的 video_asset_id。 */ void link(Long episodeId, Long mediaAssetId); diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java index 8372793..b1e7dd7 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java @@ -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 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.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); diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java index 190d4bd..26e205f 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java @@ -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 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.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.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 skipped = transcodeTaskMapper.selectList( + Wrappers.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 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.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); } diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java index 174bce7..3c5d90f 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java @@ -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); diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java index 5b7b293..c845d7a 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java @@ -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; } diff --git a/sql/patch-2026-10-22-transcode-task-id-fix.sql b/sql/patch-2026-10-22-transcode-task-id-fix.sql new file mode 100644 index 0000000..7ed71af --- /dev/null +++ b/sql/patch-2026-10-22-transcode-task-id-fix.sql @@ -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;