runtime: better error handling

check.sh

@@ -63,7 +63,6 @@ cd ${SCRIPTPATH}/examples/zigbee && cargo fmt --check
 cd ${SCRIPTPATH} && cargo clippy --all-targets --workspace --features=vulkan,zeromq,audio,flow_scheduler,tpb_scheduler,soapy,lttng,zynq,wgpu -- -D warnings
 cd ${SCRIPTPATH}/crates/futuredsp && cargo clippy --all-targets -- -D warnings
 cd ${SCRIPTPATH}/crates/macros && cargo clippy --all-targets -- -D warnings
-cd ${SCRIPTPATH}/crates/prophecy && cargo clippy --all-targets -- -D warnings
 cd ${SCRIPTPATH}/crates/remote && cargo clippy --all-targets -- -D warnings
 cd ${SCRIPTPATH}/crates/types && cargo clippy --all-targets -- -D warnings
 

examples/custom-routes/src/main.rs

@@ -73,6 +73,6 @@ async fn start_fg(State(ws): State<WebState>) {
         .build(),
     );
     let rt_handle = ws.rt.lock().unwrap().as_ref().unwrap().clone();
-    let mut fg_handle = rt_handle.start(fg).await;
+    let mut fg_handle = rt_handle.start(fg).await.unwrap();
     dbg!(fg_handle.description().await.unwrap());
 }

perf/buffer_rand/buffer_rand.rs

@@ -9,6 +9,7 @@ use futuresdr::blocks::NullSource;
 use futuresdr::runtime::buffer::slab::Slab;
 use futuresdr::runtime::scheduler::FlowScheduler;
 use futuresdr::runtime::scheduler::SmolScheduler;
+use futuresdr::runtime::Error;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Runtime;
 
