nordisk/holger

server/lib/src/promote/rows.rs

//! **The index rows of the intake chain** (lane HOL-PERF-2): one
//! [`holger_index_rows::StoredRow`] per artifact, by its sha256, at
//! `<prod_root>/lineage/index-rows/<sha256>.json`.
//!
//! A row is what the scan worker read out of the package for its format's index (the deb
//! control stanza, the rpm header, the Chart.yaml, the wheel METADATA, the npm package.json).
//! The intake stores it at INGEST (or at an airgap landing, from the bundle); the indexes the
//! chain serves (PEP 658, the npm packument, the chain views' apt, rpm-md and Helm indexes)
//! render from it, so the server never parses a package itself.
//!
//! **Written once.** The file is named by the bytes' sha256 and never replaced: a second
//! write of the same bytes finds it there. **Read once.** A read keeps the parsed row in
//! memory; a row that does not parse or check (a file damaged on disk) is no row, said in
//! the log, and the index serves the artifact without it.

use std::collections::HashMap;
use std::io::Write as _;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};

use holger_index_rows::{IndexRow, StoredRow};

use crate::lock::RwLockOrRecover as _;

/// The rows of one chain (see the module docs).
#[derive(Debug)]
pub struct RowStore {
    root: PathBuf,
    cache: RwLock<HashMap<String, Arc<StoredRow>>>,
}

impl RowStore {
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self {
            root: root.into(),
            cache: RwLock::new(HashMap::new()),
        }
    }

    fn path(&self, sha256: &str) -> Option<PathBuf> {
        // A sha256 is 64 lowercase hex digits; nothing else names a file here.
        (sha256.len() == 64
            && sha256
                .bytes()
                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)))
        .then(|| self.root.join(format!("{sha256}.json")))
    }

    /// **Store the row of the bytes `sha256`** (`size` bytes long), once. A row already
    /// stored for those bytes stays as it is.
    pub fn put(&self, sha256: &str, size: u64, row: &IndexRow) -> std::io::Result<()> {
        let path = self
            .path(sha256)
            .ok_or_else(|| std::io::Error::other(format!("'{sha256}' is not a sha256")))?;
        row.check().map_err(std::io::Error::other)?;
        let stored = StoredRow::new(sha256, size, row.clone());
        if !path.exists() {
            std::fs::create_dir_all(&self.root)?;
            let doc = serde_json::to_vec(&stored).map_err(std::io::Error::other)?;
            // A unique staging name, then a link: a concurrent writer of the same bytes
            // finds the file there, never half of it.
            let mut tmp = tempfile::NamedTempFile::new_in(&self.root)?;
            tmp.write_all(&doc)?;
            tmp.as_file().sync_all()?;
            match std::fs::hard_link(tmp.path(), &path) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
                Err(e) => return Err(e),
            }
        }
        self.cache
            .write_or_recover()
            .insert(sha256.to_string(), Arc::new(stored));
        Ok(())
    }

    /// **The row of the bytes `sha256`**, when one is stored and reads back checked.
    pub fn get(&self, sha256: &str) -> Option<Arc<StoredRow>> {
        if let Some(hit) = self.cache.read_or_recover().get(sha256) {
            return Some(Arc::clone(hit));
        }
        let path = self.path(sha256)?;
        let bytes = match std::fs::read(&path) {
            Ok(b) => b,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return None,
            Err(e) => {
                log::warn!("intake: the index row {} could not be read ({e}); the artifact is indexed without it", path.display());
                return None;
            }
        };
        match StoredRow::parse(&bytes, sha256) {
            Ok(s) => {
                let s = Arc::new(s);
                self.cache
                    .write_or_recover()
                    .insert(sha256.to_string(), Arc::clone(&s));
                Some(s)
            }
            Err(e) => {
                log::warn!(
                    "intake: the index row {} is refused ({e}); the artifact is indexed without it",
                    path.display()
                );
                None
            }
        }
    }
}