feat(gc): Implement SQLite-backed GcStorage

Cargo.toml

@@ -43,7 +43,10 @@ async-trait = "0.1.92"
 axum = "0.8.9"
 axum-extra = { version = "0.12.6", default-features = false }
 base64 = { version = "0.23.1", default-features = false, features = ["std"] }
-bigtable_rs = { git = "https://github.com/getsentry/bigtable_rs.git", rev = "072cb5871766b39c08c9a4d1f43c8d3895eeb96e", default-features = false, features = ["gcp-auth-ring", "tonic-tls-ring"] }
+bigtable_rs = { git = "https://github.com/getsentry/bigtable_rs.git", rev = "072cb5871766b39c08c9a4d1f43c8d3895eeb96e", default-features = false, features = [
+    "gcp-auth-ring",
+    "tonic-tls-ring",
+] }
 blake3 = "1.8.5"
 bytes = "1.12.0"
 bytesize = "2.4.0"
@@ -104,12 +107,16 @@ serde = { version = "1.0.228", features = ["derive"] }
 serde-vars = "0.4.0"
 serde_json = "1.0.150"
 serde_yaml = "0.9.34-deprecated"
+sqlx = "0.9.0"
 sketches-ddsketch = "0.4.2"
 tempfile = "3.27.0"
 thiserror = "2.0.18"
 thread_local = "1.1.9"
 jemalloc_pprof = "0.9.0"
-tikv-jemallocator = { version = "0.7.0", features = ["background_threads", "override_allocator_on_supported_platforms"] }
+tikv-jemallocator = { version = "0.7.0", features = [
+    "background_threads",
+    "override_allocator_on_supported_platforms",
+] }
 tikv-jemalloc-ctl = { version = "0.7.0", features = ["stats"] }
 tokio = "1.52.3"
 tokio-util = "0.7.18"

migrations/sqlite/0001_garbage_collector.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS garbage_collector (
+    object_id TEXT NOT NULL PRIMARY KEY,
+    expires_at INTEGER,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX idx_garbage_collector_expires_at
+    ON garbage_collector (expires_at);

objectstore-service/Cargo.toml

@@ -19,18 +19,35 @@ chrono = { workspace = true }
 futures-util = { workspace = true }
 gcp_auth = { workspace = true }
 humantime-serde = { workspace = true }
-objectstore-inventory-tracker = { workspace = true, features = ["kafka"], optional = true }
+objectstore-inventory-tracker = { workspace = true, features = [
+    "kafka",
+], optional = true }
 objectstore-log = { workspace = true }
 objectstore-metrics = { workspace = true }
 objectstore-types = { workspace = true }
 papaya = { workspace = true }
 quick-xml = { workspace = true, features = ["serialize"] }
 regex = { workspace = true }
-reqwest = { workspace = true, features = ["charset", "http2", "hickory-dns", "json", "multipart", "native-tls-no-alpn", "stream", "system-proxy"] }
+reqwest = { workspace = true, features = [
+    "charset",
+    "http2",
+    "hickory-dns",
+    "json",
+    "multipart",
+    "native-tls-no-alpn",
+    "stream",
+    "system-proxy",
+] }
 ring = { workspace = true }
 sentry = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+sqlx = { workspace = true, features = [
+    "runtime-tokio",
+    "tls-rustls-ring-native-roots",
+    "chrono",
+    "sqlite",
+] }
 tempfile = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

objectstore-service/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    println!("cargo:rerun-if-changed=migrations");
+}

objectstore-service/src/garbage_collector/mod.rs

@@ -0,0 +1,51 @@
+//! Persistence of object expiration metadata, read by the garbage collector to
+//! find objects whose lifetime has elapsed.
+//!
+//! [`GcStorage`] is the backend-agnostic interface. Implementations record the
+//! expiration of each object when it is written and hand back the expired ids
+//! that the collector is allowed to delete.
+
+/// [`GcStorage`] implementation backed by SQLite.
+pub mod sqlite;
+
+use std::fmt::Debug;
+
+use objectstore_types::time::Timestamp;
+
+use crate::change_stream::ChangeStreamError;
+use crate::id::ObjectId;
+
+/// Stores object expiration metadata for the garbage collector.
+///
+/// Implementations are called on the write path (via [`upsert`](Self::upsert)
+/// and [`remove`](Self::remove)) and by the collector itself (via
+/// [`list_expired`](Self::list_expired)), so they must be safe to share across
+/// tasks.
+#[async_trait::async_trait]
+pub trait GcStorage: Send + Sync + Debug {
+    /// Records that `id` expires at `expires_at`, replacing any previous deadline.
+    ///
+    /// A later deadline replaces an earlier one; a shorter deadline is ignored, so concurrent
+    /// writers cannot pull an expiration forward. `None` clears the recorded deadline and marks
+    /// the object as never expiring.
+    async fn upsert(
+        &self,
+        id: &ObjectId,
+        expires_at: Option<Timestamp>,
+    ) -> Result<(), ChangeStreamError>;
+
+    /// Forgets `id` entirely.
+    ///
+    /// Best-effort: failures are not reported to the caller.
+    async fn remove(&self, id: &ObjectId);
+
+    /// Returns up to `limit` ids whose expiration is at or before `now`,
+    /// ordered by earliest expiry first.
+    ///
+    /// Objects without an expiration are never returned.
+    async fn list_expired(
+        &self,
+        now: Timestamp,
+        limit: u64,
+    ) -> Result<Vec<ObjectId>, ChangeStreamError>;
+}