feat: add checksum-aware incremental sync

src/client/client_cli.c

@@ -461,6 +461,14 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
         return -1;
       free(config->compress_choice);
       config->compress_choice = dup;
+      if (strcmp(config->compress_choice, "zstd") == 0)
+        config->use_compression = true;
+      else if (strcmp(config->compress_choice, "none") == 0)
+        config->use_compression = false;
+      else {
+        fprintf(stderr, "Error: --compress-choice must be 'zstd' or 'none'\n");
+        return -1;
+      }
     } else if (strcmp(argv[i], "--compress-level") == 0 && i + 1 < argc) {
       int val;
       if (!parse_positive_int(argv[++i], &val)) {

src/client/client_send.c

@@ -67,7 +67,8 @@ static int send_delete_manifest(int fd, ArrayList* manifest) {
   return 0;
 }
 
-static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) {
+static int incremental_check(Client* client, File* file, const Config* config,
+                             DeltaSignature** out_sig) {
   *out_sig = NULL;
   if (!send_status(client->file_descriptor, STATUS_CHECK))
     return -1;
@@ -79,6 +80,12 @@ static int incremental_check(Client* client, File* file, DeltaSignature** out_si
     return -1;
   if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime)))
     return -1;
+  if (config->checksum) {
+    uint64_t checksum;
+    if (!file_checksum(file, &checksum) ||
+        !send_n_data(client->file_descriptor, &checksum, sizeof(checksum)))
+      return -1;
+  }
   Status s;
   if (!receive_status(client->file_descriptor, &s))
     return -1;
@@ -176,7 +183,7 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
   // Incremental path: use sendfile for the actual data if enabled and no compression
   if (use_sendfile) {
     DeltaSignature* sig = NULL;
-    int rc = incremental_check(client, file, &sig);
+    int rc = incremental_check(client, file, config, &sig);
     if (rc == 1) {
       delta_signature_destroy(sig);
       return 1;
@@ -202,7 +209,7 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
   // Incremental path with single_calls (supports compression and delta)
   file_send_fn send_fn = (file_send_fn)file_send_single_calls;
   DeltaSignature* sig = NULL;
-  int rc = incremental_check(client, file, &sig);
+  int rc = incremental_check(client, file, config, &sig);
   if (rc < 0) {
     delta_signature_destroy(sig);
     return -1;

src/server/server.c

@@ -39,7 +39,7 @@ int receive_files(Config* config, int fd) {
       if (file == NULL && !skipped)
         return -1;
       if (config->save_to_disk)
-        file_save_to_disk(config->receive_root_directory, file, NULL);
+        file_save_to_disk(config->receive_root_directory, file, config);
       file_destroy(file);
     } else if (status == STATUS_CHUNK) {
       Chunk* chunk = receive_chunk_data(fd, config);
@@ -49,7 +49,7 @@ int receive_files(Config* config, int fd) {
       }
       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], NULL);
+          file_save_to_disk(config->receive_root_directory, chunk->items[i], config);
       }
       chunk_destroy(chunk);
     } else if (status == STATUS_CHECK_BATCH) {
@@ -88,7 +88,7 @@ int receive_files(Config* config, int fd) {
         return -1;
       }
       if (config->save_to_disk)
-        file_save_to_disk(config->receive_root_directory, file, NULL);
+        file_save_to_disk(config->receive_root_directory, file, config);
       file_destroy(file);
     }
   next:
@@ -142,9 +142,12 @@ void handler(int file_descriptor) {
       close(file_descriptor);
       return;
     }
-    thrd_join(receiver, NULL);
-    thrd_join(writer, NULL);
-    send_status(file_descriptor, STATUS_OK);
+    int receiver_result;
+    int writer_result;
+    thrd_join(receiver, &receiver_result);
+    thrd_join(writer, &writer_result);
+    if (receiver_result == thrd_success && writer_result == thrd_success)
+      send_status(file_descriptor, STATUS_OK);
     pipeline_context_receiver_destroy(context);
   } else {
     receive_files(config, file_descriptor);

src/shared/config.c

@@ -469,6 +469,13 @@ Config* config_receive(int file_descriptor) {
   config->compress_choice = receive_str(file_descriptor);
   if (config->compress_choice == NULL)
     goto error;
+  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);
+    send_status(file_descriptor, STATUS_ERROR);
+    goto error;
+  }
   config->address = NULL;
   config->bind_address = NULL;
   config->ipv6 = false;

src/shared/config.h

@@ -128,7 +128,7 @@ typedef struct Config {
   char* compress_choice;
 } Config;
 
-#define PROTOCOL_VERSION "1.3.0"
+#define PROTOCOL_VERSION "2.2.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 
 Config* config_create(void);

src/shared/delta.c

@@ -27,6 +27,10 @@ uint32_t delta_xxhash32(const void* data, uint32_t len) {
   return XXH32(data, len, 0);
 }
 
+uint64_t delta_xxhash64(const void* data, size_t len) {
+  return XXH64(data, len, 0);
+}
+
 DeltaSignature* delta_signature_create(const void* old_file_data, uint64_t old_file_size,
                                        uint32_t block_size) {
   if (old_file_data == NULL || old_file_size == 0 || block_size == 0)

src/shared/delta.h

@@ -72,5 +72,6 @@ bool delta_is_worthwhile(const Delta* delta, uint64_t new_file_size);
 
 uint32_t delta_adler32(const void* data, uint32_t len);
 uint32_t delta_xxhash32(const void* data, uint32_t len);
+uint64_t delta_xxhash64(const void* data, size_t len);
 
 #endif

src/shared/file.c

@@ -21,6 +21,19 @@
 #include "protocol.h"
 #include "utils.h"
 
+bool file_checksum(File* file, uint64_t* checksum) {
+  if (!file || !checksum || !file->data)
+    return false;
+  if (file->data->size == 0) {
+    *checksum = delta_xxhash64("", 0);
+    return true;
+  }
+  if (!file->data->data && !file_load_data(file))
+    return false;
+  *checksum = delta_xxhash64(file->data->data, file->data->size);
+  return true;
+}
+
 File* file_create(const char* path) {
   File* file = (File*)malloc(sizeof(File));
   if (file == NULL) {
@@ -421,12 +434,18 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
 
   unsigned long long check_size;
   long long check_mtime;
+  uint64_t check_checksum = 0;
   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;
   }
+  if (config->checksum && !receive_n_data(fd, &check_checksum, sizeof(check_checksum))) {
+    free(check_path);
+    send_status(fd, STATUS_ERROR);
+    return NULL;
+  }
 
   if (has_path_traversal(check_path)) {
     log_message(LOG_LEVEL_ERROR, "Path traversal detected: %s", check_path);
@@ -440,8 +459,17 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
   bool has_old_file = (full_path && lstat(full_path, &st) == 0);
   unsigned long long old_size = has_old_file ? (unsigned long long)st.st_size : 0;
 
-  bool match = has_old_file && (unsigned long long)st.st_size == check_size &&
-               (long long)st.st_mtime == check_mtime;
+  bool match = has_old_file && (unsigned long long)st.st_size == check_size;
+  if (match && config->checksum) {
+    void* old_data = old_size > 0 ? old_data_from_path(full_path, old_size) : NULL;
+    uint64_t old_checksum = old_size == 0 ? delta_xxhash64("", 0) : 0;
+    if (old_data)
+      old_checksum = delta_xxhash64(old_data, (size_t)old_size);
+    match = (old_size == 0 || old_data) && old_checksum == check_checksum;
+    free(old_data);
+  } else if (match) {
+    match = (long long)st.st_mtime == check_mtime;
+  }
 
   if (match) {
     if (!send_status(fd, STATUS_OK)) {
@@ -664,6 +692,11 @@ File* file_receive(const Config* config, int file_descriptor) {
   char* path = receive_str(file_descriptor);
   if (path == NULL)
     return NULL;
+  if (path[0] == '\0' || has_path_traversal(path)) {
+    log_message(LOG_LEVEL_ERROR, "Invalid received file path: %s", path);
+    free(path);
+    return NULL;
+  }
   File* file = file_create(path);
   free(path);
   if (file == NULL)
@@ -715,13 +748,20 @@ int receive_manifest(int fd, const Config* config, int* next_status) {
   int count;
   if (!receive_int(fd, &count))
     return -1;
+  if (count < 0 || count > MAX_MANIFEST_ENTRIES)
+    return -1;
   ArrayList* manifest = array_list_create(free);
-  if (manifest) {
-    for (int i = 0; i < count; i++) {
-      char* s = receive_str(fd);
-      if (s)
-        array_list_add(manifest, s);
+  if (!manifest)
+    return -1;
+  for (int i = 0; i < count; i++) {
+    char* s = receive_str(fd);
+    if (!s || s[0] == '\0' || has_path_traversal(s) || !array_list_add(manifest, s)) {
+      free(s);
+      array_list_delete(manifest);
+      return -1;
     }
+  }
+  if (manifest) {
     fprintf(stderr, "Deleting files not in manifest...\n");
     delete_extras(config->receive_root_directory, manifest);
     array_list_delete(manifest);

src/shared/file.h

@@ -26,6 +26,7 @@ typedef struct {
 File* file_create(const char* path);
 void file_destroy(void* item);
 bool file_load_data(File* file);
+bool file_checksum(File* file, uint64_t* checksum);
 File* file_receive(const Config* config, int file_descriptor);
 bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
                             int compression_level, bool send_path);

src/shared/metadata.c

@@ -105,11 +105,16 @@ FileMetadata* metadata_receive(int file_descriptor, int* ok) {
       *ok = 0;
     return NULL;
   }
-  if (!present) {
+  if (present == 0) {
     if (ok)
       *ok = 1;
     return NULL;
   }
+  if (present != 1) {
+    if (ok)
+      *ok = 0;
+    return NULL;
+  }
   FileMetadata* m = malloc(sizeof(FileMetadata));
   if (m == NULL) {
     if (ok)
@@ -156,6 +161,12 @@ FileMetadata* metadata_receive(int file_descriptor, int* ok) {
     return NULL;
   }
   m->mtime_nsec = (long)mtime_nsec;
+  if (mtime_nsec < 0 || mtime_nsec >= 1000000000LL || mode < 0 || uid < 0 || gid < 0) {
+    free(m);
+    if (ok)
+      *ok = 0;
+    return NULL;
+  }
   if (ok)
     *ok = 1;
   return m;

src/shared/multiprocessing.c

@@ -103,6 +103,7 @@ static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver*
 
 static void receiver_thread_fail(PipelineContextReceiver* context) {
   mtx_lock(&context->mutex);
+  context->cancelled = true;
   context->receiver_done = true;
   cnd_broadcast(&context->condition_not_empty);
   cnd_broadcast(&context->condition_not_full);

src/shared/multiprocessing.h

@@ -26,6 +26,7 @@ typedef struct {
   mtx_t mutex_progress;
   unsigned long long progress_bytes;
   bool sender_done;
+  bool cancelled;
 } PipelineContextSender;
 
 typedef struct PipelineContextReceiver {
@@ -37,6 +38,7 @@ typedef struct PipelineContextReceiver {
   cnd_t condition_not_full;
   cnd_t condition_not_empty;
   bool receiver_done;
+  bool cancelled;
 } PipelineContextReceiver;
 
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,

src/shared/protocol.c

@@ -1,6 +1,7 @@
 #include "protocol.h"
 #include "log.h"
 #include <errno.h>
+#include <limits.h>
 #include <openssl/ssl.h>
 #include <poll.h>
 #include <stdio.h>
@@ -75,6 +76,17 @@ static int io_fd(int dir_fd, int file_descriptor) {
   return (dir_fd != -1) ? dir_fd : file_descriptor;
 }
 
+static int deadline_remaining_ms(const struct timespec* deadline) {
+  struct timespec now;
+  clock_gettime(CLOCK_MONOTONIC, &now);
+  long long ns = (long long)(deadline->tv_sec - now.tv_sec) * 1000000000LL +
+                deadline->tv_nsec - now.tv_nsec;
+  if (ns <= 0)
+    return 0;
+  long long ms = (ns + 999999) / 1000000;
+  return ms > INT_MAX ? INT_MAX : (int)ms;
+}
+
 bool send_n_data(int file_descriptor, const void* data, size_t data_size) {
   log_message(LOG_LEVEL_DEBUG, "    Sending n Data: %zu", data_size);
   int fd = io_fd(io_write_fd, file_descriptor);
@@ -114,13 +126,19 @@ bool receive_n_data(int file_descriptor, void* data, size_t data_size) {
 
   size_t total_bytes_received = 0;
   while (total_bytes_received < data_size) {
-    struct timespec now;
-    clock_gettime(CLOCK_MONOTONIC, &now);
-    if (now.tv_sec > deadline.tv_sec ||
-        (now.tv_sec == deadline.tv_sec && now.tv_nsec > deadline.tv_nsec)) {
+    struct pollfd pfd = {.fd = fd, .events = POLLIN};
+    int poll_result = poll(&pfd, 1, deadline_remaining_ms(&deadline));
+    if (poll_result == 0) {
       log_message(LOG_LEVEL_ERROR, "Receive timeout after %ds", RECEIVE_TIMEOUT_SEC);
       return false;
     }
+    if (poll_result < 0) {
+      if (errno == EINTR)
+        continue;
+      return false;
+    }
+    if (pfd.revents & (POLLERR | POLLNVAL))
+      return false;
 
     ssize_t bytes_received;
     if (io_ssl)

src/shared/protocol.h

@@ -13,6 +13,7 @@
 
 /* Maximum chunk size (64 MB) — prevents unbounded allocation from the wire */
 #define MAX_CHUNK_SIZE (64ULL * 1024 * 1024)
+#define MAX_MANIFEST_ENTRIES (1024 * 1024)
 
 typedef struct ssl_st SSL;
 

src/shared/queue.h

@@ -20,6 +20,9 @@ bool queue_is_full(const Queue* queue);
 bool queue_enqueue(Queue* queue, void* item);
 bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty,
                                  cnd_t* condition_not_full);
+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);
 void* queue_dequeue(Queue* queue);
 void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty,
                                   cnd_t* condition_not_full, const bool* other_thread_done);