Merge pull request 'refactor: extract CLI parser helpers, fix validation bugs' (#200) from refactor/cli-parser-helpers into dev

Reviewed-on: #200

AGENTS.md

@@ -206,7 +206,7 @@ tea pr close <number> --repo TapTap/FastSync
 
 ## Common pitfalls
 
-- **`__thread` on shared SSL context**: io_ssl must NOT be thread-local — worker threads inherit the SSL context from the main thread. Use regular `static SSL* io_ssl`.
+- **Per-thread SSL context**: `io_ssl` is stored per-thread (`static __thread SSL* io_ssl`). Each thread that performs protocol I/O must call `io_set_ssl()` to install its own SSL object before using `send_*` / `receive_*` primitives. The main thread's SSL context is not automatically inherited by worker threads.
 - **SSL WANT_READ/WANT_WRITE retry**: Always retry on `SSL_ERROR_WANT_READ` and `SSL_ERROR_WANT_WRITE` in `send_n_data`/`receive_n_data`. Removing these breaks TLS multithreaded transfers.
 - **clang-format version**: The CI image uses clang-format 18. Always format inside the CI Docker container for exact match.
 - **Merge order matters**: Merge the most comprehensive branch first, then smaller ones, to minimize conflicts when creating a combined branch.

CMakeLists.txt

@@ -79,8 +79,9 @@ 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")
-add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} src/client/scanner.c)
+add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} src/client/scanner.c src/client/client_cli.c)
 target_include_directories(tests PRIVATE ${TEST_INCLUDES})
+target_compile_definitions(tests PRIVATE FASTSYNC_TEST_BUILD)
 target_link_libraries(tests PRIVATE ${TEST_LIBS})
 add_test(NAME unit_all COMMAND tests)
 

RSYNC_COMPAT.md

