feat(protocol): client-message channel and rsync partial exit 23 (2.30.0)

src/client/client_cli.c

@@ -433,12 +433,10 @@ static int set_stderr_mode(const char* value) {
     log_set_stderr_mode(LOG_STDERR_ERRORS);
   else if (strcmp(value, "all") == 0 || strcmp(value, "a") == 0)
     log_set_stderr_mode(LOG_STDERR_ALL);
-  else if (strcmp(value, "client") == 0 || strcmp(value, "c") == 0) {
-    log_message(LOG_LEVEL_ERROR,
-                "--stderr=client is not supported: FastSync has no client message channel");
-    return -1;
-  } else {
-    log_message(LOG_LEVEL_ERROR, "--stderr must be errors or all");
+  else if (strcmp(value, "client") == 0 || strcmp(value, "c") == 0)
+    log_set_stderr_mode(LOG_STDERR_CLIENT);
+  else {
+    log_message(LOG_LEVEL_ERROR, "--stderr must be errors, all, or client");
     return -1;
   }
   return 0;
@@ -1353,10 +1351,9 @@ static bool cli_handle_pre_negation(CliParseCtx* ctx) {
     return true;
   }
   /* "--no-msgs2stderr" is the deprecated spelling of --stderr=client (rsync
-   * 3.4.1).  FastSync has no separate client message channel, so the closest
-   * supported mode is the errors-only default. */
+   * 3.4.1); the client-message channel now exists, so it maps to `client`. */
   if (strcmp(arg, "--no-msgs2stderr") == 0)
-    return set_stderr_mode("errors") == 0;
+    return set_stderr_mode("client") == 0;
   /* "--no-motd" is a real rsync option name (client-side daemon MOTD display
    * suppression), not a negation of a "--motd" flag, so it is handled before
    * the generic --no-* negation branch. */

src/client/client_report.c

@@ -12,6 +12,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <threads.h>
 #include <time.h>
 
 /* Surface a server rejection to the user.  When the last status exchange
@@ -1051,3 +1052,109 @@ const char* delete_display_path(const Config* config, const char* path) {
     return path;
   return utils_strip_transfer_root(path, config->send_directory);
 }
+
+/* ---- --stderr=client diagnostic channel (protocol 2.30.0) ----
+ *
+ * When the client's --stderr mode is `client`, log_message() hands each of the
+ * client's own diagnostics to the sink installed here instead of writing them
+ * locally.  The sink QUEUES the text (it may be called from scanner worker
+ * threads while the sender is streaming) and the sender thread -- the sole
+ * writer of the protocol stream -- drains the queue over the wire at frame
+ * boundaries via client_flush_client_messages().  A bounded queue caps the
+ * memory a chatty run can pin; overflow falls back to local output so a
+ * diagnostic is never silently dropped. */
+#define CLIENT_MSG_MAX_QUEUED 256
+#define CLIENT_MSG_MAX_BYTES (256 * 1024)
+
+static mtx_t client_msg_mutex;
+static once_flag client_msg_mutex_once = ONCE_FLAG_INIT;
+static ArrayList* client_msg_queue = NULL; /* owns char* */
+static size_t client_msg_bytes = 0;
+/* True only while a live transfer session exists: before the connection is up
+   (or after it drops) the sink declines so log_message falls back to local
+   output, matching rsync's documented fallback. */
+static bool client_msg_active = false;
+
+static void client_msg_mutex_init(void) {
+  mtx_init(&client_msg_mutex, mtx_plain);
+}
+
+static bool client_msg_enqueue(const char* message);
+
+/* Install the global log sink for the duration of one transfer.  Safe to call
+ * more than once; the queue is created lazily. */
+void client_messages_install(void) {
+  call_once(&client_msg_mutex_once, client_msg_mutex_init);
+  mtx_lock(&client_msg_mutex);
+  if (!client_msg_queue)
+    client_msg_queue = array_list_create(free);
+  mtx_unlock(&client_msg_mutex);
+  log_set_client_msg_sink(client_msg_enqueue);
+}
+
+void client_messages_activate(bool active) {
+  client_msg_active = active;
+}
+
+/* log_message sink: takes ownership (queues) the message when a session is
+ * live; returns false otherwise so the caller writes it locally. */
+static bool client_msg_enqueue(const char* message) {
+  if (!message || message[0] == '\0')
+    return client_msg_active;
+  if (!client_msg_active)
+    return false;
+  size_t len = strlen(message);
+  call_once(&client_msg_mutex_once, client_msg_mutex_init);
+  mtx_lock(&client_msg_mutex);
+  bool queued = false;
+  if (client_msg_queue && (size_t)client_msg_queue->size < CLIENT_MSG_MAX_QUEUED &&
+      client_msg_bytes + len <= CLIENT_MSG_MAX_BYTES) {
+    char* copy = str_dup(message);
+    if (copy) {
+      if (array_list_add(client_msg_queue, copy)) {
+        client_msg_bytes += len;
+        queued = true;
+      } else {
+        free(copy);
+      }
+    }
+  }
+  mtx_unlock(&client_msg_mutex);
+  return queued;
+}
+
+/* Drain the queued diagnostics as STATUS_CLIENT_MSG frames on the sender
+ * thread.  Swaps the queue out under the mutex so a concurrent worker logging
+ * never blocks on the wire.  Must be called at a protocol frame boundary. */
+void client_flush_client_messages(int fd) {
+  if (fd < 0)
+    return;
+  call_once(&client_msg_mutex_once, client_msg_mutex_init);
+  mtx_lock(&client_msg_mutex);
+  ArrayList* pending = client_msg_queue;
+  client_msg_queue = array_list_create(free);
+  client_msg_bytes = 0;
+  mtx_unlock(&client_msg_mutex);
+  if (!pending)
+    return;
+  for (int i = 0; i < pending->size; i++) {
+    const char* message = pending->items[i];
+    if (message && message[0] != '\0' && !send_client_message(fd, message))
+      break; /* peer is gone; the rest would fail too */
+  }
+  array_list_delete(pending);
+}
+
+/* Tear down the sink after a transfer and free anything still queued. */
+void client_messages_end(void) {
+  log_set_client_msg_sink(NULL);
+  client_msg_active = false;
+  call_once(&client_msg_mutex_once, client_msg_mutex_init);
+  mtx_lock(&client_msg_mutex);
+  ArrayList* pending = client_msg_queue;
+  client_msg_queue = NULL;
+  client_msg_bytes = 0;
+  mtx_unlock(&client_msg_mutex);
+  if (pending)
+    array_list_delete(pending);
+}

src/client/client_send.c

@@ -127,6 +127,11 @@ Client* connect_transfer_client(const Config* config) {
 void disconnect_transfer_client(Client* client) {
   if (!client)
     return;
+  /* --stderr=client: push any diagnostics logged during the transfer to the
+     peer before the socket closes; once deactivated, later messages fall back
+     to local output instead of being lost. */
+  client_flush_client_messages(client->file_descriptor);
+  client_messages_activate(false);
   client_disconnect(client);
   client_delete(client);
 }
@@ -240,11 +245,16 @@ static void mark_sender_done(PipelineContextSender* context) {
    When --remove-source-files is active the receiver acknowledges each data
    file it processed, in send order: STATUS_NEXT means the file was written,
    STATUS_OK means the file was skipped/unchanged.  Skipped sources are marked
-   so the later removal pass keeps them. */
+   so the later removal pass keeps them.  `partial_out` is set when the receiver
+   reported STATUS_PARTIAL (a per-entry receiver failure): the transfer is
+   otherwise complete, so successfully stored sources are still removed and the
+   caller exits 23 (rsync's partial transfer) instead of a fatal non-zero. */
 static bool finalize_transfer(Client* client, const Config* config, ArrayList* remove_sources,
-                              bool* delete_limit_out, ReceiverStats* stats_out) {
+                              bool* delete_limit_out, bool* partial_out, ReceiverStats* stats_out) {
   if (delete_limit_out)
     *delete_limit_out = false;
+  if (partial_out)
+    *partial_out = false;
   if (!send_status(client->file_descriptor, STATUS_FINISHED))
     return false;
   /* The receiver emits its optional wire-stats frame (protocol 2.25.0) FIRST,
@@ -298,6 +308,16 @@ static bool finalize_transfer(Client* client, const Config* config, ArrayList* r
       *delete_limit_out = true;
     return true;
   }
+  /* A per-entry receiver failure the receiver chose to continue past is a
+     rsync PARTIAL transfer: everything else succeeded and the stored sources
+     may be removed, but the client must exit 23. */
+  if (status == STATUS_PARTIAL) {
+    log_message(LOG_LEVEL_WARNING,
+                "some files could not be transferred (see the server log for details)");
+    if (partial_out)
+      *partial_out = true;
+    return true;
+  }
   if (status != STATUS_OK) {
     log_server_rejection("Receiver reported transfer failure");
     return false;
@@ -815,6 +835,9 @@ static bool source_is_regular_file(const File* file) {
 
 static int send_chunk_with_removal(Client* client, Chunk* chunk, Config* config,
                                    ArrayList* remove_sources, TransferStats* stats) {
+  /* --stderr=client: this is a frame boundary, so forward any diagnostics the
+     scanner/log emitted since the previous chunk before the next frame. */
+  client_flush_client_messages(client->file_descriptor);
   if (config->use_chunk_serialization) {
     if (remove_sources) {
       for (int i = 0; i < chunk->element_count; i++) {
@@ -954,6 +977,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
   protocol_session_set_io_timeout(&session, context->config->timeout);
   protocol_session_set_ssl(&session, (SSL*)client->ssl);
   protocol_session_bind(&session);
+  client_messages_activate(true);
   if (!config_send(client->file_descriptor, context->config)) {
     pipeline_cancel(context);
     disconnect_transfer_client(client);
@@ -1117,11 +1141,14 @@ static int send_chunks_multithreaded(void* pipeline_context) {
       !send_dir_times(client, context->config, context->dir_entries))
     goto send_fail;
   bool delete_limit = false;
+  bool partial = false;
   ReceiverStats recv_stats;
   memset(&recv_stats, 0, sizeof(recv_stats));
+  client_flush_client_messages(client->file_descriptor);
   bool ok = finalize_transfer(client, context->config, context->remove_source_files, &delete_limit,
-                              &recv_stats);
+                              &partial, &recv_stats);
   context->delete_limit = delete_limit;
+  context->partial = partial;
   if (!ok && context->config->use_delete)
     log_message(LOG_LEVEL_ERROR,
                 "server reported a deletion failure (--delete); see the server log for the reason");
@@ -1823,9 +1850,12 @@ static int send_files_finalize(const Config* config, SendFilesState* state) {
   if (!send_dir_times(client, config, state->dir_entries))
     return 1;
   bool delete_limit = false;
+  bool partial = false;
   ReceiverStats recv_stats;
   memset(&recv_stats, 0, sizeof(recv_stats));
-  bool ok = finalize_transfer(client, config, state->remove_sources, &delete_limit, &recv_stats);
+  client_flush_client_messages(client->file_descriptor);
+  bool ok = finalize_transfer(client, config, state->remove_sources, &delete_limit, &partial,
+                              &recv_stats);
   if (!ok && config->use_delete)
     log_message(LOG_LEVEL_ERROR,
                 "server reported a deletion failure (--delete); see the server log for the reason");
@@ -1843,12 +1873,12 @@ static int send_files_finalize(const Config* config, SendFilesState* state) {
                    (double)state->transfer_stats.transferred_file_size / (double)BYTES_PER_MIB);
   /* A skipped source entry (--ignore-errors past an unreadable directory, or a
      dereferenced symlink with no referent) makes rsync report a partial
-     transfer (exit 23) even though the rest of the run succeeded.  A
-     --max-delete-capped commit is a successful transfer that rsync reports
+     transfer (exit 23), as does a receiver per-entry failure (STATUS_PARTIAL).
+     A --max-delete-capped commit is a successful transfer that rsync reports
      with exit code 25. */
   if (!ok)
     return 1;
-  if (state->had_scan_io)
+  if (state->had_scan_io || partial)
     return 23;
   return delete_limit ? 25 : 0;
 }
@@ -1885,7 +1915,18 @@ static void send_files_cleanup(SendFilesState* state) {
   client_set_abort_armed(false);
 }
 
+static int send_files_impl(Config* config);
+
 int send_files(Config* config) {
+  /* Install the --stderr=client sink for the whole run (it only queues while a
+     session is live) and release its queue on every return path. */
+  client_messages_install();
+  int rc = send_files_impl(config);
+  client_messages_end();
+  return rc;
+}
+
+static int send_files_impl(Config* config) {
   if (config->list_only)
     return send_list_only(config);
   if (config->dry_run)
@@ -1931,6 +1972,7 @@ int send_files(Config* config) {
   protocol_session_set_io_timeout(&session, config->timeout);
   protocol_session_set_ssl(&session, (SSL*)client->ssl);
   protocol_session_bind(&session);
+  client_messages_activate(true);
 
   int ret = 1;
   if (!send_files_prepare(config, &state))
@@ -1946,7 +1988,16 @@ send_fail:
   return ret;
 }
 
+static int send_files_multithreaded_impl(Config* config);
+
 int send_files_multithreaded(Config* config) {
+  client_messages_install();
+  int rc = send_files_multithreaded_impl(config);
+  client_messages_end();
+  return rc;
+}
+
+static int send_files_multithreaded_impl(Config* config) {
   if (!config)
     return 1;
   if (config->list_only)
@@ -2202,16 +2253,18 @@ int send_files_multithreaded(Config* config) {
   mtx_unlock(&context->mutex_scanner);
   bool sender_ok = sender_result == thrd_success;
   bool delete_limit = context->delete_limit;
+  bool partial = context->partial;
   /* A skipped source entry (--ignore-errors past an unreadable directory, or a
      dereferenced symlink with no referent) makes rsync report a partial
-     transfer (exit 23).  A --max-delete-capped commit is a successful transfer
-     that rsync reports with exit code 25. */
+     transfer (exit 23), as does a receiver per-entry failure (STATUS_PARTIAL).
+     A --max-delete-capped commit is a successful transfer that rsync reports
+     with exit code 25. */
   pipeline_context_sender_destroy(context);
   client_progress_cleanup();
   client_set_abort_armed(false);
   if (!sender_ok)
     return 1;
-  if (scan_io)
+  if (scan_io || partial)
     return 23;
   return delete_limit ? 25 : 0;
 }

src/client/client_send_internal.h

@@ -72,6 +72,13 @@ void client_progress_uptodate(const Config* config, const File* file);
 void client_progress_prepare(const Config* config, const ArrayList* plan_dirs,
                              unsigned long long plan_non_dir_count);
 bool receive_stats_record(int fd, ReceiverStats* stats, ArrayList* would_delete);
+/* --stderr=client diagnostic channel (client_report.c): install the queueing
+ * log sink for a transfer, mark the session live, flush queued diagnostics over
+ * the wire at a frame boundary, and tear the sink down. */
+void client_messages_install(void);
+void client_messages_activate(bool active);
+void client_flush_client_messages(int fd);
+void client_messages_end(void);
 
 /* client_send.c */
 void receive_daemon_motd(Client* client, const Config* config);

src/client/usage.c

@@ -297,11 +297,13 @@ void print_usage(void) {
   printf("  --max-depth <n>     Maximum directory depth (0=unlimited)\n");
   printf("  -x, --one-file-system  Do not cross filesystem boundaries\n");
   printf("  --log-file <path>, --log-file=<path>  Write log messages to file\n");
-  printf("  --stderr=MODE       Route logging to stderr: errors or all\n");
+  printf("  --stderr=MODE       Route logging: errors (default), all, or client\n");
+  printf("                      (forward the client's diagnostics to the server's\n");
+  printf("                      stderr)\n");
   printf("  --msgs2stderr       Route all messages to stderr (deprecated spelling of\n");
   printf("                      --stderr=all)\n");
-  printf("  --no-msgs2stderr    Select errors-only stderr (deprecated spelling; the\n");
-  printf("                      default)\n");
+  printf("  --no-msgs2stderr    Forward the client's diagnostics to the server\n");
+  printf("                      (deprecated spelling of --stderr=client)\n");
   printf("  --partial           Keep partial files on interrupted transfer\n");
   printf("  --partial-dir <dir> Directory for partial files (implies --partial)\n");
   printf("  -T, --temp-dir <dir>  Scratch dir for temp files before atomic install.\n");