dataflow-state: Handle non-existent deployment dirs

At process start up, Readyset will clean out the tmp/working dir,
in case the process crashed on a previous run and we need to delete
temp files. If the `working_dir` already exists by the time the
`clean_working_dir()` function is invoked, then everything is fine. If
the directory does not exist, then the call to `Path::canonicalize()`
will fail.

This problem does not exist in standalone, as the Standalone Authority
is created (and it's storage directories) much earlier than we attempt
to clean out the working dir. This is not the case in adapter/server
operation, which is where this was failing. The fix is to simply not
try to delete or canonicalize the working directory path if the
directory does not exist.

Release-Note-Core: Fix a bug in distributed readyset where
  readyset-server will fail on startup due to a missing working_dir.
Change-Id: Ibe7b0511d285b224a36ce7568673105926f1c354
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/7920
Tested-by: Buildkite CI
Reviewed-by: Johnathan Davis <jcd@readyset.io>

Author: jasobrown-rs

dataflow-state/src/persistent_state/mod.rs

@@ -131,8 +131,7 @@ const INDEX_BATCH_SIZE: usize = 10_000;
 /// to prevent any further disk use leakage. This function should be executed once,
 /// preferably at startup.
 pub fn clean_working_dir(params: &PersistenceParameters) -> Result<()> {
-    let working_dir = params.derive_working_dir_base()?;
-
+    let working_dir = params.working_dir_base();
     if working_dir.is_dir() {
         debug!("deleting any prior working files from {:?}", &working_dir);
         fs::remove_dir_all(working_dir)?
@@ -325,17 +324,17 @@ impl PersistenceParameters {
         }
     }
 
-    fn derive_working_dir_base(&self) -> Result<PathBuf> {
-        // use what the operator passed in, or default to the storage_dir
+    fn storage_dir(&self) -> PathBuf {
+        self.storage_dir.clone().unwrap_or_else(|| ".".into())
+    }
+
+    fn working_dir_base(&self) -> PathBuf {
         let base_dir = match &self.working_temp_dir {
             Some(s) => s.clone(),
-            None => self.storage_dir.clone().unwrap_or_else(|| ".".into()),
+            None => self.storage_dir(),
         };
-        // canonicalize the path else `tempdir` might get confused with a relative path.
-        let base_dir = base_dir.canonicalize()?;
 
-        // use a "parent" directory to house all the various tmp rocksdb instances
-        Ok(base_dir.join(WORKING_DIR))
+        base_dir.join(WORKING_DIR)
     }
 }
 
@@ -1477,22 +1476,23 @@ impl PersistentState {
 
         let (tmpdir, full_path) = match params.mode {
             DurabilityMode::Permanent => {
-                let mut path = params.storage_dir.clone().unwrap_or_else(|| ".".into());
+                let mut path = params.storage_dir();
                 if !path.is_dir() {
                     fs::create_dir_all(&path)?;
                 }
                 path.push(&name);
 
                 (None, path)
             }
-            _ => {
-                let working_dir = params.derive_working_dir_base()?;
-                if !working_dir.is_dir() {
+            DurabilityMode::DeleteOnExit | DurabilityMode::MemoryOnly => {
+                let working_dir = params.working_dir_base();
+                if !working_dir.exists() {
                     fs::create_dir_all(&working_dir)?;
                 }
+                let canonical_dir = working_dir.canonicalize()?;
 
-                // create the subdirectory specific to this node
-                let tempdir = Builder::new().prefix(&name).tempdir_in(working_dir)?;
+                // create the temp subdirectory specific to this node
+                let tempdir = Builder::new().prefix(&name).tempdir_in(canonical_dir)?;
                 let path = tempdir.path().to_path_buf();
 
                 (Some(tempdir), path)