harmos_operations/storage/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// The Closed Catalog Of Storage Operation Declarations. The Runtime Binds
11// Each Declaration To Its Storage Handler.
12
13/* === Modules === */
14
15mod abandon_file;
16mod append_file;
17mod capture_file;
18mod create_file;
19mod delete_file;
20#[cfg(feature = "runtime")]
21pub(crate) mod exchange;
22mod finish_file;
23mod follow_file;
24mod follow_objects;
25mod follow_production;
26mod follow_records;
27mod follow_stream;
28mod hash_file;
29mod list_directory;
30mod query;
31mod read_file;
32mod resume_file;
33mod storage_status;
34mod watch_directory;
35mod write_file;
36mod write_objects;
37mod write_records;
38mod write_schema;
39mod write_streams;
40
41/* === Re-Exports === */
42
43pub use abandon_file::AbandonFile;
44pub use append_file::AppendFile;
45pub use capture_file::CaptureFile;
46pub use create_file::CreateFile;
47pub use delete_file::DeleteFile;
48pub use finish_file::FinishFile;
49pub use follow_file::FollowFile;
50pub use follow_objects::FollowObjects;
51pub use follow_production::FollowProduction;
52pub use follow_records::{FollowBoundaries, FollowRecords, Followed};
53pub use follow_stream::FollowStream;
54pub use hash_file::HashFile;
55pub use list_directory::ListDirectory;
56pub use query::Query;
57pub use read_file::ReadFile;
58pub use resume_file::ResumeFile;
59pub use storage_status::StorageStatus;
60pub use watch_directory::WatchDirectory;
61pub use write_file::WriteFile;
62pub use write_objects::WriteObjects;
63pub use write_records::WriteRecords;
64pub use write_schema::WriteSchema;
65pub use write_streams::WriteStreams;
66
67/* === Imports === */
68
69// Standard Library Imports
70use std::time::Duration;
71// Third Party Imports
72
73// Local Imports
74
75use harmos_domain::operations::{Backoff, Jitter, OperationPolicy, RetryPolicy};
76
77/* === Constants === */
78
79pub const POLICY: OperationPolicy = OperationPolicy {
80    timeout: Duration::from_secs(60),
81    attempt_timeout: Duration::from_secs(15),
82    retries: RetryPolicy::OnTransient {
83        max_retries: 3,
84        backoff: Backoff::Exponential {
85            initial: Duration::from_millis(250),
86            maximum: Duration::from_secs(2),
87            jitter: Jitter::Deterministic,
88        },
89    },
90};
91
92pub const EVERY_SCHEMA: &str = "*";
93
94pub const STREAM_POLICY: OperationPolicy = OperationPolicy {
95    attempt_timeout: POLICY.timeout,
96    retries: RetryPolicy::Never,
97    ..POLICY
98};
99
100pub const IO_POLICY: OperationPolicy = OperationPolicy {
101    timeout: harmos_domain::operations::UNBOUNDED,
102    attempt_timeout: harmos_domain::operations::UNBOUNDED,
103    retries: RetryPolicy::Never,
104};
105
106/* === Macros === */
107
108#[harmos::catalog(harmos_path = crate)]
109pub enum Storage {
110    ListDirectory(ListDirectory),
111    WatchDirectory(WatchDirectory),
112    Query(Query),
113    ReadFile(ReadFile),
114    HashFile(HashFile),
115    CaptureFile(CaptureFile),
116    FollowProduction(FollowProduction),
117    FollowRecords(FollowRecords),
118    FollowStream(FollowStream),
119    FollowFile(FollowFile),
120    StorageStatus(StorageStatus),
121    ResumeFile(ResumeFile),
122    CreateFile(CreateFile),
123    AppendFile(AppendFile),
124    FinishFile(FinishFile),
125    AbandonFile(AbandonFile),
126    WriteRecords(WriteRecords),
127    WriteStreams(WriteStreams),
128    WriteFile(WriteFile),
129    DeleteFile(DeleteFile),
130    WriteSchema(WriteSchema),
131    WriteObjects(WriteObjects),
132    FollowObjects(FollowObjects),
133}
134
135/* === Type Aliases === */
136
137/* === Structs === */
138
139/* === Enums === */
140
141/* === Traits === */
142
143/* === Implementations === */
144
145/* === Public Functions === */
146
147/* === Private Functions === */
148
149/* === Tests === */
150
151#[cfg(feature = "runtime")]
152const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
153#[cfg(feature = "runtime")]
154fn output_timeout() -> crate::Error {
155    harmos_storage::Error::backend(
156        "storage output",
157        harmos_storage::BackendFailure::Transient,
158        std::io::Error::new(std::io::ErrorKind::TimedOut, "consumer delivery deadline exceeded"),
159    )
160    .into()
161}
162
163#[cfg(feature = "runtime")]
164fn classify(error: &crate::Error) -> harmos_domain::operations::FailureKind {
165    match error {
166        crate::Error::Storage(source) if source.backend_failure() == Some(harmos_storage::BackendFailure::Transient) => {
167            harmos_domain::operations::FailureKind::Transient
168        }
169        _ => harmos_domain::operations::FailureKind::Permanent,
170    }
171}
172
173#[cfg(all(test, feature = "runtime"))]
174mod tests {
175    use super::classify;
176    use harmos_domain::operations::FailureKind;
177    use harmos_storage::{BackendFailure, Error as StorageError};
178
179    #[test]
180    fn persistence_preserves_failure_kind() {
181        for (backend, expected) in
182            [(BackendFailure::Transient, FailureKind::Transient), (BackendFailure::Permanent, FailureKind::Permanent)]
183        {
184            let source = StorageError::backend("write", backend, std::io::Error::other("backend failed"));
185            let error = StorageError::Persistence { source: std::sync::Arc::new(source) };
186            assert_eq!(classify(&crate::Error::Storage(error)), expected);
187        }
188    }
189}
harmos_operations/storage/mod.rs

harmos_operations/storage/
mod.rs
