krabka_remote_storage/
s3.rs

1//! [`S3RemoteStorage`] is an S3-compatible object-store
2//! [`RemoteStorageManager`](crate::RemoteStorageManager), the KIP-405
3//! production backend.
4//!
5//! This backend is built on the `object_store` crate, so it works against
6//! any `S3-API` endpoint: AWS S3, `MinIO`, and Cloudflare R2. Google Cloud
7//! Storage has a dedicated native backend. See
8//! [`from_gcs_config`](S3RemoteStorage::from_gcs_config) in [`crate::gcs`],
9//! which supports keyless GKE Workload Identity instead of the legacy
10//! S3-compatibility/HMAC shim.
11//!
12//! The trait method bodies are synchronous, and they mirror Kafka's blocking
13//! `RemoteStorageManager`. The broker drives them from `spawn_blocking`.
14//! Inside them this crate blocks on the async `object_store` calls with the
15//! current Tokio runtime handle. That handle is always present inside a
16//! `spawn_blocking` worker that Tokio spawned.
17//!
18//! ## Object-key layout
19//!
20//! ```text
21//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.log
22//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.index
23//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.timeindex
24//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.snapshot
25//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.leader_epoch_checkpoint
26//! <prefix?>/<topic>-<partition>-<topic_id_base64>/<base_offset>-<segment_id_base64>.txnindex
27//! ```
28//!
29//! Keys mirror [`LocalTieredStorage`](crate::LocalTieredStorage)'s
30//! directory layout so the two backends are observationally equivalent.
31//!
32//! ## Module layout
33//!
34//! This root holds the backend's state and its `Debug` rendering. Each child
35//! module holds one concern: `client` builds and configures the store and
36//! bridges its async calls, `keys` derives every object key, `copy`, `fetch`,
37//! and `delete` hold one storage operation each, and `manager` binds those
38//! three to the trait.
39
40use {
41    krabka_object_store::{GcsConfig, ObjectStoreClient, S3Config},
42    krabka_units::prelude::{ByteSize, ByteSizeExt as _},
43};
44
45mod client;
46mod copy;
47mod delete;
48mod fetch;
49mod keys;
50mod manager;
51
52#[cfg(test)]
53mod test_support;
54
55use {crate::worm::WormArchiver, std::fmt};
56
57/// A [`RemoteStorageManager`](crate::RemoteStorageManager) backed by any
58/// S3-compatible object store.
59///
60/// Construct it with [`S3RemoteStorage::with_store`], which accepts any
61/// `ObjectStore` impl, for in-process tests. Use
62/// [`S3RemoteStorage::from_s3_config`] for the production path, which builds
63/// an `AmazonS3` client from credentials, endpoint, and bucket.
64pub struct S3RemoteStorage {
65    ops: ObjectStoreClient,
66    /// Optional key prefix. The store joins it to every object key with
67    /// `/`. This lets multiple Krabka clusters share a bucket safely.
68    prefix: Option<String>,
69    /// File-size threshold above which uploads switch to S3 multipart.
70    multipart_threshold: ByteSize,
71    /// Per-part size used by the multipart path.
72    multipart_chunk_size: ByteSize,
73    /// Upper bound on a buffered index read, checked before and during GET.
74    max_index_bytes: u64,
75    /// `Some` when this backend is a write-once archive.
76    worm: Option<WormMode>,
77    pub(crate) worm_bucket: WormBucket,
78}
79
80pub enum WormBucket {
81    S3(S3Config),
82    Gcs(GcsConfig),
83    Unverified,
84}
85
86/// Resolved WORM state: the archiver (which owns the loaded key) plus the
87/// credential posture.
88struct WormMode {
89    archiver: WormArchiver,
90    write_only: bool,
91    require_version_id: bool,
92}
93
94/// Lifts a raw byte count from [`krabka_object_store`]'s config layer into
95/// the dimensioned domain. That config layer still uses primitive types.
96///
97/// The conversion saturates and does not wrap. A `usize` above `u64::MAX`
98/// cannot occur on any target Krabka builds for.
99pub fn size_from_usize(bytes: usize) -> ByteSize {
100    ByteSize::from_bytes(u64::try_from(bytes).unwrap_or(u64::MAX))
101}
102
103impl fmt::Debug for S3RemoteStorage {
104    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
105        f.debug_struct("S3RemoteStorage")
106            .field("prefix", &self.prefix)
107            // Mode only. The archiver holds live private-key material, so it
108            // never reaches a formatter.
109            .field("worm", &self.worm.is_some())
110            .field(
111                "write_only",
112                &self.worm.as_ref().is_some_and(|worm| worm.write_only),
113            )
114            .finish_non_exhaustive()
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use {
121        super::{
122            S3RemoteStorage, size_from_usize,
123            test_support::{rsm, worm_config},
124        },
125        crate::s3::test_support::WormAccess,
126        assert2::{assert, check},
127        krabka_units::prelude::ByteSizeExt as _,
128        object_store::memory::InMemory,
129        std::{fs, sync::Arc},
130        tempfile::TempDir,
131    };
132
133    /// `usize::MAX` has no `u64` image on a hypothetical 128-bit target. The
134    /// lift saturates and does not wrap to a tiny chunk size.
135    #[test]
136    fn size_from_usize_saturates_instead_of_wrapping() {
137        check!(size_from_usize(0).bytes_usize() == 0);
138        check!(size_from_usize(usize::MAX).bytes_usize() == usize::MAX);
139    }
140
141    #[test]
142    fn storage_debug_is_nonempty() {
143        // The S3RemoteStorage Debug impl must render something (a `fmt`
144        // replaced with `Ok(())` would print nothing).
145        let dbg = format!("{:?}", rsm(None));
146        assert!(dbg.contains("S3RemoteStorage"));
147    }
148
149    #[test]
150    fn worm_debug_reports_mode_without_leaking_the_key() {
151        let keys = TempDir::new().unwrap();
152        let cfg = worm_config(keys.path(), WormAccess::WriteOnly);
153        let key_path = cfg.signing_key_path.clone().unwrap();
154        let store = S3RemoteStorage::with_store(Arc::new(InMemory::new()), None)
155            .with_worm_unchecked(&cfg)
156            .unwrap();
157
158        let rendered = format!("{store:?}");
159        check!(rendered.contains("worm: true"));
160        check!(rendered.contains("write_only: true"));
161        // Neither the key nor the path to it reaches a log line.
162        check!(!rendered.contains(&key_path.display().to_string()));
163        check!(!rendered.contains(&hex::encode(fs::read(&key_path).unwrap())));
164
165        let plain = format!("{:?}", rsm(None));
166        check!(plain.contains("worm: false"));
167        check!(plain.contains("write_only: false"));
168    }
169}
krabka_remote_storage/s3.rs

krabka_remote_storage/
s3.rs
