Skip to content

Commit 7bc9734

Browse files
authored
Abort :http streaming read when the consumer closes the channel (#98)
The :http backend's streaming request had no way to abort an in-flight socket read when the consumer closes `stream_to`. Closing the channel stopped the chunk-reader task but left the read task blocked in `eof(io)`/`readbytes!(io)` on the socket, so a streaming request whose consumer stops early (e.g. a Kubernetes watch stopped via `close(stream)` after the awaited event arrives, or a timer firing) would hang until the read-idle timeout instead of returning promptly. Add an abort-on-close watcher task, mirroring the :downloads backend: capture the connection `io`, and when `stream_to` closes, close `io` and schedule an `InterruptException` on the read task. `close(io)` alone does not wake an HTTP/2 body read parked on the flow-control timer (`_wait_h2_body_progress!` -> `timedwait`), so the scheduled interrupt is what reliably unblocks it (the same fallback :downloads uses for non-interruptible downloads). The read task swallows the abort (channel closed, or our InterruptException) and returns normally; a genuine network error or read-idle timeout arrives while `stream_to` is still open, so it still propagates. The interrupt surfaces to callers as `InvocationException("request was interrupted")` via `exec`, which `is_request_interrupted` already recognizes. Affects both HTTP.jl 1.x and 2.x (shared :http code path). Validated against a live Kubernetes watch on HTTP 1.11 and 2.5: consumer-initiated stop now returns in ~2-3s instead of hanging.
1 parent 3f9bf09 commit 7bc9734

1 file changed

Lines changed: 55 additions & 2 deletions

File tree

src/client/httplibs/juliaweb_http.jl

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -290,10 +290,16 @@ function _http_streaming_request(ctx, method, url, headers, body, timeout, bytes
290290
open_kwargs = merge(open_kwargs, (; verbose=get(ctx.client.clntoptions, :verbose, false)))
291291
end
292292

293+
# Capture the streaming connection so the abort-on-close watcher below can
294+
# unblock the read task when the consumer closes `stream_to`. (Mirrors the
295+
# `:downloads` backend, which interrupts its download task on channel close.)
296+
io_ref = Ref{Any}(nothing)
297+
293298
@sync begin
294-
@async begin
299+
read_task = @async begin
295300
try
296301
HTTP.open(method, url, headers; open_kwargs...) do io
302+
io_ref[] = io
297303
write(io, body)
298304
captured_response[] = http_response = startread(io)
299305
try
@@ -311,7 +317,15 @@ function _http_streaming_request(ctx, method, url, headers, body, timeout, bytes
311317
end
312318
catch ex
313319
close(output)
314-
rethrow(ex)
320+
# When the consumer closes `stream_to`, the watcher task below aborts
321+
# this read (close(io) + a scheduled InterruptException); that surfaces
322+
# here as an IO error or InterruptException. Swallow it so the streaming
323+
# request returns normally instead of propagating a spurious error. A
324+
# read timeout or genuine network error arrives while `stream_to` is
325+
# still open (and is not our InterruptException), so it still throws.
326+
if isopen(stream_to) && !isa(ex, InterruptException)
327+
rethrow(ex)
328+
end
315329
end
316330
end
317331

@@ -337,6 +351,45 @@ function _http_streaming_request(ctx, method, url, headers, body, timeout, bytes
337351
close(stream_to)
338352
end
339353
end
354+
355+
@async begin
356+
# Abort-on-close watcher: when the consumer closes `stream_to` (e.g. a
357+
# k8s watch is stopped via `close(stream)`), abort the read task above so
358+
# it unblocks immediately instead of hanging on the socket until the
359+
# read-idle timeout. Mirrors the `:downloads` backend, which interrupts
360+
# its download task on channel close.
361+
try
362+
while isopen(stream_to)
363+
wait(stream_to)
364+
yield()
365+
end
366+
catch ex
367+
isa(ex, InvalidStateException) || rethrow(ex)
368+
end
369+
# Best-effort close of the connection. On HTTP/2 this alone does NOT wake a
370+
# body read parked on the flow-control timer, so we also forcibly interrupt
371+
# the read task (as `:downloads` does for non-interruptible downloads).
372+
io = io_ref[]
373+
# `io` may be `nothing` if the consumer closed before the connection was
374+
# established; in practice a consumer only stops after receiving an event
375+
# (or a timer fires seconds later), so the connection is already up. This
376+
# is the same theoretical hole the `:downloads` watcher has.
377+
if io !== nothing
378+
try
379+
close(io)
380+
catch
381+
# already closed / natural EOF — nothing to abort
382+
end
383+
end
384+
# `stream_to` also closes on natural EOF (the chunk reader closes it after
385+
# the read task finishes); the `istaskdone` guard skips the interrupt in
386+
# that case. For JuliaRun's watch consumers an interrupt of an already-
387+
# finishing read is harmless anyway — it maps to `is_request_interrupted`,
388+
# the same signal a read-idle timeout produces, which they already retry on.
389+
if !istaskdone(read_task)
390+
schedule(read_task, InterruptException(); error=true)
391+
end
392+
end
340393
end
341394

342395
return http_response, output

0 commit comments

Comments
 (0)