start working on #1311

R/class_builder.R

@@ -339,6 +339,7 @@ builder_handle_error <- function(target, pipeline, scheduler, meta) {
     target$settings$error,
     continue = builder_error_continue(target, scheduler),
     abridge = scheduler$abridge(target),
+    trim = scheduler$trim(target),
     stop = builder_error_exit(target, pipeline, scheduler, meta),
     null = builder_error_null(target, pipeline, scheduler, meta),
     workspace = builder_error_exit(target, pipeline, scheduler, meta)

R/class_options.R

@@ -516,7 +516,7 @@ options_class <- R6::R6Class(
       deprecate_error_workspace(error)
       tar_assert_flag(
         error,
-        c("stop", "continue", "abridge", "workspace", "null")
+        c("stop", "continue", "null", "abridge", "trim", "workspace")
       )
     },
     validate_memory = function(memory) {

R/class_scheduler.R

@@ -22,12 +22,14 @@ scheduler_init <- function(
   )
   reporter <- reporter_init(reporter, seconds_interval = seconds_reporter)
   backoff <- tar_options$get_backoff()
+  canceled <- counter_init()
   scheduler_new(
     graph = graph,
     queue = queue,
     progress = progress,
     reporter = reporter,
-    backoff = backoff
+    backoff = backoff,
+    canceled <- canceled
   )
 }
 
@@ -52,9 +54,10 @@ scheduler_new <- function(
   queue = NULL,
   progress = NULL,
   reporter = NULL,
-  backoff = NULL
+  backoff = NULL,
+  canceled = NULL
 ) {
-  scheduler_class$new(graph, queue, progress, reporter, backoff)
+  scheduler_class$new(graph, queue, progress, reporter, backoff, canceled)
 }
 
 scheduler_class <- R6::R6Class(
@@ -68,18 +71,21 @@ scheduler_class <- R6::R6Class(
     progress = NULL,
     reporter = NULL,
     backoff = NULL,
+    canceled = NULL,
     initialize = function(
       graph = NULL,
       queue = NULL,
       progress = NULL,
       reporter = NULL,
-      backoff = NULL
+      backoff = NULL,
+      canceled = NULL
     ) {
       self$graph <- graph
       self$queue <- queue
       self$progress <- progress
       self$reporter <- reporter
       self$backoff <- backoff
+      self$canceled <- canceled
     },
     count_unfinished_deps = function(name) {
       deps <- self$graph$produce_upstream(name)
@@ -98,12 +104,18 @@ scheduler_class <- R6::R6Class(
       self$progress$abridge()
       self$queue$abridge()
     },
+    trim = function(target) {
+
+      browser()
+
+    },
     validate = function() {
       self$graph$validate()
       self$queue$validate()
       self$progress$validate()
       self$reporter$validate()
       self$backoff$validate()
+      counter_validate(self$canceled)
     }
   )
 )

R/tar_target.R

@@ -211,13 +211,17 @@
 #'   stops and throws an error. Options:
 #'   * `"stop"`: the whole pipeline stops and throws an error.
 #'   * `"continue"`: the whole pipeline keeps going.
+#'   * `"null"`: The errored target continues and returns `NULL`.
+#'     The data hash is deliberately wrong so the target is not
+#'     up to date for the next run of the pipeline.
 #'   * `"abridge"`: any currently running targets keep running,
 #'     but no new targets launch after that.
+#'   * `"trim"`: any currently running targets keep running,
+#'     and everything downstream of the error is canceled.
+#'     In addition, if the error happens in a dynamic branch,
+#'     then all not-yet-dispatched sibling branches are canceled.
 #'   (Visit <https://books.ropensci.org/targets/debugging.html>
 #'   to learn how to debug targets using saved workspaces.)
-#'   * `"null"`: The errored target continues and returns `NULL`.
-#'     The data hash is deliberately wrong so the target is not
-#'     up to date for the next run of the pipeline.
 #' @param memory Character of length 1, memory strategy.
 #'   If `"persistent"`, the target stays in memory
 #'   until the end of the pipeline (unless `storage` is `"worker"`,

R/tar_target_raw.R

@@ -120,7 +120,7 @@ tar_target_raw <- function(
   tar_assert_flag(iteration, c("vector", "list", "group"))
   tar_assert_flag(
     error,
-    c("stop", "continue", "abridge", "workspace", "null")
+    c("stop", "continue", "null", "abridge", "trim", "workspace")
   )
   deprecate_error_workspace(error)
   tar_assert_flag(memory, c("persistent", "transient"))