Streaming setting from Python API + no more relative paths (in non simple cases)

CHANGELOG.md

@@ -5,6 +5,15 @@
 ### New features
 
 * Python API `crash_limit` now accepts `"unlimited"` and `"never-restart"` (and `0` is rejected)
+* Python API now allows to enable stdio streaming
+
+### Changes
+
+* Relative `cwd`, `stdout`, `stderr` and `stream` paths are now rejected in the Python API and in job files.
+  Relative `cwd`, `stdout` and `stderr` paths are also rejected in array jobs submitted via the CLI.
+  Use an absolute path or a path starting with `%{SUBMIT_DIR}` (or `%{CWD}` for `stdout`/`stderr`).
+  A relative `--stream` path in the CLI is resolved against the submit directory.
+* The stream path cannot contain task-specific placeholders (`%{TASK_ID}`, `%{INSTANCE_ID}`, `%{CWD}`).
 
 ## v0.27.0
 

crates/hyperqueue/src/client/commands/outputlog.rs

@@ -7,7 +7,7 @@ use tako::JobId;
 
 #[derive(Parser)]
 pub struct OutputLogOpts {
-    /// Path of the log file
+    /// Path of the directory with streamed output (see `hq submit --stream`)
     path: PathBuf,
 
     /// Filter files for the given server instance

crates/hyperqueue/src/client/commands/submit/command.rs

@@ -1,6 +1,7 @@
 use super::directives::parse_hq_directives;
 use crate::client::commands::duration_doc;
 use crate::client::commands::submit::directives::parse_hq_directives_from_file;
+use crate::client::commands::submit::paths::{validate_stream_path, validate_task_paths};
 use crate::client::commands::wait::{wait_for_jobs, wait_for_jobs_with_progress};
 use crate::client::globalsettings::GlobalSettings;
 use crate::client::resources::{
@@ -316,7 +317,7 @@ pub struct SubmitJobTaskConfOpts {
     )]
     time_limit: Option<Duration>,
 
-    /// Stream the output of tasks into the given log file
+    /// Stream the output of tasks into the given directory
     #[arg(long)]
     stream: Option<PathBuf>,
 
@@ -667,6 +668,8 @@ pub async fn submit_computation(
     let task_count = ids.id_count();
 
     check_suspicious_options(&opts, task_count)?;
+    let is_array =
+        opts.conf.array.is_some() || opts.conf.each_line.is_some() || opts.conf.from_json.is_some();
 
     let JobSubmitOpts {
         commands,
@@ -765,6 +768,14 @@ pub async fn submit_computation(
         job_id,
     };
 
+    // Tasks of an array share the paths, so relative paths could not be told apart.
+    // For a single task, relative paths are resolved against the submit dir.
+    // The stream path is shared by all tasks and it is always resolved against the submit dir.
+    if is_array {
+        validate_task_paths(&request.submit_desc)?;
+    }
+    validate_stream_path(&request.submit_desc)?;
+
     send_submit_request(
         gsettings,
         session,
@@ -961,7 +972,7 @@ fn warn_unknown_placeholders(opts: &JobSubmitOpts) {
         opts.conf.stderr.as_ref().and_then(|arg| arg.path()),
         "stderr path",
     );
-    check(opts.conf.stream.as_deref(), "log path");
+    check(opts.conf.stream.as_deref(), "stream path");
     check(opts.conf.cwd.as_deref(), "working directory path");
 }
 

crates/hyperqueue/src/client/commands/submit/jobfile.rs

@@ -5,6 +5,7 @@ use crate::client::commands::submit::defs::{
     ArrayDef, JobDef, StdioDefFull, StdioDefInput, TaskDef,
 };
 use crate::client::commands::submit::defs::{PinMode as PinModeDef, TaskConfigDef};
+use crate::client::commands::submit::paths::validate_submit_paths;
 use crate::client::globalsettings::GlobalSettings;
 use crate::common::arraydef::IntArray;
 use crate::common::utils::fs::get_current_dir;
@@ -238,6 +239,7 @@ pub async fn submit_computation_from_job_file(
             })?)?
         };
     let request = build_job_submit(jdef, opts.job)?;
