Server-side fixes from DIRECT_MODE testing

Signed-off-by: Andrew Opalach <andrew@akon.city>

5 files changed, 419 insertions, 172 deletions

diff --git a/src/libsink/common.h b/src/libsink/common.h
index 254e246..475ef5f 100644
--- a/src/libsink/common.h
+++ b/src/libsink/common.h
@@ -2,6 +2,8 @@
 
 enum {
     CAMU_SINK_SET = 0,
+    CAMU_SINK_UNSET,
+    CAMU_SINK_SEQUENCE,
     CAMU_SINK_PAUSE,
     CAMU_SINK_SEEK
 };
diff --git a/src/libsink/desktop.c b/src/libsink/desktop.c
index 427a6f4..be3e797 100644
--- a/src/libsink/desktop.c
+++ b/src/libsink/desktop.c
@@ -3,7 +3,7 @@
 
 #ifdef CAMU_NO_MINIAUDIO_BACKENDS
 // @TODO: audio_null is incomplete in that it never actually reads from the buffers.
-//  This is an issue in CAMU_DIRECT_MODE because eventually the server-side packet_pool
+//  This is an issue in DIRECT_MODE because eventually the server-side packet_pool
 //  will be starved by pending audio packets.
 #define DESKTOP_NULL_AUDIO
 #endif
@@ -188,10 +188,10 @@ static void screen_callback(void *userdata, u8 op, void *opaque)
         camu_sink_reseek(&c->sink);
         break;
     case CAMU_SCREEN_AUDIO_TRACK:
-        camu_sink_adjust_audio_track(&c->sink, *(s32 *)opaque);
+        camu_sink_change_audio_track(&c->sink, *(s32 *)opaque);
         break;
     case CAMU_SCREEN_SUBTITLE_TRACK:
-        camu_sink_adjust_subtitle_track(&c->sink, *(s32 *)opaque);
+        camu_sink_change_subtitle_track(&c->sink, *(s32 *)opaque);
         break;
     case CAMU_SCREEN_SHUFFLE:
         camu_sink_shuffle(&c->sink);
@@ -219,9 +219,10 @@ static void screen_callback(void *userdata, u8 op, void *opaque)
 
 bool camu_desktop_open(struct camu_desktop *c, char *window_name)
 {
+    camu_screen_init(&c->scr);
     c->scr.callback = screen_callback;
     c->scr.userdata = c;
-    if (!(camu_screen_init(&c->scr) && camu_screen_create_window(&c->scr, window_name))) {
+    if (!camu_screen_create_window(&c->scr, window_name)) {
         log_error("Failed to create window.");
         return false;
     }
diff --git a/src/libsink/input_simulator.c b/src/libsink/input_simulator.c
index 5b9d6e6..883e631 100644
--- a/src/libsink/input_simulator.c
+++ b/src/libsink/input_simulator.c
@@ -7,7 +7,7 @@
 
 #include "input_simulator.h"
 
-// This is ignoring all thread-safety.
+// Currently ignoring all thread-safety.
 static s32 quit = 1;
 static struct nn_thread thread;
 
@@ -17,7 +17,8 @@ enum {
     TOGGLE_PAUSE,
     SEEK,
     SHUFFLE,
-    MARK, // count
+    COUNT,
+    RESEEK
 };
 
 static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata)
@@ -28,7 +29,7 @@ static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata)
         //nn_thread_sleep(NNWT_TS_FROM_USEC(2000000 + al_random_int(0, 1750000)));
         //nn_thread_sleep(NNWT_TS_FROM_USEC(60000));
         nn_thread_sleep(NNWT_TS_FROM_USEC(30000));
-        switch (al_random_int(0, MARK - 1)) {
+        switch (al_random_int(0, COUNT - 1)) {
         case SKIP: {
             s32 n = al_random_int(1, 5);
             log_info("SKIP (n: %d).", n);
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 5f61bd4..bad51d9 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -12,8 +12,6 @@
 #include "sink.h"
 #include "common.h"
 
-//#define CAMU_SINK_ONESHOT
-
 // Requested state of the sinks outputs.
 enum {
     SINK_PAUSED = 0,
@@ -55,17 +53,25 @@ enum {
     TOGGLE_PAUSE,
     SEEK,
     RESEEK,
+    AUDIO_TRACK,
+    SUBTITLE_TRACK,
     SHUFFLE,
     END
 };
 
 // Status reporting.
 enum {
-    NOTIFY_EMPTY = 1,
-    NOTIFY_NOT_EMPTY = 1 << 1,
-    NOTIFY_SEEK = 1 << 2
+    NOTIFY_CONNECTED = 1,
+    NOTIFY_DISCONNECTED = 1 << 1,
+    NOTIFY_EMPTY = 1 << 2,
+    NOTIFY_NOT_EMPTY = 1 << 3,
+    NOTIFY_ENTRY_ADDED = 1 << 4,
+    NOTIFY_SEEK = 1 << 5
 };
 
+// If the video output isn't running, the audio output might need to resolve a clock_pause().
+#define AUDIO_STOP_ON_CLOCK_PAUSE
+
 // Number of entries to keep buffered at one time.
 #define ENTRY_MAX_AGE 4
 #define SINK_LRU_MAX UINT16_MAX
@@ -130,8 +136,8 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX);
 //  video_buffer_callback()
 // CAMU_MIXER_THREADED or CAMU_SCREEN_THREADED:
 //  clock_callback()
-// client_callback()::LIANA_CLIENT_DATA
-// client_callback()::LIANA_CLIENT_EOF/ERRORED
+// client_callback(LIANA_CLIENT_DATA)
+// client_callback(LIANA_CLIENT_EOF/ERRORED)
 
 static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry)
 {
@@ -276,6 +282,17 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry)
     }
 }
 
+static struct lia_prefs *get_prefs_from_node_id(struct camu_sink *sink, u32 node_id)
+{
+    struct lia_prefs *prefs;
+    al_array_foreach_ptr(sink->node_prefs, i, prefs) {
+        if (prefs->node_id == node_id) {
+            return prefs;
+        }
+    }
+    return NULL;
+}
+
 // sink->current could be NULL.
 static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink)
 {
@@ -385,7 +402,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         return;
     }
     case ADD: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
         nn_packet_write_u8(packet, CAMU_LIST_ADD);
