flagon-io/g1t

public

Actions: keep every run attempt, re-run one job, graceful cancel, summaries

A re-run is a new attempt that keeps the one before (actions/0008: run_attempts, job_attempts): its jobs as they ended, their logs and summaries moved under {job}.{attempt}. rerun takes job (that job and those that need it; a matrix or called workflow runs again whole) and debug (RUNNER_DEBUG=1, ACTIONS_STEP_DEBUG, ACTIONS_RUNNER_DEBUG, runner.debug). run, logs and summaries read any attempt; job_log_text and run_logs return whole logs to download. Cancelling marks running jobs cancel_requested_at and tells them in the answer to their next report; they end cancelled after their cleanup steps, or are stopped outright after 5 minutes. Cancelling again, or force, stops them at once. Job summaries are kept per step, up to 1 MiB and 20 steps a job.

syntaqxcommitted 1d agoParent6369df1Browse files

138	138		/// The self-hosted runner that took it, by name.
139	139		#[serde(default)]
140	140		pub runner: Option<String>,
	141	+	/// It was cancelled and is running its `if: always()` and `cancelled()`
	142	+	/// steps and its post steps before it ends.
	143	+	#[serde(default)]
	144	+	pub cancelling: bool,
141	145		}
142	146
143	147		#[derive(Clone, Debug, Serialize, Deserialize)]
⋯
154	158		/// The environments whose protection rules hold its jobs, this attempt.
155	159		#[serde(default)]
156	160		pub pending_deployments: Vec<PendingDeployment>,
	161	+	/// Every attempt of the run, oldest first, the one shown included.
	162	+	/// `run.attempt` says which one `jobs` belong to.
	163	+	#[serde(default)]
	164	+	pub attempts: Vec<RunAttempt>,
157	165		}
158	166
	167	+	/// One attempt of a run: the first, or a re-run.
	168	+	#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
	169	+	#[serde(rename_all = "camelCase")]
	170	+	pub struct RunAttempt {
	171	+	/// From 1.
	172	+	pub attempt: u64,
	173	+	/// `queued`, `in_progress` or `completed`; earlier attempts are completed.
	174	+	pub status: String,
	175	+	pub conclusion: Option<String>,
	176	+	/// Who started it: whoever caused the run for the first, whoever re-ran
	177	+	/// it for the rest.
	178	+	pub actor: Option<String>,
	179	+	/// It ran with debug logging (`RUNNER_DEBUG=1`).
	180	+	pub debug: bool,
	181	+	pub started_at: Option<String>,
	182	+	pub finished_at: Option<String>,
	183	+	}
	184	+
	185	+	/// One job's summary: what its steps wrote to `$GITHUB_STEP_SUMMARY`, in
	186	+	/// Markdown, masked.
	187	+	#[derive(Clone, Debug, Serialize, Deserialize)]
	188	+	#[serde(rename_all = "camelCase")]
	189	+	pub struct JobSummary {
	190	+	/// The job's id, as `RunDetail.jobs` gives it for the attempt.
	191	+	pub job_id: String,
	192	+	pub name: String,
	193	+	pub steps: Vec<StepSummary>,
	194	+	}
	195	+
	196	+	#[derive(Clone, Debug, Serialize, Deserialize)]
	197	+	#[serde(rename_all = "camelCase")]
	198	+	pub struct StepSummary {
	199	+	/// The step, from 1 (post steps follow the job's own).
	200	+	pub step: u32,
	201	+	pub markdown: String,
	202	+	}
	203	+
	204	+	/// A job's whole log, for downloading: its steps, to split the text by.
	205	+	#[derive(Clone, Debug, Serialize, Deserialize)]
	206	+	#[serde(rename_all = "camelCase")]
	207	+	pub struct JobLogText {
	208	+	pub job_id: String,
	209	+	pub name: String,
	210	+	pub steps: Vec<StepState>,
	211	+	pub chunks: Vec<LogChunk>,
	212	+	/// Whether the job has finished.
	213	+	pub done: bool,
	214	+	/// Its log was left out: the run's logs reached `MAX_RUN_LOG_BYTES`.
	215	+	#[serde(default)]
	216	+	pub omitted: bool,
	217	+	}
	218	+
159	219		/// A run that needed approval before it started.
160	220		#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
161	221		#[serde(rename_all = "camelCase")]
⋯
518	578		pub repo: RepoPath,
519	579		pub viewer: Viewer,
520	580		pub id: String,
	581	+	/// An earlier attempt; the latest when absent.
	582	+	#[serde(default)]
	583	+	pub attempt: Option<u64>,
	584	+	}
	585	+
	586	+	/// `summaries`: the job summaries of a run's attempt (the latest when
	587	+	/// `attempt` is absent), jobs in the run's order, those with none left
	588	+	/// out. Returns `Outcome<Vec<JobSummary>>`.
	589	+	#[derive(Debug, Serialize, Deserialize)]
	590	+	pub struct SummariesArgs {
	591	+	pub repo: RepoPath,
	592	+	pub viewer: Viewer,
	593	+	pub id: String,
	594	+	#[serde(default)]
	595	+	pub attempt: Option<u64>,
	596	+	}
	597	+
	598	+	/// `job_log_text`: one job's whole log, any attempt's (by the id the run
	599	+	/// gave the job). Returns `Outcome<JobLogText>`.
	600	+	#[derive(Debug, Serialize, Deserialize)]
	601	+	pub struct JobLogTextArgs {
	602	+	pub repo: RepoPath,
	603	+	pub viewer: Viewer,
	604	+	pub job: String,
	605	+	}
	606	+
	607	+	/// `run_logs`: every job's whole log for an attempt of a run (the latest
	608	+	/// when `attempt` is absent), until they add up to `MAX_RUN_LOG_BYTES`;
	609	+	/// jobs past it come `omitted`, with no chunks. Returns
	610	+	/// `Outcome<Vec<JobLogText>>`.
	611	+	#[derive(Debug, Serialize, Deserialize)]
	612	+	pub struct RunLogsArgs {
	613	+	pub repo: RepoPath,
	614	+	pub viewer: Viewer,
	615	+	pub id: String,
	616	+	#[serde(default)]
	617	+	pub attempt: Option<u64>,
521	618		}
522	619
	620	+	/// The most log `run_logs` returns at once, in bytes.
	621	+	pub const MAX_RUN_LOG_BYTES: usize = 24 * 1024 * 1024;
	622	+
523	623		/// `logs`: a job's log after `after`. Returns `Outcome<JobLog>`.
524	624		#[derive(Debug, Serialize, Deserialize)]
525	625		pub struct LogsArgs {
⋯
545	645		pub inputs: serde_json::Map<String, Value>,
546	646		}
547	647
548		−	/// `cancel` and `rerun` (all jobs, or with `failed_only` the ones that did
549		−	/// not succeed). Members only. Returns `Outcome<WorkflowRun>`.
	648	+	/// `cancel` and `rerun`: every job, with `failed_only` the ones that did
	649	+	/// not succeed, or with `job` that one job (by its id in the run's latest
	650	+	/// attempt); each with the jobs that need them. `debug` runs the new
	651	+	/// attempt with debug logging. Members only. Returns `Outcome<WorkflowRun>`.
