diff --git a/docs/plans/ROADMAP.md b/docs/plans/ROADMAP.md index 2b29d6d..73a257a 100644 --- a/docs/plans/ROADMAP.md +++ b/docs/plans/ROADMAP.md @@ -52,7 +52,7 @@ thematic detail. | 14 | `14_iconify-brand-icons` | ✅ Phases 1–3 shipped + runtime-verified (Phase 4 consolidation deferred) | Additive brand/protocol logos, bundled offline. | | 15 | `15_keyboard-shortcuts` | ⬜ | Remappable shortcuts + Settings Keyboard tab + file-browser keys (F2 rename & friends). Builds on the Plan 12 settings substrate. Independent polish; do whenever. | | 16 | `16_app-updater-and-notifications` | ⬜ | In-app auto-updater (signed, GitHub Releases manifest) + desktop notifications + one-click per-user PATH install. Trust/UX fundamentals for a shipping app. | -| 17 | `17_transfer-queue-depth` | ⬜ | Real queue (bounded concurrency), pause/resume, retry, bandwidth throttle. Turns the transfer list into an actual queue. | +| 17 | `17_transfer-queue-depth` | ✅ built (all 4 phases; live-backend smoke run left) | Real queue (bounded concurrency), pause/resume, retry, bandwidth throttle. Turns the transfer list into an actual queue. | | 18 | `18_jump-hosts-proxyjump` | ⬜ | Jump hosts (ProxyJump) through the single `ssh_connect` choke point + optional cloudflared integration. Unlocks locked-down (IP-allowlisted / tunnel-fronted) servers. Consumes Plan 13's bastion E2E fixture. | | 20 | `20_hubspot-backend` | ✅ shipped (Phases 1–3, mock-verified; HubDB write-back future) | HubSpot portal as a connection, one private-app token over three surfaces: Design Manager as a real remote filesystem (Source Code API v3, draft/published roots), File Manager (Files API v3), HubDB tables as virtual CSV files (read-only). The Shopify recipe (`18_shopify-backend.md`, shipped), an even better fit. | | 21 | `21_dynamics-365-backend` | 🔄 Phase 1 shipped + mock-verified (client-credentials only; delegated OAuth + Phases 2–3 planned) | Dynamics 365/Dataverse environment as a connection: web resources are literally files in a table (`webresourceset` OData — path-like names, base64 content, publish-to-deploy). Client-credentials or delegated Entra auth (reuses `oauth.rs`). Phase 2: tables as virtual CSV + `faro-cli dynamics query` (the wp-cli-style db helper). The XrmToolBox gap-filler. | @@ -415,7 +415,16 @@ concurrency (semaphore + FIFO, reorderable), per-transfer pause/resume (park at a chunk boundary, resume re-runs the file), manual + classified auto-retry (Plan 12 error kinds), and a global token-bucket bandwidth throttle. All checkpoints live in the *shared* copy loops so the 11 backends get it for -free. Panel gains pause/retry row actions, pause-all, and a throttle input. ⬜ +free. Panel gains pause/retry row actions, pause-all, and a throttle input. ✅ +**Built** — all four phases: FIFO + semaphore admission (`transferConcurrency`, +live-adjustable), per-transfer `PauseGate`s with chunk checkpoints in every +copy loop (resume re-runs from byte 0), manual + auto retry (Network/Timeout, +5s/20s backoff, classified via `classify_message`), global `TokenBucket` +(`transferThrottleKbps`, live). Panel: pause-all, throttle input, row +pause/resume/retry/↑↓ with queue positions; Settings → Transfers for the two +defaults. Unit-tested (bucket pacing, FIFO skip-paused, checkpoint restart, +retry classification); `scripts/verify-transfers.mjs` drives the panel +headlessly. Remaining: a live-backend smoke run (real SFTP/S3 batch). ## Track R — Jump hosts & Zero-Trust connectivity (Plan 18) Locked-down servers (IP allowlists, Cloudflare Tunnels) are unreachable to diff --git a/scripts/verify-transfers.mjs b/scripts/verify-transfers.mjs new file mode 100644 index 0000000..6cd9323 --- /dev/null +++ b/scripts/verify-transfers.mjs @@ -0,0 +1,317 @@ +// Headless runtime verification for Plan 17 (transfer queue depth). Spins up +// the mock Vite build (no Rust), drives the transfer panel via window.__demo +// plus the mock transfer engine (src/mock/transfers.ts, which records the +// commands the UI invokes), and asserts on the real rendered DOM: +// counts/positions, pause/resume, pause-all, retry, reorder, throttle, and the +// mid-auto-retry rendering. Exit code 0 = all checks passed. +import { spawn } from "node:child_process"; +import { setTimeout as sleep } from "node:timers/promises"; +import { existsSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import path from "node:path"; +import puppeteer from "puppeteer-core"; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const ROOT = path.resolve(__dirname, ".."); +const OUT = + process.env.SHOT_DIR || + "C:/Users/Juan/AppData/Local/Temp/claude/C--Users-Juan-Documents-GitHub-Faro/cbf0dfb0-7a23-418f-834f-25dcd5128966/scratchpad"; +const PORT = 1425; +const URL = `http://localhost:${PORT}/`; + +const CHROME_CANDIDATES = [ + "C:/Program Files/Google/Chrome/Application/chrome.exe", + "C:/Program Files (x86)/Google/Chrome/Application/chrome.exe", + "C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe", + "C:/Program Files/Microsoft/Edge/Application/msedge.exe", +]; +const chrome = CHROME_CANDIDATES.find((p) => existsSync(p)); +if (!chrome) { + console.error("No Chrome/Edge found."); + process.exit(2); +} +const isWin = process.platform === "win32"; + +let failures = 0; +function check(name, cond, detail = "") { + const ok = !!cond; + if (!ok) failures++; + console.log(` ${ok ? "✓ PASS" : "✗ FAIL"} ${name}${detail ? " — " + detail : ""}`); +} + +async function waitForServer(url, ms = 60_000) { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + try { + const r = await fetch(url); + if (r.ok) return; + } catch {} + await sleep(500); + } + throw new Error(`mock server never came up at ${url}`); +} +function killTree(child) { + if (!child || child.killed) return; + if (isWin) spawn("taskkill", ["/pid", String(child.pid), "/T", "/F"], { stdio: "ignore" }); + else try { process.kill(-child.pid, "SIGKILL"); } catch {} +} + +async function main() { + console.log("› starting mock Vite server…"); + const server = spawn("npm", ["run", "dev:mock"], { + cwd: ROOT, + shell: true, + stdio: "ignore", + detached: !isWin, + }); + let browser; + try { + await waitForServer(URL); + browser = await puppeteer.launch({ + executablePath: chrome, + headless: true, + defaultViewport: { width: 1440, height: 900, deviceScaleFactor: 1 }, + args: ["--hide-scrollbars"], + }); + const page = await browser.newPage(); + page.on("pageerror", (e) => { + console.log(" page error:", e.message); + failures++; + }); + + const drive = (fn, ...a) => page.evaluate(fn, ...a); + const shot = async (name) => { + await page.screenshot({ path: path.join(OUT, `verify-transfers-${name}.png`) }); + console.log(" · shot", `verify-transfers-${name}.png`); + }; + + let booted = false; + for (let attempt = 0; attempt < 5 && !booted; attempt++) { + await page.goto(URL, { waitUntil: "networkidle0" }); + try { + await page.waitForFunction(() => !!window.__demo?.seedTransfers, { timeout: 8_000 }); + booted = true; + } catch { + console.log(` · boot attempt ${attempt + 1} missed __demo, reloading…`); + await sleep(1500); + } + } + if (!booted) throw new Error("app never exposed window.__demo"); + await sleep(800); // let App effects (loadInitial + initListeners) run + + // In-page helpers: locate a transfer row by its source path, click one of + // its buttons by title, and read the recorded mock transfer calls. + await drive(() => { + window.__vt = { + row(src) { + const el = [...document.querySelectorAll(".font-mono")].find( + (e) => e.textContent === src + ); + return el?.closest("div.flex.items-center.gap-3") ?? null; + }, + btn(src, title) { + const r = window.__vt.row(src); + return r?.querySelector(`button[title="${title}"]`) ?? null; + }, + click(src, title) { + window.__vt.btn(src, title)?.click(); + }, + clickHeader(title) { + document.querySelector(`button[title="${title}"]`)?.click(); + }, + calls(cmd) { + return window.__demo.transferCalls.filter((c) => c.cmd === cmd); + }, + }; + }); + const rowText = (src) => + drive((s) => window.__vt.row(s)?.innerText ?? "", src); + + const SRC = { + a1: "/srv/batch/a1.bin", + a2: "/srv/batch/a2.bin", + q1: "/srv/batch/q1.bin", + q2: "/srv/batch/q2.bin", + e1: "/srv/batch/e1.bin", + r1: "/srv/batch/r1.bin", + }; + const seedBatch = () => + drive((SRC) => { + const mk = (id, status, extra = {}) => ({ + id, + kind: "download", + source: `/srv/batch/${id}.bin`, + destination: `C:\\Users\\demo\\Downloads\\${id}.bin`, + size: 1000, + transferred: 400, + status, + startedAt: Date.now(), + ...extra, + }); + window.__demo.seedTransfers( + [ + mk("a1", "transferring"), + mk("a2", "transferring"), + mk("q1", "queued", { transferred: 0 }), + mk("q2", "queued", { transferred: 0 }), + mk("e1", "error", { transferred: 0, error: "connection reset" }), + ], + { waiting: ["q1", "q2"], concurrency: 2, throttleKbps: 0 } + ); + const st = window.__demo.useTransfers.getState(); + st.setPanelOpen(true); + return st.loadInitial(); + }, SRC); + + // ---- 1. Seeded batch: header counts + queue positions ---- + await seedBatch(); + await sleep(400); + const hdr = await drive(() => document.body.innerText); + check( + "header shows '2 active · 2 queued'", + hdr.includes("2 active · 2 queued"), + hdr.match(/\d+ active · \d+ queued/)?.[0] ?? "no badge" + ); + check("first queued row shows '#1 in queue'", (await rowText(SRC.q1)).includes("#1 in queue")); + check("second queued row shows '#2 in queue'", (await rowText(SRC.q2)).includes("#2 in queue")); + await shot("1-batch"); + + // ---- 2. Pause a transferring row → Paused; resume → transferring ---- + await drive((s) => window.__vt.click(s, "Pause"), SRC.a1); + await sleep(300); + let t = await rowText(SRC.a1); + check("paused row shows Paused label", t.includes("Paused"), t.replace(/\n/g, " | ")); + check( + "paused row offers Resume", + await drive((s) => !!window.__vt.btn(s, "Resume (restarts from byte 0)"), SRC.a1) + ); + check( + "pause invoked transfer_pause on the mock", + (await drive((s) => window.__vt.calls("transfer_pause").map((c) => c.args.transferId), SRC.a1)).includes("a1") + ); + await drive((s) => window.__vt.click(s, "Resume (restarts from byte 0)"), SRC.a1); + await sleep(300); + t = await rowText(SRC.a1); + check("resumed row is transferring again (0%)", t.includes("0%"), t.replace(/\n/g, " | ")); + check( + "resume invoked transfer_resume on the mock", + (await drive(() => window.__vt.calls("transfer_resume").map((c) => c.args.transferId))).includes("a1") + ); + + // ---- 3. Pause-all → header toggle flips; queued rows stay queued ---- + await drive(() => window.__vt.clickHeader("Pause all")); + await sleep(300); + check( + "pause-all flips the header toggle to Resume all", + await drive(() => !!document.querySelector('button[title="Resume all"]')) + ); + check("queued rows stay queued under pause-all", (await rowText(SRC.q1)).includes("#1 in queue")); + await shot("3-pause-all"); + await drive(() => window.__vt.clickHeader("Resume all")); + await sleep(300); + check( + "resume-all flips the header toggle back to Pause all", + await drive(() => !!document.querySelector('button[title="Pause all"]')) + ); + + // ---- 4. Retry an error row → transfer_retry recorded, row re-queued ---- + await drive((s) => window.__vt.click(s, "Retry"), SRC.e1); + await sleep(300); + check( + "retry invoked transfer_retry on the mock", + (await drive(() => window.__vt.calls("transfer_retry").map((c) => c.args.transferId))).includes("e1") + ); + t = await rowText(SRC.e1); + check("retried row returns to the queue tail (#3)", t.includes("#3 in queue"), t.replace(/\n/g, " | ")); + check("retried row no longer offers Retry", await drive((s) => !window.__vt.btn(s, "Retry"), SRC.e1)); + + // ---- 5. Reorder: move q2 up; FIFO ends disable the buttons ---- + await drive((s) => window.__vt.click(s, "Move up"), SRC.q2); + await sleep(300); + const moveCalls = await drive(() => window.__vt.calls("transfer_move")); + check( + "move invoked transfer_move(up) on q2", + moveCalls.some((c) => c.args.transferId === "q2" && c.args.direction === "up"), + JSON.stringify(moveCalls) + ); + check("q2 is now '#1 in queue'", (await rowText(SRC.q2)).includes("#1 in queue")); + check( + "Move up disabled at the FIFO head", + await drive((s) => window.__vt.btn(s, "Move up")?.disabled === true, SRC.q2) + ); + check( + "Move down disabled at the FIFO tail", + await drive((s) => window.__vt.btn(s, "Move down")?.disabled === true, SRC.e1) + ); + await shot("5-reorder"); + + // ---- 6. Throttle input commits → transfer_set_throttle ---- + await drive(() => { + const input = document.querySelector('input[type="number"]'); + input.focus(); + const setter = Object.getOwnPropertyDescriptor( + window.HTMLInputElement.prototype, + "value" + ).set; + setter.call(input, "512"); + input.dispatchEvent(new Event("input", { bubbles: true })); + input.blur(); + }); + await sleep(300); + const throttleCalls = await drive(() => window.__vt.calls("transfer_set_throttle")); + check( + "throttle commit invoked transfer_set_throttle(512)", + throttleCalls.some((c) => c.args.kbps === 512), + JSON.stringify(throttleCalls) + ); + check( + "store throttleKbps followed the queue event", + (await drive(() => window.__demo.useTransfers.getState().throttleKbps)) === 512 + ); + + // ---- 7. Mid-auto-retry row renders the 'retrying in Ns' text as warning ---- + await drive((SRC) => { + window.__demo.seedTransfers( + [ + { + id: "r1", + kind: "upload", + source: SRC.r1, + destination: "/srv/app/r1.bin", + size: 2048, + transferred: 512, + status: "transferring", + retryAttempt: 2, + error: "retrying in 5s (attempt 2/3)", + startedAt: Date.now(), + }, + ], + { waiting: [] } + ); + return window.__demo.useTransfers.getState().loadInitial(); + }, SRC); + await sleep(400); + t = await rowText(SRC.r1); + check("mid-auto-retry row renders the retrying text", t.includes("retrying in 5s (attempt 2/3)"), t.replace(/\n/g, " | ")); + check( + "retrying text is styled as warning, not failure", + await drive( + (s) => !!window.__vt.row(s)?.querySelector(".text-warning"), + SRC.r1 + ) + ); + await shot("7-retrying"); + + if (failures === 0) console.log("\n✅ all transfer-queue checks passed"); + else console.log(`\n❌ ${failures} check(s) failed`); + } finally { + if (browser) await browser.close(); + killTree(server); + } + process.exit(failures === 0 ? 0 : 1); +} + +main().catch((e) => { + console.error("verify crashed:", e); + process.exit(3); +}); diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 1f85e91..7d1fa24 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -167,6 +167,10 @@ windows = { version = "0.58", optional = true, features = [ "Win32_Storage_CloudFilters", ] } +[dev-dependencies] +# Paused-clock tokio tests for the transfer token bucket (Plan 17). +tokio = { version = "1", features = ["test-util"] } + [features] default = ["custom-protocol"] custom-protocol = ["tauri/custom-protocol"] diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 2e74829..eb10cb1 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -737,9 +737,114 @@ pub async fn start_upload( #[tauri::command] pub async fn cancel_transfer( transfer_id: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state + .transfers + .cancel(&transfer_id, &app) + .await + .map_err(err) +} + +/// Pause a queued or running transfer (Plan 17 Phase 2). A running one parks +/// at the next chunk boundary; resume re-runs its file from byte 0. +#[tauri::command] +pub async fn transfer_pause( + transfer_id: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state.transfers.pause(&transfer_id, &app).await.map_err(err) +} + +#[tauri::command] +pub async fn transfer_resume( + transfer_id: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state + .transfers + .resume(&transfer_id, &app) + .await + .map_err(err) +} + +/// Re-enqueue a failed or canceled transfer with its original source/ +/// destination (Plan 17 Phase 3). Same id — the panel row resets in place. +#[tauri::command] +pub async fn transfer_retry( + transfer_id: String, + app: AppHandle, state: State<'_, AppState>, ) -> Result<(), String> { - state.transfers.cancel(&transfer_id).await.map_err(err) + state.transfers.retry(&transfer_id, &app).await.map_err(err) +} + +/// Reorder a waiting transfer in the queue FIFO (Plan 17). `direction` is +/// "up" (sooner) or "down" (later); active transfers are untouched. +#[tauri::command] +pub async fn transfer_move( + transfer_id: String, + direction: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state + .transfers + .move_in_queue(&transfer_id, direction == "up", &app) + .await + .map_err(err) +} + +/// Pause admission of new transfers; running ones keep going. +#[tauri::command] +pub async fn transfer_pause_all( + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state.transfers.pause_all(&app).await; + Ok(()) +} + +#[tauri::command] +pub async fn transfer_resume_all( + app: AppHandle, + state: State<'_, AppState>, +) -> Result<(), String> { + state.transfers.resume_all(&app).await; + Ok(()) +} + +/// Live-adjust how many transfers may run at once (1..=32). +#[tauri::command] +pub async fn transfer_set_concurrency( + count: u32, + state: State<'_, AppState>, +) -> Result<(), String> { + state.transfers.set_concurrency(count as usize); + Ok(()) +} + +/// Live-adjust the global bandwidth cap in KiB/s (0 = unlimited). Takes +/// effect on the next chunk of every active transfer (Plan 17 Phase 4). +#[tauri::command] +pub async fn transfer_set_throttle( + kbps: u64, + state: State<'_, AppState>, +) -> Result<(), String> { + state.transfers.set_throttle_kbps(kbps); + Ok(()) +} + +/// Queue snapshot for the panel's initial load (waiting FIFO + pause-all + +/// concurrency + throttle). +#[tauri::command] +pub async fn transfer_queue_state( + state: State<'_, AppState>, +) -> Result { + Ok(state.transfers.queue_state().await) } #[tauri::command] diff --git a/src-tauri/src/error.rs b/src-tauri/src/error.rs index 46f5f88..8d8a7c7 100644 --- a/src-tauri/src/error.rs +++ b/src-tauri/src/error.rs @@ -117,7 +117,7 @@ fn classify_io(kind: std::io::ErrorKind) -> Option { /// Keyword heuristics over an error message. Backend-side only — the frontend /// gets the resulting `kind` and never repeats this matching. -fn classify_message(message: &str) -> ErrorKind { +pub(crate) fn classify_message(message: &str) -> ErrorKind { let s = message.to_ascii_lowercase(); let has = |needles: &[&str]| needles.iter().any(|n| s.contains(n)); diff --git a/src-tauri/src/foldersync.rs b/src-tauri/src/foldersync.rs index 204fbc6..4c7bb78 100644 --- a/src-tauri/src/foldersync.rs +++ b/src-tauri/src/foldersync.rs @@ -605,7 +605,7 @@ async fn wait_for_transfers(state: &AppState, ids: Vec) { for id in &ids { if let Some(t) = state.transfers.snapshot(id).await { match t.status { - TransferStatus::Queued | TransferStatus::Transferring => remaining += 1, + TransferStatus::Queued | TransferStatus::Transferring | TransferStatus::Paused => remaining += 1, _ => {} } } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8dcc94f..858349c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -250,6 +250,23 @@ pub fn run() { }; app.manage(state); + // Apply persisted transfer-queue settings (Plan 17). The frontend + // writes `transferConcurrency`/`transferThrottleKbps` via the + // settings table; the manager starts from those values. + { + let st = app.state::(); + if let Ok(Some(raw)) = st.db.settings_get("transferConcurrency") { + if let Ok(n) = serde_json::from_str::(&raw) { + st.transfers.set_concurrency(n); + } + } + if let Ok(Some(raw)) = st.db.settings_get("transferThrottleKbps") { + if let Ok(kbps) = serde_json::from_str::(&raw) { + st.transfers.set_throttle_kbps(kbps); + } + } + } + // Bring the Agent Bridge back up if the user left its master switch // on, so the `faro-cli agent …` path keeps working across restarts. // Spawned off the async runtime so the sync setup() returns at once. @@ -389,6 +406,15 @@ pub fn run() { commands::start_directory_upload, commands::cancel_transfer, commands::list_transfers, + commands::transfer_move, + commands::transfer_pause, + commands::transfer_resume, + commands::transfer_retry, + commands::transfer_pause_all, + commands::transfer_resume_all, + commands::transfer_set_concurrency, + commands::transfer_set_throttle, + commands::transfer_queue_state, commands::rename_path, commands::delete_path, commands::create_directory, diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs index 7830b64..3c74a4c 100644 --- a/src-tauri/src/transfer.rs +++ b/src-tauri/src/transfer.rs @@ -5,13 +5,14 @@ use crate::session::{ }; use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tauri::{AppHandle, Emitter}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::sync::Mutex; +use tokio::sync::{watch, Mutex, OwnedSemaphorePermit, Semaphore}; use tokio::task::JoinHandle; use uuid::Uuid; @@ -36,12 +37,136 @@ pub enum TransferKind { pub enum TransferStatus { Queued, Transferring, + Paused, Done, Skipped, Error, Canceled, } +/// Marker error: a paused transfer was resumed — the copy loop unwinds with +/// this and the runner re-runs the file from byte 0 (Plan 17 Phase 2). +/// Honest on every backend: no per-backend seek support needed. +#[derive(Debug)] +struct RestartFromPause; + +impl std::fmt::Display for RestartFromPause { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("transfer resumed after pause; restarting file from the beginning") + } +} + +impl std::error::Error for RestartFromPause {} + +/// Payload of the `transfer://queue` event: the FIFO of waiting transfer ids +/// plus the manager-level state the panel header renders (Plan 17). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct QueueState { + pub waiting: Vec, + pub paused_all: bool, + pub concurrency: usize, + pub throttle_kbps: u64, +} + +/// Global bandwidth cap shared by every active copy loop (Plan 17 Phase 4): +/// a token bucket refilling at `rate` bytes/sec (0 = unlimited). Because all +/// transfers draw from this one bucket, the cap is split across active +/// transfers rather than applied per transfer. +struct TokenBucket { + inner: Mutex, + rate_bps: AtomicU64, +} + +struct BucketInner { + tokens: f64, + // tokio's Instant (not std's) so `start_paused` tests drive refills. + last: tokio::time::Instant, +} + +impl TokenBucket { + fn new() -> Self { + Self { + inner: Mutex::new(BucketInner { + tokens: 0.0, + last: tokio::time::Instant::now(), + }), + rate_bps: AtomicU64::new(0), + } + } + + fn rate_kbps(&self) -> u64 { + self.rate_bps.load(Ordering::Relaxed) / 1024 + } + + fn set_rate_kbps(&self, kbps: u64) { + self.rate_bps + .store(kbps.saturating_mul(1024), Ordering::Relaxed); + } + + /// Wait until `bytes` may flow under the cap. Drawn in tranches capped at + /// one second's worth of rate (min 64 KiB) so a large chunk is charged in + /// full while a 1 KiB file never waits a whole token window. + async fn acquire(&self, bytes: u64) { + let mut remaining = bytes as f64; + while remaining > 0.0 { + let rate = self.rate_bps.load(Ordering::Relaxed); + if rate == 0 { + return; + } + let wait = { + let mut g = self.inner.lock().await; + let now = tokio::time::Instant::now(); + let elapsed = now.duration_since(g.last).as_secs_f64(); + let rate_f = rate as f64; + let cap = rate_f.max(64.0 * 1024.0); + g.tokens = (g.tokens + elapsed * rate_f).min(cap); + g.last = now; + let tranche = remaining.min(cap); + if g.tokens >= tranche { + g.tokens -= tranche; + remaining -= tranche; + continue; + } + Duration::from_secs_f64((tranche - g.tokens) / rate_f) + }; + // Cap the sleep so a live rate change takes effect promptly. + tokio::time::sleep(wait.min(Duration::from_millis(250))).await; + } + } +} + +/// A pause gate shared by the scheduler (pause-all) and individual transfers +/// (Phase 2). watch-channel based so waiters never miss a wakeup. +#[derive(Debug, Clone)] +pub struct PauseGate { + tx: watch::Sender, + // Keeps the channel open: with zero receivers `send()` silently fails. + _rx: watch::Receiver, +} + +impl PauseGate { + fn new() -> Self { + let (tx, _rx) = watch::channel(false); + Self { tx, _rx } + } + fn is_paused(&self) -> bool { + *self.tx.borrow() + } + fn set(&self, paused: bool) { + let _ = self.tx.send(paused); + } + /// Park until the gate opens. Returns immediately if already open. + async fn wait_open(&self) { + let mut rx = self.tx.subscribe(); + while *rx.borrow_and_update() { + if rx.changed().await.is_err() { + break; + } + } + } +} + #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct Transfer { @@ -54,12 +179,57 @@ pub struct Transfer { pub status: TransferStatus, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, + /// Auto-retry round in progress (1- or 2-of-2), for the panel's + /// "retrying in Ns (attempt N/3)" state (Plan 17 Phase 3). + #[serde(skip_serializing_if = "Option::is_none")] + pub retry_attempt: Option, pub started_at: i64, } +/// Everything needed to re-run a failed/canceled transfer with its already +/// policy-resolved destination (overwrite/skip/rename was applied at enqueue +/// time) — Plan 17 Phase 3 manual retry. +#[derive(Clone)] +enum RetryInfo { + Download { + session: Arc, + remote_path: String, + final_path: PathBuf, + }, + Upload { + session: Arc, + local: PathBuf, + final_remote: String, + }, +} + +/// Default bound on concurrently running transfers (Plan 17); the rest wait +/// in the FIFO as `Queued`. Overridden by the `transferConcurrency` setting. +const DEFAULT_CONCURRENCY: usize = 3; + +/// How many times a transfer auto-retries on transient (Network/Timeout) +/// errors before surfacing the failure (Plan 17 Phase 3). +const MAX_AUTO_RETRIES: u32 = 2; + pub struct TransferManager { transfers: Mutex>, tasks: Mutex>>, + /// FIFO of transfer ids waiting for a slot (Plan 17). An id is popped only + /// once it is at the front, the pause-all gate is open, and a concurrency + /// permit is available — until then it waits as `Queued`. + waiting: Mutex>, + semaphore: Arc, + concurrency: AtomicUsize, + /// Manager-level pause gate. Admission checks it; Phase 2 checkpoints do too. + pause_all: PauseGate, + /// Per-transfer pause gates, created at enqueue time (Plan 17 Phase 2). + pauses: Mutex>, + /// Original resolved inputs per transfer, for manual retry (Phase 3). + retry: Mutex>, + /// Bumped on every queue change so admission waiters re-check their turn. + queue_gen: watch::Sender, + /// Global bandwidth cap every copy loop draws from per chunk (Phase 4). + bucket: TokenBucket, } fn now_ts() -> i64 { @@ -127,6 +297,14 @@ impl TransferManager { Self { transfers: Mutex::new(HashMap::new()), tasks: Mutex::new(HashMap::new()), + waiting: Mutex::new(VecDeque::new()), + semaphore: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY)), + concurrency: AtomicUsize::new(DEFAULT_CONCURRENCY), + pause_all: PauseGate::new(), + pauses: Mutex::new(HashMap::new()), + retry: Mutex::new(HashMap::new()), + queue_gen: watch::channel(0).0, + bucket: TokenBucket::new(), } } @@ -156,16 +334,318 @@ impl TransferManager { self.transfers.lock().await.insert(t.id.clone(), t); } - pub async fn cancel(&self, id: &str) -> Result<()> { + // ---------- Queue scheduling (Plan 17) ---------- + + fn build_queue_state(&self, waiting: &VecDeque) -> QueueState { + QueueState { + waiting: waiting.iter().cloned().collect(), + paused_all: self.pause_all.is_paused(), + concurrency: self.concurrency.load(Ordering::Relaxed), + throttle_kbps: self.bucket.rate_kbps(), + } + } + + /// Current queue snapshot for the panel's initial load. + pub async fn queue_state(&self) -> QueueState { + let w = self.waiting.lock().await; + self.build_queue_state(&w) + } + + /// Emit `transfer://queue` and bump the generation so admission waiters + /// re-check whether it is their turn. + async fn bump_queue(&self, app: &AppHandle) { + let state = { + let w = self.waiting.lock().await; + self.build_queue_state(&w) + }; + self.queue_gen.send_modify(|g| *g += 1); + let _ = app.emit("transfer://queue", &state); + } + + /// Wait until this transfer is at the front of the FIFO with the pause-all + /// gate open, take a concurrency permit, and pop it from the queue. + /// Returns `None` if the id left the queue (canceled while waiting). + /// The permit is returned so the caller holds it for the transfer's life. + async fn admit(&self, id: &str) -> Option { + let mut rx = self.queue_gen.subscribe(); + loop { + { + let w = self.waiting.lock().await; + if !w.iter().any(|x| x == id) { + return None; + } + if !self.is_my_turn(&w, id).await { + drop(w); + if rx.changed().await.is_err() { + return None; + } + continue; + } + } + // My turn: take a permit, then pop. + let permit = self.semaphore.clone().acquire_owned().await.ok()?; + let mut w = self.waiting.lock().await; + if self.is_my_turn(&w, id).await { + if let Some(pos) = w.iter().position(|x| x == id) { + w.remove(pos); + } + return Some(permit); + } + // Lost a race (pause engaged mid-acquire): release, re-evaluate. + drop(permit); + if !w.iter().any(|x| x == id) { + return None; + } + drop(w); + } + } + + /// Is `id` the first waiting transfer allowed to run? Strict FIFO except + /// that per-transfer-paused rows are skipped (a paused row must not + /// head-of-line block the queue); pause-all blocks everyone. Caller must + /// hold the `waiting` lock; lock order is waiting → pauses. + async fn is_my_turn(&self, w: &VecDeque, id: &str) -> bool { + if self.pause_all.is_paused() { + return false; + } + let pauses = self.pauses.lock().await; + let first_open = w + .iter() + .position(|x| pauses.get(x).is_none_or(|g| !g.is_paused())); + first_open == w.iter().position(|x| x == id) + } + + /// Reorder a waiting transfer (active transfers are untouched). + pub async fn move_in_queue(&self, id: &str, up: bool, app: &AppHandle) -> Result<()> { + { + let mut w = self.waiting.lock().await; + let Some(pos) = w.iter().position(|x| x == id) else { + anyhow::bail!("transfer {id} is not waiting in the queue"); + }; + let swap_with = if up { + pos.checked_sub(1) + } else if pos + 1 < w.len() { + Some(pos + 1) + } else { + None + }; + if let Some(other) = swap_with { + w.swap(pos, other); + } + } + self.bump_queue(app).await; + Ok(()) + } + + /// Pause admission of new transfers (running ones keep going until Phase + /// 2's chunk checkpoints let them park too). + pub async fn pause_all(&self, app: &AppHandle) { + self.pause_all.set(true); + self.bump_queue(app).await; + } + + pub async fn resume_all(&self, app: &AppHandle) { + self.pause_all.set(false); + self.bump_queue(app).await; + } + + pub fn is_paused_all(&self) -> bool { + self.pause_all.is_paused() + } + + /// Chunk-boundary checkpoint shared by every copy loop (Plan 17). Draws + /// `bytes` from the global bandwidth bucket (Phase 4), then — when the + /// transfer (or the whole manager) is paused — parks until resumed and + /// returns `RestartFromPause` so the runner re-runs the file from byte 0. + async fn checkpoint(&self, id: &str, bytes: u64) -> Result<()> { + self.bucket.acquire(bytes).await; + let gate = self.pauses.lock().await.get(id).cloned(); + let parked = self.pause_all.is_paused() || gate.as_ref().is_some_and(|g| g.is_paused()); + if !parked { + return Ok(()); + } + // Park until BOTH gates are open (resume requires both). + loop { + self.pause_all.wait_open().await; + if let Some(g) = &gate { + g.wait_open().await; + } + if !self.pause_all.is_paused() && gate.as_ref().is_none_or(|g| !g.is_paused()) { + break; + } + } + Err(RestartFromPause.into()) + } + + /// Pause a queued or transferring transfer. A running one parks at the + /// next chunk boundary; a queued one is skipped by admission until resumed. + pub async fn pause(&self, id: &str, app: &AppHandle) -> Result<()> { + match self.get(id).await.map(|t| t.status) { + Some(TransferStatus::Transferring) | Some(TransferStatus::Queued) => {} + _ => anyhow::bail!("transfer {id} is not running or queued"), + } + let gate = self + .pauses + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| anyhow::anyhow!("transfer {id} not found"))?; + gate.set(true); + self.update(id, |t| t.status = TransferStatus::Paused).await; + if let Some(t) = self.get(id).await { + let _ = app.emit("transfer://updated", &t); + } + Ok(()) + } + + /// Resume a paused transfer. A parked one re-runs its file from byte 0; + /// a queued one re-enters FIFO admission. + pub async fn resume(&self, id: &str, app: &AppHandle) -> Result<()> { + if self.get(id).await.map(|t| t.status) != Some(TransferStatus::Paused) { + anyhow::bail!("transfer {id} is not paused"); + } + let still_queued = self.waiting.lock().await.iter().any(|x| x == id); + let gate = self + .pauses + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| anyhow::anyhow!("transfer {id} not found"))?; + self.update(id, |t| { + t.status = if still_queued { + TransferStatus::Queued + } else { + TransferStatus::Transferring + }; + t.transferred = 0; + }) + .await; + gate.set(false); + if let Some(t) = self.get(id).await { + let _ = app.emit("transfer://updated", &t); + } + // Wake admission waiters: a queued row may have become runnable. + self.bump_queue(app).await; + Ok(()) + } + + /// Re-enqueue a failed or canceled transfer with its original (already + /// policy-resolved) source/destination. Same id — the panel row resets + /// in place (Plan 17 Phase 3 manual retry). + pub async fn retry(self: &Arc, id: &str, app: &AppHandle) -> Result<()> { + match self.get(id).await.map(|t| t.status) { + Some(TransferStatus::Error) | Some(TransferStatus::Canceled) => {} + _ => anyhow::bail!("only failed or canceled transfers can be retried"), + } + let info = self + .retry + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| anyhow::anyhow!("transfer {id} cannot be retried"))?; if let Some(h) = self.tasks.lock().await.remove(id) { h.abort(); } + // A cancel-while-paused leaves the gate closed — reopen it. + if let Some(g) = self.pauses.lock().await.get(id) { + g.set(false); + } self.update(id, |t| { - if t.status == TransferStatus::Transferring || t.status == TransferStatus::Queued { + t.status = TransferStatus::Queued; + t.transferred = 0; + t.error = None; + t.retry_attempt = None; + }) + .await; + if let Some(t) = self.get(id).await { + let _ = app.emit("transfer://updated", &t); + } + self.waiting.lock().await.push_back(id.to_string()); + self.bump_queue(app).await; + let mgr = Arc::clone(self); + let id_for_task = id.to_string(); + let app_for_task = app.clone(); + let task = match info { + RetryInfo::Download { + session, + remote_path, + final_path, + } => tokio::spawn(run_download_task( + mgr, + id_for_task, + session, + remote_path, + final_path, + app_for_task, + )), + RetryInfo::Upload { + session, + local, + final_remote, + } => tokio::spawn(run_upload_task( + mgr, + id_for_task, + session, + local, + final_remote, + app_for_task, + )), + }; + self.tasks.lock().await.insert(id.to_string(), task); + Ok(()) + } + + /// Live-adjust the global bandwidth cap (KiB/s, 0 = unlimited). Takes + /// effect on the next chunk of every active transfer. + pub fn set_throttle_kbps(&self, kbps: u64) { + self.bucket.set_rate_kbps(kbps); + } + + /// Live-adjust the concurrency bound. Growing adds permits at once; + /// shrinking forgets permits as running transfers release them, so + /// in-flight transfers are never killed to satisfy the new bound. + pub fn set_concurrency(&self, n: usize) { + let n = n.clamp(1, 32); + let old = self.concurrency.swap(n, Ordering::Relaxed); + if n > old { + self.semaphore.add_permits(n - old); + } else if n < old { + let sem = Arc::clone(&self.semaphore); + tokio::spawn(async move { + if let Ok(p) = sem.acquire_many((old - n) as u32).await { + p.forget(); + } + }); + } + } + + pub async fn cancel(&self, id: &str, app: &AppHandle) -> Result<()> { + { + let mut w = self.waiting.lock().await; + if let Some(pos) = w.iter().position(|x| x == id) { + w.remove(pos); + } + } + if let Some(h) = self.tasks.lock().await.remove(id) { + h.abort(); + } + self.update(id, |t| { + if matches!( + t.status, + TransferStatus::Transferring | TransferStatus::Queued | TransferStatus::Paused + ) { t.status = TransferStatus::Canceled; } }) .await; + // There is no `transfer://canceled` event — `updated` carries the row. + if let Some(t) = self.get(id).await { + let _ = app.emit("transfer://updated", &t); + } + self.bump_queue(app).await; Ok(()) } @@ -203,6 +683,7 @@ impl TransferManager { TransferStatus::Queued }, error: None, + retry_attempt: None, started_at: now_ts(), }; self.insert(transfer.clone()).await; @@ -213,143 +694,28 @@ impl TransferManager { return Ok(id); } + self.retry.lock().await.insert( + id.clone(), + RetryInfo::Download { + session: Arc::clone(&session), + remote_path: remote_path.clone(), + final_path: final_path.clone(), + }, + ); + self.waiting.lock().await.push_back(id.clone()); + self.pauses.lock().await.insert(id.clone(), PauseGate::new()); + self.bump_queue(&app).await; + let mgr = Arc::clone(self); let id_for_task = id.clone(); - let task = tokio::spawn(async move { - let res = match &*session { - Session::Ssh(ssh) => { - mgr.run_ssh_download( - &id_for_task, - ssh.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Ftp(ftp) => { - mgr.run_ftp_download( - &id_for_task, - ftp.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Object(obj) => { - mgr.run_object_download( - &id_for_task, - obj.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Webdav(dav) => { - mgr.run_webdav_download( - &id_for_task, - dav.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Http(http) => { - mgr.run_http_download( - &id_for_task, - http.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Dropbox(dbx) => { - mgr.run_dropbox_download( - &id_for_task, - dbx.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::OneDrive(od) => { - mgr.run_onedrive_download( - &id_for_task, - od.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::GDrive(gd) => { - mgr.run_gdrive_download( - &id_for_task, - gd.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Box(bx) => { - mgr.run_box_download( - &id_for_task, - bx.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Shopify(sh) => { - mgr.run_shopify_download( - &id_for_task, - sh.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::HubSpot(hs) => { - mgr.run_hubspot_download( - &id_for_task, - hs.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Dynamics(dynm) => { - mgr.run_dynamics_download( - &id_for_task, - dynm.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - Session::Agent(agent) => { - mgr.run_agent_download( - &id_for_task, - agent.clone(), - &remote_path, - &final_path, - &app, - ) - .await - } - }; - finalize(&mgr, &id_for_task, &app, res).await; - }); + let task = tokio::spawn(run_download_task( + mgr, + id_for_task, + session, + remote_path, + final_path, + app, + )); self.tasks.lock().await.insert(id.clone(), task); Ok(id) } @@ -387,6 +753,7 @@ impl TransferManager { let bytes = base64::engine::general_purpose::STANDARD .decode(&data_b64) .context("decode chunk")?; + self.checkpoint(id, bytes.len() as u64).await?; if !bytes.is_empty() { local_file.write_all(&bytes).await?; offset += bytes.len() as u64; @@ -437,6 +804,7 @@ impl TransferManager { if n == 0 { break; } + self.checkpoint(id, n as u64).await?; local_file.write_all(&buf[..n]).await?; transferred += n as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -488,6 +856,7 @@ impl TransferManager { TransferStatus::Queued }, error: None, + retry_attempt: None, started_at: now_ts(), }; self.insert(transfer.clone()).await; @@ -498,136 +867,28 @@ impl TransferManager { return Ok(id); } + self.retry.lock().await.insert( + id.clone(), + RetryInfo::Upload { + session: Arc::clone(&session), + local: local.clone(), + final_remote: final_remote.clone(), + }, + ); + self.waiting.lock().await.push_back(id.clone()); + self.pauses.lock().await.insert(id.clone(), PauseGate::new()); + self.bump_queue(&app).await; + let mgr = Arc::clone(self); let id_for_task = id.clone(); - let task = tokio::spawn(async move { - let res = match &*session { - Session::Ssh(ssh) => { - mgr.run_ssh_upload( - &id_for_task, - ssh.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Ftp(ftp) => { - mgr.run_ftp_upload( - &id_for_task, - ftp.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Object(obj) => { - mgr.run_object_upload( - &id_for_task, - obj.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Webdav(dav) => { - mgr.run_webdav_upload( - &id_for_task, - dav.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Http(_) => { - Err(anyhow::anyhow!("HTTP source is read-only — upload not supported")) - } - Session::Dropbox(dbx) => { - mgr.run_dropbox_upload( - &id_for_task, - dbx.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::OneDrive(od) => { - mgr.run_onedrive_upload( - &id_for_task, - od.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::GDrive(gd) => { - mgr.run_gdrive_upload( - &id_for_task, - gd.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Box(bx) => { - mgr.run_box_upload( - &id_for_task, - bx.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Shopify(sh) => { - mgr.run_shopify_upload( - &id_for_task, - sh.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::HubSpot(hs) => { - mgr.run_hubspot_upload( - &id_for_task, - hs.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Dynamics(dynm) => { - mgr.run_dynamics_upload( - &id_for_task, - dynm.clone(), - &local, - &final_remote, - &app, - ) - .await - } - Session::Agent(agent) => { - mgr.run_agent_upload( - &id_for_task, - agent.clone(), - &local, - &final_remote, - &app, - ) - .await - } - }; - finalize(&mgr, &id_for_task, &app, res).await; - }); + let task = tokio::spawn(run_upload_task( + mgr, + id_for_task, + session, + local, + final_remote, + app, + )); self.tasks.lock().await.insert(id.clone(), task); Ok(id) } @@ -660,6 +921,7 @@ impl TransferManager { if n == 0 { break; } + self.checkpoint(id, n as u64).await?; let data = base64::engine::general_purpose::STANDARD.encode(&buf[..n]); let resp = session .request(Request::WriteChunk { @@ -721,6 +983,7 @@ impl TransferManager { if n == 0 { break; } + self.checkpoint(id, n as u64).await?; remote_file.write_all(&buf[..n]).await?; transferred += n as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -905,6 +1168,7 @@ impl TransferManager { let _ = (mgr_for_emit, id_for_emit, app_for_emit); + self.checkpoint(id, 0).await?; let res: Result = session .with_stream(move |stream| { let file = std::fs::File::create(&final_path) @@ -932,6 +1196,7 @@ impl TransferManager { .await; let _ = app; // progress events come at completion for FTP + self.checkpoint(id, 0).await?; let local = local_path.to_path_buf(); let remote = remote_path.to_string(); let res: Result = session @@ -978,6 +1243,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("s3 chunk for {key}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1021,6 +1287,7 @@ impl TransferManager { .with_context(|| format!("open {}", local_path.display()))?; if size <= 16 * 1024 * 1024 { + self.checkpoint(id, size).await?; let mut buf = Vec::with_capacity(size as usize); file.read_to_end(&mut buf).await?; session @@ -1059,6 +1326,7 @@ impl TransferManager { if filled == 0 { break; } + self.checkpoint(id, filled as u64).await?; let chunk = bytes::Bytes::copy_from_slice(&buf[..filled]); upload .put_part(chunk.into()) @@ -1120,6 +1388,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("webdav chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1159,6 +1428,7 @@ impl TransferManager { .await .with_context(|| format!("open {}", local_path.display()))?; let body = reqwest::Body::wrap_stream(ReaderStream::new(file)); + self.checkpoint(id, size).await?; let url = session.url_for(remote_path, false); let resp = session @@ -1213,6 +1483,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("http chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1255,6 +1526,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("dropbox chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1309,6 +1581,7 @@ impl TransferManager { // Proactive refresh covers the common case; on a hard 401 we refresh and // retry once, re-opening the file for a fresh streamed body. + self.checkpoint(id, size).await?; let mut attempt = 0; loop { let token = session.access_token().await?; @@ -1357,6 +1630,7 @@ impl TransferManager { .await; let _ = app; // single-shot API: progress is reported at completion. + self.checkpoint(id, 0).await?; let data = crate::remotefs::shopify::read_asset(&session, remote_path).await?; let mut file = tokio::fs::File::create(local_path) .await @@ -1385,6 +1659,7 @@ impl TransferManager { .await .with_context(|| format!("read {}", local_path.display()))?; let size = data.len() as u64; + self.checkpoint(id, size).await?; crate::remotefs::shopify::write_asset(&session, remote_path, &data).await?; self.update(id, |t| t.transferred = size).await; Ok(()) @@ -1405,6 +1680,7 @@ impl TransferManager { .await; let _ = app; // single-shot API: progress is reported at completion. + self.checkpoint(id, 0).await?; let data = crate::remotefs::hubspot::read_file(&session, remote_path).await?; let mut file = tokio::fs::File::create(local_path) .await @@ -1434,6 +1710,7 @@ impl TransferManager { .await .with_context(|| format!("read {}", local_path.display()))?; let size = data.len() as u64; + self.checkpoint(id, size).await?; crate::remotefs::hubspot::write_file(&session, remote_path, &data).await?; self.update(id, |t| t.transferred = size).await; Ok(()) @@ -1454,6 +1731,7 @@ impl TransferManager { .await; let _ = app; // single-shot API: progress is reported at completion. + self.checkpoint(id, 0).await?; let data = crate::remotefs::dynamics::read_file(&session, remote_path).await?; let mut file = tokio::fs::File::create(local_path) .await @@ -1482,6 +1760,7 @@ impl TransferManager { .await .with_context(|| format!("read {}", local_path.display()))?; let size = data.len() as u64; + self.checkpoint(id, size).await?; crate::remotefs::dynamics::write_file(&session, remote_path, &data).await?; self.update(id, |t| t.transferred = size).await; Ok(()) @@ -1513,6 +1792,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("onedrive chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1552,6 +1832,7 @@ impl TransferManager { .unwrap_or(4 * 1024 * 1024); if size <= simple_max { + self.checkpoint(id, size).await?; self.onedrive_simple_upload(id, &session, local_path, remote_path) .await?; } else { @@ -1646,6 +1927,7 @@ impl TransferManager { if filled == 0 { break; } + self.checkpoint(id, filled as u64).await?; let start = offset; let end = offset + filled as u64 - 1; let range = format!("bytes {start}-{end}/{size}"); @@ -1706,6 +1988,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("drive chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1748,6 +2031,7 @@ impl TransferManager { .await .with_context(|| format!("read {}", local_path.display()))?; let token = session.access_token().await?; + self.checkpoint(id, size).await?; let resp = if let Some((file_id, _)) = existing { // Update the existing file's content in place. @@ -1832,6 +2116,7 @@ impl TransferManager { let mut last_emit = Instant::now(); while let Some(chunk) = stream.next().await { let chunk = chunk.with_context(|| format!("box chunk for {remote_path}"))?; + self.checkpoint(id, chunk.len() as u64).await?; file.write_all(&chunk).await?; transferred += chunk.len() as u64; if last_emit.elapsed() > Duration::from_millis(100) { @@ -1874,6 +2159,7 @@ impl TransferManager { .await .with_context(|| format!("read {}", local_path.display()))?; let token = session.access_token().await?; + self.checkpoint(id, size).await?; let file_part = reqwest::multipart::Part::bytes(bytes).file_name(name.clone()); let (url, form) = match existing { @@ -2278,6 +2564,235 @@ fn split_ext(path: &str) -> (&str, &str) { } } +/// Transient = worth an auto-retry (Plan 12's structured error kinds): +/// network and timeout failures; auth/permission/not-found never retry. +fn is_transient(e: &anyhow::Error) -> bool { + matches!( + crate::error::classify_message(&format!("{e:#}")), + crate::error::ErrorKind::Network | crate::error::ErrorKind::Timeout + ) +} + +/// Shared download runner (Plan 17): admission → run loop → finalize → wake +/// the queue. The loop re-runs the file from byte 0 after a resume-from-pause +/// (Phase 2) and auto-retries transient errors with 5s/20s backoff (Phase 3). +/// The concurrency permit is held through backoff — a deliberate trade-off so +/// a retrying transfer keeps its slot. +async fn run_download_task( + mgr: Arc, + id: String, + session: Arc, + remote_path: String, + final_path: PathBuf, + app: AppHandle, +) { + let Some(_permit) = mgr.admit(&id).await else { + return; + }; + let mut auto_retries = 0u32; + let res = loop { + let attempt = dispatch_download(&mgr, &id, &session, &remote_path, &final_path, &app).await; + match attempt { + Err(e) if e.downcast_ref::().is_some() => { + mgr.update(&id, |t| t.transferred = 0).await; + continue; + } + Err(e) if auto_retries < MAX_AUTO_RETRIES && is_transient(&e) => { + auto_retries += 1; + let delay = if auto_retries == 1 { 5 } else { 20 }; + mgr.update(&id, |t| { + t.transferred = 0; + t.retry_attempt = Some(auto_retries); + t.error = Some(format!("retrying in {delay}s (attempt {}/3)", auto_retries + 1)); + }) + .await; + if let Some(t) = mgr.get(&id).await { + let _ = app.emit("transfer://updated", &t); + } + tokio::time::sleep(Duration::from_secs(delay)).await; + mgr.update(&id, |t| t.error = None).await; + continue; + } + other => break other, + } + }; + finalize(&mgr, &id, &app, res).await; + mgr.bump_queue(&app).await; +} + +/// Upload twin of `run_download_task`. +async fn run_upload_task( + mgr: Arc, + id: String, + session: Arc, + local: PathBuf, + final_remote: String, + app: AppHandle, +) { + let Some(_permit) = mgr.admit(&id).await else { + return; + }; + let mut auto_retries = 0u32; + let res = loop { + let attempt = dispatch_upload(&mgr, &id, &session, &local, &final_remote, &app).await; + match attempt { + Err(e) if e.downcast_ref::().is_some() => { + mgr.update(&id, |t| t.transferred = 0).await; + continue; + } + Err(e) if auto_retries < MAX_AUTO_RETRIES && is_transient(&e) => { + auto_retries += 1; + let delay = if auto_retries == 1 { 5 } else { 20 }; + mgr.update(&id, |t| { + t.transferred = 0; + t.retry_attempt = Some(auto_retries); + t.error = Some(format!("retrying in {delay}s (attempt {}/3)", auto_retries + 1)); + }) + .await; + if let Some(t) = mgr.get(&id).await { + let _ = app.emit("transfer://updated", &t); + } + tokio::time::sleep(Duration::from_secs(delay)).await; + mgr.update(&id, |t| t.error = None).await; + continue; + } + other => break other, + } + }; + finalize(&mgr, &id, &app, res).await; + mgr.bump_queue(&app).await; +} + +/// Backend dispatch for a single-file download. Extracted so the runner (and +/// Phase 3 retry) can re-invoke it. +async fn dispatch_download( + mgr: &Arc, + id: &str, + session: &Arc, + remote_path: &str, + final_path: &Path, + app: &AppHandle, +) -> Result<()> { + match &**session { + Session::Ssh(ssh) => { + mgr.run_ssh_download(id, ssh.clone(), remote_path, final_path, app) + .await + } + Session::Ftp(ftp) => { + mgr.run_ftp_download(id, ftp.clone(), remote_path, final_path, app) + .await + } + Session::Object(obj) => { + mgr.run_object_download(id, obj.clone(), remote_path, final_path, app) + .await + } + Session::Webdav(dav) => { + mgr.run_webdav_download(id, dav.clone(), remote_path, final_path, app) + .await + } + Session::Http(http) => { + mgr.run_http_download(id, http.clone(), remote_path, final_path, app) + .await + } + Session::Dropbox(dbx) => { + mgr.run_dropbox_download(id, dbx.clone(), remote_path, final_path, app) + .await + } + Session::OneDrive(od) => { + mgr.run_onedrive_download(id, od.clone(), remote_path, final_path, app) + .await + } + Session::GDrive(gd) => { + mgr.run_gdrive_download(id, gd.clone(), remote_path, final_path, app) + .await + } + Session::Box(bx) => { + mgr.run_box_download(id, bx.clone(), remote_path, final_path, app) + .await + } + Session::Shopify(sh) => { + mgr.run_shopify_download(id, sh.clone(), remote_path, final_path, app) + .await + } + Session::HubSpot(hs) => { + mgr.run_hubspot_download(id, hs.clone(), remote_path, final_path, app) + .await + } + Session::Dynamics(dynm) => { + mgr.run_dynamics_download(id, dynm.clone(), remote_path, final_path, app) + .await + } + Session::Agent(agent) => { + mgr.run_agent_download(id, agent.clone(), remote_path, final_path, app) + .await + } + } +} + +/// Backend dispatch for a single-file upload. +async fn dispatch_upload( + mgr: &Arc, + id: &str, + session: &Arc, + local: &Path, + final_remote: &str, + app: &AppHandle, +) -> Result<()> { + match &**session { + Session::Ssh(ssh) => { + mgr.run_ssh_upload(id, ssh.clone(), local, final_remote, app) + .await + } + Session::Ftp(ftp) => { + mgr.run_ftp_upload(id, ftp.clone(), local, final_remote, app) + .await + } + Session::Object(obj) => { + mgr.run_object_upload(id, obj.clone(), local, final_remote, app) + .await + } + Session::Webdav(dav) => { + mgr.run_webdav_upload(id, dav.clone(), local, final_remote, app) + .await + } + Session::Http(_) => Err(anyhow::anyhow!( + "HTTP source is read-only — upload not supported" + )), + Session::Dropbox(dbx) => { + mgr.run_dropbox_upload(id, dbx.clone(), local, final_remote, app) + .await + } + Session::OneDrive(od) => { + mgr.run_onedrive_upload(id, od.clone(), local, final_remote, app) + .await + } + Session::GDrive(gd) => { + mgr.run_gdrive_upload(id, gd.clone(), local, final_remote, app) + .await + } + Session::Box(bx) => { + mgr.run_box_upload(id, bx.clone(), local, final_remote, app) + .await + } + Session::Shopify(sh) => { + mgr.run_shopify_upload(id, sh.clone(), local, final_remote, app) + .await + } + Session::HubSpot(hs) => { + mgr.run_hubspot_upload(id, hs.clone(), local, final_remote, app) + .await + } + Session::Dynamics(dynm) => { + mgr.run_dynamics_upload(id, dynm.clone(), local, final_remote, app) + .await + } + Session::Agent(agent) => { + mgr.run_agent_upload(id, agent.clone(), local, final_remote, app) + .await + } + } +} + async fn finalize( mgr: &Arc, id: &str, @@ -2308,3 +2823,135 @@ async fn finalize( } mgr.tasks.lock().await.remove(id); } + +#[cfg(test)] +mod tests { + use super::*; + + // ---------- TokenBucket (Phase 4) ---------- + + #[tokio::test(start_paused = true)] + async fn token_bucket_caps_throughput() { + let bucket = TokenBucket::new(); + bucket.set_rate_kbps(512); // 512 KiB/s + let start = tokio::time::Instant::now(); + // 1 MiB at 512 KiB/s must take ~2s of (virtual) time, charged in full. + bucket.acquire(1024 * 1024).await; + let elapsed = start.elapsed(); + assert!(elapsed >= Duration::from_millis(1900), "too fast: {elapsed:?}"); + assert!(elapsed <= Duration::from_millis(2600), "too slow: {elapsed:?}"); + } + + #[tokio::test(start_paused = true)] + async fn token_bucket_unlimited_is_instant() { + let bucket = TokenBucket::new(); // rate 0 = unlimited + let start = tokio::time::Instant::now(); + bucket.acquire(10 * 1024 * 1024).await; + assert_eq!(start.elapsed(), Duration::ZERO); + } + + #[tokio::test(start_paused = true)] + async fn token_bucket_tiny_file_skips_full_window() { + let bucket = TokenBucket::new(); + bucket.set_rate_kbps(64); // 64 KiB/s + let start = tokio::time::Instant::now(); + bucket.acquire(1024).await; // 1 KiB → ~16ms, not a whole window + assert!(start.elapsed() < Duration::from_millis(100)); + } + + // ---------- PauseGate + checkpoint (Phase 2) ---------- + + #[tokio::test] + async fn pause_gate_parks_until_opened() { + let gate = PauseGate::new(); + gate.set(true); + let g2 = gate.clone(); + let handle = tokio::spawn(async move { g2.wait_open().await }); + for _ in 0..10 { + tokio::task::yield_now().await; + } + assert!(!handle.is_finished()); + gate.set(false); + handle.await.unwrap(); + } + + #[tokio::test] + async fn checkpoint_parks_then_signals_restart() { + let mgr = Arc::new(TransferManager::new()); + mgr.pauses.lock().await.insert("t1".into(), PauseGate::new()); + // Not paused → passes straight through. + mgr.checkpoint("t1", 128).await.unwrap(); + + mgr.pauses.lock().await.get("t1").unwrap().set(true); + let m2 = Arc::clone(&mgr); + let handle = tokio::spawn(async move { m2.checkpoint("t1", 128).await }); + // The parked task cannot finish before the gate opens (Ok needs an + // open gate, Err needs the park loop to break) — so a finished handle + // here is impossible regardless of scheduling. + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!handle.is_finished()); + mgr.pauses.lock().await.get("t1").unwrap().set(false); + let err = handle.await.unwrap().unwrap_err(); + assert!(err.downcast_ref::().is_some()); + } + + // ---------- FIFO admission (Phase 1) ---------- + + #[tokio::test] + async fn fifo_skips_paused_and_pause_all_blocks_everyone() { + let mgr = TransferManager::new(); + { + let mut w = mgr.waiting.lock().await; + w.push_back("a".to_string()); + w.push_back("b".to_string()); + w.push_back("c".to_string()); + } + mgr.pauses.lock().await.insert("a".into(), PauseGate::new()); + mgr.pauses.lock().await.insert("b".into(), PauseGate::new()); + { + let w = mgr.waiting.lock().await; + assert!(mgr.is_my_turn(&w, "a").await); + assert!(!mgr.is_my_turn(&w, "b").await); + } + // A paused front row never head-of-line blocks the queue. + mgr.pauses.lock().await.get("a").unwrap().set(true); + { + let w = mgr.waiting.lock().await; + assert!(!mgr.is_my_turn(&w, "a").await); + assert!(mgr.is_my_turn(&w, "b").await); + assert!(!mgr.is_my_turn(&w, "c").await); + } + // Pause-all blocks everyone, runnable or not. + mgr.pause_all.set(true); + { + let w = mgr.waiting.lock().await; + assert!(!mgr.is_my_turn(&w, "b").await); + } + } + + #[tokio::test(start_paused = true)] + async fn concurrency_grows_and_shrinks() { + let mgr = TransferManager::new(); + assert_eq!(mgr.semaphore.available_permits(), DEFAULT_CONCURRENCY); + mgr.set_concurrency(5); + assert_eq!(mgr.semaphore.available_permits(), 5); + mgr.set_concurrency(2); + for _ in 0..10 { + tokio::task::yield_now().await; + } + assert_eq!(mgr.semaphore.available_permits(), 2); + } + + // ---------- Retry classification (Phase 3) ---------- + + #[test] + fn transient_errors_retry_permanent_ones_dont() { + assert!(is_transient(&anyhow::anyhow!("connection reset by peer"))); + assert!(is_transient(&anyhow::anyhow!("operation timed out"))); + assert!(!is_transient(&anyhow::anyhow!("authentication failed"))); + assert!(!is_transient(&anyhow::anyhow!( + "Permission denied (os error 13)" + ))); + assert!(!is_transient(&anyhow::anyhow!("No such file or directory"))); + } +} diff --git a/src-tauri/src/virtualfs/mod.rs b/src-tauri/src/virtualfs/mod.rs index b423940..0cc8b8e 100644 --- a/src-tauri/src/virtualfs/mod.rs +++ b/src-tauri/src/virtualfs/mod.rs @@ -357,7 +357,7 @@ impl Hydrator for SessionHydrator { anyhow::bail!("hydration download failed: {}", t.error.unwrap_or_default()) } TransferStatus::Canceled => anyhow::bail!("hydration download canceled"), - TransferStatus::Queued | TransferStatus::Transferring => { + TransferStatus::Queued | TransferStatus::Transferring | TransferStatus::Paused => { tokio::time::sleep(std::time::Duration::from_millis(120)).await; } }, diff --git a/src/components/Settings.tsx b/src/components/Settings.tsx index 0a8db69..b61b2a9 100644 --- a/src/components/Settings.tsx +++ b/src/components/Settings.tsx @@ -149,6 +149,29 @@ export function Settings({ onClose }: Props) { checked={s.autoOpenTransferPanel} onChange={s.setAutoOpenTransferPanel} /> + + + + + + { let unsub: (() => void) | undefined; @@ -23,8 +51,9 @@ export function TransferQueue() { if (!panelOpen) return null; const active = transfers.filter( - (t) => t.status === "transferring" || t.status === "queued" + (t) => t.status === "transferring" || t.status === "paused" ).length; + const queued = queue.length; return (
@@ -32,12 +61,20 @@ export function TransferQueue() { Transfers - {active > 0 && ( + {(active > 0 || queued > 0) && ( - {active} active + {active} active · {queued} queued )}
+ +
); } -function Row({ t, onCancel }: { t: Transfer; onCancel: () => void }) { +/** Global bandwidth cap (KiB/s, 0 = unlimited). Commits on blur or Enter. */ +function ThrottleInput({ + value, + onCommit, +}: { + value: number; + onCommit: (kbps: number) => void; +}) { + const [draft, setDraft] = useState(String(value)); + + // Follow external changes (settings sync / other windows) while not editing. + useEffect(() => { + setDraft(String(value)); + }, [value]); + + const commit = () => { + const n = Math.max(0, parseInt(draft) || 0); + setDraft(String(n)); + if (n !== value) onCommit(n); + }; + + return ( + + setDraft(e.target.value)} + onBlur={commit} + onKeyDown={(e) => { + if (e.key === "Enter") (e.target as HTMLInputElement).blur(); + }} + className="w-14 rounded border border-border bg-bg-panel px-1.5 py-0.5 text-[11px] text-text outline-none focus:border-accent" + /> + KiB/s + + ); +} + +function Row({ + t, + queue, + onCancel, + onPause, + onResume, + onRetry, + onMove, +}: { + t: Transfer; + queue: string[]; + onCancel: () => void; + onPause: () => void; + onResume: () => void; + onRetry: () => void; + onMove: (dir: "up" | "down") => void; +}) { const Icon = t.kind === "download" ? ArrowDownToLine : ArrowUpFromLine; const pct = t.size > 0 ? Math.min(100, (t.transferred / t.size) * 100) : 0; + // Mid-auto-retry: the error text reads "retrying in Ns (attempt N/3)". + const retrying = t.retryAttempt !== undefined && !!t.error; + const queuePos = queue.indexOf(t.id); + + const statusLabel = + t.status === "transferring" + ? `${pct.toFixed(0)}%` + : t.status === "done" + ? "done" + : t.status === "error" + ? "error" + : t.status === "canceled" + ? "canceled" + : t.status === "paused" + ? "Paused" + : t.status === "queued" && queuePos >= 0 + ? `#${queuePos + 1} in queue` + : t.status; return (
@@ -99,24 +218,70 @@ function Row({ t, onCancel }: { t: Transfer; onCancel: () => void }) {
)} {t.error && ( -
{t.error}
+
+ {t.error} +
)}
{fmtSize(t.transferred)}
-
- {t.status === "transferring" - ? `${pct.toFixed(0)}%` - : t.status === "done" - ? "done" - : t.status === "error" - ? "error" - : t.status === "canceled" - ? "canceled" - : t.status} -
+
{statusLabel}
+ {t.status === "queued" && queuePos >= 0 && ( + <> + + + + )} {(t.status === "transferring" || t.status === "queued") && ( + + )} + {t.status === "paused" && ( + + )} + {(t.status === "error" || t.status === "canceled") && ( + + )} + {(t.status === "transferring" || + t.status === "queued" || + t.status === "paused") && (