1use std::collections::{BTreeMap, BTreeSet};
113use std::fmt;
114use std::num::NonZeroU16;
115use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
116use std::sync::{Arc, Mutex};
117use std::time::Duration;
118
119use runner_manager_domain::attempt::{
120 AttemptOutcome, AttemptState, FailureReason, IsolationProviderFailure, RunnerAttempt,
121};
122use runner_manager_domain::capacity::{Allocation, HostAllocator, LimitingFactor};
123use runner_manager_domain::memory::{MemoryFigures, MemoryVerdict};
124use runner_manager_domain::model::Label;
125use runner_manager_domain::model::{
126 AttemptId, Clock, Host, IdlePollInterval, Org, OwnerRepo, PolicyId, RefreshInterval,
127 ScaleTarget, Timestamp,
128};
129use runner_manager_domain::policy::{DemandTally, HostTraits, ScalePolicy, is_pool_job};
130use runner_manager_github::demand::{DemandGateway, QueuedDemand, demand_requests_per_poll};
131use runner_manager_github::rest::{
132 ActivityScope, CancelToken, InventoryError, RateLimitKind, RefreshState,
133};
134use runner_manager_github::traffic::{PrimaryPressure, RateLimitSnapshot};
135
136pub const REPOSITORY_LIST_REFRESH_MULTIPLE: u32 = 30;
156
157pub const MAX_OFFLINE_BACKOFF: Duration = Duration::from_secs(15 * 60);
166
167const MAX_BACKOFF_DOUBLINGS: u32 = 5;
173
174const JITTER_RATIO: f64 = 0.5;
182
183const RATE_LIMIT_JITTER_RATIO: f64 = 0.1;
191
192pub const MAX_IDLE_REQUESTS_PER_MINUTE_PER_TARGET: u32 = 300;
200
201pub const GITHUB_CANCELS_QUEUED_JOBS_AFTER: Duration = Duration::from_secs(24 * 60 * 60);
208
209pub const ALLOCATION_LOCK_WAIT: Duration = Duration::from_secs(5);
239
240#[derive(Debug, Clone, PartialEq, Eq)]
257pub enum PollOutcome {
258 Ready(QueuedDemand),
261 Failed(RefreshState),
264}
265
266impl PollOutcome {
267 #[must_use]
269 pub const fn reading(&self) -> Option<&QueuedDemand> {
270 match self {
271 Self::Ready(demand) => Some(demand),
272 Self::Failed(_) => None,
273 }
274 }
275
276 #[must_use]
278 pub const fn failure(&self) -> Option<&RefreshState> {
279 match self {
280 Self::Failed(state) => Some(state),
281 Self::Ready(_) => None,
282 }
283 }
284
285 #[must_use]
292 pub fn is_offline(&self) -> bool {
293 matches!(self, Self::Failed(RefreshState::Offline))
294 }
295}
296
297#[async_trait::async_trait]
305pub trait DemandSource: fmt::Debug + Send + Sync {
306 async fn poll(&self, scope: &ActivityScope) -> PollOutcome;
308}
309
310#[derive(Debug)]
315pub struct GatewayDemand<G> {
316 gateway: G,
317 cancel: CancelToken,
318}
319
320impl<G: DemandGateway> GatewayDemand<G> {
321 #[must_use]
322 pub const fn new(gateway: G, cancel: CancelToken) -> Self {
323 Self { gateway, cancel }
324 }
325
326 #[must_use]
327 pub const fn gateway(&self) -> &G {
328 &self.gateway
329 }
330}
331
332#[async_trait::async_trait]
333impl<G: DemandGateway + 'static> DemandSource for GatewayDemand<G> {
334 async fn poll(&self, scope: &ActivityScope) -> PollOutcome {
335 match self.gateway.queued_demand(scope, &self.cancel).await {
336 Ok(demand) => PollOutcome::Ready(demand),
337 Err(error) => PollOutcome::Failed(RefreshState::from_error(&error)),
341 }
342 }
343}
344
345#[async_trait::async_trait]
356pub trait RepositoryDirectory: fmt::Debug + Send + Sync {
357 async fn repositories(&self, org: &Org) -> Result<Vec<OwnerRepo>, InventoryError>;
362}
363
364#[derive(Debug, Clone)]
365struct CachedRepositories {
366 repositories: Vec<OwnerRepo>,
367 fetched_at: Timestamp,
368}
369
370#[derive(Debug)]
388pub struct RepositoryCache {
389 directory: Arc<dyn RepositoryDirectory>,
390 clock: Arc<dyn Clock>,
391 ttl: Duration,
392 entries: Mutex<BTreeMap<Org, CachedRepositories>>,
393 lookups: AtomicU64,
394}
395
396impl RepositoryCache {
397 #[must_use]
400 pub fn new(
401 directory: Arc<dyn RepositoryDirectory>,
402 clock: Arc<dyn Clock>,
403 poll: RefreshInterval,
404 ) -> Self {
405 let ttl = Duration::from_secs(u64::from(poll.as_secs()))
406 .saturating_mul(REPOSITORY_LIST_REFRESH_MULTIPLE);
407 Self {
408 directory,
409 clock,
410 ttl,
411 entries: Mutex::new(BTreeMap::new()),
412 lookups: AtomicU64::new(0),
413 }
414 }
415
416 #[must_use]
418 pub const fn ttl(&self) -> Duration {
419 self.ttl
420 }
421
422 #[must_use]
427 pub fn lookups(&self) -> u64 {
428 self.lookups.load(Ordering::SeqCst)
429 }
430
431 pub async fn scope_for(&self, target: &ScaleTarget) -> Result<ActivityScope, InventoryError> {
437 match target {
438 ScaleTarget::Repository(repository) => {
439 Ok(ActivityScope::repository(repository.clone()))
440 }
441 ScaleTarget::Organization(org) => {
442 let repositories = self.repositories_of(org).await?;
443 Ok(ActivityScope::organization(org.clone(), repositories))
444 }
445 }
446 }
447
448 async fn repositories_of(&self, org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
449 let now = self.clock.now();
450 if let Some(fresh) = self.fresh_entry(org, now) {
451 return Ok(fresh);
452 }
453
454 let repositories = self.directory.repositories(org).await?;
461 self.lookups.fetch_add(1, Ordering::SeqCst);
462 self.store(org.clone(), repositories.clone(), now);
463 Ok(repositories)
464 }
465
466 fn fresh_entry(&self, org: &Org, now: Timestamp) -> Option<Vec<OwnerRepo>> {
467 let entries = self.entries.lock().ok()?;
468 let entry = entries.get(org)?;
469 let age = now.signed_duration_since(entry.fetched_at).to_std().ok()?;
470 (age < self.ttl).then(|| entry.repositories.clone())
471 }
472
473 fn store(&self, org: Org, repositories: Vec<OwnerRepo>, fetched_at: Timestamp) {
474 if let Ok(mut entries) = self.entries.lock() {
475 entries.insert(
476 org,
477 CachedRepositories {
478 repositories,
479 fetched_at,
480 },
481 );
482 }
483 }
484}
485
486pub trait Jitter: fmt::Debug + Send + Sync {
497 fn fraction(&self) -> f64;
500}
501
502#[derive(Debug, Clone, Copy, Default)]
504pub struct RandomJitter;
505
506impl Jitter for RandomJitter {
507 fn fraction(&self) -> f64 {
508 rand::random::<f64>()
509 }
510}
511
512#[derive(Debug, Clone, Copy)]
514pub struct FixedJitter(pub f64);
515
516impl Jitter for FixedJitter {
517 fn fraction(&self) -> f64 {
518 self.0
519 }
520}
521
522#[derive(Debug, Clone, Copy, Default)]
524pub struct NoJitter;
525
526impl Jitter for NoJitter {
527 fn fraction(&self) -> f64 {
528 0.0
529 }
530}
531
532#[derive(Debug, Clone, Copy, PartialEq, Eq)]
543pub enum PollPace {
544 Nominal,
546 Idle,
549 Stretched { pressure: PrimaryPressure },
554 RateLimited { kind: RateLimitKind },
556 LockedOut,
558 Offline { consecutive: u32 },
561 Blocked,
566}
567
568impl PollPace {
569 #[must_use]
571 pub const fn is_throttled(&self) -> bool {
572 !matches!(self, Self::Nominal | Self::Idle)
573 }
574
575 #[must_use]
577 pub const fn as_str(&self) -> &'static str {
578 match self {
579 Self::Nominal => "nominal",
580 Self::Idle => "idle",
581 Self::Stretched {
582 pressure: PrimaryPressure::Normal,
583 } => "nominal",
584 Self::Stretched {
585 pressure: PrimaryPressure::Low,
586 } => "stretched_low_budget",
587 Self::Stretched {
588 pressure: PrimaryPressure::Critical,
589 } => "stretched_critical_budget",
590 Self::RateLimited {
591 kind: RateLimitKind::Primary,
592 } => "rate_limited_primary",
593 Self::RateLimited {
594 kind: RateLimitKind::Secondary,
595 } => "rate_limited_secondary",
596 Self::LockedOut => "locked_out",
597 Self::Offline { .. } => "offline",
598 Self::Blocked => "blocked",
599 }
600 }
601}
602
603impl fmt::Display for PollPace {
604 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
605 f.write_str(self.as_str())
606 }
607}
608
609#[derive(Debug, Clone, Copy, PartialEq, Eq)]
611pub struct NextPoll {
612 pub delay: Duration,
613 pub pace: PollPace,
614}
615
616#[derive(Debug, Clone)]
636pub struct PollSchedule {
637 interval: RefreshInterval,
638 idle: Duration,
641 consecutive_offline: u32,
642 offline_since: Option<Timestamp>,
643}
644
645#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
647pub struct PollConditions {
648 pub quiet: bool,
651 pub quota: Option<RateLimitSnapshot>,
653 pub requests_per_pass: u32,
657}
658
659impl PollSchedule {
660 #[must_use]
661 pub const fn new(interval: RefreshInterval) -> Self {
662 Self {
663 interval,
664 idle: Duration::from_secs(interval.as_secs() as u64),
665 consecutive_offline: 0,
666 offline_since: None,
667 }
668 }
669
670 #[must_use]
675 pub fn with_idle(mut self, idle: Duration) -> Self {
676 self.set_intervals(self.interval, idle);
677 self
678 }
679
680 pub fn set_intervals(&mut self, interval: RefreshInterval, idle: Duration) {
682 self.interval = interval;
683 self.idle = idle.min(self.nominal()).max(Self::idle_floor());
684 }
685
686 #[must_use]
687 pub const fn interval(&self) -> RefreshInterval {
688 self.interval
689 }
690
691 #[must_use]
693 pub const fn idle(&self) -> Duration {
694 self.idle
695 }
696
697 #[must_use]
700 pub const fn idle_floor() -> Duration {
701 Duration::from_secs(IdlePollInterval::MIN_SECS as u64)
702 }
703
704 #[must_use]
706 pub const fn nominal(&self) -> Duration {
707 Duration::from_secs(self.interval.as_secs() as u64)
708 }
709
710 #[must_use]
712 pub const fn consecutive_offline(&self) -> u32 {
713 self.consecutive_offline
714 }
715
716 #[must_use]
725 pub fn offline_for(&self, now: Timestamp) -> Option<Duration> {
726 let since = self.offline_since?;
727 now.signed_duration_since(since).to_std().ok()
728 }
729
730 #[must_use]
732 pub const fn floor() -> Duration {
733 Duration::from_secs(RefreshInterval::MIN_SECS as u64)
734 }
735
736 pub fn next_poll(
744 &mut self,
745 failure: Option<&RefreshState>,
746 now: Timestamp,
747 jitter: &dyn Jitter,
748 ) -> NextPoll {
749 self.next_poll_with(failure, PollConditions::default(), now, jitter)
750 }
751
752 pub fn next_poll_with(
767 &mut self,
768 failure: Option<&RefreshState>,
769 conditions: PollConditions,
770 now: Timestamp,
771 jitter: &dyn Jitter,
772 ) -> NextPoll {
773 let nominal = self.nominal();
774
775 let next = match failure {
776 None => {
777 self.recovered();
778 let pressure = conditions
779 .quota
780 .map_or(PrimaryPressure::Normal, |quota| quota.pressure(now));
781 match (pressure, conditions.quota) {
782 (PrimaryPressure::Low | PrimaryPressure::Critical, Some(quota)) => NextPoll {
783 delay: nominal
784 .saturating_mul(pressure.stretch_factor())
785 .min(quota.reset_in(now).max(nominal)),
786 pace: PollPace::Stretched { pressure },
787 },
788 _ if conditions.quiet && self.idle < nominal => NextPoll {
791 delay: self.idle.max(idle_floor_for(conditions.requests_per_pass)),
792 pace: PollPace::Idle,
793 },
794 _ => NextPoll {
795 delay: nominal,
796 pace: PollPace::Nominal,
797 },
798 }
799 }
800 Some(RefreshState::Offline) => {
801 self.consecutive_offline = self.consecutive_offline.saturating_add(1);
802 self.offline_since.get_or_insert(now);
804 NextPoll {
805 delay: self.offline_delay(nominal, jitter),
806 pace: PollPace::Offline {
807 consecutive: self.consecutive_offline,
808 },
809 }
810 }
811 Some(state @ RefreshState::RateLimited(limit)) => {
812 self.recovered();
813 let wait = retry_floor(state, now).max(nominal);
816 NextPoll {
817 delay: wait.saturating_add(
818 wait.mul_f64(RATE_LIMIT_JITTER_RATIO * jitter.fraction().clamp(0.0, 1.0)),
819 ),
820 pace: PollPace::RateLimited { kind: limit.kind },
821 }
822 }
823 Some(state @ RefreshState::LockedOut { .. }) => {
824 self.recovered();
825 NextPoll {
826 delay: retry_floor(state, now).max(nominal),
827 pace: PollPace::LockedOut,
828 }
829 }
830 Some(_) => {
836 self.recovered();
839 NextPoll {
840 delay: nominal,
841 pace: PollPace::Blocked,
842 }
843 }
844 };
845
846 debug_assert!(
847 next.delay >= Self::floor() || next.pace == PollPace::Idle,
848 "the 30-second floor is a rate-budget constraint and no charged branch may go below it"
849 );
850 debug_assert!(
851 next.delay >= Self::idle_floor(),
852 "the idle floor is a secondary-limit constraint and no branch may go below it"
853 );
854 next
855 }
856
857 fn recovered(&mut self) {
859 self.consecutive_offline = 0;
860 self.offline_since = None;
861 }
862
863 fn offline_delay(&self, nominal: Duration, jitter: &dyn Jitter) -> Duration {
864 let doublings = self
865 .consecutive_offline
866 .saturating_sub(1)
867 .min(MAX_BACKOFF_DOUBLINGS);
868 let grown = nominal.saturating_mul(1_u32 << doublings);
869 let capped = grown.min(MAX_OFFLINE_BACKOFF);
870 let spread = capped.mul_f64(JITTER_RATIO * jitter.fraction().clamp(0.0, 1.0));
876 capped.saturating_add(spread).max(Self::floor())
877 }
878}
879
880fn is_quiet(readings: &BTreeMap<ScaleTarget, PollOutcome>, report: &ReconcileReport) -> bool {
891 !readings.is_empty()
892 && readings
893 .values()
894 .all(|outcome| outcome.reading().is_some_and(QueuedDemand::is_quiet))
895 && report.started == 0
896 && report.host_capacity_unreadable == 0
899 && report.attempts_unreadable == 0
900 && report
904 .allocations
905 .iter()
906 .all(|allocation| allocation.active_owned == 0 && allocation.demand == 0)
907}
908
909fn idle_floor_for(requests_per_pass: u32) -> Duration {
911 Duration::from_secs(
912 u64::from(requests_per_pass) * 60 / u64::from(MAX_IDLE_REQUESTS_PER_MINUTE_PER_TARGET),
913 )
914}
915
916fn retry_floor(state: &RefreshState, now: Timestamp) -> Duration {
924 state.retry_delay(now).unwrap_or(PollSchedule::floor())
925}
926
927#[derive(Debug, Clone, Copy, PartialEq, Eq)]
940pub struct OfflineState {
941 pub consecutive: u32,
943 pub retry_in: Duration,
945 pub offline_for: Option<Duration>,
947}
948
949impl OfflineState {
950 #[must_use]
951 pub const fn new(consecutive: u32, retry_in: Duration) -> Self {
952 Self {
953 consecutive,
954 retry_in,
955 offline_for: None,
956 }
957 }
958
959 #[must_use]
960 pub const fn since(mut self, offline_for: Duration) -> Self {
961 self.offline_for = Some(offline_for);
962 self
963 }
964
965 #[must_use]
970 pub fn has_outlasted_the_queue(&self) -> bool {
971 self.offline_for
972 .is_some_and(|elapsed| elapsed >= GITHUB_CANCELS_QUEUED_JOBS_AFTER)
973 }
974}
975
976impl fmt::Display for OfflineState {
977 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
978 write!(
979 f,
980 "GitHub is unreachable; no new runners are being started and running \
981 runners are left alone. Retrying in {}s",
982 self.retry_in.as_secs()
983 )?;
984 if self.has_outlasted_the_queue() {
985 f.write_str(
986 ". This outage has lasted more than 24 hours, and GitHub cancels a queued \
987 job after 24 hours, so queued work has been lost",
988 )
989 } else {
990 f.write_str(
991 ". GitHub cancels a queued job after 24 hours, so an outage longer than \
992 that loses queued work",
993 )
994 }
995 }
996}
997
998#[derive(Debug, Clone, Copy)]
1004pub struct LaunchRequest<'a> {
1005 pub host: &'a Host,
1006 pub policy: &'a ScalePolicy,
1007 pub(crate) allocation_guard: &'a AllocationGuard,
1013}
1014
1015#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
1022#[error("the runner could not be started: {reason}")]
1023pub struct LaunchFailure {
1024 pub reason: FailureReason,
1025}
1026
1027#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1030pub struct ReplacementIntent {
1031 pub policy: PolicyId,
1032 pub previous_attempt: AttemptId,
1033 pub operation: &'static str,
1034}
1035
1036impl LaunchFailure {
1037 #[must_use]
1038 pub const fn new(reason: FailureReason) -> Self {
1039 Self { reason }
1040 }
1041}
1042
1043#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1046pub enum HostCapacityError {
1047 MissingHost,
1048 JournalUnreadable,
1049}
1050
1051impl HostCapacityError {
1052 #[must_use]
1053 pub const fn reason(self) -> &'static str {
1054 match self {
1055 Self::MissingHost => "host_missing",
1056 Self::JournalUnreadable => "host_journal_unreadable",
1057 }
1058 }
1059}
1060
1061#[derive(Debug)]
1062enum AllocationFailure {
1063 HostCapacity(HostCapacityError),
1064 Attempts(LaunchFailure),
1065}
1066
1067#[async_trait::async_trait]
1104pub trait RunnerLauncher: fmt::Debug + Send + Sync {
1105 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError>;
1109
1110 async fn supervise(
1115 &self,
1116 _policy: &ScalePolicy,
1117 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
1118 Ok(Vec::new())
1119 }
1120
1121 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure>;
1134
1135 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure>;
1186
1187 async fn finish_launch(
1199 &self,
1200 _policy: &ScalePolicy,
1201 attempt: RunnerAttempt,
1202 _waits: ReservationWaits,
1203 ) -> Result<RunnerAttempt, LaunchFailure> {
1204 Ok(attempt)
1205 }
1206
1207 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure>;
1215}
1216
1217#[derive(Debug)]
1228pub struct FitnessGatedLauncher {
1229 inner: Arc<dyn RunnerLauncher>,
1230 unfit: Arc<AtomicBool>,
1231}
1232
1233impl FitnessGatedLauncher {
1234 #[must_use]
1235 pub const fn new(inner: Arc<dyn RunnerLauncher>, unfit: Arc<AtomicBool>) -> Self {
1236 Self { inner, unfit }
1237 }
1238}
1239
1240#[async_trait::async_trait]
1241impl RunnerLauncher for FitnessGatedLauncher {
1242 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
1243 self.inner.host_capacity(host).await
1244 }
1245
1246 async fn supervise(
1247 &self,
1248 policy: &ScalePolicy,
1249 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
1250 self.inner.supervise(policy).await
1251 }
1252
1253 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
1254 self.inner.attempts().await
1255 }
1256
1257 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
1258 if self.unfit.load(Ordering::Acquire) {
1259 return Err(LaunchFailure::new(FailureReason::HostUnfit));
1260 }
1261 self.inner.launch(request).await
1262 }
1263
1264 async fn finish_launch(
1265 &self,
1266 policy: &ScalePolicy,
1267 attempt: RunnerAttempt,
1268 waits: ReservationWaits,
1269 ) -> Result<RunnerAttempt, LaunchFailure> {
1270 self.inner.finish_launch(policy, attempt, waits).await
1271 }
1272
1273 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
1274 self.inner.clean(attempt).await
1275 }
1276}
1277
1278pub struct AllocationGuard {
1288 _held: Box<dyn std::any::Any + Send + Sync>,
1289 launch_hold: LaunchHold,
1290}
1291
1292impl AllocationGuard {
1293 #[must_use]
1295 fn new<T: Send + Sync + 'static>(held: T) -> Self {
1296 Self {
1297 _held: Box::new(held),
1298 launch_hold: LaunchHold::default(),
1299 }
1300 }
1301
1302 fn launch_hold(&self) -> LaunchHold {
1304 self.launch_hold.clone()
1305 }
1306}
1307
1308#[derive(Clone, Default)]
1313struct LaunchHold {
1314 _held: Option<Arc<dyn std::any::Any + Send + Sync>>,
1315}
1316
1317impl fmt::Debug for AllocationGuard {
1318 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1319 f.write_str("AllocationGuard")
1320 }
1321}
1322
1323#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
1325#[error("the host-wide allocation lock is held by another allocator; no runtime was created")]
1326pub struct AllocationLockBusy;
1327
1328#[async_trait::async_trait]
1342pub trait AllocationLock: fmt::Debug + Send + Sync {
1343 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy>;
1349}
1350
1351#[derive(Debug, Default)]
1359pub struct InProcessAllocationLock {
1360 mutex: Arc<tokio::sync::Mutex<()>>,
1361}
1362
1363impl InProcessAllocationLock {
1364 #[must_use]
1365 pub fn new() -> Self {
1366 Self::default()
1367 }
1368}
1369
1370#[async_trait::async_trait]
1371impl AllocationLock for InProcessAllocationLock {
1372 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
1373 let mutex = Arc::clone(&self.mutex);
1374 let guard = mutex.lock_owned().await;
1375 Ok(AllocationGuard::new(guard))
1376 }
1377}
1378
1379#[derive(Debug, Clone)]
1394pub struct FileAllocationLock {
1395 paths: Arc<runner_manager_platform::paths::AppPaths>,
1396 wait: Duration,
1397}
1398
1399impl FileAllocationLock {
1400 #[must_use]
1401 pub const fn new(paths: Arc<runner_manager_platform::paths::AppPaths>) -> Self {
1402 Self {
1403 paths,
1404 wait: ALLOCATION_LOCK_WAIT,
1405 }
1406 }
1407
1408 #[must_use]
1409 pub const fn with_wait(mut self, wait: Duration) -> Self {
1410 self.wait = wait;
1411 self
1412 }
1413}
1414
1415#[async_trait::async_trait]
1416impl AllocationLock for FileAllocationLock {
1417 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
1418 use runner_manager_platform::lock::{HostLock, LockKind};
1419
1420 let paths = Arc::clone(&self.paths);
1421 let wait = self.wait;
1422 let held = tokio::task::spawn_blocking(move || {
1423 HostLock::acquire(&paths, LockKind::Allocation, wait)
1424 })
1425 .await;
1426
1427 match held {
1428 Ok(Ok(lock)) => Ok(AllocationGuard::new(lock)),
1429 Ok(Err(_)) | Err(_) => Err(AllocationLockBusy),
1433 }
1434 }
1435}
1436
1437#[derive(Debug)]
1465pub struct WslRecoveryAllocationLock {
1466 paths: Arc<runner_manager_platform::paths::AppPaths>,
1467 inner: Arc<dyn AllocationLock>,
1468 blocked_after_refusals: u32,
1469 blocked_after: Duration,
1470 streak_gap: Duration,
1472 in_flight: Arc<std::sync::atomic::AtomicUsize>,
1473 claim: tokio::sync::Mutex<std::sync::Weak<LaunchFenceGuard>>,
1475 refusals: Mutex<Refusals>,
1476}
1477
1478#[derive(Debug, Default)]
1480struct Refusals {
1481 streak: Option<Streak>,
1484 recorded: bool,
1485}
1486
1487#[derive(Debug, Clone, Copy)]
1488struct Streak {
1489 since: chrono::DateTime<chrono::Utc>,
1490 latest: chrono::DateTime<chrono::Utc>,
1491 count: u32,
1492}
1493
1494struct InFlight(Arc<std::sync::atomic::AtomicUsize>);
1496
1497impl InFlight {
1498 fn enter(count: &Arc<std::sync::atomic::AtomicUsize>) -> Self {
1499 count.fetch_add(1, Ordering::SeqCst);
1500 Self(Arc::clone(count))
1501 }
1502}
1503
1504impl Drop for InFlight {
1505 fn drop(&mut self) {
1506 self.0.fetch_sub(1, Ordering::SeqCst);
1507 }
1508}
1509
1510struct LaunchFenceGuard(Option<runner_manager_platform::wsl::fence::GuestLaunchClaim>);
1514
1515impl Drop for LaunchFenceGuard {
1516 fn drop(&mut self) {
1517 if let Some(claim) = self.0.take()
1518 && let Err(error) = claim.release()
1519 {
1520 tracing::warn!(
1521 %error,
1522 "the WSL launch fence could not be released; the next launch reclaims it"
1523 );
1524 }
1525 }
1526}
1527
1528impl WslRecoveryAllocationLock {
1529 #[must_use]
1530 pub fn new(
1531 paths: Arc<runner_manager_platform::paths::AppPaths>,
1532 inner: Arc<dyn AllocationLock>,
1533 ) -> Self {
1534 use runner_manager_platform::launch_health::{BLOCKED_AFTER, BLOCKED_AFTER_DEFERRALS};
1535 Self {
1536 paths,
1537 inner,
1538 blocked_after_refusals: BLOCKED_AFTER_DEFERRALS,
1539 blocked_after: BLOCKED_AFTER,
1540 streak_gap: BLOCKED_AFTER,
1541 in_flight: Arc::default(),
1542 claim: tokio::sync::Mutex::default(),
1543 refusals: Mutex::default(),
1544 }
1545 }
1546
1547 #[cfg(test)]
1550 const fn with_blocked_after(mut self, refusals: u32, after: Duration) -> Self {
1551 self.blocked_after_refusals = refusals;
1552 self.blocked_after = after;
1553 self
1554 }
1555
1556 #[cfg(test)]
1558 const fn with_streak_gap(mut self, gap: Duration) -> Self {
1559 self.streak_gap = gap;
1560 self
1561 }
1562
1563 fn refusals(&self) -> std::sync::MutexGuard<'_, Refusals> {
1564 self.refusals
1565 .lock()
1566 .unwrap_or_else(std::sync::PoisonError::into_inner)
1567 }
1568
1569 fn note_granted(&self) {
1570 let mut refusals = self.refusals();
1571 refusals.streak = None;
1572 if refusals.recorded {
1573 match runner_manager_platform::launch_health::clear_launches_blocked(&self.paths) {
1574 Ok(()) => refusals.recorded = false,
1575 Err(error) => tracing::warn!(%error, "the blocked-launch record cannot be cleared"),
1577 }
1578 }
1579 }
1580
1581 fn note_refused(&self, cause: &runner_manager_platform::launch_health::BlockCause) {
1582 use runner_manager_platform::launch_health::{Viewpoint, record_launches_blocked};
1583 use runner_manager_platform::wsl::fence::elapsed_at_least;
1584
1585 if self.in_flight.load(Ordering::SeqCst) > 0 {
1586 return;
1587 }
1588 let now = chrono::Utc::now();
1589 let mut refusals = self.refusals();
1590 let streak = match refusals.streak {
1593 Some(streak) if !elapsed_at_least(streak.latest, now, self.streak_gap) => Streak {
1594 latest: now,
1595 count: streak.count.saturating_add(1),
1596 ..streak
1597 },
1598 _ => Streak {
1599 since: now,
1600 latest: now,
1601 count: 1,
1602 },
1603 };
1604 refusals.streak = Some(streak);
1605 if streak.count < self.blocked_after_refusals
1606 || !elapsed_at_least(streak.since, now, self.blocked_after)
1607 {
1608 return;
1609 }
1610 let blocked = cause.blocked_since(streak.since, Viewpoint::Guest);
1611 if !refusals.recorded {
1612 tracing::warn!(
1613 refusals = streak.count,
1614 "runner launches are blocked {blocked}: {}",
1615 blocked.remedy
1616 );
1617 }
1618 match record_launches_blocked(&self.paths, &blocked, now) {
1621 Ok(()) => refusals.recorded = true,
1622 Err(error) => tracing::warn!(%error, "the blocked-launch record cannot be written"),
1623 }
1624 }
1625}
1626
1627fn claim_guest_fence(
1629 root: &std::path::Path,
1630) -> Result<Arc<LaunchFenceGuard>, runner_manager_platform::launch_health::BlockCause> {
1631 use runner_manager_platform::launch_health::BlockCause;
1632 use runner_manager_platform::wsl::fence::{
1633 FENCE_DIRECTORY, FenceClaim, GuestClaimAttempt, try_claim_guest_launch,
1634 };
1635
1636 match try_claim_guest_launch(root) {
1637 Ok(GuestClaimAttempt::Claimed { claim, reclaimed }) => {
1638 if let Some(reclaimed) = reclaimed {
1639 tracing::warn!(
1640 shared_root = %root.display(),
1641 %reclaimed,
1642 "reclaimed a WSL launch fence claim whose owner is gone"
1643 );
1644 }
1645 Ok(Arc::new(LaunchFenceGuard(Some(claim))))
1648 }
1649 Ok(GuestClaimAttempt::Busy) => {
1650 let directory = root.join(FENCE_DIRECTORY);
1651 Err(match FenceClaim::owner(root) {
1652 Ok(owner) => BlockCause::FenceHeld { directory, owner },
1653 Err(error) => BlockCause::FenceOwnerUnreadable {
1654 directory,
1655 detail: error.to_string(),
1656 },
1657 })
1658 }
1659 Err(error) => {
1660 Err(match FenceClaim::owner(root) {
1663 Err(owner_error) => BlockCause::FenceOwnerUnreadable {
1664 directory: root.join(FENCE_DIRECTORY),
1665 detail: owner_error.to_string(),
1666 },
1667 Ok(_) => BlockCause::FenceUnusable {
1668 detail: error.to_string(),
1669 },
1670 })
1671 }
1672 }
1673}
1674
1675#[async_trait::async_trait]
1676impl AllocationLock for WslRecoveryAllocationLock {
1677 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
1678 use runner_manager_platform::launch_health::BlockCause;
1679 use runner_manager_platform::wsl::fence::{DrainRequest, GuestRecoveryConfig};
1680
1681 let paths = Arc::clone(&self.paths);
1682 let mut shared = self.claim.lock().await;
1683 let held = shared.upgrade();
1684 let claim = tokio::task::spawn_blocking(move || {
1685 let unusable = |error: &dyn fmt::Display| BlockCause::FenceUnusable {
1686 detail: error.to_string(),
1687 };
1688 let config = match GuestRecoveryConfig::read(&paths) {
1689 Ok(Some(config)) => config,
1690 Ok(None) => return Ok(Arc::new(LaunchFenceGuard(None))),
1691 Err(error) => return Err(unusable(&error)),
1692 };
1693 let root = config.shared_root;
1694 let claim = match held {
1695 Some(held) => {
1697 if let Some(Err(error)) = held.0.as_ref().map(|claim| claim.renew()) {
1698 tracing::warn!(%error, "the WSL launch fence's age could not be renewed");
1699 }
1700 held
1701 }
1702 None => claim_guest_fence(&root)?,
1703 };
1704 match DrainRequest::read(&root) {
1705 Ok(None) => Ok(claim),
1706 Ok(Some(_)) => Err(BlockCause::DrainRequested { root }),
1707 Err(error) => Err(unusable(&error)),
1708 }
1709 })
1710 .await
1711 .map_err(|_| AllocationLockBusy)?;
1712
1713 let claim = match claim {
1714 Ok(claim) => claim,
1715 Err(cause) => {
1716 drop(shared);
1717 self.note_refused(&cause);
1718 return Err(AllocationLockBusy);
1719 }
1720 };
1721 if claim.0.is_some() {
1722 *shared = Arc::downgrade(&claim);
1723 }
1724 drop(shared);
1725 let Ok(inner) = self.inner.acquire().await else {
1726 self.note_refused(&BlockCause::AllocationLockBusy {
1727 path: self
1728 .paths
1729 .state_dir()
1730 .join(runner_manager_platform::lock::LockKind::Allocation.file_name()),
1731 });
1732 return Err(AllocationLockBusy);
1733 };
1734 self.note_granted();
1735 let in_flight = InFlight::enter(&self.in_flight);
1736 let mut guard = AllocationGuard::new(inner);
1737 guard.launch_hold = LaunchHold {
1738 _held: Some(Arc::new((claim, in_flight))),
1739 };
1740 Ok(guard)
1741 }
1742}
1743
1744#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1754pub enum OutcomeKind {
1755 CompletedJob,
1756 IdleExit,
1757 Failed,
1758 Orphaned,
1759}
1760
1761impl OutcomeKind {
1762 #[must_use]
1763 pub const fn of(outcome: &AttemptOutcome) -> Self {
1764 match outcome {
1765 AttemptOutcome::CompletedJob => Self::CompletedJob,
1766 AttemptOutcome::ExitedIdleWithoutWork => Self::IdleExit,
1767 AttemptOutcome::Failed { .. } => Self::Failed,
1768 AttemptOutcome::Orphaned => Self::Orphaned,
1769 }
1770 }
1771
1772 #[must_use]
1773 pub const fn is_failure(&self) -> bool {
1774 matches!(self, Self::Failed | Self::Orphaned)
1775 }
1776
1777 #[must_use]
1778 pub const fn as_str(&self) -> &'static str {
1779 match self {
1780 Self::CompletedJob => "completed_job",
1781 Self::IdleExit => "exited_idle_without_work",
1782 Self::Failed => "failed",
1783 Self::Orphaned => "orphaned",
1784 }
1785 }
1786}
1787
1788impl fmt::Display for OutcomeKind {
1789 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1790 f.write_str(self.as_str())
1791 }
1792}
1793
1794#[must_use]
1803pub const fn failure_reason_kind(reason: &FailureReason) -> &'static str {
1804 match reason {
1805 FailureReason::JitRequestFailed => "jit_request_failed",
1806 FailureReason::JitExpired => "jit_expired",
1807 FailureReason::RunnerPackageUnverified => "runner_package_unverified",
1808 FailureReason::RunnerVersionRejected => "runner_version_rejected",
1809 FailureReason::ProcessStartFailed => "process_start_failed",
1810 FailureReason::ProcessExitedUnexpectedly => "process_exited_unexpectedly",
1811 FailureReason::RegistrationTimedOut => "registration_timed_out",
1812 FailureReason::TerminatedAfterRegistrationTimeout => {
1813 "terminated_after_registration_timeout"
1814 }
1815 FailureReason::IsolationProvider(category) => match category {
1816 IsolationProviderFailure::Unsupported => "isolation_provider_unsupported",
1817 IsolationProviderFailure::NotInstalled => "isolation_provider_not_installed",
1818 IsolationProviderFailure::PermissionDenied => "isolation_provider_permission_denied",
1819 IsolationProviderFailure::ImageUnavailableOrIncompatible => {
1820 "isolation_image_unavailable"
1821 }
1822 IsolationProviderFailure::DiskQuotaUnavailable => "isolation_disk_quota_unavailable",
1823 IsolationProviderFailure::Degraded => "isolation_provider_degraded",
1824 IsolationProviderFailure::RuntimeOperationFailed => {
1825 "isolation_runtime_operation_failed"
1826 }
1827 IsolationProviderFailure::OwnershipMismatch => "isolation_ownership_mismatch",
1828 IsolationProviderFailure::UnsafeRuntimePath => "isolation_unsafe_runtime_path",
1829 IsolationProviderFailure::JitHandoffRejected => "isolation_jit_handoff_rejected",
1830 },
1831 FailureReason::WorkspaceCleanupDeferred => "ephemeral_workspace_could_not_be_removed",
1832 FailureReason::HostUnfit => "host_unfit",
1833 FailureReason::Other(_) => "other",
1834 }
1835}
1836
1837#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
1839pub enum RoutingRefusal {
1840 OverlappingProfiles,
1841 UnsatisfiableExclusion,
1844}
1845
1846impl RoutingRefusal {
1847 #[must_use]
1848 pub const fn as_str(self) -> &'static str {
1849 match self {
1850 Self::OverlappingProfiles => "overlapping_profile_matches",
1851 Self::UnsatisfiableExclusion => "unsatisfiable_exclusion",
1852 }
1853 }
1854}
1855
1856impl fmt::Display for RoutingRefusal {
1857 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1858 f.write_str(match self {
1859 Self::OverlappingProfiles => {
1860 "repair overlapping profile labels or require a unique profile selector"
1861 }
1862 Self::UnsatisfiableExclusion => {
1863 "a job asks for a label and for not-<that label>, or for `not-` alone; fix its \
1864 runs-on: no runner can take it"
1865 }
1866 })
1867 }
1868}
1869
1870#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1880pub enum LifecycleEvent {
1881 DemandObserved {
1883 policy: PolicyId,
1884 demand: u32,
1886 not_matched: u32,
1892 unresolvable: u32,
1901 complete: bool,
1903 },
1904 RoutingRefused {
1906 policy: PolicyId,
1907 reason: RoutingRefusal,
1908 count: u32,
1909 },
1910 TargetUnreadable {
1912 policy: PolicyId,
1913 reason: &'static str,
1914 },
1915 Allocated {
1917 policy: PolicyId,
1918 demand: u32,
1919 desired: u16,
1920 active_owned: u16,
1921 headroom: u16,
1922 to_start: u16,
1923 limiting: LimitingFactor,
1924 },
1925 MonitorOnlySkipped { policy: PolicyId },
1928 RunnerStarted {
1930 policy: PolicyId,
1931 attempt: AttemptId,
1932 },
1933 RunnerStartFailed {
1935 policy: PolicyId,
1936 reason: &'static str,
1937 },
1938 AllocationDeferred { policy: PolicyId, count: u16 },
1941 MemoryShort {
1947 policy: PolicyId,
1948 figures: MemoryFigures,
1949 started: bool,
1950 },
1951 AttemptsUnreadable { reason: &'static str },
1957 HostCapacityUnreadable { reason: HostCapacityError },
1959 AttemptCleaned {
1961 policy: PolicyId,
1962 attempt: AttemptId,
1963 outcome: OutcomeKind,
1964 },
1965 AttemptCleanFailed {
1968 policy: PolicyId,
1969 attempt: AttemptId,
1970 reason: &'static str,
1971 },
1972 ScaleDownRefused {
1974 policy: PolicyId,
1975 attempt: AttemptId,
1976 },
1977 PollScheduled { retry_in_ms: u64, pace: PollPace },
1979}
1980
1981impl LifecycleEvent {
1982 #[must_use]
1984 pub const fn name(&self) -> &'static str {
1985 match self {
1986 Self::DemandObserved { .. } => "demand_observed",
1987 Self::RoutingRefused { .. } => "routing_refused",
1988 Self::TargetUnreadable { .. } => "target_unreadable",
1989 Self::Allocated { .. } => "allocated",
1990 Self::MonitorOnlySkipped { .. } => "monitor_only_skipped",
1991 Self::RunnerStarted { .. } => "runner_started",
1992 Self::RunnerStartFailed { .. } => "runner_start_failed",
1993 Self::AllocationDeferred { .. } => "allocation_deferred",
1994 Self::MemoryShort { started: false, .. } => "memory_deferred",
1995 Self::MemoryShort { started: true, .. } => "memory_admitted_alone",
1996 Self::AttemptsUnreadable { .. } => "attempts_unreadable",
1997 Self::HostCapacityUnreadable { .. } => "host_capacity_unreadable",
1998 Self::AttemptCleaned { .. } => "attempt_cleaned",
1999 Self::AttemptCleanFailed { .. } => "attempt_clean_failed",
2000 Self::ScaleDownRefused { .. } => "scale_down_refused",
2001 Self::PollScheduled { .. } => "poll_scheduled",
2002 }
2003 }
2004
2005 #[must_use]
2007 pub const fn policy(&self) -> Option<PolicyId> {
2008 match self {
2009 Self::DemandObserved { policy, .. }
2010 | Self::RoutingRefused { policy, .. }
2011 | Self::TargetUnreadable { policy, .. }
2012 | Self::Allocated { policy, .. }
2013 | Self::MonitorOnlySkipped { policy }
2014 | Self::RunnerStarted { policy, .. }
2015 | Self::RunnerStartFailed { policy, .. }
2016 | Self::AllocationDeferred { policy, .. }
2017 | Self::MemoryShort { policy, .. }
2018 | Self::AttemptCleaned { policy, .. }
2019 | Self::AttemptCleanFailed { policy, .. }
2020 | Self::ScaleDownRefused { policy, .. } => Some(*policy),
2021 Self::PollScheduled { .. }
2022 | Self::AttemptsUnreadable { .. }
2023 | Self::HostCapacityUnreadable { .. } => None,
2024 }
2025 }
2026}
2027
2028impl fmt::Display for LifecycleEvent {
2029 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2030 match self {
2031 Self::DemandObserved {
2032 policy,
2033 demand,
2034 not_matched,
2035 unresolvable,
2036 complete,
2037 } => write!(
2038 f,
2039 "policy {policy}: {demand} queued jobs for this host{}{}{}",
2040 if *not_matched == 0 {
2041 String::new()
2042 } else {
2043 format!(", {not_matched} for other labels")
2044 },
2045 if *unresolvable == 0 {
2046 String::new()
2047 } else {
2048 format!(", {unresolvable} with an unresolvable `runs-on`")
2049 },
2050 if *complete {
2051 ""
2052 } else {
2053 " (a floor, not a total)"
2054 }
2055 ),
2056 Self::TargetUnreadable { policy, reason } => {
2057 write!(f, "policy {policy}: target unreadable ({reason})")
2058 }
2059 Self::RoutingRefused {
2060 policy,
2061 reason,
2062 count,
2063 } => write!(f, "policy {policy}: {count} queued jobs blocked: {reason}"),
2064 Self::Allocated {
2065 policy,
2066 demand,
2067 desired,
2068 active_owned,
2069 headroom,
2070 to_start,
2071 limiting,
2072 } => write!(
2073 f,
2074 "policy {policy}: demand {demand}, desired {desired}, {active_owned} in \
2075 flight, {headroom} free on this host, starting {to_start} ({limiting})"
2076 ),
2077 Self::MonitorOnlySkipped { policy } => {
2078 write!(f, "policy {policy}: monitor-only, skipped")
2079 }
2080 Self::RunnerStarted { policy, attempt } => {
2081 write!(f, "policy {policy}: started attempt {attempt}")
2082 }
2083 Self::RunnerStartFailed { policy, reason } => {
2084 write!(f, "policy {policy}: could not start a runner ({reason})")
2085 }
2086 Self::AllocationDeferred { policy, count } => write!(
2087 f,
2088 "policy {policy}: the allocation lock was held; {count} granted runners \
2089 were not created"
2090 ),
2091 Self::MemoryShort {
2092 policy,
2093 figures,
2094 started: false,
2095 } => write!(f, "policy {policy}: waiting for memory: {figures}"),
2096 Self::MemoryShort {
2097 policy,
2098 figures,
2099 started: true,
2100 } => write!(
2101 f,
2102 "policy {policy}: started although the host lacks the memory, because \
2103 nothing else runs here: {figures}"
2104 ),
2105 Self::AttemptsUnreadable { reason } => write!(
2106 f,
2107 "the host's attempt set could not be read ({reason}); nothing was started, \
2108 and this is not the same as the host being idle"
2109 ),
2110 Self::HostCapacityUnreadable { reason } => write!(
2111 f,
2112 "the host capacity could not be read ({}); new launches wait for readable host settings",
2113 reason.reason()
2114 ),
2115 Self::AttemptCleaned {
2116 policy,
2117 attempt,
2118 outcome,
2119 } => write!(f, "policy {policy}: cleaned attempt {attempt} ({outcome})"),
2120 Self::AttemptCleanFailed {
2121 policy,
2122 attempt,
2123 reason,
2124 } => write!(
2125 f,
2126 "policy {policy}: attempt {attempt} could not be cleaned ({reason}); it \
2127 will be retried"
2128 ),
2129 Self::ScaleDownRefused { policy, attempt } => write!(
2130 f,
2131 "policy {policy}: attempt {attempt} is executing a job and was not removed"
2132 ),
2133 Self::PollScheduled { retry_in_ms, pace } => {
2134 write!(f, "next poll in {retry_in_ms}ms ({pace})")
2135 }
2136 }
2137 }
2138}
2139
2140pub trait EventSink: fmt::Debug + Send + Sync {
2142 fn emit(&self, event: LifecycleEvent);
2143}
2144
2145#[derive(Debug, Clone, Copy, Default)]
2147pub struct NoEvents;
2148
2149impl EventSink for NoEvents {
2150 fn emit(&self, _event: LifecycleEvent) {}
2151}
2152
2153#[derive(Debug, Clone, Copy, Default)]
2161pub struct TracingEvents;
2162
2163impl EventSink for TracingEvents {
2164 fn emit(&self, event: LifecycleEvent) {
2165 let name = event.name();
2166 match event {
2167 LifecycleEvent::DemandObserved {
2168 policy,
2169 demand,
2170 not_matched,
2171 unresolvable,
2172 complete,
2173 } => {
2174 if demand == 0 && not_matched == 0 && unresolvable == 0 && complete {
2179 tracing::debug!(
2180 event = name,
2181 policy_id = %policy,
2182 demand,
2183 not_matched,
2184 unresolvable,
2185 count = u64::from(complete),
2186 );
2187 } else {
2188 tracing::info!(
2189 event = name,
2190 policy_id = %policy,
2191 demand,
2192 not_matched,
2193 unresolvable,
2194 count = u64::from(complete),
2195 );
2196 }
2197 }
2217 LifecycleEvent::TargetUnreadable { policy, reason } => {
2218 tracing::warn!(event = name, policy_id = %policy, reason);
2219 }
2220 LifecycleEvent::RoutingRefused {
2221 policy,
2222 reason,
2223 count,
2224 } => tracing::warn!(
2225 event = name,
2226 policy_id = %policy,
2227 reason = reason.as_str(),
2228 count,
2229 "{reason}"
2230 ),
2231 LifecycleEvent::Allocated {
2232 policy,
2233 demand,
2234 desired,
2235 active_owned,
2236 headroom,
2237 to_start,
2238 limiting,
2239 } if demand == 0 && to_start == 0 && active_owned == 0 => tracing::debug!(
2240 event = name,
2243 policy_id = %policy,
2244 demand,
2245 desired,
2246 capacity = active_owned,
2247 headroom,
2248 count = to_start,
2249 reason = %limiting,
2250 ),
2251 LifecycleEvent::Allocated {
2252 policy,
2253 demand,
2254 desired,
2255 active_owned,
2256 headroom,
2257 to_start,
2258 limiting,
2259 } => tracing::info!(
2260 event = name,
2261 policy_id = %policy,
2262 demand,
2263 desired,
2264 capacity = active_owned,
2265 headroom,
2266 count = to_start,
2267 reason = %limiting,
2268 ),
2269 LifecycleEvent::MonitorOnlySkipped { policy } => {
2270 tracing::debug!(event = name, policy_id = %policy, mode = "monitor_only");
2271 }
2272 LifecycleEvent::RunnerStarted { policy, attempt } => {
2273 tracing::info!(event = name, policy_id = %policy, attempt_id = %attempt);
2274 }
2275 LifecycleEvent::RunnerStartFailed { policy, reason } => {
2276 tracing::warn!(event = name, policy_id = %policy, reason);
2277 }
2278 LifecycleEvent::AllocationDeferred { policy, count } => {
2279 tracing::warn!(
2285 event = name,
2286 policy_id = %policy,
2287 reason = "allocation_lock_unavailable",
2288 count
2289 );
2290 }
2291 LifecycleEvent::MemoryShort {
2292 policy,
2293 figures,
2294 started,
2295 } => {
2296 let reason = if started {
2297 "larger_than_free_memory_nothing_else_running"
2298 } else {
2299 "insufficient_memory"
2300 };
2301 tracing::warn!(
2302 event = name,
2303 policy_id = %policy,
2304 reason,
2305 free_bytes = figures.free,
2306 reserved_bytes = figures.reserved,
2307 estimate_bytes = figures.estimate.bytes,
2308 estimate_source = figures.estimate.source.as_str(),
2309 margin_bytes = figures.margin,
2310 );
2311 }
2312 LifecycleEvent::AttemptsUnreadable { reason } => {
2313 tracing::warn!(event = name, reason);
2314 }
2315 LifecycleEvent::HostCapacityUnreadable { reason } => {
2316 tracing::warn!(event = name, reason = reason.reason());
2317 }
2318 LifecycleEvent::AttemptCleaned {
2319 policy,
2320 attempt,
2321 outcome,
2322 } => tracing::info!(
2323 event = name,
2324 policy_id = %policy,
2325 attempt_id = %attempt,
2326 outcome = outcome.as_str(),
2327 ),
2328 LifecycleEvent::AttemptCleanFailed {
2329 policy,
2330 attempt,
2331 reason,
2332 } => tracing::warn!(
2333 event = name,
2334 policy_id = %policy,
2335 attempt_id = %attempt,
2336 reason,
2337 ),
2338 LifecycleEvent::ScaleDownRefused { policy, attempt } => tracing::info!(
2339 event = name,
2340 policy_id = %policy,
2341 attempt_id = %attempt,
2342 attempt_state = "busy",
2343 ),
2344 LifecycleEvent::PollScheduled { retry_in_ms, pace } => {
2345 if pace == PollPace::Idle {
2346 tracing::debug!(event = name, retry_in_ms, state = pace.as_str());
2347 } else {
2348 tracing::info!(event = name, retry_in_ms, state = pace.as_str());
2349 }
2350 }
2351 }
2352 }
2353}
2354
2355#[derive(Debug, Default)]
2359pub struct EventLog {
2360 events: Mutex<Vec<LifecycleEvent>>,
2361}
2362
2363impl EventLog {
2364 #[must_use]
2365 pub fn new() -> Self {
2366 Self::default()
2367 }
2368
2369 #[must_use]
2370 pub fn events(&self) -> Vec<LifecycleEvent> {
2371 self.events.lock().map(|e| e.clone()).unwrap_or_default()
2372 }
2373
2374 #[must_use]
2376 pub fn count_of(&self, name: &str) -> usize {
2377 self.events()
2378 .iter()
2379 .filter(|event| event.name() == name)
2380 .count()
2381 }
2382}
2383
2384impl EventSink for EventLog {
2385 fn emit(&self, event: LifecycleEvent) {
2386 if let Ok(mut events) = self.events.lock() {
2387 events.push(event);
2388 }
2389 }
2390}
2391
2392#[derive(Debug)]
2395pub struct TeeEvents(pub Arc<dyn EventSink>, pub Arc<dyn EventSink>);
2396
2397impl EventSink for TeeEvents {
2398 fn emit(&self, event: LifecycleEvent) {
2399 self.0.emit(event);
2400 self.1.emit(event);
2401 }
2402}
2403
2404pub trait DemandDirector: fmt::Debug + Send + Sync {
2419 fn directed_demand(&self, policy: &ScalePolicy) -> Option<u32>;
2422
2423 fn directed_exclusions(&self, _policy: &ScalePolicy) -> BTreeSet<Label> {
2426 BTreeSet::new()
2427 }
2428
2429 fn launched(&self, policy: &ScalePolicy, attempt: AttemptId);
2432
2433 fn start_failed(&self, _policy: &ScalePolicy, _attempt: Option<AttemptId>) {}
2438
2439 fn next_waiting(&self, _policy: &ScalePolicy) -> Option<u64> {
2442 None
2443 }
2444
2445 fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
2449 0
2450 }
2451
2452 fn pass_over_older(&self, _policy: &ScalePolicy) {}
2455}
2456
2457const OLDER_LAUNCH_TURN: Duration = Duration::from_secs(2);
2464const OLDER_LAUNCH_POLL: Duration = Duration::from_millis(10);
2465
2466const LAUNCH_RETRY: Duration = Duration::from_secs(30);
2468
2469#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
2471pub struct ReservationWaits {
2472 pub order: Duration,
2474 pub lock: Duration,
2476}
2477
2478#[derive(Default)]
2480struct ReservationRound {
2481 reserved: BTreeMap<PolicyId, (u32, u32)>,
2484 exhausted: BTreeSet<PolicyId>,
2485}
2486
2487type Reserving<'a> = std::pin::Pin<
2489 Box<
2490 dyn std::future::Future<Output = (Option<Reserved>, ReservationRound, Vec<ScalePolicy>)>
2491 + Send
2492 + 'a,
2493 >,
2494>;
2495
2496struct Reserved {
2498 policy: ScalePolicy,
2499 attempt: RunnerAttempt,
2500 waits: ReservationWaits,
2501 reserved_at: std::time::Instant,
2502 _hold: LaunchHold,
2504}
2505
2506pub trait MemoryGate: fmt::Debug + Send + Sync {
2511 fn admit(&self, policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict;
2514
2515 fn admitted(&self, _policy: &ScalePolicy) {}
2518}
2519
2520#[derive(Debug, Default, Clone, Copy)]
2522pub struct NoMemoryGate;
2523
2524impl MemoryGate for NoMemoryGate {
2525 fn admit(&self, _policy: &ScalePolicy, _attempts: &[RunnerAttempt]) -> MemoryVerdict {
2526 MemoryVerdict::Unchecked
2527 }
2528}
2529
2530#[derive(Debug, Default, Clone, Copy)]
2532pub struct OwnDemand;
2533
2534impl DemandDirector for OwnDemand {
2535 fn directed_demand(&self, _policy: &ScalePolicy) -> Option<u32> {
2536 None
2537 }
2538
2539 fn launched(&self, _policy: &ScalePolicy, _attempt: AttemptId) {}
2540}
2541
2542#[derive(Clone)]
2548pub struct ReconcilerPorts {
2549 pub demand: Arc<dyn DemandSource>,
2550 pub launcher: Arc<dyn RunnerLauncher>,
2551 pub lock: Arc<dyn AllocationLock>,
2552 pub directory: Arc<dyn RepositoryDirectory>,
2553 pub clock: Arc<dyn Clock>,
2554 pub jitter: Arc<dyn Jitter>,
2555 pub events: Arc<dyn EventSink>,
2556}
2557
2558impl fmt::Debug for ReconcilerPorts {
2559 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2560 f.debug_struct("ReconcilerPorts").finish_non_exhaustive()
2561 }
2562}
2563
2564#[derive(Debug, Clone, Default)]
2571pub struct ReconcileReport {
2572 pub allocations: Vec<Allocation>,
2574 pub monitor_only: Vec<PolicyId>,
2576 pub unreadable: Vec<PolicyId>,
2578 pub targets_read: u16,
2586 pub started: u16,
2588 pub replacement_intents: u16,
2591 pub cleaned: u16,
2593 pub idle_exits: u16,
2596 pub failures: u16,
2598 pub deferred: u16,
2607 pub memory_deferred: u16,
2610 pub attempts_unreadable: u16,
2629 pub host_capacity_unreadable: u16,
2631 pub clean_failures: u16,
2633 pub failure: Option<RefreshState>,
2635 pub offline: Option<OfflineState>,
2638 pub next_poll: NextPoll,
2640 pub demand_requests: u32,
2642}
2643
2644impl ReconcileReport {
2645 #[must_use]
2672 pub const fn reached_github(&self) -> bool {
2673 self.targets_read > 0
2674 }
2675}
2676
2677impl Default for NextPoll {
2678 fn default() -> Self {
2679 Self {
2680 delay: PollSchedule::floor(),
2681 pace: PollPace::Nominal,
2682 }
2683 }
2684}
2685
2686impl ReconcileReport {
2687 #[must_use]
2689 pub fn is_offline(&self) -> bool {
2690 matches!(self.failure, Some(RefreshState::Offline))
2691 }
2692
2693 #[must_use]
2695 pub const fn offline_state(&self) -> Option<&OfflineState> {
2696 self.offline.as_ref()
2697 }
2698
2699 #[must_use]
2701 pub const fn starts_nothing(&self) -> bool {
2702 self.started == 0
2703 }
2704}
2705
2706#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
2708pub struct ScaleDownReport {
2709 pub removed: u16,
2711 pub refused_busy: u16,
2713 pub clean_failures: u16,
2715 pub retained: u16,
2718 pub attempts_unreadable: bool,
2728}
2729
2730impl ScaleDownReport {
2731 #[must_use]
2736 pub const fn is_conclusive(&self) -> bool {
2737 !self.attempts_unreadable
2738 }
2739}
2740
2741const CLEAN_RETRY_INITIAL: Duration = Duration::from_secs(30);
2749
2750const CLEAN_RETRY_MAXIMUM: Duration = Duration::from_secs(30 * 60);
2752
2753#[derive(Debug, Clone, Copy)]
2755struct CleanRetry {
2756 failures: u32,
2757 not_before: Timestamp,
2758}
2759
2760#[derive(Debug, Default)]
2764pub(crate) struct CleanBackoff {
2765 retries: BTreeMap<AttemptId, CleanRetry>,
2766}
2767
2768impl CleanBackoff {
2769 const SCHEDULE: crate::lifecycle::RetryPolicy =
2771 crate::lifecycle::RetryPolicy::bounded(u32::MAX, CLEAN_RETRY_INITIAL, CLEAN_RETRY_MAXIMUM);
2772
2773 pub(crate) fn is_due(&self, attempt: AttemptId, now: Timestamp) -> bool {
2775 self.retries
2776 .get(&attempt)
2777 .is_none_or(|retry| now >= retry.not_before)
2778 }
2779
2780 pub(crate) fn failed(&mut self, attempt: AttemptId, now: Timestamp) {
2782 let failures = self
2783 .retries
2784 .get(&attempt)
2785 .map_or(1, |retry| retry.failures.saturating_add(1));
2786 let delay = Self::SCHEDULE.delay(failures);
2787 self.retries.insert(
2788 attempt,
2789 CleanRetry {
2790 failures,
2791 not_before: now + chrono::Duration::from_std(delay).unwrap_or_default(),
2793 },
2794 );
2795 }
2796
2797 pub(crate) fn succeeded(&mut self, attempt: AttemptId) {
2799 self.retries.remove(&attempt);
2800 }
2801
2802 fn retain(&mut self, present: &BTreeSet<AttemptId>) {
2805 self.retries.retain(|attempt, _| present.contains(attempt));
2806 }
2807}
2808
2809#[derive(Debug)]
2815pub struct Reconciler {
2816 host: Host,
2817 demand: Arc<dyn DemandSource>,
2818 launcher: Arc<dyn RunnerLauncher>,
2819 lock: Arc<dyn AllocationLock>,
2820 repositories: RepositoryCache,
2821 clock: Arc<dyn Clock>,
2822 jitter: Arc<dyn Jitter>,
2823 events: Arc<dyn EventSink>,
2824 schedule: PollSchedule,
2825 clean_backoff: CleanBackoff,
2826 director: Arc<dyn DemandDirector>,
2827 launch_side: Option<Arc<tokio::sync::Notify>>,
2830 memory: Arc<dyn MemoryGate>,
2831 host_labels: Vec<Label>,
2834 seen_exclusions: Mutex<BTreeMap<PolicyId, BTreeSet<Label>>>,
2837}
2838
2839impl Reconciler {
2840 #[must_use]
2842 pub fn new(host: Host, ports: ReconcilerPorts) -> Self {
2843 let interval = host.refresh_interval;
2844 let repositories = RepositoryCache::new(
2845 Arc::clone(&ports.directory),
2846 Arc::clone(&ports.clock),
2847 interval,
2848 );
2849 Self {
2850 host,
2851 demand: ports.demand,
2852 launcher: ports.launcher,
2853 lock: ports.lock,
2854 repositories,
2855 clock: ports.clock,
2856 jitter: ports.jitter,
2857 events: ports.events,
2858 schedule: PollSchedule::new(interval),
2859 clean_backoff: CleanBackoff::default(),
2860 director: Arc::new(OwnDemand),
2861 launch_side: None,
2862 memory: Arc::new(NoMemoryGate),
2863 host_labels: Vec::new(),
2864 seen_exclusions: Mutex::default(),
2865 }
2866 }
2867
2868 #[must_use]
2871 pub fn with_host_traits(mut self, traits: HostTraits) -> Self {
2872 self.host_labels = traits.labels();
2873 self
2874 }
2875
2876 #[must_use]
2878 pub fn with_memory_gate(mut self, memory: Arc<dyn MemoryGate>) -> Self {
2879 self.memory = memory;
2880 self
2881 }
2882
2883 #[must_use]
2888 pub fn with_launch_side(mut self, signal: Arc<tokio::sync::Notify>) -> Self {
2889 self.launch_side = Some(signal);
2890 self
2891 }
2892
2893 #[must_use]
2896 pub fn with_director(mut self, director: Arc<dyn DemandDirector>) -> Self {
2897 self.director = director;
2898 self
2899 }
2900
2901 #[must_use]
2902 pub const fn host(&self) -> &Host {
2903 &self.host
2904 }
2905
2906 #[must_use]
2907 pub const fn schedule(&self) -> &PollSchedule {
2908 &self.schedule
2909 }
2910
2911 pub fn set_intervals(&mut self, active: RefreshInterval, idle: Duration) {
2913 self.host.refresh_interval = active;
2914 self.schedule.set_intervals(active, idle);
2915 }
2916
2917 #[must_use]
2919 pub const fn repositories(&self) -> &RepositoryCache {
2920 &self.repositories
2921 }
2922
2923 #[cfg(test)]
2927 pub async fn start_directed(&self, policies: &[ScalePolicy]) -> ReconcileReport {
2928 let mut report = ReconcileReport::default();
2929 let mut round = ReservationRound::default();
2930 let mut reserved = Vec::new();
2931 while let Some(next) = self
2932 .reserve_next_launch(policies, &mut round, &mut report)
2933 .await
2934 {
2935 reserved.push(next);
2936 }
2937 self.finish_reserved(reserved, &mut report).await;
2938 report
2939 }
2940
2941 pub async fn serve_launches(
2957 &self,
2958 policies: &(dyn Fn() -> Vec<ScalePolicy> + Send + Sync),
2959 signal: &tokio::sync::Notify,
2960 mut stop: tokio::sync::watch::Receiver<bool>,
2961 ) {
2962 use futures::StreamExt as _;
2963
2964 let mut bringing_up = futures::stream::FuturesUnordered::new();
2965 let mut reserving: Option<Reserving<'_>> = None;
2966 let mut retry_at: Option<tokio::time::Instant> = None;
2967 let mut known: Vec<ScalePolicy> = Vec::new();
2968 let mut stopping = *stop.borrow();
2969 loop {
2970 if stopping && reserving.is_none() && bringing_up.is_empty() {
2971 return;
2972 }
2973 tokio::select! {
2974 () = signal.notified(), if !stopping && reserving.is_none() => {
2975 reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
2976 }
2977 () = async {
2978 match retry_at {
2979 Some(at) => tokio::time::sleep_until(at).await,
2980 None => std::future::pending().await,
2981 }
2982 }, if !stopping && reserving.is_none() => {
2983 retry_at = None;
2984 reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
2985 }
2986 (reserved, round, read) = async {
2987 match reserving.as_mut() {
2988 Some(reserving) => reserving.await,
2989 None => std::future::pending().await,
2990 }
2991 } => {
2992 reserving = None;
2993 match reserved {
2994 Some(reserved) => {
2995 bringing_up.push(self.bring_up(reserved));
2996 known = read;
2997 if !stopping {
2998 reserving = Some(self.reserve_step(policies(), round));
2999 }
3000 }
3001 None => known = read,
3002 }
3003 }
3004 Some(_) = bringing_up.next(), if !bringing_up.is_empty() => {}
3005 changed = stop.changed(), if !stopping => {
3006 stopping = changed.is_err() || *stop.borrow();
3007 }
3008 }
3009 if reserving.is_none()
3012 && retry_at.is_none()
3013 && known
3014 .iter()
3015 .any(|policy| self.director.next_waiting(policy).is_some())
3016 {
3017 retry_at = Some(tokio::time::Instant::now() + LAUNCH_RETRY);
3018 }
3019 }
3020 }
3021
3022 fn reserve_step(
3024 &self,
3025 policies: Vec<ScalePolicy>,
3026 mut round: ReservationRound,
3027 ) -> Reserving<'_> {
3028 Box::pin(async move {
3029 let mut report = ReconcileReport::default();
3030 let reserved = self
3031 .reserve_next_launch(&policies, &mut round, &mut report)
3032 .await;
3033 (reserved, round, policies)
3034 })
3035 }
3036
3037 async fn reserve_next_launch(
3042 &self,
3043 policies: &[ScalePolicy],
3044 round: &mut ReservationRound,
3045 report: &mut ReconcileReport,
3046 ) -> Option<Reserved> {
3047 loop {
3048 let policy = policies
3049 .iter()
3050 .filter(|policy| {
3051 policy.owns_runners()
3052 && policy.is_owned_by(self.host.id)
3053 && policy.may_start_runners()
3054 && !round.exhausted.contains(&policy.id)
3055 })
3056 .min_by_key(|policy| self.director.next_waiting(policy).unwrap_or(u64::MAX))?;
3057 let (served, most) = round.reserved.entry(policy.id).or_default();
3058 if let Some(directed) = self.director.directed_demand(policy)
3060 && *served < directed.max(*most)
3061 {
3062 *most = directed.max(*most);
3063 let mut reserved = Vec::new();
3064 let excluded = self.director.directed_exclusions(policy);
3065 self.start_runners(policy, directed, 0, &excluded, 1, report, &mut reserved)
3066 .await;
3067 if let Some(next) = reserved.pop() {
3068 *served += 1;
3069 return Some(next);
3070 }
3071 }
3072 round.exhausted.insert(policy.id);
3073 }
3074 }
3075
3076 pub async fn reconcile(&mut self, policies: &[ScalePolicy]) -> ReconcileReport {
3090 let mut report = ReconcileReport::default();
3091 let mut reserved: Vec<Reserved> = Vec::new();
3095
3096 let mut pollable: Vec<&ScalePolicy> = Vec::new();
3098 let mut supervision_failed = BTreeSet::new();
3099 for policy in policies {
3100 if !policy.owns_runners() {
3101 report.monitor_only.push(policy.id);
3102 self.events
3103 .emit(LifecycleEvent::MonitorOnlySkipped { policy: policy.id });
3104 continue;
3105 }
3106 if !policy.is_owned_by(self.host.id) {
3107 continue;
3111 }
3112 match self.launcher.supervise(policy).await {
3113 Ok(intents) => {
3114 report.replacement_intents = report
3115 .replacement_intents
3116 .saturating_add(u16::try_from(intents.len()).unwrap_or(u16::MAX));
3117 }
3118 Err(failure) => {
3119 supervision_failed.insert(policy.id);
3120 self.report_unreadable_attempts(&mut report, &failure);
3121 continue;
3122 }
3123 }
3124 if !policy.may_start_runners() {
3125 continue;
3126 }
3127 pollable.push(policy);
3128 }
3129
3130 let readings = self.poll_targets(&pollable, &mut report).await;
3131 report.targets_read = u16::try_from(
3132 readings
3133 .values()
3134 .filter(|outcome| matches!(outcome, PollOutcome::Ready(_)))
3135 .count(),
3136 )
3137 .unwrap_or(u16::MAX);
3138 let mut routed = BTreeMap::new();
3139 for (target, outcome) in &readings {
3140 let PollOutcome::Ready(reading) = outcome else {
3141 continue;
3142 };
3143 let profiles: Vec<&ScalePolicy> = pollable
3144 .iter()
3145 .copied()
3146 .filter(|policy| &policy.target == target)
3147 .collect();
3148 let (tallies, refusals) = route_demand(policies, &profiles, reading, &self.host_labels);
3149 if let Some(first) = profiles.first() {
3150 for (reason, count) in refusals {
3151 self.events.emit(LifecycleEvent::RoutingRefused {
3152 policy: first.id,
3153 reason,
3154 count,
3155 });
3156 }
3157 }
3158 routed.insert(target.clone(), tallies);
3159 }
3160
3161 self.clean_terminal_attempts(&mut report).await;
3171
3172 for policy in policies {
3187 if !policy.owns_runners() {
3188 continue;
3194 }
3195 if !policy.is_owned_by(self.host.id) || !policy.may_start_runners() {
3196 match self.allocate_only(policy, 0, &reserved).await {
3200 Ok(allocation) => {
3201 self.emit_allocation(&allocation);
3202 report.allocations.push(allocation);
3203 }
3204 Err(failure) => self.report_allocation_failure(&mut report, &failure),
3205 }
3206 continue;
3207 }
3208 if supervision_failed.contains(&policy.id) {
3209 continue;
3210 }
3211 if let Some(directed) = self.director.directed_demand(policy) {
3220 let own = match readings.get(&policy.target) {
3221 Some(PollOutcome::Ready(reading)) => routed
3222 .get(&policy.target)
3223 .and_then(|tallies| tallies.get(&policy.id))
3224 .filter(|tally| tally.non_pool_demand() > 0)
3225 .map(|tally| (tally, reading)),
3226 _ => None,
3227 };
3228 if own.is_some() || self.launch_side.is_none() {
3229 let mut excluded = self.director.directed_exclusions(policy);
3237 let own_demand = own.map_or(0, |(tally, reading)| {
3238 let count = tally.non_pool_demand();
3239 self.observe_demand(policy, tally, count, reading);
3240 excluded.extend(tally.excluded.iter().cloned());
3241 count
3242 });
3243 self.start_runners(
3244 policy,
3245 directed,
3246 own_demand,
3247 &excluded,
3248 u16::MAX,
3249 &mut report,
3250 &mut reserved,
3251 )
3252 .await;
3253 continue;
3254 }
3255 let Some(launch_side) = &self.launch_side else {
3256 continue;
3257 };
3258 match self.allocate_only(policy, directed, &reserved).await {
3259 Ok(allocation) => {
3260 if !allocation.starts_nothing() {
3261 launch_side.notify_one();
3262 }
3263 report.allocations.push(allocation);
3264 }
3265 Err(failure) => self.report_allocation_failure(&mut report, &failure),
3266 }
3267 continue;
3268 }
3269 let Some(reading) = readings.get(&policy.target) else {
3270 debug_assert!(false, "a pollable policy's target has no reading");
3273 continue;
3274 };
3275 match reading {
3276 PollOutcome::Failed(state) => {
3277 report.unreadable.push(policy.id);
3278 self.events.emit(LifecycleEvent::TargetUnreadable {
3279 policy: policy.id,
3280 reason: unreadable_reason(state),
3281 });
3282 }
3283 PollOutcome::Ready(demand) => {
3284 let tally = routed
3285 .get(&policy.target)
3286 .and_then(|tallies| tallies.get(&policy.id))
3287 .expect("every pollable profile has a routed tally");
3288 let count = tally.demand();
3289 self.observe_demand(policy, tally, count, demand);
3290 self.start_runners(
3291 policy,
3292 count,
3293 0,
3294 &tally.excluded,
3295 u16::MAX,
3296 &mut report,
3297 &mut reserved,
3298 )
3299 .await;
3300 }
3301 }
3302 }
3303 self.finish_reserved(reserved, &mut report).await;
3304
3305 let failure = readings
3307 .values()
3308 .filter_map(PollOutcome::failure)
3309 .max_by_key(|state| severity(state))
3310 .cloned();
3311 let now = self.clock.now();
3312 let conditions = PollConditions {
3313 quiet: is_quiet(&readings, &report),
3314 requests_per_pass: report.demand_requests,
3315 quota: readings
3316 .values()
3317 .filter_map(PollOutcome::reading)
3318 .filter_map(QueuedDemand::rate_limit)
3319 .max_by_key(|quota| quota.observed_at)
3320 .copied(),
3321 };
3322 report.next_poll =
3323 self.schedule
3324 .next_poll_with(failure.as_ref(), conditions, now, self.jitter.as_ref());
3325 report.failure = failure;
3326 if let PollPace::Offline { consecutive } = report.next_poll.pace {
3330 let state = OfflineState::new(consecutive, report.next_poll.delay);
3331 report.offline = Some(match self.schedule.offline_for(now) {
3332 Some(elapsed) => state.since(elapsed),
3333 None => state,
3334 });
3335 }
3336 self.events.emit(LifecycleEvent::PollScheduled {
3337 retry_in_ms: u64::try_from(report.next_poll.delay.as_millis()).unwrap_or(u64::MAX),
3338 pace: report.next_poll.pace,
3339 });
3340
3341 report
3342 }
3343
3344 async fn poll_targets(
3352 &self,
3353 pollable: &[&ScalePolicy],
3354 report: &mut ReconcileReport,
3355 ) -> BTreeMap<ScaleTarget, PollOutcome> {
3356 let targets: BTreeSet<ScaleTarget> = pollable.iter().map(|p| p.target.clone()).collect();
3357
3358 let mut readings = BTreeMap::new();
3359 for target in targets {
3360 let outcome = match self.repositories.scope_for(&target).await {
3361 Ok(scope) => {
3362 report.demand_requests = report
3363 .demand_requests
3364 .saturating_add(demand_requests_per_poll(&scope));
3365 self.demand.poll(&scope).await
3366 }
3367 Err(error) => PollOutcome::Failed(RefreshState::from_error(&error)),
3373 };
3374 readings.insert(target, outcome);
3375 }
3376 readings
3377 }
3378
3379 async fn allocate_only(
3397 &self,
3398 policy: &ScalePolicy,
3399 demand: u32,
3400 reserved: &[Reserved],
3401 ) -> Result<Allocation, AllocationFailure> {
3402 self.allocate_over(policy, demand, reserved)
3403 .await
3404 .map(|(allocation, _)| allocation)
3405 }
3406
3407 async fn allocate_over(
3409 &self,
3410 policy: &ScalePolicy,
3411 demand: u32,
3412 reserved: &[Reserved],
3413 ) -> Result<(Allocation, Vec<RunnerAttempt>), AllocationFailure> {
3414 let mut host = self.host.clone();
3417 host.host_capacity = self
3418 .launcher
3419 .host_capacity(&self.host)
3420 .await
3421 .map_err(AllocationFailure::HostCapacity)?;
3422 let attempts = self
3423 .host_attempts(reserved)
3424 .await
3425 .map_err(AllocationFailure::Attempts)?;
3426 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
3427 let allocation = allocator.allocate(policy, demand);
3428 Ok((allocation, attempts))
3429 }
3430
3431 #[allow(clippy::too_many_arguments)]
3465 async fn start_runners(
3466 &self,
3467 policy: &ScalePolicy,
3468 demand: u32,
3469 own: u32,
3470 excluded: &BTreeSet<Label>,
3471 limit: u16,
3472 report: &mut ReconcileReport,
3473 reserved: &mut Vec<Reserved>,
3474 ) {
3475 let excluded = {
3479 let mut seen = self
3480 .seen_exclusions
3481 .lock()
3482 .unwrap_or_else(std::sync::PoisonError::into_inner);
3483 let kept = seen.entry(policy.id).or_default();
3484 kept.extend(excluded.iter().cloned());
3485 kept.clone()
3486 };
3487 let registered = policy.with_runtime_labels(|labels| {
3488 labels
3489 .with_labels(self.host_labels.iter().cloned())
3490 .excluding(&excluded)
3491 });
3492 let policy = ®istered;
3493 let intent = match self
3505 .allocate_only(policy, demand.saturating_add(own), reserved)
3506 .await
3507 {
3508 Ok(intent) => intent,
3509 Err(failure) => {
3510 self.report_allocation_failure(report, &failure);
3511 self.director.start_failed(policy, None);
3512 return;
3513 }
3514 };
3515
3516 let mut decided: Option<Allocation> = None;
3520 let mut budget = intent.to_start.min(limit);
3521
3522 while budget > 0 {
3523 let order_wait = self.let_older_launches_reserve(policy).await;
3524 let demand = if order_wait.is_zero() {
3526 demand
3527 } else {
3528 self.director.directed_demand(policy).unwrap_or(demand)
3529 }
3530 .saturating_add(own);
3531 let waiting_since = std::time::Instant::now();
3532 let guard = match self.lock.acquire().await {
3533 Ok(guard) => guard,
3534 Err(_) => {
3535 report.deferred = report.deferred.saturating_add(budget);
3538 self.events.emit(LifecycleEvent::AllocationDeferred {
3539 policy: policy.id,
3540 count: budget,
3541 });
3542 self.director.start_failed(policy, None);
3545 break;
3546 }
3547 };
3548 let lock_wait = waiting_since.elapsed();
3549
3550 let (allocation, attempts) = match self.allocate_over(policy, demand, reserved).await {
3556 Ok(decided) => decided,
3557 Err(failure) => {
3558 drop(guard);
3559 self.report_allocation_failure(report, &failure);
3560 self.director.start_failed(policy, None);
3561 break;
3562 }
3563 };
3564
3565 if decided.is_none() {
3566 budget = budget.min(allocation.to_start);
3570 decided = Some(allocation.clone());
3571 }
3572
3573 if allocation.starts_nothing() || budget == 0 {
3576 drop(guard);
3577 break;
3578 }
3579
3580 match self.memory.admit(policy, &attempts) {
3584 MemoryVerdict::Defer(figures) => {
3585 drop(guard);
3586 report.memory_deferred = report.memory_deferred.saturating_add(budget);
3587 self.events.emit(LifecycleEvent::MemoryShort {
3588 policy: policy.id,
3589 figures,
3590 started: false,
3591 });
3592 self.director.start_failed(policy, None);
3596 break;
3597 }
3598 MemoryVerdict::AdmitAlone(figures) => {
3599 self.events.emit(LifecycleEvent::MemoryShort {
3600 policy: policy.id,
3601 figures,
3602 started: true,
3603 });
3604 self.memory.admitted(policy);
3605 }
3606 MemoryVerdict::Admit(_) | MemoryVerdict::Unchecked => self.memory.admitted(policy),
3607 }
3608
3609 let created = self
3610 .launcher
3611 .launch(LaunchRequest {
3612 host: &self.host,
3613 policy,
3614 allocation_guard: &guard,
3615 })
3616 .await;
3617 let hold = guard.launch_hold();
3618 drop(guard);
3619
3620 match created {
3621 Ok(attempt) => {
3622 let id = attempt.id;
3623 self.director.launched(policy, id);
3626 budget -= 1;
3627 reserved.push(Reserved {
3631 policy: policy.clone(),
3632 attempt,
3633 waits: ReservationWaits {
3634 order: order_wait,
3635 lock: lock_wait,
3636 },
3637 reserved_at: std::time::Instant::now(),
3638 _hold: hold,
3639 });
3640 }
3641 Err(failure) => {
3642 self.note_start_failed(policy, None, &failure);
3643 break;
3644 }
3645 }
3646 }
3647
3648 let allocation = decided.unwrap_or(intent);
3649 self.emit_allocation(&allocation);
3650 report.allocations.push(allocation);
3651 }
3652
3653 fn observe_demand(
3656 &self,
3657 policy: &ScalePolicy,
3658 tally: &DemandTally,
3659 demand: u32,
3660 reading: &QueuedDemand,
3661 ) {
3662 self.events.emit(LifecycleEvent::DemandObserved {
3663 policy: policy.id,
3664 demand,
3665 not_matched: tally.not_matched,
3666 unresolvable: u32::try_from(tally.unresolvable.len()).unwrap_or(u32::MAX),
3667 complete: reading.is_complete(),
3668 });
3669 }
3670
3671 async fn let_older_launches_reserve(&self, policy: &ScalePolicy) -> Duration {
3677 let began = tokio::time::Instant::now();
3678 let mut older = self.director.older_waiting(policy);
3679 let mut limit = began + OLDER_LAUNCH_TURN;
3680 while older > 0 {
3681 if tokio::time::Instant::now() >= limit {
3682 tracing::warn!(
3683 event = "older_launch_passed_over",
3684 policy_id = %policy.id,
3685 count = older,
3686 );
3687 self.director.pass_over_older(policy);
3688 break;
3689 }
3690 tokio::time::sleep(OLDER_LAUNCH_POLL).await;
3691 let still = self.director.older_waiting(policy);
3692 if still < older {
3693 limit = tokio::time::Instant::now() + OLDER_LAUNCH_TURN;
3694 }
3695 older = still;
3696 }
3697 began.elapsed()
3698 }
3699
3700 async fn finish_reserved(&self, reserved: Vec<Reserved>, report: &mut ReconcileReport) {
3708 let finished =
3709 futures::future::join_all(reserved.into_iter().map(|reserved| self.bring_up(reserved)))
3710 .await;
3711 let started = finished.into_iter().filter(|started| *started).count();
3712 report.started = report
3713 .started
3714 .saturating_add(u16::try_from(started).unwrap_or(u16::MAX));
3715 }
3716
3717 async fn bring_up(&self, reserved: Reserved) -> bool {
3719 let id = reserved.attempt.id;
3720 let mut waits = reserved.waits;
3721 waits.lock += reserved.reserved_at.elapsed();
3722 let result = self
3723 .launcher
3724 .finish_launch(&reserved.policy, reserved.attempt, waits)
3725 .await;
3726 match &result {
3729 Ok(_) => self.events.emit(LifecycleEvent::RunnerStarted {
3730 policy: reserved.policy.id,
3731 attempt: id,
3732 }),
3733 Err(failure) => self.note_start_failed(&reserved.policy, Some(id), failure),
3734 }
3735 result.is_ok()
3736 }
3737
3738 fn note_start_failed(
3739 &self,
3740 policy: &ScalePolicy,
3741 attempt: Option<AttemptId>,
3742 failure: &LaunchFailure,
3743 ) {
3744 self.events.emit(LifecycleEvent::RunnerStartFailed {
3745 policy: policy.id,
3746 reason: failure_reason_kind(&failure.reason),
3747 });
3748 self.director.start_failed(policy, attempt);
3749 }
3750
3751 async fn host_attempts(
3764 &self,
3765 reserved: &[Reserved],
3766 ) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
3767 let mut attempts = self.launcher.attempts().await?;
3768
3769 debug_assert!(
3776 reserved
3777 .iter()
3778 .map(|reserved| reserved.attempt.id)
3779 .collect::<BTreeSet<AttemptId>>()
3780 .len()
3781 == reserved.len(),
3782 "`RunnerLauncher::launch` returned an AttemptId this pass had already seen; \
3783 the host ceiling is enforced against a set keyed on that identifier, so a \
3784 duplicate is two runtimes counted as one"
3785 );
3786
3787 let known: BTreeSet<AttemptId> = attempts.iter().map(|attempt| attempt.id).collect();
3788 attempts.extend(
3789 reserved
3790 .iter()
3791 .filter(|reserved| !known.contains(&reserved.attempt.id))
3792 .map(|reserved| reserved.attempt.clone()),
3793 );
3794 Ok(attempts)
3795 }
3796
3797 fn report_allocation_failure(&self, report: &mut ReconcileReport, failure: &AllocationFailure) {
3798 match failure {
3799 AllocationFailure::HostCapacity(reason) => {
3800 report.host_capacity_unreadable = report.host_capacity_unreadable.saturating_add(1);
3801 self.events
3802 .emit(LifecycleEvent::HostCapacityUnreadable { reason: *reason });
3803 }
3804 AllocationFailure::Attempts(failure) => {
3805 self.report_unreadable_attempts(report, failure)
3806 }
3807 }
3808 }
3809
3810 fn report_unreadable_attempts(&self, report: &mut ReconcileReport, failure: &LaunchFailure) {
3817 report.attempts_unreadable = report.attempts_unreadable.saturating_add(1);
3818 self.events.emit(LifecycleEvent::AttemptsUnreadable {
3825 reason: failure_reason_kind(&failure.reason),
3826 });
3827 }
3828
3829 async fn clean_terminal_attempts(&mut self, report: &mut ReconcileReport) {
3835 let attempts = match self.launcher.attempts().await {
3836 Ok(attempts) => attempts,
3837 Err(failure) => {
3838 self.report_unreadable_attempts(report, &failure);
3839 return;
3840 }
3841 };
3842 self.clean_backoff
3843 .retain(&attempts.iter().map(|attempt| attempt.id).collect());
3844 let now = self.clock.now();
3845 for attempt in attempts {
3846 if !attempt.state().is_concluded() {
3847 continue;
3848 }
3849 let Some(outcome) = attempt.outcome() else {
3850 continue;
3851 };
3852 if !self.clean_backoff.is_due(attempt.id, now) {
3855 continue;
3856 }
3857 let kind = OutcomeKind::of(outcome);
3858 match self.launcher.clean(attempt.id).await {
3859 Ok(()) => {
3860 self.clean_backoff.succeeded(attempt.id);
3861 report.cleaned = report.cleaned.saturating_add(1);
3862 if kind.is_failure() {
3863 report.failures = report.failures.saturating_add(1);
3864 } else if kind == OutcomeKind::IdleExit {
3865 report.idle_exits = report.idle_exits.saturating_add(1);
3870 }
3871 self.events.emit(LifecycleEvent::AttemptCleaned {
3872 policy: attempt.policy_id,
3873 attempt: attempt.id,
3874 outcome: kind,
3875 });
3876 }
3877 Err(failure) => {
3887 self.clean_backoff.failed(attempt.id, now);
3888 report.clean_failures = report.clean_failures.saturating_add(1);
3889 self.events.emit(LifecycleEvent::AttemptCleanFailed {
3890 policy: attempt.policy_id,
3891 attempt: attempt.id,
3892 reason: failure_reason_kind(&failure.reason),
3893 });
3894 }
3895 }
3896 }
3897 }
3898
3899 pub async fn scale_down(&self, policy: &ScalePolicy) -> ScaleDownReport {
3907 let mut report = ScaleDownReport::default();
3908 let attempts = match self.launcher.attempts().await {
3909 Ok(attempts) => attempts,
3910 Err(failure) => {
3911 self.events.emit(LifecycleEvent::AttemptsUnreadable {
3916 reason: failure_reason_kind(&failure.reason),
3917 });
3918 report.attempts_unreadable = true;
3919 return report;
3920 }
3921 };
3922 for attempt in attempts {
3923 if attempt.policy_id != policy.id {
3924 continue;
3925 }
3926 match attempt.state() {
3927 AttemptState::Busy => {
3928 report.refused_busy = report.refused_busy.saturating_add(1);
3929 self.events.emit(LifecycleEvent::ScaleDownRefused {
3930 policy: policy.id,
3931 attempt: attempt.id,
3932 });
3933 }
3934 state if state.is_concluded() => {
3935 let kind = attempt
3936 .outcome()
3937 .map_or(OutcomeKind::Failed, OutcomeKind::of);
3938 match self.launcher.clean(attempt.id).await {
3939 Ok(()) => {
3940 report.removed = report.removed.saturating_add(1);
3941 self.events.emit(LifecycleEvent::AttemptCleaned {
3942 policy: policy.id,
3943 attempt: attempt.id,
3944 outcome: kind,
3945 });
3946 }
3947 Err(failure) => {
3948 report.clean_failures = report.clean_failures.saturating_add(1);
3949 self.events.emit(LifecycleEvent::AttemptCleanFailed {
3950 policy: policy.id,
3951 attempt: attempt.id,
3952 reason: failure_reason_kind(&failure.reason),
3953 });
3954 }
3955 }
3956 }
3957 AttemptState::Cleaned => {}
3958 _ => report.retained = report.retained.saturating_add(1),
3961 }
3962 }
3963 report
3964 }
3965
3966 fn emit_allocation(&self, allocation: &Allocation) {
3967 self.events.emit(LifecycleEvent::Allocated {
3968 policy: allocation.policy_id,
3969 demand: allocation.demand,
3970 desired: allocation.desired,
3971 active_owned: allocation.active_owned,
3972 headroom: allocation.headroom_before,
3973 to_start: allocation.to_start,
3974 limiting: allocation.limiting_factor,
3975 });
3976 }
3977}
3978
3979fn demand_for(policy: &ScalePolicy, reading: &QueuedDemand, host_labels: &[Label]) -> DemandTally {
4003 let Some(labels) = policy.routing_labels() else {
4004 debug_assert!(
4005 false,
4006 "a monitor-only policy is skipped before the demand poll (D19)"
4007 );
4008 return DemandTally::default();
4009 };
4010 let labels = labels.with_labels(host_labels.iter().cloned());
4011
4012 match &policy.target {
4013 ScaleTarget::Repository(repository) => labels.tally(reading.jobs_for(repository)),
4014 ScaleTarget::Organization(_) => labels.tally(reading.jobs()),
4015 }
4016}
4017
4018fn route_demand(
4022 all_policies: &[ScalePolicy],
4023 profiles: &[&ScalePolicy],
4024 reading: &QueuedDemand,
4025 host_labels: &[Label],
4026) -> (
4027 BTreeMap<PolicyId, DemandTally>,
4028 BTreeMap<RoutingRefusal, u32>,
4029) {
4030 let mut tallies: BTreeMap<PolicyId, DemandTally> = profiles
4031 .iter()
4032 .map(|policy| (policy.id, demand_for(policy, reading, host_labels)))
4033 .collect();
4034 let mut refusals: BTreeMap<RoutingRefusal, u32> = BTreeMap::new();
4035 let Some(first) = profiles.first() else {
4036 return (tallies, refusals);
4037 };
4038 let owners: Vec<_> = all_policies
4040 .iter()
4041 .filter(|policy| policy.target == first.target && policy.host_id == first.host_id)
4042 .filter_map(|policy| {
4043 let labels = policy.routing_labels()?;
4044 Some((policy.id, labels.with_labels(host_labels.iter().cloned())))
4045 })
4046 .collect();
4047 let jobs: Vec<&runner_manager_domain::policy::RunsOn> = match &first.target {
4048 ScaleTarget::Repository(repository) => reading.jobs_for(repository).iter().collect(),
4049 ScaleTarget::Organization(_) => reading.jobs().collect(),
4050 };
4051 for job in jobs {
4052 if let Err(unresolvable) = job.required_labels() {
4053 if unresolvable.is_unsatisfiable_exclusion() {
4057 let meant = job.plain_labels();
4058 if !owners
4059 .iter()
4060 .any(|(_, labels)| meant.iter().all(|label| labels.contains(label)))
4061 {
4062 continue;
4063 }
4064 *refusals
4065 .entry(RoutingRefusal::UnsatisfiableExclusion)
4066 .or_default() += 1;
4067 }
4068 continue;
4069 }
4070 let matched: Vec<PolicyId> = owners
4071 .iter()
4072 .filter(|(_, labels)| labels.matches(job).is_match())
4073 .map(|(id, _)| *id)
4074 .collect();
4075 if matched.len() > 1 {
4076 *refusals
4077 .entry(RoutingRefusal::OverlappingProfiles)
4078 .or_default() += 1;
4079 let pool = job
4080 .required_labels()
4081 .is_ok_and(|required| is_pool_job(&required));
4082 for id in matched {
4083 if let Some(tally) = tallies.get_mut(&id) {
4084 tally.matched -= 1;
4085 tally.not_matched += 1;
4086 if pool {
4087 tally.pool_matched = tally.pool_matched.saturating_sub(1);
4088 }
4089 }
4090 }
4091 }
4092 }
4093 (tallies, refusals)
4094}
4095
4096const fn severity(state: &RefreshState) -> u8 {
4103 match state {
4104 RefreshState::Offline => 5,
4105 RefreshState::RateLimited(_) => 4,
4106 RefreshState::LockedOut { .. } => 3,
4107 RefreshState::Unauthorized => 2,
4108 RefreshState::Forbidden { .. } | RefreshState::Failed { .. } => 1,
4109 RefreshState::Cancelled | RefreshState::Ready(_) => 0,
4110 }
4111}
4112
4113const fn unreadable_reason(state: &RefreshState) -> &'static str {
4124 match state {
4125 RefreshState::Ready(_) => "ready",
4126 RefreshState::Offline => "offline",
4127 RefreshState::RateLimited(_) => "rate_limited",
4128 RefreshState::LockedOut { .. } => "locked_out",
4129 RefreshState::Unauthorized => "unauthorized",
4130 RefreshState::Forbidden { .. } => "forbidden",
4131 RefreshState::Failed { .. } => "failed",
4132 RefreshState::Cancelled => "cancelled",
4133 }
4134}
4135
4136#[cfg(test)]
4137mod tests {
4138 use super::*;
4139
4140 #[test]
4148 fn a_pass_that_reached_no_target_does_not_claim_it_reached_github() {
4149 let mut report = ReconcileReport::default();
4150 assert!(
4151 !report.reached_github(),
4152 "a pass that polled nothing -- every policy draining, owned elsewhere, or \
4153 monitor-only -- reached nobody. This is the case the old guard let through, and \
4154 the only one it ever let through."
4155 );
4156
4157 report.unreadable.push(PolicyId::from_u128(1));
4158 assert!(
4159 !report.reached_github(),
4160 "every target this pass tried was unreadable, so there is no contact to record"
4161 );
4162
4163 report.targets_read = 1;
4164 assert!(
4165 report.reached_github(),
4166 "one target answering is contact, whatever else failed alongside it"
4167 );
4168
4169 let mut unowned = ReconcileReport::default();
4173 unowned.unreadable.push(PolicyId::from_u128(2));
4174 unowned.allocations.push(Allocation {
4175 policy_id: PolicyId::from_u128(2),
4176 demand: 0,
4177 desired: 0,
4178 active_owned: 0,
4179 headroom_before: 0,
4180 to_start: 0,
4181 limiting_factor: LimitingFactor::Demand,
4182 });
4183 assert!(
4184 !unowned.reached_github(),
4185 "an allocation is not evidence that GitHub answered"
4186 );
4187 }
4188
4189 #[test]
4199 fn an_unauthorized_target_is_a_failure_and_not_merely_unreadable() {
4200 assert!(
4201 severity(&RefreshState::Unauthorized) > 0,
4202 "an unauthorized reading must survive `max_by_key(severity)` into `report.failure`, \
4203 or a pass where every target was refused would report no failure at all"
4204 );
4205 }
4206
4207 use std::sync::atomic::AtomicUsize;
4208
4209 use std::num::NonZeroU16;
4210
4211 use runner_manager_domain::attempt::PersistedAttempt;
4212 use runner_manager_domain::model::{CachePolicy, HostId};
4213 use runner_manager_domain::policy::PolicyMode;
4214 use runner_manager_domain::workspace::WorkspaceKind;
4215 use runner_manager_github::rest::RateLimited;
4216 use runner_manager_testkit::clock::FakeClock;
4217 use runner_manager_testkit::fixtures;
4218 use runner_manager_testkit::github::FakeGithub;
4219
4220 fn host_with(capacity: u16) -> Host {
4225 fixtures::host().capacity(capacity).build()
4226 }
4227
4228 fn repo(raw: &str) -> OwnerRepo {
4229 OwnerRepo::parse(raw).expect("a valid OWNER/REPO")
4230 }
4231
4232 use runner_manager_domain::policy::{RoutingLabels, RunsOn, pool_label};
4234
4235 fn policy(id: u128, target: &str, max: u16) -> ScalePolicy {
4236 fixtures::policy()
4237 .id(PolicyId::from_u128(id))
4238 .repository(target)
4239 .autoscale("home", max)
4240 .active()
4241 .build()
4242 }
4243
4244 const HOST_LABEL: &str = "rm-home-win-x64";
4251
4252 fn jobs(n: usize) -> Vec<RunsOn> {
4258 fixtures::queued_jobs(&[HOST_LABEL], n)
4259 }
4260
4261 #[derive(Debug, Default)]
4263 struct FakeLauncher {
4264 capacity: Mutex<Option<NonZeroU16>>,
4265 capacity_unreadable: AtomicBool,
4266 capacity_after_launch: Mutex<Option<NonZeroU16>>,
4267 attempts: Mutex<Vec<RunnerAttempt>>,
4268 next_id: AtomicU64,
4269 launches: AtomicUsize,
4270 cleaned: Mutex<Vec<AttemptId>>,
4271 yields_before_recording: usize,
4274 forgetful: bool,
4278 fail_next: Mutex<Option<FailureReason>>,
4279 attempts_fail: Mutex<bool>,
4282 clean_fails: bool,
4285 replacements: Mutex<Vec<ReplacementIntent>>,
4286 registered: Mutex<Vec<Vec<String>>>,
4288 }
4289
4290 impl FakeLauncher {
4291 fn new() -> Self {
4292 Self::default()
4293 }
4294
4295 fn with_yields(mut self, yields: usize) -> Self {
4296 self.yields_before_recording = yields;
4297 self
4298 }
4299
4300 fn forgetful() -> Self {
4301 Self {
4302 forgetful: true,
4303 ..Self::default()
4304 }
4305 }
4306
4307 fn seeded(self, attempts: Vec<RunnerAttempt>) -> Self {
4308 *self.attempts.lock().unwrap() = attempts;
4309 self
4310 }
4311
4312 fn launches(&self) -> usize {
4313 self.launches.load(Ordering::SeqCst)
4314 }
4315
4316 fn snapshot(&self) -> Vec<RunnerAttempt> {
4317 self.attempts.lock().unwrap().clone()
4318 }
4319
4320 fn live_count(&self) -> usize {
4321 self.snapshot()
4322 .iter()
4323 .filter(|a| a.counts_against_capacity())
4324 .count()
4325 }
4326
4327 fn fail_next(&self, reason: FailureReason) {
4328 *self.fail_next.lock().unwrap() = Some(reason);
4329 }
4330
4331 fn fail_attempts(&self, failing: bool) {
4332 *self.attempts_fail.lock().unwrap() = failing;
4333 }
4334
4335 fn refusing_cleanup(attempts: Vec<RunnerAttempt>) -> Self {
4336 Self {
4337 clean_fails: true,
4338 ..Self::default()
4339 }
4340 .seeded(attempts)
4341 }
4342
4343 fn cleaned(&self) -> Vec<AttemptId> {
4344 self.cleaned.lock().unwrap().clone()
4345 }
4346
4347 fn replacing(self, intent: ReplacementIntent) -> Self {
4348 self.replacements.lock().unwrap().push(intent);
4349 self
4350 }
4351 }
4352
4353 #[async_trait::async_trait]
4354 impl RunnerLauncher for FakeLauncher {
4355 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
4356 if self.capacity_unreadable.load(Ordering::SeqCst) {
4357 return Err(HostCapacityError::JournalUnreadable);
4358 }
4359 Ok(self.capacity.lock().unwrap().unwrap_or(host.host_capacity))
4360 }
4361
4362 async fn supervise(
4363 &self,
4364 policy: &ScalePolicy,
4365 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
4366 let mut replacements = self.replacements.lock().unwrap();
4367 let selected: Vec<_> = replacements
4368 .extract_if(.., |intent| intent.policy == policy.id)
4369 .collect();
4370 if !selected.is_empty() {
4371 let retired: BTreeSet<_> = selected
4372 .iter()
4373 .map(|intent| intent.previous_attempt)
4374 .collect();
4375 self.attempts
4376 .lock()
4377 .unwrap()
4378 .retain(|attempt| !retired.contains(&attempt.id));
4379 }
4380 Ok(selected)
4381 }
4382
4383 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
4384 if *self.attempts_fail.lock().unwrap() {
4385 return Err(LaunchFailure::new(FailureReason::Other(
4386 "the journal could not be read".into(),
4387 )));
4388 }
4389 Ok(self.snapshot())
4390 }
4391
4392 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
4393 if let Some(reason) = self.fail_next.lock().unwrap().take() {
4394 return Err(LaunchFailure::new(reason));
4395 }
4396 for _ in 0..self.yields_before_recording {
4398 tokio::task::yield_now().await;
4399 }
4400 let id =
4401 AttemptId::from_u128(u128::from(self.next_id.fetch_add(1, Ordering::SeqCst) + 1));
4402 let created = RunnerAttempt::allocate(
4403 id,
4404 request.policy.id,
4405 "runtime/p/a",
4406 request.host.created_at,
4407 );
4408 self.registered.lock().unwrap().push(
4409 request
4410 .policy
4411 .routing_labels()
4412 .map(RoutingLabels::as_registration_labels)
4413 .unwrap_or_default(),
4414 );
4415 self.launches.fetch_add(1, Ordering::SeqCst);
4416 if let Some(capacity) = self.capacity_after_launch.lock().unwrap().take() {
4417 *self.capacity.lock().unwrap() = Some(capacity);
4418 }
4419 if !self.forgetful {
4420 self.attempts.lock().unwrap().push(created.clone());
4421 }
4422 Ok(created)
4423 }
4424
4425 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
4426 if self.clean_fails {
4427 return Err(LaunchFailure::new(FailureReason::Other(
4428 "the runtime directory is locked".into(),
4429 )));
4430 }
4431 self.cleaned.lock().unwrap().push(attempt);
4432 let mut attempts = self.attempts.lock().unwrap();
4433 attempts.retain(|a| a.id != attempt);
4434 Ok(())
4435 }
4436 }
4437
4438 #[derive(Debug, Default)]
4440 struct FakeDemand {
4441 outcome: Mutex<Option<PollOutcome>>,
4442 per_target: Mutex<BTreeMap<ScaleTarget, PollOutcome>>,
4444 scopes: Mutex<Vec<ActivityScope>>,
4445 }
4446
4447 impl FakeDemand {
4448 fn ready(count: u32, repository: &OwnerRepo) -> Self {
4449 let fake = Self::default();
4450 fake.set(PollOutcome::Ready(QueuedDemand::of(
4451 repository.clone(),
4452 jobs(count as usize),
4453 )));
4454 fake
4455 }
4456
4457 fn failing(state: RefreshState) -> Self {
4458 let fake = Self::default();
4459 fake.set(PollOutcome::Failed(state));
4460 fake
4461 }
4462
4463 fn set(&self, outcome: PollOutcome) {
4464 *self.outcome.lock().unwrap() = Some(outcome);
4465 }
4466
4467 fn set_for(&self, target: &ScaleTarget, outcome: PollOutcome) {
4469 self.per_target
4470 .lock()
4471 .unwrap()
4472 .insert(target.clone(), outcome);
4473 }
4474
4475 fn polls(&self) -> Vec<ActivityScope> {
4476 self.scopes.lock().unwrap().clone()
4477 }
4478 }
4479
4480 #[async_trait::async_trait]
4481 impl DemandSource for FakeDemand {
4482 async fn poll(&self, scope: &ActivityScope) -> PollOutcome {
4483 self.scopes.lock().unwrap().push(scope.clone());
4484 if let Some(outcome) = self.per_target.lock().unwrap().get(scope.target()) {
4485 return outcome.clone();
4486 }
4487 self.outcome
4488 .lock()
4489 .unwrap()
4490 .clone()
4491 .unwrap_or(PollOutcome::Ready(QueuedDemand::default()))
4492 }
4493 }
4494
4495 #[derive(Debug, Default)]
4496 struct FakeDirectory {
4497 repositories: Vec<OwnerRepo>,
4498 calls: AtomicUsize,
4499 }
4500
4501 impl FakeDirectory {
4502 fn of(repositories: Vec<OwnerRepo>) -> Self {
4503 Self {
4504 repositories,
4505 calls: AtomicUsize::new(0),
4506 }
4507 }
4508
4509 fn calls(&self) -> usize {
4510 self.calls.load(Ordering::SeqCst)
4511 }
4512 }
4513
4514 #[async_trait::async_trait]
4515 impl RepositoryDirectory for FakeDirectory {
4516 async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
4517 self.calls.fetch_add(1, Ordering::SeqCst);
4518 Ok(self.repositories.clone())
4519 }
4520 }
4521
4522 #[derive(Debug)]
4528 struct CountingLock {
4529 inner: InProcessAllocationLock,
4530 concurrent: Arc<AtomicUsize>,
4531 peak: Arc<AtomicUsize>,
4532 acquisitions: Arc<AtomicUsize>,
4533 }
4534
4535 impl CountingLock {
4536 fn new() -> Self {
4537 Self {
4538 inner: InProcessAllocationLock::new(),
4539 concurrent: Arc::new(AtomicUsize::new(0)),
4540 peak: Arc::new(AtomicUsize::new(0)),
4541 acquisitions: Arc::new(AtomicUsize::new(0)),
4542 }
4543 }
4544
4545 fn peak(&self) -> usize {
4546 self.peak.load(Ordering::SeqCst)
4547 }
4548
4549 fn acquisitions(&self) -> usize {
4550 self.acquisitions.load(Ordering::SeqCst)
4551 }
4552 }
4553
4554 #[derive(Debug)]
4555 struct CountingGuard {
4556 _inner: AllocationGuard,
4557 concurrent: Arc<AtomicUsize>,
4558 }
4559
4560 impl Drop for CountingGuard {
4561 fn drop(&mut self) {
4562 self.concurrent.fetch_sub(1, Ordering::SeqCst);
4563 }
4564 }
4565
4566 #[async_trait::async_trait]
4567 impl AllocationLock for CountingLock {
4568 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4569 let inner = self.inner.acquire().await?;
4570 self.acquisitions.fetch_add(1, Ordering::SeqCst);
4571 let now = self.concurrent.fetch_add(1, Ordering::SeqCst) + 1;
4572 self.peak.fetch_max(now, Ordering::SeqCst);
4573 Ok(AllocationGuard::new(CountingGuard {
4574 _inner: inner,
4575 concurrent: Arc::clone(&self.concurrent),
4576 }))
4577 }
4578 }
4579
4580 #[derive(Debug, Default)]
4583 struct NoLock;
4584
4585 #[async_trait::async_trait]
4586 impl AllocationLock for NoLock {
4587 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4588 Ok(AllocationGuard::new(()))
4589 }
4590 }
4591
4592 #[tokio::test]
4593 async fn a_wsl_drain_request_fences_the_exact_pre_launch_boundary() {
4594 use runner_manager_platform::wsl::fence::DrainRequest;
4595
4596 let (_local, shared, paths) = guest_paths();
4597 DrainRequest::new(4, chrono::Utc::now())
4598 .write(shared.path())
4599 .unwrap();
4600 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4601
4602 assert!(lock.acquire().await.is_err());
4603 }
4604
4605 #[tokio::test]
4606 async fn a_guest_launch_claim_is_released_with_its_allocation_guard() {
4607 use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;
4608
4609 let (_local, shared, paths) = guest_paths();
4610 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4611 let guard = lock.acquire().await.unwrap();
4612 assert!(shared.path().join(FENCE_DIRECTORY).exists());
4613 drop(guard);
4614 assert!(!shared.path().join(FENCE_DIRECTORY).exists());
4615 }
4616
4617 #[derive(Clone, Default)]
4619 struct Logs(Arc<Mutex<Vec<u8>>>);
4620
4621 impl std::io::Write for Logs {
4622 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
4623 self.0.lock().unwrap().extend_from_slice(buf);
4624 Ok(buf.len())
4625 }
4626 fn flush(&mut self) -> std::io::Result<()> {
4627 Ok(())
4628 }
4629 }
4630
4631 #[tokio::test]
4634 async fn a_launch_fence_that_cannot_be_released_is_logged_with_its_path() {
4635 use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;
4636
4637 let (_local, shared, paths) = guest_paths();
4638 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4639 let guard = lock.acquire().await.unwrap();
4640 let fence = shared.path().join(FENCE_DIRECTORY);
4643 std::fs::remove_dir_all(&fence).unwrap();
4644 std::fs::write(&fence, b"not a directory").unwrap();
4645
4646 let logs = Logs::default();
4647 let writer = logs.clone();
4648 let subscriber = tracing_subscriber::fmt()
4649 .with_writer(move || writer.clone())
4650 .with_ansi(false)
4651 .finish();
4652 tracing::subscriber::with_default(subscriber, || drop(guard));
4653
4654 let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
4655 assert!(text.contains("WARN"), "{text}");
4656 assert!(text.contains("could not be released"), "{text}");
4657 assert!(text.contains(FENCE_DIRECTORY), "the path is named: {text}");
4658 }
4659
4660 fn guest_paths() -> (
4661 tempfile::TempDir,
4662 tempfile::TempDir,
4663 Arc<runner_manager_platform::paths::AppPaths>,
4664 ) {
4665 use runner_manager_platform::paths::AppPaths;
4666 use runner_manager_platform::wsl::fence::GuestRecoveryConfig;
4667
4668 let local = tempfile::tempdir().unwrap();
4669 let shared = tempfile::tempdir().unwrap();
4670 let paths = Arc::new(AppPaths::rooted_at(local.path()));
4671 paths.create_all().unwrap();
4672 GuestRecoveryConfig::new(shared.path().to_path_buf())
4673 .write(&paths)
4674 .unwrap();
4675 (local, shared, paths)
4676 }
4677
4678 #[tokio::test]
4681 async fn a_stale_guest_claim_no_longer_blocks_the_next_launch() {
4682 use runner_manager_platform::wsl::fence::{FENCE_DIRECTORY, OWNER_FILE};
4683
4684 let (_local, shared, paths) = guest_paths();
4685 let fence = shared.path().join(FENCE_DIRECTORY);
4686 std::fs::create_dir_all(&fence).unwrap();
4687 let pid = std::process::id();
4688 std::fs::write(
4689 fence.join(OWNER_FILE),
4690 format!(
4691 r#"{{"schema_version":1,"kind":"guest_launch","generation":null,"process_id":{pid},"acquired_at":"2026-09-24T00:47:18.724Z","identity":{{"pid":{pid},"start_token":"linux:earlier-boot:1"}}}}"#
4692 ),
4693 )
4694 .unwrap();
4695 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4696
4697 let guard = lock.acquire().await.expect("the stale claim was reclaimed");
4698 drop(guard);
4699 assert!(!fence.exists());
4700 }
4701
4702 #[tokio::test]
4703 async fn refusals_that_persist_are_recorded_and_a_grant_clears_them() {
4704 use runner_manager_platform::launch_health::recorded_launches_blocked;
4705 use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};
4706
4707 let (_local, shared, paths) = guest_paths();
4708 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4709 .with_blocked_after(3, Duration::ZERO);
4710 let windows =
4712 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4713 .unwrap()
4714 .unwrap();
4715
4716 for _ in 0..2 {
4717 assert!(lock.acquire().await.is_err());
4718 }
4719 assert_eq!(
4720 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4721 None,
4722 "not yet"
4723 );
4724 assert!(lock.acquire().await.is_err());
4725 let blocked = recorded_launches_blocked(&paths, chrono::Utc::now())
4726 .unwrap()
4727 .expect("three refusals in a row are recorded");
4728 assert!(
4729 blocked.reason.contains("Windows WSL recovery"),
4730 "{blocked:?}"
4731 );
4732
4733 windows.release().unwrap();
4734 let guard = lock.acquire().await.unwrap();
4735 assert_eq!(
4736 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4737 None
4738 );
4739 drop(guard);
4740 }
4741
4742 #[tokio::test]
4743 async fn refusals_while_this_process_is_launching_are_its_own_contention() {
4744 use runner_manager_platform::launch_health::recorded_launches_blocked;
4745
4746 use runner_manager_platform::wsl::fence::DrainRequest;
4747
4748 let (_local, shared, paths) = guest_paths();
4749 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4750 .with_blocked_after(1, Duration::ZERO);
4751 let launching = lock.acquire().await.unwrap();
4752 DrainRequest::new(1, chrono::Utc::now())
4753 .write(shared.path())
4754 .unwrap();
4755 for _ in 0..3 {
4756 assert!(lock.acquire().await.is_err());
4757 }
4758 assert_eq!(
4759 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4760 None
4761 );
4762 drop(launching);
4763 }
4764
4765 #[tokio::test]
4769 async fn launches_in_progress_share_the_fence_until_the_last_ends() {
4770 use runner_manager_platform::wsl::fence::{DrainRequest, FenceClaim, FenceOwnerKind};
4771
4772 let (_local, shared, paths) = guest_paths();
4773 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4774 let claimed = |root: &std::path::Path| FenceClaim::owner(root).unwrap().is_some();
4775
4776 let first = lock.acquire().await.unwrap();
4777 let first_runner = first.launch_hold();
4778 drop(first);
4779 assert!(claimed(shared.path()), "held while its runner comes up");
4780
4781 let claimed_at = FenceClaim::owner(shared.path())
4782 .unwrap()
4783 .unwrap()
4784 .acquired_at;
4785 tokio::time::sleep(Duration::from_millis(5)).await;
4786 let second = lock.acquire().await.expect("this process owns the fence");
4787 assert!(
4788 FenceClaim::owner(shared.path())
4789 .unwrap()
4790 .unwrap()
4791 .acquired_at
4792 > claimed_at,
4793 "a claim launches keep sharing is not one stuck for that long"
4794 );
4795 let second_runner = second.launch_hold();
4796 drop(second);
4797 drop(first_runner);
4798 assert!(claimed(shared.path()), "the other launch is still running");
4799
4800 DrainRequest::new(1, chrono::Utc::now())
4802 .write(shared.path())
4803 .unwrap();
4804 assert!(lock.acquire().await.is_err());
4805 assert!(claimed(shared.path()));
4806 assert!(
4807 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4808 .unwrap()
4809 .is_none()
4810 );
4811 drop(second_runner);
4812 assert!(!claimed(shared.path()), "the last one lets it go");
4813 assert!(
4814 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4815 .unwrap()
4816 .is_some()
4817 );
4818 }
4819
4820 #[tokio::test]
4823 async fn refusals_with_quiet_spells_between_them_never_add_up() {
4824 use runner_manager_platform::launch_health::recorded_launches_blocked;
4825 use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};
4826
4827 let (_local, shared, paths) = guest_paths();
4828 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4829 .with_blocked_after(3, Duration::ZERO)
4830 .with_streak_gap(Duration::ZERO);
4831 let windows =
4832 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4833 .unwrap()
4834 .unwrap();
4835 for _ in 0..5 {
4836 assert!(lock.acquire().await.is_err());
4837 }
4838 assert_eq!(
4839 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4840 None
4841 );
4842 windows.release().unwrap();
4843 }
4844
4845 #[tokio::test]
4848 async fn a_claim_refused_by_a_drain_request_is_released() {
4849 use runner_manager_platform::wsl::fence::{DrainRequest, FENCE_DIRECTORY};
4850
4851 let (_local, shared, paths) = guest_paths();
4852 DrainRequest::new(4, chrono::Utc::now())
4853 .write(shared.path())
4854 .unwrap();
4855 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4856 assert!(lock.acquire().await.is_err());
4857 assert!(!shared.path().join(FENCE_DIRECTORY).exists());
4858 }
4859
4860 #[derive(Debug, Default)]
4862 struct HeldLock;
4863
4864 #[async_trait::async_trait]
4865 impl AllocationLock for HeldLock {
4866 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4867 Err(AllocationLockBusy)
4868 }
4869 }
4870
4871 struct Harness {
4873 launcher: Arc<FakeLauncher>,
4874 demand: Arc<FakeDemand>,
4875 events: Arc<EventLog>,
4876 clock: Arc<FakeClock>,
4877 reconciler: Reconciler,
4878 }
4879
4880 impl Harness {
4881 fn build(
4882 host: Host,
4883 launcher: Arc<FakeLauncher>,
4884 demand: Arc<FakeDemand>,
4885 lock: Arc<dyn AllocationLock>,
4886 ) -> Self {
4887 let events = Arc::new(EventLog::new());
4888 let clock = Arc::new(FakeClock::default());
4889 let reconciler = Reconciler::new(
4890 host,
4891 ReconcilerPorts {
4892 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
4893 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
4894 lock,
4895 directory: Arc::new(FakeDirectory::default()),
4896 clock: Arc::clone(&clock) as Arc<dyn Clock>,
4897 jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
4898 events: Arc::clone(&events) as Arc<dyn EventSink>,
4899 },
4900 );
4901 Self {
4902 launcher,
4903 demand,
4904 events,
4905 clock,
4906 reconciler,
4907 }
4908 }
4909
4910 fn simple(capacity: u16, demand_count: u32, target: &str) -> Self {
4911 let launcher = Arc::new(FakeLauncher::new());
4912 let demand = Arc::new(FakeDemand::ready(demand_count, &repo(target)));
4913 Self::build(
4914 host_with(capacity),
4915 launcher,
4916 demand,
4917 Arc::new(InProcessAllocationLock::new()),
4918 )
4919 }
4920 }
4921
4922 fn attempt_in(state: AttemptState, id: u128, policy: u128) -> RunnerAttempt {
4923 let outcome = state.is_terminal().then(|| match state {
4924 AttemptState::Failed => {
4925 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
4926 }
4927 AttemptState::Orphaned => AttemptOutcome::Orphaned,
4928 _ => AttemptOutcome::CompletedJob,
4929 });
4930 RunnerAttempt::from_persisted(PersistedAttempt {
4931 id: AttemptId::from_u128(id),
4932 policy_id: PolicyId::from_u128(policy),
4933 github_runner_id: None,
4934 state,
4935 outcome,
4936 process_id: None,
4937 execution: runner_manager_domain::execution::AttemptExecution::Native {
4938 process_id: None,
4939 },
4940 runtime_path: "runtime/p/a".into(),
4941 workspace_kind: WorkspaceKind::Ephemeral,
4942 workspace_slot: None,
4943 created_at: fixtures::created_at(),
4944 terminal_at: state.is_terminal().then(fixtures::created_at),
4945 last_state_change_at: fixtures::created_at(),
4946 })
4947 .expect("a state/outcome pair the domain accepts")
4948 }
4949
4950 fn concluded(id: u128, policy: u128, outcome: AttemptOutcome) -> RunnerAttempt {
4952 RunnerAttempt::from_persisted(PersistedAttempt {
4953 id: AttemptId::from_u128(id),
4954 policy_id: PolicyId::from_u128(policy),
4955 github_runner_id: None,
4956 state: outcome.terminal_state(),
4957 outcome: Some(outcome),
4958 process_id: None,
4959 execution: runner_manager_domain::execution::AttemptExecution::Native {
4960 process_id: None,
4961 },
4962 runtime_path: "runtime/p/a".into(),
4963 workspace_kind: WorkspaceKind::Ephemeral,
4964 workspace_slot: None,
4965 created_at: fixtures::created_at(),
4966 terminal_at: Some(fixtures::created_at()),
4967 last_state_change_at: fixtures::created_at(),
4968 })
4969 .expect("a state/outcome pair the domain accepts")
4970 }
4971
4972 #[tokio::test]
5001 async fn three_polls_of_one_still_queued_run_yield_exactly_one_attempt() {
5002 let mut harness = Harness::simple(4, 1, "acme/app");
5003 let policy = policy(1, "acme/app", 4);
5004
5005 for poll in 1..=3 {
5006 let report = harness
5007 .reconciler
5008 .reconcile(std::slice::from_ref(&policy))
5009 .await;
5010 assert_eq!(
5011 harness.launcher.launches(),
5012 1,
5013 "poll {poll} started another runner for a job already being served; the \
5014 `- active_owned_runners` term reached `HostAllocator` as a set this host \
5015 does not hold"
5016 );
5017 assert_eq!(report.allocations.len(), 1);
5018 let allocation = &report.allocations[0];
5019 assert_eq!(allocation.demand, 1, "poll {poll}: still queued at GitHub");
5020 if poll == 1 {
5021 assert_eq!(allocation.to_start, 1);
5022 assert_eq!(report.started, 1);
5023 } else {
5024 assert_eq!(allocation.active_owned, 1, "poll {poll}");
5025 assert_eq!(allocation.to_start, 0, "poll {poll}");
5026 assert_eq!(report.started, 0, "poll {poll}");
5027 }
5028 }
5029 assert_eq!(harness.launcher.live_count(), 1);
5030 }
5031
5032 #[tokio::test]
5033 async fn draining_parallel_workflow_starts_runners_for_the_remaining_queue() {
5034 let mut harness = Harness::simple(10, 8, "acme/app");
5035 let policy = policy(1, "acme/app", 10);
5036 let first = harness
5037 .reconciler
5038 .reconcile(std::slice::from_ref(&policy))
5039 .await;
5040 assert_eq!(first.started, 8);
5041
5042 let attempts = harness.launcher.snapshot();
5044 *harness.launcher.attempts.lock().unwrap() = attempts
5045 .iter()
5046 .take(3)
5047 .map(|attempt| attempt_in(AttemptState::Busy, attempt.id.as_uuid().as_u128(), 1))
5048 .collect();
5049 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
5050 repo("acme/app"),
5051 jobs(2),
5052 )));
5053
5054 for poll in 0..3 {
5055 let report = harness
5056 .reconciler
5057 .reconcile(std::slice::from_ref(&policy))
5058 .await;
5059 assert_eq!(report.started, if poll == 0 { 2 } else { 0 });
5060 assert_eq!(harness.launcher.live_count(), 5);
5061 }
5062 }
5063
5064 #[tokio::test]
5075 async fn a_launcher_whose_attempts_never_appear_cannot_wedge_the_pass() {
5076 let launcher = Arc::new(FakeLauncher::forgetful());
5077 let mut harness = Harness::build(
5078 host_with(64),
5079 Arc::clone(&launcher),
5080 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
5081 Arc::new(InProcessAllocationLock::new()),
5082 );
5083
5084 let report = harness
5085 .reconciler
5086 .reconcile(&[policy(1, "acme/app", 8)])
5087 .await;
5088
5089 assert_eq!(
5090 report.started, 3,
5091 "the pass is bounded by the grant it was given, not by the attempt set catching \
5092 up with it"
5093 );
5094 assert_eq!(launcher.launches(), 3);
5095 assert!(
5096 launcher.snapshot().is_empty(),
5097 "the launcher never recorded anything, which is the whole point of the fixture"
5098 );
5099 }
5100
5101 #[tokio::test]
5113 async fn two_policies_cannot_exceed_host_capacity_even_when_the_launcher_lags() {
5114 let launcher = Arc::new(FakeLauncher::forgetful());
5115 let demand = Arc::new(FakeDemand::default());
5116 demand.set_for(
5117 &ScaleTarget::repository("acme/left").unwrap(),
5118 PollOutcome::Ready(QueuedDemand::of(repo("acme/left"), jobs(3))),
5119 );
5120 demand.set_for(
5121 &ScaleTarget::repository("acme/right").unwrap(),
5122 PollOutcome::Ready(QueuedDemand::of(repo("acme/right"), jobs(3))),
5123 );
5124 let mut harness = Harness::build(
5125 host_with(3),
5126 Arc::clone(&launcher),
5127 demand,
5128 Arc::new(InProcessAllocationLock::new()),
5129 );
5130
5131 let report = harness
5132 .reconciler
5133 .reconcile(&[policy(1, "acme/left", 3), policy(2, "acme/right", 3)])
5134 .await;
5135
5136 assert_eq!(
5137 report.started, 3,
5138 "host_capacity is 3 and two policies each allowed 3 started {} runners \
5139 between them; the second policy's grant was computed from a set that did \
5140 not yet contain the first policy's launches",
5141 report.started
5142 );
5143 assert_eq!(launcher.launches(), 3);
5144 }
5145
5146 #[tokio::test]
5157 async fn an_unreadable_attempt_set_starts_nothing_and_is_not_read_as_an_idle_host() {
5158 let launcher = Arc::new(FakeLauncher::new());
5159 let mut harness = Harness::build(
5160 host_with(8),
5161 Arc::clone(&launcher),
5162 Arc::new(FakeDemand::ready(4, &repo("acme/app"))),
5163 Arc::new(InProcessAllocationLock::new()),
5164 );
5165 let policy = policy(1, "acme/app", 8);
5166
5167 launcher.fail_attempts(true);
5168 let unreadable = harness
5169 .reconciler
5170 .reconcile(std::slice::from_ref(&policy))
5171 .await;
5172
5173 assert_eq!(
5174 unreadable.started, 0,
5175 "nothing may be decided from a set that was not read"
5176 );
5177 assert_eq!(launcher.launches(), 0);
5178 assert!(unreadable.attempts_unreadable > 0, "and the pass says so");
5179 assert!(
5180 unreadable.allocations.is_empty(),
5181 "no allocation is reported either: there was no set to compute one from, and \
5182 an allocation of zero would claim a decision nobody made"
5183 );
5184 assert!(harness.events.count_of("attempts_unreadable") > 0);
5185
5186 launcher.fail_attempts(false);
5188 let readable = harness
5189 .reconciler
5190 .reconcile(std::slice::from_ref(&policy))
5191 .await;
5192 assert_eq!(
5193 readable.started, 4,
5194 "the difference between the two passes is only whether the set could be read"
5195 );
5196 assert_eq!(readable.attempts_unreadable, 0);
5197 }
5198
5199 #[tokio::test]
5209 async fn a_cleanup_that_cannot_succeed_is_reported_rather_than_retried_in_silence() {
5210 let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
5211 1,
5212 1,
5213 AttemptOutcome::ExitedIdleWithoutWork,
5214 )]));
5215 let mut harness = Harness::build(
5216 host_with(4),
5217 Arc::clone(&launcher),
5218 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5219 Arc::new(InProcessAllocationLock::new()),
5220 );
5221
5222 let report = harness
5223 .reconciler
5224 .reconcile(&[policy(1, "acme/app", 4)])
5225 .await;
5226
5227 assert_eq!(report.cleaned, 0);
5228 assert_eq!(report.clean_failures, 1);
5229 assert_eq!(harness.events.count_of("attempt_clean_failed"), 1);
5230 assert_eq!(
5231 harness.events.count_of("attempt_cleaned"),
5232 0,
5233 "and it is not reported as cleaned"
5234 );
5235 assert_eq!(
5236 launcher.snapshot().len(),
5237 1,
5238 "the attempt is still there, so the retry is real -- what changed is that it \
5239 is no longer silent"
5240 );
5241
5242 let reasons: Vec<&'static str> = harness
5245 .events
5246 .events()
5247 .into_iter()
5248 .filter_map(|event| match event {
5249 LifecycleEvent::AttemptCleanFailed { reason, .. } => Some(reason),
5250 _ => None,
5251 })
5252 .collect();
5253 assert_eq!(reasons, vec!["other"]);
5254 }
5255
5256 #[tokio::test]
5259 async fn a_failing_cleanup_backs_off_exponentially_per_attempt() {
5260 let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
5261 1,
5262 1,
5263 AttemptOutcome::ExitedIdleWithoutWork,
5264 )]));
5265 let mut harness = Harness::build(
5266 host_with(4),
5267 Arc::clone(&launcher),
5268 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5269 Arc::new(InProcessAllocationLock::new()),
5270 );
5271 let policies = [policy(1, "acme/app", 4)];
5272 let start = harness.clock.now();
5273 let at = |offset: Duration| start + chrono::Duration::from_std(offset).unwrap();
5274 let mut failures_at = async |offset: Duration| {
5275 harness.clock.set(at(offset));
5276 harness.reconciler.reconcile(&policies).await.clean_failures
5277 };
5278
5279 assert_eq!(failures_at(Duration::ZERO).await, 1);
5280 assert_eq!(
5282 failures_at(CLEAN_RETRY_INITIAL - Duration::from_secs(1)).await,
5283 0,
5284 "a failed cleanup was retried on the very next pass"
5285 );
5286 assert_eq!(failures_at(CLEAN_RETRY_INITIAL).await, 1);
5287 let second = CLEAN_RETRY_INITIAL + CLEAN_RETRY_INITIAL * 2;
5289 assert_eq!(
5290 failures_at(second - Duration::from_secs(1)).await,
5291 0,
5292 "the delay did not double after a second failure"
5293 );
5294 assert_eq!(failures_at(second).await, 1);
5295 assert_eq!(harness.events.count_of("attempt_clean_failed"), 3);
5296
5297 let mut backoff = CleanBackoff::default();
5299 let id = AttemptId::from_u128(9);
5300 for _ in 0..40 {
5301 backoff.failed(id, start);
5302 }
5303 assert!(!backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM - Duration::from_secs(1))));
5304 assert!(backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM)));
5305 backoff.succeeded(id);
5306 assert!(
5307 backoff.is_due(id, start),
5308 "a cleaned attempt kept its backoff"
5309 );
5310 }
5311
5312 #[tokio::test]
5325 async fn an_unreadable_attempt_set_makes_scale_down_inconclusive_rather_than_empty() {
5326 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5327 attempt_in(AttemptState::Busy, 1, 1),
5328 concluded(2, 1, AttemptOutcome::CompletedJob),
5329 ]));
5330 let harness = Harness::build(
5331 host_with(4),
5332 Arc::clone(&launcher),
5333 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5334 Arc::new(InProcessAllocationLock::new()),
5335 );
5336 let policy = policy(1, "acme/app", 4);
5337
5338 launcher.fail_attempts(true);
5339 let blind = harness.reconciler.scale_down(&policy).await;
5340
5341 assert!(!blind.is_conclusive(), "the machine was never read");
5342 assert_ne!(
5343 blind,
5344 ScaleDownReport::default(),
5345 "a scale-down that could not see the host must not be equal to one that saw \
5346 an idle host; that equality is the whole finding"
5347 );
5348 assert_eq!(blind.removed, 0);
5349 assert_eq!(
5350 blind.refused_busy, 0,
5351 "and this zero means `unknown`, not `none`"
5352 );
5353 assert_eq!(harness.events.count_of("attempts_unreadable"), 1);
5354
5355 launcher.fail_attempts(false);
5357 let seeing = harness.reconciler.scale_down(&policy).await;
5358
5359 assert!(seeing.is_conclusive());
5360 assert_eq!(seeing.removed, 1, "the concluded attempt was reclaimed");
5361 assert_eq!(seeing.refused_busy, 1, "and the busy one was left alone");
5362 assert_ne!(
5363 seeing, blind,
5364 "the difference between the two is only whether the set could be read"
5365 );
5366 }
5367
5368 #[tokio::test]
5376 async fn demand_below_min_capacity_is_raised_to_min_capacity() {
5377 let mut warm = ScalePolicy::new(
5378 PolicyId::from_u128(1),
5379 ScaleTarget::repository("acme/app").unwrap(),
5380 1,
5381 fixtures::HOST_ID,
5382 PolicyMode::autoscale(
5383 fixtures::routing_labels("home"),
5384 2,
5385 NonZeroU16::new(5).expect("non-zero"),
5386 )
5387 .expect("min <= max"),
5388 CachePolicy::default(),
5389 );
5390 warm.activate().expect("pending -> active");
5391
5392 let mut harness = Harness::simple(8, 0, "acme/app");
5393 let report = harness.reconciler.reconcile(&[warm]).await;
5394
5395 assert_eq!(
5396 report.allocations[0].demand, 0,
5397 "GitHub reported no queued runs"
5398 );
5399 assert_eq!(
5400 report.allocations[0].desired, 2,
5401 "min_capacity raised the target above demand"
5402 );
5403 assert_eq!(
5404 report.allocations[0].limiting_factor,
5405 LimitingFactor::MinCapacity
5406 );
5407 assert_eq!(report.started, 2, "and two runners were actually started");
5408 assert_eq!(harness.launcher.live_count(), 2);
5409 }
5410
5411 #[derive(Debug)]
5418 struct FixedMemory {
5419 free: u64,
5420 estimate: u64,
5421 admitted: std::sync::atomic::AtomicU32,
5422 }
5423
5424 impl MemoryGate for FixedMemory {
5425 fn admit(&self, _policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict {
5426 let running: Vec<runner_manager_domain::memory::Running> = attempts
5427 .iter()
5428 .filter(|attempt| attempt.counts_against_capacity())
5429 .map(|_| runner_manager_domain::memory::Running {
5430 estimate: self.estimate,
5431 current_rss: None,
5432 })
5433 .collect();
5434 runner_manager_domain::memory::admit(
5435 Some(self.free),
5436 &running,
5437 runner_manager_domain::memory::Estimate::choose(None, Some(self.estimate)),
5438 GIB,
5439 )
5440 }
5441
5442 fn admitted(&self, _policy: &ScalePolicy) {
5443 self.admitted
5444 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
5445 }
5446 }
5447
5448 const GIB: u64 = 1 << 30;
5449
5450 fn with_memory(harness: Harness, free: u64, estimate: u64) -> (Harness, Arc<FixedMemory>) {
5451 let gate = Arc::new(FixedMemory {
5452 free,
5453 estimate,
5454 admitted: std::sync::atomic::AtomicU32::new(0),
5455 });
5456 let Harness {
5457 launcher,
5458 demand,
5459 events,
5460 clock,
5461 reconciler,
5462 } = harness;
5463 let reconciler = reconciler.with_memory_gate(Arc::clone(&gate) as Arc<dyn MemoryGate>);
5464 (
5465 Harness {
5466 launcher,
5467 demand,
5468 events,
5469 clock,
5470 reconciler,
5471 },
5472 gate,
5473 )
5474 }
5475
5476 #[tokio::test]
5477 async fn a_start_the_memory_cannot_hold_is_deferred_and_the_demand_kept() {
5478 let (mut harness, gate) = with_memory(Harness::simple(8, 5, "acme/app"), 10 * GIB, 4 * GIB);
5482 let report = harness
5483 .reconciler
5484 .reconcile(&[policy(1, "acme/app", 5)])
5485 .await;
5486
5487 assert_eq!(
5488 report.allocations[0].to_start, 5,
5489 "the ceilings granted five"
5490 );
5491 assert_eq!(report.started, 2);
5492 assert_eq!(report.memory_deferred, 3, "deferred, not failed");
5493 assert_eq!(report.failures, 0);
5494 assert_eq!(gate.admitted.load(std::sync::atomic::Ordering::Relaxed), 2);
5495 let deferred: Vec<MemoryFigures> = harness
5496 .events
5497 .events()
5498 .into_iter()
5499 .filter_map(|event| match event {
5500 LifecycleEvent::MemoryShort {
5501 figures,
5502 started: false,
5503 ..
5504 } => Some(figures),
5505 _ => None,
5506 })
5507 .collect();
5508 assert_eq!(deferred.len(), 1);
5509 assert_eq!(deferred[0].free, 10 * GIB);
5510 assert_eq!(deferred[0].reserved, 8 * GIB);
5511 assert_eq!(deferred[0].needed(), 5 * GIB);
5512
5513 let again = harness
5516 .reconciler
5517 .reconcile(&[policy(1, "acme/app", 5)])
5518 .await;
5519 assert_eq!(again.started, 0);
5520 assert!(again.memory_deferred > 0);
5521 assert_eq!(harness.launcher.live_count(), 2);
5522 }
5523
5524 #[tokio::test]
5525 async fn a_runner_bigger_than_the_free_memory_starts_when_nothing_else_runs() {
5526 let (mut harness, _) = with_memory(Harness::simple(8, 2, "acme/app"), GIB, 4 * GIB);
5527 let report = harness
5528 .reconciler
5529 .reconcile(&[policy(1, "acme/app", 5)])
5530 .await;
5531
5532 assert_eq!(report.started, 1, "no starvation, but only one");
5533 assert_eq!(report.memory_deferred, 1);
5534 let names: Vec<&str> = harness
5535 .events
5536 .events()
5537 .iter()
5538 .map(LifecycleEvent::name)
5539 .filter(|name| name.starts_with("memory_"))
5540 .collect();
5541 assert_eq!(names, ["memory_admitted_alone", "memory_deferred"]);
5542 }
5543
5544 #[tokio::test]
5545 async fn without_a_memory_gate_every_granted_runner_starts() {
5546 let mut harness = Harness::simple(8, 5, "acme/app");
5547 let report = harness
5548 .reconciler
5549 .reconcile(&[policy(1, "acme/app", 5)])
5550 .await;
5551 assert_eq!(report.started, 5);
5552 assert_eq!(report.memory_deferred, 0);
5553 }
5554
5555 #[tokio::test]
5560 async fn demand_above_max_capacity_is_clamped_to_max_capacity() {
5561 let mut harness = Harness::simple(100, 10, "acme/app");
5562 let report = harness
5563 .reconciler
5564 .reconcile(&[policy(1, "acme/app", 3)])
5565 .await;
5566
5567 assert_eq!(report.allocations[0].demand, 10);
5568 assert_eq!(
5569 report.allocations[0].desired, 3,
5570 "max_capacity beats demand"
5571 );
5572 assert_eq!(report.started, 3);
5573 assert_eq!(
5574 report.allocations[0].limiting_factor,
5575 LimitingFactor::MaxCapacity
5576 );
5577 }
5578
5579 #[tokio::test]
5580 async fn demand_below_min_capacity_starts_nothing_in_v1() {
5581 let mut harness = Harness::simple(8, 0, "acme/app");
5584 let report = harness
5585 .reconciler
5586 .reconcile(&[policy(1, "acme/app", 4)])
5587 .await;
5588
5589 assert_eq!(report.allocations[0].desired, 0);
5590 assert_eq!(report.started, 0);
5591 assert!(report.starts_nothing());
5592 }
5593
5594 #[tokio::test]
5595 async fn lifecycle_replacement_intent_is_consumed_by_the_ordinary_allocator() {
5596 let policy = policy(1, "octo/repo", 1);
5597 let previous = attempt_in(AttemptState::Starting, 41, 1);
5598 let intent = ReplacementIntent {
5599 policy: policy.id,
5600 previous_attempt: previous.id,
5601 operation: "exit_before_acceptance_replacement",
5602 };
5603 let launcher = Arc::new(FakeLauncher::new().seeded(vec![previous]).replacing(intent));
5604 let demand = Arc::new(FakeDemand::ready(1, &repo("octo/repo")));
5605 let mut harness = Harness::build(
5606 host_with(1),
5607 Arc::clone(&launcher),
5608 demand,
5609 Arc::new(InProcessAllocationLock::new()),
5610 );
5611
5612 let report = harness.reconciler.reconcile(&[policy]).await;
5613
5614 assert_eq!(report.replacement_intents, 1);
5615 assert_eq!(report.started, 1);
5616 assert_eq!(launcher.launches(), 1);
5617 assert_eq!(launcher.live_count(), 1);
5618 }
5619
5620 #[tokio::test]
5621 async fn zero_host_headroom_starts_nothing_at_maximum_demand() {
5622 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5623 attempt_in(AttemptState::Busy, 1, 1),
5624 attempt_in(AttemptState::Busy, 2, 1),
5625 ]));
5626 let demand = Arc::new(FakeDemand::ready(u32::from(u16::MAX), &repo("acme/app")));
5627 let mut harness = Harness::build(
5628 host_with(2),
5629 launcher,
5630 demand,
5631 Arc::new(InProcessAllocationLock::new()),
5632 );
5633
5634 let report = harness
5635 .reconciler
5636 .reconcile(&[policy(1, "acme/app", 2)])
5637 .await;
5638 assert_eq!(report.started, 0);
5639 assert_eq!(report.allocations[0].headroom_before, 0);
5640 assert_eq!(harness.launcher.launches(), 0);
5641 }
5642
5643 #[tokio::test]
5644 async fn headroom_smaller_than_the_per_policy_allowance_wins() {
5645 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5648 attempt_in(AttemptState::Busy, 1, 99),
5649 attempt_in(AttemptState::Busy, 2, 99),
5650 attempt_in(AttemptState::Idle, 3, 99),
5651 attempt_in(AttemptState::Starting, 4, 99),
5652 ]));
5653 let demand = Arc::new(FakeDemand::ready(5, &repo("acme/app")));
5654 let mut harness = Harness::build(
5655 host_with(6),
5656 launcher,
5657 demand,
5658 Arc::new(InProcessAllocationLock::new()),
5659 );
5660
5661 let report = harness
5662 .reconciler
5663 .reconcile(&[policy(1, "acme/app", 5)])
5664 .await;
5665 assert_eq!(
5666 report.allocations[0].desired, 5,
5667 "its own ceiling allows five"
5668 );
5669 assert_eq!(report.started, 2, "the host has two slots free");
5670 assert_eq!(
5671 report.allocations[0].limiting_factor,
5672 LimitingFactor::HostCapacity
5673 );
5674 assert_eq!(harness.launcher.live_count(), 6);
5675 }
5676
5677 #[tokio::test]
5678 async fn the_idle_host_assertion_holds() {
5679 let mut harness = Harness::simple(8, 0, "acme/app");
5682 let report = harness
5683 .reconciler
5684 .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/app", 4)])
5685 .await;
5686
5687 assert_eq!(report.started, 0);
5688 assert_eq!(harness.launcher.launches(), 0);
5689 assert!(harness.launcher.snapshot().is_empty());
5690 assert_eq!(
5691 harness
5692 .launcher
5693 .snapshot()
5694 .iter()
5695 .filter(|a| !a.is_terminal())
5696 .count(),
5697 0
5698 );
5699 }
5700
5701 #[tokio::test]
5706 async fn live_capacity_increase_is_used_without_rebuilding_the_reconciler() {
5707 let mut harness = Harness::simple(1, 5, "acme/app");
5708 let policies = [policy(1, "acme/app", 5)];
5709 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);
5710
5711 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(4);
5712 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 3);
5713 assert_eq!(harness.launcher.live_count(), 4);
5714 }
5715
5716 #[tokio::test]
5717 async fn live_capacity_decrease_keeps_busy_runners_and_waits_for_headroom() {
5718 let mut harness = Harness::simple(5, 5, "acme/app");
5719 let busy: Vec<_> = (1..=3)
5720 .map(|id| attempt_in(AttemptState::Busy, id, 1))
5721 .collect();
5722 *harness.launcher.attempts.lock().unwrap() = busy.clone();
5723 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
5724 let policies = [policy(1, "acme/app", 5)];
5725
5726 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 0);
5727 assert_eq!(harness.launcher.snapshot(), busy);
5728 assert!(harness.launcher.cleaned().is_empty());
5729
5730 harness.launcher.attempts.lock().unwrap().truncate(1);
5732 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);
5733 assert_eq!(harness.launcher.live_count(), 2);
5734 }
5735
5736 #[tokio::test]
5737 async fn live_capacity_is_rechecked_between_launches_in_the_same_pass() {
5738 let mut harness = Harness::simple(5, 5, "acme/app");
5739 *harness.launcher.capacity_after_launch.lock().unwrap() = NonZeroU16::new(1);
5740
5741 let report = harness
5742 .reconciler
5743 .reconcile(&[policy(1, "acme/app", 5)])
5744 .await;
5745 assert_eq!(report.started, 1);
5746 assert_eq!(harness.launcher.live_count(), 1);
5747 }
5748
5749 #[tokio::test]
5750 async fn live_capacity_unreadable_blocks_launches_until_the_journal_recovers() {
5751 let mut harness = Harness::simple(5, 5, "acme/app");
5752 harness
5753 .launcher
5754 .capacity_unreadable
5755 .store(true, Ordering::SeqCst);
5756 let policies = [policy(1, "acme/app", 5)];
5757 let report = harness.reconciler.reconcile(&policies).await;
5758 assert_eq!(report.started, 0);
5759 assert_eq!(report.host_capacity_unreadable, 1);
5760 assert_eq!(report.attempts_unreadable, 0);
5761 assert_eq!(harness.events.count_of("attempts_unreadable"), 0);
5762 assert!(
5763 harness
5764 .events
5765 .events()
5766 .contains(&LifecycleEvent::HostCapacityUnreadable {
5767 reason: HostCapacityError::JournalUnreadable,
5768 })
5769 );
5770
5771 harness
5772 .launcher
5773 .capacity_unreadable
5774 .store(false, Ordering::SeqCst);
5775 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
5776 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 2);
5777 }
5778
5779 #[tokio::test]
5780 async fn unreadable_local_state_does_not_treat_a_busy_host_as_idle() {
5781 for host_failure in [true, false] {
5782 let mut harness = Harness::simple(5, 0, "acme/app");
5783 harness
5784 .reconciler
5785 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
5786 *harness.launcher.attempts.lock().unwrap() = vec![attempt_in(AttemptState::Busy, 1, 1)];
5787 harness
5788 .launcher
5789 .capacity_unreadable
5790 .store(host_failure, Ordering::SeqCst);
5791 harness.launcher.fail_attempts(!host_failure);
5792 let report = harness
5793 .reconciler
5794 .reconcile(&[policy(1, "acme/app", 5)])
5795 .await;
5796 assert_eq!(report.host_capacity_unreadable, u16::from(host_failure));
5797 assert_eq!(report.next_poll.pace, PollPace::Nominal);
5798 assert_eq!(harness.launcher.live_count(), 1);
5799 }
5800 }
5801
5802 #[derive(Debug)]
5803 struct HostEditOnLock {
5804 inner: InProcessAllocationLock,
5805 launcher: Arc<FakeLauncher>,
5806 unreadable: bool,
5807 }
5808
5809 #[async_trait::async_trait]
5810 impl AllocationLock for HostEditOnLock {
5811 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
5812 let guard = self.inner.acquire().await?;
5813 *self.launcher.capacity.lock().unwrap() = NonZeroU16::new(1);
5814 self.launcher
5815 .capacity_unreadable
5816 .store(self.unreadable, Ordering::SeqCst);
5817 Ok(guard)
5818 }
5819 }
5820
5821 #[tokio::test]
5822 async fn live_capacity_edits_and_read_failures_while_waiting_for_the_lock_block_launches() {
5823 for unreadable in [false, true] {
5824 let launcher =
5825 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
5826 let lock = Arc::new(HostEditOnLock {
5827 inner: InProcessAllocationLock::new(),
5828 launcher: Arc::clone(&launcher),
5829 unreadable,
5830 });
5831 let mut harness = Harness::build(
5832 host_with(5),
5833 Arc::clone(&launcher),
5834 Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
5835 lock,
5836 );
5837 let report = harness
5838 .reconciler
5839 .reconcile(&[policy(1, "acme/app", 5)])
5840 .await;
5841 assert_eq!(report.started, 0);
5842 assert_eq!(launcher.live_count(), 1);
5843 assert!(launcher.cleaned().is_empty());
5844 assert_eq!(report.host_capacity_unreadable, u16::from(unreadable));
5845 assert_eq!(report.attempts_unreadable, 0);
5846 }
5847 }
5848
5849 #[tokio::test(flavor = "current_thread")]
5872 async fn two_policies_reconciling_concurrently_never_exceed_host_capacity() {
5873 let lock = Arc::new(CountingLock::new());
5874 let (launches, live) =
5875 two_policies_concurrently(Arc::clone(&lock) as Arc<dyn AllocationLock>, None).await;
5876
5877 assert_eq!(
5878 launches, 3,
5879 "the sum across policies must never exceed host_capacity, and each policy is \
5880 individually within its own max_capacity of 3"
5881 );
5882 assert_eq!(live, 3, "and no duplicate runner survived the race");
5883 assert_eq!(
5884 lock.peak(),
5885 1,
5886 "the allocation lock had one holder at a time; without that the read of the \
5887 headroom and the creation of the runtime are not atomic"
5888 );
5889 assert!(
5890 (3..=5).contains(&lock.acquisitions()),
5891 "the lock is taken before *each* runtime, not once per pass: three runtimes \
5892 means at least three holds, and at most one further hold per policy to \
5893 discover the host filled up underneath it. It was taken {} times",
5894 lock.acquisitions()
5895 );
5896 }
5897
5898 #[tokio::test(flavor = "current_thread")]
5905 async fn without_the_allocation_lock_two_policies_oversubscribe_the_host() {
5906 let (launches, _) =
5907 two_policies_concurrently(Arc::new(NoLock) as Arc<dyn AllocationLock>, None).await;
5908
5909 assert!(
5910 launches > 3,
5911 "with no serialisation both policies must be able to spend the same headroom; \
5912 they started {launches} runners on a host of 3. If this is ever 3, the \
5913 contention window closed and `two_policies_reconciling_concurrently_never_\
5914 exceed_host_capacity` has stopped proving anything"
5915 );
5916 }
5917
5918 async fn two_policies_concurrently(
5922 lock: Arc<dyn AllocationLock>,
5923 live_capacity: Option<NonZeroU16>,
5924 ) -> (usize, usize) {
5925 let launcher = Arc::new(FakeLauncher::new().with_yields(4));
5926 *launcher.capacity.lock().unwrap() = live_capacity;
5927 let host = host_with(3);
5928
5929 let mut left = Harness::build(
5930 host.clone(),
5931 Arc::clone(&launcher),
5932 Arc::new(FakeDemand::ready(3, &repo("acme/left"))),
5933 Arc::clone(&lock),
5934 )
5935 .reconciler;
5936 let mut right = Harness::build(
5937 host,
5938 Arc::clone(&launcher),
5939 Arc::new(FakeDemand::ready(3, &repo("acme/right"))),
5940 Arc::clone(&lock),
5941 )
5942 .reconciler;
5943
5944 let a = policy(1, "acme/left", 3);
5945 let b = policy(2, "acme/right", 3);
5946
5947 let left = tokio::spawn(async move { left.reconcile(&[a]).await });
5948 let right = tokio::spawn(async move { right.reconcile(&[b]).await });
5949 let (_, _) = (left.await.unwrap(), right.await.unwrap());
5950
5951 (launcher.launches(), launcher.live_count())
5952 }
5953
5954 #[tokio::test(flavor = "current_thread")]
5955 async fn live_capacity_is_shared_by_concurrent_targets_with_stale_startup_hosts() {
5956 let (launches, live) =
5957 two_policies_concurrently(Arc::new(InProcessAllocationLock::new()), NonZeroU16::new(2))
5958 .await;
5959 assert_eq!(launches, 2);
5960 assert_eq!(live, 2);
5961 }
5962
5963 #[tokio::test]
5975 async fn a_monitor_only_policy_under_maximum_demand_starts_nothing_and_polls_nothing() {
5976 let gateway = FakeGithub::new().with_queued_jobs(repo("acme/app"), jobs(10_000));
5977 let gateway = Arc::new(GatewayDemand::new(gateway, CancelToken::new()));
5978 let launcher = Arc::new(FakeLauncher::new());
5979 let events = Arc::new(EventLog::new());
5980
5981 let mut reconciler = Reconciler::new(
5982 host_with(10),
5983 ReconcilerPorts {
5984 demand: Arc::clone(&gateway) as Arc<dyn DemandSource>,
5985 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
5986 lock: Arc::new(InProcessAllocationLock::new()),
5987 directory: Arc::new(FakeDirectory::default()),
5988 clock: Arc::new(FakeClock::default()),
5989 jitter: Arc::new(NoJitter),
5990 events: Arc::clone(&events) as Arc<dyn EventSink>,
5991 },
5992 );
5993
5994 let monitor = fixtures::policy()
5995 .id(PolicyId::from_u128(1))
5996 .repository("acme/app")
5997 .monitor_only()
5998 .active()
5999 .build();
6000
6001 let report = reconciler.reconcile(&[monitor]).await;
6002
6003 assert_eq!(report.started, 0);
6004 assert_eq!(launcher.launches(), 0);
6005 assert_eq!(report.monitor_only, vec![PolicyId::from_u128(1)]);
6006 assert_eq!(
6007 report.demand_requests, 0,
6008 "a monitor-only policy spends nothing from the shared hourly ceiling"
6009 );
6010 assert!(
6011 gateway.gateway().calls().is_empty(),
6012 "a monitor-only policy issued a demand request: {:?}",
6013 gateway.gateway().calls()
6014 );
6015 assert_eq!(events.count_of("monitor_only_skipped"), 1);
6016 assert_eq!(
6017 events.count_of("demand_observed"),
6018 0,
6019 "and it contributed no demand"
6020 );
6021 }
6022
6023 #[tokio::test]
6041 async fn a_monitor_only_policy_sharing_a_target_is_still_skipped_entirely() {
6042 let lock = Arc::new(CountingLock::new());
6043 let launcher = Arc::new(FakeLauncher::new());
6044 let events = Arc::new(EventLog::new());
6045 let mut reconciler = Reconciler::new(
6046 host_with(4),
6047 ReconcilerPorts {
6048 demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
6049 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
6050 lock: Arc::clone(&lock) as Arc<dyn AllocationLock>,
6051 directory: Arc::new(FakeDirectory::default()),
6052 clock: Arc::new(FakeClock::default()),
6053 jitter: Arc::new(NoJitter),
6054 events: Arc::clone(&events) as Arc<dyn EventSink>,
6055 },
6056 );
6057
6058 let watcher = fixtures::policy()
6059 .id(PolicyId::from_u128(2))
6060 .repository("acme/app")
6061 .monitor_only()
6062 .active()
6063 .build();
6064
6065 let report = reconciler
6066 .reconcile(&[policy(1, "acme/app", 4), watcher])
6067 .await;
6068
6069 assert_eq!(report.started, 2, "the autoscale policy is served normally");
6070 assert_eq!(report.monitor_only, vec![PolicyId::from_u128(2)]);
6071 assert_eq!(
6072 events.count_of("demand_observed"),
6073 1,
6074 "the demand observation belongs to the autoscale policy alone"
6075 );
6076 assert!(
6077 report
6078 .allocations
6079 .iter()
6080 .all(|a| a.policy_id == PolicyId::from_u128(1)),
6081 "a monitor-only policy was allocated for: {:?}",
6082 report.allocations
6083 );
6084 assert_eq!(
6085 lock.acquisitions(),
6086 2,
6087 "one hold per runtime created, and none on behalf of the monitor-only policy. \
6088 It was three before the budget was checked at the top of the loop rather than \
6089 after the re-read, which cost every policy a surplus hold to discover there \
6090 was nothing left to grant"
6091 );
6092 }
6093
6094 #[tokio::test]
6095 async fn the_monitor_only_refusal_is_asserted_on_the_mode_not_on_a_missing_ceiling() {
6096 let monitor = fixtures::monitor_only_policy();
6101 assert!(!monitor.owns_runners());
6102 assert_eq!(monitor.max_capacity(), None);
6103
6104 let host = host_with(10);
6105 let attempts: Vec<RunnerAttempt> = Vec::new();
6106 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
6107 let allocation = allocator.allocate(&monitor, 10_000);
6108 assert_eq!(allocation.limiting_factor, LimitingFactor::MonitorOnly);
6109 assert_eq!(allocation.to_start, 0);
6110 assert_eq!(
6111 allocator.headroom(),
6112 10,
6113 "and it consumes no headroom, so an autoscale policy on the same host is \
6114 unaffected"
6115 );
6116 }
6117
6118 #[tokio::test]
6126 async fn a_surplus_attempt_is_cleaned_as_an_idle_exit_and_not_as_a_failure() {
6127 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
6128 concluded(1, 1, AttemptOutcome::ExitedIdleWithoutWork),
6129 concluded(
6130 2,
6131 1,
6132 AttemptOutcome::failed(FailureReason::JitRequestFailed),
6133 ),
6134 concluded(3, 1, AttemptOutcome::CompletedJob),
6135 ]));
6136 let mut harness = Harness::build(
6137 host_with(4),
6138 Arc::clone(&launcher),
6139 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
6140 Arc::new(InProcessAllocationLock::new()),
6141 );
6142
6143 let report = harness
6144 .reconciler
6145 .reconcile(&[policy(1, "acme/app", 4)])
6146 .await;
6147
6148 assert_eq!(report.cleaned, 3);
6149 assert_eq!(report.idle_exits, 1, "the surplus case, counted apart");
6150 assert_eq!(
6151 report.failures, 1,
6152 "only the failed attempt is a failure; the idle exit and the completed job are \
6153 not"
6154 );
6155 assert_eq!(launcher.cleaned().len(), 3);
6156 assert!(launcher.snapshot().is_empty());
6157
6158 let cleaned: Vec<OutcomeKind> = harness
6159 .events
6160 .events()
6161 .into_iter()
6162 .filter_map(|event| match event {
6163 LifecycleEvent::AttemptCleaned { outcome, .. } => Some(outcome),
6164 _ => None,
6165 })
6166 .collect();
6167 assert!(cleaned.contains(&OutcomeKind::IdleExit));
6168 assert!(
6169 !OutcomeKind::IdleExit.is_failure(),
6170 "an idle exit rendered as a failure sends an operator hunting a fault that does \
6171 not exist"
6172 );
6173 }
6174
6175 #[tokio::test]
6178 async fn scale_down_removes_nothing_from_a_busy_attempt() {
6179 let busy = attempt_in(AttemptState::Busy, 1, 1);
6180 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
6181 busy.clone(),
6182 attempt_in(AttemptState::Starting, 2, 1),
6183 concluded(3, 1, AttemptOutcome::CompletedJob),
6184 ]));
6185 let harness = Harness::build(
6186 host_with(4),
6187 Arc::clone(&launcher),
6188 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
6189 Arc::new(InProcessAllocationLock::new()),
6190 );
6191
6192 let report = harness
6193 .reconciler
6194 .scale_down(&policy(1, "acme/app", 4))
6195 .await;
6196
6197 assert_eq!(report.refused_busy, 1);
6198 assert_eq!(
6199 report.retained, 1,
6200 "the `starting` attempt is not ended either"
6201 );
6202 assert_eq!(report.removed, 1, "only the concluded attempt is reclaimed");
6203
6204 let after = launcher.snapshot();
6205 let still_busy = after
6206 .iter()
6207 .find(|a| a.id == AttemptId::from_u128(1))
6208 .expect("the busy attempt is still there");
6209 assert_eq!(
6210 still_busy.state(),
6211 AttemptState::Busy,
6212 "scale-down removed nothing from a runner that is executing a job, and left it \
6213 busy"
6214 );
6215 assert!(!launcher.cleaned().contains(&AttemptId::from_u128(1)));
6216
6217 let mut busy = busy;
6221 assert!(matches!(
6222 busy.clean(fixtures::created_at()),
6223 Err(runner_manager_domain::attempt::AttemptError::BusyCannotBeCleaned)
6224 ));
6225 assert_eq!(harness.events.count_of("scale_down_refused"), 1);
6226 }
6227
6228 #[test]
6233 fn the_default_interval_is_sixty_seconds_and_the_floor_is_thirty() {
6234 assert_eq!(RefreshInterval::DEFAULT_SECS, 60);
6235 assert_eq!(RefreshInterval::MIN_SECS, 30);
6236 assert_eq!(PollSchedule::floor(), Duration::from_secs(30));
6237 assert!(
6238 RefreshInterval::from_secs(29).is_err(),
6239 "the floor is a rate-budget constraint, and a caller must not be able to write \
6240 a shorter interval at all"
6241 );
6242
6243 let mut schedule = PollSchedule::new(RefreshInterval::default());
6244 let next = schedule.next_poll(None, fixtures::created_at(), &NoJitter);
6245 assert_eq!(next.delay, Duration::from_secs(60));
6246 assert_eq!(next.pace, PollPace::Nominal);
6247
6248 let mut floored = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
6249 assert_eq!(
6250 floored
6251 .next_poll(None, fixtures::created_at(), &NoJitter)
6252 .delay,
6253 Duration::from_secs(30)
6254 );
6255 }
6256
6257 #[test]
6261 fn a_rate_limit_increases_the_delay_and_names_itself() {
6262 let now = fixtures::created_at();
6263 let mut schedule = PollSchedule::new(RefreshInterval::default());
6264
6265 let limited = RefreshState::RateLimited(RateLimited {
6266 kind: RateLimitKind::Secondary,
6267 retry_after: Some(Duration::from_secs(300)),
6268 remaining: None,
6269 reset_unix_secs: None,
6270 });
6271 let next = schedule.next_poll(Some(&limited), now, &NoJitter);
6272
6273 assert_eq!(next.delay, Duration::from_secs(300));
6274 assert_eq!(
6275 next.pace,
6276 PollPace::RateLimited {
6277 kind: RateLimitKind::Secondary
6278 },
6279 "the increase is reported, never hidden"
6280 );
6281 assert!(next.pace.is_throttled());
6282 assert_eq!(next.pace.as_str(), "rate_limited_secondary");
6283 }
6284
6285 #[test]
6288 fn the_retry_delay_is_an_absolute_floor_and_never_an_addend() {
6289 let now = fixtures::created_at();
6290 let mut schedule = PollSchedule::new(RefreshInterval::default());
6291
6292 let limited = RefreshState::RateLimited(RateLimited {
6293 kind: RateLimitKind::Primary,
6294 retry_after: Some(Duration::from_secs(300)),
6295 remaining: Some(0),
6296 reset_unix_secs: None,
6297 });
6298
6299 for _ in 0..5 {
6302 let next = schedule.next_poll(Some(&limited), now, &NoJitter);
6303 assert_eq!(
6304 next.delay,
6305 Duration::from_secs(300),
6306 "the delay is `max(interval, retry_delay)`; `interval + retry_delay` would \
6307 have compounded on every successive retry"
6308 );
6309 }
6310
6311 let brief = RefreshState::RateLimited(RateLimited {
6314 kind: RateLimitKind::Secondary,
6315 retry_after: Some(Duration::from_secs(5)),
6316 remaining: None,
6317 reset_unix_secs: None,
6318 });
6319 let next = schedule.next_poll(Some(&brief), now, &NoJitter);
6320 assert_eq!(
6321 next.delay,
6322 Duration::from_secs(60),
6323 "a short `retry-after` may not drop the loop below its own interval"
6324 );
6325 assert!(next.delay >= PollSchedule::floor());
6326 }
6327
6328 #[test]
6329 fn no_branch_of_the_schedule_can_go_below_the_thirty_second_floor() {
6330 let now = fixtures::created_at();
6331 let states = [
6332 None,
6333 Some(RefreshState::Offline),
6334 Some(RefreshState::RateLimited(RateLimited {
6335 kind: RateLimitKind::Secondary,
6336 retry_after: Some(Duration::from_secs(1)),
6337 remaining: None,
6338 reset_unix_secs: None,
6339 })),
6340 Some(RefreshState::LockedOut {
6341 retry_after: Duration::from_secs(1),
6342 }),
6343 Some(RefreshState::Unauthorized),
6344 Some(RefreshState::Forbidden { message: None }),
6345 Some(RefreshState::Failed {
6346 status: Some(500),
6347 message: "server error".into(),
6348 }),
6349 Some(RefreshState::Cancelled),
6350 ];
6351
6352 for state in &states {
6353 let mut schedule = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
6354 let next = schedule.next_poll(state.as_ref(), now, &NoJitter);
6355 assert!(
6356 next.delay >= PollSchedule::floor(),
6357 "{state:?} scheduled a poll {}ms away, under the 30-second floor",
6358 next.delay.as_millis()
6359 );
6360 }
6361 }
6362
6363 #[test]
6364 fn an_offline_run_backs_off_with_jitter_and_a_recovery_resets_it() {
6365 let now = fixtures::created_at();
6366 let mut schedule = PollSchedule::new(RefreshInterval::default());
6367
6368 let mut previous = Duration::ZERO;
6370 for consecutive in 1..=6_u32 {
6371 let next = schedule.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
6372 assert_eq!(next.pace, PollPace::Offline { consecutive });
6373 assert!(
6374 next.delay >= previous,
6375 "the back-off must not shrink while the outage continues"
6376 );
6377 assert!(next.delay >= Duration::from_secs(60));
6378 previous = next.delay;
6379 }
6380 assert!(previous <= MAX_OFFLINE_BACKOFF, "and it is capped");
6381
6382 let mut jittered = PollSchedule::new(RefreshInterval::default());
6385 let none = jittered.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
6386 let mut jittered = PollSchedule::new(RefreshInterval::default());
6387 let full = jittered.next_poll(Some(&RefreshState::Offline), now, &FixedJitter(0.999));
6388 assert!(full.delay > none.delay);
6389 assert!(full.delay <= none.delay.mul_f64(1.0 + JITTER_RATIO));
6390
6391 assert_eq!(schedule.consecutive_offline(), 6);
6393 let recovered = schedule.next_poll(None, now, &NoJitter);
6394 assert_eq!(recovered.pace, PollPace::Nominal);
6395 assert_eq!(recovered.delay, Duration::from_secs(60));
6396 assert_eq!(schedule.consecutive_offline(), 0);
6397 }
6398
6399 #[test]
6400 fn the_offline_state_states_the_twenty_four_hour_bound() {
6401 assert_eq!(
6402 GITHUB_CANCELS_QUEUED_JOBS_AFTER,
6403 Duration::from_secs(24 * 60 * 60)
6404 );
6405
6406 let brief = OfflineState::new(1, Duration::from_secs(120));
6407 let rendered = brief.to_string();
6408 assert!(rendered.contains("24 hours"), "{rendered}");
6409 assert!(rendered.contains("Retrying in 120s"), "{rendered}");
6410 assert!(!brief.has_outlasted_the_queue());
6411
6412 let long = brief.since(GITHUB_CANCELS_QUEUED_JOBS_AFTER + Duration::from_secs(1));
6413 assert!(long.has_outlasted_the_queue());
6414 assert!(
6415 long.to_string().contains("queued work has been lost"),
6416 "{long}"
6417 );
6418
6419 assert!(!OfflineState::new(9, Duration::from_secs(60)).has_outlasted_the_queue());
6421 }
6422
6423 fn quota(remaining: u64, reset_in_secs: i64, now: Timestamp) -> RateLimitSnapshot {
6428 RateLimitSnapshot {
6429 limit: 5_000,
6430 remaining,
6431 used: Some(5_000 - remaining),
6432 reset_unix_secs: u64::try_from(now.timestamp() + reset_in_secs).expect("positive"),
6433 observed_at: now,
6434 }
6435 }
6436
6437 fn quiet() -> PollConditions {
6438 PollConditions {
6439 quiet: true,
6440 quota: None,
6441 requests_per_pass: 0,
6442 }
6443 }
6444
6445 #[test]
6446 fn a_quiet_pass_polls_at_the_idle_interval_and_a_busy_one_at_the_active_interval() {
6447 let now = fixtures::created_at();
6448 let mut schedule =
6449 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6450
6451 let idle = schedule.next_poll_with(None, quiet(), now, &NoJitter);
6452 assert_eq!(idle.delay, Duration::from_secs(10));
6453 assert_eq!(idle.pace, PollPace::Idle);
6454 assert!(!idle.pace.is_throttled(), "idle is faster, not a slowdown");
6455
6456 let busy = schedule.next_poll_with(None, PollConditions::default(), now, &NoJitter);
6457 assert_eq!(busy.delay, Duration::from_secs(60));
6458 assert_eq!(busy.pace, PollPace::Nominal);
6459 }
6460
6461 #[test]
6462 fn a_schedule_never_told_an_idle_interval_keeps_its_active_one() {
6463 let now = fixtures::created_at();
6464 let mut schedule = PollSchedule::new(RefreshInterval::default());
6465 assert_eq!(
6466 schedule.next_poll_with(None, quiet(), now, &NoJitter).delay,
6467 Duration::from_secs(60)
6468 );
6469 }
6470
6471 #[test]
6472 fn an_idle_interval_is_clamped_between_its_floor_and_the_active_interval() {
6473 let schedule = PollSchedule::new(RefreshInterval::from_secs(30).expect("the floor"))
6474 .with_idle(Duration::from_secs(120));
6475 assert_eq!(
6476 schedule.idle(),
6477 Duration::from_secs(30),
6478 "an idle interval above the active one would make quiet targets slower to react"
6479 );
6480 let schedule =
6481 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(1));
6482 assert_eq!(schedule.idle(), PollSchedule::idle_floor());
6483 }
6484
6485 #[test]
6488 fn a_low_quota_stretches_polling_and_wins_over_idle() {
6489 let now = fixtures::created_at();
6490 let mut schedule =
6491 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6492
6493 let healthy = schedule.next_poll_with(
6494 None,
6495 PollConditions {
6496 quiet: true,
6497 quota: Some(quota(1_000, 3_000, now)),
6498 requests_per_pass: 0,
6499 },
6500 now,
6501 &NoJitter,
6502 );
6503 assert_eq!(healthy.pace, PollPace::Idle, "a fifth left is still normal");
6504
6505 let low = schedule.next_poll_with(
6506 None,
6507 PollConditions {
6508 quiet: true,
6509 quota: Some(quota(999, 3_000, now)),
6510 requests_per_pass: 0,
6511 },
6512 now,
6513 &NoJitter,
6514 );
6515 assert_eq!(low.delay, Duration::from_secs(120));
6516 assert_eq!(
6517 low.pace,
6518 PollPace::Stretched {
6519 pressure: PrimaryPressure::Low
6520 }
6521 );
6522 assert!(low.pace.is_throttled());
6523 assert_eq!(low.pace.as_str(), "stretched_low_budget");
6524
6525 let critical = schedule.next_poll_with(
6526 None,
6527 PollConditions {
6528 quiet: false,
6529 quota: Some(quota(100, 3_000, now)),
6530 requests_per_pass: 0,
6531 },
6532 now,
6533 &NoJitter,
6534 );
6535 assert_eq!(critical.delay, Duration::from_secs(240));
6536 assert_eq!(critical.pace.as_str(), "stretched_critical_budget");
6537 }
6538
6539 #[test]
6540 fn a_stretch_never_outlasts_the_quotas_reset_nor_undercuts_the_active_interval() {
6541 let now = fixtures::created_at();
6542 let mut schedule = PollSchedule::new(RefreshInterval::default());
6543
6544 let soon = schedule.next_poll_with(
6545 None,
6546 PollConditions {
6547 quiet: false,
6548 quota: Some(quota(10, 90, now)),
6549 requests_per_pass: 0,
6550 },
6551 now,
6552 &NoJitter,
6553 );
6554 assert_eq!(
6555 soon.delay,
6556 Duration::from_secs(90),
6557 "the bucket refills in 90 s; waiting 240 s buys nothing"
6558 );
6559
6560 let imminent = schedule.next_poll_with(
6561 None,
6562 PollConditions {
6563 quiet: false,
6564 quota: Some(quota(10, 5, now)),
6565 requests_per_pass: 0,
6566 },
6567 now,
6568 &NoJitter,
6569 );
6570 assert_eq!(imminent.delay, Duration::from_secs(60));
6571
6572 let expired = schedule.next_poll_with(
6573 None,
6574 PollConditions {
6575 quiet: true,
6576 quota: Some(quota(10, -5, now)),
6577 requests_per_pass: 0,
6578 },
6579 now,
6580 &NoJitter,
6581 );
6582 assert_eq!(
6583 expired.pace,
6584 PollPace::Nominal,
6585 "a reading from a window that has reset says nothing; the schedule here had no \
6586 idle interval, so it is the active one"
6587 );
6588 }
6589
6590 #[test]
6591 fn an_idle_target_too_large_to_poll_that_often_stretches_its_idle_interval() {
6592 let now = fixtures::created_at();
6593 let mut schedule =
6594 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6595 let small = schedule.next_poll_with(
6596 None,
6597 PollConditions {
6598 quiet: true,
6599 quota: None,
6600 requests_per_pass: 40,
6601 },
6602 now,
6603 &NoJitter,
6604 );
6605 assert_eq!(small.delay, Duration::from_secs(10), "ten repositories fit");
6606 let organization = schedule.next_poll_with(
6607 None,
6608 PollConditions {
6609 quiet: true,
6610 quota: None,
6611 requests_per_pass: 400,
6612 },
6613 now,
6614 &NoJitter,
6615 );
6616 assert_eq!(
6617 organization.delay,
6618 Duration::from_secs(80),
6619 "400 requests a pass at 300 a minute is one pass every 80 s"
6620 );
6621 assert_eq!(organization.pace, PollPace::Idle);
6622 }
6623
6624 #[test]
6625 fn a_rate_limit_wait_is_jittered_upward_only() {
6626 let now = fixtures::created_at();
6627 let mut schedule = PollSchedule::new(RefreshInterval::default());
6628 let limited = RefreshState::RateLimited(RateLimited {
6629 kind: RateLimitKind::Secondary,
6630 retry_after: Some(Duration::from_secs(300)),
6631 remaining: None,
6632 reset_unix_secs: None,
6633 });
6634 assert_eq!(
6635 schedule
6636 .next_poll(Some(&limited), now, &FixedJitter(0.5))
6637 .delay,
6638 Duration::from_secs(315),
6639 "every loop shares one quiet period; without jitter they all wake together"
6640 );
6641 assert_eq!(
6642 schedule.next_poll(Some(&limited), now, &NoJitter).delay,
6643 Duration::from_secs(300),
6644 "never shorter than GitHub asked"
6645 );
6646 }
6647
6648 #[tokio::test]
6649 async fn demand_this_host_did_not_serve_keeps_the_target_off_the_idle_pace() {
6650 let launcher =
6653 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 9, 9)]));
6654 let demand = Arc::new(FakeDemand::ready(2, &repo("acme/app")));
6655 let mut harness = Harness::build(
6656 host_with(1),
6657 launcher,
6658 demand,
6659 Arc::new(InProcessAllocationLock::new()),
6660 );
6661 harness
6662 .reconciler
6663 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6664 let policy = policy(1, "acme/app", 4);
6665
6666 let report = harness
6667 .reconciler
6668 .reconcile(std::slice::from_ref(&policy))
6669 .await;
6670
6671 assert_eq!(report.started, 0, "the host is full");
6672 assert_eq!(
6673 report.next_poll.pace,
6674 PollPace::Nominal,
6675 "matched demand it could not serve is work waiting, not idle"
6676 );
6677 }
6678
6679 #[tokio::test]
6680 async fn a_target_with_nothing_to_do_is_polled_at_the_idle_interval() {
6681 let mut harness = Harness::simple(4, 0, "acme/app");
6682 harness
6683 .reconciler
6684 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6685 let policy = policy(1, "acme/app", 4);
6686
6687 let report = harness
6688 .reconciler
6689 .reconcile(std::slice::from_ref(&policy))
6690 .await;
6691
6692 assert_eq!(report.next_poll.pace, PollPace::Idle);
6693 assert_eq!(report.next_poll.delay, Duration::from_secs(10));
6694 }
6695
6696 #[tokio::test]
6697 async fn a_target_whose_runs_are_changing_is_polled_at_the_active_interval() {
6698 let mut harness = Harness::simple(4, 0, "acme/app");
6699 harness
6700 .reconciler
6701 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6702 harness.demand.set(PollOutcome::Ready(
6703 QueuedDemand::of(repo("acme/app"), Vec::new()).with_busy(repo("acme/app")),
6704 ));
6705 let policy = policy(1, "acme/app", 4);
6706
6707 let report = harness
6708 .reconciler
6709 .reconcile(std::slice::from_ref(&policy))
6710 .await;
6711
6712 assert_eq!(report.next_poll.pace, PollPace::Nominal);
6713 assert_eq!(report.next_poll.delay, Duration::from_secs(60));
6714 }
6715
6716 #[tokio::test]
6717 async fn a_target_with_a_live_runner_is_polled_at_the_active_interval() {
6718 let launcher =
6719 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
6720 let demand = Arc::new(FakeDemand::ready(0, &repo("acme/app")));
6721 let mut harness = Harness::build(
6722 host_with(4),
6723 launcher,
6724 demand,
6725 Arc::new(InProcessAllocationLock::new()),
6726 );
6727 harness
6728 .reconciler
6729 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6730 let policy = policy(1, "acme/app", 4);
6731
6732 let report = harness
6733 .reconciler
6734 .reconcile(std::slice::from_ref(&policy))
6735 .await;
6736
6737 assert_eq!(
6738 report.next_poll.pace,
6739 PollPace::Nominal,
6740 "a job is running: its follow-on work is what the active interval watches for"
6741 );
6742 }
6743
6744 #[tokio::test]
6745 async fn a_pass_that_starts_a_runner_is_not_idle() {
6746 let mut harness = Harness::simple(4, 2, "acme/app");
6747 harness
6748 .reconciler
6749 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6750 let policy = policy(1, "acme/app", 4);
6751
6752 let report = harness
6753 .reconciler
6754 .reconcile(std::slice::from_ref(&policy))
6755 .await;
6756
6757 assert!(report.started > 0);
6758 assert_eq!(report.next_poll.pace, PollPace::Nominal);
6759 }
6760
6761 #[tokio::test]
6762 async fn the_quota_reported_with_a_reading_stretches_the_next_poll() {
6763 let mut harness = Harness::simple(4, 0, "acme/app");
6764 harness
6765 .reconciler
6766 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6767 let now = harness.clock.now();
6768 harness.demand.set(PollOutcome::Ready(
6769 QueuedDemand::of(repo("acme/app"), Vec::new())
6770 .with_rate_limit(Some(quota(200, 1_800, now))),
6771 ));
6772 let policy = policy(1, "acme/app", 4);
6773
6774 let report = harness
6775 .reconciler
6776 .reconcile(std::slice::from_ref(&policy))
6777 .await;
6778
6779 assert_eq!(
6780 report.next_poll.pace,
6781 PollPace::Stretched {
6782 pressure: PrimaryPressure::Critical
6783 }
6784 );
6785 assert_eq!(report.next_poll.delay, Duration::from_secs(240));
6786 }
6787
6788 #[tokio::test]
6797 async fn an_unreachable_github_starts_nothing_retains_everything_and_backs_off() {
6798 let live = vec![
6799 attempt_in(AttemptState::Busy, 1, 1),
6800 attempt_in(AttemptState::Starting, 2, 1),
6801 ];
6802 let launcher = Arc::new(FakeLauncher::new().seeded(live.clone()));
6803 let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
6804 let mut harness = Harness::build(
6805 host_with(8),
6806 Arc::clone(&launcher),
6807 Arc::clone(&demand),
6808 Arc::new(InProcessAllocationLock::new()),
6809 );
6810 let policy = policy(1, "acme/app", 8);
6811
6812 let report = harness
6813 .reconciler
6814 .reconcile(std::slice::from_ref(&policy))
6815 .await;
6816
6817 assert!(report.is_offline());
6818 assert_eq!(report.started, 0, "no new runner during an outage");
6819 assert_eq!(launcher.launches(), 0);
6820 assert_eq!(
6821 launcher.snapshot(),
6822 live,
6823 "existing runner processes are retained, untouched"
6824 );
6825 assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
6826 assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
6827 assert!(report.next_poll.delay >= Duration::from_secs(60));
6828 let offline = report.offline_state().expect("an offline state to display");
6829 assert!(offline.to_string().contains("24 hours"));
6830
6831 demand.set(PollOutcome::Ready(QueuedDemand::of(
6834 repo("acme/app"),
6835 jobs(2),
6836 )));
6837 let recovered = harness.reconciler.reconcile(&[policy]).await;
6838
6839 assert!(!recovered.is_offline());
6840 assert_eq!(recovered.next_poll.pace, PollPace::Nominal);
6841 assert_eq!(
6842 recovered.started, 1,
6843 "two queued jobs and one starting runner leave one job uncovered"
6844 );
6845 assert_eq!(recovered.allocations[0].active_owned, 2);
6846 assert_eq!(launcher.live_count(), 3);
6847 }
6848
6849 #[tokio::test]
6859 async fn one_offline_target_does_not_stop_a_reachable_one() {
6860 let mut harness = Harness::simple(8, 0, "acme/app");
6861 harness.demand.set_for(
6862 &ScaleTarget::repository("acme/app").unwrap(),
6863 PollOutcome::Ready(QueuedDemand::of(repo("acme/app"), jobs(2))),
6864 );
6865 harness.demand.set_for(
6866 &ScaleTarget::repository("acme/broken").unwrap(),
6867 PollOutcome::Failed(RefreshState::Offline),
6868 );
6869
6870 let report = harness
6871 .reconciler
6872 .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/broken", 4)])
6873 .await;
6874
6875 assert_eq!(
6876 report.started, 2,
6877 "the reachable target was served; an unreachable sibling repository must not \
6878 idle the host"
6879 );
6880 assert_eq!(report.unreadable, vec![PolicyId::from_u128(2)]);
6881 assert_eq!(harness.demand.polls().len(), 2, "both targets were polled");
6882
6883 assert!(report.is_offline());
6885 assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
6886 }
6887
6888 #[tokio::test]
6903 async fn a_day_long_outage_says_that_queued_work_has_already_been_lost() {
6904 let clock = Arc::new(FakeClock::default());
6905 let launcher = Arc::new(FakeLauncher::new());
6906 let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
6907 let mut reconciler = Reconciler::new(
6908 host_with(4),
6909 ReconcilerPorts {
6910 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
6911 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
6912 lock: Arc::new(InProcessAllocationLock::new()),
6913 directory: Arc::new(FakeDirectory::default()),
6914 clock: Arc::clone(&clock) as Arc<dyn Clock>,
6915 jitter: Arc::new(NoJitter),
6916 events: Arc::new(NoEvents),
6917 },
6918 );
6919 let policy = policy(1, "acme/app", 4);
6920
6921 let first = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6923 let state = first.offline_state().expect("an offline state");
6924 assert!(!state.has_outlasted_the_queue());
6925 assert!(
6926 state
6927 .to_string()
6928 .contains("an outage longer than that loses"),
6929 "{state}"
6930 );
6931
6932 clock.advance_secs(24 * 60 * 60 + 60);
6934 let later = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6935 let state = later.offline_state().expect("an offline state");
6936 assert!(state.has_outlasted_the_queue());
6937 assert!(
6938 state.to_string().contains("queued work has been lost"),
6939 "{state}"
6940 );
6941 assert_eq!(launcher.launches(), 0, "and still nothing was started");
6942
6943 demand.set(PollOutcome::Ready(QueuedDemand::of(
6946 repo("acme/app"),
6947 jobs(0),
6948 )));
6949 let recovered = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6950 assert!(recovered.offline_state().is_none());
6951 assert_eq!(reconciler.schedule().offline_for(clock.now()), None);
6952
6953 demand.set(PollOutcome::Failed(RefreshState::Offline));
6954 let again = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6955 assert!(
6956 !again
6957 .offline_state()
6958 .expect("an offline state")
6959 .has_outlasted_the_queue(),
6960 "a new outage must not inherit the age of the one before it"
6961 );
6962 }
6963
6964 #[tokio::test]
6980 async fn the_file_allocation_lock_excludes_a_second_holder_and_releases_on_drop() {
6981 let root = tempfile::tempdir().expect("a temporary directory");
6982 let paths = Arc::new(runner_manager_platform::paths::AppPaths::rooted_at(
6983 root.path(),
6984 ));
6985 let lock = FileAllocationLock::new(paths).with_wait(Duration::from_millis(50));
6986
6987 let held = lock.acquire().await.expect("an uncontended lock is free");
6988 assert!(
6989 matches!(lock.acquire().await, Err(AllocationLockBusy)),
6990 "a second holder was admitted; on Unix the lock is per open file description \
6991 and on Windows the share mode denies write, so this must be refused even \
6992 from inside the same process"
6993 );
6994
6995 drop(held);
6996 let regained = lock.acquire().await;
6997 assert!(
6998 regained.is_ok(),
6999 "dropping the guard is the only release there is, so a guard that does not \
7000 hold the `HostLock` leaves it held forever"
7001 );
7002 }
7003
7004 #[test]
7005 fn tee_events_reaches_both_sinks() {
7006 let left = Arc::new(EventLog::new());
7010 let right = Arc::new(EventLog::new());
7011 let tee = TeeEvents(
7012 Arc::clone(&left) as Arc<dyn EventSink>,
7013 Arc::clone(&right) as Arc<dyn EventSink>,
7014 );
7015
7016 tee.emit(LifecycleEvent::MonitorOnlySkipped {
7017 policy: PolicyId::from_u128(1),
7018 });
7019
7020 assert_eq!(left.count_of("monitor_only_skipped"), 1);
7021 assert_eq!(right.count_of("monitor_only_skipped"), 1);
7022 }
7023
7024 #[tokio::test]
7029 async fn the_repository_list_refreshes_far_more_slowly_than_the_demand_poll() {
7030 let clock = Arc::new(FakeClock::default());
7031 let directory = Arc::new(FakeDirectory::of(vec![repo("acme/one"), repo("acme/two")]));
7032 let cache = RepositoryCache::new(
7033 Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
7034 Arc::clone(&clock) as Arc<dyn Clock>,
7035 RefreshInterval::default(),
7036 );
7037 let target = ScaleTarget::organization("acme").unwrap();
7038
7039 assert_eq!(
7040 cache.ttl(),
7041 Duration::from_secs(60 * u64::from(REPOSITORY_LIST_REFRESH_MULTIPLE))
7042 );
7043
7044 for _ in 0..REPOSITORY_LIST_REFRESH_MULTIPLE {
7046 let scope = cache.scope_for(&target).await.unwrap();
7047 assert_eq!(scope.repositories().len(), 2);
7048 clock.advance_secs(60);
7049 }
7050 assert_eq!(
7051 directory.calls(),
7052 1,
7053 "re-listing an organization at demand-poll frequency is what exhausts the \
7054 shared request budget"
7055 );
7056 assert_eq!(cache.lookups(), 1);
7057
7058 cache.scope_for(&target).await.unwrap();
7060 assert_eq!(directory.calls(), 2);
7061 }
7062
7063 #[tokio::test]
7064 async fn a_repository_target_never_consults_the_directory() {
7065 let directory = Arc::new(FakeDirectory::of(vec![repo("acme/other")]));
7066 let cache = RepositoryCache::new(
7067 Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
7068 Arc::new(FakeClock::default()) as Arc<dyn Clock>,
7069 RefreshInterval::default(),
7070 );
7071 let target = ScaleTarget::repository("acme/app").unwrap();
7072
7073 let scope = cache.scope_for(&target).await.unwrap();
7074 assert_eq!(scope.repositories(), &[repo("acme/app")]);
7075 assert_eq!(directory.calls(), 0);
7076 }
7077
7078 #[tokio::test]
7079 async fn two_policies_on_one_target_cost_one_demand_poll_not_two() {
7080 let mut harness = Harness::simple(8, 4, "o/r");
7084 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7085 repo("o/r"),
7086 [
7087 fixtures::queued_job(&[HOST_LABEL]),
7088 fixtures::queued_job(&[HOST_LABEL]),
7089 fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
7090 fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
7091 ],
7092 )));
7093 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7094 isolated.activate().unwrap();
7095 let report = harness
7096 .reconciler
7097 .reconcile(&[policy(1, "o/r", 2), isolated])
7098 .await;
7099
7100 assert_eq!(harness.demand.polls().len(), 1);
7101 assert_eq!(report.targets_read, 1, "profiles share one target reading");
7102 assert_eq!(
7103 report.demand_requests,
7104 runner_manager_github::demand::DEMAND_REQUESTS_PER_REPOSITORY_PER_POLL,
7105 "one repository's worth of demand requests, not two policies' worth. Read from the constant rather than written as a literal so that repricing the poll cannot silently turn this into an assertion about the wrong thing"
7106 );
7107 assert_eq!(
7108 report.started, 4,
7109 "native and isolated profiles each get two"
7110 );
7111 }
7112
7113 #[tokio::test]
7114 async fn subset_labels_without_selector_scale_only_the_compatible_local_profile() {
7115 use runner_manager_domain::model::{Arch, HostLabel, Os};
7116 use runner_manager_domain::policy::RoutingLabels;
7117 let mut harness = Harness::build(
7118 fixtures::host()
7119 .os(Os::MacOs)
7120 .architecture(Arch::Arm64)
7121 .capacity(10)
7122 .build(),
7123 Arc::new(FakeLauncher::new()),
7124 Arc::new(FakeDemand::ready(0, &repo("o/r"))),
7125 Arc::new(InProcessAllocationLock::new()),
7126 );
7127 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7128 repo("o/r"),
7129 [
7130 fixtures::queued_job(&["self-hosted", "macOS", "ARM64"]),
7131 fixtures::queued_job(&["SELF-HOSTED", "MACOS", "arm64"]),
7132 fixtures::queued_job(&["self-hosted", "macOS", "X64"]),
7133 fixtures::queued_job(&["self-hosted", "Windows", "ARM64"]),
7134 fixtures::queued_job(&["self-hosted", "macOS", "ARM64", "gpu"]),
7135 fixtures::queued_job(&["rm-other-osx-arm64", "macOS", "ARM64"]),
7136 ],
7137 )));
7138 let mut local = fixtures::policy()
7139 .id(PolicyId::from_u128(1))
7140 .repository("o/r")
7141 .autoscale("macmini", 10)
7142 .routing_labels(RoutingLabels::derive(
7143 &HostLabel::new("macmini").unwrap(),
7144 Os::MacOs,
7145 Arch::Arm64,
7146 ))
7147 .active()
7148 .build();
7149 for label in ["self-hosted", "macos", "arm64", "extra"] {
7150 local.add_routing_label(fixtures::label(label)).unwrap();
7151 }
7152 let mut foreign = local.to_persisted();
7153 foreign.id = PolicyId::from_u128(2);
7154 foreign.host_id = HostId::from_u128(0xdead);
7155 let foreign = ScalePolicy::from_persisted(foreign).unwrap();
7156 let report = harness.reconciler.reconcile(&[local, foreign]).await;
7157 assert_eq!(report.started, 2);
7158 assert_eq!(report.targets_read, 1);
7159 assert!(harness.events.events().iter().any(|event| matches!(
7160 event,
7161 LifecycleEvent::DemandObserved {
7162 demand: 2,
7163 not_matched: 4,
7164 ..
7165 }
7166 )));
7167 assert!(
7168 !harness
7169 .events
7170 .events()
7171 .iter()
7172 .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
7173 );
7174 }
7175
7176 #[tokio::test]
7177 async fn isolated_overlap_never_falls_back_to_native_in_either_state_or_order() {
7178 use runner_manager_domain::execution::{
7179 Backend, ExecutionPolicy, ImageReference, ResourceLimits,
7180 };
7181 let mut native = policy(1, "o/r", 2);
7182 native
7183 .add_routing_label(fixtures::label("self-hosted"))
7184 .unwrap();
7185 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7186 isolated
7187 .add_routing_label(fixtures::label("self-hosted"))
7188 .unwrap();
7189 isolated
7190 .set_execution_policy(ExecutionPolicy::Isolated {
7191 backend: Backend::Oci,
7192 image: ImageReference::new(format!(
7193 "registry.example/runner@sha256:{}",
7194 "a".repeat(64)
7195 ))
7196 .unwrap(),
7197 resources: ResourceLimits {
7198 cpu_millis: 1000,
7199 memory_mib: 512,
7200 disk_mib: 1024,
7201 },
7202 })
7203 .unwrap();
7204 assert!(!isolated.may_start_runners());
7205 let mut active_isolated = isolated.clone();
7206 active_isolated.activate().unwrap();
7207 for profiles in [
7208 [native.clone(), isolated.clone()],
7209 [isolated.clone(), native.clone()],
7210 [native.clone(), active_isolated.clone()],
7211 [active_isolated, native.clone()],
7212 ] {
7213 let mut harness = Harness::simple(4, 0, "o/r");
7214 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7215 repo("o/r"),
7216 [fixtures::queued_job(&["SELF-HOSTED"])],
7217 )));
7218 let report = harness.reconciler.reconcile(&profiles).await;
7219 assert_eq!(report.started, 0);
7220 assert!(harness.events.events().iter().any(|event| matches!(
7221 event,
7222 LifecycleEvent::RoutingRefused {
7223 reason: RoutingRefusal::OverlappingProfiles,
7224 count: 1,
7225 ..
7226 }
7227 )));
7228 }
7229 let mut harness = Harness::simple(4, 0, "o/r");
7232 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7233 repo("o/r"),
7234 [fixtures::queued_job(&["rm-home-win-x64-py-isolated"])],
7235 )));
7236 assert_eq!(
7237 harness
7238 .reconciler
7239 .reconcile(&[native, isolated])
7240 .await
7241 .started,
7242 0
7243 );
7244 }
7245
7246 #[tokio::test]
7247 async fn shared_optional_labels_do_not_duplicate_native_and_isolated_demand() {
7248 let mut harness = Harness::simple(2, 0, "o/r");
7249 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7250 repo("o/r"),
7251 [
7252 fixtures::queued_job(&[HOST_LABEL, "gpu"]),
7253 fixtures::queued_job(&["rm-home-win-x64-py-isolated", "gpu"]),
7254 fixtures::queued_job(&["gpu"]),
7255 fixtures::unresolvable_job(),
7256 ],
7257 )));
7258 let mut native = policy(1, "o/r", 2);
7259 native.add_routing_label(fixtures::label("gpu")).unwrap();
7260 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7261 isolated.add_routing_label(fixtures::label("gpu")).unwrap();
7262 isolated.activate().unwrap();
7263
7264 let report = harness.reconciler.reconcile(&[native, isolated]).await;
7265 assert_eq!(harness.demand.polls().len(), 1);
7266 assert_eq!(report.targets_read, 1);
7267 assert_eq!(report.started, 2);
7268 assert_eq!(report.allocations.len(), 2);
7269 assert!(
7270 report
7271 .allocations
7272 .iter()
7273 .all(|allocation| allocation.to_start == 1)
7274 );
7275 assert!(harness.events.events().iter().any(|event| matches!(
7276 event,
7277 LifecycleEvent::RoutingRefused {
7278 reason: RoutingRefusal::OverlappingProfiles,
7279 count: 1,
7280 ..
7281 }
7282 )));
7283 assert_eq!(
7284 harness
7285 .events
7286 .events()
7287 .iter()
7288 .filter(|event| matches!(
7289 event,
7290 LifecycleEvent::DemandObserved {
7291 unresolvable: 1,
7292 ..
7293 }
7294 ))
7295 .count(),
7296 2,
7297 "an expression stays unresolvable for both profiles"
7298 );
7299 }
7300
7301 #[tokio::test]
7302 async fn an_exclusion_starts_a_runner_registered_with_it_only_where_it_holds() {
7303 let queued = || {
7304 PollOutcome::Ready(QueuedDemand::of(
7305 repo("o/r"),
7306 [
7307 fixtures::queued_job(&[HOST_LABEL, "not-wsl", "not-gpu"]),
7308 fixtures::queued_job(&[HOST_LABEL]),
7309 ],
7310 ))
7311 };
7312
7313 let mut native = Harness::simple(4, 0, "o/r");
7316 native.reconciler = native.reconciler.with_host_traits(HostTraits::wsl(false));
7317 native.demand.set(queued());
7318 let report = native.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7319 assert_eq!(report.started, 2);
7320 for labels in native.launcher.registered.lock().unwrap().iter() {
7321 assert_eq!(
7322 labels,
7323 &vec![HOST_LABEL.to_string(), "not-gpu".into(), "not-wsl".into()]
7324 );
7325 }
7326
7327 native.demand.set(PollOutcome::Ready(QueuedDemand::of(
7330 repo("o/r"),
7331 fixtures::queued_jobs(&[HOST_LABEL], 4),
7332 )));
7333 native.launcher.registered.lock().unwrap().clear();
7334 native.reconciler.reconcile(&[policy(1, "o/r", 6)]).await;
7335 let later = native.launcher.registered.lock().unwrap().clone();
7336 assert!(!later.is_empty());
7337 assert!(
7338 later
7339 .iter()
7340 .all(|labels| labels.contains(&"not-gpu".to_string()))
7341 );
7342
7343 let mut wsl = Harness::simple(4, 0, "o/r");
7345 wsl.reconciler = wsl.reconciler.with_host_traits(HostTraits::wsl(true));
7346 wsl.demand.set(queued());
7347 let report = wsl.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7348 assert_eq!(report.started, 1);
7349 assert_eq!(
7350 wsl.launcher.registered.lock().unwrap().as_slice(),
7351 [vec![HOST_LABEL.to_string(), "wsl".into()]]
7352 );
7353
7354 let mut gpu = Harness::simple(4, 0, "o/r");
7356 gpu.demand.set(queued());
7357 let mut carrying = policy(1, "o/r", 4);
7358 carrying.add_routing_label(fixtures::label("gpu")).unwrap();
7359 let report = gpu.reconciler.reconcile(&[carrying]).await;
7360 assert_eq!(report.started, 1);
7361 assert_eq!(
7362 gpu.launcher.registered.lock().unwrap().as_slice(),
7363 [vec![HOST_LABEL.to_string(), "gpu".into()]]
7364 );
7365 }
7366
7367 #[tokio::test]
7368 async fn without_exclusions_or_traits_a_runner_registers_exactly_its_profile_labels() {
7369 let mut harness = Harness::simple(4, 2, "o/r");
7370 let report = harness.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7371 assert_eq!(report.started, 2);
7372 assert_eq!(
7373 harness.launcher.registered.lock().unwrap().as_slice(),
7374 [vec![HOST_LABEL.to_string()], vec![HOST_LABEL.to_string()]]
7375 );
7376 }
7377
7378 #[tokio::test]
7379 async fn a_job_no_runner_can_satisfy_is_refused_by_name_not_counted() {
7380 let mut harness = Harness::simple(4, 0, "o/r");
7381 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7382 repo("o/r"),
7383 [
7384 fixtures::queued_job(&[HOST_LABEL, "not-"]),
7385 fixtures::queued_job(&["rm-other-linux-x64", "gpu", "not-gpu"]),
7387 ],
7388 )));
7389 let report = harness.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7390 assert_eq!(report.started, 0);
7391 assert!(harness.events.events().iter().any(|event| matches!(
7392 event,
7393 LifecycleEvent::RoutingRefused {
7394 reason: RoutingRefusal::UnsatisfiableExclusion,
7395 count: 1,
7396 ..
7397 }
7398 )));
7399 assert!(harness.events.events().iter().any(|event| matches!(
7400 event,
7401 LifecycleEvent::DemandObserved {
7402 demand: 0,
7403 unresolvable: 2,
7404 ..
7405 }
7406 )));
7407 }
7408
7409 #[tokio::test]
7410 async fn missing_required_selector_labels_do_not_match_and_overlaps_show_remedies() {
7411 let mut harness = Harness::simple(4, 0, "o/r");
7412 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7413 repo("o/r"),
7414 [fixtures::queued_job(&[
7415 HOST_LABEL,
7416 "rm-home-win-x64-py-isolated",
7417 ])],
7418 )));
7419 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7420 isolated.activate().unwrap();
7421 let report = harness
7422 .reconciler
7423 .reconcile(&[policy(1, "o/r", 2), isolated])
7424 .await;
7425 assert_eq!(report.started, 0);
7426 assert_eq!(
7427 harness
7428 .events
7429 .events()
7430 .iter()
7431 .filter(|event| matches!(
7432 event,
7433 LifecycleEvent::DemandObserved {
7434 demand: 0,
7435 not_matched: 1,
7436 ..
7437 }
7438 ))
7439 .count(),
7440 2
7441 );
7442
7443 let mut superset = policy(1, "o/r", 2);
7447 assert!(
7448 superset
7449 .add_routing_label(fixtures::label("rm-home-win-x64-py-isolated"))
7450 .unwrap()
7451 );
7452 let mut sibling = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7453 sibling.activate().unwrap();
7454 let mut unique = Harness::simple(4, 0, "o/r");
7455 unique.demand.set(PollOutcome::Ready(QueuedDemand::of(
7456 repo("o/r"),
7457 [fixtures::queued_job(&[
7458 HOST_LABEL,
7459 "rm-home-win-x64-py-isolated",
7460 ])],
7461 )));
7462 let report = unique.reconciler.reconcile(&[superset, sibling]).await;
7463 assert_eq!(report.started, 1);
7464 assert_eq!(
7465 report
7466 .allocations
7467 .iter()
7468 .find(|allocation| allocation.to_start == 1)
7469 .unwrap()
7470 .policy_id,
7471 PolicyId::from_u128(1)
7472 );
7473 assert!(
7474 !unique
7475 .events
7476 .events()
7477 .iter()
7478 .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
7479 );
7480
7481 let mut overlap = Harness::simple(4, 1, "o/r");
7482 let report = overlap
7483 .reconciler
7484 .reconcile(&[policy(1, "o/r", 2), policy(2, "o/r", 2)])
7485 .await;
7486 assert_eq!(report.started, 0);
7487 assert!(overlap.events.events().iter().any(|event| matches!(
7488 event,
7489 LifecycleEvent::RoutingRefused {
7490 reason: RoutingRefusal::OverlappingProfiles,
7491 count: 1,
7492 ..
7493 }
7494 )));
7495 assert!(
7496 RoutingRefusal::OverlappingProfiles
7497 .to_string()
7498 .contains("repair")
7499 );
7500 }
7501
7502 #[tokio::test]
7507 async fn a_failed_launch_stops_the_run_and_is_reported_without_free_text() {
7508 let launcher = Arc::new(FakeLauncher::new());
7509 launcher.fail_next(FailureReason::Other("token ghp_0123456789abcdef".into()));
7510 let mut harness = Harness::build(
7511 host_with(4),
7512 Arc::clone(&launcher),
7513 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
7514 Arc::new(InProcessAllocationLock::new()),
7515 );
7516
7517 let report = harness
7518 .reconciler
7519 .reconcile(&[policy(1, "acme/app", 4)])
7520 .await;
7521 assert_eq!(report.started, 0);
7522 assert_eq!(report.allocations[0].to_start, 3, "the decision stands");
7523
7524 let failures: Vec<&'static str> = harness
7525 .events
7526 .events()
7527 .into_iter()
7528 .filter_map(|event| match event {
7529 LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
7530 _ => None,
7531 })
7532 .collect();
7533 assert_eq!(failures, vec!["other"]);
7534 assert!(
7535 !failures[0].contains("ghp_"),
7536 "an event carried a `FailureReason::Other` detail verbatim"
7537 );
7538 }
7539
7540 #[tokio::test]
7545 async fn an_unfit_host_registers_no_runner_and_reports_host_unfit() {
7546 let launcher = Arc::new(FakeLauncher::new());
7547 let unfit = Arc::new(AtomicBool::new(true));
7548 let events = Arc::new(EventLog::new());
7549 let mut reconciler = Reconciler::new(
7550 host_with(4),
7551 ReconcilerPorts {
7552 demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
7553 launcher: Arc::new(FitnessGatedLauncher::new(
7554 Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
7555 Arc::clone(&unfit),
7556 )),
7557 lock: Arc::new(InProcessAllocationLock::new()),
7558 directory: Arc::new(FakeDirectory::default()),
7559 clock: Arc::new(FakeClock::default()) as Arc<dyn Clock>,
7560 jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
7561 events: Arc::clone(&events) as Arc<dyn EventSink>,
7562 },
7563 );
7564 let policies = [policy(1, "acme/app", 4)];
7565
7566 let report = reconciler.reconcile(&policies).await;
7567 assert_eq!(report.started, 0);
7568 assert_eq!(launcher.launches(), 0, "the inner launcher must not run");
7569 let reasons: Vec<&'static str> = events
7570 .events()
7571 .into_iter()
7572 .filter_map(|event| match event {
7573 LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
7574 _ => None,
7575 })
7576 .collect();
7577 assert_eq!(reasons, ["host_unfit"]);
7578
7579 unfit.store(false, Ordering::Release);
7580 let report = reconciler.reconcile(&policies).await;
7581 assert_eq!(report.started, 2);
7582 assert_eq!(launcher.launches(), 2);
7583 }
7584
7585 #[tokio::test]
7586 async fn a_held_allocation_lock_starts_nothing_and_says_so() {
7587 let launcher = Arc::new(FakeLauncher::new());
7588 let mut harness = Harness::build(
7589 host_with(4),
7590 Arc::clone(&launcher),
7591 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
7592 Arc::new(HeldLock),
7593 );
7594
7595 let report = harness
7596 .reconciler
7597 .reconcile(&[policy(1, "acme/app", 4)])
7598 .await;
7599 assert_eq!(report.started, 0);
7600 assert_eq!(
7601 report.deferred, 3,
7602 "three runners were granted and none was created; `deferred` counts grants, \
7603 not policies -- it reported `1` when a policy that launched two of five and \
7604 then lost the lock had left three unstarted"
7605 );
7606 assert_eq!(launcher.launches(), 0);
7607 assert_eq!(harness.events.count_of("allocation_deferred"), 1);
7608 assert!(
7609 harness
7610 .events
7611 .events()
7612 .iter()
7613 .any(|event| matches!(event, LifecycleEvent::AllocationDeferred { count: 3, .. })),
7614 "the event carries the same number the report does"
7615 );
7616 assert_eq!(
7617 report.allocations.len(),
7618 1,
7619 "the intent is still reported, so an operator staring at a queue sees why \
7620 nothing started"
7621 );
7622 }
7623
7624 #[tokio::test]
7625 async fn a_foreign_or_draining_policy_is_reported_by_name_and_polls_nothing() {
7626 let mut harness = Harness::simple(8, 5, "acme/app");
7627
7628 let foreign = fixtures::policy()
7629 .id(PolicyId::from_u128(1))
7630 .repository("acme/app")
7631 .host(HostId::from_u128(0xdead))
7632 .autoscale("office", 4)
7633 .active()
7634 .build();
7635 let mut draining = policy(2, "acme/app", 4);
7636 draining.request_disable().unwrap();
7637
7638 let report = harness.reconciler.reconcile(&[foreign, draining]).await;
7639
7640 assert_eq!(report.started, 0);
7641 assert_eq!(
7642 harness.demand.polls().len(),
7643 0,
7644 "neither can act on an answer"
7645 );
7646 let factors: Vec<LimitingFactor> = report
7647 .allocations
7648 .iter()
7649 .map(|a| a.limiting_factor)
7650 .collect();
7651 assert!(factors.contains(&LimitingFactor::ForeignHost));
7652 assert!(factors.contains(&LimitingFactor::NotReconciling));
7653 }
7654
7655 #[tokio::test]
7656 async fn an_unreadable_repository_list_makes_the_target_unreadable_not_empty() {
7657 #[derive(Debug)]
7660 struct BrokenDirectory;
7661
7662 #[async_trait::async_trait]
7663 impl RepositoryDirectory for BrokenDirectory {
7664 async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
7665 Err(InventoryError::Cancelled)
7666 }
7667 }
7668
7669 let launcher = Arc::new(FakeLauncher::new());
7670 let events = Arc::new(EventLog::new());
7671 let mut reconciler = Reconciler::new(
7672 host_with(4),
7673 ReconcilerPorts {
7674 demand: Arc::new(FakeDemand::default()),
7675 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
7676 lock: Arc::new(InProcessAllocationLock::new()),
7677 directory: Arc::new(BrokenDirectory),
7678 clock: Arc::new(FakeClock::default()),
7679 jitter: Arc::new(NoJitter),
7680 events: Arc::clone(&events) as Arc<dyn EventSink>,
7681 },
7682 );
7683
7684 let org_policy = fixtures::policy()
7685 .id(PolicyId::from_u128(1))
7686 .organization("acme")
7687 .autoscale("home", 4)
7688 .active()
7689 .build();
7690
7691 let report = reconciler.reconcile(&[org_policy]).await;
7692 assert_eq!(report.started, 0);
7693 assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
7694 assert_eq!(events.count_of("target_unreadable"), 1);
7695 }
7696
7697 fn every_event() -> Vec<LifecycleEvent> {
7709 let policy = PolicyId::from_u128(0xabcd_ef01);
7710 let attempt = AttemptId::from_u128(0x1234_5678);
7711 vec![
7712 LifecycleEvent::DemandObserved {
7713 policy,
7714 demand: u32::MAX,
7715 not_matched: u32::MAX,
7716 unresolvable: u32::MAX,
7717 complete: false,
7718 },
7719 LifecycleEvent::TargetUnreadable {
7720 policy,
7721 reason: unreadable_reason(&RefreshState::Failed {
7722 status: Some(500),
7723 message: "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz"
7724 .into(),
7725 }),
7726 },
7727 LifecycleEvent::Allocated {
7728 policy,
7729 demand: u32::MAX,
7730 desired: u16::MAX,
7731 active_owned: 7,
7732 headroom: 9,
7733 to_start: 2,
7734 limiting: LimitingFactor::HostCapacity,
7735 },
7736 LifecycleEvent::MonitorOnlySkipped { policy },
7737 LifecycleEvent::RunnerStarted { policy, attempt },
7738 LifecycleEvent::RunnerStartFailed {
7739 policy,
7740 reason: failure_reason_kind(&FailureReason::Other(
7741 "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
7742 )),
7743 },
7744 LifecycleEvent::AllocationDeferred { policy, count: 4 },
7745 LifecycleEvent::AttemptsUnreadable {
7746 reason: failure_reason_kind(&FailureReason::Other(
7747 "x-api-key: ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
7748 )),
7749 },
7750 LifecycleEvent::HostCapacityUnreadable {
7751 reason: HostCapacityError::MissingHost,
7752 },
7753 LifecycleEvent::HostCapacityUnreadable {
7754 reason: HostCapacityError::JournalUnreadable,
7755 },
7756 LifecycleEvent::AttemptCleanFailed {
7757 policy,
7758 attempt,
7759 reason: failure_reason_kind(&FailureReason::ProcessExitedUnexpectedly),
7760 },
7761 LifecycleEvent::AttemptCleaned {
7762 policy,
7763 attempt,
7764 outcome: OutcomeKind::IdleExit,
7765 },
7766 LifecycleEvent::ScaleDownRefused { policy, attempt },
7767 LifecycleEvent::PollScheduled {
7768 retry_in_ms: 900_000,
7769 pace: PollPace::RateLimited {
7770 kind: RateLimitKind::Primary,
7771 },
7772 },
7773 ]
7774 }
7775
7776 #[test]
7785 fn no_emitted_event_can_carry_a_credential() {
7786 use runner_manager_platform::logging::redact;
7787
7788 for event in every_event() {
7789 let displayed = event.to_string();
7790 assert_eq!(
7791 redact(&displayed),
7792 displayed,
7793 "`{}` renders something `d1`'s sink would have to redact",
7794 event.name()
7795 );
7796
7797 let debugged = format!("{event:?}");
7798 assert_eq!(
7799 redact(&debugged),
7800 debugged,
7801 "`{}`'s Debug renders something `d1`'s sink would have to redact",
7802 event.name()
7803 );
7804 }
7805
7806 let secret = "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz";
7809 assert_ne!(
7810 redact(secret),
7811 secret,
7812 "the scan above proves nothing if `redact` no longer recognises a credential"
7813 );
7814 }
7815
7816 #[test]
7817 fn every_field_name_this_sink_emits_is_one_d1_allows() {
7818 use runner_manager_platform::logging::is_field_allowed;
7819
7820 for field in [
7824 "event",
7825 "policy_id",
7826 "attempt_id",
7827 "attempt_state",
7828 "demand",
7829 "desired",
7830 "capacity",
7831 "headroom",
7832 "count",
7833 "reason",
7834 "outcome",
7835 "mode",
7836 "lock",
7837 "retry_in_ms",
7838 "state",
7839 ] {
7840 assert!(
7841 is_field_allowed(field),
7842 "`{field}` is not on `d1`'s allow-list, so this sink would emit \
7843 `[redacted]` in its place and the line would lose its meaning"
7844 );
7845 }
7846 }
7847
7848 #[test]
7849 fn every_failure_reason_has_a_credential_free_kind() {
7850 for reason in FailureReason::ALL {
7851 let kind = failure_reason_kind(&reason);
7852 assert!(!kind.is_empty());
7853 assert!(
7854 kind.chars().all(|c| c.is_ascii_lowercase() || c == '_'),
7855 "`{kind}` is not a fixed identifier"
7856 );
7857 }
7858 assert_eq!(
7859 failure_reason_kind(&FailureReason::Other("ghp_secret".into())),
7860 "other",
7861 "the detail of an `Other` reason never reaches an event"
7862 );
7863 let categories = [
7864 IsolationProviderFailure::NotInstalled,
7865 IsolationProviderFailure::PermissionDenied,
7866 IsolationProviderFailure::ImageUnavailableOrIncompatible,
7867 IsolationProviderFailure::DiskQuotaUnavailable,
7868 IsolationProviderFailure::RuntimeOperationFailed,
7869 ];
7870 let mut kinds = std::collections::BTreeSet::new();
7871 for category in categories {
7872 let kind = failure_reason_kind(&FailureReason::IsolationProvider(category));
7873 assert_ne!(kind, "other");
7874 assert!(!kind.contains("secret"));
7875 kinds.insert(kind);
7876 }
7877 assert_eq!(kinds.len(), categories.len());
7878 }
7879
7880 fn production_half_of(source: &str) -> String {
7895 let production = source
7896 .split_once("\n#[cfg(test)]")
7897 .map_or(source, |(production, _)| production);
7898 production
7899 .lines()
7900 .filter(|line| !line.trim_start().starts_with("//"))
7901 .collect::<Vec<_>>()
7902 .join("\n")
7903 }
7904
7905 fn this_file_above_its_tests_without_prose() -> String {
7907 production_half_of(include_str!("reconcile.rs"))
7908 }
7909
7910 fn normalise_for_scan(text: &str) -> String {
7929 text.to_ascii_lowercase().replace(['_', ' '], "")
7930 }
7931
7932 const FORBIDDEN: &[&str] = &[
7942 concat!("fn ", "acquire", "_job"),
7943 concat!("fn ", "claim", "_job"),
7944 concat!("fn ", "lease", "_job"),
7945 concat!("fn ", "reserve", "_job"),
7946 concat!("fn ", "ack", "nowledge"),
7947 concat!("struct ", "Job", "Lease"),
7948 concat!("struct ", "Job", "Claim"),
7949 concat!("struct ", "Job", "Reservation"),
7950 ];
7951
7952 fn forbidden_shape_in(source: &str) -> Option<&'static str> {
7953 let haystack = normalise_for_scan(source);
7954 FORBIDDEN
7955 .iter()
7956 .copied()
7957 .find(|forbidden| haystack.contains(&normalise_for_scan(forbidden)))
7958 }
7959
7960 #[test]
7994 fn nothing_in_this_crate_reserves_or_claims_a_job() {
7995 const SOURCES: &[(&str, &str)] = &[
7996 ("dependency_caches.rs", include_str!("dependency_caches.rs")),
7997 ("docker_cleanup.rs", include_str!("docker_cleanup.rs")),
7998 (
7999 "docker_cleanup/http.rs",
8000 include_str!("docker_cleanup/http.rs"),
8001 ),
8002 (
8003 "docker_cleanup/policy.rs",
8004 include_str!("docker_cleanup/policy.rs"),
8005 ),
8006 (
8007 "docker_cleanup/proxy.rs",
8008 include_str!("docker_cleanup/proxy.rs"),
8009 ),
8010 (
8011 "docker_cleanup/tests.rs",
8012 include_str!("docker_cleanup/tests.rs"),
8013 ),
8014 ("lib.rs", include_str!("lib.rs")),
8015 ("lifecycle.rs", include_str!("lifecycle.rs")),
8016 (
8017 "lifecycle/windows_hyperv.rs",
8018 include_str!("lifecycle/windows_hyperv.rs"),
8019 ),
8020 ("oci.rs", include_str!("oci.rs")),
8021 ("package.rs", include_str!("package.rs")),
8022 ("reconcile.rs", include_str!("reconcile.rs")),
8023 ];
8024
8025 fn walk(directory: &std::path::Path, prefix: &str, found: &mut Vec<String>) {
8026 for entry in std::fs::read_dir(directory).expect("the crate's own src/ is readable") {
8027 let entry = entry.expect("a readable directory entry");
8028 let name = entry.file_name().to_string_lossy().into_owned();
8029 let joined = if prefix.is_empty() {
8032 name.clone()
8033 } else {
8034 format!("{prefix}/{name}")
8035 };
8036 if entry.path().is_dir() {
8037 walk(&entry.path(), &joined, found);
8038 } else if name.ends_with(".rs") {
8039 found.push(joined);
8040 }
8041 }
8042 }
8043
8044 let mut listed: Vec<&str> = SOURCES.iter().map(|(name, _)| *name).collect();
8045 listed.sort_unstable();
8046 let mut on_disk = Vec::new();
8047 walk(
8048 std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/src")),
8049 "",
8050 &mut on_disk,
8051 );
8052 on_disk.sort_unstable();
8053 assert_eq!(
8054 listed, on_disk,
8055 "a source file was added or removed; this scan claims to cover the whole crate \
8056 and a stale list makes that claim false"
8057 );
8058
8059 for (name, source) in SOURCES {
8060 assert_eq!(
8061 forbidden_shape_in(&production_half_of(source)),
8062 None,
8063 "{name} names a forbidden shape: there is no `AcquireJobs` equivalent over \
8064 REST, and a local lease coordinates this host with itself and with nothing \
8065 else. If an owner decision restored one, that decision belongs in this \
8066 module's documentation and in this test before it belongs in the code"
8067 );
8068 }
8069
8070 assert!(
8073 forbidden_shape_in("async fn acquire_jobs(&self) -> Vec<Job> { todo!() }").is_some(),
8074 "the scan above proves nothing if the needles no longer match"
8075 );
8076 }
8077
8078 #[test]
8106 fn the_label_predicate_is_b1s_and_this_module_only_applies_it() {
8107 let production = this_file_above_its_tests_without_prose();
8108
8109 assert!(
8110 production.contains("DemandTally"),
8111 "the reconciliation loop must tally queued jobs against this policy's routing \
8112 labels. A production half that named `DemandTally` nowhere would mean the \
8113 label filtering had been removed, and a repository whose jobs target \
8114 `ubuntu-latest` would drive its policy toward `max_capacity` again"
8115 );
8116
8117 for second_implementation in ["RunsOnMatch", "UnresolvableRunsOn"] {
8118 assert!(
8119 !production.contains(second_implementation),
8120 "the reconciliation loop names `{second_implementation}`, which is the \
8121 vocabulary of deciding one job's `runs-on` -- and `RoutingLabels::tally` \
8122 has already decided it. This module applies the predicate and does not \
8123 re-implement it; if an owner decision changed that, it belongs in this \
8124 module's documentation and in this test before it belongs in the code"
8125 );
8126 }
8127 }
8128
8129 #[test]
8137 fn demand_is_the_queued_jobs_this_policy_can_actually_serve() {
8138 let policy = policy(1, "acme/app", 10);
8139 let reading = QueuedDemand::of(
8140 repo("acme/app"),
8141 [
8142 fixtures::queued_job(&[HOST_LABEL]),
8143 fixtures::queued_job(&[HOST_LABEL]),
8144 fixtures::queued_job(&["ubuntu-latest"]),
8145 fixtures::unresolvable_job(),
8146 ],
8147 );
8148
8149 let tally = demand_for(&policy, &reading, &[]);
8150
8151 assert_eq!(
8152 tally.demand(),
8153 2,
8154 "only the jobs whose required labels this policy carries are demand"
8155 );
8156 assert_eq!(
8157 tally.not_matched, 1,
8158 "a `ubuntu-latest` job is somebody else's work; counting it would start a \
8159 runner that idles until it times out"
8160 );
8161 assert_eq!(
8162 tally.unresolvable.len(),
8163 1,
8164 "an unresolvable `runs-on` is never demand and never discarded"
8165 );
8166 }
8167
8168 #[test]
8171 fn an_organization_policy_tallies_every_repository_its_scope_covers() {
8172 let mut per_repository = BTreeMap::new();
8173 per_repository.insert(repo("acme/left"), fixtures::queued_jobs(&[HOST_LABEL], 3));
8174 per_repository.insert(repo("acme/right"), fixtures::queued_jobs(&[HOST_LABEL], 4));
8175 let reading = QueuedDemand::new(per_repository);
8176
8177 let repository_policy = policy(1, "acme/left", 10);
8178 assert_eq!(
8179 demand_for(&repository_policy, &reading, &[]).demand(),
8180 3,
8181 "a repository target reads its own repository's queue and not the aggregate"
8182 );
8183
8184 let org_policy = fixtures::policy()
8185 .id(PolicyId::from_u128(2))
8186 .organization("acme")
8187 .autoscale("home", 10)
8188 .active()
8189 .build();
8190 assert_eq!(
8191 demand_for(&org_policy, &reading, &[]).demand(),
8192 7,
8193 "an organization policy serves any repository in its scope, so its demand is \
8194 the whole aggregate's"
8195 );
8196 }
8197
8198 #[test]
8199 fn the_accepted_over_count_is_bounded_by_the_two_ceilings_and_nothing_else() {
8200 let host = host_with(2);
8205 let policy = policy(1, "acme/app", 5);
8206 let attempts: Vec<RunnerAttempt> = Vec::new();
8207 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
8208
8209 let allocation = allocator.allocate(&policy, u32::MAX);
8210 assert_eq!(allocation.desired, 5, "max_capacity beats reported demand");
8211 assert_eq!(allocation.to_start, 2, "host_capacity beats max_capacity");
8212 assert_eq!(allocation.limiting_factor, LimitingFactor::HostCapacity);
8213 }
8214
8215 #[derive(Debug, Default)]
8218 struct FixedDirector {
8219 directed: Mutex<BTreeMap<PolicyId, u32>>,
8220 started: Mutex<Vec<(PolicyId, AttemptId)>>,
8221 failed: Mutex<Vec<(PolicyId, Option<AttemptId>)>>,
8222 older: Mutex<std::collections::VecDeque<usize>>,
8224 passed_over: AtomicUsize,
8225 orders: Mutex<BTreeMap<PolicyId, u64>>,
8227 }
8228
8229 impl DemandDirector for FixedDirector {
8230 fn directed_demand(&self, policy: &ScalePolicy) -> Option<u32> {
8231 self.directed.lock().unwrap().get(&policy.id).copied()
8232 }
8233
8234 fn launched(&self, policy: &ScalePolicy, attempt: AttemptId) {
8235 self.started.lock().unwrap().push((policy.id, attempt));
8236 }
8237
8238 fn start_failed(&self, policy: &ScalePolicy, attempt: Option<AttemptId>) {
8239 self.failed.lock().unwrap().push((policy.id, attempt));
8240 }
8241
8242 fn next_waiting(&self, policy: &ScalePolicy) -> Option<u64> {
8243 self.orders.lock().unwrap().get(&policy.id).copied()
8244 }
8245
8246 fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
8247 let mut older = self.older.lock().unwrap();
8248 if older.len() > 1 {
8249 older.pop_front().unwrap_or(0)
8250 } else {
8251 older.front().copied().unwrap_or(0)
8252 }
8253 }
8254
8255 fn pass_over_older(&self, _policy: &ScalePolicy) {
8256 self.passed_over.fetch_add(1, Ordering::SeqCst);
8257 self.older.lock().unwrap().clear();
8258 }
8259 }
8260
8261 #[derive(Debug)]
8264 struct HeldBringUp {
8265 inner: Arc<FakeLauncher>,
8266 barrier: Arc<tokio::sync::Barrier>,
8267 fail: Option<FailureReason>,
8268 waits: Mutex<Vec<ReservationWaits>>,
8269 }
8270
8271 #[async_trait::async_trait]
8272 impl RunnerLauncher for HeldBringUp {
8273 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
8274 self.inner.host_capacity(host).await
8275 }
8276
8277 async fn supervise(
8278 &self,
8279 policy: &ScalePolicy,
8280 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
8281 self.inner.supervise(policy).await
8282 }
8283
8284 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
8285 self.inner.attempts().await
8286 }
8287
8288 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
8289 self.inner.launch(request).await
8290 }
8291
8292 async fn finish_launch(
8293 &self,
8294 _policy: &ScalePolicy,
8295 attempt: RunnerAttempt,
8296 waits: ReservationWaits,
8297 ) -> Result<RunnerAttempt, LaunchFailure> {
8298 self.waits.lock().unwrap().push(waits);
8299 self.barrier.wait().await;
8300 match &self.fail {
8301 Some(reason) => Err(LaunchFailure::new(reason.clone())),
8302 None => Ok(attempt),
8303 }
8304 }
8305
8306 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
8307 self.inner.clean(attempt).await
8308 }
8309 }
8310
8311 fn directed_reconciler(
8312 capacity: u16,
8313 launcher: Arc<dyn RunnerLauncher>,
8314 lock: Arc<dyn AllocationLock>,
8315 director: Arc<FixedDirector>,
8316 ) -> Reconciler {
8317 Reconciler::new(
8318 host_with(capacity),
8319 ReconcilerPorts {
8320 demand: Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
8321 launcher,
8322 lock,
8323 directory: Arc::new(FakeDirectory::default()),
8324 clock: Arc::new(FakeClock::default()),
8325 jitter: Arc::new(NoJitter),
8326 events: Arc::new(NoEvents),
8327 },
8328 )
8329 .with_director(director as Arc<dyn DemandDirector>)
8330 }
8331
8332 #[tokio::test]
8335 async fn two_targets_bring_up_their_runners_at_once() {
8336 let fake = Arc::new(FakeLauncher::new());
8337 let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
8338 let barrier = Arc::new(tokio::sync::Barrier::new(2));
8339 let held = Arc::new(HeldBringUp {
8340 inner: Arc::clone(&fake),
8341 barrier,
8342 fail: None,
8343 waits: Mutex::default(),
8344 });
8345 let director = Arc::new(FixedDirector::default());
8346 let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
8347 director.directed.lock().unwrap().insert(one.id, 1);
8348 director.directed.lock().unwrap().insert(two.id, 1);
8349 let first = directed_reconciler(
8350 2,
8351 Arc::clone(&held) as _,
8352 Arc::clone(&lock),
8353 Arc::clone(&director),
8354 );
8355 let second = directed_reconciler(2, held as _, lock, director);
8356
8357 let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
8358 tokio::join!(
8359 first.start_directed(std::slice::from_ref(&one)),
8360 second.start_directed(std::slice::from_ref(&two)),
8361 )
8362 })
8363 .await
8364 .expect("neither waits for the other's bring-up");
8365 assert_eq!((a.started, b.started), (1, 1));
8366 }
8367
8368 #[tokio::test]
8371 async fn a_runner_coming_up_holds_its_slot_against_another_target() {
8372 let fake = Arc::new(FakeLauncher::new());
8373 let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
8374 let barrier = Arc::new(tokio::sync::Barrier::new(2));
8375 let held = Arc::new(HeldBringUp {
8376 inner: Arc::clone(&fake),
8377 barrier: Arc::clone(&barrier),
8378 fail: None,
8379 waits: Mutex::default(),
8380 });
8381 let director = Arc::new(FixedDirector::default());
8382 let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
8383 director.directed.lock().unwrap().insert(one.id, 1);
8384 director.directed.lock().unwrap().insert(two.id, 1);
8385 let first = directed_reconciler(
8386 1,
8387 Arc::clone(&held) as _,
8388 Arc::clone(&lock),
8389 Arc::clone(&director),
8390 );
8391 let second = directed_reconciler(1, held as _, lock, director);
8392
8393 let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
8394 tokio::join!(first.start_directed(std::slice::from_ref(&one)), async {
8395 while fake.live_count() == 0 {
8396 tokio::task::yield_now().await;
8397 }
8398 let report = second.start_directed(std::slice::from_ref(&two)).await;
8399 barrier.wait().await;
8400 report
8401 })
8402 })
8403 .await
8404 .expect("the first is let go");
8405 assert_eq!((a.started, b.started), (1, 0));
8406 assert_eq!(fake.live_count(), 1);
8407 }
8408
8409 #[tokio::test]
8412 async fn a_failed_bring_up_is_a_failed_start() {
8413 let fake = Arc::new(FakeLauncher::new());
8414 let held = Arc::new(HeldBringUp {
8415 inner: fake,
8416 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8417 fail: Some(FailureReason::JitRequestFailed),
8418 waits: Mutex::default(),
8419 });
8420 let director = Arc::new(FixedDirector::default());
8421 let pool = policy(1, "acme/app", 2);
8422 director.directed.lock().unwrap().insert(pool.id, 1);
8423 let reconciler = directed_reconciler(
8424 2,
8425 held as _,
8426 Arc::new(InProcessAllocationLock::new()),
8427 Arc::clone(&director),
8428 );
8429
8430 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8431
8432 assert_eq!(report.started, 0);
8433 let started = director.started.lock().unwrap()[0].1;
8434 assert_eq!(
8435 *director.failed.lock().unwrap(),
8436 vec![(pool.id, Some(started))]
8437 );
8438 }
8439
8440 fn scripted_older(director: &FixedDirector, looks: &[(usize, usize)]) {
8441 *director.older.lock().unwrap() = looks
8442 .iter()
8443 .flat_map(|(value, count)| std::iter::repeat_n(*value, *count))
8444 .collect();
8445 }
8446
8447 #[tokio::test(start_paused = true)]
8451 async fn a_younger_launch_reserves_after_older_ones_while_they_make_progress() {
8452 let held = Arc::new(HeldBringUp {
8453 inner: Arc::new(FakeLauncher::new()),
8454 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8455 fail: None,
8456 waits: Mutex::default(),
8457 });
8458 let director = Arc::new(FixedDirector::default());
8459 let pool = policy(1, "acme/app", 3);
8460 director.directed.lock().unwrap().insert(pool.id, 1);
8461 let reconciler = directed_reconciler(
8462 5,
8463 Arc::clone(&held) as _,
8464 Arc::new(InProcessAllocationLock::new()),
8465 Arc::clone(&director),
8466 );
8467
8468 let looks =
8471 (OLDER_LAUNCH_TURN.as_millis() * 3 / 4 / OLDER_LAUNCH_POLL.as_millis()) as usize;
8472 scripted_older(&director, &[(3, looks), (2, looks), (1, looks), (0, 1)]);
8473 let began = tokio::time::Instant::now();
8474 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8475 assert_eq!(report.started, 1);
8476 assert_eq!(
8477 began.elapsed(),
8478 OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap()
8479 );
8480 assert_eq!(director.passed_over.load(Ordering::SeqCst), 0);
8481 assert_eq!(
8482 held.waits.lock().unwrap()[0].order,
8483 OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap(),
8484 "the wait is in the launch's timings"
8485 );
8486 }
8487
8488 #[tokio::test]
8491 async fn a_pass_hands_a_directed_launch_to_the_launch_side() {
8492 let fake = Arc::new(FakeLauncher::new());
8493 let director = Arc::new(FixedDirector::default());
8494 let pool = policy(1, "acme/app", 3);
8495 director.directed.lock().unwrap().insert(pool.id, 1);
8496 let signal = Arc::new(tokio::sync::Notify::new());
8497 let mut reconciler = directed_reconciler(
8498 5,
8499 Arc::clone(&fake) as _,
8500 Arc::new(InProcessAllocationLock::new()),
8501 Arc::clone(&director),
8502 )
8503 .with_launch_side(Arc::clone(&signal));
8504
8505 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8506
8507 assert_eq!(report.started, 0);
8508 assert_eq!(fake.live_count(), 0);
8509 assert_eq!(report.allocations[0].to_start, 1);
8510 tokio::time::timeout(Duration::from_secs(1), signal.notified())
8511 .await
8512 .expect("the launch side is told");
8513
8514 director.directed.lock().unwrap().insert(pool.id, 0);
8516 reconciler.reconcile(std::slice::from_ref(&pool)).await;
8517 assert!(
8518 tokio::time::timeout(Duration::from_millis(50), signal.notified())
8519 .await
8520 .is_err()
8521 );
8522 }
8523
8524 fn pool_profile(id: u128, max: u16) -> ScalePolicy {
8526 let labels = fixtures::routing_labels("home")
8527 .with_labels([pool_label(runner_manager_domain::model::Os::Windows)]);
8528 fixtures::policy()
8529 .id(PolicyId::from_u128(id))
8530 .repository("acme/app")
8531 .autoscale("home", max)
8532 .routing_labels(labels)
8533 .active()
8534 .build()
8535 }
8536
8537 fn pool_queue(pool: usize, plain: usize) -> Arc<FakeDemand> {
8540 let mut queue = fixtures::queued_jobs(&[HOST_LABEL, "rm-pool-windows"], pool);
8541 queue.extend(fixtures::queued_jobs(&[HOST_LABEL], plain));
8542 let demand = Arc::new(FakeDemand::default());
8543 demand.set(PollOutcome::Ready(QueuedDemand::of(
8544 repo("acme/app"),
8545 queue,
8546 )));
8547 demand
8548 }
8549
8550 fn reconciler_reading(
8551 demand: Arc<FakeDemand>,
8552 launcher: Arc<FakeLauncher>,
8553 director: Arc<FixedDirector>,
8554 ) -> Reconciler {
8555 Reconciler::new(
8556 host_with(10),
8557 ReconcilerPorts {
8558 demand,
8559 launcher,
8560 lock: Arc::new(InProcessAllocationLock::new()),
8561 directory: Arc::new(FakeDirectory::default()),
8562 clock: Arc::new(FakeClock::default()),
8563 jitter: Arc::new(NoJitter),
8564 events: Arc::new(NoEvents),
8565 },
8566 )
8567 .with_director(director as Arc<dyn DemandDirector>)
8568 }
8569
8570 #[tokio::test]
8575 async fn a_directed_pool_profile_serves_the_jobs_that_ask_for_no_pool() {
8576 let (profile, demand) = (pool_profile(1, 5), pool_queue(1, 2));
8577 let fake = Arc::new(FakeLauncher::new());
8578 let director = Arc::new(FixedDirector::default());
8579 director.directed.lock().unwrap().insert(profile.id, 1);
8580 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director);
8581
8582 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8583 assert_eq!(
8584 report.started, 3,
8585 "one runner for the launch, one for each job that asks for no pool"
8586 );
8587 }
8588
8589 #[tokio::test]
8592 async fn a_directed_pool_profile_leaves_pool_jobs_to_the_launches() {
8593 let (profile, demand) = (pool_profile(1, 5), pool_queue(2, 0));
8594 let fake = Arc::new(FakeLauncher::new());
8595 let director = Arc::new(FixedDirector::default());
8596 director.directed.lock().unwrap().insert(profile.id, 0);
8597 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director);
8598
8599 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8600
8601 assert_eq!(report.started, 0);
8602 assert_eq!(fake.live_count(), 0);
8603 }
8604
8605 #[tokio::test]
8608 async fn a_pass_with_a_launch_side_starts_the_non_pool_jobs_runner() {
8609 let (profile, demand) = (pool_profile(1, 5), pool_queue(0, 1));
8610 let fake = Arc::new(FakeLauncher::new());
8611 let director = Arc::new(FixedDirector::default());
8612 director.directed.lock().unwrap().insert(profile.id, 0);
8613 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director)
8614 .with_launch_side(Arc::new(tokio::sync::Notify::new()));
8615
8616 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8617
8618 assert_eq!(report.started, 1);
8619 assert_eq!(fake.live_count(), 1);
8620
8621 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8623 assert_eq!(report.started, 0);
8624 assert_eq!(fake.live_count(), 1);
8625 }
8626
8627 #[tokio::test(start_paused = true)]
8630 async fn an_older_launch_that_cannot_reserve_is_passed_over_once() {
8631 let fake = Arc::new(FakeLauncher::new());
8632 let director = Arc::new(FixedDirector::default());
8633 let pool = policy(1, "acme/app", 3);
8634 director.directed.lock().unwrap().insert(pool.id, 1);
8635 let reconciler = directed_reconciler(
8636 5,
8637 Arc::clone(&fake) as _,
8638 Arc::new(InProcessAllocationLock::new()),
8639 Arc::clone(&director),
8640 );
8641
8642 scripted_older(&director, &[(1, 1)]);
8643 let began = tokio::time::Instant::now();
8644 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8645 assert_eq!(report.started, 1, "started anyway");
8646 assert!(began.elapsed() >= OLDER_LAUNCH_TURN);
8647 assert!(began.elapsed() < OLDER_LAUNCH_TURN + Duration::from_secs(1));
8648 assert_eq!(director.passed_over.load(Ordering::SeqCst), 1);
8649
8650 director.directed.lock().unwrap().insert(pool.id, 2);
8651 let began = tokio::time::Instant::now();
8652 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8653 assert_eq!(report.started, 1);
8654 assert_eq!(began.elapsed(), Duration::ZERO);
8655 }
8656
8657 #[tokio::test]
8660 async fn a_targets_profiles_reserve_oldest_launch_first() {
8661 let director = Arc::new(FixedDirector::default());
8662 let (younger, older) = (policy(1, "acme/app", 3), policy(2, "acme/app", 3));
8663 director.directed.lock().unwrap().insert(younger.id, 1);
8664 director.directed.lock().unwrap().insert(older.id, 1);
8665 director.orders.lock().unwrap().insert(younger.id, 9);
8666 director.orders.lock().unwrap().insert(older.id, 4);
8667 let reconciler = directed_reconciler(
8668 5,
8669 Arc::new(FakeLauncher::new()) as _,
8670 Arc::new(InProcessAllocationLock::new()),
8671 Arc::clone(&director),
8672 );
8673
8674 let report = reconciler
8675 .start_directed(&[younger.clone(), older.clone()])
8676 .await;
8677
8678 assert_eq!(report.started, 2);
8679 let order: Vec<PolicyId> = director
8680 .started
8681 .lock()
8682 .unwrap()
8683 .iter()
8684 .map(|(policy, _)| *policy)
8685 .collect();
8686 assert_eq!(order, vec![older.id, younger.id]);
8687 }
8688
8689 fn serving<'a>(
8691 reconciler: &'a Reconciler,
8692 pool: &ScalePolicy,
8693 signal: &'a tokio::sync::Notify,
8694 stop: tokio::sync::watch::Receiver<bool>,
8695 ) -> impl std::future::Future<Output = ()> + 'a {
8696 let pool = pool.clone();
8697 async move {
8698 let policies = move || vec![pool.clone()];
8699 reconciler.serve_launches(&policies, signal, stop).await;
8700 }
8701 }
8702
8703 #[tokio::test]
8707 async fn a_launch_accepted_during_a_batch_starts_without_waiting_for_it() {
8708 let fake = Arc::new(FakeLauncher::new());
8709 let held = Arc::new(HeldBringUp {
8710 inner: Arc::clone(&fake),
8711 barrier: Arc::new(tokio::sync::Barrier::new(2)),
8712 fail: None,
8713 waits: Mutex::default(),
8714 });
8715 let director = Arc::new(FixedDirector::default());
8716 let pool = policy(1, "acme/app", 3);
8717 director.directed.lock().unwrap().insert(pool.id, 1);
8718 let reconciler = directed_reconciler(
8719 5,
8720 held as _,
8721 Arc::new(InProcessAllocationLock::new()),
8722 Arc::clone(&director),
8723 );
8724 let signal = tokio::sync::Notify::new();
8725 let (stop, stopped) = tokio::sync::watch::channel(false);
8726
8727 tokio::time::timeout(Duration::from_secs(10), async {
8728 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8729 signal.notify_one();
8730 while director.started.lock().unwrap().is_empty() {
8731 tokio::task::yield_now().await;
8732 }
8733 director.directed.lock().unwrap().insert(pool.id, 2);
8734 signal.notify_one();
8735 while director.started.lock().unwrap().len() < 2 {
8736 tokio::task::yield_now().await;
8737 }
8738 stop.send_replace(true);
8739 })
8740 })
8741 .await
8742 .expect("the second came up beside the first, and stopping ends it");
8743 assert_eq!(fake.live_count(), 2);
8744 }
8745
8746 #[tokio::test]
8749 async fn a_launch_during_a_batch_takes_no_slot_that_batch_holds() {
8750 let fake = Arc::new(FakeLauncher::new());
8751 let held = Arc::new(HeldBringUp {
8752 inner: Arc::clone(&fake),
8753 barrier: Arc::new(tokio::sync::Barrier::new(2)),
8754 fail: None,
8755 waits: Mutex::default(),
8756 });
8757 let barrier = Arc::clone(&held.barrier);
8758 let director = Arc::new(FixedDirector::default());
8759 let pool = policy(1, "acme/app", 3);
8760 director.directed.lock().unwrap().insert(pool.id, 1);
8761 let reconciler = directed_reconciler(
8762 1,
8763 held as _,
8764 Arc::new(InProcessAllocationLock::new()),
8765 Arc::clone(&director),
8766 );
8767 let signal = tokio::sync::Notify::new();
8768 let (stop, stopped) = tokio::sync::watch::channel(false);
8769
8770 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8771 signal.notify_one();
8772 while director.started.lock().unwrap().is_empty() {
8773 tokio::task::yield_now().await;
8774 }
8775 director.directed.lock().unwrap().insert(pool.id, 2);
8776 signal.notify_one();
8777 for _ in 0..100 {
8778 tokio::task::yield_now().await;
8779 }
8780 assert_eq!(
8781 director.started.lock().unwrap().len(),
8782 1,
8783 "the host is full"
8784 );
8785 assert_eq!(fake.live_count(), 1);
8786 stop.send_replace(true);
8787 barrier.wait().await;
8788 });
8789 }
8790
8791 #[tokio::test(start_paused = true)]
8795 async fn a_reserved_runner_comes_up_before_the_next_reservation_ends() {
8796 let held = Arc::new(HeldBringUp {
8797 inner: Arc::new(FakeLauncher::new()),
8798 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8799 fail: None,
8800 waits: Mutex::default(),
8801 });
8802 let director = Arc::new(FixedDirector::default());
8803 let pool = policy(1, "acme/app", 3);
8804 director.directed.lock().unwrap().insert(pool.id, 2);
8805 let reconciler = directed_reconciler(
8806 5,
8807 Arc::clone(&held) as _,
8808 Arc::new(InProcessAllocationLock::new()),
8809 Arc::clone(&director),
8810 );
8811 let signal = tokio::sync::Notify::new();
8812 let (stop, stopped) = tokio::sync::watch::channel(false);
8813
8814 scripted_older(&director, &[(0, 1), (1, 1)]);
8816 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8817 signal.notify_one();
8818 tokio::time::sleep(Duration::from_millis(1)).await;
8819 assert_eq!(
8820 held.waits.lock().unwrap().len(),
8821 1,
8822 "the first is coming up"
8823 );
8824 tokio::time::sleep(OLDER_LAUNCH_TURN + Duration::from_secs(1)).await;
8825 assert_eq!(held.waits.lock().unwrap().len(), 2);
8826 stop.send_replace(true);
8827 });
8828 }
8829
8830 #[tokio::test(start_paused = true)]
8833 async fn a_launch_that_did_not_start_is_retried_and_a_stop_ends_it() {
8834 let held = Arc::new(HeldBringUp {
8835 inner: Arc::new(FakeLauncher::new()),
8836 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8837 fail: Some(FailureReason::JitRequestFailed),
8838 waits: Mutex::default(),
8839 });
8840 let director = Arc::new(FixedDirector::default());
8841 let pool = policy(1, "acme/app", 3);
8842 director.directed.lock().unwrap().insert(pool.id, 1);
8843 director.orders.lock().unwrap().insert(pool.id, 1);
8844 let reconciler = directed_reconciler(
8845 5,
8846 held as _,
8847 Arc::new(InProcessAllocationLock::new()),
8848 Arc::clone(&director),
8849 );
8850 let signal = tokio::sync::Notify::new();
8851 let (stop, stopped) = tokio::sync::watch::channel(false);
8852
8853 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8854 signal.notify_one();
8855 tokio::time::sleep(Duration::from_millis(1)).await;
8856 assert_eq!(director.failed.lock().unwrap().len(), 1);
8857 director.directed.lock().unwrap().insert(pool.id, 2);
8859 tokio::time::sleep(LAUNCH_RETRY - Duration::from_secs(1)).await;
8860 assert_eq!(director.failed.lock().unwrap().len(), 1, "not yet");
8861 tokio::time::sleep(Duration::from_secs(2)).await;
8862 assert_eq!(director.failed.lock().unwrap().len(), 2, "retried");
8863
8864 stop.send_replace(true);
8865 signal.notify_one();
8866 });
8867 assert_eq!(director.failed.lock().unwrap().len(), 2);
8868 }
8869
8870 #[tokio::test]
8874 async fn a_directed_policy_starts_only_what_it_was_directed_to_within_its_ceilings() {
8875 let demand = pool_queue(5, 0);
8876 let launcher = Arc::new(FakeLauncher::new());
8877 let director = Arc::new(FixedDirector::default());
8878 let mut reconciler = Reconciler::new(
8879 host_with(10),
8880 ReconcilerPorts {
8881 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
8882 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
8883 lock: Arc::new(InProcessAllocationLock::new()),
8884 directory: Arc::new(FakeDirectory::default()),
8885 clock: Arc::new(FakeClock::default()),
8886 jitter: Arc::new(NoJitter),
8887 events: Arc::new(NoEvents),
8888 },
8889 )
8890 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8891 let pool = pool_profile(1, 2);
8892
8893 director.directed.lock().unwrap().insert(pool.id, 0);
8895 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8896 assert_eq!(report.started, 0);
8897
8898 director.directed.lock().unwrap().insert(pool.id, 1);
8900 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8901 assert_eq!(report.started, 1);
8902 assert_eq!(director.started.lock().unwrap().len(), 1);
8903
8904 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8906 assert_eq!(report.started, 0);
8907
8908 director.directed.lock().unwrap().insert(pool.id, 3);
8910 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8911 assert_eq!(report.started, 1);
8912 assert_eq!(launcher.live_count(), 2);
8913
8914 let polls = demand.polls().len();
8916 director.directed.lock().unwrap().insert(pool.id, 0);
8917 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8918 assert_eq!(report.started, 0);
8919 let fresh = Arc::new(FakeLauncher::new());
8920 let starter = Reconciler::new(
8921 host_with(10),
8922 ReconcilerPorts {
8923 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
8924 launcher: Arc::clone(&fresh) as Arc<dyn RunnerLauncher>,
8925 lock: Arc::new(InProcessAllocationLock::new()),
8926 directory: Arc::new(FakeDirectory::default()),
8927 clock: Arc::new(FakeClock::default()),
8928 jitter: Arc::new(NoJitter),
8929 events: Arc::new(NoEvents),
8930 },
8931 )
8932 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8933 let wide = pool_profile(2, 6);
8934 director.directed.lock().unwrap().insert(wide.id, 6);
8935 let report = starter.start_directed(std::slice::from_ref(&wide)).await;
8936 assert_eq!(report.started, 6, "six launches, one call");
8937 assert_eq!(demand.polls().len(), polls, "GitHub was not polled");
8938 assert_eq!(
8940 starter
8941 .start_directed(&[policy(3, "acme/app", 2)])
8942 .await
8943 .started,
8944 0
8945 );
8946
8947 let failing = Arc::new(FakeLauncher::new());
8949 let mut reconciler = Reconciler::new(
8950 host_with(10),
8951 ReconcilerPorts {
8952 demand: Arc::new(FakeDemand::failing(RefreshState::Offline)),
8953 launcher: Arc::clone(&failing) as Arc<dyn RunnerLauncher>,
8954 lock: Arc::new(InProcessAllocationLock::new()),
8955 directory: Arc::new(FakeDirectory::default()),
8956 clock: Arc::new(FakeClock::default()),
8957 jitter: Arc::new(NoJitter),
8958 events: Arc::new(NoEvents),
8959 },
8960 )
8961 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8962 director.directed.lock().unwrap().insert(pool.id, 1);
8963 assert_eq!(reconciler.reconcile(&[pool]).await.started, 1);
8964
8965 let own = Arc::new(FakeLauncher::new());
8967 let mut reconciler = Reconciler::new(
8968 host_with(10),
8969 ReconcilerPorts {
8970 demand: Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
8971 launcher: Arc::clone(&own) as Arc<dyn RunnerLauncher>,
8972 lock: Arc::new(InProcessAllocationLock::new()),
8973 directory: Arc::new(FakeDirectory::default()),
8974 clock: Arc::new(FakeClock::default()),
8975 jitter: Arc::new(NoJitter),
8976 events: Arc::new(NoEvents),
8977 },
8978 )
8979 .with_director(Arc::new(OwnDemand));
8980 assert_eq!(
8981 reconciler
8982 .reconcile(&[policy(1, "acme/app", 2)])
8983 .await
8984 .started,
8985 2
8986 );
8987 }
8988}