aft/
build_breaker.rs

1//! Durable, domain-scoped admission breaker for background cold builds.
2//!
3//! The breaker deliberately measures only transactional extraction credit supplied
4//! by its caller. It never treats database size, row count, cursor movement, or
5//! SQLite page reuse as progress.
6
7use crate::db::{SqliteStore, TrackedConnection};
8use rusqlite::{params, Connection, OptionalExtension};
9use std::path::{Path, PathBuf};
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::sync::Mutex;
12use std::time::{SystemTime, UNIX_EPOCH};
13
14pub const ZERO_CREDIT_DEATH_LIMIT: u64 = 3;
15pub const CREDITED_DEATH_LIMIT: u64 = 6;
16pub const IN_BUILD_BURN_LIMIT_MS: u64 = 30 * 60 * 1_000;
17pub const TRIP_TTL_MS: u64 = 24 * 60 * 60 * 1_000;
18pub const ATTEMPT_MARKER_HEARTBEAT_INTERVAL_MS: u64 = 5_000;
19pub const ATTEMPT_MARKER_RECENT_HEARTBEAT_MS: u64 = 15_000;
20pub const TEMP_DELETE_AGE_FLOOR_MS: u64 = 24 * 60 * 60 * 1_000;
21pub const SWEEP_AMBIGUITY_TTL_MS: u64 = 7 * 24 * 60 * 60 * 1_000;
22pub const SWEEP_STAT_CHECK_CAP: usize = 64;
23pub const BREAKER_CONFIGURATION_VERSION: &str = "v1";
24
25static NEXT_ATTEMPT: AtomicU64 = AtomicU64::new(1);
26
27// Thread-local, not process-global: libtest runs sibling tests in parallel and
28// they open breakers of their own, so a shared counter cannot support the exact
29// open-count assertions the health rollup test makes about its own thread.
30#[cfg(test)]
31thread_local! {
32    static OPEN_CALLS_FOR_TEST: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
33}
34#[cfg(test)]
35static FAIL_NEXT_ACTIVE_SUSPENSIONS_FOR_TEST: std::sync::atomic::AtomicBool =
36    std::sync::atomic::AtomicBool::new(false);
37
38/// Every expensive background build must choose one explicit domain. The enum is
39/// intentionally exhaustive so new schedulers cannot silently bypass the breaker.
40#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
41pub enum BuildDomain {
42    CallgraphCold,
43    SearchCold,
44    SemanticSeed,
45    Tier2Scan,
46}
47
48impl BuildDomain {
49    pub const ALL: [Self; 4] = [
50        Self::CallgraphCold,
51        Self::SearchCold,
52        Self::SemanticSeed,
53        Self::Tier2Scan,
54    ];
55
56    pub const fn as_str(self) -> &'static str {
57        match self {
58            Self::CallgraphCold => "callgraph_cold",
59            Self::SearchCold => "search_cold",
60            Self::SemanticSeed => "semantic_seed",
61            Self::Tier2Scan => "tier2_scan",
62        }
63    }
64
65    fn from_persisted(value: &str) -> Option<Self> {
66        match value {
67            "callgraph_cold" => Some(Self::CallgraphCold),
68            "search_cold" => Some(Self::SearchCold),
69            "semantic_seed" => Some(Self::SemanticSeed),
70            "tier2_scan" => Some(Self::Tier2Scan),
71            _ => None,
72        }
73    }
74}
75
76/// Durable namespace. Configure generations and cache keys do not appear here:
77/// they may invalidate a staging cursor but must not launder death history.
78#[derive(Clone, Debug, Eq, PartialEq, Hash)]
79pub struct BreakerKey {
80    pub root_id: String,
81    pub domain: BuildDomain,
82    pub corpus_fingerprint: String,
83}
84
85impl BreakerKey {
86    pub fn new(
87        root_id: impl Into<String>,
88        domain: BuildDomain,
89        corpus_fingerprint: impl Into<String>,
90    ) -> Self {
91        Self {
92            root_id: root_id.into(),
93            domain,
94            corpus_fingerprint: corpus_fingerprint.into(),
95        }
96    }
97}
98
99#[derive(Clone, Debug, Eq, PartialEq)]
100pub struct BuildAttempt {
101    pub attempt_id: String,
102    pub start_committed_extracted_bytes: u64,
103}
104
105#[derive(Clone, Debug, Eq, PartialEq)]
106pub struct BuildSuspension {
107    pub domain: BuildDomain,
108    pub reason: String,
109    pub death_count: u64,
110    pub suspended_since_unix_ms: u64,
111}
112
113impl BuildSuspension {
114    pub fn age_millis_at(&self, now_ms: u64) -> u64 {
115        now_ms.saturating_sub(self.suspended_since_unix_ms)
116    }
117
118    pub fn age_seconds_at(&self, now_ms: u64) -> u64 {
119        self.age_millis_at(now_ms) / 1_000
120    }
121}
122
123#[derive(Clone, Debug, Eq, PartialEq)]
124pub enum BreakerAdmission {
125    Admitted(BuildAttempt),
126    Suspended(BuildSuspension),
127}
128
129#[derive(Debug)]
130pub enum BuildBreakerError {
131    Sqlite(rusqlite::Error),
132}
133
134impl From<rusqlite::Error> for BuildBreakerError {
135    fn from(error: rusqlite::Error) -> Self {
136        Self::Sqlite(error)
137    }
138}
139
140impl std::fmt::Display for BuildBreakerError {
141    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
142        match self {
143            Self::Sqlite(error) => write!(formatter, "build breaker database error: {error}"),
144        }
145    }
146}
147
148impl std::error::Error for BuildBreakerError {}
149
150pub type Result<T> = std::result::Result<T, BuildBreakerError>;
151
152/// SQLite-backed, root/domain/fingerprint-isolated death history.
153pub struct BuildDeathBreaker {
154    path: PathBuf,
155    /// One connection retains SQLite's initialized schema/WAL state for callers
156    /// that repeatedly inspect a breaker, while the mutex keeps its non-Sync
157    /// connection safe when a health thread and a build overlap. Build paths may
158    /// open the same WAL file independently: readers do not block writers, and
159    /// each connection has a five-second busy timeout for contested operations.
160    connection: Mutex<TrackedConnection>,
161}
162
163impl BuildDeathBreaker {
164    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
165        #[cfg(test)]
166        OPEN_CALLS_FOR_TEST.with(|calls| calls.set(calls.get() + 1));
167        let path = path.into();
168        if let Some(parent) = path.parent() {
169            std::fs::create_dir_all(parent).map_err(|error| {
170                BuildBreakerError::Sqlite(rusqlite::Error::ToSqlConversionFailure(Box::new(error)))
171            })?;
172        }
173        let connection = TrackedConnection::open(&path, SqliteStore::BreakerFile)?;
174        connection.busy_timeout(std::time::Duration::from_secs(5))?;
175        connection.execute_batch(
176            "PRAGMA journal_mode=WAL;
177                 PRAGMA synchronous=NORMAL;
178                 CREATE TABLE IF NOT EXISTS breaker_records (
179                    root_id TEXT NOT NULL,
180                    domain TEXT NOT NULL,
181                    corpus_fingerprint TEXT NOT NULL,
182                    configuration_version TEXT NOT NULL,
183                    zero_credit_deaths INTEGER NOT NULL DEFAULT 0,
184                    credited_deaths INTEGER NOT NULL DEFAULT 0,
185                    in_build_burn_ms INTEGER NOT NULL DEFAULT 0,
186                    suspended_reason TEXT,
187                    suspended_since_ms INTEGER,
188                    suspended_until_ms INTEGER,
189                    PRIMARY KEY(root_id, domain, corpus_fingerprint)
190                 );
191                 CREATE TABLE IF NOT EXISTS breaker_attempts (
192                    root_id TEXT NOT NULL,
193                    domain TEXT NOT NULL,
194                    corpus_fingerprint TEXT NOT NULL,
195                    attempt_id TEXT NOT NULL,
196                    start_committed_extracted_bytes INTEGER NOT NULL,
197                    death_charged INTEGER NOT NULL DEFAULT 0,
198                    PRIMARY KEY(root_id, domain, corpus_fingerprint, attempt_id)
199                 );",
200        )?;
201        Ok(Self {
202            path,
203            connection: Mutex::new(connection),
204        })
205    }
206
207    pub fn path(&self) -> &Path {
208        &self.path
209    }
210
211    pub fn admit(
212        &self,
213        key: &BreakerKey,
214        committed_extracted_bytes: u64,
215    ) -> Result<BreakerAdmission> {
216        self.admit_at(key, committed_extracted_bytes, unix_millis_now())
217    }
218
219    pub fn admit_at(
220        &self,
221        key: &BreakerKey,
222        committed_extracted_bytes: u64,
223        now_ms: u64,
224    ) -> Result<BreakerAdmission> {
225        self.with_connection(|conn| {
226            let tx = conn.transaction()?;
227            ensure_record(&tx, key)?;
228            let suspension = suspension_in_tx(&tx, key, now_ms)?;
229            if let Some(suspension) = suspension {
230                tx.commit()?;
231                return Ok(BreakerAdmission::Suspended(suspension));
232            }
233            let attempt = BuildAttempt {
234                attempt_id: format!(
235                    "{}-{}-{}",
236                    std::process::id(),
237                    now_ms,
238                    NEXT_ATTEMPT.fetch_add(1, Ordering::Relaxed)
239                ),
240                start_committed_extracted_bytes: committed_extracted_bytes,
241            };
242            tx.execute(
243                "INSERT INTO breaker_attempts(
244                    root_id, domain, corpus_fingerprint, attempt_id, start_committed_extracted_bytes
245                 ) VALUES(?1, ?2, ?3, ?4, ?5)",
246                params![
247                    key.root_id,
248                    key.domain.as_str(),
249                    key.corpus_fingerprint,
250                    attempt.attempt_id,
251                    committed_extracted_bytes as i64,
252                ],
253            )?;
254            tx.commit()?;
255            Ok(BreakerAdmission::Admitted(attempt))
256        })
257    }
258
259    /// Charge one *already attributed* exact-process death. Callers must only use
260    /// this after validating a durable expensive-phase marker and reading the
261    /// staging metadata counter. Repeating the same attempt is idempotent.
262    pub fn record_attributed_death(
263        &self,
264        key: &BreakerKey,
265        attempt_id: &str,
266        committed_extracted_bytes_at_death: u64,
267        durable_burn_ms: u64,
268    ) -> Result<Option<BuildSuspension>> {
269        self.record_attributed_death_at(
270            key,
271            attempt_id,
272            committed_extracted_bytes_at_death,
273            durable_burn_ms,
274            unix_millis_now(),
275        )
276    }
277
278    pub fn record_attributed_death_at(
279        &self,
280        key: &BreakerKey,
281        attempt_id: &str,
282        committed_extracted_bytes_at_death: u64,
283        durable_burn_ms: u64,
284        now_ms: u64,
285    ) -> Result<Option<BuildSuspension>> {
286        self.with_connection(|conn| {
287            let tx = conn.transaction()?;
288            ensure_record(&tx, key)?;
289            let attempt = tx
290                .query_row(
291                    "SELECT start_committed_extracted_bytes, death_charged
292                     FROM breaker_attempts
293                     WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3 AND attempt_id = ?4",
294                    params![key.root_id, key.domain.as_str(), key.corpus_fingerprint, attempt_id],
295                    |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)? != 0)),
296                )
297                .optional()?;
298            let Some((start, already_charged)) = attempt else {
299                tx.commit()?;
300                return Ok(None);
301            };
302            if already_charged {
303                let current = suspension_in_tx(&tx, key, now_ms)?;
304                tx.commit()?;
305                return Ok(current);
306            }
307
308            // A backward counter is integrity ambiguity, not free credit. It is
309            // deliberately charged as zero credit so it cannot erase history.
310            let credited = committed_extracted_bytes_at_death > start.max(0) as u64;
311            tx.execute(
312                "UPDATE breaker_attempts SET death_charged = 1
313                 WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3 AND attempt_id = ?4",
314                params![key.root_id, key.domain.as_str(), key.corpus_fingerprint, attempt_id],
315            )?;
316            tx.execute(
317                "UPDATE breaker_records
318                 SET zero_credit_deaths = zero_credit_deaths + ?4,
319                     credited_deaths = credited_deaths + ?5,
320                     in_build_burn_ms = in_build_burn_ms + ?6
321                 WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
322                params![
323                    key.root_id,
324                    key.domain.as_str(),
325                    key.corpus_fingerprint,
326                    i64::from(!credited),
327                    i64::from(credited),
328                    durable_burn_ms.min(i64::MAX as u64) as i64,
329                ],
330            )?;
331            let suspension = trip_if_needed(&tx, key, now_ms)?;
332            tx.commit()?;
333            Ok(suspension)
334        })
335    }
336
337    /// Credit cannot reset the burn ceiling. This is used for live heartbeat
338    /// checkpoints whose elapsed interval has already been durably bounded.
339    pub fn record_durable_burn_at(
340        &self,
341        key: &BreakerKey,
342        durable_burn_ms: u64,
343        now_ms: u64,
344    ) -> Result<Option<BuildSuspension>> {
345        self.with_connection(|conn| {
346            let tx = conn.transaction()?;
347            ensure_record(&tx, key)?;
348            tx.execute(
349                "UPDATE breaker_records SET in_build_burn_ms = in_build_burn_ms + ?4
350                 WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
351                params![
352                    key.root_id,
353                    key.domain.as_str(),
354                    key.corpus_fingerprint,
355                    durable_burn_ms.min(i64::MAX as u64) as i64,
356                ],
357            )?;
358            let suspension = trip_if_needed(&tx, key, now_ms)?;
359            tx.commit()?;
360            Ok(suspension)
361        })
362    }
363
364    /// A ready pointer flip is the only automatic full reset. Staged commits do
365    /// not call this method and therefore cannot launder a death loop.
366    pub fn record_ready_publication(&self, key: &BreakerKey) -> Result<()> {
367        self.with_connection(|conn| {
368            let tx = conn.transaction()?;
369            ensure_record(&tx, key)?;
370            reset_record(&tx, key)?;
371            tx.commit()?;
372            Ok(())
373        })
374    }
375
376    /// Doctor and force-rebuild use this explicit, tuple-scoped full reset.
377    pub fn explicit_reset(&self, key: &BreakerKey) -> Result<()> {
378        self.record_ready_publication(key)
379    }
380
381    pub fn suspension(&self, key: &BreakerKey) -> Result<Option<BuildSuspension>> {
382        self.suspension_at(key, unix_millis_now())
383    }
384
385    pub fn suspension_at(&self, key: &BreakerKey, now_ms: u64) -> Result<Option<BuildSuspension>> {
386        self.with_connection(|conn| {
387            let tx = conn.transaction()?;
388            ensure_record(&tx, key)?;
389            let suspension = suspension_in_tx(&tx, key, now_ms)?;
390            tx.commit()?;
391            Ok(suspension)
392        })
393    }
394
395    pub fn active_suspensions_for_root(&self, root_id: &str) -> Result<Vec<BuildSuspension>> {
396        self.active_suspensions_for_root_at(root_id, unix_millis_now())
397    }
398
399    /// Read every still-active domain suspension for one root from the durable
400    /// breaker rows. Health and doctor snapshots use this instead of inferring a
401    /// suspension from transient worker state, so every surface reports the same
402    /// persisted reason and counters.
403    pub fn active_suspensions_for_root_at(
404        &self,
405        root_id: &str,
406        now_ms: u64,
407    ) -> Result<Vec<BuildSuspension>> {
408        #[cfg(test)]
409        if FAIL_NEXT_ACTIVE_SUSPENSIONS_FOR_TEST.swap(false, Ordering::SeqCst) {
410            return Err(BuildBreakerError::Sqlite(rusqlite::Error::InvalidQuery));
411        }
412        self.with_connection(|conn| {
413            let mut statement = conn.prepare(
414                "SELECT domain, zero_credit_deaths, credited_deaths, suspended_reason, suspended_since_ms
415                 FROM breaker_records
416                 WHERE root_id = ?1
417                   AND configuration_version = ?2
418                   AND suspended_reason IS NOT NULL
419                   AND suspended_since_ms IS NOT NULL
420                   AND suspended_until_ms > ?3
421                 ORDER BY domain",
422            )?;
423            let rows = statement.query_map(
424                params![root_id, BREAKER_CONFIGURATION_VERSION, now_ms.min(i64::MAX as u64) as i64],
425                |row| {
426                    Ok((
427                        row.get::<_, String>(0)?,
428                        row.get::<_, i64>(1)?,
429                        row.get::<_, i64>(2)?,
430                        row.get::<_, String>(3)?,
431                        row.get::<_, i64>(4)?,
432                    ))
433                },
434            )?;
435            let mut suspensions = Vec::new();
436            for row in rows {
437                let (domain, zero_credit_deaths, credited_deaths, reason, since) = row?;
438                let Some(domain) = BuildDomain::from_persisted(&domain) else {
439                    continue;
440                };
441                suspensions.push(BuildSuspension {
442                    domain,
443                    reason,
444                    death_count: (zero_credit_deaths.max(0) as u64)
445                        .saturating_add(credited_deaths.max(0) as u64),
446                    suspended_since_unix_ms: since.max(0) as u64,
447                });
448            }
449            Ok(suspensions)
450        })
451    }
452
453    #[cfg(test)]
454    pub(crate) fn reset_open_calls_for_test() {
455        OPEN_CALLS_FOR_TEST.with(|calls| calls.set(0));
456    }
457
458    #[cfg(test)]
459    pub(crate) fn open_calls_for_test() -> u64 {
460        OPEN_CALLS_FOR_TEST.with(|calls| calls.get())
461    }
462
463    #[cfg(test)]
464    pub(crate) fn fail_next_active_suspensions_for_test() {
465        FAIL_NEXT_ACTIVE_SUSPENSIONS_FOR_TEST.store(true, Ordering::SeqCst);
466    }
467
468    fn with_connection<T>(&self, work: impl FnOnce(&mut Connection) -> Result<T>) -> Result<T> {
469        let mut connection = self
470            .connection
471            .lock()
472            .unwrap_or_else(|error| error.into_inner());
473        work(&mut connection)
474    }
475}
476
477fn ensure_record(tx: &rusqlite::Transaction<'_>, key: &BreakerKey) -> Result<()> {
478    tx.execute(
479        "INSERT INTO breaker_records(root_id, domain, corpus_fingerprint, configuration_version)
480         VALUES(?1, ?2, ?3, ?4)
481         ON CONFLICT(root_id, domain, corpus_fingerprint) DO UPDATE SET
482           configuration_version = excluded.configuration_version,
483           zero_credit_deaths = 0,
484           credited_deaths = 0,
485           in_build_burn_ms = 0,
486           suspended_reason = NULL,
487           suspended_since_ms = NULL,
488           suspended_until_ms = NULL
489         WHERE breaker_records.configuration_version != excluded.configuration_version",
490        params![
491            key.root_id,
492            key.domain.as_str(),
493            key.corpus_fingerprint,
494            BREAKER_CONFIGURATION_VERSION,
495        ],
496    )?;
497    Ok(())
498}
499
500fn reset_record(tx: &rusqlite::Transaction<'_>, key: &BreakerKey) -> Result<()> {
501    tx.execute(
502        "UPDATE breaker_records
503         SET zero_credit_deaths = 0, credited_deaths = 0, in_build_burn_ms = 0,
504             suspended_reason = NULL, suspended_since_ms = NULL, suspended_until_ms = NULL
505         WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
506        params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
507    )?;
508    Ok(())
509}
510
511fn suspension_in_tx(
512    tx: &rusqlite::Transaction<'_>,
513    key: &BreakerKey,
514    now_ms: u64,
515) -> Result<Option<BuildSuspension>> {
516    let record = tx.query_row(
517        "SELECT zero_credit_deaths, credited_deaths, in_build_burn_ms,
518                    suspended_reason, suspended_since_ms, suspended_until_ms
519             FROM breaker_records
520             WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
521        params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
522        |row| {
523            Ok((
524                row.get::<_, i64>(0)? as u64,
525                row.get::<_, i64>(1)? as u64,
526                row.get::<_, i64>(2)? as u64,
527                row.get::<_, Option<String>>(3)?,
528                row.get::<_, Option<i64>>(4)?,
529                row.get::<_, Option<i64>>(5)?,
530            ))
531        },
532    )?;
533    let (zero_credit_deaths, credited_deaths, _burn, reason, since, until) = record;
534    let (Some(reason), Some(since), Some(until)) = (reason, since, until) else {
535        return Ok(None);
536    };
537    if until.max(0) as u64 <= now_ms {
538        // TTL permits another probe while retaining all historical tallies.
539        tx.execute(
540            "UPDATE breaker_records
541             SET suspended_reason = NULL, suspended_since_ms = NULL, suspended_until_ms = NULL
542             WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
543            params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
544        )?;
545        return Ok(None);
546    }
547    Ok(Some(BuildSuspension {
548        domain: key.domain,
549        reason,
550        death_count: zero_credit_deaths.saturating_add(credited_deaths),
551        suspended_since_unix_ms: since.max(0) as u64,
552    }))
553}
554
555fn trip_if_needed(
556    tx: &rusqlite::Transaction<'_>,
557    key: &BreakerKey,
558    now_ms: u64,
559) -> Result<Option<BuildSuspension>> {
560    let (zero_credit_deaths, credited_deaths, burn) = tx.query_row(
561        "SELECT zero_credit_deaths, credited_deaths, in_build_burn_ms
562         FROM breaker_records WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
563        params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
564        |row| {
565            Ok((
566                row.get::<_, i64>(0)? as u64,
567                row.get::<_, i64>(1)? as u64,
568                row.get::<_, i64>(2)? as u64,
569            ))
570        },
571    )?;
572    let reason = if zero_credit_deaths >= ZERO_CREDIT_DEATH_LIMIT {
573        Some("zero_credit_death_limit")
574    } else if credited_deaths >= CREDITED_DEATH_LIMIT {
575        Some("credited_death_limit")
576    } else if burn >= IN_BUILD_BURN_LIMIT_MS {
577        Some("in_build_burn_limit")
578    } else {
579        None
580    };
581    let Some(reason) = reason else {
582        return Ok(None);
583    };
584    let existing = tx.query_row(
585        "SELECT suspended_since_ms FROM breaker_records
586             WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
587        params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
588        |row| row.get::<_, Option<i64>>(0),
589    )?;
590    let since = existing
591        .unwrap_or(now_ms.min(i64::MAX as u64) as i64)
592        .max(0) as u64;
593    tx.execute(
594        "UPDATE breaker_records
595         SET suspended_reason = ?4, suspended_since_ms = ?5, suspended_until_ms = ?6
596         WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
597        params![
598            key.root_id,
599            key.domain.as_str(),
600            key.corpus_fingerprint,
601            reason,
602            since as i64,
603            now_ms.saturating_add(TRIP_TTL_MS).min(i64::MAX as u64) as i64,
604        ],
605    )?;
606    Ok(Some(BuildSuspension {
607        domain: key.domain,
608        reason: reason.to_string(),
609        death_count: zero_credit_deaths.saturating_add(credited_deaths),
610        suspended_since_unix_ms: since,
611    }))
612}
613
614fn unix_millis_now() -> u64 {
615    SystemTime::now()
616        .duration_since(UNIX_EPOCH)
617        .unwrap_or_default()
618        .as_millis()
619        .min(u128::from(u64::MAX)) as u64
620}
621
622#[cfg(test)]
623#[path = "build_breaker_audit_tests.rs"]
624mod audit_matrix_tests;
625
626#[cfg(test)]
627mod tests {
628    use super::*;
629    use tempfile::tempdir;
630
631    fn breaker() -> BuildDeathBreaker {
632        BuildDeathBreaker::open(tempdir().unwrap().keep().join("breaker.sqlite")).unwrap()
633    }
634
635    fn key(domain: BuildDomain) -> BreakerKey {
636        BreakerKey::new("root-a", domain, "corpus-a")
637    }
638
639    fn admitted(breaker: &BuildDeathBreaker, key: &BreakerKey, now: u64) -> BuildAttempt {
640        match breaker.admit_at(key, 10, now).unwrap() {
641            BreakerAdmission::Admitted(attempt) => attempt,
642            BreakerAdmission::Suspended(suspension) => {
643                panic!("unexpected suspension: {suspension:?}")
644            }
645        }
646    }
647
648    fn durable_tallies(breaker: &BuildDeathBreaker, key: &BreakerKey) -> (u64, u64, u64) {
649        Connection::open(breaker.path())
650            .unwrap()
651            .query_row(
652                "SELECT zero_credit_deaths, credited_deaths, in_build_burn_ms
653                 FROM breaker_records
654                 WHERE root_id = ?1 AND domain = ?2 AND corpus_fingerprint = ?3",
655                params![key.root_id, key.domain.as_str(), key.corpus_fingerprint],
656                |row| {
657                    Ok((
658                        row.get::<_, i64>(0)? as u64,
659                        row.get::<_, i64>(1)? as u64,
660                        row.get::<_, i64>(2)? as u64,
661                    ))
662                },
663            )
664            .unwrap()
665    }
666
667    #[test]
668    fn three_zero_credit_deaths_trip_once_and_are_idempotent() {
669        let breaker = breaker();
670        let key = key(BuildDomain::CallgraphCold);
671        let mut final_attempt = None;
672        for now in 1..=3 {
673            let attempt = admitted(&breaker, &key, now);
674            let suspension = breaker
675                .record_attributed_death_at(&key, &attempt.attempt_id, 10, 0, now)
676                .unwrap();
677            if now == 3 {
678                final_attempt = Some((attempt, suspension.unwrap()));
679            } else {
680                assert!(suspension.is_none());
681            }
682        }
683        let (attempt, suspension) = final_attempt.unwrap();
684        assert_eq!(suspension.reason, "zero_credit_death_limit");
685        assert_eq!(suspension.death_count, 3);
686        assert_eq!(
687            breaker
688                .record_attributed_death_at(&key, &attempt.attempt_id, 10, 0, 4)
689                .unwrap(),
690            Some(suspension),
691            "recovery may repeat marker cleanup but must not charge it twice"
692        );
693    }
694
695    #[test]
696    fn one_batch_per_death_still_trips_after_six_credited_attempts() {
697        let breaker = breaker();
698        let key = key(BuildDomain::CallgraphCold);
699        for now in 1..=5 {
700            let attempt = admitted(&breaker, &key, now);
701            assert!(breaker
702                .record_attributed_death_at(&key, &attempt.attempt_id, 11, 0, now)
703                .unwrap()
704                .is_none());
705        }
706        assert_eq!(durable_tallies(&breaker, &key), (0, 5, 0));
707        assert!(breaker.suspension_at(&key, 6).unwrap().is_none());
708
709        let sixth = admitted(&breaker, &key, 6);
710        let suspension = breaker
711            .record_attributed_death_at(&key, &sixth.attempt_id, 11, 0, 6)
712            .unwrap()
713            .unwrap();
714        assert_eq!(suspension.reason, "credited_death_limit");
715        assert_eq!(suspension.death_count, 6);
716        assert_eq!(durable_tallies(&breaker, &key), (0, 6, 0));
717    }
718
719    #[test]
720    fn ttl_lifts_only_suspension_and_retains_death_history() {
721        let breaker = breaker();
722        let key = key(BuildDomain::CallgraphCold);
723        for now in 1..=3 {
724            let attempt = admitted(&breaker, &key, now);
725            breaker
726                .record_attributed_death_at(&key, &attempt.attempt_id, 10, 0, now)
727                .unwrap();
728        }
729        assert_eq!(durable_tallies(&breaker, &key), (3, 0, 0));
730
731        let BreakerAdmission::Admitted(retry) =
732            breaker.admit_at(&key, 10, TRIP_TTL_MS + 4).unwrap()
733        else {
734            panic!("expired suspension must admit exactly one probe");
735        };
736        assert_eq!(
737            durable_tallies(&breaker, &key),
738            (3, 0, 0),
739            "TTL expiry lifts scheduling without erasing durable history"
740        );
741
742        let suspension = breaker
743            .record_attributed_death_at(&key, &retry.attempt_id, 10, 0, TRIP_TTL_MS + 5)
744            .unwrap()
745            .unwrap();
746        assert_eq!(suspension.reason, "zero_credit_death_limit");
747        assert_eq!(suspension.death_count, 4);
748        assert_eq!(durable_tallies(&breaker, &key), (4, 0, 0));
749    }
750
751    #[test]
752    fn root_and_domain_histories_are_isolated() {
753        for tripped_domain in BuildDomain::ALL {
754            let breaker = breaker();
755            let tripped = BreakerKey::new(
756                format!("root-{}", tripped_domain.as_str()),
757                tripped_domain,
758                "corpus-a",
759            );
760            for now in 1..=3 {
761                let attempt = admitted(&breaker, &tripped, now);
762                breaker
763                    .record_attributed_death_at(&tripped, &attempt.attempt_id, 10, 0, now)
764                    .unwrap();
765            }
766            let report = breaker.suspension_at(&tripped, 4).unwrap().unwrap();
767            assert_eq!(report.domain, tripped_domain);
768            assert_eq!(report.death_count, 3);
769
770            let mut siblings = BuildDomain::ALL
771                .into_iter()
772                .filter(|domain| *domain != tripped_domain)
773                .map(|domain| BreakerKey::new(tripped.root_id.clone(), domain, "corpus-a"))
774                .collect::<Vec<_>>();
775            siblings.push(BreakerKey::new(
776                format!("{}-sibling", tripped.root_id),
777                tripped_domain,
778                "corpus-a",
779            ));
780
781            for (index, sibling) in siblings.iter().enumerate() {
782                assert!(breaker.suspension_at(sibling, 4).unwrap().is_none());
783                let attempt = admitted(&breaker, sibling, 10 + index as u64);
784                breaker
785                    .record_attributed_death_at(
786                        sibling,
787                        &attempt.attempt_id,
788                        10,
789                        0,
790                        10 + index as u64,
791                    )
792                    .unwrap();
793                assert_eq!(durable_tallies(&breaker, sibling), (1, 0, 0));
794            }
795
796            breaker.explicit_reset(&siblings[0]).unwrap();
797            assert_eq!(durable_tallies(&breaker, &siblings[0]), (0, 0, 0));
798            assert!(breaker.suspension_at(&tripped, 20).unwrap().is_some());
799
800            breaker.explicit_reset(&tripped).unwrap();
801            assert!(breaker.suspension_at(&tripped, 21).unwrap().is_none());
802            assert_eq!(durable_tallies(&breaker, &siblings[1]), (1, 0, 0));
803        }
804    }
805
806    #[test]
807    fn burn_limit_trips_without_counter_credit() {
808        let breaker = breaker();
809        let key = key(BuildDomain::Tier2Scan);
810        let suspension = breaker
811            .record_durable_burn_at(&key, IN_BUILD_BURN_LIMIT_MS, 77)
812            .unwrap()
813            .unwrap();
814        assert_eq!(suspension.reason, "in_build_burn_limit");
815        assert_eq!(suspension.domain, BuildDomain::Tier2Scan);
816    }
817
818    #[test]
819    fn durable_trip_agrees_across_navigation_inspect_and_health_snapshots() {
820        let storage = tempdir().unwrap();
821        let root = storage.path().join("root");
822        std::fs::create_dir_all(&root).unwrap();
823        let project_key = crate::search_index::artifact_cache_key(&root);
824        let breaker_path = storage
825            .path()
826            .join("callgraph")
827            .join(&project_key)
828            .join("build-breaker.sqlite");
829        let breaker = BuildDeathBreaker::open(&breaker_path).unwrap();
830        let key = BreakerKey::new(
831            root.display().to_string(),
832            BuildDomain::CallgraphCold,
833            "corpus-a",
834        );
835        let trip_at = 1_000_000;
836        let mut trip_decision = None;
837        for death_at in (trip_at - 2)..=trip_at {
838            let attempt = admitted(&breaker, &key, death_at);
839            trip_decision = breaker
840                .record_attributed_death_at(&key, &attempt.attempt_id, 10, 0, death_at)
841                .unwrap();
842        }
843        let trip_decision = trip_decision.expect("third zero-credit death trips the breaker");
844        let snapshot_at = trip_at + 5_000;
845        let durable_rows = breaker
846            .active_suspensions_for_root_at(&root.display().to_string(), snapshot_at)
847            .unwrap();
848        assert_eq!(durable_rows, vec![trip_decision.clone()]);
849        let suspension = &durable_rows[0];
850
851        let navigation = crate::commands::callgraph_store_adapter::suspended_response_at(
852            "surface-agreement",
853            "callers",
854            suspension,
855            snapshot_at,
856        );
857        assert_eq!(navigation.data["code"], "build_suspended");
858        assert_eq!(
859            navigation.data["message"],
860            "callers: build_suspended domain=callgraph_cold deaths=3 age_ms=5000 reason=zero_credit_death_limit; run doctor reset-build-breaker to resume"
861        );
862
863        let manager = crate::inspect::InspectManager::new();
864        manager.record_tier2_build_suspension_for_test(
865            crate::inspect::InspectCategory::DeadCode,
866            suspension.clone(),
867        );
868        assert_eq!(
869            manager.tier2_builder_state_detail_at(
870                crate::inspect::InspectCategory::DeadCode,
871                snapshot_at,
872            ),
873            "suspended domain=callgraph_cold deaths=3 age_s=5 reason=zero_credit_death_limit"
874        );
875
876        let config = crate::config::Config {
877            project_root: Some(root.clone()),
878            storage_dir: Some(storage.path().to_path_buf()),
879            ..crate::config::Config::default()
880        };
881        let context = crate::context::AppContext::new(
882            Box::new(crate::parser::TreeSitterProvider::new()),
883            config,
884        );
885        // Contexts without a bound root must not start unrelated heavy work while
886        // this test reads the cached health projection of the durable breaker row.
887        context.set_heavy_root_work_allowed(false);
888        assert_eq!(context.storage_dir(), storage.path());
889        context.refresh_build_suspensions_for_health_at(&root, Some(&project_key), snapshot_at);
890        let health = context.try_health_snapshot(&root);
891        assert_eq!(health.suspended_domains.len(), 1);
892        assert_eq!(health.suspended_domains[0].domain, "callgraph_cold");
893        assert_eq!(
894            health.suspended_domains[0].reason,
895            "zero_credit_death_limit"
896        );
897        assert_eq!(health.suspended_domains[0].death_count, 3);
898        assert_eq!(health.suspended_domains[0].age_s, 5);
899
900        assert!(breaker
901            .active_suspensions_for_root_at(&root.display().to_string(), trip_at + TRIP_TTL_MS + 5,)
902            .unwrap()
903            .is_empty());
904        assert!(matches!(
905            breaker
906                .admit_at(&key, 10, trip_at + TRIP_TTL_MS + 5)
907                .unwrap(),
908            BreakerAdmission::Admitted(_)
909        ));
910    }
911}
aft/build_breaker.rs

aft/
build_breaker.rs
