From c4b36583f577de9be1131000e13717e79b08e716 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Mon, 7 Sep 2026 15:57:32 +0300
Subject: [PATCH 1/4] Serialise buffered stream access across coroutines

php_stream_fill_read_buffer() derives a target address and a length from
readbuf, readpos and writepos, hands them to ops->read, and credits the
result back to writepos. Upstream relies on that sequence being
indivisible. The async wrapper parks the coroutine inside ops->read
(plain_wrapper.c, ZEND_ASYNC_SUSPEND), so a second coroutine reading the
same handle derived the same address and the same length from a state the
first one had not finished updating: both reads filled one buffer offset,
and both credits landed on one writepos.

Two coroutines reading one handle therefore lost and duplicated data --
1024 distinct records out of 2000 in the regression test -- and writepos
could pass readbuflen, after which readbuflen - writepos wrapped around
zero. ops->read then got a length near 2^64, which the async layer clamps
to INT_MAX and hands to uv_fs_read against an 8 KB allocation: a heap
overflow on Linux, and on Windows the "Read of 18446744073709543424 bytes
failed" reported in true-async/php-async#280.

A per-stream lock now covers the whole derive-read-credit sequence.
php_stream_read(), php_stream_fill_read_buffer(), php_stream_get_line(),
php_stream_get_record(), php_stream_write() and php_stream_seek() take it;
it is recursive for its owner, because the buffered API re-enters itself
(a seek emulated through reads, a userspace wrapper reading its own
stream).

The lock is an event whose callback vector is the queue of waiters, in
arrival order. Releasing it invokes the head callback directly, which
takes ownership inside that call, so a coroutine arriving between the
release and the wakeup finds the lock taken and queues behind instead of
overtaking. The event keeps an order-preserving del_callback of its own:
the core one fills the hole with the last element, which would run the
queue as a stack. php_stream carries one pointer; owner, recursion depth
and the queue live in the event.

The event also answers whether the stream survived the wait. A waiter
holds a reference on it until its own wakeup, and closing a stream marks
the event closed before freeing anything, so a waiter that wakes to a
closed event returns without touching the stream. This works for every
wrapper, including the socket and userspace ones that park outside
ops->read.

fill_read_buffer also refuses to call ops->read when writepos is past the
end of the buffer, so a residual accounting error degrades to a failed
read rather than to a read into unallocated memory.

With async off, outside a coroutine and in scheduler context every lock
call is a no-op, so synchronous behaviour is unchanged.
---
 main/php_streams.h     |   4 +
 main/streams/streams.c | 288 ++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 286 insertions(+), 6 deletions(-)

diff --git a/main/php_streams.h b/main/php_streams.h
index cdc7d1b67eb7..7ef996f8624d 100644
--- a/main/php_streams.h
+++ b/main/php_streams.h
@@ -244,6 +244,10 @@ struct _php_stream  {
 	/* how much data to read when filling buffer */
 	size_t chunk_size;
 
+	/* Async: serialises access to the buffer fields above. Owner, depth and
+	 * waiters live in the event; see stream_buffer_lock() in streams.c. */
+	struct _zend_async_event_s *buffer_lock;
+
 #if ZEND_DEBUG
 	const char *open_filename;
 	uint32_t open_lineno;
diff --git a/main/streams/streams.c b/main/streams/streams.c
index 6fcdc7f8a284..fefdf2fba75f 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -427,6 +427,15 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 			stream->readbuf = NULL;
 		}
 
+		/* Nothing would notify the waiters again; closing tells each of them,
+		 * once it runs, that the stream is gone. */
+		if (stream->buffer_lock != NULL) {
+			zend_async_event_t *lock = stream->buffer_lock;
+			stream->buffer_lock = NULL;
+			ZEND_ASYNC_CALLBACKS_NOTIFY_AND_CLOSE(lock, NULL, NULL);
+			ZEND_ASYNC_EVENT_RELEASE(lock);
+		}
+
 		if (stream->is_persistent && (close_options & PHP_STREAM_FREE_PERSISTENT)) {
 			/* we don't work with *stream but need its value for comparison */
 			zend_hash_apply_with_argument(&EG(persistent_list), _php_stream_free_persistent, stream);
@@ -455,7 +464,193 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 
 /* {{{ generic stream operations */
 
-PHPAPI zend_result php_stream_fill_read_buffer(php_stream *stream, size_t size)
+/* {{{ Buffer lock */
+
+typedef struct {
+	zend_async_event_t base;
+	php_stream *stream;
+	zend_coroutine_t *owner;
+	/* The buffered API re-enters itself: a seek emulated through reads, a
+	 * userspace wrapper reading its own stream. */
+	uint32_t depth;
+} php_stream_buffer_lock_t;
+
+
+static void stream_buffer_unlock(php_stream *stream);
+
+static bool stream_buffer_event_add_callback(zend_async_event_t *event, zend_async_event_callback_t *callback)
+{
+	return zend_async_callbacks_push(event, callback);
+}
+
+/* zend_async_callbacks_remove() fills the hole with the last element, which
+ * would run this queue as a stack. */
+static bool stream_buffer_event_del_callback(zend_async_event_t *event, zend_async_event_callback_t *callback)
+{
+	zend_async_callbacks_vector_t *vector = &event->callbacks;
+
+	for (uint32_t i = 0; i < vector->length; ++i) {
+		if (vector->data[i] != callback) {
+			continue;
+		}
+
+		if (vector->current_iterator != NULL && *vector->current_iterator > i) {
+			(*vector->current_iterator)--;
+		}
+
+		vector->length--;
+
+		if (i < vector->length) {
+			memmove(&vector->data[i], &vector->data[i + 1],
+					(vector->length - i) * sizeof(zend_async_event_callback_t *));
+		}
+
+		callback->dispose(callback, event);
+
+		return true;
+	}
+
+	return false;
+}
+
+static bool stream_buffer_event_start(zend_async_event_t *event)
+{
+	return true;
+}
+
+static bool stream_buffer_event_stop(zend_async_event_t *event)
+{
+	return true;
+}
+
+static bool stream_buffer_event_dispose(zend_async_event_t *event)
+{
+	zend_async_callbacks_free(event);
+	efree(event);
+
+	return true;
+}
+
+/* Ownership passes inside the wakeup, so a coroutine arriving meanwhile
+ * queues behind instead of overtaking. */
+static void stream_buffer_lock_grant(zend_async_event_t *event, zend_async_event_callback_t *callback,
+		void *result, zend_object *exception)
+{
+	if (!ZEND_ASYNC_EVENT_IS_CLOSED(event)) {
+		php_stream_buffer_lock_t *lock = (php_stream_buffer_lock_t *) event;
+
+		lock->owner = ((zend_coroutine_event_callback_t *) callback)->coroutine;
+		lock->depth = 1;
+	}
+
+	zend_async_waker_callback_resolve(event, callback, result, exception);
+}
+
+static php_stream_buffer_lock_t *stream_buffer_lock_create(php_stream *stream)
+{
+	php_stream_buffer_lock_t *lock = ecalloc(1, sizeof(php_stream_buffer_lock_t));
+
+	lock->stream = stream;
+	lock->base.ref_count = 1;
+	lock->base.add_callback = stream_buffer_event_add_callback;
+	lock->base.del_callback = stream_buffer_event_del_callback;
+	lock->base.start = stream_buffer_event_start;
+	lock->base.stop = stream_buffer_event_stop;
+	lock->base.dispose = stream_buffer_event_dispose;
+
+	/* A queued waiter is not IO: it must not count as live work for deadlock
+	 * detection. */
+	ZEND_ASYNC_EVENT_SET_HIDDEN(&lock->base);
+
+	stream->buffer_lock = &lock->base;
+
+	return lock;
+}
+
+/* false: the lock is NOT held, and the caller must fail its operation. */
+static bool stream_buffer_lock(php_stream *stream)
+{
+	if (ZEND_ASYNC_IS_OFF || ZEND_ASYNC_IS_SCHEDULER_CONTEXT) {
+		return true;
+	}
+
+	zend_coroutine_t *coroutine = ZEND_ASYNC_CURRENT_COROUTINE;
+
+	if (coroutine == NULL) {
+		return true;
+	}
+
+	php_stream_buffer_lock_t *lock = (php_stream_buffer_lock_t *) stream->buffer_lock;
+
+	if (lock == NULL) {
+		lock = stream_buffer_lock_create(stream);
+	} else if (lock->owner == coroutine) {
+		lock->depth++;
+		return true;
+	}
+
+	while (lock->owner != NULL) {
+		ZEND_ASYNC_WAKER_NEW(coroutine);
+		zend_async_resume_when(coroutine, &lock->base, false, stream_buffer_lock_grant, NULL);
+
+		const bool resumed = ZEND_ASYNC_SUSPEND();
+		/* Our subscription holds a reference, so the event is readable even
+		 * when the stream is already gone. */
+		const bool closed = ZEND_ASYNC_EVENT_IS_CLOSED(&lock->base);
+
+		zend_async_waker_clean(coroutine);
+
+		if (closed) {
+			return false;
+		}
+
+		if (!resumed || UNEXPECTED(EG(exception) != NULL)) {
+			/* The lock may already be ours, and nobody else would release it. */
+			if (lock->owner == coroutine) {
+				stream_buffer_unlock(stream);
+			}
+
+			return false;
+		}
+
+		if (lock->owner == coroutine) {
+			return true;
+		}
+	}
+
+	lock->owner = coroutine;
+	lock->depth = 1;
+
+	return true;
+}
+
+static void stream_buffer_unlock(php_stream *stream)
+{
+	php_stream_buffer_lock_t *lock = (php_stream_buffer_lock_t *) stream->buffer_lock;
+
+	if (lock == NULL || lock->owner != ZEND_ASYNC_CURRENT_COROUTINE) {
+		return;
+	}
+
+	if (--lock->depth > 0) {
+		return;
+	}
+
+	lock->owner = NULL;
+
+	/* Wake the head only: notifying the event would wake every waiter, and all
+	 * but one would go back to sleep. The vector does not move here — a waiter
+	 * leaves it from its own coroutine. */
+	if (lock->base.callbacks.length > 0) {
+		zend_async_event_callback_t *callback = lock->base.callbacks.data[0];
+
+		callback->callback(&lock->base, callback, NULL, NULL);
+	}
+}
+
+/* }}} */
+
+static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t size)
 {
 	/* allocate/fill the buffer */
 
@@ -605,6 +800,13 @@ PHPAPI zend_result php_stream_fill_read_buffer(php_stream *stream, size_t size)
 						stream->is_persistent);
 			}
 
