harmos_operations/storage/
follow_production.rs

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

harmos_operations/storage/
follow_production.rs
