From 9d64ae9a5abe592a587e3588717ee2f21d8b2a8a Mon Sep 17 00:00:00 2001 From: Skila <239451985+Skila1@users.noreply.github.com> Date: Thu, 20 Aug 2026 17:20:15 +1000 Subject: [PATCH] Fix silent auto-skip with a reason-aware single-consumer queue. Failed and stuck items now advance once without consuming unrelated queue tracks. Retries, notices, and health stay out of this change. Co-authored-by: Cursor --- cogs/basic.py | 5 +- cogs/listeners.py | 15 +- tests/test_playback_transitions.py | 386 ++++++++++++++ voicelink/__init__.py | 1 + voicelink/events.py | 30 +- voicelink/exceptions.py | 23 + voicelink/ipc/methods.py | 9 +- voicelink/playback.py | 812 +++++++++++++++++++++++++++++ voicelink/player.py | 429 +++++++++++++-- voicelink/pool.py | 93 +++- voicelink/queue.py | 156 ++++-- voicelink/views/controller.py | 3 + 12 files changed, 1846 insertions(+), 116 deletions(-) create mode 100644 tests/test_playback_transitions.py create mode 100644 voicelink/playback.py diff --git a/cogs/basic.py b/cogs/basic.py index 13c4fff0..d3f31a58 100644 --- a/cogs/basic.py +++ b/cogs/basic.py @@ -345,7 +345,7 @@ async def forceplay(self, ctx: commands.Context, *, query: str, start: str = "0" if player.queue._repeat.mode == voicelink.LoopType.TRACK: await player.set_repeat(voicelink.LoopType.OFF) - await player.stop() if player.is_playing else await player.do_next() + await player.stop(intent=voicelink.AttemptIntent.FORCEPLAY) if player.is_playing else await player.do_next() @commands.hybrid_command(name="pause", aliases=get_aliases("pause")) @commands.dynamic_cooldown(cooldown_check, commands.BucketType.guild) @@ -418,6 +418,7 @@ async def skip(self, ctx: commands.Context, index: int = 0): if index: player.queue.skipto(index) + player.queue.prepare_user_reselect() await send_localized_message(ctx, "player.controls.skip.success", ctx.author) if player.queue._repeat.mode == voicelink.LoopType.TRACK: @@ -447,9 +448,11 @@ async def back(self, ctx: commands.Context, index: int = 1): if not player.is_playing: player.queue.backto(index) + player.queue.prepare_user_reselect() await player.do_next() else: player.queue.backto(index + 1) + player.queue.prepare_user_reselect() await player.stop() await send_localized_message(ctx, "player.controls.back.success", ctx.author) diff --git a/cogs/listeners.py b/cogs/listeners.py index c714e17f..937ca232 100644 --- a/cogs/listeners.py +++ b/cogs/listeners.py @@ -91,7 +91,7 @@ async def restore_last_session_players(self) -> None: decoded_track = voicelink.Track.decode(track_id) requester = channel.guild.get_member(track_data.get("requester_id")) track = voicelink.Track(track_id=track_id, info=decoded_track, requester=requester) - player.queue._queue.append(track) + player.queue.put(track) # Restore queue settings. player.queue._position = queue_data.get("position", 0) - 1 @@ -131,21 +131,16 @@ async def restore_last_session_players(self) -> None: func.logger.error("Failed to remove session file: %s", Config.LAST_SESSION_FILE_DIR, exc_info=del_error) @commands.Cog.listener() - async def on_voicelink_track_end(self, player: voicelink.Player, track, _): - await player.do_next() + async def on_voicelink_track_end(self, player: voicelink.Player, track, reason): + await player.handle_track_end(track, reason) @commands.Cog.listener() async def on_voicelink_track_stuck(self, player: voicelink.Player, track, _): - await asyncio.sleep(10) - await player.do_next() + await player.handle_track_stuck(track) @commands.Cog.listener() async def on_voicelink_track_exception(self, player: voicelink.Player, track, error: dict): - try: - player._track_is_stuck = True - await player.context.send(f"{error['message']} The next song will begin in the next 5 seconds.", delete_after=10) - except: - pass + await player.handle_track_exception(track, error) @commands.Cog.listener() async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState): diff --git a/tests/test_playback_transitions.py b/tests/test_playback_transitions.py new file mode 100644 index 00000000..53fb3a17 --- /dev/null +++ b/tests/test_playback_transitions.py @@ -0,0 +1,386 @@ +import os +import sys +import unittest +from types import SimpleNamespace + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from voicelink.enums import LoopType +from voicelink.exceptions import NodeException +from voicelink.playback import ( + AttemptState, + EventDisposition, + PendingEventKind, + PlaybackAdvanceReason, + PlaybackSession, + QueueItem, + TerminalSource, + encoded_from_payload, +) +from voicelink.queue import FairQueue, Queue + + +class FakeTrack: + def __init__(self, encoded, title="song", position=0, end_time=None, requester=None, uri=None): + self.track_id = encoded + self.title = title + self.position = position + self.end_time = end_time + self.requester = requester + self.uri = uri or f"https://example.test/{encoded}" + self.author = "artist" + self.data = {"track_id": encoded, "requester_id": getattr(requester, "id", 0)} + + +class FakeRequester: + def __init__(self, bot=False, user_id=1): + self.bot = bot + self.id = user_id + + +class PlaybackTransitionTests(unittest.TestCase): + def setUp(self): + self.session = PlaybackSession() + self.item_a = QueueItem(1, FakeTrack("enc-a", "A", position=60_000, end_time=180_000)) + self.item_b = QueueItem(2, FakeTrack("enc-b", "B")) + self.item_c = QueueItem(3, FakeTrack("enc-c", "C")) + + def _start(self, item, encoded=None): + attempt = self.session.create_attempt(item, start_pos=item.track.position or 0) + self.session.mark_play_in_flight() + self.assertTrue(self.session.validate_play_completion(attempt.play_seq, item.item_id, attempt.attempt_id)) + self.session.arm_after_play_success() + self.session.apply_track_start(encoded or item.track.track_id) + return attempt + + def test_encoded_payload_snapshot(self): + self.assertEqual(encoded_from_payload({"track": {"encoded": "abc"}}), "abc") + self.assertEqual(encoded_from_payload({"encodedTrack": "xyz"}), "xyz") + + def test_early_track_start_is_pending_then_applied(self): + attempt = self.session.create_attempt(self.item_a) + self.session.mark_play_in_flight() + disposition = self.session.enqueue_or_apply(PendingEventKind.TRACK_START, encoded="enc-a") + self.assertEqual(disposition, EventDisposition.PENDING) + self.assertEqual(attempt.state, AttemptState.STARTING) + pending = self.session.arm_after_play_success() + self.assertEqual(len(pending), 1) + self.assertTrue(self.session.apply_track_start(pending[0].encoded)) + self.assertEqual(attempt.state, AttemptState.STARTED) + self.assertEqual(self.session.on_watchdog(), "IGNORE") + + def test_early_load_failed_processed_once_after_arm(self): + self.session.create_attempt(self.item_a) + self.session.mark_play_in_flight() + self.session.enqueue_or_apply(PendingEventKind.TRACK_END, encoded="enc-a", reason="loadFailed") + pending = self.session.arm_after_play_success() + self.assertEqual(pending[0].reason, "loadFailed") + decision = self.session.on_load_failed() + self.assertEqual(decision, "ADVANCE") + self.assertEqual(self.session.on_load_failed(), "IGNORE") + self.assertTrue(self.item_a.fail_exhausted) + self.assertFalse(self.session.policy.should_retry(self.item_a, TerminalSource.LOAD_FAILED)) + + def test_patch_failure_discards_pending_start(self): + attempt = self.session.create_attempt(self.item_a) + self.session.mark_play_in_flight() + self.session.enqueue_or_apply(PendingEventKind.TRACK_START, encoded="enc-a") + discarded = self.session.fail_play_patch() + self.assertEqual(len(discarded), 1) + self.assertFalse(attempt.armed) + self.assertEqual(attempt.state, AttemptState.STARTING) + + def test_stuck_versus_finished_one_terminal(self): + self._start(self.item_a) + self.assertEqual(self.session.on_stuck(), "ADVANCE") + self.assertEqual(self.session.on_finished(), "IGNORE") + self.assertTrue(self.item_a.fail_exhausted) + self.assertFalse(self.session.policy.should_retry(self.item_a, TerminalSource.TRACK_STUCK)) + + def test_finished_then_stuck_no_second_advance(self): + self._start(self.item_a) + self.assertEqual(self.session.on_finished(), "ADVANCE") + self.assertEqual(self.session.on_stuck(), "IGNORE") + + def test_leave_during_teardown_no_advance(self): + self._start(self.item_a) + self.session.user_leave() + self.assertFalse(self.session.desired_connected) + self.assertTrue(self.session.tearing_down) + self.assertEqual(self.session.on_load_failed(), "IGNORE") + self.assertEqual(len(self.session.queue_advance_logs), 0) + + def test_stale_play_completion_does_not_mutate(self): + self._start(self.item_a) + old_attempt = self.session.attempt + old_seq = old_attempt.play_seq + self.session.user_skip() + self.session.commit_advance(PlaybackAdvanceReason.MANUAL_SKIP, self.item_b) + b_attempt = self.session.attempt + self.session.mark_play_in_flight() + self.session.arm_after_play_success() + self.assertFalse(self.session.validate_play_completion(old_seq, 1, old_attempt.attempt_id)) + self.assertEqual(self.session.current_item.item_id, 2) + self.assertEqual(self.session.attempt.attempt_id, b_attempt.attempt_id) + + def test_forceplay_replaced_predicate(self): + self._start(self.item_a) + self.session.begin_forceplay(self.item_b.item_id) + self.assertEqual(self.session.on_replaced(), "FORCEPLAY_ADVANCE") + self.session.commit_advance(PlaybackAdvanceReason.FORCEPLAY, self.item_b) + self.assertEqual(self.session.current_item.item_id, 2) + self.assertEqual(len(self.session.queue_advance_logs), 1) + + def test_generic_replaced_does_not_consume(self): + self._start(self.item_a) + self.assertEqual(self.session.on_replaced(), "IGNORE") + self.assertEqual(len(self.session.queue_advance_logs), 0) + self.assertEqual(self.session.current_item.item_id, 1) + + def test_history_once_per_logical_item(self): + self.session.create_attempt(self.item_a) + self.session.mark_play_in_flight() + self.assertTrue(self.session.record_history_if_needed()) + self.assertFalse(self.session.record_history_if_needed()) + self.assertEqual(self.session.history_writes, [1]) + + def test_separate_occurrences_have_independent_history(self): + copy = QueueItem(9, FakeTrack("enc-a", "A")) + self.session.create_attempt(self.item_a) + self.session.record_history_if_needed() + self.session.commit_advance(PlaybackAdvanceReason.FINISHED, copy) + self.session.record_history_if_needed() + self.assertEqual(self.session.history_writes, [1, 9]) + + def test_watchdog_suspended_during_disconnect(self): + self.session.create_attempt(self.item_a) + self.session.mark_play_in_flight() + self.session.arm_after_play_success() + self.session.watchdog_suspended = True + self.assertEqual(self.session.on_watchdog(), "IGNORE") + + def test_abandoned_map_is_capped(self): + for i in range(40): + item = QueueItem(i + 1, FakeTrack(f"e{i}")) + self.session.create_attempt(item) + self.session.remember_stale(item.item_id, i + 1) + self.assertLessEqual(len(self.session.stale_ops), 32) + + def test_prune_stale_play_seq_none_and_empty(self): + self.session.prune_stale(play_seq=5) + self.assertEqual(self.session.stale_ops, {}) + self.session.remember_stale(10, 1) + self.session.remember_stale(20, 2) + self.session.prune_stale(play_seq=None) + self.assertEqual(self.session.stale_ops, {10: 1, 20: 2}) + self.session.prune_stale(item_id=10) + self.assertEqual(self.session.stale_ops, {20: 2}) + + def test_prune_stale_removes_older_ops_without_mutating_while_iterating(self): + self.session.remember_stale(1, 1) + self.session.remember_stale(2, 3) + self.session.remember_stale(3, 5) + self.session.remember_stale(4, 8) + self.session.prune_stale(play_seq=5) + self.assertNotIn(1, self.session.stale_ops) + self.assertNotIn(2, self.session.stale_ops) + self.assertEqual(self.session.stale_ops[3], 5) + self.assertEqual(self.session.stale_ops[4], 8) + + def test_loadfailed_advances_once_and_prunes(self): + self._start(self.item_a) + self.assertEqual(self.session.on_load_failed(), "ADVANCE") + self.assertTrue(self.item_a.fail_exhausted) + self.assertEqual(self.item_a.retry_count, 0) + + self.session.remember_stale(99, 1) + commit = self.session.commit_advance(PlaybackAdvanceReason.LOAD_FAILED, self.item_b) + self.assertEqual(commit.from_item.item_id, 1) + self.assertEqual(commit.to_item.item_id, 2) + self.assertEqual(len(self.session.queue_advance_logs), 1) + self.assertEqual(self.session.current_item.item_id, 2) + + empty = PlaybackSession() + lone = QueueItem(1, FakeTrack("enc-a", "A")) + empty.create_attempt(lone) + empty.mark_play_in_flight() + empty.arm_after_play_success() + empty.apply_track_start("enc-a") + self.assertEqual(empty.on_load_failed(), "ADVANCE") + empty.remember_stale(50, 0) + empty.remember_stale(51, 2) + from_seq = empty.attempt.play_seq + commit_empty = empty.commit_advance(PlaybackAdvanceReason.LOAD_FAILED, None) + self.assertIsNone(commit_empty.to_item) + self.assertEqual(len(empty.queue_advance_logs), 1) + self.assertNotIn(50, empty.stale_ops) + self.assertEqual(empty.stale_ops.get(51), 2) + self.assertEqual(empty.stale_ops.get(1), from_seq) + + def test_failed_item_does_not_consume_unrelated_queue(self): + self._start(self.item_a) + self.assertEqual(self.session.on_load_failed(), "ADVANCE") + commit = self.session.commit_advance(PlaybackAdvanceReason.LOAD_FAILED, self.item_b) + self.assertEqual(commit.from_item.item_id, 1) + self.assertEqual(commit.to_item.item_id, 2) + self.assertEqual(self.item_c.item_id, 3) + self.assertEqual(len(self.session.queue_advance_logs), 1) + + def test_node_exception_play_kind(self): + play_err = NodeException("x", kind="PLAYER_PLAY") + load_err = NodeException("y", kind="LOADTRACKS") + self.assertTrue(play_err.is_play_patch) + self.assertFalse(load_err.is_play_patch) + + +class QueueWrapperTests(unittest.TestCase): + def setUp(self): + self.queue = Queue(100, True, lambda *a: "full") + self.a = FakeTrack("enc-a", "A") + self.b = FakeTrack("enc-b", "B") + self.x1 = FakeTrack("enc-x", "X", uri="https://x/1") + self.x2 = FakeTrack("enc-x", "X", uri="https://x/1") + + def test_public_helpers_return_track_not_queueitem(self): + self.queue.put(self.a) + self.queue.put(self.b) + current = self.queue.get() + self.assertIsInstance(current, FakeTrack) + self.assertTrue(all(isinstance(t, FakeTrack) for t in self.queue.tracks())) + self.assertTrue(all(isinstance(t, FakeTrack) for t in self.queue.history(incTrack=True))) + self.assertTrue(all(isinstance(row, dict) for row in self.queue.session_track_data())) + self.assertIn("trackId", self.queue.ipc_track_payloads()[0]) + + def test_duplicate_identical_songs_have_independent_identity(self): + self.queue.put(self.x1) + self.queue.put(self.x2) + first = self.queue.get_item() + second = self.queue.get_item() + first.mark_retry() + first.mark_exhausted() + self.assertNotEqual(first.item_id, second.item_id) + self.assertTrue(second.can_retry()) + self.assertFalse(first.can_retry()) + + def test_track_loop_success_mints_new_item_id(self): + self.queue.put(self.a) + first = self.queue.get_item() + old_id = first.item_id + self.queue._repeat.set_mode(LoopType.TRACK) + again = self.queue.get_item() + self.assertIs(again.track, first.track) + self.assertNotEqual(again.item_id, old_id) + self.assertEqual(again.retry_count, 0) + + def test_fail_exhausted_track_loop_escapes(self): + self.queue.put(self.a) + self.queue.put(self.b) + current = self.queue.get_item() + current.mark_exhausted() + self.queue._repeat.set_mode(LoopType.TRACK) + nxt = self.queue.get_item(force_next=True, skip_exhausted=True) + self.assertEqual(nxt.track.title, "B") + + def test_successful_queue_wrap_unchanged(self): + self.queue.put(self.a) + self.queue.put(self.b) + self.queue.get() + self.queue.get() + self.queue._repeat.set_mode(LoopType.QUEUE) + wrapped = self.queue.get() + self.assertEqual(wrapped.title, "A") + + def test_fairqueue_uses_track_requester(self): + fq = FairQueue(100, True, lambda *a: "full") + r1 = FakeRequester(user_id=1) + r2 = FakeRequester(user_id=2) + fq.put(FakeTrack("a", requester=r1)) + fq.put(FakeTrack("b", requester=r2)) + fq.put(FakeTrack("c", requester=r1)) + for track in fq.tracks(incTrack=True): + self.assertIsInstance(track, FakeTrack) + self.assertTrue(len(fq._queue) >= 1) + + def test_caller_loop_modes_are_independent(self): + skip_queue = Queue(10, True, lambda *a: "full") + skip_queue.put(self.a) + skip_queue.put(self.b) + skip_queue._repeat.set_mode(LoopType.TRACK) + skip_queue._repeat.set_mode(LoopType.OFF) + self.assertEqual(skip_queue._repeat.mode, LoopType.OFF) + + dropdown = Queue(10, True, lambda *a: "full") + dropdown.put(self.a) + dropdown.put(self.b) + dropdown.get() + dropdown._repeat.set_mode(LoopType.TRACK) + dropdown.skipto(1) + self.assertEqual(dropdown._repeat.mode, LoopType.TRACK) + + def test_caller_track_loop_order_in_sources(self): + from pathlib import Path + + root = Path(__file__).resolve().parents[1] + basic = (root / "cogs" / "basic.py").read_text(encoding="utf-8") + controller = (root / "voicelink" / "views" / "controller.py").read_text(encoding="utf-8") + ipc = (root / "voicelink" / "ipc" / "methods.py").read_text(encoding="utf-8") + + skip_src = basic[basic.find("async def skip"): basic.find("async def back")] + self.assertLess(skip_src.find("set_repeat"), skip_src.find("await player.stop()")) + + back_src = basic[basic.find("async def back"): basic.find("async def seek")] + self.assertGreater(back_src.rfind("set_repeat"), back_src.find("await player.stop()")) + + force_src = basic[basic.find("async def forceplay"): basic.find("async def pause")] + self.assertLess(force_src.find("set_repeat"), force_src.find("AttemptIntent.FORCEPLAY")) + + tracks_src = controller[controller.find("class Tracks"): controller.find("class Effects")] + self.assertIn("prepare_user_reselect", tracks_src) + self.assertIn("await self.player.stop()", tracks_src) + self.assertNotIn("set_repeat", tracks_src) + + skip_ipc = ipc[ipc.find("async def skipTo"): ipc.find("async def backTo")] + self.assertLess(skip_ipc.find("set_repeat"), skip_ipc.find("await player.stop()")) + back_ipc = ipc[ipc.find("async def backTo"): ipc.find("async def moveTrack")] + self.assertNotIn("set_repeat", back_ipc) + + +class EventSnapshotTests(unittest.IsolatedAsyncioTestCase): + async def test_event_objects_snapshot_track(self): + from voicelink.events import TrackEndEvent, TrackStartEvent + + track = FakeTrack("enc-a", "A") + player = SimpleNamespace(_current=track, _ending_track=track) + start = TrackStartEvent({"type": "TrackStartEvent", "track": {"encoded": "enc-a"}}, player) + self.assertIs(start.track, track) + player._current = FakeTrack("enc-b", "B") + self.assertEqual(start.track.title, "A") + + player._current = track + end = TrackEndEvent({"reason": "finished", "track": {"encoded": "enc-a"}}, player) + player._current = None + self.assertEqual(end.track.title, "A") + self.assertEqual(end.reason, "finished") + + +NAME = "Playback Transitions" +DESCRIPTION = ( + "Reason-aware queue advancement: failed and stuck items advance once " + "without consuming unrelated tracks." +) + + +def run() -> bool: + loader = unittest.defaultTestLoader + suite = unittest.TestSuite() + suite.addTests(loader.loadTestsFromTestCase(PlaybackTransitionTests)) + suite.addTests(loader.loadTestsFromTestCase(QueueWrapperTests)) + suite.addTests(loader.loadTestsFromTestCase(EventSnapshotTests)) + result = unittest.TextTestRunner(verbosity=1).run(suite) + return result.wasSuccessful() + + +if __name__ == "__main__": + unittest.main() diff --git a/voicelink/__init__.py b/voicelink/__init__.py index 50cd96ef..7a2282cd 100644 --- a/voicelink/__init__.py +++ b/voicelink/__init__.py @@ -28,6 +28,7 @@ from .config import Config from .enums import SearchType, LoopType, TrackRecType, ReconnectStrategy +from .playback import AttemptIntent from .events import * from .exceptions import * from .filters import * diff --git a/voicelink/events.py b/voicelink/events.py index 17404b63..eb61d062 100644 --- a/voicelink/events.py +++ b/voicelink/events.py @@ -27,9 +27,21 @@ from discord.ext.commands import Bot from typing import TYPE_CHECKING +from .playback import encoded_from_payload + if TYPE_CHECKING: from .player import Player + +def _snapshot_event_track(player: "Player", data: dict): + encoded = encoded_from_payload(data) + current = getattr(player, "_current", None) + if current is not None: + if not encoded or getattr(current, "track_id", None) == encoded: + return current + ending = getattr(player, "_ending_track", None) + return ending if ending is not None else current + class VoicelinkEvent: """The base class for all events dispatched by a node. Every event must be formatted within your bot's code as a listener. @@ -54,13 +66,15 @@ class TrackStartEvent(VoicelinkEvent): def __init__(self, data: dict, player: Player): self.player: Player = player - self.track = self.player._current + self.encoded = encoded_from_payload(data) + self.track = _snapshot_event_track(player, data) # on_voicelink_track_start(player, track) self.handler_args = self.player, self.track def __repr__(self) -> str: - return f"" + track_id = getattr(self.track, "track_id", None) + return f"" class TrackEndEvent(VoicelinkEvent): @@ -71,15 +85,17 @@ class TrackEndEvent(VoicelinkEvent): def __init__(self, data: dict, player: Player): self.player: Player = player - self.track = self.player._ending_track + self.encoded = encoded_from_payload(data) + self.track = _snapshot_event_track(player, data) self.reason: str = data["reason"] # on_voicelink_track_end(player, track, reason) self.handler_args = self.player, self.track, self.reason def __repr__(self) -> str: + track_id = getattr(self.track, "track_id", None) return ( - f"" ) @@ -93,7 +109,8 @@ class TrackStuckEvent(VoicelinkEvent): def __init__(self, data: dict, player: Player): self.player: Player = player - self.track = self.player._ending_track + self.encoded = encoded_from_payload(data) + self.track = _snapshot_event_track(player, data) self.threshold: float = data["thresholdMs"] # on_voicelink_track_stuck(player, track, threshold) @@ -112,7 +129,8 @@ class TrackExceptionEvent(VoicelinkEvent): def __init__(self, data: dict, player: Player): self.player: Player = player - self.track = self.player._ending_track + self.encoded = encoded_from_payload(data) + self.track = _snapshot_event_track(player, data) self.exception: dict = data.get("exception", { "severity": "", "message": "", diff --git a/voicelink/exceptions.py b/voicelink/exceptions.py index 59c0570b..0f4a1940 100644 --- a/voicelink/exceptions.py +++ b/voicelink/exceptions.py @@ -28,6 +28,29 @@ class VoicelinkException(Exception): class NodeException(Exception): """Base exception for nodes.""" + def __init__( + self, + message: str = "Getting errors from Lavalink REST api", + *, + method: str = None, + path: str = None, + kind: str = None, + status: int = None, + body: str = None, + node_id: str = None, + ): + super().__init__(message) + self.method = method + self.path = path + self.kind = kind + self.status = status + self.body = body + self.node_id = node_id + + @property + def is_play_patch(self) -> bool: + return self.kind == "PLAYER_PLAY" + class NodeCreationError(NodeException): """There was a problem while creating the node.""" diff --git a/voicelink/ipc/methods.py b/voicelink/ipc/methods.py index 8b2c9348..09dc82e6 100644 --- a/voicelink/ipc/methods.py +++ b/voicelink/ipc/methods.py @@ -6,7 +6,7 @@ from discord import User, Member from discord.ext import commands -from voicelink import Player, Track, Playlist, NodePool, LoopType, Filters, Config, MongoDBHandler, LangHandler, LYRICS_PLATFORMS +from voicelink import Player, Track, Playlist, NodePool, LoopType, Filters, Config, MongoDBHandler, LangHandler, LYRICS_PLATFORMS, AttemptIntent from voicelink.utils import TempCtx if TYPE_CHECKING: @@ -124,7 +124,7 @@ async def initPlayer(player: Player, member: Member, data: Dict) -> Dict: "avatarUrl": member.display_avatar.url, "name": member.name } for member in player.channel.members ], - "tracks": [ {"trackId": track.track_id, "requesterId": str(track.requester.id)} for track in player.queue._queue ], + "tracks": player.queue.ipc_track_payloads(), "repeatMode": player.queue.repeat.lower(), "channelName": player.channel.name, "currentQueuePosition": player.queue._position + (0 if player.is_playing else 1), @@ -177,6 +177,7 @@ async def skipTo(player: Player, member: Member, data: Dict) -> None: index = data.get("index", 1) if index > 1: player.queue.skipto(index) + player.queue.prepare_user_reselect() if player.queue._repeat.mode == LoopType.TRACK: await player.set_repeat(LoopType.OFF) @@ -198,9 +199,11 @@ async def backTo(player: Player, member: Member, data: Dict) -> None: index = data.get("index", 1) if not player.is_playing: player.queue.backto(index) + player.queue.prepare_user_reselect() await player.do_next() else: player.queue.backto(index + 1) + player.queue.prepare_user_reselect() await player.stop() @require_permission() @@ -226,7 +229,7 @@ async def addTracks(player: Player, member: Member, data: Dict) -> None: elif _type == "forcePlay": await player.add_track(tracks, at_front=True) if player.is_playing: - return await player.stop() + return await player.stop(intent=AttemptIntent.FORCEPLAY) elif _type == "addNext": await player.add_track(tracks, at_front=True) diff --git a/voicelink/playback.py b/voicelink/playback.py new file mode 100644 index 00000000..31d6cd67 --- /dev/null +++ b/voicelink/playback.py @@ -0,0 +1,812 @@ +"""MIT License + +Copyright (c) 2023 - present Vocard Development + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Any, Deque, Dict, Iterable, List, Optional, Tuple + +TRACK_START_TIMEOUT = 15 +EXCEPTION_END_FALLBACK = 2 +MAX_PENDING_EVENTS = 8 +MAX_ITEM_RETRIES = 1 +STALE_OP_CAP = 32 +PLAYBACK_LOGGER_NAME = "vocard.playback" + + +class PlaybackAdvanceReason(Enum): + FINISHED = "FINISHED" + MANUAL_SKIP = "MANUAL_SKIP" + FORCEPLAY = "FORCEPLAY" + PLAYBACK_EXCEPTION = "PLAYBACK_EXCEPTION" + LOAD_FAILED = "LOAD_FAILED" + TRACK_STUCK = "TRACK_STUCK" + REPLACED = "REPLACED" + NODE_DISCONNECT = "NODE_DISCONNECT" + VOICE_DISCONNECT = "VOICE_DISCONNECT" + PLAY_REQUEST_FAILED = "PLAY_REQUEST_FAILED" + CLEANUP = "CLEANUP" + AUTOPLAY = "AUTOPLAY" + UNKNOWN = "UNKNOWN" + + def __str__(self) -> str: + return self.value + + +class AttemptState(Enum): + STARTING = "STARTING" + STARTED = "STARTED" + FAILED = "FAILED" + RETRYING = "RETRYING" + ENDING = "ENDING" + ENDED = "ENDED" + RECOVERING = "RECOVERING" + + +class AttemptIntent(Enum): + NONE = "NONE" + SKIP = "SKIP" + STOP = "STOP" + FORCEPLAY = "FORCEPLAY" + + +class TerminalSource(Enum): + FINISHED = "FINISHED" + LOAD_FAILED = "LOAD_FAILED" + TRACK_STUCK = "TRACK_STUCK" + WATCHDOG = "WATCHDOG" + EXCEPTION_FALLBACK = "EXCEPTION_FALLBACK" + SKIP = "SKIP" + STOP = "STOP" + CLEANUP = "CLEANUP" + PLAY_REST = "PLAY_REST" + RECOVERY_ABANDON = "RECOVERY_ABANDON" + + +class PendingEventKind(Enum): + TRACK_START = "TRACK_START" + TRACK_END = "TRACK_END" + TRACK_EXCEPTION = "TRACK_EXCEPTION" + TRACK_STUCK = "TRACK_STUCK" + + +class ForceplayOpState(Enum): + PENDING_STOP = "PENDING_STOP" + ADVANCING = "ADVANCING" + DONE = "DONE" + STALE = "STALE" + + +class EventDisposition(Enum): + APPLY = "APPLY" + PENDING = "PENDING" + IGNORE = "IGNORE" + AMBIGUOUS = "AMBIGUOUS" + + +FAILURE_TERMINALS = { + TerminalSource.LOAD_FAILED, + TerminalSource.TRACK_STUCK, + TerminalSource.WATCHDOG, + TerminalSource.EXCEPTION_FALLBACK, + TerminalSource.PLAY_REST, +} + +ADVANCE_REASON_FOR_SOURCE = { + TerminalSource.FINISHED: PlaybackAdvanceReason.FINISHED, + TerminalSource.LOAD_FAILED: PlaybackAdvanceReason.LOAD_FAILED, + TerminalSource.TRACK_STUCK: PlaybackAdvanceReason.TRACK_STUCK, + TerminalSource.WATCHDOG: PlaybackAdvanceReason.PLAY_REQUEST_FAILED, + TerminalSource.EXCEPTION_FALLBACK: PlaybackAdvanceReason.PLAYBACK_EXCEPTION, + TerminalSource.SKIP: PlaybackAdvanceReason.MANUAL_SKIP, + TerminalSource.STOP: PlaybackAdvanceReason.CLEANUP, + TerminalSource.CLEANUP: PlaybackAdvanceReason.CLEANUP, + TerminalSource.PLAY_REST: PlaybackAdvanceReason.PLAY_REQUEST_FAILED, + TerminalSource.RECOVERY_ABANDON: PlaybackAdvanceReason.CLEANUP, +} + + +def encoded_from_payload(data: Optional[dict]) -> Optional[str]: + if not data: + return None + track = data.get("track") + if isinstance(track, dict): + return track.get("encoded") or track.get("encodedTrack") + if isinstance(track, str): + return track + return data.get("encodedTrack") + + +def snapshot_exception(data: Optional[dict]) -> dict: + if not data: + return {"severity": "", "message": "", "cause": ""} + exception = data.get("exception") + if isinstance(exception, dict): + return { + "severity": exception.get("severity", ""), + "message": exception.get("message", ""), + "cause": exception.get("cause", ""), + } + return {"severity": "", "message": "", "cause": ""} + + +@dataclass +class QueueItem: + item_id: int + track: Any + retry_count: int = 0 + fail_exhausted: bool = False + history_recorded: bool = False + + def can_retry(self) -> bool: + return self.retry_count < MAX_ITEM_RETRIES and not self.fail_exhausted + + def mark_retry(self) -> None: + self.retry_count += 1 + if self.retry_count >= MAX_ITEM_RETRIES: + self.fail_exhausted = True + + def mark_exhausted(self) -> None: + self.fail_exhausted = True + + def begin_new_cycle(self, new_item_id: int) -> None: + self.item_id = new_item_id + self.retry_count = 0 + self.fail_exhausted = False + self.history_recorded = False + + def clear_failure_for_user_select(self, new_item_id: int) -> None: + self.begin_new_cycle(new_item_id) + + +@dataclass +class PendingEvent: + kind: PendingEventKind + encoded: Optional[str] = None + reason: Optional[str] = None + exception: Optional[dict] = None + threshold: Optional[float] = None + payload: Optional[dict] = None + + +@dataclass +class PlaybackAttempt: + attempt_id: int + item_id: int + encoded_track_id: Optional[str] + play_seq: int + state: AttemptState = AttemptState.STARTING + armed: bool = False + play_in_flight: bool = False + intent: AttemptIntent = AttemptIntent.NONE + start_pos: int = 0 + end_time: Optional[int] = None + terminal_source: Optional[TerminalSource] = None + recorded_exception: Optional[dict] = None + pending: Deque[PendingEvent] = field(default_factory=deque) + previous_encoded: Optional[str] = None + + @property + def is_terminal(self) -> bool: + return self.terminal_source is not None or self.state in ( + AttemptState.FAILED, + AttemptState.ENDED, + AttemptState.ENDING, + ) + + def claim_terminal(self, source: TerminalSource) -> bool: + if self.terminal_source is not None: + return False + self.terminal_source = source + if source == TerminalSource.FINISHED: + self.state = AttemptState.ENDED + elif source in (TerminalSource.SKIP, TerminalSource.STOP, TerminalSource.CLEANUP, TerminalSource.RECOVERY_ABANDON): + self.state = AttemptState.ENDING + else: + self.state = AttemptState.FAILED + self.armed = False + self.play_in_flight = False + return True + + def enqueue_pending(self, event: PendingEvent) -> None: + if len(self.pending) >= MAX_PENDING_EVENTS: + self.pending.popleft() + self.pending.append(event) + + def take_pending(self) -> List[PendingEvent]: + events = list(self.pending) + self.pending.clear() + return events + + def discard_pending(self) -> None: + self.pending.clear() + + +@dataclass +class RetryReservation: + item_id: int + source: TerminalSource + cancelled: bool = False + + +@dataclass +class ForceplayOp: + op_id: int + target_item_id: int + superseded_item_id: Optional[int] + superseded_attempt_id: Optional[int] + state: ForceplayOpState = ForceplayOpState.PENDING_STOP + + +@dataclass +class AdvanceCommit: + reason: PlaybackAdvanceReason + from_item: Optional[QueueItem] + to_item: Optional[QueueItem] + from_attempt_id: Optional[int] + logs: List[Dict[str, Any]] = field(default_factory=list) + side_effect: str = "QUEUE_ADVANCE" + + +class PlaybackPolicy: + """Pure playback transition rules used by Player and unit tests.""" + + def encoded_matches(self, attempt: Optional[PlaybackAttempt], encoded: Optional[str]) -> bool: + if attempt is None: + return False + if not encoded or not attempt.encoded_track_id: + return True + return encoded == attempt.encoded_track_id + + def classify_event( + self, + attempt: Optional[PlaybackAttempt], + kind: PendingEventKind, + encoded: Optional[str], + ) -> EventDisposition: + if attempt is None: + return EventDisposition.IGNORE + if encoded and attempt.encoded_track_id and encoded != attempt.encoded_track_id: + return EventDisposition.IGNORE + if attempt.play_in_flight and not attempt.armed: + return EventDisposition.PENDING + if not attempt.armed and attempt.state == AttemptState.STARTING: + return EventDisposition.PENDING if attempt.play_in_flight else EventDisposition.IGNORE + if attempt.is_terminal and kind != PendingEventKind.TRACK_EXCEPTION: + return EventDisposition.IGNORE + if ( + attempt.previous_encoded + and encoded + and encoded == attempt.previous_encoded + and encoded == attempt.encoded_track_id + and attempt.state == AttemptState.STARTING + ): + return EventDisposition.AMBIGUOUS + return EventDisposition.APPLY + + def should_retry(self, item: Optional[QueueItem], source: TerminalSource) -> bool: + return False + + def forceplay_accepts_replaced( + self, + op: Optional[ForceplayOp], + attempt: Optional[PlaybackAttempt], + item_id: Optional[int], + ) -> bool: + if op is None or attempt is None or item_id is None: + return False + if op.state != ForceplayOpState.PENDING_STOP: + return False + if op.superseded_item_id != item_id: + return False + if op.superseded_attempt_id != attempt.attempt_id: + return False + return True + + +class PlaybackSession: + """Player-facing session state. Safe to construct without Discord/Lavalink.""" + + def __init__(self) -> None: + self._attempt_ids = 0 + self._play_seqs = 0 + self._forceplay_ids = 0 + self.attempt: Optional[PlaybackAttempt] = None + self.current_item: Optional[QueueItem] = None + self.reservation: Optional[RetryReservation] = None + self.forceplay: Optional[ForceplayOp] = None + self.desired_connected: bool = True + self.tearing_down: bool = False + self.watchdog_suspended: bool = False + self.stale_ops: Dict[int, int] = {} + self.reconcile_generation: int = 0 + self.pending_reconcile: bool = False + self.logs: List[Dict[str, Any]] = [] + self.queue_advance_logs: List[Dict[str, Any]] = [] + self.side_effects: List[str] = [] + self.history_writes: List[int] = [] + self.policy = PlaybackPolicy() + self.last_ended_encoded: Optional[str] = None + + def _next_attempt_id(self) -> int: + self._attempt_ids += 1 + return self._attempt_ids + + def next_play_seq(self) -> int: + self._play_seqs += 1 + return self._play_seqs + + def _next_forceplay_id(self) -> int: + self._forceplay_ids += 1 + return self._forceplay_ids + + def log(self, event: str, **fields: Any) -> None: + record = {"event": event, **fields} + self.logs.append(record) + if event == "QUEUE_ADVANCE": + self.queue_advance_logs.append(record) + + def remember_stale(self, item_id: int, play_seq: int) -> None: + self.stale_ops[item_id] = play_seq + while len(self.stale_ops) > STALE_OP_CAP: + oldest = next(iter(self.stale_ops)) + self.stale_ops.pop(oldest, None) + + def prune_stale(self, item_id: Optional[int] = None, play_seq: Optional[int] = None) -> None: + if item_id is not None: + recorded = self.stale_ops.get(item_id) + if recorded is None or play_seq is None or recorded <= play_seq: + self.stale_ops.pop(item_id, None) + if play_seq is not None: + for key, seq in [(k, s) for k, s in self.stale_ops.items() if s <= play_seq]: + if key != (self.current_item.item_id if self.current_item else None): + if self.attempt and key == self.attempt.item_id and self.attempt.play_seq > play_seq: + continue + if key not in ( + {self.attempt.item_id} if self.attempt else set() + ) and key not in ( + {self.forceplay.target_item_id, self.forceplay.superseded_item_id} + if self.forceplay else set() + ): + if seq < play_seq: + self.stale_ops.pop(key, None) + + def create_attempt( + self, + item: QueueItem, + *, + start_pos: int = 0, + previous_encoded: Optional[str] = None, + state: AttemptState = AttemptState.STARTING, + ) -> PlaybackAttempt: + encoded = getattr(item.track, "track_id", None) + attempt = PlaybackAttempt( + attempt_id=self._next_attempt_id(), + item_id=item.item_id, + encoded_track_id=encoded, + play_seq=self.next_play_seq(), + state=state, + armed=False, + start_pos=start_pos, + end_time=getattr(item.track, "end_time", None), + previous_encoded=previous_encoded, + ) + self.attempt = attempt + self.current_item = item + return attempt + + def mark_play_in_flight(self) -> Optional[int]: + if not self.attempt: + return None + self.attempt.play_in_flight = True + return self.attempt.play_seq + + def validate_play_completion(self, play_seq: int, item_id: int, attempt_id: int) -> bool: + if self.tearing_down or not self.desired_connected: + return False + if not self.attempt or not self.current_item: + return False + if self.attempt.is_terminal: + return False + if self.attempt.play_seq != play_seq: + return False + if self.attempt.attempt_id != attempt_id: + return False + if self.current_item.item_id != item_id: + return False + if item_id in self.stale_ops and self.stale_ops[item_id] >= play_seq: + return False + return True + + def arm_after_play_success(self) -> List[PendingEvent]: + if not self.attempt: + return [] + self.attempt.armed = True + self.attempt.play_in_flight = False + return self.attempt.take_pending() + + def fail_play_patch(self) -> List[PendingEvent]: + if not self.attempt: + return [] + self.attempt.play_in_flight = False + pending = list(self.attempt.pending) + self.attempt.discard_pending() + return pending + + def record_history_if_needed(self) -> bool: + item = self.current_item + if item is None or item.history_recorded: + return False + requester = getattr(item.track, "requester", None) + if requester is not None and getattr(requester, "bot", False): + item.history_recorded = True + return False + item.history_recorded = True + self.history_writes.append(item.item_id) + return True + + def enqueue_or_apply( + self, + kind: PendingEventKind, + *, + encoded: Optional[str] = None, + reason: Optional[str] = None, + exception: Optional[dict] = None, + threshold: Optional[float] = None, + payload: Optional[dict] = None, + ) -> EventDisposition: + disposition = self.policy.classify_event(self.attempt, kind, encoded) + if disposition == EventDisposition.PENDING and self.attempt: + self.attempt.enqueue_pending(PendingEvent( + kind=kind, + encoded=encoded, + reason=reason, + exception=exception, + threshold=threshold, + payload=payload, + )) + if disposition == EventDisposition.AMBIGUOUS: + self.log( + "AMBIGUOUS_EVENT", + kind=kind.value, + encoded=encoded, + attempt_id=self.attempt.attempt_id if self.attempt else None, + item_id=self.attempt.item_id if self.attempt else None, + ) + return disposition + + def apply_track_start(self, encoded: Optional[str] = None) -> bool: + if not self.attempt or self.attempt.is_terminal: + return False + if not self.policy.encoded_matches(self.attempt, encoded): + return False + if self.attempt.state in (AttemptState.STARTING, AttemptState.RECOVERING): + self.attempt.state = AttemptState.STARTED + self.log( + "TRACK_START", + item_id=self.attempt.item_id, + attempt_id=self.attempt.attempt_id, + ) + return True + return False + + def apply_track_exception(self, exception: Optional[dict] = None) -> bool: + if not self.attempt or self.attempt.is_terminal: + return False + if self.attempt.recorded_exception: + return False + self.attempt.recorded_exception = exception or {} + self.log( + "TRACK_EXCEPTION", + item_id=self.attempt.item_id, + attempt_id=self.attempt.attempt_id, + ) + return True + + def claim(self, source: TerminalSource) -> bool: + if not self.attempt: + return False + return self.attempt.claim_terminal(source) + + def begin_retry_reservation(self, source: TerminalSource) -> Optional[RetryReservation]: + if not self.current_item: + return None + self.reservation = RetryReservation(item_id=self.current_item.item_id, source=source) + return self.reservation + + def cancel_reservation(self) -> bool: + if not self.reservation or self.reservation.cancelled: + return False + self.reservation.cancelled = True + return True + + def consume_reservation(self) -> Optional[RetryReservation]: + reservation = self.reservation + if reservation is None or reservation.cancelled: + self.reservation = None + return None + if self.current_item is None or reservation.item_id != self.current_item.item_id: + self.reservation = None + return None + self.reservation = None + return reservation + + def commit_retry(self) -> Optional[PlaybackAttempt]: + item = self.current_item + reservation = self.consume_reservation() + if item is None or reservation is None: + return None + if not item.can_retry(): + return None + old = self.attempt + previous = old.encoded_track_id if old else None + if old: + old.state = AttemptState.ENDED + old.discard_pending() + item.mark_retry() + attempt = self.create_attempt(item, start_pos=getattr(item.track, "position", 0) or 0, previous_encoded=previous) + self.side_effects.append("RETRY") + self.log( + "RETRY", + item_id=item.item_id, + attempt_id=attempt.attempt_id, + retry_count=item.retry_count, + ) + return attempt + + def commit_advance( + self, + reason: PlaybackAdvanceReason, + next_item: Optional[QueueItem], + ) -> AdvanceCommit: + from_item = self.current_item + from_attempt = self.attempt + if from_attempt: + from_attempt.discard_pending() + if not from_attempt.is_terminal: + from_attempt.state = AttemptState.ENDED + self.last_ended_encoded = from_attempt.encoded_track_id + if from_item: + self.remember_stale(from_item.item_id, from_attempt.play_seq if from_attempt else 0) + self.reservation = None + if self.forceplay and self.forceplay.state == ForceplayOpState.ADVANCING: + self.forceplay.state = ForceplayOpState.DONE + self.current_item = next_item + to_attempt = None + if next_item is not None: + start = getattr(next_item.track, "position", 0) or 0 + to_attempt = self.create_attempt( + next_item, + start_pos=start, + previous_encoded=self.last_ended_encoded, + ) + else: + self.attempt = None + log_fields = { + "from_item": from_item.item_id if from_item else None, + "to_item": next_item.item_id if next_item else None, + "attempt_id": from_attempt.attempt_id if from_attempt else None, + "reason": str(reason), + "retry_count": from_item.retry_count if from_item else 0, + } + if from_item: + log_fields["track"] = getattr(from_item.track, "title", None) + if from_item is not None: + self.log("QUEUE_ADVANCE", **log_fields) + self.side_effects.append("QUEUE_ADVANCE") + if next_item is None: + self.prune_stale(play_seq=from_attempt.play_seq if from_attempt else None) + return AdvanceCommit( + reason=reason, + from_item=from_item, + to_item=next_item, + from_attempt_id=from_attempt.attempt_id if from_attempt else None, + ) + + def decide_after_failure(self, source: TerminalSource) -> str: + if self.tearing_down or not self.desired_connected: + return "TEARDOWN" + if self.policy.should_retry(self.current_item, source): + self.begin_retry_reservation(source) + return "RETRY" + if self.current_item: + self.current_item.mark_exhausted() + return "ADVANCE" + + def user_skip(self) -> str: + if self.tearing_down or not self.desired_connected: + return "TEARDOWN" + if self.reservation and not self.reservation.cancelled: + self.cancel_reservation() + return "ADVANCE" + if self.attempt and not self.attempt.is_terminal: + started = self.attempt.state == AttemptState.STARTED + in_flight = self.attempt.play_in_flight + play_seq = self.attempt.play_seq + item_id = self.attempt.item_id + if started: + self.attempt.intent = AttemptIntent.SKIP + if in_flight: + self.remember_stale(item_id, play_seq) + return "STOP_THEN_ADVANCE" + if self.attempt.claim_terminal(TerminalSource.SKIP): + if in_flight: + self.remember_stale(item_id, play_seq) + self.attempt.intent = AttemptIntent.SKIP + return "ADVANCE" + return "IGNORE" + if self.attempt and self.attempt.is_terminal and self.current_item: + return "ADVANCE" + return "IGNORE" + + def user_leave(self) -> None: + self.desired_connected = False + self.tearing_down = True + self.cancel_reservation() + if self.attempt: + self.attempt.intent = AttemptIntent.STOP + self.attempt.discard_pending() + if not self.attempt.is_terminal: + self.attempt.claim_terminal(TerminalSource.STOP) + if self.attempt.play_in_flight: + self.remember_stale(self.attempt.item_id, self.attempt.play_seq) + self.forceplay = None + + def begin_forceplay(self, target_item_id: int) -> ForceplayOp: + superseded_item = self.current_item.item_id if self.current_item else None + superseded_attempt = self.attempt.attempt_id if self.attempt else None + if self.reservation: + self.cancel_reservation() + if self.attempt and self.attempt.play_in_flight: + self.remember_stale(self.attempt.item_id, self.attempt.play_seq) + if self.attempt: + self.attempt.intent = AttemptIntent.FORCEPLAY + op = ForceplayOp( + op_id=self._next_forceplay_id(), + target_item_id=target_item_id, + superseded_item_id=superseded_item, + superseded_attempt_id=superseded_attempt, + ) + self.forceplay = op + return op + + def on_replaced(self) -> str: + if self.policy.forceplay_accepts_replaced( + self.forceplay, + self.attempt, + self.current_item.item_id if self.current_item else None, + ): + if self.attempt and self.attempt.claim_terminal(TerminalSource.SKIP): + self.forceplay.state = ForceplayOpState.ADVANCING + return "FORCEPLAY_ADVANCE" + return "RECONCILE" + self.log("TRACK_END", reason="replaced", attempt_id=self.attempt.attempt_id if self.attempt else None) + return "IGNORE" + + def on_stopped(self) -> str: + if not self.attempt: + return "IGNORE" + intent = self.attempt.intent + if intent == AttemptIntent.STOP or self.tearing_down or not self.desired_connected: + self.attempt.claim_terminal(TerminalSource.STOP) + return "TEARDOWN" + if intent == AttemptIntent.FORCEPLAY: + if self.forceplay and self.forceplay.state == ForceplayOpState.PENDING_STOP: + if self.attempt.claim_terminal(TerminalSource.SKIP): + self.forceplay.state = ForceplayOpState.ADVANCING + return "FORCEPLAY_ADVANCE" + return "IGNORE" + if intent == AttemptIntent.SKIP: + if self.attempt.claim_terminal(TerminalSource.SKIP): + return "ADVANCE" + return "IGNORE" + self.log("TRACK_END", reason="stopped", attempt_id=self.attempt.attempt_id) + return "IGNORE" + + def on_finished(self) -> str: + if not self.attempt: + return "IGNORE" + if self.attempt.state != AttemptState.STARTED: + self.log( + "AMBIGUOUS_EVENT", + kind="TRACK_END", + reason="finished", + attempt_id=self.attempt.attempt_id, + item_id=self.attempt.item_id, + ) + return "IGNORE" + if self.attempt.claim_terminal(TerminalSource.FINISHED): + return "ADVANCE" + return "IGNORE" + + def on_load_failed(self) -> str: + if not self.attempt: + return "IGNORE" + if not self.attempt.claim_terminal(TerminalSource.LOAD_FAILED): + return "IGNORE" + return self.decide_after_failure(TerminalSource.LOAD_FAILED) + + def on_stuck(self) -> str: + if not self.attempt: + return "IGNORE" + if not self.attempt.claim_terminal(TerminalSource.TRACK_STUCK): + return "IGNORE" + return self.decide_after_failure(TerminalSource.TRACK_STUCK) + + def on_watchdog(self) -> str: + if self.watchdog_suspended or self.tearing_down or not self.desired_connected: + return "IGNORE" + if not self.attempt or self.attempt.state != AttemptState.STARTING or not self.attempt.armed: + return "IGNORE" + if self.attempt.intent != AttemptIntent.NONE: + return "IGNORE" + if not self.attempt.claim_terminal(TerminalSource.WATCHDOG): + return "IGNORE" + return self.decide_after_failure(TerminalSource.WATCHDOG) + + def on_exception_fallback(self) -> str: + if not self.attempt: + return "IGNORE" + if self.attempt.is_terminal: + return "IGNORE" + if not self.attempt.claim_terminal(TerminalSource.EXCEPTION_FALLBACK): + return "IGNORE" + return self.decide_after_failure(TerminalSource.EXCEPTION_FALLBACK) + + def on_cleanup(self) -> str: + if self.tearing_down or not self.desired_connected: + if self.attempt: + self.attempt.claim_terminal(TerminalSource.CLEANUP) + return "TEARDOWN" + if self.attempt: + self.attempt.claim_terminal(TerminalSource.CLEANUP) + return "RECOVER" + + def retry_start_position(self, *, recovery_after_started: bool, last_position: int = 0) -> int: + item = self.current_item + intended = getattr(item.track, "position", 0) or 0 if item else 0 + if recovery_after_started: + return last_position or intended + return intended + + def intended_end_time(self) -> Optional[int]: + if not self.current_item: + return None + return getattr(self.current_item.track, "end_time", None) + + def mark_reconcile(self) -> int: + self.pending_reconcile = True + self.reconcile_generation += 1 + return self.reconcile_generation + + def take_reconcile(self, generation: int) -> bool: + if not self.pending_reconcile: + return False + if generation != self.reconcile_generation: + return False + if self.tearing_down or not self.desired_connected: + return False + self.pending_reconcile = False + return True diff --git a/voicelink/player.py b/voicelink/player.py index bb21ae1e..ac376847 100644 --- a/voicelink/player.py +++ b/voicelink/player.py @@ -50,14 +50,32 @@ from .objects import Track, Playlist from .filters import Filter, Filters from .enums import SearchType, LoopType, RequestMethod -from .events import VoicelinkEvent, TrackEndEvent, TrackStartEvent, TrackExceptionEvent -from .exceptions import VoicelinkException, FilterInvalidArgument, TrackInvalidPosition, FilterTagAlreadyInUse, DuplicateTrack +from .events import VoicelinkEvent, TrackStartEvent +from .exceptions import VoicelinkException, FilterInvalidArgument, TrackInvalidPosition, FilterTagAlreadyInUse, DuplicateTrack, NodeException from .placeholders import PlayerPlaceholder from .queue import Queue, QUEUE_TYPES from .mongodb import MongoDBHandler from .language import LangHandler from .views import InteractiveController from .utils import format_ms, dispatch_message +from .playback import ( + AttemptIntent, + AttemptState, + EventDisposition, + EXCEPTION_END_FALLBACK, + PendingEventKind, + PlaybackAdvanceReason, + PlaybackSession, + TerminalSource, + TRACK_START_TIMEOUT, +) + +_FAILURE_ADVANCE_REASONS = ( + PlaybackAdvanceReason.LOAD_FAILED, + PlaybackAdvanceReason.TRACK_STUCK, + PlaybackAdvanceReason.PLAY_REQUEST_FAILED, + PlaybackAdvanceReason.PLAYBACK_EXCEPTION, +) if TYPE_CHECKING: from .ipc import IPCClient @@ -134,11 +152,11 @@ def __init__( self._node = NodePool.get_node() self._current: Optional[Track] = None + self._current_item = None self._filters: Filters = Filters() self._paused: bool = False self._is_connected: bool = False self._ping: float = 0.0 - self._track_is_stuck = False self._position: int = 0 self._last_position: int = 0 @@ -146,6 +164,15 @@ def __init__( self._ending_track: Optional[Track] = None self._voice_state: dict = {} + self._playback = PlaybackSession() + self._attempt_lock = asyncio.Lock() + self._queue_lock = asyncio.Lock() + self._playback_log = logging.getLogger("vocard.playback") + self._watchdog_task: Optional[asyncio.Task] = None + self._exception_fallback_task: Optional[asyncio.Task] = None + self._desired_connected: bool = True + self._tearing_down: bool = False + self._had_started: bool = False self.controller: Union[Message, PartialMessage] = None self._updating: bool = False @@ -253,7 +280,7 @@ def data(self) -> dict: "guild_id": self._guild.id, "channel_id": self.channel.id, "queue": { - "tracks": [track.data for track in self.queue._queue], + "tracks": self.queue.session_track_data(), "position": self.queue._position, "repeat_mode": self.queue._repeat.current.name, "repeat_position": self.queue._repeat_position @@ -323,10 +350,10 @@ def build_embed(self, current_track: Track = None): return PlayerPlaceholder.build_embed(embed_form, self._ph) - async def send(self, method: RequestMethod, query: str = None, data: Union[Dict, str] = {}) -> Dict: + async def send(self, method: RequestMethod, query: str = None, data: Union[Dict, str] = {}, kind: str = None) -> Dict: """Sends an HTTP request to the node with the given method, query, and data.""" uri: str = f"sessions/{self._node._session_id}/players/{self._guild.id}" + (f"?{query}" if query else "") - return await self._node.send(method, query=uri, data=data) + return await self._node.send(method, query=uri, data=data, kind=kind) async def _update_state(self, data: dict) -> None: """Updates the player's state based on the provided data.""" @@ -384,40 +411,221 @@ async def on_voice_state_update(self, data: dict): await self._dispatch_voice_update({**self._voice_state, "event": data}) + def _sync_current(self) -> None: + self._playback.desired_connected = self._desired_connected + self._playback.tearing_down = self._tearing_down + self._current_item = self._playback.current_item + self._current = self._current_item.track if self._current_item else None + + def _detach_attempt_tasks_locked(self) -> List[asyncio.Task]: + tasks = [] + for name in ("_watchdog_task", "_exception_fallback_task"): + task: Optional[asyncio.Task] = getattr(self, name) + setattr(self, name, None) + if task and not task.done(): + task.cancel() + tasks.append(task) + return tasks + + def suspend_playback_watchdog(self) -> None: + self._playback.watchdog_suspended = True + task = self._watchdog_task + self._watchdog_task = None + if task and not task.done(): + task.cancel() + + def _log_playback(self, event: str, **fields) -> None: + self._playback.log(event, guild_id=getattr(self._guild, "id", None), **fields) + extras = " ".join(f"{key}={value}" for key, value in fields.items() if value is not None) + self._playback_log.info("%s %s", event, extras) + async def _dispatch_event(self, data: dict): """Dispatches an event based on the type of event data received.""" event_type = data.get("type") + if not event_type: + return event: VoicelinkEvent = getattr(events, event_type)(data, self) - - if isinstance(event, TrackEndEvent) and event.reason != "replaced": - self._current = None - - if isinstance(event, TrackExceptionEvent) and event.exception["message"] == "This content isn’t available.": - if self._node.yt_ratelimit: - await self._node.yt_ratelimit.flag_active_token() - event.dispatch(self._bot) if isinstance(event, TrackStartEvent): - self._ending_track = self._current + self._ending_track = event.track or self._current + await self.handle_track_start(event.track, encoded=event.encoded) self._logger.debug(f"Player in {self.guild.name}({self.guild.id}) dispatched event {event_type}.") - async def do_next(self): - """Processes the next track in the queue.""" - if self._current or self.is_playing or not self.channel: + async def handle_track_start(self, track, encoded: str = None) -> None: + encoded = encoded or getattr(track, "track_id", None) + applied = False + tasks: List[asyncio.Task] = [] + async with self._attempt_lock: + disposition = self._playback.enqueue_or_apply(PendingEventKind.TRACK_START, encoded=encoded) + if disposition not in (EventDisposition.APPLY, EventDisposition.AMBIGUOUS): + return + applied = self._playback.apply_track_start(encoded) + if applied: + self._had_started = True + self._sync_current() + tasks = self._detach_attempt_tasks_locked() + await self._absorb_cancelled(tasks) + if applied: + self._log_playback("TRACK_START", item_id=getattr(self._current_item, "item_id", None), track=getattr(track, "title", None)) + + async def handle_track_end(self, track, reason: str, encoded: str = None) -> None: + encoded = encoded or getattr(track, "track_id", None) + async with self._attempt_lock: + disposition = self._playback.enqueue_or_apply( + PendingEventKind.TRACK_END, encoded=encoded, reason=reason + ) + if disposition != EventDisposition.APPLY: + return + tasks = self._detach_attempt_tasks_locked() + if reason == "finished": + decision = self._playback.on_finished() + reason_enum = PlaybackAdvanceReason.FINISHED + elif reason == "loadFailed": + decision = self._playback.on_load_failed() + reason_enum = PlaybackAdvanceReason.LOAD_FAILED + elif reason == "replaced": + decision = self._playback.on_replaced() + reason_enum = PlaybackAdvanceReason.REPLACED + elif reason == "stopped": + decision = self._playback.on_stopped() + reason_enum = PlaybackAdvanceReason.MANUAL_SKIP + elif reason == "cleanup": + decision = self._playback.on_cleanup() + reason_enum = PlaybackAdvanceReason.CLEANUP + else: + self._log_playback("TRACK_END", reason=reason or "UNKNOWN") + decision = "IGNORE" + reason_enum = PlaybackAdvanceReason.UNKNOWN + self._sync_current() + await self._absorb_cancelled(tasks) + self._log_playback("TRACK_END", reason=reason, item_id=getattr(self._current_item, "item_id", None)) + await self._run_decision(decision, reason_enum) + + async def handle_track_stuck(self, track, threshold=None, encoded: str = None) -> None: + encoded = encoded or getattr(track, "track_id", None) + async with self._attempt_lock: + disposition = self._playback.enqueue_or_apply( + PendingEventKind.TRACK_STUCK, encoded=encoded, threshold=threshold + ) + if disposition != EventDisposition.APPLY: + return + tasks = self._detach_attempt_tasks_locked() + decision = self._playback.on_stuck() + self._sync_current() + await self._absorb_cancelled(tasks) + self._log_playback("TRACK_STUCK", item_id=getattr(self._current_item, "item_id", None)) + await self._run_decision(decision, PlaybackAdvanceReason.TRACK_STUCK) + + async def handle_track_exception(self, track, error: dict, encoded: str = None) -> None: + encoded = encoded or getattr(track, "track_id", None) + start_fallback = False + async with self._attempt_lock: + disposition = self._playback.enqueue_or_apply( + PendingEventKind.TRACK_EXCEPTION, encoded=encoded, exception=error + ) + if disposition != EventDisposition.APPLY: + return + started = self._playback.apply_track_exception(error) + start_fallback = started and self._exception_fallback_task is None + if start_fallback: + attempt_id = self._playback.attempt.attempt_id if self._playback.attempt else None + item_id = self._playback.attempt.item_id if self._playback.attempt else None + self._exception_fallback_task = self.bot.loop.create_task( + self._exception_fallback_worker(attempt_id, item_id) + ) + self._log_playback("TRACK_EXCEPTION", item_id=getattr(self._current_item, "item_id", None)) + + async def _exception_fallback_worker(self, attempt_id: Optional[int], item_id: Optional[int]) -> None: + try: + await asyncio.sleep(EXCEPTION_END_FALLBACK) + async with self._attempt_lock: + if not self._playback.attempt or self._playback.attempt.attempt_id != attempt_id: + return + if self._playback.attempt.item_id != item_id: + return + decision = self._playback.on_exception_fallback() + tasks = self._detach_attempt_tasks_locked() + self._sync_current() + await self._absorb_cancelled(tasks) + await self._run_decision(decision, PlaybackAdvanceReason.PLAYBACK_EXCEPTION) + except asyncio.CancelledError: return - - if self._paused: - self._paused = False - if self._track_is_stuck: - await asyncio.sleep(10) - self._track_is_stuck = False + async def _watchdog_worker(self, attempt_id: int, item_id: int, play_seq: int) -> None: + try: + await asyncio.sleep(TRACK_START_TIMEOUT) + async with self._attempt_lock: + if self._playback.watchdog_suspended: + return + if not self._playback.attempt or self._playback.attempt.attempt_id != attempt_id: + return + if self._playback.attempt.play_seq != play_seq or self._playback.attempt.item_id != item_id: + return + decision = self._playback.on_watchdog() + tasks = self._detach_attempt_tasks_locked() + self._sync_current() + await self._absorb_cancelled(tasks) + await self._run_decision(decision, PlaybackAdvanceReason.PLAY_REQUEST_FAILED) + except asyncio.CancelledError: + return - if not self.guild.me.voice: + async def _absorb_cancelled(self, tasks: List[asyncio.Task]) -> None: + if not tasks: + return + await asyncio.gather(*tasks, return_exceptions=True) + + async def _run_decision(self, decision: str, reason: PlaybackAdvanceReason) -> None: + if decision in (None, "IGNORE"): + return + if decision == "TEARDOWN": + return + if decision in ("ADVANCE", "FORCEPLAY_ADVANCE", "RETRY"): + await self._commit_and_play( + reason, + skip_exhausted=reason in _FAILURE_ADVANCE_REASONS, + ) + return + + async def _commit_and_play(self, reason: PlaybackAdvanceReason, *, skip_exhausted: bool = False) -> None: + cancelled = [] + async with self._attempt_lock: + async with self._queue_lock: + cancelled = self._detach_attempt_tasks_locked() + next_item = self.queue.get_item( + force_next=skip_exhausted and self.queue._repeat.mode == LoopType.TRACK, + skip_exhausted=skip_exhausted, + ) + commit = self._playback.commit_advance(reason, next_item) + self._sync_current() + self._had_started = False + if commit.from_item is not None: + self._log_playback( + "QUEUE_ADVANCE", + from_item=commit.from_item.item_id, + to_item=commit.to_item.item_id if commit.to_item else None, + attempt_id=commit.from_attempt_id, + reason=str(reason), + retry_count=commit.from_item.retry_count, + track=getattr(commit.from_item.track, "title", None), + ) + await self._absorb_cancelled(cancelled) + await self._after_new_item() + if self._current_item and self._desired_connected and not self._tearing_down: + await self._issue_play() + elif not self._current_item and self.autoplay and not self._tearing_down: + if await self.get_recommendations(): + await self.do_next() + + async def _after_new_item(self) -> None: + if not self.channel: + return + if self._paused: + self._paused = False + if self.guild and self.guild.me and not self.guild.me.voice: await self.connect(timeout=0.0, reconnect=True) - + self.pause_votes.clear() self.resume_votes.clear() self.skip_votes.clear() @@ -425,37 +633,132 @@ async def do_next(self): self.shuffle_votes.clear() self.stop_votes.clear() - track = self.queue.get() - - if not track: - if self.autoplay and await self.get_recommendations(): - return await self.do_next() + if not self._current_item: if self.queue.is_empty: self._schedule_inactive_cleanup_timer() else: - try: - await self.play(track, start=track.position) - except Exception as e: - self._logger.error(f"Something went wrong while playing music in {self.guild.name}({self.guild.id})", exc_info=e) - await asyncio.sleep(5) - return await self.do_next() - - if not track.requester.bot: - self._bot.loop.create_task(MongoDBHandler.update_user(track.requester.id, { - "$push": {"history": {"$each": [track.track_id], "$slice": -25}} - })) + self._cancel_inactive_cleanup_timer() await self.invoke_controller() await self.update_voice_status() + track = self._current if self.is_ipc_connected: await self.send_ws({ - "op": "trackUpdate", + "op": "trackUpdate", "currentQueuePosition": self.queue._position if track else self.queue._position + 1, "trackId": track.track_id if track else None, "isPaused": self._paused }) + async def _issue_play(self, *, recovery: bool = False) -> None: + async with self._attempt_lock: + item = self._playback.current_item + attempt = self._playback.attempt + if not item or not attempt or self._tearing_down or not self._desired_connected: + return + play_seq = attempt.play_seq + attempt_id = attempt.attempt_id + item_id = item.item_id + start = self._playback.retry_start_position( + recovery_after_started=recovery and self._had_started, + last_position=int(self._last_position or 0), + ) + end_time = self._playback.intended_end_time() + track = item.track + self._playback.mark_play_in_flight() + self._playback.watchdog_suspended = False + self._log_playback("PLAY_REQUEST", item_id=item_id, attempt_id=attempt_id, play_seq=play_seq, start=start) + try: + await self.play(track, start=start, end=end_time or 0) + except asyncio.CancelledError: + raise + except NodeException as e: + if not e.is_play_patch: + self._logger.error( + f"Non-play REST error during playback in {self.guild.name}({self.guild.id})", + exc_info=e, + ) + return + async with self._attempt_lock: + valid = self._playback.validate_play_completion(play_seq, item_id, attempt_id) + self._playback.fail_play_patch() + if valid and self._playback.claim(TerminalSource.PLAY_REST): + decision = self._playback.decide_after_failure(TerminalSource.PLAY_REST) + else: + decision = "IGNORE" + self._sync_current() + await self._run_decision(decision, PlaybackAdvanceReason.PLAY_REQUEST_FAILED) + return + except Exception as e: + self._logger.error(f"Something went wrong while playing music in {self.guild.name}({self.guild.id})", exc_info=e) + async with self._attempt_lock: + if self._playback.validate_play_completion(play_seq, item_id, attempt_id) and self._playback.claim(TerminalSource.PLAY_REST): + decision = self._playback.decide_after_failure(TerminalSource.PLAY_REST) + else: + decision = "IGNORE" + await self._run_decision(decision, PlaybackAdvanceReason.PLAY_REQUEST_FAILED) + return + + pending = [] + start_watchdog = False + async with self._attempt_lock: + if not self._playback.validate_play_completion(play_seq, item_id, attempt_id): + return + pending = self._playback.arm_after_play_success() + if self._playback.record_history_if_needed() and track.requester and not track.requester.bot: + self._bot.loop.create_task(MongoDBHandler.update_user(track.requester.id, { + "$push": {"history": {"$each": [track.track_id], "$slice": -25}} + })) + if self._playback.attempt and self._playback.attempt.state == AttemptState.STARTING: + start_watchdog = True + self._watchdog_task = self.bot.loop.create_task( + self._watchdog_worker(attempt_id, item_id, play_seq) + ) + self._sync_current() + for event in pending: + await self._apply_pending_event(event) + if start_watchdog and self._playback.attempt and self._playback.attempt.state != AttemptState.STARTING: + task = self._watchdog_task + self._watchdog_task = None + if task and not task.done(): + task.cancel() + await self._absorb_cancelled([task]) + + async def _apply_pending_event(self, event) -> None: + if event.kind == PendingEventKind.TRACK_START: + await self.handle_track_start(self._current, encoded=event.encoded) + elif event.kind == PendingEventKind.TRACK_END: + await self.handle_track_end(self._current, event.reason or "unknown", encoded=event.encoded) + elif event.kind == PendingEventKind.TRACK_EXCEPTION: + await self.handle_track_exception(self._current, event.exception or {}, encoded=event.encoded) + elif event.kind == PendingEventKind.TRACK_STUCK: + await self.handle_track_stuck(self._current, event.threshold, encoded=event.encoded) + + async def do_next(self): + """Processes the next track in the queue.""" + if self._tearing_down or not self.channel: + return + if self._current or self.is_playing: + return + await self._commit_and_play(PlaybackAdvanceReason.FINISHED) + + async def request_skip(self) -> None: + async with self._attempt_lock: + decision = self._playback.user_skip() + if decision == "STOP_THEN_ADVANCE": + try: + await self.send(method=RequestMethod.PATCH, data={"encodedTrack": None}, kind="PLAYER_STOP") + except Exception: + await self._commit_and_play(PlaybackAdvanceReason.MANUAL_SKIP) + return + if decision == "ADVANCE": + try: + await self.send(method=RequestMethod.PATCH, data={"encodedTrack": None}, kind="PLAYER_STOP") + except Exception: + pass + await self._commit_and_play(PlaybackAdvanceReason.MANUAL_SKIP) + async def invoke_controller(self): """Sends or updates the music controller message in the designated channel.""" if not self.settings.get('controller', True) or self._updating or not self.channel: @@ -511,6 +814,13 @@ async def is_position_fresh(self): async def teardown(self): """Cleans up the player and associated resources.""" + self._desired_connected = False + self._tearing_down = True + async with self._attempt_lock: + self._playback.user_leave() + cancelled = self._detach_attempt_tasks_locked() + self._sync_current() + await self._absorb_cancelled(cancelled) try: await MongoDBHandler.update_settings(self.guild.id, {"$set": { "last_active": (timeNow := round(time.time())), @@ -559,15 +869,35 @@ async def connect(self, *, timeout: float, reconnect: bool, self_deaf: bool = Tr """Connects the player to a voice channel.""" await self.guild.change_voice_state(channel=self.channel, self_deaf=True, self_mute=self_mute) self._node._players[self.guild.id] = self + self._desired_connected = True + self._tearing_down = False self._is_connected = True if self.channel: self._logger.debug(f"Player in {self.guild.name}({self.guild.id}) has been connected to {self.channel.name}({self.channel.id}).") - async def stop(self): + async def stop(self, *, intent: AttemptIntent = AttemptIntent.SKIP): """Stops the currently playing track.""" - self._current = None - await self.send(method=RequestMethod.PATCH, data={'encodedTrack': None}) + should_advance = False + async with self._attempt_lock: + if intent == AttemptIntent.FORCEPLAY: + target = self.queue.peek_next_item() + if target: + self._playback.begin_forceplay(target.item_id) + elif self._playback.attempt: + self._playback.attempt.intent = intent + if intent == AttemptIntent.SKIP and self._playback.attempt.state != AttemptState.STARTED: + decision = self._playback.user_skip() + should_advance = decision == "ADVANCE" + if self._playback.attempt and self._playback.attempt.play_in_flight: + self._playback.remember_stale(self._playback.attempt.item_id, self._playback.attempt.play_seq) + try: + await self.send(method=RequestMethod.PATCH, data={'encodedTrack': None}, kind="PLAYER_STOP") + except Exception: + if intent == AttemptIntent.SKIP: + should_advance = True + if should_advance: + await self._commit_and_play(PlaybackAdvanceReason.MANUAL_SKIP) async def disconnect(self, *, force: bool = False): """Disconnects the player from voice.""" @@ -613,11 +943,12 @@ async def play( if end or track.end_time: data["endTime"] = str(end or track.end_time) - await self.send(method=RequestMethod.PATCH, query=f"noReplace={ignore_if_playing}", data=data) + await self.send(method=RequestMethod.PATCH, query=f"noReplace={ignore_if_playing}", data=data, kind="PLAYER_PLAY") if self._node.yt_ratelimit: await self._node.yt_ratelimit.handle_request() - self._current = track + if not self._current: + self._current = track self._logger.debug(f"Player in {self.guild.name}({self.guild.id}) playing {track.title} from uri {track.uri} with a length of {track.length}") return self._current @@ -658,7 +989,7 @@ def _schedule_inactive_cleanup_timer(self) -> None: async def add_track(self, raw_tracks: Union[Track, List[Track]], *, start_time: int = 0, end_time: int = 0, at_front: bool = False, duplicate: bool = True) -> int: """Adds one or more tracks to the queue.""" tracks: List[Track] = [] - _duplicate_tracks = [] if self.queue._allow_duplicate and duplicate else [track.uri for track in self.queue._queue] + _duplicate_tracks = [] if self.queue._allow_duplicate and duplicate else self.queue.queued_uris() raw_tracks = raw_tracks[0] if isinstance(raw_tracks, List) and len(raw_tracks) == 1 else raw_tracks try: diff --git a/voicelink/pool.py b/voicelink/pool.py index 78e7a552..e0d9b1b0 100644 --- a/voicelink/pool.py +++ b/voicelink/pool.py @@ -256,24 +256,89 @@ async def _handle_payload(self, data: dict) -> None: elif op == "playerUpdate": await player._update_state(data) - async def send(self, method: RequestMethod, query: str, data: Union[dict, str] = {}) -> dict: + def _rest_kind(self, method: RequestMethod, query: str, data: Union[dict, str]) -> str: + q = (query or "").lower() + if "loadtracks" in q: + return "LOADTRACKS" + if q.startswith("sessions/") and "/players/" not in q: + return "SESSION" + if "/players/" in q: + if method == RequestMethod.GET: + return "PLAYER_GET" + if method == RequestMethod.DELETE: + return "PLAYER_DELETE" + if method == RequestMethod.PATCH and isinstance(data, dict): + if "encodedTrack" in data: + return "PLAYER_STOP" if data.get("encodedTrack") is None else "PLAYER_PLAY" + return "PLAYER_PATCH" + return "REST" + + def _truncate_body(self, raw: str) -> str: + encoded = (raw or "").encode("utf-8", "replace")[:512] + return encoded.decode("utf-8", "replace") + + async def send( + self, + method: RequestMethod, + query: str, + data: Union[dict, str] = {}, + kind: str = None, + ) -> dict: if not self._available: raise NodeNotAvailable(f"The node '{self._identifier}' is unavailable.") uri: str = f"{self._rest_uri}/{NODE_VERSION}/{query}" - async with self._session.request( - method=method.value, - url=uri, - headers={"Authorization": self._password}, - json=data - ) as resp: - if resp.status >= 300: - raise NodeException(f"Getting errors from Lavalink REST api") - - if method == RequestMethod.DELETE: - return await resp.json(content_type=None) - - return await resp.json() + rest_kind = kind or self._rest_kind(method, query, data) + timeout = aiohttp.ClientTimeout(total=30, connect=10) + try: + async with self._session.request( + method=method.value, + url=uri, + headers={"Authorization": self._password}, + json=data if not isinstance(data, str) else None, + timeout=timeout, + ) as resp: + raw = await resp.text() + if resp.status >= 300: + raise NodeException( + "Getting errors from Lavalink REST api", + method=method.value, + path=query, + kind=rest_kind, + status=resp.status, + body=self._truncate_body(raw), + node_id=self._identifier, + ) + + if method == RequestMethod.DELETE: + if not raw: + return {} + return await resp.json(content_type=None) + + if not raw: + return {} + try: + return await resp.json(content_type=None) + except Exception: + return {} + except NodeException: + raise + except asyncio.TimeoutError as exc: + raise NodeException( + "Lavalink REST request timed out", + method=method.value, + path=query, + kind=rest_kind, + node_id=self._identifier, + ) from exc + except aiohttp.ClientError as exc: + raise NodeException( + "Lavalink REST request failed", + method=method.value, + path=query, + kind=rest_kind, + node_id=self._identifier, + ) from exc async def connect(self) -> Node: """Initiates a connection with a Lavalink node and adds it to the node pool.""" diff --git a/voicelink/queue.py b/voicelink/queue.py index dc05ea07..7553e649 100644 --- a/voicelink/queue.py +++ b/voicelink/queue.py @@ -24,11 +24,12 @@ import discord from itertools import cycle -from typing import Optional, Tuple, Callable, Dict, List +from typing import Optional, Tuple, Callable, Dict, List, Union from .exceptions import QueueFull, OutofList from .objects import Track from .enums import LoopType +from .playback import QueueItem class LoopTypeCycle: def __init__(self) -> None: @@ -61,50 +62,121 @@ def __str__(self) -> str: class Queue: def __init__(self, size: int, allow_duplicate: bool, get_msg: Callable[[str], str]) -> None: - self._queue: List[Track] = [] + self._queue: List[QueueItem] = [] self._position: int = 0 self._size: int = size self._repeat: LoopTypeCycle = LoopTypeCycle() self._repeat_position: int = 0 self._allow_duplicate: bool = allow_duplicate + self._item_ids: int = 0 self.get_msg = get_msg - def get(self) -> Optional[Track]: - track = None + def _alloc_id(self) -> int: + self._item_ids += 1 + return self._item_ids + + def _wrap(self, item: Union[Track, QueueItem]) -> QueueItem: + if isinstance(item, QueueItem): + if item.item_id is None: + item.item_id = self._alloc_id() + return item + return QueueItem(item_id=self._alloc_id(), track=item) + + def _unwrap(self, item: Union[Track, QueueItem, None]) -> Optional[Track]: + if item is None: + return None + if isinstance(item, QueueItem): + return item.track + return item + + def _item_at(self, index: int) -> QueueItem: + return self._queue[index] + + def queued_uris(self) -> List[str]: + return [item.track.uri for item in self._queue] + + def session_track_data(self) -> List[dict]: + return [item.track.data for item in self._queue] + + def ipc_track_payloads(self) -> List[dict]: + return [ + { + "trackId": item.track.track_id, + "requesterId": str(getattr(getattr(item.track, "requester", None), "id", "")), + } + for item in self._queue + ] + + def peek_next_item(self) -> Optional[QueueItem]: + try: + return self._queue[self._position] + except IndexError: + return None + + def current_slot_item(self) -> Optional[QueueItem]: try: - track = self._queue[self._position - 1 if self._repeat.mode == LoopType.TRACK else self._position] - if self._repeat.mode != LoopType.TRACK: + return self._queue[self._position - 1] + except IndexError: + return None + + def get_item(self, *, force_next: bool = False, skip_exhausted: bool = False) -> Optional[QueueItem]: + use_track_loop = self._repeat.mode == LoopType.TRACK and not force_next + if skip_exhausted and use_track_loop: + use_track_loop = False + + if use_track_loop: + try: + item = self._queue[self._position - 1] + item.begin_new_cycle(self._alloc_id()) + return item + except IndexError: + return None + + scanned = 0 + limit = max(len(self._queue), 1) + 1 + while scanned < limit: + scanned += 1 + try: + item = self._queue[self._position] self._position += 1 - except: - if self._repeat.mode == LoopType.QUEUE: - try: - track = self._queue[self._repeat_position] - self._position = self._repeat_position + 1 - except IndexError: - self._repeat.set_mode(LoopType.OFF) + except IndexError: + if self._repeat.mode == LoopType.QUEUE: + try: + item = self._queue[self._repeat_position] + self._position = self._repeat_position + 1 + except IndexError: + self._repeat.set_mode(LoopType.OFF) + return None + else: + return None + if skip_exhausted and item.fail_exhausted: + continue + return item + return None - return track + def get(self, *, force_next: bool = False, skip_exhausted: bool = False) -> Optional[Track]: + return self._unwrap(self.get_item(force_next=force_next, skip_exhausted=skip_exhausted)) - def put(self, item: Track) -> int: + def put(self, item: Union[Track, QueueItem]) -> int: if self.count >= self._size: raise QueueFull(self.get_msg("queue.errors.queueFull").format(self._size)) - self._queue.append(item) + self._queue.append(self._wrap(item)) return self.count - def put_at_front(self, item: Track) -> int: + def put_at_front(self, item: Union[Track, QueueItem]) -> int: if self.count >= self._size: raise QueueFull(self.get_msg("queue.errors.queueFull").format(self._size)) - self._queue.insert(self._position, item) + self._queue.insert(self._position, self._wrap(item)) return 1 - def put_at_index(self, index: int, item: Track) -> None: + def put_at_index(self, index: int, item: Union[Track, QueueItem]) -> None: if self.count >= self._size: raise QueueFull(self.get_msg("queue.errors.queueFull").format(self._size)) - return self._queue.insert(self._position - 1 + index, item) + return self._queue.insert(self._position - 1 + index, self._wrap(item)) def skipto(self, index: int) -> None: if not 0 < index <= self.count: @@ -118,6 +190,14 @@ def backto(self, index: int) -> None: else: self._position -= index + def prepare_user_reselect(self) -> Optional[QueueItem]: + item = self.peek_next_item() + if item is None: + item = self.current_slot_item() + if item is not None: + item.clear_failure_for_user_select(self._alloc_id()) + return item + def history_clear(self, is_playing: bool) -> None: self._queue[:self._position - 1 if is_playing else self._position] = [] self._position = 1 if is_playing else 0 @@ -126,17 +206,25 @@ def clear(self) -> None: del self._queue[self._position:] def replace(self, queue_type: str, replacement: list) -> None: + wrapped: List[QueueItem] = [] + existing = {id(item.track): item for item in self._queue} + for entry in replacement: + if isinstance(entry, QueueItem): + wrapped.append(entry) + else: + prior = existing.get(id(entry)) + wrapped.append(prior if prior is not None else self._wrap(entry)) if queue_type == "queue": self.clear() - self._queue += replacement + self._queue += wrapped elif queue_type == "history": - self._queue[:self._position] = replacement + self._queue[:self._position] = wrapped def swap(self, track_index1: int, track_index2: int) -> Tuple[Track, Track]: try: adjusted_position = self._position - 1 self._queue[adjusted_position + track_index1], self._queue[adjusted_position + track_index2] = self._queue[adjusted_position + track_index2], self._queue[adjusted_position + track_index1] - return self._queue[adjusted_position + track_index1], self._queue[adjusted_position + track_index2] + return self._unwrap(self._queue[adjusted_position + track_index1]), self._unwrap(self._queue[adjusted_position + track_index2]) except IndexError: raise OutofList(self.get_msg("queue.errors.outOfList")) @@ -148,7 +236,7 @@ def move(self, target: int, to: int) -> Optional[Track]: item = self._queue[self._position + target - 1] self._queue.remove(item) self.put_at_index(to, item) - return item + return self._unwrap(item) except: raise OutofList(self.get_msg("queue.errors.outOfList")) @@ -162,12 +250,13 @@ def remove(self, index: int, index2: int = None, member: discord.Member = None) index, index2 = index2, index try: - removed_tracks: Dict[str, Track] = {} - for i, track in enumerate(self._queue[pos + index: pos + index2 + 1]): + removed_tracks: Dict[int, Track] = {} + for i, item in enumerate(list(self._queue[pos + index: pos + index2 + 1])): + track = self._unwrap(item) if member and track.requester != member: continue - self._queue.remove(track) + self._queue.remove(item) removed_tracks[pos + index + i] = track return removed_tracks @@ -176,13 +265,13 @@ def remove(self, index: int, index2: int = None, member: discord.Member = None) def history(self, incTrack: bool = False) -> List[Track]: if incTrack: - return self._queue[:self._position] - return self._queue[:self._position - 1] + return [self._unwrap(item) for item in self._queue[:self._position]] + return [self._unwrap(item) for item in self._queue[:self._position - 1]] def tracks(self, incTrack: bool = False) -> List[Track]: if incTrack: - return self._queue[self._position - 1:] - return self._queue[self._position:] + return [self._unwrap(item) for item in self._queue[self._position - 1:]] + return [self._unwrap(item) for item in self._queue[self._position:]] @property def count(self) -> int: @@ -205,14 +294,15 @@ def __init__(self, size: int, allow_duplicate: bool, get_msg) -> None: super().__init__(size, allow_duplicate, get_msg) self._set = set() - def put(self, item: Track) -> int: + def put(self, item: Union[Track, QueueItem]) -> int: if len(self._queue) >= self._size: raise QueueFull(self.get_msg("queue.errors.queueFull").format(self._size)) tracks = self.tracks(incTrack=True) lastIndex = len(tracks) + incoming = item.track if isinstance(item, QueueItem) else item for track in reversed(tracks): - if track.requester == item.requester: + if track.requester == incoming.requester: break lastIndex -= 1 self._set.clear() diff --git a/voicelink/views/controller.py b/voicelink/views/controller.py index 8de15a81..a0ec590e 100644 --- a/voicelink/views/controller.py +++ b/voicelink/views/controller.py @@ -135,9 +135,11 @@ async def callback(self, interaction: discord.Interaction): if not self.player.is_playing: self.player.queue.backto(1) + self.player.queue.prepare_user_reselect() await self.player.do_next() else: self.player.queue.backto(2) + self.player.queue.prepare_user_reselect() await self.player.stop() await self.send(interaction, "player.controls.back.success", interaction.user) @@ -420,6 +422,7 @@ async def callback(self, interaction: discord.Interaction): await interaction.response.defer() self.player.queue.skipto(int(self.values[0].split(". ")[0])) + self.player.queue.prepare_user_reselect() await self.player.stop() if self.player.settings.get("controller_msg", True):