Fix mangled worker stdout/stderr (#10)

ElianHugh · web-flow · commit 24049dc20372 · 2026-01-09T16:55:24.000+11:00
- Sink worker stderr/stdout into temporary file managed by hotwater
- Iteratively read and tail the file during the file watcher loop
diff --git a/R/engine.R b/R/engine.R
@@ -4,7 +4,7 @@
 
 new_engine <- function(config) {
     stopifnot(is_config(config))
-    structure(
+    eng <- structure(
         list2env(
             list(
                 runner = NULL,
@@ -17,11 +17,19 @@ new_engine <- function(config) {
                         config$socket_port
                     ),
                     autostart = FALSE
-                )
+                ),
+                logpath = tempfile(),
+                logpos = 0L
             )
         ),
         class = c("hotwater_engine", "environment")
     )
+
+    reg.finalizer(eng, function(e) {
+        try(unlink(e$logpath), silent = TRUE)
+    }, onexit=TRUE)
+
+    eng
 }
 
 run_engine <- function(engine) {
@@ -30,7 +38,10 @@ run_engine <- function(engine) {
         teardown_engine(engine)
         buildup_engine(engine)
     }
-    on.exit({ teardown_engine(engine) }) # nolint: brace_linter.
+    on.exit({
+        teardown_engine(engine)
+        try(unlink(engine$logpath), silent = TRUE)
+    }) # nolint: brace_linter.
 
     cli_welcome()
     buildup_engine(engine)
@@ -45,6 +56,7 @@ run_engine <- function(engine) {
 
     repeat {
         Sys.sleep(0.05) # todo, allow this to be configured at some point
+        drain_runner_log(engine)
         current_state <- watch_directory(
             engine,
             current_state,
@@ -62,6 +74,11 @@ buildup_engine <- function(engine) {
     stopifnot(is_engine(engine))
 
     cli_server_start_progress(engine)
+
+    # just in case the logfile was deleted
+    cat("", file = engine$logpath)
+    engine$logpos <- 0L
+
     res <- new_runner(engine)
 
     if (engine$publisher$listener[[1L]][["state"]] != "started") {
@@ -76,6 +93,7 @@ buildup_engine <- function(engine) {
     }
 
     cli_watching_directory(engine)
+    drain_runner_log(engine)
 }
 
 teardown_engine <- function(engine) {
@@ -93,4 +111,53 @@ teardown_engine <- function(engine) {
 
 is_engine <- function(x) {
     inherits(x, "hotwater_engine")
+}
+
+drain_runner_log <- function(engine) {
+    logpath <- engine$logpath
+    logpos <- engine$logpos
+
+    if (!file.exists(logpath)) {
+        return()
+    }
+
+    size <- file.info(logpath)$size
+    if (is.na(size)) {
+        return()
+    }
+
+    if (size < logpos) {
+        engine$logpos <- 0L
+        logpos <- 0L
+    }
+
+    if (size <= logpos) {
+        return()
+    }
+
+    con <- file(logpath, open = "rb")
+    on.exit(close(con), add = TRUE)
+
+    seek(con, where = logpos, origin = "start")
+    data <- readChar(con, nchars = size - logpos, useBytes = TRUE)
+
+    engine$logpos <- size
+
+    if (nzchar(data)) {
+        data <- gsub(
+            "=== HOTWATER_ERROR_BEGIN ===\\s*([\\s\\S]*?)\\s*=== HOTWATER_ERROR_END ===",
+            cli::col_red("\\1"),
+            data,
+            perl = TRUE
+        )
+
+        data <- gsub(
+            "=== HOTWATER_WARNING_BEGIN ===\\s*([\\s\\S]*?)\\s*=== HOTWATER_WARNING_END ===",
+            cli::col_yellow("\\1"),
+            data,
+            perl = TRUE
+        )
+
+        nanonext::write_stdout(data)
+    }
 }
diff --git a/R/mirai.R b/R/mirai.R
@@ -9,7 +9,7 @@ new_runner <- function(engine) {
         dispatcher = FALSE,
         resilience = FALSE,
         autoexit = get_kill_signal(),
-        output = TRUE,
+        output = FALSE,
         .compute = engine$config$runner_compute
     )
 
@@ -18,26 +18,63 @@ new_runner <- function(engine) {
     mdware <- middleware(engine)
     mod <- file.path(getwd(), engine$config$entry_path)
     host <- engine$config$host
+    logpath <- engine$logpath
 
     engine$runner <- mirai::mirai(
         {
-            if (requireNamespace("box", quietly = TRUE)) {
-                box::set_script_path(mod)
-            }
-            plumber::pr_run(
-                mdware(plumber::pr(path)),
-                port = port,
-                host = host,
-                quiet = TRUE,
-                debug = TRUE
+
+            con <- file(logpath, open = "at")
+            sink(con)
+            sink(con, type = "message")
+
+            on.exit(
+                {
+                    try(sink(type = "message"), silent = TRUE)
+                    try(sink(), silent = TRUE)
+                    try(close(con), silent = TRUE)
+                },
+                add = TRUE
             )
+
+            withCallingHandlers(
+                tryCatch(
+                    {
+                        if (requireNamespace("box", quietly = TRUE)) {
+                            box::set_script_path(mod)
+                        }
+                        plumber::pr_run(
+                            mdware(plumber::pr(path)),
+                            port = port,
+                            host = host,
+                            quiet = TRUE,
+                            debug = TRUE
+                        )
+                    },
+                    error = function(e) {
+                        cat("=== HOTWATER_ERROR_BEGIN ===\n", file = con)
+                        cat(conditionMessage(e), "\n", file = con)
+                        cat("=== HOTWATER_ERROR_END ===\n", file = con)
+                        flush(con)
+                        stop(e)
+                    }
+                ),
+                warning = function(w) {
+                    cat("=== HOTWATER_WARNING_BEGIN ===\n", file = con)
+                    cat(conditionMessage(w), "\n", file = con)
+                    cat("=== HOTWATER_WARNING_END ===\n", file = con)
+                    flush(con)
+                    invokeRestart("muffleWarning")
+                }
+            )
+
         },
         .args = list(
             port = port,
             path = path,
             host = host,
             mdware = mdware,
-            mod = mod
+            mod = mod,
+            logpath = logpath
         ),
         .compute = engine$config$runner_compute
     )
diff --git a/tests/testthat/test-engine.R b/tests/testthat/test-engine.R
@@ -29,3 +29,15 @@ test_that("can kill engine", {
     expect_false(is_runner_alive(engine))
     cleanup_test_engine(engine)
 })
+
+test_that('logs are cleared', {
+    engine <- new_test_engine()
+    logpath <- engine$logpath
+    new_runner(engine)
+    expect_true(file.exists(logpath))
+    kill_engine(engine)
+    cleanup_test_engine(engine)
+    rm(engine)
+    gc()
+    expect_false(file.exists(logpath))
+})
diff --git a/tests/testthat/test-logging.R b/tests/testthat/test-logging.R
@@ -0,0 +1,82 @@
+# mock the nanonext fn to test if we are
+# handling the sink properly
+capture_write_stdout <- function(code) {
+    out <- character()
+    old <- nanonext::write_stdout
+    assignInNamespace(
+        "write_stdout",
+        function(x) out <<- c(out, x),
+        ns = "nanonext"
+    )
+    on.exit(assignInNamespace("write_stdout", old, ns = "nanonext"), add = TRUE)
+    code()
+    out
+}
+
+test_that('log files are created/removed', {
+    engine <- new_test_engine()
+    logpath <- engine$logpath
+    new_runner(engine)
+    expect_true(file.exists(logpath))
+    kill_engine(engine)
+    cleanup_test_engine(engine)
+    rm(engine)
+    gc()
+    expect_false(file.exists(logpath))
+})
+
+test_that("stdout drain forwards log contents", {
+    engine <- new_test_engine()
+    logpath <- engine$logpath
+
+    cat("Hello World!\n", file = logpath)
+
+    out <- capture_write_stdout(function() {
+    drain_runner_log(engine)
+    })
+
+    expect_true(any(grepl("Hello World!", out)))
+
+
+    cleanup_test_engine(engine)
+    rm(engine)
+    gc()
+})
+
+test_that("markers are removed", {
+    engine <- new_test_engine()
+    logpath <- engine$logpath
+    cat(
+        "=== HOTWATER_ERROR_BEGIN ===\nfoo\n=== HOTWATER_ERROR_END ===\n",
+        file = logpath
+    )
+
+    out <- capture_write_stdout(function() drain_runner_log(engine))
+    txt <- paste(out, collapse = "")
+
+    expect_true(grepl("foo", txt))
+    expect_false(grepl("HOTWATER_ERROR_BEGIN", txt))
+    expect_false(grepl("HOTWATER_ERROR_END", txt))
+
+
+    cleanup_test_engine(engine)
+    rm(engine)
+    gc()
+})
+
+
+test_that("engine doesn't reprint old content", {
+    engine <- new_test_engine()
+    logpath <- engine$logpath
+    cat("A\n", file = logpath)
+
+    out1 <- capture_write_stdout(function() drain_runner_log(engine))
+    out2 <- capture_write_stdout(function() drain_runner_log(engine))
+
+    expect_true(any(grepl("A", out1)))
+    expect_false(any(grepl("A", out2)))
+
+    cleanup_test_engine(engine)
+    rm(engine)
+    gc()
+})
diff --git a/tests/testthat/test-middleware.R b/tests/testthat/test-middleware.R
@@ -74,9 +74,17 @@ test_that("is_plumber_running works", {
 test_that("autoreloader is attached", {
     engine <- new_test_engine()
     new_runner(engine)
+
+    i <- 1L
+    while (i < 20L && !is_plumber_running(engine)) {
+        i <- i + 1L
+        Sys.sleep(0.5)
+    }
+
     resp <- sprintf("%s:%s", engine$config$host, engine$config$port) |>
         httr2::request() |>
         httr2::req_perform()
+
     expect_no_error(
         httr2::resp_check_content_type(
             resp,
