forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
22.0 KiB
1//! The webhook delivery worker: claims due deliveries, sends them with bounded concurrency, and
2//! records each result. A failed attempt is retried on the backoff schedule until the sixth
3//! attempt, after which the delivery is `gave_up`.
4//!
5//! The signature is computed at send time from the decrypted secret, so a rotated secret applies
6//! to retries too. The headers queued at enqueue time never carry a signature or a delivery GUID;
7//! both are added here. After each send, the headers actually sent replace the stored ones, so the
8//! delivery log shows what the receiver got.
9
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use forge_core::Error;
14use forge_core::time::{now_stamp, stamp_after, stamp_ago};
15use forge_core::webhooks::{AttemptOutcome, Delivery};
16use futures_util::StreamExt;
17use serde_json::{Map, Value};
18use time::OffsetDateTime;
19use time::format_description::well_known::Rfc3339;
20use tokio::sync::Notify;
21
22use super::client::{HookClient, SendOutcome, WebhookPolicy, sign, validate_url};
23use crate::state::AppState;
24
25/// Handle that wakes the worker. `emit` calls `notify_one` after queueing deliveries.
26pub type WebhookWaker = Arc<Notify>;
27
28/// Deliveries claimed per pass.
29const BATCH: u32 = 16;
30/// Deliveries sent at once.
31const CONCURRENCY: usize = 4;
32/// The shortest sleep, so a due time that has already passed does not spin the loop.
33const MIN_SLEEP: Duration = Duration::from_secs(1);
34/// The longest sleep. It only guards against a missed wake, it is not polling.
35const BACKSTOP: Duration = Duration::from_secs(15 * 60);
36/// Deliveries older than this are purged.
37const RETENTION: Duration = Duration::from_secs(14 * 24 * 60 * 60);
38/// How often the purge runs.
39const PURGE_EVERY: Duration = Duration::from_secs(24 * 60 * 60);
40/// Delay before the next attempt, indexed by the attempt that just failed (attempt 1 waits 1m).
41/// After the sixth attempt there is no retry.
42const BACKOFF: [Duration; 5] = [
43 Duration::from_secs(60),
44 Duration::from_secs(5 * 60),
45 Duration::from_secs(30 * 60),
46 Duration::from_secs(2 * 60 * 60),
47 Duration::from_secs(6 * 60 * 60),
48];
49/// Headers the worker owns. Anything stored under these names is replaced at send time.
50const WORKER_HEADERS: [&str; 3] = ["Content-Type", "X-Forge-Delivery", "X-Forge-Signature-256"];
51
52/// Starts the worker task and returns the handle that wakes it. The task runs until the process
53/// exits; it never panics on a bad delivery.
54pub fn spawn(state: AppState) -> WebhookWaker {
55 let waker = Arc::clone(&state.webhook_waker);
56 let task_waker = Arc::clone(&waker);
57 tokio::spawn(run(state, task_waker));
58 waker
59}
60
61async fn run(state: AppState, waker: WebhookWaker) {
62 let policy = match WebhookPolicy::from_config(&state.config.webhooks) {
63 Ok(policy) => policy,
64 Err(e) => {
65 // Fall back to the strictest policy: no internal hosts and no plain HTTP.
66 tracing::error!(error = %e, "webhooks: invalid [webhooks] config, using the default policy");
67 WebhookPolicy {
68 allow_http: false,
69 allowed_hosts: Vec::new(),
70 }
71 }
72 };
73 let client = match HookClient::new(policy.clone()) {
74 Ok(client) => client,
75 Err(e) => {
76 tracing::error!(error = %e, "webhooks: cannot build the HTTP client, deliveries stop");
77 return;
78 }
79 };
80
81 let mut last_purge: Option<Instant> = None;
82 loop {
83 purge_if_due(&state, &mut last_purge).await;
84 let claimed = match process_due(&state, &client, &policy, &now_stamp()).await {
85 Ok(claimed) => claimed,
86 Err(e) => {
87 tracing::error!(error = %e, "webhooks: claiming deliveries failed");
88 0
89 }
90 };
91 // A full batch may mean more is due, so go straight back for it.
92 if claimed < usize::try_from(BATCH).unwrap_or(usize::MAX) {
93 // Wake when the earliest pending delivery is due (a claimed one wakes at its lease
94 // expiry), or at the next purge. With nothing pending, sleep up to `BACKSTOP`.
95 let next = match state.store.next_delivery_due().await {
96 Ok(next) => next,
97 Err(e) => {
98 tracing::error!(error = %e, "webhooks: could not read the next due time");
99 None
100 }
101 };
102 let nap = next
103 .as_deref()
104 .map_or(BACKSTOP, |due| sleep_for(due, OffsetDateTime::now_utc()));
105 let purge_nap = last_purge.map_or(Duration::ZERO, |at| {
106 PURGE_EVERY.saturating_sub(at.elapsed())
107 });
108 tokio::select! {
109 () = waker.notified() => {}
110 () = tokio::time::sleep(nap) => {}
111 () = tokio::time::sleep(purge_nap) => {}
112 }
113 }
114 }
115}
116
117/// How long the worker sleeps until `next_due` (an RFC 3339 stamp, the earliest pending
118/// delivery), clamped to `MIN_SLEEP..=BACKSTOP`. An unparsable stamp gives `MIN_SLEEP`, so it is
119/// retried promptly rather than stalling.
120fn sleep_for(next_due: &str, now: OffsetDateTime) -> Duration {
121 let Ok(due) = OffsetDateTime::parse(next_due, &Rfc3339) else {
122 return MIN_SLEEP;
123 };
124 let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
125 until.clamp(MIN_SLEEP, BACKSTOP)
126}
127
128/// Claims up to [`BATCH`] due deliveries at `now`, sends them, and records each result. Returns
129/// how many were claimed.
130pub(crate) async fn process_due(
131 state: &AppState,
132 client: &HookClient,
133 policy: &WebhookPolicy,
134 now: &str,
135) -> forge_core::Result<usize> {
136 let claimed = state.store.claim_due_deliveries(now, BATCH).await?;
137 let count = claimed.len();
138 futures_util::stream::iter(claimed)
139 .for_each_concurrent(CONCURRENCY, |delivery| async move {
140 let outcome = attempt(state, client, policy, &delivery).await;
141 if let Err(e) = state
142 .store
143 .record_delivery_result(&delivery.id, &outcome)
144 .await
145 {
146 tracing::error!(error = %e, delivery_id = %delivery.id, "webhooks: recording the result failed");
147 }
148 })
149 .await;
150 Ok(count)
151}
152
153/// One send attempt for a claimed delivery. Never fails: every problem becomes an outcome.
154async fn attempt(
155 state: &AppState,
156 client: &HookClient,
157 policy: &WebhookPolicy,
158 delivery: &Delivery,
159) -> AttemptOutcome {
160 let hook = match state.store.webhook(&delivery.hook_id).await {
161 Ok(Some(hook)) => hook,
162 Ok(None) => return give_up("webhook was deleted"),
163 Err(e) => return retry(delivery, &format!("loading the webhook failed: {e}")),
164 };
165 let secret = match state.store.webhook_secret(&hook.id, &state.key).await {
166 Ok(secret) => secret,
167 Err(Error::NotFound) => return give_up("webhook was deleted"),
168 Err(Error::Crypto(e)) => return give_up(&format!("cannot open the webhook secret: {e}")),
169 Err(e) => return retry(delivery, &format!("loading the webhook secret failed: {e}")),
170 };
171 // Policy can change between save and send, so the URL is checked again here.
172 if let Err(e) = validate_url(&hook.url, policy) {
173 return retry(delivery, &e.to_string());
174 }
175 let body = delivery.request_body.clone().into_bytes();
176 let headers = request_headers(delivery, &secret, &body);
177 let sent = client.send(&hook.url, &headers, body).await;
178 let mut result = outcome(delivery, sent);
179 result.request_headers = Some(headers_json(&headers));
180 result
181}
182
183/// The headers as a JSON object, the form the store keeps. Stored after each send so the delivery
184/// log shows the GUID and signature the receiver got.
185fn headers_json(headers: &[(String, String)]) -> String {
186 let map: Map<String, Value> = headers
187 .iter()
188 .map(|(name, value)| (name.clone(), Value::String(value.clone())))
189 .collect();
190 Value::Object(map).to_string()
191}
192
193/// The stored headers, minus the ones the worker owns, plus the fresh GUID, content type and
194/// signature.
195fn request_headers(delivery: &Delivery, secret: &str, body: &[u8]) -> Vec<(String, String)> {
196 let mut headers: Vec<(String, String)> = match serde_json::from_str::<Map<String, Value>>(
197 &delivery.request_headers,
198 ) {
199 Ok(map) => map
200 .into_iter()
201 .filter_map(|(name, value)| match value {
202 Value::String(value) => Some((name, value)),
203 _ => None,
204 })
205 .collect(),
206 Err(e) => {
207 tracing::error!(error = %e, delivery_id = %delivery.id, "webhooks: stored headers are not a JSON object");
208 Vec::new()
209 }
210 };
211 headers.retain(|(name, _)| {
212 !WORKER_HEADERS
213 .iter()
214 .any(|owned| owned.eq_ignore_ascii_case(name))
215 });
216 headers.push(("Content-Type".to_owned(), "application/json".to_owned()));
217 headers.push(("X-Forge-Delivery".to_owned(), delivery.guid.clone()));
218 headers.push((
219 "X-Forge-Signature-256".to_owned(),
220 sign(secret.as_bytes(), body),
221 ));
222 headers
223}
224
225/// Turns the send result into the outcome to record. Only a 2xx status counts as success.
226fn outcome(delivery: &Delivery, sent: SendOutcome) -> AttemptOutcome {
227 let ok = matches!(sent.status, Some(200..=299));
228 let mut response_map = Map::new();
229 for (name, value) in sent.headers {
230 response_map.insert(name, Value::String(value));
231 }
232 AttemptOutcome {
233 ok,
234 status: sent.status,
235 response_headers: Some(Value::Object(response_map).to_string()),
236 response_body: Some(sent.body),
237 error: sent.error,
238 duration_ms: Some(sent.duration_ms),
239 next_attempt_at: if ok {
240 None
241 } else {
242 next_attempt_at(delivery.attempt)
243 },
244 request_headers: None,
245 }
246}
247
248/// A failed attempt that will be retried, when the schedule has a slot for it.
249fn retry(delivery: &Delivery, error: &str) -> AttemptOutcome {
250 AttemptOutcome {
251 error: Some(error.to_owned()),
252 next_attempt_at: next_attempt_at(delivery.attempt),
253 ..AttemptOutcome::default()
254 }
255}
256
257/// A failure that will not be retried.
258fn give_up(error: &str) -> AttemptOutcome {
259 AttemptOutcome {
260 error: Some(error.to_owned()),
261 ..AttemptOutcome::default()
262 }
263}
264
265/// When the attempt after `attempt` is due, or `None` once the schedule is used up.
266fn next_attempt_at(attempt: u32) -> Option<String> {
267 let index = usize::try_from(attempt.checked_sub(1)?).ok()?;
268 let delay = BACKOFF.get(index)?;
269 Some(stamp_after(*delay))
270}
271
272/// Deletes deliveries older than [`RETENTION`], at most once per [`PURGE_EVERY`].
273async fn purge_if_due(state: &AppState, last: &mut Option<Instant>) {
274 if last.is_some_and(|at| at.elapsed() < PURGE_EVERY) {
275 return;
276 }
277 *last = Some(Instant::now());
278 let cutoff = stamp_ago(RETENTION);
279 match state.store.purge_deliveries_before(&cutoff).await {
280 Ok(0) => {}
281 Ok(removed) => tracing::info!(removed, "webhooks: purged old deliveries"),
282 Err(e) => tracing::error!(error = %e, "webhooks: purging old deliveries failed"),
283 }
284}
285
286#[cfg(test)]
287mod tests {
288 use forge_core::Visibility;
289 use forge_core::webhooks::{Delivery, DeliveryState, Webhook, WebhookEvent};
290 use tokio::io::{AsyncReadExt, AsyncWriteExt};
291 use tokio::net::TcpListener;
292 use tokio::sync::mpsc;
293
294 use super::*;
295 use crate::test_support::{KEY, test_state};
296
297 /// A request as the local test server saw it.
298 struct Seen {
299 headers: Vec<(String, String)>,
300 body: Vec<u8>,
301 }
302
303 /// Serves one request per connection, reporting each request on the channel, and answers
304 /// with `status`.
305 async fn server(status: u16) -> (String, mpsc::UnboundedReceiver<Seen>) {
306 let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
307 let addr = listener.local_addr().expect("addr");
308 let (tx, rx) = mpsc::unbounded_channel();
309 tokio::spawn(async move {
310 loop {
311 let Ok((mut socket, _)) = listener.accept().await else {
312 return;
313 };
314 let tx = tx.clone();
315 tokio::spawn(async move {
316 let mut buf = Vec::new();
317 let mut chunk = [0u8; 1024];
318 let head_end = loop {
319 if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
320 break pos + 4;
321 }
322 match socket.read(&mut chunk).await {
323 Ok(0) | Err(_) => return,
324 Ok(n) => buf.extend_from_slice(&chunk[..n]),
325 }
326 };
327 let head = String::from_utf8_lossy(&buf[..head_end]).into_owned();
328 let headers: Vec<(String, String)> = head
329 .lines()
330 .skip(1)
331 .filter_map(|line| {
332 line.split_once(':')
333 .map(|(k, v)| (k.trim().to_owned(), v.trim().to_owned()))
334 })
335 .collect();
336 let length: usize = headers
337 .iter()
338 .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
339 .and_then(|(_, v)| v.parse().ok())
340 .unwrap_or(0);
341 while buf.len() < head_end + length {
342 match socket.read(&mut chunk).await {
343 Ok(0) | Err(_) => break,
344 Ok(n) => buf.extend_from_slice(&chunk[..n]),
345 }
346 }
347 let body = buf[head_end..].to_vec();
348 let _ = tx.send(Seen { headers, body });
349 let reply = format!(
350 "HTTP/1.1 {status} X\r\nContent-Length: 2\r\nConnection: close\r\n\r\nno"
351 );
352 let _ = socket.write_all(reply.as_bytes()).await;
353 let _ = socket.shutdown().await;
354 });
355 }
356 });
357 (format!("http://127.0.0.1:{}/hook", addr.port()), rx)
358 }
359
360 fn local_policy() -> WebhookPolicy {
361 WebhookPolicy::from_config(&forge_core::config::WebhookConfig {
362 allow_http: true,
363 allowed_hosts: vec!["127.0.0.1/32".into()],
364 })
365 .expect("policy")
366 }
367
368 /// A repo with one hook pointing at `url`, subscribed to push, and one queued delivery.
369 async fn queued(state: &AppState, url: &str) -> (Webhook, Delivery, tempfile::TempDir) {
370 let tmp = tempfile::tempdir().expect("tempdir");
371 let owner = state
372 .store
373 .create_user("alice", None, "unused", false)
374 .await
375 .expect("user");
376 let repo = state
377 .store
378 .create_repo(&owner.id, "proj", "", Visibility::Public, "main")
379 .await
380 .expect("repo");
381 let sealed = forge_core::secret::seal(&KEY, b"s3cret").expect("seal");
382 let hook = state
383 .store
384 .create_webhook(
385 &repo.id,
386 &owner.id,
387 url,
388 sealed.as_bytes(),
389 &[WebhookEvent::Push],
390 true,
391 )
392 .await
393 .expect("hook");
394 let headers = r#"{"Content-Type":"application/json","X-Forge-Event":"push","X-Forge-Delivery":"stale","X-Forge-Signature-256":"sha256=stale"}"#;
395 let delivery = state
396 .store
397 .enqueue_delivery(
398 &hook.id,
399 WebhookEvent::Push,
400 "",
401 headers,
402 r#"{"ref":"refs/heads/main"}"#,
403 )
404 .await
405 .expect("enqueue");
406 (hook, delivery, tmp)
407 }
408
409 #[tokio::test]
410 async fn delivers_with_fresh_signature_and_headers() {
411 let tmp = tempfile::tempdir().expect("tempdir");
412 let state = test_state(tmp.path()).await;
413 let (url, mut seen) = server(204).await;
414 let (hook, delivery, _dir) = queued(&state, &url).await;
415 let client = HookClient::new(local_policy()).expect("client");
416
417 let claimed = process_due(&state, &client, &local_policy(), &now_stamp())
418 .await
419 .expect("process");
420 assert_eq!(claimed, 1);
421
422 let request = seen.recv().await.expect("request arrives");
423 let header = |name: &str| {
424 request
425 .headers
426 .iter()
427 .find(|(k, _)| k.eq_ignore_ascii_case(name))
428 .map(|(_, v)| v.clone())
429 };
430 assert_eq!(request.body, br#"{"ref":"refs/heads/main"}"#);
431 assert_eq!(
432 header("x-forge-signature-256"),
433 Some(sign(b"s3cret", &request.body))
434 );
435 assert_eq!(header("x-forge-delivery"), Some(delivery.guid.clone()));
436 assert_eq!(header("x-forge-event"), Some("push".into()));
437 assert_eq!(header("content-type"), Some("application/json".into()));
438
439 let done = state
440 .store
441 .delivery(&delivery.id)
442 .await
443 .expect("lookup")
444 .expect("row");
445 assert_eq!(done.state, DeliveryState::Ok);
446 assert_eq!(done.response_status, Some(204));
447 assert!(done.next_attempt_at.is_none());
448 assert_eq!(done.hook_id, hook.id);
449
450 // The stored request headers are what was sent, including the GUID and signature.
451 let stored: Value = serde_json::from_str(&done.request_headers).expect("json");
452 assert_eq!(stored["X-Forge-Delivery"], delivery.guid.as_str());
453 assert_eq!(
454 stored["X-Forge-Signature-256"],
455 sign(b"s3cret", &request.body).as_str()
456 );
457 assert_eq!(stored["X-Forge-Event"], "push");
458 }
459
460 #[tokio::test]
461 async fn failure_schedules_a_retry_and_six_attempts_give_up() {
462 let tmp = tempfile::tempdir().expect("tempdir");
463 let state = test_state(tmp.path()).await;
464 let (url, _seen) = server(500).await;
465 let (_hook, delivery, _dir) = queued(&state, &url).await;
466 let client = HookClient::new(local_policy()).expect("client");
467
468 process_due(&state, &client, &local_policy(), &now_stamp())
469 .await
470 .expect("process");
471 let failed = state
472 .store
473 .delivery(&delivery.id)
474 .await
475 .expect("lookup")
476 .expect("row");
477 assert_eq!(failed.state, DeliveryState::Failed);
478 assert_eq!(failed.response_status, Some(500));
479 let due = failed.next_attempt_at.clone().expect("retry scheduled");
480 let expected = stamp_after(Duration::from_secs(60));
481 // Both stamps are RFC 3339 UTC, so a text comparison orders them; allow a few seconds.
482 let parse = |s: &str| s.parse::<jiff::Timestamp>().expect("stamp").as_second();
483 let drift = parse(&due) - parse(&expected);
484 assert!(
485 drift.abs() <= 5,
486 "retry at {due}, expected about {expected}"
487 );
488
489 // Each later attempt is claimed at its due time. Attempts 2 through 6 all fail.
490 let far_future = stamp_after(Duration::from_secs(10 * 24 * 60 * 60));
491 let mut last = failed;
492 for attempt in 2..=6u32 {
493 process_due(&state, &client, &local_policy(), &far_future)
494 .await
495 .expect("process");
496 last = state
497 .store
498 .delivery(&delivery.id)
499 .await
500 .expect("lookup")
501 .expect("row");
502 assert_eq!(last.attempt, attempt);
503 }
504 assert_eq!(last.state, DeliveryState::GaveUp);
505 assert!(last.next_attempt_at.is_none());
506 }
507
508 #[tokio::test]
509 async fn invalid_url_is_a_failed_attempt_not_a_send() {
510 let tmp = tempfile::tempdir().expect("tempdir");
511 let state = test_state(tmp.path()).await;
512 // Loopback is not in the allow list of this policy, so validation refuses it.
513 let strict = WebhookPolicy::from_config(&forge_core::config::WebhookConfig::default())
514 .expect("policy");
515 let (_hook, delivery, _dir) = queued(&state, "https://127.0.0.1/hook").await;
516 let client = HookClient::new(strict.clone()).expect("client");
517 assert!(validate_url("https://127.0.0.1/hook", &strict).is_err());
518
519 process_due(&state, &client, &strict, &now_stamp())
520 .await
521 .expect("process");
522 let row = state
523 .store
524 .delivery(&delivery.id)
525 .await
526 .expect("lookup")
527 .expect("row");
528 assert_eq!(row.state, DeliveryState::Failed);
529 assert!(row.error.is_some());
530 assert!(row.next_attempt_at.is_some());
531 }
532
533 #[test]
534 fn sleep_stops_at_the_next_due_time_and_never_spins() {
535 use time::macros::datetime;
536
537 let now = datetime!(2026-10-15 12:00:00 UTC);
538 let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
539 // Due in 10 minutes: sleep until then.
540 let in_10m = stamp(now + time::Duration::minutes(10));
541 assert_eq!(sleep_for(&in_10m, now), Duration::from_secs(600));
542 // Due later than the backstop: wake at the backstop.
543 let in_1h = stamp(now + time::Duration::hours(1));
544 assert_eq!(sleep_for(&in_1h, now), BACKSTOP);
545 // Already due, or a lease that has just passed: wait the minimum.
546 let past = stamp(now - time::Duration::seconds(5));
547 assert_eq!(sleep_for(&past, now), MIN_SLEEP);
548 // A stamp that does not parse also waits the minimum.
549 assert_eq!(sleep_for("soon", now), MIN_SLEEP);
550 }
551
552 #[test]
553 fn backoff_schedule_matches_the_contract() {
554 assert_eq!(next_attempt_at(0), None);
555 assert!(next_attempt_at(1).is_some());
556 assert!(next_attempt_at(5).is_some());
557 assert_eq!(next_attempt_at(6), None);
558 assert_eq!(BACKOFF[0], Duration::from_secs(60));
559 assert_eq!(BACKOFF[4], Duration::from_secs(6 * 60 * 60));
560 }
561}