diff --git a/.docker.env.sample b/.docker.env.sample
index 743cc06133..995e1254af 100644
--- a/.docker.env.sample
+++ b/.docker.env.sample
@@ -10,3 +10,8 @@ DOCSRS_TOOLCHAIN=nightly
 # for the registry watcher, automatically queued reqbuidls.
 DOCSRS_MAX_QUEUED_REBUILDS: 10
 
+# optional overrides for local ElasticMQ testing
+# DOCSRS_SQS_QUEUE_URL=http://elasticmq:9324/queue/docsrs-events
+# DOCSRS_SQS_QUEUE_REGION=elasticmq
+# DOCSRS_SQS_ENDPOINT_URL=http://elasticmq:9324
+# DOCSRS_SQS_ACTIVE=false
diff --git a/.env.sample b/.env.sample
index b6974b702c..b67400de0c 100644
--- a/.env.sample
+++ b/.env.sample
@@ -20,5 +20,11 @@ SENTRY_ENVIRONMENT=dev
 # https://forge.rust-lang.org/infra/docs/rustc-ci.html#try-builds
 DOCSRS_TOOLCHAIN=nightly
 
-# NOTE: when running services in docker-compose, you can override the settings in 
+# NOTE: when running services in docker-compose, you can override the settings in
 # `.docker.env`, you'll fine an example in `.docker.env.sample`.
