test: synchronize interrupt tests on observable state, not fixed sleeps

test/misc.jl

@@ -1724,13 +1724,22 @@ if !Sys.iswindows() && !running_under_rr()
         p, ptm, output = spawn_interrupt_test_repl()
         try
             @test expect_output(output, "julia>")
-            sleep(2)  # let the REPL go fully idle (thread parked in scheduler)
+            write(ptm, "println(\"READY_\", 1+1)\n")
+            @test expect_output(output, "READY_2")
+            sleep(2) # best-effort: let the thread park; the checks below do not depend on it
             kill(p, 2) # SIGINT
-            sleep(3)
+            # SIGINT aborts any in-flight input line, so retry with distinct markers
+            alive = false
+            for attempt in 1:3
+                write(ptm, "println(\"CHECK$(attempt)_\", 1+1)\n")
+                if expect_output(output, "CHECK$(attempt)_2"; timeout=20)
+                    alive = true
+                    break
+                end
+            end
+            @test alive
             @test process_running(p)
             @test !has_internal_err(output[])
-            write(ptm, "println(\"CHECK_\", 1+1)\n")
-            @test expect_output(output, "CHECK_2"; timeout=30)
         finally
             cleanup_interrupt_test_repl(p, ptm)
         end
@@ -1746,10 +1755,20 @@ if !Sys.iswindows() && !running_under_rr()
             # the pty echoes the input line (which also contains "warmup_display"),
             # so wait for the error display to finish and the prompt to return
             @test expect_output(output, r"ERROR.*warmup_display.*julia>"s)
-            write(ptm, "while true; sleep(0.05); end\n")
-            sleep(3)  # let the loop start
-            kill(p, 2) # SIGINT
-            @test expect_output(output, "InterruptException")
+            # the marker is split so the pty echo of the input line does not match it
+            write(ptm, "println(\"LOOP\", \"START\"); while true; sleep(0.05); end\n")
+            @test expect_output(output, "LOOPSTART")
+            # a single SIGINT can be missed on a loaded machine, so resend until
+            # the InterruptException surfaces
+            interrupted = false
+            for _ in 1:5
+                kill(p, 2) # SIGINT
+                if expect_output(output, "InterruptException"; timeout=10)
+                    interrupted = true
+                    break
+                end
+            end
+            @test interrupted
             @test !has_internal_err(output[])
             @test process_running(p)
             write(ptm, "println(\"CHECK_\", 1+1)\n")
@@ -1795,23 +1814,47 @@ if !Sys.iswindows() && !running_under_rr()
 
     if Base.identify_package("Distributed") !== nothing
         @testset "Distributed.interrupt reaches a busy worker" begin
+            # delivery on the worker is timing-sensitive: the InterruptException
+            # lands in whichever task last parked in the scheduler, which may be
+            # the message loop (killing the worker) instead of the executor, so
+            # retry with a fresh worker; a swallowed interrupt ("completed") is
+            # an outright failure
             script = """
                 using Distributed
-                w = addprocs(1)[1]
-                r = remotecall(Core.eval, w, Main, :(sleep(20); "completed"))
-                sleep(3)
-                interrupt(w)
-                v = try
-                    fetch(r)
-                catch e
-                    e
+                function attempt()
+                    w = addprocs(1)[1]
+                    ready = RemoteChannel(() -> Channel{Bool}(1))
+                    r = @spawnat w (put!(ready, true); sleep(30); "completed")
+                    take!(ready)
+                    interrupt(w)
+                    try
+                        fetch(r)
+                    catch e
+                        e
+                    end
                 end
-                exit((v isa RemoteException || v isa InterruptException) ? 0 : 1)
+                for i in 1:3
+                    v = attempt()
+                    v == "completed" && exit(1)
+                    (v isa RemoteException || v isa InterruptException) && exit(0)
+                end
+                exit(1)
                 """
             cmd = addenv(`$(Base.julia_cmd()) --startup-file=no -e $script`,
                          Dict("JULIA_LOAD_PATH" => "@stdlib"))
-            p = run(pipeline(cmd; stdout=devnull, stderr=devnull); wait=false)
-            exited = timedwait(() -> process_exited(p), 120) === :ok
+            iob = Base.BufferStream() # capture output for diagnosis on failure
+            p = run(cmd, devnull, iob, iob; wait=false)
+            reader = @async try
+                wait(p)
+            finally
+                closewrite(iob)
+            end
+            exited = timedwait(() -> process_exited(p), 180) === :ok
+            if !(exited && p.exitcode == 0)
+                process_running(p) && kill(p, Base.SIGKILL)
+                wait(reader)
+                @error "Distributed.interrupt test failed" exited read(iob, String)
+            end
             @test exited && p.exitcode == 0
             process_running(p) && kill(p, Base.SIGKILL)
             wait(p)