runtime/

offload.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
// Copyright (c) 2026 2youg1 and the sprawling contributors

//! The shared shrink primitive: lossy-but-restorable,
//! owned here and nowhere else. Four invariants, each enforced in code:
//! store-before-cut; substitute (hint included) never larger than the
//! original nor the cap; only lossy transforms store; the substitute
//! always carries a materialized read-only rest path. CAS dedup makes
//! repeated offloads of the same bytes idempotent.

use std::path::Path;

use kernel::{Address, AxCode, AxError, Locator};
use storage::Cas;

/// The directory inside the room that holds materialized rest files.
pub const REST_DIR: &str = ".rest";

/// Where offloaded bytes live: the CAS for the original, and the room
/// whose `REST_DIR` holds the materialized rest file. The room travels
/// as its city address because the model's read tool resolves every
/// path from the city root. `origin` is the run and building the
/// original is pinned for, so that a `cas:` read of it is judged where
/// the output was made.
pub struct OffloadSite<'a> {
    pub cas: &'a mut Cas,
    pub city_root: &'a Path,
    pub room: &'a Address,
    pub origin: storage::BlockOrigin,
}

/// The outcome: a substitute that fits the cap, the original pinned in
/// CAS, and a read-only path the model can keep reading from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffloadRecord {
    pub substitute: Vec<u8>,
    pub original: Locator,
    pub rest_path: String,
    pub original_len: u64,
}

fn hint_line(total: u64, rest_path: &str, locator: &Locator) -> String {
    format!("\n[offloaded: total {total} bytes; rest at {rest_path}; original {locator}]")
}

/// Writes the rest file under the room and answers its city address,
/// `<room>/.rest/rest-<hash>.dat`: the spelling the model's read tool
/// admits and resolves from the city root, and the same bytes in the
/// ledger on every machine.
fn materialized(
    bytes: &[u8],
    site: &OffloadSite<'_>,
    locator: &Locator,
) -> Result<String, AxError> {
    let name = rest_file_name(locator);
    let room = site.room.as_str();
    materialize(bytes, &site.city_root.join(room).join(REST_DIR), &name)?;
    Ok(format!("{room}/{REST_DIR}/{name}"))
}

fn rest_file_name(locator: &Locator) -> String {
    let mut tail: String = locator
        .to_string()
        .chars()
        .rev()
        .take(16)
        .collect::<String>()
        .chars()
        .rev()
        .collect();
    tail.retain(|c| c.is_ascii_alphanumeric());
    format!("rest-{tail}.dat")
}

fn materialize(bytes: &[u8], dir: &Path, name: &str) -> Result<(), AxError> {
    let io = |err: std::io::Error| {
        AxError::failure(
            AxCode::StorageFatal,
            "materialize rest file",
            err.to_string(),
        )
        .with_recovery(
            "free space on the disk holding the city, or fix the permissions of the \
             run's environment directory, then ask for the result again",
        )
    };
    let path = dir.join(name);
    if path.exists() {
        return Ok(());
    }
    std::fs::create_dir_all(dir).map_err(io)?;
    std::fs::write(&path, bytes).map_err(io)?;
    let mut perms = std::fs::metadata(&path).map_err(io)?.permissions();
    perms.set_readonly(true);
    std::fs::set_permissions(&path, perms).map_err(io)?;
    Ok(())
}

/// The way back from a cut: the original pinned in the CAS and
/// materialized as a read-only rest file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Tee {
    pub(crate) original: Locator,
    pub(crate) rest_path: String,
}

/// Pins the full original before anything is cut from it: invariants 1
/// and 4 in one call, so no cutter can run without the way back
/// existing first. Repeating it on the same bytes is idempotent.
pub(crate) fn tee(bytes: &[u8], site: &mut OffloadSite<'_>) -> Result<Tee, AxError> {
    let hash = site
        .cas
        .put_for(bytes, &site.origin)
        .map_err(storage::StorageError::into_ax)?;
    let original = Locator::cas(hash);
    let rest_path = materialized(bytes, site, &original)?;
    Ok(Tee {
        original,
        rest_path,
    })
}

