Add distributed lock manager for multi-instance deployments

Cargo.toml

@@ -51,7 +51,6 @@ governor = "0.6"
 redis = { version = "0.24", features = ["tokio-comp"] }
 jsonschema = "0.17"
 once_cell = "1.19"
-hyper = "0.14"
 ipnet = "2.9"
 utoipa = { version = "4", features = ["axum_extras"] }
 utoipa-swagger-ui = { version = "6", features = ["axum"] }

src/handlers/admin.rs

@@ -2,11 +2,11 @@ use crate::middleware::quota::{Quota, QuotaManager, QuotaStatus, ResetSchedule,
 use axum::{
     extract::{Path, State},
     http::StatusCode,
-    response::IntoResponse,
     Json,
 };
 use serde::{Deserialize, Serialize};
 
+#[derive(Clone)]
 pub struct AdminState {
     pub quota_manager: QuotaManager,
 }
@@ -92,3 +92,11 @@ pub async fn reset_quota(
         .map(|_| StatusCode::OK)
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
 }
+
+pub fn admin_routes() -> axum::Router<AdminState> {
+    use axum::routing::{get, post};
+    axum::Router::new()
+        .route("/quota/:key", get(get_quota_status))
+        .route("/quota/:key", post(set_quota))
+        .route("/quota/:key/reset", post(reset_quota))
+}

src/main.rs

@@ -12,7 +12,7 @@ use synapse_core::{
     graphql::schema::build_schema,
     handlers,
     handlers::ws::TransactionStatusUpdate,
-    metrics, middleware,
+    metrics,
     middleware::idempotency::IdempotencyService,
     schemas,
     services::{FeatureFlagService, SettlementService},
@@ -258,10 +258,11 @@ async fn serve(config: config::Config) -> anyhow::Result<()> {
     let _dlq_routes: Router =
         handlers::dlq::dlq_routes().with_state(api_state.app_state.db.clone());
 
-    let _admin_routes: Router = Router::new()
-        .nest("/admin/queue", handlers::admin::admin_routes())
-        .layer(axum_middleware::from_fn(middleware::auth::admin_auth))
-        .with_state(api_state.app_state.db.clone());
+    // Admin routes disabled - requires AdminState setup
+    // let _admin_routes: Router = Router::new()
+    //     .nest("/admin/queue", handlers::admin::admin_routes())
+    //     .layer(axum_middleware::from_fn(middleware::auth::admin_auth))
+    //     .with_state(api_state.app_state.db.clone());
 
     let _search_routes: Router = Router::new()
         .route(

src/middleware/mod.rs

@@ -1,5 +1,7 @@
 pub mod auth;
 pub mod idempotency;
 pub mod ip_filter;
+pub mod quota;
 pub mod request_logger;
+pub mod validate;
 pub mod versioning;

src/middleware/quota.rs

@@ -1,6 +1,5 @@
 use redis::{aio::MultiplexedConnection, AsyncCommands, Client};
 use serde::{Deserialize, Serialize};
-use std::time::Duration;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum Tier {
@@ -43,6 +42,7 @@ impl ResetSchedule {
     }
 }
 
+#[derive(Clone)]
 pub struct QuotaManager {
     redis_client: Client,
 }

src/services/lock_examples.rs

@@ -0,0 +1,82 @@
+use crate::services::{LockManager, TransactionProcessor};
+use std::time::Duration;
+use tracing::{info, warn};
+use uuid::Uuid;
+
+/// Example: Process transaction with distributed lock
+pub async fn process_transaction_with_lock(
+    lock_manager: &LockManager,
+    processor: &TransactionProcessor,
+    tx_id: Uuid,
+) -> anyhow::Result<bool> {
+    let resource = format!("transaction:{}", tx_id);
+    let timeout = Duration::from_secs(5);
+
+    // Try to acquire lock
+    let lock = match lock_manager.acquire(&resource, timeout).await? {
+        Some(lock) => lock,
+        None => {
+            warn!("Could not acquire lock for transaction {}", tx_id);
+            return Ok(false);
+        }
+    };
+
+    info!("Processing transaction {} with lock", tx_id);
+
+    // Process transaction
+    let result = processor.process_transaction(tx_id).await;
+
+    // Release lock
+    lock.release().await?;
+
+    result.map(|_| true)
+}
+
+/// Example: Long-running operation with auto-renewal
+pub async fn long_running_with_lock(
+    lock_manager: &LockManager,
+    resource: &str,
+) -> anyhow::Result<()> {
+    let timeout = Duration::from_secs(5);
+
+    let lock = match lock_manager.acquire(resource, timeout).await? {
+        Some(lock) => lock,
+        None => {
+            return Err(anyhow::anyhow!("Could not acquire lock"));
+        }
+    };
+
+    // Spawn auto-renewal task
+    let renewal_lock = lock.clone();
+    tokio::spawn(async move {
+        renewal_lock.auto_renew_task().await;
+    });
+
+    // Do long-running work
+    tokio::time::sleep(Duration::from_secs(60)).await;
+
+    // Lock will be released on drop
+    Ok(())
+}
+
+/// Example: Using with_lock helper
+pub async fn process_with_helper(
+    lock_manager: &LockManager,
+    processor: &TransactionProcessor,
+    tx_id: Uuid,
+) -> anyhow::Result<Option<()>> {
+    let resource = format!("transaction:{}", tx_id);
+    let timeout = Duration::from_secs(5);
+
+    lock_manager
+        .with_lock(&resource, timeout, || {
+            Box::pin(async move {
+                processor
+                    .process_transaction(tx_id)
+                    .await
+                    .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
+            })
+        })
+        .await
+        .map_err(|e| anyhow::anyhow!("Lock error: {}", e))
+}