+			/* The length below is unsigned: a writepos past the end of the
+			 * buffer wraps it around zero. */
+			if (UNEXPECTED((size_t) stream->writepos > stream->readbuflen)) {
+				stream->fatal_error = 1;
+				return FAILURE;
+			}
+
 			justread = stream->ops->read(stream, (char*)stream->readbuf + stream->writepos,
 					stream->readbuflen - stream->writepos
 					);
@@ -629,7 +831,20 @@ PHPAPI zend_result php_stream_fill_read_buffer(php_stream *stream, size_t size)
 	return retval;
 }
 
-PHPAPI ssize_t php_stream_read(php_stream *stream, char *buf, size_t size)
+PHPAPI zend_result php_stream_fill_read_buffer(php_stream *stream, size_t size)
+{
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		return FAILURE;
+	}
+
+	const zend_result result = stream_fill_read_buffer_locked(stream, size);
+
+	stream_buffer_unlock(stream);
+
+	return result;
+}
+
+static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size)
 {
 	ssize_t toread = 0, didread = 0;
 
@@ -670,7 +885,7 @@ PHPAPI ssize_t php_stream_read(php_stream *stream, char *buf, size_t size)
 				break;
 			}
 		} else {
-			if (php_stream_fill_read_buffer(stream, size) != SUCCESS) {
+			if (stream_fill_read_buffer_locked(stream, size) != SUCCESS) {
 				if (didread == 0) {
 					return -1;
 				}
@@ -713,6 +928,19 @@ PHPAPI ssize_t php_stream_read(php_stream *stream, char *buf, size_t size)
 	return didread;
 }
 
+PHPAPI ssize_t php_stream_read(php_stream *stream, char *buf, size_t size)
+{
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		return -1;
+	}
+
+	const ssize_t didread = stream_read_locked(stream, buf, size);
+
+	stream_buffer_unlock(stream);
+
+	return didread;
+}
+
 /* Like php_stream_read(), but reading into a zend_string buffer. This has some similarity
  * to the copy_to_mem() operation, but only performs a single direct read. */
 PHPAPI zend_string *php_stream_read_to_str(php_stream *stream, size_t len)
@@ -844,7 +1072,7 @@ PHPAPI const char *php_stream_locate_eol(php_stream *stream, zend_string *buf)
 /* If buf == NULL, the buffer will be allocated automatically and will be of an
  * appropriate length to hold the line, regardless of the line length, memory
  * permitting */
-PHPAPI char *php_stream_get_line(php_stream *stream, char *buf, size_t maxlen,
+static char *stream_get_line_locked(php_stream *stream, char *buf, size_t maxlen,
 		size_t *returned_len)
 {
 	size_t avail = 0;
@@ -963,6 +1191,22 @@ PHPAPI char *php_stream_get_line(php_stream *stream, char *buf, size_t maxlen,
 	return bufstart;
 }
 
+PHPAPI char *php_stream_get_line(php_stream *stream, char *buf, size_t maxlen, size_t *returned_len)
+{
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		if (returned_len != NULL) {
+			*returned_len = 0;
+		}
+		return NULL;
+	}
+
+	char *result = stream_get_line_locked(stream, buf, maxlen, returned_len);
+
+	stream_buffer_unlock(stream);
+
+	return result;
+}
+
 #define STREAM_BUFFERED_AMOUNT(stream) \
 	((size_t)(((stream)->writepos) - (stream)->readpos))
 
@@ -991,7 +1235,7 @@ static const char *_php_stream_search_delim(
 	}
 }
 
-PHPAPI zend_string *php_stream_get_record(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len)
+static zend_string *stream_get_record_locked(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len)
 {
 	zend_string	*ret_buf;				/* returned buffer */
 	const char *found_delim = NULL;
@@ -1081,6 +1325,19 @@ PHPAPI zend_string *php_stream_get_record(php_stream *stream, size_t maxlen, con
 	return ret_buf;
 }
 
+PHPAPI zend_string *php_stream_get_record(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len)
+{
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		return NULL;
+	}
+
+	zend_string *result = stream_get_record_locked(stream, maxlen, delim, delim_len);
+
+	stream_buffer_unlock(stream);
+
+	return result;
+}
+
 /* Writes a buffer directly to a stream, using multiple of the chunk size */
 static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size_t count)
 {
@@ -1244,12 +1501,18 @@ PHPAPI ssize_t php_stream_write(php_stream *stream, const char *buf, size_t coun
 		return (ssize_t) -1;
 	}
 
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		return (ssize_t) -1;
+	}
+
 	if (stream->writefilters.head) {
 		bytes = php_stream_write_filtered(stream, buf, count, PSFS_FLAG_NORMAL);
 	} else {
 		bytes = php_stream_write_buffer(stream, buf, count);
 	}
 
+	stream_buffer_unlock(stream);
+
 	if (bytes) {
 		stream->flags |= PHP_STREAM_FLAG_WAS_WRITTEN;
 	}
@@ -1339,7 +1602,7 @@ static bool php_stream_has_notifier(php_stream *stream)
 	return context && context->notifier;
 }
 
