objectstore_service/backend/
in_memory.rs

1//! In-memory backend for tests.
2//!
3//! This provides a [`Backend`](super::common::Backend) backed by a `HashMap`,
4//! removing the need for filesystem tempdir management in unit tests. The
5//! backend is [`Clone`] so tests can hold a handle for direct inspection while
6//! the service owns a boxed copy.
7
8use std::collections::{BTreeMap, HashMap};
9use std::sync::{Arc, Mutex};
10use std::time::SystemTime;
11
12use objectstore_types::range::ByteRange;
13use objectstore_types::time::Timestamp;
14
15use bytes::{Bytes, BytesMut};
16use futures_util::TryStreamExt;
17use objectstore_types::metadata::Metadata;
18
19use super::common::{
20    DeleteResponse, GetResponse, HighVolumeBackend, MultipartUploadBackend, PutResponse, TieredGet,
21    TieredMetadata, TieredUpdate, TieredWrite, Tombstone,
22};
23use crate::change_stream::{ChangeStream, NoopStream, flush_change_stream};
24use crate::error::{Error, ErrorKind, Result};
25use crate::id::ObjectId;
26use crate::multipart::{
27    AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
28    ListPartsResponse, Part, PartNumber, UploadId, UploadPartResponse,
29};
30use crate::stream::ClientStream;
31
32/// An entry in the in-memory store.
33#[derive(Clone, Debug)]
34enum StoreEntry {
35    Object(Metadata, Bytes),
36    Tombstone(Tombstone),
37}
38
39impl StoreEntry {
40    fn is_expired(&self, now: Timestamp) -> bool {
41        match self {
42            StoreEntry::Object(metadata, _) => metadata.is_expired(now),
43            StoreEntry::Tombstone(tombstone) => tombstone.is_expired(now),
44        }
45    }
46
47    /// Number of bytes an entry occupies (payload plus serialized metadata)
48    pub(crate) fn stored_size(&self) -> usize {
49        match self {
50            StoreEntry::Object(metadata, payload) => json_len(metadata) + payload.len(),
51            // A tombstone carries no payload: its bytes are the redirect target plus the
52            // deadline stored alongside it.
53            StoreEntry::Tombstone(tombstone) => {
54                tombstone.target.as_storage_path().to_string().len()
55                    + tombstone.time_expires.map_or(0, |d| json_len(&d))
56            }
57        }
58    }
59}
60
61type Store = HashMap<ObjectId, StoreEntry>;
62
63#[derive(Clone, Debug)]
64struct MultipartUpload {
65    metadata: Metadata,
66    parts: BTreeMap<PartNumber, UploadedPart>,
67}
68
69#[derive(Clone, Debug)]
70struct UploadedPart {
71    etag: String,
72    data: Bytes,
73    uploaded_at: SystemTime,
74}
75
76type MultipartStore = HashMap<(ObjectId, UploadId), MultipartUpload>;
77
78/// In-memory [`Backend`](super::common::Backend) backed by a `HashMap`.
79///
80/// Removes the need for filesystem tempdir management in unit tests. The
81/// backend is [`Clone`] so tests can hold a handle for direct inspection while
82/// the service owns a boxed copy.
83#[derive(Debug, Clone)]
84pub struct InMemoryBackend {
85    name: &'static str,
86    store: Arc<Mutex<Store>>,
87    multipart_store: Arc<Mutex<MultipartStore>>,
88    change_stream: Arc<dyn ChangeStream>,
89}
90
91impl InMemoryBackend {
92    /// Creates a new `InMemoryBackend` with the given diagnostic `name`.
93    pub fn new(name: &'static str) -> Self {
94        Self {
95            name,
96            store: Arc::new(Mutex::new(HashMap::new())),
97            multipart_store: Arc::new(Mutex::new(HashMap::new())),
98            change_stream: Arc::new(NoopStream),
99        }
100    }
101
102    /// Publishes this backend's changes to `change_stream`.
103    pub fn with_change_stream(mut self, change_stream: Arc<dyn ChangeStream>) -> Self {
104        self.change_stream = change_stream;
105        self
106    }
107
108    /// Returns the stored entry for `id`, for direct inspection in tests.
109    pub fn get(&self, id: &ObjectId) -> Entry {
110        match self.store.lock().unwrap().get(id).cloned() {
111            None => Entry::NotFound,
112            Some(StoreEntry::Tombstone(tombstone)) => Entry::Tombstone(tombstone),
113            Some(StoreEntry::Object(metadata, bytes)) => Entry::Object(metadata, bytes),
114        }
115    }
116
117    /// Returns `true` if the backend contains an entry for the given id.
118    pub fn contains(&self, id: &ObjectId) -> bool {
119        self.store.lock().unwrap().contains_key(id)
120    }
121
122    /// Returns `true` if the backend has no stored objects.
123    pub fn is_empty(&self) -> bool {
124        self.store.lock().unwrap().is_empty()
125    }
126
127    /// Removes an entry directly, bypassing the `Backend` trait.
128    ///
129    /// Useful for simulating partial failures (e.g. orphan tombstones).
130    pub fn remove(&self, id: &ObjectId) {
131        self.store.lock().unwrap().remove(id);
132    }
133}
134
135#[async_trait::async_trait]
136impl super::common::Backend for InMemoryBackend {
137    fn name(&self) -> &'static str {
138        self.name
139    }
140
141    fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> {
142        Ok(self)
143    }
144
145    async fn put_object(
146        &self,
147        id: &ObjectId,
148        metadata: &Metadata,
149        stream: ClientStream,
150        _access_time: Timestamp,
151    ) -> Result<PutResponse> {
152        let bytes: BytesMut = stream.try_collect().await?;
153        let entry = StoreEntry::Object(metadata.clone(), bytes.freeze());
154        let size = entry.stored_size();
155        self.store.lock().unwrap().insert(id.clone(), entry);
156        self.change_stream
157            .write(id, size as u64, metadata.time_expires);
158        Ok(())
159    }
160
161    async fn get_object(
162        &self,
163        id: &ObjectId,
164        access_time: Timestamp,
165        range: Option<ByteRange>,
166    ) -> Result<GetResponse> {
167        let entry = self.store.lock().unwrap().get(id).cloned();
168        match entry {
169            None => Ok(None),
170            Some(entry) if entry.is_expired(access_time) => Ok(None),
171            Some(StoreEntry::Tombstone(_)) => Err(ErrorKind::UnexpectedTombstone.into()),
172            Some(StoreEntry::Object(mut metadata, bytes)) => {
173                let total = bytes.len() as u64;
174                metadata.size = Some(bytes.len());
175                let (content_range, payload) = match range {
176                    Some(range) => {
177                        let content_range = range
178                            .resolve(total)
179                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
180                        let sliced =
181                            bytes.slice(content_range.start as usize..=content_range.end as usize);
182                        (Some(content_range), sliced)
183                    }
184                    None => (None, bytes),
185                };
186                Ok(Some((
187                    metadata,
188                    content_range,
189                    crate::stream::single(payload),
190                )))
191            }
192        }
193    }
194
195    async fn set_expiry(
196        &self,
197        id: &ObjectId,
198        expire_at: Timestamp,
199        access_time: Timestamp,
200    ) -> Result<bool> {
201        let outcome = {
202            let mut store = self.store.lock().unwrap();
203            match store.get_mut(id) {
204                Some(StoreEntry::Object(metadata, _)) => {
205                    extend_expiry(&mut metadata.time_expires, expire_at, access_time)
206                }
207                _ => ExpiryOutcome::Rejected,
208            }
209        };
210
211        if outcome == ExpiryOutcome::Extended {
212            self.change_stream.update(id, Some(expire_at));
213        }
214
215        Ok(outcome.is_satisfied())
216    }
217
218    async fn delete_object(
219        &self,
220        id: &ObjectId,
221        _access_time: Timestamp,
222    ) -> Result<DeleteResponse> {
223        if self.store.lock().unwrap().remove(id).is_some() {
224            self.change_stream.delete(id);
225        }
226        Ok(())
227    }
228
229    async fn join(&self) {
230        flush_change_stream(&self.change_stream).await;
231    }
232}
233
234#[async_trait::async_trait]
235impl HighVolumeBackend for InMemoryBackend {
236    async fn put_non_tombstone(
237        &self,
238        id: &ObjectId,
239        metadata: &Metadata,
240        payload: Bytes,
241        access_time: Timestamp,
242    ) -> Result<Option<Tombstone>> {
243        let mut store = self.store.lock().unwrap();
244        if let Some(StoreEntry::Tombstone(tombstone)) = store.get(id)
245            && !tombstone.is_expired(access_time)
246        {
247            return Ok(Some(tombstone.clone()));
248        }
249
250        let mut metadata = metadata.clone();
251        metadata.size = Some(payload.len());
252        let expires_at = metadata.time_expires;
253        let entry = StoreEntry::Object(metadata, payload);
254        let size = entry.stored_size();
255        store.insert(id.clone(), entry);
256        self.change_stream.write(id, size as u64, expires_at);
257        Ok(None)
258    }
259
260    async fn get_tiered_object(
261        &self,
262        id: &ObjectId,
263        access_time: Timestamp,
264        range: Option<ByteRange>,
265    ) -> Result<TieredGet> {
266        let entry = self.store.lock().unwrap().get(id).cloned();
267        Ok(match entry {
268            None => TieredGet::NotFound,
269            Some(entry) if entry.is_expired(access_time) => TieredGet::NotFound,
270            Some(StoreEntry::Tombstone(tombstone)) => TieredGet::Tombstone(tombstone),
271            Some(StoreEntry::Object(mut metadata, bytes)) => {
272                let total = bytes.len() as u64;
273                metadata.size = Some(bytes.len());
274                let (content_range, payload) = match range {
275                    Some(range) => {
276                        let content_range = range
277                            .resolve(total)
278                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
279                        let sliced =
280                            bytes.slice(content_range.start as usize..=content_range.end as usize);
281                        (Some(content_range), sliced)
282                    }
283                    None => (None, bytes),
284                };
285                TieredGet::Object(metadata, content_range, crate::stream::single(payload))
286            }
287        })
288    }
289
290    async fn get_tiered_metadata(
291        &self,
292        id: &ObjectId,
293        access_time: Timestamp,
294    ) -> Result<TieredMetadata> {
295        let entry = self.store.lock().unwrap().get(id).cloned();
296        Ok(match entry {
297            None => TieredMetadata::NotFound,
298            Some(entry) if entry.is_expired(access_time) => TieredMetadata::NotFound,
299            Some(StoreEntry::Tombstone(tombstone)) => TieredMetadata::Tombstone(tombstone),
300            Some(StoreEntry::Object(metadata, _bytes)) => TieredMetadata::Object(metadata),
301        })
302    }
303
304    async fn delete_non_tombstone(
305        &self,
306        id: &ObjectId,
307        access_time: Timestamp,
308    ) -> Result<Option<Tombstone>> {
309        let mut store = self.store.lock().unwrap();
310        if let Some(StoreEntry::Tombstone(tombstone)) = store.get(id).cloned()
311            && !tombstone.is_expired(access_time)
312        {
313            return Ok(Some(tombstone));
314        }
315
316        if store.remove(id).is_some() {
317            self.change_stream.delete(id);
318        }
319        Ok(None)
320    }
321
322    async fn compare_and_update(
323        &self,
324        id: &ObjectId,
325        current: Option<&ObjectId>,
326        update: TieredUpdate,
327        access_time: Timestamp,
328    ) -> Result<bool> {
329        let TieredUpdate::SetExpiry(expire_at) = update;
330        let outcome = {
331            let mut store = self.store.lock().unwrap();
332            match (store.get_mut(id), current) {
333                (Some(StoreEntry::Object(metadata, _)), None) => {
334                    extend_expiry(&mut metadata.time_expires, expire_at, access_time)
335                }
336                (Some(StoreEntry::Tombstone(t)), Some(target)) if t.target == *target => {
337                    extend_expiry(&mut t.time_expires, expire_at, access_time)
338                }
339                _ => ExpiryOutcome::Rejected,
340            }
341        };
342
343        if outcome == ExpiryOutcome::Extended {
344            self.change_stream.update(id, Some(expire_at));
345        }
346
347        Ok(outcome.is_satisfied())
348    }
349
350    async fn compare_and_write(
351        &self,
352        id: &ObjectId,
353        current: Option<&ObjectId>,
354        write: TieredWrite,
355        access_time: Timestamp,
356    ) -> Result<bool> {
357        let mut store = self.store.lock().unwrap();
358
359        let actual = store.get(id);
360        let matches_current = matches_redirect(actual, current, access_time);
361        let matches_next = matches_redirect(actual, write.target(), access_time);
362
363        if matches_current {
364            match write {
365                TieredWrite::Tombstone(tombstone) => {
366                    let expires_at = tombstone.time_expires;
367                    let entry = StoreEntry::Tombstone(tombstone);
368                    let size = entry.stored_size();
369                    store.insert(id.clone(), entry);
370                    self.change_stream.write(id, size as u64, expires_at);
371                }
372                TieredWrite::Object(metadata, payload) => {
373                    let expires_at = metadata.time_expires;
374                    let entry = StoreEntry::Object(metadata, payload);
375                    let size = entry.stored_size();
376                    store.insert(id.clone(), entry);
377                    self.change_stream.write(id, size as u64, expires_at);
378                }
379                TieredWrite::Delete => {
380                    if store.remove(id).is_some() {
381                        self.change_stream.delete(id);
382                    }
383                }
384            }
385        }
386
387        Ok(matches_current || matches_next)
388    }
389}
390
391#[async_trait::async_trait]
392impl MultipartUploadBackend for InMemoryBackend {
393    async fn initiate_multipart(
394        &self,
395        id: &ObjectId,
396        metadata: &Metadata,
397    ) -> Result<InitiateMultipartResponse> {
398        let upload_id = UploadId::new(uuid::Uuid::now_v7().to_string())?;
399        let upload = MultipartUpload {
400            metadata: metadata.clone(),
401            parts: BTreeMap::new(),
402        };
403        self.multipart_store
404            .lock()
405            .unwrap()
406            .insert((id.clone(), upload_id.clone()), upload);
407        Ok(upload_id)
408    }
409
410    async fn upload_part(
411        &self,
412        id: &ObjectId,
413        upload_id: &UploadId,
414        part_number: PartNumber,
415        _content_length: u64,
416        _content_md5: Option<&str>,
417        body: ClientStream,
418    ) -> Result<UploadPartResponse> {
419        let data: BytesMut = body.try_collect().await?;
420        let data = data.freeze();
421        let etag = format!("\"etag-{part_number}-{}\"", data.len());
422
423        let mut store = self.multipart_store.lock().unwrap();
424        let upload = store
425            .get_mut(&(id.clone(), upload_id.clone()))
426            .ok_or_else(|| {
427                Error::new(
428                    ErrorKind::BackendFailure,
429                    "in-memory multipart upload not found",
430                )
431            })?;
432
433        upload.parts.insert(
434            part_number,
435            UploadedPart {
436                etag: etag.clone(),
437                data,
438                uploaded_at: SystemTime::now(),
439            },
440        );
441
442        Ok(etag)
443    }
444
445    async fn list_parts(
446        &self,
447        id: &ObjectId,
448        upload_id: &UploadId,
449        max_parts: Option<u32>,
450        part_number_marker: Option<PartNumber>,
451    ) -> Result<ListPartsResponse> {
452        let store = self.multipart_store.lock().unwrap();
453        let upload = store.get(&(id.clone(), upload_id.clone())).ok_or_else(|| {
454            Error::new(
455                ErrorKind::BackendFailure,
456                "in-memory multipart upload not found",
457            )
458        })?;
459
460        let iter = upload
461            .parts
462            .iter()
463            .filter(|(pn, _)| part_number_marker.is_none_or(|marker| **pn > marker));
464
465        let max = max_parts.unwrap_or(u32::MAX) as usize;
466        let all: Vec<_> = iter.collect();
467        let is_truncated = all.len() > max;
468        let page: Vec<_> = all.into_iter().take(max).collect();
469
470        let next_part_number_marker = if is_truncated {
471            page.last().map(|(pn, _)| **pn)
472        } else {
473            None
474        };
475
476        let parts = page
477            .into_iter()
478            .map(|(pn, part)| Part {
479                part_number: *pn,
480                etag: part.etag.clone(),
481                last_modified: part.uploaded_at,
482                size: part.data.len() as u64,
483            })
484            .collect();
485
486        Ok(ListPartsResponse {
487            parts,
488            is_truncated,
489            next_part_number_marker,
490        })
491    }
492
493    async fn abort_multipart(
494        &self,
495        id: &ObjectId,
496        upload_id: &UploadId,
497    ) -> Result<AbortMultipartResponse> {
498        self.multipart_store
499            .lock()
500            .unwrap()
501            .remove(&(id.clone(), upload_id.clone()));
502        Ok(())
503    }
504
505    async fn complete_multipart(
506        &self,
507        id: &ObjectId,
508        upload_id: &UploadId,
509        parts: Vec<CompletedPart>,
510        _access_time: Timestamp,
511    ) -> Result<CompleteMultipartResponse> {
512        let key = (id.clone(), upload_id.clone());
513
514        // TODO: validate that parts are in ascending part_number order and reject with
515        // InvalidPartOrder if not (matches S3/GCS behavior). Needs a proper client error variant.
516
517        // Validate and assemble while holding the multipart lock, but don't
518        // remove the upload yet — a failed validation must leave it intact so
519        // the client can retry.
520        let (metadata, payload) = {
521            let store = self.multipart_store.lock().unwrap();
522            let upload = store.get(&key).ok_or_else(|| {
523                Error::new(
524                    ErrorKind::BackendFailure,
525                    "in-memory multipart upload not found",
526                )
527            })?;
528
529            for completed in &parts {
530                match upload.parts.get(&completed.part_number) {
531                    None => {
532                        return Ok(Some(crate::multipart::CompleteMultipartError {
533                            code: "InvalidPart".into(),
534                            message: format!(
535                                "part number {} was not uploaded",
536                                completed.part_number
537                            ),
538                        }));
539                    }
540                    Some(stored) if stored.etag != completed.etag => {
541                        return Ok(Some(crate::multipart::CompleteMultipartError {
542                            code: "InvalidPart".into(),
543                            message: format!(
544                                "etag mismatch for part {}: expected {}, got {}",
545                                completed.part_number, stored.etag, completed.etag
546                            ),
547                        }));
548                    }
549                    _ => {}
550                }
551            }
552
553            let mut payload = BytesMut::new();
554            for completed in &parts {
555                let stored = &upload.parts[&completed.part_number];
556                payload.extend_from_slice(&stored.data);
557            }
558
559            let mut metadata = upload.metadata.clone();
560            metadata.size = Some(payload.len());
561
562            (metadata, payload.freeze())
563        };
564
565        let expires_at = metadata.time_expires;
566        let entry = StoreEntry::Object(metadata, payload);
567        let size = entry.stored_size();
568        self.store.lock().unwrap().insert(id.clone(), entry);
569        self.change_stream.write(id, size as u64, expires_at);
570
571        self.multipart_store.lock().unwrap().remove(&key);
572
573        Ok(None)
574    }
575}
576
577/// Serialized length of `value`, or `0` if it cannot be serialized.
578fn json_len<T: serde::Serialize>(value: &T) -> usize {
579    serde_json::to_string(value).map_or(0, |json| json.len())
580}
581
582/// Returns `true` if `entry` matches the expected tombstone redirect state.
583///
584/// - `expected = None`: matches any non-tombstone (absent or inline object).
585/// - `expected = Some(target)`: matches a tombstone whose redirect target equals `target`.
586fn matches_redirect(
587    entry: Option<&StoreEntry>,
588    expected: Option<&ObjectId>,
589    now: Timestamp,
590) -> bool {
591    match entry {
592        None | Some(StoreEntry::Object(..)) => expected.is_none(),
593        Some(StoreEntry::Tombstone(tombstone)) => match expected {
594            None => tombstone.is_expired(now),
595            Some(target) => tombstone.target == *target && !tombstone.is_expired(now),
596        },
597    }
598}
599
600/// What [`extend_expiry`] did to an entry's deadline.
601///
602/// Callers need to tell a deadline that actually moved apart from one that already
603/// covered the request, because only the former is a change worth reporting to the
604/// [`ChangeStream`].
605#[derive(Clone, Copy, Debug, PartialEq, Eq)]
606enum ExpiryOutcome {
607    /// The deadline could not be extended. None is set, or it has already passed.
608    Rejected,
609    /// The stored deadline already covered `expire_at`. Nothing was written.
610    AlreadySatisfied,
611    /// The stored deadline moved out to `expire_at`.
612    Extended,
613}
614
615impl ExpiryOutcome {
616    /// Whether the entry now expires no earlier than the requested deadline.
617    fn is_satisfied(self) -> bool {
618        !matches!(self, Self::Rejected)
619    }
620}
621
622/// Extends an active expiry time to `expire_at` where valid.
623fn extend_expiry(
624    field: &mut Option<Timestamp>,
625    expire_at: Timestamp,
626    now: Timestamp,
627) -> ExpiryOutcome {
628    let Some(time_expires) = *field else {
629        return ExpiryOutcome::Rejected; // entries without a deadline cannot be extended
630    };
631
632    if time_expires < now {
633        ExpiryOutcome::Rejected // already expired
634    } else if time_expires >= expire_at {
635        ExpiryOutcome::AlreadySatisfied
636    } else {
637        *field = Some(expire_at);
638        ExpiryOutcome::Extended
639    }
640}
641
642/// Type returned by [`InMemoryBackend::get`] for direct inspection of stored entries.
643#[derive(Clone, Debug)]
644pub enum Entry {
645    /// No entry exists at this key.
646    NotFound,
647    /// A real object with its metadata and payload bytes.
648    Object(Metadata, Bytes),
649    /// A redirect tombstone indicating the real object lives in the long-term backend.
650    Tombstone(Tombstone),
651}
652
653impl Entry {
654    /// Returns `true` if the entry is [`Entry::NotFound`].
655    pub fn is_not_found(&self) -> bool {
656        matches!(self, Entry::NotFound)
657    }
658
659    /// Returns `true` if the entry is [`Entry::Object`].
660    pub fn is_object(&self) -> bool {
661        matches!(self, Entry::Object(_, _))
662    }
663
664    /// Returns `true` if the entry is [`Entry::Tombstone`].
665    pub fn is_tombstone(&self) -> bool {
666        matches!(self, Entry::Tombstone(_))
667    }
668
669    /// Panics unless the entry is [`Entry::NotFound`].
670    pub fn expect_not_found(&self) {
671        match self {
672            Entry::NotFound => (),
673            _ => panic!("expected not found entry, got {self:?}"),
674        }
675    }
676
677    /// Returns the metadata and payload bytes, panicking if the entry is not [`Entry::Object`].
678    pub fn expect_object(&self) -> (Metadata, Bytes) {
679        match self {
680            Entry::Object(metadata, bytes) => (metadata.clone(), bytes.clone()),
681            _ => panic!("expected object entry, got {self:?}"),
682        }
683    }
684
685    /// Returns the tombstone, panicking if the entry is not [`Entry::Tombstone`].
686    pub fn expect_tombstone(&self) -> Tombstone {
687        match self {
688            Entry::Tombstone(tombstone) => tombstone.clone(),
689            _ => panic!("expected tombstone entry, got {self:?}"),
690        }
691    }
692}
693
694#[cfg(test)]
695mod tests {
696    use std::num::NonZeroU32;
697    use std::time::Duration;
698
699    use objectstore_types::metadata::ExpirationPolicy;
700    use objectstore_types::scope::{Scope, Scopes};
701
702    #[cfg(feature = "storage-cogs")]
703    use objectstore_inventory_tracker::OpType;
704    #[cfg(feature = "storage-cogs")]
705    use objectstore_inventory_tracker::test_utils::DummyProducer;
706
707    use super::*;
708    use crate::backend::common::Backend;
709    use crate::id::ObjectContext;
710    use crate::stream;
711
712    fn make_id() -> ObjectId {
713        ObjectId::random(ObjectContext {
714            usecase: "testing".into(),
715            scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
716        })
717    }
718
719    #[tokio::test]
720    async fn set_expiry() {
721        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
722        for policy in [
723            ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
724            ExpirationPolicy::TimeToIdle(Duration::from_hours(1)),
725        ] {
726            let backend = InMemoryBackend::new("test");
727            let id = make_id();
728            let original_expiry = access_time + Duration::from_hours(1);
729            let metadata = Metadata {
730                expiration_policy: policy,
731                time_expires: Some(original_expiry),
732                custom: [("preserved".into(), "yes".into())].into(),
733                ..Default::default()
734            };
735            backend
736                .put_object(&id, &metadata, stream::single("payload"), access_time)
737                .await
738                .unwrap();
739
740            let requested = original_expiry + Duration::from_hours(1) + Duration::from_nanos(999);
741            assert!(
742                backend
743                    .set_expiry(&id, requested, access_time)
744                    .await
745                    .unwrap()
746            );
747            let (updated, payload) = backend.get(&id).expect_object();
748            assert_eq!(updated.expiration_policy, policy);
749            assert_eq!(updated.custom, metadata.custom);
750            assert_eq!(payload, Bytes::from_static(b"payload"));
751            assert_eq!(updated.time_expires, Some(requested));
752
753            assert!(
754                backend
755                    .set_expiry(&id, original_expiry, access_time)
756                    .await
757                    .unwrap()
758            );
759            assert_eq!(
760                backend.get(&id).expect_object().0.time_expires,
761                updated.time_expires
762            );
763        }
764    }
765
766    #[tokio::test]
767    async fn set_expiry_rejected() {
768        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
769        let backend = InMemoryBackend::new("test");
770        let absent = make_id();
771        assert!(
772            !backend
773                .set_expiry(&absent, access_time + Duration::from_hours(1), access_time)
774                .await
775                .unwrap()
776        );
777
778        let manual = make_id();
779        backend
780            .put_object(
781                &manual,
782                &Metadata::default(),
783                stream::single("manual"),
784                access_time,
785            )
786            .await
787            .unwrap();
788        assert!(
789            !backend
790                .set_expiry(&manual, access_time + Duration::from_hours(1), access_time)
791                .await
792                .unwrap()
793        );
794
795        let expired = make_id();
796        let metadata = Metadata {
797            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
798            time_expires: Some(access_time - Duration::from_secs(1)),
799            ..Default::default()
800        };
801        backend
802            .put_object(&expired, &metadata, stream::single("expired"), access_time)
803            .await
804            .unwrap();
805        let deadline = metadata.time_expires.unwrap();
806        for time in [deadline - Duration::from_secs(1), deadline] {
807            assert!(
808                backend
809                    .get_object(&expired, time, None)
810                    .await
811                    .unwrap()
812                    .is_some()
813            );
814        }
815        assert!(
816            backend
817                .get_object(&expired, access_time, None)
818                .await
819                .unwrap()
820                .is_none()
821        );
822        assert!(
823            !backend
824                .set_expiry(&expired, access_time + Duration::from_hours(1), access_time)
825                .await
826                .unwrap()
827        );
828    }
829
830    #[tokio::test]
831    async fn redirect_expiry() {
832        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
833        let backend = InMemoryBackend::new("test");
834        let id = make_id();
835        let target = make_id();
836        let other = make_id();
837        let old_expiry = access_time + Duration::from_hours(1);
838        backend
839            .compare_and_write(
840                &id,
841                None,
842                TieredWrite::Tombstone(Tombstone {
843                    target: target.clone(),
844                    time_expires: Some(old_expiry),
845                }),
846                access_time,
847            )
848            .await
849            .unwrap();
850
851        assert!(
852            backend
853                .get_object(&id, access_time, None)
854                .await
855                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
856        );
857        assert!(
858            backend
859                .get_metadata(&id, access_time)
860                .await
861                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
862        );
863
864        let new_expiry = old_expiry + Duration::from_hours(1);
865        assert!(
866            !backend
867                .compare_and_update(
868                    &id,
869                    Some(&other),
870                    TieredUpdate::SetExpiry(new_expiry),
871                    access_time,
872                )
873                .await
874                .unwrap()
875        );
876        assert_eq!(
877            backend.get(&id).expect_tombstone().time_expires,
878            Some(old_expiry)
879        );
880        assert!(
881            backend
882                .compare_and_update(
883                    &id,
884                    Some(&target),
885                    TieredUpdate::SetExpiry(new_expiry),
886                    access_time
887                )
888                .await
889                .unwrap()
890        );
891        assert_eq!(
892            backend.get(&id).expect_tombstone().time_expires,
893            Some(new_expiry)
894        );
895    }
896
897    #[tokio::test]
898    async fn multipart_single_part() {
899        let backend = InMemoryBackend::new("test");
900        let id = make_id();
901        let metadata = Metadata {
902            content_type: "text/plain".into(),
903            expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_hours(1)),
904            origin: Some("203.0.113.42".into()),
905            custom: [("foo".into(), "bar".into())].into(),
906            ..Default::default()
907        };
908
909        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
910
911        let data = b"hello, multipart world!";
912        let etag = backend
913            .upload_part(
914                &id,
915                &upload_id,
916                NonZeroU32::new(1).unwrap(),
917                data.len() as u64,
918                None,
919                stream::single(data.to_vec()),
920            )
921            .await
922            .unwrap();
923
924        let result = backend
925            .complete_multipart(
926                &id,
927                &upload_id,
928                vec![CompletedPart {
929                    part_number: NonZeroU32::new(1).unwrap(),
930                    etag,
931                }],
932                Timestamp::now(),
933            )
934            .await
935            .unwrap();
936        assert!(result.is_none(), "expected no error on complete");
937
938        let (meta, _, body) = backend
939            .get_object(&id, Timestamp::now(), None)
940            .await
941            .unwrap()
942            .unwrap();
943        let payload = stream::read_to_vec(body).await.unwrap();
944        assert_eq!(payload, data);
945        assert_eq!(meta.content_type, "text/plain".to_string());
946        assert_eq!(
947            meta.expiration_policy,
948            ExpirationPolicy::TimeToIdle(Duration::from_hours(1))
949        );
950        assert_eq!(meta.origin, Some("203.0.113.42".into()));
951        assert_eq!(meta.custom, [("foo".into(), "bar".into())].into());
952    }
953
954    #[tokio::test]
955    async fn multipart_multiple_parts() {
956        let backend = InMemoryBackend::new("test");
957        let id = make_id();
958        let metadata = Metadata::default();
959
960        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
961
962        let part1 = b"aaaa".to_vec();
963        let part2 = b"bbbb".to_vec();
964        let part3 = b"cc".to_vec();
965
966        let etag1 = backend
967            .upload_part(
968                &id,
969                &upload_id,
970                NonZeroU32::new(1).unwrap(),
971                part1.len() as u64,
972                None,
973                stream::single(part1.clone()),
974            )
975            .await
976            .unwrap();
977        let etag2 = backend
978            .upload_part(
979                &id,
980                &upload_id,
981                NonZeroU32::new(2).unwrap(),
982                part2.len() as u64,
983                None,
984                stream::single(part2.clone()),
985            )
986            .await
987            .unwrap();
988        let etag3 = backend
989            .upload_part(
990                &id,
991                &upload_id,
992                NonZeroU32::new(3).unwrap(),
993                part3.len() as u64,
994                None,
995                stream::single(part3.clone()),
996            )
997            .await
998            .unwrap();
999
1000        let result = backend
1001            .complete_multipart(
1002                &id,
1003                &upload_id,
1004                vec![
1005                    CompletedPart {
1006                        part_number: NonZeroU32::new(1).unwrap(),
1007                        etag: etag1,
1008                    },
1009                    CompletedPart {
1010                        part_number: NonZeroU32::new(2).unwrap(),
1011                        etag: etag2,
1012                    },
1013                    CompletedPart {
1014                        part_number: NonZeroU32::new(3).unwrap(),
1015                        etag: etag3,
1016                    },
1017                ],
1018                Timestamp::now(),
1019            )
1020            .await
1021            .unwrap();
1022        assert!(result.is_none());
1023
1024        let (_, _, body) = backend
1025            .get_object(&id, Timestamp::now(), None)
1026            .await
1027            .unwrap()
1028            .unwrap();
1029        let payload = stream::read_to_vec(body).await.unwrap();
1030        assert_eq!(payload, b"aaaabbbbcc");
1031    }
1032
1033    #[tokio::test]
1034    async fn multipart_list_parts() {
1035        let backend = InMemoryBackend::new("test");
1036        let id = make_id();
1037        let metadata = Metadata::default();
1038
1039        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1040
1041        let etag1 = backend
1042            .upload_part(
1043                &id,
1044                &upload_id,
1045                NonZeroU32::new(1).unwrap(),
1046                3,
1047                None,
1048                stream::single(b"aaa".to_vec()),
1049            )
1050            .await
1051            .unwrap();
1052        let etag2 = backend
1053            .upload_part(
1054                &id,
1055                &upload_id,
1056                NonZeroU32::new(2).unwrap(),
1057                3,
1058                None,
1059                stream::single(b"bbb".to_vec()),
1060            )
1061            .await
1062            .unwrap();
1063
1064        let list = backend
1065            .list_parts(&id, &upload_id, None, None)
1066            .await
1067            .unwrap();
1068        assert_eq!(list.parts.len(), 2);
1069        assert_eq!(list.parts[0].part_number.get(), 1);
1070        assert_eq!(list.parts[0].etag, etag1);
1071        assert_eq!(list.parts[0].size, 3);
1072        assert_eq!(list.parts[1].part_number.get(), 2);
1073        assert_eq!(list.parts[1].etag, etag2);
1074        assert_eq!(list.parts[1].size, 3);
1075
1076        // Pagination
1077        let page1 = backend
1078            .list_parts(&id, &upload_id, Some(1), None)
1079            .await
1080            .unwrap();
1081        assert_eq!(page1.parts.len(), 1);
1082        assert_eq!(page1.parts[0].part_number.get(), 1);
1083        assert!(page1.is_truncated);
1084        assert!(page1.next_part_number_marker.is_some());
1085
1086        let page2 = backend
1087            .list_parts(&id, &upload_id, Some(1), page1.next_part_number_marker)
1088            .await
1089            .unwrap();
1090        assert_eq!(page2.parts.len(), 1);
1091        assert_eq!(page2.parts[0].part_number.get(), 2);
1092
1093        backend.abort_multipart(&id, &upload_id).await.unwrap();
1094    }
1095
1096    #[tokio::test]
1097    async fn multipart_abort() {
1098        let backend = InMemoryBackend::new("test");
1099        let id = make_id();
1100        let metadata = Metadata::default();
1101
1102        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1103
1104        backend
1105            .upload_part(
1106                &id,
1107                &upload_id,
1108                NonZeroU32::new(1).unwrap(),
1109                5,
1110                None,
1111                stream::single(b"hello".to_vec()),
1112            )
1113            .await
1114            .unwrap();
1115
1116        backend.abort_multipart(&id, &upload_id).await.unwrap();
1117
1118        let result = backend
1119            .get_object(&id, Timestamp::now(), None)
1120            .await
1121            .unwrap();
1122        assert!(result.is_none(), "object should not exist after abort");
1123    }
1124
1125    #[tokio::test]
1126    async fn multipart_invalid_etag() {
1127        let backend = InMemoryBackend::new("test");
1128        let id = make_id();
1129        let metadata = Metadata::default();
1130
1131        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1132
1133        let etag = backend
1134            .upload_part(
1135                &id,
1136                &upload_id,
1137                NonZeroU32::new(1).unwrap(),
1138                5,
1139                None,
1140                stream::single(b"hello".to_vec()),
1141            )
1142            .await
1143            .unwrap();
1144
1145        let result = backend
1146            .complete_multipart(
1147                &id,
1148                &upload_id,
1149                vec![CompletedPart {
1150                    part_number: NonZeroU32::new(1).unwrap(),
1151                    etag: "wrong-etag".into(),
1152                }],
1153                Timestamp::now(),
1154            )
1155            .await
1156            .unwrap();
1157        assert!(result.is_some(), "expected error for bad etag");
1158        assert_eq!(result.unwrap().code, "InvalidPart");
1159
1160        // Upload must survive a failed complete so the client can retry.
1161        let result = backend
1162            .complete_multipart(
1163                &id,
1164                &upload_id,
1165                vec![CompletedPart {
1166                    part_number: NonZeroU32::new(1).unwrap(),
1167                    etag,
1168                }],
1169                Timestamp::now(),
1170            )
1171            .await
1172            .unwrap();
1173        assert!(result.is_none(), "retry with correct etag should succeed");
1174    }
1175
1176    #[tokio::test]
1177    async fn multipart_missing_part() {
1178        let backend = InMemoryBackend::new("test");
1179        let id = make_id();
1180        let metadata = Metadata::default();
1181
1182        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1183
1184        let etag = backend
1185            .upload_part(
1186                &id,
1187                &upload_id,
1188                NonZeroU32::new(1).unwrap(),
1189                5,
1190                None,
1191                stream::single(b"hello".to_vec()),
1192            )
1193            .await
1194            .unwrap();
1195
1196        let result = backend
1197            .complete_multipart(
1198                &id,
1199                &upload_id,
1200                vec![CompletedPart {
1201                    part_number: NonZeroU32::new(99).unwrap(),
1202                    etag: "whatever".into(),
1203                }],
1204                Timestamp::now(),
1205            )
1206            .await
1207            .unwrap();
1208        assert!(result.is_some(), "expected error for missing part");
1209        assert_eq!(result.unwrap().code, "InvalidPart");
1210
1211        // Upload must survive a failed complete so the client can retry.
1212        let result = backend
1213            .complete_multipart(
1214                &id,
1215                &upload_id,
1216                vec![CompletedPart {
1217                    part_number: NonZeroU32::new(1).unwrap(),
1218                    etag,
1219                }],
1220                Timestamp::now(),
1221            )
1222            .await
1223            .unwrap();
1224        assert!(result.is_none(), "retry with correct part should succeed");
1225    }
1226
1227    #[cfg(feature = "storage-cogs")]
1228    fn backend_with_change_stream() -> (InMemoryBackend, DummyProducer) {
1229        use crate::change_stream::CostTrackerStreamConfig;
1230
1231        let (streams, producer) = crate::change_stream::dummy_factory();
1232        let change_stream = streams.build(Some(&CostTrackerStreamConfig {
1233            shared_resource_id: "in_memory_objectstore".into(),
1234            sample_rate: 1.0,
1235        }));
1236        (
1237            InMemoryBackend::new("test").with_change_stream(change_stream),
1238            producer,
1239        )
1240    }
1241
1242    #[cfg(feature = "storage-cogs")]
1243    #[tokio::test]
1244    async fn change_stream_reports_writes_and_deletes() {
1245        let (backend, producer) = backend_with_change_stream();
1246        let id = make_id();
1247        let metadata = Metadata::default();
1248        let payload = b"hello";
1249
1250        backend
1251            .put_object(
1252                &id,
1253                &metadata,
1254                stream::single(payload.to_vec()),
1255                Timestamp::now(),
1256            )
1257            .await
1258            .unwrap();
1259        backend.delete_object(&id, Timestamp::now()).await.unwrap();
1260        // The object is already gone, so this reports nothing.
1261        backend.delete_object(&id, Timestamp::now()).await.unwrap();
1262
1263        let records = producer.records();
1264        assert_eq!(records.len(), 2);
1265        assert_eq!(records[0].op_type, OpType::Write);
1266        assert_eq!(records[0].app_feature, "testing");
1267        assert_eq!(
1268            records[0].size,
1269            Some((json_len(&metadata) + payload.len()) as u64),
1270            "the reported size covers metadata as well as the payload"
1271        );
1272        assert!(json_len(&metadata) > 0, "metadata must contribute bytes");
1273        assert_eq!(records[1].op_type, OpType::Delete);
1274    }
1275
1276    #[cfg(feature = "storage-cogs")]
1277    #[tokio::test]
1278    async fn change_stream_reports_expiry_extension_as_update() {
1279        let (backend, producer) = backend_with_change_stream();
1280        let id = make_id();
1281        let access_time = Timestamp::now();
1282        let expires = access_time + Duration::from_secs(3600);
1283        let metadata = Metadata {
1284            time_expires: Some(expires),
1285            ..Default::default()
1286        };
1287
1288        backend
1289            .put_object(
1290                &id,
1291                &metadata,
1292                stream::single(b"hello".to_vec()),
1293                access_time,
1294            )
1295            .await
1296            .unwrap();
1297        producer.clear();
1298
1299        let extended = expires + Duration::from_secs(3600);
1300        assert!(
1301            backend
1302                .set_expiry(&id, extended, access_time)
1303                .await
1304                .unwrap()
1305        );
1306
1307        let records = producer.records();
1308        assert_eq!(records.len(), 1);
1309        assert_eq!(records[0].op_type, OpType::Update);
1310
1311        // A deadline that already covers the request writes nothing, so it reports nothing.
1312        producer.clear();
1313        assert!(
1314            backend
1315                .set_expiry(&id, expires + Duration::from_secs(60), access_time)
1316                .await
1317                .unwrap()
1318        );
1319        assert!(producer.records().is_empty());
1320    }
1321
1322    #[cfg(feature = "storage-cogs")]
1323    #[tokio::test]
1324    async fn change_stream_reports_tombstone_expiry_extension() {
1325        let (backend, producer) = backend_with_change_stream();
1326        let id = make_id();
1327        let target = make_id();
1328        let access_time = Timestamp::now();
1329        let expires = access_time + Duration::from_secs(3600);
1330
1331        backend
1332            .compare_and_write(
1333                &id,
1334                None,
1335                TieredWrite::Tombstone(Tombstone {
1336                    target: target.clone(),
1337                    time_expires: Some(expires),
1338                }),
1339                access_time,
1340            )
1341            .await
1342            .unwrap();
1343        producer.clear();
1344
1345        let extended = expires + Duration::from_secs(3600);
1346        assert!(
1347            backend
1348                .compare_and_update(
1349                    &id,
1350                    Some(&target),
1351                    TieredUpdate::SetExpiry(extended),
1352                    access_time,
1353                )
1354                .await
1355                .unwrap()
1356        );
1357
1358        let records = producer.records();
1359        assert_eq!(records.len(), 1);
1360        assert_eq!(records[0].op_type, OpType::Update);
1361    }
1362}
objectstore_service/backend/in_memory.rs

objectstore_service/backend/
in_memory.rs
