kcode_k1_web_cache/

lib.rs

use kcode_k1_transaction_id::TxId;
use kcode_k1_web_package::{DependencySelector, SourceFile, SourcePackage, WebFamily, WebId};
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

mod storage;

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct Digest(pub [u8; 32]);
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct SourceIdentity {
    pub generation: u64,
    pub digest: Digest,
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ResolutionIdentity {
    pub generation: u64,
    pub digest: Digest,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResolutionEntry {
    pub family: WebFamily,
    pub selector: DependencySelector,
    pub resolved: WebId,
    pub winning: TxId,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResolutionManifest {
    candidate: WebId,
    source: SourceIdentity,
    projection_cursor: Option<TxId>,
    entries: Vec<ResolutionEntry>,
}

impl ResolutionManifest {
    pub fn new(
        candidate: &SourcePackage,
        source: SourceIdentity,
        projection_cursor: Option<TxId>,
        mut entries: Vec<ResolutionEntry>,
    ) -> Result<Self, String> {
        entries.sort_by(|a, b| (&a.family, &a.selector).cmp(&(&b.family, &b.selector)));
        let value = Self {
            candidate: candidate.id().clone(),
            source,
            projection_cursor,
            entries,
        };
        value.validate(candidate)?;
        Ok(value)
    }
    pub fn entries(&self) -> &[ResolutionEntry] {
        &self.entries
    }
    pub fn route(
        &self,
        family: &WebFamily,
        selector: &DependencySelector,
    ) -> Option<&ResolutionEntry> {
        self.entries
            .iter()
            .find(|entry| &entry.family == family && &entry.selector == selector)
    }
    fn validate(&self, package: &SourcePackage) -> Result<(), String> {
        if self.source.generation == 0
            || self.candidate != *package.id()
            || self.source.digest != package_digest(package)
        {
            return Err("manifest source identity does not match candidate".into());
        }
        let mut expected = package
            .dependencies()
            .iter()
            .map(|dependency| {
                WebFamily::new(dependency.authority(), dependency.name().to_owned())
                    .map(|family| (family, dependency.selector().clone()))
                    .map_err(|e| e.to_string())
            })
            .collect::<Result<Vec<_>, _>>()?;
        expected.sort();
        let actual = self
            .entries
            .iter()
            .map(|entry| (entry.family.clone(), entry.selector.clone()))
            .collect::<Vec<_>>();
        let invalid = self.entries.iter().any(|entry| {
            entry.resolved.family() != &entry.family
                || !entry.selector.matches(entry.resolved.version())
        });
        if invalid || actual != expected {
            return Err("resolution entries are invalid".into());
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CheckIdentity {
    pub cache_epoch: u64,
    pub source: SourceIdentity,
    pub resolution: ResolutionIdentity,
    pub graph: Digest,
    pub projection_cursor: Option<TxId>,
    pub checker_executable: Digest,
    pub container_image: Digest,
    pub chromium_version: String,
    pub route_revision: String,
    pub harness_revision: String,
    pub check_policy_revision: String,
    pub command_policy: String,
}

pub struct UserWebCache {
    root: PathBuf,
    epoch: u64,
    identity: Vec<u8>,
}
#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Saved {
    epoch: u64,
    source_counter: u64,
    source: Option<(u64, [u8; 32], [u8; 32])>,
    resolution_counter: u64,
    resolution: Option<(u64, [u8; 32], Option<[u8; 12]>)>,
}
struct OpenTimer(Instant);
impl Drop for OpenTimer {
    fn drop(&mut self) {
        if self.0.elapsed() > Duration::from_millis(100) {
            eprintln!("{{\"level\":\"warning\",\"event\":\"k1_web_cache_open_slow\"}}");
        }
    }
}

impl UserWebCache {
    pub fn open(
        cache_root: impl AsRef<Path>,
        user: TxId,
        boot: &str,
        schema: &str,
        checker_policy: &str,
    ) -> Result<Self, String> {
        let _timer = OpenTimer(Instant::now());
        nonempty(&[boot, schema, checker_policy])?;
        storage::directory(cache_root.as_ref())?;
        let root = cache_root.as_ref().join(user.to_string());
        let identity = cache_identity(boot, schema, checker_policy);
        if let Ok((saved, _)) = load(&root, &identity, true) {
            return Ok(Self {
                root,
                epoch: saved.epoch,
                identity,
            });
        }
        let epoch = recover_epoch(&root)
            .unwrap_or(0)
            .checked_add(1)
            .ok_or("cache epoch exhausted")?;
        initialize(&root, epoch, &identity)?;
        Ok(Self {
            root,
            epoch,
            identity,
        })
    }
    pub fn reset(&mut self, boot: &str, schema: &str, checker_policy: &str) -> Result<(), String> {
        nonempty(&[boot, schema, checker_policy])?;
        let epoch = self.epoch.checked_add(1).ok_or("cache epoch exhausted")?;
        let identity = cache_identity(boot, schema, checker_policy);
        initialize(&self.root, epoch, &identity)?;
        (self.epoch, self.identity) = (epoch, identity);
        Ok(())
    }
    pub fn root(&self) -> &Path {
        &self.root
    }
    pub fn source_root(&self) -> PathBuf {
        self.root.join("source")
    }
    pub fn resolution_root(&self) -> PathBuf {
        self.root.join("resolution")
    }
    pub fn diagnostics_root(&self) -> PathBuf {
        self.root.join("diagnostics")
    }
    pub const fn epoch(&self) -> u64 {
        self.epoch
    }
    pub fn source_identity(&self) -> Result<Option<SourceIdentity>, String> {
        Ok(source_identity(&self.state()?))
    }
    pub fn resolution_identity(&self) -> Result<Option<ResolutionIdentity>, String> {
        Ok(resolution_identity(&self.state()?))
    }
    pub fn materialize(&self, package: &SourcePackage) -> Result<SourceIdentity, String> {
        let (mut saved, files) = load(&self.root, &self.identity, false)?;
        self.same_epoch(&saved)?;
        let source = package_digest(package);
        let tree = files_digest(package.files());
        if let Some(current) = source_identity(&saved)
            && current.digest == source
            && saved.source.is_some_and(|value| value.2 == tree.0)
            && files_digest(&files) == tree
        {
            return Ok(current);
        }
        let generation = saved
            .source_counter
            .checked_add(1)
            .ok_or("source generation exhausted")?;
        let identity = SourceIdentity {
            generation,
            digest: source,
        };
        saved.source_counter = generation;
        saved.source = Some((generation, source.0, tree.0));
        saved.resolution = None;
        self.mutate(|| {
            storage::replace_source(&self.source_root(), package.files())?;
            storage::replace_empty(&self.resolution_root())?;
            storage::replace_empty(&self.diagnostics_root())?;
            storage::atomic_write(&self.root.join("control/state"), &encode_saved(&saved))?;
            storage::remove_file(&self.root.join("control/receipt"))
        })?;
        Ok(identity)
    }
    pub fn record_resolution(
        &self,
        manifest: &ResolutionManifest,
    ) -> Result<ResolutionIdentity, String> {
        let mut saved = self.state()?;
        if source_identity(&saved) != Some(manifest.source) {
            return Err("resolution source identity is stale".into());
        }
        let bytes = encode_manifest(manifest);
        let digest = hash(&bytes);
        if let Some(current) = resolution_identity(&saved)
            && current.digest == digest
            && resolution_cursor(&saved) == manifest.projection_cursor
        {
            return Ok(current);
        }
        let generation = saved
            .resolution_counter
            .checked_add(1)
            .ok_or("resolution generation exhausted")?;
        let identity = ResolutionIdentity { generation, digest };
        saved.resolution_counter = generation;
        saved.resolution = Some((
            generation,
            digest.0,
            manifest.projection_cursor.map(|id| *id.as_bytes()),
        ));
        self.mutate(|| {
            storage::replace_manifest(&self.resolution_root(), &bytes)?;
            storage::replace_empty(&self.diagnostics_root())?;
            storage::atomic_write(&self.root.join("control/state"), &encode_saved(&saved))?;
            storage::remove_file(&self.root.join("control/receipt"))
        })?;
        Ok(identity)
    }
    pub fn has_check(&self, identity: &CheckIdentity) -> Result<bool, String> {
        let saved = self.state()?;
        let receipt = check_receipt(identity)?;
        if !self.current(&saved, identity) {
            return Ok(false);
        }
        Ok(storage::read(&self.root.join("control/receipt"))?.as_deref() == Some(&receipt.0))
    }
    pub fn record_check(&self, identity: &CheckIdentity) -> Result<(), String> {
        let saved = self.state()?;
        let receipt = check_receipt(identity)?;
        if !self.current(&saved, identity) {
            return Err("check identity is stale".into());
        }
        self.mutate(|| storage::atomic_write(&self.root.join("control/receipt"), &receipt.0))
    }
    fn state(&self) -> Result<Saved, String> {
        let (saved, _) = load(&self.root, &self.identity, true)?;
        self.same_epoch(&saved)?;
        Ok(saved)
    }
    fn same_epoch(&self, saved: &Saved) -> Result<(), String> {
        (saved.epoch == self.epoch)
            .then_some(())
            .ok_or_else(|| "cache epoch changed".into())
    }
    fn current(&self, saved: &Saved, identity: &CheckIdentity) -> bool {
        identity.cache_epoch == self.epoch
            && source_identity(saved) == Some(identity.source)
            && resolution_identity(saved) == Some(identity.resolution)
            && resolution_cursor(saved) == identity.projection_cursor
    }
    fn mutate<T>(&self, operation: impl FnOnce() -> Result<T, String>) -> Result<T, String> {
        fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(self.root.join("control/dirty"))
            .map_err(|e| e.to_string())?;
        let value = operation()?;
        fs::remove_file(self.root.join("control/dirty")).map_err(|e| e.to_string())?;
        Ok(value)
    }
}

fn initialize(root: &Path, epoch: u64, identity: &[u8]) -> Result<(), String> {
    let saved = Saved {
        epoch,
        source_counter: 0,
        source: None,
        resolution_counter: 0,
        resolution: None,
    };
    storage::replace_directory(root, |stage| {
        for name in ["control", "source", "resolution", "diagnostics"] {
            fs::create_dir(stage.join(name)).map_err(|e| e.to_string())?;
        }
        fs::write(stage.join("control/identity"), identity).map_err(|e| e.to_string())?;
        fs::write(stage.join("control/state"), encode_saved(&saved)).map_err(|e| e.to_string())
    })
}
fn load(root: &Path, identity: &[u8], exact: bool) -> Result<(Saved, Vec<SourceFile>), String> {
    let files = storage::inspect(root)?;
    let control = root.join("control");
    if storage::read(&control.join("dirty"))?.is_some() {
        return Err("cache mutation is incomplete".into());
    }
    if storage::read(&control.join("identity"))?.as_deref() != Some(identity) {
        return Err("cache identity mismatch".into());
    }
    let saved =
        decode_saved(&storage::read(&control.join("state"))?.ok_or("missing cache state")?)?;
    if exact
        && !saved
            .source
            .map_or(files.is_empty(), |value| files_digest(&files).0 == value.2)
    {
        return Err("materialized source mismatch".into());
    }
    match (
        saved.resolution,
        storage::read(&root.join("resolution/manifest"))?,
    ) {
        (Some(value), Some(bytes)) if hash(&bytes).0 == value.1 => {}
        (None, None) => {}
        _ => return Err("resolution manifest mismatch".into()),
    }
    let invalid_receipt = storage::read(&control.join("receipt"))?
        .is_some_and(|receipt| receipt.len() != 32 || saved.resolution.is_none());
    if invalid_receipt {
        return Err("invalid check receipt".into());
    }
    Ok((saved, files))
}

fn encode_manifest(value: &ResolutionManifest) -> Vec<u8> {
    let id = &value.candidate;
    let cursor = value
        .projection_cursor
        .map_or_else(|| "-".into(), |item| item.to_string());
    let mut text = format!(
        "K1WEBRESOLUTION1\n{}\n{}\n{}\n{}\n{}\n{}\n",
        id.family().authority(),
        id.family().logical_name(),
        id.version(),
        value.source.generation,
        hex(value.source.digest),
        cursor
    );
    for entry in &value.entries {
        text.push_str(&format!(
            "{}\t{}\t{}\t{}\t{}\n",
            entry.family.authority(),
            entry.family.logical_name(),
            entry.selector,
            entry.resolved.version(),
            entry.winning
        ));
    }
    text.into_bytes()
}
fn encode_saved(value: &Saved) -> Vec<u8> {
    serde_json::to_vec(value).expect("saved state serializes")
}
fn decode_saved(bytes: &[u8]) -> Result<Saved, String> {
    let value: Saved = serde_json::from_slice(bytes).map_err(|_| "invalid cache state")?;
    if encode_saved(&value) != bytes
        || value
            .source
            .is_some_and(|item| item.0 == 0 || item.0 > value.source_counter)
        || value
            .resolution
            .is_some_and(|item| item.0 == 0 || item.0 > value.resolution_counter)
    {
        return Err("noncanonical cache state".into());
    }
    Ok(value)
}
fn source_identity(value: &Saved) -> Option<SourceIdentity> {
    value.source.map(|item| SourceIdentity {
        generation: item.0,
        digest: Digest(item.1),
    })
}
fn resolution_identity(value: &Saved) -> Option<ResolutionIdentity> {
    value.resolution.map(|item| ResolutionIdentity {
        generation: item.0,
        digest: Digest(item.1),
    })
}
fn resolution_cursor(value: &Saved) -> Option<TxId> {
    value
        .resolution
        .and_then(|item| item.2.map(TxId::from_bytes))
}
fn package_digest(package: &SourcePackage) -> Digest {
    let mut hasher = Sha256::new();
    append(
        &mut hasher,
        package
            .id()
            .family()
            .authority()
            .transaction_id()
            .as_bytes(),
    );
    append(&mut hasher, package.id().family().logical_name().as_bytes());
    append(&mut hasher, package.id().version().to_string().as_bytes());
    append_files(&mut hasher, package.files());
    Digest(hasher.finalize().into())
}
fn files_digest(files: &[SourceFile]) -> Digest {
    let mut hasher = Sha256::new();
    append_files(&mut hasher, files);
    Digest(hasher.finalize().into())
}
fn append_files(hasher: &mut Sha256, files: &[SourceFile]) {
    for file in files {
        append(hasher, file.path().as_bytes());
        append(hasher, file.bytes());
    }
}
fn check_receipt(value: &CheckIdentity) -> Result<Digest, String> {
    nonempty(&[
        &value.chromium_version,
        &value.route_revision,
        &value.harness_revision,
        &value.check_policy_revision,
        &value.command_policy,
    ])?;
    let mut hasher = Sha256::new();
    for number in [
        value.cache_epoch,
        value.source.generation,
        value.resolution.generation,
    ] {
        append(&mut hasher, &number.to_le_bytes());
    }
    for digest in [
        value.source.digest,
        value.resolution.digest,
        value.graph,
        value.checker_executable,
        value.container_image,
    ] {
        append(&mut hasher, &digest.0);
    }
    match value.projection_cursor {
        Some(cursor) => {
            append(&mut hasher, &[1]);
            append(&mut hasher, cursor.as_bytes());
        }
        None => append(&mut hasher, &[0]),
    }
    for text in [
        &value.chromium_version,
        &value.route_revision,
        &value.harness_revision,
        &value.check_policy_revision,
        &value.command_policy,
    ] {
        append(&mut hasher, text.as_bytes());
    }
    Ok(Digest(hasher.finalize().into()))
}
fn recover_epoch(root: &Path) -> Option<u64> {
    serde_json::from_slice::<Saved>(&storage::read(&root.join("control/state")).ok().flatten()?)
        .ok()
        .map(|value| value.epoch)
}
fn cache_identity(boot: &str, schema: &str, policy: &str) -> Vec<u8> {
    serde_json::to_vec(&(boot, schema, policy)).expect("cache identity serializes")
}
fn hash(bytes: &[u8]) -> Digest {
    Digest(Sha256::digest(bytes).into())
}
fn append(hasher: &mut Sha256, bytes: &[u8]) {
    hasher.update((bytes.len() as u64).to_le_bytes());
    hasher.update(bytes);
}
fn nonempty(values: &[&str]) -> Result<(), String> {
    (!values.iter().any(|value| value.is_empty()))
        .then_some(())
        .ok_or_else(|| "identity values must be nonempty".into())
}
fn hex(value: Digest) -> String {
    value.0.iter().map(|byte| format!("{byte:02x}")).collect()
}