max / makenotwork

4.1 KB · 115 lines History Blame Raw

1	//! Database access. Connection comes from `ops_core::sqlite`; migrations are
2	//! per-crate (the `sqlx::migrate!` path is compile-time, relative to here).
3
4	use anyhow::Result;
5	use sqlx::SqlitePool;
6	use std::path::Path;
7
8	pub use ops_core::sqlite::connect;
9
10	pub async fn open(path: &Path) -> Result<SqlitePool> {
11	let pool = connect(path).await?;
12	sqlx::migrate!("./migrations").run(&pool).await?;
13	Ok(pool)
14	}
15
16	/// Reconcile state left `running` by a previous process. The finalizer and the
17	/// per-target tasks die with the daemon, so any `running` build/target/step row
18	/// at startup is orphaned — nothing will ever stamp it terminal. Mark them all
19	/// `failed` so `/state` reflects reality and a re-run isn't blocked by a ghost.
20	/// Returns the number of build rows reconciled (for logging). Idempotent: a
21	/// clean DB sweeps zero rows.
22	pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
23	let now = chrono::Utc::now().to_rfc3339();
24	let mut tx = pool.begin().await?;
25	sqlx::query("UPDATE step_runs SET status = 'failed', finished_at = ? WHERE status = 'running'")
26	.bind(&now)
27	.execute(&mut *tx)
28	.await?;
29	sqlx::query(
30	"UPDATE target_runs SET status = 'failed', current_step = NULL, \
31	error = COALESCE(error, 'daemon restarted while running'), finished_at = ? \
32	WHERE status = 'running'",
33	)
34	.bind(&now)
35	.execute(&mut *tx)
36	.await?;
37	let builds = sqlx::query(
38	"UPDATE builds SET status = 'failed', finished_at = ? WHERE status = 'running'",
39	)
40	.bind(&now)
41	.execute(&mut *tx)
42	.await?
43	.rows_affected();
44	tx.commit().await?;
45	Ok(builds)
46	}
47
48	#[cfg(test)]
49	mod tests {
50	use super::*;
51
52	#[tokio::test]
53	async fn recovery_marks_orphaned_running_failed_and_leaves_terminal_rows() {
54	let dir = tempfile::tempdir().unwrap();
55	let pool = open(&dir.path().join("t.db")).await.unwrap();
56
57	// One running build with a running target+step, and one already-ok build.
58	let now = chrono::Utc::now().to_rfc3339();
59	let running: i64 = sqlx::query_scalar(
60	"INSERT INTO builds (app, version, status, created_at) VALUES ('demo','0.1.0','running',?) RETURNING id",
61	)
62	.bind(&now).fetch_one(&pool).await.unwrap();
63	let tr: i64 = sqlx::query_scalar(
64	"INSERT INTO target_runs (build_id, app, version, target, status, started_at) \
65	VALUES (?, 'demo','0.1.0','linux/x86_64','running',?) RETURNING id",
66	)
67	.bind(running)
68	.bind(&now)
69	.fetch_one(&pool)
70	.await
71	.unwrap();
72	sqlx::query(
73	"INSERT INTO step_runs (target_run_id, step, status, log_ref, started_at) \
74	VALUES (?, 'build','running','x',?)",
75	)
76	.bind(tr)
77	.bind(&now)
78	.execute(&pool)
79	.await
80	.unwrap();
81	sqlx::query(
82	"INSERT INTO builds (app, version, status, created_at) VALUES ('demo','0.0.9','ok',?)",
83	)
84	.bind(&now)
85	.execute(&pool)
86	.await
87	.unwrap();
88
89	let reconciled = recover_orphaned_running(&pool).await.unwrap();
90	assert_eq!(reconciled, 1, "only the running build is reconciled");
91
92	// The orphaned chain is now failed; the ok build is untouched.
93	let running_left: i64 =
94	sqlx::query_scalar("SELECT COUNT(*) FROM builds WHERE status='running'")
95	.fetch_one(&pool)
96	.await
97	.unwrap();
98	assert_eq!(running_left, 0);
99	let ok_left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds WHERE status='ok'")
100	.fetch_one(&pool)
101	.await
102	.unwrap();
103	assert_eq!(ok_left, 1);
104	let err: Option<String> = sqlx::query_scalar("SELECT error FROM target_runs WHERE id = ?")
105	.bind(tr)
106	.fetch_one(&pool)
107	.await
108	.unwrap();
109	assert_eq!(err.as_deref(), Some("daemon restarted while running"));
110
111	// Idempotent: a second sweep finds nothing.
112	assert_eq!(recover_orphaned_running(&pool).await.unwrap(), 0);
113	}
114	}
115