/build/source/nativelink-scheduler/src/unsatisfiable_tracker.rs

Source
// Copyright 2026 Trace Machina, Inc. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Tracks how long each action property shape has gone without any worker
//! able to run it.
//!
//! Time is kept per shape rather than per operation, so a build whose actions
//! all queue together waits once rather than once per action. An action is
//! only failed once its shape is due *and* the action has itself been queued
//! for the timeout, so one that arrives while its shape is already due is not
//! failed on sight. The tracker therefore also knows when the youngest such
//! action becomes eligible, so the matching loop can wake for it, and it only
//! backs off for a shape whose eligible actions could not be retired.

use core::time::Duration;
use std::collections::HashMap;
use std::time::SystemTime;

use crate::match_outcome::PropertyShape;

/// How often the same shape is logged while it stays unsatisfiable.
const WARN_INTERVAL: Duration = Duration::from_mins(1);

/// Shortest wait before re-running a matching pass on behalf of a shape.
const MIN_RECHECK: Duration = Duration::from_secs(1);

/// Most doublings of `MIN_RECHECK` while a due shape keeps failing to be
/// retired, so a stuck action wakes the loop at most every 32 seconds.
const MAX_DUE_BACKOFF_DOUBLINGS: u32 = 5;

#[derive(Debug)]
struct ShapeState {
    first_unsatisfiable: SystemTime,
    last_seen: SystemTime,
    /// Fleet generation when the shape was last seen unsatisfiable.
    fleet_generation: u64,
    seen_this_pass: bool,
    seen_last_pass: bool,
    last_warned: Option<SystemTime>,
    /// Passes in which the shape was due and an eligible action of it was
    /// seen but could not be failed (the per-pass cap, or a version
    /// conflict). Passes where every action was merely too young do not
    /// count: nothing was stuck, it just was not time yet.
    due_passes: u32,
    /// Whether the shape was due in the pass being recorded.
    due_this_pass: bool,
    /// Actions of this shape seen this pass that had waited the timeout.
    eligible_this_pass: u32,
    /// Actions of this shape failed this pass (see `note_failed`).
    failed_this_pass: u32,
    /// When the youngest action of this shape seen this pass will have
    /// waited the timeout, so the loop can wake for it. Kept until the
    /// shape is next seen, because `next_deadline` runs after `end_pass`.
    earliest_eligible: Option<SystemTime>,
}

/// What the matching pass should do with one unsatisfiable action.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Observation {
    /// How long the shape has been unsatisfiable.
    pub waited: Duration,
    /// The shape has been unsatisfiable for the configured timeout.
    pub is_due: bool,
    /// The action has itself been queued for the configured timeout. It is
    /// failed only when this and `is_due` both hold.
    pub action_eligible: bool,
    /// The shape has not been logged recently.
    pub should_warn: bool,
}

#[derive(Debug)]
pub struct UnsatisfiableTracker {
    /// `None` never fails an action.
    timeout: Option<Duration>,
    shapes: HashMap<PropertyShape, ShapeState>,
    seen_this_pass: u64,
}

impl UnsatisfiableTracker {
    #[must_use]
    pub fn new(timeout: Option<Duration>) -> Self {
        Self {
            timeout,
            shapes: HashMap::new(),
            seen_this_pass: 0,
        }
    }

