Fix unreachability detection for streams out of blocking nodes

Signed-off-by: Michael X. Grey <grey@openrobotics.org>

Author: mxgrey

src/callback.rs

@@ -19,7 +19,7 @@ use crate::{
     BlockingCallback, AsyncCallback, Channel, ChannelQueue,
     OperationRoster, StreamPack, Input, Provider, ProvideOnce,
     AddOperation, OperateCallback, ManageInput, OperationError,
-    OrBroken, OperateTask,
+    OrBroken, OperateTask, UnusedStreams, ManageDisposal,
 };
 
 use bevy::{
@@ -115,10 +115,17 @@ impl<'a> CallbackRequest<'a> {
         &mut self,
         session: Entity,
         response: Response,
+        unused_streams: UnusedStreams,
     ) -> Result<(), OperationError> {
+        if !unused_streams.streams.is_empty() {
+            self.world.get_entity_mut(self.source).or_broken()?
+                .emit_disposal(session, unused_streams.into(), self.roster);
+        }
+
         self.world
             .get_entity_mut(self.target).or_broken()?
             .give_input(session, response, self.roster)?;
+
         Ok(())
     }
 
@@ -200,9 +207,12 @@ where
         }, input.world);
         self.system.apply_deferred(&mut input.world);
 
-        Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
+        let mut unused_streams = UnusedStreams::new(input.source);
+        Streams::process_buffer(
+            streams, input.source, session, &mut unused_streams, input.world, input.roster
+        )?;
 
-        input.give_response(session, response)
+        input.give_response(session, response, unused_streams)
     }
 }
 
@@ -321,8 +331,12 @@ where
             let response = (self)(BlockingCallback {
                 request, streams: streams.clone(), source: input.source, session,
             });
-            Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
-            input.give_response(session, response)
+
+            let mut unused_streams = UnusedStreams::new(input.source);
+            Streams::process_buffer(
+                streams, input.source, session, &mut unused_streams, input.world, input.roster
+            )?;
+            input.give_response(session, response, unused_streams)
         };
         Callback::new(MapCallback { callback })
     }

src/chain.rs

@@ -23,7 +23,7 @@ use std::error::Error;
 
 use crate::{
     UnusedTarget, AddOperation, Node, InputSlot, Builder,
-    StreamPack, Provider, ProvideOnce, Scope,
+    StreamPack, Provider, ProvideOnce, Scope, StreamOf,
     AsMap, IntoBlockingMap, IntoAsyncMap, Output, Noop,
     ForkTargetStorage, StreamTargetMap, ScopeSettings, CreateCancelFilter,
     CreateDisposalFilter,
@@ -417,6 +417,17 @@ impl<'w, 's, 'a, 'b, T: 'static + Send + Sync> Chain<'w, 's, 'a, 'b, T> {
     }
 }
 
+impl<'w, 's, 'a, 'b, T> Chain<'w, 's, 'a, 'b, StreamOf<T>>
+where
+    T: 'static + Send + Sync,
+{
+    /// When the output value is wrapped in a [`StreamOf`] container, this will
+    /// strip it out of that wrapper.
+    pub fn inner(self) -> Chain<'w, 's, 'a, 'b, T> {
+        self.map_block(|v| v.0)
+    }
+}
+
 impl<'w, 's, 'a, 'b, T, E> Chain<'w, 's, 'a, 'b, Result<T, E>>
 where
     T: 'static + Send + Sync,

src/disposal.rs

@@ -28,7 +28,7 @@ use std::collections::HashMap;
 
 use crate::{
     OperationRoster, operation::ScopeStorage, Cancellation, UnhandledErrors,
-    DisposalFailure,
+    DisposalFailure, ImpulseMarker,
 };
 
 #[derive(Debug, Clone)]
@@ -100,6 +100,10 @@ pub enum DisposalCause {
 
     /// A scope was cancelled so its output has been disposed.
     Scope(Cancellation),
+
+    /// One or more streams from a node never emitted any signal. This can lead
+    /// to unexpected
+    UnusedStreams(UnusedStreams)
 }
 
 /// A variant of [`DisposalCause`]
@@ -196,7 +200,6 @@ impl From<TaskDespawned> for DisposalCause {
         Self::TaskDespawned(value)
     }
 }
-
 /// A variant of [`DisposalCause`]
 #[derive(Debug)]
 pub struct PoisonedMutexDisposal {
@@ -210,6 +213,26 @@ impl From<PoisonedMutexDisposal> for DisposalCause {
     }
 }
 
