Change ended entry behavior in list and sink

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

9 files changed, 173 insertions, 189 deletions

diff --git a/src/buffer/video_null.h b/src/buffer/video_null.h
index 8423b7c..60e73f1 100644
--- a/src/buffer/video_null.h
+++ b/src/buffer/video_null.h
@@ -64,6 +64,7 @@ static void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_co
 
 static void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_codec_packet *packet)
 {
+    // This packet gets freed in the client.
     (void)buf;
     (void)packet;
 }
diff --git a/src/codec/ffmpeg/decoder.h b/src/codec/ffmpeg/decoder.h
index db7a6f8..c6fb5fc 100644
--- a/src/codec/ffmpeg/decoder.h
+++ b/src/codec/ffmpeg/decoder.h
@@ -6,7 +6,7 @@
 #include "../codec.h"
 
 #ifndef CAMU_SINK_NO_VIDEO
-#define CAMU_FF_DECODER_HWACCEL
+//#define CAMU_FF_DECODER_HWACCEL
 #endif
 
 struct camu_ff_decoder {
diff --git a/src/fruits/cmv/cmv.c b/src/fruits/cmv/cmv.c
index e539f01..86b6f75 100644
--- a/src/fruits/cmv/cmv.c
+++ b/src/fruits/cmv/cmv.c
@@ -182,7 +182,7 @@ s32 window_system_main(s32 argc, str *argv)
     if (local) {
         for (s32 i = 1; i < argc; i++) {
             struct nn_packet *packet = nn_packet_create();
-#ifdef CAMU_HAVE_PORTAL
+#if defined CAMU_HAVE_PORTAL
             if (al_str_at(&argv[i], 0) == ';' || camu_is_url(&argv[i], 0)) {
                 nn_packet_write_u8(packet, CAMU_RESOURCE_SIMPLE_SEARCH);
 #elif defined NAUNET_HAS_CURL
diff --git a/src/fruits/common.h b/src/fruits/common.h
index fb9a127..2df5f73 100644
--- a/src/fruits/common.h
+++ b/src/fruits/common.h
@@ -13,5 +13,8 @@ static str CAMU_TEST_CONTROL_PATH = al_str_c("/tmp/camu_control_sock");
 
 AL_UNUSED_VARIABLE_POP
 
+//#define CAMU_TEST_TYPE NNWT_SOCKET_UNIX
+//#define CAMU_TEST_ADDR CAMU_TEST_PATH
+
 #define CAMU_TEST_TYPE NNWT_SOCKET_TCP
 #define CAMU_TEST_ADDR CAMU_TEST_IP
diff --git a/src/liana/client.c b/src/liana/client.c
index 726f8ff..0aec3e5 100644
--- a/src/liana/client.c
+++ b/src/liana/client.c
@@ -179,12 +179,15 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream)
             .pause = LIANA_PAUSE_NONE
         };
         client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time);
-        bool unconfigured = client->mask == 0;
-        if (unconfigured) {
+        struct lia_reconnect_info rec = {
+            .reconnect = true,
+            .unconfigured = client->mask == 0
+        };
+        if (rec.unconfigured) {
             al_assert(client->connection_id == 0);
             al_log_warn("liana", "Handling reconnect on unconfigured client.");
         }
-        client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &unconfigured);
+        client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &rec);
     } else {
         al_assert(client->connection_id == 0);
     }
@@ -212,8 +215,12 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream *
     } else {
         lia_vcr_close_all(&client->vcr);
     }
+    struct lia_reconnect_info rec = {
+        .reconnect = client->reconnect == RECONNECT_ON_CONNECTION_CLOSED,
+        .unconfigured = client->mask == 0
+    };
     // We need to account for REMOVE_BUFFERS possibly running the event loop to wait.