-PHPAPI int php_stream_seek(php_stream *stream, zend_off_t offset, int whence)
+static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence)
 {
 	if (stream->fclose_stdiocast == PHP_STREAM_FCLOSE_FOPENCOOKIE) {
 		/* flush can call seek internally so we need to prevent an infinite loop */
@@ -1446,6 +1709,19 @@ PHPAPI int php_stream_seek(php_stream *stream, zend_off_t offset, int whence)
 	return -1;
 }
 
+PHPAPI int php_stream_seek(php_stream *stream, zend_off_t offset, int whence)
+{
+	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+		return -1;
+	}
+
+	const int result = stream_seek_locked(stream, offset, whence);
+
+	stream_buffer_unlock(stream);
+
+	return result;
+}
+
 PHPAPI int php_stream_set_option(php_stream *stream, int option, int value, void *ptrparam)
 {
 	int ret = PHP_STREAM_OPTION_RETURN_NOTIMPL;

From 021317dc9784d726ae6a2ba44f8e113c6def3532 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Mon, 7 Sep 2026 23:42:06 +0300
Subject: [PATCH 2/4] Hold the buffer lock through a close that frees the
 stream
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A coroutine parked inside ops->read or ops->write can find the stream gone
when it returns: another coroutine closed the handle, php_stdiop_close
deferred the free because the parked reader still held a pin, and the unpin
on the way out of ops->read performed it. The lock helpers reached the lock
through stream->buffer_lock, so unlocking read a freed php_stream.

stream_buffer_lock() now hands the caller the lock itself and takes a
reference on it, and stream_buffer_unlock() takes that pointer. The lock
outlives the stream, and its closed flag is what the buffered paths test
after every call that can park, instead of touching the stream again.

php_stream_write_buffer() needed the same: on a close during a parked write
its "out:" label read stream->eof and PHP_STREAM_CONTEXT(stream) from freed
memory. Verified with a poisoned free — SIGSEGV in php_stream_write_buffer
at the notify_completed line, from fwrite() in a coroutine while another
one closed the handle; clean afterwards.

Regression coverage: tests/io/086-close_during_io.phpt in
true-async/php-async, which crashes on the previous code under a poisoned
free and under a sanitizer build.
---
 main/streams/streams.c | 132 +++++++++++++++++++++++++++--------------
 1 file changed, 87 insertions(+), 45 deletions(-)

diff --git a/main/streams/streams.c b/main/streams/streams.c
index fefdf2fba75f..400d0f6ec9d2 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -468,7 +468,6 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 
 typedef struct {
 	zend_async_event_t base;
-	php_stream *stream;
 	zend_coroutine_t *owner;
 	/* The buffered API re-enters itself: a seek emulated through reads, a
 	 * userspace wrapper reading its own stream. */
@@ -476,7 +475,12 @@ typedef struct {
 } php_stream_buffer_lock_t;
 
 
-static void stream_buffer_unlock(php_stream *stream);
+static void stream_buffer_unlock(php_stream_buffer_lock_t *lock);
+
+static bool stream_buffer_lock_closed(php_stream_buffer_lock_t *lock)
+{
+	return lock != NULL && ZEND_ASYNC_EVENT_IS_CLOSED(&lock->base);
+}
 
 static bool stream_buffer_event_add_callback(zend_async_event_t *event, zend_async_event_callback_t *callback)
 {
@@ -550,7 +554,6 @@ static php_stream_buffer_lock_t *stream_buffer_lock_create(php_stream *stream)
 {
 	php_stream_buffer_lock_t *lock = ecalloc(1, sizeof(php_stream_buffer_lock_t));
 
-	lock->stream = stream;
 	lock->base.ref_count = 1;
 	lock->base.add_callback = stream_buffer_event_add_callback;
 	lock->base.del_callback = stream_buffer_event_del_callback;
@@ -568,8 +571,9 @@ static php_stream_buffer_lock_t *stream_buffer_lock_create(php_stream *stream)
 }
 
 /* false: the lock is NOT held, and the caller must fail its operation. */
-static bool stream_buffer_lock(php_stream *stream)
+static bool stream_buffer_lock(php_stream *stream, php_stream_buffer_lock_t **held)
 {
+	*held = NULL;
 	if (ZEND_ASYNC_IS_OFF || ZEND_ASYNC_IS_SCHEDULER_CONTEXT) {
 		return true;
 	}
@@ -584,8 +588,14 @@ static bool stream_buffer_lock(php_stream *stream)
 
 	if (lock == NULL) {
 		lock = stream_buffer_lock_create(stream);
-	} else if (lock->owner == coroutine) {
+	}
+
+	/* Keep the event alive independently of the stream, including for the
+	 * owner: ops->read/write may return after fclose() freed the stream. */
+	ZEND_ASYNC_EVENT_ADD_REF(&lock->base);
+	if (lock->owner == coroutine) {
 		lock->depth++;
+		*held = lock;
 		return true;
 	}
 
@@ -601,39 +611,39 @@ static bool stream_buffer_lock(php_stream *stream)
 		zend_async_waker_clean(coroutine);
 
 		if (closed) {
+			ZEND_ASYNC_EVENT_RELEASE(&lock->base);
 			return false;
 		}
 
 		if (!resumed || UNEXPECTED(EG(exception) != NULL)) {
 			/* The lock may already be ours, and nobody else would release it. */
-			if (lock->owner == coroutine) {
-				stream_buffer_unlock(stream);
-			}
+			stream_buffer_unlock(lock);
 
 			return false;
 		}
 
 		if (lock->owner == coroutine) {
+			*held = lock;
 			return true;
 		}
 	}
 
 	lock->owner = coroutine;
 	lock->depth = 1;
+	*held = lock;
 
 	return true;
 }
 
-static void stream_buffer_unlock(php_stream *stream)
+static void stream_buffer_unlock(php_stream_buffer_lock_t *lock)
 {
-	php_stream_buffer_lock_t *lock = (php_stream_buffer_lock_t *) stream->buffer_lock;
-
-	if (lock == NULL || lock->owner != ZEND_ASYNC_CURRENT_COROUTINE) {
+	if (lock == NULL) {
 		return;
 	}
 
-	if (--lock->depth > 0) {
-		return;
+	if (stream_buffer_lock_closed(lock) || lock->owner != ZEND_ASYNC_CURRENT_COROUTINE
+			|| --lock->depth > 0) {
+		goto release;
 	}
 
 	lock->owner = NULL;
@@ -646,11 +656,14 @@ static void stream_buffer_unlock(php_stream *stream)
 
 		callback->callback(&lock->base, callback, NULL, NULL);
 	}
+
+release:
+	ZEND_ASYNC_EVENT_RELEASE(&lock->base);
 }
 
 /* }}} */
 
-static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t size)
+static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t size, php_stream_buffer_lock_t *lock)
 {
 	/* allocate/fill the buffer */
 
@@ -675,6 +688,10 @@ static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t siz
 
 			/* read a chunk into a bucket */
 			justread = stream->ops->read(stream, chunk_buf, stream->chunk_size);
+			if (stream_buffer_lock_closed(lock)) {
+				efree(chunk_buf);
+				return FAILURE;
+			}
 			if (justread < 0 && stream->writepos == stream->readpos) {
 				efree(chunk_buf);
 				retval = FAILURE;
@@ -833,18 +850,19 @@ static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t siz
 
 PHPAPI zend_result php_stream_fill_read_buffer(php_stream *stream, size_t size)
 {
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		return FAILURE;
 	}
 
-	const zend_result result = stream_fill_read_buffer_locked(stream, size);
+	const zend_result result = stream_fill_read_buffer_locked(stream, size, lock);
 
-	stream_buffer_unlock(stream);
+	stream_buffer_unlock(lock);
 
 	return result;
 }
 
-static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size)
+static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size, php_stream_buffer_lock_t *lock)
 {
 	ssize_t toread = 0, didread = 0;
 
@@ -876,6 +894,9 @@ static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size)
 
 		if (!stream->readfilters.head && ((stream->flags & PHP_STREAM_FLAG_NO_BUFFER) || stream->chunk_size == 1)) {
 			toread = stream->ops->read(stream, buf, size);
+			if (stream_buffer_lock_closed(lock)) {
+				return didread ? didread : -1;
+			}
 			if (toread < 0) {
 				/* Report an error if the read failed and we did not read any data
 				 * before that. Otherwise return the data we did read. */
@@ -885,7 +906,10 @@ static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size)
 				break;
 			}
 		} else {
-			if (stream_fill_read_buffer_locked(stream, size) != SUCCESS) {
+			if (stream_fill_read_buffer_locked(stream, size, lock) != SUCCESS) {
+				if (stream_buffer_lock_closed(lock)) {
+					return didread ? didread : -1;
+				}
 				if (didread == 0) {
 					return -1;
 				}
@@ -930,13 +954,14 @@ static ssize_t stream_read_locked(php_stream *stream, char *buf, size_t size)
 
 PHPAPI ssize_t php_stream_read(php_stream *stream, char *buf, size_t size)
 {
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		return -1;
 	}
 
-	const ssize_t didread = stream_read_locked(stream, buf, size);
+	const ssize_t didread = stream_read_locked(stream, buf, size, lock);
 
-	stream_buffer_unlock(stream);
+	stream_buffer_unlock(lock);
 
 	return didread;
 }
@@ -1073,7 +1098,7 @@ PHPAPI const char *php_stream_locate_eol(php_stream *stream, zend_string *buf)
  * appropriate length to hold the line, regardless of the line length, memory
  * permitting */
 static char *stream_get_line_locked(php_stream *stream, char *buf, size_t maxlen,
-		size_t *returned_len)
+		size_t *returned_len, php_stream_buffer_lock_t *lock)
 {
 	size_t avail = 0;
 	size_t current_buf_size = 0;
@@ -1163,7 +1188,8 @@ static char *stream_get_line_locked(php_stream *stream, char *buf, size_t maxlen
 				}
 			}
 
-			if (php_stream_fill_read_buffer(stream, toread) == FAILURE && stream->fatal_error) {
+			zend_result result = php_stream_fill_read_buffer(stream, toread);
+			if (stream_buffer_lock_closed(lock) || (result == FAILURE && stream->fatal_error)) {
 				if (grow_mode) {
 					efree(bufstart);
 				}
@@ -1193,16 +1219,17 @@ static char *stream_get_line_locked(php_stream *stream, char *buf, size_t maxlen
 
 PHPAPI char *php_stream_get_line(php_stream *stream, char *buf, size_t maxlen, size_t *returned_len)
 {
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		if (returned_len != NULL) {
 			*returned_len = 0;
 		}
 		return NULL;
 	}
 
-	char *result = stream_get_line_locked(stream, buf, maxlen, returned_len);
+	char *result = stream_get_line_locked(stream, buf, maxlen, returned_len, lock);
 
-	stream_buffer_unlock(stream);
+	stream_buffer_unlock(lock);
 
 	return result;
 }
@@ -1235,7 +1262,8 @@ static const char *_php_stream_search_delim(
 	}
 }
 
-static zend_string *stream_get_record_locked(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len)
+static zend_string *stream_get_record_locked(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len,
+		php_stream_buffer_lock_t *lock)
 {
 	zend_string	*ret_buf;				/* returned buffer */
 	const char *found_delim = NULL;
@@ -1260,7 +1288,8 @@ static zend_string *stream_get_record_locked(php_stream *stream, size_t maxlen,
 
 		to_read_now = MIN(maxlen - buffered_len, stream->chunk_size);
 
-		if (php_stream_fill_read_buffer(stream, buffered_len + to_read_now) == FAILURE && stream->fatal_error) {
+		zend_result result = php_stream_fill_read_buffer(stream, buffered_len + to_read_now);
+		if (stream_buffer_lock_closed(lock) || (result == FAILURE && stream->fatal_error)) {
 			return NULL;
 		}
 
@@ -1327,19 +1356,21 @@ static zend_string *stream_get_record_locked(php_stream *stream, size_t maxlen,
 
 PHPAPI zend_string *php_stream_get_record(php_stream *stream, size_t maxlen, const char *delim, size_t delim_len)
 {
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		return NULL;
 	}
 
-	zend_string *result = stream_get_record_locked(stream, maxlen, delim, delim_len);
+	zend_string *result = stream_get_record_locked(stream, maxlen, delim, delim_len, lock);
 
-	stream_buffer_unlock(stream);
+	stream_buffer_unlock(lock);
 
 	return result;
 }
 
 /* Writes a buffer directly to a stream, using multiple of the chunk size */
-static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size_t count)
+static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size_t count,
+		php_stream_buffer_lock_t *lock)
 {
 	ssize_t didwrite = 0;
 	ssize_t retval;
@@ -1364,6 +1395,13 @@ static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size
 
 	while (count > 0) {
 		ssize_t justwrote = stream->ops->write(stream, buf, MIN(chunk_size, count));
+
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			/* Another coroutine closed the stream while this write was parked:
+			 * ops->write freed it, so neither eof nor position exist any more. */
+			return didwrite;
+		}
+
 		if (justwrote <= 0) {
 			/* If we already successfully wrote some bytes and a write error occurred
 			 * later, report the successfully written bytes. */
@@ -1395,7 +1433,8 @@ static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size
  * This may trigger a real write to the stream.
  * Returns the number of bytes consumed from buf by the first filter in the chain.
  * */
-static ssize_t php_stream_write_filtered(php_stream *stream, const char *buf, size_t count, int flags)
+static ssize_t php_stream_write_filtered(php_stream *stream, const char *buf, size_t count, int flags,
+		php_stream_buffer_lock_t *lock)
 {
 	size_t consumed = 0;
 	php_stream_bucket *bucket;
@@ -1432,7 +1471,9 @@ static ssize_t php_stream_write_filtered(php_stream *stream, const char *buf, si
 			 * underlying stream */
 			while (brig_inp->head) {
 				bucket = brig_inp->head;
-				if (php_stream_write_buffer(stream, bucket->buf, bucket->buflen) < 0) {
+				if (stream_buffer_lock_closed(lock)) {
+					consumed = (ssize_t) -1;
+				} else if (php_stream_write_buffer(stream, bucket->buf, bucket->buflen, lock) < 0) {
 					consumed = (ssize_t) -1;
 				}
 
@@ -1471,7 +1512,7 @@ static int php_stream_flush_ex(php_stream *stream, bool closing)
 	int ret = 0;
 
 	if (stream->writefilters.head && stream->ops->write) {
-		php_stream_write_filtered(stream, NULL, 0, closing ? PSFS_FLAG_FLUSH_CLOSE : PSFS_FLAG_FLUSH_INC );
+		php_stream_write_filtered(stream, NULL, 0, closing ? PSFS_FLAG_FLUSH_CLOSE : PSFS_FLAG_FLUSH_INC, NULL);
 	}
 
 	stream->flags &= ~PHP_STREAM_FLAG_WAS_WRITTEN;
@@ -1501,21 +1542,21 @@ PHPAPI ssize_t php_stream_write(php_stream *stream, const char *buf, size_t coun
 		return (ssize_t) -1;
 	}
 
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		return (ssize_t) -1;
 	}
 
 	if (stream->writefilters.head) {
-		bytes = php_stream_write_filtered(stream, buf, count, PSFS_FLAG_NORMAL);
+		bytes = php_stream_write_filtered(stream, buf, count, PSFS_FLAG_NORMAL, lock);
 	} else {
-		bytes = php_stream_write_buffer(stream, buf, count);
+		bytes = php_stream_write_buffer(stream, buf, count, lock);
 	}
 
-	stream_buffer_unlock(stream);
-
-	if (bytes) {
+	if (bytes && !stream_buffer_lock_closed(lock)) {
 		stream->flags |= PHP_STREAM_FLAG_WAS_WRITTEN;
 	}
+	stream_buffer_unlock(lock);
 
 	return bytes;
 }
@@ -1711,13 +1752,14 @@ static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence)
 
 PHPAPI int php_stream_seek(php_stream *stream, zend_off_t offset, int whence)
 {
-	if (UNEXPECTED(!stream_buffer_lock(stream))) {
+	php_stream_buffer_lock_t *lock;
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
 		return -1;
 	}
 
 	const int result = stream_seek_locked(stream, offset, whence);
 
-	stream_buffer_unlock(stream);
+	stream_buffer_unlock(lock);
 
 	return result;
 }

From eb091a2fe686494732bd57e9e91265f386f55d4f Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Mon, 7 Sep 2026 23:58:16 +0300
Subject: [PATCH 3/4] Close three more paths that outlive the stream

php_stream_flush() reaches the write-filter chain, and that chain writes
through ops->write, which parks. It held no lock, so a close arriving
during the parked write left php_stream_write_buffer() reading a freed
stream. flush() now takes the lock and hands it down, and stops when the
stream is gone.

stream_seek_locked() called php_stream_flush() and then read
stream->writefilters.head regardless of the outcome. It now returns -1
when the flush lost the stream.

A persistent stream outlives the request, but its lock and the callback
vector inside it come from the request allocator: the next request would
have read a dangling pointer out of stream->buffer_lock and written a
reference into freed memory. The request-end pass over the persistent list
drops the lock, so the next request builds its own.
---
 main/streams/streams.c | 47 ++++++++++++++++++++++++++++++++++--------
 1 file changed, 38 insertions(+), 9 deletions(-)

diff --git a/main/streams/streams.c b/main/streams/streams.c
index 400d0f6ec9d2..5f45327ef2f0 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -86,6 +86,15 @@ fprintf(stderr, "forget_persistent: %s:%p\n", stream->ops->label, stream);
 		stream->ctx = NULL;
 	}
 
+	/* The lock and its callback vector come from the request allocator, while
+	 * the stream itself outlives the request. No coroutine is left to hold it
+	 * here, so drop it and let the next request build its own. */
+	if (stream->buffer_lock != NULL) {
+		zend_async_event_t *lock = stream->buffer_lock;
+		stream->buffer_lock = NULL;
+		lock->dispose(lock);
+	}
+
 	return 0;
 }
 
@@ -263,7 +272,8 @@ static int _php_stream_free_persistent(zval *zv, void *pStream)
 	return le->ptr == pStream;
 }
 
-static int php_stream_flush_ex(php_stream *stream, bool closing);
+typedef struct _php_stream_buffer_lock php_stream_buffer_lock_t;
+static int php_stream_flush_ex(php_stream *stream, bool closing, php_stream_buffer_lock_t *lock);
 
 PHPAPI int php_stream_free(php_stream *stream, int close_options) /* {{{ */
 {
@@ -351,7 +361,7 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 
 	if (stream->flags & PHP_STREAM_FLAG_WAS_WRITTEN || stream->writefilters.head) {
 		/* make sure everything is saved */
-		php_stream_flush_ex(stream, true);
+		php_stream_flush_ex(stream, true, NULL);
 	}
 
 	/* If not called from the resource dtor, remove the stream from the resource list. */
@@ -466,13 +476,13 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 
 /* {{{ Buffer lock */
 
-typedef struct {
+struct _php_stream_buffer_lock {
 	zend_async_event_t base;
 	zend_coroutine_t *owner;
 	/* The buffered API re-enters itself: a seek emulated through reads, a
 	 * userspace wrapper reading its own stream. */
 	uint32_t depth;
-} php_stream_buffer_lock_t;
+};
 
 
 static void stream_buffer_unlock(php_stream_buffer_lock_t *lock);
@@ -1507,12 +1517,16 @@ static ssize_t php_stream_write_filtered(php_stream *stream, const char *buf, si
 	return consumed;
 }
 
-static int php_stream_flush_ex(php_stream *stream, bool closing)
+static int php_stream_flush_ex(php_stream *stream, bool closing, php_stream_buffer_lock_t *lock)
 {
 	int ret = 0;
 
 	if (stream->writefilters.head && stream->ops->write) {
-		php_stream_write_filtered(stream, NULL, 0, closing ? PSFS_FLAG_FLUSH_CLOSE : PSFS_FLAG_FLUSH_INC, NULL);
+		php_stream_write_filtered(stream, NULL, 0, closing ? PSFS_FLAG_FLUSH_CLOSE : PSFS_FLAG_FLUSH_INC, lock);
+
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			return -1;
+		}
 	}
 
 	stream->flags &= ~PHP_STREAM_FLAG_WAS_WRITTEN;
@@ -1525,7 +1539,17 @@ static int php_stream_flush_ex(php_stream *stream, bool closing)
 }
 
 PHPAPI int php_stream_flush(php_stream *stream) {
-	return php_stream_flush_ex(stream, false);
+	php_stream_buffer_lock_t *lock;
+
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
+		return -1;
+	}
+
+	const int ret = php_stream_flush_ex(stream, false, lock);
+
+	stream_buffer_unlock(lock);
+
+	return ret;
 }
 
 PHPAPI ssize_t php_stream_write(php_stream *stream, const char *buf, size_t count)
@@ -1643,7 +1667,7 @@ static bool php_stream_has_notifier(php_stream *stream)
 	return context && context->notifier;
 }
 
-static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence)
+static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence, php_stream_buffer_lock_t *lock)
 {
 	if (stream->fclose_stdiocast == PHP_STREAM_FCLOSE_FOPENCOOKIE) {
 		/* flush can call seek internally so we need to prevent an infinite loop */
@@ -1659,6 +1683,11 @@ static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence)
 
 	if (stream->writefilters.head) {
 		php_stream_flush(stream);
+
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			return -1;
+		}
+
 		if (!php_stream_are_filters_seekable(stream->writefilters.head, is_start_seeking,
 				PHP_STREAM_FILTER_WRITE)) {
 			return -1;
@@ -1757,7 +1786,7 @@ PHPAPI int php_stream_seek(php_stream *stream, zend_off_t offset, int whence)
 		return -1;
 	}
 
-	const int result = stream_seek_locked(stream, offset, whence);
+	const int result = stream_seek_locked(stream, offset, whence, lock);
 
 	stream_buffer_unlock(lock);
 

From e6e032135bb114a795fa0fa35a4b5b0614176c27 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Tue, 8 Sep 2026 11:49:29 +0300
Subject: [PATCH 4/4] Lock the filter chain, the cast and the copy helpers

php_stream_filter_append_ex(), php_stream_filter_flush(), php_stream_cast(),
_php_stream_copy_to_mem(), php_stream_copy_fallback() and _php_stream_passthru()
reached the read buffer and the stream position without the buffer lock. A
coroutine parked in ops->read or ops->write saw them move its buffer, and a
close during that park freed the stream they returned to: lost and duplicated
bytes from the first three, a segfault or a double free from the rest.

They now take the same lock as the buffered read and write paths and test after
every call that can park whether the stream is still there. filter.c and cast.c
reach the lock through php_stream_buffer_lock_acquire(), _release() and
_is_closed(); a copy locks its two handles in address order, so two copies
running in opposite directions cannot deadlock.

php_stream_filter_flush() also grew the read buffer without updating readbuflen,
after which writepos could pass the end of the buffer and the next fill computed
its free space as a wrapped unsigned length. That defect is in the filter code
alone and reproduces without coroutines; the test covering it is in
ext/standard/tests/filters/.
---
 ext/standard/streamsfuncs.c                   |  16 +-
 .../stream_filter_remove_buffer_length.phpt   |  79 +++++++++
 main/php_streams.h                            |  17 +-
 main/streams/cast.c                           |  23 ++-
 main/streams/filter.c                         |  83 ++++++++--
 main/streams/php_stream_filter_api.h          |   4 +
 main/streams/streams.c                        | 154 +++++++++++++++++-
 7 files changed, 351 insertions(+), 25 deletions(-)
 create mode 100644 ext/standard/tests/filters/stream_filter_remove_buffer_length.phpt

diff --git a/ext/standard/streamsfuncs.c b/ext/standard/streamsfuncs.c
index 186d0dfdb5a5..4c964900d9fc 100644
--- a/ext/standard/streamsfuncs.c
+++ b/ext/standard/streamsfuncs.c
@@ -1372,7 +1372,13 @@ static void apply_filter_to_stream(bool append, INTERNAL_FUNCTION_PARAMETERS)
 		if (append) {
 			zend_result ret = php_stream_filter_append_ex(&stream->readfilters, filter);
 			if (ret != SUCCESS) {
-				php_stream_filter_remove(filter, 1);
+				if (filter->chain == NULL) {
+					/* Never linked: the stream was closed before the append
+					 * started, and the chain to unlink from is gone with it. */
+					php_stream_filter_free(filter);
+				} else {
+					php_stream_filter_remove(filter, 1);
+				}
 				RETURN_FALSE;
 			}
 		} else {
@@ -1389,7 +1395,13 @@ static void apply_filter_to_stream(bool append, INTERNAL_FUNCTION_PARAMETERS)
 		if (append) {
 			zend_result ret = php_stream_filter_append_ex(&stream->writefilters, filter);
 			if (ret != SUCCESS) {
-				php_stream_filter_remove(filter, 1);
+				if (filter->chain == NULL) {
+					/* Never linked: the stream was closed before the append
+					 * started, and the chain to unlink from is gone with it. */
+					php_stream_filter_free(filter);
+				} else {
+					php_stream_filter_remove(filter, 1);
+				}
 				RETURN_FALSE;
 			}
 		} else {
diff --git a/ext/standard/tests/filters/stream_filter_remove_buffer_length.phpt b/ext/standard/tests/filters/stream_filter_remove_buffer_length.phpt
new file mode 100644
index 000000000000..df74c3a9ba16
--- /dev/null
+++ b/ext/standard/tests/filters/stream_filter_remove_buffer_length.phpt
@@ -0,0 +1,79 @@
+--TEST--
+Flushing a read filter keeps the read buffer length in step with the buffer
+--FILE--
+<?php
+
+/* The filter holds everything it is fed and releases it in one bucket when the
+ * chain is flushed, so the flush has to grow the stream read buffer. */
+class Hoarder extends php_user_filter
+{
+    private string $held = '';
+
+    public function filter($in, $out, &$consumed, $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            $this->held .= $bucket->data;
+        }
+
+        if ($closing) {
+            if ($this->held === '') {
+                return PSFS_FEED_ME;
+            }
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held));
+            $this->held = '';
+            return PSFS_PASS_ON;
+        }
+
+        if (strlen($this->held) > 1) {
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held[0]));
+            $this->held = substr($this->held, 1);
+            return PSFS_PASS_ON;
+        }
+
+        return PSFS_FEED_ME;
+    }
+}
+
+stream_filter_register('hoarder', 'Hoarder');
+
+const SIZE = 3200000;
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'filter_buflen_');
+file_put_contents($tmpfile, str_repeat('0123456789abcdef', SIZE / 16));
+
+$handle = fopen($tmpfile, 'r');
+$filter = stream_filter_append($handle, 'hoarder', STREAM_FILTER_READ);
+
+$taken = 0;
+for ($i = 0; $i < 12; $i++) {
+    $taken += strlen(fread($handle, 1));
+}
+
+var_dump(stream_filter_remove($filter));
+
+$rest = 0;
+while (true) {
+    $chunk = fread($handle, 4096);
+    if ($chunk === false || $chunk === '') {
+        break;
+    }
+    $rest += strlen($chunk);
+}
+
+printf("delivered: %d of %d\n", $taken + $rest, SIZE);
+printf("eof: %s\n", var_export(feof($handle), true));
+
+fclose($handle);
+
+?>
+--CLEAN--
+<?php
+foreach (glob(sys_get_temp_dir() . '/filter_buflen_*') as $leftover) {
+    unlink($leftover);
+}
+?>
+--EXPECT--
+bool(true)
+delivered: 3200000 of 3200000
+eof: true
diff --git a/main/php_streams.h b/main/php_streams.h
index 7ef996f8624d..6c77819c2c57 100644
--- a/main/php_streams.h
+++ b/main/php_streams.h
@@ -266,8 +266,23 @@ struct _php_stream  {
 #define PHP_STREAM_FCLOSE_FDOPEN	1
 #define PHP_STREAM_FCLOSE_FOPENCOOKIE 2
 
-/* allocate a new stream for a particular ops */
 BEGIN_EXTERN_C()
+
+/* Async: serialises the buffered stream API - the read buffer, the position and
+ * the filter chain - across coroutines; unrelated to php_stream_lock(), which is
+ * the advisory file lock. acquire() returns false when the caller
+ * must abandon its operation without touching the stream; on true it stores a
+ * lock that release() must be given back, NULL included: a caller outside a
+ * coroutine has nothing to serialise. The lock is recursive for its owner and
+ * holds a reference of its own, so is_closed() still answers after a call that
+ * parked, including one that returned to a stream fclose() has freed. */
+typedef struct _php_stream_buffer_lock php_stream_buffer_lock_t;
+PHPAPI bool php_stream_buffer_lock_acquire(php_stream *stream, php_stream_buffer_lock_t **held);
+PHPAPI void php_stream_buffer_lock_release(php_stream_buffer_lock_t *lock);
+/* True once the stream behind the lock has been closed and freed. */
+PHPAPI bool php_stream_buffer_lock_is_closed(php_stream_buffer_lock_t *lock);
+
+/* allocate a new stream for a particular ops */
 PHPAPI php_stream *_php_stream_alloc(const php_stream_ops *ops, void *abstract,
 		const char *persistent_id, const char *mode STREAMS_DC);
 END_EXTERN_C()
diff --git a/main/streams/cast.c b/main/streams/cast.c
index c93b9d9747f8..cd74081b901b 100644
--- a/main/streams/cast.c
+++ b/main/streams/cast.c
@@ -195,12 +195,31 @@ PHPAPI zend_result php_stream_cast(php_stream *stream, int castas, void **ret, i
 
 	/* synchronize our buffer (if possible) */
 	if (ret && castas != PHP_STREAM_AS_FD_FOR_SELECT && castas != PHP_STREAM_AS_FD_FOR_COPY) {
+		php_stream_buffer_lock_t *lock;
+
+		/* The seek moves the descriptor a parked reader is about to credit its
+		 * bytes against, and the reset drops the bytes it has not taken yet. */
+		if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+			return FAILURE;
+		}
+
 		php_stream_flush(stream);
-		if (stream->ops->seek && (stream->flags & PHP_STREAM_FLAG_NO_SEEK) == 0) {
+
+		if (!php_stream_buffer_lock_is_closed(lock) && stream->ops->seek
+				&& (stream->flags & PHP_STREAM_FLAG_NO_SEEK) == 0) {
 			zend_off_t dummy;
 
 			stream->ops->seek(stream, stream->position, SEEK_SET, &dummy);
-			stream->readpos = stream->writepos = 0;
+			if (!php_stream_buffer_lock_is_closed(lock)) {
+				stream->readpos = stream->writepos = 0;
+			}
+		}
+
+		const bool closed = php_stream_buffer_lock_is_closed(lock);
+		php_stream_buffer_lock_release(lock);
+
+		if (UNEXPECTED(closed)) {
+			return FAILURE;
 		}
 	}
 
diff --git a/main/streams/filter.c b/main/streams/filter.c
index fdc5dfc00ad2..c380b9625696 100644
--- a/main/streams/filter.c
+++ b/main/streams/filter.c
@@ -348,9 +348,33 @@ PHPAPI void php_stream_filter_prepend(php_stream_filter_chain *chain, php_stream
 	php_stream_filter_prepend_ex(chain, filter);
 }
 
+static void php_stream_filter_flush_drain(php_stream_bucket_brigade *inp, php_stream_bucket_brigade *outp)
+{
+	php_stream_bucket *bucket;
+
+	while ((bucket = inp->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+	while ((bucket = outp->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+}
+
 PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, php_stream_filter *filter)
 {
 	php_stream *stream = chain->stream;
+	php_stream_buffer_lock_t *lock;
+
+	/* Both the acquire and the wind below suspend, and the filter is still
+	 * outside the chain here: a close landing meanwhile frees the chain, not the
+	 * filter, and the caller keeps a filter it can still free. The wind itself
+	 * reallocates readbuf and resets the positions a parked reader credits its
+	 * bytes to, which is what the lock is for. */
+	if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+		return FAILURE;
+	}
 
 	filter->prev = chain->tail;
 	filter->next = NULL;
@@ -375,6 +399,13 @@ PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, p
 		php_stream_bucket_append(brig_inp, bucket);
 		status = filter->fops->filter(stream, filter, brig_inp, brig_outp, &consumed, PSFS_FLAG_NORMAL);
 
+		if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+			/* A userspace filter suspended, and the stream was closed meanwhile. */
+			php_stream_filter_flush_drain(brig_inp, brig_outp);
+			php_stream_buffer_lock_release(lock);
+			return FAILURE;
+		}
+
 		if (stream->readpos + consumed > (uint32_t)stream->writepos) {
 			/* No behaving filter should cause this. */
 			status = PSFS_ERR_FATAL;
@@ -382,18 +413,10 @@ PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, p
 
 		switch (status) {
 			case PSFS_ERR_FATAL:
-				while (brig_in.head) {
-					bucket = brig_in.head;
-					php_stream_bucket_unlink(bucket);
-					php_stream_bucket_delref(bucket);
-				}
-				while (brig_out.head) {
-					bucket = brig_out.head;
-					php_stream_bucket_unlink(bucket);
-					php_stream_bucket_delref(bucket);
-				}
+				php_stream_filter_flush_drain(&brig_in, &brig_out);
 				php_stream_warn(stream, FilterFailed,
 						"Filter failed to process pre-buffered data");
+				php_stream_buffer_lock_release(lock);
 				return FAILURE;
 			case PSFS_FEED_ME:
 				/* We don't actually need data yet,
@@ -433,12 +456,18 @@ PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, p
 		}
 	}
 
+	php_stream_buffer_lock_release(lock);
+
 	return SUCCESS;
 }
 
 PHPAPI void php_stream_filter_append(php_stream_filter_chain *chain, php_stream_filter *filter)
 {
 	if (php_stream_filter_append_ex(chain, filter) != SUCCESS) {
+		if (filter->chain != chain) {
+			/* Never linked: the stream was closed before the append started. */
+			return;
+		}
 		if (chain->head == filter) {
 			chain->head = NULL;
 			chain->tail = NULL;
@@ -468,15 +497,33 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 	chain = filter->chain;
 	stream = chain->stream;
 
+	php_stream_buffer_lock_t *lock;
+
+	/* The flush writes into the read buffer and out through ops->write, both of
+	 * which belong to whoever holds the lock. */
+	if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+		return FAILURE;
+	}
+
 	for(current = filter; current; current = current->next) {
 		php_stream_filter_status_t status;
 
 		status = current->fops->filter(stream, current, inp, outp, NULL, flags);
+		if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+			/* A userspace filter suspended, and the stream was closed meanwhile. */
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
+			return FAILURE;
+		}
 		if (status == PSFS_FEED_ME) {
 			/* We've flushed the data far enough */
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
 			return SUCCESS;
 		}
 		if (status == PSFS_ERR_FATAL) {
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
 			return FAILURE;
 		}
 		/* Otherwise we have data available to PASS_ON
@@ -499,6 +546,7 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 
 	if (flushed_size == 0) {
 		/* Unlikely, but possible */
+		php_stream_buffer_lock_release(lock);
 		return SUCCESS;
 	}
 
@@ -511,8 +559,10 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 			stream->readpos = 0;
 		}
 		if (flushed_size > (stream->readbuflen - stream->writepos)) {
-			/* Grow the buffer */
-			stream->readbuf = perealloc(stream->readbuf, stream->writepos + flushed_size + stream->chunk_size, stream->is_persistent);
+			/* Grow the buffer. readbuflen follows the allocation: the next fill
+			 * derives its free space from it and would wrap a stale one. */
+			stream->readbuflen = stream->writepos + flushed_size + stream->chunk_size;
+			stream->readbuf = perealloc(stream->readbuf, stream->readbuflen, stream->is_persistent);
 		}
 		while ((bucket = inp->head)) {
 			memcpy(stream->readbuf + stream->writepos, bucket->buf, bucket->buflen);
@@ -524,6 +574,13 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 		/* Send flushed data to the stream */
 		while ((bucket = inp->head)) {
 			ssize_t count = stream->ops->write(stream, bucket->buf, bucket->buflen);
+			if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+				/* fclose() from another coroutine freed the stream while the
+				 * write was parked; the position below is gone with it. */
+				php_stream_filter_flush_drain(inp, outp);
+				php_stream_buffer_lock_release(lock);
+				return FAILURE;
+			}
 			if (count > 0) {
 				stream->position += count;
 			}
@@ -532,6 +589,8 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 		}
 	}
 
+	php_stream_buffer_lock_release(lock);
+
 	return SUCCESS;
 }
 
diff --git a/main/streams/php_stream_filter_api.h b/main/streams/php_stream_filter_api.h
index a8ebfc57f8bc..e8e3c02a2bcd 100644
--- a/main/streams/php_stream_filter_api.h
+++ b/main/streams/php_stream_filter_api.h
@@ -138,6 +138,10 @@ BEGIN_EXTERN_C()
 PHPAPI void php_stream_filter_prepend(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI void php_stream_filter_prepend_ex(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI void php_stream_filter_append(php_stream_filter_chain *chain, php_stream_filter *filter);
+/* Appends the filter to the chain and winds any buffered data through it.
+ * On FAILURE the filter may or may not have reached the chain: filter->chain is
+ * the chain when it did, and NULL when the stream was closed before the append
+ * started. A caller that unlinks the filter must test it. */
 PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finish);
 PHPAPI php_stream_filter *php_stream_filter_remove(php_stream_filter *filter, bool call_dtor);
diff --git a/main/streams/streams.c b/main/streams/streams.c
index 5f45327ef2f0..a2dcb84ad65d 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -272,7 +272,6 @@ static int _php_stream_free_persistent(zval *zv, void *pStream)
 	return le->ptr == pStream;
 }
 
-typedef struct _php_stream_buffer_lock php_stream_buffer_lock_t;
 static int php_stream_flush_ex(php_stream *stream, bool closing, php_stream_buffer_lock_t *lock);
 
 PHPAPI int php_stream_free(php_stream *stream, int close_options) /* {{{ */
@@ -671,6 +670,47 @@ static void stream_buffer_unlock(php_stream_buffer_lock_t *lock)
 	ZEND_ASYNC_EVENT_RELEASE(&lock->base);
 }
 
