fix(thread): owner-tagged cross-thread dispatch (#6185 Tier 2)

crates/perry-runtime/src/agent.rs

@@ -0,0 +1,163 @@
+//! Agent (JS heap owner) identity for cross-thread queue dispatch. Issue #6185
+//! Tier 2.
+//!
+//! ## The problem this solves
+//!
+//! Perry's cross-thread plumbing keeps three process-global queues —
+//! `PENDING_THREAD_RESULTS` (`thread.rs`), `TIMER_QUEUE` / `CALLBACK_TIMERS` /
+//! `INTERVAL_TIMERS` (`timer.rs`) — whose entries carry raw pointers
+//! (`*mut Promise`, closure handles, argument `f64`s that may be NaN-boxed
+//! pointers) into *some* thread's arena. Their drains were unconditional: any
+//! thread that ran the await loop / microtask pump drained *everything*. A
+//! `perry/thread` worker doing async work could therefore steal a main-thread
+//! completion, deserialize it into the worker's arena, resolve a main-heap
+//! promise with a worker-heap pointer, and fire main-heap timer closures on the
+//! worker. When the worker exits, its arena is unmapped and the main thread
+//! reads freed memory. The `unsafe impl Send` on those queue entries carried a
+//! "only accessed from the pump thread" comment that nothing enforced.
+//!
+//! ## Why not tag with `ThreadId`
+//!
+//! The obvious fix — tag each entry with the `ThreadId` that enqueued it and
+//! have drains skip entries they don't own — breaks Android. There, the
+//! compiled TypeScript (and therefore every `setTimeout` / `setInterval` call)
+//! runs on the `perry-native` thread, but the timer pump fires from the UI
+//! thread via `nativePumpTick` (`perry-ui-android/src/app.rs`). A blanket
+//! owner-skip would leave every Android timer un-fired forever. That's the
+//! blocker that deferred the narrow slice of this fix earlier.
+//!
+//! ## The model: agents, and pumps that act on their behalf
+//!
+//! The unit that matters is not the OS thread but the **heap** — the arena the
+//! queued pointers live in. We call it an *agent*.
+//!
+//! - The program starts with one agent, [`PRIMARY_AGENT`]. Every thread that
+//!   runs user JS on the primary heap (the process main thread; Android's
+//!   `perry-native` thread) belongs to it.
+//! - Each `perry/thread` worker gets its **own** agent id at thread entry (see
+//!   [`enter_worker_agent`]), because it gets its own arena and GC.
+//! - A thread that has *not* been given an agent of its own is, by definition,
+//!   not a JS heap owner: it is a **pump** running on behalf of the primary
+//!   agent. Android's UI thread is exactly this. So [`current_agent`] falls back
+//!   to [`PRIMARY_AGENT`] rather than inventing an id.
+//!
+//! That fallback is what makes the Android path keep working unchanged: the UI
+//! thread resolves to `PRIMARY_AGENT`, which is precisely the agent whose timers
+//! `perry-native` scheduled. Meanwhile a worker resolves to its own id, so its
+//! drains can only ever touch entries it enqueued itself.
+//!
+//! Queue entries are tagged with [`current_agent`] at *enqueue* time and drains
+//! filter on [`current_agent`] at *drain* time. Foreign entries are left in
+//! place for their owner (never silently dropped mid-flight); entries belonging
+//! to an agent that has exited are purged by [`retire_agent`], since that
+//! agent's arena is gone and nothing can ever settle them.
+
+use std::cell::Cell;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Identity of a JS heap (arena + GC) that queued cross-thread work.
+pub type AgentId = u64;
+
+/// The agent that owns the process's initial JS heap: the main thread, and on
+/// Android the `perry-native` thread. Threads with no agent of their own (pump
+/// threads such as Android's UI thread) act on this agent's behalf.
+pub const PRIMARY_AGENT: AgentId = 0;
+
+/// Source of worker agent ids. Starts at 1 — 0 is reserved for
+/// [`PRIMARY_AGENT`], which is never handed out here.
+static NEXT_AGENT: AtomicU64 = AtomicU64::new(1);
+
+thread_local! {
+    /// The agent this thread *owns*. `None` means "no heap of my own" — the
+    /// main/`perry-native` JS thread (which owns the primary heap implicitly)
+    /// and any pure pump thread both leave this unset, and so resolve to
+    /// [`PRIMARY_AGENT`] via [`current_agent`].
+    static CURRENT_AGENT: Cell<Option<AgentId>> = const { Cell::new(None) };
+}
+
+/// Claim a fresh agent id for the calling thread. Called once at the top of
+/// every `perry/thread` worker (`spawn` / `parallelMap` / `parallelFilter`),
+/// before the worker can allocate or enqueue anything, so every pointer it
+/// puts into a global queue is tagged as its own.
+///
+/// Returns the new id so the caller can hand it to [`retire_agent`] at exit.
+pub fn enter_worker_agent() -> AgentId {
+    let id = NEXT_AGENT.fetch_add(1, Ordering::Relaxed);
+    CURRENT_AGENT.with(|slot| slot.set(Some(id)));
+    id
+}
+
+/// The agent whose queued work the calling thread may touch: its own if it is a
+/// worker, otherwise [`PRIMARY_AGENT`] (it is a JS thread on the primary heap,
+/// or a pump acting for it).
+pub fn current_agent() -> AgentId {
+    CURRENT_AGENT
+        .with(|slot| slot.get())
+        .unwrap_or(PRIMARY_AGENT)
+}
+
+/// True when `owner` is work the calling thread is allowed to drain and run.
+#[inline]
+pub fn owns(owner: AgentId) -> bool {
+    owner == current_agent()
+}
+
+/// Retire a worker agent at thread exit: its arena is about to be unmapped, so
+/// any queue entry still tagged with it points at memory that is about to go
+/// away. Purge those entries rather than leaving them for a drain that can
+/// never legally run.
+///
+/// Purging is safe precisely because nothing else can own them: the entries name
+/// pointers into *this* agent's arena, and no other thread may dereference them.
+pub fn retire_agent(id: AgentId) {
+    debug_assert_ne!(
+        id, PRIMARY_AGENT,
+        "the primary agent outlives the process; it is never retired"
+    );
+    crate::timer::purge_agent_timers(id);
+    crate::thread::purge_agent_thread_results(id);
+    // Deliberately do NOT clear `CURRENT_AGENT`. Clearing it would make
+    // `current_agent()` fall back to `PRIMARY_AGENT` for the rest of this
+    // thread's life — i.e. a worker that has just torn down its heap would
+    // start reporting itself as an owner of the *main* thread's queued work,
+    // reintroducing exactly the cross-heap drain this fix removes. A retired
+    // worker keeps its (now dead) id and therefore owns nothing at all.
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A plain thread that never claims a worker agent — the main thread, or a
+    /// pump like Android's UI thread — resolves to the primary agent, so it
+    /// keeps draining exactly the work the primary JS thread scheduled. This is
+    /// the property that keeps `nativePumpTick` delivering Android timers.
+    #[test]
+    fn unclaimed_threads_pump_for_the_primary_agent() {
+        assert_eq!(current_agent(), PRIMARY_AGENT);
+        assert!(owns(PRIMARY_AGENT));
+
+        let pump_thread_agent = std::thread::spawn(current_agent).join().unwrap();
+        assert_eq!(pump_thread_agent, PRIMARY_AGENT);
+    }
+
+    /// A worker claims its own agent, so it neither drains the primary agent's
+    /// work nor collides with a sibling worker.
+    #[test]
+    fn workers_get_distinct_agents_and_disown_primary_work() {
+        let a = std::thread::spawn(|| {
+            let id = enter_worker_agent();
+            assert_eq!(current_agent(), id);
+            assert!(!owns(PRIMARY_AGENT), "a worker must not own primary work");
+            id
+        })
+        .join()
+        .unwrap();
+
+        let b = std::thread::spawn(|| enter_worker_agent()).join().unwrap();
+
+        assert_ne!(a, PRIMARY_AGENT);
+        assert_ne!(b, PRIMARY_AGENT);
+        assert_ne!(a, b, "sibling workers must not share an agent");
+    }
+}

crates/perry-runtime/src/agent_dispatch_tests.rs

@@ -0,0 +1,202 @@
+//! #6185 Tier 2 — owner-tagged cross-thread dispatch.
+//!
+//! These exercise the three process-global queues the 2026-07-09 GC audit
+//! flagged (`PENDING_THREAD_RESULTS`, `TIMER_QUEUE` / `CALLBACK_TIMERS` /
+//! `INTERVAL_TIMERS`) at the Rust level: an entry enqueued by one agent must be
+//! invisible to every other agent's drain, and must survive — not be silently
+//! eaten — when a foreign agent pumps.
+//!
+//! The end-to-end JS behavior (a worker's `await` no longer stealing the main
+//! thread's `spawn` completion) rides on top of this and is covered by the
+//! compiled integration test in
+//! `crates/perry/tests/issue_6185_thread_owner_tagged_dispatch.rs`.
+
+use crate::agent::{current_agent, enter_worker_agent, owns, retire_agent, PRIMARY_AGENT};
+
+/// The timer queues are process-global, and `cargo test` runs these in parallel
+/// threads of one process — so two tests that both schedule on the PRIMARY agent
+/// would see each other's timers in `active_timeout_resource_count()`. Serialize
+/// the ones that assert on primary-agent counts.
+static TIMER_QUEUE_TESTS: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// The Android invariant, stated as a test: `nativePumpTick` runs on the UI
+/// thread, which never claims a worker agent, so it must resolve to the same
+/// agent as the `perry-native` thread that scheduled the timers. If this ever
+/// regresses to a raw-`ThreadId` comparison, every Android timer stops firing —
+/// the exact blocker that deferred this fix the first time round.
+#[test]
+fn a_pump_thread_shares_the_primary_agent_with_the_js_thread() {
+    // The "perry-native" thread: runs JS, never claims a worker agent.
+    let js_thread = std::thread::spawn(current_agent).join().unwrap();
+    // The UI thread: pure pump, also never claims a worker agent.
+    let ui_pump_thread = std::thread::spawn(current_agent).join().unwrap();
+
+    assert_eq!(js_thread, PRIMARY_AGENT);
+    assert_eq!(
+        ui_pump_thread, js_thread,
+        "the UI pump must own the JS thread's work, or Android timers never fire"
+    );
+    // And it can in fact claim that work.
+    assert!(std::thread::spawn(|| owns(PRIMARY_AGENT)).join().unwrap());
+}
+
+/// A worker owns its own agent and disowns the primary agent's work — this is
+/// what stops a worker's await loop from draining the main thread's queues.
+#[test]
+fn a_worker_disowns_the_primary_agents_work() {
+    let (worker_agent, worker_owns_primary) = std::thread::spawn(|| {
+        let id = enter_worker_agent();
+        let owns_primary = owns(PRIMARY_AGENT);
+        retire_agent(id);
+        (id, owns_primary)
+    })
+    .join()
+    .unwrap();
+
+    assert_ne!(worker_agent, PRIMARY_AGENT);
+    assert!(
+        !worker_owns_primary,
+        "a worker draining primary-agent work is the #6185 use-after-free"
+    );
+}
+
+/// A timer scheduled by the primary agent must not be *fired* by a worker — and
+/// must still be there afterwards for its real owner. The pre-fix drain would
+/// have run the callback on the worker (cross-heap call) or, with a naive
+/// filter-and-drop, silently eaten it.
+#[test]
+fn a_worker_neither_fires_nor_eats_a_primary_agent_timer() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+    // Schedule on the primary agent (this test thread).
+    let before = crate::timer::active_timeout_resource_count();
+    let id = crate::timer::js_set_timeout_callback(0, 50_000.0);
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before + 1,
+        "the owning agent sees its own timer as an active handle"
+    );
+
+    // A worker pumps the timer queues. It must fire nothing of ours...
+    let fired_on_worker = std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        let fired = crate::timer::js_callback_timer_tick() + crate::timer::js_interval_timer_tick();
+        // ...and must not count our timer as work keeping ITS loop alive.
+        let visible = crate::timer::active_timeout_resource_count();
+        let has_pending = crate::timer::js_callback_timer_has_pending();
+        retire_agent(agent);
+        (fired, visible, has_pending)
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(fired_on_worker.0, 0, "a worker must not fire our timers");
+    assert_eq!(
+        fired_on_worker.1, 0,
+        "a worker must not see our timer as its active handle"
+    );
+    assert_eq!(
+        fired_on_worker.2, 0,
+        "a foreign timer must not keep a worker's event loop alive"
+    );
+
+    // The timer is still ours, untouched, after the foreign pump.
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before + 1,
+        "a foreign pump must not consume the owner's timer"
+    );
+
+    crate::timer::clearTimeout(id);
+}
+
+/// The GC root scanner must walk only the collecting agent's own timers.
+///
+/// This is the subtlest half of the fix. `scan_timer_roots_mut` runs on every GC
+/// cycle and, on an *evacuating* cycle, REWRITES the slots it visits to
+/// forwarding addresses in the collecting thread's arena. Walking a foreign
+/// agent's timer there would mark through a heap this collector does not own
+/// (racing that agent's collector on the same `GcHeader` bits) and rewrite its
+/// promise/closure slots to point into the wrong arena.
+///
+/// So a worker collecting while the primary agent holds a timer must visit zero
+/// of the primary agent's slots — while the owner still roots its own.
+#[test]
+fn a_workers_gc_scan_never_visits_another_agents_timer_slots() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+
+    // A promise timer on the primary agent. Its `promise` (a raw pointer into
+    // THIS arena) and `value` slots are scanned unconditionally, which is what
+    // makes it a clean probe for "did the collector touch a foreign slot?".
+    let baseline = {
+        let mut n = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| n += 1);
+        n
+    };
+    let _promise = crate::timer::js_set_timeout(50_000.0);
+    let visited_by_owner = {
+        let mut n = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| n += 1);
+        n
+    };
+    assert!(
+        visited_by_owner > baseline,
+        "the owning agent must root its own timer's slots, or the promise is swept"
+    );
+
+    // The same scan on a worker must not see any of it.
+    let visited_on_worker = std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        let mut visited = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| visited += 1);
+        retire_agent(agent);
+        visited
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(
+        visited_on_worker, 0,
+        "a worker's GC must not mark (or evacuate-rewrite) the primary agent's timer slots"
+    );
+
+    crate::timer::test_clear_all_timer_scanner_roots();
+}
+
+/// Retiring a worker purges the timers it scheduled: its arena is unmapped, so
+/// nothing can ever legally run those callbacks. Pre-fix they stayed on the
+/// global queue and fired on whatever thread pumped next — reading freed memory.
+#[test]
+fn retiring_a_worker_purges_the_timers_it_scheduled() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+    let before = crate::timer::active_timeout_resource_count();
+
+    std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        // A worker scheduling a timer: the callback closure would live in this
+        // worker's arena.
+        crate::timer::js_set_timeout_callback(0, 60_000.0);
+        crate::timer::setInterval(0, 60_000.0);
+        assert!(
+            crate::timer::active_timeout_resource_count() >= 2,
+            "the worker sees its own timers while it is alive"
+        );
+        retire_agent(agent);
+        assert_eq!(
+            crate::timer::active_timeout_resource_count(),
+            0,
+            "retiring the agent drops the timers it owned"
+        );
+        assert!(
+            !owns(PRIMARY_AGENT),
+            "a retired worker must not fall back to owning the primary agent's work"
+        );
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before,
+        "a dead worker's timers must not linger on the primary agent's queue"
+    );
+}

