diff --git a/app/src-tauri/src/commands/project_commands.rs b/app/src-tauri/src/commands/project_commands.rs index e82d011..e00e4a2 100644 --- a/app/src-tauri/src/commands/project_commands.rs +++ b/app/src-tauri/src/commands/project_commands.rs @@ -1449,6 +1449,16 @@ async fn start_project_container_locked( log::warn!("Failed to sync AWS credentials for project {}: {}", project.id, e); } + // Marketplace items sync in the background — see `spawn_project_sync` + // for why the start never waits on it or fails because of it. + crate::marketplace::spawn_project_sync( + app_handle.clone(), + state.marketplace.clone(), + state.settings_store.get(), + project.clone(), + container_id.clone(), + ); + Ok(container_id) }.await; diff --git a/app/src-tauri/src/docker/exec.rs b/app/src-tauri/src/docker/exec.rs index 7804d92..a285b4b 100644 --- a/app/src-tauri/src/docker/exec.rs +++ b/app/src-tauri/src/docker/exec.rs @@ -434,7 +434,8 @@ fn container_join(dir: &str, name: &str) -> String { /// Write `data` into the container at `/` with `mode`. /// /// For small, generated files — migration uses it for the `tar -T` include -/// list, which can be too long to pass as argv. Anything large should be +/// list, which can be too long to pass as argv, and the marketplace sync for +/// its payload tar and script. Anything large should be /// streamed through an attached exec's stdin instead, since this buffers the /// whole payload in memory twice (once raw, once tarred). pub async fn upload_bytes_to_container( @@ -446,9 +447,10 @@ pub async fn upload_bytes_to_container( ) -> Result { let docker = get_docker()?; - // Root-owned on purpose: the only caller is migration, whose `tar -T` list - // is read back as root. The mtime still gets stamped so the file doesn't - // read as 1970. + // Root-owned on purpose: migration's `tar -T` list is read back as root, + // and the marketplace sync uploads into a `claude`-owned directory it + // prepares first, so `claude` can still read and delete the files. The + // mtime still gets stamped so the file doesn't read as 1970. let tar_buf = build_single_file_tar(file_name, data, mode, 0, 0, now_epoch_secs())?; docker diff --git a/app/src-tauri/src/marketplace/mod.rs b/app/src-tauri/src/marketplace/mod.rs index 22d46be..30eae88 100644 --- a/app/src-tauri/src/marketplace/mod.rs +++ b/app/src-tauri/src/marketplace/mod.rs @@ -16,12 +16,13 @@ pub(crate) mod test_support; use std::collections::{BTreeSet, HashMap}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; +use tauri::Emitter; use tokio::sync::oneshot; use crate::models::marketplace::{ - CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport, + effective_installs, CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport, }; use crate::models::{AppSettings, Project}; use catalog::{item_fingerprint, parse_catalog}; @@ -38,6 +39,9 @@ pub struct MarketplaceManager { /// Serialises writers of the bare caches (fetch, pins, cache removal) so /// concurrent refreshes never race on gix ref locks (pre-flight F11a). repo_lock: tokio::sync::Mutex<()>, + /// One lock per project, held for a whole `sync_project`, so a start sync + /// and Apply now never run `sync.sh` in one container at once (F11b). + sync_locks: Mutex>>>, } /// Project ids become file names; anything outside this set is not persisted. @@ -58,6 +62,7 @@ impl MarketplaceManager { reports: Mutex::new(HashMap::new()), gh_login_cancel: tokio::sync::Mutex::new(None), repo_lock: tokio::sync::Mutex::new(()), + sync_locks: Mutex::new(HashMap::new()), } } @@ -71,6 +76,16 @@ impl MarketplaceManager { &self.repo_lock } + /// The project's sync lock; see `sync_project`. + pub fn sync_lock(&self, project_id: &str) -> Arc> { + self.sync_locks + .lock() + .unwrap() + .entry(project_id.to_string()) + .or_default() + .clone() + } + pub fn snapshot(&self, marketplace_id: &str) -> Option { self.snapshots.lock().unwrap().get(marketplace_id).cloned() } @@ -329,6 +344,93 @@ pub fn compute_updates( out } +fn project_installs(settings: &AppSettings, project: &Project) -> Vec { + effective_installs( + &settings.global_marketplace_installs, + &project.marketplace_disabled, + &project.marketplace_installs, + ) +} + +/// Build the project's payload and sync it into its running container. The +/// report is stored (and persisted) whatever happens. Holds the project's sync +/// lock throughout, so concurrent syncs of one project run one after another. +pub async fn sync_project( + mgr: &MarketplaceManager, + settings: &AppSettings, + project: &Project, + container_id: &str, +) -> SyncReport { + let lock = mgr.sync_lock(&project.id); + let _sync_guard = lock.lock().await; + + let installs = project_installs(settings, project); + let marketplaces = settings.marketplaces.clone(); + let root = mgr.data_root().to_path_buf(); + let built = tokio::task::spawn_blocking(move || { + payload::build_payload(&payload::PayloadInput { + installs: &installs, + marketplaces: &marketplaces, + data_root: &root, + }) + }) + .await + .map_err(|e| format!("Building the marketplace payload failed: {e}")) + .and_then(|r| r); + + let report = match built { + Ok(p) => { + let items = p.manifest["items"].as_array().map_or(0, Vec::len); + log::debug!( + "Marketplace sync for project {}: {} item(s) in the payload, {} skipped on the host", + project.id, + items, + p.skipped.len() + ); + let result = sync::sync_container(container_id, &p).await; + sync::with_payload_skips(sync::report_from_result(result), &p.skipped) + } + Err(e) => sync::report_from_result(Err(e)), + }; + mgr.put_report(&project.id, report.clone()); + report +} + +/// A project with no items that has never been synced has nothing to add and +/// nothing to remove, so its start does not wait on a sync at all. +pub fn should_sync(mgr: &MarketplaceManager, settings: &AppSettings, project: &Project) -> bool { + !project_installs(settings, project).is_empty() || mgr.report(&project.id).is_some() +} + +/// Sync in the background after a container start. The sync waits for the +/// entrypoint to finish (which can include a two-minute `claude update`), and +/// its failure must never fail the start — so the start never awaits it. +pub fn spawn_project_sync( + app: tauri::AppHandle, + mgr: Arc, + settings: AppSettings, + project: Project, + container_id: String, +) { + if !should_sync(&mgr, &settings, &project) { + return; + } + tauri::async_runtime::spawn(async move { + let report = sync_project(&mgr, &settings, &project, &container_id).await; + if !report.errors.is_empty() { + log::warn!( + "Marketplace sync for project {} reported errors: {:?}", + project.id, + report.errors + ); + } + let _ = app.emit( + SYNC_FINISHED_EVENT, + serde_json::json!({ "project_id": project.id, "report": report }), + ); + }); +} + /// All commits referenced by installs, per marketplace (for `git::set_pins`). pub fn pins_by_marketplace( settings: &AppSettings, @@ -560,4 +662,79 @@ mod tests { "slot is free after cancel" ); } + + #[test] + fn a_project_that_never_had_items_is_not_synced() { + let data = tempfile::tempdir().unwrap(); + let mgr = MarketplaceManager::new(data.path().to_path_buf()); + let settings = settings_with("https://example.invalid/r.git"); + let project = crate::models::Project::new("p".into(), vec![]); + assert!(!should_sync(&mgr, &settings, &project)); + } + + #[test] + fn a_project_with_items_or_a_previous_sync_is_synced() { + let data = tempfile::tempdir().unwrap(); + let mgr = MarketplaceManager::new(data.path().to_path_buf()); + let mut settings = settings_with("https://example.invalid/r.git"); + let project = crate::models::Project::new("p".into(), vec![]); + settings.global_marketplace_installs = vec![install(ItemKind::Agent, "a", &"a".repeat(40))]; + assert!(should_sync(&mgr, &settings, &project), "global items apply"); + + // Everything was uninstalled since the last sync: the container still + // holds the old files, so it must be synced to remove them. + settings.global_marketplace_installs.clear(); + mgr.put_report(&project.id, SyncReport::default()); + assert!(should_sync(&mgr, &settings, &project)); + } + + #[test] + fn a_project_whose_only_item_is_disabled_is_not_synced() { + let data = tempfile::tempdir().unwrap(); + let mgr = MarketplaceManager::new(data.path().to_path_buf()); + let mut settings = settings_with("https://example.invalid/r.git"); + let inst = install(ItemKind::Agent, "a", &"a".repeat(40)); + let mut project = crate::models::Project::new("p".into(), vec![]); + project.marketplace_disabled = vec![inst.item_ref()]; + settings.global_marketplace_installs = vec![inst]; + assert!(!should_sync(&mgr, &settings, &project)); + } + + #[test] + fn sync_locks_are_per_project() { + let mgr = MarketplaceManager::new(std::env::temp_dir()); + let a1 = mgr.sync_lock("a"); + let a2 = mgr.sync_lock("a"); + let b = mgr.sync_lock("b"); + assert!(Arc::ptr_eq(&a1, &a2), "one lock per project"); + assert!(!Arc::ptr_eq(&a1, &b), "projects do not block each other"); + } + + #[tokio::test] + async fn sync_project_waits_for_the_projects_sync_lock() { + // Pre-flight F11b: a start sync and Apply now must never run sync.sh + // in the same container at once. + let data = tempfile::tempdir().unwrap(); + let mgr = MarketplaceManager::new(data.path().to_path_buf()); + let settings = settings_with("https://example.invalid/r.git"); + let project = crate::models::Project::new("p".into(), vec![]); + + let lock = mgr.sync_lock(&project.id); + let guard = lock.lock().await; + let blocked = tokio::time::timeout( + std::time::Duration::from_millis(300), + sync_project(&mgr, &settings, &project, "no-such-container"), + ) + .await; + assert!(blocked.is_err(), "sync must wait while another sync holds the lock"); + assert_eq!(mgr.report(&project.id), None, "nothing ran while blocked"); + drop(guard); + + // No Docker (or no such container) here: the failure becomes a stored + // report instead of an error. + let report = sync_project(&mgr, &settings, &project, "no-such-container").await; + assert_eq!(report.errors.len(), 1, "{report:?}"); + assert!(!report.finished_at.is_empty()); + assert_eq!(mgr.report(&project.id), Some(report)); + } } diff --git a/app/src-tauri/src/marketplace/sync.rs b/app/src-tauri/src/marketplace/sync.rs index 699faac..7794c8c 100644 --- a/app/src-tauri/src/marketplace/sync.rs +++ b/app/src-tauri/src/marketplace/sync.rs @@ -1,9 +1,256 @@ //! Pushes a project's marketplace payload into its container and runs the //! sync script there (spec §4). +use std::time::Duration; + +use super::payload::Payload; +use crate::docker::exec::{exec_oneshot_as, exec_oneshot_streams_as, upload_bytes_to_container}; +use crate::models::marketplace::{SkippedItem, SyncReport}; + /// Where the payload and the script are uploaded. Owned by `claude`. pub const INCOMING_DIR: &str = "/home/claude/.claude/triple-c/marketplace/incoming"; /// The sync script. Shipped with the app and uploaded on every sync, so a new /// app version reaches existing containers without an image migration. pub const SYNC_SCRIPT: &str = include_str!("sync.sh"); + +/// True once the entrypoint has finished: its last step execs this exact +/// command line. Before that it may still be merging `settings.json` or running +/// `claude update`, both of which the sync would race. +const READY_PROBE: &str = "pgrep -x -f 'su -s /bin/bash claude -c exec sleep infinity' >/dev/null"; +const READY_TIMEOUT: Duration = Duration::from_secs(180); +const READY_POLL: Duration = Duration::from_secs(2); + +/// Run as root: `~/.claude` is a volume and `triple-c/` may not exist yet, and +/// the uploads below are root-owned files in a directory `claude` must own so +/// the script can delete them. +const PREPARE_SCRIPT: &str = r#"set -e +d=/home/claude/.claude/triple-c/marketplace/incoming +mkdir -p "$d" +chown -R claude:claude /home/claude/.claude/triple-c +rm -f "$d/payload.tar" "$d/sync.sh""#; + +fn sh(script: &str) -> Vec { + vec!["sh".to_string(), "-c".to_string(), script.to_string()] +} + +/// The readiness probe, run as root. +fn ready_probe_cmd() -> Vec { + sh(READY_PROBE) +} + +/// The sync script invocation, run as `claude`. +fn run_script_cmd() -> Vec { + vec!["sh".to_string(), format!("{INCOMING_DIR}/sync.sh")] +} + +fn run_script_env() -> Vec { + vec!["HOME=/home/claude".to_string()] +} + +async fn wait_until_ready(container_id: &str) -> Result<(), String> { + let deadline = tokio::time::Instant::now() + READY_TIMEOUT; + loop { + let (_, code) = exec_oneshot_as(container_id, "root", ready_probe_cmd(), vec![]).await?; + if code == 0 { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "The container did not finish starting within {} seconds, so marketplace items \ + were not applied. They are applied on the next start, or with Apply now.", + READY_TIMEOUT.as_secs() + )); + } + tokio::time::sleep(READY_POLL).await; + } +} + +/// The last `max` bytes of `text`, trimmed, never splitting a character. +fn tail(text: &str, max: usize) -> &str { + let text = text.trim(); + if text.len() <= max { + return text; + } + let mut start = text.len() - max; + while !text.is_char_boundary(start) { + start += 1; + } + &text[start..] +} + +/// Wait for readiness, upload the payload and the script, run the script as +/// `claude`, and return its report. +pub async fn sync_container(container_id: &str, payload: &Payload) -> Result { + wait_until_ready(container_id).await?; + + let (out, code) = exec_oneshot_as(container_id, "root", sh(PREPARE_SCRIPT), vec![]).await?; + if code != 0 { + return Err(format!( + "Could not prepare the container for the marketplace sync: {}", + tail(&out, 500) + )); + } + upload_bytes_to_container( + container_id, + INCOMING_DIR, + "payload.tar", + &payload.tar, + 0o644, + ) + .await?; + upload_bytes_to_container( + container_id, + INCOMING_DIR, + "sync.sh", + SYNC_SCRIPT.as_bytes(), + 0o755, + ) + .await?; + + let (stdout, stderr, code) = + exec_oneshot_streams_as(container_id, "claude", run_script_cmd(), run_script_env()).await?; + parse_report(&stdout).map_err(|e| { + format!( + "The marketplace sync script failed (exit {code}): {e}. {}", + tail(&stderr, 500) + ) + }) +} + +/// The script's report is the last non-empty line of stdout. +pub fn parse_report(stdout: &str) -> Result { + let line = stdout + .lines() + .rev() + .map(str::trim) + .find(|l| !l.is_empty()) + .ok_or_else(|| "the sync script printed no report".to_string())?; + serde_json::from_str(line) + .map_err(|e| format!("the sync script's report could not be read: {e}")) +} + +/// A sync never fails its caller: an error becomes a report that says so. +pub fn report_from_result(r: Result) -> SyncReport { + let mut report = match r { + Ok(report) => report, + Err(e) => SyncReport { + errors: vec![e], + ..Default::default() + }, + }; + report.finished_at = chrono::Utc::now().to_rfc3339(); + report +} + +/// Items the host left out of the payload (invalid, missing from the cache, …) +/// never reach the script, so the stored report lists them ahead of its own. +pub fn with_payload_skips(mut report: SyncReport, payload_skipped: &[SkippedItem]) -> SyncReport { + let mut skipped = payload_skipped.to_vec(); + skipped.append(&mut report.skipped); + report.skipped = skipped; + report +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_report_is_the_last_non_empty_stdout_line() { + let out = "noise\n{\"installed\":[\"agent:a\"],\"errors\":[]}\n\n"; + let r = parse_report(out).unwrap(); + assert_eq!(r.installed, vec!["agent:a"]); + assert!(r.skipped.is_empty()); + } + + #[test] + fn missing_or_garbled_reports_are_errors() { + assert!(parse_report("").unwrap_err().contains("no report")); + assert!(parse_report("not json\n") + .unwrap_err() + .contains("could not be read")); + } + + #[test] + fn a_failed_sync_becomes_a_report() { + // A failed sync becomes a report with the error in it — never an Err + // that could propagate into container start. + let r = report_from_result(Err("container went away".into())); + assert_eq!(r.errors, vec!["container went away"]); + assert!(!r.finished_at.is_empty()); + + let ok = report_from_result(Ok(SyncReport { + installed: vec!["hook:h".into()], + ..Default::default() + })); + assert_eq!(ok.installed, vec!["hook:h"]); + assert!(chrono::DateTime::parse_from_rfc3339(&ok.finished_at).is_ok()); + } + + #[test] + fn the_embedded_script_is_the_sync_script() { + assert!(SYNC_SCRIPT.starts_with("#!/bin/sh")); + assert!(SYNC_SCRIPT.contains("MARKETPLACE_INCOMING")); + } + + #[test] + fn readiness_probes_the_entrypoints_final_exec() { + assert_eq!( + ready_probe_cmd(), + vec![ + "sh", + "-c", + "pgrep -x -f 'su -s /bin/bash claude -c exec sleep infinity' >/dev/null" + ] + ); + assert_eq!(READY_POLL, Duration::from_secs(2)); + assert_eq!(READY_TIMEOUT, Duration::from_secs(180)); + } + + #[test] + fn the_incoming_dir_is_prepared_for_claude() { + // The uploads are root-owned, so the directory must exist and belong + // to claude before they land (claude extracts and deletes them). + assert!(PREPARE_SCRIPT.contains(INCOMING_DIR)); + assert!(PREPARE_SCRIPT.contains("mkdir -p")); + assert!(PREPARE_SCRIPT.contains("chown -R claude:claude /home/claude/.claude/triple-c")); + } + + #[test] + fn the_script_runs_as_claude_with_home_set() { + assert_eq!( + run_script_cmd(), + vec!["sh".to_string(), format!("{INCOMING_DIR}/sync.sh")] + ); + assert_eq!(run_script_env(), vec!["HOME=/home/claude"]); + } + + #[test] + fn payload_skips_come_before_the_scripts_own() { + use crate::models::marketplace::SkippedItem; + let payload_skip = SkippedItem { + item: "agent:a".into(), + reason: "invalid".into(), + }; + let script_skip = SkippedItem { + item: "hook:h".into(), + reason: "no jq".into(), + }; + let report = SyncReport { + skipped: vec![script_skip.clone()], + ..Default::default() + }; + let merged = with_payload_skips(report, &[payload_skip.clone()]); + assert_eq!(merged.skipped, vec![payload_skip, script_skip]); + } + + #[test] + fn long_output_is_tailed_on_a_char_boundary() { + assert_eq!(tail(" short \n", 10), "short"); + let s = format!("{}é", "x".repeat(20)); + let t = tail(&s, 1); + assert!(s.ends_with(t)); + assert!(t.len() <= 2); + } +}