+/* Two streams are locked in address order, so that two copies running in
+ * opposite directions cannot each hold what the other waits for. */
+static bool stream_buffer_lock_pair(php_stream *first, php_stream *second,
+		php_stream_buffer_lock_t **first_lock, php_stream_buffer_lock_t **second_lock)
+{
+	php_stream *lower = first < second ? first : second;
+	php_stream *upper = first < second ? second : first;
+	php_stream_buffer_lock_t *lower_lock, *upper_lock;
+
+	if (UNEXPECTED(!stream_buffer_lock(lower, &lower_lock))) {
+		return false;
+	}
+
+	if (UNEXPECTED(!stream_buffer_lock(upper, &upper_lock))) {
+		stream_buffer_unlock(lower_lock);
+		return false;
+	}
+
+	*first_lock = first < second ? lower_lock : upper_lock;
+	*second_lock = first < second ? upper_lock : lower_lock;
+
+	return true;
+}
+
+/* Contract in php_streams.h. The filter chain and the cast live in other files
+ * and reach the lock through these. */
+PHPAPI bool php_stream_buffer_lock_acquire(php_stream *stream, php_stream_buffer_lock_t **held)
+{
+	return stream_buffer_lock(stream, held);
+}
+
+PHPAPI void php_stream_buffer_lock_release(php_stream_buffer_lock_t *lock)
+{
+	stream_buffer_unlock(lock);
+}
+
+PHPAPI bool php_stream_buffer_lock_is_closed(php_stream_buffer_lock_t *lock)
+{
+	return stream_buffer_lock_closed(lock);
+}
+
 /* }}} */
 
 static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t size, php_stream_buffer_lock_t *lock)
@@ -721,6 +761,13 @@ static zend_result stream_fill_read_buffer_locked(php_stream *stream, size_t siz
 			for (filter = stream->readfilters.head; filter; filter = filter->next) {
 				status = filter->fops->filter(stream, filter, brig_inp, brig_outp, NULL, flags);
 
+				if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+					/* A userspace filter suspended, and the close freed the
+					 * chain this loop walks. */
+					status = PSFS_ERR_FATAL;
+					break;
+				}
+
 				if (status != PSFS_PASS_ON) {
 					break;
 				}
@@ -1392,6 +1439,10 @@ static ssize_t php_stream_write_buffer(php_stream *stream, const char *buf, size
 		stream->readpos = stream->writepos = 0;
 
 		stream->ops->seek(stream, stream->position, SEEK_SET, &stream->position);
+
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			return -1;
+		}
 	}
 
 	bool old_eof = stream->eof;
@@ -1463,6 +1514,13 @@ static ssize_t php_stream_write_filtered(php_stream *stream, const char *buf, si
 		status = filter->fops->filter(stream, filter, brig_inp, brig_outp,
 				filter == stream->writefilters.head ? &consumed : NULL, flags);
 
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			/* A userspace filter suspended, and the close freed the chain this
+			 * loop walks. */
+			status = PSFS_ERR_FATAL;
+			break;
+		}
+
 		if (status != PSFS_PASS_ON) {
 			break;
 		}
