Change ended entry behavior in list and sink

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

2 files changed, 131 insertions, 130 deletions

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;