diff --git a/src/main/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelector.java b/src/main/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelector.java index 6b78e7c..6872523 100644 --- a/src/main/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelector.java +++ b/src/main/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelector.java @@ -17,6 +17,9 @@ */ public class SequentialPieceSelector implements PieceSelector { + /** Headers/indexes regularly span several torrent pieces, not just piece zero. */ + private static final int HEADER_WINDOW_PIECES = 16; + private int totalPieces; private volatile boolean isMp4; @@ -46,6 +49,10 @@ public void initSelector(int totalPieces) { * subtitles and other files that happen to share the torrent. */ public void configureMovieRange(int firstPiece, int lastPiece, boolean mp4) { + configureMovieRange(firstPiece, lastPiece, mp4 ? "movie.mp4" : "movie.mkv"); + } + + public void configureMovieRange(int firstPiece, int lastPiece, String fileName) { if (totalPieces <= 0) { return; } @@ -53,13 +60,24 @@ public void configureMovieRange(int firstPiece, int lastPiece, boolean mp4) { int last = Math.max(first, Math.min(lastPiece, totalPieces - 1)); this.movieFirstPiece = first; this.movieLastPiece = last; - this.isMp4 = mp4; + this.isMp4 = needsTailMetadata(fileName); this.movieRangeConfigured = true; if (playheadPiece < first || playheadPiece > last) { this.playheadPiece = first; } } + private boolean needsTailMetadata(String fileName) { + if (fileName == null) { + return false; + } + String lower = fileName.toLowerCase(java.util.Locale.ROOT); + return lower.endsWith(".mp4") || lower.endsWith(".m4v") + || lower.endsWith(".mov") || lower.endsWith(".3gp") + || lower.endsWith(".3g2") || lower.endsWith(".mkv") + || lower.endsWith(".webm"); + } + public boolean isMovieRangeConfigured() { return movieRangeConfigured; } @@ -100,6 +118,10 @@ public int getPlayheadPiece() { return playheadPiece; } + public int getMovieFirstPiece() { + return movieFirstPiece; + } + @Override public IntStream getNextPieces(BitSet relevantChunks, PieceStatistics pieceStatistics) { if (relevantChunks.isEmpty()) { @@ -108,17 +130,25 @@ public IntStream getNextPieces(BitSet relevantChunks, PieceStatistics pieceStati IntStream.Builder builder = IntStream.builder(); - // The movie header has to come first whatever the playhead says - piece zero - // may belong to an nfo or sample in a multi-file torrent. - // decoded, or even probed, without it. + // The complete header window has to come first. MKV/WebM EBML data is at the + // beginning; AVI and MPEG containers also need their opening structures; and an + // MP4 moov atom can be much larger than a single torrent piece. int firstPiece = movieFirstPiece; - if (relevantChunks.get(firstPiece)) { - builder.add(firstPiece); - } - int lastPiece = movieLastPiece >= firstPiece ? movieLastPiece : totalPieces - 1; - if (isMp4 && totalPieces > 1 && relevantChunks.get(lastPiece)) { - builder.add(lastPiece); + int headerEnd = Math.min(lastPiece, firstPiece + HEADER_WINDOW_PIECES - 1); + relevantChunks.stream() + .filter(i -> i >= firstPiece && i <= headerEnd) + .forEach(builder::add); + + // ISO-BMFF containers may keep their moov/index at the end; Matroska/WebM often + // keep seek Cues there too. Fetch a window rather than one last piece because + // real-world metadata regularly spans several torrent pieces. + int tailStart = Math.max(firstPiece, lastPiece - HEADER_WINDOW_PIECES + 1); + if (isMp4) { + relevantChunks.stream() + .filter(i -> i >= tailStart && i <= lastPiece) + .filter(i -> i > headerEnd) + .forEach(builder::add); } int playhead = playheadPiece; @@ -126,25 +156,22 @@ public IntStream getNextPieces(BitSet relevantChunks, PieceStatistics pieceStati // From the playhead to the end of the file... relevantChunks.stream() .filter(i -> i >= playhead) - .filter(i -> shouldKeepPiece(i, firstPiece, lastPiece)) + .filter(i -> shouldKeepPiece(i, firstPiece, headerEnd, tailStart, lastPiece)) .forEach(builder::add); // ...then everything skipped over, so a seek does not abandon it for good. relevantChunks.stream() .filter(i -> i < playhead) - .filter(i -> shouldKeepPiece(i, firstPiece, lastPiece)) + .filter(i -> shouldKeepPiece(i, firstPiece, headerEnd, tailStart, lastPiece)) .forEach(builder::add); return builder.build(); } - private boolean shouldKeepPiece(int pieceIndex, int firstPiece, int lastPiece) { - if (pieceIndex == firstPiece) { - return false; - } - if (isMp4 && pieceIndex == lastPiece) { - return false; - } - return true; + private boolean shouldKeepPiece(int pieceIndex, int firstPiece, int headerEnd, + int tailStart, int lastPiece) { + boolean inHeader = pieceIndex >= firstPiece && pieceIndex <= headerEnd; + boolean inTail = isMp4 && pieceIndex >= tailStart && pieceIndex <= lastPiece; + return !inHeader && !inTail; } } diff --git a/src/main/java/com/filmexa/stream/modules/download/worker/TorrentDownloadWorker.java b/src/main/java/com/filmexa/stream/modules/download/worker/TorrentDownloadWorker.java index bd4a25f..8be2efc 100644 --- a/src/main/java/com/filmexa/stream/modules/download/worker/TorrentDownloadWorker.java +++ b/src/main/java/com/filmexa/stream/modules/download/worker/TorrentDownloadWorker.java @@ -40,6 +40,8 @@ @Slf4j public class TorrentDownloadWorker { + private static final int STARTUP_WINDOW_PIECES = 2; + /** How often we ask the DHT for peers again while a download is getting nowhere. */ private static final long PEER_TRIGGER_INTERVAL_MS = 5_000; @@ -294,15 +296,15 @@ && configureMovieRange(movieId, selector, runtime, torrentId)) { progress = ((double) downloaded / total) * 100.0; } - // 3. 42 Rule: Check if 10 MB buffer reached - boolean streamReady = false; - if (downloaded >= 25 * 1024 * 1024) { - streamReady = true; - } + // Aggregate downloaded bytes can be scattered near the end after a + // seek. Do not advertise a playable startup buffer until the opening + // pieces containing the container header are actually verified. + boolean streamReady = downloaded >= 25 * 1024 * 1024 + && isStartupWindowDownloaded(runtime, torrentId, selector); if (streamReady && !readyToStreamMarked.get()) { readyToStreamMarked.set(true); - log.info("Movie {} reached 10 MB buffer! Ready to stream.", movieId); + log.info("Movie {} has a verified startup buffer and is ready to stream.", movieId); movieDownloadRepository.findByMovieId(movieId).ifPresent(download -> { download.setStatus(DownloadStatus.READY_TO_STREAM); @@ -398,6 +400,32 @@ && configureMovieRange(movieId, selector, runtime, torrentId)) { } } + private boolean isStartupWindowDownloaded(BtRuntime runtime, TorrentId torrentId, + SequentialPieceSelector selector) { + if (torrentId == null || !selector.isMovieRangeConfigured()) { + return false; + } + try { + Optional descriptor = + runtime.service(TorrentRegistry.class).getDescriptor(torrentId); + if (descriptor.isEmpty() || descriptor.get().getDataDescriptor() == null) { + return false; + } + Bitfield bitfield = descriptor.get().getDataDescriptor().getBitfield(); + int first = selector.getMovieFirstPiece(); + int end = Math.min(bitfield.getPiecesTotal(), first + STARTUP_WINDOW_PIECES); + for (int piece = first; piece < end; piece++) { + if (!bitfield.isComplete(piece)) { + return false; + } + } + return end > first; + } catch (RuntimeException e) { + log.debug("Could not verify startup pieces: {}", e.getMessage()); + return false; + } + } + /** The info hash bt knows the torrent by, so we can drive its peer lookups. */ private TorrentId torrentIdOf(String magnetUrl) { try { @@ -438,7 +466,7 @@ private boolean configureMovieRange(Long movieId, SequentialPieceSelector select } if (first >= 0) { String name = movie.getPathElements().get(movie.getPathElements().size() - 1); - selector.configureMovieRange(first, last, name.toLowerCase().endsWith(".mp4")); + selector.configureMovieRange(first, last, name); log.info("Movie {} occupies torrent pieces {}..{} ({})", movieId, first, last, name); return true; } @@ -457,7 +485,7 @@ private TorrentFile largestVideoFile(DataDescriptor data) { return files.stream() .filter(file -> { String name = file.getPathElements().get(file.getPathElements().size() - 1).toLowerCase(); - return name.matches(".*\\.(mp4|mkv|avi|mov|m4v|webm|wmv|flv|mpg|mpeg|ts)$"); + return name.matches(".*\\.(mp4|mkv|avi|mov|m4v|webm|wmv|flv|mpg|mpeg|ts|m2ts|mts|m2v|vob|ogv|ogg|3gp|3g2)$"); }) .max(java.util.Comparator.comparingLong(TorrentFile::getSize)) .orElse(null); diff --git a/src/main/java/com/filmexa/stream/modules/streaming/serviceImpl/StreamServiceImpl.java b/src/main/java/com/filmexa/stream/modules/streaming/serviceImpl/StreamServiceImpl.java index c79dca1..8fbe04c 100644 --- a/src/main/java/com/filmexa/stream/modules/streaming/serviceImpl/StreamServiceImpl.java +++ b/src/main/java/com/filmexa/stream/modules/streaming/serviceImpl/StreamServiceImpl.java @@ -68,7 +68,9 @@ public class StreamServiceImpl implements StreamService { private static final double COMPLETE_FILE_RATIO = 0.95; private static final Set VIDEO_EXTENSIONS = - Set.of("mp4", "mkv", "avi", "mov", "m4v", "webm", "wmv", "flv", "mpg", "mpeg", "ts"); + Set.of("mp4", "mkv", "avi", "mov", "m4v", "webm", "wmv", "flv", + "mpg", "mpeg", "ts", "m2ts", "mts", "m2v", "vob", "ogv", "ogg", + "3gp", "3g2"); /** * Subtitles are offered in English, French and Arabic only. Deriving the set from @@ -230,13 +232,13 @@ public Segment prepareSegment(Long movieId, int height, int segmentIndex) { throw new NotFoundException("Segment " + segmentIndex + " is outside movie " + movieId); } - // For the first segment ffmpeg is the authoritative readiness check. Mapping a - // timestamp to torrent pieces is only an estimate (especially for VBR MP4), and - // can reject a segment that the decoder can already read. A not-yet-readable - // source becomes a retryable 503 in Ffmpeg rather than holding the session in - // PREPARING indefinitely. Later segments still use the piece gate for fast seeks. + // ffmpeg is the authoritative readiness check. Mapping a timestamp to a byte or + // torrent-piece fraction is only an estimate for VBR media and is especially + // inaccurate for MKV clusters. Tell the torrent where playback moved, then let + // ffmpeg either produce the segment or return a retryable 503. if (segmentIndex > 0) { - requireDownloaded(movieId, info, segmentIndex); + double segmentStart = (double) segmentIndex * properties.getSegmentSeconds(); + requestSeek(movieId, info, segmentStart); } return new Segment(movieId, info, resolution, segmentIndex, originalAudioIndex(movieId, info)); } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 650c1eb..1aa58fc 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -88,7 +88,7 @@ app.stream.ffprobe-path=${FFPROBE_PATH:ffprobe} app.stream.preset=${STREAM_PRESET:ultrafast} app.stream.max-concurrent-transcodes=${STREAM_MAX_TRANSCODES:2} app.stream.encoder-threads=${STREAM_ENCODER_THREADS:1} -app.stream.transcode-timeout-seconds=${STREAM_TRANSCODE_TIMEOUT:120} +app.stream.transcode-timeout-seconds=${STREAM_TRANSCODE_TIMEOUT:30} app.stream.probe-timeout-seconds=${STREAM_PROBE_TIMEOUT:30} app.stream.token-secret=${STREAM_TOKEN_SECRET:${SECURITY_JWT_SECRET_KEY}} app.stream.token-ttl-minutes=${STREAM_TOKEN_TTL_MINUTES:180} diff --git a/src/test/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelectorTest.java b/src/test/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelectorTest.java index bc34fd6..067cf00 100644 --- a/src/test/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelectorTest.java +++ b/src/test/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelectorTest.java @@ -25,16 +25,13 @@ void testMp4SequentialOrder_piece0First_lastPieceSecond() { // 1. Piece 0 must be FIRST assertEquals(0, order.get(0)); - // 2. Last piece (9) must be SECOND (for MP4 moov atom) - assertEquals(9, order.get(1)); - - // 3. Middle pieces in sequential order (1, 2, 3, 4, 5, 6, 7, 8) - assertEquals(List.of(0, 9, 1, 2, 3, 4, 5, 6, 7, 8), order); + // A small movie fits entirely in the header window, so every piece is sequential. + assertEquals(List.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9), order); } @Test void seekingForwardAsksForThatPartFirst_thenFillsInWhatWasSkipped() { - int totalPieces = 10; + int totalPieces = 100; SequentialPieceSelector selector = new SequentialPieceSelector(false); selector.initSelector(totalPieces); @@ -42,14 +39,15 @@ void seekingForwardAsksForThatPartFirst_thenFillsInWhatWasSkipped() { availablePieces.set(0, totalPieces); // Three quarters in: the viewer jumped towards the end of the film. - assertEquals(7, selector.seekToFraction(0.75)); + assertEquals(75, selector.seekToFraction(0.75)); List order = selector.getNextPieces(availablePieces, null) .boxed() .toList(); - // The header still comes first, then the seeked-to part, then the skipped middle. - assertEquals(List.of(0, 7, 8, 9, 1, 2, 3, 4, 5, 6), order); + assertEquals(List.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15), + order.subList(0, 16)); + assertEquals(75, order.get(16)); } @Test @@ -84,8 +82,40 @@ void movieFractionsAreMappedInsideItsMultiFileTorrentRange() { availablePieces.set(0, 20); List order = selector.getNextPieces(availablePieces, null).boxed().toList(); - assertEquals(4, order.get(0)); - assertEquals(13, order.get(1)); - assertEquals(9, order.get(2)); + assertEquals(List.of(4, 5, 6, 7, 8, 9, 10, 11, 12, 13), order.subList(0, 10)); + } + + @Test + void largeMp4PrioritisesBothHeaderAndTailWindows() { + SequentialPieceSelector selector = new SequentialPieceSelector(false); + selector.initSelector(200); + selector.configureMovieRange(10, 189, "feature.MOV"); + + BitSet available = new BitSet(200); + available.set(0, 200); + List order = selector.getNextPieces(available, null).boxed().toList(); + + assertEquals(10, order.get(0)); + assertEquals(25, order.get(15)); + assertEquals(174, order.get(16)); + assertEquals(189, order.get(31)); + } + + @Test + void webmPrioritisesOpeningHeaderAndEndingCues() { + SequentialPieceSelector selector = new SequentialPieceSelector(false); + selector.initSelector(200); + selector.configureMovieRange(10, 189, "feature.webm"); + selector.seekToFraction(0.75); + + BitSet available = new BitSet(200); + available.set(0, 200); + List order = selector.getNextPieces(available, null).boxed().toList(); + + assertEquals(10, order.get(0)); + assertEquals(25, order.get(15)); + assertEquals(174, order.get(16)); + assertEquals(189, order.get(31)); + assertEquals(145, order.get(32)); } } diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index c24cc01..61c8475 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -68,7 +68,7 @@ app.stream.ffprobe-path=${FFPROBE_PATH:ffprobe} app.stream.preset=${STREAM_PRESET:ultrafast} app.stream.max-concurrent-transcodes=${STREAM_MAX_TRANSCODES:2} app.stream.encoder-threads=${STREAM_ENCODER_THREADS:1} -app.stream.transcode-timeout-seconds=${STREAM_TRANSCODE_TIMEOUT:120} +app.stream.transcode-timeout-seconds=${STREAM_TRANSCODE_TIMEOUT:30} app.stream.probe-timeout-seconds=${STREAM_PROBE_TIMEOUT:30} app.stream.token-secret=${STREAM_TOKEN_SECRET:${SECURITY_JWT_SECRET_KEY}} app.stream.token-ttl-minutes=${STREAM_TOKEN_TTL_MINUTES:180}