Merge pull request #119 from mrc-ide/mrc-4724

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: orderly2
 Title: Orderly Next Generation
-Version: 1.99.9
+Version: 1.99.10
 Authors@R: c(person("Rich", "FitzJohn", role = c("aut", "cre"),
                     email = "[email protected]"),
              person("Robert", "Ashton", role = "aut"),

R/outpack_index.R

@@ -12,13 +12,14 @@ outpack_index <- R6::R6Class(
       private$path_ <- path
     },
 
-    rebuild = function() {
-      private$data_ <- index_update(private$path_, NULL, TRUE)
+    rebuild = function(progress = NULL) {
+      private$data_ <- index_update(private$path_, NULL, TRUE, progress)
       invisible(self)
     },
 
-    refresh = function() {
-      private$data_ <- index_update(private$path_, private$data_, FALSE)
+    refresh = function(progress = NULL) {
+      private$data_ <- index_update(private$path_, private$data_, FALSE,
+                                    progress)
       invisible(self)
     },
 
@@ -55,45 +56,59 @@ outpack_index <- R6::R6Class(
   ))
 
 
-index_update <- function(root_path, prev, skip_cache) {
+index_update <- function(root_path, prev, skip_cache, progress) {
+  progress <- progress %||% getOption("orderly_index_progress", TRUE)
   path_index <- file.path(root_path, ".outpack", "index", "outpack.rds")
 
   if (length(prev) == 0 && file.exists(path_index) && !skip_cache) {
-    prev <- readRDS(path_index)
+    prev <- tryCatch(readRDS(path_index), error = function(e) {
+      cli::cli_alert_warning("outpack index corrupted, rebuilding")
+      NULL
+    })
   }
 
   data <- prev
-  data$location <- read_locations(root_path, data$location)
-  data$metadata <- read_metadata(root_path, data$metadata)
+  data$metadata <- read_metadata(root_path, data$metadata, progress)
+  data$location <- read_locations(root_path, data$location, progress)
   data$unpacked <- data$location$packet[data$location$location == local]
 
   if (!identical(data, prev)) {
     fs::dir_create(dirname(path_index))
-    saveRDS(data, path_index)
+    saverds_atomic(data, path_index, allow_fail = TRUE)
   }
 
   data
 }
 
 
-read_metadata <- function(root_path, prev) {
+read_metadata <- function(root_path, prev, progress) {
   path <- file.path(root_path, ".outpack", "metadata")
   id_new <- setdiff(dir(path), names(prev))
 
   if (length(id_new) == 0) {
     return(prev)
   }
 
+  if (progress) {
+    cli::cli_progress_bar("Reading metadata", total = length(id_new))
+  }
   files <- file.path(path, id_new)
-  new <- lapply(files, outpack_metadata_core_read)
+  new <- vector("list", length(id_new))
+  for (i in seq_along(id_new)) {
+    new[[i]] <- outpack_metadata_core_read(files[[i]])
+    if (progress) {
+      cli::cli_progress_update()
+    }
+  }
+
   names(new) <- id_new
   ret <- c(prev, new)
   ret[order(names(ret))]
   ret
 }
 
 
-read_locations <- function(root_path, prev) {
+read_locations <- function(root_path, prev, progress) {
   if (is.null(prev)) {
     prev <- data_frame(packet = character(),
                        time = empty_time(),
@@ -104,7 +119,9 @@ read_locations <- function(root_path, prev) {
   location_path <- fs::dir_ls(file.path(root_path, ".outpack", "location"),
                               type = "directory")
   location_name <- basename(location_path)
-  new <- do.call(rbind, lapply(location_name, read_location, root_path, prev))
+  new <- do.call(
+    rbind,
+    lapply(location_name, read_location, root_path, prev, progress))
   ret <- rbind(prev, new)
   ## Always sort by location, then id
   ret <- ret[order(match(ret$location, location_name), ret$packet), ]
@@ -114,15 +131,28 @@ read_locations <- function(root_path, prev) {
 }
 
 
-read_location <- function(location_name, root_path, prev) {
+read_location <- function(location_name, root_path, prev, progress) {
   path <- file.path(root_path, ".outpack", "location", location_name)
   packets <- dir(path, re_id)
   is_new <- !(packets %in% prev$packet[prev$location == location_name])
   if (!any(is_new)) {
     return(NULL)
   }
 
-  dat <- lapply(file.path(path, packets[is_new]), jsonlite::read_json)
+  id_new <- packets[is_new]
+  dat <- vector("list", length(id_new))
+  files <- file.path(path, id_new)
+  if (progress) {
+    cli::cli_progress_bar(sprintf("Reading location '%s'", location_name),
+                          total = length(id_new))
+  }
+  for (i in seq_along(id_new)) {
+    dat[[i]] <- jsonlite::read_json(files[[i]])
+    if (progress) {
+      cli::cli_progress_update()
+    }
+  }
+
   data_frame(packet = vcapply(dat, "[[", "packet"),
              time = num_to_time(vnapply(dat, "[[", "time")),
              hash = vcapply(dat, "[[", "hash"),

R/outpack_metadata.R

@@ -166,9 +166,10 @@ outpack_metadata_core <- function(id, root, call = NULL) {
 }
 
 
+
+metadata_core_names <- c("id", "name", "parameters", "time", "files", "depends")
 outpack_metadata_core_read <- function(path) {
-  keep <- c("id", "name", "parameters", "time", "files", "depends")
-  data <- jsonlite::read_json(path)[keep]
+  data <- jsonlite::read_json(path)[metadata_core_names]
   outpack_metadata_core_deserialise(data)
 }
 

R/outpack_tools.R

@@ -208,7 +208,13 @@ orderly_metadata_extract <- function(..., extract = NULL, root = NULL,
   }
   extract <- parse_extract(extract, environment())
 
-  meta <- lapply(ids, orderly_metadata, root = root)
+  is_core_metadata <-
+    vlapply(extract$from, function(el) el[[1]] %in% metadata_core_names)
+  if (all(is_core_metadata)) {
+    meta <- lapply(ids, outpack_metadata_core, root = root)
+  } else {
+    meta <- lapply(ids, orderly_metadata, root = root)
+  }
 
   envir <- environment()
   ret <- data_frame(id = ids)

R/util.R

@@ -615,3 +615,20 @@ file_canonical_case <- function(path, workdir) {
   }
   paste(ret, collapse = "/")
 }
+
+
+saverds_atomic <- function(data, path, allow_fail = FALSE) {
+  tmp <- tempfile(pattern = sub("\\.rds", "", basename(path)),
+                  tmpdir = dirname(path),
+                  fileext = ".rds")
+  saveRDS(data, tmp)
+  if (allow_fail) {
+    tryCatch(
+      fs::file_move(tmp, path),
+      error = function(e) unlink(tmp))
+  } else {
+    tryCatch(
+      fs::file_move(tmp, path),
+      finally = unlink(tmp))
+  }
+}

tests/testthat/helper-orderly.R

@@ -1,6 +1,7 @@
 options(outpack.schema_validate =
           requireNamespace("jsonvalidate", quietly = TRUE) &&
-          packageVersion("jsonvalidate") >= "1.4.0")
+          packageVersion("jsonvalidate") >= "1.4.0",
+        orderly_index_progress = FALSE)
 
 
 test_prepare_orderly_example <- function(examples, ...) {

tests/testthat/test-outpack-index.R

@@ -0,0 +1,36 @@
+test_that("reading metadata reports progress if requested", {
+  root <- create_temporary_root()
+  ids <- create_random_packet_chain(5, root = root)
+
+  mock_progress_bar <- mockery::mock()
+  mock_progress_update <- mockery::mock()
+  mockery::stub(read_metadata, "cli::cli_progress_bar", mock_progress_bar)
+  mockery::stub(read_metadata, "cli::cli_progress_update", mock_progress_update)
+
+  read_metadata(root$path, NULL, FALSE)
+  mockery::expect_called(mock_progress_bar, 0)
+  mockery::expect_called(mock_progress_update, 0)
+
+  read_metadata(root$path, NULL, TRUE)
+  mockery::expect_called(mock_progress_bar, 1)
+  mockery::expect_called(mock_progress_update, 5)
+})
+
+
+test_that("reading location reports progress if requested", {
+  root <- create_temporary_root()
+  ids <- create_random_packet_chain(5, root = root)
+
+  mock_progress_bar <- mockery::mock()
+  mock_progress_update <- mockery::mock()
+  mockery::stub(read_location, "cli::cli_progress_bar", mock_progress_bar)
+  mockery::stub(read_location, "cli::cli_progress_update", mock_progress_update)
+
+  read_location("local", root$path, NULL, FALSE)
+  mockery::expect_called(mock_progress_bar, 0)
+  mockery::expect_called(mock_progress_update, 0)
+
+  read_location("local", root$path, NULL, TRUE)
+  mockery::expect_called(mock_progress_bar, 1)
+  mockery::expect_called(mock_progress_update, 5)
+})

tests/testthat/test-outpack-metadata.R

@@ -122,3 +122,12 @@ test_that("Sensible error if metadata file not found", {
     orderly_metadata(outpack_id(), root),
     "Packet '.+' not found in outpack index")
 })
+
+
+test_that("corrupted metadata is replaced automatically", {
+  root <- create_temporary_root()
+  id <- create_random_packet(root)
+  file.create(file.path(root$path, ".outpack", "index", "outpack.rds"))
+  expect_message(outpack_index$new(root$path)$refresh(),
+                 "outpack index corrupted, rebuilding")
+})

tests/testthat/test-util.R

@@ -306,3 +306,18 @@ test_that("can convert files to canonical case", {
   expect_equal(file_canonical_case("A/win~1/C", tmp), NA_character_)
   expect_equal(file_canonical_case(c("a/b/c", "a/b/d"), tmp), c("a/b/c", NA))
 })
+
+
+test_that("can gracefully cope with rds save failure", {
+  mock_move <- mockery::mock(stop("some error"), cycle = TRUE)
+  mockery::stub(saverds_atomic, "fs::file_move", mock_move)
+  tmp <- withr::local_tempdir()
+  path <- file.path(tmp, "file.rds")
+  expect_silent(
+    saverds_atomic(NULL, path, allow_fail = TRUE))
+  expect_equal(dir(tmp), character())
+  expect_error(
+    saverds_atomic(NULL, path, allow_fail = FALSE),
+    "some error")
+  expect_equal(dir(tmp), character())
+})