diff --git a/server/src/routes.rs b/server/src/routes.rs index f55dfec..573f705 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -549,28 +549,68 @@ struct ImportableRow { /// 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. /// -/// Registering is what has to be atomic; the work itself does not. Each id -/// runs on its own task and settles on its own event, because 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. +/// Registering is what has to be atomic; the work itself does not. The +/// batch runs as one command on the machine -- see +/// [`crate::session::import::delete`] for why it is not one per id -- but +/// each row still settles on its own event from its own outcome, because +/// 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( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { let setup = setup_by_id(&manager, &id)?; - for session in body.sessions { - let transport = crate::session::transport::Transport::for_setup(&setup); - let target = session.clone(); - in_background( - &manager, - id.clone(), - session, - Operation::Deleting, - async move { crate::session::import::delete(&transport, &target).await }, - ); - } + let transport = crate::session::transport::Transport::for_setup(&setup); + // Registered before anything is spawned, so the 202 is only sent once + // every row is already showing "deleting" -- a phone that refetches the + // instant it gets the reply cannot catch a row that has not started. + let running: Vec<(String, crate::session::pending::InFlight)> = body + .sessions + .iter() + .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) } @@ -872,9 +912,15 @@ async fn delete_session( if let Some((setup, session)) = &foreign { let setup = setup_by_id(&manager, 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 - .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}"); } manager.delete_session(&id).map_err(bad_request)?; diff --git a/server/src/session/import.rs b/server/src/session/import.rs index c43f6dc..67f460d 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -17,7 +17,9 @@ //! reason: an enrolled token must not be able to turn into "read me this //! arbitrary path". -use anyhow::{Context, Result, bail, ensure}; +use std::collections::HashMap; + +use anyhow::{Context, Result}; use serde::Serialize; use serde_json::Value; @@ -684,9 +686,42 @@ fn push_assistant(events: &mut Vec, content: &Value) { } } -/// Deletes one of the sessions [`list`] reported. +/// Removes each id it is given and prints one `\t` 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 /// matters more here: this one removes something. /// @@ -694,7 +729,48 @@ fn push_assistant(events: &mut Vec, content: &Value) { /// keeps no copy: the JSONL *is* the session, so deleting it ends any /// chance of resuming that conversation, including from an ai-app session /// 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>> { + // 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>) = + 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 // 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 @@ -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. // `context_of` below already resolved an id the cheap way; this is the // 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 - // 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. - let script = r#" -for f in "$HOME"/.claude/projects/*/"$1".jsonl; do - [ -f "$f" ] || continue - rm -f "$f" || exit 1 - printf '%s\n' "$f" -done -"#; - let launch = Launch::new( - "sh", - vec![ - "-c".to_string(), - script.to_string(), - "sh".to_string(), - id.to_string(), - ], - None, - ); - // Nothing on stdout means the loop found no such file. Said here rather - // than by exiting non-zero, because a non-zero exit is reported as the - // machine being unreachable -- which is a different thing from the - // session not being there, and only one of them is worth retrying. - let removed = transport + // + // 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. + // + // Each id prints its own verdict rather than the loop exiting on the + // first failure: with a batch, exiting would leave every id after it + // unexplained. See [`DELETE_SCRIPT`] for what the words mean. + let mut args = vec![ + "-c".to_string(), + DELETE_SCRIPT.to_string(), + "sh".to_string(), + ]; + args.extend(safe.iter().map(|id| (*id).clone())); + let launch = Launch::new("sh", args, None); + + // A failure to run the script at all is the machine being unreachable, + // which is true of every id in the batch rather than of any one of + // them -- so it is returned as the error, not written into each row. + let reported = transport .capture(&launch) .await - .with_context(|| format!("deleting Claude Code session {id}"))?; - if removed.trim().is_empty() { - bail!("no Claude Code session {id} on that machine"); + .with_context(|| format!("deleting {} Claude Code sessions", safe.len()))?; + + 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. @@ -940,6 +1034,43 @@ mod tests { 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. const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk\ YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="; diff --git a/server/src/session/transport.rs b/server/src/session/transport.rs index 99ca972..b075c75 100644 --- a/server/src/session/transport.rs +++ b/server/src/session/transport.rs @@ -21,31 +21,12 @@ use std::path::{Path, PathBuf}; use std::process::Stdio; -use std::sync::LazyLock; use anyhow::{Context, Result}; use tokio::process::Child; -use tokio::sync::Semaphore; 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 = LazyLock::new(|| Semaphore::new(4)); - /// What a driver needs run in order to exist as a process. /// /// Deliberately just the three things every transport can carry. Anything diff --git a/server/src/setups.rs b/server/src/setups.rs index 61d3543..b3d3a82 100644 --- a/server/src/setups.rs +++ b/server/src/setups.rs @@ -163,13 +163,6 @@ pub fn tidy(value: &str) -> Option { /// Runs a launch to completion and returns its stdout. impl Transport { pub async fn capture(&self, launch: &Launch) -> Result { - // 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 output = child .wait_with_output()