| 1 | //! Off-event-loop Lane control submission (#1888, #4022). |
| 2 | //! |
| 3 | //! `/lane interrupt` performs Runtime teardown — `tmux kill-session`, worktree |
| 4 | //! TTL cleanup, an advisory lock — none of which may run on the TUI composer |
| 5 | //! thread. Making the verb CLI-only would have been a capability regression, so |
| 6 | //! the slash surface *submits* instead: validation and availability are decided |
| 7 | //! synchronously (so a bad id is still refused immediately), the work is handed |
| 8 | //! to a dedicated worker thread, and the caller gets a typed `queued` receipt |
| 9 | //! carrying a ticket. The terminal receipt — `transitioned`, `no_change`, |
| 10 | //! `conflict`, or `failed` — arrives under that same ticket and is drained by |
| 11 | //! the UI on its next tick. |
| 12 | //! |
| 13 | //! A `queued` receipt is never reported as success. It says exactly what has |
| 14 | //! happened: nothing yet. |
| 15 | |
| 16 | use std::collections::VecDeque; |
| 17 | use std::path::PathBuf; |
| 18 | use std::sync::atomic::{AtomicU64, Ordering}; |
| 19 | use std::sync::{Arc, Condvar, Mutex}; |
| 20 | |
| 21 | use codewhale_lane::control::{ |
| 22 | ControlContext, ControlFailure, ControlFailureKind, ControlOperation, ControlReceipt, |
| 23 | ControlSurface, execute_lane_control_in, parse_target, |
| 24 | }; |
| 25 | |
| 26 | /// Maximum submissions that may be in flight before the queue refuses. |
| 27 | /// |
| 28 | /// Interrupts are operator-initiated keystrokes, not a stream: a handful of |
| 29 | /// pending teardowns is already pathological, and an unbounded queue would let |
| 30 | /// a stuck `tmux` call accumulate work the operator can neither see nor cancel. |
| 31 | pub const MAX_PENDING: usize = 16; |
| 32 | |
| 33 | #[derive(Debug)] |
| 34 | struct Submission { |
| 35 | ticket: String, |
| 36 | operation: ControlOperation, |
| 37 | raw_target: Option<String>, |
| 38 | registry_root: Option<PathBuf>, |
| 39 | } |
| 40 | |
| 41 | #[derive(Debug, Default)] |
| 42 | struct Shared { |
| 43 | pending: VecDeque<Submission>, |
| 44 | completed: Vec<ControlReceipt>, |
| 45 | /// Tickets accepted but not yet completed. Bounds the queue and lets a |
| 46 | /// duplicate interrupt of the same Lane be answered without re-queuing. |
| 47 | in_flight: Vec<(String, String)>, |
| 48 | shutdown: bool, |
| 49 | /// Whether the draining thread exists yet. Spawned on first submission so |
| 50 | /// that constructing an `App` (which tests do many times) does not create a |
| 51 | /// thread that will never be used. |
| 52 | worker_started: bool, |
| 53 | } |
| 54 | |
| 55 | /// A bounded, off-loop worker for durable Lane control verbs. |
| 56 | #[derive(Debug, Clone)] |
| 57 | pub struct LaneControlQueue { |
| 58 | shared: Arc<(Mutex<Shared>, Condvar)>, |
| 59 | tickets: Arc<AtomicU64>, |
| 60 | } |
| 61 | |
| 62 | impl Default for LaneControlQueue { |
| 63 | fn default() -> Self { |
| 64 | Self::new() |
| 65 | } |
| 66 | } |
| 67 | |
| 68 | impl LaneControlQueue { |
| 69 | /// Create a queue with no worker attached. Submissions accumulate until a |
| 70 | /// worker drains them, which is what the tests use to observe the queue |
| 71 | /// deterministically. |
| 72 | #[must_use] |
| 73 | pub fn new() -> Self { |
| 74 | Self { |
| 75 | shared: Arc::new((Mutex::new(Shared::default()), Condvar::new())), |
| 76 | tickets: Arc::new(AtomicU64::new(1)), |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | /// Start the draining thread if it is not already running. |
| 81 | /// |
| 82 | /// A plain OS thread, not a tokio task: the work is blocking by nature |
| 83 | /// (subprocess + file lock) and must not occupy an async executor slot. |
| 84 | /// Called on first submission so an idle `App` costs nothing. |
| 85 | fn ensure_worker(&self, shared: &mut Shared) { |
| 86 | if shared.worker_started { |
| 87 | return; |
| 88 | } |
| 89 | shared.worker_started = true; |
| 90 | let worker = self.clone(); |
| 91 | if std::thread::Builder::new() |
| 92 | .name("lane-control".to_string()) |
| 93 | .spawn(move || worker.run()) |
| 94 | .is_err() |
| 95 | { |
| 96 | // Could not spawn: leave the flag clear so a later submission can |
| 97 | // retry rather than queueing into a queue nothing will drain. |
| 98 | shared.worker_started = false; |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | /// Submit a Lane control verb for off-loop execution. |
| 103 | /// |
| 104 | /// Everything that can be decided without blocking is decided here and |
| 105 | /// returned synchronously: unknown verb, malformed target, unavailable |
| 106 | /// backend, saturated queue. Only the blocking teardown is deferred. |
| 107 | #[must_use] |
| 108 | pub fn submit( |
| 109 | &self, |
| 110 | operation: ControlOperation, |
| 111 | raw_target: Option<&str>, |
| 112 | registry_root: Option<PathBuf>, |
| 113 | ) -> ControlReceipt { |
| 114 | let descriptor = operation.descriptor(); |
| 115 | let surface = ControlSurface::Slash; |
| 116 | |
| 117 | // #1888: a surface must never advertise a backend that does not exist. |
| 118 | // A verb that is declared but unbuilt (`restart`, `resume`) or not |
| 119 | // offered on this surface can never succeed no matter what the durable |
| 120 | // store looks like, so it is refused here with its typed reason rather |
| 121 | // than answered `queued` for work that will never run. The stores are |
| 122 | // deliberately assumed present for this check: whether the registry |
| 123 | // exists is a *runtime* fact the executor probes on the worker thread, |
| 124 | // and probing it here would put a filesystem answer on the composer |
| 125 | // thread and make an empty workspace look like an unbuilt feature. |
| 126 | let availability = descriptor.availability(surface, ControlContext::new(true, true)); |
| 127 | if !availability.is_available() { |
| 128 | return ControlReceipt::unavailable(descriptor, surface, availability); |
| 129 | } |
| 130 | |
| 131 | // Validate the target on the calling thread: a typo must be refused |
| 132 | // now, not silently queued and refused a tick later. |
| 133 | let target = match parse_target(descriptor, raw_target) { |
| 134 | Ok(target) => target, |
| 135 | Err(failure) => { |
| 136 | return ControlReceipt::rejected(descriptor, surface, None, failure); |
| 137 | } |
| 138 | }; |
| 139 | |
| 140 | let (lock, condvar) = &*self.shared; |
| 141 | let mut shared = match lock.lock() { |
| 142 | Ok(shared) => shared, |
| 143 | // A poisoned queue means the worker panicked mid-teardown. Refuse |
| 144 | // rather than submitting into a queue nothing will drain. |
| 145 | Err(_) => { |
| 146 | return ControlReceipt::failed( |
| 147 | descriptor, |
| 148 | surface, |
| 149 | target, |
| 150 | ControlFailure::backend("Lane control worker is not running"), |
| 151 | ); |
| 152 | } |
| 153 | }; |
| 154 | |
| 155 | if shared.pending.len() >= MAX_PENDING { |
| 156 | return ControlReceipt::rejected( |
| 157 | descriptor, |
| 158 | surface, |
| 159 | target, |
| 160 | ControlFailure::new( |
| 161 | ControlFailureKind::Saturated, |
| 162 | format!( |
| 163 | "Lane control queue is full ({MAX_PENDING} pending); \ |
| 164 | nothing was submitted" |
| 165 | ), |
| 166 | ), |
| 167 | ); |
| 168 | } |
| 169 | |
| 170 | // Re-submitting the same Lane while a teardown is in flight is a |
| 171 | // conflict, not a second teardown. |
| 172 | if let Some(target) = target.as_ref() |
| 173 | && let Some((ticket, _)) = shared |
| 174 | .in_flight |
| 175 | .iter() |
| 176 | .find(|(_, id)| *id == target.id) |
| 177 | .cloned() |
| 178 | { |
| 179 | return ControlReceipt::rejected( |
| 180 | descriptor, |
| 181 | surface, |
| 182 | Some(target.clone()), |
| 183 | ControlFailure::conflict(format!( |
| 184 | "{} is already in flight under ticket {ticket}", |
| 185 | target.id |
| 186 | )), |
| 187 | ); |
| 188 | } |
| 189 | |
| 190 | let ticket = format!("lane-ctl-{}", self.tickets.fetch_add(1, Ordering::Relaxed)); |
| 191 | if let Some(target) = target.as_ref() { |
| 192 | shared.in_flight.push((ticket.clone(), target.id.clone())); |
| 193 | } |
| 194 | shared.pending.push_back(Submission { |
| 195 | ticket: ticket.clone(), |
| 196 | operation, |
| 197 | raw_target: raw_target.map(str::to_string), |
| 198 | registry_root, |
| 199 | }); |
| 200 | // Tests drive `run_once` directly and never start the thread; in |
| 201 | // production the first submission starts it. |
| 202 | if cfg!(not(test)) { |
| 203 | self.ensure_worker(&mut shared); |
| 204 | } |
| 205 | condvar.notify_one(); |
| 206 | drop(shared); |
| 207 | |
| 208 | ControlReceipt::queued(descriptor, surface, target, ticket) |
| 209 | } |
| 210 | |
| 211 | /// Take every terminal receipt produced since the last drain. |
| 212 | #[must_use] |
| 213 | pub fn drain_completed(&self) -> Vec<ControlReceipt> { |
| 214 | let (lock, _) = &*self.shared; |
| 215 | match lock.lock() { |
| 216 | Ok(mut shared) => std::mem::take(&mut shared.completed), |
| 217 | Err(_) => Vec::new(), |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | /// How many submissions are waiting. Test-only: nothing in the UI reads |
| 222 | /// backpressure yet, and an unused public accessor would be a claim the |
| 223 | /// build does not back. |
| 224 | #[cfg(test)] |
| 225 | #[must_use] |
| 226 | pub fn pending_len(&self) -> usize { |
| 227 | let (lock, _) = &*self.shared; |
| 228 | lock.lock().map(|shared| shared.pending.len()).unwrap_or(0) |
| 229 | } |
| 230 | |
| 231 | /// Execute one queued submission if there is one, on the calling thread. |
| 232 | /// |
| 233 | /// This is the worker's body, exposed so tests can drain deterministically |
| 234 | /// without a thread. Returns `false` when the queue was empty. |
| 235 | pub fn run_once(&self) -> bool { |
| 236 | let (lock, _) = &*self.shared; |
| 237 | let Some(submission) = ({ |
| 238 | let Ok(mut shared) = lock.lock() else { |
| 239 | return false; |
| 240 | }; |
| 241 | shared.pending.pop_front() |
| 242 | }) else { |
| 243 | return false; |
| 244 | }; |
| 245 | |
| 246 | // The blocking call. Deliberately outside the lock so a slow teardown |
| 247 | // never blocks a submission or a drain. |
| 248 | let receipt = execute_lane_control_in( |
| 249 | ControlSurface::Cli, |
| 250 | submission.operation, |
| 251 | submission.raw_target.as_deref(), |
| 252 | submission.registry_root.as_deref(), |
| 253 | ); |
| 254 | // The work ran off the composer thread, but it was *requested* from the |
| 255 | // slash surface; the receipt must say so rather than impersonating the |
| 256 | // CLI. |
| 257 | let mut receipt = receipt.with_ticket(submission.ticket.clone()); |
| 258 | receipt.surface = ControlSurface::Slash; |
| 259 | |
| 260 | if let Ok(mut shared) = lock.lock() { |
| 261 | shared |
| 262 | .in_flight |
| 263 | .retain(|(ticket, _)| *ticket != submission.ticket); |
| 264 | shared.completed.push(receipt); |
| 265 | } |
| 266 | true |
| 267 | } |
| 268 | |
| 269 | fn run(&self) { |
| 270 | let (lock, condvar) = &*self.shared; |
| 271 | loop { |
| 272 | { |
| 273 | let Ok(mut shared) = lock.lock() else { |
| 274 | return; |
| 275 | }; |
| 276 | while shared.pending.is_empty() && !shared.shutdown { |
| 277 | let Ok(next) = condvar.wait(shared) else { |
| 278 | return; |
| 279 | }; |
| 280 | shared = next; |
| 281 | } |
| 282 | if shared.shutdown && shared.pending.is_empty() { |
| 283 | return; |
| 284 | } |
| 285 | } |
| 286 | self.run_once(); |
| 287 | } |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | #[cfg(test)] |
| 292 | mod tests { |
| 293 | use super::*; |
| 294 | use codewhale_lane::{LaneRegistry, LaneStatus, LifecycleOutcome, RuntimeBackendKind}; |
| 295 | |
| 296 | fn seeded() -> (tempfile::TempDir, String) { |
| 297 | let dir = tempfile::tempdir().unwrap(); |
| 298 | let registry = LaneRegistry::open(dir.path()).unwrap(); |
| 299 | let record = registry |
| 300 | .create_pending( |
| 301 | Some("stopship".into()), |
| 302 | Some("stopship".into()), |
| 303 | Some("4022".into()), |
| 304 | None, |
| 305 | RuntimeBackendKind::Inline, |
| 306 | None, |
| 307 | ) |
| 308 | .unwrap(); |
| 309 | let id = record.id.clone(); |
| 310 | (dir, id) |
| 311 | } |
| 312 | |
| 313 | /// #4022: the production slash path returns immediately with a typed |
| 314 | /// `queued` receipt and only then performs the blocking teardown. |
| 315 | #[test] |
| 316 | fn slash_interrupt_returns_queued_then_completes_off_loop() { |
| 317 | let (dir, id) = seeded(); |
| 318 | let queue = LaneControlQueue::new(); |
| 319 | |
| 320 | let submitted = queue.submit( |
| 321 | ControlOperation::LaneInterrupt, |
| 322 | Some(id.as_str()), |
| 323 | Some(dir.path().to_path_buf()), |
| 324 | ); |
| 325 | assert_eq!(submitted.outcome, LifecycleOutcome::Queued); |
| 326 | assert_eq!(submitted.surface, ControlSurface::Slash); |
| 327 | assert!(submitted.ticket.is_some()); |
| 328 | assert!( |
| 329 | !submitted.is_error(), |
| 330 | "queued is not an error, but it is also not success" |
| 331 | ); |
| 332 | // Nothing has happened to durable state yet. |
| 333 | assert_eq!( |
| 334 | LaneRegistry::open(dir.path()) |
| 335 | .unwrap() |
| 336 | .load(&id) |
| 337 | .unwrap() |
| 338 | .status, |
| 339 | LaneStatus::Pending |
| 340 | ); |
| 341 | |
| 342 | assert!(queue.run_once()); |
| 343 | let completed = queue.drain_completed(); |
| 344 | assert_eq!(completed.len(), 1); |
| 345 | assert_eq!(completed[0].ticket, submitted.ticket); |
| 346 | assert_eq!(completed[0].outcome, LifecycleOutcome::Transitioned); |
| 347 | assert_eq!( |
| 348 | completed[0].surface, |
| 349 | ControlSurface::Slash, |
| 350 | "the receipt reports who asked, not which thread ran it" |
| 351 | ); |
| 352 | assert_eq!( |
| 353 | LaneRegistry::open(dir.path()) |
| 354 | .unwrap() |
| 355 | .load(&id) |
| 356 | .unwrap() |
| 357 | .status, |
| 358 | LaneStatus::Stopped |
| 359 | ); |
| 360 | } |
| 361 | |
| 362 | /// A malformed id is refused on the calling thread, not queued. |
| 363 | #[test] |
| 364 | fn invalid_targets_are_refused_synchronously() { |
| 365 | let queue = LaneControlQueue::new(); |
| 366 | for bad in [None, Some(""), Some("../escape"), Some("a b")] { |
| 367 | let receipt = queue.submit(ControlOperation::LaneInterrupt, bad, None); |
| 368 | assert_eq!( |
| 369 | receipt.failure.as_ref().map(|failure| failure.kind), |
| 370 | Some(ControlFailureKind::InvalidTarget), |
| 371 | "{bad:?}" |
| 372 | ); |
| 373 | assert!(receipt.ticket.is_none()); |
| 374 | } |
| 375 | assert_eq!(queue.pending_len(), 0); |
| 376 | } |
| 377 | |
| 378 | /// #1888: a verb with no backend must be refused on the calling thread. |
| 379 | /// Queueing it would answer `queued` — a receipt that implies work is |
| 380 | /// under way — for work that can never run. |
| 381 | #[test] |
| 382 | fn declared_but_unbuilt_verbs_are_refused_without_queueing() { |
| 383 | let queue = LaneControlQueue::new(); |
| 384 | for operation in [ControlOperation::LaneRestart, ControlOperation::LaneResume] { |
| 385 | let receipt = queue.submit(operation, Some("lane-a1b2c3d4"), None); |
| 386 | assert_eq!(receipt.outcome, LifecycleOutcome::Rejected); |
| 387 | assert_eq!( |
| 388 | receipt.availability.reason(), |
| 389 | Some(codewhale_lane::UnavailableReason::BackendNotImplemented), |
| 390 | "{operation:?}" |
| 391 | ); |
| 392 | assert!(receipt.ticket.is_none()); |
| 393 | assert!(!receipt.retryable); |
| 394 | } |
| 395 | assert_eq!(queue.pending_len(), 0, "nothing may have been enqueued"); |
| 396 | } |
| 397 | |
| 398 | /// #1888: a saturated queue refuses with a typed, retryable receipt rather |
| 399 | /// than growing without bound or blocking the composer. |
| 400 | #[test] |
| 401 | fn a_saturated_queue_refuses_without_submitting() { |
| 402 | let queue = LaneControlQueue::new(); |
| 403 | for index in 0..MAX_PENDING { |
| 404 | let receipt = queue.submit( |
| 405 | ControlOperation::LaneInterrupt, |
| 406 | Some(format!("lane-{index:08}").as_str()), |
| 407 | None, |
| 408 | ); |
| 409 | assert_eq!(receipt.outcome, LifecycleOutcome::Queued); |
| 410 | } |
| 411 | assert_eq!(queue.pending_len(), MAX_PENDING); |
| 412 | |
| 413 | let refused = queue.submit(ControlOperation::LaneInterrupt, Some("lane-overflow"), None); |
| 414 | assert_eq!(refused.outcome, LifecycleOutcome::Rejected); |
| 415 | assert_eq!( |
| 416 | refused.failure.as_ref().map(|failure| failure.kind), |
| 417 | Some(ControlFailureKind::Saturated) |
| 418 | ); |
| 419 | assert!(refused.retryable, "saturation clears on its own"); |
| 420 | assert!(refused.ticket.is_none()); |
| 421 | assert_eq!( |
| 422 | queue.pending_len(), |
| 423 | MAX_PENDING, |
| 424 | "the refused submission must not have been enqueued" |
| 425 | ); |
| 426 | } |
| 427 | |
| 428 | /// Re-pressing interrupt while a teardown is in flight is a conflict, not |
| 429 | /// a second teardown of the same Lane. |
| 430 | #[test] |
| 431 | fn a_duplicate_submission_for_one_lane_is_a_conflict() { |
| 432 | let queue = LaneControlQueue::new(); |
| 433 | let first = queue.submit(ControlOperation::LaneInterrupt, Some("lane-a1b2c3d4"), None); |
| 434 | assert_eq!(first.outcome, LifecycleOutcome::Queued); |
| 435 | |
| 436 | let second = queue.submit(ControlOperation::LaneInterrupt, Some("lane-a1b2c3d4"), None); |
| 437 | assert_eq!( |
| 438 | second.failure.as_ref().map(|failure| failure.kind), |
| 439 | Some(ControlFailureKind::Conflict) |
| 440 | ); |
| 441 | assert_eq!(queue.pending_len(), 1); |
| 442 | } |
| 443 | |
| 444 | /// #4022 (TUI responsiveness): submission must not perform the blocking |
| 445 | /// work. A queue with no worker still returns promptly, which is only |
| 446 | /// possible if `submit` never touches the Runtime. |
| 447 | #[test] |
| 448 | fn submission_does_not_block_on_runtime_teardown() { |
| 449 | let (dir, id) = seeded(); |
| 450 | let queue = LaneControlQueue::new(); |
| 451 | let started = std::time::Instant::now(); |
| 452 | let receipt = queue.submit( |
| 453 | ControlOperation::LaneInterrupt, |
| 454 | Some(id.as_str()), |
| 455 | Some(dir.path().to_path_buf()), |
| 456 | ); |
| 457 | let elapsed = started.elapsed(); |
| 458 | assert_eq!(receipt.outcome, LifecycleOutcome::Queued); |
| 459 | assert!( |
| 460 | elapsed < std::time::Duration::from_millis(250), |
| 461 | "submission took {elapsed:?}; it must not run teardown inline" |
| 462 | ); |
| 463 | // Proof it really was deferred: durable state is untouched until a |
| 464 | // worker runs, and no worker has. |
| 465 | assert_eq!( |
| 466 | LaneRegistry::open(dir.path()) |
| 467 | .unwrap() |
| 468 | .load(&id) |
| 469 | .unwrap() |
| 470 | .status, |
| 471 | LaneStatus::Pending |
| 472 | ); |
| 473 | } |
| 474 | |
| 475 | /// The fence still refuses under the registry lock when the work finally |
| 476 | /// runs off-loop. |
| 477 | #[test] |
| 478 | fn a_stale_fence_conflicts_when_the_queued_work_runs() { |
| 479 | let (dir, id) = seeded(); |
| 480 | let queue = LaneControlQueue::new(); |
| 481 | let submitted = queue.submit( |
| 482 | ControlOperation::LaneInterrupt, |
| 483 | Some(format!("{id}@99").as_str()), |
| 484 | Some(dir.path().to_path_buf()), |
| 485 | ); |
| 486 | assert_eq!(submitted.outcome, LifecycleOutcome::Queued); |
| 487 | |
| 488 | assert!(queue.run_once()); |
| 489 | let completed = queue.drain_completed(); |
| 490 | assert_eq!(completed.len(), 1); |
| 491 | assert_eq!(completed[0].outcome, LifecycleOutcome::Rejected); |
| 492 | assert_eq!( |
| 493 | completed[0].failure.as_ref().map(|failure| failure.kind), |
| 494 | Some(ControlFailureKind::Conflict) |
| 495 | ); |
| 496 | assert_eq!( |
| 497 | LaneRegistry::open(dir.path()) |
| 498 | .unwrap() |
| 499 | .load(&id) |
| 500 | .unwrap() |
| 501 | .status, |
| 502 | LaneStatus::Pending, |
| 503 | "a refused fence must not have torn anything down" |
| 504 | ); |
| 505 | } |
| 506 | |
| 507 | /// A completed ticket frees its slot, so the queue does not leak capacity. |
| 508 | #[test] |
| 509 | fn completing_a_submission_frees_its_in_flight_slot() { |
| 510 | let (dir, id) = seeded(); |
| 511 | let queue = LaneControlQueue::new(); |
| 512 | let _ = queue.submit( |
| 513 | ControlOperation::LaneInterrupt, |
| 514 | Some(id.as_str()), |
| 515 | Some(dir.path().to_path_buf()), |
| 516 | ); |
| 517 | assert!(queue.run_once()); |
| 518 | let _ = queue.drain_completed(); |
| 519 | |
| 520 | let again = queue.submit( |
| 521 | ControlOperation::LaneInterrupt, |
| 522 | Some(id.as_str()), |
| 523 | Some(dir.path().to_path_buf()), |
| 524 | ); |
| 525 | assert_eq!( |
| 526 | again.outcome, |
| 527 | LifecycleOutcome::Queued, |
| 528 | "the slot must be reusable once the ticket completes" |
| 529 | ); |
| 530 | } |
| 531 | } |
| 532 |