REPL upload source package (#26915)

instead of (in addition to, for now) storing the source code for REPL queries in `_module_versions`, store it in S3.

also remove unnecessary Options

GitOrigin-RevId: e6d6acb36e57d6556d08544491b67d6f7ae9915d

Author: ldanilek

crates/application/src/application_function_runner/mod.rs

@@ -1272,9 +1272,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                     source_maps.insert(module_path.module_path.clone(), source_map);
                 }
 
-                let source_package_id = module.source_package_id.ok_or_else(|| {
-                    anyhow::anyhow!("Source package is required to execute actions")
-                })?;
+                let source_package_id = module.source_package_id;
                 let source_package = SourcePackageModel::new(&mut tx)
                     .get(source_package_id)
                     .await?
@@ -1674,7 +1672,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
         &self,
         udf_config: UdfConfig,
         new_modules: Vec<ModuleConfig>,
-        source_package: Option<SourcePackage>,
+        source_package: SourcePackage,
     ) -> anyhow::Result<Result<BTreeMap<CanonicalizedModulePath, AnalyzedModule>, JsError>> {
         // We use the latest environment variables at the time of the deployment
         // this is not transactional with the rest of the deploy.
@@ -1715,9 +1713,6 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                     source_maps.insert(path.clone(), source_map);
                 }
             }
-            let source_package = source_package.ok_or_else(|| {
-                anyhow::anyhow!("Source package is required to analyze action modules")
-            })?;
 
             // Fetch source and external_deps presigned URI first
             let source_uri_future = self

crates/application/src/lib.rs

