diff --git a/ext/standard/io_poll.c b/ext/standard/io_poll.c
index 5632fc270c56..dd0b27ae0067 100644
--- a/ext/standard/io_poll.c
+++ b/ext/standard/io_poll.c
@@ -17,6 +17,7 @@
 #include "zend_exceptions.h"
 #include "php_network.h"
 #include "php_poll.h"
+#include "io_poll.h"
 #include "io_poll_arginfo.h"
 #include "io_poll_decl.h"
 #include "ext/date/php_time.h"
@@ -54,14 +55,17 @@ typedef struct php_io_poll_watcher_object {
 	uint32_t triggered_events;
 	zval data;
 	bool active;
+	bool closed; /* Deactivated because its stream was closed */
 	php_io_poll_context_object *context; /* Back reference to Context object */
+	php_socket_t fd; /* Registered fd, SOCK_ERR when inactive */
+	php_stream *stream; /* Watched stream, NULL when not registered in its watcher list */
 	zend_object std;
 } php_io_poll_watcher_object;
 
 /* Context object structure */
 struct php_io_poll_context_object {
 	php_poll_ctx *ctx;
-	HashTable *watchers; /* Maps handle pointer -> watcher object */
+	HashTable *watchers; /* Maps fd -> watcher object */
 	zend_object std;
 };
 
@@ -266,7 +270,10 @@ static zend_object *php_io_poll_watcher_create_object(zend_class_entry *ce)
 	intern->watched_events = 0;
 	intern->triggered_events = 0;
 	intern->active = false;
+	intern->closed = false;
 	intern->context = NULL;
+	intern->fd = SOCK_ERR;
+	intern->stream = NULL;
 	ZVAL_NULL(&intern->data);
 
 	return &intern->std;
@@ -285,12 +292,88 @@ static zend_object *php_io_poll_context_create_object(zend_class_entry *ce)
 	return &intern->std;
 }
 