/// Shrinks `bytes` to at most `cap_bytes`. Callers only come here for
/// lossy cases (`len > cap`); a lossless call is refused — invariant 3
/// says only lossy transforms may store.
pub fn offload(
    bytes: &[u8],
    cap_bytes: u64,
    site: &mut OffloadSite<'_>,
) -> Result<OffloadRecord, AxError> {
    let original_len = u64::try_from(bytes.len()).map_err(|_| {
        AxError::failure(AxCode::InvalidArgs, "offload result", "length exceeds u64").with_recovery(
            "ask the tool for less at a time: this result is larger than a byte \
                 count this city can hold",
        )
    })?;
    if original_len <= cap_bytes {
        return Err(AxError::failure(
            AxCode::InvalidArgs,
            "offload result",
            "lossless input: only lossy transforms store (invariant 3)",
        )
        .with_recovery(
            "pass this result through unchanged; it already fits the cap, and offload \
             stores only what it has to cut",
        ));
    }
    // Invariant 1: the full original enters the CAS before any cut.
    // Invariant 4: the substitute always points at a readable rest file.
    let Tee {
        original,
        rest_path,
    } = tee(bytes, site)?;
    let hint = hint_line(original_len, &rest_path, &original);
    let hint_len = u64::try_from(hint.len()).map_err(|_| {
        AxError::failure(AxCode::InvalidArgs, "offload result", "hint exceeds u64").with_recovery(
            "report this against runtime::offload: the hint line names one path \
                     and two byte counts and cannot reach this length",
        )
    })?;
    let head_budget = cap_bytes.checked_sub(hint_len).ok_or_else(|| {
        AxError::failure(
            AxCode::InvalidArgs,
            "offload result",
            format!("cap {cap_bytes} cannot even hold the {hint_len}-byte hint"),
        )
        .with_recovery("raise the cap or skip offload for this result")
    })?;
    let head_len = usize::try_from(head_budget.min(original_len)).map_err(|_| {
        AxError::failure(AxCode::InvalidArgs, "offload result", "head exceeds usize").with_recovery(
            "lower `[tool] result_cap_bytes`: the kept head would be longer than \
                 this machine can address",
        )
    })?;
    let mut substitute = bytes.get(..head_len).unwrap_or_default().to_vec();
    substitute.extend_from_slice(hint.as_bytes());
    // Invariant 2: substitute (hint included) within both bounds.
    let substitute_len = u64::try_from(substitute.len()).map_err(|_| {
        AxError::failure(
            AxCode::InvalidArgs,
            "offload result",
            "substitute exceeds u64",
        )
        .with_recovery(
            "lower `[tool] result_cap_bytes`: the shortened result is longer than a \
             byte count this city can hold",
        )
    })?;
    if substitute_len > cap_bytes || substitute_len > original_len {
        return Err(AxError::failure(
            AxCode::InvalidArgs,
            "offload result",
            "substitute outgrew its bounds (invariant 2)",
        )
        .with_recovery(
            "report this against runtime::offload: the shortened result must stay \
             within both the cap and the original length, and it did not",
        ));
    }
    Ok(OffloadRecord {
        substitute,
        original,
        rest_path,
        original_len,
    })
}

/// Restores the rest file from the CAS after external cleanup; the bytes
/// are the original, verbatim (A7's third assertion).
pub fn rematerialize(locator: &Locator, site: &mut OffloadSite<'_>) -> Result<String, AxError> {
    let Locator::Cas { hash, .. } = locator else {
        return Err(AxError::failure(
            AxCode::InvalidArgs,
            "rematerialize rest file",
            "locator is not cas-addressed",
        )
        .with_recovery(
            "pass the `cas:` locator the offload record carries under `original`; a \
             `file:` locator names a path that cleanup already removed",
        ));
    };
    let bytes = site.cas.get(hash).map_err(storage::StorageError::into_ax)?;
    materialized(&bytes, site, locator)
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::indexing_slicing,
    reason = "test code"
)]
pub(crate) mod tests {
    use super::*;

    pub(crate) fn origin() -> storage::BlockOrigin {
        storage::BlockOrigin {
            run: kernel::RunId::from_bytes([7; 16]),
            building: kernel::Address::parse("lab").unwrap(),
        }
    }

    fn site(dir: &tempfile::TempDir) -> (Cas, Address) {
        let cas = Cas::open(&dir.path().join("cas")).unwrap();
        (cas, Address::parse("room").unwrap())
    }

