prepare first release with graceful shutdown + containerfile + push to

hub script

.dockerignore

@@ -0,0 +1,4 @@
+target/
+.env
+.devcontainer/
+docs/

Cargo.lock

@@ -786,7 +786,7 @@ checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
 [[package]]
 name = "herald"
-version = "0.1.0"
+version = "1.0.0"
 dependencies = [
  "anyhow",
  "axum",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "herald"
-version = "0.1.0"
+version = "1.0.0"
 edition = "2024"
 
 [dependencies]

Containerfile

@@ -0,0 +1,12 @@
+FROM rust:1.96 as builder
+
+WORKDIR /app
+COPY . .
+RUN cargo build --release
+
+
+FROM debian:trixie-slim
+
+WORKDIR /app
+COPY --from=builder /app/target/release/herald .
+CMD [ "./herald" ]

scripts/build.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+IMAGE="tintounn/herald"
+
+if [ -z "${CI_COMMIT_TAG:-}" ]; then
+    echo "Error: CI_COMMIT_TAG is not set" >&2
+    exit 1
+fi
+
+TAG="${CI_COMMIT_TAG}"
+
+echo "Building ${IMAGE}:${TAG}..."
+buildah build \
+    --file Containerfile \
+    --tag "docker.io/${IMAGE}:${TAG}" \
+    .
+
+echo "Pushing ${IMAGE}:${TAG}..."
+buildah push "${IMAGE}:${TAG}"
+
+echo "Done: ${IMAGE}:${TAG}"

src/api.rs

@@ -1,5 +1,4 @@
-use anyhow::anyhow;
-use axum::body::{Body, Bytes, to_bytes};
+use axum::body::{Bytes, to_bytes};
 use axum::extract::{FromRef, FromRequest, State};
 use axum::http::Request;
 use axum::response::IntoResponse;
@@ -15,12 +14,14 @@ use tower::ServiceBuilder;
 use tower_http::trace::TraceLayer;
 use tracing::{info, instrument};
 
+use tokio_util::sync::CancellationToken;
+
 use crate::consts::{GITEA_EVENT_TYPE_HEADER_NAME, GITEA_SIG_HEADER_NAME, MAX_WEBHOOK_BODY_SIZE};
 use crate::errors::AppError;
 use crate::gitea::WebhookType;
 use crate::state::AppState;
 
-pub async fn start(app_state: AppState) -> anyhow::Result<()> {
+pub async fn start(app_state: AppState, shutdown: CancellationToken) -> anyhow::Result<()> {
     let http_port = app_state.config.http_port;
 
     let app = Router::new()
@@ -38,8 +39,12 @@ pub async fn start(app_state: AppState) -> anyhow::Result<()> {
     info!("Listening API on port {}", http_port);
 
     axum::serve(listener, app)
+        .with_graceful_shutdown(async move { shutdown.cancelled().await })
         .await
         .map_err(anyhow::Error::from)
+        .inspect(|_| info!("API shutting down complete"))?;
+
+    Ok(())
 }
 
 async fn root() -> &'static str {

src/bot.rs

@@ -5,6 +5,7 @@ use crate::{
 };
 use serde::Deserialize;
 use std::{sync::Arc, time::Duration};
+use tokio_util::sync::CancellationToken;
 use tracing::{error, info, instrument};
 
 #[derive(Deserialize, Debug)]
@@ -54,13 +55,23 @@ impl Bot {
     pub async fn start(
         &self,
         mut rx: tokio::sync::mpsc::Receiver<WebhookType>,
+        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();
 
-        while let Some(wb) = rx.recv().await {
+        loop {
+            let wb = tokio::select! {
+                biased;
+                _ = shutdown.cancelled() => break,
+                msg = rx.recv() => match msg {
+                    Some(wb) => wb,
+                    None => break,
+                },
+            };
+
             // Drain completed tasks to avoid the JoinSet growing unbounded
             while let Some(res) = tasks.try_join_next() {
                 if let Err(e) = res {
@@ -82,7 +93,7 @@ impl Bot {
         // properly before returning
         tasks.join_all().await;
 
-        info!("Bot shutting down, channel closed");
+        info!("Bot shutting down complete");
         Ok(())
     }
 

src/main.rs

@@ -1,5 +1,8 @@
 use crate::{bot::Bot, gitea::WebhookType, state::AppState};
+
 use dotenvy::dotenv;
+use tokio::signal::unix::{SignalKind, signal};
+use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -54,11 +57,32 @@ async fn run() -> anyhow::Result<()> {
         "Starting Herald"
     );
 
+    let shutdown = CancellationToken::new();
+
     let bot = Bot::new(config.clone())?;
     let (tx, rx) = tokio::sync::mpsc::channel::<WebhookType>(config.bot_max_concurrent * 2);
     let app_state = AppState { bot_tx: tx, config };
 
-    tokio::try_join!(bot.start(rx), api::start(app_state))?;
+    let signal = async {
+        let mut sigterm = signal(SignalKind::terminate())?;
+        let mut sigint = signal(SignalKind::interrupt())?;
+        tokio::select! {
+            _ = sigterm.recv() => info!("Received SIGTERM"),
+            _ = sigint.recv() => info!("Received SIGINT"),
+        }
+
+        info!("Shutting down...");
+        shutdown.cancel();
+        anyhow::Ok(())
+    };
+
+    tokio::try_join!(
+        bot.start(rx, shutdown.clone()),
+        api::start(app_state, shutdown.clone()),
+        signal
+    )?;
+
+    info!("Shutdown complete");
 
     Ok(())
 }