Fix build metadata race condition

crates/crates_io_database/src/schema.rs

@@ -989,6 +989,8 @@ diesel::table! {
         bin_names -> Nullable<Array<Nullable<Text>>>,
         /// message associated with a yanked version
         yank_message -> Nullable<Text>,
+        /// This is the same as `num` without the optional "build metadata" part (except for some versions that were published before we started validating this).
+        num_no_build -> Varchar,
     }
 }
 

crates/crates_io_database_dump/src/dump-db.toml

@@ -220,6 +220,7 @@ dependencies = ["crates", "users"]
 id = "public"
 crate_id = "public"
 num = "public"
+num_no_build = "public"
 updated_at = "public"
 created_at = "public"
 downloads = "public"

crates/crates_io_database_dump/src/snapshots/crates_io_database_dump__tests__sql_scripts@export.sql.snap

@@ -17,7 +17,7 @@ BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY;
     \copy "crates_keywords" ("crate_id", "keyword_id") TO 'data/crates_keywords.csv' WITH CSV HEADER
     \copy (SELECT "crate_id", "created_at", "created_by", "owner_id", "owner_kind" FROM "crate_owners" WHERE NOT deleted) TO 'data/crate_owners.csv' WITH CSV HEADER
 
-    \copy "versions" ("bin_names", "checksum", "crate_id", "crate_size", "created_at", "downloads", "features", "has_lib", "id", "license", "links", "num", "published_by", "rust_version", "updated_at", "yanked") TO 'data/versions.csv' WITH CSV HEADER
+    \copy "versions" ("bin_names", "checksum", "crate_id", "crate_size", "created_at", "downloads", "features", "has_lib", "id", "license", "links", "num", "num_no_build", "published_by", "rust_version", "updated_at", "yanked") TO 'data/versions.csv' WITH CSV HEADER
     \copy "default_versions" ("crate_id", "version_id") TO 'data/default_versions.csv' WITH CSV HEADER
     \copy "dependencies" ("crate_id", "default_features", "explicit_name", "features", "id", "kind", "optional", "req", "target", "version_id") TO 'data/dependencies.csv' WITH CSV HEADER
     \copy "version_downloads" ("date", "downloads", "version_id") TO 'data/version_downloads.csv' WITH CSV HEADER

crates/crates_io_database_dump/src/snapshots/crates_io_database_dump__tests__sql_scripts@import.sql.snap

@@ -59,7 +59,7 @@ BEGIN;
     \copy "crates_categories" ("category_id", "crate_id") FROM 'data/crates_categories.csv' WITH CSV HEADER
     \copy "crates_keywords" ("crate_id", "keyword_id") FROM 'data/crates_keywords.csv' WITH CSV HEADER
     \copy "crate_owners" ("crate_id", "created_at", "created_by", "owner_id", "owner_kind") FROM 'data/crate_owners.csv' WITH CSV HEADER
-    \copy "versions" ("bin_names", "checksum", "crate_id", "crate_size", "created_at", "downloads", "features", "has_lib", "id", "license", "links", "num", "published_by", "rust_version", "updated_at", "yanked") FROM 'data/versions.csv' WITH CSV HEADER
+    \copy "versions" ("bin_names", "checksum", "crate_id", "crate_size", "created_at", "downloads", "features", "has_lib", "id", "license", "links", "num", "num_no_build", "published_by", "rust_version", "updated_at", "yanked") FROM 'data/versions.csv' WITH CSV HEADER
     \copy "default_versions" ("crate_id", "version_id") FROM 'data/default_versions.csv' WITH CSV HEADER
     \copy "dependencies" ("crate_id", "default_features", "explicit_name", "features", "id", "kind", "optional", "req", "target", "version_id") FROM 'data/dependencies.csv' WITH CSV HEADER
     \copy "version_downloads" ("date", "downloads", "version_id") FROM 'data/version_downloads.csv' WITH CSV HEADER

migrations/2024-10-24-133507_add-num-no-build-column/down.sql

@@ -0,0 +1,2 @@
+alter table versions
+    drop num_no_build;

migrations/2024-10-24-133507_add-num-no-build-column/up.sql