@@ -19,7 +20,7 @@ fn connect(
     dst: usize,
     dst_port: &'static str,
     slab: bool,
-) -> Result<()> {
+) -> std::result::Result<(), Error> {
     if slab {
         fg.connect_stream_with_type(src, src_port, dst, dst_port, Slab::new())
     } else {

perf/buffer_size/buffer_size.rs

@@ -2,7 +2,6 @@ use clap::Parser;
 use std::time;
 
 use futuresdr::anyhow::{Context, Result};
-// use futuresdr::blocks::Copy;
 use futuresdr::blocks::CopyRand;
 use futuresdr::blocks::Head;
 use futuresdr::blocks::NullSink;
@@ -11,6 +10,7 @@ use futuresdr::runtime::buffer::circular::Circular;
 use futuresdr::runtime::buffer::slab::Slab;
 use futuresdr::runtime::scheduler::FlowScheduler;
 use futuresdr::runtime::scheduler::SmolScheduler;
+use futuresdr::runtime::Error;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Runtime;
 
@@ -22,7 +22,7 @@ fn connect(
     dst_port: &'static str,
     slab: bool,
     min_bytes: usize,
-) -> Result<()> {
+) -> std::result::Result<(), Error> {
     if slab {
         fg.connect_stream_with_type(src, src_port, dst, dst_port, Slab::with_size(min_bytes))
     } else {

src/runtime/block.rs

@@ -179,13 +179,13 @@ impl<T: Kernel + Send + 'static> TypedBlockWrapper<T> {
                 if i < mio.inputs().len() {
                     i
                 } else {
-                    return Err(Error::InvalidHandler(PortId::Index(i)));
+                    return Err(Error::InvalidMessagePort(None, PortId::Index(i)));
                 }
             }
             PortId::Name(n) => match mio.input_name_to_id(&n) {
                 Some(s) => s,
                 None => {
-                    return Err(Error::InvalidHandler(PortId::Name(n)));
+                    return Err(Error::InvalidMessagePort(None, PortId::Name(n)));
                 }
             },
         };
@@ -311,7 +311,7 @@ impl<T: Kernel + Send + 'static> TypedBlockWrapper<T> {
                         )
                         .await
                         {
-                            Err(Error::InvalidHandler(port_id)) => {
+                            Err(Error::InvalidMessagePort(_, port_id)) => {
                                 error!(
                                     "{}: BlockMessage::Call -> Invalid Handler {port_id:?}.",
                                     meta.instance_name().unwrap(),
@@ -356,7 +356,8 @@ impl<T: Kernel + Send + 'static> TypedBlockWrapper<T> {
                                     "{}: Error in callback. Terminating.",
                                     meta.instance_name().unwrap(),
                                 );
-                                let _ = tx.send(Err(Error::InvalidHandler(port_id)));
+                                let _ = tx
+                                    .send(Err(Error::InvalidMessagePort(Some(block_id), port_id)));
                                 main_inbox
                                     .send(FlowgraphMessage::BlockError {
                                         block_id,

src/runtime/flowgraph.rs

@@ -52,7 +52,7 @@ impl Flowgraph {
         src_port: impl Into<PortId>,
         dst_block: usize,
         dst_port: impl Into<PortId>,
-    ) -> Result<()> {
+    ) -> Result<(), Error> {
         self.topology.as_mut().unwrap().connect_stream(
             src_block,
             src_port.into(),
@@ -70,7 +70,7 @@ impl Flowgraph {
         dst_block: usize,
         dst_port: impl Into<PortId>,
         buffer: B,
-    ) -> Result<()> {
+    ) -> Result<(), Error> {
         self.topology.as_mut().unwrap().connect_stream(
             src_block,
             src_port.into(),
@@ -87,7 +87,7 @@ impl Flowgraph {
         src_port: impl Into<PortId>,
         dst_block: usize,
         dst_port: impl Into<PortId>,
-    ) -> Result<()> {
+    ) -> Result<(), Error> {
         self.topology.as_mut().unwrap().connect_message(
             src_block,
             src_port.into(),
@@ -152,8 +152,8 @@ impl FlowgraphHandle {
                 tx,
             })
             .await
-            .map_err(|_| Error::InvalidBlock)?;
-        rx.await.map_err(|_| Error::HandlerError)?
+            .or(Err(Error::InvalidBlock(block_id)))?;
+        rx.await.or(Err(Error::HandlerError))?
     }
 
     /// Call message handler
@@ -172,7 +172,7 @@ impl FlowgraphHandle {
                 tx,
             })
             .await
-            .map_err(|_| Error::InvalidBlock)?;
+            .map_err(|_| Error::InvalidBlock(block_id))?;
         rx.await.map_err(|_| Error::HandlerError)?
     }
 

src/runtime/mod.rs

@@ -205,21 +205,30 @@ pub enum BlockMessage {
 #[non_exhaustive]
 pub enum Error {
     /// Block does not exist
-    #[error("Block does not exist")]
-    InvalidBlock,
+    #[error("Block {0} does not exist")]
+    InvalidBlock(usize),
     /// Flowgraph does not exist or terminated
     #[error("Flowgraph terminated")]
     FlowgraphTerminated,
-    /// Handler does not exist
-    #[error("Handler does not exist (Id {0:?})")]
-    InvalidHandler(PortId),
-    /// Error in Handler
+    /// Message port does not exist
+    #[error("Block {0:?} does not have message port ({1:?})")]
+    InvalidMessagePort(Option<usize>, PortId),
+    /// Stream port does not exist
+    #[error("Block {0} does not have stream port ({1:?})")]
+    InvalidStreamPort(usize, PortId),
+    /// Connect Error
+    #[error("Connect Error {0}, {1:?} -> {2}, {3:?}")]
+    ConnectError(usize, PortId, usize, PortId),
+    /// Error in handler
     #[error("Error in handler")]
     HandlerError,
     /// Block is already terminated
     #[error("Block already terminated")]
     BlockTerminated,
     /// Runtime error
-    #[error("Error in runtime")]
-    RuntimeError,
+    #[error("Runtime error ({0})")]
+    RuntimeError(String),
+    /// Validation error
+    #[error("Validation error {0}")]
+    ValidationError(String),
 }