refactor: consolidate transfer infrastructure

CMakeLists.txt

@@ -8,34 +8,30 @@ set(CMAKE_C_STANDARD_REQUIRED ON)
 
 add_compile_options(-Wall -g -O3)
 
-# --- Sanitizer option ---
 set(SANITIZER "none" CACHE STRING "Sanitizer to enable (address, thread, undefined, none)")
 set_property(CACHE SANITIZER PROPERTY STRINGS address thread undefined none)
-
 if(SANITIZER STREQUAL "address")
-	add_compile_options(-fsanitize=address -fno-omit-frame-pointer -g)
-	add_link_options(-fsanitize=address)
+  add_compile_options(-fsanitize=address -fno-omit-frame-pointer -g)
+  add_link_options(-fsanitize=address)
 elseif(SANITIZER STREQUAL "thread")
-	add_compile_options(-fsanitize=thread -fno-omit-frame-pointer -g)
-	add_link_options(-fsanitize=thread)
+  add_compile_options(-fsanitize=thread -fno-omit-frame-pointer -g)
+  add_link_options(-fsanitize=thread)
 elseif(SANITIZER STREQUAL "undefined")
-	add_compile_options(-fsanitize=undefined -fno-omit-frame-pointer -g)
-	add_link_options(-fsanitize=undefined)
+  add_compile_options(-fsanitize=undefined -fno-omit-frame-pointer -g)
+  add_link_options(-fsanitize=undefined)
 elseif(NOT SANITIZER STREQUAL "none")
-	message(FATAL_ERROR "Unknown sanitizer: ${SANITIZER}. Supported values: address, thread, undefined, none")
+  message(FATAL_ERROR "Unknown sanitizer: ${SANITIZER}. Supported values: address, thread, undefined, none")
 endif()
 
-# --- Strict warnings option ---
 option(STRICT_WARNINGS "Enable strict warnings (Wextra, Wpedantic, Werror)" OFF)
 if(STRICT_WARNINGS)
-	add_compile_options(-Wextra -Wpedantic -Werror)
+  add_compile_options(-Wextra -Wpedantic -Werror)
 endif()
 
-# --- Coverage option ---
 option(ENABLE_COVERAGE "Enable gcov coverage" OFF)
 if(ENABLE_COVERAGE)
-	add_compile_options(--coverage -fprofile-arcs -ftest-coverage -O0 -g)
-	add_link_options(--coverage)
+  add_compile_options(--coverage -fprofile-arcs -ftest-coverage -O0 -g)
+  add_link_options(--coverage)
 endif()
 
 include(FetchContent)
@@ -49,55 +45,102 @@ FetchContent_MakeAvailable(xxhash)
 
 set(THREADS_PREFER_PTHREAD_FLAG ON)
 find_package(Threads REQUIRED)
-
 find_library(ZSTD_LIBRARY zstd)
 if(NOT ZSTD_LIBRARY)
-	message(FATAL_ERROR "zstd library not found. Ensure it is in your nix-shell!")
+  message(FATAL_ERROR "zstd library not found. Ensure it is in your nix-shell!")
 endif()
-
 find_package(OpenSSL REQUIRED)
 
-file(GLOB SHARED_SRCS "src/shared/*.c")
-file(GLOB SERVER_SRCS "src/server/*.c")
-file(GLOB CLIENT_SRCS "src/client/*.c")
+set(SHARED_SRCS
+  src/shared/array_list.c
+  src/shared/chunk.c
+  src/shared/compression.c
+  src/shared/config.c
+  src/shared/data.c
+  src/shared/delta.c
+  src/shared/file.c
+  src/shared/log.c
+  src/shared/metadata.c
+  src/shared/multiprocessing.c
+  src/shared/protocol.c
+  src/shared/queue.c
+  src/shared/transport_ssh.c
+  src/shared/transport_tcp.c
+  src/shared/transport_tls.c
+  src/shared/utils.c
+)
+set(SERVER_SRCS src/server/server.c)
+set(CLIENT_SRCS
+  src/client/client_cli.c
+  src/client/client_send.c
+  src/client/scanner.c
+  src/client/usage.c
+)
+
+function(configure_fastsync_target target)
+  target_include_directories(${target} PRIVATE src/shared src/server src/client)
+  target_link_libraries(${target} PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash)
+endfunction()
 
-# --- Main executables ---
 add_executable(server ${SERVER_SRCS} ${SHARED_SRCS})
-target_include_directories(server PRIVATE src/shared src/server src/client)
-target_link_libraries(server PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash)
-
+configure_fastsync_target(server)
 add_executable(client ${CLIENT_SRCS} ${SHARED_SRCS})
-target_include_directories(client PRIVATE src/shared src/server src/client)
-target_link_libraries(client PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash)
+configure_fastsync_target(client)
 
-# --- Testing ---
 enable_testing()
