Fix leaks and data loss when user filter leaves buckets on input brigade

NEWS

@@ -23,6 +23,10 @@ PHP                                                                        NEWS
     or never terminates). (Ilia Alshanetsky)
   . Fixed memory leak in JIT ASSIGN_DIM when the assignment fails. (Jorg Sowa)
 
+- Streams:
+  . Fixed leaks and data loss when user filter leaves buckets on input
+    brigade. (ndossche, Matthias Goergens, Jakub Zelenka)
+
 19 Nov 2026, PHP 8.4.28
 
 - Core:

ext/standard/tests/filters/user_filter_feed_me_across_reads.phpt

@@ -0,0 +1,39 @@
+--TEST--
+Input kept back by a read filter with PSFS_FEED_ME survives across reads and is released on close
+--SKIPIF--
+<?php
+if (!function_exists('stream_socket_pair') || !defined('STREAM_PF_UNIX') || PHP_OS_FAMILY === 'Windows') die('skip no unix socket pair');
+?>
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        if (!$closing && strlen($data) < 8) {
+            if ($data !== '') {
+                stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            }
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+[$r, $w] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+stream_set_blocking($r, false);
+stream_filter_append($r, "defer", STREAM_FILTER_READ);
+fwrite($w, "abc");
+var_dump(fread($r, 100));
+fwrite($w, "defgh");
+var_dump(fread($r, 100));
+fwrite($w, "xy");
+var_dump(fread($r, 100));
+fclose($r);
+fclose($w);
+?>
+--EXPECT--
+string(0) ""
+string(10) "[abcdefgh]"
+string(0) ""

ext/standard/tests/filters/user_filter_feed_me_chained.phpt

@@ -0,0 +1,34 @@
+--TEST--
+Input kept by a PSFS_FEED_ME filter is re-presented to that filter only, not to the filters before it
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $calls = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $this->calls++;
+        $buckets = [];
+        while ($b = stream_bucket_make_writeable($in)) { $buckets[] = $b; }
+        if ($this->calls < 3) {
+            foreach (array_reverse($buckets) as $b) { stream_bucket_prepend($in, $b); }
+            return PSFS_FEED_ME;
+        }
+        foreach ($buckets as $b) { stream_bucket_append($out, $b); }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_chained.bin";
+file_put_contents($f, str_repeat("a", 8192) . str_repeat("b", 8192) . str_repeat("c", 8192));
+$fp = fopen($f, 'r');
+stream_filter_append($fp, "string.rot13", STREAM_FILTER_READ);
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+$data = stream_get_contents($fp);
+fclose($fp);
+var_dump($data === str_repeat("n", 8192) . str_repeat("o", 8192) . str_repeat("p", 8192));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_chained.bin");
+?>
+--EXPECT--
+bool(true)

ext/standard/tests/filters/user_filter_feed_me_closing_warning.phpt

@@ -0,0 +1,29 @@
+--TEST--
+Input kept back by a PSFS_FEED_ME filter on the closing call is discarded with a warning
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $buckets = [];
+        while ($b = stream_bucket_make_writeable($in)) { $buckets[] = $b; $consumed += $b->datalen; }
+        foreach (array_reverse($buckets) as $b) { stream_bucket_prepend($in, $b); }
+        return PSFS_FEED_ME;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_closing_warning.bin";
+$fp = fopen($f, 'w');
+stream_filter_append($fp, "defer", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "hello"));
+fclose($fp);
+var_dump(file_get_contents($f));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_closing_warning.bin");
+?>
+--EXPECTF--
+int(5)
+
+Warning: fclose(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+string(0) ""

ext/standard/tests/filters/user_filter_feed_me_keeps_input.phpt

@@ -0,0 +1,32 @@
+--TEST--
+User filter returning PSFS_FEED_ME keeps buckets it put back on the input brigade
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $calls = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $this->calls++;
+        $buckets = [];
+        while ($b = stream_bucket_make_writeable($in)) { $buckets[] = $b; }
+        if ($this->calls < 3) {
+            foreach ($buckets as $b) { stream_bucket_prepend($in, $b); }
+            return PSFS_FEED_ME;
+        }
+        foreach ($buckets as $b) { stream_bucket_append($out, $b); }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_keeps_input.bin";
+file_put_contents($f, str_repeat("a", 8192) . str_repeat("b", 8192) . str_repeat("c", 8192));
+$fp = fopen($f, 'r');
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+var_dump(strlen(stream_get_contents($fp)));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_keeps_input.bin");
+?>
+--EXPECT--
+int(24576)

ext/standard/tests/filters/user_filter_feed_me_read_closing.phpt

@@ -0,0 +1,31 @@
+--TEST--
+Input kept back by a read filter with PSFS_FEED_ME is re-presented on the closing call
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        if (!$closing) {
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_read_closing.bin";
+file_put_contents($f, "abcdefghij");
+$fp = fopen($f, 'r');
+stream_set_chunk_size($fp, 4);
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+var_dump(stream_get_contents($fp));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_read_closing.bin");
+?>
+--EXPECT--
+string(12) "[abcdefghij]"

ext/standard/tests/filters/user_filter_feed_me_untouched_input.phpt

@@ -0,0 +1,35 @@
+--TEST--
+User filter returning PSFS_FEED_ME without touching the input brigade sees the original data again
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $calls = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $this->calls++;
+        if ($this->calls < 3) {
+            return PSFS_FEED_ME;
+        }
+        while ($b = stream_bucket_make_writeable($in)) {
+            stream_bucket_append($out, $b);
+        }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_untouched_input.bin";
+file_put_contents($f, str_repeat("a", 8192) . str_repeat("b", 8192) . str_repeat("c", 8192));
+$fp = fopen($f, 'r');
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+$data = stream_get_contents($fp);
+fclose($fp);
+var_dump(strlen($data), substr_count($data, "a"), substr_count($data, "b"), substr_count($data, "c"));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_untouched_input.bin");
+?>
+--EXPECT--
+int(24576)
+int(8192)
+int(8192)
+int(8192)

ext/standard/tests/filters/user_filter_feed_me_write.phpt

@@ -0,0 +1,39 @@
+--TEST--
+Write filter returning PSFS_FEED_ME sees the input it kept back again with the next write
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $kept = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        /* the bytes kept back from the previous call were already reported as consumed */
+        $consumed += strlen($data) - $this->kept;
+        if (!$closing && strlen($data) < 8) {
+            $this->kept = strlen($data);
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            return PSFS_FEED_ME;
+        }
+        $this->kept = 0;
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_write.bin";
+$fp = fopen($f, 'w');
+stream_filter_append($fp, "defer", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "abc"), fwrite($fp, "def"), fwrite($fp, "ghi"), fwrite($fp, "jk"));
+fclose($fp);
+var_dump(file_get_contents($f));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_write.bin");
+?>
+--EXPECT--
+int(3)
+int(3)
+int(3)
+int(2)
+string(15) "[abcdefghi][jk]"

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)

