fix: address PR #200 review issues

- Restore PROTOCOL_VERSION to a forward-compatible 2.1.0 and document wire format
- Add NULL guard to config_delete
- Add pipeline cancellation flag and cancellation-aware queue helper
- Join running threads before destroying pipeline contexts on creation failure
- Fix NULL dereference and memory leaks in manifest/chunk handling
- Fix TLS/TCP socket fd leak on connect error paths
- Add compression-level range validation (1-22)
- Close previous log file before opening a new one
- Use getline for unbounded pattern-file lines
- Fix thread-unsafe localtime() and add log level bounds check
- Fix file_load_data to clean up data on read size mismatch
- Add hard ceiling to decompression buffer growth
- Fix mkdir_r bounds check and restore glob comments
- Add send_str NULL guard and mutex-protect bandwidth limiter
- Update AGENTS.md for per-thread io_ssl contract

src/client/client_cli.c

@@ -276,13 +276,16 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
       if (set_nonneg_int_option(&config->max_depth, argv[++i], "--max-depth") != 0)
         return -1;
     } else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) {
+      if (config->log_file) {
+        fclose(config->log_file);
+        config->log_file = NULL;
+        log_set_file(NULL);
+      }
       FILE* lf = fopen(argv[++i], "a");
       if (!lf) {
         fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno));
         return -1;
       }
-      if (config->log_file)
-        fclose(config->log_file);
       config->log_file = lf;
       log_set_file(lf);
     } else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) {
@@ -427,6 +430,10 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "--compress-level") == 0 && i + 1 < argc) {
       if (set_positive_int_option(&config->compression_level, argv[++i], "--compress-level") != 0)
         return -1;
+      if (config->compression_level < 1 || config->compression_level > 22) {
+        fprintf(stderr, "Error: --compress-level must be between 1 and 22\n");
+        return -1;
+      }
     } else if (argv[i][0] == '-') {
       fprintf(stderr, "Unknown option: %s\n", argv[i]);
       print_usage();
@@ -600,8 +607,10 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     fprintf(stderr, "Error: could not open pattern file '%s': %s\n", filepath, strerror(errno));
     return -1;
   }
-  char line[4096];
-  while (fgets(line, sizeof(line), fp)) {
+  char* line = NULL;
+  size_t line_size = 0;
+  ssize_t n;
+  while ((n = getline(&line, &line_size, fp)) != -1) {
     char* p = line;
     while (*p == ' ' || *p == '\t')
       p++;
@@ -615,6 +624,7 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     char** tmp = realloc(*patterns, (*count + 1) * sizeof(char*));
     if (!tmp) {
       fprintf(stderr, "Error: memory allocation failed for pattern file\n");
+      free(line);
       fclose(fp);
       return -1;
     }
@@ -622,11 +632,13 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     char* dup = str_dup(p);
     if (!dup) {
       fprintf(stderr, "Error: memory allocation failed for pattern file\n");
+      free(line);
       fclose(fp);
       return -1;
     }
     (*patterns)[(*count)++] = dup;
   }
+  free(line);
   fclose(fp);
   return 0;
 }

src/client/client_send.c

@@ -346,6 +346,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
     }
     if (send_chunk(client, current_chunk, context->config) != 0) {
       fprintf(stderr, "Error: unexpected error while sending chunk\n");
+      chunk_destroy(current_chunk);
       client_disconnect(client);
       client_delete(client);
       mtx_lock(&context->mutex_progress);
@@ -385,13 +386,31 @@ static int scan_directory_multithreaded(void* pipeline_context) {
         const char* p = current_chunk->items[i]->path;
         if (*p == '/')
           p++;
-        array_list_add(context->manifest, str_dup(p));
+        char* manifest_entry = str_dup(p);
+        if (!manifest_entry) {
+          log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry");
+          mtx_unlock(&context->mutex_scanner);
+          context->cancelled = true;
+          cnd_broadcast(&context->condition_not_full_scanner);
+          cnd_broadcast(&context->condition_not_empty_scanner);
+          parallel_scanner_destroy(scanner);
+          return thrd_error;
+        }
+        array_list_add(context->manifest, manifest_entry);
       }
       mtx_unlock(&context->mutex_scanner);
     }