crates/perry-runtime/src/atomics.rs

@@ -775,13 +775,17 @@ pub extern "C" fn js_atomics_wait_async(
     }
     crate::thread::thread_job_begin();
     let promise_usize = promise as usize;
+    // #6185: the promise belongs to the agent calling `waitAsync`. The futex
+    // waiter below owns no heap (it never runs JS), so it cannot infer the
+    // right agent from itself — capture it here, on the calling thread.
+    let owner_agent = crate::agent::current_agent();
     std::thread::spawn(move || {
         let outcome = crate::atomics_futex::block(handle, deadline);
         let result = match outcome {
             crate::atomics_futex::WaitOutcome::Ok => "ok",
             _ => "timed-out",
         };
-        crate::thread::queue_promise_string_result(promise_usize, result);
+        crate::thread::queue_promise_string_result(owner_agent, promise_usize, result);
     });
     wait_async_result(true, crate::value::js_nanbox_pointer(promise as i64))
 }

crates/perry-runtime/src/lib.rs

@@ -31,6 +31,9 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 static GLOBAL: std::alloc::System = std::alloc::System;
 
 pub mod abi_trampoline;
+pub mod agent;
+#[cfg(test)]
+mod agent_dispatch_tests;
 pub mod app_group;
 pub mod arena;
 pub mod array;