read metadata fields from _modules, not _module_versions (#25355)

stop reading the following fields from `_module_versions`: `version`, `analyzeResult`, `sourcePackageId`, `environment`.
these fields exist on `_modules` and db-verifier confirms that they match, so we read them instead.

now we only need to look up `_module_versions` when reading source code, source maps, or creation time.

GitOrigin-RevId: e18ac9a83c8f90c88dccf43de0d66051ef10a67b

Author: ldanilek

crates/application/src/application_function_runner/mod.rs

@@ -126,10 +126,13 @@ use model::{
         types::FileStorageEntry,
         FileStorageId,
     },
-    modules::module_versions::{
-        AnalyzedModule,
-        ModuleSource,
-        SourceMap,
+    modules::{
+        module_versions::{
+            AnalyzedModule,
+            ModuleSource,
+            SourceMap,
+        },
+        ModuleModel,
     },
     scheduled_jobs::VirtualSchedulerModel,
     session_requests::{
@@ -1189,16 +1192,15 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
         let udf_server_version = path_and_args.npm_version().clone();
         // We should not be missing the module given we validated the path above
         // which requires the module to exist.
-        let module_version = self
-            .module_cache
-            .get_module(&mut tx, name.module().clone())
+        let module = ModuleModel::new(&mut tx)
+            .get_metadata(name.module().clone())
             .await?
-            .context("Missing a valid module_version")?;
+            .context("Missing a valid module")?;
         let (log_line_sender, log_line_receiver) = mpsc::unbounded();
 
         let inert_identity = tx.inert_identity();
-        let timer = function_total_timer(module_version.environment, UdfType::Action);
-        let completion_result = match module_version.environment {
+        let timer = function_total_timer(module.environment, UdfType::Action);
+        let completion_result = match module.environment {
             ModuleEnvironment::Isolate => {
                 // TODO: This is the only use case of clone. We should get rid of clone,
                 // when we deprecate that codepath.
@@ -1215,7 +1217,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                             unix_timestamp,
                             context.clone(),
                             vec![log_line].into(),
-                            module_version.environment,
+                            module.environment,
                         )
                     },
                 )
@@ -1236,6 +1238,13 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                 })
             },
             ModuleEnvironment::Node => {
+                // We should not be missing the module given we validated the path above
+                // which requires the module to exist.
+                let module_version = self
+                    .module_cache
+                    .get_module(&mut tx, name.module().clone())
+                    .await?
+                    .context("Missing a valid module_version")?;
                 let _request_guard = self
                     .node_action_limiter
                     .acquire_permit_with_timeout(&self.runtime)
@@ -1245,7 +1254,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                     source_maps.insert(name.module().clone(), source_map);
                 }
 
-                let source_package_id = module_version.source_package_id.ok_or_else(|| {
+                let source_package_id = module.source_package_id.ok_or_else(|| {
                     anyhow::anyhow!("Source package is required to execute actions")
                 })?;
                 let source_package = SourcePackageModel::new(&mut tx)
@@ -1320,7 +1329,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                             unix_timestamp,
                             context.clone(),
                             vec![log_line].into(),
-                            module_version.environment,
+                            module.environment,
                         )
                     },
                 )
