Wip sink changes around errored/ended buffers

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

2 files changed, 307 insertions, 202 deletions

diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index f410636..aa9542e 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -25,6 +25,8 @@ enum {
     BUFFER_INIT = 0,
     // Set but not configured.
     BUFFER_QUEUED,
+    // Errored buffer was removed, treat it as empty from now on.
+    BUFFER_DETACHED,
     // Ready to receive data.
     BUFFER_CONFIGURED,
     // The next call to add can add the buffer.
@@ -32,7 +34,9 @@ enum {
     // Treat the buffer like it's added, even if it might not be.
     BUFFER_ADDED,
     // Effectively SET_OR_BUFFERED but not addable until after a reset.
-    BUFFER_ENDED
+    BUFFER_ENDED,
+    // Same as BUFFER_ENDED but will be detached on a reset.
+    BUFFER_ERRORED
 };
 
 // Command queue commands.
@@ -54,18 +58,20 @@ enum {
     END
 };
 
-#define SINK_LRU_MAX UINT16_MAX
-
 // Number of entries to keep buffered at one time.
 #define ENTRY_MAX_AGE 4
-
-#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b)
+#define SINK_LRU_MAX UINT16_MAX
+AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX);
 
 // Store connection number in the upper 16 bits so IDs don't conflict after a server restart.
 // If the sink disconnects but the server didn't restart, this will invalidate IDs that _do_ map
-// to the same resource. So it's a trade off.
-#define LOCAL_ENTRY_ID(sink, id) (((u64)(sink)->connection_number) << 47 | (u64)(id))
+// to the same resource. That could be wasteful. It's also currently (ab)used by reseek().
+#define LOCAL_ENTRY_ID(sink, id) (((u64)(sink)->connection_number) << 48 | (u64)(id))
 #define REMOTE_ENTRY_ID(id) ((u32)(id & 0x7fffffff))
+#define CONNECTION_NUMBER(id) ((id >> 48) & 0xffff)
+
+// Only sink->target can ever be 0xb00b.
+#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b)
 
 // printf format for entries.
 #ifdef AL_DEBUG
@@ -80,21 +86,25 @@ enum {
 #define VIDEO_STATE(entry) ((entry)->video.state)
 
 // If a buffer is still INIT or QUEUED after the entry is configured, it's "empty".
-#define BUFFER_EMPTY(buf) ((buf)->state == BUFFER_INIT || (buf)->state == BUFFER_QUEUED)
-#define AUDIO_EMPTY(entry) BUFFER_EMPTY(&(entry)->audio)
-#define VIDEO_EMPTY(entry) BUFFER_EMPTY(&(entry)->video)
+// Also, if a buffer errors, it will be detached in CLIENT_REMOVE_BUFFERS.
+// Empty is a state that cannot change while a buffer is being used (push()/read()).
+#define AUDIO_EMPTY(entry) (AUDIO_STATE(entry) <= BUFFER_DETACHED)
+#define VIDEO_EMPTY(entry) (VIDEO_STATE(entry) <= BUFFER_DETACHED)
 
-#define AUDIO_NOT_ADDED(entry) ((entry)->audio.state != BUFFER_ADDED)
-#define VIDEO_NOT_ADDED(entry) ((entry)->video.state != BUFFER_ADDED)
+#define AUDIO_ENDED(entry) (AUDIO_STATE(entry) >= BUFFER_ENDED)
+#define VIDEO_ENDED(entry) (VIDEO_STATE(entry) >= BUFFER_ENDED)
+#define ENTRY_ENDED(entry) ((AUDIO_EMPTY(entry) || AUDIO_ENDED(entry)) && (VIDEO_EMPTY(entry) || VIDEO_ENDED(entry)))
 
-#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio))
-#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video))
+// IGNORED = ENDED or EMPTY.
+#define AUDIO_ADDED_OR_IGNORED(entry) (AUDIO_STATE(entry) >= BUFFER_ADDED || AUDIO_EMPTY(entry))
+#define VIDEO_ADDED_OR_IGNORED(entry) (VIDEO_STATE(entry) >= BUFFER_ADDED || VIDEO_EMPTY(entry))
 
-#define AUDIO_ENDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->audio))
-#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video))
+#define AUDIO_ENDED_OR_EMPTY(entry) (AUDIO_ENDED(entry) || AUDIO_EMPTY(entry))
+#define VIDEO_ENDED_OR_EMPTY(entry) (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry))
 
-#define VIDEO_STREAM(entry) (entry)->video.buf.stream
-#define VIDEO_IS_SINGLE_FRAME(entry) (entry)->video.buf.single_frame
+#define AUDIO_STREAM(entry) ((al_assert(!AUDIO_EMPTY(entry)), (entry)->audio.buf.stream))
+#define VIDEO_STREAM(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.stream))
+#define VIDEO_IS_SINGLE_FRAME(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.single_frame))
 
 #if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
 #define BLOCKING_SLEEP(delay) nn_thread_sleep(delay)
@@ -102,6 +112,17 @@ enum {
 #define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay)
 #endif
 
+// Functions that might happen on separate threads.
+// CAMU_MIXER_THREADED:
+//  audio_buffer_callback()
+//  mixer_callback()
+// CAMU_SCREEN_THREADED:
+//  video_buffer_callback()
+// CAMU_MIXER_THREADED or CAMU_SCREEN_THREADED:
+//  clock_callback()
+// client_callback()::LIANA_CLIENT_DATA
+// client_callback()::LIANA_CLIENT_EOF/ERRORED
+
 static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry)
 {
 #ifdef CAMU_MIXER_THREADED
@@ -163,7 +184,7 @@ static inline void add_entry_video_buffer(struct camu_sink_entry *entry)
 
 static void remove_entry_audio_buffer(struct camu_sink_entry *entry)
 {
-    al_assert(AUDIO_STATE(entry) != BUFFER_ENDED);
+    al_assert(!AUDIO_ENDED(entry));
     al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
     switch (AUDIO_STATE(entry)) {
     case BUFFER_ADDED:
@@ -191,7 +212,7 @@ static void remove_entry_audio_buffer(struct camu_sink_entry *entry)
 static void remove_entry_video_buffer(struct camu_sink_entry *entry)
 {
     // Don't assert !entry->ended here because of single frame handling.
-    al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
+    al_assert(!VIDEO_ENDED(entry));
     al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
     switch (VIDEO_STATE(entry)) {
     case BUFFER_ADDED:
@@ -214,17 +235,16 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry)
 // It's possible for some of an entry's buffers to be ENDED while others are still ADDED.
 // This means entry->ended and BUFFER_ENDED have two distinct considerations.
 // entry->ended: Completely ignored and needs special handling in CLIENT_REMOVE_BUFFERS.
-// BUFFER_ENDED: No-op'd in remove_entry_buffers() and add_or_queue_entry() but otherwise unchanged.
+// BUFFER_ENDED: No-op'd in remove_entry_buffers(), add_or_queue_entry() and not added by do_add_entry(), otherwise unchanged.
+//  Another note: remove_entry_buffers() and add_or_queue_entry() are only ever called by switch_to().
+//   switch_to() -> add_or_queue_entry().
+//   switch_to() -> maybe_add_to_previous() -> (possibly delayed)remove_entry_buffers().
 static void remove_entry_buffers(struct camu_sink_entry *entry)
 {
     log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry));
     al_assert(!entry->ended);
-    if (AUDIO_STATE(entry) != BUFFER_ENDED) {
-        remove_entry_audio_buffer(entry);
-    }
-    if (VIDEO_STATE(entry) != BUFFER_ENDED) {
-        remove_entry_video_buffer(entry);
-    }
+    if (!AUDIO_ENDED(entry)) remove_entry_audio_buffer(entry);
+    if (!VIDEO_ENDED(entry)) remove_entry_video_buffer(entry);
 }
 
 static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry);
@@ -237,12 +257,12 @@ static void add_or_queue_entry(struct camu_sink_entry *entry)
     al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
     if (AUDIO_STATE(entry) == BUFFER_INIT) {
         AUDIO_STATE(entry) = BUFFER_QUEUED;
-    } else if (AUDIO_STATE(entry) != BUFFER_ENDED) {
+    } else if (!AUDIO_ENDED(entry)) {
         add_audio_if_set_and_buffered(entry);
     }
     if (VIDEO_STATE(entry) == BUFFER_INIT) {
         VIDEO_STATE(entry) = BUFFER_QUEUED;
-    } else if (VIDEO_STATE(entry) != BUFFER_ENDED) {
+    } else if (!VIDEO_ENDED(entry)) {
         add_video_if_set_and_buffered(entry);
     }
 }
@@ -258,12 +278,12 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry)
 }
 
 #ifdef CAMU_SINK_LOCAL
