fix: harden queue and receiver error handling

src/shared/multiprocessing.c

@@ -101,7 +101,20 @@ static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver*
   return true;
 }
 
+static void receiver_thread_fail(PipelineContextReceiver* context) {
+  mtx_lock(&context->mutex);
+  context->receiver_done = true;
+  cnd_broadcast(&context->condition_not_empty);
+  cnd_broadcast(&context->condition_not_full);
+  mtx_unlock(&context->mutex);
+}
+
 int receive_thread(void* pipeline_context) {
+#define RECEIVE_THREAD_FAIL()                                                                      \
+  do {                                                                                             \
+    receiver_thread_fail(context);                                                                 \
+    return thrd_error;                                                                             \
+  } while (0)
   PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context;
   if (context->ssl)
     io_set_ssl(context->ssl);
@@ -112,53 +125,52 @@ int receive_thread(void* pipeline_context) {
 
   Status status;
   if (!receive_status(file_descriptor, &status))
-    return thrd_error;
+    RECEIVE_THREAD_FAIL();
   while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK ||
          status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) {
     if (status == STATUS_KEEPALIVE) {
-      send_status(file_descriptor, STATUS_KEEPALIVE);
+      if (!send_status(file_descriptor, STATUS_KEEPALIVE))
+        RECEIVE_THREAD_FAIL();
       goto next;
     }
     if (status == STATUS_ABORT) {
       log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
-      return thrd_error;
+      RECEIVE_THREAD_FAIL();
     }
     if (status == STATUS_CHECK) {
       bool skipped;
       File* file = receive_incremental_check(file_descriptor, config, &skipped);
       if (!skipped) {
         if (file == NULL)
-          return thrd_error;
+          RECEIVE_THREAD_FAIL();
         queue_enqueue_multithreaded(context->queue, file, &context->mutex,
                                     &context->condition_not_empty, &context->condition_not_full);
       }
     } else if (status == STATUS_CHUNK) {
       if (!receive_chunk_enqueue(file_descriptor, context))
-        return thrd_error;
+        RECEIVE_THREAD_FAIL();
     } else if (status == STATUS_CHECK_BATCH) {
       int count;
       if (!receive_int(file_descriptor, &count))
-        return thrd_error;
+        RECEIVE_THREAD_FAIL();
       for (int i = 0; i < count; i++) {
         char* check_path = receive_str(file_descriptor);
         if (!check_path)
-          return thrd_error;
+          RECEIVE_THREAD_FAIL();
         unsigned long long check_size;
         long long check_mtime;
         if (!receive_n_data(file_descriptor, &check_size, sizeof(check_size)) ||
             !receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime))) {
           free(check_path);
-          return thrd_error;
+          RECEIVE_THREAD_FAIL();
         }
         char* full_path = path_cat(config->receive_root_directory, check_path);
         struct stat st;
         bool has_old = full_path && lstat(full_path, &st) == 0;
         bool match = has_old && (unsigned long long)st.st_size == check_size &&
                      (long long)st.st_mtime == check_mtime;
-        if (match)
+        if (!send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT))
-          send_status(file_descriptor, STATUS_OK);
+          RECEIVE_THREAD_FAIL();
-        else
-          send_status(file_descriptor, STATUS_NEXT);
         free(full_path);
         free(check_path);
       }
@@ -170,21 +182,24 @@ int receive_thread(void* pipeline_context) {
                                     &context->condition_not_empty, &context->condition_not_full);
       } else {
         log_message(LOG_LEVEL_ERROR, "Failed to receive file");
-        return thrd_error;
+        RECEIVE_THREAD_FAIL();
       }
     }
   next:
     if (!receive_status(file_descriptor, &status))
-      return thrd_error;
+      RECEIVE_THREAD_FAIL();
   }
   if (status == STATUS_MANIFEST) {
     if (receive_manifest(file_descriptor, config, &status) != 0)
-      return thrd_error;
+      RECEIVE_THREAD_FAIL();
   }
+  if (status != STATUS_FINISHED)
+    RECEIVE_THREAD_FAIL();
   mtx_lock(&context->mutex);
   context->receiver_done = true;
   cnd_signal(&context->condition_not_empty);
   mtx_unlock(&context->mutex);
+#undef RECEIVE_THREAD_FAIL
   return thrd_success;
 }
 

src/shared/queue.c

@@ -1,4 +1,5 @@
 #include <stdbool.h>
