newton-core 0.7.5

//! Content-addressed object resolution over a [`UserStorageBackend`].
//!
//! Objects live in the configured backend under `objects/<cid>` and every read
//! from it is integrity-checked before the bytes are returned. With a
//! [`LocalObjectCache`], verified objects are kept on local disk and the backend is
//! read only on a miss.

use std::{fmt::Display, sync::Arc};

use cid::{multihash::Multihash, Cid};
use eyre::{Context, Result};
use tracing::{error, instrument, Span};

use super::{
    backend::UserStorageBackend,
    local_cache::{FillOutcome, LocalObjectCache},
};
use crate::cid::{cid_v1_raw_cid, parse_persisted_cid, sha2_256_multihash, verify_raw_cid, DAG_PB_CODEC, RAW_CODEC};

/// Backend prefix under which content-addressed objects are stored.
const OBJECTS_PREFIX: &str = "objects";
/// Trusted SHA2-256 multihash sidecars for migrated DAG-PB file bytes.
const OBJECT_CHECKSUMS_PREFIX: &str = "object-checksums";

/// Resolve content-addressed objects (WASM, Rego, schemas) by CID.
///
/// Raw payloads are verified directly against their CID. DAG-PB payloads are
/// verified against the checksum sidecar created by the explicit migration.
#[derive(Clone)]
pub struct PersistedImmutableDataResolver {
    backend: Arc<dyn UserStorageBackend>,
    /// Local disk tier in front of `backend`, holding objects already verified.
    local: Option<Arc<LocalObjectCache>>,
}

impl std::fmt::Debug for PersistedImmutableDataResolver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PersistedImmutableDataResolver").finish_non_exhaustive()
    }
}

impl PersistedImmutableDataResolver {
    /// Build a resolver over `backend`.
    pub fn new(backend: Arc<dyn UserStorageBackend>) -> Self {
        Self { backend, local: None }
    }

    /// Serve objects from `local` and read the backend only on a miss.
    #[must_use]
    pub fn with_local_cache(mut self, local: Arc<LocalObjectCache>) -> Self {
        self.local = Some(local);
        self
    }

    /// Backend key for a content-addressed object: `objects/<cid>`.
    pub fn object_key(cid: impl Display) -> String {
        format!("{OBJECTS_PREFIX}/{cid}")
    }

    /// Backend key for a DAG-PB object's SHA2-256 multihash sidecar.
    pub fn checksum_key(cid: impl Display) -> String {
        format!("{OBJECT_CHECKSUMS_PREFIX}/{cid}.sha256")
    }

    /// Resolve `cid` to its caller-facing content.
    pub async fn get(&self, cid: &str) -> Result<Vec<u8>> {
        let parsed = parse_persisted_cid(cid).map_err(|error| eyre::eyre!(error))?;
        self.get_cid(&parsed).await
    }

    /// Resolve an already-parsed `cid` to its caller-facing content.
    pub async fn get_cid(&self, cid: &Cid) -> Result<Vec<u8>> {
        self.get_optional(cid)
            .await?
            .ok_or_else(|| eyre::eyre!("object {cid} not found in persisted immutable data backend"))
    }

    /// Probe the local cache, then the configured backend, distinguishing absence from
    /// failure. A backend read on a miss is kept in the local cache.
    pub async fn get_optional(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
        self.fetch(cid, true).await
    }

    /// Like [`Self::get_optional`], but a backend read is not kept in the local cache.
    /// For untrusted callers (the public object endpoint), so outside traffic cannot
    /// evict the objects tasks use.
    pub async fn get_optional_without_caching(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
        self.fetch(cid, false).await
    }

