From 067d2084b940d53aa8edc59397bb1f51a3f5ee98 Mon Sep 17 00:00:00 2001
From: Yehor Smoliakov <yehors@ukr.net>
Date: Sun, 11 Oct 2026 02:02:41 -0700
Subject: [PATCH 1/4] test: reproduce cancelled timer retention and measure
 PostgreSQL caches

---
 .github/workflows/async-ecosystem.yml |   7 +
 benches/timer_cancellation.py         | 227 ++++++++++++++++++++++++
 tests/ecosystem/test_postgresql.py    |  45 +++++
 tests/test_timer_cancellation.py      | 246 ++++++++++++++++++++++++++
 4 files changed, 525 insertions(+)
 create mode 100644 benches/timer_cancellation.py
 create mode 100644 tests/test_timer_cancellation.py

diff --git a/.github/workflows/async-ecosystem.yml b/.github/workflows/async-ecosystem.yml
index d7908b7..be3b996 100644
--- a/.github/workflows/async-ecosystem.yml
+++ b/.github/workflows/async-ecosystem.yml
@@ -112,5 +112,12 @@ jobs:
         run: |
           uv run --no-sync python -c "import asyncpg, sqlalchemy, uvloop, platform; from importlib.metadata import version; print(platform.platform(), platform.python_version()); print({p: version(p) for p in ('asyncpg', 'sqlalchemy', 'uvloop')})"
           docker exec ${{ job.services.postgres.id }} postgres --version
+      - name: Measure cancelled timers and closed statement caches
+        run: |
+          for event_loop in asyncio uvloop rsloop; do
+            uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case timers
+            uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case timeout
+            uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case postgres --count 500
+          done
       - name: Compare PostgreSQL lifecycles
         run: uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem -ra