550	652		#[derive(Debug, Serialize, Deserialize)]
551	653		pub struct RunActionArgs {
552	654		pub actor: User,
⋯
554	656		pub id: String,
555	657		#[serde(default)]
556	658		pub failed_only: bool,
	659	+	#[serde(default)]
	660	+	pub job: Option<String>,
	661	+	#[serde(default)]
	662	+	pub debug: bool,
	663	+	/// `cancel`: stop running jobs outright, without their cleanup steps.
	664	+	#[serde(default)]
	665	+	pub force: bool,
557	666		}
558	667
559	668		/// `set_workflow_enabled`. Members only. Returns `Outcome<Workflow>`.

1	+	-- Runs as GitHub keeps them: every attempt with its own jobs and logs, a
2	+	-- step's job summary ($GITHUB_STEP_SUMMARY), debug re-runs, and cancelling
3	+	-- that lets a job clean up. See src/plan.rs (rerun, cancel_run, job_report)
4	+	-- and src/views.rs (run, logs, summaries).
5	+
6	+	-- A job's summary, a step at a time: the Markdown each step wrote to
7	+	-- $GITHUB_STEP_SUMMARY, masked by the runner. Kept under the same id as the
8	+	-- job's logs (a live job's, or an earlier attempt's: job_attempts.log_id).
9	+	CREATE TABLE IF NOT EXISTS job_summaries (
10	+	job_id TEXT NOT NULL,
11	+	step INTEGER NOT NULL,
12	+	markdown TEXT NOT NULL,
13	+	PRIMARY KEY (job_id, step)
14	+	);
15	+
16	+	-- A run's earlier attempts: how each ended, and who started it.
17	+	CREATE TABLE IF NOT EXISTS run_attempts (
18	+	run_id TEXT NOT NULL,
19	+	attempt INTEGER NOT NULL,
20	+	repo_id TEXT NOT NULL,
21	+	conclusion TEXT,
22	+	-- Who started this attempt: the run's actor for the first, whoever
23	+	-- re-ran it for the rest.
24	+	actor TEXT,
25	+	debug INTEGER NOT NULL DEFAULT 0,
26	+	started_at TEXT,
27	+	finished_at TEXT,
28	+	PRIMARY KEY (run_id, attempt)
29	+	);
30	+
31	+	-- Each job of an earlier attempt as it ended. `id` is the snapshot's own
32	+	-- id, shown as the job's id when that attempt is viewed; `log_id` is where
33	+	-- its logs and summary are kept: the snapshot's id once the job ran again,
34	+	-- or the live job's while that one still carries the result (a job a
35	+	-- "re-run failed jobs" left alone).
36	+	CREATE TABLE IF NOT EXISTS job_attempts (
37	+	id TEXT PRIMARY KEY,
38	+	run_id TEXT NOT NULL,
39	+	repo_id TEXT NOT NULL,
40	+	attempt INTEGER NOT NULL,
41	+	job_id TEXT NOT NULL,
42	+	log_id TEXT NOT NULL,
43	+	key TEXT NOT NULL,
44	+	ordinal INTEGER NOT NULL DEFAULT 0,
45	+	name TEXT NOT NULL,
46	+	needs TEXT NOT NULL DEFAULT '[]',
47	+	status TEXT NOT NULL,
48	+	conclusion TEXT,
49	+	steps TEXT NOT NULL DEFAULT '[]',
50	+	annotations TEXT NOT NULL DEFAULT '[]',
51	+	reason TEXT,
52	+	environment TEXT,
53	+	labels TEXT,
54	+	runner_name TEXT,
55	+	started_at TEXT,
56	+	finished_at TEXT
57	+	);
58	+	CREATE INDEX IF NOT EXISTS job_attempts_by_run ON job_attempts (run_id, attempt);
59	+	CREATE INDEX IF NOT EXISTS job_attempts_by_log ON job_attempts (log_id);
60	+
61	+	-- Who started the current attempt (the run's actor until it is re-run),
62	+	-- and whether it runs with debug logging.
63	+	ALTER TABLE runs ADD COLUMN triggering_actor TEXT;
64	+	ALTER TABLE runs ADD COLUMN debug INTEGER NOT NULL DEFAULT 0;
65	+
66	+	-- Cancelling: a running job is told to stop and given time to run its
67	+	-- `if: always()` and `cancelled()` steps and its post steps. Past the grace
68	+	-- period it is stopped outright.
69	+	ALTER TABLE jobs ADD COLUMN cancel_requested_at TEXT;

29	29		/// The most a single log report may add.
30	30		const MAX_CHUNK_BYTES: usize = 256 * 1024;
31	31		const MAX_ANNOTATIONS: usize = 50;
	32	+	/// The most one step's job summary keeps, in bytes, and how many steps of a
	33	+	/// job may have one, as on GitHub.
	34	+	pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
	35	+	pub(crate) const MAX_SUMMARIES: u32 = 20;
	36	+
	37	+	/// Whether a step's summary of `adding` bytes is kept: a job keeps up to
	38	+	/// 20 steps' summaries (`steps` it has, `mine` this step's bytes so far)
	39	+	/// of up to 1 MiB each.
	40	+	pub(crate) fn summary_fits(steps: u32, mine: Option<usize>, adding: usize) -> bool {
	41	+	if adding == 0 {
	42	+	return false;
	43	+	}
	44	+	match mine {
	45	+	Some(used) => used + adding <= MAX_SUMMARY_BYTES,
	46	+	None => steps < MAX_SUMMARIES && adding <= MAX_SUMMARY_BYTES,
	47	+	}
	48	+	}
32	49
33	50		/// The repository whose unfinished runs `event` stops, by id: one deleted,
34	51		/// or archived (not unarchived).
⋯
98	115		/// Whether its concurrency group cancels what it replaces.
99	116		#[serde(default)]
100	117		pub cancel_in_progress: u32,
	118	+	/// Who started the current attempt, once it is re-run (migration 0008).
	119	+	#[serde(default)]
	120	+	pub triggering_actor: Option<String>,
	121	+	/// The current attempt runs with debug logging.
	122	+	#[serde(default)]
	123	+	pub debug: u32,
101	124		}
102	125
103	126		impl RunRow {
⋯
192	215		pub concurrency_group: Option<String>,
193	216		#[serde(default)]
194	217		pub cancel_in_progress: u32,
	218	+	/// When it was told to stop, while it runs its cleanup steps
	219	+	/// (migration 0008).
	220	+	#[serde(default)]
	221	+	pub cancel_requested_at: Option<String>,
	222	+	}
	223	+
	224	+	/// How long a cancelled job has to run its `if: always()` and `cancelled()`
	225	+	/// steps and its post steps before it is stopped outright, as on GitHub.
	226	+	pub const CANCEL_GRACE_MS: u64 = 5 * 60 * 1000;
	227	+
	228	+	/// Which jobs a re-run runs again.
	229	+	#[derive(Clone, Copy, Debug, PartialEq)]
	230	+	pub enum Rerun<'a> {
	231	+	All,
	232	+	/// Those that did not succeed.
	233	+	Failed,
	234	+	/// One job's key.
	235	+	Job(&'a str),
	236	+	}
	237	+
	238	+	/// The keys a re-run runs again, in `order`: those `which` names and,
	239	+	/// transitively, every key that needs one of them. `needs` gives a key's
	240	+	/// needs; `succeeded` whether every job of a key succeeded.
	241	+	pub fn rerun_keys(order: &[&str], needs: impl Fn(&str) -> Vec<String>, succeeded: impl Fn(&str) -> bool, which: Rerun) -> Vec<String> {
	242	+	let mut again: Vec<String> = Vec::new();
	243	+	for key in order {
	244	+	let named = match which {
	245	+	Rerun::All => true,
	246	+	Rerun::Failed => !succeeded(key),
	247	+	Rerun::Job(job) => *key == job,
	248	+	};
	249	+	if named \|\| needs(key).iter().any(\|need\| again.contains(need)) {
	250	+	again.push((*key).to_owned());
	251	+	}
	252	+	}
	253	+	again
	254	+	}
	255	+
	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`).
	258	+	pub fn top_key(key: &str) -> &str {
	259	+	key.split('/').next().unwrap_or(key)
	260	+	}
	261	+
	262	+	/// Debug logging for a job, as a re-run with it turned on gives GitHub's:
	263	+	/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and
	264	+	/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of
	265	+	/// those names (`::debug::` lines shown, and how each step's `if` read).
	266	+	pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {
	267	+	variables.insert("RUNNER_DEBUG".into(), json!("1"));
	268	+	variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));
	269	+	variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));
	270	+	runner["debug"] = json!("1");
195	271		}
196	272
197	273		impl JobRow {
⋯
572	648		for other in &others {
573	649		if cancel_in_progress \|\| other.status == "pending" {
574	650		// A newer run replaces a waiting one, as on GitHub.
575		−	self.cancel_run(other, "A newer run in the same concurrency group replaced it.").await?;
	651	+	self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;
576	652		}
577	653		}
578	654		if !cancel_in_progress && others.iter().any(\|other\| other.status != "pending") {
⋯
1363	1439		Box::pin(self.advance(&job.run_id)).await
1364	1440		}
1365	1441
1366		−	/// Cancels a job, stopping its sandbox if it has one.
	1442	+	/// Cancels a job. One that is running is told to stop (the answer to
	1443	+	/// its next report), and runs its `if: always()` and `cancelled()`
	1444	+	/// steps and its post steps before it ends `cancelled`; one that has
	1445	+	/// not started is cancelled at once.
1367	1446		async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
	1447	+	if job.status == "in_progress" && job.started_at.is_some() {
	1448	+	self.db
	1449	+	.prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
	1450	+	.bind(&[now().into(), reason.into(), job.id.as_str().into()])?
	1451	+	.run()
	1452	+	.await?;
	1453	+	return Ok(());
	1454	+	}
	1455	+	self.hard_stop(job, reason).await
	1456	+	}
	1457	+
	1458	+	/// Cancels a job outright, stopping its sandbox if it has one.
	1459	+	async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
1368	1460		// A self-hosted runner hears it was cancelled on its next poll.
1369	1461		if job.status == "in_progress" && job.runner_id.is_none() {
1370	1462		let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
⋯
1600	1692		Ok(())
1601	1693		}
1602	1694
1603		−	/// Cancels a run: its waiting and queued jobs, and stops its running ones.
1604		−	pub async fn cancel_run(&self, run: &RunRow, reason: &str) -> Result<()> {
	1695	+	/// Cancels a run: its waiting and queued jobs at once, and its running
	1696	+	/// ones gracefully (`stop_job`), or outright when `force`.
	1697	+	pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
1605	1698		self.db
1606	1699		.prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
1607	1700		.bind(&[run.id.as_str().into()])?
1608	1701		.run()
1609	1702		.await?;
1610	1703		for job in self.job_rows(&run.id).await?.iter().filter(\|job\| job.status != "completed") {
1611		−	self.stop_job(job, reason).await?;
	1704	+	if force {
	1705	+	self.hard_stop(job, reason).await?;
	1706	+	} else {
	1707	+	self.stop_job(job, reason).await?;
	1708	+	}
1612	1709		}
1613	1710		if run.status == "pending" \|\| run.status == "action_required" {
1614	1711		self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
⋯
1627	1724		.await?
1628	1725		.results::<RunRow>()?;
1629	1726		for run in runs {
1630		−	self.cancel_run(&run, "The repository was archived or deleted.").await?;
	1727	+	self.cancel_run(&run, "The repository was archived or deleted.", true).await?;
1631	1728		}
1632	1729		Ok(())
1633	1730		}
⋯
1640	1737		if run.status == "completed" {
1641	1738		return Ok(fail(FailureCode::Conflict, "The run has already finished."));
1642	1739		}
1643		−	self.cancel_run(&run, &format!("{} cancelled the run.", a.actor.username)).await?;
	1740	+	// Cancelling a run that is already cancelling stops its jobs
	1741	+	// outright, without waiting for their cleanup steps.
	1742	+	let force = a.force \|\| run.conclusion.as_deref() == Some("cancelled");
	1743	+	let reason = if force {
	1744	+	format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
	1745	+	} else {
	1746	+	format!("{} cancelled the run.", a.actor.username)
	1747	+	};
	1748	+	self.cancel_run(&run, &reason, force).await?;
1644	1749		self.run_summary(&run.id).await
1645	1750		}
1646	1751
1647		−	/// Runs again: every job, or with `failed_only` those that did not
1648		−	/// succeed and the jobs that need them.
	1752	+	/// Runs again, as a new attempt: every job, with `failed_only` those
	1753	+	/// that did not succeed, or with `job` that one; each with the jobs that
	1754	+	/// need them. The attempt that ends is kept, its jobs and their logs and
	1755	+	/// summaries with it (`job_attempts`, `run_attempts`).
1649	1756		pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
1650	1757		if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
1651	1758		return Ok(Outcome::Fail(refused));
1652	1759		}
1653		−	let run = check!(self.run_in(&a.repo, &a.id).await?);
	1760	+	// A job alone (`POST …/jobs/{job}/rerun`) names its run.
	1761	+	let run_id = match (&a.job, a.id.is_empty()) {
	1762	+	(Some(job), true) => {
	1763	+	#[derive(Deserialize)]
	1764	+	struct Of {
	1765	+	run_id: String,
	1766	+	}
	1767	+	let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
	1768	+	match of {
	1769	+	Some(of) => of.run_id,
	1770	+	None => return Ok(fail(FailureCode::NotFound, "No such job.")),
	1771	+	}
	1772	+	}
	1773	+	_ => a.id.clone(),
	1774	+	};
	1775	+	let run = check!(self.run_in(&a.repo, &run_id).await?);
1654	1776		if run.status != "completed" {
1655	1777		return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
1656	1778		}
⋯
1667	1789		}
1668	1790		let jobs = self.job_rows(&run.id).await?;
1669	1791		let workflow = workflow::parse(&run.source).ok();
1670		−	// Which keys run again: failed ones and, transitively, those needing them.
1671		−	let mut again: Vec<String> = Vec::new();
1672		−	for key in workflow.as_ref().map(\|w\| w.job_order()).unwrap_or_default() {
1673		−	let rows: Vec<&JobRow> = jobs.iter().filter(\|j\| j.key == key).collect();
1674		−	let failed = rows.iter().any(\|row\| row.conclusion.as_deref() != Some("success"));
1675		−	let needs_again = rows.first().is_some_and(\|row\| row.needs().iter().any(\|need\| again.contains(need)));
1676		−	if !a.failed_only \|\| failed \|\| needs_again {
1677		−	again.push(key.to_owned());
1678		−	}
1679		−	}
	1792	+	let target = match &a.job {
	1793	+	Some(id) => match jobs.iter().find(\|job\| &job.id == id) {
	1794	+	Some(job) => Some(top_key(&job.key).to_owned()),
	1795	+	None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
	1796	+	},
	1797	+	None => None,
	1798	+	};
	1799	+	let which = match (&target, a.failed_only) {
	1800	+	(Some(key), _) => Rerun::Job(key),
	1801	+	(None, true) => Rerun::Failed,
	1802	+	(None, false) => Rerun::All,
	1803	+	};
	1804	+	let order = workflow.as_ref().map(\|w\| w.job_order()).unwrap_or_default();
	1805	+	let again = rerun_keys(
	1806	+	&order,
	1807	+	\|key\| jobs.iter().find(\|j\| j.key == key).map(JobRow::needs).unwrap_or_default(),
	1808	+	\|key\| jobs.iter().filter(\|j\| j.key == key).all(\|j\| j.conclusion.as_deref() == Some("success")),
	1809	+	which,
	1810	+	);
1680	1811		if again.is_empty() {
1681	1812		return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
1682	1813		}
1683		−	let mut statements = Vec::new();
	1814	+	// The jobs that run again, a called workflow's with the job calling it.
	1815	+	let rerun_ids: Vec<&str> = jobs.iter().filter(\|job\| again.iter().any(\|key\| key == top_key(&job.key))).map(\|job\| job.id.as_str()).collect();
	1816	+	let ids = serde_json::to_string(&rerun_ids)?;
	1817	+	let ended = run.attempt.to_string();
	1818	+	let ended = ended.as_str();
	1819	+	let mut statements = vec![
	1820	+	// The attempt that ends, as it ended.
	1821	+	self.db
	1822	+	.prepare(
	1823	+	"INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
	1824	+	SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
	1825	+	)
	1826	+	.bind(&[run.id.as_str().into()])?,
	1827	+	// Earlier attempts that showed a job's logs from where it ran
	1828	+	// then now find them where they move to.
	1829	+	self.db
	1830	+	.prepare("UPDATE job_attempts SET log_id = log_id \|\| '.' \|\| ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
	1831	+	.bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
	1832	+	// Each of its jobs: one that runs again keeps its logs under
	1833	+	// `{id}.{attempt}`; one left alone is still the live job's.
	1834	+	self.db
	1835	+	.prepare(
	1836	+	"INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
	1837	+	steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
	1838	+	SELECT id \|\| '.' \|\| ?1, run_id, repo_id, ?1, id,
	1839	+	CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id \|\| '.' \|\| ?1 ELSE id END,
	1840	+	key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
	1841	+	FROM jobs WHERE run_id = ?2 ORDER BY rowid",
	1842	+	)
	1843	+	.bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
	1844	+	self.db
	1845	+	.prepare("UPDATE logs SET job_id = job_id \|\| '.' \|\| ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
	1846	+	.bind(&[ended.into(), ids.as_str().into()])?,
	1847	+	self.db
	1848	+	.prepare("UPDATE job_summaries SET job_id = job_id \|\| '.' \|\| ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
	1849	+	.bind(&[ended.into(), ids.as_str().into()])?,
	1850	+	];
1684	1851		for key in &again {
1685		−	statements.push(self.db.prepare("DELETE FROM logs WHERE job_id IN (SELECT id FROM jobs WHERE run_id = ? AND key = ?)").bind(&[run.id.as_str().into(), key.as_str().into()])?);
1686	1852		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()])?);
1687	1853		// The jobs of a workflow it called are made again when it calls it again.
1688	1854		statements.push(
1689	1855		self.db
1690		−	.prepare("DELETE FROM logs WHERE job_id IN (SELECT id FROM jobs WHERE run_id = ? AND key LIKE ?)")
1691		−	.bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1692		−	);
1693		−	statements.push(
1694		−	self.db
1695	1856		.prepare("DELETE FROM jobs WHERE run_id = ? AND key LIKE ?")
1696	1857		.bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1697	1858		);
⋯
1701	1862		"UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
1702	1863		matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
1703	1864		labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,
1704		−	concurrency_group = NULL, cancel_in_progress = 0 WHERE run_id = ? AND key = ?",
	1865	+	concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
1705	1866		)
1706	1867		.bind(&[run.id.as_str().into(), key.as_str().into()])?,
1707	1868		);
1708	1869		}
1709	1870		statements.push(
1710	1871		self.db
1711		−	.prepare("UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL WHERE id = ?")
1712		−	.bind(&[run.id.as_str().into()])?,
	1872	+	.prepare(
	1873	+	"UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
	1874	+	triggering_actor = ?, debug = ? WHERE id = ?",
	1875	+	)
	1876	+	.bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
1713	1877		);
1714	1878		self.db.batch(statements).await?;
1715	1879		if let Some(run) = self.run_row(&run.id).await? {
⋯
1867	2031		// machine rather than g1t's sandbox.
1868	2032		let mut variables = info.variables(&job.key);
1869	2033		variables.insert("GITHUB_RETENTION_DAYS".into(), json!(retention_days.to_string()));
1870		−	let runner = match &job.runner_id {
	2034	+	let mut runner = match &job.runner_id {
1871	2035		Some(id) => self.runner_context_for(id, &mut variables).await?,
1872	2036		None => runner_context(),
1873	2037		};
	2038	+	// A re-run with debug logging: what GitHub sets for one.
	2039	+	if run.debug != 0 {
	2040	+	debug_logging(&mut variables, &mut runner);
	2041	+	}
1874	2042
1875	2043		// Where to check out: a pull request's fork, or the repository.
1876	2044		let clone_url = match run.pull {
⋯
2030	2198		}
2031	2199		self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2032	2200		}
	2201	+	"summary" => {
	2202	+	// `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
	2203	+	// step's summary, up to 1 MiB a step and 20 steps a job.
	2204	+	let step = report["step"].as_u64().unwrap_or(0) as u32;
	2205	+	let markdown = report["markdown"].as_str().unwrap_or_default();
	2206	+	#[derive(Deserialize)]
	2207	+	struct Held {
	2208	+	steps: u32,
	2209	+	mine: Option<f64>,
	2210	+	}
	2211	+	let held = self
	2212	+	.db
	2213	+	.prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
	2214	+	.bind(&[step.into(), job.id.as_str().into()])?
	2215	+	.first::<Held>(None)
	2216	+	.await?;
	2217	+	let (steps, mine) = held.map_or((0, None), \|held\| (held.steps, held.mine.map(\|n\| n as usize)));
	2218	+	if summary_fits(steps, mine, markdown.len()) {
	2219	+	self.db
	2220	+	.prepare(
	2221	+	"INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
	2222	+	ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown \|\| excluded.markdown",
	2223	+	)
	2224	+	.bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
	2225	+	.run()
	2226	+	.await?;
	2227	+	}
	2228	+	self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
	2229	+	}
