forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
18.0 KiB
1//! The runner against a mock forge (hyper on a local TCP port) with `HostExecutor`.
2//!
3//! Enabled for `cargo test` by the self dev-dependency in Cargo.toml, which turns on the
4//! `test-host-executor` feature. There is no cfg gate here, so a build without the feature fails
5//! to compile instead of silently skipping these tests.
6
7use std::collections::{BTreeMap, HashMap};
8use std::convert::Infallible;
9use std::path::{Path, PathBuf};
10use std::process::Command as StdCommand;
11use std::sync::{Arc, Mutex};
12use std::time::{Duration, Instant};
13
14use bytes::Bytes;
15use forge_ci::proto::{FinishRequest, JobAssignment, Outcome, RegisterResponse};
16use forge_runner::config::{Limits, RunnerConfig};
17use forge_runner::executor::HostExecutor;
18use forge_runner::logs::PumpConfig;
19use forge_runner::runner::{RunOptions, Runner};
20use http_body_util::{BodyExt, Full};
21use hyper::body::Incoming;
22use hyper::header::{AUTHORIZATION, HeaderMap};
23use hyper::server::conn::http1;
24use hyper::service::service_fn;
25use hyper::{Method, Request, Response, StatusCode};
26use hyper_util::rt::TokioIo;
27use tokio::net::TcpListener;
28use tokio_util::sync::CancellationToken;
29
30const RUNNER_SECRET: &str = "runner-secret-for-tests";
31const SECRET_VALUE: &str = "s3cr3t-value-echoed-by-job";
32const LOG_SIZE: &str = "X-Forge-Log-Size";
33const LOG_OFFSET: &str = "X-Forge-Log-Offset";
34
35/// What the mock forge saw and what it should do.
36#[derive(Default)]
37struct MockState {
38 /// Jobs handed out by claim, in order.
39 pending: Vec<JobAssignment>,
40 /// Every job by id, so job calls can check the job token.
41 by_id: HashMap<String, JobAssignment>,
42 /// The stored log, exactly as the forge would keep it.
43 log: Vec<u8>,
44 log_posts: usize,
45 conflicts: usize,
46 /// On log POST number `.0`, the forge first stores the first `.1` bytes of the body and then
47 /// answers 409, as if an earlier attempt had been applied but its response was lost.
48 ghost: Option<(usize, usize)>,
49 /// On log POST number `n`, the forge answers 200 but stores nothing, as if the batch was
50 /// dropped after it was acknowledged. A later POST then gets 409 with the smaller size.
51 drop_post: Option<usize>,
52 heartbeats: usize,
53 /// Heartbeats from number `n` on answer `cancel: true`.
54 cancel_from_heartbeat: Option<usize>,
55 artifacts: BTreeMap<String, Vec<u8>>,
56 finished: HashMap<String, FinishRequest>,
57}
58
59type Shared = Arc<Mutex<MockState>>;
60
61struct Reply {
62 status: StatusCode,
63 log_size: Option<u64>,
64 body: Vec<u8>,
65}
66
67impl Reply {
68 fn empty(status: StatusCode) -> Self {
69 Self {
70 status,
71 log_size: None,
72 body: Vec::new(),
73 }
74 }
75
76 fn json<T: serde::Serialize>(value: &T) -> Self {
77 Self {
78 status: StatusCode::OK,
79 log_size: None,
80 body: serde_json::to_vec(value).unwrap_or_default(),
81 }
82 }
83}
84
85fn bearer(headers: &HeaderMap) -> String {
86 headers
87 .get(AUTHORIZATION)
88 .and_then(|v| v.to_str().ok())
89 .and_then(|v| v.strip_prefix("Bearer "))
90 .unwrap_or_default()
91 .to_owned()
92}
93
94fn route(
95 st: &mut MockState,
96 method: &Method,
97 path: &str,
98 auth: &str,
99 offset: Option<u64>,
100 body: &[u8],
101) -> Reply {
102 let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
103 match (method.as_str(), parts.as_slice()) {
104 ("POST", ["api", "v1", "ci", "runners", "register"]) => Reply::json(&RegisterResponse {
105 runner_id: "runner-1".to_owned(),
106 runner_secret: RUNNER_SECRET.to_owned(),
107 }),
108 ("POST", ["api", "v1", "ci", "runners", "claim"]) => {
109 if auth != RUNNER_SECRET {
110 return Reply::empty(StatusCode::UNAUTHORIZED);
111 }
112 if st.pending.is_empty() {
113 return Reply::empty(StatusCode::NO_CONTENT);
114 }
115 let job = st.pending.remove(0);
116 Reply::json(&job)
117 }
118 ("POST" | "PUT" | "GET", ["api", "v1", "ci", "jobs", job_id, rest @ ..]) => {
119 let Some(job) = st.by_id.get(*job_id) else {
120 return Reply::empty(StatusCode::NOT_FOUND);
121 };
122 if auth != job.job_token {
123 return Reply::empty(StatusCode::UNAUTHORIZED);
124 }
125 match (method.as_str(), rest) {
126 ("POST", ["logs"]) => log_post(st, offset, body),
127 ("POST", ["heartbeat"]) => {
128 st.heartbeats += 1;
129 let cancel = st.cancel_from_heartbeat.is_some_and(|n| st.heartbeats >= n);
130 Reply::json(&forge_ci::proto::HeartbeatResponse { cancel })
131 }
132 ("PUT", ["artifacts", name]) => {
133 let valid = !name.is_empty()
134 && name.len() <= 128
135 && !name.starts_with('.')
136 && name
137 .bytes()
138 .all(|b| b.is_ascii_alphanumeric() || b"._+-".contains(&b));
139 if !valid {
140 return Reply::empty(StatusCode::BAD_REQUEST);
141 }
142 st.artifacts.insert((*name).to_owned(), body.to_vec());
143 Reply::empty(StatusCode::OK)
144 }
145 ("POST", ["finish"]) => match serde_json::from_slice::<FinishRequest>(body) {
146 Ok(req) => {
147 st.finished.insert((*job_id).to_owned(), req);
148 Reply::empty(StatusCode::OK)
149 }
150 Err(_) => Reply::empty(StatusCode::BAD_REQUEST),
151 },
152 _ => Reply::empty(StatusCode::NOT_FOUND),
153 }
154 }
155 _ => Reply::empty(StatusCode::NOT_FOUND),
156 }
157}
158
159/// The forge's log rules: append at size, accept an idempotent retry, otherwise 409 with the size.
160fn log_post(st: &mut MockState, offset: Option<u64>, body: &[u8]) -> Reply {
161 let Some(offset) = offset else {
162 return Reply::empty(StatusCode::BAD_REQUEST);
163 };
164 st.log_posts += 1;
165 if st.drop_post == Some(st.log_posts) {
166 return Reply::empty(StatusCode::OK);
167 }
168 if let Some((nth, prefix)) = st.ghost
169 && st.log_posts == nth
170 {
171 let n = prefix.min(body.len());
172 st.log.extend_from_slice(&body[..n]);
173 }
174 let size = st.log.len() as u64;
175 if offset == size {
176 st.log.extend_from_slice(body);
177 return Reply::empty(StatusCode::OK);
178 }
179 let end = offset + body.len() as u64;
180 if end == size && st.log[offset as usize..] == *body {
181 return Reply::empty(StatusCode::OK);
182 }
183 st.conflicts += 1;
184 Reply {
185 status: StatusCode::CONFLICT,
186 log_size: Some(size),
187 body: Vec::new(),
188 }
189}
190
191async fn handle(
192 state: Shared,
193 req: Request<Incoming>,
194) -> Result<Response<Full<Bytes>>, Infallible> {
195 let method = req.method().clone();
196 let path = req.uri().path().to_owned();
197 let auth = bearer(req.headers());
198 let offset = req
199 .headers()
200 .get(LOG_OFFSET)
201 .and_then(|v| v.to_str().ok())
202 .and_then(|v| v.parse::<u64>().ok());
203 let body = req
204 .into_body()
205 .collect()
206 .await
207 .map(|c| c.to_bytes().to_vec())
208 .unwrap_or_default();
209 // Slow claims down a little so an idle runner does not spin.
210 if path.ends_with("/claim") {
211 tokio::time::sleep(Duration::from_millis(20)).await;
212 }
213 let reply = {
214 let mut st = match state.lock() {
215 Ok(st) => st,
216 Err(poisoned) => poisoned.into_inner(),
217 };
218 route(&mut st, &method, &path, &auth, offset, &body)
219 };
220 let mut builder = Response::builder().status(reply.status);
221 if let Some(size) = reply.log_size {
222 builder = builder.header(LOG_SIZE, size.to_string());
223 }
224 let response = builder
225 .header("content-type", "application/json")
226 .body(Full::new(Bytes::from(reply.body)))
227 .unwrap_or_else(|_| Response::new(Full::new(Bytes::new())));
228 Ok(response)
229}
230
231/// Starts the mock forge on a random local port and returns its base URL.
232async fn start_mock(state: Shared) -> String {
233 let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
234 let addr = listener.local_addr().expect("addr");
235 tokio::spawn(async move {
236 loop {
237 let Ok((stream, _)) = listener.accept().await else {
238 break;
239 };
240 let st = state.clone();
241 tokio::spawn(async move {
242 let _ = http1::Builder::new()
243 .serve_connection(
244 TokioIo::new(stream),
245 service_fn(move |req| handle(st.clone(), req)),
246 )
247 .await;
248 });
249 }
250 });
251 format!("http://{addr}")
252}
253
254/// A git repo with one commit. Returns the commit id.
255fn make_repo(dir: &Path) -> String {
256 let repo = dir.join("repo");
257 std::fs::create_dir_all(&repo).expect("mkdir");
258 std::fs::write(repo.join("README.md"), "hello\n").expect("write");
259 let git = |args: &[&str]| {
260 let status = StdCommand::new("git")
261 .arg("-C")
262 .arg(&repo)
263 .args(args)
264 .status()
265 .expect("run git");
266 assert!(status.success(), "git {args:?} failed");
267 };
268 git(&["init", "-q"]);
269 git(&["add", "-A"]);
270 git(&[
271 "-c",
272 "user.name=test",
273 "-c",
274 "user.email=test@example.com",
275 "commit",
276 "-q",
277 "-m",
278 "init",
279 ]);
280 // Let a file:// fetch ask for the exact commit by id.
281 git(&["config", "uploadpack.allowAnySHA1InWant", "true"]);
282 let out = StdCommand::new("git")
283 .arg("-C")
284 .arg(&repo)
285 .args(["rev-parse", "HEAD"])
286 .output()
287 .expect("rev-parse");
288 String::from_utf8_lossy(&out.stdout).trim().to_owned()
289}
290
291fn job(job_id: &str, commit: &str, clone_url: &str, run: &str) -> JobAssignment {
292 JobAssignment {
293 job_id: job_id.to_owned(),
294 job_token: format!("token-for-{job_id}"),
295 repo: "owner/repo".to_owned(),
296 clone_url: clone_url.to_owned(),
297 commit: commit.to_owned(),
298 git_ref: "refs/heads/main".to_owned(),
299 job_name: "build".to_owned(),
300 image: "docker.io/library/alpine:3".to_owned(),
301 run: run.to_owned(),
302 env: BTreeMap::from([("GREETING".to_owned(), "hello-env".to_owned())]),
303 secrets: BTreeMap::from([("MY_TOKEN".to_owned(), SECRET_VALUE.to_owned())]),
304 artifacts: vec!["dist/**".to_owned()],
305 timeout_minutes: 5,
306 network: false,
307 }
308}
309
310fn config(url: &str, work_dir: PathBuf) -> RunnerConfig {
311 RunnerConfig {
312 url: url.to_owned(),
313 runner_id: "runner-1".to_owned(),
314 runner_secret: RUNNER_SECRET.to_owned(),
315 name: "test-runner".to_owned(),
316 labels: vec!["linux".to_owned()],
317 work_dir,
318 concurrency: 1,
319 limits: Limits::default(),
320 executor: forge_runner::config::ExecutorKind::default(),
321 images: forge_runner::config::ImagesConfig::default(),
322 sandbox: forge_runner::config::SandboxConfig::default(),
323 registries: Vec::new(),
324 }
325}
326
327fn fast_options() -> RunOptions {
328 RunOptions {
329 heartbeat_every: Duration::from_millis(50),
330 claim_backoff: Duration::from_millis(50),
331 pump: PumpConfig {
332 flush_every: Duration::from_millis(50),
333 retries: 3,
334 retry_delay: Duration::from_millis(10),
335 },
336 timeout_override: None,
337 }
338}
339
340/// Polls `check` until it returns true or the deadline passes.
341async fn wait_until(what: &str, mut check: impl FnMut() -> bool) {
342 let deadline = Instant::now() + Duration::from_secs(60);
343 while !check() {
344 assert!(Instant::now() < deadline, "timed out waiting for {what}");
345 tokio::time::sleep(Duration::from_millis(20)).await;
346 }
347}
348
349struct Harness {
350 state: Shared,
351 stop: CancellationToken,
352 runner_task: tokio::task::JoinHandle<()>,
353 _dir: tempfile::TempDir,
354}
355
356/// Starts the mock, a runner with HostExecutor, and queues one job.
357async fn start(
358 run: &str,
359 mut options: RunOptions,
360 setup: impl FnOnce(&mut MockState),
361 timeout: Option<Duration>,
362) -> (Harness, String) {
363 let dir = tempfile::tempdir().expect("tempdir");
364 let commit = make_repo(dir.path());
365 let clone_url = format!("file://{}", dir.path().join("repo").display());
366 let job_id = "job-1";
367 let state: Shared = Arc::new(Mutex::new(MockState::default()));
368 {
369 let mut st = state.lock().expect("lock");
370 let j = job(job_id, &commit, &clone_url, run);
371 st.by_id.insert(j.job_id.clone(), j.clone());
372 st.pending.push(j);
373 setup(&mut st);
374 }
375 let url = start_mock(state.clone()).await;
376 options.timeout_override = timeout;
377 let cfg = config(&url, dir.path().join("work"));
378 let runner = Runner::new(cfg, Arc::new(HostExecutor), options).expect("runner");
379 let stop = CancellationToken::new();
380 let runner_task = {
381 let stop = stop.clone();
382 tokio::spawn(async move { runner.run(stop).await })
383 };
384 (
385 Harness {
386 state,
387 stop,
388 runner_task,
389 _dir: dir,
390 },
391 job_id.to_owned(),
392 )
393}
394
395async fn finish_of(h: &Harness, job_id: &str) -> FinishRequest {
396 let state = h.state.clone();
397 let id = job_id.to_owned();
398 wait_until("the job to finish", || {
399 state
400 .lock()
401 .map(|st| st.finished.contains_key(&id))
402 .unwrap_or(false)
403 })
404 .await;
405 h.state
406 .lock()
407 .expect("lock")
408 .finished
409 .get(job_id)
410 .cloned()
411 .expect("finish recorded")
412}
413
414async fn shut_down(h: Harness) {
415 h.stop.cancel();
416 let _ = tokio::time::timeout(Duration::from_secs(20), h.runner_task).await;
417}
418
419fn log_text(h: &Harness) -> String {
420 String::from_utf8_lossy(&h.state.lock().expect("lock").log).into_owned()
421}
422
423#[tokio::test]
424async fn full_flow_masks_secrets_and_uploads_artifacts() {
425 let run = "mkdir -p dist && echo built > dist/out.txt && echo \"token is $MY_TOKEN\" && \
426 echo \"b64 $(printf %s \"$MY_TOKEN\" | base64)\" && echo \"env is $GREETING\"";
427 let (h, job_id) = start(run, fast_options(), |_| {}, None).await;
428
429 let finish = finish_of(&h, &job_id).await;
430 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
431 assert_eq!(finish.exit_code, Some(0));
432
433 let log = log_text(&h);
434 assert!(log.starts_with("forge-runner "), "info line first: {log:?}");
435 assert!(log.contains("token is ***"), "masked: {log:?}");
436 assert!(log.contains("env is hello-env"), "{log:?}");
437 assert!(!log.contains(SECRET_VALUE), "secret leaked into the log");
438 let b64 = base64_of(SECRET_VALUE);
439 assert!(
440 !log.contains(&b64),
441 "base64 of the secret leaked into the log"
442 );
443
444 let artifact = h
445 .state
446 .lock()
447 .expect("lock")
448 .artifacts
449 .get("dist_out.txt")
450 .cloned();
451 assert_eq!(artifact.as_deref(), Some(&b"built\n"[..]));
452
453 shut_down(h).await;
454}
455
456fn base64_of(text: &str) -> String {
457 use base64::Engine;
458 base64::engine::general_purpose::STANDARD.encode(text)
459}
460
461#[tokio::test]
462async fn heartbeat_cancel_stops_the_job() {
463 let mut options = fast_options();
464 options.heartbeat_every = Duration::from_millis(50);
465 let (h, job_id) = start(
466 "echo start; sleep 60",
467 options,
468 |st| st.cancel_from_heartbeat = Some(2),
469 None,
470 )
471 .await;
472
473 let started = Instant::now();
474 let finish = finish_of(&h, &job_id).await;
475 assert_eq!(finish.outcome, Outcome::Cancelled, "{finish:?}");
476 assert!(
477 started.elapsed() < Duration::from_secs(30),
478 "the job was not stopped"
479 );
480 shut_down(h).await;
481}
482
483#[tokio::test]
484async fn timeout_fails_the_job() {
485 let (h, job_id) = start(
486 "sleep 60",
487 fast_options(),
488 |_| {},
489 Some(Duration::from_millis(400)),
490 )
491 .await;
492
493 let finish = finish_of(&h, &job_id).await;
494 assert_eq!(finish.outcome, Outcome::Failure, "{finish:?}");
495 assert!(
496 finish
497 .message
498 .as_deref()
499 .unwrap_or_default()
500 .starts_with("timed out"),
501 "{finish:?}"
502 );
503 shut_down(h).await;
504}
505
506#[tokio::test]
507async fn log_offset_mismatch_resyncs_without_loss_or_duplicates() {
508 // Log POST #2 is applied by the forge for its first 3 bytes, then answered 409. The runner
509 // must drop those 3 bytes and continue, so the stored log has every line exactly once.
510 let run = "echo first; sleep 0.3; echo second; sleep 0.3; echo third";
511 let (h, job_id) = start(run, fast_options(), |st| st.ghost = Some((2, 3)), None).await;
512
513 let finish = finish_of(&h, &job_id).await;
514 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
515
516 let log = log_text(&h);
517 assert!(log.ends_with("first\nsecond\nthird\n"), "log was {log:?}");
518 assert_eq!(
519 log.matches("second").count(),
520 1,
521 "duplicate or lost bytes: {log:?}"
522 );
523 let conflicts = h.state.lock().expect("lock").conflicts;
524 assert!(conflicts >= 1, "the 409 path never ran");
525 shut_down(h).await;
526}
527
528#[tokio::test]
529async fn acknowledged_batch_the_forge_lost_is_resent_in_order() {
530 // Log POST #2 is acknowledged but never stored. The next POST gets 409 with the smaller size,
531 // and the runner must resend the lost bytes from its retained buffer, with no gap marker.
532 let run = "echo first; sleep 0.3; echo second; sleep 0.3; echo third";
533 let (h, job_id) = start(run, fast_options(), |st| st.drop_post = Some(2), None).await;
534
535 let finish = finish_of(&h, &job_id).await;
536 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
537
538 let log = log_text(&h);
539 assert!(log.ends_with("first\nsecond\nthird\n"), "log was {log:?}");
540 assert_eq!(log.matches("second").count(), 1, "duplicate bytes: {log:?}");
541 assert!(
542 !log.contains("[log gap]"),
543 "resend should not need a gap: {log:?}"
544 );
545 let conflicts = h.state.lock().expect("lock").conflicts;
546 assert!(conflicts >= 1, "the 409 path never ran");
547 shut_down(h).await;
548}