-    client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->reconnect);
+    client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &rec);
     if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) {
         // If reconnect() errors, this will close the client on recursion.
         client->reconnect = RECONNECT_SIGNAL_CLIENT;
diff --git a/src/liana/client.h b/src/liana/client.h
index 4e2ff9e..eb84051 100644
--- a/src/liana/client.h
+++ b/src/liana/client.h
@@ -18,6 +18,11 @@ enum {
     LIANA_CLIENT_CLOSED
 };
 
+struct lia_reconnect_info {
+    bool reconnect;
+    bool unconfigured;
+};
+
 struct lia_prefs {
     u8 enabled_mask;
     s8 audio_lang;
diff --git a/src/liana/list.c b/src/liana/list.c
index 3e2a350..d5eef48 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -43,22 +43,6 @@ void lia_list_init(struct lia_list *list, str *name)
     list->cmd = NULL;
 }
 
-static bool assume_ended(struct lia_list_entry *entry, u64 at)
-{
-    if (entry->duration == LIANA_TIMESTAMP_INVALID) return false;
-    if (entry->ended || entry->duration == 0) return true;
-    if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) {
-        if (entry->offset > entry->duration) {
-            return true;
-        }
-        if (at < entry->start) return false;
-        if (at - entry->start >= entry->duration - entry->offset) {
-            return true;
-        }
-    }
-    return false;
-}
-
 #define META_OPAQUE(op) ((u8[]){ op })
 
 static void signal_meta(struct lia_list *list, struct lia_list_entry *entry, u8 meta)
@@ -169,11 +153,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
         u8 pause;
         u64 at = LIANA_TIMESTAMP_INVALID;
         u64 seek_pos = current->offset;
-        bool ended = assume_ended(current, now);
-        if (ended && current->duration != LIANA_TIMESTAMP_INVALID) {
-            seek_pos = current->duration;
-        }
-        if (ended || current->paused_at != LIANA_TIMESTAMP_INVALID) {
+        if (current->paused_at != LIANA_TIMESTAMP_INVALID) {
             pause = LIANA_PAUSE_NONE;
         } else {
             at = now + LIANA_BASE_DELAY;
@@ -188,7 +168,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
             .at = at,
             .seek_pos = seek_pos,
             .pause = pause,
-            .ended = ended
+            .ended = current->ended
         };
         sink_set(sink, current, list->current, &time);
     }
@@ -303,19 +283,13 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         // Target was never started before.
         target->start = at;
     } else {
-        // An ended entry may never have been paused, but a non-ended entry that wasn't set
-        // cannot be unpaused. Checking assume_ended(target) should be safe here as long
-        // as it can't go from true to false (consideration for seek?).
-        //al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID);
+        // Target is paused or static.
+        al_assert(target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID);
     }
 
-    // These are not equivalent to current/target->ended.
-    bool current_ended = assume_ended(current, now);
-    bool target_ended = assume_ended(target, now);
-
-    // If current is ended or paused.
-    if (current_ended || current->paused_at != LIANA_TIMESTAMP_INVALID) {
-        if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) {
+    // If current->duration = LIANA_TIMESTAMP_INVALID, handling of a static entry happens on the sink.
+    if (current->duration == 0 || current->paused_at != LIANA_TIMESTAMP_INVALID) {
+        if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) {
             // Swap entries and ignore their clocks.
             pause = LIANA_PAUSE_NONE;
         } else {
@@ -324,7 +298,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
             pause = LIANA_PAUSE_RESUME;
         }
     } else { // Current is playing.
-        if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) {
+        if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) {
             // Pause-swap current, don't touch target's clock.
             pause = LIANA_PAUSE_PAUSE;
         } else {
@@ -334,13 +308,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         }
     }
 