@@ -396,7 +413,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         break;
     }
     case SKIP: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
@@ -407,7 +424,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         break;
     }
     case TOGGLE_PAUSE: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
@@ -418,7 +435,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         break;
     }
     case SEEK: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
@@ -434,8 +451,30 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         if (sink->conn) nn_rpc_conn_disconnect(sink->conn);
         break;
     }
+    case AUDIO_TRACK: {
+        if (!sink->conn) return;
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+        struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id));
+        s32 i = prefs->index.audio + cmd->v.i;
+        if (i <= prefs->index.audio_max && i >= prefs->index.audio_min) {
+            prefs->index.audio = i;
+            nn_rpc_conn_disconnect(sink->conn);
+        }
+        break;
+    }
+    case SUBTITLE_TRACK: {
+        if (!sink->conn) return;
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+        struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id));
+        s32 i = prefs->index.subtitles + cmd->v.i;
+        if (i <= prefs->index.subtitles_max && i >= prefs->index.subtitles_min) {
+            prefs->index.subtitles = i;
+            nn_rpc_conn_disconnect(sink->conn);
+        }
+        break;
+    }
     case SHUFFLE: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
         nn_packet_write_u8(packet, CAMU_LIST_SHUFFLE);
@@ -443,7 +482,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         break;
     }
     case END: {
-        if (!sink->connected) return;
+        if (!sink->conn) return;
         struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
         nn_packet_write_str(packet, &sink->default_list);
@@ -485,21 +524,20 @@ static void mixer_callback(void *userdata, u8 op)
     }
 }
 
-static s32 entry_lru_compare(const void *a, const void *b)
+static s32 lru_compare_high_to_low(const void *a, const void *b)
 {
-    struct camu_sink_entry *aa = *((struct camu_sink_entry **)a);
-    struct camu_sink_entry *bb = *((struct camu_sink_entry **)b);
-    if (aa->lru > bb->lru) return -1;
-    else if (aa->lru < bb->lru) return 1;
+    struct camu_sink_entry *entry1 = *(struct camu_sink_entry **)a;
+    struct camu_sink_entry *entry2 = *(struct camu_sink_entry **)b;
+    if (entry1->lru > entry2->lru) return -1;
+    else if (entry1->lru < entry2->lru) return 1;
     return 0;
 }
 
 static void maybe_cleanup_old_entries(struct camu_sink *sink)
 {
-    al_array_sort(sink->entries, struct camu_sink_entry *, entry_lru_compare);
     // We check size <= MAX_AGE in the loops because sink->lru is not
     // indicative of the amount of entries we have loaded.
-    // The most obvious reason being it's incremented when moving back
+    // The most obvious reason being that it's incremented when moving back
     // and forth between two entries. As well as for buffer and queue operations.
     // We have to handle sink->lru wrapping in a step before the default case.
     // 0 65532 65533 65534 65535
@@ -507,26 +545,46 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
     // 0 1 2 65534 65535
     // 0 1 2 3 65535
     // 0 1 2 3 4
+    al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare_high_to_low);
+    array(struct camu_sink_entry *) cleanup;
+    al_array_init(cleanup);
     struct camu_sink_entry *entry;
     al_array_foreach_rev(sink->entries, i, entry) {
-        if (sink->entries.count <= ENTRY_MAX_AGE) return;
+        if (sink->entries.count <= ENTRY_MAX_AGE) {
+            goto out;
+        }
         u16 entry_age = (SINK_LRU_MAX - entry->lru) + sink->lru;
         if (entry->lru > sink->lru && entry_age > ENTRY_MAX_AGE) {
             al_array_remove_at(sink->entries, i);
-            maybe_disconnect_entry(entry);
+            al_assert(!al_array_contains(sink->entries, entry));
+            al_array_push(cleanup, entry);
         }
     }
     // Make sure we don't have to consider wrapping in the second loop.
-    if (sink->lru < ENTRY_MAX_AGE) return;
-    al_array_foreach_rev(sink->entries, i, entry) {
-        al_assert(sink->lru >= entry->lru);
-        if (sink->entries.count <= ENTRY_MAX_AGE) return;
-        u16 entry_age = sink->lru - entry->lru;
-        if (entry_age > ENTRY_MAX_AGE) {
-            al_array_remove_at(sink->entries, i);
-            maybe_disconnect_entry(entry);
+    if (sink->lru >= ENTRY_MAX_AGE) {
+        al_array_foreach_rev(sink->entries, i, entry) {
+            al_assert(sink->lru >= entry->lru);
+            if (sink->entries.count <= ENTRY_MAX_AGE) {
+                goto out;
+            }
+            u16 entry_age = sink->lru - entry->lru;
+            if (entry_age > ENTRY_MAX_AGE) {
+                al_array_remove_at(sink->entries, i);
+                al_assert(!al_array_contains(sink->entries, entry));
+                al_array_push(cleanup, entry);
+            }
         }
     }
+out:
+    // Why we can't disconnect an entry while iterating sink->entries:
+    // -> maybe_disconnect_entry() -> nn_packet_stream_disconnect() -> connection_closed_callback().
+    // -> In either CLIENT_REMOVE_BUFFERS or CLIENT_CLOSED, BLOCKING_SLEEP() runs the event loop.
+    // -> An entry other than the disconnected entry gets removed from sink->entries.
+    // -> On the next loop `i` is invalid.
+    al_array_foreach(cleanup, i, entry) {
+        maybe_disconnect_entry(entry);
+    }
+    al_array_free(cleanup);
 }
 
 static void maybe_run_previous(struct camu_sink *sink)
@@ -549,16 +607,16 @@ static void maybe_run_previous(struct camu_sink *sink)
 // being at the point it's freed.
 static void run_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key)
 {
-    bool removed = false;
+    bool ran = false;
     struct camu_sink_entry *previous;
     al_array_foreach(sink->previous, i, previous) {
         if (previous == key) {
             maybe_run_previous(sink);
-            removed = true;
+            ran = true;
             break;
         }
     }
-    log_trace("run_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed));
+    log_trace("run_previous_if_contains("ENTRY_FMT"), ran: %s.", ENTRY_ARG(key), BOOLSTR(ran));
 }
 
 static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry)