    /// Traced as `storage.object.fetch`; `storage.cache` is hit|miss|shared|off. A failed
    /// fetch is logged here with its full error chain, once, so callers can return only
    /// the outer context to clients.
    #[instrument(
        name = "storage.object.fetch",
        target = "newton::task_evaluation",
        level = "debug",
        skip(self),
        fields(
            storage.backend = self.backend.backend_name(),
            storage.object.cid = %cid,
            storage.cache = tracing::field::Empty,
        )
    )]
    async fn fetch(&self, cid: &Cid, fill: bool) -> Result<Option<Vec<u8>>> {
        let span = Span::current();
        let Some(local) = &self.local else {
            span.record("storage.cache", "off");
            return log_failure(
                cid,
                self.fetch_verified(cid).await.map(|found| found.map(|(data, _)| data)),
            );
        };
        let name = cid.to_string();
        if let Some(bytes) = local.read(&name).await {
            span.record("storage.cache", "hit");
            return Ok(Some(bytes));
        }
        if !fill {
            span.record("storage.cache", "miss");
            return log_failure(
                cid,
                self.fetch_verified(cid).await.map(|found| found.map(|(data, _)| data)),
            );
        }
        // One backend read per CID however many tasks miss at once; the tasks that
        // waited share its outcome, failure and absence included.
        let mut slot = local.fill_lock(&name).await;
        if let Some(outcome) = slot.as_ref() {
            span.record("storage.cache", "shared");
            return match outcome {
                Ok(found) => Ok(found.as_deref().map(<[u8]>::to_vec)),
                Err(message) => Err(eyre::eyre!("{message}")),
            };
        }
        if let Some(bytes) = local.read(&name).await {
            span.record("storage.cache", "hit");
            return Ok(Some(bytes));
        }
        span.record("storage.cache", "miss");
        let fetched = log_failure(cid, self.fetch_verified(cid).await);
        let outcome: FillOutcome = match &fetched {
            Ok(found) => Ok(found.as_ref().map(|(data, _)| Arc::from(data.as_slice()))),
            Err(e) => Err(Arc::from(e.to_string())),
        };
        *slot = Some(outcome);
        let fetched = fetched?;
        if let Some((data, expected)) = &fetched {
            local.write(&name, expected, data).await;
        }
        Ok(fetched.map(|(data, _)| data))
    }

    /// Read `cid` from the backend and verify it, returning the bytes and the SHA2-256
    /// multihash they were verified against.
    async fn fetch_verified(&self, cid: &Cid) -> Result<Option<(Vec<u8>, Multihash<64>)>> {
        let key = Self::object_key(cid);
        let Some(data) = self
            .backend
            .get_optional(&key)
            .await
            .wrap_err_with(|| format!("resolve object {cid}"))?
        else {
            return Ok(None);
        };
        let expected = match cid.codec() {
            RAW_CODEC => {
                verify_raw_cid(cid, &data).map_err(|e| eyre::eyre!("integrity check failed for {cid}: {e}"))?;
                sha2_256_multihash(&data)
            }
            DAG_PB_CODEC => {
                let checksum_key = Self::checksum_key(cid);
                let expected = self
                    .backend
                    .get_optional(&checksum_key)
                    .await
                    .wrap_err_with(|| format!("resolve DAG-PB object checksum {cid}"))?
                    .ok_or_else(|| eyre::eyre!("DAG-PB object {cid} is missing its checksum sidecar"))?;
                let expected = Multihash::<64>::from_bytes(&expected)
                    .map_err(|error| eyre::eyre!("invalid checksum multihash for DAG-PB object {cid}: {error}"))?;
                if expected != sha2_256_multihash(&data) {
                    eyre::bail!("integrity check failed for DAG-PB object {cid}: SHA2-256 checksum mismatch");
                }
                expected
            }
            codec => eyre::bail!("unsupported CID codec for {cid}: 0x{codec:x}"),
        };

        Ok(Some((data, expected)))
    }

    /// Resolve `cid` to UTF-8 text (schemas, Rego source).
    pub async fn get_text(&self, cid: &str) -> Result<String> {
        let data = self.get(cid).await?;
        String::from_utf8(data).wrap_err_with(|| format!("object {cid} is not valid UTF-8"))
    }

    /// Store `data` content-addressed, returning its CIDv1.
    ///
    /// Idempotent: re-storing identical bytes yields the same CID and overwrites
    /// the same key with identical content.
    pub async fn put(&self, data: &[u8]) -> Result<String> {
        let cid = cid_v1_raw_cid(data);
        // Not written to the local cache: uploads are not task reads, and caching them
        // would let uploaders evict the objects tasks use. The first task read fills it.
        self.backend.put(&Self::object_key(cid), data).await?;
        Ok(cid.to_string())
    }
}

