Облачные приложения на Rust: руководство по архитектуре микросервисов

// Типичный микросервис на Rust занимает всего 20-30 МБ памяти
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let app = App::new()
        .service(health_check)
        .service(get_products);
    
    HttpServer::new(move || app.clone())
        .bind("0.0.0.0:8080")?
        .run()
        .await?;
    
    Ok(())
}

// Пример автоматического освобождения ресурсов
// Соединение с базой автоматически закроется в конце блока
{
    let conn = pool.get().await?;
    let result = conn.query("SELECT * FROM products", &[]).await?;
    // Работаем с result...
} // Здесь conn автоматически освобождается, 
  // даже если произошла ошибка

// Базовый шаблон микросервиса на Actix + Tokio
#[actix_web::main] // Это макрос, который использует Tokio под капотом
async fn main() -> std::io::Result<()> {
    let app_state = web::Data::new(AppState {
        db_pool: create_connection_pool().await,
        config: load_configuration(),
    });
 
    HttpServer::new(move || {
        App::new()
            .app_data(app_state.clone())
            .wrap(middleware::Logger::default())
            .service(health::check)
            .service(
                web::scope("/api/v1")
                    .service(products::routes())
                    .service(orders::routes())
            )
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

async fn get_product(id: Uuid, db: &PgPool) -> Result<Product, ServiceError> {
    let product = sqlx::query_as!(
        Product,
        "SELECT * FROM products WHERE id = $1",
        id
    )
    .fetch_optional(db)
    .await
    .map_err(|e| ServiceError::DatabaseError(e.to_string()))?;
    
    product.ok_or(ServiceError::NotFound(format!("Product {}", id)))
}

// Определяем трейт репозитория
trait ProductRepository: Send + Sync {
    async fn find_by_id(&self, id: Uuid) -> Result<Option<Product>, Error>;
    async fn save(&self, product: &Product) -> Result<(), Error>;
}
 
// Сервисный слой работает с абстракцией репозитория
struct ProductService<R: ProductRepository> {
    repository: R,
}
 
impl<R: ProductRepository> ProductService<R> {
    async fn get_product(&self, id: Uuid) -> Result<Product, ServiceError> {
        self.repository.find_by_id(id)
            .await
            .map_err(|e| ServiceError::Repository(e))?
            .ok_or(ServiceError::NotFound(format!("Product {}", id)))
    }
}

src/
  main.rs         # Точка входа
  config.rs       # Конфигурация
  errors.rs       # Типы ошибок
  api/            # API-слой
    mod.rs
    health.rs
    products.rs
  services/       # Сервисный слой
    mod.rs
    product.rs
  repositories/   # Слой данных
    mod.rs
    postgres/
      mod.rs
      product.rs
  domain/         # Доменные модели
    mod.rs
    product.rs

async fn call_user_service(client: &Client, id: Uuid) -> Result<User, Error> {
    static BREAKER: Lazy<CircuitBreaker> = Lazy::new(|| {
        CircuitBreakerBuilder::default()
            .failure_threshold(5)
            .success_threshold(2)
            .wait_duration(Duration::from_secs(10))
            .build()
    });
 
    // Вызов через Circuit Breaker
    BREAKER.call(|| async {
        client.get(&format!("http://user-service/users/{}", id))
            .send()
            .await?
            .json::<User>()
            .await
    }).await
}

// Ограничиваем конкурентные запросы к внешнему сервису
let semaphore = Arc::new(Semaphore::new(10)); // Максимум 10 запросов
 
async fn call_external_api(semaphore: Arc<Semaphore>, data: &Data) -> Result<Response, Error> {
    // Получаем разрешение из семафора
    let permit = semaphore.acquire().await?;
    
    // Выполняем запрос
    let response = reqwest::Client::new()
        .post("https://external-api.com/endpoint")
        .json(data)
        .send()
        .await?;
    
    // Разрешение автоматически освобождается когда permit выходит из области видимости
    drop(permit); 
    
    Ok(response)
}

// Модель конфигурации
#[derive(Debug, Deserialize)]
struct Settings {
    server: ServerSettings,
    database: DatabaseSettings,
    redis: Option<RedisSettings>,
}
 
// Загрузка конфигурации из разных источников
fn load_configuration() -> Result<Settings, ConfigError> {
    let mut settings = config::Config::default();
    
    // Порядок важен - каждый следующий источник перезаписывает предыдущий
    settings.merge(config::File::with_name("config/default"))?;
    settings.merge(config::File::with_name("config/local").required(false))?;
    
    // Переменные окружения имеют высший приоритет
    settings.merge(config::Environment::with_prefix("APP"))?;
    
    settings.try_into()
}

// Создаем главный объект приложения
struct Application {
    product_service: ProductService<PostgresRepository>,
    order_service: OrderService<PostgresRepository, EmailNotifier>,
}
 
impl Application {
    fn new(config: &Settings) -> Self {
        let db_pool = create_db_pool(&config.database);
        let email_client = create_email_client(&config.email);
        
        let product_repo = PostgresRepository::new(db_pool.clone());
        let order_repo = PostgresRepository::new(db_pool);
        let notifier = EmailNotifier::new(email_client);
        
        Self {
            product_service: ProductService::new(product_repo),
            order_service: OrderService::new(order_repo, notifier),
        }
    }
}

// Мок-репозиторий для тестов
#[derive(Default)]
struct MockProductRepo {
    products: Mutex<HashMap<Uuid, Product>>,
}
 
impl ProductRepository for MockProductRepo {
    async fn find_by_id(&self, id: Uuid) -> Result<Option<Product>, Error> {
        let products = self.products.lock().unwrap();
        Ok(products.get(&id).cloned())
    }
    
    async fn save(&self, product: &Product) -> Result<(), Error> {
        let mut products = self.products.lock().unwrap();
        products.insert(product.id, product.clone());
        Ok(())
    }
}
 
#[tokio::test]
async fn test_get_product_success() {
    // Arrange
    let repo = MockProductRepo::default();
    let product = Product { id: Uuid::new_v4(), name: "Test".to_string(), price: 10.0 };
    repo.products.lock().unwrap().insert(product.id, product.clone());
    
    let service = ProductService::new(repo);
    
    // Act
    let result = service.get_product(product.id).await;
    
    // Assert
    assert!(result.is_ok());
    assert_eq!(result.unwrap().name, "Test");
}

async fn create_order(
    order_data: OrderData,
    product_client: &ProductClient,
    payment_client: &PaymentClient,
) -> Result<Order, ServiceError> {
    // 1. Проверяем наличие товара
    let product = product_client
        .reserve_product(order_data.product_id, order_data.quantity)
        .await?;
        
    // 2. Создаем заказ
    let order = Order::new(order_data, product);
    order_repository.save(&order).await?;
    
    // 3. Обрабатываем оплату
    match payment_client.process_payment(&order).await {
        Ok(_) => {
            // Заказ успешно создан и оплачен
            Ok(order)
        }
        Err(e) => {
            // Компенсирующая транзакция - отменяем резервацию товара
            product_client.cancel_reservation(order_data.product_id, order_data.quantity).await?;
            Err(ServiceError::PaymentFailed(e.to_string()))
        }
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .service(health_check)
            .service(
                web::scope("/api/v1")
                    .service(get_products)
                    .service(create_product)
                    .service(update_product)
            )
            .default_service(web::to(|| async { HttpResponse::NotFound() }))
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}
 
#[get("/products")]
async fn get_products(db: web::Data<DbPool>) -> impl Responder {
    match db.execute(|conn| repository::get_all_products(conn)).await {
        Ok(products) => HttpResponse::Ok().json(products),
        Err(e) => {
            log::error!("Failed to get products: {}", e);
            HttpResponse::InternalServerError().finish()
        }
    }
}

#[tokio::main]
async fn main() {
    let db_pool = create_db_pool().await;
 
    let app = Router::new()
        .route("/health", get(health_check))
        .route("/api/v1/products", get(get_products).post(create_product))
        .route("/api/v1/products/:id", put(update_product))
        .with_state(AppState { db: db_pool });
 
    axum::Server::bind(&"0.0.0.0:8080".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}
 
async fn get_products(
    State(state): State<AppState>,
) -> Result<Json<Vec<Product>>, ApiError> {
    let products = sqlx::query_as!(
        Product,
        "SELECT * FROM products"
    )
    .fetch_all(&state.db)
    .await?;
 
    Ok(Json(products))
}

// Модуль маршрутов для продуктов
pub fn product_routes() -> Scope {
    web::scope("/products")
        .service(get_all)
        .service(get_by_id)
        .service(create)
        .service(update)
        .service(delete)
}
 
// Модуль маршрутов для заказов
pub fn order_routes() -> Scope {
    web::scope("/orders")
        .service(get_all)
        .service(get_by_id)
        .service(create)
        .service(cancel)
}

// Tokio
#[tokio::main]
async fn main() {
    let handles: Vec<_> = (0..10).map(|i| {
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            println!("Task {} completed", i);
            i
        })
    }).collect();
    
    for handle in handles {
        let result = handle.await.unwrap();
        println!("Got result: {}", result);
    }
}
 
// async-std
#[async_std::main]
async fn main() {
    let handles: Vec<_> = (0..10).map(|i| {
        async_std::task::spawn(async move {
            async_std::task::sleep(Duration::from_millis(100)).await;
            println!("Task {} completed", i);
            i
        })
    }).collect();
    
    for handle in handles {
        let result = handle.await;
        println!("Got result: {}", result);
    }
}

async fn publish_order(channel: &Channel, order: &Order) -> Result<(), Error> {
    let payload = serde_json::to_vec(order)?;
    
    channel
        .basic_publish(
            "orders-exchange", // exchange
            "new-order",       // routing key
            BasicPublishOptions::default(),
            payload,
            BasicProperties::default()
                .with_delivery_mode(2) // persistent
                .with_content_type("application/json".into()),
        )
        .await?
        .await?; // дважды await для подтверждения публикации
    
    Ok(())
}
 
async fn start_order_consumer(
    channel: &Channel, 
    order_service: Arc<dyn OrderService>
) -> Result<(), Error> {
    channel
        .basic_consume(
            "orders-queue",
            "order-processor",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?
        .set_delegate(move |delivery: Delivery| {
            let order_service = order_service.clone();
            
            async move {
                match delivery {
                    Ok(delivery) => {
                        let order: Order = match serde_json::from_slice(&delivery.data) {
                            Ok(order) => order,
                            Err(e) => {
                                eprintln!("Failed to parse order: {}", e);
                                delivery.nack(BasicNackOptions::default()).await.unwrap();
                                return;
                            }
                        };
                        
                        match order_service.process_order(order).await {
                            Ok(_) => delivery.ack(BasicAckOptions::default()).await.unwrap(),
                            Err(e) => {
                                eprintln!("Failed to process order: {}", e);
                                delivery.nack(BasicNackOptions { requeue: true, ..Default::default() })
                                    .await.unwrap();
                            }
                        }
                    }
                    Err(e) => eprintln!("Failed to receive delivery: {}", e),
                }
            }
        });
    
    Ok(())
}

async fn publish_to_kafka(
    producer: &FutureProducer, 
    topic: &str, 
    order: &Order
) -> Result<(), Error> {
    let payload = serde_json::to_vec(order)?;
    let key = order.id.to_string();
    
    producer.send(
        FutureRecord::to(topic)
            .payload(&payload)
            .key(&key),
        Duration::from_secs(0),
    ).await?;
    
    Ok(())
}

pub struct JwtMiddleware;
 
impl<S> Transform<S, ServiceRequest> for JwtMiddleware
where
    S: Service<ServiceRequest, Response = ServiceResponse, Error = Error> + 'static,
{
    type Response = ServiceResponse;
    type Error = Error;
    type Transform = JwtMiddlewareService<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;
 
    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(JwtMiddlewareService { service }))
    }
}
 
pub struct JwtMiddlewareService<S> {
    service: S,
}
 
impl<S> Service<ServiceRequest> for JwtMiddlewareService<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse, Error = Error> + 'static,
{
    type Response = ServiceResponse;
    type Error = Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
 
    fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.service.poll_ready(ctx)
    }
 
    fn call(&self, req: ServiceRequest) -> Self::Future {
        // Извлекаем токен из заголовка
        let auth_header = req.headers().get("Authorization");
        
        if let Some(auth_header) = auth_header {
            if let Ok(auth_str) = auth_header.to_str() {
                if auth_str.starts_with("Bearer ") {
                    let token = &auth_str[7..];
                    
                    // Проверяем JWT
                    match validate_token(token) {
                        Ok(claims) => {
                            // Сохраняем данные пользователя в запросе
                            req.extensions_mut().insert(claims);
                            let fut = self.service.call(req);
                            return Box::pin(async move {
                                fut.await
                            });
                        }
                        Err(_) => {
                            return Box::pin(async {
                                Ok(ServiceResponse::new(
                                    req.into_parts().0,
                                    HttpResponse::Unauthorized().finish()
                                ))
                            });
                        }
                    }
                }
            }
        }
        
        Box::pin(async {
            Ok(ServiceResponse::new(
                req.into_parts().0,
                HttpResponse::Unauthorized().finish()
            ))
        })
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct Claims {
    sub: String,       // user id
    exp: usize,        // expiration time
    roles: Vec<String>, // user roles
}
 
fn check_permission(claims: &Claims, required_role: &str) -> bool {
    claims.roles.iter().any(|role| role == required_role)
}
 
async fn create_product(
    req: HttpRequest,
    product: web::Json<ProductDto>,
    db: web::Data<DbPool>,
) -> Result<HttpResponse, Error> {
    // Получаем данные пользователя из запроса
    let claims = req.extensions().get::<Claims>().unwrap();
    
    // Проверяем права доступа
    if !check_permission(claims, "product_admin") {
        return Ok(HttpResponse::Forbidden().finish());
    }
    
    // Создаем продукт
    let product_id = create_product_in_db(&db, &product).await?;
    
    Ok(HttpResponse::Created().json(json!({ "id": product_id })))
}

// SQLx с проверкой запросов на этапе компиляции
async fn get_products_by_category(
    pool: &PgPool,
    category_id: i32,
) -> Result<Vec<Product>, sqlx::Error> {
    sqlx::query_as!(
        Product,
        r#"
        SELECT id, name, price, stock, category_id
        FROM products
        WHERE category_id = $1
        "#,
        category_id
    )
    .fetch_all(pool)
    .await
}

async fn transfer_money(
    pool: &PgPool,
    from_account: i32,
    to_account: i32,
    amount: Decimal,
) -> Result<(), Error> {
    let mut tx = pool.begin().await?;
    
    // Проверяем баланс
    let balance = sqlx::query_scalar!(
        "SELECT balance FROM accounts WHERE id = $1 FOR UPDATE",
        from_account
    )
    .fetch_one(&mut tx)
    .await?;
    
    if balance < amount {
        return Err(Error::InsufficientFunds);
    }
    
    // Списываем с одного счета
    sqlx::query!(
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
        amount, from_account
    )
    .execute(&mut tx)
    .await?;
    
    // Зачисляем на другой счет
    sqlx::query!(
        "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
        amount, to_account
    )
    .execute(&mut tx)
    .await?;
    
    // Фиксируем транзакцию
    tx.commit().await?;
    
    Ok(())
}

let pool = PgPoolOptions::new()
    .max_connections(10)        // Максимум 10 соединений
    .min_connections(2)         // Минимум 2 соединения
    .max_lifetime(Duration::from_secs(30 * 60)) // 30 минут
    .idle_timeout(Duration::from_secs(10 * 60)) // 10 минут простоя
    .connect("postgres://user:password@localhost/db")
    .await?;

async fn get_product_with_cache(
    id: Uuid, 
    db_pool: &PgPool,
    redis: &RedisConnection,
) -> Result<Product, Error> {
    // Пытаемся получить из кеша
    let cache_key = format!("product:{}", id);
    let cached: Option<String> = redis.get(&cache_key).await?;
    
    if let Some(json) = cached {
        if let Ok(product) = serde_json::from_str(&json) {
            return Ok(product);
        }
    }
    
    // Если нет в кеше, берем из БД
    let product = sqlx::query_as!(
        Product,
        "SELECT * FROM products WHERE id = $1",
        id
    )
    .fetch_optional(db_pool)
    .await?
    .ok_or(Error::NotFound)?;
    
    // Сохраняем в кеш на 15 минут
    let json = serde_json::to_string(&product)?;
    redis.set_ex(&cache_key, json, 900).await?;
    
    Ok(product)
}

// Сервис, изменяющий данные
async fn update_product(
    id: Uuid,
    data: ProductUpdate,
    db: &PgPool,
    redis: &RedisConnection,
) -> Result<(), Error> {
    // Обновляем в БД
    sqlx::query!(
        "UPDATE products SET name = $1, price = $2 WHERE id = $3",
        data.name, data.price, id
    )
    .execute(db)
    .await?;
    
    // Инвалидируем кеш
    let cache_key = format!("product:{}", id);
    redis.del(&cache_key).await?;
    
    // Уведомляем другие сервисы
    redis.publish(
        "product_updates", 
        serde_json::to_string(&ProductUpdateEvent { id, operation: "UPDATE" })?
    ).await?;
    
    Ok(())
}
 
// Сервис, подписанный на обновления
async fn start_cache_invalidation_listener(redis: RedisConnection) {
    let mut pubsub = redis.into_pubsub();
    pubsub.subscribe("product_updates").await.unwrap();
    
    while let Some(msg) = pubsub.on_message().next().await {
        let payload: String = msg.get_payload().unwrap();
        if let Ok(event) = serde_json::from_str::<ProductUpdateEvent>(&payload) {
            // Инвалидируем локальный кеш
            let cache_key = format!("product:{}", event.id);
            CACHE.remove(&cache_key).await;
        }
    }
}

// Определяем события домена
#[derive(Debug, Clone, Serialize, Deserialize)]
enum OrderEvent {
    Created {
        id: Uuid,
        customer_id: Uuid,
        items: Vec<OrderItem>,
        created_at: DateTime<Utc>,
    },
    PaymentReceived {
        amount: Money,
        transaction_id: String,
        received_at: DateTime<Utc>,
    },
    Shipped {
        tracking_number: String,
        shipped_at: DateTime<Utc>,
    },
    Cancelled {
        reason: String,
        cancelled_at: DateTime<Utc>,
    },
}
 
// Агрегат восстанавливает свое состояние из событий
#[derive(Debug, Default)]
struct Order {
    id: Option<Uuid>,
    customer_id: Option<Uuid>,
    items: Vec<OrderItem>,
    status: OrderStatus,
    payment: Option<Payment>,
    tracking_number: Option<String>,
    created_at: Option<DateTime<Utc>>,
}
 
impl Order {
    fn apply(&mut self, event: OrderEvent) {
        match event {
            OrderEvent::Created { id, customer_id, items, created_at } => {
                self.id = Some(id);
                self.customer_id = Some(customer_id);
                self.items = items;
                self.status = OrderStatus::Created;
                self.created_at = Some(created_at);
            },
            OrderEvent::PaymentReceived { amount, transaction_id, received_at } => {
                self.payment = Some(Payment {
                    amount,
                    transaction_id,
                    received_at,
                });
                self.status = OrderStatus::Paid;
            },
            OrderEvent::Shipped { tracking_number, shipped_at } => {
                self.tracking_number = Some(tracking_number);
                self.status = OrderStatus::Shipped;
            },
            OrderEvent::Cancelled { reason: _, cancelled_at: _ } => {
                self.status = OrderStatus::Cancelled;
            }
        }
    }
}

async fn store_events(
    pool: &PgPool, 
    stream_id: &str, 
    events: Vec<OrderEvent>,
    expected_version: i64,
) -> Result<i64, Error> {
    let mut tx = pool.begin().await?;
    
    // Проверяем версию
    let current_version: i64 = sqlx::query_scalar!(
        "SELECT COALESCE(MAX(version), -1) FROM event_store WHERE stream_id = $1",
        stream_id
    )
    .fetch_one(&mut tx)
    .await?;
    
    if current_version != expected_version {
        return Err(Error::ConcurrencyConflict);
    }
    
    // Сохраняем события
    let mut new_version = expected_version;
    for event in events {
        new_version += 1;
        let event_type = get_event_type(&event);
        let event_data = serde_json::to_value(&event)?;
        
        sqlx::query!(
            r#"
            INSERT INTO event_store (stream_id, version, event_type, event_data, created_at)
            VALUES ($1, $2, $3, $4, $5)
            "#,
            stream_id,
            new_version,
            event_type,
            event_data,
            Utc::now()
        )
        .execute(&mut tx)
        .await?;
    }
    
    tx.commit().await?;
    
    Ok(new_version)
}

// Обработчик команды
async fn handle_create_order(
    cmd: CreateOrderCommand,
    event_store: &EventStore,
    db_pool: &PgPool,
) -> Result<Uuid, Error> {
    // Генерируем ID для нового заказа
    let order_id = Uuid::new_v4();
    
    // Создаем событие
    let event = OrderEvent::Created {
        id: order_id,
        customer_id: cmd.customer_id,
        items: cmd.items,
        created_at: Utc::now(),
    };
    
    // Сохраняем событие
    let stream_id = format!("order:{}", order_id);
    event_store.store_events(&stream_id, vec![event.clone()], -1).await?;
    
    // Обновляем проекцию для чтения
    update_order_projection(db_pool, &event).await?;
    
    Ok(order_id)
}
 
// Обновление проекции для чтения
async fn update_order_projection(
    pool: &PgPool,
    event: &OrderEvent,
) -> Result<(), Error> {
    match event {
        OrderEvent::Created { id, customer_id, items, created_at } => {
            let total = calculate_total(items);
            
            sqlx::query!(
                r#"
                INSERT INTO order_read_model 
                (id, customer_id, status, total_amount, created_at)
                VALUES ($1, $2, $3, $4, $5)
                "#,
                id,
                customer_id,
                "created",
                total,
                created_at
            )
            .execute(pool)
            .await?;
            
            // Сохраняем элементы заказа
            for item in items {
                sqlx::query!(
                    r#"
                    INSERT INTO order_items_read_model
                    (order_id, product_id, quantity, price)
                    VALUES ($1, $2, $3, $4)
                    "#,
                    id,
                    item.product_id,
                    item.quantity,
                    item.price
                )
                .execute(pool)
                .await?;
            }
        },
        // Обработка других событий...
        _ => { /* ... */ }
    }
    
    Ok(())
}

async fn get_product_details(
    product_id: String,
    product_client: &mut ProductServiceClient<Channel>,
) -> Result<ProductDetails, Error> {
    let request = Request::new(GetProductRequest {
        id: product_id.clone(),
    });
    
    let response = product_client
        .get_product(request)
        .await?
        .into_inner();
    
    Ok(ProductDetails {
        id: response.id,
        name: response.name,
        price: response.price,
        stock: response.stock,
    })
}

async fn fetch_order_details(
  client: &Client,
  order_id: &str,
  auth_token: &str,
) -> Result<Order, Error> {
  let response = client
      .get(&format!("http://order-service/api/orders/{}", order_id))
      .header("Authorization", format!("Bearer {}", auth_token))
      .send()
      .await?;
      
  if !response.status().is_success() {
      return Err(Error::ServiceUnavailable(
          format!("Failed to fetch order: {}", response.status())
      ));
  }
  
  let order = response.json::<Order>().await?;
  Ok(order)
}

async fn call_with_retry<F, Fut, T, E>(
  operation: F,
  max_attempts: usize,
  base_delay: Duration,
) -> Result<T, E>
where
  F: Fn() -> Fut,
  Fut: Future<Output = Result<T, E>>,
  E: std::fmt::Debug,
{
  let mut attempts = 0;
  let mut delay = base_delay;
  
  loop {
      attempts += 1;
      match operation().await {
          Ok(value) => return Ok(value),
          Err(e) => {
              if attempts >= max_attempts {
                  return Err(e);
              }
              
              log::warn!("Operation failed: {:?}. Retrying in {:?}...", e, delay);
              tokio::time::sleep(delay).await;
              
              // Экспоненциальная задержка с небольшим случайным компонентом
              delay = delay * 2 + Duration::from_millis(rand::random::<u64>() % 100);
          }
      }
  }
}

let order = call_with_retry(
  || fetch_order_details(client, &order_id, &token),
  3,
  Duration::from_millis(100)
).await?;

struct CircuitBreaker {
  failures: AtomicUsize,
  last_failure: AtomicU64,
  threshold: usize,
  timeout: Duration,
}
 
impl CircuitBreaker {
  fn new(threshold: usize, timeout: Duration) -> Self {
      Self {
          failures: AtomicUsize::new(0),
          last_failure: AtomicU64::new(0),
          threshold,
          timeout,
      }
  }
  
  async fn call<F, T, E>(&self, operation: F) -> Result<T, E>
  where
      F: Future<Output = Result<T, E>>,
  {
      // Проверяем, открыт ли прерыватель
      let failures = self.failures.load(Ordering::SeqCst);
      let last_failure = self.last_failure.load(Ordering::SeqCst);
      let now = SystemTime::now()
          .duration_since(UNIX_EPOCH)
          .unwrap()
          .as_secs();
      
      if failures >= self.threshold && (now - last_failure) < self.timeout.as_secs() {
          // Прерыватель открыт, отклоняем запрос
          return Err(Error::CircuitOpen)?;
      }
      
      // Выполняем операцию
      match operation.await {
          Ok(value) => {
              // Успешная операция - сбрасываем счетчик ошибок
              self.failures.store(0, Ordering::SeqCst);
              Ok(value)
          }
          Err(e) => {
              // Увеличиваем счетчик ошибок
              self.failures.fetch_add(1, Ordering::SeqCst);
              self.last_failure.store(now, Ordering::SeqCst);
              Err(e)
          }
      }
  }
}

#[derive(Debug, Deserialize)]
struct Config {
  server: ServerConfig,
  database: DatabaseConfig,
  services: ServicesConfig,
  auth: AuthConfig,
}
 
#[derive(Debug, Deserialize)]
struct ServicesConfig {
  product_service_url: String,
  order_service_url: String,
  retry_attempts: usize,
  timeout_ms: u64,
}
 
fn load_config() -> Result<Config, ConfigError> {
  let mut builder = config::Config::builder();
  
  // Базовая конфигурация
  builder = builder.add_source(config::File::with_name("config/default"));
  
  // Конфигурация окружения (dev, test, prod)
  let env = std::env::var("APP_ENV").unwrap_or_else(|_| "development".to_string());
  builder = builder.add_source(config::File::with_name(&format!("config/{}", env)).required(false));
  
  // Локальные переопределения для разработчика
  builder = builder.add_source(config::File::with_name("config/local").required(false));
  
  // Переменные окружения с префиксом APP_
  builder = builder.add_source(config::Environment::with_prefix("APP").separator("__"));
  
  // Собираем и конвертируем
  builder.build()?.try_deserialize()
}

async fn process_order(
  ctx: &Context,
  order_id: &str,
  db: &PgPool,
  client: &Client,
) -> Result<(), Error> {
  // Создаем дочерний спан
  let span = tracer
      .span_builder("process_order")
      .with_attributes(vec![KeyValue::new("order_id", order_id.to_string())])
      .start_with_context(&tracer, ctx);
  
  // Выполняем код внутри этого спана
  let _guard = ctx.clone().attach_span(span);
  
  // Получаем данные заказа
  let order = get_order_from_db(ctx, order_id, db).await?;
  
  // Резервируем товары
  reserve_products(ctx, &order, client).await?;
  
  // Обрабатываем оплату
  process_payment(ctx, &order, client).await?;
  
  Ok(())
}

fn inject_context(ctx: &Context, request: &mut RequestBuilder) -> RequestBuilder {
  let mut headers = HeaderMap::new();
  let propagator = TraceContextPropagator::new();
  propagator.inject_context(ctx, &mut headers);
  
  for (key, value) in headers.iter() {
      if let Ok(value_str) = value.to_str() {
          request = request.header(key.as_str(), value_str);
      }
  }
  
  request
}

fn setup_logging() {
  // Форматирование в JSON для лучшей интеграции с ELK/Loki
  let formatting_layer = tracing_subscriber::fmt::layer()
      .json()
      .with_target(true)
      .with_level(true)
      .with_current_span(true);
  
  // Установка фильтра уровня логирования из переменной окружения
  let filter_layer = EnvFilter::try_from_env("LOG_LEVEL")
      .unwrap_or_else(|_| EnvFilter::new("info"));
  
  tracing_subscriber::registry()
      .with(filter_layer)
      .with(formatting_layer)
      .init();
  
  debug!("Logging initialized");
}

fn init_tracer() -> Result<Tracer, Error> {
  // Инициализация экспортера Jaeger
  let exporter = opentelemetry_jaeger::new_pipeline()
      .with_service_name("product-service")
      .with_agent_endpoint("jaeger-agent:6831")
      .install_batch(opentelemetry::runtime::Tokio)?;
  
  // Получаем трейсер
  let tracer = opentelemetry::global::tracer("product-service");
  
  // Интегрируем с tracing
  let telemetry = tracing_opentelemetry::layer().with_tracer(tracer.clone());
  tracing_subscriber::registry()
      .with(telemetry)
      .try_init()?;
  
  Ok(tracer)
}

#[derive(Debug, thiserror::Error)]
enum ServiceError {
  #[error("Database error: {0}")]
  Database(#[from] sqlx::Error),
  
  #[error("External service error: {0}")]
  ExternalService(String),
  
  #[error("Not found: {0}")]
  NotFound(String),
}
 
// При обработке ошибки
match result {
  Ok(data) => Ok(data),
  Err(e) => {
      // Логируем с контекстом
      tracing::error!(
          error.type = %std::any::type_name::<E>(),
          error.display = %e,
          order_id = %order_id,
          user_id = %user_id,
          "Failed to process order"
      );
      Err(e.into())
  }
}

use prometheus::{register_counter, register_histogram, Counter, Histogram};
 
// Определяем метрики
lazy_static! {
  static ref HTTP_REQUESTS_TOTAL: Counter = register_counter!(
      "http_requests_total",
      "Total number of HTTP requests"
  ).unwrap();
  
  static ref HTTP_REQUEST_DURATION: Histogram = register_histogram!(
      "http_request_duration_seconds",
      "HTTP request duration in seconds",
      vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
  ).unwrap();
}
 
// Middleware для сбора метрик
async fn metrics_middleware(req: Request, next: Next) -> Response {
  // Увеличиваем счетчик запросов
  HTTP_REQUESTS_TOTAL.inc();
  
  // Засекаем время
  let start = Instant::now();
  
  // Выполняем запрос
  let response = next.run(req).await;
  
  // Записываем длительность
  let duration = start.elapsed().as_secs_f64();
  HTTP_REQUEST_DURATION.observe(duration);
  
  response
}

async fn metrics_handler() -> Result<String, Error> {
  let encoder = TextEncoder::new();
  let mut buffer = Vec::new();
  encoder.encode(&prometheus::gather(), &mut buffer)?;
  Ok(String::from_utf8(buffer)?)
}

// Небезопасный код - нет проверки алгоритма!
let token_data = decode::<Claims>(
    &token,
    &DecodingKey::from_secret(secret.as_bytes()),
    &Validation::default(), // Здесь была ошибка - стандартная валидация разрешает любой алгоритм
)?;

// Безопасный код - явно указываем алгоритм
let mut validation = Validation::new(Algorithm::HS256);
validation.validate_exp = true;
 
let token_data = decode::<Claims>(
    &token,
    &DecodingKey::from_secret(secret.as_bytes()),
    &validation,
)?;

// Фрагмент проблемного кода с unsafe
pub fn get_connection(&self) -> Result<PooledConnection, Error> {
  let mut connections = self.connections.lock().unwrap();
  
  // Извлекаем соединение из пула или создаем новое
  let conn = if let Some(conn) = connections.pop() {
      conn
  } else if self.connection_count.load(Ordering::SeqCst) < self.max_connections {
      self.connection_count.fetch_add(1, Ordering::SeqCst);
      
      // Здесь был опасный код для создания соединения
      match self.connect() {
          Ok(conn) => conn,
          Err(e) => {
              self.connection_count.fetch_sub(1, Ordering::SeqCst);
              return Err(e);
          }
      }
  } else {
      return Err(Error::PoolExhausted);
  };
  
  // Проверка соединения перед возвратом
  unsafe {
      // Здесь был опасный код, который мог привести к двойному освобождению
      // при определенных условиях гонки данных
  }
  
  Ok(PooledConnection::new(conn, self.return_channel.clone()))
}

// Пример мучений с SOAP API
async fn call_legacy_soap_api(xml_payload: &str) -> Result<String, Error> {
  let client = reqwest::Client::new();
  
  let response = client
      .post("https://legacy-service/soap")
      .header("Content-Type", "text/xml; charset=utf-8")
      .header("SOAPAction", ""urn:GetData"")
      .body(xml_payload.to_string())
      .send()
      .await?;
      
  if !response.status().is_success() {
      let error_text = response.text().await?;
      return Err(Error::ExternalService(format!("SOAP error: {}", error_text)));
  }
  
  let response_text = response.text().await?;
  
  // Теперь самое веселое - парсинг XML ответа
  let document = roxmltree::Document::parse(&response_text)?;
  
  // Извлекаем нужные данные из XML с кучей неймспейсов
  // ...а здесь было много боли...
  
  Ok(result)
}

// Пример излишне усложненного кода
type Result<T> = std::result::Result<T, Error>;
type DynHandler<C> = Arc<dyn Fn(C) -> BoxFuture<'static, Result<Response>> + Send + Sync>;
type MiddlewareFunc<C> = Box<dyn Fn(C, Next<C>) -> BoxFuture<'static, Result<Response>> + Send + Sync>;

// Пример реального мониторинга памяти в продакшене:
// Go-сервис vs Rust-сервис (аналогичная функциональность)
// 
// Go:    ~78 МБ при 100 RPS, ~145 МБ при 1000 RPS
// Rust:  ~22 МБ при 100 RPS, ~38 МБ при 1000 RPS