@@ -572,6 +630,7 @@ static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_
             break;
         }
     }
+    al_assert(!al_array_contains(sink->previous, entry));
     log_trace("maybe_remove_from_previous("ENTRY_FMT"), removed: %s.", ENTRY_ARG(entry), BOOLSTR(removed));
     return removed;
 }
@@ -600,6 +659,7 @@ static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool
     );
     // Clear the screen if skipping from a video to an audio-only entry.
     if (VIDEO_ENDED_OR_EMPTY(entry)) refresh_video_output(sink);
+    sink->notify_status |= NOTIFY_ENTRY_ADDED;
 }
 
 // Call this after setting state to ADDED because this entry might be in previous.
@@ -688,7 +748,10 @@ static void ensure_static_video_removed(struct camu_sink_entry *entry)
     if (!VIDEO_EMPTY(entry) && VIDEO_IS_STATIC(entry)) {
         remove_entry_video_buffer(entry);
         struct camu_sink *sink = entry->sink;
+        // Unlock to wait, like in CLIENT_REMOVE_BUFFERS.
+        nn_mutex_unlock(&sink->lock);
         while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); }
+        nn_mutex_lock(&sink->lock);
     }
 }
 
@@ -705,7 +768,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
         if (suspended) {
             al_assert(suspended == current);
             // It should be impossible for a buffer to be QUEUED while it's entry is suspended.
-            // Even for a static video buffer because we wouldn't know if it was static yet.
+            // Even for a static video buffer because we wouldn't have known if it was static yet.
             al_assert(AUDIO_STATE(suspended) != BUFFER_QUEUED);
             al_assert(VIDEO_STATE(suspended) != BUFFER_QUEUED);
             ensure_static_video_removed(suspended);
@@ -722,7 +785,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
     if (dangling_target) log_trace("Ignoring dangling target.");
     bool stop_video = dangling_target;
     if (!dangling_target) {
-        target->audio.ignore_paused = false;
+        target->audio.ignore_pause = false;
         if (!sink->local && !target->paused) {
             camu_audio_buffer_resync(&target->audio.buf);
         }
@@ -761,9 +824,11 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta
     al_assert(target != current);
     al_assert(!sink->target);
     sink->target = target;
+    // Checking !current here allows the list to send set(PAUSE_BOTH) from handle_add_sink().
     bool immediate = !current || ENTRY_ENDED(current);
     if (current) {
-        current->audio.ignore_paused = true;
+        current->audio.ignore_pause = true;
+        camu_clock_set_pause_for_swap(&current->clock);
         immediate |= camu_clock_pause(&current->clock, at);
     }
     if (immediate) {
@@ -772,11 +837,13 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta
     }
 }
 
+//#define SINK_ONESHOT
+
 static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
     log_debug("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry));
     run_previous_if_contains(sink, entry);
-#ifdef CAMU_SINK_ONESHOT
+#ifdef SINK_ONESHOT
     sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL);
     return false;
 #endif
@@ -815,22 +882,27 @@ static void audio_buffer_callback(void *userdata, u8 op)
         lia_vcr_uncork(entry->audio.track);
         break;
     case CAMU_BUFFER_PAUSED: // Comes from audio read() thread.
+#ifndef AUDIO_STOP_ON_CLOCK_PAUSE
         nn_mutex_lock(&sink->lock);
         // entry->paused could plausibly be false here if the lock was held
         // by pause_command_callback() to resume. This can be simulated by
         // calling list_toggle_pause() twice in server/list_action_callback()
         // for each sink request. Spaced by an nn_event_loop_sleep(~15500us)
         // (no_video, MINIAUDIO_LOW_LATENCY mode).
-        if (!entry->audio.ignore_paused && entry->paused) {
+        if (!entry->audio.ignore_pause && entry->paused) {
             log_info("Audio buffer paused.");
             queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO));
         }
         nn_mutex_unlock(&sink->lock);
+#endif
         break;
     case CAMU_BUFFER_EOF:
     case CAMU_BUFFER_ERRORED: {
         bool error = op == CAMU_BUFFER_ERRORED;
         if (error) {
+            // There's two main considerations for whether to EJECT_ENTRY on error or not.
+            // 1. If the other buffer would have continued working, that's an unoptimal user experience.
+            // 2. If the server continues sending frames for the erorred buffer, that's wasteful.
             log_error("Audio buffer errored.");
         } else {
             log_debug("Audio EOF.");
@@ -873,7 +945,7 @@ static void video_buffer_callback(void *userdata, u8 op)
         lia_vcr_uncork(entry->video.track);
         break;
     case CAMU_BUFFER_EOF:
-    case CAMU_BUFFER_ERRORED: {
+    case CAMU_BUFFER_ERRORED: { // See notes in audio_buffer_callback().
         bool error = op == CAMU_BUFFER_ERRORED;
         if (error) {
             log_error("Video buffer errored.");
@@ -889,7 +961,6 @@ static void video_buffer_callback(void *userdata, u8 op)
             return;
         }
         // Video state could be ADDED, SET_OR_BUFFERED, or CONFIGURED.
-        // See note about threaded outputs in audio_buffer_callback(EOF|ERRORED).
         if (VIDEO_STATE(entry) == BUFFER_ADDED) {
             remove_entry_video_buffer(entry);
         }
@@ -920,6 +991,9 @@ static void clock_callback(void *userdata, u8 op)
                 sink->target = NULL;
             } else if (entry->paused) {
                 queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
+#ifdef AUDIO_STOP_ON_CLOCK_PAUSE
+                queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO));
+#endif
             }
         }
         nn_mutex_unlock(&sink->lock);
@@ -943,7 +1017,7 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s
     camu_audio_buffer_set_latency(&entry->audio.buf, audio);
     camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
     camu_video_buffer_set_latency(&entry->video.buf, video);
-    log_info("video_latency: %fs (%u frames), audio_latency: %fs.", video, frames, audio);
+    log_debug("video_latency: %fs (%u frames), audio_latency: %fs.", video, frames, audio);
     if (sink->local) {
         camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video);
         // When the video buffer starts the clock, we have to consider the audio