+    validate_submit_paths(&request.submit_desc)?;
     send_submit_request(gsettings, session, request, false, false, None).await
 }
 

crates/hyperqueue/src/client/commands/submit/mod.rs

@@ -2,6 +2,7 @@ pub mod command;
 pub mod defs;
 pub mod directives;
 mod jobfile;
+pub mod paths;
 
 pub use command::SubmitJobTaskConfOpts;
 pub use command::{JobSubmitOpts, submit_computation};

crates/hyperqueue/src/client/commands/submit/paths.rs

@@ -0,0 +1,194 @@
+use crate::common::placeholders::{
+    CWD_PLACEHOLDER, INSTANCE_ID_PLACEHOLDER, SUBMIT_DIR_PLACEHOLDER, StringPart,
+    TASK_ID_PLACEHOLDER, parse_resolvable_string,
+};
+use crate::transfer::messages::{
+    JobSubmitDescription, JobTaskDescription, TaskDescription, TaskKind,
+};
+use std::path::Path;
+use tako::program::StdioDef;
+
+const ABSOLUTE_PLACEHOLDERS: [&str; 1] = [SUBMIT_DIR_PLACEHOLDER];
+const ABSOLUTE_OUTPUT_PLACEHOLDERS: [&str; 2] = [SUBMIT_DIR_PLACEHOLDER, CWD_PLACEHOLDER];
+/// The stream path is shared by all tasks of a job, so it cannot depend on a task.
+const TASK_PLACEHOLDERS: [&str; 3] = [
+    TASK_ID_PLACEHOLDER,
+    INSTANCE_ID_PLACEHOLDER,
+    CWD_PLACEHOLDER,
+];
+
+fn check_path(path: &Path, name: &str, placeholders: &[&str]) -> anyhow::Result<()> {
+    if path.is_absolute() {
+        return Ok(());
+    }
+    let starts_with_absolute_placeholder = path.to_str().is_some_and(|path| {
+        matches!(
+            parse_resolvable_string(path).first(),
+            Some(StringPart::Placeholder(placeholder)) if placeholders.contains(placeholder)
+        )
+    });
+    if starts_with_absolute_placeholder {
+        return Ok(());
+    }
+    anyhow::bail!(
+        "Path `{}` in `{name}` is not absolute. Use an absolute path or a path that starts with {}.",
+        path.display(),
+        placeholders
+            .iter()
+            .map(|placeholder| format!("`%{{{placeholder}}}`"))
+            .collect::<Vec<_>>()
+            .join(" or ")
+    )
+}
+
+fn check_stdio(stdio: &StdioDef, name: &str) -> anyhow::Result<()> {
+    match stdio {
+        StdioDef::File { path, .. } => check_path(path, name, &ABSOLUTE_OUTPUT_PLACEHOLDERS),
+        StdioDef::Null | StdioDef::Pipe => Ok(()),
+    }
+}
+
+fn check_task(desc: &TaskDescription) -> anyhow::Result<()> {
+    match &desc.kind {
+        TaskKind::ExternalProgram(program) => {
+            let program = &program.program;
+            check_path(&program.cwd, "cwd", &ABSOLUTE_PLACEHOLDERS)?;
+            check_stdio(&program.stdout, "stdout")?;
+            check_stdio(&program.stderr, "stderr")
+        }
+    }
+}
+
+/// Checks that all task paths in the submit description are absolute or start with a placeholder.
+/// The stream path is not checked.
+pub fn validate_task_paths(desc: &JobSubmitDescription) -> anyhow::Result<()> {
+    match &desc.task_desc {
+        JobTaskDescription::Array { task_desc, .. } => check_task(task_desc),
+        JobTaskDescription::Graph { tasks, .. } => {
+            for task in tasks {
+                check_task(&task.task_desc)?;
+            }
+            Ok(())
+        }
+    }
+}
+
+fn check_stream_placeholders(path: &Path) -> anyhow::Result<()> {
+    let Some(path_str) = path.to_str() else {
+        return Ok(());
+    };
+    for part in parse_resolvable_string(path_str) {
+        if let StringPart::Placeholder(placeholder) = part
+            && TASK_PLACEHOLDERS.contains(&placeholder)
+        {
+            anyhow::bail!(
+                "Path `{}` in `stream` contains placeholder `%{{{placeholder}}}` that depends on a task. \
+                 The stream path is shared by all tasks of a job.",
+                path.display()
+            );
+        }
+    }
+    Ok(())
+}
+
+/// Checks that the stream path does not contain task-specific placeholders.
+/// A relative stream path is allowed; it is resolved against the submit dir.
+pub fn validate_stream_path(desc: &JobSubmitDescription) -> anyhow::Result<()> {
+    if let Some(path) = &desc.stream_path {
+        check_stream_placeholders(path)?;
+    }
+    Ok(())
+}
+
+/// Checks that all paths in the submit description (including the stream path)
+/// are absolute or start with a placeholder.
+pub fn validate_submit_paths(desc: &JobSubmitDescription) -> anyhow::Result<()> {
+    validate_task_paths(desc)?;
+    validate_stream_path(desc)?;
+    if let Some(path) = &desc.stream_path {
+        check_path(path, "stream", &ABSOLUTE_PLACEHOLDERS)?;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        ABSOLUTE_OUTPUT_PLACEHOLDERS, ABSOLUTE_PLACEHOLDERS, check_path, check_stream_placeholders,
+    };
+    use std::path::Path;
+
+    fn ok(path: &str, placeholders: &[&str]) -> bool {
+        check_path(Path::new(path), "cwd", placeholders).is_ok()
+    }
+
+    #[test]
+    fn accept_absolute_paths_and_absolute_placeholders() {
+        for path in [
+            "/",
+            "/foo/bar",
+            "%{SUBMIT_DIR}",
+            "%{SUBMIT_DIR}/%{TASK_ID}/x",
+        ] {
+            assert!(ok(path, &ABSOLUTE_PLACEHOLDERS), "{path}");
+        }
+        assert!(ok("%{CWD}/%{TASK_ID}.out", &ABSOLUTE_OUTPUT_PLACEHOLDERS));
+    }
+
+    #[test]
+    fn reject_relative_paths() {
+        for path in [
+            "",
+            "foo",
+            "foo/bar",
+            "./foo",
+            "../foo",
+            "a/%{SUBMIT_DIR}",
+            "%foo",
+        ] {
+            assert!(!ok(path, &ABSOLUTE_PLACEHOLDERS), "{path}");
+        }
+    }
+
+    #[test]
+    fn reject_placeholders_that_do_not_produce_an_absolute_path() {
+        for path in [
+            "%{TASK_ID}/x",
+            "%{JOB_ID}",
+            "%{INSTANCE_ID}/x",
+            "%{SERVER_UID}/x",
+            "%{UNKNOWN}/x",
+        ] {
+            assert!(!ok(path, &ABSOLUTE_OUTPUT_PLACEHOLDERS), "{path}");
+        }
+    }
+
+    #[test]
+    fn cwd_placeholder_only_in_output_paths() {
+        assert!(!ok("%{CWD}/x", &ABSOLUTE_PLACEHOLDERS));
+        assert!(ok("%{CWD}/x", &ABSOLUTE_OUTPUT_PLACEHOLDERS));
+    }
+
+    #[test]
+    fn stream_placeholders() {
+        for path in [
+            "foo",
+            "/foo",
+            "%{SUBMIT_DIR}/foo",
+            "foo-%{JOB_ID}",
+            "%{SERVER_UID}/x",
+        ] {
+            assert!(check_stream_placeholders(Path::new(path)).is_ok(), "{path}");
+        }
+        for path in [
+            "foo-%{TASK_ID}",
+            "%{SUBMIT_DIR}/%{INSTANCE_ID}",
+            "%{CWD}/foo",
+        ] {
+            assert!(
+                check_stream_placeholders(Path::new(path)).is_err(),
+                "{path}"
+            );
+        }
+    }
+}

crates/pyhq/python/hyperqueue/ffi/protocol.py

@@ -53,3 +53,4 @@ class JobDescription:
     name: Optional[str]
     tasks: List[TaskDescription]
     max_fails: Optional[int]
+    stream_path: Optional[str] = None