fix(filemanager): sequencer ordering

lib/workload/stateless/stacks/filemanager/database/migrations/0003_add_sequencer_index.sql

@@ -0,0 +1,4 @@
+-- Create an index for ordering `s3_object`s in ascending order.
+-- TODO this should be created `concurrently` when running migrations without transactions is released:
+-- https://github.com/launchbadge/sqlx/issues/767
+create index sequencer_index on s3_object (sequencer);

lib/workload/stateless/stacks/filemanager/filemanager/Cargo.toml

@@ -69,6 +69,7 @@ aws_lambda_events = "0.15"
 
 [dev-dependencies]
 lazy_static = "1"
+rand = "0.8"
 
 aws-smithy-runtime-api = "1"
 aws-smithy-mocks-experimental = "0.2"

lib/workload/stateless/stacks/filemanager/filemanager/src/queries/list.rs

@@ -1,9 +1,10 @@
 //! Query builder involving list operations on the database.
 //!
 
-use sea_orm::{EntityTrait, FromQueryResult, PaginatorTrait, QuerySelect, Select};
+use sea_orm::{EntityTrait, FromQueryResult, PaginatorTrait, QueryOrder, QuerySelect, Select};
 
 use crate::database::entities::object::Entity as ObjectEntity;
+use crate::database::entities::s3_object::Column as S3Column;
 use crate::database::entities::s3_object::Entity as S3ObjectEntity;
 use crate::database::Client;
 use crate::error::Error::OverflowError;
@@ -46,7 +47,7 @@ impl<'a> ListQueryBuilder<'a, S3ObjectEntity> {
 
     /// Build a select query for finding values from s3 objects.
     pub fn build_object() -> Select<S3ObjectEntity> {
-        S3ObjectEntity::find()
+        S3ObjectEntity::find().order_by_asc(S3Column::Sequencer)
     }
 }
 
@@ -117,7 +118,7 @@ mod tests {
 
     use crate::database::aws::migration::tests::MIGRATOR;
     use crate::database::Client;
-    use crate::queries::tests::initialize_database;
+    use crate::queries::tests::{initialize_database, initialize_database_reorder};
 
     use super::*;
 
@@ -175,7 +176,7 @@ mod tests {
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn test_list_s3_objects(pool: PgPool) {
         let client = Client::from_pool(pool);
-        let entries = initialize_database(&client, 10).await;
+        let entries = initialize_database_reorder(&client, 10).await;
 
         let builder = ListQueryBuilder::<S3ObjectEntity>::new(&client);
         let result = builder.all().await.unwrap();
@@ -192,7 +193,7 @@ mod tests {
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn test_paginate_s3_objects(pool: PgPool) {
         let client = Client::from_pool(pool);
-        let entries = initialize_database(&client, 10).await;
+        let entries = initialize_database_reorder(&client, 10).await;
 
         let builder = ListQueryBuilder::<S3ObjectEntity>::new(&client);
 

lib/workload/stateless/stacks/filemanager/filemanager/src/queries/mod.rs

@@ -9,6 +9,8 @@ pub(crate) mod tests {
     use std::ops::Add;
 
     use chrono::{DateTime, Days};
+    use rand::seq::SliceRandom;
+    use rand::thread_rng;
     use sea_orm::Set;
     use sea_orm::{ActiveModelTrait, TryIntoModel};
     use strum::EnumCount;
@@ -21,13 +23,39 @@ pub(crate) mod tests {
     use crate::database::Client;
     use crate::uuid::UuidGenerator;
 
+    /// Initialize the database state for testing and shuffle entries to simulate
+    /// out of order events.
+    pub(crate) async fn initialize_database_reorder(
+        client: &Client,
+        n: usize,
+    ) -> Vec<(Object, S3Object)> {
+        let mut data = initialize_database_with_shuffle(client, n, true).await;
+
+        // Return the correct ordering for test purposes
+        data.sort_by(|(_, a), (_, b)| a.sequencer.cmp(&b.sequencer));
+
+        data
+    }
+
     /// Initialize database state for testing.
     pub(crate) async fn initialize_database(client: &Client, n: usize) -> Vec<(Object, S3Object)> {
+        initialize_database_with_shuffle(client, n, false).await
+    }
+
+    async fn initialize_database_with_shuffle(
+        client: &Client,
+        n: usize,
+        shuffle: bool,
+    ) -> Vec<(Object, S3Object)> {
         let mut output = vec![];
 
-        for index in 0..n {
-            let (object, s3_object) = generate_entry(index);
+        let mut entries: Vec<_> = (0..n).map(generate_entry).collect();
+
+        if shuffle {
+            entries.shuffle(&mut thread_rng());
+        }
 
+        for (object, s3_object) in entries {
             object
                 .clone()
                 .insert(client.connection_ref())

lib/workload/stateless/stacks/filemanager/filemanager/src/routes/list.rs

@@ -222,7 +222,7 @@ mod tests {
     use crate::database::aws::migration::tests::MIGRATOR;
     use crate::database::entities::object::Model as Object;
     use crate::database::entities::s3_object::Model as S3Object;
-    use crate::queries::tests::initialize_database;
+    use crate::queries::tests::{initialize_database, initialize_database_reorder};
     use crate::routes::list::{ListCount, ListResponse};
     use crate::routes::{api_router, AppState};
 
@@ -331,7 +331,7 @@ mod tests {
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn list_s3_objects_api(pool: PgPool) {
         let state = AppState::from_pool(pool);
-        let entries = initialize_database(state.client(), 10).await;
+        let entries = initialize_database_reorder(state.client(), 10).await;
 
         let app = api_router(state);
         let response = app
@@ -365,7 +365,7 @@ mod tests {
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn list_s3_objects_api_paginate(pool: PgPool) {
         let state = AppState::from_pool(pool);
-        let entries = initialize_database(state.client(), 10).await;
+        let entries = initialize_database_reorder(state.client(), 10).await;
 
         let app = api_router(state);
         let response = app
@@ -399,7 +399,7 @@ mod tests {
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn list_s3_objects_api_zero_page_size(pool: PgPool) {
         let state = AppState::from_pool(pool);
-        let entries = initialize_database(state.client(), 10).await;
+        let entries = initialize_database_reorder(state.client(), 10).await;
 
         let app = api_router(state);
         let response = app