diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index d28c98e..44eff67 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -1,4 +1,4 @@ -FROM rust:1.98-trixie +FROM rust:1.99-trixie ARG USERNAME=dev ARG USER_UID=1000 @@ -15,6 +15,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ cmake \ pkg-config \ clang \ + valkey-server \ + valkey-tools \ && rm -rf /var/lib/apt/lists/* RUN groupadd --gid ${USER_GID:-1000} $USERNAME \ diff --git a/.woodpecker/release.yml b/.woodpecker/release.yml index 812b4d8..4dcbf08 100644 --- a/.woodpecker/release.yml +++ b/.woodpecker/release.yml @@ -10,8 +10,9 @@ steps: - /data/woodpecker-builds:/data commands: - echo $DOCKER_PASSWORD | buildah login docker.io -u $DOCKER_USERNAME --password-stdin - - chmod +x scripts/build.sh + - chmod +x scripts/build.sh scripts/build-website.sh - bash scripts/build.sh + - bash scripts/build-website.sh environment: DOCKER_USERNAME: from_secret: docker_username diff --git a/.woodpecker/tests.yml b/.woodpecker/tests.yml index ec1e054..fa79918 100644 --- a/.woodpecker/tests.yml +++ b/.woodpecker/tests.yml @@ -4,19 +4,19 @@ when: steps: - name: fmt - image: rust:1.98 + image: rust:1.99 commands: - rustup component add rustfmt - cargo fmt --all -- --check - name: clippy - image: rust:1.98 + image: rust:1.99 commands: - rustup component add clippy - cargo clippy --workspace --all-targets --all-features -- -D warnings - name: test - image: rust:1.98 + image: rust:1.99 commands: - cargo test --workspace --all-targets diff --git a/Cargo.lock b/Cargo.lock index cfc72d7..fe89039 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -189,12 +189,75 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "apalis" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "504d52557a16b7b202941660f339c9182910d498cac40f93d15f6b31e6bef290" +dependencies = [ + "apalis-core", + "futures", + "pin-project-lite", + "serde", + "thiserror", + "tower", + "tracing", + "tracing-futures", +] + +[[package]] +name = "apalis-core" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3948d2051795c769e6a44a46549879af39d5e8e4fb113e0011cf99f7cd21b657" +dependencies = [ + "futures", + "futures-timer", + "pin-project-lite", + "serde", + "serde_json", + "thiserror", + "tower", + "ulid", +] + +[[package]] +name = "apalis-redis" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "871367e3dfff2accda824ccc6e4f46f8f155c865065a450c466a7f2835541d55" +dependencies = [ + "apalis-core", + "chrono", + "futures", + "log", + "redis", + "serde", + "thiserror", + "tokio", +] + +[[package]] +name = "arc-swap" +version = "1.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" +dependencies = [ + "rustversion", +] + [[package]] name = "atomic-waker" version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + [[package]] name = "axum" version = "0.8.9" @@ -247,6 +310,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", +] + [[package]] name = "backtrace" version = "0.3.76" @@ -381,6 +453,32 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "num-traits", + "serde", + "windows-link", +] + +[[package]] +name = "combine" +version = "4.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +dependencies = [ + "bytes", + "futures-core", + "memchr", + "pin-project-lite", + "tokio", + "tokio-util", +] + [[package]] name = "convert_case" version = "0.10.0" @@ -621,6 +719,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -637,6 +750,17 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.34" @@ -666,12 +790,19 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" +[[package]] +name = "futures-timer" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968" + [[package]] name = "futures-util" version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -929,6 +1060,30 @@ dependencies = [ "tower-service", ] +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "icu_collections" version = "2.3.0" @@ -1281,12 +1436,40 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + [[package]] name = "objc2" version = "0.6.4" @@ -1715,6 +1898,31 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redis" +version = "0.32.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "014cc767fefab6a3e798ca45112bccad9c6e0e218fbd49720042716c73cfef44" +dependencies = [ + "arc-swap", + "backon", + "bytes", + "cfg-if", + "combine", + "futures-channel", + "futures-util", + "itoa", + "num-bigint", + "percent-encoding", + "pin-project-lite", + "ryu", + "sha1_smol", + "socket2", + "tokio", + "tokio-util", + "url", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2149,6 +2357,12 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2210,9 +2424,11 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" [[package]] name = "steep-server" -version = "1.2.0" +version = "1.3.0" dependencies = [ "anyhow", + "apalis", + "apalis-redis", "axum", "devcontainer-rs", "dotenvy", @@ -2220,6 +2436,7 @@ dependencies = [ "hex", "metrics", "metrics-exporter-prometheus", + "redis", "reqwest", "ring", "sentry", @@ -2445,6 +2662,7 @@ dependencies = [ "pin-project-lite", "sync_wrapper", "tokio", + "tokio-util", "tower-layer", "tower-service", "tracing", @@ -2530,6 +2748,15 @@ dependencies = [ "valuable", ] +[[package]] +name = "tracing-futures" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2" +dependencies = [ + "tracing", +] + [[package]] name = "tracing-log" version = "0.2.0" @@ -2565,6 +2792,16 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand", + "web-time", +] + [[package]] name = "uname" version = "0.1.1" @@ -2777,6 +3014,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-root-certs" version = "1.0.9" @@ -2808,6 +3055,41 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "windows-link" version = "0.2.1" @@ -2823,6 +3105,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index aeaa9f3..ee4b5e9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,9 @@ hex = "0.4" bytes = "1.1" metrics = "0.24" metrics-exporter-prometheus = { version = "0.18", default-features = false, features = ["http-listener"] } +apalis = { version = "0.7", features = ["catch-panic", "limit"] } +apalis-redis = "0.7" +redis = { version = "0.32", features = ["tokio-comp", "connection-manager"] } [profile.release] debug = 1 diff --git a/README.md b/README.md index 2a00232..7a35eaf 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -![Steep — Let your code steep](banniere_readme.png) +![Steep — Let your code steep](public/banniere_readme.png) # Steep @@ -38,6 +38,7 @@ Steep reads its configuration from environment variables (a `.env` file is suppo | `GITEA_URL` | Base URL of your Gitea instance | | `GITEA_TOKEN` | Gitea API token | | `GITEA_TIMEOUT` | Gitea API request timeout in seconds | +| `REDIS_URL` | Redis connection URL used for the durable review queue (e.g. `redis://127.0.0.1:6379`) | | `METRICS_BIND_ADDR` | *(optional)* Bind address for the Prometheus metrics endpoint (e.g. `0.0.0.0:9100`). If unset, the metrics exporter is disabled. | | `SENTRY_DSN` | *(optional)* Sentry DSN for error tracking | | `RUST_LOG` | *(optional)* Log level, defaults to `info` | @@ -128,8 +129,9 @@ Set `METRICS_BIND_ADDR` (e.g. `0.0.0.0:9100`) to enable it. The metrics are then |---|---|---| | `steep_webhooks_received_total` | counter | Total webhooks received (label: `event_type`) | | `steep_webhooks_duplicate_total` | counter | Webhooks rejected as duplicates (label: `event_type`) | -| `steep_webhooks_channel_full_total` | counter | Webhooks dropped because the bot channel was full (label: `event_type`) | +| `steep_webhooks_enqueue_failed_total` | counter | Webhooks dropped because enqueueing to Redis failed (label: `event_type`) | | `steep_bot_tasks_active` | gauge | Bot tasks currently in progress | +| `steep_bot_tasks_skipped_total` | counter | Reviews skipped because they were already done (label: `event_type`) | | `steep_bot_tasks_completed_total` | counter | Bot tasks completed successfully (label: `event_type`) | | `steep_bot_tasks_failed_total` | counter | Bot tasks that failed (label: `event_type`) | | `steep_openrouter_cost_cents_total` | counter | Total OpenRouter cost in cents (divide by 100 for USD) | diff --git a/crates/steep-server/Cargo.toml b/crates/steep-server/Cargo.toml index 4165ef4..b6a466a 100644 --- a/crates/steep-server/Cargo.toml +++ b/crates/steep-server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "steep-server" -version = "1.2.0" +version = "1.3.0" edition = "2024" [dependencies] @@ -25,4 +25,7 @@ hex = { workspace = true } metrics = { workspace = true } metrics-exporter-prometheus = { workspace = true } devcontainer-rs = { path = "../devcontainer-rs" } -tempfile = "3" \ No newline at end of file +tempfile = "3" +apalis = { workspace = true } +apalis-redis = { workspace = true } +redis = { workspace = true } diff --git a/crates/steep-server/src/api.rs b/crates/steep-server/src/api.rs index 64ca4f2..844c4ba 100644 --- a/crates/steep-server/src/api.rs +++ b/crates/steep-server/src/api.rs @@ -14,7 +14,10 @@ use tracing::{info, instrument}; use tokio_util::sync::CancellationToken; +use apalis::prelude::Storage; + use crate::consts::{GITEA_EVENT_TYPE_HEADER_NAME, GITEA_SIG_HEADER_NAME, MAX_WEBHOOK_BODY_SIZE}; +use crate::dedup::{Dedup, WEBHOOK_TTL}; use crate::errors::AppError; use crate::gitea::WebhookType; use crate::metrics; @@ -61,15 +64,22 @@ async fn webhook( metrics::webhook_received(event_type); let event_id = wb.event_id(); - if app_state.bot.check_and_mark(event_id).await { + let webhook_key = Dedup::webhook_key(event_id); + if !app_state + .dedup + .claim(&webhook_key, WEBHOOK_TTL) + .await + .map_err(anyhow::Error::from)? + { metrics::webhook_duplicate(event_type); return Err(AppError::AlreadyProcessedErr); } - if app_state.bot_tx.try_send(wb).is_err() { - app_state.bot.unmark(event_id).await; - metrics::webhook_channel_full(event_type); - return Err(AppError::ChannelFullErr); + let mut storage = app_state.storage.clone(); + if let Err(err) = storage.push(wb).await { + app_state.dedup.release(&webhook_key).await.ok(); + metrics::webhook_enqueue_failed(event_type); + return Err(AppError::EnqueueErr(anyhow::Error::from(err))); } Ok((StatusCode::CREATED, "Task started")) @@ -109,7 +119,7 @@ where }); }); - let webhook = parse_webhook(&type_header, &app_state.bot.name(), &body_bytes)?; + let webhook = parse_webhook(&type_header, &app_state.bot_name, &body_bytes)?; Ok(WebhookExtract(webhook)) } } diff --git a/crates/steep-server/src/bot.rs b/crates/steep-server/src/bot.rs index 596e1c8..fdb8c1c 100644 --- a/crates/steep-server/src/bot.rs +++ b/crates/steep-server/src/bot.rs @@ -1,12 +1,13 @@ use crate::{ + dedup::{Dedup, REVIEW_TTL}, gitea::{GiteaAPI, WebhookType}, metrics, open_router::OpenRouterClient, sandbox::SandboxConfig, }; +use apalis::prelude::{Data, Error, Monitor, WorkerBuilder, WorkerBuilderExt, WorkerFactoryFn}; +use apalis_redis::RedisStorage; use serde::{Deserialize, Deserializer}; -use std::{collections::HashSet, sync::Arc}; -use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use tracing::{error, info, instrument}; @@ -120,87 +121,64 @@ where #[derive(Clone)] pub struct Bot { - bot_name: String, gitea_api: GiteaAPI, open_router_client: OpenRouterClient, max_concurrent: usize, open_router_model: String, sandbox: SandboxConfig, - actions_handled: Arc>>, + dedup: Dedup, } impl Bot { pub fn new( - bot_name: String, gitea_api: GiteaAPI, open_router_client: OpenRouterClient, max_concurrent: usize, open_router_model: String, sandbox: SandboxConfig, + dedup: Dedup, ) -> Self { Self { - bot_name, gitea_api, open_router_client, max_concurrent, open_router_model, sandbox, - actions_handled: Arc::new(Mutex::new(HashSet::new())), + dedup, } } - pub fn name(&self) -> String { - self.bot_name.clone() - } - + /// Consomme la file Redis jusqu'à l'annulation de `shutdown`. + /// + /// Les jobs en cours sont laissés se terminer, et ceux qui n'ont pas encore + /// été acquittés restent dans Redis : ils seront repris au prochain démarrage. pub async fn start( &self, - mut rx: tokio::sync::mpsc::Receiver, + storage: RedisStorage, shutdown: CancellationToken, ) -> anyhow::Result<()> { info!("Bot started"); - let sem = Arc::new(tokio::sync::Semaphore::new(self.max_concurrent)); - let mut tasks = tokio::task::JoinSet::new(); + let worker = WorkerBuilder::new("steep-review") + .concurrency(self.max_concurrent) + .data(self.clone()) + .backend(storage) + .build_fn(handle_review); - loop { - let wb = tokio::select! { - biased; - _ = shutdown.cancelled() => break, - msg = rx.recv() => match msg { - Some(wb) => wb, - None => break, - }, - }; - - 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(); - metrics::increment_task_active(); - - tasks.spawn(async move { - self_clone.exec(wb).await; - drop(permit); - metrics::decrement_task_active(); - }); - } - - tasks.join_all().await; + Monitor::new() + .register(worker) + .run_with_signal(async move { + shutdown.cancelled().await; + Ok(()) + }) + .await?; info!("Bot shutting down complete"); Ok(()) } #[instrument(skip(self, webhook), fields(repo, pr))] - pub async fn exec(&self, webhook: WebhookType) { - let event_type_str = webhook.event_type_str(); - + pub async fn exec(&self, webhook: WebhookType) -> anyhow::Result<()> { match &webhook { WebhookType::Review(p) => { tracing::Span::current().record("repo", &p.repository.full_name); @@ -210,111 +188,55 @@ impl Bot { let tools = crate::sandbox::tools::for_webhook(&webhook); - let exec_result = match webhook { - WebhookType::Review(review_payload) => crate::bot_actions::review::exec_review( - &self.gitea_api, - &self.open_router_client, - &self.open_router_model, - &self.sandbox, - tools, - review_payload, - ), - } - .await; - - match exec_result { - Ok(_) => { - metrics::task_completed(event_type_str); - info!("Task completed"); - } - Err(err) => { - metrics::task_failed(event_type_str); - error!(%err, "Task error"); - sentry_anyhow::capture_anyhow(&err); + match webhook { + WebhookType::Review(review_payload) => { + crate::bot_actions::review::exec_review( + &self.gitea_api, + &self.open_router_client, + &self.open_router_model, + &self.sandbox, + tools, + review_payload, + ) + .await } } } - - pub async fn check_and_mark(&self, event_id: u64) -> bool { - let mut action_handled_lock = self.actions_handled.lock().await; - - !action_handled_lock.insert(event_id) - } - - pub async fn unmark(&self, event_id: u64) { - let mut action_handled_lock = self.actions_handled.lock().await; - - action_handled_lock.remove(&event_id); - } } -#[cfg(test)] -mod tests { - use super::*; +async fn handle_review(webhook: WebhookType, bot: Data) -> Result<(), Error> { + let event_type = webhook.event_type_str(); + let review_key = Dedup::review_key(webhook.event_id()); - fn make_actions_handled() -> Arc>> { - Arc::new(Mutex::new(HashSet::new())) + match bot.dedup.is_marked(&review_key).await { + Ok(true) => { + metrics::task_skipped(event_type); + info!("Review already done, skipping"); + return Ok(()); + } + Ok(false) => {} + Err(err) => error!(%err, "Dedup lookup failed, reviewing anyway"), } - async fn check_and_mark(actions_handled: &Arc>>, event_id: u64) -> bool { - let mut lock = actions_handled.lock().await; - !lock.insert(event_id) + metrics::increment_task_active(); + let result = bot.exec(webhook).await; + metrics::decrement_task_active(); + + match result { + Ok(()) => { + metrics::task_completed(event_type); + info!("Task completed"); + + if let Err(err) = bot.dedup.mark(&review_key, REVIEW_TTL).await { + error!(%err, "Failed to mark review as done"); + } + } + Err(err) => { + metrics::task_failed(event_type); + error!(%err, "Task error"); + sentry_anyhow::capture_anyhow(&err); + } } - async fn unmark(actions_handled: &Arc>>, event_id: u64) { - let mut lock = actions_handled.lock().await; - lock.remove(&event_id); - } - - #[tokio::test] - async fn test_check_and_mark_first_call_returns_false() { - let actions_handled = make_actions_handled(); - assert!(!check_and_mark(&actions_handled, 1).await); - } - - #[tokio::test] - async fn test_check_and_mark_second_call_returns_true() { - let actions_handled = make_actions_handled(); - check_and_mark(&actions_handled, 1).await; - assert!(check_and_mark(&actions_handled, 1).await); - } - - #[tokio::test] - async fn test_check_and_mark_different_ids_are_independent() { - let actions_handled = make_actions_handled(); - check_and_mark(&actions_handled, 1).await; - assert!(!check_and_mark(&actions_handled, 2).await); - } - - #[tokio::test] - async fn test_unmark_allows_reprocessing() { - let actions_handled = make_actions_handled(); - check_and_mark(&actions_handled, 1).await; - unmark(&actions_handled, 1).await; - assert!(!check_and_mark(&actions_handled, 1).await); - } - - #[tokio::test] - async fn test_unmark_nonexistent_id_is_noop() { - let actions_handled = make_actions_handled(); - unmark(&actions_handled, 99).await; - assert!(!check_and_mark(&actions_handled, 99).await); - } - - #[tokio::test] - async fn test_check_and_mark_concurrent_same_id() { - let actions_handled = make_actions_handled(); - let actions_handled2 = Arc::clone(&actions_handled); - - let t1 = tokio::spawn(async move { check_and_mark(&actions_handled, 42).await }); - let t2 = tokio::spawn(async move { check_and_mark(&actions_handled2, 42).await }); - - let (r1, r2) = tokio::join!(t1, t2); - let results = [r1.unwrap(), r2.unwrap()]; - - // exactement un seul des deux doit retourner false (non traité) - assert_eq!(results.iter().filter(|&&r| !r).count(), 1); - // l'autre doit retourner true (déjà traité) - assert_eq!(results.iter().filter(|&&r| r).count(), 1); - } + Ok(()) } diff --git a/crates/steep-server/src/dedup.rs b/crates/steep-server/src/dedup.rs new file mode 100644 index 0000000..b414859 --- /dev/null +++ b/crates/steep-server/src/dedup.rs @@ -0,0 +1,120 @@ +use std::time::Duration; + +use redis::aio::ConnectionManager; + +pub const WEBHOOK_TTL: Duration = Duration::from_secs(24 * 60 * 60); +pub const REVIEW_TTL: Duration = Duration::from_secs(7 * 24 * 60 * 60); + +#[derive(Clone)] +pub struct Dedup { + conn: ConnectionManager, +} + +impl Dedup { + pub fn new(conn: ConnectionManager) -> Self { + Self { conn } + } + + pub fn webhook_key(event_id: u64) -> String { + format!("steep:webhook:{event_id}") + } + + pub fn review_key(event_id: u64) -> String { + format!("steep:review:{event_id}") + } + + pub async fn claim(&self, key: &str, ttl: Duration) -> redis::RedisResult { + let set: Option = redis::cmd("SET") + .arg(key) + .arg(1) + .arg("NX") + .arg("EX") + .arg(ttl.as_secs()) + .query_async(&mut self.conn.clone()) + .await?; + + Ok(set.is_some()) + } + + pub async fn mark(&self, key: &str, ttl: Duration) -> redis::RedisResult<()> { + let _: String = redis::cmd("SET") + .arg(key) + .arg(1) + .arg("EX") + .arg(ttl.as_secs()) + .query_async(&mut self.conn.clone()) + .await?; + + Ok(()) + } + + pub async fn is_marked(&self, key: &str) -> redis::RedisResult { + let exists: i64 = redis::cmd("EXISTS") + .arg(key) + .query_async(&mut self.conn.clone()) + .await?; + + Ok(exists > 0) + } + + pub async fn release(&self, key: &str) -> redis::RedisResult<()> { + let _: i64 = redis::cmd("DEL") + .arg(key) + .query_async(&mut self.conn.clone()) + .await?; + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn webhook_key_is_namespaced() { + assert_eq!(Dedup::webhook_key(42), "steep:webhook:42"); + } + + #[test] + fn review_key_is_namespaced() { + assert_eq!(Dedup::review_key(42), "steep:review:42"); + } + + #[test] + fn the_two_keys_for_the_same_event_differ() { + assert_ne!(Dedup::webhook_key(7), Dedup::review_key(7)); + } + + async fn redis_dedup() -> Dedup { + let url = std::env::var("REDIS_URL").expect("REDIS_URL must be set"); + Dedup::new(apalis_redis::connect(url).await.expect("connect to redis")) + } + + #[tokio::test] + #[ignore = "requires a running Redis (REDIS_URL)"] + async fn claim_then_claim_is_a_duplicate() { + let dedup = redis_dedup().await; + let key = "steep:test:claim"; + dedup.release(key).await.unwrap(); + + assert!(dedup.claim(key, Duration::from_secs(30)).await.unwrap()); + assert!(!dedup.claim(key, Duration::from_secs(30)).await.unwrap()); + + dedup.release(key).await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires a running Redis (REDIS_URL)"] + async fn mark_then_is_marked() { + let dedup = redis_dedup().await; + let key = "steep:test:mark"; + dedup.release(key).await.unwrap(); + + assert!(!dedup.is_marked(key).await.unwrap()); + dedup.mark(key, Duration::from_secs(30)).await.unwrap(); + assert!(dedup.is_marked(key).await.unwrap()); + + dedup.release(key).await.unwrap(); + } +} diff --git a/crates/steep-server/src/env.rs b/crates/steep-server/src/env.rs index a7752de..771c490 100644 --- a/crates/steep-server/src/env.rs +++ b/crates/steep-server/src/env.rs @@ -11,6 +11,7 @@ pub struct EnvConfig { pub gitea_url: String, pub gitea_token: String, pub gitea_timeout: u64, + pub redis_url: String, pub metrics_bind_addr: Option, pub sandbox_max_iterations: usize, } @@ -25,6 +26,7 @@ pub fn load_config() -> anyhow::Result { 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()?; + let redis_url = try_get_env("REDIS_URL")?; let metrics_bind_addr = std::env::var("METRICS_BIND_ADDR").ok(); let sandbox_max_iterations = std::env::var("SANDBOX_MAX_ITERATIONS") .ok() @@ -41,6 +43,7 @@ pub fn load_config() -> anyhow::Result { gitea_url, gitea_token, gitea_timeout, + redis_url, metrics_bind_addr, sandbox_max_iterations, }) diff --git a/crates/steep-server/src/errors.rs b/crates/steep-server/src/errors.rs index 589854b..b06bda2 100644 --- a/crates/steep-server/src/errors.rs +++ b/crates/steep-server/src/errors.rs @@ -1,4 +1,3 @@ -use anyhow::anyhow; use axum::response::IntoResponse; use reqwest::StatusCode; @@ -22,8 +21,8 @@ pub enum AppError { #[error("WebHook have bad action")] InvalidActionErr, - #[error("Channel full")] - ChannelFullErr, + #[error("Could not enqueue task")] + EnqueueErr(#[source] anyhow::Error), #[error("Already processed")] AlreadyProcessedErr, @@ -62,11 +61,11 @@ impl IntoResponse for AppError { "WebHook sig header is invalid".to_string(), ), AppError::AlreadyProcessedErr => (StatusCode::OK, "Already processed".to_string()), - AppError::ChannelFullErr => { - sentry_anyhow::capture_anyhow(&anyhow!("Max concurrent tasks reached")); + AppError::EnqueueErr(err) => { + sentry_anyhow::capture_anyhow(&err); ( StatusCode::SERVICE_UNAVAILABLE, - "Max concurrent tasks reached".to_string(), + "Could not enqueue task".to_string(), ) } AppError::Other(err) => { diff --git a/crates/steep-server/src/gitea.rs b/crates/steep-server/src/gitea.rs index 7dbbfcc..e3e3112 100644 --- a/crates/steep-server/src/gitea.rs +++ b/crates/steep-server/src/gitea.rs @@ -1,7 +1,7 @@ use std::time::Duration; use futures_util::stream::TryStreamExt; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use tokio::io::AsyncReadExt; use tokio_util::io::StreamReader; @@ -281,7 +281,7 @@ async fn read_capped(response: reqwest::Response) -> anyhow::Result { Ok(String::from_utf8_lossy(&buf).into_owned()) } -#[derive(Debug)] +#[derive(Debug, Serialize, Deserialize)] pub enum WebhookType { Review(ReviewPayload), } @@ -300,7 +300,7 @@ impl WebhookType { } } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Serialize, Debug)] pub struct ReviewPayload { pub action: String, pub pull_request: PullRequest, @@ -308,13 +308,13 @@ pub struct ReviewPayload { pub comment: Comment, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Serialize, Debug)] pub struct PullRequest { pub number: u64, pub title: String, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Serialize, Debug)] pub struct Comment { pub id: u64, pub body: String, @@ -325,7 +325,7 @@ pub struct User { pub login: String, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Serialize, Debug)] pub struct Repository { pub full_name: String, pub clone_url: String, diff --git a/crates/steep-server/src/main.rs b/crates/steep-server/src/main.rs index b18de22..3558457 100644 --- a/crates/steep-server/src/main.rs +++ b/crates/steep-server/src/main.rs @@ -2,6 +2,7 @@ use std::time::Duration; use crate::{ bot::Bot, + dedup::Dedup, gitea::{GiteaAPI, WebhookType}, open_router::OpenRouterClient, sandbox::SandboxConfig, @@ -18,6 +19,7 @@ mod api; mod bot; mod bot_actions; mod consts; +mod dedup; mod env; mod errors; mod gitea; @@ -99,19 +101,24 @@ async fn run() -> anyhow::Result<()> { ); } + let redis_conn = apalis_redis::connect(config.redis_url.clone()).await?; + let storage: apalis_redis::RedisStorage = + apalis_redis::RedisStorage::new(redis_conn.clone()); + let dedup = Dedup::new(redis_conn); + let bot = Bot::new( - gitea_user.login, gitea_api, open_router_client, config.bot_max_concurrent, config.open_router_model.clone(), sandbox, + dedup.clone(), ); - let (tx, rx) = tokio::sync::mpsc::channel::(config.bot_max_concurrent * 2); let app_state = AppState { - bot_tx: tx, - bot: bot.clone(), + storage: storage.clone(), + bot_name: gitea_user.login.clone(), + dedup, config, }; @@ -136,7 +143,7 @@ async fn run() -> anyhow::Result<()> { tokio::select! { result = async { tokio::try_join!( - bot.start(rx, shutdown.clone()), + bot.start(storage, shutdown.clone()), api::start(app_state, shutdown.clone()), signal ) diff --git a/crates/steep-server/src/metrics.rs b/crates/steep-server/src/metrics.rs index e23c36e..ff1771c 100644 --- a/crates/steep-server/src/metrics.rs +++ b/crates/steep-server/src/metrics.rs @@ -10,8 +10,8 @@ pub fn webhook_duplicate(event_type: &str) { counter!("steep_webhooks_duplicate_total", "event_type" => event_type.to_string()).increment(1); } -pub fn webhook_channel_full(event_type: &str) { - counter!("steep_webhooks_channel_full_total", "event_type" => event_type.to_string()) +pub fn webhook_enqueue_failed(event_type: &str) { + counter!("steep_webhooks_enqueue_failed_total", "event_type" => event_type.to_string()) .increment(1); } @@ -23,6 +23,10 @@ pub fn decrement_task_active() { gauge!("steep_bot_tasks_active").decrement(1.0); } +pub fn task_skipped(event_type: &str) { + counter!("steep_bot_tasks_skipped_total", "event_type" => event_type.to_string()).increment(1); +} + pub fn task_completed(event_type: &str) { counter!("steep_bot_tasks_completed_total", "event_type" => event_type.to_string()) .increment(1); @@ -48,15 +52,20 @@ pub fn describe() { "Webhooks rejected as duplicates" ); describe_counter!( - "steep_webhooks_channel_full_total", + "steep_webhooks_enqueue_failed_total", Unit::Count, - "Webhooks dropped because the bot channel was full" + "Webhooks dropped because enqueueing to Redis failed" ); describe_gauge!( "steep_bot_tasks_active", Unit::Count, "Bot tasks currently in progress" ); + describe_counter!( + "steep_bot_tasks_skipped_total", + Unit::Count, + "Reviews skipped because they were already done" + ); describe_counter!( "steep_bot_tasks_completed_total", Unit::Count, diff --git a/crates/steep-server/src/state.rs b/crates/steep-server/src/state.rs index 7ef7b47..a6857d7 100644 --- a/crates/steep-server/src/state.rs +++ b/crates/steep-server/src/state.rs @@ -1,8 +1,11 @@ -use crate::{bot::Bot, env::EnvConfig, gitea::WebhookType}; +use apalis_redis::RedisStorage; + +use crate::{dedup::Dedup, env::EnvConfig, gitea::WebhookType}; #[derive(Clone)] pub struct AppState { - pub bot_tx: tokio::sync::mpsc::Sender, - pub bot: Bot, + pub storage: RedisStorage, + pub bot_name: String, + pub dedup: Dedup, pub config: EnvConfig, } diff --git a/logo.png b/logo.png new file mode 100644 index 0000000..ba1b3cf Binary files /dev/null and b/logo.png differ diff --git a/public/Caddyfile b/public/Caddyfile new file mode 100644 index 0000000..9fecb4d --- /dev/null +++ b/public/Caddyfile @@ -0,0 +1,5 @@ +:80 { + root * /usr/share/caddy + encode zstd gzip + file_server +} diff --git a/public/Containerfile b/public/Containerfile new file mode 100644 index 0000000..f07906b --- /dev/null +++ b/public/Containerfile @@ -0,0 +1,4 @@ +FROM caddy:2-alpine + +COPY Caddyfile /etc/caddy/Caddyfile +COPY index.html favicon.png banniere_readme.png /usr/share/caddy/ diff --git a/banniere_readme.png b/public/banniere_readme.png similarity index 100% rename from banniere_readme.png rename to public/banniere_readme.png diff --git a/public/favicon.png b/public/favicon.png new file mode 100644 index 0000000..e7017a4 Binary files /dev/null and b/public/favicon.png differ diff --git a/public/index.html b/public/index.html new file mode 100644 index 0000000..a7670af --- /dev/null +++ b/public/index.html @@ -0,0 +1,644 @@ + + + + + + + + + + Steep — Let your code steep + + + + +
+ +
+ +
+
+
+ Steep — Let your code steep +
+
+
+

AI code review · built for Gitea

+

Code reviews with the full picture.

+
+
+

Steep explores pull requests inside an isolated Dev Container, then leaves focused, line-by-line feedback right where your team works.

+
+ Get started + How it works +
+
+
+
+ +
+
+
+

Thoughtful by design

+

Useful reviews. No noisy pipeline.

+

Bring an AI reviewer into your Gitea workflow with a single comment.

+
+
+
+ +

Reviews on demand

+

Mention Steep in a pull-request comment and get an AI review streamed back to the PR. No CI hook to wire, no pipeline to learn.

+
+
+ +

Sandboxed by design

+

Every review runs in an ephemeral Dev Container with read-only tools. The network is cut after setup, and the container is removed when the review ends.

+
+
+ +

Lightweight & observable

+

A lean Rust service designed for fast reviews, with Prometheus metrics and Sentry error tracking.

+
+
+
+
+ +
+
+
+

From comment to context

+
+

Your code stays in its environment.

+

Steep gives the model tools to inspect the repository instead of sending the source code in the prompt.

+
+ Set up Steep +
+
+
+ 01 +

Steep hears your request

A Gitea pull-request comment triggers a review when you mention the bot.

+
+
+ 02 +

A disposable workspace spins up

Steep clones the pull-request head and starts the repository's Dev Container—or a safe default when none is configured.

+
+
+ 03 +

The model explores, read-only

It can inspect files with tools such as ls, read_file, and grep. Setup hooks run first; then network access is removed.

+
+
+ 04 +

Feedback lands on the changed lines

Steep posts actionable findings with severity labels, plus a concise summary of what the PR does well.

+
+
+
+
+ +
+
+
+

What’s brewing

+

The roadmap

+

From safer reviews to an assistant that can help move work forward.

+
+
    +
  • 1.2Sandboxed pull-request reviews.
    • +
  • 1.2.1A default Dev Container when a repository has no configuration.
    • +
  • 1.3Type-safe routing between review and implementation requests.
    • +
  • 2.0Implement requested changes from issues or existing PRs, then create or update a PR automatically.
    • +
  • 2.1Project instructions from AGENTS.md and separate review/implementation models.
    • +
  • 3.0Automatically investigate and fix project errors reported by Sentry.
    • +
+
+
+ +
+
+
+

A few minutes to get steeping

+

Bring Steep to your Gitea.

+

Follow these steps in order. You will need a Rust toolchain, a reachable Docker or Podman daemon, an OpenRouter API key, and a Gitea account for Steep.

+
+
+
+

1. Install and configure

+
    +
  1. + Clone the source + 
    git clone https://gitea.qpismont.fr/qpismont/herald.git
+cd herald
    +
    2. +
  2. + Create a Gitea token +

    Create a token for the account Steep should act as. Give it access to the repositories it will review, including permission to read pull requests and publish comments/reviews. Steep uses this token to identify the account; you will use that account's username when asking for a review.

    +
    3. +
  3. + Configure every required setting + 
    cp .env.example .env
+$EDITOR .env
    +

    Set all nine required variables:

    +
      +
    • GITEA_URL — base URL, e.g. https://gitea.example.com (no /api/v1)
      • +
    • GITEA_TOKEN — the token created above
      • +
    • GITEA_TIMEOUT — API timeout in seconds
      • +
    • OPEN_ROUTER_API_KEY — your OpenRouter key
      • +
    • OPEN_ROUTER_MODEL — an available OpenRouter model ID
      • +
    • OPEN_ROUTER_TIMEOUT — request timeout in seconds
      • +
    • BOT_MAX_CONCURRENT — maximum simultaneous reviews
      • +
    • HTTP_PORT — web server port; use 3001 for the commands below
      • +
    • WEBHOOK_SIG_HEADER_SECRET — generate this in the next step
      • +
    +

    Keep the example's numeric values for the timeouts and concurrency, or replace them with positive integers. Do not commit your completed .env.

    +
    4. +
  4. + Generate and configure the webhook secret + 
    openssl rand -hex 32
    +

    Copy the output into WEBHOOK_SIG_HEADER_SECRET in .env. You will enter the exact same value in Gitea in the next step.

    +
    5. +
  5. + Start the daemon, build, and launch Steep + 
    podman system service --time=0 &
+export DOCKER_HOST=unix:///run/user/$(id -u)/podman/podman.sock
+cargo build --release -p steep-server
+./target/release/steep-server
    +

    Keep this process running. With Docker, omit the Podman command and use its socket; the default is unix:///var/run/docker.sock. Optional settings such as metrics and Sentry can be added later.

    +
    6. +
+ Explore the source on Gitea +
+ +
+

2. Add the Gitea webhook

+
    +
  1. + Choose an address Gitea can reach +

    In the repository's Settings → Webhooks, add a Gitea webhook. Set the target to Steep's public/reachable address and port:

    +
    http://steep.example.com:3001/webhook
    Replace the host and scheme with yours. Do not use localhost unless Gitea and Steep share that network.
    +
    2. +
  2. + Select the event and set the secret +

    Enable at least Pull Request Comment. Paste the generated value into Gitea's Secret field; it must exactly match WEBHOOK_SIG_HEADER_SECRET.

    +
    3. +
  3. + Allow the connection +

    Make sure Gitea can connect to Steep on port 3001 (or your configured HTTP_PORT), including through any firewall or reverse proxy. Steep must also be able to reach Gitea's API, OpenRouter, and the container daemon.

    +
    4. +
  4. + Trigger a review +

    Open a pull request and add a comment that starts with @username, using the username of the Gitea account behind GITEA_TOKEN. For example:

    + 
    @steep review this PR
    +

    The mention must be at the very start of the comment. If nothing happens, check the webhook's delivery history in Gitea and Steep's logs.

    +
    5. +
+
+
+
+
+
+ +
+
+ STEEP · Let your code steep. + Built with Rust · Source on Gitea +
+
+ + + diff --git a/scripts/build-website.sh b/scripts/build-website.sh new file mode 100755 index 0000000..4697daf --- /dev/null +++ b/scripts/build-website.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +set -euo pipefail + +IMAGE="docker.io/tintounn/steep-website" + +if [ -z "${CI_COMMIT_TAG:-}" ]; then + echo "Error: CI_COMMIT_TAG is not set" >&2 + exit 1 +fi + +TAG="${CI_COMMIT_TAG}" +IMAGE_REF="${IMAGE}:${TAG}" + +echo "Building ${IMAGE_REF}..." +buildah build \ + --file public/Containerfile \ + --tag "${IMAGE_REF}" \ + public + +echo "Pushing ${IMAGE_REF}..." +buildah push "${IMAGE_REF}" + +echo "Done: ${IMAGE_REF}"