@@ -0,0 +1,274 @@
+# Rsync Feature Compatibility
+
+This document maps rsync's full feature set to FastSync's current implementation status.
+
+## Summary
+
+| Status | Count | Description |
+|--------|-------|-------------|
+| ✅ Implemented | 39 | Feature works end-to-end |
+| 🔀 Alt Arg | 5 | Functionality exists but under different flag/semantics |
+| ⚠️ Partial | 30 | Flag parsed/stored but behavior incomplete |
+| ❌ Not Implemented | 62 | Flag not recognized or no behavior |
+| **Total** | **136** | |
+
+---
+
+## 1. General Options
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-a`, `--archive` | Archive mode is -rlptgoD | 🔀 Alt Arg | Maps to -c -m -M (compression + multithread + metadata) |
+| `-v`, `--verbose` | Increase verbosity | ✅ Implemented | Sets `log_level=DEBUG` |
+| `-q`, `--quiet` | Suppress non-error messages | ✅ Implemented | `quiet` config field |
+| `-h`, `--help` | Show help | ✅ Implemented | Prints usage and exits |
+| `-V`, `--version` | Print version | ❌ Not Implemented | |
+| `--info=FLAGS` | Fine-grained info verbosity | ⚠️ Partial | `info_level` stored, not wired |
+| `--debug=FLAGS` | Fine-grained debug verbosity | ⚠️ Partial | `debug_level` stored, not wired |
+| `--stderr=MODE` | Change stderr output mode | ❌ Not Implemented | |
+| `--no-motd` | Suppress daemon MOTD | ❌ Not Implemented | |
+| `--exclude=PATTERN` | Exclude files matching pattern | ✅ Implemented | Glob matching in scanner |
+| `--include=PATTERN` | Include files matching pattern | ✅ Implemented | Glob matching in scanner |
+| `-C`, `--cvs-exclude` | Auto-ignore CVS files | ⚠️ Partial | `cvs_exclude` stored, not wired |
+
+## 2. Modifying Output
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--stats` | Give transfer stats | ✅ Implemented | Prints file/byte counts |
+| `-h`, `--human-readable` | Human-readable numbers | ⚠️ Partial | `human_readable` stored, not wired |
+| `-i`, `--itemize-changes` | Per-file change summary | ⚠️ Partial | `itemize_changes` stored, not wired |
+| `--progress` | Show progress | ✅ Implemented | Progress callback in sender |
+| `-P` | Same as --partial --progress | ❌ Not Implemented | |
+| `--out-format=FORMAT` | Custom output format | ⚠️ Partial | `out_format` stored, not wired |
+| `--log-file=FILE` | Log to file | ✅ Implemented | `log_file` config field |
+| `--log-file-format=FMT` | Log format | ❌ Not Implemented | |
+| `--8-bit-output` | Leave high-bit chars unescaped | ❌ Not Implemented | |
+| `--list-only` | List files instead of copying | ⚠️ Partial | `list_only` stored, not wired |
+
+## 3. File Selection
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--exclude-from=FILE` | Read exclude patterns from file | ✅ Implemented | Reads patterns from file |
+| `--include-from=FILE` | Read include patterns from file | ✅ Implemented | Reads patterns from file |
+| `--filter=RULE` | Add file-filtering rule | ⚠️ Partial | `filters` ArrayList stored, not wired |
+| `--files-from=FILE` | Read source file list from file | ⚠️ Partial | `files_from` stored, not wired |
+| `-0`, `--from0` | Delimit *-from files with NULs | ❌ Not Implemented | |
+| `--max-size=SIZE` | Skip files larger than SIZE | ✅ Implemented | `max_size` in scanner |
+| `--min-size=SIZE` | Skip files smaller than SIZE | ✅ Implemented | `min_size` in scanner |
+| `-I`, `--ignore-times` | Don't skip files matching size+time | ❌ Not Implemented | |
+| `--size-only` | Skip based on size only | ❌ Not Implemented | |
+| `-@`, `--modify-window=NUM` | Mod-time comparison accuracy | ❌ Not Implemented | |
+| `--existing` | Skip creating new files on receiver | ❌ Not Implemented | |
+| `--ignore-existing` | Skip updating existing files | ❌ Not Implemented | |
+| `--remove-source-files` | Sender removes synced files | ❌ Not Implemented | |
+
+## 4. Directory Options
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-r`, `--recursive` | Recurse into directories | ✅ Implemented | Default behavior |
+| `-R`, `--relative` | Use relative path names | ⚠️ Partial | `relative` stored, not wired |
+| `--no-implied-dirs` | Don't send implied dirs with -R | ❌ Not Implemented | |
+| `-d`, `--dirs` | Transfer dirs without recursing | ❌ Not Implemented | |
+| `--mkpath` | Create missing path components | ❌ Not Implemented | |
+
+## 5. Transfer Modifications
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-u`, `--update` | Skip files newer on receiver | ⚠️ Partial | `update` stored, not wired |
+| `--inplace` | Update files in-place | ✅ Implemented | Direct write mode |
+| `--append` | Append data to shorter files | ⚠️ Partial | `append` stored, not wired |
+| `--append-verify` | Append with old-data checksum | ⚠️ Partial | `append_verify` stored, not wired |
+| `-W`, `--whole-file` | Copy whole file (no delta) | ❌ Not Implemented | |
+| `--block-size=SIZE` | Force checksum block-size | ⚠️ Partial | Parsed as `--delta-block`; controls delta transfer block size |
+
+## 6. Destination Handling
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-n`, `--dry-run` | Trial run with no changes | ✅ Implemented | `dry_run` config field |
+| `-b`, `--backup` | Make backups of overwritten files | ✅ Implemented | Backup before overwrite |
+| `--backup-dir=DIR` | Backup directory hierarchy | ✅ Implemented | `backup_dir` config field |
+| `--suffix=SUFFIX` | Backup suffix (default ~) | ✅ Implemented | `suffix` config field |
+| `--delay-updates` | Put updated files in place at end | ❌ Not Implemented | |
+
+## 7. Deletion
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--delete` | Delete extraneous files from dest | ✅ Implemented | `use_delete` config field |
+| `--delete-before` | Delete before transfer | ⚠️ Partial | `delete_before` stored, not wired |
+| `--delete-during` | Delete during transfer | ❌ Not Implemented | |
+| `--delete-delay` | Find deletions during, delete after | ❌ Not Implemented | |
+| `--delete-after` | Delete after transfer | ⚠️ Partial | `delete_after` stored, not wired |
+| `--delete-excluded` | Also delete excluded files | ⚠️ Partial | `delete_excluded` stored, not wired |
+| `--max-delete=NUM` | Max files to delete | ⚠️ Partial | `max_delete` stored, not wired |
+| `--ignore-errors` | Delete even with I/O errors | ❌ Not Implemented | |
+| `--force` | Force deletion of non-empty dirs | ❌ Not Implemented | |
+| `--prune-empty-dirs` | Prune empty dir chains | ⚠️ Partial | `prune_empty_dirs` stored, not wired |
+
+## 8. Metadata Preservation
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-M`, `--preserve` | Preserve file metadata | ✅ Implemented | Mode, uid, gid, mtime |
+| `-p`, `--perms` | Preserve permissions | 🔀 Alt Arg | `-p` means SSH port; permissions preserved via `-M`/`--preserve` |
+| `-o`, `--owner` | Preserve owner | ✅ Implemented | Part of -M |
+| `-g`, `--group` | Preserve group | ✅ Implemented | Part of -M |
+| `-t`, `--times` | Preserve modification times | ✅ Implemented | Part of -M |
+| `-E`, `--executability` | Preserve executability | ❌ Not Implemented | |
+| `--chmod=CHMOD` | Affect file permissions | ❌ Not Implemented | |
+| `-A`, `--acls` | Preserve ACLs | ⚠️ Partial | `preserve_acls` stored, not wired |
+| `-X`, `--xattrs` | Preserve extended attributes | ⚠️ Partial | `preserve_xattrs` stored, not wired |
+| `-H`, `--hard-links` | Preserve hard links | ⚠️ Partial | `preserve_hard_links` stored, not wired |
+| `-D` | Same as --devices --specials | 🔀 Alt Arg | Maps to --devices only (no --specials) |
+| `--devices` | Preserve device files | ⚠️ Partial | `preserve_devices` stored, not wired |
+| `--specials` | Preserve special files | ❌ Not Implemented | |
+| `--copy-devices` | Copy device contents as file | ❌ Not Implemented | |
+| `--write-devices` | Write to devices as files | ❌ Not Implemented | |
+| `-U`, `--atimes` | Preserve access times | ❌ Not Implemented | |
+| `-N`, `--crtimes` | Preserve create times | ❌ Not Implemented | |
+| `-O`, `--omit-dir-times` | Omit dirs from --times | ❌ Not Implemented | |
+| `-J`, `--omit-link-times` | Omit symlinks from --times | ❌ Not Implemented | |
+| `--super` | Receiver attempts super-user activities | ❌ Not Implemented | |
+| `--fake-super` | Store/recover privileged attrs via xattrs | ❌ Not Implemented | |
+
+## 9. Symlink Handling
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-l`, `--links` | Copy symlinks as symlinks | ⚠️ Partial | Scanner includes symlinks; target path not transmitted |
+| `-L`, `--copy-links` | Transform symlink to referent | ✅ Implemented | `copy_links` config field |
+| `--copy-unsafe-links` | Transform unsafe symlinks | ✅ Implemented | `copy_unsafe_links` config field |
+| `--safe-links` | Ignore symlinks outside tree | ✅ Implemented | `safe_links` config field |
+| `--munge-links` | Munge symlinks for safety | ❌ Not Implemented | |
+| `-k`, `--copy-dirlinks` | Transform symlink to dir | ❌ Not Implemented | |
+| `-K`, `--keep-dirlinks` | Treat symlinked dir as dir | ❌ Not Implemented | |
+
+## 10. Sparse & Device
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-S`, `--sparse` | Sparse block handling | ✅ Implemented | `preserve_sparse` config field |
+| `--preallocate` | Allocate dest files before writing | ❌ Not Implemented | |
+
+## 11. Checksum & Comparison
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-c`, `--checksum` | Skip based on checksum | 🔀 Alt Arg | `-c` means compression; `--checksum` stored and forwarded to scanner |
+| `--checksum-choice=STR` | Choose checksum algorithm | ❌ Not Implemented | xxHash used internally |
+| `--compare-dest=DIR` | Compare dest files relative to DIR | ⚠️ Partial | `compare_dest` stored, not wired |
+| `--copy-dest=DIR` | Include copies of unchanged files | ⚠️ Partial | `copy_dest` stored, not wired |
+| `--link-dest=DIR` | Hardlink to files when unchanged | ⚠️ Partial | `link_dest` stored, not wired |
+| `--fuzzy`, `--no-fuzzy` | Find similar file for basis | ❌ Not Implemented | |
+
+## 12. Compression
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-z`, `--compress` | Compress file data | 🔀 Alt Arg | Always uses zstd (rsync supports multiple algorithms) |
+| `--compress-choice=STR` | Choose compression algorithm | ⚠️ Partial | `compress_choice` stored, always zstd |
+| `--compress-level=NUM` | Set compression level | ✅ Implemented | 1-22, default 5 |
+| `--compress-threads=NUM` | Set compression threads | ❌ Not Implemented | |
+| `--skip-compress=LIST` | Skip compress for suffixes | ❌ Not Implemented | Internal skip for hardcoded types; not user-configurable |
+
+## 13. Connectivity
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `-e`, `--rsh=COMMAND` | Remote shell to use | ✅ Implemented | SSH transport support |
+| `--rsync-path=PROGRAM` | rsync binary on remote | ✅ Implemented | `rsync_path` config field |
+| `--port=PORT` | Alternate daemon port | ✅ Implemented | `server_port` config field |
+| `--sockopts=OPTIONS` | Custom TCP options | ❌ Not Implemented | |
+| `--blocking-io` | Use blocking I/O for remote shell | ❌ Not Implemented | |
+| `--outbuf=N\|L\|B` | Set output buffering | ❌ Not Implemented | |
+| `--address=ADDRESS` | Bind address for outgoing socket | ✅ Implemented | `address` config field |
+| `-4`, `--ipv4` | Prefer IPv4 | ✅ Implemented | `ipv4` config field |
+| `-6`, `--ipv6` | Prefer IPv6 | ✅ Implemented | `ipv6` config field |
+
+## 14. Daemon Mode
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--daemon` | Run as rsync daemon | ⚠️ Partial | `daemon` stored, not wired |
+| `--config=FILE` | Alternate rsyncd.conf file | ⚠️ Partial | `daemon_config` stored, not wired |
+| `--dparam=OVERRIDE` | Override global daemon config | ❌ Not Implemented | |
+| `--no-detach` | Don't detach from parent | ❌ Not Implemented | |
+| `--password-file=FILE` | Read daemon password from file | ❌ Not Implemented | |
+| `--early-input=FILE` | Use FILE for daemon early exec | ❌ Not Implemented | |
+
+## 15. Safety & Security
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| Path escape detection | Ensure files stay within root | ✅ Implemented | `has_path_traversal()` + realpath |
+| Symlink-safe delete | Skip symlinks in delete walk | ✅ Implemented | `delete_extras_walk()` |
+| Protocol version check | Verify compatible versions | ✅ Implemented | `config_receive()` |
+| Max data/string/chunk sizes | Prevent OOM attacks | ✅ Implemented | Per-message limits |
+| Per-connection memory limit | 1GB per connection | ✅ Implemented | `MAX_CONNECTION_MEMORY` |
+| `--trust-sender` | Trust remote sender's file list | ❌ Not Implemented | |
+| `--old-args` | Disable modern arg protection | ❌ Not Implemented | |
+| `--ignore-missing-args` | Ignore missing source args | ❌ Not Implemented | |
+| `--delete-missing-args` | Delete missing source args | ❌ Not Implemented | |
+
+## 16. Batch Operations
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--write-batch=FILE` | Write batched update to file | ❌ Not Implemented | |
+| `--only-write-batch=FILE` | Write batch without updating dest | ❌ Not Implemented | |
+| `--read-batch=FILE` | Read batched update from file | ❌ Not Implemented | |
+
+## 17. Advanced
+
+| Flag | Rsync Description | FastSync Status | Notes |
+|------|-------------------|-----------------|-------|
+| `--stop-after=MINS` | Stop after N minutes | ❌ Not Implemented | |
+| `--stop-at=TIME` | Stop at specified time | ❌ Not Implemented | |
+| `--fsync` | Fsync every written file | ❌ Not Implemented | |
+| `--protocol=NUM` | Force older protocol version | ❌ Not Implemented | |
+| `--iconv=CONVERT_SPEC` | Charset conversion | ❌ Not Implemented | |
+| `--checksum-seed=NUM` | Set checksum seed | ❌ Not Implemented | |
+| `-s`, `--secluded-args` | Use protocol to send args | ❌ Not Implemented | |
+| `--no-OPTION` | Turn off implied option | ❌ Not Implemented | |
+
+---
+
+## Recommendations: Top Features to Implement Next
+
+Ranked by user demand, implementation complexity, and interoperability impact:
+
+| Priority | Feature | Effort | Impact |
+|----------|---------|--------|--------|
+| 1 | `--whole-file` / `-W` | Low | High — users expect opt-out of delta |
+| 2 | `--ignore-times` / `-I` | Low | Medium — useful for forcing re-transfer |
+| 3 | `--size-only` | Low | Medium — common migration scenario |
+| 4 | `--existing` / `--ignore-existing` | Low | Medium — common sync patterns |
+| 5 | `--remove-source-files` | Low | High — common for moves/backup |
+| 6 | `--delete-during` | Medium | High — performance improvement |
+| 7 | `--delay-updates` | Medium | High — atomic updates |
+| 8 | `--chmod` | Low | Medium — permission flexibility |
+| 9 | `--executability` / `-E` | Low | Low — simple flag |
+| 10 | `--skip-compress` | Low | Medium — performance tuning |
+
+---
+
+## FastSync-Specific Features (Not in rsync)
+
+| Feature | Description |
+|---------|-------------|
+| `-m` | Multithreaded pipeline (scanner/loader/sender) |
+| `-s` | Chunk serialization mode |
+| `-f` / `--sendfile` | Zero-copy sendfile() syscall (TCP only) |
+| `-c [level]` | zstd compression level (1-22) |
+| `--chunk-size` | Configurable chunk size |
+| `--queue-size` | Pipeline queue capacity |
+| `--tls` | TLS encryption (mutual auth) |
+| `--fastsync-server-path` | Path to fastsync-server binary |
+| `--server-host` / `--server-port` | Direct TCP connection |
+| Incremental sync | Skip unchanged files (size+mtime) |
+| Delta transfer | Block-level delta for changed files |

