diff --git a/ext/standard/tests/filters/user_filter_no_consume_leak.phpt b/ext/standard/tests/filters/user_filter_no_consume_leak.phpt
new file mode 100644
index 0000000000000..c13b69c5892cd
--- /dev/null
+++ b/ext/standard/tests/filters/user_filter_no_consume_leak.phpt
@@ -0,0 +1,22 @@
+--TEST--
+User filter that does not consume any input bucket leaks
+--FILE--
+<?php
+
+class noop_filter extends php_user_filter {
+	public function filter($in, $out, &$consumed, bool $closing): int {
+		return PSFS_PASS_ON;
+	}
+}
+
+stream_filter_register("noop_filter", "noop_filter");
+
+stream_filter_append(STDOUT, "noop_filter");
+
+$out = fwrite(STDOUT, "Hello\n");
+var_dump($out);
+
+?>
+--EXPECTF--
+Warning: fwrite(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+int(0)
diff --git a/ext/standard/user_filters.c b/ext/standard/user_filters.c
index 962eaaba7d6b0..0028441df7ffb 100644
--- a/ext/standard/user_filters.c
+++ b/ext/standard/user_filters.c
@@ -190,9 +190,25 @@ php_stream_filter_status_t userfilter_filter(
 	}
 
 	if (buckets_in->head) {
+		php_stream_bucket *bucket;
+		do {
+			bucket = buckets_in->head;
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		} while (buckets_in->head);
 		php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade");
 	}
 
+	/* Filter could've broken contract and added buckets anyway. */
+	if (ret == PSFS_FEED_ME && buckets_out->head) {
+		php_stream_bucket *bucket;
+		do {
+			bucket = buckets_out->head;
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		} while (buckets_out->head);
+	}
+
 	/* filter resources are cleaned up by the stream destructor,
 	 * keeping a reference to the stream resource here would prevent it
 	 * from being destroyed properly */
diff --git a/main/streams/filter.c b/main/streams/filter.c
index 1ce03d41513cd..f6e47405dedec 100644
--- a/main/streams/filter.c
+++ b/main/streams/filter.c
@@ -355,13 +355,6 @@ PHPAPI int php_stream_filter_append_ex(php_stream_filter_chain *chain, php_strea
 				   Reset stream's internal read buffer since the filter is "holding" it. */
 				stream->readpos = 0;
 				stream->writepos = 0;
-
-				/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-				while (brig_out.head) {
-					bucket = brig_out.head;
-					php_stream_bucket_unlink(bucket);
-					php_stream_bucket_delref(bucket);
-				}
 				break;
 			case PSFS_PASS_ON:
 				/* If any data is consumed, we cannot rely upon the existing read buffer,
diff --git a/main/streams/streams.c b/main/streams/streams.c
index 372ed6635c3f0..6628ec76c3fe2 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -631,12 +631,6 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 					/* when a filter needs feeding, there is no brig_out to deal with.
 					 * we simply continue the loop; if the caller needs more data,
 					 * we will read again, otherwise out job is done here */
-
-					/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-					while ((bucket = brig_outp->head)) {
-						php_stream_bucket_unlink(bucket);
-						php_stream_bucket_delref(bucket);
-					}
 					break;
 
 				case PSFS_ERR_FATAL:
@@ -1275,16 +1269,10 @@ static ssize_t _php_stream_write_filtered(php_stream *stream, const char *buf, s
 			/* some fatal error.  Theoretically, the stream is borked, so all
 			 * further writes should fail. */
 			consumed = (ssize_t) -1;
-			ZEND_FALLTHROUGH;
+			break;
 
 		case PSFS_FEED_ME:
 			/* need more data before we can push data through to the stream */
-			/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-			while (brig_inp->head) {
-				bucket = brig_inp->head;
-				php_stream_bucket_unlink(bucket);
-				php_stream_bucket_delref(bucket);
-			}
 			break;
 	}