Fix streaming hangs and enable streaming on Julia 1.10 (#68)

`src/Streaming.jl` was compiled out below Julia 1.12 because streaming calls
hung there. The hang reproduces on a bidirectional stream that moves ~1000
messages with `-t 4`, roughly one run in three on Julia 1.10, and it is not a
Julia version problem: the same wedge, plus a second one at end of stream, also
reproduces on 1.12 given enough runs.

Both come from the same place. read_callback pauses libcurl's send direction
whenever the request buffer runs dry, and each `curl_easy_pause(CURLPAUSE_CONT)`
the request pump issues to resume it is one-shot. libcurl can drop a transfer on
the floor around one of those un-pauses: it stops servicing the transfer's
socket entirely, so the peer's response bytes pile up unread in the receive
queue - 150KB of them in one captured hang, exactly the responses the caller was
still waiting for - while `curl_multi_socket_action` keeps reporting CURLM_OK
without consuming any. Nothing but another un-pause recovers it; driving the
multi handle, by socket event or by timeout, does not. So a caller that waits on
responses before sending more deadlocks until its deadline.

An in-flight request-streaming call now re-issues the un-pause every 50ms while
libcurl holds none of its request data, instead of relying on a single one
landing. `curl_easy_pause` returns immediately when the transfer is not paused,
and an un-pause with an empty request buffer just draws one read_callback that
pauses straight back, so a healthy call pays only for the call itself.

The end-of-stream path had the mirror-image bug, and it was reachable on every
version: it un-paused and only then closed `req.request_c` in its `finally`, so
a read_callback landing in between still saw an open stream, paused again, and
left the transfer paused with nobody to un-pause it - all data exchanged, both
peers done, call hung until the deadline. It now closes the stream first, so the
callback the un-pause draws can only take its end-of-stream branch.

Streaming is enabled on all supported versions, the version guard around the
streaming testsets is gone, and the docs no longer describe streaming as 1.12
only. "Don't Stick User Tasks" is marked broken below 1.12: it is a unary test
that only sat in the guarded block, and the stickiness comes from arming the
deadline watchdog with `Timer(cb, delay)`, whose `@async` marks the scheduling
task sticky on those versions (JuliaLang/julia#41324).

Verified against the Go test server on 1.10.11 (libcurl 8.4.0) and 1.12.6
(libcurl 8.15.0): 1000 bidirectional stress runs across both versions and both
close orderings with no hang, the full suite 12 times per version plus `-t 1`,
`-t 2` and `-t 8` on 1.10, and Runic clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: s-celles

CONTRIBUTING.md

@@ -60,9 +60,7 @@ JULIA_GRPCCLIENT_TEST_START_SERVER=go julia --project test/runtests.jl
 If you already have a server listening on `localhost:8001`, plain `julia --project test/runtests.jl`
 is enough. `GRPC_TEST_SERVER_HOST` and `GRPC_TEST_SERVER_PORT` point the suite elsewhere.
 
-CI covers Julia 1.10, 1.12, and nightly across Linux, Windows, and macOS. Note that
-`src/Streaming.jl` is compiled only on Julia 1.12 and newer, so streaming code and its tests need
-matching version guards.
+CI covers Julia 1.10, 1.12, and nightly across Linux, Windows, and macOS.
 
 ## Pull requests
 

skills/grpcclient-jl-dev/SKILL.md

@@ -13,16 +13,14 @@ description: Work on the gRPCClient.jl package itself. Covers running the Go tes
 | `src/Curl.jl` | The libcurl transport: `gRPCCURL`, `gRPCRequest`, callbacks, socket watchers, concurrency semaphore, deadline watchdog, `grpc_cancel` |
 | `src/gRPC.jl` | Public handle lifecycle, `gRPCServiceClient`, request framing, the generic `grpc_async_await` |
 | `src/Unary.jl` | Unary methods and `gRPCAsyncChannelResponse` |
-| `src/Streaming.jl` | Streaming methods and the request and response pump tasks; included only on Julia 1.12 and newer |
+| `src/Streaming.jl` | Streaming methods and the request and response pump tasks |
 | `src/ProtoBuf.jl` | The ProtoBuf.jl codegen hook that emits `*_Client` constructors |
 | `test/proto/test.proto` | One service with all four RPC variants |
 | `test/gen/` | Checked-in generated stubs used by the suite |
 | `test/go/` | The Go reference server every test and benchmark runs against |
 | `test/python/` | Generated Python stubs, regenerated alongside the Julia ones |
 | `utils/gRPCClientUtils.jl` | Separate package for benchmarks, stress workloads, and memory profiling |
 
-Streaming is conditionally compiled: `src/Streaming.jl` is included only under `@static if VERSION >= v"1.12"`, so anything added there needs a matching version guard in the tests.
-
 ## Formatting
 
 The repository is formatted with [Runic.jl](https://github.com/fredrikekre/Runic.jl), and contributed code is expected to be run through it before a pull request goes up:

skills/grpcclient-jl/SKILL.md

@@ -5,7 +5,7 @@ description: Call gRPC services from Julia with gRPCClient.jl. Covers generating
 
 # gRPCClient.jl
 
-A Julia gRPC client built on libcurl for HTTP/2 transport, with client stubs generated by ProtoBuf.jl. Version 1.0.4. Requires Julia 1.10; streaming RPC requires Julia 1.12 or newer and is compiled out below that.
+A Julia gRPC client built on libcurl for HTTP/2 transport, with client stubs generated by ProtoBuf.jl. Version 1.0.4. Requires Julia 1.10.
 
 Exported names: `grpc_init`, `grpc_shutdown`, `grpc_global_handle`, `grpc_register_service_codegen`, `grpc_async_request`, `grpc_async_await`, `grpc_sync_request`, `grpc_cancel`, `gRPCCURL`, `gRPCRequest`, `gRPCServiceClient`, `gRPCAsyncChannelResponse`, `gRPCException`, `gRPCServiceCallException`.
 
@@ -89,7 +89,7 @@ The channel form never throws at submission time; a failed request arrives as a
 
 ## Streaming RPC
 
-Requires Julia 1.12 or newer. Requests and responses move through `Channel`s, and the streaming side of the call is chosen by the RPC definition, not by the call site:
+Requests and responses move through `Channel`s, and the streaming side of the call is chosen by the RPC definition, not by the call site:
 
 | RPC variant | Call shape | `grpc_async_await` |
 |---|---|---|
@@ -131,7 +131,6 @@ Which call raises matters when writing error handling. `grpc_async_request` thro
 | `MethodError` on a keyword named `max_receive_message_length` | The keyword is `max_recieve_message_length` |
 | `ArgumentError: Invalid field of gRPCConnectionOptions: x` | Unknown per-call option keyword, most often a misspelling |
 | `DEADLINE_EXCEEDED` under load with a healthy server | The 10 second default deadline includes time queued for one of the handle's 16 concurrency slots; see `references/concurrency.md` |
-| Streaming functions undefined | Julia is older than 1.12; a warning was emitted at load |
 | A client or bidi stream never finishes | The request channel was never closed, so the server is still waiting for end of stream |
 | `INVALID_ARGUMENT` mentioning the authorization header | `token` and an `authorization` metadata entry were both set |
 | Call hangs forever | `deadline = Inf` with no `grpc_cancel`, which is the documented behavior of that setting |

skills/grpcclient-jl/references/streaming.md

@@ -1,7 +1,5 @@
 # Streaming RPC
 
-Requires Julia 1.12 or newer. On older versions `src/Streaming.jl` is not included at all, a warning is emitted at load time, and the streaming `grpc_async_request` methods simply do not exist, so the failure looks like a `MethodError` rather than a version check.
-
 Every streaming variant follows the same rhythm: create the channels, start the request, move messages, then await for errors. `grpc_async_request` returns a `gRPCRequest` immediately in all three cases.
 
 ## Client streaming, many requests to one response

src/Curl.jl

@@ -343,6 +343,11 @@ end
 # watchdog only wins when libcurl has wedged (see the watchdog comment in gRPCRequest).
 const GRPC_DEADLINE_GRACE = 0.25
 
+# How often (seconds) an in-flight request-streaming call re-issues
+# `curl_easy_pause(CURLPAUSE_CONT)` while libcurl has our request buffer drained. See
+# the `unpause` field in gRPCRequest for why this is needed.
+const GRPC_UNPAUSE_INTERVAL = 0.05
+
 mutable struct gRPCRequest
     # CURL multi lock for exclusive access to the easy handle after its added to the multi
     lock::ReentrantLock
@@ -407,6 +412,30 @@ mutable struct gRPCRequest
     # Client-side deadline watchdog, see the comment in the constructor
     timer::Union{Nothing, Timer}
 
+    # Repeating un-pause for a request-streaming call, `nothing` for every other kind.
+    #
+    # read_callback pauses libcurl's send direction whenever the request buffer runs
+    # dry, which for a streaming request is every time the caller has nothing queued,
+    # and `curl_easy_pause(CURLPAUSE_CONT)` from the request pump is what resumes it.
+    # Every un-pause the pump issues is one-shot, and libcurl can drop the transfer on
+    # the floor around one: it stops servicing the transfer's socket entirely, so the
+    # peer's response bytes pile up unread in the receive queue while
+    # curl_multi_socket_action keeps reporting CURLM_OK without consuming any of them.
+    # Nothing but another un-pause shakes it loose - driving the multi handle, by socket
+    # event or by timeout, does not - so a bidirectional stream whose caller is waiting
+    # on responses before it sends anything more deadlocks until its deadline expires.
+    # This is what made streaming unusable on Julia 1.10, whose LibCURL_jll is pinned to
+    # the 8.4 series: https://github.com/JuliaIO/gRPCClient.jl/issues/68
+    #
+    # So rather than rely on a single un-pause landing, keep re-issuing it for as long as
+    # such a call is in flight and libcurl holds none of our request data. That covers
+    # the end of the stream too, where the un-pause is what draws the read_callback that
+    # returns 0 and closes the request. curl_easy_pause returns immediately without
+    # doing anything when the transfer is not paused, and an un-pause with an empty
+    # request buffer just draws one read_callback that pauses straight back, so a healthy
+    # call pays only for the call itself while a stalled one recovers within an interval.
+    unpause::Union{Nothing, Timer}
+
     # Absolute `time()` at which this request's deadline expires, Inf when there is no
     # deadline. grpc_async_await uses it to attribute a transport error that landed after
     # the deadline to DEADLINE_EXCEEDED rather than INTERNAL: when libcurl tears a
@@ -637,10 +666,37 @@ mutable struct gRPCRequest
             grpc,
             false,
             watchdog,
+            nothing,
             expiry,
         )
         preserve_handle(req)
 
+        # Arm the repeating un-pause once `req` exists, see the `unpause` field above.
+        # Only a request-streaming call ever pauses, so nothing else needs one.
+        if isstreaming_request(req)
+            req.unpause = Timer(
+                GRPC_UNPAUSE_INTERVAL;
+                interval = GRPC_UNPAUSE_INTERVAL,
+            ) do _
+                try
+                    lock(grpc.lock) do
+                        # cleanup_request runs under this lock and frees the easy
+                        # handle, so re-check both before touching it
+                        (req.completed || !grpc.running) && return
+                        # Only once libcurl has taken every byte of the request buffer,
+                        # which is exactly when there is nothing in flight for the
+                        # un-pause to disturb. While the pump has a batch staged the
+                        # buffer is non-empty, so this stays out of the handoff the pump
+                        # is in the middle of driving itself.
+                        req.request_ptr == req.request.size &&
+                            curl_easy_pause(req.easy, CURLPAUSE_CONT)
+                    end
+                catch err
+                    @error("streaming un-pause: unexpected error", err, maxlog = 1_000)
+                end
+            end
+        end
+
         req_p = pointer_from_objref(req)
         curl_easy_setopt(easy_handle, CURLOPT_PRIVATE, req_p)
 
@@ -671,6 +727,7 @@ mutable struct gRPCRequest
                 # shut-down handle is a submission-time FAILED_PRECONDITION per the
                 # contract at the top of this constructor.
                 isnothing(watchdog) || close(watchdog)
+                isnothing(req.unpause) || close(req.unpause)
                 curl_easy_cleanup(easy_handle)
                 curl_slist_free_all(headers)
                 unpreserve_handle(req)
@@ -747,6 +804,7 @@ mutable struct gRPCRequest
             grpc,
             true,
             nothing,
+            nothing,
             expiry,
         )
 
@@ -1307,6 +1365,9 @@ function cleanup_request(grpc::gRPCCURL, req::gRPCRequest)
     req.completed = true
     # Stop the deadline watchdog
     isnothing(req.timer) || close(req.timer)
+    # Stop the repeating streaming un-pause. It is a *repeating* Timer, so leaving it
+    # open would keep the libuv event loop alive for the rest of the process.
+    isnothing(req.unpause) || close(req.unpause)
     # First remove from the multi
     curl_multi_remove_handle(grpc.multi, req.easy)
     # Cleanup the easy handle

src/Streaming.jl

@@ -68,6 +68,15 @@ function grpc_async_stream_request(
             # wakes; the re-check under req.lock keeps curl_easy_pause off the freed
             # easy handle).
             if !req.completed
+                # Mark the request stream closed before the un-pause below, not just in
+                # the `finally` further down. read_callback only returns 0 (which is what
+                # ends the request) once it sees this closed, and the un-pause is what
+                # draws that callback: with the close left until afterwards, a callback
+                # landing in between still sees an open stream, pauses again, and the
+                # transfer sits paused with nobody left to un-pause it, so the call hangs
+                # until its deadline even though both sides are done.
+                close(req.request_c)
+
                 # Wait for any request data to be flushed by curl
                 wait(req.curl_done_reading)
 

src/gRPCClient.jl

@@ -85,13 +85,7 @@ include("Curl.jl")
 include("gRPC.jl")
 include("Unary.jl")
 
-# Streaming only supported on >= 1.12
-@static if VERSION >= v"1.12"
-    include("Streaming.jl")
-else
-    @warn "Julia $(VERSION) <= 1.12, streaming support is disabled: https://github.com/JuliaIO/gRPCClient.jl/issues/68"
-end
-
+include("Streaming.jl")
 include("ProtoBuf.jl")
 
 export grpc_init