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',