From 10dbc18d0f281be4c6938ae1981d99cd2f6e2b84 Mon Sep 17 00:00:00 2001 From: Jamie Pine <32987599+jamiepine@users.noreply.github.com> Date: Mon, 5 Oct 2026 04:40:19 -0700 Subject: [PATCH] Replica fetch: resumable .part transfers, bandwidth cap, pause, and progress (#3109) * Resume interrupted replica fetches from a .part file A replica transfer now lands in ...part and stays there when the stream breaks or the daemon stops. The next attempt asks the owner for the artifact's tail from the part's length over the new FetchSnapshotFrom and FetchDatabaseFrom requests, whose header still names the whole artifact; a header that names different bytes discards the part and starts over, and the finished file is validated against the header checksum before it replaces the live replica as before. Parts are removed only when the owner's listing no longer carries their generation. A nonzero offset serves the on-disk snapshot without the pre-fetch save, so a resumed copy finishes against the same bytes. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Cap replica transfer bandwidth with a persisted daemon setting replication.max_bytes_per_sec in spacedrive.json bounds every replica artifact served and received through one token bucket (zero is unlimited); the cap applies to transfers already in flight. It is set through config.app.update and sd config set replication.max_bytes_per_sec (bytes, or K/M/G suffixes), read back through sd config get and show. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Pause and resume replica fetching sd sources replication pause stops this device starting replica transfers and halts any in flight at its next chunk, keeping the partial file; resume continues from it and runs a replication pass against every connected device at once. The switch is replication.paused in the daemon config, so a paused daemon stays paused across a restart, and a pass that runs while paused still refreshes listings, owner facts and fleet totals. A pause does not count as a fetch failure, so no backoff accrues. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Show replica fetch progress in sources list, status and events Every fetch in flight is tracked with bytes on disk, total and a recent rate. sources.list carries it on the replica's row (or a replica_fetching row for a first copy that has no share yet), core.status and the new mounts.replication_status query summarize it with the pause switch and cap, and the daemon emits a typed ReplicationProgress event about once a second per transfer plus a finished one when it stops. sd sources list and sd status print the figures; ts-client types regenerated. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Document resumable replica fetch, the cap, pause and progress Source replication gets its own section in networking.mdx with the .part contract, the wire change, the cap, the pause switch and the progress fields; file-sync.mdx points there; the replication plan marks its cleanup item superseded on the replica side and distinct from the restart refetch; PROJECT_STATUS.md gains the register row. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Keep the part when a resumed request gets no header; cap sidecar pages A resumed request that fails outright (dropped link, or an owner too old to know the offset variants) says nothing about the bytes on disk, so the part now stays and the whole-file retry lands on the same name. Sidecar pages count against the bandwidth cap on both ends, as the setting's contract says. Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Add the replication field to the test harness config Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> * Check the pause switch after the cap's wait, not before Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> --------- Co-authored-by: capy-ai-staging[bot] <305915985+capy-ai-staging[bot]@users.noreply.github.com> --- PROJECT_STATUS.md | 1 + apps/cli/src/domains/config/mod.rs | 97 +- apps/cli/src/domains/events/mod.rs | 20 + apps/cli/src/domains/source/mod.rs | 43 +- apps/cli/src/main.rs | 50 +- apps/cli/src/util/output.rs | 26 + core/src/config/app_config.rs | 21 + core/src/config/mod.rs | 3 +- core/src/infra/event/mod.rs | 7 + core/src/lib.rs | 1 + core/src/ops/config/app/get.rs | 16 + core/src/ops/config/app/update.rs | 15 +- core/src/ops/core/status/output.rs | 3 + core/src/ops/core/status/query.rs | 1 + core/src/ops/mounts/mod.rs | 4 + core/src/ops/mounts/replication_set_paused.rs | 78 ++ core/src/ops/mounts/replication_status.rs | 35 + core/src/ops/sources/list/output.rs | 38 + core/src/ops/sources/list/query.rs | 8 + core/src/service/mounts/mod.rs | 16 + core/src/service/mounts/peer.rs | 832 ++++++++++++++++-- core/src/service/mounts/replication.rs | 402 +++++++++ .../src/service/network/protocol/byterange.rs | 304 ++++--- core/src/testing/integration_utils.rs | 1 + core/tests/helpers/sync_harness.rs | 1 + docs/core/file-sync.mdx | 7 + docs/core/networking.mdx | 63 ++ .../2026-09-19-incremental-replication.md | 10 + packages/ts-client/src/generated/types.ts | 84 +- 29 files changed, 1971 insertions(+), 216 deletions(-) create mode 100644 core/src/ops/mounts/replication_set_paused.rs create mode 100644 core/src/ops/mounts/replication_status.rs create mode 100644 core/src/service/mounts/replication.rs diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md index e6da3662c..b258f2b3f 100644 --- a/PROJECT_STATUS.md +++ b/PROJECT_STATUS.md @@ -203,6 +203,7 @@ Recent landed. | Pairing bugs from the integration triage | fixed 2026-10-04 (Fable, SPAC-7); `file_copy_pull_test`, `watcher_test`, `sync_backfill_test` green on Linux | Fable | Three product bugs SPAC-6 located. Pull transfers always failed verification: the receiver hashed every byte with blake3 while the sender sent the 16-hex sampled content hash, and the push receiver never verified at all; `TransferComplete.final_checksum` is now the full blake3 of the file bytes in both directions (`ContentHashGenerator::generate_integrity_hash`, computed while streaming on the sender), compared on both receivers, with a corrupted file deleted and reported as `TransferError`. The Linux watcher turned a move out of the watched root into a Modify of a path that no longer existed: `RawEventKind::Rename` now carries inotify's From/To/Both, a lone MOVED_FROM becomes Remove and a lone MOVED_TO becomes Create after a 200 ms pairing window, and a buffered Modify is dropped when its path is removed. Two devices with the same hostname collided on `devices.slug` during sync apply (collision handling existed on insert only, not on update), so backfill failed every 5 s forever; the slug now gets a suffix from the device id on both paths, and three consecutive backfill failures log an error and pause sync instead of looping. `test_volume_resource_events_on_sync` now accepts the coarse `Refresh` that #3061 made backfill emit. Still red in `sd-fs-watcher --lib` and out of scope: `test_file_deletion_events` reads the write's buffered Modify before the Remove, and `a_shallow_watch_delivers_events_for_immediate_children` is filtered because `tempfile` dirs contain `.tmp`. Next: bug 4 (no reconnect after in-process restart) in its own thread; decide whether `core_tests.yml` runs on pull requests. | | Nightly CLI channel | in progress 2026-10-04 (Fable, SPAC-10); workflow, updater and docs in review | Fable | Every push to `sources` builds `sd` and `sd-daemon` (release, default features, macOS ARM64 and Linux x86-64) and republishes them on the rolling `nightly` prerelease with sha256 files, its tag moved to the built commit (`.github/workflows/nightly.yml`; the branch is one line in the trigger). `sd update` on `update.channel = nightly` fetches that release by tag and compares the release's target commit with the commit compiled into the binary (`SD_GIT_SHA` from `apps/cli/build.rs`, shown by `sd --version`); the stable channel still compares the latest release tag with the package version. The asset matcher strips the platform and expects exactly `sd` (or `sd-cli`) and `sd-daemon`, so the two no longer collide; downloads are checked against their `.sha256`; replacement writes a temp file and renames over the running binary; a launchd-managed daemon is restarted with `launchctl kickstart`. `sd update --yes` is the non-interactive path and `sd update install-timer` writes a launchd agent (systemd timer on Linux) that runs it hourly. Default `update.repo` is now `spacedriveapp/spacedrive`. Next: merge, watch the first real nightly run, opt both Macs in. | | `core_tests.yml` on pull requests | landed 2026-10-05 (Fable, SPAC-11); unit and integration jobs green on the first Blacksmith run | Fable | The workflow was dispatch-only with the two traps SPAC-4 removed from `ci.yml` (`maximize-build-space`, runner label in the `target` field). It now runs on push and pull requests against `sources` as two `blacksmith-4vcpu-ubuntu-2404` jobs: unit (`cargo xtask test-core --unit` plus `sd-fs-watcher --lib`, 12 min cold) and integration (`cargo xtask test-core --integration`, 17 suites, 49 min cold of which 46 min is the suites themselves; the six two-core suites take 4 to 6 min each by design). `xtask test-core` gained `--unit` and `--integration`. Each job has its own `Swatinem/rust-cache` key, `scripts/setup.sh` installs libav, bun 1.3.4 with `--frozen-lockfile` serves the TypeScript bridge suites, and the loop-device volume helper now reports a missing `losetup` device as the skip reason (Blacksmith grants passwordless sudo, so the volume suites ran for real). The two `sd-fs-watcher --lib` failures are fixed: `EventFilters::should_skip` matched `.tmp` anywhere in the path and hid every file under a `tempfile` dir, so temp and system filters now look at the file name only; `test_file_deletion_events` drains the file's events until Remove and asserts no Create slipped in, instead of assuming the write's buffered Modify arrives before the delete. Next: cut the integration wall time by splitting the two-core suites across jobs if 50 min proves too slow for the PR loop. | +| Replica fetch: resume, cap, pause, progress | in progress 2026-10-05 (Fable, SPAC-14); four commits in review against `sources` | Fable | Bugs 5 and 6 from SPAC-12: pairing the Studio with the MacBook pulled 2.9 GB of source stores at 200 KB/s over ship wifi, and stopping the daemon threw the partial away. A transfer now lands in `mounts-remote//...part` and survives failures and restarts; the next attempt resumes from its length over two new byterange requests, `FetchSnapshotFrom` and `FetchDatabaseFrom`, whose header still names the whole artifact, and the finished file is validated against the header hash before it replaces the live replica. A part is removed only when the owner's listing no longer carries its generation. `replication.max_bytes_per_sec` (daemon config, `sd config set`, zero unlimited) caps serving and receiving through one token bucket; `sd sources replication pause` / `resume` stops new fetches and halts in-flight ones with their parts kept, persisted in `replication.paused`. `sources.list` rows carry the in-flight transfer (bytes, total, rate), `core.status` and `mounts.replication_status` summarize, and a typed `ReplicationProgress` event is emitted about once a second. Both Macs must run a nightly with the `From` requests for resume to work; an older owner answers them with an error and the fetch starts from zero. Documented under Source replication in `docs/core/networking.mdx`; related to but distinct from P1 "restart refetch" in the replication plan. Next: live test on the Studio/MacBook pair with a cap set before reconnect. | | Library join over the network | in progress 2026-10-05 (Fable, SPAC-13); fixes and two new suites green on a Linux VM | Fable | SPAC-12 paired a fresh Mac Studio with the MacBook library and `join-remote` backfilled the wrong way: the member went `Uninitialized -> Backfilling{joiner}` and adopted the empty copy's snapshot (its Default space row was overwritten), the member's catch-up loop ran against the joiner every 6 to 15 s forever (no watermark is ever written for a peer with no device-owned rows), the reverse `RegisterDeviceRequest` timed out so the joiner never got the member's device row and so never found a sync partner, and `library info` on the joiner mixed peer totals with `Sources 0`. Fixes: a `sync_state` marker in `sync.db` says whether a library copy still waits for its first backfill, written only when a copy is created from another device's library, so every library that already holds state starts `Ready` and never requests a snapshot (`SharedChangeRequest.include_snapshot`, catch-up never escalates to a full backfill); per-peer catch-up completion is recorded in `sync.db` even when zero rows arrive and the loop no longer re-polls a peer on a 60 s timer; the registration round trip carries `needs_initial_state` and returns the responder's device record in the response (a copy still waiting for its own backfill refuses to seed a joiner; `sync_setup` fails instead of reporting `devices_registered: true` when the record is missing); connections a protocol dials now get a stream handler too (`TrackOutboundConnection` from `get_or_create_connection`), which is why the peer's request on a connection we opened used to hang 30 s; fleet totals in `library info` add only summaries of devices in the library's device table and include the peers' source counts. Tests: `library_join_test` (two daemons, real pairing and join) and `sync_catchup_test` (in-process). Docs: pairing, devices, CLI overview, library sync setup, library sync. Next: live check on the Macs (delete `My Library 1.sdlibrary` on the Studio and join again: MacBook stays `Ready`, Studio gets both device rows and the MacBook's spaces, no repeating "Never caught up with this peer" lines after the first). | ## Product sequence diff --git a/apps/cli/src/domains/config/mod.rs b/apps/cli/src/domains/config/mod.rs index 0bfabda14..b9d8c24cb 100644 --- a/apps/cli/src/domains/config/mod.rs +++ b/apps/cli/src/domains/config/mod.rs @@ -4,6 +4,10 @@ use comfy_table::{presets::UTF8_BORDERS_ONLY, Table}; use std::path::PathBuf; use crate::config::CliConfig; +use sd_client::CoreClient; +use sd_core::ops::config::app::{ + get::AppConfigOutput, GetAppConfigQueryInput, UpdateAppConfigInput, UpdateAppConfigOutput, +}; #[derive(Subcommand, Debug)] pub enum ConfigCmd { @@ -11,7 +15,8 @@ pub enum ConfigCmd { Show, /// Get a configuration value Get { - /// Configuration key (e.g., "update.repo", "update.channel") + /// Configuration key (e.g., "update.repo", "update.channel", + /// "replication.max_bytes_per_sec") key: String, }, /// Set a configuration value @@ -23,7 +28,39 @@ pub enum ConfigCmd { }, } -pub async fn run(data_dir: PathBuf, cmd: ConfigCmd) -> Result<()> { +/// Daemon settings read and written through the running daemon, so a +/// change applies to transfers already in flight and lands in +/// spacedrive.json rather than the CLI's own file. +const REPLICATION_MAX_BYTES_PER_SEC: &str = "replication.max_bytes_per_sec"; + +async fn daemon_config(socket_addr: &str) -> Result { + let core = CoreClient::new(socket_addr.to_string()); + core.query(&GetAppConfigQueryInput, None) + .await + .map_err(|e| anyhow::anyhow!("daemon config unavailable: {e}")) +} + +/// Parse a byte rate: a bare count of bytes, or a count with a K, M or G +/// suffix in powers of 1024, so `sd config set replication.max_bytes_per_sec 200K` +/// reads as 200 KiB/s. Zero lifts the cap. +fn parse_byte_rate(value: &str) -> Result { + let value = value.trim(); + let (digits, scale) = match value.chars().last() { + Some('k') | Some('K') => (&value[..value.len() - 1], 1u64 << 10), + Some('m') | Some('M') => (&value[..value.len() - 1], 1u64 << 20), + Some('g') | Some('G') => (&value[..value.len() - 1], 1u64 << 30), + _ => (value, 1), + }; + let count: u64 = digits + .trim() + .parse() + .map_err(|_| anyhow::anyhow!("'{value}' is not a byte rate (e.g. 0, 500000, 200K, 2M)"))?; + count + .checked_mul(scale) + .ok_or_else(|| anyhow::anyhow!("'{value}' is too large")) +} + +pub async fn run(data_dir: PathBuf, socket_addr: String, cmd: ConfigCmd) -> Result<()> { let mut config = CliConfig::load(&data_dir)?; match cmd { @@ -43,6 +80,14 @@ pub async fn run(data_dir: PathBuf, cmd: ConfigCmd) -> Result<()> { table.add_row(vec!["update.repo", &config.update.repo]); table.add_row(vec!["update.channel", &config.update.channel]); + // Daemon settings, when the daemon answers. + if let Ok(daemon) = daemon_config(&socket_addr).await { + table.add_row(vec![ + REPLICATION_MAX_BYTES_PER_SEC, + &format_rate(daemon.replication.max_bytes_per_sec), + ]); + } + println!("{}", table); println!(); println!( @@ -58,11 +103,37 @@ pub async fn run(data_dir: PathBuf, cmd: ConfigCmd) -> Result<()> { .unwrap_or_else(|| "(not set)".to_string()), "update.repo" => config.update.repo.clone(), "update.channel" => config.update.channel.clone(), + REPLICATION_MAX_BYTES_PER_SEC => daemon_config(&socket_addr) + .await? + .replication + .max_bytes_per_sec + .to_string(), _ => return Err(anyhow::anyhow!("Unknown config key: {}", key)), }; println!("{}", value); } ConfigCmd::Set { key, value } => match key.as_str() { + REPLICATION_MAX_BYTES_PER_SEC => { + let rate = parse_byte_rate(&value)?; + let core = CoreClient::new(socket_addr.clone()); + let input = UpdateAppConfigInput { + replication_max_bytes_per_sec: Some(rate), + ..UpdateAppConfigInput::default() + }; + let out: UpdateAppConfigOutput = serde_json::from_value( + core.action(&input, None) + .await + .map_err(|e| anyhow::anyhow!("daemon refused the change: {e}"))?, + )?; + if !out.success { + return Err(anyhow::anyhow!(out.message)); + } + println!( + "Set {} = {}", + REPLICATION_MAX_BYTES_PER_SEC, + format_rate(rate) + ); + } "update.repo" => { config.set_update_repo(value.clone(), &data_dir)?; println!("Set update.repo = {}", value); @@ -77,3 +148,25 @@ pub async fn run(data_dir: PathBuf, cmd: ConfigCmd) -> Result<()> { Ok(()) } + +fn format_rate(bytes_per_sec: u64) -> String { + if bytes_per_sec == 0 { + return "0 (unlimited)".to_string(); + } + format!("{bytes_per_sec} B/s") +} + +#[cfg(test)] +mod tests { + use super::parse_byte_rate; + + #[test] + fn byte_rates_take_binary_suffixes() { + assert_eq!(parse_byte_rate("0").unwrap(), 0); + assert_eq!(parse_byte_rate("500000").unwrap(), 500_000); + assert_eq!(parse_byte_rate("200K").unwrap(), 200 << 10); + assert_eq!(parse_byte_rate("2m").unwrap(), 2 << 20); + assert_eq!(parse_byte_rate("1G").unwrap(), 1 << 30); + assert!(parse_byte_rate("fast").is_err()); + } +} diff --git a/apps/cli/src/domains/events/mod.rs b/apps/cli/src/domains/events/mod.rs index e691ac44c..490fa3e24 100644 --- a/apps/cli/src/domains/events/mod.rs +++ b/apps/cli/src/domains/events/mod.rs @@ -366,6 +366,26 @@ fn summarize_event(event: &Event) -> String { // Config events Event::ConfigChanged { .. } => "Configuration changed".to_string(), + Event::ReplicationProgress { transfer } => format!( + "Replica {} from {}: {}{}", + transfer.root.display(), + transfer.device_label, + crate::util::output::format_transfer( + transfer.bytes, + transfer.total, + transfer.bytes_per_sec + ), + if transfer.finished { + if transfer.bytes == transfer.total { + " (complete)" + } else { + " (stopped)" + } + } else { + "" + } + ), + // Custom events Event::Custom { event_type, data } => { format!("Custom event: {} - {:?}", event_type, data) diff --git a/apps/cli/src/domains/source/mod.rs b/apps/cli/src/domains/source/mod.rs index df47e619c..e8ae63e05 100644 --- a/apps/cli/src/domains/source/mod.rs +++ b/apps/cli/src/domains/source/mod.rs @@ -6,6 +6,7 @@ use std::path::PathBuf; use crate::util::prelude::*; use crate::context::Context; +use sd_core::ops::mounts::{MountsReplicationSetPausedInput, MountsReplicationSetPausedOutput}; use sd_core::ops::sources::{ freeze::action::{FreezeSourceInput, FreezeSourceOutput}, list::{output::SourceInfo, query::ListSourcesInput}, @@ -27,6 +28,18 @@ pub enum SourceCmd { Verify(SourceVerifyArgs), /// Rename a source or change its capture policy Update(SourceUpdateArgs), + /// Control how this device copies paired devices' source indexes + #[command(subcommand)] + Replication(ReplicationCmd), +} + +#[derive(Subcommand, Debug)] +pub enum ReplicationCmd { + /// Stop fetching replicas: no new transfer starts and any in flight + /// stops, keeping its partial file. Persists across daemon restarts. + Pause, + /// Fetch replicas again, continuing partial transfers where they stopped + Resume, } #[derive(Args, Debug)] @@ -101,16 +114,27 @@ pub async fn run(ctx: &Context, cmd: SourceCmd) -> Result<()> { table.load_preset(UTF8_BORDERS_ONLY); table.set_header(vec!["ID", "Name", "Type", "Records", "Status", "Root"]); for source in sources { + let mut status = if source.attached { + source.status.clone() + } else { + format!("{} (detached)", source.status) + }; + if let Some(transfer) = &source.transfer { + status = format!( + "{status}: fetching {}", + crate::util::output::format_transfer( + transfer.bytes, + transfer.total, + transfer.bytes_per_sec + ) + ); + } table.add_row(vec![ source.id.to_string(), source.name.clone(), source.data_type.clone(), source.item_count.to_string(), - if source.attached { - source.status.clone() - } else { - format!("{} (detached)", source.status) - }, + status, source.root.clone().unwrap_or_default(), ]); } @@ -153,6 +177,15 @@ pub async fn run(ctx: &Context, cmd: SourceCmd) -> Result<()> { ); }); } + SourceCmd::Replication(cmd) => { + let input = MountsReplicationSetPausedInput { + paused: matches!(cmd, ReplicationCmd::Pause), + }; + let out: MountsReplicationSetPausedOutput = execute_core_action!(ctx, input); + print_output!(ctx, &out, |o: &MountsReplicationSetPausedOutput| { + println!("{}", o.message); + }); + } SourceCmd::Freeze(args) => { let input = FreezeSourceInput { source_id: args.source_id, diff --git a/apps/cli/src/main.rs b/apps/cli/src/main.rs index e5899f76c..51355a493 100644 --- a/apps/cli/src/main.rs +++ b/apps/cli/src/main.rs @@ -8,22 +8,7 @@ use sd_client::{ }; use std::path::Path; -fn format_bytes(bytes: u64) -> String { - const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"]; - let mut size = bytes as f64; - let mut unit_index = 0; - - while size >= 1024.0 && unit_index < UNITS.len() - 1 { - size /= 1024.0; - unit_index += 1; - } - - if unit_index == 0 { - format!("{} {}", bytes, UNITS[unit_index]) - } else { - format!("{:.1} {}", size, UNITS[unit_index]) - } -} +use crate::util::output::{format_bytes, format_transfer}; /// Validate instance name to prevent path traversal attacks fn validate_instance_name(instance: &str) -> Result<(), String> { @@ -486,8 +471,9 @@ async fn main() -> Result<()> { } } Commands::Config(cmd) => { - // Config management doesn't need the client - config_cmd::run(data_dir, cmd).await?; + // CLI settings need no client; daemon settings reach the daemon + // only for the keys that live there. + config_cmd::run(data_dir, socket_addr, cmd).await?; } Commands::Daemon(cmd) => { // Daemon management doesn't need the client, handle directly @@ -724,6 +710,34 @@ async fn run_client_command( "○ Stopped" }; services_table.add_row(vec!["File Sharing", share_status]); + let replication = &status.replication; + let replication_status = if replication.paused { + "○ Paused".to_string() + } else if replication.transfers.is_empty() { + "● Idle".to_string() + } else { + format!( + "● Replicating {} source{}, {}", + replication.transfers.len(), + if replication.transfers.len() == 1 { + "" + } else { + "s" + }, + format_transfer( + replication.bytes(), + replication.total(), + replication.bytes_per_sec() + ) + ) + }; + services_table.add_row(vec!["Replication", &replication_status]); + if replication.max_bytes_per_sec > 0 { + services_table.add_row(vec![ + " Bandwidth cap", + &format!("{}/s", format_bytes(replication.max_bytes_per_sec)), + ]); + } println!("{}", services_table); println!(); diff --git a/apps/cli/src/util/output.rs b/apps/cli/src/util/output.rs index 523af607b..b0586636a 100644 --- a/apps/cli/src/util/output.rs +++ b/apps/cli/src/util/output.rs @@ -3,3 +3,29 @@ use serde::Serialize; pub fn print_json(data: &T) { println!("{}", serde_json::to_string_pretty(data).unwrap()); } + +/// Bytes in binary units, as `sd status` and `sd sources list` print them. +pub fn format_bytes(bytes: u64) -> String { + const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"]; + let mut size = bytes as f64; + let mut unit_index = 0; + while size >= 1024.0 && unit_index < UNITS.len() - 1 { + size /= 1024.0; + unit_index += 1; + } + if unit_index == 0 { + format!("{} {}", bytes, UNITS[unit_index]) + } else { + format!("{:.1} {}", size, UNITS[unit_index]) + } +} + +/// One replica fetch as a phrase: `553.0 MB of 2.9 GB, 212.0 KB/s`. +pub fn format_transfer(bytes: u64, total: u64, bytes_per_sec: u64) -> String { + format!( + "{} of {}, {}/s", + format_bytes(bytes), + format_bytes(total), + format_bytes(bytes_per_sec) + ) +} diff --git a/core/src/config/app_config.rs b/core/src/config/app_config.rs index 4bcdf2135..c7a49a88a 100644 --- a/core/src/config/app_config.rs +++ b/core/src/config/app_config.rs @@ -45,6 +45,26 @@ pub struct AppConfig { /// Mounted-source behaviour #[serde(default)] pub mounts: MountsConfig, + + /// Replica transfers to and from paired devices + #[serde(default)] + pub replication: ReplicationConfig, +} + +/// Settings for the source replicas exchanged with paired devices. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplicationConfig { + /// Ceiling on replica bytes moved per second, serving and fetching + /// combined. Zero is unlimited. An initial replica runs to gigabytes + /// and starts on its own when a device connects, so a metered link + /// needs this set before pairing. + #[serde(default)] + pub max_bytes_per_sec: u64, + /// Whether this device fetches replicas at all. Paused, it starts no + /// transfer and stops any in flight, keeping the partial file; it still + /// serves its own sources to peers. + #[serde(default)] + pub paused: bool, } /// Settings for sources served as a mounted drive. @@ -268,6 +288,7 @@ impl AppConfig { logging: LoggingConfig::default(), proxy_pairing: ProxyPairingConfig::default(), mounts: MountsConfig::default(), + replication: ReplicationConfig::default(), } } diff --git a/core/src/config/mod.rs b/core/src/config/mod.rs index 17b8baf4e..f87f351ac 100644 --- a/core/src/config/mod.rs +++ b/core/src/config/mod.rs @@ -11,7 +11,8 @@ pub mod app_config; pub mod migration; pub use app_config::{ - AppConfig, JobLoggingConfig, LogStreamConfig, LoggingConfig, ProxyPairingConfig, ServiceConfig, + AppConfig, JobLoggingConfig, LogStreamConfig, LoggingConfig, ProxyPairingConfig, + ReplicationConfig, ServiceConfig, }; pub use migration::Migrate; diff --git a/core/src/infra/event/mod.rs b/core/src/infra/event/mod.rs index 8989af88d..e1936a1af 100644 --- a/core/src/infra/event/mod.rs +++ b/core/src/infra/event/mod.rs @@ -288,6 +288,13 @@ pub enum Event { field: String, }, + /// A replica fetch from a paired device moved, about once a second per + /// transfer, with a last emission marked `finished` when it leaves the + /// active set, complete or stopped. + ReplicationProgress { + transfer: crate::service::mounts::replication::ReplicaTransferProgress, + }, + // Custom events for extensibility Custom { event_type: String, diff --git a/core/src/lib.rs b/core/src/lib.rs index 06fa945ce..b2b82d748 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -265,6 +265,7 @@ impl Core { } // Serve indexed sources as mountable shares. + crate::service::mounts::replication::configure(&config.read().await.replication); let cache_max_bytes = config.read().await.mounts.cache_max_bytes; if let Err(e) = crate::service::mounts::start(context.clone(), cache_max_bytes).await { error!("Failed to start mounts share: {}", e); diff --git a/core/src/ops/config/app/get.rs b/core/src/ops/config/app/get.rs index 4014341d5..ac4296da9 100644 --- a/core/src/ops/config/app/get.rs +++ b/core/src/ops/config/app/get.rs @@ -42,6 +42,9 @@ pub struct AppConfigOutput { /// Mounted-source behaviour pub mounts: MountsConfigOutput, + /// Replica transfers with paired devices + pub replication: ReplicationConfigOutput, + /// Daemon logging configuration pub logging: LoggingConfigOutput, @@ -81,6 +84,15 @@ pub struct MountsConfigOutput { pub cache_max_bytes: u64, } +/// Replication configuration output +#[derive(Debug, Clone, Serialize, Deserialize, Type)] +pub struct ReplicationConfigOutput { + /// Bytes per second across every replica transfer; zero is unlimited. + pub max_bytes_per_sec: u64, + /// Whether this device has stopped fetching replicas. + pub paused: bool, +} + /// Logging configuration output #[derive(Debug, Clone, Serialize, Deserialize, Type)] pub struct LoggingConfigOutput { @@ -118,6 +130,10 @@ impl From<&AppConfig> for AppConfigOutput { mounts: MountsConfigOutput { cache_max_bytes: config.mounts.cache_max_bytes, }, + replication: ReplicationConfigOutput { + max_bytes_per_sec: config.replication.max_bytes_per_sec, + paused: config.replication.paused, + }, services: ServiceConfigOutput { networking_enabled: config.services.networking_enabled, volume_monitoring_enabled: config.services.volume_monitoring_enabled, diff --git a/core/src/ops/config/app/update.rs b/core/src/ops/config/app/update.rs index 41985901c..d30ad4b87 100644 --- a/core/src/ops/config/app/update.rs +++ b/core/src/ops/config/app/update.rs @@ -14,7 +14,7 @@ use crate::{ /// Input for updating app configuration /// All fields are optional for partial updates -#[derive(Debug, Clone, Serialize, Deserialize, Type)] +#[derive(Debug, Clone, Default, Serialize, Deserialize, Type)] pub struct UpdateAppConfigInput { /// Whether telemetry is enabled #[serde(skip_serializing_if = "Option::is_none")] @@ -38,6 +38,11 @@ pub struct UpdateAppConfigInput { #[serde(skip_serializing_if = "Option::is_none")] pub mounts_cache_max_bytes: Option, + /// Ceiling on replica bytes per second, serving and fetching combined; + /// zero lifts it. Applies to transfers already in flight. + #[serde(skip_serializing_if = "Option::is_none")] + pub replication_max_bytes_per_sec: Option, + /// Whether networking is enabled #[serde(skip_serializing_if = "Option::is_none")] pub networking_enabled: Option, @@ -199,6 +204,14 @@ impl CoreAction for UpdateAppConfigAction { } } + if let Some(max_bytes_per_sec) = self.input.replication_max_bytes_per_sec { + if config.replication.max_bytes_per_sec != max_bytes_per_sec { + config.replication.max_bytes_per_sec = max_bytes_per_sec; + changes.push("replication_max_bytes_per_sec"); + crate::service::mounts::replication::configure(&config.replication); + } + } + if let Some(networking_enabled) = self.input.networking_enabled { if config.services.networking_enabled != networking_enabled { config.services.networking_enabled = networking_enabled; diff --git a/core/src/ops/core/status/output.rs b/core/src/ops/core/status/output.rs index d8dbb81f7..9960e7eba 100644 --- a/core/src/ops/core/status/output.rs +++ b/core/src/ops/core/status/output.rs @@ -17,6 +17,9 @@ pub struct CoreStatus { pub services: ServiceStatus, pub network: NetworkStatus, pub system: SystemInfo, + /// Replica fetches from paired devices: the pause switch, the cap and + /// every transfer in flight. + pub replication: crate::service::mounts::replication::ReplicationStatus, } #[derive(Debug, Clone, Serialize, Deserialize, Type)] diff --git a/core/src/ops/core/status/query.rs b/core/src/ops/core/status/query.rs index 360f2e4e4..5a6689950 100644 --- a/core/src/ops/core/status/query.rs +++ b/core/src/ops/core/status/query.rs @@ -153,6 +153,7 @@ impl CoreQuery for CoreStatusQuery { services, network: network_status, system, + replication: crate::service::mounts::replication::status(), }) } } diff --git a/core/src/ops/mounts/mod.rs b/core/src/ops/mounts/mod.rs index 4fb053eb3..c327cfdc8 100644 --- a/core/src/ops/mounts/mod.rs +++ b/core/src/ops/mounts/mod.rs @@ -4,6 +4,8 @@ pub mod cache_clear; pub mod cache_status; pub mod read_trace; +pub mod replication_set_paused; +pub mod replication_status; pub mod status; pub mod sync_peers; pub mod trace_set; @@ -11,6 +13,8 @@ pub mod trace_set; pub use cache_clear::*; pub use cache_status::*; pub use read_trace::*; +pub use replication_set_paused::*; +pub use replication_status::*; pub use status::*; pub use sync_peers::*; pub use trace_set::*; diff --git a/core/src/ops/mounts/replication_set_paused.rs b/core/src/ops/mounts/replication_set_paused.rs new file mode 100644 index 000000000..187473a0d --- /dev/null +++ b/core/src/ops/mounts/replication_set_paused.rs @@ -0,0 +1,78 @@ +//! Pause and resume replica fetching on this device. +//! +//! Paused, the daemon starts no replica transfer and stops any in flight at +//! its next chunk, keeping the partial file so a resume continues from it. +//! Listings, owner facts and fleet totals still refresh, and the device +//! still serves its own sources. The switch is persisted in the daemon +//! config, so a paused daemon stays paused across a restart. + +use crate::{ + config::AppConfig, + infra::action::{error::ActionError, CoreAction}, + service::mounts::replication, +}; +use serde::{Deserialize, Serialize}; +use specta::Type; +use std::sync::Arc; + +#[derive(Debug, Clone, Serialize, Deserialize, Type, Default)] +pub struct MountsReplicationSetPausedInput { + pub paused: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Type)] +pub struct MountsReplicationSetPausedOutput { + pub paused: bool, + pub message: String, +} + +pub struct MountsReplicationSetPausedAction { + input: MountsReplicationSetPausedInput, +} + +impl CoreAction for MountsReplicationSetPausedAction { + type Input = MountsReplicationSetPausedInput; + type Output = MountsReplicationSetPausedOutput; + + fn from_input(input: Self::Input) -> std::result::Result { + Ok(Self { input }) + } + + async fn execute( + self, + context: Arc, + ) -> std::result::Result { + let paused = self.input.paused; + let mut config = AppConfig::load_from(&context.data_dir) + .map_err(|e| ActionError::Internal(format!("Failed to load config: {e}")))?; + if config.replication.paused != paused { + config.replication.paused = paused; + config + .save() + .map_err(|e| ActionError::Internal(format!("Failed to save config: {e}")))?; + } + replication::set_paused(paused); + if !paused { + crate::service::mounts::resync_connected(context); + } + Ok(MountsReplicationSetPausedOutput { + paused, + message: if paused { + "replication paused; transfers in flight stop and keep their partial files" + .to_string() + } else { + "replication resumed; deferred transfers continue from their partial files" + .to_string() + }, + }) + } + + fn action_kind(&self) -> &'static str { + "mounts.replication_set_paused" + } +} + +crate::register_core_action!( + MountsReplicationSetPausedAction, + "mounts.replication_set_paused" +); diff --git a/core/src/ops/mounts/replication_status.rs b/core/src/ops/mounts/replication_status.rs new file mode 100644 index 000000000..2f9450ee2 --- /dev/null +++ b/core/src/ops/mounts/replication_status.rs @@ -0,0 +1,35 @@ +//! What replication is doing right now: paused or not, the cap, and every +//! replica fetch in flight with bytes, total and rate. + +use crate::{ + infra::query::{CoreQuery, QueryResult}, + service::mounts::replication::{self, ReplicationStatus}, +}; +use serde::{Deserialize, Serialize}; +use specta::Type; +use std::sync::Arc; + +#[derive(Debug, Clone, Serialize, Deserialize, Type, Default)] +pub struct MountsReplicationStatusInput {} + +#[derive(Debug, Clone, Serialize, Deserialize, Type)] +pub struct MountsReplicationStatusQuery; + +impl CoreQuery for MountsReplicationStatusQuery { + type Input = MountsReplicationStatusInput; + type Output = ReplicationStatus; + + fn from_input(_input: Self::Input) -> QueryResult { + Ok(Self) + } + + async fn execute( + self, + _context: Arc, + _session: crate::infra::api::SessionContext, + ) -> QueryResult { + Ok(replication::status()) + } +} + +crate::register_core_query!(MountsReplicationStatusQuery, "mounts.replication_status"); diff --git a/core/src/ops/sources/list/output.rs b/core/src/ops/sources/list/output.rs index cf618a841..739da9d53 100644 --- a/core/src/ops/sources/list/output.rs +++ b/core/src/ops/sources/list/output.rs @@ -42,6 +42,11 @@ pub struct SourceInfo { pub device_id: Option, /// The owning device's display name, for a replica. pub device_label: Option, + /// The fetch bringing this replica up to date, while one is in flight. + /// Present on a replica being refreshed and on a source whose first + /// copy has not finished, which has no other row to appear in. + #[serde(default)] + pub transfer: Option, } impl SourceInfo { @@ -88,6 +93,7 @@ impl SourceInfo { last_seen_at, device_id: None, device_label: None, + transfer: None, } } @@ -123,6 +129,37 @@ impl SourceInfo { last_seen_at: synced_at, device_id: Some(device_id), device_label: Some(device_label.to_string()), + transfer: crate::service::mounts::replication::transfer(entry.info.id), + } + } + + /// A source whose first replica is still arriving. Nothing else lists it + /// yet: there is no share to browse and no manifest entry until the + /// artifact validates, but the transfer itself is worth seeing. + pub fn from_transfer( + transfer: &crate::service::mounts::replication::ReplicaTransferProgress, + ) -> Self { + let name = transfer + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| transfer.source_id.to_string()); + Self { + id: transfer.source_id, + name, + data_type: "filesystem".to_string(), + adapter_id: None, + item_count: 0, + last_synced: None, + status: "replica_fetching".to_string(), + root: Some(transfer.root.to_string_lossy().into_owned()), + volume_uuid: None, + attached: false, + total_bytes: None, + last_seen_at: None, + device_id: Some(transfer.device_id), + device_label: Some(transfer.device_label.clone()), + transfer: Some(transfer.clone()), } } @@ -154,6 +191,7 @@ impl SourceInfo { last_seen_at: synced_at, device_id: Some(share.device_id), device_label: Some(share.device_label.clone()), + transfer: crate::service::mounts::replication::transfer(share.info.id), } } } diff --git a/core/src/ops/sources/list/query.rs b/core/src/ops/sources/list/query.rs index 876b0f6ec..cff8746a2 100644 --- a/core/src/ops/sources/list/query.rs +++ b/core/src/ops/sources/list/query.rs @@ -130,6 +130,14 @@ impl LibraryQuery for ListSourcesQuery { &entry, )); } + + // A first copy still in flight has no share and no manifest + // entry; its row is the transfer. + for transfer in crate::service::mounts::replication::transfers() { + if !sources.iter().any(|source| source.id == transfer.source_id) { + sources.push(SourceInfo::from_transfer(&transfer)); + } + } } Ok(sources) diff --git a/core/src/service/mounts/mod.rs b/core/src/service/mounts/mod.rs index bfd497ef9..1457cea18 100644 --- a/core/src/service/mounts/mod.rs +++ b/core/src/service/mounts/mod.rs @@ -16,6 +16,7 @@ pub mod attach; pub mod cache; pub mod peer; pub mod provider; +pub mod replication; pub mod smb; pub mod trace; mod webdav; @@ -178,6 +179,21 @@ async fn watch_peers(context: Arc) { } } +/// Run a replication pass against every connected device now, outside the +/// reconnect debounce. Resuming from a pause calls this so deferred +/// transfers start without waiting for the next refresh tick. +pub fn resync_connected(context: Arc) { + tokio::spawn(async move { + let Some(networking) = context.networking.read().await.clone() else { + return; + }; + for (device_id, label) in connected_devices(&networking).await { + peer::allow_next_sync(device_id); + spawn_sync(&context, device_id, label); + } + }); +} + async fn connected_devices( networking: &Arc, ) -> Vec<(uuid::Uuid, String)> { diff --git a/core/src/service/mounts/peer.rs b/core/src/service/mounts/peer.rs index 962d2624a..9e858cc23 100644 --- a/core/src/service/mounts/peer.rs +++ b/core/src/service/mounts/peer.rs @@ -266,6 +266,7 @@ async fn restore_from(base: &std::path::Path) -> (usize, usize) { while let Ok(Some(dir)) = dirs.next_entry().await { let replica_dir = dir.path(); + sweep_legacy_temporaries(&replica_dir).await; let manifest = match tokio::fs::read(manifest_path(&replica_dir)).await { Ok(bytes) => match serde_json::from_slice::(&bytes) { Ok(manifest) => manifest, @@ -323,6 +324,20 @@ async fn restore_from(base: &std::path::Path) -> (usize, usize) { (loaded, known) } +/// Builds before resumable transfers wrote to `.tmp`, which nothing +/// can continue; a `.part` file is kept, since the next sync decides from its +/// name whether the owner still serves those bytes. +async fn sweep_legacy_temporaries(replica_dir: &Path) { + let Ok(mut dir) = tokio::fs::read_dir(replica_dir).await else { + return; + }; + while let Ok(Some(entry)) = dir.next_entry().await { + if entry.path().extension().is_some_and(|ext| ext == "tmp") { + let _ = tokio::fs::remove_file(entry.path()).await; + } + } +} + /// Load a replica's artifact into an arena: the snapshot when one exists, /// else a delivered database rebuilt entry by entry. Either way the share /// serves identically; the artifact kind is a transport detail. @@ -626,6 +641,7 @@ pub async fn fetch_sidecars( anyhow::bail!("expected a sidecar"); }; anyhow::ensure!(len <= MAX_TILE_LEN, "sidecar of {len} bytes exceeds limit"); + super::replication::throttle().acquire(len).await; let mut webp = vec![0; len as usize]; body.read_exact(&mut webp).await?; page.push(SidecarRow { @@ -700,6 +716,9 @@ async fn pull_sidecars( }; let mut copied = 0; loop { + if super::replication::paused() { + anyhow::bail!("replication paused after {copied} sidecars"); + } let (store, rows) = fetch_sidecars(context, device_id, info.id, after, MAX_SIDECAR_PAGE).await?; let count = rows.len(); @@ -838,31 +857,312 @@ fn refresh_share_facts( })) } -/// Stream a declared number of bytes into a temporary artifact, hashing as -/// they land. A short stream or a checksum mismatch fails the transfer with -/// the temporary file still in place for the caller to remove; nothing here -/// touches the published artifact. +/// A transfer in progress on disk, named +/// `...part`. The name carries what the +/// next header must match for the bytes to be worth keeping, so a restart +/// decides without the owner and without a sidecar file. +#[derive(Debug, Clone, PartialEq)] +struct PartFile { + path: PathBuf, + generation: u64, + prefix: [u8; 8], + have: u64, +} + +fn part_path(replica_dir: &Path, source_id: Uuid, generation: u64, checksum: &[u8; 32]) -> PathBuf { + replica_dir.join(format!( + "{}.{generation}.{}.part", + source_id.simple(), + hex::encode(&checksum[..8]) + )) +} + +/// The source, generation and checksum prefix a `.part` file name encodes. +fn parse_part_name(name: &str) -> Option<(Uuid, u64, [u8; 8])> { + let rest = name.strip_suffix(".part")?; + let (source, rest) = rest.split_once('.')?; + let (generation, prefix) = rest.split_once('.')?; + let source_id = Uuid::try_parse(source).ok()?; + let generation = generation.parse().ok()?; + let prefix: [u8; 8] = hex::decode(prefix).ok()?.try_into().ok()?; + Some((source_id, generation, prefix)) +} + +/// The partial transfer on disk for `source_id`, if any. Two attempts can +/// only leave two parts when the owner's generation moved between them, so +/// everything but the most recently written one is removed. +async fn find_part(replica_dir: &Path, source_id: Uuid) -> Option { + let mut dir = tokio::fs::read_dir(replica_dir).await.ok()?; + let mut found: Option<(PartFile, SystemTime)> = None; + while let Ok(Some(entry)) = dir.next_entry().await { + let name = entry.file_name(); + let Some((id, generation, prefix)) = name.to_str().and_then(parse_part_name) else { + continue; + }; + if id != source_id { + continue; + } + let Ok(meta) = entry.metadata().await else { + continue; + }; + let part = PartFile { + path: entry.path(), + generation, + prefix, + have: meta.len(), + }; + let modified = meta.modified().unwrap_or(UNIX_EPOCH); + match &found { + Some((_, at)) if *at >= modified => { + let _ = tokio::fs::remove_file(&part.path).await; + } + Some((older, _)) => { + let _ = tokio::fs::remove_file(&older.path).await; + found = Some((part, modified)); + } + None => found = Some((part, modified)), + } + } + found.map(|(part, _)| part) +} + +/// Drop partial transfers the owner's listing no longer describes: a part +/// whose source is gone, or whose generation is not the one the owner holds +/// now, can never complete against the bytes the owner would serve. A part +/// that still matches stays, however long the owner has been away. +async fn prune_parts(replica_dir: &Path, sources: &[RemoteSourceInfo]) { + let Ok(mut dir) = tokio::fs::read_dir(replica_dir).await else { + return; + }; + while let Ok(Some(entry)) = dir.next_entry().await { + let name = entry.file_name(); + let Some((source_id, generation, _)) = name.to_str().and_then(parse_part_name) else { + continue; + }; + let current = sources + .iter() + .find(|info| info.id == source_id) + .is_some_and(|info| info.generation == generation); + if !current { + tracing::debug!( + source = %source_id, + generation, + "partial transfer no longer matches the owner's generation; removed" + ); + let _ = tokio::fs::remove_file(entry.path()).await; + } + } +} + +/// What an owner answers an artifact request with: the whole artifact's +/// identity, whatever offset the bytes behind it start at. +#[derive(Debug, Clone, Copy, PartialEq)] +struct ArtifactHeader { + len: u64, + generation: u64, + checksum: [u8; 32], +} + +impl ArtifactHeader { + /// Whether a partial file was cut from these same bytes. An owner that + /// predates header identity names nothing, so nothing of its resumes. + fn continues(&self, part: &PartFile) -> bool { + self.generation != 0 + && self.checksum != [0u8; 32] + && part.generation == self.generation + && part.prefix[..] == self.checksum[..8] + && part.have <= self.len + } +} + +/// Why a fetch did not publish. A request the owner refused or could not +/// open moved no bytes, so the caller may try another artifact kind; a +/// transfer that failed partway has its partial file on disk and should +/// simply be retried. +#[derive(Debug)] +enum FetchError { + Unavailable(anyhow::Error), + Failed(anyhow::Error), +} + +impl std::fmt::Display for FetchError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + FetchError::Unavailable(err) | FetchError::Failed(err) => write!(f, "{err}"), + } + } +} + +type ArtifactBody = Box; + +/// Bring one artifact's bytes onto disk, continuing a `.part` from an +/// earlier attempt when the owner still serves the same bytes. +/// +/// `open(offset)` asks the owner for the artifact's tail from `offset`; the +/// header it returns describes the whole artifact. A part the header does +/// not continue is discarded and the fetch starts over from zero. A resumed +/// request that gets no header at all (a dropped link, or an owner too old +/// to know the offset variants) also starts over, but keeps the part, since +/// nothing has said its bytes are wrong. Every other failure leaves the +/// partial file in place for the next attempt. On success the file holds +/// every byte the header declared and matches its checksum. +async fn transfer_artifact( + replica_dir: &Path, + owner: &TransferOwner<'_>, + mut open: F, +) -> Result<(PathBuf, ArtifactHeader), FetchError> +where + F: FnMut(u64) -> Fut, + Fut: std::future::Future>, +{ + let source_id = owner.info.id; + let mut part = find_part(replica_dir, source_id).await; + let mut opened = match &part { + Some(existing) => match open(existing.have).await { + Ok(opened) => Some(opened), + Err(err) => { + tracing::debug!(source = %source_id, %err, "resumed request refused; starting over"); + None + } + }, + None => None, + }; + if let Some(existing) = part.take() { + match &opened { + Some((header, _)) if header.continues(&existing) => part = Some(existing), + Some(_) => { + tracing::info!( + source = %source_id, + had = existing.have, + "partial transfer does not match what the owner serves now; starting over" + ); + let _ = tokio::fs::remove_file(&existing.path).await; + opened = None; + } + // A refused resume says nothing about the bytes: the link may + // have dropped, or the owner may predate the offset requests. + // The part stays for the next attempt; this one starts a whole + // transfer, which lands on the same name when the header has + // not moved and supersedes the old part when it has. + None => {} + } + } + let (header, mut body) = match opened { + Some(opened) => opened, + None => open(0).await.map_err(FetchError::Unavailable)?, + }; + if header.len > MAX_SNAPSHOT_LEN { + return Err(FetchError::Unavailable(anyhow::anyhow!( + "declared artifact of {} bytes exceeds the transfer bound", + header.len + ))); + } + + let (path, offset) = match part { + Some(part) => (part.path, part.have), + None => ( + part_path(replica_dir, source_id, header.generation, &header.checksum), + 0, + ), + }; + let reporter = super::replication::TransferReporter::start( + owner.events.clone(), + owner.device_id, + owner.device_label, + owner.info, + offset, + header.len, + ); + receive_artifact( + &mut body, + &path, + offset, + header.len, + header.checksum, + &reporter, + ) + .await + .map_err(FetchError::Failed)?; + Ok((path, header)) +} + +/// Who a transfer is from and about, for progress reporting. The event bus +/// is optional so the transfer logic tests without a core. +struct TransferOwner<'a> { + events: Option>, + device_id: Uuid, + device_label: &'a str, + info: &'a RemoteSourceInfo, +} + +/// Append a stream to a partial artifact from `offset` to `len`, hashing the +/// bytes already on disk and the new ones alike. A short stream fails the +/// transfer with the file in place for the next attempt; a checksum +/// mismatch removes it, since no continuation can mend it. async fn receive_artifact( body: &mut R, - tmp_path: &std::path::Path, + part_path: &Path, + offset: u64, len: u64, expected_checksum: [u8; 32], + reporter: &super::replication::TransferReporter, ) -> anyhow::Result<()> { use tokio::io::AsyncWriteExt; - let mut file = tokio::fs::File::create(tmp_path).await?; + let mut hasher = blake3::Hasher::new(); - let mut remaining = len; - let mut buf = vec![0u8; 256 * 1024]; - while remaining > 0 { - let want = remaining.min(buf.len() as u64) as usize; - let n = body.read(&mut buf[..want]).await?; - if n == 0 { - anyhow::bail!("stream ended {remaining} bytes short of the {len} the header declared"); + let mut file = if offset > 0 { + let mut file = tokio::fs::OpenOptions::new() + .read(true) + .append(true) + .open(part_path) + .await?; + let mut hashed = 0u64; + let mut buf = vec![0u8; 256 * 1024]; + while hashed < offset { + let want = (offset - hashed).min(buf.len() as u64) as usize; + let n = file.read(&mut buf[..want]).await?; + if n == 0 { + anyhow::bail!("partial artifact is {hashed} bytes, not the {offset} expected"); + } + hasher.update(&buf[..n]); + hashed += n as u64; } - hasher.update(&buf[..n]); - file.write_all(&buf[..n]).await?; - remaining -= n as u64; + file + } else { + tokio::fs::File::create(part_path).await? + }; + + let mut remaining = len - offset; + let mut buf = vec![0u8; 256 * 1024]; + let streamed = async { + while remaining > 0 { + let want = remaining.min(buf.len() as u64) as usize; + let n = body.read(&mut buf[..want]).await?; + if n == 0 { + anyhow::bail!( + "stream ended {remaining} bytes short of the {len} the header declared" + ); + } + // Checked after the cap's wait, so a pause landing during a + // throttled sleep drops this chunk instead of writing it late; + // the part stays an exact prefix either way. + super::replication::throttle().acquire(n as u64).await; + if super::replication::paused() { + anyhow::bail!("replication paused; {remaining} bytes still to fetch"); + } + hasher.update(&buf[..n]); + file.write_all(&buf[..n]).await?; + remaining -= n as u64; + reporter.advance(n as u64); + } + anyhow::Ok(()) } + .await; + // Whatever landed is kept for the next attempt, so it must reach the + // file before the handle goes: a dropped tokio file loses pending writes. + let flushed = file.flush().await; + streamed?; + flushed?; file.sync_all().await?; // A zeroed checksum is an owner that predates the field; the parse @@ -870,6 +1170,7 @@ async fn receive_artifact( if expected_checksum != [0u8; 32] { let received = *hasher.finalize().as_bytes(); if received != expected_checksum { + let _ = tokio::fs::remove_file(part_path).await; anyhow::bail!("delivered bytes do not match the checksum the header declared"); } } @@ -983,56 +1284,68 @@ async fn fetch_database_and_publish( context: &Arc, device_id: Uuid, device_label: &str, - replica_dir: &std::path::Path, + replica_dir: &Path, info: &RemoteSourceInfo, -) -> anyhow::Result<()> { +) -> Result<(), FetchError> { let started = std::time::Instant::now(); - let (response, mut body) = request( - context, + let source_id = info.id; + let owner = TransferOwner { + events: Some(context.events.clone()), device_id, - &ByteRangeRequest::FetchDatabase { source_id: info.id }, - ) - .await?; - let (len, delivered, checksum) = match response { - ByteRangeResponse::DatabaseHeader { - len, - generation, - checksum, - } => (len, generation, checksum), - other => anyhow::bail!("unexpected response: {other:?}"), + device_label, + info, }; - if len > MAX_SNAPSHOT_LEN { - anyhow::bail!("declared artifact of {len} bytes exceeds the transfer bound"); - } + let (part, header) = transfer_artifact(replica_dir, &owner, |offset| async move { + let req = if offset == 0 { + ByteRangeRequest::FetchDatabase { source_id } + } else { + ByteRangeRequest::FetchDatabaseFrom { source_id, offset } + }; + let (response, body) = request(context, device_id, &req).await?; + match response { + ByteRangeResponse::DatabaseHeader { + len, + generation, + checksum, + } => Ok(( + ArtifactHeader { + len, + generation, + checksum, + }, + body, + )), + other => anyhow::bail!("unexpected response: {other:?}"), + } + }) + .await?; - let tmp_path = replica_dir.join(format!("{}.tmp", Uuid::now_v7().simple())); - if let Err(err) = receive_artifact(&mut body, &tmp_path, len, checksum).await { - let _ = tokio::fs::remove_file(&tmp_path).await; - return Err(err); - } - - // The parse-and-rebuild is the gate: the temporary database must open + // The parse-and-rebuild is the gate: the complete database must open // read-only and yield an arena before it may replace anything. - let opened = sd_store::SourceManager::open_file_read_only(&tmp_path).await; - let db = match opened { + let db = match sd_store::SourceManager::open_file_read_only(&part).await { Ok(db) => db, Err(err) => { - let _ = tokio::fs::remove_file(&tmp_path).await; - anyhow::bail!("delivered database refused to open: {err}"); + let _ = tokio::fs::remove_file(&part).await; + return Err(FetchError::Failed(anyhow::anyhow!( + "delivered database refused to open: {err}" + ))); } }; let index = match arena_from_database(&db, &info.root).await { Ok(index) => index, Err(err) => { - let _ = tokio::fs::remove_file(&tmp_path).await; - return Err(err); + db.pool().close().await; + let _ = tokio::fs::remove_file(&part).await; + return Err(FetchError::Failed(err)); } }; // The pool must release its handle before the file moves. db.pool().close().await; let db_path = replica_dir.join(format!("{}.db", info.id.simple())); - tokio::fs::rename(&tmp_path, &db_path).await?; + tokio::fs::rename(&part, &db_path) + .await + .map_err(|err| FetchError::Failed(err.into()))?; // A database replica supersedes any arena-snapshot artifact the source // had before it was delivered this way. let _ = @@ -1041,7 +1354,7 @@ async fn fetch_database_and_publish( let share = Arc::new(RemoteShare { device_id, device_label: device_label.to_string(), - generation: delivered, + generation: header.generation, info: info.clone(), index: Arc::new(TokioRwLock::new(index)), synced_at_secs: now_secs(), @@ -1049,8 +1362,8 @@ async fn fetch_database_and_publish( shares_map().write().await.insert(info.id, share); tracing::info!( source = %info.id, - generation = delivered, - bytes = len, + generation = header.generation, + bytes = header.len, elapsed_ms = started.elapsed().as_millis() as u64, "replica database received and published" ); @@ -1058,44 +1371,52 @@ async fn fetch_database_and_publish( } /// Fetch one source's snapshot into the device's replica directory and -/// publish it. Every failure path removes its temporary file and leaves the -/// previously published artifact and share untouched. +/// publish it. A failed transfer leaves its partial file for the next +/// attempt and the previously published artifact and share untouched. async fn fetch_and_publish( context: &Arc, device_id: Uuid, device_label: &str, - replica_dir: &std::path::Path, + replica_dir: &Path, info: &RemoteSourceInfo, -) -> anyhow::Result<()> { +) -> Result<(), FetchError> { let started = std::time::Instant::now(); - let (response, mut body) = request( - context, + let source_id = info.id; + let owner = TransferOwner { + events: Some(context.events.clone()), device_id, - &ByteRangeRequest::FetchSnapshot { source_id: info.id }, - ) - .await?; - let (len, delivered, checksum) = match response { - ByteRangeResponse::SnapshotHeader { - len, - generation, - checksum, - } => (len, generation, checksum), - other => anyhow::bail!("unexpected response: {other:?}"), + device_label, + info, }; - if len > MAX_SNAPSHOT_LEN { - anyhow::bail!("declared artifact of {len} bytes exceeds the transfer bound"); - } - - let tmp_path = replica_dir.join(format!("{}.tmp", Uuid::now_v7().simple())); - if let Err(err) = receive_artifact(&mut body, &tmp_path, len, checksum).await { - let _ = tokio::fs::remove_file(&tmp_path).await; - return Err(err); - } + let (part, header) = transfer_artifact(replica_dir, &owner, |offset| async move { + let req = if offset == 0 { + ByteRangeRequest::FetchSnapshot { source_id } + } else { + ByteRangeRequest::FetchSnapshotFrom { source_id, offset } + }; + let (response, body) = request(context, device_id, &req).await?; + match response { + ByteRangeResponse::SnapshotHeader { + len, + generation, + checksum, + } => Ok(( + ArtifactHeader { + len, + generation, + checksum, + }, + body, + )), + other => anyhow::bail!("unexpected response: {other:?}"), + } + }) + .await?; // An owner that predates header identity sends zero; the listing's // generation is then the only name these bytes have. - let generation = if delivered != 0 { - delivered + let generation = if header.generation != 0 { + header.generation } else { info.generation }; @@ -1105,13 +1426,14 @@ async fn fetch_and_publish( device_label, info, generation, - &tmp_path, + &part, ) - .await?; + .await + .map_err(FetchError::Failed)?; tracing::info!( source = %info.id, generation, - bytes = len, + bytes = header.len, elapsed_ms = started.elapsed().as_millis() as u64, "replica artifact received and published" ); @@ -1169,6 +1491,7 @@ pub async fn sync_device( let replica_dir = replica_dir(context, device_id); tokio::fs::create_dir_all(&replica_dir).await?; + prune_parts(&replica_dir, &sources).await; let mut synced = 0usize; for info in &sources { @@ -1186,6 +1509,12 @@ pub async fn sync_device( continue; } } + // Paused, the pass still refreshed the listing, facts and totals + // above; what it does not do is move an artifact. + if super::replication::paused() { + tracing::debug!(source = %info.id, "replication paused; transfer deferred"); + continue; + } if !fetch_due(info.id) { continue; } @@ -1193,12 +1522,13 @@ pub async fn sync_device( let fetched = if info.nested { // A nested source travels as its own database, which carries // exactly its records. An owner too old to export one still - // answers on the snapshot path. + // answers on the snapshot path; a transfer that broke partway + // is retried as a database, since its partial file is waiting. match fetch_database_and_publish(context, device_id, &device_label, &replica_dir, info) .await { Ok(()) => Ok(()), - Err(err) => { + Err(FetchError::Unavailable(err)) => { tracing::debug!( source = %info.id, %err, @@ -1206,6 +1536,7 @@ pub async fn sync_device( ); fetch_and_publish(context, device_id, &device_label, &replica_dir, info).await } + Err(err) => Err(err), } } else { fetch_and_publish(context, device_id, &device_label, &replica_dir, info).await @@ -1215,6 +1546,10 @@ pub async fn sync_device( note_fetch_outcome(info.id, true); synced += 1; } + Err(err) if super::replication::paused() => { + // A pause is not the owner failing; no backoff accrues. + tracing::info!(source = %info.id, %err, "artifact fetch stopped by pause"); + } Err(err) => { note_fetch_outcome(info.id, false); tracing::warn!( @@ -1226,8 +1561,10 @@ pub async fn sync_device( } } - for info in &sources { - copy_sidecars(context, device_id, info); + if !super::replication::paused() { + for info in &sources { + copy_sidecars(context, device_id, info); + } } // Persist the inventory beside its artifacts. A source whose fetch @@ -1744,6 +2081,333 @@ mod tests { ); } + /// An owner that answers `transfer_artifact` from a byte buffer, cutting + /// the stream after `cut` bytes of each answer when asked to, the way a + /// killed daemon or a dropped link does. + struct ScriptedOwner { + bytes: Vec, + generation: u64, + cuts: std::collections::VecDeque>, + offsets: Vec, + } + + impl ScriptedOwner { + fn header(&self) -> ArtifactHeader { + ArtifactHeader { + len: self.bytes.len() as u64, + generation: self.generation, + checksum: *blake3::hash(&self.bytes).as_bytes(), + } + } + + fn open(&mut self, offset: u64) -> anyhow::Result<(ArtifactHeader, ArtifactBody)> { + self.offsets.push(offset); + let cut = self.cuts.pop_front().flatten(); + let mut tail = self.bytes[offset as usize..].to_vec(); + if let Some(cut) = cut { + tail.truncate(cut); + } + Ok((self.header(), Box::new(std::io::Cursor::new(tail)))) + } + } + + fn scripted_bytes(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() + } + + fn owner_of(info: &RemoteSourceInfo) -> TransferOwner<'_> { + TransferOwner { + events: None, + device_id: Uuid::nil(), + device_label: "owner", + info, + } + } + + /// The pause switch is process-wide, so tests that transfer run one at + /// a time; a pause flipped by one must not stop another's stream. + static TRANSFER_TESTS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + /// A stream that flips the pause switch after a prefix, the way a + /// person's `sd sources replication pause` lands mid-transfer. + struct PausingBody { + inner: std::io::Cursor>, + pause_after: u64, + } + + impl tokio::io::AsyncRead for PausingBody { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + if self.inner.position() >= self.pause_after { + super::super::replication::set_paused(true); + } + std::pin::Pin::new(&mut self.inner).poll_read(cx, buf) + } + } + + /// A pause stops a transfer at its next chunk and keeps the partial + /// file; the resumed transfer continues from it once unpaused. + #[tokio::test] + async fn a_pause_stops_a_transfer_and_keeps_its_part() { + let _serial = TRANSFER_TESTS.lock().await; + let base = tempfile::tempdir().expect("dir"); + let source_id = Uuid::now_v7(); + let source = info(source_id, "/mnt/pool/kept", 1); + let bytes = scripted_bytes(2_000_000); + let header = ArtifactHeader { + len: bytes.len() as u64, + generation: 3, + checksum: *blake3::hash(&bytes).as_bytes(), + }; + + let paused_at = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let stopped = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let bytes = bytes.clone(); + let paused_at = paused_at.clone(); + async move { + paused_at.lock().unwrap().push(offset); + let body: ArtifactBody = Box::new(PausingBody { + inner: std::io::Cursor::new(bytes[offset as usize..].to_vec()), + pause_after: 700_000, + }); + Ok((header, body)) + } + }) + .await; + assert!(super::super::replication::paused()); + assert!( + matches!(stopped, Err(FetchError::Failed(_))), + "a pause reads as a failed transfer: {stopped:?}" + ); + let part = find_part(base.path(), source_id) + .await + .expect("the partial file is kept"); + assert!( + part.have >= 700_000 && part.have < bytes.len() as u64, + "the transfer stopped partway: {} bytes", + part.have + ); + + super::super::replication::set_paused(false); + let (path, delivered) = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let bytes = bytes.clone(); + let paused_at = paused_at.clone(); + async move { + paused_at.lock().unwrap().push(offset); + let body: ArtifactBody = + Box::new(std::io::Cursor::new(bytes[offset as usize..].to_vec())); + Ok((header, body)) + } + }) + .await + .expect("resumes"); + assert_eq!(delivered, header); + assert_eq!(paused_at.lock().unwrap().as_slice(), &[0, part.have]); + assert_eq!(std::fs::read(path).expect("artifact"), bytes); + } + + /// A transfer cut partway leaves its `.part` on disk; the next attempt + /// asks for the tail from that length and the finished file matches the + /// header's checksum byte for byte. + #[tokio::test] + async fn an_interrupted_transfer_resumes_from_the_part_length() { + let _serial = TRANSFER_TESTS.lock().await; + let base = tempfile::tempdir().expect("dir"); + let source_id = Uuid::now_v7(); + let source = info(source_id, "/mnt/pool/kept", 1); + let owner = std::sync::Arc::new(std::sync::Mutex::new(ScriptedOwner { + bytes: scripted_bytes(3_000_000), + generation: 42, + cuts: [Some(1_000_000), None].into_iter().collect(), + offsets: Vec::new(), + })); + let expected = owner.lock().unwrap().header(); + + let first = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let owner = owner.clone(); + async move { owner.lock().unwrap().open(offset) } + }) + .await; + assert!( + matches!(first, Err(FetchError::Failed(_))), + "a cut stream fails the transfer: {first:?}" + ); + let part = find_part(base.path(), source_id) + .await + .expect("the partial file survives the failure"); + assert_eq!(part.have, 1_000_000); + assert_eq!(part.generation, 42); + assert_eq!(part.prefix[..], expected.checksum[..8]); + + let (path, header) = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let owner = owner.clone(); + async move { owner.lock().unwrap().open(offset) } + }) + .await + .expect("the second attempt completes"); + assert_eq!(header, expected); + assert_eq!( + owner.lock().unwrap().offsets, + vec![0, 1_000_000], + "the resumed request starts at the part's length" + ); + let delivered = std::fs::read(&path).expect("complete artifact"); + assert_eq!(blake3::hash(&delivered).as_bytes(), &expected.checksum); + assert_eq!(path, part.path, "the part file is the finished artifact"); + } + + /// A resumed request that gets no header keeps the part: a dropped link + /// or an owner too old to know the offset variants says nothing about + /// the bytes already on disk. The whole transfer that follows lands on + /// the same name and completes. + #[tokio::test] + async fn a_refused_resume_keeps_the_part_and_starts_over() { + let _serial = TRANSFER_TESTS.lock().await; + let base = tempfile::tempdir().expect("dir"); + let source_id = Uuid::now_v7(); + let source = info(source_id, "/mnt/pool/kept", 1); + let bytes = scripted_bytes(2_000); + let header = ArtifactHeader { + len: bytes.len() as u64, + generation: 8, + checksum: *blake3::hash(&bytes).as_bytes(), + }; + let part = part_path(base.path(), source_id, 8, &header.checksum); + std::fs::write(&part, &bytes[..500]).expect("part"); + + let offsets = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let refused = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let offsets = offsets.clone(); + async move { + offsets.lock().unwrap().push(offset); + anyhow::bail!("connect failed") + } + }) + .await; + assert!(matches!(refused, Err(FetchError::Unavailable(_)))); + assert_eq!( + std::fs::read(&part).expect("part kept").len(), + 500, + "a refused resume leaves the part alone" + ); + + let (path, _) = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let bytes = bytes.clone(); + let offsets = offsets.clone(); + async move { + offsets.lock().unwrap().push(offset); + if offset != 0 { + anyhow::bail!("owner too old for offsets") + } + let body: ArtifactBody = Box::new(std::io::Cursor::new(bytes)); + Ok((header, body)) + } + }) + .await + .expect("the whole transfer completes"); + assert_eq!(offsets.lock().unwrap().as_slice(), &[500, 0, 500, 0]); + assert_eq!(path, part, "the whole transfer lands on the same name"); + assert_eq!(std::fs::read(&path).expect("artifact"), bytes); + } + + /// A part cut from an older generation is discarded and the fetch starts + /// from zero, even though the resumed request was answered. + #[tokio::test] + async fn a_part_from_another_generation_starts_over() { + let _serial = TRANSFER_TESTS.lock().await; + let base = tempfile::tempdir().expect("dir"); + let source_id = Uuid::now_v7(); + let source = info(source_id, "/mnt/pool/kept", 1); + let stale = part_path(base.path(), source_id, 7, &[9u8; 32]); + std::fs::write(&stale, scripted_bytes(500)).expect("stale part"); + + let owner = std::sync::Arc::new(std::sync::Mutex::new(ScriptedOwner { + bytes: scripted_bytes(2_000), + generation: 8, + cuts: std::collections::VecDeque::new(), + offsets: Vec::new(), + })); + let (path, header) = transfer_artifact(base.path(), &owner_of(&source), |offset| { + let owner = owner.clone(); + async move { owner.lock().unwrap().open(offset) } + }) + .await + .expect("transfer"); + assert_eq!(owner.lock().unwrap().offsets, vec![500, 0]); + assert!(!stale.exists(), "the stale part is removed"); + assert_eq!( + std::fs::read(&path).expect("artifact"), + scripted_bytes(2_000) + ); + assert_eq!(header.generation, 8); + } + + /// The owner's listing is what retires a part: one whose generation the + /// owner still holds stays through any number of syncs, one it has moved + /// past goes, and one for a source the owner no longer lists goes too. + #[tokio::test] + async fn parts_are_pruned_against_the_owners_listing() { + let base = tempfile::tempdir().expect("dir"); + let kept = Uuid::now_v7(); + let moved = Uuid::now_v7(); + let gone = Uuid::now_v7(); + for (id, generation) in [(kept, 7), (moved, 3), (gone, 1)] { + std::fs::write(part_path(base.path(), id, generation, &[1u8; 32]), b"x").expect("part"); + } + let mut moved_info = info(moved, "/mnt/pool/moved", 1); + moved_info.generation = 4; + prune_parts(base.path(), &[info(kept, "/mnt/pool/kept", 1), moved_info]).await; + assert!(find_part(base.path(), kept).await.is_some()); + assert!(find_part(base.path(), moved).await.is_none()); + assert!(find_part(base.path(), gone).await.is_none()); + } + + #[test] + fn a_part_name_round_trips_its_identity() { + let source_id = Uuid::now_v7(); + let mut checksum = [0u8; 32]; + checksum[..8].copy_from_slice(&[0xde, 0xad, 0xbe, 0xef, 0, 1, 2, 3]); + let path = part_path(Path::new("/r"), source_id, 99, &checksum); + let name = path.file_name().and_then(|n| n.to_str()).expect("name"); + assert_eq!( + parse_part_name(name), + Some((source_id, 99, [0xde, 0xad, 0xbe, 0xef, 0, 1, 2, 3])) + ); + assert_eq!(parse_part_name("abc.snapshot"), None); + assert_eq!(parse_part_name("0195.tmp"), None); + } + + /// A resumed database fetch depends on the export being the same bytes + /// while the store has not changed; two exports of one store must hash + /// identically or no partial copy could ever continue. + #[tokio::test] + async fn an_unchanged_store_exports_identical_bytes() { + let store_dir = tempfile::tempdir().expect("store dir"); + let manager = sd_store::SourceManager::new(store_dir.path().to_path_buf()); + let id = Uuid::now_v7().simple().to_string(); + manager + .create(&id, &sd_store::filesystem_schema()) + .await + .expect("create"); + let db = manager.open(&id).await.expect("open"); + let out = tempfile::tempdir().expect("out"); + let mut hashes = Vec::new(); + for n in 0..2 { + let path = out.path().join(format!("export-{n}.db")); + sqlx::query("VACUUM INTO ?") + .bind(path.to_string_lossy().into_owned()) + .execute(db.pool()) + .await + .expect("export"); + hashes.push(blake3::hash(&std::fs::read(&path).expect("read"))); + } + db.pool().close().await; + assert_eq!(hashes[0], hashes[1]); + } + /// A manifest without facts restores; the device has none until its /// owner answers. #[tokio::test] diff --git a/core/src/service/mounts/replication.rs b/core/src/service/mounts/replication.rs new file mode 100644 index 000000000..5b2537e95 --- /dev/null +++ b/core/src/service/mounts/replication.rs @@ -0,0 +1,402 @@ +//! Controls over replica transfers: the bandwidth cap both ends of a fetch +//! share, the pause switch, and the progress of every fetch in flight. A +//! replica fetch is the one transfer Spacedrive starts on its own, as soon +//! as a paired device connects, and an initial copy runs to gigabytes, so +//! on a metered or shared link it needs a ceiling the person sets once and +//! the daemon keeps, a way to defer it entirely that survives a restart, +//! and a place that says how far it has got. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use specta::Type; +use tokio::time::Instant; +use uuid::Uuid; + +use crate::config::ReplicationConfig; +use crate::infra::event::{Event, EventBus}; +use crate::service::network::protocol::RemoteSourceInfo; + +/// A token bucket over bytes. The rate is the sustained ceiling and also +/// the burst: one second of credit accumulates while nothing moves, so a +/// transfer that starts after a pause does not stall for its first chunk. +/// A chunk larger than the burst passes against a full bucket and leaves it +/// in debt, which the following chunks pay off, so a low cap still moves +/// large chunks at the asked-for average. +pub struct Throttle { + bytes_per_sec: AtomicU64, + bucket: Mutex, +} + +struct Bucket { + tokens: f64, + refilled_at: Instant, +} + +impl Throttle { + pub fn new(bytes_per_sec: u64) -> Self { + Self { + bytes_per_sec: AtomicU64::new(bytes_per_sec), + bucket: Mutex::new(Bucket { + tokens: bytes_per_sec as f64, + refilled_at: Instant::now(), + }), + } + } + + /// The cap in bytes per second; zero is unlimited. + pub fn bytes_per_sec(&self) -> u64 { + self.bytes_per_sec.load(Ordering::Relaxed) + } + + /// Change the cap. Takes effect at the next chunk of every transfer in + /// flight; credit already in the bucket is clamped to the new burst. + pub fn set_bytes_per_sec(&self, bytes_per_sec: u64) { + self.bytes_per_sec.store(bytes_per_sec, Ordering::Relaxed); + let mut bucket = self.bucket.lock().unwrap(); + bucket.tokens = bucket.tokens.min(bytes_per_sec as f64); + } + + /// Wait until `bytes` may move. Returns at once while unlimited. + pub async fn acquire(&self, bytes: u64) { + loop { + let rate = self.bytes_per_sec() as f64; + if rate == 0.0 { + return; + } + let wait = { + let mut bucket = self.bucket.lock().unwrap(); + let now = Instant::now(); + let refill = now.duration_since(bucket.refilled_at).as_secs_f64() * rate; + bucket.tokens = (bucket.tokens + refill).min(rate); + bucket.refilled_at = now; + let want = bytes as f64; + if bucket.tokens >= want || bucket.tokens >= rate { + bucket.tokens -= want; + return; + } + Duration::from_secs_f64((want.min(rate) - bucket.tokens) / rate) + }; + tokio::time::sleep(wait).await; + } + } +} + +static THROTTLE: OnceLock = OnceLock::new(); + +/// The daemon's one cap, shared by every artifact and sidecar page served +/// or received, so the setting bounds what replication costs the link in +/// total. Tile fetches for a replica being browsed are not counted; they +/// are a person's own reads, not replication. +pub fn throttle() -> &'static Throttle { + THROTTLE.get_or_init(|| Throttle::new(0)) +} + +static PAUSED: AtomicBool = AtomicBool::new(false); + +/// Whether replica fetches are paused. A paused daemon still lists and +/// refreshes what owners publish, serves what it already holds, and answers +/// peers' fetches; it starts no transfer of its own, and one in flight stops +/// at its next chunk with its partial file kept. +pub fn paused() -> bool { + PAUSED.load(Ordering::Relaxed) +} + +/// Flip the pause switch in memory. The persisted value lives in the +/// daemon config, which the action that calls this writes. +pub fn set_paused(paused: bool) { + PAUSED.store(paused, Ordering::Relaxed); +} + +/// Apply the persisted settings at startup and after a config update. +pub fn configure(config: &ReplicationConfig) { + throttle().set_bytes_per_sec(config.max_bytes_per_sec); + set_paused(config.paused); +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A transfer is listed from its first byte to its last and no longer: + /// the reporter owns the row, counts from where it resumed, and clears + /// it however the fetch ends. + #[tokio::test] + async fn a_reporter_owns_its_row_for_the_life_of_the_transfer() { + let info = RemoteSourceInfo { + id: Uuid::now_v7(), + root: std::path::PathBuf::from("/mnt/pool/kept"), + volume_uuid: None, + attached: true, + entry_count: None, + total_bytes: None, + generation: 1, + dirty: false, + nested: false, + sidecar_store: Uuid::nil(), + sidecar_cursor: 0, + }; + let reporter = TransferReporter::start(None, Uuid::now_v7(), "owner", &info, 400, 1_000); + let listed = transfer(info.id).expect("listed while in flight"); + assert_eq!( + (listed.bytes, listed.total, listed.resumed_from), + (400, 1_000, 400) + ); + reporter.advance(100); + assert_eq!(transfer(info.id).expect("still listed").bytes, 500); + drop(reporter); + assert!( + transfer(info.id).is_none(), + "the row goes with the reporter" + ); + } + + /// A full bucket passes one second of bytes at once; past that, bytes + /// move at the cap. Measured against tokio's paused clock, so the test + /// asserts the limiter's arithmetic rather than the scheduler's timing. + #[tokio::test(start_paused = true)] + async fn bytes_past_the_burst_move_at_the_cap() { + let throttle = Throttle::new(1_000); + let started = Instant::now(); + throttle.acquire(1_000).await; + assert_eq!(started.elapsed(), Duration::ZERO, "the burst is free"); + + throttle.acquire(500).await; + assert_eq!(started.elapsed(), Duration::from_millis(500)); + + throttle.acquire(2_000).await; + assert_eq!( + started.elapsed(), + Duration::from_millis(1_500), + "a chunk past the burst goes once a full second of credit is in" + ); + throttle.acquire(1_000).await; + assert_eq!( + started.elapsed(), + Duration::from_millis(3_500), + "4,500 bytes after a 1,000-byte burst take 3.5 s at 1,000 B/s" + ); + } + + /// A chunk bigger than one second of the cap still passes, against a + /// full bucket, and the debt it leaves delays what follows. + #[tokio::test(start_paused = true)] + async fn an_oversized_chunk_passes_and_leaves_debt() { + let throttle = Throttle::new(100); + let started = Instant::now(); + throttle.acquire(1_000).await; + assert_eq!(started.elapsed(), Duration::ZERO); + throttle.acquire(100).await; + assert_eq!( + started.elapsed(), + Duration::from_secs(10), + "the next full bucket arrives after the debt is paid" + ); + } + + /// Zero is unlimited, and the cap can change under a transfer. + #[tokio::test(start_paused = true)] + async fn zero_is_unlimited_and_the_cap_changes_live() { + let throttle = Throttle::new(0); + let started = Instant::now(); + throttle.acquire(u64::MAX).await; + assert_eq!(started.elapsed(), Duration::ZERO); + + throttle.set_bytes_per_sec(1_000); + throttle.acquire(1_000).await; + throttle.acquire(1_000).await; + assert_eq!(started.elapsed(), Duration::from_secs(2)); + + throttle.set_bytes_per_sec(0); + throttle.acquire(1 << 40).await; + assert_eq!(started.elapsed(), Duration::from_secs(2)); + } +} + +/// One replica fetch as it stands. `bytes` counts what is on disk, +/// including what an earlier attempt left in the partial file, so a resumed +/// transfer reports from where it continues rather than from zero. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Type)] +pub struct ReplicaTransferProgress { + pub device_id: Uuid, + pub device_label: String, + pub source_id: Uuid, + /// The source's root on its owner, for naming the row. + pub root: std::path::PathBuf, + pub bytes: u64, + pub total: u64, + /// Recent transfer rate, averaged over the last few seconds. + pub bytes_per_sec: u64, + /// Where this attempt picked up; zero for a fresh transfer. + pub resumed_from: u64, + /// The transfer has left the active set: complete when `bytes` reaches + /// `total`, stopped otherwise with its partial file kept. + pub finished: bool, +} + +/// The replication picture `core.status` and the CLI summarize. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Type)] +pub struct ReplicationStatus { + pub paused: bool, + pub max_bytes_per_sec: u64, + pub transfers: Vec, +} + +impl ReplicationStatus { + pub fn bytes(&self) -> u64 { + self.transfers.iter().map(|t| t.bytes).sum() + } + + pub fn total(&self) -> u64 { + self.transfers.iter().map(|t| t.total).sum() + } + + pub fn bytes_per_sec(&self) -> u64 { + self.transfers.iter().map(|t| t.bytes_per_sec).sum() + } +} + +struct ActiveTransfer { + progress: ReplicaTransferProgress, + /// The sample the rate is measured from, replaced every window. + window: (Instant, u64), + /// The rate seen in the previous window, so a fresh one does not swing. + last_rate: u64, + announced: Instant, +} + +const RATE_WINDOW: Duration = Duration::from_secs(3); +const ANNOUNCE_EVERY: Duration = Duration::from_secs(1); + +static ACTIVE: OnceLock>> = OnceLock::new(); + +fn active() -> &'static Mutex> { + ACTIVE.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Every fetch in flight right now. +pub fn transfers() -> Vec { + let mut transfers: Vec<_> = active() + .lock() + .unwrap() + .values() + .map(|t| t.progress.clone()) + .collect(); + transfers.sort_by(|a, b| a.root.cmp(&b.root)); + transfers +} + +/// The fetch in flight for one source, if any. +pub fn transfer(source_id: Uuid) -> Option { + active() + .lock() + .unwrap() + .get(&source_id) + .map(|t| t.progress.clone()) +} + +pub fn status() -> ReplicationStatus { + ReplicationStatus { + paused: paused(), + max_bytes_per_sec: throttle().bytes_per_sec(), + transfers: transfers(), + } +} + +/// One fetch's reporter: it owns the source's entry in the active set for +/// as long as it lives, publishes a progress event about once a second, +/// and on drop announces the transfer finished, complete or not. Dropping +/// is what every exit path does, so a cancelled task reports too. +pub struct TransferReporter { + source_id: Uuid, + events: Option>, +} + +impl TransferReporter { + pub fn start( + events: Option>, + device_id: Uuid, + device_label: &str, + info: &RemoteSourceInfo, + resumed_from: u64, + total: u64, + ) -> Self { + let progress = ReplicaTransferProgress { + device_id, + device_label: device_label.to_string(), + source_id: info.id, + root: info.root.clone(), + bytes: resumed_from, + total, + bytes_per_sec: 0, + resumed_from, + finished: false, + }; + let now = Instant::now(); + active().lock().unwrap().insert( + info.id, + ActiveTransfer { + progress: progress.clone(), + window: (now, resumed_from), + last_rate: 0, + announced: now, + }, + ); + if let Some(events) = &events { + events.emit(Event::ReplicationProgress { transfer: progress }); + } + Self { + source_id: info.id, + events, + } + } + + /// Record that `bytes` more landed on disk. + pub fn advance(&self, bytes: u64) { + let announce = { + let mut active = active().lock().unwrap(); + let Some(entry) = active.get_mut(&self.source_id) else { + return; + }; + let now = Instant::now(); + entry.progress.bytes += bytes; + let since = now.duration_since(entry.window.0); + if since >= RATE_WINDOW { + entry.last_rate = + ((entry.progress.bytes - entry.window.1) as f64 / since.as_secs_f64()) as u64; + entry.window = (now, entry.progress.bytes); + } + entry.progress.bytes_per_sec = if entry.last_rate == 0 && since > Duration::ZERO { + ((entry.progress.bytes - entry.window.1) as f64 / since.as_secs_f64()) as u64 + } else { + entry.last_rate + }; + if now.duration_since(entry.announced) >= ANNOUNCE_EVERY { + entry.announced = now; + Some(entry.progress.clone()) + } else { + None + } + }; + if let (Some(events), Some(progress)) = (&self.events, announce) { + events.emit(Event::ReplicationProgress { transfer: progress }); + } + } +} + +impl Drop for TransferReporter { + fn drop(&mut self) { + let Some(entry) = active().lock().unwrap().remove(&self.source_id) else { + return; + }; + if let Some(events) = &self.events { + let mut progress = entry.progress; + progress.finished = true; + progress.bytes_per_sec = 0; + events.emit(Event::ReplicationProgress { transfer: progress }); + } + } +} diff --git a/core/src/service/network/protocol/byterange.rs b/core/src/service/network/protocol/byterange.rs index c8c36ebf7..e8b7b42b0 100644 --- a/core/src/service/network/protocol/byterange.rs +++ b/core/src/service/network/protocol/byterange.rs @@ -8,7 +8,9 @@ //! carry no session state, and cancelling an in-flight read is dropping the //! stream. Framing is `[u32 BE length][rmp_serde message]`; `Read` and //! `FetchSnapshot` responses are followed by exactly `len` raw bytes on the -//! same stream so payloads are never re-encoded through msgpack. `FetchTiles` +//! same stream so payloads are never re-encoded through msgpack, and the +//! `From` variants resume an artifact by sending its tail from an offset +//! behind a header that still names the whole file. `FetchTiles` //! is the one request answered by many frames: one `Tile` per ask, each with //! its bytes behind it, in the order the owner finishes them. @@ -84,6 +86,16 @@ pub enum ByteRangeRequest { after: u64, limit: u32, }, + /// `FetchSnapshot` resumed from `offset`: the header still names the + /// whole artifact, and the bytes that follow are its tail from `offset`. + /// A nonzero offset serves the file as it stands, without the pre-fetch + /// save, so an interrupted copy can finish against the same bytes. + FetchSnapshotFrom { source_id: Uuid, offset: u64 }, + /// `FetchDatabase` resumed from `offset`. The export is rebuilt, which + /// yields the same bytes while the store's revision has not moved; the + /// header's checksum tells the receiver whether its partial copy still + /// matches. + FetchDatabaseFrom { source_id: Uuid, offset: u64 }, } /// One file whose tile a replica wants. @@ -518,11 +530,127 @@ impl ByteRangeProtocolHandler { }, ) .await?; + crate::service::mounts::replication::throttle() + .acquire(row.tile.webp.len() as u64) + .await; send.write_all(&row.tile.webp).await?; } Ok(()) } + /// One source's arena snapshot from `offset`. A fresh fetch saves live + /// arena state first so the replica does not start a session stale; a + /// resumed one must not, since a new save would change the bytes the + /// receiver is partway through. + async fn serve_snapshot( + &self, + source_id: Uuid, + offset: u64, + send: &mut W, + ) -> anyhow::Result<()> { + let cache = self.context.volume_index(); + let Some(source) = cache.sources().into_iter().find(|s| s.id == source_id) else { + return write_frame(send, &ByteRangeResponse::Error("unknown source".into())).await; + }; + if source.attached && offset == 0 { + if let Err(err) = cache.save_snapshot(&source.root).await { + tracing::debug!("pre-fetch snapshot save failed: {err}"); + } + } + let Some(snapshot_path) = cache.source_snapshot_path(source_id) else { + return write_frame( + send, + &ByteRangeResponse::Error("source has no persistent snapshot".into()), + ) + .await; + }; + let file = match tokio::fs::File::open(&snapshot_path).await { + Ok(f) => f, + Err(err) => { + return write_frame( + send, + &ByteRangeResponse::Error(format!("snapshot unavailable: {err}")), + ) + .await; + } + }; + serve_file_with_identity(send, file, None, false, offset).await + } + + /// A consistent export of one source's database from `offset`. + async fn serve_database( + &self, + source_id: Uuid, + offset: u64, + send: &mut W, + ) -> anyhow::Result<()> { + let cache = self.context.volume_index(); + if !cache.sources().into_iter().any(|s| s.id == source_id) { + return write_frame(send, &ByteRangeResponse::Error("unknown source".into())).await; + } + let Some(db) = cache.read_store(source_id).await else { + return write_frame( + send, + &ByteRangeResponse::Error("source has no readable store".into()), + ) + .await; + }; + let Some(dirs) = cache.source_dirs() else { + return write_frame( + send, + &ByteRangeResponse::Error("no persistent source layout".into()), + ) + .await; + }; + let source_dir = dirs.source_dir(source_id); + + // The generation names the store revision this export starts + // from, read before the export runs: a write landing mid-export + // moves the next listing instead of mislabeling this delivery + // as current. + let generation = match db.revision().await { + Ok(revision) => revision_generation(revision), + Err(err) => { + return write_frame( + send, + &ByteRangeResponse::Error(format!("store revision unreadable: {err}")), + ) + .await; + } + }; + + // VACUUM INTO produces a consistent, compact single-file copy + // from one read transaction; walkers keep writing meanwhile. The + // copy is byte-identical for an unchanged store, which is what + // lets a resumed fetch continue against a fresh export. + let export_path = source_dir.join(format!("export-{}.db", Uuid::now_v7().simple())); + let exported = sqlx::query("VACUUM INTO ?") + .bind(export_path.to_string_lossy().into_owned()) + .execute(db.pool()) + .await; + if let Err(err) = exported { + let _ = tokio::fs::remove_file(&export_path).await; + return write_frame( + send, + &ByteRangeResponse::Error(format!("export failed: {err}")), + ) + .await; + } + + let served = match tokio::fs::File::open(&export_path).await { + Ok(file) => serve_file_with_identity(send, file, Some(generation), true, offset).await, + Err(err) => { + write_frame( + send, + &ByteRangeResponse::Error(format!("export unavailable: {err}")), + ) + .await + } + }; + let _ = tokio::fs::remove_file(&export_path).await; + served + } + async fn respond( &self, request: ByteRangeRequest, @@ -684,102 +812,16 @@ impl ByteRangeProtocolHandler { .await } ByteRangeRequest::FetchSnapshot { source_id } => { - let cache = self.context.volume_index(); - let Some(source) = cache.sources().into_iter().find(|s| s.id == source_id) else { - return write_frame(send, &ByteRangeResponse::Error("unknown source".into())) - .await; - }; - // Persist live arena state first so the replica isn't stale - // by a whole session; failure falls back to the on-disk file. - if source.attached { - if let Err(err) = cache.save_snapshot(&source.root).await { - tracing::debug!("pre-fetch snapshot save failed: {err}"); - } - } - let Some(snapshot_path) = cache.source_snapshot_path(source_id) else { - return write_frame( - send, - &ByteRangeResponse::Error("source has no persistent snapshot".into()), - ) - .await; - }; - let file = match tokio::fs::File::open(&snapshot_path).await { - Ok(f) => f, - Err(err) => { - return write_frame( - send, - &ByteRangeResponse::Error(format!("snapshot unavailable: {err}")), - ) - .await; - } - }; - serve_file_with_identity(send, file, None, false).await + self.serve_snapshot(source_id, 0, send).await + } + ByteRangeRequest::FetchSnapshotFrom { source_id, offset } => { + self.serve_snapshot(source_id, offset, send).await } ByteRangeRequest::FetchDatabase { source_id } => { - let cache = self.context.volume_index(); - if !cache.sources().into_iter().any(|s| s.id == source_id) { - return write_frame(send, &ByteRangeResponse::Error("unknown source".into())) - .await; - } - let Some(db) = cache.read_store(source_id).await else { - return write_frame( - send, - &ByteRangeResponse::Error("source has no readable store".into()), - ) - .await; - }; - let Some(dirs) = cache.source_dirs() else { - return write_frame( - send, - &ByteRangeResponse::Error("no persistent source layout".into()), - ) - .await; - }; - let source_dir = dirs.source_dir(source_id); - - // The generation names the store revision this export starts - // from, read before the export runs: a write landing mid-export - // moves the next listing instead of mislabeling this delivery - // as current. - let generation = match db.revision().await { - Ok(revision) => revision_generation(revision), - Err(err) => { - return write_frame( - send, - &ByteRangeResponse::Error(format!("store revision unreadable: {err}")), - ) - .await; - } - }; - - // VACUUM INTO produces a consistent, compact single-file copy - // from one read transaction; walkers keep writing meanwhile. - let export_path = source_dir.join(format!("export-{}.db", Uuid::now_v7().simple())); - let exported = sqlx::query("VACUUM INTO ?") - .bind(export_path.to_string_lossy().into_owned()) - .execute(db.pool()) - .await; - if let Err(err) = exported { - let _ = tokio::fs::remove_file(&export_path).await; - return write_frame( - send, - &ByteRangeResponse::Error(format!("export failed: {err}")), - ) - .await; - } - - let served = match tokio::fs::File::open(&export_path).await { - Ok(file) => serve_file_with_identity(send, file, Some(generation), true).await, - Err(err) => { - write_frame( - send, - &ByteRangeResponse::Error(format!("export unavailable: {err}")), - ) - .await - } - }; - let _ = tokio::fs::remove_file(&export_path).await; - served + self.serve_database(source_id, 0, send).await + } + ByteRangeRequest::FetchDatabaseFrom { source_id, offset } => { + self.serve_database(source_id, offset, send).await } ByteRangeRequest::DeviceFacts => { let config = match self.context.device_manager.config() { @@ -803,16 +845,19 @@ impl ByteRangeProtocolHandler { } } -/// Hash an open file, emit its identity header, and stream exactly the bytes -/// measured. Identity comes from the open handle, not the path: a writer -/// renaming a new file into place cannot change what this stream carries, -/// and the header must name what the stream holds rather than what the -/// listing that prompted the fetch believed. +/// Hash an open file, emit its identity header, and stream its bytes from +/// `offset` to the end. Identity comes from the open handle, not the path: a +/// writer renaming a new file into place cannot change what this stream +/// carries, and the header must name what the stream holds rather than what +/// the listing that prompted the fetch believed. The header always describes +/// the whole file, so a receiver resuming at `offset` validates the same +/// generation and checksum a fresh fetch would. async fn serve_file_with_identity( send: &mut W, file: tokio::fs::File, generation: Option, database: bool, + offset: u64, ) -> anyhow::Result<()> { let meta = file.metadata().await?; let len = meta.len(); @@ -825,6 +870,15 @@ async fn serve_file_with_identity( ) .await; } + if offset > len { + return write_frame( + send, + &ByteRangeResponse::Error(format!( + "offset {offset} is past the end of a {len}-byte artifact" + )), + ) + .await; + } let generation = generation.unwrap_or_else(|| { meta.modified() .map(|mtime| crate::infra::source_version::source_version(len, mtime)) @@ -837,7 +891,7 @@ async fn serve_file_with_identity( use std::io::{Seek, SeekFrom}; let mut hasher = blake3::Hasher::new(); std::io::copy(&mut &std_file, &mut hasher)?; - (&std_file).seek(SeekFrom::Start(0))?; + (&std_file).seek(SeekFrom::Start(offset))?; Ok((std_file, *hasher.finalize().as_bytes())) }) .await?; @@ -869,7 +923,7 @@ async fn serve_file_with_identity( write_frame(send, &header).await?; let mut buf = vec![0u8; 256 * 1024]; - let mut sent = 0u64; + let mut sent = offset; while sent < len { let want = (len - sent).min(buf.len() as u64) as usize; let n = file.read(&mut buf[..want]).await?; @@ -878,6 +932,9 @@ async fn serve_file_with_identity( // accounting fails the transfer. break; } + crate::service::mounts::replication::throttle() + .acquire(n as u64) + .await; send.write_all(&buf[..n]).await?; sent += n as u64; } @@ -1127,6 +1184,49 @@ mod tests { assert_eq!(listed.last_seen_at, observed_at); } + /// A resumed artifact request answers with the whole file's identity and + /// only the bytes from the offset, so a receiver appending to a partial + /// copy validates against the same header a fresh fetch would. An offset + /// past the end is refused rather than served as an empty success. + #[tokio::test] + async fn a_resumed_artifact_serves_its_tail_behind_a_whole_file_header() { + let dir = tempfile::tempdir().expect("dir"); + let path = dir.path().join("artifact"); + let bytes: Vec = (0..600_000u32).map(|i| (i % 253) as u8).collect(); + std::fs::write(&path, &bytes).expect("write"); + let checksum = *blake3::hash(&bytes).as_bytes(); + + let mut wire = Vec::new(); + let file = tokio::fs::File::open(&path).await.expect("open"); + serve_file_with_identity(&mut wire, file, Some(5), false, 400_000) + .await + .expect("serve"); + let mut reader = wire.as_slice(); + match read_frame(&mut reader).await.expect("header") { + ByteRangeResponse::SnapshotHeader { + len, + generation, + checksum: declared, + } => { + assert_eq!(len, bytes.len() as u64, "the header names the whole file"); + assert_eq!(generation, 5); + assert_eq!(declared, checksum); + } + other => panic!("unexpected header: {other:?}"), + } + assert_eq!(reader, &bytes[400_000..], "only the tail follows"); + + let mut refused = Vec::new(); + let file = tokio::fs::File::open(&path).await.expect("open"); + serve_file_with_identity(&mut refused, file, None, true, 600_001) + .await + .expect("answered"); + assert!(matches!( + read_frame(&mut refused.as_slice()).await.expect("frame"), + ByteRangeResponse::Error(_) + )); + } + /// Facts cross the frame codec intact, and a classification this build /// does not know arrives as unknown instead of failing the response. #[tokio::test] diff --git a/core/src/testing/integration_utils.rs b/core/src/testing/integration_utils.rs index 46af5b4c6..d9fd0165b 100644 --- a/core/src/testing/integration_utils.rs +++ b/core/src/testing/integration_utils.rs @@ -224,6 +224,7 @@ impl TestConfigBuilder { logging: crate::config::app_config::LoggingConfig::default(), proxy_pairing: crate::config::app_config::ProxyPairingConfig::default(), mounts: crate::config::app_config::MountsConfig::default(), + replication: crate::config::app_config::ReplicationConfig::default(), } } diff --git a/core/tests/helpers/sync_harness.rs b/core/tests/helpers/sync_harness.rs index 2a1fe32c7..5f52fbee3 100644 --- a/core/tests/helpers/sync_harness.rs +++ b/core/tests/helpers/sync_harness.rs @@ -72,6 +72,7 @@ impl TestConfigBuilder { }, proxy_pairing: sd_core::config::app_config::ProxyPairingConfig::default(), mounts: sd_core::config::app_config::MountsConfig::default(), + replication: sd_core::config::app_config::ReplicationConfig::default(), }; config.save()?; diff --git a/docs/core/file-sync.mdx b/docs/core/file-sync.mdx index f3ab239d8..b9b502aa1 100644 --- a/docs/core/file-sync.mdx +++ b/docs/core/file-sync.mdx @@ -131,6 +131,13 @@ destination support it. Provider adapters may choose multipart uploads, server-side copies, or local streaming, but they report the same operation state. +Index replication between paired devices is not file sync, but it is the one +transfer Spacedrive starts on its own, and it follows the same rules: a +partial copy survives a restart and resumes from its length, a bandwidth cap +and a pause switch are persisted daemon settings, and progress is visible in +`sd status` and `sd sources list`. See +[Source replication](/core/networking#source-replication). + ## Deletion safety Deletion propagation is off unless the relationship explicitly enables it. diff --git a/docs/core/networking.mdx b/docs/core/networking.mdx index 1b1b6e569..c96671107 100644 --- a/docs/core/networking.mdx +++ b/docs/core/networking.mdx @@ -96,6 +96,69 @@ sidecars through one provider boundary. It supports: Control messages do not carry large file bodies. This keeps queries responsive while previews, mounts, and transfers stream in parallel. +## Source replication + +When a paired device connects, this device copies the index of every source +the peer serves into `mounts-remote//` and loads it as a read-only +replica. The copy is an arena snapshot for a source at its volume root, or a +single-file database export for a nested source. The owner's response header +names the whole artifact: its length, its generation, and the blake3 hash of +its bytes. A delivery must match that hash and parse before it replaces the +replica already published. + +### Partial transfers + +A transfer lands in `...part` and stays +there when the stream breaks or the daemon stops. The next attempt asks the +owner for the artifact's tail from the part's length (`FetchSnapshotFrom` or +`FetchDatabaseFrom`). The owner answers with the same whole-artifact header +and streams only the bytes past the offset; a resumed snapshot request skips +the owner's pre-fetch save so the bytes stay the ones the part was cut from. +A header whose generation or hash prefix does not match the part discards it +and starts over. A part is removed only when the owner's listing no longer +carries its generation, so an owner that is offline for a week leaves the +part waiting. The finished file is validated against the header hash before +it replaces the live replica. + +Both peers must run a build that knows the `From` requests. An owner that +does not answers them with an error, and the receiver starts the transfer +from zero as before. + +### Bandwidth cap + +`replication.max_bytes_per_sec` in the daemon config bounds replica bytes +moved per second, artifacts and thumbnail sidecar pages, serving and +receiving combined, through one token bucket with a one-second burst. Zero is unlimited. Set it with +`sd config set replication.max_bytes_per_sec 200K` (bytes, or K, M, G in +powers of 1024); the change applies to transfers already in flight. An +initial replica runs to gigabytes and starts as soon as a device connects, so +set the cap before pairing on a metered or shared link. + +### Pause + +`sd sources replication pause` stops this device starting replica transfers +and halts any in flight at its next chunk, keeping the partial file. The +daemon still refreshes peer listings, owner facts and fleet totals, still +serves its own sources to peers, and still answers remote operations. +`sd sources replication resume` continues each partial file where it stopped +and runs a replication pass against every connected device at once. The +switch is `replication.paused` in the daemon config, so a paused daemon stays +paused across a restart. A pause is not a fetch failure: no backoff accrues. + +### Progress + +Every fetch in flight reports `bytes` on disk (including what an earlier +attempt left in the part), `total`, a `bytes_per_sec` rate averaged over the +last few seconds, and `resumed_from`. `sources.list` carries the figures on +the replica's row, or on a `replica_fetching` row for a first copy that has +no replica yet. `core.status` and `mounts.replication_status` return the +pause switch, the cap, and every transfer; `sd status` prints them as one +line (`Replicating 2 sources, 553.0 MB of 2.9 GB, 212.0 KB/s`), and +`sd sources list` appends `fetching ...` to the status column. The daemon +emits `ReplicationProgress { transfer }` about once a second per transfer +and once more with `finished: true` when the transfer leaves the active set, +complete or stopped. + ## Sync traffic Library assertions and source generations use separate streams and watermarks. diff --git a/docs/plans/2026-09-19-incremental-replication.md b/docs/plans/2026-09-19-incremental-replication.md index 4c8e82a6d..abc7933e1 100644 --- a/docs/plans/2026-09-19-incremental-replication.md +++ b/docs/plans/2026-09-19-incremental-replication.md @@ -252,6 +252,16 @@ This executes R3's "clean up failed temporary transfers". - Both: remove the temporary file on every exit path of a transfer, cancellation included, through a guard that deletes on drop. +Superseded on the replica side (2026-10-05): a transfer now lands in +`...part` and is kept across failures and +restarts, resuming from its length over `FetchSnapshotFrom` and +`FetchDatabaseFrom`. A part is removed when the owner's listing no longer +carries its generation. The owner-side export cleanup stands. This is +related to but distinct from "A restart must not re-copy what the replica +already has" above: that item is a complete replica re-fetched because its +generation is not visible before restore finishes; this is an incomplete +transfer resumed instead of restarted (SPAC-12, bug 5). + ### Volume-root sources Volume-root sources replicate as arena snapshots, with a file-metadata diff --git a/packages/ts-client/src/generated/types.ts b/packages/ts-client/src/generated/types.ts index e5db02491..9db63bf80 100644 --- a/packages/ts-client/src/generated/types.ts +++ b/packages/ts-client/src/generated/types.ts @@ -99,6 +99,10 @@ services: ServiceConfigOutput; * Mounted-source behaviour */ mounts: MountsConfigOutput; +/** + * Replica transfers with paired devices + */ +replication: ReplicationConfigOutput; /** * Daemon logging configuration */ @@ -498,7 +502,12 @@ is_fast_operation: boolean; */ copy_method: CopyMethod }; -export type CoreStatus = { version: string; built_at: string; library_count: number; device_info: DeviceInfo; libraries: LibraryInfo[]; services: ServiceStatus; network: NetworkStatus; system: SystemInfo }; +export type CoreStatus = { version: string; built_at: string; library_count: number; device_info: DeviceInfo; libraries: LibraryInfo[]; services: ServiceStatus; network: NetworkStatus; system: SystemInfo; +/** + * Replica fetches from paired devices: the pause switch, the cap and + * every transfer in flight. + */ +replication: ReplicationStatus }; /** * Input for creating a new folder @@ -1119,7 +1128,13 @@ resource_type: string; /** * The deleted resource's ID */ -resource_id: string } } | { ConfigChanged: { field: string } } | { Custom: { event_type: string } }; +resource_id: string } } | { ConfigChanged: { field: string } } | +/** + * A replica fetch from a paired device moved, about once a second per + * transfer, with a last emission marked `finished` when it leaves the + * active set, complete or stopped. + */ +{ ReplicationProgress: { transfer: ReplicaTransferProgress } } | { Custom: { event_type: string } }; /** * Event category for grouping related events @@ -3113,6 +3128,12 @@ rereads: number; p50_micros: number; p95_micros: number; max_micros: number; siz export type MountsReadTraceInput = Record; +export type MountsReplicationSetPausedInput = { paused: boolean }; + +export type MountsReplicationSetPausedOutput = { paused: boolean; message: string }; + +export type MountsReplicationStatusInput = Record; + export type MountsStatus = { running: boolean; /** * WebDAV base, kept as plain interop and as a measurement baseline. @@ -3669,6 +3690,48 @@ export type ReplaceReason = */ "newer"; +/** + * One replica fetch as it stands. `bytes` counts what is on disk, + * including what an earlier attempt left in the partial file, so a resumed + * transfer reports from where it continues rather than from zero. + */ +export type ReplicaTransferProgress = { device_id: string; device_label: string; source_id: string; +/** + * The source's root on its owner, for naming the row. + */ +root: string; bytes: number; total: number; +/** + * Recent transfer rate, averaged over the last few seconds. + */ +bytes_per_sec: number; +/** + * Where this attempt picked up; zero for a fresh transfer. + */ +resumed_from: number; +/** + * The transfer has left the active set: complete when `bytes` reaches + * `total`, stopped otherwise with its partial file kept. + */ +finished: boolean }; + +/** + * Replication configuration output + */ +export type ReplicationConfigOutput = { +/** + * Bytes per second across every replica transfer; zero is unlimited. + */ +max_bytes_per_sec: number; +/** + * Whether this device has stopped fetching replicas. + */ +paused: boolean }; + +/** + * The replication picture `core.status` and the CLI summarize. + */ +export type ReplicationStatus = { paused: boolean; max_bytes_per_sec: number; transfers: ReplicaTransferProgress[] }; + export type ResetDataInput = { /** * Confirmation flag to prevent accidental data loss @@ -4106,7 +4169,13 @@ device_id: string | null; /** * The owning device's display name, for a replica. */ -device_label: string | null }; +device_label: string | null; +/** + * The fetch bringing this replica up to date, while one is in flight. + * Present on a replica being refreshed and on a source whose first + * copy has not finished, which has no other row to appear in. + */ +transfer?: ReplicaTransferProgress | null }; export type SourceItem = { id: string; external_id: string; title: string; preview: string | null; subtitle: string | null }; @@ -4752,6 +4821,11 @@ language?: string | null; * restart, because nothing in it is authoritative. */ mounts_cache_max_bytes?: number | null; +/** + * Ceiling on replica bytes per second, serving and fetching combined; + * zero lifts it. Applies to transfers already in flight. + */ +replication_max_bytes_per_sec?: number | null; /** * Whether networking is enabled */ @@ -5380,6 +5454,7 @@ export type CoreAction = | { type: 'models.whisper.delete'; input: DeleteWhisperModelInput; output: DeleteWhisperModelOutput } | { type: 'models.whisper.download'; input: DownloadWhisperModelInput; output: DownloadWhisperModelOutput } | { type: 'mounts.cache_clear'; input: MountsCacheClearInput; output: MountsCacheClearOutput } + | { type: 'mounts.replication_set_paused'; input: MountsReplicationSetPausedInput; output: MountsReplicationSetPausedOutput } | { type: 'mounts.sync_peers'; input: MountsSyncPeersInput; output: MountsSyncPeersOutput } | { type: 'mounts.trace_set'; input: MountsTraceSetInput; output: MountsTraceSetOutput } | { type: 'navigation.set_focus'; input: SetNavigationFocusInput; output: SetNavigationFocusOutput } @@ -5474,6 +5549,7 @@ export type CoreQuery = | { type: 'models.whisper.list'; input: ListWhisperModelsInput; output: ListWhisperModelsOutput } | { type: 'mounts.cache_status'; input: MountsCacheStatusInput; output: MountsCacheStatus } | { type: 'mounts.read_trace'; input: MountsReadTraceInput; output: MountsReadTrace } + | { type: 'mounts.replication_status'; input: MountsReplicationStatusInput; output: ReplicationStatus } | { type: 'mounts.status'; input: MountsStatusInput; output: MountsStatus } | { type: 'navigation.focus'; input: NavigationFocusInput; output: NavigationFocusOutput } | { type: 'network.devices.list'; input: ListPairedDevicesInput; output: ListPairedDevicesOutput } @@ -5578,6 +5654,7 @@ export const WIRE_METHODS = { 'models.whisper.delete': 'action:models.whisper.delete.input', 'models.whisper.download': 'action:models.whisper.download.input', 'mounts.cache_clear': 'action:mounts.cache_clear.input', + 'mounts.replication_set_paused': 'action:mounts.replication_set_paused.input', 'mounts.sync_peers': 'action:mounts.sync_peers.input', 'mounts.trace_set': 'action:mounts.trace_set.input', 'navigation.set_focus': 'action:navigation.set_focus.input', @@ -5672,6 +5749,7 @@ export const WIRE_METHODS = { 'models.whisper.list': 'query:models.whisper.list', 'mounts.cache_status': 'query:mounts.cache_status', 'mounts.read_trace': 'query:mounts.read_trace', + 'mounts.replication_status': 'query:mounts.replication_status', 'mounts.status': 'query:mounts.status', 'navigation.focus': 'query:navigation.focus', 'network.devices.list': 'query:network.devices.list',