refactor(webhook): refresh dispatcher to integrate with new obs and config
zhenyi committed
Modified lib/webhook/Cargo.toml lib/webhook/Cargo.toml +4 −0 View lib/webhook/Cargo.toml
@@ -9,15 +9,19 @@
99
1010[dependencies]
1111async-trait.workspace = true
12config.workspace = true
1213db.workspace = true
1314gitsync.workspace = true
1415hex.workspace = true
1516hmac.workspace = true
17metrics.workspace = true
18rand.workspace = true
1619reqwest.workspace = true
1720sea-orm.workspace = true
1821serde.workspace = true
1922serde_json.workspace = true
2023sha2.workspace = true
2124time.workspace = true
25tokio.workspace = true
2226tracing.workspace = true
2327uuid.workspace = true
No newline at end of file
Modified lib/webhook/dispatcher.rs lib/webhook/dispatcher.rs +328 −58 View lib/webhook/dispatcher.rs
@@ -1,39 +1,147 @@
11use crate::payload::build_push_payload;
22use crate::sign::sign_hmac_sha256;
33use async_trait::async_trait;
4use config::webhook::WebhookConfig;
45use db::conn::Db;
56use db::repo::repos;
67use db::webhook::repo_webhooks;
78use db::webhook::webhook_deliveries;
89use gitsync::{PostReceiveTask, WebhookTx};
10use metrics::counter;
911use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
1012use std::error::Error;
13use std::sync::Arc;
14use std::sync::atomic::{AtomicBool, Ordering};
1115use std::time::Duration;
1216use time::OffsetDateTime;
17use tokio::sync::{Mutex, Semaphore};
18use tokio::task::JoinSet;
19use tokio::time::sleep;
1320use tracing::{debug, warn};
1421use uuid::Uuid;
1522
16pub const DEFAULT_WEBHOOK_TIMEOUT: Duration = Duration::from_secs(10);
23/// Terminal outcome of a delivery attempt, distinguishing a successful HTTP
24/// response from a failure with a response (4xx/5xx) or a transport error.
25#[derive(Debug, Clone)]
26enum DeliveryOutcome {
27 Success { status: i32, body: Option<String> },
28 Failed { status: Option<i32>, body: Option<String> },
29}
1730
18type DeliveryResult = Result<(Option<i32>, Option<String>), Box<dyn Error + Send + Sync>>;
31/// Retry parameters derived from [`WebhookConfig`]; passable in isolation for tests.
32#[derive(Debug, Clone)]
33pub struct RetryPolicy {
34 pub max_attempts: u32,
35 pub base_delay: Duration,
36 pub max_delay: Duration,
37}
1938
39impl RetryPolicy {
40 fn from_config(cfg: &WebhookConfig) -> Self {
41 Self {
42 max_attempts: cfg.max_attempts,
43 base_delay: cfg.base_delay,
44 max_delay: cfg.max_delay,
45 }
46 }
47}
48
49/// Async, retry-capable webhook dispatcher.
50///
51/// Implements [`gitsync::WebhookTx`] so it can replace [`gitsync::NoopWebhookTx`].
52/// Each (webhook, ref_update) is dispatched on a spawned task; the trait method
53/// returns `Ok(())` immediately so the gitsync consumer is never blocked.
54/// Final outcomes (Success or Failed) are recorded exactly once per pair.
2055#[derive(Clone)]
2156pub struct WebhookDispatcher {
2257 db: Db,
2358 http: reqwest::Client,
24 timeout: Duration,
59 retry: RetryPolicy,
60 concurrency: Arc<Semaphore>,
61 in_flight: Arc<Mutex<JoinSet<()>>>,
62 draining: Arc<AtomicBool>,
2563}
2664
2765impl WebhookDispatcher {
66 /// Low-level constructor. Useful for unit tests that need to inject a custom
67 /// `reqwest::Client` or `RetryPolicy`.
2868 #[must_use]
29 pub fn new(db: Db, http: reqwest::Client, timeout: Duration) -> Self {
30 Self { db, http, timeout }
69 pub fn new(db: Db, http: reqwest::Client, retry: RetryPolicy) -> Self {
70 Self::with_concurrency(db, http, retry, 32)
71 }
72
73 /// Constructor with an explicit concurrency ceiling.
74 #[must_use]
75 pub fn with_concurrency(
76 db: Db,
77 http: reqwest::Client,
78 retry: RetryPolicy,
79 max_concurrent: usize,
80 ) -> Self {
81 Self {
82 db,
83 http,
84 retry,
85 concurrency: Arc::new(Semaphore::new(max_concurrent.max(1))),
86 in_flight: Arc::new(Mutex::new(JoinSet::new())),
87 draining: Arc::new(AtomicBool::new(false)),
88 }
89 }
90
91 /// Build an HTTP client from [`WebhookConfig`] and return a ready dispatcher.
92 /// Returns an error only if `reqwest::Client::builder().build()` fails, which
93 /// is unlikely for valid configs but reported honestly regardless.
94 pub fn from_config(cfg: &WebhookConfig, db: Db) -> Result<Self, reqwest::Error> {
95 Self::from_config_shared(cfg, db)
3196 }
3297
98 /// Convenience for tests that want a dispatcher without configuring an HTTP
99 /// client: builds a fresh `reqwest::Client` with the workspace defaults.
33100 #[must_use]
34101 pub fn with_defaults(db: Db) -> Self {
35 Self::new(db, reqwest::Client::new(), DEFAULT_WEBHOOK_TIMEOUT)
102 let cfg = WebhookConfig::default();
103 Self::from_config_shared(&cfg, db).expect("default webhook config always builds")
36104 }
105
106 fn from_config_shared(cfg: &WebhookConfig, db: Db) -> Result<Self, reqwest::Error> {
107 let mut builder = reqwest::Client::builder()
108 .timeout(cfg.request_timeout)
109 .pool_max_idle_per_host(cfg.pool_max_idle_per_host);
110 if let Some(t) = cfg.connect_timeout {
111 builder = builder.connect_timeout(t);
112 }
113 if let Some(t) = cfg.pool_idle_timeout {
114 builder = builder.pool_idle_timeout(t);
115 }
116 let redirect = if cfg.follow_redirects {
117 reqwest::redirect::Policy::limited(5)
118 } else {
119 reqwest::redirect::Policy::none()
120 };
121 let http = builder.redirect(redirect).build()?;
122 Ok(Self::with_concurrency(
123 db,
124 http,
125 RetryPolicy::from_config(cfg),
126 cfg.max_concurrent_dispatches,
127 ))
128 }
129
130 /// Wait for all currently-spawned dispatch tasks to finish. New calls to
131 /// [`Self::dispatch`] after this are recorded inline as `Failed` (without
132 /// spawning) to preserve the single-row invariant. Idempotent.
133 pub async fn drain(&self) {
134 self.draining.store(true, Ordering::Release);
135 let mut set = {
136 let mut guard = self.in_flight.lock().await;
137 std::mem::take(&mut *guard)
138 };
139 while let Some(res) = set.join_next().await {
140 if let Err(e) = res {
141 warn!(error = %e, "webhook dispatch task join error during drain");
142 }
143 }
144 }
37145}
38146
39147#[async_trait]
@@ -68,6 +176,7 @@
68176 .await
69177 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("load webhooks: {e}")))?;
70178
179 let draining = self.draining.load(Ordering::Acquire);
71180 let mut dispatched = 0usize;
72181 for webhook in &webhooks {
73182 if !event_enabled(&webhook.events) {
@@ -76,70 +185,231 @@
76185 for update in &task.refs {
77186 let payload = serde_json::to_vec(&build_push_payload(task, &repo, update))?;
78187 let signature = sign_hmac_sha256(&payload, &webhook.secret);
79 let result = self.post(&webhook.url, &signature, &payload).await;
80 self.record_delivery(webhook, &payload, &result).await;
188 if draining {
189 // Drain started: do not spawn. Record the outcome inline so
190 // operators still see a row, marked as failed with the
191 // drain reason.
192 let outcome = DeliveryOutcome::Failed {
193 status: None,
194 body: Some("dispatcher draining; new deliveries rejected".into()),
195 };
196 persist_delivery(&self.db, webhook.id, &payload, &outcome).await;
197 } else {
198 self.spawn_delivery(webhook.id, webhook.url.clone(), signature, payload);
199 }
81200 dispatched += 1;
82201 }
83202 }
84203 Ok(dispatched)
85204 }
86205
87 async fn post(&self, url: &str, signature: &str, payload: &[u8]) -> DeliveryResult {
88 let response = self
89 .http
90 .post(url)
91 .timeout(self.timeout)
92 .header(reqwest::header::CONTENT_TYPE, "application/json")
93 .header("X-Hub-Signature-256", format!("sha256={signature}"))
94 .header("X-Gitea-Signature", format!("sha256={signature}"))
95 .body(payload.to_vec())
96 .send()
97 .await
98 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("http post {url}: {e}")))?;
206 fn spawn_delivery(
207 &self,
208 webhook_id: Uuid,
209 url: String,
210 signature: String,
211 payload: Vec<u8>,
212 ) {
213 let db = self.db.clone();
214 let http = self.http.clone();
215 let retry = self.retry.clone();
216 let concurrency = self.concurrency.clone();
217 let in_flight = self.in_flight.clone();
218 // Try to register with the JoinSet without awaiting on the mutex; if
219 // the dispatcher is currently being drained (lock contended by drain)
220 // fall back to a detached task so we never block the consumer.
221 let mut guard = match in_flight.try_lock() {
222 Ok(g) => g,
223 Err(_) => {
224 // Drain holds the lock; register nothing, run detached so the
225 // consumer is never blocked. The associated row is still
226 // written by the task itself.
227 tokio::spawn(async move {
228 let Ok(_permit) = concurrency.acquire_owned().await else {
229 return;
230 };
231 let outcome = run_retry(&http, &url, &signature, &payload, &retry).await;
232 persist_delivery(&db, webhook_id, &payload, &outcome).await;
233 });
234 return;
235 }
236 };
237 guard.spawn(async move {
238 let permit = match concurrency.acquire_owned().await {
239 Ok(p) => p,
240 Err(_) => return,
241 };
242 let outcome = run_retry(&http, &url, &signature, &payload, &retry).await;
243 persist_delivery(&db, webhook_id, &payload, &outcome).await;
244 drop(permit);
245 });
246 }
247}
248
249/// Run the retry loop. Returns the terminal outcome to be recorded.
250async fn run_retry(
251 http: &reqwest::Client,
252 url: &str,
253 signature: &str,
254 payload: &[u8],
255 retry: &RetryPolicy,
256) -> DeliveryOutcome {
257 let mut last_failed: Option<DeliveryOutcome> = None;
258
259 for attempt in 1..=retry.max_attempts {
260 match post_once(http, url, signature, payload).await {
261 Ok((status, body, retry_after)) => {
262 let code = status.unwrap_or(0);
263 if (200..400).contains(&code) {
264 counter!("webhook.deliveries", "result" => "success").increment(1);
265 return DeliveryOutcome::Success { status: code, body };
266 }
267 last_failed = Some(DeliveryOutcome::Failed {
268 status,
269 body: body.clone(),
270 });
271 if is_retryable_status(code) {
272 debug!(
273 attempt,
274 status = code,
275 url = %url,
276 "webhook returned retryable status"
277 );
278 if attempt < retry.max_attempts {
279 sleep(backoff_delay(retry, attempt, retry_after)).await;
280 }
281 continue;
282 }
283 // Terminal 4xx. Record and stop.
284 counter!("webhook.deliveries", "result" => "failed").increment(1);
285 return DeliveryOutcome::Failed { status, body };
286 }
287 Err(e) => {
288 debug!(attempt, url = %url, error = %e, "webhook transport error");
289 last_failed = Some(DeliveryOutcome::Failed {
290 status: None,
291 body: Some(e.to_string()),
292 });
293 }
294 }
99295
100 let status = response.status().as_u16() as i32;
101 let body = response
102 .text()
103 .await
104 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("read response: {e}")))?;
105 Ok((Some(status), Some(body)))
296 if attempt < retry.max_attempts {
297 sleep(backoff_delay(retry, attempt, None)).await;
298 }
106299 }
107300
108 #[allow(clippy::manual_async_fn)]
109 async fn record_delivery(
110 &self,
111 webhook: &repo_webhooks::Model,
112 payload: &[u8],
113 result: &DeliveryResult,
114 ) {
115 let (status, response_status, response_body) = match result {
116 Ok((code, body)) => (
117 webhook_deliveries::WebhookDeliveryStatus::Success,
118 *code,
119 body.clone(),
120 ),
121 Err(e) => (
122 webhook_deliveries::WebhookDeliveryStatus::Failed,
123 None,
124 Some(e.to_string()),
125 ),
126 };
301 counter!("webhook.deliveries", "result" => "failed").increment(1);
302 last_failed.unwrap_or(DeliveryOutcome::Failed {
303 status: None,
304 body: Some("max attempts exhausted".into()),
305 })
306}
127307
128 let delivery = webhook_deliveries::ActiveModel {
129 id: Set(Uuid::now_v7()),
130 webhook_id: Set(webhook.id),
131 event_type: Set("push".to_string()),
132 payload: Set(String::from_utf8_lossy(payload).into_owned()),
133 status: Set(status),
134 response_status: Set(response_status),
135 response_body: Set(response_body),
136 created_at: Set(OffsetDateTime::now_utc()),
137 };
308/// One HTTP attempt. Returns (status_code, response_body, parsed_retry_after).
309async fn post_once(
310 http: &reqwest::Client,
311 url: &str,
312 signature: &str,
313 payload: &[u8],
314) -> Result<(Option<i32>, Option<String>, Option<Duration>), Box<dyn Error + Send + Sync>> {
315 let response = http
316 .post(url)
317 .header(reqwest::header::CONTENT_TYPE, "application/json")
318 .header("X-Hub-Signature-256", format!("sha256={signature}"))
319 .header("X-Gitea-Signature", format!("sha256={signature}"))
320 .body(payload.to_vec())
321 .send()
322 .await
323 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("http post {url}: {e}")))?;
324
325 let retry_after = response
326 .headers()
327 .get(reqwest::header::RETRY_AFTER)
328 .and_then(|v| v.to_str().ok())
329 .and_then(parse_retry_after_seconds);
138330
139 if let Err(e) = delivery.insert(&self.db.writer()).await {
140 warn!(error = %e, webhook_id = %webhook.id, "failed to record webhook delivery");
331 let status = response.status().as_u16() as i32;
332 let body = response
333 .text()
334 .await
335 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("read response: {e}")))?;
336 Ok((Some(status), Some(body), retry_after))
337}
338
339/// Parse RFC 7231 Retry-After into a Duration (seconds). Both delta-seconds and
340/// HTTP-date forms are accepted. Returns None on any parse failure.
341fn parse_retry_after_seconds(value: &str) -> Option<Duration> {
342 if let Ok(secs) = value.trim().parse::<u64>() {
343 return Some(Duration::from_secs(secs));
344 }
345 // HTTP-date form: e.g. "Wed, 21 Oct 2015 07:28:00 GMT".
346 if let Ok(ts) = time::OffsetDateTime::parse(
347 value.trim(),
348 &time::format_description::well_known::Rfc2822,
349 ) {
350 let now = time::OffsetDateTime::now_utc();
351 if let Ok(diff) = (ts - now).try_into() {
352 return Some(diff);
141353 }
142354 }
355 None
356}
357
358fn is_retryable_status(code: i32) -> bool {
359 code == 408 || code == 429 || (500..600).contains(&code)
360}
361
362/// Full-jitter exponential backoff: cap = min(max_delay, base * 2^(attempt-1));
363/// sleep = rand(0..=cap).
364fn backoff_delay(retry: &RetryPolicy, attempt: u32, retry_after: Option<Duration>) -> Duration {
365 // Honor Retry-After when present, capped to max_delay to defend against
366 // misbehaving / hostile servers; fall back to full-jitter exponential.
367 if let Some(s) = retry_after {
368 return s.min(retry.max_delay).max(retry.base_delay);
369 }
370 let exp = retry.base_delay.saturating_mul(1u32 << (attempt - 1).min(20));
371 let cap = exp.min(retry.max_delay);
372 let millis = cap.as_millis().min(u128::from(u64::MAX)) as u64;
373 if millis == 0 {
374 return retry.base_delay;
375 }
376 let jittered = rand::random::<u64>() % (millis + 1);
377 Duration::from_millis(jittered)
378}
379
380async fn persist_delivery(
381 db: &Db,
382 webhook_id: Uuid,
383 payload: &[u8],
384 outcome: &DeliveryOutcome,
385) {
386 let (status, response_status, response_body) = match outcome {
387 DeliveryOutcome::Success { status, body } => (
388 webhook_deliveries::WebhookDeliveryStatus::Success,
389 Some(*status),
390 body.clone(),
391 ),
392 DeliveryOutcome::Failed { status, body } => (
393 webhook_deliveries::WebhookDeliveryStatus::Failed,
394 *status,
395 body.clone(),
396 ),
397 };
398
399 let delivery = webhook_deliveries::ActiveModel {
400 id: Set(Uuid::now_v7()),
401 webhook_id: Set(webhook_id),
402 event_type: Set("push".to_string()),
403 payload: Set(String::from_utf8_lossy(payload).into_owned()),
404 status: Set(status),
405 response_status: Set(response_status),
406 response_body: Set(response_body),
407 created_at: Set(OffsetDateTime::now_utc()),
408 };
409
410 if let Err(e) = delivery.insert(&db.writer()).await {
411 warn!(error = %e, webhook_id = %webhook_id, "failed to record webhook delivery");
412 }
143413}
144414
145415fn event_enabled(events: &sea_orm::JsonValue) -> bool {
@@ -151,4 +421,4 @@
151421 }
152422 arr.iter()
153423 .any(|e| e.as_str() == Some("push") || e.as_str() == Some("*"))
154}
424}
No newline at end of file