server/lib/src/promote/catalog.rs
//! **The AVAILABLE catalog** (HOL-PERF-1): the intake record's reduction, in memory.
//!
//! The intake record ([`super::intake::IntakeLog`]) is one JSON-lines file per coordinate,
//! and it stays the truth. Every index the chain serves (the pip simple index, the npm
//! packument, the chain views' sets) used to be a scan of that whole directory per
//! request: every file read and JSON-parsed, 40 ms at 6 000 AVAILABLE artifacts and
//! linear in the chain's size (HOL-PERF-FORMATS). The catalog holds what the scan
//! computed:
//!
//! * the status of every coordinate, by its file key;
//! * the AVAILABLE set, and three indexes over it: the pip projects, the npm packages
//! and the first path segment a chain view names (`apt/`, `maven/`, ...), each prefix
//! with the generation it last changed at.
//!
//! **Built once, then kept.** The first read that needs the whole set scans the record
//! once (a restart rebuilds it from the files). Every append of THIS process updates it
//! in place, under the coordinate's own lock (the chain moves one artifact at a time),
//! and bumps the generation when the AVAILABLE set changed.
//!
//! **Another process over the same store** (a CLI's `holger sync invalidate`, a
//! `bundle import`) changes the record too. Every append therefore also writes one line,
//! `<writer> <key>`, to the journal beside the record (`lineage/intake.journal`, append
//! only). A read first compares the journal's length with what it has applied (one
//! `stat`); lines another writer appended name the coordinates to read again from their
//! files. So the download gate and the indexes see another process's change at their
//! next request, and a quarantine made by a CLI is never served from a stale catalog.
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::io::{Read as _, Seek as _, Write as _};
use std::path::{Path, PathBuf};
use std::sync::{RwLock, RwLockReadGuard};
use traits::ArtifactId;
use super::intake::{
id_from_label, normalize_project, python_project_of, ArtifactState, IntakeEntry, IntakeStatus,
};
use crate::lock::RwLockOrRecover as _;
/// The reduction of one record file's text: the last entry's state.
pub(crate) fn reduce(text: &str) -> Option<IntakeStatus> {
let last: IntakeEntry = text
.lines()
.filter(|l| !l.trim().is_empty())
.filter_map(|l| serde_json::from_str::<IntakeEntry>(l).ok())
.next_back()?;
Some(status_of(last))
}
/// The status a record stands at after `last`.
pub(crate) fn status_of(last: IntakeEntry) -> IntakeStatus {
let held_from = (last.to == ArtifactState::Quarantined).then_some(last.from);
IntakeStatus {
state: last.to,
held_from,
sha256: last.sha256.clone(),
last,
}
}
/// An AVAILABLE coordinate with no namespace and no version: the raw paths the
/// production repository, the npm door and the chain views serve.
fn plain(id: &ArtifactId) -> bool {
id.namespace.as_deref().unwrap_or("").is_empty() && id.version.is_empty()
}
/// One chain view's set: the AVAILABLE files under `<prefix>`, the prefix stripped.
#[derive(Default)]
struct Prefix {
/// The catalog generation this set last changed at.
generation: u64,
files: BTreeMap<String, ArtifactId>,
}
#[derive(Default)]
struct State {
/// Journal bytes applied (complete lines only).
consumed: u64,
/// Bumped on every change of the AVAILABLE set.
generation: u64,
by_key: HashMap<String, (Option<ArtifactId>, IntakeStatus)>,
/// The AVAILABLE coordinates, by file key (the record's order).
available: BTreeMap<String, ArtifactId>,
/// pip: normalized project → (coordinate path → sha256).
pip: BTreeMap<String, BTreeMap<String, String>>,
/// npm: package → the file keys of its AVAILABLE versions.
npm: HashMap<String, BTreeSet<String>>,
/// The chain views: `<first path segment>/` → its set.
prefixes: HashMap<String, Prefix>,
}
impl State {
/// Set `key`'s status; keep the AVAILABLE indexes and the generation in step.
fn apply(&mut self, key: &str, status: IntakeStatus) {
let id = id_from_label(&status.last.artifact, key);
let was = self.by_key.get(key).and_then(|(id, st)| {
(st.state == ArtifactState::Available).then(|| (id.clone(), st.sha256.clone()))
});
let now =
(status.state == ArtifactState::Available).then(|| (id.clone(), status.sha256.clone()));
self.by_key.insert(key.to_string(), (id, status));
if was == now {
return;
}
self.generation += 1;
if let Some((Some(old), _)) = was {
self.unindex(key, &old);
}
if let Some((Some(new), sha)) = now {
self.index(key, new, sha);
}
}
fn index(&mut self, key: &str, id: ArtifactId, sha256: String) {
if plain(&id) {
let base = id.name.rsplit('/').next().unwrap_or(&id.name);
if let Some(project) = python_project_of(base) {
self.pip
.entry(project)
.or_default()
.insert(id.name.clone(), sha256);
}
if let Some((package, _)) = super::npm::coords_of(&id) {
self.npm.entry(package).or_default().insert(key.to_string());
}
if let Some((first, rel)) = id.name.split_once('/') {
if !rel.is_empty() {
let p = self.prefixes.entry(format!("{first}/")).or_default();
p.files.insert(rel.to_string(), id.clone());
p.generation = self.generation;
}
}
}
self.available.insert(key.to_string(), id);
}
fn unindex(&mut self, key: &str, id: &ArtifactId) {
self.available.remove(key);
if !plain(id) {
return;
}
let base = id.name.rsplit('/').next().unwrap_or(&id.name);
if let Some(project) = python_project_of(base) {
if let Some(files) = self.pip.get_mut(&project) {
files.remove(&id.name);
if files.is_empty() {
self.pip.remove(&project);
}
}
}
if let Some((package, _)) = super::npm::coords_of(id) {
if let Some(keys) = self.npm.get_mut(&package) {
keys.remove(key);
if keys.is_empty() {
self.npm.remove(&package);
}
}
}
if let Some((first, rel)) = id.name.split_once('/') {
if let Some(p) = self.prefixes.get_mut(&format!("{first}/")) {
p.files.remove(rel);
p.generation = self.generation;
}
}
}
fn status(&self, key: &str) -> Option<IntakeStatus> {
self.by_key.get(key).map(|(_, st)| st.clone())
}
fn with_status(&self, key: &str, id: &ArtifactId) -> Option<(ArtifactId, IntakeStatus)> {
self.by_key.get(key).map(|(_, st)| (id.clone(), st.clone()))
}
}
/// **The catalog of one intake record** (see the module docs).
pub(crate) struct Catalog {
/// `lineage/intake`: the record's files.
root: PathBuf,
/// `lineage/intake.journal`: one `<writer> <key>` line per append, any process.
journal: PathBuf,
/// This catalog's writer tag in the journal: `<pid>.<instance>`.
me: String,
/// `None` until the first read that needs the whole set.
state: RwLock<Option<State>>,
}
impl std::fmt::Debug for Catalog {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Catalog")
.field("root", &self.root)
.field("me", &self.me)
.finish_non_exhaustive()
}
}
fn journal_len(path: &Path) -> u64 {
std::fs::metadata(path).map(|m| m.len()).unwrap_or(0)
}
impl Catalog {
pub(crate) fn new(root: PathBuf) -> Self {
static INSTANCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let n = INSTANCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let journal = root.with_extension("journal");
Self {
root,
journal,
me: format!("{}.{n}", std::process::id()),
state: RwLock::new(None),
}
}
/// **This process appended `entry` to `key`'s record** (already on disk): say so in
/// the journal for the other processes, and apply it here when the catalog is built.
pub(crate) fn applied(&self, key: &str, entry: &IntakeEntry) {
let line = format!("{} {key}\n", self.me);
let wrote = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&self.journal)
.and_then(|mut f| f.write_all(line.as_bytes()));
if let Err(e) = wrote {
log::warn!(
"intake: the catalog journal {} could not be written ({e}); another process over this store sees {key}'s change only after its restart",
self.journal.display()
);
}
let mut g = self.state.write_or_recover();
if let Some(s) = g.as_mut() {
s.apply(key, status_of(entry.clone()));
}
}
/// The whole record, scanned once.
fn load(&self) -> State {
let mut s = State {
consumed: journal_len(&self.journal),
..State::default()
};
if let Ok(rd) = std::fs::read_dir(&self.root) {
for e in rd.flatten() {
let name = e.file_name().to_string_lossy().to_string();
let Some(key) = name.strip_suffix(".jsonl") else {
continue;
};
let Some(st) = std::fs::read_to_string(e.path())
.ok()
.as_deref()
.and_then(reduce)
else {
continue;
};
s.apply(key, st);
}
}
s.generation += 1;
s
}
/// Apply the journal lines other writers appended since the last read.
fn catch_up(&self, s: &mut State, len: u64) {
let mut tail = Vec::new();
let read = std::fs::File::open(&self.journal).and_then(|mut f| {
f.seek(std::io::SeekFrom::Start(s.consumed))?;
f.take(len.saturating_sub(s.consumed))
.read_to_end(&mut tail)
});
if read.is_err() {
return;
}
// Complete lines only: a line still being written is read next time.
let Some(end) = tail.iter().rposition(|b| *b == b'\n') else {
return;
};
let text = String::from_utf8_lossy(&tail[..=end]).into_owned();
s.consumed += (end + 1) as u64;
let theirs: BTreeSet<&str> = text
.lines()
.filter_map(|l| l.split_once(' '))
.filter(|(who, _)| *who != self.me)
.map(|(_, key)| key)
.collect();
for key in theirs {
let path = self.root.join(format!("{key}.jsonl"));
if let Some(st) = std::fs::read_to_string(path)
.ok()
.as_deref()
.and_then(reduce)
{
s.apply(key, st);
}
}
}
/// The catalog, built and current: one `stat` of the journal when nothing changed.
fn fresh(&self) -> RwLockReadGuard<'_, Option<State>> {
let len = journal_len(&self.journal);
{
let g = self.state.read_or_recover();
if g.as_ref().is_some_and(|s| s.consumed == len) {
return g;
}
}
{
let mut g = self.state.write_or_recover();
match g.as_mut() {
// A journal shorter than what was applied was replaced: build again.
Some(s) if s.consumed <= len => self.catch_up(s, len),
_ => *g = Some(self.load()),
}
}
self.state.read_or_recover()
}
fn read<T>(&self, f: impl FnOnce(&State) -> T) -> T {
let g = self.fresh();
f(g.as_ref().expect("fresh() builds the catalog"))
}
/// `key`'s status when the catalog is built (`None` = not built: read the file).
pub(crate) fn status_if_built(&self, key: &str) -> Option<Option<IntakeStatus>> {
if self.state.read_or_recover().is_none() {
return None;
}
Some(self.read(|s| s.status(key)))
}
/// Every coordinate with a record, by file key.
pub(crate) fn all(&self) -> Vec<(String, IntakeStatus)> {
self.read(|s| {
let mut out: Vec<(String, IntakeStatus)> = s
.by_key
.iter()
.map(|(k, (_, st))| (k.clone(), st.clone()))
.collect();
out.sort_by(|a, b| a.0.cmp(&b.0));
out
})
}
/// The AVAILABLE set, in key order.
pub(crate) fn available(&self) -> Vec<(ArtifactId, IntakeStatus)> {
self.read(|s| {
s.available
.iter()
.filter_map(|(k, id)| s.with_status(k, id))
.collect()
})
}
/// The generation of the AVAILABLE set: it changes whenever the set does.
pub(crate) fn generation(&self) -> u64 {
self.read(|s| s.generation)
}
/// The pip projects with an AVAILABLE distribution.
pub(crate) fn pip_projects(&self) -> Vec<String> {
self.read(|s| s.pip.keys().cloned().collect())
}
/// The AVAILABLE distributions of `project` (normalized here): (path, sha256).
pub(crate) fn pip_files(&self, project: &str) -> Vec<(String, String)> {
let want = normalize_project(project);
self.read(|s| {
s.pip
.get(&want)
.map(|f| f.iter().map(|(p, h)| (p.clone(), h.clone())).collect())
.unwrap_or_default()
})
}
/// The AVAILABLE versions' coordinates and statuses of the npm `package`.
pub(crate) fn npm_versions(&self, package: &str) -> Vec<(ArtifactId, IntakeStatus)> {
self.read(|s| {
s.npm
.get(package)
.map(|keys| {
keys.iter()
.filter_map(|k| s.available.get(k).and_then(|id| s.with_status(k, id)))
.collect()
})
.unwrap_or_default()
})
}
/// The generation `prefix`'s set (`<name>/`) last changed at; 0 when it never held a file.
pub(crate) fn prefix_generation(&self, prefix: &str) -> u64 {
self.read(|s| match Self::first_segment(prefix) {
Some(p) => s.prefixes.get(p).map(|p| p.generation).unwrap_or(0),
None => s.generation,
})
}
/// The AVAILABLE files under `prefix` (`<name>/`), the prefix stripped, with the
/// generation the set is at.
pub(crate) fn available_under(&self, prefix: &str) -> (u64, Vec<(String, ArtifactId)>) {
self.read(|s| match Self::first_segment(prefix) {
Some(p) => s
.prefixes
.get(p)
.map(|p| {
(
p.generation,
p.files
.iter()
.map(|(r, id)| (r.clone(), id.clone()))
.collect(),
)
})
.unwrap_or((0, Vec::new())),
// A prefix deeper than one segment: the AVAILABLE set, filtered.
None => (
s.generation,
s.available
.values()
.filter(|id| plain(id))
.filter_map(|id| {
id.name
.strip_prefix(prefix)
.filter(|r| !r.is_empty())
.map(|r| (r.to_string(), id.clone()))
})
.collect(),
),
})
}
/// `prefix` when it is one path segment and its slash (`apt/`), the catalog's key.
fn first_segment(prefix: &str) -> Option<&str> {
let seg = prefix.strip_suffix('/')?;
(!seg.is_empty() && !seg.contains('/')).then_some(prefix)
}
}