    /// Records that an action with this shape, first submitted at
    /// `insert_timestamp`, was found unsatisfiable.
    pub fn observe(
        &mut self,
        shape: PropertyShape,
        now: SystemTime,
        fleet_generation: u64,
        insert_timestamp: SystemTime,
    ) -> Observation {
        self.seen_this_pass += 1;
        let timeout = self.timeout;
        let state = self.shapes.entry(shape).or_insert(ShapeState {
            first_unsatisfiable: now,
            last_seen: now,
            fleet_generation,
            seen_this_pass: false,
            seen_last_pass: false,
            last_warned: None,
            due_passes: 0,
            due_this_pass: false,
            eligible_this_pass: 0,
            failed_this_pass: 0,
            earliest_eligible: None,
        });
        let first_this_pass = !state.seen_this_pass;
        if first_this_pass {
            // A shape kept from an earlier build only carries over the time
            // it was actually queued, not the idle gap since.
            if !state.seen_last_pass {
                let gap = now.duration_since(state.last_seen).unwrap_or_default();
                state.first_unsatisfiable += gap;
            }
            state.due_this_pass = false;
            state.eligible_this_pass = 0;
            state.failed_this_pass = 0;
            state.earliest_eligible = None;
        }
        state.last_seen = now;
        state.fleet_generation = fleet_generation;
        state.seen_this_pass = true;

        let should_warn = state.last_warned.is_none_or(|last_warned| {125
            now.duration_since(last_warned).unwrap_or_default() >= WARN_INTERVAL
        });
        if should_warn {
            state.last_warned = Some(now);
        }

        let waited = now
            .duration_since(state.first_unsatisfiable)
            .unwrap_or_default();
        let is_due = timeout.is_some_and(|timeout| waited143 >= timeout143);
        state.due_this_pass |= is_due;

        // A clock that went backwards reads as not yet waited, the safe
        // direction; `now` never being reached reads the same way.
        let action_eligible = timeout.is_some_and(|timeout| {143
            now.duration_since(insert_timestamp)
                .is_ok_and(|action_waited| action_waited >= timeout)
        });
        if action_eligible {
            state.eligible_this_pass += 1;
        } else if let Some(eligible_at) =
            timeout.and_then(|timeout| insert_timestamp93.checked_add93(timeout93))
        {
            state.earliest_eligible = Some(
                state
                    .earliest_eligible
                    .map_or(eligible_at, |earliest| earliest20.min20(eligible_at20)),
            );
        }
        Observation {
            waited,
            is_due,
            action_eligible,
            should_warn,
        }
    }

    /// Records that an action with this shape was failed this pass, so a
    /// pass that retired every eligible action does not read as stuck.
    pub fn note_failed(&mut self, shape: &PropertyShape) {
        if let Some(state) = self.shapes.get_mut(shape) {
            state.failed_this_pass += 1;
        }0
    }

    /// Whether a shape that stays unsatisfiable is ever failed.
    #[must_use]
    pub const fn fails_actions(&self) -> bool {
        self.timeout.is_some()
    }

    /// Closes a matching pass and returns how many unsatisfiable actions it
    /// saw.
    ///
    /// A shape that was not seen is kept for one timeout period so its clock
    /// keeps running and it stays due for actions that arrive one after
    /// another; each such action still waits its own minimum before it is
    /// failed. The shape is dropped at once if the fleet changed, because a
    /// new worker may be able to run it.
    ///
    /// A due shape backs off (see `next_deadline`) only when an eligible
    /// action of it was seen but not failed this pass.
    pub fn end_pass(&mut self, now: SystemTime, fleet_generation: u64) -> u64 {
        let linger = self.timeout.unwrap_or_default();
        self.shapes.retain(|_, state| {139
            if state.seen_this_pass
                && state.due_this_pass
                && state.eligible_this_pass > state.failed_this_pass
            {
                state.due_passes = state.due_passes.saturating_add(1);
            }
            state.seen_last_pass = state.seen_this_pass;
            state.seen_this_pass = false;
            state.seen_last_pass
                || (state.fleet_generation == fleet_generation
                    && now.duration_since(state.last_seen).unwrap_or_default() < linger)
        });
        core::mem::take(&mut self.seen_this_pass)
    }

