isb_daemon/daemon/
builds.rs

1//! The `build_*` and `registry_*` tools: builds run in the daemon (it alone
2//! pushes to the local registry), and their logs are read back in pieces,
3//! so a CLI or an agent follows a build that outlives any one request.
4
5use std::collections::{BTreeMap, VecDeque};
6use std::path::PathBuf;
7use std::sync::{Arc, Condvar, Mutex};
8use std::time::{Duration, Instant};
9
10use serde::Deserialize;
11use serde_json::{Value, json};
12
13use crate::build::{BuildRequest, Builder, BuiltImage};
14use crate::client::Client;
15use crate::error::{Error, Result};
16use crate::org::OrgId;
17use crate::server::{Caller, Registry, Tool};
18use crate::stack::Controller;
19
20use super::policy::RemotePolicy;
21
22/// Lines kept per build; older ones are dropped (and counted).
23const MAX_LINES: usize = 20_000;
24/// Finished builds kept for their logs.
25const KEEP_FINISHED: usize = 50;
26
27struct Job {
28    org: OrgId,
29    app: String,
30    tag: String,
31    started: u64,
32    state: Mutex<JobState>,
33    changed: Condvar,
34}
35
36#[derive(Default)]
37struct JobState {
38    /// Line numbers start at `first`.
39    first: usize,
40    lines: VecDeque<String>,
41    result: Option<std::result::Result<BuiltImage, String>>,
42    finished: Option<u64>,
43}
44
45impl Job {
46    fn push(&self, line: &str) {
47        let mut s = self.state.lock().unwrap();
48        s.lines.push_back(line.to_string());
49        if s.lines.len() > MAX_LINES {
50            s.lines.pop_front();
51            s.first += 1;
52        }
53        self.changed.notify_all();
54    }
55
56    fn summary(&self, id: &str) -> Value {
57        let s = self.state.lock().unwrap();
58        let mut v = json!({
59            "id": id, "org": self.org, "app": self.app, "tag": self.tag,
60            "started_at": self.started, "finished_at": s.finished,
61            "state": match &s.result { None => "running", Some(Ok(_)) => "succeeded", Some(Err(_)) => "failed" },
62        });
63        match &s.result {
64            Some(Ok(b)) => {
65                v["image"] = json!(b.image);
66                v["digest"] = json!(b.digest);
67            }
68            Some(Err(e)) => v["error"] = json!(e),
69            None => {}
70        }
71        v
72    }
73}
74
75#[derive(Default)]
76struct Jobs {
77    by_id: Mutex<BTreeMap<String, Arc<Job>>>,
78}
79
80impl Jobs {
81    fn get(&self, id: &str, org: &OrgId) -> Result<Arc<Job>> {
82        self.by_id
83            .lock()
84            .unwrap()
85            .get(id)
86            .filter(|j| j.org == *org)
87            .cloned()
88            .ok_or_else(|| Error::NotFound(format!("build {id} in org {org}")))
89    }
90
91    fn add(&self, id: String, j: Arc<Job>) {
92        let mut m = self.by_id.lock().unwrap();
93        m.insert(id, j);
94        // Drop the oldest finished builds beyond the limit.
95        let finished: Vec<String> = m
96            .iter()
97            .filter(|(_, j)| j.state.lock().unwrap().result.is_some())
98            .map(|(k, _)| k.clone())
99            .collect();
100        if finished.len() > KEEP_FINISHED {
101            for k in finished.iter().take(finished.len() - KEEP_FINISHED) {
102                m.remove(k);
103            }
104        }
105    }
106}
107
108/// What the build tools need from the daemon.
109pub struct Ctx {
110    pub client: Client,
111    pub policy: RemotePolicy,
112    pub ctl: Controller,
113}
114
115fn args<T: serde::de::DeserializeOwned>(v: Value) -> Result<T> {
116    serde_json::from_value(v).map_err(|e| Error::invalid(format!("bad arguments: {e}")))
117}
118
119fn org_of(o: &Option<String>) -> Result<OrgId> {
120    match o {
121        Some(o) => OrgId::new(o.clone()),
122        None => Ok(OrgId::default_org()),
123    }
124}
125
126fn obj(mut props: Value, required: &[&str]) -> Value {
127    props["org"] =
128        json!({"type": "string", "description": "The org to act in (default: default)."});
129    json!({"type": "object", "properties": props, "required": required, "additionalProperties": false})
130}
131
132/// A build id: time-ordered and unguessable enough for a log handle (the
133/// org check is what guards it).
134fn new_build_id() -> String {
135    format!(
136        "b{}-{}{}",
137        crate::stack::now_secs(),
138        crate::stack::new_id(),
139        crate::stack::new_id()
140    )
141}
142
143/// The build tools and the registry tools.
144#[expect(
145    clippy::too_many_lines,
146    reason = "predates the lint ratchet; split it when next changed"
147)]
148pub fn register(r: &mut Registry, ctx: Ctx) -> Result<()> {
149    // Builds that were running when the last daemon stopped left their
150    // sandboxes behind, holding quota and build cache volumes.
151    let c = ctx.client.clone();
152    let _ = std::thread::Builder::new()
153        .name("isb-reap-builds".into())
154        .spawn(move || {
155            crate::build::reap_stale_everywhere(&c, &mut |l| eprintln!("isb serve: {l}"));
156        });
157    let ctx = Arc::new(ctx);
158    let jobs = Arc::new(Jobs::default());
159    let ro = json!({"readOnlyHint": true, "openWorldHint": false});
160    let write = json!({"destructiveHint": false, "openWorldHint": true});
161    let destructive = json!({"destructiveHint": true, "openWorldHint": false});
162
163    let (c2, j2) = (ctx.clone(), jobs.clone());
164    r.register(
165        Tool::new(
166            "build_run",
167            "Start a build of a source directory into an image in the org's local registry. It runs in a fresh sandbox in the org (a VM with untrusted=true), never on the host, and is pushed as <org>/<app>:<tag>. Returns an id at once: follow it with build_logs. The result's `image` (registry:APP:TAG@DIGEST) goes in a compose `image:`.",
168            obj(
169                json!({
170                    "app": {"type": "string", "description": "The app: names the repository (<org>/<app>) and the build cache. [a-z0-9][a-z0-9._-]*, up to 40 characters."},
171                    "context": {"type": "string", "description": "Absolute path of the source directory on the host (read, never written). Remote callers: under a --bind-root."},
172                    "subdir": {"type": "string", "description": "Build from this subdirectory of the context."},
173                    "builder": {"type": "string", "enum": ["railpack", "nixpacks", "dockerfile"], "description": "How to build (default railpack, or dockerfile when `dockerfile` is set)."},
174                    "dockerfile": {"type": "string", "description": "Dockerfile path relative to the context (default Dockerfile)."},
175                    "target": {"type": "string", "description": "Dockerfile stage to build."},
176                    "args": {"type": "object", "additionalProperties": {"type": "string"}, "description": "Build arguments (Dockerfile ARGs; environment for railpack and nixpacks)."},
177                    "tag": {"type": "string", "description": "The tag to push (default latest)."},
178                    "untrusted": {"type": "boolean", "description": "Build in a VM (its own kernel)."},
179                    "timeout": {"type": "string", "description": "Longest the build may take, e.g. 30m (the default)."}
180                }),
181                &["app", "context"],
182            ),
183            move |a, c| build_run(&c2, &j2, a, c),
184        )
185        .title("Start a build")
186        .annotations(write.clone()),
187    )?;
188
189    let j2 = jobs.clone();
190    r.register(
191        Tool::new(
192            "build_logs",
193            "A build's state and its log lines from `since` (a line number; 0 for all). `wait` (seconds, at most 30) holds the call until there are new lines or the build ends. Done when `state` is succeeded (then `image` and `digest`) or failed (`error`).",
194            obj(
195                json!({
196                    "id": {"type": "string"},
197                    "since": {"type": "integer", "minimum": 0},
198                    "wait": {"type": "integer", "minimum": 0, "maximum": 30}
199                }),
200                &["id"],
201            ),
202            move |a, _c| {
203                #[derive(Deserialize)]
204                #[serde(deny_unknown_fields)]
205                struct A {
206                    #[serde(default)]
207                    org: Option<String>,
208                    id: String,
209                    #[serde(default)]
210                    since: usize,
211                    #[serde(default)]
212                    wait: u64,
213                }
214                let a: A = args(a)?;
215                let org = org_of(&a.org)?;
216                let job = j2.get(&a.id, &org)?;
217                let until = Instant::now() + Duration::from_secs(a.wait.min(30));
218                let mut s = job.state.lock().unwrap();
219                while s.result.is_none() && s.first + s.lines.len() <= a.since {
220                    let left = until.saturating_duration_since(Instant::now());
221                    if left.is_zero() {
222                        break;
223                    }
224                    s = job.changed.wait_timeout(s, left).unwrap().0;
225                }
226                let from = a.since.max(s.first);
227                let lines: Vec<&String> = s.lines.iter().skip(from - s.first).collect();
228                let next = s.first + s.lines.len();
229                let dropped = a.since < s.first;
230                let lines = json!(lines);
231                drop(s);
232                let mut v = job.summary(&a.id);
233                v["lines"] = lines;
234                v["next"] = json!(next);
235                if dropped {
236                    v["truncated"] = json!(true);
237                }
238                Ok(v)
239            },
240        )
241        .title("Follow a build")
242        .annotations(ro.clone()),
243    )?;
244
245    let j2 = jobs.clone();
246    r.register(
247        Tool::new(
248            "build_list",
249            "The org's recent builds (running and finished), newest first.",
250            obj(json!({}), &[]),
251            move |a, _c| {
252                #[derive(Deserialize)]
253                #[serde(deny_unknown_fields)]
254                struct A {
255                    #[serde(default)]
256                    org: Option<String>,
257                }
258                let a: A = args(a)?;
259                let org = org_of(&a.org)?;
260                let m = j2.by_id.lock().unwrap();
261                let mut out: Vec<Value> = m
262                    .iter()
263                    .filter(|(_, j)| j.org == org)
264                    .map(|(id, j)| j.summary(id))
265                    .collect();
266                out.reverse();
267                Ok(json!({"builds": out}))
268            },
269        )
270        .title("List builds")
271        .annotations(ro.clone()),
272    )?;
273
274    let c2 = ctx.clone();
275    r.register(
276        Tool::new(
277            "registry_list",
278            "The org's images in the local registry: each app's tags with their digests and push times, newest first. Run one with image: registry:APP:TAG (or @DIGEST) in a compose file.",
279            obj(json!({}), &[]),
280            move |a, _c| {
281                #[derive(Deserialize)]
282                #[serde(deny_unknown_fields)]
283                struct A {
284                    #[serde(default)]
285                    org: Option<String>,
286                }
287                let a: A = args(a)?;
288                let org = org_of(&a.org)?;
289                let reg = crate::registry::Registry::shared(&c2.client)?;
290                Ok(json!({
291                    "registry": crate::registry::status_json(&reg),
292                    "repositories": reg.list(Some(&org))?,
293                }))
294            },
295        )
296        .title("List images")
297        .annotations(ro.clone()),
298    )?;
299
300    let c2 = ctx.clone();
301    r.register(
302        Tool::new(
303            "registry_gc",
304            "Retention for the whole local registry (platform admins): per repository keep the newest `keep` tags (default 10) and every image a deployed stack runs or would roll back to, delete the rest, and reclaim their storage. dry_run reports only.",
305            obj(
306                json!({
307                    "keep": {"type": "integer", "minimum": 0},
308                    "dry_run": {"type": "boolean"}
309                }),
310                &[],
311            ),
312            move |a, _c| {
313                #[derive(Deserialize)]
314                #[serde(deny_unknown_fields)]
315                struct A {
316                    #[serde(default)]
317                    #[allow(dead_code)]
318                    org: Option<String>,
319                    keep: Option<usize>,
320                    #[serde(default)]
321                    dry_run: bool,
322                }
323                let a: A = args(a)?;
324                let reg = crate::registry::Registry::shared(&c2.client)?;
325                let protected = crate::registry::protected_by(&c2.ctl.definitions());
326                let mut lines = Vec::new();
327                let rep = reg.gc(
328                    a.keep.unwrap_or(crate::registry::DEFAULT_KEEP),
329                    &protected,
330                    a.dry_run,
331                    &mut |l| lines.push(l.to_string()),
332                )?;
333                let mut v = serde_json::to_value(rep)?;
334                v["log"] = json!(lines);
335                Ok(v)
336            },
337        )
338        .title("Registry retention")
339        .annotations(destructive),
340    )?;
341    Ok(())
342}
343
344#[expect(
345    clippy::too_many_lines,
346    reason = "predates the lint ratchet; split it when next changed"
347)]
348fn build_run(ctx: &Ctx, jobs: &Arc<Jobs>, a: Value, c: &Caller) -> Result<Value> {
349    #[derive(Deserialize)]
350    #[serde(deny_unknown_fields)]
351    struct A {
352        #[serde(default)]
353        org: Option<String>,
354        app: String,
355        context: PathBuf,
356        #[serde(default)]
357        subdir: Option<String>,
358        #[serde(default)]
359        builder: Option<String>,
360        #[serde(default)]
361        dockerfile: Option<String>,
362        #[serde(default)]
363        target: Option<String>,
364        #[serde(default)]
365        args: BTreeMap<String, String>,
366        #[serde(default)]
367        tag: Option<String>,
368        #[serde(default)]
369        untrusted: bool,
370        #[serde(default)]
371        timeout: Option<String>,
372    }
373    let a: A = args(a)?;
374    let org = org_of(&a.org)?;
375    if !a.context.is_absolute() {
376        return Err(Error::invalid("context must be an absolute path"));
377    }
378    if !c.is_trusted() {
379        ctx.policy.check_base_dir(&a.context)?;
380    }
381    let builder = match (a.builder.as_deref(), &a.dockerfile) {
382        (None, Some(_)) | (Some("dockerfile"), _) => Builder::Dockerfile {
383            path: a.dockerfile.clone().unwrap_or_else(|| "Dockerfile".into()),
384            target: a.target.clone(),
385        },
386        (None, None) | (Some("railpack"), None) => Builder::Railpack,
387        (Some("nixpacks"), None) => Builder::Nixpacks,
388        (Some("buildpacks"), None) => Builder::Buildpacks { builder: None },
389        (Some(b), _) => {
390            return Err(Error::invalid(format!(
391                "builder {b:?}: railpack, nixpacks or dockerfile (a dockerfile path needs the dockerfile builder)"
392            )));
393        }
394    };
395    let req = BuildRequest {
396        org: org.clone(),
397        app: a.app.clone(),
398        context: a.context.clone(),
399        subdir: a.subdir.clone(),
400        builder,
401        args: a.args.into_iter().collect(),
402        tag: a.tag.clone().unwrap_or_else(|| "latest".into()),
403        untrusted: a.untrusted,
404        cache: None,
405    };
406    let mut opts = crate::build::BuildOptions::default();
407    if let Some(t) = &a.timeout {
408        opts.timeout = crate::flex::parse_duration(t).map_err(Error::invalid)?;
409    }
410    // Refuse what cannot start before answering with an id.
411    crate::registry::Registry::shared(&ctx.client)?;
412    let id = new_build_id();
413    let job = Arc::new(Job {
414        org: org.clone(),
415        app: req.app.clone(),
416        tag: req.tag.clone(),
417        started: crate::stack::now_secs(),
418        state: Mutex::new(JobState::default()),
419        changed: Condvar::new(),
420    });
421    jobs.add(id.clone(), job.clone());
422    let client = ctx.client.clone();
423    let ctl = ctx.ctl.clone();
424    let who = c.to_string();
425    let stack = crate::stack::qualified(&org, &format!("build:{}", req.app));
426    ctl.note(
427        "info",
428        &stack,
429        format!("build {id} of {}:{} started by {who}", req.app, req.tag),
430    );
431    std::thread::Builder::new()
432        .name(format!("isb-build-{id}"))
433        .spawn(move || {
434            let r = crate::build::run_with(&client, &req, &opts, &mut |l| job.push(l));
435            let (level, msg) = match &r {
436                Ok(b) => (
437                    "info",
438                    format!("build of {}:{} succeeded: {}", req.app, req.tag, b.image),
439                ),
440                Err(e) => (
441                    "error",
442                    format!("build of {}:{} failed: {e}", req.app, req.tag),
443                ),
444            };
445            if let Err(e) = &r {
446                job.push(&format!("error: {e}"));
447            }
448            ctl.note(level, &stack, msg);
449            let mut s = job.state.lock().unwrap();
450            s.result = Some(r.map_err(|e| e.to_string()));
451            s.finished = Some(crate::stack::now_secs());
452            job.changed.notify_all();
453        })
454        .map_err(Error::Io)?;
455    Ok(json!({"id": id, "org": org, "app": a.app, "tag": a.tag.unwrap_or_else(|| "latest".into())}))
456}
isb_daemon/daemon/builds.rs

isb_daemon/daemon/
builds.rs