-    // Pause current and set the held flag indicating it should be resumed
-    // if it becomes the target of a skip.
-    //
-    // @TODO: Why not hold an ended entry? The idea of ended entries being
-    // unpaused but never held seems like a over-complication.
-    // sink-side ended should only reference buffer state? i.e. pause the entries clock even if ended.
-    if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) {
+    // Pause current to be resumed if it becomes the target of a skip (hold).
+    if (current->duration != 0 && current->paused_at == LIANA_TIMESTAMP_INVALID) {
         current->paused_at = at;
         if (current->paused_at < current->start) {
             current->paused_at = current->start;
@@ -350,14 +319,13 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         current->held = true;
     }
 
-    trace("skipto (#%u-#%u): pause: %hhu, held: %s, current_ended: %s, target_ended: %s.",
-        current->id, target->id, pause, BOOLSTR(current->held), BOOLSTR(current_ended), BOOLSTR(target_ended));
+    trace("skipto(#%u-#%u): pause: %hhu, held: %s.", current->id, target->id, pause, BOOLSTR(current->held));
 
     struct lia_timing time = {
         .at = at,
         .seek_pos = target->offset,
         .pause = pause,
-        .ended = target_ended
+        .ended = target->ended
     };
 
     set_current(list, target, index, &time);
@@ -382,13 +350,12 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
     }
 
     struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);
-    al_assert(entry && !entry->held);
+    if (!entry || entry->ended || entry->duration == 0) return;
+    al_assert(!entry->held);
 
     u64 now = nn_get_timestamp();
     u8 pause = (entry->paused_at == LIANA_TIMESTAMP_INVALID) ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME;
     u64 at;
-    bool ended = assume_ended(entry, now);
-    if (ended) return;
 
     switch (pause) {
     case LIANA_PAUSE_PAUSE:
@@ -409,13 +376,13 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
         break;
     }
 
-    trace("toggle_pause (#%u): pts: %f, pause: %hhu.", entry->id, pts, pause);
+    trace("toggle_pause(#%u): pts: %f, pause: %hhu.", entry->id, pts, pause);
 
     struct lia_timing time = {
         .at = at,
         .seek_pos = LIANA_TIMESTAMP_INVALID,
         .pause = pause,
-        .ended = false
+        .ended = entry->ended
     };
 
     struct lia_list_sink *sink;
@@ -436,7 +403,7 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
     struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);
     al_assert(entry);
 
-    if (entry->duration == LIANA_TIMESTAMP_INVALID || entry->duration == 0) {
+    if (entry->duration == 0 || entry->duration == LIANA_TIMESTAMP_INVALID) {
         warn("Skipping seek on entry with no duration.");
         return;
     }
@@ -455,13 +422,13 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
 
     list->idle = false;
 
-    trace("seek (#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause);
+    trace("seek(#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause);
 
     struct lia_timing time = {
         .at = at,
         .seek_pos = pos,
         .pause = pause,
-        .ended = false
+        .ended = entry->ended
     };
 
     struct lia_list_sink *sink;
@@ -491,7 +458,7 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_id)
         return true;
     }
 
-    trace("end (#%u).", entry->id);
+    trace("end(#%u).", entry->id);
 
     entry->ended = true;
     entry->offset = entry->duration;
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index d3647f1..5038566 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -57,7 +57,7 @@ enum {
 #define SINK_LRU_MAX UINT16_MAX
 
 // Number of entries to keep buffered at one time.
-#define ENTRY_MAX_AGE 3
+#define ENTRY_MAX_AGE 4
 
 // printf format for entries.
 #if defined AL_WE_64BIT
@@ -205,9 +205,10 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry)
     }
 }
 
-// It's possible for some of an entries buffers to be ENDED while others are still
-// ADDED and playing. We handle that by making remove_entry_buffers() and
-// add_audio/video_if_set_and_buffered() no-ops for ENDED buffers.
+// 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, needs special consideration in CLIENT_REMOVE_BUFFERS.
+// BUFFER_ENDED: No-op'd in remove_entry_buffers() and add_or_queue_entry() but otherwise unchanged.
 static void remove_entry_buffers(struct camu_sink_entry *entry)
 {
     trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry));
