From d81202f905cc24691552564fd1d6d0c6831716cc Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Wed, 16 Sep 2026 19:49:04 +0100 Subject: [PATCH] fix(streaming): hitches in VoDs & Clips resulting from ffmpeg concatenation (#44) - Resolves #43 Reviewed-on: https://gitea.defaoite.dev/dylan/vod-system/pulls/44 --- docker-compose.dev.yml | 5 +- nginx/Dockerfile | 4 +- .../ddf/vodsystem/services/MediaService.java | 2 +- .../services/media/StreamActionsService.java | 114 ++++-------------- .../resources/application-local.properties | 4 +- .../media/StreamActionsServiceTest.java | 67 +++++----- 6 files changed, 60 insertions(+), 136 deletions(-) diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 6f9a5c9..4c4fd41 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -20,11 +20,12 @@ services: context: ./nginx dockerfile: Dockerfile container_name: vod_rtmp - network_mode: host env_file: - .env + extra_hosts: + - "host.docker.internal:host-gateway" environment: - - WEB_SERVER_URL=http://localhost:8080 + - WEB_SERVER_URL=http://host.docker.internal:8080 ports: - "1935:1935" volumes: diff --git a/nginx/Dockerfile b/nginx/Dockerfile index 08239a0..d27b7fa 100644 --- a/nginx/Dockerfile +++ b/nginx/Dockerfile @@ -1,6 +1,6 @@ -FROM tiangolo/nginx-rtmp +FROM tiangolo/nginx-rtmp:latest-2026-09-07 -RUN apt-get update && \ +RUN apt-get update -o Acquire::Check-Valid-Until=false && \ apt-get install -y --no-install-recommends gettext-base && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* diff --git a/src/main/java/com/ddf/vodsystem/services/MediaService.java b/src/main/java/com/ddf/vodsystem/services/MediaService.java index 712e0e0..2c90773 100644 --- a/src/main/java/com/ddf/vodsystem/services/MediaService.java +++ b/src/main/java/com/ddf/vodsystem/services/MediaService.java @@ -239,7 +239,7 @@ public class MediaService { onSuccess.run(); }).exceptionally(ex -> { job.setState(JobState.FAILED); - job.setErrorOutput(ex.getMessage()); + job.setErrorOutput("FFMPEG failed to save stream segments"); logger.error("Stream save job {} failed: {}", job.getUuid(), ex.getMessage()); return null; }); diff --git a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java index 553ce78..7619dab 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -16,6 +16,7 @@ import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; import java.util.stream.Stream; /** @@ -42,13 +43,14 @@ public class StreamActionsService { * A resolved section of a stream, ready to be muxed by {@link #saveSection}. * * @param segments the segments to concatenate, in playback order; never empty - * @param trimOffset seconds to trim from the start of {@code segments.getFirst()} - * so playback begins at the requested moment, or {@code 0} - * when the first segment already starts there - * @param duration seconds of footage to keep, measured from the trim point. - * When the requested start fell in a recording gap and the - * leading segment was dropped, this is measured from the - * later real start so the section still ends where requested + * @param trimOffset seconds between the start of {@code segments.getFirst()} and + * the requested start, or {@code 0} when the first segment + * already starts there. Segments are copied whole, so this is + * lead-in the section carries before the requested moment, not + * footage that gets trimmed away + * @param duration seconds of footage to keep, measured from the start of + * {@code segments.getFirst()} so the section ends at the + * requested end time with the lead-in included */ public record SegmentSelection(List segments, float trimOffset, float duration) {} @@ -78,8 +80,6 @@ public class StreamActionsService { long firstMs = parseTimestampMs(segments.getFirst()); float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); - // A first segment shorter than the trim offset means the requested start - // lands in a gap after it; drop it and begin at the next recorded footage. if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { segments.removeFirst(); if (segments.isEmpty()) { @@ -89,18 +89,19 @@ public class StreamActionsService { trimOffset = 0f; } - // Measure duration from where playback actually begins so a dropped - // leading segment shortens the section rather than overrunning endTime. - long effectiveStartMs = firstMs + (long) (trimOffset * 1000); - float duration = (endTime.toEpochMilli() - effectiveStartMs) / 1000f; + float duration = (endTime.toEpochMilli() - firstMs) / 1000f; return new SegmentSelection(segments, trimOffset, duration); } /** - * Muxes a {@link SegmentSelection} to {@code outputFile}, trimming the leading - * segment to {@code selection.trimOffset()} and keeping {@code selection.duration()} - * seconds of footage. + * Muxes a {@link SegmentSelection} to {@code outputFile}, keeping + * {@code selection.duration()} seconds of footage. + *

+ * Segments are concatenated byte-wise through ffmpeg's {@code concat:} protocol, + * which preserves the continuous timestamps nginx writes across fragments. The + * leading segment is copied whole rather than trimmed, so playback begins up to + * {@code selection.trimOffset()} seconds before the requested moment. *

* Runs asynchronously; failures (process-launch errors, ffmpeg failures, * interruption) complete the returned future exceptionally rather than being @@ -131,37 +132,19 @@ public class StreamActionsService { "trimOffset must be >= 0 and duration must be > 0 (got " + trimOffset + ", " + duration + ")")); } - Path headFragment = null; - Path concatList = null; try { - // Only the first segment is trimmed, so re-encode just that fragment and - // stream-copy the rest. A zero offset means the first segment already - // starts at the requested point, so re-encoding it would be wasted work. - List fragments; - if (trimOffset > 0) { - headFragment = Files.createTempFile("reencode-section-", ".ts"); - encodeHead(segments.getFirst(), trimOffset, headFragment); - - fragments = new ArrayList<>(); - fragments.add(headFragment); - fragments.addAll(segments.subList(1, segments.size())); - } else { - fragments = segments; - } - - // The concat demuxer (unlike the concat: byte protocol) rebases each - // fragment's timestamps to run continuously across the joins, absorbing - // the discontinuity between the PTS-0 head and the copied fragments. - concatList = writeConcatList(fragments); + String concatInput = "concat:" + segments.stream() + .map(p -> p.toAbsolutePath().toString()) + .collect(Collectors.joining("|")); List command = List.of( "ffmpeg", "-y", - "-f", "concat", - "-safe", "0", - "-i", concatList.toAbsolutePath().toString(), + "-progress", "pipe:1", + "-i", concatInput, "-t", String.valueOf(duration), "-c", "copy", + "-avoid_negative_ts", "make_zero", outputFile.toAbsolutePath().toString() ); @@ -180,9 +163,6 @@ public class StreamActionsService { } catch (RuntimeException e) { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); - } finally { - deleteTempQuietly(headFragment); - deleteTempQuietly(concatList); } } @@ -257,52 +237,4 @@ public class StreamActionsService { throw new IOException("Failed to probe segment duration: " + segment, e); } } - - private void encodeHead(Path firstSegment, float trimOffset, Path head) - throws IOException, InterruptedException { - - List command = List.of( - "ffmpeg", - "-y", - "-i", firstSegment.toAbsolutePath().toString(), - "-ss", String.valueOf(trimOffset), - "-c:v", "libx264", - "-preset", "veryfast", - "-crf", "20", - "-pix_fmt", "yuv420p", - "-c:a", "aac", - "-b:a", "160k", - "-muxdelay", "0", - "-muxpreload", "0", - "-f", "mpegts", - head.toAbsolutePath().toString() - ); - - logger.debug("Re-encoding head fragment '{}' from {}s", firstSegment, trimOffset); - commandRunner.run(command, line -> { /* head pass is short; no progress reporting */ }); - - if (!Files.exists(head) || Files.size(head) == 0) { - throw new IOException("Head re-encode produced no output — trimOffset " + trimOffset - + "s may exceed the duration of " + firstSegment); - } - } - - private Path writeConcatList(List fragments) throws IOException { - Path listFile = Files.createTempFile("concat-list-", ".txt"); - List lines = fragments.stream() - .map(p -> "file '" + p.toAbsolutePath().toString().replace("'", "'\\''") + "'") - .toList(); - Files.write(listFile, lines); - return listFile; - } - - private void deleteTempQuietly(Path tempFile) { - if (tempFile != null) { - try { - Files.deleteIfExists(tempFile); - } catch (IOException e) { - logger.warn("Failed to delete temp file '{}': {}", tempFile, e.toString()); - } - } - } } diff --git a/src/main/resources/application-local.properties b/src/main/resources/application-local.properties index ec2e735..cd797d7 100644 --- a/src/main/resources/application-local.properties +++ b/src/main/resources/application-local.properties @@ -3,4 +3,6 @@ spring.docker.compose.file=docker-compose.dev.yml media.temp.inputs=./videos/inputs media.temp.outputs=./videos/outputs media.users=${MEDIA_DATA} -media.streams=${STREAM_DATA} \ No newline at end of file +media.streams=${STREAM_DATA} + +logging.level.com.ddf.vodsystem=DEBUG \ No newline at end of file diff --git a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java index eef3327..89a5a18 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -57,23 +57,6 @@ class StreamActionsServiceTest { when(metadataService.getVideoMetadata(any())).thenReturn(CompletableFuture.completedFuture(meta)); } - /** - * Stubs the two-arg {@link CommandRunner#run} so the head re-encode pass - * actually writes a non-empty file at its output path — {@code encodeHead} - * verifies its output exists and is non-empty before continuing. The concat - * pass (no {@code mpegts} muxer) is left untouched. Returns {@code output} - * for both invocations. - */ - private void stubRunnerProducingHead(CommandOutput output) throws IOException, InterruptedException { - when(commandRunner.run(anyList(), any())).thenAnswer(inv -> { - List cmd = inv.getArgument(0); - if (cmd.contains("mpegts")) { - Files.writeString(Path.of(cmd.getLast()), "x"); - } - return output; - }); - } - // --------------------------------------------------------------- // selectSegments: missing / invalid directories // --------------------------------------------------------------- @@ -247,8 +230,10 @@ class StreamActionsServiceTest { assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) .containsExactly("30.ts", "33.ts"); assertThat(selection.trimOffset()).isEqualTo(1.5f); - // effective start equals the requested 31.5s, so duration runs to endTime: 40 - 31.5 - assertThat(selection.duration()).isEqualTo(8.5f); + // 30.ts is copied whole, so the section starts at 30s rather than the + // requested 31.5s; duration runs from there to endTime: 40 - 30. The + // 1.5s trimOffset is carried as lead-in, not cut away. + assertThat(selection.duration()).isEqualTo(10f); } @Test @@ -355,7 +340,7 @@ class StreamActionsServiceTest { ProgressTracker progress = new ProgressTracker(); CommandOutput commandOutput = new CommandOutput(); - stubRunnerProducingHead(commandOutput); + when(commandRunner.run(anyList(), any())).thenReturn(commandOutput); var future = service.saveSection( new StreamActionsService.SegmentSelection(List.of(segmentA, segmentB), 5.0f, 12.0f), output, progress); @@ -363,24 +348,29 @@ class StreamActionsServiceTest { assertThat(future.get()).isSameAs(commandOutput); assertThat(progress.isComplete()).isTrue(); - // With a non-zero trim offset saveSection issues two ffmpeg passes: the - // head re-encode (trims the first segment via -ss) followed by the - // concat-demuxer stream copy (-f concat / -t / -c copy). + // The section is one stream-copy pass over the byte-concatenated segments: + // the concat: protocol joins them without rebasing the timestamps nginx + // wrote across fragments, which is what keeps the joins seamless. @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); - Mockito.verify(commandRunner, Mockito.times(2)).run(captor.capture(), any()); - List headCommand = captor.getAllValues().get(0); - List concatCommand = captor.getAllValues().get(1); + Mockito.verify(commandRunner, Mockito.times(1)).run(captor.capture(), any()); + List command = captor.getValue(); - assertThat(headCommand).containsSequence("-ss", "5.0"); - assertThat(concatCommand).containsSequence("-f", "concat"); - assertThat(concatCommand).containsSequence("-t", "12.0"); - assertThat(concatCommand).containsSequence("-c", "copy"); - assertThat(concatCommand).endsWith(output.toAbsolutePath().toString()); + assertThat(command).containsSequence( + "-i", "concat:" + segmentA.toAbsolutePath() + "|" + segmentB.toAbsolutePath()); + assertThat(command).containsSequence("-t", "12.0"); + assertThat(command).containsSequence("-c", "copy"); + assertThat(command).containsSequence("-avoid_negative_ts", "make_zero"); + assertThat(command).endsWith(output.toAbsolutePath().toString()); + + // The leading segment is copied whole, so nothing seeks and the concat + // demuxer (-f concat) is deliberately not used. + assertThat(command).doesNotContain("-ss"); + assertThat(command).doesNotContainSequence("-f", "concat"); } @Test - void saveSection_zeroTrimOffset_skipsHeadReencodeAndRunsSingleConcatPass() throws Exception { + void saveSection_zeroTrimOffset_buildsTheSameSingleCopyPass() throws Exception { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); CommandOutput commandOutput = new CommandOutput(); @@ -394,14 +384,14 @@ class StreamActionsServiceTest { assertThat(future.get()).isSameAs(commandOutput); assertThat(progress.isComplete()).isTrue(); - // trimOffset 0 means the first segment already starts at the requested point: - // no head re-encode pass, just a single concat-demuxer copy (no -ss). + // trimOffset no longer changes the command shape: with or without lead-in + // the section is the same single copy pass over the joined segments. @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); Mockito.verify(commandRunner, Mockito.times(1)).run(captor.capture(), any()); List command = captor.getValue(); - assertThat(command).containsSequence("-f", "concat"); + assertThat(command).containsSequence("-i", "concat:/tmp/a.ts|/tmp/b.ts"); assertThat(command).containsSequence("-t", "10.0"); assertThat(command).doesNotContain("-ss"); } @@ -411,8 +401,7 @@ class StreamActionsServiceTest { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - // With a non-zero trim offset the head re-encode is the first ffmpeg pass; - // its failure must surface. + // The copy pass is the only ffmpeg invocation, so its failure must surface. when(commandRunner.run(anyList(), any())).thenThrow(new IOException("ffmpeg boom")); var future = service.saveSection( @@ -442,8 +431,8 @@ class StreamActionsServiceTest { void saveSection_nullOutputFile_completesExceptionally() { ProgressTracker progress = new ProgressTracker(); - // trimOffset 0 skips the head pass; building the concat command dereferences - // the null output path, so no ffmpeg pass ever runs. + // Building the command dereferences the null output path, so no ffmpeg + // pass ever runs. var future = service.saveSection( new StreamActionsService.SegmentSelection(List.of(Path.of("/tmp/a.ts")), 0f, 10f), null, progress);