diff --git a/packages/doeff-vm-core/src/arena.rs b/packages/doeff-vm-core/src/arena.rs index 29967d26..7d7fc006 100644 --- a/packages/doeff-vm-core/src/arena.rs +++ b/packages/doeff-vm-core/src/arena.rs @@ -1,13 +1,50 @@ //! Fiber arena for stable fiber IDs within a run. +use std::sync::{Arc, Mutex}; + use crate::continuation::{DetachedFiber, DetachedFiberChain}; use crate::error::VMError; use crate::ids::FiberId; use crate::segment::Fiber; +/// Channel through which a dropped `DetachedFiberChain` returns its arena +/// slot indices for reuse (#497). +/// +/// This is allocator bookkeeping, NOT fiber ownership (ISSUE-VM-001 G1 / +/// SPEC-VM-021): no Fiber, chain, or continuation ever flows through it. +/// The chain's fibers move into Continuation ownership at detach and are +/// destroyed by the chain's own Drop; only the now-permanently-vacant slot +/// indices are reported here so the arena can return them to its free list +/// instead of stranding them until run end. Reports may arrive from +/// arbitrary Python dealloc points, on any thread — hence the Mutex. +#[derive(Debug, Default)] +pub struct SlotReclaimQueue { + dropped_slot_indices: Mutex>, +} + +impl SlotReclaimQueue { + /// Called from `DetachedFiberChain::drop` with the slot indices the + /// chain still owned when it was abandoned. + pub(crate) fn report_dropped_slots(&self, indices: impl Iterator) { + let mut queue = self.dropped_slot_indices.lock().unwrap(); + queue.extend(indices); + } + + /// Drain all pending reports. Cheap when empty (no allocation). + fn take_pending(&self) -> Vec { + let mut queue = self.dropped_slot_indices.lock().unwrap(); + std::mem::take(&mut *queue) + } +} + pub struct FiberArena { fibers: Vec>, free_list: Vec, + /// Reclaim reports from detached chains dropped without reattachment + /// (#497). Drained back into `free_list` before allocating. Replaced + /// wholesale on `clear()` so a chain outliving its run session cannot + /// poison the next session's free list. + slot_reclaim: Arc, } impl FiberArena { @@ -15,10 +52,12 @@ impl FiberArena { FiberArena { fibers: Vec::new(), free_list: Vec::new(), + slot_reclaim: Arc::new(SlotReclaimQueue::default()), } } pub fn alloc(&mut self, fiber: Fiber) -> FiberId { + self.reclaim_dropped_chain_slots(); if let Some(idx) = self.free_list.pop() { self.fibers[idx] = Some(fiber); FiberId::from_index(idx) @@ -29,6 +68,31 @@ impl FiberArena { } } + /// Return slots abandoned by dropped detached chains to the free list. + /// + /// A detached chain owns its arena slots (single-location law): they + /// stay vacant-reserved while the continuation is live. When the chain + /// is dropped without reattachment its Drop impl reports the indices to + /// `slot_reclaim`; this reconciliation makes them allocatable again + /// instead of stranding until `clear()` at run end (#497). + pub fn reclaim_dropped_chain_slots(&mut self) { + for idx in self.slot_reclaim.take_pending() { + #[cfg(feature = "invariant-checks")] + { + if !matches!(self.fibers.get(idx), Some(None)) { + panic!( + "arena: dropped-chain slot {idx} is not vacant-reserved \ + (single-location law violated by reclaim)" + ); + } + if self.free_list.contains(&idx) { + panic!("arena: dropped-chain slot {idx} is already on the free list"); + } + } + self.free_list.push(idx); + } + } + pub fn free(&mut self, id: FiberId) { if let Some(slot) = self.fibers.get_mut(id.index()) { if slot.take().is_some() { @@ -58,6 +122,7 @@ impl FiberArena { let mut chain = DetachedFiberChain::new(head, last_fiber, detached); let _ = chain.set_tail_parent(None); + chain.arm_slot_reclaim(Arc::clone(&self.slot_reclaim)); Ok(chain) } @@ -151,6 +216,10 @@ impl FiberArena { pub fn clear(&mut self) { self.fibers.clear(); self.free_list.clear(); + // Detach from chains that outlive this session: their late drops + // report into the replaced queue, which nobody drains — a stale + // index from a previous session must never reach the new free list. + self.slot_reclaim = Arc::new(SlotReclaimQueue::default()); } pub fn shrink_to_fit(&mut self) { @@ -350,4 +419,92 @@ mod tests { Some(unrelated) ); } + + #[test] + fn test_dropped_chain_slots_are_reclaimed_on_next_alloc() { + // #497: a detached chain dropped without reattachment (abort-style + // handler, scheduler cancellation) must return its slots to the + // free list instead of stranding them until run end. + let mut arena = FiberArena::new(); + + let boundary = arena.alloc(Fiber::new(None)); + let body = arena.alloc(Fiber::new(Some(boundary))); + + let chain = arena.detach_chain(body, boundary).unwrap(); + assert_eq!(arena.len(), 0); + assert_eq!(arena.slot_count(), 2); + + drop(chain); + + let reused_a = arena.alloc(Fiber::new(None)); + let reused_b = arena.alloc(Fiber::new(None)); + assert!(reused_a.index() < 2, "first alloc must reuse a reclaimed slot"); + assert!(reused_b.index() < 2, "second alloc must reuse a reclaimed slot"); + assert_eq!( + arena.slot_count(), + 2, + "slot vector must not grow after an abandoned chain is dropped" + ); + } + + #[test] + fn test_abort_loop_slot_count_is_bounded() { + // #497 regression shape: repeated detach-then-drop cycles (one per + // abandoned dispatch) must keep the slot vector bounded. + let mut arena = FiberArena::new(); + for _ in 0..100 { + let boundary = arena.alloc(Fiber::new(None)); + let body = arena.alloc(Fiber::new(Some(boundary))); + let chain = arena.detach_chain(body, boundary).unwrap(); + drop(chain); + } + assert_eq!(arena.len(), 0); + assert_eq!( + arena.slot_count(), + 2, + "100 abandoned chains must reuse the same two slots" + ); + } + + #[test] + fn test_chain_dropped_after_clear_does_not_poison_next_session() { + // A chain can outlive its run session (a Python K held across + // run()). Its late drop must not inject stale indices into the + // next session's free list. + let mut arena = FiberArena::new(); + let boundary = arena.alloc(Fiber::new(None)); + let body = arena.alloc(Fiber::new(Some(boundary))); + let chain = arena.detach_chain(body, boundary).unwrap(); + + arena.clear(); // run session ends while the chain is still owned outside + + let live = arena.alloc(Fiber::new(None)); + drop(chain); // stale report goes to the orphaned queue + + let next = arena.alloc(Fiber::new(None)); + assert_ne!(next, live, "stale reclaim must not hand out a live slot"); + assert!(arena.get(live).is_some(), "live fiber must survive stale drops"); + assert_eq!(arena.len(), 2); + } + + #[test] + fn test_attached_chain_does_not_reclaim_slots() { + // Consuming a chain via attach_chain must NOT report its slots: + // the fibers are live in the arena again. + let mut arena = FiberArena::new(); + let boundary = arena.alloc(Fiber::new(None)); + let body = arena.alloc(Fiber::new(Some(boundary))); + let chain = arena.detach_chain(body, boundary).unwrap(); + + arena.attach_chain(chain, None).unwrap(); + arena.reclaim_dropped_chain_slots(); + + assert_eq!(arena.len(), 2); + let fresh = arena.alloc(Fiber::new(None)); + assert_eq!( + fresh.index(), + 2, + "no slot may be recycled while its fiber is live" + ); + } } diff --git a/packages/doeff-vm-core/src/continuation.rs b/packages/doeff-vm-core/src/continuation.rs index 568b4d1f..e1b297a5 100644 --- a/packages/doeff-vm-core/src/continuation.rs +++ b/packages/doeff-vm-core/src/continuation.rs @@ -16,11 +16,16 @@ //! construction: `Option::take()` returns `Some` first time, `None` after. //! The VM does not store continuations; for exception recovery during handler //! dispatch, it keeps a `Py` reference (a Python handle, not a -//! continuation) — see vm.rs `pending_handler_k_handle`. +//! continuation) owned by the dispatch itself — a local in +//! eval_perform/eval_perform_with_k, then `Frame::Program.handler_k_handle` +//! or `EvalReturnContinuation::ExpandReturn.handler_k_handle` (#492). + +use std::sync::Arc; use pyo3::prelude::*; use pyo3::types::PyDict; +use crate::arena::SlotReclaimQueue; use crate::ids::FiberId; use crate::ir_stream::StreamSourceLocation; use crate::memory_stats; @@ -55,6 +60,14 @@ pub struct DetachedFiberChain { head: FiberId, last_fiber: FiberId, fibers: Vec, + /// Where to report still-owned arena slot indices if this chain is + /// dropped without being reattached (#497). Armed by + /// `FiberArena::detach_chain`; `None` for chains constructed outside an + /// arena (tests). Carries bare slot indices only — never fibers or + /// continuations (ISSUE-VM-001 G1); the chain never touches the arena + /// directly (Drop can fire at arbitrary Python dealloc points, + /// including from another thread). + slot_reclaim_queue: Option>, } impl DetachedFiberChain { @@ -63,9 +76,17 @@ impl DetachedFiberChain { head, last_fiber, fibers, + slot_reclaim_queue: None, } } + /// Arm slot reclamation: on drop, any fibers this chain still owns are + /// reported to `queue` so the owning arena can return their slots to + /// its free list (#497). + pub(crate) fn arm_slot_reclaim(&mut self, queue: Arc) { + self.slot_reclaim_queue = Some(queue); + } + pub fn head(&self) -> FiberId { self.head } @@ -78,8 +99,10 @@ impl DetachedFiberChain { &self.fibers } - pub fn into_fibers(self) -> Vec { - self.fibers + pub fn into_fibers(mut self) -> Vec { + // Drain in place: `self` then drops with no owned fibers, so the + // Drop impl reports nothing — the caller now owns the fibers. + std::mem::take(&mut self.fibers) } pub fn set_parent(&mut self, id: FiberId, parent: Option) -> bool { @@ -267,6 +290,27 @@ impl DetachedFiberChain { } } +impl Drop for DetachedFiberChain { + /// A chain dropped while still owning fibers was abandoned without + /// reattachment (abort-style handler, scheduler cancellation, dropped + /// parked K). Report the owned slot indices to the arena's reclaim + /// queue so they return to the free list instead of stranding as + /// vacant-reserved until run end (#497). Consumption paths + /// (`into_fibers`, `append`) drain `fibers` first, so they report + /// nothing here. + fn drop(&mut self) { + if self.fibers.is_empty() { + return; + } + let Some(queue) = &self.slot_reclaim_queue else { + // Chain never belonged to an arena (direct construction in + // tests) — there is no free list to return slots to. + return; + }; + queue.report_dropped_slots(self.fibers.iter().map(|entry| entry.id.index())); + } +} + // --------------------------------------------------------------------------- // Continuation — the detached fiber chain // --------------------------------------------------------------------------- @@ -452,6 +496,28 @@ impl OwnedControlContinuation { // PyK — Python-visible continuation handle (sole owner) // --------------------------------------------------------------------------- +/// GC note (#500): PyK deliberately implements NEITHER `__traverse__` nor +/// `__clear__`, unlike the other Py-holding pyclasses. +/// +/// - `__clear__` is unsafe by construction: a live PyK is the SOLE owner of +/// a detached fiber chain (SPEC-VM-021 move-only invariant), and during a +/// handler dispatch the VM may still recover that chain through a +/// dispatch-owned `Py` handle to reattach it for exception +/// propagation (#492). Dropping the chain from the GC's `tp_clear` while +/// such a handle exists would destroy the one-shot ownership invariant +/// mid-dispatch. +/// - `__traverse__` alone is not implementable with the current internals: +/// the Python references inside the chain live behind `PyShared` handles +/// in fibers → frames → `dyn IRStream` streams → `Value`s, none of which +/// expose a GC-visit API. Walking them would require threading a visitor +/// through the whole frame/stream abstraction. +/// +/// Consequence: the GC treats PyK as an opaque leaf. Cycles that merely +/// pass through a PyK handle held in a DoExpr field (Resume/Transfer/...) +/// are still detected via those classes' `__traverse__`; a cycle that +/// closes through PyK's interior (e.g. a suspended generator captured in +/// the chain referencing the PyK itself) remains uncollectable until the +/// continuation is consumed or the PyK is dropped by refcount. #[pyclass(name = "K")] pub struct PyK { continuation: Option, diff --git a/packages/doeff-vm-core/src/do_ctrl.rs b/packages/doeff-vm-core/src/do_ctrl.rs index ba9458c0..965dbfc8 100644 --- a/packages/doeff-vm-core/src/do_ctrl.rs +++ b/packages/doeff-vm-core/src/do_ctrl.rs @@ -68,7 +68,10 @@ pub enum DoCtrl { /// Install an observer and execute body. /// observer: Value::Callable — called synchronously with (effect) on every perform. - /// Return value ignored. Original effect always proceeds. + /// Return value ignored. If the observer raises, the dispatch is aborted + /// and the exception propagates like a handler error — the effect does + /// NOT proceed (fail-fast, #506). On success the original effect always + /// proceeds unchanged. /// body: DoExpr — evaluated under the observer. WithObserve { observer: Value, body: Box }, diff --git a/packages/doeff-vm-core/src/frame.rs b/packages/doeff-vm-core/src/frame.rs index eeeb21f4..9b783187 100644 --- a/packages/doeff-vm-core/src/frame.rs +++ b/packages/doeff-vm-core/src/frame.rs @@ -149,7 +149,19 @@ pub enum EvalReturnContinuation { continuation: InterceptorContinuation, }, /// Expand: inner expr evaluated, result must be Value::Stream → push as frame. - ExpandReturn, + /// + /// `handler_k_handle` is Some only when this Expand came from a handler + /// dispatch (eval_perform / eval_perform_with_k on the generator-handler + /// path, e.g. the @do wrapper's `Expand(Apply(Pure(thunk), []))`). The + /// handle owns the perform-site continuation chain for the duration of + /// the deferred handler construction; it is consumed on EVERY exit from + /// this frame — moved into `Frame::Program.handler_k_handle` on the value + /// path, or used to discontinue the perform-site chain on the raise path + /// (see step_raise). Storing it here (not in a VM-global slot) ties its + /// lifetime to the dispatch that created it. + ExpandReturn { + handler_k_handle: Option>, + }, } impl EvalReturnContinuation { @@ -169,7 +181,7 @@ impl EvalReturnContinuation { | EvalReturnContinuation::TailResumeReturn | EvalReturnContinuation::ReturnToContinuation { .. } | EvalReturnContinuation::EvalInScopeReturn { .. } - | EvalReturnContinuation::ExpandReturn => None, + | EvalReturnContinuation::ExpandReturn { .. } => None, } } } diff --git a/packages/doeff-vm-core/src/vm.rs b/packages/doeff-vm-core/src/vm.rs index 83cb914c..1b3fe12a 100644 --- a/packages/doeff-vm-core/src/vm.rs +++ b/packages/doeff-vm-core/src/vm.rs @@ -29,26 +29,21 @@ pub struct VM { pub segments: FiberArena, pub var_store: VarStore, pub current_segment: Option, - /// Transient slot used by `eval_perform`/`eval_perform_with_k` to thread - /// a reference to the handler's PyK object through to `push_stream_value`. - /// Set before calling a handler callable; consumed when the resulting - /// Stream frame is pushed (the handle ends up in - /// `Frame::Program.handler_k_handle`). - /// - /// This is a Python handle (Py), NOT a continuation. The chain lives - /// inside the PyK — if the handler raises, the VM borrows this handle, - /// calls `PyK::take()`, and reattaches the recovered chain. - /// Always None outside the handler-dispatch window. - pub pending_handler_k_handle: Option>, } +// NOTE (#492): there is deliberately NO VM-global `pending_handler_k_handle` +// slot. The handler-dispatch PyK handle is owned by the dispatch itself +// (a local in eval_perform/eval_perform_with_k) and, for deferred handler +// construction, by the `EvalReturnContinuation::ExpandReturn` frame — so its +// lifetime is structurally tied to the dispatch window and cannot leak into +// an unrelated later frame. + impl VM { pub fn new() -> Self { VM { segments: FiberArena::new(), var_store: VarStore::new(), current_segment: None, - pending_handler_k_handle: None, } } @@ -56,7 +51,6 @@ impl VM { self.segments.clear(); self.var_store.clear_run_local(); self.current_segment = None; - self.pending_handler_k_handle = None; } pub fn end_active_run_session(&mut self) { @@ -65,7 +59,6 @@ impl VM { self.var_store.clear_run_local(); self.var_store.shrink_run_local_to_fit(); self.current_segment = None; - self.pending_handler_k_handle = None; } pub fn alloc_segment(&mut self, fiber: Fiber) -> FiberId { diff --git a/packages/doeff-vm-core/src/vm/invariants.rs b/packages/doeff-vm-core/src/vm/invariants.rs index fb4ee0c5..23865a35 100644 --- a/packages/doeff-vm-core/src/vm/invariants.rs +++ b/packages/doeff-vm-core/src/vm/invariants.rs @@ -183,17 +183,9 @@ impl VM { fn inv_detached_chains(&self, violations: &mut Vec) -> DetachedView { let mut detached_ids: HashSet = HashSet::new(); - // Roots visible from the VM itself (pending k handle). - if let Some(handle) = &self.pending_handler_k_handle { - self.scan_k_handle( - handle, - "VM.pending_handler_k_handle", - &mut detached_ids, - violations, - ); - } - // Roots inside live arena fibers (frame k handles + scope values). + // There is no VM-global pending-handle slot (#492): every dispatch + // k handle lives either in a Program frame or an ExpandReturn frame. let live_ids: Vec = self.segments.iter().map(|(id, _)| id).collect(); for id in live_ids { if let Some(fiber) = self.segments.get(id) { @@ -272,6 +264,19 @@ impl VM { ); } } + Frame::EvalReturn(cont) => { + if let EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } = cont.as_ref() + { + self.scan_k_handle( + handle, + &format!("{origin} (ExpandReturn.handler_k_handle)"), + detached_ids, + violations, + ); + } + } Frame::LexicalScope { bindings, var_overrides, @@ -410,13 +415,23 @@ impl VM { } // Recurse into frames carried by detached fibers that may - // carry their own handler_k_handle. + // carry their own handler_k_handle (Program frames and + // in-flight ExpandReturn frames alike). for frame in &entry.fiber.frames { - if let Frame::Program { - handler_k_handle: Some(nested_handle), - .. - } = frame - { + let nested_handle = match frame { + Frame::Program { + handler_k_handle: Some(handle), + .. + } => Some(handle), + Frame::EvalReturn(cont) => match cont.as_ref() { + EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } => Some(handle), + _ => None, + }, + _ => None, + }; + if let Some(nested_handle) = nested_handle { self.scan_k_handle( nested_handle, &format!("{origin} → detached fiber {:?}", entry.id), @@ -479,6 +494,7 @@ impl VM { #[cfg(test)] mod tests { + use crate::frame::{EvalReturnContinuation, Frame}; use crate::ids::Marker; use crate::segment::{Fiber, Handler}; use crate::value::Value; @@ -544,17 +560,24 @@ mod tests { fn properly_detached_chain_holds_invariants() { // Python::attach auto-initializes with the `auto-initialize` feature. let mut vm = VM::new(); + let root = vm.alloc_segment(Fiber::new(None)); let boundary = vm.alloc_segment(Fiber::new(None)); let body = vm.alloc_segment(Fiber::new(Some(boundary))); let chain = vm.segments.detach_chain(body, boundary).unwrap(); let k = crate::continuation::Continuation::from_chain(chain); + // Root the handle where dispatch state lives now (#492): an + // in-flight ExpandReturn frame, not a VM-global slot. pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .unwrap(); - vm.pending_handler_k_handle = Some(py_k); + vm.segments.get_mut(root).unwrap().push_frame(Frame::EvalReturn( + Box::new(EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(py_k), + }), + )); }); assert!(vm.check_invariants().is_ok()); } @@ -564,6 +587,7 @@ mod tests { use crate::continuation::{Continuation, DetachedFiber, DetachedFiberChain}; // Python::attach auto-initializes with the `auto-initialize` feature. let mut vm = VM::new(); + let root = vm.alloc_segment(Fiber::new(None)); // A fiber that is LIVE in the arena... let live = vm.alloc_segment(Fiber::new(None)); // ...and simultaneously claimed by a detached chain — the @@ -583,7 +607,11 @@ mod tests { crate::continuation::PyK::from_continuation(k), ) .unwrap(); - vm.pending_handler_k_handle = Some(py_k); + vm.segments.get_mut(root).unwrap().push_frame(Frame::EvalReturn( + Box::new(EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(py_k), + }), + )); }); let violations = vm.check_invariants().unwrap_err(); assert!(violations diff --git a/packages/doeff-vm-core/src/vm/step.rs b/packages/doeff-vm-core/src/vm/step.rs index e4cf69fa..264d4312 100644 --- a/packages/doeff-vm-core/src/vm/step.rs +++ b/packages/doeff-vm-core/src/vm/step.rs @@ -173,8 +173,30 @@ impl VM { } } _ => { - // Non-program frames can't handle errors — pop and propagate - seg.frames.pop(); + // Non-program frames can't handle errors — pop and propagate. + // + // Exception (#492): an ExpandReturn frame carrying a + // handler-dispatch k handle marks an in-flight deferred + // handler construction (the @do wrapper's + // Expand(Apply(Pure(thunk), [])) shape) whose evaluation + // raised — e.g. an arity TypeError from fn(*args) inside + // eval_apply. The handle is still owned here, so route the + // exception to the perform site's dynamic scope (recover the + // chain and discontinue it), exactly like the synchronous + // recovery arm in eval_perform. Dropping the handle instead + // would strand the perform-site chain; leaving it (the old + // VM-global slot design) let the NEXT Expand adopt it and + // deliver a later unrelated exception into the abandoned + // continuation. + let popped = seg.frames.pop(); + if let Some(Frame::EvalReturn(eval_return)) = popped { + if let EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } = *eval_return + { + return self.recover_from_k_handle(handle, error, error_context); + } + } continue_raise(error, error_context) } } @@ -190,23 +212,7 @@ impl VM { DoCtrl::Eval { expr } => continue_eval(*expr, error_context), - DoCtrl::Expand { expr } => { - // Evaluate inner, expect Value::Stream, push as frame - match *expr { - DoCtrl::Pure { value } => self.push_stream_value(value, error_context), - other => { - // Push ExpandReturn frame so we intercept the result - if let Some(seg_id) = self.current_segment { - if let Some(seg) = self.segments.get_mut(seg_id) { - seg.push_frame(Frame::EvalReturn(Box::new( - EvalReturnContinuation::ExpandReturn, - ))); - } - } - continue_eval(other, error_context) - } - } - } + DoCtrl::Expand { expr } => self.eval_expand(*expr, None, error_context), DoCtrl::Apply { f, args } => self.eval_apply(*f, args, error_context), @@ -385,12 +391,50 @@ impl VM { // Helpers // ------------------------------------------------------------------- + /// Evaluate an Expand's inner expr: expect Value::Stream, push as frame. + /// + /// `handler_k_handle` is Some only when the Expand is a handler-dispatch + /// result (the @do wrapper's `Expand(Apply(Pure(thunk), []))` shape from + /// eval_perform/eval_perform_with_k). The handle rides in the ExpandReturn + /// frame so that BOTH exits of the deferred evaluation consume it: the + /// value path moves it onto the resulting Program frame, and the raise + /// path (step_raise popping the ExpandReturn frame) discontinues the + /// perform-site chain with the exception (#492). + fn eval_expand( + &mut self, + expr: DoCtrl, + handler_k_handle: Option>, + error_context: Option>, + ) -> StepResult { + match expr { + DoCtrl::Pure { value } => { + self.push_stream_value(value, handler_k_handle, error_context) + } + other => { + // Push ExpandReturn frame so we intercept the result + if let Some(seg_id) = self.current_segment { + if let Some(seg) = self.segments.get_mut(seg_id) { + seg.push_frame(Frame::EvalReturn(Box::new( + EvalReturnContinuation::ExpandReturn { handler_k_handle }, + ))); + } + } + continue_eval(other, error_context) + } + } + } + /// Push a Value::Stream as a new Program frame on the current fiber. - fn push_stream_value(&mut self, value: Value, error_context: Option>) -> StepResult { - // Consume any k handle stashed by eval_perform/eval_perform_with_k - // so the resulting Program frame can recover the original perform-site - // chain (via PyK.take()) when its stream raises an uncaught exception. - let k_handle = self.pending_handler_k_handle.take(); + /// + /// `k_handle` is the handler-dispatch PyK handle (if any) so the resulting + /// Program frame can recover the original perform-site chain (via + /// PyK.take()) when its stream raises an uncaught exception. + fn push_stream_value( + &mut self, + value: Value, + k_handle: Option>, + error_context: Option>, + ) -> StepResult { match value { Value::Stream(stream) => { if let Some(seg_id) = self.current_segment { @@ -476,8 +520,22 @@ impl VM { } }; - // 1. Call ALL observers in the chain (synchronous, return value ignored) - self.call_all_observers(current, &effect); + // 1. Call ALL observers in the chain (synchronous, return value + // ignored). An observer exception FAILS FAST: it aborts the + // dispatch and propagates like a handler error — the effect is + // NOT delivered to any handler (#506). A dead tracing/audit + // layer must be loud, not silently dropped. + if let Err(err) = self.call_all_observers(current, &effect) { + return match err { + VMError::UncaughtException { exception } => { + // Python exception from the observer — raise it at the + // perform site so try/except around the yield sees it, + // exactly like a synchronous handler exception. + continue_raise(exception, error_context) + } + err => error_result(err, error_context), + }; + } // 2. Proceed to handler let result = match self.perform_effect(&effect) { @@ -494,49 +552,50 @@ impl VM { // (SPEC-VM-021). We keep a Py handle so that, if the // handler raises before consuming `k`, we can borrow the PyK, // take() the chain, and reattach it for exception propagation - // (OCaml 5 semantics). The handle is stashed in - // `pending_handler_k_handle` so `push_stream_value` can attach - // it to the resulting Program frame. - let k_value = pyo3::Python::attach(|py| { + // (OCaml 5 semantics). The handle is a dispatch-local: every + // arm below consumes or drops it, so it cannot outlive the + // dispatch that created it (#492). + let (k_value, handle) = pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .expect("failed to allocate PyK"); let handle = py_k.clone_ref(py); - self.pending_handler_k_handle = Some(handle); - Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())) + ( + Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())), + handle, + ) }); let outcome = handler_callable.call_handler(vec![effect, k_value]); - // If push_stream_value never ran (because outcome was Err or - // non-Expand), take the handle so it doesn't leak into a future - // handler call. - let leftover_handle = self.pending_handler_k_handle.take(); - match outcome { + Ok(DoCtrl::Expand { expr }) => { + // Deferred handler construction (the @do wrapper returns + // Expand(Apply(Pure(thunk), [])) before fn(*args) runs). + // Thread the handle through eval_expand so it lands on the + // ExpandReturn frame (or the Program frame for Pure) and + // is consumed on both the value and the raise path. + self.eval_expand(*expr, Some(handle), error_context) + } Ok(doctrl) => { - // Only restore the handle for Expand results (which will - // reach push_stream_value). For Pure/other results that - // bypass push_stream_value, dropping the handle prevents - // the stale-backup-leak bug (the chain stays in PyK and - // is freed when the Python K object is GC'd). - if matches!(doctrl, DoCtrl::Expand { .. }) { - self.pending_handler_k_handle = leftover_handle; - } + // Non-Expand result bypasses push_stream_value: drop the + // handle so it can't leak into a future frame (the + // stale-backup-leak bug). The chain stays in PyK and is + // freed when the Python K object is GC'd. + drop(handle); continue_eval(doctrl, error_context) } Err(VMError::UncaughtException { exception }) => { - // Synchronous Python exception from call_handler itself - // (the @do wrapper's Expand construction raised). Recover - // the chain from the PyK handle. - match leftover_handle { - Some(handle) => { - self.recover_from_k_handle(handle, exception, error_context) - } - None => continue_raise(exception, error_context), - } + // Synchronous Python exception from call_handler itself. + // The handle is still owned here, so route the exception + // to the perform site's dynamic scope: recover the chain + // from the PyK handle and raise into it (OCaml 5: + // discontinue k exn). If the handler already consumed k, + // the PyK is empty and the raise lands on current_segment + // (the outer scope). + self.recover_from_k_handle(handle, exception, error_context) } Err(err) => error_result(err, error_context), } @@ -589,7 +648,13 @@ impl VM { } /// Walk the entire chain and call all observers synchronously. - fn call_all_observers(&self, start: FiberId, effect: &Value) { + /// + /// Observer RETURN VALUES are ignored, but observer ERRORS are not: + /// the first failing observer aborts the walk and its error is + /// propagated by the caller like a handler error (fail-fast, #506). + /// Observers outside the failing one (further up the chain) are not + /// called for this effect. + fn call_all_observers(&self, start: FiberId, effect: &Value) -> Result<(), VMError> { let mut cursor = Some(start); while let Some(fid) = cursor { let Some(seg) = self.segments.get(fid) else { @@ -597,11 +662,12 @@ impl VM { }; if seg.is_intercept_boundary() { if let Some(observer) = seg.intercept_handler().cloned() { - let _ = observer.call(vec![effect.clone()]); + observer.call(vec![effect.clone()])?; } } cursor = seg.parent; } + Ok(()) } /// Evaluate WithObserve: install observer boundary, create body fiber, evaluate body. @@ -738,33 +804,32 @@ impl VM { if handler_callable.is_generator_handler() { // Generator handler path — see eval_perform for rationale. - let k_value = pyo3::Python::attach(|py| { + let (k_value, handle) = pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .expect("failed to allocate PyK"); let handle = py_k.clone_ref(py); - self.pending_handler_k_handle = Some(handle); - Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())) + ( + Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())), + handle, + ) }); let outcome = handler_callable.call_handler(vec![effect, k_value]); - let leftover_handle = self.pending_handler_k_handle.take(); match outcome { + Ok(DoCtrl::Expand { expr }) => { + self.eval_expand(*expr, Some(handle), error_context) + } Ok(doctrl) => { - if matches!(doctrl, DoCtrl::Expand { .. }) { - self.pending_handler_k_handle = leftover_handle; - } + drop(handle); continue_eval(doctrl, error_context) } - Err(VMError::UncaughtException { exception }) => match leftover_handle { - Some(handle) => { - self.recover_from_k_handle(handle, exception, error_context) - } - None => continue_raise(exception, error_context), - }, + Err(VMError::UncaughtException { exception }) => { + self.recover_from_k_handle(handle, exception, error_context) + } Err(err) => error_result(err, error_context), } } else { @@ -849,7 +914,9 @@ impl VM { } } EvalReturnContinuation::TailResumeReturn => continue_send(value, error_context), - EvalReturnContinuation::ExpandReturn => self.push_stream_value(value, error_context), + EvalReturnContinuation::ExpandReturn { handler_k_handle } => { + self.push_stream_value(value, handler_k_handle, error_context) + } _ => { // Other EvalReturn variants — TODO continue_send(value, error_context) diff --git a/packages/doeff-vm-core/src/vm_tests.rs b/packages/doeff-vm-core/src/vm_tests.rs index 673c3a61..feb03df0 100644 --- a/packages/doeff-vm-core/src/vm_tests.rs +++ b/packages/doeff-vm-core/src/vm_tests.rs @@ -1343,13 +1343,233 @@ mod tests { Err(err) => panic!("expected Ok, got error: {:?}", err), } - // Critical assertion: pending_handler_k_handle must be None. - // With the stale-backup bug, the handle would persist and - // attach to the next unrelated Program frame. + // Critical assertion: no dispatch k handle may survive anywhere in + // VM state. The VM-global pending_handler_k_handle slot was removed + // (#492) — the handle is dispatch-local and, for non-Expand results, + // dropped inside eval_perform — so the only places a handle can live + // are Program / ExpandReturn frames. Scan them all. assert!( - vm.pending_handler_k_handle.is_none(), - "stale-backup-leak: pending_handler_k_handle must be None \ + no_leaked_k_handle(&vm), + "stale-backup-leak: no frame may carry a handler k handle \ after a generator handler returns a non-Expand DoCtrl" ); } + + /// True if no live arena fiber carries a handler-dispatch k handle + /// (Program.handler_k_handle or ExpandReturn.handler_k_handle). + fn no_leaked_k_handle(vm: &VM) -> bool { + vm.segments.iter().all(|(_, fiber)| { + fiber.frames.iter().all(|frame| match frame { + Frame::Program { + handler_k_handle, .. + } => handler_k_handle.is_none(), + Frame::EvalReturn(cont) => !matches!( + cont.as_ref(), + crate::frame::EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(_), + } + ), + _ => true, + }) + }) + } + + // ----------------------------------------------------------------------- + // Test 14 (#492): a deferred handler-construction failure (the @do + // wrapper's Expand(Apply(Pure(thunk), [])) shape, where the thunk raises + // an arity TypeError inside eval_apply) must not leave a stale k handle + // behind. Pre-fix, the stale handle was adopted by the NEXT Expand frame, + // and a later unrelated exception was delivered INTO the abandoned + // perform-site continuation, whose return value was then substituted as + // the unrelated program's result. + // ----------------------------------------------------------------------- + + #[test] + fn test_deferred_expand_apply_error_routes_to_perform_site_and_leaves_no_stale_handle() { + use crate::continuation::PyK; + + /// Callable that raises when called — models fn(*args) blowing up + /// with an arity TypeError inside the @do wrapper's thunk. + #[derive(Debug)] + struct RaisingThunk; + + impl Callable for RaisingThunk { + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn call(&self, _args: Vec) -> Result { + Err(VMError::UncaughtException { + exception: Value::String("arity TypeError".into()), + }) + } + } + + /// Generator-like handler that returns the deferred-construction + /// shape WITHOUT touching k: Expand(Apply(Pure(RaisingThunk), [])). + /// The k stays inside the PyK the VM allocated — exactly the @do + /// wrong-arity situation from #492. + #[derive(Debug)] + struct DeferredCrashingHandler; + + impl Callable for DeferredCrashingHandler { + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn call(&self, _args: Vec) -> Result { + Err(VMError::internal("use call_handler")) + } + fn is_generator_handler(&self) -> bool { + true + } + fn call_handler(&self, args: Vec) -> Result { + // Sanity: k arrives as Opaque(PyK) on the generator path. + pyo3::Python::attach(|py| match &args[1] { + Value::Opaque(obj) => { + obj.bind(py).cast::().expect("expected PyK"); + } + other => panic!("expected Opaque(PyK), got {:?}", other), + }); + Ok(DoCtrl::Expand { + expr: Box::new(DoCtrl::Apply { + f: Box::new(DoCtrl::Pure { + value: Value::Callable(Arc::new(RaisingThunk) as CallableRef), + }), + args: vec![], + }), + }) + } + } + + /// Victim body: performs, and its throw() arm swallows ONLY the + /// unrelated crash (like the Python repro's `except RuntimeError`), + /// returning a recognizable substituted value. The construction + /// TypeError — correctly discontinued into the victim's chain — + /// propagates through. If the stale-handle bug is present, the later + /// unrelated crash lands HERE and Int(666) becomes the unrelated + /// program's "result". + #[derive(Debug)] + struct VictimStream { + state: u8, + } + + impl IRStream for VictimStream { + fn resume(&mut self, value: Value) -> StreamStep { + match self.state { + 0 => { + self.state = 1; + StreamStep::Instruction(DoCtrl::Perform { + effect: Value::String("ping".into()), + }) + } + _ => StreamStep::Done(value), + } + } + fn throw(&mut self, e: Value) -> StreamStep { + match &e { + Value::String(s) if s == "unrelated crash" => { + StreamStep::Done(Value::Int(666)) + } + _ => StreamStep::Error(e), + } + } + } + + /// Crasher body: raises an unrelated error, uncaught. + #[derive(Debug)] + struct CrasherStream; + + impl IRStream for CrasherStream { + fn resume(&mut self, _value: Value) -> StreamStep { + StreamStep::Error(Value::String("unrelated crash".into())) + } + fn throw(&mut self, e: Value) -> StreamStep { + StreamStep::Error(e) + } + } + + /// Main: installs the bad handler around the victim, CATCHES the + /// construction error (throw → continues), then runs the crasher as + /// an unrelated subprogram. The crasher's error must propagate out. + #[derive(Debug)] + struct MainStream { + state: u8, + h: Option, + b: Option>, + caught: Option, + } + + impl IRStream for MainStream { + fn resume(&mut self, value: Value) -> StreamStep { + match self.state { + 0 => { + self.state = 1; + StreamStep::Instruction(DoCtrl::WithHandler { + handler: self.h.take().unwrap(), + body: self.b.take().unwrap(), + }) + } + 1 => { + // WithHandler returned normally — the construction + // error was NOT raised. That would itself be a bug. + StreamStep::Error(Value::String( + "expected construction TypeError, got value".into(), + )) + } + 2 => { + self.state = 3; + StreamStep::Instruction(DoCtrl::Expand { + expr: Box::new(DoCtrl::Pure { + value: Value::Stream(IRStreamRef::new( + Box::new(CrasherStream) as Box, + )), + }), + }) + } + _ => StreamStep::Done(value), + } + } + fn throw(&mut self, e: Value) -> StreamStep { + if self.state == 1 { + // Caught the handler-construction error — proceed to the + // unrelated crasher, like the Python repro's try/except. + self.state = 2; + self.caught = Some(e); + self.resume(Value::Unit) + } else { + StreamStep::Error(e) + } + } + } + + let mut vm = setup_vm_with_stream(MainStream { + state: 0, + h: Some(Value::Callable( + Arc::new(DeferredCrashingHandler) as CallableRef, + )), + b: Some(expand_stream(VictimStream { state: 0 })), + caught: None, + }); + + let result = run_to_completion(&mut vm); + + // The unrelated crash must propagate out as an uncaught error — + // NOT be swallowed by the abandoned victim continuation (which + // would substitute Ok(Int(666)) here). + match result { + Err(VMError::UncaughtException { exception }) => match exception { + Value::String(s) => assert_eq!(s, "unrelated crash"), + other => panic!("expected String exception, got {:?}", other), + }, + Ok(other) => panic!( + "unrelated crash was swallowed; wrong-scope substituted result: {:?}", + other + ), + Err(err) => panic!("expected UncaughtException, got {:?}", err), + } + + assert!( + no_leaked_k_handle(&vm), + "stale k handle survived the deferred handler-construction failure" + ); + } } diff --git a/packages/doeff-vm/src/do_expr.rs b/packages/doeff-vm/src/do_expr.rs index cf08b0f3..5bd8def8 100644 --- a/packages/doeff-vm/src/do_expr.rs +++ b/packages/doeff-vm/src/do_expr.rs @@ -2,9 +2,29 @@ //! //! These replace the plain Python classes in `doeff/program.py`. //! The VM classifies them via `downcast` (not tag-based `getattr`). +//! +//! ## GC integration (#500) +//! +//! Every class that holds `Py` / `Py` fields implements +//! `__traverse__` so CPython's cycle collector can see through it — +//! without it, any reference cycle through a program node is permanently +//! uncollectable. `__clear__` is deliberately NOT implemented: these +//! classes are `frozen` (no `&mut self` access, required by the VM's +//! immutable-program invariant), so their field references cannot be +//! dropped in-place. That is sound for collection: field cycles cannot be +//! constructed among frozen nodes alone (fields are set once at +//! construction), so every real cycle routes through at least one mutable +//! Python object (instance `__dict__`, list, generator frame, ...) whose +//! `tp_clear` breaks the cycle once `__traverse__` has made it visible. +//! +//! Known limitation: the pyo3 `dict` slot (`#[pyclass(dict)]`, used by +//! defp for `__doeff_body__` metadata) is NOT reachable from +//! `__traverse__` in pyo3 0.28, so a cycle routed exclusively through a +//! program node's instance `__dict__` is still invisible to the GC. use doeff_vm_core::continuation::PyK; use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; /// Pure(value) — return a value immediately. #[pyclass(name = "Pure", frozen, dict, module = "doeff_vm.doeff_vm")] @@ -29,6 +49,10 @@ impl PyPure { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.value.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.value) + } } /// Perform(effect) — perform an effect (trigger handler lookup). @@ -54,6 +78,10 @@ impl PyPerform { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.effect.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.effect) + } } /// Resume(k, value) — resume continuation with value (non-tail, handler stays alive). @@ -78,6 +106,11 @@ impl PyResume { fn __repr__(&self) -> &'static str { "Resume(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.value) + } } /// Transfer(k, value) — resume continuation with value (tail, handler done). @@ -102,6 +135,11 @@ impl PyTransfer { fn __repr__(&self) -> &'static str { "Transfer(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.value) + } } /// Apply(f, args) — call f(args). @@ -128,6 +166,11 @@ impl PyApply { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.f.clone_ref(py), self.args.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.f)?; + visit.call(&self.args) + } } /// Expand(expr) — evaluate inner expr to Stream, then run it. @@ -152,6 +195,10 @@ impl PyExpand { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.expr.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.expr) + } } /// Pass(effect, k) — handler doesn't handle, forward to outer. @@ -176,6 +223,11 @@ impl PyPass { fn __repr__(&self) -> &'static str { "Pass(effect, k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.effect)?; + visit.call(&self.continuation) + } } /// WithHandler(handler, body) — install handler and run body under it. @@ -216,6 +268,11 @@ impl PyWithHandler { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.handler.clone_ref(py), self.body.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.handler)?; + visit.call(&self.body) + } } /// ResumeThrow(k, exception) — throw exception into continuation (non-tail). @@ -240,6 +297,11 @@ impl PyResumeThrow { fn __repr__(&self) -> &'static str { "ResumeThrow(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.exception) + } } /// TransferThrow(k, exception) — throw exception into continuation (tail). @@ -264,6 +326,11 @@ impl PyTransferThrow { fn __repr__(&self) -> &'static str { "TransferThrow(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.exception) + } } /// WithObserve(observer, body) — install observer and run body under it. @@ -290,6 +357,11 @@ impl PyWithObserve { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.observer.clone_ref(py), self.body.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.observer)?; + visit.call(&self.body) + } } /// GetTraceback(k) — query traceback from continuation without consuming it. @@ -309,6 +381,10 @@ impl PyGetTraceback { fn __repr__(&self) -> &'static str { "GetTraceback(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetExecutionContext() — get current execution context. @@ -354,6 +430,10 @@ impl PyGetHandlers { fn __repr__(&self) -> &'static str { "GetHandlers(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetBoundaries(k) — extract the interleaved handler/observer boundary @@ -380,6 +460,10 @@ impl PyGetBoundaries { fn __repr__(&self) -> &'static str { "GetBoundaries(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetOuterHandlers — extract handlers installed ABOVE the current handler. @@ -426,4 +510,8 @@ impl PyTailEval { let e = self.expr.bind(py).repr()?; Ok(format!("TailEval({})", e)) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.expr) + } } diff --git a/packages/doeff-vm/src/python_generator_stream.rs b/packages/doeff-vm/src/python_generator_stream.rs index fbfd34fc..6cecf081 100644 --- a/packages/doeff-vm/src/python_generator_stream.rs +++ b/packages/doeff-vm/src/python_generator_stream.rs @@ -7,6 +7,7 @@ use pyo3::exceptions::PyStopIteration; use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; use pyo3::types::PyString; use doeff_vm_core::do_ctrl::DoCtrl; @@ -18,7 +19,18 @@ use doeff_vm_core::value::Value; /// The Rust side uses `is_instance_of::()` for classification. /// /// Yielding an EffectBase from a generator is implicitly treated as Perform(effect). -#[pyclass(name = "EffectBase", subclass, dict, module = "doeff_vm.doeff_vm")] +/// +/// GC note (#500): EffectBase deliberately does NOT declare a pyo3 `dict` +/// slot. pyo3 0.28 cannot visit a pyclass dict slot from `__traverse__` +/// (the method only receives `&self`), so a base-owned `__dict__` would be +/// invisible to the cycle collector — any reference cycle through an +/// effect's attributes would be permanently uncollectable. Without a base +/// dict slot, Python subclasses (every real effect is a subclass) get a +/// CPython-managed `__dict__` which `subtype_traverse`/`subtype_clear` +/// handle natively, making cycles through effect attributes collectable. +/// Consequence: direct `EffectBase()` instances have no `__dict__` and +/// reject attribute assignment — effects must be subclasses (they are). +#[pyclass(name = "EffectBase", subclass, module = "doeff_vm.doeff_vm")] #[derive(Debug)] pub struct PyEffectBase; @@ -70,6 +82,17 @@ impl PythonCallable { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.callable.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.callable) + } + + fn __clear__(&mut self, py: Python<'_>) { + // Break cycles through the wrapped callable. The object is GC + // garbage at this point; any buggy post-clear use fails loudly + // ("'NoneType' object is not callable"). + self.callable = py.None(); + } } impl doeff_vm_core::value::Callable for PythonCallable { @@ -161,6 +184,16 @@ impl PyIRStream { tail_resume_lines: tail_resume_lines.unwrap_or_default(), } } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.generator) + } + + fn __clear__(&mut self, py: Python<'_>) { + // Break cycles through the wrapped generator. The object is GC + // garbage at this point; any buggy post-clear use fails loudly. + self.generator = py.None(); + } } /// A Python generator wrapped as an IRStream. diff --git a/packages/doeff-vm/src/result.rs b/packages/doeff-vm/src/result.rs index 0fc8780a..257925e4 100644 --- a/packages/doeff-vm/src/result.rs +++ b/packages/doeff-vm/src/result.rs @@ -1,4 +1,10 @@ use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; + +// GC note (#500): both result classes implement `__traverse__` so the cycle +// collector can see through them. They are `frozen`, so `__clear__` is not +// possible (no `&mut self`); see the module doc of `do_expr.rs` for why +// traverse-only is sound for frozen value holders. #[pyclass(frozen, name = "Ok", module = "doeff_vm.doeff_vm")] pub struct PyResultOk { @@ -43,6 +49,10 @@ impl PyResultOk { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.value.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.value) + } } #[pyclass(frozen, name = "Err", module = "doeff_vm.doeff_vm")] @@ -104,4 +114,9 @@ impl PyResultErr { ), )) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.error)?; + visit.call(&self.captured_traceback) + } } diff --git a/packages/doeff-vm/tests/test_memory_stats.py b/packages/doeff-vm/tests/test_memory_stats.py index f600b824..0ddd0935 100644 --- a/packages/doeff-vm/tests/test_memory_stats.py +++ b/packages/doeff-vm/tests/test_memory_stats.py @@ -64,3 +64,44 @@ def scenario(): ] assert doeff_vm.vm_live_counts() == before + + +def test_arena_slots_reclaimed_when_handler_abandons_continuation() -> None: + """#497: a handler that never resumes k (abort-style) drops the detached + chain; the chain's arena slots must return to the free list within the + run. Before reclamation every abort stranded ~2 vacant-reserved slots, + so head fiber indices grew ~2 per abort (max index ~2*N); with + reclamation the same few slots are reused and indices stay bounded. + """ + n_aborts = 200 + head_indices: list[int] = [] + + @do + def abort_handler(effect, k): + if isinstance(effect, SyntheticQuery): + head_indices.append(k.to_dict()["head"]) + return "aborted" + yield doeff_vm.Pass(effect, k) + + @do + def body(): + yield SyntheticQuery(key="x") + return "unreachable" + + @do + def scenario(): + result = None + for _ in range(n_aborts): + result = yield doeff_vm.WithHandler(abort_handler, body()) + return result + + before = doeff_vm.vm_live_counts() + + assert run(scenario()) == "aborted" + + assert doeff_vm.vm_live_counts() == before + assert len(head_indices) == n_aborts + assert max(head_indices) <= 8, ( + f"arena slots stranded: max head fiber index {max(head_indices)} " + f"after {n_aborts} aborted dispatches (expected bounded slot reuse)" + ) diff --git a/packages/doeff-vm/tests/test_pyvm.py b/packages/doeff-vm/tests/test_pyvm.py index e7a5054a..d36bf2fd 100644 --- a/packages/doeff-vm/tests/test_pyvm.py +++ b/packages/doeff-vm/tests/test_pyvm.py @@ -228,3 +228,98 @@ def test_vm_live_counts_return_to_baseline_after_run() -> None: assert run(simple_program()) == 42 assert doeff_vm.vm_live_counts() == before + + +class TestGcCycleCollection: + """Regression tests for #500: Py-holding pyclasses must implement the GC + protocol (__traverse__) so reference cycles through doeff_vm objects are + collectable. Before the fix, any cycle through Pure(...) or through an + EffectBase instance's attributes was permanently uncollectable. + """ + + @staticmethod + def _collect_and_check(ref) -> bool: + import gc + + for _ in range(3): + gc.collect() + return ref() is None + + def test_pure_python_control_cycle_is_collected(self) -> None: + """Sanity control: a pure-Python two-object cycle is collectable.""" + import weakref + + class Holder: + pass + + a, b = Holder(), Holder() + a.x = b + b.x = a + ref = weakref.ref(a) + del a, b + assert self._collect_and_check(ref) + + def test_cycle_through_pure_is_collected(self) -> None: + """o -> Pure(o) -> o must be collected (issue #500 repro).""" + import weakref + + class Holder: + pass + + o = Holder() + p = doeff_vm.Pure(o) + o.cycle = p + ref = weakref.ref(o) + del o, p + assert self._collect_and_check(ref) + + def test_cycle_through_effect_base_instance_is_collected(self) -> None: + """A cycle through an EffectBase subclass instance's attributes + must be collected (issue #500 repro, EffectBase variant).""" + import weakref + + class Holder: + pass + + class CycleEffect(doeff_vm.EffectBase): + pass + + e = CycleEffect() + o = Holder() + e.o = o + o.e = e + ref = weakref.ref(o) + del e, o + assert self._collect_and_check(ref) + + def test_cycle_between_two_effect_base_instances_is_collected(self) -> None: + """Two EffectBase instances referencing each other must be collected.""" + import weakref + + class EffA(doeff_vm.EffectBase): + pass + + class EffB(doeff_vm.EffectBase): + pass + + a, b = EffA(), EffB() + a.x = b + b.x = a + ref = weakref.ref(a) + del a, b + assert self._collect_and_check(ref) + + def test_cycle_through_with_handler_and_apply_is_collected(self) -> None: + """Cycles through composite nodes (WithHandler/Apply/Perform fields) + must be visible to the GC via __traverse__.""" + import weakref + + class Holder: + pass + + o = Holder() + node = doeff_vm.WithHandler(lambda _e, _k: None, doeff_vm.Apply(print, [o])) + o.cycle = node + ref = weakref.ref(o) + del o, node + assert self._collect_and_check(ref) diff --git a/tests/core/test_vm_fiber_ownership_g1.py b/tests/core/test_vm_fiber_ownership_g1.py index 4ea837b4..7bb665c8 100644 --- a/tests/core/test_vm_fiber_ownership_g1.py +++ b/tests/core/test_vm_fiber_ownership_g1.py @@ -57,6 +57,22 @@ def test_g1_arena_exposes_explicit_detach_attach_operations() -> None: assert "DetachedFiberChain" in source +def test_g1_slot_reclaim_carries_indices_not_fibers() -> None: + """#497: a detached chain dropped without reattachment reports its + arena slot indices through `SlotReclaimQueue` so the arena can reuse + them. That channel is allocator bookkeeping, not fiber ownership: it + must carry bare slot indices (usize) only — never fibers, chains, or + continuations. Fibers still move into Continuation ownership at detach + and drop normally with the chain (ISSUE-VM-001 G1 / SPEC-VM-021).""" + arena_source = _runtime_source(ARENA_RS) + body = _struct_body(arena_source, "SlotReclaimQueue") + + assert "Mutex>" in body + assert "Fiber" not in body + assert "Chain" not in body + assert "Continuation" not in body + + def test_g1_dispatch_and_step_do_not_construct_queue_backed_continuations() -> None: source = "\n".join( [ diff --git a/tests/test_handler_exception_catchable.py b/tests/test_handler_exception_catchable.py index 7ce1821b..39c6a826 100644 --- a/tests/test_handler_exception_catchable.py +++ b/tests/test_handler_exception_catchable.py @@ -17,8 +17,9 @@ from dataclasses import dataclass +import pytest from doeff_core_effects.handlers import try_handler -from doeff_vm import Err +from doeff_vm import Err, WithHandler from doeff import ( EffectBase, @@ -242,3 +243,85 @@ def program(): assert isinstance(result, Err), f"Expected Err, got {result!r}" assert isinstance(result.error, RuntimeError) assert "outer handler failed" in str(result.error) + + +# --------------------------------------------------------------------------- +# Tests (#492): deferred @do handler-construction failure must not leave a +# stale perform-site continuation behind. +# +# A wrong-arity @do handler matters because the @do wrapper returns +# Expand(Apply(Pure(thunk), [])) immediately; the arity TypeError only fires +# later, inside eval_apply, when the thunk calls fn(*args) — i.e. outside the +# synchronous dispatch recovery. Pre-fix, the perform-site k handle leaked +# into a VM-global slot, was adopted by the NEXT @do call's Expand frame, and +# a later unrelated exception was delivered INTO the long-abandoned victim +# continuation — whose return value was then silently substituted as the +# unrelated program's result (the 2026-07-07 exit-0 incident class). +# --------------------------------------------------------------------------- + + +@do +def _do_wrong_arity_handler(effect): # wrong arity: handlers take (effect, k) + yield None + + +def test_do_handler_wrong_arity_type_error_catchable(): + """The deferred arity TypeError routes to the perform site's dynamic scope + and is catchable around WithHandler — same as the plain-handler case.""" + + @do + def victim(): + yield Ping(label="x") + return "victim: no exception" + + @do + def program(): + try: + result = yield WithHandler(_do_wrong_arity_handler, victim()) + return result + except TypeError as e: + return f"caught: {e}" + + result = run(program()) + assert result.startswith("caught: "), f"Expected caught TypeError, got {result!r}" + assert "positional argument" in result + + +def test_stale_handle_not_adopted_by_next_program(): + """#492 repro: after catching the arity TypeError, a later unrelated + uncaught RuntimeError must propagate out of run() — NOT be delivered into + the abandoned victim continuation (which would substitute the victim's + return value as the crasher's result).""" + + caught_type_errors: list[TypeError] = [] + + @do + def victim(): + try: + yield Ping(label="x") + return "victim: no exception" + except RuntimeError as e: + return f"VICTIM CAUGHT WRONG-SCOPE EXCEPTION: {e}" + + @do + def crasher(): + if True: + raise RuntimeError("crash in unrelated subprogram") + yield # unreachable, makes it a generator + + @do + def main(): + try: + yield WithHandler(_do_wrong_arity_handler, victim()) + except TypeError as e: + caught_type_errors.append(e) + r = yield crasher() + return ("crasher returned", r) + + with pytest.raises(RuntimeError, match="crash in unrelated subprogram"): + run(main()) + + assert len(caught_type_errors) == 1, ( + "the deferred handler-construction arity TypeError must surface " + f"exactly once at the dispatch's dynamic scope, got {caught_type_errors!r}" + ) diff --git a/tests/test_with_observe_visibility.py b/tests/test_with_observe_visibility.py index 670805d8..a9e97884 100644 --- a/tests/test_with_observe_visibility.py +++ b/tests/test_with_observe_visibility.py @@ -91,3 +91,57 @@ def prog(): tell_msgs = [o for o in observed if "tell from handler" in o] assert slog_msgs, f"Observer missed handler body slog. Observed: {observed}" assert tell_msgs, f"Observer missed handler body Tell. Observed: {observed}" + + +def test_observer_exception_fails_fast(): + """Regression test for #506: an exception raised by an observer must + abort the dispatch and propagate like a handler error — run() raises + instead of returning the handled value. A dead tracing/audit layer must + be loud, not silently swallowed. + """ + import pytest + + class Ping(EffectBase): + pass + + def exploding_observer(effect): + raise RuntimeError("observer exploded") + + @do + def h(effect, k): + return (yield Resume(k, "handled")) + + @do + def body(): + return (yield Ping()) + + program = WithObserve(VMCallable(exploding_observer), _program_handler(h)(body())) + with pytest.raises(RuntimeError, match="observer exploded"): + run(program) + + +def test_observer_exception_is_catchable_at_perform_site(): + """#506: the observer exception is raised at the perform site, so a + try/except around the yield can catch it (same semantics as a + synchronous handler exception).""" + + class Ping(EffectBase): + pass + + def exploding_observer(effect): + raise RuntimeError("observer exploded") + + @do + def h(effect, k): + return (yield Resume(k, "handled")) + + @do + def body(): + try: + yield Ping() + except RuntimeError as exc: + return f"caught: {exc}" + return "not raised" + + program = WithObserve(VMCallable(exploding_observer), _program_handler(h)(body())) + assert run(program) == "caught: observer exploded"