-static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *entry)
+static void local_pause(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
-    if (entry->held) return;
     if (!camu_clock_is_paused(&entry->clock)) {
         entry->paused = true;
         camu_clock_pause(&entry->clock, 0);
+        log_info("Clock paused.");
         // Audio stop will be handled by a BUFFER_PAUSED callback.
         if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) {
 #ifndef CAMU_SINK_NO_VIDEO
@@ -274,16 +294,17 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
     } else {
         entry->paused = false;
         camu_clock_resume(&entry->clock, 0);
-        if (!AUDIO_EMPTY(entry) && sink->audio.state == SINK_PAUSED) {
-            sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
-            sink->audio.state = SINK_PLAYING;
-        }
+        log_info("Clock resumed.");
         if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) {
 #ifndef CAMU_SINK_NO_VIDEO
             sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL);
 #endif
             sink->video.state = SINK_PLAYING;
         }
+        if (!AUDIO_EMPTY(entry) && sink->audio.state == SINK_PAUSED) {
+            sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
+            sink->audio.state = SINK_PLAYING;
+        }
     }
 }
 #endif
@@ -292,8 +313,10 @@ static inline s32 get_sequence_for_command(struct camu_sink_entry *entry)
 {
     // SEQUENCE_ANY resolves order on the server.
     s32 sequence = LIANA_SEQUENCE_ANY;
-    // If we're local this could only lead to feeling like your inputs were eaten.
+    // If local, this could only lead to feeling like your inputs were eaten.
 #ifndef CAMU_SINK_LOCAL
+    // Entry is sent as an argument via get_entry_for_command(), so it can only
+    // be a valid entry or NULL and not a dangling target.
     if (entry) sequence = entry->sequence;
 #else
     (void)entry;
@@ -395,23 +418,25 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
     }
     case SKIP: {
         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);
         nn_packet_write_u8(packet, CAMU_LIST_SKIP);
-        nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque));
+        nn_packet_write_s32(packet, get_sequence_for_command(entry));
         nn_packet_write_s32(packet, (s32)cmd->value.i);
         nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
         break;
     }
     case TOGGLE_PAUSE: {
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
 #ifdef CAMU_SINK_LOCAL
-        sink_local_pause(sink, (struct camu_sink_entry *)cmd->opaque);
+        if (!entry->held) local_pause(sink, entry);
 #else
         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_TOGGLE_PAUSE);
-        nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque));
+        nn_packet_write_s32(packet, get_sequence_for_command(entry));
         nn_packet_write_f64(packet, cmd->value.f);
         nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
 #endif
