diff --git a/crates/server/src/console/pages/auth_pages.rs b/crates/server/src/console/pages/auth_pages.rs
index b869e5434..f590cf1a4 100755
--- a/crates/server/src/console/pages/auth_pages.rs
+++ b/crates/server/src/console/pages/auth_pages.rs
@@ -9,12 +9,14 @@ use axum::Form;
 use axum::extract::{ConnectInfo, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{Html, IntoResponse, Response};
+use extenddb_storage::management_store::RateLimitStore;
 use serde::Deserialize;
 
 use crate::console::ConsoleState;
 use crate::console::html;
 use crate::management::CallerIdentity;
-use crate::rate_limit;
+use crate::management::auth::log_password_check_error;
+use crate::rate_limit::{self, LoginAttempt, LoginRefusal};
 
 use super::extract_session_token;
 
@@ -35,41 +37,24 @@ pub async fn login_submit(
     ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Form(form): Form<LoginForm>,
 ) -> Response {
-    let source_ip = addr.ip().to_string();
-
-    // Rate limit / lockout check before attempting authentication.
-    if let Err(msg) =
-        rate_limit::check_login_allowed(&*state.catalog_store, &form.username, Some(&source_ip))
-            .await
-    {
-        return (
-            StatusCode::TOO_MANY_REQUESTS,
-            Html(login_html(Some(&msg), &state.listen_url)),
-        )
-            .into_response();
-    }
+    // Reserve the attempt before any password is checked. Limits are per
+    // username and peer address, so failures from elsewhere do not lock this
+    // address out.
+    let attempt = match rate_limit::begin_login(&*state.catalog_store, &form.username, addr).await {
+        Ok(attempt) => attempt,
+        Err(refusal) => return refusal_page(refusal, &state.listen_url),
+    };
 
-    // Try admin auth first.
+    // Try admin auth first, then IAM user auth. A storage error (including a
+    // timed-out wait for a bcrypt slot) is answered 500 and recorded as
+    // nothing: no password was checked.
+    let checked = check_credentials(&form.username, &form.password, &*state.catalog_store).await;
     let identity =
-        match try_admin_login(&form.username, &form.password, &*state.catalog_store).await {
-            AdminLoginResult::Authenticated(id) => Some(id),
-            AdminLoginResult::WrongPassword => None, // Admin exists, wrong password — do NOT fall through.
-            AdminLoginResult::NotFound => {
-                // Admin not found — try IAM user auth.
-                try_iam_login(&form.username, &form.password, &*state.catalog_store).await
-            }
+        match settle_attempt(attempt, checked, &*state.catalog_store, &state.listen_url).await {
+            Ok(identity) => identity,
+            Err(page) => return page,
         };
 
-    let Some(identity) = identity else {
-        rate_limit::record_failed_login(&*state.catalog_store, &form.username, Some(&source_ip))
-            .await;
-        return (
-            StatusCode::OK,
-            Html(login_html(Some("Invalid credentials"), &state.listen_url)),
-        )
-            .into_response();
-    };
-
     let (token, _csrf) = state.sessions.create(identity).await;
     let cookie = format!(
         "extenddb_session={token}; Path=/console; HttpOnly; SameSite=Strict; Max-Age=28800"
@@ -139,37 +124,203 @@ enum AdminLoginResult {
     NotFound,
 }
 
+/// The login page with a server error, answered when the store failed before
+/// or while the password was checked. Nothing is recorded against the caller.
+fn internal_error(listen_url: &str) -> Response {
+    (
+        StatusCode::INTERNAL_SERVER_ERROR,
+        Html(login_html(
+            Some("Internal error. Please try again later."),
+            listen_url,
+        )),
+    )
+        .into_response()
+}
+
+/// The login page answered to a wrong password or an unknown principal.
+fn invalid_credentials(listen_url: &str) -> Response {
+    (
+        StatusCode::OK,
+        Html(login_html(Some("Invalid credentials"), listen_url)),
+    )
+        .into_response()
+}
+
+/// The login page answered to an attempt the rate limiter did not admit.
+fn refusal_page(refusal: LoginRefusal, listen_url: &str) -> Response {
+    match refusal {
+        LoginRefusal::TooManyAttempts(msg) => (
+            StatusCode::TOO_MANY_REQUESTS,
+            Html(login_html(Some(msg), listen_url)),
+        )
+            .into_response(),
+        LoginRefusal::InvalidCredentials => invalid_credentials(listen_url),
+        LoginRefusal::Internal => internal_error(listen_url),
+    }
+}
+
+/// Check the form's credentials, admin first and then IAM user. `Ok(None)`
+/// is a wrong password or an unknown principal; `Err(())` is a storage
+/// error, including a timed-out wait for a bcrypt slot.
+async fn check_credentials(
+    username: &str,
+    password: &str,
+    store: &dyn extenddb_storage::CatalogStore,
+) -> Result<Option<CallerIdentity>, ()> {
+    match try_admin_login(username, password, store).await? {
+        AdminLoginResult::Authenticated(id) => Ok(Some(id)),
+        // Admin exists, wrong password: do NOT fall through.
+        AdminLoginResult::WrongPassword => Ok(None),
+        // Admin not found: try IAM user auth.
+        AdminLoginResult::NotFound => try_iam_login(username, password, store).await,
+    }
+}
+
+/// Settle an admitted attempt with the result of its password check. A
+/// wrong password is recorded as a failure and answered with the form's
+/// error. A storage error is answered 500 and recorded as nothing; dropping
+/// the attempt releases its reservation.
+async fn settle_attempt(
+    attempt: LoginAttempt,
+    checked: Result<Option<CallerIdentity>, ()>,
+    rate_limiter: &dyn RateLimitStore,
+    listen_url: &str,
+) -> Result<CallerIdentity, Response> {
+    match checked {
+        Ok(Some(identity)) => Ok(identity),
+        Ok(None) => {
+            attempt.record_failure(rate_limiter).await;
+            Err(invalid_credentials(listen_url))
+        }
+        Err(()) => Err(internal_error(listen_url)),
+    }
+}
+
+/// `Err(())` is a storage error; the caller answers 500 and records nothing.
 async fn try_admin_login(
     username: &str,
     password: &str,
     store: &dyn extenddb_storage::management_store::AdminStore,
-) -> AdminLoginResult {
-    let Ok(result) = store.verify_admin_password(username, password).await else {
-        return AdminLoginResult::NotFound;
-    };
-    match result {
+) -> Result<AdminLoginResult, ()> {
+    let result = store
+        .verify_admin_password(username, password)
+        .await
+        .map_err(|e| log_password_check_error("Console login: DB error during admin auth", &e))?;
+    Ok(match result {
         Some(true) => AdminLoginResult::Authenticated(CallerIdentity::Admin(username.to_owned())),
         Some(false) => AdminLoginResult::WrongPassword,
         None => AdminLoginResult::NotFound,
-    }
+    })
 }
 
+/// `Ok(None)` is a wrong password or an unknown user; `Err(())` is a storage
+/// error, which the caller answers 500 and records as nothing.
 async fn try_iam_login(
     username: &str,
     password: &str,
     store: &dyn extenddb_storage::management_store::ManagementStore,
-) -> Option<CallerIdentity> {
-    let (acct_id, uname) = username.split_once('/')?;
+) -> Result<Option<CallerIdentity>, ()> {
+    let Some((acct_id, uname)) = username.split_once('/') else {
+        return Ok(None);
+    };
     let ok = store
         .verify_iam_user_password(acct_id, uname, password)
         .await
-        .ok()?;
-    if ok {
-        Some(CallerIdentity::IamUser {
-            account_id: acct_id.to_owned(),
-            user_name: uname.to_owned(),
-        })
-    } else {
-        None
+        .map_err(|e| {
+            log_password_check_error("Console login: DB error during IAM user auth", &e)
+        })?;
+    Ok(ok.then(|| CallerIdentity::IamUser {
+        account_id: acct_id.to_owned(),
+        user_name: uname.to_owned(),
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use extenddb_storage::password::VERIFY_OVERLOADED;
+
+    use super::*;
+    use crate::management::auth::tests::FailingAdmins;
+    use crate::rate_limit::PRINCIPAL_LIMITED;
+    use crate::rate_limit::tests::{MemoryStore, capture_logs, peer};
+
+    const LISTEN_URL: &str = "https://127.0.0.1:8443";
+
+    async fn body_text(response: Response) -> String {
+        let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        String::from_utf8(bytes.to_vec()).unwrap()
+    }
+
+    #[tokio::test]
+    async fn a_refused_attempt_is_answered_by_its_reason() {
+        let limited = refusal_page(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED), LISTEN_URL);
+        assert_eq!(limited.status(), StatusCode::TOO_MANY_REQUESTS);
+        assert!(body_text(limited).await.contains(PRINCIPAL_LIMITED));
+
+        // The attempts store could not be read: a server error, not a limit.
+        let internal = refusal_page(LoginRefusal::Internal, LISTEN_URL);
+        assert_eq!(internal.status(), StatusCode::INTERNAL_SERVER_ERROR);
+
+        // A principal over the length limit gets the wrong-password page.
+        let too_long = refusal_page(LoginRefusal::InvalidCredentials, LISTEN_URL);
+        assert_eq!(too_long.status(), StatusCode::OK);
+        assert!(body_text(too_long).await.contains("Invalid credentials"));
+    }
+
+    #[tokio::test]
+    async fn a_store_error_in_the_admin_check_is_an_error_not_an_unknown_admin() {
+        let admins = FailingAdmins {
+            message: "store is down",
+        };
+        let (logs, _guard) = capture_logs();
+        assert!(try_admin_login("admin", "pw", &admins).await.is_err());
+        let text = logs.text();
+        assert!(
+            text.contains("ERROR") && text.contains("Console login: DB error during admin auth"),
+            "{text}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_bcrypt_slot_timeout_is_an_error_and_is_not_logged_as_a_store_error() {
+        let admins = FailingAdmins {
+            message: VERIFY_OVERLOADED,
+        };
+        let (logs, _guard) = capture_logs();
+        assert!(try_admin_login("admin", "pw", &admins).await.is_err());
+        let text = logs.text();
+        assert!(!text.contains("DB error"), "{text}");
+    }
+
+    // Each test uses its own source address: the in-flight registry is
+    // process-wide.
+
+    #[tokio::test]
+    async fn a_store_error_during_the_password_check_answers_500_and_records_nothing() {
+        let attempts = MemoryStore::default();
+        let attempt = rate_limit::begin_login(&attempts, "admin", peer("198.51.100.90"))
+            .await
+            .unwrap();
+        let page = settle_attempt(attempt, Err(()), &attempts, LISTEN_URL)
+            .await
+            .unwrap_err();
+        assert_eq!(page.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        assert_eq!(attempts.row_count(), 0);
+    }
+
+    #[tokio::test]
+    async fn a_wrong_password_answers_the_form_error_and_records_one_failure() {
+        let attempts = MemoryStore::default();
+        let attempt = rate_limit::begin_login(&attempts, "admin", peer("198.51.100.91"))
+            .await
+            .unwrap();
+        let page = settle_attempt(attempt, Ok(None), &attempts, LISTEN_URL)
+            .await
+            .unwrap_err();
+        assert_eq!(page.status(), StatusCode::OK);
+        assert!(body_text(page).await.contains("Invalid credentials"));
+        assert_eq!(attempts.rows_for("admin", "198.51.100.91"), 1);
     }
 }
diff --git a/crates/server/src/management/account.rs b/crates/server/src/management/account.rs
index abc2d2870..09b1ae8b6 100755
--- a/crates/server/src/management/account.rs
+++ b/crates/server/src/management/account.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::Serialize;
@@ -40,10 +40,11 @@ struct AccountEntry {
 pub async fn create_account(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     axum::Json(body): axum::Json<CreateAccountRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -85,9 +86,10 @@ pub async fn create_account(
 pub async fn list_accounts(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -117,10 +119,11 @@ pub async fn list_accounts(
 pub async fn delete_account(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(id): Path<String>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/admin.rs b/crates/server/src/management/admin.rs
index 919f3918e..9b1d82a32 100755
--- a/crates/server/src/management/admin.rs
+++ b/crates/server/src/management/admin.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -34,10 +34,11 @@ pub struct ChangePasswordRequest {
 pub async fn create_admin(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     axum::Json(body): axum::Json<CreateAdminRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -91,9 +92,10 @@ pub async fn create_admin(
 pub async fn list_admins(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -123,13 +125,14 @@ pub async fn list_admins(
 pub async fn delete_admin(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(name): Path<String>,
 ) -> Response {
     let caller = match authenticate_admin(
         &headers,
         &*state.catalog_store,
         &*state.catalog_store,
-        None,
+        addr,
     )
     .await
     {
@@ -164,11 +167,12 @@ pub async fn delete_admin(
 pub async fn change_admin_password(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(name): Path<String>,
     axum::Json(body): axum::Json<ChangePasswordRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/assume_role.rs b/crates/server/src/management/assume_role.rs
index 0be4b5295..c6eb1bd56 100755
--- a/crates/server/src/management/assume_role.rs
+++ b/crates/server/src/management/assume_role.rs
@@ -69,14 +69,8 @@ pub async fn assume_role(
     body: axum::body::Bytes,
 ) -> Response {
     // Authenticate via Basic auth — admin only.
-    let source_ip = crate::authorization::condition_source_ip(&addr);
-    if let Err(e) = authenticate_admin(
-        &headers,
-        &*state.catalog_store,
-        &*state.catalog_store,
-        Some(&source_ip),
-    )
-    .await
+    if let Err(e) =
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -182,7 +176,7 @@ pub async fn assume_role(
         request_tags,
         &state.region,
         state.secure_transport,
-        Some(source_ip.as_str()),
+        Some(crate::authorization::condition_source_ip(&addr).as_str()),
         user_agent.as_deref(),
         &account_id,
         &*state.catalog_store,
diff --git a/crates/server/src/management/auth.rs b/crates/server/src/management/auth.rs
index bc1674651..521f7bbec 100755
--- a/crates/server/src/management/auth.rs
+++ b/crates/server/src/management/auth.rs
@@ -7,9 +7,14 @@
 //! - Admin users: `admin_name:password` (Basic auth)
 //! - IAM users: `account_id/user_name:password` (Basic auth, for self-service)
 
+use std::net::SocketAddr;
+
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use base64::Engine;
+use extenddb_storage::management_store::{AdminStore, OpError};
+
+use crate::rate_limit::LoginRefusal;
 
 /// Maximum allowed length for the Authorization header (8 KB).
 const MAX_AUTH_HEADER_LEN: usize = 8 * 1024;
@@ -17,6 +22,35 @@ const MAX_AUTH_HEADER_LEN: usize = 8 * 1024;
 /// WWW-Authenticate header value for 401 responses (RFC 7235).
 const WWW_AUTHENTICATE_BASIC: &str = "Basic realm=\"extenddb\"";
 
+/// The 401 answered to a wrong password or an unknown principal.
+fn invalid_credentials() -> Response {
+    (
+        StatusCode::UNAUTHORIZED,
+        [(axum::http::header::WWW_AUTHENTICATE, WWW_AUTHENTICATE_BASIC)],
+        "Invalid credentials",
+    )
+        .into_response()
+}
+
+/// Log an error from a password check under `context`. A verification that
+/// found no free bcrypt slot in time is an overload, not a storage error:
+/// [`extenddb_storage::password::verify_password`] has already logged it at
+/// WARN under its own message, so it is not logged again here.
+pub(crate) fn log_password_check_error(context: &str, error: &OpError) {
+    if !extenddb_storage::password::is_overloaded(error) {
+        tracing::error!("{context}: {error:?}");
+    }
+}
+
+/// The answer to a login attempt the rate limiter did not admit.
+fn refusal_response(refusal: LoginRefusal) -> Response {
+    match refusal {
+        LoginRefusal::TooManyAttempts(msg) => (StatusCode::TOO_MANY_REQUESTS, msg).into_response(),
+        LoginRefusal::InvalidCredentials => invalid_credentials(),
+        LoginRefusal::Internal => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
+    }
+}
+
 /// Identifies the authenticated caller of a management API request.
 #[derive(Debug, Clone)]
 pub enum CallerIdentity {
@@ -35,13 +69,20 @@ pub enum CallerIdentity {
 /// tries IAM user auth (`account_id/user_name:password`).
 ///
 /// Returns the caller identity on success, or an error response on failure.
-/// Enforces per-principal lockout and per-IP rate limiting via the storage backend.
+///
+/// `peer` is the address of the connection the request arrived on. Failed
+/// attempts are counted per principal and peer address and per peer address
+/// (see [`crate::rate_limit`]); an attempt over either limit is answered 429
+/// before any password is checked. A wrong password is recorded as a failure;
+/// a storage error while checking it is answered 500 and recorded as nothing.
+/// A principal longer than [`crate::rate_limit::MAX_PRINCIPAL_LEN`] bytes is
+/// answered 401 and recorded as nothing.
 #[allow(clippy::result_large_err)] // axum Response as rejection is this module's idiom
 pub async fn authenticate(
     headers: &HeaderMap,
     store: &dyn extenddb_storage::CatalogStore,
     rate_limiter: &dyn extenddb_storage::management_store::RateLimitStore,
-    source_ip: Option<&str>,
+    peer: SocketAddr,
 ) -> Result<CallerIdentity, Response> {
     let header = headers
         .get("authorization")
@@ -103,42 +144,29 @@ pub async fn authenticate(
             .into_response()
     })?;
 
-    // Rate limit / lockout check before attempting authentication.
-    if let Err(msg) =
-        crate::rate_limit::check_login_allowed(rate_limiter, username, source_ip).await
-    {
-        return Err((StatusCode::TOO_MANY_REQUESTS, msg).into_response());
-    }
+    // Reserve the attempt before any password is checked, so a burst cannot
+    // run more checks than the limits allow.
+    let attempt = crate::rate_limit::begin_login(rate_limiter, username, peer)
+        .await
+        .map_err(refusal_response)?;
 
-    // Try admin auth first.
-    let result = try_admin_auth(username, password, store).await;
-    match result {
-        Ok(Some(identity)) => {
-            return Ok(identity);
+    // Try admin auth first. A storage error returns 500 without recording a
+    // failure: no password was checked, so nothing counts against the caller.
+    match try_admin_auth(username, password, store).await? {
+        AdminAuth::Authenticated(identity) => return Ok(identity),
+        AdminAuth::WrongPassword => {
+            attempt.record_failure(rate_limiter).await;
+            return Err(invalid_credentials());
         }
-        Err(e) => {
-            crate::rate_limit::record_failed_login(rate_limiter, username, source_ip).await;
-            return Err(e);
-        }
-        Ok(None) => {} // Admin not found, try IAM.
+        AdminAuth::NotFound => {} // Admin not found, try IAM.
     }
 
     // Try IAM user auth: username format is `account_id/user_name`.
-    let result = try_iam_user_auth(username, password, store).await;
-    match result {
-        Ok(Some(identity)) => Ok(identity),
-        Err(e) => {
-            crate::rate_limit::record_failed_login(rate_limiter, username, source_ip).await;
-            Err(e)
-        }
-        Ok(None) => {
-            crate::rate_limit::record_failed_login(rate_limiter, username, source_ip).await;
-            Err((
-                StatusCode::UNAUTHORIZED,
-                [(axum::http::header::WWW_AUTHENTICATE, WWW_AUTHENTICATE_BASIC)],
-                "Invalid credentials",
-            )
-                .into_response())
+    match try_iam_user_auth(username, password, store).await? {
+        Some(identity) => Ok(identity),
+        None => {
+            attempt.record_failure(rate_limiter).await;
+            Err(invalid_credentials())
         }
     }
 }
@@ -149,9 +177,9 @@ pub async fn authenticate_admin(
     headers: &HeaderMap,
     store: &dyn extenddb_storage::CatalogStore,
     rate_limiter: &dyn extenddb_storage::management_store::RateLimitStore,
-    source_ip: Option<&str>,
+    peer: SocketAddr,
 ) -> Result<String, Response> {
-    match authenticate(headers, store, rate_limiter, source_ip).await? {
+    match authenticate(headers, store, rate_limiter, peer).await? {
         CallerIdentity::Admin(name) => Ok(name),
         CallerIdentity::IamUser { .. } => {
             Err((StatusCode::FORBIDDEN, "Admin access required").into_response())
@@ -159,35 +187,41 @@ pub async fn authenticate_admin(
     }
 }
 
+/// Outcome of checking a password against the admin store.
+enum AdminAuth {
+    Authenticated(CallerIdentity),
+    /// The admin exists and the password is wrong; do not fall through to IAM.
+    WrongPassword,
+    /// No admin of that name; fall through to IAM user auth.
+    NotFound,
+}
+
+/// Check `password` for admin `username`. `Err` is the 500 answered to a
+/// storage error, including a timed-out wait for a bcrypt slot.
 #[allow(clippy::result_large_err)] // axum Response as rejection is this module's idiom
 async fn try_admin_auth(
     username: &str,
     password: &str,
-    store: &dyn extenddb_storage::CatalogStore,
-) -> Result<Option<CallerIdentity>, Response> {
+    store: &dyn AdminStore,
+) -> Result<AdminAuth, Response> {
     let result = store
         .verify_admin_password(username, password)
         .await
         .map_err(|e| {
-            tracing::error!("Management API: DB error during auth: {e:?}");
+            log_password_check_error("Management API: DB error during auth", &e);
             StatusCode::INTERNAL_SERVER_ERROR.into_response()
         })?;
 
-    match result {
-        Some(true) => Ok(Some(CallerIdentity::Admin(username.to_owned()))),
-        Some(false) => {
-            // Admin exists but password is wrong — don't fall through to IAM auth.
-            Err((
-                StatusCode::UNAUTHORIZED,
-                [(axum::http::header::WWW_AUTHENTICATE, WWW_AUTHENTICATE_BASIC)],
-                "Invalid credentials",
-            )
-                .into_response())
-        }
-        None => Ok(None), // Admin not found, fall through to IAM auth.
-    }
+    Ok(match result {
+        Some(true) => AdminAuth::Authenticated(CallerIdentity::Admin(username.to_owned())),
+        Some(false) => AdminAuth::WrongPassword,
+        None => AdminAuth::NotFound,
+    })
 }
 
+/// Check `password` for IAM user `account_id/user_name`. `Ok(None)` is a
+/// wrong password or an unknown user; `Err` is the 500 answered to a storage
+/// error.
 #[allow(clippy::similar_names)]
 #[allow(clippy::result_large_err)] // axum Response as rejection is this module's idiom
 async fn try_iam_user_auth(
@@ -203,7 +237,7 @@ async fn try_iam_user_auth(
         .verify_iam_user_password(acct_id, uname, password)
         .await
         .map_err(|e| {
-            tracing::error!("Management API: DB error during IAM user auth: {e:?}");
+            log_password_check_error("Management API: DB error during IAM user auth", &e);
             StatusCode::INTERNAL_SERVER_ERROR.into_response()
         })?;
 
@@ -216,3 +250,92 @@ async fn try_iam_user_auth(
         Ok(None)
     }
 }
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use extenddb_storage::management_store::{AdminEntry, OpResult};
+    use extenddb_storage::password::VERIFY_OVERLOADED;
+    use futures::future::BoxFuture;
+
+    use super::*;
+    use crate::rate_limit::PRINCIPAL_LIMITED;
+    use crate::rate_limit::tests::capture_logs;
+
+    /// An admin store whose password check fails with `OpError::Internal`
+    /// carrying `message`. Logins call nothing else on it.
+    pub(crate) struct FailingAdmins {
+        pub(crate) message: &'static str,
+    }
+
+    impl AdminStore for FailingAdmins {
+        fn create_admin(&self, _: &str, _: &str) -> BoxFuture<'_, OpResult<()>> {
+            unreachable!("a login does not create admins")
+        }
+
+        fn list_admins(&self) -> BoxFuture<'_, OpResult<Vec<AdminEntry>>> {
+            unreachable!("a login does not list admins")
+        }
+
+        fn delete_admin(&self, _: &str) -> BoxFuture<'_, OpResult<()>> {
+            unreachable!("a login does not delete admins")
+        }
+
+        fn change_admin_password(&self, _: &str, _: &str) -> BoxFuture<'_, OpResult<()>> {
+            unreachable!("a login does not change passwords")
+        }
+
+        fn verify_admin_password(&self, _: &str, _: &str) -> BoxFuture<'_, OpResult<Option<bool>>> {
+            let error = OpError::Internal(self.message.to_owned());
+            Box::pin(async move { Err(error) })
+        }
+    }
+
+    #[test]
+    fn a_refused_attempt_is_answered_by_its_reason() {
+        let limited = refusal_response(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED));
+        assert_eq!(limited.status(), StatusCode::TOO_MANY_REQUESTS);
+
+        // The attempts store could not be read: a server error, not a limit.
+        let internal = refusal_response(LoginRefusal::Internal);
+        assert_eq!(internal.status(), StatusCode::INTERNAL_SERVER_ERROR);
+
+        // A principal over the length limit gets the wrong-password answer.
+        let too_long = refusal_response(LoginRefusal::InvalidCredentials);
+        assert_eq!(too_long.status(), StatusCode::UNAUTHORIZED);
+        assert_eq!(
+            too_long.headers()[axum::http::header::WWW_AUTHENTICATE],
+            WWW_AUTHENTICATE_BASIC
+        );
+    }
+
+    #[tokio::test]
+    async fn a_store_error_in_the_admin_check_answers_500_and_is_logged_as_one() {
+        let admins = FailingAdmins {
+            message: "store is down",
+        };
+        let (logs, _guard) = capture_logs();
+        let Err(response) = try_admin_auth("admin", "pw", &admins).await else {
+            panic!("a store error is not a login outcome");
+        };
+        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        let text = logs.text();
+        assert!(
+            text.contains("ERROR") && text.contains("Management API: DB error during auth"),
+            "{text}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_bcrypt_slot_timeout_answers_500_and_is_not_logged_as_a_store_error() {
+        let admins = FailingAdmins {
+            message: VERIFY_OVERLOADED,
+        };
+        let (logs, _guard) = capture_logs();
+        let Err(response) = try_admin_auth("admin", "pw", &admins).await else {
+            panic!("an overload is not a login outcome");
+        };
+        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        let text = logs.text();
+        assert!(!text.contains("DB error"), "{text}");
+    }
+}
diff --git a/crates/server/src/management/cache_invalidate.rs b/crates/server/src/management/cache_invalidate.rs
index 5f04d9c20..4759c2742 100644
--- a/crates/server/src/management/cache_invalidate.rs
+++ b/crates/server/src/management/cache_invalidate.rs
@@ -16,7 +16,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::State;
+use axum::extract::{ConnectInfo, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -101,13 +101,14 @@ pub struct InvalidateResponse {
 pub async fn invalidate(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     body: Result<axum::Json<InvalidateRequest>, axum::extract::rejection::JsonRejection>,
 ) -> Response {
     let admin_name = match authenticate_admin(
         &headers,
         &*state.catalog_store,
         &*state.catalog_store,
-        None,
+        addr,
     )
     .await
     {
diff --git a/crates/server/src/management/cache_metrics.rs b/crates/server/src/management/cache_metrics.rs
index 44b948dda..b333cb371 100644
--- a/crates/server/src/management/cache_metrics.rs
+++ b/crates/server/src/management/cache_metrics.rs
@@ -9,7 +9,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::State;
+use axum::extract::{ConnectInfo, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde_json::json;
@@ -25,9 +25,10 @@ use super::auth::authenticate_admin;
 pub async fn auth_cache_metrics(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/iam_group.rs b/crates/server/src/management/iam_group.rs
index 54127e563..1ff916fa1 100755
--- a/crates/server/src/management/iam_group.rs
+++ b/crates/server/src/management/iam_group.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -36,11 +36,12 @@ pub struct AddMemberRequest {
 pub async fn create_group(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
     axum::Json(body): axum::Json<CreateGroupRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -65,10 +66,11 @@ pub async fn create_group(
 pub async fn list_groups(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -101,10 +103,11 @@ pub async fn list_groups(
 pub async fn delete_group(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name)): Path<(String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -152,11 +155,12 @@ pub async fn delete_group(
 pub async fn add_member(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<AddMemberRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -184,10 +188,11 @@ pub async fn add_member(
 pub async fn remove_member(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name, user_name)): Path<(String, String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/iam_policy.rs b/crates/server/src/management/iam_policy.rs
index 7328902f5..8e5de48c6 100755
--- a/crates/server/src/management/iam_policy.rs
+++ b/crates/server/src/management/iam_policy.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::Serialize;
@@ -26,12 +26,14 @@ struct PolicyEntry {
 pub async fn put_user_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name, policy_name)): Path<(String, String, String)>,
     axum::Json(document): axum::Json<Value>,
 ) -> Response {
     put_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "user",
         &user_name,
@@ -44,19 +46,22 @@ pub async fn put_user_policy(
 pub async fn list_user_policies(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
-    list_policies(&state, &headers, &account_id, "user", &user_name).await
+    list_policies(&state, &headers, addr, &account_id, "user", &user_name).await
 }
 
 pub async fn delete_user_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name, policy_name)): Path<(String, String, String)>,
 ) -> Response {
     delete_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "user",
         &user_name,
@@ -68,12 +73,14 @@ pub async fn delete_user_policy(
 pub async fn put_group_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name, policy_name)): Path<(String, String, String)>,
     axum::Json(document): axum::Json<Value>,
 ) -> Response {
     put_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "group",
         &group_name,
@@ -86,19 +93,22 @@ pub async fn put_group_policy(
 pub async fn list_group_policies(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name)): Path<(String, String)>,
 ) -> Response {
-    list_policies(&state, &headers, &account_id, "group", &group_name).await
+    list_policies(&state, &headers, addr, &account_id, "group", &group_name).await
 }
 
 pub async fn delete_group_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, group_name, policy_name)): Path<(String, String, String)>,
 ) -> Response {
     delete_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "group",
         &group_name,
@@ -110,12 +120,14 @@ pub async fn delete_group_policy(
 pub async fn put_role_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name, policy_name)): Path<(String, String, String)>,
     axum::Json(document): axum::Json<Value>,
 ) -> Response {
     put_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "role",
         &role_name,
@@ -128,19 +140,22 @@ pub async fn put_role_policy(
 pub async fn list_role_policies(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
 ) -> Response {
-    list_policies(&state, &headers, &account_id, "role", &role_name).await
+    list_policies(&state, &headers, addr, &account_id, "role", &role_name).await
 }
 
 pub async fn delete_role_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name, policy_name)): Path<(String, String, String)>,
 ) -> Response {
     delete_policy(
         &state,
         &headers,
+        addr,
         &account_id,
         "role",
         &role_name,
@@ -153,9 +168,11 @@ pub async fn delete_role_policy(
 // Shared implementation
 // ---------------------------------------------------------------------------
 
+#[allow(clippy::too_many_arguments)]
 async fn put_policy(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
@@ -163,7 +180,7 @@ async fn put_policy(
     document: &Value,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
@@ -216,12 +233,13 @@ async fn put_policy(
 async fn list_policies(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
@@ -254,13 +272,14 @@ async fn list_policies(
 async fn delete_policy(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
     policy_name: &str,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
diff --git a/crates/server/src/management/iam_role.rs b/crates/server/src/management/iam_role.rs
index a05d131b2..309d1cc5c 100755
--- a/crates/server/src/management/iam_role.rs
+++ b/crates/server/src/management/iam_role.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -51,11 +51,12 @@ pub struct TagEntry {
 pub async fn create_role(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
     axum::Json(body): axum::Json<CreateRoleRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -93,10 +94,11 @@ pub async fn create_role(
 pub async fn list_roles(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -130,10 +132,11 @@ pub async fn list_roles(
 pub async fn delete_role(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -183,11 +186,12 @@ pub async fn delete_role(
 pub async fn tag_role(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<TagRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -241,11 +245,12 @@ pub async fn tag_role(
 pub async fn untag_role(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<UntagRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -276,10 +281,11 @@ pub async fn untag_role(
 pub async fn list_role_tags(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/iam_user.rs b/crates/server/src/management/iam_user.rs
index 8e9a8d583..e93c2187f 100755
--- a/crates/server/src/management/iam_user.rs
+++ b/crates/server/src/management/iam_user.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -50,11 +50,12 @@ pub struct TagEntry {
 pub async fn create_user(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
     axum::Json(body): axum::Json<CreateUserRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -104,10 +105,11 @@ pub async fn create_user(
 pub async fn list_users(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(account_id): Path<String>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -141,10 +143,11 @@ pub async fn list_users(
 pub async fn delete_user(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -221,11 +224,12 @@ pub async fn delete_user(
 pub async fn tag_user(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<TagRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -279,11 +283,12 @@ pub async fn tag_user(
 pub async fn untag_user(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<UntagRequest>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
@@ -314,10 +319,11 @@ pub async fn untag_user(
 pub async fn list_user_tags(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return e;
     }
diff --git a/crates/server/src/management/iam_user_self.rs b/crates/server/src/management/iam_user_self.rs
index 9d7acc9e5..3bac56a0e 100755
--- a/crates/server/src/management/iam_user_self.rs
+++ b/crates/server/src/management/iam_user_self.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -63,10 +63,11 @@ fn authorize_self_service(
 pub async fn create_access_key(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
     let caller =
-        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, None).await {
+        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, addr).await {
             Ok(c) => c,
             Err(e) => return e,
         };
@@ -104,10 +105,11 @@ pub async fn create_access_key(
 pub async fn list_access_keys(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
     let caller =
-        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, None).await {
+        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, addr).await {
             Ok(c) => c,
             Err(e) => return e,
         };
@@ -144,10 +146,11 @@ pub async fn list_access_keys(
 pub async fn delete_access_key(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name, key_id)): Path<(String, String, String)>,
 ) -> Response {
     let caller =
-        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, None).await {
+        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, addr).await {
             Ok(c) => c,
             Err(e) => return e,
         };
@@ -174,11 +177,12 @@ pub async fn delete_access_key(
 pub async fn change_user_password(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<ChangePasswordRequest>,
 ) -> Response {
     let caller =
-        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, None).await {
+        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, addr).await {
             Ok(c) => c,
             Err(e) => return e,
         };
@@ -223,11 +227,12 @@ pub async fn change_user_password(
 pub async fn import_access_key(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
     axum::Json(body): axum::Json<ImportAccessKeyRequest>,
 ) -> Response {
     let caller =
-        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, None).await {
+        match authenticate(&headers, &*state.catalog_store, &*state.catalog_store, addr).await {
             Ok(c) => c,
             Err(e) => return e,
         };
diff --git a/crates/server/src/management/password.rs b/crates/server/src/management/password.rs
index 5dd2da941..b38911c29 100755
--- a/crates/server/src/management/password.rs
+++ b/crates/server/src/management/password.rs
@@ -1,44 +1,33 @@
 // Copyright 2026 ExtendDB contributors
 // SPDX-License-Identifier: Apache-2.0
 
-//! Async-safe password hashing and verification.
+//! Async-safe password hashing.
 //!
 //! bcrypt is CPU-intensive (~100ms per call at default cost). Calling it
-//! directly from an async task blocks the tokio runtime. These helpers use
-//! `spawn_blocking` to move the work to a dedicated thread pool.
-//!
-//! CB-8: A global semaphore limits concurrent bcrypt operations to prevent
-//! denial-of-service via flooding the blocking thread pool (shared with sqlx).
-
-use tokio::sync::Semaphore;
-
-/// Maximum concurrent bcrypt operations. Limits CPU saturation from
-/// concurrent login attempts without starving the sqlx blocking pool.
-static BCRYPT_SEMAPHORE: Semaphore = Semaphore::const_new(4);
+//! directly from an async task blocks the tokio runtime. This helper hashes
+//! through [`extenddb_storage::password::hash_password`], which runs bcrypt on
+//! a blocking thread under a slot of the process-wide bcrypt semaphore it
+//! shares with password verification in every storage backend, so hashing and
+//! verification together never run more than that module's bound at once.
 
 /// Hash a password with bcrypt on a blocking thread.
 ///
 /// Takes `String` (not `&str`) because the closure must own the data
 /// to move it into the `spawn_blocking` thread.
 ///
+/// Waits as long as it takes for a bcrypt slot: the caller is an authenticated
+/// admin creating or changing a password, and a queue is preferable to an
+/// error. The slot is held until bcrypt returns, even if the request is
+/// cancelled first.
+///
 /// # Errors
 ///
 /// Returns `bcrypt::BcryptError` if hashing fails or the blocking task panics.
 pub async fn hash_password(password: String) -> Result<String, bcrypt::BcryptError> {
-    // CB-8: Acquire semaphore permit before spawning blocking work.
-    // `const_new` semaphore is never closed, so `acquire` cannot fail in practice.
-    let Ok(_permit) = BCRYPT_SEMAPHORE.acquire().await else {
-        tracing::error!("bcrypt semaphore closed unexpectedly");
-        return Err(bcrypt::BcryptError::CostNotAllowed(0));
-    };
-    tokio::task::spawn_blocking(move || bcrypt::hash(password, bcrypt::DEFAULT_COST))
+    extenddb_storage::password::hash_password(password, bcrypt::DEFAULT_COST)
         .await
-        .unwrap_or_else(|e| {
-            tracing::error!("bcrypt blocking task failed: {e}");
-            Err(bcrypt::BcryptError::CostNotAllowed(0))
+        .map_err(|e| {
+            tracing::error!("bcrypt hash failed: {e:?}");
+            bcrypt::BcryptError::CostNotAllowed(0)
         })
 }
-
-// Password verification (verify_password) has been migrated to the
-// AdminStore and ManagementStore trait implementations in storage-postgres.
-// The server crate only needs hash_password for creating/updating passwords.
diff --git a/crates/server/src/management/permissions_boundary.rs b/crates/server/src/management/permissions_boundary.rs
index f1e10e9ca..dd34bcad6 100755
--- a/crates/server/src/management/permissions_boundary.rs
+++ b/crates/server/src/management/permissions_boundary.rs
@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use serde_json::Value;
@@ -18,63 +18,88 @@ use super::ops::{OpError, op_err_to_response};
 pub async fn set_user_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
     axum::Json(document): axum::Json<Value>,
 ) -> Response {
-    set_boundary(&state, &headers, &account_id, "user", &user_name, &document).await
+    set_boundary(
+        &state,
+        &headers,
+        addr,
+        &account_id,
+        "user",
+        &user_name,
+        &document,
+    )
+    .await
 }
 
 pub async fn get_user_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
-    get_boundary(&state, &headers, &account_id, "user", &user_name).await
+    get_boundary(&state, &headers, addr, &account_id, "user", &user_name).await
 }
 
 pub async fn delete_user_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, user_name)): Path<(String, String)>,
 ) -> Response {
-    delete_boundary(&state, &headers, &account_id, "user", &user_name).await
+    delete_boundary(&state, &headers, addr, &account_id, "user", &user_name).await
 }
 
 pub async fn set_role_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
     axum::Json(document): axum::Json<Value>,
 ) -> Response {
-    set_boundary(&state, &headers, &account_id, "role", &role_name, &document).await
+    set_boundary(
+        &state,
+        &headers,
+        addr,
+        &account_id,
+        "role",
+        &role_name,
+        &document,
+    )
+    .await
 }
 
 pub async fn get_role_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
 ) -> Response {
-    get_boundary(&state, &headers, &account_id, "role", &role_name).await
+    get_boundary(&state, &headers, addr, &account_id, "role", &role_name).await
 }
 
 pub async fn delete_role_boundary(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path((account_id, role_name)): Path<(String, String)>,
 ) -> Response {
-    delete_boundary(&state, &headers, &account_id, "role", &role_name).await
+    delete_boundary(&state, &headers, addr, &account_id, "role", &role_name).await
 }
 
 async fn set_boundary(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
     document: &Value,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
@@ -140,12 +165,13 @@ async fn set_boundary(
 async fn get_boundary(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
@@ -187,12 +213,13 @@ async fn get_boundary(
 async fn delete_boundary(
     state: &ManagementState,
     headers: &HeaderMap,
+    peer: std::net::SocketAddr,
     account_id: &str,
     principal_type: &str,
     principal_name: &str,
 ) -> Response {
     if let Err(e) =
-        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(headers, &*state.catalog_store, &*state.catalog_store, peer).await
     {
         return e;
     }
diff --git a/crates/server/src/management/settings.rs b/crates/server/src/management/settings.rs
index 3a676abb9..d148367e1 100755
--- a/crates/server/src/management/settings.rs
+++ b/crates/server/src/management/settings.rs
@@ -6,7 +6,7 @@
 use std::sync::Arc;
 
 use axum::Json;
-use axum::extract::{Path, State};
+use axum::extract::{ConnectInfo, Path, State};
 use axum::http::HeaderMap;
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
@@ -33,9 +33,10 @@ fn op_err_to_response(e: extenddb_storage::management_store::OpError) -> Respons
 pub async fn list_settings(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
 ) -> Response {
     if let Err(r) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return r;
     }
@@ -63,10 +64,11 @@ pub async fn list_settings(
 pub async fn get_setting(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(key): Path<String>,
 ) -> Response {
     if let Err(r) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return r;
     }
@@ -98,11 +100,12 @@ pub struct SetSettingRequest {
 pub async fn set_setting(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
+    ConnectInfo(addr): ConnectInfo<std::net::SocketAddr>,
     Path(key): Path<String>,
     Json(body): Json<SetSettingRequest>,
 ) -> Response {
     if let Err(r) =
-        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, None).await
+        authenticate_admin(&headers, &*state.catalog_store, &*state.catalog_store, addr).await
     {
         return r;
     }
diff --git a/crates/server/src/metrics_endpoint.rs b/crates/server/src/metrics_endpoint.rs
index d625c5d92..0ec7b06fd 100755
--- a/crates/server/src/metrics_endpoint.rs
+++ b/crates/server/src/metrics_endpoint.rs
@@ -23,6 +23,7 @@ use crate::AppState;
 pub(crate) async fn metrics_endpoint(
     State(state): State<Arc<AppState>>,
     headers: axum::http::HeaderMap,
+    axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo<std::net::SocketAddr>,
     axum::extract::Query(params): axum::extract::Query<extenddb_core::metrics::MetricsQuery>,
 ) -> axum::response::Response {
     // `/metrics` returns per-table CloudWatch-style dimensions (TableName /
@@ -50,7 +51,7 @@ pub(crate) async fn metrics_endpoint(
         &headers,
         catalog_store.as_ref(),
         catalog_store.as_ref(),
-        None,
+        addr,
     )
     .await
     {
diff --git a/crates/server/src/rate_limit.rs b/crates/server/src/rate_limit.rs
index 6017fe0f8..ca68e0553 100755
--- a/crates/server/src/rate_limit.rs
+++ b/crates/server/src/rate_limit.rs
@@ -1,86 +1,283 @@
 // Copyright 2026 ExtendDB contributors
 // SPDX-License-Identifier: Apache-2.0
 
-//! Login rate limiting and account lockout.
+//! Login rate limiting per user and source address.
 //!
-//! All state lives in the storage backend (e.g. `login_attempts` table in
-//! `PostgreSQL`) so multiple `ExtendDB` instances sharing a catalog see a consistent
-//! view. No in-process caching.
+//! Recorded failures live in the storage backend (the `login_attempts`
+//! table), keyed by principal and by the address the attempt came from, so
+//! every `ExtendDB` instance sharing a catalog sees the same history. Attempts
+//! that have been admitted but not yet finished are counted in a process-wide
+//! registry: an attempt reserves its slot before the password is checked, and
+//! the reservation counts against later attempts until the failure is
+//! recorded. A burst of concurrent wrong passwords therefore runs no more
+//! password checks than the cap allows.
+//!
+//! Limits are applied per source address. Failures from one address never
+//! lock a principal out of another address, so a third party who knows an
+//! admin's name cannot deny that admin the console or the management API.
+//! The source address is the peer address of the TCP connection; behind a
+//! reverse proxy that is the proxy's address for every caller, and `ExtendDB`
+//! does not read `X-Forwarded-For`, so all callers behind one proxy share
+//! one set of limits. An IPv6 peer is counted by its /64 prefix (see
+//! `limiter_source`).
+
+use std::collections::HashMap;
+use std::net::{IpAddr, Ipv6Addr, SocketAddr};
+use std::sync::{LazyLock, Mutex, MutexGuard, PoisonError};
 
 use extenddb_storage::management_store::RateLimitStore;
 
-/// Maximum failed login attempts per principal in the lookback window before
-/// the account is locked out.
-const MAX_FAILURES_PER_PRINCIPAL: i64 = 5;
+/// Maximum failed attempts for one principal from one source address in the
+/// window. Once reached, that address is refused for that principal until the
+/// oldest failure leaves the window.
+pub const MAX_FAILURES_PER_PRINCIPAL: i64 = 5;
 
-/// Maximum failed login attempts per source IP in the lookback window.
-const MAX_FAILURES_PER_IP: i64 = 20;
+/// Maximum failed attempts from one source address in the window, over all
+/// principals.
+pub const MAX_FAILURES_PER_IP: i64 = 20;
 
 /// Lookback window in seconds for counting failed attempts.
-const WINDOW_SECONDS: i64 = 900; // 15 minutes
+pub const WINDOW_SECONDS: i64 = 900; // 15 minutes
+
+/// Longest principal, in bytes, that a login attempt may name. Admin names
+/// are at most 64 bytes and `account_id/user_name` at most 141, so a longer
+/// principal cannot name anyone and is refused before the store is read.
+pub const MAX_PRINCIPAL_LEN: usize = 256;
+
+/// Message returned when the (principal, source) pair is over its limit.
+pub const PRINCIPAL_LIMITED: &str = "Too many failed login attempts. Please try again later.";
 
-/// Check whether a login attempt should be blocked due to rate limiting or
-/// account lockout. Returns `Ok(())` if the attempt may proceed, or
-/// `Err(message)` if it should be rejected.
+/// Message returned when the source address is over its limit.
+pub const SOURCE_LIMITED: &str =
+    "Too many failed login attempts from this address. Please try again later.";
+
+/// Why a login attempt was not admitted.
+#[derive(Debug, PartialEq, Eq)]
+pub enum LoginRefusal {
+    /// The principal from this source, or this source overall, has too many
+    /// recent failures. Callers answer 429 with the message.
+    TooManyAttempts(&'static str),
+    /// The attempts store could not be read. Callers answer 500; nothing is
+    /// recorded, and no password is checked.
+    Internal,
+    /// The principal is longer than [`MAX_PRINCIPAL_LEN`] bytes. Callers
+    /// answer as for a wrong password; nothing is read, reserved or recorded.
+    InvalidCredentials,
+}
+
+/// An admitted login attempt. It holds one reservation against the limits of
+/// its principal and source until it is recorded as a failure or dropped.
+///
+/// Drop it after a successful login, or when the password could not be
+/// checked, to release the reservation without recording anything.
+#[must_use = "the attempt holds a reservation until it is recorded or dropped"]
+pub struct LoginAttempt {
+    principal: String,
+    source_ip: String,
+}
+
+impl LoginAttempt {
+    /// The source key the attempt is counted under: an IPv4 address, or an
+    /// IPv6 /64 prefix.
+    #[must_use]
+    pub fn source_ip(&self) -> &str {
+        &self.source_ip
+    }
+
+    /// Record the attempt as a failed login, then release its reservation.
+    ///
+    /// The row is written before the reservation is released, so an admission
+    /// check running at the same time sees the attempt in one or the other.
+    ///
+    /// The write is fail-open: if the store cannot record the failure, the
+    /// error is logged and the limits are not tightened, which keeps logins
+    /// available during a storage outage at the cost of not counting failures
+    /// while storage is unreachable.
+    pub async fn record_failure(self, store: &dyn RateLimitStore) {
+        store
+            .record_failed_login(&self.principal, Some(&self.source_ip))
+            .await;
+        // `self` drops here, after the row is written.
+    }
+}
+
+impl Drop for LoginAttempt {
+    fn drop(&mut self) {
+        registry().release(&self.source_ip, &self.principal);
+    }
+}
+
+/// Admit a login attempt for `principal` from `peer`, or refuse it.
+///
+/// The attempt is admitted when the failures recorded for the pair in the
+/// window plus the attempts in flight for the pair (this one included) do not
+/// exceed [`MAX_FAILURES_PER_PRINCIPAL`], and the same holds for the source
+/// against [`MAX_FAILURES_PER_IP`].
 ///
 /// # Errors
 ///
-/// Returns `Err(String)` when the principal is locked out, the source IP
-/// exceeds the rate limit, or a database error occurs during the check.
-pub async fn check_login_allowed(
+/// [`LoginRefusal::InvalidCredentials`] when the principal is longer than
+/// [`MAX_PRINCIPAL_LEN`] bytes, [`LoginRefusal::TooManyAttempts`] when a
+/// limit is reached and [`LoginRefusal::Internal`] when the store cannot be
+/// read.
+pub async fn begin_login(
     store: &dyn RateLimitStore,
     principal: &str,
-    source_ip: Option<&str>,
-) -> Result<(), String> {
-    // Check per-principal lockout.
-    let principal_failures = store
-        .count_principal_failures(principal, WINDOW_SECONDS)
+    peer: SocketAddr,
+) -> Result<LoginAttempt, LoginRefusal> {
+    if principal.len() > MAX_PRINCIPAL_LEN {
+        return Err(LoginRefusal::InvalidCredentials);
+    }
+    begin_login_from(store, principal, limiter_source(&peer)).await
+}
+
+/// The source key a peer's login attempts are counted under.
+///
+/// An IPv4 peer, and an IPv6 peer carrying an IPv4-mapped address, count
+/// under the IPv4 address. Any other IPv6 peer counts under its /64 prefix
+/// (`2001:db8:1:2::/64`): a /64 is the block usually assigned to one host or
+/// site, and a client can send from any address in it, so keying on the full
+/// address would give it a fresh set of limits for every address it picks.
+///
+/// This applies to the limiter only. `aws:SourceIp` still sees the full
+/// address through [`crate::authorization::condition_source_ip`].
+fn limiter_source(peer: &SocketAddr) -> String {
+    match peer.ip().to_canonical() {
+        IpAddr::V6(v6) => {
+            let prefix = Ipv6Addr::from(u128::from(v6) & !(u128::from(u64::MAX)));
+            format!("{prefix}/64")
+        }
+        v4 @ IpAddr::V4(_) => v4.to_string(),
+    }
+}
+
+async fn begin_login_from(
+    store: &dyn RateLimitStore,
+    principal: &str,
+    source_ip: String,
+) -> Result<LoginAttempt, LoginRefusal> {
+    // Reserve first, then count. The numbers returned include this attempt
+    // and every other attempt in flight for the pair and the source at this
+    // moment. An attempt admitted earlier is still in flight (so it is in the
+    // numbers) or has written its row before releasing (so it is in the
+    // count). One that is in both is counted twice, which only refuses
+    // sooner. The reservation is released by `Drop` if the attempt is
+    // refused.
+    let (pair_in_flight, source_in_flight) = registry().reserve(&source_ip, principal);
+    let attempt = LoginAttempt {
+        principal: principal.to_owned(),
+        source_ip,
+    };
+
+    let recorded_for_pair = store
+        .count_principal_failures(&attempt.principal, &attempt.source_ip, WINDOW_SECONDS)
         .await
         .map_err(|e| {
             tracing::error!("Rate limit check failed: {e:?}");
-            "Internal error".to_owned()
+            LoginRefusal::Internal
         })?;
+    if recorded_for_pair + pair_in_flight > MAX_FAILURES_PER_PRINCIPAL {
+        // The principal is the caller's text, at most MAX_PRINCIPAL_LEN bytes
+        // here. `{:?}` escapes newlines and control characters, so it cannot
+        // start a new log line.
+        tracing::warn!(
+            "Login refused: principal {:?} from {} has {recorded_for_pair} failed attempts \
+             and {} in flight in window",
+            attempt.principal,
+            attempt.source_ip,
+            pair_in_flight - 1,
+        );
+        return Err(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED));
+    }
 
-    if principal_failures >= MAX_FAILURES_PER_PRINCIPAL {
+    let recorded_for_source = store
+        .count_ip_failures(&attempt.source_ip, WINDOW_SECONDS)
+        .await
+        .map_err(|e| {
+            tracing::error!("Rate limit check failed: {e:?}");
+            LoginRefusal::Internal
+        })?;
+    if recorded_for_source + source_in_flight > MAX_FAILURES_PER_IP {
         tracing::warn!(
-            "Account lockout: principal {principal} has {principal_failures} failed attempts in window",
+            "IP rate limit: {} has {recorded_for_source} failed attempts and {} in flight in window",
+            attempt.source_ip,
+            source_in_flight - 1,
         );
-        return Err("Too many failed login attempts. Please try again later.".to_owned());
+        return Err(LoginRefusal::TooManyAttempts(SOURCE_LIMITED));
+    }
+
+    Ok(attempt)
+}
+
+// ── In-flight registry ─────────────────────────────────────────────────
+
+#[derive(Default)]
+struct SourceInFlight {
+    total: i64,
+    by_principal: HashMap<String, i64>,
+}
+
+/// Attempts admitted and not yet recorded or dropped, by source address and
+/// principal. An entry exists only while something is in flight for it.
+#[derive(Default)]
+struct InFlightRegistry {
+    sources: Mutex<HashMap<String, SourceInFlight>>,
+}
+
+impl InFlightRegistry {
+    fn lock(&self) -> MutexGuard<'_, HashMap<String, SourceInFlight>> {
+        self.sources.lock().unwrap_or_else(PoisonError::into_inner)
     }
 
-    // Check per-IP rate limit.
-    if let Some(ip) = source_ip {
-        let ip_failures = store
-            .count_ip_failures(ip, WINDOW_SECONDS)
-            .await
-            .map_err(|e| {
-                tracing::error!("Rate limit check failed: {e:?}");
-                "Internal error".to_owned()
-            })?;
-
-        if ip_failures >= MAX_FAILURES_PER_IP {
-            tracing::warn!("IP rate limit: {ip} has {ip_failures} failed attempts in window",);
-            return Err(
-                "Too many failed login attempts from this address. Please try again later."
-                    .to_owned(),
-            );
+    /// Count one more attempt in flight for the pair and the source. Returns
+    /// the pair's and the source's in-flight counts including this attempt.
+    fn reserve(&self, source_ip: &str, principal: &str) -> (i64, i64) {
+        let mut sources = self.lock();
+        let source = sources.entry(source_ip.to_owned()).or_default();
+        source.total += 1;
+        let pair = source.by_principal.entry(principal.to_owned()).or_default();
+        *pair += 1;
+        (*pair, source.total)
+    }
+
+    fn release(&self, source_ip: &str, principal: &str) {
+        let mut sources = self.lock();
+        let Some(source) = sources.get_mut(source_ip) else {
+            return;
+        };
+        if let Some(pair) = source.by_principal.get_mut(principal) {
+            *pair -= 1;
+            if *pair <= 0 {
+                source.by_principal.remove(principal);
+            }
+        }
+        source.total -= 1;
+        if source.total <= 0 {
+            sources.remove(source_ip);
         }
     }
 
-    Ok(())
+    /// In-flight attempts for the pair, for tests.
+    #[cfg(test)]
+    fn in_flight(&self, source_ip: &str, principal: &str) -> i64 {
+        self.lock()
+            .get(source_ip)
+            .and_then(|s| s.by_principal.get(principal))
+            .copied()
+            .unwrap_or(0)
+    }
+
+    /// Whether the source has any attempt in flight, for tests.
+    #[cfg(test)]
+    fn tracks(&self, source_ip: &str) -> bool {
+        self.lock().contains_key(source_ip)
+    }
 }
 
-/// Record a failed login attempt.
-///
-/// **Design decision (fail-open):** If the storage write fails, the attempt
-/// is logged but not rejected. This preserves availability during storage outages
-/// at the cost of allowing an attacker to bypass rate limiting while storage is
-/// unreachable. The per-principal lockout check in `check_login_allowed` already
-/// fails-closed (storage error → reject).
-pub async fn record_failed_login(
-    store: &dyn RateLimitStore,
-    principal: &str,
-    source_ip: Option<&str>,
-) {
-    store.record_failed_login(principal, source_ip).await;
+fn registry() -> &'static InFlightRegistry {
+    static REGISTRY: LazyLock<InFlightRegistry> = LazyLock::new(InFlightRegistry::default);
+    &REGISTRY
 }
+
+#[cfg(test)]
+pub(crate) mod tests;
diff --git a/crates/server/src/rate_limit/tests.rs b/crates/server/src/rate_limit/tests.rs
new file mode 100644
index 000000000..7e6132a09
--- /dev/null
+++ b/crates/server/src/rate_limit/tests.rs
@@ -0,0 +1,411 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Unit tests for login rate limiting, against an attempts store in memory.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+
+use extenddb_storage::management_store::{OpError, OpResult};
+use futures::future::BoxFuture;
+use tokio::sync::{Notify, Semaphore};
+
+use super::*;
+
+/// An attempts store in memory: recorded failures as (principal, source)
+/// rows. `delay` slows the count queries so concurrent attempts overlap,
+/// and `broken` makes every read fail. With `write_gate` set, a failure
+/// write announces itself on `write_started` and then waits for a permit
+/// before the row lands.
+#[derive(Default)]
+pub(crate) struct MemoryStore {
+    rows: Mutex<Vec<(String, Option<String>)>>,
+    delay: Option<Duration>,
+    broken: AtomicBool,
+    write_gate: Option<Semaphore>,
+    write_started: Notify,
+}
+
+impl MemoryStore {
+    pub(crate) fn rows_for(&self, principal: &str, source_ip: &str) -> i64 {
+        self.rows
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(p, s)| p == principal && s.as_deref() == Some(source_ip))
+            .count() as i64
+    }
+
+    pub(crate) fn row_count(&self) -> usize {
+        self.rows.lock().unwrap().len()
+    }
+
+    async fn read(&self) -> OpResult<()> {
+        if let Some(delay) = self.delay {
+            tokio::time::sleep(delay).await;
+        }
+        if self.broken.load(Ordering::SeqCst) {
+            return Err(OpError::Internal("store is down".to_owned()));
+        }
+        Ok(())
+    }
+}
+
+impl RateLimitStore for MemoryStore {
+    fn count_principal_failures(
+        &self,
+        principal: &str,
+        source_ip: &str,
+        _window_seconds: i64,
+    ) -> BoxFuture<'_, OpResult<i64>> {
+        let principal = principal.to_owned();
+        let source_ip = source_ip.to_owned();
+        Box::pin(async move {
+            self.read().await?;
+            Ok(self.rows_for(&principal, &source_ip))
+        })
+    }
+
+    fn count_ip_failures(
+        &self,
+        source_ip: &str,
+        _window_seconds: i64,
+    ) -> BoxFuture<'_, OpResult<i64>> {
+        let source_ip = source_ip.to_owned();
+        Box::pin(async move {
+            self.read().await?;
+            let rows = self.rows.lock().unwrap();
+            Ok(rows
+                .iter()
+                .filter(|(_, s)| s.as_deref() == Some(source_ip.as_str()))
+                .count() as i64)
+        })
+    }
+
+    fn record_failed_login(&self, principal: &str, source_ip: Option<&str>) -> BoxFuture<'_, ()> {
+        let row = (principal.to_owned(), source_ip.map(str::to_owned));
+        Box::pin(async move {
+            if let Some(gate) = &self.write_gate {
+                self.write_started.notify_one();
+                let _permit = gate.acquire().await.unwrap();
+                self.rows.lock().unwrap().push(row);
+                return;
+            }
+            self.rows.lock().unwrap().push(row);
+        })
+    }
+
+    fn cleanup_old_attempts(&self, _max_age_seconds: i64) -> BoxFuture<'_, ()> {
+        Box::pin(async {})
+    }
+}
+
+pub(crate) fn peer(ip: &str) -> SocketAddr {
+    format!("{ip}:40000").parse().unwrap()
+}
+
+/// Log output collected by [`capture_logs`].
+#[derive(Clone, Default)]
+pub(crate) struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
+
+impl CapturedLogs {
+    pub(crate) fn text(&self) -> String {
+        String::from_utf8_lossy(&self.0.lock().unwrap()).into_owned()
+    }
+}
+
+impl std::io::Write for CapturedLogs {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.0.lock().unwrap().extend_from_slice(buf);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+/// Collect this thread's log output until the guard drops. Use it from a
+/// test on the default current-thread runtime, where the code under test
+/// logs on the test's own thread.
+pub(crate) fn capture_logs() -> (CapturedLogs, tracing::subscriber::DefaultGuard) {
+    let logs = CapturedLogs::default();
+    let writer = logs.clone();
+    let subscriber = tracing_subscriber::fmt()
+        .with_writer(move || writer.clone())
+        .with_ansi(false)
+        .finish();
+    (logs, tracing::subscriber::set_default(subscriber))
+}
+
+async fn fail_times(store: &MemoryStore, principal: &str, ip: &str, times: usize) {
+    for _ in 0..times {
+        let attempt = begin_login(store, principal, peer(ip)).await.unwrap();
+        attempt.record_failure(store).await;
+    }
+}
+
+// Every test uses its own addresses: the registry is process-wide.
+
+#[tokio::test]
+async fn failures_from_one_address_do_not_lock_the_principal_out_of_another() {
+    let store = MemoryStore::default();
+    fail_times(&store, "admin", "198.51.100.1", 5).await;
+    assert_eq!(store.rows_for("admin", "198.51.100.1"), 5);
+
+    // The limited address is refused for this principal.
+    assert_eq!(
+        begin_login(&store, "admin", peer("198.51.100.1"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+    // Another address is admitted, and a success leaves no row.
+    let attempt = begin_login(&store, "admin", peer("198.51.100.2"))
+        .await
+        .expect("another address is not affected");
+    drop(attempt);
+    assert_eq!(store.rows_for("admin", "198.51.100.2"), 0);
+    // The limited address may still log in as someone else.
+    let _other = begin_login(&store, "other", peer("198.51.100.1"))
+        .await
+        .expect("the limit is per principal and source");
+}
+
+#[tokio::test]
+async fn a_source_address_is_limited_across_principals() {
+    let store = MemoryStore::default();
+    for n in 0..MAX_FAILURES_PER_IP {
+        fail_times(&store, &format!("user{n}"), "198.51.100.20", 1).await;
+    }
+    assert_eq!(
+        begin_login(&store, "user-new", peer("198.51.100.20"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(SOURCE_LIMITED))
+    );
+    let _elsewhere = begin_login(&store, "user-new", peer("198.51.100.21"))
+        .await
+        .expect("another address is not affected");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn a_concurrent_burst_admits_no_more_than_the_cap() {
+    let store = Arc::new(MemoryStore {
+        delay: Some(Duration::from_millis(50)),
+        ..MemoryStore::default()
+    });
+    let mut tasks = Vec::new();
+    for _ in 0..40 {
+        let store = Arc::clone(&store);
+        tasks.push(tokio::spawn(async move {
+            begin_login(&*store, "admin", peer("198.51.100.30")).await
+        }));
+    }
+    let mut admitted = Vec::new();
+    for task in tasks {
+        if let Ok(attempt) = task.await.unwrap() {
+            admitted.push(attempt);
+        }
+    }
+    // Admitted attempts are still in flight, so exactly the cap got in.
+    assert_eq!(admitted.len() as i64, MAX_FAILURES_PER_PRINCIPAL);
+    assert_eq!(
+        registry().in_flight("198.51.100.30", "admin"),
+        MAX_FAILURES_PER_PRINCIPAL
+    );
+    assert_eq!(store.row_count(), 0, "admission records nothing");
+
+    // Once they are recorded, the address is out for this principal.
+    for attempt in admitted {
+        attempt.record_failure(&*store).await;
+    }
+    assert_eq!(store.rows_for("admin", "198.51.100.30"), 5);
+    assert!(!registry().tracks("198.51.100.30"));
+    assert_eq!(
+        begin_login(&*store, "admin", peer("198.51.100.30"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_admission_while_a_failure_is_being_written_still_counts_it() {
+    let store = Arc::new(MemoryStore {
+        write_gate: Some(Semaphore::new(0)),
+        ..MemoryStore::default()
+    });
+    let source = "198.51.100.70";
+    let mut admitted = Vec::new();
+    for _ in 0..MAX_FAILURES_PER_PRINCIPAL {
+        admitted.push(begin_login(&*store, "admin", peer(source)).await.unwrap());
+    }
+
+    // Record one failure and hold its row write open.
+    let first = admitted.pop().unwrap();
+    let writer = {
+        let store = Arc::clone(&store);
+        tokio::spawn(async move { first.record_failure(&*store).await })
+    };
+    store.write_started.notified().await;
+
+    // The failure has neither landed as a row nor finished. It must still
+    // count against the pair, so the pair stays at its cap.
+    assert_eq!(
+        begin_login(&*store, "admin", peer(source)).await.err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+
+    store.write_gate.as_ref().unwrap().add_permits(8);
+    writer.await.unwrap();
+    for attempt in admitted {
+        attempt.record_failure(&*store).await;
+    }
+    assert_eq!(store.rows_for("admin", source), MAX_FAILURES_PER_PRINCIPAL);
+    assert!(!registry().tracks(source));
+}
+
+#[tokio::test]
+async fn a_dropped_attempt_releases_its_reservation_without_a_row() {
+    let store = MemoryStore::default();
+    let attempt = begin_login(&store, "admin", peer("198.51.100.40"))
+        .await
+        .unwrap();
+    assert_eq!(registry().in_flight("198.51.100.40", "admin"), 1);
+    drop(attempt);
+    assert!(!registry().tracks("198.51.100.40"));
+    assert_eq!(store.row_count(), 0);
+}
+
+#[tokio::test]
+async fn a_store_error_is_internal_and_holds_no_reservation() {
+    let store = MemoryStore::default();
+    store.broken.store(true, Ordering::SeqCst);
+    assert_eq!(
+        begin_login(&store, "admin", peer("198.51.100.50"))
+            .await
+            .err(),
+        Some(LoginRefusal::Internal)
+    );
+    assert!(!registry().tracks("198.51.100.50"));
+}
+
+#[tokio::test]
+async fn the_refusal_log_line_escapes_the_principal() {
+    let store = MemoryStore::default();
+    let principal = "admin\nERROR forged line";
+    fail_times(&store, principal, "198.51.100.85", 5).await;
+
+    let (logs, _guard) = capture_logs();
+    assert_eq!(
+        begin_login(&store, principal, peer("198.51.100.85"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+    let text = logs.text();
+    assert!(
+        text.contains(r#"principal "admin\nERROR forged line" from 198.51.100.85"#),
+        "{text}"
+    );
+    assert!(!text.contains("\nERROR forged line"), "{text}");
+}
+
+#[tokio::test]
+async fn a_principal_over_the_length_limit_is_refused_before_the_store_is_read() {
+    let store = MemoryStore::default();
+    // Every read fails, so an attempt that reads the store is Internal.
+    store.broken.store(true, Ordering::SeqCst);
+    let too_long = "a".repeat(MAX_PRINCIPAL_LEN + 1);
+    assert_eq!(
+        begin_login(&store, &too_long, peer("198.51.100.80"))
+            .await
+            .err(),
+        Some(LoginRefusal::InvalidCredentials)
+    );
+    assert!(!registry().tracks("198.51.100.80"));
+
+    // A principal at the limit is checked against the store.
+    let at_limit = "a".repeat(MAX_PRINCIPAL_LEN);
+    assert_eq!(
+        begin_login(&store, &at_limit, peer("198.51.100.80"))
+            .await
+            .err(),
+        Some(LoginRefusal::Internal)
+    );
+}
+
+#[test]
+fn an_ipv6_peer_is_keyed_by_its_64_and_ipv4_by_its_address() {
+    assert_eq!(
+        limiter_source(&peer("[2001:db8:1:2:aaaa:bbbb:cccc:dddd]")),
+        "2001:db8:1:2::/64"
+    );
+    assert_eq!(
+        limiter_source(&peer("[::ffff:198.51.100.7]")),
+        "198.51.100.7"
+    );
+    assert_eq!(limiter_source(&peer("198.51.100.7")), "198.51.100.7");
+}
+
+#[tokio::test]
+async fn addresses_in_one_ipv6_64_share_the_pair_limit_and_another_64_does_not() {
+    let store = MemoryStore::default();
+    for n in 1..=MAX_FAILURES_PER_PRINCIPAL {
+        let attempt = begin_login(&store, "admin", peer(&format!("[2001:db8:64:1::{n:x}]")))
+            .await
+            .unwrap();
+        attempt.record_failure(&store).await;
+    }
+    assert_eq!(store.rows_for("admin", "2001:db8:64:1::/64"), 5);
+
+    // A sixth address in the same /64 is refused for the principal.
+    assert_eq!(
+        begin_login(&store, "admin", peer("[2001:db8:64:1:ffff::99]"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+    // The neighbouring /64 is a different source.
+    let _other = begin_login(&store, "admin", peer("[2001:db8:64:2::1]"))
+        .await
+        .expect("another /64 is not affected");
+}
+
+#[tokio::test]
+async fn addresses_in_one_ipv6_64_share_the_source_limit_and_another_64_does_not() {
+    let store = MemoryStore::default();
+    for n in 1..=MAX_FAILURES_PER_IP {
+        let attempt = begin_login(
+            &store,
+            &format!("user{n}"),
+            peer(&format!("[2001:db8:64:3::{n:x}]")),
+        )
+        .await
+        .unwrap();
+        attempt.record_failure(&store).await;
+    }
+    assert_eq!(
+        begin_login(&store, "user-new", peer("[2001:db8:64:3::abcd]"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(SOURCE_LIMITED))
+    );
+    let _other = begin_login(&store, "user-new", peer("[2001:db8:64:4::abcd]"))
+        .await
+        .expect("another /64 is not affected");
+}
+
+#[tokio::test]
+async fn an_ipv4_mapped_peer_counts_as_its_ipv4_address() {
+    let store = MemoryStore::default();
+    fail_times(&store, "admin", "198.51.100.60", 5).await;
+    assert_eq!(
+        begin_login(&store, "admin", peer("[::ffff:198.51.100.60]"))
+            .await
+            .err(),
+        Some(LoginRefusal::TooManyAttempts(PRINCIPAL_LIMITED))
+    );
+}
diff --git a/crates/server/src/workers.rs b/crates/server/src/workers.rs
index f519f91c3..58e3a7133 100755
--- a/crates/server/src/workers.rs
+++ b/crates/server/src/workers.rs
@@ -341,6 +341,7 @@ mod tests {
         fn count_principal_failures(
             &self,
             _principal: &str,
+            _source_ip: &str,
             _window_seconds: i64,
         ) -> BoxFuture<'_, OpResult<i64>> {
             Box::pin(async { Ok(0) })
diff --git a/crates/storage-cassandra/src/admin_store.rs b/crates/storage-cassandra/src/admin_store.rs
index bf32e3458..a7e2d599a 100755
--- a/crates/storage-cassandra/src/admin_store.rs
+++ b/crates/storage-cassandra/src/admin_store.rs
@@ -184,14 +184,8 @@ impl extenddb_storage::management_store::AdminStore for CassandraCatalogStore {
             let hash: String =
                 crate::cassandra_util::get_column(&row, "password_hash", "verify_admin_password")?;
 
-            Ok(Some(verify_bcrypt(password, hash).await))
+            let verified = extenddb_storage::password::verify_password(password, hash).await?;
+            Ok(Some(verified))
         })
     }
 }
-
-/// Verify a bcrypt password on a blocking thread (same logic as `server::password`).
-async fn verify_bcrypt(password: String, hash: String) -> bool {
-    tokio::task::spawn_blocking(move || bcrypt::verify(password, &hash).unwrap_or(false))
-        .await
-        .unwrap_or(false)
-}
diff --git a/crates/storage-cassandra/src/catalog_store.rs b/crates/storage-cassandra/src/catalog_store.rs
index e0adc3f6f..0e65c43cd 100644
--- a/crates/storage-cassandra/src/catalog_store.rs
+++ b/crates/storage-cassandra/src/catalog_store.rs
@@ -406,9 +406,11 @@ impl RateLimitStore for CassandraCatalogStore {
     fn count_principal_failures(
         &self,
         principal: &str,
+        source_ip: &str,
         window_seconds: i64,
     ) -> BoxFuture<'_, OpResult<i64>> {
         let principal = principal.to_owned();
+        let source_ip = source_ip.to_owned();
         let session = self.session.clone();
         let catalog_keyspace = self.catalog_keyspace();
         Box::pin(async move {
@@ -417,12 +419,12 @@ impl RateLimitStore for CassandraCatalogStore {
             let query = format!(
                 "SELECT COUNT(*) FROM {catalog_keyspace}.login_attempts \
                  WHERE principal = ? AND attempted_at > ? AND success = false \
-                 ALLOW FILTERING"
+                 AND source_ip = ? ALLOW FILTERING"
             );
             let result = session
                 .query_with_values(
                     &query,
-                    cdrs_tokio::query_values!(principal.as_str(), cutoff_ms),
+                    cdrs_tokio::query_values!(principal.as_str(), cutoff_ms, source_ip.as_str()),
                 )
                 .await
                 .map_err(|e| {
diff --git a/crates/storage-cassandra/src/management_store/users.rs b/crates/storage-cassandra/src/management_store/users.rs
index 3ed0a895f..d076aef40 100644
--- a/crates/storage-cassandra/src/management_store/users.rs
+++ b/crates/storage-cassandra/src/management_store/users.rs
@@ -401,12 +401,7 @@ impl CassandraCatalogStore {
             return Ok(false);
         };
 
-        let pw = password.to_owned();
-        Ok(
-            tokio::task::spawn_blocking(move || bcrypt::verify(pw, &hash).unwrap_or(false))
-                .await
-                .unwrap_or(false),
-        )
+        extenddb_storage::password::verify_password(password.to_owned(), hash).await
     }
 
     pub(crate) async fn change_user_password_impl(
diff --git a/crates/storage-mongodb/src/management_store.rs b/crates/storage-mongodb/src/management_store.rs
index 91e60e2b9..a55ab7ead 100644
--- a/crates/storage-mongodb/src/management_store.rs
+++ b/crates/storage-mongodb/src/management_store.rs
@@ -642,11 +642,7 @@ impl ManagementStore for MongoCatalogStore {
             };
 
             let hash = hash.to_owned();
-            Ok(tokio::task::spawn_blocking(move || {
-                bcrypt::verify(password, &hash).unwrap_or(false)
-            })
-            .await
-            .unwrap_or(false))
+            extenddb_storage::password::verify_password(password, hash).await
         })
     }
 
@@ -2205,9 +2201,11 @@ impl extenddb_storage::management_store::RateLimitStore for MongoCatalogStore {
     fn count_principal_failures(
         &self,
         principal: &str,
+        source_ip: &str,
         window_seconds: i64,
     ) -> BoxFuture<'_, OpResult<i64>> {
         let principal = principal.to_owned();
+        let source_ip = source_ip.to_owned();
         Box::pin(async move {
             let coll = self.catalog_db().collection::<Document>("failed_logins");
             let cutoff = time::OffsetDateTime::now_utc()
@@ -2216,6 +2214,7 @@ impl extenddb_storage::management_store::RateLimitStore for MongoCatalogStore {
             let count = coll
                 .count_documents(doc! {
                     "principal": &principal,
+                    "source_ip": &source_ip,
                     "attempted_at": { "$gte": cutoff_bson },
                 })
                 .await
@@ -2405,13 +2404,8 @@ impl extenddb_storage::management_store::AdminStore for MongoCatalogStore {
                 return Ok(None);
             };
             let hash = hash.to_owned();
-            Ok(Some(
-                tokio::task::spawn_blocking(move || {
-                    bcrypt::verify(password, &hash).unwrap_or(false)
-                })
-                .await
-                .unwrap_or(false),
-            ))
+            let verified = extenddb_storage::password::verify_password(password, hash).await?;
+            Ok(Some(verified))
         })
     }
 }
diff --git a/crates/storage-postgres/src/admin_store.rs b/crates/storage-postgres/src/admin_store.rs
index 9edaeff04..30513732e 100755
--- a/crates/storage-postgres/src/admin_store.rs
+++ b/crates/storage-postgres/src/admin_store.rs
@@ -121,14 +121,8 @@ impl extenddb_storage::management_store::AdminStore for PostgresCatalogStore {
             let Some((hash,)) = row else {
                 return Ok(None);
             };
-            Ok(Some(verify_bcrypt(password, hash).await))
+            let verified = extenddb_storage::password::verify_password(password, hash).await?;
+            Ok(Some(verified))
         })
     }
 }
-
-/// Verify a bcrypt password on a blocking thread (same logic as `server::password`).
-async fn verify_bcrypt(password: String, hash: String) -> bool {
-    tokio::task::spawn_blocking(move || bcrypt::verify(password, &hash).unwrap_or(false))
-        .await
-        .unwrap_or(false)
-}
diff --git a/crates/storage-postgres/src/catalog_store.rs b/crates/storage-postgres/src/catalog_store.rs
index d246ddf83..834c88a5a 100755
--- a/crates/storage-postgres/src/catalog_store.rs
+++ b/crates/storage-postgres/src/catalog_store.rs
@@ -344,16 +344,19 @@ impl extenddb_storage::management_store::RateLimitStore for PostgresCatalogStore
     fn count_principal_failures(
         &self,
         principal: &str,
+        source_ip: &str,
         window_seconds: i64,
     ) -> BoxFuture<'_, OpResult<i64>> {
         let principal = principal.to_owned();
+        let source_ip = source_ip.to_owned();
         Box::pin(async move {
             let row: (i64,) = sqlx::query_as(
                 "SELECT COUNT(*) FROM login_attempts \
-                 WHERE principal = $1 AND success = false \
-                 AND attempted_at > NOW() - make_interval(secs => $2)",
+                 WHERE principal = $1 AND source_ip = $2 AND success = false \
+                 AND attempted_at > NOW() - make_interval(secs => $3)",
             )
             .bind(&principal)
+            .bind(&source_ip)
             .bind(window_seconds)
             .fetch_one(&self.pool)
             .await
diff --git a/crates/storage-postgres/src/management_store/users.rs b/crates/storage-postgres/src/management_store/users.rs
index 245ad77ae..fa5dae6e3 100755
--- a/crates/storage-postgres/src/management_store/users.rs
+++ b/crates/storage-postgres/src/management_store/users.rs
@@ -229,12 +229,7 @@ impl PostgresCatalogStore {
             return Ok(false);
         };
 
-        let pw = password.to_owned();
-        Ok(
-            tokio::task::spawn_blocking(move || bcrypt::verify(pw, &hash).unwrap_or(false))
-                .await
-                .unwrap_or(false),
-        )
+        extenddb_storage::password::verify_password(password.to_owned(), hash).await
     }
 
     pub(crate) async fn change_user_password_impl(
diff --git a/crates/storage-sqlite/src/admin_store.rs b/crates/storage-sqlite/src/admin_store.rs
index 6780f317b..e0e590de8 100644
--- a/crates/storage-sqlite/src/admin_store.rs
+++ b/crates/storage-sqlite/src/admin_store.rs
@@ -123,10 +123,7 @@ impl AdminStore for SqliteCatalogStore {
             let Some((hash,)) = row else {
                 return Ok(None);
             };
-            let verified = tokio::task::spawn_blocking(move || bcrypt::verify(&password, &hash))
-                .await
-                .map_err(|e| OpError::Internal(format!("bcrypt task: {e}")))?
-                .unwrap_or(false);
+            let verified = extenddb_storage::password::verify_password(password, hash).await?;
             Ok(Some(verified))
         })
     }
diff --git a/crates/storage-sqlite/src/catalog_store.rs b/crates/storage-sqlite/src/catalog_store.rs
index 410aa46d6..17af9a254 100644
--- a/crates/storage-sqlite/src/catalog_store.rs
+++ b/crates/storage-sqlite/src/catalog_store.rs
@@ -268,18 +268,21 @@ impl RateLimitStore for SqliteCatalogStore {
     fn count_principal_failures(
         &self,
         principal: &str,
+        source_ip: &str,
         window_seconds: i64,
     ) -> BoxFuture<'_, OpResult<i64>> {
         let principal = principal.to_owned();
+        let source_ip = source_ip.to_owned();
         Box::pin(async move {
             let cutoff = format_timestamp(
                 OffsetDateTime::now_utc() - time::Duration::seconds(window_seconds),
             );
             let row: (i64,) = sqlx::query_as(
                 "SELECT COUNT(*) FROM login_attempts \
-                 WHERE principal = ? AND success = 0 AND attempted_at > ?",
+                 WHERE principal = ? AND source_ip = ? AND success = 0 AND attempted_at > ?",
             )
             .bind(&principal)
+            .bind(&source_ip)
             .bind(&cutoff)
             .fetch_one(&self.pool)
             .await
diff --git a/crates/storage-sqlite/src/management_store/users.rs b/crates/storage-sqlite/src/management_store/users.rs
index 13f1e97fa..c28f35a12 100644
--- a/crates/storage-sqlite/src/management_store/users.rs
+++ b/crates/storage-sqlite/src/management_store/users.rs
@@ -223,12 +223,7 @@ impl SqliteCatalogStore {
             // User absent or has no console password.
             return Ok(false);
         };
-        let password = password.to_owned();
-        let verified = tokio::task::spawn_blocking(move || bcrypt::verify(&password, &hash))
-            .await
-            .map_err(|e| OpError::Internal(format!("bcrypt task: {e}")))?
-            .unwrap_or(false);
-        Ok(verified)
+        extenddb_storage::password::verify_password(password.to_owned(), hash).await
     }
 
     pub(crate) async fn change_user_password_impl(
diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs
index 8c6fd57b8..b48719ee5 100755
--- a/crates/storage/src/lib.rs
+++ b/crates/storage/src/lib.rs
@@ -19,6 +19,7 @@ pub mod hooks;
 pub mod management_store;
 pub mod management_store_factory;
 pub mod operations;
+pub mod password;
 pub mod policy_migration;
 pub mod server_components;
 pub mod settings_store;
diff --git a/crates/storage/src/management_store/mod.rs b/crates/storage/src/management_store/mod.rs
index c0b753845..6adf6a87c 100755
--- a/crates/storage/src/management_store/mod.rs
+++ b/crates/storage/src/management_store/mod.rs
@@ -87,12 +87,15 @@ pub trait MetricsStore: Send + Sync {
 
 // ── Rate limit store ───────────────────────────────────────────────────
 
-/// Login rate limiting and account lockout storage.
+/// Storage for login rate limiting per user and source address.
 pub trait RateLimitStore: Send + Sync {
-    /// Count failed login attempts for a principal within the lookback window.
+    /// Count failed login attempts for a principal from one source IP within
+    /// the lookback window. Failures from other sources do not count, so a
+    /// third party cannot lock the principal out of its own address.
     fn count_principal_failures(
         &self,
         principal: &str,
+        source_ip: &str,
         window_seconds: i64,
     ) -> BoxFuture<'_, OpResult<i64>>;
 
diff --git a/crates/storage/src/password.rs b/crates/storage/src/password.rs
new file mode 100644
index 000000000..9afcc0b86
--- /dev/null
+++ b/crates/storage/src/password.rs
@@ -0,0 +1,307 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! bcrypt on a blocking thread, bounded by one process-wide semaphore.
+//!
+//! bcrypt takes about 100 ms per call at the default cost, so it runs on
+//! tokio's blocking pool. Hashing and verification share one semaphore so
+//! that no more than [`MAX_CONCURRENT_BCRYPT`] bcrypt calls run at once in
+//! this process: a burst of login requests cannot occupy more than that many
+//! cores, and the sqlx blocking threads that share the pool are not starved.
+//!
+//! Every backend's password verification goes through [`verify_password`],
+//! and the management API hashes through [`hash_password`], so the bound
+//! holds whatever store a login is checked against.
+//!
+//! A slot is held by the blocking call itself, not by the async caller. A
+//! caller whose future is dropped while bcrypt runs (a client that closes
+//! the connection mid-login) cannot release its slot early: the slot frees
+//! when bcrypt returns.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+
+use tokio::sync::{Semaphore, SemaphorePermit};
+
+use crate::management_store::{OpError, OpResult};
+
+/// Maximum bcrypt calls (hash or verify) running at once in this process.
+pub const MAX_CONCURRENT_BCRYPT: usize = 4;
+
+/// How long a verification waits for a bcrypt slot before giving up. A login
+/// that waits longer than this is answered with an error rather than queued
+/// without bound; it is not recorded as a failed attempt because no password
+/// was checked.
+pub const VERIFY_WAIT: Duration = Duration::from_secs(10);
+
+/// The `OpError::Internal` message of a verification that found no free
+/// bcrypt slot within [`VERIFY_WAIT`]. Callers match on it to log the
+/// overload apart from storage errors.
+pub const VERIFY_OVERLOADED: &str = "password verification is overloaded; try again later";
+
+static BCRYPT_SEMAPHORE: Semaphore = Semaphore::const_new(MAX_CONCURRENT_BCRYPT);
+
+/// bcrypt calls running right now (on the blocking pool).
+static RUNNING: AtomicUsize = AtomicUsize::new(0);
+
+/// The most bcrypt calls ever observed running at once in this process.
+static HIGH_WATER: AtomicUsize = AtomicUsize::new(0);
+
+fn semaphore_closed() -> OpError {
+    OpError::Internal("bcrypt semaphore closed".to_owned())
+}
+
+/// Whether `error` is the refusal [`verify_password`] returns when no bcrypt
+/// slot became free within [`VERIFY_WAIT`].
+#[must_use]
+pub fn is_overloaded(error: &OpError) -> bool {
+    matches!(error, OpError::Internal(msg) if msg == VERIFY_OVERLOADED)
+}
+
+/// Hash `password` with bcrypt at `cost` on a blocking thread, holding one
+/// bcrypt slot until bcrypt returns.
+///
+/// Waits as long as it takes for a slot: the caller is an authenticated
+/// admin creating or changing a password, and a queue is preferable to an
+/// error.
+///
+/// # Errors
+///
+/// Returns `OpError::Internal` when bcrypt refuses the input or the blocking
+/// task failed.
+pub async fn hash_password(password: String, cost: u32) -> OpResult<String> {
+    let permit = BCRYPT_SEMAPHORE
+        .acquire()
+        .await
+        .map_err(|_| semaphore_closed())?;
+    run_bcrypt(permit, move || bcrypt::hash(password, cost))
+        .await?
+        .map_err(|e| OpError::Internal(format!("bcrypt hash: {e}")))
+}
+
+/// Verify `password` against a bcrypt `hash` on a blocking thread, holding
+/// one bcrypt slot until bcrypt returns.
+///
+/// Returns `Ok(false)` for a wrong password and for a hash bcrypt cannot
+/// parse, matching how the stores treated a malformed hash before.
+///
+/// # Errors
+///
+/// Returns `OpError::Internal` with [`VERIFY_OVERLOADED`] when no bcrypt
+/// slot became free within [`VERIFY_WAIT`], and `OpError::Internal` when the
+/// blocking task failed. Callers treat both as a server error, not as a
+/// failed login.
+pub async fn verify_password(password: String, hash: String) -> OpResult<bool> {
+    let Ok(acquired) = tokio::time::timeout(VERIFY_WAIT, BCRYPT_SEMAPHORE.acquire()).await else {
+        tracing::warn!(
+            "password verification waited {}s for a bcrypt slot; rejecting",
+            VERIFY_WAIT.as_secs()
+        );
+        return Err(OpError::Internal(VERIFY_OVERLOADED.to_owned()));
+    };
+    let permit = acquired.map_err(|_| semaphore_closed())?;
+    run_bcrypt(permit, move || {
+        bcrypt::verify(&password, &hash).unwrap_or(false)
+    })
+    .await
+}
+
+/// Run one bcrypt call on the blocking pool. The permit moves into the
+/// blocking closure and is dropped there after the call returns, so the slot
+/// stays taken for the whole call even if the caller's future is dropped
+/// while it waits for the result.
+async fn run_bcrypt<T: Send + 'static>(
+    permit: SemaphorePermit<'static>,
+    work: impl FnOnce() -> T + Send + 'static,
+) -> OpResult<T> {
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        let running = RUNNING.fetch_add(1, Ordering::SeqCst) + 1;
+        HIGH_WATER.fetch_max(running, Ordering::SeqCst);
+        let out = work();
+        RUNNING.fetch_sub(1, Ordering::SeqCst);
+        out
+    })
+    .await
+    .map_err(|e| OpError::Internal(format!("bcrypt task: {e}")))
+}
+
+/// The most bcrypt calls observed running at once since the process started.
+/// Never exceeds [`MAX_CONCURRENT_BCRYPT`].
+#[must_use]
+pub fn high_water_mark() -> usize {
+    HIGH_WATER.load(Ordering::SeqCst)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const COST: u32 = 4; // cheapest bcrypt cost; keeps the tests fast
+
+    /// Tests that read `RUNNING` or the high-water mark take this lock, so
+    /// their bcrypt calls are the only ones in flight in this process.
+    static COUNTERS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+    /// Check the result of a wrong password queued behind other bcrypt calls:
+    /// refused as wrong, or refused as overloaded when the host is slow
+    /// enough that no slot freed within [`VERIFY_WAIT`].
+    fn assert_wrong_or_overloaded(result: OpResult<bool>) {
+        match result {
+            Ok(verified) => assert!(!verified, "a wrong password verified"),
+            Err(e) => assert!(is_overloaded(&e), "unexpected error: {e:?}"),
+        }
+    }
+
+    /// Wait until `n` bcrypt calls are running on the blocking pool.
+    async fn wait_until_running(n: usize) {
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+        while RUNNING.load(Ordering::SeqCst) < n {
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "{n} bcrypt calls never started"
+            );
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+    }
+
+    /// Start `MAX_CONCURRENT_BCRYPT` slow bcrypt calls through `start`, cancel
+    /// their callers while bcrypt runs, then run `MAX_CONCURRENT_BCRYPT` fast
+    /// verifications and return the peak number of bcrypt calls that ran at
+    /// once. A slot released by the cancelled caller lets a fast call start
+    /// beside a slow one still running, which pushes the peak past the bound.
+    async fn peak_after_cancelling<F, Fut>(start: F) -> usize
+    where
+        F: Fn() -> Fut,
+        Fut: std::future::Future + Send + 'static,
+        Fut::Output: Send + 'static,
+    {
+        let callers: Vec<_> = (0..MAX_CONCURRENT_BCRYPT)
+            .map(|_| tokio::spawn(start()))
+            .collect();
+        wait_until_running(MAX_CONCURRENT_BCRYPT).await;
+        for caller in &callers {
+            caller.abort();
+        }
+        for caller in callers {
+            let Err(cancelled) = caller.await else {
+                panic!("the caller finished before it was cancelled");
+            };
+            assert!(cancelled.is_cancelled());
+        }
+
+        let fast_hash = bcrypt::hash("pw", COST).unwrap();
+        let fresh: Vec<_> = (0..MAX_CONCURRENT_BCRYPT)
+            .map(|_| {
+                let hash = fast_hash.clone();
+                tokio::spawn(async move { verify_password("wrong".to_owned(), hash).await })
+            })
+            .collect();
+        for task in fresh {
+            assert_wrong_or_overloaded(task.await.unwrap());
+        }
+        // The cancelled calls run to completion on the blocking pool.
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+        while RUNNING.load(Ordering::SeqCst) > 0 {
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "bcrypt calls never finished"
+            );
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+        high_water_mark()
+    }
+
+    #[tokio::test]
+    async fn verify_accepts_the_right_password_and_refuses_a_wrong_one() {
+        let _counters = COUNTERS.lock().await;
+        let hash = bcrypt::hash("correct horse", COST).unwrap();
+        assert!(
+            verify_password("correct horse".to_owned(), hash.clone())
+                .await
+                .unwrap()
+        );
+        assert!(
+            !verify_password("battery staple".to_owned(), hash)
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn a_malformed_hash_is_a_wrong_password_not_an_error() {
+        let _counters = COUNTERS.lock().await;
+        assert!(
+            !verify_password("x".to_owned(), "not a bcrypt hash".to_owned())
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn hash_produces_a_hash_that_verifies() {
+        let _counters = COUNTERS.lock().await;
+        let hash = hash_password("correct horse".to_owned(), COST)
+            .await
+            .unwrap();
+        assert!(
+            verify_password("correct horse".to_owned(), hash)
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+    async fn a_cancelled_verification_keeps_its_slot_until_bcrypt_returns() {
+        let _counters = COUNTERS.lock().await;
+        // Cost 12 keeps each cancelled call running long after its caller is
+        // gone, so a fast call admitted early would overlap it.
+        let slow_hash = bcrypt::hash("pw", 12).unwrap();
+        let peak = peak_after_cancelling(|| {
+            let hash = slow_hash.clone();
+            async move { verify_password("wrong".to_owned(), hash).await }
+        })
+        .await;
+        assert!(
+            peak <= MAX_CONCURRENT_BCRYPT,
+            "{peak} bcrypt calls ran at once after cancelled verifications; \
+             the bound is {MAX_CONCURRENT_BCRYPT}"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+    async fn a_cancelled_hash_keeps_its_slot_until_bcrypt_returns() {
+        let _counters = COUNTERS.lock().await;
+        let peak =
+            peak_after_cancelling(|| async { hash_password("pw".to_owned(), 12).await }).await;
+        assert!(
+            peak <= MAX_CONCURRENT_BCRYPT,
+            "{peak} bcrypt calls ran at once after cancelled hashes; \
+             the bound is {MAX_CONCURRENT_BCRYPT}"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+    async fn a_burst_of_verifications_never_runs_more_than_the_bound_at_once() {
+        let _counters = COUNTERS.lock().await;
+        // Cost 10 keeps each verify around 50 ms so the burst overlaps for
+        // real; with 32 requests and 4 slots the burst takes 8 rounds.
+        let hash = bcrypt::hash("pw", 10).unwrap();
+        let mut tasks = Vec::new();
+        for _ in 0..32 {
+            let hash = hash.clone();
+            tasks.push(tokio::spawn(async move {
+                verify_password("wrong".to_owned(), hash).await
+            }));
+        }
+        for task in tasks {
+            assert_wrong_or_overloaded(task.await.unwrap());
+        }
+        let peak = high_water_mark();
+        assert!(peak >= 1, "no verification was observed running");
+        assert!(
+            peak <= MAX_CONCURRENT_BCRYPT,
+            "{peak} verifications ran at once; the bound is {MAX_CONCURRENT_BCRYPT}"
+        );
+    }
+}
diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md
index dcfda3a3d..34cba1ef0 100755
--- a/docs/design/04-component-storage.md
+++ b/docs/design/04-component-storage.md
@@ -44,7 +44,7 @@ ExtendDB defines **13 traits** across three categories:
 - `AdminStore` — admin user management
 - `SettingsStore` — runtime settings (key-value store)
 - `MetricsStore` — historical metrics persistence and query
-- `RateLimitStore` — login rate limiting and account lockout
+- `RateLimitStore`: login rate limiting per user and source address
 - `AuthorizationStore` — policy lookups for authorization decisions
 
 **Initialization and Lifecycle** (defined in `storage/src/bootstrapper.rs`):
@@ -207,12 +207,14 @@ storage.
 
 ### RateLimitStore
 
-Login rate limiting and account lockout:
+Login rate limiting per user and source address:
 - `count_principal_failures`, `count_ip_failures`, `record_failed_login`,
   `cleanup_old_attempts`
 
-Tracks failed login attempts by principal and source IP to mitigate clients
-sending excessive traffic.
+Records failed login attempts with the principal and the source address, and
+counts them per (principal, source address) pair and per source address, so
+the server can refuse an address that has failed too often for one principal
+or overall without locking the principal out of other addresses.
 
 ### AuthorizationStore
 
diff --git a/docs/manuals/01-architecture-guide.md b/docs/manuals/01-architecture-guide.md
index ea9c52da3..e6b407075 100755
--- a/docs/manuals/01-architecture-guide.md
+++ b/docs/manuals/01-architecture-guide.md
@@ -282,7 +282,7 @@ The console uses the `ManagementStore` trait for all data access — it never qu
 - Sessions stored in-memory with random tokens in HttpOnly cookies (`SameSite=Strict`, `Path=/console`)
 - CSRF tokens on all state-changing POST handlers
 - Security headers: X-Content-Type-Options (nosniff), X-Frame-Options (DENY), Referrer-Policy (strict-origin-when-cross-origin)
-- Login rate limiting with account lockout on excessive failures
+- Login rate limiting per user and source address, shared with the management API (see the security model manual)
 
 ### Unauthenticated Routes
 
diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md
index bb7a1c75d..b50944152 100755
--- a/docs/manuals/10-security-model.md
+++ b/docs/manuals/10-security-model.md
@@ -172,7 +172,30 @@ The management web console (`/console/*`) implements:
 - **Session management**: Server-side sessions with 8-hour expiry, HttpOnly cookies with `SameSite=Strict` and `Path=/console`
 - **Security headers**: X-Content-Type-Options (nosniff), X-Frame-Options (DENY), Referrer-Policy (strict-origin-when-cross-origin)
 - **HSTS**: Sent automatically (TLS is always enabled)
-- **Login rate limiting**: Failed login attempts are tracked per user. Excessive failures trigger account lockout.
+- **Login rate limiting**: Failed login attempts are counted per user and source address, and per source address, as described under [Login Rate Limiting](#login-rate-limiting).
+
+## Login Rate Limiting
+
+The console login form, the management API (HTTP Basic auth) and the `/metrics` endpoint share one set of limits on failed logins. Attempts are stored in the catalog's `login_attempts` table, so every surface sees the same history.
+
+- **5 failures per user and source address in 15 minutes.** After the fifth wrong password for a user from one address, further attempts for that user from that address are answered `429 Too Many Requests` until the oldest failure is 15 minutes old. Attempts for the same user from other addresses are not affected, so a third party who knows an admin's name cannot lock the admin out.
+- **20 failures per source address in 15 minutes**, counted over all users. This caps guessing from one address at 20 passwords per 15 minutes however many names it tries.
+- **Attempts are reserved before the password is checked.** An attempt counts against both limits from the moment it is admitted until its failure is recorded, so a burst of concurrent wrong passwords is admitted up to the cap and the rest are refused without a password check.
+- **Bounded verification.** At most 4 bcrypt operations (password checks and password hashes) run at once per process. A login that waits more than 10 seconds for a slot is answered `500` and is not counted as a failure, since no password was checked.
+
+A successful login is not recorded. A storage error before or during the password check is answered `500` and is not counted against the caller.
+
+The source address is the peer address of the TCP connection, with an IPv4-mapped IPv6 address (`::ffff:a.b.c.d`) counted as the IPv4 address. Any other IPv6 address is counted by its /64 prefix, because a /64 is the block usually assigned to one host or site and a client can send from any address in it; all addresses in one /64 share one set of limits. ExtendDB does not read `X-Forwarded-For` or any other forwarded-address header (they are caller-supplied).
+
+Callers that reach ExtendDB from one address share one set of limits. Five failures for a user from any of them refuse that user to all of them, and 20 failures in total refuse every user, until the oldest failure is 15 minutes old. Callers share an address in these cases:
+
+- **A reverse proxy or load balancer** that opens its own connections to ExtendDB, so every request arrives from the proxy's address. Put a proxy or load balancer in front of ExtendDB only when it preserves the client address, or when callers behind it are trusted not to lock each other out.
+- **The server host itself.** `extenddb manage` and a `/metrics` scraper running on the server host connect from the same address. A scraper with a stale password for an admin refuses that admin on the host after 5 failures, `extenddb manage` included.
+- **A NAT gateway**, whose egress address is shared by every client behind it.
+
+To recover an address that is refused, wait until its oldest failure is 15 minutes old, or delete the address's rows from the catalog's `login_attempts` table.
+
+A password can be guessed against one user from many addresses at 5 attempts per address per 15 minutes; only the source-address limit and the strength of the password bound that. Use long passwords for admin users and restrict which networks can reach the management and console ports.
 
 ## Provisioned Throughput Throttling
 
diff --git a/docs/manuals/11-deployment-guide.md b/docs/manuals/11-deployment-guide.md
index 77624ce38..48bae71d5 100755
--- a/docs/manuals/11-deployment-guide.md
+++ b/docs/manuals/11-deployment-guide.md
@@ -206,7 +206,7 @@ extenddb init \
 
 - [ ] Set `bind_addr` to the appropriate interface (not `0.0.0.0` unless behind a load balancer)
 - [ ] Firewall: allow only necessary ports (extenddb port, storage backend port)
-- [ ] Consider a reverse proxy (nginx, HAProxy) for TLS termination, rate limiting, and access logging
+- [ ] Consider a reverse proxy (nginx, HAProxy) for TLS termination, rate limiting, and access logging. Unless the proxy preserves the client address, login limits count the proxy's address as the source for every caller behind it (see the security model manual).
 
 ### Database (PostgreSQL)
 
diff --git a/docs/manuals/12-extending-extenddb-storage.md b/docs/manuals/12-extending-extenddb-storage.md
index 75f701d54..e77ad96b5 100755
--- a/docs/manuals/12-extending-extenddb-storage.md
+++ b/docs/manuals/12-extending-extenddb-storage.md
@@ -40,7 +40,7 @@ A new backend must implement **13 storage traits** plus the `CredentialStore` tr
 7. `AdminStore` — admin user management
 8. `SettingsStore` — runtime settings
 9. `MetricsStore` — historical metrics persistence and query
-10. `RateLimitStore` — login rate limiting and account lockout
+10. `RateLimitStore`: login rate limiting per user and source address
 11. `AuthorizationStore` — policy lookups for authorization decisions
 12. `BackupEngine` — backup and restore operations
 13. `Bootstrapper` — database initialization, destruction, migration, verification
@@ -218,10 +218,13 @@ Defined in `crates/storage/src/management_store/mod.rs`. Login rate limiting:
 
 | Method | Purpose |
 |--------|---------|
-| `record_login_attempt` | Record a login attempt (success or failure) |
-| `recent_failed_attempts` | Count recent failed attempts for lockout decisions |
+| `count_principal_failures` | Count recent failed attempts for one principal from one source address |
+| `count_ip_failures` | Count recent failed attempts from one source address, over all principals |
+| `record_failed_login` | Record a failed login attempt with its source address |
 | `cleanup_old_attempts` | Garbage-collect old login attempt records |
 
+Password verification in a backend's `AdminStore` and `ManagementStore` must go through `extenddb_storage::password::verify_password`, which bounds concurrent bcrypt work per process.
+
 ### AuthorizationStore
 
 Defined in `crates/storage/src/authorization_store.rs`. Policy lookups for authorization:
diff --git a/tests/test_rate_limiting.py b/tests/test_rate_limiting.py
index a807281d9..4f12cba49 100644
--- a/tests/test_rate_limiting.py
+++ b/tests/test_rate_limiting.py
@@ -1,11 +1,19 @@
 # Copyright 2026 ExtendDB contributors
 # SPDX-License-Identifier: Apache-2.0
 
-"""Rate limiting and account lockout integration tests.
+"""Login rate limiting integration tests.
 
-Verifies that repeated failed login attempts on the management API trigger the
-per-principal lockout enforced by the storage backend's RateLimitStore
-implementation.
+Failed logins are limited per user and source address: five failures for a
+user from one address refuse that user from that address, and twenty
+failures from one address refuse every user from it. These tests check both
+limits on the management API, the console login form and ``/metrics``,
+against the storage backend's RateLimitStore implementation.
+
+Each test sends its attempts from loopback addresses of its own
+(``127.x.y.z``, which Linux routes to the loopback interface), so the suite
+records no failure against 127.0.0.1 and can be run again straight away. A
+test skips when the endpoint is not a loopback address or the address
+cannot be bound (macOS configures only 127.0.0.1 by default).
 
 ExtendDB-only: rate limiting is not exercised against real DynamoDB.
 
@@ -18,11 +26,16 @@
 
 from __future__ import annotations
 
+import ipaddress
 import os
+import random
+import socket
 import uuid
+from urllib.parse import urlsplit
 
 import pytest
 import requests
+from requests.adapters import HTTPAdapter
 
 from management_helpers import ManagementClient
 
@@ -42,17 +55,67 @@ def _require_extenddb_env() -> tuple[str, str, str]:
 # MAX_FAILURES_PER_PRINCIPAL from crates/server/src/rate_limit.rs
 _LOCKOUT_THRESHOLD = 5
 
+# MAX_FAILURES_PER_IP from crates/server/src/rate_limit.rs
+_SOURCE_LIMIT = 20
+
+_TIMEOUT = 30
+
+
+class _SourceAddressAdapter(HTTPAdapter):
+    """Opens every connection from one local address."""
+
+    def __init__(self, source_ip: str) -> None:
+        # Set before HTTPAdapter.__init__, which calls init_poolmanager.
+        self._source_address = (source_ip, 0)
+        super().__init__()
+
+    def init_poolmanager(self, *args, **kwargs):
+        kwargs["source_address"] = self._source_address
+        super().init_poolmanager(*args, **kwargs)
+
+
+def _session_from_fresh_loopback_address(endpoint: str) -> requests.Session:
+    """A session whose connections leave from a loopback address no other
+    test uses, so the failures it records count against nobody else."""
+    host = urlsplit(endpoint).hostname or ""
+    if host != "localhost":
+        try:
+            address = ipaddress.ip_address(host)
+        except ValueError:
+            address = None
+        if address is None or address.version != 4 or not address.is_loopback:
+            pytest.skip(f"sending from 127.x.y.z needs a loopback endpoint, not {host}")
+    source_ip = "127.{}.{}.{}".format(
+        random.randint(1, 254), random.randint(0, 255), random.randint(1, 254)
+    )
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        try:
+            probe.bind((source_ip, 0))
+        except OSError as exc:
+            pytest.skip(f"cannot send from {source_ip}: {exc}")
+    session = requests.Session()
+    adapter = _SourceAddressAdapter(source_ip)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
 
 class TestRateLimiting:
     @pytest.fixture(autouse=True)
     def setup_and_teardown(self):
         endpoint, admin_user, admin_pass = _require_extenddb_env()
-        self.endpoint = endpoint
+        self.endpoint = endpoint.rstrip("/")
         self.mgmt = ManagementClient(endpoint, admin_user, admin_pass)
         self.verify = not endpoint.startswith("https://")
 
+        # Address A sends the failures; address B checks that another
+        # address is not affected by them.
+        self.address_a = _session_from_fresh_loopback_address(endpoint)
+        self.address_b = _session_from_fresh_loopback_address(endpoint)
+
         self.account_id = f"{uuid.uuid4().int % 10**12:012d}"
         self.user_name = f"ratelimit-{uuid.uuid4().hex[:8]}"
+        self.admins: list[str] = []
 
         resp = self.mgmt.create_account(self.account_id, f"ratelimit-acct-{self.account_id}")
         assert resp.status_code == 201, resp.text
@@ -61,19 +124,66 @@ def setup_and_teardown(self):
 
         yield
 
+        for name in self.admins:
+            requests.delete(
+                f"{self.mgmt.base_url}/admins/{name}",
+                auth=self.mgmt.admin_auth,
+                timeout=_TIMEOUT,
+                verify=self.verify,
+            )
         self.mgmt.delete_account(self.account_id)
+        self.address_a.close()
+        self.address_b.close()
 
-    def _management_request_with_password(self, password: str) -> requests.Response:
+    def _management_request_with_password(
+        self, password: str, session: requests.Session | None = None
+    ) -> requests.Response:
         """Hit a self-service management endpoint as the test user with the given password."""
-        return requests.get(
+        return (session or self.address_a).get(
             f"{self.mgmt.base_url}/accounts/{self.account_id}/users/{self.user_name}/access-keys",
             auth=(f"{self.account_id}/{self.user_name}", password),
-            timeout=10,
+            timeout=_TIMEOUT,
+            verify=self.verify,
+        )
+
+    def _new_admin(self) -> tuple[str, str]:
+        """Create an admin for this test only; teardown deletes it."""
+        name = f"ratelimit-{uuid.uuid4().hex[:8]}"
+        password = "AdminPass123!"
+        resp = requests.post(
+            f"{self.mgmt.base_url}/admins",
+            json={"admin_name": name, "password": password},
+            auth=self.mgmt.admin_auth,
+            timeout=_TIMEOUT,
+            verify=self.verify,
+        )
+        assert resp.status_code == 201, resp.text
+        self.admins.append(name)
+        return name, password
+
+    def _get_as(
+        self, session: requests.Session, path: str, user: str, password: str
+    ) -> requests.Response:
+        return session.get(
+            f"{self.endpoint}{path}",
+            auth=(user, password),
+            timeout=_TIMEOUT,
+            verify=self.verify,
+        )
+
+    def _console_login(
+        self, session: requests.Session, user: str, password: str
+    ) -> requests.Response:
+        return session.post(
+            f"{self.endpoint}/console/login",
+            data={"username": user, "password": password},
+            allow_redirects=False,
+            timeout=_TIMEOUT,
             verify=self.verify,
         )
 
     def test_principal_locked_out_after_threshold_failures(self):
-        """After MAX_FAILURES_PER_PRINCIPAL bad passwords the account is locked out."""
+        """After MAX_FAILURES_PER_PRINCIPAL bad passwords the user is refused from that address."""
         for i in range(_LOCKOUT_THRESHOLD):
             resp = self._management_request_with_password("wrongpassword")
             assert resp.status_code == 401, (
@@ -91,7 +201,7 @@ def test_principal_locked_out_after_threshold_failures(self):
         )
 
     def test_correct_password_still_works_before_lockout(self):
-        """Fewer than threshold failures do not lock out the account."""
+        """Fewer than threshold failures do not refuse the user."""
         for _ in range(_LOCKOUT_THRESHOLD - 1):
             resp = self._management_request_with_password("wrongpassword")
             assert resp.status_code == 401
@@ -102,3 +212,93 @@ def test_correct_password_still_works_before_lockout(self):
             f"Expected 200 with correct password before lockout, "
             f"got {resp.status_code}: {resp.text}"
         )
+
+    def test_failures_from_one_address_do_not_refuse_the_user_from_another(self):
+        """The per-user limit counts failures from one source address only."""
+        for _ in range(_LOCKOUT_THRESHOLD):
+            assert self._management_request_with_password("wrongpassword").status_code == 401
+        resp = self._management_request_with_password("ValidPass123!")
+        assert resp.status_code == 429, f"address A should be refused: {resp.text}"
+
+        resp = self._management_request_with_password("ValidPass123!", self.address_b)
+        assert resp.status_code == 200, (
+            f"Expected 200 from another address, got {resp.status_code}: {resp.text}"
+        )
+
+    def test_an_address_is_refused_after_twenty_failures_across_users(self):
+        """MAX_FAILURES_PER_IP failures for different users refuse every user from that address."""
+        run = uuid.uuid4().hex[:8]
+        for n in range(_SOURCE_LIMIT):
+            resp = self._get_as(self.address_a, "/management/admins", f"nobody-{run}-{n}", "x")
+            assert resp.status_code == 401, (
+                f"Expected 401 for user {n + 1}, got {resp.status_code}: {resp.text}"
+            )
+
+        newcomer = f"nobody-{run}-new"
+        resp = self._get_as(self.address_a, "/management/admins", newcomer, "x")
+        assert resp.status_code == 429, (
+            f"Expected 429 after {_SOURCE_LIMIT} failures from one address, "
+            f"got {resp.status_code}: {resp.text}"
+        )
+        assert "from this address" in resp.text, resp.text
+
+        resp = self._get_as(self.address_b, "/management/admins", newcomer, "x")
+        assert resp.status_code == 401, (
+            f"Expected 401 from another address, got {resp.status_code}: {resp.text}"
+        )
+
+    def test_admin_wrong_password_is_counted_on_the_management_api(self):
+        name, password = self._new_admin()
+        for i in range(_LOCKOUT_THRESHOLD):
+            resp = self._get_as(self.address_a, "/management/accounts", name, "wrongpassword")
+            assert resp.status_code == 401, (
+                f"Expected 401 on attempt {i + 1}, got {resp.status_code}: {resp.text}"
+            )
+        resp = self._get_as(self.address_a, "/management/accounts", name, password)
+        assert resp.status_code == 429, (
+            f"Expected 429 for the right password after {_LOCKOUT_THRESHOLD} failures, "
+            f"got {resp.status_code}: {resp.text}"
+        )
+
+        resp = self._get_as(self.address_b, "/management/accounts", name, password)
+        assert resp.status_code == 200, (
+            f"Expected 200 from another address, got {resp.status_code}: {resp.text}"
+        )
+
+    def test_console_login_failures_are_counted(self):
+        name, password = self._new_admin()
+        for i in range(_LOCKOUT_THRESHOLD):
+            resp = self._console_login(self.address_a, name, "wrongpassword")
+            assert resp.status_code == 200, (
+                f"Expected the login form on attempt {i + 1}, got {resp.status_code}"
+            )
+            assert "Invalid credentials" in resp.text
+        resp = self._console_login(self.address_a, name, password)
+        assert resp.status_code == 429, (
+            f"Expected 429 for the right password after {_LOCKOUT_THRESHOLD} failures, "
+            f"got {resp.status_code}"
+        )
+        assert "too many" in resp.text.lower()
+
+        resp = self._console_login(self.address_b, name, password)
+        assert resp.status_code == 303, (
+            f"Expected a redirect after login from another address, got {resp.status_code}"
+        )
+
+    def test_metrics_login_failures_are_counted(self):
+        name, password = self._new_admin()
+        for i in range(_LOCKOUT_THRESHOLD):
+            resp = self._get_as(self.address_a, "/metrics", name, "wrongpassword")
+            assert resp.status_code == 401, (
+                f"Expected 401 on attempt {i + 1}, got {resp.status_code}: {resp.text}"
+            )
+        resp = self._get_as(self.address_a, "/metrics", name, password)
+        assert resp.status_code == 429, (
+            f"Expected 429 for the right password after {_LOCKOUT_THRESHOLD} failures, "
+            f"got {resp.status_code}: {resp.text}"
+        )
+
+        resp = self._get_as(self.address_b, "/metrics", name, password)
+        assert resp.status_code == 200, (
+            f"Expected 200 from another address, got {resp.status_code}: {resp.text}"
+        )