Skip to main content

runnerd_core/scheduler/engine/
control.rs

1//! Control from the API: the `Control` handle the daemon's API holds (pause flags, which it records in the store so
2//! a restart keeps them, shutdown, a channel of `Command`s) and the engine's side of each command: pause and resume,
3//! of one repository or all, with or without cancelling their runs, reseed, cache clean, config replacement, the
4//! removal of records and the shells opened in them.
5//! Commands run between ticks or while a tick waits for GitHub, so they see the slot state a tick does (ADRs 0019,
6//! 0026). It must not serve HTTP or write the config file: the daemon does, through `Commit`.
7
8use std::collections::{BTreeMap, BTreeSet};
9use std::sync::Mutex;
10use std::time::Duration;
11
12use tokio::sync::{Notify, mpsc, oneshot};
13use tracing::{info, warn};
14
15use super::records::RecordSelection;
16use super::{Engine, Occupant};
17use crate::api;
18use crate::cache;
19use crate::config::Config;
20use crate::docker::{DockerApi, Tty, TtySize};
21use crate::github::GithubApi;
22use crate::isolation;
23use crate::scheduler::SlotRef;
24use crate::store::Store;
25use crate::sync::lock;
26
27/// How long the runner of a run a forced pause cancelled gets to report its job cancelled and exit, before it is
28/// removed: GitHub completes a cancelled run only once its runner reports it, which took 40s in a test.
29pub(super) const CANCEL_GRACE: Duration = Duration::from_secs(90);
30
31/// How the daemon is asked to stop.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
33pub enum Shutdown {
34    /// Keep scheduling.
35    No,
36    /// Start nothing new, let running jobs finish.
37    Graceful,
38    /// Kill running jobs and delete their runners.
39    Force,
40}
41
42/// What a pause, a shutdown or a runner's removal does with the jobs runners are running.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum RunningJobs {
45    /// Leave them running: a pause refuses, a shutdown waits for them, a busy runner stays.
46    Keep,
47    /// Kill them: a pause cancels their runs, a shutdown and a removal kill their runners.
48    Kill,
49}
50
51/// The repositories a pause or a resume applies to.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub enum Target {
54    /// One repository, `owner/name`.
55    Repository(String),
56    /// Every configured repository.
57    All,
58}
59
60/// Why the engine refused or failed a command from the API.
61#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
62pub enum ControlError {
63    /// No such repository, profile or cache.
64    #[error("{0}")]
65    NotFound(String),
66    /// Runners use what the command would change.
67    #[error("{message}")]
68    Busy {
69        /// Names the runners and their jobs.
70        message: String,
71        /// The runners in the way.
72        jobs: Vec<api::RunningJob>,
73    },
74    /// The command conflicts with one already in progress.
75    #[error("{0}")]
76    Conflict(String),
77    /// The command cannot apply to this target.
78    #[error("{0}")]
79    Invalid(String),
80    /// Docker or GitHub failed.
81    #[error("{0}")]
82    Failed(String),
83    /// A pause or resume holds in memory, but the job history could not record it: a restart undoes it.
84    #[error("{0}")]
85    NotRecorded(String),
86    /// The engine stopped.
87    #[error("runnerd is stopping")]
88    Stopped,
89}
90
91/// A command the engine carries out between ticks or while a tick waits for GitHub, with where its answer goes.
92#[derive(Debug)]
93pub enum Command {
94    /// Pause one repository or all; with `RunningJobs::Kill`, cancel their running jobs.
95    Pause {
96        /// What to pause.
97        target: Target,
98        /// Cancel running jobs, or refuse while there are any.
99        jobs: RunningJobs,
100        /// The answer.
101        reply: oneshot::Sender<Result<api::Paused, ControlError>>,
102    },
103    /// Resume one repository or all.
104    Resume {
105        /// What to resume.
106        target: Target,
107        /// The answer.
108        reply: oneshot::Sender<Result<(), ControlError>>,
109    },
110    /// Make slot 1's layers the profile's snapshot once its slots are free.
111    Reseed {
112        /// The profile.
113        profile: String,
114        /// The answer.
115        reply: oneshot::Sender<Result<(), ControlError>>,
116    },
117    /// Remove a slot's volume or a shared cache's, unless a runner uses it.
118    CleanCache {
119        /// The volume.
120        name: String,
121        /// The answer.
122        reply: oneshot::Sender<Result<(), ControlError>>,
123    },
124    /// Write a new, validated config to its file and run with it, with no tick in between.
125    ReplaceConfig {
126        /// The config.
127        config: Box<Config>,
128        /// The `MemoryMax` of the slice it names.
129        memory_max: Option<u64>,
130        /// Writes the file.
131        commit: Commit,
132        /// Answered once the engine runs with it: with a warning from the commit, or why it did not commit.
133        reply: oneshot::Sender<Result<Option<String>, ControlError>>,
134    },
135    /// Remove records of failed jobs.
136    RemoveRecords {
137        /// Which.
138        selection: RecordSelection,
139        /// The answer.
140        reply: oneshot::Sender<Result<api::RemovedRecords, ControlError>>,
141    },
142    /// Start a shell in a job's record, if the budget has room for it.
143    OpenShell {
144        /// The job.
145        job: u64,
146        /// Its terminal's size.
147        size: TtySize,
148        /// The shell's terminal.
149        reply: oneshot::Sender<Result<Tty, ControlError>>,
150    },
151    /// Remove a shell's container: its terminal closed.
152    CloseShell {
153        /// The container's ID.
154        container: String,
155    },
156}
157
158/// What makes a new config durable, run by the engine before it runs with the config: `Ok` once the config is in
159/// place, with a warning if it may not survive a crash; `Err` with nothing changed.
160pub struct Commit(Box<dyn FnOnce() -> Result<Option<String>, String> + Send>);
161
162impl Commit {
163    /// A commit that runs `write`, off the engine's thread.
164    pub fn new(write: impl FnOnce() -> Result<Option<String>, String> + Send + 'static) -> Commit {
165        Commit(Box::new(write))
166    }
167
168    /// Runs it.
169    pub async fn run(self) -> Result<Option<String>, String> {
170        tokio::task::spawn_blocking(self.0)
171            .await
172            .unwrap_or_else(|join_error| Err(join_error.to_string()))
173    }
174}
175
176impl std::fmt::Debug for Commit {
177    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
178        f.write_str("Commit")
179    }
180}
181
182/// The engine's end of `Control`'s commands.
183#[derive(Debug)]
184pub struct Commands(mpsc::UnboundedReceiver<Command>);
185
186impl Commands {
187    /// The next command; `None` once `Control` is gone.
188    pub async fn recv(&mut self) -> Option<Command> {
189        self.0.recv().await
190    }
191}
192
193/// Commands from the API and signals to the loop.
194#[derive(Debug)]
195pub struct Control {
196    // Locked with `sync::lock`: the guarded data stays consistent, as every critical section is a single insert,
197    // remove or read.
198    repositories: Mutex<BTreeSet<String>>,
199    paused: Mutex<BTreeSet<String>>,
200    /// Where `paused` is recorded.
201    store: Store,
202    shutdown: Mutex<Shutdown>,
203    pub(super) wake: Notify,
204    commands: mpsc::UnboundedSender<Command>,
205}
206
207impl Control {
208    /// Controls for the configured repositories, of which those in `paused` start paused, and the commands they
209    /// send the engine. A paused repository the config no longer has is forgotten, in `store` too: one added back
210    /// starts active.
211    pub fn new(
212        repositories: impl IntoIterator<Item = String>,
213        paused: impl IntoIterator<Item = String>,
214        store: Store,
215    ) -> (Control, Commands) {
216        let repositories: BTreeSet<String> = repositories.into_iter().collect();
217        let paused: BTreeSet<String> = paused
218            .into_iter()
219            .filter(|repository| repositories.contains(repository))
220            .collect();
221        // Sent at once; a write that fails leaves rows the next start filters out the same way.
222        drop(store.paused(paused.iter().cloned().collect()));
223        let (commands, receiver) = mpsc::unbounded_channel();
224        let control = Control {
225            repositories: Mutex::new(repositories),
226            paused: Mutex::new(paused),
227            store,
228            shutdown: Mutex::new(Shutdown::No),
229            wake: Notify::new(),
230            commands,
231        };
232        (control, Commands(receiver))
233    }
234
235    /// Whether a shutdown was requested, and how.
236    pub fn mode(&self) -> Shutdown {
237        *lock(&self.shutdown)
238    }
239
240    /// Pauses or resumes one repository or all, whatever they run, and records it in the store before it answers,
241    /// so a restarted daemon starts with the same ones paused. `NotFound` for a repository the config does not have.
242    /// If the record fails, the change holds until a restart and the answer is `NotRecorded`. The status shows it
243    /// at the engine's next publish: the API pauses and resumes through `pause` and `resume`, which publish before
244    /// they answer.
245    pub async fn set_paused(&self, target: &Target, paused: bool) -> Result<(), ControlError> {
246        let recorded = {
247            let repositories = lock(&self.repositories);
248            let names: Vec<&String> = match target {
249                Target::Repository(repository) => {
250                    let Some(name) = repositories.get(repository) else {
251                        return Err(not_configured(repository, repositories.iter()));
252                    };
253                    vec![name]
254                }
255                Target::All => repositories.iter().collect(),
256            };
257            let mut set = lock(&self.paused);
258            for name in names {
259                if paused {
260                    set.insert(name.clone());
261                } else {
262                    set.remove(name);
263                }
264            }
265            // Sent under the lock, so the records change in the order the set does.
266            self.store.paused(set.iter().cloned().collect())
267        };
268        self.wake.notify_one();
269        if recorded.await {
270            return Ok(());
271        }
272        let name = match target {
273            Target::Repository(repository) => repository.as_str(),
274            Target::All => "every repository",
275        };
276        let (done, undone) = if paused {
277            ("paused", "resumes it")
278        } else {
279            ("resumed", "pauses it again")
280        };
281        let message = format!(
282            "{name} is {done} until runnerd restarts, but the job history cannot record it, so a restart {undone}"
283        );
284        warn!("{message}");
285        Err(ControlError::NotRecorded(message))
286    }
287
288    /// Replaces the configured repositories; the removed ones are no longer paused, in the store too.
289    fn set_repositories(&self, repositories: BTreeSet<String>) {
290        let mut configured = lock(&self.repositories);
291        let mut paused = lock(&self.paused);
292        let before = paused.len();
293        paused.retain(|repository| repositories.contains(repository));
294        if paused.len() != before {
295            // Not waited for: a restart forgets a repository the config does not have anyway.
296            drop(self.store.paused(paused.iter().cloned().collect()));
297        }
298        *configured = repositories;
299    }
300
301    /// Asks the loop to stop; a forced request overrides a graceful one, never the other way.
302    pub fn shutdown(&self, jobs: RunningJobs) {
303        let mode = match jobs {
304            RunningJobs::Keep => Shutdown::Graceful,
305            RunningJobs::Kill => Shutdown::Force,
306        };
307        let mut shutdown = lock(&self.shutdown);
308        *shutdown = (*shutdown).max(mode);
309        drop(shutdown);
310        self.wake.notify_one();
311    }
312
313    /// Whether the repository is paused.
314    pub fn is_paused(&self, repository: &str) -> bool {
315        lock(&self.paused).contains(repository)
316    }
317
318    async fn send<T>(
319        &self,
320        command: impl FnOnce(oneshot::Sender<T>) -> Command,
321    ) -> Result<T, ControlError> {
322        let (reply, answer) = oneshot::channel();
323        self.commands
324            .send(command(reply))
325            .map_err(|_| ControlError::Stopped)?;
326        answer.await.map_err(|_| ControlError::Stopped)
327    }
328
329    /// Pauses one repository or all. With `RunningJobs::Keep`, refuses while any of them runs a job, pausing none;
330    /// with `Kill`, cancels the runs of their running jobs on GitHub, whose runners are removed once they report it
331    /// or `CANCEL_GRACE` later, and removes their other runners at once.
332    pub async fn pause(
333        &self,
334        target: Target,
335        jobs: RunningJobs,
336    ) -> Result<api::Paused, ControlError> {
337        self.send(|reply| Command::Pause {
338            target,
339            jobs,
340            reply,
341        })
342        .await?
343    }
344
345    /// Resumes one repository or all, and publishes the status before it answers, so a client that reads the status
346    /// or the events after the answer sees them resumed. Fails as `set_paused` does.
347    pub async fn resume(&self, target: Target) -> Result<(), ControlError> {
348        self.send(|reply| Command::Resume { target, reply }).await?
349    }
350
351    /// Reseeds a profile as soon as its slots are free; none of them starts a runner until then.
352    pub async fn reseed(&self, profile: &str) -> Result<(), ControlError> {
353        self.send(|reply| Command::Reseed {
354            profile: profile.to_owned(),
355            reply,
356        })
357        .await?
358    }
359
360    /// Removes a slot's volume or a shared cache's, unless a runner uses it.
361    pub async fn clean_cache(&self, name: &str) -> Result<(), ControlError> {
362        self.send(|reply| Command::CleanCache {
363            name: name.to_owned(),
364            reply,
365        })
366        .await?
367    }
368
369    /// Removes the selected records of failed jobs and their images.
370    pub async fn remove_records(
371        &self,
372        selection: RecordSelection,
373    ) -> Result<api::RemovedRecords, ControlError> {
374        self.send(|reply| Command::RemoveRecords { selection, reply })
375            .await?
376    }
377
378    /// Starts a shell in a job's record and returns its terminal. The caller closes it with `close_shell` once the
379    /// terminal ends, from either side.
380    pub async fn open_shell(&self, job: u64, size: TtySize) -> Result<Tty, ControlError> {
381        self.send(|reply| Command::OpenShell { job, size, reply })
382            .await?
383    }
384
385    /// Removes a shell's container. Not waited for, so it can be sent while the terminal is dropped.
386    pub fn close_shell(&self, container: String) {
387        let _ = self.commands.send(Command::CloseShell { container });
388    }
389
390    /// Runs `commit` to write a validated config, and if it commits, runs with the config from then on, before the
391    /// next tick. If it does not, nothing changes and the answer is `Failed`. Running jobs keep running: removed
392    /// profiles and slots go as their runners finish.
393    pub async fn replace_config(
394        &self,
395        config: Config,
396        memory_max: Option<u64>,
397        commit: Commit,
398    ) -> Result<Option<String>, ControlError> {
399        self.send(|reply| Command::ReplaceConfig {
400            config: Box::new(config),
401            memory_max,
402            commit,
403            reply,
404        })
405        .await?
406    }
407}
408
409/// Names the repositories there are, so a mistyped name shows its fix.
410fn not_configured<'name>(
411    repository: &str,
412    configured: impl Iterator<Item = &'name String>,
413) -> ControlError {
414    let configured: Vec<&str> = configured.map(String::as_str).collect();
415    let configured = if configured.is_empty() {
416        "none is configured".to_owned()
417    } else {
418        format!("configured: {}", configured.join(", "))
419    };
420    ControlError::NotFound(format!(
421        "{repository} is not a configured repository; {configured}"
422    ))
423}
424
425/// `jobs` for people to read: `job 1 on runner-a, runner-b`.
426fn list(jobs: &[api::RunningJob]) -> String {
427    jobs.iter()
428        .map(ToString::to_string)
429        .collect::<Vec<_>>()
430        .join(", ")
431}
432
433impl<G: GithubApi + 'static, D: DockerApi + 'static> Engine<G, D> {
434    /// Carries out a command from the API.
435    pub(super) async fn command(&mut self, command: Command) {
436        match command {
437            Command::Pause {
438                target,
439                jobs,
440                reply,
441            } => {
442                let _ = reply.send(self.pause(&target, jobs).await);
443            }
444            Command::Resume { target, reply } => {
445                let _ = reply.send(self.control.set_paused(&target, false).await);
446            }
447            Command::Reseed { profile, reply } => {
448                let _ = reply.send(self.request_reseed(&profile));
449            }
450            Command::CleanCache { name, reply } => {
451                let _ = reply.send(self.clean_cache(&name).await);
452            }
453            Command::ReplaceConfig {
454                config,
455                memory_max,
456                commit,
457                reply,
458            } => {
459                let answer = match commit.run().await {
460                    Ok(warning) => {
461                        self.reload(*config, memory_max);
462                        Ok(warning)
463                    }
464                    Err(message) => Err(ControlError::Failed(message)),
465                };
466                let _ = reply.send(answer);
467            }
468            Command::RemoveRecords { selection, reply } => {
469                let _ = reply.send(self.remove_records(&selection).await);
470            }
471            Command::OpenShell { job, size, reply } => {
472                let opened = self.open_shell(job, size).await;
473                // A terminal nobody waits for any more is closed at once.
474                if let Err(Ok(tty)) = reply.send(opened) {
475                    self.close_shell(&tty.container).await;
476                }
477            }
478            Command::CloseShell { container } => self.close_shell(&container).await,
479        }
480    }
481
482    /// Pauses one repository or all. A runner of theirs that holds a job, or that GitHub reports busy, refuses it
483    /// with `RunningJobs::Keep`; with `Kill`, the runs of their jobs are cancelled, and their runners removed once
484    /// they exit or `CANCEL_GRACE` later; a runner whose run cannot be cancelled is removed at once.
485    pub(super) async fn pause(
486        &mut self,
487        target: &Target,
488        running: RunningJobs,
489    ) -> Result<api::Paused, ControlError> {
490        let repositories: BTreeSet<String> = match target {
491            Target::Repository(repository) if !self.repositories.contains_key(repository) => {
492                return Err(not_configured(repository, self.repositories.keys()));
493            }
494            Target::Repository(repository) => BTreeSet::from([repository.clone()]),
495            Target::All => self.repositories.keys().cloned().collect(),
496        };
497        self.take_found();
498        let runners: Vec<(SlotRef, Occupant)> = self
499            .occupied()
500            .filter(|(_, occupant)| repositories.contains(&occupant.repository))
501            .map(|(slot, occupant)| (slot, occupant.clone()))
502            .collect();
503        let jobs: Vec<api::RunningJob> = runners
504            .iter()
505            .filter(|(_, occupant)| occupant.busy || occupant.job.is_some())
506            .map(|(_, occupant)| occupant.running_job())
507            .collect();
508        if running == RunningJobs::Keep && !jobs.is_empty() {
509            let mut by_repository: BTreeMap<&str, Vec<api::RunningJob>> = BTreeMap::new();
510            for job in &jobs {
511                by_repository
512                    .entry(&job.repository)
513                    .or_default()
514                    .push(job.clone());
515            }
516            let runs: Vec<String> = by_repository
517                .iter()
518                .map(|(repository, jobs)| format!("{repository} runs {}", list(jobs)))
519                .collect();
520            // The jobs only: each client says how to force the pause in its own terms.
521            return Err(ControlError::Busy {
522                message: runs.join("; "),
523                jobs,
524            });
525        }
526        // Paused in memory even if not recorded: a forced pause still cancels the runs it was asked to.
527        let recorded = self.control.set_paused(target, true).await;
528        match target {
529            Target::Repository(repository) => info!("{repository}: paused"),
530            Target::All => info!("paused every repository"),
531        }
532        let mut stopped = Vec::new();
533        if running == RunningJobs::Keep {
534            return recorded.map(|()| api::Paused { stopped });
535        }
536        for (slot, occupant) in runners {
537            let repository = &occupant.repository;
538            let cancelled = if let Some(job) = occupant.job {
539                self.cancel(repository, job).await
540            } else {
541                if occupant.busy {
542                    warn!(
543                        "{repository}: {} runs a job its log has not named yet; GitHub fails it once the runner is gone",
544                        occupant.runner_name
545                    );
546                }
547                false
548            };
549            if cancelled {
550                // `collect_finished` removes it once it exits, `kill_cancelled` if it does not.
551                self.cancelling.insert(
552                    occupant.runner_name.clone(),
553                    tokio::time::Instant::now() + CANCEL_GRACE,
554                );
555            } else {
556                self.stop(&slot, RunningJobs::Kill).await;
557                if self
558                    .occupied()
559                    .any(|(_, candidate)| candidate.runner_name == occupant.runner_name)
560                {
561                    continue;
562                }
563            }
564            stopped.push(occupant.running_job());
565        }
566        match recorded {
567            Ok(()) => Ok(api::Paused { stopped }),
568            Err(ControlError::NotRecorded(message)) if !stopped.is_empty() => Err(
569                ControlError::NotRecorded(format!("{message}; stopped {}", list(&stopped))),
570            ),
571            Err(error) => Err(error),
572        }
573    }
574
575    /// Removes the runners of cancelled runs that did not exit within `CANCEL_GRACE`, every tick until they are gone.
576    pub(super) async fn kill_cancelled(&mut self) {
577        let now = tokio::time::Instant::now();
578        let running: BTreeSet<String> = self
579            .occupied()
580            .map(|(_, occupant)| occupant.runner_name.clone())
581            .collect();
582        self.cancelling.retain(|name, _| running.contains(name));
583        let overdue: Vec<(SlotRef, String)> = self
584            .occupied()
585            .filter(|(_, occupant)| {
586                self.cancelling
587                    .get(&occupant.runner_name)
588                    .is_some_and(|deadline| *deadline <= now)
589            })
590            .map(|(slot, occupant)| (slot, occupant.runner_name.clone()))
591            .collect();
592        for (slot, name) in overdue {
593            warn!(
594                "{name} did not exit {}s after its run was cancelled",
595                CANCEL_GRACE.as_secs()
596            );
597            // The deadline stays until the runner is gone, so a failed removal is tried again next tick.
598            self.stop(&slot, RunningJobs::Kill).await;
599        }
600    }
601
602    /// Cancels the run of a job; whether GitHub accepted it.
603    async fn cancel(&self, repository: &str, job: u64) -> bool {
604        let run = match self.github.job(repository, job).await {
605            Ok(details) => details.run_id,
606            Err(error) => {
607                warn!("{repository}: cannot read job {job} to cancel its run: {error}");
608                return false;
609            }
610        };
611        match self.github.cancel_run(repository, run).await {
612            Ok(()) => {
613                info!("{repository}: cancelled run {run} of job {job}");
614                true
615            }
616            Err(error) => {
617                warn!("{repository}: cannot cancel run {run}: {error}");
618                false
619            }
620        }
621    }
622
623    /// Asks for a profile's reseed: none of its slots starts a runner until they are free and it ran.
624    pub(super) fn request_reseed(&mut self, profile: &str) -> Result<(), ControlError> {
625        if !self.config.profiles.contains_key(profile) {
626            return Err(ControlError::NotFound(format!(
627                "{profile} is not a configured profile"
628            )));
629        }
630        if self.is_reseeding(profile) {
631            return Err(ControlError::Conflict(format!(
632                "{profile} is reseeding already"
633            )));
634        }
635        // A shell mounts the snapshot a reseed replaces, and may stay open for hours.
636        if self.shell_profiles().contains(profile) {
637            return Err(ControlError::Conflict(format!(
638                "a shell is open in a record of {profile}, which mounts its snapshot; reseed it once the shell exits"
639            )));
640        }
641        info!("{profile}: reseeding once its slots are free");
642        self.reseed_requested.insert(profile.to_owned());
643        Ok(())
644    }
645
646    /// Removes a slot's volume or a shared cache's, unless a runner or a reseed uses it. Snapshots are rebuilt by
647    /// reseeding instead: slot layers sit on top of them.
648    pub(super) async fn clean_cache(&mut self, name: &str) -> Result<(), ControlError> {
649        let Some(volume) = cache::volumes(&self.names, &self.config)
650            .into_iter()
651            .find(|volume| volume.name == name)
652        else {
653            return Err(ControlError::NotFound(format!(
654                "{name} is not a cache volume of this config; `runnerctl caches` lists them"
655            )));
656        };
657        let users: Vec<api::RunningJob> = match &volume.kind {
658            api::CacheKind::Snapshot { profile } => {
659                return Err(ControlError::Invalid(format!(
660                    "{name} is {profile}'s snapshot, which every slot's layers sit on; reseed {profile} to rebuild it"
661                )));
662            }
663            api::CacheKind::Slot { profile, slot } => {
664                if self.reseeds.contains_key(profile) {
665                    return Err(ControlError::Conflict(format!(
666                        "{profile} is reseeding, which writes {name}"
667                    )));
668                }
669                self.occupied()
670                    .filter(|(at, _)| at.profile == *profile && at.index + 1 == *slot)
671                    .map(|(_, occupant)| occupant.running_job())
672                    .collect()
673            }
674            api::CacheKind::Pnpm | api::CacheKind::Maven | api::CacheKind::Mirror => self
675                .occupied()
676                .map(|(_, occupant)| occupant.running_job())
677                .collect(),
678        };
679        let shared = matches!(volume.kind, api::CacheKind::Pnpm | api::CacheKind::Maven);
680        if shared && !self.records.shells.is_empty() {
681            return Err(ControlError::Conflict(format!(
682                "a shell open in a record mounts {name}; clean it once the shell exits"
683            )));
684        }
685        if !users.is_empty() {
686            return Err(ControlError::Busy {
687                message: format!(
688                    "{name} is in use by {}; clean it once they finish",
689                    list(&users)
690                ),
691                jobs: users,
692            });
693        }
694        if volume.kind == api::CacheKind::Mirror && self.mirror_up {
695            cache::mirror_down(&*self.docker, &self.names)
696                .await
697                .map_err(|error| {
698                    ControlError::Failed(format!("cannot stop the mirror: {error}"))
699                })?;
700            self.mirror_up = false;
701        }
702        self.docker
703            .remove_volume(name)
704            .await
705            .map_err(|error| ControlError::Failed(format!("cannot remove {name}: {error}")))?;
706        info!("removed {name}");
707        // Measured again once idle.
708        self.trim_due = true;
709        Ok(())
710    }
711
712    /// Runs with a new, validated config. Running jobs keep running: a removed profile, or slots past a lowered
713    /// number, stay until their runners finish, and no runner starts in a profile while it has as many as its
714    /// slots. Runners started from now on get the new settings.
715    pub(super) fn reload(&mut self, config: Config, memory_max: Option<u64>) {
716        let mut old = std::mem::replace(&mut self.config, config);
717        for (name, profile) in &self.config.profiles {
718            let slots = self.slots.entry(name.clone()).or_default();
719            if slots.len() < profile.slots as usize {
720                slots.resize(profile.slots as usize, None);
721            }
722            self.retired.remove(name);
723        }
724        for (name, profile) in std::mem::take(&mut old.profiles) {
725            if !self.config.profiles.contains_key(&name) {
726                self.retired.insert(name, profile);
727            }
728        }
729        self.drop_closed_slots();
730        let names: BTreeSet<String> = self
731            .config
732            .repositories
733            .iter()
734            .map(|repository| repository.name.clone())
735            .collect();
736        let removed: Vec<String> = self
737            .repositories
738            .keys()
739            .filter(|name| !names.contains(*name))
740            .cloned()
741            .collect();
742        self.draining.extend(removed);
743        self.draining.retain(|name| !names.contains(name));
744        self.repositories.retain(|name, _| names.contains(name));
745        for name in &names {
746            self.repositories.entry(name.clone()).or_default();
747        }
748        self.control.set_repositories(names);
749        let profiles = &self.config.profiles;
750        self.reseed_requested
751            .retain(|name| profiles.contains_key(name));
752        self.isolation = isolation::backend(self.config.host.isolation);
753        self.memory_max = memory_max;
754        self.caches
755            .send_replace(cache::volumes(&self.names, &self.config));
756        info!("reloaded the config");
757    }
758}