Skip to content

fix streaming startup and seeking #156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kchaouki merged 1 commit into from
Oct 3, 2026
Merged

fix streaming startup and seeking #156

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 47 additions & 20 deletions src/main/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -46,20 +49,35 @@ 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;
}
int first = Math.max(0, Math.min(firstPiece, totalPieces - 1));
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;
}
Expand Down Expand Up @@ -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()) {
Expand All @@ -108,43 +130,48 @@ 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;

// 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;
}
}
44 changes: 36 additions & 8 deletions src/main/java/com/filmexa/stream/modules/download/worker/TorrentDownloadWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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<TorrentDescriptor> 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 {
Expand Down Expand Up @@ -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;
}
Expand All @@ -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);
Expand Down
16 changes: 9 additions & 7 deletions src/main/java/com/filmexa/stream/modules/streaming/serviceImpl/StreamServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,9 @@ public class StreamServiceImpl implements StreamService {
private static final double COMPLETE_FILE_RATIO = 0.95;

private static final Set<String> 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
Expand Down Expand Up @@ -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));
}
Expand Down
2 changes: 1 addition & 1 deletion src/main/resources/application.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
54 changes: 42 additions & 12 deletions src/test/java/com/filmexa/stream/modules/download/selector/SequentialPieceSelectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,31 +25,29 @@ 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);

BitSet availablePieces = new BitSet(totalPieces);
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<Integer> 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
Expand Down Expand Up @@ -84,8 +82,40 @@ void movieFractionsAreMappedInsideItsMultiFileTorrentRange() {
availablePieces.set(0, 20);
List<Integer> 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<Integer> 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<Integer> 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));
}
}
2 changes: 1 addition & 1 deletion src/test/resources/application.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
Loading