2033	2230		"annotation" => {
2034	2231		let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2035	2232		if annotations.len() < MAX_ANNOTATIONS {
⋯
2047	2244		.await?;
2048	2245		}
2049	2246		}
	2247	+	// Nothing to tell; the answer says whether to stop.
	2248	+	"ping" => {
	2249	+	self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
	2250	+	}
2050	2251		"done" => {
2051		−	let conclusion = report["conclusion"]
2052		−	.as_str()
2053		−	.filter(\|c\| matches!(*c, "success" \| "failure" \| "cancelled"))
2054		−	.unwrap_or("failure");
	2252	+	// A job told to stop ends cancelled, however its cleanup went.
	2253	+	let conclusion = if job.cancel_requested_at.is_some() {
	2254	+	"cancelled"
	2255	+	} else {
	2256	+	report["conclusion"]
	2257	+	.as_str()
	2258	+	.filter(\|c\| matches!(*c, "success" \| "failure" \| "cancelled"))
	2259	+	.unwrap_or("failure")
	2260	+	};
2055	2261		let outputs = report["outputs"].as_object().cloned();
2056	2262		Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2057	2263		}
2058	2264		other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2059	2265		}
2060		−	Ok(Outcome::Ok(json!({ "ok": true })))
	2266	+	// `cancelled`: the run was cancelled, and the runner should stop
	2267	+	// the step it is on and run only its cleanup steps.
	2268	+	Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
