Patrones de Arquitectura de Microservicios en Rust

Patrones de diseño para construir microservicios escalables en Rust

Jesús Pérez Lorenzo 8 min read
rust microservices architecture patterns
Explora patrones probados para construir arquitecturas de microservicios robustas usando Rust. Desde descubrimiento de servicios hasta comunicación basada en eventos, aprende cómo aprovechar las fortalezas de Rust para sistemas distribuidos.

Patrones de Arquitectura de Microservicios en Rust

La arquitectura de microservicios se ha convertido en el patrón preferido para construir aplicaciones escalables y mantenibles. Las características únicas de Rust—seguridad de memoria, abstracciones de costo cero y excelente rendimiento—lo convierten en una opción ideal para el desarrollo de microservicios. Este artículo explora patrones probados para construir arquitecturas de microservicios robustas usando Rust.

¿Por qué Rust para Microservicios?

Rust ofrece varias ventajas para microservicios:

  • Rendimiento: Rendimiento cercano a C/C++ con seguridad de memoria
  • Concurrencia: Excelente soporte async/await con tokio
  • Eficiencia de Recursos: Bajo uso de memoria y CPU
  • Confiabilidad: Garantías en tiempo de compilación reducen errores en tiempo de ejecución
  • Ecosistema: Rico ecosistema de frameworks web y librerías

Patrones de Arquitectura Core

1. Patrón de Registro de Servicios

Implementa descubrimiento de servicios usando un registro centralizado:

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use tokio::sync::RwLock;
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceInstance {
    pub id: Uuid,
    pub name: String,
    pub host: String,
    pub port: u16,
    pub health_check_url: String,
    pub metadata: HashMap<String, String>,
}

pub struct ServiceRegistry {
    services: RwLock<HashMap<String, Vec<ServiceInstance>>>,
}

impl ServiceRegistry {
    pub fn new() -> Self {
        Self {
            services: RwLock::new(HashMap::new()),
        }
    }

    pub async fn register(&self, instance: ServiceInstance) {
        let mut services = self.services.write().await;
        services
            .entry(instance.name.clone())
            .or_insert_with(Vec::new)
            .push(instance);
    }

    pub async fn discover(&self, service_name: &str) -> Option<Vec<ServiceInstance>> {
        let services = self.services.read().await;
        services.get(service_name).cloned()
    }

    pub async fn deregister(&self, service_name: &str, instance_id: Uuid) {
        let mut services = self.services.write().await;
        if let Some(instances) = services.get_mut(service_name) {
            instances.retain(|instance| instance.id != instance_id);
        }
    }
}

2. Patrón Circuit Breaker

Protege servicios de fallos en cascada:

use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;

#[derive(Debug, PartialEq)]
pub enum CircuitState {
    Closed,
    Open,
    HalfOpen,
}

pub struct CircuitBreaker {
    state: Arc<Mutex<CircuitBreakerState>>,
    failure_threshold: u32,
    timeout: Duration,
}

struct CircuitBreakerState {
    state: CircuitState,
    failure_count: u32,
    last_failure_time: Option<Instant>,
    success_count: u32,
}

impl CircuitBreaker {
    pub fn new(failure_threshold: u32, timeout: Duration) -> Self {
        Self {
            state: Arc::new(Mutex::new(CircuitBreakerState {
                state: CircuitState::Closed,
                failure_count: 0,
                last_failure_time: None,
                success_count: 0,
            })),
            failure_threshold,
            timeout,
        }
    }

    pub async fn call<F, T, E>(&self, operation: F) -> Result<T, CircuitBreakerError<E>>
    where
        F: FnOnce() -> Result<T, E>,
    {
        let mut state = self.state.lock().await;

        match state.state {
            CircuitState::Open => {
                if let Some(last_failure) = state.last_failure_time {
                    if last_failure.elapsed() > self.timeout {
                        state.state = CircuitState::HalfOpen;
                        state.success_count = 0;
                    } else {
                        return Err(CircuitBreakerError::CircuitOpen);
                    }
                }
            }
            CircuitState::HalfOpen => {
                if state.success_count >= 3 {
                    state.state = CircuitState::Closed;
                    state.failure_count = 0;
                }
            }
            CircuitState::Closed => {}
        }

        drop(state);

        match operation() {
            Ok(result) => {
                self.on_success().await;
                Ok(result)
            }
            Err(error) => {
                self.on_failure().await;
                Err(CircuitBreakerError::OperationFailed(error))
            }
        }
    }

    async fn on_success(&self) {
        let mut state = self.state.lock().await;
        match state.state {
            CircuitState::HalfOpen => {
                state.success_count += 1;
            }
            CircuitState::Closed => {
                state.failure_count = 0;
            }
            _ => {}
        }
    }

