fix: add high-water mark deletion to hls session (#1737)

Also ensures that deletion happens sequentially (no fire-and-forget)

server/src/stream/hls/HlsPlaylistMutator.test.ts

@@ -920,6 +920,112 @@ describe('HlsPlaylistMutator', () => {
     });
   });
 
+  describe('high-water mark floor protection', () => {
+    // These tests verify the invariant that HlsSession's #highestDeletedBelow
+    // relies on: the before_segment_number filter acts as a hard floor, so the
+    // playlist never references segments below (segmentNumber - segmentsToKeepBefore).
+
+    const start = dayjs('2024-10-18T14:00:00.000-0400');
+    const largeOpts = { ...defaultOpts, maxSegmentsToKeep: 20 };
+
+    it('Scenario B: segmentNumber=0 (empty _minByIp) selects segments from the start', () => {
+      // Without the high-water mark fix, stale cleanup empties _minByIp and
+      // minSegmentRequested returns 0. This causes trimPlaylist to serve segments
+      // from the very beginning of the playlist — all of which have been deleted.
+      // This test documents the behavior that #highestDeletedBelow prevents.
+      const lines = createPlaylist(100);
+
+      const result = mutator.trimPlaylist(
+        start,
+        {
+          type: 'before_segment_number',
+          segmentNumber: 0,
+          segmentsToKeepBefore: 10,
+        },
+        lines,
+        largeOpts,
+      );
+
+      // minSeg = max(0-10, 0) = 0 → all 100 segments pass the filter (≥20) → take first 20 = segs 0..19
+      expect(result.playlist).toContain('data000000.ts');
+      expect(result.playlist).toContain('data000019.ts');
+      expect(result.playlist).not.toContain('data000020.ts');
+    });
+
+    it('Scenario B fix: high-water mark as segmentNumber floors the playlist above deleted range', () => {
+      // After deleteOldSegments(190), #highestDeletedBelow = 190.
+      // Math.max(minSegmentRequested=0, highestDeletedBelow=190) = 190 is used
+      // as segmentNumber, so the playlist starts at 180 (190 - keepBefore:10)
+      // rather than 0, avoiding any reference to deleted segments.
+      const lines = createPlaylist(300);
+
+      const result = mutator.trimPlaylist(
+        start,
+        {
+          type: 'before_segment_number',
+          segmentNumber: 190,
+          segmentsToKeepBefore: 10,
+        },
+        lines,
+        largeOpts,
+      );
+
+      // minSeg = max(190-10, 0) = 180; segs 180..299 (120 ≥ 20) → take first 20 = segs 180..199
+      expect(result.playlist).toContain('data000180.ts');
+      expect(result.playlist).toContain('data000199.ts');
+      expect(result.playlist).not.toContain('data000179.ts');
+      expect(result.playlist).not.toContain('data000000.ts');
+    });
+
+    it('Scenario A: stale client removal jump still floors above deleted range', () => {
+      // Client A was at seg 100, stale cleanup removes it, leaving Client B at seg 200.
+      // deleteOldSegments(190) ran, so #highestDeletedBelow = 190.
+      // Math.max(minSegmentRequested=200, highestDeletedBelow=190) = 200.
+      // Playlist must not include segments below 190 (200 - keepBefore:10).
+      const lines = createPlaylist(300);
+
+      const result = mutator.trimPlaylist(
+        start,
+        {
+          type: 'before_segment_number',
+          segmentNumber: 200,
+          segmentsToKeepBefore: 10,
+        },
+        lines,
+        largeOpts,
+      );
+
+      // minSeg = max(200-10, 0) = 190; segs 190..299 (110 ≥ 20) → take first 20 = segs 190..209
+      expect(result.playlist).toContain('data000190.ts');
+      expect(result.playlist).toContain('data000209.ts');
+      expect(result.playlist).not.toContain('data000189.ts');
+      expect(result.playlist).not.toContain('data000100.ts');
+    });
+
+    it('floor does not over-trim when segmentNumber equals the live edge', () => {
+      // When #highestDeletedBelow and minSegmentRequested agree (normal single-client case),
+      // the playlist should include the last keepBefore segments before the current position.
+      const lines = createPlaylist(50);
+
+      const result = mutator.trimPlaylist(
+        start,
+        {
+          type: 'before_segment_number',
+          segmentNumber: 30,
+          segmentsToKeepBefore: 10,
+        },
+        lines,
+        largeOpts,
+      );
+
+      // minSeg = max(30-10, 0) = 20; segs 20..49 (30 ≥ 20) → take first 20 = segs 20..39
+      expect(result.playlist).toContain('data000020.ts');
+      expect(result.playlist).toContain('data000039.ts');
+      expect(result.playlist).not.toContain('data000019.ts');
+      expect(result.segmentCount).toBe(20);
+    });
+  });
+
   describe('integration with real test file', () => {
     it('should parse and trim the test.m3u8 file', async () => {
       const lines = (await readTestFile('test.m3u8'))

server/src/stream/hls/HlsSession.ts

@@ -25,7 +25,7 @@ import fs from 'node:fs/promises';
 import path, { basename, dirname, extname } from 'node:path';
 import type { DeepRequired } from 'ts-essentials';
 import type { BaseHlsSessionOptions } from './BaseHlsSession.js';
-import { BaseHlsSession } from './BaseHlsSession.js';
+import { BaseHlsSession, SegmentNameRegex } from './BaseHlsSession.js';
 import type { HlsPlaylistFilterOptions } from './HlsPlaylistMutator.js';
 import { HlsPlaylistMutator } from './HlsPlaylistMutator.js';
 
@@ -54,6 +54,7 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
   #lastDelete: Dayjs = dayjs().subtract(1, 'year');
   #isFirstTranscode = true;
   #lastDiscontinuitySequence: number | undefined;
+  #highestDeletedBelow: number = 0;
 
   constructor(
     channel: ChannelOrmWithTranscodeConfig,
@@ -80,7 +81,10 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
   async trimPlaylist(filterOpts?: HlsPlaylistFilterOptions) {
     filterOpts ??= {
       type: 'before_segment_number',
-      segmentNumber: this.minSegmentRequested,
+      segmentNumber: Math.max(
+        this.minSegmentRequested,
+        this.#highestDeletedBelow,
+      ),
       segmentsToKeepBefore: 10,
     };
     return Result.attemptAsync(async () => {
@@ -106,9 +110,7 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
               this.channel.uuid,
               this.channel.number,
             );
-            this.deleteOldSegments(trimResult.sequence).catch((e) =>
-              this.logger.error(e),
-            );
+            await this.deleteOldSegments(trimResult.sequence);
             this.#lastDelete = now;
           }
 
@@ -371,6 +373,10 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
   }
 
   private async deleteOldSegments(sequenceNum: number) {
+    this.#highestDeletedBelow = Math.max(
+      this.#highestDeletedBelow,
+      sequenceNum,
+    );
     const workingDirectoryFiles = await fs.readdir(this._workingDirectory);
     const segments = filter(
       seq.collect(
@@ -379,8 +385,8 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
           return ext === '.ts' || ext === '.mp4';
         }),
         (file) => {
-          const matches = file.match(/[A-z/]+(\d+)\.[ts|mp4]/);
-          if (matches && matches.length > 0) {
+          const matches = file.match(SegmentNameRegex);
+          if (matches && matches.length > 1) {
             return {
               file,
               seq: parseInt(matches[1]!),