From 310d55eb3713e3a23b3fa041715422fdc1e056f7 Mon Sep 17 00:00:00 2001 From: Josh Knapp Date: Sun, 27 Sep 2026 09:30:58 -0700 Subject: [PATCH] Marketplace: sync projects into their containers on start Waits for the entrypoint, uploads the payload and the sync script, runs it as claude and stores its report (payload skips merged in). The start hook spawns the sync in the background; a per-project lock serialises syncs of one project (pre-flight F11b). Co-Authored-By: Claude Opus 5.5 --- .../src/commands/project_commands.rs | 10 + app/src-tauri/src/docker/exec.rs | 10 +- app/src-tauri/src/marketplace/mod.rs | 181 ++++++++++++- app/src-tauri/src/marketplace/sync.rs | 247 ++++++++++++++++++ 4 files changed, 442 insertions(+), 6 deletions(-) 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); + } +}