fix(media): propagate async FFmpeg results and failures

This commit is contained in:
2026-07-12 18:49:20 +01:00
parent 688ff541da
commit 2408263988
10 changed files with 116 additions and 77 deletions

View File

@@ -222,14 +222,11 @@ public class ClipService {
} }
// Thumbnail generation can fail with error propagation // Thumbnail generation can fail with error propagation
try { thumbnailService.createThumbnail(newClipFile, thumbnailFile, 0.0f)
thumbnailService.createThumbnail(newClipFile, thumbnailFile, 0.0f); .exceptionally(ex -> {
} catch (InterruptedException e) { logger.error("Thumbnail job for user {} failed: {}", user.getId(), ex.toString());
logger.error("Thumbnail generation interrupted for user: {}", user.getId(), e); return null;
Thread.currentThread().interrupt(); });
} catch (IOException e) {
logger.error("Error generating thumbnail for user: {}", user.getId(), e);
}
clipMetadata.setTitle(title); clipMetadata.setTitle(title);
clipMetadata.setDescription(description); clipMetadata.setDescription(description);

View File

@@ -59,9 +59,8 @@ public class MediaService {
* @return the {@link Job} tracking the compression; state transitions to * @return the {@link Job} tracking the compression; state transitions to
* {@link JobState#SUCCEEDED} or {@link JobState#FAILED} asynchronously * {@link JobState#SUCCEEDED} or {@link JobState#FAILED} asynchronously
* @throws IOException if saving the upload to the temporary directory fails * @throws IOException if saving the upload to the temporary directory fails
* @throws InterruptedException if the current thread is interrupted while setting up the job
*/ */
public Job compress(MultipartFile file, ClipOptions clipOptions) throws IOException, InterruptedException { public Job compress(MultipartFile file, ClipOptions clipOptions) throws IOException {
Job job = jobRegistryService.generateJob(); Job job = jobRegistryService.generateJob();
Optional<User> user = userService.getLoggedInUser(); Optional<User> user = userService.getLoggedInUser();

View File

@@ -201,14 +201,11 @@ public class VodService {
throw new FFMPEGException("Error retrieving video metadata for vod: " + e.getMessage()); throw new FFMPEGException("Error retrieving video metadata for vod: " + e.getMessage());
} }
try { thumbnailService.createThumbnail(newVodFile, thumbnailFile, 0.0f)
thumbnailService.createThumbnail(newVodFile, thumbnailFile, 0.0f); .exceptionally(ex -> {
} catch (InterruptedException e) { logger.error("Thumbnail job for user {} failed: {}", user.getId(), ex.toString());
logger.error("Thumbnail generation interrupted for user: {}", user.getId(), e); return null;
Thread.currentThread().interrupt(); });
} catch (IOException e) {
logger.error("Error generating thumbnail for user: {}", user.getId(), e);
}
vodMetadata.setTitle(title); vodMetadata.setTitle(title);
vodMetadata.setDescription(description); vodMetadata.setDescription(description);

View File

@@ -32,18 +32,27 @@ public class CompressionService {
Path outputFile, Path outputFile,
ClipOptions clipOptions, ClipOptions clipOptions,
ProgressTracker progress ProgressTracker progress
) throws IOException, InterruptedException { ) {
logger.info("Compressing video from {} to {}", inputFile.toAbsolutePath(), outputFile.toAbsolutePath()); logger.info("Compressing video from {} to {}", inputFile.toAbsolutePath(), outputFile.toAbsolutePath());
List<String> command = buildCommand( try {
inputFile, List<String> command = buildCommand(
outputFile, inputFile,
clipOptions outputFile,
); clipOptions
CommandOutput result = commandRunner.run(command, line -> commandRunner.setProgress(line, progress, clipOptions.getDuration())); );
progress.markComplete(); CommandOutput result = commandRunner.run(command, line -> commandRunner.setProgress(line, progress, clipOptions.getDuration()));
progress.markComplete();
return CompletableFuture.completedFuture(result); return CompletableFuture.completedFuture(result);
} catch (IOException e) {
logger.error("IO error on compress call: {}", e.toString());
return CompletableFuture.failedFuture(e);
} catch (InterruptedException e) {
logger.error("Thread error on compress call: {}", e.toString());
Thread.currentThread().interrupt();
return CompletableFuture.failedFuture(e);
}
} }
private List<String> buildFilters(Float fps, Integer width, Integer height) { private List<String> buildFilters(Float fps, Integer width, Integer height) {

View File

@@ -14,7 +14,6 @@ import java.io.IOException;
import java.nio.file.Path; import java.nio.file.Path;
import java.util.List; import java.util.List;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
@Service @Service
public class MetadataService { public class MetadataService {
@@ -28,7 +27,7 @@ public class MetadataService {
} }
@Async("ffmpegTaskExecutor") @Async("ffmpegTaskExecutor")
public Future<ClipOptions> getVideoMetadata(Path file) { public CompletableFuture<ClipOptions> getVideoMetadata(Path file) {
logger.info("Getting metadata for file {}", file); logger.info("Getting metadata for file {}", file);
List<String> command = List.of( List<String> command = List.of(
@@ -53,10 +52,10 @@ public class MetadataService {
JsonNode node = mapper.readTree(outputBuilder.toString()); JsonNode node = mapper.readTree(outputBuilder.toString());
return CompletableFuture.completedFuture(parseVideoMetadata(node)); return CompletableFuture.completedFuture(parseVideoMetadata(node));
} catch (IOException e) { } catch (IOException e) {
throw new FFMPEGException("Error while getting video metadata: " + e); return CompletableFuture.failedFuture(e);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
throw new FFMPEGException("Error while getting video metadata: " + e); return CompletableFuture.failedFuture(e);
} }
} }

View File

@@ -2,6 +2,8 @@ package com.ddf.vodsystem.services.media;
import com.ddf.vodsystem.dto.CommandOutput; import com.ddf.vodsystem.dto.CommandOutput;
import com.ddf.vodsystem.dto.ProgressTracker; import com.ddf.vodsystem.dto.ProgressTracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -12,6 +14,7 @@ import java.util.concurrent.CompletableFuture;
@Service @Service
public class RemuxService { public class RemuxService {
private static final Logger logger = LoggerFactory.getLogger(RemuxService.class);
private final CommandRunner commandRunner; private final CommandRunner commandRunner;
public RemuxService(CommandRunner commandRunner) { public RemuxService(CommandRunner commandRunner) {
@@ -23,19 +26,28 @@ public class RemuxService {
File outputFile, File outputFile,
ProgressTracker remuxProgress, ProgressTracker remuxProgress,
float length float length
) throws IOException, InterruptedException { ) {
List<String> command = List.of( try {
"ffmpeg", List<String> command = List.of(
"-progress", "pipe:1", "ffmpeg",
"-y", "-progress", "pipe:1",
"-i", inputFile.getAbsolutePath(), "-y",
"-c:v", "h264", "-i", inputFile.getAbsolutePath(),
"-c:a", "aac", "-c:v", "h264",
"-f", "mp4", "-c:a", "aac",
outputFile.getAbsolutePath() "-f", "mp4",
); outputFile.getAbsolutePath()
);
return CompletableFuture.completedFuture(commandRunner.run(command, line -> return CompletableFuture.completedFuture(commandRunner.run(command, line ->
commandRunner.setProgress(line, remuxProgress, length))); commandRunner.setProgress(line, remuxProgress, length)));
} catch (IOException e) {
logger.error("IO error on remux call: {}", e.toString());
return CompletableFuture.failedFuture(e);
} catch (InterruptedException e) {
logger.error("Thread error on remux call: {}", e.toString());
Thread.currentThread().interrupt();
return CompletableFuture.failedFuture(e);
}
} }
} }

View File

@@ -1,5 +1,6 @@
package com.ddf.vodsystem.services.media; package com.ddf.vodsystem.services.media;
import com.ddf.vodsystem.dto.CommandOutput;
import com.ddf.vodsystem.dto.ProgressTracker; import com.ddf.vodsystem.dto.ProgressTracker;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -54,12 +55,11 @@ public class StreamActionsService {
* to extract * to extract
* @param outputFile the destination file the trimmed section will be * @param outputFile the destination file the trimmed section will be
* written to; overwritten if it already exists * written to; overwritten if it already exists
* @return a {@link CompletableFuture} that completes with * @return a {@link CompletableFuture} that completes with the ffmpeg
* {@code outputFile} on success, or completes exceptionally on * command output on success, or completes exceptionally on failure
* failure
*/ */
@Async("ffmpegTaskExecutor") @Async("ffmpegTaskExecutor")
public CompletableFuture<Path> saveSection( public CompletableFuture<CommandOutput> saveSection(
List<Path> segments, List<Path> segments,
float trimOffset, float trimOffset,
float duration, float duration,
@@ -87,12 +87,19 @@ public class StreamActionsService {
logger.info("Saving section with ({} segments) to '{}'", logger.info("Saving section with ({} segments) to '{}'",
segments.size(), outputFile); segments.size(), outputFile);
commandRunner.run(command, line -> commandRunner.setProgress(line, progressTracker, duration)); CommandOutput output = commandRunner.run(command, line -> commandRunner.setProgress(line, progressTracker, duration));
progressTracker.markComplete();
return CompletableFuture.completedFuture(outputFile); progressTracker.markComplete();
} catch (Exception e) { return CompletableFuture.completedFuture(output);
logger.error("Failed to save section to '{}'", outputFile, e); } catch (IOException e) {
logger.error("IO error on saveSection call: {}", e.toString());
return CompletableFuture.failedFuture(e);
} catch (InterruptedException e) {
logger.error("Thread error on saveSection call: {}", e.toString());
Thread.currentThread().interrupt();
return CompletableFuture.failedFuture(e);
} catch (RuntimeException e) {
logger.error("Failed to save section to '{}': {}", outputFile, e.toString());
return CompletableFuture.failedFuture(e); return CompletableFuture.failedFuture(e);
} }
} }
@@ -130,4 +137,4 @@ public class StreamActionsService {
// nginx hls_fragment_naming system uses seconds; convert to ms // nginx hls_fragment_naming system uses seconds; convert to ms
return value < 1_000_000_000_000L ? value * 1000L : value; return value < 1_000_000_000_000L ? value * 1000L : value;
} }
} }

View File

@@ -1,5 +1,6 @@
package com.ddf.vodsystem.services.media; package com.ddf.vodsystem.services.media;
import com.ddf.vodsystem.dto.CommandOutput;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Async;
@@ -8,6 +9,7 @@ import org.springframework.stereotype.Service;
import java.io.IOException; import java.io.IOException;
import java.nio.file.Path; import java.nio.file.Path;
import java.util.List; import java.util.List;
import java.util.concurrent.CompletableFuture;
@Service @Service
public class ThumbnailService { public class ThumbnailService {
@@ -19,17 +21,25 @@ public class ThumbnailService {
} }
@Async("ffmpegTaskExecutor") @Async("ffmpegTaskExecutor")
public void createThumbnail(Path inputFile, Path outputFile, Float timeInVideo) throws IOException, InterruptedException { public CompletableFuture<CommandOutput> createThumbnail(Path inputFile, Path outputFile, Float timeInVideo) {
logger.info("Creating thumbnail at {} seconds", timeInVideo); logger.info("Creating thumbnail at {} seconds", timeInVideo);
List<String> command = List.of( try {
"ffmpeg", List<String> command = List.of(
"-ss", timeInVideo.toString(), "ffmpeg",
"-i", inputFile.toAbsolutePath().toString(), "-ss", timeInVideo.toString(),
"-frames:v", "1", "-i", inputFile.toAbsolutePath().toString(),
outputFile.toAbsolutePath().toString() "-frames:v", "1",
); outputFile.toAbsolutePath().toString()
);
commandRunner.run(command); CommandOutput output = commandRunner.run(command);
return CompletableFuture.completedFuture(output);
} catch (IOException e) {
return CompletableFuture.failedFuture(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return CompletableFuture.failedFuture(e);
}
} }
} }

View File

@@ -53,16 +53,19 @@ class MetadataServiceTest {
private ClipOptions runMetadata() throws Exception { private ClipOptions runMetadata() throws Exception {
return metadataService.getVideoMetadata(INPUT).get(); return metadataService.getVideoMetadata(INPUT).get();
} }
private void assertMetadataFailsWith(Class<? extends Throwable> exceptionType, String messageFragment) {
private void assertMetadataFailsWith(String messageFragment) {
assertThatThrownBy(() -> metadataService.getVideoMetadata(INPUT).get()) assertThatThrownBy(() -> metadataService.getVideoMetadata(INPUT).get())
.satisfies(thrown -> { .satisfies(thrown -> {
Throwable cause = thrown instanceof ExecutionException ? thrown.getCause() : thrown; Throwable cause = thrown instanceof ExecutionException ? thrown.getCause() : thrown;
assertThat(cause).isInstanceOf(FFMPEGException.class); assertThat(cause).isInstanceOf(exceptionType);
assertThat(cause.getMessage()).contains(messageFragment); assertThat(cause.getMessage()).contains(messageFragment);
}); });
} }
private void assertMetadataFailsWith(String messageFragment) {
assertMetadataFailsWith(FFMPEGException.class, messageFragment);
}
// --------------------------------------------------------------- // ---------------------------------------------------------------
// Happy path / parsing behavior // Happy path / parsing behavior
// --------------------------------------------------------------- // ---------------------------------------------------------------
@@ -246,10 +249,10 @@ class MetadataServiceTest {
} }
@Test @Test
void getVideoMetadata_commandThrowsIOException_throwsFFMPEGException() throws Exception { void getVideoMetadata_commandThrowsIOException_completesExceptionally() throws Exception {
when(commandRunner.run(anyList())).thenThrow(new IOException("ffprobe boom")); when(commandRunner.run(anyList())).thenThrow(new IOException("ffprobe boom"));
assertMetadataFailsWith("Error while getting video metadata"); assertMetadataFailsWith(IOException.class, "ffprobe boom");
} }
@Test @Test
@@ -291,7 +294,7 @@ class MetadataServiceTest {
void getVideoMetadata_plainIOException_doesNotSetThreadInterruptFlag() throws Exception { void getVideoMetadata_plainIOException_doesNotSetThreadInterruptFlag() throws Exception {
when(commandRunner.run(anyList())).thenThrow(new IOException("ffprobe boom")); when(commandRunner.run(anyList())).thenThrow(new IOException("ffprobe boom"));
assertMetadataFailsWith("Error while getting video metadata"); assertMetadataFailsWith(IOException.class, "ffprobe boom");
assertThat(Thread.currentThread().isInterrupted()) assertThat(Thread.currentThread().isInterrupted())
.as("a plain IOException must not leave the calling thread's interrupt flag set") .as("a plain IOException must not leave the calling thread's interrupt flag set")
@@ -302,7 +305,7 @@ class MetadataServiceTest {
void getVideoMetadata_interruptedException_setsThreadInterruptFlag() throws Exception { void getVideoMetadata_interruptedException_setsThreadInterruptFlag() throws Exception {
when(commandRunner.run(anyList())).thenThrow(new InterruptedException("interrupted")); when(commandRunner.run(anyList())).thenThrow(new InterruptedException("interrupted"));
assertMetadataFailsWith("Error while getting video metadata"); assertMetadataFailsWith(InterruptedException.class, "interrupted");
assertThat(Thread.currentThread().isInterrupted()).isTrue(); assertThat(Thread.currentThread().isInterrupted()).isTrue();
} }

View File

@@ -268,17 +268,18 @@ class StreamActionsServiceTest {
// --------------------------------------------------------------- // ---------------------------------------------------------------
@Test @Test
void saveSection_success_buildsExpectedCommandAndCompletesWithOutputFile() throws Exception { void saveSection_success_buildsExpectedCommandAndCompletesWithCommandOutput() throws Exception {
Path segmentA = Path.of("/tmp/streams/a.ts"); Path segmentA = Path.of("/tmp/streams/a.ts");
Path segmentB = Path.of("/tmp/streams/b.ts"); Path segmentB = Path.of("/tmp/streams/b.ts");
Path output = Path.of("/tmp/out/clip.mp4"); Path output = Path.of("/tmp/out/clip.mp4");
ProgressTracker progress = new ProgressTracker(); ProgressTracker progress = new ProgressTracker();
CommandOutput commandOutput = new CommandOutput();
when(commandRunner.run(anyList(), any())).thenReturn(new CommandOutput()); when(commandRunner.run(anyList(), any())).thenReturn(commandOutput);
var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress); var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress);
assertThat(future.get()).isEqualTo(output); assertThat(future.get()).isSameAs(commandOutput);
assertThat(progress.isComplete()).isTrue(); assertThat(progress.isComplete()).isTrue();
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@@ -310,14 +311,16 @@ class StreamActionsServiceTest {
} }
@Test @Test
void saveSection_emptySegmentList_throwsIllegalStateException() { void saveSection_emptySegmentList_completesExceptionallyWithIllegalStateException() {
Path output = Path.of("/tmp/out/clip.mp4"); Path output = Path.of("/tmp/out/clip.mp4");
ProgressTracker progress = new ProgressTracker(); ProgressTracker progress = new ProgressTracker();
var future = service.saveSection(List.of(), 0f, 10f, output, progress); var future = service.saveSection(List.of(), 0f, 10f, output, progress);
assertThatThrownBy(future::get).isInstanceOf(ExecutionException.class); assertThatThrownBy(future::get)
.isInstanceOf(ExecutionException.class)
.hasCauseInstanceOf(IllegalStateException.class);
assertThat(progress.isComplete()).isFalse();
} }
@Test @Test
@@ -326,6 +329,9 @@ class StreamActionsServiceTest {
var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress); var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress);
assertThatThrownBy(future::get).isInstanceOf(ExecutionException.class); assertThatThrownBy(future::get)
.isInstanceOf(ExecutionException.class)
.hasCauseInstanceOf(NullPointerException.class);
assertThat(progress.isComplete()).isFalse();
} }
} }