@@ -963,7 +1037,9 @@ static void run_queue_by_opaque(struct camu_sink *sink, void *opaque)
     camu_queue_lock(sink->queue);
     struct camu_sink_cmd *cmd;
     al_array_foreach_ptr(sink->queue.a, i, cmd) {
-        if (cmd->opaque == opaque) {
+        // Only run "Sink operations", otherwise in DIRECT_MODE we could
+        // end up with an insane call stack and likely deadlock.
+        if (cmd->op < ADD && cmd->opaque == opaque) {
             handle_sink_cmd(sink, cmd);
             al_array_remove_at_iter(sink->queue.a, i);
         }
@@ -1043,6 +1119,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // This is mainly to assert DETACHED handling.
         al_assert(!VIDEO_ENDED(entry) && !AUDIO_ENDED(entry));
         evaluate_and_set_buffer_params(sink, entry);
+        struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id));
+        al_assert(prefs);
+        *prefs = entry->client.prefs;
         nn_mutex_unlock(&sink->lock);
         break;
     }
@@ -1091,7 +1170,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             }
             if (sink->target) {
                 // This is necessary to avoid re-adding an entry with an in-between clock state.
-                // See note in clock.c::camu_clock_seek().
+                // See note in clock.c:camu_clock_seek().
                 switch_to(sink, sink->target);
                 sink->target = NULL;
             } else if (rec->reconnect) {
@@ -1101,8 +1180,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         }
 
         // This entry might be in previous if it was added to previous then,
-        //  1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered.
-        //  2. it was seeked.
+        // 1. it was seeked.
+        // 2. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered.
+        // 3. it got disconnected server-side.
         run_previous_if_contains(sink, entry);
         // AUDIO/VIDEO_STATE() could be INIT at this point, even if entry = current.
 
@@ -1137,7 +1217,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         }
         // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER is not thread-safe.
         // Meaning it must be run on the event loop. So, resolve any queued REMOVE_BUFFER
-        // requests so we can safely block the loop.
+        // requests so we can safely block the loop. As of now, remove_entry_video_buffer()
+        // never queues REMOVE_BUFFER. If that were to change, we may need to reevaluate
+        // ensure_static_video_removed() based on this.
         run_queue_by_opaque(sink, entry);
 
         // Unlock to wait.
@@ -1202,7 +1284,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_STATIC(entry);
         if (!AUDIO_EMPTY(entry)) {
             camu_audio_buffer_reset(&entry->audio.buf);
-            // no_video is set to true in video BUFFER_EOF as a fail-safe. Reset it here.
+            // no_video is set to true in video BUFFER_EOF as a fail-safe, reset it here.
             camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
         }
         if (!ignore_video) {
@@ -1270,6 +1352,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 
         // If a client is closed after a failed reconnect, a static video buffer could still be added.
         if (rec->reconnect) ensure_static_video_removed(entry);
+
         al_assert(AUDIO_STATE(entry) != BUFFER_ADDED);
         al_assert(VIDEO_STATE(entry) != BUFFER_ADDED);
 
@@ -1293,10 +1376,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
                     queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
                 }
             }
-            // If current was never fully added we need to call this here.
+            // If current never got to after_add_entry(), run previous here.
             maybe_run_previous(sink);
         }
 
+        // If entry was in previous, it should have been removed in CLIENT_REMOVE_BUFFERS.
+        al_assert(!maybe_remove_from_previous(sink, entry));
+
         nn_mutex_unlock(&sink->lock);
 
         lia_client_free(&entry->client);
@@ -1318,15 +1404,28 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id)
     entry->disconnected = false;
     entry->ended = false;
 
-    camu_clock_init(&entry->clock, clock_callback, entry);
-
     entry->client.callback = client_callback;
     entry->client.userdata = entry;
     entry->client.renderer = sink->video.renderer;
-    entry->client.prefs = sink->prefs;
+    struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(id));
+    if (!prefs) {
+        prefs = &sink->prefs;
+        prefs->node_id = REMOTE_ENTRY_ID(id);
+        prefs->index.audio = -1;
+        prefs->index.audio_min = INT32_MAX;
+        prefs->index.audio_max = 0;
+        prefs->index.video = -1;
+        prefs->index.subtitles = -1;
+        prefs->index.subtitles_min = INT32_MAX;
+        prefs->index.subtitles_max = 0;
+        al_array_push(sink->node_prefs, *prefs);
+    }
+    entry->client.prefs = *prefs;
+
+    camu_clock_init(&entry->clock, clock_callback, entry);
 
     AUDIO_STATE(entry) = BUFFER_INIT;
-    entry->audio.ignore_paused = false;
+    entry->audio.ignore_pause = false;
     camu_audio_buffer_init(&entry->audio.buf, &entry->clock);
     entry->audio.buf.callback = audio_buffer_callback;
     entry->audio.buf.userdata = entry;
@@ -1342,8 +1441,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id)
     union { f64 f; u64 u; } fv = { .u = id };
     entry->video.buf.seek_pts = fv.f;
 
-    al_array_push(sink->entries, entry);
-
     return entry;
 }
 
@@ -1356,15 +1453,6 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u64 id)
     return NULL;
 }
 
-static void unset_current(struct camu_sink *sink)
-{
-    struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->lock);
-    if (current) {
-        maybe_disconnect_entry(current);
-    }
-}
-
 static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     struct nn_packet *packet, struct nn_packet *rpacket)
 {
@@ -1372,16 +1460,11 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     (void)rpacket;
 
     u8 op = nn_packet_read_u8(packet);
-    if (op == LIANA_SINK_UNSET) {
-        nn_mutex_lock(&sink->lock);
-        unset_current(sink);
-        nn_mutex_lock(&sink->lock);
-        goto out;
-    }
 
     // Liana node info.
-    str addr;
-    nn_packet_read_str(packet, &addr);
+    str addr, paddr;
+    nn_packet_read_str(packet, &paddr);
+    al_str_clone(&addr, &paddr);
     u16 port = nn_packet_read_u16(packet);
     u32 node_id = nn_packet_read_u32(packet);
 
@@ -1389,34 +1472,25 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
     s32 sequence = nn_packet_read_s32(packet);
     u64 at = nn_packet_read_u64(packet);
+    if (sink->local) at = 0;
     u64 pos = nn_packet_read_u64(packet);
     u8 pause = nn_packet_read_u8(packet);
     u32 reset_token = nn_packet_read_u32(packet);
 
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     bool create = !entry;
-    if (create) entry = create_entry(sink, id);
-    entry->sequence = sequence;
-    entry->lru = sink->lru;
-    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
-    entry->reset_token = reset_token;
     if (create) {
+        entry = create_entry(sink, id);
+        al_array_push(sink->entries, entry);
         entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
         camu_clock_set(&entry->clock, pos / 1000000.0);
-        lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos);
     }