    async fn on_failure(&self) {
        let mut state = self.state.lock().await;
        state.failure_count += 1;
        state.last_failure_time = Some(Instant::now());

        if state.failure_count >= self.failure_threshold {
            state.state = CircuitState::Open;
        }
    }
}

#[derive(Debug)]
pub enum CircuitBreakerError<E> {
    CircuitOpen,
    OperationFailed(E),
}

3. Comunicación Basada en Eventos

Implementa mensajería asíncrona entre servicios:

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use tokio::sync::broadcast;
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
    pub id: Uuid,
    pub event_type: String,
    pub source: String,
    pub timestamp: chrono::DateTime<chrono::Utc>,
    pub data: serde_json::Value,
}

pub struct EventBus {
    senders: HashMap<String, broadcast::Sender<Event>>,
}

impl EventBus {
    pub fn new() -> Self {
        Self {
            senders: HashMap::new(),
        }
    }

    pub fn subscribe(&mut self, event_type: &str) -> broadcast::Receiver<Event> {
        let sender = self
            .senders
            .entry(event_type.to_string())
            .or_insert_with(|| broadcast::channel(1000).0);

        sender.subscribe()
    }

    pub async fn publish(&mut self, event: Event) -> Result<(), Box<dyn std::error::Error>> {
        if let Some(sender) = self.senders.get(&event.event_type) {
            sender.send(event)?;
        }
        Ok(())
    }
}

// Ejemplo de uso
pub struct OrderService {
    event_bus: EventBus,
}

impl OrderService {
    pub async fn create_order(&mut self, order_data: serde_json::Value) -> Result<(), Box<dyn std::error::Error>> {
        // Lógica para crear pedido aquí

        let event = Event {
            id: Uuid::new_v4(),
            event_type: "order.created".to_string(),
            source: "order-service".to_string(),
            timestamp: chrono::Utc::now(),
            data: order_data,
        };

        self.event_bus.publish(event).await?;
        Ok(())
    }
}

Patrones de Gestión de Datos

Base de Datos por Servicio

Cada microservicio debe poseer sus datos:

use sqlx::{PgPool, Row};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct User {
    pub id: i32,
    pub email: String,
    pub name: String,
}

pub struct UserRepository {
    pool: PgPool,
}

impl UserRepository {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    pub async fn create_user(&self, email: &str, name: &str) -> Result<User, sqlx::Error> {
        let row = sqlx::query!(
            "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name",
            email,
            name
        )
        .fetch_one(&self.pool)
        .await?;

        Ok(User {
            id: row.id,
            email: row.email,
            name: row.name,
        })
    }

    pub async fn find_by_id(&self, id: i32) -> Result<Option<User>, sqlx::Error> {
        let row = sqlx::query!("SELECT id, email, name FROM users WHERE id = $1", id)
            .fetch_optional(&self.pool)
            .await?;

        Ok(row.map(|r| User {
            id: r.id,
            email: r.email,
            name: r.name,
        }))
    }
}

Patrón Saga para Transacciones Distribuidas

Implementa transacciones compensatorias:

use async_trait::async_trait;
use serde_json::Value;
use std::collections::HashMap;

#[async_trait]
pub trait SagaStep {
    async fn execute(&self, context: &mut SagaContext) -> Result<Value, SagaError>;
    async fn compensate(&self, context: &SagaContext) -> Result<(), SagaError>;
}

pub struct SagaContext {
    pub data: HashMap<String, Value>,
    pub completed_steps: Vec<String>,
}

pub struct Saga {
    steps: Vec<(String, Box<dyn SagaStep>)>,
}

impl Saga {
    pub fn new() -> Self {
        Self { steps: Vec::new() }
    }

    pub fn add_step(mut self, name: String, step: Box<dyn SagaStep>) -> Self {
        self.steps.push((name, step));
        self
    }

    pub async fn execute(&self) -> Result<SagaContext, SagaError> {
        let mut context = SagaContext {
            data: HashMap::new(),
            completed_steps: Vec::new(),
        };

        for (step_name, step) in &self.steps {
            match step.execute(&mut context).await {
                Ok(result) => {
                    context.data.insert(step_name.clone(), result);
                    context.completed_steps.push(step_name.clone());
                }
                Err(error) => {
                    // Compensar pasos completados en orden inverso
                    for completed_step_name in context.completed_steps.iter().rev() {
                        if let Some((_, completed_step)) = self
                            .steps
                            .iter()
                            .find(|(name, _)| name == completed_step_name)
                        {
                            let _ = completed_step.compensate(&context).await;
                        }
                    }
                    return Err(error);
                }
            }
        }

        Ok(context)
    }
}

#[derive(Debug)]
pub enum SagaError {
    StepFailed(String),
    CompensationFailed(String),
}

Patrones de Seguridad

Validación de Tokens JWT

use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
    pub sub: String,
    pub exp: usize,
    pub iat: usize,
    pub roles: Vec<String>,
}

