diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index 4db526fe..4b72bbea 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 145ab78c..23800cb6 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 eba00463..454b7a9c 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 6340c567..74efb33f 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 25b3d944..8f1d80df 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 54b65019..bb7dd144 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 f38a6cf9..91720c10 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 d8c0e645..fd6167a6 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 dc4b2597..161fc396 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