@@ -0,0 +1,29 @@
+alter table versions
+    add num_no_build varchar;
+
+comment on column versions.num_no_build is 'This is the same as `num` without the optional "build metadata" part (except for some versions that were published before we started validating this).';
+
+-- to be run manually:
+
+-- update versions
+--     set num_no_build = split_part(num, '+', 1);
+--
+-- with duplicates as (
+--     -- find all versions that have the same `crate_id` and `num_no_build`
+--     select crate_id, num_no_build, array_agg(num ORDER BY id) as nums
+--     from versions
+--     group by crate_id, num_no_build
+--     having count(*) > 1
+-- ),
+-- duplicates_to_update as (
+--     -- for each group of duplicates, update all versions except the one that
+--     -- doesn't have "build metadata", or the first one that was published if
+--     -- all versions have "build metadata"
+--     select crate_id, num_no_build, unnest(case when array_position(nums, num_no_build) IS NOT NULL then array_remove(nums, num_no_build) else nums[2:] end) as num
+--     from duplicates
+-- )
+-- update versions
+--     set num_no_build = duplicates_to_update.num
+--     from duplicates_to_update
+--     where versions.crate_id = duplicates_to_update.crate_id
+--     and versions.num = duplicates_to_update.num;

migrations/2024-10-24-134209_make-unique-num-not-null/down.sql

@@ -0,0 +1,2 @@
+alter table versions
+    alter column num_no_build drop not null;

migrations/2024-10-24-134209_make-unique-num-not-null/up.sql

@@ -0,0 +1,2 @@
+alter table versions
+    alter column num_no_build set not null;

migrations/2024-10-25-112826_make-unique-version-unique/down.sql

@@ -0,0 +1,2 @@
+alter table versions
+    drop constraint versions_crate_id_num_no_build_uindex;

migrations/2024-10-25-112826_make-unique-version-unique/metadata.toml

@@ -0,0 +1 @@
+run_in_transaction = false

migrations/2024-10-25-112826_make-unique-version-unique/up.sql

@@ -0,0 +1,2 @@
+create unique index concurrently if not exists versions_crate_id_num_no_build_uindex
+    on versions (crate_id, num_no_build);

src/controllers/krate/publish.rs

@@ -359,7 +359,7 @@
             let hex_cksum: String = Sha256::digest(&tarball_bytes).encode_hex();
 
             // Persist the new version of this crate
-            let version = NewVersion::builder(krate.id, &version_string)
+            let new_version = NewVersion::builder(krate.id, &version_string)
                 .features(serde_json::to_value(&features)?)
                 .maybe_license(license.as_deref())
                 // Downcast is okay because the file length must be less than the max upload size
@@ -371,8 +371,18 @@
                 .maybe_rust_version(rust_version.as_deref())
                 .has_lib(tarball_info.manifest.lib.is_some())
                 .bin_names(bin_names.as_slice())
