Lower MAX_USER_MODULES to 4096 but make it actually work (#36478)

There were a couple issues:
- Audit logging creates a ConvexArray with one entry per user module, which implies a hard limit of 8192
- Also, we would hit a TooManyReads error when replacing too many modules in one transaction.

The former is fixed by lowering MAX_USER_MODULES to be safely within the limit - 4096 for now.
This could be raised a little if necessary, but otherwise we would have to change the audit log schema.

The latter is fixed by a somewhat hacky change: since we already read out all the modules from the database,
we can read the `by_id` index instead of `by_creation_time` (which is used by full_table_scan). Then,
the additional read ranges created by `Transaction::replace` will merge with the full-table range,
instead of creating many small ones.

GitOrigin-RevId: 70d43a529b669e59af76fbb87502e61cd265f340

Author: goffrie

crates/application/src/deploy_config.rs

@@ -3,7 +3,10 @@ use std::{
         BTreeMap,
         BTreeSet,
     },
-    time::Duration,
+    time::{
+        Duration,
+        Instant,
+    },
 };
 
 use anyhow::Context;
@@ -29,16 +32,19 @@ use common::{
         Runtime,
         UnixTimestamp,
     },
+    schemas::DatabaseSchema,
     types::{
         EnvVarName,
         EnvVarValue,
         ModuleEnvironment,
         NodeDependency,
     },
