feat: keeper trait & sqlite-backed implementation

Cargo.toml

@@ -83,12 +83,21 @@ sentry-options = "1.2.1"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.150"
 serde_yaml = "0.9.34-deprecated"
+sqlx = { version = "0.9.0", features = [
+    "runtime-tokio",
+    "tls-rustls-ring-native-roots",
+    "sqlite",
+    "chrono",
+] }
 sketches-ddsketch = "0.3.1"
 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_keeper.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS ttl_keeper (
+    object_id TEXT PRIMARY KEY,
+    expiration_policy INTEGER NOT NULL DEFAULT 0,
+    duration INTEGER,
+    created_at INTEGER NOT NULL,
+    expires_at INTEGER
+);

objectstore-server/Cargo.toml

@@ -38,17 +38,40 @@ papaya = { workspace = true }
 pin-project-lite = { workspace = true }
 percent-encoding = { workspace = true }
 rand = { workspace = true }
-reqwest = { workspace = true, features = ["charset", "http2", "system-proxy", "native-tls-no-alpn"] }
+reqwest = { workspace = true, features = [
+    "charset",
+    "http2",
+    "system-proxy",
+    "native-tls-no-alpn",
+] }
 rustls = { workspace = true }
 secrecy = { workspace = true, features = ["serde"] }
-sentry = { workspace = true, features = ["tower-axum-matched-path", "tracing", "logs"] }
+sentry = { workspace = true, features = [
+    "tower-axum-matched-path",
+    "tracing",
+    "logs",
+] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 thread_local = { workspace = true }
-tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
+tokio = { workspace = true, features = [
+    "fs",
+    "macros",
+    "net",
+    "rt-multi-thread",
+    "signal",
+    "sync",
+    "time",
+] }
 tower = { workspace = true }
-tower-http = { workspace = true, features = ["catch-panic", "metrics", "set-header", "trace"] }
+tower-http = { workspace = true, features = [
+    "catch-panic",
+    "metrics",
+    "set-header",
+    "trace",
+] }
+sqlx.workspace = true
 
 [target.'cfg(target_os = "linux")'.dependencies]
 jemalloc_pprof = { workspace = true, optional = true }

objectstore-service/Cargo.toml

@@ -25,10 +25,20 @@ objectstore-metrics = { workspace = true }
 objectstore-types = { 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",
+] }
 sentry = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+sqlx = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true, features = ["io", "rt"] }

objectstore-service/src/error.rs

@@ -151,6 +151,14 @@ pub enum Error {
     /// Invalid upload ID (e.g. path traversal attempt).
     #[error(transparent)]
     InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId),
+
+    /// Errors from SQLx database migrations.
+    #[error("sqlx migration error: {0}")]
+    SqlxMigrate(#[from] sqlx::migrate::MigrateError),
+
+    /// Common SQLx database errors.
+    #[error("sqlx error: {0}")]
+    Sqlx(#[from] sqlx::Error),
 }
 
 impl Error {
@@ -211,6 +219,8 @@ impl Error {
             Self::NotImplemented => Level::ERROR,
             Self::InvalidUploadId(_) => Level::DEBUG,
             Self::Generic { .. } => Level::ERROR,
+            Self::SqlxMigrate(_) => Level::ERROR,
+            Self::Sqlx(_) => Level::ERROR,
         }
     }
 }

objectstore-service/src/keeper/mod.rs

@@ -0,0 +1,34 @@
+//! Keeper defines a trait to support extending an object retention for backends
+//! that does not support them natively (e.g., S3-compatible backend and filesystem).
+
+use objectstore_types::metadata::ExpirationPolicy;
+
+use crate::error::Result;
+use crate::id::ObjectId;
+
+/// SQLite-backed implementation of the [`Keeper`] trait.
+pub mod sqlite_backed;
+
+/// Represents the computed expiry information for an object.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObjectExpiry<'a> {
+    id: &'a ObjectId,
+    expiration_policy: ExpirationPolicy,
+}
+
+/// Object retention keeper trait.
+#[async_trait::async_trait]
+pub trait Keeper: Send + Sync {
+    /// Keep is the first step in the object retention lifecycle.
+    /// Practically speaking, it would not be kept if the `expiration_policy` is `Manual`.
+    /// The `expiration_policy` is set by the client at upload time via the supplied Metadata.
+    async fn keep(&self, id: &ObjectId, expiration_policy: ExpirationPolicy) -> Result<()>;
+
+    /// Remove is the final step in the object retention lifecycle.
+    /// It is called by a cleanup worker when the object is no longer needed.
+    async fn remove(&self, id: &ObjectId) -> Result<()>;
+
+    /// Marks an object as accessed. For `expiration_policy` of `TimeToIdle`, this will
+    /// extend the object retention.
+    async fn mark_accessed(&self, id: &ObjectId) -> Result<()>;
+}