-                .build()
-                .save(conn, &verified_email_address)?;
+                .build();
+
+            let version = match new_version.save(conn, &verified_email_address) {
+                Err(diesel::result::Error::DatabaseError(diesel::result::DatabaseErrorKind::UniqueViolation, _)) => {
+                    return Err(bad_request(format_args!(
+                        "crate version `{}` is already uploaded",
+                        new_version.num_no_build
+                    )));
+                },
+                Err(error) => return Err(error.into()),
+                Ok(version) => version,
+            };
 
             insert_version_owner_action(
                 conn,

src/models/default_versions.rs

@@ -245,6 +245,7 @@ mod tests {
             .values((
                 versions::crate_id.eq(crate_id),
                 versions::num.eq(num),
+                versions::num_no_build.eq(num),
                 versions::checksum.eq(""),
             ))
             .execute(conn)

src/models/version.rs

@@ -6,11 +6,8 @@ use crates_io_index::features::FeaturesMap;
 use diesel::prelude::*;
 use serde::Deserialize;
 
-use crate::util::errors::{bad_request, AppResult};
-
 use crate::models::{Crate, Dependency, User};
 use crate::schema::*;
-use crate::sql::split_part;
 use crate::util::diesel::Conn;
 
 // Queryable has a custom implementation below
@@ -34,6 +31,7 @@ pub struct Version {
     pub has_lib: Option<bool>,
     pub bin_names: Option<Vec<Option<String>>>,
     pub yank_message: Option<String>,
+    pub num_no_build: String,
 }
 
 impl Version {
@@ -84,6 +82,8 @@ pub struct NewVersion<'a> {
     crate_id: i32,
     #[builder(start_fn)]
     num: &'a str,
+    #[builder(default = strip_build_metadata(num))]
+    pub num_no_build: &'a str,
     created_at: Option<&'a NaiveDateTime>,
     yanked: Option<bool>,
     #[builder(default = serde_json::Value::Object(Default::default()))]
@@ -100,24 +100,10 @@ pub struct NewVersion<'a> {
 }
 
 impl NewVersion<'_> {
-    pub fn save(&self, conn: &mut impl Conn, published_by_email: &str) -> AppResult<Version> {
-        use diesel::dsl::exists;
-        use diesel::{insert_into, select};
+    pub fn save(&self, conn: &mut impl Conn, published_by_email: &str) -> QueryResult<Version> {
+        use diesel::insert_into;
 
         conn.transaction(|conn| {
-            let num_no_build = strip_build_metadata(self.num);
-
-            let already_uploaded = versions::table
-                .filter(versions::crate_id.eq(self.crate_id))
-                .filter(split_part(versions::num, "+", 1).eq(num_no_build));
-
-            if select(exists(already_uploaded)).get_result(conn)? {
-                return Err(bad_request(format_args!(
-                    "crate version `{}` is already uploaded",
-                    num_no_build
-                )));
-            }
-
             let version: Version = insert_into(versions::table).values(self).get_result(conn)?;
 
             insert_into(versions_published_by::table)

src/tests/worker/rss/sync_crate_feed.rs

@@ -69,6 +69,7 @@ async fn create_version(
         .values((
             versions::crate_id.eq(crate_id),
             versions::num.eq(version),
+            versions::num_no_build.eq(version),
             versions::created_at.eq(publish_time),
             versions::updated_at.eq(publish_time),
             versions::checksum.eq("checksum"),

src/tests/worker/rss/sync_updates_feed.rs

@@ -75,6 +75,7 @@ async fn create_version(
         .values((
             versions::crate_id.eq(crate_id),
             versions::num.eq(version),
+            versions::num_no_build.eq(version),
             versions::created_at.eq(publish_time),
             versions::updated_at.eq(publish_time),
             versions::checksum.eq("checksum"),

src/worker/jobs/archive_version_downloads.rs

@@ -397,6 +397,7 @@ mod tests {
             .values((
                 versions::crate_id.eq(crate_id),
                 versions::num.eq(num),
+                versions::num_no_build.eq(num),
                 versions::checksum.eq(""),
             ))
             .returning(versions::id)

src/worker/jobs/downloads/process_log.rs

@@ -511,6 +511,7 @@ mod tests {
             .values((
                 versions::crate_id.eq(crate_id),
                 versions::num.eq(version),
+                versions::num_no_build.eq(version),
                 versions::checksum.eq("checksum"),
             ))
             .execute(conn)

src/worker/jobs/rss/sync_crate_feed.rs

@@ -254,10 +254,12 @@ mod tests {
         version: impl Into<Cow<'static, str>>,
         publish_time: NaiveDateTime,
     ) -> impl Future<Output = i32> {
+        let version = version.into();
         let future = diesel::insert_into(versions::table)
             .values((
                 versions::crate_id.eq(crate_id),
-                versions::num.eq(version.into()),
+                versions::num.eq(version.clone()),
+                versions::num_no_build.eq(version),
                 versions::created_at.eq(publish_time),
                 versions::updated_at.eq(publish_time),
                 versions::checksum.eq("checksum"),

src/worker/jobs/rss/sync_updates_feed.rs

@@ -248,10 +248,12 @@ mod tests {
         version: impl Into<Cow<'static, str>>,
         publish_time: NaiveDateTime,
     ) -> impl Future<Output = i32> {
+        let version = version.into();
         let future = diesel::insert_into(versions::table)
             .values((
                 versions::crate_id.eq(crate_id),
-                versions::num.eq(version.into()),
+                versions::num.eq(version.clone()),
+                versions::num_no_build.eq(version),
                 versions::created_at.eq(publish_time),
                 versions::updated_at.eq(publish_time),
                 versions::checksum.eq("checksum"),