1use 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
27pub(super) const CANCEL_GRACE: Duration = Duration::from_secs(90);
30
31#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
33pub enum Shutdown {
34 No,
36 Graceful,
38 Force,
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum RunningJobs {
45 Keep,
47 Kill,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
53pub enum Target {
54 Repository(String),
56 All,
58}
59
60#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
62pub enum ControlError {
63 #[error("{0}")]
65 NotFound(String),
66 #[error("{message}")]
68 Busy {
69 message: String,
71 jobs: Vec<api::RunningJob>,
73 },
74 #[error("{0}")]
76 Conflict(String),
77 #[error("{0}")]
79 Invalid(String),
80 #[error("{0}")]
82 Failed(String),
83 #[error("{0}")]
85 NotRecorded(String),
86 #[error("runnerd is stopping")]
88 Stopped,
89}
90
91#[derive(Debug)]
93pub enum Command {
94 Pause {
96 target: Target,
98 jobs: RunningJobs,
100 reply: oneshot::Sender<Result<api::Paused, ControlError>>,
102 },
103 Resume {
105 target: Target,
107 reply: oneshot::Sender<Result<(), ControlError>>,
109 },
110 Reseed {
112 profile: String,
114 reply: oneshot::Sender<Result<(), ControlError>>,
116 },
117 CleanCache {
119 name: String,
121 reply: oneshot::Sender<Result<(), ControlError>>,
123 },
124 ReplaceConfig {
126 config: Box<Config>,
128 memory_max: Option<u64>,
130 commit: Commit,
132 reply: oneshot::Sender<Result<Option<String>, ControlError>>,
134 },
135 RemoveRecords {
137 selection: RecordSelection,
139 reply: oneshot::Sender<Result<api::RemovedRecords, ControlError>>,
141 },
142 OpenShell {
144 job: u64,
146 size: TtySize,
148 reply: oneshot::Sender<Result<Tty, ControlError>>,
150 },
151 CloseShell {
153 container: String,
155 },
156}
157
158pub struct Commit(Box<dyn FnOnce() -> Result<Option<String>, String> + Send>);
161
162impl Commit {
163 pub fn new(write: impl FnOnce() -> Result<Option<String>, String> + Send + 'static) -> Commit {
165 Commit(Box::new(write))
166 }
167
168 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#[derive(Debug)]
184pub struct Commands(mpsc::UnboundedReceiver<Command>);
185
186impl Commands {
187 pub async fn recv(&mut self) -> Option<Command> {
189 self.0.recv().await
190 }
191}
192
193#[derive(Debug)]
195pub struct Control {
196 repositories: Mutex<BTreeSet<String>>,
199 paused: Mutex<BTreeSet<String>>,
200 store: Store,
202 shutdown: Mutex<Shutdown>,
203 pub(super) wake: Notify,
204 commands: mpsc::UnboundedSender<Command>,
205}
206
207impl Control {
208 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 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 pub fn mode(&self) -> Shutdown {
237 *lock(&self.shutdown)
238 }
239
240 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 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 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 drop(self.store.paused(paused.iter().cloned().collect()));
297 }
298 *configured = repositories;
299 }
300
301 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 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 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 pub async fn resume(&self, target: Target) -> Result<(), ControlError> {
348 self.send(|reply| Command::Resume { target, reply }).await?
349 }
350
351 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 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 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 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 pub fn close_shell(&self, container: String) {
387 let _ = self.commands.send(Command::CloseShell { container });
388 }
389
390 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
409fn 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
425fn 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 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 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 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 return Err(ControlError::Busy {
522 message: runs.join("; "),
523 jobs,
524 });
525 }
526 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 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 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 self.stop(&slot, RunningJobs::Kill).await;
599 }
600 }
601
602 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 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 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 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 self.trim_due = true;
709 Ok(())
710 }
711
712 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}