Let a SIGINT leave the reaper task and restart it on the next connect

docs/protocol-notes.md

@@ -104,7 +104,10 @@ source are never read.
 - **Finalizers never do I/O**: a dropped `MySQL.Handle` enqueues its `ReapEntry` (CAS
   `:live → :pending`); the reaper (0.5 s timer, `reap_now!`, `atexit`) removes entries under
   `REAPER_LOCK`, then closes each transport after releasing the lock. `close!` retires the
-  entry first so a later finalizer is a no-op. Reseau's own poll-FD finalizer is the
+  entry first so a later finalizer is a no-op. The reaper task runs on the default pool and
+  catches nothing around its timer wait, so a Ctrl-C that lands in it leaves the task as an
+  `InterruptException` (which Base hands to the REPL); `ensure_reaper!` starts a new task on
+  the next connect when the old one is done. Reseau's own poll-FD finalizer is the
   last-resort fd reclaimer.
 - **RSA-OAEP through OpenSSL_jll's libcrypto** (`crypto.jl`): explicit SHA-1 OAEP + MGF1,
   `k - 42` plaintext cap, every handle freed in `finally`, 20k-iteration leak test. The

src/reaper.jl

@@ -36,6 +36,7 @@ const REAPER_LOCK = Threads.SpinLock()
 const REAPER_QUEUE = Ref{Union{Nothing, ReapEntry}}(nothing)
 const REAPER_QUEUE_LENGTH = Ref(0)
 const REAPER_TIMER = Ref{Union{Nothing, Timer}}(nothing)
+const REAPER_TASK = Ref{Union{Nothing, Task}}(nothing)
 const REAPER_INTERVAL_S = 0.5
 const REAPER_STATS = Ref((enqueued=0, closed=0))
 
@@ -127,15 +128,15 @@ const REAPER_SETUP_LOCK = ReentrantLock()
 # `Timer(callback)`: Base dispatches timer callbacks from an internal closure task, which
 # a trimmed executable cannot run, while `wait(::Timer)` is woken from the event loop's C
 # side. The task body and the atexit hook are `@noinline` with static call edges below.
+#
+# Nothing is caught around `wait`: a SIGINT is thrown into whichever task is resident on
+# thread 1, and when that is the reaper the InterruptException must leave the task so that
+# Base's task_done_hook can hand it to the REPL. `ensure_reaper!` replaces a dead reaper.
 @noinline function reaper_loop()
     t = REAPER_TIMER[]
     t === nothing && return nothing
     while isopen(t)
-        try
-            wait(t)
-        catch
-            break   # timer closed
-        end
+        wait(t)
         try
             reap_now!()
         catch err
@@ -147,23 +148,30 @@ end
 
 @noinline reaper_atexit() = return (try; reap_now!(); catch; end; nothing)
 
-# Starts the timer once. A ReentrantLock (not the finalizer-safe spinlock) because creating a
-# Timer and registering the atexit hook may yield.
+# Starts the timer once and a reaper task whenever none is alive. A ReentrantLock (not the
+# finalizer-safe spinlock) because creating a Timer and registering the atexit hook may yield.
 function ensure_reaper!()
     lock(REAPER_SETUP_LOCK)
     try
-        REAPER_TIMER[] === nothing || return nothing
-        REAPER_TIMER[] = Timer(REAPER_INTERVAL_S; interval=REAPER_INTERVAL_S)
+        task = REAPER_TASK[]
+        task === nothing || istaskdone(task) || return nothing
+        if REAPER_TIMER[] === nothing
+            REAPER_TIMER[] = Timer(REAPER_INTERVAL_S; interval=REAPER_INTERVAL_S)
+            atexit(reaper_atexit)
+        end
         if TRIM_CALL_EDGE[]
             reaper_loop()
             reaper_atexit()
         end
         task = Task(reaper_loop)
         task.sticky = false
-        # reaper_loop handles timer shutdown and logs reclamation errors itself.
-        # Base.errormonitor would add an untrimmable dynamic error-display task.
+        # The default pool: when an interactive pool exists the user's tasks run there, and
+        # a Ctrl-C goes to the task resident on thread 1, which must not be the reaper.
+        Threads._spawn_set_thrpool(task, :default)
+        # reaper_loop logs reclamation errors itself. Base.errormonitor would add an
+        # untrimmable dynamic error-display task.
+        REAPER_TASK[] = task
         schedule(task)
-        atexit(reaper_atexit)
     finally
         unlock(REAPER_SETUP_LOCK)
     end

test/protocol/native_tests.jl

@@ -476,6 +476,56 @@ end
     @test all(ref -> ref.value === nothing, refs)
 end
 
+@testset "reaper: an interrupt leaves the task and the next registration restarts it" begin
+    N.ensure_reaper!()
+    reaper = N.REAPER_TASK[]
+    @test reaper isa Task
+    # Off the interactive pool, where the user's work runs when one exists, so a Ctrl-C
+    # (thrown into the task resident on thread 1) does not land in the reaper.
+    @test Threads.threadpool(reaper) === :default
+    timer = N.REAPER_TIMER[]
+    # Under the timer's lock, a task linked in the timer's wait queue is parked in `wait(timer)`.
+    # The queue holds the tasks themselves on Julia <= 1.13 and WaitEntry records with an
+    # atomic `task` field from 1.14 on.
+    in_waitq(t::Task, c) = any(w -> (w isa Task ? w : (@atomic :monotonic w.task)) === t, c.waitq)
+    # Throw an InterruptException into the reaper while it is parked in `wait(timer)`, as a
+    # Ctrl-C does when the reaper is the task resident on thread 1. The timer's lock keeps
+    # the timer tick from waking the reaper between the check and the throw.
+    injected = false
+    deadline = time() + 15
+    while !injected && time() < deadline
+        @lock timer.cond begin
+            if in_waitq(reaper, timer.cond)
+                schedule(reaper, InterruptException(); error=true)
+                injected = true
+            end
+        end
+        injected || sleep(0.01)
+    end
+    @test injected
+    deadline = time() + 15
+    while !istaskdone(reaper) && time() < deadline
+        sleep(0.01)
+    end
+    # The interrupt leaves the task, so Base's task_done_hook can hand it to the REPL.
+    @test istaskfailed(reaper)
+    @test reaper.result isa InterruptException
+    N.ensure_reaper!()
+    fresh = N.REAPER_TASK[]
+    @test fresh isa Task && fresh !== reaper && !istaskdone(fresh)
+    counter = P.FaultTransport(IOBuffer())
+    entry = N.ReapEntry(counter)
+    while (@atomic entry.state) == :live
+        N.enqueue_from_finalizer!(entry) || yield()
+    end
+    deadline = time() + 15
+    while (@atomic entry.state) != :closed && time() < deadline
+        sleep(0.05)
+    end
+    @test (@atomic entry.state) == :closed
+    @test (@atomic counter.close_count) == 1
+end
+
 @testset "no descriptor growth across connect/close cycles" begin
     Sys.iswindows() && return
     multi_accept_server() do port