Error if node with git field was not built before dataflow start

binaries/daemon/src/lib.rs

@@ -1117,8 +1117,19 @@ impl Daemon {
 
         let mut stopped = Vec::new();
 
-        let node_working_dirs = build_id
-            .and_then(|build_id| self.builds.get(&build_id))
+        let build_info = build_id.and_then(|build_id| self.builds.get(&build_id));
+        let node_with_git_source = nodes.values().find(|n| n.has_git_source());
+        if let Some(git_node) = node_with_git_source {
+            if build_info.is_none() {
+                eyre::bail!(
+                    "node {} has git source, but no `dora build` was run yet\n\n\
+                    nodes with a `git` field must be built using `dora build` before starting the \
+                    dataflow",
+                    git_node.id
+                )
+            }
+        }
+        let node_working_dirs = build_info
             .map(|info| info.node_working_dirs.clone())
             .unwrap_or_default();
 
@@ -1186,12 +1197,16 @@ impl Daemon {
                     .entry(node.id.clone())
                     .or_insert_with(|| Arc::new(ArrayQueue::new(STDERR_LOG_LINES)))
                     .clone();
-                logger
-                    .log(LogLevel::Info, Some("daemon".into()), "spawning")
-                    .await;
-                let node_working_dir = node_working_dirs
-                    .get(&node_id)
-                    .cloned()
+
+                let configured_node_working_dir = node_working_dirs.get(&node_id).cloned();
+                if configured_node_working_dir.is_none() && node.has_git_source() {
+                    eyre::bail!(
+                        "node {} has git source, but no git clone directory was found for it\n\n\
+                        try running `dora build` again",
+                        node.id
+                    )
+                }
+                let node_working_dir = configured_node_working_dir
                     .or_else(|| {
                         node.deploy
                             .as_ref()

binaries/daemon/src/spawn.rs

@@ -357,7 +357,24 @@ impl PreparedNode {
 
     pub async fn spawn(mut self, logger: &mut NodeLogger<'_>) -> eyre::Result<RunningNode> {
         let mut child = match &mut self.command {
-            Some(command) => command.spawn().wrap_err(self.spawn_error_msg)?,
+            Some(command) => {
+                let std_command = command.as_std();
+                logger
+                    .log(
+                        LogLevel::Info,
+                        Some("spawner".into()),
+                        format!(
+                            "spawning `{}` in `{}`",
+                            std_command.get_program().to_string_lossy(),
+                            std_command
+                                .get_current_dir()
+                                .unwrap_or(Path::new("<unknown>"))
+                                .display(),
+                        ),
+                    )
+                    .await;
+                command.spawn().wrap_err(self.spawn_error_msg)?
+            }
             None => {
                 return Ok(RunningNode {
                     pid: None,
@@ -672,13 +689,6 @@ async fn path_spawn_command(
                     cmd
                 }
                 _ => {
-                    logger
-                        .log(
-                            LogLevel::Info,
-                            Some("spawner".into()),
-                            format!("spawning: {}", resolved_path.display()),
-                        )
-                        .await;
                     if uv {
                         let mut cmd = tokio::process::Command::new("uv");
                         cmd.arg("run");

libraries/message/src/descriptor.rs

@@ -13,7 +13,7 @@
     path::PathBuf,
 };
 
 pub const SHELL_SOURCE: &str = "shell";
 /// Set the [`Node::path`] field to this value to treat the node as a
 /// [_dynamic node_](https://docs.rs/dora-node-api/latest/dora_node_api/).
 pub const DYNAMIC_SOURCE: &str = "dynamic";
@@ -92,7 +92,7 @@
 
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct Deploy {
     /// Target machine for deployment
     pub machine: Option<String>,
     /// Working directory for the deployment
@@ -100,7 +100,7 @@
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
 pub struct Debug {
     /// Whether to publish all messages to Zenoh for debugging
     #[serde(default)]
     pub publish_all_messages_to_zenoh: bool,
@@ -469,19 +469,28 @@
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ResolvedNode {
     pub id: NodeId,
     pub name: Option<String>,
     pub description: Option<String>,
     pub env: Option<BTreeMap<String, EnvValue>>,
 
     #[serde(default)]
     pub deploy: Option<Deploy>,
 
     #[serde(flatten)]
     pub kind: CoreNodeKind,
 }
 
+impl ResolvedNode {
+    pub fn has_git_source(&self) -> bool {
+        self.kind
+            .as_custom()
+            .map(|n| n.source.is_git())
+            .unwrap_or_default()
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 #[allow(clippy::large_enum_variant)]
@@ -492,6 +501,15 @@
     Custom(CustomNode),
 }
 
+impl CoreNodeKind {
+    pub fn as_custom(&self) -> Option<&CustomNode> {
+        match self {
+            CoreNodeKind::Runtime(_) => None,
+            CoreNodeKind::Custom(custom_node) => Some(custom_node),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
 #[serde(transparent)]
 pub struct RuntimeNode {
@@ -640,6 +658,12 @@
     },
 }
 
+impl NodeSource {
+    pub fn is_git(&self) -> bool {
+        matches!(self, Self::GitBranch { .. })
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
 pub enum ResolvedNodeSource {
     Local,