+#[derive(Debug)]
+pub struct UnusedStreams {
+    /// The node which did not use all its streams
+    pub node: Entity,
+    /// The streams which went unused.
+    pub streams: Vec<&'static str>,
+}
+
+impl UnusedStreams {
+    pub fn new(node: Entity) -> Self {
+        Self { node, streams: Default::default() }
+    }
+}
+
+impl From<UnusedStreams> for DisposalCause {
+    fn from(value: UnusedStreams) -> Self {
+        Self::UnusedStreams(value)
+    }
+}
+
 pub trait ManageDisposal {
     fn emit_disposal(
         &mut self,
@@ -233,15 +256,19 @@ impl<'w> ManageDisposal for EntityMut<'w> {
         roster: &mut OperationRoster,
     ) {
         let Some(scope) = self.get::<ScopeStorage>() else {
-            let broken_node = self.id();
-            self.world_scope(|world| {
-                world
-                .get_resource_or_insert_with(|| UnhandledErrors::default())
-                .disposals
-                .push(DisposalFailure {
-                    disposal, broken_node, backtrace: Some(Backtrace::new())
+            if !self.contains::<ImpulseMarker>() {
+                // If the emitting node does not have a scope as not part of an
+                // impulse chain, then something is broken.
+                let broken_node = self.id();
+                self.world_scope(|world| {
+                    world
+                    .get_resource_or_insert_with(|| UnhandledErrors::default())
+                    .disposals
+                    .push(DisposalFailure {
+                        disposal, broken_node, backtrace: Some(Backtrace::new())
+                    });
                 });
-            });
+            }
             return;
         };
         let scope = scope.get();

src/impulse.rs

@@ -124,6 +124,7 @@ where
         let target = self.commands.spawn((
             Detached::default(),
             UnusedTarget,
+            ImpulseMarker,
         )).id();
 
         // We should automatically delete the previous step in the chain once

src/impulse/internal.rs

@@ -42,6 +42,9 @@ pub(crate) trait Impulsive {
     fn execute(request: OperationRequest) -> OperationResult;
 }
 
+#[derive(Component)]
+pub(crate) struct ImpulseMarker;
+
 pub(crate) struct AddImpulse<I: Impulsive> {
     source: Entity,
     impulse: I,
@@ -64,6 +67,7 @@ impl<I: Impulsive + 'static + Sync + Send> Command for AddImpulse<I> {
             .insert((
                 OperationExecuteStorage(perform_impulse::<I>),
                 Cancellable::new(cancel_impulse),
+                ImpulseMarker,
             ))
             .remove::<UnusedTarget>();
     }

src/impulse/map.rs

@@ -26,7 +26,7 @@ use crate::{
     Impulsive, OperationSetup, OperationRequest, SingleTargetStorage, StreamPack,
     InputBundle, OperationResult, OrBroken, Input, ManageInput,
     ChannelQueue, BlockingMap, AsyncMap, Channel, OperateTask, ActiveTasksStorage,
-    CallBlockingMapOnce, CallAsyncMapOnce,
+    CallBlockingMapOnce, CallAsyncMapOnce, UnusedStreams,
 };
 
 /// The key difference between this and [`crate::OperateBlockingMap`] is that
@@ -94,7 +94,12 @@ where
 
         let response = f.call(BlockingMap { request, streams: streams.clone(), source, session });
 
-        Streams::process_buffer(streams, source, session, world, roster)?;
+        let mut unused_streams = UnusedStreams::new(source);
+        Streams::process_buffer(
+            streams, source, session, &mut unused_streams, world, roster
+        )?;
+        // Note: We do not need to emit a disposal for any unused streams since
+        // this is only used for impulses, not workflows.
 
         world.get_entity_mut(target).or_broken()?.give_input(session, response, roster)?;
         Ok(())

src/input.rs

@@ -32,7 +32,7 @@ use backtrace::Backtrace;
 use crate::{
     OperationRoster, OperationError, OrBroken,
     DeferredRoster, Cancel, Cancellation, CancellationCause, Broken,
-    BufferSettings, RetentionPolicy, ForkTargetStorage,
+    BufferSettings, RetentionPolicy, ForkTargetStorage, UnusedTarget,
 };
 
 /// Typical container for input data accompanied by its session information.
@@ -210,8 +210,18 @@ impl<'w> ManageInput for EntityMut<'w> {
         session: Entity,
         data: T,
     ) -> Result<(), OperationError> {
-        let mut storage = self.get_mut::<InputStorage<T>>().or_broken()?;
-        storage.reverse_queue.insert(0, Input { session, data });
+        if let Some(mut storage) = self.get_mut::<InputStorage<T>>() {
+            storage.reverse_queue.insert(0, Input { session, data });
+        } else if !self.contains::<UnusedTarget>() {
+            // If the input is being fed to an unused target then we can
+            // generally ignore it, although it may indicate a bug in the user's
+            // workflow because workflow branches that end in an unused target
+            // will be spuriously dropped when the scope terminates.
+
+            // However in this case, the target is not unused but also does not
+            // have the correct input storage type. This indicates
+            None.or_broken()?;
+        }
         Ok(())
     }
 

src/operation/operate_map.rs

@@ -27,7 +27,8 @@ use crate::{
     SingleTargetStorage, StreamPack, Input, ManageInput, OperationCleanup,
     CallBlockingMap, CallAsyncMap, SingleInputStorage, OperationResult,
     OrBroken, OperationSetup, OperationRequest, OperateTask, ActiveTasksStorage,
-    OperationReachability, ReachabilityResult, InputBundle,
+    OperationReachability, ReachabilityResult, InputBundle, UnusedStreams,
+    ManageDisposal,
 };
 
 #[derive(Bundle)]
@@ -96,7 +97,14 @@ where
         let response = f.call(BlockingMap { request, streams: streams.clone(), source, session });
         map.f = Some(f);
 
-        Streams::process_buffer(streams, source, session, world, roster)?;
+        let mut unused_streams = UnusedStreams::new(source);
+        Streams::process_buffer(
+            streams, source, session, &mut unused_streams, world, roster
+        )?;
+        if !unused_streams.streams.is_empty() {
+            world.get_entity_mut(source).or_broken()?
+                .emit_disposal(session, unused_streams.into(), roster);
+        }
 
         world.get_entity_mut(target).or_broken()?.give_input(session, response, roster)?;
         Ok(())

src/request.rs

@@ -21,7 +21,7 @@ use std::future::Future;
 
 use crate::{
     UnusedTarget, StreamPack, ProvideOnce, IntoBlockingMapOnce, IntoAsyncMap,
-    Impulse, Detached, InputCommand, Cancellable, cancel_impulse,
+    Impulse, Detached, InputCommand, Cancellable, ImpulseMarker, cancel_impulse,
 };
 
 /// Extensions for creating impulse chains by making a request to a provider or
@@ -88,10 +88,14 @@ impl<'w, 's> RequestExt<'w, 's> for Commands<'w, 's> {
         let target = self.spawn((
             Detached::default(),
             UnusedTarget,
+            ImpulseMarker,
         )).id();
 
         let source = self
-            .spawn(Cancellable::new(cancel_impulse))
+            .spawn((
+                Cancellable::new(cancel_impulse),
+                ImpulseMarker,
+            ))
             // We set the parent of this source to the target so that when the
             // target gets despawned, this will also be despawned.
             .set_parent(target)

src/service/blocking.rs

@@ -18,7 +18,7 @@
 use crate::{
     BlockingService, BlockingServiceInput, IntoService, ServiceTrait, ServiceRequest,
     Input, ManageInput, ServiceBundle, OperationRequest, OperationError, StreamPack,
-    OrBroken, dispose_for_despawned_service,
+    UnusedStreams, ManageDisposal, OrBroken, dispose_for_despawned_service,
     service::builder::BlockingChosen,
 };
 
@@ -115,7 +115,9 @@ where
             request, streams: streams.clone(), provider, source, session,
         }, world);
         service.apply_deferred(world);
-        Streams::process_buffer(streams, source, session, world, roster)?;
+
+        let mut unused_streams = UnusedStreams::new(source);
+        Streams::process_buffer(streams, source, session, &mut unused_streams, world, roster)?;
 
         if let Some(mut provider_mut) = world.get_entity_mut(provider) {
             if let Some(mut storage) = provider_mut.get_mut::<BlockingServiceStorage<Request, Response, Streams>>() {
@@ -131,6 +133,11 @@ where
             // not consider this to be cancelled.
         }
 
+        if !unused_streams.streams.is_empty() {
+            world.get_entity_mut(source).or_broken()?
+                .emit_disposal(session, unused_streams.into(), roster);
+        }
+
         world.get_entity_mut(target).or_broken()?
             .give_input(session, response, roster)?;
         Ok(())