Skip to main content

max / makenotwork

11.0 KB · 361 lines History Blame Raw
1 //! Build pipeline queries: configs, builds, status updates.
2
3 use sqlx::PgPool;
4
5 use super::models::{DbBuild, DbBuildConfig};
6 use super::{BuildConfigId, BuildId, BuildStatus, OtaReleaseId, SyncAppId};
7 use crate::error::Result;
8
9 // ── Build configs ──
10
11 /// Create a new build config for a sync app.
12 #[tracing::instrument(skip_all)]
13 pub(crate) async fn create_build_config(
14 pool: &PgPool,
15 app_id: SyncAppId,
16 repo_id: super::GitRepoId,
17 build_command: &str,
18 artifact_path: &str,
19 signing_key_path: &str,
20 targets: &[String],
21 ) -> Result<DbBuildConfig> {
22 let config = sqlx::query_as::<_, DbBuildConfig>(
23 r"
24 INSERT INTO ota_build_configs (app_id, repo_id, build_command, artifact_path, signing_key_path, targets)
25 VALUES ($1, $2, $3, $4, $5, $6)
26 RETURNING *
27 ",
28 )
29 .bind(app_id)
30 .bind(repo_id)
31 .bind(build_command)
32 .bind(artifact_path)
33 .bind(signing_key_path)
34 .bind(targets)
35 .fetch_one(pool)
36 .await?;
37
38 Ok(config)
39 }
40
41 /// Get the build config for a sync app.
42 #[tracing::instrument(skip_all)]
43 pub(crate) async fn get_build_config_by_app(
44 pool: &PgPool,
45 app_id: SyncAppId,
46 ) -> Result<Option<DbBuildConfig>> {
47 let config =
48 sqlx::query_as::<_, DbBuildConfig>("SELECT * FROM ota_build_configs WHERE app_id = $1")
49 .bind(app_id)
50 .fetch_optional(pool)
51 .await?;
52
53 Ok(config)
54 }
55
56 /// Get an enabled build config by repo ID (for hook trigger lookup).
57 #[tracing::instrument(skip_all)]
58 pub(crate) async fn get_build_config_by_repo(
59 pool: &PgPool,
60 repo_id: super::GitRepoId,
61 ) -> Result<Option<DbBuildConfig>> {
62 let config = sqlx::query_as::<_, DbBuildConfig>(
63 "SELECT * FROM ota_build_configs WHERE repo_id = $1 AND enabled = true",
64 )
65 .bind(repo_id)
66 .fetch_optional(pool)
67 .await?;
68
69 Ok(config)
70 }
71
72 #[tracing::instrument(skip_all)]
73 pub(crate) async fn update_build_config(
74 pool: &PgPool,
75 config_id: BuildConfigId,
76 build_command: &str,
77 artifact_path: &str,
78 signing_key_path: &str,
79 targets: &[String],
80 enabled: bool,
81 ) -> Result<DbBuildConfig> {
82 let config = sqlx::query_as::<_, DbBuildConfig>(
83 r"
84 UPDATE ota_build_configs
85 SET build_command = $2, artifact_path = $3, signing_key_path = $4,
86 targets = $5, enabled = $6, updated_at = now()
87 WHERE id = $1
88 RETURNING *
89 ",
90 )
91 .bind(config_id)
92 .bind(build_command)
93 .bind(artifact_path)
94 .bind(signing_key_path)
95 .bind(targets)
96 .bind(enabled)
97 .fetch_one(pool)
98 .await?;
99
100 Ok(config)
101 }
102
103 /// Delete a build config (cascades to builds).
104 #[tracing::instrument(skip_all)]
105 pub(crate) async fn delete_build_config(pool: &PgPool, config_id: BuildConfigId) -> Result<bool> {
106 let result = sqlx::query("DELETE FROM ota_build_configs WHERE id = $1")
107 .bind(config_id)
108 .execute(pool)
109 .await?;
110
111 Ok(result.rows_affected() > 0)
112 }
113
114 // ── Builds ──
115
116 /// Create a new pending build.
117 #[tracing::instrument(skip_all)]
118 pub(crate) async fn create_build(
119 pool: &PgPool,
120 config_id: BuildConfigId,
121 app_id: SyncAppId,
122 version: &str,
123 tag: &str,
124 triggered_by: &str,
125 ) -> Result<DbBuild> {
126 let build = sqlx::query_as::<_, DbBuild>(
127 r"
128 INSERT INTO ota_builds (config_id, app_id, version, tag, triggered_by)
129 VALUES ($1, $2, $3, $4, $5)
130 RETURNING *
131 ",
132 )
133 .bind(config_id)
134 .bind(app_id)
135 .bind(version)
136 .bind(tag)
137 .bind(triggered_by)
138 .fetch_one(pool)
139 .await?;
140
141 Ok(build)
142 }
143
144 /// Get a build by ID.
145 #[tracing::instrument(skip_all)]
146 pub(crate) async fn get_build(pool: &PgPool, build_id: BuildId) -> Result<Option<DbBuild>> {
147 let build = sqlx::query_as::<_, DbBuild>("SELECT * FROM ota_builds WHERE id = $1")
148 .bind(build_id)
149 .fetch_optional(pool)
150 .await?;
151
152 Ok(build)
153 }
154
155 /// Return the current `octet_length` of a build's log column, plus whether
156 /// it already ends with `marker` (server-side string compare so the full
157 /// log column never travels back over the wire).
158 ///
159 /// Used by `append_log_bounded` to make the cap check without pulling the
160 /// 5 MiB log row into memory on every line.
161 #[tracing::instrument(skip_all)]
162 pub(crate) async fn get_build_log_size(
163 pool: &PgPool,
164 build_id: BuildId,
165 marker: &str,
166 ) -> Result<Option<(i64, bool)>> {
167 let row: Option<(i64, bool)> = sqlx::query_as(
168 "SELECT octet_length(log)::BIGINT, right(log, char_length($2)) = $2 \
169 FROM ota_builds WHERE id = $1",
170 )
171 .bind(build_id)
172 .bind(marker)
173 .fetch_optional(pool)
174 .await?;
175
176 Ok(row)
177 }
178
179 /// List builds for an app, newest first.
180 #[tracing::instrument(skip_all)]
181 pub(crate) async fn list_builds_by_app(
182 pool: &PgPool,
183 app_id: SyncAppId,
184 limit: i64,
185 ) -> Result<Vec<DbBuild>> {
186 let builds = sqlx::query_as::<_, DbBuild>(
187 "SELECT * FROM ota_builds WHERE app_id = $1 ORDER BY created_at DESC LIMIT $2",
188 )
189 .bind(app_id)
190 .bind(limit)
191 .fetch_all(pool)
192 .await?;
193
194 Ok(builds)
195 }
196
197 /// Check if a config has any active (pending or running) build.
198 #[tracing::instrument(skip_all)]
199 pub(crate) async fn has_active_build(pool: &PgPool, config_id: BuildConfigId) -> Result<bool> {
200 let count: (i64,) = sqlx::query_as(
201 "SELECT COUNT(*) FROM ota_builds WHERE config_id = $1 AND status IN ('pending', 'running')",
202 )
203 .bind(config_id)
204 .fetch_one(pool)
205 .await?;
206
207 Ok(count.0 > 0)
208 }
209
210 /// Update build status with conditional timestamps.
211 #[tracing::instrument(skip_all)]
212 pub(crate) async fn update_build_status(
213 pool: &PgPool,
214 build_id: BuildId,
215 status: BuildStatus,
216 error_message: Option<&str>,
217 ) -> Result<()> {
218 match status {
219 BuildStatus::Running => {
220 sqlx::query("UPDATE ota_builds SET status = $2, started_at = now() WHERE id = $1")
221 .bind(build_id)
222 .bind(status)
223 .execute(pool)
224 .await?;
225 }
226 BuildStatus::Succeeded | BuildStatus::Failed | BuildStatus::Cancelled => {
227 // Gate on a non-terminal source status so the stale-build reaper
228 // (`fail_stale_running_builds`) and a real terminal write can't
229 // race: if the reaper just flipped the row to 'failed', the
230 // successful builder's write must no-op rather than clobber it.
231 // Pending IS a legitimate source, cancelling a build that never
232 // started must still transition pending → cancelled.
233 let result = sqlx::query(
234 "UPDATE ota_builds SET status = $2, finished_at = now(), error_message = $3 WHERE id = $1 AND status IN ('pending', 'running')",
235 )
236 .bind(build_id)
237 .bind(status)
238 .bind(error_message)
239 .execute(pool)
240 .await?;
241 if result.rows_affected() == 0 {
242 tracing::warn!(build_id = %build_id, target_status = ?status,
243 "build status terminal write skipped, row already terminal (likely reaper-set)");
244 }
245 }
246 BuildStatus::Pending => {
247 sqlx::query("UPDATE ota_builds SET status = $2 WHERE id = $1")
248 .bind(build_id)
249 .bind(status)
250 .execute(pool)
251 .await?;
252 }
253 }
254
255 Ok(())
256 }
257
258 /// Atomically claim the oldest pending build if no build is currently running.
259 ///
260 /// Sets status to 'running' and started_at in a single UPDATE with a subquery,
261 /// eliminating the TOCTOU race between checking for running builds and fetching
262 /// a pending one. `FOR UPDATE SKIP LOCKED` means concurrent callers never block
263 ///; the loser gets no row.
264 #[tracing::instrument(skip_all)]
265 pub(crate) async fn claim_pending_build(pool: &PgPool) -> Result<Option<DbBuild>> {
266 let result = sqlx::query_as::<_, DbBuild>(
267 r"
268 UPDATE ota_builds
269 SET status = 'running', started_at = now()
270 WHERE id = (
271 SELECT id FROM ota_builds
272 WHERE status = 'pending'
273 AND NOT EXISTS (SELECT 1 FROM ota_builds WHERE status = 'running')
274 ORDER BY created_at ASC
275 LIMIT 1
276 FOR UPDATE SKIP LOCKED
277 )
278 RETURNING *
279 ",
280 )
281 .fetch_optional(pool)
282 .await;
283
284 // Multi-replica: the NOT EXISTS subquery races between replicas. The
285 // `ota_builds_single_running` partial unique index is the backstop,
286 // the loser's UPDATE surfaces as a 23505 unique violation, which means
287 // a peer claimed first. Treat as "nothing to claim this tick".
288 match result {
289 Ok(build) => Ok(build),
290 Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("23505") => {
291 tracing::info!(
292 "claim_pending_build lost the running-slot race; another replica claimed"
293 );
294 Ok(None)
295 }
296 Err(e) => Err(e.into()),
297 }
298 }
299
300 /// Mark any builds that have been "running" longer than the timeout as failed.
301 ///
302 /// Returns the number of builds marked as failed.
303 #[tracing::instrument(skip_all)]
304 pub(crate) async fn fail_stale_running_builds(pool: &PgPool, timeout_secs: i64) -> Result<u64> {
305 let result = sqlx::query(
306 r"
307 UPDATE ota_builds
308 SET status = 'failed',
309 finished_at = now(),
310 error_message = 'Build timed out (stale running status)'
311 WHERE status = 'running'
312 AND started_at < now() - make_interval(secs => $1)
313 ",
314 )
315 .bind(timeout_secs as f64)
316 .execute(pool)
317 .await?;
318
319 Ok(result.rows_affected())
320 }
321
322 /// Hard ceiling on the `log` column, enforced at the SQL layer in
323 /// [`append_build_log`] so no caller can grow it without bound. Set above the
324 /// 5 MiB soft cap that `build_runner::append_log_bounded` applies (with its
325 /// `[log truncated]` marker), so this is a pure backstop: the soft cap fires
326 /// first for the normal path, and this guarantees the invariant even for a
327 /// future caller that bypasses the orchestrator.
328 const HARD_LOG_CAP_BYTES: i64 = 8 * 1024 * 1024;
329
330 /// Append a line to the build log, never letting the column exceed
331 /// [`HARD_LOG_CAP_BYTES`]. The `octet_length` guard makes unbounded growth
332 /// unwritable regardless of caller, the bound lives in this one function, not
333 /// in a convention each call site must remember.
334 #[tracing::instrument(skip_all)]
335 pub(crate) async fn append_build_log(pool: &PgPool, build_id: BuildId, line: &str) -> Result<()> {
336 sqlx::query("UPDATE ota_builds SET log = log || $2 WHERE id = $1 AND octet_length(log) < $3")
337 .bind(build_id)
338 .bind(line)
339 .bind(HARD_LOG_CAP_BYTES)
340 .execute(pool)
341 .await?;
342
343 Ok(())
344 }
345
346 /// Set the release ID on a build (after successful artifact upload).
347 #[tracing::instrument(skip_all)]
348 pub(crate) async fn set_build_release(
349 pool: &PgPool,
350 build_id: BuildId,
351 release_id: OtaReleaseId,
352 ) -> Result<()> {
353 sqlx::query("UPDATE ota_builds SET release_id = $2 WHERE id = $1")
354 .bind(build_id)
355 .bind(release_id)
356 .execute(pool)
357 .await?;
358
359 Ok(())
360 }
361