+
+# optional overrides for local ElasticMQ testing
+DOCSRS_SQS_QUEUE_URL=http://localhost:9324/queue/docsrs-events
+DOCSRS_SQS_QUEUE_REGION=localhost
+DOCSRS_SQS_ENDPOINT_URL=http://localhost:9324
+DOCSRS_SQS_ACTIVE=false
diff --git a/.sqlx/query-2dc065cc08f262c937c54f9cc8629e35750da2bea995fb0c433893addb253214.json b/.sqlx/query-2dc065cc08f262c937c54f9cc8629e35750da2bea995fb0c433893addb253214.json
deleted file mode 100644
index 2109f69cf3..0000000000
--- a/.sqlx/query-2dc065cc08f262c937c54f9cc8629e35750da2bea995fb0c433893addb253214.json
+++ /dev/null
@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM builds_logs bl\n         USING builds b\n         JOIN releases r ON b.rid = r.id\n         WHERE bl.build_id = b.id AND r.crate_id = $1 AND r.version = $2;",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Int4",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "2dc065cc08f262c937c54f9cc8629e35750da2bea995fb0c433893addb253214"
-}
diff --git a/.sqlx/query-5f5fa0e89b4e13c690b1648a18e8420f7da0f0445c6e43d5d64617226c24fba1.json b/.sqlx/query-5f5fa0e89b4e13c690b1648a18e8420f7da0f0445c6e43d5d64617226c24fba1.json
new file mode 100644
index 0000000000..5d451984e9
--- /dev/null
+++ b/.sqlx/query-5f5fa0e89b4e13c690b1648a18e8420f7da0f0445c6e43d5d64617226c24fba1.json
@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM releases",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4",
+        "origin": {
+          "Table": {
+            "table": "releases",
+            "name": "id"
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "5f5fa0e89b4e13c690b1648a18e8420f7da0f0445c6e43d5d64617226c24fba1"
+}
diff --git a/.sqlx/query-66b0ba6978880b79ce7a179bbe986e6c7eed78a2d4f01f316772949f5d688f42.json b/.sqlx/query-66b0ba6978880b79ce7a179bbe986e6c7eed78a2d4f01f316772949f5d688f42.json
new file mode 100644
index 0000000000..89cbc239cc
--- /dev/null
+++ b/.sqlx/query-66b0ba6978880b79ce7a179bbe986e6c7eed78a2d4f01f316772949f5d688f42.json
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM builds_logs bl\n         USING builds b\n         WHERE bl.build_id = b.id AND b.rid = $1;",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int4"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "66b0ba6978880b79ce7a179bbe986e6c7eed78a2d4f01f316772949f5d688f42"
+}
diff --git a/.sqlx/query-7b7dd5795cddcb66b140b57157983bd73f73ecc1cf9b4fc24c457d5f26fd582b.json b/.sqlx/query-7b7dd5795cddcb66b140b57157983bd73f73ecc1cf9b4fc24c457d5f26fd582b.json
new file mode 100644
index 0000000000..effb9f9ac3
--- /dev/null
+++ b/.sqlx/query-7b7dd5795cddcb66b140b57157983bd73f73ecc1cf9b4fc24c457d5f26fd582b.json
@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM releases WHERE crate_id = $1 AND version = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4",
+        "origin": {
+          "Table": {
+            "table": "releases",
+            "name": "id"
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "7b7dd5795cddcb66b140b57157983bd73f73ecc1cf9b4fc24c457d5f26fd582b"
+}
diff --git a/.sqlx/query-014a054d852f0937191e1a54f742d4b4c454361689fb3841cc12fd7dd1094948.json b/.sqlx/query-fab139cabc0987a1f2ad706060a3f4254924db75fc7f76a6c78d17a3fc06d663.json
similarity index 65%
rename from .sqlx/query-014a054d852f0937191e1a54f742d4b4c454361689fb3841cc12fd7dd1094948.json
rename to .sqlx/query-fab139cabc0987a1f2ad706060a3f4254924db75fc7f76a6c78d17a3fc06d663.json
index 380bd9ea67..1f16abd551 100644
--- a/.sqlx/query-014a054d852f0937191e1a54f742d4b4c454361689fb3841cc12fd7dd1094948.json
+++ b/.sqlx/query-fab139cabc0987a1f2ad706060a3f4254924db75fc7f76a6c78d17a3fc06d663.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "DELETE FROM releases WHERE crate_id = $1 AND version = $2 RETURNING is_library",
+  "query": "DELETE FROM releases WHERE id = $1 RETURNING is_library",
   "describe": {
     "columns": [
       {
@@ -17,13 +17,12 @@
     ],
     "parameters": {
       "Left": [
-        "Int4",
-        "Text"
+        "Int4"
       ]
     },
     "nullable": [
       true
     ]
   },
-  "hash": "014a054d852f0937191e1a54f742d4b4c454361689fb3841cc12fd7dd1094948"
+  "hash": "fab139cabc0987a1f2ad706060a3f4254924db75fc7f76a6c78d17a3fc06d663"
 }
diff --git a/Cargo.lock b/Cargo.lock
index 48b171579a..372915df99 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -485,11 +485,11 @@ dependencies = [
  "aws-runtime",
  "aws-sdk-sts",
  "aws-smithy-async",
- "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-http 0.64.0",
+ "aws-smithy-json 0.63.0",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "aws-types",
  "bytes",
@@ -546,7 +546,7 @@ dependencies = [
  "aws-sigv4",
  "aws-smithy-async",
  "aws-smithy-eventstream",
- "aws-smithy-http",
+ "aws-smithy-http 0.64.0",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -577,12 +577,12 @@ dependencies = [
  "aws-smithy-async",
  "aws-smithy-checksums",
  "aws-smithy-eventstream",
- "aws-smithy-http",
- "aws-smithy-json",
- "aws-smithy-observability",
+ "aws-smithy-http 0.64.0",
+ "aws-smithy-json 0.63.0",
+ "aws-smithy-observability 0.3.0",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "aws-smithy-xml",
  "aws-types",
@@ -601,6 +601,31 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "aws-sdk-sqs"
+version = "1.102.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0246bf049cfc003ce44599dff955b9353758de3afa68a053da9b2c7de20a07d8"
+dependencies = [
+ "arc-swap",
+ "aws-credential-types",
+ "aws-runtime",
+ "aws-smithy-async",
+ "aws-smithy-http 0.63.6",
+ "aws-smithy-json 0.62.7",
+ "aws-smithy-observability 0.2.6",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "fastrand",
+ "http 0.2.12",
+ "http 1.5.0",
+ "regex-lite",
+ "tracing",
+]
+
 [[package]]
 name = "aws-sdk-sts"
 version = "1.112.0"
@@ -611,13 +636,13 @@ dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async",
- "aws-smithy-http",
- "aws-smithy-json",
- "aws-smithy-observability",
+ "aws-smithy-http 0.64.0",
+ "aws-smithy-json 0.63.0",
+ "aws-smithy-observability 0.3.0",
  "aws-smithy-query",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "aws-smithy-xml",
  "aws-types",
@@ -636,7 +661,7 @@ checksum = "723c2234ad7511ceef63eab016b7ba6ff7c55590fefb96fa8467af014a07309f"
 dependencies = [
  "aws-credential-types",
  "aws-smithy-eventstream",
- "aws-smithy-http",
+ "aws-smithy-http 0.64.0",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "bytes",
@@ -668,7 +693,7 @@ version = "0.65.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b67ecd999972b58e67cab052f5129906c08c25883bd0788ceefc55ef97d61307"
 dependencies = [
- "aws-smithy-http",
+ "aws-smithy-http 0.64.0",
  "aws-smithy-types",
  "bytes",
  "crc-fast",
@@ -694,6 +719,27 @@ dependencies = [
  "crc32fast",
 ]
 
+[[package]]
+name = "aws-smithy-http"
+version = "0.63.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba1ab2dc1c2c3749ead27180d333c42f11be8b0e934058fb4b2258ee8dbe5231"
+dependencies = [
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "bytes",
+ "bytes-utils",
+ "futures-core",
+ "futures-util",
+ "http 1.5.0",
+ "http-body 1.1.0",
+ "http-body-util",
+ "percent-encoding",
+ "pin-project-lite",
+ "pin-utils",
+ "tracing",
+]
+
 [[package]]
 name = "aws-smithy-http"
 version = "0.64.0"
@@ -740,6 +786,17 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "aws-smithy-json"
+version = "0.62.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "701a947f4797e52a911e114a898667c746c39feea467bbd1abd7b3721f702ffa"
+dependencies = [
+ "aws-smithy-runtime-api",
+ "aws-smithy-schema 0.1.0",
+ "aws-smithy-types",
+]
+
 [[package]]
 name = "aws-smithy-json"
 version = "0.63.0"
@@ -747,10 +804,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3dc65a121adb4b33729919fcfa14fa36fb33c1555a8f06bb0e2188dbfdc1d9ef"
 dependencies = [
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
 ]
 
+[[package]]
+name = "aws-smithy-observability"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a06c2315d173edbf1920da8ba3a7189695827002e4c0fc961973ab1c54abca9c"
+dependencies = [
+ "aws-smithy-runtime-api",
+]
+
 [[package]]
 name = "aws-smithy-observability"
 version = "0.3.0"
@@ -767,7 +833,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "512346c7212ab7436df2d77a16d976a468ae44a418835511d2a69269810aaf62"
 dependencies = [
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "aws-smithy-xml",
  "urlencoding",
@@ -780,11 +846,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b82e438d30e02a825d363bd639a9efaed68a8089d86101054b0081e7e0d3e606"
 dependencies = [
  "aws-smithy-async",
- "aws-smithy-http",
+ "aws-smithy-http 0.64.0",
  "aws-smithy-http-client",
- "aws-smithy-observability",
+ "aws-smithy-observability 0.3.0",
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "bytes",
  "fastrand",
@@ -828,6 +894,17 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "aws-smithy-schema"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7442cb268338f0eb8278140a107c046756aa01093d8ef5e99628d34ae09c94f5"
+dependencies = [
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "http 1.5.0",
+]
+
 [[package]]
 name = "aws-smithy-schema"
 version = "0.2.0"
@@ -882,7 +959,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ce84f71c72fee2cbbadde6e7d082f5fb466e3a84733855295fa7aafd1b31b7d8"
 dependencies = [
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "xmlparser",
 ]
@@ -896,7 +973,7 @@ dependencies = [
  "aws-credential-types",
  "aws-smithy-async",
  "aws-smithy-runtime-api",
- "aws-smithy-schema",
+ "aws-smithy-schema 0.2.0",
  "aws-smithy-types",
  "rustc_version",
  "tracing",
@@ -1259,10 +1336,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
 dependencies = [
  "iana-time-zone",
- "js-sys",
  "num-traits",
  "serde",
- "wasm-bindgen",
  "windows-link",
 ]
 
@@ -2144,6 +2219,7 @@ dependencies = [
  "chrono",
  "serde",
  "serde_json",
+ "test-case",
 ]
 
 [[package]]
@@ -2488,12 +2564,17 @@ name = "docs_rs_watcher"
 version = "0.6.0"
 dependencies = [
  "anyhow",
+ "async-trait",
+ "aws-config",
+ "aws-sdk-sqs",
+ "chrono",
  "clap",
  "crates-index",
  "crates-index-diff",
  "docs_rs_build_queue",
  "docs_rs_config",
  "docs_rs_context",
+ "docs_rs_crates_io",
  "docs_rs_database",
  "docs_rs_env_vars",
  "docs_rs_fastly",
@@ -2510,9 +2591,11 @@ dependencies = [
  "opentelemetry",
  "pretty_assertions",
  "rayon",
+ "serde_json",
  "sqlx",
  "tokio",
  "tracing",
+ "url",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 1684d19f5f..9e656de1d5 100644
--- a/Cargo.toml
+++ b/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.23"
 bon = { version = "3.8.1", features = ["experimental-overwritable"] }
diff --git a/README.md b/README.md
index 1adf5b1eb9..d8a9d02f44 100644
--- a/README.md
+++ b/README.md
@@ -166,8 +166,8 @@ $ just lint
 ```
 
 Linting GitHub Actions workflows requires
-[`actionlint`](https://github.com/rhysd/actionlint/blob/main/docs/install.md). If
-it is not installed, that check is skipped with a warning.
+[`actionlint`](https://github.com/rhysd/actionlint/blob/main/docs/install.md).
+If it is not installed, that check is skipped with a warning.
 
 Run all formatters with:
 
diff --git a/crates/bin/cratesfyi/src/daemon.rs b/crates/bin/cratesfyi/src/daemon.rs
index 2ea37bd6b4..f8b7f58082 100644
--- a/crates/bin/cratesfyi/src/daemon.rs
+++ b/crates/bin/cratesfyi/src/daemon.rs
@@ -4,7 +4,7 @@ use docs_rs_config::AppConfig as _;
 use docs_rs_context::Context;
 use docs_rs_watcher::{
     start_background_queue_rebuild, start_background_repository_stats_updater,
-    start_background_service_metric_collector, watch_registry,
+    start_background_service_metric_collector,
 };
 use docs_rs_web::run_web_server;
 use std::sync::Arc;
@@ -21,7 +21,7 @@ fn start_registry_watcher(
         // space this out to prevent it from clashing against the queue-builder thread on launch
         tokio::time::sleep(Duration::from_secs(30)).await;
 
-        watch_registry(&config, &context).await
+        docs_rs_watcher::watch(&config, &context).await;
     });
 
     Ok(())
diff --git a/crates/bin/docs_rs_watcher/Cargo.toml b/crates/bin/docs_rs_watcher/Cargo.toml
index 30f38a6e8b..31da9b4f28 100644
--- a/crates/bin/docs_rs_watcher/Cargo.toml
+++ b/crates/bin/docs_rs_watcher/Cargo.toml
@@ -8,6 +8,10 @@ edition.workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
+async-trait = "0.1.89"
+aws-config = { workspace = true }
+aws-sdk-sqs = { version = "1.99.0", default-features = false, features = ["default-https-client", "rt-tokio"] }
+chrono = { workspace = true }
 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"] }
@@ -16,6 +20,7 @@ crates-index-diff = { version = "31.0.0", default-features = false, features = [
 docs_rs_build_queue = { path = "../../lib/docs_rs_build_queue" }
 docs_rs_config = { path = "../../lib/docs_rs_config" }
 docs_rs_context = { path = "../../lib/docs_rs_context" }
+docs_rs_crates_io = { path = "../../lib/docs_rs_crates_io" }
 docs_rs_database = { path = "../../lib/docs_rs_database" }
 docs_rs_env_vars = { path = "../../lib/docs_rs_env_vars" }
 docs_rs_fastly = { path = "../../lib/docs_rs_fastly" }
@@ -29,9 +34,11 @@ futures-util = { workspace = true }
 itertools = { workspace = true }
 opentelemetry = { workspace = true }
 rayon = "1.6.1"
+serde_json = { workspace = true }
 sqlx = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+url = { workspace = true }
 
 [dev-dependencies]
 docs_rs_config = { path = "../../lib/docs_rs_config", features = ["testing"] }
diff --git a/crates/bin/docs_rs_watcher/src/config.rs b/crates/bin/docs_rs_watcher/src/config.rs
index 7b5f179760..5f7f105fcd 100644
--- a/crates/bin/docs_rs_watcher/src/config.rs
+++ b/crates/bin/docs_rs_watcher/src/config.rs
@@ -2,15 +2,54 @@ 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;
+
+const SQS_QUEUE_URL: &str = "DOCSRS_SQS_QUEUE_URL";
+const SQS_QUEUE_REGION: &str = "DOCSRS_SQS_QUEUE_REGION";
+
+#[derive(Debug)]
+pub struct SqsConfig {
+    pub queue_url: Url,
+    pub region: String,
+    pub endpoint_url: Option<Url>,
+    pub max_retries: u32,
+    /// temporary, to switch between the sources for the index (git index vs SQS).
+    /// true = only use SQS, don't even fetch git
+    /// false = fetch both sqs & git, use git, just log sqs.
+    pub active: bool,
+}
+
+impl SqsConfig {
+    pub(crate) fn if_configured() -> Result<Option<Self>> {
+        if maybe_env::<String>(SQS_QUEUE_URL)?.is_some()
+            && maybe_env::<String>(SQS_QUEUE_REGION)?.is_some()
+        {
+            SqsConfig::from_environment().map(Some)
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+impl AppConfig for SqsConfig {
+    fn from_environment() -> Result<Self> {
+        Ok(Self {
+            queue_url: require_env(SQS_QUEUE_URL)?,
+            region: require_env(SQS_QUEUE_REGION)?,
+            endpoint_url: maybe_env("DOCSRS_SQS_ENDPOINT_URL")?,
+            active: env("DOCSRS_SQS_ACTIVE", false)?,
+            max_retries: env("DOCSRS_SQS_MAX_RETRIES", 6u32)?,
+        })
+    }
+}
 
 #[derive(Debug)]
 pub struct Config {
+    /// registry watching config. Also used for database-synchonize
     pub registry_index_path: PathBuf,
     pub registry_url: Option<String>,
-
     /// How long to wait between registry checks
     pub delay_between_registry_fetches: Duration,
-
     // Time between 'git gc --auto' calls in seconds
     pub registry_gc_interval: u64,
 
@@ -20,15 +59,29 @@ pub struct Config {
     /// Maximum time to wait for queue row locks when deleting crates/releases.
     pub delete_lock_timeout: Duration,
 
+    pub crates_io_events: Option<SqsConfig>,
+
     pub repository: docs_rs_repository_stats::Config,
 }
 
+impl Config {
+    pub fn crates_io_events_active(&self) -> bool {
+        self.crates_io_events
+            .as_ref()
+            .map(|config| config.active)
+            .unwrap_or(false)
+    }
+}
+
 impl AppConfig for Config {
     fn from_environment() -> Result<Self> {
         let prefix: PathBuf = require_env("DOCSRS_PREFIX")?;
         Ok(Self {
             registry_index_path: env("REGISTRY_INDEX_PATH", prefix.join("crates.io-index"))?,
             registry_url: maybe_env("REGISTRY_URL")?,
+
+            crates_io_events: SqsConfig::if_configured()?,
+
             delay_between_registry_fetches: Duration::from_secs(env::<u64>(
                 "DOCSRS_DELAY_BETWEEN_REGISTRY_FETCHES",
                 60,
@@ -42,4 +95,13 @@ impl AppConfig for Config {
             repository: docs_rs_repository_stats::Config::from_environment()?,
         })
     }
+
+    #[cfg(test)]
+    fn test_config() -> Result<Self> {
+        let mut config = Self::from_environment()?;
+        if let Some(sqs_config) = &mut config.crates_io_events {
+            sqs_config.active = false;
+        }
+        Ok(config)
+    }
 }
diff --git a/crates/bin/docs_rs_watcher/src/db/delete.rs b/crates/bin/docs_rs_watcher/src/db/delete.rs
index 7a5a3fb448..b65d67bf83 100644
--- a/crates/bin/docs_rs_watcher/src/db/delete.rs
+++ b/crates/bin/docs_rs_watcher/src/db/delete.rs
@@ -5,12 +5,14 @@ use docs_rs_storage::{AsyncStorage, rustdoc_archive_path, source_archive_path};
 use docs_rs_types::{CrateId, KrateName, Version};
 use sqlx::Connection;
 use tokio::fs;
+use tracing::instrument;
 
 /// List of directories in docs.rs's underlying storage (either the database or S3) containing a
 /// subdirectory named after the crate. Those subdirectories will be deleted.
 static LIBRARY_STORAGE_PATHS_TO_DELETE: &[&str] = &["rustdoc", "rustdoc-json", "sources"];
 static OTHER_STORAGE_PATHS_TO_DELETE: &[&str] = &["sources"];
 
+#[instrument(skip_all, fields(name=%name))]
 pub async fn delete_crate(
     conn: &mut sqlx::PgConnection,
     storage: &AsyncStorage,
@@ -56,6 +58,7 @@ pub async fn delete_crate(
     Ok(())
 }
 
+#[instrument(skip_all, fields(name=%name, version=%version))]
 pub async fn delete_version(
     conn: &mut sqlx::PgConnection,
     storage: &AsyncStorage,
@@ -67,7 +70,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 +142,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 +177,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 +210,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
@@ -406,6 +426,13 @@ mod tests {
         assert!(!storage.exists(&rustdoc_archive_path(&FOO, &V1)).await?);
         assert!(!storage.exists(&rustdoc_archive_path(&FOO, &V2)).await?);
 
+        // running delete-crate again doesn't error.
+        assert!(
+            delete_crate(&mut conn, storage, env.config(), &FOO)
+                .await
+                .is_ok()
+        );
+
         Ok(())
     }
 
@@ -534,6 +561,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)?;
@@ -612,6 +646,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?;
diff --git a/crates/bin/docs_rs_watcher/src/index_watcher.rs b/crates/bin/docs_rs_watcher/src/index_watcher.rs
index 3d03d3ce94..ef5f2fceee 100644
--- a/crates/bin/docs_rs_watcher/src/index_watcher.rs
+++ b/crates/bin/docs_rs_watcher/src/index_watcher.rs
@@ -2,24 +2,62 @@ use crate::{
     Config,
     db::{delete_crate, delete_version},
     index::Index,
+    metrics::{EventSource, WatcherMetrics},
 };
 use anyhow::{Context as _, Result};
 use crates_index_diff::Change;
 use docs_rs_build_queue::PRIORITY_MANUAL_FROM_CRATES_IO;
 use docs_rs_context::Context;
+use docs_rs_crates_io::events::ChangeKind;
 use docs_rs_database::{
     crate_details::update_latest_version_id,
     service_config::{ConfigName, get_config, set_config},
 };
 use docs_rs_fastly::{Cdn, CdnBehaviour as _};
 use docs_rs_types::{CrateId, KrateName, Version};
-use tracing::{debug, error, info, warn};
+use std::time::Instant;
+use tracing::{debug, error, info, instrument, warn};
+
+trait ChangeExt {
+    fn name(&self) -> &str;
+    fn version(&self) -> Option<&str>;
+    fn kind(&self) -> ChangeKind;
+    fn first_crate_version(&self) -> &crates_index_diff::CrateVersion;
+}
+
+impl ChangeExt for Change {
+    fn first_crate_version(&self) -> &crates_index_diff::CrateVersion {
+        self.versions().first().expect("always exists")
+    }
+
+    fn name(&self) -> &str {
+        self.first_crate_version().name.as_str()
+    }
+
+    fn version(&self) -> Option<&str> {
+        if let Change::CrateDeleted { .. } = self {
+            None
+        } else {
+            Some(self.first_crate_version().version.as_str())
+        }
+    }
+
+    fn kind(&self) -> ChangeKind {
+        match *self {
+            Change::Added(_) => ChangeKind::Added,
+            Change::Yanked(_) => ChangeKind::Yanked,
+            Change::CrateDeleted { .. } => ChangeKind::CrateDeleted,
+            Change::VersionDeleted(_) => ChangeKind::VersionDeleted,
+            Change::Unyanked(_) => ChangeKind::Unyanked,
+            Change::AddedAndYanked(_) => ChangeKind::AddedAndYanked,
+        }
+    }
+}
 
 #[derive(Debug)]
 pub(crate) struct CrateVersion {
     pub name: KrateName,
     pub version: Version,
-    pub yanked: bool,
 }
 
 #[cfg(test)]
@@ -28,7 +66,6 @@ impl Default for CrateVersion {
         Self {
             name: docs_rs_types::testing::KRATE,
             version: docs_rs_types::testing::V1,
-            yanked: false,
         }
     }
 }
@@ -40,7 +77,17 @@ impl TryFrom<crates_index_diff::CrateVersion> for CrateVersion {
         Ok(Self {
             name: value.name.parse()?,
             version: value.version.parse()?,
-            yanked: value.yanked,
+        })
+    }
+}
+
+impl TryFrom<&docs_rs_crates_io::events::CrateVersion> for CrateVersion {
+    type Error = anyhow::Error;
+
+    fn try_from(value: &docs_rs_crates_io::events::CrateVersion) -> Result<Self, Self::Error> {
+        Ok(Self {
+            name: value.name.parse()?,
+            version: value.version.parse()?,
         })
     }
 }
@@ -51,7 +98,6 @@ impl From<CrateVersion> for crates_index_diff::CrateVersion {
         Self {
             name: value.name.to_string().into(),
             version: value.version.to_string().into(),
-            yanked: value.yanked,
             ..Default::default()
         }
     }
@@ -94,6 +140,7 @@ pub(crate) async fn get_new_crates(
     context: &Context,
     index: &Index,
     config: &Config,
+    metrics: &WatcherMetrics,
 ) -> Result<usize> {
     let mut conn = context.pool()?.get_async().await?;
 
@@ -115,7 +162,8 @@ pub(crate) async fn get_new_crates(
 
     debug!(last_seen_reference=%last_seen_reference, new_reference=%new_reference, "queueing changes");
 
-    let crates_added = process_changes(context, &changes, config).await;
+    metrics.record_events_received(EventSource::Git, changes.len());
+    let crates_added = process_changes(context, &changes, config, metrics).await;
 
     if let Err(err) = context.build_queue()?.reevaluate_priorities().await {
         error!(?err, "error reevaluating queued release priorities");
@@ -129,41 +177,89 @@ pub(crate) async fn get_new_crates(
     Ok(crates_added)
 }
 
-async fn process_changes(context: &Context, changes: &Vec<Change>, config: &Config) -> usize {
+async fn process_changes(
+    context: &Context,
+    changes: &Vec<Change>,
+    config: &Config,
+    metrics: &WatcherMetrics,
+) -> usize {
     let mut crates_added = 0;
 
     for change in changes {
-        match process_change(context, change, config).await {
+        let start = Instant::now();
+        let crate_name = change.name();
+        let crate_version = change.version();
+        let change_type = change.kind();
+
+        debug!(
+            target: "docs_rs_watcher::index_event",
+            source = %EventSource::Git,
+            %change_type,
+            crate_name,
+            crate_version,
+            "crates.io index event"
+        );
+
+        if config.crates_io_events_active() {
+            // just to be safe.
+            // Generally we don't even start the git-index-watcher when
+            // SQS is active.
+            // Will be removed with the git index watcher code when SQS is stable.
+            continue;
+        }
+
+        let success = match process_change(context, change, config).await {
             Ok(added) => {
+                metrics.record_change_applied(EventSource::Git, change_type);
                 if added {
                     crates_added += 1;
                 }
+                true
             }
             Err(err) => {
                 error!(?change, ?err, "failed to process change");
+                false
             }
-        }
+        };
+        metrics.record_event_processing_time(
+            EventSource::Git,
+            Some(change_type),
+            success,
+            start.elapsed(),
+        );
     }
     crates_added
 }
 
 /// Process a crate change, returning whether the change was a crate addition or not.
-async fn process_change(context: &Context, change: &Change, config: &Config) -> Result<bool> {
-    let crate_version: CrateVersion = change
-        .versions()
-        .first()
-        .expect("always exists")
-        .clone()
-        .try_into()?;
+#[instrument(skip_all, fields(name, version))]
+pub(crate) async fn process_change(
+    context: &Context,
+    change: &Change,
+    config: &Config,
+) -> Result<bool> {
+    // 1: use the `CrateVersion` from `crates-index-diff`.
+    let crate_version = change.first_crate_version();
+
+    // record name & version on the tracing span for performance instrumentation.
+    tracing::Span::current()
+        .record("name", crate_version.name.as_str())
+        .record("version", crate_version.version.as_str());
+
+    // 2: now, convert to our own internal `CrateVersion.`
+    let crate_version: CrateVersion = crate_version.clone().try_into()?;
 
     match change {
         Change::Added(_release) => process_version_added(context, &crate_version).await?,
         Change::AddedAndYanked(_release) => {
             process_version_added(context, &crate_version).await?;
-            process_version_yank_status(context, &crate_version).await?;
+            process_version_yank_status(context, &crate_version, true).await?;
         }
-        Change::Unyanked(_release) | Change::Yanked(_release) => {
-            process_version_yank_status(context, &crate_version).await?
+        Change::Unyanked(_release) => {
+            process_version_yank_status(context, &crate_version, false).await?
+        }
+        Change::Yanked(_release) => {
+            process_version_yank_status(context, &crate_version, true).await?
         }
         Change::CrateDeleted { name, .. } => {
             let name: KrateName = name.parse()?;
@@ -177,15 +273,19 @@ async fn process_change(context: &Context, change: &Change, config: &Config) ->
 }
 
 /// Processes crate changes, whether they got yanked or unyanked.
-async fn process_version_yank_status(context: &Context, release: &CrateVersion) -> Result<()> {
+pub(crate) async fn process_version_yank_status(
+    context: &Context,
+    release: &CrateVersion,
+    yanked: bool,
+) -> Result<()> {
     // FIXME: delay yanks of crates that have not yet finished building
     // https://github.com/rust-lang/docs.rs/issues/1934
-    set_yanked(context, &release.name, &release.version, release.yanked).await?;
+    set_yanked(context, &release.name, &release.version, yanked).await?;
     queue_crate_invalidation(&release.name, context.cdn.as_deref()).await;
     Ok(())
 }
 
-async fn process_version_added(context: &Context, release: &CrateVersion) -> Result<()> {
+pub(crate) async fn process_version_added(context: &Context, release: &CrateVersion) -> Result<()> {
     let build_queue = context.build_queue()?;
 
     let priority = build_queue.find_priority(&release.name).await?;
@@ -217,7 +317,7 @@ async fn process_version_added(context: &Context, release: &CrateVersion) -> Res
     Ok(())
 }
 
-async fn process_version_deleted(
+pub(crate) async fn process_version_deleted(
     context: &Context,
     config: &Config,
     release: &CrateVersion,
@@ -251,7 +351,7 @@ async fn process_version_deleted(
     Ok(())
 }
 
-async fn process_crate_deleted(
+pub(crate) async fn process_crate_deleted(
     context: &Context,
     config: &Config,
     krate: &KrateName,
@@ -343,7 +443,6 @@ mod tests {
         let krate = CrateVersion {
             name: KRATE,
             version: V1,
-            ..Default::default()
         };
 
         process_version_added(&env, &krate).await?;
@@ -354,7 +453,6 @@ mod tests {
         let krate = CrateVersion {
             name: "krate".parse()?,
             version: V2.to_string().parse()?,
-            ..Default::default()
         };
 
         process_version_added(&env, &krate).await?;
@@ -387,9 +485,8 @@ mod tests {
         let krate = CrateVersion {
             name: KRATE,
             version: V1,
-            yanked: true,
         };
-        process_version_yank_status(&env, &krate).await?;
+        process_version_yank_status(&env, &krate, true).await?;
 
         // And verify it's actually marked as yanked
         let row = sqlx::query!(
@@ -406,9 +503,8 @@ mod tests {
         let krate = CrateVersion {
             name: KRATE,
             version: V1,
-            yanked: false,
         };
-        process_version_yank_status(&env, &krate).await?;
+        process_version_yank_status(&env, &krate, false).await?;
 
         let row = sqlx::query!(
             "SELECT yanked
@@ -471,7 +567,6 @@ mod tests {
         let krate = CrateVersion {
             name: KRATE,
             version: V2,
-            ..Default::default()
         };
         process_version_deleted(&env, env.config(), &krate).await?;
 
@@ -501,23 +596,20 @@ mod tests {
         let krate1 = CrateVersion {
             name: KRATE,
             version: V1,
-            ..Default::default()
         };
         let krate2 = CrateVersion {
             name: "krate2".parse()?,
             version: V1,
-            ..Default::default()
         };
         let krate_already_present = CrateVersion {
             name: "krate_already_present".parse()?,
             version: V1,
-            ..Default::default()
         };
         let non_existing_version = CrateVersion {
             name: "krate_already_present".parse()?,
             version: V2,
-            ..Default::default()
         };
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
         let added = process_changes(
             &env,
             &vec![
@@ -531,6 +623,7 @@ mod tests {
                 Change::VersionDeleted(non_existing_version.into()),
             ],
             env.config(),
+            &metrics,
         )
         .await;
 
diff --git a/crates/bin/docs_rs_watcher/src/lib.rs b/crates/bin/docs_rs_watcher/src/lib.rs
index 833a6c6885..337388d2b3 100644
--- a/crates/bin/docs_rs_watcher/src/lib.rs
+++ b/crates/bin/docs_rs_watcher/src/lib.rs
@@ -3,8 +3,10 @@ pub mod consistency;
 mod db;
 mod index;
 pub mod index_watcher;
+mod metrics;
 mod rebuilds;
 mod service_metrics;
+mod subscriber;
 #[cfg(test)]
 mod testing;
 
@@ -13,7 +15,11 @@ pub use db::{delete_crate, delete_version};
 pub use index::Index;
 pub use rebuilds::queue_rebuilds;
 
-use crate::{index_watcher::get_new_crates, service_metrics::OtelServiceMetrics};
+use crate::{
+    index_watcher::get_new_crates,
+    metrics::{EventSource, WatcherMetrics},
+    service_metrics::OtelServiceMetrics,
+};
 use anyhow::Result;
 use docs_rs_context::Context;
 use docs_rs_utils::start_async_cron;
@@ -21,12 +27,66 @@ use std::{sync::Arc, time::Duration};
 use tokio::time::{self, Instant};
 use tracing::{debug, error, info, trace};
 
+/// main index-watcher / subscriber loop.
+/// mostly wraps either the git index watcher loop, or the sqs subscriber loop.
+/// Only here so unexpected errors lead to a sentry report & restart instead of
+/// the daemon / watcher just stopping.
+pub async fn watch(config: &Config, context: &Context) {
+    let metrics = WatcherMetrics::new(context.meter_provider());
+
+    // NOTE: for now we don't have a graceful shutdown.
+    // Since we currently always lock the queue & builds before deploys, that's
+    // not a problem.
+    // But I assume with the new AWS infra we need to solve this at some point so we don't loose
+    // events.
+
+    loop {
+        if config.crates_io_events_active() {
+            if let Err(err) = crate::subscriber::run_sqs_subscriber(config, context, &metrics).await
+            {
+                error!(?err, "unexpected error watching SQS, will retry");
+                time::sleep(Duration::from_secs(10)).await;
+            }
+        } else {
+            // intermediate mode:
+            // - still fetch from git for events
+            // - listen so SQS, and log the events so we can test SQS connection, and compare events
+            //
+            // We don't retry on unespected SQS errors yet.
+
+            let registry_watcher = crate::watch_registry(config, context, &metrics);
+            tokio::pin!(registry_watcher);
+
+            let registry_result = tokio::select! {
+                result = &mut registry_watcher => result,
+                sqs_result = crate::subscriber::run_sqs_subscriber(config, context, &metrics) => {
+                    // Unexpected SQS errors stop the test subscriber, but the registry watcher
+                    // remains the authoritative source and must keep running.
+                    if let Err(err) = sqs_result {
+                        error!(?err, "error setting up SQS test subscriber");
+                    }
+                    registry_watcher.await
+                }
+            };
+
+            if let Err(err) = registry_result {
+                // unexpected index watcher errors lead to a report & retry.
+                error!(?err, "unexpected error watching registry, will retry");
+                time::sleep(Duration::from_secs(10)).await;
+            }
+        }
+    }
+}
+
 /// Run the registry watcher
 /// NOTE: this should only be run once, otherwise crates would be added
 /// to the queue multiple times.
-pub async fn watch_registry(config: &Config, context: &Context) -> Result<()> {
+async fn watch_registry(
+    config: &Config,
+    context: &Context,
+    metrics: &WatcherMetrics,
+) -> Result<()> {
     let mut last_gc = Instant::now();
-
     let queue = context.build_queue()?;
 
     loop {
@@ -36,9 +96,10 @@ pub async fn watch_registry(config: &Config, context: &Context) -> Result<()> {
             debug!("Checking new crates");
             let index = Index::from_config(config).await?;
 
-            match get_new_crates(context, &index, config).await {
+            match get_new_crates(context, &index, config, metrics).await {
                 Ok(n) => debug!("{} crates added to queue", n),
                 Err(e) => {
+                    metrics.record_poll_error(EventSource::Git);
                     error!(?e, "Failed to get new crates");
                 }
             }
diff --git a/crates/bin/docs_rs_watcher/src/main.rs b/crates/bin/docs_rs_watcher/src/main.rs
index ebc4f728f9..d4138a3758 100644
--- a/crates/bin/docs_rs_watcher/src/main.rs
+++ b/crates/bin/docs_rs_watcher/src/main.rs
@@ -82,7 +82,7 @@ impl CommandLine {
                 // which should only run once, and all the time.
                 docs_rs_watcher::start_background_service_metric_collector(&ctx).await?;
 
-                docs_rs_watcher::watch_registry(&config, &ctx).await?;
+                docs_rs_watcher::watch(&config, &ctx).await;
             }
             Self::Queue { subcommand } => subcommand.handle_args(config, ctx).await?,
             Self::Database { subcommand } => subcommand.handle_args(config, ctx).await?,
diff --git a/crates/bin/docs_rs_watcher/src/metrics.rs b/crates/bin/docs_rs_watcher/src/metrics.rs
new file mode 100644
index 0000000000..56fad5fa4d
--- /dev/null
+++ b/crates/bin/docs_rs_watcher/src/metrics.rs
@@ -0,0 +1,118 @@
+use docs_rs_crates_io::events::ChangeKind;
+use docs_rs_opentelemetry::AnyMeterProvider;
+use opentelemetry::{
+    KeyValue,
+    metrics::{Counter, Histogram},
+};
+use std::{fmt, time::Duration};
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) enum EventSource {
+    Git,
+    Sqs,
+}
+
+impl EventSource {
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Git => "git",
+            Self::Sqs => "sqs",
+        }
+    }
+}
+
+impl fmt::Display for EventSource {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct WatcherMetrics {
+    events_received_total: Counter<u64>,
+    poll_errors_total: Counter<u64>,
+    changes_applied_total: Counter<u64>,
+    event_processing_time: Histogram<f64>,
+    event_lag: Histogram<f64>,
+}
+
+impl WatcherMetrics {
+    pub(crate) fn new(meter_provider: &AnyMeterProvider) -> Self {
+        let meter = meter_provider.meter("watcher");
+        const PREFIX: &str = "docsrs.watcher";
+        Self {
+            events_received_total: meter
+                .u64_counter(format!("{PREFIX}.events_received_total"))
+                .with_unit("1")
+                .build(),
+            poll_errors_total: meter
+                .u64_counter(format!("{PREFIX}.poll_errors_total"))
+                .with_unit("1")
+                .build(),
+            changes_applied_total: meter
+                .u64_counter(format!("{PREFIX}.changes_applied_total"))
+                .with_unit("1")
+                .build(),
+            event_processing_time: meter
+                .f64_histogram(format!("{PREFIX}.event_processing_time"))
+                .with_boundaries(vec![
+                    0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0,
+                    45.0, 55.0, 60.0, 65.0, 90.0, 120.0,
+                ])
+                .with_unit("s")
+                .build(),
+            event_lag: meter
+                .f64_histogram(format!("{PREFIX}.event_lag"))
+                .with_boundaries(vec![
+                    0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 300.0, 900.0, 3600.0,
+                ])
+                .with_unit("s")
+                .build(),
+        }
+    }
+
+    pub(crate) fn record_change_applied(&self, source: EventSource, kind: ChangeKind) {
+        self.changes_applied_total.add(
+            1,
+            &[
+                KeyValue::new("source", source.as_str()),
+                KeyValue::new("type", kind.as_str()),
+            ],
+        );
+    }
+
+    pub(crate) fn record_event_lag(&self, source: EventSource, duration: Duration) {
+        self.event_lag.record(
+            duration.as_secs_f64(),
+            &[KeyValue::new("source", source.as_str())],
+        );
+    }
+
+    pub(crate) fn record_event_processing_time(
+        &self,
+        source: EventSource,
+        kind: Option<ChangeKind>,
+        success: bool,
+        duration: Duration,
+    ) {
+        let result = if success { "ok" } else { "err" };
+        self.event_processing_time.record(
+            duration.as_secs_f64(),
+            &[
+                KeyValue::new("source", source.as_str()),
+                KeyValue::new("type", kind.map(ChangeKind::as_str).unwrap_or("unknown")),
+                KeyValue::new("result", result),
+            ],
+        );
+    }
+
+    pub(crate) fn record_events_received(&self, source: EventSource, count: usize) {
+        self.events_received_total
+            .add(count as u64, &[KeyValue::new("source", source.as_str())]);
+    }
+
+    pub(crate) fn record_poll_error(&self, source: EventSource) {
+        self.poll_errors_total
+            .add(1, &[KeyValue::new("source", source.as_str())]);
+    }
+}
diff --git a/crates/bin/docs_rs_watcher/src/subscriber.rs b/crates/bin/docs_rs_watcher/src/subscriber.rs
new file mode 100644
index 0000000000..977d2e02c6
--- /dev/null
+++ b/crates/bin/docs_rs_watcher/src/subscriber.rs
@@ -0,0 +1,695 @@
+use crate::{
+    Config,
+    index_watcher::{
+        process_crate_deleted, process_version_added, process_version_deleted,
+        process_version_yank_status,
+    },
+    metrics::{EventSource, WatcherMetrics},
+};
+use anyhow::{Context as _, Result};
+use async_trait::async_trait;
+use aws_config::{BehaviorVersion, Region, retry::RetryConfig};
+use aws_sdk_sqs::{Client, types::Message};
+use chrono::Utc;
+use docs_rs_context::Context;
+use docs_rs_crates_io::events::{IndexChangeEventV1, IndexChangeV1};
+use docs_rs_types::KrateName;
+use docs_rs_utils::retry_async;
+use std::time::{Duration, Instant};
+use tokio::time;
+use tracing::{debug, error, instrument, warn};
+
+/// wait-time (long polling):
+///
+/// How long should the request be kept open when there are no messages.
+/// SQS only accepts values in the range 0..=20 seconds.
+const WAIT_TIME: Duration = Duration::from_secs(20);
+
+/// when one long-polling request is finished, how long to sleep before starting the next?
+const SLEEP_BETWEEN_REQUESTS: Duration = Duration::from_secs(1);
+
+/// How regularly to recheck the priorities of queued crates.
+/// Right now only runs `deprioritize_workspaces`.
+const DELAY_BETWEEN_PRIORITY_RECHECK: Duration = Duration::from_secs(60);
+
+/// visibility timeout:
+/// SQS visibility timeout is the period after a consumer receives a message during
+/// which that message is hidden from other consumers, and if it is not deleted before
+/// the timeout expires, it becomes visible again for redelivery.
+///
+/// Should be longer than the longest time our server takes to handle a message.
+const VISIBILITY_TIMEOUT: Duration = Duration::from_secs(600);
+
+#[async_trait]
+trait SqsActions: Sync {
+    async fn delete_message(&self, queue_url: &str, receipt_handle: &str) -> Result<()>;
+}
+
+#[async_trait]
+impl SqsActions for Client {
+    async fn delete_message(&self, queue_url: &str, receipt_handle: &str) -> Result<()> {
+        self.delete_message()
+            .queue_url(queue_url)
+            .receipt_handle(receipt_handle)
+            .send()
+            .await
+            .context("error deleting SQS message")?;
+        Ok(())
+    }
+}
+
+pub(crate) async fn run_sqs_subscriber(
+    config: &Config,
+    context: &Context,
+    metrics: &WatcherMetrics,
+) -> Result<()> {
+    let Some(sqs_config) = &config.crates_io_events else {
+        warn!("missing sqs config, disabling crates.io SQS subscriber");
+        return Ok(());
+    };
+    let mut last_priority_recheck = Instant::now();
+    let queue = context.build_queue()?;
+
+    debug!("creating SQS client...");
+    let shared_config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+    let mut client_config = aws_sdk_sqs::config::Builder::from(&shared_config)
+        .retry_config(RetryConfig::standard().with_max_attempts(sqs_config.max_retries))
+        .region(Region::new(sqs_config.region.to_string()));
+    if let Some(endpoint_url) = &sqs_config.endpoint_url {
+        client_config = client_config.endpoint_url(endpoint_url.to_string());
+    }
+    let client = Client::from_conf(client_config.build());
+
+    let queue_url = sqs_config.queue_url.to_string();
+
+    loop {
+        if queue.is_locked().await? {
+            debug!("Queue is locked, skipping checking new crates");
+            time::sleep(WAIT_TIME).await;
+            continue;
+        }
+
+        debug!("receiving messages...");
+        let messages = match client
+            .receive_message()
+            .queue_url(&queue_url)
+            // confirm that we want to do batches.
+            // important because it's a FIFO queue:
+            // NOTE: when we start retrying tasks with a FIFO queute.
+            // important: return on on the first erroring message, don't
+            // handle the rest of the batch.
+            .max_number_of_messages(10)
+            .wait_time_seconds(WAIT_TIME.as_secs() as i32)
+            .visibility_timeout(VISIBILITY_TIMEOUT.as_secs() as i32)
+            .send()
+            .await
+        {
+            Ok(response) => response.messages().to_vec(),
+            Err(err) => {
+                // NOTE: right now we handle the change-events like the old
+                // git index: on error just skip over the event, handle the next.
+                // Future improvement: retry the task for retryable errors.
+                metrics.record_poll_error(EventSource::Sqs);
+                error!(?err, queue_url, "error receiving messages from sqs");
+                time::sleep(WAIT_TIME).await;
+                continue;
+            }
+        };
+        process_messages(&client, &queue_url, context, config, metrics, messages).await;
+
+        if last_priority_recheck.elapsed() >= DELAY_BETWEEN_PRIORITY_RECHECK {
+            if let Err(err) = queue.reevaluate_priorities().await {
+                error!(?err, "error reevaluating queued release priorities");
+            }
+
+            last_priority_recheck = Instant::now();
+        }
+
+        time::sleep(SLEEP_BETWEEN_REQUESTS).await;
+    }
+}
+
+async fn process_messages(
+    client: &dyn SqsActions,
+    queue_url: &str,
+    context: &Context,
+    config: &Config,
+    metrics: &WatcherMetrics,
+    messages: Vec<Message>,
+) {
+    let batch_start = Instant::now();
+    let mut error_logged = false;
+
+    for message in &messages {
+        handle_message_body(context, config, metrics, message.body.as_deref()).await;
+        if let Some(receipt_handle) = message.receipt_handle.as_deref()
+            && let Err(err) = client.delete_message(queue_url, receipt_handle).await
+        {
+            error!(?err, receipt_handle, "error deleting message from queue");
+        }
+
+        if !error_logged && batch_start.elapsed() >= VISIBILITY_TIMEOUT {
+            // NOTE: When the message is still in the queue ( not deleted here) after
+            // the visibility-timeout is reached, SQS will redeliver it, assuming that
+            // something went wrong.
+            // So in these cases we'll get duplicate messages.
+            error!(
+                messages = ?messages,
+                visibility_timetout = VISIBILITY_TIMEOUT.as_secs_f64(),
+                "handling message batch took longer than the visibility timeout!"
+            );
+            error_logged = true;
+        }
+    }
+}
+
+async fn handle_message_body(
+    context: &Context,
+    config: &Config,
+    metrics: &WatcherMetrics,
+    body: Option<&str>,
+) {
+    let Some(body) = body else {
+        return;
+    };
+    if let Err(err) = process_sqs_event(context, config, metrics, body).await {
+        // Match the git-index watcher behavior for the initial rollout: record and skip
+        // failed events instead of letting one event block the FIFO queue indefinitely.
+        error!(?err, body, "error handling message, skipping event");
+    }
+}
+
+#[instrument(skip_all)]
+async fn process_sqs_event(
+    context: &Context,
+    config: &Config,
+    metrics: &WatcherMetrics,
+    body: &str,
+) -> Result<()> {
+    metrics.record_events_received(EventSource::Sqs, 1);
+
+    let start = Instant::now();
+    let event: IndexChangeEventV1 = match serde_json::from_str(body) {
+        Ok(event) => event,
+        Err(err) => {
+            metrics.record_event_processing_time(EventSource::Sqs, None, false, start.elapsed());
+            return Err(err).context("error parsing event from json");
+        }
+    };
+
+    debug!(
+        target: "docs_rs_watcher::index_event",
+        source = %EventSource::Sqs,
+        event_id = %event.id,
+        occurred_at = %event.occurred_at,
+        change_type = %event.change.kind(),
+        crate_name = event.change.name(),
+        crate_version = event.change.version().unwrap_or_default(),
+        "crates.io index event"
+    );
+
+    if let Ok(lag) = (Utc::now() - event.occurred_at).to_std() {
+        metrics.record_event_lag(EventSource::Sqs, lag);
+    }
+
+    let processing_result = if config.crates_io_events_active() {
+        retry_async(
+            || {
+                let change = event.change.clone();
+                async move { process_change(context, &change, config).await }
+            },
+            3,
+        )
+        .await
+        .context("error processing change")
+    } else {
+        Ok(())
+    };
+
+    metrics.record_event_processing_time(
+        EventSource::Sqs,
+        Some(event.change.kind()),
+        processing_result.is_ok(),
+        start.elapsed(),
+    );
+    processing_result?;
+
+    if config.crates_io_events_active() {
+        metrics.record_change_applied(EventSource::Sqs, event.change.kind());
+    }
+
+    Ok(())
+}
+
+/// Process a crate change
+#[instrument(skip(context, config))]
+pub(crate) async fn process_change(
+    context: &Context,
+    change: &IndexChangeV1,
+    config: &Config,
+) -> Result<()> {
+    match change {
+        IndexChangeV1::Added(crate_version) => {
+            process_version_added(context, &crate_version.try_into()?).await?
+        }
+        IndexChangeV1::Yanked(crate_version) => {
+            process_version_yank_status(context, &crate_version.try_into()?, true).await?
+        }
+        IndexChangeV1::Unyanked(crate_version) => {
+            process_version_yank_status(context, &crate_version.try_into()?, false).await?
+        }
+        IndexChangeV1::CrateDeleted { name, .. } => {
+            let name: KrateName = name.parse()?;
+            process_crate_deleted(context, config, &name).await?
+        }
+        IndexChangeV1::VersionDeleted(crate_version) => {
+            process_version_deleted(context, config, &crate_version.try_into()?).await?
+        }
+    };
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::testing::TestEnvironment;
+    use docs_rs_config::AppConfig as _;
+    use docs_rs_crates_io::events::CrateVersion;
+    use docs_rs_types::{
+        Version,
+        testing::{KRATE, V1, V2},
+    };
+    use pretty_assertions::assert_eq;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct FakeSqsActions {
+        deleted: Mutex<Vec<String>>,
+    }
+
+    #[async_trait]
+    impl SqsActions for FakeSqsActions {
+        async fn delete_message(&self, _queue_url: &str, receipt_handle: &str) -> Result<()> {
+            self.deleted.lock().unwrap().push(receipt_handle.into());
+            Ok(())
+        }
+    }
+
+    fn added_event_json(name: &KrateName, version: &Version) -> String {
+        serde_json::to_string(&serde_json::json!({
+            "id":"evt_123",
+            "occurred_at":"2026-06-01T12:00:00Z",
+            "type":"added",
+            "payload":{
+                "name": name.to_string(),
+                "vers": version.to_string(),
+            }
+        }))
+        .unwrap()
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_added_queues_crate() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+
+        process_change(
+            &env,
+            &IndexChangeV1::Added(CrateVersion {
+                name: KRATE.to_string(),
+                version: V1.to_string(),
+            }),
+            env.config(),
+        )
+        .await?;
+
+        let queue = env.build_queue()?.queued_crates().await?;
+        assert_eq!(queue.len(), 1);
+        assert_eq!(queue[0].name, KRATE);
+        assert_eq!(queue[0].version, V1);
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_yanked_updates_release() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        let id = env
+            .fake_release()
+            .await
+            .name(KRATE)
+            .version(V1)
+            .create()
+            .await?;
+
+        process_change(
+            &env,
+            &IndexChangeV1::Yanked(CrateVersion {
+                name: KRATE.to_string(),
+                version: V1.to_string(),
+            }),
+            env.config(),
+        )
+        .await?;
+
+        let yanked = sqlx::query_scalar!(
+            "SELECT yanked
+             FROM releases
+             WHERE id = $1",
+            id.0
+        )
+        .fetch_one(&mut *conn)
+        .await?;
+        assert_eq!(yanked, Some(true));
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_unyanked_updates_release() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        let id = env
+            .fake_release()
+            .await
+            .name(KRATE)
+            .version(V1)
+            .yanked(true)
+            .create()
+            .await?;
+
+        process_change(
+            &env,
+            &IndexChangeV1::Unyanked(CrateVersion {
+                name: KRATE.to_string(),
+                version: V1.to_string(),
+            }),
+            env.config(),
+        )
+        .await?;
+
+        let row = sqlx::query!(
+            "SELECT yanked
+             FROM releases
+             WHERE id = $1",
+            id.0
+        )
+        .fetch_one(&mut *conn)
+        .await?;
+        assert_eq!(row.yanked, Some(false));
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_crate_deleted_removes_crate() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        env.fake_release()
+            .await
+            .name(KRATE)
+            .version(V1)
+            .create()
+            .await?;
+
+        process_change(
+            &env,
+            &IndexChangeV1::CrateDeleted {
+                name: KRATE.to_string(),
+            },
+            env.config(),
+        )
+        .await?;
+
+        let row = sqlx::query!(
+            "SELECT id
+             FROM crates
+             WHERE name = $1",
+            KRATE as _
+        )
+        .fetch_optional(&mut *conn)
+        .await?;
+        assert!(row.is_none());
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_added_is_idempotent() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let change = IndexChangeV1::Added(CrateVersion {
+            name: KRATE.to_string(),
+            version: V1.to_string(),
+        });
+
+        process_change(&env, &change, env.config()).await?;
+        process_change(&env, &change, env.config()).await?;
+
+        let queue = env.build_queue()?.queued_crates().await?;
+        assert_eq!(queue.len(), 1);
+        assert_eq!(queue[0].name, KRATE);
+        assert_eq!(queue[0].version, V1);
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_version_deleted_removes_release() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let mut conn = env.async_conn().await?;
+
+        let rid_1 = env
+            .fake_release()
+            .await
+            .name(KRATE)
+            .version(V1)
+            .create()
+            .await?;
+        env.fake_release()
+            .await
+            .name(KRATE)
+            .version(V2)
+            .create()
+            .await?;
+
+        process_change(
+            &env,
+            &IndexChangeV1::VersionDeleted(CrateVersion {
+                name: KRATE.to_string(),
+                version: V2.to_string(),
+            }),
+            env.config(),
+        )
+        .await?;
+
+        assert_eq!(
+            sqlx::query_scalar!("SELECT id FROM releases")
+                .fetch_all(&mut *conn)
+                .await?,
+            vec![rid_1.0]
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_change_version_deleted_is_idempotent() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        env.fake_release()
+            .await
+            .name(KRATE)
+            .version(V1)
+            .create()
+            .await?;
+        let change = IndexChangeV1::VersionDeleted(CrateVersion {
+            name: KRATE.to_string(),
+            version: V1.to_string(),
+        });
+
+        process_change(&env, &change, env.config()).await?;
+        process_change(&env, &change, env.config()).await?;
+
+        assert!(
+            sqlx::query_scalar!("SELECT id FROM releases")
+                .fetch_all(&mut *env.async_conn().await?)
+                .await?
+                .is_empty()
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_sqs_event_dispatches_added_event() -> Result<()> {
+        let mut config = Config::test_config()?;
+        if let Some(sqs_config) = &mut config.crates_io_events {
+            sqs_config.active = true;
+        }
+        let env = TestEnvironment::builder().config(config).build().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        process_sqs_event(&env, env.config(), &metrics, &added_event_json(&KRATE, &V1)).await?;
+
+        let queue = env.build_queue()?.queued_crates().await?;
+        assert_eq!(queue.len(), 1);
+        assert_eq!(queue[0].name, KRATE);
+        assert_eq!(queue[0].version, V1);
+        let collected = env.collected_metrics();
+        let applied_metric =
+            collected.get_metric("watcher", "docsrs.watcher.changes_applied_total")?;
+        let applied = applied_metric.get_u64_counter();
+        let change_type = applied
+            .attributes()
+            .find(|kv| kv.key.as_str() == "type")
+            .unwrap()
+            .value
+            .to_string();
+        assert_eq!(change_type, "added");
+        assert_eq!(applied.value(), 1);
+        let lag_metric = collected.get_metric("watcher", "docsrs.watcher.event_lag")?;
+        assert_eq!(lag_metric.get_f64_histogram().count(), 1);
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_sqs_event_respects_sqs_active() -> Result<()> {
+        let mut config = Config::test_config()?;
+        if let Some(sqs_config) = &mut config.crates_io_events {
+            sqs_config.active = false;
+        }
+        let env = TestEnvironment::builder().config(config).build().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        process_sqs_event(&env, env.config(), &metrics, &added_event_json(&KRATE, &V1)).await?;
+
+        assert!(env.build_queue()?.queued_crates().await?.is_empty());
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_sqs_event_rejects_invalid_json() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        let err = process_sqs_event(&env, env.config(), &metrics, "{not json").await;
+
+        assert!(err.is_err());
+        let err = format!("{:?}", err.unwrap_err());
+        assert!(
+            err.contains("error parsing event from json"),
+            "unexpected error: {err}"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_handle_message_body_acknowledges_success() -> Result<()> {
+        let config = Config::test_config()?;
+        let env = TestEnvironment::builder().config(config).build().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        handle_message_body(
+            &env,
+            env.config(),
+            &metrics,
+            Some(&added_event_json(&KRATE, &V1)),
+        )
+        .await;
+        let collected = env.collected_metrics();
+        let processing_metric =
+            collected.get_metric("watcher", "docsrs.watcher.event_processing_time")?;
+        assert_eq!(processing_metric.get_f64_histogram().count(), 1);
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_handle_message_body_records_failed_processing() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        handle_message_body(&env, env.config(), &metrics, Some("{bad json")).await;
+        let collected = env.collected_metrics();
+        let processing_metric =
+            collected.get_metric("watcher", "docsrs.watcher.event_processing_time")?;
+        let processing = processing_metric.get_f64_histogram();
+        assert_eq!(processing.count(), 1);
+        assert!(processing.attributes().any(|attribute| {
+            attribute.key.as_str() == "result" && attribute.value.to_string() == "err"
+        }));
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_handle_message_body_acknowledges_missing_body() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+
+        handle_message_body(&env, env.config(), &metrics, None).await;
+        assert!(env.build_queue()?.queued_crates().await?.is_empty());
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_messages_skips_errors_and_continues_batch() -> Result<()> {
+        let config = Config::test_config()?;
+        let env = TestEnvironment::builder().config(config).build().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+        let client = FakeSqsActions::default();
+        let messages = vec![
+            Message::builder()
+                .body(added_event_json(&KRATE, &V1))
+                .receipt_handle("success-1")
+                .build(),
+            Message::builder()
+                .body("{bad json")
+                .receipt_handle("failure")
+                .build(),
+            Message::builder()
+                .body(added_event_json(&KRATE, &V2))
+                .receipt_handle("success-2")
+                .build(),
+        ];
+
+        process_messages(&client, "queue-url", &env, env.config(), &metrics, messages).await;
+
+        assert_eq!(
+            *client.deleted.lock().unwrap(),
+            vec!["success-1", "failure", "success-2"]
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_process_messages_without_body_is_acknowledged() -> Result<()> {
+        let config = Config::test_config()?;
+        let env = TestEnvironment::builder().config(config).build().await?;
+        let metrics = WatcherMetrics::new(&env.context().meter_provider);
+        let client = FakeSqsActions::default();
+
+        process_messages(
+            &client,
+            "queue-url",
+            &env,
+            env.config(),
+            &metrics,
+            vec![Message::builder().receipt_handle("missing-body").build()],
+        )
+        .await;
+
+        assert_eq!(
+            *client.deleted.lock().unwrap(),
+            vec!["missing-body".to_string()]
+        );
+        Ok(())
+    }
+}
diff --git a/crates/bin/docs_rs_web/src/lib.rs b/crates/bin/docs_rs_web/src/lib.rs
index edd28d7c30..41fd32eaa2 100644
--- a/crates/bin/docs_rs_web/src/lib.rs
+++ b/crates/bin/docs_rs_web/src/lib.rs
@@ -1,4 +1,3 @@
-#![recursion_limit = "256"]
 #![allow(
     // clippy::cognitive_complexity,
     // TODO: `AxumNope::Redirect(EscapedURI, CachePolicy)` is too big.
diff --git a/crates/lib/docs_rs_crates_io/Cargo.toml b/crates/lib/docs_rs_crates_io/Cargo.toml
index 497536d66e..6ac6ab55d2 100644
--- a/crates/lib/docs_rs_crates_io/Cargo.toml
+++ b/crates/lib/docs_rs_crates_io/Cargo.toml
@@ -9,11 +9,12 @@ repository.workspace = true
 edition.workspace = true
 
 [dependencies]
-chrono = { version = "0.4", features = ["serde"] }
-serde = { version = "1", features = ["derive"] }
+chrono = { workspace = true }
+serde = { workspace = true }
 
 [dev-dependencies]
-serde_json = "1.0"
+serde_json = { workspace = true }
+test-case = { workspace = true }
 
 [lints]
 workspace = true
diff --git a/crates/lib/docs_rs_crates_io/src/events.rs b/crates/lib/docs_rs_crates_io/src/events.rs
index f90484abc2..0169f02972 100644
--- a/crates/lib/docs_rs_crates_io/src/events.rs
+++ b/crates/lib/docs_rs_crates_io/src/events.rs
@@ -1,6 +1,35 @@
 use chrono::{DateTime, Utc};
 use std::fmt;
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ChangeKind {
+    Added,
+    AddedAndYanked,
+    Unyanked,
+    Yanked,
+    CrateDeleted,
+    VersionDeleted,
+}
+
+impl ChangeKind {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Added => "added",
+            Self::AddedAndYanked => "added_and_yanked",
+            Self::Unyanked => "unyanked",
+            Self::Yanked => "yanked",
+            Self::CrateDeleted => "crate_deleted",
+            Self::VersionDeleted => "version_deleted",
+        }
+    }
+}
+
+impl fmt::Display for ChangeKind {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
 /// A change that can happen to a crate on our index.
 #[derive(Clone, serde::Serialize, serde::Deserialize, Eq, PartialEq, Debug)]
 #[serde(tag = "type", content = "payload", rename_all = "snake_case")]
@@ -57,17 +86,41 @@ impl IndexChangeV1 {
             _ => None,
         }
     }
+
+    pub fn name(&self) -> &str {
+        match self {
+            IndexChangeV1::Added(crate_version) => &crate_version.name,
+            IndexChangeV1::Unyanked(crate_version) => &crate_version.name,
+            IndexChangeV1::Yanked(crate_version) => &crate_version.name,
+            IndexChangeV1::CrateDeleted { name } => name,
+            IndexChangeV1::VersionDeleted(crate_version) => &crate_version.name,
+        }
+    }
+
+    pub fn version(&self) -> Option<&str> {
+        match self {
+            IndexChangeV1::Added(crate_version) => Some(&crate_version.version),
+            IndexChangeV1::Unyanked(crate_version) => Some(&crate_version.version),
+            IndexChangeV1::Yanked(crate_version) => Some(&crate_version.version),
+            IndexChangeV1::CrateDeleted { .. } => None,
+            IndexChangeV1::VersionDeleted(crate_version) => Some(&crate_version.version),
+        }
+    }
+
+    pub fn kind(&self) -> ChangeKind {
+        match *self {
+            IndexChangeV1::Added(_) => ChangeKind::Added,
+            IndexChangeV1::Yanked(_) => ChangeKind::Yanked,
+            IndexChangeV1::CrateDeleted { .. } => ChangeKind::CrateDeleted,
+            IndexChangeV1::VersionDeleted(_) => ChangeKind::VersionDeleted,
+            IndexChangeV1::Unyanked(_) => ChangeKind::Unyanked,
+        }
+    }
 }
 
 impl fmt::Display for IndexChangeV1 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(match *self {
-            IndexChangeV1::Added(_) => "added",
-            IndexChangeV1::Yanked(_) => "yanked",
-            IndexChangeV1::CrateDeleted { .. } => "crate deleted",
-            IndexChangeV1::VersionDeleted(_) => "version deleted",
-            IndexChangeV1::Unyanked(_) => "unyanked",
-        })
+        self.kind().fmt(f)
     }
 }
 
diff --git a/crates/lib/docs_rs_storage/Cargo.toml b/crates/lib/docs_rs_storage/Cargo.toml
index 1980060474..ba7060c8bd 100644
--- a/crates/lib/docs_rs_storage/Cargo.toml
+++ b/crates/lib/docs_rs_storage/Cargo.toml
@@ -16,11 +16,7 @@ testing = [
 anyhow = { workspace = true }
 async-compression = { version = "0.4.32", features = ["bzip2", "deflate", "gzip", "tokio", "zstd"] }
 async-stream = { workspace = true }
-# 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"] }
+aws-config = { workspace = true }
 aws-sdk-s3 = { version = "1.3.0", default-features = false, features = ["default-https-client", "rt-tokio"] }
 aws-smithy-types-convert = { version = "0.61.0", features = ["convert-chrono"] }
 base64 = { workspace = true }
diff --git a/docker-compose.yml b/docker-compose.yml
index 87f605d7fc..6e695ba457 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -19,6 +19,7 @@
 #     - repo-stats updater
 #     - cdn invalidator
 #     - release-rebuild-enqueuer
+# * `elasticmq` -> local SQS-compatible queue for watcher testing
 #
 # optional profile: `metrics`:
 # * `opentelemetry` -> a debug opentelemetry receiver
@@ -120,6 +121,7 @@ x-registry-watcher: &registry-watcher
     depends_on:
         - db
         - s3
+        - elasticmq
     volumes:
         - "./ignored/docker-registry-watcher/prefix:/opt/docsrs/prefix"
         - crates-io-index:/opt/docsrs/crates.io-index
@@ -132,6 +134,10 @@ x-registry-watcher: &registry-watcher
         REGISTRY_INDEX_PATH: /opt/docsrs/crates.io-index
         # configure the rebuild-queuer,
         DOCSRS_MAX_QUEUED_REBUILDS: ${DOCSRS_MAX_QUEUED_REBUILDS:-10}
+        DOCSRS_SQS_QUEUE_URL: ${DOCSRS_SQS_QUEUE_URL:-http://elasticmq:9324/queue/docsrs-events}
+        DOCSRS_SQS_QUEUE_REGION: ${DOCSRS_SQS_QUEUE_REGION:-elasticmq}
+        DOCSRS_SQS_ENDPOINT_URL: ${DOCSRS_SQS_ENDPOINT_URL:-http://elasticmq:9324}
+        DOCSRS_SQS_ACTIVE: ${DOCSRS_SQS_ACTIVE:-false}
     env_file:
         - .docker.env
 
@@ -169,6 +175,17 @@ services:
             # watcher-CLI should not be run as background daemon, just manually
             - manual
 
+    elasticmq:
+        image: softwaremill/elasticmq
+        ports:
+            - "127.0.0.1:9324:9324"
+        volumes:
+            - "./dockerfiles/elasticmq.conf:/opt/elasticmq.conf:ro"
+        command: ["-Dconfig.file=/opt/elasticmq.conf"]
+        healthcheck:
+            <<: *healthcheck-interval
+            test: curl --silent --fail http://localhost:9324/health
+
     builder-a:
         <<: *builder
         volumes:
diff --git a/dockerfiles/elasticmq.conf b/dockerfiles/elasticmq.conf
new file mode 100644
index 0000000000..fb77fac15d
--- /dev/null
+++ b/dockerfiles/elasticmq.conf
@@ -0,0 +1,18 @@
+include classpath("application.conf")
+
+node-address {
+  protocol = http
+  host = "*"
+  port = 9324
+  context-path = ""
+}
+
+rest-sqs {
+  enabled = true
+  bind-port = 9324
+  bind-hostname = "0.0.0.0"
+}
+
+queues {
+  docsrs-events { }
+}
diff --git a/justfiles/utils.just b/justfiles/utils.just
index db9458e2cc..fa8663e82b 100644
--- a/justfiles/utils.just
+++ b/justfiles/utils.just
@@ -2,11 +2,31 @@
 _ensure_db_and_s3_are_running: _touch-docker-env
     # dependencies in the docker-cli file are ignored
     # here. Instead we explicitly start any dependent services first.
-    docker compose up -d db s3 --wait
+    docker compose up -d db s3 elasticmq --wait
 
 _touch-docker-env:
     touch .docker.env
 
+send-sqs-payload:
+    #!/usr/bin/env bash
+    set -euo pipefail
+
+    payload='{
+      "id": "evt_1",
+      "occurred_at": "2026-07-02T12:00:00Z",
+      "type": "added",
+      "payload": {
+        "name": "demo-crate",
+        "vers": "1.2.3"
+      }
+    }'
+
+    aws sqs send-message \
+      --endpoint-url $DOCSRS_SQS_ENDPOINT_URL \
+      --region elasticmq \
+      --queue-url $DOCSRS_SQS_QUEUE_URL \
+      --message-body "$payload"
+
 # helper recipe to ensure a CLI tool is installed.
 # * Accepts multiple names
 # * uses `cargo binstall` if it exists.