From e3d643a0aa57c458bff63da0487dc1f4cd7a0edb Mon Sep 17 00:00:00 2001
From: seks99x <omarelsayed161@gmail.com>
Date: Fri, 6 Nov 2026 16:20:17 +0300
Subject: [PATCH 1/4] Add --stream-source support for single-pass FIFO/pipe
 delta transfers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Enable rsync to read directly from a local named pipe (FIFO) or unseekable
stream and perform full rolling-checksum delta transfers against an existing
destination basis file without staging to temporary storage or changing the
wire protocol.

Key changes:
- Add --stream-source, --stream-max-size=SIZE, --stream-timeout=SECONDS,
  and --stream-pipe-sz=SIZE CLI options to options.c, usage.c, and rsync.1.md.
- Disguise the local FIFO source as S_IFREG with st_size = stream_max_size
  in make_file() (flist.c) so the receiver/generator negotiates normal
  basis checksums (s->sums) without protocol modifications.
- Implement sequential stream windowing in map_file() and map_ptr()
  (fileio.c): disallow backward seeks (read_start != map->p_fd_offset),
  multiplex reads with select() to enforce stream_timeout, clamp
  map->file_size dynamically on early EOF (n == 0), and optionally resize
  the kernel pipe ring buffer via F_SETPIPE_SZ on Linux.
- Update match_sums() and hash_search() (match.c) to pre-read stream
  data when s->count == 0 (missing/empty basis file) and safely clamp the
  sliding window and tail flush upon early EOF.
- Verify true EOF in send_files() (sender.c) via a non-blocking read probe
  when a stream reaches stream_max_size, aborting with IOERR_GENERAL on
  overflow, and reconcile F_LENGTH(file) and stats.total_size to the actual
  bytes read before progress and transfer logging.
- Ensure receive_data() (receiver.c) calls do_ftruncate(fd, offset) in
  --inplace mode when an early-EOF stream ends shorter than the basis file.
- Block incompatible options (-r, -a, -d, -c, --append, --files-from,
  --remove-source-files, batch mode) and file-size-dependent options
  (--size-only, --max-size, --min-size, --fuzzy, --preallocate),
  while enforcing ignore_times = 1 and clearing update_only, preserve_links,
  preserve_specials, and preserve_devices.
- Add Python test suites (stream-source_test.py and stream-early-eof_test.py)
  covering delta matching, --inplace truncation, early EOF, exact ceiling,
  ceiling overflow, idle timeout, and option validation.
---
 fileio.c                           |  74 ++++++-
 flist.c                            |   6 +
 main.c                             |   6 +
 match.c                            |  69 ++++++-
 options.c                          |  83 ++++++++
 rsync.1.md                         | 103 ++++++++++
 rsync.h                            |   5 +
 sender.c                           |  41 ++++
 testsuite/stream-early-eof_test.py | 295 ++++++++++++++++++++++++++++
 testsuite/stream-source_test.py    | 302 +++++++++++++++++++++++++++++
 usage.c                            |   5 +
 11 files changed, 974 insertions(+), 15 deletions(-)
 create mode 100644 testsuite/stream-early-eof_test.py
 create mode 100644 testsuite/stream-source_test.py

diff --git a/fileio.c b/fileio.c
index b8db29b85..2ebb2a620 100644
--- a/fileio.c
+++ b/fileio.c
@@ -35,6 +35,8 @@
 
 extern int sparse_files;
 extern struct stats stats;
+extern int stream_timeout;
+extern int stream_pipe_sz;
 
 OFF_T preallocated_len = 0;
 
@@ -330,6 +332,20 @@ struct map_struct *map_file(int fd, OFF_T len, int32 read_size, int32 blk_size)
 	map->file_size = len;
 	map->def_window_size = ALIGNED_LENGTH(read_size);
 
+	if (is_stream()) {
+#ifdef F_SETPIPE_SZ
+		if (fcntl(map->fd, F_SETPIPE_SZ, stream_pipe_sz) < 0) {
+			if (stream_pipe_sz != DEFAULT_STREAM_PIPE_SZ || errno != EPERM)
+				rsyserr(FWARNING, errno, "Warning: could not resize pipe buffer to %d bytes, using OS default",
+						stream_pipe_sz);
+		} else if (INFO_GTE(MISC, 1)) {
+			rprintf(FINFO, "Stream pipe buffer resized to %d bytes\n", stream_pipe_sz);
+		}
+#else
+		rprintf(FWARNING, "Warning: F_SETPIPE_SZ is not supported on this platform, using OS default pipe buffer size\n");
+#endif
+	}
+
 	return map;
 }
 
@@ -386,21 +402,63 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 		exit_cleanup(RERR_FILEIO);
 	}
 
-	if (map->p_fd_offset != read_start) {
-		OFF_T ret = do_lseek(map->fd, read_start, SEEK_SET);
-		if (ret != read_start) {
-			rsyserr(FERROR, errno, "lseek returned %s, not %s",
-				big_num(ret), big_num(read_start));
+    if (map->p_fd_offset != read_start) {
+        if (is_stream()) {
+			rprintf(FERROR, "stream-source illegal seek from %s to %s\n",
+				big_num(map->p_fd_offset), big_num(read_start));
 			exit_cleanup(RERR_FILEIO);
-		}
-		map->p_fd_offset = read_start;
-	}
+		} else {
+            OFF_T ret = do_lseek(map->fd, read_start, SEEK_SET);
+            if (ret != read_start) {
+                rsyserr(FERROR, errno, "lseek returned %s, not %s",
+                    big_num(ret), big_num(read_start));
+                exit_cleanup(RERR_FILEIO);
+            }
+            map->p_fd_offset = read_start;
+        }
+    }
 	map->p_offset = window_start;
 	map->p_len = window_size;
 
 	while (read_size > 0) {
+		if (is_stream() && stream_timeout > 0) {
+			fd_set rfds;
+			struct timeval tv;
+
+			FD_ZERO(&rfds);
+			FD_SET(map->fd, &rfds);
+			tv.tv_sec = stream_timeout;
+			tv.tv_usec = 0;
+
+			int res = select(map->fd + 1, &rfds, NULL, NULL, &tv);
+			if (res == 0) {
+				rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
+					stream_timeout);
+				exit_cleanup(RERR_TIMEOUT);
+			} else if (res < 0) {
+				if (errno == EINTR)
+					continue;
+				rsyserr(FERROR, errno, "select failed on stream");
+				exit_cleanup(RERR_FILEIO);
+			}
+		}
 		int32 nread = read(map->fd, map->p + read_offset, read_size);
 		if (nread <= 0) {
+			/* Retry if a blocking pipe read was interrupted by a signal. */
+			if (nread < 0 && errno == EINTR)
+				continue;
+			/* On early EOF from a FIFO/pipe, shrink file_size and window
+			 * length to the actual bytes read so far and zero the tail. */
+			if (is_stream() && nread == 0) {
+				if (INFO_GTE(MISC, 1)) {
+					rprintf(FINFO, "Stream reached early EOF at %s bytes (ceiling was %s)\n",
+						big_num(map->p_fd_offset), big_num(map->file_size));
+				}
+				map->file_size = map->p_fd_offset;
+				map->p_len = read_offset;
+				memset(map->p + read_offset, 0, read_size);
+				break;
+			}
 			if (!map->status)
 				map->status = nread ? errno : ENODATA;
 			/* The best we can do is zero the buffer -- the file
diff --git a/flist.c b/flist.c
index 282fe81e9..9c6407ef2 100644
--- a/flist.c
+++ b/flist.c
@@ -86,6 +86,7 @@ extern uid_t our_uid;
 extern struct stats stats;
 extern char *filesfrom_host;
 extern char *usermap, *groupmap;
+extern OFF_T stream_max_size;
 
 extern struct name_num_item *file_sum_nni;
 
@@ -1704,6 +1705,11 @@ struct file_struct *make_file(const char *fname, struct file_list *flist,
 			full_fname(thisname));
 		return NULL;
 	}
+	if (is_stream() && S_ISFIFO(st.st_mode)) {
+		/* Disguise the FIFO pipe as a regular file globally */
+		st.st_mode = (st.st_mode & ~S_IFMT) | S_IFREG;
+		st.st_size = stream_max_size;
+	}
 
 	if (filter_level == NO_FILTERS)
 		goto skip_filters;
diff --git a/main.c b/main.c
index 63c35a15c..33f3b5e8d 100644
--- a/main.c
+++ b/main.c
@@ -95,6 +95,7 @@ extern int default_af_hint;
 extern int stdout_format_has_i;
 extern int trust_sender_filter;
 extern int trust_sender_args;
+extern int stream_source_mode;
 extern struct stats stats;
 extern char *stdout_format;
 extern char *logfile_format;