src/client/client_cli.c

@@ -12,6 +12,7 @@
 #include <stdlib.h>
 #include <string.h>
 
+#ifndef FASTSYNC_TEST_BUILD
 /* Parse environment variables for source/destination directories and save-to-disk flag. */
 static void parse_environment(const char** out_env_source, const char** out_env_dest,
                               bool* out_save_to_disk) {
@@ -22,6 +23,7 @@ static void parse_environment(const char** out_env_source, const char** out_env_
   if (env_save && (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0))
     *out_save_to_disk = true;
 }
+#endif
 
 /* Parse a string as a positive integer, returning true on success. */
 static bool parse_positive_int(const char* s, int* out_val) {
@@ -49,16 +51,50 @@ static bool parse_nonneg_int(const char* s, int* out_val) {
   return true;
 }
 
+/* Duplicate a string argument into *dest, freeing the old value. Returns 0 on success, -1 on
+ * failure. */
+static int set_string_option(char** dest, const char* value, const char* option_name) {
+  char* dup = str_dup(value);
+  if (!dup) {
+    fprintf(stderr, "Error: memory allocation failed for %s\n", option_name);
+    return -1;
+  }
+  free(*dest);
+  *dest = dup;
+  return 0;
+}
+
+/* Parse a string as a positive integer into *dest. Returns 0 on success, -1 on error. */
+static int set_positive_int_option(int* dest, const char* value, const char* option_name) {
+  if (!parse_positive_int(value, dest)) {
+    fprintf(stderr, "Error: %s must be a positive integer\n", option_name);
+    return -1;
+  }
+  return 0;
+}
+
+/* Parse a string as a non-negative integer into *dest. Returns 0 on success, -1 on error. */
+static int set_nonneg_int_option(int* dest, const char* value, const char* option_name) {
+  if (!parse_nonneg_int(value, dest)) {
+    fprintf(stderr, "Error: %s must be a non-negative integer\n", option_name);
+    return -1;
+  }
+  return 0;
+}
+
 static void print_usage(void);
 static int read_patterns_from_file(const char* filepath, char*** patterns, int* count);
 
 /* Parse CLI arguments into config. Returns 0 on success, -1 on error, 1 for help/clean-exit. */
-static int parse_args(Config* config, int argc, char* argv[], int* positional_args,
+int parse_args(Config* config, int argc, char* argv[], int* positional_args,
                int* positional_count) {
   for (int i = 1; i < argc; i++) {
     if (strcmp(argv[i], "--help") == 0) {
       print_usage();
       return 1;
+    } else if (strcmp(argv[i], "-V") == 0 || strcmp(argv[i], "--version") == 0) {
+      printf("fastsync version %s\n", PROTOCOL_VERSION);
+      return 1;
     } else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--archive") == 0) {
       config->use_compression = true;
       config->use_multithreading = true;
@@ -71,6 +107,10 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
         fprintf(stderr, "Error: invalid --port/-p value: %s\n", argv[i]);
         return -1;
       }
+      if (config->ssh_port > 65535) {
+        fprintf(stderr, "Error: SSH port must be 1-65535\n");
+        return -1;
+      }
     } else if (strcmp(argv[i], "--delete") == 0) {
       config->use_delete = true;
     } else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
@@ -100,21 +140,47 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
       }
       config->include_patterns[config->include_count++] = dup;
     } else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) {
-      config->max_size = strtoull(argv[++i], NULL, 10);
+      char* end;
+      errno = 0;
+      unsigned long long val = strtoull(argv[++i], &end, 10);
+      if (errno != 0 || *end != '\0') {
+        fprintf(stderr, "Error: --max-size must be a non-negative integer\n");
+        return -1;
+      }
+      config->max_size = val;
     } else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) {
-      config->min_size = strtoull(argv[++i], NULL, 10);
+      char* end;
+      errno = 0;
+      unsigned long long val = strtoull(argv[++i], &end, 10);
+      if (errno != 0 || *end != '\0') {
+        fprintf(stderr, "Error: --min-size must be a non-negative integer\n");
+        return -1;
+      }
+      config->min_size = val;
     } else if (strcmp(argv[i], "--incremental") == 0) {
       config->use_incremental = true;
     } else if (strcmp(argv[i], "--delta") == 0) {
       config->use_delta = true;
     } else if (strcmp(argv[i], "--delta-block") == 0 && i + 1 < argc) {
-      unsigned long long val = strtoull(argv[++i], NULL, 10);
+      char* end;
+      errno = 0;
+      unsigned long long val = strtoull(argv[++i], &end, 10);
+      if (errno != 0 || *end != '\0') {
+        fprintf(stderr, "Error: --delta-block must be a positive integer\n");
+        return -1;
+      }
       if (val >= DELTA_BLOCK_SIZE_MIN && val <= DELTA_BLOCK_SIZE_MAX)
         config->delta_block_size = (uint32_t)val;
       else
         fprintf(stderr, "Warning: --delta-block value %llu out of range, using default\n", val);
     } else if (strcmp(argv[i], "--delta-max") == 0 && i + 1 < argc) {
-      unsigned long long val = strtoull(argv[++i], NULL, 10);
+      char* end;
+      errno = 0;
+      unsigned long long val = strtoull(argv[++i], &end, 10);
+      if (errno != 0 || *end != '\0') {
+        fprintf(stderr, "Error: --delta-max must be a positive integer\n");
+        return -1;
+      }
       if (val >= DELTA_MIN_FILE_SIZE)
         config->delta_max_file_size = val;
       else
@@ -126,27 +192,21 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
         char* end_ptr;
         long level = strtol(argv[i + 1], &end_ptr, 10);
         if (*end_ptr == '\0') {
+          if (level < 1 || level > 22) {
+            fprintf(stderr, "Error: compression level must be 1-22\n");
+            return -1;
+          }
           config->compression_level = (int)level;
           log_message(LOG_LEVEL_INFO, "Set Compression level to %ld", level);
           i++;
         }
       }
     } else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --source-dir\n");
+      if (set_string_option(&config->send_directory, argv[++i], "--source-dir") != 0)
         return -1;
-      }
-      free(config->send_directory);
-      config->send_directory = dup;
     } else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --dest-dir\n");
+      if (set_string_option(&config->receive_root_directory, argv[++i], "--dest-dir") != 0)
         return -1;
-      }
-      free(config->receive_root_directory);
-      config->receive_root_directory = dup;
     } else if (strcmp(argv[i], "--save-to-disk") == 0) {
       config->save_to_disk = true;
     } else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) {
@@ -162,18 +222,17 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
       config->use_chunk_serialization = true;
       log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization");
     } else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --server-host\n");
+      if (set_string_option(&config->server_host, argv[++i], "--server-host") != 0)
         return -1;
-      }
-      free(config->server_host);
-      config->server_host = dup;
     } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
       if (!parse_positive_int(argv[++i], &config->server_port)) {
         fprintf(stderr, "Error: invalid --server-port value: %s\n", argv[i]);
         return -1;
       }
+      if (config->server_port > 65535) {
+        fprintf(stderr, "Error: server port must be 1-65535\n");
+        return -1;
+      }
     } else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) {
       char* end;
       errno = 0;
@@ -191,70 +250,50 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "--progress") == 0) {
       config->show_progress = true;
     } else if (strcmp(argv[i], "--chunk-size") == 0 && i + 1 < argc) {
-      unsigned long long val = strtoull(argv[++i], NULL, 10);
-      if (val > 0)
+      char* end;
+      errno = 0;
+      unsigned long long val = strtoull(argv[++i], &end, 10);
+      if (errno != 0 || *end != '\0' || val == 0) {
+        fprintf(stderr, "Error: --chunk-size must be a positive integer\n");
+        return -1;
+      }
       config->chunk_size = val;
     } else if (strcmp(argv[i], "--tls") == 0) {
       config->use_tls = true;
     } else if (strcmp(argv[i], "--cert") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --cert\n");
+      if (set_string_option(&config->tls_cert, argv[++i], "--cert") != 0)
         return -1;
-      }
-      free(config->tls_cert);
-      config->tls_cert = dup;
     } else if (strcmp(argv[i], "--key") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --key\n");
+      if (set_string_option(&config->tls_key, argv[++i], "--key") != 0)
         return -1;
-      }
-      free(config->tls_key);
-      config->tls_key = dup;
     } else if (strcmp(argv[i], "--ca") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --ca\n");
+      if (set_string_option(&config->tls_ca, argv[++i], "--ca") != 0)
         return -1;
-      }
-      free(config->tls_ca);
-      config->tls_ca = dup;
     } else if (strcmp(argv[i], "--timeout") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_positive_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: --timeout must be a positive integer\n");
+      if (set_positive_int_option(&config->timeout, argv[++i], "--timeout") != 0)
         return -1;
-      }
-      config->timeout = val;
     } else if (strcmp(argv[i], "--contimeout") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_positive_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: --contimeout must be a positive integer\n");
+      if (set_positive_int_option(&config->contimeout, argv[++i], "--contimeout") != 0)
         return -1;
-      }
-      config->contimeout = val;
     } else if (strcmp(argv[i], "-q") == 0 || strcmp(argv[i], "--quiet") == 0 ||
                strcmp(argv[i], "--silent") == 0) {
       config->quiet = true;
     } else if (strcmp(argv[i], "--backup") == 0) {
       config->backup = true;
     } else if (strcmp(argv[i], "--backup-dir") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --backup-dir\n");
+      if (set_string_option(&config->backup_dir, argv[++i], "--backup-dir") != 0)
         return -1;
-      }
-      free(config->backup_dir);
-      config->backup_dir = dup;
     } else if (strcmp(argv[i], "--stats") == 0) {
       config->stats = true;
     } else if (strcmp(argv[i], "--max-depth") == 0 && i + 1 < argc) {
-      if (!parse_nonneg_int(argv[++i], &config->max_depth)) {
-        fprintf(stderr, "Error: --max-depth must be a non-negative integer\n");
+      if (set_nonneg_int_option(&config->max_depth, argv[++i], "--max-depth") != 0)
         return -1;
-      }
     } else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) {
+      if (config->log_file) {
+        fclose(config->log_file);
+        config->log_file = NULL;
+        log_set_file(NULL);
+      }
       FILE* lf = fopen(argv[++i], "a");
       if (!lf) {
         fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno));
