Allow atomic in-place track overwrite via override

DESCRIPTION

@@ -1,7 +1,7 @@
 Type: Package
 Package: naryn
 Title: Native Access Medical Record Retriever for High Yield Analytics
-Version: 2.7.1
+Version: 2.7.2
 Authors@R: c(
     person("Misha", "Hoichman", , "misha@hoichman.com", role = "aut"),
     person("Aviezer", "Lifshitz", , "aviezer.lifshitz@weizmann.ac.il", role = c("aut", "cre")),

NEWS.md

@@ -1,3 +1,8 @@
+# naryn 2.7.2
+
+* `emr_track.create` and `emr_track.import` can now rewrite a track in its own db when `override = TRUE`. Previously `override` only covered shadowing a track from another db, and rewriting in place required `emr_track.rm()` first - which left the track missing for the whole rebuild, so anything reading it in that window failed.
+* Track writes are now staged to a temporary file and renamed into place. `rename(2)` replaces the target atomically, so a concurrent reader always sees either the complete previous track or the complete new one. `emr_track.import` already staged, but unlinked the target before the move, which reopened the same window.
+
 # naryn 2.7.1
 
 * Merged the CRAN-portability fixes from 2.6.32-2.6.34 onto the new locking mechanism introduced in 2.7.0.

R/track.R

@@ -195,7 +195,10 @@ emr_track.addto <- function(track, src, force = FALSE, remove_unknown = FALSE) {
 #' implicitly based on track expressions. See also 'iterator' section.
 #' @param keepref If 'TRUE' references are preserved in the iterator
 #' @param filter Iterator filter
-#' @param override Boolean indicating whether the creation intends to override an existing track (default FALSE)
+#' @param override Boolean indicating whether the write intends to replace an existing track (default FALSE).
+#' Covers both shadowing a track that lives in another db and rewriting one in this db. The new track is
+#' written to a staging file and renamed into place, so readers never observe the track missing or partial
+#' - there is no need to \code{emr_track.rm()} it first.
 #'
 #' @return None.
 #'
@@ -347,7 +350,10 @@ emr_track.ids <- function(track) {
 #' @param space db dir string (path), one of the paths supplied in emr_db.connect
 #' @param categorical if 'TRUE' track is marked as categorical
 #' @param src file name or data-frame containing the track records
-#' @param override Boolean indicating whether the creation intends to override an existing track (default FALSE)
+#' @param override Boolean indicating whether the write intends to replace an existing track (default FALSE).
+#' Covers both shadowing a track that lives in another db and rewriting one in this db. The new track is
+#' written to a staging file and renamed into place, so readers never observe the track missing or partial
+#' - there is no need to \code{emr_track.rm()} it first.
 #' @param remove_unknown if 'TRUE', removes unknown ids (ids that are not present at 'patients.dob' track) from the data. Otherwise, an error is thrown.
 #'
 #' @return None.

src/NRImport.cpp

@@ -1,6 +1,8 @@
 #include <stdlib.h>
 #include <unistd.h>
 
+#include <string>
+
 #include "EMRDb.h"
 #include "EMRTrack.h"
 #include "FileUtils.h"
@@ -57,9 +59,12 @@ SEXP emr_import(SEXP _track, SEXP _db_id, SEXP _categorical, SEXP _src, SEXP _ad
                 verror("%s directory is not set", db_id.c_str());
             }
 
-            //Error only if exists in the same db, otherwise try overriding
-            if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id == db_id)) {
-                verror("Track %s already exists", trackname.c_str());
+            // Rewriting a track in its own db needs override, same as shadowing one from
+            // another db. The write below is staged and renamed into place, so an in-place
+            // rewrite never leaves readers with a missing or half-written track - callers no
+            // longer have to emr_track.rm() first and expose that window themselves.
+            if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id == db_id) && !toverride) {
+                verror("Track %s already exists, see override argument", trackname.c_str());
             }
             //User must explicitly pass an overriding argument
             if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id != db_id) && !toverride) {
@@ -90,7 +95,9 @@ SEXP emr_import(SEXP _track, SEXP _db_id, SEXP _categorical, SEXP _src, SEXP _ad
 
             track_filename = db_id + string("/") + trackname + EMRDb::TRACK_FILE_EXT;
 
-            if (access(track_filename.c_str(), F_OK) != -1)
+            // A stray file with no track registered for it: still refuse to clobber it silently,
+            // but override is an explicit "rewrite this track", so let it through.
+            if (!toverride && access(track_filename.c_str(), F_OK) != -1)
                 verror("File %s already exists", track_filename.c_str());
         }
 
@@ -174,15 +181,22 @@ SEXP emr_import(SEXP _track, SEXP _db_id, SEXP _categorical, SEXP _src, SEXP _ad
         }
 
 
-        if (access(track_filename.c_str(), F_OK) != -1) {
-            string tmp_filename = track_filename + ".tmp";
+        // Stage next to the target (same directory, so same filesystem) and rename into place.
+        // rename(2) replaces an existing file atomically, so a concurrent reader sees either the
+        // complete previous track or the complete new one - never a partial or absent file. The
+        // unlink that used to precede the move reopened exactly that window, and is not needed:
+        // rename replaces the target by itself. The pid in the suffix keeps two writers of the
+        // same track from clobbering each other's staging file.
+        string tmp_filename = track_filename + ".tmp." + std::to_string(getpid());
+        try {
             EMRTrack::serialize(tmp_filename.c_str(), categorical ? EMRTrack::IS_CATEGORICAL : 0, data);
-            unlink(track_filename.c_str());
             FileUtils::move_file(tmp_filename.c_str(), track_filename.c_str());
-        } else {
-            EMRTrack::serialize(track_filename.c_str(), categorical, data);
+        } catch (...) {
+            unlink(tmp_filename.c_str());
+            throw;
         }
-
+
+
         if (has_overlap){
             g_db->unload_track(trackname.c_str(), true, true);
         }

src/NRTrackCreate.cpp

@@ -11,8 +11,13 @@
 #undef error
 #endif
 
+#include <unistd.h>
+
+#include <string>
+
 #include "EMRDb.h"
 #include "EMRTrack.h"
+#include "FileUtils.h"
 #include "naryn.h"
 #include "NRTrackExpressionScanner.h"
 
@@ -46,10 +51,14 @@ SEXP emr_track_create(SEXP _track, SEXP _db_id, SEXP _categorical, SEXP _expr, S
 
         string trackname = { CHAR(Rf_asChar(_track)) };
 
-        if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id == db_id)){
-            verror("Track %s already exists", trackname.c_str());
+        // Rewriting a track in its own db needs override, same as shadowing one from another
+        // db. The write below is staged and renamed into place, so an in-place rewrite never
+        // leaves readers with a missing or half-written track - callers no longer have to
+        // emr_track.rm() first and expose that window themselves.
+        if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id == db_id) && !toverride){
+            verror("Track %s already exists, see override argument", trackname.c_str());
         }
