Repository: eclipse-zenoh-flow/zenoh-flow (language: rust)

Issue:
[Bug] Built-in Zenoh source subscribes at creation
### Describe the bug

The built-in Zenoh source, which internally wraps subscriber(s), declares its subscriber(s) on Zenoh when it is created.
Because, in Zenoh-Flow, a newly created data flow is not started immediately -- it requires a separate explicit action -- this means that the publishers will start storing the publications but, as long as the data flow is not started, this data will never be consumed.

This will cause congestion and other nefarious effects.

### To reproduce

1. Start a Zenoh-Flow daemon.
2. Create a data flow with a built-in Zenoh source (e.g. the getting-started data flow): `zfctl instance create path/to/flow.yml`
3. Start publishing on one of the key expressions subscribed by the Zenoh source.
4. Start the data flow: `zfctl instance start <uuid>`.

Note that the flow will already produce results, indicating that the publications were stored by the built-in Zenoh source.

### System info

Any platform.


Discussion / hints:

Fix patch:
diff --git a/zenoh-flow-runtime/src/runners/builtin/zenoh/source.rs b/zenoh-flow-runtime/src/runners/builtin/zenoh/source.rs
index 923197e1..bad03a57 100644
--- a/zenoh-flow-runtime/src/runners/builtin/zenoh/source.rs
+++ b/zenoh-flow-runtime/src/runners/builtin/zenoh/source.rs
@@ -72,9 +72,6 @@ impl<'a> ZenohSource<'a> {
             futs: Arc::new(Mutex::new(Vec::with_capacity(key_exprs.len()))),
         };
 
-        // NOTE: Calling this function avoids repeating code initialising the `futs` and `subscribers`.
-        zenoh_source.on_resume().await?;
-
         Ok(zenoh_source)
     }
 }
diff --git a/zenoh-flow-runtime/src/runners/mod.rs b/zenoh-flow-runtime/src/runners/mod.rs
index 64f89fab..4e20839d 100644
--- a/zenoh-flow-runtime/src/runners/mod.rs
+++ b/zenoh-flow-runtime/src/runners/mod.rs
@@ -26,18 +26,12 @@ use tracing::Instrument;
 use zenoh_flow_commons::{NodeId, Result};
 use zenoh_flow_nodes::prelude::Node;
 
-enum State {
-    Uninitialized,
-    Initialized,
-}
-
 /// A `Runner` takes care of running a `Node`.
 ///
 /// Each Runner runs in a separate task.
 pub(crate) struct Runner {
     pub(crate) id: NodeId,
     node: Arc<dyn Node>,
-    state: State,
     handle: Option<JoinHandle<()>>,
     // The `_library` field is used solely for its `Arc`. We need to keep track of how many `Runners` are using the
     // `Library` such that once that number reaches 0, we drop the library.
@@ -59,7 +53,6 @@ impl Runner {
         Self {
             id,
             node,
-            state: State::Uninitialized,
             handle: None,
             _library: library,
         }
@@ -79,12 +72,10 @@ impl Runner {
             return Ok(());
         }
 
-        if matches!(self.state, State::Initialized) {
-            self.node
-                .on_resume()
-                .await
-                .with_context(|| format!("{}: call to `on_resume` failed", self.id))?;
-        }
+        self.node
+            .on_resume()
+            .await
+            .with_context(|| format!("{}: call to `on_resume` failed", self.id))?;
 
         let id = self.id.clone();
         let node = self.node.clone();
@@ -108,7 +99,6 @@ impl Runner {
             .instrument(iteration_span),
         ));
 
-        self.state = State::Initialized;
         Ok(())
     }
 


Test patch: