diff --git a/server/src/services/hls.service.ts b/server/src/services/hls.service.ts index 949aab12fb..eda24f53b5 100644 --- a/server/src/services/hls.service.ts +++ b/server/src/services/hls.service.ts @@ -24,8 +24,8 @@ type AssetWithStreamInfo = { videoStream: VideoStreamInfo & { timeBase: number } @Injectable() export class HlsService extends BaseService { - private pendingSegments = new PendingEvents<'HlsSegmentResult'>(15_000); - private pendingSessions = new PendingEvents<'HlsSessionResult'>(5000); + private pendingSegments = new PendingEvents<'HlsSegmentResult'>({ timeoutMs: 15_000 }); + private pendingSessions = new PendingEvents<'HlsSessionResult'>({ timeoutMs: 5000 }); private sessions = new Map(); @OnEvent({ name: 'HlsSessionResult', server: true, workers: [ImmichWorker.Api] }) @@ -60,6 +60,8 @@ export class HlsService extends BaseService { throw new NotFoundException('Asset is not yet ready for streaming'); } + // Sharing the sessionId allows only one microservices worker to successfully insert to the session table. + // The microservices worker that creates a session owns the transcoding lifecycle for it. const sessionId = this.cryptoRepository.randomUUID(); this.websocketRepository.serverSend('HlsSessionRequest', { sessionId, assetId, ownerId: auth.user.id }); await this.pendingSessions.wait(sessionId); diff --git a/server/src/services/transcoding.service.ts b/server/src/services/transcoding.service.ts index 506be153ce..c913765a2b 100644 --- a/server/src/services/transcoding.service.ts +++ b/server/src/services/transcoding.service.ts @@ -81,6 +81,7 @@ export class TranscodingService extends BaseService { this.cleanupInterval ??= setInterval(() => void this.removeInactiveSessions(), HLS_CLEANUP_INTERVAL_MS); this.websocketRepository.serverSend('HlsSessionResult', { sessionId }); } catch (error) { + // If insertion failed due to a PK constraint, another worker has already created a session for this ID. if (!isVideoStreamSessionPkConstraint(error)) { this.logger.error(`Failed to create HLS session ${sessionId}: ${error}`); this.websocketRepository.serverSend('HlsSessionResult', { sessionId, error: 'Failed to create HLS session' }); diff --git a/server/src/utils/event.ts b/server/src/utils/event.ts index adb8c8f1df..77698b903f 100644 --- a/server/src/utils/event.ts +++ b/server/src/utils/event.ts @@ -2,8 +2,11 @@ import { ArgOf, EmitEvent } from 'src/repositories/event.repository'; export class PendingEvents extends { error?: string } ? T : never }[EmitEvent]> { private pending = new Map>[]; timeout: NodeJS.Timeout }>(); + private timeoutMs: number; - constructor(private timeoutMs: number) {} + constructor({ timeoutMs }: { timeoutMs: number }) { + this.timeoutMs = timeoutMs; + } wait(key: string): Promise> { const completer = Promise.withResolvers>();