Skip to content

feat(change-stream): ExpiryIndex for Sqlite backend #690

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
aldy505 wants to merge 1 commit into from
Closed

feat(change-stream): ExpiryIndex for Sqlite backend #690

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 177 additions & 1 deletion objectstore-service/src/change_stream/garbage_collector_sqlite.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use crate::change_stream::{ChangeStream, ChangeStreamError};
use crate::change_stream::{ChangeStream, ChangeStreamError, ExpiryIndex};
use crate::id::ObjectId;
use async_trait::async_trait;
use objectstore_types::time::Timestamp;
Expand Down Expand Up @@ -122,6 +122,36 @@ impl ChangeStream for SqliteGarbageCollectorStream {
}
}

#[async_trait]
impl ExpiryIndex for SqliteGarbageCollectorStream {
async fn list_expired(
&self,
now: Timestamp,
limit: u64,
) -> Result<Vec<ObjectId>, Box<dyn std::error::Error + Send + Sync>> {

@lcian lcian Oct 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Normally we use thiserror which helps creating structured errors, I would prefer if we could use it here too, it could be something like ExpiryIndexError with the different variants.

let pool = self.pool.clone();
let query_result: Result<Vec<String>, sqlx::Error> = sqlx::query_scalar::<_, String>(
"SELECT object_id FROM garbage_collector WHERE expires_at IS NOT NULL AND expires_at <= ? ORDER BY expires_at LIMIT ?",
)
.bind(i64::try_from(now.as_secs()).ok())
.bind(i64::try_from(limit).unwrap_or(i64::MAX))
.fetch_all(&pool)
.await;

match query_result {
Ok(rows) => {
let mut ids = Vec::new();
for row in rows {
let id = row.clone();
ids.push(ObjectId::from_storage_path(&id).unwrap());

@cursor cursor Bot Oct 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Expired listing panics on bad ids

Medium Severity

list_expired calls ObjectId::from_storage_path and unwraps. One stored path that does not round-trip — an empty key, empty usecase, or a row written outside this code — panics the whole listing instead of returning the Result the method already uses.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 2e5604a. Configure here.

@lcian lcian Oct 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even though this should never happen, here we should log an error and continue, probably also deleting the bad row.
We certainly don't want to panic, and we also don't want to return an Error as Cursor suggests as that would stop the whole list_expired and you would find the same bad record on the next run.

}
Ok(ids)
}
Err(error) => Err(Box::new(error)),
}
}
}

#[cfg(test)]
mod tests {
use crate::id::ObjectContext;
Expand Down Expand Up @@ -507,4 +537,150 @@ mod tests {
assert_eq!(stored_expiry(&stream, &first).await, Some(None));
assert_eq!(stored_expiry(&stream, &second).await, Some(None));
}

#[tokio::test]
async fn list_expired_returns_only_past_deadlines() {
let (_tempdir, stream) = create_test_stream().await;
let past = create_test_id("past");
let future = create_test_id("future");
let no_expiry = create_test_id("no-expiry");

stream
.begin_write(&past, Some(Timestamp::from_unix_secs(1_000).unwrap()))
.await
.unwrap();
stream
.begin_write(&future, Some(Timestamp::from_unix_secs(9_000).unwrap()))
.await
.unwrap();
stream.begin_write(&no_expiry, None).await.unwrap();

// Only the past deadline is listed: the future deadline and the NULL expiry
// are both excluded, and the returned id round-trips through the storage path.
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
vec![past.clone()]
);
}

#[tokio::test]
async fn list_expired_includes_deadline_equal_to_now() {
let (_tempdir, stream) = create_test_stream().await;
let due = create_test_id("due");
let later = create_test_id("later");

stream
.begin_write(&due, Some(Timestamp::from_unix_secs(2_000).unwrap()))
.await
.unwrap();
stream
.begin_write(&later, Some(Timestamp::from_unix_secs(2_001).unwrap()))
.await
.unwrap();

// The contract is inclusive: a deadline exactly equal to `now` is listed.
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(2_000).unwrap(), 10)
.await
.unwrap(),
vec![due.clone()]
);
}

