WIP: Blobs

crates/bench/benches/special.rs

@@ -144,7 +144,7 @@ fn serialize_benchmarks<
         PageManager::new_for_test().into(),
         PageEvictionPolicy::NeverEvict,
     );
-    let mut blob_store = spacetimedb_table::blob_store::HashMapBlobStore::default();
+    let blob_store = spacetimedb_table::tiered::BlobSet::new_for_test();
 
     let ptrs = data_pv
         .elements

crates/core/src/host/host_controller.rs

@@ -50,6 +50,7 @@ use spacetimedb_sats::hash::Hash;
 use spacetimedb_schema::auto_migrate::{ponder_migrate, AutoMigrateError, MigrationPolicy, PrettyPrintStyle};
 use spacetimedb_schema::def::{ModuleDef, RawModuleDefVersion};
 use spacetimedb_table::page_pool::PagePool;
+use spacetimedb_table::tiered::{ByteBudget, MemoryObjectStore, TieredStorageConfig};
 use std::future::Future;
 use std::ops::Deref;
 use std::sync::Arc;
@@ -1251,7 +1252,7 @@ impl Host {
                 EmptyHistory::new(),
                 None,
                 Some(tx_metrics_queue),
-                page_pool.clone(),
+                TieredStorageConfig::new_for_memory_database(page_pool.clone()),
             )?,
             db::Storage::Disk => {
                 // Replay from the local state.
@@ -1262,20 +1263,26 @@ impl Host {
                     owner_identity: database.owner_identity,
                 };
                 let persistence = persistence.persistence(&persistence_db, replica_id).await?;
+                // TODO(disk-tables): Figure out where the config comes from.
+                let storage = TieredStorageConfig {
+                    page_pool: page_pool.clone(),
+                    local_object_store: Box::new(MemoryObjectStore::default()),
+                    remote_object_store: None,
+                    resident_budget: ByteBudget::unlimited(),
+                };
                 // Loading a database from persistent storage involves heavy
                 // blocking I/O. `asyncify` to avoid blocking the async worker.
                 let (db, clients) = asyncify({
                     let database_identity = database.database_identity;
                     let owner_identity = database.owner_identity;
-                    let page_pool = page_pool.clone();
                     move || {
                         RelationalDB::open(
                             database_identity,
                             owner_identity,
                             history,
                             Some(persistence),
                             Some(tx_metrics_queue),
-                            page_pool,
+                            storage,
                         )
                     }
                 })
@@ -1508,7 +1515,7 @@ impl Host {
             EmptyHistory::new(),
             None,
             None,
-            page_pool,
+            TieredStorageConfig::new_for_module_check(page_pool),
         )?;
         let memory_observer: Arc<dyn MemoryObserver> = Arc::new(());
 

crates/datastore/src/error.rs

@@ -8,10 +8,10 @@ use spacetimedb_sats::{AlgebraicType, AlgebraicValue};
 use spacetimedb_schema::def::error::LibError;
 #[cfg(feature = "durability")]
 use spacetimedb_snapshot::SnapshotError;
-use spacetimedb_table::tiered::PageError;
 use spacetimedb_table::{
     bflatn_to, read_column,
     table::{self, ReadViaBsatnError, UniqueConstraintViolation},
+    tiered::{BlobError, DeleteRowError, PageError},
 };
 use thiserror::Error;
 
@@ -39,6 +39,12 @@ pub enum DatastoreError {
     #[error(transparent)]
     Page(#[from] PageError),
 
+    #[error(transparent)]
+    TableStorage(#[from] table::TableError),
+
+    #[error(transparent)]
+    Blob(#[from] BlobError),
+
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
@@ -146,7 +152,7 @@ impl From<table::InsertError> for DatastoreError {
             table::InsertError::Duplicate(e) => TableError::from(e).into(),
             table::InsertError::Bflatn(e) => TableError::from(e).into(),
             table::InsertError::IndexError(e) => IndexError::from(e).into(),
-            table::InsertError::Page(e) => e.into(),
+            table::InsertError::Table(e) => e.into(),
         }
     }
 }
@@ -157,6 +163,12 @@ impl From<bflatn_to::Error> for DatastoreError {
     }
 }
 
+impl From<DeleteRowError> for DatastoreError {
+    fn from(err: DeleteRowError) -> Self {
+        table::TableError::from(err).into()
+    }
+}
+
 #[cfg(feature = "durability")]
 impl From<SnapshotError> for DatastoreError {
     fn from(e: SnapshotError) -> Self {