@@ -419,10 +444,10 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
     }
     case SEEK: {
         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);
         nn_packet_write_u8(packet, CAMU_LIST_SEEK);
-        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         nn_packet_write_s32(packet, entry->sequence);
         nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id));
         nn_packet_write_u64(packet, cmd->value.u);
@@ -430,8 +455,9 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         break;
     }
     case RESEEK: {
-        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
-        lia_client_reseek(&entry->client);
+        if (!sink->conn) return;
+        // Crude way to trigger "re-add sink to list".
+        nn_rpc_conn_disconnect(sink->conn);
         break;
     }
     case SHUFFLE: {
@@ -444,10 +470,10 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
     }
     case END: {
         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);
         nn_packet_write_u8(packet, CAMU_LIST_END);
-        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
         nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id));
         nn_packet_write_u32(packet, (u32)cmd->value.u);
         nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
@@ -471,10 +497,11 @@ static void queue_signal_callback(void *userdata)
 static void mixer_callback(void *userdata, u8 op)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
+    // @TODO: Isn't the idea of MIXER_EMPTY to not STOP audio when we know the mixer will be empty?
     if (op == CAMU_MIXER_EMPTY) {
         log_info("Mixer empty.");
-        // We need to sync with do_add_entry() because the order of
-        // the START/STOP's in the queue matters.
+        // Regardless of if we are checking an entry's state here, we have to sync with
+        // do_add_entry() because the order of START/STOPs in the queue matters.
         nn_mutex_lock(&sink->lock);
         // This feels a bit too loose.
         if (!(sink->current && AUDIO_STATE(sink->current) == BUFFER_ADDED)) {
@@ -503,7 +530,6 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
     // indicative of the amount of entries we have loaded.
     // The most obvious reason being it's incremented when moving back
     // and forth between two entries. As well as for buffer and queue operations.
-    u16 max_age = ENTRY_MAX_AGE;
     // We have to handle sink->lru wrapping in a step before the default case.
     // 0 65532 65533 65534 65535
     // 0 1 65533 65534 65535
@@ -512,21 +538,22 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
     // 0 1 2 3 4
     struct camu_sink_entry *entry;
     al_array_foreach_rev(sink->entries, i, entry) {
-        if (entry->lru > sink->lru && ((SINK_LRU_MAX - entry->lru) + 1) + sink->lru >= max_age) {
+        if (sink->entries.count <= ENTRY_MAX_AGE) return;
+        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);
         }
-        if (sink->entries.count <= max_age) return;
     }
-    // Make sure we don't have to consider wrapping here.
-    if (sink->lru >= max_age) {
-        al_array_foreach_rev(sink->entries, i, entry) {
-            al_assert(sink->lru >= entry->lru);
-            if (sink->lru - entry->lru >= max_age) {
-                al_array_remove_at(sink->entries, i);
-                maybe_disconnect_entry(entry);
-            }
-            if (sink->entries.count <= max_age) return;
+    // 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);
         }
     }
 }
@@ -537,12 +564,21 @@ static void maybe_remove_previous(struct camu_sink *sink)
     struct camu_sink_entry *previous;
     al_array_foreach(sink->previous, i, previous) {
         remove_entry_buffers(previous);
+        // Cleanup entries from old connections. This is especially important to
+        // keep reseek() from being overly wasteful.
+        if (CONNECTION_NUMBER(previous->id) != sink->connection_number) {
+            queue_cmd(sink, (struct camu_sink_cmd){
+                .op = EJECT_ENTRY,
+                .opaque = previous
+            });
+        }
     }
     sink->previous.count = 0;
 }
 
 // Due to the looseness of the previous queue, we may have to explicitly remove an
-// entry if it becomes incorrect to attempt removing it's buffers.
+// entry if it becomes incorrect to attempt removing it's buffers. The most obvious example
+// being at the point it's freed.
 static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key)
 {
     bool removed = false;
@@ -573,45 +609,49 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
     al_array_push(sink->previous, previous);
 }
 
-// Call this after setting the buffer's state to ADDED because this entry might be in previous.
-static void do_add_entry(struct camu_sink_entry *entry)
+static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool skip_video)
 {
-    // Single frames are unconditionally added in add_video_if_set_and_buffered().
-    if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
-        add_entry_video_buffer(entry);
-    }
-    if (!AUDIO_EMPTY(entry)) {
-        add_entry_audio_buffer(entry);
-    }
     maybe_remove_previous(entry->sink);
     queue_cmd(entry->sink, (struct camu_sink_cmd){
-        .op = (VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry) || entry->paused) ? STOP : START,
+        .op = (skip_video || entry->paused) ? STOP : START,
         .value.i = CAMU_SINK_VIDEO
     });
-    if (VIDEO_EMPTY(entry)) {
+    if (VIDEO_ENDED_OR_EMPTY(entry)) {
         // Clear the screen if skipping from a video to an audio-only entry.
         refresh_video_output(entry->sink);
     }
     queue_cmd(entry->sink, (struct camu_sink_cmd){
-        .op = (AUDIO_EMPTY(entry) || entry->paused) ? STOP : START,
+        .op = (skip_audio || entry->paused) ? STOP : START,
         .value.i = CAMU_SINK_AUDIO
     });
 }
 
+// Call this after setting state to ADDED because this entry might be in previous.
+static void do_add_entry(struct camu_sink_entry *entry)
+{
+    bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry);
+    // Single frames are unconditionally added in add_video_if_set_and_buffered().
+    bool skip_video = VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
+    // Whether to add audio or video first could be a consideration for responsiveness.
+    if (!skip_video) add_entry_video_buffer(entry);
+    if (!skip_audio) add_entry_audio_buffer(entry);
+    after_add_entry(entry, skip_audio, skip_video);
+}
+
 void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 {
     al_assert(!entry->ended);
     al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
     al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED);
     al_assert(AUDIO_STATE(entry) != BUFFER_ADDED);
-    al_assert(AUDIO_STATE(entry) != BUFFER_ENDED);
+    al_assert(!AUDIO_ENDED(entry));
     switch (AUDIO_STATE(entry)) {
     case BUFFER_CONFIGURED:
         AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
         break;
     case BUFFER_SET_OR_BUFFERED:
         AUDIO_STATE(entry) = BUFFER_ADDED;
-        if (VIDEO_ADDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) {
+        if (VIDEO_ADDED_OR_IGNORED(entry) || VIDEO_IS_SINGLE_FRAME(entry)) {
             do_add_entry(entry);
         }
         break;
@@ -625,7 +665,7 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
     al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
     al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
     al_assert(VIDEO_STATE(entry) != BUFFER_ADDED);
-    al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
+    al_assert(!VIDEO_ENDED(entry));
     switch (VIDEO_STATE(entry)) {
     case BUFFER_CONFIGURED:
         VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
@@ -634,8 +674,11 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
         VIDEO_STATE(entry) = BUFFER_ADDED;
         if (VIDEO_IS_SINGLE_FRAME(entry)) {
             add_entry_video_buffer(entry);
-            if (AUDIO_EMPTY(entry)) maybe_remove_previous(entry->sink);
-        } else if (AUDIO_ADDED_OR_EMPTY(entry)) {
+            bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry);
+            if (skip_audio) {
+                after_add_entry(entry, skip_audio, true);
+            }
+        } else if (AUDIO_ADDED_OR_IGNORED(entry)) {
             do_add_entry(entry);
         }
         break;
@@ -644,26 +687,31 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
 
 static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
 {
-    log_trace("switch_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(sink->current));
+    struct camu_sink_entry *current = sink->current;
+
+    log_trace("switch_to("ENTRY_FMT"(ended: %s)), current: "ENTRY_FMT"(ended: %s).",
+        ENTRY_ARG(target), BOOLSTR(ENTRY_IS_VALID(target) ? target->ended : false),
+        ENTRY_ARG(current), BOOLSTR(current ? current->ended : false));
 
     bool ensure_removed = false;
-    struct camu_sink_entry *current = sink->current;
     if (current) {
         struct camu_sink_entry *detached = sink->detached;
         al_assert(current != target);
         ensure_removed = detached || current->ended;
         if (detached) {
             al_assert(detached == current);
+            // A buffer's state being QUEUED should be impossible while it's
+            // entry is reconnecting.
+            al_assert(AUDIO_STATE(detached) != BUFFER_QUEUED);
+            al_assert(VIDEO_STATE(detached) != BUFFER_QUEUED);
             sink->detached = NULL;
-            log_debug("Unset detached as a substitute for remove.");
-            // This should only matter if detached was unconfigured.
-            if (AUDIO_EMPTY(detached)) AUDIO_STATE(detached) = BUFFER_INIT;
-            if (VIDEO_EMPTY(detached)) VIDEO_STATE(detached) = BUFFER_INIT;
+            log_warn("Unset detached as a substitute for remove.");
         } else if (!current->ended) {
             maybe_add_to_previous(sink, current, target);
         }
     }
 
+    // @TODO: Cleanup stop_video? and log_trace lengths.
     bool stop_video = false;
     bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
     if (!dangling_target) {
@@ -671,19 +719,22 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
             remove_previous_if_contains(sink, target);
             add_or_queue_entry(target);
         } else {
-            if (!VIDEO_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) {
+            log_trace("Target ended in switch_to().");
+            if (!VIDEO_ENDED_OR_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) {
                 add_video_if_set_and_buffered(target);
             }
             stop_video = true;
         }
+    } else {
+        log_trace("Ignored dangling target.");
     }
 
     if (ensure_removed) {
-        if (!VIDEO_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) {
+        if (!VIDEO_ENDED_OR_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) {
             remove_entry_video_buffer(current);
         }
-        al_assert(AUDIO_NOT_ADDED(current));
-        al_assert(VIDEO_NOT_ADDED(current));
+        al_assert(AUDIO_STATE(current) != BUFFER_ADDED);
+        al_assert(VIDEO_STATE(current) != BUFFER_ADDED);
     }
 
     if (!dangling_target) {
@@ -706,6 +757,20 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
     }
 }
 
+// @TODO: Could entry->ended be redundant? What about entry->held?
+//  - The catalyst for this is having to check entry->ended in CLIENT_REMOVE_BUFFERS breaking the ability
+//    to maintain the same state if called consecutively.
+//  - As long as end_entry_and_advance_queue() evaluates target, what is the point of
+//    ended except to say all of an entry's buffers are ended.
+//     (except to guess what the server thinks, which was it's original purpose).
+//  - Knowing if the outputs are paused is clearly necessary but what about it's relationship to held?
+//   - entry->paused basically means preempt any queued skip action.
+//   - On video or audio paused, check if SINK_EMPTY or PAUSED and run target (currently not handled at all).
+//   - Fix SINK_LOCAL relying on entry->paused hint from server (rely on clock state?).
+//    entry->paused reduced to nothing but a hint about if the entry is paused when skipping.
+//    Only consideration is for the sink outputs to never stay paused when playing and vise versa
+//    - Then entry->paused actually doesn't matter (sink state is above entry anyway and we should be able to
+//      rely on the clock per-entry).
 #ifndef CAMU_SINK_LOCAL
 static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at)
 {
@@ -787,27 +852,26 @@ static void audio_buffer_callback(void *userdata, u8 op)
         nn_mutex_unlock(&sink->lock);
         break;
     case CAMU_BUFFER_EOF:
-        log_debug("Audio EOF.");
+    case CAMU_BUFFER_ERRORED: {
+        bool error = op == CAMU_BUFFER_ERRORED;
+        log_debug(error ? "Audio buffer errored." : "Audio EOF.");
         nn_mutex_lock(&sink->lock);
-        // Having threaded outputs means anything could have happened while waiting
-        // on the lock above. If we were locked in CLIENT_REMOVE_BUFFERS, state could very
-        // well be CONFIGURED here.
+        // EOF and ERRORED come from the outputs read() thread. So, having threaded
+        // outputs means anything could have happened while waiting on the lock above.
+        // For example, if we were locked in CLIENT_REMOVE_BUFFERS, state could have
+        // dropped all the way to CONFIGURED before we acquired the lock here.
+        // This should also maintain a consistent state in the more common case of switch_to()
+        // right before a buffer EOF.
         if (AUDIO_STATE(entry) == BUFFER_ADDED) {
             remove_entry_audio_buffer(entry);
         }
-        AUDIO_STATE(entry) = BUFFER_ENDED;
+        AUDIO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED;
         if (VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) {
             end_entry_and_advance_queue(sink, entry);
         }
         nn_mutex_unlock(&sink->lock);
         break;
-    case CAMU_BUFFER_ERRORED:
-        log_error("Audio buffer errored.");
-        queue_cmd(sink, (struct camu_sink_cmd){
-            .op = EJECT_ENTRY,
-            .opaque = entry
-        });
-        break;
+    }
     }
 }
 
@@ -829,20 +893,23 @@ static void video_buffer_callback(void *userdata, u8 op)
         lia_vcr_uncork(entry->video.track);
         break;
     case CAMU_BUFFER_EOF:
-        log_debug("Video EOF.");
+    case CAMU_BUFFER_ERRORED: {
+        bool error = op == CAMU_BUFFER_ERRORED;
+        log_debug(error ? "Video buffer errored." : "Video EOF.");
         nn_mutex_lock(&sink->lock);
-        // This buffer's state could be ADDED, SET_OR_BUFFERED, or CONFIGURED.
         if (!AUDIO_EMPTY(entry)) {
             camu_audio_buffer_set_no_video(&entry->audio.buf, true);
         }
-        if (VIDEO_IS_SINGLE_FRAME(entry)) {
+        if (VIDEO_IS_SINGLE_FRAME(entry) && !error) {
             nn_mutex_unlock(&sink->lock);
             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);
         }
-        VIDEO_STATE(entry) = BUFFER_ENDED;
+        VIDEO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED;
         bool swapped = false;
         if (AUDIO_ENDED_OR_EMPTY(entry)) {
             swapped = end_entry_and_advance_queue(sink, entry);
@@ -855,13 +922,7 @@ static void video_buffer_callback(void *userdata, u8 op)
             });
         }
         break;
-    case CAMU_BUFFER_ERRORED:
-        log_error("Video buffer errored.");
-        queue_cmd(sink, (struct camu_sink_cmd){
-            .op = EJECT_ENTRY,
-            .opaque = entry
-        });
-        break;
+    }
     }
 }
 
@@ -889,6 +950,7 @@ static void clock_callback(void *userdata, u8 op)
 
 static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
+    // @TODO: Video can't be ENDED here right?
     bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
     f64 avg_frame_duration = entry->video.buf.avg_frame_duration;
 #ifdef CAMU_SINK_LOCAL
@@ -896,8 +958,8 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s
         f64 audio = camu_mixer_get_latency(sink->audio.mixer);
         struct camu_renderer *renderer = sink->video.renderer;
         f64 video = renderer->get_latency(renderer) * avg_frame_duration;
-        // Start either the audio or video early so we can start the clock
-        // as soon as possible while keeping A/V sync.
+        // Start either the audio or video early so we can start the clock as
+        // soon as possible while keeping A/V sync.
         if (audio > video) {
             camu_video_buffer_set_latency(&entry->video.buf, video - audio);
         } else if (video > audio) {
@@ -1008,15 +1070,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     }
     case LIANA_CLIENT_DATA: {
         struct camu_codec_frame *frame = (struct camu_codec_frame *)opaque;
+        // Even though !AUDIO/VIDEO_EMPTY() is a value that cannot change at this point, we
+        // still have to lock because AUDIO/VIDEO_STATE() is not atomic.
         switch (stream->type) {
         case CAMU_STREAM_AUDIO:
-            al_assert(!AUDIO_EMPTY(entry));
+            nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock);
             camu_audio_buffer_push(&entry->audio.buf, frame);
-            return;
+            break;
         case CAMU_STREAM_VIDEO:
-            al_assert(!VIDEO_EMPTY(entry));
+            nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock);
             camu_video_buffer_push(&entry->video.buf, frame);
-            return;
+            break;
         default:
             camu_codec_frame_discard(frame);
             break;
@@ -1037,22 +1101,38 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         nn_mutex_lock(&sink->lock);
         log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current));
 
-        // Immediately switch to a potential target to avoid excessive delay/catchup
-        // that would be caused by this entry being re-added with an in-between clock state.
-        if (sink->target) {
-            switch_to(sink, sink->target);
-            sink->target = NULL;
+        if (entry == sink->current) {
+            if (!entry->ended) {
+                // It should only be possible for a buffer to be INIT if entry is not current or ended.
+                // Ended case is: (audio or video buffer empty) -> entry switched off of -> entry ended -> switch_to()'d.
+                //  - The empty buffer doesn't get re-QUEUED because the entry is ended.
+                al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
+                al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
+            }
+            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().
+                switch_to(sink, sink->target);
+                sink->target = NULL;
+            }
+        }
+
+        // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers.
+        if (entry == sink->current && rec->reconnect) {
+            sink->detached = entry;
         }
 
         // 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.
         remove_previous_if_contains(sink, entry);
+        // AUDIO/VIDEO_STATE() could be INIT after this point, even if entry = current.
 
-        // If this entry is still current on CLIENT_RECONNECTED, re-add buffers.
-        if (rec->reconnect && entry == sink->current) {
-            sink->detached = entry;
-        }
+        // We should treat CLIENT_REMOVE_BUFFERS as a function that removes an entry's buffers and
+        // resets it's buffered state. For a non-empty entry that means remove_entry_audio/video_buffer()
+        // twice and for an empty buffer, knock it down to BUFFER_INIT.
+        if (AUDIO_STATE(entry) == BUFFER_QUEUED) AUDIO_STATE(entry) = BUFFER_INIT;
+        if (VIDEO_STATE(entry) == BUFFER_QUEUED) VIDEO_STATE(entry) = BUFFER_INIT;
 
         // Ignore unconfigured entries.
         if (rec->unconfigured) {
@@ -1062,79 +1142,95 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             return;
         }
 
-        // An empty buffer is guaranteed to not be held.
-        if (!AUDIO_ENDED_OR_EMPTY(entry)) {
+        // An empty buffer is guaranteed to not be held. An ended buffer is already
+        // removed and will be further handled at the end of this case.
+        bool skip_audio = AUDIO_EMPTY(entry);
+        if (!skip_audio && !AUDIO_ENDED(entry)) {
             // Remove for re-add in CLIENT_RECONNECTED.
             remove_entry_audio_buffer(entry);
             // Remove again for another add in BUFFER_BUFFERED.
             remove_entry_audio_buffer(entry);
         }
-        bool skip_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry);
-        if (!VIDEO_ENDED_OR_EMPTY(entry)) { // Single frames will never be ENDED.
-            if (!skip_video) {
-                remove_entry_video_buffer(entry);
-                remove_entry_video_buffer(entry);
-            } else {
-                // We are keeping the frame, so don't request a duplicate.
-                entry->client.mask &= ~(1 << VIDEO_STREAM(entry)->index);
-            }
+        bool skip_video = VIDEO_EMPTY(entry);
+        // Don't remove a single_frame if we are reconnecting, unless it's errored.
+        skip_video = skip_video || (rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry) && VIDEO_STATE(entry) != BUFFER_ERRORED);
+        if (!skip_video && !VIDEO_ENDED(entry)) {
+            remove_entry_video_buffer(entry);
+            remove_entry_video_buffer(entry);
         }
-        // Resolve any queued REMOVE_BUFFER requests before blocking.
-        // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER happens from the sink's
-        // command queue. So, this is necessary for safely blocking the loop here.
+        // 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.
         run_queue_by_opaque(sink, entry);
 
         // Unlock to wait.
         nn_mutex_unlock(&sink->lock);
 
-        while (entry_audio_buffer_held(entry) || (!skip_video && entry_video_buffer_held(entry))) {
+        while ((!skip_audio && entry_audio_buffer_held(entry)) || (!skip_video && entry_video_buffer_held(entry))) {
             BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000));
         }
+        // At this point we can be sure that the entry's buffers are no longer in use.
 
-        // Reset possible ENDED state here in case the entry ended at some point after unlocking above.
+        // Re-lock to check an entries ended state. As it could have been set at some point after unlocking to wait.
         nn_mutex_lock(&sink->lock);
 
+        if (rec->reconnect) {
+            // Detach errored buffers. If we detach both streams, the entry will be closed.
+            if (AUDIO_STATE(entry) == BUFFER_ERRORED) {
+                al_array_push(rec->detached, AUDIO_STREAM(entry));
+            }
+            if (VIDEO_STATE(entry) == BUFFER_ERRORED) {
+                al_array_push(rec->detached, VIDEO_STREAM(entry));
+            } else if (!VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) {
+                // Don't request a duplicate single_frame.
+                rec->mask &= ~(1 << VIDEO_STREAM(entry)->index);
+            }
+        }
+
+        // Finalize the removal of the buffers by handling ended (ENDED or ERRORED) buffers.
+        // ENDED: Set to CONFIGURED to emulate the two removes earlier in this case.
+        // ERRORED: Set to DETACHED and they will now be considered empty.
         if (AUDIO_STATE(entry) == BUFFER_ENDED) {
             al_assert(!AUDIO_EMPTY(entry));
             AUDIO_STATE(entry) = BUFFER_CONFIGURED;
-        } else if (entry->ended && AUDIO_EMPTY(entry)) {
-            // If entry is ended, it may have forewent a remove_entry_buffers().
-            AUDIO_STATE(entry) = BUFFER_INIT;
+        } else if (AUDIO_STATE(entry) == BUFFER_ERRORED) {
+            AUDIO_STATE(entry) = BUFFER_DETACHED;
         }
         if (VIDEO_STATE(entry) == BUFFER_ENDED) {
             al_assert(!VIDEO_EMPTY(entry));
             VIDEO_STATE(entry) = BUFFER_CONFIGURED;
-        } else if (entry->ended && VIDEO_EMPTY(entry)) {
-            VIDEO_STATE(entry) = BUFFER_INIT;
+        } else if (VIDEO_STATE(entry) == BUFFER_ERRORED) {
+            VIDEO_STATE(entry) = BUFFER_DETACHED;
         }