-            
+
         // User must explicitly pass an overriding argument
         if (g_db->track(trackname) && (g_db->track_info(trackname)->db_id != db_id) && !toverride){
             verror("Track %s already exists in db %s, see override argument", trackname.c_str(), g_db->track_info(trackname)->db_id.c_str());
@@ -85,7 +94,19 @@ SEXP emr_track_create(SEXP _track, SEXP _db_id, SEXP _categorical, SEXP _expr, S
 			g_naryn->verify_max_data_size(data.data.size(), "Result");
 		}
 
-        EMRTrack::serialize(track_filename.c_str(), categorical, data);
+        // Stage next to the target (same directory, so same filesystem) and rename into place.
+        // rename(2) replaces an existing file atomically, so a concurrent reader sees either the
+        // complete previous track or the complete new one - never a partial or absent file. The
+        // pid in the suffix keeps two writers of the same track from clobbering each other's
+        // staging file; the rename itself then just decides who wins.
+        string tmp_filename = track_filename + ".tmp." + std::to_string(getpid());
+        try {
+            EMRTrack::serialize(tmp_filename.c_str(), categorical ? EMRTrack::IS_CATEGORICAL : 0, data);
+            FileUtils::move_file(tmp_filename.c_str(), track_filename.c_str());
+        } catch (...) {
+            unlink(tmp_filename.c_str());
+            throw;
+        }
 
         if (has_overlap) {
             g_db->unload_track(trackname.c_str(), true, true);

tests/testthat/test-track.overwrite.R

@@ -0,0 +1,48 @@
+load_test_db()
+
+# Rewriting a track in its own db used to be impossible: both emr_track.import and
+# emr_track.create errored with "Track already exists" regardless of `override`, which only
+# covered shadowing a track from another db. Callers had to emr_track.rm() first, leaving the
+# track absent for the whole rebuild - anything reading it in that window failed.
+
+test_that("emr_track.import can rewrite a track in the same db with override", {
+    a <- emr_extract("track1", keepref = TRUE, names = "value")
+    emr_track.import("ovr_track", "global", categorical = FALSE, src = a)
+    withr::defer(emr_track.rm("ovr_track", force = TRUE))
+
+    expect_error(emr_track.import("ovr_track", "global", categorical = FALSE, src = a))
+
+    b <- emr_extract("track2", keepref = TRUE, names = "value")
+    emr_track.import("ovr_track", "global", categorical = FALSE, src = b, override = TRUE)
+
+    expect_true(emr_track.exists("ovr_track"))
+    got <- emr_extract("ovr_track", keepref = TRUE, names = "value")
+    expect_equal(got$value, b$value)
+})
+
+test_that("emr_track.create can rewrite a track in the same db with override", {
+    emr_track.create("ovr_ctrack", "global", categorical = FALSE, expr = "track1")
+    withr::defer(emr_track.rm("ovr_ctrack", force = TRUE))
+    before <- emr_extract("ovr_ctrack", names = "value")$value
+
+    expect_error(emr_track.create("ovr_ctrack", "global", categorical = FALSE, expr = "track1 + 1"))
+
+    # Same expression shifted by one, so the comparison stays inside a single iterator shape
+    # and only tests that the rewrite actually replaced the stored data.
+    emr_track.create("ovr_ctrack", "global", categorical = FALSE, expr = "track1 + 1", override = TRUE)
+
+    expect_true(emr_track.exists("ovr_ctrack"))
+    expect_equal(emr_extract("ovr_ctrack", names = "value")$value, before + 1)
+})
+
+test_that("overwriting leaves no staging files behind", {
+    a <- emr_extract("track1", keepref = TRUE, names = "value")
+    emr_track.import("ovr_tmp_track", "global", categorical = FALSE, src = a)
+    withr::defer(emr_track.rm("ovr_tmp_track", force = TRUE))
+    emr_track.import("ovr_tmp_track", "global", categorical = FALSE, src = a, override = TRUE)
+    emr_track.create("ovr_tmp_ctrack", "global", categorical = FALSE, expr = "track1", override = TRUE)
+    withr::defer(emr_track.rm("ovr_tmp_ctrack", force = TRUE))
+
+    db_dir <- emr_db.ls()[1]
+    expect_length(list.files(db_dir, pattern = "\\.tmp\\."), 0)
+})