harmos_operations/storage/
query.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// A Declarative Storage Query: A Source At A Full Cursor, Ordered Transforms
11// And A Final Row Slice. Results Cross The Wire As Arrow IPC.
12
13/* === Modules === */
14
15/* === Re-Exports === */
16
17/* === Imports === */
18
19// Standard Library Imports
20
21// Third Party Imports
22
23// Local Imports
24use super::IO_POLICY;
25use harmos_domain::MemorySize;
26use harmos_domain::model::storage::StorageCursor;
27use harmos_domain::model::storage::codec::{Nested, Repeated, product};
28use harmos_domain::model::storage::query::transform::Transform;
29use harmos_domain::model::storage::query::{QueryCompletion, QueryOutput, RowSlice, Source};
30use harmos_domain::operations::ReplaySafety;
31#[cfg(feature = "runtime")]
32use harmos_sdk::operations::access::GrantAccess;
33#[cfg(feature = "runtime")]
34use harmos_storage::{ErasedCatalog, StorageManager};
35
36/* === Constants === */
37
38/* === Macros === */
39
40product!(Query {
41    at: Nested<StorageCursor> => 1,
42    source: Nested<Source> => 2,
43    transforms: Repeated<Transform> => 3,
44    slice: Nested<RowSlice> => 4,
45});
46
47/* === Type Aliases === */
48
49/* === Structs === */
50
51#[harmos::operation(
52    harmos_path = crate,
53    resources = [StorageManager<ErasedCatalog>, GrantAccess],
54    codec = manual,
55    id = "storage.query",
56    response = QueryCompletion,
57    policy = IO_POLICY,
58    output = QueryOutput,
59    visibility = Shared,
60    replay = ReplaySafety::Idempotent,
61    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "Query"),
62    grant = Self::grant_for,
63)]
64#[derive(Clone, Debug)]
65pub struct Query {
66    pub at: StorageCursor,
67    pub source: Source,
68    pub transforms: Vec<Transform>,
69    pub slice: RowSlice,
70}
71
72/* === Enums === */
73
74/* === Traits === */
75
76/* === Implementations === */
77
78impl Query {
79    fn grant_for(request: &Query) -> Option<harmos_domain::plugin::Grant> {
80        Some(harmos_domain::plugin::Grant::of(<Self as harmos_domain::OperationDeclaration>::ID, request.source.grant_scope()))
81    }
82}
83
84impl MemorySize for Query {
85    fn memory_size(&self) -> usize {
86        std::mem::size_of::<Self>()
87            .saturating_add(self.at.memory_size().saturating_sub(std::mem::size_of_val(&self.at)))
88            .saturating_add(self.source.memory_size().saturating_sub(std::mem::size_of_val(&self.source)))
89            .saturating_add(self.transforms.memory_size().saturating_sub(std::mem::size_of_val(&self.transforms)))
90            .saturating_add(self.slice.memory_size().saturating_sub(std::mem::size_of_val(&self.slice)))
91    }
92}
93
94#[cfg(feature = "runtime")]
95mod execution {
96    use super::*;
97
98    use crate::storage::IO_TIMEOUT;
99    use crate::{Error, Result};
100    use harmos_domain::plugin::Grant;
101    use harmos_sdk::operations::{Operation, OperationDeclaration};
102    use harmos_storage::query::QueryCompletion;
103
104    // Standard Library Imports
105
106    // Third Party Imports
107
108    // Local Imports
109
110    impl Operation for super::Query {
111        type Error = Error;
112
113        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<QueryCompletion>> {
114            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
115            let grants = context.resource::<harmos_sdk::operations::access::GrantAccess>();
116            let output = context.streams();
117
118            let host = grants;
119            let ctx = &host.0;
120            let grant = Grant::of(Query::ID, self.source.grant_scope());
121            ctx.admit(&grant).map_err(Error::Access)?;
122            let mut reader = storage
123                .query(harmos_storage::Query {
124                    at: self.at.clone(),
125                    source: self.source.clone(),
126                    transforms: self.transforms.clone(),
127                    slice: self.slice,
128                })
129                .await?;
130            while let Some(item) = reader.next().await? {
131                ctx.admit(&grant).map_err(Error::Access)?;
132                let bytes = item.output_memory_size();
133                tokio::time::timeout(IO_TIMEOUT, output.send_with(bytes, || Ok(item.output())))
134                    .await
135                    .map_err(|_| crate::storage::output_timeout())??;
136            }
137            Ok(reader.completion()?.into())
138        }
139        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
140            crate::storage::classify(error)
141        }
142    }
143}
144
145/* === Public Functions === */
146
147/* === Private Functions === */
148
149/* === Tests === */
harmos_operations/storage/query.rs

harmos_operations/storage/
query.rs
