From aaad2fb9864facff4e75a37fa4acca58d53ffbec Mon Sep 17 00:00:00 2001 From: Reinaldy Rafli Date: Sun, 1 Nov 2026 20:51:03 +0700 Subject: [PATCH] feat(change-stream): generic garbage collector --- objectstore-service/src/change_stream/gc.rs | 233 +++++++++++++++++++ objectstore-service/src/change_stream/mod.rs | 1 + 2 files changed, 234 insertions(+) create mode 100644 objectstore-service/src/change_stream/gc.rs diff --git a/objectstore-service/src/change_stream/gc.rs b/objectstore-service/src/change_stream/gc.rs new file mode 100644 index 00000000..92e7ad0f --- /dev/null +++ b/objectstore-service/src/change_stream/gc.rs @@ -0,0 +1,233 @@ +//! A [`ChangeStream`] that keeps the garbage collector's expiration index in sync. +//! +//! The garbage collector deletes objects once their deadline has passed, so it needs to +//! know about every object the backends write and every object they delete explicitly. +//! [`GcChangeStream`] forwards the relevant lifecycle events to a [`GcStorage`]. + +use std::sync::Arc; +use std::time::Duration; + +use objectstore_types::time::Timestamp; +use tonic::async_trait; + +use crate::change_stream::{ChangeStream, ChangeStreamError}; +use crate::garbage_collector::GcStorage; +use crate::id::ObjectId; + +/// Mirrors object lifetimes into [`GcStorage`] for the garbage collector. +/// +/// Writes and updates are announced through [`begin_write`](ChangeStream::begin_write) and +/// [`begin_update`](ChangeStream::begin_update); both record `expires_at` because the +/// collector only needs the current deadline, and [`GcStorage::upsert`] already implements +/// the required "a shorter deadline never replaces a longer one" semantics. Explicit +/// deletions are handled in [`commit_delete`](ChangeStream::commit_delete). +/// +/// The `commit_*` hooks for writes and updates are intentionally left as no-ops: the +/// deadline is recorded before the storage operation runs, so an object cannot be written +/// without a corresponding index entry. Recording it earlier only means a failed write may +/// leave a deadline for an object that never existed, which the collector can safely ignore +/// (and which a later, longer write overwrites). +/// +/// All storage calls are awaited inline, so [`join`](ChangeStream::join) returns +/// immediately; there is no buffered work to drain. +#[derive(Debug)] +pub struct GcChangeStream { + /// Expiration index that receives the forwarded changes. + pub storage: Arc, +} + +#[async_trait] +impl ChangeStream for GcChangeStream { + async fn begin_write( + &self, + id: &ObjectId, + expires_at: Option, + ) -> Result<(), ChangeStreamError> { + self.storage.upsert(id, expires_at).await + } + + async fn begin_update( + &self, + id: &ObjectId, + expires_at: Option, + ) -> Result<(), ChangeStreamError> { + self.storage.upsert(id, expires_at).await + } + + async fn commit_delete(&self, id: &ObjectId) { + self.storage.remove(id).await + } + + async fn join(&self, _timeout: Duration) {} +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use objectstore_types::scope::{Scope, Scopes}; + + use super::*; + use crate::id::ObjectContext; + + fn ts(secs: u64) -> Timestamp { + Timestamp::from_unix_secs(secs).unwrap() + } + + fn id_for(key: &str) -> ObjectId { + ObjectId::new( + ObjectContext { + usecase: "testing".parse().unwrap(), + scopes: Scopes::from_iter([ + Scope::create("organization", "17").unwrap(), + Scope::create("project", "42").unwrap(), + ]), + }, + key.to_owned(), + ) + } + + /// A [`GcStorage`] that records the calls it receives instead of persisting anything. + #[derive(Debug, Default)] + struct RecordingStorage { + calls: Mutex>, + fail_upsert: bool, + } + + #[derive(Debug, PartialEq)] + enum Call { + Upsert(ObjectId, Option), + Remove(ObjectId), + } + + impl RecordingStorage { + fn calls(&self) -> Vec { + std::mem::take(&mut *self.calls.lock().unwrap()) + } + } + + #[async_trait] + impl GcStorage for RecordingStorage { + async fn upsert( + &self, + id: &ObjectId, + expires_at: Option, + ) -> Result<(), ChangeStreamError> { + if self.fail_upsert { + return Err(ChangeStreamError::Unavailable("storage down".into())); + } + self.calls + .lock() + .unwrap() + .push(Call::Upsert(id.clone(), expires_at)); + Ok(()) + } + + async fn remove(&self, id: &ObjectId) { + self.calls.lock().unwrap().push(Call::Remove(id.clone())); + } + + async fn list_expired( + &self, + _now: Timestamp, + _limit: u64, + ) -> Result, ChangeStreamError> { + Ok(Vec::new()) + } + } + + fn stream(storage: Arc) -> GcChangeStream { + GcChangeStream { + storage: storage as Arc, + } + } + + #[tokio::test] + async fn begin_write_records_expiration() { + let storage = Arc::new(RecordingStorage::default()); + let id = id_for("written"); + + stream(storage.clone()) + .begin_write(&id, Some(ts(100))) + .await + .unwrap(); + + assert_eq!(storage.calls(), vec![Call::Upsert(id, Some(ts(100)))]); + } + + #[tokio::test] + async fn begin_update_records_expiration() { + let storage = Arc::new(RecordingStorage::default()); + let id = id_for("updated"); + + stream(storage.clone()) + .begin_update(&id, Some(ts(200))) + .await + .unwrap(); + + assert_eq!(storage.calls(), vec![Call::Upsert(id, Some(ts(200)))]); + } + + #[tokio::test] + async fn begin_hooks_forward_cleared_expiration() { + let storage = Arc::new(RecordingStorage::default()); + let id = id_for("permanent"); + + stream(storage.clone()) + .begin_update(&id, None) + .await + .unwrap(); + + assert_eq!(storage.calls(), vec![Call::Upsert(id, None)]); + } + + #[tokio::test] + async fn commits_do_not_touch_storage() { + let storage = Arc::new(RecordingStorage::default()); + let stream = stream(storage.clone()); + let id = id_for("committed"); + + stream.commit_write(&id, 42, Some(ts(100))).await; + stream.commit_update(&id, Some(ts(200))).await; + + assert!(storage.calls().is_empty()); + } + + #[tokio::test] + async fn commit_delete_removes_object() { + let storage = Arc::new(RecordingStorage::default()); + let id = id_for("deleted"); + + stream(storage.clone()).commit_delete(&id).await; + + assert_eq!(storage.calls(), vec![Call::Remove(id)]); + } + + #[tokio::test] + async fn begin_hooks_propagate_storage_failures() { + let storage = Arc::new(RecordingStorage { + fail_upsert: true, + ..Default::default() + }); + let stream = stream(storage.clone()); + let id = id_for("failing"); + + let write = stream.begin_write(&id, Some(ts(100))).await; + let update = stream.begin_update(&id, Some(ts(100))).await; + + // A failing `begin_*` must fail the request before storage is touched. + assert!(matches!(write, Err(ChangeStreamError::Unavailable(_)))); + assert!(matches!(update, Err(ChangeStreamError::Unavailable(_)))); + assert!(storage.calls().is_empty()); + } + + #[tokio::test] + async fn join_returns_without_waiting_for_timeout() { + let stream = stream(Arc::new(RecordingStorage::default())); + + // `join` has nothing buffered, so it must return well before the timeout it is given. + tokio::time::timeout(Duration::from_secs(1), stream.join(Duration::from_secs(60))) + .await + .expect("join should return immediately"); + } +} diff --git a/objectstore-service/src/change_stream/mod.rs b/objectstore-service/src/change_stream/mod.rs index 37f11790..f95e014a 100644 --- a/objectstore-service/src/change_stream/mod.rs +++ b/objectstore-service/src/change_stream/mod.rs @@ -20,6 +20,7 @@ use crate::id::ObjectId; #[cfg(feature = "storage-cogs")] mod cost_tracker; mod factory; +pub mod gc; #[cfg(feature = "storage-cogs")] pub use cost_tracker::CostTrackerStream;