Repository: streamlink/streamlink

Issue:
stream.hls: Could the "No new segments in playlist" judgment be longer or customizable?
### Checklist

- [X] This is a plugin issue and not a different kind of issue
- [X] [I have read the contribution guidelines](https://github.com/streamlink/streamlink/blob/master/CONTRIBUTING.md#contributing-to-streamlink)
- [X] [I have checked the list of open and recently closed plugin issues](https://github.com/streamlink/streamlink/issues?q=is%3Aissue+label%3A%22plugin+issue%22)
- [X] [I have checked the commit log of the master branch](https://github.com/streamlink/streamlink/commits/master)

### Streamlink version

Latest stable release

### Description

The SHOWROOM plugin in Streamlink seems to have been updated from 6.0.0 and a new feature has been introduced: to check if the segments in the playlist are exhausted or not to determine to stop streaming. Like this:

```
[download] Written 272.81 MiB to [Filename].ts (29m39s @ 145.41 KiB/s)
[stream.hls][debug] Writing segment 896 to output
[stream.hls][debug] Segment 896 complete
[download] Written 273.10 MiB to [Filename].ts (29m41s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m42s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m43s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m44s @ 129.23 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m45s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m46s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m47s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m47s @ 99.49 KiB/s)
[stream.hls][warning] No new segments in playlist for more than 8.00s. Stopping...
[stream.segmented][debug] Closing worker thread
[stream.segmented][debug] Closing writer thread

[cli][info] Stream ended
[cli][info] Closing currently open stream...
```

As far as I have observed, the time above (8.00s) is not a fixed number, it could be 6/8/10s in other runs. And I have two questions:

1. I didn't see any instruction to let users to configure the time value, so could it be able to customize? I'd be very grateful if you could make this possible in further updates.

2. If you don't have the plan to make it open to customization, could you please make it a bit longer? One minute might be better for SHOWROOM, because it is not rare that SHOWROOM broadcasters happen to have network problems, which will make a short period of stop in updating playlist. And the current time can easily cause Streamlink to stop recording before the stream restarts.



### Debug log

```text
...
[download] Written 272.81 MiB to [Filename].ts (29m39s @ 145.41 KiB/s)
[stream.hls][debug] Writing segment 896 to output
[stream.hls][debug] Segment 896 complete
[download] Written 273.10 MiB to [Filename].ts (29m41s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m42s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m43s @ 143.74 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m44s @ 129.23 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m45s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m46s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m47s @ 114.84 KiB/s)
[stream.hls][debug] Reloading playlist
[download] Written 273.10 MiB to [Filename].ts (29m47s @ 99.49 KiB/s)
[stream.hls][warning] No new segments in playlist for more than 8.00s. Stopping...
[stream.segmented][debug] Closing worker thread
[stream.segmented][debug] Closing writer thread

[cli][info] Stream ended
[cli][info] Closing currently open stream...
```


Discussion:
Lytexx: @bastimeyer 
i ran both 6.0.1 and this PR on the same stream at the same time

the 6.0.1 one eventually failed with this error
> stream.hls][warning] No new segments in playlist for more than 2.00s. Stopping...

youtube stream set to ultra low latency mode seem to have set a target duration of 1

the one running this PR using `--hls-segment-queue-threshold 0` kept running
bastimeyer: Could you please link one of those YT live streams?
Lytexx: > Could you please link one of those YT live streams?

sure, i used this one https://www.youtube.com/watch?v=mgBKCSJZtdc
you can see which latency a streamer has set in the "stats for nerds" thing that you can open with right click on the player

Patch:
diff --git a/src/streamlink/session.py b/src/streamlink/session.py
index 2811e9b1..973ad088 100644
--- a/src/streamlink/session.py
+++ b/src/streamlink/session.py
@@ -241,6 +241,7 @@ class Streamlink:
             "hls-duration": None,
             "hls-playlist-reload-attempts": 3,
             "hls-playlist-reload-time": "default",
+            "hls-segment-queue-threshold": 3,
             "hls-segment-stream-data": False,
             "hls-segment-ignore-names": [],
             "hls-segment-key-uri": None,
@@ -405,6 +406,10 @@ class Streamlink:
                 - ``segment``: duration of the last segment
                 - ``live-edge``: sum of segment durations of the ``hls-live-edge`` value minus one
                 - ``default``: the playlist's target duration
+            * - hls-segment-queue-threshold
+              - ``float``
+              - ``3``
+              - Factor of the playlist's targetduration which sets the threshold for stopping early on missing segments
             * - hls-segment-stream-data
               - ``bool``
               - ``False``
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index 80ad04d0..e2cf4f63 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -151,7 +151,7 @@ class HLSStreamWriter(SegmentedStreamWriter):
 
         return request_params
 
-    def put(self, sequence: Sequence):
+    def put(self, sequence: Optional[Sequence]):
         if self.closed:
             return
 
@@ -289,7 +289,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
     writer: "HLSStreamWriter"
     stream: "HLSStream"
 
-    SEGMENT_QUEUE_TIMING_THRESHOLD_FACTOR = 2
+    SEGMENT_QUEUE_TIMING_THRESHOLD_MIN = 5.0
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
@@ -304,6 +304,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
         self.playlist_reload_time: float = 6
         self.playlist_reload_time_override = self.session.options.get("hls-playlist-reload-time")
         self.playlist_reload_retries = self.session.options.get("hls-playlist-reload-attempts")
+        self.segment_queue_timing_threshold_factor = self.session.options.get("hls-segment-queue-threshold")
         self.live_edge = self.session.options.get("hls-live-edge")
         self.duration_offset_start = int(self.stream.start_offset + (self.session.options.get("hls-start-offset") or 0))
         self.duration_limit = self.stream.duration or (
@@ -401,7 +402,13 @@ class HLSStreamWorker(SegmentedStreamWorker):
         return sequence.num >= self.playlist_sequence
 
     def _segment_queue_timing_threshold_reached(self) -> bool:
-        threshold = self.playlist_targetduration * self.SEGMENT_QUEUE_TIMING_THRESHOLD_FACTOR
+        if self.segment_queue_timing_threshold_factor <= 0:
+            return False
+
+        threshold = max(
+            self.SEGMENT_QUEUE_TIMING_THRESHOLD_MIN,
+            self.playlist_targetduration * self.segment_queue_timing_threshold_factor,
+        )
         if now() <= self.playlist_sequences_last + timedelta(seconds=threshold):
             return False
 
diff --git a/src/streamlink_cli/argparser.py b/src/streamlink_cli/argparser.py
index 2f8adbd8..ab8189bd 100644
--- a/src/streamlink_cli/argparser.py
+++ b/src/streamlink_cli/argparser.py
@@ -953,6 +953,24 @@ def build_parser():
         Default is default.
         """,
     )
+    transport_hls.add_argument(
+        "--hls-segment-queue-threshold",
+        metavar="FACTOR",
+        type=num(float, ge=0),
+        help="""
+        The multiplication factor of the HLS playlist's target duration after which the stream will be stopped early
+        if no new segments were queued after refreshing the playlist (multiple times). The target duration defines the
+        maximum duration a single segment can have, meaning new segments must be available during this time frame,
+        otherwise playback issues can occur.
+
+        The intention of this queue threshold is to be able to stop early when the end of a stream doesn't get
+        announced by the server, so Streamlink doesn't have to wait until a read-timeout occurs. See --stream-timeout.
+
+        Set to ``0`` to disable.
+
+        Default is 3.
+        """,
+    )
     transport_hls.add_argument(
         "--hls-segment-ignore-names",
         metavar="NAMES",
@@ -1376,6 +1394,7 @@ _ARGUMENT_TO_SESSIONOPTION: List[Tuple[str, str, Optional[Callable[[Any], Any]]]
     ("hls_duration", "hls-duration", None),
     ("hls_playlist_reload_attempts", "hls-playlist-reload-attempts", None),
     ("hls_playlist_reload_time", "hls-playlist-reload-time", None),
+    ("hls_segment_queue_threshold", "hls-segment-queue-threshold", None),
     ("hls_segment_stream_data", "hls-segment-stream-data", None),
     ("hls_segment_ignore_names", "hls-segment-ignore-names", None),
     ("hls_segment_key_uri", "hls-segment-key-uri", None),


Test patch:
diff --git a/tests/stream/test_hls.py b/tests/stream/test_hls.py
index 1886f4a6..0550af6e 100644
--- a/tests/stream/test_hls.py
+++ b/tests/stream/test_hls.py
@@ -218,7 +218,7 @@ class TestHLSStreamWorker(TestMixinStreamHLS, unittest.TestCase):
             frozen_time.tick(ONE_SECOND)
             self.await_playlist_reload(1)
 
-            assert worker.handshake_wait.wait_ready(1), "Arrives at first wait() call"
+            assert worker.handshake_wait.wait_ready(1), "Arrives at wait() call #1"
             assert worker.playlist_sequence == 1, "Updates the sequence number"
             assert worker.playlist_sequences_last == EPOCH + ONE_SECOND, "Updates the last queue time"
             assert worker.playlist_targetduration == 5.0
@@ -228,30 +228,117 @@ class TestHLSStreamWorker(TestMixinStreamHLS, unittest.TestCase):
             self.await_playlist_wait(1)
             self.await_playlist_reload(1)
 
-            assert worker.handshake_wait.wait_ready(1), "Arrives at second wait() call"
+            assert worker.handshake_wait.wait_ready(1), "Arrives at wait() call #2"
             assert worker.playlist_sequence == 2, "Updates the sequence number again"
             assert worker.playlist_sequences_last == EPOCH + ONE_SECOND + targetduration, "Updates the last queue time again"
             assert worker.playlist_targetduration == 5.0
 
+            for num in range(3, 6):
+                # trigger next reload when the target duration has passed
+                frozen_time.tick(targetduration)
+                self.await_playlist_wait(1)
+                self.await_playlist_reload(1)
+
+                assert worker.handshake_wait.wait_ready(1), f"Arrives at wait() call #{num}"
+                assert worker.playlist_sequence == 2, "Sequence number is unchanged"
+                assert worker.playlist_sequences_last == EPOCH + ONE_SECOND + targetduration, "Last queue time is unchanged"
+                assert worker.playlist_targetduration == 5.0
+
+            assert mock_log.warning.call_args_list == []
+
             # trigger next reload when the target duration has passed
             frozen_time.tick(targetduration)
             self.await_playlist_wait(1)
             self.await_playlist_reload(1)
 
-            assert worker.handshake_wait.wait_ready(1), "Arrives at third wait() call"
-            assert worker.playlist_sequence == 2, "Sequence number is unchanged"
-            assert worker.playlist_sequences_last == EPOCH + ONE_SECOND + targetduration, "Last queue time is unchanged"
-            assert worker.playlist_targetduration == 5.0
+            self.await_read(read_all=True)
+            self.await_close(1)
 
-            # trigger next reload when the target duration has passed
-            frozen_time.tick(targetduration)
-            self.await_playlist_wait(1)
+            assert mock_log.warning.call_args_list == [call("No new segments in playlist for more than 15.00s. Stopping...")]
+
+    def test_segment_queue_timing_threshold_reached_ignored(self) -> None:
+        thread, segments = self.subject(
+            start=False,
+            options={"hls-segment-queue-threshold": 0},
+            playlists=[
+                # no EXT-X-ENDLIST, last mocked playlist response will be repreated forever
+                Playlist(0, targetduration=5, segments=[Segment(0)]),
+            ],
+        )
+        worker: EventedHLSStreamWorker = thread.reader.worker
+        targetduration = ONE_SECOND * 5
+
+        with freezegun.freeze_time(EPOCH) as frozen_time:
+            self.start()
+
+            assert worker.handshake_reload.wait_ready(1), "Loads playlist for the first time"
+            assert worker.playlist_sequence == -1, "Initial sequence number"
+            assert worker.playlist_sequences_last == EPOCH, "Sets the initial last queue time"
+
+            # first playlist reload has taken one second
+            frozen_time.tick(ONE_SECOND)
             self.await_playlist_reload(1)
 
-            assert worker.handshake_wait.wait_ready(1), "Arrives at fourth wait() call"
-            assert worker.playlist_sequence == 2, "Sequence number is unchanged"
-            assert worker.playlist_sequences_last == EPOCH + ONE_SECOND + targetduration, "Last queue time is unchanged"
+            assert worker.handshake_wait.wait_ready(1), "Arrives at first wait() call"
+            assert worker.playlist_sequence == 1, "Updates the sequence number"
+            assert worker.playlist_sequences_last == EPOCH + ONE_SECOND, "Updates the last queue time"
             assert worker.playlist_targetduration == 5.0
+            assert self.await_read() == self.content(segments)
+
+            # keep reloading a couple of times
+            for num in range(10):
+                frozen_time.tick(targetduration)
+                self.await_playlist_wait(1)
+                self.await_playlist_reload(1)
+
+                assert worker.handshake_wait.wait_ready(1), f"Arrives at wait() #{num + 1}"
+                assert worker.playlist_sequence == 1, "Sequence number is unchanged"
+                assert worker.playlist_sequences_last == EPOCH + ONE_SECOND, "Last queue time is unchanged"
+
+        assert self.thread.data == [], "No new data"
+        assert worker.is_alive()
+
+        # make stream end gracefully to avoid any unnecessary thread blocking
+        self.thread.reader.writer.put(None)
+
+    def test_segment_queue_timing_threshold_reached_min(self) -> None:
+        thread, segments = self.subject(
+            start=False,
+            playlists=[
+                # no EXT-X-ENDLIST, last mocked playlist response will be repreated forever
+                Playlist(0, targetduration=1, segments=[Segment(0)]),
+            ],
+        )
+        worker: EventedHLSStreamWorker = thread.reader.worker
+        targetduration = ONE_SECOND
+
+        with freezegun.freeze_time(EPOCH) as frozen_time, \
+             patch("streamlink.stream.hls.log") as mock_log:
+            self.start()
+
+            assert worker.handshake_reload.wait_ready(1), "Loads playlist for the first time"
+            assert worker.playlist_sequence == -1, "Initial sequence number"
+            assert worker.playlist_sequences_last == EPOCH, "Sets the initial last queue time"
+
+            # first playlist reload has taken one second
+            frozen_time.tick(ONE_SECOND)
+            self.await_playlist_reload(1)
+
+            assert worker.handshake_wait.wait_ready(1), "Arrives at wait() call #1"
+            assert worker.playlist_sequence == 1, "Updates the sequence number"
+            assert worker.playlist_sequences_last == EPOCH + ONE_SECOND, "Updates the last queue time"
+            assert worker.playlist_targetduration == 1.0
+
+            for num in range(2, 7):
+                # trigger next reload when the target duration has passed
+                frozen_time.tick(targetduration)
+                self.await_playlist_wait(1)
+                self.await_playlist_reload(1)
+
+                assert worker.handshake_wait.wait_ready(1), f"Arrives at wait() call #{num}"
+                assert worker.playlist_sequence == 1, "Sequence number is unchanged"
+                assert worker.playlist_sequences_last == EPOCH + ONE_SECOND, "Last queue time is unchanged"
+                assert worker.playlist_targetduration == 1.0
 
             assert mock_log.warning.call_args_list == []
 
@@ -260,10 +347,7 @@ class TestHLSStreamWorker(TestMixinStreamHLS, unittest.TestCase):
             self.await_playlist_wait(1)
             self.await_playlist_reload(1)
 
-            self.await_read(read_all=True)
-            self.await_close(1)
-
-            assert mock_log.warning.call_args_list == [call("No new segments in playlist for more than 10.00s. Stopping...")]
+            assert mock_log.warning.call_args_list == [call("No new segments in playlist for more than 5.00s. Stopping...")]
 
     def test_playlist_reload_offset(self) -> None:
         thread, segments = self.subject(