diff --git a/benches/timer_cancellation.py b/benches/timer_cancellation.py
new file mode 100644
index 0000000..f6af0c1
--- /dev/null
+++ b/benches/timer_cancellation.py
@@ -0,0 +1,227 @@
+"""Fresh-process lifetime/RSS and timer throughput probe for issue #118.
+
+Run with the same release build and interpreter for before/after throughput.
+PostgreSQL requires RSLOOP_POSTGRES_DSN pointing to an isolated database.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import gc
+import importlib
+import json
+import os
+import platform
+import statistics
+import sys
+import time
+import tracemalloc
+import weakref
+from importlib.metadata import version
+from pathlib import Path
+
+
+class Payload:
+    def __init__(self):
+        self.blob = bytearray(4096)
+
+    def fire(self):
+        raise AssertionError("cancelled timer fired")
+
+
+def rss_bytes():
+    if sys.platform == "linux":
+        for line in Path("/proc/self/status").read_text().splitlines():
+            if line.startswith("VmRSS:"):
+                return int(line.split()[1]) * 1024
+    return None
+
+
+def live(references):
+    gc.collect()
+    return sum(reference() is not None for reference in references)
+
+
+async def memory_case(count, mode):
+    loop = asyncio.get_running_loop()
+    sentinel = loop.call_later(1800, lambda: None)
+    references = []
+    gc.collect()
+    rss_before = rss_bytes()
+    tracemalloc.start()
+    before = tracemalloc.get_traced_memory()[0]
+    started = time.perf_counter()
+    try:
+        if mode == "timers":
+            payload = None
+            for _ in range(count):
+                payload = Payload()
+                references.append(weakref.ref(payload))
+                loop.call_later(3600, payload.fire).cancel()
+            del payload
+        else:
+
+            async def request():
+                async with asyncio.timeout(3600):
+                    await asyncio.sleep(0)
+                return Payload()
+
+            for offset in range(0, count, 500):
+                tasks = [
+                    loop.create_task(request()) for _ in range(min(500, count - offset))
+                ]
+                await asyncio.gather(*tasks)
+                references.extend(weakref.ref(task) for task in tasks)
+                del tasks
+        immediate = live(references)
+        held_immediate = tracemalloc.get_traced_memory()[0] - before
+        rss_immediate = rss_bytes()
+        for _ in range(3):
+            await asyncio.sleep(0)
+        after_turn = live(references)
+        held_after_turn = tracemalloc.get_traced_memory()[0] - before
+        # Subtract only this probe's weakref/list storage, as in the issue.
+        probe_bytes = sys.getsizeof(references) + sum(
+            sys.getsizeof(ref) for ref in references
+        )
+        return {
+            "count": count,
+            "alive_immediately": immediate,
+            "alive_after_turn": after_turn,
+            "python_bytes_immediately": held_immediate - probe_bytes,
+            "python_bytes_after_turn": held_after_turn - probe_bytes,
+            "traced_peak_bytes": tracemalloc.get_traced_memory()[1],
+            "rss_before": rss_before,
+            "rss_immediately": rss_immediate,
+            "rss_after_turn": rss_bytes(),
+            "seconds": time.perf_counter() - started,
+        }
+    finally:
+        tracemalloc.stop()
+        sentinel.cancel()
+
+
+def postgres_counts():
+    gc.collect()
+    counts = dict.fromkeys(("_StatementCache", "PreparedStatementState", "Protocol"), 0)
+    for obj in gc.get_objects():
+        name = type(obj).__name__
+        if name in counts:
+            counts[name] += 1
+    return counts
+
+
+async def postgres_case(count):
+    asyncpg = importlib.import_module("asyncpg")
+
+    loop = asyncio.get_running_loop()
+    sentinel = loop.call_later(1800, lambda: None)
+    before = postgres_counts()
+    rss_before = rss_bytes()
+    started = time.perf_counter()
+
+    async def connection():
+        conn = await asyncpg.connect(
+            os.environ["RSLOOP_POSTGRES_DSN"], max_cached_statement_lifetime=3600
+        )
+        try:
+            for index in range(3):
+                async with asyncio.timeout(3600):
+                    assert (
+                        await conn.fetchval(f"SELECT $1::int + {index}", index)
+                        == 2 * index
+                    )
+        finally:
+            await conn.close()
+
+    try:
+        for offset in range(0, count, 10):
+            await asyncio.gather(
+                *(connection() for _ in range(min(10, count - offset)))
+            )
+        for _ in range(3):
+            await asyncio.sleep(0)
+        after = postgres_counts()
+        return {
+            "count": count,
+            "seconds": time.perf_counter() - started,
+            "live_object_delta": {name: after[name] - before[name] for name in before},
+            "rss_before": rss_before,
+            "rss_after": rss_bytes(),
+            "asyncpg": version("asyncpg"),
+        }
+    finally:
+        sentinel.cancel()
+
+
+async def performance_case(count):
+    # Reuse the repository's workloads, without tracemalloc or GC in timed spans.
+    from scheduler_workloads import bench_timers
+
+    samples = {}
+    for mode in (
+        "timers_retained",
+        "timers_discarded",
+        "timers_cancelled",
+        "timers_mixed",
+    ):
+        await bench_timers("probe", 1000, 1000, mode)
+        samples[mode] = [
+            (await bench_timers("probe", count, 1000, mode)).seconds for _ in range(7)
+        ]
+    return {
+        "count": count,
+        "samples_seconds": samples,
+        "median_seconds": {
+            mode: statistics.median(values) for mode, values in samples.items()
+        },
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--loop", choices=("asyncio", "rsloop", "uvloop"), required=True
+    )
+    parser.add_argument(
+        "--case",
+        choices=("timers", "timeout", "postgres", "performance"),
+        required=True,
+    )
+    parser.add_argument("--count", type=int, default=20000)
+    args = parser.parse_args()
+    if args.count <= 0:
+        parser.error("--count must be positive")
+    if args.loop == "asyncio":
+        factory = asyncio.new_event_loop
+    else:
+        factory = __import__(args.loop).new_event_loop
+    loop = factory()
+    try:
+        if args.case == "postgres":
+            operation = postgres_case(args.count)
+        elif args.case == "performance":
+            operation = performance_case(args.count)
+        else:
+            operation = memory_case(args.count, args.case)
+        result = loop.run_until_complete(operation)
+    finally:
+        loop.close()
+    print(
+        json.dumps(
+            dict(
+                loop=args.loop,
+                case=args.case,
+                python=platform.python_version(),
+                platform=platform.platform(),
+                loop_version=version(args.loop) if args.loop != "asyncio" else None,
+                **result,
+            ),
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/ecosystem/test_postgresql.py b/tests/ecosystem/test_postgresql.py
index 092e890..ba40425 100644
--- a/tests/ecosystem/test_postgresql.py
+++ b/tests/ecosystem/test_postgresql.py
@@ -1,6 +1,7 @@
 """Real PostgreSQL contracts; opt in with a dedicated RSLOOP_POSTGRES_DSN."""
 
 import asyncio
+import gc
 import os
 import uuid
 from contextlib import asynccontextmanager
@@ -255,3 +256,47 @@ async def main():
             await observer.close()
 
     loop.run_until_complete(asyncio.wait_for(main(), 20))
+
+
+def test_closed_connections_release_statement_cache_before_expiry(loop, postgres_dsn):
+    """A live earlier timer must not pin cache expiry callbacks after close (#118)."""
+    names = {"_StatementCache", "PreparedStatementState", "Protocol"}
+
+    def counts():
+        gc.collect()
+        result = dict.fromkeys(names, 0)
+        for obj in gc.get_objects():
+            name = type(obj).__name__
+            if name in result:
+                result[name] += 1
+        return result
+
+    async def main():
+        sentinel = loop.call_later(1800, lambda: None)
+        before = counts()
+
+        async def connection():
+            conn = await asyncpg.connect(
+                postgres_dsn, max_cached_statement_lifetime=3600
+            )
+            try:
+                for index in range(3):
+                    assert (
+                        await conn.fetchval(f"SELECT $1::int + {index}", index)
+                        == 2 * index
+                    )
+            finally:
+                await conn.close()
+
+        try:
+            for _ in range(5):
+                await asyncio.gather(*(connection() for _ in range(10)))
+            for _ in range(3):
+                await asyncio.sleep(0)
+            after = counts()
+            delta = {name: after[name] - before[name] for name in names}
+            assert all(value <= 0 for value in delta.values()), delta
+        finally:
+            sentinel.cancel()
+
+    loop.run_until_complete(asyncio.wait_for(main(), 30))
diff --git a/tests/test_timer_cancellation.py b/tests/test_timer_cancellation.py
new file mode 100644
index 0000000..452a954
--- /dev/null
+++ b/tests/test_timer_cancellation.py
@@ -0,0 +1,246 @@
+"""Cancelled timers release Python ownership before their heap deadline (#118)."""
+
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import gc
+import threading
+import weakref
+
+import pytest
+import rsloop
+
+
+@pytest.fixture(params=["asyncio", "rsloop", "uvloop"])
+def loop(request):
+    factory = {
+        "asyncio": asyncio.new_event_loop,
+        "rsloop": rsloop.new_event_loop,
+    }.get(request.param)
+    if factory is None:
+        factory = pytest.importorskip("uvloop").new_event_loop
+    event_loop = factory()
+    yield event_loop
+    event_loop.close()
+
+
+class Payload:
+    def __init__(self):
+        self.blob = bytearray(4096)
+
+    def __call__(self, *args):
+        pytest.fail("cancelled callback executed")
+
+
+def make_timer(loop, ownership, many=False):
+    payload = Payload()
+    reference = weakref.ref(payload)
+    if ownership == "callback":
+        handle = loop.call_later(3600, payload)
+    elif ownership == "args":
+        args = (payload, payload) if many else (payload,)
+        handle = loop.call_later(3600, lambda *args: None, *args)
+    else:
+        var = contextvars.ContextVar("timer_payload")
+        context = contextvars.Context()
+        context.run(var.set, payload)
+        handle = loop.call_later(3600, lambda: None, context=context)
+    return handle, reference
+
+
+@pytest.mark.parametrize("ownership", ["callback", "args", "context"])
+@pytest.mark.parametrize("running", [False, True])
+def test_cancel_releases_python_references_immediately(loop, ownership, running):
+    def check():
+        sentinel = loop.call_later(1800, lambda: None)
+        try:
+            handle, reference = make_timer(loop, ownership)
+            assert reference() is not None
+            handle.cancel()
+            handle.cancel()
+            # No GC or loop turn is needed for these acyclic payloads.
+            if ownership == "context" and not isinstance(loop, rsloop.Loop):
+                # CPython and uvloop retain Context on the user-held handle.
+                assert reference() is not None
+            else:
+                assert reference() is None
+            assert handle.cancelled()
+        finally:
+            sentinel.cancel()
+
+    async def main():
+        check()
+
+    if running:
+        loop.run_until_complete(main())
+    else:
+        check()
+
+
+def test_cancel_releases_many_arguments_and_captured_context(loop):
+    sentinel = loop.call_later(1800, lambda: None)
+    try:
+        handle, reference = make_timer(loop, "args", many=True)
+        handle.cancel()
+        assert reference() is None
+        var = contextvars.ContextVar("implicit_timer_payload")
+        payload = Payload()
+        reference = weakref.ref(payload)
+        token = var.set(payload)
+        handle = loop.call_later(3600, lambda: None)
+        var.reset(token)
+        del payload
+        handle.cancel()
+        if isinstance(loop, rsloop.Loop):
+            assert reference() is None
+        else:
+            assert reference() is not None
+    finally:
+        sentinel.cancel()
+
+
+def test_cancelled_burst_behind_live_timer_releases_all_payloads(loop):
+    async def main():
+        sentinel = loop.call_later(1800, lambda: None)
+        try:
+            references = []
+            for _ in range(1000):
+                handle, reference = make_timer(loop, "callback")
+                references.append(reference)
+                handle.cancel()
+            await asyncio.sleep(0)
+            gc.collect()
+            assert all(reference() is None for reference in references)
+        finally:
+            sentinel.cancel()
+
+    loop.run_until_complete(main())
+
+
+@pytest.mark.parametrize("mode", ["timeout", "wait_for"])
+def test_completed_timeout_tasks_and_results_are_released(loop, mode):
+    if mode == "timeout" and not hasattr(asyncio, "timeout"):
+        pytest.skip("asyncio.timeout requires Python 3.11")
+
+    async def request():
+        if mode == "timeout":
+            async with asyncio.timeout(3600):
+                await asyncio.sleep(0)
+        else:
+            await asyncio.wait_for(asyncio.sleep(0), 3600)
+        return Payload()
+
+    async def main():
+        sentinel = loop.call_later(1800, lambda: None)
+        try:
+            tasks = [loop.create_task(request()) for _ in range(100)]
+            await asyncio.gather(*tasks)
+            references = [weakref.ref(task) for task in tasks]
+            references.extend(weakref.ref(task.result()) for task in tasks)
+            del tasks
+            # Release gather's completion callbacks; no deadline may expire.
+            for _ in range(3):
+                await asyncio.sleep(0)
+            gc.collect()
+            assert all(reference() is None for reference in references)
+        finally:
+            sentinel.cancel()
+
+    loop.run_until_complete(main())
+
+
+def test_cancel_finalizers_can_reenter_and_close_loop(loop):
+    if not isinstance(loop, rsloop.Loop):
+        pytest.skip("reentrant close is a separate rsloop lifecycle contract")
+    events = []
+    handles = []
+
+    class Callback:
+        def __call__(self):
+            pytest.fail("cancelled callback executed")
+
+        def __del__(self):
+            events.append("released")
+            handles[0].cancel()
+            if not loop.is_closed():
+                loop.call_later(0, lambda: None).cancel()
+                loop.close()
+
+    handles.append(loop.call_later(3600, Callback()))
+    handles[0].cancel()
+    released_immediately = events == ["released"]
+    if not released_immediately:
+        loop.close()
+    assert released_immediately
+    assert handles[0].cancelled()
+    assert loop.is_closed()
+
+
+def test_cancel_after_dequeue_skips_callback(loop):
+    events = []
+    when = loop.time()
+    # call_later's deadlines give the cancellation callback a strictly earlier
+    # position on loops without stable ordering for equal deadlines.
+    loop.call_at(when - 1, lambda: handle.cancel())
+    handle = loop.call_at(when, lambda: events.append("unexpected"))
+    loop.run_until_complete(asyncio.sleep(0))
+    assert events == []
+
+
+def test_concurrent_repeated_cancel_releases_payload_once(loop):
+    handle, reference = make_timer(loop, "callback")
+    barrier = threading.Barrier(9)
+
+    def cancel():
+        barrier.wait(timeout=5)
+        for _ in range(100):
+            handle.cancel()
+
+    threads = [threading.Thread(target=cancel) for _ in range(8)]
+    for thread in threads:
+        thread.start()
+    barrier.wait(timeout=5)
+    for thread in threads:
+        thread.join(timeout=5)
+        assert not thread.is_alive()
+    assert handle.cancelled()
+    assert reference() is None
+
+
+def test_cancel_running_callback_preserves_arguments_and_context(loop):
+    entered = threading.Event()
+    cancelled = threading.Event()
+    var = contextvars.ContextVar("running_timer_value")
+    context = contextvars.Context()
+    context.run(var.set, "captured")
+    events = []
+
+    def callback(payload):
+        entered.set()
+        assert cancelled.wait(timeout=5)
+        assert payload.blob == bytearray(4096)
+        assert var.get() == "captured"
+        events.append("finished")
+
+    payload = Payload()
+    reference = weakref.ref(payload)
+    handle = loop.call_later(0, callback, payload, context=context)
+    del payload
+
+    def cancel():
+        if entered.wait(timeout=5):
+            handle.cancel()
+        cancelled.set()
+
+    thread = threading.Thread(target=cancel)
+    thread.start()
+    try:
+        loop.run_until_complete(asyncio.sleep(0))
+        assert events == ["finished"]
+        gc.collect()
+        assert reference() is None
+    finally:
+        cancelled.set()
+        thread.join(timeout=5)
+        assert not thread.is_alive()

From dd9698df02d274795333c0f53a3c13cafa3c40a4 Mon Sep 17 00:00:00 2001
From: Yehor Smoliakov <yehors@ukr.net>
Date: Sun, 11 Oct 2026 02:08:31 -0700
Subject: [PATCH 2/4] fix: detach cancelled timer payloads and compact queued
 tombstones

---
 benches/timer_cancellation.py    |  16 +++
 src/engine/callbacks.rs          | 193 ++++++++++++++++++++++++++++++-
 src/engine/commands.rs           |   3 +-
 src/engine/loop_core.rs          |  92 +++++++++++++--
 src/engine/timer_entry.rs        |  81 +++++++++++--
 tests/test_timer_cancellation.py |  69 ++++++++++-
 6 files changed, 423 insertions(+), 31 deletions(-)

diff --git a/benches/timer_cancellation.py b/benches/timer_cancellation.py
index f6af0c1..c7c676d 100644
--- a/benches/timer_cancellation.py
+++ b/benches/timer_cancellation.py
@@ -157,6 +157,7 @@ async def connection():
 
 async def performance_case(count):
     # Reuse the repository's workloads, without tracemalloc or GC in timed spans.
+    from compare_event_loops import bench_callbacks
     from scheduler_workloads import bench_timers
 
     samples = {}
@@ -170,6 +171,10 @@ async def performance_case(count):
         samples[mode] = [
             (await bench_timers("probe", count, 1000, mode)).seconds for _ in range(7)
         ]
+    await bench_callbacks("probe", 1000)
+    samples["call_soon"] = [
+        (await bench_callbacks("probe", count)).seconds for _ in range(7)
+    ]
     return {
         "count": count,
         "samples_seconds": samples,
@@ -191,6 +196,17 @@ def main():
     )
     parser.add_argument("--count", type=int, default=20000)
     args = parser.parse_args()
+    if args.case == "timeout" and not hasattr(asyncio, "timeout"):
+        print(
+            json.dumps(
+                {
+                    "loop": args.loop,
+                    "case": args.case,
+                    "skipped": "asyncio.timeout requires Python 3.11",
+                }
+            )
+        )
+        return
     if args.count <= 0:
         parser.error("--count must be positive")
     if args.loop == "asyncio":
diff --git a/src/engine/callbacks.rs b/src/engine/callbacks.rs
index fafce30..43df01d 100644
--- a/src/engine/callbacks.rs
+++ b/src/engine/callbacks.rs
@@ -1,8 +1,8 @@
 //! Callback values and Python handle wrappers used by the loop engine.
 
 use std::sync::{
-    Arc, Weak,
-    atomic::{AtomicBool, Ordering},
+    Arc, Mutex, Weak,
+    atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 
 use pyo3::{prelude::*, types::PyTuple};
@@ -341,6 +341,103 @@ impl PyHandle {
     }
 }
 
+/// Mutable ownership exists only for timers: ordinary ready callbacks keep
+/// their allocation size and lock-free invocation path.
+pub struct TimerCallback {
+    id: CallbackId,
+    state: Mutex<TimerState>,
+    cancelled: AtomicBool,
+    cancelled_timers: Arc<AtomicUsize>,
+}
+
+struct TimerState {
+    ready: Option<ReadyCallback>,
+    queued: bool,
+}
+
+impl TimerCallback {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub fn new(ready: ReadyCallback, cancelled_timers: Arc<AtomicUsize>) -> Self {
+        Self {
+            id: ready.id(),
+            state: Mutex::new(TimerState {
+                ready: Some(ready),
+                queued: true,
+            }),
+            cancelled: AtomicBool::new(false),
+            cancelled_timers,
+        }
+    }
+
+    #[inline]
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub fn id(&self) -> CallbackId {
+        self.id
+    }
+
+    #[inline]
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub fn cancelled(&self) -> bool {
+        self.cancelled.load(Ordering::Relaxed)
+    }
+
+    /// Linearize cancellation and detach Python ownership under the same lock
+    /// as invocation. Drop outside it: DECREF can run reentrant finalizers.
+    /// The Python handle calls this while attached; Rust callers can also use
+    /// PyO3's deferred decref when detached.
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub fn cancel(&self) {
+        let ready = {
+            let mut state = self.state.lock().expect("poisoned timer state");
+            if self.cancelled.swap(true, Ordering::Relaxed) {
+                return;
+            }
+            if state.queued {
+                self.cancelled_timers.fetch_add(1, Ordering::Relaxed);
+            }
+            state.ready.take()
+        };
+        drop(ready);
+    }
+
+    /// Queue removal and cancellation share the lock so repeated cancellation,
+    /// cross-thread removal and loop shutdown count each tombstone once.
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub(super) fn dequeue(&self) {
+        let mut state = self.state.lock().expect("poisoned timer state");
+        if state.queued && self.cancelled() {
+            self.cancelled_timers.fetch_sub(1, Ordering::Relaxed);
+        }
+        state.queued = false;
+    }
+
+    /// The invocation owns its payload until completion. A concurrent cancel
+    /// can prevent a not-yet-claimed invocation, but cannot invalidate a
+    /// running callback's arguments/context. No lock crosses Python
+    /// execution.
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))]
+    pub(super) fn take_ready(&self) -> Option<ReadyCallback> {
+        self.state
+            .lock()
+            .expect("poisoned timer state")
+            .ready
+            .take()
+    }
+}
+
+impl Drop for TimerCallback {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<TimerCallback as Drop>")
+    )]
+    fn drop(&mut self) {
+        let state = self.state.get_mut().expect("poisoned timer state");
+        if state.queued && *self.cancelled.get_mut() {
+            self.cancelled_timers.fetch_sub(1, Ordering::Relaxed);
+        }
+    }
+}
+
 /// Python's cancellable `asyncio.TimerHandle` equivalent.
 ///
 /// The weak callback reference lets cancellation update a live scheduled
