Skip to main content

onevcs/
pool.rs

1//! The pool: warm worktree slots an identity keeps between sessions.
2//!
3//! A slot is structurally a run root that survives `session close`. It lives under
4//! `<identity dir>/pool/<n>/` with its own `--shared --no-checkout` clone — never
5//! several slots on one clone, for the per-clone lock reason `workspace.rs`'s header
6//! gives — its worktree, and a `slot.json` naming the lender it borrows from. A session
7//! placed on one works in it exactly as it works in a run root: the occupancy lease,
8//! the process census, the dirty-tree preservation, the hand-back and the stray-work
9//! refusal all address the slot directory as the run root, and the record names it.
10//!
11//! What differs is the two ends. Taking a slot swaps the branch **in place** in the
12//! existing worktree, so an incremental build sees a branch diff rather than an empty
13//! tree; returning one detaches the worktree onto the base, resets it, and cleans the
14//! untracked files without touching the ignored ones — `.gitignore` being the
15//! repository's own declaration of what is build output — so the next session finds
16//! `target/`, `node_modules/` and `.venv/` still there.
17//!
18//! **A slot is idle iff no session record in state `open` names it and its maintenance
19//! claim is void** — null, or naming a process that is no longer running. That is the
20//! same record-based proof [`crate::workspace`]'s reclamation uses for run roots, and
21//! deliberately not a second liveness test: an open record whose owner exited holds its
22//! slot until `session close` or `onevcs sweep` forgets it, because a session opened
23//! from the command line has no owner process from the instant its token is printed.
24//!
25//! Sizing is the host's, per identity, in `$ONEVCS_HOME/workspaces.yml`
26//! ([`crate::workspaces`]): each host can afford a different amount of concurrent disk.
27//! The pool is lazy — a slot is cut when a session needs one and fewer than `pool`
28//! exist — so an identity that only ever has one concurrent task only ever has one
29//! slot. Past the pool, sessions overflow into disposable run roots under `runs/`
30//! exactly as before, up to `overflow`, and are then refused with
31//! [`Error::PoolExhausted`] rather than made to wait.
32
33use std::path::{Path, PathBuf};
34
35use serde::{Deserialize, Serialize};
36
37use crate::error::{self, Error, Result};
38use crate::session::{Lifecycle, SessionRequest, SessionToken};
39use crate::store::{self, Resolution};
40use crate::workspace::{self, ProcessStart, Record, Ref};
41use crate::workspaces::{self, Bound, Overrides, Resolved};
42use crate::{git, guidance, ids, lock, processes};
43
44mod maintain;
45
46pub use maintain::{
47    pool_maintain, IdentityMaintenance, IdentityOutcome, MaintainReport, SlotMaintenance,
48    SlotOutcome,
49};
50
51/// The version of the slot record this build writes and reads.
52pub const SLOT_VERSION: u32 = 1;
53
54/// A stored instant, checked where it is read.
55///
56/// The check is in the conversion, so a slot record carrying a `created`, a
57/// `last_maintained` or a claim's `since` that is not a timestamp this build can order
58/// by does not deserialize at all — it is a broken slot rather than a status line
59/// carrying whatever was in the file. The shape is the one [`ids::timestamp`] writes,
60/// which is what every other stored instant in this crate is held to.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(try_from = "String", into = "String")]
63pub(crate) struct Timestamp(String);
64
65impl TryFrom<String> for Timestamp {
66    type Error = String;
67
68    fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
69        match ids::is_timestamp(&value) {
70            true => Ok(Timestamp(value)),
71            false => Err(format!(
72                "{value:?} is not a timestamp this build can order by (RFC3339, millisecond, \
73                 UTC)"
74            )),
75        }
76    }
77}
78
79impl From<Timestamp> for String {
80    fn from(stamp: Timestamp) -> Self {
81        stamp.0
82    }
83}
84
85impl Timestamp {
86    /// Now, as [`ids::timestamp`] writes it.
87    fn now() -> Self {
88        Timestamp(ids::timestamp())
89    }
90}
91
92/// What one slot records about itself, in `slot.json`.
93///
94/// The fields a maintenance verb fills — `maintaining`, `last_maintained` and
95/// `last_outcome` — are declared and read here and first written by that verb, so a
96/// consumer reading `pool status` against them meets one shape from the day slots exist.
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub(crate) struct SlotRecord {
99    /// The schema version this record was written at.
100    pub version: u32,
101    /// The slot's number, which is its directory's name.
102    pub number: u32,
103    /// The identity key the slot belongs to.
104    pub identity: String,
105    /// The execution checkout the slot's clone borrows from — its lender. A slot is
106    /// bound to it: a session cloning from another checkout never takes this slot.
107    pub execution_checkout: PathBuf,
108    /// When the slot was cut.
109    pub created: Timestamp,
110    /// The claim the maintenance verb holds while it runs in the slot, or `null`.
111    pub maintaining: Option<Claim>,
112    /// When maintenance last finished, or `null` where none has run.
113    pub last_maintained: Option<Timestamp>,
114    /// How the last maintenance ended, or `null` where none has run.
115    pub last_outcome: Option<MaintenanceOutcome>,
116}
117
118/// The claim a maintenance run writes onto the slot it is working in.
119///
120/// Its owner *is* the process running the command, so a claim whose process is gone is
121/// void — cleared by the next reader rather than honoured — and the slot is idle again.
122#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
123pub(crate) struct Claim {
124    /// The process running the maintenance command.
125    pub pid: u32,
126    /// Its creation identity, so a later process wearing the same pid is not it.
127    pub started: ProcessStart,
128    /// When the claim was written.
129    pub since: Timestamp,
130}
131
132impl Claim {
133    /// Whether the process that wrote this claim is that same process, still running.
134    fn is_live(&self) -> bool {
135        workspace::process_started(self.pid) == Some(self.started)
136    }
137}
138
139/// How a maintenance run ended.
140#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
141#[serde(rename_all = "kebab-case")]
142pub enum MaintenanceOutcome {
143    /// The command exited zero inside its bound.
144    Succeeded,
145    /// The command exited non-zero, or was ended by a signal, inside its bound.
146    Failed {
147        /// The exit status, or `None` where a signal ended it.
148        exit: Option<i32>,
149    },
150    /// The bound elapsed first and the command was stopped.
151    TimedOut,
152}
153
154/// Where one slot is in its life, as `pool status` reports it.
155#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
156#[serde(tag = "state", rename_all = "kebab-case")]
157pub enum SlotState {
158    /// No open session names it and no live maintenance claims it: the next session
159    /// may take it.
160    Idle,
161    /// An open session record names it as its run root.
162    InUse {
163        /// The session working in it.
164        session: SessionToken,
165    },
166    /// A maintenance run whose process is still running claims it.
167    Maintaining {
168        /// The process running the maintenance command.
169        pid: u32,
170        /// When it claimed the slot, RFC3339 — checked where the record was read, and
171        /// spelled here as the amendment declares it.
172        // llmlint: ignore[invalid_states_unrepresentable] the approved amendment fixes
173        // this field as `String`, the way `Acknowledgement::recorded_at` is fixed; the
174        // value came through `Timestamp`'s conversion at the read boundary, so nothing
175        // that is not a timestamp reaches here.
176        since: String,
177    },
178    /// Its clone or worktree is missing or not a repository, or its record is
179    /// unreadable. The next session that would take it recreates it in place.
180    Broken {
181        /// What is wrong with it.
182        reason: String,
183    },
184}
185
186/// One slot, as `pool status` reports it.
187#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
188pub struct SlotStatus {
189    /// The slot's number.
190    pub number: u32,
191    /// The slot directory: the run root of every session placed on it.
192    pub path: PathBuf,
193    /// The execution checkout the slot is bound to. Empty where its record could not
194    /// be read.
195    pub execution_checkout: PathBuf,
196    /// Where it is in its life.
197    pub state: SlotState,
198    /// When maintenance last finished, RFC3339 — checked where the record was read,
199    /// and spelled here as the amendment declares it.
200    // llmlint: ignore[invalid_states_unrepresentable] the approved amendment fixes this
201    // field as `Option<String>`, the way `Acknowledgement::recorded_at` is fixed; the
202    // value came through `Timestamp`'s conversion at the read boundary.
203    pub last_maintained: Option<String>,
204    /// How the last maintenance ended.
205    pub last_outcome: Option<MaintenanceOutcome>,
206}
207
208/// How many sessions one identity can admit right now, and whether one request would
209/// be placed.
210///
211/// Advisory: it reads the same records `open` reads and applies the request's overrides
212/// over the same layering, and `open` is what decides. `admitted` is true for a request
213/// whose pinned branch an open session already holds, because that open resumes in
214/// place and places nothing.
215#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
216pub struct WorkspaceCapacity {
217    /// The identity key.
218    pub identity: String,
219    /// The pool size the request resolves to.
220    pub pool: u32,
221    /// How many slots exist, whatever their state.
222    pub slots: u32,
223    /// How many of them are idle. A broken slot is neither this nor in use — `slots`
224    /// exceeds the three counts by the broken ones — and a session still takes it,
225    /// recreating it in place, which `admits` and `admitted` count.
226    pub idle: u32,
227    /// How many an open session names.
228    pub in_use: u32,
229    /// How many a live maintenance run claims.
230    pub maintaining: u32,
231    /// The overflow bound the request resolves to.
232    pub overflow: Bound,
233    /// How many sessions of the identity are open under `runs/`, which is what the
234    /// overflow bound counts.
235    pub overflow_in_use: u32,
236    /// How many more opens the identity admits now, across its slots and its overflow.
237    pub admits: Bound,
238    /// Whether *this* request would be placed now.
239    pub admitted: bool,
240}
241
242/// What `pool status` reports: the capacity, and every slot.
243#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
244pub struct PoolStatus {
245    /// The identity's capacity, resolved with no per-open override.
246    pub capacity: WorkspaceCapacity,
247    /// Every slot, by number.
248    pub slots: Vec<SlotStatus>,
249}
250
251/// What `pool prune` did: which slots it removed, and which it kept and why.
252#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
253pub struct PruneReport {
254    /// The slots removed, by number.
255    pub removed: Vec<u32>,
256    /// The slots kept, each with the reason.
257    pub kept: Vec<(u32, String)>,
258}
259
260/// The pool directory of one identity.
261pub(crate) fn pool_dir(identity_root: &Path) -> PathBuf {
262    identity_root.join("pool")
263}
264
265/// The lock identity that serializes placement over one identity's pool.
266///
267/// Held from the survey to the record's save, so two opens cannot both take one idle
268/// slot or both count the same overflow headroom. Short by construction — a shared
269/// clone and a checkout — and never held across the fetch, which happens before it.
270pub(crate) fn placement_identity(pool: &Path) -> String {
271    format!("pool:{}", pool.display())
272}
273
274/// Where one slot's record is written.
275fn record_path(slot: &Path) -> PathBuf {
276    slot.join("slot.json")
277}
278
279/// Read one slot's record, or say why it cannot be acted on.
280///
281/// Serde proves the shape and nothing else, and every field here is handed to git or
282/// to the filesystem afterwards — so a record that disagrees with the directory it
283/// was read from is refused here, the way a session record naming a different token
284/// than its file is: one declaring another version, numbered for another slot,
285/// belonging to another identity, or lending from a path that is not absolute is a
286/// broken slot rather than one to place a session on.
287fn read_record(
288    slot: &Path,
289    number: u32,
290    identity: &str,
291) -> std::result::Result<SlotRecord, String> {
292    let path = record_path(slot);
293    let raw = std::fs::read_to_string(&path)
294        .map_err(|e| format!("its record at {} cannot be read: {e}", path.display()))?;
295    let record: SlotRecord = serde_json::from_str(&raw)
296        .map_err(|e| format!("its record at {} is malformed: {e}", path.display()))?;
297    if record.version != SLOT_VERSION {
298        return Err(format!(
299            "its record at {} declares version {}; this build reads version {SLOT_VERSION}",
300            path.display(),
301            record.version
302        ));
303    }
304    if record.number != number {
305        return Err(format!(
306            "its record at {} is for slot {}, not for slot {number}",
307            path.display(),
308            record.number
309        ));
310    }
311    if record.identity != identity {
312        return Err(format!(
313            "its record at {} belongs to {:?}, not to {identity:?}",
314            path.display(),
315            record.identity
316        ));
317    }
318    if !record.execution_checkout.is_absolute() {
319        return Err(format!(
320            "its record at {} names an execution checkout at {}, which is not an absolute path",
321            path.display(),
322            record.execution_checkout.display()
323        ));
324    }
325    Ok(record)
326}
327
328/// Write one slot's record whole.
329fn write_record(slot: &Path, record: &SlotRecord) -> Result<()> {
330    let path = record_path(slot);
331    let json = serde_json::to_string_pretty(record).map_err(error::at("serialize", &path))?;
332    crate::home::atomic_write(&path, &format!("{json}\n"))
333}
334
335/// One slot as the survey found it.
336#[derive(Debug, Clone)]
337struct Surveyed {
338    number: u32,
339    dir: PathBuf,
340    /// Its record, where it read.
341    record: Option<SlotRecord>,
342    state: SlotState,
343}
344
345impl Surveyed {
346    fn clone_dir(&self) -> PathBuf {
347        self.dir.join("clone")
348    }
349
350    fn worktree(&self) -> PathBuf {
351        self.dir.join("worktree")
352    }
353
354    fn lender(&self) -> Option<&Path> {
355        self.record
356            .as_ref()
357            .map(|record| record.execution_checkout.as_path())
358    }
359
360    fn takeable(&self) -> bool {
361        matches!(self.state, SlotState::Idle | SlotState::Broken { .. })
362    }
363
364    fn status(&self) -> SlotStatus {
365        SlotStatus {
366            number: self.number,
367            path: self.dir.clone(),
368            execution_checkout: self.lender().map(Path::to_path_buf).unwrap_or_default(),
369            state: self.state.clone(),
370            last_maintained: self
371                .record
372                .as_ref()
373                .and_then(|record| record.last_maintained.clone())
374                .map(String::from),
375            last_outcome: self.record.as_ref().and_then(|record| record.last_outcome),
376        }
377    }
378}
379
380/// Everything one placement or one report reads about an identity's pool, read once.
381struct Survey {
382    pool: PathBuf,
383    slots: Vec<Surveyed>,
384    /// Every open record of the identity, whichever kind of root it names.
385    open: Vec<Record>,
386}
387
388impl Survey {
389    /// The open sessions of the identity placed under `runs/`, which is what the
390    /// overflow bound counts. A stale open record counts until it is closed or
391    /// forgotten, and the refusal names it with its liveness so an operator can tell.
392    fn overflow(&self) -> Vec<&Record> {
393        self.open
394            .iter()
395            .filter(|record| record.slot.is_none())
396            .collect()
397    }
398
399    fn count(&self, wanted: impl Fn(&SlotState) -> bool) -> u32 {
400        u32::try_from(self.slots.iter().filter(|slot| wanted(&slot.state)).count())
401            .unwrap_or(u32::MAX)
402    }
403}
404
405/// Read every slot of an identity, and decide the state of each.
406///
407/// A claim whose process is gone is void and is cleared here, by the reader, so a
408/// maintenance run that died leaves nothing a later reader has to reason about.
409fn survey(identity: &str, pool: &Path) -> Result<Survey> {
410    let open: Vec<Record> = workspace::all()?
411        .into_iter()
412        .filter(|record| record.identity == identity && record.state == Lifecycle::Open)
413        .collect();
414    let mut slots = Vec::new();
415    // A pool nothing has cut a slot under yet is empty; one that is there and cannot
416    // be read is refused naming it, because every answer below — idle, in use, how
417    // many exist — would otherwise be answered from a listing nobody got.
418    let entries = match std::fs::read_dir(pool) {
419        Ok(entries) => entries,
420        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
421            return Ok(Survey {
422                pool: pool.to_path_buf(),
423                slots,
424                open,
425            })
426        }
427        Err(e) => return Err(error::at("read the pool at", pool)(e)),
428    };
429    for entry in entries {
430        // llmlint: ignore[changed_behavior_has_e2e] uncovered: a listing that yields
431        // an entry it cannot even name. No interface this crate exposes can produce
432        // one — the entries here are directories this crate created — so a journey for
433        // it would be a fixture standing in for the kernel's `readdir`; it is refused
434        // rather than skipped for the reason the directory itself is.
435        let entry = entry.map_err(error::at("read the pool at", pool))?;
436        let dir = entry.path();
437        // Anything that is not a numbered directory is not a slot: a stray file an
438        // operator dropped here is neither counted nor removed.
439        let Some(number) = entry
440            .file_name()
441            .to_str()
442            .and_then(|name| name.parse::<u32>().ok())
443            .filter(|number| *number > 0)
444        else {
445            continue;
446        };
447        if !dir.is_dir() {
448            continue;
449        }
450        let (record, state) = state_of(&dir, number, identity, &open);
451        slots.push(Surveyed {
452            number,
453            dir,
454            record,
455            state,
456        });
457    }
458    slots.sort_by_key(|slot| slot.number);
459    Ok(Survey {
460        pool: pool.to_path_buf(),
461        slots,
462        open,
463    })
464}
465
466/// The state of one slot: broken, in use, maintaining, or idle — in that order, because
467/// each answer is decided by something the next would have to trust.
468fn state_of(
469    dir: &Path,
470    number: u32,
471    identity: &str,
472    open: &[Record],
473) -> (Option<SlotRecord>, SlotState) {
474    let record = match read_record(dir, number, identity) {
475        Ok(record) => record,
476        Err(reason) => return (None, SlotState::Broken { reason }),
477    };
478    let clone = dir.join("clone");
479    let worktree = dir.join("worktree");
480    if !git::is_repo(&clone) {
481        return (
482            Some(record),
483            SlotState::Broken {
484                reason: format!(
485                    "its clone at {} is missing or not a repository",
486                    clone.display()
487                ),
488            },
489        );
490    }
491    if !git::is_repo(&worktree) {
492        return (
493            Some(record),
494            SlotState::Broken {
495                reason: format!(
496                    "its worktree at {} is missing or not a repository",
497                    worktree.display()
498                ),
499            },
500        );
501    }
502    if let Some(holder) = open.iter().find(|held| held.run_root == dir) {
503        return (
504            Some(record),
505            SlotState::InUse {
506                session: SessionToken(holder.token.to_string()),
507            },
508        );
509    }
510    if let Some(claim) = record.maintaining.clone() {
511        if claim.is_live() {
512            return (
513                Some(record),
514                SlotState::Maintaining {
515                    pid: claim.pid,
516                    since: claim.since.into(),
517                },
518            );
519        }
520        // Void: the process that wrote it is gone. Cleared best-effort, because the
521        // answer is the same whether or not the write lands.
522        let mut cleared = record.clone();
523        cleared.maintaining = None;
524        let _ = write_record(dir, &cleared);
525        return (Some(cleared), SlotState::Idle);
526    }
527    (Some(record), SlotState::Idle)
528}
529
530/// The local branches of a slot's clone that its lender does not reach: the ones a
531/// hand-back could not copy, which is the only way a branch outlives a return there.
532///
533/// A clone that is not there holds nothing, which is the one negative answer that is
534/// a fact rather than a failure. One that is there and cannot be listed is `Err`,
535/// carrying git's refusal: a count nobody got is not a count of none, and what this
536/// decides is whether a slot may be removed.
537fn retained_branches(
538    clone: &Path,
539    lender: Option<&Path>,
540) -> std::result::Result<Vec<String>, String> {
541    if !git::is_repo(clone) {
542        return Ok(Vec::new());
543    }
544    let branches = git::branches(clone).map_err(|e| e.to_string())?;
545    Ok(branches
546        .into_iter()
547        .filter(|branch| {
548            let reached = match (git::tip(clone, &format!("refs/heads/{branch}")), lender) {
549                (Some(tip), Some(lender)) => git::refs_reach(lender, &tip),
550                _ => false,
551            };
552            !reached
553        })
554        .collect())
555}
556
557/// Why an idle slot is kept rather than removed, or `None` where it may go.
558///
559/// `judge` is the checkout a slot with no readable lender is judged against — the
560/// requester's execution checkout at a placement, the publication checkout at a prune
561/// — so a slot whose record is gone can still be shown to retain nothing, rather than
562/// standing broken for ever because nothing could say.
563fn keeps(slot: &Surveyed, judge: &Path) -> Option<String> {
564    match &slot.state {
565        SlotState::InUse { session } => {
566            return Some(format!("session {} is working in it", session.0));
567        }
568        SlotState::Maintaining { pid, since } => {
569            return Some(format!(
570                "a maintenance run (pid {pid}) has claimed it since {since}"
571            ));
572        }
573        SlotState::Idle | SlotState::Broken { .. } => {}
574    }
575    // llmlint: ignore[changed_behavior_has_e2e] uncovered: git declining to list the
576    // branches of a repository this crate has just read as one. No interface this
577    // crate exposes produces that, and what it would prove is that the slot is
578    // *kept*, which is the answer every other unknown here resolves to.
579    let lender = slot.lender().unwrap_or(judge);
580    let retained = match retained_branches(&slot.clone_dir(), Some(lender)) {
581        Ok(retained) => retained,
582        Err(reason) => {
583            return Some(format!(
584                "its clone could not be asked what it retains, and a count nobody got is not \
585                 a count of none: {reason}"
586            ))
587        }
588    };
589    if retained.is_empty() {
590        return None;
591    }
592    let names: Vec<&str> = retained.iter().map(String::as_str).collect();
593    Some(format!(
594        "its clone retains {} — {} its lender {} would not take, and removing the slot is \
595         what would make {} unreachable",
596        guidance::listed(&names),
597        match retained.len() {
598            1 => "a branch",
599            _ => "branches",
600        },
601        lender.display(),
602        match retained.len() {
603            1 => "it",
604            _ => "them",
605        },
606    ))
607}
608
609/// Remove one slot that is proven idle, under the exclusive take that proves nobody is
610/// inside it right now.
611fn remove(slot: &Surveyed) -> std::result::Result<(), String> {
612    let Some(_exclusive) = lock::try_exclusive(&workspace::occupancy_identity(&slot.dir))
613        .map_err(|e| e.to_string())?
614    else {
615        return Err("a command is working in it right now".to_owned());
616    };
617    std::fs::remove_dir_all(&slot.dir).map_err(|e| format!("it could not be removed: {e}"))
618}
619
620/// Shed every idle slot above the file's pool size, highest number first, keeping any
621/// that retains a branch.
622///
623/// Against the **file's** pool and never a per-process override: `--pool` on one open
624/// and `ONEVCS_POOL` in a dispatch's environment place that one session, and a
625/// per-process value that removed slots would take the warm pool down every time a
626/// node was handed one.
627fn shed(survey: &mut Survey, file_pool: u32, judge: &Path) {
628    let mut removed = Vec::new();
629    for slot in sheddable(survey, file_pool, judge) {
630        if remove(slot).is_ok() {
631            removed.push(slot.number);
632        }
633    }
634    survey.slots.retain(|slot| !removed.contains(&slot.number));
635}
636
637/// The slots a shed against `file_pool` removes, in the order it removes them:
638/// takeable ones above the pool size, highest number first, passing over any that is
639/// kept for what it retains.
640///
641/// One answer for the shed itself and for the capacity that predicts it, so the
642/// advisory read cannot count a slot as gone that the shed keeps.
643fn sheddable<'a>(survey: &'a Survey, file_pool: u32, judge: &Path) -> Vec<&'a Surveyed> {
644    let mut surplus = u32::try_from(survey.slots.len())
645        .unwrap_or(u32::MAX)
646        .saturating_sub(file_pool);
647    let mut shed = Vec::new();
648    for slot in survey.slots.iter().rev() {
649        if surplus == 0 {
650            break;
651        }
652        if !slot.takeable() || keeps(slot, judge).is_some() {
653            continue;
654        }
655        shed.push(slot);
656        surplus -= 1;
657    }
658    shed
659}
660
661/// Where a session was placed.
662#[derive(Debug)]
663pub(crate) enum Placement {
664    /// On a slot, whose directory is the session's run root.
665    Slot {
666        /// Its number.
667        number: u32,
668        /// Its directory.
669        dir: PathBuf,
670        /// How much of it the session found already there.
671        cut: Cut,
672        /// The exclusive take that keeps every other placement, prune and shed off it
673        /// until the session's record names it. Dropped by the caller after the save.
674        _held: lock::Guard,
675        /// The serialization over the whole pool, dropped with the take above.
676        _serial: lock::Guard,
677    },
678    /// Under `runs/`, exactly as every session was placed before there were slots.
679    RunRoot {
680        /// The serialization over the whole pool, held through the record's save so
681        /// two overflow opens cannot both count the same headroom. `None` where the
682        /// identity is not pooled and nothing was counted.
683        _serial: Option<lock::Guard>,
684    },
685}
686
687/// Everything a placement decision needs to know about the request.
688pub(crate) struct Ask<'a> {
689    pub resolution: &'a Resolution,
690    pub execution: &'a Path,
691    pub resolved: &'a Resolved,
692    pub pinned: Option<&'a Ref>,
693    pub identity_root: &'a Path,
694}
695
696/// Decide where a session goes, in the manager's ruled order and never waiting.
697///
698/// Surplus idle slots are shed first; then an idle slot bound to this lender — the one
699/// the pinned branch's most recent closed session used, where there is one; then a new
700/// slot at the lowest free number while fewer than `pool` exist; then a run root under
701/// `runs/` while the identity's overflow admits one; then a refusal naming every holder.
702/// A `--pool 0` skips the slots and spends the overflow, because the cap is about disk
703/// and a session that opted out of reuse still spends it.
704pub(crate) fn place(ask: &Ask<'_>) -> Result<Placement> {
705    let pool = pool_dir(ask.identity_root);
706    crate::home::ensure_dir(&pool)?;
707    let serial = lock::exclusive(&placement_identity(&pool))?;
708    let mut survey = survey(&ask.resolution.key, &pool)?;
709    shed(&mut survey, ask.resolved.file_pool, ask.execution);
710    let wanted = ask.resolved.pool.value;
711    if wanted > 0 {
712        let taken = match take_idle(&survey, ask)? {
713            Some(taken) => Some(taken),
714            None => match recreate_broken(&survey, ask)? {
715                Some(recreated) => Some(recreated),
716                None if u32::try_from(survey.slots.len()).unwrap_or(u32::MAX) < wanted => {
717                    Some(cut_new(&survey, ask)?)
718                }
719                None => None,
720            },
721        };
722        if let Some(taken) = taken {
723            return Ok(Placement::Slot {
724                number: taken.number,
725                dir: taken.dir.clone(),
726                cut: taken.cut,
727                _held: taken.held,
728                _serial: serial,
729            });
730        }
731    }
732    let overflow_in_use = u32::try_from(survey.overflow().len()).unwrap_or(u32::MAX);
733    if ask.resolved.overflow.value.admits(overflow_in_use) {
734        return Ok(Placement::RunRoot {
735            _serial: Some(serial),
736        });
737    }
738    Err(exhausted(ask, &survey))
739}
740
741/// How much of a slot a placement found already there, which is how much `open` has
742/// still to build.
743#[derive(Debug, Clone, Copy, PartialEq, Eq)]
744pub(crate) enum Cut {
745    /// Nothing: the clone and the worktree are cut from the lender.
746    Fresh,
747    /// The clone, with every ref it retains; only the worktree is cut, from that
748    /// clone — a broken slot whose clone is still a repository.
749    KeptClone,
750    /// Both, standing as the last return left them: the branch is swapped in place.
751    Warm,
752}
753
754impl Cut {
755    /// Whether the session's opening event reports the slot as cut for it rather
756    /// than taken warm: a worktree built from nothing is, whatever the clone kept.
757    pub(crate) fn created(self) -> bool {
758        self != Cut::Warm
759    }
760}
761
762/// A slot taken for a placement: its number, its directory, how much of it was
763/// there, and the exclusive take.
764struct Taken {
765    number: u32,
766    dir: PathBuf,
767    cut: Cut,
768    held: lock::Guard,
769}
770
771/// The idle slot of this lender the session takes, preferring the pinned branch's own.
772fn take_idle(survey: &Survey, ask: &Ask<'_>) -> Result<Option<Taken>> {
773    let preferred = match ask.pinned {
774        Some(branch) => last_slot_of(&ask.resolution.key, branch)?,
775        None => None,
776    };
777    let mut candidates: Vec<&Surveyed> = survey
778        .slots
779        .iter()
780        .filter(|slot| slot.state == SlotState::Idle && slot.lender() == Some(ask.execution))
781        .collect();
782    candidates.sort_by_key(|slot| (Some(slot.number) != preferred, slot.number));
783    for slot in candidates {
784        // The exclusive take is what closes the window between the survey and the
785        // record: a prune or another placement that has it is inside this slot now.
786        let Some(held) = lock::try_exclusive(&workspace::occupancy_identity(&slot.dir))? else {
787            continue;
788        };
789        // A slot whose last session closed without returning it — a publication closes
790        // the record, and `session close` returns the tree afterwards — may still have
791        // that session's worker inside; the same census the close refuses on skips it.
792        if !returned(&slot.worktree()) && !processes::holding(&slot.dir).is_empty() {
793            continue;
794        }
795        return Ok(Some(Taken {
796            number: slot.number,
797            dir: slot.dir.clone(),
798            cut: Cut::Warm,
799            held,
800        }));
801    }
802    Ok(None)
803}
804
805/// A broken slot of this lender — or of no readable lender — recreated in place,
806/// keeping whatever of it is still usable.
807///
808/// Only what is broken is rebuilt, because a slot's clone is where a branch the
809/// hand-back could not copy lives. A clone that is still a repository is kept with
810/// every ref it retains: an unreadable record is rewritten beside it, and a worktree
811/// that is missing or not a repository is cut again from that clone. Only a clone that
812/// is itself unusable — missing, or not a repository git can read — is removed, and
813/// there is nothing it could be asked to retain.
814fn recreate_broken(survey: &Survey, ask: &Ask<'_>) -> Result<Option<Taken>> {
815    for slot in &survey.slots {
816        if !matches!(slot.state, SlotState::Broken { .. }) {
817            continue;
818        }
819        let clone = slot.clone_dir();
820        let worktree = slot.worktree();
821        // The lender the slot is bound to: its record's, or — where the record could
822        // not be read — the checkout its clone borrows objects from. A slot bound to
823        // another lender is not this request's to rebuild any more than to take.
824        let lender = slot
825            .lender()
826            .map(Path::to_path_buf)
827            .or_else(|| lender_of(&clone));
828        if lender.is_some_and(|lender| lender != ask.execution) {
829            continue;
830        }
831        let Some(held) = lock::try_exclusive(&workspace::occupancy_identity(&slot.dir))? else {
832            continue;
833        };
834        let cut = if git::is_repo(&clone) {
835            if git::is_repo(&worktree) {
836                Cut::Warm
837            } else {
838                if worktree.exists() {
839                    std::fs::remove_dir_all(&worktree)
840                        .map_err(error::at("remove the broken worktree at", &worktree))?;
841                }
842                // The clone still registers the worktree it lost; a fresh `worktree
843                // add` at the same path is refused until that registration is gone.
844                git::worktree_prune(&clone)?;
845                Cut::KeptClone
846            }
847        } else {
848            std::fs::remove_dir_all(&slot.dir)
849                .map_err(error::at("remove the broken slot at", &slot.dir))?;
850            std::fs::create_dir(&slot.dir).map_err(error::at("create", &slot.dir))?;
851            Cut::Fresh
852        };
853        write_record(&slot.dir, &fresh_record(slot.number, ask))?;
854        return Ok(Some(Taken {
855            number: slot.number,
856            dir: slot.dir.clone(),
857            cut,
858            held,
859        }));
860    }
861    Ok(None)
862}
863
864/// The checkout a slot's clone borrows its objects from, read off the clone itself
865/// for a slot whose record cannot say.
866///
867/// A shared clone records its lender's object store in `objects/info/alternates`,
868/// which for a checkout is `<checkout>/.git/objects`; anything else there is not a
869/// lender this crate can name, and answers `None`.
870fn lender_of(clone: &Path) -> Option<PathBuf> {
871    let alternates = git::objects_dir(clone).ok()?.join("info/alternates");
872    let recorded = std::fs::read_to_string(alternates).ok()?;
873    let objects = PathBuf::from(recorded.lines().next()?.trim());
874    let git_dir = objects
875        .file_name()
876        .is_some_and(|name| name == "objects")
877        .then(|| objects.parent())
878        .flatten()?;
879    git_dir
880        .file_name()
881        .is_some_and(|name| name == ".git")
882        .then(|| git_dir.parent().map(Path::to_path_buf))
883        .flatten()
884}
885
886/// A new slot at the lowest free number.
887///
888/// `create_dir` rather than `create_dir_all`, so a number two placements race for is
889/// taken by exactly one of them and the other moves on to the next.
890fn cut_new(survey: &Survey, ask: &Ask<'_>) -> Result<Taken> {
891    let mut number: u32 = 1;
892    loop {
893        if survey.slots.iter().any(|slot| slot.number == number) {
894            number += 1;
895            continue;
896        }
897        let dir = survey.pool.join(number.to_string());
898        match std::fs::create_dir(&dir) {
899            Ok(()) => {}
900            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
901                number += 1;
902                continue;
903            }
904            Err(e) => return Err(error::at("create", &dir)(e)),
905        }
906        let held = lock::try_exclusive(&workspace::occupancy_identity(&dir))?.ok_or_else(|| {
907            error::invalid(format!(
908                "the new slot {} is already occupied before it was cut",
909                dir.display()
910            ))
911        })?;
912        write_record(&dir, &fresh_record(number, ask))?;
913        return Ok(Taken {
914            number,
915            dir,
916            cut: Cut::Fresh,
917            held,
918        });
919    }
920}
921
922fn fresh_record(number: u32, ask: &Ask<'_>) -> SlotRecord {
923    SlotRecord {
924        version: SLOT_VERSION,
925        number,
926        identity: ask.resolution.key.clone(),
927        execution_checkout: ask.execution.to_path_buf(),
928        created: Timestamp::now(),
929        maintaining: None,
930        last_maintained: None,
931        last_outcome: None,
932    }
933}
934
935/// The slot the most recent closed session of this identity on `branch` used, when
936/// there is one: a retry pinned to a preserved branch goes back where its predecessor
937/// built.
938fn last_slot_of(identity: &str, branch: &Ref) -> Result<Option<u32>> {
939    let mut newest: Option<(std::time::SystemTime, u32)> = None;
940    for record in workspace::all()? {
941        let Some(number) = record.slot else {
942            continue;
943        };
944        if record.identity != identity
945            || record.branch != *branch
946            || record.state != Lifecycle::Closed
947        {
948            continue;
949        }
950        let written = workspace::record_path(&record.token)
951            .ok()
952            .and_then(|path| std::fs::metadata(path).ok())
953            .and_then(|meta| meta.modified().ok())
954            .unwrap_or(std::time::UNIX_EPOCH);
955        if newest.is_none_or(|(when, _)| written > when) {
956            newest = Some((written, number));
957        }
958    }
959    Ok(newest.map(|(_, number)| number))
960}
961
962/// Whether one slot is idle now: no open session record names it and no live
963/// maintenance claims it — the one proof a placement takes a slot on.
964pub(crate) fn is_idle(identity: &str, slot: &Path) -> Result<bool> {
965    let Some(pool) = slot.parent() else {
966        return Ok(false);
967    };
968    Ok(survey(identity, pool)?
969        .slots
970        .iter()
971        .any(|surveyed| surveyed.dir == slot && surveyed.state == SlotState::Idle))
972}
973
974/// Whether a slot's worktree stands as a return leaves it: detached, and clean.
975pub(crate) fn returned(worktree: &Path) -> bool {
976    git::current_branch(worktree).is_ok_and(|branch| branch == workspace::DETACHED_HEAD)
977        && git::is_dirty(worktree).is_ok_and(|dirty| !dirty)
978}
979
980/// Why no session can be placed now: the identity, the resolved limits with their
981/// sources, and every holder — the way the close's occupancy refusal names processes.
982fn exhausted(ask: &Ask<'_>, survey: &Survey) -> Error {
983    let resolved = ask.resolved;
984    let slots = survey.slots.len();
985    let idle = survey.count(|state| *state == SlotState::Idle);
986    let in_use = survey.count(|state| matches!(state, SlotState::InUse { .. }));
987    let maintaining = survey.count(|state| matches!(state, SlotState::Maintaining { .. }));
988    let overflow = survey.overflow();
989    let mut holders: Vec<String> = Vec::new();
990    for slot in &survey.slots {
991        match &slot.state {
992            SlotState::InUse { session } => {
993                let held = survey
994                    .open
995                    .iter()
996                    .find(|record| *record.token == *session.0);
997                holders.push(match held {
998                    Some(record) => format!("slot {}: {}", slot.number, describe(record)),
999                    None => format!("slot {}: session {}", slot.number, session.0),
1000                });
1001            }
1002            SlotState::Maintaining { pid, since } => holders.push(format!(
1003                "slot {}: a maintenance run, pid {pid}, since {since}",
1004                slot.number
1005            )),
1006            SlotState::Idle => holders.push(format!(
1007                "slot {}: idle, bound to {}",
1008                slot.number,
1009                slot.lender()
1010                    .map(|lender| lender.display().to_string())
1011                    .unwrap_or_else(|| "(unknown)".to_owned())
1012            )),
1013            SlotState::Broken { reason } => {
1014                holders.push(format!("slot {}: broken — {reason}", slot.number));
1015            }
1016        }
1017    }
1018    for record in &overflow {
1019        holders.push(format!("under runs/: {}", describe(record)));
1020    }
1021    let file = workspaces::default_path()
1022        .map(|path| path.display().to_string())
1023        .unwrap_or_else(|_| "$ONEVCS_HOME/workspaces.yml".to_owned());
1024    Error::PoolExhausted {
1025        reason: format!(
1026            "no session of {identity} can be placed now: its pool is {pool} (from {pool_from}) — \
1027             {slots} created, {idle} idle, {in_use} in use, {maintaining} maintaining — and its \
1028             overflow is {overflow} (from {overflow_from}) with {overflow_in_use} in use. \
1029             {holding}. Close a session, raise pool or overflow for this identity in {file}, \
1030             or open this one with `--overflow unlimited`",
1031            identity = ask.resolution.key,
1032            pool = resolved.pool.value,
1033            pool_from = resolved.pool.from,
1034            overflow = resolved.overflow.value,
1035            overflow_from = resolved.overflow.from,
1036            overflow_in_use = overflow.len(),
1037            holding = match holders.is_empty() {
1038                true => "Nothing holds it".to_owned(),
1039                false => format!("Held by: {}", holders.join("; ")),
1040            },
1041        ),
1042    }
1043}
1044
1045/// One holder, as the refusal names it: token, branch, worktree, owner pid, liveness.
1046fn describe(record: &Record) -> String {
1047    format!(
1048        "session {} on {:?} in {}, opened by pid {} ({})",
1049        record.token,
1050        record.branch,
1051        record.worktree.display(),
1052        record.owner_pid,
1053        record.liveness().as_str(),
1054    )
1055}
1056
1057/// Which slot, if any, `request` resolves to as its request-level view: the identity,
1058/// its execution checkout, and the resolved policy.
1059struct Asked {
1060    resolution: Resolution,
1061    execution: PathBuf,
1062    resolved: Resolved,
1063    identity_root: PathBuf,
1064}
1065
1066fn asked(request: &SessionRequest) -> Result<Asked> {
1067    let registry = store::load()?;
1068    let resolution = store::resolve(&registry, &request.repo)?;
1069    let execution = workspace::execution_checkout(
1070        &registry,
1071        &resolution,
1072        request.execution_checkout.as_deref(),
1073    )?;
1074    let resolved = workspaces::resolve_for(
1075        &resolution,
1076        Overrides {
1077            pool: request.pool,
1078            overflow: request.overflow,
1079        },
1080    )?;
1081    let identity_root = workspace::identity_dir(&resolution.key)?;
1082    Ok(Asked {
1083        resolution,
1084        execution,
1085        resolved,
1086        identity_root,
1087    })
1088}
1089
1090/// The capacity one request meets, read the way `open` reads it.
1091fn capacity_of(asked: &Asked, survey: &Survey, request: &SessionRequest) -> WorkspaceCapacity {
1092    let resolved = &asked.resolved;
1093    let slots = u32::try_from(survey.slots.len()).unwrap_or(u32::MAX);
1094    let idle = survey.count(|state| *state == SlotState::Idle);
1095    let takeable =
1096        survey.count(|state| matches!(state, SlotState::Idle | SlotState::Broken { .. }));
1097    let in_use = survey.count(|state| matches!(state, SlotState::InUse { .. }));
1098    let maintaining = survey.count(|state| matches!(state, SlotState::Maintaining { .. }));
1099    let overflow_in_use = u32::try_from(survey.overflow().len()).unwrap_or(u32::MAX);
1100    // What the next open finds after the shed it performs first: exactly the slots
1101    // the shed would remove are gone — never one it keeps for what it retains — and
1102    // the rest stay.
1103    let shed = u32::try_from(sheddable(survey, resolved.file_pool, &asked.execution).len())
1104        .unwrap_or(u32::MAX);
1105    let takeable_after = takeable.saturating_sub(shed);
1106    let slots_after = slots.saturating_sub(shed);
1107    let can_create = resolved.pool.value.saturating_sub(slots_after);
1108    let admits = match resolved.pool.value {
1109        0 => resolved.overflow.value.headroom(overflow_in_use),
1110        _ => match resolved.overflow.value.headroom(overflow_in_use) {
1111            Bound::Unlimited => Bound::Unlimited,
1112            Bound::Bounded(room) => Bound::Bounded(
1113                takeable_after
1114                    .saturating_add(can_create)
1115                    .saturating_add(room),
1116            ),
1117        },
1118    };
1119    let resumes = request.branch.as_ref().is_some_and(|pinned| {
1120        survey.open.iter().any(|record| {
1121            *record.branch == **pinned
1122                && record.execution_checkout == asked.execution
1123                && record.run_root.is_dir()
1124                && record.clone.is_dir()
1125        })
1126    });
1127    let mine = survey.slots.iter().any(|slot| {
1128        slot.takeable() && slot.lender().is_none_or(|lender| lender == asked.execution)
1129    });
1130    let admitted = resumes
1131        || (resolved.pool.value > 0 && (mine || can_create > 0))
1132        || resolved.overflow.value.admits(overflow_in_use);
1133    WorkspaceCapacity {
1134        identity: asked.resolution.key.clone(),
1135        pool: resolved.pool.value,
1136        slots,
1137        idle,
1138        in_use,
1139        maintaining,
1140        overflow: resolved.overflow.value,
1141        overflow_in_use,
1142        admits,
1143        admitted,
1144    }
1145}
1146
1147/// How many sessions the request's identity admits right now, and whether the request
1148/// itself would be placed.
1149pub fn workspace_capacity(request: &SessionRequest) -> Result<WorkspaceCapacity> {
1150    let asked = asked(request)?;
1151    let survey = survey(&asked.resolution.key, &pool_dir(&asked.identity_root))?;
1152    Ok(capacity_of(&asked, &survey, request))
1153}
1154
1155/// The pool of one repository: its capacity with no per-open override, and every slot.
1156pub fn pool_status(repo: &str) -> Result<PoolStatus> {
1157    let request = SessionRequest {
1158        repo: repo.to_owned(),
1159        branch: None,
1160        branch_name: None,
1161        branch_prefix: None,
1162        base: None,
1163        execution_checkout: None,
1164        pool: None,
1165        overflow: None,
1166        labels: Default::default(),
1167        refuse_conflicts: false,
1168    };
1169    let asked = asked(&request)?;
1170    let survey = survey(&asked.resolution.key, &pool_dir(&asked.identity_root))?;
1171    Ok(PoolStatus {
1172        capacity: capacity_of(&asked, &survey, &request),
1173        slots: survey.slots.iter().map(Surveyed::status).collect(),
1174    })
1175}
1176
1177/// Remove every slot of one repository nobody is in — idle, or broken — whose clone
1178/// retains no branch, and say why each other one was kept.
1179///
1180/// A broken slot goes the way an idle one does because it is takeable the way an idle
1181/// one is: the next open would recreate it in place, and what that recreation keeps is
1182/// exactly the retained branch that keeps a slot here.
1183pub fn pool_prune(repo: &str) -> Result<PruneReport> {
1184    let request = SessionRequest {
1185        repo: repo.to_owned(),
1186        branch: None,
1187        branch_name: None,
1188        branch_prefix: None,
1189        base: None,
1190        execution_checkout: None,
1191        pool: None,
1192        overflow: None,
1193        labels: Default::default(),
1194        refuse_conflicts: false,
1195    };
1196    let asked = asked(&request)?;
1197    let pool = pool_dir(&asked.identity_root);
1198    let mut report = PruneReport {
1199        removed: Vec::new(),
1200        kept: Vec::new(),
1201    };
1202    if !pool.is_dir() {
1203        return Ok(report);
1204    }
1205    // Serialized against placement, so a slot this is removing is not one an open is
1206    // in the middle of taking.
1207    let _serial = lock::exclusive(&placement_identity(&pool))?;
1208    let survey = survey(&asked.resolution.key, &pool)?;
1209    for slot in &survey.slots {
1210        if let Some(why) = keeps(slot, &asked.resolution.publication) {
1211            report.kept.push((slot.number, why));
1212            continue;
1213        }
1214        match remove(slot) {
1215            Ok(()) => report.removed.push(slot.number),
1216            Err(why) => report.kept.push((slot.number, why)),
1217        }
1218    }
1219    Ok(report)
1220}