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

aft/
build_breaker.rs
