مدیریت خطا در Async در Rust؛ از یک تابع ساده تا یک Health Monitor واقعی

Error: request failed

use anyhow::{Context, Result};
use reqwest::Client;

pub async fn fetch_status(url: &str) -> Result<u16> {
    let client = Client::builder()
        .user_agent("drunkleen-async/0.1")
        .build()
        .context("failed to build HTTP client")?;

    let response = client
        .get(url)
        .send()
        .await
        .with_context(|| format!("request to {url} failed"))?;

    // Expose only the HTTP status so the caller decides what to do.
    Ok(response.status().as_u16())
}

#[tokio::main]
async fn main() -> Result<()> {
    let url = "https://drunkleen.com";
    let status = fetch_status(url).await?;
    println!("status for {url}: {}", status);
    Ok(())
}

use thiserror::Error;

#[derive(Debug, Error)]
pub enum MonitorError {
    #[error("network issue: {0}")]
    Network(#[from] reqwest::Error),
    #[error("timed out after {0:?}")]
    Timeout(std::time::Duration),
    #[error("unexpected status {status} for {url}")]
    UnexpectedStatus { status: u16, url: String },
}

[package]
name = "async-errors"
edition = "2024"

[dependencies]
anyhow = "1"
thiserror = "1"
reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }

use anyhow::{Context, Result};
use tokio::{
    task::JoinSet,
    time::{timeout, Duration},
};

pub async fn gather(urls: &[String]) -> Result<Vec<u16>> {
    let mut set = JoinSet::new();

    for url in urls.iter().cloned() {
        set.spawn(async move {
            let status = timeout(Duration::from_secs(3), crate::fetch_status(&url))
                .await
                .context("task aborted because of timeout")??;
            Ok::<u16, anyhow::Error>(status)
        });
    }

    let mut statuses = Vec::with_capacity(urls.len());
    while let Some(res) = set.join_next().await {
        match res {
            Ok(Ok(code)) => statuses.push(code),
            Ok(Err(err)) => eprintln!("worker failed: {err:#}"),
            Err(join_err) => eprintln!("task panicked: {join_err}"),
        }
    }

    Ok(statuses)
}

use anyhow::Result;
use tokio::{select, sync::mpsc};

pub async fn watch_with_alerts(url: String) -> Result<()> {
    let (tx, mut rx) = mpsc::channel(4);
    let worker = tokio::spawn(async move {
        fetch_status(&url).await?;
        Ok::<_, anyhow::Error>(())
    });

    select! {
        biased;
        res = worker => {
            res??;
        }
        msg = rx.recv() => {
            if let Some(alert) = msg {
                anyhow::bail!("alert channel said: {alert}");
            }
        }
    }
    Ok(())
}

use tokio::time::{sleep, Duration};
use anyhow::Result;

pub async fn retry_fetch_status(url: &str) -> Result<u16> {
    let mut attempts = 0;
    let mut delay = Duration::from_millis(200);

    loop {
        match fetch_status(url).await {
            Ok(code) => return Ok(code),
            Err(err) if attempts < 3 => {
                eprintln!(
                    "attempt={} url={} failed: {err:#}; retrying in {:?}",
                    attempts + 1,
                    url,
                    delay
                );
                sleep(delay).await;
                delay *= 2;
                attempts += 1;
            }
            Err(err) => return Err(err),
        }
    }
}

use tracing::{error, info};
use tracing_subscriber::{fmt, EnvFilter};

pub fn init_tracing() {
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into());
    fmt().with_env_filter(filter).init();
}

use anyhow::Result;

pub async fn monitored_fetch(url: &str) -> Result<u16> {
    info!(target = "monitor", %url, "starting fetch");
    match fetch_status(url).await {
        Ok(code) => {
            info!(target = "monitor", %url, code, "fetch completed");
            Ok(code)
        }
        Err(err) => {
            error!(target = "monitor", %url, error = %err, "fetch failed");
            Err(err)
        }
    }
}

[package]
name = "leen-health"
edition = "2024"

[dependencies]
anyhow = "1"
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "1"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }

src/
 ├── main.rs
 ├── monitor.rs
 └── config.rs

use anyhow::{Context, Result};
use serde::Deserialize;

#[derive(Debug, Deserialize, Clone)]
pub struct Endpoint {
    pub name: String,
    pub url: String,
    pub expected_status: u16,
}

#[derive(Debug, Deserialize, Clone)]
pub struct Config {
    pub endpoints: Vec<Endpoint>,
}

impl Config {
    pub fn load(path: &str) -> Result<Self> {
        let data = std::fs::read_to_string(path)
            .with_context(|| format!("cannot read {path}"))?;
        Ok(serde_json::from_str(&data)
            .with_context(|| format!("{path} is not valid JSON"))?)
    }
}

