Handling failures
In pipeflow: Fast Interactive Data Analysis Pipelines

knitr::opts_chunk$set(
    comment = "#",
    prompt = FALSE,
    tidy = FALSE,
    cache = FALSE,
    collapse = TRUE
)

old <- options(width = 100L)

In practice, steps fail: data is missing, a model does not converge, or an input violates an assumption. This vignette shows how {pipeflow} reports such failures, how a run can be continued for unaffected steps, and how to pick up the pieces once the cause is fixed.

A pipeline that fails

We use a small pipeline whose check step fails if the data exceeds a limit.

library(pipeflow)

pip <- pip_new("analysis") |>
    pip_add("data", \(n = 5) seq_len(n)) |>
    pip_add("check", \(x = ~data, limit = 3) {
        if (max(x) > limit) stop("data exceeds limit")
        x
    }) |>
    pip_add("model", \(x = ~check) mean(x)) |>
    pip_add("total", \(x = ~data) sum(x))

With n = 5 and limit = 3, the check step will fail, and the model step downstream of it cannot run. The total step, however, only needs the data step.

Stop at the first error

By default, pip_run() stops at the first step that fails.

pip_run(pip)

pip

pip_run_state(pip)

The pipeline stays in a defined state. The failed step is marked as failed, the steps that were not reached are still new, and the run state indicates that the last run ended with an error.

Let's inspect the error object returned by tryCatch().

err <- tryCatch(pip_run(pip, lgr = NULL), error = \(e) e)

class(err)
err$step

class(err$parent)
err$parent$message

The error is of class pipeflow_step_error. Its field step names the failed step and the field parent holds the original condition, which keeps its own class, so that errors can be handled by class:

Continue after errors

In the previous run, the total step was not run, although it does not depend on the failed step, because the run stopped early. With on_error = "continue" in pip_run(), a failing step is still marked as failed but the run can continue for all other steps that do not depend on it:

pip$reset()

suppressWarnings(pip_run(pip, on_error = "continue"))

pip

pip_run_state(pip)

Now total was calculated, while model was skipped, because its input depends on the failed step. The out column of a failed step holds the condition, which is printed in short form as <kind/class>.

To work with it programmatically, we can extract it as usual:

out <- pip[["check", "out"]]

class(out)
out$message

Let's inspect the warning object returned by tryCatch().

w <- tryCatch(
    pip_run(pip, lgr = NULL, on_error = "continue"),
    warning = \(w) w
)

names(w$failed)
w$skipped

The warning is signalled after all states and outputs are stored, so catching it with tryCatch() does not lose any results.

Mitigate failed inputs with `try()`

Sometimes a step should run even if one of its inputs failed, for example a report that mentions what is missing or if a fallback is available. Such an argument is marked by wrapping the step reference in try():

pip_add(
    pip,
    "report",
    \(total = ~total, m = ~try(model)) {
        if (inherits(m, "pipeflow_failure")) {
            paste("total:", total, "- no model:", m$message)
        } else {
            paste("total:", total, "- mean:", m)
        }
    }
)

suppressWarnings(pip_run(pip, on_error = "continue"))

pip[["report", "out"]]

Instead of the output of model, the m argument received a failure object, a condition of class pipeflow_failure, which stands in for the missing output. The message tells why the input is not available. It has the fields

step: the step that provided the input (model),
failed_step: the step that actually failed (check), and
parent: the original error condition.

A step is only run despite a failed input if all its failed inputs are marked with try(). Everything else is skipped, as model above. Note that try() has an effect only in a continuing run. With on_error = "stop", the run ends at the failed step anyway.

Recover from failures

Both skipped and failed steps are retried in the next run, and so are steps that got a failure object. Steps that were completed are not run again. Let's fix the cause by raising the limit and re-run:

pip_set_params(pip, list(limit = 10))
pip_run(pip)

pip

The run log shows that data and total were skipped, since they were completed already. The failed step check and the skipped step model were executed now, and so was report, because it had received a failure object instead of the real output. The report now contains the model result and the run state is ready again.

pip[["report", "out"]]

pip_run_state(pip)

Failures in split pipelines

Failures also work with the execution modes split and reduce, which are described in the vignette on split, map, and reduce. Let's split a small data set by group and let the spread step fail for groups with a single row.

d <- data.frame(g = c("a", "a", "b", "b", "c"), y = c(1, 2, 3, 5, 8))

pip <- pip_new("groups") |>
    pip_add("data", \(data = NULL) data) |>
    pip_add("by_group", \(data = ~data) split(data, data$g), exec = "split") |>
    pip_add("spread", \(d = ~by_group) {
        if (nrow(d) < 2) stop("need at least 2 rows")
        diff(range(d$y))
    }) |>
    pip_add(
        "combine",
        \(s = ~try(spread)) {
            if (inherits(s, "pipeflow_failure")) {
                return(s$message)
            }
            unlist(s)
        },
        exec = "reduce"
    )

pip_set_params(pip, list(data = d))

If one partition (group c) fails, the whole spread step fails. The condition tells the failed key and keeps the original one as parent:

suppressWarnings(pip_run(pip, on_error = "continue"))

pip

err <- pip[["spread", "out"]]
class(err)
err$key
err$parent$message

The combine step reduces the outputs of all groups. Since the input spread failed and is marked with try(), combine ran once and received the failure object, with the key in the message:

pip[["combine", "out"]]

Likewise, if the pipeline is run with on_error = "stop", the error is a pipeflow_step_error as before, with the pipeflow_key_error as its parent. Without try() in the combine step, it would just be skipped.

Failures are currently tracked per step, not per key, so the groups a and b that were fine are not available as results of a run that failed in group c. Once the cause is fixed, here by removing the group, the pipeline completes:

pip_set_params(pip, list(data = d[d$g != "c", ]))
pip_run(pip)

pip[["combine", "out"]]

Summary

By default, a failing step stops the run with a pipeflow_step_error that names the step and holds the original condition in parent.
pip_run(on_error = "continue") marks the step as failed, stores its condition as output, skips the steps that depend on it, and runs everything else. The run state becomes continued, and a pipeflow_run_failed warning lists both failed and skipped steps.
A reference wrapped in try(), e.g. x = ~try(step), makes a step run despite a failed input. The argument then receives a pipeflow_failure condition.
The next run retries any failed and skipped steps (and those that received failure objects).
For steps that map over partitions, the condition of the failed partition is a pipeflow_key_error with the field key. A reduce step with a try() reference receives the failure.