fix(graph): bound concurrency caps by the semaphore permit limit
Semaphore::new panics above tokio's MAX_PERMITS, so a fixed or templated map cap (or settings.max_concurrency) that large crashed the run instead of erroring. The validator rejects both at load, resolve_max_concurrency rejects every arm at run time, and the executor checks the graph setting before building its own semaphore, so validation mode makes no difference. Also corrects run_child_agent_with_graph_inputs' contract comment: inputs for a config-only agent are rejected, not ignored.
This commit is contained in:
4 files changed
+134
-3
No files matched your search
@@ -565,7 +565,8 @@ impl Drop for CancelOnDrop {
|
|||||||
|
|
||||||
/// Runs a child agent to completion. When the child is a graph agent,
|
/// Runs a child agent to completion. When the child is a graph agent,
|
||||||
/// `graph_inputs` is overlaid on its `initial_state` before the graph
|
/// `graph_inputs` is overlaid on its `initial_state` before the graph
|
||||||
/// starts; it is ignored for config-only agents.
|
/// starts. Passing inputs to a config-only agent is rejected before the
|
||||||
|
/// child runs, since nothing could consume them.
|
||||||
fn run_child_agent_with_graph_inputs(
|
fn run_child_agent_with_graph_inputs(
|
||||||
mut child_ctx: RequestContext,
|
mut child_ctx: RequestContext,
|
||||||
initial_input: Input,
|
initial_input: Input,
|
||||||
|
|||||||
@@ -109,6 +109,13 @@ impl GraphExecutor {
|
|||||||
let max_iterations = graph.settings.max_loop_iterations;
|
let max_iterations = graph.settings.max_loop_iterations;
|
||||||
let graph_timeout = graph.settings.timeout.and_then(wall_clock);
|
let graph_timeout = graph.settings.timeout.and_then(wall_clock);
|
||||||
let max_concurrency = graph.settings.max_concurrency;
|
let max_concurrency = graph.settings.max_concurrency;
|
||||||
|
if max_concurrency > Semaphore::MAX_PERMITS {
|
||||||
|
bail!(
|
||||||
|
"Graph '{}': settings.max_concurrency {max_concurrency} exceeds the runtime limit of {}",
|
||||||
|
graph.name,
|
||||||
|
Semaphore::MAX_PERMITS
|
||||||
|
);
|
||||||
|
}
|
||||||
let graph = Arc::new(graph);
|
let graph = Arc::new(graph);
|
||||||
let start = Instant::now();
|
let start = Instant::now();
|
||||||
|
|
||||||
@@ -1362,6 +1369,30 @@ nodes:
|
|||||||
assert!(err.contains("sleeper"), "error should name frontier: {err}");
|
assert!(err.contains("sleeper"), "error should name frontier: {err}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn settings_max_concurrency_above_semaphore_limit_errors_instead_of_panicking() {
|
||||||
|
let ws = TestWorkspace::new();
|
||||||
|
let yaml = r#"
|
||||||
|
name: oversized_cap_test
|
||||||
|
start: done
|
||||||
|
settings:
|
||||||
|
validate_before_run: false
|
||||||
|
nodes:
|
||||||
|
done:
|
||||||
|
type: end
|
||||||
|
output: "done"
|
||||||
|
"#;
|
||||||
|
let mut graph: Graph = serde_yaml::from_str(yaml).unwrap();
|
||||||
|
graph.settings.max_concurrency = Semaphore::MAX_PERMITS + 1;
|
||||||
|
let mut ctx = make_ctx();
|
||||||
|
let err = GraphExecutor::new(graph, &ws.dir)
|
||||||
|
.execute(&mut ctx, create_abort_signal())
|
||||||
|
.await
|
||||||
|
.expect_err("an oversized cap is an error, not a panic");
|
||||||
|
let msg = format!("{err:#}");
|
||||||
|
assert!(msg.contains("exceeds the runtime limit of"), "{msg}");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn graph_timeout_zero_is_unbounded() {
|
async fn graph_timeout_zero_is_unbounded() {
|
||||||
if !cmd_available("bash") {
|
if !cmd_available("bash") {
|
||||||
|
|||||||
+40
-2
@@ -337,7 +337,7 @@ fn resolve_max_concurrency(
|
|||||||
Some(ConcurrencyCap::Fixed(0)) => {
|
Some(ConcurrencyCap::Fixed(0)) => {
|
||||||
bail!("map node '{node_id}': max_concurrency 0; expected a positive integer")
|
bail!("map node '{node_id}': max_concurrency 0; expected a positive integer")
|
||||||
}
|
}
|
||||||
Some(ConcurrencyCap::Fixed(n)) => return Ok(*n),
|
Some(ConcurrencyCap::Fixed(n)) => return bounded_cap(node_id, *n),
|
||||||
Some(ConcurrencyCap::Template(t)) => t,
|
Some(ConcurrencyCap::Template(t)) => t,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -359,7 +359,7 @@ fn resolve_max_concurrency(
|
|||||||
};
|
};
|
||||||
|
|
||||||
match resolved {
|
match resolved {
|
||||||
Some(n) if n >= 1 => Ok(n),
|
Some(n) if n >= 1 => bounded_cap(node_id, n),
|
||||||
_ => Err(anyhow!(
|
_ => Err(anyhow!(
|
||||||
"map node '{node_id}': max_concurrency template \"{template}\" resolved to \
|
"map node '{node_id}': max_concurrency template \"{template}\" resolved to \
|
||||||
{value} ({}); expected a positive integer",
|
{value} ({}); expected a positive integer",
|
||||||
@@ -368,6 +368,18 @@ fn resolve_max_concurrency(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `Semaphore::new` panics above `MAX_PERMITS`; a cap that large is a
|
||||||
|
/// configuration error, not a crash.
|
||||||
|
fn bounded_cap(node_id: &str, n: usize) -> Result<usize> {
|
||||||
|
if n > Semaphore::MAX_PERMITS {
|
||||||
|
bail!(
|
||||||
|
"map node '{node_id}': max_concurrency {n} exceeds the runtime limit of {}",
|
||||||
|
Semaphore::MAX_PERMITS
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Ok(n)
|
||||||
|
}
|
||||||
|
|
||||||
/// Pre-provision one teammate identity per item before any chain starts, so a
|
/// Pre-provision one teammate identity per item before any chain starts, so a
|
||||||
/// fast sibling can message one still waiting on the semaphore (the message
|
/// fast sibling can message one still waiting on the semaphore (the message
|
||||||
/// queues in the pre-created inbox). The identity lives for the item's whole
|
/// queues in the pre-created inbox). The identity lives for the item's whole
|
||||||
@@ -686,6 +698,32 @@ mod tests {
|
|||||||
assert!(msg.contains("map node 'm': max_concurrency \"4\""), "{msg}");
|
assert!(msg.contains("map node 'm': max_concurrency \"4\""), "{msg}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resolve_max_concurrency_fixed_above_semaphore_limit_errors() {
|
||||||
|
let state = StateManager::new(HashMap::new());
|
||||||
|
let cap = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS + 1));
|
||||||
|
let msg = resolve(cap, &state).unwrap_err().to_string();
|
||||||
|
assert_eq!(
|
||||||
|
msg,
|
||||||
|
format!(
|
||||||
|
"map node 'm': max_concurrency {} exceeds the runtime limit of {}",
|
||||||
|
Semaphore::MAX_PERMITS + 1,
|
||||||
|
Semaphore::MAX_PERMITS
|
||||||
|
)
|
||||||
|
);
|
||||||
|
// The limit itself is fine.
|
||||||
|
let cap = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS));
|
||||||
|
assert_eq!(resolve(cap, &state).unwrap(), Semaphore::MAX_PERMITS);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resolve_max_concurrency_template_above_semaphore_limit_errors() {
|
||||||
|
let state = state_with("budget", Value::from((Semaphore::MAX_PERMITS as u64) + 1));
|
||||||
|
let cap = Some(ConcurrencyCap::Template("{{budget}}".into()));
|
||||||
|
let msg = resolve(cap, &state).unwrap_err().to_string();
|
||||||
|
assert!(msg.contains("exceeds the runtime limit of"), "{msg}");
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn resolve_max_concurrency_accepts_padded_template() {
|
fn resolve_max_concurrency_accepts_padded_template() {
|
||||||
let state = state_with("budget", Value::from(3));
|
let state = state_with("budget", Value::from(3));
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ use anyhow::{Result, bail};
|
|||||||
use std::collections::{BTreeMap, HashSet, VecDeque};
|
use std::collections::{BTreeMap, HashSet, VecDeque};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use tokio::sync::Semaphore;
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct ValidationError {
|
pub struct ValidationError {
|
||||||
@@ -537,6 +538,12 @@ impl GraphValidator {
|
|||||||
deadlock the executor",
|
deadlock the executor",
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
if graph.settings.max_concurrency > Semaphore::MAX_PERMITS {
|
||||||
|
result.error(ValidationError::new(format!(
|
||||||
|
"settings.max_concurrency must be <= {} (got {}); the runtime cannot allocate more permits",
|
||||||
|
Semaphore::MAX_PERMITS, graph.settings.max_concurrency
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
for (node_id, node) in &graph.nodes {
|
for (node_id, node) in &graph.nodes {
|
||||||
if let NodeType::Map(m) = &node.node_type
|
if let NodeType::Map(m) = &node.node_type
|
||||||
@@ -548,6 +555,18 @@ impl GraphValidator {
|
|||||||
would deadlock the executor",
|
would deadlock the executor",
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
if let NodeType::Map(m) = &node.node_type
|
||||||
|
&& let Some(ConcurrencyCap::Fixed(n)) = m.max_concurrency
|
||||||
|
&& n > Semaphore::MAX_PERMITS
|
||||||
|
{
|
||||||
|
result.error(ValidationError::with_node(
|
||||||
|
node_id,
|
||||||
|
format!(
|
||||||
|
"map node's `max_concurrency` must be <= {} (got {n}); the runtime cannot allocate more permits",
|
||||||
|
Semaphore::MAX_PERMITS
|
||||||
|
),
|
||||||
|
));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -3158,6 +3177,24 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn settings_max_concurrency_above_semaphore_limit_errors() {
|
||||||
|
let mut graph = graph_with(vec![("e", end_node("e"))], "e");
|
||||||
|
graph.settings.max_concurrency = Semaphore::MAX_PERMITS + 1;
|
||||||
|
|
||||||
|
let result = validator().validate(&graph);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
result.errors.iter().any(|e| e.message.contains(&format!(
|
||||||
|
"settings.max_concurrency must be <= {} (got {})",
|
||||||
|
Semaphore::MAX_PERMITS,
|
||||||
|
Semaphore::MAX_PERMITS + 1
|
||||||
|
))),
|
||||||
|
"expected graph-level max_concurrency upper-bound error: {:?}",
|
||||||
|
result.errors
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn settings_max_concurrency_default_is_valid() {
|
fn settings_max_concurrency_default_is_valid() {
|
||||||
let graph = graph_with(vec![("e", end_node("e"))], "e");
|
let graph = graph_with(vec![("e", end_node("e"))], "e");
|
||||||
@@ -3458,6 +3495,30 @@ mod tests {
|
|||||||
assert_eq!(w[1].node_id.as_deref(), Some("l"));
|
assert_eq!(w[1].node_id.as_deref(), Some("l"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn map_max_concurrency_above_semaphore_limit_errors() {
|
||||||
|
let mut map = map_node_basic("m", "br", Some("end"));
|
||||||
|
if let NodeType::Map(ref mut mm) = map.node_type {
|
||||||
|
mm.max_concurrency = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS + 1));
|
||||||
|
}
|
||||||
|
let branch = llm_with_state_updates("br", &[("output", "{{output}}")], None);
|
||||||
|
let graph = graph_with(
|
||||||
|
vec![("m", map), ("br", branch), ("end", end_node("end"))],
|
||||||
|
"m",
|
||||||
|
);
|
||||||
|
|
||||||
|
let result = validator().validate(&graph);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
result.errors.iter().any(|e| e
|
||||||
|
.message
|
||||||
|
.contains(&format!("must be <= {}", Semaphore::MAX_PERMITS))
|
||||||
|
&& e.node_id.as_deref() == Some("m")),
|
||||||
|
"expected map max_concurrency upper-bound error: {:?}",
|
||||||
|
result.errors
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn map_max_concurrency_zero_errors() {
|
fn map_max_concurrency_zero_errors() {
|
||||||
let mut map = map_node_basic("m", "br", Some("end"));
|
let mut map = map_node_basic("m", "br", Some("end"));
|
||||||
|
|||||||
Reference in new issue
Block a user