Fix bug where domains aren't re-entered when handlers are executed

DESCRIPTION

@@ -41,4 +41,4 @@ Config/Needs/website: rsconnect
 Encoding: UTF-8
 Language: en-US
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.1
+RoxygenNote: 7.3.2

R/domains.R

@@ -48,11 +48,6 @@ promiseDomain <- list(
       stop("A single `then` call cannot combine `onFinally` with `onFulfilled`/`onRejected`")
     }
 
-    # TODO: All wrapped functions should also be rewritten to reenter the domain
-    # jcheng 2019-07-26: Actually, this seems not to be necessary--the domain
-    # is getting reentered during callbacks. But I can't figure out now how it's
-    # happening.
-
     domain <- current_promise_domain()
 
     shouldWrapFinally <- !is.null(onFinally) && !is.null(domain) && !is.null(domain$wrapOnFinally)
@@ -76,7 +71,21 @@ promiseDomain <- list(
       onFulfilled = if (shouldWrapFulfilled) domain$wrapOnFulfilled(onFulfilled) else onFulfilled,
       onRejected = if (shouldWrapRejected) domain$wrapOnRejected(onRejected) else onRejected
     )
-    results[!vapply(results, is.null, logical(1))]
+    results <- results[!vapply(results, is.null, logical(1))]
+    if (!is.null(domain)) {
+      # If there's a domain, ensure that before any callback is invoked, we
+      # reenter the domain. This is important for this kind of code:
+      #
+      #     with_promise_domain(domain, {
+      #       async_sleep(0.1) %...>% {
+      #         async_sleep(0.1) %...>% {
+      #           # Without re-entry, this would be outside the domain!
+      #         }
+      #       }
+      #     })
+      results <- lapply(results, wrap_callback_reenter, domain = domain)
+    }
+    results
   },
   onError = function(error) {
     domain <- current_promise_domain()
@@ -86,6 +95,30 @@ promiseDomain <- list(
   }
 )
 
