harmos_operations/storage/
follow_production.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, Unsigned, product};
24use harmos_domain::model::storage::files::{FileBatch, FileProgress, StagedFile};
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!(FollowProduction {
36    file: Nested<StagedFile> => 1,
37    offset: Unsigned => 2,
38});
39
40/* === Type Aliases === */
41
42/* === Structs === */
43
44// Follow An Existing Staged Production Without Reinterpreting Its Identity.
45#[harmos::operation(
46    harmos_path = crate,
47    resources = [StorageManager<ErasedCatalog>, GrantAccess],
48    codec = manual,
49    id = "storage.follow_production",
50    response = FileProgress,
51    policy = IO_POLICY,
52    output = FileBatch,
53    visibility = Shared,
54    replay = ReplaySafety::Unknown,
55    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "FollowProduction"),
56    grant = Self::grant_for,
57)]
58#[derive(Clone, Debug, PartialEq, Eq)]
59pub struct FollowProduction {
60    pub file: StagedFile,
61    pub offset: u64,
62}
63
64/* === Enums === */
65
66/* === Traits === */
67
68/* === Implementations === */
69
70impl FollowProduction {
71    fn grant_for(request: &FollowProduction) -> Option<harmos_domain::plugin::Grant> {
72        Some(harmos_domain::plugin::Grant::of(<Self as harmos_domain::OperationDeclaration>::ID, request.file.id.as_str().to_owned()))
73    }
74}
75
76impl MemorySize for FollowProduction {
77    fn memory_size(&self) -> usize {
78        std::mem::size_of::<Self>()
79            .saturating_add(self.file.memory_size().saturating_sub(std::mem::size_of_val(&self.file)))
80            .saturating_add(self.offset.memory_size().saturating_sub(std::mem::size_of_val(&self.offset)))
81    }
82}
83
84#[cfg(feature = "runtime")]
85mod execution {
86    use super::*;
87
88    // Standard Library Imports
89
90    // Third Party Imports
91
92    // Local Imports
93    use crate::storage::IO_TIMEOUT;
94    use crate::{Error, Result};
95    use harmos_sdk::operations::{Operation, OperationDeclaration};
96    use harmos_storage::FileProgress;
97
98    impl Operation for super::FollowProduction {
99        type Error = Error;
100
101        #[tracing::instrument(target = "harmos::storage", name = "storage.operation.follow_production", level = "trace", skip_all)]
102        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<FileProgress>> {
103            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
104            let grants = context.resource::<harmos_sdk::operations::access::GrantAccess>();
105            let output = context.streams();
106
107            let host = grants;
108            let ctx = &host.0;
109            let grant = harmos_domain::plugin::Grant::of(FollowProduction::ID, self.file.id.as_str().to_owned());
110            ctx.admit(&grant).map_err(Error::Access)?;
111            let mut follower = storage.follow_production(&self.file, self.offset).await?;
112            let batch_bytes = storage.settings().response.batch_bytes;
113            while let Some(batch) = follower.next_batch(batch_bytes).await? {
114                ctx.admit(&grant).map_err(Error::Access)?;
115                let response = harmos_sdk::operations::Response::from_owner(batch.owner());
116                tokio::time::timeout(IO_TIMEOUT, output.send_owned(response)).await.map_err(|_| crate::storage::output_timeout())??;
117            }
118            Ok(follower.progress().into())
119        }
120
121        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
122            crate::storage::classify(error)
123        }
124    }
125}
126
127/* === Public Functions === */
128
129/* === Private Functions === */
130
131/* === Tests === */
harmos_operations/storage/follow_production.rs

harmos_operations/storage/
follow_production.rs
