Skip to content

Commit 085c0db

Browse files
fix(service): Start resumable upload expiry at commit
Co-Authored-By: Lorenzo Cian <17258265+lcian@users.noreply.github.com>
1 parent 96d9a96 commit 085c0db

1 file changed

Lines changed: 212 additions & 7 deletions

File tree

‎objectstore-service/src/backend/tiered.rs‎

Lines changed: 212 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,13 @@
107107
//! as the source of truth for whether the upload logically exists.
108108
//! This marker is consumed upon cancellation or the first submission of a final chunk, making
109109
//! finalization one-shot.
110+
//!
111+
//! Time-based expiration of a resumable upload counts from its commit. Until then, the long-term
112+
//! blob carries a placeholder deadline of at least seven days, which outlives the marker, so the
113+
//! blob cannot expire before the upload completes. At commit, the blob is extended if the
114+
//! requested deadline is later, and only then is the tombstone committed with that deadline.
115+
//! A blob whose placeholder exceeds the requested deadline stays unreachable once the tombstone
116+
//! expires and is reclaimed when its own deadline passes.
110117
111118
use std::num::NonZeroU64;
112119
use std::sync::Arc;
@@ -146,6 +153,11 @@ const BACKEND_SIZE_THRESHOLD: usize = 1024 * 1024; // 1 MiB
146153
/// Fixed lifetime of an ongoing upload marker.
147154
const RESUMABLE_UPLOAD_TTL: Duration = Duration::from_hours(5 * 24);
148155

156+
/// Minimum lifetime of a resumable upload's long-term blob before it is committed.
157+
///
158+
/// Longer than [`RESUMABLE_UPLOAD_TTL`], so the blob outlives any session that can still commit.
159+
const RESUMABLE_BLOB_MIN_TTL: Duration = Duration::from_hours(7 * 24);
160+
149161
/// Amount of time for which a `Change` generated by a `complete_multipart` operation is kept in the `Assembling`
150162
/// state before becoming eligible for cleanup by the `ChangeLog` recovery process.
151163
/// This allows the client to retry the `complete_multipart` operation upon any failures for at least this long,
@@ -304,6 +316,39 @@ impl TieredStorage {
304316
Ok(())
305317
}
306318