pub struct JwtValidator {
    decoding_key: DecodingKey,
    validation: Validation,
}

impl JwtValidator {
    pub fn new(secret: &[u8]) -> Self {
        let mut validation = Validation::new(Algorithm::HS256);
        validation.required_spec_claims = ["exp", "iat", "sub"].iter().map(|s| s.to_string()).collect();

        Self {
            decoding_key: DecodingKey::from_secret(secret),
            validation,
        }
    }

    pub fn validate_token(&self, token: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
        let token_data = decode::<Claims>(token, &self.decoding_key, &self.validation)?;
        Ok(token_data.claims)
    }
}

Monitoreo y Observabilidad

Trazabilidad Distribuida

use tracing::{info, instrument, Span};
use tracing_opentelemetry::OpenTelemetrySpanExt;

pub struct OrderService {
    user_service_client: UserServiceClient,
    payment_service_client: PaymentServiceClient,
}

impl OrderService {
    #[instrument(skip(self))]
    pub async fn process_order(&self, order_id: &str, user_id: &str) -> Result<(), ServiceError> {
        let span = Span::current();
        span.set_attribute("order.id", order_id.to_string());
        span.set_attribute("user.id", user_id.to_string());

        info!("Iniciando procesamiento de pedido para orden: {}", order_id);

        // Validar usuario
        let user = self.user_service_client.get_user(user_id).await?;
        span.set_attribute("user.email", user.email.clone());

        // Procesar pago
        let payment_result = self.payment_service_client
            .process_payment(&user.payment_method)
            .await?;

        span.set_attribute("payment.transaction_id", payment_result.transaction_id);

        info!("Pedido procesado exitosamente: {}", order_id);
        Ok(())
    }
}

Estrategias de Pruebas

Pruebas de Integración con TestContainers

#[cfg(test)]
mod tests {
    use super::*;
    use testcontainers::{clients::Cli, images::postgres::Postgres, Container};
    use sqlx::PgPool;

    async fn setup_database() -> (Cli, Container<'_, Postgres>, PgPool) {
        let docker = Cli::default();
        let postgres_container = docker.run(Postgres::default());

        let connection_string = format!(
            "postgres://postgres:postgres@127.0.0.1:{}/postgres",
            postgres_container.get_host_port_ipv4(5432)
        );

        let pool = PgPool::connect(&connection_string).await.unwrap();

        // Ejecutar migraciones
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();

        (docker, postgres_container, pool)
    }

    #[tokio::test]
    async fn test_user_repository() {
        let (_docker, _container, pool) = setup_database().await;
        let repo = UserRepository::new(pool);

        let user = repo.create_user("test@example.com", "Usuario Prueba").await.unwrap();
        assert_eq!(user.email, "test@example.com");

        let found_user = repo.find_by_id(user.id).await.unwrap().unwrap();
        assert_eq!(found_user.email, user.email);
    }
}

Patrones de Despliegue

Verificaciones de Salud

use axum::{http::StatusCode, response::Json, routing::get, Router};
use serde_json::{json, Value};

async fn health_check() -> Result<Json<Value>, StatusCode> {
    // Verificar conectividad de base de datos, servicios externos, etc.
    Ok(Json(json!({
        "status": "healthy",
        "timestamp": chrono::Utc::now(),
        "version": env!("CARGO_PKG_VERSION")
    })))
}

async fn readiness_check() -> Result<Json<Value>, StatusCode> {
    // Verificar si el servicio está listo para aceptar tráfico
    Ok(Json(json!({
        "status": "ready",
        "timestamp": chrono::Utc::now()
    })))
}

pub fn create_health_router() -> Router {
    Router::new()
        .route("/health", get(health_check))
        .route("/ready", get(readiness_check))
}

Conclusión

Construir microservicios con Rust requiere una consideración cuidadosa de los patrones arquitectónicos. Los patrones cubiertos en este artículo proporcionan una base sólida para crear microservicios escalables y mantenibles:

  1. Descubrimiento de Servicios: Esencial para la comunicación dinámica entre servicios
  2. Circuit Breakers: Previenen fallos en cascada
  3. Arquitectura Basada en Eventos: Permiten acoplamiento débil entre servicios
  4. Patrón Saga: Maneja transacciones distribuidas de manera segura
  5. Seguridad: Implementa autenticación y autorización adecuadas
  6. Observabilidad: Monitorea y rastrea interacciones entre servicios

El sistema de tipos de Rust y las características de rendimiento lo convierten en una excelente opción para microservicios, especialmente cuando se combina con estos patrones arquitectónicos probados.

¿Preguntas o comentarios? No dudes en contactarme en LinkedIn o GitHub.

[cookies-prefs-title]

Usamos cookies para que este sitio funcione, entender el uso del servicio y apoyar acciones de marketing. Política de cookies para más información.