More robust disconnect for errored entries

 - Improve >1 width character handling in tui input.

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

10 files changed, 143 insertions, 84 deletions

diff --git a/flake.lock b/flake.lock
index a44a7d9..133fbee 100644
--- a/flake.lock
+++ b/flake.lock
@@ -23,11 +23,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1740845322,
-        "narHash": "sha256-AXEgFj3C0YJhu9k1OhbRhiA6FnDr81dQZ65U3DhaWpw=",
+        "lastModified": 1741056285,
+        "narHash": "sha256-/JKDMVqq8PIqcGonBVKbKq1SooV3kzGmv+cp3rKAgPA=",
         "owner": "nix-community",
         "repo": "home-manager",
-        "rev": "fcac3d6d88302a5e64f6cb8014ac785e08874c8d",
+        "rev": "70fbbf05a5594b0a72124ab211bff1d502c89e3f",
         "type": "github"
       },
       "original": {
@@ -91,11 +91,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1740828860,
-        "narHash": "sha256-cjbHI+zUzK5CPsQZqMhE3npTyYFt9tJ3+ohcfaOF/WM=",
+        "lastModified": 1741010256,
+        "narHash": "sha256-WZNlK/KX7Sni0RyqLSqLPbK8k08Kq7H7RijPJbq9KHM=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "303bd8071377433a2d8f76e684ec773d70c5b642",
+        "rev": "ba487dbc9d04e0634c64e3b1f0d25839a0a68246",
         "type": "github"
       },
       "original": {
diff --git a/flake.nix b/flake.nix
index 4f767db..2dea414 100644
--- a/flake.nix
+++ b/flake.nix
@@ -233,7 +233,7 @@
           xxHash
           vulkan-headers
           # Many more things need to be fixed for notcurses on Windows.
-          ((ncurses.override { enableStatic = true; withCxx = false; withTermlib = false; }).overrideAttrs (oldAttrs: rec {
+          ((ncurses.override { enableStatic = true; withCxx = true; withTermlib = false; }).overrideAttrs (oldAttrs: rec {
             version = "6.5.20250118";
             src = fetchurl {
               url = "https://invisible-island.net/archives/ncurses/ncurses-${lib.versions.majorMinor version}.tar.gz";
@@ -241,7 +241,14 @@
             };
             #configureFlags = lib.remove "--enable-sp-funcs" (lib.remove "--enable-term-driver" oldAttrs.configureFlags) ++ [
             configureFlags = oldAttrs.configureFlags ++ [
-            #  "--enable-exp-win32"
+              "--without-pthread"
+              "--without-ada"
+              "--enable-interop"
+              "--enable-database"
+              "--disable-home-terminfo"
+              "--disable-rpath"
+              "--disable-symlinks"
+              #"--enable-exp-win32"
             ];
             patches = (lib.lists.forEach [
                 #{ date = "20240427"; hash = "qiAf4WrIiihsLaekMPJFZFh8qdCUrp+9U9zmlrG4lSw="; }
diff --git a/src/fruits/cmc/ui/pane_search.c b/src/fruits/cmc/ui/pane_search.c
index e75caa5..e273504 100644
--- a/src/fruits/cmc/ui/pane_search.c
+++ b/src/fruits/cmc/ui/pane_search.c
@@ -37,6 +37,8 @@ void cmc_sp_add_search(struct cmc_ui *ui, struct cmc_search *search)
 
 static bool handle_text_input(struct cmc_search_tab *tab, struct ncinput *input)
 {
+    bool entered = false;
+
     switch (input->id) {
     case NCKEY_BACKSPACE:
         if (tab->cursor > 0) {
@@ -59,7 +61,8 @@ static bool handle_text_input(struct cmc_search_tab *tab, struct ncinput *input)
         tab->input_active = false;
         // Erase text on next render.
         tab->input_changed = true;
-        return true;
+        entered = true;
+        break;
     default:
         if (ncinput_ctrl_p(input)) {
             switch (input->id) {
@@ -111,7 +114,15 @@ static bool handle_text_input(struct cmc_search_tab *tab, struct ncinput *input)
 
         break;
     }
-    return false;
+
+    tab->visual_cursor = 0;
+    if (tab->cursor > 0) {
+        for (u32 i = 0; i < tab->cursor; i++) {
+            tab->visual_cursor += al_wchar_width(al_wstr_at(&tab->input_text, i));
+        }
+    }
+
+    return entered;
 }
 
 bool cmc_sp_handle_input(struct cmc_ui *ui, struct ncinput *input)
@@ -142,7 +153,7 @@ bool cmc_sp_handle_input(struct cmc_ui *ui, struct ncinput *input)
                 tab->input_changed = true;
                 s32 absx = ncplane_abs_x(tab->input);
                 s32 absy = ncplane_abs_y(tab->input);
-                notcurses_cursor_enable(ui->nc, absy, absx + tab->cursor);
+                notcurses_cursor_enable(ui->nc, absy, absx + tab->visual_cursor);
                 break;
             }
         }
@@ -174,7 +185,7 @@ void cmc_sp_render(struct cmc_ui *ui)
         if (tab->input_active) {
             s32 absx = ncplane_abs_x(tab->input);
             s32 absy = ncplane_abs_y(tab->input);
-            notcurses_cursor_enable(ui->nc, absy, absx + tab->cursor);
+            notcurses_cursor_enable(ui->nc, absy, absx + tab->visual_cursor);
         }
     }
 
diff --git a/src/fruits/cmc/ui/ui.h b/src/fruits/cmc/ui/ui.h
index 996bb73..2f73a81 100644
--- a/src/fruits/cmc/ui/ui.h
+++ b/src/fruits/cmc/ui/ui.h
@@ -21,6 +21,7 @@ struct cmc_search_tab {
     struct cmc_search *search;
     struct ncplane *input;
     u32 cursor;
+    u32 visual_cursor;
     bool input_active;
     wstr input_text;
     bool input_changed;
diff --git a/src/liana/client.c b/src/liana/client.c
index 54b5584..b53fb7e 100644
--- a/src/liana/client.c
+++ b/src/liana/client.c
@@ -13,7 +13,8 @@
 enum {
     RECONNECT_NONE = 0,
     RECONNECT_ON_CONNECTION_CLOSED,
-    RECONNECT_SIGNAL_CLIENT
+    RECONNECT_SIGNAL_CLIENT,
+    RECONNECT_DISCONNECTED
 };
 
 static void data_packet_callback(void *userdata, struct nn_packet_stream *stream, struct nn_packet *packet)
@@ -259,8 +260,9 @@ void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop,
 
 void lia_client_seek(struct lia_client *client, u64 pos, u64 at)
 {
+    if (client->reconnect == RECONNECT_DISCONNECTED) return;
     // If reconnect = ON_CONNECTION_CLOSED or SIGNAL_CLIENT, we are safe to edit pos
-    // and at inplace because they aren't evaluated until connection_callback().
+    // and at in-place because they aren't evaluated until connection_callback().
     client->pos = pos;
     client->at = at;
     if (client->reconnect == RECONNECT_NONE) {
@@ -277,7 +279,8 @@ void lia_client_reseek(struct lia_client *client)
 void lia_client_disconnect(struct lia_client *client)
 {
     u8 reconnect = client->reconnect;
-    client->reconnect = RECONNECT_NONE;
+    al_assert(reconnect != RECONNECT_DISCONNECTED);
+    client->reconnect = RECONNECT_DISCONNECTED;
     if (reconnect != RECONNECT_ON_CONNECTION_CLOSED) {
         nn_packet_stream_disconnect(&client->data);
     }
diff --git a/src/liana/list.c b/src/liana/list.c
index cccca8e..650b6c5 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -40,7 +40,7 @@ void lia_list_init(struct lia_list *list, str *name)
     al_array_init(list->entries);
     al_array_init(list->sinks);
     al_array_init(list->command_queue);
-    list->active_cmd = NULL;
+    list->cmd = NULL;
 }
 
 static bool assume_ended(struct lia_list_entry *entry, u64 at)
@@ -73,10 +73,18 @@ static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_e
     if (status == LIANA_ENTRY_ERRORED) {
         if (sequence >= 0) {
             al_array_remove_at(list->entries, (u32)sequence);
-            if (sequence == list->current && (u32)list->current == list->entries.count) {
-                // This maps to the behavior of only skipping ahead on errors.
+            if (list->cmd->sequence == sequence) {
+                // @TODO: This command has to be effectively discarded.
+                list->cmd->sequence = -1;
+            } else if (list->cmd->sequence > sequence) {
+                list->cmd->sequence--;
+            }
+            if (list->current >= sequence) {
                 list->current--;
-                list->idle = true;
+                // This maps to the behavior of only skipping ahead on errors.
+                if ((u32)list->current == list->entries.count) {
+                    list->idle = true;
+                }
             }
         }
         *error = true;
@@ -391,7 +399,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
 static bool handle_skip(struct lia_list *list, s32 sequence, s32 n)
 {
     if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
-    struct lia_list_cmd *cmd = list->active_cmd;
+    struct lia_list_cmd *cmd = list->cmd;
     cmd->op = SKIPTO;
     cmd->sequence = sequence;
     cmd->arg0.i = sequence + n;
@@ -556,7 +564,7 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_id)
             list->queued = -1;
             signal_meta(list, queued, LIANA_META_CURRENT_CHANGED);
         } else if (next < size) {
-            struct lia_list_cmd *cmd = list->active_cmd;
+            struct lia_list_cmd *cmd = list->cmd;
             cmd->op = SKIPTO;
             cmd->sequence = sequence;
             cmd->arg0.i = next;
@@ -573,7 +581,7 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_id)
 static bool adjust_current(struct lia_list *list, struct lia_list_entry *previous)
 {
     al_assert(list->current >= 0);
-    struct lia_list_cmd *cmd = list->active_cmd;
+    struct lia_list_cmd *cmd = list->cmd;
     struct lia_list_entry *entry;
     al_array_foreach(list->entries, i, entry) {
         if (entry->opaque == previous->opaque) {
@@ -646,11 +654,11 @@ static bool handle_shuffle(struct lia_list *list)
 static void run_queue(struct lia_list *list)
 {
     // @TODO: What does current = -1/currentless really mean.
-    if (!list->active_cmd) {
+    if (!list->cmd) {
         if (!list->command_queue.count) return;
-        al_array_pop_at(list->command_queue, 0, list->active_cmd);
+        al_array_pop_at(list->command_queue, 0, list->cmd);
     }
-    struct lia_list_cmd *cmd = list->active_cmd;
+    struct lia_list_cmd *cmd = list->cmd;
     switch (cmd->op) {
     case ADD_SINK:
         if (!handle_add_sink(list, cmd->sink)) {
@@ -711,7 +719,7 @@ static void run_queue(struct lia_list *list)
         break;
     }
     al_free(cmd);
-    list->active_cmd = NULL;
+    list->cmd = NULL;
     pump_queue(list);
 }
 
@@ -876,7 +884,7 @@ void lia_list_free(struct lia_list *list)
         al_free(cmd);
     }
     al_array_free(list->command_queue);
-    if (list->active_cmd) al_free(list->active_cmd);
+    if (list->cmd) al_free(list->cmd);
 
     struct lia_list_entry *entry;
     al_array_foreach(list->entries, i, entry) {
diff --git a/src/liana/list.h b/src/liana/list.h
index a60790a..0d80dde 100644
--- a/src/liana/list.h
+++ b/src/liana/list.h
@@ -113,7 +113,7 @@ struct lia_list {
     array(struct lia_list_entry *) entries;
     array(struct lia_list_sink *) sinks;
     array(struct lia_list_cmd *) command_queue;
-    struct lia_list_cmd *active_cmd;
+    struct lia_list_cmd *cmd;
     void (*callback)(void *, u8, struct lia_list_entry *, void *);
     void *userdata;
 };
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 3528e6b..3f337d5 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -35,12 +35,15 @@ enum {
 
 // Command queue commands.
 enum {
+    // Sink operations.
     START = 0,
     STOP,
     ADD_BUFFER,
     REMOVE_BUFFER,
-    CLEAR,
+    CLEAR_BUFFERS,
+    ERROR_OUT,
     CLOSE,
+    // List actions.
     TOGGLE_PAUSE,
     SEEK,
     SKIP,
@@ -224,6 +227,16 @@ static void add_or_queue_entry(struct camu_sink_entry *entry)
 #endif
 }
 
+// Disconnecting a packet stream twice before a reconnect is an error.
+static void maybe_disconnect_entry(struct camu_sink_entry *entry)
+{
+    if (!entry->disconnected) {
+        // disconnect() could free entry.
+        entry->disconnected = true;
+        lia_client_disconnect(&entry->client);
+    }
+}
+
 #ifdef CAMU_SINK_LOCAL
 static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
@@ -342,7 +355,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         }
         break;
     }
-    case CLEAR: {
+    case CLEAR_BUFFERS: {
         switch (cmd->value.i) {
         case CAMU_SINK_AUDIO:
             sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_AUDIO, NULL);
@@ -355,6 +368,11 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         }
         break;
     }
+    case ERROR_OUT: {
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+        maybe_disconnect_entry(entry);
+        break;
+    }
     case CLOSE: {
         nn_signal_stop(&sink->queue_signal);
         sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
@@ -472,7 +490,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     )
 {
     sink->loop = loop;
-    nn_mutex_init(&sink->mutex);
+    nn_mutex_init(&sink->lock);
     nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
     nn_signal_start(&sink->queue_signal);
     camu_queue_init(sink->queue);
@@ -580,7 +598,7 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
         // 0 1 2 3 4
         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);
+            maybe_disconnect_entry(entry);
         }
         if (sink->entries.count <= ENTRY_MAX_AGE) return;
     }
@@ -592,7 +610,7 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
             al_assert(sink->lru >= entry->lru);
             if (sink->lru - entry->lru >= ENTRY_MAX_AGE) {
                 al_array_remove_at(sink->entries, i);
-                lia_client_disconnect(&entry->client);
+                maybe_disconnect_entry(entry);
             }
             if (sink->entries.count <= ENTRY_MAX_AGE) return;
         }
@@ -809,9 +827,9 @@ static void audio_buffer_callback(void *userdata, u8 op)
     switch (op) {
     case CAMU_BUFFER_BUFFERED:
         lia_vcr_set_buffered(entry->audio.track);
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         add_audio_if_set_and_buffered(entry);
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     case CAMU_BUFFER_CORK:
         lia_vcr_cork(entry->audio.track);
@@ -820,7 +838,7 @@ static void audio_buffer_callback(void *userdata, u8 op)
         lia_vcr_uncork(entry->audio.track);
         break;
     case CAMU_BUFFER_PAUSED:
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         if (!entry->audio.ignore_paused && entry->buffers_paused) {
             al_log_info("sink", "Audio buffer paused.");
             queue_cmd(sink, (struct camu_sink_cmd){
@@ -828,10 +846,10 @@ static void audio_buffer_callback(void *userdata, u8 op)
                 .value.i = CAMU_SINK_AUDIO
             });
         }
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     case CAMU_BUFFER_EOF: {
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         al_log_info("sink", "Audio EOF.");
         // @TODO: This is not well synced. EOF can happen at any time
         //  while other stuff is happening in the sink. For example
@@ -853,11 +871,15 @@ static void audio_buffer_callback(void *userdata, u8 op)
         if (end_entry) {
             end_entry_and_advance_queue(sink, entry);
         }
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     }
     case CAMU_BUFFER_ERRORED:
         al_log_error("sink", "Audio buffer errored.");
+        queue_cmd(sink, (struct camu_sink_cmd){
+            .op = ERROR_OUT,
+            .opaque = entry
+        });
         break;
     }
 }
@@ -870,9 +892,9 @@ static void video_buffer_callback(void *userdata, u8 op)
     switch (op) {
     case CAMU_BUFFER_BUFFERED: {
         lia_vcr_set_buffered(entry->video.track);
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         add_video_if_set_and_buffered(entry);
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     }
     case CAMU_BUFFER_CORK:
@@ -884,7 +906,7 @@ static void video_buffer_callback(void *userdata, u8 op)
     case CAMU_BUFFER_EOF: {
         bool swapped = false;
         bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         al_log_info("sink", "Video EOF.");
         if (!AUDIO_EMPTY(entry)) {
             camu_audio_buffer_set_no_video(&entry->audio.buf, true);
@@ -899,7 +921,7 @@ static void video_buffer_callback(void *userdata, u8 op)
                 swapped = end_entry_and_advance_queue(sink, entry);
             }
         }
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         if (!swapped && !single_frame) {
             queue_cmd(sink, (struct camu_sink_cmd){
                 .op = STOP,
@@ -910,6 +932,10 @@ static void video_buffer_callback(void *userdata, u8 op)
     }
     case CAMU_BUFFER_ERRORED:
         al_log_error("sink", "Video buffer errored.");
+        queue_cmd(sink, (struct camu_sink_cmd){
+            .op = ERROR_OUT,
+            .opaque = entry
+        });
         break;
     }
 }
@@ -923,7 +949,7 @@ static void clock_callback(void *userdata, u8 op)
 #ifdef CAMU_SINK_TRACE
         al_log_info("sink", "clock_callback(), target: 0x%llx", sink->target);
 #endif
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         if (entry == sink->current) {
             if (sink->target) {
                 switch_to(sink, sink->target);
@@ -937,7 +963,7 @@ static void clock_callback(void *userdata, u8 op)
 #endif
             }
         }
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
     }
 }
 
@@ -1004,10 +1030,10 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             entry->audio.track = (struct lia_vcr_track *)opaque;
             if (!camu_audio_buffer_configure(&entry->audio.buf, stream, sink->audio.mixer)) {
                 al_log_warn("sink", "Audio buffer failed to configure.");
-                lia_client_disconnect(&entry->client);
+                maybe_disconnect_entry(entry);
                 return;
             }
-            nn_mutex_lock(&sink->mutex);
+            nn_mutex_lock(&sink->lock);
             if (entry->audio.state == BUFFER_QUEUED) {
                 entry->audio.state = BUFFER_SET_OR_BUFFERED;
             } else if (entry->audio.state == BUFFER_INIT) {
@@ -1015,17 +1041,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             } else {
                 al_assert(false);
             }
-            nn_mutex_unlock(&sink->mutex);
+            nn_mutex_unlock(&sink->lock);
             break;
 #ifndef CAMU_SINK_NO_VIDEO
         case CAMU_STREAM_VIDEO:
             entry->video.track = (struct lia_vcr_track *)opaque;
             if (!camu_video_buffer_configure(&entry->video.buf, stream, sink->video.renderer)) {
                 al_log_warn("sink", "Video buffer failed to configure.");
-                lia_client_disconnect(&entry->client);
+                maybe_disconnect_entry(entry);
                 return;
             }
-            nn_mutex_lock(&sink->mutex);
+            nn_mutex_lock(&sink->lock);
             if (entry->video.state == BUFFER_QUEUED) {
                 entry->video.state = BUFFER_SET_OR_BUFFERED;
             } else if (entry->video.state == BUFFER_INIT) {
@@ -1033,7 +1059,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             } else {
                 al_assert(false);
             }
-            nn_mutex_unlock(&sink->mutex);
+            nn_mutex_unlock(&sink->lock);
             break;
         case CAMU_STREAM_SUBTITLE:
             if (!camu_video_buffer_configure_subtitles(&entry->video.buf, stream)) {
@@ -1051,9 +1077,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     }
     case LIANA_CLIENT_CONFIGURE_COMPLETE: {
         // All present buffers were configured.
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         evaluate_and_set_buffer_params(sink, entry);
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     }
     case LIANA_CLIENT_DATA: {
@@ -1093,7 +1119,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_REMOVE_BUFFERS: {
         bool reconnect = *(bool *)opaque;
 
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
 
 #ifdef CAMU_SINK_TRACE
         al_log_info("sink", "remove_buffers(%s), entry == current: %s", BOOLSTR(reconnect), BOOLSTR(entry == sink->current));
@@ -1133,7 +1159,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // MIXER_THREADED_START_STOP is not set.
         run_queue_by_opaque(sink, entry);
 
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
 
         while ( // Block until buffers are removed.
 #ifndef CAMU_SINK_NO_VIDEO
@@ -1145,7 +1171,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 
         // Reset possible ENDED state here in case the entry ended at
         // some point after unlocking to block above.
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
 
         if (reconnect && entry->ended) {
             entry->ended = false;
@@ -1172,7 +1198,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         }
 #endif
 
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
 
         if (reconnect) {
             if (!AUDIO_EMPTY(entry)) {
@@ -1189,10 +1215,10 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     }
     case LIANA_CLIENT_RESUME_AT: {
         struct lia_timing *time = (struct lia_timing *)opaque;
-        nn_mutex_lock(&sink->mutex);
 #ifdef CAMU_SINK_TRACE
         al_log_info("sink", "resume_at(%llu, %llu)", time->seek_pos, time->at);
 #endif
+        nn_mutex_lock(&sink->lock);
 #if defined LIANA_LIST_SCUFFED_LOOP && !defined CAMU_SINK_NO_VIDEO
         struct camu_video_buffer *buf = &entry->video.buf;
         if (time->seek_pos == 0 && buf->last_pts >= 0.0) {
@@ -1203,11 +1229,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 #else
         camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
 #endif
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     }
     case LIANA_CLIENT_RECONNECTED: {
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
         if (entry == sink->reconnecting) {
             al_assert(entry == sink->current);
             if (entry->audio.state > BUFFER_QUEUED) {
@@ -1226,7 +1252,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 #endif
             sink->reconnecting = NULL;
         }
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
         break;
     }
     case LIANA_CLIENT_EOF: {
@@ -1251,7 +1277,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     }
     case LIANA_CLIENT_CLOSED: {
         // LIANA_CLIENT_REMOVE_BUFFERS has been called on this entry before we're here.
-        nn_mutex_lock(&sink->mutex);
+        nn_mutex_lock(&sink->lock);
 
         bool removed;
         al_array_remove_checked(sink->entries, entry, removed);
@@ -1288,7 +1314,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             sink->queued = NULL;
         }
 
-        nn_mutex_unlock(&sink->mutex);
+        nn_mutex_unlock(&sink->lock);
 
         lia_client_free(&entry->client);
         camu_audio_buffer_free(&entry->audio.buf);
@@ -1308,6 +1334,7 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id)
     struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry);
     entry->sink = sink;
     entry->id = id;
+    entry->disconnected = false;
     entry->ended = false;
 
     camu_clock_init(&entry->clock, clock_callback, entry);
@@ -1353,7 +1380,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     (void)conn;
     (void)rpacket;
 
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
 
     u8 op = nn_packet_read_u8(packet);
     if (op == LIANA_SINK_UNSET) {
@@ -1503,7 +1530,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
 #endif
 
 out:
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     if (op != LIANA_SINK_BUFFER) {
         maybe_cleanup_old_entries(sink);
     }
@@ -1529,7 +1556,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     if (!entry) goto out;
     al_assert(entry->sequence == sequence);
 
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
 #ifdef CAMU_SINK_LOCAL
     (void)at;
     (void)pause;
@@ -1563,7 +1590,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
         break;
     }
 #endif
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
 
 out:
     nn_packet_stream_return_packet(conn->stream, packet);
@@ -1670,13 +1697,13 @@ bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str
 
 struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     return sink->current;
 }
 
 void camu_sink_return_current(struct camu_sink *sink)
 {
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
 }
 
 void camu_sink_skip(struct camu_sink *sink, s32 n)
@@ -1696,9 +1723,9 @@ void camu_sink_shuffle(struct camu_sink *sink)
 
 void camu_sink_toggle_pause(struct camu_sink *sink)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     if (current) {
         queue_cmd(sink, (struct camu_sink_cmd){
             .op = TOGGLE_PAUSE,
@@ -1710,9 +1737,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
 
 void camu_sink_seek(struct camu_sink *sink, f64 precent)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = SEEK,
         .value.f = precent,
@@ -1734,9 +1761,9 @@ static f64 tmp_get_entry_pts(struct camu_sink_entry *entry)
 
 void camu_sink_relative_seek(struct camu_sink *sink, f64 offset)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     if (!current) return;
     f64 pts = tmp_get_entry_pts(current) + offset;
     f64 duration = current->client.duration / 1000000.0;
@@ -1749,9 +1776,9 @@ void camu_sink_relative_seek(struct camu_sink *sink, f64 offset)
 
 void camu_sink_reseek(struct camu_sink *sink)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     if (!current) return;
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = RESEEK,
@@ -1780,9 +1807,9 @@ static char status[128];
 
 void camu_sink_status(struct camu_sink *sink)
 {
-    nn_mutex_lock(&sink->mutex);
+    nn_mutex_lock(&sink->lock);
     struct camu_sink_entry *current = sink->current;
-    nn_mutex_unlock(&sink->mutex);
+    nn_mutex_unlock(&sink->lock);
     if (!current) {
         al_log_info("sink", "Nothing playing.");
         return;
@@ -1819,7 +1846,7 @@ void camu_sink_stop(struct camu_sink *sink)
         .value.i = CAMU_SINK_AUDIO
     });
     queue_cmd(sink, (struct camu_sink_cmd){
-        .op = CLEAR,
+        .op = CLEAR_BUFFERS,
         .value.i = CAMU_SINK_AUDIO
     });
     queue_cmd(sink, (struct camu_sink_cmd){
@@ -1835,7 +1862,7 @@ void camu_sink_close(struct camu_sink *sink)
     struct camu_sink_entry *entry;
     al_array_foreach_rev(sink->entries, i, entry) {
         al_array_remove_at(sink->entries, i);
-        lia_client_disconnect(&entry->client);
+        maybe_disconnect_entry(entry);
     }
 }
 
@@ -1848,7 +1875,7 @@ void camu_sink_free(struct camu_sink *sink)
     al_array_free(sink->entries);
     nn_rpc_free(&sink->client);
     camu_queue_free(sink->queue);
-    nn_mutex_destroy(&sink->mutex);
+    nn_mutex_destroy(&sink->lock);
     al_str_free(&sink->addr);
     al_str_free(&sink->name);
 }
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 4e97f3a..ae8e2e4 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -43,6 +43,7 @@ struct camu_sink_entry {
     struct lia_client client;
     s32 sequence;
     u16 lru;
+    bool disconnected;
     bool ended;
     u32 reset_id;
     struct camu_clock clock;
@@ -78,7 +79,7 @@ struct camu_sink {
     u16 port;
     struct nn_rpc client;
     struct nn_rpc_connection *conn;
-    struct nn_mutex mutex;
+    struct nn_mutex lock;
     struct nn_timer periodic_timer;
     struct nn_signal queue_signal;
     queue(struct camu_sink_cmd) queue;
diff --git a/src/server/server.c b/src/server/server.c
index 10204df..d91dd6f 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -278,7 +278,7 @@ static struct cch_entry *entry_from_post(struct camu_server *server, struct camu
         }
     }
     if (!entry) {
-        al_log_warn("server", "Failed to load resource %.*s %u.", al_str_fmt(&post->unique_id), index);
+        al_log_error("server", "Failed to load post media %.*s %u.", al_str_fmt(&post->unique_id), index);
         return NULL;
     }
     entry->handler->maybe_spawn_worker(entry->handler, 0);
@@ -418,6 +418,7 @@ static void list_callback(void *userdata, u8 op, struct lia_list_entry *entry, v
             send_clients_entry_seeked(server, entry->list, entry);
             break;
         case LIANA_META_ENTRY_ERRORED:
+            al_log_error("server", "Failed to load: %ls.", entry->name.data);
             break;
         case LIANA_META_ORDER_CHANGED:
             break;