@@ -218,6 +218,7 @@ use model::{
     source_packages::{
         types::SourcePackage,
         upload_download::upload_package,
+        SourcePackageModel,
     },
     udf_config::{
         types::UdfConfig,
@@ -340,7 +341,7 @@ pub struct ApplyConfigArgs {
     pub schema_id: Option<String>,
     pub modules: Vec<ModuleConfig>,
     pub udf_config: UdfConfig,
-    pub source_package: Option<SourcePackage>,
+    pub source_package: SourcePackage,
     pub analyze_results: BTreeMap<CanonicalizedModulePath, AnalyzedModule>,
 }
 
@@ -1526,7 +1527,7 @@ impl<RT: Runtime> Application<RT> {
         &self,
         udf_config: UdfConfig,
         new_modules: Vec<ModuleConfig>,
-        source_package: Option<SourcePackage>,
+        source_package: SourcePackage,
     ) -> anyhow::Result<Result<BTreeMap<CanonicalizedModulePath, AnalyzedModule>, JsError>> {
         self.runner
             .analyze(udf_config, new_modules, source_package)
@@ -1746,7 +1747,7 @@ impl<RT: Runtime> Application<RT> {
                 config_metadata.clone(),
                 modules,
                 udf_config,
-                source_package,
+                Some(source_package),
                 analyze_results,
                 schema_id,
             )
@@ -1837,7 +1838,7 @@ impl<RT: Runtime> Application<RT> {
         &self,
         modules: &Vec<ModuleConfig>,
         external_deps_id_and_pkg: Option<(ExternalDepsPackageId, ExternalDepsPackage)>,
-    ) -> anyhow::Result<Option<SourcePackage>> {
+    ) -> anyhow::Result<SourcePackage> {
         // If there are any node actions, turn on the lambdas.
         if modules
             .iter()
@@ -1879,12 +1880,12 @@ impl<RT: Runtime> Application<RT> {
         tracing::info!("Source package size: {}", package_size);
         log_source_package_size_bytes_total(package_size);
 
-        Ok(Some(SourcePackage {
+        Ok(SourcePackage {
             storage_key,
             sha256,
             external_deps_package_id,
             package_size,
-        }))
+        })
     }
 
     // Clear all records for specified tables concurrently, potentially taking
@@ -1915,6 +1916,11 @@ impl<RT: Runtime> Application<RT> {
             )
             .await?;
 
+        // Write (and commit) the module source to S3.
+        // This will become a dangling reference since the _modules entry won't
+        // be committed to the database, but we have to deal with those anyway.
+        let source_package = self.upload_package(&vec![module.clone()], None).await?;
+
         let mut tx = self.begin(identity.clone()).await?;
 
         // Use the last pushed version. If there hasn't been a push
@@ -1936,7 +1942,11 @@ impl<RT: Runtime> Application<RT> {
         };
 
         let analyze_results = self
-            .analyze(udf_config.clone(), vec![module.clone()], None)
+            .analyze(
+                udf_config.clone(),
+                vec![module.clone()],
+                source_package.clone(),
+            )
             .await?
             .map_err(|js_error| {
                 let metadata = ErrorMetadata::bad_request(
@@ -1963,16 +1973,20 @@ impl<RT: Runtime> Application<RT> {
         }
         let analyzed_function = analyzed_function.context("Missing default export.")?;
 
+        let source_package_id = SourcePackageModel::new(&mut tx).put(source_package).await?;
+
         // 3. Add the module
         ModuleModel::new(&mut tx)
-            .put_standalone(
+            .put(
                 CanonicalizedComponentModulePath {
                     component: ComponentId::Root,
                     module_path: module_path.clone(),
                 },
                 module.source,
+                source_package_id,
                 module.source_map,
-                analyzed_module,
+                Some(analyzed_module),
+                ModuleEnvironment::Isolate,
             )
             .await?;
 

crates/application/src/test_helpers.rs

@@ -332,7 +332,7 @@ impl<RT: Runtime> Application<RT> {
                 ConfigMetadata::new(),
                 test_source.clone(),
                 udf_config,
-                source_package,
+                Some(source_package),
                 analyze_results,
                 Some(schema_id),
             )

crates/application/src/tests/source_package.rs

@@ -1,6 +1,5 @@
 use std::collections::BTreeMap;
 
-use anyhow::Context;
 use common::{
     components::ComponentId,
     runtime::Runtime,
@@ -49,10 +48,7 @@ async fn test_source_package(rt: ProdRuntime) -> anyhow::Result<()> {
         storage_key,
         sha256,
         ..
-    } = application
-        .upload_package(&package, None)
-        .await?
-        .context("With functions should upload")?;
+    } = application.upload_package(&package, None).await?;
 
     let result =
         download_package(application.modules_storage().clone(), storage_key, sha256).await?;

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

@@ -3,10 +3,7 @@ use std::{
     sync::Arc,
 };
 
-use anyhow::{
-    anyhow,
-    Context,
-};
+use anyhow::anyhow;
 use common::{
     components::ComponentId,
     document::ParsedDocument,
@@ -93,12 +90,10 @@ async fn download_module_source_from_package<RT: Runtime>(
     tx: &mut Transaction<RT>,
     modules_storage: Arc<dyn Storage>,
     modules_tablet: TabletId,
-    source_package_id: Option<SourcePackageId>,
+    source_package_id: SourcePackageId,
 ) -> anyhow::Result<HashMap<(ResolvedDocumentId, ModuleVersion), FullModuleSource>> {
     let mut result = HashMap::new();
-    let source_package = SourcePackageModel::new(tx)
-        .get(source_package_id.context("source package missing")?)
-        .await?;
+    let source_package = SourcePackageModel::new(tx).get(source_package_id).await?;
     let mut package = download_package(
         modules_storage,
         source_package.storage_key.clone(),

crates/isolate/src/tests/internal.rs

@@ -21,6 +21,7 @@ use keybroker::{
 };
 use model::{
     config::ConfigModel,
+    source_packages::SourcePackageModel,
     udf_config::types::UdfConfig,
 };
 use must_let::must_let;
@@ -65,6 +66,11 @@ async fn test_udf_visibility(rt: TestRuntime) -> anyhow::Result<()> {
         .isolate
         .analyze(udf_config, modules_by_path, BTreeMap::new())
         .await??;
+
+    let source_package = SourcePackageModel::new(&mut tx)
+        .get_latest()
+        .await?
+        .unwrap();
     drop(tx);
 
     // Newer version + analyze results
@@ -74,7 +80,7 @@ async fn test_udf_visibility(rt: TestRuntime) -> anyhow::Result<()> {
             config_metadata.clone(),
             module_configs.clone(),
             UdfConfig::new_for_test(&t.rt, post_internal_npm_version.clone()),
-            None,
+            Some(source_package.into_value()),
             analyze_results.clone(),
             None,
         )

crates/local_backend/src/deploy_config.rs

@@ -439,11 +439,7 @@ pub async fn push_config_handler(
         .await?;
     let end_upload_source_package = Instant::now();
     // Verify that we have not exceeded the max zipped or unzipped file size
-    let combined_pkg_size = source_package
-        .as_ref()
-        .map(|pkg| pkg.package_size)
-        .unwrap_or(PackageSize::default())
-        + external_deps_pkg_size;
+    let combined_pkg_size = source_package.package_size + external_deps_pkg_size;
     combined_pkg_size.verify_size()?;
 
     let udf_config = UdfConfig {
@@ -506,7 +502,7 @@ pub async fn analyze_modules_with_auth_config(
     application: &Application<ProdRuntime>,
     udf_config: UdfConfig,
     modules: Vec<ModuleConfig>,
-    source_package: Option<SourcePackage>,
+    source_package: SourcePackage,
 ) -> anyhow::Result<(
     Option<ModuleConfig>,
     BTreeMap<CanonicalizedModulePath, AnalyzedModule>,
@@ -536,7 +532,7 @@ pub async fn analyze_modules(
     application: &Application<ProdRuntime>,
     udf_config: UdfConfig,
     modules: Vec<ModuleConfig>,
-    source_package: Option<SourcePackage>,
+    source_package: SourcePackage,
 ) -> anyhow::Result<BTreeMap<CanonicalizedModulePath, AnalyzedModule>> {
     let num_dep_modules = modules.iter().filter(|m| m.path.is_deps()).count();
     anyhow::ensure!(

crates/local_backend/src/deploy_config2.rs

@@ -396,8 +396,7 @@ pub async fn start_push_handler(
     let app_pkg = st
         .application
         .upload_package(&app_modules, external_deps_id_and_pkg.clone())
-        .await?
-        .context("No package for app?")?;
+        .await?;
     total_size += app_pkg.package_size;
     component_definition_packages.insert(ComponentDefinitionPath::root(), app_pkg);
 
@@ -406,8 +405,7 @@ pub async fn start_push_handler(
         let component_pkg = st
             .application
             .upload_package(&component_modules, None)
-            .await?
-            .context("No package for component?")?;
+            .await?;
         total_size += component_pkg.package_size;
         anyhow::ensure!(component_definition_packages
             .insert(component_def.definition_path.clone(), component_pkg)
@@ -423,7 +421,7 @@ pub async fn start_push_handler(
         &st.application,
         config.udf_config.clone(),
         config.app_definition.functions.clone(),
-        Some(app_pkg.clone()),
+        app_pkg.clone(),
     )
     .await?;
 
@@ -475,7 +473,7 @@ pub async fn start_push_handler(
             &st.application,
             config.udf_config.clone(),
             component_def.functions.clone(),
-            Some(component_pkg.clone()),
+            component_pkg.clone(),
         )
         .await?;
         anyhow::ensure!(component_analysis_by_def_path

crates/model/src/config/module_loader.rs

@@ -74,9 +74,7 @@ pub mod test_module_loader {
             tx: &mut Transaction<RT>,
             module_metadata: ParsedDocument<ModuleMetadata>,
         ) -> anyhow::Result<Arc<FullModuleSource>> {
-            let source_package_id = module_metadata
-                .source_package_id
-                .with_context(|| format!("Missing source package {}", module_metadata.id()))?;
+            let source_package_id = module_metadata.source_package_id;
             let source_package = SourcePackageModel::new(tx)
                 .get(source_package_id)
                 .await?

crates/model/src/modules/mod.rs

@@ -229,7 +229,7 @@ impl<'a, RT: Runtime> ModuleModel<'a, RT> {
                     module_path: path.clone(),
                 },
                 module.source,
-                source_package_id,
+                source_package_id.context("missing source_package_id")?,
                 module.source_map,
                 analyze_result,
                 module.environment,
@@ -390,46 +390,12 @@ impl<'a, RT: Runtime> ModuleModel<'a, RT> {
         Ok(Some(module_metadata))
     }
 
-    /// Write a isolate-environment module to _module_versions, without source
-    /// package.
-    ///
-    /// This transaction must never be committed.
-    /// All future reads of modules on this transaction will read from the
-    /// database instead of the source package.
-    pub async fn put_standalone(
-        &mut self,
-        path: CanonicalizedComponentModulePath,
-        source: ModuleSource,
-        source_map: Option<SourceMap>,
-        analyze_result: AnalyzedModule,
-    ) -> anyhow::Result<()> {
-        if !(self.tx.identity().is_admin() || self.tx.identity().is_system()) {
-            anyhow::bail!(unauthorized_error("put_standalone_module"));
-        }
-        if path.module_path.is_system() {
-            anyhow::bail!("You cannot push a function under '_system/'");
-        }
-        let component = path.component;
-        let sha256 = hash_module_source(&source, source_map.as_ref());
-        let (module_id, version) = self
-            .put_module_metadata(
-                path,
-                None,
-                Some(analyze_result),
-                ModuleEnvironment::Isolate,
-                sha256,
-            )
-            .await?;
-        self.put_module_source_into_db(module_id, version, source, source_map, component)
-            .await
-    }
-
     /// Put a module's source at a given path.
     pub async fn put(
         &mut self,
         path: CanonicalizedComponentModulePath,
         source: ModuleSource,
-        source_package_id: Option<SourcePackageId>,
+        source_package_id: SourcePackageId,
         source_map: Option<SourceMap>,
         analyze_result: Option<AnalyzedModule>,
         environment: ModuleEnvironment,
@@ -456,7 +422,7 @@ impl<'a, RT: Runtime> ModuleModel<'a, RT> {
     async fn put_module_metadata(
         &mut self,
         path: CanonicalizedComponentModulePath,
-        source_package_id: Option<SourcePackageId>,
+        source_package_id: SourcePackageId,
         analyze_result: Option<AnalyzedModule>,
         environment: ModuleEnvironment,
         sha256: Sha256Digest,