ext/standard/tests/filters/user_filter_pass_on_leftover_read.phpt

@@ -0,0 +1,38 @@
+--TEST--
+Input left on the brigade by a read filter returning PSFS_PASS_ON is discarded with a warning
+--FILE--
+<?php
+class lines extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; $consumed += $b->datalen; }
+        $p = strrpos($data, "\n");
+        if ($p === false) {
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, substr($data, 0, $p + 1)));
+        if ($p + 1 < strlen($data)) {
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, substr($data, $p + 1)));
+        }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("lines", "lines");
+$f = __DIR__ . "/user_filter_pass_on_leftover_read.bin";
+file_put_contents($f, "abc\ndef\nghi\n");
+$fp = fopen($f, 'r');
+stream_set_chunk_size($fp, 6);
+stream_filter_append($fp, "lines", STREAM_FILTER_READ);
+var_dump(stream_get_contents($fp));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_pass_on_leftover_read.bin");
+?>
+--EXPECTF--
+Warning: stream_get_contents(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+string(10) "abc
+f
+ghi
+"

ext/standard/tests/filters/user_filter_unprocessed_buckets_reentrant.phpt

@@ -0,0 +1,34 @@
+--TEST--
+Buckets appended to the input brigade by an error handler during the unprocessed buckets warning do not leak
+--FILE--
+<?php
+class reentrant extends php_user_filter {
+    public static bool $armed = false;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        if (!self::$armed) {
+            self::$armed = true;
+            $GLOBALS['brigade'] = $in;
+            $GLOBALS['bucket'] = stream_bucket_new($this->stream, "refilled");
+        }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("reentrant", "reentrant");
+
+set_error_handler(function (int $severity, string $message): bool {
+    if (str_contains($message, "Unprocessed filter buckets")) {
+        stream_bucket_append($GLOBALS['brigade'], $GLOBALS['bucket']);
+        echo "Handled warning\n";
+        return true;
+    }
+    return false;
+});
+
+$fp = fopen("php://memory", "w+");
+stream_filter_append($fp, "reentrant", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "input"));
+fclose($fp);
+?>
+--EXPECT--
+Handled warning
+int(0)