hive-c0re: graceful agent stop — quiesce harness, flush state, then stop
This commit is contained in:
parent
bf65345b83
commit
03ea5d601b
8 changed files with 202 additions and 2 deletions
|
|
@ -40,8 +40,9 @@ somewhere."
|
||||||
| `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. |
|
| `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. |
|
||||||
| `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, manager `restart` MCP tool. |
|
| `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, manager `restart` MCP tool. |
|
||||||
| `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. After a successful file write, emits `CapabilitiesChanged` or `ToolGroupsChanged` SSE snapshot so the P3RM1SS10NS tab updates live. |
|
| `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. After a successful file write, emits `CapabilitiesChanged` or `ToolGroupsChanged` SSE snapshot so the P3RM1SS10NS tab updates live. |
|
||||||
|
| `GracefulStop` | Quiesce then stop a container (the `?graceful=1` path on `/kill/<agent>`). Signals the harness (its next `Recv` returns `GracefulStop` — the inbound fence — so it runs one stop-checkpoint turn to flush durable `/state`, then exits), waits for it to drain (bounded by a 3-min timeout → hard-stop fallback), then runs the normal container-stop teardown. Queued so it can't race an in-flight rebuild for the same agent. |
|
||||||
|
|
||||||
**Intentionally not queued** (sub-second ops): `start`, `stop`, `kill`.
|
**Intentionally not queued** (sub-second ops): the *hard* `start`, `stop`, `kill`. (A *graceful* stop is the `GracefulStop` kind above — it takes a checkpoint turn, so it rides the queue.)
|
||||||
|
|
||||||
### Dedup
|
### Dedup
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -173,6 +173,22 @@ fn synthetic_continue() -> hive_sh4re::DeliveredMessage {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Synthetic message that drives the single stop-checkpoint turn when c0re
|
||||||
|
/// signals a graceful stop. The agent gets one final turn to flush durable
|
||||||
|
/// `/state` before the container is stopped; new inbound is already fenced.
|
||||||
|
fn graceful_stop_message() -> hive_sh4re::DeliveredMessage {
|
||||||
|
hive_sh4re::DeliveredMessage {
|
||||||
|
from: "graceful-stop".into(),
|
||||||
|
body: "You are being gracefully stopped — the container will shut down after this turn, \
|
||||||
|
and new inbound messages are already fenced. Flush anything worth keeping to your \
|
||||||
|
durable /state files now, then end your turn. Do not start new long-running work."
|
||||||
|
.into(),
|
||||||
|
id: 0,
|
||||||
|
redelivered: false,
|
||||||
|
in_reply_to: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ---------- surface trait ----------
|
// ---------- surface trait ----------
|
||||||
|
|
||||||
/// What a `Recv` long-poll returned. Decoupled from the per-role
|
/// What a `Recv` long-poll returned. Decoupled from the per-role
|
||||||
|
|
@ -188,6 +204,10 @@ enum RecvOutcome {
|
||||||
/// retries; the surface impl is responsible for tracing the
|
/// retries; the surface impl is responsible for tracing the
|
||||||
/// detail before returning this.
|
/// detail before returning this.
|
||||||
TransportError,
|
TransportError,
|
||||||
|
/// c0re signalled a graceful stop for this agent. The serve loop runs
|
||||||
|
/// one stop-checkpoint turn (flush durable `/state`), reports
|
||||||
|
/// `GracefulStopComplete`, and exits so the container can be stopped.
|
||||||
|
GracefulStop,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Wire surface abstraction. `AgentSurface` is the only impl — the trait
|
/// Wire surface abstraction. `AgentSurface` is the only impl — the trait
|
||||||
|
|
@ -211,6 +231,12 @@ trait Surface {
|
||||||
/// Either field is `None` when the underlying request errors.
|
/// Either field is `None` when the underlying request errors.
|
||||||
fn post_turn_counts(socket: &Path) -> impl Future<Output = (Option<u64>, Option<u64>)>;
|
fn post_turn_counts(socket: &Path) -> impl Future<Output = (Option<u64>, Option<u64>)>;
|
||||||
|
|
||||||
|
/// Tell c0re the graceful-stop checkpoint is done and the harness is
|
||||||
|
/// exiting its serve loop (fire-and-forget; logs on error). Lets the
|
||||||
|
/// `GracefulStop` orchestration stop the container without waiting out
|
||||||
|
/// its timeout fallback.
|
||||||
|
fn graceful_stop_complete(socket: &Path) -> impl Future<Output = ()>;
|
||||||
|
|
||||||
/// Send a message addressed to `<parent>` (broker resolves the
|
/// Send a message addressed to `<parent>` (broker resolves the
|
||||||
/// sentinel via `topology::parent_of` at delivery time; root
|
/// sentinel via `topology::parent_of` at delivery time; root
|
||||||
/// agents/manager fall through to operator).
|
/// agents/manager fall through to operator).
|
||||||
|
|
@ -259,6 +285,18 @@ impl Surface for AgentSurface {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn graceful_stop_complete(socket: &Path) {
|
||||||
|
match client::request::<_, AgentResponse>(socket, &AgentRequest::GracefulStopComplete).await
|
||||||
|
{
|
||||||
|
Ok(AgentResponse::Ok) => {}
|
||||||
|
Ok(AgentResponse::Err { message }) => {
|
||||||
|
tracing::warn!(%message, "graceful_stop_complete rejected by broker");
|
||||||
|
}
|
||||||
|
Ok(other) => tracing::warn!(?other, "graceful_stop_complete unexpected response"),
|
||||||
|
Err(e) => tracing::warn!(error = ?e, "graceful_stop_complete transport error"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async fn inbox_unread(socket: &Path) -> u64 {
|
async fn inbox_unread(socket: &Path) -> u64 {
|
||||||
match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await {
|
match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await {
|
||||||
Ok(AgentResponse::Status { unread }) => unread,
|
Ok(AgentResponse::Status { unread }) => unread,
|
||||||
|
|
@ -318,6 +356,7 @@ impl Surface for AgentSurface {
|
||||||
RecvOutcome::Message(first)
|
RecvOutcome::Message(first)
|
||||||
}
|
}
|
||||||
Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty,
|
Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty,
|
||||||
|
Ok(AgentResponse::GracefulStop) => RecvOutcome::GracefulStop,
|
||||||
Ok(AgentResponse::Err { message }) => {
|
Ok(AgentResponse::Err { message }) => {
|
||||||
tracing::warn!(%message, "recv error");
|
tracing::warn!(%message, "recv error");
|
||||||
RecvOutcome::TransportError
|
RecvOutcome::TransportError
|
||||||
|
|
@ -473,6 +512,26 @@ async fn serve_loop<S: Surface>(
|
||||||
// No backoff: the long-poll wait is itself the throttle.
|
// No backoff: the long-poll wait is itself the throttle.
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
RecvOutcome::GracefulStop => {
|
||||||
|
// c0re fenced our inbox and wants a clean stop. Run one
|
||||||
|
// checkpoint turn so the agent flushes durable /state,
|
||||||
|
// report completion, then exit the loop → the harness
|
||||||
|
// process ends and the container can be stopped.
|
||||||
|
tracing::info!(
|
||||||
|
"graceful stop signalled — running stop-checkpoint turn, then exiting"
|
||||||
|
);
|
||||||
|
let _ = handle_turn::<S>(
|
||||||
|
socket,
|
||||||
|
&bus,
|
||||||
|
stats.as_ref(),
|
||||||
|
files,
|
||||||
|
&turn_lock,
|
||||||
|
graceful_stop_message(),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
S::graceful_stop_complete(socket).await;
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
|
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
|
||||||
|
|
|
||||||
|
|
@ -79,6 +79,10 @@ impl From<hive_sh4re::Response> for SocketReply {
|
||||||
hive_sh4re::Response::HostJournal { content } => Self::HostJournal(content),
|
hive_sh4re::Response::HostJournal { content } => Self::HostJournal(content),
|
||||||
hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules),
|
hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules),
|
||||||
hive_sh4re::Response::Containers { containers } => Self::Containers(containers),
|
hive_sh4re::Response::Containers { containers } => Self::Containers(containers),
|
||||||
|
// A graceful stop is pending — the inbox is fenced. If claude polls
|
||||||
|
// `recv` during its stop-checkpoint turn, present it as an empty
|
||||||
|
// inbox (the serve loop is the real handler of this signal).
|
||||||
|
hive_sh4re::Response::GracefulStop => Self::Messages(Vec::new()),
|
||||||
hive_sh4re::Response::AgentMeta {
|
hive_sh4re::Response::AgentMeta {
|
||||||
name,
|
name,
|
||||||
running,
|
running,
|
||||||
|
|
|
||||||
|
|
@ -185,6 +185,13 @@ pub(crate) async fn dispatch_shared(
|
||||||
}
|
}
|
||||||
hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent),
|
hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent),
|
||||||
hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
|
hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
|
||||||
|
hive_sh4re::Request::GracefulStopComplete => {
|
||||||
|
// Harness drained + is exiting: clear the fence so the
|
||||||
|
// `GracefulStop` orchestration (which polls this flag) proceeds
|
||||||
|
// to stop the container without waiting out its timeout.
|
||||||
|
coord.clear_graceful_stop(agent);
|
||||||
|
hive_sh4re::Response::Ok
|
||||||
|
}
|
||||||
hive_sh4re::Request::GetHostJournal {
|
hive_sh4re::Request::GetHostJournal {
|
||||||
unit,
|
unit,
|
||||||
container,
|
container,
|
||||||
|
|
@ -221,6 +228,15 @@ async fn handle_recv(
|
||||||
wait_seconds: Option<u64>,
|
wait_seconds: Option<u64>,
|
||||||
max: Option<u32>,
|
max: Option<u32>,
|
||||||
) -> hive_sh4re::Response {
|
) -> hive_sh4re::Response {
|
||||||
|
// Graceful-stop fence: while a graceful stop is pending for this agent,
|
||||||
|
// return `GracefulStop` instead of polling the broker. The harness runs
|
||||||
|
// one stop-checkpoint turn then exits; new sends keep queueing in the
|
||||||
|
// broker for the agent's next start. Checked before the (blocking) poll
|
||||||
|
// so a flag set between polls is seen on the next Recv — the orchestration
|
||||||
|
// also fires a transient wake to break an in-flight long-poll.
|
||||||
|
if coord.is_graceful_stop_pending(agent) {
|
||||||
|
return hive_sh4re::Response::GracefulStop;
|
||||||
|
}
|
||||||
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
|
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
|
||||||
match coord
|
match coord
|
||||||
.broker
|
.broker
|
||||||
|
|
|
||||||
|
|
@ -125,6 +125,12 @@ pub struct Coordinator {
|
||||||
/// the dashboard's `agents_crashing` banner warning via
|
/// the dashboard's `agents_crashing` banner warning via
|
||||||
/// `recent_crash_counts`, which prunes entries older than its window.
|
/// `recent_crash_counts`, which prunes entries older than its window.
|
||||||
recent_crashes: Mutex<HashMap<String, Vec<std::time::Instant>>>,
|
recent_crashes: Mutex<HashMap<String, Vec<std::time::Instant>>>,
|
||||||
|
/// Agents with a graceful stop in progress. Set by the `GracefulStop`
|
||||||
|
/// orchestration; read by `agent_server::handle_recv`, which returns
|
||||||
|
/// `Response::GracefulStop` (instead of polling the broker) while an
|
||||||
|
/// agent is in this set — the inbound fence. Cleared when the agent
|
||||||
|
/// reports `GracefulStopComplete` or the container is stopped.
|
||||||
|
graceful_stop_pending: Mutex<HashSet<String>>,
|
||||||
/// Unified wire-facing event channel feeding the dashboard SSE
|
/// Unified wire-facing event channel feeding the dashboard SSE
|
||||||
/// stream. Carries broker messages (mirrored from `broker.subscribe`
|
/// stream. Carries broker messages (mirrored from `broker.subscribe`
|
||||||
/// by the forwarder task in `main.rs`) and dashboard-only mutation
|
/// by the forwarder task in `main.rs`) and dashboard-only mutation
|
||||||
|
|
@ -450,6 +456,7 @@ impl Coordinator {
|
||||||
transient: Mutex::new(HashMap::new()),
|
transient: Mutex::new(HashMap::new()),
|
||||||
recent_transient: Mutex::new(HashMap::new()),
|
recent_transient: Mutex::new(HashMap::new()),
|
||||||
recent_crashes: Mutex::new(HashMap::new()),
|
recent_crashes: Mutex::new(HashMap::new()),
|
||||||
|
graceful_stop_pending: Mutex::new(HashSet::new()),
|
||||||
dashboard_events,
|
dashboard_events,
|
||||||
event_seq: AtomicU64::new(0),
|
event_seq: AtomicU64::new(0),
|
||||||
meta_updates_active: AtomicU64::new(0),
|
meta_updates_active: AtomicU64::new(0),
|
||||||
|
|
@ -1128,6 +1135,28 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Mark `name` as having a graceful stop in progress. While set,
|
||||||
|
/// `agent_server::handle_recv` returns `Response::GracefulStop` for
|
||||||
|
/// this agent instead of polling the broker (the inbound fence).
|
||||||
|
pub fn mark_graceful_stop(&self, name: &str) {
|
||||||
|
self.graceful_stop_pending
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.insert(name.to_owned());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Whether a graceful stop is pending for `name`.
|
||||||
|
#[must_use]
|
||||||
|
pub fn is_graceful_stop_pending(&self, name: &str) -> bool {
|
||||||
|
self.graceful_stop_pending.lock().unwrap().contains(name)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Clear the graceful-stop flag for `name` (agent reported
|
||||||
|
/// `GracefulStopComplete`, or the stop finished / was abandoned).
|
||||||
|
pub fn clear_graceful_stop(&self, name: &str) {
|
||||||
|
self.graceful_stop_pending.lock().unwrap().remove(name);
|
||||||
|
}
|
||||||
|
|
||||||
/// Set of agents whose transient was cleared within the last
|
/// Set of agents whose transient was cleared within the last
|
||||||
/// `grace` seconds — i.e. agents the operator just acted on,
|
/// `grace` seconds — i.e. agents the operator just acted on,
|
||||||
/// whose stop the crash watcher should NOT classify as a crash.
|
/// whose stop the crash watcher should NOT classify as a crash.
|
||||||
|
|
|
||||||
|
|
@ -6,12 +6,21 @@
|
||||||
//! spinner); destroy delegates to `actions::destroy` (optionally purging).
|
//! spinner); destroy delegates to `actions::destroy` (optionally purging).
|
||||||
|
|
||||||
use axum::{
|
use axum::{
|
||||||
extract::{Form, Path as AxumPath, State},
|
extract::{Form, Path as AxumPath, Query, State},
|
||||||
http::StatusCode,
|
http::StatusCode,
|
||||||
response::{IntoResponse, Response},
|
response::{IntoResponse, Response},
|
||||||
};
|
};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
|
|
||||||
|
/// Query params for `post_kill`. `?graceful=1` routes to the graceful-stop
|
||||||
|
/// orchestration (quiesce the harness, flush `/state`, then container stop)
|
||||||
|
/// instead of an immediate hard stop. Defaults false → today's hard kill.
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
pub(super) struct KillParams {
|
||||||
|
#[serde(default)]
|
||||||
|
graceful: bool,
|
||||||
|
}
|
||||||
|
|
||||||
use super::{AppState, error_response, guard_agent_name, strip_container_prefix};
|
use super::{AppState, error_response, guard_agent_name, strip_container_prefix};
|
||||||
use crate::{actions, lifecycle};
|
use crate::{actions, lifecycle};
|
||||||
|
|
||||||
|
|
@ -75,11 +84,28 @@ where
|
||||||
pub(super) async fn post_kill(
|
pub(super) async fn post_kill(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
AxumPath(name): AxumPath<String>,
|
AxumPath(name): AxumPath<String>,
|
||||||
|
Query(params): Query<KillParams>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
let logical = strip_container_prefix(&name);
|
let logical = strip_container_prefix(&name);
|
||||||
if let Some(reject) = guard_agent_name(&state, &logical).await {
|
if let Some(reject) = guard_agent_name(&state, &logical).await {
|
||||||
return reject;
|
return reject;
|
||||||
}
|
}
|
||||||
|
if params.graceful {
|
||||||
|
// Graceful stop: enqueue the quiesce orchestration (signal the harness
|
||||||
|
// → one stop-checkpoint turn → drain → container stop, with a timeout
|
||||||
|
// fallback to a hard stop). Serialised through the rebuild queue so it
|
||||||
|
// can't race an in-flight rebuild for the same agent, and its per-step
|
||||||
|
// progress surfaces on the queue snapshot + build log.
|
||||||
|
state.coord.rebuild_queue.enqueue(
|
||||||
|
crate::rebuild_queue::QueueKind::GracefulStop,
|
||||||
|
logical,
|
||||||
|
crate::rebuild_queue::QueueSource::Manual,
|
||||||
|
"manual via dashboard graceful stop".to_owned(),
|
||||||
|
None,
|
||||||
|
);
|
||||||
|
state.coord.emit_rebuild_queue_snapshot();
|
||||||
|
return (StatusCode::OK, "ok").into_response();
|
||||||
|
}
|
||||||
// Manager is stoppable from the dashboard like any other
|
// Manager is stoppable from the dashboard like any other
|
||||||
// agent. The host's dashboard server keeps running (it's
|
// agent. The host's dashboard server keeps running (it's
|
||||||
// hive-c0re, not the manager container), per-agent approvals
|
// hive-c0re, not the manager container), per-agent approvals
|
||||||
|
|
|
||||||
|
|
@ -42,6 +42,10 @@ pub enum QueueKind {
|
||||||
/// Serialised through the queue so concurrent dashboard batch-apply
|
/// Serialised through the queue so concurrent dashboard batch-apply
|
||||||
/// actions for different agents never race on the shared JSON file.
|
/// actions for different agents never race on the shared JSON file.
|
||||||
PermChange,
|
PermChange,
|
||||||
|
/// Gracefully stop a container: signal the harness to run one
|
||||||
|
/// stop-checkpoint turn (flush durable `/state`), wait for it to drain,
|
||||||
|
/// then `nixos-container stop`. Falls back to a hard stop on timeout.
|
||||||
|
GracefulStop,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl QueueKind {
|
impl QueueKind {
|
||||||
|
|
@ -54,6 +58,7 @@ impl QueueKind {
|
||||||
QueueKind::StartupSweep => "startup_sweep",
|
QueueKind::StartupSweep => "startup_sweep",
|
||||||
QueueKind::Restart => "restart",
|
QueueKind::Restart => "restart",
|
||||||
QueueKind::PermChange => "perm_change",
|
QueueKind::PermChange => "perm_change",
|
||||||
|
QueueKind::GracefulStop => "graceful_stop",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -662,6 +667,12 @@ impl RebuildQueue {
|
||||||
/// signal the worker exits after its current entry finishes; pending
|
/// signal the worker exits after its current entry finishes; pending
|
||||||
/// `Queued` entries are dropped (they'll either be replayed by the
|
/// `Queued` entries are dropped (they'll either be replayed by the
|
||||||
/// startup sweep on next boot or left for an operator to re-queue).
|
/// startup sweep on next boot or left for an operator to re-queue).
|
||||||
|
/// Max time the `GracefulStop` worker waits for the harness to run its
|
||||||
|
/// stop-checkpoint turn + drain before falling back to a hard container stop.
|
||||||
|
/// Generous — a checkpoint turn can take a while — but bounded so a wedged
|
||||||
|
/// agent never blocks the stop indefinitely.
|
||||||
|
const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
|
||||||
|
|
||||||
pub async fn run_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
|
pub async fn run_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
|
||||||
let mut shutdown = coord.shutdown_rx();
|
let mut shutdown = coord.shutdown_rx();
|
||||||
loop {
|
loop {
|
||||||
|
|
@ -833,9 +844,49 @@ async fn dispatch(
|
||||||
crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default();
|
crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default();
|
||||||
crate::auto_update::rebuild_agent(coord, name, ¤t_rev, Some(entry.id)).await
|
crate::auto_update::rebuild_agent(coord, name, ¤t_rev, Some(entry.id)).await
|
||||||
}
|
}
|
||||||
|
(QueueKind::GracefulStop, _) => run_graceful_stop(coord, entry).await,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Run one `GracefulStop` entry: signal the harness to quiesce (it returns
|
||||||
|
/// `GracefulStop` on its next `Recv`, runs one stop-checkpoint turn to flush
|
||||||
|
/// durable `/state`, then exits), wait for it to drain — bounded by
|
||||||
|
/// `GRACEFUL_STOP_TIMEOUT` so a wedged agent can't block forever — then stop
|
||||||
|
/// the container with the same teardown as a plain kill.
|
||||||
|
async fn run_graceful_stop(
|
||||||
|
coord: &std::sync::Arc<crate::coordinator::Coordinator>,
|
||||||
|
entry: &QueueEntry,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
let name = &entry.agent;
|
||||||
|
let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Stopping);
|
||||||
|
// Signal the harness; the kick breaks an idle long-poll so it's seen promptly.
|
||||||
|
coord.set_queue_step(Some(entry.id), "graceful stop: signalling agent");
|
||||||
|
coord.mark_graceful_stop(name);
|
||||||
|
coord.kick_agent(name, "graceful stop requested");
|
||||||
|
// Wait for the harness to drain (it clears the flag via `GracefulStopComplete`)
|
||||||
|
// or fall back to a hard stop after the timeout. The single queue worker is
|
||||||
|
// intentionally held for the duration — graceful stops are infrequent.
|
||||||
|
coord.set_queue_step(Some(entry.id), "graceful stop: waiting for agent to drain");
|
||||||
|
let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
|
||||||
|
while coord.is_graceful_stop_pending(name) {
|
||||||
|
if std::time::Instant::now() >= deadline {
|
||||||
|
tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||||
|
}
|
||||||
|
coord.clear_graceful_stop(name);
|
||||||
|
// Stop the container — same teardown as a plain kill.
|
||||||
|
coord.set_queue_step(Some(entry.id), "nixos-container stop");
|
||||||
|
crate::lifecycle::kill(name).await?;
|
||||||
|
coord.unregister_agent(name);
|
||||||
|
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
|
||||||
|
agent: name.clone(),
|
||||||
|
});
|
||||||
|
coord.rescan_containers_and_emit().await;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Run one `MetaUpdate` entry: bump the meta flake's locks for the
|
/// Run one `MetaUpdate` entry: bump the meta flake's locks for the
|
||||||
/// requested inputs, then enqueue a cascade of `Rebuild` entries
|
/// requested inputs, then enqueue a cascade of `Rebuild` entries
|
||||||
/// (with `parent_id` set to this entry's id) for every agent affected
|
/// (with `parent_id` set to this entry's id) for every agent affected
|
||||||
|
|
|
||||||
|
|
@ -529,6 +529,11 @@ pub enum Request {
|
||||||
/// crashed-mid-turn sessions. See
|
/// crashed-mid-turn sessions. See
|
||||||
/// `docs/conventions.md::Broker delivery + ack cycle`.
|
/// `docs/conventions.md::Broker delivery + ack cycle`.
|
||||||
RequeueInflight,
|
RequeueInflight,
|
||||||
|
/// Harness → c0re: "I saw the `GracefulStop` signal, ran my
|
||||||
|
/// stop-checkpoint turn (durable `/state` flushed) and am exiting my
|
||||||
|
/// serve loop now." Lets the `GracefulStop` orchestration stop the
|
||||||
|
/// container immediately instead of waiting out its timeout fallback.
|
||||||
|
GracefulStopComplete,
|
||||||
/// *(capability-gated: `read_host_journal`)* Fetch recent lines
|
/// *(capability-gated: `read_host_journal`)* Fetch recent lines
|
||||||
/// from the host journal. Filters are all optional; omitting all
|
/// from the host journal. Filters are all optional; omitting all
|
||||||
/// returns the last `lines` entries from the global journal.
|
/// returns the last `lines` entries from the global journal.
|
||||||
|
|
@ -700,6 +705,15 @@ pub enum Response {
|
||||||
/// status. Ordered by topology depth (parents before children), then
|
/// status. Ordered by topology depth (parents before children), then
|
||||||
/// alphabetically within each depth tier.
|
/// alphabetically within each depth tier.
|
||||||
Containers { containers: Vec<ContainerInfo> },
|
Containers { containers: Vec<ContainerInfo> },
|
||||||
|
/// `Recv` result when a graceful stop is pending for this agent
|
||||||
|
/// (set by hive-c0re's `GracefulStop` orchestration). Returned in
|
||||||
|
/// place of `Messages` — it doubles as the inbound fence: the harness
|
||||||
|
/// stops consuming normal inbox messages and instead runs one
|
||||||
|
/// stop-checkpoint turn (flush durable `/state`), then reports
|
||||||
|
/// `GracefulStopComplete` and exits its serve loop so the container
|
||||||
|
/// can be stopped cleanly. New sends keep queueing in the broker for
|
||||||
|
/// the agent's next start.
|
||||||
|
GracefulStop,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Backwards-compatible response aliases.
|
/// Backwards-compatible response aliases.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue