fix(server): add hint header for segment after init.mp4 (#28867)

* add hint header for segment after init.mp4

* use zod

* actually validate

* update openapi

* linting

mobile/openapi/lib/api/assets_api.dart

@@ -1067,7 +1067,9 @@ class AssetsApi {
   /// * [String] key:
   ///
   /// * [String] slug:
-  Future<Response> getSegmentWithHttpInfo(String filename, String id, String sessionId, int variantIndex, { String? key, String? slug, Future<void>? abortTrigger, }) async {
+  ///
+  /// * [int] xImmichHlsMsn:
+  Future<Response> getSegmentWithHttpInfo(String filename, String id, String sessionId, int variantIndex, { String? key, String? slug, int? xImmichHlsMsn, Future<void>? abortTrigger, }) async {
     // ignore: prefer_const_declarations
     final apiPath = r'/assets/{id}/video/stream/{sessionId}/{variantIndex}/{filename}'
       .replaceAll('{filename}', filename)
@@ -1089,6 +1091,10 @@ class AssetsApi {
       queryParams.addAll(_queryParams('', 'slug', slug));
     }
 
+    if (xImmichHlsMsn != null) {
+      headerParams[r'x-immich-hls-msn'] = parameterToString(xImmichHlsMsn);
+    }
+
     const contentTypes = <String>[];
 
 
@@ -1121,8 +1127,10 @@ class AssetsApi {
   /// * [String] key:
   ///
   /// * [String] slug:
-  Future<MultipartFile?> getSegment(String filename, String id, String sessionId, int variantIndex, { String? key, String? slug, Future<void>? abortTrigger, }) async {
-    final response = await getSegmentWithHttpInfo(filename, id, sessionId, variantIndex, key: key, slug: slug, abortTrigger: abortTrigger,);
+  ///
+  /// * [int] xImmichHlsMsn:
+  Future<MultipartFile?> getSegment(String filename, String id, String sessionId, int variantIndex, { String? key, String? slug, int? xImmichHlsMsn, Future<void>? abortTrigger, }) async {
+    final response = await getSegmentWithHttpInfo(filename, id, sessionId, variantIndex, key: key, slug: slug, xImmichHlsMsn: xImmichHlsMsn, abortTrigger: abortTrigger,);
     if (response.statusCode >= HttpStatus.badRequest) {
       throw ApiException(response.statusCode, await _decodeBodyBytes(response));
     }

open-api/immich-openapi-specs.json

@@ -5026,6 +5026,16 @@
               "maximum": 9007199254740991,
               "type": "integer"
             }
+          },
+          {
+            "name": "x-immich-hls-msn",
+            "required": false,
+            "in": "header",
+            "schema": {
+              "minimum": 0,
+              "maximum": 9007199254740991,
+              "type": "integer"
+            }
           }
         ],
         "responses": {

packages/sdk/src/fetch-client.ts

@@ -4472,13 +4472,14 @@ export function getMediaPlaylist({ id, key, sessionId, slug, variantIndex }: {
 /**
  * Get HLS segment or init file
  */
-export function getSegment({ filename, id, key, sessionId, slug, variantIndex }: {
+export function getSegment({ filename, id, key, sessionId, slug, variantIndex, xImmichHlsMsn }: {
     filename: string;
     id: string;
     key?: string;
     sessionId: string;
     slug?: string;
     variantIndex: number;
+    xImmichHlsMsn?: number;
 }, opts?: Oazapfts.RequestOpts) {
     return oazapfts.ok(oazapfts.fetchBlob<{
         status: 200;
@@ -4487,7 +4488,10 @@ export function getSegment({ filename, id, key, sessionId, slug, variantIndex }:
         key,
         slug
     }))}`, {
-        ...opts
+        ...opts,
+        headers: oazapfts.mergeHeaders(opts?.headers, {
+            "x-immich-hls-msn": xImmichHlsMsn
+        })
     }));
 }
 /**

server/src/controllers/video-stream.controller.ts

@@ -1,11 +1,17 @@
-import { Controller, Delete, Get, Header, HttpCode, HttpStatus, Next, Param, Res } from '@nestjs/common';
+import { Controller, Delete, Get, Header, Headers, HttpCode, HttpStatus, Next, Param, Res } from '@nestjs/common';
 import { ApiProduces, ApiTags } from '@nestjs/swagger';
 import { NextFunction, Response } from 'express';
+import { ZodValidationException } from 'nestjs-zod';
 import { HLS_PLAYLIST_CONTENT_TYPE } from 'src/constants';
 import { Endpoint, HistoryBuilder } from 'src/decorators';
 import { AuthDto } from 'src/dtos/auth.dto';
-import { HlsSegmentParamDto, HlsSessionParamDto, HlsVariantParamDto } from 'src/dtos/streaming.dto';
-import { ApiTag, Permission, RouteKey } from 'src/enum';
+import {
+  HlsSegmentHeaderDto,
+  HlsSegmentParamDto,
+  HlsSessionParamDto,
+  HlsVariantParamDto,
+} from 'src/dtos/streaming.dto';
+import { ApiTag, ImmichHeader, Permission, RouteKey } from 'src/enum';
 import { Auth, Authenticated, FileResponse } from 'src/middleware/auth.guard';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { HlsService } from 'src/services/hls.service';
@@ -59,10 +65,21 @@ export class VideoStreamController {
   async getSegment(
     @Auth() auth: AuthDto,
     @Param() { id, sessionId, variantIndex, filename }: HlsSegmentParamDto,
+    @Headers() headers: HlsSegmentHeaderDto,
     @Res() res: Response,
     @Next() next: NextFunction,
   ) {
-    await sendFile(res, next, () => this.service.getSegment(auth, id, sessionId, variantIndex, filename), this.logger);
+    try {
+      headers = HlsSegmentHeaderDto.create(headers);
+    } catch (error) {
+      throw new ZodValidationException(error);
+    }
+    await sendFile(
+      res,
+      next,
+      () => this.service.getSegment(auth, id, sessionId, variantIndex, filename, headers[ImmichHeader.HlsInitSegment]),
+      this.logger,
+    );
   }
 
   @Delete(':id/video/stream/:sessionId')

server/src/dtos/streaming.dto.ts

@@ -1,4 +1,5 @@
 import { createZodDto } from 'nestjs-zod';
+import { ImmichHeader } from 'src/enum';
 import z from 'zod';
 
 const HlsSessionParamSchema = z.object({
@@ -24,3 +25,10 @@ const HlsSegmentParamSchema = z.object({
 });
 
 export class HlsSegmentParamDto extends createZodDto(HlsSegmentParamSchema) {}
+
+const HlsSegmentHeaderSchema = z.object({
+  // Lets the client hint at which segment will be loaded after init.mp4.
+  [ImmichHeader.HlsInitSegment]: z.coerce.number().int().min(0).optional(),
+});
+
+export class HlsSegmentHeaderDto extends createZodDto(HlsSegmentHeaderSchema) {}

server/src/enum.ts

@@ -24,6 +24,7 @@ export enum ImmichHeader {
   SharedLinkSlug = 'x-immich-share-slug',
   Checksum = 'x-immich-checksum',
   CorrelationId = 'X-Correlation-ID',
+  HlsInitSegment = 'x-immich-hls-msn',
 }
 
 export enum ImmichQuery {

server/src/services/hls.service.spec.ts

@@ -256,7 +256,7 @@ describe(HlsService.name, () => {
       });
     });
 
-    it('returns lastRequested + 1 for init.mp4 after a segment has been served', async () => {
+    it('returns lastRequested + 1 for init.mp4 without a target segment', async () => {
       await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s');
       mocks.websocket.serverSend.mockClear();
       await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4');
@@ -313,6 +313,35 @@ describe(HlsService.name, () => {
         NotFoundException,
       );
     });
+
+    it('uses the target segment for init.mp4 when provided', async () => {
+      await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4', 7);
+      expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
+        sessionId,
+        variantIndex,
+        segmentIndex: 7,
+      });
+    });
+
+    it('prefers the target segment over the lastRequested + 1 fallback', async () => {
+      await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s'); // fallback would be 6
+      mocks.websocket.serverSend.mockClear();
+      await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4', 12);
+      expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
+        sessionId,
+        variantIndex,
+        segmentIndex: 12,
+      });
+    });
+
+    it('ignores the target segment for media segment requests (the filename wins)', async () => {
+      await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s', 99);
+      expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
+        sessionId,
+        variantIndex,
+        segmentIndex: 5,
+      });
+    });
   });
 
   describe('endSession', () => {

server/src/services/hls.service.ts

@@ -82,7 +82,14 @@ export class HlsService extends BaseService {
     return this.generateMediaPlaylist(asset);
   }
 
-  async getSegment(auth: AuthDto, assetId: string, sessionId: string, variantIndex: number, filename: string) {
+  async getSegment(
+    auth: AuthDto,
+    assetId: string,
+    sessionId: string,
+    variantIndex: number,
+    filename: string,
+    initSegment?: number,
+  ) {
     await this.requireAccess({ auth, permission: Permission.AssetView, ids: [assetId] });
 
     const session = await this.videoStreamRepository.getSession(sessionId);
@@ -99,7 +106,7 @@ export class HlsService extends BaseService {
     });
 
     const apiSession = this.trackSession(sessionId, variantIndex);
-    const segmentIndex = this.getSegmentIndex(apiSession, filename);
+    const segmentIndex = this.getSegmentIndex(apiSession, filename, initSegment);
     this.websocketRepository.serverSend('HlsHeartbeat', { sessionId, variantIndex, segmentIndex });
 
     if (await this.storageRepository.checkFileExists(path, constants.R_OK)) {
@@ -172,9 +179,13 @@ export class HlsService extends BaseService {
     return `${sessionId}:${variantIndex}:${segmentIndex}`;
   }
 
-  private getSegmentIndex(session: ApiSession, filename: string) {
+  private getSegmentIndex(session: ApiSession, filename: string, initSegment?: number) {
     if (filename.endsWith('.mp4')) {
-      return (session.lastRequestedSegment ?? -1) + 1;
+      // We need to know where to start transcoding, but the init.mp4 has no segment number in its name.
+      // We can infer this from the last requested segment, but this can be inaccurate given the client
+      // can load cached segments without reaching out to the server. `initSegment` acts as a hint to
+      // remove ambiguity when possible.
+      return initSegment ?? (session.lastRequestedSegment ?? -1) + 1;
     }
     const segmentIndex = Number.parseInt(HLS_SEGMENT_FILENAME_REGEX.exec(filename)![1]);
     session.lastRequestedSegment = segmentIndex;

server/src/services/transcoding.service.ts

@@ -30,6 +30,7 @@ type Session = {
   ownerId: string;
   paused: boolean;
   process: ChildProcess | null;
+  starting: boolean;
   startSegment: number | null;
   variantIndex: number | null;
 };
@@ -75,6 +76,7 @@ export class TranscodingService extends BaseService {
         ownerId,
         paused: false,
         process: null,
+        starting: false,
         startSegment: null,
         variantIndex: null,
       });
@@ -145,11 +147,19 @@ export class TranscodingService extends BaseService {
     } else if (session.process) {
       this.resumeTranscode(session);
       return;
+    } else if (session.starting) {
+      this.logger.debug(`Session ${sessionId} is already starting a transcode, skipping duplicate start request`);
+      return;
     }
 
-    const process = await this.startTranscode(session, variantIndex, segmentIndex);
-    if (process) {
-      session.process = process;
+    session.starting = true;
+    try {
+      const process = await this.startTranscode(session, variantIndex, segmentIndex);
+      if (process) {
+        session.process = process;
+      }
+    } finally {
+      session.starting = false;
     }
   }