| 254 | 254 | | } |
| 255 | 255 | | |
| 256 | 256 | | /// The key under `jobs:` a job belongs to: a called workflow's jobs |
| 257 | | − | /// (`build/test`) run again with the job that calls it (`build`). |
| 257 | + | /// (`build/test`, or `build[1]/test` for a matrix's second call) run again |
| 258 | + | /// with the job that calls it (`build`). |
| 258 | 259 | | pub fn top_key(key: &str) -> &str { |
| 259 | | − | key.split('/').next().unwrap_or(key) |
| 260 | + | let first = key.split('/').next().unwrap_or(key); |
| 261 | + | first.split('[').next().unwrap_or(first) |
| 262 | + | } |
| 263 | + | |
| 264 | + | /// What the keys of the jobs a caller's call makes start with: the |
| 265 | + | /// caller's key, and for a matrix the combination's index after it |
| 266 | + | /// (`build[1]`), so each combination's call has jobs of its own. A job id |
| 267 | + | /// has no `[`, so the two never meet. |
| 268 | + | fn call_prefix(key: &str, index: Option<usize>) -> String { |
| 269 | + | match index { |
| 270 | + | Some(index) => format!("{key}[{index}]"), |
| 271 | + | None => key.to_owned(), |
| 272 | + | } |
| 273 | + | } |
| 274 | + | |
| 275 | + | /// Whether a run may call one more reusable workflow, `file`, `depth` |
| 276 | + | /// levels below the workflow it started from, having called `called` |
| 277 | + | /// already (one name per call; repeats are one workflow). As on GitHub: ten |
| 278 | + | /// levels of workflows at most, the top one and nine reusable ones below |
| 279 | + | /// it, and 50 different reusable workflows in a run. |
| 280 | + | fn call_allowed(depth: u64, called: &[&str], file: &str) -> std::result::Result<(), String> { |
| 281 | + | if depth > MAX_CALL_DEPTH { |
| 282 | + | return Err(format!( |
| 283 | + | "`{file}` would be the {}th level of workflows: a run connects at most ten, the workflow it started from and nine levels of reusable workflows below it.", |
| 284 | + | depth + 1 |
| 285 | + | )); |
| 286 | + | } |
| 287 | + | let mut unique: Vec<&str> = called.to_vec(); |
| 288 | + | unique.push(file); |
| 289 | + | unique.sort_unstable(); |
| 290 | + | unique.dedup(); |
| 291 | + | if unique.len() > MAX_CALLED_WORKFLOWS { |
| 292 | + | return Err(format!("Calling `{file}` would make {} reusable workflows in this run: a run calls at most {MAX_CALLED_WORKFLOWS} different ones.", unique.len())); |
| 293 | + | } |
| 294 | + | Ok(()) |
| 260 | 295 | | } |
| 261 | 296 | | |
| 297 | + | /// What a job is shown as: its `name` (else its id), under the job whose |
| 298 | + | /// call made it (`caller`), read with its matrix when it has an |
| 299 | + | /// expression, else with the matrix's values after it. |
| 300 | + | fn shown_name(job: &workflow::Job, caller: Option<&str>, combination: &Map<String, Value>, scope: &Scope) -> String { |
| 301 | + | let base = job.name.clone().unwrap_or(job.id.clone()); |
| 302 | + | let base = match caller { |
| 303 | + | Some(caller) => format!("{caller} / {base}"), |
| 304 | + | None => base, |
| 305 | + | }; |
| 306 | + | if expr::has_expression(&base) { |
| 307 | + | expr::interpolate(&base, scope).unwrap_or(base) |
| 308 | + | } else if job.matrix.is_some() { |
| 309 | + | matrix::job_name(&base, combination) |
| 310 | + | } else { |
| 311 | + | base |
| 312 | + | } |
| 313 | + | } |
| 314 | + | |
| 262 | 315 | | /// Debug logging for a job, as a re-run with it turned on gives GitHub's: |
| 263 | 316 | | /// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and |
| 264 | 317 | | /// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of |
| ⋯ |
| 279 | 332 | | self.call.as_deref().and_then(|call| serde_json::from_str(call).ok()) |
| 280 | 333 | | } |
| 281 | 334 | | |
| 335 | + | /// For a job that calls a reusable workflow: what the keys of the jobs |
| 336 | + | /// its call makes start with (`call_prefix`). |
| 337 | + | pub fn call_prefix(&self) -> Option<String> { |
| 338 | + | let call = self.call().filter(|call| call["role"] == "caller")?; |
| 339 | + | Some(call["prefix"].as_str().unwrap_or(&self.key).to_owned()) |
| 340 | + | } |
| 341 | + | |
| 342 | + | /// A job that calls a reusable workflow and has not called it yet: it |
| 343 | + | /// waits for room under its matrix's `max-parallel`, or in its |
| 344 | + | /// concurrency group. |
| 345 | + | pub fn holds_call(&self) -> bool { |
| 346 | + | self.status == "waiting" && self.call_prefix().is_some() |
| 347 | + | } |
| 348 | + | |
| 282 | 349 | | /// For a job of a called workflow: that workflow, the job's own id in |
| 283 | 350 | | /// it, and the job. |
| 284 | 351 | | pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> { |
| ⋯ |
| 293 | 360 | | /// (name in `needs`, key of the jobs). |
| 294 | 361 | | type Unit = (String, workflow::Job, Vec<(String, String)>); |
| 295 | 362 | | |
| 296 | | − | /// How deep reusable workflows may call one another, as on GitHub. |
| 297 | | − | const MAX_CALL_DEPTH: u64 = 4; |
| 363 | + | /// How many levels of reusable workflows may sit below the workflow a run |
| 364 | + | /// started from, as on GitHub (ten levels of workflows in all). |
| 365 | + | const MAX_CALL_DEPTH: u64 = 9; |
| 366 | + | /// How many different reusable workflows a run may call, as on GitHub. |
| 367 | + | const MAX_CALLED_WORKFLOWS: usize = 50; |
| 298 | 368 | | |
| 299 | 369 | | /// What the jobs of one key came to, for `needs.<key>`. |
| 300 | 370 | | fn key_result(rows: &[&JobRow]) -> &'static str { |
| ⋯ |
| 755 | 825 | | let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure"; |
| 756 | 826 | | for (key, job, needs) in &units { |
| 757 | 827 | | let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect(); |
| 758 | | − | if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting") { |
| 828 | + | if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting" && !row.holds_call()) { |
| 759 | 829 | | continue; |
| 760 | 830 | | } |
| 761 | 831 | | let needed: Vec<(&String, Vec<&JobRow>)> = |
| ⋯ |
| 769 | 839 | | } |
| 770 | 840 | | // A job that called a workflow finishes with that workflow's jobs. |
| 771 | 841 | | for row in jobs.iter().filter(|row| row.status == "calling") { |
| 842 | + | let prefix = row.call_prefix().unwrap_or(row.key.clone()); |
| 772 | 843 | | let children: Vec<&JobRow> = jobs |
| 773 | 844 | | .iter() |
| 774 | | − | .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(row.key.as_str()))) |
| 845 | + | .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(prefix.as_str()))) |
| 775 | 846 | | .collect(); |
| 776 | 847 | | if !children.is_empty() && children.iter().all(|child| child.status == "completed") { |
| 777 | | − | self.finish_call(row, &children).await?; |
| 848 | + | self.finish_call(&run, row, &children).await?; |
| 778 | 849 | | changed = true; |
| 779 | 850 | | } |
| 780 | 851 | | } |
| 852 | + | // Calls that wait start while their matrix's `max-parallel` and |
| 853 | + | // their concurrency group have room. |
| 854 | + | let mut calling: std::collections::HashMap<&str, u32> = std::collections::HashMap::new(); |
| 855 | + | for row in jobs.iter().filter(|row| row.status == "calling") { |
| 856 | + | *calling.entry(row.key.as_str()).or_default() += 1; |
| 857 | + | } |
| 858 | + | for row in jobs.iter().filter(|row| row.holds_call()) { |
| 859 | + | let running = calling.get(row.key.as_str()).copied().unwrap_or(0); |
| 860 | + | if row.max_parallel.is_some_and(|max| running >= max) || self.group_busy(row).await? { |
| 861 | + | continue; |
| 862 | + | } |
| 863 | + | if self.start_call(row).await? { |
| 864 | + | calling.insert(row.key.as_str(), running + 1); |
| 865 | + | changed = true; |
| 866 | + | } |
| 867 | + | } |
| 781 | 868 | | if !changed { |
| 782 | 869 | | break; |
| 783 | 870 | | } |
| ⋯ |
| 794 | 881 | | let mut contexts = Self::base_contexts(run, &vars, &job.id); |
| 795 | 882 | | // A called workflow's jobs read the inputs they were called with. |
| 796 | 883 | | let call = row.call(); |
| 797 | | − | let parent = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)); |
| 884 | + | // Shown under the job whose call made it, by that job's name. |
| 885 | + | let caller = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| { |
| 886 | + | c["title"].as_str().map(str::to_owned).or_else(|| c["parent"].as_str().map(|parent| parent.replace('/', " / "))) |
| 887 | + | }); |
| 798 | 888 | | if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") { |
| 799 | 889 | | contexts.insert("inputs".into(), call["inputs"].clone()); |
| 800 | 890 | | } |
| ⋯ |
| 825 | 915 | | Ok(true) => {} |
| 826 | 916 | | Ok(false) => return self.skip_job(row, None).await, |
| 827 | 917 | | Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await, |
| 828 | | − | } |
| 829 | | − | if let Some(uses) = &job.uses { |
| 830 | | − | return self.call_workflow(run, job, row, uses, &scope).await; |
| 831 | 918 | | } |
| 832 | | − | |
| 833 | 919 | | // Its matrix, which may come from a needed job's outputs. |
| 834 | 920 | | let combinations = match &job.matrix { |
| 835 | 921 | | None => vec![Map::new()], |
| ⋯ |
| 845 | 931 | | } |
| 846 | 932 | | } |
| 847 | 933 | | }; |
| 934 | + | // A job that calls a reusable workflow calls it once for each. |
| 935 | + | if let Some(uses) = &job.uses { |
| 936 | + | return self.call_workflow(run, job, row, uses, &contexts, &combinations, caller.as_deref()).await; |
| 937 | + | } |
| 848 | 938 | | let total = combinations.len(); |
| 849 | 939 | | let raw = job.raw.as_object().cloned().unwrap_or_default(); |
| 850 | 940 | | // Whether pull requests from forks may use self-hosted runners here, |
| ⋯ |
| 865 | 955 | | status: Status::Success, |
| 866 | 956 | | hash_files: None, |
| 867 | 957 | | }; |
| 868 | | − | let base_name = job.name.clone().unwrap_or(job.id.clone()); |
| 869 | | − | // A called workflow's job is shown under the job that called it. |
| 870 | | − | let base_name = match &parent { |
| 871 | | − | Some(parent) => format!("{} / {base_name}", parent.replace('/', " / ")), |
| 872 | | − | None => base_name, |
| 873 | | − | }; |
| 874 | | − | let name = if expr::has_expression(&base_name) { |
| 875 | | − | expr::interpolate(&base_name, &scope).unwrap_or(base_name) |
| 876 | | − | } else if job.matrix.is_some() { |
| 877 | | − | matrix::job_name(&base_name, combination) |
| 878 | | − | } else { |
| 879 | | − | base_name |
| 880 | | − | }; |
| 958 | + | let name = shown_name(job, caller.as_deref(), combination, &scope); |
| 881 | 959 | | let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null); |
| 882 | 960 | | let labels = match &runs_on { |
| 883 | 961 | | Value::String(label) => vec![label.clone()], |
| ⋯ |
| 1066 | 1144 | | for (group, cancel) in groups { |
| 1067 | 1145 | | let others = self |
| 1068 | 1146 | | .db |
| 1069 | | − | .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('queued', 'pending', 'in_progress')") |
| 1147 | + | // A call that waits in the group, or calls, is in it too. |
| 1148 | + | .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('waiting', 'queued', 'pending', 'in_progress', 'calling')") |
| 1070 | 1149 | | .bind(&[run.repo_id.as_str().into(), group.as_str().into()])? |
| 1071 | 1150 | | .all() |
| 1072 | 1151 | | .await? |
| 1073 | 1152 | | .results::<JobRow>()?; |
| 1074 | 1153 | | for other in others.iter().filter(|other| !mine.contains(&other.id)) { |
| 1075 | | − | if *cancel || other.status == "queued" { |
| 1154 | + | if *cancel || matches!(other.status.as_str(), "queued" | "waiting") { |
| 1076 | 1155 | | self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?; |
| 1077 | 1156 | | if !moved.contains(&other.run_id) { |
| 1078 | 1157 | | moved.push(other.run_id.clone()); |
| ⋯ |
| 1087 | 1166 | | } |
| 1088 | 1167 | | |
| 1089 | 1168 | | /// A job that calls a reusable workflow, in the repository or another |
| 1090 | | − | /// (reach.rs): that workflow's jobs join the run under it, with the |
| 1091 | | − | /// inputs and secrets it passes. |
| 1092 | | − | async fn call_workflow(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, uses: &str, scope: &Scope<'_>) -> Result<()> { |
| 1169 | + | /// (reach.rs): once for each combination of its matrix, as on GitHub, |
| 1170 | + | /// each with the inputs and secrets it passes read with that |
| 1171 | + | /// combination's `matrix`. Each call waits (`holds_call`) until its |
| 1172 | + | /// matrix's `max-parallel` and its concurrency group have room, when |
| 1173 | + | /// `start_call` puts the workflow's jobs in the run under it. |
| 1174 | + | #[allow(clippy::too_many_arguments)] |
| 1175 | + | async fn call_workflow( |
| 1176 | + | &self, |
| 1177 | + | run: &RunRow, |
| 1178 | + | job: &workflow::Job, |
| 1179 | + | row: &JobRow, |
| 1180 | + | uses: &str, |
| 1181 | + | contexts: &Map<String, Value>, |
| 1182 | + | combinations: &[Map<String, Value>], |
| 1183 | + | caller: Option<&str>, |
| 1184 | + | ) -> Result<()> { |
| 1093 | 1185 | | let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1; |
| 1094 | 1186 | | if depth > MAX_CALL_DEPTH { |
| 1095 | | − | return self.fail_job(row, &format!("Reusable workflows call each other more than {MAX_CALL_DEPTH} deep.")).await; |
| 1187 | + | return self.fail_job(row, &call_allowed(depth, &[], uses).unwrap_err()).await; |
| 1096 | 1188 | | } |
| 1097 | 1189 | | let (file, source, origin) = match self.called_workflow(run, row, uses).await? { |
| 1098 | 1190 | | Ok(found) => found, |
| 1099 | 1191 | | Err(why) => return self.fail_job(row, &why).await, |
| 1100 | 1192 | | }; |
| 1193 | + | let jobs = self.job_rows(&run.id).await?; |
| 1194 | + | // Each call so far, by the workflow it called. |
| 1195 | + | let called_paths: Vec<String> = jobs |
| 1196 | + | .iter() |
| 1197 | + | .filter(|other| other.call_prefix().is_some()) |
| 1198 | + | .filter_map(JobRow::call) |
| 1199 | + | .filter_map(|call| call["path"].as_str().map(str::to_owned)) |
| 1200 | + | .collect(); |
| 1201 | + | let called_already: Vec<&str> = called_paths.iter().map(String::as_str).collect(); |
| 1202 | + | if let Err(why) = call_allowed(depth, &called_already, &file) { |
| 1203 | + | return self.fail_job(row, &why).await; |
| 1204 | + | } |
| 1101 | 1205 | | let called = match workflow::parse(&source) { |
| 1102 | 1206 | | Ok(called) => called, |
| 1103 | 1207 | | Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await, |
| 1104 | 1208 | | }; |
| 1105 | 1209 | | let Some(trigger) = called.trigger("workflow_call") else { |
| 1106 | 1210 | | return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await; |
| 1107 | | − | }; |
| 1108 | | − | // Inputs: what the caller passes, else the called workflow's defaults. |
| 1109 | | − | let given = match job.raw.get("with") { |
| 1110 | | − | Some(with) => match expr::interpolate_value(with, scope) { |
| 1111 | | − | Ok(Value::Object(given)) => given, |
| 1112 | | − | Ok(_) => Map::new(), |
| 1113 | | − | Err(problem) => return self.fail_job(row, &format!("Its `with` does not read: {problem}")).await, |
| 1114 | | − | }, |
| 1115 | | − | None => Map::new(), |
| 1116 | 1211 | | }; |
| 1117 | | − | let mut inputs = Map::new(); |
| 1118 | | − | for (name, spec) in &trigger.inputs { |
| 1119 | | − | let value = given.get(name).cloned().or_else(|| spec.get("default").cloned()).unwrap_or(Value::Null); |
| 1120 | | − | if value.is_null() && spec.get("required").and_then(Value::as_bool) == Some(true) { |
| 1121 | | − | return self.fail_job(row, &format!("`{file}` needs the input `{name}`.")).await; |
| 1122 | | − | } |
| 1123 | | − | inputs.insert(name.clone(), value); |
| 1124 | | − | } |
| 1125 | | − | for (name, value) in given { |
| 1126 | | − | inputs.entry(name).or_insert(value); |
| 1127 | | − | } |
| 1128 | 1212 | | // Secrets: none but the job's token unless `secrets:` passes them, |
| 1129 | 1213 | | // by name or with `inherit`; read when each job starts (job_spec). |
| 1130 | 1214 | | let outer = row.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned()); |
| 1131 | | − | let secrets = crate::reach::secrets_plan(&job.raw, scope.contexts, outer); |
| 1132 | | − | if let Some(name) = crate::reach::missing_secrets(&called.raw, &secrets).first() { |
| 1133 | | − | return self.fail_job(row, &format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`.")).await; |
| 1215 | + | let total = combinations.len(); |
| 1216 | + | let mut groups: Vec<(String, bool)> = Vec::new(); |
| 1217 | + | let mut statements = Vec::new(); |
| 1218 | + | for (index, combination) in combinations.iter().enumerate() { |
| 1219 | + | let mut contexts = contexts.clone(); |
| 1220 | + | contexts.insert("matrix".into(), Value::Object(combination.clone())); |
| 1221 | + | contexts.insert( |
| 1222 | + | "strategy".into(), |
| 1223 | + | json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }), |
| 1224 | + | ); |
| 1225 | + | let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None }; |
| 1226 | + | let name = shown_name(job, caller, combination, &scope); |
| 1227 | + | // Inputs: what `with:` passes, as the called workflow types them. |
| 1228 | + | let given = match job.raw.get("with") { |
| 1229 | + | Some(with) => match expr::interpolate_value(with, &scope) { |
| 1230 | + | Ok(Value::Object(given)) => Ok(given), |
| 1231 | + | Ok(_) => Ok(Map::new()), |
| 1232 | + | Err(problem) => Err(format!("Its `with` does not read: {problem}")), |
| 1233 | + | }, |
| 1234 | + | None => Ok(Map::new()), |
| 1235 | + | }; |
| 1236 | + | let inputs = given.and_then(|given| crate::reach::call_inputs(&file, &trigger.inputs, &given)); |
| 1237 | + | let secrets = crate::reach::secrets_plan(&job.raw, &contexts, outer.clone()); |
| 1238 | + | let problem = match &inputs { |
| 1239 | + | Err(why) => Some(why.clone()), |
| 1240 | + | Ok(_) => crate::reach::missing_secrets(&called.raw, &secrets) |
| 1241 | + | .first() |
| 1242 | + | .map(|name| format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`.")) |
| 1243 | + | .or_else(|| { |
| 1244 | + | crate::reach::undeclared_secrets(&called.raw, &secrets) |
| 1245 | + | .first() |
| 1246 | + | .map(|name| format!("Invalid secret: `{name}` is not defined in `{file}`.")) |
| 1247 | + | }), |
| 1248 | + | }; |
| 1249 | + | // Its own concurrency group, which it holds while its workflow runs. |
| 1250 | + | let group = job |
| 1251 | + | .concurrency |
| 1252 | + | .as_ref() |
| 1253 | + | .and_then(|c| expr::interpolate(&c.group, &scope).ok()) |
| 1254 | + | .map(|group| group.trim().to_owned()) |
| 1255 | + | .filter(|group| !group.is_empty()); |
| 1256 | + | let cancel = job |
| 1257 | + | .concurrency |
| 1258 | + | .as_ref() |
| 1259 | + | .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok()) |
| 1260 | + | .is_some_and(|value| expr::truthy(&value)); |
| 1261 | + | if problem.is_none() |
| 1262 | + | && let Some(group) = &group |
| 1263 | + | && !groups.iter().any(|(known, _)| known == group) |
| 1264 | + | { |
| 1265 | + | groups.push((group.clone(), cancel)); |
| 1266 | + | } |
| 1267 | + | let call = json!({ |
| 1268 | + | "role": "caller", "path": file, "source": source, "origin": origin, "depth": depth, |
| 1269 | + | "prefix": call_prefix(&row.key, job.matrix.is_some().then_some(index)), "title": name, |
| 1270 | + | "inputs": inputs.unwrap_or_default(), "secrets": secrets, "fail_fast": job.fail_fast, |
| 1271 | + | }); |
| 1272 | + | let (status, conclusion, finished) = match &problem { |
| 1273 | + | Some(_) => ("completed", Some("failure"), Some(now())), |
| 1274 | + | None => ("waiting", None, None), |
| 1275 | + | }; |
| 1276 | + | let values: Vec<worker::wasm_bindgen::JsValue> = vec![ |
| 1277 | + | name.as_str().into(), |
| 1278 | + | serde_json::to_string(combination)?.into(), |
| 1279 | + | status.into(), |
| 1280 | + | optional(conclusion), |
| 1281 | + | optional(problem.as_deref()), |
| 1282 | + | job.max_parallel.map(|max| max.max(1)).map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), |
| 1283 | + | optional(finished.as_deref()), |
| 1284 | + | optional(group.as_deref()), |
| 1285 | + | u32::from(cancel).into(), |
| 1286 | + | serde_json::to_string(&call)?.into(), |
| 1287 | + | ]; |
| 1288 | + | if index == 0 { |
| 1289 | + | let mut bound = values; |
| 1290 | + | bound.push(row.id.as_str().into()); |
| 1291 | + | statements.push( |
| 1292 | + | self.db |
| 1293 | + | .prepare( |
| 1294 | + | "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, max_parallel = ?, finished_at = ?, |
| 1295 | + | concurrency_group = ?, cancel_in_progress = ?, call = ? WHERE id = ?", |
| 1296 | + | ) |
| 1297 | + | .bind(&bound)?, |
| 1298 | + | ); |
| 1299 | + | } else { |
| 1300 | + | let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![ |
| 1301 | + | new_id("job", now_ms()).into(), |
| 1302 | + | row.run_id.as_str().into(), |
| 1303 | + | row.repo_id.as_str().into(), |
| 1304 | + | row.namespace.as_str().into(), |
| 1305 | + | row.key.as_str().into(), |
| 1306 | + | (index as u32).into(), |
| 1307 | + | row.needs.as_str().into(), |
| 1308 | + | ]; |
| 1309 | + | bound.extend(values); |
| 1310 | + | statements.push( |
| 1311 | + | self.db |
| 1312 | + | .prepare( |
| 1313 | + | "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, name, matrix, status, conclusion, reason, |
| 1314 | + | max_parallel, finished_at, concurrency_group, cancel_in_progress, call) |
| 1315 | + | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 1316 | + | ) |
| 1317 | + | .bind(&bound)?, |
| 1318 | + | ); |
| 1319 | + | } |
| 1320 | + | } |
| 1321 | + | self.db.batch(statements).await?; |
| 1322 | + | self.replace_in_groups(run, &row.key, &groups).await |
| 1323 | + | } |
| 1324 | + | |
| 1325 | + | /// Whether another job of `row`'s concurrency group runs, or calls a |
| 1326 | + | /// workflow, so that `row` waits for it. |
| 1327 | + | async fn group_busy(&self, row: &JobRow) -> Result<bool> { |
| 1328 | + | let Some(group) = &row.concurrency_group else { return Ok(false) }; |
| 1329 | + | let busy = self |
| 1330 | + | .db |
| 1331 | + | .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?") |
| 1332 | + | .bind(&[row.repo_id.as_str().into(), group.as_str().into(), row.id.as_str().into()])? |
| 1333 | + | .first::<Count>(None) |
| 1334 | + | .await? |
| 1335 | + | .map_or(0, |count| count.n); |
| 1336 | + | Ok(busy > 0) |
| 1337 | + | } |
| 1338 | + | |
| 1339 | + | /// A call that waited (`holds_call`) is made: the called workflow's |
| 1340 | + | /// jobs join the run under it, named under its name and keyed under its |
| 1341 | + | /// prefix. False when it no longer waits. |
| 1342 | + | async fn start_call(&self, row: &JobRow) -> Result<bool> { |
| 1343 | + | let Some(call) = row.call() else { return Ok(false) }; |
| 1344 | + | let Some(prefix) = row.call_prefix() else { return Ok(false) }; |
| 1345 | + | let Some(called) = call["source"].as_str().and_then(|source| workflow::parse(source).ok()) else { return Ok(false) }; |
| 1346 | + | let file = call["path"].as_str().unwrap_or_default(); |
| 1347 | + | let started = self |
| 1348 | + | .db |
| 1349 | + | .prepare("UPDATE jobs SET status = 'calling', reason = ?, started_at = ? WHERE id = ? AND status = 'waiting' RETURNING id") |
| 1350 | + | .bind(&[format!("Calls `{file}`.").into(), now().into(), row.id.as_str().into()])? |
| 1351 | + | .first::<Value>(None) |
| 1352 | + | .await?; |
| 1353 | + | if started.is_none() { |
| 1354 | + | return Ok(false); |
| 1134 | 1355 | | } |
| 1135 | 1356 | | let mut statements = Vec::new(); |
| 1136 | 1357 | | for called_job in &called.jobs { |
| 1137 | | − | let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{}/{n}", row.key)).collect(); |
| 1138 | | − | let call = json!({ |
| 1139 | | − | "role": "callee", "parent": row.key, "job": called_job.id, "path": file, |
| 1140 | | − | "source": source, "inputs": inputs, "depth": depth, "origin": origin, "secrets": secrets, |
| 1358 | + | let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{prefix}/{n}")).collect(); |
| 1359 | + | let callee = json!({ |
| 1360 | + | "role": "callee", "parent": prefix, "title": row.name, "job": called_job.id, "path": file, |
| 1361 | + | "source": call["source"], "inputs": call["inputs"], "depth": call["depth"], "origin": call["origin"], "secrets": call["secrets"], |
| 1141 | 1362 | | }); |
| 1142 | 1363 | | statements.push( |
| 1143 | 1364 | | self.db |
| ⋯ |
| 1147 | 1368 | | row.run_id.as_str().into(), |
| 1148 | 1369 | | row.repo_id.as_str().into(), |
| 1149 | 1370 | | row.namespace.as_str().into(), |
| 1150 | | − | format!("{}/{}", row.key, called_job.id).into(), |
| 1371 | + | format!("{prefix}/{}", called_job.id).into(), |
| 1151 | 1372 | | format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(), |
| 1152 | 1373 | | serde_json::to_string(&needs)?.into(), |
| 1153 | | − | serde_json::to_string(&call)?.into(), |
| 1374 | + | serde_json::to_string(&callee)?.into(), |
| 1154 | 1375 | | ])?, |
| 1155 | 1376 | | ); |
| 1156 | 1377 | | } |
| 1157 | | − | statements.push( |
| 1158 | | − | self.db |
| 1159 | | − | .prepare("UPDATE jobs SET status = 'calling', call = ?, reason = ?, started_at = ? WHERE id = ?") |
| 1160 | | − | .bind(&[ |
| 1161 | | − | serde_json::to_string(&json!({ "role": "caller", "path": file, "source": source }))?.into(), |
| 1162 | | − | format!("Calls `{file}`.").into(), |
| 1163 | | − | now().into(), |
| 1164 | | − | row.id.as_str().into(), |
| 1165 | | − | ])?, |
| 1166 | | − | ); |
| 1167 | | − | self.db.batch(statements).await?; |
| 1378 | + | if !statements.is_empty() { |
| 1379 | + | self.db.batch(statements).await?; |
| 1380 | + | } |
| 1381 | + | Ok(true) |
| 1382 | + | } |
| 1383 | + | |
| 1384 | + | /// The runs with a call waiting in `group`, but `run_id`, moved along |
| 1385 | + | /// now that a job of the group has finished. |
| 1386 | + | async fn wake_group(&self, repo_id: &str, group: &str, run_id: &str) -> Result<()> { |
| 1387 | + | #[derive(Deserialize)] |
| 1388 | + | struct RunId { |
| 1389 | + | run_id: String, |
| 1390 | + | } |
| 1391 | + | let waiting = self |
| 1392 | + | .db |
| 1393 | + | .prepare("SELECT DISTINCT run_id FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'waiting' AND run_id != ?") |
| 1394 | + | .bind(&[repo_id.into(), group.into(), run_id.into()])? |
| 1395 | + | .all() |
| 1396 | + | .await? |
| 1397 | + | .results::<RunId>()?; |
| 1398 | + | for other in waiting { |
| 1399 | + | Box::pin(self.advance(&other.run_id)).await?; |
| 1400 | + | } |
| 1168 | 1401 | | Ok(()) |
| 1169 | 1402 | | } |
| 1170 | 1403 | | |
| 1171 | 1404 | | /// A job that called a workflow, finished with its jobs: their result, |
| 1172 | | − | /// and the outputs the workflow declares. |
| 1173 | | − | async fn finish_call(&self, row: &JobRow, children: &[&JobRow]) -> Result<()> { |
| 1405 | + | /// and the outputs the workflow declares. A failed call of a fail-fast |
| 1406 | + | /// matrix stops the matrix's other calls. |
| 1407 | + | async fn finish_call(&self, run: &RunRow, row: &JobRow, children: &[&JobRow]) -> Result<()> { |
| 1174 | 1408 | | let call = row.call().unwrap_or_default(); |
| 1175 | 1409 | | let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok()); |
| 1176 | 1410 | | let mut jobs_context = Map::new(); |
| ⋯ |
| 1206 | 1440 | | } |
| 1207 | 1441 | | } |
| 1208 | 1442 | | let conclusion = key_result(children); |
| 1209 | | − | self.db |
| 1210 | | − | .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling'") |
| 1443 | + | let finished = self |
| 1444 | + | .db |
| 1445 | + | .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling' RETURNING id") |
| 1211 | 1446 | | .bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])? |
| 1212 | | − | .run() |
| 1447 | + | .first::<Value>(None) |
| 1213 | 1448 | | .await?; |
| 1449 | + | if finished.is_none() { |
| 1450 | + | return Ok(()); |
| 1451 | + | } |
| 1452 | + | if conclusion == "failure" && row.matrix.as_deref().is_some_and(|m| m != "{}") && call["fail_fast"] != json!(false) { |
| 1453 | + | let siblings = self |
| 1454 | + | .db |
| 1455 | + | .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'") |
| 1456 | + | .bind(&[row.run_id.as_str().into(), row.key.as_str().into()])? |
| 1457 | + | .all() |
| 1458 | + | .await? |
| 1459 | + | .results::<JobRow>()?; |
| 1460 | + | for sibling in siblings { |
| 1461 | + | self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?; |
| 1462 | + | } |
| 1463 | + | } |
| 1464 | + | if let Some(group) = &row.concurrency_group { |
| 1465 | + | self.wake_group(&run.repo_id, group, &run.id).await?; |
| 1466 | + | } |
| 1214 | 1467 | | Ok(()) |
| 1215 | 1468 | | } |
| 1216 | 1469 | | |
| ⋯ |
| 1296 | 1549 | | if let Some(group) = &job.concurrency_group { |
| 1297 | 1550 | | let running = self |
| 1298 | 1551 | | .db |
| 1299 | | − | .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'in_progress' AND id != ?") |
| 1552 | + | .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?") |
| 1300 | 1553 | | .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])? |
| 1301 | 1554 | | .first::<Count>(None) |
| 1302 | 1555 | | .await? |
| ⋯ |
| 1420 | 1673 | | // fail-fast: one failed combination stops the rest of its matrix. |
| 1421 | 1674 | | if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") { |
| 1422 | 1675 | | let run = self.run_row(&job.run_id).await?; |
| 1423 | | − | let fail_fast = run |
| 1424 | | − | .as_ref() |
| 1425 | | − | .and_then(|run| workflow::parse(&run.source).ok()) |
| 1426 | | − | .and_then(|workflow| workflow.jobs.into_iter().find(|j| j.id == job.key)) |
| 1427 | | − | .is_none_or(|j| j.fail_fast); |
| 1676 | + | // As its workflow, or the workflow it is a job of, says. |
| 1677 | + | let fail_fast = run.as_ref().and_then(|run| job_spec(run, &job)).is_none_or(|j| j.fail_fast); |
| 1428 | 1678 | | if fail_fast { |
| 1429 | 1679 | | let siblings = self |
| 1430 | 1680 | | .db |
| ⋯ |
| 1438 | 1688 | | } |
| 1439 | 1689 | | } |
| 1440 | 1690 | | } |
| 1691 | + | // A call waiting in its concurrency group, in another run, may start. |
| 1692 | + | if let Some(group) = &job.concurrency_group { |
| 1693 | + | self.wake_group(&job.repo_id, group, &job.run_id).await?; |
| 1694 | + | } |
| 1441 | 1695 | | Box::pin(self.advance(&job.run_id)).await |
| 1442 | 1696 | | } |
| 1443 | 1697 | | |
| ⋯ |
| 1446 | 1700 | | /// steps and its post steps before it ends `cancelled`; one that has |
| 1447 | 1701 | | /// not started is cancelled at once. |
| 1448 | 1702 | | async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> { |
| 1703 | + | // A job calling a workflow stops with that workflow's jobs. |
| 1704 | + | if job.status == "calling" |
| 1705 | + | && let Some(prefix) = job.call_prefix() |
| 1706 | + | { |
| 1707 | + | let under = format!("{prefix}/"); |
| 1708 | + | for child in self.job_rows(&job.run_id).await?.iter().filter(|child| child.key.starts_with(&under) && child.status != "completed") { |
| 1709 | + | Box::pin(self.stop_job(child, reason)).await?; |
| 1710 | + | } |
| 1711 | + | } |
| 1449 | 1712 | | if job.status == "in_progress" && job.started_at.is_some() { |
| 1450 | 1713 | | self.db |
| 1451 | 1714 | | .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'") |
| ⋯ |
| 1852 | 2115 | | ]; |
| 1853 | 2116 | | for key in &again { |
| 1854 | 2117 | | statements.push(self.db.prepare("DELETE FROM jobs WHERE run_id = ? AND key = ? AND ordinal > 0").bind(&[run.id.as_str().into(), key.as_str().into()])?); |
| 1855 | | − | // The jobs of a workflow it called are made again when it calls it again. |
| 2118 | + | // The jobs of a workflow it called (each matrix combination's |
| 2119 | + | // under `key[i]/`) are made again when it calls it again. |
| 1856 | 2120 | | statements.push( |
| 1857 | 2121 | | self.db |
| 1858 | | − | .prepare("DELETE FROM jobs WHERE run_id = ? AND key LIKE ?") |
| 1859 | | − | .bind(&[run.id.as_str().into(), format!("{key}/%").into()])?, |
| 2122 | + | .prepare("DELETE FROM jobs WHERE run_id = ? AND (key LIKE ? OR key LIKE ?)") |
| 2123 | + | .bind(&[run.id.as_str().into(), format!("{key}/%").into(), format!("{key}[%").into()])?, |
| 1860 | 2124 | | ); |
| 1861 | 2125 | | statements.push( |
| 1862 | 2126 | | self.db |
| ⋯ |
| 1948 | 2212 | | let (default, pull_requests) = self.token_policy(&run.repo_id).await?; |
| 1949 | 2213 | | let mut permissions = spec.permissions(&workflow, default); |
| 1950 | 2214 | | if let Some(parent) = &job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)) { |
| 1951 | | − | let top = parent.split('/').next().unwrap_or(parent); |
| 2215 | + | let top = top_key(parent); |
| 1952 | 2216 | | if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) { |
| 1953 | 2217 | | permissions = permissions.capped_by(&caller_job.permissions(&caller, default)); |
| 1954 | 2218 | | } |
| ⋯ |
| 2327 | 2591 | | self.finish_job(&job.id, "failure", Some(&reason), None).await?; |
| 2328 | 2592 | | } |
| 2329 | 2593 | | } |
| 2594 | + | // Calls of reusable workflows waiting in a concurrency group, in |
| 2595 | + | // case the job they waited for was cancelled in another run. |
| 2596 | + | #[derive(Deserialize)] |
| 2597 | + | struct RunId { |
| 2598 | + | run_id: String, |
| 2599 | + | } |
| 2600 | + | let held = self |
| 2601 | + | .db |
| 2602 | + | .prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20") |
| 2603 | + | .all() |
| 2604 | + | .await? |
| 2605 | + | .results::<RunId>()?; |
| 2606 | + | for run in held { |
| 2607 | + | self.advance(&run.run_id).await?; |
| 2608 | + | } |
| 2330 | 2609 | | if let Err(error) = self.sweep_runners(now_ms).await { |
| 2331 | 2610 | | worker::console_error!("actions: the runners' sweep failed: {error}"); |
| 2332 | 2611 | | } |
| ⋯ |
| 2479 | 2758 | | mod reruns { |
| 2480 | 2759 | | use serde_json::{Map, Value, json}; |
| 2481 | 2760 | | |
| 2482 | | − | use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, debug_logging, rerun_keys, summary_fits, top_key}; |
| 2761 | + | use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, Scope, Status, call_allowed, call_prefix, debug_logging, rerun_keys, shown_name, summary_fits, top_key}; |
| 2483 | 2762 | | |
| 2484 | 2763 | | /// build ← test ← deploy, and lint on its own. |
| 2485 | 2764 | | fn keys(which: Rerun, failed: &[&str]) -> Vec<String> { |
| ⋯ |
| 2512 | 2791 | | assert_eq!(top_key("build/test"), "build"); |
| 2513 | 2792 | | assert_eq!(top_key("build/inner/test"), "build"); |
| 2514 | 2793 | | assert_eq!(top_key("lint"), "lint"); |
| 2794 | + | // A matrix's calls each have their own jobs, under `key[i]/`. |
| 2795 | + | assert_eq!(top_key("build[1]/test"), "build"); |
| 2796 | + | assert_eq!(top_key("build/inner[0]/test"), "build"); |
| 2797 | + | assert_eq!(call_prefix("build", Some(1)), "build[1]"); |
| 2798 | + | assert_eq!(call_prefix("build/inner", None), "build/inner"); |
| 2799 | + | } |
| 2800 | + | |
| 2801 | + | #[test] |
| 2802 | + | fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() { |
| 2803 | + | assert!(call_allowed(1, &[], "a.yml").is_ok()); |
| 2804 | + | assert!(call_allowed(9, &[], "a.yml").is_ok()); |
| 2805 | + | let deep = call_allowed(10, &[], "a.yml").unwrap_err(); |
| 2806 | + | assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}"); |
| 2807 | + | let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect(); |
| 2808 | + | let fifty: Vec<&str> = names.iter().map(String::as_str).collect(); |
| 2809 | + | // A workflow called again, as by each combination of a matrix, is one. |
| 2810 | + | assert!(call_allowed(1, &fifty, "w3.yml").is_ok()); |
| 2811 | + | assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different")); |
| 2812 | + | } |
| 2813 | + | |
| 2814 | + | #[test] |
| 2815 | + | fn a_matrixs_calls_are_named_by_the_caller_and_its_values() { |
| 2816 | + | let workflow = super::workflow::parse( |
| 2817 | + | "on: push\njobs:\n build:\n name: Build\n strategy:\n matrix: { os: [linux], node: [20] }\n uses: ./.g1t/workflows/b.yml\n plain:\n uses: ./.g1t/workflows/b.yml\n test:\n name: Test ${{ matrix.os }}\n strategy:\n matrix: { os: [linux] }\n runs-on: ubuntu-latest\n steps: [{ run: 'true' }]", |
| 2818 | + | ) |
| 2819 | + | .unwrap(); |
| 2820 | + | let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap(); |
| 2821 | + | let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap(); |
| 2822 | + | let mut contexts = Map::new(); |
| 2823 | + | contexts.insert("matrix".into(), Value::Object(combination.clone())); |
| 2824 | + | let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None }; |
| 2825 | + | assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)"); |
| 2826 | + | assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain"); |
| 2827 | + | // A called workflow's job, under the call that made it. |
| 2828 | + | assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain"); |
| 2829 | + | assert_eq!(shown_name(job("test"), Some("Build"), &combination, &scope), "Build / Test linux"); |
| 2515 | 2830 | | } |
| 2516 | 2831 | | |
| 2517 | 2832 | | #[test] |