Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: read placements only once the queue client has connected

This commit is contained in:
atlas 2026-11-07 18:40:39 +01:00 • committed by mara
commit f86d90a91d

View file

@ -790,10 +790,7 @@ async fn update_meta(
.to_owned(),
);
};
await_queue_connection(|| wanted.is_connected()).await?;
let placements = placements(wanted, &meta.hives)
.await
.map_err(|e| format!("{e:#}"))?;
let placements = read_placements(wanted, &meta.hives).await?;
let agents = placed_agents(queue, &placements)
.await
.map_err(|e| format!("{e:#}"))?;
@ -837,6 +834,18 @@ async fn await_queue_connection(connected: impl Fn() -> bool) -> Result<(), Stri
))
}
/// [`placements`], once the queue client has connected: the read hangs on a
/// client that has not. `Err` is the node's failure.
async fn read_placements(
wanted: &wanted::WantedWriter,
hives: &[String],
) -> Result<std::collections::BTreeSet<(String, String)>, String> {
await_queue_connection(|| wanted.is_connected()).await?;
placements(wanted, hives)
.await
.map_err(|e| format!("{e:#}"))
}
/// The `UpdateMeta` node's outcome from what [`update_meta`] decided, with the
/// deploys declared into `builder` (the scheduler inserts them under the node
/// when it completes). Synchronous: the builder is not `Sync`, so it cannot be
@ -4562,6 +4571,29 @@ mod tests {
"unclear message: {err}"
);
}
/// The placements read waits for the client first: over a client that
/// never connected it fails with the bounded-wait message, not with the
/// `client state: Pending` the read itself would give.
#[tokio::test(start_paused = true)]
async fn reading_placements_waits_on_a_client_that_never_connected() {
let client = async_nats::ConnectOptions::new()
.retry_on_initial_connect()
.connect("127.0.0.1:1")
.await
.expect("retry_on_initial_connect returns without waiting for a real connection");
let wanted = crate::wanted::WantedWriter::new(client);
let hives = ["h1".to_owned()];
let err = super::read_placements(&wanted, &hives)
.await
.expect_err("never connected");
assert!(
err.contains("still not connected after 60s"),
"unexpected error: {err}"
);
assert!(!err.contains("Pending"), "reached the read: {err}");
}
}
#[cfg(test)]