Make local node executor run a server instead of a one-shot process (#34778)

This PR makes node actions in self-hosted backend way faster. 100ms -> 2ms for a simple action that logs a line.

Instead of starting a one-shot node process to execute every single node action, which requires downloading the source and any external packages specified, this change starts a node server in `LocalNodeExecutor` that caches source and external packages similar to how the AWS lambda implementation works.

GitOrigin-RevId: 0421a6182e811cd879c69738ad23222ebb32c824

Author: emmaling27

Cargo.lock

@@ -150,7 +150,6 @@ tokio = { version = "1", features = [ "full" ] }
 tokio-metrics = { version = "0.3.1" }
 tokio-metrics-collector = { version = "0.2.1" }
 tokio-postgres = { version = "0.7.10", features = [ "with-serde_json-1" ] }
-tokio-process-stream = { version = "0.4.0" }
 tokio-stream = { version = "0.1", features = [ "io-util", "sync", "signal" ] }
 tokio-tungstenite = { version = "0.26.2", features = [ "native-tls-vendored" ] }
 tokio-util = { version = "0.7.13", features = [ "io", "rt", "io-util" ] }

Cargo.toml

@@ -99,6 +99,7 @@ model = { path = "../model", features = ["testing"] }
 must-let = { workspace = true }
 node_executor = { path = "../../crates/node_executor", features = ["testing"] }
 openidconnect = { workspace = true }
+portpicker = { workspace = true }
 pretty_assertions = { workspace = true }
 proptest = { workspace = true }
 proptest-derive = { workspace = true }

crates/application/Cargo.toml

@@ -87,8 +87,9 @@ use model::{
     virtual_system_mapping,
 };
 use node_executor::{
-    local::LocalNodeExecutor,
+    noop::NoopNodeExecutor,
     Actions,
+    NodeExecutor,
 };
 use storage::Storage;
 use value::{
@@ -117,6 +118,7 @@ pub static OBJECTS_TABLE_COMPONENT: ComponentId = ComponentId::test_user();
 pub struct ApplicationFixtureArgs {
     pub tp: Option<TestPersistence>,
     pub event_logger: Option<Arc<dyn UsageEventLogger>>,
+    pub node_executor: Option<Arc<dyn NodeExecutor>>,
 }
 
 impl ApplicationFixtureArgs {
@@ -126,6 +128,13 @@ impl ApplicationFixtureArgs {
             ..Default::default()
         }
     }
+
+    pub fn with_node_executor(node_executor: Arc<dyn NodeExecutor>) -> Self {
+        Self {
+            node_executor: Some(node_executor),
+            ..Default::default()
+        }
+    }
 }
 
 #[async_trait]
@@ -227,8 +236,9 @@ impl<RT: Runtime> ApplicationTestExt<RT> for Application<RT> {
             database: database.clone(),
         };
 
-        let node_process_timeout = *ACTION_USER_TIMEOUT + Duration::from_secs(5);
-        let node_executor = Arc::new(LocalNodeExecutor::new(node_process_timeout)?);
+        let node_executor = args
+            .node_executor
+            .unwrap_or_else(|| Arc::new(NoopNodeExecutor::new()));
         let actions = Actions::new(
             node_executor,
             convex_origin.clone(),

crates/application/src/test_helpers.rs

@@ -1,4 +1,8 @@
-use std::collections::BTreeMap;
+use std::{
+    collections::BTreeMap,
+    sync::Arc,
+    time::Duration,
+};
 
 use common::types::{
     ModuleEnvironment,
@@ -8,10 +12,14 @@ use model::{
     config::types::ModuleConfig,
     udf_config::types::UdfConfig,
 };
+use node_executor::local::LocalNodeExecutor;
 use runtime::prod::ProdRuntime;
 
 use crate::{
-    test_helpers::ApplicationTestExt,
+    test_helpers::{
+        ApplicationFixtureArgs,
+        ApplicationTestExt,
+    },
     tests::NODE_SOURCE,
     Application,
 };
@@ -56,7 +64,13 @@ export default {
 // This test requires prod runtime since it analyzes node modules.
 #[convex_macro::prod_rt_test]
 async fn test_analyze(rt: ProdRuntime) -> anyhow::Result<()> {
-    let application = Application::new_for_tests(&rt).await?;
+    let application = Application::new_for_tests_with_args(
+        &rt,
+        ApplicationFixtureArgs::with_node_executor(Arc::new(
+            LocalNodeExecutor::new(Duration::from_secs(10)).await?,
+        )),
+    )
+    .await?;
     let modules = vec![
         ModuleConfig {
             path: "a.js".parse()?,
@@ -162,7 +176,13 @@ export { hello, internalHello };
 }
 "#;
 
-    let application = Application::new_for_tests(&rt).await?;
+    let application = Application::new_for_tests_with_args(
+        &rt,
+        ApplicationFixtureArgs::with_node_executor(Arc::new(
+            LocalNodeExecutor::new(Duration::from_secs(10)).await?,
+        )),
+    )
+    .await?;
     let modules = vec![
         ModuleConfig {
             path: "isolate_source.js".parse()?,
@@ -218,7 +238,13 @@ export { hello, internalHello };
 // This test requires prod runtime since it analyzes node modules.
 #[convex_macro::prod_rt_test]
 async fn test_analyze_crons(rt: ProdRuntime) -> anyhow::Result<()> {
-    let application = Application::new_for_tests(&rt).await?;
+    let application = Application::new_for_tests_with_args(
+        &rt,
+        ApplicationFixtureArgs::with_node_executor(Arc::new(
+            LocalNodeExecutor::new(Duration::from_secs(10)).await?,
+        )),
+    )
+    .await?;
     let modules = vec![
         ModuleConfig {
             path: "a.js".parse()?,

crates/application/src/tests/analyze.rs

@@ -173,7 +173,7 @@ pub async fn make_app(
     };
 
     let node_process_timeout = *ACTION_USER_TIMEOUT + Duration::from_secs(5);
-    let node_executor = Arc::new(LocalNodeExecutor::new(node_process_timeout)?);
+    let node_executor = Arc::new(LocalNodeExecutor::new(node_process_timeout).await?);
     let actions = Actions::new(
         node_executor,
         config.convex_origin_url()?,

crates/local_backend/src/lib.rs

@@ -18,21 +18,23 @@ common = { path = "../common" }
 errors = { path = "../errors" }
 fastrace = { workspace = true }
 futures = { workspace = true }
+futures-async-stream = { workspace = true }
 home = { workspace = true }
 http = { workspace = true }
 isolate = { path = "../isolate" }
 keybroker = { path = "../keybroker" }
 maplit = { workspace = true }
 metrics = { path = "../metrics" }
 model = { path = "../model" }
+portpicker = { workspace = true }
+reqwest = { version = "0.11", features = ["json"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 sourcemap = { workspace = true }
 storage = { path = "../storage" }
 sync_types = { package = "convex_sync_types", path = "../convex/sync_types" }
 tempfile = { workspace = true }
 tokio = { workspace = true }
-tokio-process-stream = { workspace = true }
 tracing = { workspace = true }
 udf = { path = "../udf" }
 value = { path = "../value" }