Merge pull request 'Starting impl Sentry and tracing' (#3) from tracing into main

Reviewed-on: #3

.devcontainer/Dockerfile

@@ -11,6 +11,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     curl \
     git \
     build-essential \
+    libssl-dev \
     cmake \
     pkg-config \
     clang \

Cargo.toml

@@ -11,8 +11,14 @@ tokio-util = "0.7"
 futures-util = "0.3"
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
+sentry = { version = "0.48", features = ["tower-axum-matched-path"] }
+sentry-anyhow = "0.48"
 openrouter-rs = "0.10"
 dotenvy = "0.15"
+tower = "0.5"
+tower-http = {version = "0.6", features = ["trace"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features=["env-filter"] }
 axum = "0.8"
 anyhow = "1.0"
 thiserror = "2.0"

src/api.rs

@@ -1,13 +1,19 @@
-use axum::body::{Bytes, to_bytes};
+use anyhow::anyhow;
+use axum::body::{Body, Bytes, to_bytes};
 use axum::extract::{FromRef, FromRequest, State};
+use axum::http::Request;
 use axum::response::IntoResponse;
 use axum::routing::{get, post};
 use axum::{Json, Router};
 use hmac::{Hmac, KeyInit, Mac};
 use reqwest::StatusCode;
+use sentry::integrations::tower::{NewSentryLayer, SentryHttpLayer};
 use serde_json::Value;
 use sha2::Sha256;
 use subtle::ConstantTimeEq;
+use tower::ServiceBuilder;
+use tower_http::trace::TraceLayer;
+use tracing::{info, instrument};
 
 use crate::consts::{GITEA_EVENT_TYPE_HEADER_NAME, GITEA_SIG_HEADER_NAME, MAX_WEBHOOK_BODY_SIZE};
 use crate::errors::AppError;
@@ -20,9 +26,17 @@ pub async fn start(app_state: AppState) -> anyhow::Result<()> {
     let app = Router::new()
         .route("/", get(root))
         .route("/webhook", post(webhook))
+        .layer(
+            ServiceBuilder::new()
+                .layer(NewSentryLayer::<Request<_>>::new_from_top())
+                .layer(SentryHttpLayer::new())
+                .layer(TraceLayer::new_for_http()),
+        )
         .with_state(app_state);
 
     let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", http_port)).await?;
+    info!("Listening API on port {}", http_port);
+
     axum::serve(listener, app)
         .await
         .map_err(anyhow::Error::from)
@@ -32,15 +46,17 @@ async fn root() -> &'static str {
     "Hi, i'm Herald :)"
 }
 
+#[instrument(skip(app_state), fields(webhook_type), err)]
 async fn webhook(
     State(app_state): State<AppState>,
     WebhookExtract(wb): WebhookExtract,
 ) -> Result<impl IntoResponse, AppError> {
+    tracing::Span::current().record("webhook_type", tracing::field::debug(&wb));
+
     app_state
         .bot_tx
-        .send(wb)
+        .try_send(wb)
-        .await
+        .map_err(|_| AppError::ChannelFullErr)?;
-        .map_err(anyhow::Error::from)?;
 
     Ok((StatusCode::CREATED, "Task started"))
 }
@@ -54,6 +70,7 @@ where
 {
     type Rejection = AppError;
 
+    #[instrument(skip(req, state), err)]
     async fn from_request(req: axum::extract::Request, state: &S) -> Result<Self, Self::Rejection> {
         let app_state = AppState::from_ref(state);
         let headers = req.headers();
@@ -68,6 +85,16 @@ where
             &body_bytes,
         )?;
 
+        let body_str = String::from_utf8_lossy(&body_bytes).into_owned();
+        sentry::configure_scope(|scope| {
+            scope.add_event_processor(move |mut event| {
+                let mut request = event.request.take().unwrap_or_default();
+                request.data = Some(body_str.clone());
+                event.request = Some(request);
+                Some(event)
+            });
+        });
+
         let webhook = parse_webhook(&type_header, &app_state.config.bot_name, &body_bytes)?;
         Ok(WebhookExtract(webhook))
     }

src/bot.rs

@@ -1,11 +1,11 @@
-use serde::Deserialize;
-use std::time::Duration;
-
 use crate::{
     env::EnvConfig,
     gitea::{GiteaAPI, WebhookType},
     open_router::OpenRouterClient,
 };
+use serde::Deserialize;
+use std::{sync::Arc, time::Duration};
+use tracing::{error, info, instrument};
 
 #[derive(Deserialize, Debug)]
 pub struct ReviewResult {
@@ -22,49 +22,13 @@ pub struct ReviewItem {
     pub message: String,
 }
 
-/// Map a filename to a markdown language identifier for syntax highlighting.
+#[derive(Clone)]
-fn lang_from_filename(filename: &str) -> &str {
-    match std::path::Path::new(filename)
-        .extension()
-        .and_then(|e| e.to_str())
-        .unwrap_or("")
-    {
-        "rs" => "rust",
-        "py" => "python",
-        "js" | "mjs" => "javascript",
-        "ts" => "typescript",
-        "jsx" => "jsx",
-        "tsx" => "tsx",
-        "go" => "go",
-        "java" => "java",
-        "kt" | "kts" => "kotlin",
-        "scala" => "scala",
-        "c" | "h" => "c",
-        "cpp" | "cc" | "cxx" | "hpp" | "hxx" => "cpp",
-        "rb" => "ruby",
-        "php" => "php",
-        "swift" => "swift",
-        "sh" | "bash" | "zsh" => "bash",
-        "sql" => "sql",
-        "html" | "htm" => "html",
-        "css" => "css",
-        "scss" | "sass" => "scss",
-        "json" => "json",
-        "yaml" | "yml" => "yaml",
-        "xml" => "xml",
-        "toml" => "toml",
-        "md" | "mdx" => "markdown",
-        "dockerfile" | "Dockerfile" => "dockerfile",
-        "Makefile" => "makefile",
-        _ => "",
-    }
-}
-
 pub struct Bot {
     config: EnvConfig,
     gitea_api: GiteaAPI,
     open_router_client: OpenRouterClient,
     http_client: reqwest::Client,
+    max_concurrent: usize,
 }
 
 impl Bot {
@@ -79,6 +43,7 @@ impl Bot {
                 &config.open_router_model,
                 open_router_timeout,
             )?,
+            max_concurrent: config.bot_max_concurrent,
             config,
             http_client: reqwest::Client::builder()
                 .timeout(Duration::from_secs(gitea_timeout))
@@ -90,14 +55,40 @@ impl Bot {
         &self,
         mut rx: tokio::sync::mpsc::Receiver<WebhookType>,
     ) -> anyhow::Result<()> {
+        info!("Bot started");
+
+        let sem = Arc::new(tokio::sync::Semaphore::new(self.max_concurrent));
+        let mut tasks = tokio::task::JoinSet::new();
+
         while let Some(wb) = rx.recv().await {
-            self.exec(wb).await;
+            // Drain completed tasks to avoid the JoinSet growing unbounded
+            while let Some(res) = tasks.try_join_next() {
+                if let Err(e) = res {
+                    error!("Task panicked: {e}");
+                }
+            }
+
+            info!(queued = rx.len(), active = tasks.len(), "Webhook received");
+            let permit = sem.clone().acquire_owned().await?;
+            let self_clone = self.clone();
+
+            tasks.spawn(async move {
+                self_clone.exec(wb).await;
+                drop(permit);
+            });
         }
 
+        // When all webhook tasks have completed, we can safely exit
+        // properly before returning
+        tasks.join_all().await;
+
+        info!("Bot shutting down, channel closed");
         Ok(())
     }
 
+    #[instrument(skip(self))]
     pub async fn exec(&self, webhook: WebhookType) {
+        tracing::Span::current().record("webhook_type", tracing::field::debug(&webhook));
         let exec_result = match webhook {
             WebhookType::Review(review_payload) => crate::bot_actions::review::exec_review(
                 &self.gitea_api,
@@ -110,8 +101,11 @@ impl Bot {
         .await;
 
         match exec_result {
-            Ok(_) => println!("Task completed"),
+            Ok(_) => info!("Task completed"),
-            Err(err) => println!("{}", err),
+            Err(err) => {
+                error!(%err, "Task error");
+                sentry_anyhow::capture_anyhow(&err);
+            }
         }
     }
 }

src/bot_actions/review.rs

@@ -1,22 +1,30 @@
 use futures_util::stream::TryStreamExt;
 use tokio::io::AsyncReadExt;
 use tokio_util::io::StreamReader;
+use tracing::instrument;
 
 use crate::{
     bot::ReviewResult,
     consts::{BOT_PROCESS_MSG, MAX_DIFF_SIZE, REVIEW_PROMPT},
-    errors::AppError,
     gitea::{GiteaAPI, ReviewPayload},
     open_router::OpenRouterClient,
 };
 
+#[instrument(skip(gitea_api, open_router_client, http_client, review_payload), err)]
 pub async fn exec_review(
     gitea_api: &GiteaAPI,
     open_router_client: &OpenRouterClient,
     http_client: &reqwest::Client,
     model: &str,
     review_payload: ReviewPayload,
-) -> Result<(), AppError> {
+) -> anyhow::Result<()> {
+    tracing::info!(
+        repo = %review_payload.repository.full_name,
+        pr = review_payload.pull_request.number,
+        action = %review_payload.action,
+        "Starting review"
+    );
+
     let new_comment = gitea_api
         .comment(
             &BOT_PROCESS_MSG.replace("{model}", model),
@@ -41,9 +49,12 @@ pub async fn exec_review(
 
         review_result.cost = chat_result.cost;
 
+        let final_review_markdown = review_result_to_markdown(&review_result);
+
         gitea_api
             .post_pull_request_review(
                 &review_result,
+                &final_review_markdown,
                 &review_payload.repository.full_name,
                 review_payload.pull_request.number,
             )
@@ -53,20 +64,22 @@ pub async fn exec_review(
     }
     .await;
 
-    let edit_msg = match bot_result {
+    match bot_result {
-        Ok(bot_result) => review_result_to_markdown(&bot_result),
+        Ok(_) => {
-        Err(e) => format!("Error while reviewing: {}", e),
+            gitea_api
-    };
+                .delete_comment(&review_payload.repository.full_name, new_comment.id)
-
+                .await
-    gitea_api
+        }
-        .edit_comment(
+        Err(e) => {
-            &edit_msg,
+            gitea_api
-            &review_payload.repository.full_name,
+                .edit_comment(
-            new_comment.id,
+                    &format!("Error while reviewing: {}", e),
-        )
+                    &review_payload.repository.full_name,
-        .await?;
+                    new_comment.id,
-
+                )
-    Ok(())
+                .await
+        }
+    }
 }
 
 fn review_result_to_markdown(review_result: &ReviewResult) -> String {

src/env.rs

@@ -1,5 +1,4 @@
 use anyhow::anyhow;
-use dotenvy::dotenv;
 
 #[derive(Clone)]
 pub struct EnvConfig {
@@ -9,20 +8,20 @@ pub struct EnvConfig {
     pub open_router_model: String,
     pub open_router_timeout: u64,
     pub bot_name: String,
+    pub bot_max_concurrent: usize,
     pub gitea_url: String,
     pub gitea_token: String,
     pub gitea_timeout: u64,
 }
 
 pub fn load_config() -> anyhow::Result<EnvConfig> {
-    dotenv().ok();
-
     let http_port = try_get_env("HTTP_PORT")?.parse()?;
     let bot_name = try_get_env("BOT_NAME")?;
     let webhook_secret = try_get_env("WEBHOOK_SIG_HEADER_SECRET")?;
     let open_router_api_key = try_get_env("OPEN_ROUTER_API_KEY")?;
     let open_router_model = try_get_env("OPEN_ROUTER_MODEL")?;
     let open_router_timeout = try_get_env("OPEN_ROUTER_TIMEOUT")?.parse()?;
+    let bot_max_concurrent = try_get_env("BOT_MAX_CONCURRENT")?.parse()?;
     let gitea_url = try_get_env("GITEA_URL")?;
     let gitea_token = try_get_env("GITEA_TOKEN")?;
     let gitea_timeout = try_get_env("GITEA_TIMEOUT")?.parse()?;
@@ -34,13 +33,14 @@ pub fn load_config() -> anyhow::Result<EnvConfig> {
         open_router_api_key,
         open_router_model,
         open_router_timeout,
+        bot_max_concurrent,
         gitea_url,
         gitea_token,
         gitea_timeout,
     })
 }
 
-fn try_get_env(key: &str) -> anyhow::Result<String> {
+pub fn try_get_env(key: &str) -> anyhow::Result<String> {
     let env = std::env::var(key)?;
 
     if env.trim().is_empty() {

src/errors.rs

@@ -1,3 +1,4 @@
+use anyhow::anyhow;
 use axum::response::IntoResponse;
 use reqwest::StatusCode;
 
@@ -21,6 +22,9 @@ pub enum AppError {
     #[error("WebHook have bad action")]
     InvalidActionErr,
 
+    #[error("Channel full")]
+    ChannelFullErr,
+
     #[error(transparent)]
     BadJsonStructErr(#[from] serde_json::Error),
 
@@ -54,10 +58,20 @@ impl IntoResponse for AppError {
                 StatusCode::UNAUTHORIZED,
                 "WebHook sig header is invalid".to_string(),
             ),
-            AppError::Other(_) => (
+            AppError::ChannelFullErr => {
-                StatusCode::INTERNAL_SERVER_ERROR,
+                sentry_anyhow::capture_anyhow(&anyhow!("Max concurrent tasks reached"));
-                "Internal server error".to_string(),
+                (
-            ),
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "Max concurrent tasks reached".to_string(),
+                )
+            }
+            AppError::Other(err) => {
+                sentry_anyhow::capture_anyhow(&err);
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "Internal server error".to_string(),
+                )
+            }
         }
         .into_response()
     }

src/gitea.rs

@@ -2,12 +2,11 @@ use std::time::Duration;
 
 use serde::Deserialize;
 use serde_json::{Value, json};
+use tracing::instrument;
 
-use crate::{
+use crate::{bot::ReviewResult, errors::AppError};
-    bot::{ReviewItem, ReviewResult},
-    errors::AppError,
-};
 
+#[derive(Clone)]
 pub struct GiteaAPI {
     base_url: String,
     client: reqwest::Client,
@@ -30,6 +29,7 @@ impl GiteaAPI {
         })
     }
 
+    #[instrument(skip(self), err)]
     pub async fn comment(
         &self,
         body: &str,
@@ -57,6 +57,7 @@ impl GiteaAPI {
         res.json::<Comment>().await.map_err(anyhow::Error::from)
     }
 
+    #[instrument(skip(self), err)]
     pub async fn edit_comment(
         &self,
         body: &str,
@@ -84,9 +85,30 @@ impl GiteaAPI {
         Ok(())
     }
 
+    #[instrument(skip(self), err)]
+    pub async fn delete_comment(&self, full_name: &str, comment_id: u64) -> anyhow::Result<()> {
+        let url = format!(
+            "{}/api/v1/repos/{}/issues/comments/{}",
+            self.base_url, full_name, comment_id
+        );
+
+        let res = self.client.delete(url).send().await?;
+
+        if !res.status().is_success() {
+            return Err(anyhow::anyhow!(
+                "Failed to delete comment: {}",
+                res.status()
+            ));
+        }
+
+        Ok(())
+    }
+
+    #[instrument(skip(self, review_result), err)]
     pub async fn post_pull_request_review(
         &self,
         review_result: &ReviewResult,
+        final_comment: &str,
         full_name: &str,
         index: u64,
     ) -> anyhow::Result<()> {
@@ -95,7 +117,7 @@ impl GiteaAPI {
             self.base_url, full_name, index
         );
 
-        let comments = &&review_result
+        let comments = &review_result
             .reviews
             .iter()
             .filter(|r| r.line.is_some())
@@ -117,7 +139,7 @@ impl GiteaAPI {
             .post(url)
             .json(&json!({
                 "event": "COMMENT",
-                "body": review_result.comment,
+                "body": final_comment,
                 "comments": comments
             }))
             .send()

src/main.rs

@@ -1,4 +1,7 @@
 use crate::{bot::Bot, gitea::WebhookType, state::AppState};
+use dotenvy::dotenv;
+use tracing::{info, warn};
+use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 
 mod api;
 mod bot;
@@ -10,13 +13,49 @@ mod gitea;
 mod open_router;
 mod state;
 
-#[tokio::main]
+fn main() -> anyhow::Result<()> {
-async fn main() -> anyhow::Result<()> {
+    dotenv().ok();
+
+    tracing_subscriber::registry()
+        .with(fmt::layer())
+        .with(
+            EnvFilter::try_from_default_env() // lit RUST_LOG depuis l'env
+                .unwrap_or_else(|_| EnvFilter::new("info")),
+        )
+        .init();
+
+    let _sentry_guard = if let Ok(sentry_dsn) = env::try_get_env("SENTRY_DSN") {
+        info!("Initialize sentry");
+
+        Some(sentry::init((
+            sentry_dsn,
+            sentry::ClientOptions {
+                release: sentry::release_name!(),
+                send_default_pii: true,
+                ..Default::default()
+            },
+        )))
+    } else {
+        warn!("SENTRY_DSN not set, sentry will not be initialized");
+        None
+    };
+
+    tokio::runtime::Runtime::new()?.block_on(run())
+}
+
+async fn run() -> anyhow::Result<()> {
     let config = env::load_config()?;
+
+    info!(
+        port = config.http_port,
+        model = %config.open_router_model,
+        gitea_url = %config.gitea_url,
+        bot_name = %config.bot_name,
+        "Starting Herald"
+    );
+
     let bot = Bot::new(config.clone())?;
-
+    let (tx, rx) = tokio::sync::mpsc::channel::<WebhookType>(config.bot_max_concurrent * 2);
-    let (tx, rx) = tokio::sync::mpsc::channel::<WebhookType>(1);
-
     let app_state = AppState { bot_tx: tx, config };
 
     tokio::try_join!(bot.start(rx), api::start(app_state))?;

src/open_router.rs

@@ -1,12 +1,14 @@
 use std::time::Duration;
 
 use openrouter_rs::{Message, api::chat::ChatCompletionRequest};
+use tracing::instrument;
 
 pub struct ChatResult {
     pub message: String,
     pub cost: Option<f64>,
 }
 
+#[derive(Clone)]
 pub struct OpenRouterClient {
     client: openrouter_rs::OpenRouterClient,
     model: String,
@@ -27,6 +29,7 @@ impl OpenRouterClient {
         })
     }
 
+    #[instrument(skip(self), err)]
     pub async fn chat(&self, msg: &str) -> anyhow::Result<ChatResult> {
         let request = ChatCompletionRequest::builder()
             .model(&self.model)