fix(spooler): Fix datetime comparison

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Use `matches_any_origin` to scrub HTTP hosts in spans. ([#3939](https://github.com/getsentry/relay/pull/3939)).
 - Keep frames from both ends of the stacktrace when trimming frames. ([#3905](https://github.com/getsentry/relay/pull/3905))
+- Use `UnixTimestamp` instead of `DateTime` when sorting envelopes from disk. ([#4025](https://github.com/getsentry/relay/pull/4025))
 
 **Features**:
 

relay-server/src/services/buffer/envelope_store/sqlite.rs

@@ -13,6 +13,7 @@ use crate::Envelope;
 use futures::stream::StreamExt;
 use hashbrown::HashSet;
 use relay_base_schema::project::{ParseProjectKeyError, ProjectKey};
+use relay_common::time::UnixTimestamp;
 use relay_config::Config;
 use sqlx::migrate::MigrateError;
 use sqlx::query::Query;
@@ -369,10 +370,13 @@ impl SqliteEnvelopeStore {
             }
         }
 
-        // We sort envelopes by `received_at`.
+        // We sort envelopes by `received_at` in ascending order.
+        //
         // Unfortunately we have to do this because SQLite `DELETE` with `RETURNING` doesn't
         // return deleted rows in a specific order.
-        extracted_envelopes.sort_by_key(|a| a.received_at());
+        extracted_envelopes.sort_by_key(|a| {
+            UnixTimestamp::from_datetime(a.received_at()).unwrap_or(UnixTimestamp::now())
+        });
 
         Ok(extracted_envelopes)
     }
@@ -518,12 +522,10 @@ pub fn build_count_all<'a>() -> Query<'a, Sqlite, SqliteArguments<'a>> {
 
 #[cfg(test)]
 mod tests {
-    use hashbrown::HashSet;
     use std::time::Duration;
     use tokio::time::sleep;
 
     use relay_base_schema::project::ProjectKey;
-    use relay_event_schema::protocol::EventId;
 
     use super::*;
     use crate::services::buffer::testutils::utils::{mock_envelopes, setup_db};
@@ -538,21 +540,30 @@ mod tests {
 
         // We insert 10 envelopes.
         let envelopes = mock_envelopes(10);
-        let envelope_ids: HashSet<EventId> =
-            envelopes.iter().filter_map(|e| e.event_id()).collect();
         assert!(envelope_store
             .insert_many(envelopes.iter().map(|e| e.as_ref().try_into().unwrap()))
             .await
             .is_ok());
 
-        // We check that if we load more than the limit, we still get back at most 10.
+        // We check that if we load 5, we get the newest 5.
+        let extracted_envelopes = envelope_store
+            .delete_many(own_key, sampling_key, 5)
+            .await
+            .unwrap();
+        assert_eq!(extracted_envelopes.len(), 5);
+        for (i, extracted_envelope) in extracted_envelopes.iter().enumerate().take(5) {
+            assert_eq!(extracted_envelope.event_id(), envelopes[5..][i].event_id());
+        }
+
+        // We check that if we load more than the envelopes stored on disk, we still get back at
+        // most 5.
         let extracted_envelopes = envelope_store
-            .delete_many(own_key, sampling_key, 15)
+            .delete_many(own_key, sampling_key, 10)
             .await
             .unwrap();
-        assert_eq!(envelopes.len(), 10);
-        for envelope in extracted_envelopes {
-            assert!(envelope_ids.contains(&envelope.event_id().unwrap()));
+        assert_eq!(extracted_envelopes.len(), 5);
+        for (i, extracted_envelope) in extracted_envelopes.iter().enumerate().take(5) {
+            assert_eq!(extracted_envelope.event_id(), envelopes[0..5][i].event_id());
         }
     }