    /// How long until the loop should run again for a shape seen in the
    /// last pass, if any: when the shape comes due, when its youngest action
    /// becomes eligible, or, for a due shape whose eligible actions could
    /// not be retired, a backoff that doubles up to 32 seconds.
    #[must_use]
    pub fn next_deadline(&self, now: SystemTime) -> Option<Duration> {
        let timeout134 = self.timeout?36.3k;
        self.shapes
            .values()
            .filter(|state| state.seen_last_pass)
            .filter_map(|state| {44
                let waited = now
                    .duration_since(state.first_unsatisfiable)
                    .unwrap_or_default();
                let shape_term = if waited < timeout {
                    Some(timeout.saturating_sub(waited))
                } else if state.eligible_this_pass > state.failed_this_pass {
                    // Still queued though eligible, so wake less and less
                    // often for it.
                    Some(MIN_RECHECK * 2u32.pow(state.due_passes.min(MAX_DUE_BACKOFF_DOUBLINGS)))
                } else {
                    // Due, but nothing of it is stuck: only a young action can
                    // change anything, and that is covered below.
                    None
                };
                let eligible_term = state
                    .earliest_eligible
                    .map(|eligible_at| eligible_at34.duration_since34(now34).unwrap_or_default34());
                match (shape_term, eligible_term) {
                    (Some(a), Some(b)) => Some(a.min(b)),
                    (a, b) => a.or(b),
                }
                .map(|deadline| deadline41.max41(MIN_RECHECK))
            })
            .min()
    }
}

Coverage Report

Created: 2026-11-13 23:29

