diff --git a/docs/plans/2026-10-22-transcode-progress-fix.md b/docs/plans/2026-10-22-transcode-progress-fix.md new file mode 100644 index 0000000..3ad3f3a --- /dev/null +++ b/docs/plans/2026-10-22-transcode-progress-fix.md @@ -0,0 +1,29 @@ +# 转码错误截断与进度上报 + +## 问题 + +- ffmpeg 错误输出超过 `error_msg varchar(500)`,失败回写 UPDATE 再次失败, + 任务停在 RUNNING/0。 +- worker 未解析 ffmpeg progress,前端只能看到 0。 +- 缺少 RUNNING 超时收口。 + +## 修复 + +1. SQL patch 将 transcode_task/media_thumbnail 的 error_msg 改为 TEXT,可重复执行。 +2. 新增统一错误消息截断,最多保留 500 字符并附总长度;即使补丁尚未执行, + 也不会超过旧 varchar(500) 限制。 +3. ffmpeg 使用 `-progress` 临时文件,worker 节流为约 1 秒写 transcode_task.progress。 +4. rendition 阶段映射 0-90%,封面/缩略图 90-99%,成功后 100。 +5. 失败保留最后进度;RUNNING 超过 10 分钟无 updated_at 变化标记 FAILED。 +6. 手动重转复用最新任务,SUCCESS/FAILED/CANCELED 原地重置为 QUEUED、progress=0。 +7. 成功写 rendition 前替换同媒资旧产物,避免重复重转唯一键冲突。 + +## 风险 + +- 进度写入只允许 RUNNING 状态,避免覆盖已完成/失败状态。 +- 历史僵尸任务由 worker 下一轮自动收口,无需手工 SQL。 + +## 验证 + +- 错误截断、进度更新、僵尸重置、重复触发幂等测试。 +- `mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test`。 diff --git a/docs/plans/PROGRESS.md b/docs/plans/PROGRESS.md index d96fb64..daefa51 100644 --- a/docs/plans/PROGRESS.md +++ b/docs/plans/PROGRESS.md @@ -25549,3 +25549,79 @@ $ mvn -pl fruitshorts-admin-api -am test 解读:精度字符串、空态、video_url 反查命中、文件存在补登记及原有转码链全部通过。 +## 2026-10-22 转码错误截断与进度上报 + +- 分支:`feature/transcode-progress-fix` +- 基线:`develop` / `origin/develop` = `f9bc6ee` +- 计划:`docs/plans/2026-10-22-transcode-progress-fix.md` + +现场根因已确认:ffmpeg 错误信息超过 error_msg varchar(500),回写失败使任务 +停留在 RUNNING/0;同时现有 runner 未解析 ffmpeg progress,缺少 RUNNING 超时自愈。 + +### TDD 红测 + +```text +$ mvn -pl fruitshorts-transcode-worker -am \ + -Dtest=TranscodeServiceTriggerTest,TranscodeWorkerTest,FfmpegTranscodeRunnerSmokeTest \ + -Dsurefire.failIfNoSpecifiedTests=false test +[ERROR] TranscodeRequest 找不到 progressListener +[ERROR] 找不到符号: 方法 updateProgress(Long,int) +[ERROR] 找不到符号: 方法 recoverStaleRunningTasks() +[INFO] BUILD FAILURE +``` + +解读:红测先锁定进度回调、进度落库和僵尸自愈接口,生产代码尚未实现。 + +### 实现 + +- error_msg 写库前统一截断为最多 500 字符,附 `truncated,total=...`; + 即使数据库补丁尚未执行,也不会超过旧 varchar(500)。 +- ffmpeg 使用临时 `-progress` 文件,每 500ms 解析 out_time, + worker 最多每秒写一次 transcode_task.progress;只更新 RUNNING 状态。 +- rendition 阶段映射 0-90%,封面/缩略图 90-99%,成功事务置 100; + 失败保留最后进度。 +- RUNNING 且 updated_at 超过 10 分钟未变化时自动标记 FAILED。 +- 手动重转复用最新任务,SUCCESS/FAILED/CANCELED 均原任务重置为 QUEUED、progress=0。 +- 成功写 rendition 前删除同媒资旧产物,避免重复重转唯一键冲突。 +- 聚合 VO 和任务 VO 增加 updatedAt,供前端展示最近更新时间。 + +### SQL 补丁 + +文件:`sql/patch-2026-10-22-transcode-error-msg.sql` + +PM 按项目现有 patch 方式执行(本任务未执行数据库变更),例如使用项目当前 +PostgreSQL 连接串: + +```text +$ psql "$DB_URL" -f sql/patch-2026-10-22-transcode-error-msg.sql +``` + +脚本只将 `transcode_task.error_msg` 与 `media_thumbnail.error_msg` +改为 TEXT,可重复执行。 + +### 定向测试 + +```text +$ mvn -pl fruitshorts-transcode-worker -am \ + -Dtest=TranscodeServiceTriggerTest,TranscodeWorkerTest,FfmpegTranscodeRunnerSmokeTest \ + -Dsurefire.failIfNoSpecifiedTests=false test +[INFO] Tests run: 16, 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 +``` + +解读:真实 ffmpeg smoke 已有进度回调且达到 90% 阶段点;长错误截断、 +进度更新、僵尸任务收口、重复触发复用均通过。 + diff --git a/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminService.java b/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminService.java index 1a5bfd7..914f9d2 100644 --- a/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminService.java +++ b/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminService.java @@ -65,6 +65,7 @@ public class EpisodeTranscodeAdminService { result.setProgress(0); result.setRetryCount(0); result.setErrorMsg(""); + result.setUpdatedAt(episode.getUpdatedAt()); return result; } result.setTaskId(idString(task.getId())); @@ -72,6 +73,7 @@ public class EpisodeTranscodeAdminService { result.setProgress(task.getProgress()); result.setRetryCount(task.getRetryCount()); result.setErrorMsg(task.getErrorMsg()); + result.setUpdatedAt(task.getUpdatedAt()); return result; } diff --git a/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/vo/EpisodeTranscodeRespVO.java b/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/vo/EpisodeTranscodeRespVO.java index 6399012..8a5f08f 100644 --- a/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/vo/EpisodeTranscodeRespVO.java +++ b/fruitshorts-admin-api/src/main/java/com/fruitshorts/admin/vo/EpisodeTranscodeRespVO.java @@ -7,6 +7,7 @@ import com.fruitshorts.media.vo.MediaThumbnailRespVO; import io.swagger.v3.oas.annotations.media.Schema; import lombok.Data; +import java.time.LocalDateTime; import java.util.List; /** 后台剧集编辑页一次性获取的转码状态。 */ @@ -21,6 +22,7 @@ public class EpisodeTranscodeRespVO { private Integer progress; private Integer retryCount; private String errorMsg; + private LocalDateTime updatedAt; private List renditions = List.of(); private MediaThumbnailRespVO thumbnails; } diff --git a/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/controller/EpisodeTranscodeContractTest.java b/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/controller/EpisodeTranscodeContractTest.java index d05650d..f75b5f7 100644 --- a/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/controller/EpisodeTranscodeContractTest.java +++ b/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/controller/EpisodeTranscodeContractTest.java @@ -9,6 +9,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.test.web.servlet.MockMvc; +import java.time.LocalDateTime; + import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; @@ -38,6 +40,7 @@ class EpisodeTranscodeContractTest { vo.setTaskId(taskId); vo.setStatus("SUCCESS"); vo.setProgress(100); + vo.setUpdatedAt(LocalDateTime.of(2026, 10, 22, 15, 10)); given(service.detail(Long.parseLong(episodeId))).willReturn(vo); mockMvc.perform(get("/admin/episodes/" + episodeId + "/transcode")) @@ -47,7 +50,8 @@ class EpisodeTranscodeContractTest { .andExpect(jsonPath("$.data.mediaAssetId").value(mediaAssetId)) .andExpect(jsonPath("$.data.taskId").value(taskId)) .andExpect(jsonPath("$.data.status").value("SUCCESS")) - .andExpect(jsonPath("$.data.progress").value(100)); + .andExpect(jsonPath("$.data.progress").value(100)) + .andExpect(jsonPath("$.data.updatedAt").value("2026-10-22T15:10:00")); } @Test diff --git a/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminServiceTest.java b/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminServiceTest.java index b7f3d88..f231237 100644 --- a/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminServiceTest.java +++ b/fruitshorts-admin-api/src/test/java/com/fruitshorts/admin/service/EpisodeTranscodeAdminServiceTest.java @@ -11,6 +11,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.util.List; +import java.time.LocalDateTime; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.given; @@ -77,6 +78,7 @@ class EpisodeTranscodeAdminServiceTest { task.setMediaAssetId(MEDIA_ASSET_ID); task.setStatus("SUCCESS"); task.setProgress(100); + task.setUpdatedAt(LocalDateTime.of(2026, 10, 22, 15, 10)); given(outputService.latestTask(MEDIA_ASSET_ID)).willReturn(task); var result = service.detail(EPISODE_ID); @@ -84,6 +86,7 @@ class EpisodeTranscodeAdminServiceTest { assertThat(result.getEpisodeId()).isEqualTo("2113150431755636737"); assertThat(result.getMediaAssetId()).isEqualTo("2113150398670966786"); assertThat(result.getTaskId()).isEqualTo("2113150398935207938"); + assertThat(result.getUpdatedAt()).isEqualTo(task.getUpdatedAt()); } @Test diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/ThumbnailRecordService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/ThumbnailRecordService.java index 9c3baaa..965ed99 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/ThumbnailRecordService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/ThumbnailRecordService.java @@ -33,6 +33,7 @@ public class ThumbnailRecordService { .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; diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeErrorTruncator.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeErrorTruncator.java new file mode 100644 index 0000000..38b4170 --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeErrorTruncator.java @@ -0,0 +1,23 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.service; + +/** 转码错误消息统一截断,避免数据库写入失败。 */ +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 TranscodeErrorTruncator() { + } + + public static String truncate(String message) { + if (message == null || message.isEmpty() || message.length() <= MAX_ERROR_CHARS) { + 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; + } +} 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 a59f135..190d4bd 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 @@ -42,6 +42,10 @@ 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; + private static final int MAX_RUNNING_PROGRESS = 99; + private static final String STALE_RUNNING_REASON = + "任务 RUNNING 超过 10 分钟无进度更新,已自动标记失败,请重新转码"; private final TranscodeTaskMapper transcodeTaskMapper; private final MediaRenditionMapper mediaRenditionMapper; @@ -83,7 +87,7 @@ public class TranscodeService { return response(latest.getId(), latest.getStatus()); } if (latest != null && isRetryable(latest.getStatus())) { - resetFailedTask(latest, sourceUrl); + resetTaskForRetry(latest, sourceUrl); return response(latest.getId(), STATUS_QUEUED); } @@ -162,6 +166,8 @@ public class TranscodeService { public void succeed(Long taskId, List renditions, MediaThumbnail thumbnail) { TranscodeTask task = getExisting(taskId); + mediaRenditionMapper.delete(Wrappers.lambdaQuery() + .eq(MediaRendition::getMediaAssetId, task.getMediaAssetId())); for (MediaRendition rendition : renditions) { rendition.setMediaAssetId(task.getMediaAssetId()); rendition.setDramaId(task.getDramaId()); @@ -173,10 +179,10 @@ public class TranscodeService { if (thumbnail != null) { try { thumbnailRecordService.upsert(task, thumbnail); - thumbnailError = thumbnail.getErrorMsg() == null - ? "" : thumbnail.getErrorMsg(); + thumbnailError = TranscodeErrorTruncator.truncate(thumbnail.getErrorMsg()); } catch (Exception ex) { - thumbnailError = "缩略图登记失败:" + exceptionMessage(ex); + thumbnailError = TranscodeErrorTruncator.truncate( + "缩略图登记失败:" + exceptionMessage(ex)); log.warn("缩略图登记失败 task={}:{}", taskId, ex.getMessage()); } } @@ -202,12 +208,35 @@ public class TranscodeService { task.setStatus(STATUS_QUEUED); } task.setRetryCount(retry + 1); - task.setErrorMsg(errorMessage == null ? "" : errorMessage); + task.setErrorMsg(TranscodeErrorTruncator.truncate(errorMessage)); task.setFinishedAt(LocalDateTime.now()); transcodeTaskMapper.updateById(task); return task.getStatus(); } + /** worker 实时上报 RUNNING 任务进度,最多到 99,100 由成功事务统一写入。 */ + public void updateProgress(Long taskId, int progress) { + if (taskId == null) { + return; + } + int safeProgress = Math.min(MAX_RUNNING_PROGRESS, Math.max(0, progress)); + transcodeTaskMapper.update(null, Wrappers.lambdaUpdate() + .eq(TranscodeTask::getId, taskId) + .eq(TranscodeTask::getStatus, STATUS_RUNNING) + .set(TranscodeTask::getProgress, safeProgress)); + } + + /** RUNNING 超过阈值且 updated_at 长期未变时自动收口,避免僵尸任务永久占位。 */ + public int recoverStaleRunningTasks() { + LocalDateTime now = LocalDateTime.now(); + return transcodeTaskMapper.update(null, Wrappers.lambdaUpdate() + .eq(TranscodeTask::getStatus, STATUS_RUNNING) + .lt(TranscodeTask::getUpdatedAt, now.minusMinutes(STALE_RUNNING_MINUTES)) + .set(TranscodeTask::getStatus, STATUS_FAILED) + .set(TranscodeTask::getErrorMsg, STALE_RUNNING_REASON) + .set(TranscodeTask::getFinishedAt, now)); + } + public List listRenditions(Long episodeId) { return mediaRenditionMapper.selectList(Wrappers.lambdaQuery() .eq(MediaRendition::getEpisodeId, episodeId) @@ -314,14 +343,16 @@ public class TranscodeService { } private boolean isRetryable(String status) { - return STATUS_FAILED.equals(status) || "CANCELED".equals(status); + return STATUS_SUCCESS.equals(status) + || STATUS_FAILED.equals(status) || "CANCELED".equals(status); } - private void resetFailedTask(TranscodeTask task, String sourceUrl) { + private void resetTaskForRetry(TranscodeTask task, String sourceUrl) { boolean hasSource = sourceUrl != null && !sourceUrl.isBlank(); transcodeTaskMapper.update(null, Wrappers.lambdaUpdate() .eq(TranscodeTask::getId, task.getId()) - .in(TranscodeTask::getStatus, List.of(STATUS_FAILED, "CANCELED")) + .in(TranscodeTask::getStatus, + List.of(STATUS_SUCCESS, STATUS_FAILED, "CANCELED")) .set(TranscodeTask::getStatus, STATUS_QUEUED) .set(TranscodeTask::getProgress, 0) .set(TranscodeTask::getRetryCount, 0) diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegProgressMonitor.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegProgressMonitor.java new file mode 100644 index 0000000..9c950f4 --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegProgressMonitor.java @@ -0,0 +1,105 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.service.transcode; + +import com.fruitshorts.common.exception.ErrorCode; +import com.fruitshorts.common.exception.ServiceException; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntConsumer; + +/** 解析 ffmpeg -progress 临时文件并节流上报。 */ +final class FfmpegProgressMonitor { + + private static final long MICROS_PER_SECOND = 1_000_000L; + + private FfmpegProgressMonitor() { + } + + static Path createTempFile() { + try { + return Files.createTempFile("fruitshorts-ffmpeg-progress-", ".log"); + } catch (IOException ex) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + "创建 ffmpeg 进度文件失败:" + ex.getMessage()); + } + } + + static void report( + Path progressFile, int durationSec, + AtomicInteger lastReported, IntConsumer listener) { + if (!Files.isRegularFile(progressFile)) { + return; + } + String content; + try { + content = Files.readString(progressFile, StandardCharsets.UTF_8); + } catch (IOException ex) { + return; + } + long microseconds = extractOutTimeMicros(content); + if (microseconds < 0) { + return; + } + int percent = durationSec <= 0 ? 0 + : (int) Math.min(100, microseconds * 100 / (durationSec * MICROS_PER_SECOND)); + if (percent > lastReported.get()) { + lastReported.set(percent); + listener.accept(percent); + } + } + + static void deleteQuietly(Path path) { + if (path == null) { + return; + } + try { + Files.deleteIfExists(path); + } catch (IOException ignored) { + // 临时进度文件删不掉不影响业务结果 + } + } + + private static long extractOutTimeMicros(String content) { + long result = -1L; + for (String line : content.split("\\R")) { + String[] pair = line.trim().split("=", 2); + if (pair.length != 2) { + continue; + } + if ("out_time_us".equals(pair[0]) || "out_time_ms".equals(pair[0])) { + result = Math.max(result, parseLong(pair[1])); + } else if ("out_time".equals(pair[0])) { + result = Math.max(result, parseTimestampMicros(pair[1])); + } + } + return result; + } + + private static long parseTimestampMicros(String timestamp) { + String[] parts = timestamp.trim().split(":"); + if (parts.length != 3) { + return -1L; + } + try { + double seconds = Integer.parseInt(parts[0]) * 3600.0 + + Integer.parseInt(parts[1]) * 60.0 + + Double.parseDouble(parts[2]); + return Math.round(seconds * MICROS_PER_SECOND); + } catch (NumberFormatException ex) { + return -1L; + } + } + + private static long parseLong(String value) { + try { + return Long.parseLong(value.trim()); + } catch (NumberFormatException ex) { + return -1L; + } + } +} 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 3f911cf..174bce7 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 @@ -15,6 +15,8 @@ import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntConsumer; import java.util.stream.Stream; /** @@ -35,6 +37,11 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { private static final String VARIANT_PLAYLIST = "index.m3u8"; private static final String POSTER_FILE = "poster.jpg"; private static final int STDERR_TAIL = 800; + private static final int TRANSCODE_PROGRESS_END = 90; + private static final int POSTER_PROGRESS_START = 92; + private static final int POSTER_PROGRESS_END = 95; + private static final int THUMBNAIL_PROGRESS_END = 99; + private static final long PROCESS_POLL_MILLIS = 500; private final FfmpegThumbnailGenerator thumbnailGenerator = new FfmpegThumbnailGenerator(); @Override @@ -46,14 +53,23 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { createDirectories(request.outputDir()); int durationSec = probeDuration(request); SourceDimensions source = probeSourceDimensions(request); + reportProgress(request.progressListener(), 1); List renditions = new ArrayList<>(request.ladder().size()); - for (LadderRung rung : request.ladder()) { - renditions.add(transcodeRung(request, rung, source)); + int totalRungs = Math.max(1, request.ladder().size()); + 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)); } + reportProgress(request.progressListener(), POSTER_PROGRESS_START); writeMasterPlaylist(request, renditions); ProducedRendition poster = extractPoster(request); + reportProgress(request.progressListener(), POSTER_PROGRESS_END); + reportProgress(request.progressListener(), 96); ProducedThumbnail thumbnail = thumbnailGenerator.generate(request, durationSec); + reportProgress(request.progressListener(), THUMBNAIL_PROGRESS_END); log.info("转码完成 task={}:{} 档 HLS + 封面 + 缩略图状态={},时长 {}s", request.taskId(), renditions.size(), thumbnail == null ? "DISABLED" : thumbnail.status(), durationSec); @@ -62,7 +78,8 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { } private ProducedRendition transcodeRung( - TranscodeRequest request, LadderRung rung, SourceDimensions source) { + TranscodeRequest request, LadderRung rung, SourceDimensions source, + int durationSec, int progressStart, int progressEnd) { String relativeDir = rung.quality() + "/"; Path rungDir = request.outputDir().resolve(rung.quality()); createDirectories(rungDir); @@ -70,8 +87,10 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { String scaleFilter = source.landscape() ? "scale=-2:" + shortEdge + ":flags=lanczos,setsar=1" : "scale=" + shortEdge + ":-2:flags=lanczos,setsar=1"; + Path progressFile = FfmpegProgressMonitor.createTempFile(); List command = List.of( request.ffmpegPath(), "-y", + "-progress", progressFile.toString(), "-nostats", "-i", request.sourceFile().toString(), // 按源方向只用短边定档;-2 保证自动推导的一边为偶数,setsar=1 避免播放器再次拉伸 "-vf", scaleFilter, @@ -85,7 +104,15 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { "-hls_playlist_type", "vod", "-hls_segment_filename", rungDir.resolve("seg_%03d.ts").toString(), rungDir.resolve(VARIANT_PLAYLIST).toString()); - runProcess(command, request.timeoutSeconds(), "转码 " + rung.quality() + " 失败"); + try { + runProcessWithProgress(command, request.timeoutSeconds(), + "转码 " + rung.quality() + " 失败", progressFile, durationSec, + rawProgress -> reportProgress(request.progressListener(), + mapProgress(rawProgress, progressStart, progressEnd))); + reportProgress(request.progressListener(), progressEnd); + } finally { + FfmpegProgressMonitor.deleteQuietly(progressFile); + } Path playlist = rungDir.resolve(VARIANT_PLAYLIST); if (!Files.exists(playlist)) { @@ -99,6 +126,17 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { sizeOf(rungDir), countSegments(rungDir)); } + private int mapProgress(int rawProgress, int start, int end) { + int safe = Math.min(100, Math.max(0, rawProgress)); + return start + (end - start) * safe / 100; + } + + private void reportProgress(IntConsumer listener, int progress) { + if (listener != null) { + listener.accept(Math.min(99, Math.max(0, progress))); + } + } + private int effectiveShortEdge(LadderRung rung, SourceDimensions source) { int rungShortEdge = Math.min(rung.width(), rung.height()); int capped = Math.min(rungShortEdge, source.shortEdge()); @@ -247,6 +285,47 @@ public class FfmpegTranscodeRunner implements TranscodeRunner { } } + /** 运行 ffmpeg 并轮询 -progress 文件,把 out_time 映射成 0-100。 */ + private String runProcessWithProgress( + List command, int timeoutSeconds, String errorPrefix, + Path progressFile, int durationSec, IntConsumer listener) { + Path logFile = null; + try { + logFile = Files.createTempFile("fruitshorts-ffmpeg-", ".log"); + Process process = new ProcessBuilder(command) + .redirectErrorStream(true) + .redirectOutput(logFile.toFile()) + .start(); + AtomicInteger lastReported = new AtomicInteger(-1); + long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L; + while (!process.waitFor(PROCESS_POLL_MILLIS, TimeUnit.MILLISECONDS)) { + FfmpegProgressMonitor.report(progressFile, durationSec, lastReported, listener); + if (System.currentTimeMillis() > deadline) { + process.destroyForcibly(); + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + ":超时(" + timeoutSeconds + "s)"); + } + } + FfmpegProgressMonitor.report(progressFile, durationSec, lastReported, listener); + String output = Files.readString(logFile, StandardCharsets.UTF_8); + if (process.exitValue() != 0) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + ":" + tail(output)); + } + return output; + } catch (ServiceException ex) { + throw ex; + } catch (IOException ex) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + ":无法执行 ffmpeg(" + ex.getMessage() + ")"); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new ServiceException(ErrorCode.INTERNAL_ERROR, errorPrefix + ":被中断"); + } finally { + FfmpegProgressMonitor.deleteQuietly(logFile); + } + } + /** * 起进程并等结果;非 0 退出把输出尾部带进异常,方便定位是片源问题还是参数问题。 * diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java index 814b096..8a1decf 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java @@ -4,6 +4,7 @@ package com.fruitshorts.media.service.transcode; import java.nio.file.Path; import java.util.List; +import java.util.function.IntConsumer; /** * 转码执行端口:把"调 ffmpeg 出多码率 HLS + 封面"封在一层后面。 @@ -30,14 +31,25 @@ public interface TranscodeRunner { record TranscodeRequest(Long taskId, Path sourceFile, Path outputDir, String objectKeyPrefix, List ladder, String ffmpegPath, String ffprobePath, int segmentSeconds, int posterSecond, int timeoutSeconds, - SpriteOptions sprite) { + SpriteOptions sprite, IntConsumer progressListener) { + + public TranscodeRequest(Long taskId, Path sourceFile, Path outputDir, String objectKeyPrefix, + List ladder, String ffmpegPath, String ffprobePath, + int segmentSeconds, int posterSecond, int timeoutSeconds, + SpriteOptions sprite) { + this(taskId, sourceFile, outputDir, objectKeyPrefix, ladder, ffmpegPath, ffprobePath, + segmentSeconds, posterSecond, timeoutSeconds, sprite, ignored -> { + }); + } /** 兼容未启用缩略图的既有调用。 */ public TranscodeRequest(Long taskId, Path sourceFile, Path outputDir, String objectKeyPrefix, List ladder, String ffmpegPath, String ffprobePath, int segmentSeconds, int posterSecond, int timeoutSeconds) { this(taskId, sourceFile, outputDir, objectKeyPrefix, ladder, ffmpegPath, ffprobePath, - segmentSeconds, posterSecond, timeoutSeconds, SpriteOptions.disabled()); + segmentSeconds, posterSecond, timeoutSeconds, SpriteOptions.disabled(), + ignored -> { + }); } } 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 0e897ba..5b7b293 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 @@ -9,6 +9,7 @@ import com.fruitshorts.media.service.MediaService; import com.fruitshorts.media.service.MediaStorageService; import com.fruitshorts.media.service.MediaTranscodeProfileService; import com.fruitshorts.media.service.EpisodeTranscodeCompletionPort; +import com.fruitshorts.media.service.TranscodeErrorTruncator; import com.fruitshorts.media.service.TranscodeService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -19,6 +20,8 @@ import org.springframework.stereotype.Component; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.IntConsumer; /** * 转码 worker:消费 `transcode_task`,产出多码率 HLS + 封面。 @@ -38,6 +41,8 @@ import java.util.List; @ConditionalOnProperty(name = "fruitshorts.media.transcode.worker-enabled", havingValue = "true") public class TranscodeWorker { + private static final long PROGRESS_WRITE_INTERVAL_MS = 1000L; + private final TranscodeService transcodeService; private final MediaService mediaService; private final MediaStorageService storageService; @@ -49,6 +54,10 @@ public class TranscodeWorker { @Scheduled(fixedDelayString = "${fruitshorts.media.transcode.poll-delay-ms:5000}") public void poll() { try { + int recovered = transcodeService.recoverStaleRunningTasks(); + if (recovered > 0) { + log.warn("已自动收口 {} 个超时 RUNNING 转码任务", recovered); + } processOne(); } catch (Exception ex) { // 轮询里不能让异常逃出去,否则调度线程会被取消,整个 worker 静默停摆 @@ -82,7 +91,8 @@ public class TranscodeWorker { task.getId(), sourceFile, outputDir, objectKeyPrefix, ladder(profile), profileService.ffmpegPath(), profileService.ffprobePath(), profile.getSegmentSeconds(), profileService.posterSecond(), - profileService.timeoutSeconds(), spriteOptions(profile))); + profileService.timeoutSeconds(), spriteOptions(profile), + progressListener(task.getId()))); transcodeService.succeed(task.getId(), toRenditions(result), toThumbnail(task, result.thumbnail())); @@ -140,6 +150,17 @@ public class TranscodeWorker { sprite.getColumns(), sprite.getThumbWidth()); } + private IntConsumer progressListener(Long taskId) { + AtomicLong lastWriteAt = new AtomicLong(0L); + return progress -> { + long now = System.currentTimeMillis(); + if (now - lastWriteAt.get() >= PROGRESS_WRITE_INTERVAL_MS || progress >= 99) { + lastWriteAt.set(now); + transcodeService.updateProgress(taskId, progress); + } + }; + } + private MediaThumbnail toThumbnail( TranscodeTask task, TranscodeRunner.ProducedThumbnail produced) { if (produced == null) { @@ -159,7 +180,7 @@ public class TranscodeWorker { thumbnail.setDurationSec(produced.durationSec()); thumbnail.setSizeBytes(produced.sizeBytes()); thumbnail.setStatus(produced.status()); - thumbnail.setErrorMsg(produced.errorMessage()); + thumbnail.setErrorMsg(TranscodeErrorTruncator.truncate(produced.errorMessage())); return thumbnail; } @@ -169,7 +190,8 @@ public class TranscodeWorker { return; } if (MediaThumbnail.STATUS_FAILED.equals(thumbnail.status())) { - log.warn("主转码成功但缩略图失败:{}", thumbnail.errorMessage()); + log.warn("主转码成功但缩略图失败:{}", + TranscodeErrorTruncator.truncate(thumbnail.errorMessage())); return; } log.info("缩略图完成:{} 帧,{}x{}", thumbnail.frameCount(), diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/vo/TranscodeTaskRespVO.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/vo/TranscodeTaskRespVO.java index ac51407..dd6913d 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/vo/TranscodeTaskRespVO.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/vo/TranscodeTaskRespVO.java @@ -26,6 +26,7 @@ public class TranscodeTaskRespVO { private LocalDateTime startedAt; private LocalDateTime finishedAt; private LocalDateTime createdAt; + private LocalDateTime updatedAt; public static TranscodeTaskRespVO from(TranscodeTask task) { if (task == null) { @@ -45,6 +46,7 @@ public class TranscodeTaskRespVO { vo.setStartedAt(task.getStartedAt()); vo.setFinishedAt(task.getFinishedAt()); vo.setCreatedAt(task.getCreatedAt()); + vo.setUpdatedAt(task.getUpdatedAt()); return vo; } } diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java index a0f9a43..532dbf6 100644 --- a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java @@ -12,6 +12,7 @@ import java.nio.file.Path; import java.util.Comparator; import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -40,13 +41,15 @@ class FfmpegTranscodeRunnerSmokeTest { Path output = work.resolve("out"); TranscodeRunner runner = new FfmpegTranscodeRunner(); + AtomicInteger maxProgress = new AtomicInteger(); // 只跑两档(360p/480p)保持秒级;四档阶梯的逻辑在 TranscodeWorkerTest 里断言 TranscodeRunner.TranscodeResult result = runner.run(new TranscodeRunner.TranscodeRequest( 1L, sample, output, "transcoded/smoke", List.of(new TranscodeRunner.LadderRung("360p", 640, 360, 400, 64), new TranscodeRunner.LadderRung("480p", 854, 480, 800, 64)), ffmpeg, ffprobe, 2, 1, 120, - new TranscodeRunner.SpriteOptions(true, 1, 10, 160))); + new TranscodeRunner.SpriteOptions(true, 1, 10, 160), + value -> maxProgress.accumulateAndGet(value, Math::max))); assertThat(Files.exists(output.resolve("master.m3u8"))).isTrue(); assertThat(Files.exists(output.resolve("poster.jpg"))).isTrue(); @@ -60,6 +63,7 @@ class FfmpegTranscodeRunnerSmokeTest { assertThat(rendition.sizeBytes()).isGreaterThan(0L); } assertThat(result.durationSec()).isBetween(1, 5); + assertThat(maxProgress.get()).isGreaterThanOrEqualTo(90); assertThat(result.thumbnail()).isNotNull(); assertThat(result.thumbnail().status()).isEqualTo("READY"); assertThat(result.thumbnail().thumbWidth()).isEqualTo(160); diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeServiceTriggerTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeServiceTriggerTest.java index f52e045..998de70 100644 --- a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeServiceTriggerTest.java +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeServiceTriggerTest.java @@ -21,7 +21,6 @@ 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; @@ -54,24 +53,22 @@ class TranscodeServiceTriggerTest { } @Test - void should_create_once_and_reuse_queued_task_when_retrying_successful_asset() { + void should_reset_completed_task_in_place_and_reuse_it_when_triggered_again() { 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)); + queued.setId(10L); + given(taskMapper.selectOne(any())).willReturn(success, queued); + given(taskMapper.update(any(), any())).willReturn(1); 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.getTaskId()).isEqualTo(10L); assertThat(first.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED); - assertThat(second.getTaskId()).isEqualTo(100L); + assertThat(second.getTaskId()).isEqualTo(10L); assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED); - verify(taskMapper).insert(any(TranscodeTask.class)); + verify(taskMapper, never()).insert(any(TranscodeTask.class)); + verify(taskMapper).update(any(), any()); } @Test @@ -111,17 +108,39 @@ class TranscodeServiceTriggerTest { TranscodeTask running = task(41L, TranscodeService.STATUS_RUNNING); running.setRetryCount(2); given(taskMapper.selectById(41L)).willReturn(running); + String longReason = "ffmpeg 输出异常:" + "x".repeat(5000); - String status = transcodeService.fail( - 41L, "ffmpeg 退出码 1:Invalid data found"); + String status = transcodeService.fail(41L, longReason); 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"); + assertThat(running.getErrorMsg()).contains("ffmpeg 输出异常"); + assertThat(running.getErrorMsg()).contains("truncated"); + assertThat(running.getErrorMsg()).contains("total=" + longReason.length()); + assertThat(running.getErrorMsg().length()).isLessThanOrEqualTo(500); verify(taskMapper).updateById(running); } + @Test + void should_update_progress_only_for_running_task() { + given(taskMapper.update(any(), any())).willReturn(1); + + transcodeService.updateProgress(51L, 46); + + verify(taskMapper).update(any(), any()); + } + + @Test + void should_mark_stale_running_task_failed() { + given(taskMapper.update(any(), any())).willReturn(2); + + int recovered = transcodeService.recoverStaleRunningTasks(); + + assertThat(recovered).isEqualTo(2); + verify(taskMapper).update(any(), any()); + } + private TranscodeTask task(Long id, String status) { TranscodeTask task = new TranscodeTask(); task.setId(id); diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java index dbb4588..0335614 100644 --- a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java @@ -72,6 +72,7 @@ class TranscodeWorkerTest { @Test void should_register_renditions_and_mark_asset_ready_when_transcode_succeeds() { given(transcodeService.claimNext()).willReturn(queuedTask()); + runner.progress = 45; assertThat(worker.processOne()).isTrue(); @@ -92,6 +93,7 @@ class TranscodeWorkerTest { eq("/files/transcoded/88/poster.jpg"), eq(120)); verify(episodeCompletionPort).markEpisodeReady( 12L, "/files/transcoded/88/master.m3u8", 120); + verify(transcodeService).updateProgress(TASK_ID, 45); verify(transcodeService, never()).fail(any(), anyString()); } @@ -110,11 +112,12 @@ class TranscodeWorkerTest { @Test void should_keep_main_transcode_successful_when_thumbnail_generation_fails() { given(transcodeService.claimNext()).willReturn(queuedTask()); + String thumbnailError = "tile 滤镜失败:" + "x".repeat(5000); runner.thumbnail = new TranscodeRunner.ProducedThumbnail( "transcoded/88/thumbnails/thumbnails.vtt", "transcoded/88/thumbnails/sprite_", 1, 10, 1, 160, 0, 120, 120, 0L, - MediaThumbnail.STATUS_FAILED, "tile 滤镜失败"); + MediaThumbnail.STATUS_FAILED, thumbnailError); worker.processOne(); @@ -128,6 +131,8 @@ class TranscodeWorkerTest { assertThat(thumbnail.getValue().getStatus()) .isEqualTo(MediaThumbnail.STATUS_FAILED); assertThat(thumbnail.getValue().getErrorMsg()).contains("tile"); + assertThat(thumbnail.getValue().getErrorMsg()).contains("truncated"); + assertThat(thumbnail.getValue().getErrorMsg().length()).isLessThanOrEqualTo(500); verify(mediaService).markReady(eq(ASSET_ID), anyString(), anyString(), eq(120)); verify(transcodeService, never()).fail(any(), anyString()); } @@ -190,12 +195,16 @@ class TranscodeWorkerTest { private TranscodeRequest request; private RuntimeException failure; private ProducedThumbnail thumbnail; + private Integer progress; private boolean called; @Override public TranscodeResult run(TranscodeRequest request) { this.request = request; this.called = true; + if (progress != null) { + request.progressListener().accept(progress); + } if (failure != null) { throw failure; } diff --git a/sql/patch-2026-10-22-transcode-error-msg.sql b/sql/patch-2026-10-22-transcode-error-msg.sql new file mode 100644 index 0000000..5b6319e --- /dev/null +++ b/sql/patch-2026-10-22-transcode-error-msg.sql @@ -0,0 +1,12 @@ +-- Author: fruit.com.cn +-- 扩展转码错误信息列,允许保留 ffmpeg / ffprobe 关键错误上下文(可重复执行) + +BEGIN; + +ALTER TABLE IF EXISTS transcode_task + ALTER COLUMN error_msg TYPE TEXT USING error_msg::TEXT; + +ALTER TABLE IF EXISTS media_thumbnail + ALTER COLUMN error_msg TYPE TEXT USING error_msg::TEXT; + +COMMIT;