Vakyartha documentation

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3use std::collections::VecDeque;
4use std::sync::{Arc, Mutex};
5
6use tokio_util::sync::CancellationToken;
7
8use vak_core::Core;
9use vak_llm::stream;
10use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
11use vak_llm::{EventStream, LlmError, Provider};
12
13struct Scripted {
14 responses: Mutex<VecDeque<AssistantMessage>>,
15}
16
17#[async_trait::async_trait]
18impl Provider for Scripted {
19 fn name(&self) -> &str {
20 "scripted"
21 }
22
23 async fn stream(
24 &self,
25 _request: ChatRequest,
26 _cancel: CancellationToken,
27 ) -> Result<EventStream, LlmError> {
28 let next = self.responses.lock().unwrap().pop_front();
29 let (mut sink, rx) = stream::channel(64);
30 match next {
31 Some(m) => {
32 sink.push(stream::StreamEvent::Start { partial: m.clone() });
33 sink.close_message(m).await;
34 }
35 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
36 }
37 Ok(rx)
38 }
39}
40
41fn text(t: &str) -> AssistantMessage {
42 AssistantMessage {
43 content: vec![ContentBlock::text(t)],
44 stop_reason: vak_llm::types::StopReason::EndTurn,
45 usage: Usage {
46 input_tokens: 7,
47 output_tokens: 3,
48 ..Default::default()
49 },
50 model: "test-model".into(),
51 response_id: None,
52 }
53}
54
55type Captured = Arc<Mutex<Vec<(Option<String>, serde_json::Value)>>>;
56
57/// Records (authorization header, body) for every POST. With `fail_first`,
58/// the first `n` requests answer 503 before succeeding — exercising the
59/// delivery retry/backoff path.
60async fn spawn_receiver_failing(fail_first: u32) -> (String, Captured, Arc<Mutex<u32>>) {
61 let captured: Captured = Arc::new(Mutex::new(Vec::new()));
62 let seen = Arc::new(Mutex::new(0u32));
63 let cap = captured.clone();
64 let count = seen.clone();
65 let app = axum::Router::new().route(
66 "/hook",
67 axum::routing::post(
68 move |headers: axum::http::HeaderMap,
69 axum::Json(body): axum::Json<serde_json::Value>| async move {
70 let auth = headers
71 .get(axum::http::header::AUTHORIZATION)
72 .and_then(|v| v.to_str().ok())
73 .map(String::from);
74 cap.lock().unwrap().push((auth, body));
75 let mut n = count.lock().unwrap();
76 *n += 1;
77 if *n <= fail_first {
78 axum::http::StatusCode::BAD_GATEWAY
79 } else {
80 axum::http::StatusCode::OK
81 }
82 },
83 ),
84 );
85 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
86 let addr = listener.local_addr().unwrap();
87 tokio::spawn(async move {
88 axum::serve(listener, app).await.unwrap();
89 });
90 (format!("http://{addr}"), captured, seen)
91}
92
93async fn spawn_receiver() -> (String, Captured) {
94 let (base, captured, _seen) = spawn_receiver_failing(0).await;
95 (base, captured)
96}
97
98fn git_seed(cwd: &std::path::Path) {
99 let run = |args: &[&str]| {
100 let out = std::process::Command::new("git")
101 .args(args)
102 .current_dir(cwd)
103 .env("GIT_AUTHOR_NAME", "t")
104 .env("GIT_AUTHOR_EMAIL", "t@t")
105 .env("GIT_COMMITTER_NAME", "t")
106 .env("GIT_COMMITTER_EMAIL", "t@t")
107 .output()
108 .unwrap();
109 assert!(out.status.success(), "git {args:?} failed");
110 };
111 run(&["init", "-q"]);
112 std::fs::write(cwd.join("README.md"), "seed\n").unwrap();
113 run(&["add", "."]);
114 run(&["commit", "-q", "-m", "seed"]);
115}
116
117struct Gateway {
118 base: String,
119 token: String,
120 cwd: std::path::PathBuf,
121 _server: tokio::task::JoinHandle<()>,
122}
123
124/// Gateway enabled purely through trusted project config — exercises the
125/// `[gateway]` config path end to end.
126async fn spawn_with_config(provider: Arc<dyn Provider>, gateway_toml: &str) -> Gateway {
127 let dir = tempfile::tempdir().unwrap();
128 let cwd = dir.path().to_path_buf();
129 let project = cwd.join(".vak");
130 std::fs::create_dir_all(&project).unwrap();
131 let full_toml = format!("permission_mode = \"full-access\"\n{gateway_toml}");
132 std::fs::write(project.join("config.toml"), full_toml).unwrap();
133
134 vak_config::paths::isolate_home_for_tests();
135 let core = Core::new_with_trust(cwd.clone(), true).unwrap();
136 core.set_sessions_home(dir.path().join("home"));
137 core.set_provider_instance(provider);
138 core.set_permission_mode(vak_config::PermissionMode::FullAccess);
139 core.set_tool_worker_exe(std::path::PathBuf::from(env!(
140 "CARGO_BIN_EXE_vak-tool-worker"
141 )));
142 std::mem::forget(dir);
143
144 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
145 let addr = listener.local_addr().unwrap();
146 let (app, token) = vak_server::secured_router_with(core, false);
147 let server = tokio::spawn(async move {
148 axum::serve(listener, app).await.unwrap();
149 });
150 Gateway {
151 base: format!("http://{addr}"),
152 token,
153 cwd,
154 _server: server,
155 }
156}
157
158fn client_with(token: &str) -> reqwest::Client {
159 reqwest::ClientBuilder::new()
160 .default_headers({
161 let mut h = reqwest::header::HeaderMap::new();
162 h.insert(
163 reqwest::header::AUTHORIZATION,
164 format!("Bearer {token}").parse().unwrap(),
165 );
166 h
167 })
168 .build()
169 .unwrap()
170}
171
172async fn run_nightly(gw: &Gateway) -> () {
173 let client = client_with(&gw.token);
174 git_seed(&gw.cwd);
175
176 client
177 .post(format!("{}/tasks", gw.base))
178 .json(&serde_json::json!({
179 "name": "nightly",
180 "prompt": "what is the nightly status?",
181 "interval_secs": 3600,
182 "deliver_to": "webhook:ci"
183 }))
184 .send()
185 .await
186 .unwrap();
187 let tasks: serde_json::Value = client
188 .get(format!("{}/tasks", gw.base))
189 .send()
190 .await
191 .unwrap()
192 .json()
193 .await
194 .unwrap();
195 let tid = tasks["tasks"][0]["id"].as_str().unwrap().to_string();
196 client
197 .post(format!("{}/tasks/{tid}/run-now", gw.base))
198 .send()
199 .await
200 .unwrap();
201}
202
203#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
204async fn webhook_delivery_posts_run_output() {
205 let (rx_base, captured) = spawn_receiver().await;
206 let toml = format!(
207 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\n"
208 );
209 let gw = spawn_with_config(
210 Arc::new(Scripted {
211 responses: Mutex::new(VecDeque::from(vec![text("built ok")])),
212 }),
213 &toml,
214 )
215 .await;
216 run_nightly(&gw).await;
217
218 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
219 while std::time::Instant::now() < deadline {
220 if !captured.lock().unwrap().is_empty() {
221 break;
222 }
223 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
224 }
225 let cap = captured.lock().unwrap();
226 assert_eq!(cap.len(), 1, "exactly one delivery expected");
227 let (auth, body) = &cap[0];
228 assert!(auth.is_none(), "no token_env configured: must post bare");
229 let text = body["text"].as_str().unwrap_or_default();
230 assert!(text.contains("routine 'nightly' finished"), "{body}");
231 assert!(text.contains("built ok"), "real answer delivered: {body}");
232 assert_eq!(body["delivery"]["kind"], "task_summary");
233 assert_eq!(body["delivery"]["fallback_markdown"], body["text"]);
234 assert_eq!(body["job_id"], body["delivery"]["job_id"]);
235}
236
237#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
238async fn webhook_missing_token_fails_closed() {
239 let (rx_base, captured) = spawn_receiver().await;
240 let toml = format!(
241 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\ntoken_env = \"GATEWAY_TEST_UNSET_TOKEN\"\n"
242 );
243 let gw = spawn_with_config(
244 Arc::new(Scripted {
245 responses: Mutex::new(VecDeque::from(vec![text("secret output")])),
246 }),
247 &toml,
248 )
249 .await;
250
251 // Sanity: the credential really is absent from this environment.
252 assert!(std::env::var("GATEWAY_TEST_UNSET_TOKEN").is_err());
253
254 run_nightly(&gw).await;
255 let client = client_with(&gw.token);
256
257 // The run itself still completes and records its answer; only the
258 // delivery is withheld.
259 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
260 loop {
261 assert!(
262 std::time::Instant::now() < deadline,
263 "task never recorded a summary"
264 );
265 let t: serde_json::Value = client
266 .get(format!("{}/tasks", gw.base))
267 .send()
268 .await
269 .unwrap()
270 .json()
271 .await
272 .unwrap();
273 if t["tasks"][0]["last_summary"] == "secret output" {
274 break;
275 }
276 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
277 }
278 tokio::time::sleep(std::time::Duration::from_millis(500)).await;
279 assert!(
280 captured.lock().unwrap().is_empty(),
281 "missing credential must fail closed: nothing posted"
282 );
283 let outbox = gw.cwd.join("home/delivery/jobs");
284 let pending = std::fs::read_dir(&outbox)
285 .unwrap()
286 .filter_map(Result::ok)
287 .find_map(|entry| std::fs::read_to_string(entry.path()).ok())
288 .expect("failed delivery remains in the durable outbox");
289 assert!(pending.contains("\"state\":\"pending\""), "{pending}");
290 assert!(
291 pending.contains("secret output"),
292 "exact output survives: {pending}"
293 );
294 let inbox = std::fs::read_to_string(gw.cwd.join("home/inbox.jsonl")).unwrap();
295 assert!(
296 inbox.contains("secret output"),
297 "inbox is recorded before push"
298 );
299}
300
301#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
302async fn webhook_retries_transient_5xx_and_succeeds() {
303 // One 503, then success: delivery must survive the blip.
304 let (rx_base, captured, seen) = spawn_receiver_failing(1).await;
305 let toml = format!(
306 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\n"
307 );
308 let gw = spawn_with_config(
309 Arc::new(Scripted {
310 responses: Mutex::new(VecDeque::from(vec![text("retry built ok")])),
311 }),
312 &toml,
313 )
314 .await;
315 run_nightly(&gw).await;
316
317 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
318 loop {
319 assert!(
320 std::time::Instant::now() < deadline,
321 "delivery never landed despite retries"
322 );
323 let n = *seen.lock().unwrap();
324 if n >= 2 {
325 break;
326 }
327 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
328 }
329 tokio::time::sleep(std::time::Duration::from_millis(600)).await;
330 {
331 let cap = captured.lock().unwrap();
332 // This receiver records EVERY post including the one that answered
333 // 503, so two captures = attempt(503) + retry(200). Exactly one
334 // retry happened and it carried the same payload.
335 assert_eq!(cap.len(), 2, "one failed attempt then one successful retry");
336 assert_eq!(cap[0].1, cap[1].1, "retry reuses the identical payload");
337 }
338
339 // A permanently failing receiver (all attempts 5xx) reports failure
340 // instead of hanging or pretending success.
341 let (rx_base2, _captured2, seen2) = spawn_receiver_failing(u32::MAX).await;
342 let toml2 = format!(
343 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base2}/hook\"\n"
344 );
345 let gw2 = spawn_with_config(
346 Arc::new(Scripted {
347 responses: Mutex::new(VecDeque::from(vec![text("never delivered")])),
348 }),
349 &toml2,
350 )
351 .await;
352 run_nightly(&gw2).await;
353 let deadline2 = std::time::Instant::now() + std::time::Duration::from_secs(20);
354 loop {
355 assert!(
356 std::time::Instant::now() < deadline2,
357 "webhook exhausted retries without giving up"
358 );
359 if *seen2.lock().unwrap() >= 3 {
360 break;
361 }
362 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
363 }
364}
365

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3use std::collections::VecDeque;
4use std::sync::{Arc, Mutex};
5
6use tokio_util::sync::CancellationToken;
7
8use vak_core::Core;
9use vak_llm::stream;
10use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
11use vak_llm::{EventStream, LlmError, Provider};
12
13struct Scripted {
14 responses: Mutex<VecDeque<AssistantMessage>>,
15}
16
17#[async_trait::async_trait]
18impl Provider for Scripted {
19 fn name(&self) -> &str {
20 "scripted"
21 }
22
23 async fn stream(
24 &self,
25 _request: ChatRequest,
26 _cancel: CancellationToken,
27 ) -> Result<EventStream, LlmError> {
28 let next = self.responses.lock().unwrap().pop_front();
29 let (mut sink, rx) = stream::channel(64);
30 match next {
31 Some(m) => {
32 sink.push(stream::StreamEvent::Start { partial: m.clone() });
33 sink.close_message(m).await;
34 }
35 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
36 }
37 Ok(rx)
38 }
39}
40
41fn text(t: &str) -> AssistantMessage {
42 AssistantMessage {
43 content: vec![ContentBlock::text(t)],
44 stop_reason: vak_llm::types::StopReason::EndTurn,
45 usage: Usage {
46 input_tokens: 7,
47 output_tokens: 3,
48 ..Default::default()
49 },
50 model: "test-model".into(),
51 response_id: None,
52 }
53}
54
55type Captured = Arc<Mutex<Vec<(Option<String>, serde_json::Value)>>>;
56
57/// Records (authorization header, body) for every POST. With `fail_first`,
58/// the first `n` requests answer 503 before succeeding — exercising the
59/// delivery retry/backoff path.
60async fn spawn_receiver_failing(fail_first: u32) -> (String, Captured, Arc<Mutex<u32>>) {
61 let captured: Captured = Arc::new(Mutex::new(Vec::new()));
62 let seen = Arc::new(Mutex::new(0u32));
63 let cap = captured.clone();
64 let count = seen.clone();
65 let app = axum::Router::new().route(
66 "/hook",
67 axum::routing::post(
68 move |headers: axum::http::HeaderMap,
69 axum::Json(body): axum::Json<serde_json::Value>| async move {
70 let auth = headers
71 .get(axum::http::header::AUTHORIZATION)
72 .and_then(|v| v.to_str().ok())
73 .map(String::from);
74 cap.lock().unwrap().push((auth, body));
75 let mut n = count.lock().unwrap();
76 *n += 1;
77 if *n <= fail_first {
78 axum::http::StatusCode::BAD_GATEWAY
79 } else {
80 axum::http::StatusCode::OK
81 }
82 },
83 ),
84 );
85 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
86 let addr = listener.local_addr().unwrap();
87 tokio::spawn(async move {
88 axum::serve(listener, app).await.unwrap();
89 });
90 (format!("http://{addr}"), captured, seen)
91}
92
93async fn spawn_receiver() -> (String, Captured) {
94 let (base, captured, _seen) = spawn_receiver_failing(0).await;
95 (base, captured)
96}
97
98fn git_seed(cwd: &std::path::Path) {
99 let run = |args: &[&str]| {
100 let out = std::process::Command::new("git")
101 .args(args)
102 .current_dir(cwd)
103 .env("GIT_AUTHOR_NAME", "t")
104 .env("GIT_AUTHOR_EMAIL", "t@t")
105 .env("GIT_COMMITTER_NAME", "t")
106 .env("GIT_COMMITTER_EMAIL", "t@t")
107 .output()
108 .unwrap();
109 assert!(out.status.success(), "git {args:?} failed");
110 };
111 run(&["init", "-q"]);
112 std::fs::write(cwd.join("README.md"), "seed\n").unwrap();
113 run(&["add", "."]);
114 run(&["commit", "-q", "-m", "seed"]);
115}
116
117struct Gateway {
118 base: String,
119 token: String,
120 cwd: std::path::PathBuf,
121 _server: tokio::task::JoinHandle<()>,
122}
123
124/// Gateway enabled purely through trusted project config — exercises the
125/// `[gateway]` config path end to end.
126async fn spawn_with_config(provider: Arc<dyn Provider>, gateway_toml: &str) -> Gateway {
127 let dir = tempfile::tempdir().unwrap();
128 let cwd = dir.path().to_path_buf();
129 let project = cwd.join(".vak");
130 std::fs::create_dir_all(&project).unwrap();
131 let full_toml = format!("permission_mode = \"full-access\"\n{gateway_toml}");
132 std::fs::write(project.join("config.toml"), full_toml).unwrap();
133
134 vak_config::paths::isolate_home_for_tests();
135 let core = Core::new_with_trust(cwd.clone(), true).unwrap();
136 core.set_sessions_home(dir.path().join("home"));
137 core.set_provider_instance(provider);
138 core.set_permission_mode(vak_config::PermissionMode::FullAccess);
139 core.set_tool_worker_exe(std::path::PathBuf::from(env!(
140 "CARGO_BIN_EXE_vak-tool-worker"
141 )));
142 std::mem::forget(dir);
143
144 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
145 let addr = listener.local_addr().unwrap();
146 let (app, token) = vak_server::secured_router_with(core, false);
147 let server = tokio::spawn(async move {
148 axum::serve(listener, app).await.unwrap();
149 });
150 Gateway {
151 base: format!("http://{addr}"),
152 token,
153 cwd,
154 _server: server,
155 }
156}
157
158fn client_with(token: &str) -> reqwest::Client {
159 reqwest::ClientBuilder::new()
160 .default_headers({
161 let mut h = reqwest::header::HeaderMap::new();
162 h.insert(
163 reqwest::header::AUTHORIZATION,
164 format!("Bearer {token}").parse().unwrap(),
165 );
166 h
167 })
168 .build()
169 .unwrap()
170}
171
172async fn run_nightly(gw: &Gateway) -> () {
173 let client = client_with(&gw.token);
174 git_seed(&gw.cwd);
175
176 client
177 .post(format!("{}/tasks", gw.base))
178 .json(&serde_json::json!({
179 "name": "nightly",
180 "prompt": "what is the nightly status?",
181 "interval_secs": 3600,
182 "deliver_to": "webhook:ci"
183 }))
184 .send()
185 .await
186 .unwrap();
187 let tasks: serde_json::Value = client
188 .get(format!("{}/tasks", gw.base))
189 .send()
190 .await
191 .unwrap()
192 .json()
193 .await
194 .unwrap();
195 let tid = tasks["tasks"][0]["id"].as_str().unwrap().to_string();
196 client
197 .post(format!("{}/tasks/{tid}/run-now", gw.base))
198 .send()
199 .await
200 .unwrap();
201}
202
203#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
204async fn webhook_delivery_posts_run_output() {
205 let (rx_base, captured) = spawn_receiver().await;
206 let toml = format!(
207 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\n"
208 );
209 let gw = spawn_with_config(
210 Arc::new(Scripted {
211 responses: Mutex::new(VecDeque::from(vec![text("built ok")])),
212 }),
213 &toml,
214 )
215 .await;
216 run_nightly(&gw).await;
217
218 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
219 while std::time::Instant::now() < deadline {
220 if !captured.lock().unwrap().is_empty() {
221 break;
222 }
223 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
224 }
225 let cap = captured.lock().unwrap();
226 assert_eq!(cap.len(), 1, "exactly one delivery expected");
227 let (auth, body) = &cap[0];
228 assert!(auth.is_none(), "no token_env configured: must post bare");
229 let text = body["text"].as_str().unwrap_or_default();
230 assert!(text.contains("routine 'nightly' finished"), "{body}");
231 assert!(text.contains("built ok"), "real answer delivered: {body}");
232 assert_eq!(body["delivery"]["kind"], "task_summary");
233 assert_eq!(body["delivery"]["fallback_markdown"], body["text"]);
234 assert_eq!(body["job_id"], body["delivery"]["job_id"]);
235}
236
237#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
238async fn webhook_missing_token_fails_closed() {
239 let (rx_base, captured) = spawn_receiver().await;
240 let toml = format!(
241 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\ntoken_env = \"GATEWAY_TEST_UNSET_TOKEN\"\n"
242 );
243 let gw = spawn_with_config(
244 Arc::new(Scripted {
245 responses: Mutex::new(VecDeque::from(vec![text("secret output")])),
246 }),
247 &toml,
248 )
249 .await;
250
251 // Sanity: the credential really is absent from this environment.
252 assert!(std::env::var("GATEWAY_TEST_UNSET_TOKEN").is_err());
253
254 run_nightly(&gw).await;
255 let client = client_with(&gw.token);
256
257 // The run itself still completes and records its answer; only the
258 // delivery is withheld.
259 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
260 loop {
261 assert!(
262 std::time::Instant::now() < deadline,
263 "task never recorded a summary"
264 );
265 let t: serde_json::Value = client
266 .get(format!("{}/tasks", gw.base))
267 .send()
268 .await
269 .unwrap()
270 .json()
271 .await
272 .unwrap();
273 if t["tasks"][0]["last_summary"] == "secret output" {
274 break;
275 }
276 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
277 }
278 tokio::time::sleep(std::time::Duration::from_millis(500)).await;
279 assert!(
280 captured.lock().unwrap().is_empty(),
281 "missing credential must fail closed: nothing posted"
282 );
283 let outbox = gw.cwd.join("home/delivery/jobs");
284 let pending = std::fs::read_dir(&outbox)
285 .unwrap()
286 .filter_map(Result::ok)
287 .find_map(|entry| std::fs::read_to_string(entry.path()).ok())
288 .expect("failed delivery remains in the durable outbox");
289 assert!(pending.contains("\"state\":\"pending\""), "{pending}");
290 assert!(
291 pending.contains("secret output"),
292 "exact output survives: {pending}"
293 );
294 let inbox = std::fs::read_to_string(gw.cwd.join("home/inbox.jsonl")).unwrap();
295 assert!(
296 inbox.contains("secret output"),
297 "inbox is recorded before push"
298 );
299}
300
301#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
302async fn webhook_retries_transient_5xx_and_succeeds() {
303 // One 503, then success: delivery must survive the blip.
304 let (rx_base, captured, seen) = spawn_receiver_failing(1).await;
305 let toml = format!(
306 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base}/hook\"\n"
307 );
308 let gw = spawn_with_config(
309 Arc::new(Scripted {
310 responses: Mutex::new(VecDeque::from(vec![text("retry built ok")])),
311 }),
312 &toml,
313 )
314 .await;
315 run_nightly(&gw).await;
316
317 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
318 loop {
319 assert!(
320 std::time::Instant::now() < deadline,
321 "delivery never landed despite retries"
322 );
323 let n = *seen.lock().unwrap();
324 if n >= 2 {
325 break;
326 }
327 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
328 }
329 tokio::time::sleep(std::time::Duration::from_millis(600)).await;
330 {
331 let cap = captured.lock().unwrap();
332 // This receiver records EVERY post including the one that answered
333 // 503, so two captures = attempt(503) + retry(200). Exactly one
334 // retry happened and it carried the same payload.
335 assert_eq!(cap.len(), 2, "one failed attempt then one successful retry");
336 assert_eq!(cap[0].1, cap[1].1, "retry reuses the identical payload");
337 }
338
339 // A permanently failing receiver (all attempts 5xx) reports failure
340 // instead of hanging or pretending success.
341 let (rx_base2, _captured2, seen2) = spawn_receiver_failing(u32::MAX).await;
342 let toml2 = format!(
343 "[gateway]\nenabled = true\n[gateway.outbound.webhooks.ci]\nurl = \"{rx_base2}/hook\"\n"
344 );
345 let gw2 = spawn_with_config(
346 Arc::new(Scripted {
347 responses: Mutex::new(VecDeque::from(vec![text("never delivered")])),
348 }),
349 &toml2,
350 )
351 .await;
352 run_nightly(&gw2).await;
353 let deadline2 = std::time::Instant::now() + std::time::Duration::from_secs(20);
354 loop {
355 assert!(
356 std::time::Instant::now() < deadline2,
357 "webhook exhausted retries without giving up"
358 );
359 if *seen2.lock().unwrap() >= 3 {
360 break;
361 }
362 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
363 }
364}
365

Webhook Delivery

Indexing the workspace…

Webhook Delivery