@@ -1763,7 +1821,8 @@ static int stream_seek_locked(php_stream *stream, zend_off_t offset, int whence,
 		char tmp[1024];
 		ssize_t didread;
 		while (offset > 0) {
-			if ((didread = php_stream_read(stream, tmp, MIN(offset, sizeof(tmp)))) <= 0) {
+			if ((didread = php_stream_read(stream, tmp, MIN(offset, sizeof(tmp)))) <= 0
+					|| UNEXPECTED(stream_buffer_lock_closed(lock))) {
 				return -1;
 			}
 			offset -= didread;
@@ -1846,6 +1905,11 @@ PHPAPI ssize_t _php_stream_passthru(php_stream * stream STREAMS_DC)
 	size_t bcount = 0;
 	char buf[8192];
 	ssize_t b;
+	php_stream_buffer_lock_t *lock;
+
+	if (UNEXPECTED(!stream_buffer_lock(stream, &lock))) {
+		return -1;
+	}
 
 	if (php_stream_mmap_possible(stream)) {
 		char *p;
@@ -1859,19 +1923,38 @@ PHPAPI ssize_t _php_stream_passthru(php_stream * stream STREAMS_DC)
 				if (0 < (b = PHPWRITE(p + bcount, MIN(mapped - bcount, INT_MAX)))) {
 					bcount += b;
 				}
+				/* The output write suspends, and the mapping is gone with the
+				 * stream a close freed meanwhile. */
+				if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+					break;
+				}
 			} while (b > 0 && mapped > bcount);
 
-			php_stream_mmap_unmap_ex(stream, mapped);
+			if (!stream_buffer_lock_closed(lock)) {
+				php_stream_mmap_unmap_ex(stream, mapped);
+			}
+
+			stream_buffer_unlock(lock);
 
 			return bcount;
 		}
 	}
 
 	while ((b = php_stream_read(stream, buf, sizeof(buf))) > 0) {
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			break;
+		}
 		PHPWRITE(buf, b);
 		bcount += b;