-
-    // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect().
-    nn_mutex_lock(&sink->lock);
-
-    // lia_client_connect() can fail and call connection_closed_callback() in-line. Meaning this entry
-    // could already be disconnected here.
-    if (!al_array_contains(sink->entries, entry)) {
-        unset_current(sink);
-        goto out;
-    }
-
-    struct camu_sink_entry *current = sink->current;
+    entry->sequence = sequence;
+    entry->lru = sink->lru;
+    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
+    entry->reset_token = reset_token;
 
     if (op == LIANA_SINK_BUFFER) {
         log_trace("buffered("ENTRY_FMT"), created: %s.", ENTRY_ARG(entry), BOOLSTR(create));
@@ -1426,12 +1500,32 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
         goto out;
     }
 
-    if (sink->local) at = 0;
+    nn_mutex_lock(&sink->lock);
 
-    // For target to be set that must mean current is set, armed to pause, and not ended.
+    // For target to be set that means: current is set, armed to pause, and not ended.
+    struct camu_sink_entry *current = sink->current;
     struct camu_sink_entry *prev_target = sink->target;
+
+    // Go against the list and try to handle this case with a pause_and_swap_to() as it's much cleaner.
+    if (current && (CONNECTION_NUMBER(current->id) != CONNECTION_NUMBER(entry->id)) &&
+        (REMOTE_ENTRY_ID(current->id) == REMOTE_ENTRY_ID(entry->id))) {
+        if (prev_target) {
+            sink->target = entry;
+            // Don't assume anything about the list-side pause state of current.
+            if (pause == LIANA_PAUSE_RESUME) {
+                // This will result in a slight jump due to differing `at`s.
+                camu_clock_resume(&entry->clock, at);
+            }
+            nn_mutex_unlock(&sink->lock);
+            goto out;
+        } else if (!current->paused) {
+            pause += LIANA_PAUSE_PAUSE;
+        }
+    }
+
     log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry),
         lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
+
     switch (pause) {
     case LIANA_PAUSE_NONE:
         if (prev_target) {
@@ -1499,14 +1593,61 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
         break;
     }
 
-out:
     nn_mutex_unlock(&sink->lock);
+
+out:
+    if (create) {
+        lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos);
+    }
+
+    al_str_free(&addr);
+
     if (op != LIANA_SINK_BUFFER) {
         maybe_cleanup_old_entries(sink);
     }
 
+    return false;
+}
+
+static bool unset_command_callback(void *userdata, struct nn_rpc_connection *conn,
+    struct nn_packet *packet, struct nn_packet *rpacket)
+{
+    struct camu_sink *sink = (struct camu_sink *)userdata;
+    (void)rpacket;
+
+    nn_packet_stream_return_packet(conn->stream, packet);
+
+    nn_mutex_lock(&sink->lock);
+    struct camu_sink_entry *current = sink->current;
+    nn_mutex_unlock(&sink->lock);
+
+    log_trace("unset("ENTRY_FMT")", ENTRY_ARG(current));
+
+    if (current) {
+        maybe_disconnect_entry(current);
+    }
+
+    return false;
+}
+
+static bool sequence_command_callback(void *userdata, struct nn_rpc_connection *conn,
+    struct nn_packet *packet, struct nn_packet *rpacket)
+{
+    struct camu_sink *sink = (struct camu_sink *)userdata;
+    (void)rpacket;
+
+    u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
+    s32 sequence = nn_packet_read_s32(packet);
+
     nn_packet_stream_return_packet(conn->stream, packet);
 
+    struct camu_sink_entry *entry = get_entry_from_id(sink, id);
+    if (!entry) goto out;
+    entry->sequence = sequence;
+
+    log_trace("sequence("ENTRY_FMT"), sequence: %d", ENTRY_ARG(entry), sequence);
+
+out:
     return false;
 }
 
@@ -1519,23 +1660,32 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
     s32 sequence = nn_packet_read_s32(packet);
     u64 at = nn_packet_read_u64(packet);
+    if (sink->local) at = 0;
     u8 pause = nn_packet_read_u8(packet);
 
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     if (!entry) goto out;
-    nn_mutex_lock(&sink->lock);
-    // As long as the list discards skips with a non-current sequence, this should hold true.
+    // List-side we discard skip/pause commands with a non-current sequence and send SINK_SEQUENCE
+    // when an erroring command is reduced to only a sequence change (in terms of what the sink
+    // cares about). That being said, as of now, there's no technical reason to enforce this.
     al_assert(entry->sequence == sequence);
+
+    nn_mutex_lock(&sink->lock);
+
     log_trace("pause("ENTRY_FMT"), %s, audio_state: %hhu, video_state: %hhu.",
         ENTRY_ARG(entry), lia_pause_op_name(pause), AUDIO_STATE(entry), VIDEO_STATE(entry));