@@ -1481,6 +1482,11 @@ static int start_client(int argc, char *argv[])
 		if (path) { /* source is remote */
 			char *dummy_host;
 			int dummy_port = 0;
+			if (stream_source_mode) {
+				rprintf(FERROR,
+					"--stream-source cannot be used with a remote source\n");
+				exit_cleanup(RERR_SYNTAX);
+			}
 			*argv = path;
 			remote_argv = argv;
 			remote_argc = argc;
diff --git a/match.c b/match.c
index 6f17b354e..b5a2b1bb0 100644
--- a/match.c
+++ b/match.c
@@ -75,6 +75,13 @@ static int32 *hash_table;
 
 #define BIG_SUM2HASH(sum) ((sum)%tablesize)
 
+/* Return true if a stream hit EOF before reaching the expected length 'len'
+ * (map_ptr() shrinks buf->file_size to the actual bytes read on pipe EOF). */
+static int stream_early_EOF(struct map_struct *buf, OFF_T len)
+{
+	return is_stream() && buf->file_size < len;
+}
+
 static void build_hash_table(struct sum_struct *s)
 {
 	static uint32 alloc_size;
@@ -187,6 +194,20 @@ static void hash_search(int f,struct sum_struct *s,
 	k = (int32)MIN(len, (OFF_T)s->blength);
 
 	map = (schar *)map_ptr(buf, 0, k);
+	if (stream_early_EOF(buf, len)) {
+		len = buf->file_size;
+		k = (int32)MIN(len, (OFF_T)s->blength);
+		if (DEBUG_GTE(DELTASUM, 2)) {
+			rprintf(FINFO, "stream early EOF at offset=0 len=%s k=%ld\n",
+				big_num(len), (long)k);
+		}
+	}
+
+	offset = aligned_offset = aligned_i = 0;
+
+	end = len + 1 - s->sums[s->count-1].len;
+	if (offset >= end)
+		goto finish_search;
 
 	sum = get_checksum1((char *)map, k);
 	s1 = sum & 0xFFFF;
@@ -194,10 +215,6 @@ static void hash_search(int f,struct sum_struct *s,
 	if (DEBUG_GTE(DELTASUM, 3))
 		rprintf(FINFO, "sum=%.8x k=%ld\n", sum, (long)k);
 
-	offset = aligned_offset = aligned_i = 0;
-
-	end = len + 1 - s->sums[s->count-1].len;
-
 	if (DEBUG_GTE(DELTASUM, 3)) {
 		rprintf(FINFO, "hash search s->blength=%ld len=%s count=%s\n",
 			(long)s->blength, big_num(len), big_num(s->count));
@@ -317,6 +334,8 @@ static void hash_search(int f,struct sum_struct *s,
 							backup = 0;
 						map = (schar *)map_ptr(buf, aligned_offset - backup, l + backup)
 						    + backup;
+						if (stream_early_EOF(buf, aligned_offset + l))
+							goto check_want_i;
 						sum = get_checksum1((char *)map, l);
 						if (sum != s->sums[i].sum1)
 							goto check_want_i;
@@ -352,6 +371,8 @@ static void hash_search(int f,struct sum_struct *s,
 			offset += s->sums[i].len - 1;
 			k = (int32)MIN((OFF_T)s->blength, len-offset);
 			map = (schar *)map_ptr(buf, offset, k);
+			if (stream_early_EOF(buf, len))
+				k = (int32)MIN((OFF_T)s->blength, buf->file_size-offset);
 			sum = get_checksum1((char *)map, k);
 			s1 = sum & 0xFFFF;
 			s2 = sum >> 16;
@@ -368,8 +389,16 @@ static void hash_search(int f,struct sum_struct *s,
 				OFF_T skip = MIN((OFF_T)(k - 1), end - 1 - offset);
 				if (skip > 0) {
 					offset += skip;
+					/* Anchor the map_ptr window at last_match (offset - backup) so
+					 * un-flushed literal bytes are not evicted from the sliding window,
+					 * which would force an illegal backward lseek() on a stream. */
+					backup = (int32)(offset - last_match);
+					if (backup < 0)
+						backup = 0;
 					k = (int32)MIN((OFF_T)s->blength, len-offset);
-					map = (schar *)map_ptr(buf, offset, k);
+					map = (schar *)map_ptr(buf, offset - backup, k + backup) + backup;
+					if (stream_early_EOF(buf, len))
+						k = (int32)MIN((OFF_T)s->blength, buf->file_size-offset);
 					sum = get_checksum1((char *)map, k);
 					s1 = sum & 0xFFFF;
 					s2 = sum >> 16;
@@ -389,6 +418,16 @@ static void hash_search(int f,struct sum_struct *s,
 		/* Trim off the first byte from the checksum */
 		more = offset + k < len;
 		map = (schar *)map_ptr(buf, offset - backup, k + more + backup) + backup;
+		if (stream_early_EOF(buf, len)) {
+			len = buf->file_size;
+			end = len + 1 - s->sums[s->count-1].len;
+			k = (int32)MIN((OFF_T)s->blength, len - offset);
+			more = offset + k < len;
+			if (DEBUG_GTE(DELTASUM, 2)) {
+				rprintf(FINFO, "stream early EOF at offset=%s len=%s end=%s k=%ld\n",
+					big_num(offset), big_num(len), big_num(end), (long)k);
+			}
+		}
 		s1 -= map[0] + CHAR_OFFSET;
 		s2 -= k * (map[0]+CHAR_OFFSET);
 
@@ -409,8 +448,13 @@ static void hash_search(int f,struct sum_struct *s,
 			matched(f, s, buf, offset - s->blength, -2);
 	} while (++offset < end);
 
+  finish_search:
 	matched(f, s, buf, len, -1);
-	map_ptr(buf, len-1, 1);
+	/* Ensure the map window has read through the final byte of the file.
+	 * Guard with len > 0 because an empty stream (0-byte early EOF) shrinks
+	 * len to 0, where len - 1 would pass a negative offset (-1) to map_ptr(). */
+	if (len > 0)
+		map_ptr(buf, len-1, 1);
 }
 
 
@@ -480,8 +524,19 @@ void match_sums(int f, struct sum_struct *s, struct map_struct *buf, OFF_T len)
 	} else {
 		OFF_T j;
 		/* by doing this in pieces we avoid too many seeks */
-		for (j = last_match + CHUNK_SIZE; j < len; j += CHUNK_SIZE)
+		for (j = last_match + CHUNK_SIZE; ; j += CHUNK_SIZE) {
+			/* Pre-read the next chunk on a stream before calling matched() so
+			 * that if the pipe hits early EOF mid-chunk, buf->file_size and len
+			 * shrink before matched() tries to flush up to 'j'. */
+			if (is_stream()) {
+				map_ptr(buf, last_match, (int32)MIN((OFF_T)CHUNK_SIZE, len - last_match));
+				if (buf->file_size < len)
+					len = buf->file_size;
+			}
+			if (j >= len)
+				break;
 			matched(f, s, buf, j, -2);
+		}
 		matched(f, s, buf, len, -1);
 	}
 
diff --git a/options.c b/options.c
index 4be82eed6..b7193c62c 100644
--- a/options.c
+++ b/options.c
@@ -117,6 +117,12 @@ int human_readable = 1;
 int recurse = 0;
 int mkpath_dest_arg = 0;
 int allow_inc_recurse = 1;
+int stream_source_mode = 0;
+int stream_timeout = DEFAULT_STREAM_TIMEOUT;
+OFF_T stream_max_size = 0;
+int stream_pipe_sz = -1;
+char *stream_max_size_arg = NULL;
+char *stream_pipe_sz_arg = NULL;
 int xfer_dirs = -1;
 int am_daemon = 0;
 /* Set after a successful per-module chroot ("use chroot = yes") in
@@ -724,6 +730,10 @@ static struct poptOption long_options[] = {
   {"max-size",         0,  POPT_ARG_STRING, &max_size_arg, OPT_MAX_SIZE, 0, 0 },
   {"min-size",         0,  POPT_ARG_STRING, &min_size_arg, OPT_MIN_SIZE, 0, 0 },
   {"max-alloc",        0,  POPT_ARG_STRING, &max_alloc_arg, 0, 0, 0 },
+  {"stream-source",    0,  POPT_ARG_VAL,    &stream_source_mode, 1, 0, 0 },
+  {"stream-max-size",  0,  POPT_ARG_STRING, &stream_max_size_arg, 0, 0, 0 },
+  {"stream-timeout",   0,  POPT_ARG_INT,    &stream_timeout, 0, 0, 0 },
+  {"stream-pipe-sz",   0,  POPT_ARG_STRING, &stream_pipe_sz_arg, 0, 0, 0 },
   {"sparse",          'S', POPT_ARG_VAL,    &sparse_files, 1, 0, 0 },
   {"no-sparse",        0,  POPT_ARG_VAL,    &sparse_files, 0, 0, 0 },
   {"no-S",             0,  POPT_ARG_VAL,    &sparse_files, 0, 0, 0 },
@@ -2726,6 +2736,79 @@ int parse_arguments(int *argc_p, const char ***argv_p)
 			}
 		}
 	}
+	if (stream_source_mode) {
+		if (am_daemon || am_server) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used in server or daemon mode\n");
+			goto cleanup;
+		}
+		if (!stream_max_size_arg) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source requires a positive --stream-max-size\n");
+			goto cleanup;
+		}
+		if ((stream_max_size = parse_size_arg(stream_max_size_arg, 'b', "stream-max-size", 1, -1, False)) < 0)
+			goto cleanup;
+
+		if (stream_timeout == -1)
+			stream_timeout = DEFAULT_STREAM_TIMEOUT;
+		else if (stream_timeout < 0)
+			stream_timeout = 0;
+
+		if (!stream_pipe_sz_arg)
+			stream_pipe_sz = DEFAULT_STREAM_PIPE_SZ;
+		else {
+			ssize_t sz = parse_size_arg(stream_pipe_sz_arg, 'b', "stream-pipe-sz",
+						    MIN_STREAM_PIPE_SZ, MAX_STREAM_PIPE_SZ, False);
+			if (sz < 0)
+				goto cleanup;
+			stream_pipe_sz = (int)sz;
+		}
+		if (recurse || xfer_dirs > 1) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with --recursive (-r), --archive (-a), or --dirs (-d)\n");
+			goto cleanup;
+		}
+		if (append_mode) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with --append\n");
+			goto cleanup;
+		}
+		if (always_checksum) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with --checksum (-c)\n");
+			goto cleanup;
+		}
+		if (size_only || max_size_arg || min_size_arg || fuzzy_basis || preallocate_files) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with file-size-dependent options (--size-only, --max-size, --min-size, --fuzzy, --preallocate)\n");
+			goto cleanup;
+		}
+		if (files_from) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with --files-from\n");
+			goto cleanup;
+		}
+		if (remove_source_files) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with --remove-%s-files\n",
+				remove_source_files == 1 ? "source" : "sent");
+			goto cleanup;
+		}
+		if (write_batch || read_batch) {
+			snprintf(err_buf, sizeof err_buf,
+				"--stream-source cannot be used with batch mode\n");
+			goto cleanup;
+		}
+		ignore_times = 1;
+		update_only = 0;
+		preserve_specials = 0;
+		preserve_devices = 0;
+	} else if (stream_max_size_arg || stream_pipe_sz_arg || stream_timeout != DEFAULT_STREAM_TIMEOUT) {
+		snprintf(err_buf, sizeof err_buf,
+			"stream option(s) cannot be used without --stream-source\n");
+		goto cleanup;
+	}
 
 	if (trust_sender || am_server || read_batch)
 		trust_sender_args = trust_sender_filter = 1;
diff --git a/rsync.1.md b/rsync.1.md
index e447ed3eb..981b51ec4 100644
--- a/rsync.1.md
+++ b/rsync.1.md
@@ -618,6 +618,10 @@ has its own detailed description later in this manpage.
 --max-size=SIZE          don't transfer any file larger than SIZE
 --min-size=SIZE          don't transfer any file smaller than SIZE
 --max-alloc=SIZE         change a limit relating to memory alloc
+--stream-source          treat source as a live FIFO/pipe stream
+--stream-max-size=SIZE   max stream size ceiling (required with --stream-source)
+--stream-timeout=SECONDS abort stream if idle for SECONDS (default 30, 0=none)
+--stream-pipe-sz=SIZE    set pipe buffer size on Linux (default 1M, 64K-16M)
 --partial                keep partially transferred files
 --partial-dir=DIR        put a partially transferred file into DIR
 --delay-updates          put all updated files into place at end
@@ -4172,6 +4176,105 @@ sign) if you want the local shell to expand it.
     If a priority change is unsupported or fails, rsync warns and continues the
     transfer.
 
+0.  `--stream-source`
+
+    This option tells rsync to read a single named pipe (FIFO) on the local
+    sender and transfer its contents as a regular file to the destination.
+    Because pipes are non-seekable and report a length of 0 in `stat(2)`, this
+    mode requires [`--stream-max-size`](#opt) to define an upper byte ceiling
+    for the transfer.
+
+    When `--stream-source` is enabled, rsync automatically implies
+    [`--ignore-times`](#opt) (`-I`) and clears `--update`, `--links`,
+    `--specials`, and `--devices` so that the stream is never skipped by
+    quick-check comparisons or copied as a special FIFO node.  Both whole-file
+    transfers ([`--whole-file`](#opt)) and block-level delta transfers
+    (`--no-whole-file` and [`--inplace`](#opt)) are supported over
+    unidirectional streams without seeking on the source file descriptor.
+
+    Because a FIFO can only be read once sequentially and its final size is not
+    known until EOF, `--stream-source` cannot be combined with directory
+    recursion (`-r`, `-a`, `-d`, `--files-from`), multi-pass or append options
+    (`-c`, `--append`, `--append-verify`, batch mode), `--remove-source-files`,
+    or options whose semantics require a known source size up front
+    (`--size-only`, `--max-size`, `--min-size`, `-y`/`--fuzzy`,
+    `--preallocate`).  While `-a` is rejected because it implies `-r` and
+    `--specials`, individual metadata options such as `-p`, `-t`, `-o`, `-g`,
+    and `--chmod` can be used normally.
+
+    If the source argument is a symbolic link pointing to a pipe (such as
+    `/dev/stdin` or a shell process substitution `<(cmd)`), you must also
+    specify [`--copy-links`](#opt) (`-L`) so that rsync dereferences the
+    symlink to the underlying FIFO rather than skipping it as a non-regular
+    file.
+
+    For example, streaming a named pipe to a remote host using block-level
+    delta transfers in-place:
+
+    >     rsync --stream-source --stream-max-size=10G --no-whole-file --inplace \
+    >         /tmp/db_dump.fifo host:/backups/db_dump.sql
+
+    Or streaming from standard input or a process substitution using `-L`:
+
+    >     pg_dump mydb | rsync -L --stream-source --stream-max-size=10G \
+    >         /dev/stdin host:/backups/mydb.sql
+    >
+    >     rsync -L --stream-source --stream-max-size=500M \
+    >         <(tar -cf - /etc) host:/backups/etc.tar
+
+    This option is restricted to a local sender (local-to-local, remote-shell
+    push, or rsync-daemon push) and cannot be used when the sender is remote.
+
+0.  `--stream-max-size=SIZE`
+
+    This option specifies the maximum byte ceiling to read from a source FIFO
+    when [`--stream-source`](#opt) is active.  It is mandatory whenever
+    `--stream-source` is used.
+
+    `SIZE` acts as an upper bound rather than an exact file size:
+
+    - If the input stream reaches EOF before `SIZE` bytes are read, the
+      transfer completes cleanly at EOF and the destination file is sized to
+      the exact number of bytes received.  On the sender side, transfer
+      statistics (`--stats`) are reconciled to the actual number of bytes read
+      at EOF.  However, because `SIZE` is transmitted in the initial file list
+      without modifying the wire protocol, the advertised ceiling still
+      controls `%l` format logging, transfer progress percentages, and
+      receiver-side (or daemon-side) total-size statistics.
+    - If the input stream produces more than `SIZE` bytes, rsync stops reading
+      at `SIZE` bytes, logs a warning indicating that `--stream-max-size` was
+      exceeded, and finalizes the destination file at that ceiling length.
+
+    The `SIZE` value can be suffixed with a unit multiplier such as `K` (or
+    `KiB`), `M` (or `MiB`), or `G` (or `GiB`).  See the [`--max-size`](#opt)
+    option for a full description of all available suffixes.
+
+0.  `--stream-timeout=SECONDS`
+
+    This option specifies the maximum number of seconds that the sender will
+    wait for data when reading from a source FIFO when [`--stream-source`](#opt)
+    is active.  If no data arrives from the upstream writer process within the
+    specified time, rsync aborts the transfer with an error.
+
+    The default timeout is 30 seconds.  Specifying a value of 0 (or a negative
+    value) disables the stream read timeout, causing rsync to wait indefinitely
+    for pipe data.
+
+0.  `--stream-pipe-sz=SIZE`
+
+    This option instructs rsync to resize the kernel pipe buffer
+    (`fcntl(fd, F_SETPIPE_SZ, SIZE)`) on the source FIFO on Linux when
+    [`--stream-source`](#opt) is active.  Increasing the pipe buffer size
+    reduces context switching and prevents the upstream writer process from
+    blocking frequently during high-throughput delta transfers.
+
+    The default pipe buffer size is 1048576 bytes (1 MiB).  The specified
+    `SIZE` must be between 65536 bytes (64 KiB) and 16777216 bytes (16 MiB).
+    If the kernel rejects the requested size (for example, when `SIZE` exceeds
+    `/proc/sys/fs/pipe-max-size` and the user lacks `CAP_SYS_RESOURCE`), rsync
+    continues with the existing kernel pipe buffer size.  On non-Linux systems
+    where `F_SETPIPE_SZ` is unavailable, this option has no effect.
+
 ## DAEMON OPTIONS
 
 The options allowed when starting an rsync daemon are as follows:
diff --git a/rsync.h b/rsync.h
index adf9bf1f1..4a088cc29 100644
--- a/rsync.h
+++ b/rsync.h
@@ -26,6 +26,11 @@
 #define RSYNC_RSH_ENV "RSYNC_RSH"
 #define RSYNC_RSH_IO_ENV "RSYNC_RSH_IO"
 
+#define DEFAULT_STREAM_TIMEOUT 30
+#define DEFAULT_STREAM_PIPE_SZ (1024 * 1024)
+#define MIN_STREAM_PIPE_SZ (64 * 1024)
+#define MAX_STREAM_PIPE_SZ (16 * 1024 * 1024)
+
 #define RSYNC_NAME "rsync"
 /* RSYNCD_SYSCONF is now set in config.h */
 #define RSYNCD_USERCONF "rsyncd.conf"
diff --git a/sender.c b/sender.c
index c1b903960..080d13a09 100644
--- a/sender.c
+++ b/sender.c
@@ -26,6 +26,7 @@ extern int do_xfers;
 extern int open_noatime;
 extern int am_server;
 extern int am_daemon;
+extern int am_sender;
 extern int local_server;
 extern int inc_recurse;
 extern int log_before_transfer;
@@ -63,6 +64,10 @@ extern BOOL want_progress_now;
 extern struct stats stats;
 extern struct file_list *cur_flist, *first_flist, *dir_flist;
 extern char num_dev_ino_buf[4 + 8 + 8];
+extern int stream_source_mode;
+extern int stream_timeout;
+extern OFF_T stream_max_size;
+extern int stream_pipe_sz;
 
 BOOL extra_flist_sending_enabled;
 
@@ -206,6 +211,11 @@ static int secure_sender_parent_fd(struct file_struct *file, const char *fname,
 #endif
 }
 
+int is_stream()
+{
+	return stream_source_mode && am_sender && !am_server;
+}
+
 /* Go through the do_*() wrapper rather than a raw unlinkat(): it carries the
  * dry_run no-op and the read-only/list-only refusal that do_unlink() applies
  * on the non-fd path, plus the missing-AT_FDCWD fallback. */
@@ -743,6 +753,17 @@ void send_files(int f_in, int f_out)
 			exit_cleanup(RERR_FILEIO);
 		}
 
+		if (is_stream()) {
+			if (!S_ISFIFO(st.st_mode)) {
+				rprintf(FERROR, "stream-source requires a named pipe (FIFO): %s\n",
+					full_fname(fname));
+				close(fd);
+				exit_cleanup(RERR_FILEIO);
+			}
+			st.st_mode = (st.st_mode & ~S_IFMT) | S_IFREG;
+			st.st_size = stream_max_size;
+		}
+
 		if (IS_DEVICE(st.st_mode)) {
 			if (!copy_devices) {
 				rprintf(FERROR, "attempt to copy device contents without --copy-devices\n");
@@ -787,6 +808,26 @@ void send_files(int f_in, int f_out)
 		set_compression(fname);
 
 		match_sums(f_xfer, s, mbuf, st.st_size);
+		/* Update the stats for right tracking. */
+		if (is_stream() && mbuf) {
+			if (mbuf->file_size < st.st_size) {
+				OFF_T diff = st.st_size - mbuf->file_size;
+				stats.total_size -= diff;
+				stats.total_transferred_size -= diff;
+				st.st_size = mbuf->file_size;
+			} else {
+				char extra;
+				ssize_t n;
+				do {
+					n = read(fd, &extra, 1);
+				} while (n < 0 && errno == EINTR);
+				if (n > 0) {
+					rprintf(FWARNING,
+						"Warning: stream source exceeded --stream-max-size=%s, stopping at limit\n",
+						big_num(stream_max_size));
+				}
+			}
+		}
 		if (INFO_GTE(PROGRESS, 1))
 			end_progress(st.st_size);
 		else if (want_progress_now)
diff --git a/testsuite/stream-early-eof_test.py b/testsuite/stream-early-eof_test.py
new file mode 100644
index 000000000..1ad0b0555
--- /dev/null
+++ b/testsuite/stream-early-eof_test.py
@@ -0,0 +1,295 @@
+#!/usr/bin/env python3
+"""Regression test for --stream-source EOF length updates and stats accuracy."""
+
+import os
+from pathlib import Path
+import re
+import threading
+
+from rsyncfns import (
+    FROMDIR, TODIR, assert_same, rmtree, run_rsync, test_fail,
+)
+
+if not hasattr(os, 'mkfifo'):
+    test_skipped('os.mkfifo is not available on this platform')
+
+_probe = run_rsync('--help', capture_output=True, check=False)
+if _probe.returncode != 0 or '--stream-source' not in _probe.stdout:
+    test_skipped('rsync binary does not support --stream-source')
+
+WIN_SZ = 263168
+BLOCK = 700
+TAIL_LEN = 123
+C_BYTE = 100
+
+
+def make_det_bytes(length: int, seed: int) -> bytes:
+    out = bytearray(length)
+    x = (seed & 0xFFFFFFFF) or 0x12345678
+    for i in range(length):
+        x ^= (x << 13) & 0xFFFFFFFF
+        x ^= (x >> 17) & 0xFFFFFFFF
+        x ^= (x << 5) & 0xFFFFFFFF
+        out[i] = x & 0xFF
+    return bytes(out)
+
+
+def make_spread_colliding_block(delta: int, fill: int = C_BYTE) -> bytes:
+    b = bytearray([fill]) * BLOCK
+    b[0] = fill + delta
+    b[349] = fill - delta
+    b[350] = fill - delta
+    b[699] = fill + delta
+    return bytes(b)
+
+
+def parse_stats(label: str, out: str):
+    m_delta = re.search(
+        r'total:\s+matches=(\d+)\s+hash_hits=(\d+)\s+false_alarms=(\d+)\s+data=(\d+)',
+        out,
+    )
+    m_fsize = re.search(r'Total file size:\s*([\d,.]+)\s*bytes', out)
+    m_lit = re.search(r'Literal data:\s*([\d,.]+)\s*bytes', out)
+    m_match = re.search(r'Matched data:\s*([\d,.]+)\s*bytes', out)
+    if not m_delta or not m_fsize or not m_lit or not m_match:
+        test_fail(f"[{label}] could not find deltasum2/stats in rsync output:\n{out}")
+    return {
+        'matches': int(m_delta.group(1)),
+        'hash_hits': int(m_delta.group(2)),
+        'false_alarms': int(m_delta.group(3)),
+        'delta_data': int(m_delta.group(4)),
+        'total_file_size': int(m_fsize.group(1).replace(',', '').replace('.', '')),
+        'literal': int(m_lit.group(1).replace(',', '').replace('.', '')),
+        'matched': int(m_match.group(1).replace(',', '').replace('.', '')),
+    }
+
+
+def run_stream_rsync(label: str, fifo_path: Path, payload: bytes, dst_path: Path, *extra_args):
+    writer_exc = []
+
+    def writer():
+        try:
+            fd = os.open(fifo_path, os.O_WRONLY)
+            try:
+                view = memoryview(payload)
+                total = 0
+                while total < len(view):
+                    written = os.write(fd, view[total : total + 65536])
+                    total += written
+            finally:
+                os.close(fd)
+        except Exception as exc:
+            writer_exc.append(exc)
+
+    t = threading.Thread(target=writer, daemon=True)
+    t.start()
+    proc = run_rsync(
+        '--stream-source',
+        '--stream-max-size=10M',
+        '--no-whole-file',
+        f'--block-size={BLOCK}',
+        '--no-compress',
+        '--stats',
+        '--debug=deltasum2',
+        *extra_args,
+        f'{fifo_path}',
+        f'{dst_path}',
+        capture_output=True,
+        check=False,
+    )
+    t.join(timeout=10)
+    if proc.returncode != 0:
+        test_fail(
+            f"[FAIL] {label}: rsync exited with code {proc.returncode}\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    if writer_exc:
+        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
+    return proc
+
+
+def assert_matches_regular(
+    label: str,
+    fifo_path: Path,
+    payload: bytes,
+    basis: bytes,
+    *extra_args,
+):
+    print(f"[RUN] {label} (payload={len(payload)}B, basis={len(basis)}B)", flush=True)
+    src_reg = FROMDIR / 'ref_src.bin'
+    dst_reg = TODIR / 'dst_reg.bin'
+    dst_stream = TODIR / 'dst_stream.bin'
+
+    src_reg.write_bytes(payload)
+    dst_reg.write_bytes(basis)
+    dst_stream.write_bytes(basis)
+
+    proc_reg = run_rsync(
+        '-I',
+        '--no-whole-file',
+        f'--block-size={BLOCK}',
+        '--no-compress',
+        '--stats',
+        '--debug=deltasum2',
+        *extra_args,
+        f'{src_reg}',
+        f'{dst_reg}',
+        capture_output=True,
+    )
+    proc_stream = run_stream_rsync(label, fifo_path, payload, dst_stream, *extra_args)
+
+    s_reg = parse_stats(label, proc_reg.stdout + proc_reg.stderr)
+    s_stream = parse_stats(label, proc_stream.stdout + proc_stream.stderr)
+
+    actual_len = len(payload)
+
+    if s_stream['literal'] + s_stream['matched'] != actual_len:
+        test_fail(
+            f"[FAIL] {label}: stream length not updated! "
+            f"literal ({s_stream['literal']}) + matched ({s_stream['matched']}) "
+            f"!= actual stream length ({actual_len})"
+        )
+
+    if s_stream['delta_data'] != s_stream['literal']:
+        test_fail(
+            f"[FAIL] {label}: deltasum2 data={s_stream['delta_data']} "
+            f"!= literal={s_stream['literal']}"
+        )
+
+    for k in ('matches', 'hash_hits', 'false_alarms', 'delta_data', 'literal', 'matched'):
+        if s_reg[k] != s_stream[k]:
+            test_fail(
+                f"[FAIL] {label}: stream stat '{k}' differs from regular file!\n"
+                f"  regular = {s_reg}\n"
+                f"  stream  = {s_stream}"
+            )
+
+    assert_same(dst_stream, src_reg, label=label)
+    print(f"[PASS] {label}: {s_stream}", flush=True)
+    return s_stream
+
+
+rmtree(FROMDIR)
+rmtree(TODIR)
+FROMDIR.mkdir(parents=True, exist_ok=True)
+TODIR.mkdir(parents=True, exist_ok=True)
+
+fifo = FROMDIR / 'stream.fifo'
+os.mkfifo(fifo)
+
+basis_10 = make_det_bytes(BLOCK * 10, seed=101)
+prefix_350 = make_det_bytes(350, seed=202)
+tail_123 = make_det_bytes(TAIL_LEN, seed=303)
+basis_short_tail = basis_10 + tail_123
+random_800 = make_det_bytes(800, seed=404)
+zeros_basis_10 = b'\x00' * (BLOCK * 10)
+
+# ---------------------------------------------------------------------------
+# Part 1: Early-EOF (< 263,168B initial window)
+# ---------------------------------------------------------------------------
+assert_matches_regular(
+    'Early 1: 0-byte EOF against zero basis',
+    fifo, b'', zeros_basis_10,
+)
+
+assert_matches_regular(
+    'Early 2: 100B stream < last basis block',
+    fifo, b'\x00' * 100, zeros_basis_10,
+)
+
+assert_matches_regular(
+    'Early 3: exact 123B short tail block match at offset=0',
+    fifo, tail_123, basis_short_tail,
+)
+
+assert_matches_regular(
+    'Early 4: exact 1-block match at EOF',
+    fifo, basis_10[:BLOCK], basis_10,
+)
+
+assert_matches_regular(
+    'Early 5: exact 3-block match at EOF',
+    fifo, basis_10[: 3 * BLOCK], basis_10,
+)
+
+assert_matches_regular(
+    'Early 6: unaligned 350B literal + 1-block match at EOF',
+    fifo, prefix_350 + basis_10[:BLOCK], basis_10,
+)
+
+assert_matches_regular(
+    'Early 7: short tail basis + unmatched 800B stream',
+    fifo, random_800, basis_short_tail,
+)
+
+assert_matches_regular(
+    'Early 8: 350B literal + final short basis block (123B) at EOF',
+    fifo, prefix_350 + tail_123, basis_short_tail,
+)
+
+head_100 = make_det_bytes(100, seed=505) + (b'\x00' * (BLOCK * 2))
+assert_matches_regular(
+    'Early 9: --inplace zero-block realignment at early EOF',
+    fifo, head_100, zeros_basis_10,
+    '--inplace',
+)
+
+# ---------------------------------------------------------------------------
+# Part 2: Late-EOF (> 263,168B initial def_window_size)
+# ---------------------------------------------------------------------------
+blk0 = make_det_bytes(BLOCK, seed=606)
+basis_2 = blk0 + tail_123
+random_win = make_det_bytes(WIN_SZ, seed=707)
+random_577 = make_det_bytes(577, seed=808)
+
+s_late1 = assert_matches_regular(
+    'Late 1: EOF after 1st window + 123B short tail match',
+    fifo, random_win + tail_123, basis_2,
+)
+
+s_late2 = assert_matches_regular(
+    'Late 2: exact 263,168B window boundary EOF',
+    fifo, random_win[: WIN_SZ - TAIL_LEN] + tail_123, basis_2,
+)
+
+s_late3 = assert_matches_regular(
+    'Late 3: 263,168B window boundary + 1B EOF in null_hash',
+    fifo, random_win[: WIN_SZ - TAIL_LEN + 1] + tail_123, basis_2,
+)
+
+s_late4 = assert_matches_regular(
+    'Late 4: 263,168B window boundary - 1B EOF',
+    fifo, random_win[: WIN_SZ - TAIL_LEN - 1] + tail_123, basis_2,
+)
+
+s_late5 = assert_matches_regular(
+    'Late 5: match at 262,467B + post-matched() hits EOF across boundary',
+    fifo,
+    random_win[: WIN_SZ - BLOCK - 1] + blk0 + random_577[:78] + tail_123,
+    basis_2,
+)
+
+s_late6 = assert_matches_regular(
+    'Late 6: match before boundary + null_hash EOF at boundary',
+    fifo,
+    random_win[: WIN_SZ - 1400] + blk0 + random_577 + tail_123,
+    basis_2,
+)
+
+zeros_basis_large = (b'\x00' * (BLOCK * 400)) + tail_123
+s_late7 = assert_matches_regular(
+    'Late 7: --inplace realignment lookahead hits EOF across boundary',
+    fifo,
+    random_win[: WIN_SZ - 708] + (b'\x00' * 720),
+    zeros_basis_large,
+    '--inplace',
+)
+
+coll_block = make_spread_colliding_block(5)
+basis_coll_tail = (coll_block * 4) + tail_123
+flood_with_tail = bytes([C_BYTE]) * (WIN_SZ - 101) + tail_123
+s_late8 = assert_matches_regular(
+    'Late 8: collision skip hits EOF across boundary + clamps k for tail_123',
+    fifo, flood_with_tail, basis_coll_tail,
+)
diff --git a/testsuite/stream-source_test.py b/testsuite/stream-source_test.py
new file mode 100644
index 000000000..6e5efde61
--- /dev/null
+++ b/testsuite/stream-source_test.py
@@ -0,0 +1,302 @@
+"""Regression test for --stream-source option validation, bounds, and refused flags."""
+
+import os
+from pathlib import Path
+import threading
+
+from rsyncfns import (
+    FROMDIR, TODIR, assert_same, rmtree, run_rsync, test_fail, test_skipped,
+)
+
+if not hasattr(os, 'mkfifo'):
+    test_skipped('os.mkfifo is not available on this platform')
+
+_probe = run_rsync('--help', capture_output=True, check=False)
+if _probe.returncode != 0 or '--stream-source' not in _probe.stdout:
+    test_skipped('rsync binary does not support --stream-source')
+
+
+def expect_fail(label: str, *args, expected_err: str = ''):
+    print(f"[RUN] {label}", flush=True)
+    proc = run_rsync(*args, capture_output=True, check=False)
+    if proc.returncode == 0:
+        test_fail(
+            f"[FAIL] {label}: expected non-zero exit code, got 0\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    combined = (proc.stdout + proc.stderr).lower()
+    if expected_err and expected_err.lower() not in combined:
+        test_fail(
+            f"[FAIL] {label}: expected '{expected_err}' in output, got:\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    print(f"[PASS] {label} (exit={proc.returncode})", flush=True)
+    return proc
+
+
+def run_stream_transfer(label: str, fifo_path: Path, payload: bytes, dst_path: Path, *extra_args):
+    print(f"[RUN] {label}", flush=True)
+    writer_exc = []
+
+    def writer():
+        try:
+            fd = os.open(fifo_path, os.O_WRONLY)
+            try:
+                view = memoryview(payload)
+                total = 0
+                while total < len(view):
+                    written = os.write(fd, view[total : total + 65536])
+                    total += written
+            except BrokenPipeError:
+                pass
+            finally:
+                os.close(fd)
+        except Exception as exc:
+            writer_exc.append(exc)
+
+    t = threading.Thread(target=writer, daemon=True)
+    t.start()
+    proc = run_rsync(
+        '--stream-source',
+        *extra_args,
+        f'{fifo_path}',
+        f'{dst_path}',
+        capture_output=True,
+        check=False,
+    )
+    t.join(timeout=10)
+    if proc.returncode != 0:
+        test_fail(
+            f"[FAIL] {label}: rsync exited with code {proc.returncode}\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    if writer_exc:
+        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
+    print(f"[PASS] {label}", flush=True)
+    return proc
+
+
+rmtree(FROMDIR)
+rmtree(TODIR)
+FROMDIR.mkdir(parents=True, exist_ok=True)
+TODIR.mkdir(parents=True, exist_ok=True)
+
+fifo = FROMDIR / 'stream.fifo'
+try:
+    os.mkfifo(fifo)
+except OSError as exc:
+    test_skipped(f"cannot create FIFO in test filesystem: {exc}")
+
+reg_src = FROMDIR / 'reg.bin'
+reg_src.write_bytes(b'regular file payload')
+dst = TODIR / 'dst.bin'
+
+# ---------------------------------------------------------------------------
+# Part 1: Help output (--help includes stream options and FIFO usage)
+# ---------------------------------------------------------------------------
+label = "Main --help includes --stream-source options and FIFO usage"
+print(f"[RUN] {label}", flush=True)
+proc_main_help = run_rsync('--help', capture_output=True, check=False)
+if proc_main_help.returncode != 0:
+    test_fail(f"[FAIL] {label}: exited with {proc_main_help.returncode}\n{proc_main_help.stderr}")
+for expected in (
+    'FIFO DEST',
+    '--stream-source',
+    '--stream-max-size=SIZE',
+    '--stream-timeout=SECONDS',
+    '--stream-pipe-sz=SIZE',
+):
+    if expected not in proc_main_help.stdout:
+        test_fail(f"[FAIL] {label}: missing '{expected}' in help output:\n{proc_main_help.stdout}")
+print(f"[PASS] {label}", flush=True)
+
+# ---------------------------------------------------------------------------
+# Part 2: Sub-options used without --stream-source must be refused
+# ---------------------------------------------------------------------------
+for sub_opt in ('--stream-max-size=1M', '--stream-timeout=5', '--stream-pipe-sz=1M'):
+    expect_fail(
+        f"Refuse {sub_opt} without --stream-source",
+        sub_opt, f'{reg_src}', f'{dst}',
+        expected_err='--stream-source',
+    )
+
+# ---------------------------------------------------------------------------
+# Part 3: --stream-max-size validation (missing, <= 0, invalid, truncation log)
+# ---------------------------------------------------------------------------
+expect_fail(
+    "Missing --stream-max-size with --stream-source",
+    '--stream-source', f'{fifo}', f'{dst}',
+    expected_err='--stream-max-size',
+)
+
+for bad_max in ('0', '0B', '-1', '-10M', 'invalid'):
+    expect_fail(
+        f"Invalid --stream-max-size={bad_max}",
+        '--stream-source', f'--stream-max-size={bad_max}', f'{fifo}', f'{dst}',
+    )
+
+# Stream exceeding --stream-max-size logs warning and stops at limit
+if dst.exists():
+    dst.unlink()
+over_payload = b'X' * 4096
+ref_1k = FROMDIR / 'ref_1k.bin'
+ref_1k.write_bytes(over_payload[:1024])
+
+proc_over = run_stream_transfer(
+    "Stream exceeding --stream-max-size=1024 logs and stops at limit",
+    fifo, over_payload, dst,
+    '--stream-max-size=1024',
+)
+combined_over = (proc_over.stdout + proc_over.stderr).lower()
+if 'exceeded' not in combined_over and 'limit' not in combined_over:
+    test_fail(
+        f"[FAIL] Expected warning when stream exceeds --stream-max-size, got:\n"
+        f"--- stdout ---\n{proc_over.stdout}\n"
+        f"--- stderr ---\n{proc_over.stderr}"
+    )
+assert_same(dst, ref_1k, label="stream-max-size=1024 cap")
+
+# ---------------------------------------------------------------------------
+# Part 4: --stream-pipe-sz min/max bounds (64KB .. 16MB)
+# ---------------------------------------------------------------------------
+for bad_pipe_sz in ('0', '-1', '32K', '65535', '17M', '16777217', 'invalid'):
+    expect_fail(
+        f"Out-of-bounds --stream-pipe-sz={bad_pipe_sz}",
+        '--stream-source', '--stream-max-size=1M', f'--stream-pipe-sz={bad_pipe_sz}',
+        f'{fifo}', f'{dst}',
+    )
+
+payload_4k = b'A' * 4096
+ref_4k = FROMDIR / 'ref_4k.bin'
+ref_4k.write_bytes(payload_4k)
+
+for valid_pipe_sz in ('64K', '65536', '1M', '16M', '16777216'):
+    if dst.exists():
+        dst.unlink()
+    run_stream_transfer(
+        f"Valid --stream-pipe-sz={valid_pipe_sz}",
+        fifo, payload_4k, dst,
+        '--stream-max-size=1M', f'--stream-pipe-sz={valid_pipe_sz}',
+    )
+    assert_same(dst, ref_4k, label=f"pipe-sz={valid_pipe_sz}")
+
+# ---------------------------------------------------------------------------
+# Part 5: --stream-timeout (<= 0 disables timeout; positive expires on stall)
+# ---------------------------------------------------------------------------
+for no_timeout in ('0', '-1'):
+    if dst.exists():
+        dst.unlink()
+    run_stream_transfer(
+        f"Disabled timeout with --stream-timeout={no_timeout}",
+        fifo, payload_4k, dst,
+        '--stream-max-size=1M', f'--stream-timeout={no_timeout}',
+    )
+    assert_same(dst, ref_4k, label=f"timeout={no_timeout}")
+
+stall_stop = threading.Event()
+
+
+def stalled_writer():
+    try:
+        fd = os.open(fifo, os.O_WRONLY)
+        try:
+            os.write(fd, b'partial bytes before stall')
+            stall_stop.wait(timeout=5)
+        finally:
+            os.close(fd)
+    except Exception:
+        pass
+
+
+t_stall = threading.Thread(target=stalled_writer, daemon=True)
+t_stall.start()
+try:
+    expect_fail(
+        "Stalled stream hits --stream-timeout=1",
+        '--stream-source', '--stream-max-size=1M', '--stream-timeout=1',
+        f'{fifo}', f'{dst}',
+        expected_err='timed out',
+    )
+finally:
+    stall_stop.set()
+    t_stall.join(timeout=5)
+
+# ---------------------------------------------------------------------------
+# Part 6: Refused / incompatible options with --stream-source
+# ---------------------------------------------------------------------------
+refused_flags = [
+    ('-r',),
+    ('-a',),
+    ('-d',),
+    ('--append',),
+    ('--append-verify',),
+    ('-c',),
+    ('--checksum',),
+    ('--size-only',),
+    ('--max-size=1M',),
+    ('--min-size=1K',),
+    ('-y',),
+    ('--fuzzy',),
+    ('--remove-source-files',),
+    ('--preallocate',),
+]
+
+for flags in refused_flags:
+    flag_str = ' '.join(flags)
+    expect_fail(
+        f"Refuse incompatible flag {flag_str} with --stream-source",
+        '--stream-source', '--stream-max-size=1M', *flags, f'{fifo}', f'{dst}',
+    )
+
+flist_file = FROMDIR / 'flist.txt'
+flist_file.write_text('stream.fifo\n')
+expect_fail(
+    "Refuse --files-from with --stream-source",
+    '--stream-source', '--stream-max-size=1M', f'--files-from={flist_file}',
+    f'{FROMDIR}', f'{TODIR}',
+)
+
+batch_prefix = TODIR / 'batch_out'
+expect_fail(
+    "Refuse --write-batch with --stream-source",
+    '--stream-source', '--stream-max-size=1M', f'--write-batch={batch_prefix}',
+    f'{fifo}', f'{dst}',
+)
+
+expect_fail(
+    "Refuse multiple sources with --stream-source",
+    '--stream-source', '--stream-max-size=1M', f'{fifo}', f'{reg_src}', f'{TODIR}/',
+)
+
+expect_fail(
+    "Refuse --daemon with --stream-source",
+    '--daemon', '--stream-source', '--stream-max-size=1M',
+    expected_err='--stream-source',
+)
+
+expect_fail(
+    "Refuse --server with --stream-source",
+    '--server', '--stream-source', '--stream-max-size=1M', '.', f'{dst}',
+    expected_err='--stream-source',
+)
+
+expect_fail(
+    "Refuse remote shell source pull (HOST:SRC) with --stream-source",
+    '--stream-source', '--stream-max-size=1M', f'localhost:{fifo}', f'{dst}',
+    expected_err='--stream-source',
+)
+
+expect_fail(
+    "Refuse remote daemon source pull (HOST::SRC) with --stream-source",
+    '--stream-source', '--stream-max-size=1M', 'localhost::module/stream.fifo', f'{dst}',
+    expected_err='--stream-source',
+)
+
+expect_fail(
+    "Refuse rsync:// URL source pull with --stream-source",
+    '--stream-source', '--stream-max-size=1M', 'rsync://localhost/module/stream.fifo', f'{dst}',
+    expected_err='--stream-source',
+)
diff --git a/usage.c b/usage.c
index de586839c..9dd9d8627 100644
--- a/usage.c
+++ b/usage.c
@@ -355,8 +355,13 @@ void usage(enum logcode F)
   rprintf(F,"  or   rsync [OPTION]... [USER@]HOST:SRC [DEST]\n");
   rprintf(F,"  or   rsync [OPTION]... [USER@]HOST::SRC [DEST]\n");
   rprintf(F,"  or   rsync [OPTION]... rsync://[USER@]HOST[:PORT]/SRC [DEST]\n");
+  rprintf(F,"  or   rsync [OPTION]... --stream-source --stream-max-size=SIZE FIFO DEST\n");
+  rprintf(F,"  or   rsync [OPTION]... --stream-source --stream-max-size=SIZE FIFO [USER@]HOST:DEST\n");
+  rprintf(F,"  or   rsync [OPTION]... --stream-source --stream-max-size=SIZE FIFO [USER@]HOST::DEST\n");
+  rprintf(F,"  or   rsync [OPTION]... --stream-source --stream-max-size=SIZE FIFO rsync://[USER@]HOST[:PORT]/DEST\n");
   rprintf(F,"The ':' usages connect via remote shell, while '::' & 'rsync://' usages connect\n");
   rprintf(F,"to an rsync daemon, and require SRC or DEST to start with a module name.\n");
+  rprintf(F,"The '--stream-source' usages require a single local FIFO/pipe source.\n");
   rprintf(F,"\n");
   rprintf(F,"Options\n");
 #include "help-rsync.h"

From 6504328b55cb677a403995c0d31fc524da4ca753 Mon Sep 17 00:00:00 2001
From: seks99x <omarelsayed161@gmail.com>
Date: Sat, 7 Nov 2026 01:42:49 +0300
Subject: [PATCH 2/4] Fix stream-source post-EOF select() hang on macOS/NetBSD
 and scan-build warning

On macOS (XNU) and NetBSD, named FIFOs clear their EOF socket state (SBS_CANTRCVMORE) after a 0-byte read() is consumed so the FIFO can be reused by future writers. If map_ptr() is called again before the caller clamps its length to the shrunk buf->file_size, a subsequent select() on the FIFO sees zero writers and no EOF flag, blocking until stream_timeout expires.

Track stream EOF via eof_reached in fileio.c (reset in map_file()) so map_ptr() and send_files() never invoke select() or read() on the stream descriptor once a 0-byte read() has been observed. Also guard buf before dereferencing buf->file_size in match_sums() to silence a Clang scan-build null-dereference warning.
---
 fileio.c | 9 ++++++---
 match.c  | 4 ++--
 sender.c | 3 ++-
 3 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/fileio.c b/fileio.c
index 2ebb2a620..266a51161 100644
--- a/fileio.c
+++ b/fileio.c
@@ -38,6 +38,7 @@ extern struct stats stats;
 extern int stream_timeout;
 extern int stream_pipe_sz;
 
+int eof_reached = 0;
 OFF_T preallocated_len = 0;
 
 static OFF_T sparse_seek = 0;
@@ -333,6 +334,7 @@ struct map_struct *map_file(int fd, OFF_T len, int32 read_size, int32 blk_size)
 	map->def_window_size = ALIGNED_LENGTH(read_size);
 
 	if (is_stream()) {
+		eof_reached = 0;
 #ifdef F_SETPIPE_SZ
 		if (fcntl(map->fd, F_SETPIPE_SZ, stream_pipe_sz) < 0) {
 			if (stream_pipe_sz != DEFAULT_STREAM_PIPE_SZ || errno != EPERM)
@@ -421,7 +423,7 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 	map->p_len = window_size;
 
 	while (read_size > 0) {
-		if (is_stream() && stream_timeout > 0) {
+		if (is_stream() && !eof_reached && stream_timeout > 0) {
 			fd_set rfds;
 			struct timeval tv;
 
@@ -442,7 +444,7 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 				exit_cleanup(RERR_FILEIO);
 			}
 		}
-		int32 nread = read(map->fd, map->p + read_offset, read_size);
+		int32 nread = eof_reached ? 0 : read(map->fd, map->p + read_offset, read_size);
 		if (nread <= 0) {
 			/* Retry if a blocking pipe read was interrupted by a signal. */
 			if (nread < 0 && errno == EINTR)
@@ -450,10 +452,11 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 			/* On early EOF from a FIFO/pipe, shrink file_size and window
 			 * length to the actual bytes read so far and zero the tail. */
 			if (is_stream() && nread == 0) {
-				if (INFO_GTE(MISC, 1)) {
+				if (!eof_reached && INFO_GTE(MISC, 1)) {
 					rprintf(FINFO, "Stream reached early EOF at %s bytes (ceiling was %s)\n",
 						big_num(map->p_fd_offset), big_num(map->file_size));
 				}
+				eof_reached = 1;
 				map->file_size = map->p_fd_offset;
 				map->p_len = read_offset;
 				memset(map->p + read_offset, 0, read_size);
diff --git a/match.c b/match.c
index b5a2b1bb0..1ffdad3b2 100644
--- a/match.c
+++ b/match.c
@@ -79,7 +79,7 @@ static int32 *hash_table;
  * (map_ptr() shrinks buf->file_size to the actual bytes read on pipe EOF). */
 static int stream_early_EOF(struct map_struct *buf, OFF_T len)
 {
-	return is_stream() && buf->file_size < len;
+	return is_stream() && buf && buf->file_size < len;
 }
 
 static void build_hash_table(struct sum_struct *s)
@@ -528,7 +528,7 @@ void match_sums(int f, struct sum_struct *s, struct map_struct *buf, OFF_T len)
 			/* Pre-read the next chunk on a stream before calling matched() so
 			 * that if the pipe hits early EOF mid-chunk, buf->file_size and len
 			 * shrink before matched() tries to flush up to 'j'. */
-			if (is_stream()) {
+			if (is_stream() && buf) {
 				map_ptr(buf, last_match, (int32)MIN((OFF_T)CHUNK_SIZE, len - last_match));
 				if (buf->file_size < len)
 					len = buf->file_size;
diff --git a/sender.c b/sender.c
index 080d13a09..30e777aaa 100644
--- a/sender.c
+++ b/sender.c
@@ -46,6 +46,7 @@ extern int whole_file;
 extern int allowed_lull;
 extern int copy_devices;
 extern int preserve_xattrs;
+extern int eof_reached;
 extern int protocol_version;
 extern int remove_source_files;
 extern int updating_basis_file;
@@ -815,7 +816,7 @@ void send_files(int f_in, int f_out)
 				stats.total_size -= diff;
 				stats.total_transferred_size -= diff;
 				st.st_size = mbuf->file_size;
-			} else {
+			} else if (!eof_reached) {
 				char extra;
 				ssize_t n;
 				do {

From cfa7746242fdfd5c065ea929453653cfe35a7fc7 Mon Sep 17 00:00:00 2001
From: seks99x <omarelsayed161@gmail.com>
Date: Tue, 10 Nov 2026 22:12:12 +0300
Subject: [PATCH 3/4] Address review feedback on --stream-source validation and
 EOF handling

- Verify EOF or excess stream bytes in match_sums() before finalising the
  transfer checksum so streams exceeding --stream-max-size fail with
  RERR_FILEIO (exit code 11) without truncating or overwriting existing
  destination files, and stalled streams at the size boundary honour
  --stream-timeout via poll().
- Track --stream-timeout explicitly with OPT_STREAM_TIMEOUT instead of a
  -1 sentinel so negative values disable the timeout consistently with 0,
  matching the documented behaviour.
- Reject multiple source arguments in start_client() before opening any
  FIFO when --stream-source is active.
- Update rsync.1.md to document pre-finalisation abort behaviour on
  oversized streams and the single-source requirement.
- Expand stream-source_test.py to cover exact-ceiling transfers, oversized
  stream aborts preserving destination files, paused/stalled stream
  timeouts, and writerless multi-FIFO rejection in both argument orders.
---
 fileio.c                        |  54 +++++---
 main.c                          |   4 +
 match.c                         |  23 ++++
 options.c                       |  15 ++-
 rsync.1.md                      |  29 ++--
 sender.c                        |  23 +---
 testsuite/stream-source_test.py | 232 ++++++++++++++++++++++++++++----
 7 files changed, 297 insertions(+), 83 deletions(-)

diff --git a/fileio.c b/fileio.c
index 266a51161..0a9a1605a 100644
--- a/fileio.c
+++ b/fileio.c
@@ -21,6 +21,7 @@
 
 #include "rsync.h"
 #include "inums.h"
+#include <poll.h>
 
 #ifndef ENODATA
 #define ENODATA EAGAIN
@@ -316,6 +317,37 @@ int skip_matched(int fd, OFF_T offset, const char *buf, int len)
 	return 0;
 }
 
+void wait_stream_fd(int fd)
+{
+	if (!is_stream() || eof_reached || stream_timeout <= 0)
+		return;
+
+	int timeout_ms = stream_timeout > INT_MAX / 1000
+		       ? INT_MAX
+		       : stream_timeout * 1000;
+
+	for (;;) {
+		struct pollfd pfd;
+		pfd.fd = fd;
+		pfd.events = POLLIN;
+		pfd.revents = 0;
+
+		int res = poll(&pfd, 1, timeout_ms);
+		if (res == 0) {
+			rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
+				stream_timeout);
+			exit_cleanup(RERR_TIMEOUT);
+		}
+		if (res < 0) {
+			if (errno == EINTR)
+				continue;
+			rsyserr(FERROR, errno, "poll failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		break;
+	}
+}
+
 /* This provides functionality somewhat similar to mmap() but using read().
  * It gives sliding window access to a file.  mmap() is not used because of
  * the possibility of another program (such as a mailer) truncating the
@@ -423,27 +455,7 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 	map->p_len = window_size;
 
 	while (read_size > 0) {
-		if (is_stream() && !eof_reached && stream_timeout > 0) {
-			fd_set rfds;
-			struct timeval tv;
-
-			FD_ZERO(&rfds);
-			FD_SET(map->fd, &rfds);
-			tv.tv_sec = stream_timeout;
-			tv.tv_usec = 0;
-
-			int res = select(map->fd + 1, &rfds, NULL, NULL, &tv);
-			if (res == 0) {
-				rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
-					stream_timeout);
-				exit_cleanup(RERR_TIMEOUT);
-			} else if (res < 0) {
-				if (errno == EINTR)
-					continue;
-				rsyserr(FERROR, errno, "select failed on stream");
-				exit_cleanup(RERR_FILEIO);
-			}
-		}
+		wait_stream_fd(map->fd);
 		int32 nread = eof_reached ? 0 : read(map->fd, map->p + read_offset, read_size);
 		if (nread <= 0) {
 			/* Retry if a blocking pipe read was interrupted by a signal. */
diff --git a/main.c b/main.c
index 33f3b5e8d..dfc72344a 100644
--- a/main.c
+++ b/main.c
@@ -1569,6 +1569,10 @@ static int start_client(int argc, char *argv[])
 		char *dummy_host;
 		int dummy_port = rsync_port;
 		int i;
+		if (stream_source_mode && argc > 1) {
+            rprintf(FERROR, "--stream-source cannot be used with multiple sources\n");
+            exit_cleanup(RERR_SYNTAX);
+        }
 		if (!argv[0][0])
 			goto invalid_empty;
 		/* For local source, extra source args must not have hostspec. */
diff --git a/match.c b/match.c
index 1ffdad3b2..c8838b4f7 100644
--- a/match.c
+++ b/match.c
@@ -28,6 +28,9 @@ extern int append_mode;
 extern struct name_num_item *xfer_sum_nni;
 extern int xfer_sum_len;
 
+extern int eof_reached;
+extern OFF_T stream_max_size;
+
 int updating_basis_file;
 char sender_file_sum[MAX_DIGEST_LEN];
 
@@ -540,6 +543,26 @@ void match_sums(int f, struct sum_struct *s, struct map_struct *buf, OFF_T len)
 		matched(f, s, buf, len, -1);
 	}
 
+	if (is_stream() && buf && !eof_reached) {
+		char extra;
+		ssize_t n;
+		wait_stream_fd(buf->fd);
+		do {
+			n = read(buf->fd, &extra, 1);
+		} while (n < 0 && errno == EINTR);
+		if (n < 0) {
+			rsyserr(FERROR, errno, "read failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		if (n > 0) {
+			rprintf(FERROR,
+				"stream source exceeded --stream-max-size=%s\n",
+				big_num(stream_max_size));
+			exit_cleanup(RERR_FILEIO);
+		}
+		eof_reached = 1;
+	}
+
 	sum_end(sender_file_sum);
 
 	/* If we had a read error, send a bad checksum.  We use all bits
diff --git a/options.c b/options.c
index b7193c62c..be0b42cb5 100644
--- a/options.c
+++ b/options.c
@@ -119,6 +119,7 @@ int mkpath_dest_arg = 0;
 int allow_inc_recurse = 1;
 int stream_source_mode = 0;
 int stream_timeout = DEFAULT_STREAM_TIMEOUT;
+static int stream_timeout_seen = 0;
 OFF_T stream_max_size = 0;
 int stream_pipe_sz = -1;
 char *stream_max_size_arg = NULL;
@@ -611,7 +612,7 @@ enum {OPT_SERVER = 1000, OPT_DAEMON, OPT_SENDER, OPT_EXCLUDE, OPT_EXCLUDE_FROM,
       OPT_USERMAP, OPT_GROUPMAP, OPT_CHOWN, OPT_BWLIMIT, OPT_STDERR,
       OPT_OLD_COMPRESS, OPT_NEW_COMPRESS, OPT_NO_COMPRESS, OPT_OLD_ARGS,
       OPT_STOP_AFTER, OPT_STOP_AT, OPT_NICE, OPT_NO_NICE,
-      OPT_REFUSED_BASE = 9000};
+      OPT_STREAM_TIMEOUT, OPT_REFUSED_BASE = 9000};
 
 static struct poptOption long_options[] = {
   /* longName, shortName, argInfo, argPtr, value, descrip, argDesc */
@@ -732,7 +733,7 @@ static struct poptOption long_options[] = {
   {"max-alloc",        0,  POPT_ARG_STRING, &max_alloc_arg, 0, 0, 0 },
   {"stream-source",    0,  POPT_ARG_VAL,    &stream_source_mode, 1, 0, 0 },
   {"stream-max-size",  0,  POPT_ARG_STRING, &stream_max_size_arg, 0, 0, 0 },
-  {"stream-timeout",   0,  POPT_ARG_INT,    &stream_timeout, 0, 0, 0 },
+  {"stream-timeout",   0,  POPT_ARG_INT,    &stream_timeout, OPT_STREAM_TIMEOUT, 0, 0 },
   {"stream-pipe-sz",   0,  POPT_ARG_STRING, &stream_pipe_sz_arg, 0, 0, 0 },
   {"sparse",          'S', POPT_ARG_VAL,    &sparse_files, 1, 0, 0 },
   {"no-sparse",        0,  POPT_ARG_VAL,    &sparse_files, 0, 0, 0 },
@@ -1878,6 +1879,10 @@ int parse_arguments(int *argc_p, const char ***argv_p)
 			break;
 		}
 
+		case OPT_STREAM_TIMEOUT:
+			stream_timeout_seen = 1;
+			break;
+
 		case OPT_APPEND:
 			if (am_server)
 				append_mode++;
@@ -2750,9 +2755,7 @@ int parse_arguments(int *argc_p, const char ***argv_p)
 		if ((stream_max_size = parse_size_arg(stream_max_size_arg, 'b', "stream-max-size", 1, -1, False)) < 0)
 			goto cleanup;
 
-		if (stream_timeout == -1)
-			stream_timeout = DEFAULT_STREAM_TIMEOUT;
-		else if (stream_timeout < 0)
+		if (stream_timeout <= 0)
 			stream_timeout = 0;
 
 		if (!stream_pipe_sz_arg)
@@ -2804,7 +2807,7 @@ int parse_arguments(int *argc_p, const char ***argv_p)
 		update_only = 0;
 		preserve_specials = 0;
 		preserve_devices = 0;
-	} else if (stream_max_size_arg || stream_pipe_sz_arg || stream_timeout != DEFAULT_STREAM_TIMEOUT) {
+	} else if (stream_max_size_arg || stream_pipe_sz_arg || stream_timeout_seen) {
 		snprintf(err_buf, sizeof err_buf,
 			"stream option(s) cannot be used without --stream-source\n");
 		goto cleanup;
diff --git a/rsync.1.md b/rsync.1.md
index 981b51ec4..d5caad89d 100644
--- a/rsync.1.md
+++ b/rsync.1.md
@@ -4222,8 +4222,9 @@ sign) if you want the local shell to expand it.
     >     rsync -L --stream-source --stream-max-size=500M \
     >         <(tar -cf - /etc) host:/backups/etc.tar
 
-    This option is restricted to a local sender (local-to-local, remote-shell
-    push, or rsync-daemon push) and cannot be used when the sender is remote.
+    This option requires exactly one source argument and is restricted to a
+    local sender (local-to-local, remote-shell push, or rsync-daemon push);
+    multiple sources and remote senders are rejected.
 
 0.  `--stream-max-size=SIZE`
 
@@ -4233,17 +4234,19 @@ sign) if you want the local shell to expand it.
 
     `SIZE` acts as an upper bound rather than an exact file size:
 
-    - If the input stream reaches EOF before `SIZE` bytes are read, the
-      transfer completes cleanly at EOF and the destination file is sized to
-      the exact number of bytes received.  On the sender side, transfer
-      statistics (`--stats`) are reconciled to the actual number of bytes read
-      at EOF.  However, because `SIZE` is transmitted in the initial file list
-      without modifying the wire protocol, the advertised ceiling still
-      controls `%l` format logging, transfer progress percentages, and
-      receiver-side (or daemon-side) total-size statistics.
-    - If the input stream produces more than `SIZE` bytes, rsync stops reading
-      at `SIZE` bytes, logs a warning indicating that `--stream-max-size` was
-      exceeded, and finalizes the destination file at that ceiling length.
+    - If the input stream reaches EOF at or before `SIZE` bytes are read, the
+      transfer completes cleanly and the destination file is sized to the exact
+      number of bytes received.  On the sender side, transfer statistics
+      (`--stats`) are reconciled to the actual number of bytes read at EOF.
+      However, because `SIZE` is transmitted in the initial file list without
+      modifying the wire protocol, the advertised ceiling still controls `%l`
+      format logging, transfer progress percentages, and receiver-side (or
+      daemon-side) total-size statistics.
+    - If the input stream produces more than `SIZE` bytes, rsync aborts the
+      transfer with a file I/O error (exit code 11) before finalizing the
+      update, leaving any existing destination file untouched (unless
+      [`--inplace`](#opt) is active, which writes directly to the destination
+      file during transfer).
 
     The `SIZE` value can be suffixed with a unit multiplier such as `K` (or
     `KiB`), `M` (or `MiB`), or `G` (or `GiB`).  See the [`--max-size`](#opt)
diff --git a/sender.c b/sender.c
index 30e777aaa..04b2b9273 100644
--- a/sender.c
+++ b/sender.c
@@ -810,24 +810,11 @@ void send_files(int f_in, int f_out)
 
 		match_sums(f_xfer, s, mbuf, st.st_size);
 		/* Update the stats for right tracking. */
-		if (is_stream() && mbuf) {
-			if (mbuf->file_size < st.st_size) {
-				OFF_T diff = st.st_size - mbuf->file_size;
-				stats.total_size -= diff;
-				stats.total_transferred_size -= diff;
-				st.st_size = mbuf->file_size;
-			} else if (!eof_reached) {
-				char extra;
-				ssize_t n;
-				do {
-					n = read(fd, &extra, 1);
-				} while (n < 0 && errno == EINTR);
-				if (n > 0) {
-					rprintf(FWARNING,
-						"Warning: stream source exceeded --stream-max-size=%s, stopping at limit\n",
-						big_num(stream_max_size));
-				}
-			}
+		if (is_stream() && mbuf && mbuf->file_size < st.st_size) {
+			OFF_T diff = st.st_size - mbuf->file_size;
+			stats.total_size -= diff;
+			stats.total_transferred_size -= diff;
+			st.st_size = mbuf->file_size;
 		}
 		if (INFO_GTE(PROGRESS, 1))
 			end_progress(st.st_size);
diff --git a/testsuite/stream-source_test.py b/testsuite/stream-source_test.py
index 6e5efde61..3daa35ca8 100644
--- a/testsuite/stream-source_test.py
+++ b/testsuite/stream-source_test.py
@@ -2,10 +2,13 @@
 
 import os
 from pathlib import Path
+import resource
+import subprocess
 import threading
+import time
 
 from rsyncfns import (
-    FROMDIR, TODIR, assert_same, rmtree, run_rsync, test_fail, test_skipped,
+    FROMDIR, TODIR, assert_same, rmtree, rsync_argv, run_rsync, test_fail, test_skipped,
 )
 
 if not hasattr(os, 'mkfifo'):
@@ -16,9 +19,93 @@
     test_skipped('rsync binary does not support --stream-source')
 
 
-def expect_fail(label: str, *args, expected_err: str = ''):
+def run_high_fd_stream_test():
+    target_fds = 2048
+    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    if soft < target_fds + 64:
+        new_soft = min(hard, target_fds + 128) if hard != resource.RLIM_INFINITY else target_fds + 128
+        if new_soft < target_fds + 64:
+            print(f"[SKIP] High-FD test requires RLIMIT_NOFILE >= {target_fds + 64} (hard={hard})", flush=True)
+            return
+        resource.setrlimit(resource.RLIMIT_NOFILE, (new_soft, hard))
+
+    label = f"High inherited descriptors (>= {target_fds}) stream transfer"
+    print(f"[RUN] {label}", flush=True)
+    if dst.exists():
+        dst.unlink()
+
+    created_fds = []
+    occupied_fds = []
+    writer_exc = []
+
+    def writer():
+        try:
+            fd = os.open(fifo, os.O_WRONLY)
+            try:
+                os.write(fd, payload_4k)
+            finally:
+                os.close(fd)
+        except Exception as exc:
+            writer_exc.append(exc)
+
+    pipe_r, pipe_w = os.pipe()
+    created_fds.extend([pipe_r, pipe_w])
+    try:
+        while True:
+            dup_fd = os.dup(pipe_r)
+            created_fds.append(dup_fd)
+            if dup_fd >= target_fds:
+                break
+
+        for fd in range(3, target_fds + 1):
+            try:
+                os.fstat(fd)
+                occupied_fds.append(fd)
+            except OSError:
+                pass
+
+        t = threading.Thread(target=writer, daemon=True)
+        t.start()
+        argv = rsync_argv(
+            '--stream-source',
+            '--stream-max-size=1M',
+            '--stream-timeout=10',
+            f'{fifo}',
+            f'{dst}',
+        )
+        proc = subprocess.run(
+            argv,
+            pass_fds=occupied_fds,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        t.join(timeout=10)
+    finally:
+        for fd in created_fds:
+            os.close(fd)
+
+    if proc.returncode != 0:
+        test_fail(
+            f"[FAIL] {label}: rsync exited with code {proc.returncode}\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    if writer_exc:
+        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
+    assert_same(dst, ref_4k, label="high-fd stream transfer")
+    print(f"[PASS] {label}", flush=True)
+
+
+def expect_fail(label: str, *args, expected_err: str = '', timeout: float = 30.0):
     print(f"[RUN] {label}", flush=True)
-    proc = run_rsync(*args, capture_output=True, check=False)
+    argv = rsync_argv(*args)
+    proc = subprocess.run(
+        argv,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
     if proc.returncode == 0:
         test_fail(
             f"[FAIL] {label}: expected non-zero exit code, got 0\n"
@@ -36,7 +123,7 @@ def expect_fail(label: str, *args, expected_err: str = ''):
     return proc
 
 
-def run_stream_transfer(label: str, fifo_path: Path, payload: bytes, dst_path: Path, *extra_args):
+def run_stream_transfer(label: str, fifo_path: Path, payload: bytes, dst_path: Path, *extra_args, expect_success: bool = True):
     print(f"[RUN] {label}", flush=True)
     writer_exc = []
 
@@ -67,15 +154,21 @@ def writer():
         check=False,
     )
     t.join(timeout=10)
-    if proc.returncode != 0:
+    if writer_exc:
+        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
+    if expect_success and proc.returncode != 0:
         test_fail(
             f"[FAIL] {label}: rsync exited with code {proc.returncode}\n"
             f"--- stdout ---\n{proc.stdout}\n"
             f"--- stderr ---\n{proc.stderr}"
         )
-    if writer_exc:
-        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
-    print(f"[PASS] {label}", flush=True)
+    if not expect_success and proc.returncode == 0:
+        test_fail(
+            f"[FAIL] {label}: expected non-zero exit code, got 0\n"
+            f"--- stdout ---\n{proc.stdout}\n"
+            f"--- stderr ---\n{proc.stderr}"
+        )
+    print(f"[PASS] {label} (exit={proc.returncode})", flush=True)
     return proc
 
 
@@ -85,8 +178,10 @@ def writer():
 TODIR.mkdir(parents=True, exist_ok=True)
 
 fifo = FROMDIR / 'stream.fifo'
+fifo2 = FROMDIR / 'stream2.fifo'
 try:
     os.mkfifo(fifo)
+    os.mkfifo(fifo2)
 except OSError as exc:
     test_skipped(f"cannot create FIFO in test filesystem: {exc}")
 
@@ -116,7 +211,7 @@ def writer():
 # ---------------------------------------------------------------------------
 # Part 2: Sub-options used without --stream-source must be refused
 # ---------------------------------------------------------------------------
-for sub_opt in ('--stream-max-size=1M', '--stream-timeout=5', '--stream-pipe-sz=1M'):
+for sub_opt in ('--stream-max-size=1M', '--stream-timeout=30', '--stream-pipe-sz=1M'):
     expect_fail(
         f"Refuse {sub_opt} without --stream-source",
         sub_opt, f'{reg_src}', f'{dst}',
@@ -124,7 +219,7 @@ def writer():
     )
 
 # ---------------------------------------------------------------------------
-# Part 3: --stream-max-size validation (missing, <= 0, invalid, truncation log)
+# Part 3: --stream-max-size validation (missing, <= 0, invalid, exact, excess)
 # ---------------------------------------------------------------------------
 expect_fail(
     "Missing --stream-max-size with --stream-source",
@@ -138,26 +233,72 @@ def writer():
         '--stream-source', f'--stream-max-size={bad_max}', f'{fifo}', f'{dst}',
     )
 
-# Stream exceeding --stream-max-size logs warning and stops at limit
+# Exact-limit stream completes cleanly without hanging
+exact_1k = b'E' * 1024
+ref_exact_1k = FROMDIR / 'ref_exact_1k.bin'
+ref_exact_1k.write_bytes(exact_1k)
 if dst.exists():
     dst.unlink()
-over_payload = b'X' * 4096
-ref_1k = FROMDIR / 'ref_1k.bin'
-ref_1k.write_bytes(over_payload[:1024])
 
+run_stream_transfer(
+    "Exact-limit stream (--stream-max-size=1024) completes cleanly",
+    fifo, exact_1k, dst,
+    '--stream-max-size=1024', '--stream-timeout=5',
+)
+assert_same(dst, ref_exact_1k, label="exact stream-max-size=1024")
+
+# Stream exceeding --stream-max-size fails before finalisation and does not commit truncated output
+sentinel_bytes = b'ORIGINAL_INTACT_DESTINATION_CONTENT'
+ref_sentinel = FROMDIR / 'ref_sentinel.bin'
+ref_sentinel.write_bytes(sentinel_bytes)
+dst.write_bytes(sentinel_bytes)
+
+over_payload = b'X' * 4096
 proc_over = run_stream_transfer(
-    "Stream exceeding --stream-max-size=1024 logs and stops at limit",
+    "Stream exceeding --stream-max-size=1024 fails without committing truncated output",
     fifo, over_payload, dst,
-    '--stream-max-size=1024',
+    '--stream-max-size=1024', '--stream-timeout=5',
+    expect_success=False,
 )
 combined_over = (proc_over.stdout + proc_over.stderr).lower()
-if 'exceeded' not in combined_over and 'limit' not in combined_over:
+if 'exceeded' not in combined_over:
     test_fail(
-        f"[FAIL] Expected warning when stream exceeds --stream-max-size, got:\n"
+        f"[FAIL] Expected error mentioning 'exceeded' when stream exceeds --stream-max-size, got:\n"
         f"--- stdout ---\n{proc_over.stdout}\n"
         f"--- stderr ---\n{proc_over.stderr}"
     )
-assert_same(dst, ref_1k, label="stream-max-size=1024 cap")
+assert_same(dst, ref_sentinel, label="existing dst preserved on oversized stream failure")
+
+# Exact-limit stream that stalls before EOF hits --stream-timeout before finalisation
+dst.write_bytes(sentinel_bytes)
+exact_stall_stop = threading.Event()
+
+
+def exact_limit_stalled_writer():
+    try:
+        fd = os.open(fifo, os.O_WRONLY)
+        try:
+            os.write(fd, exact_1k)
+            exact_stall_stop.wait(timeout=5)
+        finally:
+            os.close(fd)
+    except Exception:
+        pass
+
+
+t_exact_stall = threading.Thread(target=exact_limit_stalled_writer, daemon=True)
+t_exact_stall.start()
+try:
+    expect_fail(
+        "Exact-limit stalled stream hits --stream-timeout=1 before finalisation",
+        '--stream-source', '--stream-max-size=1024', '--stream-timeout=1',
+        f'{fifo}', f'{dst}',
+        expected_err='timed out',
+    )
+finally:
+    exact_stall_stop.set()
+    t_exact_stall.join(timeout=5)
+assert_same(dst, ref_sentinel, label="existing dst preserved on exact-limit stall timeout")
 
 # ---------------------------------------------------------------------------
 # Part 4: --stream-pipe-sz min/max bounds (64KB .. 16MB)
@@ -184,17 +325,46 @@ def writer():
     assert_same(dst, ref_4k, label=f"pipe-sz={valid_pipe_sz}")
 
 # ---------------------------------------------------------------------------
-# Part 5: --stream-timeout (<= 0 disables timeout; positive expires on stall)
+# Part 5: --stream-timeout (<= 0 disables timeout on paused stream; > 0 expires)
 # ---------------------------------------------------------------------------
 for no_timeout in ('0', '-1'):
+    label = f"Paused stream succeeds with disabled --stream-timeout={no_timeout}"
+    print(f"[RUN] {label}", flush=True)
     if dst.exists():
         dst.unlink()
-    run_stream_transfer(
-        f"Disabled timeout with --stream-timeout={no_timeout}",
-        fifo, payload_4k, dst,
-        '--stream-max-size=1M', f'--stream-timeout={no_timeout}',
+    writer_exc = []
+
+    def delayed_writer():
+        try:
+            fd = os.open(fifo, os.O_WRONLY)
+            try:
+                os.write(fd, payload_4k[:2048])
+                time.sleep(1.2)
+                os.write(fd, payload_4k[2048:])
+            finally:
+                os.close(fd)
+        except Exception as exc:
+            writer_exc.append(exc)
+
+    t_delayed = threading.Thread(target=delayed_writer, daemon=True)
+    t_delayed.start()
+    proc_nt = run_rsync(
+        '--stream-source', '--stream-max-size=1M', f'--stream-timeout={no_timeout}',
+        f'{fifo}', f'{dst}',
+        capture_output=True,
+        check=False,
     )
+    t_delayed.join(timeout=10)
+    if writer_exc:
+        test_fail(f"[FAIL] {label}: FIFO writer failed: {writer_exc[0]}")
+    if proc_nt.returncode != 0:
+        test_fail(
+            f"[FAIL] {label}: rsync exited with code {proc_nt.returncode}\n"
+            f"--- stdout ---\n{proc_nt.stdout}\n"
+            f"--- stderr ---\n{proc_nt.stderr}"
+        )
     assert_same(dst, ref_4k, label=f"timeout={no_timeout}")
+    print(f"[PASS] {label}", flush=True)
 
 stall_stop = threading.Event()
 
@@ -266,9 +436,19 @@ def stalled_writer():
     f'{fifo}', f'{dst}',
 )
 
+# Reject multiple sources before opening either FIFO
+expect_fail(
+    "Refuse multiple FIFO sources (fifo, fifo2) before opening either FIFO",
+    '--stream-source', '--stream-max-size=1M', f'{fifo}', f'{fifo2}', f'{TODIR}/',
+    expected_err='multiple',
+    timeout=5.0,
+)
+
 expect_fail(
-    "Refuse multiple sources with --stream-source",
+    "Refuse multiple mixed sources (fifo, reg_src) with --stream-source",
     '--stream-source', '--stream-max-size=1M', f'{fifo}', f'{reg_src}', f'{TODIR}/',
+    expected_err='multiple',
+    timeout=5.0,
 )
 
 expect_fail(
@@ -300,3 +480,5 @@ def stalled_writer():
     '--stream-source', '--stream-max-size=1M', 'rsync://localhost/module/stream.fifo', f'{dst}',
     expected_err='--stream-source',
 )
+
+run_high_fd_stream_test()

From 491669e1c1cd0a0fbedcb8c4d0a6215bc666fb5b Mon Sep 17 00:00:00 2001
From: seks99x <omarelsayed161@gmail.com>
Date: Tue, 10 Nov 2026 23:31:31 +0300
Subject: [PATCH 4/4] Use select() on macOS in wait_stream_fd() to avoid XNU
 FIFO poll() hang

- Fall back to select() under __APPLE__ in wait_stream_fd() because XNU's
  poll() fails to wake on FIFO EOF when the pipe buffer is empty, causing
  0-byte and exact-ceiling --stream-source transfers to time out on macOS.
- Enable _DARWIN_UNLIMITED_SELECT and expand FD_SETSIZE to 65536 before
  including headers in fileio.c alongside an explicit bounds check so
  high inherited file descriptors remain supported.
---
 fileio.c | 36 ++++++++++++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/fileio.c b/fileio.c
index 0a9a1605a..a56d600b2 100644
--- a/fileio.c
+++ b/fileio.c
@@ -18,6 +18,10 @@
  * You should have received a copy of the GNU General Public License along
  * with this program; if not, visit the http://fsf.org website.
  */
+#ifdef __APPLE__
+#define _DARWIN_UNLIMITED_SELECT 1
+#define FD_SETSIZE 65536
+#endif
 
 #include "rsync.h"
 #include "inums.h"
@@ -322,6 +326,37 @@ void wait_stream_fd(int fd)
 	if (!is_stream() || eof_reached || stream_timeout <= 0)
 		return;
 
+#ifdef __APPLE__
+	if (fd < 0 || fd >= FD_SETSIZE) {
+		rprintf(FERROR, "stream fd %d exceeds FD_SETSIZE (%d)\n",
+			fd, (int)FD_SETSIZE);
+		exit_cleanup(RERR_FILEIO);
+	}
+
+	for (;;) {
+		fd_set rfds;
+		struct timeval tv;
+
+		FD_ZERO(&rfds);
+		FD_SET(fd, &rfds);
+		tv.tv_sec = stream_timeout;
+		tv.tv_usec = 0;
+
+		int res = select(fd + 1, &rfds, NULL, NULL, &tv);
+		if (res == 0) {
+			rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
+				stream_timeout);
+			exit_cleanup(RERR_TIMEOUT);
+		}
+		if (res < 0) {
+			if (errno == EINTR)
+				continue;
+			rsyserr(FERROR, errno, "select failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		break;
+	}
+#else
 	int timeout_ms = stream_timeout > INT_MAX / 1000
 		       ? INT_MAX
 		       : stream_timeout * 1000;
@@ -346,6 +381,7 @@ void wait_stream_fd(int fd)
 		}
 		break;
 	}
+#endif
 }
 
 /* This provides functionality somewhat similar to mmap() but using read().