Sch: Propagate a failure to dependents before deleting the failed task

src/sch/Sch.jl

@@ -1162,8 +1162,10 @@ function finish_task!(ctx, state, node, thunk_failed, ready::Vector{Thunk})
         # permanently unresolved.  Handle the three steps directly instead.
         fill_registered_futures!(state, node, true)
         node.sch_accessible = false
-        delete_unused_task!(state, node)
+        # Dependents read this error with load_result, so propagate before
+        # the result can be deleted
         schedule_dependents!(state, node, true, ready)
+        delete_unused_task!(state, node)
     else
         # Success path: seal dependents (collecting newly-ready ones into
         # `ready` for the caller to schedule outside state.lock), fulfill
@@ -1197,9 +1199,8 @@ function delete_unused_task!(state, thunk; recycle::Bool=false)
 end
 # `recycle=true` returns the Thunk to the global pool. Only callers for whom
 # the futures AND dependents Treiber lists are already sealed-and-drained may
-# pass it (success-path finish_task! and unref_thunk!): the failure paths run
-# schedule_dependents! AFTER deletion, and a recycled thunk's re-SEALED
-# dependents head would silently skip failure propagation.
+# pass it (success-path finish_task! and unref_thunk!): a recycled thunk's
+# re-SEALED dependents head would silently skip failure propagation.
 function task_delete!(state, thunk; recycle::Bool=false)
     clear_result!(state, thunk)
     @atomic thunk.valid = false

src/sch/util.jl

@@ -346,7 +346,6 @@ function set_failed!(state, origin::Thunk, thunk::Thunk=origin; ex=nothing)
 
     fill_registered_futures!(state, thunk, true)
     thunk.sch_accessible = false
-    delete_unused_task!(state, thunk)
 
     # Seal the dependents list and propagate failure transitively.
     # schedule_dependents! with failed=true immediately calls set_failed! for
@@ -358,6 +357,9 @@ function set_failed!(state, origin::Thunk, thunk::Thunk=origin; ex=nothing)
     failed_ready = Thunk[]
     schedule_dependents!(state, thunk, true, failed_ready)
     @assert isempty(failed_ready) "set_failed! produced ready thunks on the failure path"
+
+    # Only now: each dependent above read this thunk's error with load_result
+    delete_unused_task!(state, thunk)
 end
 
 "Internal utility, useful for debugging scheduler state."

test/thunk.jl

@@ -271,6 +271,18 @@ end
             c = @spawn b+2
             @test_throws_unwrap (Dagger.DTaskFailedException, ErrorException) fetch(c)
         end
+        @testset "dependent chain, only the last task referenced" begin
+            # The upstream tasks can be deleted as soon as they fail
+            function chain()
+                s = @spawn sleep(1)
+                a = @spawn (s->error("Test"))(s)
+                b = @spawn a+1
+                return @spawn b+2
+            end
+            c = chain()
+            GC.gc(); GC.gc()
+            @test_throws_unwrap (Dagger.DTaskFailedException, ErrorException) fetch(c)
+        end
         @testset "single input" begin
             a = @spawn 1+1
             b = @spawn (a->error("Test"))(a)