Rotate V8 isolate on heap growth or fragmentation (#4684)

# Description of Changes

While testing #4663, I discovered the server would crash from a V8 out
of memory error after processing many requests. Before #4663, this would
not happen. I theorized that because we now have a single JS worker that
can process an unbounded number of reducer calls over its lifetime, any
V8 heap retention that would previously have been spread across several
pooled isolates now accumulates in one isolate.

This patch now periodically collects heap statistics and forces GC or
replaces the isolate if memory cannot be reclaimed. This greatly reduces
the risk of hitting the V8 heap limit and crashing the server. It
doesn't remove the risk entirely however. But this risk was still
present before we switched to a single worker model in #4663. In order
to remove the risk of crashing the server entirely, we would need to run
V8 in a separate process.

# API and ABI breaking changes

None

# Expected complexity level and risk

3

# Testing

TBD

Author: joshua-spacetime

Cargo.lock

@@ -64,6 +64,7 @@ hex.workspace = true
 hostname.workspace = true
 http.workspace = true
 http-body-util.workspace = true
+humantime.workspace = true
 hyper.workspace = true
 imara-diff.workspace = true
 indexmap.workspace = true

crates/core/Cargo.toml

@@ -1,6 +1,8 @@
 use std::path::Path;
+use std::time::Duration;
 use std::{fmt, io};
 
+use serde::Deserialize;
 use spacetimedb_lib::ConnectionId;
 use spacetimedb_paths::cli::{ConfigDir, PrivKeyPath, PubKeyPath};
 use spacetimedb_paths::server::{ConfigToml, MetadataTomlPath};
@@ -131,6 +133,8 @@ pub struct ConfigFile {
     pub certificate_authority: Option<CertificateAuthority>,
     #[serde(default)]
     pub logs: LogConfig,
+    #[serde(default)]
+    pub v8_heap_policy: V8HeapPolicyConfig,
 }
 
 impl ConfigFile {
@@ -169,6 +173,139 @@ pub struct LogConfig {
     pub directives: Vec<String>,
 }
 
+#[derive(Clone, Copy, Debug, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct V8HeapPolicyConfig {
+    #[serde(default = "def_req_interval", deserialize_with = "de_nz_u64")]
+    pub heap_check_request_interval: Option<u64>,
+    #[serde(default = "def_time_interval", deserialize_with = "de_nz_duration")]
+    pub heap_check_time_interval: Option<Duration>,
+    #[serde(default = "def_gc_trigger", deserialize_with = "de_fraction")]
+    pub heap_gc_trigger_fraction: f64,
+    #[serde(default = "def_retire", deserialize_with = "de_fraction")]
+    pub heap_retire_fraction: f64,
+    #[serde(default, rename = "heap-limit-mb", deserialize_with = "de_limit_mb")]
+    pub heap_limit_bytes: Option<usize>,
+}
+
+impl Default for V8HeapPolicyConfig {
+    fn default() -> Self {
+        Self {
+            heap_check_request_interval: def_req_interval(),
+            heap_check_time_interval: def_time_interval(),
+            heap_gc_trigger_fraction: def_gc_trigger(),
+            heap_retire_fraction: def_retire(),
+            heap_limit_bytes: None,
+        }
+    }
+}
+
+impl V8HeapPolicyConfig {
+    pub fn normalized(mut self) -> Self {
+        if self.heap_retire_fraction < self.heap_gc_trigger_fraction {
+            log::warn!(
+                "v8-heap-policy.heap-retire-fraction ({}) is below \
+                 v8-heap-policy.heap-gc-trigger-fraction ({}); using the GC trigger fraction for both",
+                self.heap_retire_fraction,
+                self.heap_gc_trigger_fraction,
+            );
+            self.heap_retire_fraction = self.heap_gc_trigger_fraction;
+        }
+
+        self
+    }
+}
+
+/// Default number of requests between V8 heap checks.
+fn def_req_interval() -> Option<u64> {
+    Some(65_536)
+}
+
+/// Default wall-clock interval between V8 heap checks.
+fn def_time_interval() -> Option<Duration> {
+    Some(Duration::from_secs(30))
+}
+
+/// Default heap fill fraction that triggers a GC.
+fn def_gc_trigger() -> f64 {
+    0.67
+}
+
+/// Default heap fill fraction that retires the worker after a GC.
+fn def_retire() -> f64 {
+    0.75
+}
+
+fn de_nz_u64<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let value = u64::deserialize(deserializer)?;
+    Ok((value != 0).then_some(value))
+}
+
+fn de_nz_duration<'de, D>(deserializer: D) -> Result<Option<Duration>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    #[derive(serde::Deserialize)]
+    #[serde(untagged)]
+    enum DurationValue {
+        String(String),
+        Seconds(u64),
+    }
+
+    let duration = match DurationValue::deserialize(deserializer)? {
+        DurationValue::String(value) => humantime::parse_duration(&value).map_err(serde::de::Error::custom)?,
+        DurationValue::Seconds(value) => Duration::from_secs(value),
+    };
+
+    Ok((!duration.is_zero()).then_some(duration))
+}
+
+fn de_fraction<'de, D>(deserializer: D) -> Result<f64, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    #[derive(serde::Deserialize)]
+    #[serde(untagged)]
+    enum FractionValue {
+        Integer(u64),
+        Float(f64),
+    }
+
+    let value = match FractionValue::deserialize(deserializer)? {
+        FractionValue::Integer(value) => value as f64,
+        FractionValue::Float(value) => value,
+    };
+
+    if value.is_finite() && (0.0..=1.0).contains(&value) {
+        Ok(value)
+    } else {
+        Err(serde::de::Error::custom(format!(
+            "expected a fraction between 0.0 and 1.0, got {value}"
+        )))
+    }
+}
+
+fn de_limit_mb<'de, D>(deserializer: D) -> Result<Option<usize>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let value = u64::deserialize(deserializer)?;
+    if value == 0 {
+        return Ok(None);
+    }
+
+    let bytes = value
+        .checked_mul(1024 * 1024)
+        .ok_or_else(|| serde::de::Error::custom("heap-limit-mb is too large"))?;
+
+    usize::try_from(bytes)
+        .map(Some)
+        .map_err(|_| serde::de::Error::custom("heap-limit-mb does not fit in usize"))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -270,4 +407,41 @@ mod tests {
             .check_compatibility_and_update(mkmeta_pre(2, 1, 0, "rc1"))
             .unwrap_err();
     }