-
-# Common test libraries
-set(TEST_LIBS Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash)
-set(TEST_INCLUDES tests src/shared src/server src/client)
-
-# Monolithic test binary (backward compatible)
-file(GLOB TEST_SRCS "tests/test_*.c" "tests/runner.c")
+set(TEST_SRCS
+  tests/runner.c
+  tests/test_array_list.c
+  tests/test_chunk.c
+  tests/test_client_cli.c
+  tests/test_compression.c
+  tests/test_config.c
+  tests/test_data.c
+  tests/test_delta.c
+  tests/test_file.c
+  tests/test_file_sendfile.c
+  tests/test_fuzz_smoke.c
+  tests/test_glob.c
+  tests/test_log.c
+  tests/test_metadata.c
+  tests/test_multiprocessing.c
+  tests/test_property.c
+  tests/test_protocol.c
+  tests/test_queue.c
+  tests/test_robustness.c
+  tests/test_scanner.c
+  tests/test_server.c
+  tests/test_shared_utils.c
+  tests/test_stress.c
+  tests/test_transport_ssh.c
+  tests/test_transport_tcp.c
+  tests/test_transport_tls.c
+)
 add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} src/client/scanner.c src/client/client_cli.c)
-target_include_directories(tests PRIVATE ${TEST_INCLUDES})
+configure_fastsync_target(tests)
+target_include_directories(tests PRIVATE tests)
 target_compile_definitions(tests PRIVATE FASTSYNC_TEST_BUILD)
-target_link_libraries(tests PRIVATE ${TEST_LIBS})
 add_test(NAME unit_all COMMAND tests)
 
-# --- Fuzz targets (requires clang) ---
 option(ENABLE_FUZZ "Build fuzz targets (requires clang)" OFF)
 if(ENABLE_FUZZ)
   if(NOT CMAKE_C_COMPILER_ID MATCHES "Clang")
     message(FATAL_ERROR "ENABLE_FUZZ requires Clang (compiler is ${CMAKE_C_COMPILER_ID})")
   endif()
-  file(GLOB FUZZ_SRCS "tests/fuzz/*.c")
-	foreach(FUZZ_SRC ${FUZZ_SRCS})
-		get_filename_component(FUZZ_NAME ${FUZZ_SRC} NAME_WE)
-		add_executable(${FUZZ_NAME} ${FUZZ_SRC} ${SHARED_SRCS})
-		target_include_directories(${FUZZ_NAME} PRIVATE ${TEST_INCLUDES})
-		target_compile_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer)
-		target_link_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined)
-		target_link_libraries(${FUZZ_NAME} PRIVATE ${TEST_LIBS})
-	endforeach()
+  set(FUZZ_SRCS
+    tests/fuzz/fuzz_chunk_deserialize.c
+    tests/fuzz/fuzz_compress_decompress.c
+    tests/fuzz/fuzz_delta_deserialize.c
+    tests/fuzz/fuzz_delta_signature_deserialize.c
+    tests/fuzz/fuzz_glob_match.c
+    tests/fuzz/fuzz_metadata_from_buf.c
+  )
+  foreach(FUZZ_SRC ${FUZZ_SRCS})
+    get_filename_component(FUZZ_NAME ${FUZZ_SRC} NAME_WE)
+    add_executable(${FUZZ_NAME} ${FUZZ_SRC} ${SHARED_SRCS})
+    configure_fastsync_target(${FUZZ_NAME})
+    target_include_directories(${FUZZ_NAME} PRIVATE tests)
+    target_compile_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer)
+    target_link_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined)
+  endforeach()
 endif()

src/server/server.c

@@ -1,7 +1,4 @@
-#include "array_list.h"
-#include "chunk.h"
 #include "config.h"
-#include "data.h"
 #include "file.h"
 #include "log.h"
 #include "multiprocessing.h"
@@ -29,9 +26,12 @@ static bool path_is_within(const char* root, const char* path) {
   return strncmp(root, path, n) == 0 && (path[n] == '\0' || path[n] == '/');
 }
 
-static bool valid_batch_path(const char* path) {
-  return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path) &&
-         strchr(path, '\0') == path + strlen(path);
+static bool save_received_file(File* file, void* context) {
+  Config* config = context;
+  if (config->save_to_disk && !file_save_to_disk(config->receive_root_directory, file, config))
+    return false;
+  file_destroy(file);
+  return true;
 }
 
 static bool __attribute__((unused)) configure_authorization(const char* root) {
@@ -53,116 +53,7 @@ static bool __attribute__((unused)) configure_authorization(const char* root) {
 }
 
 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 ||
-         status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) {
-    if (status == STATUS_KEEPALIVE) {
-      send_status(fd, STATUS_KEEPALIVE);
-      goto next;
-    }
-    if (status == STATUS_ABORT) {
-      log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
-      return -1;
-    }
-    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, config)) {
-        file_destroy(file);
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      file_destroy(file);
-    } else if (status == STATUS_CHUNK) {
-      Chunk* chunk = receive_chunk_data(fd, config);
-      if (chunk == NULL) {
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      for (int i = 0; i < chunk->element_count; i++) {
-        if (config->save_to_disk &&
-            !file_save_to_disk(config->receive_root_directory, chunk->items[i], config)) {
-          chunk_destroy(chunk);
-          send_status(fd, STATUS_ERROR);
-          return -1;
-        }
-      }
-      chunk_destroy(chunk);
-    } else if (status == STATUS_CHECK_BATCH) {
-      int count;
-      /* Batch framing has no checksum field yet; never silently downgrade a
-         checksum-enabled transfer into mtime-only matching. */
-      if (config->checksum || !receive_int(fd, &count) || count < 0 || count > MAX_MANIFEST_ENTRIES)
-        return -1;
-      for (int i = 0; i < count; i++) {
-        char* check_path = receive_str(fd);
-        if (!check_path)
-          return -1;
-        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);
-          return -1;
-        }
-        if (!valid_batch_path(check_path)) {
-          free(check_path);
-          send_status(fd, STATUS_ERROR);
-          return -1;
-        }
-        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;
-        bool sent = send_status(fd, match ? STATUS_OK : STATUS_NEXT);
-        free(full_path);
-        free(check_path);
-        if (!sent)
-          return -1;
-      }
-      goto next;
-    } 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, config)) {
-        file_destroy(file);
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      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(fd, STATUS_ERROR);
-    return -1;
-  }
-  send_status(fd, STATUS_OK);
-  return 0;
+  return receive_files_common(config, fd, save_received_file, config, true);
 }
 
 void handler(int file_descriptor) {

src/shared/config.c

@@ -173,103 +173,39 @@ void config_delete(Config* config) {
   free(config);
 }
 
-/* Wire format order (must match config_receive and be updated when PROTOCOL_VERSION bumps):
- * version, send_directory, receive_root_directory, save_to_disk, use_multithreading,
- * use_chunk_serialization, use_compression, use_metadata, compression_level, chunk_size,
- * use_sendfile, use_delete, use_incremental, use_delta, delta_block_size, delta_max_file_size,
- * backup, backup_dir, follow_symlinks, copy_links, safe_links, copy_unsafe_links,
- * preserve_hard_links, preserve_acls, preserve_xattrs, preserve_devices, preserve_sparse,
- * update, inplace, append, append_verify, delete_excluded, delete_after, max_delete, relative,
- * prune_empty_dirs, temp_dir, partial, partial_dir, suffix, delete_before, checksum,
- * compress_choice, status
- */
+static bool config_send_string(int file_descriptor, const char* value, bool optional) {
+  return send_str(file_descriptor, value ? value : (optional ? "" : NULL));
+}
+
+#define CONFIG_SEND_VERSION(field)                                                                 \
+  do {                                                                                             \
+    if (!config_send_string(file_descriptor, config->field, false))                                \
+      return false;                                                                                \
+  } while (0);
+#define CONFIG_SEND_STRING(field)                                                                  \
+  do {                                                                                             \
+    if (!config_send_string(file_descriptor, config->field, false))                                \
+      return false;                                                                                \
+  } while (0);
+#define CONFIG_SEND_OPTIONAL_STRING(field)                                                         \
+  do {                                                                                             \
+    if (!config_send_string(file_descriptor, config->field, true))                                 \
+      return false;                                                                                \
+  } while (0);
+#define CONFIG_SEND_INTEGER(field)                                                                 \
+  do {                                                                                             \
+    if (!send_int(file_descriptor, config->field))                                                 \
+      return false;                                                                                \
+  } while (0);
+#define CONFIG_SEND_DATA(field)                                                                    \
+  do {                                                                                             \
+    if (!send_n_data(file_descriptor, &config->field, sizeof(config->field)))                      \
+      return false;                                                                                \
+  } while (0);
+
 bool config_send(int file_descriptor, const Config* config) {
-  if (!send_str(file_descriptor, config->version))
-    return false;
-  if (!send_str(file_descriptor, config->send_directory))
-    return false;
-  if (!send_str(file_descriptor, config->receive_root_directory))
-    return false;
-  if (!send_int(file_descriptor, config->save_to_disk))
-    return false;
-  if (!send_int(file_descriptor, config->use_multithreading))
-    return false;
-  if (!send_int(file_descriptor, config->use_chunk_serialization))
-    return false;
-  if (!send_int(file_descriptor, config->use_compression))
-    return false;
-  if (!send_int(file_descriptor, config->use_metadata))
-    return false;
-  if (!send_int(file_descriptor, config->compression_level))
-    return false;
-  if (!send_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size)))
-    return false;
-  if (!send_int(file_descriptor, config->use_sendfile))
-    return false;
-  if (!send_int(file_descriptor, config->use_delete))
-    return false;
-  if (!send_int(file_descriptor, config->use_incremental))
-    return false;
-  if (!send_int(file_descriptor, config->use_delta))
-    return false;
-  if (!send_n_data(file_descriptor, &config->delta_block_size, sizeof(config->delta_block_size)))
-    return false;
-  if (!send_n_data(file_descriptor, &config->delta_max_file_size, sizeof(unsigned long long)))
-    return false;
-  if (!send_int(file_descriptor, config->backup))
-    return false;
-  if (!send_str(file_descriptor, config->backup_dir ? config->backup_dir : ""))
-    return false;
-  if (!send_int(file_descriptor, config->follow_symlinks))
-    return false;
-  if (!send_int(file_descriptor, config->copy_links))
-    return false;
-  if (!send_int(file_descriptor, config->safe_links))
-    return false;
-  if (!send_int(file_descriptor, config->copy_unsafe_links))
-    return false;
-  if (!send_int(file_descriptor, config->preserve_hard_links))
-    return false;
-  if (!send_int(file_descriptor, config->preserve_acls))
-    return false;
-  if (!send_int(file_descriptor, config->preserve_xattrs))
-    return false;
-  if (!send_int(file_descriptor, config->preserve_devices))
-    return false;
-  if (!send_int(file_descriptor, config->preserve_sparse))
-    return false;
-  if (!send_int(file_descriptor, config->update))
-    return false;
-  if (!send_int(file_descriptor, config->inplace))
-    return false;
-  if (!send_int(file_descriptor, config->append))
-    return false;
-  if (!send_int(file_descriptor, config->append_verify))
-    return false;
-  if (!send_int(file_descriptor, config->delete_excluded))
-    return false;
-  if (!send_int(file_descriptor, config->delete_after))
-    return false;
-  if (!send_n_data(file_descriptor, &config->max_delete, sizeof(config->max_delete)))
-    return false;
-  if (!send_int(file_descriptor, config->relative))
-    return false;
-  if (!send_int(file_descriptor, config->prune_empty_dirs))
-    return false;
-  if (!send_str(file_descriptor, config->temp_dir ? config->temp_dir : ""))
-    return false;
-  if (!send_int(file_descriptor, config->partial))
-    return false;
-  if (!send_str(file_descriptor, config->partial_dir ? config->partial_dir : ""))
-    return false;
-  if (!send_str(file_descriptor, config->suffix ? config->suffix : ""))
-    return false;
-  if (!send_int(file_descriptor, config->delete_before))
-    return false;
-  if (!send_int(file_descriptor, config->checksum))
-    return false;
-  if (!send_str(file_descriptor, config->compress_choice ? config->compress_choice : ""))
-    return false;
+  CONFIG_WIRE_FIELDS(CONFIG_SEND_VERSION, CONFIG_SEND_STRING, CONFIG_SEND_OPTIONAL_STRING,
+                     CONFIG_SEND_INTEGER, CONFIG_SEND_DATA)
   Status status;
   if (!receive_status(file_descriptor, &status))
     return false;
@@ -280,160 +216,60 @@ bool config_send(int file_descriptor, const Config* config) {
   return true;
 }
 
-/* Wire format order: see the comment above config_send. */
+#undef CONFIG_SEND_VERSION
+#undef CONFIG_SEND_STRING
+#undef CONFIG_SEND_OPTIONAL_STRING
+#undef CONFIG_SEND_INTEGER
+#undef CONFIG_SEND_DATA
+
+static bool config_receive_string(int file_descriptor, char** destination) {
+  char* value = receive_str(file_descriptor);
+  if (!value)
+    return false;
+  *destination = value;
+  return true;
+}
+
+#define CONFIG_RECEIVE_VERSION(field)                                                              \
+  do {                                                                                             \
+    free(config->field);                                                                           \
+    config->field = receive_str(file_descriptor);                                                  \
+    if (!config->field)                                                                            \
+      goto error;                                                                                  \
+    if (strcmp(config->field, PROTOCOL_VERSION) != 0) {                                            \
+      fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", config->field,          \
+              PROTOCOL_VERSION);                                                                   \
+      config_delete(config);                                                                       \
+      send_status(file_descriptor, STATUS_ERROR);                                                  \
+      return NULL;                                                                                 \
+    }                                                                                              \
+  } while (0);
+#define CONFIG_RECEIVE_STRING(field)                                                               \
+  do {                                                                                             \
+    if (!config_receive_string(file_descriptor, &config->field))                                   \
+      goto error;                                                                                  \
+  } while (0);
+#define CONFIG_RECEIVE_OPTIONAL_STRING(field) CONFIG_RECEIVE_STRING(field)
+#define CONFIG_RECEIVE_INTEGER(field)                                                              \
+  do {                                                                                             \
+    if (!receive_int(file_descriptor, &tmp))                                                       \
+      goto error;                                                                                  \
+    config->field = tmp;                                                                           \
+  } while (0);
+#define CONFIG_RECEIVE_DATA(field)                                                                 \
+  do {                                                                                             \
+    if (!receive_n_data(file_descriptor, &config->field, sizeof(config->field)))                   \
+      goto error;                                                                                  \
+  } while (0);
+
 Config* config_receive(int file_descriptor) {
   Config* config = (Config*)malloc(sizeof(Config));
   if (config == NULL)
     return NULL;
   config_set_defaults(config);
-  free(config->version);
-  config->version = receive_str(file_descriptor);
-  if (!config->version) {
-    free(config->server_host);
-    free(config);
-    return NULL;
-  }
-  if (strcmp(config->version, PROTOCOL_VERSION) != 0) {
-    fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", config->version,
-            PROTOCOL_VERSION);
-    free(config->version);
-    free(config->server_host);
-    free(config);
-    send_status(file_descriptor, STATUS_ERROR);
-    return NULL;
-  }
-  config->send_directory = receive_str(file_descriptor);
-  if (!config->send_directory) {
-    free(config->version);
-    free(config->server_host);
-    free(config);
-    return NULL;
-  }
-  config->receive_root_directory = receive_str(file_descriptor);
-  if (!config->receive_root_directory) {
-    free(config->version);
-    free(config->send_directory);
-    free(config->server_host);
-    free(config);
-    return NULL;
-  }
   int tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->save_to_disk = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_multithreading = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_chunk_serialization = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_compression = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_metadata = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->compression_level = tmp;
-  if (!receive_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size)))
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_sendfile = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_delete = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_incremental = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->use_delta = tmp;
-  if (!receive_n_data(file_descriptor, &config->delta_block_size, sizeof(config->delta_block_size)))
-    goto error;
-  if (!receive_n_data(file_descriptor, &config->delta_max_file_size, sizeof(unsigned long long)))
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->backup = tmp;
-  config->backup_dir = receive_str(file_descriptor);
-  if (config->backup_dir == NULL)
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->follow_symlinks = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->copy_links = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->safe_links = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->copy_unsafe_links = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->preserve_hard_links = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->preserve_acls = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->preserve_xattrs = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->preserve_devices = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->preserve_sparse = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->update = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->inplace = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->append = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->append_verify = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->delete_excluded = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->delete_after = tmp;
-  if (!receive_n_data(file_descriptor, &config->max_delete, sizeof(config->max_delete)))
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->relative = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->prune_empty_dirs = tmp;
-  config->temp_dir = receive_str(file_descriptor);
-  if (config->temp_dir == NULL)
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->partial = tmp;
-  config->partial_dir = receive_str(file_descriptor);
-  if (config->partial_dir == NULL)
-    goto error;
-  config->suffix = receive_str(file_descriptor);
-  if (config->suffix == NULL)
-    goto error;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->delete_before = tmp;
-  if (!receive_int(file_descriptor, &tmp))
-    goto error;
-  config->checksum = tmp;
-  config->compress_choice = receive_str(file_descriptor);
-  if (config->compress_choice == NULL)
-    goto error;
+  CONFIG_WIRE_FIELDS(CONFIG_RECEIVE_VERSION, CONFIG_RECEIVE_STRING, CONFIG_RECEIVE_OPTIONAL_STRING,
+                     CONFIG_RECEIVE_INTEGER, CONFIG_RECEIVE_DATA)
   if (config->compress_choice[0] != '\0' && strcmp(config->compress_choice, "zstd") != 0 &&
       strcmp(config->compress_choice, "none") != 0) {
     fprintf(stderr, "Unsupported compression choice: %s\n", config->compress_choice);
@@ -445,15 +281,12 @@ Config* config_receive(int file_descriptor) {
   return config;
 
 error:
-  free(config->version);
-  free(config->send_directory);
-  free(config->receive_root_directory);
-  free(config->server_host);
-  free(config->backup_dir);
-  free(config->temp_dir);
-  free(config->partial_dir);
-  free(config->suffix);
-  free(config->compress_choice);
-  free(config);
+  config_delete(config);
   return NULL;
 }
+
+#undef CONFIG_RECEIVE_VERSION
+#undef CONFIG_RECEIVE_STRING
+#undef CONFIG_RECEIVE_OPTIONAL_STRING
+#undef CONFIG_RECEIVE_INTEGER
+#undef CONFIG_RECEIVE_DATA

src/shared/config.h

@@ -128,6 +128,53 @@ typedef struct Config {
   char* compress_choice;
 } Config;
 
+/* Keep the on-wire field order in one place. The first three strings require
+ * values when sent; optional strings are encoded as empty strings when NULL. */
+#define CONFIG_WIRE_FIELDS(VERSION, STRING, OPTIONAL_STRING, INTEGER, DATA)                        \
+  VERSION(version)                                                                                 \
+  STRING(send_directory)                                                                           \
+  STRING(receive_root_directory)                                                                   \
+  INTEGER(save_to_disk)                                                                            \
+  INTEGER(use_multithreading)                                                                      \
+  INTEGER(use_chunk_serialization)                                                                 \
+  INTEGER(use_compression)                                                                         \
+  INTEGER(use_metadata)                                                                            \
+  INTEGER(compression_level)                                                                       \
+  DATA(chunk_size)                                                                                 \
+  INTEGER(use_sendfile)                                                                            \
+  INTEGER(use_delete)                                                                              \
+  INTEGER(use_incremental)                                                                         \
+  INTEGER(use_delta)                                                                               \
+  DATA(delta_block_size)                                                                           \
+  DATA(delta_max_file_size)                                                                        \
+  INTEGER(backup)                                                                                  \
+  OPTIONAL_STRING(backup_dir)                                                                      \
+  INTEGER(follow_symlinks)                                                                         \
+  INTEGER(copy_links)                                                                              \
+  INTEGER(safe_links)                                                                              \
+  INTEGER(copy_unsafe_links)                                                                       \
+  INTEGER(preserve_hard_links)                                                                     \
+  INTEGER(preserve_acls)                                                                           \
+  INTEGER(preserve_xattrs)                                                                         \
+  INTEGER(preserve_devices)                                                                        \
+  INTEGER(preserve_sparse)                                                                         \
+  INTEGER(update)                                                                                  \
+  INTEGER(inplace)                                                                                 \
+  INTEGER(append)                                                                                  \
+  INTEGER(append_verify)                                                                           \
+  INTEGER(delete_excluded)                                                                         \
+  INTEGER(delete_after)                                                                            \
+  DATA(max_delete)                                                                                 \
+  INTEGER(relative)                                                                                \
+  INTEGER(prune_empty_dirs)                                                                        \
+  OPTIONAL_STRING(temp_dir)                                                                        \
+  INTEGER(partial)                                                                                 \
+  OPTIONAL_STRING(partial_dir)                                                                     \
+  OPTIONAL_STRING(suffix)                                                                          \
+  INTEGER(delete_before)                                                                           \
+  INTEGER(checksum)                                                                                \
+  OPTIONAL_STRING(compress_choice)
+
 #define PROTOCOL_VERSION "2.2.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 

src/shared/multiprocessing.c

@@ -13,11 +13,107 @@
 #include <stdlib.h>
 #include <string.h>
 #include <threads.h>
+#include <sys/stat.h>
 
 static bool valid_batch_path(const char* path) {
   return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path);
 }
 
+static bool handle_batch_checks(int file_descriptor, const Config* config) {
+  int count;
+  if (config->checksum || !receive_int(file_descriptor, &count) || count < 0 ||
+      count > MAX_MANIFEST_ENTRIES)
+    return false;
+  for (int i = 0; i < count; i++) {
+    char* check_path = receive_str(file_descriptor);
+    if (!check_path)
+      return false;
+    unsigned long long check_size;
+    long long check_mtime;
+    bool received = receive_n_data(file_descriptor, &check_size, sizeof(check_size)) &&
+                    receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime));
+    if (!received || !valid_batch_path(check_path)) {
+      free(check_path);
+      if (received)
+        send_status(file_descriptor, STATUS_ERROR);
+      return false;
+    }
+    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;
+    bool sent = send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT);
+    free(full_path);
+    free(check_path);
+    if (!sent)
+      return false;
+  }
+  return true;
+}
+
+int receive_files_common(const Config* config, int file_descriptor, ReceivedFileHandler handler,
+                         void* context, bool send_completion_status) {
+  Status status;
+  if (!receive_status(file_descriptor, &status))
+    return -1;
+  while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK ||
+         status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) {
+    if (status == STATUS_KEEPALIVE) {
+      if (!send_status(file_descriptor, STATUS_KEEPALIVE))
+        return -1;
+    } else if (status == STATUS_ABORT) {
+      log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
+      return -1;
+    } else if (status == STATUS_CHECK_BATCH) {
+      if (!handle_batch_checks(file_descriptor, config))
+        return -1;
+    } else {
+      if (status == STATUS_CHUNK) {
+        Chunk* chunk = receive_chunk_data(file_descriptor, config);
+        if (!chunk)
+          return -1;
+        for (int i = 0; i < chunk->element_count; i++) {
+          File* file = chunk->items[i];
+          chunk->items[i] = NULL;
+          if (!handler(file, context)) {
+            file_destroy(file);
+            chunk_destroy(chunk);
+            return -1;
+          }
+        }
+        chunk_destroy(chunk);
+      } else {
+        bool skipped = false;
+        File* file = status == STATUS_CHECK
+                         ? receive_incremental_check(file_descriptor, config, &skipped)
+                         : file_receive(config, file_descriptor);
+        if (!skipped) {
+          if (!file || !handler(file, context)) {
+            file_destroy(file);
+            if (status == STATUS_NEXT)
+              log_message(LOG_LEVEL_ERROR, "Failed to receive file");
+            return -1;
+          }
+        }
+      }
+    }
+    if (!receive_status(file_descriptor, &status))
+      return -1;
+  }
+  if (status == STATUS_MANIFEST && receive_manifest(file_descriptor, config, &status) != 0)
+    return -1;
+  if (status != STATUS_FINISHED) {
+    log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
+    if (send_completion_status)
+      send_status(file_descriptor, STATUS_ERROR);
+    return -1;
+  }
+  if (send_completion_status && !send_status(file_descriptor, STATUS_OK))
+    return -1;
+  return 0;
+}
+
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
                                                       Queue* queue_loader) {
   PipelineContextSender* context = malloc(sizeof(PipelineContextSender));
@@ -137,41 +233,14 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver* context) {
   free(context);
 }
 
-static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* context) {
-  Chunk* chunk = receive_chunk_data(file_descriptor, context->config);
-  if (chunk == NULL)
-    return false;
-
-  for (int i = 0; i < chunk->element_count; i++) {
-    File* file = chunk->items[i];
-    chunk->items[i] = NULL;
-    if (!queue_enqueue_multithreaded_cancel(context->queue, file, &context->mutex,
-                                            &context->condition_not_empty,
-                                            &context->condition_not_full, &context->cancelled)) {
-      file_destroy(file);
-      chunk_destroy(chunk);
-      return false;
-    }
-  }
-  chunk_destroy(chunk);
-  return true;
-}
-
-static void receiver_thread_fail(PipelineContextReceiver* context) {
-  mtx_lock(&context->mutex);
-  atomic_store(&context->cancelled, true);
-  context->receiver_done = true;
-  cnd_broadcast(&context->condition_not_empty);
-  cnd_broadcast(&context->condition_not_full);
-  mtx_unlock(&context->mutex);
+static bool enqueue_received_file(File* file, void* context) {
+  PipelineContextReceiver* receiver = context;
+  return queue_enqueue_multithreaded_cancel(receiver->queue, file, &receiver->mutex,
+                                            &receiver->condition_not_empty,
+                                            &receiver->condition_not_full, &receiver->cancelled);
 }
 
 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);
