WIP: watcher: add sqs sdk, add first draft for event payload structs

Cargo.toml

@@ -30,6 +30,11 @@ edition = "2024"
 anyhow = { version = "1.0.42", features = ["backtrace"] }
 askama = "0.16.0"
 async-stream = "0.3.5"
+# The default `rustls` feature pulls in the legacy hyper 0.14 + rustls 0.21
+# stack via `aws-smithy-runtime/tls-rustls`, which includes the vulnerable
+# `rustls-webpki` v0.101.x. Using only `default-https-client` avoids this by
+# using the modern rustls 0.23 + hyper 1.x stack instead.
+aws-config = { version = "1.0.0", default-features = false, features = ["default-https-client", "rt-tokio"] }
 axum-extra = { version = "0.12.0", features = ["middleware", "routing", "typed-header"] }
 base64 = "0.22"
 bon = { version = "3.8.1", features = ["experimental-overwritable"] }

crates/bin/docs_rs_watcher/Cargo.toml

@@ -8,13 +8,16 @@ edition.workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
+aws-config = { workspace = true }
+aws-sdk-sqs = { version = "1.99.0", default-features = false, features = ["default-https-client", "rt-tokio"] }
 clap = { workspace = true }
 # NOTE: on the new infra, switch back from `git-https-reqwest` to `git-https` (curl) once the curl version is new enough
 crates-index = { version = "3.0.0", default-features = false, features = ["git", "git-https-reqwest", "git-performance", "parallel"] }
 # NOTE: on the new infra, switch back from `http-reqwest` to `http-curl` once the curl version is new enough
 crates-index-diff = { version = "30.0.0", default-features = false, features = ["http-reqwest", "max-performance", "semver"] }
 docs_rs_build_queue = { path = "../../lib/docs_rs_build_queue" }
 docs_rs_config = { path = "../../lib/docs_rs_config" }
+docs_rs_crates_io = { path = "../../lib/docs_rs_crates_io" }
 docs_rs_context = { path = "../../lib/docs_rs_context" }
 docs_rs_database = { path = "../../lib/docs_rs_database" }
 docs_rs_env_vars = { path = "../../lib/docs_rs_env_vars" }
@@ -27,11 +30,13 @@ docs_rs_types = { path = "../../lib/docs_rs_types" }
 docs_rs_utils = { path = "../../lib/docs_rs_utils" }
 futures-util = { workspace = true }
 itertools = { workspace = true }
+serde_json = { workspace = true }
 opentelemetry = { workspace = true }
 rayon = "1.6.1"
 sqlx = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+url = { workspace = true }
 
 [dev-dependencies]
 docs_rs_config = { path = "../../lib/docs_rs_config", features = ["testing"] }

crates/bin/docs_rs_watcher/src/config.rs

@@ -2,11 +2,15 @@ use anyhow::Result;
 use docs_rs_config::AppConfig;
 use docs_rs_env_vars::{env, maybe_env, require_env};
 use std::{path::PathBuf, time::Duration};
