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::{RunEnd, 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 auto_update: true,
325 }
326}
327
328fn fast_options() -> RunOptions {
329 RunOptions {
330 heartbeat_every: Duration::from_millis(50),
331 claim_backoff: Duration::from_millis(50),
332 pump: PumpConfig {
333 flush_every: Duration::from_millis(50),
334 retries: 3,
335 retry_delay: Duration::from_millis(10),
336 },
337 timeout_override: None,
338 }
339}
340
341/// Polls `check` until it returns true or the deadline passes.
342async fn wait_until(what: &str, mut check: impl FnMut() -> bool) {
343 let deadline = Instant::now() + Duration::from_secs(60);
344 while !check() {
345 assert!(Instant::now() < deadline, "timed out waiting for {what}");
346 tokio::time::sleep(Duration::from_millis(20)).await;
347 }
348}
349
350struct Harness {
351 state: Shared,
352 stop: CancellationToken,
353 runner_task: tokio::task::JoinHandle<RunEnd>,
354 _dir: tempfile::TempDir,
355}
356
357/// Starts the mock, a runner with HostExecutor, and queues one job.
358async fn start(
359 run: &str,
360 mut options: RunOptions,
361 setup: impl FnOnce(&mut MockState),
362 timeout: Option<Duration>,
363) -> (Harness, String) {
364 let dir = tempfile::tempdir().expect("tempdir");
365 let commit = make_repo(dir.path());
366 let clone_url = format!("file://{}", dir.path().join("repo").display());
367 let job_id = "job-1";
368 let state: Shared = Arc::new(Mutex::new(MockState::default()));
369 {
370 let mut st = state.lock().expect("lock");
371 let j = job(job_id, &commit, &clone_url, run);
372 st.by_id.insert(j.job_id.clone(), j.clone());
373 st.pending.push(j);
374 setup(&mut st);
375 }
376 let url = start_mock(state.clone()).await;
377 options.timeout_override = timeout;
378 let cfg = config(&url, dir.path().join("work"));
379 let runner = Runner::new(cfg, Arc::new(HostExecutor), options).expect("runner");
380 let stop = CancellationToken::new();
381 let runner_task = {
382 let stop = stop.clone();
383 tokio::spawn(async move { runner.run(stop).await })
384 };
385 (
386 Harness {
387 state,
388 stop,
389 runner_task,
390 _dir: dir,
391 },
392 job_id.to_owned(),
393 )
394}
395
396async fn finish_of(h: &Harness, job_id: &str) -> FinishRequest {
397 let state = h.state.clone();
398 let id = job_id.to_owned();
399 wait_until("the job to finish", || {
400 state
401 .lock()
402 .map(|st| st.finished.contains_key(&id))
403 .unwrap_or(false)
404 })
405 .await;
406 h.state
407 .lock()
408 .expect("lock")
409 .finished
410 .get(job_id)
411 .cloned()
412 .expect("finish recorded")
413}
414
415async fn shut_down(h: Harness) {
416 h.stop.cancel();
417 let _ = tokio::time::timeout(Duration::from_secs(20), h.runner_task).await;
418}
419
420fn log_text(h: &Harness) -> String {
421 String::from_utf8_lossy(&h.state.lock().expect("lock").log).into_owned()
422}
423
424#[tokio::test]
425async fn full_flow_masks_secrets_and_uploads_artifacts() {
426 let run = "mkdir -p dist && echo built > dist/out.txt && echo \"token is $MY_TOKEN\" && \
427 echo \"b64 $(printf %s \"$MY_TOKEN\" | base64)\" && echo \"env is $GREETING\"";
428 let (h, job_id) = start(run, fast_options(), |_| {}, None).await;
429
430 let finish = finish_of(&h, &job_id).await;
431 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
432 assert_eq!(finish.exit_code, Some(0));
433
434 let log = log_text(&h);
435 assert!(log.starts_with("forge-runner "), "info line first: {log:?}");
436 assert!(log.contains("token is ***"), "masked: {log:?}");
437 assert!(log.contains("env is hello-env"), "{log:?}");
438 assert!(!log.contains(SECRET_VALUE), "secret leaked into the log");
439 let b64 = base64_of(SECRET_VALUE);
440 assert!(
441 !log.contains(&b64),
442 "base64 of the secret leaked into the log"
443 );
444
445 let artifact = h
446 .state
447 .lock()
448 .expect("lock")
449 .artifacts
450 .get("dist_out.txt")
451 .cloned();
452 assert_eq!(artifact.as_deref(), Some(&b"built\n"[..]));
453
454 shut_down(h).await;
455}
456
457fn base64_of(text: &str) -> String {
458 use base64::Engine;
459 base64::engine::general_purpose::STANDARD.encode(text)
460}
461
462#[tokio::test]
463async fn heartbeat_cancel_stops_the_job() {
464 let mut options = fast_options();
465 options.heartbeat_every = Duration::from_millis(50);
466 let (h, job_id) = start(
467 "echo start; sleep 60",
468 options,
469 |st| st.cancel_from_heartbeat = Some(2),
470 None,
471 )
472 .await;
473
474 let started = Instant::now();
475 let finish = finish_of(&h, &job_id).await;
476 assert_eq!(finish.outcome, Outcome::Cancelled, "{finish:?}");
477 assert!(
478 started.elapsed() < Duration::from_secs(30),
479 "the job was not stopped"
480 );
481 shut_down(h).await;
482}
483
484#[tokio::test]
485async fn timeout_fails_the_job() {
486 let (h, job_id) = start(
487 "sleep 60",
488 fast_options(),
489 |_| {},
490 Some(Duration::from_millis(400)),
491 )
492 .await;
493
494 let finish = finish_of(&h, &job_id).await;
495 assert_eq!(finish.outcome, Outcome::Failure, "{finish:?}");
496 assert!(
497 finish
498 .message
499 .as_deref()
500 .unwrap_or_default()
501 .starts_with("timed out"),
502 "{finish:?}"
503 );
504 shut_down(h).await;
505}
506
507#[tokio::test]
508async fn log_offset_mismatch_resyncs_without_loss_or_duplicates() {
509 // Log POST #2 is applied by the forge for its first 3 bytes, then answered 409. The runner
510 // must drop those 3 bytes and continue, so the stored log has every line exactly once.
511 let run = "echo first; sleep 0.3; echo second; sleep 0.3; echo third";
512 let (h, job_id) = start(run, fast_options(), |st| st.ghost = Some((2, 3)), None).await;
513
514 let finish = finish_of(&h, &job_id).await;
515 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
516
517 let log = log_text(&h);
518 assert!(log.ends_with("first\nsecond\nthird\n"), "log was {log:?}");
519 assert_eq!(
520 log.matches("second").count(),
521 1,
522 "duplicate or lost bytes: {log:?}"
523 );
524 let conflicts = h.state.lock().expect("lock").conflicts;
525 assert!(conflicts >= 1, "the 409 path never ran");
526 shut_down(h).await;
527}
528
529#[tokio::test]
530async fn acknowledged_batch_the_forge_lost_is_resent_in_order() {
531 // Log POST #2 is acknowledged but never stored. The next POST gets 409 with the smaller size,
532 // and the runner must resend the lost bytes from its retained buffer, with no gap marker.
533 let run = "echo first; sleep 0.3; echo second; sleep 0.3; echo third";
534 let (h, job_id) = start(run, fast_options(), |st| st.drop_post = Some(2), None).await;
535
536 let finish = finish_of(&h, &job_id).await;
537 assert_eq!(finish.outcome, Outcome::Success, "{finish:?}");
538
539 let log = log_text(&h);
540 assert!(log.ends_with("first\nsecond\nthird\n"), "log was {log:?}");
541 assert_eq!(log.matches("second").count(), 1, "duplicate bytes: {log:?}");
542 assert!(
543 !log.contains("[log gap]"),
544 "resend should not need a gap: {log:?}"
545 );
546 let conflicts = h.state.lock().expect("lock").conflicts;
547 assert!(conflicts >= 1, "the 409 path never ran");
548 shut_down(h).await;
549}