+wrap_callback_reenter <- function(callback, domain) {
+  force(callback)
+  force(domain)
+
+  wrapper <- function(...) {
+    reenter_promise_domain(domain, callback(...))
+  }
+
+  # There are parts of this package that will inspect formals() to see if
+  # there's a `.visible` parameter in the callback. So it's important to have
+  # the returned wrapper have the same formals as the original callback.
+  wrap_with_signature(wrapper, formals(callback))
+}
+
+wrap_with_signature <- function(func, formal_args) {
+  # func must have a `...` signature
+  stopifnot("..." %in% names(formals(func)))
+
+  args <- names(formal_args)
+  recall <- rlang::call2(func, !!!rlang::set_names(lapply(args, as.symbol), args))
+
+  rlang::new_function(formal_args, recall)
+}
+
 globals <- new.env(parent = emptyenv())
 
 current_promise_domain <- function() {

tests/testthat/common.R

@@ -19,31 +19,40 @@ ext_promise <- function() {
 
 # Block until all pending later tasks have executed
 # wait_for_it <- function(timeout = if (on_ci) 60 else 30) {
-wait_for_it <- function(timeout = if (on_ci) 60 else 30) {
+wait_for_it <- function(p = NULL, timeout = if (on_ci) 60 else 30) {
   start <- Sys.time()
+
+  err <- NULL
+  if (!is.null(p)) {
+    p %...!% (function(reason) err <<- reason)
+  }
+
   while (!loop_empty()) {
     if (difftime(Sys.time(), start, units = "secs") > timeout) {
       stop("Waited too long")
     }
     run_now()
     Sys.sleep(0.01)
   }
+
+  if (!is.null(err)) {
+    withRestarts(
+      stop(err),
+      continue_test = function(e) NULL
+    )
+  }
 }
 
 # Block until the promise is resolved/rejected. If resolved, return the value.
 # If rejected, throw (yes throw, not return) the error.
 extract <- function(promise) {
   promise_value <- NULL
-  error <- NULL
+
   promise %...>%
-    (function(value) promise_value <<- value) %...!%
-    (function(reason) error <<- reason)
+    (function(value) promise_value <<- value) %>%
+    wait_for_it()
 
-  wait_for_it()
-  if (!is.null(error))
-    stop(error)
-  else
-    promise_value
+  promise_value
 }
 
 resolve_later <- function(value, delaySecs) {
@@ -54,7 +63,12 @@ resolve_later <- function(value, delaySecs) {
 # Prevent "Unhandled promise error" warning that happens if you don't handle the
 # rejection of a promise
 squelch_unhandled_promise_error <- function(promise) {
-  promise %...!% {}
+  promise %...!% (function(reason) {
+    if (inherits(reason, "failure")) {
+      # Don't squelch test failures
+      stop(reason)
+    }
+  })
 }
 
 .GlobalEnv$.Last <- function() {

tests/testthat/test-aplus-2-2.R

@@ -27,12 +27,10 @@ describe("2.2. The `then` Method", {
       x <- NULL
       p <- ext_promise()
 
-      p$promise %>% then(function(value) { x <<- value })
-      wait_for_it()
+      p$promise %>% then(function(value) { x <<- value }) %>% wait_for_it()
       expect_identical(x, NULL)
 
-      p$resolve(10)
-      wait_for_it()
+      p$resolve(10) %>% wait_for_it()
       expect_identical(x, 10)
     })
     it("2.2.2.2. it must not be called before promise is fulfilled.", {
@@ -47,8 +45,8 @@ describe("2.2. The `then` Method", {
       x <- NULL
       p <- ext_promise()
 
-      p$promise %>% then(onRejected = function(reason) { x <<- reason })
-      wait_for_it()
+      p$promise %>% then(onRejected = function(reason) { x <<- reason }) %>%
+        wait_for_it()
       expect_identical(x, NULL)
 
       p$reject(simpleError("boom"))
@@ -59,9 +57,9 @@ describe("2.2. The `then` Method", {
   describe("2.2.4. onFulfilled or onRejected must not be called until the execution context stack contains only platform code. [3.1].", {
     it(" ", {
       x <- NULL
-      promise(~resolve(TRUE)) %>% then(function(value) {x <<- value})
+      p <- promise(~resolve(TRUE)) %>% then(function(value) {x <<- value})
       expect_identical(x, NULL)
-      wait_for_it()
+      p %>% wait_for_it()
       expect_identical(x, TRUE)
     })
   })
@@ -74,7 +72,7 @@ describe("2.2. The `then` Method", {
       callbacks_called <- 0L
       results <- new.env(parent = emptyenv())
 
-      lapply(1:10, function(i) {
+      all_promises <- lapply(1:10, function(i) {
         results[[as.character(i)]] <- p$promise %>%
           then(function(value) {
             callbacks_called <<- callbacks_called + 1L
@@ -84,7 +82,7 @@ describe("2.2. The `then` Method", {
       })
 
       p$resolve(cars)
-      wait_for_it()
+      promise_all(.list = all_promises) %>% wait_for_it()
 
       lapply(as.list(results), function(x) {
         expect_identical(extract(x), cars)
@@ -97,7 +95,7 @@ describe("2.2. The `then` Method", {
     callbacks_called <- 0L
     results <- new.env(parent = emptyenv())
 
-    lapply(1:10, function(i) {
+    all_promises <- lapply(1:10, function(i) {
       results[[as.character(i)]] <- p$promise %>%
         catch(function(err) {
           callbacks_called <<- callbacks_called + 1L
@@ -107,7 +105,7 @@ describe("2.2. The `then` Method", {
     })
 
     p$reject(simpleError("an error"))
-    wait_for_it()
+    promise_all(.list = all_promises) %>% wait_for_it()
 
     lapply(as.list(results), function(x) {
       expect_identical(extract(x), simpleError("an error"))

tests/testthat/test-cpp.R

@@ -0,0 +1,28 @@
+describe("C++ interface", {
+  it("basically works", {
+    promise(function(resolve, reject) {
+      asyncFib(resolve, reject, 3)
+    }) %...>%
+      { expect_identical(., 2) } %>%
+      wait_for_it()
+  })
+
+  it("works with domains", {
+    cd <- create_counting_domain()
+
+    expect_true(is.null(current_promise_domain()))
+    with_promise_domain(cd, {
+      promise(function(resolve, reject) {
+        asyncFib(resolve, reject, 3)
+      }) %...>%
+        {
+          expect_identical(., 2)
+          promise_resolve(TRUE) %...>% {
+            expect_true(!is.null(current_promise_domain()))
+            expect_identical(cd$counts$onFulfilledCalled, 3L)
+          }
+        } %>%
+        wait_for_it()
+    })
+  })
+})