319+
/// Resolves the deadline of a completed resumable upload, counted from its commit.
320+
///
321+
/// Extends the long-term blob if `now + expires_in` is later than its placeholder deadline,
322+
/// and returns the deadline the redirect tombstone must carry. Returns `None` without a
323+
/// time-based policy.
324+
///
325+
/// Fails if the long-term backend does not confirm the deadline, so the caller never commits a
326+
/// tombstone that could outlive its blob.
327+
async fn commit_upload_expiry(
328+
&self,
329+
revision: &ObjectId,
330+
expires_in: Option<Duration>,
331+
) -> Result<Option<Timestamp>> {
332+
let Some(expires_in) = expires_in else {
333+
return Ok(None);
334+
};
335+
336+
let now = Timestamp::now();
337+
let target = ExpiryTarget::At(now.saturating_add(expires_in));
338+
match self
339+
.inner
340+
.long_term
341+
.set_expiry(revision, target.into(), now)
342+
.await?
343+
{
344+
SetExpiryResponse::Satisfied(deadline) => Ok(Some(deadline)),
345+
SetExpiryResponse::NotFound | SetExpiryResponse::Rejected => Err(Error::new(
346+
ErrorKind::BackendFailure,
347+
"long-term backend did not confirm the expiry of a completed resumable upload",
348+
)),
349+
}
350+
}
351+
307352
/// Returns the name of the backend corresponding to the given routing choice.
308353
fn backend_type(&self, choice: &BackendChoice) -> &'static str {
309354
match choice {
@@ -445,7 +490,10 @@ type LongTermBackendToken = BackendToken;
445490
struct TieredResumableToken {
446491
inner: LongTermBackendToken,
447492
revision: String,
448-
time_expires: Option<Timestamp>,
493+
/// Duration of a time-based expiration policy, applied from the commit time.
494+
///
495+
/// `None` for manual expiration.
496+
expires_in: Option<Duration>,
449497
}
450498

451499
impl TieredResumableToken {
@@ -492,10 +540,17 @@ impl Backend for TieredStorage {
492540
}
493541

494542
let revision = new_long_term_revision(id);
543+
// The commit sets the real deadline, so the blob must not expire before then.
544+
let blob_metadata = Metadata {
545+
time_expires: metadata
546+
.time_expires
547+
.map(|expiry| expiry.max(Timestamp::now().saturating_add(RESUMABLE_BLOB_MIN_TTL))),
548+
..metadata.clone()
549+
};
495550
let Some(inner) = self
496551
.inner
497552
.long_term
498-
.create_upload_session(&revision, metadata, upload_length)
553+
.create_upload_session(&revision, &blob_metadata, upload_length)
499554
.await?
500555
else {
501556
return Ok(None);
@@ -527,7 +582,7 @@ impl Backend for TieredStorage {
527582
let token = TieredResumableToken {
528583
revision: revision.key,
529584
inner,
530-
time_expires: metadata.time_expires,
585+
expires_in: metadata.expiration_policy.expires_in(),
531586
};
532587
Ok(Some(serde_json::to_string(&token).context(
533588
ErrorKind::Internal,
@@ -550,7 +605,7 @@ impl Backend for TieredStorage {
550605
backend_type = self.backend_type(&BackendChoice::LongTerm),
551606
);
552607
let tiered = TieredResumableToken::decode(&session.backend_token)?;
553-
let time_expires = tiered.time_expires;
608+
let expires_in = tiered.expires_in;
554609
let inner_session = tiered.into_inner_session(session);
555610
let id = &session.object_id;
556611
let revision = &inner_session.object_id;
@@ -635,6 +690,10 @@ impl Backend for TieredStorage {
635690
}
636691
guard.advance(ChangePhase::Written);
637692

693+
// The blob must cover the final deadline before the tombstone makes it reachable. On
694+
// failure, the guard reclaims the unreferenced blob.
695+
let time_expires = self.commit_upload_expiry(revision, expires_in).await?;
696+
638697
let written = self
639698
.inner
640699
.high_volume
@@ -1439,8 +1498,8 @@ mod tests {
14391498
Ok(())
14401499
}
14411500

1442-
#[tokio::test]
1443-
async fn resumable_bigtable_and_gcs() -> anyhow::Result<()> {
1501+
/// Builds tiered storage over the Bigtable and GCS emulators.
1502+
async fn make_bigtable_and_gcs_storage() -> anyhow::Result<TieredStorage> {
14441503
let streams = ChangeStreamFactory::default();
14451504
let lt = GcsBackend::new(
14461505
GcsConfig {
@@ -1465,7 +1524,16 @@ mod tests {
14651524
&streams,
14661525
)
14671526
.await?;
1468-
let storage = TieredStorage::new(Box::new(hv), Box::new(lt), Box::new(NoopChangeLog));
1527+
Ok(TieredStorage::new(
1528+
Box::new(hv),
1529+
Box::new(lt),
1530+
Box::new(NoopChangeLog),
1531+
))
1532+
}
1533+
1534+
#[tokio::test]
1535+
async fn resumable_bigtable_and_gcs() -> anyhow::Result<()> {
1536+
let storage = make_bigtable_and_gcs_storage().await?;
14691537
let id = make_id(&format!("tiered-resumable-{}", uuid::Uuid::now_v7()));
14701538
let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1];
14711539
let token =
@@ -1620,6 +1688,143 @@ mod tests {
16201688
assert!(!hv.contains(&id));
16211689
}
16221690

1691+
/// TTL metadata as resolved by a session created at `created`.
1692+
fn ttl_metadata(created: Timestamp, ttl: Duration) -> Metadata {
1693+
Metadata {
1694+
expiration_policy: ExpirationPolicy::TimeToLive(ttl),
1695+
time_created: Some(created),
1696+
time_expires: Some(created + ttl),
1697+
..Default::default()
1698+
}
1699+
}
1700+
1701+
/// Uploads `payload` in a single final chunk and returns the upload's revision.
1702+
async fn upload_resumable(
1703+
storage: &TieredStorage,
1704+
id: &ObjectId,
1705+
metadata: &Metadata,
1706+
payload: &[u8],
1707+
) -> Result<ObjectId> {
1708+
let token = resumable_token(storage, id, metadata, payload.len() as u64).await;
1709+
let progress = storage
1710+
.put_chunk(
1711+
&token,
1712+
0,
1713+
payload.len() as u64,
1714+
stream::single(payload.to_vec()),
1715+
)
1716+
.await?;
1717+
assert_eq!(progress, UploadProgress::Complete);
1718+
Ok(upload_revision(&token))
1719+
}
1720+
1721+
#[tokio::test]
1722+
async fn resumable_expiry_counts_from_commit() {
1723+
let (storage, hv, lt, _) = make_tiered_storage();
1724+
let id = make_id("resumable-expiry");
1725+
let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1];
1726+
// The session was created long enough ago that its own TTL deadline has passed.
1727+
let metadata = ttl_metadata(
1728+
Timestamp::now() - Duration::from_hours(3),
1729+
Duration::from_hours(1),
1730+
);
1731+
1732+
let before = Timestamp::now();
1733+
let revision = upload_resumable(&storage, &id, &metadata, &payload)
1734+
.await
1735+
.unwrap();
1736+
let after = Timestamp::now();
1737+
1738+
let deadline = hv.get(&id).expect_tombstone().time_expires.unwrap();
1739+
assert!(deadline >= before + Duration::from_hours(1));
1740+
assert!(deadline <= after + Duration::from_hours(1));
1741+
// The placeholder already covers the deadline, so the blob keeps it.
1742+
let blob_expiry = lt.get(&revision).expect_object().0.time_expires.unwrap();
1743+
assert!(blob_expiry >= before + RESUMABLE_BLOB_MIN_TTL);
1744+
1745+
let (metadata, _, body) = storage
1746+
.get_object(&id, Timestamp::now(), None)
1747+
.await
1748+
.unwrap()
1749+
.unwrap();
1750+
assert_eq!(metadata.time_expires, Some(deadline));
1751+
assert_eq!(stream::read_to_vec(body).await.unwrap(), payload);
1752+
}
1753+
1754+
#[tokio::test]
1755+
async fn resumable_expiry_extends_blob_beyond_placeholder() {
1756+
let (storage, hv, lt, _) = make_tiered_storage();
1757+
let id = make_id("resumable-expiry-long");
1758+
let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1];
1759+
let ttl = RESUMABLE_BLOB_MIN_TTL + Duration::from_hours(24);
1760+
let metadata = ttl_metadata(Timestamp::now() - Duration::from_secs(5), ttl);
1761+
1762+
let revision = upload_resumable(&storage, &id, &metadata, &payload)
1763+
.await
1764+
.unwrap();
1765+
1766+
let deadline = hv.get(&id).expect_tombstone().time_expires.unwrap();
1767+
assert!(deadline > metadata.time_expires.unwrap());
1768+
assert_eq!(
1769+
lt.get(&revision).expect_object().0.time_expires,
1770+
Some(deadline)
1771+
);
1772+
}
1773+
1774+
#[tokio::test]
1775+
async fn resumable_expiry_counts_from_commit_bigtable_and_gcs() -> anyhow::Result<()> {
1776+
let storage = make_bigtable_and_gcs_storage().await?;
1777+
let id = make_id(&format!("tiered-resumable-expiry-{}", uuid::Uuid::now_v7()));
1778+
let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1];
1779+
let metadata = ttl_metadata(
1780+
Timestamp::now() - Duration::from_hours(3),
1781+
Duration::from_hours(1),
1782+
);
1783+
1784+
let before = Timestamp::now();
1785+
let revision = upload_resumable(&storage, &id, &metadata, &payload).await?;
1786+
let after = Timestamp::now();
1787+
1788+
let metadata = storage
1789+
.get_metadata(&id, Timestamp::now())
1790+
.await?
1791+
.expect("committed upload is readable");
1792+
let deadline = metadata.time_expires.unwrap();
1793+
assert!(deadline >= before + Duration::from_hours(1));
1794+
assert!(deadline <= after + Duration::from_hours(1));
1795+
assert_eq!(
1796+
metadata.expiration_policy,
1797+
ExpirationPolicy::TimeToLive(Duration::from_hours(1))
1798+
);
1799+
1800+
let blob = storage
1801+
.inner
1802+
.long_term
1803+
.get_metadata(&revision, Timestamp::now())
1804+
.await?
1805+
.unwrap();
1806+
assert!(blob.time_expires.unwrap() >= before + RESUMABLE_BLOB_MIN_TTL);
1807+
Ok(())
1808+
}
1809+
1810+
#[tokio::test]
1811+
async fn resumable_commit_requires_confirmed_expiry() {
1812+
let (storage, hv, _, events) = tiered_with_expiry_hooks(false, true);
1813+
let id = make_id("resumable-expiry-rejected");
1814+
let payload = vec![b'a'; BACKEND_SIZE_THRESHOLD + 1];
1815+
let metadata = ttl_metadata(Timestamp::now(), Duration::from_hours(1));
1816+
1817+
let error = upload_resumable(&storage, &id, &metadata, &payload)
1818+
.await
1819+
.unwrap_err();
1820+
assert_eq!(error.kind(), ErrorKind::BackendFailure);
1821+
assert_eq!(events.lock().unwrap().as_slice(), &["lt"]);
1822+
assert!(
1823+
!hv.inner.contains(&id),
1824+
"no tombstone without a confirmed deadline"
1825+
);
1826+
}
1827+
16231828
#[derive(Clone, Debug)]
16241829
struct ExpiryHook {
16251830
label: &'static str,

0 commit comments

Comments
 (0)