Explicitly track sink-side entry ended state

The server/list-side of this change is not done so there are likely bugs
in the set command logic. It works mostly fine for now though.

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

12 files changed, 206 insertions, 178 deletions

diff --git a/flake.lock b/flake.lock
index e08bef1..6acf619 100644
--- a/flake.lock
+++ b/flake.lock
@@ -41,11 +41,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1732482255,
-        "narHash": "sha256-GUffLwzawz5WRVfWaWCg78n/HrBJrOG7QadFY6rtV8A=",
+        "lastModified": 1733073798,
+        "narHash": "sha256-7wEIiBbm1gkHchJCK6Qo7yHV/HZfj8qEhtD80/9HDLg=",
         "owner": "nix-community",
         "repo": "home-manager",
-        "rev": "a9953635d7f34e7358d5189751110f87e3ac17da",
+        "rev": "441fae847ddfe20f9f9a4c47345691a205bb772c",
         "type": "github"
       },
       "original": {
@@ -57,11 +57,11 @@
     },
     "nixos-hardware": {
       "locked": {
-        "lastModified": 1732483221,
-        "narHash": "sha256-kF6rDeCshoCgmQz+7uiuPdREVFuzhIorGOoPXMalL2U=",
+        "lastModified": 1733066523,
+        "narHash": "sha256-aQorWITXZu7b095UwnpUvcGt9dNJie/GO9r4hZfe2sU=",
         "owner": "NixOS",
         "repo": "nixos-hardware",
-        "rev": "45348ad6fb8ac0e8415f6e5e96efe47dd7f39405",
+        "rev": "fe01780d356d70fd119a19277bff71d3e78dad00",
         "type": "github"
       },
       "original": {
@@ -110,11 +110,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1732521221,
-        "narHash": "sha256-2ThgXBUXAE1oFsVATK1ZX9IjPcS4nKFOAjhPNKuiMn0=",
+        "lastModified": 1732837521,
+        "narHash": "sha256-jNRNr49UiuIwaarqijgdTR2qLPifxsVhlJrKzQ8XUIE=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "4633a7c72337ea8fd23a4f2ba3972865e3ec685d",
+        "rev": "970e93b9f82e2a0f3675757eb0bfc73297cc6370",
         "type": "github"
       },
       "original": {
diff --git a/hardware/cmsb-001/configuration.nix b/hardware/cmsb-001/configuration.nix
index 799065a..ca0da74 100644
--- a/hardware/cmsb-001/configuration.nix
+++ b/hardware/cmsb-001/configuration.nix
@@ -38,7 +38,6 @@
     "usbhid"
     "usb_storage"
     "vc4"
-    "v3d"
     "bcm2835_dma"
     "i2c_bcm2835"
     "reset-raspberrypi"
diff --git a/src/buffer/audio.c b/src/buffer/audio.c
index ac698bb..89f1b70 100644
--- a/src/buffer/audio.c
+++ b/src/buffer/audio.c
@@ -211,6 +211,7 @@ void camu_audio_buffer_unpause(struct camu_audio_buffer *buf)
 // flush() always comes from the same thread as push().
 void camu_audio_buffer_flush(struct camu_audio_buffer *buf)
 {
+    al_log_debug("audio_buffer", "Flush requested.");
     if (buf->fmt.resampler_needed) {
         s32 sample_count = buf->resamp->flush(buf->resamp);
         if (sample_count > 0) {
@@ -226,7 +227,6 @@ void camu_audio_buffer_flush(struct camu_audio_buffer *buf)
         buf->buffered = true;
     }
     al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELAXED);
-    al_log_debug("audio_buffer", "Flush requested.");
 }
 
 // Not thread-safe, must be called while the buffer is not being read from or written to.
diff --git a/src/buffer/clock.c b/src/buffer/clock.c
index cee0ec7..e1813f7 100644
--- a/src/buffer/clock.c
+++ b/src/buffer/clock.c
@@ -5,7 +5,6 @@
 
 #define RUNNING 0.0
 #define PAUSED -DBL_MAX
-#define ENDED DBL_MAX
 
 void camu_clock_init(struct camu_clock *clock, void (*callback)(void *, u8), void *userdata)
 {
@@ -75,16 +74,6 @@ bool camu_clock_is_paused(struct camu_clock *clock)
     return al_atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED) == PAUSED;
 }
 
-void camu_clock_end(struct camu_clock *clock)
-{
-    al_atomic_store(f64)(&clock->pause, ENDED, AL_ATOMIC_RELAXED);
-}
-
-bool camu_clock_is_ended(struct camu_clock *clock)
-{
-    return al_atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED) == ENDED;
-}
-
 f64 camu_clock_get_base_pts(struct camu_clock *clock)
 {
     return clock->base;
@@ -93,7 +82,7 @@ f64 camu_clock_get_base_pts(struct camu_clock *clock)
 f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset)
 {
     f64 pause = al_atomic_load(f64)(&clock->pause, AL_ATOMIC_ACQUIRE);
-    if (pause == PAUSED || pause == ENDED) return -1.0;
+    if (pause == PAUSED) return -1.0;
     f64 current = aki_get_tick();
     f64 tick = al_atomic_load(f64)(&clock->tick, AL_ATOMIC_RELAXED);
     if (tick == -1.0) {
diff --git a/src/buffer/clock.h b/src/buffer/clock.h
index 17edb0d..64f4ddb 100644
--- a/src/buffer/clock.h
+++ b/src/buffer/clock.h
@@ -44,8 +44,6 @@ void camu_clock_offset(struct camu_clock *clock, f64 amount);
 void camu_clock_pause(struct camu_clock *clock, u64 ts);
 void camu_clock_resume(struct camu_clock *clock, u64 ts);
 bool camu_clock_is_paused(struct camu_clock *clock);
-void camu_clock_end(struct camu_clock *clock);
-bool camu_clock_is_ended(struct camu_clock *clock);
 
 f64 camu_clock_get_base_pts(struct camu_clock *clock);
 f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset);
diff --git a/src/buffer/video.c b/src/buffer/video.c
index 30e19c7..92bcb2d 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -202,6 +202,7 @@ void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, AVPacket *pk
 // flush() always comes from the same thread as push().
 void camu_video_buffer_flush(struct camu_video_buffer *buf)
 {
+    al_log_debug("video_buffer", "Flush requested.");
     buf->queue->flush(buf->queue);
     if (!buf->buffered) {
         f64 have = buf->queue->count(buf->queue) * buf->avg_frame_duration;
diff --git a/src/fruits/cmsrv/ui.c b/src/fruits/cmsrv/ui.c
index e1e686f..7ae19d5 100644
--- a/src/fruits/cmsrv/ui.c
+++ b/src/fruits/cmsrv/ui.c
@@ -3,7 +3,7 @@
 
 #include "ui.h"
 
-#define LOG_RATIO 2.1
+#define LOG_RATIO 1.5
 
 static s32 resize_cb(struct ncplane *p)
 {
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 80414fd..ede3889 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -10,7 +10,9 @@
 #include "sink.h"
 #include "common.h"
 
+#ifndef CAMU_SINK_NO_VIDEO
 #include "../render/renderer_libplacebo.h"
+#endif
 
 enum {
     SINK_EMPTY = 0,
@@ -95,6 +97,7 @@ static bool entry_buffers_held(struct camu_sink_entry *entry)
 
 static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
+    al_assert(!entry->ended);
     if (entry->audio.state == BUFFER_ADDED) {
         sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
@@ -122,6 +125,7 @@ static void remove_entry_video_buffer(struct camu_sink *sink, struct camu_sink_e
 
 static void remove_entry_buffers(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
+    al_assert(!entry->ended);
 #ifndef CAMU_SINK_NO_VIDEO
     remove_entry_video_buffer(sink, entry);
 #endif
@@ -133,8 +137,9 @@ static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry);
 static void add_video_if_set_and_buffered(struct camu_sink_entry *entry);
 #endif
 
-static void set_or_queue_entry(struct camu_sink_entry *entry)
+static void add_or_queue_entry(struct camu_sink_entry *entry)
 {
+    al_assert(!entry->ended);
     if (entry->audio.state == BUFFER_INIT) {
         entry->audio.state = BUFFER_QUEUED;
     } else {
@@ -315,6 +320,46 @@ static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
     aki_signal_send(&sink->queue_signal);
 }
 
+static void mixer_callback(void *userdata, u8 op)
+{
+    struct camu_sink *sink = (struct camu_sink *)userdata;
+    if (op == CAMU_MIXER_EMPTY) {
+        al_log_debug("sink", "Mixer empty.");
+        queue_cmd(sink, (struct camu_sink_cmd){
+            .op = STOP,
+            .value.i = CAMU_SINK_AUDIO
+        });
+    }
+}
+
+bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop,
+    struct camu_mixer *mixer
+#ifndef CAMU_SINK_NO_VIDEO
+    , struct camu_renderer *renderer
+#endif
+    )
+{
+    sink->loop = loop;
+    aki_mutex_init(&sink->mutex);
+    aki_signal_init(&sink->queue_signal, queue_signal_callback, sink);
+    aki_signal_start(&sink->queue_signal, sink->loop);
+    camu_queue_init(sink->queue);
+    sink->queued = NULL;
+    sink->current = 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;
+#ifndef CAMU_SINK_NO_VIDEO
+    sink->video.renderer = renderer;
+    sink->video.state = SINK_PAUSED;
+#endif
+    return true;
+}
+
 static void maybe_remove_previous(struct camu_sink *sink)
 {
     struct camu_sink_entry *previous;
@@ -324,28 +369,74 @@ static void maybe_remove_previous(struct camu_sink *sink)
     sink->previous.size = 0;
 }
 
-static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *entry, struct camu_sink_entry *current)
+static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target)
 {
-    al_assert(entry != current);
+    al_assert(previous != target);
+    // If our target is ended, remove previous immediately.
+    if (target->ended) {
+        remove_entry_buffers(sink, previous);
+        return;
+    }
     struct camu_sink_entry *rentry;
     al_array_foreach_rev(sink->previous, i, rentry) {
-        if (rentry == current) {
-            // If the entry we are about to add is in previous,
-            // remove it immediately.
-            remove_entry_buffers(sink, current);
-            al_array_remove_at(sink->previous, i);
+        // If the entry we are about to add is in previous, run the queue now.
+        if (rentry == target) {
+            maybe_remove_previous(sink);
+            return;
         }
     }
     // Don't accept duplicates.
     al_array_foreach_rev(sink->previous, i, rentry) {
-        if (rentry == entry) return;
+        if (rentry == previous) return;
+    }
+    al_array_push(sink->previous, previous);
+}
+
+static s32 lru_compare(const void *a, const void *b)
+{
+    struct camu_sink_entry *aa = *((struct camu_sink_entry **)a);
+    struct camu_sink_entry *bb = *((struct camu_sink_entry **)b);
+    if (aa->lru > bb->lru) return -1;
+    else if (aa->lru < bb->lru) return 1;
+    return 0;
+}
+
+static void maybe_cleanup_old_entries(struct camu_sink *sink)
+{
+    // We check size <= MAX_AGE in the loop because sink->lru
+    // is not indicative of the amount of entries we have loaded.
+    // There are various reasons for this but the most obvious is
+    // that it's incremented for buffer and queue operations.
+    //
+    // Handle sink->lru wrapping.
+    // 0 65532 65533 65534 65535
+    // 0 1 65533 65534 65535
+    // 0 1 2 65534 65535
+    // 0 1 2 3 65535
+    // 0 1 2 3 4
+    al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare);
+    struct camu_sink_entry *entry;
+    al_array_foreach_rev(sink->entries, i, entry) {
+        if (entry->lru > sink->lru && (UINT16_MAX - (entry->lru - 1)) + sink->lru >= ENTRY_MAX_AGE) {
+            al_array_remove_at(sink->entries, i);
+            lia_client_disconnect(&entry->client);
+        }
+        if (sink->entries.size <= ENTRY_MAX_AGE) return;
+    }
+    if (sink->lru >= ENTRY_MAX_AGE) {
+        al_array_foreach_rev(sink->entries, i, entry) {
+            if (sink->lru - entry->lru >= ENTRY_MAX_AGE) {
+                al_array_remove_at(sink->entries, i);
+                lia_client_disconnect(&entry->client);
+            }
+            if (sink->entries.size <= ENTRY_MAX_AGE) return;
+        }
     }
-    al_array_push(sink->previous, entry);
 }
 
 void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 {
-    al_assert(entry->audio.state != BUFFER_ADDED);
+    al_assert(!entry->ended && entry->audio.state != BUFFER_ADDED);
     if (entry->audio.state == BUFFER_CONFIGURED) {
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
     } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
@@ -370,7 +461,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 #ifndef CAMU_SINK_NO_VIDEO
 void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
 {
-    al_assert(entry->video.state != BUFFER_ADDED);
+    al_assert(!entry->ended && entry->video.state != BUFFER_ADDED);
     if (entry->video.state == BUFFER_CONFIGURED) {
         entry->video.state = BUFFER_SET_OR_BUFFERED;
     } else if (entry->video.state == BUFFER_SET_OR_BUFFERED) {
@@ -391,16 +482,22 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
 static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
     al_log_info("sink", "Entry ended.");
-    camu_clock_end(&entry->clock);
+    entry->ended = true;
+    // TODO: Can it make sense for this entry to be in previous?
+    maybe_remove_previous(sink);
     if (sink->target) {
-        set_or_queue_entry(sink->target);
+        if (!sink->target->ended) {
+            add_or_queue_entry(sink->target);
+        }
         sink->current = sink->target;
         sink->target = NULL;
         return true;
     } else if (sink->queued) {
         // TODO: What is the right behavior if sink->queued
         // and sink->target are both set.
-        set_or_queue_entry(sink->queued);
+        if (!sink->queued->ended) {
+            add_or_queue_entry(sink->queued);
+        }
         sink->current = sink->queued;
         sink->queued = NULL;
         return true;
@@ -509,17 +606,51 @@ static void video_buffer_callback(void *userdata, u8 op)
 }
 #endif
 
-static void switch_to(struct camu_sink *sink, struct camu_sink_entry *entry)
+static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
 {
     if (sink->current) {
-        if (camu_clock_is_ended(&entry->clock)) {
-            remove_entry_buffers(sink, sink->current);
+        struct camu_sink_entry *current = sink->current;
+        if (current->ended) {
+            bool single_frame = camu_video_buffer_is_single_frame(&current->video.buf);
+            if (single_frame) {
+                remove_entry_video_buffer(sink, current);
+            }
+            al_assert(current->audio.state != BUFFER_ADDED && current->video.state != BUFFER_ADDED);
         } else {
-            maybe_add_to_previous(sink, sink->current, entry);
+            maybe_add_to_previous(sink, current, target);
+        }
+    }
+    if (!target->ended) {
+        add_or_queue_entry(target);
+    }
+    sink->current = target;
+}
+
+static void set_target_and_pause(struct camu_sink *sink, struct camu_sink_entry *target, u64 at)
+{
+    if (!sink->current || sink->current->ended) {
+        if (!target->ended) {
+            add_or_queue_entry(target);
+        }
+        sink->current = target;
+    } else {
+        sink->target = target;
+        camu_clock_pause(&sink->current->clock, at);
+    }
+}
+
+static void clock_callback(void *userdata, u8 op)
+{
+    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
+    struct camu_sink *sink = entry->sink;
+    if (op == CAMU_CLOCK_PAUSED) {
+        aki_mutex_lock(&sink->mutex);
+        if (sink->target) {
+            switch_to(sink, sink->target);
+            sink->target = NULL;
         }
+        aki_mutex_unlock(&sink->mutex);
     }
-    set_or_queue_entry(entry);
-    sink->current = entry;
 }
 
 static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *entry)
@@ -633,10 +764,12 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_SUBTITLE: {
         switch (stream->type) {
         case CAMU_STREAM_SUBTITLE: {
+#ifndef CAMU_SINK_NO_VIDEO
 #ifdef CAMU_HAVE_FFMPEG
             AVPacket *pkt = (AVPacket *)opaque;
             camu_video_buffer_push_subtitle(&entry->video.buf, pkt);
 #endif
+#endif
             break;
         }
         }
@@ -699,11 +832,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         aki_mutex_lock(&sink->mutex);
         if (entry == sink->target) {
             if (sink->current) {
-                remove_entry_buffers(sink, sink->current);
+                if (!sink->current->ended) {
+                    remove_entry_buffers(sink, sink->current);
+                }
                 sink->current = NULL;
             }
             sink->target = NULL;
         } else if (entry == sink->current) {
+            // current's buffers will already be removed.
             sink->current = NULL;
             if (sink->target) {
                 switch_to(sink, sink->target);
@@ -737,103 +873,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     }
 }
 
-static void mixer_callback(void *userdata, u8 op)
-{
-    struct camu_sink *sink = (struct camu_sink *)userdata;
-    if (op == CAMU_MIXER_EMPTY) {
-        al_log_debug("sink", "Mixer empty.");
-        queue_cmd(sink, (struct camu_sink_cmd){
-            .op = STOP,
-            .value.i = CAMU_SINK_AUDIO
-        });
-    }
-}
-
-bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop,
-    struct camu_mixer *mixer
-#ifndef CAMU_SINK_NO_VIDEO
-    , struct camu_renderer *renderer
-#endif
-    )
-{
-    sink->loop = loop;
-    aki_mutex_init(&sink->mutex);
-    aki_signal_init(&sink->queue_signal, queue_signal_callback, sink);
-    aki_signal_start(&sink->queue_signal, sink->loop);
-    camu_queue_init(sink->queue);
-    sink->queued = NULL;
-    sink->current = 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;
-#ifndef CAMU_SINK_NO_VIDEO
-    sink->video.renderer = renderer;
-    sink->video.state = SINK_PAUSED;
-#endif
-    return true;
-}
-
-static s32 lru_compare(const void *a, const void *b)
-{
-    struct camu_sink_entry *aa = *((struct camu_sink_entry **)a);
-    struct camu_sink_entry *bb = *((struct camu_sink_entry **)b);
-    if (aa->lru > bb->lru) return -1;
-    else if (aa->lru < bb->lru) return 1;
-    return 0;
-}
-
-static void maybe_cleanup_old_entries(struct camu_sink *sink)
-{
-    // We check size <= MAX_AGE in the loop because sink->lru
-    // is not indicative of the amount of entries we have loaded.
-    // There are various reasons for this but the most obvious is
-    // that it's incremented for buffer and queue operations.
-    //
-    // Handle sink->lru wrapping.
-    // 0 65532 65533 65534 65535
-    // 0 1 65533 65534 65535
-    // 0 1 2 65534 65535
-    // 0 1 2 3 65535
-    // 0 1 2 3 4
-    al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare);
-    struct camu_sink_entry *entry;
-    al_array_foreach_rev(sink->entries, i, entry) {
-        if (entry->lru > sink->lru && (UINT16_MAX - (entry->lru - 1)) + sink->lru >= ENTRY_MAX_AGE) {
-            al_array_remove_at(sink->entries, i);
-            lia_client_disconnect(&entry->client);
-        }
-        if (sink->entries.size <= ENTRY_MAX_AGE) return;
-    }
-    if (sink->lru >= ENTRY_MAX_AGE) {
-        al_array_foreach_rev(sink->entries, i, entry) {
-            if (sink->lru - entry->lru >= ENTRY_MAX_AGE) {
-                al_array_remove_at(sink->entries, i);
-                lia_client_disconnect(&entry->client);
-            }
-            if (sink->entries.size <= ENTRY_MAX_AGE) return;
-        }
-    }
-}
-
-
-static void clock_callback(void *userdata, u8 op)
-{
-    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
-    struct camu_sink *sink = entry->sink;
-    if (op == CAMU_CLOCK_PAUSED) {
-        aki_mutex_lock(&sink->mutex);
-        if (sink->target) {
-            switch_to(sink, sink->target);
-            sink->target = NULL;
-        }
-        aki_mutex_unlock(&sink->mutex);
-    }
-}
-
 static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, bool *created)
 {
     struct camu_sink_entry *entry;
@@ -846,6 +885,7 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id,
 
     entry = al_alloc_object(struct camu_sink_entry);
     entry->id = id;
+    entry->ended = false;
     entry->sink = sink;
 
     camu_clock_init(&entry->clock, clock_callback, entry);
@@ -884,9 +924,12 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
 
     u8 op = aki_packet_read_u8(packet);
 #ifdef CAMU_SINK_LOCAL
+    // TODO: look over this.
     if (op == LIANA_SINK_UNSET) {
         if (sink->current) {
-            remove_entry_buffers(sink, sink->current);
+            if (!sink->current->ended) {
+                remove_entry_buffers(sink, sink->current);
+            }
             sink->current = NULL;
         }
         goto out;
@@ -931,17 +974,17 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
 #ifdef CAMU_SINK_LOCAL
     (void)at;
     (void)pause;
-    if (sink->current) {
-        if (!camu_clock_is_paused(&sink->current->clock)) {
-            camu_clock_pause(&sink->current->clock, 0);
-        }
-        maybe_add_to_previous(sink, sink->current, entry);
+    if (sink->current && !sink->current->ended && !camu_clock_is_paused(&sink->current->clock)) {
+        camu_clock_pause(&sink->current->clock, 0);
     }
-    set_or_queue_entry(entry);
-    sink->current = entry;
+    switch_to(sink, entry);
     // This will resume a user paused stream.
     camu_clock_resume(&entry->clock, 0);
 #else
+    // PAUSE_NONE and PAUSE_PAUSE mean the server expects the entry
+    // being set to be ended. Not acting accordingly here is the
+    // only place where local entry->ended and the servers expectation
+    // being mismatched can cause issues.
     switch (pause) {
     case LIANA_PAUSE_NONE:
         if (sink->target) {
@@ -949,6 +992,9 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
             sink->target = NULL;
         }
         if (entry != sink->current) {
+            if (camu_clock_is_paused(&entry->clock)) { // TMP
+                camu_clock_resume(&entry->clock, at);
+            }
             switch_to(sink, entry);
         }
         break;
@@ -965,22 +1011,27 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
         }
         break;
     case LIANA_PAUSE_PAUSE: {
-        // For target to be set that must mean that current is set and armed to pause.
         struct camu_sink_entry *prev_target = sink->target;
         if (prev_target) {
+            // For target to be set that must mean that current is set,
+            // armed to pause, and not ended.
+            al_assert(sink->current && !sink->current->ended);
             if (prev_target == entry) {
                 sink->target = NULL;
             } else {
                 sink->target = entry;
             }
-            camu_clock_pause(&prev_target->clock, at);
+            // The targets clock is only guaranteed to be resumed
+            // if it was set in the PAUSE_BOTH case. I feel like
+            // this needs to be simpler.
+            if (!camu_clock_is_paused(&prev_target->clock)) { // TMP
+                camu_clock_pause(&prev_target->clock, at);
+            }
         } else {
-            if (!sink->current || camu_clock_is_ended(&sink->current->clock)) {
-                switch_to(sink, entry);
-            } else {
-                sink->target = entry;
-                camu_clock_pause(&sink->current->clock, at);
+            if (camu_clock_is_paused(&entry->clock)) { // TMP
+                camu_clock_resume(&entry->clock, at);
             }
+            set_target_and_pause(sink, entry, at);
         }
         break;
     }
@@ -988,20 +1039,18 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
         camu_clock_resume(&entry->clock, at);
         struct camu_sink_entry *prev_target = sink->target;
         if (prev_target) {
+            al_assert(sink->current && !sink->current->ended);
             if (prev_target == entry) {
                 sink->target = NULL;
             } else {
                 sink->target = entry;
                 camu_audio_buffer_unpause(&entry->audio.buf);
             }
-            camu_clock_pause(&prev_target->clock, at);
-        } else {
-            if (!sink->current || camu_clock_is_ended(&sink->current->clock)) {
-                switch_to(sink, entry);
-            } else {
-                sink->target = entry;
-                camu_clock_pause(&sink->current->clock, at);
+            if (!camu_clock_is_paused(&prev_target->clock)) { // TMP
+                camu_clock_pause(&prev_target->clock, at);
             }
+        } else {
+            set_target_and_pause(sink, entry, at);
         }
         break;
     }
@@ -1078,6 +1127,7 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con
     if (!current) goto out;
 
     if (current->sequence == sequence) {
+        current->ended = false;
 #ifdef CAMU_SINK_LOCAL
         (void)at;
         lia_client_seek(&current->client, pos, 0);
@@ -1229,12 +1279,6 @@ void camu_sink_offset_volume(struct camu_sink *sink, f64 amount)
 
 void camu_sink_stop(struct camu_sink *sink)
 {
-    aki_mutex_lock(&sink->mutex);
-    if (sink->current) {
-        remove_entry_buffers(sink, sink->current);
-        sink->current = NULL;
-    }
-    aki_mutex_unlock(&sink->mutex);
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = STOP,
         .value.i = CAMU_SINK_AUDIO
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 18ef134..a9a6da0 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -40,6 +40,7 @@ struct camu_sink_entry {
     u16 id;
     s32 sequence;
     u16 lru;
+    bool ended;
     struct camu_clock clock;
     struct lia_client client;
     struct {
diff --git a/src/server/common.h b/src/server/common.h
index ef24601..0d47f48 100644
--- a/src/server/common.h
+++ b/src/server/common.h
@@ -15,9 +15,6 @@ extern str *CAMU_UNIX_LOCAL;
 #define CAMU_LOCAL_TYPE AKI_SOCKET_TCP
 #define CAMU_LOCAL_ADDR CAMU_SERVER_IP
 
-//#define CAMU_LOCAL_TYPE AKI_SOCKET_UNIX
-//#define CAMU_LOCAL_ADDR CAMU_UNIX_PATH
-
 enum {
     CAMU_NODE = 0,
     CAMU_CLIENT,
diff --git a/src/server/server.c b/src/server/server.c
index e43a57f..57d7d7b 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -474,12 +474,12 @@ static void connection_closed_callback(void *userdata, struct aki_rpc_connection
     al_array_foreach(server->sinks, i, sink) {
         if (sink->conn == conn) {
             al_log_info("server", "Sink removed.");
-            al_array_remove_at(server->sinks, i);
             struct lia_list *list;
             al_array_foreach(server->lists, j, list) {
                 lia_list_remove_sink(list, sink);
             }
             cleanup_sink(sink);
+            al_array_remove_at(server->sinks, i);
             break;
         }
     }
diff --git a/src/sink/common.h b/src/sink/common.h
index ed84c39..e59c780 100644
--- a/src/sink/common.h
+++ b/src/sink/common.h
@@ -79,7 +79,6 @@ static bool camu_default_sink_callback(struct camu_screen *scr, struct camu_mixe
         break;
     case CAMU_SINK_EXIT:
         return false;
-
     }
     return true;
 }