Ensure failed nodes are deleted from worker graph

The worker's local task graph should be cleaned up by removing nodes after they
succeed or fail, including the children of those nodes. We had a subtle edge
case where if marking a *child* task as failed returned Err(...) -- for example
due to the server being unavailable -- we would fail to actually remove the
parent from the graph, instead early exiting (indeed, killing an entire worker
thread).

This commit instead logs, but otherwise ignores, errors on marking tasks as
failed on the server: if we don't succeed, we may be forced to rerun the task
later as the server doesn't know of the failure. This failure *should* be
transient though, so it should be OK if we end up rerunning the task as a
result. I suspect that we don't 100% handle this situation ideally today -- for
example, if a task *always* results in some failure on the server, we will loop
indefinitely trying to run it. But I'm not sure why that would happen, and it
seems like something that the server is better suited than worker nodes to
handle (especially since worker nodes are ideally transient state-wise).

The server (likely after the recent mutex removal) will currently return
'database is busy' errors on some requests, which is a separate bug from the one
being fixed here, but exacerbates the situations in which this bug arises.

Author: Mark-Simulacrum

src/runner/graph.rs

@@ -223,6 +223,7 @@ impl TasksGraph {
             .graph
             .neighbors_directed(node, Direction::Incoming)
             .collect::<Vec<_>>();
+        let mut last_err = None;
         for child in children.drain(..) {
             // Don't recursively mark a child as failed if this is not the only parent of the child
             let parents = self
@@ -237,7 +238,14 @@ impl TasksGraph {
                 );
                 continue;
             }
-            self.mark_as_failed(child, ex, db, state, config, error, result, worker)?;
+            let res = self.mark_as_failed(child, ex, db, state, config, error, result, worker);
+            if let Err(err) = res {
+                // Regardless of whether this child failed, we still want to
+                // (try to) mark the other children as failed as well as the
+                // current node.
+                log::error!("Failure to mark child as failed; skipping: {:?}", err);
+                last_err = Some(err);
+            }
         }
 
         // We need to mark_as_completed the node here (if it's a task),
@@ -257,6 +265,14 @@ impl TasksGraph {
 
         self.mark_as_completed(node);
 
+        // If we would have treated this as a successful mark_as_failed, then
+        // don't.
+        if res.is_ok() {
+            if let Some(err) = last_err {
+                return Err(err);
+            }
+        }
+
         res
     }
 

src/runner/worker.rs

@@ -93,7 +93,7 @@ impl<'a, DB: WriteResults + Sync> Worker<'a, DB> {
                             }
                         }
 
-                        guard.mark_as_failed(
+                        if let Err(e) = guard.mark_as_failed(
                             id,
                             self.ex,
                             self.db,
@@ -102,7 +102,13 @@ impl<'a, DB: WriteResults + Sync> Worker<'a, DB> {
                             &e,
                             result,
                             &self.name,
-                        )?;
+                        ) {
+                            // We don't return an Err(...) from the loop here,
+                            // as this failure shouldn't bring down this worker
+                            // thread -- it can continue to do useful work.
+                            error!("Failed to mark node {:?} as failed", id);
+                            utils::report_failure(&e);
+                        }
                     } else {
                         guard.mark_as_completed(id);
                     }