urma_runtime/
disk_plan.rs

1use crate::error::{Context, Error, ensure};
2use crate::publication::PublicPlan;
3use crate::storage;
4use crate::{disk_order::Order, disk_writer::DiskWriter};
5use crate::{
6    node::Node,
7    plan::{PlanLimits, PublicationPlan},
8    planner::Planner,
9    validation::Validation,
10};
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13use std::{
14    fs::File,
15    io::{Read, Seek, SeekFrom},
16    path::{Path, PathBuf},
17};
18use urma_chain::observation::Chain;
19use urma_core::multipart::Geometry;
20use urma_identity::identity::IdentitySigner;
21
22#[derive(Clone, Serialize, Deserialize)]
23#[serde(deny_unknown_fields)]
24pub struct DiskPlan {
25    pub version: u8,
26    pub chain: Chain,
27    pub author: String,
28    pub root_txid: String,
29    pub total_fee: u64,
30    pub maximum_fee: u64,
31    pub record_count: u32,
32    pub records_hash: String,
33    pub index_hash: String,
34    #[serde(skip)]
35    directory: PathBuf,
36}
37
38impl DiskPlan {
39    pub const MAX_RECORDS: u32 = Geometry::MAX_NODES;
40    pub(crate) const MAX_PAIR_BYTES: usize = 2 * 1024 * 1024;
41
42    pub fn prepare_multipart(
43        node: &Node,
44        signer: &impl IdentitySigner,
45        reader: &mut impl Read,
46        length: u64,
47        profile: [u8; 8],
48        limits: PlanLimits,
49        directory: &Path,
50    ) -> Result<Self, Error> {
51        let geometry = urma_core::multipart::Geometry::new(length)?;
52        let planner = Planner::streaming(node, signer, limits, geometry.nodes())?;
53        let writer = DiskWriter::new(directory, planner)?;
54        writer.prepare(reader, geometry, length, profile)
55    }
56
57    pub(crate) fn from_metadata(
58        metadata: &PublicationPlan,
59        count: u32,
60        records_hash: String,
61        index_hash: String,
62        directory: &Path,
63    ) -> Self {
64        Self {
65            version: 1,
66            chain: metadata.chain,
67            author: metadata.author.clone(),
68            root_txid: metadata.root_txid.clone(),
69            total_fee: metadata.total_fee,
70            maximum_fee: metadata.maximum_fee,
71            record_count: count,
72            records_hash,
73            index_hash,
74            directory: directory.to_owned(),
75        }
76    }
77
78    pub fn directory(&self) -> &Path {
79        &self.directory
80    }
81
82    pub fn load(directory: &Path) -> Result<Self, Error> {
83        let mut plan: Self =
84            serde_json::from_slice(&storage::read_bounded(&directory.join("plan.json"), 16384)?)?;
85        plan.directory = directory.to_owned();
86        plan.validate()?;
87        Ok(plan)
88    }
89
90    pub fn id(&self) -> Result<String, Error> {
91        Ok(hex::encode(Sha256::digest(serde_json::to_vec(self)?)))
92    }
93
94    pub fn record(&self, index: u32) -> Result<PublicPlan, Error> {
95        ensure!(index < self.record_count, "record index exceeds plan");
96        let mut offsets = File::open(self.directory.join("index.bin"))?;
97        offsets.seek(SeekFrom::Start(u64::from(index) * 12))?;
98        let (offset, length) = read_index(&mut offsets)?;
99        let mut records = File::open(self.directory.join("records.bin"))?;
100        records.seek(SeekFrom::Start(offset))?;
101        read_pair(&mut records, length)
102    }
103
104    pub fn validate(&self) -> Result<(), Error> {
105        ensure!(
106            self.version == 1 && self.record_count > 0 && self.record_count <= Self::MAX_RECORDS,
107            "invalid disk plan version or record count"
108        );
109        ensure!(
110            self.maximum_fee > 0 && self.total_fee <= self.maximum_fee,
111            "plan fee budget exceeded"
112        );
113        let mut index = File::open(self.directory.join("index.bin"))?;
114        let mut records = File::open(self.directory.join("records.bin"))?;
115        ensure!(
116            records.metadata()?.len()
117                <= u64::from(self.record_count) * u64::try_from(Self::MAX_PAIR_BYTES)?,
118            "disk plan exceeds bounded record capacity"
119        );
120        ensure!(
121            index.metadata()?.len() == u64::from(self.record_count) * 12,
122            "invalid disk plan index length"
123        );
124        ensure!(
125            digest(&mut index)? == self.index_hash && digest(&mut records)? == self.records_hash,
126            "disk plan digest mismatch"
127        );
128        let mut offset = 0u64;
129        let mut state = Validation::new();
130        let mut order = Order::new();
131        for _ in 0..self.record_count {
132            let (actual, length) = read_index(&mut index)?;
133            ensure!(actual == offset, "noncanonical disk plan offset");
134            let bytes =
135                state.append(&read_pair(&mut records, length)?, self.chain, &self.author)?;
136            order.append(&bytes, state.last_txid()?)?;
137            offset = offset
138                .checked_add(u64::from(length))
139                .context("disk plan offset overflow")?;
140        }
141        ensure!(
142            offset == records.metadata()?.len(),
143            "trailing disk plan bytes"
144        );
145        order.finish(self.record_count)?;
146        state.finish(self.total_fee, &self.root_txid)
147    }
148}
149
150fn read_index(reader: &mut impl Read) -> Result<(u64, u32), Error> {
151    let mut offset = [0u8; 8];
152    let mut length = [0u8; 4];
153    reader.read_exact(&mut offset)?;
154    reader.read_exact(&mut length)?;
155    Ok((u64::from_le_bytes(offset), u32::from_le_bytes(length)))
156}
157
158fn read_pair(reader: &mut impl Read, length: u32) -> Result<PublicPlan, Error> {
159    ensure!(
160        length > 0 && usize::try_from(length)? <= DiskPlan::MAX_PAIR_BYTES,
161        "disk plan record exceeds capacity"
162    );
163    let mut bytes = vec![0; usize::try_from(length)?];
164    reader.read_exact(&mut bytes)?;
165    Ok(serde_json::from_slice(&bytes)?)
166}
167
168fn digest(file: &mut File) -> Result<String, Error> {
169    let hash = urma_io::digest(file)?;
170    file.rewind()?;
171    Ok(hex::encode(hash))
172}
urma_runtime/disk_plan.rs

urma_runtime/
disk_plan.rs
