fix(renderer): release retired renderer I/O drivers

Require Tokio 1.53.1 so dropping worker wake receivers releases their registered wakers even while browser-context senders remain live. This frees stopped renderer I/O drivers instead of retaining an eventpoll/eventfd pair per closed target.

Cover deferred services, live services, and deferred-to-live initialization while preserving a live peer receiver.

Cargo.lock

@@ -5080,9 +5080,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.52.1"
+version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
  "bytes",
  "libc",

moli-renderer-v8/Cargo.toml

@@ -96,7 +96,10 @@ serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 serde_v8 = "=0.320.0"
 strum = { version = "0.28.0", features = ["derive"] }
-tokio = { version = "1.51.0", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
+# Tokio >=1.53 releases a dropped mpsc receiver's waker even while context-wide
+# senders remain (tokio-rs/tokio#8095). Older versions retain stopped renderer
+# I/O drivers and leak an eventpoll/eventfd pair per closed target on Linux.
+tokio = { version = "1.53.1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
 tracing = "0.1.41"
 url = "2.5.7"
 uuid = { version = "1.23.1", default-features = false }

moli-renderer-v8/src/runtime/browser_context_runtime.rs

@@ -941,6 +941,9 @@ impl RendererWorkerContextRuntime {
     }
 }
 
+#[cfg(test)]
+mod owner_wake_tests;
+
 #[cfg(test)]
 mod tests {
     use std::time::Duration;

moli-renderer-v8/src/runtime/browser_context_runtime/owner_wake_tests.rs

@@ -0,0 +1,104 @@
+use std::{
+    sync::Arc,
+    task::{Context, Wake, Waker},
+};
+
+use super::RendererBrowserContextRuntime;
+use crate::{
+    service_worker_runtime::service_worker_owner_wake_channel,
+    shared_worker_runtime::shared_worker_owner_wake_channel,
+};
+
+enum WorkerServices {
+    Deferred,
+    AlreadyLive,
+    InitializedAfterRegistration,
+}
+
+struct ReceiverWake;
+
+// A static Waker::noop() cannot witness whether the channel retains its owner.
+#[allow(clippy::manual_noop_waker)]
+impl Wake for ReceiverWake {
+    fn wake(self: Arc<Self>) {}
+}
+
+fn assert_retired_receiver_releases_waker(services: WorkerServices) {
+    let context = RendererBrowserContextRuntime::new();
+    let initialize_workers = || {
+        context.inner.shared_worker_runtime.get_or_init();
+        context.inner.service_worker_runtime.get_or_init();
+    };
+    if matches!(services, WorkerServices::AlreadyLive) {
+        initialize_workers();
+    }
+
+    let (shared_tx, mut shared_rx) = shared_worker_owner_wake_channel();
+    let (service_tx, mut service_rx) = service_worker_owner_wake_channel();
+    context.add_shared_worker_owner_wake_sender(shared_tx);
+    context.add_service_worker_owner_wake_sender(service_tx);
+    let (peer_shared_tx, mut peer_shared_rx) = shared_worker_owner_wake_channel();
+    let (peer_service_tx, mut peer_service_rx) = service_worker_owner_wake_channel();
+    context.add_shared_worker_owner_wake_sender(peer_shared_tx);
+    context.add_service_worker_owner_wake_sender(peer_service_tx);
+
+    if matches!(services, WorkerServices::InitializedAfterRegistration) {
+        initialize_workers();
+    }
+
+    let shared_wake = Arc::new(ReceiverWake);
+    let shared_weak = Arc::downgrade(&shared_wake);
+    let service_wake = Arc::new(ReceiverWake);
+    let service_weak = Arc::downgrade(&service_wake);
+    {
+        let shared_waker = Waker::from(shared_wake);
+        let service_waker = Waker::from(service_wake);
+        assert!(
+            shared_rx
+                .poll_recv(&mut Context::from_waker(&shared_waker))
+                .is_pending()
+        );
+        assert!(
+            service_rx
+                .poll_recv(&mut Context::from_waker(&service_waker))
+                .is_pending()
+        );
+    }
+    drop(shared_rx);
+    drop(service_rx);
+
+    // The context (including its senders) stays live. Before Tokio 1.53,
+    // receiver drop retained the last block_on waker and its I/O driver.
+    // Assert before any new registration/wake can happen to prune old routes.
+    assert!(
+        shared_weak.upgrade().is_none(),
+        "retired SharedWorker receiver retained its waker"
+    );
+    assert!(
+        service_weak.upgrade().is_none(),
+        "retired ServiceWorker receiver retained its waker"
+    );
+    assert!(matches!(
+        peer_shared_rx.try_recv(),
+        Err(tokio::sync::mpsc::error::TryRecvError::Empty)
+    ));
+    assert!(matches!(
+        peer_service_rx.try_recv(),
+        Err(tokio::sync::mpsc::error::TryRecvError::Empty)
+    ));
+}
+
+#[test]
+fn retired_deferred_worker_wake_receivers_release_wakers_while_context_lives() {
+    assert_retired_receiver_releases_waker(WorkerServices::Deferred);
+}
+
+#[test]
+fn retired_live_worker_wake_receivers_release_wakers_without_affecting_peer() {
+    assert_retired_receiver_releases_waker(WorkerServices::AlreadyLive);
+}
+
+#[test]
+fn retired_worker_wake_receivers_release_wakers_after_service_initialization() {
+    assert_retired_receiver_releases_waker(WorkerServices::InitializedAfterRegistration);
+}