@@ -225,16 +226,17 @@ static void add_video_if_set_and_buffered(struct camu_sink_entry *entry);
 
 static void add_or_queue_entry(struct camu_sink_entry *entry)
 {
+    trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry));
+    al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED);
+    al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
     if (AUDIO_STATE(entry) == BUFFER_INIT) {
         AUDIO_STATE(entry) = BUFFER_QUEUED;
-    } else {
-        al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED);
+    } else if (AUDIO_STATE(entry) != BUFFER_ENDED) {
         add_audio_if_set_and_buffered(entry);
     }
     if (VIDEO_STATE(entry) == BUFFER_INIT) {
         VIDEO_STATE(entry) = BUFFER_QUEUED;
-    } else {
-        al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
+    } else if (VIDEO_STATE(entry) != BUFFER_ENDED) {
         add_video_if_set_and_buffered(entry);
     }
 }
@@ -490,16 +492,16 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     sink->current = NULL;
     sink->queued = NULL;
     sink->target = NULL;
-    sink->reconnecting = NULL;
+    sink->detached = NULL;
     al_array_init(sink->previous);
     al_array_init(sink->entries);
     sink->lru = 0;
     mixer->callback = mixer_callback;
     mixer->userdata = sink;
-    sink->audio.mixer = mixer;
     sink->audio.state = SINK_PAUSED;
-    sink->video.renderer = renderer;
+    sink->audio.mixer = mixer;
     sink->video.state = SINK_PAUSED;
+    sink->video.renderer = renderer;
     return true;
 }
 
@@ -557,31 +559,27 @@ static void maybe_remove_previous(struct camu_sink *sink)
     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.
 static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key)
 {
+    bool removed = false;
     struct camu_sink_entry *previous;
     al_array_foreach(sink->previous, i, previous) {
         if (previous == key) {
             maybe_remove_previous(sink);
+            removed = true;
             break;
         }
     }
-}
-
-// 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.
-// An obvious example of this is at the point an entry gets freed.
-static void maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry)
-{
-    trace("maybe_remove_from_previous("ENTRY_FMT").", ENTRY_ARG(entry));
-    al_array_remove_all(sink->previous, entry);
+    trace("remove_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed));
 }
 
 // Every call to maybe_add_to_previous() must map to a remove_entry_buffers().
 static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous,
     struct camu_sink_entry *target)
 {
-    trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(previous), ENTRY_ARG(target), AUDIO_STATE(previous), VIDEO_STATE(previous));
+    trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT").", ENTRY_ARG(previous), ENTRY_ARG(target));
     al_assert(previous != target);
     al_assert(!previous->ended);
     // If none of the entry's buffers are added, we don't care about adding it to previous.
@@ -624,10 +622,8 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
     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);
     switch (AUDIO_STATE(entry)) {
-    case BUFFER_ENDED:
-        warn("Tried to add an ended audio buffer.");
-        return;
     case BUFFER_CONFIGURED:
         AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
         break;
@@ -646,17 +642,18 @@ 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);
-    if (VIDEO_STATE(entry) == BUFFER_ENDED) {
-        warn("Tried to add an ended video buffer.");
-        return;
-    } else if (VIDEO_STATE(entry) == BUFFER_CONFIGURED) {
+    al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
+    switch (VIDEO_STATE(entry)) {
+    case BUFFER_CONFIGURED:
         VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
-    } else if (VIDEO_STATE(entry) == BUFFER_SET_OR_BUFFERED) {
+        break;
+    case BUFFER_SET_OR_BUFFERED:
         VIDEO_STATE(entry) = BUFFER_ADDED;
         if (VIDEO_IS_SINGLE_FRAME(entry)) add_entry_video_buffer(entry);
         if (AUDIO_ADDED_OR_EMPTY(entry)) {
             do_add_entry(entry);
         }
+        break;
     }
 }
 