-    if (sink->local) at = 0;
+
     switch (pause) {
     case LIANA_PAUSE_PAUSE: {
         entry->paused = true;
-        if (camu_clock_pause(&entry->clock, at)) {
-            if (entry == sink->current) {
+        bool immediate = camu_clock_pause(&entry->clock, at);
+        if (entry == sink->current) {
+            if (immediate) {
                 queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
             }
+            refresh_video_output(sink);
         }
         log_info("Clock paused.");
         // Audio will be stopped in a BUFFER_PAUSED callback.
@@ -1560,11 +1710,10 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
         break;
     }
     }
+
     nn_mutex_unlock(&sink->lock);
 
 out:
-    nn_packet_stream_return_packet(conn->stream, packet);
-
     return false;
 }
 
@@ -1580,25 +1729,34 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn
     u64 pos = nn_packet_read_u64(packet);
     u32 reset_token = nn_packet_read_u32(packet);
 
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     if (!entry) goto out;
-    nn_mutex_lock(&sink->lock);
+    // List-side seek() allows a non-current sequence. See comment in pause_command_callback().
     entry->sequence = sequence;
+
+    nn_mutex_lock(&sink->lock);
+
     log_trace("seek("ENTRY_FMT", %.2f), reset_token: %u.", ENTRY_ARG(entry), pos / 1000000.0, reset_token);
+
     entry->reset_token = reset_token;
     sink->notify_status |= NOTIFY_SEEK;
+    refresh_video_output(sink);
+
     nn_mutex_unlock(&sink->lock);
+
     // The rest of the seek is handled in CLIENT_REMOVE_BUFFERS/RESUME_AT/RECONNECTED.
     lia_client_seek(&entry->client, pos, at);
 
 out:
-    nn_packet_stream_return_packet(conn->stream, packet);
-
     return false;
 }
 
 static struct nn_rpc_command commands[] = {
     { .op = CAMU_SINK_SET, .callback = set_command_callback, .userdata = NULL },
+    { .op = CAMU_SINK_UNSET, .callback = unset_command_callback, .userdata = NULL },
+    { .op = CAMU_SINK_SEQUENCE, .callback = sequence_command_callback, .userdata = NULL },
     { .op = CAMU_SINK_PAUSE, .callback = pause_command_callback, .userdata = NULL },
     { .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL }
 };
@@ -1606,17 +1764,17 @@ static struct nn_rpc_command commands[] = {
 static void identify_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
+    nn_packet_stream_return_packet(conn->stream, packet);
 #ifndef CAMU_DIRECT_MODE
     if (sink->type == NNWT_SOCKET_UNIX) {
-        log_info("Sink connected to %.*s.", al_str_x(&sink->addr));
+        log_info("Connected to %.*s.", al_str_x(&sink->addr));
     } else {
-        log_info("Sink connected to %.*s:%hu.", al_str_x(&sink->addr), sink->port);
+        log_info("Connected to %.*s:%hu.", al_str_x(&sink->addr), sink->port);
     }
 #else
     (void)sink;
     log_info("Sink directly bridged to server.");
 #endif
-    nn_packet_stream_return_packet(conn->stream, packet);
 }
 
 static void identify_on_connection(struct camu_sink *sink)
@@ -1630,49 +1788,71 @@ static void identify_on_connection(struct camu_sink *sink)
 static void connection_callback(void *userdata, struct nn_rpc_connection *conn)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
-    nn_timer_stop(&sink->reconnect_timer);
-    if (sink->conn) al_assert(sink->conn == conn);
     sink->conn = conn;
-    sink->connected = true;
-    refresh_video_output(sink); // For OSD.
+    sink->connecting = false;
     sink->connection_number = al_u16_inc_wrap(sink->connection_number);
     if (sink->connection_number == 0) sink->connection_number = 1;
+    nn_timer_stop(&sink->reconnect_timer);
+    sink->notify_status |= NOTIFY_CONNECTED;
+    refresh_video_output(sink);
+}
+
+static void ready_callback(void *userdata, struct nn_rpc_connection *conn)
+{
+    struct camu_sink *sink = (struct camu_sink *)userdata;
+    (void)conn;
     identify_on_connection(sink);
 }
 
+static inline void rpc_connect(struct camu_sink *sink)
+{
+#ifdef CAMU_DIRECT_MODE
+    nn_multiplex_direct_connect(sink->client.conn->stream, CAMU_MULTIPLEX_RPC);
+#else
+    sink->connecting = nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port);
+#endif
+}
+
+static inline void rpc_reconnect(struct camu_sink *sink)
+{
+#ifdef CAMU_DIRECT_MODE
+    nn_multiplex_direct_reconnect(sink->client.conn->stream);
+#else
+    sink->connecting = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
+#endif
+}
+
 static void reconnect_timer_callback(void *userdata, struct nn_timer *timer)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
     (void)timer;
-    if (sink->conn) {
+    if (sink->connecting) {
         // If the client was connecting, reconnect() will force a disconnect before reconnecting.
         log_warn("Forcing reconnect due to timeout.");
     }
-    sink->conn = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
+    rpc_reconnect(sink);
 }
 
 static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
-    // @TODO: This is broken for an immediately failing reconnect().
-    //  If the nn_rpc_reconnect() in this function fails and recurses on connection_closed_callback(),
-    //  sink->conn will be NULL and we will not attempt to reconnect.
-    bool reconnect = sink->conn != NULL;
-    bool disconnected = sink->conn && sink->connection_number > 0;
+    bool was_connected = sink->conn && sink->connection_number > 0;
+    sink->connecting = false;
     if (sink->conn) {
         al_assert(sink->conn == conn);
         sink->conn = NULL;
-        sink->connected = false;
-    } else {
-        al_assert(!reconnect || !sink->connected);
+        sink->notify_status |= NOTIFY_DISCONNECTED;
+        refresh_video_output(sink);
     }
-    if (reconnect) {
-        if (disconnected) {
+    if (!sink->closed) {
+        if (was_connected) {
             log_warn("Connection to server closed, attempting reconnect...");
-            nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
+#ifndef CAMU_DIRECT_MODE
+            nn_timer_again(&sink->reconnect_timer);
+#endif
+            rpc_reconnect(sink);
         } else {
             log_warn("Failed to connect to server, trying again...");
-            nn_timer_again(&sink->reconnect_timer);
         }
     }
 }
@@ -1693,13 +1873,15 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     struct camu_mixer *mixer, struct camu_renderer *renderer)
 {
     sink->loop = loop;
-    nn_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink);
+    nn_rpc_init(&sink->client, sink->loop, connection_callback, ready_callback, connection_closed_callback, sink);
     sink->conn = NULL;
+    sink->closed = false;
+    sink->connecting = false;
     sink->connection_number = 0;
     nn_mutex_init(&sink->lock);
     nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink);
