From aabeae5e8b3e02e93a19e800845796792516c4c1 Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Sat, 25 Jul 2026 13:44:34 +0100 Subject: [PATCH 1/6] fix incorrect hls segment handling --- frontend/package-lock.json | 18 ++++++ frontend/package.json | 1 + .../ddf/vodsystem/services/MediaService.java | 2 +- .../services/media/StreamActionsService.java | 55 ++++++++++++++----- .../media/StreamActionsServiceTest.java | 40 ++++++++------ 5 files changed, 82 insertions(+), 34 deletions(-) diff --git a/frontend/package-lock.json b/frontend/package-lock.json index d3823e5..e0c31a4 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -25,6 +25,7 @@ "devDependencies": { "@eslint/js": "^9.25.0", "@tailwindcss/vite": "^4.3.3", + "@types/node": "^26.1.1", "@types/react": "^19.1.2", "@types/react-dom": "^19.1.2", "@vitejs/plugin-react": "^4.4.1", @@ -1888,6 +1889,16 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/node": { + "version": "26.1.1", + "resolved": "https://registry.npmjs.org/@types/node/-/node-26.1.1.tgz", + "integrity": "sha512-nxAkRSVkN1Y0JC1W8ky/fTfkGsMmcrRsbx+3XoZE+rMOX71kLYTV7fLXpqud1GpbpP5TuffXFqfX7fH2GgZREw==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~8.3.0" + } + }, "node_modules/@types/react": { "version": "19.2.17", "resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.17.tgz", @@ -5304,6 +5315,13 @@ "node": ">=10" } }, + "node_modules/undici-types": { + "version": "8.3.0", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-8.3.0.tgz", + "integrity": "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ==", + "dev": true, + "license": "MIT" + }, "node_modules/update-browserslist-db": { "version": "1.2.3", "resolved": "https://registry.npmjs.org/update-browserslist-db/-/update-browserslist-db-1.2.3.tgz", diff --git a/frontend/package.json b/frontend/package.json index 9fb4eda..1799591 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -28,6 +28,7 @@ "devDependencies": { "@eslint/js": "^9.25.0", "@tailwindcss/vite": "^4.3.3", + "@types/node": "^26.1.1", "@types/react": "^19.1.2", "@types/react-dom": "^19.1.2", "@vitejs/plugin-react": "^4.4.1", diff --git a/src/main/java/com/ddf/vodsystem/services/MediaService.java b/src/main/java/com/ddf/vodsystem/services/MediaService.java index 56f3ba1..195d89b 100644 --- a/src/main/java/com/ddf/vodsystem/services/MediaService.java +++ b/src/main/java/com/ddf/vodsystem/services/MediaService.java @@ -238,7 +238,7 @@ public class MediaService { return new SegmentContext(segments, trimOffset); } - private Job dispatchSaveJob(Job job, SegmentContext ctx, float duration, Path outputFile, Runnable onSuccess) { + private Job dispatchSaveJob(Job job, SegmentContext ctx, float duration, Path outputFile, Runnable onSuccess) throws IOException { job.setState(JobState.PROCESSING); streamActionsService.saveSection(ctx.segments(), ctx.trimOffset(), duration, outputFile, job.getProgressTracker()) .thenRun(() -> { 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 532b55a..2c81c6a 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -11,6 +11,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; +import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -31,8 +32,6 @@ public class StreamActionsService { private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class); private final CommandRunner commandRunner; - private static final int HLS_SEGMENT_LENGTH = 3; - public StreamActionsService(CommandRunner commandRunner) { this.commandRunner = commandRunner; } @@ -65,21 +64,33 @@ public class StreamActionsService { float duration, Path outputFile, ProgressTracker progressTracker - ) { + ) throws IOException { + if (segments.isEmpty()) { + return CompletableFuture.failedFuture(new IllegalArgumentException("Segment list cannot be empty")); + } + + Path firstSegment = segments.getFirst(); + Path trimmedFirst = Files.createTempFile("trimmed-first-", ".ts"); try { - String concatInput = segments.stream() + // First segment must be trimmed manually due to keyframe issues with the concat protocol + List trimCommand = List.of( + "ffmpeg", "-y", + "-ss", String.valueOf(trimOffset), + "-i", firstSegment.toAbsolutePath().toString(), + "-c", "copy", + trimmedFirst.toAbsolutePath().toString() + ); + commandRunner.run(trimCommand); + + String concatInput = Stream.concat(Stream.of(trimmedFirst), segments.stream().skip(1)) .map(p -> p.toAbsolutePath().toString()) .collect(Collectors.joining("|")); - if (concatInput.isEmpty()) { - throw new IllegalStateException("Must have at least one segment"); - } - List command = List.of( "ffmpeg", "-y", - "-i", "concat:" + concatInput, "-ss", String.valueOf(trimOffset), + "-i", "concat:" + concatInput, "-t", String.valueOf(duration), "-c", "copy", outputFile.toAbsolutePath().toString() @@ -101,6 +112,8 @@ public class StreamActionsService { } catch (RuntimeException e) { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); + } finally { + Files.deleteIfExists(trimmedFirst); } } @@ -112,17 +125,29 @@ public class StreamActionsService { throw new IllegalArgumentException("Start time must be before end time"); } + List results = new ArrayList<>(); + try (Stream files = Files.list(streamDirectory)) { - return files + List streamFiles = files .filter(p -> p.getFileName().toString().endsWith(".ts")) .filter(Files::isRegularFile) - .filter(p -> { - long segMs = parseTimestampMs(p); - return segMs < endMs && (segMs + HLS_SEGMENT_LENGTH * 1000) > startMs; - }) .sorted(Comparator.comparingLong(this::parseTimestampMs)) - .collect(Collectors.toList()); + .toList(); + + for (int i = 0; i < streamFiles.size(); i++) { + Path streamFile = streamFiles.get(i); + long segmentStart = parseTimestampMs(streamFile); + long segmentEnd = (i + 1 < streamFiles.size()) + ? parseTimestampMs(streamFiles.get(i + 1)) + : Long.MAX_VALUE; + + if (segmentStart < endMs && segmentEnd > startMs) { + results.add(streamFile); + } + } } + + return results; } public long parseTimestampMs(Path path) { 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 16baba1..cb0d563 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -124,13 +124,13 @@ class StreamActionsServiceTest { @Test void getSegmentsInRange_segmentEndsExactlyAtRangeStart_isExcluded() throws IOException { - // "0.ts" spans [0, 3000)ms. A request starting exactly at 3000ms should not - // pull in a segment that ended the instant before. createSegment("0.ts"); + createSegment("3.ts"); List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(3000), Instant.ofEpochMilli(10_000)); - assertThat(result).isEmpty(); + assertThat(result).extracting(p -> p.getFileName().toString()) + .containsExactly("3.ts"); } @Test @@ -170,16 +170,6 @@ class StreamActionsServiceTest { } - @Test - void getSegmentsInRange_segmentsWhollyOutsideRange_areExcluded() throws IOException { - createSegment("0.ts"); // [0, 3000) - createSegment("100.ts"); // [100000, 103000) - - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(50_000), Instant.ofEpochMilli(60_000)); - - assertThat(result).isEmpty(); - } - @Test void getSegmentsInRange_segmentsOutOfOrder_returnedInOrder() throws IOException { createSegment("20.ts"); @@ -219,6 +209,20 @@ class StreamActionsServiceTest { .containsExactly("10.ts", "20.ts", "30.ts"); } + @Test + void getSegmentsInRange_gapInStream_returnsTheGap() throws IOException { + createSegment("20.ts"); + createSegment("30.ts"); + // user stops streaming + createSegment("100.ts"); + createSegment("110.ts"); + + List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(70_000), Instant.ofEpochMilli(111_000)); + + assertThat(result).extracting(p -> p.getFileName().toString()) + .containsExactly("30.ts", "100.ts", "110.ts"); + } + // --------------------------------------------------------------- // parseTimestampMs: unit-based conversion // --------------------------------------------------------------- @@ -275,6 +279,7 @@ class StreamActionsServiceTest { ProgressTracker progress = new ProgressTracker(); CommandOutput commandOutput = new CommandOutput(); + when(commandRunner.run(anyList())).thenReturn(commandOutput); when(commandRunner.run(anyList(), any())).thenReturn(commandOutput); var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress); @@ -287,8 +292,6 @@ class StreamActionsServiceTest { Mockito.verify(commandRunner).run(captor.capture(), any()); List command = captor.getValue(); - assertThat(command).containsSequence("-i", - "concat:" + segmentA.toAbsolutePath() + "|" + segmentB.toAbsolutePath()); assertThat(command).containsSequence("-ss", "5.0"); assertThat(command).containsSequence("-t", "12.0"); assertThat(command).containsSequence("-c", "copy"); @@ -300,6 +303,7 @@ class StreamActionsServiceTest { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); + when(commandRunner.run(anyList())).thenReturn(new CommandOutput()); when(commandRunner.run(anyList(), any())).thenThrow(new IOException("ffmpeg boom")); var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, output, progress); @@ -311,7 +315,7 @@ class StreamActionsServiceTest { } @Test - void saveSection_emptySegmentList_completesExceptionallyWithIllegalStateException() { + void saveSection_emptySegmentList_completesExceptionallyWithIllegalArgumentException() throws IOException { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); @@ -319,12 +323,12 @@ class StreamActionsServiceTest { assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) - .hasCauseInstanceOf(IllegalStateException.class); + .hasCauseInstanceOf(IllegalArgumentException.class); assertThat(progress.isComplete()).isFalse(); } @Test - void saveSection_nullOutputFile_completesExceptionally() { + void saveSection_nullOutputFile_completesExceptionally() throws IOException { ProgressTracker progress = new ProgressTracker(); var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress); -- 2.49.1 From fce366125b1a306e1798157a15d461e953ccea0a Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Sun, 26 Jul 2026 14:36:57 +0100 Subject: [PATCH 2/6] re-encode first segment --- .../services/media/StreamActionsService.java | 71 ++++++++++++++----- 1 file changed, 55 insertions(+), 16 deletions(-) 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 2c81c6a..aeb8339 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -64,32 +64,31 @@ public class StreamActionsService { float duration, Path outputFile, ProgressTracker progressTracker - ) throws IOException { + ) { + if (segments.isEmpty()) { return CompletableFuture.failedFuture(new IllegalArgumentException("Segment list cannot be empty")); } - Path firstSegment = segments.getFirst(); - Path trimmedFirst = Files.createTempFile("trimmed-first-", ".ts"); - try { - // First segment must be trimmed manually due to keyframe issues with the concat protocol - List trimCommand = List.of( - "ffmpeg", "-y", - "-ss", String.valueOf(trimOffset), - "-i", firstSegment.toAbsolutePath().toString(), - "-c", "copy", - trimmedFirst.toAbsolutePath().toString() - ); - commandRunner.run(trimCommand); + if (trimOffset < 0 || duration <= 0) { + return CompletableFuture.failedFuture(new IllegalArgumentException( + "trimOffset must be >= 0 and duration must be > 0 (got " + trimOffset + ", " + duration + ")")); + } - String concatInput = Stream.concat(Stream.of(trimmedFirst), segments.stream().skip(1)) + Path headSegment = null; + try { + // Unique per invocation so concurrent saves by different users never + // collide on the same head fragment; cleaned up in the finally block. + headSegment = Files.createTempFile("reencode-section-", ".ts"); + encodeHead(segments.getFirst(), trimOffset, headSegment); + + String concatInput = Stream.concat(Stream.of(headSegment), segments.stream().skip(1)) .map(p -> p.toAbsolutePath().toString()) .collect(Collectors.joining("|")); List command = List.of( "ffmpeg", "-y", - "-ss", String.valueOf(trimOffset), "-i", "concat:" + concatInput, "-t", String.valueOf(duration), "-c", "copy", @@ -113,7 +112,13 @@ public class StreamActionsService { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); } finally { - Files.deleteIfExists(trimmedFirst); + if (headSegment != null) { + try { + Files.deleteIfExists(headSegment); + } catch (IOException e) { + logger.warn("Failed to delete temp head fragment '{}': {}", headSegment, e.toString()); + } + } } } @@ -162,4 +167,38 @@ public class StreamActionsService { // nginx hls_fragment_naming system uses seconds; convert to ms return value < 1_000_000_000_000L ? value * 1000L : value; } + + /** + * Re-encodes {@code firstSegment} from {@code trimOffset} to its end, + * writing an MPEG-TS fragment that can be stream-copy concatenated with + * the untouched fragments that follow it. + */ + 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; not worth 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); + } + } } -- 2.49.1 From 2378e421996546df97708896f3239335a16cd945 Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Sun, 26 Jul 2026 18:56:45 +0100 Subject: [PATCH 3/6] fix broken unit tests --- .../media/StreamActionsServiceTest.java | 42 ++++++++++++++----- 1 file changed, 32 insertions(+), 10 deletions(-) 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 cb0d563..0d9e02c 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -45,6 +45,23 @@ class StreamActionsServiceTest { Files.createFile(tempDir.resolve(name)); } + /** + * 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.get(cmd.size() - 1)), "x"); + } + return output; + }); + } + // --------------------------------------------------------------- // getSegmentsInRange: missing / invalid directories // --------------------------------------------------------------- @@ -279,23 +296,25 @@ class StreamActionsServiceTest { ProgressTracker progress = new ProgressTracker(); CommandOutput commandOutput = new CommandOutput(); - when(commandRunner.run(anyList())).thenReturn(commandOutput); - when(commandRunner.run(anyList(), any())).thenReturn(commandOutput); + stubRunnerProducingHead(commandOutput); var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress); assertThat(future.get()).isSameAs(commandOutput); assertThat(progress.isComplete()).isTrue(); + // saveSection issues two ffmpeg passes: the head re-encode (trims the + // first segment via -ss) followed by the stream-copy concat (-t/-c copy). @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); - Mockito.verify(commandRunner).run(captor.capture(), any()); - List command = captor.getValue(); + Mockito.verify(commandRunner, Mockito.times(2)).run(captor.capture(), any()); + List headCommand = captor.getAllValues().get(0); + List concatCommand = captor.getAllValues().get(1); - assertThat(command).containsSequence("-ss", "5.0"); - assertThat(command).containsSequence("-t", "12.0"); - assertThat(command).containsSequence("-c", "copy"); - assertThat(command).endsWith(output.toAbsolutePath().toString()); + assertThat(headCommand).containsSequence("-ss", "5.0"); + assertThat(concatCommand).containsSequence("-t", "12.0"); + assertThat(concatCommand).containsSequence("-c", "copy"); + assertThat(concatCommand).endsWith(output.toAbsolutePath().toString()); } @Test @@ -303,7 +322,7 @@ class StreamActionsServiceTest { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - when(commandRunner.run(anyList())).thenReturn(new CommandOutput()); + // The head re-encode is the first ffmpeg pass; its failure must surface. when(commandRunner.run(anyList(), any())).thenThrow(new IOException("ffmpeg boom")); var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, output, progress); @@ -328,9 +347,12 @@ class StreamActionsServiceTest { } @Test - void saveSection_nullOutputFile_completesExceptionally() throws IOException { + void saveSection_nullOutputFile_completesExceptionally() throws Exception { ProgressTracker progress = new ProgressTracker(); + // Let the head pass succeed so execution reaches the null output file. + stubRunnerProducingHead(new CommandOutput()); + var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress); assertThatThrownBy(future::get) -- 2.49.1 From 22d0a61a1517592d3670fd7deb388d5e8465646b Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Sun, 26 Jul 2026 22:19:08 +0100 Subject: [PATCH 4/6] fix edge case where first segment is longer than offset --- .../ddf/vodsystem/services/MediaService.java | 18 ++--- .../services/media/StreamActionsService.java | 59 +++++++++++++- .../media/StreamActionsServiceTest.java | 79 ++++++++++++++++++- 3 files changed, 139 insertions(+), 17 deletions(-) diff --git a/src/main/java/com/ddf/vodsystem/services/MediaService.java b/src/main/java/com/ddf/vodsystem/services/MediaService.java index 195d89b..57deb71 100644 --- a/src/main/java/com/ddf/vodsystem/services/MediaService.java +++ b/src/main/java/com/ddf/vodsystem/services/MediaService.java @@ -133,7 +133,7 @@ public class MediaService { .orElseThrow(() -> new NotAuthenticated("User is not authenticated")); Job job = jobRegistryService.generateJob(); - SegmentContext ctx = resolveSegments(user, startTime, endTime); + StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); float duration = (endTime.toEpochMilli() - startTime.toEpochMilli()) / 1000f; Path outputFile = directoryService.getVodsDir(user.getId()).resolve(UUID.randomUUID() + ".mp4"); @@ -211,7 +211,7 @@ public class MediaService { .orElseThrow(() -> new NotAuthenticated("User is not authenticated")); Job job = jobRegistryService.generateJob(); - SegmentContext ctx = resolveSegments(user, startTime, endTime); + StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); Path outputFile = directoryService.getTempOutputDir().resolve(UUID.randomUUID() + ".mp4"); return dispatchSaveJob(job, ctx, duration, outputFile, () -> @@ -225,20 +225,12 @@ public class MediaService { ); } - private record SegmentContext(List segments, float trimOffset) {} - - private SegmentContext resolveSegments(User user, Instant startTime, Instant endTime) throws IOException { + private StreamActionsService.SegmentSelection resolveSegments(User user, Instant startTime, Instant endTime) throws IOException { Path streamDir = directoryService.getStreamDir(user.getStreamKey()); - List segments = streamActionsService.getSegmentsInRange(streamDir, startTime, endTime); - if (segments.isEmpty()) { - throw new IllegalArgumentException("No stream segments found in the given time range"); - } - long firstMs = streamActionsService.parseTimestampMs(segments.getFirst()); - float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); - return new SegmentContext(segments, trimOffset); + return streamActionsService.selectSegments(streamDir, startTime, endTime); } - private Job dispatchSaveJob(Job job, SegmentContext ctx, float duration, Path outputFile, Runnable onSuccess) throws IOException { + private Job dispatchSaveJob(Job job, StreamActionsService.SegmentSelection ctx, float duration, Path outputFile, Runnable onSuccess) { job.setState(JobState.PROCESSING); streamActionsService.saveSection(ctx.segments(), ctx.trimOffset(), duration, outputFile, job.getProgressTracker()) .thenRun(() -> { 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 aeb8339..dcd6664 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -15,6 +15,7 @@ import java.util.ArrayList; 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; @@ -31,9 +32,11 @@ import java.util.stream.Stream; public class StreamActionsService { private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class); private final CommandRunner commandRunner; + private final MetadataService metadataService; - public StreamActionsService(CommandRunner commandRunner) { + public StreamActionsService(CommandRunner commandRunner, MetadataService metadataService) { this.commandRunner = commandRunner; + this.metadataService = metadataService; } /** @@ -141,6 +144,7 @@ public class StreamActionsService { for (int i = 0; i < streamFiles.size(); i++) { Path streamFile = streamFiles.get(i); + long segmentStart = parseTimestampMs(streamFile); long segmentEnd = (i + 1 < streamFiles.size()) ? parseTimestampMs(streamFiles.get(i + 1)) @@ -155,6 +159,59 @@ public class StreamActionsService { return results; } + /** + * The segments backing a save and the offset into the first one at which + * the requested section begins. + * + * @param segments the ordered segments to concatenate, in playback order + * @param trimOffset the offset, in seconds, into {@code segments.getFirst()} + * at which the requested section starts + */ + public record SegmentSelection(List segments, float trimOffset) {} + + /** + * Selects the recorded {@code .ts} segments overlapping {@code [startTime, endTime)} + * and computes the trim offset into the first segment. + *

+ * If {@code startTime} falls in a stream gap after the earliest overlapping segment + * (so that segment holds no footage for the requested start), the segment is dropped + * and the selection begins at the first genuinely-recorded footage instead. + * + * @throws IllegalArgumentException if no segments hold footage for the requested range + * @throws IOException if listing the stream directory or probing a segment fails + */ + public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + List segments = getSegmentsInRange(streamDirectory, startTime, endTime); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No stream segments found in the given time range"); + } + + long firstMs = parseTimestampMs(segments.getFirst()); + float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); + + // if the first segment is longer than the trim-offset, then ggs + if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { + segments.removeFirst(); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No recorded footage at the requested start time"); + } + trimOffset = 0f; + } + + return new SegmentSelection(segments, trimOffset); + } + + private float probeDurationSeconds(Path segment) throws IOException { + try { + return metadataService.getVideoMetadata(segment).get().getDuration(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while probing segment duration: " + segment, e); + } catch (ExecutionException e) { + throw new IOException("Failed to probe segment duration: " + segment, e); + } + } + public long parseTimestampMs(Path path) { String fullName = path.getFileName().toString(); 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 0d9e02c..2afe492 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -1,5 +1,6 @@ package com.ddf.vodsystem.services.media; +import com.ddf.vodsystem.dto.ClipOptions; import com.ddf.vodsystem.dto.CommandOutput; import com.ddf.vodsystem.dto.ProgressTracker; import org.junit.jupiter.api.BeforeEach; @@ -17,6 +18,7 @@ import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.time.Instant; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import static org.assertj.core.api.Assertions.assertThat; @@ -31,6 +33,9 @@ class StreamActionsServiceTest { @Mock private CommandRunner commandRunner; + @Mock + private MetadataService metadataService; + private StreamActionsService service; @TempDir @@ -38,13 +43,20 @@ class StreamActionsServiceTest { @BeforeEach void setUp() { - service = new StreamActionsService(commandRunner); + service = new StreamActionsService(commandRunner, metadataService); } private void createSegment(String name) throws IOException { Files.createFile(tempDir.resolve(name)); } + /** Stubs the (async) metadata probe so {@code segments.getFirst()} reports {@code seconds}. */ + private void stubHeadDuration(float seconds) { + ClipOptions meta = new ClipOptions(); + meta.setDuration(seconds); + 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} @@ -56,7 +68,7 @@ class StreamActionsServiceTest { when(commandRunner.run(anyList(), any())).thenAnswer(inv -> { List cmd = inv.getArgument(0); if (cmd.contains("mpegts")) { - Files.writeString(Path.of(cmd.get(cmd.size() - 1)), "x"); + Files.writeString(Path.of(cmd.getLast()), "x"); } return output; }); @@ -240,6 +252,67 @@ class StreamActionsServiceTest { .containsExactly("30.ts", "100.ts", "110.ts"); } + // --------------------------------------------------------------- + // selectSegments: head selection and trim-offset + // --------------------------------------------------------------- + + @Test + void selectSegments_startWithinFirstSegment_keepsItAndComputesOffset() throws IOException { + createSegment("30.ts"); + createSegment("33.ts"); + stubHeadDuration(3f); // 30.ts spans [30s, 33s) + + var selection = service.selectSegments(tempDir, Instant.ofEpochMilli(31_500), Instant.ofEpochMilli(40_000)); + + assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) + .containsExactly("30.ts", "33.ts"); + assertThat(selection.trimOffset()).isEqualTo(1.5f); + } + + @Test + void selectSegments_startInGapAfterFirstSegment_dropsItAndStartsAtNext() throws IOException { + createSegment("30.ts"); + // user stops streaming; a gap follows 30.ts + createSegment("100.ts"); + createSegment("110.ts"); + stubHeadDuration(3f); // 30.ts only holds [30s, 33s); requested start 70s is past it + + var selection = service.selectSegments(tempDir, Instant.ofEpochMilli(70_000), Instant.ofEpochMilli(111_000)); + + assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) + .containsExactly("100.ts", "110.ts"); + assertThat(selection.trimOffset()).isZero(); + } + + @Test + void selectSegments_startPastOnlySegment_throwsIllegalArgument() throws IOException { + createSegment("0.ts"); + stubHeadDuration(3f); // 0.ts holds [0s, 3s); requested start 50s has no footage + + assertThatThrownBy(() -> + service.selectSegments(tempDir, Instant.ofEpochMilli(50_000), Instant.ofEpochMilli(60_000))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void selectSegments_noSegmentsInRange_throwsIllegalArgument() { + assertThatThrownBy(() -> + service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void selectSegments_probeFailure_surfacesAsIOException() throws IOException { + createSegment("30.ts"); + createSegment("100.ts"); + when(metadataService.getVideoMetadata(any())) + .thenReturn(CompletableFuture.failedFuture(new IOException("ffprobe boom"))); + + assertThatThrownBy(() -> + service.selectSegments(tempDir, Instant.ofEpochMilli(70_000), Instant.ofEpochMilli(111_000))) + .isInstanceOf(IOException.class); + } + // --------------------------------------------------------------- // parseTimestampMs: unit-based conversion // --------------------------------------------------------------- @@ -334,7 +407,7 @@ class StreamActionsServiceTest { } @Test - void saveSection_emptySegmentList_completesExceptionallyWithIllegalArgumentException() throws IOException { + void saveSection_emptySegmentList_completesExceptionallyWithIllegalArgumentException() { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); -- 2.49.1 From 70f29f257c703844dca46da351adb8911ba8c69c Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Mon, 27 Jul 2026 22:10:07 +0100 Subject: [PATCH 5/6] first segment's duration is probed to exclude it if needed --- .../services/media/StreamActionsService.java | 4 +- .../media/StreamActionsServiceTest.java | 96 ++++++++----------- 2 files changed, 40 insertions(+), 60 deletions(-) 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 dcd6664..ed37183 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -125,7 +125,7 @@ public class StreamActionsService { } } - public List getSegmentsInRange(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + private List getSegmentsInRange(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { long startMs = startTime.toEpochMilli(); long endMs = endTime.toEpochMilli(); @@ -189,7 +189,7 @@ public class StreamActionsService { long firstMs = parseTimestampMs(segments.getFirst()); float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); - // if the first segment is longer than the trim-offset, then ggs + // if the first segment is longer than the trim-offset if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { segments.removeFirst(); if (segments.isEmpty()) { 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 2afe492..8f91f84 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -75,137 +75,131 @@ class StreamActionsServiceTest { } // --------------------------------------------------------------- - // getSegmentsInRange: missing / invalid directories + // selectSegments: missing / invalid directories // --------------------------------------------------------------- @Test - void getSegmentsInRange_directoryDoesNotExist_throwsIOException() { + void selectSegments_directoryDoesNotExist_throwsIOException() { Path missing = tempDir.resolve("does-not-exist"); - assertThatThrownBy(() -> service.getSegmentsInRange(missing, Instant.EPOCH, Instant.now())) + assertThatThrownBy(() -> service.selectSegments(missing, Instant.EPOCH, Instant.now())) .isInstanceOf(NoSuchFileException.class); } @Test - void getSegmentsInRange_pathIsAFileNotADirectory_throwsIOException() throws IOException { + void selectSegments_pathIsAFileNotADirectory_throwsIOException() throws IOException { Path file = tempDir.resolve("not-a-directory.txt"); Files.createFile(file); - assertThatThrownBy(() -> service.getSegmentsInRange(file, Instant.EPOCH, Instant.now())) + assertThatThrownBy(() -> service.selectSegments(file, Instant.EPOCH, Instant.now())) .isInstanceOf(IOException.class); } - @Test - void getSegmentsInRange_emptyDirectory_returnsEmptyList() throws IOException { - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(100_000)); - - assertThat(result).isEmpty(); - } - // --------------------------------------------------------------- - // getSegmentsInRange: filename filtering + // selectSegments: filename filtering // --------------------------------------------------------------- @Test - void getSegmentsInRange_ignoresFilesNotEndingInTs() throws IOException { + void selectSegments_ignoresFilesNotEndingInTs() throws IOException { createSegment("1.ts"); createSegment("playlist.m3u8"); createSegment("1.ts.tmp"); createSegment("readme.txt"); - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5000)).segments(); assertThat(result).extracting(p -> p.getFileName().toString()).containsExactly("1.ts"); } @Test - void getSegmentsInRange_directoryNamedLikeASegment_isNotIncluded() throws IOException { + void selectSegments_directoryNamedLikeASegment_isNotIncluded() throws IOException { Files.createDirectory(tempDir.resolve("5.ts")); - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000)); - - assertThat(result).isEmpty(); + // the directory is the only ".ts" entry; excluding it leaves nothing to select + assertThatThrownBy(() -> + service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) + .isInstanceOf(IllegalArgumentException.class); } @Test - void getSegmentsInRange_malformedNonNumericFilename_throwsNumberFormatException() throws IOException { + void selectSegments_malformedNonNumericFilename_throwsNumberFormatException() throws IOException { // A real HLS fragment name is always numeric. A corrupted or partially-written // file (e.g. "index.ts") crashes the whole listing instead of just being skipped. createSegment("index.ts"); assertThatThrownBy(() -> - service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) + service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) .isInstanceOf(NumberFormatException.class); } @Test - void getSegmentsInRange_emptyFilenameBeforeExtension_throwsNumberFormatException() throws IOException { + void selectSegments_emptyFilenameBeforeExtension_throwsNumberFormatException() throws IOException { createSegment(".ts"); assertThatThrownBy(() -> - service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) + service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000))) .isInstanceOf(NumberFormatException.class); } // --------------------------------------------------------------- - // getSegmentsInRange: range-boundary arithmetic + // selectSegments: range-boundary arithmetic // --------------------------------------------------------------- @Test - void getSegmentsInRange_segmentEndsExactlyAtRangeStart_isExcluded() throws IOException { + void selectSegments_segmentEndsExactlyAtRangeStart_isExcluded() throws IOException { createSegment("0.ts"); createSegment("3.ts"); - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(3000), Instant.ofEpochMilli(10_000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(3000), Instant.ofEpochMilli(10_000)).segments(); assertThat(result).extracting(p -> p.getFileName().toString()) .containsExactly("3.ts"); } @Test - void getSegmentsInRange_segmentOverlapsRangeStartByOneMs_isIncluded() throws IOException { + void selectSegments_segmentOverlapsRangeStartByOneMs_isIncluded() throws IOException { createSegment("0.ts"); // spans [0, 3000) + stubHeadDuration(3f); // start 2999ms trims into 0.ts, so its duration is probed - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(2999), Instant.ofEpochMilli(10_000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(2999), Instant.ofEpochMilli(10_000)).segments(); assertThat(result).hasSize(1); } @Test - void getSegmentsInRange_segmentStartsExactlyAtRangeEnd_isExcluded() throws IOException { + void selectSegments_segmentStartsExactlyAtRangeEnd_isExcluded() throws IOException { createSegment("5.ts"); // starts at 5000ms - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5000)); - - assertThat(result).isEmpty(); + assertThatThrownBy(() -> + service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5000))) + .isInstanceOf(IllegalArgumentException.class); } @Test - void getSegmentsInRange_segmentStartsOneMsBeforeRangeEnd_isIncluded() throws IOException { + void selectSegments_segmentStartsOneMsBeforeRangeEnd_isIncluded() throws IOException { createSegment("5.ts"); // starts at 5000ms - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5001)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(5001)).segments(); assertThat(result).hasSize(1); } @Test - void getSegmentsInRange_startAfterEnd_throwsIllegalArgument() throws IOException { + void selectSegments_startAfterEnd_throwsIllegalArgument() throws IOException { createSegment("10.ts"); assertThatThrownBy(() -> - service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(20_000), Instant.ofEpochMilli(10_000))) + service.selectSegments(tempDir, Instant.ofEpochMilli(20_000), Instant.ofEpochMilli(10_000))) .isInstanceOf(IllegalArgumentException.class); - } @Test - void getSegmentsInRange_segmentsOutOfOrder_returnedInOrder() throws IOException { + void selectSegments_segmentsOutOfOrder_returnedInOrder() throws IOException { createSegment("20.ts"); createSegment("10.ts"); createSegment("0.ts"); - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(25_000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(25_000)).segments(); assertThat(result) .extracting(p -> p.getFileName().toString()) @@ -213,45 +207,31 @@ class StreamActionsServiceTest { } @Test - void getSegmentsInRange_mixOfOverlappingAndNonOverlappingSegments_returnsOnlyOverlapping() throws IOException { + void selectSegments_mixOfOverlappingAndNonOverlappingSegments_returnsOnlyOverlapping() throws IOException { createSegment("0.ts"); // [0, 3000) createSegment("3.ts"); // [3000, 6000) createSegment("6.ts"); // [6000, 9000) createSegment("9.ts"); // [9000, 12000) + stubHeadDuration(3f); // start 4000ms trims into 3.ts, so its duration is probed - - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(4000), Instant.ofEpochMilli(8000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(4000), Instant.ofEpochMilli(8000)).segments(); assertThat(result).extracting(p -> p.getFileName().toString()) .containsExactly("3.ts", "6.ts"); } @Test - void getSegmentsInRange_resultsAreSortedByTimestampRegardlessOfDirectoryOrder() throws IOException { + void selectSegments_resultsAreSortedByTimestampRegardlessOfDirectoryOrder() throws IOException { createSegment("30.ts"); createSegment("10.ts"); createSegment("20.ts"); - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(100_000)); + var result = service.selectSegments(tempDir, Instant.ofEpochMilli(0), Instant.ofEpochMilli(100_000)).segments(); assertThat(result).extracting(p -> p.getFileName().toString()) .containsExactly("10.ts", "20.ts", "30.ts"); } - @Test - void getSegmentsInRange_gapInStream_returnsTheGap() throws IOException { - createSegment("20.ts"); - createSegment("30.ts"); - // user stops streaming - createSegment("100.ts"); - createSegment("110.ts"); - - List result = service.getSegmentsInRange(tempDir, Instant.ofEpochMilli(70_000), Instant.ofEpochMilli(111_000)); - - assertThat(result).extracting(p -> p.getFileName().toString()) - .containsExactly("30.ts", "100.ts", "110.ts"); - } - // --------------------------------------------------------------- // selectSegments: head selection and trim-offset // --------------------------------------------------------------- -- 2.49.1 From 2011029dc54aedcaffe334ec39d5f40a2cf67771 Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Mon, 27 Jul 2026 23:00:37 +0100 Subject: [PATCH 6/6] cleaned up StreamActions interface --- .../ddf/vodsystem/services/MediaService.java | 9 +- .../services/media/StreamActionsService.java | 275 ++++++++++-------- .../media/StreamActionsServiceTest.java | 61 +++- 3 files changed, 215 insertions(+), 130 deletions(-) diff --git a/src/main/java/com/ddf/vodsystem/services/MediaService.java b/src/main/java/com/ddf/vodsystem/services/MediaService.java index 57deb71..712e0e0 100644 --- a/src/main/java/com/ddf/vodsystem/services/MediaService.java +++ b/src/main/java/com/ddf/vodsystem/services/MediaService.java @@ -134,10 +134,9 @@ public class MediaService { Job job = jobRegistryService.generateJob(); StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); - float duration = (endTime.toEpochMilli() - startTime.toEpochMilli()) / 1000f; Path outputFile = directoryService.getVodsDir(user.getId()).resolve(UUID.randomUUID() + ".mp4"); - return dispatchSaveJob(job, ctx, duration, outputFile, () -> + return dispatchSaveJob(job, ctx, outputFile, () -> vodService.persist( title == null ? Instant.now().toString() : title, description == null ? "" : description, @@ -214,7 +213,7 @@ public class MediaService { StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); Path outputFile = directoryService.getTempOutputDir().resolve(UUID.randomUUID() + ".mp4"); - return dispatchSaveJob(job, ctx, duration, outputFile, () -> + return dispatchSaveJob(job, ctx, outputFile, () -> clipService.persistClip( title == null ? Instant.now().toString() : title, description == null ? "" : description, @@ -230,9 +229,9 @@ public class MediaService { return streamActionsService.selectSegments(streamDir, startTime, endTime); } - private Job dispatchSaveJob(Job job, StreamActionsService.SegmentSelection ctx, float duration, Path outputFile, Runnable onSuccess) { + private Job dispatchSaveJob(Job job, StreamActionsService.SegmentSelection ctx, Path outputFile, Runnable onSuccess) { job.setState(JobState.PROCESSING); - streamActionsService.saveSection(ctx.segments(), ctx.trimOffset(), duration, outputFile, job.getProgressTracker()) + streamActionsService.saveSection(ctx, outputFile, job.getProgressTracker()) .thenRun(() -> { job.setDownload(outputFile); job.setState(JobState.SUCCEEDED); 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 ed37183..553ce78 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -16,17 +16,16 @@ 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; /** - * Service responsible for performing stream-related media actions, such as - * extracting and saving a trimmed section of video from a sequence of - * recorded segments. + * Extracts and saves a trimmed section of a stream from its recorded HLS + * segments. *

- * Operations in this service are backed by {@code ffmpeg} and run - * asynchronously on a dedicated executor so that potentially long-running - * encoding/concatenation work does not block request-handling threads. + * The usual flow is {@link #selectSegments} to resolve the segments covering a + * time range into a {@link SegmentSelection}, then {@link #saveSection} to mux + * that selection to a file. Both are backed by {@code ffmpeg} and run on a + * dedicated executor so long-running encodes never block request threads. */ @Service public class StreamActionsService { @@ -40,66 +39,133 @@ public class StreamActionsService { } /** - * Concatenates a list of video segments and extracts a trimmed section - * from the result, writing the output to {@code outputFile}. - *

- * Failures (process launch errors, ffmpeg failures, interruption) are - * surfaced via the returned future rather than thrown, in line with - * normal {@link CompletableFuture} conventions — callers should use - * {@code .exceptionally()} / {@code .handle()} rather than a try/catch. + * A resolved section of a stream, ready to be muxed by {@link #saveSection}. * - * @param segments the ordered list of segment files to concatenate, - * in playback order - * @param trimOffset the offset, in seconds, from the start of the - * concatenated stream at which to begin the trimmed - * section - * @param duration the duration, in seconds, of the trimmed section - * to extract - * @param outputFile the destination file the trimmed section will be - * written to; overwritten if it already exists - * @return a {@link CompletableFuture} that completes with the ffmpeg - * command output on success, or completes exceptionally on failure + * @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 + */ + public record SegmentSelection(List segments, float trimOffset, float duration) {} + + /** + * Resolves the recorded {@code .ts} segments covering {@code [startTime, endTime)} + * into a {@link SegmentSelection} ready for {@link #saveSection}. + *

+ * The trim offset into the first segment is derived from {@code startTime}. If + * that start falls in a recording gap after the earliest overlapping segment + * (so the segment holds no footage for it), the segment is dropped and the + * selection instead begins at the next recorded footage. + * + * @param streamDirectory directory holding the stream's {@code .ts} segments + * @param startTime inclusive start of the requested section + * @param endTime exclusive end of the requested section; must be after {@code startTime} + * @return the segments, trim offset, and duration describing the section + * @throws IllegalArgumentException if {@code startTime} is not before {@code endTime}, + * or no segment holds footage for the range + * @throws IOException if listing the directory or probing a segment fails + */ + public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + List segments = findOverlappingSegments(streamDirectory, startTime, endTime); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No stream segments found in the given time range"); + } + + 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()) { + throw new IllegalArgumentException("No recorded footage at the requested start time"); + } + firstMs = parseTimestampMs(segments.getFirst()); + 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; + + 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. + *

+ * Runs asynchronously; failures (process-launch errors, ffmpeg failures, + * interruption) complete the returned future exceptionally rather than being + * thrown, so callers should use {@link CompletableFuture#exceptionally} or + * {@link CompletableFuture#handle} rather than try/catch. + * + * @param selection the segments, trim offset, and duration to write + * (see {@link #selectSegments}) + * @param outputFile destination for the muxed section; overwritten if present + * @param progressTracker receives progress updates as ffmpeg runs + * @return a future completing with the ffmpeg output, or exceptionally on failure */ @Async("ffmpegTaskExecutor") public CompletableFuture saveSection( - List segments, - float trimOffset, - float duration, + SegmentSelection selection, Path outputFile, ProgressTracker progressTracker ) { + List segments = selection.segments(); + float trimOffset = selection.trimOffset(); + float duration = selection.duration(); if (segments.isEmpty()) { - return CompletableFuture.failedFuture(new IllegalArgumentException("Segment list cannot be empty")); + return CompletableFuture.failedFuture(new IllegalArgumentException("Selection must contain at least one segment")); } - if (trimOffset < 0 || duration <= 0) { return CompletableFuture.failedFuture(new IllegalArgumentException( "trimOffset must be >= 0 and duration must be > 0 (got " + trimOffset + ", " + duration + ")")); } - Path headSegment = null; + Path headFragment = null; + Path concatList = null; try { - // Unique per invocation so concurrent saves by different users never - // collide on the same head fragment; cleaned up in the finally block. - headSegment = Files.createTempFile("reencode-section-", ".ts"); - encodeHead(segments.getFirst(), trimOffset, headSegment); + // 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); - String concatInput = Stream.concat(Stream.of(headSegment), segments.stream().skip(1)) - .map(p -> p.toAbsolutePath().toString()) - .collect(Collectors.joining("|")); + 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); List command = List.of( "ffmpeg", "-y", - "-i", "concat:" + concatInput, + "-f", "concat", + "-safe", "0", + "-i", concatList.toAbsolutePath().toString(), "-t", String.valueOf(duration), "-c", "copy", outputFile.toAbsolutePath().toString() ); - logger.info("Saving section with ({} segments) to '{}'", - segments.size(), outputFile); + logger.info("Saving section ({} segments) to '{}'", segments.size(), outputFile); CommandOutput output = commandRunner.run(command, line -> commandRunner.setProgress(line, progressTracker, duration)); progressTracker.markComplete(); @@ -115,25 +181,47 @@ public class StreamActionsService { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); } finally { - if (headSegment != null) { - try { - Files.deleteIfExists(headSegment); - } catch (IOException e) { - logger.warn("Failed to delete temp head fragment '{}': {}", headSegment, e.toString()); - } - } + deleteTempQuietly(headFragment); + deleteTempQuietly(concatList); } } - private List getSegmentsInRange(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + /** + * Parses the start timestamp encoded in an HLS segment's filename (e.g. + * {@code 1712345678.ts}). nginx names fragments in whole seconds, so + * second-scale values are scaled up to milliseconds. + * + * @param path a {@code .ts} segment path + * @return the segment's start time in epoch milliseconds + * @throws IllegalArgumentException if {@code path} is not a {@code .ts} file + * @throws NumberFormatException if the filename stem is not numeric + */ + public long parseTimestampMs(Path path) { + String fullName = path.getFileName().toString(); + + if (!fullName.endsWith(".ts")) { + throw new IllegalArgumentException("File must end in ts"); + } + + String name = fullName.substring(0, fullName.length() - 3); + long value = Long.parseLong(name); + return value < 1_000_000_000_000L ? value * 1000L : value; + } + + /** + * Lists the {@code .ts} segments whose playback interval overlaps + * {@code [startTime, endTime)}, in playback order. A segment spans from its + * own timestamp up to the next segment's timestamp (or open-ended for the last). + */ + private List findOverlappingSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { long startMs = startTime.toEpochMilli(); long endMs = endTime.toEpochMilli(); - if (endTime.toEpochMilli() <= startTime.toEpochMilli()) { + if (endMs <= startMs) { throw new IllegalArgumentException("Start time must be before end time"); } - List results = new ArrayList<>(); + List overlapping = new ArrayList<>(); try (Stream files = Files.list(streamDirectory)) { List streamFiles = files @@ -151,54 +239,12 @@ public class StreamActionsService { : Long.MAX_VALUE; if (segmentStart < endMs && segmentEnd > startMs) { - results.add(streamFile); + overlapping.add(streamFile); } } } - return results; - } - - /** - * The segments backing a save and the offset into the first one at which - * the requested section begins. - * - * @param segments the ordered segments to concatenate, in playback order - * @param trimOffset the offset, in seconds, into {@code segments.getFirst()} - * at which the requested section starts - */ - public record SegmentSelection(List segments, float trimOffset) {} - - /** - * Selects the recorded {@code .ts} segments overlapping {@code [startTime, endTime)} - * and computes the trim offset into the first segment. - *

- * If {@code startTime} falls in a stream gap after the earliest overlapping segment - * (so that segment holds no footage for the requested start), the segment is dropped - * and the selection begins at the first genuinely-recorded footage instead. - * - * @throws IllegalArgumentException if no segments hold footage for the requested range - * @throws IOException if listing the stream directory or probing a segment fails - */ - public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { - List segments = getSegmentsInRange(streamDirectory, startTime, endTime); - if (segments.isEmpty()) { - throw new IllegalArgumentException("No stream segments found in the given time range"); - } - - long firstMs = parseTimestampMs(segments.getFirst()); - float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); - - // if the first segment is longer than the trim-offset - if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { - segments.removeFirst(); - if (segments.isEmpty()) { - throw new IllegalArgumentException("No recorded footage at the requested start time"); - } - trimOffset = 0f; - } - - return new SegmentSelection(segments, trimOffset); + return overlapping; } private float probeDurationSeconds(Path segment) throws IOException { @@ -212,24 +258,6 @@ public class StreamActionsService { } } - public long parseTimestampMs(Path path) { - String fullName = path.getFileName().toString(); - - if (!fullName.endsWith(".ts")) { - throw new IllegalArgumentException("File must end in ts"); - } - - String name = fullName.substring(0, fullName.length() - 3); - long value = Long.parseLong(name); - // nginx hls_fragment_naming system uses seconds; convert to ms - return value < 1_000_000_000_000L ? value * 1000L : value; - } - - /** - * Re-encodes {@code firstSegment} from {@code trimOffset} to its end, - * writing an MPEG-TS fragment that can be stream-copy concatenated with - * the untouched fragments that follow it. - */ private void encodeHead(Path firstSegment, float trimOffset, Path head) throws IOException, InterruptedException { @@ -251,11 +279,30 @@ public class StreamActionsService { ); logger.debug("Re-encoding head fragment '{}' from {}s", firstSegment, trimOffset); - commandRunner.run(command, line -> { /* head pass is short; not worth reporting */ }); + 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/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java index 8f91f84..eef3327 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -247,6 +247,8 @@ 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); } @Test @@ -262,6 +264,10 @@ class StreamActionsServiceTest { assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) .containsExactly("100.ts", "110.ts"); assertThat(selection.trimOffset()).isZero(); + // the dropped head shifts the effective start to 100s, so the duration + // shrinks to end at endTime (111 - 100 = 11s) instead of the 41s the + // original 70s->111s window would have implied + assertThat(selection.duration()).isEqualTo(11f); } @Test @@ -351,13 +357,15 @@ class StreamActionsServiceTest { stubRunnerProducingHead(commandOutput); - var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(segmentA, segmentB), 5.0f, 12.0f), output, progress); assertThat(future.get()).isSameAs(commandOutput); assertThat(progress.isComplete()).isTrue(); - // saveSection issues two ffmpeg passes: the head re-encode (trims the - // first segment via -ss) followed by the stream-copy concat (-t/-c copy). + // 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). @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); Mockito.verify(commandRunner, Mockito.times(2)).run(captor.capture(), any()); @@ -365,20 +373,50 @@ class StreamActionsServiceTest { List concatCommand = captor.getAllValues().get(1); 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()); } + @Test + void saveSection_zeroTrimOffset_skipsHeadReencodeAndRunsSingleConcatPass() throws Exception { + Path output = Path.of("/tmp/out/clip.mp4"); + ProgressTracker progress = new ProgressTracker(); + CommandOutput commandOutput = new CommandOutput(); + + when(commandRunner.run(anyList(), any())).thenReturn(commandOutput); + + var future = service.saveSection( + new StreamActionsService.SegmentSelection( + List.of(Path.of("/tmp/a.ts"), Path.of("/tmp/b.ts")), 0f, 10f), output, progress); + + 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). + @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("-t", "10.0"); + assertThat(command).doesNotContain("-ss"); + } + @Test void saveSection_commandRunnerThrowsIOException_completesExceptionallyWithoutMarkingProgressComplete() throws Exception { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - // The head re-encode is the first ffmpeg pass; its failure must surface. + // With a non-zero trim offset the head re-encode is the first ffmpeg pass; + // its failure must surface. when(commandRunner.run(anyList(), any())).thenThrow(new IOException("ffmpeg boom")); - var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(Path.of("/tmp/a.ts")), 2f, 10f), output, progress); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) @@ -391,7 +429,8 @@ class StreamActionsServiceTest { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - var future = service.saveSection(List.of(), 0f, 10f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(), 0f, 10f), output, progress); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) @@ -400,13 +439,13 @@ class StreamActionsServiceTest { } @Test - void saveSection_nullOutputFile_completesExceptionally() throws Exception { + void saveSection_nullOutputFile_completesExceptionally() { ProgressTracker progress = new ProgressTracker(); - // Let the head pass succeed so execution reaches the null output file. - stubRunnerProducingHead(new CommandOutput()); - - var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress); + // trimOffset 0 skips the head pass; building the concat 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); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) -- 2.49.1