feat: keeper trait & sqlite-backed implementation

Cargo.toml

@@ -84,12 +84,16 @@ sentry-options = "1.2.4"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.150"
 serde_yaml = "0.9.34-deprecated"
+sqlx = "0.9.0"
 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,
+    time_created INTEGER NOT NULL,
+    time_expires INTEGER
+);

objectstore-service/Cargo.toml

@@ -29,6 +29,7 @@ reqwest = { workspace = true, features = ["charset", "http2", "hickory-dns", "js
 sentry = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+sqlx = { workspace = true, features = ["runtime-tokio", "tls-rustls-ring-native-roots", "sqlite", "chrono"] }
 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,33 @@
+//! 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<()>;
+
+    /// Update an object to a new expiration policy (and thus, new expiration time).
+    async fn update(&self, id: &ObjectId, expiration_policy: ExpirationPolicy) -> Result<()>;
+}