From 35118f62b230bbf7dc06c344b17ba4ba439e1121 Mon Sep 17 00:00:00 2001 From: krynju Date: Fri, 3 Jul 2026 16:45:13 +0200 Subject: [PATCH 1/2] fix: JSON.jl 1.x watch events + fail-fast on stream processor death MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related fixes for silent k8s watch stalls: 1. `kuber_obj` / `_kuber_obj` accepted only `Dict{String,Any}`. Under JSON.jl 1.x, watch event objects parse to `JSON.Object{String,Any}` (an `AbstractDict`), so the first event of every watch threw a MethodError inside the stream processor. Widen to `AbstractDict` — `convert(::Type{<:APIModel}, j)` already handles any AbstractDict. Follow-up to the JSON 1.x support in #66, which missed this entry point. 2. `watch(streamprocessor, ctx, watched, ...)` ran the processor in a bare `@async` under `@sync`. If the processor threw (e.g. the MethodError above), its task died but `@sync` kept waiting on the still-running HTTP watch task — a deaf watch that surfaces no error until the apiserver drops the connection (potentially 30+ minutes), while events pile up unconsumed in the buffered stream. Close the stream in a `finally` on the processor task too — symmetric with the watcher task — so processor death aborts the watch promptly and the caller sees the real exception. This mirrors client-go's reflector, where any watchHandler error exits ListAndWatch and triggers a fresh relist+rewatch. Diagnosed live: JobLoops' pod reflector went permanently stale (jobs stuck in Submitted while pods ran) because every watch stream died silently on its first event under JSON 1.x. Co-Authored-By: Claude Fable 5 --- Project.toml | 2 +- src/helpers.jl | 4 ++-- src/simpleapi.jl | 11 ++++++++++- 3 files changed, 13 insertions(+), 4 deletions(-) diff --git a/Project.toml b/Project.toml index d0c229b..9919848 100644 --- a/Project.toml +++ b/Project.toml @@ -4,7 +4,7 @@ authors = ["JuliaHub Inc."] keywords = ["kubernetes", "client"] license = "MIT" desc = "Julia Kubernetes Client" -version = "0.7.9" +version = "0.7.10" [deps] Dates = "ade2ca70-3891-5945-98fb-dc099432e06a" diff --git a/src/helpers.jl b/src/helpers.jl index d7eb2da..51dd737 100644 --- a/src/helpers.jl +++ b/src/helpers.jl @@ -243,9 +243,9 @@ end # OpenAPI conversions insist that JSONs objects are always `Dict{String,Any}`. # To ensure that for a user supplied Dict, we serialize that to string and parse it back as json. -kuber_obj(ctx::KuberContext, j::Dict{String,Any}) = kuber_obj(ctx, JSON.json(j)) +kuber_obj(ctx::KuberContext, j::AbstractDict) = kuber_obj(ctx, JSON.json(j)) kuber_obj(ctx::KuberContext, data::String) = _kuber_obj(ctx, _parse_json(data)) -_kuber_obj(ctx::KuberContext, j::Dict{String,Any}) = convert(kind_to_type(ctx, j["kind"], get(j, "apiVersion", nothing)), j) +_kuber_obj(ctx::KuberContext, j::AbstractDict) = convert(kind_to_type(ctx, j["kind"], get(j, "apiVersion", nothing)), j) show(io::IO, ctx::KuberContext) = print(io, "Kubernetes namespace ", ctx.namespace, " at ", ctx.client.root) diff --git a/src/simpleapi.jl b/src/simpleapi.jl index 04943fa..c20d15e 100644 --- a/src/simpleapi.jl +++ b/src/simpleapi.jl @@ -40,7 +40,16 @@ function watch(streamprocessor::Function, ctx::KuberContext, watched::Function, finally close(stream) end - @async streamprocessor(stream) + @async try + streamprocessor(stream) + finally + # Symmetric to the watcher task above: if the stream processor + # dies (e.g. an exception while converting an event), close the + # stream so the HTTP watch task aborts too. Otherwise `@sync` + # silently waits for the connection to end while events pile up + # unconsumed — a deaf watch with no error surfaced. + close(stream) + end end end end From 75a620fb6142afc6880c2688df1ed19af0744f05 Mon Sep 17 00:00:00 2001 From: krynju Date: Fri, 3 Jul 2026 17:12:41 +0200 Subject: [PATCH 2/2] test: regression tests for deaf-watch fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Watch Events: convert each received event object with kuber_obj — exercises the AbstractDict path (JSON.Object under JSON.jl 1.x) that previously threw MethodError on the first event of every watch. - New "Watch processor failure aborts watch" testset (no cluster needed): a throwing streamprocessor must propagate its error out of watch() promptly instead of leaving @sync waiting on the producer task forever. Verified to hang on the unfixed code. Co-Authored-By: Claude Fable 5 --- test/runtests.jl | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/test/runtests.jl b/test/runtests.jl index 96851bb..adebc9c 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -314,6 +314,15 @@ function test_versioned(ctx, testid) @test !isempty(events) for event in events @test isa(event, Union{Typedefs.CoreV1.WatchEvent,Typedefs.CoreV1.PodList}) + # Watch event objects parse to `JSON.Object` under JSON.jl 1.x — + # an AbstractDict, not a Dict{String,Any}. `kuber_obj` must accept + # them: a MethodError here used to kill the stream processor and + # leave the watch silently deaf (see the watch-processor-failure + # test below for the propagation side). + if isa(event, Typedefs.CoreV1.WatchEvent) + obj = Kuber.kuber_obj(ctx, event.object) + @test isa(obj, OpenAPI.APIModel) + end end end end @@ -350,6 +359,36 @@ function test_all() end end +function test_watch_processor_failure() + # A `streamprocessor` that throws must abort the watch promptly and + # propagate the error. Before the fix, the processor task died silently + # while `@sync` kept waiting on the (long-running) `watched` task — a + # deaf watch: events kept buffering with no error surfaced until the + # server dropped the connection. + ctx = KuberContext() # only passed through to KuberWatchContext; no server needed + producer = (watchctx) -> begin + # mimic the HTTP watch task: keep streaming events; ends only when the + # stream is closed under it (put! on a closed channel throws) + i = 0 + while true + put!(watchctx.stream, (i += 1)) + sleep(0.05) + end + end + t0 = time() + @test_throws Exception watch(ctx, producer) do stream + take!(stream) + error("processor failure") + end + # must fail fast — processor death closes the stream, which kills the + # producer's next put! — not linger until the producer would have ended + @test (time() - t0) < 10.0 +end + +@testset "Watch processor failure aborts watch" begin + test_watch_processor_failure() +end + test_all() # Shutdown the kubectl proxy if we are running on github CI.