+
+    #[test]
+    fn v8_heap_policy_defaults_when_omitted() {
+        let config: ConfigFile = toml::from_str("").unwrap();
+
+        assert_eq!(config.v8_heap_policy.heap_check_request_interval, Some(65_536));
+        assert_eq!(
+            config.v8_heap_policy.heap_check_time_interval,
+            Some(Duration::from_secs(30))
+        );
+        assert_eq!(config.v8_heap_policy.heap_gc_trigger_fraction, 0.67);
+        assert_eq!(config.v8_heap_policy.heap_retire_fraction, 0.75);
+        assert_eq!(config.v8_heap_policy.heap_limit_bytes, None);
+    }
+
+    #[test]
+    fn v8_heap_policy_parses_from_toml() {
+        let toml = r#"
+            [v8-heap-policy]
+            heap-check-request-interval = 0
+            heap-check-time-interval = "45s"
+            heap-gc-trigger-fraction = 0.6
+            heap-retire-fraction = 0.8
+            heap-limit-mb = 256
+        "#;
+
+        let config: ConfigFile = toml::from_str(toml).unwrap();
+
+        assert_eq!(config.v8_heap_policy.heap_check_request_interval, None);
+        assert_eq!(
+            config.v8_heap_policy.heap_check_time_interval,
+            Some(Duration::from_secs(45))
+        );
+        assert_eq!(config.v8_heap_policy.heap_gc_trigger_fraction, 0.6);
+        assert_eq!(config.v8_heap_policy.heap_retire_fraction, 0.8);
+        assert_eq!(config.v8_heap_policy.heap_limit_bytes, Some(256 * 1024 * 1024));
+    }
 }

crates/core/src/config.rs

@@ -3,6 +3,7 @@ use super::scheduler::SchedulerStarter;
 use super::wasmtime::WasmtimeRuntime;
 use super::{Scheduler, UpdateDatabaseResult};
 use crate::client::{ClientActorId, ClientName};
+use crate::config::V8HeapPolicyConfig;
 use crate::database_logger::DatabaseLogger;
 use crate::db::persistence::PersistenceProvider;
 use crate::db::relational_db::{self, spawn_view_cleanup_loop, DiskSizeFn, RelationalDB, Txdata};