@@ -263,12 +302,8 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
       config->log_file = lf;
       log_set_file(lf);
     } else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_positive_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: --queue-size must be a positive integer\n");
+      if (set_positive_int_option(&config->queue_size, argv[++i], "--queue-size") != 0)
         return -1;
-      }
-      config->queue_size = val;
     } else if (strcmp(argv[i], "--exclude-from") == 0 && i + 1 < argc) {
       if (read_patterns_from_file(argv[++i], &config->exclude_patterns, &config->exclude_count) !=
           0)
@@ -280,13 +315,9 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "--partial") == 0) {
       config->partial = true;
     } else if (strcmp(argv[i], "--fastsync-server-path") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --fastsync-server-path\n");
+      if (set_string_option(&config->fastsync_server_path, argv[++i], "--fastsync-server-path") !=
+          0)
         return -1;
-      }
-      free(config->fastsync_server_path);
-      config->fastsync_server_path = dup;
     } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
       set_log_level(LOG_LEVEL_DEBUG);
     } else if (strcmp(argv[i], "-l") == 0 || strcmp(argv[i], "--links") == 0) {
@@ -310,15 +341,14 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "-i") == 0 || strcmp(argv[i], "--itemize-changes") == 0) {
       config->itemize_changes = true;
     } else if (strcmp(argv[i], "--out-format") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->out_format, argv[++i], "--out-format") != 0)
         return -1;
-      free(config->out_format);
-      config->out_format = dup;
     } else if (strcmp(argv[i], "--info") == 0 && i + 1 < argc) {
-      config->info_level = atoi(argv[++i]);
+      if (set_nonneg_int_option(&config->info_level, argv[++i], "--info") != 0)
+        return -1;
     } else if (strcmp(argv[i], "--debug") == 0 && i + 1 < argc) {
-      config->debug_level = atoi(argv[++i]);
+      if (set_nonneg_int_option(&config->debug_level, argv[++i], "--debug") != 0)
+        return -1;
     } else if (strcmp(argv[i], "--list-only") == 0) {
       config->list_only = true;
     } else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--human-readable") == 0) {
@@ -336,12 +366,8 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "--delete-after") == 0) {
       config->delete_after = true;
     } else if (strcmp(argv[i], "--max-delete") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_nonneg_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: --max-delete must be a non-negative integer\n");
+      if (set_nonneg_int_option(&config->max_delete, argv[++i], "--max-delete") != 0)
         return -1;
-      }
-      config->max_delete = val;
     } else if (strcmp(argv[i], "--filter") == 0 && i + 1 < argc) {
       if (!config->filters)
         config->filters = array_list_create(free);
@@ -350,11 +376,8 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
         return -1;
       array_list_add(config->filters, dup);
     } else if (strcmp(argv[i], "--files-from") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->files_from, argv[++i], "--files-from") != 0)
         return -1;
-      free(config->files_from);
-      config->files_from = dup;
     } else if (strcmp(argv[i], "--cvs-exclude") == 0) {
       config->cvs_exclude = true;
     } else if (strcmp(argv[i], "--prune-empty-dirs") == 0) {
@@ -363,82 +386,44 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
       config->relative = true;
     } else if (strcmp(argv[i], "-e") == 0 || strcmp(argv[i], "--rsh") == 0) {
       if (i + 1 < argc) {
-        char* dup = str_dup(argv[++i]);
-        if (!dup)
+        if (set_string_option(&config->rsh_command, argv[++i], "-e/--rsh") != 0)
           return -1;
-        free(config->rsh_command);
-        config->rsh_command = dup;
       } else {
         fprintf(stderr, "Error: -e/--rsh requires a command argument\n");
         return -1;
       }
     } else if (strcmp(argv[i], "--rsync-path") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->rsync_path, argv[++i], "--rsync-path") != 0)
         return -1;
-      free(config->rsync_path);
-      config->rsync_path = dup;
     } else if (strcmp(argv[i], "--temp-dir") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->temp_dir, argv[++i], "--temp-dir") != 0)
         return -1;
-      free(config->temp_dir);
-      config->temp_dir = dup;
     } else if (strcmp(argv[i], "--compare-dest") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->compare_dest, argv[++i], "--compare-dest") != 0)
         return -1;
-      free(config->compare_dest);
-      config->compare_dest = dup;
     } else if (strcmp(argv[i], "--copy-dest") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->copy_dest, argv[++i], "--copy-dest") != 0)
         return -1;
-      free(config->copy_dest);
-      config->copy_dest = dup;
     } else if (strcmp(argv[i], "--link-dest") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->link_dest, argv[++i], "--link-dest") != 0)
         return -1;
-      free(config->link_dest);
-      config->link_dest = dup;
     } else if (strcmp(argv[i], "--partial-dir") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --partial-dir\n");
+      if (set_string_option(&config->partial_dir, argv[++i], "--partial-dir") != 0)
         return -1;
-      }
-      free(config->partial_dir);
-      config->partial_dir = dup;
     } else if (strcmp(argv[i], "--suffix") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup) {
-        fprintf(stderr, "Error: memory allocation failed for --suffix\n");
+      if (set_string_option(&config->suffix, argv[++i], "--suffix") != 0)
         return -1;
-      }
-      free(config->suffix);
-      config->suffix = dup;
     } else if (strcmp(argv[i], "--delete-before") == 0) {
       config->delete_before = true;
     } else if (strcmp(argv[i], "-T") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_positive_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: -T must be a positive integer\n");
+      if (set_positive_int_option(&config->timeout, argv[++i], "-T") != 0)
         return -1;
-      }
-      config->timeout = val;
     } else if (strcmp(argv[i], "--address") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->address, argv[++i], "--address") != 0)
         return -1;
-      free(config->address);
-      config->address = dup;
     } else if (strcmp(argv[i], "--bind-address") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->bind_address, argv[++i], "--bind-address") != 0)
         return -1;
-      free(config->bind_address);
-      config->bind_address = dup;
     } else if (strcmp(argv[i], "--ipv6") == 0) {
       config->ipv6 = true;
     } else if (strcmp(argv[i], "--ipv4") == 0) {
@@ -446,28 +431,22 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
     } else if (strcmp(argv[i], "--daemon") == 0) {
       config->daemon = true;
     } else if (strcmp(argv[i], "--config") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->daemon_config, argv[++i], "--config") != 0)
         return -1;