#[tokio::test]
async fn list_expired_orders_by_oldest_deadline_and_respects_limit() {
let (_tempdir, stream) = create_test_stream().await;
let newest = create_test_id("newest");
let oldest = create_test_id("oldest");
let middle = create_test_id("middle");

// Seeded scrambled relative to deadline order on purpose.
stream
.begin_write(&newest, Some(Timestamp::from_unix_secs(3_000).unwrap()))
.await
.unwrap();
stream
.begin_write(&oldest, Some(Timestamp::from_unix_secs(1_000).unwrap()))
.await
.unwrap();
stream
.begin_write(&middle, Some(Timestamp::from_unix_secs(2_000).unwrap()))
.await
.unwrap();

// Oldest deadline first, capped at `limit`...
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 2)
.await
.unwrap(),
vec![oldest.clone(), middle.clone()]
);
// ...and the full set once the limit allows it.
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
vec![oldest.clone(), middle.clone(), newest.clone()]
);
}

#[tokio::test]
async fn list_expired_tracks_begin_update() {
let (_tempdir, stream) = create_test_stream().await;
let moved = create_test_id("moved");

stream
.begin_write(&moved, Some(Timestamp::from_unix_secs(1_000).unwrap()))
.await
.unwrap();
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
vec![moved.clone()]
);

// Extending the deadline past `now` drops the row out of the listing.
stream
.begin_update(&moved, Some(Timestamp::from_unix_secs(9_000).unwrap()))
.await
.unwrap();
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
Vec::<ObjectId>::new()
);

// Shortening it back below `now` makes the row listed again.
stream
.begin_update(&moved, Some(Timestamp::from_unix_secs(2_000).unwrap()))
.await
.unwrap();
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
vec![moved.clone()]
);

// Clearing the expiry leaves a NULL row, which is never listed.
stream.begin_update(&moved, None).await.unwrap();
assert_eq!(
stream
.list_expired(Timestamp::from_unix_secs(5_000).unwrap(), 10)
.await
.unwrap(),
Vec::<ObjectId>::new()
);
}
}
12 changes: 12 additions & 0 deletions objectstore-service/src/change_stream/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,3 +184,15 @@ impl ChangeStream for NoopStream {
fn scope_id(id: &ObjectId, scope: &str) -> Option<u64> {
id.scopes().get_value(scope)?.parse().ok()
}

/// A change stream that can list expired objects.
#[async_trait::async_trait]
pub trait ExpiryIndex: ChangeStream {

@lcian lcian Oct 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we should mix the ExpiryIndex with the ChangeStream, rather, one should use the other.

I'm thinking we could model this as something like:

trait GcStorage: Send + Sync + Debug {
    async fn upsert(&self, id: &ObjectId, expires_at: Option<Timestamp>) -> Result<(), ChangeStreamError>;
    async fn remove(&self, id: &ObjectId);
    async fn list_expired(&self, now: Timestamp, limit: u64) -> Result<Vec<ObjectId>, ChangeStreamError>;
}

Then the GcChangeStream would depend on it, like so:

#[derive(Debug)]
struct GcChangeStream {
    strorage: Arc<dyn GcStorage>
}

#[async_trait]
impl ChangeStream for GcChangeStream {
    async fn begin_write(&self, id: &ObjectId, exp: Option<Timestamp>) -> Result<(), ChangeStreamError> {
        self.storage.upsert(id, exp).await
    }
    // ...
}

And similarly, you would construct construct your GC task that loops through the records and deletes stuff by passing it the same Arc<dyn GcStorage>.

What do you think?

@aldy505 aldy505 Oct 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This makes more sense. But that means I need to change how I implemented it on #653

@lcian lcian Oct 27, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, sorry about that but I only noticed this now that we're adding more capabilities to this struct.
Could you refactor it? 🙏

/// Object ids whose stored deadline is at or before `now`, oldest first,
/// at most `limit` per call. Never returns non-expiring (NULL) rows.
async fn list_expired(
&self,
now: Timestamp,
limit: u64,
) -> Result<Vec<ObjectId>, Box<dyn std::error::Error + Send + Sync>>;
}
Loading