    #[test]
    fn a7_roundtrip_all_four_assertions() {
        let dir = tempfile::tempdir().unwrap();
        let (mut cas, room) = site(&dir);
        let mut s = OffloadSite {
            cas: &mut cas,
            city_root: dir.path(),
            room: &room,
            origin: origin(),
        };
        let original: Vec<u8> = (0..40_000u32)
            .map(|i| u8::try_from(i % 251).unwrap())
            .collect();
        let record = offload(&original, 4_096, &mut s).unwrap();
        // Substitute within both bounds, hint included.
        assert!(record.substitute.len() <= 4_096);
        assert!(record.substitute.len() < original.len());
        assert!(
            String::from_utf8_lossy(&record.substitute).contains("[offloaded: total 40000 bytes")
        );
        assert!(
            String::from_utf8_lossy(&record.substitute).contains("; rest at room/.rest/rest-"),
            "the hint names the rest file by its address in the room"
        );
        // The rest path serves the full original.
        assert_eq!(
            std::fs::read(dir.path().join(&record.rest_path)).unwrap(),
            original
        );
        // The CAS serves the full original by locator.
        let Locator::Cas { hash, .. } = &record.original else {
            panic!("expected a cas locator");
        };
        assert_eq!(s.cas.get(hash).unwrap(), original);
        // External cleanup, then rematerialize: bytes identical.
        let mut perms = std::fs::metadata(dir.path().join(&record.rest_path))
            .unwrap()
            .permissions();
        #[allow(clippy::permissions_set_readonly_false, reason = "test cleanup")]
        perms.set_readonly(false);
        std::fs::set_permissions(dir.path().join(&record.rest_path), perms).unwrap();
        std::fs::remove_file(dir.path().join(&record.rest_path)).unwrap();
        let back = rematerialize(&record.original, &mut s).unwrap();
        assert_eq!(std::fs::read(dir.path().join(&back)).unwrap(), original);
    }

    #[test]
    fn the_rest_address_is_one_the_read_tool_admits_and_resolves() {
        let dir = tempfile::tempdir().unwrap();
        let (mut cas, room) = site(&dir);
        let mut s = OffloadSite {
            cas: &mut cas,
            city_root: dir.path(),
            room: &room,
            origin: origin(),
        };
        let original = vec![3u8; 30_000];
        let record = offload(&original, 2_048, &mut s).unwrap();
        let admitted =
            crate::tools::admit(&record.rest_path, "read", &|_| kernel::ReadVerdict::Open).unwrap();
        assert_eq!(
            std::fs::read(dir.path().join(admitted.as_str())).unwrap(),
            original,
            "the read tool resolves the rest address from the city root"
        );
    }

    #[test]
    fn the_original_is_pinned_for_the_run_and_building_it_was_cut_for() {
        let dir = tempfile::tempdir().unwrap();
        let (mut cas, room) = site(&dir);
        let mut s = OffloadSite {
            cas: &mut cas,
            city_root: dir.path(),
            room: &room,
            origin: origin(),
        };
        let record = offload(&vec![3u8; 30_000], 2_048, &mut s).unwrap();
        let Locator::Cas { hash, .. } = &record.original else {
            panic!("expected a cas locator");
        };
        assert_eq!(s.cas.origins(hash).unwrap(), vec![origin()]);
    }

    #[test]
    fn same_bytes_offload_to_the_same_locator() {
        let dir = tempfile::tempdir().unwrap();
        let (mut cas, room) = site(&dir);
        let mut s = OffloadSite {
            cas: &mut cas,
            city_root: dir.path(),
            room: &room,
            origin: origin(),
        };
        let original = vec![7u8; 30_000];
        let one = offload(&original, 2_048, &mut s).unwrap();
        let two = offload(&original, 2_048, &mut s).unwrap();
        assert_eq!(
            one.original, two.original,
            "CAS dedup makes offload idempotent"
        );
        assert_eq!(one.rest_path, two.rest_path);
    }

    #[test]
    fn lossless_input_and_hopeless_caps_are_refused() {
        let dir = tempfile::tempdir().unwrap();
        let (mut cas, room) = site(&dir);
        let mut s = OffloadSite {
            cas: &mut cas,
            city_root: dir.path(),
            room: &room,
            origin: origin(),
        };
        let err = offload(b"small", 4_096, &mut s).unwrap_err();
        assert_eq!(*err.code(), AxCode::InvalidArgs);
        let big = vec![1u8; 20_000];
        let err = offload(&big, 16, &mut s).unwrap_err();
        assert!(err.subject().contains("hint"));
    }
}