Delete a batch of imports in one command, not one connection each

Replaces the connection throttle from the previous commit, which was a
workaround: it made a batch slower without changing that N sessions
meant N ssh connections, and the cap it picked was a guess at somebody
else's sshd config.

The batch now goes out as a single invocation. import::delete takes the
whole list, the remote script loops over the ids and prints one
`<id>\t<state>` line each, and the route settles every row from its own
line. So a batch of any size is one connection and cannot exceed
MaxStartups however many rows are selected -- and it is faster, since
it stopped paying a handshake per session.

Each id still reports on its own: deleted, missing, or failed, kept
apart because only "failed" is worth retrying. Ids the machine never
mentioned -- a connection that dropped part-way -- are reported as
unknown rather than defaulting to either answer, and a malformed id
fails only itself.

Verified end-to-end against a fake ssh that counts connections: a
9-session batch used one, every row settled, both copies of a session
recorded under two project directories went, and the id that was not
there failed with a message saying so rather than a connection error.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
irisandClaude Opus 5 committed 2026-09-01 00:32:52 -04:00
1 parent 7a1a462caf
commit dce1799802
4 files changed
+225 -74

No files matched your search

+61 -15
View File
@@ -549,28 +549,68 @@ struct ImportableRow {
/// id is registered as in flight before the 202 goes back, so the answer to /// id is registered as in flight before the 202 goes back, so the answer to
/// "did my batch start" is one answer for the batch. /// "did my batch start" is one answer for the batch.
/// ///
/// Registering is what has to be atomic; the work itself does not. Each id /// Registering is what has to be atomic; the work itself does not. The
/// runs on its own task and settles on its own event, because six deletes /// batch runs as one command on the machine -- see
/// that must all succeed or all roll back is not something a filesystem /// [`crate::session::import::delete`] for why it is not one per id -- but
/// offers, and pretending otherwise would mean holding five sessions /// each row still settles on its own event from its own outcome, because
/// hostage to the one that failed. /// six deletes that must all succeed or all roll back is not something a
/// filesystem offers, and pretending otherwise would mean holding five
/// sessions hostage to the one that failed.
async fn delete_importable( async fn delete_importable(
State(manager): State<Arc<SessionManager>>, State(manager): State<Arc<SessionManager>>,
UrlPath(id): UrlPath<String>, UrlPath(id): UrlPath<String>,
axum::Json(body): axum::Json<DeleteBatch>, axum::Json(body): axum::Json<DeleteBatch>,
) -> Result<StatusCode, ApiError> { ) -> Result<StatusCode, ApiError> {
let setup = setup_by_id(&manager, &id)?; let setup = setup_by_id(&manager, &id)?;
for session in body.sessions {
let transport = crate::session::transport::Transport::for_setup(&setup); let transport = crate::session::transport::Transport::for_setup(&setup);
let target = session.clone(); // Registered before anything is spawned, so the 202 is only sent once
in_background( // every row is already showing "deleting" -- a phone that refetches the
&manager, // instant it gets the reply cannot catch a row that has not started.
id.clone(), let running: Vec<(String, crate::session::pending::InFlight)> = body
session, .sessions
Operation::Deleting, .iter()
async move { crate::session::import::delete(&transport, &target).await }, .map(|session| {
(
session.clone(),
manager.pending().begin(&id, session, Operation::Deleting),
)
})
.collect();
let sessions = body.sessions;
tokio::spawn(async move {
// One failure here is the machine being unreachable, which is true
// of every row rather than of any one of them, so they all say so.
let outcomes = match crate::session::import::delete(&transport, &sessions).await {
Ok(outcomes) => outcomes,
Err(err) => {
let message = format!("{err:#}");
tracing::warn!(
"deleting {} sessions on {id} failed: {message}",
running.len()
); );
for (_, flight) in running {
flight.failed(message.clone());
} }
return;
}
};
for (session, flight) in running {
match outcomes.get(&session) {
Some(Ok(())) => {
tracing::info!("deleting {session} on {id}: done");
flight.succeeded();
}
Some(Err(message)) => {
tracing::warn!("deleting {session} on {id} failed: {message}");
flight.failed(message.clone());
}
// `delete` promises an entry per id, so this is a bug
// rather than a state -- but a row stuck on "deleting"
// for ever is a worse answer than one that says so.
None => flight.failed(format!("nothing was reported about {session}")),
}
}
});
Ok(StatusCode::ACCEPTED) Ok(StatusCode::ACCEPTED)
} }
@@ -872,9 +912,15 @@ async fn delete_session(
if let Some((setup, session)) = &foreign { if let Some((setup, session)) = &foreign {
let setup = setup_by_id(&manager, setup)?; let setup = setup_by_id(&manager, setup)?;
let transport = crate::session::transport::Transport::for_setup(&setup); let transport = crate::session::transport::Transport::for_setup(&setup);
crate::session::import::delete(&transport, session) // A batch of one: the same call, so there is one description of
// what deleting a foreign transcript means. Its outcome is this
// request's outcome, since there is only the one row.
crate::session::import::delete(&transport, std::slice::from_ref(session))
.await .await
.map_err(bad_request)?; .map_err(bad_request)?
.remove(session)
.unwrap_or_else(|| Err(format!("nothing was reported about {session}")))
.map_err(|message| bad_request(anyhow::anyhow!("{message}")))?;
tracing::info!("deleted Claude Code session {session} with ai-app session {id}"); tracing::info!("deleted Claude Code session {session} with ai-app session {id}");
} }
manager.delete_session(&id).map_err(bad_request)?; manager.delete_session(&id).map_err(bad_request)?;
+164 -33
View File
@@ -17,7 +17,9 @@
//! reason: an enrolled token must not be able to turn into "read me this //! reason: an enrolled token must not be able to turn into "read me this
//! arbitrary path". //! arbitrary path".
use anyhow::{Context, Result, bail, ensure}; use std::collections::HashMap;
use anyhow::{Context, Result};
use serde::Serialize; use serde::Serialize;
use serde_json::Value; use serde_json::Value;
@@ -684,9 +686,42 @@ fn push_assistant(events: &mut Vec<Event>, content: &Value) {
} }
} }
/// Deletes one of the sessions [`list`] reported. /// Removes each id it is given and prints one `<id>\t<state>` line per id.
/// ///
/// By id, resolved here against what the machine actually has, so the /// The three states are every way removing one id can end: `deleted` if at
/// least one file went, `missing` if the glob matched nothing, `failed` if
/// an `rm` refused. "Not there" is deliberately kept apart from "it broke"
/// rather than folded together by the shell -- only one of them is worth
/// retrying, and the caller is what knows how to word either.
///
/// Every copy of each id, not the first. The same id can name a file under
/// two project directories -- see the de-duplication in `parse_listing` --
/// and stopping at the first left the other behind, so the row came back on
/// the next listing after a delete that had reported success. `failed`
/// therefore sticks once set: one copy removed and another refused is not a
/// success.
///
/// Ids arrive as arguments rather than in the script text, so nothing here
/// is shell syntax; `is_session_id` is what keeps one from globbing its way
/// out of the projects directory.
const DELETE_SCRIPT: &str = r#"
for id do
state=missing
for f in "$HOME"/.claude/projects/*/"$id".jsonl; do
[ -f "$f" ] || continue
if rm -f "$f"; then
[ "$state" = failed ] || state=deleted
else
state=failed
fi
done
printf '%s\t%s\n' "$id" "$state"
done
"#;
/// Deletes sessions [`list`] reported, and says what happened to each.
///
/// By id, resolved on the machine against what it actually has, so the
/// caller never names a file -- the same rule importing follows, and it /// caller never names a file -- the same rule importing follows, and it
/// matters more here: this one removes something. /// matters more here: this one removes something.
/// ///
@@ -694,7 +729,48 @@ fn push_assistant(events: &mut Vec<Event>, content: &Value) {
/// keeps no copy: the JSONL *is* the session, so deleting it ends any /// keeps no copy: the JSONL *is* the session, so deleting it ends any
/// chance of resuming that conversation, including from an ai-app session /// chance of resuming that conversation, including from an ai-app session
/// that was already importing it. /// that was already importing it.
pub async fn delete(transport: &Transport, id: &str) -> Result<()> { ///
/// The whole batch in one invocation, which over ssh is the difference
/// between one connection and one per session. Six deletes started in the
/// same tick were six `ssh` processes racing to authenticate, and a batch
/// big enough to pass the remote sshd's `MaxStartups` (10 unauthenticated
/// connections, by default, before it begins refusing) had rows come back
/// as `Connection closed by … port 2222` -- a row reporting a delete that
/// never ran, for a reason that has nothing to do with the session. One
/// connection cannot exceed that however many ids are selected.
///
/// Still one outcome per id, because a batch is not a transaction: six
/// removals that must all succeed or all roll back is not something a
/// filesystem offers, and the caller settles each row from its own line.
/// Every requested id gets an entry, so an id the machine said nothing
/// about is reported as such rather than defaulting to either answer.
pub async fn delete(
transport: &Transport,
ids: &[String],
) -> Result<HashMap<String, Result<(), String>>> {
// Refused here rather than on the machine: `is_session_id` is what
// keeps an id from walking out of the projects directory, and a bad
// one must never reach the glob. It fails only itself -- one malformed
// id is not a reason to leave the other five in place.
let (safe, mut outcomes): (Vec<&String>, HashMap<String, Result<(), String>>) =
ids.iter().fold(
(Vec::new(), HashMap::new()),
|(mut safe, mut outcomes), id| {
if is_session_id(id) {
safe.push(id);
} else {
outcomes.insert(
id.clone(),
Err(format!("not a Claude Code session id: {id}")),
);
}
(safe, outcomes)
},
);
if safe.is_empty() {
return Ok(outcomes);
}
// The file name *is* the id, so the machine can find it by name. This // The file name *is* the id, so the machine can find it by name. This
// used to call `list` and search its output, which is correct and costs // used to call `list` and search its output, which is correct and costs
// a full read of every transcript on the machine -- around four seconds // a full read of every transcript on the machine -- around four seconds
@@ -702,40 +778,58 @@ pub async fn delete(transport: &Transport, id: &str) -> Result<()> {
// best part of a minute doing nothing but re-reading the same files. // best part of a minute doing nothing but re-reading the same files.
// `context_of` below already resolved an id the cheap way; this is the // `context_of` below already resolved an id the cheap way; this is the
// same lookup, and the two now agree. // same lookup, and the two now agree.
ensure!(is_session_id(id), "not a Claude Code session id: {id}"); //
// Every copy, not the first. The same id can name a file under two // Every copy of each id, not the first. The same id can name a file
// project directories -- see the de-duplication in `parse_listing` -- // under two project directories -- see the de-duplication in
// and stopping at the first left the other behind, so the row came back // `parse_listing` -- and stopping at the first left the other behind,
// on the next listing after a delete that had reported success. // so the row came back on the next listing after a delete that had
let script = r#" // reported success.
for f in "$HOME"/.claude/projects/*/"$1".jsonl; do //
[ -f "$f" ] || continue // Each id prints its own verdict rather than the loop exiting on the
rm -f "$f" || exit 1 // first failure: with a batch, exiting would leave every id after it
printf '%s\n' "$f" // unexplained. See [`DELETE_SCRIPT`] for what the words mean.
done let mut args = vec![
"#;
let launch = Launch::new(
"sh",
vec![
"-c".to_string(), "-c".to_string(),
script.to_string(), DELETE_SCRIPT.to_string(),
"sh".to_string(), "sh".to_string(),
id.to_string(), ];
], args.extend(safe.iter().map(|id| (*id).clone()));
None, let launch = Launch::new("sh", args, None);
);
// Nothing on stdout means the loop found no such file. Said here rather // A failure to run the script at all is the machine being unreachable,
// than by exiting non-zero, because a non-zero exit is reported as the // which is true of every id in the batch rather than of any one of
// machine being unreachable -- which is a different thing from the // them -- so it is returned as the error, not written into each row.
// session not being there, and only one of them is worth retrying. let reported = transport
let removed = transport
.capture(&launch) .capture(&launch)
.await .await
.with_context(|| format!("deleting Claude Code session {id}"))?; .with_context(|| format!("deleting {} Claude Code sessions", safe.len()))?;
if removed.trim().is_empty() {
bail!("no Claude Code session {id} on that machine"); for line in reported.lines() {
let Some((id, state)) = line.trim().split_once('\t') else {
continue;
};
outcomes.insert(
id.to_string(),
match state {
"deleted" => Ok(()),
"missing" => Err(format!("no Claude Code session {id} on that machine")),
_ => Err(format!("couldn't remove Claude Code session {id}")),
},
);
} }
Ok(()) // Anything the machine did not mention. The connection can drop
// part-way through the loop, and an id whose line never arrived is one
// nobody knows the fate of -- which is its own answer, and must not be
// read as either a success or a clean "not there".
for id in safe {
outcomes.entry(id.clone()).or_insert_with(|| {
Err(format!(
"couldn't tell whether Claude Code session {id} was deleted -- the machine \
stopped answering part-way through the batch"
))
});
}
Ok(outcomes)
} }
/// Whether an id is one of ours to put in a shell glob. /// Whether an id is one of ours to put in a shell glob.
@@ -940,6 +1034,43 @@ mod tests {
assert!(!is_session_id(&"a".repeat(65))); assert!(!is_session_id(&"a".repeat(65)));
} }
/// One batch, one invocation, one verdict per id -- including for the
/// two cases a single-id delete never had to keep apart from the rest:
/// an id recorded under two project directories (both copies must go,
/// and it still reports once) and an id that is not there at all.
///
/// Runs the real script against a temporary `$HOME`, because what is
/// being checked is the shell, not the Rust around it.
#[test]
fn a_batch_deletes_every_copy_and_reports_each_id_once() {
let home = tempfile::tempdir().expect("tempdir");
let projects = home.path().join(".claude/projects");
let one = "5ecf21da-d53f-4a11-9c0d-000000000100";
let twice = "5ecf21da-d53f-4a11-9c0d-000000000200";
let absent = "5ecf21da-d53f-4a11-9c0d-000000000300";
for (project, id) in [("a", one), ("a", twice), ("b", twice)] {
let dir = projects.join(project);
std::fs::create_dir_all(&dir).expect("project dir");
std::fs::write(dir.join(format!("{id}.jsonl")), "{}\n").expect("transcript");
}
let output = std::process::Command::new("sh")
.args(["-c", DELETE_SCRIPT, "sh", one, twice, absent])
.env("HOME", home.path())
.output()
.expect("run the delete script");
assert!(output.status.success());
assert_eq!(
String::from_utf8_lossy(&output.stdout),
format!("{one}\tdeleted\n{twice}\tdeleted\n{absent}\tmissing\n"),
);
// The second copy is the one a per-id delete used to leave behind.
assert!(!projects.join("b").join(format!("{twice}.jsonl")).exists());
assert!(!projects.join("a").join(format!("{twice}.jsonl")).exists());
assert!(!projects.join("a").join(format!("{one}.jsonl")).exists());
}
/// A 1x1 PNG, base64 -- the smallest thing with a real header. /// A 1x1 PNG, base64 -- the smallest thing with a real header.
const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk\ const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk\
YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="; YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==";
-19
View File
@@ -21,31 +21,12 @@
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::process::Stdio; use std::process::Stdio;
use std::sync::LazyLock;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use tokio::process::Child; use tokio::process::Child;
use tokio::sync::Semaphore;
use crate::config::SshConfig; use crate::config::SshConfig;
/// How many [`Transport::capture`] calls may have an ssh connection open at
/// once, across every setup.
///
/// `capture` is what a batch (deleting several imports, listing several
/// machines) fans out over -- one child `ssh` process per call, all started
/// within the same tick. Nothing here throttled that, so a batch large
/// enough to open more connections than the remote sshd's default
/// `MaxStartups` (10, before it starts randomly refusing) has some of
/// them come back as "Connection closed" -- not a real failure of the
/// operation, just too many handshakes landing on the listener at once. Four
/// keeps a batch comfortably under that ceiling while still overlapping the
/// network round trips. Long-lived processes (`Transport::spawn`, a
/// session's own child) don't take a permit: they hold it for the session's
/// lifetime rather than for one round trip, which would starve every other
/// probe behind it.
pub(crate) static CAPTURE_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(4));
/// What a driver needs run in order to exist as a process. /// What a driver needs run in order to exist as a process.
/// ///
/// Deliberately just the three things every transport can carry. Anything /// Deliberately just the three things every transport can carry. Anything
-7
View File
@@ -163,13 +163,6 @@ pub fn tidy(value: &str) -> Option<String> {
/// Runs a launch to completion and returns its stdout. /// Runs a launch to completion and returns its stdout.
impl Transport { impl Transport {
pub async fn capture(&self, launch: &Launch) -> Result<String> { pub async fn capture(&self, launch: &Launch) -> Result<String> {
// See `CAPTURE_PERMITS`: caps how many of these run their ssh
// connection at once, so a batch doesn't open more than the remote
// sshd tolerates before it starts dropping them.
let _permit = super::session::transport::CAPTURE_PERMITS
.acquire()
.await
.expect("capture semaphore is never closed");
let child = self.spawn(launch, super::session::transport::Streams::Piped)?; let child = self.spawn(launch, super::session::transport::Streams::Piped)?;
let output = child let output = child
.wait_with_output() .wait_with_output()