Merge pull request #49 from hyperledger/multiplex-operation-updates-fix

Multiplex operation updates fix

Author: Quantumplation

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-balius"
-version = "0.6.0"
+version = "0.6.1"
 description = "Helpers to write contracts for the FireFly Cardano connector"
 license-file.workspace = true
 publish = false

firefly-balius/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-cardanoconnect"
-version = "0.6.0"
+version = "0.6.1"
 description = "An implementation of the FireFly Connector API for Cardano"
 license-file.workspace = true
 publish = false

firefly-cardanoconnect/Cargo.toml

@@ -10,7 +10,7 @@ use axum::{
 };
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use tracing::{Level, error, instrument, warn};
+use tracing::{Level, error, info, instrument, warn};
 
 use crate::{
     AppState,
@@ -30,6 +30,7 @@ async fn handle_socket(
         IncomingMessage::Listen(ListenMessage { topic }) => topic,
         other => bail!("unexpected first message: {:?}", other),
     };
+    info!("caller listening on topic {topic}");
     let mut subscription = stream_manager.subscribe(&topic).await?;
 
     while let Some(batch) = subscription.recv().await {
@@ -238,7 +239,9 @@ pub async fn handle_socket_upgrade(
     State(app_state): State<AppState>,
     ws: WebSocketUpgrade,
 ) -> Response {
+    info!("Incoming websocket connection");
     ws.on_upgrade(|socket| async move {
+        info!("Websocket connection succeeded");
         if let Err(error) = handle_socket(app_state, socket).await {
             warn!("socket error: {:?}", error);
         }

firefly-cardanoconnect/src/routes/ws.rs

@@ -283,7 +283,7 @@ impl StreamDispatcherWorker {
         let mut batch_sink = None;
         loop {
             let (batch, ack_rx, last_op_id, hwms) = select! {
-                batch = self.build_batch(&mut operation_update_source), if batch_sink.is_some() && !self.listeners.is_empty() => batch,
+                batch = self.build_batch(&mut operation_update_source), if batch_sink.is_some() => batch,
                 Some(change) = state_change_source.recv() => {
                     match change {
                         StreamDispatcherStateChange::NewSettings(settings) => {
@@ -386,7 +386,7 @@ impl StreamDispatcherWorker {
         }
         while events.len() < self.batch_size {
             select! {
-                _ = self.collect_contract_events(hwms, events) => {}
+                _ = self.collect_contract_events(hwms, events), if !self.listeners.is_empty() => {}
                 Ok(()) = operation_update_source.changed() => {
                     self.collect_operation_events(last_op, events).await;
                 }
@@ -542,13 +542,15 @@ mod tests {
     use crate::{
         blockchain::BlockchainClient,
         contracts::ContractManager,
+        operations::{Operation, OperationId, OperationStatus},
         persistence,
         streams::{
-            BlockReference, Listener, ListenerFilter, ListenerType, Stream, mux::Multiplexer,
+            BatchEvent, BlockReference, Listener, ListenerFilter, ListenerType, Stream,
+            mux::Multiplexer,
         },
     };
     use firefly_server::apitypes::ApiError;
-    use tokio::sync::watch;
+    use tokio::{sync::watch, time::timeout};
 
     #[tokio::test]
     async fn should_ack_events() -> Result<(), ApiError> {
@@ -649,4 +651,59 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn should_surface_operation_updates_without_listeners() -> Result<(), ApiError> {
+        let blockchain = Arc::new(BlockchainClient::mock().await);
+        let contracts = Arc::new(ContractManager::none());
+        let persistence = persistence::init(&persistence::PersistenceConfig::Mock).await?;
+        let operation_update_sink = watch::Sender::new(None);
+        let mux = Multiplexer::new(
+            blockchain.clone(),
+            contracts,
+            persistence.clone(),
+            operation_update_sink.clone(),
+        )
+        .await?;
+
+        let stream = Stream {
+            id: "stream_id".to_string().into(),
+            name: "Some Stream".into(),
+            batch_size: 5,
+            batch_timeout: Duration::from_millis(100),
+        };
+        persistence.write_stream(&stream).await?;
+        mux.handle_stream_write(&stream).await?;
+
+        let mut subscription = mux.subscribe("Some Stream").await?;
+
+        let operation = Operation {
+            id: OperationId::from("op".to_string()),
+            status: OperationStatus::Pending,
+            tx_id: None,
+            contract_address: None,
+        };
+
+        let update_id = persistence.write_operation(&operation).await?;
+        operation_update_sink.send_replace(Some(update_id.clone()));
+
+        // We receive a new batch with the operation update
+        let batch = subscription.recv().await.unwrap();
+        assert_eq!(batch.batch_number, 1);
+        assert_eq!(batch.events, vec![BatchEvent::Receipt(operation)]);
+        batch.ack();
+
+        // We don't receive any additional batches
+        assert!(
+            timeout(Duration::from_millis(500), subscription.recv())
+                .await
+                .is_err()
+        );
+
+        // The checkpoint is updated
+        let checkpoint = persistence.read_checkpoint(&stream.id).await?.unwrap();
+        assert_eq!(checkpoint.last_operation_id, Some(update_id));
+
+        Ok(())
+    }
 }

firefly-cardanoconnect/src/streams/mux.rs

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-cardanosigner"
-version = "0.6.0"
+version = "0.6.1"
 description = "A service managing keys and signing for the FireFly Cardano connector"
 license-file.workspace = true
 publish = false

firefly-cardanosigner/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-server"
-version = "0.6.0"
+version = "0.6.1"
 description = "Internal library with shared code for services"
 license-file.workspace = true
 publish = false

firefly-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-cardano-demo"
-version = "0.6.0"
+version = "0.6.1"
 description = "A demo of the firefly-cardanoconnect API"
 license-file.workspace = true
 publish = false

scripts/demo/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-cardano-deploy-contract"
-version = "0.6.0"
+version = "0.6.1"
 description = "A script to build and deploy a Balius smart contract to the FireFly Cardano connector"
 license-file.workspace = true
 publish = false

scripts/deploy-contract/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firefly-cardano-deploy"
-version = "0.6.0"
+version = "0.6.1"
 description = "A script to build and deploy a Balius-backed API to FireFly"
 license-file.workspace = true
 publish = false