+        // Unset ended, consistent with the logic in list.
         entry->ended = false;
 
         nn_mutex_unlock(&sink->lock);
 
         break;
     }
-    case LIANA_CLIENT_RESUME_AT: {
+    case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED.
         struct lia_timing *time = (struct lia_timing *)opaque;
+#ifdef CAMU_SINK_LOCAL
+        // List has no concept of a local sink, ignore it's request.
+        time->at = 0;
+#endif
         log_trace("resume_at("ENTRY_FMT"), seek_pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->seek_pos / 1000000.0, entry->clock.paused_at);
-        // These buffers won't be re-added until after a CLIENT_RECONNECTED event.
+        nn_mutex_lock(&sink->lock);
         bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
         if (!AUDIO_EMPTY(entry)) {
             camu_audio_buffer_reset(&entry->audio.buf);
-            // no_video is set 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) {
             camu_video_buffer_reset(&entry->video.buf, time->seek_pos);
         }
-        nn_mutex_lock(&sink->lock);
 #ifdef LIANA_LIST_SCUFFED_LOOP
         if (time->seek_pos == 0) {
             camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock));
         } else {
 #endif
-#ifdef CAMU_SINK_LOCAL
-            time->at = 0;
-#endif
             camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
 #ifdef LIANA_LIST_SCUFFED_LOOP
         }
@@ -1148,40 +1244,44 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         log_trace("reconnected("ENTRY_FMT"), detached: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), ENTRY_ARG(sink->detached), AUDIO_STATE(entry), VIDEO_STATE(entry));
         if (entry == sink->detached) {
             al_assert(entry == sink->current);
+            // Let this be the only other explicit BUFFER_ENDED check, or this will
+            // become too complicated.
             al_assert(AUDIO_STATE(entry) != BUFFER_ENDED);
             al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
+            // The value of unconfigured remains consistent from CLIENT_REMOVE_BUFFERS.
             if (rec->unconfigured) {
-                // The value of unconfigured is consistent from CLIENT_REMOVE_BUFFERS.
                 al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry));
-            } else {
-                if (AUDIO_EMPTY(entry)) {
-                    AUDIO_STATE(entry) = BUFFER_QUEUED;
-                } else {
-                    add_audio_if_set_and_buffered(entry);
-                }
-                if (VIDEO_EMPTY(entry)) {
-                    VIDEO_STATE(entry) = BUFFER_QUEUED;
-                } else if (!VIDEO_IS_SINGLE_FRAME(entry)) {
-                    add_video_if_set_and_buffered(entry);
-                }
+            }
+            // An unconfigured entry would have still been queued if it was current.
+            if (AUDIO_STATE(entry) == BUFFER_INIT) {
+                AUDIO_STATE(entry) = BUFFER_QUEUED;
+            } else if (!AUDIO_EMPTY(entry)) {
+                add_audio_if_set_and_buffered(entry);
+            }
+            if (VIDEO_STATE(entry) == BUFFER_INIT) {
+                VIDEO_STATE(entry) = BUFFER_QUEUED;
+            } else if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
+                add_video_if_set_and_buffered(entry);
             }
             sink->detached = NULL;
         }
         nn_mutex_unlock(&sink->lock);
         break;
     }
-    case LIANA_CLIENT_EOF: {
+    case LIANA_CLIENT_EOF:
+    case LIANA_CLIENT_ERRORED: {
+        bool error = op == LIANA_CLIENT_ERRORED;
         switch (stream->type) {
         case CAMU_STREAM_AUDIO: {
-            if (!AUDIO_EMPTY(entry)) {
-                camu_audio_buffer_flush(&entry->audio.buf);
-            }
+            nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock);
+            camu_audio_buffer_flush(&entry->audio.buf, error);
             break;
         }
         case CAMU_STREAM_VIDEO: {
+            nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock);
             // Single frames are immediately flushed inside the buffer.
-            if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
-                camu_video_buffer_flush(&entry->video.buf);
+            if (!VIDEO_IS_SINGLE_FRAME(entry) || error) {
+                camu_video_buffer_flush(&entry->video.buf, error);
             }
             break;
         }
@@ -1189,14 +1289,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         break;
     }
     case LIANA_CLIENT_CLOSED: {
-        // CLIENT_REMOVE_BUFFERS has been called on this entry before we're here.
+        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
         nn_mutex_lock(&sink->lock);
+        // We can be assured that CLIENT_REMOVE_BUFFERS has been called on this entry.
 
-        if (VIDEO_STATE(entry) == BUFFER_ADDED) {
-            al_assert(VIDEO_IS_SINGLE_FRAME(entry));
+        // Only possible case that a buffer could still be added.
+        if (rec->reconnect && !VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) {
             remove_entry_video_buffer(entry);
             while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); }
         }
+        al_assert(AUDIO_STATE(entry) != BUFFER_ADDED);
+        al_assert(VIDEO_STATE(entry) != BUFFER_ADDED);
 
         bool removed = al_array_remove(sink->entries, entry);
         remove_from_queue_by_opaque(sink, entry);
@@ -1232,7 +1335,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         lia_client_free(&entry->client);
         camu_audio_buffer_free(&entry->audio.buf);
         camu_video_buffer_free(&entry->video.buf);
-        log_info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup");
+        log_warn("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup");
         al_free(entry);
 
         break;
@@ -1266,8 +1369,12 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id)
     camu_video_buffer_init(&entry->video.buf, &entry->clock);
     entry->video.buf.callback = video_buffer_callback;
     entry->video.buf.userdata = entry;
+    // Two entries in order could share the same pointer if the first
+    // entry was just freed. This has to be accounted for in the renderer
+    // cache or it won't update on the first frame of the new entry.
+    // This is most likely to happen when the sink reconnects to a server.
     union { f64 f; u64 u; } fv = { .u = id };