+    version::Version,
 };
 use database::{
     BootstrapComponentsModel,
     IndexModel,
+    OccRetryStats,
     Token,
     WriteSource,
     SCHEMAS_TABLE,
@@ -120,7 +126,26 @@ use value::{
     TableNamespace,
 };
 
-use crate::Application;
+use crate::{
+    Application,
+    ApplyConfigArgs,
+    ConfigMetadataAndSchema,
+};
+
+pub struct PushAnalytics {
+    pub config: ConfigMetadata,
+    pub modules: Vec<ModuleConfig>,
+    pub udf_server_version: Version,
+    pub analyze_results: BTreeMap<CanonicalizedModulePath, AnalyzedModule>,
+    pub schema: Option<DatabaseSchema>,
+}
+
+pub struct PushMetrics {
+    pub build_external_deps_time: Duration,
+    pub upload_source_package_time: Duration,
+    pub analyze_time: Duration,
+    pub occ_stats: OccRetryStats,
+}
 
 impl<RT: Runtime> Application<RT> {
     #[fastrace::trace]
@@ -622,6 +647,102 @@ impl<RT: Runtime> Application<RT> {
 
         Ok(diff)
     }
+
+    /// N.B.: does not check auth
+    pub async fn push_config_no_components(
+        &self,
+        identity: Identity,
+        config_file: ConfigFile,
+        modules: Vec<ModuleConfig>,
+        udf_server_version: Version,
+        schema_id: Option<String>,
+        node_dependencies: Option<Vec<NodeDependencyJson>>,
+    ) -> anyhow::Result<(PushAnalytics, PushMetrics)> {
+        let begin_build_external_deps = Instant::now();
+        // Upload external node dependencies separately
+        let external_deps_id_and_pkg = if let Some(deps) = node_dependencies
+            && !deps.is_empty()
+        {
+            let deps: Vec<_> = deps.into_iter().map(NodeDependency::from).collect();
+            Some(self.build_external_node_deps(deps).await?)
+        } else {
+            None
+        };
+        let end_build_external_deps = Instant::now();
+        let external_deps_pkg_size = external_deps_id_and_pkg
+            .as_ref()
+            .map(|(_, pkg)| pkg.package_size)
+            .unwrap_or_default();
+
+        let source_package = self
+            .upload_package(&modules, external_deps_id_and_pkg)
+            .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.package_size + external_deps_pkg_size;
+        combined_pkg_size.verify_size()?;
+
+        let udf_config = UdfConfig {
+            server_version: udf_server_version,
+            // Generate a new seed and timestamp to be used at import time.
+            import_phase_rng_seed: self.runtime.rng().random(),
+            import_phase_unix_timestamp: self.runtime.unix_timestamp(),
+        };
+        let begin_analyze = Instant::now();
+        // Note: This is not transactional with the rest of the deploy to avoid keeping
+        // a transaction open for a long time.
+        let mut tx = self.begin(Identity::system()).await?;
+        let user_environment_variables = EnvironmentVariablesModel::new(&mut tx).get_all().await?;
+        let system_env_var_overrides = system_env_var_overrides(&mut tx).await?;
+        drop(tx);
+        // Run analyze to make sure the new modules are valid.
+        let (auth_module, analyze_results) = self
+            .analyze_modules_with_auth_config(
+                udf_config.clone(),
+                modules.clone(),
+                source_package.clone(),
+                user_environment_variables,
+                system_env_var_overrides,
+            )
+            .await?;
+        let end_analyze = Instant::now();
+        let (
+            ConfigMetadataAndSchema {
+                config_metadata,
+                schema,
+            },
+            occ_stats,
+        ) = self
+            .apply_config_with_retries(
+                identity.clone(),
+                ApplyConfigArgs {
+                    auth_module,
+                    config_file,
+                    schema_id,
+                    modules: modules.clone(),
+                    udf_config: udf_config.clone(),
+                    source_package,
+                    analyze_results: analyze_results.clone(),
+                },
+            )
+            .await?;
+
+        Ok((
+            PushAnalytics {
+                config: config_metadata,
+                modules,
+                udf_server_version: udf_config.server_version,
+                analyze_results,
+                schema,
+            },
+            PushMetrics {
+                build_external_deps_time: end_build_external_deps - begin_build_external_deps,
+                upload_source_package_time: end_upload_source_package - end_build_external_deps,
+                analyze_time: end_analyze - begin_analyze,
+                occ_stats,
+            },
+        ))
+    }
 }
 
 struct ApplicationInitializerEvaluator<'a, RT: Runtime> {

crates/application/src/tests/mod.rs

@@ -10,6 +10,7 @@ mod http_action;
 mod indexes;
 mod mutation;
 mod occ_retries;
+mod push;
 mod query_cache;
 mod returns_validation;
 mod scheduled_jobs;

crates/application/src/tests/push.rs

@@ -0,0 +1,80 @@
+use common::knobs::MAX_USER_MODULES;
+use keybroker::Identity;
+use model::config::types::ConfigFile;
+use runtime::testing::TestRuntime;
+
+use crate::{
+    deploy_config::{
+        AppDefinitionConfigJson,
+        ModuleJson,
+        StartPushRequest,
+    },
+    test_helpers::ApplicationTestExt as _,
+    Application,
+};
+
+fn make_modules() -> Vec<ModuleJson> {
+    let mut functions: Vec<_> = (0..*MAX_USER_MODULES)
+        .map(|i| ModuleJson {
+            environment: None,
+            source_map: None,
+            path: format!("mod{i}.js"),
+            source: format!("// {i}"),
+        })
+        .collect();
+    functions.extend((0..*MAX_USER_MODULES).map(|i| ModuleJson {
+        environment: None,
+        source_map: None,
+        path: format!("_deps/mod{i}.js"),
+        source: format!("// dep {i}"),
+    }));
+    functions
+}
+
+#[convex_macro::test_runtime]
+async fn test_max_size_push(rt: TestRuntime) -> anyhow::Result<()> {
+    let application = Application::new_for_tests(&rt).await?;
+    for _ in 0..2 {
+        application
+            .run_test_push(StartPushRequest {
+                admin_key: "".into(),
+                dry_run: false,
+                functions: "convex/".into(),
+                app_definition: AppDefinitionConfigJson {
+                    definition: None,
+                    dependencies: vec![],
+                    schema: None,
+                    functions: make_modules(),
+                    udf_server_version: "1.3939.3939".into(),
+                },
+                component_definitions: vec![],
+                node_dependencies: vec![],
+            })
+            .await?;
+    }
+    Ok(())
+}
+
+#[convex_macro::test_runtime]
+async fn test_max_size_push_no_components(rt: TestRuntime) -> anyhow::Result<()> {
+    let application = Application::new_for_tests(&rt).await?;
+    for _ in 0..2 {
+        application
+            .push_config_no_components(
+                Identity::system(),
+                ConfigFile {
+                    auth_info: None,
+                    functions: "convex/".into(),
+                },
+                make_modules()
+                    .into_iter()
+                    .map(|m| m.try_into().unwrap())
+                    .collect(),
+                "1.3939.3939".parse().unwrap(),
+                None,
+                None,
+            )
+            .await?;
+    }
+    Ok(())
+}

crates/common/src/knobs.rs

@@ -1151,7 +1151,7 @@ pub static MAX_ECHO_BYTES: LazyLock<usize> =
 
 /// The limit on the number of user modules in a push bundle.
 pub static MAX_USER_MODULES: LazyLock<usize> =
-    LazyLock::new(|| env_config("MAX_USER_MODULES", 10000));
+    LazyLock::new(|| env_config("MAX_USER_MODULES", 4096));
 
 /// Percentage of request traces that should sampled.
 ///

crates/isolate/src/environment/analyze.rs

@@ -325,7 +325,7 @@ impl AnalyzeEnvironment {
                 let module_config = self
                     .modules
                     .get(path)
-                    .ok_or(anyhow!("could not find module config in environment"))?;
+                    .context("could not find module config in environment")?;
                 let source_map = module_config
                     .source_map
                     .as_ref()