+		/* The output write suspends as well, so the next read would go to a
+		 * stream that is no longer there. */
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			break;
+		}
 	}
 
+	stream_buffer_unlock(lock);
+
 	if (b < 0 && bcount == 0) {
 		return b;
 	}
@@ -1896,12 +1979,20 @@ PHPAPI zend_string *_php_stream_copy_to_mem(php_stream *src, size_t maxlen, bool
 		maxlen = 0;
 	}
 
+	php_stream_buffer_lock_t *lock;
+
+	/* php_stream_read() reports what it delivered before parking, so without
+	 * the lock held across the loop the next round reads a freed stream. */
+	if (UNEXPECTED(!stream_buffer_lock(src, &lock))) {
+		return NULL;
+	}
+
 	if (maxlen > 0 && maxlen < 4 * CHUNK_SIZE) {
 		result = zend_string_alloc(maxlen, persistent);
 		ptr = ZSTR_VAL(result);
 		while ((len < maxlen) && !php_stream_eof(src)) {
 			ret = php_stream_read(src, ptr, maxlen - len);
-			if (ret <= 0) {
+			if (ret <= 0 || stream_buffer_lock_closed(lock)) {
 				// TODO: Propagate error?
 				break;
 			}
@@ -1920,6 +2011,7 @@ PHPAPI zend_string *_php_stream_copy_to_mem(php_stream *src, size_t maxlen, bool
 			zend_string_free(result);
 			result = NULL;
 		}
+		stream_buffer_unlock(lock);
 		return result;
 	}
 
@@ -1944,6 +2036,9 @@ PHPAPI zend_string *_php_stream_copy_to_mem(php_stream *src, size_t maxlen, bool
 	const int min_room = CHUNK_SIZE / 4;
 	// TODO: Propagate error?
 	while ((ret = php_stream_read(src, ptr, buflen - len)) > 0) {
+		if (UNEXPECTED(stream_buffer_lock_closed(lock))) {
+			break;
+		}
 		len += ret;
 		if (len + min_room >= buflen) {
 			if (maxlen == len) {
@@ -1968,6 +2063,8 @@ PHPAPI zend_string *_php_stream_copy_to_mem(php_stream *src, size_t maxlen, bool
 		result = NULL;
 	}
 
+	stream_buffer_unlock(lock);
+
 	return result;
 }
 
@@ -1976,11 +2073,18 @@ static zend_result php_stream_copy_fallback(php_stream *src, php_stream *dest, s
 {
 	char buf[CHUNK_SIZE];
 	size_t haveread = 0;
+	php_stream_buffer_lock_t *src_lock, *dest_lock;
 
 	if (maxlen == PHP_STREAM_COPY_ALL) {
 		maxlen = 0;
 	}
 
+	/* Either handle can be closed while the other side of the copy is parked. */
+	if (UNEXPECTED(!stream_buffer_lock_pair(src, dest, &src_lock, &dest_lock))) {
+		*len = 0;
+		return FAILURE;
+	}
+
 	while (1) {
 		size_t readchunk = sizeof(buf);
 		ssize_t didread;
@@ -1991,9 +2095,12 @@ static zend_result php_stream_copy_fallback(php_stream *src, php_stream *dest, s
 		}
 
 		didread = php_stream_read(src, buf, readchunk);
-		if (didread <= 0) {
+		const bool src_closed = stream_buffer_lock_closed(src_lock);
+		if (didread <= 0 || src_closed) {
 			*len = haveread;
-			return didread < 0 ? FAILURE : SUCCESS;
+			stream_buffer_unlock(dest_lock);
+			stream_buffer_unlock(src_lock);
+			return didread < 0 || src_closed ? FAILURE : SUCCESS;
 		}
 
 		size_t towrite = didread;
@@ -2002,8 +2109,10 @@ static zend_result php_stream_copy_fallback(php_stream *src, php_stream *dest, s
 
 		while (towrite) {
 			ssize_t didwrite = php_stream_write(dest, writeptr, towrite);
-			if (didwrite <= 0) {
+			if (didwrite <= 0 || stream_buffer_lock_closed(dest_lock)) {
 				*len = haveread - towrite;
+				stream_buffer_unlock(dest_lock);
+				stream_buffer_unlock(src_lock);
 				return FAILURE;
 			}
 
@@ -2016,6 +2125,9 @@ static zend_result php_stream_copy_fallback(php_stream *src, php_stream *dest, s
 		}
 	}
 
+	stream_buffer_unlock(dest_lock);
+	stream_buffer_unlock(src_lock);
+
 	*len = haveread;
 	return SUCCESS;
 }
@@ -2033,6 +2145,15 @@ PHPAPI zend_result _php_stream_copy_to_stream_ex(php_stream *src, php_stream *de
 		return SUCCESS;
 	}
 
+	php_stream_buffer_lock_t *src_lock, *dest_lock;
+
+	/* php_io_copy() below writes the positions of both handles back after it
+	 * returns, so it is held against a close the same way the fallback is. */
+	if (UNEXPECTED(!stream_buffer_lock_pair(src, dest, &src_lock, &dest_lock))) {
+		*len = 0;
+		return FAILURE;
+	}
+
 	/* Try optimized fd-level copy if both streams support it and their read buffers
 	 * are empty, so the fd offsets match the logical stream positions */
 	if (!php_stream_is(src, PHP_STREAM_IS_USERSPACE) && !php_stream_is(dest, PHP_STREAM_IS_USERSPACE) &&
@@ -2057,6 +2178,13 @@ PHPAPI zend_result _php_stream_copy_to_stream_ex(php_stream *src, php_stream *de
 			size_t copied = 0;
 			zend_result result = php_io_copy(&src_copy_fd, &dest_copy_fd, io_maxlen, &copied);
 
+			if (UNEXPECTED(stream_buffer_lock_closed(src_lock) || stream_buffer_lock_closed(dest_lock))) {
+				stream_buffer_unlock(dest_lock);
+				stream_buffer_unlock(src_lock);
+				*len = copied;
+				return FAILURE;
+			}
+
 			src->position += copied;
 			dest->position += copied;
 
@@ -2065,12 +2193,22 @@ PHPAPI zend_result _php_stream_copy_to_stream_ex(php_stream *src, php_stream *de
 			php_stream_set_option(dest, PHP_STREAM_OPTION_ALIGN_POSITION, 0, &dest->position);
 
 			*len = copied;
+			stream_buffer_unlock(dest_lock);
+			stream_buffer_unlock(src_lock);
 			return result;
 		}
 	}
 
 fallback:
-	return php_stream_copy_fallback(src, dest, maxlen, len);
+	;
+	/* The fallback takes the same two locks again: they are recursive for their
+	 * owner. */
+	const zend_result fallback_result = php_stream_copy_fallback(src, dest, maxlen, len);
+
+	stream_buffer_unlock(dest_lock);
+	stream_buffer_unlock(src_lock);
+
+	return fallback_result;
 }
 
 /* Returns the number of bytes moved.