Merge pull request #527 from hivtools/nm-42

[NM-42] Make it impossible to accidentally destroy redis data
hivtools · Sep 25, 2024 · 85d3e87 · 85d3e87
2 parents cb828e7 + ef3d90e
commit 85d3e87
Show file tree

Hide file tree

Showing 9 changed files with 454 additions and 415 deletions.
diff --git a/NEWS.md b/NEWS.md
@@ -4,6 +4,13 @@
 
 # hintr 1.2.1
 
+* Make it harder to accidentally delete redis data by:
+   * Not starting workers automatically when you create a queue
+   * Only removing the workers a queue creates by default
+   * Only deleting data if you explicitly opt in
+
+# hintr 1.2.1
+
 * Add new endpoint `/download/result/path/<id>` to return path to the download file on disk
 
 # hintr 1.2.0

diff --git a/R/api.R b/R/api.R
@@ -94,7 +94,6 @@ api <- function(queue_id = NULL, workers = 2,
                      inputs_dir = inputs_dir,
                      health_check_interval = health_check_interval)
   rrq::rrq_worker_delete_exited(controller = queue$controller)
-  logger <- porcelain::porcelain_logger(log_level)
   api_build(queue, logger = logger)
 }
 

diff --git a/R/queue.R b/R/queue.R
@@ -3,21 +3,25 @@ Queue <- R6::R6Class(
   cloneable = FALSE,
   public = list(
     root = NULL,
-    cleanup_on_exit = NULL,
+    stop_workers_on_exit = NULL,
+    delete_data_on_exit = FALSE,
     controller = NULL,
+    worker_ids = NULL,
     results_dir = NULL,
     inputs_dir = NULL,
 
     health_check_interval = NULL,
     next_health_check = NULL,
 
-    initialize = function(queue_id = NULL, workers = 2,
-                          cleanup_on_exit = workers > 0,
+    initialize = function(queue_id = NULL, workers = 0,
+                          stop_workers_on_exit = workers > 0,
+                          delete_data_on_exit = FALSE,
                           results_dir = tempdir(),
                           inputs_dir = NULL,
                           timeout = Inf,
                           health_check_interval = 0) {
-      self$cleanup_on_exit <- cleanup_on_exit
+      self$stop_workers_on_exit <- stop_workers_on_exit
+      self$delete_data_on_exit <- delete_data_on_exit
       self$results_dir <- results_dir
 
       message(t_("QUEUE_CONNECTING", list(redis = redux::redis_config()$url)))
@@ -66,11 +70,12 @@ Queue <- R6::R6Class(
 
     start = function(workers, timeout) {
       if (workers > 0L) {
-        worker_manager <- rrq::rrq_worker_spawn(workers,
-                                                controller = self$controller)
+        worker_manager <- rrq::rrq_worker_spawn(
+          workers, controller = self$controller)
+        self$worker_ids <- worker_manager$id
         if (is.finite(timeout) && timeout > 0) {
           rrq::rrq_message_send_and_wait("TIMEOUT_SET", timeout,
-                                         worker_manager$id,
+                                         self$worker_ids,
                                          controller = self$controller)
         }
       }
@@ -168,17 +173,26 @@ Queue <- R6::R6Class(
 
     ## Not part of the api exposed functions, used in tests
     destroy = function() {
+      message(sprintf(
+        "Deleting all redis data for queue '%s'",
+        self$controller$queue_id))
       rrq::rrq_destroy(delete = TRUE, controller = self$controller)
     },
 
     cleanup = function() {
       if (!is.null(self$controller)) {
         clear_cache(self$controller$queue_id)
-        if (self$cleanup_on_exit) {
-          message(t_("QUEUE_STOPPING_WORKERS"))
+        if (self$delete_data_on_exit) {
+          message("Stopping all workers")
+          rrq::rrq_worker_detect_exited(controller = self$controller)
           worker_stop(type = "kill", controller = self$controller)
           self$destroy()
           self$controller <- NULL
+        } else if (self$stop_workers_on_exit && !is.null(self$worker_ids)) {
+          message(t_("QUEUE_STOPPING_WORKERS"))
+          worker_stop(self$worker_ids, type = "kill",
+                      controller = self$controller)
+          self$controller <- NULL
         }
       }
     }

diff --git a/scripts/redis b/scripts/redis
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 set -e
-CONTAINER_NAME=hintr_redis
+CONTAINER_NAME=hintr-redis
 if [ "$1" = "start" ]; then
     docker run --rm -d --name=$CONTAINER_NAME -p 127.0.0.1:6379:6379 redis
 elif [ "$1" = "stop" ]; then

diff --git a/tests/testthat/helper-queue.R b/tests/testthat/helper-queue.R
@@ -25,8 +25,10 @@ MockQueue <- R6::R6Class(
   )
 )
 
-test_queue <- function(workers = 0) {
-  queue <- Queue$new(workers = workers, timeout = 300)
+test_queue <- function(workers = 0, delete_data_on_exit = TRUE) {
+  queue <- Queue$new(workers = workers,
+                     timeout = 300,
+                     delete_data_on_exit = delete_data_on_exit)
   withr::defer_parent({
     message("cleaning up workers")
     queue$cleanup()
@@ -48,9 +50,10 @@ create_blocking_worker <- function(controller, worker_name = NULL) {
                        worker_id = worker_name)
 }
 
-test_queue_result <- function(model = mock_model, calibrate = mock_calibrate,
+test_queue_result <- function(model = mock_model,
+                              calibrate = mock_calibrate,
                               clone_output = TRUE) {
-  queue <- Queue$new(workers = 1, timeout = 300)
+  queue <- Queue$new(workers = 1, timeout = 300, delete_data_on_exit = TRUE)
   withr::defer_parent({
     message("cleaning up workers")
     queue$cleanup()
@@ -141,8 +144,7 @@ setup_prerun_queue <- function() {
   )
 
   list(
-    queue = Queue$new(workers = 0,
-                      inputs_dir = inputs_dir,
+    queue = Queue$new(inputs_dir = inputs_dir,
                       results_dir = output_dir),
     payload = jsonlite::toJSON(payload)
   )