-    queue_enqueue_multithreaded(context->queue_scanner, current_chunk, &context->mutex_scanner,
-                                &context->condition_not_empty_scanner,
-                                &context->condition_not_full_scanner);
+    if (!queue_enqueue_multithreaded_cancel(
+            context->queue_scanner, current_chunk, &context->mutex_scanner,
+            &context->condition_not_empty_scanner, &context->condition_not_full_scanner,
+            &context->cancelled)) {
+      chunk_destroy(current_chunk);
+      context->cancelled = true;
+      cnd_broadcast(&context->condition_not_full_scanner);
+      cnd_broadcast(&context->condition_not_empty_scanner);
+      parallel_scanner_destroy(scanner);
+      return thrd_error;
+    }
   }
   mtx_lock(&context->mutex_scanner);
   context->scanner_done = true;
@@ -427,9 +446,16 @@ static int load_files_multithreaded(void* pipeline_context) {
         }
       }
     }
-    queue_enqueue_multithreaded(context->queue_loader, chunk, &context->mutex_loader,
-                                &context->condition_not_empty_loader,
-                                &context->condition_not_full_loader);
+    if (!queue_enqueue_multithreaded_cancel(context->queue_loader, chunk, &context->mutex_loader,
+                                            &context->condition_not_empty_loader,
+                                            &context->condition_not_full_loader,
+                                            &context->cancelled)) {
+      chunk_destroy(chunk);
+      context->cancelled = true;
+      cnd_broadcast(&context->condition_not_full_loader);
+      cnd_broadcast(&context->condition_not_empty_loader);
+      return thrd_error;
+    }
   }
 }
 
