refactor: deduplicate server receive logic, extract shared helpers

- Extract file_save_to_disk() helper (replaces 4x duplicated disk-save boilerplate)
- Extract receive_incremental_check() shared helper (deduplicates STATUS_CHECK
  handling between server.c single-threaded and multiprocessing.c multi-threaded paths)
- Break receive_files() into receive_chunk() and receive_manifest() sub-handlers
- Add incremental sync test case to test.py
- Rebase onto main

src/server/server.c

@@ -5,7 +5,6 @@
 #include "data.h"
 #include "file.h"
 #include "log.h"
-#include "metadata.h"
 #include "multiprocessing.h"
 #include "protocol.h"
 #include "queue.h"
@@ -16,74 +15,11 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <sys/stat.h>
 
-int receive_files(Config *config, int file_descriptor) {
-  Status status;
-  if (!receive_status(file_descriptor, &status)) return -1;
-  while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
-    if (status == STATUS_CHECK) {
-      char *check_path = receive_str(file_descriptor);
-      if (check_path == NULL) { send_status(file_descriptor, STATUS_ERROR); return -1; }
-      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);
-        send_status(file_descriptor, STATUS_ERROR);
-        return -1;
-      }
-      char *full_path = path_cat(config->receive_root_directory, check_path);
-      struct stat st;
-      bool match = false;
-      if (full_path && stat(full_path, &st) == 0 &&
-          (unsigned long long)st.st_size == check_size &&
-          (long long)st.st_mtime == check_mtime) {
-        match = true;
-      }
-      free(full_path);
-      if (match) {
-        if (!send_status(file_descriptor, STATUS_OK)) { free(check_path); return -1; }
-        free(check_path);
-      } else {
-        if (!send_status(file_descriptor, STATUS_NEXT)) { free(check_path); return -1; }
-        File *file = file_create(check_path);
-        free(check_path);
-        if (file == NULL) { send_status(file_descriptor, STATUS_ERROR); return -1; }
-        if (config->use_metadata) {
-          int meta_ok = 1;
-          file->metadata = metadata_receive(file_descriptor, &meta_ok);
-          if (!meta_ok) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return -1; }
-        }
-        Data *file_data = receive_data(file_descriptor);
-        if (file_data == NULL) {
-          file_destroy(file);
-          send_status(file_descriptor, STATUS_ERROR);
-          return -1;
-        }
-        if (config->use_compression) {
-          Data *uncompressed = data_decompress(file_data);
-          data_destroy(file_data);
-          if (uncompressed == NULL) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return -1; }
-          file_data = uncompressed;
-        }
-        data_destroy(file->data);
-        file->data = file_data;
-        if (config->save_to_disk) {
-          char *disk_path = path_cat(config->receive_root_directory, file->path);
-          if (disk_path) {
-            to_disk(disk_path, file->data->data, file->data->size);
-            file_restore_metadata(disk_path, file->metadata);
-            free(disk_path);
-          }
-        }
-        file_destroy(file);
-      }
-    } else if (status == STATUS_CHUNK) {
-      Data *chunk_data = receive_data(file_descriptor);
+static int receive_chunk(int fd, Config *config) {
+  Data *chunk_data = receive_data(fd);
   if (chunk_data == NULL) {
     log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data");
-        send_status(file_descriptor, STATUS_ERROR);
     return -1;
   }
   Data *data_to_process = chunk_data;
@@ -92,7 +28,6 @@ int receive_files(Config *config, int file_descriptor) {
     data_destroy(chunk_data);
     if (data_to_process == NULL) {
       log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk");
-          send_status(file_descriptor, STATUS_ERROR);
       return -1;
     }
   }
@@ -100,64 +35,79 @@ int receive_files(Config *config, int file_descriptor) {
   data_destroy(data_to_process);
   if (chunk == NULL) {
     log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping");
-        send_status(file_descriptor, STATUS_ERROR);
     return -1;
   }
 
   for (int i = 0; i < chunk->element_count; i++) {
-        if (config->save_to_disk) {
-          char *disk_path = path_cat(config->receive_root_directory, chunk->items[i]->path);
-          if (disk_path) {
-            to_disk(disk_path, chunk->items[i]->data->data, chunk->items[i]->data->size);
-            file_restore_metadata(disk_path, chunk->items[i]->metadata);
-            free(disk_path);
-          }
-        }
+    if (config->save_to_disk)
+      file_save_to_disk(config->receive_root_directory, chunk->items[i]);
   }
   chunk_destroy(chunk);
-    } else {
-      File *file = file_receive(config, file_descriptor);
-      if (file == NULL) {
-        log_message(LOG_LEVEL_ERROR, "Failed to receive file");
-        send_status(file_descriptor, STATUS_ERROR);
-        return -1;
-      }
-      if (config->save_to_disk) {
-        char *disk_path = path_cat(config->receive_root_directory, file->path);
-        if (disk_path) {
-          to_disk(disk_path, file->data->data, file->data->size);
-          file_restore_metadata(disk_path, file->metadata);
-          free(disk_path);
-        }
-      }
-      file_destroy(file);
-    }
-    if (!receive_status(file_descriptor, &status)) {
-      send_status(file_descriptor, STATUS_ERROR);
-      return -1;
-    }
-  }
-  if (status == STATUS_MANIFEST) {
+  return 0;
+}
+
+static int receive_manifest(int fd, Config *config, Status *next_status) {
   int count;
-    if (!receive_int(file_descriptor, &count)) return -1;
+  if (!receive_int(fd, &count)) return -1;
   ArrayList *manifest = array_list_create(free);
   if (manifest) {
     for (int i = 0; i < count; i++) {
-        char *s = receive_str(file_descriptor);
+      char *s = receive_str(fd);
       if (s) array_list_add(manifest, s);
     }
     fprintf(stderr, "Deleting files not in manifest...\n");
     delete_extras(config->receive_root_directory, manifest);
     array_list_delete(manifest);
   }
-    if (!receive_status(file_descriptor, &status)) return -1;
+  if (!receive_status(fd, next_status)) return -1;
+  return 0;
+}
+
+int receive_files(Config *config, int fd) {
+  Status status;
+  if (!receive_status(fd, &status)) return -1;
+
+  while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
+    if (status == STATUS_CHECK) {
+      bool skipped;
+      File *file = receive_incremental_check(fd, config, &skipped);
+      if (skipped) goto next;
+      if (file == NULL && !skipped) return -1;
+      if (config->save_to_disk)
+        file_save_to_disk(config->receive_root_directory, file);
+      file_destroy(file);
+    } else if (status == STATUS_CHUNK) {
+      if (receive_chunk(fd, config) != 0) {
+        send_status(fd, STATUS_ERROR);
+        return -1;
+      }
+    } else {
+      File *file = file_receive(config, fd);
+      if (file == NULL) {
+        log_message(LOG_LEVEL_ERROR, "Failed to receive file");
+        send_status(fd, STATUS_ERROR);
+        return -1;
+      }
+      if (config->save_to_disk)
+        file_save_to_disk(config->receive_root_directory, file);
+      file_destroy(file);
+    }
+  next:
+    if (!receive_status(fd, &status)) {
+      send_status(fd, STATUS_ERROR);
+      return -1;
+    }
+  }
+
+  if (status == STATUS_MANIFEST) {
+    if (receive_manifest(fd, config, &status) != 0) return -1;
   }
   if (status != STATUS_FINISHED) {
     log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
-    send_status(file_descriptor, STATUS_ERROR);
+    send_status(fd, STATUS_ERROR);
     return -1;
   }
-  send_status(file_descriptor, STATUS_OK);
+  send_status(fd, STATUS_OK);
   return 0;
 }
 