2061	2269		}
2062	2270
2063	2271		// --- Every minute ---------------------------------------------------------------
⋯
2075	2283		let silent = job.seen_at.as_deref().is_some_and(\|seen\| seen < before(SILENT_MS).as_str());
2076	2284		let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2077	2285		let over = job.started_at.as_deref().is_some_and(\|started\| started < before(limit).as_str());
2078		−	if over {
	2286	+	let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(\|at\| at < before(CANCEL_GRACE_MS).as_str());
	2287	+	if cancel_overdue {
	2288	+	// Cancelled, and still going after its grace period.
	2289	+	self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
	2290	+	self.advance(&job.run_id).await?;
	2291	+	} else if over {
2079	2292		let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
2080	2293		// A self-hosted runner is told to stop on its next poll.
2081	2294		if job.runner_id.is_none() {
⋯
2231	2444		assert_eq!(deployment_outcome(&[None]), None);
2232	2445		}
2233	2446		}
	2447	+
	2448	+	#[cfg(test)]
	2449	+	mod reruns {
	2450	+	use serde_json::{Map, Value, json};
	2451	+
	2452	+	use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, debug_logging, rerun_keys, summary_fits, top_key};
	2453	+
	2454	+	/// build ← test ← deploy, and lint on its own.
	2455	+	fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
	2456	+	let order = ["build", "lint", "test", "deploy"];
	2457	+	let needs = \|key: &str\| -> Vec<String> {
	2458	+	match key {
	2459	+	"test" => vec!["build".into()],
	2460	+	"deploy" => vec!["test".into()],
	2461	+	_ => Vec::new(),
	2462	+	}
	2463	+	};
	2464	+	rerun_keys(&order, needs, \|key\| !failed.contains(&key), which)
	2465	+	}
	2466	+
	2467	+	#[test]
	2468	+	fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
	2469	+	assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
	2470	+	assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
	2471	+	assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
	2472	+	assert!(keys(Rerun::Failed, &[]).is_empty());
	2473	+	// One job, whatever it came to, and what depends on it.
	2474	+	assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
	2475	+	assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
	2476	+	assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
	2477	+	assert!(keys(Rerun::Job("missing"), &[]).is_empty());
	2478	+	}
	2479	+
	2480	+	#[test]
	2481	+	fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
	2482	+	assert_eq!(top_key("build/test"), "build");
	2483	+	assert_eq!(top_key("build/inner/test"), "build");
	2484	+	assert_eq!(top_key("lint"), "lint");
	2485	+	}
	2486	+
	2487	+	#[test]
	2488	+	fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
	2489	+	assert!(summary_fits(0, None, 10));
	2490	+	assert!(!summary_fits(0, None, 0), "nothing to keep");
	2491	+	assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
	2492	+	assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
	2493	+	assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
	2494	+	assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
	2495	+	assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
	2496	+	}
	2497	+
	2498	+	#[test]
	2499	+	fn a_debug_rerun_sets_what_github_sets() {
	2500	+	let mut variables = Map::new();
	2501	+	let mut runner = json!({ "name": "g1t", "debug": "" });
	2502	+	debug_logging(&mut variables, &mut runner);
	2503	+	assert_eq!(variables["RUNNER_DEBUG"], "1");
	2504	+	assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
	2505	+	assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
	2506	+	assert_eq!(runner["debug"], Value::String("1".into()));
	2507	+	assert_eq!(runner["name"], "g1t");
	2508	+	}
	2509	+	}

1		−	//! Reading: workflows, runs, a run's jobs, and a job's log.
	1	+	//! Reading: workflows, runs (any attempt), a run's jobs, a job's log and
	2	+	//! its steps' summaries.
2	3
3	4		use g1t_actions::workflow::{self, Severity};
4	5		use g1t_contracts::access::Capability;
5	6		use g1t_contracts::checks::{ActionsChecksArgs, MAX_COMMITS};
6	7		use g1t_contracts::actions::{
7		−	Annotation, Job, JobLog, LogChunk, LogsArgs, RunArgs, RunDetail, RunsArgs, SetWorkflowEnabledArgs, StepState, Workflow, WorkflowNote,
8		−	WorkflowRun, WorkflowsArgs,
	8	+	Annotation, Job, JobLog, JobLogText, JobLogTextArgs, JobSummary, LogChunk, LogsArgs, MAX_RUN_LOG_BYTES, RunArgs, RunAttempt, RunDetail,
	9	+	RunLogsArgs, RunsArgs, SetWorkflowEnabledArgs, StepState, StepSummary, SummariesArgs, Workflow, WorkflowNote, WorkflowRun, WorkflowsArgs,
9	10		};
10	11		use g1t_contracts::{FailureCode, Outcome};
11	12		use serde::Deserialize;
⋯
37	38		.unwrap_or_default()
38	39		}
39	40
	41	+	/// A job of an earlier attempt, as it ended (migration 0008).
	42	+	#[derive(Clone, Deserialize)]
	43	+	pub struct AttemptJobRow {
	44	+	pub id: String,
	45	+	pub run_id: String,
	46	+	pub log_id: String,
	47	+	pub key: String,
	48	+	pub name: String,
	49	+	pub needs: String,
	50	+	pub status: String,
	51	+	pub conclusion: Option<String>,
	52	+	pub steps: String,
	53	+	pub annotations: String,
	54	+	pub reason: Option<String>,
	55	+	pub environment: Option<String>,
	56	+	pub labels: Option<String>,
	57	+	pub runner_name: Option<String>,
	58	+	pub started_at: Option<String>,
	59	+	pub finished_at: Option<String>,
	60	+	}
	61	+
	62	+	fn attempt_job_view(row: AttemptJobRow) -> Job {
	63	+	Job {
	64	+	needs: serde_json::from_str(&row.needs).unwrap_or_default(),
	65	+	id: row.id,
	66	+	run_id: row.run_id,
	67	+	key: row.key,
	68	+	name: row.name,
	69	+	status: row.status,
	70	+	conclusion: row.conclusion,
	71	+	steps: serde_json::from_str::<Vec<StepState>>(&row.steps).unwrap_or_default(),
	72	+	annotations: serde_json::from_str::<Vec<Annotation>>(&row.annotations).unwrap_or_default(),
	73	+	reason: row.reason,
	74	+	started_at: row.started_at,
	75	+	finished_at: row.finished_at,
	76	+	environment: row.environment,
	77	+	self_hosted: row.labels.is_some(),
	78	+	runner: row.runner_name,
	79	+	cancelling: false,
	80	+	}
	81	+	}
	82	+
	83	+	/// An earlier attempt of a run, as it ended.
	84	+	#[derive(Clone, Deserialize)]
	85	+	struct AttemptRow {
	86	+	attempt: u64,
	87	+	conclusion: Option<String>,
	88	+	actor: Option<String>,
	89	+	debug: u32,
	90	+	started_at: Option<String>,
	91	+	finished_at: Option<String>,
	92	+	}
	93	+
	94	+	/// One attempt's jobs, each with where its logs and summary are kept.
	95	+	struct AttemptJobs {
	96	+	jobs: Vec<Job>,
	97	+	log_ids: Vec<String>,
	98	+	}
	99	+
