Sch: Fix task error-related assertion

Author: jpsamaroo

src/sch/Sch.jl

@@ -610,7 +610,22 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
         end
         task = pop!(state.ready)
         timespan_start(ctx, :schedule, task.id, (;thunk_id=task.id))
-        @assert !haskey(state.cache, task)
+        if haskey(state.cache, task)
+            if haskey(state.errored, task)
+                # An error was eagerly propagated to this task
+                finish_failed!(state, task)
+            else
+                # This shouldn't have happened
+                iob = IOBuffer()
+                println(iob, "Scheduling inconsistency: Task being scheduled is already cached!")
+                println(iob, "  Task: $(task.id)")
+                println(iob, "  Cache Entry: $(typeof(state.cache[task]))")
+                ex = SchedulingException(String(take!(iob)))
+                state.cache[task] = ex
+                state.errored[task] = true
+            end
+            @goto pop_task
+        end
         opts = merge(ctx.options, task.options)
         sig = signature(task, state)
 

src/sch/util.jl

@@ -145,14 +145,17 @@ function set_failed!(state, origin, thunk=origin)
     filter!(x->x!==thunk, state.ready)
     state.cache[thunk] = ThunkFailedException(thunk, origin, state.cache[origin])
     state.errored[thunk] = true
+    finish_failed!(state, thunk, origin)
+end
+function finish_failed!(state, thunk, origin=nothing)
     fill_registered_futures!(state, thunk, true)
     if haskey(state.waiting_data, thunk)
         for dep in state.waiting_data[thunk]
             haskey(state.waiting, dep) &&
                 delete!(state.waiting, dep)
             haskey(state.errored, dep) &&
                 continue
-            set_failed!(state, origin, dep)
+            origin !== nothing && set_failed!(state, origin, dep)
         end
         delete!(state.waiting_data, thunk)
     end