feat(client): --port alias, --threads=N, graceful abort, keepalive

src/client/client_cli.c

@@ -12,6 +12,7 @@
 #include "identity.h"
 #include "log.h"
 #include "protocol.h"
+#include "scanner.h"
 #include "stop_condition.h"
 #include "transport_tcp.h"
 #include "transport_tls.h"
@@ -27,6 +28,26 @@
 #include <stdlib.h>
 #include <string.h>
 
+/* Async-signal-safe abort flag set by the SIGINT/SIGTERM handler.  Exposed via
+ * client_send.h so the send loops can poll it.  Defined here (not in
+ * client_send.c) so the unit-test binary, which compiles this file but not
+ * client_send.c, still links the symbol. */
+volatile sig_atomic_t client_abort_requested = 0;
+
+#ifndef FASTSYNC_TEST_BUILD
+/* Signal handler: perform NO work beyond storing the flag.  Logging, protocol
+ * I/O and the STATUS_ABORT frame are all done later on the normal send path,
+ * which is not async-signal-safe.  Only the production client installs it. */
+static void client_signal_handler(int signo) {
+  (void)signo;
+  client_abort_requested = 1;
+}
+#endif
+
+bool client_abort_pending(void) {
+  return client_abort_requested != 0;
+}
+
 #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,
@@ -141,6 +162,23 @@ static int set_checksum_seed(Config* config, const char* value) {
   return 0;
 }
 
