readyset: Verify estimated snapshot size at startup

When starting Readyset, try to determine how much space a snapshot of
the upstream will consume, and fail the startup verification if it looks
as though we won't have enough space to accommodate the entire snapshot.

Our verification relies on a few assumptions:

- The reported sizes of the upstream tables are basically correct
- All databases (MySQL) or schemas (PostgreSQL) will be replicated
- The tables will take about twice the space when stored in Readyset

Skip this verification check if the user has specified any tables via
the --replication-tables or --replication-tables-ignore arguments.  We
can consider a more granular approach in the future.

Fixes: REA-5937
Release-Note-Core: Add pre-snapshot verification checks to help find
  potential configuration issues earlier. The argument --verify can be
  used to run only the verification checks and then exit. The argument
  --verify-skip can be used to skip the verification checks and continue
  with normal startup.
Change-Id: I1a59c7bad465f87b59f1c1ce6062616d779579bb
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10601
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: davisjc

Cargo.lock

@@ -132,6 +132,7 @@ derive_more = { version = "2.0.1", features = [
     "display",
 ] }
 diff = "0.1.13"
+dir-size = "0.1"
 enum-display-derive = "0.1.1"
 enum-kinds = "0.5.1"
 enum_dispatch = "0.3.13"
@@ -241,6 +242,7 @@ streaming-iterator = "0.1"
 stringprep = "0.1.4"
 strum = { version = "0.27", features = ["derive"] }
 syn = { version = "2.0.87", default-features = false }
+sysinfo = "0.37"
 temp-dir = "0.1"
 tempfile = "3.10.1"
 test-strategy = "0.4.3"

Cargo.toml

@@ -746,11 +746,11 @@ pub struct WorkerOptions {
 }
 
 impl WorkerOptions {
-    pub fn storage_dir(&self) -> Option<PathBuf> {
-        if let Some(s) = &self.storage_dir {
-            return Some(s.to_path_buf());
-        }
-        None
+    pub fn storage_dir(&self, deployment: &str) -> PathBuf {
+        self.storage_dir
+            .clone()
+            .unwrap_or_else(|| PathBuf::from("."))
+            .join(deployment)
     }
 
     pub fn working_dir(&self) -> Option<PathBuf> {
@@ -837,14 +837,22 @@ mod tests {
     /// Test setting the storage directory
     #[test]
     fn storage_and_db_dirs() {
+        let deployment = "readyset.db";
+
         // test the unset variant
         let worker_opts = Wrapper::parse_from(["test"]).worker_opts;
-        assert_eq!(None, worker_opts.storage_dir());
+        assert_eq!(
+            PathBuf::from(".").join(deployment),
+            worker_opts.storage_dir(deployment)
+        );
 
         // the test the --storage-dir cli flag
         let storage_dir = "/tmp/cli-flag-storage";
         let worker_opts = Wrapper::parse_from(["test", "--storage-dir", storage_dir]).worker_opts;
-        assert_eq!(Some(PathBuf::from(storage_dir)), worker_opts.storage_dir());
+        assert_eq!(
+            PathBuf::from(storage_dir).join(deployment),
+            worker_opts.storage_dir(deployment)
+        );
     }
 
     #[test]

readyset-server/src/lib.rs

@@ -1,5 +1,4 @@
 use std::net::{IpAddr, SocketAddr};
-use std::path::PathBuf;
 use std::process;
 use std::sync::Arc;
 use std::time::Duration;
@@ -221,12 +220,7 @@ fn main() -> anyhow::Result<()> {
         info!(%volume_id);
     }
 
-    let deployment_dir = opts
-        .worker_options
-        .storage_dir()
-        .unwrap_or_else(|| PathBuf::from("."))
-        .join(&opts.deployment);
-
+    let deployment_dir = opts.worker_options.storage_dir(&opts.deployment);
     let authority = opts.authority.clone();
     let authority_addr = match authority {
         AuthorityType::Standalone => deployment_dir

readyset-server/src/main.rs

@@ -14,13 +14,15 @@ antithesis_sdk = { workspace = true }
 anyhow = { workspace = true }
 clap = { workspace = true, features = ["derive", "env"] }
 crossbeam-skiplist = { workspace = true }
+dir-size = { workspace = true }
 futures-util = { workspace = true }
 fail = { workspace = true }
 failpoint-macros = { path = "../failpoint-macros" }
 metrics = { workspace = true }
 mysql_common = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
+sysinfo = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tokio-native-tls = { workspace = true }

readyset/Cargo.toml

@@ -793,10 +793,7 @@ where
 
         let deployment_dir = options
             .server_worker_options
-            .storage_dir()
-            .unwrap_or_else(|| PathBuf::from("."))
-            .join(&options.deployment);
-
+            .storage_dir(&options.deployment);
         let upstream_config = options.server_worker_options.replicator_config.clone();
 
         if options.cleanup {