Skip to main content

mitosis/
lifecycle.rs

1use crate::clones::{
2    destroy_for, install_for, mark, restart_or_undo, sleep, stop_and_remove, suspect_warning,
3};
4use crate::ctx::{Ctx, InFlightOp};
5use crate::error::{ErrorCode, MitosisError, Result};
6use crate::events::{self, reason};
7use crate::names::{is_clone_id, random_hex};
8use crate::pg::{control_state, psql, running_server, stop, ControlState, RunningPg};
9use crate::proc_util::{has_exited, is_alive, process_start_time};
10use crate::protocol::{GcFailure, GcReport};
11use crate::state::{CloneRow, CloneStatus};
12use crate::templates::{
13    folder_stem, is_build_socket_name, now, parse_gen_dir, parse_leftover, Leftover,
14};
15#[cfg(target_os = "macos")]
16use nix::errno::Errno;
17#[cfg(target_os = "macos")]
18use nix::sys::event::{EvFlags, EventFilter, FilterFlag, KEvent, Kqueue};
19use std::collections::HashSet;
20use std::fs::OpenOptions;
21use std::io::{ErrorKind, Write};
22use std::path::{Path, PathBuf};
23use std::sync::Arc;
24use std::time::Duration;
25
26/// Allowed difference, in seconds, between a recorded and a measured process start time.
27const START_SLACK: u64 = 1;
28/// How long a parent watcher keeps retrying a clone that is still starting (a slow create).
29const STARTING_WAIT: Duration = Duration::from_secs(60);
30const DESTROY_RETRY_DELAY: Duration = Duration::from_millis(500);
31/// How often a parent watcher wakes to check that its clone still exists.
32#[cfg(target_os = "macos")]
33const WATCH_RECHECK: Duration = Duration::from_secs(30);
34/// gc drops a branch template no clone uses after this long without a `clone --branch` for it.
35const BRANCH_UNUSED_SECS: i64 = 14 * 86_400;
36
37fn internal(msg: impl Into<String>) -> MitosisError {
38    MitosisError::new(ErrorCode::Internal, msg)
39}
40
41/// Appends one line to `logs/daemon.log`; lifecycle work has no caller to report errors to.
42pub(crate) fn log(ctx: &Ctx, msg: &str) {
43    if let Ok(mut f) = OpenOptions::new()
44        .create(true)
45        .append(true)
46        .open(ctx.home.logs_dir().join("daemon.log"))
47    {
48        let _ = writeln!(f, "[{}] {msg}", now());
49    }
50}
51
52fn start_matches(pid: i32, start_time: u64) -> bool {
53    process_start_time(pid).is_some_and(|t| t.abs_diff(start_time) <= START_SLACK)
54}
55
56/// True while `pid` runs and, when a start time was recorded, is still the same process.
57pub fn parent_alive(pid: i32, start_time: Option<u64>) -> bool {
58    is_alive(pid) && start_time.is_none_or(|t| start_matches(pid, t))
59}
60
61/// True while the clone's recorded postmaster still runs: same pid, exactly the same kernel start
62/// time (both recorded from the kernel at create). A dead child is reaped, not counted.
63fn postmaster_running(row: &CloneRow) -> bool {
64    match (row.pg_pid, row.pg_start_time) {
65        (Some(pid), Some(t)) => !has_exited(pid) && process_start_time(pid) == Some(t),
66        _ => false,
67    }
68}
69
70fn failed(report: &mut GcReport, item: impl Into<String>, e: &MitosisError) {
71    report.failed.push(GcFailure {
72        item: item.into(),
73        error: e.message.clone(),
74    });
75}
76
77fn in_flight(ctx: &Ctx, id: &str) -> bool {
78    ctx.in_flight
79        .lock()
80        .unwrap_or_else(|e| e.into_inner())
81        .contains_key(id)
82}
83
84/// A name's port and password are kept this long after its last create.
85const RESERVATION_KEEP_SECS: i64 = 30 * 24 * 3600;
86
87/// Destroys Ready and Sleeping clones whose TTL has run out; returns the ids destroyed.
88pub async fn reap_expired(ctx: &Ctx, now: i64) -> Result<Vec<String>> {
89    let rows = ctx.store().list_clones()?;
90    let mut reaped = Vec::new();
91    // Creating/Starting rows belong to an in-flight create; the next pass gets them.
92    for row in rows.iter().filter(|r| {
93        matches!(r.status, CloneStatus::Ready | CloneStatus::Sleeping)
94            && !r.protected
95            && r.expires_at <= now
96    }) {
97        if in_flight(ctx, &row.id) {
98            continue;
99        }
100        match destroy_for(ctx, &row.id, reason::TTL).await {
101            Ok(id) => reaped.push(id),
102            Err(e) if e.code == ErrorCode::CloneNotFound => {}
103            Err(e) => log(ctx, &format!("reaper: clone {}: {e}", row.id)),
104        }
105    }
106    Ok(reaped)
107}
108
109/// Client backends other than this check's own session, from `pg_stat_activity`: the users of
110/// the direct port, whom the router cannot see. Blocking work runs off the runtime.
111pub(crate) async fn client_backends(ctx: &Ctx, row: &CloneRow) -> Result<usize> {
112    let install = install_for(ctx, &row.dir).ok_or_else(|| {
113        internal(format!(
114            "no Postgres install found for {}",
115            row.dir.display()
116        ))
117    })?;
118    let (sock, port, db) = (row.socket_dir.clone(), row.port, row.database.clone());
119    let out = tokio::task::spawn_blocking(move || {
120        psql(
121            &install,
122            &sock,
123            port,
124            &db,
125            "select count(*) from pg_stat_activity \
126             where backend_type = 'client backend' and pid <> pg_backend_pid()",
127        )
128    })
129    .await
130    .map_err(|e| internal(format!("background task failed: {e}")))??;
131    out.trim()
132        .parse()
133        .map_err(|_| internal(format!("unexpected pg_stat_activity count {out:?}")))
134}
135
136/// Ready and Sleeping clones (no operation running) with no connections right now, each with
137/// the time its idle spell began: the latest of its last router activity, the last time a check
138/// saw a client backend, its creation and this daemon's start. A Ready clone with a client
139/// backend counts as active at `now` and is left out. Clones are read from the store, so router
140/// entries of clones that are gone play no part.
141pub(crate) async fn idle_clones(ctx: &Ctx, now: i64) -> Result<Vec<(CloneRow, i64)>> {
142    // From the wall clock, not `now`: a caller may pass a later `now` to look ahead.
143    let started = crate::templates::now()
144        .saturating_sub(i64::try_from(ctx.started_at.elapsed().as_secs()).unwrap_or(0));
145    let rows = ctx.store().list_clones()?;
146    let mut idle = Vec::new();
147    for row in rows
148        .into_iter()
149        .filter(|r| matches!(r.status, CloneStatus::Ready | CloneStatus::Sleeping))
150    {
151        if in_flight(ctx, &row.id) {
152            continue;
153        }
154        let (conns, last) = {
155            let r = ctx.router();
156            (r.open_conns(&row.id), r.last_active(&row.id))
157        };
158        if conns > 0 {
159            continue;
160        }
161        if row.status == CloneStatus::Ready {
162            match client_backends(ctx, &row).await {
163                Ok(0) => {}
164                Ok(_) => {
165                    ctx.router().last_active.insert(row.id.clone(), now);
166                    continue;
167                }
168                Err(e) => {
169                    // A crashed server fails the check every pass until gc recovers it: say it once
170                    // there, not here every 30 s.
171                    if postmaster_running(&row) {
172                        log(ctx, &format!("idle check of clone {}: {e}", row.id));
173                    }
174                    continue;
175                }
176            }
177        }
178        let since = last.unwrap_or(0).max(row.created_at).max(started);
179        idle.push((row, since));
180    }
181    Ok(idle)
182}
183
184/// The reaper's idle pass: destroys unnamed, unprotected clones idle for `idle_reap` (reason
185/// `idle`), then puts Ready clones idle for `idle_sleep` to sleep. Sleeping needs the router
186/// listening (it is what wakes them). Returns `(slept, destroyed)` ids.
187pub async fn idle_pass(ctx: &Ctx, now: i64) -> Result<(Vec<String>, Vec<String>)> {
188    let cfg = ctx.config();
189    let sleep_after = cfg.idle_sleep.filter(|_| crate::router::listening(ctx));
190    let (mut slept, mut reaped) = (Vec::new(), Vec::new());
191    if sleep_after.is_none() && cfg.idle_reap.is_none() {
192        return Ok((slept, reaped));
193    }
194    let longer = |idle: i64, d: Option<Duration>| {
195        d.is_some_and(|d| idle >= i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
196    };
197    for (row, since) in idle_clones(ctx, now).await? {
198        let idle = now.saturating_sub(since);
199        if row.name.is_none() && !row.protected && longer(idle, cfg.idle_reap) {
200            match destroy_for(ctx, &row.id, reason::IDLE).await {
201                Ok(id) => reaped.push(id),
202                Err(e) if e.code == ErrorCode::CloneNotFound => {}
203                Err(e) => log(ctx, &format!("idle reap: clone {}: {e}", row.id)),
204            }
205        } else if row.status == CloneStatus::Ready && longer(idle, sleep_after) {
206            match sleep(ctx, &row.id, reason::IDLE).await {
207                Ok(()) => slept.push(row.id),
208                Err(e) => log(ctx, &format!("idle sleep: clone {}: {e}", row.id)),
209            }
210        }
211    }
212    Ok((slept, reaped))
213}
214
215/// Watches `parent_pid` on a thread; when it exits, destroys the clone on the current tokio runtime.
216/// The thread ends early once the clone is gone or being destroyed some other way.
217/// Must be called from within the runtime. A second watch for the same clone is a no-op.
218pub fn watch_parent(ctx: Arc<Ctx>, clone_id: String, parent_pid: i32) {
219    let parent_start_time = match ctx.store().get_clone(&clone_id) {
220        Ok(Some(row)) => row.parent_start_time,
221        Ok(None) => return,
222        Err(e) => {
223            log(&ctx, &format!("parent watch for clone {clone_id}: {e}"));
224            None
225        }
226    };
227    let handle = match tokio::runtime::Handle::try_current() {
228        Ok(h) => h,
229        Err(e) => {
230            return log(
231                &ctx,
232                &format!("parent watch for clone {clone_id}: no tokio runtime: {e}"),
233            )
234        }
235    };
236    if !ctx
237        .watching
238        .lock()
239        .unwrap_or_else(|e| e.into_inner())
240        .insert(clone_id.clone())
241    {
242        return;
243    }
244    let watch = Watch { ctx, id: clone_id };
245    std::thread::spawn(move || {
246        if wait_for_parent_exit(&watch, parent_pid, parent_start_time) {
247            // If the runtime is gone the future is dropped, and `watch` with it.
248            handle.spawn(destroy_after_parent_exit(watch));
249        }
250    });
251}
252
253/// One parent watcher; leaves `Ctx::watching` when dropped, on every exit path.
254struct Watch {
255    ctx: Arc<Ctx>,
256    id: String,
257}
258
259impl Drop for Watch {
260    fn drop(&mut self) {
261        self.ctx
262            .watching
263            .lock()
264            .unwrap_or_else(|e| e.into_inner())
265            .remove(&self.id);
266    }
267}
268
269/// False once the clone is gone or already being torn down, so its watcher can stop.
270fn still_watched(ctx: &Ctx, id: &str) -> bool {
271    if ctx
272        .in_flight
273        .lock()
274        .unwrap_or_else(|e| e.into_inner())
275        .get(id)
276        == Some(&InFlightOp::Destroy)
277    {
278        return false;
279    }
280    match ctx.store().get_clone(id) {
281        // `get_clone` also matches names; a row with another id means ours is gone.
282        Ok(Some(row)) => row.id == id && row.status != CloneStatus::Stopping,
283        Ok(None) => false,
284        Err(_) => true, // keep watching; the next check may succeed
285    }
286}
287
288#[cfg(target_os = "macos")]
289/// Registers a NOTE_EXIT watch on `pid`; `None` when the pid is already gone.
290fn register_exit(pid: i32) -> nix::Result<Option<Kqueue>> {
291    let kq = Kqueue::new()?;
292    let ev = KEvent::new(
293        pid as usize,
294        EventFilter::EVFILT_PROC,
295        EvFlags::EV_ADD | EvFlags::EV_ONESHOT,
296        FilterFlag::NOTE_EXIT,
297        0,
298        0,
299    );
300    // An empty event list makes kevent only register, reporting failures as its own error.
301    match kq.kevent(&[ev], &mut [], None) {
302        Err(Errno::ESRCH) => Ok(None),
303        Err(e) => Err(e),
304        Ok(_) => Ok(Some(kq)),
305    }
306}
307
308/// Blocks until the parent exits (true), or until the clone no longer needs watching (false).
309/// Linux has no kqueue: this polls once a second.
310#[cfg(not(target_os = "macos"))]
311fn wait_for_parent_exit(watch: &Watch, pid: i32, start_time: Option<u64>) -> bool {
312    loop {
313        if !parent_alive(pid, start_time) {
314            return true;
315        }
316        if !still_watched(&watch.ctx, &watch.id) {
317            return false;
318        }
319        std::thread::sleep(Duration::from_secs(1));
320    }
321}
322
323/// Blocks until the parent exits (true), or until the clone no longer needs watching (false).
324/// Uses kqueue, waking every `WATCH_RECHECK`; polls once a second if kqueue is unavailable.
325#[cfg(target_os = "macos")]
326fn wait_for_parent_exit(watch: &Watch, pid: i32, start_time: Option<u64>) -> bool {
327    let mut kq = match register_exit(pid) {
328        Ok(None) => return true,
329        Ok(Some(kq)) => Some(kq),
330        Err(e) => {
331            log(
332                &watch.ctx,
333                &format!(
334                    "parent watch for clone {}: kqueue failed ({e}); polling",
335                    watch.id
336                ),
337            );
338            None
339        }
340    };
341    let recheck = libc::timespec {
342        tv_sec: WATCH_RECHECK.as_secs() as libc::time_t,
343        tv_nsec: 0,
344    };
345    let mut out = [KEvent::new(
346        0,
347        EventFilter::EVFILT_PROC,
348        EvFlags::empty(),
349        FilterFlag::empty(),
350        0,
351        0,
352    )];
353    loop {
354        // Also covers an exit just before registering, and a pid reused by another process.
355        if !parent_alive(pid, start_time) {
356            return true;
357        }
358        if !still_watched(&watch.ctx, &watch.id) {
359            return false;
360        }
361        match &kq {
362            Some(q) => match q.kevent(&[], &mut out, Some(recheck)) {
363                Ok(n) if n > 0 => return true,
364                Ok(_) | Err(Errno::EINTR) => {}
365                Err(e) => {
366                    log(
367                        &watch.ctx,
368                        &format!(
369                            "parent watch for clone {}: kevent failed ({e}); polling",
370                            watch.id
371                        ),
372                    );
373                    kq = None;
374                }
375            },
376            None => std::thread::sleep(Duration::from_secs(1)),
377        }
378    }
379}
380
381async fn destroy_after_parent_exit(watch: Watch) {
382    let (ctx, id) = (&watch.ctx, &watch.id);
383    let deadline = tokio::time::Instant::now() + STARTING_WAIT;
384    loop {
385        match destroy_for(ctx, id, reason::PARENT).await {
386            Ok(_) => break,
387            Err(e) if e.code == ErrorCode::CloneNotFound => break,
388            // destroy refuses a clone whose create has not finished; a slow create can take a while.
389            Err(e)
390                if e.code == ErrorCode::InvalidRequest
391                    && tokio::time::Instant::now() < deadline =>
392            {
393                tokio::time::sleep(DESTROY_RETRY_DELAY).await
394            }
395            Err(e) => {
396                log(
397                    ctx,
398                    &format!("parent of clone {id} exited, but destroy failed: {e}"),
399                );
400                break;
401            }
402        }
403    }
404}
405
406/// Brings the store and the home folder back in line after a crash; safe while requests are served.
407/// Clones with an in-flight create and templates with a running build are left alone.
408/// A failure on one entry is logged to `logs/daemon.log` and the rest still run. Sleeping clones
409/// that nothing would wake (router not listening) are woken at the end, within the budget.
410pub async fn reconcile(ctx: Arc<Ctx>, now: i64) -> Result<GcReport> {
411    let (mut report, to_wake) = reconcile_deferred(ctx.clone(), now).await?;
412    report.warnings.extend(wake_sleepers(&ctx, to_wake).await);
413    Ok(report)
414}
415
416/// `reconcile` without the wakes: also returns the sleeping clones to wake, for the daemon's
417/// startup gc, which hands them to `wake_sleepers` on a task once its socket accepts.
418pub async fn reconcile_deferred(ctx: Arc<Ctx>, now: i64) -> Result<(GcReport, Vec<String>)> {
419    let mut report = GcReport::default();
420    let mut to_wake = Vec::new();
421    let ids: Vec<String> = ctx
422        .store()
423        .list_clones()?
424        .into_iter()
425        .map(|r| r.id)
426        .collect();
427    for id in ids {
428        if in_flight(&ctx, &id) {
429            continue;
430        }
431        // Re-read after the in-flight check: a create that just finished shows its final status.
432        let row = match ctx.store().get_clone(&id)? {
433            Some(row) if row.id == id => row,
434            _ => continue,
435        };
436        if let Err(e) = reconcile_clone(&ctx, &row, now, &mut report, &mut to_wake).await {
437            log(&ctx, &format!("gc: clone {id}: {e}"));
438            failed(&mut report, id, &e);
439        }
440    }
441    // Before the orphan pass, which then removes the dropped templates' generation folders.
442    drop_branch_templates(&ctx, now, &mut report).await;
443    remove_orphans(&ctx, &mut report).await;
444    emit_kept(&ctx);
445    drop_unused_reservations(&ctx, now);
446    events::emit(
447        &ctx,
448        events::GC,
449        None,
450        None,
451        serde_json::json!({
452            "adopted": report.adopted.len(),
453            "destroyed": report.destroyed.len(),
454            "orphan_folders_removed": report.orphan_folders_removed.len(),
455            "records_dropped": report.records_dropped.len(),
456            "failed": report.failed.len(),
457        }),
458    );
459    Ok((report, to_wake))
460}
461
462/// Wakes the sleeping clones `ids` (from `reconcile_deferred`) one by one, most recently active
463/// first, while the router is not listening and the budget has room for one more running server;
464/// returns the warnings (also logged). The rest stay asleep with one warning naming them.
465pub async fn wake_sleepers(ctx: &Arc<Ctx>, ids: Vec<String>) -> Vec<String> {
466    let mut warnings = Vec::new();
467    let mut rows: Vec<(CloneRow, std::time::SystemTime)> = Vec::new();
468    for id in ids {
469        let row = match ctx.store().get_clone(&id) {
470            Ok(Some(row)) if row.status == CloneStatus::Sleeping => row,
471            _ => continue,
472        };
473        // A data folder's mtime moves when its server stops (postmaster.pid goes): the sleep
474        // time, which follows the clone's last activity.
475        let dir = row.dir.clone();
476        let stopped = tokio::task::spawn_blocking(move || std::fs::metadata(dir)?.modified())
477            .await
478            .ok()
479            .and_then(|r| r.ok())
480            .unwrap_or(std::time::UNIX_EPOCH);
481        rows.push((row, stopped));
482    }
483    rows.sort_by_key(|(_, stopped)| std::cmp::Reverse(*stopped));
484    let mut left: Vec<String> = Vec::new();
485    let mut why = String::new();
486    for (row, _) in rows {
487        let label = row.name.clone().unwrap_or_else(|| row.id.clone());
488        if !left.is_empty() {
489            left.push(label);
490            continue;
491        }
492        // The router came up meanwhile (its bind retry): a connection wakes the rest.
493        if crate::router::listening(ctx) {
494            break;
495        }
496        match crate::budget::wake_blocked(ctx).await {
497            Ok(None) => {}
498            Ok(Some(limit)) => why = format!("waking more would exceed {limit}"),
499            Err(e) => why = format!("the budget check failed: {e}"),
500        }
501        if !why.is_empty() {
502            left.push(label);
503            continue;
504        }
505        match crate::clones::wake(ctx, &row.id).await {
506            Ok((woken, ws)) => {
507                let suspect = woken.suspect.then(|| suspect_warning(&woken));
508                for w in ws.into_iter().chain(suspect) {
509                    log(ctx, &format!("gc: clone {}: {w}", row.id));
510                    warnings.push(w);
511                }
512            }
513            Err(e) => {
514                let w = format!(
515                    "clone {label} sleeps and could not be started ({e}); mitosis wake {label} \
516                     starts it"
517                );
518                log(ctx, &format!("gc: {w}"));
519                warnings.push(w);
520            }
521        }
522    }
523    if !left.is_empty() {
524        let w = format!(
525            "sleeping clones left asleep ({why}): {}; mitosis wake <clone> (or mitosis env \
526             <clone> --direct) starts one",
527            left.join(", ")
528        );
529        log(ctx, &format!("gc: {w}"));
530        warnings.push(w);
531    }
532    warnings
533}
534
535/// True when git says `branch` does not exist in the repository of `dir` (`rev-parse --verify`
536/// exits 1); false when it exists or git cannot tell. Blocking.
537fn branch_deleted(dir: &Path, branch: &str) -> bool {
538    std::process::Command::new("git")
539        .arg("-C")
540        .arg(dir)
541        .args(["rev-parse", "--verify", "--quiet"])
542        .arg(format!("refs/heads/{branch}"))
543        .stdin(std::process::Stdio::null())
544        .stdout(std::process::Stdio::null())
545        .stderr(std::process::Stdio::null())
546        .status()
547        .is_ok_and(|s| s.code() == Some(1))
548}
549
550/// Drops the rows of branch templates no clone uses whose branch is gone from the base's
551/// repository or that were not used for `BRANCH_UNUSED_SECS`; logs, never fails.
552async fn drop_branch_templates(ctx: &Ctx, now: i64, report: &mut GcReport) {
553    let rows = match ctx.store().list_templates() {
554        Ok(ts) => ts.into_iter().filter(|t| t.base.is_some()),
555        Err(e) => return log(ctx, &format!("gc: list templates: {e}")),
556    };
557    for t in rows {
558        let (Some(base), Some(branch)) = (t.base.clone(), t.branch.clone()) else {
559            continue;
560        };
561        let unused = t
562            .last_used
563            .is_none_or(|u| now.saturating_sub(u) >= BRANCH_UNUSED_SECS);
564        let repo = ctx
565            .store()
566            .get_template(&base)
567            .ok()
568            .flatten()
569            .and_then(|b| b.migrate)
570            .map(|m| m.cwd);
571        let gone = match repo.filter(|_| !unused) {
572            Some(dir) => tokio::task::spawn_blocking(move || branch_deleted(&dir, &branch))
573                .await
574                .unwrap_or(false),
575            None => false,
576        };
577        if !unused && !gone {
578            continue;
579        }
580        // Under the write guard: no clone creation is between reading the template and cloning it.
581        let _swap = ctx.template_swap.write().await;
582        if ctx
583            .building
584            .lock()
585            .unwrap_or_else(|e| e.into_inner())
586            .contains(&t.name)
587        {
588            continue;
589        }
590        // Re-read under the guard: a `clone --branch` may have used it since the first read.
591        let fresh = match ctx.store().get_template(&t.name) {
592            Ok(Some(f)) => f,
593            Ok(None) => continue,
594            Err(e) => {
595                log(ctx, &format!("gc: drop template {}: {e}", t.name));
596                failed(report, t.name.clone(), &e);
597                continue;
598            }
599        };
600        let unused = fresh
601            .last_used
602            .is_none_or(|u| now.saturating_sub(u) >= BRANCH_UNUSED_SECS);
603        if !unused && !gone {
604            continue;
605        }
606        let res = {
607            let store = ctx.store();
608            match store.list_clones() {
609                Ok(cs) if cs.iter().any(|c| c.template == t.name) => continue,
610                Ok(_) => store.delete_template(&t.name),
611                Err(e) => Err(e),
612            }
613        };
614        match res {
615            Ok(()) => {
616                let why = if gone { "branch deleted" } else { "unused" };
617                log(ctx, &format!("gc: dropped template {} ({why})", t.name));
618                let build_log = crate::templates::build_log_path(&ctx.home, &t.name);
619                // `remove_path` ignores NotFound and names the path in its error.
620                if let Ok(Err(e)) =
621                    tokio::task::spawn_blocking(move || remove_path(&build_log)).await
622                {
623                    log(ctx, &format!("gc: {e}"));
624                }
625                events::emit(
626                    ctx,
627                    events::TEMPLATE_PRUNED,
628                    None,
629                    Some(&t.name),
630                    serde_json::json!({ "dropped": true, "cause": why }),
631                );
632                report.records_dropped.push(format!("template {}", t.name));
633            }
634            Err(e) => {
635                log(ctx, &format!("gc: drop template {}: {e}", t.name));
636                failed(report, t.name.clone(), &e);
637            }
638        }
639    }
640}
641
642/// Drops name reservations unused for 30 days whose name has no clone; logs, never fails. Run by
643/// gc and, for a long-running daemon, once a day by the reaper.
644pub fn drop_unused_reservations(ctx: &Ctx, now: i64) {
645    match ctx
646        .store()
647        .drop_stale_reservations(now.saturating_sub(RESERVATION_KEEP_SECS))
648    {
649        Ok(names) if !names.is_empty() => log(
650            ctx,
651            &format!("gc: dropped unused reservations: {}", names.join(", ")),
652        ),
653        Err(e) => log(ctx, &format!("gc: reservations: {e}")),
654        _ => {}
655    }
656}
657
658async fn reconcile_clone(
659    ctx: &Arc<Ctx>,
660    row: &CloneRow,
661    now: i64,
662    report: &mut GcReport,
663    to_wake: &mut Vec<String>,
664) -> Result<()> {
665    let dir = ctx.home.clones_dir().join(&row.id);
666    if !dir.exists() {
667        // A destroy that started since the in-flight check owns it now.
668        let Ok(_mark) = mark(ctx, &row.id, InFlightOp::Destroy) else {
669            return Ok(());
670        };
671        stop_and_remove(ctx, row, reason::GC).await?; // also clears its socket folder and log
672        report.records_dropped.push(row.id.clone());
673        return Ok(());
674    }
675    if row.status == CloneStatus::Sleeping {
676        return reconcile_sleeping(ctx, row, now, report, to_wake).await;
677    }
678    let running = row.status == CloneStatus::Ready && postmaster_running(row);
679    // Whether the machine rebooted or the server just died (a full disk, a manual stop, a
680    // logout), a clone someone may want back is restarted, never destroyed outright.
681    let kept = row.name.is_some() || row.protected || row.parent_pid.is_none();
682    if row.status == CloneStatus::Ready && !running && kept {
683        return recover_dead(ctx, row, now, report).await;
684    }
685    let keep = running
686        && row.expires_at > now
687        && row
688            .parent_pid
689            .is_none_or(|p| parent_alive(p, row.parent_start_time));
690    if !keep {
691        let Ok(_mark) = mark(ctx, &row.id, InFlightOp::Destroy) else {
692            return Ok(());
693        };
694        // Stops only a server identified exactly (recorded pid + start time, or cwd in `dir`).
695        stop_and_remove(ctx, row, reason::GC).await?;
696        report.destroyed.push(row.id.clone());
697        return Ok(());
698    }
699    report.adopted.push(row.id.clone());
700    if let Some(pid) = row.parent_pid {
701        watch_parent(ctx.clone(), row.id.clone(), pid);
702    }
703    Ok(())
704}
705
706/// A sleeping clone keeps sleeping under the TTL and parent rules of a running one. A server
707/// left running by a wake or sleep that crashed half-way is stopped, so Sleeping means stopped.
708/// While the router is not listening (off, or its port taken) nothing would wake it on connect,
709/// so it goes on `to_wake` (see `wake_sleepers`).
710async fn reconcile_sleeping(
711    ctx: &Arc<Ctx>,
712    row: &CloneRow,
713    now: i64,
714    report: &mut GcReport,
715    to_wake: &mut Vec<String>,
716) -> Result<()> {
717    let parent_gone = row
718        .parent_pid
719        .is_some_and(|p| !parent_alive(p, row.parent_start_time));
720    if parent_gone || row.expires_at <= now {
721        let Ok(_mark) = mark(ctx, &row.id, InFlightOp::Destroy) else {
722            return Ok(());
723        };
724        stop_and_remove(ctx, row, reason::GC).await?;
725        report.destroyed.push(row.id.clone());
726        return Ok(());
727    }
728    if postmaster_running(row) {
729        let Ok(_mark) = mark(ctx, &row.id, InFlightOp::Create) else {
730            return Ok(());
731        };
732        let install = install_for(ctx, &row.dir).ok_or_else(|| {
733            internal(format!(
734                "no Postgres install found for {}",
735                row.dir.display()
736            ))
737        })?;
738        let expected = row
739            .pg_pid
740            .zip(row.pg_start_time)
741            .map(|(pid, start_time)| RunningPg { pid, start_time });
742        let (dir, t) = (row.dir.clone(), ctx.config().stop_timeout);
743        tokio::task::spawn_blocking(move || stop(&install, &dir, expected, t))
744            .await
745            .map_err(|e| internal(format!("background task failed: {e}")))??;
746        let mut row = row.clone();
747        (row.pg_pid, row.pg_start_time) = (None, None);
748        ctx.store().update_clone(&row)?;
749    }
750    if !crate::router::listening(ctx) {
751        to_wake.push(row.id.clone());
752    }
753    report.adopted.push(row.id.clone());
754    if let Some(pid) = row.parent_pid {
755        watch_parent(ctx.clone(), row.id.clone(), pid);
756    }
757    Ok(())
758}
759
760/// True when a non-durable clone's cluster did not shut down cleanly (fsync was off, so recent
761/// writes may be lost). Durable clones are never unclean in this sense. The server must be down.
762pub(crate) async fn unclean(ctx: &Ctx, row: &CloneRow) -> Result<bool> {
763    if row.durable {
764        return Ok(false);
765    }
766    let install = install_for(ctx, &row.dir).ok_or_else(|| {
767        internal(format!(
768            "no Postgres install found for {}",
769            row.dir.display()
770        ))
771    })?;
772    let dir = row.dir.clone();
773    let state = tokio::task::spawn_blocking(move || control_state(&install, &dir))
774        .await
775        .map_err(|e| internal(format!("background task failed: {e}")))??;
776    Ok(state != ControlState::ShutDown)
777}
778
779/// Records that the clone may hold lost or corrupt writes.
780pub(crate) fn mark_suspect(ctx: &Ctx, row: &mut CloneRow) -> Result<()> {
781    row.suspect = true;
782    ctx.store().update_clone(row)?;
783    events::emit_clone(
784        ctx,
785        events::CLONE_SUSPECT,
786        row,
787        serde_json::json!({ "cause": "unclean shutdown with fsync off" }),
788    );
789    Ok(())
790}
791
792/// Destroys a clone gc decided to drop, unless another operation owns it now.
793async fn drop_clone(ctx: &Ctx, row: &CloneRow, report: &mut GcReport) -> Result<()> {
794    let Ok(_mark) = mark(ctx, &row.id, InFlightOp::Destroy) else {
795        return Ok(());
796    };
797    stop_and_remove(ctx, row, reason::RECOVERY).await?;
798    report.destroyed.push(row.id.clone());
799    Ok(())
800}
801
802/// A named, protected or parentless Ready clone whose server is gone (a reboot or a crash). With
803/// the router listening it is marked Sleeping (the router or `clone --name` wakes it), else
804/// restarted as in v0.2; a non-durable one that did not shut down cleanly is suspect (fsync was
805/// off), and a suspect clone without a name is destroyed instead, as is an expired one or one
806/// whose parent exited.
807async fn recover_dead(
808    ctx: &Arc<Ctx>,
809    row: &CloneRow,
810    now: i64,
811    report: &mut GcReport,
812) -> Result<()> {
813    let parent_gone = row
814        .parent_pid
815        .is_some_and(|p| !parent_alive(p, row.parent_start_time));
816    if parent_gone || row.expires_at <= now {
817        return drop_clone(ctx, row, report).await;
818    }
819    let Ok(mark) = mark(ctx, &row.id, InFlightOp::Create) else {
820        return Ok(());
821    };
822    let mut row = row.clone();
823    if unclean(ctx, &row).await? {
824        if row.name.is_none() {
825            drop(mark);
826            log(
827                ctx,
828                &format!(
829                    "gc: clone {} did not shut down cleanly with fsync off; destroyed",
830                    row.id
831                ),
832            );
833            return drop_clone(ctx, &row, report).await;
834        }
835        mark_suspect(ctx, &mut row)?;
836    }
837    if crate::router::listening(ctx) {
838        // Not started now: the first router connection (or `clone --name`) wakes it.
839        row.status = CloneStatus::Sleeping;
840        (row.pg_pid, row.pg_start_time) = (None, None);
841        ctx.store().update_clone(&row)?;
842        events::emit_clone(
843            ctx,
844            events::RECOVERY,
845            &row,
846            serde_json::json!({ "restarted": false, "sleeping": true, "suspect": row.suspect }),
847        );
848        report.adopted.push(row.id.clone());
849        if let Some(w) = row.suspect.then(|| suspect_warning(&row)) {
850            log(ctx, &format!("gc: clone {}: {w}", row.id));
851            report.warnings.push(w);
852        }
853        if let Some(pid) = row.parent_pid {
854            watch_parent(ctx.clone(), row.id.clone(), pid);
855        }
856        return Ok(());
857    }
858    let old_port = row.port;
859    match restart_or_undo(ctx, &mut row).await {
860        Ok(warnings) => {
861            events::emit_clone(
862                ctx,
863                events::RECOVERY,
864                &row,
865                serde_json::json!({
866                    "restarted": true,
867                    "suspect": row.suspect,
868                    "port_moved": row.port != old_port,
869                }),
870            );
871            report.adopted.push(row.id.clone());
872            let suspect = row.suspect.then(|| suspect_warning(&row));
873            for w in warnings.into_iter().chain(suspect) {
874                log(ctx, &format!("gc: clone {}: {w}", row.id));
875                report.warnings.push(w);
876            }
877            if let Some(pid) = row.parent_pid {
878                watch_parent(ctx.clone(), row.id.clone(), pid);
879            }
880            Ok(())
881        }
882        // Kept, not destroyed: it may hold data someone wants. `restart_or_undo` put the old
883        // identity back, so the next daemon start or gc tries again.
884        Err(e) => Err(e),
885    }
886}
887
888fn entries(dir: &Path) -> Vec<(String, PathBuf)> {
889    let Ok(rd) = std::fs::read_dir(dir) else {
890        return Vec::new();
891    };
892    rd.flatten()
893        .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
894        .collect()
895}
896
897fn remove_path(path: &Path) -> Result<()> {
898    let res = if path.is_dir() {
899        std::fs::remove_dir_all(path)
900    } else {
901        std::fs::remove_file(path)
902    };
903    match res {
904        Err(e) if e.kind() != ErrorKind::NotFound => {
905            Err(internal(format!("remove {}: {e}", path.display())))
906        }
907        _ => Ok(()),
908    }
909}
910
911/// Stops the Postgres serving `dir`, if any, then deletes the folder.
912async fn remove_data_dir(ctx: &Ctx, dir: PathBuf) -> Result<()> {
913    let install = install_for(ctx, &dir);
914    let timeout = ctx.config().stop_timeout;
915    let backend = ctx.backend.clone();
916    tokio::task::spawn_blocking(move || {
917        // No row, so the server (if any) is the process whose cwd is this folder.
918        if running_server(&dir, None)?.is_some() {
919            let install = install.ok_or_else(|| {
920                internal(format!(
921                    "no Postgres install found to stop the server in {}",
922                    dir.display()
923                ))
924            })?;
925            stop(&install, &dir, None, timeout)?;
926        }
927        // A stray file under a clone or generation name is not the back end's.
928        if dir.is_dir() {
929            backend.remove_dir(&dir)
930        } else {
931            remove_path(&dir)
932        }
933    })
934    .await
935    .map_err(|e| internal(format!("background task failed: {e}")))?
936}
937
938/// One `template.generation_kept` event per folder the back end left in place because clones of it
939/// still exist (ZFS snapshots with clones); gc tries each again on its next run.
940pub(crate) fn emit_kept(ctx: &Ctx) {
941    for dir in ctx.backend.take_kept() {
942        log(
943            ctx,
944            &format!("gc: kept {}: clones still depend on it", dir.display()),
945        );
946        events::emit(
947            ctx,
948            events::GENERATION_KEPT,
949            None,
950            None,
951            serde_json::json!({ "reason": "clones_depend", "dir": dir.display().to_string() }),
952        );
953    }
954}
955
956/// Leaves an entry gc does not recognise alone, with one log line.
957fn foreign(ctx: &Ctx, path: &Path) {
958    log(
959        ctx,
960        &format!(
961            "gc: left {} alone: Mitosis did not create it",
962            path.display()
963        ),
964    );
965}
966
967/// Step 6: folders in `clones/`, `sockets/` and `templates/` that nothing owns any more.
968/// Only names Mitosis itself creates are touched; anything else is logged and left alone.
969async fn remove_orphans(ctx: &Ctx, report: &mut GcReport) {
970    let done = |report: &mut GcReport, path: PathBuf, res: Result<()>| match res {
971        Ok(()) => report.orphan_folders_removed.push(path),
972        Err(e) => {
973            log(ctx, &format!("gc: orphan {}: {e}", path.display()));
974            failed(report, path.display().to_string(), &e);
975        }
976    };
977
978    for (id, path) in entries(&ctx.home.clones_dir()) {
979        if !is_clone_id(&id) {
980            foreign(ctx, &path);
981            continue;
982        }
983        // Store and in-flight set are read under the store lock, as `clones::reserve` writes them.
984        let owned = {
985            let store = ctx.store();
986            store
987                .get_clone(&id)
988                .map(|r| r.is_some() || in_flight(ctx, &id))
989        };
990        match owned {
991            Ok(false) => {}
992            Ok(true) => continue,
993            Err(e) => {
994                log(
995                    ctx,
996                    &format!("gc: orphan check for {}: {e}", path.display()),
997                );
998                continue;
999            }
1000        }
1001        let res = remove_data_dir(ctx, path.clone()).await;
1002        done(report, path, res);
1003    }
1004
1005    for (name, path) in entries(&ctx.home.sockets_dir()) {
1006        let res = if diff_tag(&name, "d", "").is_some() {
1007            continue; // a diff's, maybe running now; daemon start removes leftovers
1008        } else if is_build_socket_name(&name) {
1009            // Holding the build set keeps a new build from taking its guard and reusing this folder.
1010            let building = ctx.building.lock().unwrap_or_else(|e| e.into_inner());
1011            if !building.is_empty() {
1012                continue;
1013            }
1014            remove_path(&path)
1015        } else if is_clone_id(&name) {
1016            // Under the store lock so a concurrent create cannot claim this id meanwhile.
1017            let store = ctx.store();
1018            match store.get_clone(&name) {
1019                Ok(Some(_)) => continue,
1020                Ok(None) if in_flight(ctx, &name) => continue,
1021                Ok(None) => remove_path(&path),
1022                Err(e) => Err(e),
1023            }
1024        } else {
1025            foreign(ctx, &path);
1026            continue;
1027        };
1028        done(report, path, res);
1029    }
1030
1031    // `logs/upgrade-*` work folders: an upgrade runs as a build, so with no build running they
1032    // are leftovers of a crash. The build set is held across the removal, as for build sockets.
1033    for (name, path) in entries(&ctx.home.logs_dir()) {
1034        if !name.starts_with("upgrade-") || !path.is_dir() {
1035            continue;
1036        }
1037        let building = ctx.building.lock().unwrap_or_else(|e| e.into_inner());
1038        if !building.is_empty() {
1039            break;
1040        }
1041        let res = remove_path(&path);
1042        drop(building);
1043        done(report, path, res);
1044    }
1045
1046    let templates_dir = ctx.home.templates_dir();
1047    // Read guard: a commit renames a folder to `<name>@<gen>` and inserts its row under the write
1048    // guard, so a generation folder is never seen here without its row.
1049    let swap = ctx.template_swap.read().await;
1050    let gen_dirs = match generation_folders(ctx) {
1051        Ok(d) => d,
1052        Err(e) => {
1053            // Without the rows nothing in templates/ can be judged; leave it all for the next gc.
1054            log(ctx, &format!("gc: list generations: {e}"));
1055            failed(report, templates_dir.display().to_string(), &e);
1056            return;
1057        }
1058    };
1059    let mut leftovers = Vec::new();
1060    let mut rowless = Vec::new();
1061    for (file, path) in entries(&templates_dir) {
1062        if gen_dirs.contains(&file) {
1063            continue;
1064        }
1065        if diff_tag(&file, ".diff-", "").is_some() {
1066            continue; // a diff's, maybe running now; daemon start removes leftovers
1067        } else if diff_tag(&file, GC_TRASH, "").is_some() {
1068            // Moved aside by a gc that stopped before removing it.
1069            let res = remove_data_dir(ctx, path.clone()).await;
1070            done(report, path, res);
1071        } else if let Some(l) = parse_leftover(&file) {
1072            leftovers.push((l, path));
1073        } else if parse_gen_dir(&file).is_some() || migrated_rowless(ctx, &file, &path) {
1074            // A generation folder with no row: a commit that crashed, or a prune that did not
1075            // finish (also the plain-named folder of a v0.1 store's migrated generation).
1076            rowless.push(path);
1077        } else {
1078            foreign(ctx, &path);
1079        }
1080    }
1081    restore_old_templates(ctx, &templates_dir, &gen_dirs, &mut leftovers, report);
1082    drop(swap);
1083    for path in rowless {
1084        // A clone that read the row before a prune dropped it may still be copying the folder
1085        // under the read guard: the write guard waits for it, and the check is repeated there.
1086        // Only the rename is guarded; stopping a stray server can take a while and would stall
1087        // every commit and create.
1088        let moved = {
1089            let _swap = ctx.template_swap.write().await;
1090            let file = path.file_name().map(|f| f.to_string_lossy().into_owned());
1091            match generation_folders(ctx) {
1092                Ok(dirs) if file.as_ref().is_some_and(|f| dirs.contains(f)) => continue,
1093                Ok(_) => {
1094                    let trash = templates_dir.join(format!("{GC_TRASH}{}", random_hex(4)));
1095                    ctx.backend
1096                        .rename_dir(&path, &trash)
1097                        .map(|()| trash)
1098                        .map_err(|e| {
1099                            internal(format!("move {} aside: {}", path.display(), e.message))
1100                        })
1101                }
1102                Err(e) => Err(e),
1103            }
1104        };
1105        let res = match moved {
1106            Ok(trash) => remove_data_dir(ctx, trash).await,
1107            Err(e) => Err(e),
1108        };
1109        done(report, path, res);
1110    }
1111    for (l, path) in leftovers {
1112        // A running build owns its own leftovers (its side folder, the folders its prune removes).
1113        let tpl = match &l {
1114            Leftover::Building(tpl) | Leftover::Old(tpl, _) => tpl,
1115        };
1116        // `tpl` is a folder stem: a branch template's build is known by its full name.
1117        let remove = !ctx
1118            .building
1119            .lock()
1120            .unwrap_or_else(|e| e.into_inner())
1121            .iter()
1122            .any(|b| folder_stem(b) == *tpl);
1123        if remove {
1124            let res = remove_data_dir(ctx, path.clone()).await;
1125            done(report, path, res);
1126        }
1127    }
1128}
1129
1130/// Prefix of a row-less generation folder gc moved aside to remove (`.gc-<8 hex>`).
1131const GC_TRASH: &str = ".gc-";
1132
1133/// The file names of every generation folder the store knows (they all sit in `templates/`).
1134fn generation_folders(ctx: &Ctx) -> Result<HashSet<String>> {
1135    let store = ctx.store();
1136    let mut dirs = HashSet::new();
1137    for t in store.list_templates()? {
1138        dirs.extend(
1139            store
1140                .list_generations(&t.name)?
1141                .into_iter()
1142                .filter_map(|g| g.dir.file_name().map(|f| f.to_string_lossy().into_owned())),
1143        );
1144    }
1145    Ok(dirs)
1146}
1147
1148/// True for `templates/<name>` named like a template that exists, holding a cluster, that no
1149/// generation row names (the caller checked): a v0.1 store's migrated generation 1 whose prune
1150/// crashed between dropping the row and moving the folder.
1151fn migrated_rowless(ctx: &Ctx, file: &str, path: &Path) -> bool {
1152    crate::names::validate_name("template", file).is_ok()
1153        && path.join("PG_VERSION").is_file()
1154        && ctx.store().get_template(file).ok().flatten().is_some()
1155}
1156
1157/// The 8-hex tag of a name `diff` makes (`<prefix><tag><suffix>`), else `None`.
1158fn diff_tag<'a>(name: &'a str, prefix: &str, suffix: &str) -> Option<&'a str> {
1159    name.strip_prefix(prefix)
1160        .and_then(|r| r.strip_suffix(suffix))
1161        .filter(|t| t.len() == 8 && t.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')))
1162}
1163
1164/// Removes what a diff left behind when the daemon crashed or stopped under it: its base folder
1165/// `templates/.diff-<tag>` (stopping the server in it first), `sockets/d<tag>` and
1166/// `logs/diff-<tag>.log`. Only at daemon start, before the socket binds: no diff runs then, and
1167/// nothing records the running ones. Logs, never fails; returns what it removed.
1168pub async fn remove_diff_leftovers(ctx: &Ctx) -> Vec<PathBuf> {
1169    let mut removed = Vec::new();
1170    let found = [
1171        (ctx.home.templates_dir(), ".diff-", ""),
1172        (ctx.home.sockets_dir(), "d", ""),
1173        (ctx.home.logs_dir(), "diff-", ".log"),
1174    ]
1175    .into_iter()
1176    .flat_map(|(dir, prefix, suffix)| {
1177        entries(&dir)
1178            .into_iter()
1179            .filter(move |(name, _)| diff_tag(name, prefix, suffix).is_some())
1180            .map(move |(_, path)| (path, prefix == ".diff-"))
1181    })
1182    .collect::<Vec<_>>();
1183    for (path, base) in found {
1184        let res = if base {
1185            remove_data_dir(ctx, path.clone()).await
1186        } else {
1187            remove_path(&path)
1188        };
1189        match res {
1190            Ok(()) => removed.push(path),
1191            Err(e) => log(ctx, &format!("diff leftover {}: {e}", path.display())),
1192        }
1193    }
1194    if !removed.is_empty() {
1195        log(ctx, &format!("removed {} diff leftover(s)", removed.len()));
1196    }
1197    removed
1198}
1199
1200/// A v0.1 swap that crashed between its two renames leaves `.<name>.old-*` but no `templates/<name>`.
1201/// When an upgraded store still has a generation in `templates/<name>`, the newest such folder is
1202/// moved back (and dropped from `leftovers`) before the rest are removed. v0.2 commits never move
1203/// a live folder, so its `.old-` folders are only pruned generations, whose rows are gone.
1204fn restore_old_templates(
1205    ctx: &Ctx,
1206    templates_dir: &Path,
1207    gen_dirs: &HashSet<String>,
1208    leftovers: &mut Vec<(Leftover, PathBuf)>,
1209    report: &mut GcReport,
1210) {
1211    let mut newest: std::collections::HashMap<String, (i64, usize)> = Default::default();
1212    for (i, (l, _)) in leftovers.iter().enumerate() {
1213        if let Leftover::Old(name, ts) = l {
1214            let e = newest.entry(name.clone()).or_insert((*ts, i));
1215            if *ts > e.0 {
1216                *e = (*ts, i);
1217            }
1218        }
1219    }
1220    let mut handled = Vec::new();
1221    for (name, (_, i)) in newest {
1222        let final_dir = templates_dir.join(&name);
1223        // The build set is held across the rename, so no build can swap this template meanwhile.
1224        let building = ctx.building.lock().unwrap_or_else(|e| e.into_inner());
1225        if building.contains(&name) || final_dir.exists() || !gen_dirs.contains(&name) {
1226            continue;
1227        }
1228        let old = &leftovers[i].1;
1229        match ctx.backend.rename_dir(old, &final_dir) {
1230            Ok(()) => {
1231                drop(building);
1232                log(
1233                    ctx,
1234                    &format!("gc: restored template {name} from {}", old.display()),
1235                );
1236                handled.push(i);
1237            }
1238            Err(e) => {
1239                drop(building);
1240                let e = internal(format!(
1241                    "restore {} to {}: {e}",
1242                    old.display(),
1243                    final_dir.display()
1244                ));
1245                log(ctx, &format!("gc: {e}"));
1246                failed(report, old.display().to_string(), &e);
1247                handled.push(i); // keep it on disk: it may be the only copy
1248            }
1249        }
1250    }
1251    handled.sort_unstable_by(|a, b| b.cmp(a));
1252    for i in handled {
1253        leftovers.remove(i);
1254    }
1255}
1256
1257#[cfg(test)]
1258mod tests {
1259    use super::*;
1260    use crate::clones::destroy;
1261    use crate::paths::Home;
1262
1263    fn tmp_ctx() -> (tempfile::TempDir, Arc<Ctx>) {
1264        // Under /tmp: short paths, on the default back end.
1265        let tmp = tempfile::Builder::new()
1266            .prefix("mt")
1267            .tempdir_in(crate::paths::test_tmp())
1268            .unwrap();
1269        let ctx = Ctx::open(
1270            Home::at(tmp.path().to_path_buf()),
1271            crate::ctx::Config::default(),
1272        )
1273        .unwrap();
1274        (tmp, Arc::new(ctx))
1275    }
1276
1277    /// Inserts a row and creates its (empty) clone folder; no Postgres involved.
1278    fn fake_clone(ctx: &Ctx, id: &str, status: CloneStatus) -> CloneRow {
1279        let t = now();
1280        let row = CloneRow {
1281            id: id.into(),
1282            name: None,
1283            template: "main".into(),
1284            dir: ctx.home.clones_dir().join(id),
1285            port: 54320,
1286            socket_dir: ctx.home.sockets_dir().join(id),
1287            password: "x".repeat(32),
1288            database: "app".into(),
1289            pg_pid: None,
1290            pg_start_time: None,
1291            parent_pid: None,
1292            parent_start_time: None,
1293            created_at: t,
1294            expires_at: t + 7200,
1295            durable: false,
1296            status,
1297            generation: 1,
1298            protected: false,
1299            suspect: false,
1300        };
1301        ctx.store().insert_clone(&row).unwrap();
1302        std::fs::create_dir(&row.dir).unwrap();
1303        row
1304    }
1305
1306    #[tokio::test]
1307    async fn reconcile_skips_in_flight_clones() {
1308        let (_tmp, ctx) = tmp_ctx();
1309        let row = fake_clone(&ctx, "cb0550a", CloneStatus::Creating);
1310        let m = mark(&ctx, &row.id, InFlightOp::Create).unwrap();
1311        let report = reconcile(ctx.clone(), now()).await.unwrap();
1312        assert!(
1313            report.destroyed.is_empty() && report.orphan_folders_removed.is_empty(),
1314            "{report:?}"
1315        );
1316        assert!(row.dir.is_dir() && ctx.store().get_clone(&row.id).unwrap().is_some());
1317
1318        drop(m); // the create "crashed": now it is debris
1319        let report = reconcile(ctx.clone(), now()).await.unwrap();
1320        assert_eq!(report.destroyed, vec![row.id.clone()]);
1321        assert!(!row.dir.exists() && ctx.store().get_clone(&row.id).unwrap().is_none());
1322    }
1323
1324    #[tokio::test]
1325    async fn reap_and_destroy_respect_in_flight_ops() {
1326        let (_tmp, ctx) = tmp_ctx();
1327        let row = fake_clone(&ctx, "cb0550b", CloneStatus::Ready);
1328        {
1329            let _m = mark(&ctx, &row.id, InFlightOp::Destroy).unwrap();
1330            assert!(reap_expired(&ctx, now() + 10_000).await.unwrap().is_empty());
1331            assert_eq!(
1332                destroy(&ctx, &row.id).await.unwrap(),
1333                row.id,
1334                "a second destroy is a quiet success"
1335            );
1336            assert!(row.dir.is_dir(), "the running destroy owns the folder");
1337        }
1338        {
1339            let _m = mark(&ctx, &row.id, InFlightOp::Create).unwrap();
1340            assert_eq!(
1341                destroy(&ctx, &row.id).await.unwrap_err().code,
1342                ErrorCode::InvalidRequest
1343            );
1344        }
1345        assert!(
1346            mark(&ctx, &row.id, InFlightOp::Destroy).is_ok(),
1347            "marks are released on drop"
1348        );
1349        assert_eq!(
1350            reap_expired(&ctx, now() + 10_000).await.unwrap(),
1351            vec![row.id.clone()]
1352        );
1353        assert!(!row.dir.exists());
1354    }
1355
1356    #[tokio::test]
1357    async fn destroy_during_wake_or_sleep_says_busy() {
1358        let (_tmp, ctx) = tmp_ctx();
1359        let row = fake_clone(&ctx, "cb0550c", CloneStatus::Sleeping);
1360        let _m = mark(&ctx, &row.id, InFlightOp::Create).unwrap(); // what a wake or sleep holds
1361        let e = destroy(&ctx, &row.id).await.unwrap_err();
1362        assert_eq!(e.code, ErrorCode::InvalidRequest);
1363        assert!(
1364            e.message.contains("waking or going to sleep"),
1365            "{}",
1366            e.message
1367        );
1368        assert!(!e.message.contains("still starting"), "{}", e.message);
1369    }
1370
1371    #[tokio::test]
1372    async fn destroy_forgets_router_activity() {
1373        let (_tmp, ctx) = tmp_ctx();
1374        let row = fake_clone(&ctx, "cb0550d", CloneStatus::Ready);
1375        ctx.router().last_active.insert(row.id.clone(), now());
1376        destroy(&ctx, &row.id).await.unwrap();
1377        assert_eq!(ctx.router().last_active(&row.id), None);
1378    }
1379
1380    #[tokio::test]
1381    async fn idle_check_of_a_dead_server_is_not_logged() {
1382        let (_tmp, ctx) = tmp_ctx();
1383        let row = fake_clone(&ctx, "cb0550e", CloneStatus::Ready); // no server runs
1384        let idle = idle_clones(&ctx, now()).await.unwrap();
1385        assert!(
1386            idle.iter().all(|(r, _)| r.id != row.id),
1387            "a failed check is not idle"
1388        );
1389        let log =
1390            std::fs::read_to_string(ctx.home.logs_dir().join("daemon.log")).unwrap_or_default();
1391        assert!(!log.contains("idle check"), "{log}");
1392    }
1393
1394    #[test]
1395    fn unused_reservations_are_dropped() {
1396        let (_tmp, ctx) = tmp_ctx();
1397        let t = now();
1398        let res = |name: &str, last_used: i64| crate::state::ReservationRow {
1399            name: name.into(),
1400            port: 54321,
1401            password: "x".repeat(32),
1402            template: "main".into(),
1403            last_used,
1404        };
1405        ctx.store()
1406            .upsert_reservation(&res("old", t - RESERVATION_KEEP_SECS - 10))
1407            .unwrap();
1408        ctx.store()
1409            .upsert_reservation(&res("recent", t - 10))
1410            .unwrap();
1411        drop_unused_reservations(&ctx, t);
1412        assert!(ctx.store().get_reservation("old").unwrap().is_none());
1413        assert!(ctx.store().get_reservation("recent").unwrap().is_some());
1414    }
1415
1416    #[tokio::test]
1417    async fn diff_leftovers_go_at_daemon_start_only() {
1418        let (_tmp, ctx) = tmp_ctx();
1419        let home = &ctx.home;
1420        let base = home.templates_dir().join(".diff-0123abcd");
1421        let sock = home.sockets_dir().join("d0123abcd");
1422        let log = home.logs_dir().join("diff-0123abcd.log");
1423        let not_diff = home.templates_dir().join(".diff-notours");
1424        std::fs::create_dir(&base).unwrap();
1425        std::fs::create_dir(&sock).unwrap();
1426        std::fs::write(&log, "x").unwrap();
1427        std::fs::create_dir(&not_diff).unwrap();
1428        // gc may run beside a diff: it leaves them be, and does not call them foreign.
1429        reconcile(ctx.clone(), now()).await.unwrap();
1430        assert!(base.is_dir() && sock.is_dir() && log.is_file());
1431        let daemon_log = std::fs::read_to_string(home.logs_dir().join("daemon.log")).unwrap();
1432        assert!(!daemon_log.contains("0123abcd"), "{daemon_log}");
1433
1434        let removed = remove_diff_leftovers(&ctx).await;
1435        assert_eq!(removed.len(), 3, "{removed:?}");
1436        assert!(!base.exists() && !sock.exists() && !log.exists());
1437        assert!(not_diff.is_dir(), "not a name diff makes");
1438    }
1439
1440    #[tokio::test]
1441    async fn gc_removes_rowless_and_crash_leftover_folders() {
1442        let (_tmp, ctx) = tmp_ctx();
1443        let home = &ctx.home;
1444        let src = crate::protocol::TemplateSource::Seed {
1445            command: "true".into(),
1446            cwd: None,
1447            path: None,
1448        };
1449        ctx.store()
1450            .insert_template(&crate::state::TemplateRow::new("old", "app", src, None, 0))
1451            .unwrap();
1452        // A v0.1 store's migrated generation whose prune crashed after its row went.
1453        let migrated = home.templates_dir().join("old");
1454        std::fs::create_dir(&migrated).unwrap();
1455        std::fs::write(migrated.join("PG_VERSION"), "17\n").unwrap();
1456        let rowless = home.templates_dir().join("main@3");
1457        std::fs::create_dir(&rowless).unwrap();
1458        let trash = home.templates_dir().join(".gc-0123abcd");
1459        std::fs::create_dir(&trash).unwrap();
1460        let upgrade = home.logs_dir().join("upgrade-main-1700000000");
1461        std::fs::create_dir(&upgrade).unwrap();
1462        // A plain folder of no template, or with no cluster in it, stays.
1463        let foreign = home.templates_dir().join("notes");
1464        std::fs::create_dir(&foreign).unwrap();
1465
1466        let report = reconcile(ctx.clone(), now()).await.unwrap();
1467        for p in [&migrated, &rowless, &trash, &upgrade] {
1468            assert!(!p.exists(), "{} left: {report:?}", p.display());
1469            assert!(report.orphan_folders_removed.contains(p), "{report:?}");
1470        }
1471        assert!(foreign.is_dir());
1472        let left: Vec<_> = entries(&home.templates_dir())
1473            .into_iter()
1474            .map(|(n, _)| n)
1475            .filter(|n| n.starts_with(GC_TRASH))
1476            .collect();
1477        assert!(left.is_empty(), "{left:?}");
1478    }
1479
1480    // The daemon runs these on a multi-thread runtime, so they must be Send.
1481    #[allow(dead_code)]
1482    fn futures_are_send(ctx: &'static Ctx, arc: Arc<Ctx>) {
1483        fn send<T: Send>(_: T) {}
1484        send(reap_expired(ctx, 0));
1485        send(reconcile(arc.clone(), 0));
1486        send(idle_pass(ctx, 0));
1487        send(remove_diff_leftovers(ctx));
1488        send(destroy_after_parent_exit(Watch {
1489            ctx: arc,
1490            id: String::new(),
1491        }));
1492    }
1493}