Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion objectstore-server/src/endpoints/objects.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
3 changes: 3 additions & 0 deletions objectstore-service/docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions objectstore-service/src/backend/bigtable.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1298,6 +1298,7 @@ impl HighVolumeBackend for BigTableBackend {
access_time: Timestamp,
) -> Result<SetExpiryResponse> {
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
Expand Down
33 changes: 30 additions & 3 deletions objectstore-service/src/backend/common.rs
Original file line number Diff line number Diff line change
Expand Up @@ -92,15 +92,41 @@ pub struct ExpiryUpdate {
pub target: ExpiryTarget,
/// Maximum remaining lifetime. `None` means no limit.
pub max: Option<Duration>,
/// 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<ExpiryTarget> 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.
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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
Expand Down
74 changes: 63 additions & 11 deletions objectstore-service/src/backend/gcs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Option<GcsObject>> {
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<Option<GcsObject>> {
let metadata_opt = self
.with_retry("get_metadata", || async {
let resp = self
Expand All @@ -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.
Expand Down Expand Up @@ -1082,7 +1093,12 @@ impl Backend for GcsBackend {
access_time: Timestamp,
) -> Result<SetExpiryResponse> {
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 {
Expand Down Expand Up @@ -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] {
Expand Down
68 changes: 59 additions & 9 deletions objectstore-service/src/backend/in_memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<ExpiryOutcome> + Send + Sync,
) -> Result<SetExpiryResponse> {
let try_extend = |store: &Store| -> Result<(ExpiryOutcome, Option<StoreEntry>)> {
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)))
Expand Down Expand Up @@ -268,10 +270,17 @@ impl super::common::Backend for InMemoryBackend {
target: ExpiryUpdate,
access_time: Timestamp,
) -> Result<SetExpiryResponse> {
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
}

Expand Down Expand Up @@ -546,7 +555,8 @@ impl HighVolumeBackend for InMemoryBackend {
access_time: Timestamp,
) -> Result<SetExpiryResponse> {
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)
}
Expand Down Expand Up @@ -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<Timestamp>,
Expand All @@ -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
}

Expand Down Expand Up @@ -1245,7 +1257,7 @@ mod tests {

let restricted = ExpiryUpdate {
max: Some(Duration::ZERO),
target,
..target.into()
};
assert_eq!(
backend
Expand All @@ -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(),
Expand Down Expand Up @@ -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();
Expand Down
Loading
Loading