krabka_object_store/
ops.rs

1//! The shared object-store operation surface.
2//!
3//! This module holds an async, mockable [`ObjectOps`] trait and its single
4//! concrete implementation [`ObjectStoreClient`] over `object_store`. Consumers
5//! route their put, get, delete, and multipart calls through it, so the
6//! operation logic lives in one place. That logic includes the
7//! multipart-threshold branch and the `object_store::Error` to
8//! [`ObjectStoreError`] mapping.
9
10/// Precondition for a write, re-exported so callers need not depend on
11/// `object_store` directly.
12pub use object_store::PutMode;
13use {
14    bytes::Bytes,
15    futures_util::stream::BoxStream,
16    object_store::{GetRange, ObjectMeta, path::Path},
17};
18
19mod client;
20
21#[cfg(test)]
22mod tests;
23
24pub use self::client::ObjectStoreClient;
25use crate::error::ObjectStoreError;
26
27/// A listing delivered one object at a time.
28///
29/// The stream is `'static` because the backend's own listing is: it owns the
30/// paging state rather than borrowing the store, so a caller can hold the
31/// stream past the borrow it was created from.
32pub type ObjectStream = BoxStream<'static, Result<ObjectMeta, ObjectStoreError>>;
33
34/// How one put should behave.
35#[derive(Debug, Clone, Default, PartialEq, Eq)]
36pub struct PutRequest {
37    /// Precondition for the write. Defaults to `PutMode::Overwrite`.
38    pub mode: PutMode,
39    /// Whether to compute a SHA-256 digest of the payload during upload.
40    /// Defaults to `false`; only the WORM archive path needs it, and hashing
41    /// every tiered byte is not free.
42    pub digest: bool,
43}
44
45/// What a put produced.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct PutOutcome {
48    /// Bytes written.
49    pub size_bytes: u64,
50    /// Whole-object SHA-256, present only when `PutRequest::digest` was set.
51    pub sha256: Option<[u8; 32]>,
52    /// Backend entity tag, when the backend returned one.
53    pub e_tag: Option<String>,
54    /// Version id on a versioned bucket, when the backend returned one.
55    pub version_id: Option<String>,
56    /// Whether the backend applied an atomic create precondition to this put.
57    /// Multipart uploads return `false`: an atomic empty-object reservation
58    /// serializes creators, while bucket retention protects the completed
59    /// version.
60    pub create_precondition: bool,
61}
62
63/// Async object-store operations. The trait is `Send + Sync`, so tasks can
64/// share it.
65///
66/// The trait stays dyn-safe and `#[automock]`-able. It expresses multipart
67/// upload as [`ObjectOps::put_from_path`] over a filesystem path, not over a
68/// generic reader. The trait thus mocks cleanly for mutation-testable IO
69/// decision logic.
70#[krabka_macros::object_ops]
71#[cfg_attr(test, mockall::automock)]
72#[async_trait::async_trait]
73pub trait ObjectOps: Send + Sync {
74    /// Single-PUT an in-memory payload under the preconditions in `req`.
75    async fn put(
76        &self,
77        key: &Path,
78        bytes: Bytes,
79        req: PutRequest,
80    ) -> Result<PutOutcome, ObjectStoreError>;
81
82    /// Upload a local file. The method uses single-PUT below `threshold`
83    /// bytes, and streaming multipart in `chunk_size` parts at or above
84    /// `threshold`.
85    put_from_path!();
86
87    /// Fetch a whole object.
88    async fn get(&self, key: &Path) -> Result<Bytes, ObjectStoreError>;
89
90    /// Fetch a whole object with a byte limit enforced before buffering its body.
91    ///
92    /// Bind the GET to the HEAD version or etag when the backend provides one,
93    /// and reject an oversized response while streaming even without either.
94    async fn get_capped(&self, key: &Path, max_bytes: u64) -> Result<Bytes, ObjectStoreError>;
95
96    /// Fetch a byte range of an object.
97    async fn get_range(&self, key: &Path, range: GetRange) -> Result<Bytes, ObjectStoreError>;
98
99    /// Fetch object metadata, such as the size and the etag.
100    async fn head(&self, key: &Path) -> Result<ObjectMeta, ObjectStoreError>;
101
102    /// List objects under an optional prefix, collecting the whole listing.
103    ///
104    /// A prefix holding millions of keys costs that much memory here. Prefer
105    /// [`ObjectOps::list_stream`] whenever the caller can consume the listing
106    /// as it arrives.
107    async fn list(&self, prefix: Option<Path>) -> Result<Vec<ObjectMeta>, ObjectStoreError>;
108
109    /// List objects under an optional prefix as a stream, one object at a
110    /// time, so a caller never holds the whole listing at once.
111    fn list_stream(&self, prefix: Option<Path>) -> ObjectStream;
112
113    /// Delete an object.
114    async fn delete(&self, key: &Path) -> Result<(), ObjectStoreError>;
115}
krabka_object_store/ops.rs

krabka_object_store/
ops.rs
