From f43e791ad75d937d5aeb0e44111b3ce3cd8b4d34 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 09:49:43 +0000 Subject: [PATCH] fix(service): Start resumable upload expiry at commit The long-term blob of a resumable upload keeps the deadline resolved at session creation. When the final chunk lands, the blob's deadline is moved to the commit time plus the policy's duration before the HV tombstone is written with the confirmed deadline. The blob may already be past its initial deadline at that point, so ExpiryUpdate gains an explicit include_expired flag that the long-term backends with resumable uploads (GCS, local-fs, in-memory) honor. Other backends and the public tiered set_expiry reject it as unsupported. The session token now carries the required ExpirationPolicy, so tokens issued before this change fail to decode instead of committing without an expiry. Fixes FS-564 Co-Authored-By: Lorenzo Cian <17258265+lcian@users.noreply.github.com> --- objectstore-server/src/endpoints/objects.rs | 5 +- objectstore-service/docs/architecture.md | 3 + objectstore-service/src/backend/bigtable.rs | 1 + objectstore-service/src/backend/common.rs | 33 ++- objectstore-service/src/backend/gcs.rs | 74 +++++- objectstore-service/src/backend/in_memory.rs | 68 ++++- objectstore-service/src/backend/local_fs.rs | 64 ++++- .../src/backend/s3_compatible.rs | 1 + objectstore-service/src/backend/tiered.rs | 242 +++++++++++++++++- 9 files changed, 452 insertions(+), 39 deletions(-) diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index 4db526fee..4b72bbea3 100644 --- a/objectstore-server/src/endpoints/objects.rs +++ b/objectstore-server/src/endpoints/objects.rs @@ -131,7 +131,10 @@ async fn object_patch( }; let max = state.config.usecases.get_max_expiry(id.usecase()); - let update = ExpiryUpdate { target, max }; + let update = ExpiryUpdate { + max, + ..target.into() + }; match service.set_expiry(id, update, access_time).await? { SetExpiryResponse::Satisfied(_) => Ok(StatusCode::NO_CONTENT), diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 145ab78cf..23800cb6a 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -339,6 +339,9 @@ Not all backends support resumable uploads. A backend returns no session when it particular upload; this is a routine outcome rather than an error. Acceptance can depend on the declared size, the metadata, or whether resuming is possible in principle. +A time-based expiration policy declared at session creation counts from the upload's completion, +not from session creation, so a slow upload does not shorten the object's lifetime. + The offset returned by every chunk response is authoritative. Session creation reports each upload's granularity in bytes, or zero when it imposes none. For a positive granularity, a backend may accept the full request body but persist only the largest aligned prefix, so a client must not diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index eba004630..454b7a9c1 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -1298,6 +1298,7 @@ impl HighVolumeBackend for BigTableBackend { access_time: Timestamp, ) -> Result { let TieredUpdate::SetExpiry(expiry_target) = update; + expiry_target.ensure_unexpired_only()?; let path = id.as_storage_path().to_string().into_bytes(); // Inline extension needs metadata and payload from the same read so a diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs index 6340c5679..74efb33fd 100644 --- a/objectstore-service/src/backend/common.rs +++ b/objectstore-service/src/backend/common.rs @@ -92,15 +92,41 @@ pub struct ExpiryUpdate { pub target: ExpiryTarget, /// Maximum remaining lifetime. `None` means no limit. pub max: Option, + /// Also updates an object whose deadline has already passed but which still exists. + /// + /// The tiered backend sets this when committing a resumable upload: the long-term blob's + /// deadline was resolved when the session was created and may pass before the upload + /// completes, while the blob only becomes reachable once the commit extends it. + /// + /// Only long-term backends with resumable uploads honor this; others return + /// [`ErrorKind::Unsupported`]. Defaults to `false`. + pub include_expired: bool, } impl From for ExpiryUpdate { fn from(target: ExpiryTarget) -> Self { - Self { target, max: None } + Self { + target, + max: None, + include_expired: false, + } } } impl ExpiryUpdate { + /// Returns [`ErrorKind::Unsupported`] if this update sets [`include_expired`](Self::include_expired). + /// + /// For backends that only update unexpired objects. + pub(super) fn ensure_unexpired_only(&self) -> Result<()> { + if self.include_expired { + return Err(Error::new( + ErrorKind::Unsupported, + "expiry updates of expired objects are not supported", + )); + } + Ok(()) + } + /// Resolves and validates the requested deadline using the stored object metadata. /// /// Returns `None` if the creation-time anchor is required but unavailable. @@ -236,7 +262,8 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { /// /// Returns [`SetExpiryResponse::Satisfied`] when extended or already satisfied, /// [`SetExpiryResponse::NotFound`] when observed absent or expired, or - /// [`SetExpiryResponse::Rejected`] when ineligible or conflicting. + /// [`SetExpiryResponse::Rejected`] when ineligible or conflicting. With + /// [`ExpiryUpdate::include_expired`], expired objects that still exist are updated instead. /// Limit violations return [`ErrorKind::InvalidMetadata`]; backend failures also /// return errors. Limits are checked before reporting an already-satisfied request. async fn set_expiry( @@ -779,8 +806,8 @@ mod tests { ExpiryTarget::FromCreation(Duration::from_hours(3)), ] { let update = ExpiryUpdate { - target, max: Some(Duration::from_hours(1)), + ..target.into() }; assert_eq!( update diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 25b3d9440..8f1d80dfd 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -651,12 +651,30 @@ impl GcsBackend { } /// Fetches GCS object metadata without modifying the object. - #[tracing::instrument(level = "debug", fields(%object_url), skip(self))] + /// + /// Returns `None` for missing objects and for objects past their expiry. async fn get_gcs_metadata( &self, object_url: &Url, access_time: Timestamp, ) -> Result> { + let metadata_opt = self.get_gcs_metadata_unfiltered(object_url).await?; + + // Filter already expired objects but leave them to garbage collection + if metadata_opt + .as_ref() + .is_some_and(|metadata| metadata.is_expired(access_time)) + { + objectstore_log::debug!("Object found but past expiry"); + return Ok(None); + } + + Ok(metadata_opt) + } + + /// Fetches GCS object metadata, including objects past their expiry. + #[tracing::instrument(level = "debug", fields(%object_url), skip(self))] + async fn get_gcs_metadata_unfiltered(&self, object_url: &Url) -> Result> { let metadata_opt = self .with_retry("get_metadata", || async { let resp = self @@ -682,18 +700,11 @@ impl GcsBackend { }) .await?; - let Some(gcs_metadata) = metadata_opt else { + if metadata_opt.is_none() { objectstore_log::debug!("Object not found"); - return Ok(None); - }; - - // Filter already expired objects but leave them to garbage collection - if gcs_metadata.is_expired(access_time) { - objectstore_log::debug!("Object found but past expiry"); - return Ok(None); } - Ok(Some(gcs_metadata)) + Ok(metadata_opt) } /// Moves an object's `customTime` and optionally updates its metadata in the same PATCH. @@ -1082,7 +1093,12 @@ impl Backend for GcsBackend { access_time: Timestamp, ) -> Result { let object_url = self.object_url(id)?; - let Some(object) = self.get_gcs_metadata(&object_url, access_time).await? else { + let object = if target.include_expired { + self.get_gcs_metadata_unfiltered(&object_url).await? + } else { + self.get_gcs_metadata(&object_url, access_time).await? + }; + let Some(object) = object else { return Ok(SetExpiryResponse::NotFound); }; let Some(current_expiry) = object.custom_time else { @@ -2647,6 +2663,42 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_set_expiry_include_expired() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id(); + let now = Timestamp::now(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)), + time_expires: Some(now - Duration::from_hours(2)), + ..Default::default() + }; + backend + .put_object(&id, &metadata, stream::single("payload"), now) + .await?; + + let target = ExpiryTarget::At(now + Duration::from_hours(1)); + assert_eq!( + backend.set_expiry(&id, target.into(), now).await?, + SetExpiryResponse::NotFound + ); + + let update = ExpiryUpdate { + include_expired: true, + ..target.into() + }; + assert_eq!( + backend.set_expiry(&id, update, now).await?, + SetExpiryResponse::Satisfied(now + Duration::from_hours(1)) + ); + let stored = backend + .get_metadata(&id, now) + .await? + .expect("extended object is readable"); + assert_eq!(stored.time_expires, Some(now + Duration::from_hours(1))); + Ok(()) + } + #[tokio::test] async fn test_set_expiry() -> Result<()> { for is_ttl in [false, true] { diff --git a/objectstore-service/src/backend/in_memory.rs b/objectstore-service/src/backend/in_memory.rs index 54b65019d..bb7dd144b 100644 --- a/objectstore-service/src/backend/in_memory.rs +++ b/objectstore-service/src/backend/in_memory.rs @@ -135,15 +135,17 @@ impl InMemoryBackend { /// Applies `extend` to `id`'s entry, calling [`ChangeStream::begin_update`] first. /// /// Like a conditional write, this is rejected if the entry changed in the meantime. + /// Expired entries are only passed to `extend` if `include_expired` is set. async fn update_expiry( &self, id: &ObjectId, access_time: Timestamp, + include_expired: bool, extend: impl Fn(&mut StoreEntry) -> Result + Send + Sync, ) -> Result { let try_extend = |store: &Store| -> Result<(ExpiryOutcome, Option)> { match store.get(id) { - Some(entry) if !entry.is_expired(access_time) => { + Some(entry) if include_expired || !entry.is_expired(access_time) => { let mut updated = entry.clone(); let outcome = extend(&mut updated)?; Ok((outcome, Some(updated))) @@ -268,10 +270,17 @@ impl super::common::Backend for InMemoryBackend { target: ExpiryUpdate, access_time: Timestamp, ) -> Result { - self.update_expiry(id, access_time, |entry| match entry { - StoreEntry::Object(metadata, _) => extend_object_expiry(metadata, target, access_time), - StoreEntry::Tombstone(_) => Ok(ExpiryOutcome::Rejected), - }) + self.update_expiry( + id, + access_time, + target.include_expired, + |entry| match entry { + StoreEntry::Object(metadata, _) => { + extend_object_expiry(metadata, target, access_time) + } + StoreEntry::Tombstone(_) => Ok(ExpiryOutcome::Rejected), + }, + ) .await } @@ -546,7 +555,8 @@ impl HighVolumeBackend for InMemoryBackend { access_time: Timestamp, ) -> Result { let TieredUpdate::SetExpiry(expiry_target) = update; - self.update_expiry(id, access_time, |entry| match (entry, current) { + expiry_target.ensure_unexpired_only()?; + self.update_expiry(id, access_time, false, |entry| match (entry, current) { (StoreEntry::Object(metadata, _), None) => { extend_object_expiry(metadata, expiry_target, access_time) } @@ -902,6 +912,8 @@ fn extend_object_expiry( /// Resolves `target` and extends an active expiry time where valid. /// +/// Expired deadlines are only extended if [`ExpiryUpdate::include_expired`] is set. +/// /// Eligibility is checked before target resolution. fn extend_expiry( field: &mut Option, @@ -913,7 +925,7 @@ fn extend_expiry( return Ok(ExpiryOutcome::Rejected); // entries without a deadline cannot be extended }; - if time_expires < access_time { + if time_expires < access_time && !target.include_expired { return Ok(ExpiryOutcome::NotFound); // already expired } @@ -1245,7 +1257,7 @@ mod tests { let restricted = ExpiryUpdate { max: Some(Duration::ZERO), - target, + ..target.into() }; assert_eq!( backend @@ -1257,7 +1269,7 @@ mod tests { ); let target = ExpiryUpdate { max: Some(Duration::from_hours(100)), - target, + ..target.into() }; assert_eq!( backend.set_expiry(&id, target, access_time).await.unwrap(), @@ -1331,6 +1343,44 @@ mod tests { } } + #[tokio::test] + async fn set_expiry_include_expired() { + let backend = InMemoryBackend::new("test"); + let id = make_id(); + let now = Timestamp::now(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)), + time_created: Some(now - Duration::from_hours(3)), + time_expires: Some(now - Duration::from_hours(2)), + ..Default::default() + }; + backend + .put_object(&id, &metadata, stream::single("payload"), now) + .await + .unwrap(); + + let target = ExpiryTarget::At(now + Duration::from_hours(1)); + assert_eq!( + backend.set_expiry(&id, target.into(), now).await.unwrap(), + SetExpiryResponse::NotFound + ); + + let update = ExpiryUpdate { + include_expired: true, + ..target.into() + }; + assert_eq!( + backend.set_expiry(&id, update, now).await.unwrap(), + SetExpiryResponse::Satisfied(now + Duration::from_hours(1)) + ); + let (stored, _) = backend.get(&id).expect_object(); + assert_eq!(stored.time_expires, Some(now + Duration::from_hours(1))); + assert_eq!( + stored.expiration_policy, + ExpirationPolicy::TimeToLive(Duration::from_hours(4)) + ); + } + #[tokio::test] async fn set_expiry_rejected() { let access_time = Timestamp::from_unix_secs(1_700_000_000).unwrap(); diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index f38a6cf9a..91720c109 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -252,7 +252,12 @@ impl Backend for LocalFsBackend { let _guard = self.locks.acquire(id).await?; let path = self.path(id); - let Some(object) = ObjectFile::try_open(&path, access_time).await? else { + let object = if target.include_expired { + ObjectFile::try_open_unfiltered(&path).await? + } else { + ObjectFile::try_open(&path, access_time).await? + }; + let Some(object) = object else { return Ok(SetExpiryResponse::NotFound); }; let ObjectFile { @@ -995,8 +1000,23 @@ where } impl ObjectFile { - /// Opens an object file, returning `None` when it does not exist. + /// Opens an object file, returning `None` when it does not exist or has expired. async fn try_open(path: &Path, access_time: Timestamp) -> Result> { + let object = Self::try_open_unfiltered(path).await?; + if object + .as_ref() + .is_some_and(|object| object.metadata.is_expired(access_time)) + { + objectstore_log::debug!("Object found but past expiry"); + return Ok(None); + } + Ok(object) + } + + /// Opens an object file, returning `None` only when it does not exist. + /// + /// Unlike [`try_open`](Self::try_open), this also returns objects past their expiry. + async fn try_open_unfiltered(path: &Path) -> Result> { let file = match OpenOptions::new().read(true).open(path).await { Ok(file) => file, Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), @@ -1006,11 +1026,6 @@ impl ObjectFile { let mut reader = BufReader::new(file); let (mut metadata, preamble_len) = read_metadata_preamble(&mut reader).await?; - if metadata.is_expired(access_time) { - objectstore_log::debug!("Object found but past expiry"); - return Ok(None); - } - let preamble_len = preamble_len as u64; let file_len = reader .get_ref() @@ -1578,6 +1593,41 @@ mod tests { assert_eq!(draft_count(&backend, &id), 0); } + #[tokio::test] + async fn set_expiry_include_expired() { + let (_tempdir, backend) = make_backend(); + let id = make_id(); + let now = Timestamp::now(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)), + time_created: Some(now - Duration::from_hours(3)), + time_expires: Some(now - Duration::from_hours(2)), + ..Default::default() + }; + backend + .put_object(&id, &metadata, stream::single("payload"), now) + .await + .unwrap(); + + let target = ExpiryTarget::At(now + Duration::from_hours(1)); + assert_eq!( + backend.set_expiry(&id, target.into(), now).await.unwrap(), + SetExpiryResponse::NotFound + ); + + let update = ExpiryUpdate { + include_expired: true, + ..target.into() + }; + assert_eq!( + backend.set_expiry(&id, update, now).await.unwrap(), + SetExpiryResponse::Satisfied(now + Duration::from_hours(1)) + ); + let (stored, _, payload) = backend.get_object(&id, now, None).await.unwrap().unwrap(); + assert_eq!(stored.time_expires, Some(now + Duration::from_hours(1))); + assert_eq!(stream::read_to_vec(payload).await.unwrap(), b"payload"); + } + #[tokio::test] async fn set_expiry() { for is_ttl in [false, true] { diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index d8c0e6454..fd6167a69 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -469,6 +469,7 @@ impl Backend for S3CompatibleBackend { target: ExpiryUpdate, access_time: Timestamp, ) -> Result { + target.ensure_unexpired_only()?; let Some((mut metadata, _, response)) = self .request_object(Method::HEAD, id, access_time, None) .await? diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index dc4b25977..161fc396c 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -107,6 +107,12 @@ //! as the source of truth for whether the upload logically exists. //! This marker is consumed upon cancellation or the first submission of a final chunk, making //! finalization one-shot. +//! +//! Time-based expiration counts from the commit. When the session is created, the long-term blob +//! gets the deadline resolved from the request, which may pass while the upload is ongoing. Before +//! committing the tombstone, the final chunk moves the blob's deadline to the commit time plus the +//! policy's duration, using [`ExpiryUpdate::include_expired`] since the blob may already be past +//! its initial deadline. The tombstone then carries the confirmed deadline. use std::num::NonZeroU64; use std::sync::Arc; @@ -117,7 +123,7 @@ use base64::Engine as _; use bytes::Bytes; use futures_util::StreamExt; use objectstore_metrics::TimerGuard; -use objectstore_types::metadata::Metadata; +use objectstore_types::metadata::{ExpirationPolicy, Metadata}; use objectstore_types::range::ByteRange; use objectstore_types::resumable::UploadProgress; use objectstore_types::time::Timestamp; @@ -304,6 +310,42 @@ impl TieredStorage { Ok(()) } + /// Moves a completed resumable upload's deadline to the commit time plus `expires_in`. + /// + /// The blob still carries the deadline resolved at session creation, which may already have + /// passed, so the update includes expired blobs. Returns the deadline the redirect tombstone + /// must carry, or `None` without a time-based policy. + /// + /// Fails if the long-term backend does not confirm the deadline, so the caller never commits a + /// tombstone without a matching blob deadline. + async fn commit_upload_expiry( + &self, + revision: &ObjectId, + expires_in: Option, + ) -> Result> { + let Some(expires_in) = expires_in else { + return Ok(None); + }; + + let now = Timestamp::now(); + let update = ExpiryUpdate { + include_expired: true, + ..ExpiryTarget::At(now.saturating_add(expires_in)).into() + }; + match self + .inner + .long_term + .set_expiry(revision, update, now) + .await? + { + SetExpiryResponse::Satisfied(deadline) => Ok(Some(deadline)), + SetExpiryResponse::NotFound | SetExpiryResponse::Rejected => Err(Error::new( + ErrorKind::BackendFailure, + "long-term backend did not confirm the expiry of a completed resumable upload", + )), + } + } + /// Returns the name of the backend corresponding to the given routing choice. fn backend_type(&self, choice: &BackendChoice) -> &'static str { match choice { @@ -445,7 +487,11 @@ type LongTermBackendToken = BackendToken; struct TieredResumableToken { inner: LongTermBackendToken, revision: String, - time_expires: Option, + /// Expiration policy requested for the object, applied again from the commit time. + /// + /// Required, so tokens issued before this field existed fail to decode instead of + /// committing without an expiry. + expiration_policy: ExpirationPolicy, } impl TieredResumableToken { @@ -527,7 +573,7 @@ impl Backend for TieredStorage { let token = TieredResumableToken { revision: revision.key, inner, - time_expires: metadata.time_expires, + expiration_policy: metadata.expiration_policy, }; Ok(Some(serde_json::to_string(&token).context( ErrorKind::Internal, @@ -550,7 +596,7 @@ impl Backend for TieredStorage { backend_type = self.backend_type(&BackendChoice::LongTerm), ); let tiered = TieredResumableToken::decode(&session.backend_token)?; - let time_expires = tiered.time_expires; + let expires_in = tiered.expiration_policy.expires_in(); let inner_session = tiered.into_inner_session(session); let id = &session.object_id; let revision = &inner_session.object_id; @@ -635,6 +681,10 @@ impl Backend for TieredStorage { } guard.advance(ChangePhase::Written); + // The blob must carry the final deadline before the tombstone makes it reachable. On + // failure, the guard reclaims the unreferenced blob. + let time_expires = self.commit_upload_expiry(revision, expires_in).await?; + let written = self .inner .high_volume @@ -826,6 +876,9 @@ impl Backend for TieredStorage { target: ExpiryUpdate, access_time: Timestamp, ) -> Result { + // Only resumable commits update expired blobs, and they call the long-term backend + // directly. Logical objects stay unreachable once expired. + target.ensure_unexpired_only()?; match self .inner .high_volume @@ -1439,8 +1492,8 @@ mod tests { Ok(()) } - #[tokio::test] - async fn resumable_bigtable_and_gcs() -> anyhow::Result<()> { + /// Builds tiered storage over the Bigtable and GCS emulators. + async fn make_bigtable_and_gcs_storage() -> anyhow::Result { let streams = ChangeStreamFactory::default(); let lt = GcsBackend::new( GcsConfig { @@ -1465,7 +1518,16 @@ mod tests { &streams, ) .await?; - let storage = TieredStorage::new(Box::new(hv), Box::new(lt), Box::new(NoopChangeLog)); + Ok(TieredStorage::new( + Box::new(hv), + Box::new(lt), + Box::new(NoopChangeLog), + )) + } + + #[tokio::test] + async fn resumable_bigtable_and_gcs() -> anyhow::Result<()> { + let storage = make_bigtable_and_gcs_storage().await?; let id = make_id(&format!("tiered-resumable-{}", uuid::Uuid::now_v7())); let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1]; let token = @@ -1620,6 +1682,170 @@ mod tests { assert!(!hv.contains(&id)); } + /// TTL metadata as resolved by a session created at `created`. + fn ttl_metadata(created: Timestamp, ttl: Duration) -> Metadata { + Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(ttl), + time_created: Some(created), + time_expires: Some(created + ttl), + ..Default::default() + } + } + + /// Uploads `payload` in a single final chunk and returns the upload's revision. + async fn upload_resumable( + storage: &TieredStorage, + id: &ObjectId, + metadata: &Metadata, + payload: &[u8], + ) -> Result { + let token = resumable_token(storage, id, metadata, payload.len() as u64).await; + let progress = storage + .put_chunk( + &token, + 0, + payload.len() as u64, + stream::single(payload.to_vec()), + ) + .await?; + assert_eq!(progress, UploadProgress::Complete); + Ok(upload_revision(&token)) + } + + #[tokio::test] + async fn resumable_expiry_counts_from_commit() { + let (storage, hv, lt, _) = make_tiered_storage(); + let id = make_id("resumable-expiry"); + let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1]; + // The session started long enough ago that its initial deadline has passed. + let metadata = ttl_metadata( + Timestamp::now() - Duration::from_hours(3), + Duration::from_hours(1), + ); + + let before = Timestamp::now(); + let revision = upload_resumable(&storage, &id, &metadata, &payload) + .await + .unwrap(); + let after = Timestamp::now(); + + let deadline = hv.get(&id).expect_tombstone().time_expires.unwrap(); + assert!(deadline >= before + Duration::from_hours(1)); + assert!(deadline <= after + Duration::from_hours(1)); + let (blob, _) = lt.get(&revision).expect_object(); + assert_eq!(blob.time_expires, Some(deadline)); + // The session's creation time is kept, so the TTL covers the upload time too. + assert_eq!(blob.time_created, metadata.time_created); + assert_eq!( + blob.expiration_policy, + ExpirationPolicy::TimeToLive(Duration::from_hours(4)) + ); + + let (metadata, _, body) = storage + .get_object(&id, Timestamp::now(), None) + .await + .unwrap() + .unwrap(); + assert_eq!(metadata.time_expires, Some(deadline)); + assert_eq!(stream::read_to_vec(body).await.unwrap(), payload); + } + + #[tokio::test] + async fn resumable_manual_expiry_commits_without_deadline() { + let (storage, hv, lt, _) = make_tiered_storage(); + let id = make_id("resumable-manual"); + let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1]; + + let revision = upload_resumable(&storage, &id, &Metadata::default(), &payload) + .await + .unwrap(); + + assert_eq!(hv.get(&id).expect_tombstone().time_expires, None); + assert_eq!(lt.get(&revision).expect_object().0.time_expires, None); + } + + #[tokio::test] + async fn resumable_expiry_counts_from_commit_bigtable_and_gcs() -> anyhow::Result<()> { + let storage = make_bigtable_and_gcs_storage().await?; + let id = make_id(&format!("tiered-resumable-expiry-{}", uuid::Uuid::now_v7())); + let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1]; + let metadata = ttl_metadata( + Timestamp::now() - Duration::from_hours(3), + Duration::from_hours(1), + ); + + let before = Timestamp::now(); + let revision = upload_resumable(&storage, &id, &metadata, &payload).await?; + let after = Timestamp::now(); + + let metadata = storage + .get_metadata(&id, Timestamp::now()) + .await? + .expect("committed upload is readable"); + let deadline = metadata.time_expires.unwrap(); + assert!(deadline >= before + Duration::from_hours(1)); + assert!(deadline <= after + Duration::from_hours(1)); + // GCS reports the finalize time as creation time, so the TTL stays as requested. + assert_eq!( + metadata.expiration_policy, + ExpirationPolicy::TimeToLive(Duration::from_hours(1)) + ); + + let blob = storage + .inner + .long_term + .get_metadata(&revision, Timestamp::now()) + .await? + .unwrap(); + assert_eq!(blob.time_expires, Some(deadline)); + Ok(()) + } + + #[test] + fn resumable_token_requires_expiration_policy() { + let legacy = r#"{"inner":"token","revision":"key/rev","time_expires":null}"#; + assert_eq!( + TieredResumableToken::decode(&legacy.to_owned()) + .unwrap_err() + .kind(), + ErrorKind::UnknownUploadSession + ); + } + + #[tokio::test] + async fn resumable_commit_requires_confirmed_expiry() { + let (storage, hv, _, events) = tiered_with_expiry_hooks(false, true); + let id = make_id("resumable-expiry-rejected"); + let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1]; + let metadata = ttl_metadata(Timestamp::now(), Duration::from_hours(1)); + + let error = upload_resumable(&storage, &id, &metadata, &payload) + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::BackendFailure); + assert_eq!(events.lock().unwrap().as_slice(), &["lt"]); + assert!( + !hv.inner.contains(&id), + "no tombstone without a confirmed deadline" + ); + } + + #[tokio::test] + async fn set_expiry_rejects_include_expired() { + let (storage, _, _, _) = make_tiered_storage(); + let id = make_id("tiered-include-expired"); + let update = ExpiryUpdate { + include_expired: true, + ..ExpiryTarget::At(Timestamp::now() + Duration::from_hours(1)).into() + }; + + let error = storage + .set_expiry(&id, update, Timestamp::now()) + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::Unsupported); + } + #[derive(Clone, Debug)] struct ExpiryHook { label: &'static str, @@ -1750,8 +1976,8 @@ mod tests { .await .unwrap(); let update = ExpiryUpdate { - target: ExpiryTarget::FromCreation(Duration::from_hours(1)), max: Some(Duration::ZERO), + ..ExpiryTarget::FromCreation(Duration::from_hours(1)).into() }; assert_eq!( storage