swarm: bash task status keys expire after 72 hours
Nothing deletes the key of a finished task, so the bucket would grow with every task an agent runs. The shared bucket helper takes an optional max age, applied when the bucket is created; a live task's republishing keeps its own key.
This commit is contained in:
parent
1779fa0568
commit
1a698f1a97
4 changed files with 35 additions and 3 deletions
|
|
@ -195,7 +195,9 @@ status kept apart from the output. A task's status is one key, `<agent>.<task>`,
|
||||||
`bash-task-status` bucket: its `state` (`pending`, `running`, `done`, `timed_out`,
|
`bash-task-status` bucket: its `state` (`pending`, `running`, `done`, `timed_out`,
|
||||||
`interrupted` or `killed`), the command cut to 200 characters, when it began and when it
|
`interrupted` or `killed`), the command cut to 200 characters, when it began and when it
|
||||||
entered its state, the exit status, and `ts`, the time of the publish. A live task republishes
|
entered its state, the exit status, and `ts`, the time of the publish. A live task republishes
|
||||||
every 60 seconds, and a reader treats a live value older than three of those as stale. The output
|
every 60 seconds, and a reader treats a live value older than three of those as stale. A key
|
||||||
|
expires 72 hours after its last write, since nothing deletes a finished task's key, and a
|
||||||
|
live task's republishing keeps its own. The output
|
||||||
goes to `$SWARM.term.<agent>.bash.<task>`, with a `Swarm-Stream` header of `stdout` or
|
goes to `$SWARM.term.<agent>.bash.<task>`, with a `Swarm-Stream` header of `stdout` or
|
||||||
`stderr`, into the stream `term-bash-<agent>` that swarm-controller creates for every declared
|
`stderr`, into the stream `term-bash-<agent>` that swarm-controller creates for every declared
|
||||||
agent, for 24 hours or 64 MiB like the subagent rows. The agent's credential may publish
|
agent, for 24 hours or 64 MiB like the subagent rows. The agent's credential may publish
|
||||||
|
|
|
||||||
|
|
@ -32,16 +32,19 @@ pub(crate) fn output_config(
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Open the status bucket, creating it if nothing has yet. `history: 1`: a
|
/// Open the status bucket, creating it if nothing has yet. `history: 1`: a
|
||||||
/// reader wants each task's current state, not every state it has been in.
|
/// reader wants each task's current state, not every state it has been in. A
|
||||||
|
/// key expires [`swarm_wire::bash_task::STATUS_MAX_AGE`] after its last write,
|
||||||
|
/// since nothing deletes the key of a finished task.
|
||||||
pub async fn open_status(
|
pub async fn open_status(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
|
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
|
||||||
let js = async_nats::jetstream::new(client.clone());
|
let js = async_nats::jetstream::new(client.clone());
|
||||||
crate::kv::open_or_create(
|
crate::kv::open_or_create_expiring(
|
||||||
&js,
|
&js,
|
||||||
swarm_wire::bash_task::BUCKET,
|
swarm_wire::bash_task::BUCKET,
|
||||||
"bash-task-status",
|
"bash-task-status",
|
||||||
"The state of each bash task, as its agent's bash MCP last published it".to_owned(),
|
"The state of each bash task, as its agent's bash MCP last published it".to_owned(),
|
||||||
|
Some(swarm_wire::bash_task::STATUS_MAX_AGE),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,19 @@ pub(crate) async fn open_or_create(
|
||||||
bucket: &str,
|
bucket: &str,
|
||||||
label: &str,
|
label: &str,
|
||||||
description: String,
|
description: String,
|
||||||
|
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
||||||
|
open_or_create_expiring(js, bucket, label, description, None).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// [`open_or_create`] with a `max_age`: a key nobody writes for that long goes
|
||||||
|
/// away, for a bucket whose keys come from a publisher that never deletes. The
|
||||||
|
/// age applies to a bucket created here; one that already exists keeps its own.
|
||||||
|
pub(crate) async fn open_or_create_expiring(
|
||||||
|
js: &async_nats::jetstream::Context,
|
||||||
|
bucket: &str,
|
||||||
|
label: &str,
|
||||||
|
description: String,
|
||||||
|
max_age: Option<std::time::Duration>,
|
||||||
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
||||||
match js.get_key_value(bucket).await {
|
match js.get_key_value(bucket).await {
|
||||||
Ok(store) => Ok(store),
|
Ok(store) => Ok(store),
|
||||||
|
|
@ -22,6 +35,7 @@ pub(crate) async fn open_or_create(
|
||||||
bucket: bucket.to_owned(),
|
bucket: bucket.to_owned(),
|
||||||
description,
|
description,
|
||||||
history: 1,
|
history: 1,
|
||||||
|
max_age: max_age.unwrap_or_default(),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,11 @@ pub const HEARTBEAT_S: i64 = 60;
|
||||||
/// A `running` status whose [`Status::ts`] is older than this is stale.
|
/// A `running` status whose [`Status::ts`] is older than this is stale.
|
||||||
pub const STALE_AFTER_S: i64 = 3 * HEARTBEAT_S;
|
pub const STALE_AFTER_S: i64 = 3 * HEARTBEAT_S;
|
||||||
|
|
||||||
|
/// How long a task's key stays in the bucket after its last write. A live task
|
||||||
|
/// is republished every [`HEARTBEAT_S`], which keeps its key; a finished one
|
||||||
|
/// is only worth listing for a few days, and nothing else deletes the key.
|
||||||
|
pub const STATUS_MAX_AGE: std::time::Duration = std::time::Duration::from_hours(72);
|
||||||
|
|
||||||
/// The longest [`Status::cmd`], in characters; a longer command is cut.
|
/// The longest [`Status::cmd`], in characters; a longer command is cut.
|
||||||
pub const CMD_MAX_CHARS: usize = 200;
|
pub const CMD_MAX_CHARS: usize = 200;
|
||||||
|
|
||||||
|
|
@ -218,6 +223,14 @@ mod tests {
|
||||||
assert!(!status(State::Done, 10_000).is_running(10_000));
|
assert!(!status(State::Done, 10_000).is_running(10_000));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A live task's key must outlive the gap between two heartbeats and the
|
||||||
|
/// stale window by a wide margin, or a running task could expire.
|
||||||
|
#[test]
|
||||||
|
fn a_key_outlives_the_stale_window_by_far() {
|
||||||
|
let stale = std::time::Duration::from_secs(STALE_AFTER_S.unsigned_abs());
|
||||||
|
assert!(STATUS_MAX_AGE >= stale * 100);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn a_long_command_is_cut_with_an_ellipsis() {
|
fn a_long_command_is_cut_with_an_ellipsis() {
|
||||||
assert_eq!(clip_cmd("ls"), "ls");
|
assert_eq!(clip_cmd("ls"), "ls");
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue