From 2450b90dd06f9fc41c4fd1d5bc5beb82d1265b49 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 8 Aug 2026 20:27:26 +0200 Subject: [PATCH] feat: add checksum-aware incremental sync --- README.md | 8 +++-- src/client/client_cli.c | 8 +++++ src/client/client_send.c | 13 +++++-- src/server/server.c | 15 +++++---- src/shared/config.c | 7 ++++ src/shared/config.h | 2 +- src/shared/delta.c | 4 +++ src/shared/delta.h | 1 + src/shared/file.c | 54 ++++++++++++++++++++++++++---- src/shared/file.h | 1 + src/shared/metadata.c | 13 ++++++- src/shared/multiprocessing.c | 1 + src/shared/multiprocessing.h | 2 ++ src/shared/protocol.c | 26 +++++++++++--- src/shared/protocol.h | 1 + src/shared/queue.h | 3 ++ tests/integration/test_features.py | 20 +++++++++++ tests/test_delta.c | 5 +++ tests/test_metadata.c | 14 ++++++++ tests/test_server.c | 16 +++++++++ 20 files changed, 189 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index 353133c..f692060 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,7 @@ A high-performance file synchronization system with SSH and TCP transport, TLS e | `STATUS_NEXT` | Ready for next file (per-file mode) | | `STATUS_CHUNK` | Following data is a serialized chunk | | `STATUS_MANIFEST` | Following data is a file manifest (for `--delete`) | -| `STATUS_CHECK` | Incremental check: client sends file path + size + mtime, server responds with OK (skip) or NEXT (send) | +| `STATUS_CHECK` | Incremental check: client sends file path + size + mtime and, when negotiated, checksum; server responds with OK (skip) or NEXT (send) | | `STATUS_CHECK_BATCH` | Batch incremental check: multiple file checks sent in one message | | `STATUS_KEEPALIVE` | Keep-alive heartbeat to detect stalled connections | | `STATUS_ABORT` | Abort signal: client interrupts, server cleans up and exits | @@ -95,7 +95,9 @@ Abort (`STATUS_ABORT`) may be sent at any point. On receipt the server cleans up ### Protocol Version -`1.3.0` — server and client must match. Mismatch results in `STATUS_ERROR`. +`2.2.0` — server and client must match. This version adds a 64-bit XXH64 checksum to checksum-enabled `STATUS_CHECK` messages and validates the negotiated compression choice (`zstd` or `none`). Older clients and servers must not be mixed with this version; mismatch results in `STATUS_ERROR`. + +Config negotiation is sender-driven: the client serializes transfer options and the server applies them while receiving and writing files. `--checksum` compares size and content checksum instead of timestamps. `--compress-choice zstd` enables zstd; `none` disables it. Unsupported choices are rejected during config exchange. ## Command-Line Arguments @@ -186,7 +188,7 @@ Abort (`STATUS_ABORT`) may be sent at any point. On receipt the server cleans up 2. **Chunking** — files accumulated until `chunk_size` threshold, then flushed 3. **Compression** — streaming zstd via `ZSTD_compressStream2` / `ZSTD_decompressStream` 4. **Network protocol** — status-code-driven exchange with metadata packing, keep-alive, and abort support -5. **Incremental check** — client sends `STATUS_CHECK` + path + size + mtime; server compares against destination. Can be batched via `STATUS_CHECK_BATCH` for reduced round-trips. +5. **Incremental check** — client sends `STATUS_CHECK` + path + size + mtime and, with `--checksum`, XXH64 content checksum; server compares against destination. Can be batched via `STATUS_CHECK_BATCH` for reduced round-trips. 6. **Bandwidth limiting** — token-bucket algorithm with `nanosleep` throttling on 64 KB write chunks 7. **Metadata restoration** — `chmod()`, `chown()`, `utimensat()` on the receiving side 8. **`--delete`** — sender tracks all sent paths; receiver walks destination tree and removes unlisted files/directories diff --git a/src/client/client_cli.c b/src/client/client_cli.c index d1149fc..a144c25 100644 --- a/src/client/client_cli.c +++ b/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)) { diff --git a/src/client/client_send.c b/src/client/client_send.c index 620cb57..1d5468f 100644 --- a/src/client/client_send.c +++ b/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; diff --git a/src/server/server.c b/src/server/server.c index f8fdb54..704537b 100644 --- a/src/server/server.c +++ b/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); diff --git a/src/shared/config.c b/src/shared/config.c index 6d52cc4..3ad4aa4 100644 --- a/src/shared/config.c +++ b/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; diff --git a/src/shared/config.h b/src/shared/config.h index 9c78505..986a5c4 100644 --- a/src/shared/config.h +++ b/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); diff --git a/src/shared/delta.c b/src/shared/delta.c index 0305fdc..da03280 100644 --- a/src/shared/delta.c +++ b/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) diff --git a/src/shared/delta.h b/src/shared/delta.h index 96cf0a6..0d8ddb3 100644 --- a/src/shared/delta.h +++ b/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 diff --git a/src/shared/file.c b/src/shared/file.c index 2b730f8..e6c14e0 100644 --- a/src/shared/file.c +++ b/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); diff --git a/src/shared/file.h b/src/shared/file.h index b0a1b4e..27cbadb 100644 --- a/src/shared/file.h +++ b/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); diff --git a/src/shared/metadata.c b/src/shared/metadata.c index 4134c13..7f246df 100644 --- a/src/shared/metadata.c +++ b/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; diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index f5019a9..dcf5588 100644 --- a/src/shared/multiprocessing.c +++ b/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); diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index aa53a13..443720a 100644 --- a/src/shared/multiprocessing.h +++ b/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, diff --git a/src/shared/protocol.c b/src/shared/protocol.c index 0b32f96..4a8ff98 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -1,6 +1,7 @@ #include "protocol.h" #include "log.h" #include +#include #include #include #include @@ -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) diff --git a/src/shared/protocol.h b/src/shared/protocol.h index 9173fd3..d432503 100644 --- a/src/shared/protocol.h +++ b/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; diff --git a/src/shared/queue.h b/src/shared/queue.h index 8bd6ba4..3d03039 100644 --- a/src/shared/queue.h +++ b/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); diff --git a/tests/integration/test_features.py b/tests/integration/test_features.py index 63a5bf6..9635fc7 100644 --- a/tests/integration/test_features.py +++ b/tests/integration/test_features.py @@ -193,6 +193,26 @@ class TestIncremental: content = f.read() assert b"modified content" in content, f"Modified content not transferred: {content[:50]}" + def test_checksum_detects_same_size_and_mtime_change(self, shared_server): + clean_dir(DEST_DIR) + result, _ = run_client(SOURCE_DIR, DEST_DIR, flags=["-M"], port=shared_server.port) + assert result.returncode == 0 + + received = get_dest_received_dir(DEST_DIR, SOURCE_DIR) + source_file = os.path.join(SOURCE_DIR, "small.txt") + received_file = os.path.join(received, "small.txt") + source_stat = os.stat(source_file) + with open(received_file, "wb") as f: + f.write(b"different!\n") + os.utime(received_file, (source_stat.st_atime, source_stat.st_mtime)) + + result, _ = run_client(SOURCE_DIR, DEST_DIR, + flags=["-M", "--incremental", "--checksum"], + port=shared_server.port) + assert result.returncode == 0, f"Checksum sync failed: {result.stderr[:200]}" + with open(received_file, "rb") as f: + assert f.read() == b"hello world\n" + class TestDelete: def test_delete_removes_extra_files(self, shared_server): diff --git a/tests/test_delta.c b/tests/test_delta.c index 09a6467..7030ff2 100644 --- a/tests/test_delta.c +++ b/tests/test_delta.c @@ -35,6 +35,10 @@ static void test_xxhash32_different_data() { EXPECT_TRUE(ha != hb); } +static void test_xxhash64_different_data() { + EXPECT_TRUE(delta_xxhash64("AAAA", 4) != delta_xxhash64("BBBB", 4)); +} + static void test_signature_roundtrip() { char old_data[4096]; for (int i = 0; i < 4096; i++) @@ -328,6 +332,7 @@ void test_delta() { test_adler32_different_data(); test_xxhash32_basic(); test_xxhash32_different_data(); + test_xxhash64_different_data(); test_signature_roundtrip(); test_delta_identical_files(); test_delta_small_edit(); diff --git a/tests/test_metadata.c b/tests/test_metadata.c index 6a27660..e73e922 100644 --- a/tests/test_metadata.c +++ b/tests/test_metadata.c @@ -109,6 +109,19 @@ static void test_metadata_send_null() { close(p[1]); } +static void test_metadata_rejects_invalid_values() { + int p[2]; + EXPECT_EQ_INT(pipe(p), 0); + io_set_fds(p[0], p[1]); + int32_t present = 2; + EXPECT_TRUE(send_n_data(p[1], &present, sizeof(present))); + int ok = 1; + EXPECT_NULL(metadata_receive(p[0], &ok)); + EXPECT_EQ_INT(ok, 0); + close(p[0]); + close(p[1]); +} + static void test_file_restore_metadata() { const char* path = "temp_meta_restore_test.txt"; const char* content = "test content"; @@ -137,5 +150,6 @@ void test_metadata() { test_metadata_from_buf_null(); test_metadata_send_receive_roundtrip(); test_metadata_send_null(); + test_metadata_rejects_invalid_values(); test_file_restore_metadata(); } diff --git a/tests/test_server.c b/tests/test_server.c index 9226e00..38c99d9 100644 --- a/tests/test_server.c +++ b/tests/test_server.c @@ -171,10 +171,26 @@ static void test_receive_files_abort() { } } +static void test_receive_manifest_rejects_traversal() { + Config* cfg = config_create(); + EXPECT_NOT_NULL(cfg); + cfg->receive_root_directory = str_dup("/tmp/dst"); + int p[2]; + EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0); + io_set_fds(p[0], p[1]); + EXPECT_TRUE(send_int(p[1], 1)); + EXPECT_TRUE(send_str(p[1], "../outside")); + EXPECT_EQ_INT(receive_manifest(p[0], cfg, NULL), -1); + close(p[0]); + close(p[1]); + config_delete(cfg); +} + void test_server() { if (!is_running_under_valgrind()) { test_receive_files_finished(); test_receive_files_single_file(); test_receive_files_abort(); + test_receive_manifest_rejects_traversal(); } }