snapshot: Remote synchronization

Cargo.toml

@@ -249,6 +249,7 @@ thin-vec = "0.2.13"
 thiserror = "1.0.37"
 tokio = { version = "1.37", features = ["full"] }
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
+tokio-stream = "0.1.17"
 tokio-tungstenite = { version = "0.26.2", features = ["native-tls"] }
 tokio-util = { version = "0.7.4", features = ["time"] }
 toml = "0.8"

crates/core/src/db/relational_db.rs

@@ -1591,7 +1591,7 @@ pub mod tests_utils {
             Ok((db, local))
         }
 
-        fn open_db(
+        pub fn open_db(
             root: &ReplicaDir,
             history: impl durability::History<TxData = Txdata>,
             durability: Option<(Arc<Durability>, DiskSizeFn)>,

crates/fs-utils/src/compression.rs

@@ -4,7 +4,7 @@ use std::io::{BufReader, Read, Seek, SeekFrom};
 use zstd_framed;
 use zstd_framed::{ZstdReader, ZstdWriter};
 
-const ZSTD_MAGIC_BYTES: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD];
+pub const ZSTD_MAGIC_BYTES: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD];
 
 /// Helper struct to keep track of the number of files compressed using each algorithm
 #[derive(Debug, Copy, Clone, PartialEq, Default)]

crates/fs-utils/src/dir_trie.rs

@@ -75,7 +75,7 @@ impl DirTrie {
         Ok(Self { root })
     }
 
-    fn file_path(&self, file_id: &FileId) -> PathBuf {
+    pub fn file_path(&self, file_id: &FileId) -> PathBuf {
         // TODO(perf, bikeshedding): avoid allocating a `String`.
         let file_id_hex = hex::encode(file_id);
 

crates/snapshot/Cargo.toml

@@ -16,6 +16,21 @@ spacetimedb-paths.workspace = true
 spacetimedb-fs-utils.workspace = true
 
 blake3.workspace = true
+bytes.workspace = true
+futures.workspace = true
 hex.workspace = true
 log.workspace = true
+tempfile.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["io-util"] }
+tokio-stream.workspace = true
+tokio-util = { workspace = true, features = ["io"] }
+zstd-framed.workspace = true
+
+[dev-dependencies]
+spacetimedb-core = { workspace = true, features = ["test"] }
+spacetimedb-schema.workspace = true
+
+anyhow.workspace = true
+env_logger.workspace = true
+pretty_assertions = { workspace = true, features = ["unstable"] }

crates/snapshot/src/lib.rs

@@ -48,6 +48,8 @@ use std::{
     path::PathBuf,
 };
 
+pub mod remote;
+
 #[derive(Debug, Copy, Clone)]
 /// An object which may be associated with an error during snapshotting.
 pub enum ObjectType {
@@ -139,21 +141,21 @@ pub const SNAPSHOT_FILE_EXT: &str = "snapshot_bsatn";
 /// File extension of snapshots which have been marked invalid by [`SnapshotRepository::invalidate_newer_snapshots`].
 pub const INVALID_SNAPSHOT_DIR_EXT: &str = "invalid_snapshot";
 
-#[derive(Serialize, Deserialize)]
+#[derive(Clone, Serialize, Deserialize)]
 /// The hash and refcount of a single blob in the blob store.
 struct BlobEntry {
     hash: BlobHash,
     uses: u32,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Clone, Serialize, Deserialize)]
 /// A snapshot of a single table, containing the hashes of all its resident pages.
 struct TableEntry {
     table_id: TableId,
     pages: Vec<blake3::Hash>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Clone, Serialize, Deserialize)]
 pub struct Snapshot {
     /// A magic number: must be equal to [`MAGIC`].
     magic: [u8; 4],
@@ -452,6 +454,11 @@ impl Snapshot {
             .collect()
     }
 
+    /// The number of objects in this snapshot, both blobs and pages.
+    pub fn total_objects(&self) -> usize {
+        self.blobs.len() + self.tables.iter().map(|table| table.pages.len()).sum::<usize>()
+    }
+
     /// Obtain an iterator over the [`blake3::Hash`]es of all objects
     /// this snapshot is referring to.
     pub fn objects(&self) -> impl Iterator<Item = blake3::Hash> + '_ {