vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: bound forge calls in job executors

Browse source 
forgejo-api's client has no request timeout, so a forge that accepts a
connection and never answers held a forge-calling job node forever:
the node never reached Done or Failed, and kept its executor task.

Every executor arm that awaits the forge now runs its work under the
same deadline the periodic passes use. On elapsing, the node fails
through its existing Outcome::Failed path with
"<node>: no answer from the forge within 120s".

Nodes: create_repo, create_forge_user, add_repo_member,
grant_docs_read, init_agent_config_repo, mint_agent_forge_token,
mint_hive_forge_token.

The bound is per node, not per request: the nodes make 1 to about 15
sequential forge and store requests each (mint_hive_forge_token the
most), and the requests live inside forge::Client methods the periodic
passes share. One 2-minute bound covers both, so #5065's
within_pass_deadline / PASS_DEADLINE become within_forge_deadline /
FORGE_DEADLINE, with a log line that no longer says "retrying next
tick" (true of a pass, not of a node).

The InitAgentConfigRepo arm is lifted out of run_swarm_node, like its
siblings, to keep that function under clippy::too_many_lines.

Refs #5066
This commit is contained in:
atlas 2026-10-09 20:00:19 +02:00 • committed by mara
commit 6f8a6ab8e0
8 changed files with 190 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

8
swarm-controller/src/config_pr.rs
View file

