diff --git a/crates/spt-proto/src/emit.rs b/crates/spt-proto/src/emit.rs index 6ccd60ed..b4a799cb 100644 --- a/crates/spt-proto/src/emit.rs +++ b/crates/spt-proto/src/emit.rs @@ -182,30 +182,180 @@ macro_rules! emit_block { /// attempted. That is an observability gap, accepted here only because the /// alternative is a panic, and it belongs in the conversion notes so no later /// reader rediscovers it as a defect. +/// +/// It honours a [`Capture`] active on the calling thread: a captured emission is +/// composed exactly as it would have been written and held for the capture's +/// owner to replay, so a line a callee emits deep inside a captured unit of work +/// lands in that unit's block rather than interleaving with its siblings'. // [impl->REQ-EMIT-SINGLE-WRITE] +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] #[macro_export] macro_rules! emit_line_err { ($($arg:tt)*) => {{ - let _ = $crate::emit::write_line( - &mut ::std::io::stderr().lock(), - &::std::format!($($arg)*), - ); + $crate::emit::line_to($crate::emit::Stream::Err, &::std::format!($($arg)*)); }}; } /// [`emit_block!`] aimed at this process's stderr, on the same terms as -/// [`emit_line_err!`]: separate name, error discarded. +/// [`emit_line_err!`]: separate name, error discarded, capture honoured. // [impl->REQ-EMIT-SINGLE-WRITE] #[macro_export] macro_rules! emit_block_err { ($($arg:tt)*) => {{ - let _ = $crate::emit::write_block( - &mut ::std::io::stderr().lock(), - &::std::format!($($arg)*), + $crate::emit::deliver( + $crate::emit::Stream::Err, + &$crate::emit::compose_block(&::std::format!($($arg)*)), ); }}; } +/// `eprintln!`, capture-aware: the rendered text plus one newline, verbatim (no +/// escaping — this is the drop-in for prose and for lines whose payload may carry +/// a caller's own multi-line error text), delivered as ONE write to stderr or to +/// the calling thread's active [`Capture`]. The write error is discarded on the +/// same terms as [`emit_line_err!`]. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +#[macro_export] +macro_rules! cap_eprintln { + ($($arg:tt)*) => {{ + let mut __s = ::std::format!($($arg)*); + __s.push('\n'); + $crate::emit::deliver($crate::emit::Stream::Err, &__s); + }}; +} + +/// `println!`, capture-aware — [`cap_eprintln!`] aimed at stdout. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +#[macro_export] +macro_rules! cap_println { + ($($arg:tt)*) => {{ + let mut __s = ::std::format!($($arg)*); + __s.push('\n'); + $crate::emit::deliver($crate::emit::Stream::Out, &__s); + }}; +} + +/// Which process handle an emission is aimed at. A [`Capture`] keeps the answer +/// per chunk so a replay puts every chunk back on the handle it was meant for: +/// an adapter's stdout notice must not arrive on stderr just because it was held. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Stream { + Out, + Err, +} + +thread_local! { + /// The calling thread's active capture, if any. Thread-local on purpose: the + /// unit of isolation is the thread doing one unit of work, and a process-wide + /// sink would put sibling threads' lines back into one interleaved stream. + static CAPTURE: std::cell::RefCell>> = + const { std::cell::RefCell::new(None) }; +} + +/// The single-line path of [`emit_line_err!`] with the capture check: refuse an +/// interior break (debug), compose, then [`deliver`]. +// [impl->REQ-EMIT-SINGLE-WRITE] +pub fn line_to(stream: Stream, rendered: &str) { + debug_assert!( + !has_interior_break(rendered), + "emit_line! got an interior line break — use emit_block! if the emission \ + really is multi-line; the release build escapes it to {ESCAPED_NEWLINE} \ + so one emission stays one logical line: {rendered:?}" + ); + deliver(stream, &compose_line(rendered)); +} + +/// Hand one COMPOSED emission to the calling thread's active [`Capture`], or — +/// with none active — write it to `stream`'s process handle as one call, the +/// write error discarded. Every capture-aware macro funnels through here, so the +/// capture decision is made in exactly one place. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +pub fn deliver(stream: Stream, composed: &str) { + let held = CAPTURE.with(|c| match c.borrow_mut().as_mut() { + Some(chunks) => { + chunks.push((stream, composed.to_string())); + true + } + None => false, + }); + if held { + return; + } + let _ = match stream { + Stream::Out => write_composed(&mut io::stdout().lock(), composed), + Stream::Err => write_composed(&mut io::stderr().lock(), composed), + }; +} + +/// Hold every capture-aware emission made on THIS thread until [`Capture::finish`]. +/// +/// The seam that lets N units of work run on N threads and still print as N +/// whole blocks: each thread begins a capture, does its work, and hands the +/// chunks to its parent, which replays them when the unit is done. Only +/// emissions that reach [`deliver`] are held — a bare `eprintln!` or a child +/// process on an inherited handle bypasses it, which is why the converted paths +/// say so at their macros rather than trusting this type alone. +/// +/// Not nestable: beginning a capture while one is active on the thread would +/// silently steal the outer capture's later lines, so it panics instead. +/// Dropping an unfinished capture discards what it held and ends it, so a +/// panicking unit cannot leave the thread capturing forever. +#[derive(Debug)] +#[must_use = "a capture holds output until finish(); dropping it discards the output"] +pub struct Capture { + _not_send: std::marker::PhantomData<*const ()>, +} + +impl Capture { + /// Begin capturing on the calling thread. + pub fn begin() -> Capture { + CAPTURE.with(|c| { + let mut slot = c.borrow_mut(); + assert!(slot.is_none(), "emit::Capture is not nestable on one thread"); + *slot = Some(Vec::new()); + }); + Capture { + _not_send: std::marker::PhantomData, + } + } + + /// End the capture and return every held chunk in emission order. + pub fn finish(self) -> Vec<(Stream, String)> { + let chunks = CAPTURE.with(|c| c.borrow_mut().take()).unwrap_or_default(); + std::mem::forget(self); + chunks + } +} + +impl Drop for Capture { + fn drop(&mut self) { + CAPTURE.with(|c| c.borrow_mut().take()); + } +} + +/// Replay captured chunks, each onto the handle it was aimed at, in order. +/// Generic over both writers so the replay is unit-pinnable without touching the +/// process's real handles. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +pub fn replay_to( + chunks: &[(Stream, String)], + out: &mut O, + err: &mut E, +) { + for (stream, text) in chunks { + let _ = match stream { + Stream::Out => write_composed(out, text), + Stream::Err => write_composed(err, text), + }; + } +} + +/// [`replay_to`] onto this process's real stdout/stderr. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +pub fn replay(chunks: &[(Stream, String)]) { + replay_to(chunks, &mut io::stdout().lock(), &mut io::stderr().lock()); +} + /// A writer that counts the calls it receives and keeps every buffer verbatim. /// /// Public because the property this module promises is a claim about CALL COUNT, @@ -540,4 +690,100 @@ mod tests { emit_block!(&mut w2, "TOKEN:a: line one\nline two\n").expect("write"); assert_eq!(w2.sole_write(), b"TOKEN:a: line one\nline two\n"); } + + /// Every capture-aware spelling lands in the active capture, composed the + /// way it would have been written and tagged with the handle it was aimed + /// at — including `emit_line_err!`, the macro a callee in another crate + /// uses, which is the reason the capture lives in this module at all. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn a_capture_holds_every_spelling_in_order_with_its_stream() { + let cap = Capture::begin(); + crate::emit_line_err!("TOKEN:a: {}", 1); + crate::cap_println!("notice for {}", "a"); + crate::cap_eprintln!("FAIL:a: detail\nsecond line"); + crate::emit_block_err!("BLOCK:a\nrest"); + let chunks = cap.finish(); + assert_eq!( + chunks, + vec![ + (Stream::Err, "TOKEN:a: 1\n".to_string()), + (Stream::Out, "notice for a\n".to_string()), + // cap_eprintln! is eprintln!'s drop-in: verbatim, never escaped. + (Stream::Err, "FAIL:a: detail\nsecond line\n".to_string()), + (Stream::Err, "BLOCK:a\nrest\n".to_string()), + ] + ); + } + + /// THE ISOLATION PROPERTY: two threads emitting in lock-step, each under its + /// own capture, each get back exactly their own lines. A process-wide sink + /// would hand both of them the interleaved union, which is the defect the + /// per-adapter blocks exist to retire. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn captures_on_two_threads_never_see_each_others_lines() { + let barrier = std::sync::Arc::new(std::sync::Barrier::new(2)); + let run = |who: &'static str, barrier: std::sync::Arc| { + std::thread::spawn(move || { + let cap = Capture::begin(); + for i in 0..5 { + barrier.wait(); + crate::emit_line_err!("LINE:{who}: {i}"); + } + cap.finish() + }) + }; + let a = run("a", barrier.clone()); + let b = run("b", barrier); + let a = a.join().expect("thread a"); + let b = b.join().expect("thread b"); + let expect = |who: &str| -> Vec<(Stream, String)> { + (0..5) + .map(|i| (Stream::Err, format!("LINE:{who}: {i}\n"))) + .collect() + }; + assert_eq!(a, expect("a")); + assert_eq!(b, expect("b")); + } + + /// A dropped (never finished) capture ends: the thread is not left capturing + /// forever, so a unit that panics mid-work cannot swallow the next unit's + /// output on a reused thread. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn a_dropped_capture_ends_and_a_new_one_starts_empty() { + { + let _cap = Capture::begin(); + crate::emit_line_err!("LOST:a"); + } + let cap = Capture::begin(); + crate::emit_line_err!("KEPT:b"); + assert_eq!(cap.finish(), vec![(Stream::Err, "KEPT:b\n".to_string())]); + } + + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + #[should_panic(expected = "not nestable")] + fn a_nested_capture_is_refused_not_silently_stolen() { + let _outer = Capture::begin(); + let _inner = Capture::begin(); + } + + /// Replay puts each chunk back on the handle it was aimed at, one write per + /// chunk, in order. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn replay_routes_each_chunk_to_its_own_handle() { + let chunks = vec![ + (Stream::Err, "E1\n".to_string()), + (Stream::Out, "O1\n".to_string()), + (Stream::Err, "E2\n".to_string()), + ]; + let mut out = CountingWriter::new(); + let mut err = CountingWriter::new(); + replay_to(&chunks, &mut out, &mut err); + assert_eq!(out.writes, vec![b"O1\n".to_vec()]); + assert_eq!(err.writes, vec![b"E1\n".to_vec(), b"E2\n".to_vec()]); + } } diff --git a/crates/spt/src/cli.rs b/crates/spt/src/cli.rs index f6d1c40f..fe979add 100644 --- a/crates/spt/src/cli.rs +++ b/crates/spt/src/cli.rs @@ -21196,6 +21196,15 @@ fn cmd_adapter(action: AdapterCmd, json: bool) -> i32 { "ADAPTER_ADD:{}:{:?}:{:?} (registered)", record.name, record.kind, record.mode ); + // An install is the first update (releases#62): heal a declared entry + // binary that arrived without its exec bit, loud, before its service is + // asked to start. + // [impl->REQ-ADAPTER-ENTRY-EXEC-BIT] + spt_runtime::entry_exec::force_entry_exec( + &manifest, + std::path::Path::new(&record.source_dir), + &record.name, + ); // Installing an adapter never requires restarting spt to bring its // service up (the operator addition folded into REQ-RESIDENT-SERVICE). nudge_adapter_service(&manifest, &record.name); @@ -21498,10 +21507,17 @@ fn cmd_adapter(action: AdapterCmd, json: bool) -> i32 { /// `remove_dir_all(dest)` + blanket-extract, which on Windows fails the WHOLE /// update the instant a single install-dir binary is still running (the field /// brick). Identical files — and their still-running binaries — are left -/// untouched. Additive: a file dropped by the new version is left in place -/// (adapter file deletions are rare; a stale unreferenced file is harmless), so a -/// running binary that vanished from the manifest is never yanked mid-update. +/// untouched. +/// +/// A file the new version dropped is handled by WHERE it lives (releases#278). +/// Outside `strings/` the swap is additive: the file stays, so a running binary +/// that vanished from the archive is never yanked mid-update. Under `strings/` +/// the install MIRRORS the archive: a dropped file is pruned once the swap has +/// committed. A stale file there is not harmless — a harness discovers +/// `strings/` by scanning it, so a retired skill left behind keeps firing. The +/// daemon-coordinated apply runs the same plan, so both routes prune alike. // [impl->REQ-ADAPTER-LIVE-UPDATE] +// [impl->REQ-ADAPTER-UPDATE-PRUNES-STRINGS] fn apply_release_crc_swap( staged: &std::path::Path, dest: &std::path::Path, @@ -21767,7 +21783,7 @@ fn nudge_adapter_service(manifest: &spt_runtime::manifest::Manifest, adapter: &s ) { Ok(o) => o, Err(_) => { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_SERVICE_PENDING:{adapter}: service declared, but the daemon is not \ running — it starts at the next daemon boot" ); @@ -21794,8 +21810,8 @@ fn nudge_adapter_service(manifest: &spt_runtime::manifest::Manifest, adapter: &s } }; match o.detail { - Some(d) => eprintln!("ADAPTER_SERVICE:{}: {what} ({d})", o.option), - None => eprintln!("ADAPTER_SERVICE:{}: {what}", o.option), + Some(d) => spt_proto::cap_eprintln!("ADAPTER_SERVICE:{}: {what} ({d})", o.option), + None => spt_proto::cap_eprintln!("ADAPTER_SERVICE:{}: {what}", o.option), } } } @@ -21807,7 +21823,7 @@ fn nudge_serving_registry() { if let Err(error) = spt_daemon::servehost::reconcile(&spt_daemon::endpoint::seed_socket_name()) { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_WEB_PENDING: serving registry reconciliation failed: {error}; \ retry at the next daemon start" ); @@ -22072,7 +22088,7 @@ fn run_update_post_step( // No post-step → today's behavior: an applied update fires the message. if adapter_applied { if let Some(notice) = manifest_message { - println!("{notice}"); + spt_proto::cap_println!("{notice}"); } } return true; @@ -22099,7 +22115,7 @@ fn run_update_post_step( let emit_fallback = |adapter_applied: bool, msg: &Option| { if adapter_applied { if let Some(notice) = msg { - println!("{notice}"); + spt_proto::cap_println!("{notice}"); } } }; @@ -22116,18 +22132,18 @@ fn run_update_post_step( PostNotice::None => {} PostNotice::ManifestMessage => { if let Some(notice) = manifest_message { - println!("{notice}"); + spt_proto::cap_println!("{notice}"); } } PostNotice::Custom(text) => { - println!("{}", crate::helpfmt::render(text, color).trim_end()); + spt_proto::cap_println!("{}", crate::helpfmt::render(text, color).trim_end()); } } true } Ok(out) => { let detail = out.stderr.trim(); - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_POST_FAIL:{adapter_name}: post-step exited {:?}{}", out.status_code, if detail.is_empty() { @@ -22140,7 +22156,7 @@ fn run_update_post_step( false } Err(e) => { - eprintln!("ADAPTER_UPDATE_POST_FAIL:{adapter_name}: post-step did not run: {e}"); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_POST_FAIL:{adapter_name}: post-step did not run: {e}"); emit_fallback(adapter_applied, &manifest_message); false } @@ -22260,6 +22276,101 @@ fn adapter_update_exit(outcomes: &[(String, AdapterUpdateOutcome)]) -> i32 { 0 } +/// Run `work(i)` for every adapter in `names` on its own thread and return the +/// outcomes in `names` order (releases#335). Each thread runs under an +/// [`spt_proto::emit::Capture`]; its held output is printed as ONE block the +/// moment that adapter finishes, so two adapters' lines never interleave and a +/// slow adapter never holds back a fast one's block. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +fn fan_out_adapter_updates(names: &[String], work: F) -> Vec<(String, AdapterUpdateOutcome)> +where + F: Fn(usize) -> AdapterUpdateOutcome + Sync, +{ + fan_out_adapter_updates_to(names, work, &mut |chunks| spt_proto::emit::replay(chunks)) +} + +/// [`fan_out_adapter_updates`] with the block sink as a parameter, so a unit can +/// observe exactly which blocks arrive, whole, and in what order. +/// +/// A thread that PANICS is still an answer: its adapter reports +/// `ADAPTER_UPDATE_FAIL:: update thread panicked` in its own block and +/// counts as `Failed`, never as a missing summary line — the sweep's one-line- +/// per-adapter contract holds for the adapter that broke as much as for the +/// rest. A thread that cannot be STARTED runs its adapter inline on the parent, +/// loudly, rather than dropping it. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +fn fan_out_adapter_updates_to( + names: &[String], + work: F, + on_block: &mut R, +) -> Vec<(String, AdapterUpdateOutcome)> +where + F: Fn(usize) -> AdapterUpdateOutcome + Sync, + R: FnMut(&[(spt_proto::emit::Stream, String)]), +{ + type Done = (usize, Vec<(spt_proto::emit::Stream, String)>, AdapterUpdateOutcome); + let run_one = |i: usize| -> Done { + let cap = spt_proto::emit::Capture::begin(); + let outcome = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(i))) { + Ok(o) => o, + Err(_) => { + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: update thread panicked", names[i]); + AdapterUpdateOutcome::Failed + } + }; + (i, cap.finish(), outcome) + }; + + let mut slots: Vec> = names.iter().map(|_| None).collect(); + std::thread::scope(|s| { + let (tx, rx) = std::sync::mpsc::channel::(); + let mut inline: Vec = Vec::new(); + for (i, name) in names.iter().enumerate() { + let tx = tx.clone(); + let run_one = &run_one; + let spawned = std::thread::Builder::new() + .name(format!("adapter-update-{name}")) + .spawn_scoped(s, move || { + let _ = tx.send(run_one(i)); + }); + if let Err(e) = spawned { + spt_proto::cap_eprintln!( + "ADAPTER_UPDATE_INLINE:{name}: could not start its thread ({e}) — \ + updating it on the parent after the others" + ); + inline.push(i); + } + } + drop(tx); + for (i, chunks, outcome) in rx { + on_block(&chunks); + slots[i] = Some(outcome); + } + for i in inline { + let (i, chunks, outcome) = run_one(i); + on_block(&chunks); + slots[i] = Some(outcome); + } + }); + names + .iter() + .zip(slots) + .map(|(n, o)| (n.clone(), o.unwrap_or(AdapterUpdateOutcome::Failed))) + .collect() +} + +/// Serializes the COMMIT half of an adapter update — the swap into the install +/// dir, the registry re-register, and the service/serving nudges — across the +/// per-adapter threads (releases#335). The fetch, verify, floor peek and +/// post-step stay parallel; that is where the time goes. The registry write is a +/// read-modify-write of one shared file with no lock of its own, so two threads +/// re-registering at once would each write back a registry missing the other's +/// record (REQ-HAZARD-INFO-RMW-LOST-UPDATE class); the daemon-coordinated apply +/// and the nudges talk to one shared daemon, and ordering them is cheaper than +/// proving every daemon arm reentrant. +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +static ADAPTER_COMMIT_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + /// The `adapter update` / `update adapters` engine (REQ-UPD-9, /// REQ-UPDATE-ADAPTERS-VERB): parse the comma-list, fail-fast on unknown names /// BEFORE any update, then per-adapter — failure-isolated, one summary line @@ -22323,25 +22434,28 @@ fn cmd_adapter_update(adapters: &std::path::Path, names: Option<&str>, core_vers return 0; } - let mut outcomes: Vec<(String, AdapterUpdateOutcome)> = Vec::new(); - for (record, manifest) in selected { + // ONE THREAD PER ADAPTER (releases#335): N adapters cost the slowest fetch + + // post-step, not the sum. Each thread's output is held and printed as one + // block when that adapter finishes; the summary below keeps selection order. + let names: Vec = selected.iter().map(|(r, _)| r.name.clone()).collect(); + let outcomes = fan_out_adapter_updates(&names, |i| { + let (record, manifest) = selected[i]; let gh_release = manifest .update .as_ref() .is_some_and(|u| u.avenue == UpdateAvenue::GhRelease); - let outcome = if gh_release { + if gh_release { update_one_adapter(adapters, record, manifest, core_version) } else { // Loud skip — a local-path/dev registration has nothing to pull. - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_SKIP:{}: local registration (no [update] avenue = \ \"gh_release\") — skipped", record.name ); AdapterUpdateOutcome::SkippedLocal - }; - outcomes.push((record.name.clone(), outcome)); - } + } + }); for (name, outcome) in &outcomes { eprintln!("{}", render_adapter_update_summary(name, outcome)); @@ -22372,7 +22486,7 @@ fn update_one_adapter( None => { // Validation requires `repo`; a record reaching here without one // is corrupt — never silently fetch from nowhere. - eprintln!("ADAPTER_UPDATE_FAIL:{}: gh_release missing repo", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: gh_release missing repo", record.name); return AdapterUpdateOutcome::Failed; } }; @@ -22386,7 +22500,7 @@ fn update_one_adapter( let latest = match gh_latest_release_version(repo, transport) { Ok(v) => v, Err(e) => { - eprintln!("ADAPTER_UPDATE_FAIL:{}: latest release: {e}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: latest release: {e}", record.name); return AdapterUpdateOutcome::Failed; } }; @@ -22413,7 +22527,7 @@ fn update_one_adapter( // 2. Version-compare: only a NEWER release pulls. A no-op falls THROUGH to // the post-step below — it never `continue`s (ADR-0029 unconditional). if version_is_newer(&latest, installed) { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE:{}: {installed} -> {latest} (fetching {asset} from {repo})", record.name ); @@ -22421,10 +22535,10 @@ fn update_one_adapter( // 3. Fetch the release .spt (reuse the REQ-INSTALL-9 primitive). The // fetched archive lands under adapters/_github/ as raw bytes we // can verify BEFORE it is extracted+registered. - let staged = match stage_gh_release_archive(adapters, repo, &latest, asset, transport) { + let staged = match stage_gh_release_archive(adapters, &record.name, repo, &latest, asset, transport) { Ok(s) => s, Err(e) => { - eprintln!("ADAPTER_UPDATE_FAIL:{}: fetch: {e}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: fetch: {e}", record.name); return AdapterUpdateOutcome::Failed; } }; @@ -22435,18 +22549,18 @@ fn update_one_adapter( // key, proceed on HTTPS + GitHub trust (the --release acquisition model). if let Some(key_hex) = update.signing_key.as_deref() { if let Err(e) = verify_staged_archive( - adapters, repo, &latest, asset, &staged, key_hex, transport, + &adapter_fetch_scratch(adapters, &record.name), repo, &latest, asset, &staged, key_hex, transport, ) { let _ = std::fs::remove_file(&staged); - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_REJECTED:{}: signature verification failed: {e}", record.name ); return AdapterUpdateOutcome::Failed; } - eprintln!("ADAPTER_UPDATE_VERIFIED:{}: signature OK", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_VERIFIED:{}: signature OK", record.name); } else { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_UNSIGNED:{}: no signing_key — trusting HTTPS + GitHub", record.name ); @@ -22465,7 +22579,7 @@ fn update_one_adapter( // [impl->REQ-UPDATE-REFUSAL-EXIT-DISTINCT] if let Err(reason) = staged_floor_ok(&staged, &dest, &record.name, core_version) { let _ = std::fs::remove_file(&staged); - eprintln!("ADAPTER_UPDATE_REFUSED:{}: {reason}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_REFUSED:{}: {reason}", record.name); return AdapterUpdateOutcome::Refused { reason }; } @@ -22476,7 +22590,7 @@ fn update_one_adapter( // the WHOLE update the moment one install-dir binary is still running. if let Err(e) = std::fs::create_dir_all(&dest) { let _ = std::fs::remove_file(&staged); - eprintln!("ADAPTER_UPDATE_FAIL:{}: create dest: {e}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: create dest: {e}", record.name); return AdapterUpdateOutcome::Failed; } // W3d routing: a live endpoint on this adapter may hold a resident @@ -22493,10 +22607,17 @@ fn update_one_adapter( // though: a daemon that is not running is supervising nothing, so // there is nothing to quiesce and the direct swap is correct. // [impl->REQ-RESIDENT-SERVICE] + // The commit half is serialized across the per-adapter threads + // (releases#335); held to the end of this apply arm, released before + // the post-step. + // [impl->REQ-ADAPTER-UPDATE-PARALLEL] + let _commit = ADAPTER_COMMIT_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let service_needs_daemon = effective_manifest.service.is_some() && spt_daemon::is_running(); let applied = if adapter_has_live_endpoint(&record.name) || service_needs_daemon { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_LIVE:{}: live endpoint(s) or a supervised service — \ daemon-coordinated apply", record.name @@ -22507,7 +22628,7 @@ fn update_one_adapter( }; let _ = std::fs::remove_file(&staged); if let Err(e) = applied { - eprintln!("ADAPTER_UPDATE_FAIL:{}: apply: {e}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: apply: {e}", record.name); return AdapterUpdateOutcome::Failed; } // THE SECOND COMPARISON SITE, and it is load-bearing @@ -22522,7 +22643,11 @@ fn update_one_adapter( // [impl->REQ-ADAPTER-FLOOR-VS-STAGED-CORE] match registry::register_with_core(adapters, &dest, now_ms(), core_version) { Ok((r, nm)) => { - eprintln!("ADAPTER_UPDATE_DONE:{} (re-registered)", r.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_DONE:{} (re-registered)", r.name); + // Before the nudges: a service started from an entry binary + // that arrived without its exec bit would fail to launch. + // [impl->REQ-ADAPTER-ENTRY-EXEC-BIT] + spt_runtime::entry_exec::force_entry_exec(&nm, &dest, &r.name); adapter_applied = true; new_version = latest.clone(); // A re-registration is declared intent that the manifest or @@ -22533,15 +22658,24 @@ fn update_one_adapter( effective_manifest = nm; } Err(e) => { - eprintln!("ADAPTER_UPDATE_FAIL:{}: re-register: {e}", record.name); + spt_proto::cap_eprintln!("ADAPTER_UPDATE_FAIL:{}: re-register: {e}", record.name); return AdapterUpdateOutcome::Failed; } } } else { - eprintln!( + spt_proto::cap_eprintln!( "ADAPTER_UPDATE_UPTODATE:{}: installed {installed}, latest {latest}", record.name ); + // No swap ran, and a content swap could never have healed a MODE-only + // difference anyway (releases#62): an entry binary that arrived without + // its exec bit heals here, on the next update run of any outcome. + // [impl->REQ-ADAPTER-ENTRY-EXEC-BIT] + spt_runtime::entry_exec::force_entry_exec( + manifest, + std::path::Path::new(&record.source_dir), + &record.name, + ); } // 6. [update.post] — the avenue-agnostic delegated post-step (ADR-0029), @@ -22657,10 +22791,17 @@ fn version_is_newer(a: &str, b: &str) -> bool { /// `_github` home (NOT yet extracted/registered), so they can be verified before /// trust. Returns the staged archive path. `transport` selects HTTPS vs the `gh` /// CLI (private-repo path). (REQ-UPD-9, REQ-ADAPTER-GH-TRANSPORT) +/// +/// Both the staged file and the `gh` download scratch are keyed by ADAPTER, not +/// by repo or asset alone (releases#335): the sweep runs one thread per adapter, +/// and two adapters sharing a repo — or both defaulting to `adapter.spt` — would +/// otherwise download into, verify, and delete one another's bytes. // [impl->REQ-UPD-9] // [impl->REQ-ADAPTER-GH-TRANSPORT] +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] fn stage_gh_release_archive( adapters: &std::path::Path, + adapter: &str, repo: &str, tag: &str, asset: &str, @@ -22670,12 +22811,20 @@ fn stage_gh_release_archive( let github = adapters.join("_github"); std::fs::create_dir_all(&github).map_err(|e| e.to_string())?; let tag_v = format!("v{tag}"); - let bytes = fetch_release_asset_bytes(repo, Some(&tag_v), asset, transport, &github)?; - let staged = github.join(format!("{safe}.update.spt")); + let scratch = adapter_fetch_scratch(adapters, adapter); + let bytes = fetch_release_asset_bytes(repo, Some(&tag_v), asset, transport, &scratch)?; + let staged = github.join(format!("{safe}.{adapter}.update.spt")); std::fs::write(&staged, &bytes).map_err(|e| e.to_string())?; Ok(staged) } +/// The per-adapter `gh release download` scratch dir under `_github` +/// (releases#335 — see [`stage_gh_release_archive`]). +// [impl->REQ-ADAPTER-UPDATE-PARALLEL] +fn adapter_fetch_scratch(adapters: &std::path::Path, adapter: &str) -> std::path::PathBuf { + adapters.join("_github").join(format!(".fetch-{adapter}")) +} + /// Fail-closed verification of a staged `.spt` against the adapter's declared /// Ed25519 `signing_key`: fetch the detached signature published beside the asset /// (`.sig`, lowercase-hex) and verify the staged bytes against it. The @@ -22686,7 +22835,7 @@ fn stage_gh_release_archive( // [impl->REQ-UPD-9] // [impl->REQ-ADAPTER-GH-TRANSPORT] fn verify_staged_archive( - adapters: &std::path::Path, + scratch: &std::path::Path, repo: &str, tag: &str, asset: &str, @@ -22697,9 +22846,8 @@ fn verify_staged_archive( let key = spt_daemon::parse_verifying_key(key_hex).map_err(|e| e.to_string())?; let bytes = std::fs::read(staged).map_err(|e| e.to_string())?; let tag_v = format!("v{tag}"); - let github = adapters.join("_github"); let sig_bytes = - fetch_release_asset_bytes(repo, Some(&tag_v), &format!("{asset}.sig"), transport, &github)?; + fetch_release_asset_bytes(repo, Some(&tag_v), &format!("{asset}.sig"), transport, scratch)?; let sig_hex = String::from_utf8(sig_bytes).map_err(|e| e.to_string())?; spt_daemon::verify_detached(&bytes, sig_hex.trim(), &key).map_err(|e| e.to_string()) } @@ -26242,6 +26390,142 @@ mod tests { ); } + /// releases#335: every adapter's work runs AT ONCE. Proven on state, not on a + /// wall-clock budget: each unit waits (bounded) until all three are inside + /// `work` together. A serial loop can never get there, so each unit would + /// time out and report `Failed`. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn adapter_fan_out_runs_every_adapter_concurrently() { + let names: Vec = ["a", "b", "c"].iter().map(|s| s.to_string()).collect(); + let inside = std::sync::Mutex::new(0usize); + let all_in = std::sync::Condvar::new(); + let outcomes = fan_out_adapter_updates_to( + &names, + |_| { + let mut n = inside.lock().unwrap(); + *n += 1; + all_in.notify_all(); + let (n, timeout) = all_in + .wait_timeout_while(n, std::time::Duration::from_secs(30), |n| *n < 3) + .unwrap(); + drop(n); + if timeout.timed_out() { + AdapterUpdateOutcome::Failed + } else { + AdapterUpdateOutcome::UpToDate { version: "1.0.0".to_string() } + } + }, + &mut |_| {}, + ); + for (name, outcome) in &outcomes { + assert!( + matches!(outcome, AdapterUpdateOutcome::UpToDate { .. }), + "{name}: {outcome:?} — the three units were never inside work together" + ); + } + } + + /// releases#335 OUTPUT ISOLATION: three adapters emit interleaved in time, + /// yet each arrives as ONE whole block holding only its own lines, blocks + /// arrive in FINISH order, and the outcomes come back in SELECTION order — + /// the order the summary lines print in. + /// + /// Deterministic, no sleeps: a barrier makes all three emit step `s` before + /// any emits `s + 1` (the interleave is forced, not hoped for), and each + /// unit may only RETURN on its turn, which the block sink advances — so + /// "fast" finishes first and "slow" last whatever the runner's load. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn adapter_fan_out_prints_each_adapter_as_one_whole_block() { + use spt_proto::emit::Stream; + let names: Vec = ["slow", "mid", "fast"].iter().map(|s| s.to_string()).collect(); + let step = std::sync::Barrier::new(3); + let turn = std::sync::Mutex::new(0usize); + let turned = std::sync::Condvar::new(); + let mut blocks: Vec> = Vec::new(); + let outcomes = fan_out_adapter_updates_to( + &names, + |i| { + let name = &names[i]; + for s in 0..3 { + step.wait(); + spt_proto::cap_eprintln!("STEP:{name}: {s}"); + } + spt_proto::cap_println!("notice from {name}"); + // Finish rank: fast (i=2) first, slow (i=0) last. + let rank = 2 - i; + let (_t, timeout) = turned + .wait_timeout_while( + turn.lock().unwrap(), + std::time::Duration::from_secs(30), + |t| *t != rank, + ) + .unwrap(); + if timeout.timed_out() { + return AdapterUpdateOutcome::Failed; + } + AdapterUpdateOutcome::UpToDate { version: format!("{i}.0.0") } + }, + &mut |chunks| { + blocks.push(chunks.to_vec()); + *turn.lock().unwrap() += 1; + turned.notify_all(); + }, + ); + + assert_eq!(blocks.len(), 3, "one block per adapter"); + for block in &blocks { + let owner = block[0].1.split(':').nth(1).unwrap_or_default().to_string(); + let expect: Vec<(Stream, String)> = (0..3) + .map(|s| (Stream::Err, format!("STEP:{owner}: {s}\n"))) + .chain(std::iter::once((Stream::Out, format!("notice from {owner}\n")))) + .collect(); + assert_eq!(block, &expect, "a block holds exactly its own adapter's lines, whole"); + } + let finish_order: Vec = blocks + .iter() + .map(|b| b[0].1.split(':').nth(1).unwrap_or_default().to_string()) + .collect(); + assert_eq!(finish_order, vec!["fast", "mid", "slow"], "printed as each finishes"); + let summary_order: Vec<&str> = outcomes.iter().map(|(n, _)| n.as_str()).collect(); + assert_eq!(summary_order, vec!["slow", "mid", "fast"], "outcomes keep selection order"); + assert_eq!( + outcomes[0].1, + AdapterUpdateOutcome::UpToDate { version: "0.0.0".to_string() }, + "each outcome is filed under its own adapter" + ); + } + + /// A panicking adapter thread is still ONE summary line: `Failed`, with its + /// reason in its own block; its siblings are untouched. + // [unit->REQ-ADAPTER-UPDATE-PARALLEL] + #[test] + fn adapter_fan_out_turns_a_panicking_thread_into_a_failed_outcome() { + let names: Vec = ["ok", "boom"].iter().map(|s| s.to_string()).collect(); + let mut blocks: Vec> = Vec::new(); + let outcomes = fan_out_adapter_updates_to( + &names, + |i| { + if i == 1 { + panic!("deliberate test panic"); + } + AdapterUpdateOutcome::UpToDate { version: "1.0.0".to_string() } + }, + &mut |chunks| blocks.push(chunks.to_vec()), + ); + assert_eq!(outcomes[1], ("boom".to_string(), AdapterUpdateOutcome::Failed)); + assert!(matches!(outcomes[0].1, AdapterUpdateOutcome::UpToDate { .. })); + assert!( + blocks + .iter() + .flatten() + .any(|(_, l)| l == "ADAPTER_UPDATE_FAIL:boom: update thread panicked\n"), + "the panic is reported in the adapter's own block: {blocks:?}" + ); + assert_eq!(adapter_update_exit(&outcomes), 1); + } + // [unit->REQ-UPDATE-ADAPTERS-VERB] isolation + aggregation: one failure // makes the sweep exit nonzero without hiding the others' outcomes; loud // local-path skips and up-to-dates are success. Summary lines are