apt: unknown architectures and uploads past the component cap are refused

crates/core/src/format.rs

@@ -143,6 +143,14 @@ pub trait Format: Send + Sync {
         None
     }
 
+    /// The fields of [`Self::manifest_summary`] of which a hosted repository
+    /// holds a bounded number of distinct values
+    /// ([`forge_plugin_api::FormatPlugin::summary_caps`]): a publish or an
+    /// import that would add a value past one is refused. None by default.
+    fn summary_caps(&self) -> &'static [forge_plugin_api::SummaryCap] {
+        &[]
+    }
+
     /// Whether `id` is a well-formed package id of this format. Read routes
     /// answer 404 for anything else, and a proxy never puts such an id into
     /// an upstream URL: an id is pasted into URL paths, so `/`, `..`, `?`,

crates/core/src/repo/hosted.rs

@@ -876,6 +876,16 @@ impl HostedWrite {
             ))
         };
 
+        // What the version's summary will be: an import adds what the source
+        // says of it (apt's component).
+        let mut manifest_summary = manifest_summary;
+        if let (Some(serde_json::Value::Object(summary)), Some(fields)) = (
+            manifest_summary.as_mut(),
+            import.and_then(|options| options.summary_fields.as_ref()),
+        ) {
+            summary.extend(fields.clone());
+        }
+
         // An import of a package the source named otherwise stores nothing.
         if let Some((id, version)) = import.and_then(|options| options.expected.as_ref()) {
             let same_version = self
@@ -908,6 +918,21 @@ impl HostedWrite {
             policy.authorize(context).await?;
         }
 
+        // A version past one of the format's caps is refused before anything
+        // reaches S3 (and an import's plan says so); the check in the
+        // transaction below stays the authority.
+        {
+            let mut conn = self.pool.acquire().await.map_err(RepoError::from)?;
+            self.check_summary_caps(
+                &mut conn,
+                repo_row.id,
+                &parsed,
+                manifest_summary.as_ref(),
+                false,
+            )
+            .await?;
+        }
+
         // A duplicate without redeploy is answered before anything reaches
         // S3: a CI job that re-pushes an unchanged version with
         // `--skip-duplicate` on every build would otherwise upload it each
@@ -977,6 +1002,14 @@ impl HostedWrite {
         let mut retried = false;
         let (mut tx, component_id, replaced, outcome) = loop {
             let mut tx = forge_db::begin(&self.pool).await.map_err(RepoError::from)?;
+            self.check_summary_caps(
+                &mut tx,
+                repo_row.id,
+                &parsed,
+                manifest_summary.as_ref(),
+                true,
+            )
+            .await?;
 
             // Read the duplicate-check *inside* the transaction so it serializes
             // with the subsequent insert/redeploy write against concurrent pushes
@@ -1085,13 +1118,6 @@ impl HostedWrite {
             }
         };
 
-        let mut manifest_summary = manifest_summary;
-        if let (Some(serde_json::Value::Object(summary)), Some(fields)) = (
-            manifest_summary.as_mut(),
-            import.and_then(|options| options.summary_fields.as_ref()),
-        ) {
-            summary.extend(fields.clone());
-        }
         if let (Some(id), Some(summary)) = (component_id, manifest_summary.as_ref()) {
             component_dao::set_manifest_summary(&mut *tx, id, summary)
                 .await
@@ -1151,6 +1177,69 @@ impl HostedWrite {
         done(outcome)
     }
 
+    /// Refuse a version whose summary would add a value to the repository
+    /// past one of the format's caps ([`Format::summary_caps`]: apt's
+    /// components), naming the cap. A value the repository's other versions
+    /// hold already (the one a redeploy replaces left out) passes. With
+    /// `lock`, in the publish transaction, a new value is counted under the
+    /// repository's value lock
+    /// ([`repository_dao::lock_repository_values`]), so that two publishes
+    /// of new values never both take the last place.
+    async fn check_summary_caps(
+        &self,
+        conn: &mut sqlx::PgConnection,
+        repository_id: i64,
+        parsed: &crate::domain::ParsedUpload,
+        summary: Option<&serde_json::Value>,
+        lock: bool,
+    ) -> Result<(), WriteError> {
+        for cap in self.format.summary_caps() {
+            let Some(value) = summary
+                .and_then(|s| s.get(cap.field))
+                .and_then(|v| v.as_str())
+            else {
+                continue;
+            };
+            let mut values = component_dao::summary_values(
+                &mut *conn,
+                repository_id,
+                cap.field,
+                value,
+                &parsed.id,
+                &parsed.version,
+            )
+            .await
+            .map_err(RepoError::from)?;
+            if !values.0 && lock {
+                repository_dao::lock_repository_values(&mut *conn, repository_id)
+                    .await
+                    .map_err(RepoError::from)?;
+                values = component_dao::summary_values(
+                    &mut *conn,
+                    repository_id,
+                    cap.field,
+                    value,
+                    &parsed.id,
+                    &parsed.version,
+                )
+                .await
+                .map_err(RepoError::from)?;
+            }
+            let (present, count) = values;
+            if !present && usize::try_from(count).unwrap_or(usize::MAX) >= cap.max {
+                return Err(WriteError::InvalidPackage(
+                    crate::domain::FormatError::InvalidPackage(format!(
+                        "the repository holds {count} {what}, the most it may hold ({max}): \
+                         {value:?} would be one more",
+                        what = cap.what,
+                        max = cap.max,
+                    )),
+                ));
+            }
+        }
+        Ok(())
+    }
+
     /// Insert a brand-new component + `.nupkg`/`.nuspec` assets, increfing both
     /// blobs, all within `tx`.
     #[allow(clippy::too_many_arguments)]

crates/db/src/component_dao.rs

@@ -575,6 +575,45 @@ where
     Ok(())
 }
 
+/// The distinct values of the string field `field` of the manifest
+/// summaries of `repository_id`'s versions, leaving out version `name`
+/// `version` (one a redeploy replaces): whether `value` is among them, and
+/// how many there are. A publish checks a format's caps on them
+/// (`forge_plugin_api::SummaryCap`).
+pub async fn summary_values<'e, E>(
+    executor: E,
+    repository_id: i64,
+    field: &str,
+    value: &str,
+    name: &str,
+    version: &str,
+) -> sqlx::Result<(bool, i64)>
+where
+    E: PgExecutor<'e>,
+{
+    // A runtime query: the field is a parameter, which the offline data of
+    // the macros checks no better.
+    let (present, count): (Option<bool>, i64) = sqlx::query_as(
+        r#"
+        SELECT bool_or(v = $3), COUNT(DISTINCT v)
+        FROM (
+            SELECT manifest_summary->>$2 AS v
+            FROM component
+            WHERE repository_id = $1 AND NOT (name = $4 AND version = $5)
+        ) AS summary
+        WHERE v IS NOT NULL
+        "#,
+    )
+    .bind(repository_id)
+    .bind(field)
+    .bind(value)
+    .bind(name)
+    .bind(version)
+    .fetch_one(executor)
+    .await?;
+    Ok((present.unwrap_or(false), count))
+}
+
 /// Give an imported component the times its source recorded: `published_at`
 /// becomes its creation time, and `downloaded_at` its (and its package
 /// asset's, at `package_path`) last download unless Forge saw a later one.

crates/db/src/repository_dao.rs

@@ -650,6 +650,24 @@ where
     Ok(row.is_some())
 }
 
+/// Lock a repository row `FOR NO KEY UPDATE` inside the caller's
+/// transaction: publishes that add a new value of a capped summary field
+/// (`component_dao::summary_values`) take it before they count the values,
+/// so that two of them never both pass the cap. It does not conflict with
+/// the `FOR KEY SHARE` lock of every `component` insert, so other publishes
+/// do not wait for it. Returns `false` when the row does not exist.
+pub async fn lock_repository_values<'e, E>(executor: E, id: i64) -> sqlx::Result<bool>
+where
+    E: sqlx::PgExecutor<'e>,
+{
+    let row: Option<i64> =
+        sqlx::query_scalar("SELECT id FROM repository WHERE id = $1 FOR NO KEY UPDATE")
+            .bind(id)
+            .fetch_optional(executor)
+            .await?;
+    Ok(row.is_some())
+}
+
 /// Delete a repository by id. Returns whether a row was deleted.
 ///
 /// Every dependent row cascades via `ON DELETE CASCADE` — `group_member` rows