@@ -355,7 +452,7 @@ impl PyHandle {
 pub struct PyTimerHandle {
     callback_id: CallbackId,
     when: f64,
-    callback: Weak<ReadyCallback>,
+    callback: Weak<TimerCallback>,
     cancelled: AtomicBool,
 }
 
@@ -363,7 +460,7 @@ impl PyTimerHandle {
     #[inline]
     /// Creates a timer handle for a callback scheduled at loop time `when`.
     #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PyTimerHandle"))]
-    pub fn new(callback_id: CallbackId, when: f64, callback: &Arc<ReadyCallback>) -> Self {
+    pub fn new(callback_id: CallbackId, when: f64, callback: &Arc<TimerCallback>) -> Self {
         Self {
             callback_id,
             when,
@@ -518,11 +615,92 @@ mod tests {
         });
     }
 
+    #[test]
+    fn timer_cancellation_racing_queue_removal_counts_once() {
+        crate::initialize_python_for_tests();
+        Python::attach(|py| {
+            let count = Arc::new(AtomicUsize::new(0));
+            for id in 0..128 {
+                let timer = TimerCallback::new(
+                    callback_with_args(py, id, PyTuple::empty(py).unbind()),
+                    Arc::clone(&count),
+                );
+                std::thread::scope(|scope| {
+                    scope.spawn(|| {
+                        timer.cancel();
+                        timer.cancel();
+                    });
+                    scope.spawn(|| {
+                        timer.dequeue();
+                        timer.dequeue();
+                    });
+                });
+                assert!(timer.cancelled());
+                assert!(timer.take_ready().is_none());
+                assert_eq!(count.load(Ordering::Relaxed), 0);
+            }
+        });
+    }
+
+    #[test]
+    fn timer_claim_racing_cancel_never_invokes_twice_or_loses_ownership() {
+        crate::initialize_python_for_tests();
+        Python::attach(|py| {
+            for id in 0..128 {
+                let count = Arc::new(AtomicUsize::new(0));
+                let timer = TimerCallback::new(
+                    callback_with_args(py, id, PyTuple::empty(py).unbind()),
+                    Arc::clone(&count),
+                );
+                timer.dequeue();
+                let claimed = std::thread::scope(|scope| {
+                    scope.spawn(|| timer.cancel());
+                    scope
+                        .spawn(|| timer.take_ready())
+                        .join()
+                        .expect("claim thread")
+                });
+                if let Some(callback) = claimed {
+                    assert_eq!(
+                        callback
+                            .invoke(py)
+                            .expect("claimed invocation")
+                            .extract::<Vec<i32>>(py)
+                            .expect("result"),
+                        Vec::<i32>::new()
+                    );
+                }
+                assert!(timer.take_ready().is_none());
+                assert_eq!(count.load(Ordering::Relaxed), 0);
+            }
+        });
+    }
+
+    #[test]
+    fn dropping_queued_cancelled_timer_restores_count() {
+        crate::initialize_python_for_tests();
+        Python::attach(|py| {
+            let count = Arc::new(AtomicUsize::new(0));
+            let timer = TimerCallback::new(
+                callback_with_args(py, 1, PyTuple::empty(py).unbind()),
+                Arc::clone(&count),
+            );
+            timer.cancel();
+            timer.cancel();
+            assert_eq!(count.load(Ordering::Relaxed), 1);
+            drop(timer);
+            assert_eq!(count.load(Ordering::Relaxed), 0);
+        });
+    }
+
     #[test]
     fn timer_handle_cancels_live_callback_and_tolerates_a_dropped_one() {
         crate::initialize_python_for_tests();
         Python::attach(|py| {
-            let ready = Arc::new(callback_with_args(py, 12, PyTuple::empty(py).unbind()));
+            let ready = Arc::new(TimerCallback::new(
+                callback_with_args(py, 12, PyTuple::empty(py).unbind()),
+                Arc::new(AtomicUsize::new(0)),
+            ));
             let timer = PyTimerHandle::new(12, 1.25, &ready);
             assert!((timer.when() - 1.25).abs() < f64::EPSILON);
             assert!(!timer.cancelled());
@@ -535,7 +713,10 @@ mod tests {
                 "<TimerHandle id=12 when=1.250000 cancelled=true>"
             );
 
-            let dropped = Arc::new(callback_with_args(py, 13, PyTuple::empty(py).unbind()));
+            let dropped = Arc::new(TimerCallback::new(
+                callback_with_args(py, 13, PyTuple::empty(py).unbind()),
+                Arc::new(AtomicUsize::new(0)),
+            ));
             let dropped_timer = PyTimerHandle::new(13, 2.0, &dropped);
             drop(dropped);
             dropped_timer
diff --git a/src/engine/commands.rs b/src/engine/commands.rs
index 6ffb0bc..945fb0e 100644
--- a/src/engine/commands.rs
+++ b/src/engine/commands.rs
@@ -7,7 +7,7 @@ use std::{
 
 use pyo3::prelude::*;
 
-use super::callbacks::{PyHandle, ReadyCallback};
+use super::callbacks::{PyHandle, ReadyCallback, TimerCallback};
 use crate::{
     fd_ops::RawFd,
     transport::{
@@ -23,6 +23,7 @@ use crate::{
 /// GIL.
 pub enum ReadyItem {
     Callback(Arc<ReadyCallback>),
+    TimerCallback(Arc<TimerCallback>),
     HandleCallback(Py<PyHandle>),
     FutureSetResult(Box<FutureCompletion>),
     FutureSetException(Box<FutureCompletion>),
diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs
index 5d3a671..eef4fe0 100644
--- a/src/engine/loop_core.rs
+++ b/src/engine/loop_core.rs
@@ -12,7 +12,7 @@ use std::{
     pin::Pin,
     sync::{
         Arc, Mutex,
-        atomic::{AtomicBool, AtomicU64, Ordering},
+        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
         mpsc,
     },
     task::{Context, Poll, Waker},
@@ -30,7 +30,7 @@ use pyo3::{
 #[cfg(unix)]
 use super::commands::{ConnectCompletion, TcpReaderStart};
 use super::{
-    callbacks::{CallbackArgs, CallbackId, CallbackKind, ReadyCallback},
+    callbacks::{CallbackArgs, CallbackId, CallbackKind, ReadyCallback, TimerCallback},
     commands::{
         AcceptedConnection, FutureCompletion, LoopCommand, LoopFutureCommand, LoopIoCommand,
         LoopRunCommand, LoopTransportCommand, ReadyItem,
@@ -232,6 +232,14 @@ impl<'a> LocalTimers<'a> {
                     .expect("poisoned pending timers"),
             );
         }
+        // Like asyncio, amortize linear heap rebuilding over large bursts.
+        // Small queues retain at most 100 tombstones; larger queues compact
+        // once more than half of their entries are cancelled. These relaxed
+        // atomics are statistical hints; per-timer ownership uses a mutex.
+        let cancelled = self.core.cancelled_timers.load(Ordering::Relaxed);
+        if cancelled > 100 && cancelled > self.heap.len() / 2 {
+            self.heap.compact_cancelled(ready);
+        }
         let now = Instant::now();
         while self
             .heap
@@ -241,7 +249,8 @@ impl<'a> LocalTimers<'a> {
             let entry = self.heap.pop().expect("timer heap was nonempty");
             // Even cancelled callbacks are released by the ready drain, never
             // while the heap is mutably borrowed: finalizers can schedule work.
-            ready.push_back(ReadyItem::Callback(entry.callback));
+            entry.callback.dequeue();
+            ready.push_back(ReadyItem::TimerCallback(entry.callback));
         }
     }
 
@@ -382,6 +391,7 @@ pub struct LoopCore {
     active_ready_dispatch: Mutex<Option<ActiveReadyDispatch>>,
     pending_timers: Mutex<BinaryHeap<TimerEntry>>,
     pending_timers_dirty: AtomicBool,
+    cancelled_timers: Arc<AtomicUsize>,
     // Wakes the loop thread when a producer enqueues a ready item. Held in an
     // Arc so the park future (`WaitForWake`) can own a clone under `py.detach`.
     wake: Arc<LoopWake>,
@@ -414,6 +424,7 @@ impl LoopCore {
             active_ready_dispatch: Mutex::new(None),
             pending_timers: Mutex::new(BinaryHeap::new()),
             pending_timers_dirty: AtomicBool::new(false),
+            cancelled_timers: Arc::new(AtomicUsize::new(0)),
             wake: Arc::new(LoopWake::new()),
         });
 
@@ -609,16 +620,19 @@ impl LoopCore {
         callback: Py<PyAny>,
         args: Py<PyTuple>,
         context: Option<Py<PyAny>>,
-    ) -> PyResult<(Arc<ReadyCallback>, f64)> {
+    ) -> PyResult<(Arc<TimerCallback>, f64)> {
         let (captured, context_needs_run) = capture_context(py, context)?;
-        let ready = Arc::new(ReadyCallback::new(
-            py,
-            self.next_callback_id(),
-            CallbackKind::Timer,
-            callback,
-            args,
-            captured,
-            context_needs_run,
+        let ready = Arc::new(TimerCallback::new(
+            ReadyCallback::new(
+                py,
+                self.next_callback_id(),
+                CallbackKind::Timer,
+                callback,
+                args,
+                captured,
+                context_needs_run,
+            ),
+            Arc::clone(&self.cancelled_timers),
         ));
 
         let deadline = Instant::now() + delay;
@@ -832,6 +846,14 @@ impl LoopCore {
                             break;
                         }
                     }
+                    ReadyItem::TimerCallback(timer) => {
+                        if let Some(callback) = timer.take_ready() {
+                            if let Some(err) = self.execute_ready(py, Some(&loop_obj), &callback)? {
+                                ready_error = Some(err);
+                                break;
+                            }
+                        }
+                    }
                     ReadyItem::FutureSetResult(payload) => {
                         let FutureCompletion { future, value } = *payload;
                         let future = future.bind(py);
@@ -1503,6 +1525,9 @@ impl LoopCore {
                 .try_enqueue_local_ready(ReadyItem::Callback(callback))
                 .or_else(|item| self.try_enqueue_active_ready(item))
                 .map_err(|item| match item {
+                    ReadyItem::TimerCallback(_) => {
+                        unreachable!("ordinary enqueue cannot produce timers")
+                    }
                     ReadyItem::Callback(callback) => LoopCommand::ScheduleReady(callback),
                     ReadyItem::HandleCallback(handle) => LoopCommand::ScheduleReadyHandle(handle),
                     ReadyItem::Stop => LoopCommand::RequestStop,
@@ -1757,6 +1782,49 @@ mod wake_tests {
         });
     }
 
+    #[test]
+    fn cancelled_timer_burst_is_compacted_behind_an_earlier_live_timer() {
+        crate::initialize_python_for_tests();
+        Python::attach(|py| {
+            let core = LoopCore::new();
+            core.schedule_timer(
+                py,
+                Duration::from_secs(1800),
+                py.None(),
+                PyTuple::empty(py).unbind(),
+                None,
+            )
+            .expect("sentinel");
+            for _ in 0..256 {
+                let (callback, _) = core
+                    .schedule_timer(
+                        py,
+                        Duration::from_secs(3600),
+                        py.None(),
+                        PyTuple::empty(py).unbind(),
+                        None,
+                    )
+                    .expect("timer");
+                callback.cancel();
+                callback.cancel();
+            }
+            let mut local = LocalTimers::new(&core);
+            let mut ready = VecDeque::new();
+            local.collect(&mut ready);
+            let mut remaining = BinaryHeap::new();
+            local.heap.drain_into(&mut remaining);
+            assert_eq!(
+                remaining.len(),
+                1,
+                "cancelled shells retained behind sentinel"
+            );
+            drop(local);
+            drop(remaining);
+            drop(ready);
+            core.close().expect("close loop");
+        });
+    }
+
     struct WakeCounter(AtomicUsize);
 
     impl ArcWake for WakeCounter {
diff --git a/src/engine/timer_entry.rs b/src/engine/timer_entry.rs
index d2df46f..3448bf9 100644
--- a/src/engine/timer_entry.rs
+++ b/src/engine/timer_entry.rs
@@ -24,7 +24,7 @@ fn compare_timer_parts<T: Ord>(
 pub(super) struct TimerEntry {
     pub(super) when: Instant,
     pub(super) seq: u64,
-    pub(super) callback: Arc<super::callbacks::ReadyCallback>,
+    pub(super) callback: Arc<super::callbacks::TimerCallback>,
 }
 
 /// Immediately due timers normally arrive in deadline order and use a FIFO.
@@ -48,6 +48,30 @@ impl TimerQueue {
         self.ordered.is_empty() && self.unordered.is_empty()
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))]
+    pub(super) fn len(&self) -> usize {
+        self.ordered.len() + self.unordered.len()
+    }
+
+    /// Retain live entries in linear time, rebuilding the heap only after a
+    /// cancellation burst. Transfer removed shells to the ready drain rather
+    /// than dropping shared callback ownership under a queue borrow.
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))]
+    pub(super) fn compact_cancelled(&mut self, ready: &mut VecDeque<super::commands::ReadyItem>) {
+        let mut retain = |entry: &TimerEntry| {
+            if !entry.callback.cancelled() {
+                return true;
+            }
+            entry.callback.dequeue();
+            ready.push_back(super::commands::ReadyItem::TimerCallback(Arc::clone(
+                &entry.callback,
+            )));
+            false
+        };
+        self.ordered.retain(&mut retain);
+        self.unordered.retain(retain);
+    }
+
     #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))]
     pub(super) fn push(&mut self, entry: TimerEntry, immediate: bool) {
         // TimerEntry's comparison is reversed for the min-deadline heap.
@@ -203,20 +227,23 @@ mod tests {
     use pyo3::{prelude::*, types::PyTuple};
 
     use super::{TimerEntry, TimerQueue};
-    use crate::engine::callbacks::{CallbackKind, ReadyCallback};
+    use crate::engine::callbacks::{CallbackKind, ReadyCallback, TimerCallback};
 
     fn timer_entry(py: Python<'_>, when: Instant, seq: u64) -> TimerEntry {
         TimerEntry {
             when,
             seq,
-            callback: Arc::new(ReadyCallback::new(
-                py,
-                seq,
-                CallbackKind::Timer,
-                py.None(),
-                PyTuple::empty(py).unbind(),
-                py.None(),
-                false,
+            callback: Arc::new(TimerCallback::new(
+                ReadyCallback::new(
+                    py,
+                    seq,
+                    CallbackKind::Timer,
+                    py.None(),
+                    PyTuple::empty(py).unbind(),
+                    py.None(),
+                    false,
+                ),
+                Arc::new(std::sync::atomic::AtomicUsize::new(0)),
             )),
         }
     }
@@ -246,6 +273,40 @@ mod tests {
         });
     }
 
+    #[test]
+    fn compaction_preserves_live_deadlines_and_ties_in_both_queues() {
+        crate::initialize_python_for_tests();
+        Python::attach(|py| {
+            let now = Instant::now();
+            let mut queue = TimerQueue::new();
+            // FIFO prefix plus scattered heap arrivals, with equal deadlines.
+            for (seq, delay, immediate, cancelled) in [
+                (0, 0, true, false),
+                (1, 0, true, true),
+                (2, 0, true, false),
+                (3, 30, false, true),
+                (4, 10, false, false),
+                (5, 10, false, false),
+                (6, 20, false, true),
+            ] {
+                let entry = timer_entry(py, now + Duration::from_millis(delay), seq);
+                if cancelled {
+                    entry.callback.cancel();
+                }
+                queue.push(entry, immediate);
+            }
+            let mut removed = std::collections::VecDeque::new();
+            queue.compact_cancelled(&mut removed);
+            assert_eq!(removed.len(), 3);
+            assert_eq!(queue.len(), 4);
+            let mut live = Vec::new();
+            while let Some(entry) = queue.pop() {
+                live.push(entry.seq);
+            }
+            assert_eq!(live, [0, 2, 4, 5]);
+        });
+    }
+
     proptest::proptest! {
         #[test]
         fn timer_queue_matches_heap_through_insert_pop_and_merge(
diff --git a/tests/test_timer_cancellation.py b/tests/test_timer_cancellation.py
index 452a954..ab7c04a 100644
--- a/tests/test_timer_cancellation.py
+++ b/tests/test_timer_cancellation.py
@@ -150,7 +150,8 @@ async def main():
     loop.run_until_complete(main())
 
 
-def test_cancel_finalizers_can_reenter_and_close_loop(loop):
+@pytest.mark.parametrize("ownership", ["callback", "args", "context"])
+def test_cancel_finalizers_can_reenter_and_close_loop(loop, ownership):
     if not isinstance(loop, rsloop.Loop):
         pytest.skip("reentrant close is a separate rsloop lifecycle contract")
     events = []
@@ -167,7 +168,16 @@ def __del__(self):
                 loop.call_later(0, lambda: None).cancel()
                 loop.close()
 
-    handles.append(loop.call_later(3600, Callback()))
+    if ownership == "callback":
+        handles.append(loop.call_later(3600, Callback()))
+    elif ownership == "args":
+        handles.append(loop.call_later(3600, lambda arg: None, Callback()))
+    else:
+        var = contextvars.ContextVar("finalizer_payload")
+        context = contextvars.Context()
+        context.run(var.set, Callback())
+        handles.append(loop.call_later(3600, lambda: None, context=context))
+        del context
     handles[0].cancel()
     released_immediately = events == ["released"]
     if not released_immediately:
@@ -244,3 +254,58 @@ def cancel():
         cancelled.set()
         thread.join(timeout=5)
         assert not thread.is_alive()
+
+
+@pytest.mark.parametrize("mode", ["timeout", "wait_for"])
+def test_expired_timeout_still_cancels_awaited_work(loop, mode):
+    if mode == "timeout" and not hasattr(asyncio, "timeout"):
+        pytest.skip("asyncio.timeout requires Python 3.11")
+
+    async def main():
+        finalized = []
+
+        async def work():
+            try:
+                await asyncio.Future()
+            finally:
+                finalized.append("cancelled")
+
+        with pytest.raises(asyncio.TimeoutError):
+            if mode == "timeout":
+                async with asyncio.timeout(0.01):
+                    await work()
+            else:
+                await asyncio.wait_for(work(), 0.01)
+        assert finalized == ["cancelled"]
+        assert len(asyncio.all_tasks()) == 1
+
+    loop.run_until_complete(main())
+
+
+def test_periodic_timers_do_not_retain_cancelled_payloads(loop):
+    async def main():
+        started = [asyncio.Event(), asyncio.Event()]
+
+        async def heartbeat(index):
+            started[index].set()
+            while True:
+                await asyncio.sleep(0.01)
+
+        beats = [loop.create_task(heartbeat(index)) for index in range(2)]
+        try:
+            await asyncio.gather(*(event.wait() for event in started))
+            references = []
+            for _ in range(100):
+                handle, reference = make_timer(loop, "callback")
+                handle.cancel()
+                references.append(reference)
+            for _ in range(3):
+                await asyncio.sleep(0.01)
+            gc.collect()
+            assert all(reference() is None for reference in references)
+        finally:
+            for beat in beats:
+                beat.cancel()
+            await asyncio.gather(*beats, return_exceptions=True)
+
+    loop.run_until_complete(main())

From 5fda7bcfa0e095e629f14ca53e1941aee0bc341f Mon Sep 17 00:00:00 2001
From: Yehor Smoliakov <yehors@ukr.net>
Date: Sun, 11 Oct 2026 02:13:30 -0700
Subject: [PATCH 3/4] test: support Python 3.10 PostgreSQL probes and document
 lifetime measurements

---
 benches/timer_cancellation.py     |  12 +-
 docs/cancelled-timer-lifetimes.md | 182 ++++++++++++++++++++++++++++++
 2 files changed, 189 insertions(+), 5 deletions(-)
 create mode 100644 docs/cancelled-timer-lifetimes.md

diff --git a/benches/timer_cancellation.py b/benches/timer_cancellation.py
index c7c676d..c067a3a 100644
--- a/benches/timer_cancellation.py
+++ b/benches/timer_cancellation.py
@@ -127,11 +127,13 @@ async def connection():
         )
         try:
             for index in range(3):
-                async with asyncio.timeout(3600):
-                    assert (
-                        await conn.fetchval(f"SELECT $1::int + {index}", index)
-                        == 2 * index
-                    )
+                query = conn.fetchval(f"SELECT $1::int + {index}", index)
+                if hasattr(asyncio, "timeout"):
+                    async with asyncio.timeout(3600):
+                        value = await query
+                else:
+                    value = await asyncio.wait_for(query, 3600)
+                assert value == 2 * index
         finally:
             await conn.close()
 
diff --git a/docs/cancelled-timer-lifetimes.md b/docs/cancelled-timer-lifetimes.md
new file mode 100644
index 0000000..d8a8d14
--- /dev/null
+++ b/docs/cancelled-timer-lifetimes.md
@@ -0,0 +1,182 @@
+# Cancelled timer ownership (issue #118)
+
+[Issue #118](https://github.com/RustedBytes/rsloop/issues/118) reports that
+`TimerHandle.cancel()` retains callbacks, arguments and captured context behind
+an earlier live timer. The original deadline then determines the lifetime of
+finished `asyncio.timeout()` tasks and closed asyncpg statement caches.
+
+## Cause and fix
+
+Previously, `ReadyCallback::cancel()` only changed an atomic flag. The timer
+queue visited only expired or cancelled entries at its head; an earlier active
+timer hid every cancelled entry behind it.
+
+Timers now use a separate `TimerCallback` with an optional `ReadyCallback`
+payload behind a mutex. Cancellation marks the timer, accounts for its queued
+shell once and takes the payload. The payload is dropped **after unlocking**,
+while the Python handle's `cancel()` call is still attached to Python. This
+releases callback, args and context before returning for an unclaimed timer.
+Repeated cancellation is harmless, including cancellation by a finalizer.
+
+Dispatch claims the payload under the same mutex, then invokes it without any
+lock. If dispatch wins a race with cancellation, that invocation owns its
+callback, arguments and context until it finishes. Cancellation prevents future
+invocation and cannot invalidate an already claimed callback's objects.
+
+The queue counts cancelled shells under per-timer synchronization. Queue
+removal decrements the count once; dropping a queued cancelled timer during
+shutdown also balances the count. The atomic loads used by heap inspection and
+compaction are statistical hints, not synchronization for Python ownership.
+
+Collection compacts both the immediate FIFO and future-deadline heap when the
+cancelled count exceeds 100 **and** half the queue size, matching asyncio's
+amortization thresholds. Live deadlines and insertion order for ties survive
+compaction. Removed shells transfer to the ready drain so shared callback
+ownership is never dropped under a timer queue borrow. Large cancellation
+bursts therefore stop accumulating behind a long-lived earlier timer at the
+next collection. At or below the thresholds, small or predominantly live queues
+can still contain payload-free cancelled entries. A stopped loop retains shells
+until its next run or close; Python payload release does not wait for that run.
+
+The existing `ReadyCallback`, Python `Handle` and `call_soon` paths retain their
+layout and lock-free invocation. Synchronization and extra state are timer-only;
+no dependencies or public Python signatures change.
+
+## TDD and behavioral comparisons
+
+Baseline: master `5cfc0594b734d60e0b1db44033e6175ac202868d` (0.1.63).
+The test-only PR commit is
+[`067d208`](https://github.com/RustedBytes/rsloop/commit/067d2084b940d53aa8edc59397bb1f51a3f5ee98).
+
+Before the implementation, 12 lifetime regressions failed on rsloop while
+asyncio/uvloop controls passed; two rsloop-specific reentrant-close comparisons
+were skipped on controls. A Rust regression also observed **257** queued entries
+(one live sentinel plus 256 cancelled timers) where it expected **1**.
+
+The expanded regressions cover callback/one-arg/many-arg/context ownership;
+explicit and captured contexts; stopped and running loops; retained handles;
+earlier live and periodic timers; completed `timeout()`/`wait_for()` tasks and
+results; expired timeout cancellation; cancellation after heap dequeue; repeated
+cross-thread cancellation; cancel versus dispatch/removal races; invocation
+ownership during cancellation; finalizer reentry through scheduling,
+cancellation and close; and compaction ordering in both queue types.
+
+One control difference is intentional: CPython and uvloop clear callback and
+args on cancellation but keep Context on a user-held handle. rsloop releases all
+three payload components when safe. Control tests record that difference rather
+than assuming identical Context ownership.
+
+## Memory measurements
+
+The checked-in probe `benches/timer_cancellation.py` runs each case in a fresh
+process. A live sentinel has an 1800 s deadline; cancelled timers, timeout blocks
+and statement-cache expiry timers have 3600 s deadlines. No measured deadline
+expires. Counts are observed after GC and three zero-delay loop turns, allowing
+`gather` completion callbacks to release their normal transient ownership.
+
+Local measurements: Linux x86_64, glibc 2.39, CPython 3.14.7, epoll fallback;
+both builds compiled from source in release mode with the pinned
+`nightly-2026-09-25` toolchain and locked Cargo dependencies.
+Python bytes subtract the probe's list/weakref storage, following the issue's
+method. RSS includes native allocations and allocator high-water retention.
+
+| Local case (20,000 objects) | Before | After |
+|---|---:|---:|
+| Cancelled timer payloads alive immediately / after turns | 20,000 / 20,000 | 0 / 0 |
+| Timer Python bytes retained after turns | 87,223,891 B | 307 B |
+| Timer RSS before workload → after turns | 26.2 → 125.2 MiB | 26.2 → 33.1 MiB |
+| Finished timeout tasks alive after turns | 20,000 | 0 |
+| Timeout Python bytes retained after turns | 101,011,784 B | 29,893 B |
+| Timeout RSS before workload → after turns | 26.1 → 150.4 MiB | 26.0 → 34.1 MiB |
+
+Immediately after the last `gather`, the fixed timeout probe sees 500 tasks
+from its last batch, as do asyncio and uvloop. These disappear after completion
+callbacks drain. This transient is separate from cancelled timer retention.
+
+Real PostgreSQL measurements ran in GitHub Actions with PostgreSQL 16.15,
+asyncpg 0.32.0, uvloop 0.23.0, CPython 3.14.8, Linux x86_64/glibc 2.39.
+Each process closes 500 connections, in groups of 10, after three distinct cached
+parameterized queries per connection, each under `asyncio.timeout()`.
+
+| Objects remaining after 500 closed connections | asyncio | uvloop | rsloop before | rsloop after |
+|---|---:|---:|---:|---:|
+| `_StatementCache` | 0 | 0 | 500 | 0 |
+| `PreparedStatementState` | 0 | 0 | 1500 | 0 |
+| `Protocol` | 0 | 0 | 0 | 0 |
+
+The 50-connection regression failed before the fix with 50 caches and 150 states;
+after the fix all **39 PostgreSQL lifecycle tests passed**, including existing
+SQLAlchemy transaction, pool exhaustion, concurrent writer and disconnect tests.
+[Baseline CI](https://github.com/RustedBytes/rsloop/actions/runs/38126971249)
+and [fixed CI](https://github.com/RustedBytes/rsloop/actions/runs/38127314821)
+contain JSON measurements and the regression results.
+
+These are fresh-process bounded reproductions, not a repeat of the author's
+300 s Granian load test. They do not establish production steady-state RSS.
+PostgreSQL was available in CI; local server setup was unavailable in the
+restricted environment. Python 3.10 uses `wait_for()` around its queries and
+explicitly skips the standalone `asyncio.timeout()` probe, which requires 3.11.
+
+## Timer performance and tradeoff
+
+Same local release builds and interpreter; 1,000,000 operations per sample,
+1,000-timer batches, one warmup, seven samples per process, two before/after
+pairs with reversed order. The table uses the median of 14 samples per build.
+Tracemalloc/GC probes are not enabled in timed spans. These are scheduler
+microbenchmarks, not application throughput claims.
+
+| Workload | Before (s) | After (s) | Time change |
+|---|---:|---:|---:|
+| `call_soon` callback burst | 0.18194 | 0.18122 | -0.4% |
+| Zero-delay timers, handles retained | 0.29481 | 0.31178 | +5.8% |
+| Zero-delay timers, handles discarded | 0.25788 | 0.28745 | +11.5% |
+| Cancelled zero-delay timers | 0.26746 | 0.28294 | +5.8% |
+| Cancelled scattered future timers | 0.37740 | 0.36704 | -2.7% |
+
+The extra timer synchronization costs approximately 15–30 ns per operation in
+these local workloads. The `call_soon` difference is within measurement noise;
+its hot path has no added mutex. The mixed-timer result is not evidence of a
+general speed improvement. Safe ownership transfer is the reason for the timer
+tradeoff; release-mode and downstream measurements should guide later changes.
+
+## Validation and reproduction
+
+Local validation:
+
+- New differential tests: 51 passed, 6 skipped (the three reentrant-close
+  ownership cases are rsloop-specific and skip on both controls).
+- Free-threaded CPython 3.14.8: 19 new rsloop tests plus 5 existing local-timer
+  tests passed. The Rust synchronization tests also create real concurrent
+  threads while the Python loop thread is attached.
+- Full Python core suite: 628 passed, 12 skipped; two Unix-socket cases fail with
+  `PermissionError` in the local sandbox and fail identically on the baseline.
+- Downstream ecosystem suite: 78 passed, 39 PostgreSQL cases skipped locally;
+  PostgreSQL executes against the real CI service.
+- Rust default suite: 336 passed; 15 environment-restricted io_uring/Unix cases
+  fail locally. All-features suite: 432 passed; 19 restricted cases fail locally.
+- Clippy: `cargo clippy --all-targets --all-features --locked -- -D warnings`
+  passed. Cargo fmt, Ruff and targeted Pyright passed.
+
+The first local incremental all-features link failed with missing generated
+monomorphization symbols; rerunning with `CARGO_INCREMENTAL=0` resolved the link.
+The initial test-only CI all-features build passed with the standard command;
+this local build-cache issue did not require a source workaround.
+
+```sh
+# Use the locked test dependencies and build the project first.
+uv sync --locked --no-default-groups --group ecosystem --no-install-project
+uv run --no-sync maturin develop --release --locked
+uv run --no-sync python -m pytest tests/test_timer_cancellation.py tests/test_local_timers.py
+uv run --no-sync python scripts/run_rust_tests.py
+uv run --no-sync python scripts/run_rust_tests.py --all-features
+cargo clippy --all-targets --all-features --locked -- -D warnings
+cargo fmt --check
+
+# Run each line in a fresh process; repeat for asyncio and uvloop.
+uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case timers
+uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case timeout
+uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case performance --count 1000000
+# Set RSLOOP_POSTGRES_DSN to a dedicated test database for these commands.
+uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case postgres --count 500
+uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem
+```

From 5db78d458758692d2867988b6635e5a65c780458 Mon Sep 17 00:00:00 2001
From: Yehor Smoliakov <yehors@ukr.net>
Date: Sun, 11 Oct 2026 12:23:42 +0300
Subject: [PATCH 4/4] Delete docs/cancelled-timer-lifetimes.md

---
 docs/cancelled-timer-lifetimes.md | 182 ------------------------------
 1 file changed, 182 deletions(-)
 delete mode 100644 docs/cancelled-timer-lifetimes.md

diff --git a/docs/cancelled-timer-lifetimes.md b/docs/cancelled-timer-lifetimes.md
deleted file mode 100644
index d8a8d14..0000000
--- a/docs/cancelled-timer-lifetimes.md
+++ /dev/null
@@ -1,182 +0,0 @@
-# Cancelled timer ownership (issue #118)
-
-[Issue #118](https://github.com/RustedBytes/rsloop/issues/118) reports that
-`TimerHandle.cancel()` retains callbacks, arguments and captured context behind
-an earlier live timer. The original deadline then determines the lifetime of
-finished `asyncio.timeout()` tasks and closed asyncpg statement caches.
-
-## Cause and fix
-
-Previously, `ReadyCallback::cancel()` only changed an atomic flag. The timer
-queue visited only expired or cancelled entries at its head; an earlier active
-timer hid every cancelled entry behind it.
-
-Timers now use a separate `TimerCallback` with an optional `ReadyCallback`
-payload behind a mutex. Cancellation marks the timer, accounts for its queued
-shell once and takes the payload. The payload is dropped **after unlocking**,
-while the Python handle's `cancel()` call is still attached to Python. This
-releases callback, args and context before returning for an unclaimed timer.
-Repeated cancellation is harmless, including cancellation by a finalizer.
-
-Dispatch claims the payload under the same mutex, then invokes it without any
-lock. If dispatch wins a race with cancellation, that invocation owns its
-callback, arguments and context until it finishes. Cancellation prevents future
-invocation and cannot invalidate an already claimed callback's objects.
-
-The queue counts cancelled shells under per-timer synchronization. Queue
-removal decrements the count once; dropping a queued cancelled timer during
-shutdown also balances the count. The atomic loads used by heap inspection and
-compaction are statistical hints, not synchronization for Python ownership.
-
-Collection compacts both the immediate FIFO and future-deadline heap when the
-cancelled count exceeds 100 **and** half the queue size, matching asyncio's
-amortization thresholds. Live deadlines and insertion order for ties survive
-compaction. Removed shells transfer to the ready drain so shared callback
-ownership is never dropped under a timer queue borrow. Large cancellation
-bursts therefore stop accumulating behind a long-lived earlier timer at the
-next collection. At or below the thresholds, small or predominantly live queues
-can still contain payload-free cancelled entries. A stopped loop retains shells
-until its next run or close; Python payload release does not wait for that run.
-
-The existing `ReadyCallback`, Python `Handle` and `call_soon` paths retain their
-layout and lock-free invocation. Synchronization and extra state are timer-only;
-no dependencies or public Python signatures change.
-
-## TDD and behavioral comparisons
-
-Baseline: master `5cfc0594b734d60e0b1db44033e6175ac202868d` (0.1.63).
-The test-only PR commit is
-[`067d208`](https://github.com/RustedBytes/rsloop/commit/067d2084b940d53aa8edc59397bb1f51a3f5ee98).
-
-Before the implementation, 12 lifetime regressions failed on rsloop while
-asyncio/uvloop controls passed; two rsloop-specific reentrant-close comparisons
-were skipped on controls. A Rust regression also observed **257** queued entries
-(one live sentinel plus 256 cancelled timers) where it expected **1**.
-
-The expanded regressions cover callback/one-arg/many-arg/context ownership;
-explicit and captured contexts; stopped and running loops; retained handles;
-earlier live and periodic timers; completed `timeout()`/`wait_for()` tasks and
-results; expired timeout cancellation; cancellation after heap dequeue; repeated
-cross-thread cancellation; cancel versus dispatch/removal races; invocation
-ownership during cancellation; finalizer reentry through scheduling,
-cancellation and close; and compaction ordering in both queue types.
-
-One control difference is intentional: CPython and uvloop clear callback and
-args on cancellation but keep Context on a user-held handle. rsloop releases all
-three payload components when safe. Control tests record that difference rather
-than assuming identical Context ownership.
-
-## Memory measurements
-
-The checked-in probe `benches/timer_cancellation.py` runs each case in a fresh
-process. A live sentinel has an 1800 s deadline; cancelled timers, timeout blocks
-and statement-cache expiry timers have 3600 s deadlines. No measured deadline
-expires. Counts are observed after GC and three zero-delay loop turns, allowing
-`gather` completion callbacks to release their normal transient ownership.
-
-Local measurements: Linux x86_64, glibc 2.39, CPython 3.14.7, epoll fallback;
-both builds compiled from source in release mode with the pinned
-`nightly-2026-09-25` toolchain and locked Cargo dependencies.
-Python bytes subtract the probe's list/weakref storage, following the issue's
-method. RSS includes native allocations and allocator high-water retention.
-
-| Local case (20,000 objects) | Before | After |
-|---|---:|---:|
-| Cancelled timer payloads alive immediately / after turns | 20,000 / 20,000 | 0 / 0 |
-| Timer Python bytes retained after turns | 87,223,891 B | 307 B |
-| Timer RSS before workload → after turns | 26.2 → 125.2 MiB | 26.2 → 33.1 MiB |
-| Finished timeout tasks alive after turns | 20,000 | 0 |
-| Timeout Python bytes retained after turns | 101,011,784 B | 29,893 B |
-| Timeout RSS before workload → after turns | 26.1 → 150.4 MiB | 26.0 → 34.1 MiB |
-
-Immediately after the last `gather`, the fixed timeout probe sees 500 tasks
-from its last batch, as do asyncio and uvloop. These disappear after completion
-callbacks drain. This transient is separate from cancelled timer retention.
-
-Real PostgreSQL measurements ran in GitHub Actions with PostgreSQL 16.15,
-asyncpg 0.32.0, uvloop 0.23.0, CPython 3.14.8, Linux x86_64/glibc 2.39.
-Each process closes 500 connections, in groups of 10, after three distinct cached
-parameterized queries per connection, each under `asyncio.timeout()`.
-
-| Objects remaining after 500 closed connections | asyncio | uvloop | rsloop before | rsloop after |
-|---|---:|---:|---:|---:|
-| `_StatementCache` | 0 | 0 | 500 | 0 |
-| `PreparedStatementState` | 0 | 0 | 1500 | 0 |
-| `Protocol` | 0 | 0 | 0 | 0 |
-
-The 50-connection regression failed before the fix with 50 caches and 150 states;
-after the fix all **39 PostgreSQL lifecycle tests passed**, including existing
-SQLAlchemy transaction, pool exhaustion, concurrent writer and disconnect tests.
-[Baseline CI](https://github.com/RustedBytes/rsloop/actions/runs/38126971249)
-and [fixed CI](https://github.com/RustedBytes/rsloop/actions/runs/38127314821)
-contain JSON measurements and the regression results.
-
-These are fresh-process bounded reproductions, not a repeat of the author's
-300 s Granian load test. They do not establish production steady-state RSS.
-PostgreSQL was available in CI; local server setup was unavailable in the
-restricted environment. Python 3.10 uses `wait_for()` around its queries and
-explicitly skips the standalone `asyncio.timeout()` probe, which requires 3.11.
-
-## Timer performance and tradeoff
-
-Same local release builds and interpreter; 1,000,000 operations per sample,
-1,000-timer batches, one warmup, seven samples per process, two before/after
-pairs with reversed order. The table uses the median of 14 samples per build.
-Tracemalloc/GC probes are not enabled in timed spans. These are scheduler
-microbenchmarks, not application throughput claims.
-
-| Workload | Before (s) | After (s) | Time change |
-|---|---:|---:|---:|
-| `call_soon` callback burst | 0.18194 | 0.18122 | -0.4% |
-| Zero-delay timers, handles retained | 0.29481 | 0.31178 | +5.8% |
-| Zero-delay timers, handles discarded | 0.25788 | 0.28745 | +11.5% |
-| Cancelled zero-delay timers | 0.26746 | 0.28294 | +5.8% |
-| Cancelled scattered future timers | 0.37740 | 0.36704 | -2.7% |
-
-The extra timer synchronization costs approximately 15–30 ns per operation in
-these local workloads. The `call_soon` difference is within measurement noise;
-its hot path has no added mutex. The mixed-timer result is not evidence of a
-general speed improvement. Safe ownership transfer is the reason for the timer
-tradeoff; release-mode and downstream measurements should guide later changes.
-
-## Validation and reproduction
-
-Local validation:
-
-- New differential tests: 51 passed, 6 skipped (the three reentrant-close
-  ownership cases are rsloop-specific and skip on both controls).
-- Free-threaded CPython 3.14.8: 19 new rsloop tests plus 5 existing local-timer
-  tests passed. The Rust synchronization tests also create real concurrent
-  threads while the Python loop thread is attached.
-- Full Python core suite: 628 passed, 12 skipped; two Unix-socket cases fail with
-  `PermissionError` in the local sandbox and fail identically on the baseline.
-- Downstream ecosystem suite: 78 passed, 39 PostgreSQL cases skipped locally;
-  PostgreSQL executes against the real CI service.
-- Rust default suite: 336 passed; 15 environment-restricted io_uring/Unix cases
-  fail locally. All-features suite: 432 passed; 19 restricted cases fail locally.
-- Clippy: `cargo clippy --all-targets --all-features --locked -- -D warnings`
-  passed. Cargo fmt, Ruff and targeted Pyright passed.
-
-The first local incremental all-features link failed with missing generated
-monomorphization symbols; rerunning with `CARGO_INCREMENTAL=0` resolved the link.
-The initial test-only CI all-features build passed with the standard command;
-this local build-cache issue did not require a source workaround.
-
-```sh
-# Use the locked test dependencies and build the project first.
-uv sync --locked --no-default-groups --group ecosystem --no-install-project
-uv run --no-sync maturin develop --release --locked
-uv run --no-sync python -m pytest tests/test_timer_cancellation.py tests/test_local_timers.py
-uv run --no-sync python scripts/run_rust_tests.py
-uv run --no-sync python scripts/run_rust_tests.py --all-features
-cargo clippy --all-targets --all-features --locked -- -D warnings
-cargo fmt --check
-
-# Run each line in a fresh process; repeat for asyncio and uvloop.
-uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case timers
-uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case timeout
-uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case performance --count 1000000
-# Set RSLOOP_POSTGRES_DSN to a dedicated test database for these commands.
-uv run --no-sync python benches/timer_cancellation.py --loop rsloop --case postgres --count 500
-uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem
-```