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 places_plain_jobs(&self, _policy: &ScalePolicy) -> bool {
2450 false
2451 }
2452
2453 fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
2457 0
2458 }
2459
2460 fn pass_over_older(&self, _policy: &ScalePolicy) {}
2463}
2464
2465const OLDER_LAUNCH_TURN: Duration = Duration::from_secs(2);
2472const OLDER_LAUNCH_POLL: Duration = Duration::from_millis(10);
2473
2474const LAUNCH_RETRY: Duration = Duration::from_secs(30);
2476
2477#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
2479pub struct ReservationWaits {
2480 pub order: Duration,
2482 pub lock: Duration,
2484}
2485
2486#[derive(Default)]
2488struct ReservationRound {
2489 reserved: BTreeMap<PolicyId, (u32, u32)>,
2492 exhausted: BTreeSet<PolicyId>,
2493}
2494
2495type Reserving<'a> = std::pin::Pin<
2497 Box<
2498 dyn std::future::Future<Output = (Option<Reserved>, ReservationRound, Vec<ScalePolicy>)>
2499 + Send
2500 + 'a,
2501 >,
2502>;
2503
2504struct Reserved {
2506 policy: ScalePolicy,
2507 attempt: RunnerAttempt,
2508 waits: ReservationWaits,
2509 reserved_at: std::time::Instant,
2510 _hold: LaunchHold,
2512}
2513
2514pub trait MemoryGate: fmt::Debug + Send + Sync {
2519 fn admit(&self, policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict;
2522
2523 fn admitted(&self, _policy: &ScalePolicy) {}
2526}
2527
2528#[derive(Debug, Default, Clone, Copy)]
2530pub struct NoMemoryGate;
2531
2532impl MemoryGate for NoMemoryGate {
2533 fn admit(&self, _policy: &ScalePolicy, _attempts: &[RunnerAttempt]) -> MemoryVerdict {
2534 MemoryVerdict::Unchecked
2535 }
2536}
2537
2538#[derive(Debug, Default, Clone, Copy)]
2540pub struct OwnDemand;
2541
2542impl DemandDirector for OwnDemand {
2543 fn directed_demand(&self, _policy: &ScalePolicy) -> Option<u32> {
2544 None
2545 }
2546
2547 fn launched(&self, _policy: &ScalePolicy, _attempt: AttemptId) {}
2548}
2549
2550#[derive(Clone)]
2556pub struct ReconcilerPorts {
2557 pub demand: Arc<dyn DemandSource>,
2558 pub launcher: Arc<dyn RunnerLauncher>,
2559 pub lock: Arc<dyn AllocationLock>,
2560 pub directory: Arc<dyn RepositoryDirectory>,
2561 pub clock: Arc<dyn Clock>,
2562 pub jitter: Arc<dyn Jitter>,
2563 pub events: Arc<dyn EventSink>,
2564}
2565
2566impl fmt::Debug for ReconcilerPorts {
2567 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2568 f.debug_struct("ReconcilerPorts").finish_non_exhaustive()
2569 }
2570}
2571
2572#[derive(Debug, Clone, Default)]
2579pub struct ReconcileReport {
2580 pub allocations: Vec<Allocation>,
2582 pub monitor_only: Vec<PolicyId>,
2584 pub unreadable: Vec<PolicyId>,
2586 pub targets_read: u16,
2594 pub started: u16,
2596 pub replacement_intents: u16,
2599 pub cleaned: u16,
2601 pub idle_exits: u16,
2604 pub failures: u16,
2606 pub deferred: u16,
2615 pub memory_deferred: u16,
2618 pub attempts_unreadable: u16,
2637 pub host_capacity_unreadable: u16,
2639 pub clean_failures: u16,
2641 pub failure: Option<RefreshState>,
2643 pub offline: Option<OfflineState>,
2646 pub next_poll: NextPoll,
2648 pub demand_requests: u32,
2650}
2651
2652impl ReconcileReport {
2653 #[must_use]
2680 pub const fn reached_github(&self) -> bool {
2681 self.targets_read > 0
2682 }
2683}
2684
2685impl Default for NextPoll {
2686 fn default() -> Self {
2687 Self {
2688 delay: PollSchedule::floor(),
2689 pace: PollPace::Nominal,
2690 }
2691 }
2692}
2693
2694impl ReconcileReport {
2695 #[must_use]
2697 pub fn is_offline(&self) -> bool {
2698 matches!(self.failure, Some(RefreshState::Offline))
2699 }
2700
2701 #[must_use]
2703 pub const fn offline_state(&self) -> Option<&OfflineState> {
2704 self.offline.as_ref()
2705 }
2706
2707 #[must_use]
2709 pub const fn starts_nothing(&self) -> bool {
2710 self.started == 0
2711 }
2712}
2713
2714#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
2716pub struct ScaleDownReport {
2717 pub removed: u16,
2719 pub refused_busy: u16,
2721 pub clean_failures: u16,
2723 pub retained: u16,
2726 pub attempts_unreadable: bool,
2736}
2737
2738impl ScaleDownReport {
2739 #[must_use]
2744 pub const fn is_conclusive(&self) -> bool {
2745 !self.attempts_unreadable
2746 }
2747}
2748
2749const CLEAN_RETRY_INITIAL: Duration = Duration::from_secs(30);
2757
2758const CLEAN_RETRY_MAXIMUM: Duration = Duration::from_secs(30 * 60);
2760
2761#[derive(Debug, Clone, Copy)]
2763struct CleanRetry {
2764 failures: u32,
2765 not_before: Timestamp,
2766}
2767
2768#[derive(Debug, Default)]
2772pub(crate) struct CleanBackoff {
2773 retries: BTreeMap<AttemptId, CleanRetry>,
2774}
2775
2776impl CleanBackoff {
2777 const SCHEDULE: crate::lifecycle::RetryPolicy =
2779 crate::lifecycle::RetryPolicy::bounded(u32::MAX, CLEAN_RETRY_INITIAL, CLEAN_RETRY_MAXIMUM);
2780
2781 pub(crate) fn is_due(&self, attempt: AttemptId, now: Timestamp) -> bool {
2783 self.retries
2784 .get(&attempt)
2785 .is_none_or(|retry| now >= retry.not_before)
2786 }
2787
2788 pub(crate) fn failed(&mut self, attempt: AttemptId, now: Timestamp) {
2790 let failures = self
2791 .retries
2792 .get(&attempt)
2793 .map_or(1, |retry| retry.failures.saturating_add(1));
2794 let delay = Self::SCHEDULE.delay(failures);
2795 self.retries.insert(
2796 attempt,
2797 CleanRetry {
2798 failures,
2799 not_before: now + chrono::Duration::from_std(delay).unwrap_or_default(),
2801 },
2802 );
2803 }
2804
2805 pub(crate) fn succeeded(&mut self, attempt: AttemptId) {
2807 self.retries.remove(&attempt);
2808 }
2809
2810 fn retain(&mut self, present: &BTreeSet<AttemptId>) {
2813 self.retries.retain(|attempt, _| present.contains(attempt));
2814 }
2815}
2816
2817#[derive(Debug)]
2823pub struct Reconciler {
2824 host: Host,
2825 demand: Arc<dyn DemandSource>,
2826 launcher: Arc<dyn RunnerLauncher>,
2827 lock: Arc<dyn AllocationLock>,
2828 repositories: RepositoryCache,
2829 clock: Arc<dyn Clock>,
2830 jitter: Arc<dyn Jitter>,
2831 events: Arc<dyn EventSink>,
2832 schedule: PollSchedule,
2833 clean_backoff: CleanBackoff,
2834 director: Arc<dyn DemandDirector>,
2835 launch_side: Option<Arc<tokio::sync::Notify>>,
2838 memory: Arc<dyn MemoryGate>,
2839 host_labels: Vec<Label>,
2842 seen_exclusions: Mutex<BTreeMap<PolicyId, BTreeSet<Label>>>,
2845}
2846
2847impl Reconciler {
2848 #[must_use]
2850 pub fn new(host: Host, ports: ReconcilerPorts) -> Self {
2851 let interval = host.refresh_interval;
2852 let repositories = RepositoryCache::new(
2853 Arc::clone(&ports.directory),
2854 Arc::clone(&ports.clock),
2855 interval,
2856 );
2857 Self {
2858 host,
2859 demand: ports.demand,
2860 launcher: ports.launcher,
2861 lock: ports.lock,
2862 repositories,
2863 clock: ports.clock,
2864 jitter: ports.jitter,
2865 events: ports.events,
2866 schedule: PollSchedule::new(interval),
2867 clean_backoff: CleanBackoff::default(),
2868 director: Arc::new(OwnDemand),
2869 launch_side: None,
2870 memory: Arc::new(NoMemoryGate),
2871 host_labels: Vec::new(),
2872 seen_exclusions: Mutex::default(),
2873 }
2874 }
2875
2876 #[must_use]
2879 pub fn with_host_traits(mut self, traits: HostTraits) -> Self {
2880 self.host_labels = traits.labels();
2881 self
2882 }
2883
2884 #[must_use]
2886 pub fn with_memory_gate(mut self, memory: Arc<dyn MemoryGate>) -> Self {
2887 self.memory = memory;
2888 self
2889 }
2890
2891 #[must_use]
2896 pub fn with_launch_side(mut self, signal: Arc<tokio::sync::Notify>) -> Self {
2897 self.launch_side = Some(signal);
2898 self
2899 }
2900
2901 #[must_use]
2904 pub fn with_director(mut self, director: Arc<dyn DemandDirector>) -> Self {
2905 self.director = director;
2906 self
2907 }
2908
2909 #[must_use]
2910 pub const fn host(&self) -> &Host {
2911 &self.host
2912 }
2913
2914 #[must_use]
2915 pub const fn schedule(&self) -> &PollSchedule {
2916 &self.schedule
2917 }
2918
2919 pub fn set_intervals(&mut self, active: RefreshInterval, idle: Duration) {
2921 self.host.refresh_interval = active;
2922 self.schedule.set_intervals(active, idle);
2923 }
2924
2925 #[must_use]
2927 pub const fn repositories(&self) -> &RepositoryCache {
2928 &self.repositories
2929 }
2930
2931 #[cfg(test)]
2935 pub async fn start_directed(&self, policies: &[ScalePolicy]) -> ReconcileReport {
2936 let mut report = ReconcileReport::default();
2937 let mut round = ReservationRound::default();
2938 let mut reserved = Vec::new();
2939 while let Some(next) = self
2940 .reserve_next_launch(policies, &mut round, &mut report)
2941 .await
2942 {
2943 reserved.push(next);
2944 }
2945 self.finish_reserved(reserved, &mut report).await;
2946 report
2947 }
2948
2949 pub async fn serve_launches(
2965 &self,
2966 policies: &(dyn Fn() -> Vec<ScalePolicy> + Send + Sync),
2967 signal: &tokio::sync::Notify,
2968 mut stop: tokio::sync::watch::Receiver<bool>,
2969 ) {
2970 use futures::StreamExt as _;
2971
2972 let mut bringing_up = futures::stream::FuturesUnordered::new();
2973 let mut reserving: Option<Reserving<'_>> = None;
2974 let mut retry_at: Option<tokio::time::Instant> = None;
2975 let mut known: Vec<ScalePolicy> = Vec::new();
2976 let mut stopping = *stop.borrow();
2977 loop {
2978 if stopping && reserving.is_none() && bringing_up.is_empty() {
2979 return;
2980 }
2981 tokio::select! {
2982 () = signal.notified(), if !stopping && reserving.is_none() => {
2983 reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
2984 }
2985 () = async {
2986 match retry_at {
2987 Some(at) => tokio::time::sleep_until(at).await,
2988 None => std::future::pending().await,
2989 }
2990 }, if !stopping && reserving.is_none() => {
2991 retry_at = None;
2992 reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
2993 }
2994 (reserved, round, read) = async {
2995 match reserving.as_mut() {
2996 Some(reserving) => reserving.await,
2997 None => std::future::pending().await,
2998 }
2999 } => {
3000 reserving = None;
3001 match reserved {
3002 Some(reserved) => {
3003 bringing_up.push(self.bring_up(reserved));
3004 known = read;
3005 if !stopping {
3006 reserving = Some(self.reserve_step(policies(), round));
3007 }
3008 }
3009 None => known = read,
3010 }
3011 }
3012 Some(_) = bringing_up.next(), if !bringing_up.is_empty() => {}
3013 changed = stop.changed(), if !stopping => {
3014 stopping = changed.is_err() || *stop.borrow();
3015 }
3016 }
3017 if reserving.is_none()
3020 && retry_at.is_none()
3021 && known
3022 .iter()
3023 .any(|policy| self.director.next_waiting(policy).is_some())
3024 {
3025 retry_at = Some(tokio::time::Instant::now() + LAUNCH_RETRY);
3026 }
3027 }
3028 }
3029
3030 fn reserve_step(
3032 &self,
3033 policies: Vec<ScalePolicy>,
3034 mut round: ReservationRound,
3035 ) -> Reserving<'_> {
3036 Box::pin(async move {
3037 let mut report = ReconcileReport::default();
3038 let reserved = self
3039 .reserve_next_launch(&policies, &mut round, &mut report)
3040 .await;
3041 (reserved, round, policies)
3042 })
3043 }
3044
3045 async fn reserve_next_launch(
3050 &self,
3051 policies: &[ScalePolicy],
3052 round: &mut ReservationRound,
3053 report: &mut ReconcileReport,
3054 ) -> Option<Reserved> {
3055 loop {
3056 let policy = policies
3057 .iter()
3058 .filter(|policy| {
3059 policy.owns_runners()
3060 && policy.is_owned_by(self.host.id)
3061 && policy.may_start_runners()
3062 && !round.exhausted.contains(&policy.id)
3063 })
3064 .min_by_key(|policy| self.director.next_waiting(policy).unwrap_or(u64::MAX))?;
3065 let (served, most) = round.reserved.entry(policy.id).or_default();
3066 if let Some(directed) = self.director.directed_demand(policy)
3068 && *served < directed.max(*most)
3069 {
3070 *most = directed.max(*most);
3071 let mut reserved = Vec::new();
3072 let excluded = self.director.directed_exclusions(policy);
3073 self.start_runners(policy, directed, 0, &excluded, 1, report, &mut reserved)
3074 .await;
3075 if let Some(next) = reserved.pop() {
3076 *served += 1;
3077 return Some(next);
3078 }
3079 }
3080 round.exhausted.insert(policy.id);
3081 }
3082 }
3083
3084 pub async fn reconcile(&mut self, policies: &[ScalePolicy]) -> ReconcileReport {
3098 let mut report = ReconcileReport::default();
3099 let mut reserved: Vec<Reserved> = Vec::new();
3103
3104 let mut pollable: Vec<&ScalePolicy> = Vec::new();
3106 let mut supervision_failed = BTreeSet::new();
3107 for policy in policies {
3108 if !policy.owns_runners() {
3109 report.monitor_only.push(policy.id);
3110 self.events
3111 .emit(LifecycleEvent::MonitorOnlySkipped { policy: policy.id });
3112 continue;
3113 }
3114 if !policy.is_owned_by(self.host.id) {
3115 continue;
3119 }
3120 match self.launcher.supervise(policy).await {
3121 Ok(intents) => {
3122 report.replacement_intents = report
3123 .replacement_intents
3124 .saturating_add(u16::try_from(intents.len()).unwrap_or(u16::MAX));
3125 }
3126 Err(failure) => {
3127 supervision_failed.insert(policy.id);
3128 self.report_unreadable_attempts(&mut report, &failure);
3129 continue;
3130 }
3131 }
3132 if !policy.may_start_runners() {
3133 continue;
3134 }
3135 pollable.push(policy);
3136 }
3137
3138 let readings = self.poll_targets(&pollable, &mut report).await;
3139 report.targets_read = u16::try_from(
3140 readings
3141 .values()
3142 .filter(|outcome| matches!(outcome, PollOutcome::Ready(_)))
3143 .count(),
3144 )
3145 .unwrap_or(u16::MAX);
3146 let mut routed = BTreeMap::new();
3147 for (target, outcome) in &readings {
3148 let PollOutcome::Ready(reading) = outcome else {
3149 continue;
3150 };
3151 let profiles: Vec<&ScalePolicy> = pollable
3152 .iter()
3153 .copied()
3154 .filter(|policy| &policy.target == target)
3155 .collect();
3156 let (tallies, refusals) = route_demand(policies, &profiles, reading, &self.host_labels);
3157 if let Some(first) = profiles.first() {
3158 for (reason, count) in refusals {
3159 self.events.emit(LifecycleEvent::RoutingRefused {
3160 policy: first.id,
3161 reason,
3162 count,
3163 });
3164 }
3165 }
3166 routed.insert(target.clone(), tallies);
3167 }
3168
3169 self.clean_terminal_attempts(&mut report).await;
3179
3180 for policy in policies {
3195 if !policy.owns_runners() {
3196 continue;
3202 }
3203 if !policy.is_owned_by(self.host.id) || !policy.may_start_runners() {
3204 match self.allocate_only(policy, 0, &reserved).await {
3208 Ok(allocation) => {
3209 self.emit_allocation(&allocation);
3210 report.allocations.push(allocation);
3211 }
3212 Err(failure) => self.report_allocation_failure(&mut report, &failure),
3213 }
3214 continue;
3215 }
3216 if supervision_failed.contains(&policy.id) {
3217 continue;
3218 }
3219 if let Some(directed) = self.director.directed_demand(policy) {
3229 let own = match readings.get(&policy.target) {
3230 Some(PollOutcome::Ready(reading))
3231 if !self.director.places_plain_jobs(policy) =>
3232 {
3233 routed
3234 .get(&policy.target)
3235 .and_then(|tallies| tallies.get(&policy.id))
3236 .filter(|tally| tally.non_pool_demand() > 0)
3237 .map(|tally| (tally, reading))
3238 }
3239 _ => None,
3240 };
3241 if own.is_some() || self.launch_side.is_none() {
3242 let mut excluded = self.director.directed_exclusions(policy);
3250 let own_demand = own.map_or(0, |(tally, reading)| {
3251 let count = tally.non_pool_demand();
3252 self.observe_demand(policy, tally, count, reading);
3253 excluded.extend(tally.excluded.iter().cloned());
3254 count
3255 });
3256 self.start_runners(
3257 policy,
3258 directed,
3259 own_demand,
3260 &excluded,
3261 u16::MAX,
3262 &mut report,
3263 &mut reserved,
3264 )
3265 .await;
3266 continue;
3267 }
3268 let Some(launch_side) = &self.launch_side else {
3269 continue;
3270 };
3271 match self.allocate_only(policy, directed, &reserved).await {
3272 Ok(allocation) => {
3273 if !allocation.starts_nothing() {
3274 launch_side.notify_one();
3275 }
3276 report.allocations.push(allocation);
3277 }
3278 Err(failure) => self.report_allocation_failure(&mut report, &failure),
3279 }
3280 continue;
3281 }
3282 let Some(reading) = readings.get(&policy.target) else {
3283 debug_assert!(false, "a pollable policy's target has no reading");
3286 continue;
3287 };
3288 match reading {
3289 PollOutcome::Failed(state) => {
3290 report.unreadable.push(policy.id);
3291 self.events.emit(LifecycleEvent::TargetUnreadable {
3292 policy: policy.id,
3293 reason: unreadable_reason(state),
3294 });
3295 }
3296 PollOutcome::Ready(demand) => {
3297 let tally = routed
3298 .get(&policy.target)
3299 .and_then(|tallies| tallies.get(&policy.id))
3300 .expect("every pollable profile has a routed tally");
3301 let count = tally.demand();
3302 self.observe_demand(policy, tally, count, demand);
3303 self.start_runners(
3304 policy,
3305 count,
3306 0,
3307 &tally.excluded,
3308 u16::MAX,
3309 &mut report,
3310 &mut reserved,
3311 )
3312 .await;
3313 }
3314 }
3315 }
3316 self.finish_reserved(reserved, &mut report).await;
3317
3318 let failure = readings
3320 .values()
3321 .filter_map(PollOutcome::failure)
3322 .max_by_key(|state| severity(state))
3323 .cloned();
3324 let now = self.clock.now();
3325 let conditions = PollConditions {
3326 quiet: is_quiet(&readings, &report),
3327 requests_per_pass: report.demand_requests,
3328 quota: readings
3329 .values()
3330 .filter_map(PollOutcome::reading)
3331 .filter_map(QueuedDemand::rate_limit)
3332 .max_by_key(|quota| quota.observed_at)
3333 .copied(),
3334 };
3335 report.next_poll =
3336 self.schedule
3337 .next_poll_with(failure.as_ref(), conditions, now, self.jitter.as_ref());
3338 report.failure = failure;
3339 if let PollPace::Offline { consecutive } = report.next_poll.pace {
3343 let state = OfflineState::new(consecutive, report.next_poll.delay);
3344 report.offline = Some(match self.schedule.offline_for(now) {
3345 Some(elapsed) => state.since(elapsed),
3346 None => state,
3347 });
3348 }
3349 self.events.emit(LifecycleEvent::PollScheduled {
3350 retry_in_ms: u64::try_from(report.next_poll.delay.as_millis()).unwrap_or(u64::MAX),
3351 pace: report.next_poll.pace,
3352 });
3353
3354 report
3355 }
3356
3357 async fn poll_targets(
3365 &self,
3366 pollable: &[&ScalePolicy],
3367 report: &mut ReconcileReport,
3368 ) -> BTreeMap<ScaleTarget, PollOutcome> {
3369 let targets: BTreeSet<ScaleTarget> = pollable.iter().map(|p| p.target.clone()).collect();
3370
3371 let mut readings = BTreeMap::new();
3372 for target in targets {
3373 let outcome = match self.repositories.scope_for(&target).await {
3374 Ok(scope) => {
3375 report.demand_requests = report
3376 .demand_requests
3377 .saturating_add(demand_requests_per_poll(&scope));
3378 self.demand.poll(&scope).await
3379 }
3380 Err(error) => PollOutcome::Failed(RefreshState::from_error(&error)),
3386 };
3387 readings.insert(target, outcome);
3388 }
3389 readings
3390 }
3391
3392 async fn allocate_only(
3410 &self,
3411 policy: &ScalePolicy,
3412 demand: u32,
3413 reserved: &[Reserved],
3414 ) -> Result<Allocation, AllocationFailure> {
3415 self.allocate_over(policy, demand, reserved)
3416 .await
3417 .map(|(allocation, _)| allocation)
3418 }
3419
3420 async fn allocate_over(
3422 &self,
3423 policy: &ScalePolicy,
3424 demand: u32,
3425 reserved: &[Reserved],
3426 ) -> Result<(Allocation, Vec<RunnerAttempt>), AllocationFailure> {
3427 let mut host = self.host.clone();
3430 host.host_capacity = self
3431 .launcher
3432 .host_capacity(&self.host)
3433 .await
3434 .map_err(AllocationFailure::HostCapacity)?;
3435 let attempts = self
3436 .host_attempts(reserved)
3437 .await
3438 .map_err(AllocationFailure::Attempts)?;
3439 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
3440 let allocation = allocator.allocate(policy, demand);
3441 Ok((allocation, attempts))
3442 }
3443
3444 #[allow(clippy::too_many_arguments)]
3478 async fn start_runners(
3479 &self,
3480 policy: &ScalePolicy,
3481 demand: u32,
3482 own: u32,
3483 excluded: &BTreeSet<Label>,
3484 limit: u16,
3485 report: &mut ReconcileReport,
3486 reserved: &mut Vec<Reserved>,
3487 ) {
3488 let excluded = {
3492 let mut seen = self
3493 .seen_exclusions
3494 .lock()
3495 .unwrap_or_else(std::sync::PoisonError::into_inner);
3496 let kept = seen.entry(policy.id).or_default();
3497 kept.extend(excluded.iter().cloned());
3498 kept.clone()
3499 };
3500 let registered = policy.with_runtime_labels(|labels| {
3501 labels
3502 .with_labels(self.host_labels.iter().cloned())
3503 .excluding(&excluded)
3504 });
3505 let policy = ®istered;
3506 let intent = match self
3518 .allocate_only(policy, demand.saturating_add(own), reserved)
3519 .await
3520 {
3521 Ok(intent) => intent,
3522 Err(failure) => {
3523 self.report_allocation_failure(report, &failure);
3524 self.director.start_failed(policy, None);
3525 return;
3526 }
3527 };
3528
3529 let mut decided: Option<Allocation> = None;
3533 let mut budget = intent.to_start.min(limit);
3534
3535 while budget > 0 {
3536 let order_wait = self.let_older_launches_reserve(policy).await;
3537 let demand = if order_wait.is_zero() {
3539 demand
3540 } else {
3541 self.director.directed_demand(policy).unwrap_or(demand)
3542 }
3543 .saturating_add(own);
3544 let waiting_since = std::time::Instant::now();
3545 let guard = match self.lock.acquire().await {
3546 Ok(guard) => guard,
3547 Err(_) => {
3548 report.deferred = report.deferred.saturating_add(budget);
3551 self.events.emit(LifecycleEvent::AllocationDeferred {
3552 policy: policy.id,
3553 count: budget,
3554 });
3555 self.director.start_failed(policy, None);
3558 break;
3559 }
3560 };
3561 let lock_wait = waiting_since.elapsed();
3562
3563 let (allocation, attempts) = match self.allocate_over(policy, demand, reserved).await {
3569 Ok(decided) => decided,
3570 Err(failure) => {
3571 drop(guard);
3572 self.report_allocation_failure(report, &failure);
3573 self.director.start_failed(policy, None);
3574 break;
3575 }
3576 };
3577
3578 if decided.is_none() {
3579 budget = budget.min(allocation.to_start);
3583 decided = Some(allocation.clone());
3584 }
3585
3586 if allocation.starts_nothing() || budget == 0 {
3589 drop(guard);
3590 break;
3591 }
3592
3593 match self.memory.admit(policy, &attempts) {
3597 MemoryVerdict::Defer(figures) => {
3598 drop(guard);
3599 report.memory_deferred = report.memory_deferred.saturating_add(budget);
3600 self.events.emit(LifecycleEvent::MemoryShort {
3601 policy: policy.id,
3602 figures,
3603 started: false,
3604 });
3605 self.director.start_failed(policy, None);
3609 break;
3610 }
3611 MemoryVerdict::AdmitAlone(figures) => {
3612 self.events.emit(LifecycleEvent::MemoryShort {
3613 policy: policy.id,
3614 figures,
3615 started: true,
3616 });
3617 self.memory.admitted(policy);
3618 }
3619 MemoryVerdict::Admit(_) | MemoryVerdict::Unchecked => self.memory.admitted(policy),
3620 }
3621
3622 let created = self
3623 .launcher
3624 .launch(LaunchRequest {
3625 host: &self.host,
3626 policy,
3627 allocation_guard: &guard,
3628 })
3629 .await;
3630 let hold = guard.launch_hold();
3631 drop(guard);
3632
3633 match created {
3634 Ok(attempt) => {
3635 let id = attempt.id;
3636 self.director.launched(policy, id);
3639 budget -= 1;
3640 reserved.push(Reserved {
3644 policy: policy.clone(),
3645 attempt,
3646 waits: ReservationWaits {
3647 order: order_wait,
3648 lock: lock_wait,
3649 },
3650 reserved_at: std::time::Instant::now(),
3651 _hold: hold,
3652 });
3653 }
3654 Err(failure) => {
3655 self.note_start_failed(policy, None, &failure);
3656 break;
3657 }
3658 }
3659 }
3660
3661 let allocation = decided.unwrap_or(intent);
3662 self.emit_allocation(&allocation);
3663 report.allocations.push(allocation);
3664 }
3665
3666 fn observe_demand(
3669 &self,
3670 policy: &ScalePolicy,
3671 tally: &DemandTally,
3672 demand: u32,
3673 reading: &QueuedDemand,
3674 ) {
3675 self.events.emit(LifecycleEvent::DemandObserved {
3676 policy: policy.id,
3677 demand,
3678 not_matched: tally.not_matched,
3679 unresolvable: u32::try_from(tally.unresolvable.len()).unwrap_or(u32::MAX),
3680 complete: reading.is_complete(),
3681 });
3682 }
3683
3684 async fn let_older_launches_reserve(&self, policy: &ScalePolicy) -> Duration {
3690 let began = tokio::time::Instant::now();
3691 let mut older = self.director.older_waiting(policy);
3692 let mut limit = began + OLDER_LAUNCH_TURN;
3693 while older > 0 {
3694 if tokio::time::Instant::now() >= limit {
3695 tracing::warn!(
3696 event = "older_launch_passed_over",
3697 policy_id = %policy.id,
3698 count = older,
3699 );
3700 self.director.pass_over_older(policy);
3701 break;
3702 }
3703 tokio::time::sleep(OLDER_LAUNCH_POLL).await;
3704 let still = self.director.older_waiting(policy);
3705 if still < older {
3706 limit = tokio::time::Instant::now() + OLDER_LAUNCH_TURN;
3707 }
3708 older = still;
3709 }
3710 began.elapsed()
3711 }
3712
3713 async fn finish_reserved(&self, reserved: Vec<Reserved>, report: &mut ReconcileReport) {
3721 let finished =
3722 futures::future::join_all(reserved.into_iter().map(|reserved| self.bring_up(reserved)))
3723 .await;
3724 let started = finished.into_iter().filter(|started| *started).count();
3725 report.started = report
3726 .started
3727 .saturating_add(u16::try_from(started).unwrap_or(u16::MAX));
3728 }
3729
3730 async fn bring_up(&self, reserved: Reserved) -> bool {
3732 let id = reserved.attempt.id;
3733 let mut waits = reserved.waits;
3734 waits.lock += reserved.reserved_at.elapsed();
3735 let result = self
3736 .launcher
3737 .finish_launch(&reserved.policy, reserved.attempt, waits)
3738 .await;
3739 match &result {
3742 Ok(_) => self.events.emit(LifecycleEvent::RunnerStarted {
3743 policy: reserved.policy.id,
3744 attempt: id,
3745 }),
3746 Err(failure) => self.note_start_failed(&reserved.policy, Some(id), failure),
3747 }
3748 result.is_ok()
3749 }
3750
3751 fn note_start_failed(
3752 &self,
3753 policy: &ScalePolicy,
3754 attempt: Option<AttemptId>,
3755 failure: &LaunchFailure,
3756 ) {
3757 self.events.emit(LifecycleEvent::RunnerStartFailed {
3758 policy: policy.id,
3759 reason: failure_reason_kind(&failure.reason),
3760 });
3761 self.director.start_failed(policy, attempt);
3762 }
3763
3764 async fn host_attempts(
3777 &self,
3778 reserved: &[Reserved],
3779 ) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
3780 let mut attempts = self.launcher.attempts().await?;
3781
3782 debug_assert!(
3789 reserved
3790 .iter()
3791 .map(|reserved| reserved.attempt.id)
3792 .collect::<BTreeSet<AttemptId>>()
3793 .len()
3794 == reserved.len(),
3795 "`RunnerLauncher::launch` returned an AttemptId this pass had already seen; \
3796 the host ceiling is enforced against a set keyed on that identifier, so a \
3797 duplicate is two runtimes counted as one"
3798 );
3799
3800 let known: BTreeSet<AttemptId> = attempts.iter().map(|attempt| attempt.id).collect();
3801 attempts.extend(
3802 reserved
3803 .iter()
3804 .filter(|reserved| !known.contains(&reserved.attempt.id))
3805 .map(|reserved| reserved.attempt.clone()),
3806 );
3807 Ok(attempts)
3808 }
3809
3810 fn report_allocation_failure(&self, report: &mut ReconcileReport, failure: &AllocationFailure) {
3811 match failure {
3812 AllocationFailure::HostCapacity(reason) => {
3813 report.host_capacity_unreadable = report.host_capacity_unreadable.saturating_add(1);
3814 self.events
3815 .emit(LifecycleEvent::HostCapacityUnreadable { reason: *reason });
3816 }
3817 AllocationFailure::Attempts(failure) => {
3818 self.report_unreadable_attempts(report, failure)
3819 }
3820 }
3821 }
3822
3823 fn report_unreadable_attempts(&self, report: &mut ReconcileReport, failure: &LaunchFailure) {
3830 report.attempts_unreadable = report.attempts_unreadable.saturating_add(1);
3831 self.events.emit(LifecycleEvent::AttemptsUnreadable {
3838 reason: failure_reason_kind(&failure.reason),
3839 });
3840 }
3841
3842 async fn clean_terminal_attempts(&mut self, report: &mut ReconcileReport) {
3848 let attempts = match self.launcher.attempts().await {
3849 Ok(attempts) => attempts,
3850 Err(failure) => {
3851 self.report_unreadable_attempts(report, &failure);
3852 return;
3853 }
3854 };
3855 self.clean_backoff
3856 .retain(&attempts.iter().map(|attempt| attempt.id).collect());
3857 let now = self.clock.now();
3858 for attempt in attempts {
3859 if !attempt.state().is_concluded() {
3860 continue;
3861 }
3862 let Some(outcome) = attempt.outcome() else {
3863 continue;
3864 };
3865 if !self.clean_backoff.is_due(attempt.id, now) {
3868 continue;
3869 }
3870 let kind = OutcomeKind::of(outcome);
3871 match self.launcher.clean(attempt.id).await {
3872 Ok(()) => {
3873 self.clean_backoff.succeeded(attempt.id);
3874 report.cleaned = report.cleaned.saturating_add(1);
3875 if kind.is_failure() {
3876 report.failures = report.failures.saturating_add(1);
3877 } else if kind == OutcomeKind::IdleExit {
3878 report.idle_exits = report.idle_exits.saturating_add(1);
3883 }
3884 self.events.emit(LifecycleEvent::AttemptCleaned {
3885 policy: attempt.policy_id,
3886 attempt: attempt.id,
3887 outcome: kind,
3888 });
3889 }
3890 Err(failure) => {
3900 self.clean_backoff.failed(attempt.id, now);
3901 report.clean_failures = report.clean_failures.saturating_add(1);
3902 self.events.emit(LifecycleEvent::AttemptCleanFailed {
3903 policy: attempt.policy_id,
3904 attempt: attempt.id,
3905 reason: failure_reason_kind(&failure.reason),
3906 });
3907 }
3908 }
3909 }
3910 }
3911
3912 pub async fn scale_down(&self, policy: &ScalePolicy) -> ScaleDownReport {
3920 let mut report = ScaleDownReport::default();
3921 let attempts = match self.launcher.attempts().await {
3922 Ok(attempts) => attempts,
3923 Err(failure) => {
3924 self.events.emit(LifecycleEvent::AttemptsUnreadable {
3929 reason: failure_reason_kind(&failure.reason),
3930 });
3931 report.attempts_unreadable = true;
3932 return report;
3933 }
3934 };
3935 for attempt in attempts {
3936 if attempt.policy_id != policy.id {
3937 continue;
3938 }
3939 match attempt.state() {
3940 AttemptState::Busy => {
3941 report.refused_busy = report.refused_busy.saturating_add(1);
3942 self.events.emit(LifecycleEvent::ScaleDownRefused {
3943 policy: policy.id,
3944 attempt: attempt.id,
3945 });
3946 }
3947 state if state.is_concluded() => {
3948 let kind = attempt
3949 .outcome()
3950 .map_or(OutcomeKind::Failed, OutcomeKind::of);
3951 match self.launcher.clean(attempt.id).await {
3952 Ok(()) => {
3953 report.removed = report.removed.saturating_add(1);
3954 self.events.emit(LifecycleEvent::AttemptCleaned {
3955 policy: policy.id,
3956 attempt: attempt.id,
3957 outcome: kind,
3958 });
3959 }
3960 Err(failure) => {
3961 report.clean_failures = report.clean_failures.saturating_add(1);
3962 self.events.emit(LifecycleEvent::AttemptCleanFailed {
3963 policy: policy.id,
3964 attempt: attempt.id,
3965 reason: failure_reason_kind(&failure.reason),
3966 });
3967 }
3968 }
3969 }
3970 AttemptState::Cleaned => {}
3971 _ => report.retained = report.retained.saturating_add(1),
3974 }
3975 }
3976 report
3977 }
3978
3979 fn emit_allocation(&self, allocation: &Allocation) {
3980 self.events.emit(LifecycleEvent::Allocated {
3981 policy: allocation.policy_id,
3982 demand: allocation.demand,
3983 desired: allocation.desired,
3984 active_owned: allocation.active_owned,
3985 headroom: allocation.headroom_before,
3986 to_start: allocation.to_start,
3987 limiting: allocation.limiting_factor,
3988 });
3989 }
3990}
3991
3992fn demand_for(policy: &ScalePolicy, reading: &QueuedDemand, host_labels: &[Label]) -> DemandTally {
4016 let Some(labels) = policy.routing_labels() else {
4017 debug_assert!(
4018 false,
4019 "a monitor-only policy is skipped before the demand poll (D19)"
4020 );
4021 return DemandTally::default();
4022 };
4023 let labels = labels.with_labels(host_labels.iter().cloned());
4024
4025 match &policy.target {
4026 ScaleTarget::Repository(repository) => labels.tally(reading.jobs_for(repository)),
4027 ScaleTarget::Organization(_) => labels.tally(reading.jobs()),
4028 }
4029}
4030
4031fn route_demand(
4035 all_policies: &[ScalePolicy],
4036 profiles: &[&ScalePolicy],
4037 reading: &QueuedDemand,
4038 host_labels: &[Label],
4039) -> (
4040 BTreeMap<PolicyId, DemandTally>,
4041 BTreeMap<RoutingRefusal, u32>,
4042) {
4043 let mut tallies: BTreeMap<PolicyId, DemandTally> = profiles
4044 .iter()
4045 .map(|policy| (policy.id, demand_for(policy, reading, host_labels)))
4046 .collect();
4047 let mut refusals: BTreeMap<RoutingRefusal, u32> = BTreeMap::new();
4048 let Some(first) = profiles.first() else {
4049 return (tallies, refusals);
4050 };
4051 let owners: Vec<_> = all_policies
4053 .iter()
4054 .filter(|policy| policy.target == first.target && policy.host_id == first.host_id)
4055 .filter_map(|policy| {
4056 let labels = policy.routing_labels()?;
4057 Some((policy.id, labels.with_labels(host_labels.iter().cloned())))
4058 })
4059 .collect();
4060 let jobs: Vec<&runner_manager_domain::policy::RunsOn> = match &first.target {
4061 ScaleTarget::Repository(repository) => reading.jobs_for(repository).iter().collect(),
4062 ScaleTarget::Organization(_) => reading.jobs().collect(),
4063 };
4064 for job in jobs {
4065 if let Err(unresolvable) = job.required_labels() {
4066 if unresolvable.is_unsatisfiable_exclusion() {
4070 let meant = job.plain_labels();
4071 if !owners
4072 .iter()
4073 .any(|(_, labels)| meant.iter().all(|label| labels.contains(label)))
4074 {
4075 continue;
4076 }
4077 *refusals
4078 .entry(RoutingRefusal::UnsatisfiableExclusion)
4079 .or_default() += 1;
4080 }
4081 continue;
4082 }
4083 let matched: Vec<PolicyId> = owners
4084 .iter()
4085 .filter(|(_, labels)| labels.matches(job).is_match())
4086 .map(|(id, _)| *id)
4087 .collect();
4088 if matched.len() > 1 {
4089 *refusals
4090 .entry(RoutingRefusal::OverlappingProfiles)
4091 .or_default() += 1;
4092 let pool = job
4093 .required_labels()
4094 .is_ok_and(|required| is_pool_job(&required));
4095 for id in matched {
4096 if let Some(tally) = tallies.get_mut(&id) {
4097 tally.matched -= 1;
4098 tally.not_matched += 1;
4099 if pool {
4100 tally.pool_matched = tally.pool_matched.saturating_sub(1);
4101 }
4102 }
4103 }
4104 }
4105 }
4106 (tallies, refusals)
4107}
4108
4109const fn severity(state: &RefreshState) -> u8 {
4116 match state {
4117 RefreshState::Offline => 5,
4118 RefreshState::RateLimited(_) => 4,
4119 RefreshState::LockedOut { .. } => 3,
4120 RefreshState::Unauthorized => 2,
4121 RefreshState::Forbidden { .. } | RefreshState::Failed { .. } => 1,
4122 RefreshState::Cancelled | RefreshState::Ready(_) => 0,
4123 }
4124}
4125
4126const fn unreadable_reason(state: &RefreshState) -> &'static str {
4137 match state {
4138 RefreshState::Ready(_) => "ready",
4139 RefreshState::Offline => "offline",
4140 RefreshState::RateLimited(_) => "rate_limited",
4141 RefreshState::LockedOut { .. } => "locked_out",
4142 RefreshState::Unauthorized => "unauthorized",
4143 RefreshState::Forbidden { .. } => "forbidden",
4144 RefreshState::Failed { .. } => "failed",
4145 RefreshState::Cancelled => "cancelled",
4146 }
4147}
4148
4149#[cfg(test)]
4150mod tests {
4151 use super::*;
4152
4153 #[test]
4161 fn a_pass_that_reached_no_target_does_not_claim_it_reached_github() {
4162 let mut report = ReconcileReport::default();
4163 assert!(
4164 !report.reached_github(),
4165 "a pass that polled nothing -- every policy draining, owned elsewhere, or \
4166 monitor-only -- reached nobody. This is the case the old guard let through, and \
4167 the only one it ever let through."
4168 );
4169
4170 report.unreadable.push(PolicyId::from_u128(1));
4171 assert!(
4172 !report.reached_github(),
4173 "every target this pass tried was unreadable, so there is no contact to record"
4174 );
4175
4176 report.targets_read = 1;
4177 assert!(
4178 report.reached_github(),
4179 "one target answering is contact, whatever else failed alongside it"
4180 );
4181
4182 let mut unowned = ReconcileReport::default();
4186 unowned.unreadable.push(PolicyId::from_u128(2));
4187 unowned.allocations.push(Allocation {
4188 policy_id: PolicyId::from_u128(2),
4189 demand: 0,
4190 desired: 0,
4191 active_owned: 0,
4192 headroom_before: 0,
4193 to_start: 0,
4194 limiting_factor: LimitingFactor::Demand,
4195 });
4196 assert!(
4197 !unowned.reached_github(),
4198 "an allocation is not evidence that GitHub answered"
4199 );
4200 }
4201
4202 #[test]
4212 fn an_unauthorized_target_is_a_failure_and_not_merely_unreadable() {
4213 assert!(
4214 severity(&RefreshState::Unauthorized) > 0,
4215 "an unauthorized reading must survive `max_by_key(severity)` into `report.failure`, \
4216 or a pass where every target was refused would report no failure at all"
4217 );
4218 }
4219
4220 use std::sync::atomic::AtomicUsize;
4221
4222 use std::num::NonZeroU16;
4223
4224 use runner_manager_domain::attempt::PersistedAttempt;
4225 use runner_manager_domain::model::{CachePolicy, HostId};
4226 use runner_manager_domain::policy::PolicyMode;
4227 use runner_manager_domain::workspace::WorkspaceKind;
4228 use runner_manager_github::rest::RateLimited;
4229 use runner_manager_testkit::clock::FakeClock;
4230 use runner_manager_testkit::fixtures;
4231 use runner_manager_testkit::github::FakeGithub;
4232
4233 fn host_with(capacity: u16) -> Host {
4238 fixtures::host().capacity(capacity).build()
4239 }
4240
4241 fn repo(raw: &str) -> OwnerRepo {
4242 OwnerRepo::parse(raw).expect("a valid OWNER/REPO")
4243 }
4244
4245 use runner_manager_domain::policy::{RoutingLabels, RunsOn, pool_label};
4247
4248 fn policy(id: u128, target: &str, max: u16) -> ScalePolicy {
4249 fixtures::policy()
4250 .id(PolicyId::from_u128(id))
4251 .repository(target)
4252 .autoscale("home", max)
4253 .active()
4254 .build()
4255 }
4256
4257 const HOST_LABEL: &str = "rm-home-win-x64";
4264
4265 fn jobs(n: usize) -> Vec<RunsOn> {
4271 fixtures::queued_jobs(&[HOST_LABEL], n)
4272 }
4273
4274 #[derive(Debug, Default)]
4276 struct FakeLauncher {
4277 capacity: Mutex<Option<NonZeroU16>>,
4278 capacity_unreadable: AtomicBool,
4279 capacity_after_launch: Mutex<Option<NonZeroU16>>,
4280 attempts: Mutex<Vec<RunnerAttempt>>,
4281 next_id: AtomicU64,
4282 launches: AtomicUsize,
4283 cleaned: Mutex<Vec<AttemptId>>,
4284 yields_before_recording: usize,
4287 forgetful: bool,
4291 fail_next: Mutex<Option<FailureReason>>,
4292 attempts_fail: Mutex<bool>,
4295 clean_fails: bool,
4298 replacements: Mutex<Vec<ReplacementIntent>>,
4299 registered: Mutex<Vec<Vec<String>>>,
4301 }
4302
4303 impl FakeLauncher {
4304 fn new() -> Self {
4305 Self::default()
4306 }
4307
4308 fn with_yields(mut self, yields: usize) -> Self {
4309 self.yields_before_recording = yields;
4310 self
4311 }
4312
4313 fn forgetful() -> Self {
4314 Self {
4315 forgetful: true,
4316 ..Self::default()
4317 }
4318 }
4319
4320 fn seeded(self, attempts: Vec<RunnerAttempt>) -> Self {
4321 *self.attempts.lock().unwrap() = attempts;
4322 self
4323 }
4324
4325 fn launches(&self) -> usize {
4326 self.launches.load(Ordering::SeqCst)
4327 }
4328
4329 fn snapshot(&self) -> Vec<RunnerAttempt> {
4330 self.attempts.lock().unwrap().clone()
4331 }
4332
4333 fn live_count(&self) -> usize {
4334 self.snapshot()
4335 .iter()
4336 .filter(|a| a.counts_against_capacity())
4337 .count()
4338 }
4339
4340 fn fail_next(&self, reason: FailureReason) {
4341 *self.fail_next.lock().unwrap() = Some(reason);
4342 }
4343
4344 fn fail_attempts(&self, failing: bool) {
4345 *self.attempts_fail.lock().unwrap() = failing;
4346 }
4347
4348 fn refusing_cleanup(attempts: Vec<RunnerAttempt>) -> Self {
4349 Self {
4350 clean_fails: true,
4351 ..Self::default()
4352 }
4353 .seeded(attempts)
4354 }
4355
4356 fn cleaned(&self) -> Vec<AttemptId> {
4357 self.cleaned.lock().unwrap().clone()
4358 }
4359
4360 fn replacing(self, intent: ReplacementIntent) -> Self {
4361 self.replacements.lock().unwrap().push(intent);
4362 self
4363 }
4364 }
4365
4366 #[async_trait::async_trait]
4367 impl RunnerLauncher for FakeLauncher {
4368 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
4369 if self.capacity_unreadable.load(Ordering::SeqCst) {
4370 return Err(HostCapacityError::JournalUnreadable);
4371 }
4372 Ok(self.capacity.lock().unwrap().unwrap_or(host.host_capacity))
4373 }
4374
4375 async fn supervise(
4376 &self,
4377 policy: &ScalePolicy,
4378 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
4379 let mut replacements = self.replacements.lock().unwrap();
4380 let selected: Vec<_> = replacements
4381 .extract_if(.., |intent| intent.policy == policy.id)
4382 .collect();
4383 if !selected.is_empty() {
4384 let retired: BTreeSet<_> = selected
4385 .iter()
4386 .map(|intent| intent.previous_attempt)
4387 .collect();
4388 self.attempts
4389 .lock()
4390 .unwrap()
4391 .retain(|attempt| !retired.contains(&attempt.id));
4392 }
4393 Ok(selected)
4394 }
4395
4396 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
4397 if *self.attempts_fail.lock().unwrap() {
4398 return Err(LaunchFailure::new(FailureReason::Other(
4399 "the journal could not be read".into(),
4400 )));
4401 }
4402 Ok(self.snapshot())
4403 }
4404
4405 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
4406 if let Some(reason) = self.fail_next.lock().unwrap().take() {
4407 return Err(LaunchFailure::new(reason));
4408 }
4409 for _ in 0..self.yields_before_recording {
4411 tokio::task::yield_now().await;
4412 }
4413 let id =
4414 AttemptId::from_u128(u128::from(self.next_id.fetch_add(1, Ordering::SeqCst) + 1));
4415 let created = RunnerAttempt::allocate(
4416 id,
4417 request.policy.id,
4418 "runtime/p/a",
4419 request.host.created_at,
4420 );
4421 self.registered.lock().unwrap().push(
4422 request
4423 .policy
4424 .routing_labels()
4425 .map(RoutingLabels::as_registration_labels)
4426 .unwrap_or_default(),
4427 );
4428 self.launches.fetch_add(1, Ordering::SeqCst);
4429 if let Some(capacity) = self.capacity_after_launch.lock().unwrap().take() {
4430 *self.capacity.lock().unwrap() = Some(capacity);
4431 }
4432 if !self.forgetful {
4433 self.attempts.lock().unwrap().push(created.clone());
4434 }
4435 Ok(created)
4436 }
4437
4438 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
4439 if self.clean_fails {
4440 return Err(LaunchFailure::new(FailureReason::Other(
4441 "the runtime directory is locked".into(),
4442 )));
4443 }
4444 self.cleaned.lock().unwrap().push(attempt);
4445 let mut attempts = self.attempts.lock().unwrap();
4446 attempts.retain(|a| a.id != attempt);
4447 Ok(())
4448 }
4449 }
4450
4451 #[derive(Debug, Default)]
4453 struct FakeDemand {
4454 outcome: Mutex<Option<PollOutcome>>,
4455 per_target: Mutex<BTreeMap<ScaleTarget, PollOutcome>>,
4457 scopes: Mutex<Vec<ActivityScope>>,
4458 }
4459
4460 impl FakeDemand {
4461 fn ready(count: u32, repository: &OwnerRepo) -> Self {
4462 let fake = Self::default();
4463 fake.set(PollOutcome::Ready(QueuedDemand::of(
4464 repository.clone(),
4465 jobs(count as usize),
4466 )));
4467 fake
4468 }
4469
4470 fn failing(state: RefreshState) -> Self {
4471 let fake = Self::default();
4472 fake.set(PollOutcome::Failed(state));
4473 fake
4474 }
4475
4476 fn set(&self, outcome: PollOutcome) {
4477 *self.outcome.lock().unwrap() = Some(outcome);
4478 }
4479
4480 fn set_for(&self, target: &ScaleTarget, outcome: PollOutcome) {
4482 self.per_target
4483 .lock()
4484 .unwrap()
4485 .insert(target.clone(), outcome);
4486 }
4487
4488 fn polls(&self) -> Vec<ActivityScope> {
4489 self.scopes.lock().unwrap().clone()
4490 }
4491 }
4492
4493 #[async_trait::async_trait]
4494 impl DemandSource for FakeDemand {
4495 async fn poll(&self, scope: &ActivityScope) -> PollOutcome {
4496 self.scopes.lock().unwrap().push(scope.clone());
4497 if let Some(outcome) = self.per_target.lock().unwrap().get(scope.target()) {
4498 return outcome.clone();
4499 }
4500 self.outcome
4501 .lock()
4502 .unwrap()
4503 .clone()
4504 .unwrap_or(PollOutcome::Ready(QueuedDemand::default()))
4505 }
4506 }
4507
4508 #[derive(Debug, Default)]
4509 struct FakeDirectory {
4510 repositories: Vec<OwnerRepo>,
4511 calls: AtomicUsize,
4512 }
4513
4514 impl FakeDirectory {
4515 fn of(repositories: Vec<OwnerRepo>) -> Self {
4516 Self {
4517 repositories,
4518 calls: AtomicUsize::new(0),
4519 }
4520 }
4521
4522 fn calls(&self) -> usize {
4523 self.calls.load(Ordering::SeqCst)
4524 }
4525 }
4526
4527 #[async_trait::async_trait]
4528 impl RepositoryDirectory for FakeDirectory {
4529 async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
4530 self.calls.fetch_add(1, Ordering::SeqCst);
4531 Ok(self.repositories.clone())
4532 }
4533 }
4534
4535 #[derive(Debug)]
4541 struct CountingLock {
4542 inner: InProcessAllocationLock,
4543 concurrent: Arc<AtomicUsize>,
4544 peak: Arc<AtomicUsize>,
4545 acquisitions: Arc<AtomicUsize>,
4546 }
4547
4548 impl CountingLock {
4549 fn new() -> Self {
4550 Self {
4551 inner: InProcessAllocationLock::new(),
4552 concurrent: Arc::new(AtomicUsize::new(0)),
4553 peak: Arc::new(AtomicUsize::new(0)),
4554 acquisitions: Arc::new(AtomicUsize::new(0)),
4555 }
4556 }
4557
4558 fn peak(&self) -> usize {
4559 self.peak.load(Ordering::SeqCst)
4560 }
4561
4562 fn acquisitions(&self) -> usize {
4563 self.acquisitions.load(Ordering::SeqCst)
4564 }
4565 }
4566
4567 #[derive(Debug)]
4568 struct CountingGuard {
4569 _inner: AllocationGuard,
4570 concurrent: Arc<AtomicUsize>,
4571 }
4572
4573 impl Drop for CountingGuard {
4574 fn drop(&mut self) {
4575 self.concurrent.fetch_sub(1, Ordering::SeqCst);
4576 }
4577 }
4578
4579 #[async_trait::async_trait]
4580 impl AllocationLock for CountingLock {
4581 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4582 let inner = self.inner.acquire().await?;
4583 self.acquisitions.fetch_add(1, Ordering::SeqCst);
4584 let now = self.concurrent.fetch_add(1, Ordering::SeqCst) + 1;
4585 self.peak.fetch_max(now, Ordering::SeqCst);
4586 Ok(AllocationGuard::new(CountingGuard {
4587 _inner: inner,
4588 concurrent: Arc::clone(&self.concurrent),
4589 }))
4590 }
4591 }
4592
4593 #[derive(Debug, Default)]
4596 struct NoLock;
4597
4598 #[async_trait::async_trait]
4599 impl AllocationLock for NoLock {
4600 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4601 Ok(AllocationGuard::new(()))
4602 }
4603 }
4604
4605 #[tokio::test]
4606 async fn a_wsl_drain_request_fences_the_exact_pre_launch_boundary() {
4607 use runner_manager_platform::wsl::fence::DrainRequest;
4608
4609 let (_local, shared, paths) = guest_paths();
4610 DrainRequest::new(4, chrono::Utc::now())
4611 .write(shared.path())
4612 .unwrap();
4613 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4614
4615 assert!(lock.acquire().await.is_err());
4616 }
4617
4618 #[tokio::test]
4619 async fn a_guest_launch_claim_is_released_with_its_allocation_guard() {
4620 use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;
4621
4622 let (_local, shared, paths) = guest_paths();
4623 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4624 let guard = lock.acquire().await.unwrap();
4625 assert!(shared.path().join(FENCE_DIRECTORY).exists());
4626 drop(guard);
4627 assert!(!shared.path().join(FENCE_DIRECTORY).exists());
4628 }
4629
4630 #[derive(Clone, Default)]
4632 struct Logs(Arc<Mutex<Vec<u8>>>);
4633
4634 impl std::io::Write for Logs {
4635 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
4636 self.0.lock().unwrap().extend_from_slice(buf);
4637 Ok(buf.len())
4638 }
4639 fn flush(&mut self) -> std::io::Result<()> {
4640 Ok(())
4641 }
4642 }
4643
4644 #[tokio::test]
4647 async fn a_launch_fence_that_cannot_be_released_is_logged_with_its_path() {
4648 use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;
4649
4650 let (_local, shared, paths) = guest_paths();
4651 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4652 let guard = lock.acquire().await.unwrap();
4653 let fence = shared.path().join(FENCE_DIRECTORY);
4656 std::fs::remove_dir_all(&fence).unwrap();
4657 std::fs::write(&fence, b"not a directory").unwrap();
4658
4659 let logs = Logs::default();
4660 let writer = logs.clone();
4661 let subscriber = tracing_subscriber::fmt()
4662 .with_writer(move || writer.clone())
4663 .with_ansi(false)
4664 .finish();
4665 tracing::subscriber::with_default(subscriber, || drop(guard));
4666
4667 let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
4668 assert!(text.contains("WARN"), "{text}");
4669 assert!(text.contains("could not be released"), "{text}");
4670 assert!(text.contains(FENCE_DIRECTORY), "the path is named: {text}");
4671 }
4672
4673 fn guest_paths() -> (
4674 tempfile::TempDir,
4675 tempfile::TempDir,
4676 Arc<runner_manager_platform::paths::AppPaths>,
4677 ) {
4678 use runner_manager_platform::paths::AppPaths;
4679 use runner_manager_platform::wsl::fence::GuestRecoveryConfig;
4680
4681 let local = tempfile::tempdir().unwrap();
4682 let shared = tempfile::tempdir().unwrap();
4683 let paths = Arc::new(AppPaths::rooted_at(local.path()));
4684 paths.create_all().unwrap();
4685 GuestRecoveryConfig::new(shared.path().to_path_buf())
4686 .write(&paths)
4687 .unwrap();
4688 (local, shared, paths)
4689 }
4690
4691 #[tokio::test]
4694 async fn a_stale_guest_claim_no_longer_blocks_the_next_launch() {
4695 use runner_manager_platform::wsl::fence::{FENCE_DIRECTORY, OWNER_FILE};
4696
4697 let (_local, shared, paths) = guest_paths();
4698 let fence = shared.path().join(FENCE_DIRECTORY);
4699 std::fs::create_dir_all(&fence).unwrap();
4700 let pid = std::process::id();
4701 std::fs::write(
4702 fence.join(OWNER_FILE),
4703 format!(
4704 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"}}}}"#
4705 ),
4706 )
4707 .unwrap();
4708 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4709
4710 let guard = lock.acquire().await.expect("the stale claim was reclaimed");
4711 drop(guard);
4712 assert!(!fence.exists());
4713 }
4714
4715 #[tokio::test]
4716 async fn refusals_that_persist_are_recorded_and_a_grant_clears_them() {
4717 use runner_manager_platform::launch_health::recorded_launches_blocked;
4718 use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};
4719
4720 let (_local, shared, paths) = guest_paths();
4721 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4722 .with_blocked_after(3, Duration::ZERO);
4723 let windows =
4725 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4726 .unwrap()
4727 .unwrap();
4728
4729 for _ in 0..2 {
4730 assert!(lock.acquire().await.is_err());
4731 }
4732 assert_eq!(
4733 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4734 None,
4735 "not yet"
4736 );
4737 assert!(lock.acquire().await.is_err());
4738 let blocked = recorded_launches_blocked(&paths, chrono::Utc::now())
4739 .unwrap()
4740 .expect("three refusals in a row are recorded");
4741 assert!(
4742 blocked.reason.contains("Windows WSL recovery"),
4743 "{blocked:?}"
4744 );
4745
4746 windows.release().unwrap();
4747 let guard = lock.acquire().await.unwrap();
4748 assert_eq!(
4749 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4750 None
4751 );
4752 drop(guard);
4753 }
4754
4755 #[tokio::test]
4756 async fn refusals_while_this_process_is_launching_are_its_own_contention() {
4757 use runner_manager_platform::launch_health::recorded_launches_blocked;
4758
4759 use runner_manager_platform::wsl::fence::DrainRequest;
4760
4761 let (_local, shared, paths) = guest_paths();
4762 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4763 .with_blocked_after(1, Duration::ZERO);
4764 let launching = lock.acquire().await.unwrap();
4765 DrainRequest::new(1, chrono::Utc::now())
4766 .write(shared.path())
4767 .unwrap();
4768 for _ in 0..3 {
4769 assert!(lock.acquire().await.is_err());
4770 }
4771 assert_eq!(
4772 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4773 None
4774 );
4775 drop(launching);
4776 }
4777
4778 #[tokio::test]
4782 async fn launches_in_progress_share_the_fence_until_the_last_ends() {
4783 use runner_manager_platform::wsl::fence::{DrainRequest, FenceClaim, FenceOwnerKind};
4784
4785 let (_local, shared, paths) = guest_paths();
4786 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4787 let claimed = |root: &std::path::Path| FenceClaim::owner(root).unwrap().is_some();
4788
4789 let first = lock.acquire().await.unwrap();
4790 let first_runner = first.launch_hold();
4791 drop(first);
4792 assert!(claimed(shared.path()), "held while its runner comes up");
4793
4794 let claimed_at = FenceClaim::owner(shared.path())
4795 .unwrap()
4796 .unwrap()
4797 .acquired_at;
4798 tokio::time::sleep(Duration::from_millis(5)).await;
4799 let second = lock.acquire().await.expect("this process owns the fence");
4800 assert!(
4801 FenceClaim::owner(shared.path())
4802 .unwrap()
4803 .unwrap()
4804 .acquired_at
4805 > claimed_at,
4806 "a claim launches keep sharing is not one stuck for that long"
4807 );
4808 let second_runner = second.launch_hold();
4809 drop(second);
4810 drop(first_runner);
4811 assert!(claimed(shared.path()), "the other launch is still running");
4812
4813 DrainRequest::new(1, chrono::Utc::now())
4815 .write(shared.path())
4816 .unwrap();
4817 assert!(lock.acquire().await.is_err());
4818 assert!(claimed(shared.path()));
4819 assert!(
4820 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4821 .unwrap()
4822 .is_none()
4823 );
4824 drop(second_runner);
4825 assert!(!claimed(shared.path()), "the last one lets it go");
4826 assert!(
4827 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4828 .unwrap()
4829 .is_some()
4830 );
4831 }
4832
4833 #[tokio::test]
4836 async fn refusals_with_quiet_spells_between_them_never_add_up() {
4837 use runner_manager_platform::launch_health::recorded_launches_blocked;
4838 use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};
4839
4840 let (_local, shared, paths) = guest_paths();
4841 let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
4842 .with_blocked_after(3, Duration::ZERO)
4843 .with_streak_gap(Duration::ZERO);
4844 let windows =
4845 FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
4846 .unwrap()
4847 .unwrap();
4848 for _ in 0..5 {
4849 assert!(lock.acquire().await.is_err());
4850 }
4851 assert_eq!(
4852 recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
4853 None
4854 );
4855 windows.release().unwrap();
4856 }
4857
4858 #[tokio::test]
4861 async fn a_claim_refused_by_a_drain_request_is_released() {
4862 use runner_manager_platform::wsl::fence::{DrainRequest, FENCE_DIRECTORY};
4863
4864 let (_local, shared, paths) = guest_paths();
4865 DrainRequest::new(4, chrono::Utc::now())
4866 .write(shared.path())
4867 .unwrap();
4868 let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
4869 assert!(lock.acquire().await.is_err());
4870 assert!(!shared.path().join(FENCE_DIRECTORY).exists());
4871 }
4872
4873 #[derive(Debug, Default)]
4875 struct HeldLock;
4876
4877 #[async_trait::async_trait]
4878 impl AllocationLock for HeldLock {
4879 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
4880 Err(AllocationLockBusy)
4881 }
4882 }
4883
4884 struct Harness {
4886 launcher: Arc<FakeLauncher>,
4887 demand: Arc<FakeDemand>,
4888 events: Arc<EventLog>,
4889 clock: Arc<FakeClock>,
4890 reconciler: Reconciler,
4891 }
4892
4893 impl Harness {
4894 fn build(
4895 host: Host,
4896 launcher: Arc<FakeLauncher>,
4897 demand: Arc<FakeDemand>,
4898 lock: Arc<dyn AllocationLock>,
4899 ) -> Self {
4900 let events = Arc::new(EventLog::new());
4901 let clock = Arc::new(FakeClock::default());
4902 let reconciler = Reconciler::new(
4903 host,
4904 ReconcilerPorts {
4905 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
4906 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
4907 lock,
4908 directory: Arc::new(FakeDirectory::default()),
4909 clock: Arc::clone(&clock) as Arc<dyn Clock>,
4910 jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
4911 events: Arc::clone(&events) as Arc<dyn EventSink>,
4912 },
4913 );
4914 Self {
4915 launcher,
4916 demand,
4917 events,
4918 clock,
4919 reconciler,
4920 }
4921 }
4922
4923 fn simple(capacity: u16, demand_count: u32, target: &str) -> Self {
4924 let launcher = Arc::new(FakeLauncher::new());
4925 let demand = Arc::new(FakeDemand::ready(demand_count, &repo(target)));
4926 Self::build(
4927 host_with(capacity),
4928 launcher,
4929 demand,
4930 Arc::new(InProcessAllocationLock::new()),
4931 )
4932 }
4933 }
4934
4935 fn attempt_in(state: AttemptState, id: u128, policy: u128) -> RunnerAttempt {
4936 let outcome = state.is_terminal().then(|| match state {
4937 AttemptState::Failed => {
4938 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
4939 }
4940 AttemptState::Orphaned => AttemptOutcome::Orphaned,
4941 _ => AttemptOutcome::CompletedJob,
4942 });
4943 RunnerAttempt::from_persisted(PersistedAttempt {
4944 id: AttemptId::from_u128(id),
4945 policy_id: PolicyId::from_u128(policy),
4946 github_runner_id: None,
4947 state,
4948 outcome,
4949 process_id: None,
4950 execution: runner_manager_domain::execution::AttemptExecution::Native {
4951 process_id: None,
4952 },
4953 runtime_path: "runtime/p/a".into(),
4954 workspace_kind: WorkspaceKind::Ephemeral,
4955 workspace_slot: None,
4956 created_at: fixtures::created_at(),
4957 terminal_at: state.is_terminal().then(fixtures::created_at),
4958 last_state_change_at: fixtures::created_at(),
4959 })
4960 .expect("a state/outcome pair the domain accepts")
4961 }
4962
4963 fn concluded(id: u128, policy: u128, outcome: AttemptOutcome) -> RunnerAttempt {
4965 RunnerAttempt::from_persisted(PersistedAttempt {
4966 id: AttemptId::from_u128(id),
4967 policy_id: PolicyId::from_u128(policy),
4968 github_runner_id: None,
4969 state: outcome.terminal_state(),
4970 outcome: Some(outcome),
4971 process_id: None,
4972 execution: runner_manager_domain::execution::AttemptExecution::Native {
4973 process_id: None,
4974 },
4975 runtime_path: "runtime/p/a".into(),
4976 workspace_kind: WorkspaceKind::Ephemeral,
4977 workspace_slot: None,
4978 created_at: fixtures::created_at(),
4979 terminal_at: Some(fixtures::created_at()),
4980 last_state_change_at: fixtures::created_at(),
4981 })
4982 .expect("a state/outcome pair the domain accepts")
4983 }
4984
4985 #[tokio::test]
5014 async fn three_polls_of_one_still_queued_run_yield_exactly_one_attempt() {
5015 let mut harness = Harness::simple(4, 1, "acme/app");
5016 let policy = policy(1, "acme/app", 4);
5017
5018 for poll in 1..=3 {
5019 let report = harness
5020 .reconciler
5021 .reconcile(std::slice::from_ref(&policy))
5022 .await;
5023 assert_eq!(
5024 harness.launcher.launches(),
5025 1,
5026 "poll {poll} started another runner for a job already being served; the \
5027 `- active_owned_runners` term reached `HostAllocator` as a set this host \
5028 does not hold"
5029 );
5030 assert_eq!(report.allocations.len(), 1);
5031 let allocation = &report.allocations[0];
5032 assert_eq!(allocation.demand, 1, "poll {poll}: still queued at GitHub");
5033 if poll == 1 {
5034 assert_eq!(allocation.to_start, 1);
5035 assert_eq!(report.started, 1);
5036 } else {
5037 assert_eq!(allocation.active_owned, 1, "poll {poll}");
5038 assert_eq!(allocation.to_start, 0, "poll {poll}");
5039 assert_eq!(report.started, 0, "poll {poll}");
5040 }
5041 }
5042 assert_eq!(harness.launcher.live_count(), 1);
5043 }
5044
5045 #[tokio::test]
5046 async fn draining_parallel_workflow_starts_runners_for_the_remaining_queue() {
5047 let mut harness = Harness::simple(10, 8, "acme/app");
5048 let policy = policy(1, "acme/app", 10);
5049 let first = harness
5050 .reconciler
5051 .reconcile(std::slice::from_ref(&policy))
5052 .await;
5053 assert_eq!(first.started, 8);
5054
5055 let attempts = harness.launcher.snapshot();
5057 *harness.launcher.attempts.lock().unwrap() = attempts
5058 .iter()
5059 .take(3)
5060 .map(|attempt| attempt_in(AttemptState::Busy, attempt.id.as_uuid().as_u128(), 1))
5061 .collect();
5062 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
5063 repo("acme/app"),
5064 jobs(2),
5065 )));
5066
5067 for poll in 0..3 {
5068 let report = harness
5069 .reconciler
5070 .reconcile(std::slice::from_ref(&policy))
5071 .await;
5072 assert_eq!(report.started, if poll == 0 { 2 } else { 0 });
5073 assert_eq!(harness.launcher.live_count(), 5);
5074 }
5075 }
5076
5077 #[tokio::test]
5088 async fn a_launcher_whose_attempts_never_appear_cannot_wedge_the_pass() {
5089 let launcher = Arc::new(FakeLauncher::forgetful());
5090 let mut harness = Harness::build(
5091 host_with(64),
5092 Arc::clone(&launcher),
5093 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
5094 Arc::new(InProcessAllocationLock::new()),
5095 );
5096
5097 let report = harness
5098 .reconciler
5099 .reconcile(&[policy(1, "acme/app", 8)])
5100 .await;
5101
5102 assert_eq!(
5103 report.started, 3,
5104 "the pass is bounded by the grant it was given, not by the attempt set catching \
5105 up with it"
5106 );
5107 assert_eq!(launcher.launches(), 3);
5108 assert!(
5109 launcher.snapshot().is_empty(),
5110 "the launcher never recorded anything, which is the whole point of the fixture"
5111 );
5112 }
5113
5114 #[tokio::test]
5126 async fn two_policies_cannot_exceed_host_capacity_even_when_the_launcher_lags() {
5127 let launcher = Arc::new(FakeLauncher::forgetful());
5128 let demand = Arc::new(FakeDemand::default());
5129 demand.set_for(
5130 &ScaleTarget::repository("acme/left").unwrap(),
5131 PollOutcome::Ready(QueuedDemand::of(repo("acme/left"), jobs(3))),
5132 );
5133 demand.set_for(
5134 &ScaleTarget::repository("acme/right").unwrap(),
5135 PollOutcome::Ready(QueuedDemand::of(repo("acme/right"), jobs(3))),
5136 );
5137 let mut harness = Harness::build(
5138 host_with(3),
5139 Arc::clone(&launcher),
5140 demand,
5141 Arc::new(InProcessAllocationLock::new()),
5142 );
5143
5144 let report = harness
5145 .reconciler
5146 .reconcile(&[policy(1, "acme/left", 3), policy(2, "acme/right", 3)])
5147 .await;
5148
5149 assert_eq!(
5150 report.started, 3,
5151 "host_capacity is 3 and two policies each allowed 3 started {} runners \
5152 between them; the second policy's grant was computed from a set that did \
5153 not yet contain the first policy's launches",
5154 report.started
5155 );
5156 assert_eq!(launcher.launches(), 3);
5157 }
5158
5159 #[tokio::test]
5170 async fn an_unreadable_attempt_set_starts_nothing_and_is_not_read_as_an_idle_host() {
5171 let launcher = Arc::new(FakeLauncher::new());
5172 let mut harness = Harness::build(
5173 host_with(8),
5174 Arc::clone(&launcher),
5175 Arc::new(FakeDemand::ready(4, &repo("acme/app"))),
5176 Arc::new(InProcessAllocationLock::new()),
5177 );
5178 let policy = policy(1, "acme/app", 8);
5179
5180 launcher.fail_attempts(true);
5181 let unreadable = harness
5182 .reconciler
5183 .reconcile(std::slice::from_ref(&policy))
5184 .await;
5185
5186 assert_eq!(
5187 unreadable.started, 0,
5188 "nothing may be decided from a set that was not read"
5189 );
5190 assert_eq!(launcher.launches(), 0);
5191 assert!(unreadable.attempts_unreadable > 0, "and the pass says so");
5192 assert!(
5193 unreadable.allocations.is_empty(),
5194 "no allocation is reported either: there was no set to compute one from, and \
5195 an allocation of zero would claim a decision nobody made"
5196 );
5197 assert!(harness.events.count_of("attempts_unreadable") > 0);
5198
5199 launcher.fail_attempts(false);
5201 let readable = harness
5202 .reconciler
5203 .reconcile(std::slice::from_ref(&policy))
5204 .await;
5205 assert_eq!(
5206 readable.started, 4,
5207 "the difference between the two passes is only whether the set could be read"
5208 );
5209 assert_eq!(readable.attempts_unreadable, 0);
5210 }
5211
5212 #[tokio::test]
5222 async fn a_cleanup_that_cannot_succeed_is_reported_rather_than_retried_in_silence() {
5223 let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
5224 1,
5225 1,
5226 AttemptOutcome::ExitedIdleWithoutWork,
5227 )]));
5228 let mut harness = Harness::build(
5229 host_with(4),
5230 Arc::clone(&launcher),
5231 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5232 Arc::new(InProcessAllocationLock::new()),
5233 );
5234
5235 let report = harness
5236 .reconciler
5237 .reconcile(&[policy(1, "acme/app", 4)])
5238 .await;
5239
5240 assert_eq!(report.cleaned, 0);
5241 assert_eq!(report.clean_failures, 1);
5242 assert_eq!(harness.events.count_of("attempt_clean_failed"), 1);
5243 assert_eq!(
5244 harness.events.count_of("attempt_cleaned"),
5245 0,
5246 "and it is not reported as cleaned"
5247 );
5248 assert_eq!(
5249 launcher.snapshot().len(),
5250 1,
5251 "the attempt is still there, so the retry is real -- what changed is that it \
5252 is no longer silent"
5253 );
5254
5255 let reasons: Vec<&'static str> = harness
5258 .events
5259 .events()
5260 .into_iter()
5261 .filter_map(|event| match event {
5262 LifecycleEvent::AttemptCleanFailed { reason, .. } => Some(reason),
5263 _ => None,
5264 })
5265 .collect();
5266 assert_eq!(reasons, vec!["other"]);
5267 }
5268
5269 #[tokio::test]
5272 async fn a_failing_cleanup_backs_off_exponentially_per_attempt() {
5273 let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
5274 1,
5275 1,
5276 AttemptOutcome::ExitedIdleWithoutWork,
5277 )]));
5278 let mut harness = Harness::build(
5279 host_with(4),
5280 Arc::clone(&launcher),
5281 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5282 Arc::new(InProcessAllocationLock::new()),
5283 );
5284 let policies = [policy(1, "acme/app", 4)];
5285 let start = harness.clock.now();
5286 let at = |offset: Duration| start + chrono::Duration::from_std(offset).unwrap();
5287 let mut failures_at = async |offset: Duration| {
5288 harness.clock.set(at(offset));
5289 harness.reconciler.reconcile(&policies).await.clean_failures
5290 };
5291
5292 assert_eq!(failures_at(Duration::ZERO).await, 1);
5293 assert_eq!(
5295 failures_at(CLEAN_RETRY_INITIAL - Duration::from_secs(1)).await,
5296 0,
5297 "a failed cleanup was retried on the very next pass"
5298 );
5299 assert_eq!(failures_at(CLEAN_RETRY_INITIAL).await, 1);
5300 let second = CLEAN_RETRY_INITIAL + CLEAN_RETRY_INITIAL * 2;
5302 assert_eq!(
5303 failures_at(second - Duration::from_secs(1)).await,
5304 0,
5305 "the delay did not double after a second failure"
5306 );
5307 assert_eq!(failures_at(second).await, 1);
5308 assert_eq!(harness.events.count_of("attempt_clean_failed"), 3);
5309
5310 let mut backoff = CleanBackoff::default();
5312 let id = AttemptId::from_u128(9);
5313 for _ in 0..40 {
5314 backoff.failed(id, start);
5315 }
5316 assert!(!backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM - Duration::from_secs(1))));
5317 assert!(backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM)));
5318 backoff.succeeded(id);
5319 assert!(
5320 backoff.is_due(id, start),
5321 "a cleaned attempt kept its backoff"
5322 );
5323 }
5324
5325 #[tokio::test]
5338 async fn an_unreadable_attempt_set_makes_scale_down_inconclusive_rather_than_empty() {
5339 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5340 attempt_in(AttemptState::Busy, 1, 1),
5341 concluded(2, 1, AttemptOutcome::CompletedJob),
5342 ]));
5343 let harness = Harness::build(
5344 host_with(4),
5345 Arc::clone(&launcher),
5346 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
5347 Arc::new(InProcessAllocationLock::new()),
5348 );
5349 let policy = policy(1, "acme/app", 4);
5350
5351 launcher.fail_attempts(true);
5352 let blind = harness.reconciler.scale_down(&policy).await;
5353
5354 assert!(!blind.is_conclusive(), "the machine was never read");
5355 assert_ne!(
5356 blind,
5357 ScaleDownReport::default(),
5358 "a scale-down that could not see the host must not be equal to one that saw \
5359 an idle host; that equality is the whole finding"
5360 );
5361 assert_eq!(blind.removed, 0);
5362 assert_eq!(
5363 blind.refused_busy, 0,
5364 "and this zero means `unknown`, not `none`"
5365 );
5366 assert_eq!(harness.events.count_of("attempts_unreadable"), 1);
5367
5368 launcher.fail_attempts(false);
5370 let seeing = harness.reconciler.scale_down(&policy).await;
5371
5372 assert!(seeing.is_conclusive());
5373 assert_eq!(seeing.removed, 1, "the concluded attempt was reclaimed");
5374 assert_eq!(seeing.refused_busy, 1, "and the busy one was left alone");
5375 assert_ne!(
5376 seeing, blind,
5377 "the difference between the two is only whether the set could be read"
5378 );
5379 }
5380
5381 #[tokio::test]
5389 async fn demand_below_min_capacity_is_raised_to_min_capacity() {
5390 let mut warm = ScalePolicy::new(
5391 PolicyId::from_u128(1),
5392 ScaleTarget::repository("acme/app").unwrap(),
5393 1,
5394 fixtures::HOST_ID,
5395 PolicyMode::autoscale(
5396 fixtures::routing_labels("home"),
5397 2,
5398 NonZeroU16::new(5).expect("non-zero"),
5399 )
5400 .expect("min <= max"),
5401 CachePolicy::default(),
5402 );
5403 warm.activate().expect("pending -> active");
5404
5405 let mut harness = Harness::simple(8, 0, "acme/app");
5406 let report = harness.reconciler.reconcile(&[warm]).await;
5407
5408 assert_eq!(
5409 report.allocations[0].demand, 0,
5410 "GitHub reported no queued runs"
5411 );
5412 assert_eq!(
5413 report.allocations[0].desired, 2,
5414 "min_capacity raised the target above demand"
5415 );
5416 assert_eq!(
5417 report.allocations[0].limiting_factor,
5418 LimitingFactor::MinCapacity
5419 );
5420 assert_eq!(report.started, 2, "and two runners were actually started");
5421 assert_eq!(harness.launcher.live_count(), 2);
5422 }
5423
5424 #[derive(Debug)]
5431 struct FixedMemory {
5432 free: u64,
5433 estimate: u64,
5434 admitted: std::sync::atomic::AtomicU32,
5435 }
5436
5437 impl MemoryGate for FixedMemory {
5438 fn admit(&self, _policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict {
5439 let running: Vec<runner_manager_domain::memory::Running> = attempts
5440 .iter()
5441 .filter(|attempt| attempt.counts_against_capacity())
5442 .map(|_| runner_manager_domain::memory::Running {
5443 estimate: self.estimate,
5444 current_rss: None,
5445 })
5446 .collect();
5447 runner_manager_domain::memory::admit(
5448 Some(self.free),
5449 &running,
5450 runner_manager_domain::memory::Estimate::choose(None, Some(self.estimate)),
5451 GIB,
5452 )
5453 }
5454
5455 fn admitted(&self, _policy: &ScalePolicy) {
5456 self.admitted
5457 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
5458 }
5459 }
5460
5461 const GIB: u64 = 1 << 30;
5462
5463 fn with_memory(harness: Harness, free: u64, estimate: u64) -> (Harness, Arc<FixedMemory>) {
5464 let gate = Arc::new(FixedMemory {
5465 free,
5466 estimate,
5467 admitted: std::sync::atomic::AtomicU32::new(0),
5468 });
5469 let Harness {
5470 launcher,
5471 demand,
5472 events,
5473 clock,
5474 reconciler,
5475 } = harness;
5476 let reconciler = reconciler.with_memory_gate(Arc::clone(&gate) as Arc<dyn MemoryGate>);
5477 (
5478 Harness {
5479 launcher,
5480 demand,
5481 events,
5482 clock,
5483 reconciler,
5484 },
5485 gate,
5486 )
5487 }
5488
5489 #[tokio::test]
5490 async fn a_start_the_memory_cannot_hold_is_deferred_and_the_demand_kept() {
5491 let (mut harness, gate) = with_memory(Harness::simple(8, 5, "acme/app"), 10 * GIB, 4 * GIB);
5495 let report = harness
5496 .reconciler
5497 .reconcile(&[policy(1, "acme/app", 5)])
5498 .await;
5499
5500 assert_eq!(
5501 report.allocations[0].to_start, 5,
5502 "the ceilings granted five"
5503 );
5504 assert_eq!(report.started, 2);
5505 assert_eq!(report.memory_deferred, 3, "deferred, not failed");
5506 assert_eq!(report.failures, 0);
5507 assert_eq!(gate.admitted.load(std::sync::atomic::Ordering::Relaxed), 2);
5508 let deferred: Vec<MemoryFigures> = harness
5509 .events
5510 .events()
5511 .into_iter()
5512 .filter_map(|event| match event {
5513 LifecycleEvent::MemoryShort {
5514 figures,
5515 started: false,
5516 ..
5517 } => Some(figures),
5518 _ => None,
5519 })
5520 .collect();
5521 assert_eq!(deferred.len(), 1);
5522 assert_eq!(deferred[0].free, 10 * GIB);
5523 assert_eq!(deferred[0].reserved, 8 * GIB);
5524 assert_eq!(deferred[0].needed(), 5 * GIB);
5525
5526 let again = harness
5529 .reconciler
5530 .reconcile(&[policy(1, "acme/app", 5)])
5531 .await;
5532 assert_eq!(again.started, 0);
5533 assert!(again.memory_deferred > 0);
5534 assert_eq!(harness.launcher.live_count(), 2);
5535 }
5536
5537 #[tokio::test]
5538 async fn a_runner_bigger_than_the_free_memory_starts_when_nothing_else_runs() {
5539 let (mut harness, _) = with_memory(Harness::simple(8, 2, "acme/app"), GIB, 4 * GIB);
5540 let report = harness
5541 .reconciler
5542 .reconcile(&[policy(1, "acme/app", 5)])
5543 .await;
5544
5545 assert_eq!(report.started, 1, "no starvation, but only one");
5546 assert_eq!(report.memory_deferred, 1);
5547 let names: Vec<&str> = harness
5548 .events
5549 .events()
5550 .iter()
5551 .map(LifecycleEvent::name)
5552 .filter(|name| name.starts_with("memory_"))
5553 .collect();
5554 assert_eq!(names, ["memory_admitted_alone", "memory_deferred"]);
5555 }
5556
5557 #[tokio::test]
5558 async fn without_a_memory_gate_every_granted_runner_starts() {
5559 let mut harness = Harness::simple(8, 5, "acme/app");
5560 let report = harness
5561 .reconciler
5562 .reconcile(&[policy(1, "acme/app", 5)])
5563 .await;
5564 assert_eq!(report.started, 5);
5565 assert_eq!(report.memory_deferred, 0);
5566 }
5567
5568 #[tokio::test]
5573 async fn demand_above_max_capacity_is_clamped_to_max_capacity() {
5574 let mut harness = Harness::simple(100, 10, "acme/app");
5575 let report = harness
5576 .reconciler
5577 .reconcile(&[policy(1, "acme/app", 3)])
5578 .await;
5579
5580 assert_eq!(report.allocations[0].demand, 10);
5581 assert_eq!(
5582 report.allocations[0].desired, 3,
5583 "max_capacity beats demand"
5584 );
5585 assert_eq!(report.started, 3);
5586 assert_eq!(
5587 report.allocations[0].limiting_factor,
5588 LimitingFactor::MaxCapacity
5589 );
5590 }
5591
5592 #[tokio::test]
5593 async fn demand_below_min_capacity_starts_nothing_in_v1() {
5594 let mut harness = Harness::simple(8, 0, "acme/app");
5597 let report = harness
5598 .reconciler
5599 .reconcile(&[policy(1, "acme/app", 4)])
5600 .await;
5601
5602 assert_eq!(report.allocations[0].desired, 0);
5603 assert_eq!(report.started, 0);
5604 assert!(report.starts_nothing());
5605 }
5606
5607 #[tokio::test]
5608 async fn lifecycle_replacement_intent_is_consumed_by_the_ordinary_allocator() {
5609 let policy = policy(1, "octo/repo", 1);
5610 let previous = attempt_in(AttemptState::Starting, 41, 1);
5611 let intent = ReplacementIntent {
5612 policy: policy.id,
5613 previous_attempt: previous.id,
5614 operation: "exit_before_acceptance_replacement",
5615 };
5616 let launcher = Arc::new(FakeLauncher::new().seeded(vec![previous]).replacing(intent));
5617 let demand = Arc::new(FakeDemand::ready(1, &repo("octo/repo")));
5618 let mut harness = Harness::build(
5619 host_with(1),
5620 Arc::clone(&launcher),
5621 demand,
5622 Arc::new(InProcessAllocationLock::new()),
5623 );
5624
5625 let report = harness.reconciler.reconcile(&[policy]).await;
5626
5627 assert_eq!(report.replacement_intents, 1);
5628 assert_eq!(report.started, 1);
5629 assert_eq!(launcher.launches(), 1);
5630 assert_eq!(launcher.live_count(), 1);
5631 }
5632
5633 #[tokio::test]
5634 async fn zero_host_headroom_starts_nothing_at_maximum_demand() {
5635 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5636 attempt_in(AttemptState::Busy, 1, 1),
5637 attempt_in(AttemptState::Busy, 2, 1),
5638 ]));
5639 let demand = Arc::new(FakeDemand::ready(u32::from(u16::MAX), &repo("acme/app")));
5640 let mut harness = Harness::build(
5641 host_with(2),
5642 launcher,
5643 demand,
5644 Arc::new(InProcessAllocationLock::new()),
5645 );
5646
5647 let report = harness
5648 .reconciler
5649 .reconcile(&[policy(1, "acme/app", 2)])
5650 .await;
5651 assert_eq!(report.started, 0);
5652 assert_eq!(report.allocations[0].headroom_before, 0);
5653 assert_eq!(harness.launcher.launches(), 0);
5654 }
5655
5656 #[tokio::test]
5657 async fn headroom_smaller_than_the_per_policy_allowance_wins() {
5658 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
5661 attempt_in(AttemptState::Busy, 1, 99),
5662 attempt_in(AttemptState::Busy, 2, 99),
5663 attempt_in(AttemptState::Idle, 3, 99),
5664 attempt_in(AttemptState::Starting, 4, 99),
5665 ]));
5666 let demand = Arc::new(FakeDemand::ready(5, &repo("acme/app")));
5667 let mut harness = Harness::build(
5668 host_with(6),
5669 launcher,
5670 demand,
5671 Arc::new(InProcessAllocationLock::new()),
5672 );
5673
5674 let report = harness
5675 .reconciler
5676 .reconcile(&[policy(1, "acme/app", 5)])
5677 .await;
5678 assert_eq!(
5679 report.allocations[0].desired, 5,
5680 "its own ceiling allows five"
5681 );
5682 assert_eq!(report.started, 2, "the host has two slots free");
5683 assert_eq!(
5684 report.allocations[0].limiting_factor,
5685 LimitingFactor::HostCapacity
5686 );
5687 assert_eq!(harness.launcher.live_count(), 6);
5688 }
5689
5690 #[tokio::test]
5691 async fn the_idle_host_assertion_holds() {
5692 let mut harness = Harness::simple(8, 0, "acme/app");
5695 let report = harness
5696 .reconciler
5697 .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/app", 4)])
5698 .await;
5699
5700 assert_eq!(report.started, 0);
5701 assert_eq!(harness.launcher.launches(), 0);
5702 assert!(harness.launcher.snapshot().is_empty());
5703 assert_eq!(
5704 harness
5705 .launcher
5706 .snapshot()
5707 .iter()
5708 .filter(|a| !a.is_terminal())
5709 .count(),
5710 0
5711 );
5712 }
5713
5714 #[tokio::test]
5719 async fn live_capacity_increase_is_used_without_rebuilding_the_reconciler() {
5720 let mut harness = Harness::simple(1, 5, "acme/app");
5721 let policies = [policy(1, "acme/app", 5)];
5722 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);
5723
5724 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(4);
5725 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 3);
5726 assert_eq!(harness.launcher.live_count(), 4);
5727 }
5728
5729 #[tokio::test]
5730 async fn live_capacity_decrease_keeps_busy_runners_and_waits_for_headroom() {
5731 let mut harness = Harness::simple(5, 5, "acme/app");
5732 let busy: Vec<_> = (1..=3)
5733 .map(|id| attempt_in(AttemptState::Busy, id, 1))
5734 .collect();
5735 *harness.launcher.attempts.lock().unwrap() = busy.clone();
5736 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
5737 let policies = [policy(1, "acme/app", 5)];
5738
5739 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 0);
5740 assert_eq!(harness.launcher.snapshot(), busy);
5741 assert!(harness.launcher.cleaned().is_empty());
5742
5743 harness.launcher.attempts.lock().unwrap().truncate(1);
5745 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);
5746 assert_eq!(harness.launcher.live_count(), 2);
5747 }
5748
5749 #[tokio::test]
5750 async fn live_capacity_is_rechecked_between_launches_in_the_same_pass() {
5751 let mut harness = Harness::simple(5, 5, "acme/app");
5752 *harness.launcher.capacity_after_launch.lock().unwrap() = NonZeroU16::new(1);
5753
5754 let report = harness
5755 .reconciler
5756 .reconcile(&[policy(1, "acme/app", 5)])
5757 .await;
5758 assert_eq!(report.started, 1);
5759 assert_eq!(harness.launcher.live_count(), 1);
5760 }
5761
5762 #[tokio::test]
5763 async fn live_capacity_unreadable_blocks_launches_until_the_journal_recovers() {
5764 let mut harness = Harness::simple(5, 5, "acme/app");
5765 harness
5766 .launcher
5767 .capacity_unreadable
5768 .store(true, Ordering::SeqCst);
5769 let policies = [policy(1, "acme/app", 5)];
5770 let report = harness.reconciler.reconcile(&policies).await;
5771 assert_eq!(report.started, 0);
5772 assert_eq!(report.host_capacity_unreadable, 1);
5773 assert_eq!(report.attempts_unreadable, 0);
5774 assert_eq!(harness.events.count_of("attempts_unreadable"), 0);
5775 assert!(
5776 harness
5777 .events
5778 .events()
5779 .contains(&LifecycleEvent::HostCapacityUnreadable {
5780 reason: HostCapacityError::JournalUnreadable,
5781 })
5782 );
5783
5784 harness
5785 .launcher
5786 .capacity_unreadable
5787 .store(false, Ordering::SeqCst);
5788 *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
5789 assert_eq!(harness.reconciler.reconcile(&policies).await.started, 2);
5790 }
5791
5792 #[tokio::test]
5793 async fn unreadable_local_state_does_not_treat_a_busy_host_as_idle() {
5794 for host_failure in [true, false] {
5795 let mut harness = Harness::simple(5, 0, "acme/app");
5796 harness
5797 .reconciler
5798 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
5799 *harness.launcher.attempts.lock().unwrap() = vec![attempt_in(AttemptState::Busy, 1, 1)];
5800 harness
5801 .launcher
5802 .capacity_unreadable
5803 .store(host_failure, Ordering::SeqCst);
5804 harness.launcher.fail_attempts(!host_failure);
5805 let report = harness
5806 .reconciler
5807 .reconcile(&[policy(1, "acme/app", 5)])
5808 .await;
5809 assert_eq!(report.host_capacity_unreadable, u16::from(host_failure));
5810 assert_eq!(report.next_poll.pace, PollPace::Nominal);
5811 assert_eq!(harness.launcher.live_count(), 1);
5812 }
5813 }
5814
5815 #[derive(Debug)]
5816 struct HostEditOnLock {
5817 inner: InProcessAllocationLock,
5818 launcher: Arc<FakeLauncher>,
5819 unreadable: bool,
5820 }
5821
5822 #[async_trait::async_trait]
5823 impl AllocationLock for HostEditOnLock {
5824 async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
5825 let guard = self.inner.acquire().await?;
5826 *self.launcher.capacity.lock().unwrap() = NonZeroU16::new(1);
5827 self.launcher
5828 .capacity_unreadable
5829 .store(self.unreadable, Ordering::SeqCst);
5830 Ok(guard)
5831 }
5832 }
5833
5834 #[tokio::test]
5835 async fn live_capacity_edits_and_read_failures_while_waiting_for_the_lock_block_launches() {
5836 for unreadable in [false, true] {
5837 let launcher =
5838 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
5839 let lock = Arc::new(HostEditOnLock {
5840 inner: InProcessAllocationLock::new(),
5841 launcher: Arc::clone(&launcher),
5842 unreadable,
5843 });
5844 let mut harness = Harness::build(
5845 host_with(5),
5846 Arc::clone(&launcher),
5847 Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
5848 lock,
5849 );
5850 let report = harness
5851 .reconciler
5852 .reconcile(&[policy(1, "acme/app", 5)])
5853 .await;
5854 assert_eq!(report.started, 0);
5855 assert_eq!(launcher.live_count(), 1);
5856 assert!(launcher.cleaned().is_empty());
5857 assert_eq!(report.host_capacity_unreadable, u16::from(unreadable));
5858 assert_eq!(report.attempts_unreadable, 0);
5859 }
5860 }
5861
5862 #[tokio::test(flavor = "current_thread")]
5885 async fn two_policies_reconciling_concurrently_never_exceed_host_capacity() {
5886 let lock = Arc::new(CountingLock::new());
5887 let (launches, live) =
5888 two_policies_concurrently(Arc::clone(&lock) as Arc<dyn AllocationLock>, None).await;
5889
5890 assert_eq!(
5891 launches, 3,
5892 "the sum across policies must never exceed host_capacity, and each policy is \
5893 individually within its own max_capacity of 3"
5894 );
5895 assert_eq!(live, 3, "and no duplicate runner survived the race");
5896 assert_eq!(
5897 lock.peak(),
5898 1,
5899 "the allocation lock had one holder at a time; without that the read of the \
5900 headroom and the creation of the runtime are not atomic"
5901 );
5902 assert!(
5903 (3..=5).contains(&lock.acquisitions()),
5904 "the lock is taken before *each* runtime, not once per pass: three runtimes \
5905 means at least three holds, and at most one further hold per policy to \
5906 discover the host filled up underneath it. It was taken {} times",
5907 lock.acquisitions()
5908 );
5909 }
5910
5911 #[tokio::test(flavor = "current_thread")]
5918 async fn without_the_allocation_lock_two_policies_oversubscribe_the_host() {
5919 let (launches, _) =
5920 two_policies_concurrently(Arc::new(NoLock) as Arc<dyn AllocationLock>, None).await;
5921
5922 assert!(
5923 launches > 3,
5924 "with no serialisation both policies must be able to spend the same headroom; \
5925 they started {launches} runners on a host of 3. If this is ever 3, the \
5926 contention window closed and `two_policies_reconciling_concurrently_never_\
5927 exceed_host_capacity` has stopped proving anything"
5928 );
5929 }
5930
5931 async fn two_policies_concurrently(
5935 lock: Arc<dyn AllocationLock>,
5936 live_capacity: Option<NonZeroU16>,
5937 ) -> (usize, usize) {
5938 let launcher = Arc::new(FakeLauncher::new().with_yields(4));
5939 *launcher.capacity.lock().unwrap() = live_capacity;
5940 let host = host_with(3);
5941
5942 let mut left = Harness::build(
5943 host.clone(),
5944 Arc::clone(&launcher),
5945 Arc::new(FakeDemand::ready(3, &repo("acme/left"))),
5946 Arc::clone(&lock),
5947 )
5948 .reconciler;
5949 let mut right = Harness::build(
5950 host,
5951 Arc::clone(&launcher),
5952 Arc::new(FakeDemand::ready(3, &repo("acme/right"))),
5953 Arc::clone(&lock),
5954 )
5955 .reconciler;
5956
5957 let a = policy(1, "acme/left", 3);
5958 let b = policy(2, "acme/right", 3);
5959
5960 let left = tokio::spawn(async move { left.reconcile(&[a]).await });
5961 let right = tokio::spawn(async move { right.reconcile(&[b]).await });
5962 let (_, _) = (left.await.unwrap(), right.await.unwrap());
5963
5964 (launcher.launches(), launcher.live_count())
5965 }
5966
5967 #[tokio::test(flavor = "current_thread")]
5968 async fn live_capacity_is_shared_by_concurrent_targets_with_stale_startup_hosts() {
5969 let (launches, live) =
5970 two_policies_concurrently(Arc::new(InProcessAllocationLock::new()), NonZeroU16::new(2))
5971 .await;
5972 assert_eq!(launches, 2);
5973 assert_eq!(live, 2);
5974 }
5975
5976 #[tokio::test]
5988 async fn a_monitor_only_policy_under_maximum_demand_starts_nothing_and_polls_nothing() {
5989 let gateway = FakeGithub::new().with_queued_jobs(repo("acme/app"), jobs(10_000));
5990 let gateway = Arc::new(GatewayDemand::new(gateway, CancelToken::new()));
5991 let launcher = Arc::new(FakeLauncher::new());
5992 let events = Arc::new(EventLog::new());
5993
5994 let mut reconciler = Reconciler::new(
5995 host_with(10),
5996 ReconcilerPorts {
5997 demand: Arc::clone(&gateway) as Arc<dyn DemandSource>,
5998 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
5999 lock: Arc::new(InProcessAllocationLock::new()),
6000 directory: Arc::new(FakeDirectory::default()),
6001 clock: Arc::new(FakeClock::default()),
6002 jitter: Arc::new(NoJitter),
6003 events: Arc::clone(&events) as Arc<dyn EventSink>,
6004 },
6005 );
6006
6007 let monitor = fixtures::policy()
6008 .id(PolicyId::from_u128(1))
6009 .repository("acme/app")
6010 .monitor_only()
6011 .active()
6012 .build();
6013
6014 let report = reconciler.reconcile(&[monitor]).await;
6015
6016 assert_eq!(report.started, 0);
6017 assert_eq!(launcher.launches(), 0);
6018 assert_eq!(report.monitor_only, vec![PolicyId::from_u128(1)]);
6019 assert_eq!(
6020 report.demand_requests, 0,
6021 "a monitor-only policy spends nothing from the shared hourly ceiling"
6022 );
6023 assert!(
6024 gateway.gateway().calls().is_empty(),
6025 "a monitor-only policy issued a demand request: {:?}",
6026 gateway.gateway().calls()
6027 );
6028 assert_eq!(events.count_of("monitor_only_skipped"), 1);
6029 assert_eq!(
6030 events.count_of("demand_observed"),
6031 0,
6032 "and it contributed no demand"
6033 );
6034 }
6035
6036 #[tokio::test]
6054 async fn a_monitor_only_policy_sharing_a_target_is_still_skipped_entirely() {
6055 let lock = Arc::new(CountingLock::new());
6056 let launcher = Arc::new(FakeLauncher::new());
6057 let events = Arc::new(EventLog::new());
6058 let mut reconciler = Reconciler::new(
6059 host_with(4),
6060 ReconcilerPorts {
6061 demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
6062 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
6063 lock: Arc::clone(&lock) as Arc<dyn AllocationLock>,
6064 directory: Arc::new(FakeDirectory::default()),
6065 clock: Arc::new(FakeClock::default()),
6066 jitter: Arc::new(NoJitter),
6067 events: Arc::clone(&events) as Arc<dyn EventSink>,
6068 },
6069 );
6070
6071 let watcher = fixtures::policy()
6072 .id(PolicyId::from_u128(2))
6073 .repository("acme/app")
6074 .monitor_only()
6075 .active()
6076 .build();
6077
6078 let report = reconciler
6079 .reconcile(&[policy(1, "acme/app", 4), watcher])
6080 .await;
6081
6082 assert_eq!(report.started, 2, "the autoscale policy is served normally");
6083 assert_eq!(report.monitor_only, vec![PolicyId::from_u128(2)]);
6084 assert_eq!(
6085 events.count_of("demand_observed"),
6086 1,
6087 "the demand observation belongs to the autoscale policy alone"
6088 );
6089 assert!(
6090 report
6091 .allocations
6092 .iter()
6093 .all(|a| a.policy_id == PolicyId::from_u128(1)),
6094 "a monitor-only policy was allocated for: {:?}",
6095 report.allocations
6096 );
6097 assert_eq!(
6098 lock.acquisitions(),
6099 2,
6100 "one hold per runtime created, and none on behalf of the monitor-only policy. \
6101 It was three before the budget was checked at the top of the loop rather than \
6102 after the re-read, which cost every policy a surplus hold to discover there \
6103 was nothing left to grant"
6104 );
6105 }
6106
6107 #[tokio::test]
6108 async fn the_monitor_only_refusal_is_asserted_on_the_mode_not_on_a_missing_ceiling() {
6109 let monitor = fixtures::monitor_only_policy();
6114 assert!(!monitor.owns_runners());
6115 assert_eq!(monitor.max_capacity(), None);
6116
6117 let host = host_with(10);
6118 let attempts: Vec<RunnerAttempt> = Vec::new();
6119 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
6120 let allocation = allocator.allocate(&monitor, 10_000);
6121 assert_eq!(allocation.limiting_factor, LimitingFactor::MonitorOnly);
6122 assert_eq!(allocation.to_start, 0);
6123 assert_eq!(
6124 allocator.headroom(),
6125 10,
6126 "and it consumes no headroom, so an autoscale policy on the same host is \
6127 unaffected"
6128 );
6129 }
6130
6131 #[tokio::test]
6139 async fn a_surplus_attempt_is_cleaned_as_an_idle_exit_and_not_as_a_failure() {
6140 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
6141 concluded(1, 1, AttemptOutcome::ExitedIdleWithoutWork),
6142 concluded(
6143 2,
6144 1,
6145 AttemptOutcome::failed(FailureReason::JitRequestFailed),
6146 ),
6147 concluded(3, 1, AttemptOutcome::CompletedJob),
6148 ]));
6149 let mut harness = Harness::build(
6150 host_with(4),
6151 Arc::clone(&launcher),
6152 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
6153 Arc::new(InProcessAllocationLock::new()),
6154 );
6155
6156 let report = harness
6157 .reconciler
6158 .reconcile(&[policy(1, "acme/app", 4)])
6159 .await;
6160
6161 assert_eq!(report.cleaned, 3);
6162 assert_eq!(report.idle_exits, 1, "the surplus case, counted apart");
6163 assert_eq!(
6164 report.failures, 1,
6165 "only the failed attempt is a failure; the idle exit and the completed job are \
6166 not"
6167 );
6168 assert_eq!(launcher.cleaned().len(), 3);
6169 assert!(launcher.snapshot().is_empty());
6170
6171 let cleaned: Vec<OutcomeKind> = harness
6172 .events
6173 .events()
6174 .into_iter()
6175 .filter_map(|event| match event {
6176 LifecycleEvent::AttemptCleaned { outcome, .. } => Some(outcome),
6177 _ => None,
6178 })
6179 .collect();
6180 assert!(cleaned.contains(&OutcomeKind::IdleExit));
6181 assert!(
6182 !OutcomeKind::IdleExit.is_failure(),
6183 "an idle exit rendered as a failure sends an operator hunting a fault that does \
6184 not exist"
6185 );
6186 }
6187
6188 #[tokio::test]
6191 async fn scale_down_removes_nothing_from_a_busy_attempt() {
6192 let busy = attempt_in(AttemptState::Busy, 1, 1);
6193 let launcher = Arc::new(FakeLauncher::new().seeded(vec![
6194 busy.clone(),
6195 attempt_in(AttemptState::Starting, 2, 1),
6196 concluded(3, 1, AttemptOutcome::CompletedJob),
6197 ]));
6198 let harness = Harness::build(
6199 host_with(4),
6200 Arc::clone(&launcher),
6201 Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
6202 Arc::new(InProcessAllocationLock::new()),
6203 );
6204
6205 let report = harness
6206 .reconciler
6207 .scale_down(&policy(1, "acme/app", 4))
6208 .await;
6209
6210 assert_eq!(report.refused_busy, 1);
6211 assert_eq!(
6212 report.retained, 1,
6213 "the `starting` attempt is not ended either"
6214 );
6215 assert_eq!(report.removed, 1, "only the concluded attempt is reclaimed");
6216
6217 let after = launcher.snapshot();
6218 let still_busy = after
6219 .iter()
6220 .find(|a| a.id == AttemptId::from_u128(1))
6221 .expect("the busy attempt is still there");
6222 assert_eq!(
6223 still_busy.state(),
6224 AttemptState::Busy,
6225 "scale-down removed nothing from a runner that is executing a job, and left it \
6226 busy"
6227 );
6228 assert!(!launcher.cleaned().contains(&AttemptId::from_u128(1)));
6229
6230 let mut busy = busy;
6234 assert!(matches!(
6235 busy.clean(fixtures::created_at()),
6236 Err(runner_manager_domain::attempt::AttemptError::BusyCannotBeCleaned)
6237 ));
6238 assert_eq!(harness.events.count_of("scale_down_refused"), 1);
6239 }
6240
6241 #[test]
6246 fn the_default_interval_is_sixty_seconds_and_the_floor_is_thirty() {
6247 assert_eq!(RefreshInterval::DEFAULT_SECS, 60);
6248 assert_eq!(RefreshInterval::MIN_SECS, 30);
6249 assert_eq!(PollSchedule::floor(), Duration::from_secs(30));
6250 assert!(
6251 RefreshInterval::from_secs(29).is_err(),
6252 "the floor is a rate-budget constraint, and a caller must not be able to write \
6253 a shorter interval at all"
6254 );
6255
6256 let mut schedule = PollSchedule::new(RefreshInterval::default());
6257 let next = schedule.next_poll(None, fixtures::created_at(), &NoJitter);
6258 assert_eq!(next.delay, Duration::from_secs(60));
6259 assert_eq!(next.pace, PollPace::Nominal);
6260
6261 let mut floored = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
6262 assert_eq!(
6263 floored
6264 .next_poll(None, fixtures::created_at(), &NoJitter)
6265 .delay,
6266 Duration::from_secs(30)
6267 );
6268 }
6269
6270 #[test]
6274 fn a_rate_limit_increases_the_delay_and_names_itself() {
6275 let now = fixtures::created_at();
6276 let mut schedule = PollSchedule::new(RefreshInterval::default());
6277
6278 let limited = RefreshState::RateLimited(RateLimited {
6279 kind: RateLimitKind::Secondary,
6280 retry_after: Some(Duration::from_secs(300)),
6281 remaining: None,
6282 reset_unix_secs: None,
6283 });
6284 let next = schedule.next_poll(Some(&limited), now, &NoJitter);
6285
6286 assert_eq!(next.delay, Duration::from_secs(300));
6287 assert_eq!(
6288 next.pace,
6289 PollPace::RateLimited {
6290 kind: RateLimitKind::Secondary
6291 },
6292 "the increase is reported, never hidden"
6293 );
6294 assert!(next.pace.is_throttled());
6295 assert_eq!(next.pace.as_str(), "rate_limited_secondary");
6296 }
6297
6298 #[test]
6301 fn the_retry_delay_is_an_absolute_floor_and_never_an_addend() {
6302 let now = fixtures::created_at();
6303 let mut schedule = PollSchedule::new(RefreshInterval::default());
6304
6305 let limited = RefreshState::RateLimited(RateLimited {
6306 kind: RateLimitKind::Primary,
6307 retry_after: Some(Duration::from_secs(300)),
6308 remaining: Some(0),
6309 reset_unix_secs: None,
6310 });
6311
6312 for _ in 0..5 {
6315 let next = schedule.next_poll(Some(&limited), now, &NoJitter);
6316 assert_eq!(
6317 next.delay,
6318 Duration::from_secs(300),
6319 "the delay is `max(interval, retry_delay)`; `interval + retry_delay` would \
6320 have compounded on every successive retry"
6321 );
6322 }
6323
6324 let brief = RefreshState::RateLimited(RateLimited {
6327 kind: RateLimitKind::Secondary,
6328 retry_after: Some(Duration::from_secs(5)),
6329 remaining: None,
6330 reset_unix_secs: None,
6331 });
6332 let next = schedule.next_poll(Some(&brief), now, &NoJitter);
6333 assert_eq!(
6334 next.delay,
6335 Duration::from_secs(60),
6336 "a short `retry-after` may not drop the loop below its own interval"
6337 );
6338 assert!(next.delay >= PollSchedule::floor());
6339 }
6340
6341 #[test]
6342 fn no_branch_of_the_schedule_can_go_below_the_thirty_second_floor() {
6343 let now = fixtures::created_at();
6344 let states = [
6345 None,
6346 Some(RefreshState::Offline),
6347 Some(RefreshState::RateLimited(RateLimited {
6348 kind: RateLimitKind::Secondary,
6349 retry_after: Some(Duration::from_secs(1)),
6350 remaining: None,
6351 reset_unix_secs: None,
6352 })),
6353 Some(RefreshState::LockedOut {
6354 retry_after: Duration::from_secs(1),
6355 }),
6356 Some(RefreshState::Unauthorized),
6357 Some(RefreshState::Forbidden { message: None }),
6358 Some(RefreshState::Failed {
6359 status: Some(500),
6360 message: "server error".into(),
6361 }),
6362 Some(RefreshState::Cancelled),
6363 ];
6364
6365 for state in &states {
6366 let mut schedule = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
6367 let next = schedule.next_poll(state.as_ref(), now, &NoJitter);
6368 assert!(
6369 next.delay >= PollSchedule::floor(),
6370 "{state:?} scheduled a poll {}ms away, under the 30-second floor",
6371 next.delay.as_millis()
6372 );
6373 }
6374 }
6375
6376 #[test]
6377 fn an_offline_run_backs_off_with_jitter_and_a_recovery_resets_it() {
6378 let now = fixtures::created_at();
6379 let mut schedule = PollSchedule::new(RefreshInterval::default());
6380
6381 let mut previous = Duration::ZERO;
6383 for consecutive in 1..=6_u32 {
6384 let next = schedule.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
6385 assert_eq!(next.pace, PollPace::Offline { consecutive });
6386 assert!(
6387 next.delay >= previous,
6388 "the back-off must not shrink while the outage continues"
6389 );
6390 assert!(next.delay >= Duration::from_secs(60));
6391 previous = next.delay;
6392 }
6393 assert!(previous <= MAX_OFFLINE_BACKOFF, "and it is capped");
6394
6395 let mut jittered = PollSchedule::new(RefreshInterval::default());
6398 let none = jittered.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
6399 let mut jittered = PollSchedule::new(RefreshInterval::default());
6400 let full = jittered.next_poll(Some(&RefreshState::Offline), now, &FixedJitter(0.999));
6401 assert!(full.delay > none.delay);
6402 assert!(full.delay <= none.delay.mul_f64(1.0 + JITTER_RATIO));
6403
6404 assert_eq!(schedule.consecutive_offline(), 6);
6406 let recovered = schedule.next_poll(None, now, &NoJitter);
6407 assert_eq!(recovered.pace, PollPace::Nominal);
6408 assert_eq!(recovered.delay, Duration::from_secs(60));
6409 assert_eq!(schedule.consecutive_offline(), 0);
6410 }
6411
6412 #[test]
6413 fn the_offline_state_states_the_twenty_four_hour_bound() {
6414 assert_eq!(
6415 GITHUB_CANCELS_QUEUED_JOBS_AFTER,
6416 Duration::from_secs(24 * 60 * 60)
6417 );
6418
6419 let brief = OfflineState::new(1, Duration::from_secs(120));
6420 let rendered = brief.to_string();
6421 assert!(rendered.contains("24 hours"), "{rendered}");
6422 assert!(rendered.contains("Retrying in 120s"), "{rendered}");
6423 assert!(!brief.has_outlasted_the_queue());
6424
6425 let long = brief.since(GITHUB_CANCELS_QUEUED_JOBS_AFTER + Duration::from_secs(1));
6426 assert!(long.has_outlasted_the_queue());
6427 assert!(
6428 long.to_string().contains("queued work has been lost"),
6429 "{long}"
6430 );
6431
6432 assert!(!OfflineState::new(9, Duration::from_secs(60)).has_outlasted_the_queue());
6434 }
6435
6436 fn quota(remaining: u64, reset_in_secs: i64, now: Timestamp) -> RateLimitSnapshot {
6441 RateLimitSnapshot {
6442 limit: 5_000,
6443 remaining,
6444 used: Some(5_000 - remaining),
6445 reset_unix_secs: u64::try_from(now.timestamp() + reset_in_secs).expect("positive"),
6446 observed_at: now,
6447 }
6448 }
6449
6450 fn quiet() -> PollConditions {
6451 PollConditions {
6452 quiet: true,
6453 quota: None,
6454 requests_per_pass: 0,
6455 }
6456 }
6457
6458 #[test]
6459 fn a_quiet_pass_polls_at_the_idle_interval_and_a_busy_one_at_the_active_interval() {
6460 let now = fixtures::created_at();
6461 let mut schedule =
6462 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6463
6464 let idle = schedule.next_poll_with(None, quiet(), now, &NoJitter);
6465 assert_eq!(idle.delay, Duration::from_secs(10));
6466 assert_eq!(idle.pace, PollPace::Idle);
6467 assert!(!idle.pace.is_throttled(), "idle is faster, not a slowdown");
6468
6469 let busy = schedule.next_poll_with(None, PollConditions::default(), now, &NoJitter);
6470 assert_eq!(busy.delay, Duration::from_secs(60));
6471 assert_eq!(busy.pace, PollPace::Nominal);
6472 }
6473
6474 #[test]
6475 fn a_schedule_never_told_an_idle_interval_keeps_its_active_one() {
6476 let now = fixtures::created_at();
6477 let mut schedule = PollSchedule::new(RefreshInterval::default());
6478 assert_eq!(
6479 schedule.next_poll_with(None, quiet(), now, &NoJitter).delay,
6480 Duration::from_secs(60)
6481 );
6482 }
6483
6484 #[test]
6485 fn an_idle_interval_is_clamped_between_its_floor_and_the_active_interval() {
6486 let schedule = PollSchedule::new(RefreshInterval::from_secs(30).expect("the floor"))
6487 .with_idle(Duration::from_secs(120));
6488 assert_eq!(
6489 schedule.idle(),
6490 Duration::from_secs(30),
6491 "an idle interval above the active one would make quiet targets slower to react"
6492 );
6493 let schedule =
6494 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(1));
6495 assert_eq!(schedule.idle(), PollSchedule::idle_floor());
6496 }
6497
6498 #[test]
6501 fn a_low_quota_stretches_polling_and_wins_over_idle() {
6502 let now = fixtures::created_at();
6503 let mut schedule =
6504 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6505
6506 let healthy = schedule.next_poll_with(
6507 None,
6508 PollConditions {
6509 quiet: true,
6510 quota: Some(quota(1_000, 3_000, now)),
6511 requests_per_pass: 0,
6512 },
6513 now,
6514 &NoJitter,
6515 );
6516 assert_eq!(healthy.pace, PollPace::Idle, "a fifth left is still normal");
6517
6518 let low = schedule.next_poll_with(
6519 None,
6520 PollConditions {
6521 quiet: true,
6522 quota: Some(quota(999, 3_000, now)),
6523 requests_per_pass: 0,
6524 },
6525 now,
6526 &NoJitter,
6527 );
6528 assert_eq!(low.delay, Duration::from_secs(120));
6529 assert_eq!(
6530 low.pace,
6531 PollPace::Stretched {
6532 pressure: PrimaryPressure::Low
6533 }
6534 );
6535 assert!(low.pace.is_throttled());
6536 assert_eq!(low.pace.as_str(), "stretched_low_budget");
6537
6538 let critical = schedule.next_poll_with(
6539 None,
6540 PollConditions {
6541 quiet: false,
6542 quota: Some(quota(100, 3_000, now)),
6543 requests_per_pass: 0,
6544 },
6545 now,
6546 &NoJitter,
6547 );
6548 assert_eq!(critical.delay, Duration::from_secs(240));
6549 assert_eq!(critical.pace.as_str(), "stretched_critical_budget");
6550 }
6551
6552 #[test]
6553 fn a_stretch_never_outlasts_the_quotas_reset_nor_undercuts_the_active_interval() {
6554 let now = fixtures::created_at();
6555 let mut schedule = PollSchedule::new(RefreshInterval::default());
6556
6557 let soon = schedule.next_poll_with(
6558 None,
6559 PollConditions {
6560 quiet: false,
6561 quota: Some(quota(10, 90, now)),
6562 requests_per_pass: 0,
6563 },
6564 now,
6565 &NoJitter,
6566 );
6567 assert_eq!(
6568 soon.delay,
6569 Duration::from_secs(90),
6570 "the bucket refills in 90 s; waiting 240 s buys nothing"
6571 );
6572
6573 let imminent = schedule.next_poll_with(
6574 None,
6575 PollConditions {
6576 quiet: false,
6577 quota: Some(quota(10, 5, now)),
6578 requests_per_pass: 0,
6579 },
6580 now,
6581 &NoJitter,
6582 );
6583 assert_eq!(imminent.delay, Duration::from_secs(60));
6584
6585 let expired = schedule.next_poll_with(
6586 None,
6587 PollConditions {
6588 quiet: true,
6589 quota: Some(quota(10, -5, now)),
6590 requests_per_pass: 0,
6591 },
6592 now,
6593 &NoJitter,
6594 );
6595 assert_eq!(
6596 expired.pace,
6597 PollPace::Nominal,
6598 "a reading from a window that has reset says nothing; the schedule here had no \
6599 idle interval, so it is the active one"
6600 );
6601 }
6602
6603 #[test]
6604 fn an_idle_target_too_large_to_poll_that_often_stretches_its_idle_interval() {
6605 let now = fixtures::created_at();
6606 let mut schedule =
6607 PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
6608 let small = schedule.next_poll_with(
6609 None,
6610 PollConditions {
6611 quiet: true,
6612 quota: None,
6613 requests_per_pass: 40,
6614 },
6615 now,
6616 &NoJitter,
6617 );
6618 assert_eq!(small.delay, Duration::from_secs(10), "ten repositories fit");
6619 let organization = schedule.next_poll_with(
6620 None,
6621 PollConditions {
6622 quiet: true,
6623 quota: None,
6624 requests_per_pass: 400,
6625 },
6626 now,
6627 &NoJitter,
6628 );
6629 assert_eq!(
6630 organization.delay,
6631 Duration::from_secs(80),
6632 "400 requests a pass at 300 a minute is one pass every 80 s"
6633 );
6634 assert_eq!(organization.pace, PollPace::Idle);
6635 }
6636
6637 #[test]
6638 fn a_rate_limit_wait_is_jittered_upward_only() {
6639 let now = fixtures::created_at();
6640 let mut schedule = PollSchedule::new(RefreshInterval::default());
6641 let limited = RefreshState::RateLimited(RateLimited {
6642 kind: RateLimitKind::Secondary,
6643 retry_after: Some(Duration::from_secs(300)),
6644 remaining: None,
6645 reset_unix_secs: None,
6646 });
6647 assert_eq!(
6648 schedule
6649 .next_poll(Some(&limited), now, &FixedJitter(0.5))
6650 .delay,
6651 Duration::from_secs(315),
6652 "every loop shares one quiet period; without jitter they all wake together"
6653 );
6654 assert_eq!(
6655 schedule.next_poll(Some(&limited), now, &NoJitter).delay,
6656 Duration::from_secs(300),
6657 "never shorter than GitHub asked"
6658 );
6659 }
6660
6661 #[tokio::test]
6662 async fn demand_this_host_did_not_serve_keeps_the_target_off_the_idle_pace() {
6663 let launcher =
6666 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 9, 9)]));
6667 let demand = Arc::new(FakeDemand::ready(2, &repo("acme/app")));
6668 let mut harness = Harness::build(
6669 host_with(1),
6670 launcher,
6671 demand,
6672 Arc::new(InProcessAllocationLock::new()),
6673 );
6674 harness
6675 .reconciler
6676 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6677 let policy = policy(1, "acme/app", 4);
6678
6679 let report = harness
6680 .reconciler
6681 .reconcile(std::slice::from_ref(&policy))
6682 .await;
6683
6684 assert_eq!(report.started, 0, "the host is full");
6685 assert_eq!(
6686 report.next_poll.pace,
6687 PollPace::Nominal,
6688 "matched demand it could not serve is work waiting, not idle"
6689 );
6690 }
6691
6692 #[tokio::test]
6693 async fn a_target_with_nothing_to_do_is_polled_at_the_idle_interval() {
6694 let mut harness = Harness::simple(4, 0, "acme/app");
6695 harness
6696 .reconciler
6697 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6698 let policy = policy(1, "acme/app", 4);
6699
6700 let report = harness
6701 .reconciler
6702 .reconcile(std::slice::from_ref(&policy))
6703 .await;
6704
6705 assert_eq!(report.next_poll.pace, PollPace::Idle);
6706 assert_eq!(report.next_poll.delay, Duration::from_secs(10));
6707 }
6708
6709 #[tokio::test]
6710 async fn a_target_whose_runs_are_changing_is_polled_at_the_active_interval() {
6711 let mut harness = Harness::simple(4, 0, "acme/app");
6712 harness
6713 .reconciler
6714 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6715 harness.demand.set(PollOutcome::Ready(
6716 QueuedDemand::of(repo("acme/app"), Vec::new()).with_busy(repo("acme/app")),
6717 ));
6718 let policy = policy(1, "acme/app", 4);
6719
6720 let report = harness
6721 .reconciler
6722 .reconcile(std::slice::from_ref(&policy))
6723 .await;
6724
6725 assert_eq!(report.next_poll.pace, PollPace::Nominal);
6726 assert_eq!(report.next_poll.delay, Duration::from_secs(60));
6727 }
6728
6729 #[tokio::test]
6730 async fn a_target_with_a_live_runner_is_polled_at_the_active_interval() {
6731 let launcher =
6732 Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
6733 let demand = Arc::new(FakeDemand::ready(0, &repo("acme/app")));
6734 let mut harness = Harness::build(
6735 host_with(4),
6736 launcher,
6737 demand,
6738 Arc::new(InProcessAllocationLock::new()),
6739 );
6740 harness
6741 .reconciler
6742 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6743 let policy = policy(1, "acme/app", 4);
6744
6745 let report = harness
6746 .reconciler
6747 .reconcile(std::slice::from_ref(&policy))
6748 .await;
6749
6750 assert_eq!(
6751 report.next_poll.pace,
6752 PollPace::Nominal,
6753 "a job is running: its follow-on work is what the active interval watches for"
6754 );
6755 }
6756
6757 #[tokio::test]
6758 async fn a_pass_that_starts_a_runner_is_not_idle() {
6759 let mut harness = Harness::simple(4, 2, "acme/app");
6760 harness
6761 .reconciler
6762 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6763 let policy = policy(1, "acme/app", 4);
6764
6765 let report = harness
6766 .reconciler
6767 .reconcile(std::slice::from_ref(&policy))
6768 .await;
6769
6770 assert!(report.started > 0);
6771 assert_eq!(report.next_poll.pace, PollPace::Nominal);
6772 }
6773
6774 #[tokio::test]
6775 async fn the_quota_reported_with_a_reading_stretches_the_next_poll() {
6776 let mut harness = Harness::simple(4, 0, "acme/app");
6777 harness
6778 .reconciler
6779 .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
6780 let now = harness.clock.now();
6781 harness.demand.set(PollOutcome::Ready(
6782 QueuedDemand::of(repo("acme/app"), Vec::new())
6783 .with_rate_limit(Some(quota(200, 1_800, now))),
6784 ));
6785 let policy = policy(1, "acme/app", 4);
6786
6787 let report = harness
6788 .reconciler
6789 .reconcile(std::slice::from_ref(&policy))
6790 .await;
6791
6792 assert_eq!(
6793 report.next_poll.pace,
6794 PollPace::Stretched {
6795 pressure: PrimaryPressure::Critical
6796 }
6797 );
6798 assert_eq!(report.next_poll.delay, Duration::from_secs(240));
6799 }
6800
6801 #[tokio::test]
6810 async fn an_unreachable_github_starts_nothing_retains_everything_and_backs_off() {
6811 let live = vec![
6812 attempt_in(AttemptState::Busy, 1, 1),
6813 attempt_in(AttemptState::Starting, 2, 1),
6814 ];
6815 let launcher = Arc::new(FakeLauncher::new().seeded(live.clone()));
6816 let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
6817 let mut harness = Harness::build(
6818 host_with(8),
6819 Arc::clone(&launcher),
6820 Arc::clone(&demand),
6821 Arc::new(InProcessAllocationLock::new()),
6822 );
6823 let policy = policy(1, "acme/app", 8);
6824
6825 let report = harness
6826 .reconciler
6827 .reconcile(std::slice::from_ref(&policy))
6828 .await;
6829
6830 assert!(report.is_offline());
6831 assert_eq!(report.started, 0, "no new runner during an outage");
6832 assert_eq!(launcher.launches(), 0);
6833 assert_eq!(
6834 launcher.snapshot(),
6835 live,
6836 "existing runner processes are retained, untouched"
6837 );
6838 assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
6839 assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
6840 assert!(report.next_poll.delay >= Duration::from_secs(60));
6841 let offline = report.offline_state().expect("an offline state to display");
6842 assert!(offline.to_string().contains("24 hours"));
6843
6844 demand.set(PollOutcome::Ready(QueuedDemand::of(
6847 repo("acme/app"),
6848 jobs(2),
6849 )));
6850 let recovered = harness.reconciler.reconcile(&[policy]).await;
6851
6852 assert!(!recovered.is_offline());
6853 assert_eq!(recovered.next_poll.pace, PollPace::Nominal);
6854 assert_eq!(
6855 recovered.started, 1,
6856 "two queued jobs and one starting runner leave one job uncovered"
6857 );
6858 assert_eq!(recovered.allocations[0].active_owned, 2);
6859 assert_eq!(launcher.live_count(), 3);
6860 }
6861
6862 #[tokio::test]
6872 async fn one_offline_target_does_not_stop_a_reachable_one() {
6873 let mut harness = Harness::simple(8, 0, "acme/app");
6874 harness.demand.set_for(
6875 &ScaleTarget::repository("acme/app").unwrap(),
6876 PollOutcome::Ready(QueuedDemand::of(repo("acme/app"), jobs(2))),
6877 );
6878 harness.demand.set_for(
6879 &ScaleTarget::repository("acme/broken").unwrap(),
6880 PollOutcome::Failed(RefreshState::Offline),
6881 );
6882
6883 let report = harness
6884 .reconciler
6885 .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/broken", 4)])
6886 .await;
6887
6888 assert_eq!(
6889 report.started, 2,
6890 "the reachable target was served; an unreachable sibling repository must not \
6891 idle the host"
6892 );
6893 assert_eq!(report.unreadable, vec![PolicyId::from_u128(2)]);
6894 assert_eq!(harness.demand.polls().len(), 2, "both targets were polled");
6895
6896 assert!(report.is_offline());
6898 assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
6899 }
6900
6901 #[tokio::test]
6916 async fn a_day_long_outage_says_that_queued_work_has_already_been_lost() {
6917 let clock = Arc::new(FakeClock::default());
6918 let launcher = Arc::new(FakeLauncher::new());
6919 let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
6920 let mut reconciler = Reconciler::new(
6921 host_with(4),
6922 ReconcilerPorts {
6923 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
6924 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
6925 lock: Arc::new(InProcessAllocationLock::new()),
6926 directory: Arc::new(FakeDirectory::default()),
6927 clock: Arc::clone(&clock) as Arc<dyn Clock>,
6928 jitter: Arc::new(NoJitter),
6929 events: Arc::new(NoEvents),
6930 },
6931 );
6932 let policy = policy(1, "acme/app", 4);
6933
6934 let first = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6936 let state = first.offline_state().expect("an offline state");
6937 assert!(!state.has_outlasted_the_queue());
6938 assert!(
6939 state
6940 .to_string()
6941 .contains("an outage longer than that loses"),
6942 "{state}"
6943 );
6944
6945 clock.advance_secs(24 * 60 * 60 + 60);
6947 let later = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6948 let state = later.offline_state().expect("an offline state");
6949 assert!(state.has_outlasted_the_queue());
6950 assert!(
6951 state.to_string().contains("queued work has been lost"),
6952 "{state}"
6953 );
6954 assert_eq!(launcher.launches(), 0, "and still nothing was started");
6955
6956 demand.set(PollOutcome::Ready(QueuedDemand::of(
6959 repo("acme/app"),
6960 jobs(0),
6961 )));
6962 let recovered = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6963 assert!(recovered.offline_state().is_none());
6964 assert_eq!(reconciler.schedule().offline_for(clock.now()), None);
6965
6966 demand.set(PollOutcome::Failed(RefreshState::Offline));
6967 let again = reconciler.reconcile(std::slice::from_ref(&policy)).await;
6968 assert!(
6969 !again
6970 .offline_state()
6971 .expect("an offline state")
6972 .has_outlasted_the_queue(),
6973 "a new outage must not inherit the age of the one before it"
6974 );
6975 }
6976
6977 #[tokio::test]
6993 async fn the_file_allocation_lock_excludes_a_second_holder_and_releases_on_drop() {
6994 let root = tempfile::tempdir().expect("a temporary directory");
6995 let paths = Arc::new(runner_manager_platform::paths::AppPaths::rooted_at(
6996 root.path(),
6997 ));
6998 let lock = FileAllocationLock::new(paths).with_wait(Duration::from_millis(50));
6999
7000 let held = lock.acquire().await.expect("an uncontended lock is free");
7001 assert!(
7002 matches!(lock.acquire().await, Err(AllocationLockBusy)),
7003 "a second holder was admitted; on Unix the lock is per open file description \
7004 and on Windows the share mode denies write, so this must be refused even \
7005 from inside the same process"
7006 );
7007
7008 drop(held);
7009 let regained = lock.acquire().await;
7010 assert!(
7011 regained.is_ok(),
7012 "dropping the guard is the only release there is, so a guard that does not \
7013 hold the `HostLock` leaves it held forever"
7014 );
7015 }
7016
7017 #[test]
7018 fn tee_events_reaches_both_sinks() {
7019 let left = Arc::new(EventLog::new());
7023 let right = Arc::new(EventLog::new());
7024 let tee = TeeEvents(
7025 Arc::clone(&left) as Arc<dyn EventSink>,
7026 Arc::clone(&right) as Arc<dyn EventSink>,
7027 );
7028
7029 tee.emit(LifecycleEvent::MonitorOnlySkipped {
7030 policy: PolicyId::from_u128(1),
7031 });
7032
7033 assert_eq!(left.count_of("monitor_only_skipped"), 1);
7034 assert_eq!(right.count_of("monitor_only_skipped"), 1);
7035 }
7036
7037 #[tokio::test]
7042 async fn the_repository_list_refreshes_far_more_slowly_than_the_demand_poll() {
7043 let clock = Arc::new(FakeClock::default());
7044 let directory = Arc::new(FakeDirectory::of(vec![repo("acme/one"), repo("acme/two")]));
7045 let cache = RepositoryCache::new(
7046 Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
7047 Arc::clone(&clock) as Arc<dyn Clock>,
7048 RefreshInterval::default(),
7049 );
7050 let target = ScaleTarget::organization("acme").unwrap();
7051
7052 assert_eq!(
7053 cache.ttl(),
7054 Duration::from_secs(60 * u64::from(REPOSITORY_LIST_REFRESH_MULTIPLE))
7055 );
7056
7057 for _ in 0..REPOSITORY_LIST_REFRESH_MULTIPLE {
7059 let scope = cache.scope_for(&target).await.unwrap();
7060 assert_eq!(scope.repositories().len(), 2);
7061 clock.advance_secs(60);
7062 }
7063 assert_eq!(
7064 directory.calls(),
7065 1,
7066 "re-listing an organization at demand-poll frequency is what exhausts the \
7067 shared request budget"
7068 );
7069 assert_eq!(cache.lookups(), 1);
7070
7071 cache.scope_for(&target).await.unwrap();
7073 assert_eq!(directory.calls(), 2);
7074 }
7075
7076 #[tokio::test]
7077 async fn a_repository_target_never_consults_the_directory() {
7078 let directory = Arc::new(FakeDirectory::of(vec![repo("acme/other")]));
7079 let cache = RepositoryCache::new(
7080 Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
7081 Arc::new(FakeClock::default()) as Arc<dyn Clock>,
7082 RefreshInterval::default(),
7083 );
7084 let target = ScaleTarget::repository("acme/app").unwrap();
7085
7086 let scope = cache.scope_for(&target).await.unwrap();
7087 assert_eq!(scope.repositories(), &[repo("acme/app")]);
7088 assert_eq!(directory.calls(), 0);
7089 }
7090
7091 #[tokio::test]
7092 async fn two_policies_on_one_target_cost_one_demand_poll_not_two() {
7093 let mut harness = Harness::simple(8, 4, "o/r");
7097 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7098 repo("o/r"),
7099 [
7100 fixtures::queued_job(&[HOST_LABEL]),
7101 fixtures::queued_job(&[HOST_LABEL]),
7102 fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
7103 fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
7104 ],
7105 )));
7106 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7107 isolated.activate().unwrap();
7108 let report = harness
7109 .reconciler
7110 .reconcile(&[policy(1, "o/r", 2), isolated])
7111 .await;
7112
7113 assert_eq!(harness.demand.polls().len(), 1);
7114 assert_eq!(report.targets_read, 1, "profiles share one target reading");
7115 assert_eq!(
7116 report.demand_requests,
7117 runner_manager_github::demand::DEMAND_REQUESTS_PER_REPOSITORY_PER_POLL,
7118 "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"
7119 );
7120 assert_eq!(
7121 report.started, 4,
7122 "native and isolated profiles each get two"
7123 );
7124 }
7125
7126 #[tokio::test]
7127 async fn subset_labels_without_selector_scale_only_the_compatible_local_profile() {
7128 use runner_manager_domain::model::{Arch, HostLabel, Os};
7129 use runner_manager_domain::policy::RoutingLabels;
7130 let mut harness = Harness::build(
7131 fixtures::host()
7132 .os(Os::MacOs)
7133 .architecture(Arch::Arm64)
7134 .capacity(10)
7135 .build(),
7136 Arc::new(FakeLauncher::new()),
7137 Arc::new(FakeDemand::ready(0, &repo("o/r"))),
7138 Arc::new(InProcessAllocationLock::new()),
7139 );
7140 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7141 repo("o/r"),
7142 [
7143 fixtures::queued_job(&["self-hosted", "macOS", "ARM64"]),
7144 fixtures::queued_job(&["SELF-HOSTED", "MACOS", "arm64"]),
7145 fixtures::queued_job(&["self-hosted", "macOS", "X64"]),
7146 fixtures::queued_job(&["self-hosted", "Windows", "ARM64"]),
7147 fixtures::queued_job(&["self-hosted", "macOS", "ARM64", "gpu"]),
7148 fixtures::queued_job(&["rm-other-osx-arm64", "macOS", "ARM64"]),
7149 ],
7150 )));
7151 let mut local = fixtures::policy()
7152 .id(PolicyId::from_u128(1))
7153 .repository("o/r")
7154 .autoscale("macmini", 10)
7155 .routing_labels(RoutingLabels::derive(
7156 &HostLabel::new("macmini").unwrap(),
7157 Os::MacOs,
7158 Arch::Arm64,
7159 ))
7160 .active()
7161 .build();
7162 for label in ["self-hosted", "macos", "arm64", "extra"] {
7163 local.add_routing_label(fixtures::label(label)).unwrap();
7164 }
7165 let mut foreign = local.to_persisted();
7166 foreign.id = PolicyId::from_u128(2);
7167 foreign.host_id = HostId::from_u128(0xdead);
7168 let foreign = ScalePolicy::from_persisted(foreign).unwrap();
7169 let report = harness.reconciler.reconcile(&[local, foreign]).await;
7170 assert_eq!(report.started, 2);
7171 assert_eq!(report.targets_read, 1);
7172 assert!(harness.events.events().iter().any(|event| matches!(
7173 event,
7174 LifecycleEvent::DemandObserved {
7175 demand: 2,
7176 not_matched: 4,
7177 ..
7178 }
7179 )));
7180 assert!(
7181 !harness
7182 .events
7183 .events()
7184 .iter()
7185 .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
7186 );
7187 }
7188
7189 #[tokio::test]
7190 async fn isolated_overlap_never_falls_back_to_native_in_either_state_or_order() {
7191 use runner_manager_domain::execution::{
7192 Backend, ExecutionPolicy, ImageReference, ResourceLimits,
7193 };
7194 let mut native = policy(1, "o/r", 2);
7195 native
7196 .add_routing_label(fixtures::label("self-hosted"))
7197 .unwrap();
7198 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7199 isolated
7200 .add_routing_label(fixtures::label("self-hosted"))
7201 .unwrap();
7202 isolated
7203 .set_execution_policy(ExecutionPolicy::Isolated {
7204 backend: Backend::Oci,
7205 image: ImageReference::new(format!(
7206 "registry.example/runner@sha256:{}",
7207 "a".repeat(64)
7208 ))
7209 .unwrap(),
7210 resources: ResourceLimits {
7211 cpu_millis: 1000,
7212 memory_mib: 512,
7213 disk_mib: 1024,
7214 },
7215 })
7216 .unwrap();
7217 assert!(!isolated.may_start_runners());
7218 let mut active_isolated = isolated.clone();
7219 active_isolated.activate().unwrap();
7220 for profiles in [
7221 [native.clone(), isolated.clone()],
7222 [isolated.clone(), native.clone()],
7223 [native.clone(), active_isolated.clone()],
7224 [active_isolated, native.clone()],
7225 ] {
7226 let mut harness = Harness::simple(4, 0, "o/r");
7227 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7228 repo("o/r"),
7229 [fixtures::queued_job(&["SELF-HOSTED"])],
7230 )));
7231 let report = harness.reconciler.reconcile(&profiles).await;
7232 assert_eq!(report.started, 0);
7233 assert!(harness.events.events().iter().any(|event| matches!(
7234 event,
7235 LifecycleEvent::RoutingRefused {
7236 reason: RoutingRefusal::OverlappingProfiles,
7237 count: 1,
7238 ..
7239 }
7240 )));
7241 }
7242 let mut harness = Harness::simple(4, 0, "o/r");
7245 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7246 repo("o/r"),
7247 [fixtures::queued_job(&["rm-home-win-x64-py-isolated"])],
7248 )));
7249 assert_eq!(
7250 harness
7251 .reconciler
7252 .reconcile(&[native, isolated])
7253 .await
7254 .started,
7255 0
7256 );
7257 }
7258
7259 #[tokio::test]
7260 async fn shared_optional_labels_do_not_duplicate_native_and_isolated_demand() {
7261 let mut harness = Harness::simple(2, 0, "o/r");
7262 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7263 repo("o/r"),
7264 [
7265 fixtures::queued_job(&[HOST_LABEL, "gpu"]),
7266 fixtures::queued_job(&["rm-home-win-x64-py-isolated", "gpu"]),
7267 fixtures::queued_job(&["gpu"]),
7268 fixtures::unresolvable_job(),
7269 ],
7270 )));
7271 let mut native = policy(1, "o/r", 2);
7272 native.add_routing_label(fixtures::label("gpu")).unwrap();
7273 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7274 isolated.add_routing_label(fixtures::label("gpu")).unwrap();
7275 isolated.activate().unwrap();
7276
7277 let report = harness.reconciler.reconcile(&[native, isolated]).await;
7278 assert_eq!(harness.demand.polls().len(), 1);
7279 assert_eq!(report.targets_read, 1);
7280 assert_eq!(report.started, 2);
7281 assert_eq!(report.allocations.len(), 2);
7282 assert!(
7283 report
7284 .allocations
7285 .iter()
7286 .all(|allocation| allocation.to_start == 1)
7287 );
7288 assert!(harness.events.events().iter().any(|event| matches!(
7289 event,
7290 LifecycleEvent::RoutingRefused {
7291 reason: RoutingRefusal::OverlappingProfiles,
7292 count: 1,
7293 ..
7294 }
7295 )));
7296 assert_eq!(
7297 harness
7298 .events
7299 .events()
7300 .iter()
7301 .filter(|event| matches!(
7302 event,
7303 LifecycleEvent::DemandObserved {
7304 unresolvable: 1,
7305 ..
7306 }
7307 ))
7308 .count(),
7309 2,
7310 "an expression stays unresolvable for both profiles"
7311 );
7312 }
7313
7314 #[tokio::test]
7315 async fn an_exclusion_starts_a_runner_registered_with_it_only_where_it_holds() {
7316 let queued = || {
7317 PollOutcome::Ready(QueuedDemand::of(
7318 repo("o/r"),
7319 [
7320 fixtures::queued_job(&[HOST_LABEL, "not-wsl", "not-gpu"]),
7321 fixtures::queued_job(&[HOST_LABEL]),
7322 ],
7323 ))
7324 };
7325
7326 let mut native = Harness::simple(4, 0, "o/r");
7329 native.reconciler = native.reconciler.with_host_traits(HostTraits::wsl(false));
7330 native.demand.set(queued());
7331 let report = native.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7332 assert_eq!(report.started, 2);
7333 for labels in native.launcher.registered.lock().unwrap().iter() {
7334 assert_eq!(
7335 labels,
7336 &vec![HOST_LABEL.to_string(), "not-gpu".into(), "not-wsl".into()]
7337 );
7338 }
7339
7340 native.demand.set(PollOutcome::Ready(QueuedDemand::of(
7343 repo("o/r"),
7344 fixtures::queued_jobs(&[HOST_LABEL], 4),
7345 )));
7346 native.launcher.registered.lock().unwrap().clear();
7347 native.reconciler.reconcile(&[policy(1, "o/r", 6)]).await;
7348 let later = native.launcher.registered.lock().unwrap().clone();
7349 assert!(!later.is_empty());
7350 assert!(
7351 later
7352 .iter()
7353 .all(|labels| labels.contains(&"not-gpu".to_string()))
7354 );
7355
7356 let mut wsl = Harness::simple(4, 0, "o/r");
7358 wsl.reconciler = wsl.reconciler.with_host_traits(HostTraits::wsl(true));
7359 wsl.demand.set(queued());
7360 let report = wsl.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7361 assert_eq!(report.started, 1);
7362 assert_eq!(
7363 wsl.launcher.registered.lock().unwrap().as_slice(),
7364 [vec![HOST_LABEL.to_string(), "wsl".into()]]
7365 );
7366
7367 let mut gpu = Harness::simple(4, 0, "o/r");
7369 gpu.demand.set(queued());
7370 let mut carrying = policy(1, "o/r", 4);
7371 carrying.add_routing_label(fixtures::label("gpu")).unwrap();
7372 let report = gpu.reconciler.reconcile(&[carrying]).await;
7373 assert_eq!(report.started, 1);
7374 assert_eq!(
7375 gpu.launcher.registered.lock().unwrap().as_slice(),
7376 [vec![HOST_LABEL.to_string(), "gpu".into()]]
7377 );
7378 }
7379
7380 #[tokio::test]
7381 async fn without_exclusions_or_traits_a_runner_registers_exactly_its_profile_labels() {
7382 let mut harness = Harness::simple(4, 2, "o/r");
7383 let report = harness.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7384 assert_eq!(report.started, 2);
7385 assert_eq!(
7386 harness.launcher.registered.lock().unwrap().as_slice(),
7387 [vec![HOST_LABEL.to_string()], vec![HOST_LABEL.to_string()]]
7388 );
7389 }
7390
7391 #[tokio::test]
7392 async fn a_job_no_runner_can_satisfy_is_refused_by_name_not_counted() {
7393 let mut harness = Harness::simple(4, 0, "o/r");
7394 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7395 repo("o/r"),
7396 [
7397 fixtures::queued_job(&[HOST_LABEL, "not-"]),
7398 fixtures::queued_job(&["rm-other-linux-x64", "gpu", "not-gpu"]),
7400 ],
7401 )));
7402 let report = harness.reconciler.reconcile(&[policy(1, "o/r", 4)]).await;
7403 assert_eq!(report.started, 0);
7404 assert!(harness.events.events().iter().any(|event| matches!(
7405 event,
7406 LifecycleEvent::RoutingRefused {
7407 reason: RoutingRefusal::UnsatisfiableExclusion,
7408 count: 1,
7409 ..
7410 }
7411 )));
7412 assert!(harness.events.events().iter().any(|event| matches!(
7413 event,
7414 LifecycleEvent::DemandObserved {
7415 demand: 0,
7416 unresolvable: 2,
7417 ..
7418 }
7419 )));
7420 }
7421
7422 #[tokio::test]
7423 async fn missing_required_selector_labels_do_not_match_and_overlaps_show_remedies() {
7424 let mut harness = Harness::simple(4, 0, "o/r");
7425 harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
7426 repo("o/r"),
7427 [fixtures::queued_job(&[
7428 HOST_LABEL,
7429 "rm-home-win-x64-py-isolated",
7430 ])],
7431 )));
7432 let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7433 isolated.activate().unwrap();
7434 let report = harness
7435 .reconciler
7436 .reconcile(&[policy(1, "o/r", 2), isolated])
7437 .await;
7438 assert_eq!(report.started, 0);
7439 assert_eq!(
7440 harness
7441 .events
7442 .events()
7443 .iter()
7444 .filter(|event| matches!(
7445 event,
7446 LifecycleEvent::DemandObserved {
7447 demand: 0,
7448 not_matched: 1,
7449 ..
7450 }
7451 ))
7452 .count(),
7453 2
7454 );
7455
7456 let mut superset = policy(1, "o/r", 2);
7460 assert!(
7461 superset
7462 .add_routing_label(fixtures::label("rm-home-win-x64-py-isolated"))
7463 .unwrap()
7464 );
7465 let mut sibling = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
7466 sibling.activate().unwrap();
7467 let mut unique = Harness::simple(4, 0, "o/r");
7468 unique.demand.set(PollOutcome::Ready(QueuedDemand::of(
7469 repo("o/r"),
7470 [fixtures::queued_job(&[
7471 HOST_LABEL,
7472 "rm-home-win-x64-py-isolated",
7473 ])],
7474 )));
7475 let report = unique.reconciler.reconcile(&[superset, sibling]).await;
7476 assert_eq!(report.started, 1);
7477 assert_eq!(
7478 report
7479 .allocations
7480 .iter()
7481 .find(|allocation| allocation.to_start == 1)
7482 .unwrap()
7483 .policy_id,
7484 PolicyId::from_u128(1)
7485 );
7486 assert!(
7487 !unique
7488 .events
7489 .events()
7490 .iter()
7491 .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
7492 );
7493
7494 let mut overlap = Harness::simple(4, 1, "o/r");
7495 let report = overlap
7496 .reconciler
7497 .reconcile(&[policy(1, "o/r", 2), policy(2, "o/r", 2)])
7498 .await;
7499 assert_eq!(report.started, 0);
7500 assert!(overlap.events.events().iter().any(|event| matches!(
7501 event,
7502 LifecycleEvent::RoutingRefused {
7503 reason: RoutingRefusal::OverlappingProfiles,
7504 count: 1,
7505 ..
7506 }
7507 )));
7508 assert!(
7509 RoutingRefusal::OverlappingProfiles
7510 .to_string()
7511 .contains("repair")
7512 );
7513 }
7514
7515 #[tokio::test]
7520 async fn a_failed_launch_stops_the_run_and_is_reported_without_free_text() {
7521 let launcher = Arc::new(FakeLauncher::new());
7522 launcher.fail_next(FailureReason::Other("token ghp_0123456789abcdef".into()));
7523 let mut harness = Harness::build(
7524 host_with(4),
7525 Arc::clone(&launcher),
7526 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
7527 Arc::new(InProcessAllocationLock::new()),
7528 );
7529
7530 let report = harness
7531 .reconciler
7532 .reconcile(&[policy(1, "acme/app", 4)])
7533 .await;
7534 assert_eq!(report.started, 0);
7535 assert_eq!(report.allocations[0].to_start, 3, "the decision stands");
7536
7537 let failures: Vec<&'static str> = harness
7538 .events
7539 .events()
7540 .into_iter()
7541 .filter_map(|event| match event {
7542 LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
7543 _ => None,
7544 })
7545 .collect();
7546 assert_eq!(failures, vec!["other"]);
7547 assert!(
7548 !failures[0].contains("ghp_"),
7549 "an event carried a `FailureReason::Other` detail verbatim"
7550 );
7551 }
7552
7553 #[tokio::test]
7558 async fn an_unfit_host_registers_no_runner_and_reports_host_unfit() {
7559 let launcher = Arc::new(FakeLauncher::new());
7560 let unfit = Arc::new(AtomicBool::new(true));
7561 let events = Arc::new(EventLog::new());
7562 let mut reconciler = Reconciler::new(
7563 host_with(4),
7564 ReconcilerPorts {
7565 demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
7566 launcher: Arc::new(FitnessGatedLauncher::new(
7567 Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
7568 Arc::clone(&unfit),
7569 )),
7570 lock: Arc::new(InProcessAllocationLock::new()),
7571 directory: Arc::new(FakeDirectory::default()),
7572 clock: Arc::new(FakeClock::default()) as Arc<dyn Clock>,
7573 jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
7574 events: Arc::clone(&events) as Arc<dyn EventSink>,
7575 },
7576 );
7577 let policies = [policy(1, "acme/app", 4)];
7578
7579 let report = reconciler.reconcile(&policies).await;
7580 assert_eq!(report.started, 0);
7581 assert_eq!(launcher.launches(), 0, "the inner launcher must not run");
7582 let reasons: Vec<&'static str> = events
7583 .events()
7584 .into_iter()
7585 .filter_map(|event| match event {
7586 LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
7587 _ => None,
7588 })
7589 .collect();
7590 assert_eq!(reasons, ["host_unfit"]);
7591
7592 unfit.store(false, Ordering::Release);
7593 let report = reconciler.reconcile(&policies).await;
7594 assert_eq!(report.started, 2);
7595 assert_eq!(launcher.launches(), 2);
7596 }
7597
7598 #[tokio::test]
7599 async fn a_held_allocation_lock_starts_nothing_and_says_so() {
7600 let launcher = Arc::new(FakeLauncher::new());
7601 let mut harness = Harness::build(
7602 host_with(4),
7603 Arc::clone(&launcher),
7604 Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
7605 Arc::new(HeldLock),
7606 );
7607
7608 let report = harness
7609 .reconciler
7610 .reconcile(&[policy(1, "acme/app", 4)])
7611 .await;
7612 assert_eq!(report.started, 0);
7613 assert_eq!(
7614 report.deferred, 3,
7615 "three runners were granted and none was created; `deferred` counts grants, \
7616 not policies -- it reported `1` when a policy that launched two of five and \
7617 then lost the lock had left three unstarted"
7618 );
7619 assert_eq!(launcher.launches(), 0);
7620 assert_eq!(harness.events.count_of("allocation_deferred"), 1);
7621 assert!(
7622 harness
7623 .events
7624 .events()
7625 .iter()
7626 .any(|event| matches!(event, LifecycleEvent::AllocationDeferred { count: 3, .. })),
7627 "the event carries the same number the report does"
7628 );
7629 assert_eq!(
7630 report.allocations.len(),
7631 1,
7632 "the intent is still reported, so an operator staring at a queue sees why \
7633 nothing started"
7634 );
7635 }
7636
7637 #[tokio::test]
7638 async fn a_foreign_or_draining_policy_is_reported_by_name_and_polls_nothing() {
7639 let mut harness = Harness::simple(8, 5, "acme/app");
7640
7641 let foreign = fixtures::policy()
7642 .id(PolicyId::from_u128(1))
7643 .repository("acme/app")
7644 .host(HostId::from_u128(0xdead))
7645 .autoscale("office", 4)
7646 .active()
7647 .build();
7648 let mut draining = policy(2, "acme/app", 4);
7649 draining.request_disable().unwrap();
7650
7651 let report = harness.reconciler.reconcile(&[foreign, draining]).await;
7652
7653 assert_eq!(report.started, 0);
7654 assert_eq!(
7655 harness.demand.polls().len(),
7656 0,
7657 "neither can act on an answer"
7658 );
7659 let factors: Vec<LimitingFactor> = report
7660 .allocations
7661 .iter()
7662 .map(|a| a.limiting_factor)
7663 .collect();
7664 assert!(factors.contains(&LimitingFactor::ForeignHost));
7665 assert!(factors.contains(&LimitingFactor::NotReconciling));
7666 }
7667
7668 #[tokio::test]
7669 async fn an_unreadable_repository_list_makes_the_target_unreadable_not_empty() {
7670 #[derive(Debug)]
7673 struct BrokenDirectory;
7674
7675 #[async_trait::async_trait]
7676 impl RepositoryDirectory for BrokenDirectory {
7677 async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
7678 Err(InventoryError::Cancelled)
7679 }
7680 }
7681
7682 let launcher = Arc::new(FakeLauncher::new());
7683 let events = Arc::new(EventLog::new());
7684 let mut reconciler = Reconciler::new(
7685 host_with(4),
7686 ReconcilerPorts {
7687 demand: Arc::new(FakeDemand::default()),
7688 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
7689 lock: Arc::new(InProcessAllocationLock::new()),
7690 directory: Arc::new(BrokenDirectory),
7691 clock: Arc::new(FakeClock::default()),
7692 jitter: Arc::new(NoJitter),
7693 events: Arc::clone(&events) as Arc<dyn EventSink>,
7694 },
7695 );
7696
7697 let org_policy = fixtures::policy()
7698 .id(PolicyId::from_u128(1))
7699 .organization("acme")
7700 .autoscale("home", 4)
7701 .active()
7702 .build();
7703
7704 let report = reconciler.reconcile(&[org_policy]).await;
7705 assert_eq!(report.started, 0);
7706 assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
7707 assert_eq!(events.count_of("target_unreadable"), 1);
7708 }
7709
7710 fn every_event() -> Vec<LifecycleEvent> {
7722 let policy = PolicyId::from_u128(0xabcd_ef01);
7723 let attempt = AttemptId::from_u128(0x1234_5678);
7724 vec![
7725 LifecycleEvent::DemandObserved {
7726 policy,
7727 demand: u32::MAX,
7728 not_matched: u32::MAX,
7729 unresolvable: u32::MAX,
7730 complete: false,
7731 },
7732 LifecycleEvent::TargetUnreadable {
7733 policy,
7734 reason: unreadable_reason(&RefreshState::Failed {
7735 status: Some(500),
7736 message: "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz"
7737 .into(),
7738 }),
7739 },
7740 LifecycleEvent::Allocated {
7741 policy,
7742 demand: u32::MAX,
7743 desired: u16::MAX,
7744 active_owned: 7,
7745 headroom: 9,
7746 to_start: 2,
7747 limiting: LimitingFactor::HostCapacity,
7748 },
7749 LifecycleEvent::MonitorOnlySkipped { policy },
7750 LifecycleEvent::RunnerStarted { policy, attempt },
7751 LifecycleEvent::RunnerStartFailed {
7752 policy,
7753 reason: failure_reason_kind(&FailureReason::Other(
7754 "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
7755 )),
7756 },
7757 LifecycleEvent::AllocationDeferred { policy, count: 4 },
7758 LifecycleEvent::AttemptsUnreadable {
7759 reason: failure_reason_kind(&FailureReason::Other(
7760 "x-api-key: ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
7761 )),
7762 },
7763 LifecycleEvent::HostCapacityUnreadable {
7764 reason: HostCapacityError::MissingHost,
7765 },
7766 LifecycleEvent::HostCapacityUnreadable {
7767 reason: HostCapacityError::JournalUnreadable,
7768 },
7769 LifecycleEvent::AttemptCleanFailed {
7770 policy,
7771 attempt,
7772 reason: failure_reason_kind(&FailureReason::ProcessExitedUnexpectedly),
7773 },
7774 LifecycleEvent::AttemptCleaned {
7775 policy,
7776 attempt,
7777 outcome: OutcomeKind::IdleExit,
7778 },
7779 LifecycleEvent::ScaleDownRefused { policy, attempt },
7780 LifecycleEvent::PollScheduled {
7781 retry_in_ms: 900_000,
7782 pace: PollPace::RateLimited {
7783 kind: RateLimitKind::Primary,
7784 },
7785 },
7786 ]
7787 }
7788
7789 #[test]
7798 fn no_emitted_event_can_carry_a_credential() {
7799 use runner_manager_platform::logging::redact;
7800
7801 for event in every_event() {
7802 let displayed = event.to_string();
7803 assert_eq!(
7804 redact(&displayed),
7805 displayed,
7806 "`{}` renders something `d1`'s sink would have to redact",
7807 event.name()
7808 );
7809
7810 let debugged = format!("{event:?}");
7811 assert_eq!(
7812 redact(&debugged),
7813 debugged,
7814 "`{}`'s Debug renders something `d1`'s sink would have to redact",
7815 event.name()
7816 );
7817 }
7818
7819 let secret = "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz";
7822 assert_ne!(
7823 redact(secret),
7824 secret,
7825 "the scan above proves nothing if `redact` no longer recognises a credential"
7826 );
7827 }
7828
7829 #[test]
7830 fn every_field_name_this_sink_emits_is_one_d1_allows() {
7831 use runner_manager_platform::logging::is_field_allowed;
7832
7833 for field in [
7837 "event",
7838 "policy_id",
7839 "attempt_id",
7840 "attempt_state",
7841 "demand",
7842 "desired",
7843 "capacity",
7844 "headroom",
7845 "count",
7846 "reason",
7847 "outcome",
7848 "mode",
7849 "lock",
7850 "retry_in_ms",
7851 "state",
7852 ] {
7853 assert!(
7854 is_field_allowed(field),
7855 "`{field}` is not on `d1`'s allow-list, so this sink would emit \
7856 `[redacted]` in its place and the line would lose its meaning"
7857 );
7858 }
7859 }
7860
7861 #[test]
7862 fn every_failure_reason_has_a_credential_free_kind() {
7863 for reason in FailureReason::ALL {
7864 let kind = failure_reason_kind(&reason);
7865 assert!(!kind.is_empty());
7866 assert!(
7867 kind.chars().all(|c| c.is_ascii_lowercase() || c == '_'),
7868 "`{kind}` is not a fixed identifier"
7869 );
7870 }
7871 assert_eq!(
7872 failure_reason_kind(&FailureReason::Other("ghp_secret".into())),
7873 "other",
7874 "the detail of an `Other` reason never reaches an event"
7875 );
7876 let categories = [
7877 IsolationProviderFailure::NotInstalled,
7878 IsolationProviderFailure::PermissionDenied,
7879 IsolationProviderFailure::ImageUnavailableOrIncompatible,
7880 IsolationProviderFailure::DiskQuotaUnavailable,
7881 IsolationProviderFailure::RuntimeOperationFailed,
7882 ];
7883 let mut kinds = std::collections::BTreeSet::new();
7884 for category in categories {
7885 let kind = failure_reason_kind(&FailureReason::IsolationProvider(category));
7886 assert_ne!(kind, "other");
7887 assert!(!kind.contains("secret"));
7888 kinds.insert(kind);
7889 }
7890 assert_eq!(kinds.len(), categories.len());
7891 }
7892
7893 fn production_half_of(source: &str) -> String {
7908 let production = source
7909 .split_once("\n#[cfg(test)]")
7910 .map_or(source, |(production, _)| production);
7911 production
7912 .lines()
7913 .filter(|line| !line.trim_start().starts_with("//"))
7914 .collect::<Vec<_>>()
7915 .join("\n")
7916 }
7917
7918 fn this_file_above_its_tests_without_prose() -> String {
7920 production_half_of(include_str!("reconcile.rs"))
7921 }
7922
7923 fn normalise_for_scan(text: &str) -> String {
7942 text.to_ascii_lowercase().replace(['_', ' '], "")
7943 }
7944
7945 const FORBIDDEN: &[&str] = &[
7955 concat!("fn ", "acquire", "_job"),
7956 concat!("fn ", "claim", "_job"),
7957 concat!("fn ", "lease", "_job"),
7958 concat!("fn ", "reserve", "_job"),
7959 concat!("fn ", "ack", "nowledge"),
7960 concat!("struct ", "Job", "Lease"),
7961 concat!("struct ", "Job", "Claim"),
7962 concat!("struct ", "Job", "Reservation"),
7963 ];
7964
7965 fn forbidden_shape_in(source: &str) -> Option<&'static str> {
7966 let haystack = normalise_for_scan(source);
7967 FORBIDDEN
7968 .iter()
7969 .copied()
7970 .find(|forbidden| haystack.contains(&normalise_for_scan(forbidden)))
7971 }
7972
7973 #[test]
8007 fn nothing_in_this_crate_reserves_or_claims_a_job() {
8008 const SOURCES: &[(&str, &str)] = &[
8009 ("dependency_caches.rs", include_str!("dependency_caches.rs")),
8010 ("docker_cleanup.rs", include_str!("docker_cleanup.rs")),
8011 (
8012 "docker_cleanup/http.rs",
8013 include_str!("docker_cleanup/http.rs"),
8014 ),
8015 (
8016 "docker_cleanup/policy.rs",
8017 include_str!("docker_cleanup/policy.rs"),
8018 ),
8019 (
8020 "docker_cleanup/proxy.rs",
8021 include_str!("docker_cleanup/proxy.rs"),
8022 ),
8023 (
8024 "docker_cleanup/tests.rs",
8025 include_str!("docker_cleanup/tests.rs"),
8026 ),
8027 ("lib.rs", include_str!("lib.rs")),
8028 ("lifecycle.rs", include_str!("lifecycle.rs")),
8029 (
8030 "lifecycle/windows_hyperv.rs",
8031 include_str!("lifecycle/windows_hyperv.rs"),
8032 ),
8033 ("oci.rs", include_str!("oci.rs")),
8034 ("package.rs", include_str!("package.rs")),
8035 ("reconcile.rs", include_str!("reconcile.rs")),
8036 ];
8037
8038 fn walk(directory: &std::path::Path, prefix: &str, found: &mut Vec<String>) {
8039 for entry in std::fs::read_dir(directory).expect("the crate's own src/ is readable") {
8040 let entry = entry.expect("a readable directory entry");
8041 let name = entry.file_name().to_string_lossy().into_owned();
8042 let joined = if prefix.is_empty() {
8045 name.clone()
8046 } else {
8047 format!("{prefix}/{name}")
8048 };
8049 if entry.path().is_dir() {
8050 walk(&entry.path(), &joined, found);
8051 } else if name.ends_with(".rs") {
8052 found.push(joined);
8053 }
8054 }
8055 }
8056
8057 let mut listed: Vec<&str> = SOURCES.iter().map(|(name, _)| *name).collect();
8058 listed.sort_unstable();
8059 let mut on_disk = Vec::new();
8060 walk(
8061 std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/src")),
8062 "",
8063 &mut on_disk,
8064 );
8065 on_disk.sort_unstable();
8066 assert_eq!(
8067 listed, on_disk,
8068 "a source file was added or removed; this scan claims to cover the whole crate \
8069 and a stale list makes that claim false"
8070 );
8071
8072 for (name, source) in SOURCES {
8073 assert_eq!(
8074 forbidden_shape_in(&production_half_of(source)),
8075 None,
8076 "{name} names a forbidden shape: there is no `AcquireJobs` equivalent over \
8077 REST, and a local lease coordinates this host with itself and with nothing \
8078 else. If an owner decision restored one, that decision belongs in this \
8079 module's documentation and in this test before it belongs in the code"
8080 );
8081 }
8082
8083 assert!(
8086 forbidden_shape_in("async fn acquire_jobs(&self) -> Vec<Job> { todo!() }").is_some(),
8087 "the scan above proves nothing if the needles no longer match"
8088 );
8089 }
8090
8091 #[test]
8119 fn the_label_predicate_is_b1s_and_this_module_only_applies_it() {
8120 let production = this_file_above_its_tests_without_prose();
8121
8122 assert!(
8123 production.contains("DemandTally"),
8124 "the reconciliation loop must tally queued jobs against this policy's routing \
8125 labels. A production half that named `DemandTally` nowhere would mean the \
8126 label filtering had been removed, and a repository whose jobs target \
8127 `ubuntu-latest` would drive its policy toward `max_capacity` again"
8128 );
8129
8130 for second_implementation in ["RunsOnMatch", "UnresolvableRunsOn"] {
8131 assert!(
8132 !production.contains(second_implementation),
8133 "the reconciliation loop names `{second_implementation}`, which is the \
8134 vocabulary of deciding one job's `runs-on` -- and `RoutingLabels::tally` \
8135 has already decided it. This module applies the predicate and does not \
8136 re-implement it; if an owner decision changed that, it belongs in this \
8137 module's documentation and in this test before it belongs in the code"
8138 );
8139 }
8140 }
8141
8142 #[test]
8150 fn demand_is_the_queued_jobs_this_policy_can_actually_serve() {
8151 let policy = policy(1, "acme/app", 10);
8152 let reading = QueuedDemand::of(
8153 repo("acme/app"),
8154 [
8155 fixtures::queued_job(&[HOST_LABEL]),
8156 fixtures::queued_job(&[HOST_LABEL]),
8157 fixtures::queued_job(&["ubuntu-latest"]),
8158 fixtures::unresolvable_job(),
8159 ],
8160 );
8161
8162 let tally = demand_for(&policy, &reading, &[]);
8163
8164 assert_eq!(
8165 tally.demand(),
8166 2,
8167 "only the jobs whose required labels this policy carries are demand"
8168 );
8169 assert_eq!(
8170 tally.not_matched, 1,
8171 "a `ubuntu-latest` job is somebody else's work; counting it would start a \
8172 runner that idles until it times out"
8173 );
8174 assert_eq!(
8175 tally.unresolvable.len(),
8176 1,
8177 "an unresolvable `runs-on` is never demand and never discarded"
8178 );
8179 }
8180
8181 #[test]
8184 fn an_organization_policy_tallies_every_repository_its_scope_covers() {
8185 let mut per_repository = BTreeMap::new();
8186 per_repository.insert(repo("acme/left"), fixtures::queued_jobs(&[HOST_LABEL], 3));
8187 per_repository.insert(repo("acme/right"), fixtures::queued_jobs(&[HOST_LABEL], 4));
8188 let reading = QueuedDemand::new(per_repository);
8189
8190 let repository_policy = policy(1, "acme/left", 10);
8191 assert_eq!(
8192 demand_for(&repository_policy, &reading, &[]).demand(),
8193 3,
8194 "a repository target reads its own repository's queue and not the aggregate"
8195 );
8196
8197 let org_policy = fixtures::policy()
8198 .id(PolicyId::from_u128(2))
8199 .organization("acme")
8200 .autoscale("home", 10)
8201 .active()
8202 .build();
8203 assert_eq!(
8204 demand_for(&org_policy, &reading, &[]).demand(),
8205 7,
8206 "an organization policy serves any repository in its scope, so its demand is \
8207 the whole aggregate's"
8208 );
8209 }
8210
8211 #[test]
8212 fn the_accepted_over_count_is_bounded_by_the_two_ceilings_and_nothing_else() {
8213 let host = host_with(2);
8218 let policy = policy(1, "acme/app", 5);
8219 let attempts: Vec<RunnerAttempt> = Vec::new();
8220 let mut allocator = HostAllocator::from_attempts(&host, &attempts);
8221
8222 let allocation = allocator.allocate(&policy, u32::MAX);
8223 assert_eq!(allocation.desired, 5, "max_capacity beats reported demand");
8224 assert_eq!(allocation.to_start, 2, "host_capacity beats max_capacity");
8225 assert_eq!(allocation.limiting_factor, LimitingFactor::HostCapacity);
8226 }
8227
8228 #[derive(Debug, Default)]
8231 struct FixedDirector {
8232 directed: Mutex<BTreeMap<PolicyId, u32>>,
8233 started: Mutex<Vec<(PolicyId, AttemptId)>>,
8234 failed: Mutex<Vec<(PolicyId, Option<AttemptId>)>>,
8235 older: Mutex<std::collections::VecDeque<usize>>,
8237 passed_over: AtomicUsize,
8238 orders: Mutex<BTreeMap<PolicyId, u64>>,
8240 places_plain: AtomicBool,
8242 }
8243
8244 impl DemandDirector for FixedDirector {
8245 fn directed_demand(&self, policy: &ScalePolicy) -> Option<u32> {
8246 self.directed.lock().unwrap().get(&policy.id).copied()
8247 }
8248
8249 fn launched(&self, policy: &ScalePolicy, attempt: AttemptId) {
8250 self.started.lock().unwrap().push((policy.id, attempt));
8251 }
8252
8253 fn start_failed(&self, policy: &ScalePolicy, attempt: Option<AttemptId>) {
8254 self.failed.lock().unwrap().push((policy.id, attempt));
8255 }
8256
8257 fn next_waiting(&self, policy: &ScalePolicy) -> Option<u64> {
8258 self.orders.lock().unwrap().get(&policy.id).copied()
8259 }
8260
8261 fn places_plain_jobs(&self, _policy: &ScalePolicy) -> bool {
8262 self.places_plain.load(Ordering::SeqCst)
8263 }
8264
8265 fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
8266 let mut older = self.older.lock().unwrap();
8267 if older.len() > 1 {
8268 older.pop_front().unwrap_or(0)
8269 } else {
8270 older.front().copied().unwrap_or(0)
8271 }
8272 }
8273
8274 fn pass_over_older(&self, _policy: &ScalePolicy) {
8275 self.passed_over.fetch_add(1, Ordering::SeqCst);
8276 self.older.lock().unwrap().clear();
8277 }
8278 }
8279
8280 #[derive(Debug)]
8283 struct HeldBringUp {
8284 inner: Arc<FakeLauncher>,
8285 barrier: Arc<tokio::sync::Barrier>,
8286 fail: Option<FailureReason>,
8287 waits: Mutex<Vec<ReservationWaits>>,
8288 }
8289
8290 #[async_trait::async_trait]
8291 impl RunnerLauncher for HeldBringUp {
8292 async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
8293 self.inner.host_capacity(host).await
8294 }
8295
8296 async fn supervise(
8297 &self,
8298 policy: &ScalePolicy,
8299 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
8300 self.inner.supervise(policy).await
8301 }
8302
8303 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
8304 self.inner.attempts().await
8305 }
8306
8307 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
8308 self.inner.launch(request).await
8309 }
8310
8311 async fn finish_launch(
8312 &self,
8313 _policy: &ScalePolicy,
8314 attempt: RunnerAttempt,
8315 waits: ReservationWaits,
8316 ) -> Result<RunnerAttempt, LaunchFailure> {
8317 self.waits.lock().unwrap().push(waits);
8318 self.barrier.wait().await;
8319 match &self.fail {
8320 Some(reason) => Err(LaunchFailure::new(reason.clone())),
8321 None => Ok(attempt),
8322 }
8323 }
8324
8325 async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
8326 self.inner.clean(attempt).await
8327 }
8328 }
8329
8330 fn directed_reconciler(
8331 capacity: u16,
8332 launcher: Arc<dyn RunnerLauncher>,
8333 lock: Arc<dyn AllocationLock>,
8334 director: Arc<FixedDirector>,
8335 ) -> Reconciler {
8336 Reconciler::new(
8337 host_with(capacity),
8338 ReconcilerPorts {
8339 demand: Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
8340 launcher,
8341 lock,
8342 directory: Arc::new(FakeDirectory::default()),
8343 clock: Arc::new(FakeClock::default()),
8344 jitter: Arc::new(NoJitter),
8345 events: Arc::new(NoEvents),
8346 },
8347 )
8348 .with_director(director as Arc<dyn DemandDirector>)
8349 }
8350
8351 #[tokio::test]
8354 async fn two_targets_bring_up_their_runners_at_once() {
8355 let fake = Arc::new(FakeLauncher::new());
8356 let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
8357 let barrier = Arc::new(tokio::sync::Barrier::new(2));
8358 let held = Arc::new(HeldBringUp {
8359 inner: Arc::clone(&fake),
8360 barrier,
8361 fail: None,
8362 waits: Mutex::default(),
8363 });
8364 let director = Arc::new(FixedDirector::default());
8365 let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
8366 director.directed.lock().unwrap().insert(one.id, 1);
8367 director.directed.lock().unwrap().insert(two.id, 1);
8368 let first = directed_reconciler(
8369 2,
8370 Arc::clone(&held) as _,
8371 Arc::clone(&lock),
8372 Arc::clone(&director),
8373 );
8374 let second = directed_reconciler(2, held as _, lock, director);
8375
8376 let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
8377 tokio::join!(
8378 first.start_directed(std::slice::from_ref(&one)),
8379 second.start_directed(std::slice::from_ref(&two)),
8380 )
8381 })
8382 .await
8383 .expect("neither waits for the other's bring-up");
8384 assert_eq!((a.started, b.started), (1, 1));
8385 }
8386
8387 #[tokio::test]
8390 async fn a_runner_coming_up_holds_its_slot_against_another_target() {
8391 let fake = Arc::new(FakeLauncher::new());
8392 let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
8393 let barrier = Arc::new(tokio::sync::Barrier::new(2));
8394 let held = Arc::new(HeldBringUp {
8395 inner: Arc::clone(&fake),
8396 barrier: Arc::clone(&barrier),
8397 fail: None,
8398 waits: Mutex::default(),
8399 });
8400 let director = Arc::new(FixedDirector::default());
8401 let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
8402 director.directed.lock().unwrap().insert(one.id, 1);
8403 director.directed.lock().unwrap().insert(two.id, 1);
8404 let first = directed_reconciler(
8405 1,
8406 Arc::clone(&held) as _,
8407 Arc::clone(&lock),
8408 Arc::clone(&director),
8409 );
8410 let second = directed_reconciler(1, held as _, lock, director);
8411
8412 let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
8413 tokio::join!(first.start_directed(std::slice::from_ref(&one)), async {
8414 while fake.live_count() == 0 {
8415 tokio::task::yield_now().await;
8416 }
8417 let report = second.start_directed(std::slice::from_ref(&two)).await;
8418 barrier.wait().await;
8419 report
8420 })
8421 })
8422 .await
8423 .expect("the first is let go");
8424 assert_eq!((a.started, b.started), (1, 0));
8425 assert_eq!(fake.live_count(), 1);
8426 }
8427
8428 #[tokio::test]
8431 async fn a_failed_bring_up_is_a_failed_start() {
8432 let fake = Arc::new(FakeLauncher::new());
8433 let held = Arc::new(HeldBringUp {
8434 inner: fake,
8435 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8436 fail: Some(FailureReason::JitRequestFailed),
8437 waits: Mutex::default(),
8438 });
8439 let director = Arc::new(FixedDirector::default());
8440 let pool = policy(1, "acme/app", 2);
8441 director.directed.lock().unwrap().insert(pool.id, 1);
8442 let reconciler = directed_reconciler(
8443 2,
8444 held as _,
8445 Arc::new(InProcessAllocationLock::new()),
8446 Arc::clone(&director),
8447 );
8448
8449 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8450
8451 assert_eq!(report.started, 0);
8452 let started = director.started.lock().unwrap()[0].1;
8453 assert_eq!(
8454 *director.failed.lock().unwrap(),
8455 vec![(pool.id, Some(started))]
8456 );
8457 }
8458
8459 fn scripted_older(director: &FixedDirector, looks: &[(usize, usize)]) {
8460 *director.older.lock().unwrap() = looks
8461 .iter()
8462 .flat_map(|(value, count)| std::iter::repeat_n(*value, *count))
8463 .collect();
8464 }
8465
8466 #[tokio::test(start_paused = true)]
8470 async fn a_younger_launch_reserves_after_older_ones_while_they_make_progress() {
8471 let held = Arc::new(HeldBringUp {
8472 inner: Arc::new(FakeLauncher::new()),
8473 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8474 fail: None,
8475 waits: Mutex::default(),
8476 });
8477 let director = Arc::new(FixedDirector::default());
8478 let pool = policy(1, "acme/app", 3);
8479 director.directed.lock().unwrap().insert(pool.id, 1);
8480 let reconciler = directed_reconciler(
8481 5,
8482 Arc::clone(&held) as _,
8483 Arc::new(InProcessAllocationLock::new()),
8484 Arc::clone(&director),
8485 );
8486
8487 let looks =
8490 (OLDER_LAUNCH_TURN.as_millis() * 3 / 4 / OLDER_LAUNCH_POLL.as_millis()) as usize;
8491 scripted_older(&director, &[(3, looks), (2, looks), (1, looks), (0, 1)]);
8492 let began = tokio::time::Instant::now();
8493 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8494 assert_eq!(report.started, 1);
8495 assert_eq!(
8496 began.elapsed(),
8497 OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap()
8498 );
8499 assert_eq!(director.passed_over.load(Ordering::SeqCst), 0);
8500 assert_eq!(
8501 held.waits.lock().unwrap()[0].order,
8502 OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap(),
8503 "the wait is in the launch's timings"
8504 );
8505 }
8506
8507 #[tokio::test]
8510 async fn a_pass_hands_a_directed_launch_to_the_launch_side() {
8511 let fake = Arc::new(FakeLauncher::new());
8512 let director = Arc::new(FixedDirector::default());
8513 let pool = policy(1, "acme/app", 3);
8514 director.directed.lock().unwrap().insert(pool.id, 1);
8515 let signal = Arc::new(tokio::sync::Notify::new());
8516 let mut reconciler = directed_reconciler(
8517 5,
8518 Arc::clone(&fake) as _,
8519 Arc::new(InProcessAllocationLock::new()),
8520 Arc::clone(&director),
8521 )
8522 .with_launch_side(Arc::clone(&signal));
8523
8524 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8525
8526 assert_eq!(report.started, 0);
8527 assert_eq!(fake.live_count(), 0);
8528 assert_eq!(report.allocations[0].to_start, 1);
8529 tokio::time::timeout(Duration::from_secs(1), signal.notified())
8530 .await
8531 .expect("the launch side is told");
8532
8533 director.directed.lock().unwrap().insert(pool.id, 0);
8535 reconciler.reconcile(std::slice::from_ref(&pool)).await;
8536 assert!(
8537 tokio::time::timeout(Duration::from_millis(50), signal.notified())
8538 .await
8539 .is_err()
8540 );
8541 }
8542
8543 fn pool_profile(id: u128, max: u16) -> ScalePolicy {
8545 let labels = fixtures::routing_labels("home")
8546 .with_labels([pool_label(runner_manager_domain::model::Os::Windows)]);
8547 fixtures::policy()
8548 .id(PolicyId::from_u128(id))
8549 .repository("acme/app")
8550 .autoscale("home", max)
8551 .routing_labels(labels)
8552 .active()
8553 .build()
8554 }
8555
8556 fn pool_queue(pool: usize, plain: usize) -> Arc<FakeDemand> {
8559 let mut queue = fixtures::queued_jobs(&[HOST_LABEL, "rm-pool-windows"], pool);
8560 queue.extend(fixtures::queued_jobs(&[HOST_LABEL], plain));
8561 let demand = Arc::new(FakeDemand::default());
8562 demand.set(PollOutcome::Ready(QueuedDemand::of(
8563 repo("acme/app"),
8564 queue,
8565 )));
8566 demand
8567 }
8568
8569 fn reconciler_reading(
8570 demand: Arc<FakeDemand>,
8571 launcher: Arc<FakeLauncher>,
8572 director: Arc<FixedDirector>,
8573 ) -> Reconciler {
8574 Reconciler::new(
8575 host_with(10),
8576 ReconcilerPorts {
8577 demand,
8578 launcher,
8579 lock: Arc::new(InProcessAllocationLock::new()),
8580 directory: Arc::new(FakeDirectory::default()),
8581 clock: Arc::new(FakeClock::default()),
8582 jitter: Arc::new(NoJitter),
8583 events: Arc::new(NoEvents),
8584 },
8585 )
8586 .with_director(director as Arc<dyn DemandDirector>)
8587 }
8588
8589 #[tokio::test]
8594 async fn a_directed_pool_profile_serves_the_jobs_that_ask_for_no_pool() {
8595 let (profile, demand) = (pool_profile(1, 5), pool_queue(1, 2));
8596 let fake = Arc::new(FakeLauncher::new());
8597 let director = Arc::new(FixedDirector::default());
8598 director.directed.lock().unwrap().insert(profile.id, 1);
8599 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director);
8600
8601 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8602 assert_eq!(
8603 report.started, 3,
8604 "one runner for the launch, one for each job that asks for no pool"
8605 );
8606 }
8607
8608 #[tokio::test]
8612 async fn a_coordinator_that_places_plain_jobs_leaves_them_to_its_launches() {
8613 let (profile, demand) = (pool_profile(1, 5), pool_queue(0, 2));
8614 let fake = Arc::new(FakeLauncher::new());
8615 let director = Arc::new(FixedDirector::default());
8616 director.directed.lock().unwrap().insert(profile.id, 1);
8617 director.places_plain.store(true, Ordering::SeqCst);
8618 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director);
8619
8620 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8621
8622 assert_eq!(report.started, 1, "the launch's runner only");
8623 }
8624
8625 #[tokio::test]
8628 async fn a_directed_pool_profile_leaves_pool_jobs_to_the_launches() {
8629 let (profile, demand) = (pool_profile(1, 5), pool_queue(2, 0));
8630 let fake = Arc::new(FakeLauncher::new());
8631 let director = Arc::new(FixedDirector::default());
8632 director.directed.lock().unwrap().insert(profile.id, 0);
8633 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director);
8634
8635 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8636
8637 assert_eq!(report.started, 0);
8638 assert_eq!(fake.live_count(), 0);
8639 }
8640
8641 #[tokio::test]
8644 async fn a_pass_with_a_launch_side_starts_the_non_pool_jobs_runner() {
8645 let (profile, demand) = (pool_profile(1, 5), pool_queue(0, 1));
8646 let fake = Arc::new(FakeLauncher::new());
8647 let director = Arc::new(FixedDirector::default());
8648 director.directed.lock().unwrap().insert(profile.id, 0);
8649 let mut reconciler = reconciler_reading(demand, Arc::clone(&fake), director)
8650 .with_launch_side(Arc::new(tokio::sync::Notify::new()));
8651
8652 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8653
8654 assert_eq!(report.started, 1);
8655 assert_eq!(fake.live_count(), 1);
8656
8657 let report = reconciler.reconcile(std::slice::from_ref(&profile)).await;
8659 assert_eq!(report.started, 0);
8660 assert_eq!(fake.live_count(), 1);
8661 }
8662
8663 #[tokio::test(start_paused = true)]
8666 async fn an_older_launch_that_cannot_reserve_is_passed_over_once() {
8667 let fake = Arc::new(FakeLauncher::new());
8668 let director = Arc::new(FixedDirector::default());
8669 let pool = policy(1, "acme/app", 3);
8670 director.directed.lock().unwrap().insert(pool.id, 1);
8671 let reconciler = directed_reconciler(
8672 5,
8673 Arc::clone(&fake) as _,
8674 Arc::new(InProcessAllocationLock::new()),
8675 Arc::clone(&director),
8676 );
8677
8678 scripted_older(&director, &[(1, 1)]);
8679 let began = tokio::time::Instant::now();
8680 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8681 assert_eq!(report.started, 1, "started anyway");
8682 assert!(began.elapsed() >= OLDER_LAUNCH_TURN);
8683 assert!(began.elapsed() < OLDER_LAUNCH_TURN + Duration::from_secs(1));
8684 assert_eq!(director.passed_over.load(Ordering::SeqCst), 1);
8685
8686 director.directed.lock().unwrap().insert(pool.id, 2);
8687 let began = tokio::time::Instant::now();
8688 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8689 assert_eq!(report.started, 1);
8690 assert_eq!(began.elapsed(), Duration::ZERO);
8691 }
8692
8693 #[tokio::test]
8696 async fn a_targets_profiles_reserve_oldest_launch_first() {
8697 let director = Arc::new(FixedDirector::default());
8698 let (younger, older) = (policy(1, "acme/app", 3), policy(2, "acme/app", 3));
8699 director.directed.lock().unwrap().insert(younger.id, 1);
8700 director.directed.lock().unwrap().insert(older.id, 1);
8701 director.orders.lock().unwrap().insert(younger.id, 9);
8702 director.orders.lock().unwrap().insert(older.id, 4);
8703 let reconciler = directed_reconciler(
8704 5,
8705 Arc::new(FakeLauncher::new()) as _,
8706 Arc::new(InProcessAllocationLock::new()),
8707 Arc::clone(&director),
8708 );
8709
8710 let report = reconciler
8711 .start_directed(&[younger.clone(), older.clone()])
8712 .await;
8713
8714 assert_eq!(report.started, 2);
8715 let order: Vec<PolicyId> = director
8716 .started
8717 .lock()
8718 .unwrap()
8719 .iter()
8720 .map(|(policy, _)| *policy)
8721 .collect();
8722 assert_eq!(order, vec![older.id, younger.id]);
8723 }
8724
8725 fn serving<'a>(
8727 reconciler: &'a Reconciler,
8728 pool: &ScalePolicy,
8729 signal: &'a tokio::sync::Notify,
8730 stop: tokio::sync::watch::Receiver<bool>,
8731 ) -> impl std::future::Future<Output = ()> + 'a {
8732 let pool = pool.clone();
8733 async move {
8734 let policies = move || vec![pool.clone()];
8735 reconciler.serve_launches(&policies, signal, stop).await;
8736 }
8737 }
8738
8739 #[tokio::test]
8743 async fn a_launch_accepted_during_a_batch_starts_without_waiting_for_it() {
8744 let fake = Arc::new(FakeLauncher::new());
8745 let held = Arc::new(HeldBringUp {
8746 inner: Arc::clone(&fake),
8747 barrier: Arc::new(tokio::sync::Barrier::new(2)),
8748 fail: None,
8749 waits: Mutex::default(),
8750 });
8751 let director = Arc::new(FixedDirector::default());
8752 let pool = policy(1, "acme/app", 3);
8753 director.directed.lock().unwrap().insert(pool.id, 1);
8754 let reconciler = directed_reconciler(
8755 5,
8756 held as _,
8757 Arc::new(InProcessAllocationLock::new()),
8758 Arc::clone(&director),
8759 );
8760 let signal = tokio::sync::Notify::new();
8761 let (stop, stopped) = tokio::sync::watch::channel(false);
8762
8763 tokio::time::timeout(Duration::from_secs(10), async {
8764 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8765 signal.notify_one();
8766 while director.started.lock().unwrap().is_empty() {
8767 tokio::task::yield_now().await;
8768 }
8769 director.directed.lock().unwrap().insert(pool.id, 2);
8770 signal.notify_one();
8771 while director.started.lock().unwrap().len() < 2 {
8772 tokio::task::yield_now().await;
8773 }
8774 stop.send_replace(true);
8775 })
8776 })
8777 .await
8778 .expect("the second came up beside the first, and stopping ends it");
8779 assert_eq!(fake.live_count(), 2);
8780 }
8781
8782 #[tokio::test]
8785 async fn a_launch_during_a_batch_takes_no_slot_that_batch_holds() {
8786 let fake = Arc::new(FakeLauncher::new());
8787 let held = Arc::new(HeldBringUp {
8788 inner: Arc::clone(&fake),
8789 barrier: Arc::new(tokio::sync::Barrier::new(2)),
8790 fail: None,
8791 waits: Mutex::default(),
8792 });
8793 let barrier = Arc::clone(&held.barrier);
8794 let director = Arc::new(FixedDirector::default());
8795 let pool = policy(1, "acme/app", 3);
8796 director.directed.lock().unwrap().insert(pool.id, 1);
8797 let reconciler = directed_reconciler(
8798 1,
8799 held as _,
8800 Arc::new(InProcessAllocationLock::new()),
8801 Arc::clone(&director),
8802 );
8803 let signal = tokio::sync::Notify::new();
8804 let (stop, stopped) = tokio::sync::watch::channel(false);
8805
8806 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8807 signal.notify_one();
8808 while director.started.lock().unwrap().is_empty() {
8809 tokio::task::yield_now().await;
8810 }
8811 director.directed.lock().unwrap().insert(pool.id, 2);
8812 signal.notify_one();
8813 for _ in 0..100 {
8814 tokio::task::yield_now().await;
8815 }
8816 assert_eq!(
8817 director.started.lock().unwrap().len(),
8818 1,
8819 "the host is full"
8820 );
8821 assert_eq!(fake.live_count(), 1);
8822 stop.send_replace(true);
8823 barrier.wait().await;
8824 });
8825 }
8826
8827 #[tokio::test(start_paused = true)]
8831 async fn a_reserved_runner_comes_up_before_the_next_reservation_ends() {
8832 let held = Arc::new(HeldBringUp {
8833 inner: Arc::new(FakeLauncher::new()),
8834 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8835 fail: None,
8836 waits: Mutex::default(),
8837 });
8838 let director = Arc::new(FixedDirector::default());
8839 let pool = policy(1, "acme/app", 3);
8840 director.directed.lock().unwrap().insert(pool.id, 2);
8841 let reconciler = directed_reconciler(
8842 5,
8843 Arc::clone(&held) as _,
8844 Arc::new(InProcessAllocationLock::new()),
8845 Arc::clone(&director),
8846 );
8847 let signal = tokio::sync::Notify::new();
8848 let (stop, stopped) = tokio::sync::watch::channel(false);
8849
8850 scripted_older(&director, &[(0, 1), (1, 1)]);
8852 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8853 signal.notify_one();
8854 tokio::time::sleep(Duration::from_millis(1)).await;
8855 assert_eq!(
8856 held.waits.lock().unwrap().len(),
8857 1,
8858 "the first is coming up"
8859 );
8860 tokio::time::sleep(OLDER_LAUNCH_TURN + Duration::from_secs(1)).await;
8861 assert_eq!(held.waits.lock().unwrap().len(), 2);
8862 stop.send_replace(true);
8863 });
8864 }
8865
8866 #[tokio::test(start_paused = true)]
8869 async fn a_launch_that_did_not_start_is_retried_and_a_stop_ends_it() {
8870 let held = Arc::new(HeldBringUp {
8871 inner: Arc::new(FakeLauncher::new()),
8872 barrier: Arc::new(tokio::sync::Barrier::new(1)),
8873 fail: Some(FailureReason::JitRequestFailed),
8874 waits: Mutex::default(),
8875 });
8876 let director = Arc::new(FixedDirector::default());
8877 let pool = policy(1, "acme/app", 3);
8878 director.directed.lock().unwrap().insert(pool.id, 1);
8879 director.orders.lock().unwrap().insert(pool.id, 1);
8880 let reconciler = directed_reconciler(
8881 5,
8882 held as _,
8883 Arc::new(InProcessAllocationLock::new()),
8884 Arc::clone(&director),
8885 );
8886 let signal = tokio::sync::Notify::new();
8887 let (stop, stopped) = tokio::sync::watch::channel(false);
8888
8889 tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
8890 signal.notify_one();
8891 tokio::time::sleep(Duration::from_millis(1)).await;
8892 assert_eq!(director.failed.lock().unwrap().len(), 1);
8893 director.directed.lock().unwrap().insert(pool.id, 2);
8895 tokio::time::sleep(LAUNCH_RETRY - Duration::from_secs(1)).await;
8896 assert_eq!(director.failed.lock().unwrap().len(), 1, "not yet");
8897 tokio::time::sleep(Duration::from_secs(2)).await;
8898 assert_eq!(director.failed.lock().unwrap().len(), 2, "retried");
8899
8900 stop.send_replace(true);
8901 signal.notify_one();
8902 });
8903 assert_eq!(director.failed.lock().unwrap().len(), 2);
8904 }
8905
8906 #[tokio::test]
8910 async fn a_directed_policy_starts_only_what_it_was_directed_to_within_its_ceilings() {
8911 let demand = pool_queue(5, 0);
8912 let launcher = Arc::new(FakeLauncher::new());
8913 let director = Arc::new(FixedDirector::default());
8914 let mut reconciler = Reconciler::new(
8915 host_with(10),
8916 ReconcilerPorts {
8917 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
8918 launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
8919 lock: Arc::new(InProcessAllocationLock::new()),
8920 directory: Arc::new(FakeDirectory::default()),
8921 clock: Arc::new(FakeClock::default()),
8922 jitter: Arc::new(NoJitter),
8923 events: Arc::new(NoEvents),
8924 },
8925 )
8926 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8927 let pool = pool_profile(1, 2);
8928
8929 director.directed.lock().unwrap().insert(pool.id, 0);
8931 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8932 assert_eq!(report.started, 0);
8933
8934 director.directed.lock().unwrap().insert(pool.id, 1);
8936 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8937 assert_eq!(report.started, 1);
8938 assert_eq!(director.started.lock().unwrap().len(), 1);
8939
8940 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8942 assert_eq!(report.started, 0);
8943
8944 director.directed.lock().unwrap().insert(pool.id, 3);
8946 let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
8947 assert_eq!(report.started, 1);
8948 assert_eq!(launcher.live_count(), 2);
8949
8950 let polls = demand.polls().len();
8952 director.directed.lock().unwrap().insert(pool.id, 0);
8953 let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
8954 assert_eq!(report.started, 0);
8955 let fresh = Arc::new(FakeLauncher::new());
8956 let starter = Reconciler::new(
8957 host_with(10),
8958 ReconcilerPorts {
8959 demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
8960 launcher: Arc::clone(&fresh) as Arc<dyn RunnerLauncher>,
8961 lock: Arc::new(InProcessAllocationLock::new()),
8962 directory: Arc::new(FakeDirectory::default()),
8963 clock: Arc::new(FakeClock::default()),
8964 jitter: Arc::new(NoJitter),
8965 events: Arc::new(NoEvents),
8966 },
8967 )
8968 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8969 let wide = pool_profile(2, 6);
8970 director.directed.lock().unwrap().insert(wide.id, 6);
8971 let report = starter.start_directed(std::slice::from_ref(&wide)).await;
8972 assert_eq!(report.started, 6, "six launches, one call");
8973 assert_eq!(demand.polls().len(), polls, "GitHub was not polled");
8974 assert_eq!(
8976 starter
8977 .start_directed(&[policy(3, "acme/app", 2)])
8978 .await
8979 .started,
8980 0
8981 );
8982
8983 let failing = Arc::new(FakeLauncher::new());
8985 let mut reconciler = Reconciler::new(
8986 host_with(10),
8987 ReconcilerPorts {
8988 demand: Arc::new(FakeDemand::failing(RefreshState::Offline)),
8989 launcher: Arc::clone(&failing) as Arc<dyn RunnerLauncher>,
8990 lock: Arc::new(InProcessAllocationLock::new()),
8991 directory: Arc::new(FakeDirectory::default()),
8992 clock: Arc::new(FakeClock::default()),
8993 jitter: Arc::new(NoJitter),
8994 events: Arc::new(NoEvents),
8995 },
8996 )
8997 .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
8998 director.directed.lock().unwrap().insert(pool.id, 1);
8999 assert_eq!(reconciler.reconcile(&[pool]).await.started, 1);
9000
9001 let own = Arc::new(FakeLauncher::new());
9003 let mut reconciler = Reconciler::new(
9004 host_with(10),
9005 ReconcilerPorts {
9006 demand: Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
9007 launcher: Arc::clone(&own) as Arc<dyn RunnerLauncher>,
9008 lock: Arc::new(InProcessAllocationLock::new()),
9009 directory: Arc::new(FakeDirectory::default()),
9010 clock: Arc::new(FakeClock::default()),
9011 jitter: Arc::new(NoJitter),
9012 events: Arc::new(NoEvents),
9013 },
9014 )
9015 .with_director(Arc::new(OwnDemand));
9016 assert_eq!(
9017 reconciler
9018 .reconcile(&[policy(1, "acme/app", 2)])
9019 .await
9020 .started,
9021 2
9022 );
9023 }
9024}