40	100		fn job_view(row: JobRow) -> Job {
41	101		let needs = row.needs();
	102	+	let cancelling = row.cancel_requested_at.is_some() && row.status != "completed";
42	103		Job {
	104	+	cancelling,
43	105		id: row.id,
44	106		run_id: row.run_id,
45	107		key: row.key,
⋯
154	216		Ok(Outcome::Ok(rows.iter().map(RunRow::summary).collect()))
155	217		}
156	218
	219	+	/// Every attempt of a run, oldest first, the current one last.
	220	+	async fn attempts_of(&self, run: &RunRow) -> Result<Vec<RunAttempt>> {
	221	+	let mut attempts: Vec<RunAttempt> = self
	222	+	.db
	223	+	.prepare("SELECT attempt, conclusion, actor, debug, started_at, finished_at FROM run_attempts WHERE run_id = ? ORDER BY attempt")
	224	+	.bind(&[run.id.as_str().into()])?
	225	+	.all()
	226	+	.await?
	227	+	.results::<AttemptRow>()?
	228	+	.into_iter()
	229	+	.filter(\|row\| row.attempt < run.attempt)
	230	+	.map(\|row\| RunAttempt {
	231	+	attempt: row.attempt,
	232	+	status: "completed".to_owned(),
	233	+	conclusion: row.conclusion,
	234	+	actor: row.actor,
	235	+	debug: row.debug != 0,
	236	+	started_at: row.started_at,
	237	+	finished_at: row.finished_at,
	238	+	})
	239	+	.collect();
	240	+	attempts.push(RunAttempt {
	241	+	attempt: run.attempt,
	242	+	status: run.status.clone(),
	243	+	conclusion: run.conclusion.clone().filter(\|_\| run.status == "completed"),
	244	+	actor: run.triggering_actor.clone().or_else(\|\| run.actor.clone()),
	245	+	debug: run.debug != 0,
	246	+	started_at: run.started_at.clone(),
	247	+	finished_at: run.finished_at.clone(),
	248	+	});
	249	+	Ok(attempts)
	250	+	}
	251	+
	252	+	/// The jobs of one attempt: the live ones for the current attempt, the
	253	+	/// kept ones for an earlier attempt. None for an attempt it never had.
	254	+	async fn attempt_jobs(&self, run: &RunRow, attempt: Option<u64>) -> Result<Option<AttemptJobs>> {
	255	+	match attempt.filter(\|n\| *n != run.attempt) {
	256	+	None => {
	257	+	let jobs: Vec<Job> = self.job_rows(&run.id).await?.into_iter().map(job_view).collect();
	258	+	let log_ids = jobs.iter().map(\|job\| job.id.clone()).collect();
	259	+	Ok(Some(AttemptJobs { jobs, log_ids }))
	260	+	}
	261	+	Some(n) if n == 0 \|\| n > run.attempt => Ok(None),
	262	+	Some(n) => {
	263	+	let rows = self
	264	+	.db
	265	+	.prepare("SELECT * FROM job_attempts WHERE run_id = ? AND attempt = ? ORDER BY rowid")
	266	+	.bind(&[run.id.as_str().into(), (n as f64).into()])?
	267	+	.all()
	268	+	.await?
	269	+	.results::<AttemptJobRow>()?;
	270	+	let log_ids = rows.iter().map(\|row\| row.log_id.clone()).collect();
	271	+	Ok(Some(AttemptJobs { jobs: rows.into_iter().map(attempt_job_view).collect(), log_ids }))
	272	+	}
	273	+	}
	274	+	}
	275	+
157	276		pub async fn run(&self, a: RunArgs) -> Result<Outcome<RunDetail>> {
158	277		if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
159	278		return Ok(fail(FailureCode::NotFound, "There is no such repository."));
160	279		}
161	280		let run = check!(self.run_in(&a.repo, &a.id).await?);
162		−	let jobs: Vec<Job> = self.job_rows(&run.id).await?.into_iter().map(job_view).collect();
163		−	let pending_deployments = self.pending_for(&run, &a.viewer).await?;
	281	+	let Some(AttemptJobs { jobs, .. }) = self.attempt_jobs(&run, a.attempt).await? else {
	282	+	return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
	283	+	};
	284	+	let attempts = self.attempts_of(&run).await?;
164	285		let mut summary = run.summary();
	286	+	// An earlier attempt: as it ended, and nothing waits for it.
	287	+	if let Some(earlier) = a.attempt.filter(\|n\| *n != run.attempt).and_then(\|n\| attempts.iter().find(\|at\| at.attempt == n)) {
	288	+	summary.attempt = earlier.attempt;
	289	+	summary.status = "completed".to_owned();
	290	+	summary.conclusion = earlier.conclusion.clone();
	291	+	summary.started_at = earlier.started_at.clone();
	292	+	summary.finished_at = earlier.finished_at.clone();
	293	+	return Ok(Outcome::Ok(RunDetail {
	294	+	notes: notes(&run.source),
	295	+	run: summary,
	296	+	jobs,
	297	+	approval: run.approval(),
	298	+	pending_deployments: Vec::new(),
	299	+	attempts,
	300	+	}));
	301	+	}
	302	+	let pending_deployments = self.pending_for(&run, &a.viewer).await?;
165	303		// Jobs held at an environment's rules: the run waits, as on GitHub.
166	304		if matches!(summary.status.as_str(), "queued" \| "in_progress") && jobs.iter().any(\|job\| job.status == "pending") {
167	305		summary.status = "waiting".to_owned();
⋯
172	310		jobs,
173	311		approval: run.approval(),
174	312		pending_deployments,
	313	+	attempts,
175	314		}))
176	315		}
177	316
178		−	pub async fn logs(&self, a: LogsArgs) -> Result<Outcome<JobLog>> {
179		−	if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
180		−	return Ok(fail(FailureCode::NotFound, "There is no such repository."));
181		−	}
182		−	let job = self
	317	+	/// A job of the repository's runs by the id a run's page gives it: a
	318	+	/// live job, or one of an earlier attempt. Its name, steps, where its
	319	+	/// logs and summary are kept, and whether it has finished.
	320	+	async fn job_for_log(&self, repo: &g1t_contracts::repos::RepoPath, id: &str) -> Result<Option<(Job, String, bool)>> {
	321	+	let full = format!("{}/{}", repo.namespace, repo.name);
	322	+	let live = self
183	323		.db
184	324		.prepare("SELECT jobs.* FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ? AND lower(runs.repo) = lower(?)")
185		−	.bind(&[a.job.as_str().into(), format!("{}/{}", a.repo.namespace, a.repo.name).into()])?
	325	+	.bind(&[id.into(), full.as_str().into()])?
186	326		.first::<JobRow>(None)
187	327		.await?;
188		−	let Some(job) = job else {
189		−	return Ok(fail(FailureCode::NotFound, "No such job."));
190		−	};
	328	+	if let Some(job) = live {
	329	+	let done = job.status == "completed";
	330	+	let view = job_view(job);
	331	+	let log_id = view.id.clone();
	332	+	return Ok(Some((view, log_id, done)));
	333	+	}
	334	+	let kept = self
	335	+	.db
	336	+	.prepare(
	337	+	"SELECT job_attempts.* FROM job_attempts JOIN runs ON runs.id = job_attempts.run_id
	338	+	WHERE job_attempts.id = ? AND lower(runs.repo) = lower(?)",
	339	+	)
	340	+	.bind(&[id.into(), full.as_str().into()])?
	341	+	.first::<AttemptJobRow>(None)
	342	+	.await?;
	343	+	Ok(kept.map(\|row\| {
	344	+	let log_id = row.log_id.clone();
	345	+	(attempt_job_view(row), log_id, true)
	346	+	}))
	347	+	}
	348	+
	349	+	async fn log_chunks(&self, log_id: &str, after: u64, limit: u32) -> Result<Vec<LogChunk>> {
191	350		#[derive(Deserialize)]
192	351		struct Row {
193	352		seq: u64,
194	353		step: u32,
195	354		text: String,
196	355		}
197		−	let chunks = self
	356	+	Ok(self
198	357		.db
199		−	.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 500")
200		−	.bind(&[job.id.as_str().into(), (a.after as f64).into()])?
	358	+	.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
	359	+	.bind(&[log_id.into(), (after as f64).into(), limit.into()])?
201	360		.all()
202	361		.await?
203	362		.results::<Row>()?
204	363		.into_iter()
205	364		.map(\|row\| LogChunk { seq: row.seq, step: row.step, text: row.text })
206		−	.collect();
207		−	Ok(Outcome::Ok(JobLog {
208		−	chunks,
209		−	done: job.status == "completed",
210		−	}))
	365	+	.collect())
	366	+	}
	367	+
	368	+	/// Every chunk of a job's log, a page at a time.
	369	+	async fn whole_log(&self, log_id: &str) -> Result<Vec<LogChunk>> {
	370	+	let mut chunks: Vec<LogChunk> = Vec::new();
	371	+	loop {
	372	+	let after = chunks.last().map_or(0, \|chunk\| chunk.seq);
	373	+	let page = self.log_chunks(log_id, after, 500).await?;
	374	+	let more = page.len() == 500;
	375	+	chunks.extend(page);
	376	+	if !more {
	377	+	return Ok(chunks);
	378	+	}
	379	+	}
	380	+	}
	381	+
	382	+	pub async fn logs(&self, a: LogsArgs) -> Result<Outcome<JobLog>> {
	383	+	if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
	384	+	return Ok(fail(FailureCode::NotFound, "There is no such repository."));
	385	+	}
	386	+	let Some((_, log_id, done)) = self.job_for_log(&a.repo, &a.job).await? else {
	387	+	return Ok(fail(FailureCode::NotFound, "No such job."));
	388	+	};
	389	+	let chunks = self.log_chunks(&log_id, a.after, 500).await?;
	390	+	Ok(Outcome::Ok(JobLog { chunks, done }))
	391	+	}
	392	+
	393	+	/// `job_log_text`: a job's whole log, to download.
	394	+	pub async fn job_log_text(&self, a: JobLogTextArgs) -> Result<Outcome<JobLogText>> {
	395	+	if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
	396	+	return Ok(fail(FailureCode::NotFound, "There is no such repository."));
	397	+	}
	398	+	let Some((job, log_id, done)) = self.job_for_log(&a.repo, &a.job).await? else {
	399	+	return Ok(fail(FailureCode::NotFound, "No such job."));
	400	+	};
	401	+	let chunks = self.whole_log(&log_id).await?;
	402	+	Ok(Outcome::Ok(JobLogText { job_id: job.id, name: job.name, steps: job.steps, chunks, done, omitted: false }))
	403	+	}
	404	+
	405	+	/// `run_logs`: every job's whole log for one attempt, to download as one
	406	+	/// archive.
	407	+	pub async fn run_logs(&self, a: RunLogsArgs) -> Result<Outcome<Vec<JobLogText>>> {
	408	+	if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
	409	+	return Ok(fail(FailureCode::NotFound, "There is no such repository."));
	410	+	}
	411	+	let run = check!(self.run_in(&a.repo, &a.id).await?);
	412	+	let Some(AttemptJobs { jobs, log_ids }) = self.attempt_jobs(&run, a.attempt).await? else {
	413	+	return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
	414	+	};
	415	+	let mut out = Vec::with_capacity(jobs.len());
	416	+	let mut total = 0usize;
	417	+	for (job, log_id) in jobs.into_iter().zip(log_ids) {
	418	+	let done = job.status == "completed";
	419	+	let omitted = total >= MAX_RUN_LOG_BYTES;
	420	+	let chunks = if omitted { Vec::new() } else { self.whole_log(&log_id).await? };
	421	+	total += chunks.iter().map(\|chunk\| chunk.text.len()).sum::<usize>();
	422	+	out.push(JobLogText { job_id: job.id, name: job.name, steps: job.steps, chunks, done, omitted });
	423	+	}
	424	+	Ok(Outcome::Ok(out))
	425	+	}
	426	+
	427	+	/// `summaries`: what each job's steps wrote to `$GITHUB_STEP_SUMMARY`.
	428	+	pub async fn summaries(&self, a: SummariesArgs) -> Result<Outcome<Vec<JobSummary>>> {
	429	+	if self.visible_repo(&a.repo, &a.viewer).await?.is_none() {
	430	+	return Ok(fail(FailureCode::NotFound, "There is no such repository."));
	431	+	}
	432	+	let run = check!(self.run_in(&a.repo, &a.id).await?);
	433	+	let Some(AttemptJobs { jobs, log_ids }) = self.attempt_jobs(&run, a.attempt).await? else {
	434	+	return Ok(fail(FailureCode::NotFound, format!("This run has no attempt {}.", a.attempt.unwrap_or_default())));
	435	+	};
	436	+	#[derive(Deserialize)]
	437	+	struct Row {
	438	+	job_id: String,
	439	+	step: u32,
	440	+	markdown: String,
	441	+	}
	442	+	let rows = self
	443	+	.db
	444	+	.prepare("SELECT job_id, step, markdown FROM job_summaries WHERE job_id IN (SELECT value FROM json_each(?)) ORDER BY step")
	445	+	.bind(&[serde_json::to_string(&log_ids)?.into()])?
	446	+	.all()
	447	+	.await?
	448	+	.results::<Row>()?;
	449	+	Ok(Outcome::Ok(
	450	+	jobs.into_iter()
	451	+	.zip(log_ids)
	452	+	.filter_map(\|(job, log_id)\| {
	453	+	let steps: Vec<StepSummary> = rows
	454	+	.iter()
	455	+	.filter(\|row\| row.job_id == log_id)
	456	+	.map(\|row\| StepSummary { step: row.step, markdown: row.markdown.clone() })
	457	+	.collect();
	458	+	(!steps.is_empty()).then_some(JobSummary { job_id: job.id, name: job.name, steps })
	459	+	})
	460	+	.collect(),
	461	+	))
211	462		}
212	463
213	464		pub async fn set_workflow_enabled(&self, a: SetWorkflowEnabledArgs) -> Result<Outcome<Workflow>> {
⋯
291	542		notes: Vec::new(),
292	543		approval: run.approval(),
293	544		pending_deployments: Vec::new(),
	545	+	attempts: Vec::new(),
294	546		})
295	547		.collect())
296	548		}