-    // This is also effectively a timeout for attempted reconnects.
-    nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000));
+    // Also effectively a timeout.
+    nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(2500000));
     nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
     nn_signal_start(&sink->queue_signal);
     camu_queue_init(sink->queue);
@@ -1709,7 +1891,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     al_array_init(sink->previous);
     al_array_init(sink->entries);
     nn_timer_init(&sink->empty_timer, sink->loop, empty_timer_callback, sink);
-    nn_timer_set_repeat(&sink->empty_timer, NNWT_TS_FROM_USEC(100000));
+    nn_timer_set_repeat(&sink->empty_timer, NNWT_TS_FROM_USEC(250000));
     nn_timer_again(&sink->empty_timer);
     sink->notify_status = 0;
     // Start high to exercise the wrapping path.
@@ -1720,6 +1902,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     sink->audio.mixer = mixer;
     sink->video.state = SINK_PAUSED;
     sink->video.renderer = renderer;
+    al_array_init(sink->node_prefs);
     return true;
 }
 
@@ -1729,18 +1912,16 @@ bool camu_sink_connect(struct camu_sink *sink, str *name, u8 type, str *addr, u1
     sink->type = type;
     al_str_clone(&sink->addr, addr);
     sink->port = port;
-    sink->conn = nn_rpc_prepare_client(&sink->client);
+    nn_rpc_prepare_client(&sink->client);
     al_assert(sink->callback);
     for (u32 i = 0; i < ARRAY_SIZE(commands); i++) {
         commands[i].userdata = sink;
         nn_rpc_add_command(&sink->client, &commands[i]);
     }
-#ifdef CAMU_DIRECT_MODE
-    // Note that direct_connect() runs connection_callback() directly.
-    nn_multiplex_direct_connect(sink->conn->stream, CAMU_MULTIPLEX_RPC);
-#else
-    nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port);
+#ifndef CAMU_DIRECT_MODE
+    nn_timer_again(&sink->reconnect_timer);
 #endif
+    rpc_connect(sink);
     return true;
 }
 