@@ -666,12 +663,16 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
 
     if (sink->current) {
         struct camu_sink_entry *current = sink->current;
+        struct camu_sink_entry *detached = sink->detached;
         al_assert(current != target);
-        bool ensure_removed = sink->reconnecting || current->ended;
-        if (sink->reconnecting) {
-            al_assert(sink->reconnecting == current);
-            sink->reconnecting = NULL;
-            warn("Unset reconnecting as a substitute for remove.");
+        bool ensure_removed = detached || current->ended;
+        if (detached) {
+            al_assert(detached == current);
+            sink->detached = NULL;
+            warn("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;
         } else if (!current->ended) {
             maybe_add_to_previous(sink, current, target);
         }
@@ -721,24 +722,26 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta
     struct camu_sink_entry *current = sink->current;
     trace("pause_and_swap_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(current));
     al_assert(target != current);
-    if (!current || current->ended) {
-        switch_to(sink, target);
-    } else {
-        sink->target = target;
+    if (current) {
+        if (!current->ended) sink->target = target;
         current->audio.ignore_paused = true;
         camu_clock_pause(&current->clock, at);
     }
+    if (!current || current->ended) {
+        switch_to(sink, target);
+    }
 }
 #endif
 
 static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
     info("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry));
+    remove_previous_if_contains(sink, entry);
 #ifdef CAMU_SINK_ONESHOT
     sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL);
     return false;
 #endif
-    maybe_remove_from_previous(sink, entry);
+    // This entry's buffers cannot be added again until after a reset.
     entry->ended = true;
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = END,
@@ -750,6 +753,7 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
     return true;
 #endif
     if (sink->target) {
+        al_assert(!sink->queued);
         switch_to(sink, sink->target);
         sink->target = NULL;
         return true;
@@ -790,16 +794,14 @@ static void audio_buffer_callback(void *userdata, u8 op)
         nn_mutex_unlock(&sink->lock);
         break;
     case CAMU_BUFFER_EOF: {
-        // @TODO: Sync around BUFFER_EOF is not well tested.
         info("Audio EOF.");
         nn_mutex_lock(&sink->lock);
-        // Getting EOF on a buffer that isn't ADDED is very possible if the audio
-        // output is threaded. Even more if we had to wait on the lock above.
+        // This buffer's state could be ADDED, SET_OR_BUFFERED, or CONFIGURED.
+        // Having threaded outputs means anything could have happened while waiting on the lock.
+        // If we were waiting in CLIENT_REMOVE_BUFFERS, state could very well be CONFIGURED.
         if (AUDIO_STATE(entry) == BUFFER_ADDED) {
             remove_entry_audio_buffer(entry);
         }
-        // This assert likely doesn't matter due to the handling of the ENDED state.
-        al_assert(AUDIO_STATE(entry) == BUFFER_SET_OR_BUFFERED);
         AUDIO_STATE(entry) = BUFFER_ENDED;
         if (VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) {
             end_entry_and_advance_queue(sink, entry);
@@ -840,6 +842,7 @@ static void video_buffer_callback(void *userdata, u8 op)
         bool swapped = false;
         bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
         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);
         }
@@ -847,7 +850,6 @@ static void video_buffer_callback(void *userdata, u8 op)
             if (VIDEO_STATE(entry) == BUFFER_ADDED) {
                 remove_entry_video_buffer(entry);
             }
-            al_assert(VIDEO_STATE(entry) == BUFFER_SET_OR_BUFFERED);
             VIDEO_STATE(entry) = BUFFER_ENDED;
             if (AUDIO_ENDED_OR_EMPTY(entry)) {
                 swapped = end_entry_and_advance_queue(sink, entry);
@@ -963,7 +965,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
                 return;
             }
             nn_mutex_lock(&sink->lock);
-            al_assert(AUDIO_STATE(entry) == BUFFER_QUEUED || AUDIO_STATE(entry) == BUFFER_INIT);
+            al_assert(AUDIO_EMPTY(entry));
             if (AUDIO_STATE(entry) == BUFFER_QUEUED) {
                 AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
             } else if (AUDIO_STATE(entry) == BUFFER_INIT) {
@@ -979,7 +981,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
                 return;
             }
             nn_mutex_lock(&sink->lock);
-            al_assert(VIDEO_STATE(entry) == BUFFER_QUEUED || VIDEO_STATE(entry) == BUFFER_INIT);
+            al_assert(VIDEO_EMPTY(entry));
             if (VIDEO_STATE(entry) == BUFFER_QUEUED) {
                 VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
             } else if (VIDEO_STATE(entry) == BUFFER_INIT) {
@@ -1034,70 +1036,77 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         break;
     }
     case LIANA_CLIENT_REMOVE_BUFFERS: {
-        bool reconnect = *(bool *)opaque;
-
+        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
         nn_mutex_lock(&sink->lock);
+        trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current));
 
-        trace("remove_buffers("ENTRY_FMT", %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(reconnect), BOOLSTR(entry == sink->current));
+        // 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);
 
-        if (reconnect && entry == sink->current) {
-            sink->reconnecting = entry;
+        // If this entry is still current on CLIENT_RECONNECTED, re-add buffers.
+        if (rec->reconnect && entry == sink->current) {
+            sink->detached = entry;
         }
 
-        // Entry might be in previous here if it was added to previous then,
-        //  - it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered.
-        //  - it was seeked.
-        remove_previous_if_contains(sink, entry);
+        // Ignore unconfigured entries.
+        if (rec->unconfigured) {
+            al_assert(!entry->ended);
+            al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry));
+            nn_mutex_unlock(&sink->lock);
+            return;
+        }
 
-        if (AUDIO_STATE(entry) == BUFFER_ADDED) {
+        // An empty buffer is guaranteed to not be held.
+        if (!AUDIO_ENDED_OR_EMPTY(entry)) {
+            // Remove for re-add in CLIENT_RECONNECTED.
+            remove_entry_audio_buffer(entry);
+            // Remove again for BUFFER_BUFFERED.
             remove_entry_audio_buffer(entry);
         }
-        bool ignore_video = reconnect && VIDEO_IS_SINGLE_FRAME(entry);
-        if (!ignore_video && VIDEO_STATE(entry) == BUFFER_ADDED) {
+        // Slight optimization. A duplicate frame will still be sent but discarded in the video buffer.
+        bool ignore_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry);
+        if (!ignore_video && !VIDEO_ENDED_OR_EMPTY(entry)) {
+            remove_entry_video_buffer(entry);
             remove_entry_video_buffer(entry);
         }
-
         // Resolve any queued REMOVE_BUFFER requests before blocking.
-        // This is why we are safe to block the event loop thread even if
-        // MIXER_THREADED_START_STOP is not set.
+        // 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.
         run_queue_by_opaque(sink, entry);
 
+        // Unlock to wait.
         nn_mutex_unlock(&sink->lock);
 
         while (entry_audio_buffer_held(entry) || (!ignore_video && entry_video_buffer_held(entry))) {
             BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000));
         }
 
-        // Reset possible ENDED state here in case the entry ended at
-        // some point after unlocking to block above.
+        // Reset possible ENDED state here in case the entry ended at some point after unlocking above.
         nn_mutex_lock(&sink->lock);
 
-        if (reconnect && entry->ended) {
-            entry->ended = false;
-            if (AUDIO_STATE(entry) == BUFFER_QUEUED) {
-                AUDIO_STATE(entry) = BUFFER_INIT;
-            }
-            if (VIDEO_STATE(entry) == BUFFER_QUEUED) {
-                VIDEO_STATE(entry) = BUFFER_INIT;
-            }
-        }
-        // SET_OR_BUFFERED, ADDED, or ENDED.
-        if (AUDIO_STATE(entry) > BUFFER_CONFIGURED) {
+        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;
         }
-        if (!ignore_video && VIDEO_STATE(entry) > BUFFER_CONFIGURED) {
+        if (VIDEO_STATE(entry) == BUFFER_ENDED) {
+            al_assert(!VIDEO_EMPTY(entry));
             VIDEO_STATE(entry) = BUFFER_CONFIGURED;
+        } else if (entry->ended && !ignore_video && VIDEO_EMPTY(entry)) {
+            VIDEO_STATE(entry) = BUFFER_INIT;
         }
+        entry->ended = false;
 
         nn_mutex_unlock(&sink->lock);
 
-        if (reconnect) {
-            if (!AUDIO_EMPTY(entry)) {
-                camu_audio_buffer_reset(&entry->audio.buf);
-            }
-            if (!VIDEO_EMPTY(entry) && !ignore_video) {
-                camu_video_buffer_reset(&entry->video.buf);
-            }
+        // This buffer won't be re-added until after a CLIENT_RECONNECTED event.
+        if (rec->reconnect) {
+            if (!AUDIO_EMPTY(entry)) camu_audio_buffer_reset(&entry->audio.buf);
+            if (!ignore_video && !VIDEO_EMPTY(entry)) camu_video_buffer_reset(&entry->video.buf);
         }
 
         break;
@@ -1109,39 +1118,42 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         time->at = 0;
 #endif
         nn_mutex_lock(&sink->lock);
-#if defined LIANA_LIST_SCUFFED_LOOP
+#ifdef LIANA_LIST_SCUFFED_LOOP
         if (time->seek_pos == 0) {
             camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock));
         } else {
 #endif
             camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
-#if defined LIANA_LIST_SCUFFED_LOOP
+#ifdef LIANA_LIST_SCUFFED_LOOP
         }
 #endif
         nn_mutex_unlock(&sink->lock);
         break;
     }
     case LIANA_CLIENT_RECONNECTED: {
-        bool unconfigured = *(bool *)opaque;
+        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
         nn_mutex_lock(&sink->lock);
-        trace("reconnected("ENTRY_FMT"), reconnecting: "ENTRY_FMT".", ENTRY_ARG(entry), ENTRY_ARG(sink->reconnecting));
-        if (entry == sink->reconnecting) {
+        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);
-            if (AUDIO_STATE(entry) > BUFFER_QUEUED) {
-                al_assert(!unconfigured);
-                add_audio_if_set_and_buffered(entry);
+            al_assert(AUDIO_STATE(entry) != BUFFER_ENDED);
+            al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
+            if (rec->unconfigured) {
+                // The value of unconfigured is consistent from CLIENT_REMOVE_BUFFERS.
+                al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry));
             } else {
-                AUDIO_STATE(entry) = BUFFER_QUEUED;
-            }
-            if (VIDEO_STATE(entry) > BUFFER_QUEUED) {
-                al_assert(!unconfigured);
-                if (!VIDEO_IS_SINGLE_FRAME(entry)) {
+                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);
                 }
-            } else {
-                VIDEO_STATE(entry) = BUFFER_QUEUED;
             }
-            sink->reconnecting = NULL;
+            sink->detached = NULL;
         }
         nn_mutex_unlock(&sink->lock);
         break;
