storage/cas/
origin.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! Whose a block is: the run and the building it was put for, recorded
7//! when it is put (`crates/storage/spec/Cas.lean` §8-3).
8//!
9//! The same bytes may be put for two buildings, so a block keeps every
10//! origin it was put for, one line each, in `<dir>/from/<shard>/<hex>`.
11//! A line that does not read back - the tail of an append a crash cut -
12//! grants nothing, which is the side a read bound fails towards.
13
14use std::path::PathBuf;
15
16use kernel::{Address, B3Hash, RunId};
17
18use super::Cas;
19use crate::error::{StorageError, io_err};
20
21/// The run and the building a block was put for.
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct BlockOrigin {
24    pub run: RunId,
25    pub building: Address,
26}
27
28impl Cas {
29    /// Puts the bytes and records the origin they were put for. `Ok`
30    /// means both the object and the record are durable; a second put
31    /// with the same origin adds nothing.
32    ///
33    /// # Errors
34    /// `StorageError::Io` naming the path the store would not take.
35    pub fn put_for(&mut self, bytes: &[u8], origin: &BlockOrigin) -> Result<B3Hash, StorageError> {
36        let hash = self.put(bytes)?;
37        let line = format!("{} {}", origin.run, origin.building.as_str());
38        let (shard_dir, path) = self.origin_path(&hash);
39        let mut record = format!("{line}\n");
40        if self.vfs.exists(&path) {
41            let held = self
42                .vfs
43                .read(&path)
44                .map_err(io_err("read cas origins", &path))?;
45            if held
46                .split(|byte| *byte == b'\n')
47                .any(|recorded| recorded == line.as_bytes())
48            {
49                return Ok(hash);
50            }
51            // A crash can cut an append short; the new record starts on
52            // a line of its own, or it would read back as part of that tail.
53            if held.last().is_some_and(|byte| *byte != b'\n') {
54                record.insert(0, '\n');
55            }
56        }
57        self.vfs
58            .create_dir_all(&shard_dir)
59            .map_err(io_err("create cas origin shard", &shard_dir))?;
60        self.vfs
61            .append(&path, record.as_bytes())
62            .map_err(io_err("record cas origin", &path))?;
63        self.vfs
64            .sync_data(&path)
65            .map_err(io_err("sync cas origin", &path))?;
66        self.vfs
67            .sync_dir(&shard_dir)
68            .map_err(io_err("sync cas origin shard", &shard_dir))?;
69        Ok(hash)
70    }
71
72    /// Every origin recorded for a block, oldest first; empty for a
73    /// block put with no origin and for a hash the store never saw.
74    ///
75    /// # Errors
76    /// `StorageError::Io` when a record exists and cannot be read.
77    pub fn origins(&self, hash: &B3Hash) -> Result<Vec<BlockOrigin>, StorageError> {
78        let (_, path) = self.origin_path(hash);
79        if !self.vfs.exists(&path) {
80            return Ok(Vec::new());
81        }
82        let held = self
83            .vfs
84            .read(&path)
85            .map_err(io_err("read cas origins", &path))?;
86        Ok(held
87            .split(|byte| *byte == b'\n')
88            .filter_map(|line| {
89                let (run, building) = std::str::from_utf8(line).ok()?.split_once(' ')?;
90                Some(BlockOrigin {
91                    run: RunId::parse(run).ok()?,
92                    building: Address::parse(building).ok()?,
93                })
94            })
95            .collect())
96    }
97
98    fn origin_path(&self, hash: &B3Hash) -> (PathBuf, PathBuf) {
99        let hex = hash.to_string();
100        let shard_dir = self.dir.join("from").join(hex.get(..2).unwrap_or("00"));
101        let path = shard_dir.join(&hex);
102        (shard_dir, path)
103    }
104}
105
106#[cfg(test)]
107#[allow(clippy::unwrap_used, reason = "test code")]
108mod tests {
109    use std::io::Write;
110
111    use super::*;
112
113    #[test]
114    fn an_origin_appended_after_a_torn_tail_still_reads_back() {
115        let dir = tempfile::tempdir().unwrap();
116        let mut cas = Cas::open(dir.path()).unwrap();
117        let put_in = |building: &str| BlockOrigin {
118            run: RunId::from_bytes([7; 16]),
119            building: Address::parse(building).unwrap(),
120        };
121        let hash = cas.put_for(b"notes\n", &put_in("lab")).unwrap();
122        let (_, path) = cas.origin_path(&hash);
123        std::fs::OpenOptions::new()
124            .append(true)
125            .open(&path)
126            .unwrap()
127            .write_all(b"0000 torn")
128            .unwrap();
129
130        cas.put_for(b"notes\n", &put_in("vault")).unwrap();
131
132        assert_eq!(
133            cas.origins(&hash).unwrap(),
134            vec![put_in("lab"), put_in("vault")]
135        );
136    }
137}
storage/cas/origin.rs

storage/cas/
origin.rs
