flagon-io/g1t

public

Merge queue: tested states are deleted once their entry leaves

Each entry's combined state is pushed to g1t-queue/<entry> for its checks and merge_group workflows, and nothing ever removed it, so a busy repository's branch list filled with them (hello had 19). The queue now deletes the branch when an entry lands, fails or is taken out, as GitHub does with gh-readonly-queue/*. The repos service gains delete_branch, which speaks receive-pack with delete-refs and refuses any name not starting with g1t-, so it can never remove a person's branch. A failed delete only logs.

syntaqxcommitted 3d agoParent3e345e7Browse files

5 files+100−50/5 viewed

+2−1View BlameViewed

78	78		Once those pass, the repository's [GitHub Actions](/guides/actions/)
79	79		workflows that run `on: merge_group` run on the state too, with the same
80	80		`merge_group` event GitHub sends, on the branch `g1t-queue/<entry>`. The
81		−	entry waits for them, and lands only if they pass:
	81	+	entry waits for them, and lands only if they pass. The branch is deleted
	82	+	once the entry lands or leaves the queue. A workflow opts in like this:
82	83
83	84		```yaml
84	85		on:

+15−0View BlameViewed

414	414		pub repo_id: String,
415	415		pub branch: String,
416	416		}
	417	+
	418	+	/// Where g1t keeps branches of its own in a repository, such as the merge
	419	+	/// queue's tested states. Only these can be removed with `delete_branch`.
	420	+	pub const G1T_BRANCH_PREFIX: &str = "g1t-";
	421	+
	422	+	/// `delete_branch`: removes a branch g1t made for itself once it is done
	423	+	/// with it, never one of people's: the name must start with
	424	+	/// [`G1T_BRANCH_PREFIX`]. For services, which have no viewer. Returns
	425	+	/// `Outcome<bool>`: whether there was such a branch.
	426	+	#[derive(Debug, Serialize, Deserialize)]
	427	+	#[serde(rename_all = "camelCase")]
	428	+	pub struct DeleteBranchArgs {
	429	+	pub repo_id: String,
	430	+	pub branch: String,
	431	+	}

+28−3View BlameViewed

118	118		new: &str,
119	119		pack: Vec<u8>,
120	120		) -> Result<std::result::Result<(), String>> {
	121	+	update_ref(target, branch, old, new, Some(pack)).await
	122	+	}
	123	+
	124	+	/// Removes `branch` from the target, if it is still at `old`.
	125	+	pub(crate) async fn delete_ref(
	126	+	target: &GitAccess,
	127	+	branch: &str,
	128	+	old: &str,
	129	+	) -> Result<std::result::Result<(), String>> {
	130	+	update_ref(target, branch, Some(old), ZERO_ID, None).await
	131	+	}
	132	+
	133	+	/// One receive-pack command; a deletion sends no pack.
	134	+	async fn update_ref(
	135	+	target: &GitAccess,
	136	+	branch: &str,
	137	+	old: Option<&str>,
	138	+	new: &str,
	139	+	pack: Option<Vec<u8>>,
	140	+	) -> Result<std::result::Result<(), String>> {
121	141		let reference = format!("refs/heads/{branch}");
	142	+	let sends_pack = pack.is_some();
	143	+	let capabilities = if sends_pack { "report-status" } else { "report-status delete-refs" };
122	144		let mut body = pkt_line(&format!(
123		−	"{} {new} {reference}\0 report-status\n",
	145	+	"{} {new} {reference}\0 {capabilities}\n",
124	146		old.unwrap_or(ZERO_ID)
125	147		));
126	148		body.extend_from_slice(FLUSH);
127		−	body.extend(pack);
	149	+	if let Some(pack) = pack {
	150	+	body.extend(pack);
	151	+	}
128	152
129	153		let response = post(target, "git-receive-pack", body).await?;
130	154		let (lines, _) = read_pkt_lines(&response);
⋯
132	156		.into_iter()
133	157		.map(\|line\| String::from_utf8_lossy(line).trim_end().to_owned())
134	158		.collect();
135		−	let unpacked = lines.iter().any(\|line\| line == "unpack ok");
	159	+	// With nothing to unpack a server may not say so.
	160	+	let unpacked = !sends_pack \|\| lines.iter().any(\|line\| line == "unpack ok");
136	161		let updated = lines.iter().any(\|line\| *line == format!("ok {reference}"));
137	162		Ok(if unpacked && updated {
138	163		Ok(())

+31−0View BlameViewed

522	522		.map(\|commit\| commit.hash))
523	523		}
524	524
	525	+	async fn delete_branch(&self, a: DeleteBranchArgs) -> Result<Outcome<bool>> {
	526	+	if !a.branch.starts_with(G1T_BRANCH_PREFIX) {
	527	+	return Ok(Outcome::fail(
	528	+	FailureCode::Forbidden,
	529	+	"Only branches g1t made for itself can be deleted this way.",
	530	+	));
	531	+	}
	532	+	let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
	533	+	return Ok(not_found());
	534	+	};
	535	+	let git = self.store.open(&store_key(&repo)).await?;
	536	+	let Some(old) = git
	537	+	.branches()
	538	+	.await?
	539	+	.into_iter()
	540	+	.find(\|branch\| branch.name == a.branch)
	541	+	.map(\|branch\| branch.hash)
	542	+	else {
	543	+	return Ok(Outcome::Ok(false));
	544	+	};
	545	+	let access = git.access(Scope::Write).await?;
	546	+	if let Err(reason) = land::delete_ref(&access, &a.branch, &old).await? {
	547	+	return Ok(Outcome::fail(
	548	+	FailureCode::Conflict,
	549	+	format!("{} could not be deleted: {reason}", a.branch),
	550	+	));
	551	+	}
	552	+	Ok(Outcome::Ok(true))
	553	+	}
	554	+
525	555		async fn fork_for_pull(&self, a: ForkArgs) -> Result<Outcome<Repo>> {
526	556		let viewer = Some(a.actor.clone());
527	557		let Some(source) = self
⋯
950	980		"head" => reply(&repos.head(args(body)?).await?),
951	981		"behind" => reply(&repos.behind(args(body)?).await?),
952	982		"land" => reply(&repos.land(args(body)?).await?),
	983	+	"delete_branch" => reply(&repos.delete_branch(args(body)?).await?),
953	984		"compare" => reply(&repos.compare(args(body)?).await?),
954	985		_ => Response::error("Unknown method", 404),
955	986		}

+24−1View BlameViewed

12	12
13	13		use futures_util::future::try_join_all;
14	14		use g1t_contracts::events::{ChecksEvent, QueueChanged};
15		−	use g1t_contracts::repos::{GetByIdArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath};
	15	+	use g1t_contracts::repos::{DeleteBranchArgs, GetByIdArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath};
16	16		use g1t_contracts::time::rfc3339;
17	17		use g1t_contracts::work::*;
18	18		use g1t_contracts::{FailureCode, Outcome, User, new_id};
⋯
260	260		])?
261	261		.run()
262	262		.await?;
	263	+	self.drop_branch(entry).await;
263	264		let behind: Vec<&EntryRow> = active
264	265		.iter()
265	266		.filter(\|row\| row.state() != QueueState::Waiting && row.ahead().contains(&pull.number))
⋯
268	269		Ok(true)
269	270		}
270	271
	272	+	/// Removes an entry's tested state from the repository once it has
	273	+	/// left the queue, as GitHub does with its queue's branches. A branch
	274	+	/// left behind is untidy, not wrong, so a failure only logs.
	275	+	async fn drop_branch(&self, row: &EntryRow) {
	276	+	let deleted: Result<Outcome<bool>> = g1t_kit::call(
	277	+	&self.repos,
	278	+	"delete_branch",
	279	+	&DeleteBranchArgs {
	280	+	repo_id: row.repo_id.clone(),
	281	+	branch: row.branch(),
	282	+	},
	283	+	)
	284	+	.await;
	285	+	match deleted {
	286	+	Ok(Outcome::Ok(_)) => {}
	287	+	Ok(Outcome::Fail(failure)) => worker::console_warn!("{}: {}", row.branch(), failure.message),
	288	+	Err(error) => worker::console_warn!("{}: {error}", row.branch()),
	289	+	}
	290	+	}
	291	+
271	292		/// Sends entries back to waiting, to be tested again.
272	293		async fn retest(&self, rows: &[&EntryRow]) -> Result<()> {
273	294		for row in rows {
⋯
596	617		/// it are tested again without it, and the failure is recorded as a
597	618		/// failed check run of its pull request, so a g1t agent is sent back.
598	619		async fn eject(&self, row: &EntryRow, report: &ReportQueueArgs) -> Result<()> {
	620	+	self.drop_branch(row).await;
599	621		let active = self.entries(&row.repo_id, true).await?;
600	622		let behind: Vec<&EntryRow> = active
601	623		.iter()
⋯
762	784		.bind(&[rfc3339(now_ms()).into(), row.id.as_str().into()])?
763	785		.run()
764	786		.await?;
	787	+	self.drop_branch(&row).await;
765	788		self.record_merge(&repo, pull, &actor, row.keep_issue_open != 0, landed)
766	789		.await?;
767	790		}