merge: 修复转码错误截断与进度上报

This commit is contained in:
Zhou Weidong committed 2026-10-22 15:13:50 +08:00
commit 08eddc9ed2
18 files changed
+469 -34

No files matched your search

@@ -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`。
+76
View File
@@ -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% 阶段点;长错误截断、
进度更新、僵尸任务收口、重复触发复用均通过。
@@ -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;
}
@@ -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<MediaRenditionRespVO> renditions = List.of();
private MediaThumbnailRespVO thumbnails;
}
@@ -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
@@ -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
@@ -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;
@@ -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;
}
}
@@ -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<MediaRendition> renditions,
MediaThumbnail thumbnail) {
TranscodeTask task = getExisting(taskId);
mediaRenditionMapper.delete(Wrappers.<MediaRendition>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.<TranscodeTask>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.<TranscodeTask>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<MediaRendition> listRenditions(Long episodeId) {
return mediaRenditionMapper.selectList(Wrappers.<MediaRendition>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.<TranscodeTask>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)
@@ -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;
}
}
}
@@ -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<ProducedRendition> 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<String> 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<String> 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 退出把输出尾部带进异常,方便定位是片源问题还是参数问题。
*
@@ -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<LadderRung> 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<LadderRung> 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<LadderRung> 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 -> {
});
}
}
@@ -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(),
@@ -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;
}
}
@@ -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);
@@ -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);
@@ -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;
}
@@ -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;