Fix silent watch stalls: JSON.jl 1.x event objects + fail-fast on stream processor death

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"

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)
 

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

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.