-    entry->video.buf.reset_pts = fv.f;
+    entry->video.buf.seek_pts = fv.f;
 
     al_array_push(sink->entries, entry);
 
@@ -1293,10 +1400,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     if (op == LIANA_SINK_UNSET) {
         nn_mutex_lock(&sink->lock);
         if (sink->current) {
-            queue_cmd(sink, (struct camu_sink_cmd){
-                .op = EJECT_ENTRY,
-                .opaque = sink->current
-            });
+            maybe_disconnect_entry(sink->current);
         }
         goto out;
     }
@@ -1321,8 +1425,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     bool create = !entry;
     if (create) entry = create_entry(sink, id);
     entry->sequence = sequence;
-    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
     entry->lru = sink->lru;
+    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
     entry->reset_id = reset_id;
     if (create) {
         entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
@@ -1330,8 +1434,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
         lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos);
     }
 
-    // Don't lock before calling client_connect() or we could deadlock
-    // in CLIENT_CLOSED on a failed socket_connect().
+    // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect().
     nn_mutex_lock(&sink->lock);
 
     if (op == LIANA_SINK_BUFFER) {
@@ -1345,7 +1448,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
 #ifdef CAMU_SINK_LOCAL
     (void)at;
     al_assert(entry != current);
-    log_trace("set("ENTRY_FMT"), %s[local], created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create));
+    log_trace("set("ENTRY_FMT"), %s(local), created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create));
     if (current) {
         current->audio.ignore_paused = true;
         if (!current->paused) {
@@ -1458,29 +1561,31 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     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));
 #ifdef CAMU_SINK_LOCAL
     (void)at;
-    sink_local_pause(sink, entry);
+    if (!entry->held) local_pause(sink, entry);
 #else
     switch (pause) {
     case LIANA_PAUSE_PAUSE:
         entry->paused = true;
         camu_clock_pause(&entry->clock, at);
+        log_info("Clock paused.");
         // Audio will be stopped in a BUFFER_PAUSED callback.
         // Video will be stopped in a CLOCK_PAUSED callback.
         break;
     case LIANA_PAUSE_RESUME:
         entry->paused = false;
         camu_clock_resume(&entry->clock, at);
-        if (!AUDIO_EMPTY(entry)) {
-            camu_audio_buffer_resync(&entry->audio.buf);
+        log_info("Clock resumed.");
+        if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
             queue_cmd(entry->sink, (struct camu_sink_cmd){
                 .op = START,
-                .value.i = CAMU_SINK_AUDIO
+                .value.i = CAMU_SINK_VIDEO
             });
         }
-        if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
+        if (!AUDIO_EMPTY(entry)) {
+            camu_audio_buffer_resync(&entry->audio.buf);
             queue_cmd(entry->sink, (struct camu_sink_cmd){
                 .op = START,
-                .value.i = CAMU_SINK_VIDEO
+                .value.i = CAMU_SINK_AUDIO
             });
         }
         break;
@@ -1570,18 +1675,20 @@ static void connection_closed_callback(void *userdata, struct nn_rpc_connection
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
     bool reconnect = !sink->reconnect_timer.disabled;
+    bool disconnected = sink->conn && sink->connection_number > 0;
+    if (sink->conn) {
+        al_assert(sink->conn == conn);
+        sink->conn = NULL;
+    }
     if (reconnect) {
-        if (sink->conn && sink->connection_number > 0) {
+        if (disconnected) {
             log_info("Connection to server closed, attempting reconnect...");
+            nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
         } else {
             log_info("Failed to connect to server, trying again...");
+            nn_timer_again(&sink->reconnect_timer);
         }
     }
-    if (sink->conn) {
-        al_assert(sink->conn == conn);
-        sink->conn = NULL;
-    }
-    if (reconnect) nn_timer_again(&sink->reconnect_timer);
 }
 
 bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
@@ -1593,7 +1700,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     sink->connection_number = 0;
     nn_mutex_init(&sink->lock);
     nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink);
-    nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(2000000));
+    nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1500000));
     nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
     nn_signal_start(&sink->queue_signal);
     camu_queue_init(sink->queue);
@@ -1603,7 +1710,8 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     sink->detached = NULL;
     al_array_init(sink->previous);
     al_array_init(sink->entries);
-    sink->lru = 0;
+    // Start high to exercise the wrapping path.
+    sink->lru = SINK_LRU_MAX - 2;
     mixer->callback = mixer_callback;
     mixer->userdata = sink;
     sink->audio.state = SINK_PAUSED;
@@ -1644,6 +1752,7 @@ void camu_sink_return_current(struct camu_sink *sink)
     nn_mutex_unlock(&sink->lock);
 }
 
+// sink->current could be NULL.
 static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink)
 {
     return ENTRY_IS_VALID(sink->target) ? sink->target : sink->current;
@@ -1714,13 +1823,8 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
 
 void camu_sink_reseek(struct camu_sink *sink)
 {
-    nn_mutex_lock(&sink->lock);
-    struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->lock);
-    if (!current) return;
     queue_cmd(sink, (struct camu_sink_cmd){
-        .op = RESEEK,
-        .opaque = current
+        .op = RESEEK
     });
 }
 
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 3face58..f2ae20f 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -91,6 +91,7 @@ struct camu_sink {
     struct camu_sink_entry *current;
     struct camu_sink_entry *queued;
     struct camu_sink_entry *target;
+    // @TODO: Rename detached to reconnecing.
     struct camu_sink_entry *detached;
     array(struct camu_sink_entry *) previous;
     array(struct camu_sink_entry *) entries;