Use database-wide eviction instead of per-table

crates/datastore/src/error.rs

@@ -51,6 +51,24 @@ pub enum DatastoreError {
     Other(#[from] anyhow::Error),
 }
 
+impl DatastoreError {
+    pub(crate) fn is_resident_memory_limit(&self) -> bool {
+        match self {
+            Self::Page(PageError::MemoryLimitExceeded(_))
+            | Self::Blob(BlobError::MemoryLimitExceeded(_))
+            | Self::Table(TableError::Bflatn(bflatn_to::Error::Pages(PageError::MemoryLimitExceeded(_))))
+            | Self::Table(TableError::Bflatn(bflatn_to::Error::Blobs(BlobError::MemoryLimitExceeded(_)))) => true,
+            Self::TableStorage(error) | Self::Table(TableError::Bflatn(bflatn_to::Error::Table(error))) => {
+                match error {
+                    table::TableError::Page(error) => matches!(error.as_ref(), PageError::MemoryLimitExceeded(_)),
+                    table::TableError::Blob(error) => matches!(error.as_ref(), BlobError::MemoryLimitExceeded(_)),
+                }
+            }
+            _ => false,
+        }
+    }
+}
+
 #[derive(Error, Debug)]
 pub enum ViewError {
     #[error("view '{0}' not found")]

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -691,7 +691,17 @@ impl CommittedState {
         view_instances: ViewInstanceTxState,
         ctx: &ExecutionContext,
     ) -> Result<TxData> {
-        let prepared = match self.prepare_merge(&tx_state) {
+        let preparation = self.prepare_merge(&tx_state).or_else(|error| {
+            if error.is_resident_memory_limit() {
+                // Preparation hit the memory limit.
+                // Try to evict resident pages and then prepare again.
+                self.trim_resident()?;
+                self.prepare_merge(&tx_state)
+            } else {
+                Err(error)
+            }
+        });
+        let prepared = match preparation {
             Ok(prepared) => prepared,
             Err(error) => {
                 self.rollback(tx_state);
@@ -1112,6 +1122,25 @@ impl CommittedState {
             .set(self.storage.blobs().bytes_used_by_blobs() as _);
     }
 
+    /// Evict objects from the cache until the resident size is below the low watermark,
+    /// or no more objects can be evicted.
+    ///
+    /// TODO: Use a database-scoped LRU eviction policy instead of evicting table by table.
+    pub(super) fn trim_resident(&self) -> Result<()> {
+        let pages = self.storage.pages();
+        let target = pages.resident_budget().low_water_bytes;
+        for table in self.tables.values() {
+            if pages.resident_usage().accounted_bytes() <= target {
+                return Ok(());
+            }
+            table.trim()?;
+        }
+        if pages.resident_usage().accounted_bytes() > target {
+            self.storage.blobs().trim()?;
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "durability")]
     pub(super) fn trim(&self) -> std::result::Result<(), crate::error::TrimTableError> {
         for table in self.tables.values() {

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -4038,8 +4038,21 @@ fn insert_physically_maybe_generate<'a, const GENERATE: bool>(
     };
 
     // 1. Insert the physical row.
-    let (tx_row_ref, blob_bytes) = tx_table.insert_physically_bsatn(tx_blob_store, row)?;
-    let tx_pinned_row = tx_row_ref.into_pinned_row();
+    let insert_physical = |table: &mut Table, blobs: &mut BlobSet| {
+        table
+            .insert_physically_bsatn(blobs, row)
+            .map(|(row, bytes)| (row.into_pinned_row(), bytes))
+            .map_err(DatastoreError::from)
+    };
+    let (tx_pinned_row, blob_bytes) = match insert_physical(tx_table, tx_blob_store) {
+        Err(error) if error.is_resident_memory_limit() => {
+            // The insertion hit the memory limit.
+            // Try to evict resident pages and insert again.
+            committed_state.trim_resident()?;
+            insert_physical(tx_table, tx_blob_store)?
+        }
+        result => result?,
+    };
     // 2. Optionally: Detect, generate, write sequence values.
     let (tx_parts, gen_cols) = if GENERATE {
         // When `GENERATE` is enabled, we're instructed to deal with sequence value generation.

crates/table/src/bflatn_to.rs

@@ -628,12 +628,15 @@ pub mod test {
         let row = spacetimedb_sats::product![first.clone(), second.clone()];
         let mut page = Page::new(ty.size());
 
-        let error = unsafe { write_row_to_page(&mut page, &blobs, &visitor, &ty, &row, EvictionPolicy::Evictable) }
-            .unwrap_err();
-        assert!(matches!(error, Error::Blobs(BlobError::MemoryLimitExceeded(_))));
-        assert_eq!(page.num_rows(), 0);
-        assert_eq!(blobs.num_blobs(), 1);
-        assert_eq!(blobs.usage_counter().get(&first_hash), Some(&1));
+        // Repeat the failure to exercise granules reused from the freelist.
+        for _ in 0..2 {
+            let error = unsafe { write_row_to_page(&mut page, &blobs, &visitor, &ty, &row, EvictionPolicy::Evictable) }
+                .unwrap_err();
+            assert!(matches!(error, Error::Blobs(BlobError::MemoryLimitExceeded(_))));
+            assert_eq!(page.num_rows(), 0);
+            assert_eq!(blobs.num_blobs(), 1);
+            assert_eq!(blobs.usage_counter().get(&first_hash), Some(&1));
+        }
 
         // The failed row's page allocations must also be reusable.
         drop(first_pin);

crates/table/src/page.rs

@@ -730,12 +730,15 @@ impl<'page> VarView<'page> {
     /// Allocates a granule for a large blob object
     /// and returns a [`VarLenRef`] pointing to that granule.
     ///
-    /// The granule is not initialized by this method, and contains valid-unconstrained bytes.
-    /// It is the caller's responsibility to initialize it with a [`BlobHash`](super::blob_hash::BlobHash).
+    /// Initialize an empty header so cleanup can free the granule if writing the blob fails.
+    /// The caller must still write the blob hash before reading the granule's data.
     #[cold]
     fn alloc_blob_hash(&mut self) -> Result<VarLenRef, Error> {
         // Var-len hashes are 32 bytes, which fits within a single granule.
-        self.alloc_granule().map(VarLenRef::large_blob)
+        let granule = self.alloc_granule()?;
+        // SAFETY: The allocation is unused and aligned, and NULL terminates the list.
+        unsafe { self.write_chunk_to_granule(&[], 0, granule, PageOffset::VAR_LEN_NULL) };
+        Ok(VarLenRef::large_blob(granule))
     }
 
     /// Inserts `var_len_obj` into `blob_store`

crates/table/src/tiered/page_set.rs

@@ -208,7 +208,7 @@ impl TieredPages {
         mut f: impl FnMut() -> Result<T, PageError>,
     ) -> Result<T, PageError> {
         match f() {
-            Err(PageError::MemoryLimitExceeded(_)) => {
+            Err(PageError::MemoryLimitExceeded(_)) if !self.slots.is_empty() => {
                 self.trim(manager)?;
                 f()
             }

crates/table/src/var_len.rs

@@ -145,11 +145,8 @@ impl VarLenGranuleHeader {
     /// However, a granule can never store more than [`VarLenGranule::DATA_SIZE`] (`62`),
     /// which is `2` less than `2^6`.
     ///
-    /// We will also never allocate a `VarLenGranule` with len 0.
-    ///
-    /// This means that the `len` field of a `VarLenGranule` has two dead states,
-    /// 0 and 63. We could use these as sentinels,
-    /// but currently have no use for them.
+    /// Completed granules have a nonzero length. A reserved blob-hash granule
+    /// has length 0 until the hash is written; length 63 remains unused.
     const LEN_BITMASK: u16 = (1 << Self::LEN_BITS) - 1;
 
     /// The [`LEN_BITMASK`] will preserve all granule lengths possible.