From 98f833980d7b4b2c649b3af774ae1179be27a143 Mon Sep 17 00:00:00 2001
From: TapTap <taptap.programming@gmail.com>
Date: Sat, 15 Aug 2026 13:20:50 +0200
Subject: [PATCH] fix: handle pipeline cancellation failures

---
 src/client/client_send.c     | 12 +++++++++---
 src/shared/multiprocessing.c |  9 +++++++++
 2 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/src/client/client_send.c b/src/client/client_send.c
index 76872bd..9fc25dc 100644
--- a/src/client/client_send.c
+++ b/src/client/client_send.c
@@ -60,6 +60,7 @@ static Client* connect_transfer_client(const Config* config) {
     connected = client_connect(client, config->server_host, config->server_port);
   }
   if (!connected) {
+    client_disconnect(client);
     client_delete(client);
     return NULL;
   }
@@ -399,6 +400,13 @@ static int send_chunks_multithreaded(void* pipeline_context) {
         context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader,
         &context->condition_not_full_loader, &context->loader_done);
     if (current_chunk == NULL) {
+      if (atomic_load(&context->cancelled)) {
+        pipeline_cancel(context);
+        disconnect_transfer_client(client);
+        mark_sender_done(context);
+        protocol_session_unbind();
+        return thrd_error;
+      }
       if (context->config->use_delete) {
         if (send_delete_manifest(client->file_descriptor, context->manifest) != 0)
           goto send_fail;
@@ -523,9 +531,7 @@ static int load_files_multithreaded(void* pipeline_context) {
                                             &context->condition_not_full_loader,
                                             &context->cancelled)) {
       chunk_destroy(chunk);
-      atomic_store(&context->cancelled, true);
-      cnd_broadcast(&context->condition_not_full_loader);
-      cnd_broadcast(&context->condition_not_empty_loader);
+      pipeline_cancel(context);
       return thrd_error;
     }
   }
diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c
index 08fb1f2..49de752 100644
--- a/src/shared/multiprocessing.c
+++ b/src/shared/multiprocessing.c
@@ -183,6 +183,15 @@ int write_thread(void* pipeline_context) {
   bool save_to_disk = context->config->save_to_disk;
   char* root_directory = str_dup(context->config->receive_root_directory);
   mtx_unlock(&context->mutex);
+  if (save_to_disk && !root_directory) {
+    mtx_lock(&context->mutex);
+    atomic_store(&context->cancelled, true);
+    context->receiver_done = true;
+    cnd_broadcast(&context->condition_not_full);
+    cnd_broadcast(&context->condition_not_empty);
+    mtx_unlock(&context->mutex);
+    return thrd_error;
+  }
 
   while (true) {
     File* file =