flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh

git agents cloudflare rust forge mcp

Commit

A line of a job's log no longer shows twice when the runner sends its batch again. The runner numbers each batch of a job's log (1 for the first, counting up through the job) and a batch keeps its number when a report whose answer was lost is sent again; g1t keeps that number in the log's new `chunk` column (migration 0016), under a unique index on the job and the number, and a batch whose number it already has is left out in the same statement that takes the next `seq`. A runner that numbers nothing has every batch kept, as before, and a job run again starts its numbers afresh, its earlier log having moved aside. The service's tests run the statement against the migrations in SQLite, and the runner's against a server that fails the first report. The Actions guide says so.

Claude-Session:: https://claude.ai/code/session_01PnwTRSNaC7vEHTAuBzjhrn

Claudeagent and Claudeagentcommitted 2d agoParentb198e74Browse files

81	81		]
82	82
83	83		[[package]]
	84	+	name = "ahash"
	85	+	version = "0.8.12"
	86	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	87	+	checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
	88	+	dependencies = [
	89	+	"cfg-if",
	90	+	"once_cell",
	91	+	"version_check",
	92	+	"zerocopy",
	93	+	]
	94	+
	95	+	[[package]]
84	96		name = "aho-corasick"
85	97		version = "1.1.5"
86	98		source = "registry+https://github.com/rust-lang/crates.io-index"
⋯
740	752		]
741	753
742	754		[[package]]
	755	+	name = "fallible-iterator"
	756	+	version = "0.3.0"
	757	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	758	+	checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
	759	+
	760	+	[[package]]
	761	+	name = "fallible-streaming-iterator"
	762	+	version = "0.1.9"
	763	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	764	+	checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
	765	+
	766	+	[[package]]
743	767		name = "ff"
744	768		version = "0.14.0"
745	769		source = "registry+https://github.com/rust-lang/crates.io-index"
⋯
930	954		"g1t-kit",
931	955		"g1t-secrets",
932	956		"hex",
	957	+	"rusqlite",
933	958		"serde",
934	959		"serde_json",
935	960		"worker",
⋯
1315	1340
1316	1341		[[package]]
1317	1342		name = "hashbrown"
	1343	+	version = "0.14.5"
	1344	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	1345	+	checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
	1346	+	dependencies = [
	1347	+	"ahash",
	1348	+	]
	1349	+
	1350	+	[[package]]
	1351	+	name = "hashbrown"
1318	1352		version = "0.17.1"
1319	1353		source = "registry+https://github.com/rust-lang/crates.io-index"
1320	1354		checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
⋯
1324	1358
1325	1359		[[package]]
1326	1360		name = "hashlink"
	1361	+	version = "0.9.1"
	1362	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	1363	+	checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af"
	1364	+	dependencies = [
	1365	+	"hashbrown 0.14.5",
	1366	+	]
	1367	+
	1368	+	[[package]]
	1369	+	name = "hashlink"
1327	1370		version = "0.12.2"
1328	1371		source = "registry+https://github.com/rust-lang/crates.io-index"
1329	1372		checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb"
1330	1373		dependencies = [
1331		−	"hashbrown",
	1374	+	"hashbrown 0.17.1",
1332	1375		]
1333	1376
1334	1377		[[package]]
⋯
1624	1667		checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
1625	1668		dependencies = [
1626	1669		"equivalent",
1627		−	"hashbrown",
	1670	+	"hashbrown 0.17.1",
1628	1671		]
1629	1672
1630	1673		[[package]]
⋯
1755	1798		checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
1756	1799
1757	1800		[[package]]
	1801	+	name = "libsqlite3-sys"
	1802	+	version = "0.30.1"
	1803	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	1804	+	checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149"
	1805	+	dependencies = [
	1806	+	"cc",
	1807	+	"pkg-config",
	1808	+	"vcpkg",
	1809	+	]
	1810	+
	1811	+	[[package]]
1758	1812		name = "litemap"
1759	1813		version = "0.8.3"
1760	1814		source = "registry+https://github.com/rust-lang/crates.io-index"
⋯
2444	2498		]
2445	2499
2446	2500		[[package]]
	2501	+	name = "rusqlite"
	2502	+	version = "0.32.1"
	2503	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	2504	+	checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e"
	2505	+	dependencies = [
	2506	+	"bitflags",
	2507	+	"fallible-iterator",
	2508	+	"fallible-streaming-iterator",
	2509	+	"hashlink 0.9.1",
	2510	+	"libsqlite3-sys",
	2511	+	"smallvec",
	2512	+	]
	2513	+
	2514	+	[[package]]
