fix: close remaining PR review gaps

src/client/client_cli.c

@@ -459,7 +459,6 @@ int main(int argc, char* argv[]) {
   parse_environment(&env_source, &env_dest, &save_to_disk);
 
   int exit_code = 0;
-  bool config_owned_by_pipeline = false;
   Config* config = config_create();
   if (!config) {
     fprintf(stderr, "Error: failed to allocate config\n");
@@ -542,18 +541,14 @@ int main(int argc, char* argv[]) {
 
   /* Execute transfer */
   if (config->use_multithreading) {
-    config_owned_by_pipeline = true;
-    exit_code = send_files_multithreaded(config);
+    exit_code = send_files_multithreaded(&config);
   } else {
     exit_code = send_files(config);
   }
 
 cleanup:
   if (config) {
-    if (config->log_file)
-      fclose(config->log_file);
-    if (!config_owned_by_pipeline)
-      config_delete(config);
+    config_delete(config);
   }
   return exit_code;
 }

src/client/client_send.c

@@ -70,6 +70,8 @@ static int send_dry_run_manifest(Config* config) {
 
 /* Send the delete manifest (list of files) to the server. Returns 0 on success, -1 on failure. */
 static int send_delete_manifest(int fd, ArrayList* manifest) {
+  if (!manifest)
+    return -1;
   if (!send_status(fd, STATUS_MANIFEST))
     return -1;
   if (!send_int(fd, manifest->size))
@@ -111,17 +113,22 @@ static int incremental_check(Client* client, File* file, const Config* config,
     return 1;
   if (s == STATUS_DELTA_SIGNATURE) {
     Data* sig_data = receive_data(client->file_descriptor);
-    if (!sig_data)
+    if (!sig_data) {
+      send_status(client->file_descriptor, STATUS_ERROR);
       return -1;
+    }
     DeltaSignature* sig = delta_signature_deserialize(sig_data);
     data_destroy(sig_data);
-    if (!sig)
+    if (!sig) {
+      send_status(client->file_descriptor, STATUS_ERROR);
       return -1;
+    }
     *out_sig = sig;
     return 2;
   }
   if (s != STATUS_NEXT) {
     log_message(LOG_LEVEL_ERROR, "Unexpected server status");
+    send_status(client->file_descriptor, STATUS_ERROR);
     return -1;
   }
   return 0;
@@ -129,8 +136,10 @@ static int incremental_check(Client* client, File* file, const Config* config,
 
 static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* config) {
   Delta* delta = delta_compute(file->data->data, file->data->size, sig, config->delta_block_size);
+  /* The receiver is blocked after sending the signature.  Every local
+     fallback therefore needs the explicit NEXT response before full data. */
   if (!delta)
-    return 1;
+    return send_status(client->file_descriptor, STATUS_NEXT) ? 1 : -1;
 
   if (!delta_is_worthwhile(delta, file->data->size)) {
     delta_destroy(delta);
@@ -142,14 +151,14 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c
   Data* delta_data = delta_serialize(delta);
   delta_destroy(delta);
   if (!delta_data)
-    return -1;
+    return send_status(client->file_descriptor, STATUS_NEXT) ? 1 : -1;
 
   Data* to_send = delta_data;
   if (config->use_compression) {
     to_send = data_compress(delta_data, config->compression_level);
     data_destroy(delta_data);
     if (!to_send)
-      return -1;
+      return send_status(client->file_descriptor, STATUS_NEXT) ? 1 : -1;
   }
 
   bool ok = send_status(client->file_descriptor, STATUS_DELTA_DATA) &&
@@ -279,7 +288,8 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) {
     if (f == NULL)
       continue;
     bool stream = f->data->data == NULL && f->data->size > 0;
-    bool use_sendfile = (config->use_sendfile && !config->use_compression) || stream;
+    bool use_sendfile =
+        (config->use_sendfile && !config->use_compression) || (stream && !config->use_compression);
     int rc = send_single_file(client, f, config, config->use_incremental, use_sendfile);
     if (rc == 1)
       continue;
@@ -478,7 +488,7 @@ static int load_files_multithreaded(void* pipeline_context) {
     if (!context->config->use_sendfile) {
       for (int i = 0; i < chunk->element_count; i++) {
         File* f = chunk->items[i];
-        if (f->data->size > STREAM_THRESHOLD)
+        if (f->data->size > STREAM_THRESHOLD && !context->config->use_compression)
           continue;
         if (!file_load_data(f)) {
           log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping");
@@ -630,7 +640,7 @@ int send_files(Config* config) {
     if (!config->use_sendfile) {
       for (int i = 0; i < current_chunk->element_count; i++) {
         File* f = current_chunk->items[i];
-        if (f->data->size > STREAM_THRESHOLD)
+        if (f->data->size > STREAM_THRESHOLD && !config->use_compression)
           continue;
         if (!file_load_data(f)) {
           log_message(LOG_LEVEL_ERROR, "Failed to load file data");
@@ -697,7 +707,10 @@ send_fail:
   return 1;
 }
 
-int send_files_multithreaded(Config* config) {
+int send_files_multithreaded(Config** config_ptr) {
+  if (!config_ptr || !*config_ptr)
+    return 1;
+  Config* config = *config_ptr;
   if (config->dry_run)
     return send_dry_run_manifest(config);
 
@@ -728,8 +741,13 @@ int send_files_multithreaded(Config* config) {
     queue_destroy(q2);
     return 1;
   }
+  *config_ptr = NULL; /* context now owns config through all remaining paths */
   if (config->use_delete)
     context->manifest = array_list_create(free);
+  if (config->use_delete && !context->manifest) {
+    pipeline_context_sender_destroy(context);
+    return 1;
+  }
 
   thrd_t scanner, loader, sender;
   bool scanner_created = false;

src/client/client_send.h

@@ -7,6 +7,7 @@
 
 int send_chunk(Client* client, Chunk* chunk, Config* config);
 int send_files(Config* config);
-int send_files_multithreaded(Config* config);
+/* Takes ownership only when *config is set to NULL on return. */
+int send_files_multithreaded(Config** config);
 
 #endif

src/client/scanner.c

@@ -465,8 +465,10 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata
     if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
       continue;
     char* cur_path = path_cat(root_directory, entry->d_name);
-    if (!cur_path)
+    if (!cur_path) {
+      ps->failed = true;
       continue;
+    }
     struct stat lstats;
     if (lstat(cur_path, &lstats) != 0) {
       free(cur_path);