Merge branch 'main' into client_event_downlinks

Author: SirCipher

runtime/swimos_runtime/src/downlink/mod.rs

@@ -555,6 +555,10 @@ where
     let mut awaiting_synced: Vec<DownlinkSender> = vec![];
     let mut awaiting_linked: Vec<DownlinkSender> = vec![];
     let mut registered: Vec<DownlinkSender> = vec![];
+    // Track whether any events have been received while syncing the downlink. While this isn't the
+    // nicest thing to have, it's required to distinguish between communicating with a stateless
+    // lane and a downlink syncing with a lane that has a type which may be optional.
+    let mut sync_event = false;
 
     let result: Result<(), H::Report> = loop {
         let (event, is_active) = match task_state.as_mut().as_pin_mut() {
@@ -653,7 +657,17 @@ where
                     trace!("Entering Synced state.");
                     dl_state = ReadTaskDlState::Synced;
                     if is_active {
-                        if I::SINGLE_FRAME_STATE {
+                        // `sync_event` will be false if we're communicating with a stateless lane
+                        // as no event envelope will have been sent. However, it's valid Recon for
+                        // an empty event envelope to be sent (consider Option::None) and this must
+                        // still be sent to the downlink task.
+                        //
+                        // If we're linked to a stateless lane, then `sync_current` cannot be used
+                        // as we will not have received an event envelope as it will dispatch one
+                        // with the empty buffer and this may cause the downlink task's decoder to
+                        // fail due to reading an extant read event. Therefore, delegate the operation to
+                        // `sync_only` which will not send an event notification.
+                        if I::SINGLE_FRAME_STATE && sync_event {
                             sync_current(&mut awaiting_synced, &mut registered, &current).await;
                         } else {
                             sync_only(&mut awaiting_synced, &mut registered).await;
@@ -669,8 +683,11 @@ where
                     break Ok(());
                 }
                 Notification::Event(bytes) => {
+                    sync_event = true;
+
                     trace!("Updating the current value.");
                     current.clear();
+
                     if let Err(e) = interpretation.interpret_frame_data(bytes, &mut current) {
                         if let BadFrameResponse::Abort(report) = failure_handler.failed_with(e) {
                             break Err(report);

runtime/swimos_runtime/src/downlink/tests/value.rs

@@ -384,7 +384,7 @@ async fn shutdowm_after_attached() {
 }
 
 #[tokio::test]
-async fn shutdowm_after_corrupted_frame() {
+async fn shutdown_after_corrupted_frame() {
     let (events, result) = run_test(DownlinkOptions::empty(), |context| async move {
         let TestContext {
             mut tx,
@@ -731,7 +731,7 @@ async fn exhaust_output_buffer() {
 }
 
 #[tokio::test]
-async fn shutdowm_after_timeout_with_no_subscribers() {
+async fn shutdown_after_timeout_with_no_subscribers() {
     let ((_stop, events), result) = run_test_with_config(
         DownlinkOptions::empty(),
         DownlinkRuntimeConfig {
@@ -1079,3 +1079,100 @@ async fn receive_from_two_consumers() {
     assert_eq!(first_events, vec![DownlinkNotification::Unlinked]);
     assert_eq!(second_events, vec![DownlinkNotification::Unlinked]);
 }
+
+#[tokio::test]
+async fn sync_no_event() {
+    let (events, result) = run_test(
+        DownlinkOptions::SYNC,
+        |TestContext {
+             mut tx,
+             mut rx,
+             start_client,
+             stop,
+             mut events,
+             ..
+         }| async move {
+            expect_message(rx.recv().await, Operation::Link);
+
+            tx.link().await;
+
+            start_client.trigger();
+
+            expect_event(
+                events.next().await,
+                State::Unlinked,
+                DownlinkNotification::Linked,
+            );
+            expect_message(rx.recv().await, Operation::Sync);
+
+            tx.sync().await;
+
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Synced,
+            );
+
+            stop.trigger();
+            events.collect::<Vec<_>>().await
+        },
+    )
+    .await;
+    assert!(result.is_ok());
+    assert_eq!(
+        events,
+        vec![(State::Synced, DownlinkNotification::Unlinked)]
+    );
+}
+
+#[tokio::test]
+async fn sync_empty_event() {
+    let (events, result) = run_test(
+        DownlinkOptions::SYNC,
+        |TestContext {
+             mut tx,
+             mut rx,
+             start_client,
+             stop,
+             mut events,
+             ..
+         }| async move {
+            expect_message(rx.recv().await, Operation::Link);
+
+            tx.link().await;
+
+            start_client.trigger();
+
+            expect_event(
+                events.next().await,
+                State::Unlinked,
+                DownlinkNotification::Linked,
+            );
+            expect_message(rx.recv().await, Operation::Sync);
+
+            let content = Message::CurrentValue(Text::new(""));
+            tx.update(content.clone()).await;
+            tx.sync().await;
+
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Event { body: content },
+            );
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Synced,
+            );
+
+            stop.trigger();
+            events.collect::<Vec<_>>().await
+        },
+    )
+    .await;
+    assert!(result.is_ok());
+    assert_eq!(
+        events,
+        vec![(State::Synced, DownlinkNotification::Unlinked)]
+    );
+}