950	950		let detail: Outcome<RunDetail> = g1t_kit::call(
951	951		&self.actions,
952	952		"run",
953		−	&RunArgs { repo: repo.clone(), viewer: viewer.clone(), id: run_id },
	953	+	&RunArgs { repo: repo.clone(), viewer: viewer.clone(), id: run_id, attempt: None },
954	954		)
955	955		.await?;
956	956		let Outcome::Ok(detail) = detail else { continue };

215	215		"runs" => reply(&service.runs(args(body)?).await?),
216	216		"run" => reply(&service.run(args(body)?).await?),
217	217		"logs" => reply(&service.logs(args(body)?).await?),
	218	+	"job_log_text" => reply(&service.job_log_text(args(body)?).await?),
	219	+	"run_logs" => reply(&service.run_logs(args(body)?).await?),
	220	+	"summaries" => reply(&service.summaries(args(body)?).await?),
218	221		"dispatch" => reply(&service.dispatch(args(body)?).await?),
219	222		"repository_dispatch" => reply(&service.repository_dispatch(args(body)?).await?),
220	223		"approve_run" => reply(&service.approve_run(args(body)?).await?),

32	32		/// stay with the project.
33	33		pub const PURGED: &[&str] = &[
34	34		"DELETE FROM logs WHERE job_id IN (SELECT id FROM jobs WHERE repo_id = ?1)",
	35	+	// Earlier attempts' logs and summaries, kept under their own ids.
	36	+	"DELETE FROM logs WHERE job_id IN (SELECT id FROM job_attempts WHERE repo_id = ?1)",
	37	+	"DELETE FROM job_summaries WHERE job_id IN (SELECT id FROM jobs WHERE repo_id = ?1)",
	38	+	"DELETE FROM job_summaries WHERE job_id IN (SELECT id FROM job_attempts WHERE repo_id = ?1)",
	39	+	"DELETE FROM job_attempts WHERE repo_id = ?1",
	40	+	"DELETE FROM run_attempts WHERE repo_id = ?1",
35	41		"DELETE FROM jobs WHERE repo_id = ?1",
36	42		"DELETE FROM runs WHERE repo_id = ?1",
37	43		"DELETE FROM workflows WHERE repo_id = ?1",

845	845		let rerun: Outcome<Value> = g1t_kit::call(
846	846		&self.actions,
847	847		"rerun",
848		−	&RunActionArgs { actor: actor.clone(), repo: repo.clone(), id: run_id.to_owned(), failed_only: false },
	848	+	&RunActionArgs { actor: actor.clone(), repo: repo.clone(), id: run_id.to_owned(), failed_only: false, job: None, debug: false, force: false },
849	849		)
850	850		.await?;
851	851		Ok(match rerun {