Address feedback

Author: Jake-Shadle

crates/corrosion-tests/src/lib.rs

@@ -364,7 +364,7 @@ pub fn query_to_string(
     String::from_utf8(out).unwrap()
 }
 
-pub async fn assert_sub<T>(
+pub async fn assert_sub_event_eq<T>(
     rx: &mut tokio::sync::mpsc::Receiver<pubsub::SubscriptionEvent>,
     expected: &pubsub::TypedQueryEvent<T>,
 ) -> Option<T>

crates/corrosion-tests/tests/subscription.rs

@@ -352,12 +352,12 @@ async fn single_sub() {
     let mut cur_cid = 0;
 
     // We always get a state of the world first, in our case the DB is empty
-    corrosion_tests::assert_sub::<ServerRow>(
+    corrosion_tests::assert_sub_event_eq::<ServerRow>(
         &mut rx,
         &TypedQueryEvent::Columns(vec!["endpoint".into(), "icao".into(), "tokens".into()]),
     )
     .await;
-    corrosion_tests::assert_sub::<ServerRow>(
+    corrosion_tests::assert_sub_event_eq::<ServerRow>(
         &mut rx,
         &TypedQueryEvent::EndOfQuery {
             time: 0.,
@@ -395,7 +395,7 @@ async fn single_sub() {
             pool.broadcast_changes(&mut states).await;
             cur_cid += 1;
 
-            corrosion_tests::assert_sub(
+            corrosion_tests::assert_sub_event_eq(
                 &mut rx,
                 &TypedQueryEvent::Change(
                     ChangeType::Insert,
@@ -423,7 +423,7 @@ async fn single_sub() {
             pool.broadcast_changes(&mut states).await;
             cur_cid += 1;
 
-            corrosion_tests::assert_sub(
+            corrosion_tests::assert_sub_event_eq(
                 &mut rx,
                 &TypedQueryEvent::Change(
                     ChangeType::Update,
@@ -451,7 +451,7 @@ async fn single_sub() {
             pool.broadcast_changes(&mut states).await;
             cur_cid += 1;
 
-            corrosion_tests::assert_sub(
+            corrosion_tests::assert_sub_event_eq(
                 &mut rx,
                 &TypedQueryEvent::Change(
                     ChangeType::Delete,
@@ -521,10 +521,10 @@ async fn multiple_subs() {
     let mut cur_cid = 0;
 
     let mut assert_all = async |tqe| {
-        corrosion_tests::assert_sub::<ServerRow>(&mut orx, &tqe).await;
+        corrosion_tests::assert_sub_event_eq::<ServerRow>(&mut orx, &tqe).await;
 
         for sub in &mut multi_subs {
-            corrosion_tests::assert_sub::<ServerRow>(sub, &tqe).await;
+            corrosion_tests::assert_sub_event_eq::<ServerRow>(sub, &tqe).await;
         }
     };
 

crates/corrosion/src/db.rs

@@ -16,56 +16,58 @@ pub struct InitializedDb {
     pub actor_id: ActorId,
 }
 
-pub async fn setup(db_path: &crate::Path, schema: &str) -> eyre::Result<InitializedDb> {
-    let partial_schema = corro_types::schema::parse_sql(schema)?;
-
-    let actor_id = {
-        // we need to set auto_vacuum before any tables are created
-        let db_conn = rusqlite::Connection::open(db_path)?;
-        db_conn.execute_batch("PRAGMA auto_vacuum = INCREMENTAL")?;
-
-        let conn = CrConn::init(db_conn)?;
-        conn.query_row("SELECT crsql_site_id();", [], |row| {
-            row.get::<_, ActorId>(0)
-        })?
-    };
+impl InitializedDb {
+    pub async fn setup(db_path: &crate::Path, schema: &str) -> eyre::Result<Self> {
+        let partial_schema = corro_types::schema::parse_sql(schema)?;
+
+        let actor_id = {
+            // we need to set auto_vacuum before any tables are created
+            let db_conn = rusqlite::Connection::open(db_path)?;
+            db_conn.execute_batch("PRAGMA auto_vacuum = INCREMENTAL")?;
+
+            let conn = CrConn::init(db_conn)?;
+            conn.query_row("SELECT crsql_site_id();", [], |row| {
+                row.get::<_, ActorId>(0)
+            })?
+        };
 
-    let write_sema = Arc::new(tokio::sync::Semaphore::new(1));
-    let pool = SplitPool::create(&db_path, write_sema.clone()).await?;
+        let write_sema = Arc::new(tokio::sync::Semaphore::new(1));
+        let pool = SplitPool::create(&db_path, write_sema.clone()).await?;
 
-    let clock = Arc::new(
-        uhlc::HLCBuilder::default()
-            .with_id(actor_id.try_into().unwrap())
-            .with_max_delta(std::time::Duration::from_millis(300))
-            .build(),
-    );
+        let clock = Arc::new(
+            uhlc::HLCBuilder::default()
+                .with_id(actor_id.try_into().unwrap())
+                .with_max_delta(std::time::Duration::from_millis(300))
+                .build(),
+        );
 
-    let schema = {
-        let mut conn = pool.write_priority().await?;
+        let schema = {
+            let mut conn = pool.write_priority().await?;
 
-        let old_schema = {
-            corro_types::agent::migrate(clock.clone(), &mut conn)?;
-            let mut schema = corro_types::schema::init_schema(&conn)?;
-            schema.constrain()?;
+            let old_schema = {
+                corro_types::agent::migrate(clock.clone(), &mut conn)?;
+                let mut schema = corro_types::schema::init_schema(&conn)?;
+                schema.constrain()?;
 
-            schema
-        };
+                schema
+            };
 
-        tokio::task::block_in_place(|| update_schema(old_schema, partial_schema, &mut conn))?
-    };
+            tokio::task::block_in_place(|| update_schema(&mut conn, old_schema, partial_schema))?
+        };
 
-    Ok(InitializedDb {
-        pool,
-        clock,
-        schema: Arc::new(schema),
-        actor_id,
-    })
+        Ok(Self {
+            pool,
+            clock,
+            schema: Arc::new(schema),
+            actor_id,
+        })
+    }
 }
 
 pub fn update_schema(
+    conn: &mut WriteConn,
     old_schema: Schema,
     new_schema: Schema,
-    conn: &mut WriteConn,
 ) -> eyre::Result<Schema> {
     // clone the previous schema and apply
     let mut new_schema = {

crates/corrosion/src/persistent/executor.rs

@@ -17,7 +17,8 @@ use std::{
 
 pub type BroadcastTx = tokio::sync::mpsc::Sender<broadcast::BroadcastInput>;
 
-///
+/// A DB mutator that will broadcast changes to any subscribers when a mutation
+/// occurs that matches a subscriber's query
 #[derive(Clone)]
 pub struct BroadcastingTransactor {
     pool: SplitPool,