feat: readiness probe, shared cursor pagination, rate limit docs, and…

crates/api/src/auth.rs

@@ -172,7 +172,7 @@ fn check_auth_rate_limit(
     let ip = crate::rate_limit::client_ip(headers, peer);
     if state
         .rate_limiter()
-        .check(&ip, "auth", 10, std::time::Duration::from_secs(60))
+        .check(&ip, "auth", crate::rate_limit::AUTH_RATE_LIMIT, crate::rate_limit::AUTH_RATE_WINDOW)
     {
         Ok(())
     } else {
@@ -333,8 +333,8 @@ pub async fn resend_otp(
     if !state.rate_limiter().check(
         &format!("otp:{user_id}"),
         "otp_resend",
-        3,
-        std::time::Duration::from_secs(60 * 60),
+        crate::rate_limit::OTP_RESEND_USER_LIMIT,
+        crate::rate_limit::OTP_RESEND_USER_WINDOW,
     ) {
         return Err(ApiError::TooManyRequests(
             "too many resend attempts — wait a while and try again".into(),
@@ -344,8 +344,8 @@ pub async fn resend_otp(
     if !state.rate_limiter().check(
         &ip,
         "otp_resend_ip",
-        10,
-        std::time::Duration::from_secs(60 * 60),
+        crate::rate_limit::OTP_RESEND_IP_LIMIT,
+        crate::rate_limit::OTP_RESEND_IP_WINDOW,
     ) {
         return Err(ApiError::TooManyRequests(
             "too many resend attempts — wait a while and try again".into(),

crates/api/src/horizon.rs

@@ -387,6 +387,23 @@ impl Horizon {
             Err(ResilienceError::Exhausted(_)) => Err(ApiError::Internal),
         }
     }
+
+    // Probe Horizon root endpoint to verify node reachability.
+    pub async fn check_reachability(&self) -> Result<(), String> {
+        let url = self.base_url.trim_end_matches('/');
+        let resp = self
+            .http
+            .get(url)
+            .timeout(Duration::from_secs(3))
+            .send()
+            .await
+            .map_err(|e| format!("horizon unreachable: {e}"))?;
+        if resp.status().is_success() {
+            Ok(())
+        } else {
+            Err(format!("horizon returned HTTP {}", resp.status()))
+        }
+    }
 }
 
 // ---------------------------------------------------------------------------

crates/api/src/lib.rs

@@ -18,11 +18,12 @@ pub mod submit_validation;
 pub use error::{ApiError, ApiResult, Envelope};
 pub use state::AppState;
 
-use axum::extract::{DefaultBodyLimit, Request};
+use axum::extract::{DefaultBodyLimit, Request, State};
+use axum::http::StatusCode;
 use axum::middleware::{self, Next};
 use axum::response::{IntoResponse, Response};
 use axum::routing::{delete, get, post};
-use axum::Router;
+use axum::{Json, Router};
 use std::time::Duration;
 use tower_http::cors::{Any, CorsLayer};
 
@@ -52,6 +53,7 @@ pub fn build_router(state: AppState) -> Router {
     // together with the error handler that turns an oversized body into a 413 envelope.
     Router::new()
         .route("/health", get(health))
+        .route("/health/ready", get(health_ready))
         .route("/v1/auth/signup", post(auth::signup))
         .route("/v1/auth/verify-email", post(auth::verify_email))
         .route("/v1/auth/resend-otp", post(auth::resend_otp))
@@ -234,6 +236,53 @@ async fn health() -> &'static str {
     "ok"
 }
 
+// Readiness probe checking database and Horizon reachability.
+async fn health_ready(State(state): State<AppState>) -> impl IntoResponse {
+    let mut db_ok = false;
+    let mut horizon_ok = false;
+    let mut db_err = None;
+    let mut horizon_err = None;
+
+    match state.store().ping().await {
+        Ok(_) => db_ok = true,
+        Err(e) => db_err = Some(e.to_string()),
+    }
+
+    match state.horizon().check_reachability().await {
+        Ok(_) => horizon_ok = true,
+        Err(e) => horizon_err = Some(e),
+    }
+
+    if db_ok && horizon_ok {
+        (
+            StatusCode::OK,
+            Json(serde_json::json!({
+                "status": "ready",
+                "database": "ok",
+                "horizon": "ok"
+            })),
+        )
+    } else {
+        let mut failed = Vec::new();
+        if !db_ok {
+            failed.push("database");
+        }
+        if !horizon_ok {
+            failed.push("horizon");
+        }
+        (
+            StatusCode::SERVICE_UNAVAILABLE,
+            Json(serde_json::json!({
+                "status": "not_ready",
+                "database": if db_ok { "ok".to_string() } else { db_err.unwrap_or_else(|| "unreachable".into()) },
+                "horizon": if horizon_ok { "ok".to_string() } else { horizon_err.unwrap_or_else(|| "unreachable".into()) },
+                "failed": failed,
+                "error": format!("unreachable dependencies: {}", failed.join(", "))
+            })),
+        )
+    }
+}
+
 // NOTE: a `handle_errors` HandleErrorLayer helper lived here to convert oversized-body errors
 // into a 413 envelope. It is unnecessary with `DefaultBodyLimit` (axum renders that rejection as
 // 413 itself) and did not satisfy `Router::layer`'s Service bounds, so it was removed.

crates/api/src/rate_limit.rs

@@ -12,6 +12,24 @@ use std::time::{Duration, Instant};
 /// Cap on tracked (ip, class) buckets before expired entries are swept.
 const SWEEP_THRESHOLD: usize = 10_000;
 
+// Rate limit thresholds and fixed windows for API endpoints.
+pub const AUTH_RATE_LIMIT: u32 = 10;
+pub const AUTH_RATE_WINDOW: Duration = Duration::from_secs(60);
+pub const OTP_RESEND_USER_LIMIT: u32 = 3;
+pub const OTP_RESEND_USER_WINDOW: Duration = Duration::from_secs(3600);
+pub const OTP_RESEND_IP_LIMIT: u32 = 10;
+pub const OTP_RESEND_IP_WINDOW: Duration = Duration::from_secs(3600);
+pub const PAY_READ_LIMIT: u32 = 60;
+pub const PAY_READ_WINDOW: Duration = Duration::from_secs(60);
+pub const PAY_INTENT_LIMIT: u32 = 5;
+pub const PAY_INTENT_WINDOW: Duration = Duration::from_secs(60);
+pub const PAY_STATUS_LIMIT: u32 = 60;
+pub const PAY_STATUS_WINDOW: Duration = Duration::from_secs(60);
+pub const PAY_SIGNING_INFO_LIMIT: u32 = 60;
+pub const PAY_SIGNING_INFO_WINDOW: Duration = Duration::from_secs(60);
+pub const PAY_SUBMIT_LIMIT: u32 = 20;
+pub const PAY_SUBMIT_WINDOW: Duration = Duration::from_secs(60);
+
 /// Bucket key: the client IP plus the endpoint class it is being limited against.
 type BucketKey = (String, &'static str);
 /// Bucket value: when the current fixed window started, and hits so far within it.

crates/api/src/routes/payment_links.rs

@@ -326,11 +326,12 @@ fn check_public_rate_limit(
     peer: Option<axum::extract::ConnectInfo<std::net::SocketAddr>>,
     class: &'static str,
     limit: u32,
+    window: std::time::Duration,
 ) -> Result<(), ApiError> {
     let ip = crate::rate_limit::client_ip(headers, peer.map(|c| c.0));
     if state
         .rate_limiter()
-        .check(&ip, class, limit, std::time::Duration::from_secs(60))
+        .check(&ip, class, limit, window)
     {
         Ok(())
     } else {
@@ -347,7 +348,14 @@ pub async fn get_public_payment_link(
     peer: Option<axum::extract::ConnectInfo<std::net::SocketAddr>>,
     headers: HeaderMap,
 ) -> ApiResult<Json<Envelope<PublicPaymentLinkView>>> {
-    check_public_rate_limit(&state, &headers, peer, "pay_read", 60)?;
+    check_public_rate_limit(
+        &state,
+        &headers,
+        peer,
+        "pay_read",
+        crate::rate_limit::PAY_READ_LIMIT,
+        crate::rate_limit::PAY_READ_WINDOW,
+    )?;
     let link = state.store().get_payment_link_by_slug(&slug).await?;
     if !link.active {
         return Err(ApiError::NotFound);
@@ -390,7 +398,14 @@ pub async fn create_payment_intent(
     headers: HeaderMap,
     body: Bytes,
 ) -> ApiResult<(StatusCode, Json<Envelope<PaymentIntentView>>)> {
-    check_public_rate_limit(&state, &headers, peer, "pay_intent", 5)?;
+    check_public_rate_limit(
+        &state,
+        &headers,
+        peer,
+        "pay_intent",
+        crate::rate_limit::PAY_INTENT_LIMIT,
+        crate::rate_limit::PAY_INTENT_WINDOW,
+    )?;
     let link = state.store().get_payment_link_by_slug(&slug).await?;
     if !link.active {
         return Err(ApiError::NotFound);
@@ -463,7 +478,14 @@ pub async fn get_payment_status(
     headers: HeaderMap,
 ) -> ApiResult<Json<Envelope<PaymentStatusView>>> {
     // The pay page polls this every ~3s while waiting, so the ceiling is generous.
-    check_public_rate_limit(&state, &headers, peer, "pay_status", 60)?;
+    check_public_rate_limit(
+        &state,
+        &headers,
+        peer,
+        "pay_status",
+        crate::rate_limit::PAY_STATUS_LIMIT,
+        crate::rate_limit::PAY_STATUS_WINDOW,
+    )?;
     let link = state.store().get_payment_link_by_slug(&slug).await?;
     let payment = state
         .store()
@@ -501,7 +523,14 @@ pub async fn public_signing_info(
     peer: Option<axum::extract::ConnectInfo<std::net::SocketAddr>>,
     headers: HeaderMap,
 ) -> ApiResult<Json<Envelope<crate::routes::submit::SigningInfo>>> {
-    check_public_rate_limit(&state, &headers, peer, "pay_signing_info", 60)?;
+    check_public_rate_limit(
+        &state,
+        &headers,
+        peer,
+        "pay_signing_info",
+        crate::rate_limit::PAY_SIGNING_INFO_LIMIT,
+        crate::rate_limit::PAY_SIGNING_INFO_WINDOW,
+    )?;
     // Confirms the link exists/is active before doing any Horizon work on the caller's behalf.
     let link = state.store().get_payment_link_by_slug(&slug).await?;
     if !link.active {
@@ -552,7 +581,14 @@ pub async fn submit_payment(
     headers: HeaderMap,
     body: Bytes,
 ) -> ApiResult<(StatusCode, Json<Envelope<SubmitPaymentResponse>>)> {
-    check_public_rate_limit(&state, &headers, peer, "pay_submit", 20)?;
+    check_public_rate_limit(
+        &state,
+        &headers,
+        peer,
+        "pay_submit",
+        crate::rate_limit::PAY_SUBMIT_LIMIT,
+        crate::rate_limit::PAY_SUBMIT_WINDOW,
+    )?;
     let link = state.store().get_payment_link_by_slug(&slug).await?;
     if !link.active {
         return Err(ApiError::NotFound);

crates/api/tests/api_tests.rs

@@ -435,6 +435,97 @@ async fn health_is_public_and_ok() {
     assert_eq!(resp.status(), StatusCode::OK);
 }
 
+// Local mock Horizon server for readiness testing.
+async fn start_mock_horizon_ok() -> String {
+    let app = Router::new().route("/", axum::routing::get(|| async { "horizon ok" }));
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind mock horizon");
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    format!("http://{addr}")
+}
+
+#[tokio::test]
+async fn health_ready_returns_200_when_db_and_horizon_are_both_reachable() {
+    let Some(_) = test_state().await else {
+        return;
+    };
+    let mock_horizon = start_mock_horizon_ok().await;
+    let url = database_url().unwrap();
+    let store = Store::connect(&url).await.expect("connect");
+    let state = AppState::new(
+        store,
+        [42u8; 32],
+        StellarNetwork::Testnet,
+        mock_horizon,
+        None,
+        octo_email::EmailSender::new_captured(),
+    );
+    let app = build_router(state);
+    let resp = app.oneshot(get("/health/ready")).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let json = body_json(resp).await;
+    assert_eq!(json["status"], "ready");
+    assert_eq!(json["database"], "ok");
+    assert_eq!(json["horizon"], "ok");
+}
+
+#[tokio::test]
+async fn health_ready_returns_a_clear_503_naming_the_db_when_the_database_is_unreachable() {
+    let mock_horizon = start_mock_horizon_ok().await;
+    let dead_pool = sqlx::postgres::PgPoolOptions::new()
+        .acquire_timeout(std::time::Duration::from_millis(100))
+        .connect_lazy("postgres://postgres:wrong@127.0.0.1:1/nonexistent")
+        .unwrap();
+    let store = Store::from_pool(dead_pool);
+    let state = AppState::new(
+        store,
+        [42u8; 32],
+        StellarNetwork::Testnet,
+        mock_horizon,
+        None,
+        octo_email::EmailSender::new_captured(),
+    );
+    let app = build_router(state);
+    let resp = app.oneshot(get("/health/ready")).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+    let json = body_json(resp).await;
+    assert_eq!(json["status"], "not_ready");
+    assert_eq!(json["horizon"], "ok");
+    assert!(json["database"] != "ok");
+    let error_str = json["error"].as_str().unwrap();
+    assert!(error_str.contains("database"));
+}
+
+#[tokio::test]
+async fn health_ready_returns_a_clear_503_naming_horizon_when_horizon_is_unreachable() {
+    let Some(_) = test_state().await else {
+        return;
+    };
+    let url = database_url().unwrap();
+    let store = Store::connect(&url).await.expect("connect");
+    let state = AppState::new(
+        store,
+        [42u8; 32],
+        StellarNetwork::Testnet,
+        "http://127.0.0.1:1".into(),
+        None,
+        octo_email::EmailSender::new_captured(),
+    );
+    let app = build_router(state);
+    let resp = app.oneshot(get("/health/ready")).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+    let json = body_json(resp).await;
+    assert_eq!(json["status"], "not_ready");
+    assert_eq!(json["database"], "ok");
+    assert!(json["horizon"] != "ok");
+    let error_str = json["error"].as_str().unwrap();
+    assert!(error_str.contains("horizon"));
+}
+
 #[tokio::test]
 async fn backup_round_trips_the_opaque_blob_verbatim() {
     let Some(state) = test_state().await else {