+use url::Url;
 
 #[derive(Debug)]
 pub struct Config {
     pub registry_index_path: PathBuf,
     pub registry_url: Option<String>,
+    pub sqs_queue_url: Option<Url>,
+    pub sqs_region: Option<String>,
+    pub aws_sdk_max_retries: u32,
 
     /// How long to wait between registry checks
     pub delay_between_registry_fetches: Duration,
@@ -29,6 +33,11 @@ impl AppConfig for Config {
         Ok(Self {
             registry_index_path: env("REGISTRY_INDEX_PATH", prefix.join("crates.io-index"))?,
             registry_url: maybe_env("REGISTRY_URL")?,
+
+            sqs_queue_url: maybe_env("DOCSRS_SQS_QUEUE_URL")?,
+            sqs_region: maybe_env("DOCSRS_SQS_REGION")?,
+            aws_sdk_max_retries: env("DOCSRS_AWS_SDK_MAX_RETRIES", 6u32)?,
+
             delay_between_registry_fetches: Duration::from_secs(env::<u64>(
                 "DOCSRS_DELAY_BETWEEN_REGISTRY_FETCHES",
                 60,

crates/bin/docs_rs_watcher/src/db/delete.rs

@@ -67,7 +67,13 @@ pub async fn delete_version(
         return Ok(());
     };
 
-    let is_library = delete_version_from_database(conn, config, name, crate_id, version).await?;
+    let Some(is_library) =
+        delete_version_from_database(conn, config, name, crate_id, version).await?
+    else {
+        // release doesn't exist
+        return Ok(());
+    };
+
     let paths = if is_library {
         LIBRARY_STORAGE_PATHS_TO_DELETE
     } else {
@@ -133,7 +139,18 @@ async fn delete_version_from_database(
     name: &KrateName,
     crate_id: CrateId,
     version: &Version,
-) -> Result<bool> {
+) -> Result<Option<bool>> {
+    let Some(release_id) = sqlx::query_scalar!(
+        "SELECT id FROM releases WHERE crate_id = $1 AND version = $2",
+        crate_id as _,
+        version as _
+    )
+    .fetch_optional(&mut *conn)
+    .await?
+    else {
+        return Ok(None);
+    };
+
     let mut transaction = conn.begin().await?;
 
     let delete_lock_timeout = format!("{}ms", config.delete_lock_timeout.as_millis());
@@ -157,23 +174,23 @@ async fn delete_version_from_database(
     sqlx::query!(
         "DELETE FROM builds_logs bl
          USING builds b
-         JOIN releases r ON b.rid = r.id
-         WHERE bl.build_id = b.id AND r.crate_id = $1 AND r.version = $2;",
-        crate_id as _,
-        version as _
+         WHERE bl.build_id = b.id AND b.rid = $1;",
+        release_id as _,
     )
     .execute(&mut *transaction)
     .await?;
 
     for &(table, column) in METADATA {
-        sqlx::query(sqlx::AssertSqlSafe(
-            format!("DELETE FROM {table} WHERE {column} IN (SELECT id FROM releases WHERE crate_id = $1 AND version = $2)")))
-        .bind(crate_id).bind(version).execute(&mut *transaction).await?;
+        sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DELETE FROM {table} WHERE {column} = $1"
+        )))
+        .bind(release_id)
+        .execute(&mut *transaction)
+        .await?;
     }
     let is_library: bool = sqlx::query_scalar!(
-        "DELETE FROM releases WHERE crate_id = $1 AND version = $2 RETURNING is_library",
-        crate_id.0,
-        version as _,
+        "DELETE FROM releases WHERE id = $1 RETURNING is_library",
+        release_id as _,
     )
     .fetch_one(&mut *transaction)
     .await?
@@ -190,7 +207,7 @@ async fn delete_version_from_database(
     update_latest_version_id(&mut transaction, crate_id).await?;
 
     transaction.commit().await?;
-    Ok(is_library)
+    Ok(Some(is_library))
 }
 
 /// Returns whether any release in this crate was a library
@@ -449,6 +466,13 @@ mod tests {
             );
         }
 
+        // running delete-crate again doesn't error.
+        assert!(
+            delete_crate(&mut conn, storage, env.config(), &FOO)
+                .await
+                .is_ok()
+        );
+
         Ok(())
     }
 
@@ -613,6 +637,13 @@ mod tests {
             vec!["Peter Rabbit".to_string()]
         );
 
+        // running delete-version again doesn't fail.
+        assert!(
+            delete_version(&mut conn, storage, env.config(), &KRATE, &V1)
+                .await
+                .is_ok()
+        );
+
         // FIXME: remove for now until test frontend is async
         // let web = env.frontend();
         // assert_success("/a/2.0.0/a/", web)?;
@@ -691,6 +722,32 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_already_deleted_version_doesnt_error() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        env.fake_release()
+            .await
+            .name(&KRATE)
+            .version(V1)
+            .create()
+            .await?;
+        env.fake_release()
+            .await
+            .name(&KRATE)
+            .version(V2)
+            .create()
+            .await?;
+
+        delete_version(&mut conn, env.storage()?, env.config(), &KRATE, &V1).await?;
+        delete_version(&mut conn, env.storage()?, env.config(), &KRATE, &V1).await?;
+
+        assert!(crate_exists(&mut conn, &KRATE).await?);
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_delete_version_waits_for_locked_queue_rows() -> Result<()> {
         let env = TestEnvironment::new().await?;