@@ -1165,7 +1177,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         break;
     }
     case LIANA_CLIENT_CLOSED: {
-        // LIANA_CLIENT_REMOVE_BUFFERS has been called on this entry before we're here.
+        // CLIENT_REMOVE_BUFFERS has been called on this entry before we're here.
         nn_mutex_lock(&sink->lock);
 
         if (VIDEO_STATE(entry) == BUFFER_ADDED) {
@@ -1183,9 +1195,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             sink->target = (struct camu_sink_entry *)0xb00b;
             warn("Attempting to handle a disconnected target.");
         } else if (entry == sink->current) {
-            if (sink->reconnecting) {
-                al_assert(sink->reconnecting == sink->current);
-                sink->reconnecting = NULL;
+            if (sink->detached) {
+                al_assert(sink->detached == sink->current);
+                sink->detached = NULL;
             }
             if (sink->target) {
                 switch_to(sink, sink->target);
@@ -1210,7 +1222,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);
-        info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "error" : "cleanup");
+        info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup");
         al_free(entry);
 
         break;
@@ -1240,6 +1252,11 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id)
     camu_clock_init(&entry->clock, clock_callback, entry);
     entry->paused = false;
 
+    entry->client.callback = client_callback;
+    entry->client.userdata = entry;
+    entry->client.renderer = sink->video.renderer;
+    entry->client.prefs = sink->prefs;
+
     AUDIO_STATE(entry) = BUFFER_INIT;
     entry->audio.ignore_paused = false;
     camu_audio_buffer_init(&entry->audio.buf, &entry->clock);
@@ -1251,11 +1268,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id)
     entry->video.buf.callback = video_buffer_callback;
     entry->video.buf.userdata = entry;
 
