perf(host): export only the requested plugin package

platform/event-log/src/applications.rs

@@ -326,6 +326,62 @@ impl EventLog {
             .await
     }
 
+    /// Exact bytes of one package in a committed revision, without reading or
+    /// validating the revision's other packages. `None` when the revision does
+    /// not contain that package.
+    pub async fn load_application_package_at(
+        &self,
+        source: &DeploymentRef,
+        package: &PackageRef,
+    ) -> Result<Option<Package>, StoreError> {
+        source.validate().map_err(invalid)?;
+        if source.state_domain != self.application_domain()? {
+            return Err(invalid("foreign application state domain"));
+        }
+        package
+            .validate()
+            .map_err(|error| invalid(error.to_string()))?;
+        let deployment = source.deployment.to_string();
+        let revision = source.revision as i64;
+        let requested = package.clone();
+        let files = self
+            .connection
+            .run(move |connection| {
+                Box::pin(async move {
+                    let mut tx = connection.begin().await?;
+                    let installed: bool = sqlx::query_scalar(
+                        "SELECT EXISTS(SELECT 1 FROM application_source_packages
+                            WHERE deployment = ? AND source_revision = ? AND id = ? AND revision = ?)",
+                    )
+                    .bind(&deployment)
+                    .bind(revision)
+                    .bind(&requested.id)
+                    .bind(&requested.revision)
+                    .fetch_one(&mut *tx)
+                    .await?;
+                    let files = if installed {
+                        Some(read_package_files(&mut tx, &requested).await?)
+                    } else {
+                        None
+                    };
+                    tx.commit().await?;
+                    Ok(files)
+                })
+            })
+            .await?;
+        let Some(files) = files else {
+            return Ok(None);
+        };
+        let package = package.clone();
+        tokio::task::spawn_blocking(move || build_checked_package(files, &package).map(Some))
+            .await
+            .map_err(|error| {
+                invalid(format!(
+                    "application package validation worker failed: {error}"
+                ))
+            })?
+    }
+
     async fn read_metadata(
         &self,
         deployment: Uuid,
@@ -357,7 +413,8 @@ impl EventLog {
             return Err(invalid("invalid deployment"));
         }
         let owner = std::sync::Arc::downgrade(self.root_owner.as_ref().expect("owned source Root"));
-        self.connection
+        let raw = self
+            .connection
             .run(move |connection| {
                 Box::pin(async move {
                     let mut tx = connection.begin().await?;
@@ -366,28 +423,43 @@ impl EventLog {
                     else {
                         return Ok(None);
                     };
-                    let mut packages = Vec::new();
-                    for reference in metadata.packages {
-                        let package = read_package(&mut tx, &reference).await?;
-                        packages.push(package);
+                    let mut raw_packages = Vec::with_capacity(metadata.packages.len());
+                    for reference in &metadata.packages {
+                        let files = read_package_files(&mut tx, reference).await?;
+                        raw_packages.push((reference.clone(), files));
                     }
-                    let plan = validate(
-                        &metadata.source.input,
-                        &metadata.source.remote_exports,
-                        &packages,
-                    )?;
                     tx.commit().await?;
-                    let snapshot =
-                        SourceSnapshot::new(owner, metadata.source.reference.clone(), &plan);
-                    Ok(Some(LoadedDeployment {
-                        snapshot,
-                        source: metadata.source,
-                        packages,
-                        plan,
-                    }))
+                    Ok(Some((metadata, raw_packages)))
                 })
             })
-            .await
+            .await?;
+        let Some((metadata, raw_packages)) = raw else {
+            return Ok(None);
+        };
+        tokio::task::spawn_blocking(move || {
+            let mut packages = Vec::with_capacity(raw_packages.len());
+            for (reference, files) in raw_packages {
+                packages.push(build_checked_package(files, &reference)?);
+            }
+            let plan = validate(
+                &metadata.source.input,
+                &metadata.source.remote_exports,
+                &packages,
+            )?;
+            let snapshot = SourceSnapshot::new(owner, metadata.source.reference.clone(), &plan);
+            Ok(Some(LoadedDeployment {
+                snapshot,
+                source: metadata.source,
+                packages,
+                plan,
+            }))
+        })
+        .await
+        .map_err(|error| {
+            invalid(format!(
+                "application source validation worker failed: {error}"
+            ))
+        })?
     }
 }
 
@@ -461,38 +533,52 @@ async fn read_metadata(
     }))
 }
 
-async fn read_package(
+async fn read_package_files(
     connection: &mut SqliteConnection,
     reference: &PackageRef,
-) -> Result<Package, StoreError> {
-    let rows = sqlx::query("SELECT path,
-        CASE WHEN typeof(payload) = 'blob' THEN length(payload) END AS bytes FROM application_package_files WHERE id = ? AND revision = ? ORDER BY path")
-        .bind(&reference.id).bind(&reference.revision).fetch_all(&mut *connection).await?;
+) -> Result<BTreeMap<String, Vec<u8>>, StoreError> {
+    let rows = sqlx::query(
+        "SELECT path,
+            CASE WHEN typeof(payload) = 'blob' THEN payload END AS payload
+            FROM application_package_files WHERE id = ? AND revision = ? ORDER BY path",
+    )
+    .bind(&reference.id)
+    .bind(&reference.revision)
+    .fetch_all(&mut *connection)
+    .await?;
     if rows.is_empty() {
         return Err(invalid("invalid stored application package file count"));
     }
-    for row in &rows {
-        usize::try_from(
-            row.try_get::<Option<i64>, _>("bytes")?
-                .ok_or_else(|| invalid("stored package payload must be a BLOB"))?,
-        )
-        .map_err(|_| invalid("invalid package payload length"))?;
-        let path: &str = row.try_get("path")?;
-        maka_packages::validate_path(path).map_err(|e| invalid(e.to_string()))?;
-    }
     let mut files = BTreeMap::new();
     for row in rows {
         let path: String = row.try_get("path")?;
-        let bytes: Vec<u8> = sqlx::query_scalar("SELECT payload FROM application_package_files WHERE id = ? AND revision = ? AND path = ?")
-            .bind(&reference.id).bind(&reference.revision).bind(&path).fetch_one(&mut *connection).await?;
+        maka_packages::validate_path(&path).map_err(|e| invalid(e.to_string()))?;
+        let bytes: Vec<u8> = row
+            .try_get::<Option<Vec<u8>>, _>("payload")?
+            .ok_or_else(|| invalid("stored package payload must be a BLOB"))?;
         files.insert(path, bytes);
     }
+    Ok(files)
+}
+
+fn build_checked_package(
+    files: BTreeMap<String, Vec<u8>>,
+    reference: &PackageRef,
+) -> Result<Package, StoreError> {
     let package = Package::new(files).map_err(|e| invalid(e.to_string()))?;
     if &package.application_ref() != reference {
         return Err(invalid("application package digest mismatch"));
     }
     Ok(package)
 }
 
+async fn read_package(
+    connection: &mut SqliteConnection,
+    reference: &PackageRef,
+) -> Result<Package, StoreError> {
+    let files = read_package_files(connection, reference).await?;
+    build_checked_package(files, reference)
+}
+
 #[cfg(test)]
 mod tests;