/// Log a failed fetch with its full error chain.
fn log_failure<T>(cid: &Cid, result: Result<T>) -> Result<T> {
    if let Err(e) = &result {
        error!(cid = %cid, error = format!("{e:#}"), "object fetch failed");
    }
    result
}

// ObjectFetcher impl — gated on `rpc` feature to match the trait definition.
#[cfg(feature = "rpc")]
#[async_trait::async_trait]
impl crate::common::task::rpc::ObjectFetcher for PersistedImmutableDataResolver {
    async fn get_object(&self, cid: &str) -> eyre::Result<Vec<u8>> {
        self.get(cid).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::persisted_immutable_data::backend::LocalBackend;

    #[tokio::test]
    async fn put_then_get_roundtrips() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));

        let cid = resolver.put(b"policy rego source").await.unwrap();
        let got = resolver.get(&cid).await.unwrap();
        assert_eq!(got, b"policy rego source");
    }

    #[tokio::test]
    async fn local_cache_serves_hits_without_the_backend() {
        let backend_dir = tempfile::tempdir().unwrap();
        let cache_dir = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(backend_dir.path()));
        let cid = PersistedImmutableDataResolver::new(backend.clone())
            .put(b"policy rego source")
            .await
            .unwrap();
        let local = Arc::new(LocalObjectCache::open(cache_dir.path(), 1 << 20).unwrap());
        let resolver = PersistedImmutableDataResolver::new(backend.clone()).with_local_cache(local.clone());

        // Miss: read from the backend, verified, then kept locally.
        assert_eq!(resolver.get(&cid).await.unwrap(), b"policy rego source");
        assert!(local.read(&cid).await.is_some());

        // Hit: served even with the backend copy gone.
        let parsed = parse_persisted_cid(&cid).unwrap();
        backend
            .delete(&PersistedImmutableDataResolver::object_key(parsed))
            .await
            .unwrap();
        assert_eq!(resolver.get(&cid).await.unwrap(), b"policy rego source");
    }

    #[tokio::test]
    async fn local_cache_never_keeps_an_object_that_fails_verification() {
        let backend_dir = tempfile::tempdir().unwrap();
        let cache_dir = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(backend_dir.path()));
        let cid = PersistedImmutableDataResolver::new(backend.clone())
            .put(b"original")
            .await
            .unwrap();
        let parsed = parse_persisted_cid(&cid).unwrap();
        backend
            .put(&PersistedImmutableDataResolver::object_key(parsed), b"tampered")
            .await
            .unwrap();
        let local = Arc::new(LocalObjectCache::open(cache_dir.path(), 1 << 20).unwrap());
        let resolver = PersistedImmutableDataResolver::new(backend).with_local_cache(local.clone());

        assert!(resolver.get(&cid).await.is_err());
        assert!(local.read(&cid).await.is_none());
    }

    #[tokio::test]
    async fn uploads_and_uncached_reads_do_not_fill_the_local_cache() {
        let backend_dir = tempfile::tempdir().unwrap();
        let cache_dir = tempfile::tempdir().unwrap();
        let local = Arc::new(LocalObjectCache::open(cache_dir.path(), 1 << 20).unwrap());
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(backend_dir.path())))
            .with_local_cache(local.clone());

        let cid = resolver.put(b"uploaded").await.unwrap();
        assert!(local.read(&cid).await.is_none());

        let parsed = parse_persisted_cid(&cid).unwrap();
        let got = resolver.get_optional_without_caching(&parsed).await.unwrap();
        assert_eq!(got.as_deref(), Some(b"uploaded".as_slice()));
        assert!(local.read(&cid).await.is_none());

        // A task read fills it.
        resolver.get(&cid).await.unwrap();
        assert!(local.read(&cid).await.is_some());
    }

    #[tokio::test]
    async fn waiters_share_the_fill_outcome_instead_of_refetching() {
        let backend_dir = tempfile::tempdir().unwrap();
        let cache_dir = tempfile::tempdir().unwrap();
        let local = Arc::new(LocalObjectCache::open(cache_dir.path(), 1 << 20).unwrap());
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(backend_dir.path())))
            .with_local_cache(local.clone());
        let cid = parse_persisted_cid(&crate::cid::cid_v1_raw(b"absent")).unwrap();

        // A finished fill recorded "not found"; a task that waited on it gets that answer.
        let mut slot = local.fill_lock(&cid.to_string()).await;
        *slot = Some(Ok(None));
        let waiter = {
            let resolver = resolver.clone();
            tokio::spawn(async move { resolver.get_optional(&cid).await })
        };
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        drop(slot);
        assert!(waiter.await.unwrap().unwrap().is_none());
    }

    #[tokio::test]
    async fn missing_object_fails_backend_only() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));
        let cid = crate::cid::cid_v1_raw(b"missing");

        let error = resolver.get(&cid).await.unwrap_err();
        assert!(error.to_string().contains("not found"), "{error}");
    }

    #[tokio::test]
    async fn get_rejects_corrupted_object() {
        let tmp = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(tmp.path()));
        let resolver = PersistedImmutableDataResolver::new(backend.clone());

        let cid = resolver.put(b"original").await.unwrap();
        let parsed = parse_persisted_cid(&cid).unwrap();
        backend
            .put(&PersistedImmutableDataResolver::object_key(parsed), b"tampered")
            .await
            .unwrap();

        let err = resolver.get(&cid).await.unwrap_err();
        assert!(err.to_string().contains("integrity check failed"), "{err}");
    }

    #[tokio::test]
    async fn get_text_reads_utf8() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));
        let cid = resolver.put("схема".as_bytes()).await.unwrap();
        assert_eq!(resolver.get_text(&cid).await.unwrap(), "схема");
    }

    #[tokio::test]
    async fn dag_pb_object_requires_matching_checksum_sidecar() {
        use sha2::{Digest, Sha256};

        let tmp = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(tmp.path()));
        let resolver = PersistedImmutableDataResolver::new(backend.clone());
        let payload = b"migrated DAG-PB policy";
        let root_block_digest = Sha256::digest(b"dag-pb root block");
        let multihash = Multihash::<64>::wrap(0x12, root_block_digest.as_slice()).unwrap();
        let cid = Cid::new_v1(crate::cid::DAG_PB_CODEC, multihash);

        backend
            .put(&PersistedImmutableDataResolver::object_key(cid), payload)
            .await
            .unwrap();
        assert!(resolver
            .get(&cid.to_string())
            .await
            .unwrap_err()
            .to_string()
            .contains("checksum sidecar"));

        let checksum = sha2_256_multihash(payload).to_bytes();
        backend
            .put(&PersistedImmutableDataResolver::checksum_key(cid), &checksum)
            .await
            .unwrap();
        assert_eq!(resolver.get(&cid.to_string()).await.unwrap(), payload);

        let wrong_checksum = sha2_256_multihash(b"different payload").to_bytes();
        backend
            .put(&PersistedImmutableDataResolver::checksum_key(cid), &wrong_checksum)
            .await
            .unwrap();
        assert!(resolver
            .get(&cid.to_string())
            .await
            .unwrap_err()
            .to_string()
            .contains("checksum mismatch"));
    }
}