@ -227,9 +227,11 @@ pub fn spawn(client: Arc<Client>) -> Arc<ConfigPrCache> {
let mut ticker = tokio::time::interval(POLL_INTERVAL);
loop {
ticker.tick().await;
let Some(scan) =
crate::forge::within_pass_deadline("config-pr poll", client.list_open_config_prs())
.await
let Some(scan) = crate::forge::within_forge_deadline(
"config-pr poll",
client.list_open_config_prs(),
)
.await
else {
continue;
};

46
swarm-controller/src/forge.rs
View file

@ -231,26 +231,27 @@ where
}
}
/// The longest one periodic forge pass may run before it is abandoned until
/// its next tick. forgejo-api's client has no request timeout, so without
/// this a forge that accepts a connection and never answers holds the pass,
/// and the loop running it, forever. Under every pass's 5-minute interval and
/// far over what a pass that is making progress takes.
pub(crate) const PASS_DEADLINE: std::time::Duration = std::time::Duration::from_mins(2);
/// The longest one unit of forge work, a periodic pass or a job node, may run
/// before it is abandoned. forgejo-api's client has no request timeout, so
/// without this a forge that accepts a connection and never answers holds the
/// pass's loop, or the node, forever. Under every pass's 5-minute interval,
/// and far over what the longest node (`MintHiveForgeToken`, about fifteen
/// sequential forge and store requests) takes while it is making progress.
pub(crate) const FORGE_DEADLINE: std::time::Duration = std::time::Duration::from_mins(2);
/// Run `fut`, one periodic pass named `pass` in the log, under
/// [`PASS_DEADLINE`]. `None` means it was abandoned at the deadline; that is
/// logged here, and the caller moves on to its next tick.
pub(crate) async fn within_pass_deadline<T>(
pass: &str,
/// Run `fut`, the forge work named `work` in the log, under
/// [`FORGE_DEADLINE`]. `None` means it was abandoned at the deadline; that is
/// logged here.
pub(crate) async fn within_forge_deadline<T>(
work: &str,
fut: impl std::future::Future<Output = T>,
) -> Option<T> {
let result = tokio::time::timeout(PASS_DEADLINE, fut).await.ok();
let result = tokio::time::timeout(FORGE_DEADLINE, fut).await.ok();
if result.is_none() {
tracing::warn!(
pass,
deadline_s = PASS_DEADLINE.as_secs(),
"forge pass: no answer within the deadline; abandoned, retrying next tick"
work,
deadline_s = FORGE_DEADLINE.as_secs(),
"forge work: no answer within the deadline; abandoned"
);
}
result
@ -1411,6 +1412,15 @@ fn conflict_error() -> ForgejoError {
ForgejoError::ApiError(ApiErrorKind::Other(StatusCode::CONFLICT).into())
}
#[cfg(test)]
impl Client {
/// A client for a test's own forge at `url`, holding a placeholder token.
pub(crate) fn for_test_forge(url: url::Url) -> Self {
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap();
Self { api, url }
}
}
#[cfg(test)]
mod tests {
use super::*;
@ -1483,14 +1493,14 @@ mod tests {
#[tokio::test(start_paused = true)]
async fn a_pass_that_never_answers_is_abandoned_at_the_deadline() {
let start = tokio::time::Instant::now();
let got = within_pass_deadline("test", std::future::pending::<()>()).await;
let got = within_forge_deadline("test", std::future::pending::<()>()).await;
assert_eq!(got, None);
assert_eq!(start.elapsed(), PASS_DEADLINE);
assert_eq!(start.elapsed(), FORGE_DEADLINE);
}
#[tokio::test(start_paused = true)]
async fn a_pass_that_answers_returns_its_result() {
assert_eq!(within_pass_deadline("test", async { 7 }).await, Some(7));
assert_eq!(within_forge_deadline("test", async { 7 }).await, Some(7));
}
#[tokio::test]

2
swarm-controller/src/forge/agent_token.rs
View file

@ -370,7 +370,7 @@ pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + Send + 'static
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
loop {
ticker.tick().await;
let Some(outcome) = super::within_pass_deadline("agent forge tokens", async {
let Some(outcome) = super::within_forge_deadline("agent forge tokens", async {
let observed = client.observe_all().await?;
let legacy = super::legacy_tokens::sweepable(&observed);
client.sweep_legacy_tokens(&legacy).await;

2
swarm-controller/src/forge/docs_grant.rs
View file

@ -185,7 +185,7 @@ pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + Send + 'static
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
loop {
ticker.tick().await;
let Some(agents) = super::within_pass_deadline("docs grants", async {
let Some(agents) = super::within_forge_deadline("docs grants", async {
run_pass(&client, read_roster().await).await
})
.await

2
swarm-controller/src/forge/hive_token.rs
View file

@ -387,7 +387,7 @@ pub fn spawn(
loop {
ticker.tick().await;
let Some(outcome) =
super::within_pass_deadline("hive forge tokens", client.observe_hives(&hives))
super::within_forge_deadline("hive forge tokens", client.observe_hives(&hives))
.await
else {
continue;

2
swarm-controller/src/forge/objects.rs
View file

@ -1083,7 +1083,7 @@ pub fn spawn(client: Arc<Client>) {
loop {
ticker.tick().await;
let Some(outcome) =
super::within_pass_deadline("swarm forge objects", client.reconcile(&desired))
super::within_forge_deadline("swarm forge objects", client.reconcile(&desired))
.await
else {
continue;

2
swarm-controller/src/forge/runner_token.rs
View file

@ -77,7 +77,7 @@ pub fn spawn(client: Arc<Client>) {
let mut last_written: Option<String> = None;
loop {
ticker.tick().await;
let Some(outcome) = super::within_pass_deadline(
let Some(outcome) = super::within_forge_deadline(
"runner registration token",
pass(&client, last_written.as_deref()),
)

174
swarm-controller/src/main.rs
View file

@ -324,7 +324,7 @@ async fn run_swarm_node(
not in the swarm secret store)"
.to_owned(),
),
Some(client) => match client.create_repo(&agent).await {
Some(client) => match forge_node("create_repo", client.create_repo(&agent)).await {
Ok(full_name) => {
tracing::info!(%full_name, "swarm jobq: create_repo done");
Outcome::Done
@ -338,10 +338,12 @@ async fn run_swarm_node(
not in the swarm secret store)"
.to_owned(),
),
Some(client) => match client.ensure_agent_user(&agent).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
},
Some(client) => {
match forge_node("create_forge_user", client.ensure_agent_user(&agent)).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
}
},
SwarmNodeKind::AddRepoMember { agent } => match forge {
None => Outcome::Failed(
@ -349,23 +351,15 @@ async fn run_swarm_node(
not in the swarm secret store)"
.to_owned(),
),
Some(client) => match client.add_repo_member(&agent, &agent).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
},
Some(client) => {
match forge_node("add_repo_member", client.add_repo_member(&agent, &agent)).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
}
},
SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(forge, &agent).await,
SwarmNodeKind::InitAgentConfigRepo { agent } => match forge {
None => Outcome::Failed(
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
not in the swarm secret store)"
.to_owned(),
),
Some(client) => match client.seed_agent_config(&agent, &agent).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
},
},
SwarmNodeKind::InitAgentConfigRepo { agent } => init_agent_config_repo(forge, &agent).await,
SwarmNodeKind::MintAgentIdentity { agent } => mint_identity(&agent).await,
SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(forge, &agent).await,
SwarmNodeKind::MintAgentMatrixAccount { agent } => {
@ -395,6 +389,23 @@ async fn run_swarm_node(
(builder, outcome)
}
/// A forge-calling node's work, run under [`forge::FORGE_DEADLINE`]. Elapsing
/// is the node's own error, naming `node`, so it fails like any other refused
/// step instead of holding its executor slot.
async fn forge_node<T>(
node: &str,
work: impl std::future::Future<Output = Result<T>>,
) -> Result<T> {
forge::within_forge_deadline(node, work)
.await
.unwrap_or_else(|| {
Err(anyhow::anyhow!(
"{node}: no answer from the forge within {}s",
forge::FORGE_DEADLINE.as_secs()
))
})
}
/// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under
/// `clippy::too_many_lines`.
async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome {
@ -421,7 +432,12 @@ async fn mint_forge_token(
.to_owned(),
);
};
match client.ensure_agent_forge_token(agent).await {
match forge_node(
"mint_agent_forge_token",
client.ensure_agent_forge_token(agent),
)
.await
{
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
@ -441,7 +457,32 @@ async fn grant_docs_read(
.to_owned(),
);
};
match client.grant_docs_read(agent).await {
match forge_node("grant_docs_read", client.grant_docs_read(agent)).await {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
}
/// The `InitAgentConfigRepo` arm, lifted out for the same reason.
async fn init_agent_config_repo(
forge: Option<Arc<forge::Client>>,
agent: &str,
) -> hive_jobq::scheduler::Outcome {
use hive_jobq::scheduler::Outcome;
let Some(client) = forge else {
return Outcome::Failed(
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
not in the swarm secret store)"
.to_owned(),
);
};
match forge_node(
"init_agent_config_repo",
client.seed_agent_config(agent, agent),
)
.await
{
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
@ -471,7 +512,12 @@ async fn mint_hive_forge_token(
.to_owned(),
);
};
match client.ensure_hive_forge_token(hive).await {
match forge_node(
"mint_hive_forge_token",
client.ensure_hive_forge_token(hive),
)
.await
{
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
@ -4188,6 +4234,88 @@ mod tests {
assert_eq!(queued.node_id, node.id.get());
}
/// Every node whose first request goes to the forge, run against a forge
/// that accepts the connection and never answers: each ends `Failed`
/// with an error naming the node instead of holding its executor slot.
/// `MintAgentForgeToken` is not in the list: it logs in to the secret
/// store before its first forge request, and no test here has a store.
#[tokio::test(start_paused = true)]
async fn a_forge_node_fails_when_the_forge_never_answers() {
// Bound but never accepted: the kernel completes the handshake and
// buffers the request, and no response ever comes.
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
let forge = super::Late::<std::sync::Arc<crate::forge::Client>>::default();
assert!(
forge
.set(std::sync::Arc::new(crate::forge::Client::for_test_forge(
url
)))
.is_ok()
);
let agent = || "atlas".to_owned();
let nodes = [
("create_repo", SwarmNodeKind::CreateRepo { agent: agent() }),
(
"create_forge_user",
SwarmNodeKind::CreateForgeUser { agent: agent() },
),
(
"add_repo_member",
SwarmNodeKind::AddRepoMember { agent: agent() },
),
(
"grant_docs_read",
SwarmNodeKind::GrantDocsRead { agent: agent() },
),
(
"init_agent_config_repo",
SwarmNodeKind::InitAgentConfigRepo { agent: agent() },
),
(
"mint_hive_forge_token",
SwarmNodeKind::MintHiveForgeToken {
hive: "pr1ma".to_owned(),
},
),
];
for (label, kind) in nodes {
let mut sched = hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
);
let id = sched.append(kind, Vec::new(), None).expect("insert");
let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
let deps = WorkerDeps {
auth: None,
forge: forge.clone(),
queue: None,
wanted: None,
matrix_homeserver: None,
};
let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
run_swarm_node(id, kind, builder, deps)
})
.expect("the node just inserted is runnable");
tokio::time::timeout(std::time::Duration::from_mins(10), runner)
.await
.unwrap_or_else(|_| panic!("{label}: node still running after 10 minutes"))
.1
.expect("no growth declared, nothing to reject");
let guard = sched.lock().unwrap();
let node = guard.graph().node(id).expect("node still present");
assert_eq!(node.state, hive_jobq::State::Failed, "{label}");
let error = node.error.as_deref().unwrap_or_default();
assert!(
error.contains(label) && error.contains("no answer from the forge"),
"{label}: expected a deadline error naming the node, got {error:?}"
);
}
drop(listener);
}
#[tokio::test]
async fn create_repo_node_runs_end_to_end_and_fails_without_forge_configured() {
unsafe {