diff --git a/server/src/services/transcoding.service.spec.ts b/server/src/services/transcoding.service.spec.ts index ed3f39b0bc..738ad9c0af 100644 --- a/server/src/services/transcoding.service.spec.ts +++ b/server/src/services/transcoding.service.spec.ts @@ -25,6 +25,12 @@ describe(TranscodingService.name, () => { listener!('rename', `seg_${index}.m4s`); }; + const completeSegmentsThrough = (start: number, end: number) => { + for (let i = start; i <= end; i++) { + completeSegment(i); + } + }; + beforeEach(() => { ({ sut, mocks } = newTestService(TranscodingService)); mocks.systemMetadata.get.mockResolvedValue({ ffmpeg: { realtime: { enabled: true } } }); @@ -185,7 +191,7 @@ describe(TranscodingService.name, () => { }); it('pauses the transcode once the lead exceeds HLS_BACKPRESSURE_PAUSE_SEGMENTS', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); @@ -193,7 +199,7 @@ describe(TranscodingService.name, () => { }); it('does not pause when the lead equals the pause threshold', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); @@ -201,7 +207,7 @@ describe(TranscodingService.name, () => { }); it('resumes once the lead drops below HLS_BACKPRESSURE_RESUME_SEGMENTS', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); expect(proc.kill).toHaveBeenCalledWith('SIGSTOP'); vi.mocked(proc.kill).mockClear(); @@ -213,7 +219,7 @@ describe(TranscodingService.name, () => { }); it('stays paused while the lead is in the dead-band', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); vi.mocked(proc.kill).mockClear(); @@ -230,7 +236,7 @@ describe(TranscodingService.name, () => { }); it('is a no-op when the heartbeat omits segmentIndex', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId }); @@ -238,7 +244,7 @@ describe(TranscodingService.name, () => { }); it('resumes the paused transcode when the client requests the next in-range segment', async () => { - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); expect(proc.kill).toHaveBeenCalledWith('SIGSTOP'); vi.mocked(proc.kill).mockClear(); @@ -253,7 +259,7 @@ describe(TranscodingService.name, () => { const newProc = mockSpawn(0, '', ''); mocks.process.spawn.mockReturnValueOnce(newProc); - completeSegment(HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); + completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1); await sut.onHeartbeat({ sessionId, segmentIndex: 0 }); expect(proc.kill).toHaveBeenCalledWith('SIGSTOP'); @@ -264,6 +270,32 @@ describe(TranscodingService.name, () => { expect(newProc.kill).not.toHaveBeenCalled(); }); + + it('ignores stale segment events from the prior transcode after a backward seek', async () => { + const newProc = mockSpawn(0, '', ''); + mocks.process.spawn.mockReturnValueOnce(newProc); + + const completedAhead = HLS_BACKPRESSURE_PAUSE_SEGMENTS + 5; + completeSegmentsThrough(1, completedAhead); // seg_0 was emitted in beforeEach + + await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 1, segmentIndex: 0 }); + + vi.mocked(newProc.kill).mockClear(); + mocks.websocket.serverSend.mockClear(); + completeSegment(completedAhead + 1); + + expect(mocks.websocket.serverSend).not.toHaveBeenCalledWith( + 'HlsSegmentResult', + expect.objectContaining({ segmentIndex: completedAhead + 1 }), + ); + expect(newProc.kill).not.toHaveBeenCalled(); + + completeSegment(0); + expect(mocks.websocket.serverSend).toHaveBeenCalledWith( + 'HlsSegmentResult', + expect.objectContaining({ segmentIndex: 0 }), + ); + }); }); describe('inactivity sweeper', () => { diff --git a/server/src/services/transcoding.service.ts b/server/src/services/transcoding.service.ts index 926462dda7..f994fff4f8 100644 --- a/server/src/services/transcoding.service.ts +++ b/server/src/services/transcoding.service.ts @@ -273,6 +273,11 @@ export class TranscodingService extends BaseService { return; } const segmentIndex = Number.parseInt(match[1]); + const expected = session.lastCompletedSegment === null ? session.startSegment : session.lastCompletedSegment + 1; + // Ignore stale events from old process after seek + if (expected === null || segmentIndex !== expected) { + return; + } session.lastCompletedSegment = segmentIndex; this.websocketRepository.serverSend('HlsSegmentResult', { sessionId: session.id,