Add logging to event bus and fix test

src/app/cli/src/app.rs

@@ -457,7 +457,7 @@ pub fn configure_cli_catalog(
 
 #[tracing::instrument(level = "info", skip_all)]
 async fn initialize_components(cli_catalog: &Catalog) -> Result<(), CLIError> {
-    // TODO: For some reason, we if we do in a single transaction, there is a hangup
+    // TODO: Generalize on-startup initialization into a trait
     DatabaseTransactionRunner::new(cli_catalog.clone())
         .transactional(|transactional_catalog| async move {
             let registrator = transactional_catalog
@@ -467,11 +467,8 @@ async fn initialize_components(cli_catalog: &Catalog) -> Result<(), CLIError> {
             registrator
                 .ensure_predefined_accounts_are_registered()
                 .await
-                .map_err(CLIError::critical)
-        })
-        .await?;
-    DatabaseTransactionRunner::new(cli_catalog.clone())
-        .transactional(|transactional_catalog| async move {
+                .map_err(CLIError::critical)?;
+
             let initializer = transactional_catalog
                 .get_one::<DatasetOwnershipServiceInMemoryStateInitializer>()
                 .map_err(CLIError::critical)?;

src/infra/core/tests/tests/test_query_service_impl.rs

@@ -399,7 +399,7 @@ async fn test_dataset_tail_empty_dataset() {
     let tempdir = tempfile::tempdir().unwrap();
     let catalog = create_catalog_with_local_workspace(
         tempdir.path(),
-        MockDatasetActionAuthorizer::new().expect_check_read_a_dataset(2),
+        MockDatasetActionAuthorizer::new().expect_check_read_a_dataset(1),
     );
 
     let dataset_repo = catalog.get_one::<dyn DatasetRepository>().unwrap();

src/infra/core/tests/tests/test_sync_service_impl.rs

@@ -75,7 +75,7 @@ async fn do_test_sync(
 
     let dataset_authorizer = construct_authorizer(
         &AuthorizationExpectations {
-            d1_reads: 8,
+            d1_reads: 7,
             d2_reads: 2,
             d1_writes: 1,
             d2_writes: 4,

src/utils/event-bus/Cargo.toml

@@ -27,9 +27,10 @@ internal-error = { workspace = true }
 async-trait = "0.1"
 dill = "0.8"
 futures = "0.3"
+tracing = "0.1"
 
 [dev-dependencies]
 test-log = { version = "0.2", features = ["trace"] }
-tokio = { version = "1", default-features = false, features=["rt", "macros"] }
+tokio = { version = "1", default-features = false, features = ["rt", "macros"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 thiserror = { version = "1", default-features = false }

src/utils/event-bus/src/event_bus.rs

@@ -9,7 +9,7 @@
 
 use std::sync::Arc;
 
-use dill::{component, Catalog};
+use dill::{Builder, Catalog};
 use internal_error::InternalError;
 
 use crate::{AsyncEventHandler, EventHandler};
@@ -20,7 +20,7 @@ pub struct EventBus {
     catalog: Arc<Catalog>,
 }
 
-#[component(pub)]
+#[dill::component(pub)]
 impl EventBus {
     pub fn new(catalog: Arc<Catalog>) -> EventBus {
         Self { catalog }
@@ -37,13 +37,16 @@ impl EventBus {
     }
 
     fn sync_dispatch<TEvent: 'static + Clone>(&self, event: &TEvent) -> Result<(), InternalError> {
-        let sync_handlers = self
-            .catalog
-            .get::<dill::AllOf<dyn EventHandler<TEvent>>>()
-            .unwrap();
+        let builders = self.catalog.builders_for::<dyn EventHandler<TEvent>>();
 
-        for sync_handler in sync_handlers {
-            sync_handler.handle(event)?;
+        for b in builders {
+            tracing::debug!(
+                handler = b.instance_type_name(),
+                event = std::any::type_name::<TEvent>(),
+                "Dispatching event to sync handler"
+            );
+            let inst = b.get(&self.catalog).unwrap();
+            inst.handle(event)?;
         }
 
         Ok(())
@@ -53,17 +56,25 @@ impl EventBus {
         &self,
         event: &TEvent,
     ) -> Result<(), InternalError> {
-        let async_handlers = self
-            .catalog
-            .get::<dill::AllOf<dyn AsyncEventHandler<TEvent>>>()
-            .unwrap();
+        let builders = self.catalog.builders_for::<dyn AsyncEventHandler<TEvent>>();
 
-        let async_handler_futures: Vec<_> = async_handlers
+        let mut handlers = Vec::new();
+        for b in builders {
+            tracing::debug!(
+                handler = b.instance_type_name(),
+                event = std::any::type_name::<TEvent>(),
+                "Dispatching event to async handler"
+            );
+            let handler = b.get(&self.catalog).unwrap();
+            handlers.push(handler);
+        }
+
+        let futures: Vec<_> = handlers
             .iter()
             .map(|handler| handler.handle(event))
             .collect();
 
-        let results = futures::future::join_all(async_handler_futures).await;
+        let results = futures::future::join_all(futures).await;
         results.into_iter().try_for_each(|res| res)?;
 
         Ok(())