-    entry->client.callback = client_callback;
-    entry->client.userdata = entry;
-    entry->client.renderer = sink->video.renderer;
-    entry->client.prefs = sink->prefs;
-
     al_array_push(sink->entries, entry);
 
     return entry;
@@ -1304,44 +1316,30 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     u64 seek_pos = nn_packet_read_u64(packet);
     u8 pause = nn_packet_read_u8(packet);
     bool ended = nn_packet_read_bool(packet);
+    (void)ended;
     u32 reset_id = nn_packet_read_u32(packet);
 
-    struct camu_sink_entry *current = sink->current;
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     bool create = !entry;
     if (create) entry = create_entry(sink, id);
     entry->sequence = sequence;
-    sink->lru = al_u16_add_wrap(sink->lru, SINK_LRU_MAX);
+    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
     entry->lru = sink->lru;
     entry->reset_id = reset_id;
-
     if (create) {
+        entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
         camu_clock_set(&entry->clock, seek_pos / 1000000.0);
-        // ended here does not map to entry->ended, it means the server expects the entry to be ended.
-        if (!ended) {
-            entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
-        } else {
-#ifndef CAMU_SINK_LOCAL
-            camu_clock_resume(&entry->clock, 0);
-#endif
-        }
         lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos);
     }
 
     if (op == LIANA_SINK_BUFFER) {
         goto out;
     } else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
