canic_host/component_operation/ops/
mod.rs

1//! Component operation persistence and transport boundary.
2//! Durable publication uses the shared host I/O owner.
3
4mod progress;
5pub mod transport;
6
7pub use progress::project_progress;
8
9use crate::{
10    component_operation::{
11        ComponentOperationError,
12        model::{
13            ComponentAuthorityRecord, ComponentOperationRecord, ComponentPlanRecord,
14            ComponentProgressRecord,
15        },
16        policy::validate_label,
17        view::{ComponentObservation, ComponentProgressObservation},
18    },
19    durable_io::{read_regular_bytes, write_bytes},
20};
21use canic_core::cdk::utils::hash::sha256_hex;
22use std::path::{Path, PathBuf};
23
24const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024;
25
26/// Transport performs individual reads/commands; workflow owns ordering and retry.
27pub trait ComponentTransport {
28    /// Read current release, placement, controller and available-pool authority.
29    fn observe(
30        &mut self,
31        plan: &ComponentPlanRecord,
32    ) -> Result<ComponentObservation, ComponentOperationError>;
33    /// Read this exact operation; only typed absence permits submission.
34    fn progress(
35        &mut self,
36        plan: &ComponentPlanRecord,
37    ) -> Result<ComponentProgressObservation, ComponentOperationError>;
38    /// Submit the retained operation identity once.
39    fn submit(&mut self, plan: &ComponentPlanRecord) -> Result<(), ComponentOperationError>;
40}
41
42/// Select one confined same-Fleet local operation document.
43pub fn record_path(
44    root: &Path,
45    environment: &str,
46    fleet: &str,
47    name: &str,
48) -> Result<PathBuf, ComponentOperationError> {
49    for value in [environment, fleet, name] {
50        validate_label(value)?;
51    }
52    Ok(root
53        .join(".canic/component-operations")
54        .join(environment)
55        .join(fleet)
56        .join(format!("{name}.json")))
57}
58
59/// Hash immutable review data, excluding its self-referential digest.
60pub fn review_digest(plan: &ComponentPlanRecord) -> Result<String, ComponentOperationError> {
61    let mut canonical = plan.clone();
62    canonical.review_sha256.clear();
63    Ok(sha256_hex(&serde_json::to_vec(&canonical)?))
64}
65
66/// Create a fresh operation identity only before its first durable publication.
67pub fn new_record(
68    name: &str,
69    authority: ComponentAuthorityRecord,
70) -> Result<ComponentOperationRecord, ComponentOperationError> {
71    let mut operation_id = [0; 32];
72    getrandom::fill(&mut operation_id).map_err(std::io::Error::other)?;
73    let mut plan = ComponentPlanRecord {
74        schema_version: 1,
75        name: name.to_string(),
76        operation_id,
77        authority,
78        review_sha256: String::new(),
79    };
80    plan.review_sha256 = review_digest(&plan)?;
81    let mut record = ComponentOperationRecord {
82        document_sha256: String::new(),
83        plan,
84        submission_attempts: 0,
85        progress: None,
86    };
87    seal(&mut record)?;
88    Ok(record)
89}
90
91fn document_digest(record: &ComponentOperationRecord) -> Result<String, ComponentOperationError> {
92    let mut canonical = record.clone();
93    canonical.document_sha256.clear();
94    Ok(sha256_hex(&serde_json::to_vec(&canonical)?))
95}
96
97fn seal(record: &mut ComponentOperationRecord) -> Result<(), ComponentOperationError> {
98    record.document_sha256 = document_digest(record)?;
99    Ok(())
100}
101
102/// Read and verify one bounded current record without following a symlink.
103pub fn read(path: &Path) -> Result<Option<ComponentOperationRecord>, ComponentOperationError> {
104    let bytes = match read_regular_bytes(path, MAX_RECORD_BYTES) {
105        Ok(bytes) => bytes,
106        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
107        Err(error) => return Err(error.into()),
108    };
109    let record: ComponentOperationRecord = serde_json::from_slice(&bytes)?;
110    if record.plan.schema_version != 1
111        || record.plan.operation_id == [0; 32]
112        || review_digest(&record.plan)? != record.plan.review_sha256
113        || document_digest(&record)? != record.document_sha256
114    {
115        return Err(ComponentOperationError::Integrity);
116    }
117    if let Some(progress) = &record.progress {
118        if record.submission_attempts == 0 {
119            return Err(ComponentOperationError::Integrity);
120        }
121        crate::component_operation::policy::validate_progress(
122            &record.plan.authority,
123            None,
124            progress,
125        )?;
126    }
127    Ok(Some(record))
128}
129
130/// Atomically retain the complete document before any subsequent effect.
131pub fn write(
132    path: &Path,
133    record: &ComponentOperationRecord,
134) -> Result<(), ComponentOperationError> {
135    let bytes = serde_json::to_vec_pretty(record)?;
136    if bytes.len() > MAX_RECORD_BYTES {
137        return Err(ComponentOperationError::Integrity);
138    }
139    Ok(write_bytes(path, &bytes)?)
140}
141
142/// Persist one submission reservation, including attempts whose response is lost.
143pub fn reserve(
144    path: &Path,
145    record: &mut ComponentOperationRecord,
146) -> Result<(), ComponentOperationError> {
147    record.submission_attempts = record
148        .submission_attempts
149        .checked_add(1)
150        .ok_or(ComponentOperationError::Integrity)?;
151    seal(record)?;
152    write(path, record)
153}
154
155/// Retain correlated monotonic Root progress in the same operation document.
156pub fn retain_progress(
157    path: &Path,
158    record: &mut ComponentOperationRecord,
159    progress: ComponentProgressRecord,
160) -> Result<(), ComponentOperationError> {
161    record.progress = Some(progress);
162    seal(record)?;
163    write(path, record)
164}
canic_host/component_operation/ops/mod.rs

canic_host/component_operation/ops/
mod.rs