2447	2515		name = "russh"
2448	2516		version = "0.63.3"
2449	2517		source = "registry+https://github.com/rust-lang/crates.io-index"
⋯
3502	3570		checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
3503	3571
3504	3572		[[package]]
	3573	+	name = "vcpkg"
	3574	+	version = "0.2.15"
	3575	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	3576	+	checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
	3577	+
	3578	+	[[package]]
3505	3579		name = "version_check"
3506	3580		version = "0.9.5"
3507	3581		source = "registry+https://github.com/rust-lang/crates.io-index"
⋯
3966	4040		checksum = "57e5b818a27a4cd30884ea380857a5e56f7ec3ba24a3990a3cc0b95af3238e18"
3967	4041		dependencies = [
3968	4042		"arraydeque",
3969		−	"hashlink",
	4043	+	"hashlink 0.12.2",
3970	4044		]
3971	4045
3972	4046		[[package]]
⋯
3993	4067		]
3994	4068
3995	4069		[[package]]
	4070	+	name = "zerocopy"
	4071	+	version = "0.8.62"
	4072	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	4073	+	checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d"
	4074	+	dependencies = [
	4075	+	"zerocopy-derive",
	4076	+	]
	4077	+
	4078	+	[[package]]
	4079	+	name = "zerocopy-derive"
	4080	+	version = "0.8.62"
	4081	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	4082	+	checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a"
	4083	+	dependencies = [
	4084	+	"proc-macro2",
	4085	+	"quote",
	4086	+	"syn 2.0.119",
	4087	+	]
	4088	+
	4089	+	[[package]]
3996	4090		name = "zerofrom"
3997	4091		version = "0.1.8"
3998	4092		source = "registry+https://github.com/rust-lang/crates.io-index"

170	170		sent: Instant,
171	171		/// The problem matchers steps added, which last the whole job.
172	172		pub(crate) matchers: super::matchers::Matchers,
	173	+	/// The number of the last batch sent, 1 for the job's first: each
	174	+	/// batch carries its own, and keeps it when it is sent again, so g1t
	175	+	/// keeps a batch it already has once.
	176	+	chunk: u64,
173	177		/// Everything sent so far, so a test reads the whole log however its
174	178		/// batches fell (`buffered`).
175	179		#[cfg(test)]
⋯
188	192		last: Instant::now(),
189	193		sent: Instant::now(),
190	194		matchers: super::matchers::Matchers::default(),
	195	+	chunk: 0,
191	196		#[cfg(test)]
192	197		flushed: String::new(),
193	198		}
⋯
263	268		let text = std::mem::take(&mut self.buffer);
264	269		#[cfg(test)]
265	270		self.flushed.push_str(&text);
266		−	self.api.report(json!({ "kind": "log", "step": self.step, "text": text }));
	271	+	let report = self.log_report(text);
	272	+	self.api.report(report);
	273	+	}
	274	+
	275	+	/// The next batch of the log, numbered.
	276	+	fn log_report(&mut self, text: String) -> Value {
	277	+	self.chunk += 1;
	278	+	json!({ "kind": "log", "step": self.step, "chunk": self.chunk, "text": text })
267	279		}
268	280
269	281		pub(crate) fn steps(&self, names: &[String]) {
⋯
416	428		Log::new(Api::offline(), mask::all_variants(secrets.iter().copied()))
417	429		}
418	430
	431	+	/// A server that answers the first `fail` requests with a 500 and the
	432	+	/// rest with `{}`, and hands back each request's body.
	433	+	fn flaky_server(fail: usize, requests: usize) -> (String, std::thread::JoinHandle<Vec<Value>>) {
	434	+	use std::io::{BufRead, BufReader, Read, Write};
	435	+	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	436	+	let base = format!("http://{}", listener.local_addr().unwrap());
	437	+	let serving = std::thread::spawn(move \|\| {
	438	+	let mut bodies = Vec::new();
	439	+	for answered in 0..requests {
	440	+	let (stream, _) = listener.accept().unwrap();
	441	+	let mut reader = BufReader::new(stream);
	442	+	let mut length = 0;
	443	+	loop {
	444	+	let mut header = String::new();
	445	+	reader.read_line(&mut header).unwrap();
	446	+	if header.trim().is_empty() {
	447	+	break;
	448	+	}
	449	+	if let Some((name, value)) = header.split_once(':')
	450	+	&& name.eq_ignore_ascii_case("content-length")
	451	+	{
	452	+	length = value.trim().parse().unwrap();
	453	+	}
	454	+	}
	455	+	let mut body = vec![0; length];
	456	+	reader.read_exact(&mut body).unwrap();
	457	+	bodies.push(serde_json::from_slice::<Value>(&body).unwrap());
	458	+	let status = if answered < fail { "500 Internal Server Error" } else { "200 OK" };
	459	+	write!(reader.get_mut(), "HTTP/1.1 {status}\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{{}}").unwrap();
	460	+	}
	461	+	bodies
	462	+	});
	463	+	(base, serving)
	464	+	}
	465	+
	466	+	#[test]
	467	+	fn each_batch_of_the_log_keeps_its_number_when_sent_again() {
	468	+	let (base, serving) = flaky_server(1, 3);
	469	+	let mut log = Log::new(Api { base, job: "job_1".into(), token: "t".into() }, Vec::new());
	470	+	log.line("one");
	471	+	log.flush();
	472	+	log.line("two");
	473	+	log.flush();
	474	+	log.flush(); // Nothing waits: no batch, and no number used.
	475	+	let bodies = serving.join().unwrap();
	476	+	let sent: Vec<(u64, String)> = bodies
	477	+	.iter()
	478	+	.map(\|body\| (body["report"]["chunk"].as_u64().unwrap(), unstamped(body["report"]["text"].as_str().unwrap())))
	479	+	.collect();
	480	+	// The first batch failed and was sent again with its number; the
	481	+	// next is the job's second.
	482	+	assert_eq!(sent, [(1, "one\n".into()), (1, "one\n".into()), (2, "two\n".into())]);
	483	+	assert_eq!(bodies[0], bodies[1], "a batch sent again is the same batch");
	484	+	}
	485	+
419	486		#[test]
420	487		fn outputs_holding_a_secret_are_left_out() {
421	488		let mut log = log(&["s3cr3t-token"]);

30	30		const MAX_CHUNK_BYTES: usize = 256 * 1024;
31	31		const MAX_ANNOTATIONS: usize = 50;
32	32		/// A batch of a job's log, under the next `seq` of that job's, worked out
33		−	/// in the same statement (SQLite runs one write at a time).
34		−	const LOG_INSERT: &str =
35		−	"INSERT INTO logs (job_id, seq, step, text) SELECT ?1, COALESCE(MAX(seq), 0) + 1, ?2, ?3 FROM logs WHERE job_id = ?1";
	33	+	/// in the same statement (SQLite runs one write at a time). `?4` is the
	34	+	/// runner's number for the batch (null from runners that send none): a
	35	+	/// batch sent again, after an answer the runner did not hear, has the
	36	+	/// number of one already kept, and is not kept twice (migration 0016).
	37	+	const LOG_INSERT: &str = "INSERT INTO logs (job_id, seq, step, text, chunk)
	38	+	SELECT ?1, COALESCE(MAX(seq), 0) + 1, ?2, ?3, ?4 FROM logs WHERE job_id = ?1
	39	+	ON CONFLICT (job_id, chunk) DO NOTHING";
36	40		/// The most one step's job summary keeps, in bytes, and how many steps of a
37	41		/// job may have one, as on GitHub.
38	42		pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
⋯
2474	2478		// and lose a chunk to the primary key.
2475	2479		self.db
2476	2480		.prepare(LOG_INSERT)
2477		−	.bind(&[job.id.as_str().into(), (report["step"].as_u64().unwrap_or(0) as u32).into(), text.into()])?
	2481	+	.bind(&[
	2482	+	job.id.as_str().into(),
	2483	+	(report["step"].as_u64().unwrap_or(0) as u32).into(),
	2484	+	text.into(),
	2485	+	report["chunk"].as_u64().map_or(worker::wasm_bindgen::JsValue::NULL, \|n\| (n as f64).into()),
	2486	+	])?
2478	2487		.run()
2479	2488		.await?;
2480	2489		}
⋯
2639	2648
2640	2649
2641	2650		#[cfg(test)]
	2651	+	mod log_chunks {
	2652	+	use rusqlite::{Connection, params};
	2653	+
	2654	+	use super::LOG_INSERT;
	2655	+
	2656	+	/// The logs table as the migrations leave it.
	2657	+	fn database() -> Connection {
	2658	+	let db = Connection::open_in_memory().unwrap();
	2659	+	for name in ["0001_init.sql", "0016_log_chunks.sql"] {
	2660	+	let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
	2661	+	db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
	2662	+	}
	2663	+	db
	2664	+	}
	2665	+
	2666	+	fn log(db: &Connection, job: &str) -> Vec<(i64, i64, String)> {
	2667	+	let mut rows = db.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq").unwrap();
	2668	+	rows.query_map([job], \|row\| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect()
	2669	+	}
	2670	+
	2671	+	#[test]
	2672	+	fn a_batch_sent_again_is_kept_once() {
	2673	+	let db = database();
	2674	+	let insert = \|job: &str, step: i64, text: &str, chunk: Option<i64>\| db.execute(LOG_INSERT, params![job, step, text, chunk]).unwrap();
	2675	+	assert_eq!(insert("job_1", 0, "one\n", Some(1)), 1);
	2676	+	// The runner did not hear the answer and sent the batch again.
	2677	+	assert_eq!(insert("job_1", 0, "one\n", Some(1)), 0);
	2678	+	assert_eq!(insert("job_1", 1, "two\n", Some(2)), 1);
	2679	+	// Another job numbers its batches from 1 too.
	2680	+	assert_eq!(insert("job_2", 0, "other\n", Some(1)), 1);
	2681	+	// A runner that numbers nothing has every batch kept, as before.
	2682	+	assert_eq!(insert("job_3", 0, "a\n", None), 1);
	2683	+	assert_eq!(insert("job_3", 0, "a\n", None), 1);
	2684	+	assert_eq!(log(&db, "job_1"), [(1, 0, "one\n".into()), (2, 1, "two\n".into())]);
	2685	+	assert_eq!(log(&db, "job_2"), [(1, 0, "other\n".into())]);
	2686	+	assert_eq!(log(&db, "job_3").len(), 2);
	2687	+	// A job run again moves its log aside (`{id}.{attempt}`), and the
	2688	+	// new attempt's runner numbers from 1 again.
	2689	+	db.execute("UPDATE logs SET job_id = job_id \|\| '.1' WHERE job_id = 'job_1'", []).unwrap();
	2690	+	assert_eq!(insert("job_1", 0, "again\n", Some(1)), 1);
	2691	+	assert_eq!(log(&db, "job_1"), [(1, 0, "again\n".into())]);
	2692	+	assert_eq!(log(&db, "job_1.1").len(), 2);
	2693	+	}
	2694	+	}
	2695	+
	2696	+	#[cfg(test)]
2642	2697		mod stopping {
2643	2698		use super::stops_runs;
2644	2699		use g1t_contracts::events::Event;

1166	1166		logs** shows or hides the times; your browser remembers which.
1167	1167		- A download (the job's `log.txt` and the run's zip) has the time at the
1168	1168		start of every line, as GitHub's downloaded logs do.
	1169	+	- The runner numbers each batch of the log it sends. When it sends a
	1170	+	batch again because g1t's answer didn't reach it, g1t keeps the batch
	1171	+	once, so a line never shows twice.
1169	1172		- If the page cannot reach g1t for a moment, it says **Reconnecting to the
1170	1173		log** and tries again, waiting longer each time up to half a minute.
1171	1174		A tab you are not looking at stops asking, and catches up when you come

1	+	-- The runner's own number for each batch of a job's log (1 for its first),
2	+	-- which a batch keeps when the runner sends it again after an answer it did
3	+	-- not hear: a batch g1t already has is not kept twice. Null for runners that
4	+	-- do not number their batches; nulls never collide in a unique index.
5	+	ALTER TABLE logs ADD COLUMN chunk INTEGER;
6	+	CREATE UNIQUE INDEX logs_by_chunk ON logs (job_id, chunk);

20	20		serde_json = { workspace = true, features = ["preserve_order"] }
21	21		worker.workspace = true
22	22
	23	+	[dev-dependencies]
	24	+	rusqlite = { version = "0.32", features = ["bundled"] }
	25	+
23	26		# wasm-opt at -O1: about the same gzipped size as -O in a tenth of the
24	27		# time.
25	28		[package.metadata.wasm-pack.profile.release]