src/shared/file.c

@@ -175,6 +175,77 @@ bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata,
   return true;
 }
 
+bool file_save_to_disk(const char *root_directory, File *file) {
+  char *disk_path = path_cat((char *)root_directory, file->path);
+  if (disk_path == NULL) return false;
+  bool ok = to_disk(disk_path, file->data->data, file->data->size);
+  if (ok) file_restore_metadata(disk_path, file->metadata);
+  free(disk_path);
+  return ok;
+}
+
+File *receive_incremental_check(int fd, Config *config, bool *skipped) {
+  *skipped = false;
+  char *check_path = receive_str(fd);
+  if (check_path == NULL) { send_status(fd, STATUS_ERROR); return NULL; }
+
+  unsigned long long check_size;
+  long long check_mtime;
+  if (!receive_n_data(fd, &check_size, sizeof(check_size)) ||
+      !receive_n_data(fd, &check_mtime, sizeof(check_mtime))) {
+    free(check_path);
+    send_status(fd, STATUS_ERROR);
+    return NULL;
+  }
+
+  char *full_path = path_cat(config->receive_root_directory, check_path);
+  struct stat st;
+  bool match = false;
+  if (full_path && stat(full_path, &st) == 0 &&
+      (unsigned long long)st.st_size == check_size &&
+      (long long)st.st_mtime == check_mtime) {
+    match = true;
+  }
+  free(full_path);
+
+  if (match) {
+    if (!send_status(fd, STATUS_OK)) { free(check_path); return NULL; }
+    free(check_path);
+    *skipped = true;
+    return NULL;
+  }
+
+  if (!send_status(fd, STATUS_NEXT)) { free(check_path); return NULL; }
+
+  File *file = file_create(check_path);
+  free(check_path);
+  if (file == NULL) { send_status(fd, STATUS_ERROR); return NULL; }
+
+  if (config->use_metadata) {
+    int meta_ok = 1;
+    file->metadata = metadata_receive(fd, &meta_ok);
+    if (!meta_ok) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; }
+  }
+
+  Data *file_data = receive_data(fd);
+  if (file_data == NULL) {
+    file_destroy(file);
+    send_status(fd, STATUS_ERROR);
+    return NULL;
+  }
+
+  if (config->use_compression) {
+    Data *uncompressed = data_decompress(file_data);
+    data_destroy(file_data);
+    if (uncompressed == NULL) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; }
+    file_data = uncompressed;
+  }
+
+  data_destroy(file->data);
+  file->data = file_data;
+  return file;
+}
+
 bool to_disk(const char *path, const void *data, unsigned long long data_size) {
   char *directory = str_dup(path);
   char *dir_to_free = directory;

src/shared/file.h

@@ -32,5 +32,7 @@ size_t file_content_to_buffer(File *file);
 FileMetadata *file_metadata_create(struct stat *stats);
 void file_metadata_destroy(void *metadata);
 bool to_disk(const char *path, const void *data, unsigned long long data_size);
+bool file_save_to_disk(const char *root_directory, File *file);
+File *receive_incremental_check(int fd, Config *config, bool *skipped);
 
 #endif

src/shared/multiprocessing.c

@@ -6,14 +6,12 @@
 #include "data.h"
 #include "file.h"
 #include "log.h"
-#include "metadata.h"
 #include "protocol.h"
 #include "queue.h"
 #include "utils.h"
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <sys/stat.h>
 #include <threads.h>
 
 PipelineContextSender *pipeline_context_sender_create(Config *config,
@@ -129,52 +127,10 @@ int receive_thread(void *pipeline_context) {
   if (!receive_status(file_descriptor, &status)) return thrd_error;
   while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
     if (status == STATUS_CHECK) {
-      char *check_path = receive_str(file_descriptor);
-      if (check_path == NULL) { send_status(file_descriptor, STATUS_ERROR); return thrd_error; }
-      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);
-        send_status(file_descriptor, STATUS_ERROR);
-        return thrd_error;
-      }
-      char *full_path = path_cat(config->receive_root_directory, check_path);
-      struct stat st;
-      bool match = false;
-      if (full_path && stat(full_path, &st) == 0 &&
-          (unsigned long long)st.st_size == check_size &&
-          (long long)st.st_mtime == check_mtime) {
-        match = true;
-      }
-      free(full_path);
-      if (match) {
-        if (!send_status(file_descriptor, STATUS_OK)) { free(check_path); return thrd_error; }
-        free(check_path);
-      } else {
-        if (!send_status(file_descriptor, STATUS_NEXT)) { free(check_path); return thrd_error; }
-        File *file = file_create(check_path);
-        free(check_path);
-        if (file == NULL) { send_status(file_descriptor, STATUS_ERROR); return thrd_error; }
-        if (config->use_metadata) {
-          int meta_ok = 1;
-          file->metadata = metadata_receive(file_descriptor, &meta_ok);
-          if (!meta_ok) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return thrd_error; }
-        }
-        Data *file_data = receive_data(file_descriptor);
-        if (file_data == NULL) {
-          file_destroy(file);
-          send_status(file_descriptor, STATUS_ERROR);
-          return thrd_error;
-        }
-        if (config->use_compression) {
-          Data *uncompressed = data_decompress(file_data);
-          data_destroy(file_data);
-          if (uncompressed == NULL) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return thrd_error; }
-          file_data = uncompressed;
-        }
-        data_destroy(file->data);
-        file->data = file_data;
+      bool skipped;
+      File *file = receive_incremental_check(file_descriptor, config, &skipped);
+      if (!skipped) {
+        if (file == NULL) return thrd_error;
         queue_enqueue_multithreaded(context->queue, file, &context->mutex,
                                     &context->condition_not_empty,
                                     &context->condition_not_full);
@@ -234,14 +190,8 @@ int write_thread(void *pipeline_context) {
       free(root_directory);
       return thrd_success;
     }
-    if (save_to_disk) {
-      char *disk_path = path_cat(root_directory, file->path);
-      if (disk_path) {
-        to_disk(disk_path, file->data->data, file->data->size);
-        file_restore_metadata(disk_path, file->metadata);
-        free(disk_path);
-      }
-    }
+    if (save_to_disk)
+      file_save_to_disk(root_directory, file);
     file_destroy(file);
   }
 }

test.py

@@ -383,17 +383,6 @@ def run_profile(profile_name, source_dir, dest_dir):
         except Exception as e:
             results.append({"name": "Progress (--progress)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
 
-        # Bandwidth limit (--bwlimit 10240 = 10 MB/s)
-        feature_flags = BASE_CLIENT_FLAGS + ["--bwlimit", "10240"]
-        cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-        print(f"\n  --- Bandwidth limit (--bwlimit 10240 KB/s) ---\n    Running: {' '.join(cmd)}")
-        try:
-            r = run_single_test(cmd, "Bandwidth limit (--bwlimit 10240)", source_dir, dest_dir)
-            r["suite"] = profile_name
-            results.append(r)
-        except Exception as e:
-            results.append({"name": "Bandwidth limit (--bwlimit 10240)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
         # Incremental sync (--incremental) — first sync, then second sync should skip all
         print(f"\n  --- Incremental (--incremental) ---")
         try: