krabka_remote_storage/s3/
client.rs

1//! Construction and configuration of an [`S3RemoteStorage`], and the bridge
2//! that drives the backend's async object-store calls from a synchronous
3//! trait method.
4//!
5//! A backend comes either from an arbitrary `ObjectStore` handle through
6//! [`S3RemoteStorage::with_store`] or from an [`S3Config`] through
7//! [`S3RemoteStorage::from_s3_config`]. The chained builders
8//! [`S3RemoteStorage::with_multipart_tuning`] and
9//! [`S3RemoteStorage::with_worm`] then set the upload tuning and the archive
10//! posture.
11
12use {
13    super::{S3RemoteStorage, WormBucket, WormMode, size_from_usize},
14    crate::{
15        error::RemoteStorageError,
16        worm::{WormArchiver, WormConfig},
17    },
18    krabka_object_store::{
19        DEFAULT_MULTIPART_CHUNK_SIZE, DEFAULT_MULTIPART_THRESHOLD, ObjectStoreClient,
20        ObjectStoreConfig, ObjectStoreError, S3Config, build_object_store, verify_gcs_worm_bucket,
21        verify_s3_worm_bucket,
22    },
23    krabka_units::prelude::{ByteSize, ByteSizeExt as _},
24    object_store::ObjectStore,
25    std::{future::Future, sync::Arc, thread},
26    tokio::{runtime, runtime::Handle, task},
27};
28
29impl S3RemoteStorage {
30    /// Largest index read by default: Kafka's Java `ByteBuffer` capacity
31    /// fits in a signed `int`, so an index larger than this cannot be loaded
32    /// by Kafka. The cap applies before buffering and during streaming.
33    pub const DEFAULT_MAX_INDEX_BYTES: u64 = i32::MAX as u64;
34
35    /// Set a lower memory limit for whole index reads, including snapshots
36    /// and epoch checkpoints. Values above [`Self::DEFAULT_MAX_INDEX_BYTES`]
37    /// use that ceiling. An object above the limit returns an error; it is
38    /// never truncated. Zero permits only empty objects.
39    #[must_use]
40    pub fn with_max_index_bytes(mut self, max_bytes: u64) -> Self {
41        self.max_index_bytes = max_bytes.min(Self::DEFAULT_MAX_INDEX_BYTES);
42        self
43    }
44
45    /// Wraps an arbitrary `ObjectStore`, for example
46    /// `object_store::memory::InMemory` for tests. Use
47    /// [`Self::from_s3_config`] for the production S3 path. Multipart
48    /// tuning falls back to the [`DEFAULT_MULTIPART_THRESHOLD`] and
49    /// [`DEFAULT_MULTIPART_CHUNK_SIZE`] constants. Call
50    /// [`Self::with_multipart_tuning`] to override them in tests.
51    #[must_use]
52    pub fn with_store(store: Arc<dyn ObjectStore>, prefix: Option<String>) -> Self {
53        Self {
54            ops: ObjectStoreClient::new(store),
55            prefix,
56            multipart_threshold: ByteSize::from_bytes(DEFAULT_MULTIPART_THRESHOLD),
57            multipart_chunk_size: size_from_usize(DEFAULT_MULTIPART_CHUNK_SIZE),
58            max_index_bytes: Self::DEFAULT_MAX_INDEX_BYTES,
59            worm: None,
60            worm_bucket: WormBucket::Unverified,
61        }
62    }
63
64    /// Puts this backend into WORM archive mode.
65    ///
66    /// Every copy then seals a signed, chained `.manifest` beside the segment,
67    /// every delete is refused, and a `write_only` archive refuses remote
68    /// fetches as well.
69    ///
70    /// # Errors
71    ///
72    /// Returns an error when the bucket policy cannot be confirmed or `cfg`'s
73    /// signing key cannot be loaded.
74    pub fn with_worm(self, cfg: &WormConfig) -> Result<Self, RemoteStorageError> {
75        match &self.worm_bucket {
76            WormBucket::S3(s3) => {
77                if !s3.conditional_put {
78                    return Err(RemoteStorageError::InvalidArgument(
79                        "WORM mode requires remote_storage.s3.conditional_put = true".into(),
80                    ));
81                }
82            }
83            WormBucket::Gcs(_) => {}
84            WormBucket::Unverified => {
85                return Err(RemoteStorageError::InvalidArgument(
86                    "WORM mode requires an object-store configuration so bucket versioning and \
87                     retention can be confirmed"
88                        .into(),
89                ));
90            }
91        }
92        verify_worm_bucket(&self.worm_bucket)?;
93        self.enable_worm(cfg, true)
94    }
95
96    fn enable_worm(
97        mut self,
98        cfg: &WormConfig,
99        require_version_id: bool,
100    ) -> Result<Self, RemoteStorageError> {
101        self.worm = Some(WormMode {
102            archiver: WormArchiver::from_config(cfg)?,
103            write_only: cfg.write_only,
104            require_version_id,
105        });
106        Ok(self)
107    }
108
109    #[cfg(test)]
110    pub(super) fn with_worm_unchecked(self, cfg: &WormConfig) -> Result<Self, RemoteStorageError> {
111        self.enable_worm(cfg, false)
112    }
113
114    /// Overrides the multipart threshold and chunk size. Returns `self` for
115    /// chained calls. Tests use this to force the multipart path on small
116    /// fixtures. Production usually keeps the defaults.
117    #[must_use]
118    pub fn with_multipart_tuning(mut self, threshold: ByteSize, chunk_size: ByteSize) -> Self {
119        self.multipart_threshold = threshold;
120        self.multipart_chunk_size = chunk_size;
121        self
122    }
123
124    pub(crate) fn from_backend_config(
125        config: &ObjectStoreConfig,
126        prefix: Option<String>,
127        threshold: u64,
128        chunk_size: usize,
129        worm_bucket: WormBucket,
130    ) -> Result<Self, RemoteStorageError> {
131        let store = build_object_store(config)
132            .map_err(|error| RemoteStorageError::InvalidArgument(error.to_string()))?;
133        let mut storage = Self::with_store(store, prefix)
134            .with_multipart_tuning(ByteSize::from_bytes(threshold), size_from_usize(chunk_size));
135        storage.worm_bucket = worm_bucket;
136        Ok(storage)
137    }
138
139    /// Builds an `AmazonS3` client from `cfg` and wraps it.
140    ///
141    /// # Errors
142    ///
143    /// Returns [`RemoteStorageError::InvalidArgument`] if `object_store`'s
144    /// builder rejects the bucket, region, and endpoint combination.
145    pub fn from_s3_config(cfg: &S3Config) -> Result<Self, RemoteStorageError> {
146        Self::from_backend_config(
147            &ObjectStoreConfig::S3(cfg.clone()),
148            cfg.prefix.clone(),
149            cfg.multipart_threshold,
150            cfg.multipart_chunk_size,
151            WormBucket::S3(cfg.clone()),
152        )
153    }
154
155    /// Runs an async [`ObjectOps`](krabka_object_store::ObjectOps) call to
156    /// completion on the current Tokio runtime. Sync trait callers reach this
157    /// through `spawn_blocking`, where `Handle::current()` is always
158    /// available. The `block_on` bridge lives here, never in the substrate.
159    pub(super) fn block_os<T, F>(fut: F) -> Result<T, ObjectStoreError>
160    where
161        F: Future<Output = Result<T, ObjectStoreError>>,
162    {
163        let handle = Handle::try_current().map_err(|_| {
164            ObjectStoreError::Backend(
165                "S3RemoteStorage requires an active Tokio runtime; call from spawn_blocking".into(),
166            )
167        })?;
168        task::block_in_place(|| handle.block_on(fut))
169    }
170}
171
172fn verify_worm_bucket(bucket: &WormBucket) -> Result<(), RemoteStorageError> {
173    thread::scope(|scope| {
174        scope
175            .spawn(|| {
176                runtime::Builder::new_current_thread()
177                    .enable_all()
178                    .build()
179                    .map_err(ObjectStoreError::from)?
180                    .block_on(async {
181                        match bucket {
182                            WormBucket::S3(cfg) => verify_s3_worm_bucket(cfg).await,
183                            WormBucket::Gcs(cfg) => verify_gcs_worm_bucket(cfg).await,
184                            WormBucket::Unverified => unreachable!("checked by with_worm"),
185                        }
186                    })
187            })
188            .join()
189            .map_err(|_| ObjectStoreError::Backend("WORM bucket check panicked".into()))?
190    })?;
191    Ok(())
192}
193
194#[cfg(test)]
195mod tests {
196    use {
197        super::S3RemoteStorage,
198        crate::{
199            s3::test_support::{WormAccess, worm_config},
200            worm::WormConfig,
201        },
202        assert2::check,
203        krabka_object_store::{
204            DEFAULT_MULTIPART_CHUNK_SIZE, DEFAULT_MULTIPART_THRESHOLD, S3Config,
205        },
206        krabka_units::prelude::{ByteSizeExt as _, kibibytes, mebibytes},
207        object_store::memory::InMemory,
208        std::sync::Arc,
209        tempfile::TempDir,
210    };
211
212    /// The multipart tunables cross two seams: in from
213    /// [`krabka_object_store`]'s primitive config, and back out to the
214    /// primitive-typed `ObjectOps` substrate. Both must be lossless for every
215    /// size the config can express, so a mis-scaled conversion, such as a
216    /// stray `* 1024`, cannot silently change when a segment switches to
217    /// multipart.
218    #[test]
219    fn multipart_tuning_round_trips_through_the_primitive_seams() {
220        let store = S3RemoteStorage::with_store(Arc::new(InMemory::new()), None);
221        check!(store.multipart_threshold.bytes_u64() == DEFAULT_MULTIPART_THRESHOLD);
222        check!(store.multipart_chunk_size.bytes_usize() == DEFAULT_MULTIPART_CHUNK_SIZE);
223
224        let tuned = store.with_multipart_tuning(mebibytes(64), kibibytes(512));
225        check!(tuned.multipart_threshold.bytes_u64() == 64 * 1024 * 1024);
226        check!(tuned.multipart_chunk_size.bytes_usize() == 512 * 1024);
227    }
228
229    #[test]
230    fn worm_refuses_disabled_conditional_put_before_contacting_s3() {
231        let keys = TempDir::new().unwrap();
232        let s3 = S3Config {
233            bucket: "archive".into(),
234            region: "us-east-1".into(),
235            conditional_put: false,
236            ..Default::default()
237        };
238
239        let error = S3RemoteStorage::from_s3_config(&s3)
240            .unwrap()
241            .with_worm(&worm_config(keys.path(), WormAccess::ReadWrite))
242            .unwrap_err();
243
244        check!(error.to_string().contains("conditional_put = true"));
245    }
246
247    #[test]
248    fn worm_refuses_an_opaque_store_without_policy_config() {
249        let error = S3RemoteStorage::with_store(Arc::new(InMemory::new()), None)
250            .with_worm(&WormConfig::default())
251            .unwrap_err();
252
253        check!(error.to_string().contains("object-store configuration"));
254    }
255}
krabka_remote_storage/s3/client.rs

krabka_remote_storage/s3/
client.rs
