From 9ea9e9c6841e90180d830d02c3d2cd558014fdc8 Mon Sep 17 00:00:00 2001 From: pmaxhogan Date: Mon, 17 Aug 2026 13:58:02 -0500 Subject: [PATCH 1/2] fix(app): never freeze on tray quit during a backup Quit ran shutdown_orchestrators() inline via tauri::async_runtime::block_on straight from the RunEvent::ExitRequested callback - i.e. on the thread that owns the platform event loop. On Windows that stalls the Win32 message pump for the whole drain budget (up to RUN_LOOP_DRAIN_TIMEOUT = 20s while a backup cycle finishes), so the shell ghosts the window at ~5s, paints "Not Responding", and Windows Error Reporting kills the process at ~14s. The freeze did not even buy the graceful drain it was blocking for: the process is killed mid-drain, so the run loop is hard-killed anyway. - #299: hide the window + repaint the tray synchronously (single fast platform calls), then SPAWN the drain on the Tauri async runtime. The ExitRequested handler becomes a three-phase machine (IDLE -> DRAINING -> READY) so the event loop keeps pumping until the drain re-raises the exit itself. - #300: a distinct slate "quitting" tray icon with a stop glyph, the tooltip "Driven - quitting (finishing backup, right-click to force quit)", and a two-item menu: a disabled "Quitting - finishing current backup..." status line plus "Force quit now". apply_state and the macOS menu bar title engine both stand down while quitting so a still-draining run loop cannot repaint over it. - #301: reconcile ran one Drive round trip per pending op strictly serially (measured 65.2s for 18 ops on maxbook) while emitting no state transition at all, so the UI sat on a generic "Starting backup...". The adopt-or-requeue LOOKUPS now run with bounded concurrency (6) - they are pure reads plus the already single-flighted parent walk, and every state-DB commit still happens one at a time in the original order - and the pass reports honest per-op progress through OrchestratorState::Recovering (new ops_done/ops_total), surfaced as "Recovering - resuming 7 of 18 uploads". Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_019xKUm9vH4ifb5LHR5szy1v --- README.md | 12 +- crates/driven-core/src/executor.rs | 611 +++++++++++++++++-- crates/driven-core/src/orchestrator.rs | 25 +- crates/driven-core/src/types.rs | 31 +- src-tauri/locales/en-US.yml | 3 + src-tauri/src/lib.rs | 383 ++++++++---- src-tauri/src/menubar.rs | 20 + src-tauri/src/tray.rs | 165 ++++- ui/src/__tests__/global-progress-bar.test.ts | 21 +- ui/src/__tests__/progress-store.test.ts | 38 +- ui/src/components/GlobalProgressBar.vue | 18 +- ui/src/locales/en-US.json | 3 +- ui/src/stores/progress.ts | 17 + 13 files changed, 1159 insertions(+), 188 deletions(-) diff --git a/README.md b/README.md index 078b9df0..7e0a8d80 100644 --- a/README.md +++ b/README.md @@ -130,8 +130,12 @@ These move: check each project's current docs before relying on a cell. rules are later edited. Genuinely overlapping sources are still rejected. - Concurrent, paced executor with retries and resumable uploads. An upload interrupted by a quit or crash resumes byte-for-byte across restarts, with - the recovery shown live in the app ("Recovering interrupted upload - 8.2 GB - of 88.6 GB") instead of an unlabeled startup phase. + the recovery shown live in the app - both the per-file byte progress + ("Recovering interrupted upload - 8.2 GB of 88.6 GB") and, while the recovery + works through the interrupted uploads themselves, the op counts + ("Recovering - resuming 7 of 18 uploads") - instead of an unlabeled startup + phase. The startup recovery runs its remote lookups concurrently, so a large + backlog of interrupted uploads no longer delays the first scan by a minute. - Configurable OS priority (`low` by default) for the scan, upload reads, and bundle builds, so backups yield CPU and disk to whatever is in the foreground. - Battery and network awareness: backups defer on battery and on metered or @@ -143,6 +147,10 @@ These move: check each project's current docs before relying on a cell. - macOS menu bar live status: configurable metrics next to the tray icon while backing up (upload speed, percent, files, time remaining) and a configurable idle readout (last backup age or today's uploaded total). +- Quit never freezes the app. Quitting mid-backup hides the window instantly and + lets the current cycle finish in the background; the tray shows a distinct + "quitting" icon, and its menu offers "Force quit now" if you would rather not + wait (in-flight uploads resume from where they stopped on the next launch). - Settings organized as a sidebar of focused pages (General, Schedule & Power, Performance, a macOS/Windows platform page, Network, Privacy & Data, Advanced) with search; About is identity-only, reached from the sidebar diff --git a/crates/driven-core/src/executor.rs b/crates/driven-core/src/executor.rs index c4fc84ee..7e0a372d 100644 --- a/crates/driven-core/src/executor.rs +++ b/crates/driven-core/src/executor.rs @@ -550,20 +550,111 @@ pub fn noop_outcome_sink(_outcome: &OpOutcome) -> futures::future::BoxFuture<'st Box::pin(async {}) } -/// One progress tick of a reconcile-phase upload recovery (the streaming -/// resume of an interrupted resumable session, 2026-08-14 follow-up). Emitted -/// at resume start (with `bytes_done` = the previously-acked offset - a large -/// prefix re-read can run for minutes before the first new ack) and then on -/// every destination ack; the orchestrator throttles + rebroadcasts it as +/// One progress tick of the reconcile-phase recovery pass (2026-08-14 +/// follow-up; issue #301). Two flavours, distinguished by which pair of +/// counters moves: +/// +/// - a BYTE tick, emitted at a resumable session's resume start (with +/// `bytes_done` = the previously-acked offset - a large prefix re-read can +/// run for minutes before the first new ack) and then on every destination +/// ack; +/// - an OP tick (issue #301), emitted as each pending op's remote lookup +/// completes. Most of a reconcile's wall clock is exactly these lookups - one +/// `metadata` read or `find_by_op_uuid` per op, ~2-4s each against Drive - +/// and they move no bytes, so without this the UI sees a generic +/// "Starting backup..." for the whole pass (a measured 65s on a 18-op source). +/// +/// The orchestrator throttles + rebroadcasts either as /// [`OrchestratorState::Recovering`](crate::types::OrchestratorState). #[derive(Debug, Clone, PartialEq, Eq)] pub struct RecoverProgress { - /// Source-relative path of the file being resumed (display only). + /// Source-relative path of the file being resumed (display only). Empty on + /// an op tick, which describes the whole pass rather than one file. pub path: String, - /// Bytes the destination has acked so far. + /// Bytes the destination has acked so far. Zero on an op tick. pub bytes_done: u64, - /// The session's total byte count. + /// The session's total byte count. Zero on an op tick. pub bytes_total: u64, + /// Issue #301: pending ops of this pass recovered so far. + pub ops_done: u64, + /// Issue #301: total pending ops in this pass. + pub ops_total: u64, +} + +impl RecoverProgress { + /// A BYTE tick: one resumable session's re-read progress. The op counters + /// are left at zero - `reconcile_inner` stamps the pass's live values on + /// before forwarding, so a deep call site never has to thread them. + #[must_use] + pub fn bytes(path: String, bytes_done: u64, bytes_total: u64) -> Self { + Self { + path, + bytes_done, + bytes_total, + ops_done: 0, + ops_total: 0, + } + } + + /// An OP tick (issue #301): whole-pass op progress, no byte dimension and + /// no single file to name. Both counters are stamped by the same wrapper as + /// the byte ticks, so this is the zero value it starts from. + #[must_use] + pub fn op_tick() -> Self { + Self { + path: String::new(), + bytes_done: 0, + bytes_total: 0, + ops_done: 0, + ops_total: 0, + } + } +} + +/// How many reconcile adopt-or-requeue lookups may be in flight at once +/// (issue #301). +/// +/// Modest on purpose. These are metadata round trips, not transfers, and they +/// already pass through [`AimdPacer::permit_request`](crate::pacer::AimdPacer) +/// so a rate-limit or an open circuit breaker still gates them - the bound just +/// keeps a large pending set from queueing hundreds of permits at once. Six +/// turns the measured 18-op / 65s serial pass into roughly a sixth of the round +/// trips' wall clock while staying well inside Drive's per-user concurrency. +const RECONCILE_LOOKUP_CONCURRENCY: usize = 6; + +/// The result of ONE prefetched adopt-or-requeue lookup (issue #301). +/// +/// Each variant carries exactly what the sequential decision pass would have +/// gotten from making the call itself, so the decision logic is unchanged - it +/// just reads a value instead of awaiting one. +enum ReconcileLookup { + /// UPDATE path: the `remote.metadata(drive_file_id)` result. + Metadata(anyhow::Result), + /// CREATE path: re-deriving the parent folder chain failed. Propagated + /// verbatim (it is not classified - the sequential path mapped it straight + /// through `to_reconcile_err`). + ParentFailed(anyhow::Error), + /// CREATE path: the `find_by_op_uuid` result under the resolved parent. + Orphan(anyhow::Result>), +} + +impl ReconcileLookup { + /// Did this lookup fail in a way that makes the sequential pass ABORT the + /// source's reconcile and retry next cycle? Used to stop issuing further + /// prefetches, so a Drive outage costs one failed request rather than one + /// per pending op. + /// + /// A `ParentFailed` counts: the sequential pass propagates it unclassified, + /// which aborts the pass just the same. + fn is_retryable_failure(&self) -> bool { + match self { + ReconcileLookup::Metadata(Err(e)) | ReconcileLookup::Orphan(Err(e)) => { + reconcile_metadata_error_is_retryable(classify_drive_error(e)) + } + ReconcileLookup::ParentFailed(_) => true, + ReconcileLookup::Metadata(Ok(_)) | ReconcileLookup::Orphan(Ok(_)) => false, + } + } } /// The reconcile-phase recovery progress sink (mirrors [`OutcomeSink`]'s @@ -4934,6 +5025,170 @@ impl DefaultExecutor { } } + /// One paced UPDATE-path lookup: read the recorded object's metadata and + /// account the response with the pacer/breaker, exactly as the sequential + /// path did inline. A pure READ - safe to run concurrently with its peers + /// (issue #301). + async fn reconcile_metadata_lookup(&self, file_id: &str) -> anyhow::Result { + self.pacer.permit_request().await; + match self.remote.metadata(file_id).await { + Ok(entry) => { + self.pacer.note_response(ResponseClass::Ok); + Ok(entry) + } + Err(e) => { + self.pacer + .note_response(classify_drive_error(&e).response_class()); + Err(e) + } + } + } + + /// One paced CREATE-path lookup: re-derive the parent folder chain, then + /// search it for the orphan carrying `uuid`. + /// + /// Safe to run concurrently (issue #301). The parent walk + /// ([`Self::ensure_parents_once`]) is idempotent AND single-flighted behind + /// `parent_walk` with a per-source `parent_dirs` cache - it was built for + /// exactly this, so that concurrent uploads into one new directory cannot + /// race duplicate `ensure_folder` calls. `find_by_op_uuid` is a pure read. + /// Neither touches the state DB, so no per-op durability ordering is at + /// stake here; every commit still happens in the sequential pass below. + async fn reconcile_orphan_lookup( + &self, + source: &SourceRow, + relative_path: &RelativePath, + uuid: &str, + crypto: Option<&dyn SourceCryptoSuite>, + ) -> ReconcileLookup { + let parent_id = match self + .reconcile_parent_id(source, relative_path, crypto) + .await + { + Ok(id) => id, + Err(e) => return ReconcileLookup::ParentFailed(e), + }; + self.pacer.permit_request().await; + let found = match self + .remote + .find_by_op_uuid(&parent_id, uuid, &source.drive_context()) + .await + { + Ok(found) => { + self.pacer.note_response(ResponseClass::Ok); + Ok(found) + } + Err(e) => { + self.pacer + .note_response(classify_drive_error(&e).response_class()); + Err(e) + } + }; + ReconcileLookup::Orphan(found) + } + + /// Issue #301: run every op's adopt-or-requeue lookup with BOUNDED + /// concurrency, ahead of the sequential decision pass. + /// + /// # Why + /// + /// The lookups are where a reconcile spends its wall clock: one Drive + /// round trip per op at ~2-4s apiece. Serially that is a minute of dead + /// air before the first scan (measured on maxbook 2026-08-16: 18 pending + /// ops, 65.2s between "recovering pending ops" and the first `Scanning` + /// transition). They are also the only part of the pass that is provably + /// independent - see [`Self::reconcile_orphan_lookup`] for why the parent + /// walk is concurrency-safe, and note that NOTHING here writes to the state + /// DB. Every `delete_pending_op` / `adopt_reconciled` / `clear_file_state_*` + /// still runs one at a time, in the original op order, in the pass below, + /// so per-op durability is byte-for-byte unchanged. + /// + /// # What is deliberately excluded + /// + /// - Ops carrying a live `resumable` session. Those stream a whole file and + /// own the byte-progress ticks; they stay strictly sequential, and their + /// fall-through lookup (when a session turns out to be stale) is done + /// inline below. + /// - Ops with no `client_op_uuid` (older rows the normal queue picks up). + /// + /// # Fail-fast + /// + /// The sequential pass ABORTS the source's reconcile on the first RETRYABLE + /// lookup error (keeping the op for the next cycle). Prefetching would + /// otherwise turn one failed request during a Drive outage into N failed + /// requests. So the first future to see a retryable error sets `halt`, and + /// every future that has not started yet returns `None` - the pass below + /// then does that one lookup inline and hits the same abort. Bounded by + /// [`RECONCILE_LOOKUP_CONCURRENCY`]. + async fn prefetch_reconcile_lookups( + &self, + source: &SourceRow, + ops: &[crate::state::PendingOpRow], + crypto: Option<&dyn SourceCryptoSuite>, + ) -> Vec> { + use futures::StreamExt; + + let halt = std::sync::atomic::AtomicBool::new(false); + let halt = &halt; + + // Owned per-op inputs: the futures below must not borrow from `ops`, or + // the closure fails the higher-ranked lifetime bound `buffer_unordered` + // needs. + let prefetchable: Vec<(usize, RelativePath, String, Option)> = ops + .iter() + .enumerate() + .filter_map(|(idx, op)| { + let payload = PendingOpPayload::from_value(&op.payload_json); + // A live session is resumed sequentially; an op with no uuid is + // skipped entirely by the pass below. + if payload.resumable.is_some() { + return None; + } + payload.client_op_uuid.map(|uuid| { + ( + idx, + op.relative_path.clone(), + uuid, + payload.drive_file_id.clone(), + ) + }) + }) + .collect(); + + let mut out: Vec> = (0..ops.len()).map(|_| None).collect(); + if prefetchable.is_empty() { + return out; + } + + let results = futures::stream::iter(prefetchable) + .map(|(idx, relative_path, uuid, drive_file_id)| async move { + if halt.load(std::sync::atomic::Ordering::Acquire) { + return (idx, None); + } + let lookup = match drive_file_id.as_deref() { + Some(file_id) => { + ReconcileLookup::Metadata(self.reconcile_metadata_lookup(file_id).await) + } + None => { + self.reconcile_orphan_lookup(source, &relative_path, &uuid, crypto) + .await + } + }; + if lookup.is_retryable_failure() { + halt.store(true, std::sync::atomic::Ordering::Release); + } + (idx, Some(lookup)) + }) + .buffer_unordered(RECONCILE_LOOKUP_CONCURRENCY) + .collect::>() + .await; + + for (idx, lookup) in results { + out[idx] = lookup; + } + out + } + /// Body of [`Executor::reconcile`], split out so the trait impl stays a /// one-line fetch + delegate. async fn reconcile_inner( @@ -5153,11 +5408,47 @@ impl DefaultExecutor { // when actually needed. let mut exclusion_matcher: Option> = None; - for op in remaining { + // --- issue #301: honest, granular progress for the whole pass ------- + // The pass is dominated by one remote round trip per op, which moves no + // bytes - so the byte ticks alone left the UI on a generic + // "Starting backup..." for the entire recovery. Count ops here and stamp + // the running total onto EVERY tick (byte ticks included), so + // `OrchestratorState::Recovering` always carries both dimensions and the + // deep resume call sites do not have to thread the counters down. + let ops_total = remaining.len() as u64; + let ops_done = std::sync::atomic::AtomicU64::new(0); + let ops_done = &ops_done; + let stamped = |mut p: RecoverProgress| -> futures::future::BoxFuture<'_, ()> { + Box::pin(async move { + p.ops_done = ops_done.load(std::sync::atomic::Ordering::Relaxed); + p.ops_total = ops_total; + on_recover(p).await; + }) + }; + + // Announce the pass BEFORE the first round trip, so the UI leaves the + // indeterminate PowerCheck sweep immediately rather than after however + // long the first lookup takes. + if ops_total > 0 { + stamped(RecoverProgress::op_tick()).await; + } + + // --- issue #301: prefetch the independent lookups concurrently ------- + let mut lookups = self + .prefetch_reconcile_lookups(source, &remaining, crypto) + .await; + + for (idx, op) in remaining.into_iter().enumerate() { let payload = PendingOpPayload::from_value(&op.payload_json); + // Take this op's prefetched lookup (if any). `None` means it was + // deliberately skipped (a live resumable session / no uuid) or the + // prefetch halted early on a retryable failure - either way the + // branches below fall back to making the call inline. + let prefetched = lookups.get_mut(idx).and_then(Option::take); let Some(uuid) = payload.client_op_uuid.clone() else { // No UUID carried (older row): leave it for the normal queue. + ops_done.fetch_add(1, std::sync::atomic::Ordering::Relaxed); continue; }; @@ -5203,7 +5494,7 @@ impl DefaultExecutor { // error to ReconcileError::AuthInvalidGrant so reconcile_once's // enter_needs_reauth fires. let resumed = self - .resume_persisted(source, &op, &payload, resumable, crypto, on_recover) + .resume_persisted(source, &op, &payload, resumable, crypto, &stamped) .await .map_err(to_reconcile_err)?; match resumed { @@ -5223,6 +5514,8 @@ impl DefaultExecutor { self.adopt_reconciled(source, &op, &adopted, entry, crypto) .await .map_err(to_reconcile_err)?; + ops_done.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + stamped(RecoverProgress::op_tick()).await; continue; } None => { @@ -5236,8 +5529,18 @@ impl DefaultExecutor { if let Some(file_id) = payload.drive_file_id.clone() { // Update path: compare the existing object's appProperties. - self.pacer.permit_request().await; - match self.remote.metadata(&file_id).await { + // Issue #301: the round trip normally already happened, in + // parallel with its peers; the pacer/breaker accounting moved + // into `reconcile_metadata_lookup` with it, so the arms below no + // longer call `note_response` themselves. + let metadata = match prefetched { + Some(ReconcileLookup::Metadata(result)) => result, + // Prefetch halted (a peer hit a retryable failure) or this + // op fell through from a stale resumable session: make the + // call inline, exactly as the pre-#301 path did. + _ => self.reconcile_metadata_lookup(&file_id).await, + }; + match metadata { // R2-P1-1: a SUCCESSFUL metadata read decides the op's fate. Ok(entry) if entry @@ -5246,7 +5549,6 @@ impl DefaultExecutor { .map(|v| v == &uuid) .unwrap_or(false) => { - self.pacer.note_response(ResponseClass::Ok); // Already committed remotely; re-hash + finish. // adopt_reconciled re-derives the encrypted parent chain // (remote ensure_folder) - map an invalid_grant there. @@ -5260,7 +5562,6 @@ impl DefaultExecutor { // committed. Only THEN drop the stale op so the next scan // re-enqueues it cleanly (the prior file_state row keeps // the existing drive_file_id for the update). - self.pacer.note_response(ResponseClass::Ok); self.state.delete_pending_op(op.id).await?; } Err(e) => { @@ -5280,7 +5581,6 @@ impl DefaultExecutor { // whole account. Instead clear the stale id so the next scan // re-plans a fresh CREATE (re-upload), and drop this op. let class = classify_drive_error(&e); - self.pacer.note_response(class.response_class()); if reconcile_metadata_error_is_retryable(class) { warn!( target: TARGET, @@ -5313,20 +5613,32 @@ impl DefaultExecutor { // // R2-P1-1: reconcile_parent_id + find_by_op_uuid + adopt each // do remote awaits; map an invalid_grant to needs_reauth. - let parent_id = self - .reconcile_parent_id(source, &op.relative_path, crypto) - .await - .map_err(to_reconcile_err)?; - self.pacer.permit_request().await; - let found = match self - .remote - .find_by_op_uuid(&parent_id, &uuid, &source.drive_context()) - .await - { - Ok(found) => { - self.pacer.note_response(ResponseClass::Ok); - found + // + // Issue #301: the parent walk + lookup normally already ran, in + // parallel with its peers (both are concurrency-safe - see + // `reconcile_orphan_lookup`), and carried the pacer accounting + // with them. Fall back to an inline lookup when the prefetch was + // skipped or halted. + let lookup = match prefetched { + Some( + lookup @ (ReconcileLookup::ParentFailed(_) | ReconcileLookup::Orphan(_)), + ) => lookup, + _ => { + self.reconcile_orphan_lookup(source, &op.relative_path, &uuid, crypto) + .await } + }; + let found = match lookup { + ReconcileLookup::ParentFailed(e) => return Err(to_reconcile_err(e)), + // The update path never reaches here (it is selected by a + // recorded `drive_file_id`), so a Metadata result cannot + // belong to this branch; treat it as "no orphan found", + // which requeues the op - the safe direction. + ReconcileLookup::Metadata(_) => Ok(None), + ReconcileLookup::Orphan(found) => found, + }; + let found = match found { + Ok(found) => found, Err(e) => { // R2-P1-1 / R3-P1-2: a lookup ERROR proves nothing about // whether the create committed. For a TRANSIENT / rate- @@ -5342,7 +5654,6 @@ impl DefaultExecutor { // drive_file_id to clear, so drop the op; the next scan // re-plans a fresh CREATE from the live file. let class = classify_drive_error(&e); - self.pacer.note_response(class.response_class()); if reconcile_metadata_error_is_retryable(class) { warn!( target: TARGET, @@ -5359,6 +5670,8 @@ impl DefaultExecutor { "reconcile: find_by_op_uuid returned a definitive not-found; dropping the op so the next scan re-creates the object (R3-P1-2): {e}" ); self.state.delete_pending_op(op.id).await?; + ops_done.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + stamped(RecoverProgress::op_tick()).await; continue; } }; @@ -5374,6 +5687,13 @@ impl DefaultExecutor { } } } + + // Issue #301: this op is fully handled (adopted, requeued, or + // dropped) and durably committed. Count it and tick, so the UI's + // "Recovering - N of M" advances per op rather than only when a + // resumable session happens to move bytes. + ops_done.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + stamped(RecoverProgress::op_tick()).await; } Ok(()) } @@ -5613,11 +5933,11 @@ impl DefaultExecutor { // Flip the UI into Recovering IMMEDIATELY (2026-08-14 follow-up): a // large prefix re-read produces no destination acks for minutes, and // the state must not stay in the indeterminate PowerCheck sweep. - on_recover(RecoverProgress { - path: op.relative_path.as_str().to_string(), - bytes_done: resumable.acked_offset, - bytes_total: total, - }) + on_recover(RecoverProgress::bytes( + op.relative_path.as_str().to_string(), + resumable.acked_offset, + total, + )) .await; // Re-open the local file and check the resume-safe IDENTITY recorded @@ -5996,11 +6316,11 @@ impl DefaultExecutor { if let Some(io) = self.io_counters.as_ref() { io.add_net_wire(take as u64); } - on_recover(RecoverProgress { - path: op.relative_path.as_str().to_string(), - bytes_done: session.size, - bytes_total: session.size, - }) + on_recover(RecoverProgress::bytes( + op.relative_path.as_str().to_string(), + session.size, + session.size, + )) .await; acc.clear(); Ok(ResumePushOutcome::Completed(entry)) @@ -6030,11 +6350,11 @@ impl DefaultExecutor { } // Persist the new acked offset so a crash resumes from here. self.persist_payload(op.id, live).await?; - on_recover(RecoverProgress { - path: op.relative_path.as_str().to_string(), - bytes_done: received, - bytes_total: session.size, - }) + on_recover(RecoverProgress::bytes( + op.relative_path.as_str().to_string(), + received, + session.size, + )) .await; Ok(ResumePushOutcome::Acked) } @@ -11627,6 +11947,163 @@ mod tests { assert!(h.pacer.backoff_hits.load(Ordering::SeqCst) >= 1); } + // --- issue #301: reconcile speed + honest op progress ------------------- + + /// Issue #301: the adopt-or-requeue lookups must run CONCURRENTLY, and the + /// pass must report per-op progress. + /// + /// Evidence this replaces (maxbook 2026-08-16): 18 pending ops, one Drive + /// round trip apiece, 65.2s between "reconcile: recovering pending ops" and + /// the first `Scanning` transition - with NOT ONE state transition emitted + /// in between, so the UI sat on the generic "Starting backup...". + /// + /// Here 12 ops each cost a fixed `DELAY` on the fake store. Serially that is + /// a hard floor of `12 * DELAY`; with [`RECONCILE_LOOKUP_CONCURRENCY`] = 6 + /// it is two rounds. The bound asserted below sits between the two, so this + /// FAILS if the prefetch is ever removed or silently serialised - it is not + /// a "sleep and hope" timing test. + #[tokio::test] + async fn reconcile_lookups_run_concurrently_and_report_op_progress() { + const OPS: usize = 12; + const DELAY: std::time::Duration = std::time::Duration::from_millis(120); + + let h = harness_with_remote(InMemoryRemoteStore::new().with_slow_responses(DELAY)).await; + + // 12 create-path ops whose objects never landed (no orphan on the fake), + // so each one costs exactly one `find_by_op_uuid` and is then dropped. + for i in 0..OPS { + let rel = RelativePath::try_from(format!("f{i}.txt")).unwrap(); + h.state + .enqueue_pending_op(NewPendingOp { + source_id: h.source.id, + op_type: OP_TYPE_UPLOAD.to_string(), + relative_path: rel, + payload_json: PendingOpPayload { + client_op_uuid: Some(uuid::Uuid::new_v4().to_string()), + ..PendingOpPayload::default() + } + .to_value(), + scheduled_for: 0, + created_at: 0, + }) + .await + .unwrap(); + } + + let ticks: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + let sink = |p: RecoverProgress| -> futures::future::BoxFuture<'_, ()> { + let ticks = &ticks; + Box::pin(async move { + ticks.lock().unwrap_or_else(|e| e.into_inner()).push(p); + }) + }; + + let exec = h.executor(); + let started = std::time::Instant::now(); + exec.reconcile(&h.source, &sink).await.unwrap(); + let elapsed = started.elapsed(); + + // Serial floor is OPS * DELAY = 1440ms; two concurrent rounds is ~240ms. + // The bound is deliberately loose (a slow CI box still passes) but far + // below the serial floor. + let serial_floor = DELAY * OPS as u32; + assert!( + elapsed < serial_floor / 2, + "reconcile lookups must run concurrently: took {elapsed:?}, a serial \ + pass would need at least {serial_floor:?}" + ); + + // Every op was resolved (none landed, so all were dropped). + assert!(h + .state + .get_pending_ops_for_source(h.source.id) + .await + .unwrap() + .is_empty()); + + // Honest progress: the pass announced itself BEFORE the first round trip + // and finished at OPS of OPS. + let ticks = ticks.into_inner().unwrap_or_else(|e| e.into_inner()); + assert!(!ticks.is_empty(), "the pass must report op progress"); + assert_eq!( + ticks[0].ops_done, 0, + "the first tick announces the pass before any lookup" + ); + assert!( + ticks.iter().all(|t| t.ops_total == OPS as u64), + "every tick carries the pass total" + ); + let last = ticks.last().expect("at least one tick"); + assert_eq!( + last.ops_done, OPS as u64, + "the final tick reports every op recovered" + ); + // Monotonic: `ops_done` never goes backwards. + assert!( + ticks.windows(2).all(|w| w[0].ops_done <= w[1].ops_done), + "op progress must be monotonic" + ); + } + + /// A reconcile with NOTHING pending emits no op ticks at all (the common + /// case must stay free - no state transition churn on every clean boot). + #[tokio::test] + async fn reconcile_with_no_pending_ops_emits_no_op_ticks() { + let h = harness().await; + let ticks: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + let sink = |p: RecoverProgress| -> futures::future::BoxFuture<'_, ()> { + let ticks = &ticks; + Box::pin(async move { + ticks.lock().unwrap_or_else(|e| e.into_inner()).push(p); + }) + }; + h.executor().reconcile(&h.source, &sink).await.unwrap(); + assert!( + ticks + .into_inner() + .unwrap_or_else(|e| e.into_inner()) + .is_empty(), + "a clean boot must not emit Recovering ticks" + ); + } + + /// Issue #301 fail-fast: the prefetch stops issuing lookups once one has + /// failed RETRYABLY, so a Drive outage costs a handful of requests rather + /// than one per pending op. Drives the classifier directly - it is what the + /// prefetch's halt flag keys off. + #[test] + fn reconcile_lookup_classifies_which_failures_halt_the_prefetch() { + use driven_drive::google::DriveError as DriveStoreError; + use driven_drive::remote_store::DriveErrorClassification; + + let transient = || { + anyhow::Error::new(DriveStoreError::Classified { + kind: DriveErrorClassification::Transient5xx, + source: anyhow::anyhow!("503"), + }) + }; + assert!( + ReconcileLookup::Metadata(Err(transient())).is_retryable_failure(), + "a transient metadata failure aborts the pass, so it must halt the prefetch" + ); + assert!( + ReconcileLookup::Orphan(Err(transient())).is_retryable_failure(), + "a transient orphan lookup failure must halt the prefetch" + ); + assert!( + ReconcileLookup::ParentFailed(anyhow::anyhow!("folder chain")).is_retryable_failure(), + "a parent-walk failure propagates unclassified and aborts the pass, so it halts too" + ); + // A DEFINITIVE not-found drops the single op and the pass CONTINUES - + // halting on it would needlessly serialise the rest. + let definitive = anyhow::anyhow!("drive.dest_folder_missing"); + assert!( + !ReconcileLookup::Orphan(Err(definitive)).is_retryable_failure(), + "a definitive failure is per-op; it must NOT halt the prefetch" + ); + assert!(!ReconcileLookup::Orphan(Ok(None)).is_retryable_failure()); + } + // --- crash mid-resumable resumes via reconcile -------------------------- #[tokio::test] @@ -12195,6 +12672,12 @@ mod tests { /// follow-up): one at resume start carrying the previously-acked offset /// (the UI must flip out of the indeterminate sweep before the first new /// ack), then per destination ack, ending at bytes_done == bytes_total. + /// + /// Issue #301 added a second flavour of tick - whole-pass OP progress - + /// which now leads the stream (the pass announces itself before the first + /// round trip) and trails it (each recovered op counts). The BYTE ticks + /// asserted here are the ones carrying a path and a non-zero total; every + /// tick, of either flavour, carries the pass's op counters. #[tokio::test] async fn resume_emits_recover_progress_ticks() { let h = harness().await; @@ -12214,24 +12697,44 @@ mod tests { exec.reconcile(&h.source, &sink).await.unwrap(); let ticks = ticks.into_inner().unwrap(); + // Issue #301: the pass announces itself first, with no byte dimension. + assert_eq!( + (ticks[0].ops_done, ticks[0].ops_total, ticks[0].bytes_total), + (0, 1, 0), + "the pass announces 0 of 1 ops before the first round trip: {ticks:?}" + ); + assert!( + ticks.iter().all(|t| t.ops_total == 1), + "every tick carries the pass total: {ticks:?}" + ); + let last = ticks.last().expect("at least one tick"); + assert_eq!( + last.ops_done, 1, + "the final tick reports the op recovered: {ticks:?}" + ); + + let byte_ticks: Vec<&RecoverProgress> = + ticks.iter().filter(|t| t.bytes_total > 0).collect(); assert!( - ticks.len() >= 2, - "at least a start tick and a completion tick: {ticks:?}" + byte_ticks.len() >= 2, + "at least a resume-start tick and a completion tick: {ticks:?}" ); assert_eq!( - ticks[0].bytes_done, 0, + byte_ticks[0].bytes_done, 0, "start tick carries the acked offset" ); - assert_eq!(ticks[0].bytes_total, size); - assert_eq!(ticks[0].path, "ticks.bin"); - let last = ticks.last().unwrap(); + assert_eq!(byte_ticks[0].bytes_total, size); + assert_eq!(byte_ticks[0].path, "ticks.bin"); + let last_byte = byte_ticks.last().expect("a byte tick"); assert_eq!( - last.bytes_done, last.bytes_total, - "the final tick reports completion" + last_byte.bytes_done, last_byte.bytes_total, + "the final byte tick reports completion" ); assert!( - ticks.windows(2).all(|w| w[0].bytes_done <= w[1].bytes_done), - "ticks are monotonic: {ticks:?}" + byte_ticks + .windows(2) + .all(|w| w[0].bytes_done <= w[1].bytes_done), + "byte ticks are monotonic: {ticks:?}" ); // And the op actually resolved (adopted Synced). let row = h diff --git a/crates/driven-core/src/orchestrator.rs b/crates/driven-core/src/orchestrator.rs index fdaf490d..71cb34aa 100644 --- a/crates/driven-core/src/orchestrator.rs +++ b/crates/driven-core/src/orchestrator.rs @@ -1213,8 +1213,19 @@ impl SyncOrchestrator { let now = self.clock.now_ms(); { let mut guard = last_emit.lock().await; + // Issue #301: the pass now emits an OP tick per + // recovered pending op as well as the BYTE ticks of + // a streaming resume, so "is this the final tick" + // has to consider BOTH dimensions - a byte tick that + // finished its file is not final if ops remain, and + // an op tick (which carries 0/0 bytes) is final only + // once every op is done. Anything else throttles to + // ~1/s on the INJECTED clock (the module's + // determinism rule - never tokio/Instant time). + let is_final = + p.bytes_done >= p.bytes_total && p.ops_done >= p.ops_total; let throttled = matches!(*guard, Some(last) if now - last < RECOVER_EMIT_MIN_INTERVAL_MS) - && p.bytes_done < p.bytes_total; + && !is_final; if throttled { return; } @@ -1225,6 +1236,8 @@ impl SyncOrchestrator { path: p.path, bytes_done: p.bytes_done, bytes_total: p.bytes_total, + ops_done: p.ops_done, + ops_total: p.ops_total, }) .await; }) @@ -3465,11 +3478,11 @@ mod tests { if ticks > 1 { let total = 1_000u64; for i in 0..ticks { - on_recover(crate::executor::RecoverProgress { - path: "big/file.bin".to_string(), - bytes_done: i * total / (ticks - 1), - bytes_total: total, - }) + on_recover(crate::executor::RecoverProgress::bytes( + "big/file.bin".to_string(), + i * total / (ticks - 1), + total, + )) .await; } } diff --git a/crates/driven-core/src/types.rs b/crates/driven-core/src/types.rs index 4e623334..e608c7c1 100644 --- a/crates/driven-core/src/types.rs +++ b/crates/driven-core/src/types.rs @@ -480,22 +480,35 @@ pub enum OrchestratorState { /// Checking the power / network gates (DESIGN s5.7) before starting a /// batch. A failed gate transitions to [`OrchestratorState::Paused`]. PowerCheck, - /// The startup reconcile is recovering an interrupted upload by - /// byte-level resume (2026-08-14 follow-up): re-reading the local file - /// and pushing the unacked tail. Previously this ran invisibly inside - /// [`OrchestratorState::PowerCheck`] - for the incident's 88 GB disk - /// image that meant a multi-minute "Starting backup..." sweep while - /// 140 Mbps of upload showed nowhere in the UI. + /// The startup reconcile is recovering interrupted work (2026-08-14 + /// follow-up; issue #301): adopting or re-queuing each pending op, and + /// byte-level resuming any op that carries a live upload session. + /// Previously this ran invisibly inside [`OrchestratorState::PowerCheck`] - + /// for the incident's 88 GB disk image that meant a multi-minute + /// "Starting backup..." sweep while 140 Mbps of upload showed nowhere in + /// the UI, and (issue #301) a 22-op reconcile meant a full minute of + /// generic "Starting backup..." with nothing moving at all. Recovering { /// Source whose pending op is being recovered. source_id: SourceId, - /// Source-relative path of the file being resumed (display only). + /// Source-relative path of the op being recovered (display only). + /// Empty when the tick reports whole-pass op progress rather than one + /// file's byte progress. path: String, /// Bytes the destination has acked so far (the resume's progress - /// numerator; starts at the previously-acked offset, not 0). + /// numerator; starts at the previously-acked offset, not 0). Zero on + /// an op-progress tick, which carries no byte dimension. bytes_done: u64, - /// The session's total byte count. + /// The session's total byte count. Zero on an op-progress tick. bytes_total: u64, + /// Issue #301: pending ops of this source's pass recovered so far. + /// The reconcile spends most of its wall clock on one remote + /// round trip per op (a `metadata` read or a `find_by_op_uuid` + /// lookup), which carries no bytes at all - so without this counter + /// the UI has nothing to show for the majority of a recovery. + ops_done: u64, + /// Issue #301: total pending ops in this source's pass. + ops_total: u64, }, /// Walking + diffing one source's local tree (SPEC s6). `scanned` is a /// running count of files visited, for a live progress readout. diff --git a/src-tauri/locales/en-US.yml b/src-tauri/locales/en-US.yml index b8ee39db..727d3d93 100644 --- a/src-tauri/locales/en-US.yml +++ b/src-tauri/locales/en-US.yml @@ -27,6 +27,8 @@ tray: settings: "Settings" restore: "Restore" quit: "Quit Driven" + quitting_status: "Quitting - finishing current backup..." + force_quit: "Force quit now" menu_status: line1: "Backing up - %{percent}, %{eta} left" line1_no_eta: "Backing up - %{percent}" @@ -58,6 +60,7 @@ tray: needs_reauth: "Driven needs to sign in again" error: "Driven - error, attention needed" suspending: "Suspending..." + quitting: "Driven - quitting (finishing backup, right-click to force quit)" notifications: first_sync_complete: diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8f9738a5..85b6e43c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -63,9 +63,13 @@ mod updater; mod vss_helper; use std::path::PathBuf; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::Arc; +use driven_core::types::AccountId; use tauri::{Manager, RunEvent, WindowEvent}; use tauri_plugin_deep_link::DeepLinkExt; +use tokio::task::JoinHandle; pub use app_state::{AccountHandle, AppState, RemoteMode}; @@ -213,6 +217,109 @@ fn handle_second_launch(app: &tauri::AppHandle, argv: &[String]) { show_main_window(app); } +// ----------------------------------------------------------------------------- +// Quit lifecycle (issue #299 / #300) +// ----------------------------------------------------------------------------- + +/// No quit in progress. +const QUIT_IDLE: u8 = 0; +/// A graceful quit is running its drain OFF the event-loop thread. +const QUIT_DRAINING: u8 = 1; +/// The drain finished (or the user force-quit): the next `ExitRequested` must +/// be allowed through so the process actually exits. +const QUIT_READY: u8 = 2; + +/// The quit lifecycle phase (issue #299). +/// +/// Process-global because the quit path is inherently process-wide and is +/// driven from three places that cannot share a value: the Tauri `RunEvent` +/// callback (`&AppHandle` only), the detached drain task, and the tray menu's +/// "Force quit now" item. +static QUIT_PHASE: AtomicU8 = AtomicU8::new(QUIT_IDLE); + +/// `true` while the graceful quit drain is running. +/// +/// The tray reads this so a `StateChanged` arriving mid-drain (the run loop is +/// still finishing its cycle, so it keeps emitting transitions) cannot repaint +/// over the "quitting" icon/tooltip that [`tray::enter_quitting`] just set. +pub(crate) fn is_quitting() -> bool { + QUIT_PHASE.load(Ordering::Acquire) == QUIT_DRAINING +} + +/// Issue #300: abandon the graceful drain and exit NOW. +/// +/// Wired to the quitting-state tray menu's single enabled item. Every in-flight +/// upload is hard-cancelled by the process exit itself, which is safe because +/// the resume point is already durable: the streaming resumable uploader +/// persists `resumable.acked_offset` into the pending op's payload after EVERY +/// acked wire chunk (`executor.rs` `push_one`), so the next launch's reconcile +/// pass resumes the session from the last acked byte (at most one wire chunk is +/// re-sent). Restore jobs are likewise safe to kill - their writers stage into +/// a temp that is never promoted, so a half-restored file cannot be published. +/// +/// Marks the phase [`QUIT_READY`] BEFORE asking for the exit so the +/// `ExitRequested` this raises is not prevented by the graceful path. +pub(crate) fn force_quit(app: &tauri::AppHandle) { + QUIT_PHASE.store(QUIT_READY, Ordering::Release); + tracing::warn!( + target: "driven::app", + "force quit requested from the tray; abandoning the graceful drain (in-flight uploads resume from their persisted offset on next launch)" + ); + app.exit(0); +} + +/// Everything an explicit Quit must drive to a true stop, TAKEN out of +/// [`AppState`] synchronously on the event-loop thread so the async drain owns +/// it outright and never has to borrow the Tauri-managed state across an await. +struct ShutdownHandles { + /// Per-account task sets (run loop, watcher/event bridges, power poller). + accounts: Vec<(AccountId, Arc)>, + /// M8-P1-1: the already-cancelled in-flight restore job tasks. + restore_jobs: Vec>, + /// M9a: the periodic updater-check task, if it was started. + updater: Option>, + /// M9b: the periodic telemetry-ping task, if it was started. + telemetry: Option>, + /// 2026-08-14 follow-up: the 1 Hz io-throughput sampler, if it was started. + iostat: Option>, +} + +/// Signal every shutdown-able task and TAKE its handle, synchronously. +/// +/// Every call here is a cheap, non-blocking mutex-take or watch-send - there is +/// no `await` and no I/O - so this is safe to run on the event-loop thread. +/// Returns `None` when no [`AppState`] is managed (assembly never ran), i.e. +/// there is nothing to drain. +fn take_shutdown_handles(app: &tauri::AppHandle) -> Option { + let state = app.try_state::()?; + // Signal every orchestrator to stop AFTER its current cycle up front, so the + // concurrent drains below see the stop flag already set (each account's + // in-flight cycle winds down in parallel instead of one-at-a-time). + let accounts = state.accounts(); + for (account_id, handle) in &accounts { + tracing::info!(target: "driven::app", account_id = %account_id, "signalling graceful shutdown on quit"); + handle.orchestrator.shutdown(); + } + Some(ShutdownHandles { + accounts, + // M8-P1-1: cancel every in-flight RESTORE job up front too (mirrors the + // no-orphan AccountHandle drain). Setting each job's cancel flag makes + // its task delete the in-flight temp + emit a terminal CANCELLED status, + // so quit leaves no orphaned restore task and no partial files. + restore_jobs: state.cancel_all_restore_jobs(), + // M9a: signal + take the periodic updater-check task so the drain joins + // it too (no orphan). It is a tokio-interval task that select!s on its + // shutdown watch, so it exits promptly once signalled; the bounded drain + // still aborts-and-awaits it if it is mid-check (e.g. a slow network + // request) so quit cannot hang. + updater: state.shutdown_updater_task(), + // M9b: the periodic telemetry-ping task, same shape as the updater one. + telemetry: state.shutdown_telemetry_task(), + // 2026-08-14 follow-up: the io-throughput sampler, same shape again. + iostat: state.shutdown_iostat_task(), + }) +} + /// GRACEFULLY shut down every per-account task set on an explicit Quit /// (R-P1-1 / R3-P1-1, ROADMAP M5 "no orphaned tokio tasks"; DESIGN s5.10.2 /// in-flight drain). For each account [`AccountHandle::shutdown`] signals the @@ -222,9 +329,6 @@ fn handle_second_launch(app: &tauri::AppHandle, argv: &[String]) { /// giving an in-flight backup cycle a chance to finish rather than being killed /// mid-upload. /// -/// Runs on the Tauri async runtime via `block_on` because the Tauri event-loop -/// callback (`RunEvent`) is synchronous. -/// /// R3-P1-1 (concurrency + no outer cancellation): the per-account drains run /// CONCURRENTLY via [`futures::future::join_all`] - NOT serially - so two slow /// accounts (each run loop up to [`app_state::RUN_LOOP_DRAIN_TIMEOUT`], each @@ -236,117 +340,150 @@ fn handle_second_launch(app: &tauri::AppHandle, argv: &[String]) { /// already self-bounds (await up to its budget, then `abort()` AND await the /// aborted handle), so every per-account drain completes on its own; we let them /// all finish instead of racing an outer cancellation that could orphan a task. -fn shutdown_orchestrators(app: &tauri::AppHandle) { - let Some(state) = app.try_state::() else { +/// +/// Issue #299: this future is SPAWNED on the Tauri async runtime, never +/// `block_on`d from the `RunEvent` callback. See [`begin_graceful_quit`]. +async fn drain_shutdown_handles(handles: ShutdownHandles) { + let ShutdownHandles { + accounts, + restore_jobs, + updater, + telemetry, + iostat, + } = handles; + + // R3-P1-1: drive ALL per-account shutdowns concurrently. Each + // `handle.shutdown()` self-bounds its per-task drains and aborts-and- + // awaits anything that overruns, so no outer timeout is needed (and an + // outer timeout would risk dropping a cancellation-unsafe drain mid-abort + // -> an orphaned task). `join_all` returns only once EVERY account's + // every task is finished. + let drains = accounts.into_iter().map(|(account_id, handle)| async move { + handle.shutdown().await; + tracing::info!(target: "driven::app", account_id = %account_id, "all per-account tasks shut down (no orphans)"); + }); + futures::future::join_all(drains).await; + + // M8-P1-1 / R2-P2-2: drain every cancelled restore task with a BOUNDED, + // abort-capable budget. Each task observes its cancel flag between frames, + // deletes its in-flight temp, and exits - normally well within the budget. + // But a task stuck BEFORE it next checks the flag (e.g. blocked on a slow + // download read) would hang an explicit Quit forever if we awaited it + // unconditionally. So we await each handle up to RESTORE_JOB_DRAIN_TIMEOUT + // and, on timeout, `abort()` it and AWAIT the aborted handle so the task is + // genuinely GONE before quit proceeds (no orphan). The task's temp is + // cleaned even on the abort path because the restore writer holds a + // Drop-based temp guard (see `restore.rs` TempFileGuard), so dropping the + // aborted future removes any in-flight temp. Mirrors the M5 per-account + // `drain_or_abort` shape. The drains run concurrently so two stuck jobs do + // not sum their budgets. + let restore_drains = restore_jobs + .into_iter() + .map(|h| async move { drain_restore_handle(h).await }); + futures::future::join_all(restore_drains).await; + tracing::info!(target: "driven::app", "all in-flight restore jobs cancelled + drained (no orphans)"); + + // M9a: drain the periodic updater-check task with the SAME bounded, + // abort-capable budget so quit never hangs on a mid-check task and leaves + // no orphan. + if let Some(handle) = updater { + drain_restore_handle(handle).await; + tracing::info!(target: "driven::app", "updater periodic check task drained (no orphan)"); + } + + // M9b: drain the periodic telemetry-ping task with the SAME bounded, + // abort-capable budget so quit never hangs on a mid-ping task and leaves no + // orphan. + if let Some(handle) = telemetry { + drain_restore_handle(handle).await; + tracing::info!(target: "driven::app", "telemetry ping task drained (no orphan)"); + } + + // 2026-08-14 follow-up: drain the io-throughput sampler the same way. + if let Some(handle) = iostat { + drain_restore_handle(handle).await; + tracing::info!(target: "driven::app", "io throughput sampler drained (no orphan)"); + } + + // Stop the cosmetic tray syncing-spinner LAST - AFTER every orchestrator + // is dropped (so the per-account event bridges' broadcasts are closed and + // no further `StateChanged` can drive `apply_state` -> restart the + // spinner). Stopping it earlier would race a still-queued syncing event + // that could re-spawn the detached timer task after the stop. It is a + // pure timer loop (set_icon only) that the process exit then tears down; + // stopping it here keeps the no-orphan drain honest. + tray::stop_sync_animation(); + tracing::info!(target: "driven::app", "tray syncing animation stopped (no orphan)"); +} + +/// Issue #299: start an explicit Quit WITHOUT ever blocking the event loop. +/// +/// The old shape ran [`drain_shutdown_handles`] inline via +/// `tauri::async_runtime::block_on` straight from the `RunEvent::ExitRequested` +/// callback - i.e. on the main thread that owns the platform event loop. On +/// Windows that stalls the Win32 message pump for the whole drain budget (up to +/// `RUN_LOOP_DRAIN_TIMEOUT` = 20s while a backup cycle finishes), so the shell +/// ghosts the window at ~5s, paints "Not Responding", and Windows Error +/// Reporting kills the process at ~14s. That is worse than a slow quit: the +/// drain never completes, so the graceful semantics it was blocking FOR are +/// lost too (maxbook 2026-08-16 - two "Application Hang" events for +/// driven-app.exe, and the log has "signalling graceful shutdown on quit" with +/// no matching "all per-account tasks shut down" line). +/// +/// So: hide the window and repaint the tray synchronously (both are single fast +/// platform calls), then SPAWN the drain on the Tauri async runtime and return +/// immediately. The event loop keeps pumping the whole time, so the app can +/// never be declared hung; the caller prevents the exit and the drain re-raises +/// it via [`AppHandle::exit`](tauri::AppHandle::exit) once it is genuinely done. +fn begin_graceful_quit(app: &tauri::AppHandle) { + // Hide the window FIRST: it is the one thing the user is waiting to see, it + // is a single fast platform call, and it must happen even if there is + // nothing to drain. + if let Some(window) = app.get_webview_window(MAIN_WINDOW) { + if let Err(err) = window.hide() { + tracing::warn!(target: "driven::app", %err, "hide main window on quit failed"); + } + } + // Issue #300: the tray becomes the quit progress surface for however long + // the drain runs - distinct icon + tooltip, and a menu whose only action is + // "Force quit now". + tray::enter_quitting(app); + + let Some(handles) = take_shutdown_handles(app) else { // No managed state => no orchestrators ran => no event bridge => the // syncing spinner was never started, so there is nothing to stop. + tracing::info!(target: "driven::app", "quit with no managed state; exiting immediately"); + finish_quit(app); return; }; - // Signal every orchestrator to stop AFTER its current cycle up front, so the - // concurrent drains below see the stop flag already set (each account's - // in-flight cycle winds down in parallel instead of one-at-a-time). - let handles = state.accounts(); - for (account_id, handle) in &handles { - tracing::info!(target: "driven::app", account_id = %account_id, "signalling graceful shutdown on quit"); - handle.orchestrator.shutdown(); - } - // M8-P1-1: cancel every in-flight RESTORE job up front too (mirrors the - // no-orphan AccountHandle drain). Setting each job's cancel flag makes its - // task delete the in-flight temp + emit a terminal CANCELLED status, so quit - // leaves no orphaned restore task and no partial files. We take the handles - // here and await them in the block_on below. - let restore_handles = state.cancel_all_restore_jobs(); - // M9a: signal + take the periodic updater-check task so the drain below joins - // it too (no orphan). It is a tokio-interval task that select!s on its - // shutdown watch, so it exits promptly once signalled; the bounded drain - // below still aborts-and-awaits it if it is mid-check (e.g. a slow network - // request) so quit cannot hang. - let updater_handle = state.shutdown_updater_task(); - // M9b: signal + take the periodic telemetry-ping task so the drain below joins - // it too (no orphan). It is a tokio-interval task that select!s on its shutdown - // watch, so it exits promptly once signalled; the bounded drain below still - // aborts-and-awaits it if it is mid-ping (e.g. a slow best-effort POST) so quit - // cannot hang. - let telemetry_handle = state.shutdown_telemetry_task(); - // 2026-08-14 follow-up: signal + take the io-throughput sampler the same - // way (a 1s tokio-interval task select!ing on its shutdown watch). - let iostat_handle = state.shutdown_iostat_task(); - tauri::async_runtime::block_on(async move { - // R3-P1-1: drive ALL per-account shutdowns concurrently. Each - // `handle.shutdown()` self-bounds its per-task drains and aborts-and- - // awaits anything that overruns, so no outer timeout is needed (and an - // outer timeout would risk dropping a cancellation-unsafe drain mid-abort - // -> an orphaned task). `join_all` returns only once EVERY account's - // every task is finished. - let drains = handles.into_iter().map(|(account_id, handle)| async move { - handle.shutdown().await; - tracing::info!(target: "driven::app", account_id = %account_id, "all per-account tasks shut down (no orphans)"); - }); - futures::future::join_all(drains).await; - - // M8-P1-1 / R2-P2-2: drain every cancelled restore task with a BOUNDED, - // abort-capable budget. Each task observes its cancel flag between frames, - // deletes its in-flight temp, and exits - normally well within the budget. - // But a task stuck BEFORE it next checks the flag (e.g. blocked on a slow - // download read) would hang an explicit Quit forever if we awaited it - // unconditionally. So we await each handle up to RESTORE_JOB_DRAIN_TIMEOUT - // and, on timeout, `abort()` it and AWAIT the aborted handle so the task is - // genuinely GONE before quit proceeds (no orphan). The task's temp is - // cleaned even on the abort path because the restore writer holds a - // Drop-based temp guard (see `restore.rs` TempFileGuard), so dropping the - // aborted future removes any in-flight temp. Mirrors the M5 per-account - // `drain_or_abort` shape. The drains run concurrently so two stuck jobs do - // not sum their budgets. - let restore_drains = restore_handles - .into_iter() - .map(|h| async move { drain_restore_handle(h).await }); - futures::future::join_all(restore_drains).await; - tracing::info!(target: "driven::app", "all in-flight restore jobs cancelled + drained (no orphans)"); - - // M9a: drain the periodic updater-check task with the SAME bounded, - // abort-capable budget so quit never hangs on a mid-check task and leaves - // no orphan. - if let Some(handle) = updater_handle { - drain_restore_handle(handle).await; - tracing::info!(target: "driven::app", "updater periodic check task drained (no orphan)"); - } - // M9b: drain the periodic telemetry-ping task with the SAME bounded, - // abort-capable budget so quit never hangs on a mid-ping task and leaves no - // orphan. - if let Some(handle) = telemetry_handle { - drain_restore_handle(handle).await; - tracing::info!(target: "driven::app", "telemetry ping task drained (no orphan)"); + let app = app.clone(); + tauri::async_runtime::spawn(async move { + drain_shutdown_handles(handles).await; + + if let Some(state) = app.try_state::() { + // Issue #25 (DESIGN s5.3.1): shut the least-privilege VSS helper + // broker down (best-effort; no-op if it was never launched) so no + // elevated process outlives the app. Done AFTER the orchestrator + // drain above so no locked-file backup is mid-stream on the pipe. + state.shutdown_vss_helper(); + // DESIGN s5.3.2: the macOS mirror - shut the APFS snapshot broker + // down so no root process, and no mounted snapshot, outlives the app + // session. Same ordering rationale as the VSS sweep above. + state.shutdown_apfs_helper(); } - // 2026-08-14 follow-up: drain the io-throughput sampler the same way. - if let Some(handle) = iostat_handle { - drain_restore_handle(handle).await; - tracing::info!(target: "driven::app", "io throughput sampler drained (no orphan)"); - } - - // Stop the cosmetic tray syncing-spinner LAST - AFTER every orchestrator - // is dropped (so the per-account event bridges' broadcasts are closed and - // no further `StateChanged` can drive `apply_state` -> restart the - // spinner). Stopping it earlier would race a still-queued syncing event - // that could re-spawn the detached timer task after the stop. It is a - // pure timer loop (set_icon only) that the process exit then tears down; - // stopping it here keeps the no-orphan drain honest. - tray::stop_sync_animation(); - tracing::info!(target: "driven::app", "tray syncing animation stopped (no orphan)"); + finish_quit(&app); }); +} - // Issue #25 (DESIGN s5.3.1): shut the least-privilege VSS helper broker down - // (best-effort; no-op if it was never launched) so no elevated process - // outlives the app. Done AFTER the orchestrator drain above so no locked-file - // backup is mid-stream on the pipe. Sync (a quick pipe Shutdown), so it runs - // on this thread rather than the async runtime. - state.shutdown_vss_helper(); - - // DESIGN s5.3.2: the macOS mirror - shut the APFS snapshot broker down so no - // root process, and no mounted snapshot, outlives the app session. Same - // ordering rationale as the VSS sweep above. - state.shutdown_apfs_helper(); +/// The drain is done (or there was nothing to drain): flip the phase to +/// [`QUIT_READY`] and ask for the exit again. The `RunEvent::ExitRequested` this +/// raises sees `QUIT_READY` and lets the process go. +fn finish_quit(app: &tauri::AppHandle) { + tracing::info!(target: "driven::app", "graceful quit drain complete; exiting"); + QUIT_PHASE.store(QUIT_READY, Ordering::Release); + app.exit(0); } /// R2-P2-2: drive ONE cancelled restore task to a true stop with a bounded budget. @@ -774,19 +911,47 @@ pub fn run() { // // `RunEvent::ExitRequested.code`: // - `Some(_)` => an explicit exit (`app.exit(code)` from the tray Quit / - // `--quit`). Drain + let the process exit. + // `--quit`). Start the drain + keep the loop alive until it finishes. // - `None` => an incidental exit (the last window was closed). Since the // app is a background tray daemon, `prevent_exit()` keeps it alive so // sync survives. (The window-close handler already hid the window, but // this guards the path where the platform still raises ExitRequested.) + // + // Issue #299: the explicit-quit arm is a THREE-phase state machine keyed off + // `QUIT_PHASE`, because the drain now runs off this thread and therefore has + // to come back through this same callback to actually exit: + // IDLE -> prevent the exit, start the drain, return at once (the event + // loop keeps pumping, so the app can never be declared hung); + // DRAINING -> a second explicit quit arrived while the first is still + // draining (a repeat `--quit`, or an OS session-end). Prevent + // the exit again and let the in-flight drain finish - the tray's + // "Force quit now" is the deliberate escape hatch; + // READY -> the drain finished (or the user force-quit): do NOT prevent, + // and the process exits. app.run(|app_handle, event| { if let RunEvent::ExitRequested { code, api, .. } = &event { if code.is_none() { tracing::debug!(target: "driven::app", "incidental exit (last window closed); staying alive in tray"); api.prevent_exit(); - } else { - tracing::info!(target: "driven::app", "explicit quit; draining orchestrators"); - shutdown_orchestrators(app_handle); + return; + } + match QUIT_PHASE.compare_exchange( + QUIT_IDLE, + QUIT_DRAINING, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + tracing::info!(target: "driven::app", "explicit quit; draining orchestrators off the event loop"); + api.prevent_exit(); + begin_graceful_quit(app_handle); + } + Err(QUIT_DRAINING) => { + tracing::info!(target: "driven::app", "quit requested again while already draining; ignoring (use the tray's Force quit now)"); + api.prevent_exit(); + } + // QUIT_READY (and any value we never write): let the exit happen. + Err(_) => {} } } }); diff --git a/src-tauri/src/menubar.rs b/src-tauri/src/menubar.rs index 53990e87..d0baab20 100644 --- a/src-tauri/src/menubar.rs +++ b/src-tauri/src/menubar.rs @@ -1105,6 +1105,26 @@ async fn tick( status_gen_seen: &mut u64, was_active: &mut bool, ) { + // Issue #300: a quit is draining and the tray now belongs to the quitting + // affordance. Keeping a live "12.3 MB/s, 4 min left" title next to the + // quitting icon would contradict it, so clear the title ONCE and stop + // ticking. The `painted` cache makes the clear idempotent across the + // remaining ticks, and the status rows need no handling here - + // `tray::enter_quitting` already dropped their handles, so the block below + // would skip them anyway. + if crate::is_quitting() { + if should_paint(painted, &None) { + if let Some(tray) = app.tray_by_id(tray::TRAY_ID) { + if let Err(err) = tray.set_title(None::<&str>) { + tracing::debug!(target: TARGET, %err, "clear tray title on quit failed"); + } else { + *painted = Some(None); + } + } + } + return; + } + // Copy both statics out and drop the guards BEFORE the idle branch's // `.await` below - a std Mutex guard must never cross a suspend point. let cfg = *CONFIG.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/src-tauri/src/tray.rs b/src-tauri/src/tray.rs index b6d27c37..28f495c5 100644 --- a/src-tauri/src/tray.rs +++ b/src-tauri/src/tray.rs @@ -122,6 +122,12 @@ mod menu_id { pub const ACTIVITY: &str = "activity"; pub const RESTORE: &str = "restore"; pub const QUIT: &str = "quit"; + /// Issue #300: the disabled status line of the QUITTING menu ("Quitting - + /// finishing current backup..."). Disabled, so it never dispatches. + pub const QUITTING_STATUS: &str = "quitting_status"; + /// Issue #300: the only enabled item of the QUITTING menu - abandon the + /// graceful drain and exit now. + pub const FORCE_QUIT: &str = "force_quit"; } /// Tiny generated-tile dimensions. A 16x16 RGBA tile is a valid tray icon on @@ -161,6 +167,20 @@ pub enum TrayIcon { NetworkAttention, /// Red: an error needs attention (auth needed, decrypt failure, disk full). Error, + /// Issue #300: neutral slate with a stop square - an explicit Quit is + /// draining the current backup cycle. Deliberately NOT amber: the approved + /// mockup used amber, but amber is already spent twice over on + /// [`TrayIcon::Paused`] and [`TrayIcon::NetworkAttention`], and "quitting" + /// must not read as "paused, still running". A desaturated slate badge plus + /// the universal stop square says "winding down" by hue AND by shape, and it + /// is the only badged state that is not a saturated colour - so it reads + /// distinctly even on the macOS template path, where hue is discarded + /// entirely and only the punched glyph survives. + /// + /// Never produced by [`TrayIcon::for_state`]: quitting is a shell lifecycle + /// phase, not an [`OrchestratorState`]. It is set directly by + /// [`enter_quitting`]. + Quitting, } impl TrayIcon { @@ -223,6 +243,9 @@ impl TrayIcon { TrayIcon::NetworkAttention => [0xff, 0x8c, 0x00, 0xff], // Red - error needs attention. TrayIcon::Error => [0xdc, 0x26, 0x26, 0xff], + // Slate (zinc-500) - quitting: the one desaturated badge, so it can + // never be mistaken for the amber Paused / NetworkAttention pair. + TrayIcon::Quitting => [0x71, 0x71, 0x7a, 0xff], } } @@ -342,6 +365,7 @@ impl TrayIcon { TrayIcon::Paused => draw_pause_glyph(rgba, width, height, geom), TrayIcon::NetworkAttention => draw_bang_glyph(rgba, width, height, geom), TrayIcon::Error => draw_cross_glyph(rgba, width, height, geom), + TrayIcon::Quitting => draw_stop_glyph(rgba, width, height, geom), } } @@ -698,6 +722,19 @@ fn draw_cross_glyph(rgba: &mut [u8], width: u32, height: u32, geom: &BadgeGeom) } } +/// Draw the QUITTING glyph (issue #300): a white filled square - the universal +/// "stop" mark - centred in the badge disc. Distinct in SHAPE from the pause +/// bars, the `!`, and the `X`, so the quitting state is readable on the macOS +/// template path too (where the slate hue is discarded). +fn draw_stop_glyph(rgba: &mut [u8], width: u32, height: u32, geom: &BadgeGeom) { + let &BadgeGeom { cx, cy, r_fill } = geom; + // Half-extent chosen so the square sits comfortably inside the disc: at + // 0.42 * r_fill its corners land at ~0.59 * r_fill from the centre, well + // short of the rim even after the white contrast ring is drawn. + let half = (r_fill * 0.42).max(1.0); + fill_rect(rgba, width, height, cx, cy, half, half); +} + /// Is this pause reason a network / reachability condition (DESIGN s8.1 /// yellow-with-`!`) rather than a plain user/auto pause (DESIGN s8.1 yellow)? fn pause_reason_is_network(reason: PauseReason) -> bool { @@ -1135,6 +1172,91 @@ fn build_menu(app: &AppHandle) -> tauri::Result> { .build() } +/// Issue #300: the QUITTING tray menu - a disabled status line plus the single +/// enabled "Force quit now" escape hatch, and nothing else. +/// +/// Every normal action is deliberately absent: "Sync now" / "Pause" / "Resume" +/// would all queue work against orchestrators that are already winding down, +/// and a second "Quit" is a no-op the shell already ignores (see the +/// `QUIT_DRAINING` arm of the `RunEvent` handler). The only two things a user +/// can usefully learn or do here are "it is finishing the current backup" and +/// "stop waiting". +fn build_quitting_menu(app: &AppHandle) -> tauri::Result> { + let status = MenuItem::with_id( + app, + menu_id::QUITTING_STATUS, + rust_i18n::t!("tray.quitting_status"), + false, + None::<&str>, + )?; + let force = MenuItem::with_id( + app, + menu_id::FORCE_QUIT, + rust_i18n::t!("tray.force_quit"), + true, + None::<&str>, + )?; + MenuBuilder::new(app) + .item(&status) + .separator() + .item(&force) + .build() +} + +/// Issue #300: flip the whole tray into the QUITTING affordance - distinct +/// icon, distinct tooltip, and the [`build_quitting_menu`] two-item menu. +/// +/// Called from the shell's quit path ([`crate::begin_graceful_quit`]) on the +/// event-loop thread, BEFORE the drain is spawned, so the tray tells the truth +/// from the very first moment of a quit that may take up to +/// [`crate::app_state::RUN_LOOP_DRAIN_TIMEOUT`] to finish. +/// +/// Holds `TRAY_APPLY` for the icon swap for the same reason [`apply_state`] +/// does: a concurrent `StateChanged` must not restart the spinner after we stop +/// it. Beyond this call, [`apply_state`] additionally checks +/// [`crate::is_quitting`] and returns before touching the tray at all, so a +/// transition emitted by a still-draining run loop cannot repaint over us. +pub fn enter_quitting(app: &AppHandle) { + { + let _apply = TRAY_APPLY.lock().unwrap_or_else(|e| e.into_inner()); + // The spinner task owns the icon while it runs - stop it first or its + // next frame would overwrite the quitting icon 125ms later. + stop_sync_animation(); + let Some(tray) = app.tray_by_id(TRAY_ID) else { + tracing::warn!(target: TARGET, "tray {TRAY_ID} not found; cannot show the quitting state"); + return; + }; + if let Err(err) = tray.set_icon(Some(TrayIcon::Quitting.image())) { + tracing::warn!(target: TARGET, "set quitting tray icon failed: {err}"); + } + set_template_mode(&tray); + if let Err(err) = + tray.set_tooltip(Some(rust_i18n::t!("tray.tooltip.quitting").into_owned())) + { + tracing::warn!(target: TARGET, "set quitting tray tooltip failed: {err}"); + } + // Drop the status-row handles BEFORE swapping the menu out, so the macOS + // menu bar engine's 1 Hz tick cannot `set_text` a handle belonging to + // the menu we are about to replace (same discipline as `rebuild`). + *STATUS_ITEMS.lock().unwrap_or_else(|e| e.into_inner()) = None; + match build_quitting_menu(app) { + Ok(menu) => { + if let Err(err) = tray.set_menu(Some(menu)) { + tracing::warn!(target: TARGET, "set quitting tray menu failed: {err}"); + } + } + Err(err) => { + // The normal menu stays up. Its Quit item is now a no-op (the + // shell is already draining), which is survivable - the drain + // still finishes on its own; the user just loses the force-quit + // shortcut. + tracing::warn!(target: TARGET, "build quitting tray menu failed: {err}"); + } + } + } + tracing::info!(target: TARGET, "tray switched to the quitting state (force quit available)"); +} + /// Dispatch a tray menu click to the M5 sync commands / window show / quit /// (SPEC s12). Async commands run on the Tauri runtime so the menu callback /// returns immediately. @@ -1174,7 +1296,12 @@ fn on_menu_event(app: &AppHandle, id: &str) { show_main_window(app); navigate_hint(app, id); } + // Routes into `RunEvent::ExitRequested`, which starts the GRACEFUL quit + // off the event-loop thread (issue #299) and swaps this menu for + // `build_quitting_menu`. menu_id::QUIT => app.exit(0), + // Issue #300: abandon the drain and exit now. + menu_id::FORCE_QUIT => crate::force_quit(app), other => tracing::warn!(target: TARGET, "unknown tray menu id: {other}"), } } @@ -1379,6 +1506,19 @@ pub fn apply_state(app: &AppHandle, account_id: AccountId, state: OrchestratorSt // goes idle (the old last-writer-wins bug). The icon + tooltip reflect // the aggregate; the notification below stays PER ACCOUNT. let aggregate = aggregate_state(account_id, state.clone()); + + // Issue #300: a quit is draining. The run loop is still finishing its + // cycle, so it keeps emitting transitions - but the tray now belongs to + // the quitting affordance ([`enter_quitting`]), and repainting a syncing + // spinner over it would tell the user the quit had not registered. The + // aggregate map above is still updated (it is process state, and a + // force-quit is the only way out of here anyway), but nothing below this + // point runs: no icon, no tooltip, no spinner, and no OS notification - + // a "first sync complete" toast fired while the app is closing is noise. + if crate::is_quitting() { + return; + } + let icon = TrayIcon::for_state(&aggregate); // Drive the animation purely off the aggregate icon: start the spinner @@ -1963,6 +2103,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ] { let buf = icon.rgba_buffer(); // TILE x TILE pixels, 4 bytes (RGBA) each. @@ -1984,6 +2125,9 @@ mod tests { TrayIcon::Paused.rgba(), TrayIcon::NetworkAttention.rgba(), TrayIcon::Error.rgba(), + // Issue #300: the quitting slate must not collide with any of the + // above - especially the two ambers it deliberately avoids. + TrayIcon::Quitting.rgba(), ]; for i in 0..colours.len() { for j in (i + 1)..colours.len() { @@ -2018,6 +2162,12 @@ mod tests { ("tray.activity", "Activity"), ("tray.restore", "Restore"), ("tray.quit", "Quit Driven"), + // Issue #300: the two labels of the QUITTING menu. + ( + "tray.quitting_status", + "Quitting - finishing current backup...", + ), + ("tray.force_quit", "Force quit now"), ]; for (key, expected) in cases { let got = rust_i18n::t!(key); @@ -2059,6 +2209,8 @@ mod tests { "tray.activity", "tray.restore", "tray.quit", + "tray.quitting_status", + "tray.force_quit", "tray.menu_status.line1", "tray.menu_status.line1_no_eta", "tray.menu_status.line2", @@ -2079,6 +2231,7 @@ mod tests { "tray.tooltip.needs_reauth", "tray.tooltip.error", "tray.tooltip.suspending", + "tray.tooltip.quitting", "notifications.first_sync_complete.title", "notifications.first_sync_complete.body", "notifications.error.title", @@ -2157,6 +2310,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ]; let mut variants: Vec<(TrayIcon, Vec)> = Vec::new(); for s in states { @@ -2195,6 +2349,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ] { let Some([r, g, b]) = s.badge_color() else { panic!("non-idle state {s:?} must have a badge colour"); @@ -2222,6 +2377,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ] { let img = s.image(); assert_eq!(img.width(), base.width, "{s:?} width"); @@ -2259,6 +2415,10 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + // Issue #300: the stop square must be distinct from the pause bars, + // the `!`, and the `X` - it is the ONLY state cue that survives the + // macOS template path, where the slate hue is discarded. + TrayIcon::Quitting, ] { // Badge-only baseline (disc + ring, no glyph) vs the full glyphed icon. let mut badge_only = base.rgba.clone(); @@ -2295,6 +2455,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ] { let f0 = s.brand_rgba_frame(base, 0); let f5 = s.brand_rgba_frame(base, 5); @@ -2416,11 +2577,12 @@ mod tests { rgba.chunks_exact(4).map(|px| px[3]).collect() } - const STATES: [TrayIcon; 4] = [ + const STATES: [TrayIcon; 5] = [ TrayIcon::Syncing, TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ]; /// THE guard against a solid black square: the template source must be @@ -2537,6 +2699,7 @@ mod tests { TrayIcon::Paused, TrayIcon::NetworkAttention, TrayIcon::Error, + TrayIcon::Quitting, ] { assert_eq!( s.template_rgba_frame(base, 0), diff --git a/ui/src/__tests__/global-progress-bar.test.ts b/ui/src/__tests__/global-progress-bar.test.ts index 4fbd613f..16494155 100644 --- a/ui/src/__tests__/global-progress-bar.test.ts +++ b/ui/src/__tests__/global-progress-bar.test.ts @@ -41,13 +41,20 @@ function powerCheck(): OrchestratorState { function idle(): OrchestratorState { return { state: "idle", last_run_at: null }; } -function recovering(bytesDone: number, bytesTotal: number): OrchestratorState { +function recovering( + bytesDone: number, + bytesTotal: number, + opsDone = 0, + opsTotal = 0 +): OrchestratorState { return { state: "recovering", source_id: "src-1", path: "dev-drives/dev.vhdx", bytes_done: bytesDone, bytes_total: bytesTotal, + ops_done: opsDone, + ops_total: opsTotal, }; } function executing(p: Partial): OrchestratorState { @@ -247,6 +254,18 @@ describe("GlobalProgressBar", () => { expect(wrapper.find(PHASE_LABEL).text()).toBe("Recovering an interrupted upload..."); }); + it("names the op counts for the byte-free part of the recovery (issue #301)", async () => { + // Most of a reconcile is one remote lookup per interrupted upload - no + // bytes move, so before #301 this rendered as a generic "Starting + // backup..." for a measured 65 seconds. + const { store, wrapper } = mountBar(); + store.ingest(perAccount("a", recovering(0, 0, 7, 18))); + await settle(wrapper); + expect(wrapper.find(PHASE_LABEL).text()).toBe("Recovering - resuming 7 of 18 uploads"); + const bar = wrapper.find('[role="progressbar"]'); + expect(bar.attributes("aria-valuenow")).toBe("39"); + }); + it("shows the upload percent once execution starts", async () => { const { store, wrapper } = mountBar(); store.ingest(perAccount("a", executing({ bytes_done: 1, bytes_total: 4 }))); diff --git a/ui/src/__tests__/progress-store.test.ts b/ui/src/__tests__/progress-store.test.ts index a3c60c0c..ea005236 100644 --- a/ui/src/__tests__/progress-store.test.ts +++ b/ui/src/__tests__/progress-store.test.ts @@ -48,13 +48,20 @@ function verifying(sampled = 0): OrchestratorState { function powerCheck(): OrchestratorState { return { state: "power_check" }; } -function recovering(bytesDone: number, bytesTotal: number): OrchestratorState { +function recovering( + bytesDone: number, + bytesTotal: number, + opsDone = 0, + opsTotal = 0 +): OrchestratorState { return { state: "recovering", source_id: "src-1", path: "dev-drives/dev.vhdx", bytes_done: bytesDone, bytes_total: bytesTotal, + ops_done: opsDone, + ops_total: opsTotal, }; } function backoff(): OrchestratorState { @@ -615,4 +622,33 @@ describe("recovering state", () => { expect(store.active).toBe(true); expect(store.percent).toBeNull(); }); + + // --- issue #301: op progress for the byte-free part of the pass ----------- + + it("uses the OP counters for a determinate percent when no bytes move", () => { + // The bulk of a reconcile is one remote lookup per interrupted upload - + // work with no byte dimension at all. Before #301 that read as an + // indeterminate sweep for a measured 65s. + const store = useProgressStore(); + store.ingest(perAccount("a", recovering(0, 0, 7, 18))); + expect(store.active).toBe(true); + expect(store.phase).toBe("recovering"); + expect(store.recoveringOps).toEqual({ done: 7, total: 18 }); + expect(store.percent).toBeCloseTo(7 / 18, 5); + }); + + it("prefers the byte percent over the op percent while a resume is streaming", () => { + // A byte tick carries BOTH dimensions; bytes are the finer-grained signal + // for the file actually on the wire, so they win. + const store = useProgressStore(); + store.ingest(perAccount("a", recovering(25, 100, 3, 18))); + expect(store.percent).toBeCloseTo(0.25, 5); + }); + + it("sums op counters across accounts", () => { + const store = useProgressStore(); + store.ingest(perAccount("a", recovering(0, 0, 2, 5))); + store.ingest(perAccount("b", recovering(0, 0, 4, 10))); + expect(store.recoveringOps).toEqual({ done: 6, total: 15 }); + }); }); diff --git a/ui/src/components/GlobalProgressBar.vue b/ui/src/components/GlobalProgressBar.vue index 1c3989b3..44588f97 100644 --- a/ui/src/components/GlobalProgressBar.vue +++ b/ui/src/components/GlobalProgressBar.vue @@ -70,10 +70,20 @@ const label = computed(() => { // upload, with its real byte totals ("8.2 GB of 88.6 GB") - this used // to be an unlabelled indeterminate "Starting backup..." sweep. const r = progress.recoveringBytes; - return r.total > 0 - ? t("progress.recoveringBytes", { - done: formatBytes(r.done, locale.value), - total: formatBytes(r.total, locale.value), + if (r.total > 0) { + return t("progress.recoveringBytes", { + done: formatBytes(r.done, locale.value), + total: formatBytes(r.total, locale.value), + }); + } + // Issue #301: the rest of the pass is one remote lookup per interrupted + // upload - no bytes move, so name the OPS instead of falling back to the + // bare "Recovering an interrupted upload..." for a minute. + const ro = progress.recoveringOps; + return ro.total > 0 + ? t("progress.recoveringOps", { + done: count.format(ro.done), + total: count.format(ro.total), }) : t("progress.recovering"); } diff --git a/ui/src/locales/en-US.json b/ui/src/locales/en-US.json index 50d7e612..b70ede58 100644 --- a/ui/src/locales/en-US.json +++ b/ui/src/locales/en-US.json @@ -28,7 +28,8 @@ "verifying": "Verifying backup...", "verifyingCount": "Verifying backup - {count} files", "recovering": "Recovering an interrupted upload...", - "recoveringBytes": "Recovering interrupted upload - {done} of {total}" + "recoveringBytes": "Recovering interrupted upload - {done} of {total}", + "recoveringOps": "Recovering - resuming {done} of {total} uploads" }, "pauseBanner": { "indefinite": "Backups paused indefinitely", diff --git a/ui/src/stores/progress.ts b/ui/src/stores/progress.ts index 6333fb92..3eb6bdf4 100644 --- a/ui/src/stores/progress.ts +++ b/ui/src/stores/progress.ts @@ -219,6 +219,17 @@ export const useProgressStore = defineStore("progress", () => { total: sumOver("recovering", "bytes_total"), })); + /** Issue #301: OP progress of the reconcile-phase recovery, summed across + * accounts in `recovering`. Most of a recovery is one remote round trip per + * pending op - work that moves no bytes at all - so `recoveringBytes` above + * stays 0/0 for it and the UI had nothing to show. These counters are what + * make the phase honest ("Recovering - resuming 7 of 18 uploads") instead of + * a generic "Starting backup..." for a measured 65 seconds. */ + const recoveringOps = computed<{ done: number; total: number }>(() => ({ + done: sumOver("recovering", "ops_done"), + total: sumOver("recovering", "ops_total"), + })); + /** Aggregate execution progress across every account currently `executing`. * Scan/plan/verify carry no reliable total, so they contribute nothing here. * @@ -275,6 +286,11 @@ export const useProgressStore = defineStore("progress", () => { // resume ran). const r = recoveringBytes.value; if (r.total > 0) return clamp01(r.done / r.total); + // Issue #301: no byte dimension (the pass is doing per-op remote lookups), + // but the op counters are a real, determinate total - use them rather than + // falling back to an indeterminate sweep. + const ro = recoveringOps.value; + if (ro.total > 0) return clamp01(ro.done / ro.total); return null; }); @@ -342,6 +358,7 @@ export const useProgressStore = defineStore("progress", () => { plannedFiles, verified, recoveringBytes, + recoveringOps, percent, filesDone, filesTotal, From be55fe9effe7cc8fb5db89f3af85b66fc3f7aa90 Mon Sep 17 00:00:00 2001 From: pmaxhogan Date: Mon, 17 Aug 2026 14:26:25 -0500 Subject: [PATCH 2/2] test(core): cover the new reconcile prefetch branches The coverage gate flagged a 0.13pp Rust regression: the #301 prefetch added error branches with no test behind them. - ReconcileLookup drops the ParentFailed variant for a nested Result on Orphan (outer = the parent walk, inner = find_by_op_uuid). That removes the defensive "a Metadata result reached the create branch" arm entirely - the branch is selected by the absence of a recorded drive_file_id, so the not-prefetched fallback covers it. - reconcile_parent_walk_failure_aborts_the_pass_and_keeps_every_op: 8 nested create ops against a store whose ensure_folder always fails. Covers the parent-walk error propagation, the halt classifier's Orphan(Err) arm, and (with more ops than the concurrency bound) the prefetch's halt branch. - reconcile_adopts_an_update_whose_object_already_carries_the_op_uuid: the UPDATE path's successful metadata lookup, which had no test at all. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_019xKUm9vH4ifb5LHR5szy1v --- crates/driven-core/src/executor.rs | 249 +++++++++++++++++++++++------ 1 file changed, 197 insertions(+), 52 deletions(-) diff --git a/crates/driven-core/src/executor.rs b/crates/driven-core/src/executor.rs index 7e0a372d..3ad8df11 100644 --- a/crates/driven-core/src/executor.rs +++ b/crates/driven-core/src/executor.rs @@ -630,12 +630,13 @@ const RECONCILE_LOOKUP_CONCURRENCY: usize = 6; enum ReconcileLookup { /// UPDATE path: the `remote.metadata(drive_file_id)` result. Metadata(anyhow::Result), - /// CREATE path: re-deriving the parent folder chain failed. Propagated - /// verbatim (it is not classified - the sequential path mapped it straight - /// through `to_reconcile_err`). - ParentFailed(anyhow::Error), - /// CREATE path: the `find_by_op_uuid` result under the resolved parent. - Orphan(anyhow::Result>), + /// CREATE path, in two layers because the sequential pass treats the two + /// failures differently. The OUTER `Err` is a failed parent-folder-chain + /// walk, propagated verbatim (unclassified, straight through + /// `to_reconcile_err`, aborting the source's pass); the INNER result is + /// `find_by_op_uuid`'s, which IS classified (transient keeps the op and + /// aborts, definitive drops just this op). + Orphan(anyhow::Result>>), } impl ReconcileLookup { @@ -648,11 +649,13 @@ impl ReconcileLookup { /// which aborts the pass just the same. fn is_retryable_failure(&self) -> bool { match self { - ReconcileLookup::Metadata(Err(e)) | ReconcileLookup::Orphan(Err(e)) => { + ReconcileLookup::Metadata(Err(e)) | ReconcileLookup::Orphan(Ok(Err(e))) => { reconcile_metadata_error_is_retryable(classify_drive_error(e)) } - ReconcileLookup::ParentFailed(_) => true, - ReconcileLookup::Metadata(Ok(_)) | ReconcileLookup::Orphan(Ok(_)) => false, + // A parent-walk failure propagates unclassified and aborts the pass + // just the same, so it halts the prefetch too. + ReconcileLookup::Orphan(Err(_)) => true, + ReconcileLookup::Metadata(Ok(_)) | ReconcileLookup::Orphan(Ok(Ok(_))) => false, } } } @@ -5045,7 +5048,9 @@ impl DefaultExecutor { } /// One paced CREATE-path lookup: re-derive the parent folder chain, then - /// search it for the orphan carrying `uuid`. + /// search it for the orphan carrying `uuid`. The OUTER `Err` is a failed + /// parent walk (propagated verbatim, aborting the pass); the INNER one is + /// `find_by_op_uuid`'s, which the caller classifies. /// /// Safe to run concurrently (issue #301). The parent walk /// ([`Self::ensure_parents_once`]) is idempotent AND single-flighted behind @@ -5060,31 +5065,28 @@ impl DefaultExecutor { relative_path: &RelativePath, uuid: &str, crypto: Option<&dyn SourceCryptoSuite>, - ) -> ReconcileLookup { - let parent_id = match self + ) -> anyhow::Result>> { + let parent_id = self .reconcile_parent_id(source, relative_path, crypto) - .await - { - Ok(id) => id, - Err(e) => return ReconcileLookup::ParentFailed(e), - }; + .await?; self.pacer.permit_request().await; - let found = match self - .remote - .find_by_op_uuid(&parent_id, uuid, &source.drive_context()) - .await - { - Ok(found) => { - self.pacer.note_response(ResponseClass::Ok); - Ok(found) - } - Err(e) => { - self.pacer - .note_response(classify_drive_error(&e).response_class()); - Err(e) - } - }; - ReconcileLookup::Orphan(found) + Ok( + match self + .remote + .find_by_op_uuid(&parent_id, uuid, &source.drive_context()) + .await + { + Ok(found) => { + self.pacer.note_response(ResponseClass::Ok); + Ok(found) + } + Err(e) => { + self.pacer + .note_response(classify_drive_error(&e).response_class()); + Err(e) + } + }, + ) } /// Issue #301: run every op's adopt-or-requeue lookup with BOUNDED @@ -5169,10 +5171,10 @@ impl DefaultExecutor { Some(file_id) => { ReconcileLookup::Metadata(self.reconcile_metadata_lookup(file_id).await) } - None => { + None => ReconcileLookup::Orphan( self.reconcile_orphan_lookup(source, &relative_path, &uuid, crypto) - .await - } + .await, + ), }; if lookup.is_retryable_failure() { halt.store(true, std::sync::atomic::Ordering::Release); @@ -5619,23 +5621,23 @@ impl DefaultExecutor { // `reconcile_orphan_lookup`), and carried the pacer accounting // with them. Fall back to an inline lookup when the prefetch was // skipped or halted. - let lookup = match prefetched { - Some( - lookup @ (ReconcileLookup::ParentFailed(_) | ReconcileLookup::Orphan(_)), - ) => lookup, + let orphan = match prefetched { + Some(ReconcileLookup::Orphan(result)) => result, + // Not prefetched (this op fell through from a stale + // resumable session, or the prefetch halted on a peer's + // retryable failure). A `Metadata` result cannot belong to + // this branch - it is selected by the ABSENCE of a recorded + // `drive_file_id` - so the same fallback covers both. _ => { self.reconcile_orphan_lookup(source, &op.relative_path, &uuid, crypto) .await } }; - let found = match lookup { - ReconcileLookup::ParentFailed(e) => return Err(to_reconcile_err(e)), - // The update path never reaches here (it is selected by a - // recorded `drive_file_id`), so a Metadata result cannot - // belong to this branch; treat it as "no orphan found", - // which requeues the op - the safe direction. - ReconcileLookup::Metadata(_) => Ok(None), - ReconcileLookup::Orphan(found) => found, + let found = match orphan { + Ok(found) => found, + // The parent-folder-chain walk failed: propagate verbatim + // (an invalid_grant in there still maps to needs_reauth). + Err(e) => return Err(to_reconcile_err(e)), }; let found = match found { Ok(found) => found, @@ -12087,21 +12089,164 @@ mod tests { "a transient metadata failure aborts the pass, so it must halt the prefetch" ); assert!( - ReconcileLookup::Orphan(Err(transient())).is_retryable_failure(), + ReconcileLookup::Orphan(Ok(Err(transient()))).is_retryable_failure(), "a transient orphan lookup failure must halt the prefetch" ); assert!( - ReconcileLookup::ParentFailed(anyhow::anyhow!("folder chain")).is_retryable_failure(), + ReconcileLookup::Orphan(Err(anyhow::anyhow!("folder chain"))).is_retryable_failure(), "a parent-walk failure propagates unclassified and aborts the pass, so it halts too" ); // A DEFINITIVE not-found drops the single op and the pass CONTINUES - // halting on it would needlessly serialise the rest. let definitive = anyhow::anyhow!("drive.dest_folder_missing"); assert!( - !ReconcileLookup::Orphan(Err(definitive)).is_retryable_failure(), + !ReconcileLookup::Orphan(Ok(Err(definitive))).is_retryable_failure(), "a definitive failure is per-op; it must NOT halt the prefetch" ); - assert!(!ReconcileLookup::Orphan(Ok(None)).is_retryable_failure()); + assert!(!ReconcileLookup::Orphan(Ok(Ok(None))).is_retryable_failure()); + } + + /// Issue #301 fail-fast, end to end: when the parent-folder-chain walk + /// fails, the pass aborts with EVERY pending op kept (they retry next + /// cycle), and the prefetch stops issuing lookups rather than burning one + /// doomed round trip per op. + /// + /// `with_dest_folder_missing` fails every write-target request, which is + /// what `ensure_folder` is - so each op's parent walk fails. There are more + /// ops than [`RECONCILE_LOOKUP_CONCURRENCY`], so the ones queued behind the + /// first batch observe the halt flag and are left for the sequential pass + /// (which never reaches them - it aborts on the first). + #[tokio::test] + async fn reconcile_parent_walk_failure_aborts_the_pass_and_keeps_every_op() { + const OPS: usize = 8; + let h = harness_with_remote(InMemoryRemoteStore::new().with_dest_folder_missing()).await; + + for i in 0..OPS { + // NESTED, so the parent walk actually has a folder to ensure. + let rel = RelativePath::try_from(format!("sub/f{i}.txt")).unwrap(); + h.state + .enqueue_pending_op(NewPendingOp { + source_id: h.source.id, + op_type: OP_TYPE_UPLOAD.to_string(), + relative_path: rel, + payload_json: PendingOpPayload { + client_op_uuid: Some(uuid::Uuid::new_v4().to_string()), + ..PendingOpPayload::default() + } + .to_value(), + scheduled_for: 0, + created_at: 0, + }) + .await + .unwrap(); + } + + let err = h + .executor() + .reconcile(&h.source, &noop_recover_sink) + .await + .expect_err("a failed parent walk must abort the source's reconcile"); + assert!( + err.to_string().contains("dest_folder_missing") + || err + .chain() + .any(|c| c.to_string().contains("dest_folder_missing")), + "the parent-walk error propagates verbatim: {err:?}" + ); + + // NOTHING was dropped: a failed lookup proves nothing about whether the + // create landed, so every op is kept for the next cycle. + assert_eq!( + h.state + .get_pending_ops_for_source(h.source.id) + .await + .unwrap() + .len(), + OPS, + "an aborted pass must keep every pending op" + ); + } + + /// The UPDATE path through the prefetched lookup: a crash after the update + /// landed on the remote leaves an object already carrying the op's uuid, so + /// reconcile ADOPTS it (marks the row Synced) and drops the op. + #[tokio::test] + async fn reconcile_adopts_an_update_whose_object_already_carries_the_op_uuid() { + let h = harness().await; + let body = b"updated bytes"; + let (rel, size) = h.write_file("u.txt", body); + + // The object as it exists remotely AFTER the update committed: it + // carries this op's uuid in appProperties. + let op_uuid = uuid::Uuid::new_v4().to_string(); + let mut app = HashMap::new(); + app.insert(CLIENT_OP_UUID_KEY.to_string(), op_uuid.clone()); + let entry = h + .remote + .create( + h.source.drive_folder_id.as_str(), + "u.txt", + "application/octet-stream", + UploadBody::Bytes(Bytes::copy_from_slice(body)), + app, + ) + .await + .unwrap(); + + // The pre-update row (an UPDATE op is selected by a recorded id). + h.state + .upsert_file_state(&crate::state::FileStateRow { + source_id: h.source.id, + relative_path: rel.clone(), + size, + mtime_ns: 1, + hash_blake3: *blake3::hash(b"older bytes").as_bytes(), + drive_file_id: Some(entry.id.clone()), + drive_md5: None, + encrypted_remote_path: None, + status: FileStateStatus::Pending, + last_uploaded_at: Some(0), + last_verified_at: Some(0), + }) + .await + .unwrap(); + h.state + .enqueue_pending_op(NewPendingOp { + source_id: h.source.id, + op_type: OP_TYPE_UPLOAD.to_string(), + relative_path: rel.clone(), + payload_json: PendingOpPayload { + client_op_uuid: Some(op_uuid), + drive_file_id: Some(entry.id.clone()), + uploaded_blake3_hex: Some(hex::encode(blake3::hash(body).as_bytes())), + ..PendingOpPayload::default() + } + .to_value(), + scheduled_for: 0, + created_at: 0, + }) + .await + .unwrap(); + + h.executor() + .reconcile(&h.source, &noop_recover_sink) + .await + .unwrap(); + + let row = h + .state + .get_file_state(h.source.id, &rel) + .await + .unwrap() + .expect("file_state row"); + assert_eq!(row.status, FileStateStatus::Synced, "the update is adopted"); + assert_eq!(row.drive_file_id.as_deref(), Some(entry.id.as_str())); + assert!(h + .state + .get_pending_ops_for_source(h.source.id) + .await + .unwrap() + .is_empty()); } // --- crash mid-resumable resumes via reconcile --------------------------