objectstore_service/backend/
common.rs

1//! Shared trait definition and types for all backends.
2
3use std::fmt;
4use std::num::NonZeroU64;
5
6use objectstore_types::metadata::Metadata;
7use objectstore_types::range::{ByteRange, ContentRange};
8use objectstore_types::resumable::UploadProgress;
9use objectstore_types::time::Timestamp;
10
11use bytes::Bytes;
12
13use crate::error::{ErrorKind, Result};
14use crate::id::ObjectId;
15use crate::multipart::{
16    AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
17    ListPartsResponse, PartNumber, UploadId, UploadPartResponse,
18};
19use crate::resumable::BackendToken;
20use crate::stream::{ClientStream, PayloadStream};
21
22/// User agent string used for outgoing requests.
23///
24/// This intentionally has a "sentry" prefix so that it can easily be traced back to us.
25pub const USER_AGENT: &str = concat!("sentry-objectstore/", env!("CARGO_PKG_VERSION"));
26
27/// Backend response for put operations.
28pub type PutResponse = ();
29/// Backend response for get operations.
30pub type GetResponse = Option<(Metadata, Option<ContentRange>, PayloadStream)>;
31/// Backend response for metadata-only get operations.
32pub type MetadataResponse = Option<Metadata>;
33/// Backend response for delete operations.
34pub type DeleteResponse = ();
35
36/// Trait implemented by all storage backends.
37///
38/// Object operations take `access_time`, the timestamp of the caller's operation.
39/// Use it to decide whether an object or redirect has expired, so all steps of
40/// an operation use the same time, including retries and calls to other backends.
41/// An object is expired when its deadline is strictly earlier than `access_time`.
42/// Writes preserve the creation time and deadline in the supplied metadata.
43#[async_trait::async_trait]
44pub trait Backend: fmt::Debug + Send + Sync + 'static {
45    /// The backend name, used for diagnostics.
46    fn name(&self) -> &'static str;
47
48    /// Stores an object at the given path with the given metadata.
49    async fn put_object(
50        &self,
51        id: &ObjectId,
52        metadata: &Metadata,
53        stream: ClientStream,
54        access_time: Timestamp,
55    ) -> Result<PutResponse>;
56
57    /// Retrieves (part of) an object at the given path, returning its metadata, a description of
58    /// the part being returned, and the payload.
59    async fn get_object(
60        &self,
61        id: &ObjectId,
62        access_time: Timestamp,
63        range: Option<ByteRange>,
64    ) -> Result<GetResponse>;
65
66    /// Retrieves only the metadata for an object, without the payload.
67    async fn get_metadata(
68        &self,
69        id: &ObjectId,
70        access_time: Timestamp,
71    ) -> Result<MetadataResponse> {
72        Ok(self
73            .get_object(id, access_time, None)
74            .await?
75            .map(|(metadata, _range, _stream)| metadata))
76    }
77
78    /// Extends the deadline of an existing object with expiration policy.
79    ///
80    /// This only changes the stored deadline: the expiration policy, duration,
81    /// payload, and all other metadata remain unchanged.
82    ///
83    /// Returns `true` when the deadline was extended or was already at least as
84    /// late as `expire_at`. Returns `false` when the object is absent, expired,
85    /// manually expired, or changed concurrently.
86    async fn set_expiry(
87        &self,
88        id: &ObjectId,
89        expire_at: Timestamp,
90        access_time: Timestamp,
91    ) -> Result<bool>;
92
93    /// Deletes the object at the given path.
94    async fn delete_object(&self, id: &ObjectId, access_time: Timestamp) -> Result<DeleteResponse>;
95
96    /// Waits for any outstanding background operations to complete before shutdown.
97    ///
98    /// The default implementation is a no-op. Backends that spawn background tasks
99    /// (such as [`TieredStorage`](super::tiered::TieredStorage)) should override this
100    /// to wait for those tasks to complete.
101    async fn join(&self) {}
102
103    /// Borrows this backend as a [`MultipartUploadBackend`] if supported.
104    ///
105    /// The default returns an [`ErrorKind::Unsupported`]. Backends that implement
106    /// [`MultipartUploadBackend`] should override this to return `Ok(self)`.
107    fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> {
108        Err(ErrorKind::Unsupported.into())
109    }
110
111    /// Creates a resumable upload session for the object at `id`.
112    ///
113    /// Object metadata and its total length are declared upfront and cannot be mutated
114    /// during the upload.
115    ///
116    /// The returned string is opaque backend-defined state. [`StorageService`](crate::StorageService)
117    /// protects it before exposing the session token outside the service layer.
118    ///
119    /// Returns `Ok(None)` when this backend cannot store the described object resumably. Declining
120    /// is a routine outcome rather than an error, and the default implementation declines.
121    ///
122    /// # Errors
123    ///
124    /// Returns an error only when the backend supports resumable uploads but failed to open the
125    /// session.
126    async fn create_upload_session(
127        &self,
128        id: &ObjectId,
129        metadata: &Metadata,
130        total_length: NonZeroU64,
131    ) -> Result<Option<BackendToken>> {
132        let _ = (id, metadata, total_length);
133        Ok(None)
134    }
135
136    /// Writes a chunk of `content_length` bytes at `offset` into an open session.
137    ///
138    /// A backend may acknowledge fewer bytes than the chunk supplied, for example by persisting
139    /// only an aligned prefix. Callers must continue from the authoritative offset in the returned
140    /// [`UploadProgress`], or query [`Self::upload_offset`] after an ambiguous failure. A backend
141    /// may or may not accept a replay starting before its persisted offset.
142    ///
143    /// [`UploadProgress::Complete`] means the upload is terminal and the object is available
144    /// through this backend's normal read methods. A backend that composes another backend must
145    /// finish its own publication work before returning that outcome.
146    ///
147    /// A `content_length` of zero is valid. It writes nothing and reports the offset the backend
148    /// holds.
149    ///
150    /// Returns [`ErrorKind::UnknownUploadSession`] when `token` does not identify an open session,
151    /// and [`ErrorKind::ChunkExceedsUploadLength`] when the chunk would exceed the total length
152    /// declared when the session was created.
153    async fn put_chunk(
154        &self,
155        id: &ObjectId,
156        token: &BackendToken,
157        offset: u64,
158        content_length: u64,
159        stream: ClientStream,
160    ) -> Result<UploadProgress> {
161        let _ = (id, token, offset, content_length, stream);
162        Err(ErrorKind::Unsupported.into())
163    }
164
165    /// Reports how far the session has progressed.
166    ///
167    /// This can return [`UploadProgress::Complete`] repeatedly after the final chunk, including
168    /// when its original response was lost. A composed backend may finish pending idempotent
169    /// publication work before returning that terminal outcome.
170    ///
171    /// Returns [`ErrorKind::UnknownUploadSession`] when `token` does not identify a known session.
172    async fn upload_offset(&self, id: &ObjectId, token: &BackendToken) -> Result<UploadProgress> {
173        let _ = (id, token);
174        Err(ErrorKind::Unsupported.into())
175    }
176
177    /// Cancels an upload session, discarding whatever was uploaded.
178    ///
179    /// Returns [`ErrorKind::UnknownUploadSession`] when `token` does not identify an open session.
180    async fn cancel_upload(&self, id: &ObjectId, token: &BackendToken) -> Result<()> {
181        let _ = (id, token);
182        Err(ErrorKind::Unsupported.into())
183    }
184}
185
186/// Trait for backends that support our S3-style multipart upload protocol.
187#[async_trait::async_trait]
188pub trait MultipartUploadBackend: Backend + fmt::Debug + Send + Sync + 'static {
189    /// Initiates a new multipart upload at `id` with the given metadata.
190    async fn initiate_multipart(
191        &self,
192        id: &ObjectId,
193        metadata: &Metadata,
194    ) -> Result<InitiateMultipartResponse>;
195
196    /// Uploads a single part of the upload identified by `(id, upload_id)`.
197    async fn upload_part(
198        &self,
199        id: &ObjectId,
200        upload_id: &UploadId,
201        part_number: PartNumber,
202        content_length: u64,
203        content_md5: Option<&str>,
204        body: ClientStream,
205    ) -> Result<UploadPartResponse>;
206
207    /// Lists the parts uploaded so far for `(id, upload_id)`.
208    async fn list_parts(
209        &self,
210        id: &ObjectId,
211        upload_id: &UploadId,
212        max_parts: Option<u32>,
213        part_number_marker: Option<PartNumber>,
214    ) -> Result<ListPartsResponse>;
215
216    /// Aborts the upload identified by `(id, upload_id)`.
217    async fn abort_multipart(
218        &self,
219        id: &ObjectId,
220        upload_id: &UploadId,
221    ) -> Result<AbortMultipartResponse>;
222
223    /// Finalizes the upload identified by `(id, upload_id)` with the given
224    /// ordered list of parts.
225    ///
226    /// Note that this returns `Result<Option<CompleteMultipartError>>`.
227    /// It's therefore possible to get `Ok(Some(err))`, meaning that at the server level this will
228    /// translate to HTTP `200 OK` with an error contained in the response body.
229    /// We need to do it this way to mirror backends that also behave like this (namely S3 and
230    /// GCS).
231    async fn complete_multipart(
232        &self,
233        id: &ObjectId,
234        upload_id: &UploadId,
235        parts: Vec<CompletedPart>,
236        access_time: Timestamp,
237    ) -> Result<CompleteMultipartResponse>;
238}
239
240/// Trait for backends that support tombstone-conditional operations.
241///
242/// Only backends suitable for the high-volume tier of
243/// [`TieredStorage`](super::tiered::TieredStorage) implement this trait.
244/// The conditional methods provide atomic operations to avoid overwriting
245/// redirect tombstones.
246#[async_trait::async_trait]
247pub trait HighVolumeBackend: Backend {
248    /// Writes the object only if NO redirect tombstone exists at this key.
249    ///
250    /// Returns `None` after storing the object, or `Some(tombstone)` (skipping
251    /// the write) when a redirect tombstone is present. The returned tombstone
252    /// carries the target LT `ObjectId` so the caller can route without a
253    /// second round trip.
254    ///
255    /// Takes [`Bytes`] instead of a [`ClientStream`] because callers on this
256    /// path have already fully buffered the payload.
257    async fn put_non_tombstone(
258        &self,
259        id: &ObjectId,
260        metadata: &Metadata,
261        payload: Bytes,
262        access_time: Timestamp,
263    ) -> Result<Option<Tombstone>>;
264
265    /// Retrieves (part of) an object with explicit tombstone awareness.
266    ///
267    /// Returns [`TieredGet::Tombstone`] instead of synthesizing a tombstone
268    /// object, making the caller's routing logic a compile-time distinction.
269    async fn get_tiered_object(
270        &self,
271        id: &ObjectId,
272        access_time: Timestamp,
273        range: Option<ByteRange>,
274    ) -> Result<TieredGet>;
275
276    /// Retrieves only metadata with explicit tombstone awareness.
277    ///
278    /// Implementations should skip the payload column where possible to avoid
279    /// fetching up to 1 MiB of data just to discover a tombstone.
280    async fn get_tiered_metadata(
281        &self,
282        id: &ObjectId,
283        access_time: Timestamp,
284    ) -> Result<TieredMetadata>;
285
286    /// Deletes the object only if it is NOT a redirect tombstone.
287    ///
288    /// Returns `None` after deleting the row (or if the row was already absent),
289    /// or `Some(tombstone)` (leaving the row intact) when the object is a
290    /// redirect tombstone. The returned tombstone carries the target LT
291    /// `ObjectId` so the caller can delete from long-term storage directly,
292    /// without a second round trip.
293    async fn delete_non_tombstone(
294        &self,
295        id: &ObjectId,
296        access_time: Timestamp,
297    ) -> Result<Option<Tombstone>>;
298
299    /// Atomically mutates the row if the current redirect state matches.
300    ///
301    /// `current` determines the precondition:
302    /// - `None`: succeeds only if no live tombstone exists (row absent, inline,
303    ///   or tombstone present but logically expired).
304    /// - `Some(target)`: succeeds only if a tombstone exists whose redirect
305    ///   resolves to `target`.
306    ///
307    /// **This operation is idempotent:** if the object is already in the target
308    /// state, it returns `true`. Whether the mutation runs again is up to the
309    /// implementation.
310    ///
311    /// Returns `true` on success or idempotent match, `false` if a conflicting
312    /// state was found (another writer won the race).
313    async fn compare_and_write(
314        &self,
315        id: &ObjectId,
316        current: Option<&ObjectId>,
317        write: TieredWrite,
318        access_time: Timestamp,
319    ) -> Result<bool>;
320
321    /// Atomically updates an existing row if its kind and redirect target match.
322    ///
323    /// `current = None` requires a live inline object. `Some(target)` requires
324    /// a live redirect to exactly that target. Updates never authorize creation
325    /// of an absent row.
326    ///
327    /// Returns `true` when the update was applied or its requested state was
328    /// already satisfied. Returns `false` for an absent, expired, or conflicting
329    /// entry.
330    async fn compare_and_update(
331        &self,
332        id: &ObjectId,
333        current: Option<&ObjectId>,
334        update: TieredUpdate,
335        access_time: Timestamp,
336    ) -> Result<bool>;
337}
338
339/// Information about a redirect tombstone in the high-volume backend.
340#[derive(Clone, Debug, PartialEq, Eq)]
341pub struct Tombstone {
342    /// The [`ObjectId`] of the object in the long-term backend.
343    ///
344    /// For legacy tombstones with an empty `r` column, the HV backend resolves
345    /// this to the HV `ObjectId` itself before surfacing the tombstone to callers.
346    pub target: ObjectId,
347
348    /// The concrete deadline stored on the redirect.
349    pub time_expires: Option<Timestamp>,
350}
351
352impl Tombstone {
353    /// Returns whether the tombstone has expired at the given time.
354    pub fn is_expired(&self, now: Timestamp) -> bool {
355        self.time_expires.is_some_and(|deadline| deadline < now)
356    }
357}
358
359/// Typed response from [`HighVolumeBackend::get_tiered_object`].
360pub enum TieredGet {
361    /// A real object was found.
362    Object(Metadata, Option<ContentRange>, PayloadStream),
363    /// A redirect tombstone was found; the real object lives in the long-term backend.
364    Tombstone(Tombstone),
365    /// No entry exists at this key.
366    NotFound,
367}
368
369impl fmt::Debug for TieredGet {
370    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
371        match self {
372            TieredGet::Object(metadata, content_range, _stream) => f
373                .debug_tuple("Object")
374                .field(metadata)
375                .field(content_range)
376                .finish_non_exhaustive(),
377            TieredGet::Tombstone(info) => f.debug_tuple("Tombstone").field(info).finish(),
378            TieredGet::NotFound => write!(f, "NotFound"),
379        }
380    }
381}
382
383/// Typed metadata-only response from [`HighVolumeBackend::get_tiered_metadata`].
384#[derive(Debug)]
385pub enum TieredMetadata {
386    /// Metadata for a real object was found.
387    Object(Metadata),
388    /// A redirect tombstone was found; the real object lives in the long-term backend.
389    Tombstone(Tombstone),
390    /// No entry exists at this key.
391    NotFound,
392}
393
394/// The write operation performed by [`HighVolumeBackend::compare_and_write`].
395#[derive(Clone, Debug)]
396pub enum TieredWrite {
397    /// Write a redirect tombstone.
398    Tombstone(Tombstone),
399    /// Write inline object data.
400    Object(Metadata, Bytes),
401    /// Delete the row entirely.
402    Delete,
403}
404
405impl TieredWrite {
406    /// Returns the tombstone target if this is a tombstone write, or `None` otherwise.
407    pub fn target(&self) -> Option<&ObjectId> {
408        match self {
409            TieredWrite::Tombstone(t) => Some(&t.target),
410            _ => None,
411        }
412    }
413}
414
415/// The in-place operation performed by [`HighVolumeBackend::compare_and_update`].
416#[derive(Clone, Debug)]
417pub enum TieredUpdate {
418    /// Extend the deadline while preserving all other stored data.
419    SetExpiry(Timestamp),
420}
421
422/// Creates a reqwest client with required defaults.
423///
424/// Automatic decompression is disabled because backends store pre-compressed
425/// payloads and manage `Content-Encoding` themselves.
426pub(super) fn reqwest_client() -> reqwest::Client {
427    reqwest::Client::builder()
428        .user_agent(USER_AGENT)
429        .hickory_dns(true)
430        .http1_only()
431        .no_zstd()
432        .no_brotli()
433        .no_gzip()
434        .no_deflate()
435        .build()
436        // INVARIANT: Building fails only if the TLS backend cannot be initialized, which
437        // is checked at startup when the rustls crypto provider is installed.
438        .expect("failed to build backend HTTP client")
439}
objectstore_service/backend/common.rs

objectstore_service/backend/
common.rs