@@ -124,9 +125,9 @@ pub(crate) struct HostRuntimes {
 }
 
 impl HostRuntimes {
-    fn new(data_dir: Option<&ServerDataDir>) -> Arc<Self> {
+    fn new(data_dir: Option<&ServerDataDir>, v8_heap_policy: V8HeapPolicyConfig) -> Arc<Self> {
         let wasmtime = WasmtimeRuntime::new(data_dir);
-        let v8 = V8Runtime::default();
+        let v8 = V8Runtime::new(v8_heap_policy);
         Arc::new(Self { wasmtime, v8 })
     }
 }
@@ -210,6 +211,7 @@ impl HostController {
     pub fn new(
         data_dir: Arc<ServerDataDir>,
         default_config: db::Config,
+        v8_heap_policy: V8HeapPolicyConfig,
         program_storage: ProgramStorage,
         energy_monitor: Arc<impl EnergyMonitor>,
         persistence: Arc<dyn PersistenceProvider>,
@@ -221,7 +223,7 @@ impl HostController {
             program_storage,
             energy_monitor,
             persistence,
-            runtimes: HostRuntimes::new(Some(&data_dir)),
+            runtimes: HostRuntimes::new(Some(&data_dir), v8_heap_policy),
             data_dir,
             page_pool: PagePool::new(default_config.page_pool_max_size),
             bsatn_rlb_pool: BsatnRowListBuilderPool::new(),
@@ -1387,7 +1389,7 @@ pub async fn extract_schema(program_bytes: Box<[u8]>, host_type: HostType) -> an
     extract_schema_with_pools(
         PagePool::new(None),
         BsatnRowListBuilderPool::new(),
-        &HostRuntimes::new(None),
+        &HostRuntimes::new(None, V8HeapPolicyConfig::default()),
         program_bytes,
         host_type,
     )
@@ -1421,4 +1423,12 @@ where
         .data_size_blob_store_bytes_used_by_blobs
         .remove_label_values(db);
     let _ = WORKER_METRICS.wasm_memory_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_total_heap_size_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_total_physical_size_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_used_global_handles_size_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_used_heap_size_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_heap_size_limit_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_external_memory_bytes.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_native_contexts.remove_label_values(db);
+    let _ = WORKER_METRICS.v8_detached_contexts.remove_label_values(db);
 }

crates/core/src/host/host_controller.rs

@@ -347,8 +347,8 @@ pub enum ModuleWithInstance {
 }
 
 enum ModuleHostInner {
-    Wasm(WasmtimeModuleHost),
-    Js(V8ModuleHost),
+    Wasm(Box<WasmtimeModuleHost>),
+    Js(Box<V8ModuleHost>),
 }
 
 struct WasmtimeModuleHost {
@@ -1070,20 +1070,20 @@ impl ModuleHost {
             } => {
                 info = module.info();
                 let instance_manager = ModuleInstanceManager::new(module, Some(init_inst), database_identity);
-                Arc::new(ModuleHostInner::Wasm(WasmtimeModuleHost {
+                Arc::new(ModuleHostInner::Wasm(Box::new(WasmtimeModuleHost {
                     executor,
                     instance_manager,
-                }))
+                })))
             }
             ModuleWithInstance::Js { module, init_inst } => {
                 info = module.info();
                 let instance_lane = super::v8::JsInstanceLane::new(module.clone(), init_inst);
                 let procedure_instances = ModuleInstanceManager::new(module.clone(), None, database_identity);
-                Arc::new(ModuleHostInner::Js(V8ModuleHost {
+                Arc::new(ModuleHostInner::Js(Box::new(V8ModuleHost {
                     module,
                     instance_lane,
                     procedure_instances,
-                }))
+                })))
             }
         };
         let on_panic = Arc::new(on_panic);
@@ -1143,15 +1143,17 @@ impl ModuleHost {
         let timer_guard = self.start_call_timer(label);
 
         Ok(match &*self.inner {
-            ModuleHostInner::Wasm(WasmtimeModuleHost { executor, .. }) => {
+            ModuleHostInner::Wasm(wasm) => {
+                let executor = &wasm.executor;
                 executor
                     .run_job(async move || {
                         drop(timer_guard);
                         f().await
                     })
                     .await
             }
-            ModuleHostInner::Js(V8ModuleHost { instance_lane, .. }) => {
+            ModuleHostInner::Js(js) => {
+                let instance_lane = &js.instance_lane;
                 instance_lane
                     .run_on_thread(async move || {
                         drop(timer_guard);
@@ -1215,15 +1217,14 @@ impl ModuleHost {
         });
 
         Ok(match &*self.inner {
-            ModuleHostInner::Wasm(WasmtimeModuleHost {
-                executor,
-                instance_manager,
-            }) => {
+            ModuleHostInner::Wasm(wasm) => {
+                let executor = &wasm.executor;
+                let instance_manager = &wasm.instance_manager;
                 instance_manager
                     .with_instance(async |inst| work_wasm(timer_guard, executor, inst, arg).await)
                     .await
             }
-            ModuleHostInner::Js(V8ModuleHost { instance_lane, .. }) => work_js(timer_guard, instance_lane, arg).await,
+            ModuleHostInner::Js(js) => work_js(timer_guard, &js.instance_lane, arg).await,
         })
     }
 
@@ -1294,13 +1295,10 @@ impl ModuleHost {
         });
 
         Ok(match &*self.inner {
-            ModuleHostInner::Wasm(WasmtimeModuleHost {
-                executor,
-                instance_manager,
-            }) => {
-                instance_manager
+            ModuleHostInner::Wasm(host) => {
+                host.instance_manager
                     .with_instance(async |mut inst| {
-                        executor
+                        host.executor
                             .run_job(async move || {
                                 drop(timer_guard);
                                 (wasm(arg, &mut inst).await, inst)
@@ -1309,10 +1307,8 @@ impl ModuleHost {
                     })
                     .await
             }
-            ModuleHostInner::Js(V8ModuleHost {
-                procedure_instances, ..
-            }) => {
-                procedure_instances
+            ModuleHostInner::Js(host) => {
+                host.procedure_instances
                     .with_instance(async |inst| {
                         drop(timer_guard);
                         let res = js(arg, &inst).await;