-      free(config->daemon_config);
-      config->daemon_config = dup;
     } else if (strcmp(argv[i], "--server") == 0) {
       config->server_mode = true;
     } else if (strcmp(argv[i], "--checksum") == 0) {
       config->checksum = true;
     } else if (strcmp(argv[i], "--compress-choice") == 0 && i + 1 < argc) {
-      char* dup = str_dup(argv[++i]);
-      if (!dup)
+      if (set_string_option(&config->compress_choice, argv[++i], "--compress-choice") != 0)
         return -1;
-      free(config->compress_choice);
-      config->compress_choice = dup;
     } else if (strcmp(argv[i], "--compress-level") == 0 && i + 1 < argc) {
-      int val;
-      if (!parse_positive_int(argv[++i], &val)) {
-        fprintf(stderr, "Error: --compress-level must be a positive integer\n");
+      if (set_positive_int_option(&config->compression_level, argv[++i], "--compress-level") != 0)
+        return -1;
+      if (config->compression_level < 1 || config->compression_level > 22) {
+        fprintf(stderr, "Error: --compress-level must be between 1 and 22\n");
         return -1;
       }
-      config->compression_level = val;
     } else if (argv[i][0] == '-') {
       fprintf(stderr, "Unknown option: %s\n", argv[i]);
       print_usage();
@@ -485,6 +464,7 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar
   return 0;
 }
 
+#ifndef FASTSYNC_TEST_BUILD
 /* Validate config after parsing. Returns true if valid. */
 static bool validate_config(const Config* config) {
   if (!config->send_directory || !config->receive_root_directory) {
@@ -525,6 +505,7 @@ static bool validate_config(const Config* config) {
   }
   return true;
 }
+#endif /* FASTSYNC_TEST_BUILD */
 
 static void print_usage(void) {
   printf("Usage:\n");
@@ -633,6 +614,7 @@ static void print_usage(void) {
   printf("  --copy-dest <dir>   Copy destination\n");
   printf("  --link-dest <dir>   Link destination\n");
   printf("  --help              Show this help\n");
+  printf("  -V, --version       Show version\n");
 }
 
 static int read_patterns_from_file(const char* filepath, char*** patterns, int* count) {
@@ -641,8 +623,10 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     fprintf(stderr, "Error: could not open pattern file '%s': %s\n", filepath, strerror(errno));
     return -1;
   }
-  char line[4096];
-  while (fgets(line, sizeof(line), fp)) {
+  char* line = NULL;
+  size_t line_size = 0;
+  ssize_t n;
+  while ((n = getline(&line, &line_size, fp)) != -1) {
     char* p = line;
     while (*p == ' ' || *p == '\t')
       p++;
@@ -656,6 +640,7 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     char** tmp = realloc(*patterns, (*count + 1) * sizeof(char*));
     if (!tmp) {
       fprintf(stderr, "Error: memory allocation failed for pattern file\n");
+      free(line);
       fclose(fp);
       return -1;
     }
@@ -663,15 +648,18 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
     char* dup = str_dup(p);
     if (!dup) {
       fprintf(stderr, "Error: memory allocation failed for pattern file\n");
+      free(line);
       fclose(fp);
       return -1;
     }
     (*patterns)[(*count)++] = dup;
   }
+  free(line);
   fclose(fp);
   return 0;
 }
 
+#ifndef FASTSYNC_TEST_BUILD
 int main(int argc, char* argv[]) {
   const char* env_source = NULL;
   const char* env_dest = NULL;
@@ -777,3 +765,4 @@ cleanup:
   }
   return exit_code;
 }
+#endif /* FASTSYNC_TEST_BUILD */

src/client/client_send.c

@@ -346,6 +346,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
     }
     if (send_chunk(client, current_chunk, context->config) != 0) {
       fprintf(stderr, "Error: unexpected error while sending chunk\n");
+      chunk_destroy(current_chunk);
       client_disconnect(client);
       client_delete(client);
       mtx_lock(&context->mutex_progress);
@@ -385,13 +386,31 @@ static int scan_directory_multithreaded(void* pipeline_context) {
         const char* p = current_chunk->items[i]->path;
         if (*p == '/')
           p++;
-        array_list_add(context->manifest, str_dup(p));
+        char* manifest_entry = str_dup(p);
+        if (!manifest_entry) {
+          log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry");
+          mtx_unlock(&context->mutex_scanner);
+          context->cancelled = true;
+          cnd_broadcast(&context->condition_not_full_scanner);
+          cnd_broadcast(&context->condition_not_empty_scanner);
+          parallel_scanner_destroy(scanner);
+          return thrd_error;
+        }
+        array_list_add(context->manifest, manifest_entry);
       }
       mtx_unlock(&context->mutex_scanner);
     }
-    queue_enqueue_multithreaded(context->queue_scanner, current_chunk, &context->mutex_scanner,
-                                &context->condition_not_empty_scanner,
-                                &context->condition_not_full_scanner);
+    if (!queue_enqueue_multithreaded_cancel(
+            context->queue_scanner, current_chunk, &context->mutex_scanner,
+            &context->condition_not_empty_scanner, &context->condition_not_full_scanner,
+            &context->cancelled)) {
+      chunk_destroy(current_chunk);
+      context->cancelled = true;
+      cnd_broadcast(&context->condition_not_full_scanner);
+      cnd_broadcast(&context->condition_not_empty_scanner);
+      parallel_scanner_destroy(scanner);
+      return thrd_error;
+    }
   }
   mtx_lock(&context->mutex_scanner);
   context->scanner_done = true;
@@ -427,9 +446,16 @@ static int load_files_multithreaded(void* pipeline_context) {
         }
       }
     }
-    queue_enqueue_multithreaded(context->queue_loader, chunk, &context->mutex_loader,
+    if (!queue_enqueue_multithreaded_cancel(context->queue_loader, chunk, &context->mutex_loader,
                                             &context->condition_not_empty_loader,
-                                &context->condition_not_full_loader);
+                                            &context->condition_not_full_loader,
+                                            &context->cancelled)) {
+      chunk_destroy(chunk);
+      context->cancelled = true;
+      cnd_broadcast(&context->condition_not_full_loader);
+      cnd_broadcast(&context->condition_not_empty_loader);
+      return thrd_error;
+    }
   }
 }
 
