forge
ExploreLog in

Tahinli/forge

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