canic_core/ops/fixture_importer/
mod.rs

1//! Module: ops::fixture_importer
2//!
3//! Responsibility: validate application progress and invoke synchronous importer steps.
4//! Does not own: application records, transport orchestration or scheduling.
5//! Boundary: callback errors and invalid postconditions trap before partial writes commit.
6
7#[cfg(test)]
8mod tests;
9
10use crate::{
11    dto::fixture_provisioning::{
12        FixtureAssignment, FixtureImportError, FixtureImportFailure, FixtureImportProgress,
13        FixtureProvisioningStatus, FixtureStoreError,
14    },
15    model::fixture_importer::{
16        FixtureImportAuthority, FixtureImportLease, FixtureImporterRegistry,
17        FixtureImporterRegistryError,
18    },
19    ops::{ic::IcOps, storage::fleet_activation::FleetActivationOps},
20};
21use std::cell::RefCell;
22
23thread_local! {
24    static REGISTRY: RefCell<FixtureImporterRegistry<&'static dyn FixtureImporter>> =
25        const { RefCell::new(FixtureImporterRegistry::new()) };
26}
27
28/// Synchronous application participant for bounded import and validation steps.
29///
30/// Register once from the existing synchronous lifecycle participant, after restoring
31/// the database. Mutating methods must commit rows and their checkpoint in one message.
32/// Canic traps a returned error or invalid postcondition to roll back that message.
33/// Each validation step must inspect bounded stored data; completion must not rescan it.
34pub trait FixtureImporter: Sync {
35    /// Read the durable checkpoint without changing it; `None` means not begun.
36    fn progress(
37        &self,
38        assignment: &FixtureAssignment,
39    ) -> Result<Option<FixtureImportProgress>, FixtureImportError>;
40    /// Initialize an absent checkpoint for this exact assignment without wiping existing data.
41    fn begin(&self, assignment: &FixtureAssignment) -> Result<(), FixtureImportError>;
42    /// Apply one verified chunk and advance the same durable checkpoint exactly once.
43    fn apply_chunk(
44        &self,
45        assignment: &FixtureAssignment,
46        index: u32,
47        bytes: &[u8],
48    ) -> Result<(), FixtureImportError>;
49    /// Validate a bounded portion of stored data and eventually commit the exact receipt.
50    fn validate_step(&self, assignment: &FixtureAssignment) -> Result<(), FixtureImportError>;
51}
52
53/// Scoped ownership of one fetch; IC callback cleanup also releases this exact lease.
54pub struct ImportLease(FixtureImportLease);
55
56impl ImportLease {
57    pub fn acquire(assignment: &FixtureAssignment) -> Result<Self, FixtureImportError> {
58        let binding = &assignment.grant.binding;
59        let authority = FixtureImportAuthority {
60            target: binding.target.clone(),
61            installation: binding.installation,
62            release_build_id: binding.release_build_id,
63            content_id: binding.content_id,
64        };
65        REGISTRY
66            .with_borrow_mut(|registry| registry.acquire(authority))
67            .map(Self)
68            .map_err(registry_error)
69    }
70
71    pub fn require_current(&self) -> Result<(), FixtureImportError> {
72        if !REGISTRY.with_borrow(|registry| registry.is_current(&self.0)) {
73            return Err(FixtureImportError::Authority);
74        }
75        Ok(())
76    }
77}
78
79impl Drop for ImportLease {
80    fn drop(&mut self) {
81        REGISTRY.with_borrow_mut(|registry| registry.release(&self.0));
82    }
83}
84
85/// Project the protected source without creating a second import checkpoint.
86pub fn assignment() -> Result<Option<Box<FixtureAssignment>>, FixtureImportError> {
87    FleetActivationOps::component_fixture_assignment().map_err(runtime_error)
88}
89
90/// Observe application progress and reject mismatched or prematurely completed evidence.
91pub fn progress(
92    importer: &dyn FixtureImporter,
93    assignment: &FixtureAssignment,
94) -> Result<Option<FixtureImportProgress>, FixtureImportError> {
95    let observed = importer.progress(assignment)?;
96    if let Some(progress) = &observed {
97        validate_progress(assignment, progress)?;
98    }
99    Ok(observed)
100}
101
102pub fn validate_progress(
103    assignment: &FixtureAssignment,
104    progress: &FixtureImportProgress,
105) -> Result<(), FixtureImportError> {
106    if progress.binding != assignment.grant.binding {
107        return Err(FixtureImportError::Authority);
108    }
109    if progress.next_chunk as usize > assignment.descriptor.chunks.len() {
110        return Err(FixtureImportError::Progress);
111    }
112    if let Some(receipt) = &progress.receipt {
113        if receipt.binding != assignment.grant.binding
114            || receipt.completion_summary != assignment.descriptor.completion_summary
115        {
116            return Err(FixtureImportError::Receipt);
117        }
118        if progress.next_chunk as usize != assignment.descriptor.chunks.len() {
119            return Err(FixtureImportError::Progress);
120        }
121    }
122    Ok(())
123}
124
125pub fn status(
126    assignment: &FixtureAssignment,
127    importer: &dyn FixtureImporter,
128) -> Result<FixtureProvisioningStatus, FixtureImportError> {
129    let observed = progress(importer, assignment)?;
130    Ok(match observed {
131        Some(FixtureImportProgress {
132            receipt: Some(receipt),
133            ..
134        }) => FixtureProvisioningStatus::Complete(receipt),
135        other => FixtureProvisioningStatus::Pending(other.map(Box::new)),
136    })
137}
138
139/// Begin exactly once, enforcing the callback's local postcondition before committing.
140pub fn begin(importer: &dyn FixtureImporter, assignment: &FixtureAssignment) {
141    require_callback(importer.begin(assignment));
142    require_position(importer, assignment, 0, false);
143}
144
145/// Commit exactly one chunk and require its checkpoint in the same message.
146pub fn apply_chunk(
147    importer: &dyn FixtureImporter,
148    assignment: &FixtureAssignment,
149    index: u32,
150    bytes: &[u8],
151) {
152    require_callback(importer.apply_chunk(assignment, index, bytes));
153    let next = index
154        .checked_add(1)
155        .unwrap_or_else(|| fail(FixtureImportError::Progress));
156    require_position(importer, assignment, next, false);
157}
158
159/// Validate one bounded slice; only the application's durable receipt can finish it.
160pub fn validate_step(importer: &dyn FixtureImporter, assignment: &FixtureAssignment) {
161    require_callback(importer.validate_step(assignment));
162    let next = u32::try_from(assignment.descriptor.chunks.len())
163        .unwrap_or_else(|_| fail(FixtureImportError::Progress));
164    require_position(importer, assignment, next, true);
165}
166
167fn require_position(
168    importer: &dyn FixtureImporter,
169    assignment: &FixtureAssignment,
170    next: u32,
171    allow_receipt: bool,
172) {
173    let observed = progress(importer, assignment)
174        .unwrap_or_else(|error| fail(error))
175        .unwrap_or_else(|| fail(FixtureImportError::Progress));
176    if observed.next_chunk != next || (!allow_receipt && observed.receipt.is_some()) {
177        fail(FixtureImportError::Progress);
178    }
179}
180
181fn require_callback(result: Result<(), FixtureImportError>) {
182    if let Err(error) = result {
183        fail(error);
184    }
185}
186
187fn fail(error: FixtureImportError) -> ! {
188    IcOps::trap(format!("fixture importer callback failed: {error:?}"))
189}
190
191/// Register through the sole heap model owner.
192pub fn register(importer: &'static dyn FixtureImporter) -> Result<(), FixtureImportError> {
193    REGISTRY
194        .with_borrow_mut(|registry| registry.register(importer))
195        .map_err(registry_error)
196}
197
198/// Read the one registered participant without holding a borrow across callbacks.
199pub fn registered() -> Option<&'static dyn FixtureImporter> {
200    REGISTRY.with_borrow(FixtureImporterRegistry::importer)
201}
202
203/// Require Active runtime infrastructure independently of application data readiness.
204pub fn require_active() -> Result<(), FixtureImportError> {
205    let active = FleetActivationOps::status(false).map_err(runtime_error)?;
206    if active.phase != crate::dto::fleet_activation::FleetActivationPhase::Active {
207        return Err(FixtureImportError::Authority);
208    }
209    Ok(())
210}
211
212const fn registry_error(error: FixtureImporterRegistryError) -> FixtureImportError {
213    match error {
214        FixtureImporterRegistryError::Busy => FixtureImportError::Busy,
215        FixtureImporterRegistryError::Registration => FixtureImportError::Registration,
216    }
217}
218
219fn runtime_error(
220    error: crate::ops::storage::fleet_activation::FleetActivationOpsError,
221) -> FixtureImportError {
222    FixtureImportError::Runtime(
223        crate::InternalError::from(crate::ops::storage::StorageOpsError::from(error)).into(),
224    )
225}
226
227/// Invalidate a stale heap fetch only after the durable owner proves expiry.
228pub fn abandon_expired_fetch() {
229    REGISTRY.with_borrow_mut(FixtureImporterRegistry::abandon);
230}
231
232/// Classify returned failures; callback traps remain uncertain work for recovery.
233pub fn permanent_failure(error: FixtureImportError) -> Option<FixtureImportFailure> {
234    use FixtureImportFailure as F;
235    Some(match error {
236        FixtureImportError::Busy
237        | FixtureImportError::ImporterMissing
238        | FixtureImportError::NotReady
239        | FixtureImportError::Transport(_)
240        | FixtureImportError::Source(FixtureStoreError::NotReady) => return None,
241        FixtureImportError::Application { code } => F::Application { code },
242        FixtureImportError::Authority => F::Authority,
243        FixtureImportError::Codec(error) => F::Codec {
244            code: error.raw_code(),
245        },
246        FixtureImportError::Progress => F::Progress,
247        FixtureImportError::Receipt => F::Receipt,
248        FixtureImportError::Registration => F::Registration,
249        FixtureImportError::Runtime(error) => F::Runtime {
250            code: error.raw_code(),
251        },
252        FixtureImportError::SourceRejected(error) => F::SourceRejected {
253            code: error.raw_code(),
254        },
255        FixtureImportError::Source(error) => match error {
256            FixtureStoreError::Authority => F::SourceAuthority,
257            FixtureStoreError::Bounds => F::SourceBounds,
258            FixtureStoreError::Capacity => F::SourceCapacity,
259            FixtureStoreError::Conflict => F::SourceConflict,
260            FixtureStoreError::Content => F::SourceContent,
261            FixtureStoreError::NotFound => F::SourceNotFound,
262            FixtureStoreError::Sequence => F::SourceSequence,
263            FixtureStoreError::NotReady => unreachable!(),
264        },
265    })
266}
267
268/// Observe the existing lease without changing transport or application progress.
269#[cfg(feature = "internal-test-fixtures")]
270pub fn fetch_in_flight() -> bool {
271    REGISTRY.with_borrow(FixtureImporterRegistry::fetch_in_flight)
272}
canic_core/ops/fixture_importer/mod.rs

canic_core/ops/fixture_importer/
mod.rs
