---
title: "Handling failures"
output:
  rmarkdown::html_vignette:
    toc: true
    toc_depth: 4
description: >
  Shows what happens when a step fails, how to keep a pipeline running with
  on_error = "continue", how to react to failed inputs with ~try() references,
  and how to recover once the cause is fixed.
vignette: >
  %\VignetteIndexEntry{Handling failures}
  %\VignetteEngine{knitr::rmarkdown}
  %\VignetteEncoding{UTF-8}
---

```{r knitr-setup, include = FALSE}
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.

```{r define-pipeline}
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.

```{r run-and-stop, error = TRUE}
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()`.

```{r step-error-fields}
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:

```{r run-and-continue, warning = TRUE}
pip_reset(pip)

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:

```{r failed-output}
out <- pip[["check", "out"]]

class(out)
out$message
```

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

```{r run-failed-warning}
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()`:

```{r add-report-step}
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:

```{r recover}
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.

```{r recovered-report}
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](v05a-split-map-reduce.html). Let's split a small
data set by group and let the `spread` step fail for groups with a single row.

```{r define-split-pipeline}
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("label", \(s = ~spread) sprintf("spread: %g", s)) |>
    pip_add(
        "combine",
        \(s = ~try(spread)) {
            failed <- vapply(
                s,
                FUN = inherits,
                what = "pipeflow_failure",
                FUN.VALUE = logical(1)
            )
            unlist(s[!failed])
        },
        exec = "reduce"
    )

pip_set_params(pip, list(data = d))
```

If one partition (group `c`) fails, `spread` fails for that key only. It
gets state `failed`, but its output keeps the results of the groups `a` and
`b`. For group `c` it holds a failure object, which tells the key and keeps
the original condition as `parent` of a `pipeflow_key_error`:

```{r run-split-pipeline}
suppressWarnings(pip_run(pip, on_error = "continue"))

pip

out <- pip[["spread", "out"]]
out$a
fc <- out$c
class(fc)
fc$key
fc$message
```

The `label` step maps over `spread` like `spread` maps over the groups, so
the failure is handled per key: `label` runs for groups `a` and `b`, and is
not run for group `c`, whose output is again a failure object:

```{r map-with-failure}
pip[["label", "out"]]$a
pip[["label", "out"]]$c$message
```

The `combine` step reduces the outputs of all groups in one call. Since it
takes `spread` via `try()`, it ran and received all of them, with the
failure in group `c`:

```{r reduce-with-failure}
pip[["combine", "out"]]
```

Without `try()`, `combine` would not be run, as for any failed input. With
`s = ~try(spread)` in `label`, `label` would be called for group `c` too and
receive the failure object. If the pipeline is run with
`on_error = "stop"`, the first failed group stops the run with a
`pipeflow_step_error`, with the `pipeflow_key_error` as its `parent`.

If nothing changed in the pipeline since, the next run calls `spread` only
for the failed group `c` and keeps the results of the other groups, which
helps if the cause was outside of the pipeline, e.g. a lost connection. Here
the cause is in the data, so we fix it by removing the group. That changes
the input of all groups, so `spread` is run for all of them, and the
pipeline completes:

```{r recover-split-pipeline}
pip_set_params(pip, list(data = d[d$g != "c", ]))
pip_run(pip)

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


### Failures in nested pipelines

A pipeline can be turned into a function with `as.function()` and added as a
step to another pipeline, as described in the vignette on
[nested pipelines](v05b-nested-pipeline.html). Such a step runs a pipeline of
its own, which means it also has its own error handling.

Let's put the `spread` step of the previous section into an inner pipeline
that reports on it, and map that pipeline over the groups:

```{r define-nested-pipeline}
inner <- pip_new("inner") |>
    pip_add("data", \(data = NULL) data) |>
    pip_add("spread", \(d = ~data) {
        if (nrow(d) < 2) stop("need at least 2 rows")
        diff(range(d$y))
    }) |>
    pip_add("report", \(s = ~try(spread)) {
        if (inherits(s, "pipeflow_failure")) "no spread" else s
    })

nested <- pip_new("nested") |>
    pip_add("data", \(data = NULL) data) |>
    pip_add("by_group", \(data = ~data) split(data, data$g), exec = "split") |>
    pip_add(
        "per_group",
        as.function(inner, out = "report"),
        params = list(data = ~by_group)
    )

pip_set_params(nested, list(data = d))
```

By default, the inner pipeline is run with `on_error = "stop"`, so it ends at
the first failed inner step and the outer step fails with it. The `try()`
reference of the `report` step has no effect here, because `try()` only
matters in a *continuing* run:

```{r run-nested-stop, error = TRUE}
pip_run(nested)
```

The message names the outer step, the group and the inner step. Each of the
conditions keeps the one of the level below as its `parent`, so these can
also be read from the error:

```{r nested-error-chain}
e <- tryCatch(pip_run(nested, lgr = NULL), error = identity)
e$step               # the outer step
e$parent$key         # the group
e$parent$parent$step # the inner step

original <- e$parent$parent$parent
conditionMessage(original)
```

With `as.function(on_error = "continue")`, the inner pipeline is run as
`pip_run(on_error = "continue")` runs it: inner steps that do not depend on
the failed step still run, and `~try(spread)` gives the `report` step the
failure object. Group `c` is handled inside the inner pipeline, and the outer
run completes:

```{r run-nested-continue}
pip_replace(
    nested,
    "per_group",
    as.function(inner, out = "report", on_error = "continue"),
    params = list(data = ~by_group)
)

pip_run(nested)

nested[["per_group", "out"]]
```

The warning of the continued inner run is passed on like any other warning of
a step, so the outer run logs it and the result of every group is available.

The function still fails if one of the steps named in `out` failed or was not
run because an input failed: the inner pipeline handles what it can, and
everything else is left to the outer run, as before.

```{r run-nested-requested, error = TRUE}
pip_replace(
    nested,
    "per_group",
    as.function(
        inner,
        out = "spread",
        on_error = "continue"           # <--- this time we continue
    ),
    params = list(data = ~by_group)
)

pip_run(nested, lgr = NULL)
```

Since `out = NULL` requests the output of *all* inner steps, name the steps
you need in `out` to profit from a continuing inner run.


### 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).
* A step that maps over partitions fails per key: its output keeps the
  results of the other keys, and each failed key holds a failure object
  with the field `key`. Steps mapped downstream are not run for that key
  (unless they use `try()`), and a `reduce` step gets the output with the
  failed keys if it uses `try()`, and is skipped otherwise.
* A step made with `as.function()` runs a pipeline of its own.
  `as.function(on_error = "continue")` lets that inner pipeline continue
  after a failed inner step, which is also what makes `try()` references
  inside it work. The step still fails if one of the steps named in `out`
  failed or was not run.

```{r, include = FALSE}
options(old)
```