@@ -1765,40 +1946,46 @@ void camu_sink_add(struct camu_sink *sink, str *path)
 void camu_sink_skip(struct camu_sink *sink, s32 n)
 {
     nn_mutex_lock(&sink->lock);
-    struct camu_sink_entry *current = get_entry_for_command(sink);
+    struct camu_sink_entry *entry = get_entry_for_command(sink);
     nn_mutex_unlock(&sink->lock);
-    queue_cmd(sink, CMD(SKIP, .v.i = n, .opaque = current));
+    queue_cmd(sink, CMD(SKIP, .v.i = n, .opaque = entry));
 }
 
 void camu_sink_toggle_pause(struct camu_sink *sink)
 {
     nn_mutex_lock(&sink->lock);
-    struct camu_sink_entry *current = get_entry_for_command(sink);
+    struct camu_sink_entry *entry = get_entry_for_command(sink);
+    struct camu_sink_entry *current = (entry == sink->current) ? NULL : sink->current;
+    if (!entry) {
+        nn_mutex_unlock(&sink->lock);
+        return;
+    }
+    if (current) {
+        camu_clock_set_external_pause(&current->clock);
+    }
+    camu_clock_set_external_pause(&entry->clock);
+    f64 pts = camu_clock_get_last_pts(&entry->clock);
     nn_mutex_unlock(&sink->lock);
-    if (!current) return;
-    bool armed_for_pause = false;
-    camu_clock_external_pause(&current->clock);
-    f64 pts = camu_clock_get_pts(&current->clock, 0.0, false, &armed_for_pause);
-    queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = current));
+    queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = entry));
 }
 
 void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
 {
     nn_mutex_lock(&sink->lock);
-    struct camu_sink_entry *current = get_entry_for_command(sink);
+    struct camu_sink_entry *entry = get_entry_for_command(sink);
     u64 duration = 0;
     f64 pts = 0.0;
-    if (current) {
-        duration = current->client.duration;
-        pts = camu_clock_get_last_pts(&current->clock);
+    if (entry) {
+        duration = entry->client.duration;
+        pts = camu_clock_get_last_pts(&entry->clock);
     }
     nn_mutex_unlock(&sink->lock);
-    if (!current || duration == 0) {
+    if (!entry || duration == 0) {
         return;
     }
     struct camu_sink_cmd cmd = {
         .op = SEEK,
-        .opaque = current
+        .opaque = entry
     };
     switch (mode) {
     case CAMU_SEEK_POS: {
@@ -1813,7 +2000,6 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
         break;
     }
     case CAMU_SEEK_PERCENT: {
-        // There's probably a way to lose less precision here.
         f64 percent = *(f64 *)value;
         cmd.v.u = (u64)(duration * percent);
         break;
@@ -1827,6 +2013,24 @@ void camu_sink_reseek(struct camu_sink *sink)
     queue_cmd(sink, CMD(RESEEK));
 }
 
+void camu_sink_change_audio_track(struct camu_sink *sink, s32 n)
+{
+    nn_mutex_lock(&sink->lock);
+    struct camu_sink_entry *entry = get_entry_for_command(sink);
+    nn_mutex_unlock(&sink->lock);
+    if (!entry) return;
+    queue_cmd(sink, CMD(AUDIO_TRACK, .v.i = n, .opaque = entry));
+}
+
+void camu_sink_change_subtitle_track(struct camu_sink *sink, s32 n)
+{
+    nn_mutex_lock(&sink->lock);
+    struct camu_sink_entry *entry = get_entry_for_command(sink);
+    nn_mutex_unlock(&sink->lock);
+    if (!entry) return;
+    queue_cmd(sink, CMD(SUBTITLE_TRACK, .v.i = n, .opaque = entry));
+}
+
 void camu_sink_shuffle(struct camu_sink *sink)
 {
     queue_cmd(sink, CMD(SHUFFLE));
@@ -1836,7 +2040,21 @@ void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd)
 {
     nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    osd->connecting = !sink->connected;
+    osd->connecting = !sink->conn;
+    if (sink->notify_status & NOTIFY_DISCONNECTED) {
+        sink->notify_status &= ~NOTIFY_DISCONNECTED;
+        osd->show = true;
+        osd->shown_for |= CAMU_OSD_CONNECTING;
+    }
+    if (sink->notify_status & NOTIFY_CONNECTED) {
+        sink->notify_status &= ~NOTIFY_CONNECTED;
+        if (osd->shown_for & CAMU_OSD_CONNECTING) {
+            osd->shown_for &= ~CAMU_OSD_CONNECTING;
+            if (osd->shown_for == 0) {
+                osd->show = false;
+            }
+        }
+    }
     if (sink->notify_status & NOTIFY_EMPTY) {
         sink->notify_status &= ~NOTIFY_EMPTY;
         osd->show = true;
@@ -1846,29 +2064,45 @@ void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd)
         sink->notify_status &= ~NOTIFY_NOT_EMPTY;
         if (osd->shown_for & CAMU_OSD_EMPTY) {
             osd->shown_for &= ~CAMU_OSD_EMPTY;
-            osd->show = false;
+            if (osd->shown_for == 0) {
+                osd->show = false;
+            }
         }
     }
     if (sink->notify_status & NOTIFY_SEEK) {
         sink->notify_status &= ~NOTIFY_SEEK;
         if (!osd->show) {
             osd->flash = CAMU_OSD_FLASH_FOR(0.75);
+            osd->show = true;
         }
     }
-    nn_mutex_unlock(&sink->lock);
+    if (sink->notify_status & NOTIFY_ENTRY_ADDED) {
+        sink->notify_status &= ~NOTIFY_ENTRY_ADDED;
+        osd->no_force_render = false;
+    }
+    if (!osd->no_force_render) {
+        osd->no_force_render = sink->suspended;
+    }
     if (current) {
-        osd->paused = camu_clock_is_user_paused(&current->clock);
-        bool armed_for_pause = false;
-        osd->pts = camu_clock_get_pts(&current->clock, 0.0, false, &armed_for_pause);
+        u8 status = CAMU_CLOCK_NO_SIGNAL_PAUSE;
+        osd->pts = camu_clock_get_pts(&current->clock, 0.0, false, &status);
         if (CAMU_PTS_CONSIDER_PAUSED(osd->pts)) {
+            osd->paused = !(status & CAMU_CLOCK_PAUSE_FOR_SWAP) && osd->pts != CAMU_PTS_UNSET;
             osd->pts = camu_clock_get_last_pts(&current->clock);
+        } else {
+            osd->paused = !!(status & CAMU_CLOCK_EXTERNAL_PAUSE);
+        }
+        // Don't show paused on an image.
+        osd->paused &= VIDEO_EMPTY(current) || !(VIDEO_IS_STATIC(current) && AUDIO_EMPTY(current));
+        if (current->client.duration != LIANA_TIMESTAMP_INVALID) {
+            osd->duration = current->client.duration;
         }
-        osd->duration = current->client.duration;
     } else {
         osd->paused = false;
         osd->pts = 0.0;
         osd->duration = 0;
     }
+    nn_mutex_unlock(&sink->lock);
 }
 
 void camu_sink_stop(struct camu_sink *sink)
@@ -1883,25 +2117,30 @@ void camu_sink_stop(struct camu_sink *sink)
 void camu_sink_close(struct camu_sink *sink)
 {
     nn_timer_stop(&sink->reconnect_timer);
-    if (sink->conn) {
-        struct nn_rpc_connection *conn = sink->conn;
-        sink->conn = NULL; // Signal to connection_closed_callback() we're done.
-        nn_rpc_conn_disconnect(conn);
+    if (sink->conn || sink->connecting) {
+        sink->closed = true; // Signal to connection_closed_callback() we're done.
+        nn_rpc_disconnect(&sink->client);
     }
     nn_timer_stop(&sink->empty_timer);
+    array(struct camu_sink_entry *) cleanup;
+    al_array_init(cleanup);
+    al_array_copy(cleanup, sink->entries);
+    sink->entries.count = 0;
     struct camu_sink_entry *entry;
-    al_array_foreach_rev(sink->entries, i, entry) {
-        al_array_remove_at(sink->entries, i);
+    al_array_foreach(cleanup, i, entry) {
         maybe_disconnect_entry(entry);
     }
+    al_array_free(cleanup);
 }
 
 void camu_sink_free(struct camu_sink *sink)
 {
+    al_array_free(sink->node_prefs);
     al_assert(!sink->entries.count);
     al_array_free(sink->entries);
-    nn_rpc_free(&sink->client);
+    al_array_free(sink->previous);
     camu_queue_free(sink->queue);
+    nn_rpc_free(&sink->client);
     nn_mutex_destroy(&sink->lock);
     al_str_free(&sink->name);
     al_str_free(&sink->addr);
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 6000b4f..5cc5a2a 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -57,7 +57,7 @@ struct camu_sink_entry {
     struct {
         u8 state;
         // Don't stop audio on a BUFFER_PAUSED from this entry.
-        bool ignore_paused;
+        bool ignore_pause;
         struct camu_audio_buffer buf;
         struct lia_vcr_track *track;
     } audio;
@@ -81,11 +81,12 @@ struct camu_sink {
     u8 type;
     str addr;
     u16 port;
+    bool local;
     struct nn_rpc client;
     struct nn_rpc_connection *conn;
-    bool connected;
+    bool closed;
+    bool connecting;
     u16 connection_number;
-    bool local;
     struct nn_mutex lock;
     struct nn_timer reconnect_timer;
     struct nn_signal queue_signal;
@@ -108,6 +109,7 @@ struct camu_sink {
         struct camu_renderer *renderer;
     } video;
     struct lia_prefs prefs;
+    array(struct lia_prefs) node_prefs;
     str default_list;
     u8 (*callback)(void *, u8, u8, void *);
     void *userdata;
@@ -123,6 +125,8 @@ void camu_sink_skip(struct camu_sink *sink, s32 n);
 void camu_sink_toggle_pause(struct camu_sink *sink);
 void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode);
 void camu_sink_reseek(struct camu_sink *sink);
+void camu_sink_change_audio_track(struct camu_sink *sink, s32 n);
+void camu_sink_change_subtitle_track(struct camu_sink *sink, s32 n);
 void camu_sink_shuffle(struct camu_sink *sink);
 void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd);
 void camu_sink_stop(struct camu_sink *sink);