fix(media): 转码错误信息截断并上报进度

This commit is contained in:
Zhou Weidong committed 2026-10-22 15:12:05 +08:00
1 parent f9bc6eee6d
commit 660afaf66d
18 files changed
+469 -34

No files matched your search

@@ -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;
}
}