+/* Watcher registration helpers */
+
+static zend_always_inline zend_ulong php_io_poll_compute_ptr_key(void *ptr)
+{
+	zend_ulong key = (zend_ulong) (uintptr_t) ptr;
+	return (key >> 3) | (key << ((sizeof(key) * 8) - 3));
+}
+
+static zend_always_inline void php_io_poll_watcher_deactivate(php_io_poll_watcher_object *watcher)
+{
+	watcher->active = false;
+	watcher->context = NULL;
+	watcher->fd = SOCK_ERR;
+}
+
+static void php_io_poll_stream_watch(php_stream *stream, php_io_poll_watcher_object *watcher)
+{
+	if (!stream->poll_watchers) {
+		stream->poll_watchers = pemalloc(sizeof(HashTable), stream->is_persistent);
+		zend_hash_init(stream->poll_watchers, 4, NULL, NULL, stream->is_persistent);
+	}
+
+	zval zv;
+	ZVAL_PTR(&zv, watcher);
+	zend_hash_index_add_new(stream->poll_watchers, php_io_poll_compute_ptr_key(watcher), &zv);
+	watcher->stream = stream;
+}
+
+static void php_io_poll_stream_unwatch(php_io_poll_watcher_object *watcher)
+{
+	php_stream *stream = watcher->stream;
+
+	if (!stream) {
+		return;
+	}
+	watcher->stream = NULL;
+
+	zend_hash_index_del(stream->poll_watchers, php_io_poll_compute_ptr_key(watcher));
+	if (zend_hash_num_elements(stream->poll_watchers) == 0) {
+		zend_hash_destroy(stream->poll_watchers);
+		pefree(stream->poll_watchers, stream->is_persistent);
+		stream->poll_watchers = NULL;
+	}
+}
+
+static void php_io_poll_context_retire_watcher(
+		php_io_poll_context_object *context, php_io_poll_watcher_object *watcher)
+{
+	php_socket_t fd = watcher->fd;
+
+	php_poll_remove(context->ctx, (int) fd);
+	php_io_poll_stream_unwatch(watcher);
+	php_io_poll_watcher_deactivate(watcher);
+	zend_hash_index_del(context->watchers, (zend_ulong) fd);
+}
+
+/* Called from php_stream_free() while the fd is still open */
+PHPAPI void php_io_poll_stream_notify_close(php_stream *stream)
+{
+	HashTable *watchers = stream->poll_watchers;
+	stream->poll_watchers = NULL;
+
+	ZEND_HASH_FOREACH_VAL(watchers, zval *zv) {
+		php_io_poll_watcher_object *watcher = Z_PTR_P(zv);
+		watcher->stream = NULL;
+		watcher->closed = true;
+		if (watcher->context) {
+			php_io_poll_context_retire_watcher(watcher->context, watcher);
+		}
+	} ZEND_HASH_FOREACH_END();
+
+	zend_hash_destroy(watchers);
+	pefree(watchers, stream->is_persistent);
+}
+
 /* Object Destruction Functions */
 
 static void php_io_poll_watcher_free_object(zend_object *obj)
 {
 	php_io_poll_watcher_object *intern = PHP_POLL_WATCHER_OBJ_FROM_ZOBJ(obj);
 
+	php_io_poll_stream_unwatch(intern);
 	zval_ptr_dtor(&intern->data);
 
 	if (intern->handle) {
@@ -307,8 +390,8 @@ static void php_io_poll_context_free_object(zend_object *obj)
 	if (intern->watchers) {
 		ZEND_HASH_FOREACH_VAL(intern->watchers, zval *zv) {
 			php_io_poll_watcher_object *watcher = PHP_POLL_WATCHER_OBJ_FROM_ZOBJ(Z_OBJ_P(zv));
-			watcher->active = false;
-			watcher->context = NULL;
+			php_io_poll_stream_unwatch(watcher);
+			php_io_poll_watcher_deactivate(watcher);
 		} ZEND_HASH_FOREACH_END();
 	}
 
@@ -355,12 +438,6 @@ static HashTable *php_io_poll_context_get_gc(zend_object *obj, zval **table, int
 
 /* Utility functions */
 
-static zend_always_inline zend_ulong php_io_poll_compute_ptr_key(void *ptr)
-{
-	zend_ulong key = (zend_ulong) (uintptr_t) ptr;
-	return (key >> 3) | (key << ((sizeof(key) * 8) - 3));
-}
-
 static zend_result php_io_poll_watcher_modify_events(
 		php_io_poll_watcher_object *watcher, uint32_t events)
 {
@@ -370,16 +447,12 @@ static zend_result php_io_poll_watcher_modify_events(
 		return FAILURE;
 	}
 
-	php_socket_t fd = php_poll_handle_get_fd(watcher->handle);
-	if (fd == SOCK_ERR) {
-		zend_throw_exception(
-				php_io_poll_invalid_handle_class_entry, "Invalid handle for polling", 0);
-		return FAILURE;
-	}
-
-	/* Modify in poll context */
+	/* Re-add if the backend dropped a fired one-shot registration */
 	php_poll_ctx *poll_ctx = watcher->context->ctx;
-	if (php_poll_modify(poll_ctx, (int) fd, events, watcher) != SUCCESS) {
+	int fd = (int) watcher->fd;
+	if (php_poll_modify(poll_ctx, fd, events, watcher) != SUCCESS
+			&& (php_poll_get_error(poll_ctx) != PHP_POLL_ERR_NOTFOUND
+					|| php_poll_add(poll_ctx, fd, events, watcher) != SUCCESS)) {
 		php_poll_error err = php_poll_get_error(poll_ctx);
 		php_io_poll_throw_failed_operation(php_io_poll_failed_watcher_mod_class_entry,
 				"Failed to modify watcher in polling system", err);
@@ -634,26 +707,16 @@ PHP_METHOD(Io_Poll_Watcher, remove)
 	php_io_poll_watcher_object *intern = PHP_POLL_WATCHER_OBJ_FROM_ZV(getThis());
 
 	if (!intern->active || !intern->context) {
+		/* Closing the stream already removed it, so this is just the expected cleanup */
+		if (intern->closed) {
+			return;
+		}
 		zend_throw_exception(
 				php_io_poll_inactive_watcher_class_entry, "Cannot remove inactive watcher", 0);
 		RETURN_THROWS();
 	}
 
-	php_io_poll_context_object *context = intern->context;
-	php_poll_ctx *poll_ctx = context->ctx;
-	HashTable *watchers = context->watchers;
-	zend_ulong hash_key = php_io_poll_compute_ptr_key(intern->handle);
-	php_socket_t fd = php_poll_handle_get_fd(intern->handle);
-	if (fd != SOCK_ERR) {
-		php_poll_remove(poll_ctx, (int) fd);
-	}
-
-	intern->active = false;
-	intern->context = NULL;
-
-	if (watchers) {
-		zend_hash_index_del(watchers, hash_key);
-	}
+	php_io_poll_context_retire_watcher(intern->context, intern);
 }
 
 PHP_METHOD(Io_Poll_Context, __construct)
@@ -715,6 +778,12 @@ PHP_METHOD(Io_Poll_Context, add)
 	php_io_poll_context_object *intern = PHP_POLL_CONTEXT_OBJ_FROM_ZV(getThis());
 	php_poll_handle_object *handle = PHP_POLL_HANDLE_OBJ_FROM_ZV(handle_obj);
 
+	events = php_io_poll_event_enums_to_events(event_enums);
+	if (!events) {
+		zend_argument_type_error(2, "must be array of Event enums");
+		RETURN_THROWS();
+	}
+
 	/* Get file descriptor */
 	php_socket_t fd = php_poll_handle_get_fd(handle);
 	if (fd == SOCK_ERR) {
@@ -723,16 +792,21 @@ PHP_METHOD(Io_Poll_Context, add)
 		RETURN_THROWS();
 	}
 
+	zval *existing_zv = zend_hash_index_find(intern->watchers, (zend_ulong) fd);
+	if (existing_zv) {
+		php_io_poll_watcher_object *existing = PHP_POLL_WATCHER_OBJ_FROM_ZOBJ(Z_OBJ_P(existing_zv));
+		if (php_poll_handle_get_fd(existing->handle) == fd) {
+			zend_throw_exception(
+					php_io_poll_handle_already_watched_class_entry, "Handle already added", 0);
+			RETURN_THROWS();
+		}
+		php_io_poll_context_retire_watcher(intern, existing);
+	}
+
 	/* Create watcher object */
 	object_init_ex(return_value, php_io_poll_watcher_class_entry);
 	php_io_poll_watcher_object *watcher = PHP_POLL_WATCHER_OBJ_FROM_ZV(return_value);
 
-	events = php_io_poll_event_enums_to_events(event_enums);
-	if (!events) {
-		zend_argument_type_error(2, "must be array of Event enums");
-		RETURN_THROWS();
-	}
-
 	watcher->handle = handle;
 	watcher->watched_events = events;
 	watcher->triggered_events = 0;
@@ -758,16 +832,22 @@ PHP_METHOD(Io_Poll_Context, add)
 		RETURN_THROWS();
 	}
 
-	/* Store in our watchers map using shifted pointer as key */
+	/* Store in our watchers map */
 	zval watcher_zv;
 	ZVAL_OBJ(&watcher_zv, &watcher->std);
 	GC_ADDREF(&watcher->std);
-
-	zend_ulong hash_key = php_io_poll_compute_ptr_key(handle);
-	zend_hash_index_add_new(intern->watchers, hash_key, &watcher_zv);
+	zend_hash_index_add_new(intern->watchers, (zend_ulong) fd, &watcher_zv);
 
 	watcher->active = true;
 	watcher->context = intern;
+	watcher->fd = fd;
+
+	if (handle->ops == &php_stream_poll_handle_ops) {
+		php_stream *stream = php_stream_poll_handle_get_stream(handle);
+		if (stream) {
+			php_io_poll_stream_watch(stream, watcher);
+		}
+	}
 }
 
 PHP_METHOD(Io_Poll_Context, wait)
diff --git a/ext/standard/io_poll.h b/ext/standard/io_poll.h
new file mode 100644
index 000000000000..1b94355f27a5
--- /dev/null
+++ b/ext/standard/io_poll.h
@@ -0,0 +1,26 @@
+/*
+   +----------------------------------------------------------------------+
+   | Copyright © The PHP Group and Contributors.                          |
+   +----------------------------------------------------------------------+
+   | This source file is subject to the Modified BSD License that is      |
+   | bundled with this package in the file LICENSE, and is available      |
+   | through the World Wide Web at <https://www.php.net/license/>.        |
+   |                                                                      |
+   | SPDX-License-Identifier: BSD-3-Clause                                |
+   +----------------------------------------------------------------------+
+   | Author: Jakub Zelenka <bukka@php.net>                                |
+   +----------------------------------------------------------------------+
+*/
+
+#ifndef PHP_IO_POLL_H
+#define PHP_IO_POLL_H
+
+#include "php_streams.h"
+
+BEGIN_EXTERN_C()
+
+PHPAPI void php_io_poll_stream_notify_close(php_stream *stream);
+
+END_EXTERN_C()
+
+#endif /* PHP_IO_POLL_H */
diff --git a/ext/standard/tests/poll/poll_stream_closed_dup_fd.phpt b/ext/standard/tests/poll/poll_stream_closed_dup_fd.phpt
new file mode 100644
index 000000000000..9fafd91a46aa
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_dup_fd.phpt
@@ -0,0 +1,49 @@
+--TEST--
+Io\Poll: a watched stream closed while a duplicated fd exists is unregistered in time
+--SKIPIF--
+<?php
+if (!function_exists('proc_open')) {
+    die("skip proc_open required\n");
+}
+if (!Io\Poll\Backend::Epoll->isAvailable()) {
+    die("skip Epoll backend required\n");
+}
+?>
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+// The child keeps a duplicate of $r open, so epoll would keep the interest after
+// fclose() unless the watcher is unregistered while the fd is still open
+$ctx = new Io\Poll\Context(Io\Poll\Backend::Epoll);
+list($r, $w) = pt_new_socket_pair();
+$watcher = $ctx->add(new StreamPollHandle($r), [Io\Poll\Event::Read]);
+
+$proc = proc_open('sleep 2', [0 => $r], $pipes);
+fclose($r);
+$watcher->remove();
+unset($watcher);
+gc_collect_cycles();
+
+fwrite($w, "ping");
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+
+// A new stream that takes the same fd number gets its own registration
+list($r2, $w2) = pt_new_socket_pair();
+$watcher2 = $ctx->add(new StreamPollHandle($r2), [Io\Poll\Event::Read], "new");
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+fwrite($w2, "pong");
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+echo "Events count: ", count($events), "\n";
+var_dump($events[0]->getData());
+
+proc_terminate($proc);
+proc_close($proc);
+echo "done\n";
+?>
+--EXPECT--
+Events count: 0
+Events count: 0
+Events count: 1
+string(3) "new"
+done
diff --git a/ext/standard/tests/poll/poll_stream_closed_fd_reuse.phpt b/ext/standard/tests/poll/poll_stream_closed_fd_reuse.phpt
new file mode 100644
index 000000000000..74c359f301bf
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_fd_reuse.phpt
@@ -0,0 +1,47 @@
+--TEST--
+Io\Poll: a new stream reusing the fd number of a closed watched stream can be added
+--SKIPIF--
+<?php
+if (strtoupper(substr(PHP_OS, 0, 3)) == 'WIN') {
+    die("skip fd numbers are not reused predictably on Windows\n");
+}
+?>
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+$ctx = pt_new_stream_poll();
+
+list($r0, $w0) = pt_new_socket_pair();
+$old = $ctx->add(new StreamPollHandle($r0), [Io\Poll\Event::Read], "old");
+
+// The lowest free fd number is handed out first, so $r1 takes the number $r0 had
+fclose($r0);
+list($r1, $w1) = pt_new_socket_pair();
+
+$new = $ctx->add(new StreamPollHandle($r1), [Io\Poll\Event::Read], "new");
+var_dump($old->isActive());
+var_dump($new->isActive());
+
+fwrite($w1, "ping");
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+var_dump(count($events));
+var_dump($events[0] === $new);
+var_dump($events[0]->getData());
+
+$old->remove();
+echo "old removed\n";
+
+$new->remove();
+var_dump($new->isActive());
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+?>
+--EXPECT--
+bool(false)
+bool(true)
+int(1)
+bool(true)
+string(3) "new"
+old removed
+bool(false)
+Events count: 0
diff --git a/ext/standard/tests/poll/poll_stream_closed_fd_reuse_unwatched.phpt b/ext/standard/tests/poll/poll_stream_closed_fd_reuse_unwatched.phpt
new file mode 100644
index 000000000000..bf699c948586
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_fd_reuse_unwatched.phpt
@@ -0,0 +1,36 @@
+--TEST--
+Io\Poll: an unwatched stream reusing the fd number of a closed watched stream is never reported
+--SKIPIF--
+<?php
+if (strtoupper(substr(PHP_OS, 0, 3)) == 'WIN') {
+    die("skip fd numbers are not reused predictably on Windows\n");
+}
+?>
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+$ctx = pt_new_stream_poll();
+
+list($r0, $w0) = pt_new_socket_pair();
+$old = $ctx->add(new StreamPollHandle($r0), [Io\Poll\Event::Read], "old");
+
+// The lowest free fd number is handed out first, so $r1 takes the number $r0 had
+fclose($r0);
+list($r1, $w1) = pt_new_socket_pair();
+fwrite($w1, "ping");
+
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+var_dump($old->isActive());
+
+$old->remove();
+var_dump($old->isActive());
+var_dump(fread($r1, 10));
+?>
+--EXPECT--
+Events count: 0
+Events count: 0
+bool(false)
+bool(false)
+string(4) "ping"
diff --git a/ext/standard/tests/poll/poll_stream_closed_many.phpt b/ext/standard/tests/poll/poll_stream_closed_many.phpt
new file mode 100644
index 000000000000..3f436fc940a6
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_many.phpt
@@ -0,0 +1,52 @@
+--TEST--
+Io\Poll: many watchers closed before removal leave the context consistent
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+$ctx = pt_new_stream_poll();
+
+list($rk, $wk) = pt_new_socket_pair();
+$keep = $ctx->add(new StreamPollHandle($rk), [Io\Poll\Event::Read], "keep");
+
+list($ro, $wo) = pt_new_socket_pair();
+$oneshot = $ctx->add(new StreamPollHandle($ro), [Io\Poll\Event::Read, Io\Poll\Event::OneShot], "oneshot");
+fwrite($wo, "x");
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+var_dump(count($events), $events[0]->getData());
+
+for ($i = 0; $i < 200; $i++) {
+    list($r, $w) = pt_new_socket_pair();
+    $watcher = $ctx->add(new StreamPollHandle($r), [Io\Poll\Event::Read]);
+    fclose($r);
+    $watcher->remove();
+    fclose($w);
+}
+
+// The fired one-shot watcher stays disarmed, the kept one is still watched
+fwrite($wk, "y");
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+var_dump(count($events), $events[0]->getData());
+var_dump($keep->isActive(), $oneshot->isActive());
+
+$oneshot->modifyEvents([Io\Poll\Event::Read, Io\Poll\Event::OneShot]);
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+var_dump(count($events));
+
+list($r, $w) = pt_new_socket_pair();
+$new = $ctx->add(new StreamPollHandle($r), [Io\Poll\Event::Read], "new");
+fwrite($w, "z");
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+var_dump(count($events));
+var_dump($oneshot->isActive());
+?>
+--EXPECT--
+int(1)
+string(7) "oneshot"
+int(1)
+string(4) "keep"
+bool(true)
+bool(true)
+int(2)
+int(2)
+bool(true)
diff --git a/ext/standard/tests/poll/poll_stream_closed_multi_context.phpt b/ext/standard/tests/poll/poll_stream_closed_multi_context.phpt
new file mode 100644
index 000000000000..c14b4803bee2
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_multi_context.phpt
@@ -0,0 +1,34 @@
+--TEST--
+Io\Poll: closing a stream watched in several contexts leaves every watcher removable
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+list($r, $w) = pt_new_socket_pair();
+$handle = new StreamPollHandle($r);
+
+$ctx1 = pt_new_stream_poll();
+$ctx2 = pt_new_stream_poll();
+$w1 = $ctx1->add($handle, [Io\Poll\Event::Read]);
+$w2 = $ctx2->add($handle, [Io\Poll\Event::Read]);
+
+fwrite($w, "ping");
+fclose($r);
+
+echo "ctx1 events count: ", count($ctx1->wait(Time\Duration::fromSeconds(0))), "\n";
+echo "ctx2 events count: ", count($ctx2->wait(Time\Duration::fromSeconds(0))), "\n";
+var_dump($w1->isActive());
+var_dump($w2->isActive());
+
+$w1->remove();
+$w2->remove();
+echo "removed\n";
+
+fclose($w);
+?>
+--EXPECT--
+ctx1 events count: 0
+ctx2 events count: 0
+bool(false)
+bool(false)
+removed
diff --git a/ext/standard/tests/poll/poll_stream_closed_persistent.phpt b/ext/standard/tests/poll/poll_stream_closed_persistent.phpt
new file mode 100644
index 000000000000..b26f24b7cc9a
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_persistent.phpt
@@ -0,0 +1,28 @@
+--TEST--
+Io\Poll: a watcher on a persistent stream is unregistered at request shutdown
+--SKIPIF--
+<?php
+$srv = @stream_socket_server("tcp://127.0.0.1:0", $e1, $e2);
+if (!$srv) {
+    die("skip cannot bind loopback listener\n");
+}
+?>
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+$srv = stream_socket_server("tcp://127.0.0.1:0", $e1, $e2);
+$addr = stream_socket_get_name($srv, false);
+$p = pfsockopen("tcp://" . $addr, -1, $en, $es, 1);
+
+$ctx = pt_new_stream_poll();
+$watcher = $ctx->add(new StreamPollHandle($p), [Io\Poll\Event::Write]);
+echo "Events count: ", count($ctx->wait(Time\Duration::fromMicroseconds(100000))), "\n";
+
+// The persistent stream outlives the request, so the watcher and context are
+// intentionally left to be freed at shutdown
+echo "done\n";
+?>
+--EXPECT--
+Events count: 1
+done
diff --git a/ext/standard/tests/poll/poll_stream_closed_watcher_inactive.phpt b/ext/standard/tests/poll/poll_stream_closed_watcher_inactive.phpt
new file mode 100644
index 000000000000..7c5e943ebe5a
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_closed_watcher_inactive.phpt
@@ -0,0 +1,38 @@
+--TEST--
+Io\Poll: closing a watched stream deactivates its watcher and remove() stays harmless
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+list($r, $w) = pt_new_socket_pair();
+$ctx = pt_new_stream_poll();
+$watcher = $ctx->add(new StreamPollHandle($r), [Io\Poll\Event::Read]);
+
+fwrite($w, "ping");
+var_dump($watcher->isActive());
+
+fclose($r);
+var_dump($watcher->isActive());
+
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+
+$watcher->remove();
+$watcher->remove();
+echo "removed\n";
+var_dump($watcher->isActive());
+
+try {
+    $watcher->modifyEvents([Io\Poll\Event::Write]);
+} catch (Io\Poll\InactiveWatcherException $e) {
+    echo $e::class, ': ', $e->getMessage(), "\n";
+}
+
+fclose($w);
+?>
+--EXPECT--
+bool(true)
+bool(false)
+Events count: 0
+removed
+bool(false)
+Io\Poll\InactiveWatcherException: Cannot modify inactive watcher
diff --git a/ext/standard/tests/poll/poll_stream_handle_closed_stream.phpt b/ext/standard/tests/poll/poll_stream_handle_closed_stream.phpt
index 56b5c08fadef..5e9e6729aa98 100644
--- a/ext/standard/tests/poll/poll_stream_handle_closed_stream.phpt
+++ b/ext/standard/tests/poll/poll_stream_handle_closed_stream.phpt
@@ -19,7 +19,7 @@ var_dump(get_debug_type($handle->getStream()));
 
 try {
     $watcher->modifyEvents([Io\Poll\Event::Write]);
-} catch (Io\Poll\InvalidHandleException $e) {
+} catch (Io\Poll\InactiveWatcherException $e) {
     echo $e->getMessage(), "\n";
 }
 
@@ -37,7 +37,7 @@ var_dump($watcher->isActive());
 --EXPECT--
 bool(false)
 string(17) "resource (closed)"
-Invalid handle for polling
+Cannot modify inactive watcher
 Invalid handle for polling
 Events count: 0
 bool(false)
diff --git a/ext/standard/tests/poll/poll_stream_oneshot_rearm.phpt b/ext/standard/tests/poll/poll_stream_oneshot_rearm.phpt
new file mode 100644
index 000000000000..5ccc43113dae
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_oneshot_rearm.phpt
@@ -0,0 +1,44 @@
+--TEST--
+Io\Poll: a fired one-shot watcher stays active and is re-armed by modifyEvents()
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+list($r, $w) = pt_new_socket_pair();
+$ctx = pt_new_stream_poll();
+$handle = new StreamPollHandle($r);
+$watcher = $ctx->add($handle, [Io\Poll\Event::Read, Io\Poll\Event::OneShot], "data");
+
+fwrite($w, "a");
+echo "Events count: ", count($ctx->wait(Time\Duration::fromMicroseconds(100000))), "\n";
+var_dump(fread($r, 10));
+
+fwrite($w, "b");
+echo "Events count: ", count($ctx->wait(Time\Duration::fromSeconds(0))), "\n";
+var_dump($watcher->isActive());
+
+try {
+    $ctx->add($handle, [Io\Poll\Event::Read]);
+} catch (Io\Poll\HandleAlreadyWatchedException $e) {
+    echo $e::class, ': ', $e->getMessage(), "\n";
+}
+
+$watcher->modifyEvents([Io\Poll\Event::Read, Io\Poll\Event::OneShot]);
+$events = $ctx->wait(Time\Duration::fromMicroseconds(100000));
+echo "Events count: ", count($events), "\n";
+var_dump($events[0] === $watcher);
+var_dump(fread($r, 10));
+
+$watcher->remove();
+var_dump($watcher->isActive());
+?>
+--EXPECT--
+Events count: 1
+string(1) "a"
+Events count: 0
+bool(true)
+Io\Poll\HandleAlreadyWatchedException: Handle already added
+Events count: 1
+bool(true)
+string(1) "b"
+bool(false)
diff --git a/main/php_streams.h b/main/php_streams.h
index fb0c57ecf83d..69bd996fea4f 100644
--- a/main/php_streams.h
+++ b/main/php_streams.h
@@ -250,6 +250,8 @@ struct _php_stream  {
 	struct _php_stream *enclosing_stream; /* this is a private stream owned by enclosing_stream */
 
 	zend_llist *error_list;
+
+	HashTable *poll_watchers; /* Io\Poll watchers notified before the stream is closed */
 }; /* php_stream */
 
 #define PHP_STREAM_CONTEXT(stream) \
diff --git a/main/poll/poll_backend_kqueue.c b/main/poll/poll_backend_kqueue.c
index 19ff0ad6d228..f2c0212f4d15 100644
--- a/main/poll/poll_backend_kqueue.c
+++ b/main/poll/poll_backend_kqueue.c
@@ -279,6 +279,11 @@ static zend_result kqueue_backend_remove(php_poll_ctx *ctx, int fd)
 		return FAILURE;
 	}
 
+	/* Remove from tracking */
+	if (!ctx->raw_events) {
+		zend_hash_index_del(backend_data->fd_tracking, fd);
+	}
+
 	/* If no filters were successfully deleted, that's an error */
 	if (successful_deletes == 0) {
 		php_poll_set_error(ctx, PHP_POLL_ERR_NOTFOUND);
@@ -289,11 +294,6 @@ static zend_result kqueue_backend_remove(php_poll_ctx *ctx, int fd)
 	backend_data->fd_count--;
 	backend_data->filter_count -= successful_deletes;
 
-	/* Remove from tracking */
-	if (!ctx->raw_events) {
-		zend_hash_index_del(backend_data->fd_tracking, fd);
-	}
-
 	return SUCCESS;
 }
 
diff --git a/main/streams/streams.c b/main/streams/streams.c
index 7cd63f0038d3..0862dc7081e5 100644
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -28,6 +28,7 @@
 #include "ext/standard/basic_functions.h" /* for BG(CurrentStatFile) */
 #include "ext/standard/php_string.h" /* for php_memnstr, used by php_stream_get_record() */
 #include "ext/uri/php_uri.h"
+#include "ext/standard/io_poll.h"
 #include <stddef.h>
 #include <fcntl.h>
 #include "php_streams_int.h"
@@ -371,6 +372,11 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 			return ret;
 		}
 
+		/* Watchers must unregister while the fd is still open */
+		if (stream->poll_watchers) {
+			php_io_poll_stream_notify_close(stream);
+		}
+
 		ret = stream->ops->close(stream, preserve_handle ? 0 : 1);
 		if (!ret) {
 			ret = flush_result;
@@ -386,6 +392,10 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 	}
 
 	if (close_options & PHP_STREAM_FREE_RELEASE_STREAM) {
+		if (stream->poll_watchers) {
+			php_io_poll_stream_notify_close(stream);
+		}
+
 		while (stream->readfilters.head) {
 			if (stream->readfilters.head->res != NULL) {
 				zend_list_close(stream->readfilters.head->res);