harmos_operations/storage/
write_file.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19
20// Local Imports
21use super::IO_POLICY;
22use harmos_domain::MemorySize;
23use harmos_domain::model::storage::codec::{Nested, product};
24use harmos_domain::model::storage::{FileBatch, FileMetadata, FileReference};
25use harmos_domain::operations::ReplaySafety;
26#[cfg(feature = "runtime")]
27use harmos_sdk::operations::access::GrantAccess;
28#[cfg(feature = "runtime")]
29use harmos_storage::{ErasedCatalog, StorageManager};
30
31/* === Constants === */
32
33/* === Macros === */
34
35product!(WriteFile {
36    metadata: Nested<FileMetadata> => 1,
37});
38
39/* === Type Aliases === */
40
41/* === Structs === */
42
43// Write One Finished File From Streamed Batches.
44#[harmos::operation(
45    harmos_path = crate,
46    resources = [StorageManager<ErasedCatalog>, GrantAccess],
47    codec = manual,
48    id = "storage.write_file",
49    response = FileReference,
50    policy = IO_POLICY,
51    input = FileBatch,
52    visibility = Shared,
53    replay = ReplaySafety::Unknown,
54    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "WriteFile"),
55    grant = Self::grant_for,
56)]
57#[derive(Clone, Debug, PartialEq, Eq)]
58pub struct WriteFile {
59    pub metadata: FileMetadata,
60}
61
62/* === Enums === */
63
64/* === Traits === */
65
66/* === Implementations === */
67
68impl WriteFile {
69    fn grant_for(request: &WriteFile) -> Option<harmos_domain::plugin::Grant> {
70        Some(harmos_domain::plugin::Grant::of(<Self as harmos_domain::OperationDeclaration>::ID, request.metadata.id.as_str().to_owned()))
71    }
72}
73
74impl MemorySize for WriteFile {
75    fn memory_size(&self) -> usize {
76        std::mem::size_of::<Self>().saturating_add(self.metadata.memory_size().saturating_sub(std::mem::size_of_val(&self.metadata)))
77    }
78}
79
80#[cfg(feature = "runtime")]
81mod execution {
82    use super::*;
83
84    // Standard Library Imports
85    // Third Party Imports
86    // Local Imports
87    use crate::{Error, Result};
88    use harmos_sdk::operations::{Operation, OperationDeclaration};
89    use harmos_storage::FileReference;
90
91    impl Operation for super::WriteFile {
92        type Error = Error;
93
94        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<FileReference>> {
95            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
96            let grants = context.resource::<harmos_sdk::operations::access::GrantAccess>();
97            let input = context.streams();
98
99            let host = grants;
100            let ctx = &host.0;
101            let grant = harmos_domain::plugin::Grant::of(WriteFile::ID, self.metadata.id.as_str().to_owned());
102            ctx.admit(&grant).map_err(Error::Access)?;
103            storage.retry_transient_persistence();
104            let mut file = storage.create_file(self.metadata.clone()).await?;
105
106            let mut offset = 0u64;
107            while let Some(batch) = input.recv().await {
108                let (batch, _claims) = batch.into_owned_request()?;
109                ctx.admit(&grant).map_err(Error::Access)?;
110                if batch.offset != offset || batch.bytes.is_empty() {
111                    return Err(Error::InvalidInput {
112                        field: "file batch",
113                        reason: "nonempty batches must have contiguous byte offsets",
114                    });
115                }
116                let next = offset.checked_add(batch.bytes.len() as u64).ok_or(Error::InvalidInput {
117                    field: "file batch offset",
118                    reason: "byte offset overflow",
119                })?;
120                file.write_all(batch.bytes).await?;
121                offset = next;
122            }
123
124            ctx.admit(&grant).map_err(Error::Access)?;
125            let reference = file.finish().await?;
126            storage.await_durable(&reference.position).await?;
127            Ok(reference.into())
128        }
129
130        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
131            crate::storage::classify(error)
132        }
133    }
134}
135
136/* === Public Functions === */
137
138/* === Private Functions === */
139
140/* === Tests === */
harmos_operations/storage/write_file.rs

harmos_operations/storage/
write_file.rs
