diff --git a/CLAUDE.md b/CLAUDE.md index 0086c02d..94a1ab6f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -81,11 +81,13 @@ ant-core/src/ │ ├── link.rs # ant://
and ant://manifest/ links │ ├── file.rs # .ant file read/write │ ├── history.rs # Upload history: one .ant per upload under /uploads +│ ├── journal.rs # Crash-safe journal of a running build (.ant.partial) │ ├── wire.rs # Manifest-owned DataMap layout; pre-decode structure check │ ├── build.rs # ManifestBuilder (uploads files, embedded or compact refs) │ ├── embed.rs # Root-vs-shrunk DataMap choice for embedding │ ├── compact.rs # Plan/publish/apply compaction (embedded DataMaps -> addresses) -│ └── extract.rs # Containment-safe extraction with per-entry results +│ ├── extract.rs # Containment-safe extraction with per-entry results +│ └── output/ # Handle-relative output directories (fds on Unix, held handles on Windows) └── node/ # Node management ├── mod.rs # add_nodes, remove_node, reset ├── types.rs # DaemonConfig, DaemonStatus, NodeConfig, NodeInfo, AddNodeOpts, etc. @@ -150,7 +152,7 @@ cargo run --bin ant -- --help # Run the CLI - **CLI output**: All commands support `--json` global flag. Human-readable output by default, structured JSON when `--json` is passed. - **Hidden commands**: `ant node daemon run` is hidden (runs daemon in foreground, used internally by `daemon start`). - **REST API conflict responses**: All 409 responses include a `current_state` field so retrying clients can confirm desired state already exists. -- **Platform paths**: Use `ant_core::config::data_dir()` and `ant_core::config::log_dir()` for platform-appropriate paths. Never hardcode paths. +- **Platform paths**: Use `ant_core::config::data_dir()`, `ant_core::config::log_dir()` and `ant_core::config::download_dir()` for platform-appropriate paths. Never hardcode paths. - **OpenAPI schema**: Types exposed in the REST API derive `utoipa::ToSchema`. Use `#[schema(value_type = String)]` for `PathBuf` fields. - **Tests**: Unit tests go in `#[cfg(test)] mod tests` inside the source file. Integration tests go in `ant-core/tests/`. - **Progress reporting**: Long-running operations (e.g., binary downloads) accept a `&dyn ProgressReporter` from `ant_core::node::binary`. CLI provides a terminal-printing implementation; use `NoopProgress` in tests and daemon API handlers. @@ -161,8 +163,13 @@ cargo run --bin ant -- --help # Run the CLI is shared off-network; only the files it references live on Autonomi. `ant://` is always a plain file link. Every upload (`ant file upload`, `ant manifest create`) is also recorded as a manifest under `/uploads/`; `ant manifest list` shows them and `show`/`link`/`export`/`download` - accept a record id. Entry paths obey the portable rules in `data/manifest/path.rs` at build, - decode and extract time, and extraction never follows symlinks inside the output directory. + accept a record id. Without `--output`, `download` extracts into a new directory named after + the manifest in the user's downloads directory. Entry paths obey the portable rules in `data/manifest/path.rs` at build, + decode and extract time, and extraction never follows symlinks inside the output directory + and resolves every name from directories it holds open (`data/manifest/output/`: by + descriptor on Unix, by path under held handles on Windows). A manifest + build journals each completed upload to `.ant.partial` before paying for the next, so a + crash loses no paid upload; journals list as unfinished records. See `docs/adr/ADR-0006-file-manifests-and-links.md`. - **Log forwarding is opt-in and node-logging-dependent**: `ant node logs forward enable` is the consent act. It only forwards nodes whose `NodeConfig.log_dir` is `Some` — node file logging is diff --git a/Cargo.lock b/Cargo.lock index cca5c7c6..166fa5dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -840,6 +840,7 @@ dependencies = [ "rmp-serde", "serde", "serde_json", + "sha2", "tempfile", "tokio", "tokio-util", @@ -863,6 +864,8 @@ dependencies = [ "blake3", "bytes", "console_error_panic_hook", + "crc32fast", + "dirs", "fips204", "flate2", "fs2", @@ -880,6 +883,7 @@ dependencies = [ "rand 0.8.6", "reqwest 0.12.28", "rmp-serde", + "rustix", "saorsa-transport", "self-replace", "self_encryption", @@ -901,6 +905,7 @@ dependencies = [ "tower-http", "tracing", "tracing-subscriber", + "unicase", "unicode-normalization", "url", "utoipa", diff --git a/ant-cli/Cargo.toml b/ant-cli/Cargo.toml index 78e2b809..86a2bcb6 100644 --- a/ant-cli/Cargo.toml +++ b/ant-cli/Cargo.toml @@ -23,6 +23,7 @@ rand = "0.8" rmp-serde = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" +sha2 = "0.10" tokio = { version = "1", features = ["full"] } tokio-util = "0.7" tracing = "0.1" diff --git a/ant-cli/src/commands/data/file.rs b/ant-cli/src/commands/data/file.rs index bd1aab8a..056e825d 100644 --- a/ant-cli/src/commands/data/file.rs +++ b/ant-cli/src/commands/data/file.rs @@ -801,7 +801,7 @@ struct UploadJsonResult { /// Tell the user where the upload was recorded. fn print_recorded(manifest_id: Option<&str>) { if let Some(id) = manifest_id { - println!("Recorded as {id} (ant manifest download {id})"); + println!("Recorded as {id} (ant manifest download {id} --output )"); } } diff --git a/ant-cli/src/commands/data/manifest.rs b/ant-cli/src/commands/data/manifest.rs index 1d103d57..f3452ea4 100644 --- a/ant-cli/src/commands/data/manifest.rs +++ b/ant-cli/src/commands/data/manifest.rs @@ -17,13 +17,15 @@ use ant_core::data::{ apply_compaction, data_map_address, default_history_dir, embeddable_data_map, extract_manifest, format_timestamp, is_link, list_uploads, load_upload, manifest_filename_for, manifest_link_bytes, manifest_link_from_bytes, parse_link, plan_compaction, publish_data_maps, - read_manifest_file, record_upload, write_manifest_file, BuildEvent, BuildOptions, BuildResult, - Client, ContentRef, DownloadEvent, EntryStatus, ExtractEvent, ExtractOptions, Link, Manifest, - ManifestBuilder, ManifestError, PaymentMode, ReferenceMode, TorrentReference, UploadEvent, - UploadRecord, Visibility, MANIFEST_LINK_RECOMMENDED_MAX_BYTES, + read_manifest_file, record_upload, select_entries, write_manifest_file, BuildEvent, + BuildOptions, BuildResult, Client, ContentRef, DownloadEvent, EntryStatus, ExtractEvent, + ExtractOptions, Link, Manifest, ManifestBuilder, ManifestError, PaymentMode, RecordState, + ReferenceMode, TorrentReference, UploadEvent, UploadRecord, Visibility, + MANIFEST_LINK_RECOMMENDED_MAX_BYTES, }; use clap::Subcommand; use serde_json::json; +use sha2::{Digest, Sha256}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::info; @@ -42,6 +44,12 @@ const YES_ANSWERS: &[&str] = &["y", "yes"]; const FORCE_QUIT_EXIT_CODE: i32 = 130; /// Marks a size in `manifest show` that only the manifest's creator vouches for. const CLAIMED_SIZE_MARK: char = '*'; +/// Extension of a stopped build's partial manifest, written beside an +/// output file that already exists rather than over it. +const PARTIAL_MANIFEST_EXTENSION: &str = "partial.ant"; +/// Directory under the data directory that records the default download +/// directories whose download has not finished, so a re-run can continue. +const DOWNLOAD_RECORDS_DIR: &str = "downloads"; /// Manifest subcommands. #[derive(Subcommand, Debug)] @@ -143,8 +151,13 @@ pub enum ManifestAction { /// A `.ant` file, an `ant://manifest/...` link, or an upload id /// from `ant manifest list`. source: String, - /// Directory to extract into. Defaults to the manifest name, or - /// the current directory when it has none. + /// Directory to extract into, existing or not. Defaults to a new + /// directory named after the manifest in your downloads directory, + /// which must not exist yet unless an earlier run of this download + /// made it and did not finish. A manifest without a name, or with a + /// hidden one, needs --output. Files already in place with their + /// recorded SHA-256 are kept, so a download that stopped can be run + /// again. #[arg(short, long)] output: Option, /// Only extract these entries or directories. Repeatable. @@ -491,6 +504,9 @@ async fn create(client: &Client, args: CreateArgs, json: bool) -> anyhow::Result payment_mode: args.payment_mode, follow_symlinks: args.follow_symlinks, cancel: CancellationToken::new(), + // Each completed upload is journaled here before the next is paid + // for; a build that cannot record itself stops before paying. + history: Some(default_history_dir()?), }; // Fail before any paid upload if the manifest could not be written. let out = resolve_output(options.name.as_deref(), args.output.clone())?; @@ -547,8 +563,7 @@ async fn create(client: &Client, args: CreateArgs, json: bool) -> anyhow::Result result }; - let (result, history_id) = - record_build_result(built, |manifest| record_upload_manifest(manifest, None))?; + let (result, history_id) = build_outcome(built, &out)?; write_manifest(&result.manifest, &out, args.overwrite)?; if !json { eprintln!( @@ -571,10 +586,13 @@ async fn create(client: &Client, args: CreateArgs, json: bool) -> anyhow::Result ) } -/// Persist completed uploads before reporting cancellation or failure. -fn record_build_result( +/// Turn a build's outcome into its result and the id of the record that +/// holds it, or an error that says where the completed uploads are kept. The +/// builder has already recorded them in the history, complete or partial; +/// when it could not, a stopped build's partial manifest is written to `out`. +fn build_outcome( built: Result, - record: impl FnOnce(&Manifest) -> Option, + out: &Path, ) -> anyhow::Result<(BuildResult, Option)> { let result = match built { Ok(result) => result, @@ -583,24 +601,64 @@ fn record_build_result( cause, partial, }) => { - let history_id = record(&partial.manifest); + let at = if path.is_empty() { + String::new() + } else { + format!(" at {path:?}") + }; anyhow::bail!( - "manifest build failed at {path:?} after {} of {} file(s): {cause}{}", + "manifest build failed{at} after {} of {} file(s): {cause}{}", partial.files_uploaded, partial.total_to_upload, - partial_record_message(history_id.as_deref()), + keep_partial(&partial, out), ); } Err(error) => return Err(error.into()), }; - - let history_id = record(&result.manifest); if result.cancelled { - return Err(cancelled_build_error(&result, history_id.as_deref())); + return Err(cancelled_build_error(&result, &keep_partial(&result, out))); } + let history_id = recorded_id(&result); Ok((result, history_id)) } +/// The id of the record that holds a build's manifest, or `None`, with a +/// warning, when its record could not be written. +fn recorded_id(result: &BuildResult) -> Option { + match (&result.history_id, &result.history_error) { + (Some(id), Some(error)) => { + eprintln!( + "warning: the record of upload {id} could not be written ({error}); \ + its journal stays in the history, listed as unfinished" + ); + None + } + (id, _) => id.clone(), + } +} + +/// Where a stopped build's partial manifest is kept, for its error message: +/// its record, or `out` when the record could not be written, since a +/// journal the build could not finish may also lack the last upload. A file +/// already at `out`, which may be a complete manifest, is never replaced; +/// the partial manifest goes beside it instead. +fn keep_partial(partial: &BuildResult, out: &Path) -> String { + if let Some(id) = recorded_id(partial) { + return format!("; the partial manifest is recorded as {id}"); + } + if partial.files_uploaded == 0 { + return String::new(); + } + let beside = out.with_extension(PARTIAL_MANIFEST_EXTENSION); + let written = write_manifest(&partial.manifest, out, false) + .map(|()| out) + .or_else(|_| write_manifest(&partial.manifest, &beside, false).map(|()| beside.as_path())); + match written { + Ok(path) => format!("; the partial manifest is written to {}", path.display()), + Err(error) => format!("; the partial manifest could not be kept: {error}"), + } +} + /// The output path: the one given, or `.ant` in the current directory. fn resolve_output(name: Option<&str>, output: Option) -> anyhow::Result { if let Some(name) = name { @@ -610,32 +668,99 @@ fn resolve_output(name: Option<&str>, output: Option) -> anyhow::Result Ok(output.unwrap_or_else(|| PathBuf::from(manifest_filename_for(name)))) } -fn resolve_download_root(manifest: &Manifest, output: Option) -> anyhow::Result { +/// Where a download extracts to. +struct DownloadRoot { + path: PathBuf, + /// The record marking `path` as this manifest's unfinished default + /// download, removed once every entry is in place. + record: Option, +} + +/// The directory a download extracts into without `--output`: a new +/// directory named after the manifest, made in `parent`, the downloads +/// directory, and recorded in `records` until the download finishes. The +/// sender chooses that name, so it never picks a hidden directory or one +/// that already exists, unless an unfinished run of this same download made +/// it; a manifest without a name needs `--output`. Making the directory is +/// the check, so nothing can appear there in between. +fn default_download_root( + manifest: &Manifest, + parent: &Path, + records: Option<&Path>, +) -> anyhow::Result { manifest.validate()?; - Ok(output.unwrap_or_else(|| { - manifest - .name - .as_deref() - .map(PathBuf::from) - .unwrap_or_else(|| PathBuf::from(".")) - })) + let Some(name) = manifest.name.as_deref() else { + anyhow::bail!("this manifest has no name; pass --output to choose where to extract it"); + }; + if ant_core::data::manifest::path::is_hidden(name) { + anyhow::bail!( + "this manifest is named {name:?}, a hidden directory; \ + pass --output to choose where to extract it" + ); + } + std::fs::create_dir_all(parent) + .map_err(|e| anyhow::anyhow!("cannot create {}: {e}", parent.display()))?; + let path = parent.join(name); + match std::fs::create_dir(&path) { + Ok(()) => { + let record = records.and_then(|records| { + start_download_record(records, manifest, &path) + .inspect_err(|e| { + eprintln!("warning: this download cannot be continued if it stops: {e}") + }) + .ok() + }); + Ok(DownloadRoot { path, record }) + } + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { + let unfinished = records + .and_then(|records| download_record(records, manifest, &path).ok()) + .filter(|record| record.is_file()); + match unfinished { + Some(record) => Ok(DownloadRoot { + path, + record: Some(record), + }), + None => anyhow::bail!( + "{} already exists; choose where to extract with --output, \ + which may be that directory", + path.display() + ), + } + } + Err(e) => Err(anyhow::anyhow!("cannot create {}: {e}", path.display())), + } } -fn partial_record_message(history_id: Option<&str>) -> String { - history_id - .map(|id| format!("; the partial manifest is recorded as {id}")) - .unwrap_or_default() +/// The record marking `root` as the unfinished default download of +/// `manifest`, named after both, so no other download or directory matches. +fn download_record(records: &Path, manifest: &Manifest, root: &Path) -> anyhow::Result { + let root = std::fs::canonicalize(root).or_else(|_| std::path::absolute(root))?; + let mut key = Sha256::new(); + key.update(manifest.encode()?); + key.update(root.as_os_str().as_encoded_bytes()); + Ok(records.join(hex::encode(key.finalize()))) } -fn cancelled_build_error( - result: &ant_core::data::BuildResult, - history_id: Option<&str>, -) -> anyhow::Error { +/// Record that `root` was just made for `manifest`'s download. The record +/// holds the directory's path for anyone looking. +fn start_download_record( + records: &Path, + manifest: &Manifest, + root: &Path, +) -> anyhow::Result { + let record = download_record(records, manifest, root)?; + std::fs::create_dir_all(records)?; + std::fs::write(&record, root.display().to_string())?; + Ok(record) +} + +/// `kept` says where the partial manifest is kept, as [`keep_partial`] does. +fn cancelled_build_error(result: &ant_core::data::BuildResult, kept: &str) -> anyhow::Error { anyhow::anyhow!( - "cancelled after {} of {} file(s){}", + "cancelled after {} of {} file(s){kept}", result.files_uploaded, result.total_to_upload, - partial_record_message(history_id), ) } @@ -735,7 +860,7 @@ fn report_created( } ); if let Some(id) = history_id { - println!("Recorded as {id} (ant manifest download {id})"); + println!("Recorded as {id} (ant manifest download {id} --output )"); } if let Some((link, len)) = link_text { warn_if_long_link(len); @@ -759,6 +884,7 @@ fn list_history(json: bool) -> anyhow::Result<()> { "name": r.manifest.name, "entries": r.manifest.entries.len(), "total_size": r.manifest.total_size(), + "state": r.state, "file": r.path.display().to_string(), }) }) @@ -803,8 +929,12 @@ fn print_history_row(record: &UploadRecord) { // Drop the trailing " UTC"; the header already says so. let stamp = format_timestamp(record.recorded_at); let stamp = stamp.strip_suffix(" UTC").unwrap_or(&stamp); + let state = match record.state { + RecordState::Complete => "", + RecordState::Unfinished => " (unfinished: the files listed were uploaded)", + }; println!( - "{:<23} {:>7} {:>12} {}", + "{:<23} {:>7} {:>12} {}{state}", stamp, record.manifest.entries.len(), size, @@ -913,7 +1043,21 @@ async fn download( concurrency: NonZeroUsize, json: bool, ) -> anyhow::Result<()> { - let output_root = resolve_download_root(manifest, output)?; + // Before the default directory is made, so a selection matching nothing + // leaves nothing behind. + select_entries(manifest, &selection)?; + let records = ant_core::config::data_dir() + .ok() + .map(|dir| dir.join(DOWNLOAD_RECORDS_DIR)); + let root = match output { + Some(path) => DownloadRoot { path, record: None }, + None => default_download_root( + manifest, + &ant_core::config::download_dir()?, + records.as_deref(), + )?, + }; + let output_root = root.path.clone(); let options = ExtractOptions { output_root: output_root.clone(), selection, @@ -928,12 +1072,19 @@ async fn download( let report = if json { extract_manifest(client, manifest, &options, None).await? } else { + eprintln!("Downloading into {}", output_root.display()); let (tx, rx) = mpsc::channel(PROGRESS_CHANNEL_CAPACITY); let ui = tokio::spawn(drive_extract_progress(rx)); let report = extract_manifest(client, manifest, &options, Some(tx)).await; let _ = ui.await; report? }; + // Every entry is in place, so there is nothing left to continue. + if let Some(record) = &root.record { + if options.selection.is_empty() && report.failed() == 0 && report.cancelled() == 0 { + let _ = std::fs::remove_file(record); + } + } if json { println!( @@ -941,6 +1092,7 @@ async fn download( serde_json::to_string_pretty(&json!({ "output": output_root.display().to_string(), "written": report.written(), + "present": report.present(), "failed": report.failed(), "cancelled": report.cancelled(), "entries": report.entries, @@ -950,13 +1102,15 @@ async fn download( for e in &report.entries { match &e.status { EntryStatus::Written { bytes } => println!("written {} ({bytes} bytes)", e.name), + EntryStatus::Present { bytes } => println!("present {} ({bytes} bytes)", e.name), EntryStatus::Failed { error } => println!("failed {}: {error}", e.name), EntryStatus::Cancelled => println!("cancelled {}", e.name), } } eprintln!( - "{} written, {} failed in {:.1}s -> {}", + "{} written, {} already present, {} failed in {:.1}s -> {}", report.written(), + report.present(), report.failed(), start.elapsed().as_secs_f64(), output_root.display() @@ -992,7 +1146,7 @@ fn load_manifest(source: &str) -> anyhow::Result { } let dir = default_history_dir()?; match load_upload(&dir, source)? { - Some(record) => Ok(record.manifest), + Some(record) => Ok(record_manifest(record)), None => anyhow::bail!( "{source} is not a manifest file, a manifest link, or a recorded upload id \ (see `ant manifest list`)" @@ -1000,6 +1154,19 @@ fn load_manifest(source: &str) -> anyhow::Result { } } +/// A recorded upload's manifest, with a reminder of what an unfinished one +/// holds. +fn record_manifest(record: UploadRecord) -> Manifest { + if record.state == RecordState::Unfinished { + eprintln!( + "warning: upload {} has not finished; it holds the {} file(s) uploaded so far", + record.id, + record.manifest.entries.len() + ); + } + record.manifest +} + /// Parse the `--torrent-hash` values into one reference, or `None` if none. fn parse_torrent_hashes(hashes: &[String]) -> anyhow::Result> { let mut merged: Option = None; @@ -1176,32 +1343,73 @@ mod tests { }; assert!(action.execute_offline(false).is_err()); assert!(resolve_output(Some(name), None).is_err()); - assert!(resolve_download_root(&manifest, None).is_err()); + assert!(default_download_root(&manifest, dir.path(), None).is_err()); } assert!(!outside.with_extension("ant").exists()); - let valid = Manifest::new(Some("pack".into())); assert_eq!( resolve_output(Some("pack"), None).unwrap(), PathBuf::from("pack.ant") ); - assert_eq!( - resolve_download_root(&valid, None).unwrap(), - PathBuf::from("pack") - ); let explicit = dir.path().join("chosen/destination"); assert_eq!( resolve_output(Some("pack"), Some(explicit.clone())).unwrap(), explicit ); - assert_eq!( - resolve_download_root(&valid, Some(explicit.clone())).unwrap(), - explicit - ); - assert_eq!( - resolve_download_root(&Manifest::new(None), None).unwrap(), - PathBuf::from(".") - ); + } + + /// The sender names the manifest, so without --output a download only + /// ever extracts into a new directory it has just made in the downloads + /// directory, or into one an unfinished run of the same download made. + #[test] + fn a_download_defaults_only_to_a_new_directory_named_after_the_manifest() { + let dir = tempfile::tempdir().unwrap(); + let parent = dir.path().join("Downloads"); + let records = dir.path().join("records"); + let valid = Manifest::new(Some("pack".into())); + // A missing downloads directory is made. + let first = default_download_root(&valid, &parent, Some(&records)).unwrap(); + assert_eq!(first.path, parent.join("pack")); + assert!(first.path.is_dir()); + let record = first.record.expect("the new directory is recorded"); + assert!(record.is_file()); + + // The same download again continues into the directory it made. + let again = default_download_root(&valid, &parent, Some(&records)).unwrap(); + assert_eq!(again.path, first.path); + assert_eq!(again.record.as_ref(), Some(&record)); + + // Another manifest of the same name does not, nor does this one + // once its download has finished and the record is gone. + let mut other = Manifest::new(Some("pack".into())); + other.entries.push(ant_core::data::ManifestEntry { + path: Some("a.bin".into()), + size: None, + sha256: None, + source: ContentRef::Public { address: [1; 32] }, + }); + std::fs::remove_file(&record).unwrap(); + for manifest in [&other, &valid] { + let error = default_download_root(manifest, &parent, Some(&records)) + .err() + .unwrap(); + assert!(error.to_string().contains("already exists"), "{error}"); + } + + // Hidden and unnamed manifests need --output, and nothing is made. + for manifest in [ + Manifest::new(Some(".ssh".into())), + Manifest::new(Some("\u{200C}.ssh".into())), + Manifest::new(Some(".config".into())), + Manifest::new(None), + ] { + let error = default_download_root(&manifest, &parent, Some(&records)) + .err() + .unwrap(); + assert!(error.to_string().contains("--output"), "{error}"); + } + assert!(!parent.join(".ssh").exists()); + assert!(!parent.join(".config").exists()); } #[test] @@ -1215,6 +1423,8 @@ mod tests { chunks_stored: 3, storage_cost_atto: 0, gas_cost_wei: 0, + history_id: Some("record-id".into()), + history_error: None, }; for index in 0..3 { result.manifest.entries.push(ant_core::data::ManifestEntry { @@ -1225,15 +1435,18 @@ mod tests { address: [index; 32], }, }); + let out = PathBuf::from("unused.ant"); assert_eq!( - cancelled_build_error(&result, Some("record-id")).to_string(), + build_outcome(Ok(result.clone()), &out) + .unwrap_err() + .to_string(), "cancelled after 1 of 10 file(s); the partial manifest is recorded as record-id" ); } } #[test] - fn build_failure_and_cancellation_are_recorded_before_returning_an_error() { + fn build_failure_and_cancellation_report_where_the_build_was_recorded() { let manifest = Manifest { name: Some("partial".into()), torrent: None, @@ -1254,6 +1467,8 @@ mod tests { chunks_stored: 3, storage_cost_atto: 12, gas_cost_wei: 34, + history_id: Some("20261019T120000Z-partial".into()), + history_error: None, }; let built = if cancelled { Ok(partial) @@ -1264,23 +1479,86 @@ mod tests { partial: Box::new(partial), }) }; - let history = tempfile::tempdir().unwrap(); - let error = record_build_result(built, |manifest| { - Some(record_upload(history.path(), manifest, None).unwrap().id) - }) - .unwrap_err(); - let listing = list_uploads(history.path()).unwrap(); - assert_eq!(listing.records.len(), 1); - assert_eq!(listing.records[0].manifest, manifest); - assert!(error.to_string().contains(&listing.records[0].id)); + let out = PathBuf::from("unused.ant"); + let error = build_outcome(built, &out).unwrap_err(); + assert!(error + .to_string() + .contains("the partial manifest is recorded as 20261019T120000Z-partial")); assert!(error.to_string().contains("1 of 2 file(s)")); if !cancelled { assert!(error.to_string().contains("later.bin")); assert!(error.to_string().contains("forced failure")); } + assert!(!out.exists()); } } + /// A build whose record could not be written, as on a full disk, may have + /// a journal that lacks its last upload, so the partial manifest goes to + /// the output file instead, and no record is claimed. + #[test] + fn a_stopped_build_that_could_not_be_recorded_is_written_to_the_output() { + let dir = tempfile::tempdir().unwrap(); + let manifest = Manifest { + name: Some("partial".into()), + torrent: None, + entries: vec![ant_core::data::ManifestEntry { + path: Some("completed.bin".into()), + size: Some(10), + sha256: None, + source: ContentRef::Public { address: [1; 32] }, + }], + }; + let partial = BuildResult { + manifest: manifest.clone(), + cancelled: false, + skipped_symlinks: vec![], + files_uploaded: 1, + total_to_upload: 2, + chunks_stored: 3, + storage_cost_atto: 12, + gas_cost_wei: 34, + history_id: Some("20261019T120000Z-partial".into()), + history_error: Some("No space left on device".into()), + }; + let out = dir.path().join("partial.ant"); + let built = Err(ManifestError::BuildFailed { + path: "later.bin".into(), + cause: Box::new(ManifestError::Build("No space left on device".into())), + partial: Box::new(partial.clone()), + }); + let error = build_outcome(built, &out).unwrap_err().to_string(); + assert!(!error.contains("recorded as"), "{error}"); + assert!( + error.contains(&format!("written to {}", out.display())), + "{error}" + ); + assert_eq!(read_manifest_file(&out).unwrap(), manifest); + + // A file already at the output, which may be a complete manifest, is + // never replaced: the partial manifest goes beside it. + let complete = Manifest::new(Some("complete".into())); + write_manifest_file(&out, &complete, true).unwrap(); + let built = Err(ManifestError::BuildFailed { + path: "later.bin".into(), + cause: Box::new(ManifestError::Build("No space left on device".into())), + partial: Box::new(partial.clone()), + }); + let error = build_outcome(built, &out).unwrap_err().to_string(); + let beside = dir.path().join("partial.partial.ant"); + assert!( + error.contains(&format!("written to {}", beside.display())), + "{error}" + ); + assert_eq!(read_manifest_file(&out).unwrap(), complete); + assert_eq!(read_manifest_file(&beside).unwrap(), manifest); + + // A finished build keeps its manifest in the output anyway; only + // the record id is not claimed. + let (_, history_id) = build_outcome(Ok(partial), &out).unwrap(); + assert!(history_id.is_none()); + } + #[test] fn public_file_spec_parses_with_and_without_path() { let hex64 = "ab".repeat(32); diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index 1a3b272f..8107daea 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -20,6 +20,8 @@ blake3 = "1" base64 = "0.22" bytes = "1" hex = "0.4" +# Checksums for the frames of an upload journal (ADR-0006). +crc32fast = "1.5" rmp-serde = "1" serde = { version = "1", features = ["derive"] } serde_bytes = "0.11" @@ -30,6 +32,8 @@ sha2 = "0.10" # Portable WebRTC Direct profile shared with ant-node. saorsa-transport = { version = "0.38.2", default-features = false, features = ["webrtc"] } thiserror = "2" +# Full Unicode case folding for the portable manifest path rules (ADR-0006). +unicase = "2.9" # NFC normalisation for the portable manifest path rules (ADR-0006). unicode-normalization = "0.1" url = "2" @@ -38,6 +42,9 @@ web-time = "1.1" # Node management async-stream = { version = "0.3", optional = true } axum = { version = "0.8", optional = true } +# The user's downloads directory, the default for manifest downloads +# (ADR-0006). Already in the tree through saorsa-core. +dirs = { version = "6", optional = true } flate2 = { version = "1", optional = true } fs2 = { version = "0.4", optional = true } futures-core = { version = "0.3" } @@ -128,9 +135,12 @@ libc = "0.2" # Without this, the musl CI builds fail because the runners lack musl-compatible # OpenSSL dev libraries. Not needed on Windows (uses schannel via native-tls). openssl = { version = "0.10", features = ["vendored"] } +# Directory-relative file operations for containment-safe manifest +# extraction (ADR-0006). Already in the tree through tempfile. +rustix = { version = "1.1", features = ["fs"] } [target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_System_Console", "Win32_System_Threading"] } +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_Console", "Win32_System_Threading"] } [features] # Preserve the existing native ant-core API for ordinary consumers. WASM @@ -141,6 +151,7 @@ native = [ "ant-protocol/logging", "dep:async-stream", "dep:axum", + "dep:dirs", "dep:flate2", "dep:fs2", "dep:reqwest", diff --git a/ant-core/src/config.rs b/ant-core/src/config.rs index d982b2df..9aa74b32 100644 --- a/ant-core/src/config.rs +++ b/ant-core/src/config.rs @@ -4,6 +4,10 @@ use std::path::PathBuf; use crate::data::{CustomNetwork, DevnetManifest, EvmAddress, EvmNetwork, MultiAddr}; use crate::error::{Error, Result}; +/// The downloads directory under the home directory, where the platform +/// does not say where it is. +const DOWNLOADS_DIR_NAME: &str = "Downloads"; + /// Returns the platform-appropriate data directory for ant. /// /// - Linux: `~/.local/share/ant` @@ -57,6 +61,19 @@ pub fn log_dir() -> Result { } } +/// Returns the user's downloads directory, where downloads go unless told +/// otherwise. +/// +/// - Linux: `XDG_DOWNLOAD_DIR`, or `~/Downloads` +/// - macOS: `~/Downloads` +/// - Windows: the Downloads known folder, `%USERPROFILE%\Downloads` unless moved +pub fn download_dir() -> Result { + match dirs::download_dir() { + Some(dir) => Ok(dir), + None => Ok(home_dir()?.join(DOWNLOADS_DIR_NAME)), + } +} + /// Loads bootstrap peers from the platform-appropriate `bootstrap_peers.toml` file. /// /// Returns `Ok(Some(peers))` if the file exists and parses successfully, @@ -223,6 +240,11 @@ mod tests { assert_eq!(dir.file_name().unwrap(), "ant"); } + #[test] + fn download_dir_is_absolute() { + assert!(download_dir().unwrap().is_absolute()); + } + #[test] fn log_dir_contains_ant() { let dir = log_dir().unwrap(); diff --git a/ant-core/src/data/client/file/native.rs b/ant-core/src/data/client/file/native.rs index 052b7b97..0834c238 100644 --- a/ant-core/src/data/client/file/native.rs +++ b/ant-core/src/data/client/file/native.rs @@ -3617,6 +3617,27 @@ impl Client { }) .await } + + /// Download and decrypt a file into `writer`, with optional progress + /// events. + /// + /// For callers that open the destination themselves, such as a manifest + /// extraction that resolves every path from directories it holds open. + /// Constant memory (one decrypt batch resident at a time). Nothing is + /// renamed or removed here: on an error `writer` may hold part of the + /// file, and the caller decides what to do with it. + pub async fn file_download_to_writer( + &self, + data_map: &DataMap, + writer: &mut W, + progress: Option>, + ) -> Result { + let peer_count = self.config().close_group_size; + self.download_decrypted_chunks(data_map, progress, peer_count, None, None, |bytes| { + std::future::ready(writer.write_all(&bytes).map_err(Error::from)) + }) + .await + } } #[cfg(test)] diff --git a/ant-core/src/data/manifest/build.rs b/ant-core/src/data/manifest/build.rs index 61060446..af6653f1 100644 --- a/ant-core/src/data/manifest/build.rs +++ b/ant-core/src/data/manifest/build.rs @@ -8,24 +8,33 @@ use std::fs; use std::path::{Path, PathBuf}; -use self_encryption::MIN_ENCRYPTABLE_BYTES; +use self_encryption::{ChunkInfo, DataMap, MIN_ENCRYPTABLE_BYTES}; use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; +use xor_name::XorName; + +#[cfg(test)] +use self_encryption::{shrink_data_map, MAX_CHUNK_SIZE}; use crate::data::client::file::{UploadEvent, Visibility}; use crate::data::client::merkle::PaymentMode; use crate::data::Client; use super::embed::embeddable_data_map; +use super::journal::UploadJournal; use super::path::{check_collisions, validate_component, validate_path, PATH_SEPARATOR}; use super::{ data_map_address, ContentRef, Manifest, ManifestEntry, ManifestError, TorrentReference, - ADDRESS_LEN, MAX_MANIFEST_ENTRIES, + ADDRESS_LEN, MANIFEST_HEADER_LEN, MAX_MANIFEST_BYTES, MAX_MANIFEST_ENTRIES, + MAX_PUBLISHED_MAP_CHUNKS, SHA256_LEN, }; /// Capacity of the per-file upload progress channel. const UPLOAD_PROGRESS_CAPACITY: usize = 64; +/// Bytes the entry array's msgpack header can grow by as entries are added: +/// small counts live in the marker byte, the largest take four more bytes. +const ENTRY_ARRAY_HEADER_GROWTH: usize = std::mem::size_of::(); /// How a locally uploaded file is referenced from the manifest. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] @@ -60,6 +69,13 @@ pub struct BuildOptions { /// Cancels the build between files, or during an upload. Files already /// uploaded are kept in the partial result. pub cancel: CancellationToken, + /// Upload history directory to record the build in. A journal is + /// created there before the first paid upload and each completed upload + /// is flushed to it before the next is paid for, so a crash or a forced + /// quit loses none of them. When the build ends, complete or not, its + /// manifest becomes the record. A build that cannot create the journal + /// stops before paying for anything. + pub history: Option, } /// Progress during [`ManifestBuilder::finish`]. @@ -114,6 +130,16 @@ pub struct BuildResult { pub storage_cost_atto: u128, /// Gas paid across all uploads, in wei. pub gas_cost_wei: u128, + /// The upload history record of this build, when it was recorded. A + /// build that uploaded nothing before it stopped is not recorded. + #[serde(default)] + pub history_id: Option, + /// Why the record could not be written, when it could not. The journal + /// then stays in the history, listed as unfinished, with every upload + /// whose entry reached it. When appending to the journal is what failed, + /// the last upload is only in `manifest`, so keep that elsewhere. + #[serde(default)] + pub history_error: Option, } struct PendingFile { @@ -245,9 +271,11 @@ impl<'a> ManifestBuilder<'a> { } /// Check everything that could make the finished manifest invalid - /// before the first paid upload: file sizes, the name, and collisions - /// among all queued paths and address entries. - fn preflight(&self) -> Result<(), ManifestError> { + /// before the first paid upload: file sizes, the name, collisions among + /// all queued paths and address entries, and whether the manifest fits + /// [`MAX_MANIFEST_BYTES`] with every file at the largest entry it can + /// end as. Returns the size budget the uploads are admitted against. + fn preflight(&self) -> Result { let count = self.pending.len().saturating_add(self.entries.len()); if count > MAX_MANIFEST_ENTRIES { return Err(ManifestError::TooManyEntries { @@ -269,6 +297,7 @@ impl<'a> ManifestBuilder<'a> { )); } let mut names = Vec::with_capacity(self.pending.len() + self.entries.len()); + let mut budget = SizeBudget::new(&self.options)?; for file in &self.pending { let size = fs::metadata(&file.local)?.len(); if size < MIN_ENCRYPTABLE_BYTES as u64 { @@ -278,11 +307,15 @@ impl<'a> ManifestBuilder<'a> { ))); } names.push(file.path.clone()); + budget.queue_upload(&file.path)?; } for entry in &self.entries { names.push(entry.effective_name()?); + budget.add(entry)?; } - check_collisions(names.iter().map(String::as_str)).map_err(ManifestError::Conflict) + check_collisions(names.iter().map(String::as_str)).map_err(ManifestError::Conflict)?; + budget.ensure_fits()?; + Ok(budget) } /// Upload every queued file and assemble the manifest. @@ -297,7 +330,8 @@ impl<'a> ManifestBuilder<'a> { mut self, progress: Option>, ) -> Result { - self.preflight()?; + let mut budget = self.preflight()?; + let mut journal = self.start_journal()?; let total = self.pending.len(); let mut files_uploaded = 0; let mut chunks_stored = 0; @@ -369,23 +403,43 @@ impl<'a> ManifestBuilder<'a> { // Keep the upload's usable DataMap before any fallible cost // parsing or reference optimisation. A later error must not // discard the file that has just been paid for and uploaded. - self.entries.push(ManifestEntry { + let uploaded = ManifestEntry { path: Some(file.path.clone()), size: Some(size), sha256: Some(result.sha256), source: ContentRef::Embedded { data_map: result.data_map.clone(), }, - }); + }; + let counted = budget.uploaded(index, &uploaded); + // On disk before anything else can fail or be paid for. If it + // cannot be, the build stops rather than pay for uploads a crash + // would lose. + let journaled = journal + .as_mut() + .map_or(Ok(()), |journal| journal.append(&uploaded)); + let storage_cost = parse_atto(&result.storage_cost_atto); + self.entries.push(uploaded); files_uploaded += 1; chunks_stored += result.chunks_stored; gas_cost_wei = gas_cost_wei.saturating_add(result.gas_cost_wei); - match parse_atto(&result.storage_cost_atto) { - Ok(cost) => storage_cost_atto = storage_cost_atto.saturating_add(cost), + if let Ok(cost) = &storage_cost { + storage_cost_atto = storage_cost_atto.saturating_add(*cost); + } + if let Err(error) = journaled { + failure = Some((file.path, error)); + break; + } + let counted = match counted { + Ok(counted) => counted, Err(error) => { failure = Some((file.path, error)); break; } + }; + if let Err(error) = storage_cost { + failure = Some((file.path, error)); + break; } let source = match self @@ -398,11 +452,28 @@ impl<'a> ManifestBuilder<'a> { break; } }; - let reference = source.kind(); - // The entry just appended is always present at this point. - if let Some(entry) = self.entries.last_mut() { - entry.source = source; + // The entry just appended is always present at this point. A + // larger reference, such as an embedded root map, replaces the + // uploaded map only if the manifest still fits with it. + let Some(entry) = self.entries.last_mut() else { + break; + }; + let changed = match budget.offer_source(entry, counted, source) { + Ok(changed) => changed, + Err(error) => { + failure = Some((file.path, error)); + break; + } + }; + // Record the final reference too; the uploaded form is already + // safe, so a failure here only stops the build. + if let (true, Some(journal)) = (changed, journal.as_mut()) { + if let Err(error) = journal.append(entry) { + failure = Some((file.path, error)); + break; + } } + let reference = entry.source.kind(); if let Some(tx) = &progress { let _ = tx @@ -420,7 +491,29 @@ impl<'a> ManifestBuilder<'a> { torrent: self.options.torrent.clone(), entries: std::mem::take(&mut self.entries), }; - manifest.canonicalize()?; + // Everything canonicalising checks was checked before the first + // upload, so this should not fail. If it does, the partial result is + // still returned, and the journal keeps every completed upload. + if let Err(error) = manifest.canonicalize() { + failure.get_or_insert((String::new(), error)); + } + + let stopped_early = cancelled || failure.is_some(); + let (history_id, history_error) = match journal { + None => (None, None), + // Nothing was paid for, so there is nothing to keep. + Some(journal) if stopped_early && files_uploaded == 0 => { + let _ = journal.discard(); + (None, None) + } + Some(journal) => { + let id = journal.id().to_string(); + match journal.finish(&manifest) { + Ok(record) => (Some(record.id), None), + Err(error) => (Some(id), Some(error.to_string())), + } + } + }; let result = BuildResult { manifest, @@ -431,6 +524,8 @@ impl<'a> ManifestBuilder<'a> { chunks_stored, storage_cost_atto, gas_cost_wei, + history_id, + history_error, }; match failure { Some((path, cause)) => Err(ManifestError::BuildFailed { @@ -442,6 +537,25 @@ impl<'a> ManifestBuilder<'a> { } } + /// Create the build's journal, when it is recorded, holding the entries + /// added without uploading. Runs before the first paid upload. + fn start_journal(&self) -> Result, ManifestError> { + let Some(dir) = &self.options.history else { + return Ok(None); + }; + let mut journal = UploadJournal::create( + dir, + None, + self.options.name.as_deref(), + self.options.torrent.as_ref(), + )?; + if let Err(error) = journal.append_all(&self.entries) { + let _ = journal.discard(); + return Err(error); + } + Ok(Some(journal)) + } + /// Embedded mode always embeds a DataMap, even when the upload was /// public, and embeds the root map when it is small enough so readers /// skip the wrapper-record fetches. Compact mode records the public @@ -511,6 +625,157 @@ fn file_name_utf8(path: &Path) -> Result { .ok_or_else(|| ManifestError::Build(format!("{} has no UTF-8 file name", path.display()))) } +/// Keeps a build's manifest within [`MAX_MANIFEST_BYTES`] however its +/// uploads turn out, so a build that passes preflight never stops for size. +/// +/// Each queued file is counted at the largest entry it can end as: its own +/// upload map at the deepest child level, which bounds every map an upload +/// returns. Once uploaded, a file is counted at its real size, and a larger +/// reference, such as an embedded root map, is taken only if the manifest +/// still fits with it. Every partial or complete result can therefore be +/// encoded and recorded, and nothing is paid for that could not be. A file +/// that ends as a public address is smaller than counted, so the budget may +/// refuse a build that would just have fitted, never the reverse. +struct SizeBudget { + /// Encoded bytes the manifest can need, as counted so far. + reserved: usize, + /// Each queued file's counted entry, in queue order. + counted: Vec, +} + +impl SizeBudget { + /// The fixed part of a manifest with these options and no entries. + fn new(options: &BuildOptions) -> Result { + let empty = Manifest { + name: options.name.clone(), + torrent: options.torrent.clone(), + entries: Vec::new(), + }; + Ok(Self { + reserved: MANIFEST_HEADER_LEN + encoded_len(&empty)? + ENTRY_ARRAY_HEADER_GROWTH, + counted: Vec::new(), + }) + } + + /// Count a queued file at the largest entry it can end as. + fn queue_upload(&mut self, path: &str) -> Result<(), ManifestError> { + let counted = encoded_len(&largest_upload_entry(path))?; + self.counted.push(counted); + self.reserved = self.reserved.saturating_add(counted); + Ok(()) + } + + /// Count an entry added without uploading. + fn add(&mut self, entry: &ManifestEntry) -> Result<(), ManifestError> { + self.reserved = self.reserved.saturating_add(encoded_len(entry)?); + Ok(()) + } + + /// Refuse a manifest that could outgrow the limit. + fn ensure_fits(&self) -> Result<(), ManifestError> { + fits(self.reserved) + } + + /// Count queued file `index` at its uploaded entry from now on, and + /// return that entry's size. + fn uploaded(&mut self, index: usize, entry: &ManifestEntry) -> Result { + let len = encoded_len(entry)?; + self.reserved = self.swapped(self.counted_of(index), len); + Ok(len) + } + + /// Give `entry`, counted at `current` bytes, the reference `source` when + /// the manifest still fits with it; otherwise it keeps the one it has. + /// Returns whether the reference changed. + fn offer_source( + &mut self, + entry: &mut ManifestEntry, + current: usize, + source: ContentRef, + ) -> Result { + let previous = std::mem::replace(&mut entry.source, source); + let offered = match encoded_len(entry) { + Ok(len) => len, + Err(error) => { + entry.source = previous; + return Err(error); + } + }; + let projected = self.swapped(current, offered); + if projected <= MAX_MANIFEST_BYTES { + self.reserved = projected; + Ok(entry.source != previous) + } else { + entry.source = previous; + Ok(false) + } + } + + fn counted_of(&self, index: usize) -> usize { + self.counted.get(index).copied().unwrap_or_default() + } + + /// The reserved total with `old` bytes replaced by `new`. + fn swapped(&self, old: usize, new: usize) -> usize { + self.reserved.saturating_sub(old).saturating_add(new) + } +} + +/// Refuse a manifest of `len` encoded bytes when it exceeds the limit. +fn fits(len: usize) -> Result<(), ManifestError> { + if len > MAX_MANIFEST_BYTES { + return Err(ManifestError::TooLarge { + len, + max: MAX_MANIFEST_BYTES, + }); + } + Ok(()) +} + +/// An uploaded file's entry with the given size and reference. +fn upload_entry(path: &str, size: u64, source: ContentRef) -> ManifestEntry { + ManifestEntry { + path: Some(path.to_string()), + size: Some(size), + sha256: Some([0; SHA256_LEN]), + source, + } +} + +/// The largest entry an upload of `path` can become, with the largest size, +/// since the file can grow before it is uploaded. +fn largest_upload_entry(path: &str) -> ManifestEntry { + upload_entry( + path, + u64::MAX, + ContentRef::Embedded { + data_map: largest_upload_map(), + }, + ) +} + +/// The largest DataMap an upload returns, as the manifest encodes it: its +/// most chunk records, at the deepest child level. The manifest writes every chunk +/// record at a fixed size, so only the child level varies. +fn largest_upload_map() -> DataMap { + let infos = (0..MAX_PUBLISHED_MAP_CHUNKS) + .map(|index| ChunkInfo { + index, + dst_hash: XorName::default(), + src_hash: XorName::default(), + src_size: 0, + }) + .collect(); + DataMap::with_child(infos, usize::MAX) +} + +/// Bytes `value` takes in a manifest's msgpack encoding. +fn encoded_len(value: &T) -> Result { + rmp_serde::to_vec_named(value) + .map(|bytes| bytes.len()) + .map_err(|e| ManifestError::Encode(e.to_string())) +} + fn parse_atto(value: &str) -> Result { value .parse::() @@ -522,6 +787,154 @@ fn parse_atto(value: &str) -> Result { mod tests { use super::*; + fn embedded(data_map: DataMap) -> ContentRef { + ContentRef::Embedded { data_map } + } + + fn root_of(chunks: usize) -> DataMap { + DataMap::new( + (0..chunks) + .map(|index| ChunkInfo { + index, + dst_hash: XorName::from_content(&index.to_le_bytes()), + src_hash: XorName::from_content(&index.to_be_bytes()), + src_size: MAX_CHUNK_SIZE, + }) + .collect(), + ) + } + + #[test] + fn the_largest_upload_entry_bounds_every_real_one() { + let (shrunk, _) = shrink_data_map(root_of(1_000), |_, _| Ok(())).unwrap(); + assert!(shrunk.is_child()); + assert_eq!(shrunk.len(), MAX_PUBLISHED_MAP_CHUNKS); + let bound = encoded_len(&largest_upload_entry("dir/f.bin")).unwrap(); + for map in [shrunk, root_of(MAX_PUBLISHED_MAP_CHUNKS)] { + let real = encoded_len(&upload_entry("dir/f.bin", 12_345, embedded(map))).unwrap(); + assert!(real <= bound, "{real} > {bound}"); + } + } + + /// Preflight passes with less than one more file to spare. Every upload + /// then returns a shrunk map with a child level and is offered another + /// reference, a larger root map or a public address, and the manifest + /// never outgrows the limit. + #[test] + fn a_build_that_passes_preflight_never_stops_for_size() { + let path = format!("{}/f.bin", "d".repeat(60_000)); + let (shrunk, _) = shrink_data_map(root_of(1_000), |_, _| Ok(())).unwrap(); + let per_file = encoded_len(&largest_upload_entry(&path)).unwrap(); + for offered in [ + embedded(root_of(100)), + ContentRef::Public { + address: [7; ADDRESS_LEN], + }, + ] { + let mut budget = SizeBudget::new(&BuildOptions::default()).unwrap(); + while budget.reserved + per_file <= MAX_MANIFEST_BYTES { + budget.queue_upload(&path).unwrap(); + } + budget.ensure_fits().unwrap(); + for index in 0..budget.counted.len() { + let mut entry = upload_entry(&path, 12_345_678_901, embedded(shrunk.clone())); + let counted = budget.uploaded(index, &entry).unwrap(); + assert!(budget.reserved <= MAX_MANIFEST_BYTES); + budget + .offer_source(&mut entry, counted, offered.clone()) + .unwrap(); + assert!(budget.reserved <= MAX_MANIFEST_BYTES); + } + } + } + + /// Many long paths in the default embedded mode: as addresses they would + /// fit, embedded they cannot, so the build is refused before any upload. + #[test] + fn a_build_that_cannot_fit_embedded_is_refused_before_paying() { + let path = format!("{}/file.bin", "d".repeat(400)); + let address_len = encoded_len(&upload_entry( + &path, + 4_096, + ContentRef::Public { + address: [0; ADDRESS_LEN], + }, + )) + .unwrap(); + let mut budget = SizeBudget::new(&BuildOptions::default()).unwrap(); + let files = MAX_MANIFEST_BYTES / address_len - 1; + for _ in 0..files { + budget.queue_upload(&path).unwrap(); + } + assert!(files * address_len < MAX_MANIFEST_BYTES); + assert!(matches!( + budget.ensure_fits(), + Err(ManifestError::TooLarge { .. }) + )); + } + + #[test] + fn the_fixed_part_counts_everything_but_the_entries() { + let options = BuildOptions { + name: Some("photos".into()), + ..BuildOptions::default() + }; + let mut budget = SizeBudget::new(&options).unwrap(); + let entry = upload_entry("a.bin", 4_096, embedded(root_of(3))); + budget.add(&entry).unwrap(); + let manifest = Manifest { + name: options.name, + torrent: None, + entries: vec![entry], + }; + let actual = manifest.encode().unwrap().len(); + assert!(actual <= budget.reserved, "{actual} > {}", budget.reserved); + assert!(budget.reserved - actual <= ENTRY_ARRAY_HEADER_GROWTH); + } + + #[test] + fn a_larger_reference_is_taken_only_when_it_fits() { + let mut budget = SizeBudget::new(&BuildOptions::default()).unwrap(); + budget.queue_upload("a.bin").unwrap(); + let mut entry = upload_entry("a.bin", 4_096, embedded(largest_upload_map())); + let counted = budget.uploaded(0, &entry).unwrap(); + + // With room, a large root map replaces the uploaded map. + assert!(budget + .offer_source(&mut entry, counted, embedded(root_of(100))) + .unwrap()); + assert!( + matches!(&entry.source, ContentRef::Embedded { data_map } if data_map.len() == 100) + ); + let with_root = encoded_len(&entry).unwrap(); + + // Full to the limit: a still larger map is refused and the entry + // keeps the one it has; a smaller reference is always taken. + budget.reserved = MAX_MANIFEST_BYTES; + assert!(!budget + .offer_source(&mut entry, with_root, embedded(root_of(200))) + .unwrap()); + assert!( + matches!(&entry.source, ContentRef::Embedded { data_map } if data_map.len() == 100) + ); + assert_eq!(budget.reserved, MAX_MANIFEST_BYTES); + let public = ContentRef::Public { + address: [7; ADDRESS_LEN], + }; + assert!(budget + .offer_source(&mut entry, with_root, public.clone()) + .unwrap()); + // Offering the same reference again changes nothing. + let now = budget.reserved; + let current = encoded_len(&entry).unwrap(); + assert!(!budget + .offer_source(&mut entry, current, public.clone()) + .unwrap()); + assert_eq!(budget.reserved, now); + assert_eq!(entry.source, public); + assert!(budget.reserved < MAX_MANIFEST_BYTES); + } + #[test] fn file_name_requires_utf8_and_a_name() { assert_eq!(file_name_utf8(Path::new("a/b.txt")).unwrap(), "b.txt"); diff --git a/ant-core/src/data/manifest/extract.rs b/ant-core/src/data/manifest/extract.rs index 4a4a5194..9ed52927 100644 --- a/ant-core/src/data/manifest/extract.rs +++ b/ant-core/src/data/manifest/extract.rs @@ -2,16 +2,16 @@ //! //! Containment is a filesystem contract, not a string check. Inside the //! output root the extractor never follows a symlink, junction or reparse -//! point: every intermediate directory is created, or confirmed to be a -//! real directory, without following links, before the download, and the -//! whole parent chain and the target are checked again, without following -//! links, immediately before the final rename. The file is written through -//! a reserved temporary name in its parent. An existing target is a -//! per-entry failure unless overwrite was requested, and even then only a -//! regular file is replaced. - -use std::fs; -use std::io; +//! point, and every name is resolved from directories held open, so +//! nothing that changes while an entry downloads can redirect where it is +//! written (see the `output` module). The file is written through a +//! temporary name in its parent and renamed into place without replacing +//! anything. A regular file already at a target with the entry's recorded +//! SHA-256 is kept as that entry, so an extraction that stopped part way can +//! be run again. Any other existing target is a per-entry failure unless +//! overwrite was requested, and even then only a regular file is replaced. + +use std::io::{self, Write}; use std::num::NonZeroUsize; use std::path::{Path, PathBuf}; @@ -19,7 +19,6 @@ use futures::stream::{self, StreamExt}; use self_encryption::DataMap; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use tempfile::Builder as TempBuilder; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -27,16 +26,10 @@ use crate::data::client::file::DownloadEvent; use crate::data::error::Error; use crate::data::Client; +use super::output::OutputRoot; use super::path::PATH_SEPARATOR; use super::{ContentRef, Manifest, ManifestEntry, ManifestError, SHA256_LEN}; -#[cfg(all(test, unix))] -use std::os::unix::fs::symlink; - -/// Read buffer for checking a downloaded file's SHA-256. -const SHA256_READ_BUFFER: usize = 1024 * 1024; -/// Prefix of the temporary file an entry is downloaded into. -const TEMP_PREFIX: &str = ".ant-extract-"; /// Capacity of the per-entry download progress channel. const DOWNLOAD_PROGRESS_CAPACITY: usize = 64; @@ -92,6 +85,12 @@ pub enum EntryStatus { /// Bytes written. bytes: u64, }, + /// Already at its target with the entry's recorded SHA-256, as an + /// earlier run left it. Nothing was downloaded or written. + Present { + /// Bytes of the file. + bytes: u64, + }, /// Failed; nothing was left at the target. Failed { /// The error. @@ -123,6 +122,11 @@ impl ExtractReport { self.count(|s| matches!(s, EntryStatus::Written { .. })) } + /// Entries already present. + pub fn present(&self) -> usize { + self.count(|s| matches!(s, EntryStatus::Present { .. })) + } + /// Entries that failed. pub fn failed(&self) -> usize { self.count(|s| matches!(s, EntryStatus::Failed { .. })) @@ -187,23 +191,22 @@ pub async fn extract_manifest( ) -> Result { manifest.validate()?; let selected = select_entries(manifest, &options.selection)?; - fs::create_dir_all(&options.output_root)?; - let total = selected.len(); + let root = OutputRoot::open(&options.output_root)?; + let extraction = Extraction { + client, + root: &root, + options, + total: selected.len(), + }; + let extraction = &extraction; let outcomes: Vec<(usize, EntryOutcome)> = stream::iter(selected.into_iter().enumerate()) .map(|(index, (name, entry))| { let progress = progress.clone(); async move { - let status = extract_entry( - client, - entry, - &name, - index, - total, - options, - progress.as_ref(), - ) - .await; + let status = extraction + .extract_entry(entry, &name, index, progress.as_ref()) + .await; if let Some(tx) = &progress { let _ = tx .send(ExtractEvent::EntryFinished { @@ -226,125 +229,147 @@ pub async fn extract_manifest( }) } -async fn extract_entry( - client: &Client, - entry: &ManifestEntry, - name: &str, - index: usize, +/// What every entry of one extraction shares. +struct Extraction<'a> { + client: &'a Client, + root: &'a OutputRoot, + options: &'a ExtractOptions, + /// Number of selected entries. total: usize, - options: &ExtractOptions, - progress: Option<&mpsc::Sender>, -) -> EntryStatus { - if options.cancel.is_cancelled() { - return EntryStatus::Cancelled; - } - if let Some(tx) = progress { - let _ = tx - .send(ExtractEvent::EntryStarted { - name: name.to_string(), - index, - total, - }) - .await; - } - let work = download_entry(client, entry, name, options, progress); - tokio::select! { - _ = options.cancel.cancelled() => EntryStatus::Cancelled, - result = work => match result { - Ok(bytes) => EntryStatus::Written { bytes }, - Err(e) => EntryStatus::Failed { error: e.to_string() }, - }, - } } -async fn download_entry( - client: &Client, - entry: &ManifestEntry, - name: &str, - options: &ExtractOptions, - progress: Option<&mpsc::Sender>, -) -> Result { - let target = prepare_target(&options.output_root, name, options.overwrite)?; - let parent = target - .parent() - .ok_or_else(|| ManifestError::Build(format!("{} has no parent", target.display())))?; - - // The root map lists every chunk, so its size is the file's real size, - // unlike the creator's unverified `size` hint. The download would - // resolve the root anyway. - let data_map = client - .data_map_resolve_root(&resolve_data_map(client, &entry.source).await?) - .await?; - ensure_space(parent, data_map.original_file_size())?; - - let temp = TempBuilder::new() - .prefix(TEMP_PREFIX) - .tempfile_in(parent)? - .into_temp_path(); - - let download_progress = progress.map(|tx| { - let (dl_tx, mut dl_rx) = mpsc::channel(DOWNLOAD_PROGRESS_CAPACITY); - let tx = tx.clone(); - let name = name.to_string(); - tokio::spawn(async move { - while let Some(event) = dl_rx.recv().await { - let _ = tx - .send(ExtractEvent::Download { - name: name.clone(), - event, - }) - .await; +impl Extraction<'_> { + async fn extract_entry( + &self, + entry: &ManifestEntry, + name: &str, + index: usize, + progress: Option<&mpsc::Sender>, + ) -> EntryStatus { + if self.options.cancel.is_cancelled() { + return EntryStatus::Cancelled; + } + if let Some(tx) = progress { + let _ = tx + .send(ExtractEvent::EntryStarted { + name: name.to_string(), + index, + total: self.total, + }) + .await; + } + let work = self.download_entry(entry, name, progress); + tokio::select! { + _ = self.options.cancel.cancelled() => EntryStatus::Cancelled, + result = work => match result { + Ok(status) => status, + Err(e) => EntryStatus::Failed { error: e.to_string() }, + }, + } + } + + async fn download_entry( + &self, + entry: &ManifestEntry, + name: &str, + progress: Option<&mpsc::Sender>, + ) -> Result { + let overwrite = self.options.overwrite; + let target = self.root.target(name)?; + // Hashing a large file blocks, so keep it off the async workers. + // The size, exact or as the creator recorded it, spares reading a + // file that cannot be the entry. + if let Some(expected) = &entry.sha256 { + let size = entry.known_size().or(entry.size); + if let Some(bytes) = tokio::task::block_in_place(|| target.identical(expected, size)) { + return Ok(EntryStatus::Present { bytes }); } - }); - dl_tx - }); - - let bytes = client - .file_download_with_progress(&data_map, &temp, download_progress) - .await?; - - // Check the creator's recorded SHA-256 before the file takes its name. - // On a mismatch the temporary file is dropped, so nothing unverified is - // left behind. - if let Some(expected) = entry.sha256 { - let path = temp.to_path_buf(); - let actual = tokio::task::spawn_blocking(move || file_sha256(&path)) - .await - .map_err(|e| ManifestError::Build(format!("hashing task failed: {e}")))??; - if actual != expected { - return Err(ManifestError::Sha256Mismatch { - expected: hex::encode(expected), - actual: hex::encode(actual), + } + target.check(overwrite)?; + + // The root map lists every chunk, so its size is the file's real + // size, unlike the creator's unverified `size` hint. The download + // would resolve the root anyway. + let client = self.client; + let data_map = client + .data_map_resolve_root(&resolve_data_map(client, &entry.source).await?) + .await?; + ensure_space(target.parent_path(), data_map.original_file_size())?; + + let download_progress = progress.map(|tx| { + let (dl_tx, mut dl_rx) = mpsc::channel(DOWNLOAD_PROGRESS_CAPACITY); + let tx = tx.clone(); + let name = name.to_string(); + tokio::spawn(async move { + while let Some(event) = dl_rx.recv().await { + let _ = tx + .send(ExtractEvent::Download { + name: name.clone(), + event, + }) + .await; + } }); + dl_tx + }); + + // Dropping the staged file on any early return, including a + // cancellation, removes it. + let mut staged = target.stage()?; + let mut writer = HashingWriter::new(staged.file()?, entry.sha256.is_some()); + let bytes = client + .file_download_to_writer(&data_map, &mut writer, download_progress) + .await?; + + // Check the creator's recorded SHA-256 before the file takes its + // name, so nothing unverified is left at the target. + if let (Some(expected), Some(actual)) = (entry.sha256, writer.finish()) { + if actual != expected { + return Err(ManifestError::Sha256Mismatch { + expected: hex::encode(expected), + actual: hex::encode(actual), + }); + } } + // The commit syncs the file to disk; keep that off the async workers. + tokio::task::block_in_place(|| staged.commit(overwrite))?; + Ok(EntryStatus::Written { bytes }) } +} - // Re-check the parent chain and the target immediately before the - // rename; either may have changed while the download ran. - verify_parent_chain(&options.output_root, name)?; - check_target(&target, options.overwrite)?; - let persisted = if options.overwrite { - temp.persist(&target) - } else { - temp.persist_noclobber(&target) - }; - persisted.map_err(|e| ManifestError::Io(e.error))?; - Ok(bytes) +/// Writes through to a file, hashing what it writes when asked to, so a +/// download is checked in the same pass that stores it. +struct HashingWriter<'a, W> { + inner: &'a mut W, + hasher: Option, +} + +impl<'a, W: Write> HashingWriter<'a, W> { + fn new(inner: &'a mut W, hash: bool) -> Self { + Self { + inner, + hasher: hash.then(Sha256::new), + } + } + + /// The SHA-256 of everything written, when hashing was asked for. + fn finish(self) -> Option<[u8; SHA256_LEN]> { + self.hasher.map(|hasher| hasher.finalize().into()) + } } -/// SHA-256 of the file at `path`, read in one streaming pass. -fn file_sha256(path: &Path) -> io::Result<[u8; SHA256_LEN]> { - let mut file = fs::File::open(path)?; - let mut hasher = Sha256::new(); - let mut buffer = vec![0u8; SHA256_READ_BUFFER]; - loop { - let n = io::Read::read(&mut file, &mut buffer)?; - if n == 0 { - break; +impl Write for HashingWriter<'_, W> { + fn write(&mut self, buf: &[u8]) -> io::Result { + let written = self.inner.write(buf)?; + if let Some(hasher) = &mut self.hasher { + hasher.update(&buf[..written]); } - hasher.update(&buffer[..n]); + Ok(written) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() } - Ok(hasher.finalize().into()) } /// Fail the entry before downloading when `dir`'s filesystem cannot hold @@ -371,115 +396,27 @@ async fn resolve_data_map(client: &Client, source: &ContentRef) -> Result Result { - let mut components = name.split(PATH_SEPARATOR).peekable(); - let mut current = root.to_path_buf(); - while let Some(component) = components.next() { - current.push(component); - if components.peek().is_none() { - check_target(¤t, overwrite)?; - return Ok(current); - } - ensure_real_directory(¤t)?; - } - Err(ManifestError::Build("empty effective name".to_string())) -} - -/// Confirm, without creating anything or following links, that every -/// directory component of `name` beneath `root` is still a real directory. -fn verify_parent_chain(root: &Path, name: &str) -> Result<(), ManifestError> { - let mut components = name.split(PATH_SEPARATOR).peekable(); - let mut current = root.to_path_buf(); - while let Some(component) = components.next() { - if components.peek().is_none() { - return Ok(()); - } - current.push(component); - let meta = fs::symlink_metadata(¤t)?; - verify_directory(¤t, &meta)?; - } - Ok(()) -} - -fn ensure_real_directory(path: &Path) -> Result<(), ManifestError> { - match fs::symlink_metadata(path) { - Ok(meta) => verify_directory(path, &meta), - Err(e) if e.kind() == io::ErrorKind::NotFound => { - match fs::create_dir(path) { - Ok(()) => {} - Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {} - Err(e) => return Err(e.into()), - } - let meta = fs::symlink_metadata(path)?; - verify_directory(path, &meta) - } - Err(e) => Err(e.into()), - } -} - -fn verify_directory(path: &Path, meta: &fs::Metadata) -> Result<(), ManifestError> { - if meta.file_type().is_symlink() { - return Err(ManifestError::Build(format!( - "{} is a symlink; extraction does not follow links inside the output directory", - path.display() - ))); - } - if !meta.is_dir() { - return Err(ManifestError::Build(format!( - "{} exists and is not a directory", - path.display() - ))); - } - Ok(()) -} - -fn check_target(target: &Path, overwrite: bool) -> Result<(), ManifestError> { - match fs::symlink_metadata(target) { - Ok(meta) => { - if meta.file_type().is_symlink() { - return Err(ManifestError::Build(format!( - "{} is a symlink and will not be replaced", - target.display() - ))); - } - if !overwrite { - return Err(ManifestError::Build(format!( - "{} already exists; pass overwrite to replace it", - target.display() - ))); - } - if !meta.is_file() { - return Err(ManifestError::Build(format!( - "{} exists and is not a regular file", - target.display() - ))); - } - Ok(()) - } - Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), - Err(e) => Err(e.into()), - } -} - #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used)] mod tests { use super::*; #[test] - fn file_sha256_matches_a_one_shot_digest() { - let dir = tempfile::tempdir().unwrap(); - // Spans several read buffers, with a ragged tail. - let data: Vec = (0..(2 * SHA256_READ_BUFFER + 4321)) - .map(|i| (i % 251) as u8) - .collect(); - let path = dir.path().join("f.bin"); - fs::write(&path, &data).unwrap(); + fn hashing_writer_matches_a_one_shot_digest_and_writes_everything() { + let data: Vec = (0..100_003u32).map(|i| (i % 251) as u8).collect(); + let mut out = Vec::new(); + let mut writer = HashingWriter::new(&mut out, true); + for piece in data.chunks(4096) { + writer.write_all(piece).unwrap(); + } let expected: [u8; SHA256_LEN] = Sha256::digest(&data).into(); - assert_eq!(file_sha256(&path).unwrap(), expected); + assert_eq!(writer.finish(), Some(expected)); + assert_eq!(out, data); + + let mut unhashed = Vec::new(); + let mut writer = HashingWriter::new(&mut unhashed, false); + writer.write_all(&data).unwrap(); + assert_eq!(writer.finish(), None); } fn entry(path: &str) -> ManifestEntry { @@ -517,51 +454,4 @@ mod tests { assert_eq!(trailing.len(), 2); assert!(select_entries(&manifest, &["zzz".into()]).is_err()); } - - #[test] - fn prepare_target_creates_directories_and_refuses_existing_file() { - let root = tempfile::tempdir().unwrap(); - let target = prepare_target(root.path(), "x/y/z.txt", false).unwrap(); - assert_eq!(target, root.path().join("x").join("y").join("z.txt")); - assert!(root.path().join("x/y").is_dir()); - fs::write(&target, b"old").unwrap(); - assert!(prepare_target(root.path(), "x/y/z.txt", false).is_err()); - assert!(prepare_target(root.path(), "x/y/z.txt", true).is_ok()); - // A directory where a file should go is never replaced. - assert!(prepare_target(root.path(), "x/y", true).is_err()); - // A file where a directory should go. - assert!(prepare_target(root.path(), "x/y/z.txt/deeper", false).is_err()); - } - - #[cfg(unix)] - #[test] - fn prepare_target_never_follows_symlinks_inside_root() { - let outside = tempfile::tempdir().unwrap(); - let root = tempfile::tempdir().unwrap(); - - // Pre-existing directory symlink at the first level. - symlink(outside.path(), root.path().join("safe")).unwrap(); - let err = prepare_target(root.path(), "safe/file.txt", false).unwrap_err(); - assert!(err.to_string().contains("symlink"), "{err}"); - assert!(!outside.path().join("file.txt").exists()); - - // Deeper: a real directory containing a symlinked directory. - fs::create_dir(root.path().join("real")).unwrap(); - symlink(outside.path(), root.path().join("real/link")).unwrap(); - assert!(prepare_target(root.path(), "real/link/f", false).is_err()); - - // A symlink at the target itself, with and without overwrite. - symlink(outside.path().join("victim"), root.path().join("t.txt")).unwrap(); - assert!(prepare_target(root.path(), "t.txt", false).is_err()); - assert!(prepare_target(root.path(), "t.txt", true).is_err()); - assert!(!outside.path().join("victim").exists()); - - // A directory swapped for a symlink after preparation is caught by - // the pre-rename check. - prepare_target(root.path(), "late/file", false).unwrap(); - fs::remove_dir(root.path().join("late")).unwrap(); - symlink(outside.path(), root.path().join("late")).unwrap(); - assert!(verify_parent_chain(root.path(), "late/file").is_err()); - assert!(verify_parent_chain(root.path(), "real/f").is_ok()); - } } diff --git a/ant-core/src/data/manifest/file.rs b/ant-core/src/data/manifest/file.rs index 4f04163d..d65f883b 100644 --- a/ant-core/src/data/manifest/file.rs +++ b/ant-core/src/data/manifest/file.rs @@ -21,6 +21,11 @@ pub fn manifest_filename_for(name: Option<&str>) -> String { /// Read and decode a `.ant` file, refusing oversized files before reading. pub fn read_manifest_file(path: &Path) -> Result { + Manifest::decode(&read_manifest_bytes(path)?) +} + +/// The bytes of a `.ant` file, refusing an oversized file before reading. +pub(super) fn read_manifest_bytes(path: &Path) -> Result, ManifestError> { let len = fs::metadata(path)?.len(); if len > MAX_MANIFEST_BYTES as u64 { return Err(ManifestError::TooLarge { @@ -28,7 +33,7 @@ pub fn read_manifest_file(path: &Path) -> Result { max: MAX_MANIFEST_BYTES, }); } - Manifest::decode(&fs::read(path)?) + Ok(fs::read(path)?) } /// Encode and write a manifest atomically. Refuses to replace an existing diff --git a/ant-core/src/data/manifest/history.rs b/ant-core/src/data/manifest/history.rs index a8e5f0c1..c077da0e 100644 --- a/ant-core/src/data/manifest/history.rs +++ b/ant-core/src/data/manifest/history.rs @@ -8,8 +8,10 @@ //! When an id is already taken, which happens when uploads with the same label //! finish within one second, `-2`, `-3` and so on are appended. Records list //! newest first, and within one second by that suffix compared as a number. -//! Nothing else is stored, so the history directory is just a folder of -//! manifests that any manifest tool can read. +//! Apart from the journal of a manifest build that has not finished +//! (`.ant.partial`, see the `journal` module), nothing else is stored, so +//! the history directory is just a folder of manifests that any manifest +//! tool can read. That journal is listed as an unfinished record. use std::fs; use std::path::{Path, PathBuf}; @@ -17,11 +19,16 @@ use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; -use super::file::{read_manifest_file, write_manifest_file}; -use super::{Manifest, ManifestError, MANIFEST_EXTENSION}; +use super::file::read_manifest_bytes; +use super::journal::{journal_path, read_journal, UploadJournal, JOURNAL_SUFFIX}; +use super::path::{ + check_collisions, portable_component, validate_component, validate_path, PathError, + PATH_SEPARATOR, +}; +use super::{Manifest, ManifestEntry, ManifestError, MANIFEST_EXTENSION}; #[cfg(test)] -use super::{ContentRef, ManifestEntry}; +use super::{ContentRef, MANIFEST_FORMAT_VERSION, MANIFEST_MAGIC}; /// Directory under the data directory that holds upload records. pub const UPLOAD_HISTORY_DIR: &str = "uploads"; @@ -78,14 +85,33 @@ const FEBRUARY: u32 = 2; pub struct UploadRecord { /// Filename stem: `-