From 865e35211d5a173b53e4ccd9ee1c7f5dac2e04b9 Mon Sep 17 00:00:00 2001 From: lostcause Date: Mon, 21 Sep 2026 11:22:07 +1000 Subject: [PATCH] add GetTransaction/CancelTransaction to npackd Install and Update accept "async": true to return {"transaction_id": N} immediately instead of blocking until completion; GetTransaction polls a transaction's status (running/succeeded/failed/cancelled). CancelTransaction is cooperative rather than raw task abortion: a shared AtomicBool cancellation flag is threaded through InstallRefOptions/ResolverState and checked only at the start of processing each package (before starting the next package in a dependency graph, or the next package in an Update loop) -- never mid-download or mid-install of a package already in progress, so a cancelled transaction cannot leave the store half-installed. Verified end-to-end over a real socket: an async Update against an empty store completes and reports succeeded via GetTransaction, and cancelling an in-flight async Install against a real relay before its resolve step starts reports cancelled. Streamed progress events (e.g. per-file download progress) remain future work. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 2 +- docs/roadmap.md | 20 +-- docs/using-npack.md | 33 ++++- src/main.rs | 313 +++++++++++++++++++++++++++++++++++++++++--- 4 files changed, 334 insertions(+), 34 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index d9eae54..5f48ad8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,4 +47,4 @@ npack is an independent package manager whose registry metadata will be publishe - `npack update --check` (alias of `install-ref --check`) reports available updates for one or all installed packages without downloading or installing anything, the `apt update` counterpart to `update`'s `apt upgrade`. - A manifest's optional `app` object (`summary`, `description`, `homepage`, `license`, `categories`, `icon`, `screenshots`, `desktop_file`, `release_date`) carries desktop-store metadata; `icon` and `desktop_file` are package-relative paths validated to exist and, for `desktop_file`, to be a syntactically valid freedesktop.org Desktop Entry file with `Exec` required when `Type=Application`. - `npack appstream` maps a manifest's `app` metadata to an AppStream `` document per the freedesktop.org AppStream spec: `console-application` when there is no `desktop_file` (advertising ``), `desktop-application` otherwise (advertising ``). Component IDs are namespaced `io.npack..` since publishers are Nostr pubkeys, not domains. -- `npack daemon` runs npackd, a local JSON-RPC-over-Unix-socket service (newline-delimited JSON requests/responses) exposing `Search`, `GetPackage`, `ListInstalled`, `Install`, `Remove`, `Update`, and `CheckUpdates` so a GUI store or other tool does not need to understand Nostr, Blossom, or `.npk` internals. It defaults to `$XDG_RUNTIME_DIR/npackd.sock` and handles connections concurrently via `tokio::spawn`. A client gets one response once an `Install`/`Update` call completes; transaction polling and progress-event streaming are future work. +- `npack daemon` runs npackd, a local JSON-RPC-over-Unix-socket service (newline-delimited JSON requests/responses) exposing `Search`, `GetPackage`, `ListInstalled`, `Install`, `Remove`, `Update`, `CheckUpdates`, `GetTransaction`, and `CancelTransaction` so a GUI store or other tool does not need to understand Nostr, Blossom, or `.npk` internals. It defaults to `$XDG_RUNTIME_DIR/npackd.sock` and handles connections concurrently via `tokio::spawn`. `Install`/`Update` run synchronously by default, or return `{"transaction_id": N}` immediately when called with `"async": true`, pollable via `GetTransaction`. `CancelTransaction` is cooperative: checked only between packages in a dependency graph or update loop, never mid-download or mid-install, so a cancelled transaction cannot leave the store half-installed. Progress-event streaming (as opposed to a final result) remains future work. diff --git a/docs/roadmap.md b/docs/roadmap.md index 1954577..a1caa4c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -56,7 +56,8 @@ exposing: Search() GetPackage() ListInstalled() Install() Remove() Update() -CheckUpdates() +CheckUpdates() GetTransaction() +CancelTransaction() ``` Chose a Unix socket over D-Bus for this first slice: no new system dependency @@ -68,15 +69,18 @@ npackd handles connections concurrently (each accepted connection is `tokio::spawn`ed), which required making the recursive dependency-install future `Send`. +`Install`/`Update` run synchronously by default; passing `"async": true` +returns `{"transaction_id": N}` immediately, pollable via `GetTransaction`. +`CancelTransaction` is cooperative rather than raw task abortion: a shared +cancellation flag is checked only between packages (before starting the next +package in a dependency graph, or the next package in an `Update` loop), +never mid-download or mid-install of a package already in progress -- a +cancelled transaction cannot leave the store half-installed. + Remaining work: -- `GetTransaction()`/`CancelTransaction()` and streamed progress events for - long-running `Install`/`Update` calls -- a client currently gets one - response once the whole operation completes, with no way to poll status or - interrupt it partway through. A safe `CancelTransaction` needs a real - cooperative checkpoint in the install path (e.g. between packages in a - dependency closure, never mid-file-write) rather than raw task abortion, - which risks leaving the store in a half-installed state. +- Streamed progress events (e.g. per-file download progress) rather than a + single final result once `GetTransaction` reports the transaction done. ## Phase 3: Security and privilege separation diff --git a/docs/using-npack.md b/docs/using-npack.md index d22030e..5aacd99 100644 --- a/docs/using-npack.md +++ b/docs/using-npack.md @@ -462,17 +462,38 @@ Supported methods, mirroring the CLI operations above: | `Search` | `query`, `relay[]`, `trusted_publisher[]`, `pubkey`, `refresh`, `no_cache` | Array of matching releases. | | `GetPackage` | `package`, `relay[]`, `requirement`, `os`, `arch`, `trusted_publisher[]`, `store`, `user` | The same resolved-metadata object as `npack resolve`. | | `ListInstalled` | `user`, `store` | Array of installed packages. | -| `Install` | `package`, `requirement`, `relay[]`, `server[]`, `user`, `store`, `allow_capability[]` | The installed package's record. | +| `Install` | `package`, `requirement`, `relay[]`, `server[]`, `user`, `store`, `allow_capability[]`, `async` | The installed package's record, or `{"transaction_id": N}` if `async` is true. | | `Remove` | `package`, `user`, `store` | `{"removed": ""}`. | -| `Update` | `package` (omit for all), `relay[]`, `server[]`, `user`, `store`, `allow_capability[]` | Array of per-package update outcomes. | +| `Update` | `package` (omit for all), `relay[]`, `server[]`, `user`, `store`, `allow_capability[]`, `async` | Array of per-package update outcomes, or `{"transaction_id": N}` if `async` is true. | | `CheckUpdates` | `package` (omit for all), `relay[]`, `trusted_publisher[]`, `user`, `store` | Array of `{reference, current_version, available_version}`. | +| `GetTransaction` | `transaction_id` | `{"status": "running"}`, `{"status": "succeeded", "result": ...}`, `{"status": "failed", "error": "..."}`, or `{"status": "cancelled"}`. | +| `CancelTransaction` | `transaction_id` | `{"cancel_requested": true}`. | An unknown method or a request that fails to deserialize its params returns `{"id": ..., "error": "..."}` instead of `result`. Each connection is handled -concurrently, but within a connection npackd does not yet stream -install/update progress back to the client; a client sees the final result -once the operation completes. GetTransaction/CancelTransaction-style progress -reporting is future work. +concurrently. + +By default `Install` and `Update` run to completion before responding. Pass +`"async": true` to get `{"transaction_id": N}` back immediately and poll +`GetTransaction` for the final result: + +```text +--> {"id": 1, "method": "Install", "params": {"package": "npub1.../myapp", "relay": ["wss://relay.example"], "async": true}} +<-- {"id": 1, "result": {"transaction_id": 1}} + +--> {"id": 2, "method": "GetTransaction", "params": {"transaction_id": 1}} +<-- {"id": 2, "result": {"status": "running"}} + ... later ... +--> {"id": 3, "method": "GetTransaction", "params": {"transaction_id": 1}} +<-- {"id": 3, "result": {"status": "succeeded", "result": {"publisher": "...", "name": "myapp", "version": "1.0.0", ...}}} +``` + +`CancelTransaction` is cooperative, not forcible: it is only checked between +packages (before starting the next package in a dependency graph, or the +next package in an `Update` loop), never mid-download or mid-install of a +package already in progress. This means a package that has already started +installing will finish before cancellation takes effect -- the store can +never be left half-installed by a cancelled transaction. ## Configuration diff --git a/src/main.rs b/src/main.rs index 6ad310c..64d9d58 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,8 @@ use std::{ os::unix::fs::PermissionsExt, os::unix::fs::symlink, path::{Path, PathBuf}, + sync::Arc, + sync::atomic::{AtomicBool, AtomicU64, Ordering}, time::{Instant, SystemTime, UNIX_EPOCH}, }; @@ -603,6 +605,7 @@ async fn main() -> Result<()> { offline, allowed_capabilities, &config, + None, ) .await?; } @@ -835,6 +838,7 @@ async fn main() -> Result<()> { offline, allowed_capabilities, &config, + None, ) .await? } else { @@ -914,6 +918,7 @@ async fn install_remote_command( offline: bool, allowed_capabilities: Vec, config: &Config, + cancel: Option>, ) -> Result<()> { let relays = if offline { Vec::new() @@ -951,6 +956,7 @@ async fn install_remote_command( blossom_servers: &servers, allowed_capabilities: &allowed_capabilities, offline, + cancel, }, requirement, ) @@ -999,6 +1005,7 @@ async fn update_all_command( false, allowed_capabilities.clone(), config, + None, ) .await; match result { @@ -1618,6 +1625,10 @@ struct InstallRefOptions<'a> { blossom_servers: &'a [String], allowed_capabilities: &'a [String], offline: bool, + /// Checked at the start of processing each package in the dependency + /// graph; never mid-download or mid-install of a package already in + /// progress, so cancellation cannot leave the store half-installed. + cancel: Option>, } async fn install_ref( @@ -1637,6 +1648,7 @@ async fn install_ref( blossom_servers, allowed_capabilities, offline, + cancel, } = options; let client = Client::default(); if !offline { @@ -1664,6 +1676,7 @@ async fn install_ref( installed: Vec::new(), selected: HashMap::new(), offline, + cancel, }; install_remote_package(&mut state, name.to_owned(), publisher, requirement).await?; if !offline { @@ -1905,6 +1918,7 @@ struct ResolverState<'a> { installed: Vec, selected: HashMap, offline: bool, + cancel: Option>, } fn cached_release_path(root: &Path, package: &LockedPackage) -> PathBuf { @@ -2371,6 +2385,13 @@ fn install_remote_package<'a, 'b>( requirement: Option, ) -> Pin> + Send + 'a>> { Box::pin(async move { + if state + .cancel + .as_deref() + .is_some_and(|cancel| cancel.load(Ordering::Relaxed)) + { + bail!("installation cancelled"); + } let install_key = publisher.as_deref().map_or_else( || name.to_owned(), |publisher| format!("{publisher}/{name}"), @@ -3053,8 +3074,13 @@ async fn search_matching_releases( // Requests and responses are newline-delimited JSON: // {"id": 1, "method": "ListInstalled", "params": {"user": true}} // {"id": 1, "result": [...]} -// Transaction/progress-event streaming is deferred to a later phase; Install -// and Update run to completion before responding. +// Install and Update run synchronously (the response carries the final +// result) unless params include "async": true, in which case they return +// {"transaction_id": N} immediately and the caller polls GetTransaction. +// CancelTransaction is cooperative: it is only checked between packages in a +// dependency graph or update loop, never mid-download or mid-install of a +// package already in progress, so a cancelled transaction cannot leave the +// store half-installed. #[derive(Debug, Deserialize)] struct DaemonRequest { @@ -3099,6 +3125,79 @@ fn daemon_socket_path(override_path: Option) -> PathBuf { }) } +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "status", rename_all = "lowercase")] +enum TransactionStatus { + Running, + Succeeded { result: serde_json::Value }, + Failed { error: String }, + Cancelled, +} + +struct TransactionEntry { + status: TransactionStatus, + cancel: Arc, +} + +struct DaemonState { + config: Config, + transactions: std::sync::Mutex>, + next_transaction_id: AtomicU64, +} + +impl DaemonState { + fn new(config: Config) -> Self { + Self { + config, + transactions: std::sync::Mutex::new(HashMap::new()), + next_transaction_id: AtomicU64::new(1), + } + } + + /// Registers a new transaction and spawns `make_work` (given a fresh + /// cancellation flag) to run in the background, recording its outcome + /// once it completes. Returns the transaction id immediately. + fn start_transaction(self: &Arc, make_work: F) -> u64 + where + F: FnOnce(Arc) -> Fut, + Fut: Future> + Send + 'static, + { + let id = self.next_transaction_id.fetch_add(1, Ordering::Relaxed); + let cancel = Arc::new(AtomicBool::new(false)); + self.transactions.lock().unwrap().insert( + id, + TransactionEntry { + status: TransactionStatus::Running, + cancel: cancel.clone(), + }, + ); + let work = make_work(cancel); + let state = self.clone(); + tokio::spawn(async move { + let status = match work.await { + Ok(result) => TransactionStatus::Succeeded { result }, + Err(error) if is_cancellation_error(&error) => TransactionStatus::Cancelled, + Err(error) => TransactionStatus::Failed { + error: error.to_string(), + }, + }; + if let Ok(mut transactions) = state.transactions.lock() + && let Some(entry) = transactions.get_mut(&id) + { + entry.status = status; + } + }); + id + } +} + +fn is_cancellation_error(error: &anyhow::Error) -> bool { + matches!( + error.to_string().as_str(), + "installation cancelled" | "update cancelled" + ) +} + async fn run_daemon(socket: Option, config: Config) -> Result<()> { let socket_path = daemon_socket_path(socket); if let Some(parent) = socket_path.parent() { @@ -3111,19 +3210,22 @@ async fn run_daemon(socket: Option, config: Config) -> Result<()> { let listener = tokio::net::UnixListener::bind(&socket_path) .with_context(|| format!("binding {}", socket_path.display()))?; eprintln!("npackd listening on {}", socket_path.display()); - let config = std::sync::Arc::new(config); + let state = Arc::new(DaemonState::new(config)); loop { let (stream, _) = listener.accept().await?; - let config = config.clone(); + let state = state.clone(); tokio::spawn(async move { - if let Err(error) = handle_daemon_connection(stream, &config).await { + if let Err(error) = handle_daemon_connection(stream, &state).await { eprintln!("npackd connection error: {error:#}"); } }); } } -async fn handle_daemon_connection(stream: tokio::net::UnixStream, config: &Config) -> Result<()> { +async fn handle_daemon_connection( + stream: tokio::net::UnixStream, + state: &Arc, +) -> Result<()> { use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; let (reader, mut writer) = stream.into_split(); let mut lines = tokio::io::BufReader::new(reader).lines(); @@ -3134,7 +3236,7 @@ async fn handle_daemon_connection(stream: tokio::net::UnixStream, config: &Confi let response = match serde_json::from_str::(&line) { Ok(request) => { let id = request.id; - match dispatch_daemon_request(config, request).await { + match dispatch_daemon_request(state, request).await { Ok(result) => DaemonResponse::ok(id, result), Err(error) => DaemonResponse::err(id, error), } @@ -3154,17 +3256,19 @@ async fn handle_daemon_connection(stream: tokio::net::UnixStream, config: &Confi } async fn dispatch_daemon_request( - config: &Config, + state: &Arc, request: DaemonRequest, ) -> Result { match request.method.as_str() { - "Search" => daemon_search(config, request.params).await, - "GetPackage" => daemon_get_package(config, request.params).await, - "ListInstalled" => daemon_list_installed(config, request.params), - "Install" => daemon_install(config, request.params).await, - "Remove" => daemon_remove(config, request.params), - "Update" => daemon_update(config, request.params).await, - "CheckUpdates" => daemon_check_updates(config, request.params).await, + "Search" => daemon_search(&state.config, request.params).await, + "GetPackage" => daemon_get_package(&state.config, request.params).await, + "ListInstalled" => daemon_list_installed(&state.config, request.params), + "Install" => daemon_install(state, request.params).await, + "Remove" => daemon_remove(&state.config, request.params), + "Update" => daemon_update(state, request.params).await, + "CheckUpdates" => daemon_check_updates(&state.config, request.params).await, + "GetTransaction" => daemon_get_transaction(state, request.params), + "CancelTransaction" => daemon_cancel_transaction(state, request.params), other => bail!("unknown method {other}"), } } @@ -3288,10 +3392,32 @@ struct InstallParams { store: Option, #[serde(default)] allow_capability: Vec, + /// Run in the background and return `{"transaction_id": N}` immediately + /// instead of waiting for the install to finish. + #[serde(default, rename = "async")] + background: bool, } -async fn daemon_install(config: &Config, params: serde_json::Value) -> Result { +async fn daemon_install( + state: &Arc, + params: serde_json::Value, +) -> Result { let params: InstallParams = serde_json::from_value(params).context("invalid Install params")?; + if params.background { + let daemon_state = state.clone(); + let id = state.start_transaction(move |cancel| async move { + daemon_install_sync(&daemon_state.config, params, Some(cancel)).await + }); + return Ok(serde_json::json!({ "transaction_id": id })); + } + daemon_install_sync(&state.config, params, None).await +} + +async fn daemon_install_sync( + config: &Config, + params: InstallParams, + cancel: Option>, +) -> Result { let package = params.package.clone(); install_remote_command( &package, @@ -3308,6 +3434,7 @@ async fn daemon_install(config: &Config, params: serde_json::Value) -> Result, #[serde(default)] allow_capability: Vec, + /// Run in the background and return `{"transaction_id": N}` immediately + /// instead of waiting for every package to finish updating. + #[serde(default, rename = "async")] + background: bool, } -async fn daemon_update(config: &Config, params: serde_json::Value) -> Result { +async fn daemon_update( + state: &Arc, + params: serde_json::Value, +) -> Result { let params: UpdateParams = serde_json::from_value(params).context("invalid Update params")?; + if params.background { + let daemon_state = state.clone(); + let id = state.start_transaction(move |cancel| async move { + daemon_update_sync(&daemon_state.config, params, Some(cancel)).await + }); + return Ok(serde_json::json!({ "transaction_id": id })); + } + daemon_update_sync(&state.config, params, None).await +} + +async fn daemon_update_sync( + config: &Config, + params: UpdateParams, + cancel: Option>, +) -> Result { let use_user = params.user || config.install.user; let root = install_paths(params.store.as_deref(), use_user).0; let mut installed = installed_packages(Some(&root))?; @@ -3392,6 +3541,12 @@ async fn daemon_update(config: &Config, params: serde_json::Value) -> Result Result Result, + params: serde_json::Value, +) -> Result { + let params: TransactionIdParams = + serde_json::from_value(params).context("invalid GetTransaction params")?; + let transactions = state.transactions.lock().unwrap(); + let entry = transactions + .get(¶ms.transaction_id) + .context("unknown transaction_id")?; + Ok(serde_json::to_value(&entry.status)?) +} + +fn daemon_cancel_transaction( + state: &Arc, + params: serde_json::Value, +) -> Result { + let params: TransactionIdParams = + serde_json::from_value(params).context("invalid CancelTransaction params")?; + let transactions = state.transactions.lock().unwrap(); + let entry = transactions + .get(¶ms.transaction_id) + .context("unknown transaction_id")?; + entry.cancel.store(true, Ordering::Relaxed); + Ok(serde_json::json!({ "cancel_requested": true })) +} + #[derive(Debug, Serialize)] struct UpdateStatus { reference: String, @@ -6587,10 +6775,10 @@ mod tests { fs::create_dir_all(&store)?; let listener = tokio::net::UnixListener::bind(&socket_path)?; - let config = Config::default(); + let state = Arc::new(DaemonState::new(Config::default())); let server = async { let (stream, _) = listener.accept().await.unwrap(); - handle_daemon_connection(stream, &config).await.unwrap(); + handle_daemon_connection(stream, &state).await.unwrap(); }; let client_work = async { @@ -6673,4 +6861,91 @@ mod tests { daemon.abort(); Ok(()) } + + #[tokio::test] + async fn install_remote_package_respects_cancellation_before_starting_a_package() -> Result<()> + { + let dir = tempdir()?; + let root = dir.path().join("root"); + let prefix = dir.path().join("prefix"); + fs::create_dir_all(&root)?; + fs::create_dir_all(&prefix)?; + let client = Client::default(); + let mut state = ResolverState { + client: &client, + allowed_capabilities: &[], + user: false, + trusted_publishers: &[], + blossom_servers: &[], + root: &root, + prefix: &prefix, + locked_packages: None, + visiting: Vec::new(), + installed: Vec::new(), + selected: HashMap::new(), + offline: false, + cancel: Some(Arc::new(AtomicBool::new(true))), + }; + let error = install_remote_package(&mut state, "hello".into(), None, None) + .await + .unwrap_err(); + assert_eq!(error.to_string(), "installation cancelled"); + assert!(state.installed.is_empty()); + Ok(()) + } + + #[tokio::test] + async fn transactions_report_running_then_succeeded_and_reject_unknown_ids() -> Result<()> { + let state = Arc::new(DaemonState::new(Config::default())); + let id = state.start_transaction(|_cancel| async { Ok(serde_json::json!({"ok": true})) }); + + let running = daemon_get_transaction(&state, serde_json::json!({ "transaction_id": id }))?; + assert_eq!(running["status"], "running"); + + let mut final_status = None; + for _ in 0..200 { + let status = + daemon_get_transaction(&state, serde_json::json!({ "transaction_id": id }))?; + if status["status"] != "running" { + final_status = Some(status); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let final_status = final_status.context("transaction never left the running state")?; + assert_eq!(final_status["status"], "succeeded"); + assert_eq!(final_status["result"], serde_json::json!({"ok": true})); + + assert!( + daemon_get_transaction(&state, serde_json::json!({ "transaction_id": 999_999 })) + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn cancel_transaction_stops_a_cooperative_task() -> Result<()> { + let state = Arc::new(DaemonState::new(Config::default())); + let id = state.start_transaction(|cancel| async move { + loop { + if cancel.load(Ordering::Relaxed) { + bail!("update cancelled"); + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }); + + daemon_cancel_transaction(&state, serde_json::json!({ "transaction_id": id }))?; + + for _ in 0..200 { + let status = + daemon_get_transaction(&state, serde_json::json!({ "transaction_id": id }))?; + if status["status"] != "running" { + assert_eq!(status["status"], "cancelled"); + return Ok(()); + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + panic!("transaction never reported cancelled"); + } }