mod config;
mod monitor;

use anyhow::Result;
use clap::Parser;
use config::Config;
use monitor::run_monitor;
use tracing_subscriber::{fmt, EnvFilter};

#[derive(Debug, Parser)]
struct Cli {
    #[arg(short, long, default_value = "endpoints.json")]
    config: String,
    #[arg(long, default_value_t = 3)]
    retries: u8,
}

#[tokio::main]
async fn main() -> Result<()> {
    fmt().with_env_filter(EnvFilter::from_default_env()).init();
    let cli = Cli::parse();
    let cfg = Config::load(&cli.config)?;
    run_monitor(cfg, cli.retries).await
}

use anyhow::{Context, Result};
use reqwest::Client;
use serde::Serialize;
use std::time::Duration;
use tokio::{
    sync::mpsc,
    task::JoinSet,
    time::{sleep, timeout},
};
use tracing::{error, info, warn};

#[derive(Debug, Serialize)]
struct Report {
    name: String,
    status: String,
    code: Option<u16>,
    message: Option<String>,
}

pub async fn run_monitor(config: crate::config::Config, retries: u8) -> Result<()> {
    let client = Client::builder()
        .user_agent("drunkleen-health/0.1")
        .build()
        .context("failed to build reqwest client")?;

    let (tx, mut rx) = mpsc::channel::<Report>(16);
    let mut set = JoinSet::new();

    for endpoint in config.endpoints {
        let tx = tx.clone();
        let client = client.clone();
        set.spawn(async move {
            let mut attempt = 0;
            let mut delay = Duration::from_millis(400);
            loop {
                match timeout(Duration::from_secs(4), client.get(&endpoint.url).send()).await {
                    Err(_) => {
                        warn!(target = "monitor", name = %endpoint.name, "request timed out");
                        if attempt >= retries.into() {
                            tx.send(Report {
                                name: endpoint.name.clone(),
                                status: "timeout".into(),
                                code: None,
                                message: Some("request exceeded 4s".into()),
                            })
                            .await
                            .context("failed to send timeout report")?;
                            break;
                        }
                    }
                    Ok(Err(err)) => {
                        error!(target = "monitor", name = %endpoint.name, %err, "network error");
                        tx.send(Report {
                            name: endpoint.name.clone(),
                            status: "network_error".into(),
                            code: None,
                            message: Some(err.to_string()),
                        })
                        .await
                        .context("failed to send network report")?;
                        break;
                    }
                    Ok(Ok(response)) => {
                        let status = response.status().as_u16();
                        if status == endpoint.expected_status {
                            tx.send(Report {
                                name: endpoint.name.clone(),
                                status: "ok".into(),
                                code: Some(status),
                                message: None,
                            })
                            .await
                            .context("failed to send ok report")?;
                        } else {
                            tx.send(Report {
                                name: endpoint.name.clone(),
                                status: "mismatch".into(),
                                code: Some(status),
                                message: Some(format!(
                                    "expected {}, got {}",
                                    endpoint.expected_status, status
                                )),
                            })
                            .await
                            .context("failed to send mismatch report")?;
                        }
                        break;
                    }
                }
                attempt += 1;
                sleep(delay).await;
                delay *= 2;
            }
            Ok::<_, anyhow::Error>(())
        });
    }

    drop(tx);

    let mut reports = Vec::new();
    while let Some(report) = rx.recv().await {
        info!(
            target = "monitor",
            name = %report.name,
            status = %report.status,
            "report received"
        );
        println!("{} => {}", report.name, report.status);
        reports.push(report);
    }

    while let Some(res) = set.join_next().await {
        res??;
    }

    println!("\nJSON summary:\n{}", serde_json::to_string_pretty(&reports)?);
    Ok(())
}

{
  "endpoints": [
    { "name": "fast", "url": "https://httpbin.org/status/200", "expected_status": 200 },
    { "name": "slow", "url": "https://httpbin.org/delay/5", "expected_status": 200 },
    { "name": "broken", "url": "https://httpbin.org/status/503", "expected_status": 200 }
  ]
}

cargo run -- --config endpoints.json --retries 2

fast => ok
slow => timeout
broken => mismatch

JSON summary:
[
  {
    "name": "fast",
    "status": "ok",
    "code": 200,
    "message": null
  },
  {
    "name": "slow",
    "status": "timeout",
    "code": null,
    "message": "request exceeded 4s"
  },
  {
    "name": "broken",
    "status": "mismatch",
    "code": 503,
    "message": "expected 200, got 503"
  }
]