diff --git a/crates/datastore/src/error.rs b/crates/datastore/src/error.rs index c844b89ac7d..163379dacea 100644 --- a/crates/datastore/src/error.rs +++ b/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")] diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs index 28e4aafb0f1..433d305823b 100644 --- a/crates/datastore/src/locking_tx_datastore/committed_state.rs +++ b/crates/datastore/src/locking_tx_datastore/committed_state.rs @@ -691,7 +691,17 @@ impl CommittedState { view_instances: ViewInstanceTxState, ctx: &ExecutionContext, ) -> Result { - 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() { diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 25467f34199..3710e3a76a9 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/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. diff --git a/crates/table/src/bflatn_to.rs b/crates/table/src/bflatn_to.rs index 0e6753ea03d..fd7807167b2 100644 --- a/crates/table/src/bflatn_to.rs +++ b/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); diff --git a/crates/table/src/page.rs b/crates/table/src/page.rs index 5e52d723d3b..01c3bf77966 100644 --- a/crates/table/src/page.rs +++ b/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 { // 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` diff --git a/crates/table/src/tiered/page_set.rs b/crates/table/src/tiered/page_set.rs index d84dfdd33d8..b8987921215 100644 --- a/crates/table/src/tiered/page_set.rs +++ b/crates/table/src/tiered/page_set.rs @@ -208,7 +208,7 @@ impl TieredPages { mut f: impl FnMut() -> Result, ) -> Result { match f() { - Err(PageError::MemoryLimitExceeded(_)) => { + Err(PageError::MemoryLimitExceeded(_)) if !self.slots.is_empty() => { self.trim(manager)?; f() } diff --git a/crates/table/src/var_len.rs b/crates/table/src/var_len.rs index c331fe783b6..7af87e6630d 100644 --- a/crates/table/src/var_len.rs +++ b/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.