@@ -1366,8 +1375,8 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                 Ok(ActionCompletion {
                     outcome,
                     execution_time: start.elapsed(),
-                    environment: module_version.environment,
-                    memory_in_mb: match module_version.environment {
+                    environment: module.environment,
+                    memory_in_mb: match module.environment {
                         ModuleEnvironment::Isolate => (*ISOLATE_MAX_USER_HEAP_SIZE / (1 << 20))
                             .try_into()
                             .unwrap(),
@@ -2017,7 +2026,6 @@ impl<RT: Runtime> ActionCallbacks for ApplicationFunctionRunner<RT> {
             // Scheduling from actions is not transaction and happens at latest
             // timestamp.
             self.database.runtime().unix_timestamp(),
-            &self.module_cache,
             &mut tx,
         )
         .await?;

crates/application/src/module_cache/mod.rs

@@ -240,7 +240,7 @@ impl<RT: Runtime> ModuleLoader<RT> for ModuleCache<RT> {
         // read same is the same regardless if we hit the cache or not.
         // This is not technically needed since the module version is immutable,
         // but better safe and consistent that sorry.
-        ModuleModel::new(tx).record_module_version_read_dependency(key.0, key.1)?;
+        ModuleModel::new(tx).record_module_version_read_dependency(key.0)?;
 
         timer.finish();
         Ok(Some(result))

crates/function_runner/src/module_cache.rs

@@ -98,8 +98,7 @@ impl<RT: Runtime> ModuleLoader<RT> for FunctionRunnerModuleLoader<RT> {
         // read same is the same regardless if we hit the cache or not.
         // This is not technically needed since the module version is immutable,
         // but better safe and consistent that sorry.
-        ModuleModel::new(tx)
-            .record_module_version_read_dependency(key.module_id, key.module_version)?;
+        ModuleModel::new(tx).record_module_version_read_dependency(key.module_id)?;
 
         Ok(Some(result))
     }

crates/isolate/src/environment/action/phase.rs

@@ -23,6 +23,7 @@ use model::{
     },
     modules::{
         module_versions::ModuleVersionMetadata,
+        types::ModuleMetadata,
         ModuleModel,
     },
     udf_config::UdfConfigModel,
@@ -71,7 +72,7 @@ enum ActionPreloaded<RT: Runtime> {
     },
     Preloading,
     Ready {
-        modules: BTreeMap<CanonicalizedModulePath, Arc<ModuleVersionMetadata>>,
+        modules: BTreeMap<CanonicalizedModulePath, (ModuleMetadata, Arc<ModuleVersionMetadata>)>,
         env_vars: BTreeMap<EnvVarName, EnvVarValue>,
         rng: Option<ChaCha12Rng>,
         import_time_unix_timestamp: Option<UnixTimestamp>,
@@ -136,10 +137,10 @@ impl<RT: Runtime> ActionPhase<RT> {
             }
             let path = metadata.path.clone();
             if let Some(module) = module_loader
-                .get_module_with_metadata(&mut tx, metadata)
+                .get_module_with_metadata(&mut tx, metadata.clone())
                 .await?
             {
-                modules.insert(path, module);
+                modules.insert(path, (metadata.into_value(), module));
             }
         }
 
@@ -171,20 +172,20 @@ impl<RT: Runtime> ActionPhase<RT> {
         let ActionPreloaded::Ready { ref modules, .. } = self.preloaded else {
             anyhow::bail!("Phase not initialized");
         };
-        let module_version = modules
+        let module = modules
             .get(&module_path.clone().canonicalize())
-            .map(|m| (**m).clone());
+            .map(|(module, source)| (module, (**source).clone()));
 
-        if let Some(module_version) = module_version.as_ref() {
+        if let Some((module, _)) = module.as_ref() {
             anyhow::ensure!(
-                module_version.environment == ModuleEnvironment::Isolate,
+                module.environment == ModuleEnvironment::Isolate,
                 "Trying to execute {:?} in isolate, but it is bundled for {:?}.",
                 module_path,
-                module_version.environment
+                module.environment
             );
         };
 
-        Ok(module_version)
+        Ok(module.map(|(_, source)| source))
     }
 
     pub fn begin_execution(&mut self) -> anyhow::Result<()> {

crates/isolate/src/environment/helpers/module_loader.rs

@@ -58,7 +58,10 @@ pub trait ModuleLoader<RT: Runtime>: Sync + Send + 'static {
         tx: &mut Transaction<RT>,
         udf_path: &CanonicalizedUdfPath,
     ) -> anyhow::Result<anyhow::Result<AnalyzedFunction>> {
-        let Some(module) = self.get_module(tx, udf_path.module().clone()).await? else {
+        let Some(module) = ModuleModel::new(tx)
+            .get_metadata(udf_path.module().clone())
+            .await?
+        else {
             return Ok(Err(ErrorMetadata::bad_request(
                 "ModuleNotFound",
                 ModuleNotFoundError::new(udf_path.module().as_str()).to_string(),
@@ -69,7 +72,6 @@ pub trait ModuleLoader<RT: Runtime>: Sync + Send + 'static {
         // Dependency modules don't have AnalyzedModule.
         if !udf_path.module().is_deps() {
             let analyzed_module = module
-                .as_ref()
                 .analyze_result
                 .as_ref()
                 .ok_or_else(|| anyhow::anyhow!("Expected analyze result for {udf_path:?}"))?;

crates/isolate/src/environment/helpers/validation.rs

@@ -34,7 +34,10 @@ use model::{
         DISABLED_ERROR_MESSAGE,
         PAUSED_ERROR_MESSAGE,
     },
-    modules::module_versions::Visibility,
+    modules::{
+        module_versions::Visibility,
+        ModuleModel,
+    },
     udf_config::UdfConfigModel,
 };
 #[cfg(any(test, feature = "testing"))]
@@ -61,7 +64,6 @@ pub async fn validate_schedule_args<RT: Runtime>(
     udf_args: Vec<JsonValue>,
     scheduled_ts: UnixTimestamp,
     udf_ts: UnixTimestamp,
-    module_loader: &Arc<dyn ModuleLoader<RT>>,
     tx: &mut Transaction<RT>,
 ) -> anyhow::Result<(UdfPath, ConvexArray)> {
     // We validate the following mostly so the developer don't get the timestamp
@@ -88,8 +90,8 @@ pub async fn validate_schedule_args<RT: Runtime>(
     // We do it here instead of within transaction in order to leverage the module
     // cache.
     let canonicalized = udf_path.clone().canonicalize();
-    let module_version = module_loader
-        .get_module(tx, canonicalized.module().clone())
+    let module = ModuleModel::new(tx)
+        .get_metadata(canonicalized.module().clone())
         .await?
         .with_context(|| {
             let p = String::from(udf_path.module().clone());
@@ -104,7 +106,7 @@ pub async fn validate_schedule_args<RT: Runtime>(
     // of analyze, we should always validate in practice. We will tighten
     // the interface and make AnalyzedResult non-optional in the future.
     let function_name = canonicalized.function_name();
-    if let Some(analyze_result) = &module_version.analyze_result {
+    if let Some(analyze_result) = &module.analyze_result {
         let found = analyze_result
             .functions
             .iter()

crates/isolate/src/environment/udf/async_syscall.rs

@@ -344,7 +344,6 @@ impl<RT: Runtime> AsyncSyscallProvider<RT> for DatabaseUdfEnvironment<RT> {
             args,
             scheduled_ts,
             self.phase.unix_timestamp()?,
-            &self.module_loader,
             self.phase.tx()?,
         )
         .await

crates/isolate/src/environment/udf/mod.rs

@@ -179,7 +179,6 @@ pub struct DatabaseUdfEnvironment<RT: Runtime> {
     udf_server_version: Option<semver::Version>,
 
     phase: UdfPhase<RT>,
-    module_loader: Arc<dyn ModuleLoader<RT>>,
     file_storage: TransactionalFileStorage<RT>,
 
     query_manager: QueryManager<RT>,
@@ -353,7 +352,6 @@ impl<RT: Runtime> DatabaseUdfEnvironment<RT> {
             udf_server_version,
 
             phase: UdfPhase::new(transaction, rt, module_loader.clone(), system_env_vars),
-            module_loader,
             file_storage,
 
             query_manager: QueryManager::new(),

crates/isolate/src/environment/udf/phase.rs

@@ -31,7 +31,10 @@ use model::{
         EnvironmentVariablesModel,
         PreloadedEnvironmentVariables,
     },
-    modules::module_versions::ModuleVersionMetadata,
+    modules::{
+        module_versions::ModuleVersionMetadata,
+        ModuleModel,
+    },
     udf_config::UdfConfigModel,
 };
 use rand::SeedableRng;
@@ -145,6 +148,12 @@ impl<RT: Runtime> UdfPhase<RT> {
                 format!("Can't dynamically import {module_path:?} in a query or mutation")
             ));
         }
+        let module = with_release_permit(
+            timeout,
+            permit_slot,
+            ModuleModel::new(&mut self.tx).get_metadata(module_path.clone().canonicalize()),
+        )
+        .await?;
         let module_version = with_release_permit(
             timeout,
             permit_slot,
@@ -153,12 +162,12 @@ impl<RT: Runtime> UdfPhase<RT> {
         )
         .await?;
 
-        if let Some(module_version) = module_version.as_ref() {
+        if let Some(module) = module.as_ref() {
             anyhow::ensure!(
-                module_version.environment == ModuleEnvironment::Isolate,
+                module.environment == ModuleEnvironment::Isolate,
                 "Trying to execute {:?} in isolate, but it is bundled for {:?}.",
                 module_path,
-                module_version.environment
+                module.environment
             );
         };
 

crates/isolate/src/isolate2/runner.rs

@@ -546,7 +546,6 @@ async fn run_request<RT: Runtime>(
         shared,
         key_broker,
         execution_context,
-        module_loader,
     );
     let r: anyhow::Result<_> = try {
         // Update our shared state with the updated table mappings before reentering
@@ -783,8 +782,6 @@ struct Isolate2SyscallProvider<'a, RT: Runtime> {
 
     key_broker: KeyBroker,
     context: ExecutionContext,
-
-    module_loader: Arc<dyn ModuleLoader<RT>>,
 }
 
 impl<'a, RT: Runtime> Isolate2SyscallProvider<'a, RT> {
@@ -797,7 +794,6 @@ impl<'a, RT: Runtime> Isolate2SyscallProvider<'a, RT> {
         shared: UdfShared<RT>,
         key_broker: KeyBroker,
         context: ExecutionContext,
-        module_loader: Arc<dyn ModuleLoader<RT>>,
     ) -> Self {
         Self {
             tx,
@@ -810,7 +806,6 @@ impl<'a, RT: Runtime> Isolate2SyscallProvider<'a, RT> {
             syscall_trace: SyscallTrace::new(),
             key_broker,
             context,
-            module_loader,
         }
     }
 }
@@ -868,15 +863,7 @@ impl<'a, RT: Runtime> AsyncSyscallProvider<RT> for Isolate2SyscallProvider<'a, R
         args: Vec<JsonValue>,
         scheduled_ts: UnixTimestamp,
     ) -> anyhow::Result<(UdfPath, ConvexArray)> {
-        validate_schedule_args(
-            udf_path,
-            args,
-            scheduled_ts,
-            self.unix_timestamp,
-            &self.module_loader,
-            self.tx,
-        )
-        .await
+        validate_schedule_args(udf_path, args, scheduled_ts, self.unix_timestamp, self.tx).await
     }
 
     fn file_storage_generate_upload_url(&self) -> anyhow::Result<String> {