refactor(common): modularize Js UDF handling with plan an exec phases

crates/bin/ampd/Cargo.toml

@@ -24,6 +24,7 @@ clap.workspace = true
 common = { path = "../../core/common" }
 console-subscriber = { version = "0.5.0", default-features = false, optional = true }
 controller = { path = "../../services/controller" }
+js-runtime = { path = "../../core/js-runtime" }
 metadata-db = { path = "../../core/metadata-db" }
 metadata-db-postgres = { path = "../../services/metadata-db-postgres" }
 monitoring = { path = "../../core/monitoring" }

crates/bin/ampd/src/server_cmd.rs

@@ -78,12 +78,14 @@ pub async fn run(
         None
     };
 
+    let isolate_pool = js_runtime::isolate_pool::IsolatePool::new();
     let (addrs, server) = server::service::new(
         Arc::new(server_config),
         metadata_db,
         data_store,
         datasets_cache,
         ethcall_udfs_cache,
+        isolate_pool,
         meter,
         flight_at,
         jsonl_at,

crates/bin/ampd/src/solo_cmd.rs

@@ -151,12 +151,14 @@ pub async fn run(
         };
 
         let server_config = server_cmd::config_from_common(&config);
+        let server_isolate_pool = js_runtime::isolate_pool::IsolatePool::new();
         let (addrs, fut) = server::service::new(
             Arc::new(server_config),
             metadata_db.clone(),
             data_store.clone(),
             datasets_cache.clone(),
             ethcall_udfs_cache.clone(),
+            server_isolate_pool,
             meter.clone(),
             flight_at,
             jsonl_at,
@@ -178,13 +180,16 @@ pub async fn run(
 
     // Initialize worker
     let worker_config = worker_cmd::config_from_common(&config);
+    let isolate_pool = js_runtime::isolate_pool::IsolatePool::new();
+
     let worker_fut = worker::service::new(
         worker_config,
         build_info,
         metadata_db,
         data_store,
         datasets_cache,
         ethcall_udfs_cache,
+        isolate_pool,
         meter,
         worker_id,
         None, // Use config-based event emitter

crates/bin/ampd/src/worker_cmd.rs

@@ -59,13 +59,16 @@ pub async fn run(
     let worker_config = config_from_common(&config);
 
     // Initialize the worker (setup phase)
+    let isolate_pool = js_runtime::isolate_pool::IsolatePool::new();
+
     let worker_fut = worker::service::new(
         worker_config,
         build_info,
         metadata_db,
         data_store,
         datasets_cache,
         ethcall_udfs_cache,
+        isolate_pool,
         meter,
         node_id,
         None, // Use config-based event emitter

crates/core/common/src/amp_catalog_provider.rs

@@ -18,7 +18,6 @@ use datasets_common::{
     hash_reference::HashReference, partial_reference::PartialReference, reference::Reference,
 };
 use datasets_derived::deps::SELF_REF_KEYWORD;
-use js_runtime::isolate_pool::IsolatePool;
 
 use crate::{
     dataset_schema_provider::DatasetSchemaProvider,
@@ -53,7 +52,6 @@ pub const AMP_CATALOG_NAME: &str = "amp";
 pub struct AmpCatalogProvider {
     datasets_cache: DatasetsCache,
     ethcall_udfs_cache: EthCallUdfsCache,
-    isolate_pool: IsolatePool,
     /// Optional dependency alias overrides. When set, bare names matching
     /// a key are resolved directly to the corresponding [`HashReference`]
     /// instead of going through `PartialReference` → `Reference` → `resolve_revision`.
@@ -65,15 +63,10 @@ pub struct AmpCatalogProvider {
 
 impl AmpCatalogProvider {
     /// Creates a new catalog provider.
-    pub fn new(
-        datasets_cache: DatasetsCache,
-        ethcall_udfs_cache: EthCallUdfsCache,
-        isolate_pool: IsolatePool,
-    ) -> Self {
+    pub fn new(datasets_cache: DatasetsCache, ethcall_udfs_cache: EthCallUdfsCache) -> Self {
         Self {
             datasets_cache,
             ethcall_udfs_cache,
-            isolate_pool,
             dep_aliases: Default::default(),
             self_schema: None,
         }
@@ -127,7 +120,6 @@ impl AmpCatalogProvider {
                 name.to_string(),
                 dataset,
                 self.ethcall_udfs_cache.clone(),
-                self.isolate_pool.clone(),
             ));
             return Ok(Some(provider));
         }
@@ -158,7 +150,6 @@ impl AmpCatalogProvider {
             name.to_string(),
             dataset,
             self.ethcall_udfs_cache.clone(),
-            self.isolate_pool.clone(),
         ));
         Ok(Some(provider))
     }