b90f48e
use std::net::SocketAddr;
use std::time::Duration;
use anyhow::Context;
use axum::Router;
use ci::CiService;
use config::scoped::{ScopedConfig, WebhookServiceConfig};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tracing::Instrument as _;
use tracing::{info, warn};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
export::telemetry::init_tracing();
let cfg = WebhookServiceConfig::load().context("loading configuration")?;
if !cfg.webhook.enabled() {
info!("webhook is disabled by config; exiting");
return Ok(());
}
info!("connecting to database / redis / storage");
let state = export::AppState::with_config(cfg.clone().into_app_config())
.await
.context("building AppState")?;
let service = CiService::new(state);
info!("dependencies ready");
if let Err(err) = export::telemetry::init_meter_provider() {
warn!(error = %err, "failed to init meter provider; metrics disabled");
}
export::metrics::spawn_queue_depth_reporter(
service.state.redis.clone(),
&[
("ci.events", cfg.ci.event_queue_topic.clone()),
("ci.events.dlq", cfg.ci.event_dlq_topic.clone()),
],
);
let cancel = CancellationToken::new();
let consumer = if cfg.ci.enabled {
Some(tokio::spawn(consume_events(
service.clone(),
cancel.clone(),
)))
} else {
info!("ci is disabled by config; scheduler not started");
None
};
let scheduler = if cfg.ci.enabled {
Some(tokio::spawn(run_scheduler(
service.clone(),
cfg.ci.scheduler_interval_secs,
cancel.clone(),
)))
} else {
None
};
let outbound = tokio::spawn(run_outbound(
service.clone(),
cfg.ci.scheduler_interval_secs,
cancel.clone(),
));
let addr = SocketAddr::new(
cfg.webhook
.bind_addr()
.parse()
.context("webhook bind addr")?,
cfg.webhook.port(),
);
let listener = TcpListener::bind(addr)
.await
.with_context(|| format!("binding webhook {addr}"))?;
let runner_routes = ci::routes::runner_router().with_state(service.clone());
let router: Router = Router::new()
.nest("/api/v1", runner_routes)
.merge(ci::actions_runtime::router().with_state(service.clone()))
.merge(export::health::router_with(
Some(service.state.database.clone()),
service.state.redis.clone(),
))
.layer(export::telemetry::http_trace_layer())
.layer(export::telemetry::metrics_layer())
.layer(export::middleware::AppTraceLayer::new());
info!(%addr, "webhook scheduler listening");
let server_cancel = cancel.clone();
axum::serve(listener, router.into_make_service())
.with_graceful_shutdown(async move { server_cancel.cancelled().await })
.await
.context("webhook server error")?;
cancel.cancel();
if let Some(consumer) = consumer {
let _ = consumer.await;
}
if let Some(scheduler) = scheduler {
let _ = scheduler.await;
}
let _ = outbound.await;
info!("webhook shut down");
Ok(())
}
async fn consume_events(service: CiService, cancel: CancellationToken) {
use ci::events::{decode_event, publish_event_to_dlq};
use git::sync::queue::SyncQueue;
let topic = service.config().ci.event_queue_topic.clone();
let dlq = service.config().ci.event_dlq_topic.clone();
// BRPOP blocks longer than the Redis request timeout, so the consumer
// needs its own dedicated connection (same as the sync worker).
let queue = match SyncQueue::with_consumer_conns(service.state.redis.clone(), 1).await {
Ok(queue) => queue,
Err(error) => {
warn!(%error, "failed to open ci event consumer connection");
return;
}
};
let mut attempts: std::collections::HashMap<uuid::Uuid, u32> = std::collections::HashMap::new();
loop {
let payload = tokio::select! {
_ = cancel.cancelled() => break,
res = queue.consume(&topic, 5, 0) => match res {
Ok(Some(payload)) => payload,
Ok(None) => continue,
Err(error) => {
warn!(%error, "event consume failed");
tokio::time::sleep(Duration::from_secs(2)).await;
continue;
}
},
};
let event = match decode_event(&payload) {
Ok(event) => event,
Err(error) => {
warn!(%error, "dropping malformed ci event");
let _ = publish_event_to_dlq(&service.state.redis, &dlq, &payload).await;
continue;
}
};
let event_kind = match &event {
ci_protocol::events::CiEvent::Push(_) => "push",
ci_protocol::events::CiEvent::PullRequest(_) => "pull_request",
ci_protocol::events::CiEvent::Trigger(_) => "trigger",
};
export::metrics::ci_event(event_kind);
let event_id = match &event {
ci_protocol::events::CiEvent::Push(push) => push.id,
ci_protocol::events::CiEvent::PullRequest(pr) => pr.id,
ci_protocol::events::CiEvent::Trigger(event) => event.id,
};
let span = tracing::info_span!(
"ci.event",
event = %event_id,
request_id = tracing::field::Empty
);
if let Some(request_id) = event.request_id() {
span.record("request_id", request_id);
}
export::telemetry::set_parent_from_traceparent(&span, event.trace());
match export::middleware::with_request_id(
event.request_id(),
service.handle_event(&event).instrument(span),
)
.await
{
Ok(created) => {
attempts.remove(&event_id);
export::metrics::ci_runs_created(event_kind, created.len());
if !created.is_empty() {
info!(event = %event_id, runs = created.len(), "ci event produced runs");
}
}
Err(error) => {
let count = attempts.entry(event_id).or_insert(0);
*count += 1;
if *count > 5 {
attempts.remove(&event_id);
warn!(%error, event = %event_id, "ci event failed permanently; sending to DLQ");
let _ = publish_event_to_dlq(&service.state.redis, &dlq, &payload).await;
} else {
warn!(%error, event = %event_id, attempt = *count, "ci event failed; retrying");
tokio::time::sleep(Duration::from_secs(2 * u64::from(*count))).await;
let _ = queue.publish_raw(&topic, None, &payload).await;
}
}
}
}
info!("event consumer stopped");
}
async fn run_outbound(service: CiService, interval_secs: u64, cancel: CancellationToken) {
let interval = Duration::from_secs(interval_secs.clamp(1, 30));
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tokio::select! {
_ = cancel.cancelled() => break,
_ = ticker.tick() => {}
}
if let Err(error) = service.deliver_pending_repo_webhooks().await {
warn!(%error, "outbound webhook delivery failed");
}
}
info!("outbound webhook loop stopped");
}
async fn run_scheduler(service: CiService, interval_secs: u64, cancel: CancellationToken) {
if let Err(error) = service.discover_actions_schedules().await {
warn!(%error,"initial schedule discovery failed");
}
let interval = Duration::from_secs(interval_secs.max(1));
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut ticks: u64 = 0;
loop {
tokio::select! {
_ = cancel.cancelled() => break,
_ = ticker.tick() => {}
}
if let Err(error) = service.advance_jobs().await {
warn!(%error, "advance jobs failed");
}
if let Err(error) = service.enforce_concurrency().await {
warn!(%error, "concurrency enforcement failed");
}
if let Err(error) = service.expire_leases().await {
warn!(%error, "lease expiry failed");
}
if let Err(error) = service.apply_cancellations().await {
warn!(%error, "cancellation sweep failed");
}
if let Err(error) = service.finalize_runs().await {
warn!(%error, "run finalization failed");
}
ticks = ticks.wrapping_add(1);
if let Err(error) = service.run_due_actions_schedules().await {
warn!(%error,"scheduled workflow execution failed");
}
if ticks.is_multiple_of(3600)
&& let Err(error) = service.prune_expired().await
{
warn!(%error, "retention prune failed");
}
if ticks.is_multiple_of(30)
&& let Err(error) = service.reap_stale_runners().await
{
warn!(%error, "stale runner sweep failed");
}
}
info!("scheduler stopped");
}