@@ -180,100 +249,15 @@ int receive_thread(void* pipeline_context) {
   const Config* config = context->config;
   mtx_unlock(&context->mutex);
 
-  Status status;
-  if (!receive_status(file_descriptor, &status))
-    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) {
-      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");
-      RECEIVE_THREAD_FAIL();
-    }
-    if (status == STATUS_CHECK) {
-      bool skipped;
-      File* file = receive_incremental_check(file_descriptor, config, &skipped);
-      if (!skipped) {
-        if (file == NULL)
-          RECEIVE_THREAD_FAIL();
-        if (!queue_enqueue_multithreaded_cancel(
-                context->queue, file, &context->mutex, &context->condition_not_empty,
-                &context->condition_not_full, &context->cancelled)) {
-          file_destroy(file);
-          RECEIVE_THREAD_FAIL();
-        }
-      }
-    } else if (status == STATUS_CHUNK) {
-      if (!receive_chunk_enqueue(file_descriptor, context))
-        RECEIVE_THREAD_FAIL();
-    } else if (status == STATUS_CHECK_BATCH) {
-      int count;
-      if (config->checksum || !receive_int(file_descriptor, &count) || count < 0 ||
-          count > MAX_MANIFEST_ENTRIES)
-        RECEIVE_THREAD_FAIL();
-      for (int i = 0; i < count; i++) {
-        char* check_path = receive_str(file_descriptor);
-        if (!check_path)
-          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);
-          RECEIVE_THREAD_FAIL();
-        }
-        if (!valid_batch_path(check_path)) {
-          free(check_path);
-          if (!send_status(file_descriptor, STATUS_ERROR))
-            RECEIVE_THREAD_FAIL();
-          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 (!send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT))
-          RECEIVE_THREAD_FAIL();
-        free(full_path);
-        free(check_path);
-      }
-      goto next;
-    } else {
-      File* file = file_receive(config, file_descriptor);
-      if (file) {
-        if (!queue_enqueue_multithreaded_cancel(
-                context->queue, file, &context->mutex, &context->condition_not_empty,
-                &context->condition_not_full, &context->cancelled)) {
-          file_destroy(file);
-          receiver_thread_fail(context);
-          return thrd_error;
-        }
-      } else {
-        log_message(LOG_LEVEL_ERROR, "Failed to receive file");
-        RECEIVE_THREAD_FAIL();
-      }
-    }
-  next:
-    if (!receive_status(file_descriptor, &status))
-      RECEIVE_THREAD_FAIL();
-  }
-  if (status == STATUS_MANIFEST) {
-    if (receive_manifest(file_descriptor, config, &status) != 0)
-      RECEIVE_THREAD_FAIL();
-  }
-  if (status != STATUS_FINISHED)
-    RECEIVE_THREAD_FAIL();
+  int result = receive_files_common(config, file_descriptor, enqueue_received_file, context, false);
   mtx_lock(&context->mutex);
