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