stackless_core/state/
reaper.rs

1//! Reap-attempt bookkeeping and the per-tick reaper decision (§6).
2//!
3//! The reaper lives in the daemon, but its *decision* — reap, skip,
4//! GC, or wait out a backoff — is a pure function of state-store rows.
5//! Factoring it here keeps it testable without spawning subprocesses
6//! (the daemon's tick is the only thing that shells out), and keeps the
7//! `reap_attempts` table in core so both a reaper-spawned `down` and a
8//! manual `down` (which run the same engine path) clear it on success.
9
10use std::time::Duration;
11
12use super::error::StateError;
13use super::row::Row;
14use super::store::Store;
15
16/// A recorded failed-reap attempt — surfaced in `status`/`list` until a
17/// successful teardown clears it (invariant 4).
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct ReapAttempt {
20    pub instance: String,
21    pub attempts: i64,
22    pub last_error: String,
23    pub next_retry_at: i64,
24}
25
26/// What the reaper should do with one instance this tick.
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub enum ReapDecision {
29    /// Send it through the verified teardown path (the same as `down`).
30    Reap,
31    /// An operation holds the lock (§2/§6) — never reap mid-flight.
32    SkipLocked,
33    /// A prior reap failed and its backoff has not elapsed yet.
34    WaitBackoff { until: i64 },
35}
36
37/// The first backoff after a failure, doubling per attempt up to a cap
38/// (§6: a failed reap retries with backoff). The first retry waits one
39/// tick-ish; the cap bounds runaway instances to hourly attempts.
40const BACKOFF_BASE: Duration = Duration::from_secs(60);
41const BACKOFF_CAP: Duration = Duration::from_secs(3600);
42
43/// The 7-day tombstone GC window (D14): logs/status keep answering
44/// until it expires, then the reaper deletes the row and the logs dir.
45pub const TOMBSTONE_GC_WINDOW: Duration = Duration::from_secs(7 * 24 * 3600);
46
47impl TryFrom<&Row> for ReapAttempt {
48    type Error = StateError;
49
50    fn try_from(row: &Row) -> Result<Self, Self::Error> {
51        Ok(Self {
52            instance: row.get_string(0)?,
53            attempts: row.get_i64(1)?,
54            last_error: row.get_string(2)?,
55            next_retry_at: row.get_i64(3)?,
56        })
57    }
58}
59
60impl ReapAttempt {
61    /// Backoff delay after `attempts` consecutive failures: 60s, 120s,
62    /// 240s, … capped at 1h.
63    pub fn backoff_after(attempts: i64) -> Duration {
64        let shift = attempts.saturating_sub(1).clamp(0, 16) as u32;
65        let secs = BACKOFF_BASE
66            .as_secs()
67            .saturating_mul(1u64.checked_shl(shift).unwrap_or(u64::MAX));
68        Duration::from_secs(secs.min(BACKOFF_CAP.as_secs()))
69    }
70}
71
72impl ReapDecision {
73    /// The pure per-tick decision for one expired instance, given whether a
74    /// live operation holds its lock and its recorded prior failure (if
75    /// any). `now` is unix seconds — the caller's tick clock.
76    pub fn decide(now: i64, lock_held: bool, prior: Option<&ReapAttempt>) -> Self {
77        if lock_held {
78            return Self::SkipLocked;
79        }
80        match prior {
81            Some(attempt) if attempt.next_retry_at > now => Self::WaitBackoff {
82                until: attempt.next_retry_at,
83            },
84            _ => Self::Reap,
85        }
86    }
87}
88
89impl Store {
90    /// Record a failed reap, advancing the backoff. `attempts`
91    /// increments; `next_retry_at` is now + the doubled delay.
92    pub fn record_reap_failure(&self, instance: &str, error: &str) -> Result<(), StateError> {
93        let now = Self::now();
94        let attempts = self
95            .reap_attempt(instance)?
96            .map(|a| a.attempts + 1)
97            .unwrap_or(1);
98        let next_retry_at = now + ReapAttempt::backoff_after(attempts).as_secs() as i64;
99        self.execute(
100            "INSERT INTO reap_attempts (instance, attempts, last_error, next_retry_at)
101             VALUES (?1, ?2, ?3, ?4)
102             ON CONFLICT(instance) DO UPDATE SET
103               attempts = excluded.attempts,
104               last_error = excluded.last_error,
105               next_retry_at = excluded.next_retry_at",
106            &[
107                instance.into(),
108                attempts.into(),
109                error.into(),
110                next_retry_at.into(),
111            ],
112        )?;
113        Ok(())
114    }
115
116    /// Clear an instance's reap-failure record — a successful reap or a
117    /// successful manual `down` calls this through the engine.
118    pub fn clear_reap_failure(&self, instance: &str) -> Result<(), StateError> {
119        self.execute(
120            "DELETE FROM reap_attempts WHERE instance = ?1",
121            &[instance.into()],
122        )?;
123        Ok(())
124    }
125
126    pub fn reap_attempt(&self, instance: &str) -> Result<Option<ReapAttempt>, StateError> {
127        self.query_row(
128            "SELECT instance, attempts, last_error, next_retry_at
129             FROM reap_attempts WHERE instance = ?1",
130            &[instance.into()],
131            |row| ReapAttempt::try_from(row),
132        )
133    }
134
135    /// Tombstoned instances whose GC window has elapsed — the reaper
136    /// deletes the row (FK cascade cleans leases/locks/checkpoints) and
137    /// removes the logs dir (D14).
138    pub fn gc_due_tombstones(&self) -> Result<Vec<String>, StateError> {
139        let cutoff = Self::now() - TOMBSTONE_GC_WINDOW.as_secs() as i64;
140        self.query_map(
141            "SELECT name FROM instances
142             WHERE status = 'tombstoned' AND tombstoned_at IS NOT NULL
143               AND tombstoned_at <= ?1 ORDER BY name",
144            &[cutoff.into()],
145            |row: &Row| row.get_string(0),
146        )
147    }
148
149    /// Delete an instance row outright (the GC step). FK cascade removes
150    /// its leases, locks, checkpoints, and reap-attempt row.
151    pub fn delete_instance(&self, instance: &str) -> Result<(), StateError> {
152        self.execute("DELETE FROM instances WHERE name = ?1", &[instance.into()])?;
153        Ok(())
154    }
155}
156
157#[cfg(test)]
158mod tests {
159    use super::*;
160
161    #[test]
162    fn locked_instance_is_skipped() {
163        assert_eq!(
164            ReapDecision::decide(100, true, None),
165            ReapDecision::SkipLocked
166        );
167    }
168
169    #[test]
170    fn unlocked_with_no_prior_is_reaped() {
171        assert_eq!(ReapDecision::decide(100, false, None), ReapDecision::Reap);
172    }
173
174    #[test]
175    fn backoff_not_elapsed_waits() {
176        let prior = ReapAttempt {
177            instance: "x".into(),
178            attempts: 1,
179            last_error: "boom".into(),
180            next_retry_at: 200,
181        };
182        assert_eq!(
183            ReapDecision::decide(100, false, Some(&prior)),
184            ReapDecision::WaitBackoff { until: 200 }
185        );
186    }
187
188    #[test]
189    fn backoff_elapsed_reaps_again() {
190        let prior = ReapAttempt {
191            instance: "x".into(),
192            attempts: 3,
193            last_error: "boom".into(),
194            next_retry_at: 50,
195        };
196        assert_eq!(
197            ReapDecision::decide(100, false, Some(&prior)),
198            ReapDecision::Reap
199        );
200    }
201
202    #[test]
203    fn backoff_doubles_and_caps() {
204        assert_eq!(ReapAttempt::backoff_after(1), Duration::from_secs(60));
205        assert_eq!(ReapAttempt::backoff_after(2), Duration::from_secs(120));
206        assert_eq!(ReapAttempt::backoff_after(3), Duration::from_secs(240));
207        assert_eq!(ReapAttempt::backoff_after(100), Duration::from_secs(3600));
208    }
209}
stackless_core/state/reaper.rs

stackless_core/state/
reaper.rs
