use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::Context as _;
use serde::{Deserialize, Serialize};
use crate::config::ValidatedConfig;
use crate::datastore::Datastore;
use crate::datastore::keys::RepoKey;
use crate::out::out;
use crate::plan::repo_key;
use crate::store::RepoStore;
/// the extra minute covers the reload landing after the activation record.
pub(crate) const DRAIN_SECS: u64 = crate::nginx::WORKER_SHUTDOWN_SECS + 60;
const STORE_DIRS: [&str; 2] = ["releases", "cache/export"];
const PINS_DIR: &str = "pins";
const QUEUE_FILE: &str = "runtime/gc.queued";
/// what gc reads of each generation's manifest, kept by generation id since a generation never changes
const LIVE_CACHE: &str = "cache/gc-live";
/// a build holds it until its activation, so another activation's gc leaves the candidate alone
#[derive(Debug)]
pub struct Pin {
_file: std::fs::File,
}
/// gc only runs under the output lock, so take it under that lock too
pub(crate) fn pin(output_root: &Path, generation: &str) -> anyhow::Result<Pin> {
let dir = output_root.join(PINS_DIR);
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let path = dir.join(generation);
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.open(&path)
.with_context(|| format!("open pin {}", path.display()))?;
// shared, two builds can make the same generation
file.lock_shared()
.with_context(|| format!("pin {}", path.display()))?;
Ok(Pin { _file: file })
}
/// at most one gc waits for the output lock; a write that finds the slot taken leaves its gc to
/// that one, which collects against whatever is active once it has the lock
pub(crate) fn queue(output_root: &Path) -> anyhow::Result<Option<std::fs::File>> {
std::fs::create_dir_all(output_root.join("runtime")).context("create runtime dir")?;
let file = crate::output::lock_file(output_root, QUEUE_FILE)?;
Ok(file.try_lock().is_ok().then_some(file))
}
/// a pin nobody holds is left over from a build that ended, and goes
fn pinned(output_root: &Path, generation: &str) -> bool {
let path = output_root.join(PINS_DIR).join(generation);
let Ok(file) = std::fs::File::open(&path) else {
return false;
};
if file.try_lock().is_err() {
return true;
}
let _ = std::fs::remove_file(&path);
false
}
const WORKER: &[u8] = b"nginx: worker process";
const SHUTTING_DOWN: &[u8] = b"nginx: worker process is shutting down";
const RELOAD_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const RELOAD_POLL: std::time::Duration = std::time::Duration::from_millis(10);
/// workers of the nginx master in the pid file that still accept connections.
/// on SIGHUP the master starts new ones and sets the old ones to `is shutting down`.
pub(crate) fn active_workers(output_root: &Path) -> BTreeSet<u32> {
active_workers_in(Path::new("/proc"), output_root)
}
pub(crate) fn wait_for_reload(output_root: &Path, before: &BTreeSet<u32>) {
if !wait_for_reload_in(Path::new("/proc"), output_root, before, RELOAD_TIMEOUT) {
tracing::warn!("nginx reload did not swap workers within {RELOAD_TIMEOUT:?}");
}
}
pub(crate) fn wait_for_reload_in(
proc: &Path,
output_root: &Path,
before: &BTreeSet<u32>,
timeout: std::time::Duration,
) -> bool {
if before.is_empty() {
return true;
}
let deadline = std::time::Instant::now() + timeout;
loop {
let now = active_workers_in(proc, output_root);
if !now.is_empty() && now.is_disjoint(before) {
return true;
}
if std::time::Instant::now() >= deadline {
return false;
}
std::thread::sleep(RELOAD_POLL);
}
}
pub(crate) fn active_workers_in(proc: &Path, output_root: &Path) -> BTreeSet<u32> {
let Some(master) = std::fs::read_to_string(output_root.join("runtime/nginx.pid"))
.ok()
.and_then(|text| text.trim().parse::<u32>().ok())
else {
return BTreeSet::new();
};
let Ok(entries) = std::fs::read_dir(proc) else {
return BTreeSet::new();
};
entries
.flatten()
.filter_map(|entry| {
let pid = entry.file_name().to_str()?.parse::<u32>().ok()?;
let stat = std::fs::read_to_string(entry.path().join("stat")).ok()?;
let close = stat.rfind(')')?;
let parent: u32 = stat.get(close + 2..)?.split(' ').nth(1)?.parse().ok()?;
(parent == master).then_some(())?;
let cmd = std::fs::read(entry.path().join("cmdline")).ok()?;
(cmd.starts_with(WORKER) && !cmd.starts_with(SHUTTING_DOWN)).then_some(pid)
})
.collect()
}
/// when the oldest process that may read a retired generation started, since one started after the
/// retirement can't read it. readers by role: workers of the nginx master in the pid file, cgi
/// requests fcgiwrap runs, `git-ssh` sessions, and what those start. long-lived services such as
/// the firehose read no generation and don't count. unknown off linux or without nginx's pid file.
fn oldest_reader(output_root: &Path) -> Option<u64> {
// SAFETY: sysconf only reads a constant of the running system
let ticks = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
oldest_reader_in(Path::new("/proc"), output_root, u64::try_from(ticks).ok()?)
}
struct Process {
parent: u32,
name: String,
git_ssh: bool,
started: u64,
}
fn oldest_reader_in(proc: &Path, output_root: &Path, ticks: u64) -> Option<u64> {
let master: u32 = std::fs::read_to_string(output_root.join("runtime/nginx.pid"))
.ok()?
.trim()
.parse()
.ok()?;
let boot: u64 = std::fs::read_to_string(proc.join("stat"))
.ok()?
.lines()
.find_map(|line| line.strip_prefix("btime ")?.trim().parse().ok())?;
let mut processes = BTreeMap::new();
for entry in std::fs::read_dir(proc).ok()? {
let entry = entry.ok()?;
let Some(pid) = (entry.file_name().to_str()).and_then(|name| name.parse::<u32>().ok())
else {
continue;
};
// a process that exited since the listing reads nothing
let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else {
continue;
};
// the name is in parentheses and may hold spaces, the fields after it don't
let (open, close) = (stat.find('(')?, stat.rfind(')')?);
let fields: Vec<&str> = stat.get(close + 2..)?.split(' ').collect();
let command = std::fs::read(entry.path().join("cmdline")).unwrap_or_default();
let process = Process {
parent: fields.get(1)?.parse().ok()?,
name: stat[open + 1..close].to_string(),
git_ssh: command
.split(|&byte| byte == 0)
.any(|arg| arg == b"git-ssh"),
started: boot + fields.get(19)?.parse::<u64>().ok()? / ticks,
};
processes.insert(pid, process);
}
let me = std::process::id();
let reads = |mut pid: u32| {
while let Some(process) = processes.get(&pid) {
let parent = processes.get(&process.parent);
if process.parent == master
|| process.git_ssh
|| (process.name != "fcgiwrap"
&& parent.is_some_and(|parent| parent.name == "fcgiwrap"))
{
return true;
}
pid = process.parent;
}
false
};
let oldest = (processes.iter())
.filter(|&(&pid, _)| pid != me && reads(pid))
.map(|(_, process)| process.started)
.min();
Some(oldest.unwrap_or(u64::MAX))
}
#[derive(Deserialize, Serialize)]
struct Live {
#[serde(default)]
repos: Vec<LiveRepo>,
#[serde(default)]
stats: LiveStats,
#[serde(default)]
profile: LiveProfile,
}
#[derive(Deserialize, Serialize, Default)]
struct LiveProfile {
#[serde(default)]
atproto: BTreeMap<String, String>,
#[serde(default)]
atproto_entries: BTreeMap<String, String>,
#[serde(default)]
git_xrpc_entries: BTreeMap<String, String>,
}
#[derive(Default)]
struct Served {
tips: BTreeSet<String>,
seals: BTreeSet<String>,
git_seals: BTreeSet<String>,
}
#[derive(Deserialize, Serialize, PartialEq, Eq, PartialOrd, Ord)]
struct LiveRepo {
repo_did: String,
#[serde(default)]
release_id: Option<String>,
#[serde(default)]
git_export_hash: Option<String>,
#[serde(default)]
object_format: Option<String>,
}
#[derive(Deserialize, Serialize, Default)]
struct LiveStats {
#[serde(default)]
created_at_unix: u64,
}
impl Live {
fn load(output_root: &Path, generation: &str) -> Option<Self> {
let cached = output_root.join(LIVE_CACHE).join(generation);
if let Ok(bytes) = std::fs::read(&cached)
&& let Ok(live) = serde_json::from_slice(&bytes)
{
return Some(live);
}
let manifest = output_root
.join("generations")
.join(generation)
.join("manifest.json");
let live: Self = serde_json::from_slice(&std::fs::read(manifest).ok()?).ok()?;
if let Ok(bytes) = serde_json::to_vec(&live)
&& std::fs::create_dir_all(output_root.join(LIVE_CACHE)).is_ok()
{
let _ = crate::cas::atomic_write(&cached, &bytes);
}
Some(live)
}
fn served(&self, into: &mut BTreeMap<String, Served>) {
for (did, tip) in &self.profile.atproto {
into.entry(repo_key(did))
.or_default()
.tips
.insert(tip.clone());
}
for (did, seal) in &self.profile.atproto_entries {
into.entry(repo_key(did))
.or_default()
.seals
.insert(seal.clone());
}
for (did, seal) in &self.profile.git_xrpc_entries {
into.entry(repo_key(did))
.or_default()
.git_seals
.insert(seal.clone());
}
}
}
impl LiveRepo {
fn entries(&self, output_root: &Path) -> Vec<PathBuf> {
let dir = crate::repolock::repo_dir(output_root, &repo_key(&self.repo_did));
let release = self
.release_id
.as_ref()
.map(|id| dir.join("releases").join(id));
let store = RepoStore::open(&dir);
let format = (self.object_format.as_deref())
.and_then(|format| knot_types::ObjectFormat::from_capability(format));
let export = self.git_export_hash.iter().flat_map(|fingerprint| {
let entry = store.entry_path(crate::git_export::EXPORT_CACHE_KIND, fingerprint);
let filed =
format.map(|format| crate::partitions::filed_ref_deltas(&store, &entry, format));
std::iter::once(entry).chain(filed.into_iter().flatten())
});
release.into_iter().chain(export).collect()
}
}
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct Removed {
pub generations: usize,
pub store_entries: usize,
pub atproto_files: usize,
pub git_xrpc_files: usize,
pub staging: usize,
}
impl Removed {
pub fn is_empty(&self) -> bool {
*self == Self::default()
}
}
impl std::fmt::Display for Removed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"removed {} generations, {} releases and store entries, {} atproto files, {} git read files, {} abandoned staging directories",
self.generations,
self.store_entries,
self.atproto_files,
self.git_xrpc_files,
self.staging
)
}
}
pub async fn run(
config: Arc<ValidatedConfig>,
keep: Option<usize>,
force: bool,
) -> anyhow::Result<()> {
let keep = keep.unwrap_or(config.activate.keep_generations);
anyhow::ensure!(keep >= 1, "--keep must be at least 1");
let drain_secs = if force { 0 } else { DRAIN_SECS };
out!(
"gc: {}",
collect_locking(&config.output_root, keep, drain_secs)?
);
Ok(())
}
fn collect_locking(output_root: &Path, keep: usize, drain_secs: u64) -> anyhow::Result<Removed> {
let _output_lock = crate::output::lock(output_root)?;
crate::publish::reconcile(output_root)?;
let reader = oldest_reader(output_root);
collect_with(
output_root,
keep,
crate::publish::unix_now(),
drain_secs,
reader,
)
}
pub(crate) fn collect(output_root: &Path, keep: usize) -> anyhow::Result<Removed> {
let reader = oldest_reader(output_root);
collect_with(
output_root,
keep,
crate::publish::unix_now(),
DRAIN_SECS,
reader,
)
}
#[cfg(test)]
fn collect_at(
output_root: &Path,
keep: usize,
now: u64,
drain_secs: u64,
) -> anyhow::Result<Removed> {
collect_with(output_root, keep, now, drain_secs, None)
}
/// a retired generation stays for the drain window, or until nothing that started before its
/// retirement still runs, when `oldest_reader` says when the oldest of those started.
fn collect_with(
output_root: &Path,
keep: usize,
now: u64,
drain_secs: u64,
oldest_reader: Option<u64>,
) -> anyhow::Result<Removed> {
anyhow::ensure!(
!crate::tombstone::present(output_root),
"activation tombstone present, the served generation is unknown"
);
let mut removed = Removed::default();
let generations_root = output_root.join("generations");
let mut generations: Vec<(String, Option<Live>)> = Vec::new();
for entry in std::fs::read_dir(&generations_root).context("list generations")? {
let entry = entry?;
let Ok(name) = entry.file_name().into_string() else {
continue;
};
if name.starts_with('.') {
remove(&entry.path())?;
removed.staging += 1;
} else if entry.file_type()?.is_dir() {
let live = Live::load(output_root, &name);
generations.push((name, live));
}
}
let Some(current) = crate::publish::current_generation(output_root) else {
return Ok(removed);
};
let Some(current_created) = generations
.iter()
.find(|(name, _)| *name == current)
.and_then(|(_, live)| live.as_ref())
.map(|live| live.stats.created_at_unix)
else {
return Ok(removed);
};
let record = crate::publish::load_backend_record(output_root)?
.filter(|record| record.generation == current);
let activated_at = record
.as_ref()
.map_or(current_created, |record| record.activated_at_unix);
let superseded = record.as_ref().map_or(&[][..], |record| &record.superseded);
let on_disk = |id: &str| generations.iter().any(|(name, _)| name == id);
let mut kept: BTreeSet<String> = BTreeSet::from([current.clone()]);
kept.extend(
superseded
.iter()
.filter(|entry| now.saturating_sub(entry.at_unix) < drain_secs)
.filter(|entry| oldest_reader.is_none_or(|start| start <= entry.at_unix))
.map(|entry| entry.generation.clone()),
);
// build time only fills in generations no activation recorded.
let mut targets: Vec<&str> = superseded
.iter()
.rev()
.map(|entry| entry.generation.as_str())
.filter(|id| *id != current && on_disk(id))
.take(keep)
.collect();
let activated =
|id: &str| id == current || superseded.iter().any(|entry| entry.generation == id);
let mut unrecorded: Vec<(u64, &str)> = Vec::new();
for (name, live) in &generations {
let Some(live) = live else {
kept.insert(name.clone());
continue;
};
if activated(name) {
continue;
}
// a build made since the last activation is kept until the next one.
if live.stats.created_at_unix >= activated_at {
kept.insert(name.clone());
} else {
unrecorded.push((live.stats.created_at_unix, name));
}
}
unrecorded.sort_unstable_by(|a, b| b.cmp(a));
let missing = keep - targets.len();
targets.extend(unrecorded.into_iter().take(missing).map(|(_, name)| name));
kept.extend(targets.into_iter().map(str::to_string));
let on_the_way: Vec<String> = generations
.iter()
.map(|(name, _)| name)
.filter(|name| !kept.contains(*name) && pinned(output_root, name))
.cloned()
.collect();
kept.extend(on_the_way);
for (name, _) in &generations {
if !kept.contains(name) {
remove(&generations_root.join(name))?;
// nginx's temp dirs and logs, which a retired worker may still have open.
let runtime = output_root.join("runtime").join(name);
if runtime.exists() {
remove(&runtime)?;
}
removed.generations += 1;
}
}
if let Some(mut record) = record
&& record
.superseded
.iter()
.any(|entry| !kept.contains(&entry.generation))
{
record
.superseded
.retain(|entry| kept.contains(&entry.generation));
crate::publish::write_backend_record(output_root, &record)?;
}
// an unreadable kept manifest could name any entry, and nothing is swept.
let mut repos_live = BTreeSet::new();
let mut served = BTreeMap::new();
for (name, view) in &generations {
if kept.contains(name) {
let Some(view) = view else {
return Ok(removed);
};
repos_live.extend(&view.repos);
view.served(&mut served);
}
}
// most kept generations carry the same entries, so each is listed once
let live: BTreeSet<PathBuf> = repos_live
.into_iter()
.flat_map(|repo| repo.entries(output_root))
.collect();
sweep(&output_root.join(LIVE_CACHE), |path| {
path.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| kept.contains(name))
})?;
let chain_root = output_root.join("atproto");
let Ok(repos) = std::fs::read_dir(output_root.join("repos")) else {
return Ok(removed);
};
for repo in repos {
let repo = repo?;
for store_dir in STORE_DIRS {
let Ok(entries) = std::fs::read_dir(repo.path().join(store_dir)) else {
continue;
};
for entry in entries {
let path = entry?.path();
let staging = path
.file_name()
.is_some_and(|name| name.as_encoded_bytes().starts_with(b"."));
if staging {
remove(&path)?;
removed.staging += 1;
} else if !live.contains(&path) {
remove(&path)?;
removed.store_entries += 1;
}
}
}
let Ok(name) = repo.file_name().into_string() else {
continue;
};
// dot-files here are writes in flight, made without the output lock.
let store = RepoStore::open(&repo.path());
let kind = store.kind_dir(crate::partitions::REF_DELTAS_KIND);
removed.store_entries += sweep(&kind, |path| {
live.contains(path)
|| path
.file_name()
.is_some_and(|name| name.as_encoded_bytes().starts_with(b"."))
})?;
let mut named = served.remove(&name).unwrap_or_default();
removed.git_xrpc_files +=
collect_git_reads(&repo.path(), std::mem::take(&mut named.git_seals))?;
removed.atproto_files += collect_served(&repo.path(), &chain_root, &name, named)?;
if let Some(key) = RepoKey::encoded(&name) {
removed.store_entries += collect_cob_states(output_root, &repo.path(), &key)?;
}
}
removed.git_xrpc_files += sweep(&output_root.join(crate::git_xrpc::SHARED_BLOBS), |path| {
use std::os::unix::fs::MetadataExt as _;
std::fs::metadata(path).map_or(true, |meta| meta.nlink() > 1)
|| path
.file_name()
.is_some_and(|name| name.as_encoded_bytes().starts_with(b"."))
})?;
Ok(removed)
}
/// an unreadable seal would make its files look unreferenced. nothing gets swept.
/// a pool already swept for the same seals is left alone, so a write's gc only reads the repos it changed.
pub(crate) fn collect_git_reads(repo: &Path, mut seals: BTreeSet<String>) -> anyhow::Result<usize> {
let store = RepoStore::served(repo);
seals.extend(crate::git_xrpc::last_seal(&store));
if crate::git_xrpc::swept_for(&store, &seals) {
return Ok(0);
}
let Some(kept) = crate::git_xrpc::kept_served(&store, &seals) else {
return Ok(0);
};
let (pool, seal_dir, maps_dir) = crate::git_xrpc::served_dirs(&store);
let mut removed = sweep(&seal_dir, |path| kept.contains(path))?;
removed += sweep(&maps_dir, |path| kept.contains(path))?;
let pooled = crate::git_xrpc::POOLED.iter().map(|dir| pool.join(dir));
for dir in pooled.chain(crate::git_xrpc::paths::swept(&pool)) {
removed += sweep(&dir, |path| {
kept.contains(path)
|| path
.file_name()
.is_some_and(|name| name.as_encoded_bytes().starts_with(b"."))
})?;
}
crate::git_xrpc::mark_swept(&store, &seals)?;
Ok(removed)
}
/// an unreadable seal keeps every served file in this repo.
fn collect_served(
repo: &Path,
chain_root: &Path,
key: &str,
named: Served,
) -> anyhow::Result<usize> {
let store = RepoStore::served(repo);
let mut seals = named.seals;
seals.extend(crate::atproto::last_seal(chain_root, key));
let Some(kept) = crate::atproto::kept_served(&store, &seals, &named.tips) else {
return Ok(0);
};
let (pool, seal_dir) = crate::atproto::served_dirs(&store);
let shared: Vec<PathBuf> = crate::atproto::POOLED
.iter()
.map(|dir| pool.join(dir))
.collect();
let mut removed = sweep(&seal_dir, |path| kept.contains(path))?;
removed += sweep(&pool, |path| {
kept.contains(path) || shared.iter().any(|dir| dir == path)
})?;
for dir in &shared {
removed += sweep(dir, |path| kept.contains(path))?;
}
Ok(removed)
}
/// a non-json file is a write in flight and stays.
fn collect_cob_states(output_root: &Path, repo: &Path, key: &RepoKey) -> anyhow::Result<usize> {
let store = RepoStore::open(repo);
let records = Datastore::at(output_root);
let Some(kept) = crate::atproto::kept_cob_states(&store, &records, key) else {
return Ok(0);
};
let mut removed = 0;
for dir in crate::atproto::cob_state_dirs(&store) {
removed += sweep(&dir, |path| {
kept.contains(path) || path.extension().is_none_or(|ext| ext != "json")
})?;
}
Ok(removed)
}
fn sweep(dir: &Path, keep: impl Fn(&Path) -> bool) -> anyhow::Result<usize> {
let Ok(entries) = std::fs::read_dir(dir) else {
return Ok(0);
};
let mut removed = 0;
for entry in entries {
let path = entry?.path();
if !keep(&path) {
remove(&path)?;
removed += 1;
}
}
Ok(removed)
}
fn remove(path: &Path) -> anyhow::Result<()> {
let result = if std::fs::symlink_metadata(path)?.is_dir() {
std::fs::remove_dir_all(path)
} else {
std::fs::remove_file(path)
};
result.with_context(|| format!("remove {}", path.display()))
}
#[cfg(test)]
#[path = "tests/gc.rs"]
mod tests;