+#include <limits.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -7,6 +8,9 @@
 #include "queue.h"
 
 Queue* queue_create(int capacity, void (*destroyer)(void* item)) {
+  if (capacity <= 0)
+    return NULL;
+
   Queue* queue = (Queue*)malloc(sizeof(Queue));
   if (queue == NULL) {
     perror("ERROR: Could not allocate memory for queue structure");
@@ -61,7 +65,9 @@ bool queue_is_full(const Queue* queue) {
 static bool queue_double_capacity(Queue* queue) {
   if (queue == NULL)
     return false;
-  unsigned int new_capacity = queue->capacity * 2;
+  if (queue->capacity > INT_MAX / 2)
+    return false;
+  int new_capacity = queue->capacity * 2;
   if (new_capacity <= 1)
     new_capacity = 100;
   void** new_items = malloc(new_capacity * sizeof(void*));
@@ -103,6 +109,22 @@ bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t*
   return ok;
 }
 
+bool queue_enqueue_multithreaded_cancel(Queue* queue, void* item, mtx_t* mutex,
+                                        cnd_t* condition_not_empty, cnd_t* condition_not_full,
+                                        const bool* cancelled) {
+  mtx_lock(mutex);
+  while (queue_is_full(queue) && (cancelled == NULL || !*cancelled))
+    cnd_wait(condition_not_full, mutex);
+  if (cancelled != NULL && *cancelled) {
+    mtx_unlock(mutex);
+    return false;
+  }
+  bool ok = queue_enqueue(queue, item);
+  cnd_signal(condition_not_empty);
+  mtx_unlock(mutex);
+  return ok;
+}
+
 void* queue_dequeue(Queue* queue) {
   if (queue == NULL || queue_is_empty(queue)) {
     perror("ERROR: Could not dequeue from null or empty queue.");

tests/test_multiprocessing.c

@@ -82,7 +82,7 @@ static void test_sender_queue_capacities() {
   pipeline_context_sender_destroy(ctx);
 }
 
-/* Test that create handles zero-capacity queues */
+/* Invalid queue capacities must not create unusable pipeline queues. */
 static void test_sender_zero_capacity() {
   Config* cfg = config_create();
   EXPECT_NOT_NULL(cfg);
@@ -93,11 +93,9 @@ static void test_sender_zero_capacity() {
 
   Queue* q1 = queue_create(0, NULL);
   Queue* q2 = queue_create(0, NULL);
-  PipelineContextSender* ctx = pipeline_context_sender_create(cfg, q1, q2);
+  EXPECT_NULL(q1);
-  EXPECT_NOT_NULL(ctx);
+  EXPECT_NULL(q2);
-  EXPECT_EQ_INT(ctx->queue_scanner->capacity, 0);
+  config_delete(cfg);
-  EXPECT_EQ_INT(ctx->queue_loader->capacity, 0);
-  pipeline_context_sender_destroy(ctx);
 }
 
 /* Test receiver with zero file_descriptor */
@@ -168,6 +166,41 @@ static void test_receive_thread_finished() {
   }
 }
 
+/* A malformed terminal status must wake a writer waiting on an empty queue. */
+static void test_receive_thread_failure_wakes_writer() {
+  Config* cfg = config_create();
+  EXPECT_NOT_NULL(cfg);
+  free(cfg->version);
+  cfg->version = str_dup(PROTOCOL_VERSION);
+  cfg->send_directory = str_dup("/src");
+  cfg->receive_root_directory = str_dup("/tmp/dst");
+
+  int p[2];
+  EXPECT_EQ_INT(pipe(p), 0);
+  Queue* q = queue_create(1, file_destroy);
+  EXPECT_NOT_NULL(q);
+  PipelineContextReceiver* ctx = pipeline_context_receiver_create(cfg, q, p[0], NULL);
+  EXPECT_NOT_NULL(ctx);
+
+  thrd_t receiver;
+  thrd_t writer;
+  EXPECT_EQ_INT(thrd_create(&writer, write_thread, ctx), thrd_success);
+  EXPECT_EQ_INT(thrd_create(&receiver, receive_thread, ctx), thrd_success);
+  EXPECT_TRUE(send_status(p[1], STATUS_OK));
+  close(p[1]);
+
+  int receiver_result;
+  int writer_result;
+  EXPECT_EQ_INT(thrd_join(receiver, &receiver_result), thrd_success);
+  EXPECT_EQ_INT(thrd_join(writer, &writer_result), thrd_success);
+  EXPECT_EQ_INT(receiver_result, thrd_error);
+  EXPECT_EQ_INT(writer_result, thrd_success);
+  EXPECT_TRUE(ctx->receiver_done);
+
+  close(p[0]);
+  pipeline_context_receiver_destroy(ctx);
+}
+
 /* Test that write_thread completes cleanly when queue signals done */
 static void test_write_thread_done() {
   Config* cfg = config_create();
@@ -223,6 +256,7 @@ void test_multiprocessing() {
   test_receiver_fd_zero();
   if (!is_running_under_valgrind()) {
     test_receive_thread_finished();
+    test_receive_thread_failure_wakes_writer();
   }
   test_write_thread_done();
 }

tests/test_queue.c

@@ -56,6 +56,11 @@ static void test_queue_basic() {
   queue_destroy(q);
 }
 
+static void test_queue_rejects_invalid_capacity() {
+  EXPECT_NULL(queue_create(0, NULL));
+  EXPECT_NULL(queue_create(-1, NULL));
+}
+
 static void test_queue_resize() {
   Queue* q = queue_create(3, NULL);
   EXPECT_NOT_NULL(q);
@@ -199,6 +204,7 @@ static void test_queue_multithreaded() {
 
 void test_queue() {
   test_queue_basic();
+  test_queue_rejects_invalid_capacity();
   test_queue_resize();
   test_queue_destroyer();
   test_queue_multithreaded();