-#ifdef CAMU_SINK_LOCAL
-        camu_clock_resume(&entry->clock, 0);
-#else
-        if (pause == LIANA_PAUSE_RESUME) {
-            camu_clock_resume(&entry->clock, at);
-        }
-#endif
         sink->queued = entry;
         goto out;
     }
 
+    struct camu_sink_entry *current = sink->current;
 #ifdef CAMU_SINK_LOCAL
     (void)at;
     al_assert(entry != current);
@@ -1629,8 +1627,11 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
     nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
     u64 duration = 0;
-    if (current) duration = current->client.duration;
-    f64 pts = camu_clock_get_last_pts(&current->clock);
+    f64 pts = 0.0;
+    if (current) {
+        duration = current->client.duration;
+        pts = camu_clock_get_last_pts(&current->clock);
+    }
     nn_mutex_unlock(&sink->lock);
     if (!current || duration == 0) return;
     struct camu_sink_cmd cmd = {
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index d1bcd26..ba37241 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -88,7 +88,7 @@ struct camu_sink {
     struct camu_sink_entry *current;
     struct camu_sink_entry *queued;
     struct camu_sink_entry *target;
-    struct camu_sink_entry *reconnecting;
+    struct camu_sink_entry *detached;
     array(struct camu_sink_entry *) previous;
     array(struct camu_sink_entry *) entries;
     u16 lru;