+/* --threads=N: enable the -m pipeline and size its parallel scanner pool.
+ * Rejects a non-positive, non-numeric or oversized value up front. */
+static int set_scanner_threads_option(Config* config, const char* value) {
+  int threads;
+  if (!parse_positive_int(value, &threads)) {
+    log_message(LOG_LEVEL_ERROR, "--threads must be a positive integer");
+    return -1;
+  }
+  if (threads > MAX_SCANNER_THREADS) {
+    log_message(LOG_LEVEL_ERROR, "--threads must be between 1 and %d", MAX_SCANNER_THREADS);
+    return -1;
+  }
+  config->use_multithreading = true;
+  config->scanner_threads = threads;
+  return 0;
+}
+
 static int set_compression_threads_option(int* dest, const char* value) {
   if (set_positive_int_option(dest, value, "--compress-threads") != 0)
     return -1;
@@ -1288,10 +1326,21 @@ static bool cli_handle_transfer_flags(CliParseCtx* ctx) {
     return true;
   }
   if (opt_is(arg, "-j", "--threads")) {
+    /* Bare -j/--threads: enable the pipeline with the scanner's built-in
+       worker default (scanner_threads stays 0). */
     config->use_multithreading = true;
     log_info_message(LOG_INFO_MISC, "Enabled Multithreading");
     return true;
   }
+  if (strncmp(arg, "--threads=", 10) == 0) {
+    if (set_scanner_threads_option(config, arg + 10) != 0) {
+      ctx->exit_code = -1;
+    } else {
+      log_info_message(LOG_INFO_MISC, "Enabled Multithreading with %d scanner threads",
+                       config->scanner_threads);
+    }
+    return true;
+  }
   if (opt_is(arg, "--chunk-serialization", NULL)) {
     config->use_chunk_serialization = true;
     log_info_message(LOG_INFO_MISC, "Enabled Chunk Serialization");
@@ -1300,29 +1349,48 @@ static bool cli_handle_transfer_flags(CliParseCtx* ctx) {
   return false;
 }
 
-/* Network/IO options: --server-port, --bwlimit, --chunk-size, --log-file and
- * --stderr.  Returns true when the argument was consumed. */
+/* Parse and validate a TCP server port (--server-port, or its rsync-friendly
+ * alias --port).  Returns 0 on success, -1 (with a message) on a malformed or
+ * out-of-range value. */
+static int set_server_port_option(Config* config, const char* value, const char* option_name) {
+  int port;
+  if (!parse_positive_int(value, &port)) {
+    char* escaped = output_escape(value, false);
+    log_message(LOG_LEVEL_ERROR, "invalid %s value: %s", option_name,
+                escaped ? escaped : "<allocation failed>");
+    free(escaped);
+    return -1;
+  }
+  if (port > 65535) {
+    log_message(LOG_LEVEL_ERROR, "server port must be 1-65535");
+    return -1;
+  }
+  config->server_port = port;
+  return 0;
+}
+
+/* Network/IO options: --server-port/--port, --bwlimit, --chunk-size, --log-file
+ * and --stderr.  Returns true when the argument was consumed. */
 static bool cli_handle_io_options(CliParseCtx* ctx) {
   Config* config = ctx->config;
   const char* arg = ctx->argv[ctx->i];
-  if (opt_is(arg, "--server-port", NULL)) {
+  if (opt_is(arg, "--server-port", "--port")) {
     if (ctx->i + 1 >= ctx->argc) {
       log_message(LOG_LEVEL_ERROR, "missing argument for %s", arg);
       ctx->exit_code = -1;
       return true;
     }
-    if (!parse_positive_int(ctx->argv[++ctx->i], &config->server_port)) {
-      char* escaped = output_escape(ctx->argv[ctx->i], false);
-      log_message(LOG_LEVEL_ERROR, "invalid --server-port value: %s",
-                  escaped ? escaped : "<allocation failed>");
-      free(escaped);
+    if (set_server_port_option(config, ctx->argv[++ctx->i], arg) != 0)
       ctx->exit_code = -1;
-      return true;
-    }
-    if (config->server_port > 65535) {
-      log_message(LOG_LEVEL_ERROR, "server port must be 1-65535");
+    return true;
+  }
+  /* rsync users commonly write --port=NNNN; --server-port=NNNN is accepted too
+   * so both spellings behave identically. */
+  if (strncmp(arg, "--server-port=", 14) == 0 || strncmp(arg, "--port=", 7) == 0) {
+    const char* option_name = strncmp(arg, "--server-port=", 14) == 0 ? "--server-port" : "--port";
+    const char* value = arg + (strncmp(arg, "--server-port=", 14) == 0 ? 14 : 7);
+    if (set_server_port_option(config, value, option_name) != 0)
       ctx->exit_code = -1;
-    }
     return true;
   }
   if (opt_is(arg, "--bwlimit", NULL)) {
@@ -1975,6 +2043,17 @@ int main(int argc, char* argv[]) {
      oversized delta).  Ignore SIGPIPE so that a broken TCP connection
      surfaces as a clean write error instead of killing the client. */
   signal(SIGPIPE, SIG_IGN);
+  /* Ctrl-C / SIGTERM: set the abort flag so the send loops can send
+   * STATUS_ABORT and let the receiver clean up, instead of dying abruptly.
+   * No SA_RESTART so an in-flight poll()/read() is interrupted (EINTR), which
+   * lets the keepalive/abort checks observe the flag promptly. */
+  struct sigaction abort_action;
+  memset(&abort_action, 0, sizeof(abort_action));
+  abort_action.sa_handler = client_signal_handler;
+  sigemptyset(&abort_action.sa_mask);
+  abort_action.sa_flags = 0;
+  sigaction(SIGINT, &abort_action, NULL);
+  sigaction(SIGTERM, &abort_action, NULL);
   const char* env_source = NULL;
   const char* env_dest = NULL;
   bool save_to_disk = false;

src/client/client_send.c

@@ -885,6 +885,10 @@ static int send_delete_manifest(int fd, ArrayList* manifest, ArrayList* protecte
    unlinks) before replying, so the wait uses a generous explicit deadline
    instead of the default 60 s receive window. */
 #define DELETE_ACK_TIMEOUT_SEC 3600
+/* While waiting for the (potentially slow) receiver-side deletion, send a
+ * STATUS_KEEPALIVE at most this often so the connection is demonstrably alive
+ * and neither side's per-message timeout trips. */
+#define DELETE_ACK_KEEPALIVE_SEC 10
 
 static bool send_delete_manifest_early(Client* client, ArrayList* manifest,
                                        ArrayList* protected_prefixes, ArrayList* missing_args) {
@@ -894,8 +898,21 @@ static bool send_delete_manifest_early(Client* client, ArrayList* manifest,
       0)
     return false;
   Status ack;
-  if (!receive_status_timed(client->file_descriptor, &ack, DELETE_ACK_TIMEOUT_SEC))
+  /* The wait is long (up to an hour) and runs inline on this thread: a helper
+   * thread would race the non-thread-safe protocol send path, so keepalives are
+   * emitted from this wait loop itself.  A Ctrl-C/SIGTERM abort flag also ends
+   * the wait; the caller then best-effort sends STATUS_ABORT. */
+  if (!receive_status_keepalive(client->file_descriptor, &ack, DELETE_ACK_TIMEOUT_SEC,
+                                DELETE_ACK_KEEPALIVE_SEC, client_abort_pending)) {
+    /* A Ctrl-C/SIGTERM abort ends the wait above; tell the receiver before the
+       caller tears the connection down (best-effort). */
+    if (client_abort_pending()) {
+      log_info_message(LOG_INFO_MISC,
+                       "Abort requested while awaiting delete ack; sending STATUS_ABORT");
+      send_status(client->file_descriptor, STATUS_ABORT);
+    }
     return false;
+  }
   if (ack != STATUS_OK) {
     log_message(LOG_LEVEL_ERROR, "Server failed to delete files before the transfer");
     return false;
@@ -1490,6 +1507,19 @@ static int send_chunks_multithreaded(void* pipeline_context) {
   }
 
   while (true) {
+    /* Graceful abort (Ctrl-C/SIGTERM): tell the receiver to clean up instead of
+       dying abruptly.  Best-effort: a failed send just means the peer is gone.
+       Only reached while the session is active (config_send already succeeded). */
+    if (client_abort_pending()) {
+      log_info_message(LOG_INFO_MISC,
+                       "Abort requested; sending STATUS_ABORT to server and disconnecting");
+      send_status(client->file_descriptor, STATUS_ABORT);
+      pipeline_cancel(context);
+      disconnect_transfer_client(client);
+      mark_sender_done(context);
+      protocol_session_unbind();
+      return thrd_error;
+    }
     /* Phase 6: stop-elegantly at the next chunk boundary once the --stop-after
        / --stop-at deadline has passed.  Everything already sent is finalized by
        the completion tail below; the run still returns success. */
@@ -1624,7 +1654,9 @@ static int scan_directory_multithreaded(void* pipeline_context) {
   PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
   protocol_session_bind(&context->allocation_session);
   PreparedScanner prepared;
-  if (!prepare_scanner(context->config, 4, &prepared)) {
+  /* -j/--threads=N sizes the parallel scanner's worker pool; 0 (bare -j) lets
+   * the scanner apply its built-in default. */
+  if (!prepare_scanner(context->config, context->config->scanner_threads, &prepared)) {
     pipeline_cancel(context);
     protocol_session_unbind();
     return thrd_error;
@@ -2036,6 +2068,15 @@ int send_files(Config* config) {
      only a prefix of the source. */
   bool scan_stopped_early = false;
   while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
+    /* Graceful abort (Ctrl-C/SIGTERM): notify the receiver and clean up.  The
+       session is active (config_send already succeeded); a send failure here is
+       fine because the client is exiting anyway. */
+    if (client_abort_pending()) {
+      log_info_message(LOG_INFO_MISC, "Abort requested; sending STATUS_ABORT to server");
+      chunk_destroy(current_chunk);
+      send_status(client->file_descriptor, STATUS_ABORT);
+      goto send_fail;
+    }
     /* Phase 6: stop-elegantly at the next chunk boundary once the deadline has
        passed.  The scanner may also have stopped early itself; either way the
        completion tail below keeps everything already sent. */
@@ -2099,6 +2140,13 @@ int send_files(Config* config) {
     goto send_fail;
   if (directory_scanner_had_io_error(scanner))
     had_scan_io = true;
+  /* An abort that arrived after the last chunk must still stop the completion
+     tail (manifest/finalize) rather than let it run to success. */
+  if (client_abort_pending()) {
+    log_info_message(LOG_INFO_MISC, "Abort requested; sending STATUS_ABORT to server");
+    send_status(client->file_descriptor, STATUS_ABORT);
+    goto send_fail;
+  }
   /* Phase 6: the scanner may have stopped early (returning NULL without a
      failure) as soon as the deadline passed, so reflect that here too.  A
      deadline that cut the scan short leaves an incomplete keep-set; transmitting
@@ -2276,7 +2324,7 @@ int send_files_multithreaded(Config** config_ptr) {
          fills the protected excluded prefixes. */
       PreparedScanner prepared;
       memset(&prepared, 0, sizeof(prepared));
-      bool prepared_ok = prepare_scanner(config, 4, &prepared);
+      bool prepared_ok = prepare_scanner(config, config->scanner_threads, &prepared);
       if (prepared_ok && context->excluded_paths)
         prepared.options.excluded_paths = context->excluded_paths;
       bool prebuilt = prepared_ok && scan_paths_only(config, &prepared.options, context->manifest,

src/client/client_send.h

@@ -4,6 +4,15 @@
 #include "chunk.h"
 #include "config.h"
 #include "transport_tcp.h"
+#include <signal.h>
+#include <stdbool.h>
+
+/* Set ONLY by the client's SIGINT/SIGTERM handler (async-signal-safe: the
+ * handler stores 1 and does nothing else).  The send loops poll it via
+ * client_abort_pending() and, when set, best-effort send STATUS_ABORT so the
+ * receiver can clean up before the client exits. */
+extern volatile sig_atomic_t client_abort_requested;
+bool client_abort_pending(void);
 
 /* Both sender entry points BORROW `config` for the duration of the call; they
  * never free it, and the caller retains ownership (freeing it with

src/client/scanner.h

@@ -14,6 +14,10 @@
 #include <sys/types.h>
 #include <threads.h>
 
+/* Upper bound on the configurable parallel scanner worker count (--threads=N):
+ * keeps one transfer from spawning an unbounded pool on a very large machine. */
+#define MAX_SCANNER_THREADS 256
+
 typedef struct {
   bool use_metadata;
   /* Phase 4 metadata capture: -U/--atimes and -N/--crtimes tell the scanner to

src/client/usage.c

@@ -2,6 +2,7 @@
 #include <stdio.h>
 #include <delta.h>
 #include <chunk.h>
+#include "scanner.h"
 
 void print_usage(void) {
   printf("Usage:\n");
@@ -144,7 +145,10 @@ void print_usage(void) {
   printf("                    Delta block size in bytes (default: %d)\n", DELTA_BLOCK_SIZE_DEFAULT);
   printf("  --delta-max <n>     Max file size for delta transfer (default: %llu)\n",
          DELTA_MAX_FILE_SIZE);
-  printf("  -j, --threads       Enable multithreading\n");
+  printf("  -j, --threads[=N]   Enable the multithreaded scanner/loader/sender\n");
+  printf("                      pipeline; N (1-%d) sets the parallel scanner worker\n",
+         MAX_SCANNER_THREADS);
+  printf("                      count (bare -j/--threads uses the default)\n");
   printf("  --chunk-serialization  Enable chunk serialization (long form only)\n");
   printf("  -s, --secluded-args    Protect-args compatibility option (no effect; remote\n");
   printf("                      SSH argv is already built injection-safe)\n");
@@ -203,6 +207,7 @@ void print_usage(void) {
   printf("  --save-to-disk      Write received files to disk\n");
   printf("  --server-host <ip>  Server IP address (default: 127.0.0.1)\n");
   printf("  --server-port <n>   Server port (default: 8080)\n");
+  printf("  --port <n>          Alias for --server-port\n");
   printf("  --password-file <f> Authenticate a host::module/path daemon destination.\n");
   printf("                      The file's first user:password line supplies the\n");
   printf("                      username and password (only a SHA-256 digest of the\n");