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::num::NonZeroU64;
10use std::sync::{Arc, Mutex};
11use std::time::SystemTime;
12
13use objectstore_types::range::ByteRange;
14use objectstore_types::time::Timestamp;
15
16use bytes::{Bytes, BytesMut};
17use futures_util::TryStreamExt;
18use objectstore_types::metadata::Metadata;
19
20use crate::backend::common::{
21    self, DeleteResponse, ExpiryUpdate, GetResponse, HighVolumeBackend, MultipartUploadBackend,
22    PutResponse, SetExpiryResponse, TieredGet, TieredMetadata, TieredUpdate, TieredWrite,
23    Tombstone,
24};
25use crate::change_stream::{ChangeStream, NoopStream, flush_change_stream};
26use crate::error::{Error, ErrorKind, Result};
27use crate::id::ObjectId;
28use crate::multipart::{
29    AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
30    ListPartsResponse, Part, PartNumber, UploadId, UploadPartResponse,
31};
32use crate::resumable::{BackendToken, Session, UploadProgress};
33use crate::stream::ClientStream;
34
35/// An entry in the in-memory store.
36#[derive(Clone, Debug)]
37enum StoreEntry {
38    Object(Metadata, Bytes),
39    Tombstone(Tombstone),
40}
41
42impl StoreEntry {
43    fn is_expired(&self, now: Timestamp) -> bool {
44        match self {
45            StoreEntry::Object(metadata, _) => metadata.is_expired(now),
46            StoreEntry::Tombstone(tombstone) => tombstone.is_expired(now),
47        }
48    }
49
50    /// Number of bytes an entry occupies (payload plus serialized metadata)
51    pub(crate) fn stored_size(&self) -> usize {
52        match self {
53            StoreEntry::Object(metadata, payload) => json_len(metadata) + payload.len(),
54            // A tombstone carries no payload: its bytes are the redirect target plus the
55            // deadline stored alongside it.
56            StoreEntry::Tombstone(tombstone) => {
57                tombstone.target.as_storage_path().to_string().len()
58                    + tombstone.time_expires.map_or(0, |d| json_len(&d))
59            }
60        }
61    }
62}
63
64type Store = HashMap<ObjectId, StoreEntry>;
65
66#[derive(Clone, Debug)]
67struct MultipartUpload {
68    metadata: Metadata,
69    parts: BTreeMap<PartNumber, UploadedPart>,
70}
71
72#[derive(Clone, Debug)]
73struct UploadedPart {
74    etag: String,
75    data: Bytes,
76    uploaded_at: SystemTime,
77}
78
79#[derive(Debug)]
80struct ResumableUpload {
81    metadata: Metadata,
82    data: BytesMut,
83}
84
85// None marks a removed session for operations already waiting on its mutex.
86type ResumableSession = Arc<tokio::sync::Mutex<Option<ResumableUpload>>>;
87type ResumableStore = HashMap<(ObjectId, BackendToken), ResumableSession>;
88
89type MultipartStore = HashMap<(ObjectId, UploadId), MultipartUpload>;
90
91/// In-memory [`Backend`](super::common::Backend) backed by a `HashMap`.
92///
93/// Removes the need for filesystem tempdir management in unit tests. The
94/// backend is [`Clone`] so tests can hold a handle for direct inspection while
95/// the service owns a boxed copy.
96#[derive(Debug, Clone)]
97pub struct InMemoryBackend {
98    name: &'static str,
99    store: Arc<Mutex<Store>>,
100    multipart_store: Arc<Mutex<MultipartStore>>,
101    resumable_store: Arc<Mutex<ResumableStore>>,
102    upload_markers: Arc<Mutex<HashMap<ObjectId, Timestamp>>>,
103    change_stream: Arc<dyn ChangeStream>,
104}
105
106impl InMemoryBackend {
107    /// Creates a new `InMemoryBackend` with the given diagnostic `name`.
108    pub fn new(name: &'static str) -> Self {
109        Self {
110            name,
111            store: Arc::new(Mutex::new(HashMap::new())),
112            multipart_store: Arc::new(Mutex::new(HashMap::new())),
113            resumable_store: Arc::new(Mutex::new(HashMap::new())),
114            upload_markers: Arc::new(Mutex::new(HashMap::new())),
115            change_stream: Arc::new(NoopStream),
116        }
117    }
118
119    fn upload_session(&self, session: &Session) -> Result<ResumableSession> {
120        self.resumable_store
121            .lock()
122            .unwrap()
123            .get(&(session.object_id.clone(), session.backend_token.clone()))
124            // Clone the session handle so that the `resumable_store` mutex is released immediately.
125            .cloned()
126            .ok_or_else(|| ErrorKind::UnknownUploadSession.into())
127    }
128
129    /// Publishes this backend's changes to `change_stream`.
130    pub fn with_change_stream(mut self, change_stream: Arc<dyn ChangeStream>) -> Self {
131        self.change_stream = change_stream;
132        self
133    }
134
135    /// Returns the stored entry for `id`, for direct inspection in tests.
136    pub fn get(&self, id: &ObjectId) -> Entry {
137        match self.store.lock().unwrap().get(id).cloned() {
138            None => Entry::NotFound,
139            Some(StoreEntry::Tombstone(tombstone)) => Entry::Tombstone(tombstone),
140            Some(StoreEntry::Object(metadata, bytes)) => Entry::Object(metadata, bytes),
141        }
142    }
143
144    /// Returns `true` if the backend contains an entry for the given id.
145    pub fn contains(&self, id: &ObjectId) -> bool {
146        self.store.lock().unwrap().contains_key(id)
147    }
148
149    /// Returns `true` if the backend has no stored objects.
150    pub fn is_empty(&self) -> bool {
151        self.store.lock().unwrap().is_empty()
152    }
153
154    /// Removes an entry directly, bypassing the `Backend` trait.
155    ///
156    /// Useful for simulating partial failures (e.g. orphan tombstones).
157    pub fn remove(&self, id: &ObjectId) {
158        self.store.lock().unwrap().remove(id);
159    }
160}
161
162#[async_trait::async_trait]
163impl super::common::Backend for InMemoryBackend {
164    fn name(&self) -> &'static str {
165        self.name
166    }
167
168    fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> {
169        Ok(self)
170    }
171
172    async fn put_object(
173        &self,
174        id: &ObjectId,
175        metadata: &Metadata,
176        stream: ClientStream,
177        _access_time: Timestamp,
178    ) -> Result<PutResponse> {
179        let bytes: BytesMut = stream.try_collect().await?;
180        let entry = StoreEntry::Object(metadata.clone(), bytes.freeze());
181        let size = entry.stored_size();
182        self.store.lock().unwrap().insert(id.clone(), entry);
183        self.change_stream
184            .write(id, size as u64, metadata.time_expires);
185        Ok(())
186    }
187
188    async fn get_object(
189        &self,
190        id: &ObjectId,
191        access_time: Timestamp,
192        range: Option<ByteRange>,
193    ) -> Result<GetResponse> {
194        let entry = self.store.lock().unwrap().get(id).cloned();
195        match entry {
196            None => Ok(None),
197            Some(entry) if entry.is_expired(access_time) => Ok(None),
198            Some(StoreEntry::Tombstone(_)) => Err(ErrorKind::UnexpectedTombstone.into()),
199            Some(StoreEntry::Object(mut metadata, bytes)) => {
200                let total = bytes.len() as u64;
201                metadata.size = Some(bytes.len());
202                let (content_range, payload) = match range {
203                    Some(range) => {
204                        let content_range = range
205                            .resolve(total)
206                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
207                        let sliced =
208                            bytes.slice(content_range.start as usize..=content_range.end as usize);
209                        (Some(content_range), sliced)
210                    }
211                    None => (None, bytes),
212                };
213                Ok(Some((
214                    metadata,
215                    content_range,
216                    crate::stream::single(payload),
217                )))
218            }
219        }
220    }
221
222    async fn set_expiry(
223        &self,
224        id: &ObjectId,
225        target: ExpiryUpdate,
226        access_time: Timestamp,
227    ) -> Result<SetExpiryResponse> {
228        let outcome = {
229            let mut store = self.store.lock().unwrap();
230            match store.get_mut(id) {
231                None => ExpiryOutcome::NotFound,
232                Some(entry) if entry.is_expired(access_time) => ExpiryOutcome::NotFound,
233                Some(StoreEntry::Object(metadata, _)) => {
234                    extend_object_expiry(metadata, target, access_time)?
235                }
236                _ => ExpiryOutcome::Rejected,
237            }
238        };
239
240        if let ExpiryOutcome::Extended(expire_at) = outcome {
241            self.change_stream.update(id, Some(expire_at));
242        }
243
244        Ok(outcome.response())
245    }
246
247    async fn delete_object(
248        &self,
249        id: &ObjectId,
250        _access_time: Timestamp,
251    ) -> Result<DeleteResponse> {
252        if self.store.lock().unwrap().remove(id).is_some() {
253            self.change_stream.delete(id);
254        }
255        Ok(())
256    }
257
258    async fn create_upload_session(
259        &self,
260        id: &ObjectId,
261        metadata: &Metadata,
262        _upload_length: NonZeroU64,
263    ) -> Result<Option<BackendToken>> {
264        let token = uuid::Uuid::now_v7().to_string();
265        let upload = ResumableUpload {
266            metadata: metadata.clone(),
267            data: BytesMut::new(),
268        };
269        self.resumable_store.lock().unwrap().insert(
270            (id.clone(), token.clone()),
271            Arc::new(tokio::sync::Mutex::new(Some(upload))),
272        );
273        Ok(Some(token))
274    }
275
276    async fn put_chunk(
277        &self,
278        session: &Session,
279        offset: u64,
280        content_length: u64,
281        mut stream: ClientStream,
282    ) -> Result<UploadProgress> {
283        let upload_session = self.upload_session(session)?;
284        let mut guard = upload_session.lock().await;
285        let upload = guard.as_mut().ok_or(ErrorKind::UnknownUploadSession)?;
286        offset
287            .checked_add(content_length)
288            .filter(|end| *end <= session.upload_length.get())
289            .ok_or(ErrorKind::ChunkExceedsUploadLength {
290                offset,
291                content_length,
292                upload_length: session.upload_length.get(),
293            })?;
294        if content_length != 0 && offset != upload.data.len() as u64 {
295            return Err(ErrorKind::UploadOffsetMismatch {
296                offset: upload.data.len() as u64,
297            }
298            .into());
299        }
300
301        let mut remaining = content_length;
302        while remaining > 0 {
303            let Some(chunk) = stream.try_next().await? else {
304                break;
305            };
306            let count = remaining.min(chunk.len() as u64) as usize;
307            upload.data.extend_from_slice(&chunk[..count]);
308            remaining -= count as u64;
309        }
310
311        let offset = upload.data.len() as u64;
312        if offset != session.upload_length.get() {
313            return Ok(UploadProgress::Incomplete { offset });
314        }
315
316        let upload = guard.take().unwrap();
317        let metadata = upload.metadata;
318        let expires_at = metadata.time_expires;
319        let entry = StoreEntry::Object(metadata, upload.data.freeze());
320        let size = entry.stored_size();
321        self.store
322            .lock()
323            .unwrap()
324            .insert(session.object_id.clone(), entry);
325
326        self.change_stream
327            .write(&session.object_id, size as u64, expires_at);
328        self.resumable_store
329            .lock()
330            .unwrap()
331            .remove(&(session.object_id.clone(), session.backend_token.clone()));
332
333        Ok(UploadProgress::Complete)
334    }
335
336    async fn upload_offset(&self, session: &Session) -> Result<UploadProgress> {
337        let upload_session = self.upload_session(session)?;
338        let guard = upload_session.lock().await;
339        let upload = guard.as_ref().ok_or(ErrorKind::UnknownUploadSession)?;
340        Ok(UploadProgress::Incomplete {
341            offset: upload.data.len() as u64,
342        })
343    }
344
345    async fn cancel_upload(&self, session: &Session) -> Result<()> {
346        let upload_session = self.upload_session(session)?;
347        let mut guard = upload_session.lock().await;
348        guard.take().ok_or(ErrorKind::UnknownUploadSession)?;
349        self.resumable_store
350            .lock()
351            .unwrap()
352            .remove(&(session.object_id.clone(), session.backend_token.clone()));
353        Ok(())
354    }
355
356    async fn join(&self) {
357        flush_change_stream(&self.change_stream).await;
358    }
359}
360
361#[async_trait::async_trait]
362impl HighVolumeBackend for InMemoryBackend {
363    async fn create_upload_marker(
364        &self,
365        revision: &ObjectId,
366        time_expires: Timestamp,
367    ) -> Result<()> {
368        self.upload_markers
369            .lock()
370            .unwrap()
371            .insert(revision.clone(), time_expires);
372        Ok(())
373    }
374
375    async fn has_upload_marker(&self, revision: &ObjectId, access_time: Timestamp) -> Result<bool> {
376        Ok(self
377            .upload_markers
378            .lock()
379            .unwrap()
380            .get(revision)
381            .is_some_and(|expiry| *expiry >= access_time))
382    }
383
384    async fn delete_upload_marker(
385        &self,
386        revision: &ObjectId,
387        access_time: Timestamp,
388    ) -> Result<bool> {
389        Ok(self
390            .upload_markers
391            .lock()
392            .unwrap()
393            .remove(revision)
394            .is_some_and(|expiry| expiry >= access_time))
395    }
396
397    async fn put_non_tombstone(
398        &self,
399        id: &ObjectId,
400        metadata: &Metadata,
401        payload: Bytes,
402        access_time: Timestamp,
403    ) -> Result<Option<Tombstone>> {
404        let mut store = self.store.lock().unwrap();
405        if let Some(StoreEntry::Tombstone(tombstone)) = store.get(id)
406            && !tombstone.is_expired(access_time)
407        {
408            return Ok(Some(tombstone.clone()));
409        }
410
411        let mut metadata = metadata.clone();
412        metadata.size = Some(payload.len());
413        let expires_at = metadata.time_expires;
414        let entry = StoreEntry::Object(metadata, payload);
415        let size = entry.stored_size();
416        store.insert(id.clone(), entry);
417        self.change_stream.write(id, size as u64, expires_at);
418        Ok(None)
419    }
420
421    async fn get_tiered_object(
422        &self,
423        id: &ObjectId,
424        access_time: Timestamp,
425        range: Option<ByteRange>,
426    ) -> Result<TieredGet> {
427        let entry = self.store.lock().unwrap().get(id).cloned();
428        Ok(match entry {
429            None => TieredGet::NotFound,
430            Some(entry) if entry.is_expired(access_time) => TieredGet::NotFound,
431            Some(StoreEntry::Tombstone(tombstone)) => TieredGet::Tombstone(tombstone),
432            Some(StoreEntry::Object(mut metadata, bytes)) => {
433                let total = bytes.len() as u64;
434                metadata.size = Some(bytes.len());
435                let (content_range, payload) = match range {
436                    Some(range) => {
437                        let content_range = range
438                            .resolve(total)
439                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
440                        let sliced =
441                            bytes.slice(content_range.start as usize..=content_range.end as usize);
442                        (Some(content_range), sliced)
443                    }
444                    None => (None, bytes),
445                };
446                TieredGet::Object(metadata, content_range, crate::stream::single(payload))
447            }
448        })
449    }
450
451    async fn get_tiered_metadata(
452        &self,
453        id: &ObjectId,
454        access_time: Timestamp,
455    ) -> Result<TieredMetadata> {
456        let entry = self.store.lock().unwrap().get(id).cloned();
457        Ok(match entry {
458            None => TieredMetadata::NotFound,
459            Some(entry) if entry.is_expired(access_time) => TieredMetadata::NotFound,
460            Some(StoreEntry::Tombstone(tombstone)) => TieredMetadata::Tombstone(tombstone),
461            Some(StoreEntry::Object(metadata, _bytes)) => TieredMetadata::Object(metadata),
462        })
463    }
464
465    async fn delete_non_tombstone(
466        &self,
467        id: &ObjectId,
468        access_time: Timestamp,
469    ) -> Result<Option<Tombstone>> {
470        let mut store = self.store.lock().unwrap();
471        if let Some(StoreEntry::Tombstone(tombstone)) = store.get(id).cloned()
472            && !tombstone.is_expired(access_time)
473        {
474            return Ok(Some(tombstone));
475        }
476
477        if store.remove(id).is_some() {
478            self.change_stream.delete(id);
479        }
480        Ok(None)
481    }
482
483    async fn compare_and_update(
484        &self,
485        id: &ObjectId,
486        current: Option<&ObjectId>,
487        update: TieredUpdate,
488        access_time: Timestamp,
489    ) -> Result<SetExpiryResponse> {
490        let TieredUpdate::SetExpiry(expiry_target) = update;
491        let outcome = {
492            let mut store = self.store.lock().unwrap();
493            match (store.get_mut(id), current) {
494                (None, _) => ExpiryOutcome::NotFound,
495                (Some(entry), _) if entry.is_expired(access_time) => ExpiryOutcome::NotFound,
496                (Some(StoreEntry::Object(metadata, _)), None) => {
497                    extend_object_expiry(metadata, expiry_target, access_time)?
498                }
499                (Some(StoreEntry::Tombstone(t)), Some(target)) if t.target == *target => {
500                    extend_expiry(&mut t.time_expires, expiry_target, None, access_time)?
501                }
502                _ => ExpiryOutcome::Rejected,
503            }
504        };
505
506        if let ExpiryOutcome::Extended(expire_at) = outcome {
507            self.change_stream.update(id, Some(expire_at));
508        }
509
510        Ok(outcome.response())
511    }
512
513    async fn compare_and_write(
514        &self,
515        id: &ObjectId,
516        current: Option<&ObjectId>,
517        write: TieredWrite,
518        access_time: Timestamp,
519    ) -> Result<bool> {
520        let mut store = self.store.lock().unwrap();
521
522        let actual = store.get(id);
523        let matches_current = matches_redirect(actual, current, access_time);
524        let matches_next = matches_redirect(actual, write.target(), access_time);
525
526        if matches_current {
527            match write {
528                TieredWrite::Tombstone(tombstone) => {
529                    let expires_at = tombstone.time_expires;
530                    let entry = StoreEntry::Tombstone(tombstone);
531                    let size = entry.stored_size();
532                    store.insert(id.clone(), entry);
533                    self.change_stream.write(id, size as u64, expires_at);
534                }
535                TieredWrite::Object(metadata, payload) => {
536                    let expires_at = metadata.time_expires;
537                    let entry = StoreEntry::Object(metadata, payload);
538                    let size = entry.stored_size();
539                    store.insert(id.clone(), entry);
540                    self.change_stream.write(id, size as u64, expires_at);
541                }
542                TieredWrite::Delete => {
543                    if store.remove(id).is_some() {
544                        self.change_stream.delete(id);
545                    }
546                }
547            }
548        }
549
550        Ok(matches_current || matches_next)
551    }
552}
553
554#[async_trait::async_trait]
555impl MultipartUploadBackend for InMemoryBackend {
556    async fn initiate_multipart(
557        &self,
558        id: &ObjectId,
559        metadata: &Metadata,
560    ) -> Result<InitiateMultipartResponse> {
561        let upload_id = UploadId::new(uuid::Uuid::now_v7().to_string())?;
562        let upload = MultipartUpload {
563            metadata: metadata.clone(),
564            parts: BTreeMap::new(),
565        };
566        self.multipart_store
567            .lock()
568            .unwrap()
569            .insert((id.clone(), upload_id.clone()), upload);
570        Ok(upload_id)
571    }
572
573    async fn upload_part(
574        &self,
575        id: &ObjectId,
576        upload_id: &UploadId,
577        part_number: PartNumber,
578        _content_length: u64,
579        _content_md5: Option<&str>,
580        body: ClientStream,
581    ) -> Result<UploadPartResponse> {
582        let data: BytesMut = body.try_collect().await?;
583        let data = data.freeze();
584        let etag = format!("\"etag-{part_number}-{}\"", data.len());
585
586        let mut store = self.multipart_store.lock().unwrap();
587        let upload = store
588            .get_mut(&(id.clone(), upload_id.clone()))
589            .ok_or_else(|| {
590                Error::new(
591                    ErrorKind::BackendFailure,
592                    "in-memory multipart upload not found",
593                )
594            })?;
595
596        upload.parts.insert(
597            part_number,
598            UploadedPart {
599                etag: etag.clone(),
600                data,
601                uploaded_at: SystemTime::now(),
602            },
603        );
604
605        Ok(etag)
606    }
607
608    async fn list_parts(
609        &self,
610        id: &ObjectId,
611        upload_id: &UploadId,
612        max_parts: Option<u32>,
613        part_number_marker: Option<PartNumber>,
614    ) -> Result<ListPartsResponse> {
615        let store = self.multipart_store.lock().unwrap();
616        let upload = store.get(&(id.clone(), upload_id.clone())).ok_or_else(|| {
617            Error::new(
618                ErrorKind::BackendFailure,
619                "in-memory multipart upload not found",
620            )
621        })?;
622
623        let iter = upload
624            .parts
625            .iter()
626            .filter(|(pn, _)| part_number_marker.is_none_or(|marker| **pn > marker));
627
628        let max = max_parts.unwrap_or(u32::MAX) as usize;
629        let all: Vec<_> = iter.collect();
630        let is_truncated = all.len() > max;
631        let page: Vec<_> = all.into_iter().take(max).collect();
632
633        let next_part_number_marker = if is_truncated {
634            page.last().map(|(pn, _)| **pn)
635        } else {
636            None
637        };
638
639        let parts = page
640            .into_iter()
641            .map(|(pn, part)| Part {
642                part_number: *pn,
643                etag: part.etag.clone(),
644                last_modified: part.uploaded_at,
645                size: part.data.len() as u64,
646            })
647            .collect();
648
649        Ok(ListPartsResponse {
650            parts,
651            is_truncated,
652            next_part_number_marker,
653        })
654    }
655
656    async fn abort_multipart(
657        &self,
658        id: &ObjectId,
659        upload_id: &UploadId,
660    ) -> Result<AbortMultipartResponse> {
661        self.multipart_store
662            .lock()
663            .unwrap()
664            .remove(&(id.clone(), upload_id.clone()));
665        Ok(())
666    }
667
668    async fn complete_multipart(
669        &self,
670        id: &ObjectId,
671        upload_id: &UploadId,
672        parts: Vec<CompletedPart>,
673        _access_time: Timestamp,
674    ) -> Result<CompleteMultipartResponse> {
675        let key = (id.clone(), upload_id.clone());
676
677        // TODO: validate that parts are in ascending part_number order and reject with
678        // InvalidPartOrder if not (matches S3/GCS behavior). Needs a proper client error variant.
679
680        // Validate and assemble while holding the multipart lock, but don't
681        // remove the upload yet — a failed validation must leave it intact so
682        // the client can retry.
683        let (metadata, payload) = {
684            let store = self.multipart_store.lock().unwrap();
685            let upload = store.get(&key).ok_or_else(|| {
686                Error::new(
687                    ErrorKind::BackendFailure,
688                    "in-memory multipart upload not found",
689                )
690            })?;
691
692            for completed in &parts {
693                match upload.parts.get(&completed.part_number) {
694                    None => {
695                        return Ok(Some(crate::multipart::CompleteMultipartError {
696                            code: "InvalidPart".into(),
697                            message: format!(
698                                "part number {} was not uploaded",
699                                completed.part_number
700                            ),
701                        }));
702                    }
703                    Some(stored) if stored.etag != completed.etag => {
704                        return Ok(Some(crate::multipart::CompleteMultipartError {
705                            code: "InvalidPart".into(),
706                            message: format!(
707                                "etag mismatch for part {}: expected {}, got {}",
708                                completed.part_number, stored.etag, completed.etag
709                            ),
710                        }));
711                    }
712                    _ => {}
713                }
714            }
715
716            let mut payload = BytesMut::new();
717            for completed in &parts {
718                let stored = &upload.parts[&completed.part_number];
719                payload.extend_from_slice(&stored.data);
720            }
721
722            let mut metadata = upload.metadata.clone();
723            metadata.size = Some(payload.len());
724
725            (metadata, payload.freeze())
726        };
727
728        let expires_at = metadata.time_expires;
729        let entry = StoreEntry::Object(metadata, payload);
730        let size = entry.stored_size();
731        self.store.lock().unwrap().insert(id.clone(), entry);
732        self.change_stream.write(id, size as u64, expires_at);
733
734        self.multipart_store.lock().unwrap().remove(&key);
735
736        Ok(None)
737    }
738}
739
740/// Serialized length of `value`, or `0` if it cannot be serialized.
741fn json_len<T: serde::Serialize>(value: &T) -> usize {
742    serde_json::to_string(value).map_or(0, |json| json.len())
743}
744
745/// Returns `true` if `entry` matches the expected tombstone redirect state.
746///
747/// - `expected = None`: matches any non-tombstone (absent or inline object).
748/// - `expected = Some(target)`: matches a tombstone whose redirect target equals `target`.
749fn matches_redirect(
750    entry: Option<&StoreEntry>,
751    expected: Option<&ObjectId>,
752    now: Timestamp,
753) -> bool {
754    match entry {
755        None | Some(StoreEntry::Object(..)) => expected.is_none(),
756        Some(StoreEntry::Tombstone(tombstone)) => match expected {
757            None => tombstone.is_expired(now),
758            Some(target) => tombstone.target == *target && !tombstone.is_expired(now),
759        },
760    }
761}
762
763/// What [`extend_expiry`] did to an entry's deadline.
764///
765/// Callers need to tell a deadline that actually moved apart from one that already
766/// covered the request, because only the former is a change worth reporting to the
767/// [`ChangeStream`].
768#[derive(Clone, Copy, Debug, PartialEq, Eq)]
769enum ExpiryOutcome {
770    /// The entry is absent or expired.
771    NotFound,
772    /// The entry was not eligible or the target could not be resolved.
773    Rejected,
774    /// The stored deadline already covered `expire_at`. Nothing was written.
775    AlreadySatisfied(Timestamp),
776    /// The stored deadline moved out to `expire_at`.
777    Extended(Timestamp),
778}
779
780impl ExpiryOutcome {
781    /// Converts the internal mutation outcome to the backend response.
782    fn response(self) -> SetExpiryResponse {
783        match self {
784            Self::NotFound => SetExpiryResponse::NotFound,
785            Self::Rejected => SetExpiryResponse::Rejected,
786            Self::AlreadySatisfied(deadline) | Self::Extended(deadline) => {
787                SetExpiryResponse::Satisfied(deadline)
788            }
789        }
790    }
791}
792
793/// Extends an object's deadline and TTL together, leaving metadata intact on error.
794fn extend_object_expiry(
795    metadata: &mut Metadata,
796    target: ExpiryUpdate,
797    access_time: Timestamp,
798) -> Result<ExpiryOutcome> {
799    let Some(original_expires) = metadata.time_expires else {
800        return Ok(ExpiryOutcome::Rejected); // entry without a deadline
801    };
802
803    // Only write `updated_expires` back if `extended_expiration_policy` also succeeds
804    let mut updated_expires = Some(original_expires);
805    let outcome = extend_expiry(
806        &mut updated_expires,
807        target,
808        metadata.time_created,
809        access_time,
810    )?;
811
812    if let ExpiryOutcome::Extended(expire_at) = outcome {
813        let updated_policy = common::extended_expiration_policy(
814            metadata.expiration_policy,
815            metadata.time_created,
816            original_expires,
817            expire_at,
818        )?;
819
820        metadata.time_expires = updated_expires;
821        metadata.expiration_policy = updated_policy;
822    }
823
824    Ok(outcome)
825}
826
827/// Resolves `target` and extends an active expiry time where valid.
828///
829/// Eligibility is checked before target resolution.
830fn extend_expiry(
831    field: &mut Option<Timestamp>,
832    target: ExpiryUpdate,
833    time_created: Option<Timestamp>,
834    access_time: Timestamp,
835) -> Result<ExpiryOutcome> {
836    let Some(time_expires) = *field else {
837        return Ok(ExpiryOutcome::Rejected); // entries without a deadline cannot be extended
838    };
839
840    if time_expires < access_time {
841        return Ok(ExpiryOutcome::NotFound); // already expired
842    }
843
844    let Some(expire_at) = target.resolve(time_created, access_time)? else {
845        return Ok(ExpiryOutcome::Rejected);
846    };
847    if time_expires >= expire_at {
848        Ok(ExpiryOutcome::AlreadySatisfied(expire_at))
849    } else {
850        *field = Some(expire_at);
851        Ok(ExpiryOutcome::Extended(expire_at))
852    }
853}
854
855/// Type returned by [`InMemoryBackend::get`] for direct inspection of stored entries.
856#[derive(Clone, Debug)]
857pub enum Entry {
858    /// No entry exists at this key.
859    NotFound,
860    /// A real object with its metadata and payload bytes.
861    Object(Metadata, Bytes),
862    /// A redirect tombstone indicating the real object lives in the long-term backend.
863    Tombstone(Tombstone),
864}
865
866impl Entry {
867    /// Returns `true` if the entry is [`Entry::NotFound`].
868    pub fn is_not_found(&self) -> bool {
869        matches!(self, Entry::NotFound)
870    }
871
872    /// Returns `true` if the entry is [`Entry::Object`].
873    pub fn is_object(&self) -> bool {
874        matches!(self, Entry::Object(_, _))
875    }
876
877    /// Returns `true` if the entry is [`Entry::Tombstone`].
878    pub fn is_tombstone(&self) -> bool {
879        matches!(self, Entry::Tombstone(_))
880    }
881
882    /// Panics unless the entry is [`Entry::NotFound`].
883    pub fn expect_not_found(&self) {
884        match self {
885            Entry::NotFound => (),
886            _ => panic!("expected not found entry, got {self:?}"),
887        }
888    }
889
890    /// Returns the metadata and payload bytes, panicking if the entry is not [`Entry::Object`].
891    pub fn expect_object(&self) -> (Metadata, Bytes) {
892        match self {
893            Entry::Object(metadata, bytes) => (metadata.clone(), bytes.clone()),
894            _ => panic!("expected object entry, got {self:?}"),
895        }
896    }
897
898    /// Returns the tombstone, panicking if the entry is not [`Entry::Tombstone`].
899    pub fn expect_tombstone(&self) -> Tombstone {
900        match self {
901            Entry::Tombstone(tombstone) => tombstone.clone(),
902            _ => panic!("expected tombstone entry, got {self:?}"),
903        }
904    }
905}
906
907#[cfg(test)]
908mod tests {
909    use futures_util::StreamExt;
910    use std::num::NonZeroU32;
911    use std::time::Duration;
912
913    #[cfg(feature = "storage-cogs")]
914    use objectstore_inventory_tracker::{OpType, test_utils::DummyProducer};
915    use objectstore_types::metadata::ExpirationPolicy;
916    use objectstore_types::scope::{Scope, Scopes};
917
918    use super::*;
919    use crate::backend::common::{Backend, ExpiryTarget};
920    use crate::id::ObjectContext;
921    use crate::stream;
922
923    fn make_id() -> ObjectId {
924        ObjectId::random(ObjectContext {
925            usecase: "testing".into(),
926            scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
927        })
928    }
929
930    async fn create_session(backend: &InMemoryBackend, id: &ObjectId, length: u64) -> Session {
931        let backend_token = backend
932            .create_upload_session(id, &Metadata::default(), NonZeroU64::new(length).unwrap())
933            .await
934            .unwrap()
935            .unwrap();
936        Session {
937            object_id: id.clone(),
938            upload_length: NonZeroU64::new(length).unwrap(),
939            backend_token,
940        }
941    }
942
943    #[tokio::test]
944    async fn resumable_upload() {
945        let backend = InMemoryBackend::new("test");
946        let id = make_id();
947        let metadata = Metadata {
948            custom: [("preserved".into(), "yes".into())].into(),
949            ..Default::default()
950        };
951
952        // Upload an object and create a session for the same key.
953        backend
954            .put_object(
955                &id,
956                &Metadata::default(),
957                stream::single("old"),
958                Timestamp::now(),
959            )
960            .await
961            .unwrap();
962        let upload_length = NonZeroU64::new(3).unwrap();
963        let token = backend
964            .create_upload_session(&id, &metadata, upload_length)
965            .await
966            .unwrap()
967            .unwrap();
968        let token = Session {
969            object_id: id.clone(),
970            upload_length,
971            backend_token: token,
972        };
973        assert_eq!(
974            backend.upload_offset(&token).await.unwrap(),
975            UploadProgress::Incomplete { offset: 0 }
976        );
977
978        // Upload a prefix. The session advances, and the old object remains visible.
979        assert_eq!(
980            backend
981                .put_chunk(&token, 0, 1, stream::single("a"))
982                .await
983                .unwrap(),
984            UploadProgress::Incomplete { offset: 1 }
985        );
986        assert_eq!(backend.get(&id).expect_object().1, "old");
987        assert_eq!(
988            backend
989                .put_chunk(&token, 0, 0, stream::single(""))
990                .await
991                .unwrap(),
992            UploadProgress::Incomplete { offset: 1 }
993        );
994        let error = backend
995            .put_chunk(&token, 0, 1, stream::single("a"))
996            .await
997            .unwrap_err();
998        assert_eq!(error.kind(), ErrorKind::UploadOffsetMismatch { offset: 1 });
999
1000        // Upload the suffix. Publication replaces the object and deletes the session.
1001        assert_eq!(
1002            backend
1003                .put_chunk(&token, 1, 2, stream::single("bc"))
1004                .await
1005                .unwrap(),
1006            UploadProgress::Complete
1007        );
1008        let (actual, bytes) = backend.get(&id).expect_object();
1009        assert_eq!(bytes, "abc");
1010        assert_eq!(actual.custom, metadata.custom);
1011        assert_eq!(actual.size, metadata.size);
1012        assert_eq!(
1013            backend.upload_offset(&token).await.unwrap_err().kind(),
1014            ErrorKind::UnknownUploadSession
1015        );
1016    }
1017
1018    #[tokio::test]
1019    async fn resumable_cancel_and_invalid_sessions() {
1020        let backend = InMemoryBackend::new("test");
1021        let id = make_id();
1022        let token = create_session(&backend, &id, 3).await;
1023
1024        // The backend rejects a chunk whose declared range exceeds the session length.
1025        let error = backend
1026            .put_chunk(&token, 3, 1, stream::single("x"))
1027            .await
1028            .unwrap_err();
1029        assert_eq!(
1030            error.kind(),
1031            ErrorKind::ChunkExceedsUploadLength {
1032                offset: 3,
1033                content_length: 1,
1034                upload_length: 3
1035            }
1036        );
1037
1038        // Cancellation discards partial progress and makes the token unknown to the backend.
1039        backend
1040            .put_chunk(&token, 0, 1, stream::single("a"))
1041            .await
1042            .unwrap();
1043        backend.cancel_upload(&token).await.unwrap();
1044        assert!(!backend.contains(&id));
1045        assert_eq!(
1046            backend.upload_offset(&token).await.unwrap_err().kind(),
1047            ErrorKind::UnknownUploadSession
1048        );
1049    }
1050
1051    #[tokio::test]
1052    async fn failed_resumable_chunk_preserves_partial_progress() {
1053        let backend = InMemoryBackend::new("test");
1054        let id = make_id();
1055        let token = create_session(&backend, &id, 4).await;
1056
1057        // Persist a prefix, then disconnect after writing one byte of the next chunk.
1058        backend
1059            .put_chunk(&token, 0, 2, stream::single("ab"))
1060            .await
1061            .unwrap();
1062        let body = futures_util::stream::iter([
1063            Ok(Bytes::from_static(b"c")),
1064            Err(stream::ClientError::new(std::io::Error::other(
1065                "interrupted",
1066            ))),
1067        ])
1068        .boxed();
1069        assert_eq!(
1070            backend
1071                .put_chunk(&token, 2, 2, body)
1072                .await
1073                .unwrap_err()
1074                .kind(),
1075            ErrorKind::ClientStream
1076        );
1077
1078        // Resume from the partial byte and verify the complete object.
1079        assert_eq!(
1080            backend.upload_offset(&token).await.unwrap(),
1081            UploadProgress::Incomplete { offset: 3 }
1082        );
1083        assert_eq!(
1084            backend
1085                .put_chunk(&token, 3, 1, stream::single("d"))
1086                .await
1087                .unwrap(),
1088            UploadProgress::Complete
1089        );
1090        assert_eq!(backend.get(&id).expect_object().1, "abcd");
1091    }
1092
1093    #[tokio::test]
1094    async fn resumable_serializes_session_operations() {
1095        let backend = InMemoryBackend::new("test");
1096        let id = make_id();
1097        let token = create_session(&backend, &id, 1).await;
1098        let (sender, receiver) = tokio::sync::oneshot::channel();
1099        let body = futures_util::stream::once(async { Ok(receiver.await.unwrap()) }).boxed();
1100
1101        // An offset query waits while a chunk holds the session lock.
1102        let request = backend.put_chunk(&token, 0, 1, body);
1103        tokio::pin!(request);
1104        assert!(futures_util::poll!(&mut request).is_pending());
1105        let query = backend.upload_offset(&token);
1106        tokio::pin!(query);
1107        assert!(futures_util::poll!(&mut query).is_pending());
1108
1109        // Completion wakes the query, which observes that the session is now missing.
1110        sender.send(Bytes::from_static(b"x")).unwrap();
1111        assert_eq!(request.await.unwrap(), UploadProgress::Complete);
1112        assert_eq!(
1113            query.await.unwrap_err().kind(),
1114            ErrorKind::UnknownUploadSession
1115        );
1116        assert_eq!(backend.get(&id).expect_object().1, "x");
1117    }
1118
1119    #[cfg(feature = "storage-cogs")]
1120    #[tokio::test]
1121    async fn resumable_emits_only_publication() {
1122        let (backend, producer) = backend_with_change_stream();
1123        let id = make_id();
1124        let token = create_session(&backend, &id, 2).await;
1125
1126        // Partial session state is not reported as a stored object.
1127        backend
1128            .put_chunk(&token, 0, 1, stream::single("a"))
1129            .await
1130            .unwrap();
1131        assert!(producer.records().is_empty());
1132
1133        // Completion emits exactly one write for the published object.
1134        backend
1135            .put_chunk(&token, 1, 1, stream::single("b"))
1136            .await
1137            .unwrap();
1138        let records = producer.records();
1139        assert_eq!(records.len(), 1);
1140        assert_eq!(records[0].op_type, OpType::Write);
1141    }
1142
1143    #[tokio::test]
1144    async fn set_expiry() {
1145        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
1146        for policy in [
1147            ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
1148            ExpirationPolicy::TimeToIdle(Duration::from_hours(1)),
1149        ] {
1150            let original_expiry = access_time + Duration::from_hours(1);
1151            let requested = original_expiry + Duration::from_hours(1) + Duration::from_nanos(999);
1152            for target in [
1153                ExpiryTarget::At(requested),
1154                ExpiryTarget::FromCreation(Duration::from_hours(2) + Duration::from_nanos(999)),
1155            ] {
1156                let backend = InMemoryBackend::new("test");
1157                let id = make_id();
1158                let metadata = Metadata {
1159                    expiration_policy: policy,
1160                    time_created: Some(access_time),
1161                    time_expires: Some(original_expiry),
1162                    custom: [("preserved".into(), "yes".into())].into(),
1163                    ..Default::default()
1164                };
1165                backend
1166                    .put_object(&id, &metadata, stream::single("payload"), access_time)
1167                    .await
1168                    .unwrap();
1169
1170                let restricted = ExpiryUpdate {
1171                    max: Some(Duration::ZERO),
1172                    target,
1173                };
1174                assert_eq!(
1175                    backend
1176                        .set_expiry(&id, restricted, access_time)
1177                        .await
1178                        .unwrap_err()
1179                        .kind(),
1180                    ErrorKind::InvalidMetadata,
1181                );
1182                let target = ExpiryUpdate {
1183                    max: Some(Duration::from_hours(100)),
1184                    target,
1185                };
1186                assert_eq!(
1187                    backend.set_expiry(&id, target, access_time).await.unwrap(),
1188                    SetExpiryResponse::Satisfied(requested)
1189                );
1190                assert_eq!(
1191                    backend
1192                        .set_expiry(&id, restricted, access_time)
1193                        .await
1194                        .unwrap_err()
1195                        .kind(),
1196                    ErrorKind::InvalidMetadata,
1197                );
1198                let (updated, payload) = backend.get(&id).expect_object();
1199                let expected_policy = match policy {
1200                    ExpirationPolicy::TimeToLive(_) => {
1201                        ExpirationPolicy::TimeToLive(Duration::from_hours(2))
1202                    }
1203                    other => other,
1204                };
1205                assert_eq!(updated.expiration_policy, expected_policy);
1206                assert_eq!(updated.time_created, metadata.time_created);
1207                assert_eq!(updated.custom, metadata.custom);
1208                assert_eq!(payload, Bytes::from_static(b"payload"));
1209                assert_eq!(updated.time_expires, Some(requested));
1210
1211                assert_eq!(
1212                    backend
1213                        .set_expiry(&id, ExpiryTarget::At(original_expiry).into(), access_time)
1214                        .await
1215                        .unwrap(),
1216                    SetExpiryResponse::Satisfied(original_expiry)
1217                );
1218                assert_eq!(backend.get(&id).expect_object().0, updated);
1219            }
1220        }
1221    }
1222
1223    #[test]
1224    fn extend_object_expiry_preserves_metadata_on_error() {
1225        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
1226        let original_expiry = access_time + Duration::from_hours(1);
1227        let requested = original_expiry + Duration::from_hours(1);
1228        for (time_created, ttl) in [
1229            (
1230                Some(requested + Duration::from_secs(1)),
1231                Duration::from_hours(1),
1232            ),
1233            (None, Duration::MAX),
1234        ] {
1235            let original = Metadata {
1236                expiration_policy: ExpirationPolicy::TimeToLive(ttl),
1237                time_created,
1238                time_expires: Some(original_expiry),
1239                ..Default::default()
1240            };
1241            let mut metadata = original.clone();
1242            for _ in 0..2 {
1243                assert_eq!(
1244                    extend_object_expiry(
1245                        &mut metadata,
1246                        ExpiryTarget::At(requested).into(),
1247                        access_time,
1248                    )
1249                    .unwrap_err()
1250                    .kind(),
1251                    ErrorKind::CorruptData
1252                );
1253                assert_eq!(metadata, original);
1254            }
1255        }
1256    }
1257
1258    #[tokio::test]
1259    async fn set_expiry_rejected() {
1260        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
1261        let backend = InMemoryBackend::new("test");
1262        let absent = make_id();
1263        assert_eq!(
1264            backend
1265                .set_expiry(
1266                    &absent,
1267                    ExpiryTarget::At(access_time + Duration::from_hours(1)).into(),
1268                    access_time,
1269                )
1270                .await
1271                .unwrap(),
1272            SetExpiryResponse::NotFound
1273        );
1274
1275        let missing_creation = make_id();
1276        let original_expiry = access_time + Duration::from_hours(1);
1277        backend
1278            .put_object(
1279                &missing_creation,
1280                &Metadata {
1281                    expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
1282                    time_expires: Some(original_expiry),
1283                    ..Default::default()
1284                },
1285                stream::single("legacy"),
1286                access_time,
1287            )
1288            .await
1289            .unwrap();
1290        let absolute = original_expiry + Duration::from_hours(1);
1291        assert_eq!(
1292            backend
1293                .set_expiry(
1294                    &missing_creation,
1295                    ExpiryTarget::At(absolute).into(),
1296                    access_time
1297                )
1298                .await
1299                .unwrap(),
1300            SetExpiryResponse::Satisfied(absolute)
1301        );
1302        let updated = backend.get(&missing_creation).expect_object().0;
1303        assert_eq!(
1304            updated.expiration_policy,
1305            ExpirationPolicy::TimeToLive(Duration::from_hours(2))
1306        );
1307        assert_eq!(updated.time_created, None);
1308        assert_eq!(
1309            backend
1310                .set_expiry(
1311                    &missing_creation,
1312                    ExpiryTarget::FromCreation(Duration::ZERO).into(),
1313                    access_time,
1314                )
1315                .await
1316                .unwrap(),
1317            SetExpiryResponse::Rejected
1318        );
1319
1320        let manual = make_id();
1321        backend
1322            .put_object(
1323                &manual,
1324                &Metadata::default(),
1325                stream::single("manual"),
1326                access_time,
1327            )
1328            .await
1329            .unwrap();
1330        assert_eq!(
1331            backend
1332                .set_expiry(
1333                    &manual,
1334                    ExpiryTarget::At(access_time + Duration::from_hours(1)).into(),
1335                    access_time,
1336                )
1337                .await
1338                .unwrap(),
1339            SetExpiryResponse::Rejected
1340        );
1341
1342        let expired = make_id();
1343        let metadata = Metadata {
1344            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
1345            time_expires: Some(access_time - Duration::from_secs(1)),
1346            ..Default::default()
1347        };
1348        backend
1349            .put_object(&expired, &metadata, stream::single("expired"), access_time)
1350            .await
1351            .unwrap();
1352        let deadline = metadata.time_expires.unwrap();
1353        for time in [deadline - Duration::from_secs(1), deadline] {
1354            assert!(
1355                backend
1356                    .get_object(&expired, time, None)
1357                    .await
1358                    .unwrap()
1359                    .is_some()
1360            );
1361        }
1362        assert!(
1363            backend
1364                .get_object(&expired, access_time, None)
1365                .await
1366                .unwrap()
1367                .is_none()
1368        );
1369        assert_eq!(
1370            backend
1371                .set_expiry(
1372                    &expired,
1373                    ExpiryTarget::At(access_time + Duration::from_hours(1)).into(),
1374                    access_time,
1375                )
1376                .await
1377                .unwrap(),
1378            SetExpiryResponse::NotFound
1379        );
1380    }
1381
1382    #[tokio::test]
1383    async fn redirect_expiry() {
1384        let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap();
1385        let backend = InMemoryBackend::new("test");
1386        let id = make_id();
1387        let target = make_id();
1388        let other = make_id();
1389        let old_expiry = access_time + Duration::from_hours(1);
1390        backend
1391            .compare_and_write(
1392                &id,
1393                None,
1394                TieredWrite::Tombstone(Tombstone {
1395                    target: target.clone(),
1396                    time_expires: Some(old_expiry),
1397                }),
1398                access_time,
1399            )
1400            .await
1401            .unwrap();
1402
1403        assert!(
1404            backend
1405                .get_object(&id, access_time, None)
1406                .await
1407                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
1408        );
1409        assert!(
1410            backend
1411                .get_metadata(&id, access_time)
1412                .await
1413                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
1414        );
1415
1416        let new_expiry = old_expiry + Duration::from_hours(1);
1417        assert_eq!(
1418            backend
1419                .compare_and_update(
1420                    &id,
1421                    Some(&other),
1422                    TieredUpdate::SetExpiry(ExpiryTarget::At(new_expiry).into()),
1423                    access_time,
1424                )
1425                .await
1426                .unwrap(),
1427            SetExpiryResponse::Rejected
1428        );
1429        assert_eq!(
1430            backend.get(&id).expect_tombstone().time_expires,
1431            Some(old_expiry)
1432        );
1433        assert_eq!(
1434            backend
1435                .compare_and_update(
1436                    &id,
1437                    Some(&target),
1438                    TieredUpdate::SetExpiry(
1439                        ExpiryTarget::FromCreation(Duration::from_hours(2)).into()
1440                    ),
1441                    access_time,
1442                )
1443                .await
1444                .unwrap(),
1445            SetExpiryResponse::Rejected
1446        );
1447        assert_eq!(
1448            backend
1449                .compare_and_update(
1450                    &id,
1451                    Some(&target),
1452                    TieredUpdate::SetExpiry(ExpiryTarget::At(new_expiry).into()),
1453                    access_time
1454                )
1455                .await
1456                .unwrap(),
1457            SetExpiryResponse::Satisfied(new_expiry)
1458        );
1459        assert_eq!(
1460            backend.get(&id).expect_tombstone().time_expires,
1461            Some(new_expiry)
1462        );
1463    }
1464
1465    #[tokio::test]
1466    async fn multipart_single_part() {
1467        let backend = InMemoryBackend::new("test");
1468        let id = make_id();
1469        let metadata = Metadata {
1470            content_type: "text/plain".into(),
1471            expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_hours(1)),
1472            origin: Some("203.0.113.42".into()),
1473            custom: [("foo".into(), "bar".into())].into(),
1474            ..Default::default()
1475        };
1476
1477        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1478
1479        let data = b"hello, multipart world!";
1480        let etag = backend
1481            .upload_part(
1482                &id,
1483                &upload_id,
1484                NonZeroU32::new(1).unwrap(),
1485                data.len() as u64,
1486                None,
1487                stream::single(data.to_vec()),
1488            )
1489            .await
1490            .unwrap();
1491
1492        let result = backend
1493            .complete_multipart(
1494                &id,
1495                &upload_id,
1496                vec![CompletedPart {
1497                    part_number: NonZeroU32::new(1).unwrap(),
1498                    etag,
1499                }],
1500                Timestamp::now(),
1501            )
1502            .await
1503            .unwrap();
1504        assert!(result.is_none(), "expected no error on complete");
1505
1506        let (meta, _, body) = backend
1507            .get_object(&id, Timestamp::now(), None)
1508            .await
1509            .unwrap()
1510            .unwrap();
1511        let payload = stream::read_to_vec(body).await.unwrap();
1512        assert_eq!(payload, data);
1513        assert_eq!(meta.content_type, "text/plain".to_string());
1514        assert_eq!(
1515            meta.expiration_policy,
1516            ExpirationPolicy::TimeToIdle(Duration::from_hours(1))
1517        );
1518        assert_eq!(meta.origin, Some("203.0.113.42".into()));
1519        assert_eq!(meta.custom, [("foo".into(), "bar".into())].into());
1520    }
1521
1522    #[tokio::test]
1523    async fn multipart_multiple_parts() {
1524        let backend = InMemoryBackend::new("test");
1525        let id = make_id();
1526        let metadata = Metadata::default();
1527
1528        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1529
1530        let part1 = b"aaaa".to_vec();
1531        let part2 = b"bbbb".to_vec();
1532        let part3 = b"cc".to_vec();
1533
1534        let etag1 = backend
1535            .upload_part(
1536                &id,
1537                &upload_id,
1538                NonZeroU32::new(1).unwrap(),
1539                part1.len() as u64,
1540                None,
1541                stream::single(part1.clone()),
1542            )
1543            .await
1544            .unwrap();
1545        let etag2 = backend
1546            .upload_part(
1547                &id,
1548                &upload_id,
1549                NonZeroU32::new(2).unwrap(),
1550                part2.len() as u64,
1551                None,
1552                stream::single(part2.clone()),
1553            )
1554            .await
1555            .unwrap();
1556        let etag3 = backend
1557            .upload_part(
1558                &id,
1559                &upload_id,
1560                NonZeroU32::new(3).unwrap(),
1561                part3.len() as u64,
1562                None,
1563                stream::single(part3.clone()),
1564            )
1565            .await
1566            .unwrap();
1567
1568        let result = backend
1569            .complete_multipart(
1570                &id,
1571                &upload_id,
1572                vec![
1573                    CompletedPart {
1574                        part_number: NonZeroU32::new(1).unwrap(),
1575                        etag: etag1,
1576                    },
1577                    CompletedPart {
1578                        part_number: NonZeroU32::new(2).unwrap(),
1579                        etag: etag2,
1580                    },
1581                    CompletedPart {
1582                        part_number: NonZeroU32::new(3).unwrap(),
1583                        etag: etag3,
1584                    },
1585                ],
1586                Timestamp::now(),
1587            )
1588            .await
1589            .unwrap();
1590        assert!(result.is_none());
1591
1592        let (_, _, body) = backend
1593            .get_object(&id, Timestamp::now(), None)
1594            .await
1595            .unwrap()
1596            .unwrap();
1597        let payload = stream::read_to_vec(body).await.unwrap();
1598        assert_eq!(payload, b"aaaabbbbcc");
1599    }
1600
1601    #[tokio::test]
1602    async fn multipart_list_parts() {
1603        let backend = InMemoryBackend::new("test");
1604        let id = make_id();
1605        let metadata = Metadata::default();
1606
1607        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1608
1609        let etag1 = backend
1610            .upload_part(
1611                &id,
1612                &upload_id,
1613                NonZeroU32::new(1).unwrap(),
1614                3,
1615                None,
1616                stream::single(b"aaa".to_vec()),
1617            )
1618            .await
1619            .unwrap();
1620        let etag2 = backend
1621            .upload_part(
1622                &id,
1623                &upload_id,
1624                NonZeroU32::new(2).unwrap(),
1625                3,
1626                None,
1627                stream::single(b"bbb".to_vec()),
1628            )
1629            .await
1630            .unwrap();
1631
1632        let list = backend
1633            .list_parts(&id, &upload_id, None, None)
1634            .await
1635            .unwrap();
1636        assert_eq!(list.parts.len(), 2);
1637        assert_eq!(list.parts[0].part_number.get(), 1);
1638        assert_eq!(list.parts[0].etag, etag1);
1639        assert_eq!(list.parts[0].size, 3);
1640        assert_eq!(list.parts[1].part_number.get(), 2);
1641        assert_eq!(list.parts[1].etag, etag2);
1642        assert_eq!(list.parts[1].size, 3);
1643
1644        // Pagination
1645        let page1 = backend
1646            .list_parts(&id, &upload_id, Some(1), None)
1647            .await
1648            .unwrap();
1649        assert_eq!(page1.parts.len(), 1);
1650        assert_eq!(page1.parts[0].part_number.get(), 1);
1651        assert!(page1.is_truncated);
1652        assert!(page1.next_part_number_marker.is_some());
1653
1654        let page2 = backend
1655            .list_parts(&id, &upload_id, Some(1), page1.next_part_number_marker)
1656            .await
1657            .unwrap();
1658        assert_eq!(page2.parts.len(), 1);
1659        assert_eq!(page2.parts[0].part_number.get(), 2);
1660
1661        backend.abort_multipart(&id, &upload_id).await.unwrap();
1662    }
1663
1664    #[tokio::test]
1665    async fn multipart_abort() {
1666        let backend = InMemoryBackend::new("test");
1667        let id = make_id();
1668        let metadata = Metadata::default();
1669
1670        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1671
1672        backend
1673            .upload_part(
1674                &id,
1675                &upload_id,
1676                NonZeroU32::new(1).unwrap(),
1677                5,
1678                None,
1679                stream::single(b"hello".to_vec()),
1680            )
1681            .await
1682            .unwrap();
1683
1684        backend.abort_multipart(&id, &upload_id).await.unwrap();
1685
1686        let result = backend
1687            .get_object(&id, Timestamp::now(), None)
1688            .await
1689            .unwrap();
1690        assert!(result.is_none(), "object should not exist after abort");
1691    }
1692
1693    #[tokio::test]
1694    async fn multipart_invalid_etag() {
1695        let backend = InMemoryBackend::new("test");
1696        let id = make_id();
1697        let metadata = Metadata::default();
1698
1699        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1700
1701        let etag = backend
1702            .upload_part(
1703                &id,
1704                &upload_id,
1705                NonZeroU32::new(1).unwrap(),
1706                5,
1707                None,
1708                stream::single(b"hello".to_vec()),
1709            )
1710            .await
1711            .unwrap();
1712
1713        let result = backend
1714            .complete_multipart(
1715                &id,
1716                &upload_id,
1717                vec![CompletedPart {
1718                    part_number: NonZeroU32::new(1).unwrap(),
1719                    etag: "wrong-etag".into(),
1720                }],
1721                Timestamp::now(),
1722            )
1723            .await
1724            .unwrap();
1725        assert!(result.is_some(), "expected error for bad etag");
1726        assert_eq!(result.unwrap().code, "InvalidPart");
1727
1728        // Upload must survive a failed complete so the client can retry.
1729        let result = backend
1730            .complete_multipart(
1731                &id,
1732                &upload_id,
1733                vec![CompletedPart {
1734                    part_number: NonZeroU32::new(1).unwrap(),
1735                    etag,
1736                }],
1737                Timestamp::now(),
1738            )
1739            .await
1740            .unwrap();
1741        assert!(result.is_none(), "retry with correct etag should succeed");
1742    }
1743
1744    #[tokio::test]
1745    async fn multipart_missing_part() {
1746        let backend = InMemoryBackend::new("test");
1747        let id = make_id();
1748        let metadata = Metadata::default();
1749
1750        let upload_id = backend.initiate_multipart(&id, &metadata).await.unwrap();
1751
1752        let etag = backend
1753            .upload_part(
1754                &id,
1755                &upload_id,
1756                NonZeroU32::new(1).unwrap(),
1757                5,
1758                None,
1759                stream::single(b"hello".to_vec()),
1760            )
1761            .await
1762            .unwrap();
1763
1764        let result = backend
1765            .complete_multipart(
1766                &id,
1767                &upload_id,
1768                vec![CompletedPart {
1769                    part_number: NonZeroU32::new(99).unwrap(),
1770                    etag: "whatever".into(),
1771                }],
1772                Timestamp::now(),
1773            )
1774            .await
1775            .unwrap();
1776        assert!(result.is_some(), "expected error for missing part");
1777        assert_eq!(result.unwrap().code, "InvalidPart");
1778
1779        // Upload must survive a failed complete so the client can retry.
1780        let result = backend
1781            .complete_multipart(
1782                &id,
1783                &upload_id,
1784                vec![CompletedPart {
1785                    part_number: NonZeroU32::new(1).unwrap(),
1786                    etag,
1787                }],
1788                Timestamp::now(),
1789            )
1790            .await
1791            .unwrap();
1792        assert!(result.is_none(), "retry with correct part should succeed");
1793    }
1794
1795    #[cfg(feature = "storage-cogs")]
1796    fn backend_with_change_stream() -> (InMemoryBackend, DummyProducer) {
1797        use crate::change_stream::CostTrackerStreamConfig;
1798
1799        let (streams, producer) = crate::change_stream::dummy_factory();
1800        let change_stream = streams.build(Some(&CostTrackerStreamConfig {
1801            shared_resource_id: "in_memory_objectstore".into(),
1802            sample_rate: 1.0,
1803        }));
1804        (
1805            InMemoryBackend::new("test").with_change_stream(change_stream),
1806            producer,
1807        )
1808    }
1809
1810    #[cfg(feature = "storage-cogs")]
1811    #[tokio::test]
1812    async fn change_stream_reports_writes_and_deletes() {
1813        let (backend, producer) = backend_with_change_stream();
1814        let id = make_id();
1815        let metadata = Metadata::default();
1816        let payload = b"hello";
1817
1818        backend
1819            .put_object(
1820                &id,
1821                &metadata,
1822                stream::single(payload.to_vec()),
1823                Timestamp::now(),
1824            )
1825            .await
1826            .unwrap();
1827        backend.delete_object(&id, Timestamp::now()).await.unwrap();
1828        // The object is already gone, so this reports nothing.
1829        backend.delete_object(&id, Timestamp::now()).await.unwrap();
1830
1831        let records = producer.records();
1832        assert_eq!(records.len(), 2);
1833        assert_eq!(records[0].op_type, OpType::Write);
1834        assert_eq!(records[0].app_feature, "testing");
1835        assert_eq!(
1836            records[0].size,
1837            Some((json_len(&metadata) + payload.len()) as u64),
1838            "the reported size covers metadata as well as the payload"
1839        );
1840        assert!(json_len(&metadata) > 0, "metadata must contribute bytes");
1841        assert_eq!(records[1].op_type, OpType::Delete);
1842    }
1843
1844    #[cfg(feature = "storage-cogs")]
1845    #[tokio::test]
1846    async fn change_stream_reports_expiry_extension_as_update() {
1847        let (backend, producer) = backend_with_change_stream();
1848        let id = make_id();
1849        let access_time = Timestamp::now();
1850        let expires = access_time + Duration::from_secs(3600);
1851        let metadata = Metadata {
1852            time_expires: Some(expires),
1853            ..Default::default()
1854        };
1855
1856        backend
1857            .put_object(
1858                &id,
1859                &metadata,
1860                stream::single(b"hello".to_vec()),
1861                access_time,
1862            )
1863            .await
1864            .unwrap();
1865        producer.clear();
1866
1867        let extended = expires + Duration::from_secs(3600);
1868        assert_eq!(
1869            backend
1870                .set_expiry(&id, ExpiryTarget::At(extended).into(), access_time)
1871                .await
1872                .unwrap(),
1873            SetExpiryResponse::Satisfied(extended)
1874        );
1875
1876        let records = producer.records();
1877        assert_eq!(records.len(), 1);
1878        assert_eq!(records[0].op_type, OpType::Update);
1879
1880        // A deadline that already covers the request writes nothing, so it reports nothing.
1881        producer.clear();
1882        let requested = expires + Duration::from_secs(60);
1883        assert_eq!(
1884            backend
1885                .set_expiry(&id, ExpiryTarget::At(requested).into(), access_time)
1886                .await
1887                .unwrap(),
1888            SetExpiryResponse::Satisfied(requested)
1889        );
1890        assert!(producer.records().is_empty());
1891    }
1892
1893    #[cfg(feature = "storage-cogs")]
1894    #[tokio::test]
1895    async fn change_stream_reports_tombstone_expiry_extension() {
1896        let (backend, producer) = backend_with_change_stream();
1897        let id = make_id();
1898        let target = make_id();
1899        let access_time = Timestamp::now();
1900        let expires = access_time + Duration::from_secs(3600);
1901
1902        backend
1903            .compare_and_write(
1904                &id,
1905                None,
1906                TieredWrite::Tombstone(Tombstone {
1907                    target: target.clone(),
1908                    time_expires: Some(expires),
1909                }),
1910                access_time,
1911            )
1912            .await
1913            .unwrap();
1914        producer.clear();
1915
1916        let extended = expires + Duration::from_secs(3600);
1917        assert_eq!(
1918            backend
1919                .compare_and_update(
1920                    &id,
1921                    Some(&target),
1922                    TieredUpdate::SetExpiry(ExpiryTarget::At(extended).into()),
1923                    access_time,
1924                )
1925                .await
1926                .unwrap(),
1927            SetExpiryResponse::Satisfied(extended)
1928        );
1929
1930        let records = producer.records();
1931        assert_eq!(records.len(), 1);
1932        assert_eq!(records[0].op_type, OpType::Update);
1933    }
1934}
objectstore_service/backend/in_memory.rs

objectstore_service/backend/
in_memory.rs