+  if (result != 0)
+    atomic_store(&context->cancelled, true);
   context->receiver_done = true;
   cnd_signal(&context->condition_not_empty);
+  cnd_broadcast(&context->condition_not_full);
   mtx_unlock(&context->mutex);
-#undef RECEIVE_THREAD_FAIL
-  return thrd_success;
+  return result == 0 ? thrd_success : thrd_error;
 }
 
 int write_thread(void* pipeline_context) {

src/shared/multiprocessing.h

@@ -42,6 +42,11 @@ typedef struct PipelineContextReceiver {
   atomic_bool cancelled;
 } PipelineContextReceiver;
 
+typedef bool (*ReceivedFileHandler)(File* file, void* context);
+
+int receive_files_common(const Config* config, int file_descriptor, ReceivedFileHandler handler,
+                         void* context, bool send_completion_status);
+
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
                                                       Queue* queue_loader);
 void pipeline_context_sender_destroy(PipelineContextSender* context);

src/shared/transport_tcp.c

@@ -173,7 +173,7 @@ Client* client_create() {
   return client;
 }
 
-bool client_connect(Client* client, char* host, int port) {
+bool tcp_connect_socket(Client* client, const char* host, int port) {
   struct addrinfo hints;
   struct addrinfo* result;
   memset(&hints, 0, sizeof(hints));
@@ -226,6 +226,10 @@ bool client_connect(Client* client, char* host, int port) {
   return true;
 }
 
+bool client_connect(Client* client, char* host, int port) {
+  return tcp_connect_socket(client, host, port);
+}
+
 void client_disconnect(Client* client) {
   if (client->ssl) {
     SSL_shutdown(client->ssl);

src/shared/transport_tcp.h

@@ -29,6 +29,7 @@ void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* chil
                         const char* log_fmt);
 void server_delete(Server** server);
 Client* client_create();
+bool tcp_connect_socket(Client* client, const char* host, int port);
 bool client_connect(Client* client, char* host, int port);
 void client_disconnect(Client* client);
 void client_delete(Client* client);

src/shared/transport_tls.c

@@ -2,8 +2,6 @@
 #include "log.h"
 #include "protocol.h"
 #include "transport_tcp.h"
-#include <arpa/inet.h>
-#include <netdb.h>
 #include <openssl/err.h>
 #include <openssl/ssl.h>
 #include <signal.h>
@@ -149,53 +147,8 @@ bool server_listen_tls(Server* server, void (*handler)(int file_descriptor)) {
 
 bool client_connect_tls(Client* client, char* host, int port, const char* cert_path,
                         const char* key_path, const char* ca_path) {
-  struct addrinfo hints;
-  struct addrinfo* result;
-  memset(&hints, 0, sizeof(hints));
-  hints.ai_family = AF_UNSPEC;
-  hints.ai_socktype = SOCK_STREAM;
-  hints.ai_protocol = IPPROTO_TCP;
-
-  char port_str[16];
-  snprintf(port_str, sizeof(port_str), "%d", port);
-
-  int err = getaddrinfo(host, port_str, &hints, &result);
-  if (err != 0 || result == NULL) {
-    fprintf(stderr, "Could not resolve host: %s (%s)\n", host, gai_strerror(err));
+  if (!tcp_connect_socket(client, host, port))
     return false;
-  }
-
-  struct addrinfo* rp;
-  bool connected = false;
-  for (rp = result; rp != NULL; rp = rp->ai_next) {
-    if (client->file_descriptor >= 0)
-      close(client->file_descriptor);
-
-    client->file_descriptor = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
-    if (client->file_descriptor < 0)
-      continue;
-
-    struct timeval ct;
-    ct.tv_sec = tcp_get_contimeout_sec();
-    ct.tv_usec = 0;
-    setsockopt(client->file_descriptor, SOL_SOCKET, SO_RCVTIMEO, &ct, sizeof(ct));
-    setsockopt(client->file_descriptor, SOL_SOCKET, SO_SNDTIMEO, &ct, sizeof(ct));
-
-    memcpy(&client->address, rp->ai_addr, rp->ai_addrlen);
-    client->address_length = rp->ai_addrlen;
-
-    if (connect(client->file_descriptor, (struct sockaddr*)&client->address,
-                client->address_length) == 0) {
-      connected = true;
-      break;
-    }
-  }
-  freeaddrinfo(result);
-
-  if (!connected) {
-    perror("Could not connect to Server!");
-    return false;
-  }
 
   SSL_CTX* ctx = create_ssl_ctx(false, cert_path, key_path, ca_path);
   if (!ctx)