@@ -487,16 +513,20 @@ int send_files(Config* config) {
     client = client_create();
     if (!client || !client_connect_tls(client, config->server_host, config->server_port,
                                        config->tls_cert, config->tls_key, config->tls_ca)) {
-      if (client)
+      if (client) {
+        client_disconnect(client);
         client_delete(client);
+      }
       fprintf(stderr, "Error: could not connect to server via TLS\n");
       return 1;
     }
   } else {
     client = client_create();
     if (!client || !client_connect(client, config->server_host, config->server_port)) {
-      if (client)
+      if (client) {
+        client_disconnect(client);
         client_delete(client);
+      }
       fprintf(stderr, "Error: could not connect to server\n");
       return 1;
     }
@@ -526,7 +556,17 @@ int send_files(Config* config) {
         const char* p = current_chunk->items[i]->path;
         if (*p == '/')
           p++;
-        array_list_add(manifest, str_dup(p));
+        char* manifest_entry = str_dup(p);
+        if (!manifest_entry) {
+          log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry");
+          chunk_destroy(current_chunk);
+          array_list_delete(manifest);
+          directory_scanner_destroy(scanner);
+          client_disconnect(client);
+          client_delete(client);
+          return 1;
+        }
+        array_list_add(manifest, manifest_entry);
       }
     }
     if (!config->use_sendfile) {
@@ -625,17 +665,42 @@ int send_files_multithreaded(Config* config) {
   if (config->use_delete)
     context->manifest = array_list_create(free);
 
-  thrd_t scanner, loader, sender, progress;
-  if (thrd_create(&scanner, scan_directory_multithreaded, context) != thrd_success ||
-      thrd_create(&loader, load_files_multithreaded, context) != thrd_success ||
-      thrd_create(&sender, send_chunks_multithreaded, context) != thrd_success) {
+  thrd_t scanner, loader, sender;
+  bool scanner_created = false;
+  bool loader_created = false;
+  bool sender_created = false;
+
+  scanner_created = (thrd_create(&scanner, scan_directory_multithreaded, context) == thrd_success);
+  if (scanner_created)
+    loader_created = (thrd_create(&loader, load_files_multithreaded, context) == thrd_success);
+  if (scanner_created && loader_created)
+    sender_created = (thrd_create(&sender, send_chunks_multithreaded, context) == thrd_success);
+
+  if (!scanner_created || !loader_created || !sender_created) {
     perror("Error creating threads.\n");
+    context->cancelled = true;
+    context->scanner_done = true;
+    context->loader_done = true;
+    context->sender_done = true;
+    cnd_broadcast(&context->condition_not_full_scanner);
+    cnd_broadcast(&context->condition_not_empty_scanner);
+    cnd_broadcast(&context->condition_not_full_loader);
+    cnd_broadcast(&context->condition_not_empty_loader);
+    if (sender_created)
+      thrd_join(sender, NULL);
+    if (loader_created)
+      thrd_join(loader, NULL);
+    if (scanner_created)
+      thrd_join(scanner, NULL);
     pipeline_context_sender_destroy(context);
     return 1;
   }
 
+  thrd_t progress;
+  bool progress_created = false;
   if (config->show_progress) {
-    if (thrd_create(&progress, progress_thread_fn, context) != thrd_success) {
+    progress_created = (thrd_create(&progress, progress_thread_fn, context) == thrd_success);
+    if (!progress_created) {
       perror("Error creating progress thread.\n");
       /* Non-fatal; continue without progress reporting */
     }
@@ -646,7 +711,7 @@ int send_files_multithreaded(Config* config) {
   thrd_join(loader, NULL);
   thrd_join(sender, &sender_result);
 
-  if (config->show_progress) {
+  if (progress_created) {
     /* Signal progress thread to exit if it hasn't already */
     mtx_lock(&context->mutex_progress);
     context->sender_done = true;

src/client/scanner.c

@@ -27,10 +27,14 @@ static void dir_entry_destroy(void* item) {
 
 static DirEntry* dir_entry_create(const char* path, int depth) {
   DirEntry* de = malloc(sizeof(DirEntry));
-  if (de) {
+  if (!de)
+    return NULL;
   de->path = str_dup(path);
-    de->depth = depth;
+  if (!de->path) {
+    free(de);
+    return NULL;
   }
+  de->depth = depth;
   return de;
 }
 
@@ -62,7 +66,18 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_
   scanner->safe_links = safe_links;
   scanner->copy_unsafe_links = copy_unsafe_links;
   scanner->checksum = checksum;
-  queue_enqueue(scanner->directories, dir_entry_create(root_directory, 0));
+  DirEntry* root = dir_entry_create(root_directory, 0);
+  if (!root) {
+    queue_destroy(scanner->directories);
+    free(scanner);
+    return NULL;
+  }
+  if (!queue_enqueue(scanner->directories, root)) {
+    dir_entry_destroy(root);
+    queue_destroy(scanner->directories);
+    free(scanner);
+    return NULL;
+  }
   return scanner;
 }
 
@@ -323,9 +338,28 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata
     free(ps);
     return NULL;
   }
-  if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success ||
-      cnd_init(&ps->result_not_empty) != thrd_success ||
-      cnd_init(&ps->result_not_full) != thrd_success) {
+  int init = 0;
+  bool ok = true;
+  if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success)
+    ok = false;
+  if (ok) {
+    init++;
+    if (cnd_init(&ps->result_not_empty) != thrd_success)
+      ok = false;
+  }
+  if (ok) {
+    // cppcheck-suppress unreadVariable
+    init++;
+    if (cnd_init(&ps->result_not_full) != thrd_success)
+      ok = false;
+  }
+  if (!ok) {
+    if (init >= 3)
+      cnd_destroy(&ps->result_not_full);
+    if (init >= 2)
+      cnd_destroy(&ps->result_not_empty);
+    if (init >= 1)
+      mtx_destroy(&ps->result_mutex);
     queue_destroy(ps->result_queue);
     free(ps);
     return NULL;
@@ -564,6 +598,7 @@ void parallel_scanner_destroy(ParallelScanner* ps) {
     return;
   ps->done = true;
   cnd_signal(&ps->result_not_empty);
+  cnd_broadcast(&ps->result_not_full);
   for (int i = 0; i < ps->num_threads; i++)
     thrd_join(ps->threads[i], NULL);
   free(ps->threads);

src/server/server.c

@@ -135,11 +135,27 @@ void handler(int file_descriptor) {
       return;
     }
     thrd_t receiver, writer;
-    if (thrd_create(&receiver, receive_thread, context) != thrd_success ||
-        thrd_create(&writer, write_thread, context) != thrd_success) {
+    bool receiver_created = false;
+    bool writer_created = false;
+    receiver_created = (thrd_create(&receiver, receive_thread, context) == thrd_success);
+    if (receiver_created)
+      writer_created = (thrd_create(&writer, write_thread, context) == thrd_success);
+    if (!receiver_created || !writer_created) {
       perror("Error creating Threads");
-      pipeline_context_receiver_destroy(context);
+      if (receiver_created) {
+        mtx_lock(&context->mutex);
+        context->cancelled = true;
+        cnd_broadcast(&context->condition_not_full);
+        cnd_broadcast(&context->condition_not_empty);
+        mtx_unlock(&context->mutex);
         close(file_descriptor);
+        thrd_join(receiver, NULL);
+      } else {
+        close(file_descriptor);
+      }
+      if (writer_created)
+        thrd_join(writer, NULL);
+      pipeline_context_receiver_destroy(context);
       return;
     }
     thrd_join(receiver, NULL);

src/shared/compression.c

@@ -1,12 +1,13 @@
 #include "compression.h"
 #include "data.h"
 #include "log.h"
-#include "stdlib.h"
-#include "string.h"
+#include <stdlib.h>
+#include <string.h>
 #include <strings.h>
-#include "zstd.h"
+#include <zstd.h>
 
 #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024)
+#define MAX_DECOMPRESSED_SIZE (100ULL * 1024 * 1024) /* 100 MB hard ceiling */
 
 static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv",
                                                     ".zip", ".gz",   ".xz",  ".zst", NULL};
@@ -84,6 +85,13 @@ Data* data_decompress(Data* compressed_data) {
     dst_size = compressed_data->size * 3;
     if (dst_size < INITIAL_DECOMPRESS_BUF_SIZE)
       dst_size = INITIAL_DECOMPRESS_BUF_SIZE;
+    if (dst_size > MAX_DECOMPRESSED_SIZE)
+      dst_size = MAX_DECOMPRESSED_SIZE;
+  }
+  if (dst_size > MAX_DECOMPRESSED_SIZE) {
+    log_message(LOG_LEVEL_ERROR, "Declared decompressed size exceeds %llu bytes",
+                (unsigned long long)MAX_DECOMPRESSED_SIZE);
+    return NULL;
   }
 
   ZSTD_DCtx* dctx = ZSTD_createDCtx();
@@ -113,7 +121,16 @@ Data* data_decompress(Data* compressed_data) {
       return NULL;
     }
     if (ret > 0 && output.pos == output.size) {
+      if (buf_size >= MAX_DECOMPRESSED_SIZE) {
+        log_message(LOG_LEVEL_ERROR, "Decompressed data exceeds %llu bytes",
+                    (unsigned long long)MAX_DECOMPRESSED_SIZE);
+        ZSTD_freeDCtx(dctx);
+        data_destroy(uncompressed_data);
+        return NULL;
+      }
       buf_size *= 2;
+      if (buf_size > MAX_DECOMPRESSED_SIZE)
+        buf_size = MAX_DECOMPRESSED_SIZE;
       void* new_data = realloc(uncompressed_data->data, buf_size);
       if (!new_data) {
         log_message(LOG_LEVEL_ERROR, "Failed to grow decompression buffer");

src/shared/config.c

@@ -8,10 +8,7 @@
 #include <stdlib.h>
 #include <string.h>
 
-Config* config_create(void) {
-  Config* config = malloc(sizeof(Config));
-  if (!config)
-    return NULL;
+static void config_set_defaults(Config* config) {
   config->version = str_dup(PROTOCOL_VERSION);
   config->send_directory = NULL;
   config->receive_root_directory = NULL;
@@ -101,6 +98,13 @@ Config* config_create(void) {
   config->server_mode = false;
   config->checksum = false;
   config->compress_choice = NULL;
+}
+
+Config* config_create(void) {
+  Config* config = malloc(sizeof(Config));
+  if (!config)
+    return NULL;
+  config_set_defaults(config);
   return config;
 }
 
@@ -131,6 +135,8 @@ void config_parse_ssh_dest(Config* config) {
 }
 
 void config_delete(Config* config) {
+  if (config == NULL)
+    return;
   free(config->version);
   free(config->send_directory);
   free(config->receive_root_directory);
@@ -167,6 +173,16 @@ 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
+ */
 bool config_send(int file_descriptor, const Config* config) {
   if (!send_str(file_descriptor, config->version))
     return false;
@@ -196,7 +212,7 @@ bool config_send(int file_descriptor, const Config* config) {
     return false;
   if (!send_int(file_descriptor, config->use_delta))
     return false;
-  if (!send_int(file_descriptor, (int)config->delta_block_size))
+  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;
@@ -264,13 +280,16 @@ bool config_send(int file_descriptor, const Config* config) {
   return true;
 }
 
+/* Wire format order: see the comment above config_send. */
 Config* config_receive(int file_descriptor) {
   Config* config = (Config*)malloc(sizeof(Config));
   if (config == NULL)
     return NULL;
-  memset(config, 0, sizeof(*config));
+  config_set_defaults(config);
+  free(config->version);
   config->version = receive_str(file_descriptor);
   if (!config->version) {
+    free(config->server_host);
     free(config);
     return NULL;
   }
@@ -278,6 +297,7 @@ Config* config_receive(int file_descriptor) {
     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;
@@ -285,6 +305,7 @@ Config* config_receive(int file_descriptor) {
   config->send_directory = receive_str(file_descriptor);
   if (!config->send_directory) {
     free(config->version);
+    free(config->server_host);
     free(config);
     return NULL;
   }
@@ -292,6 +313,7 @@ Config* config_receive(int file_descriptor) {
   if (!config->receive_root_directory) {
     free(config->version);
     free(config->send_directory);
+    free(config->server_host);
     free(config);
     return NULL;
   }
@@ -328,67 +350,10 @@ Config* config_receive(int file_descriptor) {
   if (!receive_int(file_descriptor, &tmp))
     goto error;
   config->use_delta = tmp;
-  if (!receive_int(file_descriptor, &tmp))
+  if (!receive_n_data(file_descriptor, &config->delta_block_size, sizeof(config->delta_block_size)))
     goto error;
-  config->delta_block_size = (uint32_t)tmp;
   if (!receive_n_data(file_descriptor, &config->delta_max_file_size, sizeof(unsigned long long)))
     goto error;
-  config->show_progress = false;
-  config->dry_run = false;
-  config->ssh_port = 22;
-  config->transport = TRANSPORT_TCP;
-  config->ssh_destination = NULL;
-  config->fastsync_server_path = NULL;
-  config->exclude_patterns = NULL;
-  config->exclude_count = 0;
-  config->include_patterns = NULL;
-  config->include_count = 0;
-  config->max_size = 0;
-  config->min_size = 0;
-  config->use_tls = false;
-  config->tls_cert = NULL;
-  config->tls_key = NULL;
-  config->tls_ca = NULL;
-  config->timeout = 30;
-  config->contimeout = 10;
-  config->quiet = false;
-  config->stats = false;
-  config->max_depth = 0;
-  config->log_file = NULL;
-  config->queue_size = 100;
-  config->follow_symlinks = false;
-  config->copy_links = false;
-  config->safe_links = false;
-  config->copy_unsafe_links = false;
-  config->preserve_hard_links = false;
-  config->preserve_acls = false;
-  config->preserve_xattrs = false;
-  config->preserve_devices = false;
-  config->preserve_sparse = false;
-  config->itemize_changes = false;
-  config->out_format = NULL;
-  config->info_level = 0;
-  config->debug_level = 0;
-  config->list_only = false;
-  config->human_readable = false;
-  config->update = false;
-  config->inplace = false;
-  config->append = false;
-  config->append_verify = false;
-  config->delete_excluded = false;
-  config->delete_after = false;
-  config->max_delete = 0;
-  config->filters = NULL;
-  config->files_from = NULL;
-  config->cvs_exclude = false;
-  config->prune_empty_dirs = false;
-  config->relative = false;
-  config->rsh_command = NULL;
-  config->rsync_path = NULL;
-  config->temp_dir = NULL;
-  config->compare_dest = NULL;
-  config->copy_dest = NULL;
-  config->link_dest = NULL;
   if (!receive_int(file_descriptor, &tmp))
     goto error;
   config->backup = tmp;
@@ -469,15 +434,6 @@ Config* config_receive(int file_descriptor) {
   config->compress_choice = receive_str(file_descriptor);
   if (config->compress_choice == NULL)
     goto error;
-  config->address = NULL;
-  config->bind_address = NULL;
-  config->ipv6 = false;
-  config->ipv4 = false;
-  config->daemon = false;
-  config->daemon_config = NULL;
-  config->server_mode = false;
-  config->server_host = str_dup("127.0.0.1");
-  config->server_port = 8080;
   if (!send_status(file_descriptor, STATUS_OK))
     goto error;
   return config;

src/shared/config.h

@@ -128,7 +128,8 @@ typedef struct Config {
   char* compress_choice;
 } Config;
 
-#define PROTOCOL_VERSION "1.3.0"
+/* This version must be bumped whenever config_send / config_receive wire format changes. */
+#define PROTOCOL_VERSION "2.1.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 
 Config* config_create(void);

src/shared/delta.c

@@ -388,6 +388,10 @@ Delta* delta_deserialize(const Data* data) {
 
     if (type == DELTA_OP_BLOCK_MATCH) {
       if (pos + sizeof(uint32_t) * 3 > data->size) {
+        for (uint32_t k = 0; k < i; k++) {
+          if (delta->instructions[k].type == DELTA_INSTR_LITERAL)
+            free(delta->instructions[k].literal.data);
+        }
         free(delta->instructions);
         free(delta);
         return NULL;
@@ -426,6 +430,11 @@ Delta* delta_deserialize(const Data* data) {
       }
       delta->instructions[i].literal.data = malloc(lit_len);
       if (!delta->instructions[i].literal.data) {
+        log_message(LOG_LEVEL_ERROR, "Failed to allocate %u bytes for literal data", lit_len);
+        for (uint32_t k = 0; k < i; k++) {
+          if (delta->instructions[k].type == DELTA_INSTR_LITERAL)
+            free(delta->instructions[k].literal.data);
+        }
         free(delta->instructions);
         free(delta);
         return NULL;

src/shared/file.c

@@ -16,7 +16,6 @@
 #include "config.h"
 #include "data.h"
 #include "file.h"
-#include "log.h"
 #include "metadata.h"
 #include "protocol.h"
 #include "utils.h"
@@ -96,6 +95,9 @@ bool file_load_data(File* file) {
   size_t bytes_read = file_content_to_buffer(file);
   if (bytes_read != file->data->size) {
     log_message(LOG_LEVEL_ERROR, "Did not read expected amount of bytes from file");
+    free(file->data->data);
+    file->data->data = NULL;
+    file->data->size = 0;
     return false;
   }
   return true;

src/shared/log.c

@@ -18,11 +18,15 @@ void log_set_file(FILE* fp) {
 void log_message(LogLevel log_level, const char* format, ...) {
   if (log_level < current_log_level)
     return;
+  if (log_level < 0 || log_level >= (int)(sizeof(log_level_strings) / sizeof(log_level_strings[0])))
+    return;
   time_t now = time(NULL);
-  const struct tm* t = localtime(&now);
+  struct tm t;
+  if (!localtime_r(&now, &t))
+    return;
 
-  fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1,
-          t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]);
+  fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t.tm_year + 1900, t.tm_mon + 1, t.tm_mday,
+          t.tm_hour, t.tm_min, t.tm_sec, log_level_strings[log_level]);
 
   va_list args;
   va_start(args, format);
@@ -31,8 +35,8 @@ void log_message(LogLevel log_level, const char* format, ...) {
   fprintf(stderr, "\n");
 
   if (log_fp) {
-    fprintf(log_fp, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1,
-            t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]);
+    fprintf(log_fp, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t.tm_year + 1900, t.tm_mon + 1,
+            t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, log_level_strings[log_level]);
     va_start(args, format);
     vfprintf(log_fp, format, args);
     va_end(args);

src/shared/multiprocessing.c

@@ -26,19 +26,49 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que
   context->manifest = NULL;
   context->progress_bytes = 0;
   context->sender_done = false;
-  if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success ||
-      cnd_init(&context->condition_not_full_scanner) != thrd_success ||
-      cnd_init(&context->condition_not_empty_scanner) != thrd_success ||
-      mtx_init(&context->mutex_loader, mtx_plain) != thrd_success ||
-      cnd_init(&context->condition_not_full_loader) != thrd_success ||
-      mtx_init(&context->mutex_progress, mtx_plain) != thrd_success ||
-      cnd_init(&context->condition_not_empty_loader) != thrd_success) {
+  context->cancelled = false;
+  int init = 0;
+  if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_full_scanner) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_empty_scanner) != thrd_success)
+    goto fail;
+  init++;
+  if (mtx_init(&context->mutex_loader, mtx_plain) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_full_loader) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_empty_loader) != thrd_success)
+    goto fail;
+  init++;
+  if (mtx_init(&context->mutex_progress, mtx_plain) != thrd_success)
+    goto fail;
+  // cppcheck-suppress unreadVariable
+  init++;
+  return context;
+
+fail:
   perror("Error initializing synchronization objects");
+  if (init >= 6)
+    cnd_destroy(&context->condition_not_empty_loader);
+  if (init >= 5)
+    cnd_destroy(&context->condition_not_full_loader);
+  if (init >= 4)
+    mtx_destroy(&context->mutex_loader);
+  if (init >= 3)
+    cnd_destroy(&context->condition_not_empty_scanner);
+  if (init >= 2)
+    cnd_destroy(&context->condition_not_full_scanner);
+  if (init >= 1)
+    mtx_destroy(&context->mutex_scanner);
   free(context);
   return NULL;
 }
-  return context;
-}
 
 void pipeline_context_sender_destroy(PipelineContextSender* context) {
   if (context->manifest) {
@@ -67,15 +97,31 @@ PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue*
   context->file_descriptor = file_descriptor;
   context->ssl = ssl;
   context->receiver_done = false;
-  if (mtx_init(&context->mutex, mtx_plain) != thrd_success ||
-      cnd_init(&context->condition_not_full) != thrd_success ||
-      cnd_init(&context->condition_not_empty) != thrd_success) {
+  context->cancelled = false;
+  int init = 0;
+  if (mtx_init(&context->mutex, mtx_plain) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_full) != thrd_success)
+    goto fail;
+  init++;
+  if (cnd_init(&context->condition_not_empty) != thrd_success)
+    goto fail;
+  // cppcheck-suppress unreadVariable
+  init++;
+  return context;
+
+fail:
   perror("Error initializing synchronization objects");
+  if (init >= 3)
+    cnd_destroy(&context->condition_not_empty);
+  if (init >= 2)
+    cnd_destroy(&context->condition_not_full);
+  if (init >= 1)
+    mtx_destroy(&context->mutex);
   free(context);
   return NULL;
 }
-  return context;
-}
 
 void pipeline_context_receiver_destroy(PipelineContextReceiver* context) {
   config_delete(context->config);
@@ -94,8 +140,13 @@ static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver*
   for (int i = 0; i < chunk->element_count; i++) {
     File* file = chunk->items[i];
     chunk->items[i] = NULL;
-    queue_enqueue_multithreaded(context->queue, file, &context->mutex,
-                                &context->condition_not_empty, &context->condition_not_full);
+    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;
@@ -129,8 +180,12 @@ int receive_thread(void* pipeline_context) {
       if (!skipped) {
         if (file == NULL)
           return thrd_error;
-        queue_enqueue_multithreaded(context->queue, file, &context->mutex,
-                                    &context->condition_not_empty, &context->condition_not_full);
+        if (!queue_enqueue_multithreaded_cancel(
+                context->queue, file, &context->mutex, &context->condition_not_empty,
+                &context->condition_not_full, &context->cancelled)) {
+          file_destroy(file);
+          return thrd_error;
+        }
       }
     } else if (status == STATUS_CHUNK) {
       if (!receive_chunk_enqueue(file_descriptor, context))
@@ -166,8 +221,12 @@ int receive_thread(void* pipeline_context) {
     } else {
       File* file = file_receive(config, file_descriptor);
       if (file) {
-        queue_enqueue_multithreaded(context->queue, file, &context->mutex,
-                                    &context->condition_not_empty, &context->condition_not_full);
+        if (!queue_enqueue_multithreaded_cancel(
+                context->queue, file, &context->mutex, &context->condition_not_empty,
+                &context->condition_not_full, &context->cancelled)) {
+          file_destroy(file);
+          return thrd_error;
+        }
       } else {
         log_message(LOG_LEVEL_ERROR, "Failed to receive file");
         return thrd_error;

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

@@ -6,10 +6,10 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <threads.h>
 #include <time.h>
 #include <unistd.h>
 
-#define MAX_DATA_SIZE (100ULL * 1024 * 1024)          /* 100 MB max per data message */
 #define RECEIVE_TIMEOUT_SEC 60                        /* 60 second per-message timeout */
 #define MAX_CONNECTION_MEMORY (1024ULL * 1024 * 1024) /* 1 GB total per connection */
 
@@ -20,6 +20,8 @@ static __thread SSL* io_ssl;
 static unsigned long long io_bwlimit = 0;
 static long long bw_tokens = 0;
 static struct timespec bw_last_refill = {0, 0};
+static mtx_t bw_mutex;
+static once_flag bw_mutex_once = ONCE_FLAG_INIT;
 
 static __thread unsigned long long total_allocated_bytes = 0;
 
@@ -28,15 +30,24 @@ void io_set_fds(int read_fd, int write_fd) {
   io_write_fd = write_fd;
 }
 
+static void bw_mutex_init(void) {
+  mtx_init(&bw_mutex, mtx_plain);
+}
+
 void io_set_bwlimit(unsigned long long bytes_per_sec) {
+  call_once(&bw_mutex_once, bw_mutex_init);
+  mtx_lock(&bw_mutex);
   io_bwlimit = bytes_per_sec;
   bw_tokens = (long long)io_bwlimit;
   clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
+  mtx_unlock(&bw_mutex);
 }
 
 static void bw_throttle(size_t bytes_written) {
   if (io_bwlimit == 0)
     return;
+  call_once(&bw_mutex_once, bw_mutex_init);
+  mtx_lock(&bw_mutex);
 
   struct timespec now;
   clock_gettime(CLOCK_MONOTONIC, &now);
@@ -61,6 +72,7 @@ static void bw_throttle(size_t bytes_written) {
     bw_tokens = 0;
     clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
   }
+  mtx_unlock(&bw_mutex);
 }
 
 void io_set_ssl(SSL* ssl) {
@@ -177,6 +189,8 @@ static const char* status_to_string(Status status) {
 }
 
 bool send_str(int file_descriptor, const char* data) {
+  if (data == NULL)
+    return false;
   size_t size = strlen(data);
   if (!send_n_data(file_descriptor, &size, sizeof(size_t)))
     return false;
@@ -221,9 +235,9 @@ Data* receive_data(int file_descriptor) {
   unsigned long long size = 0;
   if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long)))
     return NULL;
-  if (size > MAX_DATA_SIZE) {
+  if (size > MAX_DATA_PAYLOAD_SIZE) {
     log_message(LOG_LEVEL_ERROR, "Data size %llu exceeds maximum %llu", size,
-                (unsigned long long)MAX_DATA_SIZE);
+                (unsigned long long)MAX_DATA_PAYLOAD_SIZE);
     return NULL;
   }
   if (total_allocated_bytes + size > MAX_CONNECTION_MEMORY) {

src/shared/queue.c

@@ -103,6 +103,22 @@ bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t*
   return ok;
 }
 
+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) {
+  mtx_lock(mutex);
+  while (queue_is_full(queue) && (cancelled == NULL || !*cancelled))
+    cnd_wait(condition_not_full, mutex);
+  if (cancelled != NULL && *cancelled) {
+    mtx_unlock(mutex);
+    return false;
+  }
+  bool ok = queue_enqueue(queue, item);
+  cnd_signal(condition_not_empty);
+  mtx_unlock(mutex);
+  return ok;
+}
+
 void* queue_dequeue(Queue* queue) {
   if (queue == NULL || queue_is_empty(queue)) {
     perror("ERROR: Could not dequeue from null or empty queue.");

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);

src/shared/utils.c

@@ -10,11 +10,13 @@
 #include <unistd.h>
 
 bool mkdir_r(const char* path) {
-  char* path_duplicate = malloc(strlen(path) + 1);
+  size_t path_len = strlen(path);
+  char* path_duplicate = malloc(path_len + 1);
   if (!path_duplicate)
     return false;
-  memcpy(path_duplicate, path, strlen(path) + 1);
-  char* path_current = (char*)malloc((strlen(path) + 2) * sizeof(char));
+  memcpy(path_duplicate, path, path_len + 1);
+  size_t capacity = path_len + 2;
+  char* path_current = (char*)malloc(capacity * sizeof(char));
   if (!path_current) {
     free(path_duplicate);
     return false;
@@ -33,6 +35,10 @@ bool mkdir_r(const char* path) {
   bool ok = true;
   while (part != NULL) {
     size_t part_len = strlen(part);
+    if ((size_t)(path_current_position - path_current) + part_len + 2 > capacity) {
+      ok = false;
+      break;
+    }
     memcpy(path_current_position, part, part_len);
     path_current_position += part_len;
     path_current_position[0] = '/';
@@ -64,10 +70,18 @@ char* str_dup(const char* string) {
   return new_string;
 }
 
+/* Match a glob pattern against a string. Supported wildcards:
+ *   ?      matches any single character except '/'.
+ *   *      matches any sequence of characters within one path component (no '/').
+ *   **     matches any sequence of characters, including '/' (cross-directory).
+ *   slash-star-star-slash is treated as a cross-directory wildcard when it appears between
+ * literals.
+ */
 bool glob_match(const char* pattern, const char* str) {
   while (*pattern) {
     if (*pattern == '*') {
       if (*(pattern + 1) == '*') {
+        /* globstar: match across directories */
         pattern += 2;
         if (*pattern == '\0')
           return true;
@@ -80,6 +94,7 @@ bool glob_match(const char* pattern, const char* str) {
         }
         return glob_match(pattern, str);
       }
+      /* single *: match within one path component */
       pattern++;
       while (*str && *str != '/') {
         if (glob_match(pattern, str))
@@ -94,6 +109,7 @@ bool glob_match(const char* pattern, const char* str) {
       str++;
     } else {
       if (*pattern != *str) {
+        /* allow literal / ** / rest to match any number of directories */
         if (*pattern == '/' && *(pattern + 1) == '*' && *(pattern + 2) == '*') {
           const char* rest = pattern + 3;
           if (*rest == '/')

tests/test_client_cli.c

@@ -6,6 +6,9 @@
 #include <string.h>
 #include <unistd.h>
 
+/* Declaration of parse_args from client_cli.c */
+int parse_args(Config* config, int argc, char* argv[], int* positional_args, int* positional_count);
+
 /* Test main() with --help flag (early return path, no server connection needed) */
 static void test_cli_help() {
   /* We can't easily call main() because it calls send_files which needs a server.
@@ -80,10 +83,160 @@ static void test_cli_exclude_patterns() {
   config_delete(cfg);
 }
 
+/* Test parse_args with --help returns 1 (clean exit) */
+static void test_parse_args_help() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--help"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 2, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args with -V/--version returns 1 */
+static void test_parse_args_version() {
+  Config* cfg = config_create();
+  char* argv_short[] = {"fastsync", "-V"};
+  char* argv_long[] = {"fastsync", "--version"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 2, argv_short, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 1);
+
+  ret = parse_args(cfg, 2, argv_long, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args with valid port */
+static void test_parse_args_valid_port() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "-p", "2222", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 0);
+  EXPECT_EQ_INT(cfg->ssh_port, 2222);
+  EXPECT_EQ_INT(positional_count, 2);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args rejects port > 65535 */
+static void test_parse_args_invalid_port() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "-p", "99999", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, -1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args rejects non-numeric port */
+static void test_parse_args_non_numeric_port() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "-p", "abc", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, -1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args rejects server port > 65535 */
+static void test_parse_args_invalid_server_port() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--server-port", "70000", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, -1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args rejects invalid compression level */
+static void test_parse_args_invalid_compression_level() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "-c", "25", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, -1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args accepts valid compression level */
+static void test_parse_args_valid_compression_level() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "-c", "10", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 5, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 0);
+  EXPECT_EQ_INT(cfg->compression_level, 10);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args unknown option returns error */
+static void test_parse_args_unknown_option() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--nonexistent", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 4, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, -1);
+
+  config_delete(cfg);
+}
+
+/* Test parse_args with --archive flag */
+static void test_parse_args_archive() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--archive", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+
+  int ret = parse_args(cfg, 4, argv, positional_args, &positional_count);
+  EXPECT_EQ_INT(ret, 0);
+  EXPECT_TRUE(cfg->use_compression);
+  EXPECT_TRUE(cfg->use_multithreading);
+  EXPECT_TRUE(cfg->use_metadata);
+
+  config_delete(cfg);
+}
+
 void test_client_cli() {
   test_cli_help();
   test_cli_archive_flags();
   test_cli_dry_run();
   test_cli_delete_flag();
   test_cli_exclude_patterns();
+  test_parse_args_help();
+  test_parse_args_version();
+  test_parse_args_valid_port();
+  test_parse_args_invalid_port();
+  test_parse_args_non_numeric_port();
+  test_parse_args_invalid_server_port();
+  test_parse_args_invalid_compression_level();
+  test_parse_args_valid_compression_level();
+  test_parse_args_unknown_option();
+  test_parse_args_archive();
 }