@@ -487,16 +513,20 @@ int send_files(Config* config) {
     client = client_create();
     if (!client || !client_connect_tls(client, config->server_host, config->server_port,
                                        config->tls_cert, config->tls_key, config->tls_ca)) {
-      if (client)
+      if (client) {
+        client_disconnect(client);
         client_delete(client);
+      }
       fprintf(stderr, "Error: could not connect to server via TLS\n");
       return 1;
     }
   } else {
     client = client_create();
     if (!client || !client_connect(client, config->server_host, config->server_port)) {
-      if (client)
+      if (client) {
+        client_disconnect(client);
         client_delete(client);
+      }
       fprintf(stderr, "Error: could not connect to server\n");
       return 1;
     }
@@ -526,7 +556,17 @@ int send_files(Config* config) {
         const char* p = current_chunk->items[i]->path;
         if (*p == '/')
           p++;
-        array_list_add(manifest, str_dup(p));
+        char* manifest_entry = str_dup(p);
+        if (!manifest_entry) {
+          log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry");
+          chunk_destroy(current_chunk);
+          array_list_delete(manifest);
+          directory_scanner_destroy(scanner);
+          client_disconnect(client);
+          client_delete(client);
+          return 1;
+        }
+        array_list_add(manifest, manifest_entry);
       }
     }
     if (!config->use_sendfile) {
@@ -626,16 +666,40 @@ int send_files_multithreaded(Config* config) {
     context->manifest = array_list_create(free);
 
   thrd_t scanner, loader, sender, progress;
-  if (thrd_create(&scanner, scan_directory_multithreaded, context) != thrd_success ||
-      thrd_create(&loader, load_files_multithreaded, context) != thrd_success ||
-      thrd_create(&sender, send_chunks_multithreaded, context) != thrd_success) {
+  bool scanner_created = false;
+  bool loader_created = false;
+  bool sender_created = false;
+  bool progress_created = false;
+
+  scanner_created = (thrd_create(&scanner, scan_directory_multithreaded, context) == thrd_success);
+  if (scanner_created)
+    loader_created = (thrd_create(&loader, load_files_multithreaded, context) == thrd_success);
+  if (scanner_created && loader_created)
+    sender_created = (thrd_create(&sender, send_chunks_multithreaded, context) == thrd_success);
+
+  if (!scanner_created || !loader_created || !sender_created) {
     perror("Error creating threads.\n");
+    context->cancelled = true;
+    context->scanner_done = true;
+    context->loader_done = true;
+    context->sender_done = true;
+    cnd_broadcast(&context->condition_not_full_scanner);
+    cnd_broadcast(&context->condition_not_empty_scanner);
+    cnd_broadcast(&context->condition_not_full_loader);
+    cnd_broadcast(&context->condition_not_empty_loader);
+    if (sender_created)
+      thrd_join(sender, NULL);
+    if (loader_created)
+      thrd_join(loader, NULL);
+    if (scanner_created)
+      thrd_join(scanner, NULL);
     pipeline_context_sender_destroy(context);
     return 1;
   }
 
   if (config->show_progress) {
-    if (thrd_create(&progress, progress_thread_fn, context) != thrd_success) {
+    progress_created = (thrd_create(&progress, progress_thread_fn, context) == thrd_success);
+    if (!progress_created) {
       perror("Error creating progress thread.\n");
       /* Non-fatal; continue without progress reporting */
     }

src/client/scanner.c

@@ -27,10 +27,14 @@ static void dir_entry_destroy(void* item) {
 
 static DirEntry* dir_entry_create(const char* path, int depth) {
   DirEntry* de = malloc(sizeof(DirEntry));
-  if (de) {
-    de->path = str_dup(path);
-    de->depth = depth;
+  if (!de)
+    return NULL;
+  de->path = str_dup(path);
+  if (!de->path) {
+    free(de);
+    return NULL;
   }
+  de->depth = depth;
   return de;
 }
 
@@ -62,7 +66,18 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_
   scanner->safe_links = safe_links;
   scanner->copy_unsafe_links = copy_unsafe_links;
   scanner->checksum = checksum;
-  queue_enqueue(scanner->directories, dir_entry_create(root_directory, 0));
+  DirEntry* root = dir_entry_create(root_directory, 0);
+  if (!root) {
+    queue_destroy(scanner->directories);
+    free(scanner);
+    return NULL;
+  }
+  if (!queue_enqueue(scanner->directories, root)) {
+    dir_entry_destroy(root);
+    queue_destroy(scanner->directories);
+    free(scanner);
+    return NULL;
+  }
   return scanner;
 }
 
@@ -323,9 +338,27 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata
     free(ps);
     return NULL;
   }
-  if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success ||
-      cnd_init(&ps->result_not_empty) != thrd_success ||
-      cnd_init(&ps->result_not_full) != thrd_success) {
+  int init = 0;
+  bool ok = true;
+  if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success)
+    ok = false;
+  if (ok) {
+    init++;
+    if (cnd_init(&ps->result_not_empty) != thrd_success)
+      ok = false;
+  }
+  if (ok) {
+    init++;
+    if (cnd_init(&ps->result_not_full) != thrd_success)
+      ok = false;
+  }
+  if (!ok) {
+    if (init >= 3)
+      cnd_destroy(&ps->result_not_full);
+    if (init >= 2)
+      cnd_destroy(&ps->result_not_empty);
+    if (init >= 1)
+      mtx_destroy(&ps->result_mutex);
     queue_destroy(ps->result_queue);
     free(ps);
     return NULL;
@@ -564,6 +597,7 @@ void parallel_scanner_destroy(ParallelScanner* ps) {
     return;
   ps->done = true;
   cnd_signal(&ps->result_not_empty);
+  cnd_broadcast(&ps->result_not_full);
   for (int i = 0; i < ps->num_threads; i++)
     thrd_join(ps->threads[i], NULL);
   free(ps->threads);