Line	Count	Source
1		// Copyright 2026 Trace Machina, Inc. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		//! Tracks how long each action property shape has gone without any worker
16		//! able to run it.
17		//!
18		//! Time is kept per shape rather than per operation, so a build whose actions
19		//! all queue together waits once rather than once per action. An action is
20		//! only failed once its shape is due and the action has itself been queued
21		//! for the timeout, so one that arrives while its shape is already due is not
22		//! failed on sight. The tracker therefore also knows when the youngest such
23		//! action becomes eligible, so the matching loop can wake for it, and it only
24		//! backs off for a shape whose eligible actions could not be retired.
25
26		use core::time::Duration;
27		use std::collections::HashMap;
28		use std::time::SystemTime;
29
30		use crate::match_outcome::PropertyShape;
31
32		/// How often the same shape is logged while it stays unsatisfiable.
33		const WARN_INTERVAL: Duration = Duration::from_mins(1);
34
35		/// Shortest wait before re-running a matching pass on behalf of a shape.
36		const MIN_RECHECK: Duration = Duration::from_secs(1);
37
38		/// Most doublings of `MIN_RECHECK` while a due shape keeps failing to be
39		/// retired, so a stuck action wakes the loop at most every 32 seconds.
40		const MAX_DUE_BACKOFF_DOUBLINGS: u32 = 5;
41
42		#[derive(Debug)]
43		struct ShapeState {
44		first_unsatisfiable: SystemTime,
45		last_seen: SystemTime,
46		/// Fleet generation when the shape was last seen unsatisfiable.
47		fleet_generation: u64,
48		seen_this_pass: bool,
49		seen_last_pass: bool,
50		last_warned: Option<SystemTime>,
51		/// Passes in which the shape was due and an eligible action of it was
52		/// seen but could not be failed (the per-pass cap, or a version
53		/// conflict). Passes where every action was merely too young do not
54		/// count: nothing was stuck, it just was not time yet.
55		due_passes: u32,
56		/// Whether the shape was due in the pass being recorded.
57		due_this_pass: bool,
58		/// Actions of this shape seen this pass that had waited the timeout.
59		eligible_this_pass: u32,
60		/// Actions of this shape failed this pass (see `note_failed`).
61		failed_this_pass: u32,
62		/// When the youngest action of this shape seen this pass will have
63		/// waited the timeout, so the loop can wake for it. Kept until the
64		/// shape is next seen, because `next_deadline` runs after `end_pass`.
65		earliest_eligible: Option<SystemTime>,
66		}
67
68		/// What the matching pass should do with one unsatisfiable action.
69		#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70		pub struct Observation {
71		/// How long the shape has been unsatisfiable.
72		pub waited: Duration,
73		/// The shape has been unsatisfiable for the configured timeout.
74		pub is_due: bool,
75		/// The action has itself been queued for the configured timeout. It is
76		/// failed only when this and `is_due` both hold.
77		pub action_eligible: bool,
78		/// The shape has not been logged recently.
79		pub should_warn: bool,
80		}
81
82		#[derive(Debug)]
83		pub struct UnsatisfiableTracker {
84		/// `None` never fails an action.
85		timeout: Option<Duration>,
86		shapes: HashMap<PropertyShape, ShapeState>,
87		seen_this_pass: u64,
88		}
89
90		impl UnsatisfiableTracker {
91		#[must_use]
92	355	pub fn new(timeout: Option<Duration>) -> Self {
93	355	Self {
94	355	timeout,
95	355	shapes: HashMap::new(),
96	355	seen_this_pass: 0,
97	355	}
98	355	}
99
100		/// Records that an action with this shape, first submitted at
101		/// `insert_timestamp`, was found unsatisfiable.
102	165	pub fn observe(
103	165	&mut self,
104	165	shape: PropertyShape,
105	165	now: SystemTime,
106	165	fleet_generation: u64,
107	165	insert_timestamp: SystemTime,
108	165	) -> Observation {
109	165	self.seen_this_pass += 1;
110	165	let timeout = self.timeout;
111	165	let state = self.shapes.entry(shape).or_insert(ShapeState {
112	165	first_unsatisfiable: now,
113	165	last_seen: now,
114	165	fleet_generation,
115	165	seen_this_pass: false,
116	165	seen_last_pass: false,
117	165	last_warned: None,
118	165	due_passes: 0,
119	165	due_this_pass: false,
120	165	eligible_this_pass: 0,
121	165	failed_this_pass: 0,
122	165	earliest_eligible: None,
123	165	});
124	165	let first_this_pass = !state.seen_this_pass;
125	165	if first_this_pass {
126		// A shape kept from an earlier build only carries over the time
127		// it was actually queued, not the idle gap since.
128	133	if !state.seen_last_pass {
129	41	let gap = now.duration_since(state.last_seen).unwrap_or_default();
130	41	state.first_unsatisfiable += gap;
131	92	}
132	133	state.due_this_pass = false;
133	133	state.eligible_this_pass = 0;
134	133	state.failed_this_pass = 0;
135	133	state.earliest_eligible = None;
136	32	}
137	165	state.last_seen = now;
138	165	state.fleet_generation = fleet_generation;
139	165	state.seen_this_pass = true;
140
141	165	let should_warn = state.last_warned.is_none_or(\|last_warned\| {125
142	125	now.duration_since(last_warned).unwrap_or_default() >= WARN_INTERVAL
143	125	});
144	165	if should_warn {
145	68	state.last_warned = Some(now);
146	97	}
147
148	165	let waited = now
149	165	.duration_since(state.first_unsatisfiable)
150	165	.unwrap_or_default();
151	165	let is_due = timeout.is_some_and(\|timeout\| waited143 >= timeout143 );
152	165	state.due_this_pass \|= is_due;
153
154		// A clock that went backwards reads as not yet waited, the safe
155		// direction; `now` never being reached reads the same way.
156	165	let action_eligible = timeout.is_some_and(\|timeout\| {143
157	143	now.duration_since(insert_timestamp)
158	143	.is_ok_and(\|action_waited\| action_waited >= timeout)
159	143	});
160	165	if action_eligible {
161	50	state.eligible_this_pass += 1;
162	93	} else if let Some(eligible_at) =
163	115	timeout.and_then(\|timeout\| insert_timestamp93 . checked_add93 ( timeout93 ))
164		{
165		state.earliest_eligible = Some(
166	93	state
167	93	.earliest_eligible
168	93	.map_or(eligible_at, \|earliest\| earliest20 . min20 ( eligible_at20 )),
169		);
170	22	}
171	165	Observation {
172	165	waited,
173	165	is_due,
174	165	action_eligible,
175	165	should_warn,
176	165	}
177	165	}
178
179		/// Records that an action with this shape was failed this pass, so a
180		/// pass that retired every eligible action does not read as stuck.
181	30	pub fn note_failed(&mut self, shape: &PropertyShape) {
182	30	if let Some(state) = self.shapes.get_mut(shape) {
183	30	state.failed_this_pass += 1;
184	30	}0
185	30	}
186
187		/// Whether a shape that stays unsatisfiable is ever failed.
188		#[must_use]
189	134	pub const fn fails_actions(&self) -> bool {
190	134	self.timeout.is_some()
191	134	}
192
193		/// Closes a matching pass and returns how many unsatisfiable actions it
194		/// saw.
195		///
196		/// A shape that was not seen is kept for one timeout period so its clock
197		/// keeps running and it stays due for actions that arrive one after
198		/// another; each such action still waits its own minimum before it is
199		/// failed. The shape is dropped at once if the fleet changed, because a
200		/// new worker may be able to run it.
201		///
202		/// A due shape backs off (see `next_deadline`) only when an eligible
203		/// action of it was seen but not failed this pass.
204	36.3k	pub fn end_pass(&mut self, now: SystemTime, fleet_generation: u64) -> u64 {
205	36.3k	let linger = self.timeout.unwrap_or_default();
206	36.3k	self.shapes.retain(\|_, state\| {139
207	139	if state.seen_this_pass
208	127	&& state.due_this_pass
209	47	&& state.eligible_this_pass > state.failed_this_pass
210	8	{
211	8	state.due_passes = state.due_passes.saturating_add(1);
212	131	}
213	139	state.seen_last_pass = state.seen_this_pass;
214	139	state.seen_this_pass = false;
215	139	state.seen_last_pass
216	12	\|\| (state.fleet_generation == fleet_generation
217	5	&& now.duration_since(state.last_seen).unwrap_or_default() < linger)
218	139	});
219	36.3k	core::mem::take(&mut self.seen_this_pass)
220	36.3k	}
221
222		/// How long until the loop should run again for a shape seen in the
223		/// last pass, if any: when the shape comes due, when its youngest action
224		/// becomes eligible, or, for a due shape whose eligible actions could
225		/// not be retired, a backoff that doubles up to 32 seconds.
226		#[must_use]
227	36.5k	pub fn next_deadline(&self, now: SystemTime) -> Option<Duration> {
228	36.5k	let timeout134 = self.timeout ?36.3k ;
229	134	self.shapes
230	134	.values()
231	134	.filter(\|state\| state.seen_last_pass)
232	134	.filter_map(\|state\| {44
233	44	let waited = now
234	44	.duration_since(state.first_unsatisfiable)
235	44	.unwrap_or_default();
236	44	let shape_term = if waited < timeout {
237	26	Some(timeout.saturating_sub(waited))
238	18	} else if state.eligible_this_pass > state.failed_this_pass {
239		// Still queued though eligible, so wake less and less
240		// often for it.
241	7	Some(MIN_RECHECK * 2u32.pow(state.due_passes.min(MAX_DUE_BACKOFF_DOUBLINGS)))
242		} else {
243		// Due, but nothing of it is stuck: only a young action can
244		// change anything, and that is covered below.
245	11	None
246		};
247	44	let eligible_term = state
248	44	.earliest_eligible
249	44	.map(\|eligible_at\| eligible_at34 . duration_since34 ( now34 ). unwrap_or_default34 ());
250	44	match (shape_term, eligible_term) {
251	26	(Some(a), Some(b)) => Some(a.min(b)),
252	18	(a, b) => a.or(b),
253		}
254	44	.map(\|deadline\| deadline41 . max41 (MIN_RECHECK))
255	44	})
256	134	.min()
257	36.5k	}
258		}