From bddb5063ad74e6f910b1a74344676fa02a795ddf Mon Sep 17 00:00:00 2001 From: zhom <2717306+zhom@users.noreply.github.com> Date: Sat, 3 Oct 2026 02:24:34 +0400 Subject: [PATCH] refactor: cleanup --- .github/dependabot.yml | 16 - donut-sync/src/auth/auth.guard.spec.ts | 196 +++++++++ donut-sync/src/auth/auth.guard.ts | 4 + e2e/app/Cargo.lock | 8 +- e2e/tests/motion.test.mjs | 20 +- e2e/tests/sync.test.mjs | 226 +++++++++- package.json | 2 +- src-tauri/Cargo.lock | 2 +- src-tauri/Cargo.toml | 2 +- src-tauri/build.rs | 12 +- src-tauri/src/agent.rs | 43 +- src-tauri/src/browser_runner.rs | 2 +- src-tauri/src/cloud_auth.rs | 440 ++++++++++++++++++- src-tauri/src/cloud_errors.rs | 4 + src-tauri/src/cookie_bot.rs | 380 ++++++++++++++-- src-tauri/src/fingerprint_consistency.rs | 4 + src-tauri/src/lib.rs | 5 + src-tauri/src/proxy_runner.rs | 83 ++-- src-tauri/src/proxy_server.rs | 1 + src-tauri/src/remote_handoff.rs | 303 ++++++++++--- src-tauri/src/remote_session.rs | 96 +++- src-tauri/src/sync/engine.rs | 152 ++++++- src-tauri/src/sync/mod.rs | 33 ++ src-tauri/src/sync/scheduler.rs | 19 +- src-tauri/src/sync/subscription.rs | 248 ++++++++++- src-tauri/src/team_lock.rs | 487 ++++++++++++++++++--- src-tauri/src/wayfern_manager.rs | 243 ++++++++-- src-tauri/tauri.conf.json | 2 +- src-tauri/tests/donut_proxy_integration.rs | 113 +++++ src/hooks/use-cookie-bot.ts | 23 +- src/hooks/use-team-locks.ts | 17 +- 31 files changed, 2787 insertions(+), 399 deletions(-) create mode 100644 donut-sync/src/auth/auth.guard.spec.ts diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 9b4b9fc9..e99996fd 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -69,22 +69,6 @@ updates: include: "scope" # Standalone SDKs for the local REST API - - package-ecosystem: "npm" - directory: "/sdk/node" - schedule: - interval: "weekly" - day: "saturday" - time: "09:00" - allow: - - dependency-type: "all" - groups: - sdk-node-dependencies: - patterns: - - "*" - commit-message: - prefix: "deps(sdk)" - include: "scope" - - package-ecosystem: "pip" directory: "/sdk/python" schedule: diff --git a/donut-sync/src/auth/auth.guard.spec.ts b/donut-sync/src/auth/auth.guard.spec.ts new file mode 100644 index 00000000..c2c7f63a --- /dev/null +++ b/donut-sync/src/auth/auth.guard.spec.ts @@ -0,0 +1,196 @@ +import { generateKeyPairSync } from "node:crypto"; +import { + type ExecutionContext, + Logger, + UnauthorizedException, +} from "@nestjs/common"; +import type { ConfigService } from "@nestjs/config"; +import * as jwt from "jsonwebtoken"; +import { AuthGuard } from "./auth.guard.js"; +import type { UserContext } from "./user-context.interface.js"; + +const signingKeys = generateKeyPairSync("rsa", { + modulusLength: 2048, + publicKeyEncoding: { type: "spki", format: "pem" }, + privateKeyEncoding: { type: "pkcs8", format: "pem" }, +}); + +const otherKeys = generateKeyPairSync("rsa", { + modulusLength: 2048, + publicKeyEncoding: { type: "spki", format: "pem" }, + privateKeyEncoding: { type: "pkcs8", format: "pem" }, +}); + +const USER_ID = "11111111-2222-3333-4444-555555555555"; +const SELF_HOSTED_TOKEN = "self-hosted-secret"; + +function guardWith(env: Record): AuthGuard { + const config = { + get: (key: string) => env[key], + } as unknown as ConfigService; + return new AuthGuard(config); +} + +function cloudGuard(): AuthGuard { + return guardWith({ SYNC_JWT_PUBLIC_KEY: signingKeys.publicKey }); +} + +function contextFor(token: string): { + context: ExecutionContext; + request: { headers: Record; user?: UserContext }; +} { + const request: { headers: Record; user?: UserContext } = { + headers: { authorization: `Bearer ${token}` }, + }; + const context = { + switchToHttp: () => ({ getRequest: () => request }), + } as unknown as ExecutionContext; + return { context, request }; +} + +function sign( + payload: Record, + privateKey: string = signingKeys.privateKey, + expiresIn = 300, +): string { + return jwt.sign(payload, privateKey, { algorithm: "RS256", expiresIn }); +} + +function syncToken(overrides: Record = {}): string { + return sign({ + sub: USER_ID, + prefix: `users/${USER_ID}/`, + profileLimit: 5, + scope: "sync", + ...overrides, + }); +} + +describe("AuthGuard", () => { + let warnSpy: jest.SpyInstance; + + beforeEach(() => { + warnSpy = jest + .spyOn(Logger.prototype, "warn") + .mockImplementation(() => undefined); + jest.spyOn(Logger.prototype, "log").mockImplementation(() => undefined); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it("accepts a sync token and scopes the request to its own namespace", async () => { + const { context, request } = contextFor(syncToken()); + + await expect(cloudGuard().canActivate(context)).resolves.toBe(true); + expect(request.user).toEqual({ + mode: "cloud", + prefix: `users/${USER_ID}/`, + profileLimit: 5, + sub: USER_ID, + }); + }); + + it("refuses an account access token signed with the same key", async () => { + const accessToken = sign({ + sub: USER_ID, + email: "owner@example.com", + sid: "device-1", + }); + const { context, request } = contextFor(accessToken); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + expect(request.user).toBeUndefined(); + expect(warnSpy).toHaveBeenCalled(); + }); + + it("refuses a per-profile agent token signed with the same key", async () => { + const agentToken = sign({ + sub: USER_ID, + email: "owner@example.com", + sid: "agent:run-1", + scope: "agent", + profile: "profile-1", + }); + const { context, request } = contextFor(agentToken); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + expect(request.user).toBeUndefined(); + }); + + it.each([ + ["an array", ["sync"]], + ["another case", "SYNC"], + ["an empty string", ""], + ["null", null], + ])("refuses a scope that is %s", async (_label, scope) => { + const { context } = contextFor(syncToken({ scope })); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + }); + + it("refuses a sync token signed with a different key", async () => { + const forged = sign( + { + sub: USER_ID, + prefix: `users/${USER_ID}/`, + profileLimit: 5, + scope: "sync", + }, + otherKeys.privateKey, + ); + const { context } = contextFor(forged); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + }); + + it("refuses an expired sync token", async () => { + const expired = sign( + { + sub: USER_ID, + prefix: `users/${USER_ID}/`, + profileLimit: 5, + scope: "sync", + }, + signingKeys.privateKey, + -10, + ); + const { context } = contextFor(expired); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + }); + + it("refuses a sync token whose prefix is not a single user namespace", async () => { + const { context } = contextFor(syncToken({ prefix: "users/" })); + + await expect(cloudGuard().canActivate(context)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + }); + + it("still accepts the configured token of a self-hosted server", async () => { + const guard = guardWith({ + SYNC_TOKEN: SELF_HOSTED_TOKEN, + SYNC_JWT_PUBLIC_KEY: signingKeys.publicKey, + }); + const { context, request } = contextFor(SELF_HOSTED_TOKEN); + + await expect(guard.canActivate(context)).resolves.toBe(true); + expect(request.user).toEqual({ + mode: "self-hosted", + prefix: "", + profileLimit: 0, + }); + }); +}); diff --git a/donut-sync/src/auth/auth.guard.ts b/donut-sync/src/auth/auth.guard.ts index 40fbaf3b..3298a403 100644 --- a/donut-sync/src/auth/auth.guard.ts +++ b/donut-sync/src/auth/auth.guard.ts @@ -118,6 +118,10 @@ export class AuthGuard implements CanActivate { algorithms: ["RS256"], }) as jwt.JwtPayload; + if (decoded.scope !== "sync") { + throw new Error("Token scope is not accepted"); + } + const sub = typeof decoded.sub === "string" ? decoded.sub : ""; // Validate the prefix claim SHAPE before trusting it as an S3 key // prefix. An empty/over-broad prefix would make validateKeyAccess diff --git a/e2e/app/Cargo.lock b/e2e/app/Cargo.lock index 1b6781ce..95c566a5 100644 --- a/e2e/app/Cargo.lock +++ b/e2e/app/Cargo.lock @@ -969,9 +969,9 @@ dependencies = [ [[package]] name = "cfg-expr" -version = "0.20.9" +version = "0.20.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe4ece8474b5f766c63426647e7b4b316b67431ade1036a8313cee24a03ae917" +checksum = "9ba9e9ec16c447027685b1f897b720e18e9a8afd00bd7332c483537e38086c9f" dependencies = [ "smallvec", "target-lexicon 0.13.5", @@ -1724,7 +1724,7 @@ dependencies = [ [[package]] name = "donutbrowser" -version = "0.31.2" +version = "0.31.3" dependencies = [ "aes 0.9.3", "aes-gcm 0.11.1", @@ -6516,7 +6516,7 @@ version = "7.0.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "396a35feb67335377e0251fcbc1092fc85c484bd4e3a7a54319399da127796e7" dependencies = [ - "cfg-expr 0.20.9", + "cfg-expr 0.20.10", "heck 0.5.0", "pkg-config", "toml 1.1.6+spec-1.1.0", diff --git a/e2e/tests/motion.test.mjs b/e2e/tests/motion.test.mjs index 5778852a..62172247 100644 --- a/e2e/tests/motion.test.mjs +++ b/e2e/tests/motion.test.mjs @@ -1601,12 +1601,14 @@ test("schedule displays every timezone and slot, and skipped run details work by const schedules = arguments[0], id = arguments[1]; const previous = window.fetch; window.__donutScheduleCalls = []; + window.__donutRemainingHours = 199.8; const response = value => Promise.resolve(new Response(JSON.stringify(value), { status: 200, headers: { "content-type": "application/json", "Tauri-Response": "ok" } })); window.fetch = function(input, init) { let command = ""; try { const url = new URL(typeof input === "string" ? input : input.url); if (url.hostname === "localhost" && url.protocol === "ipc:" || url.hostname === "ipc.localhost") command = decodeURIComponent(url.pathname.split("/").pop() || ""); } catch {} if (command === "get_cookie_bot_schedules") return response({ schedules }); - if (command === "get_remote_hours_quota") return response({ granted_hours: 200, remaining_hours: 199.8, used_hours: 0.2, seats: 1, per_seat_hours: 200, members: [] }); + if (command === "start_remote_session_events") { window.__donutScheduleCalls.push(command); return response(null); } + if (command === "get_remote_hours_quota") return response({ granted_hours: 200, remaining_hours: window.__donutRemainingHours, used_hours: 200 - window.__donutRemainingHours, seats: 1, per_seat_hours: 200, members: [] }); if (command === "get_cookie_bot_runs") return response({ runs: [{ id: "skipped-fixture", profile_id: id, profile_name: "Two daily slots", status: "skipped", scheduled_for: "2026-09-08T02:00:00Z", max_minutes: 30, chunks_total: 1, chunk_index: 0, sites_total: 0, sites_visited: 0, sites_failed: 0, consent_dismissed: 0, billed_seconds: 0, outcome_code: "profile_locked" }], next_before: null }); return previous.apply(window, arguments); }; @@ -1628,6 +1630,22 @@ test("schedule displays every timezone and slot, and skipped run details work by ); assert.equal(await app.visibleTextIncludes("Asia/Yerevan"), true); assert.equal(await app.visibleTextIncludes("199.8"), true); + await app.execute("window.__donutRemainingHours = 199.7;"); + await emit(app, "remote-session-snapshot", { sessions: [] }); + await app.waitForText("199.7"); + const connections = await app.execute( + "return window.__donutScheduleCalls.length;", + ); + await app.execute(` + window.__donutRemainingHours = 199.6; + window.dispatchEvent(new Event("online")); + `); + await app.waitForText("199.6"); + assert.ok( + (await app.execute("return window.__donutScheduleCalls.length;")) > + connections, + "network recovery wakes the remote session connection", + ); await app.capture("every-schedule-slot"); await app.clickText(en.cookieBot.tabs.activity, { roles: ["tab"] }); await app.clickSelector('[role="combobox"]'); diff --git a/e2e/tests/sync.test.mjs b/e2e/tests/sync.test.mjs index f7534e75..d370fd00 100644 --- a/e2e/tests/sync.test.mjs +++ b/e2e/tests/sync.test.mjs @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { mkdir, readdir, readFile, writeFile } from "node:fs/promises"; +import { mkdir, readdir, readFile, utimes, writeFile } from "node:fs/promises"; import { createServer } from "node:http"; import path from "node:path"; import test from "node:test"; @@ -80,6 +80,193 @@ async function waitFor(app, callback, description, timeoutMs = 45_000) { return app.waitFor(callback, { description, timeoutMs, intervalMs: 250 }); } +test("remote profile changes return after a long outage without another app restart", async () => { + assert.ok(syncUrl && syncToken, "Sync infrastructure was not started"); + const app = appFromEnvironment("sync-remote-return"); + let mode = "missing"; + let profileId; + let missingManifests = 0; + let failedPulls = 0; + let failedFiles = 0; + let proxyUrl; + const proxy = createServer(async (request, response) => { + try { + const chunks = []; + for await (const chunk of request) chunks.push(chunk); + const body = Buffer.concat(chunks).toString(); + const input = body ? JSON.parse(body) : {}; + if (request.url === "/failed-file") { + failedFiles += 1; + response.writeHead(503).end(); + return; + } + if (request.url.endsWith("/subscribe")) { + response.writeHead(503).end(); + return; + } + if (mode === "offline") { + if (input.key === `profiles/${profileId}/metadata.json`) { + failedPulls += 1; + } + response.writeHead(503).end(); + return; + } + if ( + mode === "missing" && + request.url.endsWith("/stat") && + input.key === `profiles/${profileId}/manifest.json` + ) { + missingManifests += 1; + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ exists: false })); + return; + } + const upstream = await fetch(`${syncUrl}${request.url}`, { + method: request.method, + headers: { + authorization: `Bearer ${syncToken}`, + "content-type": "application/json", + }, + ...(body ? { body } : {}), + signal: AbortSignal.timeout(10_000), + }); + let text = await upstream.text(); + if ( + mode === "partial" && + request.url.endsWith("/presign-download-batch") + ) { + const result = JSON.parse(text); + for (const item of result.items ?? []) { + if (item.key.endsWith("/History")) + item.url = `${proxyUrl}/failed-file`; + } + text = JSON.stringify(result); + } + response.writeHead(upstream.status, { + "content-type": "application/json", + }); + response.end(text); + } catch { + response.writeHead(502).end(); + } + }); + await new Promise((resolve) => proxy.listen(0, "127.0.0.1", resolve)); + proxyUrl = `http://127.0.0.1:${proxy.address().port}`; + + try { + await app.start(); + await configureSync(app); + const profile = await createProfile(app, "Remote Return"); + profileId = profile.id; + const profileRoot = path.join(app.dataRoot, "data", "profiles", profileId); + const data = path.join(profileRoot, "profile", "Default"); + const preferences = path.join(data, "Preferences"); + const history = path.join(data, "History"); + await mkdir(data, { recursive: true }); + await writeFile(preferences, "remote preferences"); + await writeFile(history, "remote history"); + await app.invoke("set_profile_sync_mode", { + profileId, + syncMode: "Regular", + }); + await app.invoke("request_profile_sync", { profileId }); + await waitFor( + app, + async () => + (await app.invoke("list_browser_profiles")).some( + (item) => item.id === profileId && item.last_sync != null, + ), + "the source profile finishes its upload", + ); + const manifestKey = `profiles/${profileId}/manifest.json`; + const originalManifest = await downloadRemote(manifestKey); + await app.invoke("save_sync_settings", { + syncServerUrl: proxyUrl, + syncToken, + }); + await app.close(); + + await writeFile(preferences, "stale local preferences"); + await writeFile(history, "stale local history"); + const later = new Date(Date.now() + 60_000); + await Promise.all([ + utimes(preferences, later, later), + utimes(history, later, later), + ]); + await mkdir(path.join(profileRoot, ".donut-sync"), { recursive: true }); + await writeFile( + path.join(profileRoot, ".donut-sync", "resume-state.json"), + JSON.stringify({ + profile_id: profileId, + direction: "download", + started_at: new Date().toISOString(), + completed_files: [ + "profile/Default/Preferences", + "profile/Default/History", + ], + }), + ); + await writeFile( + path.join(app.dataRoot, "data", "settings", "remote_handoff.json"), + JSON.stringify({ + [profileId]: { + session_id: "remote-return-session", + state: "pending_sync", + observed_at: Math.floor(Date.now() / 1000), + }, + }), + ); + await app.start(); + const pending = async () => + (await app.invoke("get_remote_handoff_states"))[profileId] === + "pending_sync"; + await waitFor( + app, + () => missingManifests >= 2, + "return waits for the remote manifest", + ); + assert.equal(await pending(), true); + assert.deepEqual(await downloadRemote(manifestKey), originalManifest); + + mode = "offline"; + await app.invoke("request_profile_sync", { profileId }); + await waitFor( + app, + () => failedPulls >= 6, + "return keeps trying after five failures", + 120_000, + ); + assert.equal(await pending(), true); + + mode = "partial"; + await waitFor( + app, + () => failedFiles >= 3, + "a non-critical file fails to return", + ); + assert.equal(await pending(), true); + assert.deepEqual(await downloadRemote(manifestKey), originalManifest); + + mode = "healthy"; + await waitFor( + app, + async () => !(await pending()), + "return completes without a restart", + 25_000, + ); + assert.equal(await readFile(preferences, "utf8"), "remote preferences"); + assert.equal(await readFile(history, "utf8"), "remote history"); + assert.deepEqual(await downloadRemote(manifestKey), originalManifest); + } catch (error) { + await app.capture("failure"); + throw error; + } finally { + await app.close(); + proxy.closeAllConnections(); + await new Promise((resolve) => proxy.close(resolve)); + } +}); + test("two real app devices reconcile profile files and every config entity with last-write-wins", async () => { assert.ok(syncUrl && syncToken, "Sync infrastructure was not started"); const deviceA = appFromEnvironment("sync-regular-a"); @@ -339,6 +526,43 @@ test("two real app devices reconcile profile files and every config entity with } }); +test("the first sync of an empty profile publishes a manifest and records completion", async () => { + const source = appFromEnvironment("sync-empty-source"); + try { + await source.start(); + await configureSync(source); + const profile = await createProfile(source, "Empty Profile"); + await source.invoke("set_profile_sync_mode", { + profileId: profile.id, + syncMode: "Regular", + }); + await source.invoke("request_profile_sync", { profileId: profile.id }); + const manifestKey = `profiles/${profile.id}/manifest.json`; + await waitFor( + source, + async () => { + const profiles = await source.invoke("list_browser_profiles"); + return ( + profiles.find((item) => item.id === profile.id)?.last_sync > 0 && + (await listRemote(`profiles/${profile.id}/`)).some( + (item) => item.key === manifestKey, + ) + ); + }, + "empty profile has a remote manifest and a completed sync", + ); + const manifest = JSON.parse( + (await downloadRemote(manifestKey)).toString("utf8"), + ); + assert.deepEqual(manifest.files, []); + } catch (error) { + await source.capture("failure"); + throw error; + } finally { + await source.close(); + } +}); + test("global config sealing and encrypted profile sync reject a wrong password, round-trip with the right one, and roll over", async () => { const source = appFromEnvironment("sync-encrypted-source"); const receiver = appFromEnvironment("sync-encrypted-receiver"); diff --git a/package.json b/package.json index 7318b874..757ede8d 100644 --- a/package.json +++ b/package.json @@ -2,7 +2,7 @@ "name": "donutbrowser", "private": true, "license": "AGPL-3.0", - "version": "0.31.2", + "version": "0.31.3", "type": "module", "scripts": { "predev": "pnpm licenses:generate", diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 71db0e8d..725dfe96 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -1742,7 +1742,7 @@ dependencies = [ [[package]] name = "donutbrowser" -version = "0.31.2" +version = "0.31.3" dependencies = [ "aes 0.9.3", "aes-gcm 0.11.1", diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 50526361..29937c43 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "donutbrowser" -version = "0.31.2" +version = "0.31.3" description = "Simple Yet Powerful Anti-Detect Browser" authors = ["zhom@github"] edition = "2021" diff --git a/src-tauri/build.rs b/src-tauri/build.rs index 01c9137a..f6b9c0b3 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -10,8 +10,7 @@ fn main() { // Generate tray icon PNG files from SVG (macOS template icon format) generate_tray_icons(); - #[cfg(target_os = "macos")] - { + if build_target.ends_with("-apple-darwin") { println!("cargo:rustc-link-lib=framework=CoreFoundation"); println!("cargo:rustc-link-lib=framework=CoreServices"); } @@ -72,16 +71,16 @@ fn main() { // or they crash with STATUS_ENTRYPOINT_NOT_FOUND (0xc0000139). We embed the // manifest for all targets, then suppress the duplicate for bins with /MANIFEST:NO // (tauri_build's resource-embedded manifest still takes effect for bins). - #[cfg(target_os = "windows")] - { + if build_target.contains("windows") { embed_windows_manifest(); println!("cargo:rustc-link-arg-bins=/MANIFEST:NO"); } } else { println!("cargo:warning=Skipping tauri_build: external binaries not found. This is expected when building sidecar binaries."); - #[cfg(target_os = "windows")] - embed_windows_manifest(); + if build_target.contains("windows") { + embed_windows_manifest(); + } } } @@ -139,7 +138,6 @@ fn ensure_dist_folder_exists() { println!("cargo:rerun-if-changed=../dist"); } -#[cfg(target_os = "windows")] fn embed_windows_manifest() { use std::path::PathBuf; diff --git a/src-tauri/src/agent.rs b/src-tauri/src/agent.rs index 637b6483..a011431b 100644 --- a/src-tauri/src/agent.rs +++ b/src-tauri/src/agent.rs @@ -548,7 +548,7 @@ pub async fn update_recipe( /// Delete one. pub async fn delete_recipe(id: &str) -> Result { - let outcome: RecipeDeleted = request( + let outcome: Option = request( reqwest::Method::DELETE, format!("{}/recipes/{}", base(), urlencoding::encode(id)), Vec::new(), @@ -557,7 +557,11 @@ pub async fn delete_recipe(id: &str) -> Result { ) .await .map_err(|e| agent_error("recipe delete", e))?; - Ok(outcome.deleted.unwrap_or(true)) + Ok(recipe_deleted(outcome)) +} + +fn recipe_deleted(outcome: Option) -> bool { + outcome.and_then(|reply| reply.deleted).unwrap_or(true) } #[derive(Debug, Default, Deserialize)] @@ -925,16 +929,25 @@ async fn request( return Err(format!("({status}) {text}")); } - response - .json::() + let bytes = response + .bytes() .await - .map_err(|e| format!("decode response: {e}")) + .map_err(|e| format!("decode response: {e}"))?; + decode_body::(&bytes).map_err(|e| format!("decode response: {e}")) } }) .await .map_err(|e| AgentError(cloud_errors::classify_message(&e, codes))) } +fn decode_body(bytes: &[u8]) -> Result { + if bytes.iter().all(u8::is_ascii_whitespace) { + serde_json::from_slice(b"null") + } else { + serde_json::from_slice(bytes) + } +} + // --- Tauri commands --------------------------------------------------------- // // Defined here rather than in `lib.rs` because every local precondition they @@ -1026,6 +1039,26 @@ pub fn get_agent_run_events_status() -> Option { mod tests { use super::*; + #[test] + fn a_recipe_delete_with_no_body_is_a_success() { + let empty: Option = decode_body(b"").expect("a 204 has no body"); + assert!(recipe_deleted(empty)); + let blank: Option = decode_body(b" \n").expect("whitespace is no body"); + assert!(recipe_deleted(blank)); + let refused: Option = + decode_body(br#"{"deleted":false}"#).expect("an explicit answer"); + assert!(!recipe_deleted(refused)); + let bare: Option = decode_body(b"{}").expect("an empty object"); + assert!(recipe_deleted(bare)); + } + + #[test] + fn an_empty_body_is_still_a_failure_where_a_value_is_required() { + assert!(decode_body::(b"").is_err()); + assert!(decode_body::(b"").is_err()); + assert!(decode_body::>(b"").is_err()); + } + #[test] fn a_blank_goal_is_refused_before_anything_is_spent() { for goal in ["", " ", "\n\t "] { diff --git a/src-tauri/src/browser_runner.rs b/src-tauri/src/browser_runner.rs index ff8250d4..d9d20cf5 100644 --- a/src-tauri/src/browser_runner.rs +++ b/src-tauri/src/browser_runner.rs @@ -1665,7 +1665,7 @@ async fn unwind_launch(profile: &BrowserProfile, acquired_team_lock: bool) { /// Whether a live browser process is recorded for this profile right now. /// Re-read from disk: the caller's copy predates the launch attempt. -fn browser_is_running_for(profile_id: &str) -> bool { +pub(crate) fn browser_is_running_for(profile_id: &str) -> bool { BrowserRunner::instance() .profile_manager .list_profiles() diff --git a/src-tauri/src/cloud_auth.rs b/src-tauri/src/cloud_auth.rs index b6ec6a32..54da792a 100644 --- a/src-tauri/src/cloud_auth.rs +++ b/src-tauri/src/cloud_auth.rs @@ -4,6 +4,7 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::fs; use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; use tokio::sync::Mutex; use crate::browser::ProxySettings; @@ -18,6 +19,60 @@ pub const CLOUD_SYNC_URL: &str = "https://sync.donutbrowser.com"; /// cloud API has not sent one. const DEFAULT_REQUESTS_PER_HOUR: i64 = 100; +const ACCESS_TOKEN_REFRESH_MARGIN_SECS: i64 = 15 * 60; +const SYNC_TOKEN_REFRESH_MARGIN_SECS: i64 = 120; +const DEFAULT_RETRY_AFTER: Duration = Duration::from_secs(60); +const MAX_RETRY_AFTER: Duration = Duration::from_secs(60 * 60); + +pub(crate) fn parse_retry_after(value: &str, now: chrono::DateTime) -> Option { + let value = value.trim(); + if let Ok(seconds) = value.parse::() { + return Some(Duration::from_secs(seconds).min(MAX_RETRY_AFTER)); + } + let at = chrono::DateTime::parse_from_rfc2822(value).ok()?; + let seconds = u64::try_from((at.with_timezone(&Utc) - now).num_seconds()).unwrap_or(0); + Some(Duration::from_secs(seconds).min(MAX_RETRY_AFTER)) +} + +pub(crate) fn retry_after(headers: &reqwest::header::HeaderMap) -> Duration { + headers + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| parse_retry_after(value, Utc::now())) + .unwrap_or(DEFAULT_RETRY_AFTER) +} + +fn is_refused_refresh(error: &str) -> bool { + error.contains("(401") || error.contains("No refresh token") +} + +#[derive(Debug)] +pub(crate) enum AuthorizedError { + SignedOut, + RateLimited(Duration), + Transport(String), +} + +impl AuthorizedError { + fn from_token_error(error: String) -> Self { + if error.contains("Not logged in") || is_refused_refresh(&error) { + Self::SignedOut + } else { + Self::Transport(error) + } + } +} + +impl std::fmt::Display for AuthorizedError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::SignedOut => write!(f, "not signed in"), + Self::RateLimited(wait) => write!(f, "rate limited for {}s", wait.as_secs()), + Self::Transport(error) => write!(f, "{error}"), + } + } +} + /// Capability + limit set the account is entitled to, derived from its plan. /// Mirrors the entitlement set the cloud API sends. Features are gated on these /// flags instead of a single "is paid?" boolean, so a plan like "solo" (cloud @@ -341,6 +396,7 @@ pub struct CloudAuthManager { client: Client, state: Mutex>, refresh_lock: tokio::sync::Mutex<()>, + refresh_deferred_until: std::sync::Mutex>, wayfern_token: Mutex>, /// The server's reason for refusing this desktop a Wayfern token under a /// device rule, until a token is acquired again. The engine reports a @@ -366,6 +422,7 @@ impl CloudAuthManager { client, state: Mutex::new(state), refresh_lock: tokio::sync::Mutex::new(()), + refresh_deferred_until: std::sync::Mutex::new(None), wayfern_token: Mutex::new(None), wayfern_device_refusal: std::sync::Mutex::new(None), } @@ -426,6 +483,14 @@ impl CloudAuthManager { Self::decrypt_from_file(&path, b"DBCST") } + pub(crate) fn discard_cloud_sync_token(refused: &str) { + if let Ok(Some(cached)) = Self::load_cloud_sync_token() { + if cached == refused { + let _ = fs::remove_file(Self::get_settings_dir().join("cloud_sync_token.dat")); + } + } + } + fn store_auth_state(state: &CloudAuthState) -> Result<(), String> { let path = Self::get_settings_dir().join("cloud_auth_state.json"); if let Some(parent) = path.parent() { @@ -465,10 +530,10 @@ impl CloudAuthManager { // --- JWT expiry check --- - fn is_jwt_expiring_soon(token: &str) -> bool { + fn jwt_expiry(token: &str) -> Option { let parts: Vec<&str> = token.split('.').collect(); if parts.len() != 3 { - return true; + return None; } use base64::{engine::general_purpose, Engine as _}; @@ -476,25 +541,32 @@ impl CloudAuthManager { Ok(bytes) => bytes, Err(_) => { // Try standard base64 with padding - match general_purpose::STANDARD.decode(parts[1]) { - Ok(bytes) => bytes, - Err(_) => return true, - } + general_purpose::STANDARD.decode(parts[1]).ok()? } }; - let json: serde_json::Value = match serde_json::from_slice(&payload) { - Ok(v) => v, - Err(_) => return true, - }; + let json: serde_json::Value = serde_json::from_slice(&payload).ok()?; + json.get("exp").and_then(|v| v.as_i64()) + } - let exp = match json.get("exp").and_then(|v| v.as_i64()) { - Some(exp) => exp, - None => return true, - }; + fn jwt_expires_within(token: &str, margin_secs: i64, now: i64) -> bool { + Self::jwt_expiry(token).is_none_or(|exp| exp - now < margin_secs) + } - let now = Utc::now().timestamp(); - exp - now < 120 + fn is_jwt_expiring_soon(token: &str) -> bool { + Self::jwt_expires_within( + token, + SYNC_TOKEN_REFRESH_MARGIN_SECS, + Utc::now().timestamp(), + ) + } + + fn access_token_needs_refresh(token: &str) -> bool { + Self::jwt_expires_within( + token, + ACCESS_TOKEN_REFRESH_MARGIN_SECS, + Utc::now().timestamp(), + ) } // --- API methods --- @@ -612,6 +684,16 @@ impl CloudAuthManager { pub async fn refresh_access_token(&self) -> Result<(), String> { let _guard = self.refresh_lock.lock().await; + self.refresh_while_locked().await + } + + async fn refresh_while_locked(&self) -> Result<(), String> { + if let Some(wait) = self.refresh_deferral() { + return Err(format!( + "Token refresh deferred for {}s (429 Too Many Requests)", + wait.as_secs() + )); + } log::info!("Refreshing access token (holding lock)..."); let refresh_token = @@ -629,6 +711,14 @@ impl CloudAuthManager { if !response.status().is_success() { let status = response.status(); log::warn!("Token refresh failed ({status})"); + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + self.defer_refresh(retry_after(response.headers())); + } else if status == reqwest::StatusCode::UNAUTHORIZED { + Self::forget_refresh_token(); + tauri::async_runtime::spawn(async { + CLOUD_AUTH.invalidate_session().await; + }); + } return Err(format!("Token refresh failed ({status})")); } @@ -644,6 +734,102 @@ impl CloudAuthManager { Ok(()) } + fn refresh_deferral(&self) -> Option { + let until = (*self + .refresh_deferred_until + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner))?; + until.checked_duration_since(Instant::now()) + } + + fn defer_refresh(&self, wait: Duration) { + *self + .refresh_deferred_until + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Instant::now() + wait); + } + + fn forget_refresh_token() { + let path = Self::get_settings_dir().join("cloud_refresh_token.dat"); + if path.exists() { + let _ = fs::remove_file(path); + } + } + + async fn ensure_fresh_access_token(&self) -> Result { + let token = Self::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; + if !Self::access_token_needs_refresh(&token) { + return Ok(token); + } + let _guard = self.refresh_lock.lock().await; + let current = Self::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; + if current != token && !Self::access_token_needs_refresh(¤t) { + return Ok(current); + } + self.refresh_while_locked().await?; + Self::load_access_token()?.ok_or_else(|| "Not logged in after refresh".to_string()) + } + + async fn usable_access_token(&self) -> Result { + let token = Self::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; + if !Self::access_token_needs_refresh(&token) { + return Ok(token); + } + match self.ensure_fresh_access_token().await { + Ok(fresh) => Ok(fresh), + Err(e) if e.contains("(401") => Err(e), + Err(e) => { + log::debug!("Early access token refresh did not complete: {e}"); + Ok(token) + } + } + } + + async fn refresh_after_rejection(&self, rejected: &str) -> Result { + let _guard = self.refresh_lock.lock().await; + if let Some(current) = Self::load_access_token()? { + if current != rejected && !current.is_empty() { + log::info!("Token was already refreshed by another caller, retrying..."); + return Ok(current); + } + } + self.refresh_while_locked().await?; + Self::load_access_token()?.ok_or_else(|| "Not logged in after refresh".to_string()) + } + + pub(crate) async fn authorized_request( + &self, + build: F, + ) -> Result + where + F: Fn(&Client, &str) -> reqwest::RequestBuilder + Send, + { + let token = self + .usable_access_token() + .await + .map_err(AuthorizedError::from_token_error)?; + let mut response = build(&self.client, &token) + .send() + .await + .map_err(|e| AuthorizedError::Transport(e.to_string()))?; + if response.status() == reqwest::StatusCode::UNAUTHORIZED { + let token = self + .refresh_after_rejection(&token) + .await + .map_err(AuthorizedError::from_token_error)?; + response = build(&self.client, &token) + .send() + .await + .map_err(|e| AuthorizedError::Transport(e.to_string()))?; + } + if response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + return Err(AuthorizedError::RateLimited(retry_after( + response.headers(), + ))); + } + Ok(response) + } + /// Invalidate the session: clear all auth state and notify the frontend. /// Only call this when the session is definitively dead (explicit logout /// or repeated background refresh failures). @@ -663,6 +849,10 @@ impl CloudAuthManager { // itself is retired from the account page. Self::forget_mcp_key_locally("after the session expired"); self.clear_auth().await; + crate::team_lock::PROFILE_LOCK.disconnect().await; + sync::stop_subscription(); + crate::remote_session::stop_session_events(); + crate::agent::stop_run_events(); let _ = crate::events::emit_empty("cloud-auth-expired"); } @@ -937,6 +1127,10 @@ impl CloudAuthManager { self.clear_wayfern_token().await; self.set_wayfern_device_refusal(None); + crate::team_lock::PROFILE_LOCK + .release_all_held_within(crate::team_lock::SHUTDOWN_RELEASE_TIMEOUT) + .await; + // Disconnect profile lock manager crate::team_lock::PROFILE_LOCK.disconnect().await; @@ -1124,10 +1318,23 @@ impl CloudAuthManager { state.clone() } + async fn plan_marker(&self) -> Option<(String, bool, bool)> { + let state = self.state.lock().await; + state.as_ref().map(|auth| { + let entitlements = auth.user.entitlements(); + ( + auth.user.effective_plan().to_string(), + entitlements.active, + entitlements.cloud_backup, + ) + }) + } + async fn clear_auth(&self) { let mut state = self.state.lock().await; *state = None; Self::delete_all_cloud_files(); + crate::cookie_bot::forget_enrolments(); } /// API call with 401 retry: if first attempt gets 401, refresh access token and retry once. @@ -1137,7 +1344,7 @@ impl CloudAuthManager { F: Fn(String) -> Fut + Send, Fut: std::future::Future> + Send, { - let access_token = Self::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; + let access_token = self.usable_access_token().await?; match make_request(access_token.clone()).await { Ok(result) => Ok(result), @@ -1151,9 +1358,7 @@ impl CloudAuthManager { return make_request(current_token).await; } - self.refresh_access_token().await?; - let new_token = - Self::load_access_token()?.ok_or_else(|| "Not logged in after refresh".to_string())?; + let new_token = self.refresh_after_rejection(&access_token).await?; log::info!("Token refreshed, retrying request..."); make_request(new_token).await } @@ -1446,13 +1651,11 @@ impl CloudAuthManager { // Proactively refresh the access token if it's expired or expiring soon. // This runs first so subsequent API calls use a fresh token. if let Ok(Some(token)) = Self::load_access_token() { - if Self::is_jwt_expiring_soon(&token) { - if let Err(e) = CLOUD_AUTH.refresh_access_token().await { + if Self::access_token_needs_refresh(&token) { + if let Err(e) = CLOUD_AUTH.ensure_fresh_access_token().await { log::warn!("Failed to refresh cloud access token: {e}"); // If the refresh token itself was rejected, session is irrecoverable if e.contains("(401") || e.contains("Unauthorized") { - log::warn!("Refresh token rejected — invalidating session"); - CLOUD_AUTH.invalidate_session().await; continue; } } @@ -1469,6 +1672,8 @@ impl CloudAuthManager { } } + let plan_before = CLOUD_AUTH.plan_marker().await; + // Refresh profile data periodically. A failure here leaves the cached // plan stale, which silently gates paid features, so it belongs at warn // rather than debug where the shipped log level hides it. @@ -1476,6 +1681,10 @@ impl CloudAuthManager { log::warn!("Failed to refresh cloud profile: {e}"); } + if CLOUD_AUTH.plan_marker().await != plan_before { + sync::resume_subscription(app_handle.clone()).await; + } + // Reconnect profile lock manager if needed if let Some(auth_state) = CLOUD_AUTH.get_user().await { if auth_state.user.effective_plan() != "free" @@ -2157,4 +2366,187 @@ mod tests { assert_eq!(spent_extra.recurring_limit_mb, 500); assert_eq!(spent_extra.extra_limit_mb, 0); } + + fn token_expiring_at(exp: i64) -> String { + use base64::{engine::general_purpose, Engine as _}; + let header = general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"RS256","typ":"JWT"}"#); + let payload = + general_purpose::URL_SAFE_NO_PAD.encode(serde_json::json!({ "exp": exp }).to_string()); + format!("{header}.{payload}.signature") + } + + #[test] + fn the_access_token_is_renewed_fifteen_minutes_before_it_expires() { + let now = 1_800_000_000; + let ten_minutes_left = token_expiring_at(now + 10 * 60); + let twenty_minutes_left = token_expiring_at(now + 20 * 60); + + assert!(CloudAuthManager::jwt_expires_within( + &ten_minutes_left, + ACCESS_TOKEN_REFRESH_MARGIN_SECS, + now + )); + assert!(!CloudAuthManager::jwt_expires_within( + &twenty_minutes_left, + ACCESS_TOKEN_REFRESH_MARGIN_SECS, + now + )); + assert!(!CloudAuthManager::jwt_expires_within( + &ten_minutes_left, + SYNC_TOKEN_REFRESH_MARGIN_SECS, + now + )); + assert!(CloudAuthManager::jwt_expires_within( + "not-a-jwt", + ACCESS_TOKEN_REFRESH_MARGIN_SECS, + now + )); + } + + #[test] + fn the_background_loop_renews_before_the_margin_runs_out() { + let source = include_str!("cloud_auth.rs"); + let body = source + .split("pub async fn start_sync_token_refresh_loop(") + .nth(1) + .expect("the refresh loop must exist"); + let period = body + .split("tokio::time::sleep(std::time::Duration::from_secs(") + .nth(1) + .and_then(|rest| rest.split(')').next()) + .and_then(|secs| secs.parse::().ok()) + .expect("the loop sleeps a fixed number of seconds"); + assert!(period < ACCESS_TOKEN_REFRESH_MARGIN_SECS); + assert!(body.contains("access_token_needs_refresh(&token)")); + } + + #[test] + fn a_plan_change_brings_a_stopped_sync_subscription_back() { + let source = include_str!("cloud_auth.rs"); + let body = source + .split("pub async fn start_sync_token_refresh_loop(") + .nth(1) + .expect("the refresh loop must exist"); + let before = body + .find("let plan_before = CLOUD_AUTH.plan_marker().await;") + .expect("the plan is read before the profile refresh"); + let fetch = body + .find("CLOUD_AUTH.fetch_profile().await") + .expect("the profile refresh"); + let resume = body + .find("sync::resume_subscription(app_handle.clone()).await;") + .expect("a changed plan resumes the subscription"); + assert!(before < fetch && fetch < resume); + } + + #[test] + fn retry_after_accepts_seconds_and_http_dates() { + let now = chrono::DateTime::parse_from_rfc3339("2026-09-27T10:00:00Z") + .unwrap() + .with_timezone(&Utc); + assert_eq!( + parse_retry_after("120", now), + Some(Duration::from_secs(120)) + ); + assert_eq!( + parse_retry_after("Sun, 27 Sep 2026 10:01:30 GMT", now), + Some(Duration::from_secs(90)) + ); + assert_eq!( + parse_retry_after("Sun, 27 Sep 2026 09:00:00 GMT", now), + Some(Duration::ZERO) + ); + assert_eq!(parse_retry_after("999999", now), Some(MAX_RETRY_AFTER)); + assert_eq!(parse_retry_after("soon", now), None); + } + + #[test] + fn a_missing_retry_after_falls_back_to_a_default_wait() { + let mut headers = reqwest::header::HeaderMap::new(); + assert_eq!(retry_after(&headers), DEFAULT_RETRY_AFTER); + headers.insert(reqwest::header::RETRY_AFTER, "7".parse().unwrap()); + assert_eq!(retry_after(&headers), Duration::from_secs(7)); + } + + #[test] + fn only_a_dead_session_reads_as_signed_out() { + assert!(matches!( + AuthorizedError::from_token_error("Token refresh failed (401 Unauthorized)".to_string()), + AuthorizedError::SignedOut + )); + assert!(matches!( + AuthorizedError::from_token_error("Not logged in".to_string()), + AuthorizedError::SignedOut + )); + assert!(matches!( + AuthorizedError::from_token_error("No refresh token stored".to_string()), + AuthorizedError::SignedOut + )); + assert!(matches!( + AuthorizedError::from_token_error("Failed to refresh token: connection reset".to_string()), + AuthorizedError::Transport(_) + )); + assert!(matches!( + AuthorizedError::from_token_error( + "Token refresh deferred for 30s (429 Too Many Requests)".to_string() + ), + AuthorizedError::Transport(_) + )); + } + + #[test] + fn a_refused_refresh_ends_the_session_and_a_throttled_one_waits() { + let source = include_str!("cloud_auth.rs"); + let body = source + .split("async fn refresh_while_locked(&self)") + .nth(1) + .expect("refresh_while_locked must exist"); + let body = &body[..body.find("\n }").unwrap_or(body.len())]; + let refused = body + .find("StatusCode::UNAUTHORIZED") + .expect("a 401 must be handled"); + let invalidate = body + .find("invalidate_session()") + .expect("a refused refresh must end the session"); + assert!(refused < invalidate); + assert!(body.contains("forget_refresh_token()")); + assert!(body.contains("defer_refresh(retry_after(response.headers()))")); + assert!(body.find("refresh_deferral()") < body.find(".send()")); + } + + #[test] + fn an_ended_session_stops_every_loop_that_sends_a_token() { + let source = include_str!("cloud_auth.rs"); + let body = source + .split("pub async fn invalidate_session(&self)") + .nth(1) + .expect("invalidate_session must exist"); + let body = &body[..body.find("\n }").unwrap_or(body.len())]; + for stop in [ + "crate::team_lock::PROFILE_LOCK.disconnect()", + "sync::stop_subscription()", + "crate::remote_session::stop_session_events()", + "crate::agent::stop_run_events()", + "crate::mcp_remote::stop(None)", + ] { + assert!(body.contains(stop), "invalidate_session must call {stop}"); + } + } + + #[test] + fn every_api_call_starts_from_a_token_that_is_not_about_to_expire() { + let source = include_str!("cloud_auth.rs"); + for owner in [ + "pub async fn api_call_with_retry<", + "pub(crate) async fn authorized_request<", + ] { + let body = source + .split(owner) + .nth(1) + .unwrap_or_else(|| panic!("{owner} must exist")); + let body = &body[..body.find("\n }").unwrap_or(body.len())]; + assert!(body.contains(".usable_access_token()"), "{owner}"); + assert!(body.contains("refresh_after_rejection("), "{owner}"); + } + } } diff --git a/src-tauri/src/cloud_errors.rs b/src-tauri/src/cloud_errors.rs index 9ab3e712..b51ceeec 100644 --- a/src-tauri/src/cloud_errors.rs +++ b/src-tauri/src/cloud_errors.rs @@ -120,6 +120,10 @@ pub fn transport_failure(message: &str) -> BackendFailure { } } +pub fn carries_code(body: &str) -> bool { + from_body(0, body).is_some() +} + fn code_for_status(status: u16, codes: FailureCodes) -> &'static str { match status { 400 | 422 => codes.bad_request, diff --git a/src-tauri/src/cookie_bot.rs b/src-tauri/src/cookie_bot.rs index 877d5302..1e010cc1 100644 --- a/src-tauri/src/cookie_bot.rs +++ b/src-tauri/src/cookie_bot.rs @@ -14,6 +14,7 @@ use crate::cloud_errors::{self, BackendFailure, FailureCodes}; use crate::profile::types::BrowserProfile; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::sync::OnceLock; use std::time::Duration; @@ -22,6 +23,10 @@ use std::time::Duration; /// at 02:00. pub const BOT_PLATFORMS: [&str; 3] = ["windows", "macos", "linux"]; +const MAX_PROFILE_NAME_UNITS: usize = 200; + +static ENROLLED_PROFILES: std::sync::Mutex>> = std::sync::Mutex::new(None); + const REQUEST_TIMEOUT: Duration = Duration::from_secs(20); const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); @@ -809,14 +814,18 @@ pub async fn list_schedules(scope: Option<&str>) -> Result Result, .await; match result { - Ok(envelope) => Ok(Some(envelope.schedule)), - Err(e) if e.code() == "COOKIE_BOT_NOT_ENROLLED" => Ok(None), + Ok(envelope) => { + record_enrolment(&mut enrolled_profiles(), profile_id, true); + Ok(Some(envelope.schedule)) + } + Err(e) if e.code() == "COOKIE_BOT_NOT_ENROLLED" => { + record_enrolment(&mut enrolled_profiles(), profile_id, false); + Ok(None) + } Err(e) => Err(e), } } @@ -868,14 +883,16 @@ pub async fn save_schedule( ); } - request( + let saved: CookieBotScheduleSaved = request( reqwest::Method::PUT, format!("{}/schedules/{}", base(), urlencoding::encode(profile_id)), Vec::new(), Some(body), SCHEDULE_CODES, ) - .await + .await?; + record_enrolment(&mut enrolled_profiles(), profile_id, true); + Ok(saved) } /// Re-declare the profile facts the server cannot observe for itself. @@ -885,24 +902,9 @@ pub async fn save_schedule( pub async fn update_profile_state( profile_id: &str, state: ProfileState, - timezone: Option<&str>, + identity: &ProfileIdentity, ) -> Result { - let mut body = serde_json::Map::new(); - body.insert("sync_enabled".to_string(), state.sync_enabled.into()); - body.insert("encrypted_sync".to_string(), state.encrypted_sync.into()); - body.insert("has_proxy".to_string(), state.has_proxy.into()); - body.insert( - "proxy_remote_reachable".to_string(), - state.proxy_remote_reachable.into(), - ); - body.insert( - "touch_fingerprint".to_string(), - state.touch_fingerprint.into(), - ); - body.insert("sticky_exit".to_string(), state.sticky_exit.into()); - if let Some(zone) = timezone { - body.insert("timezone".to_string(), zone.into()); - } + let body = profile_state_body(state, identity); let envelope: ScheduleEnvelope = request( reqwest::Method::POST, @@ -912,7 +914,7 @@ pub async fn update_profile_state( urlencoding::encode(profile_id) ), Vec::new(), - Some(serde_json::Value::Object(body)), + Some(body), SCHEDULE_CODES, ) .await?; @@ -934,35 +936,189 @@ pub async fn update_profile_state( /// which is the normal case, and the next edit re-declares the same facts. pub fn report_profile_state(profile: &BrowserProfile) { let profile_id = profile.id.to_string(); + if !should_report(&enrolled_profiles(), &profile_id) { + return; + } let state = profile_state(profile); + let identity = profile_identity(profile); tauri::async_runtime::spawn(async move { - if !crate::cloud_auth::CLOUD_AUTH.is_logged_in().await { - return; - } - match update_profile_state(&profile_id, state, None).await { + send_profile_state(profile_id, state, identity).await; + }); +} + +async fn send_profile_state(profile_id: String, state: ProfileState, identity: ProfileIdentity) { + if crate::cloud_auth::CLOUD_AUTH.is_logged_in().await { + match update_profile_state(&profile_id, state, &identity).await { Ok(_) => { + record_enrolment(&mut enrolled_profiles(), &profile_id, true); log::debug!("Re-declared cookie-bot profile state for {profile_id}"); } // Not enrolled is the common answer and not worth a log line at warn. - Err(e) if e.code() == "COOKIE_BOT_NOT_ENROLLED" => {} + Err(e) if e.code() == "COOKIE_BOT_NOT_ENROLLED" => { + record_enrolment(&mut enrolled_profiles(), &profile_id, false); + } Err(e) => { log::warn!("Could not re-declare cookie-bot profile state for {profile_id}: {e}"); } } - }); + } +} + +pub fn report_profiles_using_proxy(proxy_id: &str) { + report_matching(|profile| profile.proxy_id.as_deref() == Some(proxy_id)); +} + +pub fn report_profiles_using_vpn(vpn_id: &str) { + report_matching(|profile| profile.vpn_id.as_deref() == Some(vpn_id)); +} + +fn report_matching(matches: impl Fn(&BrowserProfile) -> bool) { + let Ok(profiles) = crate::profile::ProfileManager::instance().list_profiles() else { + return; + }; + for profile in profiles.iter().filter(|profile| matches(profile)) { + report_profile_state(profile); + } +} + +pub async fn report_enrolled_profiles() { + let auth = &crate::cloud_auth::CLOUD_AUTH; + if !auth.is_logged_in().await || !auth.can_use_cookie_bot().await { + return; + } + let list = match list_schedules(None).await { + Ok(list) => list, + Err(e) => { + log::debug!("Could not list cookie-bot schedules at startup: {e}"); + return; + } + }; + let Ok(profiles) = crate::profile::ProfileManager::instance().list_profiles() else { + return; + }; + for profile in enrolled_local_profiles(&list, profiles) { + send_profile_state( + profile.id.to_string(), + profile_state(&profile), + profile_identity(&profile), + ) + .await; + } +} + +fn enrolled_local_profiles( + list: &CookieBotScheduleList, + profiles: Vec, +) -> Vec { + let enrolled: HashSet<&str> = list + .schedules + .iter() + .map(|schedule| schedule.profile_id.as_str()) + .collect(); + profiles + .into_iter() + .filter(|profile| enrolled.contains(profile.id.to_string().as_str())) + .collect() +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ProfileIdentity { + pub timezone: Option, + pub platform: Option, + pub profile_name: Option, +} + +pub fn profile_identity(profile: &BrowserProfile) -> ProfileIdentity { + let name = profile.name.trim(); + ProfileIdentity { + timezone: crate::fingerprint_consistency::fingerprint_timezone(profile) + .map(|zone| zone.trim().to_string()) + .filter(|zone| !zone.is_empty()), + platform: profile + .resolved_os() + .filter(|os| BOT_PLATFORMS.contains(os)) + .map(str::to_string), + profile_name: (!name.is_empty() && name.encode_utf16().count() <= MAX_PROFILE_NAME_UNITS) + .then(|| name.to_string()), + } +} + +fn profile_state_body(state: ProfileState, identity: &ProfileIdentity) -> serde_json::Value { + let mut body = serde_json::Map::new(); + body.insert("sync_enabled".to_string(), state.sync_enabled.into()); + body.insert("encrypted_sync".to_string(), state.encrypted_sync.into()); + body.insert("has_proxy".to_string(), state.has_proxy.into()); + body.insert( + "proxy_remote_reachable".to_string(), + state.proxy_remote_reachable.into(), + ); + body.insert( + "touch_fingerprint".to_string(), + state.touch_fingerprint.into(), + ); + body.insert("sticky_exit".to_string(), state.sticky_exit.into()); + if let Some(zone) = &identity.timezone { + body.insert("timezone".to_string(), zone.as_str().into()); + } + if let Some(platform) = &identity.platform { + body.insert("platform".to_string(), platform.as_str().into()); + } + if let Some(name) = &identity.profile_name { + body.insert("profile_name".to_string(), name.as_str().into()); + } + serde_json::Value::Object(body) +} + +pub fn forget_enrolments() { + *enrolled_profiles() = None; +} + +fn enrolled_profiles() -> std::sync::MutexGuard<'static, Option>> { + ENROLLED_PROFILES + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn should_report(enrolled: &Option>, profile_id: &str) -> bool { + enrolled.as_ref().is_none_or(|ids| ids.contains(profile_id)) +} + +fn record_enrolment(enrolled: &mut Option>, profile_id: &str, is_enrolled: bool) { + if let Some(ids) = enrolled.as_mut() { + if is_enrolled { + ids.insert(profile_id.to_string()); + } else { + ids.remove(profile_id); + } + } +} + +fn record_own_schedules(enrolled: &mut Option>, list: &CookieBotScheduleList) { + if list.scope.as_deref() == Some("team") { + return; + } + *enrolled = Some( + list + .schedules + .iter() + .map(|schedule| schedule.profile_id.clone()) + .collect(), + ); } /// Turn the bot off for this profile. Safe to repeat: deleting an enrolment /// that is already gone succeeds with `deleted: false`. pub async fn delete_schedule(profile_id: &str) -> Result { - request( + let deleted: CookieBotScheduleDeleted = request( reqwest::Method::DELETE, format!("{}/schedules/{}", base(), urlencoding::encode(profile_id)), Vec::new(), None, SCHEDULE_CODES, ) - .await + .await?; + record_enrolment(&mut enrolled_profiles(), profile_id, false); + Ok(deleted) } /// Ask, without writing anything, who else already warms this profile. @@ -1388,6 +1544,166 @@ mod tests { } } + fn profile_in_zone(zone: &str) -> BrowserProfile { + let mut profile = eligible_profile(); + profile.wayfern_config = Some(crate::wayfern_manager::WayfernConfig { + fingerprint: Some(serde_json::json!({ "timezone": zone, "language": "de-DE" }).to_string()), + ..Default::default() + }); + profile + } + + #[test] + fn a_report_names_the_zone_the_platform_and_the_profile() { + let identity = profile_identity(&profile_in_zone("Europe/Berlin")); + assert_eq!( + identity, + ProfileIdentity { + timezone: Some("Europe/Berlin".to_string()), + platform: Some("macos".to_string()), + profile_name: Some("warm me".to_string()), + } + ); + + let body = profile_state_body(profile_state(&eligible_profile()), &identity); + assert_eq!(body["timezone"], "Europe/Berlin"); + assert_eq!(body["platform"], "macos"); + assert_eq!(body["profile_name"], "warm me"); + assert_eq!(body["has_proxy"], true); + assert_eq!(body["sync_enabled"], true); + } + + #[test] + fn a_report_leaves_out_what_the_server_would_refuse() { + let mut profile = eligible_profile(); + profile.host_os = Some("android".to_string()); + profile.name = "n".repeat(MAX_PROFILE_NAME_UNITS + 1); + let identity = profile_identity(&profile); + assert_eq!(identity, ProfileIdentity::default()); + + let body = profile_state_body(profile_state(&profile), &identity); + for key in ["timezone", "platform", "profile_name"] { + assert!(body.get(key).is_none(), "{key} must be omitted"); + } + assert!(body.get("sync_enabled").is_some()); + } + + #[test] + fn a_profile_name_is_measured_the_way_the_server_measures_it() { + let mut profile = eligible_profile(); + profile.name = "😀".repeat(MAX_PROFILE_NAME_UNITS / 2); + assert!(profile_identity(&profile).profile_name.is_some()); + profile.name = "😀".repeat(MAX_PROFILE_NAME_UNITS / 2 + 1); + assert!(profile_identity(&profile).profile_name.is_none()); + profile.name = " ".to_string(); + assert!(profile_identity(&profile).profile_name.is_none()); + } + + fn schedule_list(scope: Option<&str>, ids: &[&str]) -> CookieBotScheduleList { + let schedules = ids + .iter() + .map(|id| { + serde_json::from_value(serde_json::json!({ + "profile_id": id, "profile_name": "p", "platform": "macos", "enabled": true, + "run_at_minute": 0, "days_mask": 1, "timezone": "UTC", "preset": "light", + "max_minutes": 20 + })) + .expect("a minimal schedule") + }) + .collect(); + CookieBotScheduleList { + schedules, + team_id: None, + scope: scope.map(str::to_string), + } + } + + #[test] + fn only_enrolled_profiles_are_reported_once_the_enrolments_are_known() { + let mut enrolled = None; + assert!(should_report(&enrolled, "any")); + + record_own_schedules(&mut enrolled, &schedule_list(Some("mine"), &["p1"])); + assert!(should_report(&enrolled, "p1")); + assert!(!should_report(&enrolled, "p2")); + + record_enrolment(&mut enrolled, "p2", true); + assert!(should_report(&enrolled, "p2")); + record_enrolment(&mut enrolled, "p1", false); + assert!(!should_report(&enrolled, "p1")); + } + + #[test] + fn a_team_listing_does_not_replace_the_callers_own_enrolments() { + let mut enrolled = None; + record_own_schedules(&mut enrolled, &schedule_list(Some("mine"), &["mine-1"])); + record_own_schedules( + &mut enrolled, + &schedule_list(Some("team"), &["mine-1", "teammate-1"]), + ); + assert!(should_report(&enrolled, "mine-1")); + assert!(!should_report(&enrolled, "teammate-1")); + } + + #[test] + fn an_unknown_enrolment_list_is_not_filled_in_by_single_answers() { + let mut enrolled = None; + record_enrolment(&mut enrolled, "p1", false); + assert!(enrolled.is_none()); + assert!(should_report(&enrolled, "p1")); + } + + #[test] + fn startup_reports_every_enrolled_profile_this_machine_has() { + let mut here = eligible_profile(); + here.id = uuid::Uuid::from_u128(1); + let mut elsewhere = eligible_profile(); + elsewhere.id = uuid::Uuid::from_u128(2); + let list = schedule_list(Some("mine"), &[&here.id.to_string(), "not-on-this-machine"]); + let picked = enrolled_local_profiles(&list, vec![here.clone(), elsewhere]); + assert_eq!(picked.len(), 1); + assert_eq!(picked[0].id, here.id); + } + + #[test] + fn uploads_and_startup_re_declare_the_profile_state() { + let engine = include_str!("sync/engine.rs"); + let lib = include_str!("lib.rs"); + let upload_proxy = engine + .split("async fn upload_proxy(") + .nth(1) + .expect("upload_proxy must exist"); + assert!(upload_proxy[..upload_proxy.find("\n }").unwrap()] + .contains("crate::cookie_bot::report_profiles_using_proxy(&proxy.id);")); + let upload_vpn = engine + .split("async fn upload_vpn(") + .nth(1) + .expect("upload_vpn must exist"); + assert!(upload_vpn[..upload_vpn.find("\n }").unwrap()] + .contains("crate::cookie_bot::report_profiles_using_vpn(&vpn.id);")); + let profile_sync = engine + .split("async fn sync_profile_inner(") + .nth(1) + .expect("sync_profile_inner must exist"); + let profile_sync = &profile_sync[..profile_sync.find("\n }").unwrap()]; + let uploaded = profile_sync + .find(".upload_manifest(") + .expect("the manifest upload"); + let reported = profile_sync + .find("crate::cookie_bot::report_profile_state(&updated_profile);") + .expect("the report after a finished upload"); + assert!(uploaded < reported); + assert!(lib.contains("tauri::async_runtime::spawn(cookie_bot::report_enrolled_profiles());")); + let auth = include_str!("cloud_auth.rs"); + let clear = auth + .split("async fn clear_auth(&self)") + .nth(1) + .expect("clear_auth must exist"); + assert!( + clear[..clear.find("\n }").unwrap()].contains("crate::cookie_bot::forget_enrolments();") + ); + } + #[test] fn the_profile_facts_the_server_requires_are_derived_from_the_profile() { let state = profile_state(&eligible_profile()); diff --git a/src-tauri/src/fingerprint_consistency.rs b/src-tauri/src/fingerprint_consistency.rs index 154c92b6..d16c4ad4 100644 --- a/src-tauri/src/fingerprint_consistency.rs +++ b/src-tauri/src/fingerprint_consistency.rs @@ -250,6 +250,10 @@ fn fingerprint_locale(profile: &BrowserProfile) -> (Option, Option Option { + fingerprint_locale(profile).0 +} + /// A mutex whose poison is not fatal. /// /// A panic anywhere under this lock used to brick the check process-wide. diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index fe6cf2a8..966daa4c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3356,6 +3356,8 @@ pub fn run_with_builder( // with its work still in cloud storage. Signing in is the first moment // that pull can succeed, so it is where it is retried. remote_handoff::resume_pending_pulls(&app_handle_cloud); + + tauri::async_runtime::spawn(cookie_bot::report_enrolled_profiles()); } cloud_auth::CloudAuthManager::start_sync_token_refresh_loop(app_handle_cloud).await; }); @@ -3656,6 +3658,9 @@ pub fn run_with_builder( // one account holds one bridge at a time, so an instance that exits // without hanging up delays the account's next machine. mcp_remote::stop(None); + tauri::async_runtime::block_on( + team_lock::PROFILE_LOCK.release_all_held_within(team_lock::SHUTDOWN_RELEASE_TIMEOUT), + ); } #[cfg(target_os = "macos")] diff --git a/src-tauri/src/proxy_runner.rs b/src-tauri/src/proxy_runner.rs index 760f0355..df1396f6 100644 --- a/src-tauri/src/proxy_runner.rs +++ b/src-tauri/src/proxy_runner.rs @@ -373,11 +373,6 @@ pub async fn start_proxy_process_with_profile( processes.insert(id.clone(), pid); } - // Update config with PID - let mut config_with_pid = config.clone(); - config_with_pid.pid = Some(pid); - save_proxy_config(&config_with_pid)?; - // Don't wait for the child - it's detached drop(child); } @@ -447,11 +442,6 @@ pub async fn start_proxy_process_with_profile( processes.insert(id.clone(), pid); } - // Update config with PID - let mut config_with_pid = config.clone(); - config_with_pid.pid = Some(pid); - save_proxy_config(&config_with_pid)?; - drop(child); } @@ -516,43 +506,46 @@ pub async fn start_proxy_process_with_profile( } pub async fn stop_proxy_process(id: &str) -> Result> { - let config = get_proxy_config(id); + let pid = PROXY_PROCESSES + .lock() + .unwrap() + .get(id) + .copied() + .or_else(|| get_proxy_config(id).and_then(|config| config.pid)); - if let Some(config) = config { - if let Some(pid) = config.pid { - // Kill the process - #[cfg(unix)] - { - use std::process::Command; - let _ = Command::new("kill") - .arg("-TERM") - .arg(pid.to_string()) - .output(); - } - #[cfg(windows)] - { - use std::os::windows::process::CommandExt; - use std::process::Command; - const CREATE_NO_WINDOW: u32 = 0x08000000; - let _ = Command::new("taskkill") - .args(["/F", "/PID", &pid.to_string()]) - .creation_flags(CREATE_NO_WINDOW) - .output(); - } - - // Wait a bit for the process to exit - tokio::time::sleep(tokio::time::Duration::from_millis(500)).await; - - // Remove from tracking - { - let mut processes = PROXY_PROCESSES.lock().unwrap(); - processes.remove(id); - } - - // Delete the config file - delete_proxy_config(id); - return Ok(true); + if let Some(pid) = pid { + // Kill the process + #[cfg(unix)] + { + use std::process::Command; + let _ = Command::new("kill") + .arg("-TERM") + .arg(pid.to_string()) + .output(); } + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + use std::process::Command; + const CREATE_NO_WINDOW: u32 = 0x08000000; + let _ = Command::new("taskkill") + .args(["/F", "/PID", &pid.to_string()]) + .creation_flags(CREATE_NO_WINDOW) + .output(); + } + + // Wait a bit for the process to exit + tokio::time::sleep(tokio::time::Duration::from_millis(500)).await; + + // Remove from tracking + { + let mut processes = PROXY_PROCESSES.lock().unwrap(); + processes.remove(id); + } + + // Delete the config file + delete_proxy_config(id); + return Ok(true); } Ok(false) diff --git a/src-tauri/src/proxy_server.rs b/src-tauri/src/proxy_server.rs index 0e610a1d..20e9bacc 100644 --- a/src-tauri/src/proxy_server.rs +++ b/src-tauri/src/proxy_server.rs @@ -1654,6 +1654,7 @@ pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box; static STORE: RwLock> = RwLock::new(None); +static PULL_WORKERS: Mutex>>> = Mutex::new(None); + +fn with_workers(f: impl FnOnce(&mut HashMap>) -> T) -> T { + let mut guard = PULL_WORKERS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + f(guard.get_or_insert_with(HashMap::new)) +} + +struct PullWorker { + profile_id: String, + wake: Arc, +} + +impl PullWorker { + fn claim(profile_id: String) -> Option { + with_workers(|workers| { + if let Some(wake) = workers.get(&profile_id) { + wake.notify_one(); + return None; + } + let wake = Arc::new(Notify::new()); + workers.insert(profile_id.clone(), wake.clone()); + Some(Self { profile_id, wake }) + }) + } + + fn pending_session(&self) -> Option { + // Retire under the same lock as claim, so a new request cannot wake a + // worker that has already decided to exit. + with_workers(|workers| { + let pending = pending_session_for(&self.profile_id); + if pending.is_none() { + workers.remove(&self.profile_id); + } + pending + }) + } +} + +impl Drop for PullWorker { + fn drop(&mut self) { + with_workers(|workers| { + if workers + .get(&self.profile_id) + .is_some_and(|wake| Arc::ptr_eq(wake, &self.wake)) + { + workers.remove(&self.profile_id); + } + }); + } +} + +fn pull_retry_delay(attempt: u32) -> Duration { + PULL_RETRY_BASE + .saturating_mul(1u32 << attempt.min(16)) + .min(PULL_RETRY_MAX) +} fn store_path() -> std::path::PathBuf { crate::app_dirs::settings_dir().join("remote_handoff.json") @@ -168,6 +213,14 @@ pub fn state_for(profile_id: &str) -> Option { with_store(|store| store.get(profile_id).map(|entry| entry.state)) } +pub(crate) fn session_for(profile_id: &str) -> Option<(HandoffState, String)> { + with_store(|store| { + store + .get(profile_id) + .map(|entry| (entry.state, entry.session_id.clone())) + }) +} + /// The session currently holding this profile remotely, if any. /// /// Answers for a `provisioning` session too, which the drivable-session index @@ -265,9 +318,23 @@ pub fn note_ended(profile_id: &str, session_id: &str) -> bool { transitioned } -/// Drop the gate. Called only after a pull has actually completed. -pub fn clear(profile_id: &str) { - mutate(|store| store.remove(profile_id).is_some()); +pub(crate) fn pending_session_for(profile_id: &str) -> Option { + session_for(profile_id) + .filter(|(state, _)| *state == HandoffState::PendingSync) + .map(|(_, session_id)| session_id) +} + +/// A completed pull must not clear a newer session's gate. +pub(crate) fn clear_pending(profile_id: &str, session_id: &str) { + mutate(|store| { + let matches = store.get(profile_id).is_some_and(|entry| { + entry.state == HandoffState::PendingSync && entry.session_id == session_id + }); + if matches { + store.remove(profile_id); + } + matches + }); } /// Bring stored `Running` entries back in line with what the backend reports. @@ -322,11 +389,7 @@ pub fn ensure_local_launch_allowed(profile_id: &str) -> Result<(), String> { /// Restart the pull for every profile still waiting on one. /// -/// A pull can fail for as long as the machine is offline, and its retries are -/// bounded, so without this a profile could stay blocked from launching until -/// the user found the manual sync button. Called whenever the app has a cloud -/// session again, which is exactly when a previously impossible pull becomes -/// possible. +/// Existing workers wake at once when the connection or sync settings recover. pub fn resume_pending_pulls(app_handle: &tauri::AppHandle) { let pending: Vec = with_store(|store| { store @@ -336,7 +399,6 @@ pub fn resume_pending_pulls(app_handle: &tauri::AppHandle) { .collect() }); for profile_id in pending { - log::info!("Resuming the post-session pull for profile {profile_id}"); schedule_pull(app_handle.clone(), profile_id); } } @@ -364,54 +426,57 @@ fn profile_exists_locally(profile_id: &str) -> bool { /// whole attempt, so there is no window in which the user can open the stale /// copy while this is in flight. pub fn schedule_pull(app_handle: tauri::AppHandle, profile_id: String) { + let Some(worker) = PullWorker::claim(profile_id.clone()) else { + return; + }; tauri::async_runtime::spawn(async move { - for attempt in 0..PULL_ATTEMPTS { - if state_for(&profile_id) != Some(HandoffState::PendingSync) { - // A new session started, or another pull got there first. - return; - } - if attempt > 0 { - let delay = PULL_RETRY_BASE - .saturating_mul(1u32 << (attempt - 1).min(16)) - .min(PULL_RETRY_MAX); - tokio::time::sleep(delay).await; - } + run_pull_worker( + worker, + || crate::sync::pull_profile_after_remote_session(&app_handle, &profile_id), + || profile_exists_locally(&profile_id), + ) + .await; + }); +} - match crate::sync::pull_profile_after_remote_session(&app_handle, &profile_id).await { - Ok(outcome) if outcome.is_completed() => { - log::info!("Pulled remote session work for profile {profile_id}"); - clear(&profile_id); - return; - } - Ok(crate::sync::ProfileSyncOutcome::Skipped(reason)) => { - log::info!("Post-session pull for profile {profile_id} did nothing ({reason}); retrying"); - } - Ok(_) => unreachable!("is_completed covers every completed outcome"), - Err(e) => { - // A profile that was created and run entirely on a remote host may - // not exist on this device at all: nothing to pull, nothing to gate. - // The pull would fail with "not found" on every attempt and the entry - // would sit in the handoff store for ever. Clear it — the gate only - // protects a LOCAL profile from being opened over unsynced remote - // work, and there is no local profile here. - if !profile_exists_locally(&profile_id) { - log::info!( - "Clearing the post-session gate for profile {profile_id}: it has no local copy \ - (created and run remotely), so there is nothing to pull or protect" - ); - clear(&profile_id); - return; - } - log::warn!("Post-session pull for profile {profile_id} failed: {e}"); +async fn run_pull_worker(worker: PullWorker, mut pull: F, exists: E) +where + F: FnMut() -> Fut, + Fut: std::future::Future>, + E: Fn() -> bool, +{ + let mut attempt = 0u32; + while let Some(session_id) = worker.pending_session() { + let profile_id = &worker.profile_id; + match pull().await { + Ok(outcome) if outcome.is_completed() => { + log::info!("Pulled remote session work for profile {profile_id}"); + clear_pending(profile_id, &session_id); + attempt = 0; + continue; + } + Ok(crate::sync::ProfileSyncOutcome::Skipped(reason)) => { + log::debug!("Post-session pull for profile {profile_id} is waiting: {reason}"); + } + Ok(_) => unreachable!("is_completed covers every completed outcome"), + Err(e) => { + if !exists() { + clear_pending(profile_id, &session_id); + continue; } + log::warn!("Post-session pull for profile {profile_id} failed: {e}"); } } - - log::warn!( - "Could not pull remote session work for profile {profile_id} yet; it stays blocked from \ - launching locally until the pull succeeds" - ); - }); + if worker.pending_session().is_none() { + return; + } + tokio::select! { + _ = tokio::time::sleep(pull_retry_delay(attempt)) => { + attempt = attempt.saturating_add(1); + } + _ = worker.wake.notified() => attempt = 0, + } + } } /// Serialises every test that can reach [`STORE`], wherever it lives. @@ -442,6 +507,102 @@ mod tests { use super::*; use std::collections::HashSet; + #[test] + fn retries_continue_after_an_outage_longer_than_five_attempts() { + let _iso = isolated(); + note_running("p1", "s1"); + note_ended("p1", "s1"); + let worker = PullWorker::claim("p1".into()).unwrap(); + let wake = worker.wake.clone(); + let mut attempts = 0; + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(run_pull_worker( + worker, + || { + attempts += 1; + wake.notify_one(); + std::future::ready(if attempts <= 7 { + assert_eq!(state_for("p1"), Some(HandoffState::PendingSync)); + if attempts % 2 == 0 { + Ok(crate::sync::ProfileSyncOutcome::Skipped("locked elsewhere")) + } else { + Err("network unavailable".to_string()) + } + } else { + Ok(crate::sync::ProfileSyncOutcome::Completed) + }) + }, + || true, + )); + assert_eq!(attempts, 8); + assert!(ensure_local_launch_allowed("p1").is_ok()); + assert!(PullWorker::claim("p1".into()).is_some()); + } + + #[test] + fn duplicate_requests_wake_one_worker_and_a_retired_worker_cannot_remove_its_replacement() { + let _iso = isolated(); + note_running("p1", "s1"); + note_ended("p1", "s1"); + let worker = PullWorker::claim("p1".into()).unwrap(); + assert!(PullWorker::claim("p1".into()).is_none()); + assert!(futures_util::FutureExt::now_or_never(worker.wake.notified()).is_some()); + clear_pending("p1", "s1"); + assert!(worker.pending_session().is_none()); + note_running("p1", "s2"); + note_ended("p1", "s2"); + let replacement = PullWorker::claim("p1".into()).unwrap(); + drop(worker); + assert!(PullWorker::claim("p1".into()).is_none()); + assert_eq!(replacement.pending_session().as_deref(), Some("s2")); + } + + #[test] + fn completion_of_an_old_pull_keeps_the_new_session_blocked() { + let _iso = isolated(); + note_running("p1", "s1"); + note_ended("p1", "s1"); + note_running("p1", "s2"); + clear_pending("p1", "s1"); + assert_eq!(state_for("p1"), Some(HandoffState::Running)); + note_ended("p1", "s2"); + clear_pending("p1", "s1"); + assert_eq!(pending_session_for("p1").as_deref(), Some("s2")); + } + + #[test] + fn a_new_running_session_stops_the_waiting_pull() { + let _iso = isolated(); + note_running("p1", "s1"); + note_ended("p1", "s1"); + let worker = PullWorker::claim("p1".into()).unwrap(); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(run_pull_worker( + worker, + || { + note_running("p1", "s2"); + std::future::ready(Err("network unavailable".to_string())) + }, + || true, + )); + assert_eq!(state_for("p1"), Some(HandoffState::Running)); + assert!(PullWorker::claim("p1".into()).is_some()); + } + + #[test] + fn pull_retries_start_quickly_and_remain_bounded() { + assert_eq!(pull_retry_delay(0), Duration::from_secs(1)); + assert_eq!(pull_retry_delay(1), Duration::from_secs(2)); + assert_eq!(pull_retry_delay(2), Duration::from_secs(4)); + assert_eq!(pull_retry_delay(u32::MAX), Duration::from_secs(15)); + } + /// Point the store at a scratch directory and start it empty. /// /// Everything returned must outlive the test body: dropping the guard @@ -485,7 +646,7 @@ mod tests { let err = ensure_local_launch_allowed("p1").expect_err("pending work must block a launch"); assert!(err.contains("PROFILE_REMOTE_SYNC_PENDING")); - clear("p1"); + clear_pending("p1", "s1"); assert!(ensure_local_launch_allowed("p1").is_ok()); } @@ -528,7 +689,7 @@ mod tests { let _iso = isolated(); note_running("p1", "s1"); note_ended("p1", "s1"); - clear("p1"); + clear_pending("p1", "s1"); assert!(!note_ended("p1", "s1")); assert!(ensure_local_launch_allowed("p1").is_ok()); } diff --git a/src-tauri/src/remote_session.rs b/src-tauri/src/remote_session.rs index 1143ff3b..5fce5ea1 100644 --- a/src-tauri/src/remote_session.rs +++ b/src-tauri/src/remote_session.rs @@ -101,7 +101,8 @@ pub fn classify_backend_status(status: u16, body: &str) -> RemoteSessionError { 503 => RemoteSessionError::NoCapacity(message), 409 => RemoteSessionError::Conflict(message), 401..=403 => RemoteSessionError::NotAuthorised(message), - _ => RemoteSessionError::Other(message), + _ if cloud_errors::carries_code(body) => RemoteSessionError::Other(message), + _ => RemoteSessionError::Other(format!("({status}) {message}")), } } @@ -674,7 +675,10 @@ fn index_session(app: Option<&AppHandle>, session: &RemoteSessionState) { // the only frames that exist. It is deliberately keyed off `is_terminal` // rather than `is_drivable`: a provisioning host already owns the profile. if is_terminal(session) { - if crate::remote_handoff::note_ended(&profile_id, &session.session_id) { + crate::remote_handoff::note_ended(&profile_id, &session.session_id); + if crate::remote_handoff::pending_session_for(&profile_id).as_deref() + == Some(session.session_id.as_str()) + { if let Some(app) = app { crate::remote_handoff::schedule_pull(app.clone(), profile_id.clone()); } @@ -717,6 +721,9 @@ fn reindex(app: Option<&AppHandle>, sessions: &[RemoteSessionState]) { crate::remote_handoff::schedule_pull(app.clone(), profile_id); } } + if let Some(app) = app { + crate::remote_handoff::resume_pending_pulls(app); + } let next: HashMap = sessions .iter() @@ -778,10 +785,12 @@ pub const EVENT_STREAM_STATUS: &str = "remote-session-stream"; /// well above the heartbeat interval so a slow network cannot cause a churn of /// reconnects. const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(90); +const STREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); /// First reconnect delay. Doubles per failure. const RECONNECT_BASE: Duration = Duration::from_secs(1); /// Ceiling on the reconnect delay. const RECONNECT_MAX: Duration = Duration::from_secs(60); +const RECOVERY_RECONNECT_MAX: Duration = Duration::from_secs(10); /// Where the backoff restarts after an auth failure. Being signed out or /// unentitled is not something a fast retry fixes, and hammering an endpoint /// that will keep saying no is how a background task becomes a battery bug. @@ -792,6 +801,7 @@ const SHUTDOWN_POLL: Duration = Duration::from_millis(250); static STREAM_RUNNING: AtomicBool = AtomicBool::new(false); static STREAM_TASK: Mutex>> = Mutex::new(None); +static STREAM_WAKE: tokio::sync::Notify = tokio::sync::Notify::const_new(); /// One frame off the wire. #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -953,10 +963,11 @@ pub fn reconnect_delay(attempt: u32) -> Duration { .min(RECONNECT_MAX) } -/// Start receiving session transitions. Idempotent: a second call while the -/// stream is up is a no-op rather than a second socket. +/// Start or wake the one subscriber that receives session transitions. pub fn start_session_events(app: AppHandle) { + crate::remote_handoff::resume_pending_pulls(&app); if STREAM_RUNNING.swap(true, Ordering::SeqCst) { + STREAM_WAKE.notify_one(); return; } let handle = tauri::async_runtime::spawn(async move { @@ -998,6 +1009,7 @@ async fn run_session_events(app: AppHandle) { let mut last_event_id: Option = None; while STREAM_RUNNING.load(Ordering::SeqCst) { + let mut auth_failure = false; match connect_session_events(last_event_id.as_deref()).await { Ok(response) => { attempt = 0; @@ -1016,6 +1028,7 @@ async fn run_session_events(app: AppHandle) { Err(err) => { let reason = err.to_string(); if matches!(err, RemoteSessionError::NotAuthorised(_)) { + auth_failure = true; attempt = attempt.max(AUTH_BACKOFF_ATTEMPT); } log::warn!("Remote session stream could not connect: {reason}"); @@ -1026,14 +1039,30 @@ async fn run_session_events(app: AppHandle) { if !STREAM_RUNNING.load(Ordering::SeqCst) { break; } - let delay = jittered(reconnect_delay(attempt)); + let delay = jittered(session_reconnect_delay( + attempt, + !crate::remote_handoff::states().is_empty(), + auth_failure, + )); attempt = attempt.saturating_add(1); - sleep_unless_stopped(delay).await; + tokio::select! { + _ = sleep_unless_stopped(delay) => {}, + _ = STREAM_WAKE.notified() => attempt = 0, + } } log::info!("Remote session stream stopped"); } +fn session_reconnect_delay(attempt: u32, has_remote_work: bool, auth_failure: bool) -> Duration { + let delay = reconnect_delay(attempt); + if has_remote_work && !auth_failure { + delay.min(RECOVERY_RECONNECT_MAX) + } else { + delay + } +} + /// Spread reconnects so every desktop that lost the same backend does not come /// back in the same millisecond. /// @@ -1095,9 +1124,9 @@ async fn connect_session_events( request = request.header("Last-Event-ID", id); } - let response = request - .send() + let response = tokio::time::timeout(STREAM_CONNECT_TIMEOUT, request.send()) .await + .map_err(|_| "reach backend: session stream response timed out".to_string())? .map_err(|e| format!("reach backend: {e}"))?; let status = response.status().as_u16(); @@ -1127,7 +1156,10 @@ async fn consume_session_events( return Ok(()); } - let next = tokio::time::timeout(STREAM_IDLE_TIMEOUT, stream.next()).await; + let next = tokio::select! { + next = tokio::time::timeout(STREAM_IDLE_TIMEOUT, stream.next()) => next, + _ = STREAM_WAKE.notified() => return Ok(()), + }; let chunk = match next { // No heartbeat. The socket is gone even though nothing errored, which // is what a machine returning from sleep sees. @@ -1224,6 +1256,38 @@ mod tests { )); } + #[test] + fn a_failure_without_a_code_keeps_its_http_status() { + let code = |status: u16, body: &str| { + let json = classify_backend_status(status, body).to_error_json(); + serde_json::from_str::(&json).unwrap()["code"] + .as_str() + .unwrap() + .to_string() + }; + assert_eq!( + code(404, "No such remote session"), + "REMOTE_SESSION_NOT_FOUND" + ); + assert_eq!(code(400, "Bad Request"), "REMOTE_SESSION_REFUSED"); + assert_eq!(code(429, "Too Many Requests"), cloud_errors::RATE_LIMITED); + assert_eq!(code(500, "boom"), cloud_errors::UNAVAILABLE); + assert_eq!(code(502, ""), cloud_errors::UNAVAILABLE); + assert_eq!( + code(404, r#"{"message":"Not Found","statusCode":404}"#), + "REMOTE_SESSION_NOT_FOUND" + ); + } + + #[test] + fn a_failure_with_a_code_keeps_the_body_untouched() { + let body = r#"{"code":"REMOTE_SESSION_NOT_FOUND"}"#; + let RemoteSessionError::Other(kept) = classify_backend_status(404, body) else { + panic!("a 404 is not one of the typed kinds"); + }; + assert_eq!(kept, body); + } + #[test] fn conflict_is_surfaced_so_the_user_learns_the_profile_is_open() { assert!(matches!( @@ -1610,6 +1674,20 @@ mod tests { assert_eq!(reconnect_delay(u32::MAX), RECONNECT_MAX); } + #[test] + fn unfinished_remote_work_reconnects_quickly_without_retrying_auth_failures_faster() { + assert_eq!( + session_reconnect_delay(0, true, false), + Duration::from_secs(1) + ); + assert_eq!( + session_reconnect_delay(20, true, false), + Duration::from_secs(10) + ); + assert_eq!(session_reconnect_delay(20, false, false), RECONNECT_MAX); + assert_eq!(session_reconnect_delay(20, true, true), RECONNECT_MAX); + } + #[test] fn the_auth_backoff_floor_is_far_longer_than_the_first_retry() { // Being signed out or unentitled is not fixed by retrying in a second. diff --git a/src-tauri/src/sync/engine.rs b/src-tauri/src/sync/engine.rs index 9d1a29ea..aa3d3aa6 100644 --- a/src-tauri/src/sync/engine.rs +++ b/src-tauri/src/sync/engine.rs @@ -14,7 +14,7 @@ use std::collections::{HashMap, HashSet}; use std::fs; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex as StdMutex}; +use std::sync::{Arc, Mutex as StdMutex, Weak}; use std::time::Instant; use tokio::sync::{Mutex as TokioMutex, Semaphore}; @@ -39,6 +39,22 @@ impl ProfileSyncOutcome { } } +fn profile_sync_lock(profile_id: &str) -> Arc> { + static LOCKS: std::sync::OnceLock>>>> = + std::sync::OnceLock::new(); + let mut locks = LOCKS + .get_or_init(StdMutex::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + locks.retain(|_, lock| lock.strong_count() > 0); + if let Some(lock) = locks.get(profile_id).and_then(Weak::upgrade) { + return lock; + } + let lock = Arc::new(TokioMutex::new(())); + locks.insert(profile_id.to_string(), Arc::downgrade(&lock)); + lock +} + static SYNC_CANCEL_FLAGS: std::sync::LazyLock>>> = std::sync::LazyLock::new(|| StdMutex::new(HashMap::new())); @@ -605,8 +621,7 @@ impl SyncEngine { /// Reconcile a profile, stating which side wins and whether anything happened. /// - /// The outcome matters to exactly one caller: the pull that follows a remote - /// session. Every skip below returns `Ok(())` from `sync_profile`, so a caller + /// Every skip below returns `Ok(())` from `sync_profile`, so a caller /// that treated success as "the profile is now current" would clear the local /// launch gate without having downloaded a single byte — and the user would /// then open a stale profile over the session's work. `Skipped` says so. @@ -628,6 +643,38 @@ impl SyncEngine { bias: DiffBias, reencrypt: bool, ) -> SyncResult { + let profile_id = profile.id.to_string(); + let _sync_guard = profile_sync_lock(&profile_id).lock_owned().await; + let pending_session = match crate::remote_handoff::session_for(&profile_id) { + Some((crate::remote_handoff::HandoffState::Running, _)) => { + return Ok(ProfileSyncOutcome::Skipped("profile is running remotely")); + } + Some((crate::remote_handoff::HandoffState::PendingSync, session_id)) => Some(session_id), + None => None, + }; + if bias == DiffBias::PreferRemote && pending_session.is_none() { + return Ok(ProfileSyncOutcome::Skipped( + "remote work was already pulled", + )); + } + if reencrypt && pending_session.is_some() { + return Err(SyncError::InvalidData( + "Remote profile changes are pending".to_string(), + )); + } + let bias = if pending_session.is_some() { + DiffBias::PreferRemote + } else { + bias + }; + let current_profile = ProfileManager::instance() + .list_profiles() + .map_err(|e| SyncError::IoError(format!("Failed to read profile: {e}")))? + .into_iter() + .find(|current| current.id == profile.id) + .ok_or_else(|| SyncError::InvalidData(format!("Profile {profile_id} not found")))?; + let profile = ¤t_profile; + if profile.is_cross_os() { log::info!( "Cross-OS profile: {} ({}) — syncing metadata only", @@ -664,6 +711,16 @@ impl SyncEngine { // Skip if profile is locked by another team member, or by one of this // user's own remote sessions. + if pending_session.is_some() + && crate::team_lock::TEAM_LOCK + .is_locked_by_another(&profile_id) + .await + { + crate::team_lock::TEAM_LOCK + .fetch_locks() + .await + .map_err(SyncError::NetworkError)?; + } if crate::team_lock::TEAM_LOCK .is_locked_by_another(&profile.id.to_string()) .await @@ -781,11 +838,20 @@ impl SyncEngine { .await? }; + if pending_session.is_some() && remote_manifest.is_none() { + return Ok(ProfileSyncOutcome::Skipped( + "remote manifest is not available yet", + )); + } + // Compute diff let diff = compute_diff_with_bias(&local_manifest, remote_manifest.as_ref(), bias); - if diff.is_empty() && !reencrypt { + if diff.is_empty() && !reencrypt && remote_manifest.is_some() && profile.last_sync.is_some() { log::info!("Profile {} is already in sync", profile_id); + if let Some(session_id) = &pending_session { + crate::remote_handoff::clear_pending(&profile_id, session_id); + } let _ = events::emit( "profile-sync-status", serde_json::json!({ @@ -849,6 +915,11 @@ impl SyncEngine { // Perform downloads if !diff.files_to_download.is_empty() { + if pending_session.is_some() { + // The diff already excludes files whose hashes match. A resume record + // from an older session must not skip a file that changed again. + SyncResumeState::delete(&profile_dir); + } self .download_profile_files( app_handle, @@ -904,6 +975,9 @@ impl SyncEngine { !diff.files_to_download.is_empty() || !diff.files_to_delete_local.is_empty(); let final_manifest = if local_changed { let mut new_cache = HashCache::load(&cache_path); + for file in &diff.files_to_download { + new_cache.entries.remove(&file.path); + } let mut regenerated = generate_manifest(&profile_id, &profile_dir, &mut new_cache)?; new_cache.save(&cache_path)?; regenerated.encrypted = encryption_key.is_some(); @@ -914,28 +988,39 @@ impl SyncEngine { m }; - // Upload manifest.json last for atomicity - self - .upload_manifest( - &profile_id, - &final_manifest, - encryption_key.as_ref(), - &key_prefix, - ) - .await?; + if pending_session.is_some() { + // Partial downloads must remain pending, including files outside the + // critical-file list. Never publish an incomplete return as the new copy. + if !compute_diff_with_bias(&final_manifest, remote_manifest.as_ref(), bias).is_empty() { + return Err(SyncError::NetworkError( + "Remote profile download is incomplete".to_string(), + )); + } + } else { + self + .upload_manifest( + &profile_id, + &final_manifest, + encryption_key.as_ref(), + &key_prefix, + ) + .await?; + } // Sync completed successfully — clean up resume state SyncResumeState::delete(&profile_dir); // Sync associated proxy, group, and VPN - if let Some(proxy_id) = &profile.proxy_id { - let _ = self.sync_proxy(proxy_id, Some(app_handle)).await; - } - if let Some(group_id) = &profile.group_id { - let _ = self.sync_group(group_id, Some(app_handle)).await; - } - if let Some(vpn_id) = &profile.vpn_id { - let _ = self.sync_vpn(vpn_id, Some(app_handle)).await; + if pending_session.is_none() { + if let Some(proxy_id) = &profile.proxy_id { + let _ = self.sync_proxy(proxy_id, Some(app_handle)).await; + } + if let Some(group_id) = &profile.group_id { + let _ = self.sync_group(group_id, Some(app_handle)).await; + } + if let Some(vpn_id) = &profile.vpn_id { + let _ = self.sync_vpn(vpn_id, Some(app_handle)).await; + } } let mut updated_profile = profile.clone(); @@ -952,6 +1037,10 @@ impl SyncEngine { })?; let _ = events::emit("profiles-changed", ()); + if let Some(session_id) = &pending_session { + crate::remote_handoff::clear_pending(&profile_id, session_id); + } + let _ = events::emit( "profile-sync-status", serde_json::json!({ @@ -961,6 +1050,8 @@ impl SyncEngine { }), ); + crate::cookie_bot::report_profile_state(&updated_profile); + log::info!("Profile {} synced successfully", profile_id); Ok(ProfileSyncOutcome::Completed) } @@ -1791,6 +1882,7 @@ impl SyncEngine { })?; log::info!("Proxy {} uploaded", proxy.id); + crate::cookie_bot::report_profiles_using_proxy(&proxy.id); Ok(()) } @@ -2173,6 +2265,7 @@ impl SyncEngine { } log::info!("VPN {} uploaded", vpn.id); + crate::cookie_bot::report_profiles_using_vpn(&vpn.id); Ok(()) } @@ -4552,6 +4645,23 @@ pub async fn rollover_encryption_for_all_entities( mod tests { use super::*; + #[test] + fn sync_calls_share_a_lock_only_for_the_same_profile() { + let first = profile_sync_lock("sync-lock-profile-1") + .try_lock_owned() + .unwrap(); + assert!(profile_sync_lock("sync-lock-profile-1") + .try_lock_owned() + .is_err()); + assert!(profile_sync_lock("sync-lock-profile-2") + .try_lock_owned() + .is_ok()); + drop(first); + assert!(profile_sync_lock("sync-lock-profile-1") + .try_lock_owned() + .is_ok()); + } + /// The whole of issue 534, at the only place the user ever sees it. /// /// A self-hosted server signs every presigned URL against the address it uses diff --git a/src-tauri/src/sync/mod.rs b/src-tauri/src/sync/mod.rs index 5ea6f9e4..b5c9556e 100644 --- a/src-tauri/src/sync/mod.rs +++ b/src-tauri/src/sync/mod.rs @@ -62,6 +62,38 @@ pub fn stop_pipeline() { } } +pub fn stop_subscription() { + if let Ok(mut guard) = GLOBAL_SUBSCRIPTION.lock() { + if let Some(subscription) = guard.as_mut() { + subscription.stop(); + } + } +} + +pub async fn resume_subscription(app_handle: tauri::AppHandle) { + let _building = PIPELINE_LOCK.lock().await; + + let taken = GLOBAL_SUBSCRIPTION + .lock() + .ok() + .and_then(|mut guard| guard.take()); + let Some(mut subscription_manager) = taken else { + return; + }; + + if !subscription_manager.is_running() { + if let Err(e) = subscription_manager.start(app_handle).await { + log::warn!("Failed to resume sync subscription: {e}"); + } + } + + if let Ok(mut guard) = GLOBAL_SUBSCRIPTION.lock() { + if guard.is_none() { + *guard = Some(subscription_manager); + } + } +} + /// Build and start the sync pipeline. Safe to call again to restart it. /// /// Startup and `restart_sync_service` each held their own copy of this, and the @@ -80,6 +112,7 @@ pub async fn start_pipeline(app_handle: tauri::AppHandle) { let _building = PIPELINE_LOCK.lock().await; stop_pipeline(); + crate::remote_handoff::resume_pending_pulls(&app_handle); let mut subscription_manager = SubscriptionManager::new(); let Some(work_rx) = subscription_manager.take_work_receiver() else { diff --git a/src-tauri/src/sync/scheduler.rs b/src-tauri/src/sync/scheduler.rs index 6c2e294e..6ae135b3 100644 --- a/src-tauri/src/sync/scheduler.rs +++ b/src-tauri/src/sync/scheduler.rs @@ -425,7 +425,13 @@ impl SyncScheduler { received = work_rx.recv(), if work_channel_open => { match received { Some(work_item) => match work_item { - SyncWorkItem::Profile(id) => scheduler.queue_profile_sync(id).await, + SyncWorkItem::Profile(id) => { + if crate::remote_handoff::pending_session_for(&id).is_some() { + crate::remote_handoff::schedule_pull(app_handle_clone.clone(), id); + } else { + scheduler.queue_profile_sync(id).await; + } + } SyncWorkItem::Proxy(id) => scheduler.queue_proxy_sync(id).await, SyncWorkItem::Group(id) => scheduler.queue_group_sync(id).await, SyncWorkItem::Vpn(id) => scheduler.queue_vpn_sync(id).await, @@ -535,7 +541,11 @@ impl SyncScheduler { }; let result = match SyncEngine::create_from_settings(&app).await { - Ok(engine) => engine.sync_profile(&app, &profile).await, + Ok(engine) => { + engine + .sync_profile_with_bias(&app, &profile, super::DiffBias::Auto) + .await + } Err(e) => { log::error!("Failed to create sync engine: {}", e); Err(super::types::SyncError::NotConfigured) @@ -548,7 +558,7 @@ impl SyncScheduler { } match result { - Ok(()) => { + Ok(super::ProfileSyncOutcome::Completed) => { log::info!("Profile {} synced successfully", profile_id); let _ = events::emit( "profile-sync-status", @@ -558,6 +568,9 @@ impl SyncScheduler { }), ); } + Ok(super::ProfileSyncOutcome::Skipped(reason)) => { + log::debug!("Profile {profile_id} sync is waiting: {reason}"); + } Err(e) => { log::error!("Failed to sync profile {}: {}", profile_id, e); let _ = events::emit( diff --git a/src-tauri/src/sync/subscription.rs b/src-tauri/src/sync/subscription.rs index ac1c1a10..f4dab2c1 100644 --- a/src-tauri/src/sync/subscription.rs +++ b/src-tauri/src/sync/subscription.rs @@ -37,6 +37,67 @@ enum TokenSource { SelfHosted, } +const RECONNECT_BASE_DELAY: Duration = Duration::from_secs(6); +const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(600); +const STOP_POLL_INTERVAL: Duration = Duration::from_secs(1); + +#[derive(Debug)] +enum StreamFailure { + Refused(u16), + RateLimited(Duration), + Failed(String), +} + +#[derive(Debug, PartialEq, Eq)] +enum TokenStep { + Use(String), + Keep, + Wait, + Stop, +} + +fn reconnect_delay(failures: u32) -> Duration { + let doublings = failures.saturating_sub(1).min(16); + RECONNECT_BASE_DELAY + .saturating_mul(1u32 << doublings) + .min(RECONNECT_MAX_DELAY) +} + +fn is_refused_fetch(error: &str) -> bool { + error.contains("(401") + || error.contains("(403") + || error.contains("No refresh token") + || error.contains("Not logged in") +} + +fn next_token(refused: Option<&str>, fetched: Result, String>) -> TokenStep { + match fetched { + Ok(Some(token)) if refused == Some(token.as_str()) => TokenStep::Stop, + Ok(Some(token)) => TokenStep::Use(token), + Ok(None) => TokenStep::Stop, + Err(error) if is_refused_fetch(&error) => TokenStep::Stop, + Err(_) if refused.is_some() => TokenStep::Wait, + Err(_) => TokenStep::Keep, + } +} + +fn stream_failure(status: u16, headers: &reqwest::header::HeaderMap) -> StreamFailure { + match status { + 401 | 403 => StreamFailure::Refused(status), + 429 => StreamFailure::RateLimited(crate::cloud_auth::retry_after(headers)), + _ => StreamFailure::Failed(format!("SSE connection failed with status: {status}")), + } +} + +async fn sleep_while_running(running: &AtomicBool, total: Duration) { + let mut left = total; + while !left.is_zero() && running.load(Ordering::SeqCst) { + let step = left.min(STOP_POLL_INTERVAL); + sleep(step).await; + left = left.saturating_sub(step); + } +} + pub struct SyncSubscription { client: Client, base_url: String, @@ -125,41 +186,84 @@ impl SyncSubscription { let source = self.source; let work_tx = self.work_tx.clone(); let client = self.client.clone(); - let mut token = self.token.clone(); + let mut token = Some(self.token.clone()); tokio::spawn(async move { + let mut refused: Option = None; + let mut failures: u32 = 0; + while running.load(Ordering::SeqCst) { - match Self::connect_and_listen(&client, &base_url, &token, &work_tx, &running, &app_handle) - .await - { - Ok(()) => { - log::info!("SSE connection closed gracefully"); + let mut wait_at_least = Duration::ZERO; + + if let Some(current) = token.clone() { + let mut connected = false; + let outcome = Self::connect_and_listen( + &client, + &base_url, + ¤t, + &work_tx, + &running, + &mut connected, + ) + .await; + if connected { + failures = 0; } - Err(e) => { - log::warn!("SSE connection error: {e}, reconnecting in 5s"); - sleep(Duration::from_secs(5)).await; + match outcome { + Ok(()) => { + log::info!("SSE connection closed gracefully"); + } + Err(StreamFailure::Refused(status)) => { + log::warn!("Sync subscription refused with status {status}"); + if matches!(source, TokenSource::Cloud) { + crate::cloud_auth::CloudAuthManager::discard_cloud_sync_token(¤t); + } + refused = Some(current); + token = None; + } + Err(StreamFailure::RateLimited(retry)) => { + log::warn!("Sync subscription rate limited for {}s", retry.as_secs()); + wait_at_least = retry; + } + Err(StreamFailure::Failed(e)) => { + log::warn!("SSE connection error: {e}"); + } } } + if !running.load(Ordering::SeqCst) { + break; + } + + failures = failures.saturating_add(1); + let delay = reconnect_delay(failures).max(wait_at_least); + log::debug!("Reconnecting the sync subscription in {}s", delay.as_secs()); + sleep_while_running(&running, delay).await; + if running.load(Ordering::SeqCst) { - sleep(Duration::from_secs(1)).await; // Refresh the sync token before reconnecting. The token may have // expired while the stream was open (tokens last ~15 min); reusing // the construction-time token otherwise produces an endless 401 // reconnect loop until the app is restarted. - match Self::fetch_sync_token(source, &app_handle).await { - Ok(Some(fresh)) => token = fresh, - Ok(None) => { + let fetched = Self::fetch_sync_token(source, &app_handle).await; + if let Err(e) = &fetched { + log::warn!("Failed to refresh sync token: {e}"); + } + match next_token(refused.as_deref(), fetched) { + TokenStep::Use(fresh) => { + token = Some(fresh); + refused = None; + } + TokenStep::Keep | TokenStep::Wait => {} + TokenStep::Stop => { log::info!("Sync token no longer available; stopping subscription"); break; } - Err(e) => { - log::warn!("Failed to refresh sync token: {e}; retrying with the current token"); - } } } } + running.store(false, Ordering::SeqCst); log::info!("Sync subscription stopped"); }); } @@ -188,8 +292,8 @@ impl SyncSubscription { token: &str, work_tx: &mpsc::UnboundedSender, running: &Arc, - _app_handle: &tauri::AppHandle, - ) -> Result<(), String> { + connected: &mut bool, + ) -> Result<(), StreamFailure> { let url = format!("{base_url}/v1/objects/subscribe"); let response = client @@ -198,15 +302,16 @@ impl SyncSubscription { .header("Accept", "text/event-stream") .send() .await - .map_err(|e| format!("Failed to connect to SSE: {e}"))?; + .map_err(|e| StreamFailure::Failed(format!("Failed to connect to SSE: {e}")))?; if !response.status().is_success() { - return Err(format!( - "SSE connection failed with status: {}", - response.status() + return Err(stream_failure( + response.status().as_u16(), + response.headers(), )); } + *connected = true; log::info!("Connected to sync subscription"); let _ = events::emit("sync-subscription-status", "connected"); @@ -231,7 +336,7 @@ impl SyncSubscription { } } Ok(Some(Err(e))) => { - return Err(format!("SSE stream error: {e}")); + return Err(StreamFailure::Failed(format!("SSE stream error: {e}"))); } Ok(None) => { return Ok(()); @@ -395,9 +500,10 @@ impl SubscriptionManager { } pub async fn start(&mut self, app_handle: tauri::AppHandle) -> Result<(), String> { - if self.subscription.is_some() { + if self.is_running() { return Ok(()); } + self.subscription = None; let subscription = SyncSubscription::create_from_settings(&app_handle, self.work_tx.clone()).await?; @@ -425,3 +531,97 @@ impl SubscriptionManager { self.subscription.as_ref().is_some_and(|s| s.is_running()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reconnects_back_off_from_six_seconds_to_ten_minutes() { + let delays: Vec = (1..=10).map(|n| reconnect_delay(n).as_secs()).collect(); + assert_eq!(delays, vec![6, 12, 24, 48, 96, 192, 384, 600, 600, 600]); + assert_eq!(reconnect_delay(0), Duration::from_secs(6)); + assert_eq!(reconnect_delay(u32::MAX), Duration::from_secs(600)); + } + + #[test] + fn a_refused_token_is_never_sent_again() { + assert_eq!( + next_token(Some("old"), Ok(Some("old".to_string()))), + TokenStep::Stop + ); + assert_eq!( + next_token(Some("old"), Ok(Some("new".to_string()))), + TokenStep::Use("new".to_string()) + ); + assert_eq!( + next_token( + Some("old"), + Err("Failed to get sync token: timed out".to_string()) + ), + TokenStep::Wait + ); + } + + #[test] + fn a_refused_refresh_stops_the_subscription() { + for refusal in [ + "Failed to refresh cloud sync token: Token refresh failed (401 Unauthorized)", + "Failed to refresh cloud sync token: Sync token request failed (403 Forbidden): {}", + "Failed to refresh cloud sync token: No refresh token stored", + "Failed to refresh cloud sync token: Not logged in", + ] { + assert_eq!(next_token(None, Err(refusal.to_string())), TokenStep::Stop); + assert_eq!( + next_token(Some("old"), Err(refusal.to_string())), + TokenStep::Stop + ); + } + assert_eq!(next_token(None, Ok(None)), TokenStep::Stop); + } + + #[test] + fn a_transient_refresh_failure_keeps_a_token_the_server_still_accepts() { + assert_eq!( + next_token( + None, + Err("Failed to get sync token: connection reset".to_string()) + ), + TokenStep::Keep + ); + assert_eq!( + next_token(None, Ok(Some("same".to_string()))), + TokenStep::Use("same".to_string()) + ); + } + + #[test] + fn the_stream_status_decides_refusal_and_rate_limits() { + let mut headers = reqwest::header::HeaderMap::new(); + assert!(matches!( + stream_failure(401, &headers), + StreamFailure::Refused(401) + )); + assert!(matches!( + stream_failure(403, &headers), + StreamFailure::Refused(403) + )); + assert!(matches!( + stream_failure(502, &headers), + StreamFailure::Failed(_) + )); + headers.insert(reqwest::header::RETRY_AFTER, "900".parse().unwrap()); + match stream_failure(429, &headers) { + StreamFailure::RateLimited(wait) => assert_eq!(wait, Duration::from_secs(900)), + other => panic!("expected a rate limit, got {other:?}"), + } + } + + #[tokio::test] + async fn a_stopped_subscription_ends_its_wait_promptly() { + let running = AtomicBool::new(false); + let started = std::time::Instant::now(); + sleep_while_running(&running, Duration::from_secs(600)).await; + assert!(started.elapsed() < Duration::from_millis(100)); + } +} diff --git a/src-tauri/src/team_lock.rs b/src-tauri/src/team_lock.rs index 0ae8584f..f36bf272 100644 --- a/src-tauri/src/team_lock.rs +++ b/src-tauri/src/team_lock.rs @@ -1,10 +1,20 @@ -use reqwest::Client; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::time::{Duration, Instant}; use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinHandle; -use crate::cloud_auth::{CloudAuthManager, CLOUD_API_URL, CLOUD_AUTH}; +use crate::cloud_auth::{AuthorizedError, CLOUD_API_URL, CLOUD_AUTH}; + +pub const LOCK_CHANGED_EVENT: &str = "profile-lock-changed"; +pub const SHUTDOWN_RELEASE_TIMEOUT: Duration = Duration::from_secs(2); +const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30); +const RELEASE_RETRY_DELAYS: [Duration; 4] = [ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), + Duration::from_secs(8), +]; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ProfileLockInfo { @@ -30,10 +40,115 @@ struct AcquireLockResponse { locked_by_email: Option, } +#[derive(Debug, Deserialize)] +struct HeartbeatResponse { + #[serde(default)] + refreshed: bool, +} + +#[derive(Debug, PartialEq, Eq)] +enum Renewal { + Refreshed, + Lost, + Failed, +} + +#[derive(Debug, PartialEq, Eq)] +enum ReleaseStep { + Done, + Retry, + GiveUp, +} + +fn renewal_outcome(status: u16, body: &str) -> Renewal { + if !(200..300).contains(&status) { + return Renewal::Failed; + } + match serde_json::from_str::(body) { + Ok(reply) if reply.refreshed => Renewal::Refreshed, + Ok(_) => Renewal::Lost, + Err(_) => Renewal::Failed, + } +} + +fn release_step(status: u16) -> ReleaseStep { + match status { + 200..=299 | 404 => ReleaseStep::Done, + 408 | 500..=599 => ReleaseStep::Retry, + _ => ReleaseStep::GiveUp, + } +} + +fn release_retry_delay(attempt: usize) -> Option { + RELEASE_RETRY_DELAYS.get(attempt).copied() +} + +fn lock_signature(locks: &HashMap) -> BTreeSet<(String, String)> { + locks + .values() + .map(|lock| (lock.profile_id.clone(), lock.locked_by.clone())) + .collect() +} + +fn emit_lock_change(profile_id: Option<&str>, action: &str) { + let _ = crate::events::emit( + LOCK_CHANGED_EVENT, + serde_json::json!({ "profileId": profile_id, "action": action }), + ); +} + +fn lock_url(profile_id: &str) -> String { + format!("{CLOUD_API_URL}/api/profile-locks/{profile_id}") +} + +async fn release_attempt(profile_id: &str) -> Option { + let url = lock_url(profile_id); + match CLOUD_AUTH + .authorized_request(|client, token| client.delete(&url).bearer_auth(token)) + .await + { + Ok(response) => match release_step(response.status().as_u16()) { + ReleaseStep::Done => None, + ReleaseStep::Retry => Some(Duration::ZERO), + ReleaseStep::GiveUp => { + log::warn!( + "Profile lock release for {profile_id} was refused ({})", + response.status() + ); + None + } + }, + Err(AuthorizedError::SignedOut) => None, + Err(AuthorizedError::RateLimited(wait)) => Some(wait), + Err(AuthorizedError::Transport(e)) => { + log::debug!("Profile lock release for {profile_id} failed: {e}"); + Some(Duration::ZERO) + } + } +} + +async fn release_with_retry(profile_id: String, mut pending: Option) { + let mut attempt = 0; + while let Some(wait) = pending { + let Some(delay) = release_retry_delay(attempt) else { + log::warn!("Could not release the profile lock for {profile_id}"); + return; + }; + attempt += 1; + tokio::time::sleep(delay.max(wait)).await; + if PROFILE_LOCK.holds(&profile_id).await { + return; + } + pending = release_attempt(&profile_id).await; + } +} + pub struct ProfileLockManager { locks: RwLock>, + held: RwLock>, heartbeat_handle: Mutex>>, connected: Mutex, + paused_until: std::sync::Mutex>, } pub static PROFILE_LOCK: std::sync::LazyLock = @@ -46,8 +161,10 @@ impl ProfileLockManager { fn new() -> Self { Self { locks: RwLock::new(HashMap::new()), + held: RwLock::new(HashSet::new()), heartbeat_handle: Mutex::new(None), connected: Mutex::new(false), + paused_until: std::sync::Mutex::new(None), } } @@ -81,6 +198,8 @@ impl ProfileLockManager { locks.clear(); } + self.held.write().await.clear(); + { let mut c = self.connected.lock().await; *c = false; @@ -91,18 +210,47 @@ impl ProfileLockManager { *self.connected.lock().await } - pub async fn acquire_lock(&self, profile_id: &str) -> Result<(), String> { - let client = Client::new(); - let access_token = CloudAuthManager::load_access_token()? - .ok_or_else(|| crate::backend_error("PROFILE_LOCK_UNAVAILABLE"))?; + pub async fn holds(&self, profile_id: &str) -> bool { + self.held.read().await.contains(profile_id) + } - let url = format!("{CLOUD_API_URL}/api/profile-locks/{profile_id}"); - let response = client - .post(&url) - .header("Authorization", format!("Bearer {access_token}")) - .send() + async fn held_ids(&self) -> Vec { + self.held.read().await.iter().cloned().collect() + } + + async fn forget(&self, profile_id: &str) { + self.held.write().await.remove(profile_id); + self.locks.write().await.remove(profile_id); + } + + fn pause_for(&self, wait: Duration) { + *self + .paused_until + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Instant::now() + wait); + } + + fn is_paused(&self) -> bool { + self + .paused_until + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_some_and(|until| Instant::now() < until) + } + + fn note_failure(&self, error: &AuthorizedError) { + if let AuthorizedError::RateLimited(wait) = error { + self.pause_for(*wait); + } + } + + pub async fn acquire_lock(&self, profile_id: &str) -> Result<(), String> { + let url = lock_url(profile_id); + let response = CLOUD_AUTH + .authorized_request(|client, token| client.post(&url).bearer_auth(token)) .await .map_err(|e| { + self.note_failure(&e); log::warn!("Failed to acquire profile lock for {profile_id}: {e}"); crate::backend_error("PROFILE_LOCK_UNAVAILABLE") })?; @@ -127,6 +275,8 @@ impl ProfileLockManager { )); } + self.held.write().await.insert(profile_id.to_string()); + // Update local cache if let Some(user) = CLOUD_AUTH.get_user().await { let mut locks = self.locks.write().await; @@ -142,37 +292,47 @@ impl ProfileLockManager { ); } - let _ = crate::events::emit( - "profile-lock-changed", - serde_json::json!({ "profileId": profile_id, "action": "acquired" }), - ); + emit_lock_change(Some(profile_id), "acquired"); Ok(()) } - pub async fn release_lock(&self, profile_id: &str) -> Result<(), String> { - let client = Client::new(); - let access_token = - CloudAuthManager::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; + pub async fn release_lock(&self, profile_id: &str) { + self.forget(profile_id).await; + emit_lock_change(Some(profile_id), "released"); - let url = format!("{CLOUD_API_URL}/api/profile-locks/{profile_id}"); - let _ = client - .delete(&url) - .header("Authorization", format!("Bearer {access_token}")) - .send() - .await; + let pending = release_attempt(profile_id).await; + if pending.is_some() { + let profile_id = profile_id.to_string(); + tauri::async_runtime::spawn(release_with_retry(profile_id, pending)); + } + } + pub async fn release_all_held_within(&self, limit: Duration) { + let held: Vec = self.held.write().await.drain().collect(); + if held.is_empty() { + return; + } { let mut locks = self.locks.write().await; - locks.remove(profile_id); + for profile_id in &held { + locks.remove(profile_id); + } + } + for profile_id in &held { + emit_lock_change(Some(profile_id), "released"); } - let _ = crate::events::emit( - "profile-lock-changed", - serde_json::json!({ "profileId": profile_id, "action": "released" }), - ); - - Ok(()) + let releases = held.into_iter().map(|profile_id| async move { + let pending = release_attempt(&profile_id).await; + release_with_retry(profile_id, pending).await; + }); + if tokio::time::timeout(limit, futures_util::future::join_all(releases)) + .await + .is_err() + { + log::warn!("Timed out releasing profile locks"); + } } pub async fn get_locks(&self) -> Vec { @@ -195,18 +355,15 @@ impl ProfileLockManager { false } - async fn fetch_locks(&self) -> Result<(), String> { - let client = Client::new(); - let access_token = - CloudAuthManager::load_access_token()?.ok_or_else(|| "Not logged in".to_string())?; - + pub(crate) async fn fetch_locks(&self) -> Result<(), String> { let url = format!("{CLOUD_API_URL}/api/profile-locks"); - let response = client - .get(&url) - .header("Authorization", format!("Bearer {access_token}")) - .send() + let response = CLOUD_AUTH + .authorized_request(|client, token| client.get(&url).bearer_auth(token)) .await - .map_err(|e| format!("Failed to fetch locks: {e}"))?; + .map_err(|e| { + self.note_failure(&e); + format!("Failed to fetch locks: {e}") + })?; if !response.status().is_success() { return Err("Failed to fetch locks".to_string()); @@ -217,13 +374,48 @@ impl ProfileLockManager { .await .map_err(|e| format!("Failed to parse locks: {e}"))?; + if self.replace_display_locks(lock_list).await { + emit_lock_change(None, "refreshed"); + } + + Ok(()) + } + + async fn replace_display_locks(&self, lock_list: Vec) -> bool { let mut locks = self.locks.write().await; + let before = lock_signature(&locks); locks.clear(); for lock in lock_list { locks.insert(lock.profile_id.clone(), lock); } + lock_signature(&locks) != before + } - Ok(()) + async fn renew(&self, profile_id: &str) -> Result { + let url = format!("{}/heartbeat", lock_url(profile_id)); + let response = CLOUD_AUTH + .authorized_request(|client, token| client.post(&url).bearer_auth(token)) + .await?; + let status = response.status().as_u16(); + let body = response.text().await.unwrap_or_default(); + Ok(renewal_outcome(status, &body)) + } + + async fn recover_lost_lock(&self, profile_id: &str) { + if !self.holds(profile_id).await { + return; + } + if crate::browser_runner::browser_is_running_for(profile_id) { + log::warn!("Profile lock for {profile_id} was not renewed; taking it again"); + if let Err(e) = self.acquire_lock(profile_id).await { + log::warn!("Could not take the profile lock for {profile_id} again: {e}"); + self.forget(profile_id).await; + emit_lock_change(Some(profile_id), "released"); + } + } else { + self.forget(profile_id).await; + emit_lock_change(Some(profile_id), "released"); + } } async fn start_heartbeat_loop(&self) { @@ -234,38 +426,45 @@ impl ProfileLockManager { let h = tokio::spawn(async move { loop { - tokio::time::sleep(std::time::Duration::from_secs(30)).await; + tokio::time::sleep(HEARTBEAT_INTERVAL).await; if !PROFILE_LOCK.is_connected().await { break; } + if PROFILE_LOCK.is_paused() { + continue; + } + // Send heartbeat for each held lock - let held_locks: Vec = { - let locks = PROFILE_LOCK.locks.read().await; - if let Some(user) = CLOUD_AUTH.get_user().await { - locks - .values() - .filter(|l| l.locked_by == user.user.id) - .map(|l| l.profile_id.clone()) - .collect() - } else { - vec![] - } - }; + let held_locks = PROFILE_LOCK.held_ids().await; + let mut signed_out = false; for profile_id in held_locks { - let client = Client::new(); - if let Ok(Some(token)) = CloudAuthManager::load_access_token() { - let url = format!("{CLOUD_API_URL}/api/profile-locks/{profile_id}/heartbeat"); - let _ = client - .post(&url) - .header("Authorization", format!("Bearer {token}")) - .send() - .await; + match PROFILE_LOCK.renew(&profile_id).await { + Ok(Renewal::Refreshed) => {} + Ok(Renewal::Lost) => PROFILE_LOCK.recover_lost_lock(&profile_id).await, + Ok(Renewal::Failed) => { + log::debug!("Profile lock heartbeat for {profile_id} was not accepted"); + } + Err(AuthorizedError::SignedOut) => { + signed_out = true; + break; + } + Err(e) => { + PROFILE_LOCK.note_failure(&e); + log::debug!("Profile lock heartbeat for {profile_id} failed: {e}"); + if PROFILE_LOCK.is_paused() { + break; + } + } } } + if signed_out || PROFILE_LOCK.is_paused() { + continue; + } + // Refresh lock state from server if let Err(e) = PROFILE_LOCK.fetch_locks().await { log::debug!("Failed to refresh profile locks: {e}"); @@ -348,16 +547,12 @@ pub async fn acquire_team_lock_if_needed( /// Release profile lock if profile is sync-enabled and user has a paid subscription. pub async fn release_team_lock_if_needed(profile: &crate::profile::BrowserProfile) { - if !profile.is_sync_enabled() { - return; - } - if !CLOUD_AUTH.has_active_paid_subscription().await { + let profile_id = profile.id.to_string(); + if !PROFILE_LOCK.holds(&profile_id).await { return; } - if let Err(e) = PROFILE_LOCK.release_lock(&profile.id.to_string()).await { - log::warn!("Failed to release profile lock for {}: {e}", profile.id); - } + PROFILE_LOCK.release_lock(&profile_id).await; } // --- Tauri commands --- @@ -398,6 +593,154 @@ mod tests { assert_eq!(json["params"]["email"], "mate@example.com"); } + fn server_lock(profile_id: &str, holder: &str) -> ProfileLockInfo { + ProfileLockInfo { + profile_id: profile_id.to_string(), + locked_by: holder.to_string(), + locked_by_email: "owner@example.com".to_string(), + locked_at: "2026-09-27T00:00:00Z".to_string(), + expires_at: None, + } + } + + #[tokio::test] + async fn listing_the_server_locks_never_makes_them_renewal_targets() { + let manager = ProfileLockManager::new(); + manager.held.write().await.insert("taken-here".to_string()); + + let changed = manager + .replace_display_locks(vec![ + server_lock("left-by-a-crash", "user-1"), + server_lock("taken-here", "user-1"), + ]) + .await; + assert!(changed); + assert_eq!(manager.held_ids().await, vec!["taken-here".to_string()]); + assert_eq!(manager.get_locks().await.len(), 2); + + manager.replace_display_locks(vec![]).await; + assert!(manager.get_locks().await.is_empty()); + assert!(manager.holds("taken-here").await); + assert!(!manager.holds("left-by-a-crash").await); + } + + #[tokio::test] + async fn an_unchanged_listing_is_not_reported_as_a_change() { + let manager = ProfileLockManager::new(); + assert!( + manager + .replace_display_locks(vec![server_lock("p1", "user-1")]) + .await + ); + assert!( + !manager + .replace_display_locks(vec![server_lock("p1", "user-1")]) + .await + ); + assert!( + manager + .replace_display_locks(vec![server_lock("p1", "user-2")]) + .await + ); + } + + #[tokio::test] + async fn disconnecting_forgets_every_lock_taken_here() { + let manager = ProfileLockManager::new(); + manager.held.write().await.insert("p1".to_string()); + manager + .replace_display_locks(vec![server_lock("p1", "user-1")]) + .await; + manager.disconnect().await; + assert!(manager.held_ids().await.is_empty()); + assert!(manager.get_locks().await.is_empty()); + assert!(!manager.is_connected().await); + } + + #[tokio::test] + async fn releasing_with_nothing_held_returns_at_once() { + let manager = ProfileLockManager::new(); + let started = std::time::Instant::now(); + manager + .release_all_held_within(SHUTDOWN_RELEASE_TIMEOUT) + .await; + assert!(started.elapsed() < Duration::from_millis(500)); + } + + #[test] + fn a_renewal_that_did_not_refresh_means_the_lock_is_gone() { + assert_eq!( + renewal_outcome(201, r#"{"refreshed":true}"#), + Renewal::Refreshed + ); + assert_eq!( + renewal_outcome(200, r#"{"refreshed":false}"#), + Renewal::Lost + ); + assert_eq!(renewal_outcome(201, "{}"), Renewal::Lost); + assert_eq!( + renewal_outcome(502, r#"{"refreshed":false}"#), + Renewal::Failed + ); + assert_eq!(renewal_outcome(200, ""), Renewal::Failed); + } + + #[test] + fn a_release_is_retried_only_when_the_server_may_still_accept_it() { + assert_eq!(release_step(200), ReleaseStep::Done); + assert_eq!(release_step(404), ReleaseStep::Done); + assert_eq!(release_step(500), ReleaseStep::Retry); + assert_eq!(release_step(503), ReleaseStep::Retry); + assert_eq!(release_step(408), ReleaseStep::Retry); + assert_eq!(release_step(403), ReleaseStep::GiveUp); + assert_eq!(release_step(400), ReleaseStep::GiveUp); + } + + #[test] + fn release_retries_back_off_and_then_stop() { + let delays: Vec = (0..).map_while(release_retry_delay).collect(); + assert_eq!(delays.len(), 4); + assert!(delays.windows(2).all(|pair| pair[0] < pair[1])); + assert_eq!(release_retry_delay(4), None); + } + + #[test] + fn the_frontend_listens_to_the_one_event_rust_emits() { + let hook = include_str!("../../src/hooks/use-team-locks.ts"); + assert!(hook.contains(&format!("\"{LOCK_CHANGED_EVENT}\""))); + assert!(!hook.contains("team-lock-acquired")); + assert!(!hook.contains("team-lock-released")); + let source = include_str!("team_lock.rs"); + let runtime = &source[..source.find("#[cfg(test)]").unwrap_or(source.len())]; + assert_eq!(runtime.matches("crate::events::emit(").count(), 1); + assert_eq!(runtime.matches("LOCK_CHANGED_EVENT,").count(), 1); + } + + #[test] + fn releasing_depends_on_what_this_process_took_and_not_on_the_plan() { + let source = include_str!("team_lock.rs"); + let release = source + .split("pub async fn release_team_lock_if_needed(") + .nth(1) + .expect("release_team_lock_if_needed must exist"); + let body = &release[..release.find("\n}").unwrap_or(release.len())]; + assert!(body.contains("PROFILE_LOCK.holds(")); + assert!(!body.contains("has_active_paid_subscription")); + assert!(!body.contains("is_sync_enabled")); + } + + #[test] + fn the_app_releases_what_it_holds_when_it_exits() { + let lib = include_str!("lib.rs"); + let exit = lib + .split("if let tauri::RunEvent::Exit = _event {") + .nth(1) + .expect("the exit handler must exist"); + let body = &exit[..exit.find("\n }").unwrap_or(exit.len())]; + assert!(body.contains("release_all_held_within(team_lock::SHUTDOWN_RELEASE_TIMEOUT)")); + assert_eq!(SHUTDOWN_RELEASE_TIMEOUT, Duration::from_secs(2)); + } + #[test] fn a_lock_with_no_identifiable_holder_still_produces_a_code() { // Raw English here is what reaches a Russian user untranslated. diff --git a/src-tauri/src/wayfern_manager.rs b/src-tauri/src/wayfern_manager.rs index 4f9f1530..184c1e63 100644 --- a/src-tauri/src/wayfern_manager.rs +++ b/src-tauri/src/wayfern_manager.rs @@ -762,47 +762,21 @@ impl HeadlessWayfern { page_ws_url: String::new(), }; - if let Err(e) = manager + let targets = match manager .wait_for_cdp_ready_while_running(port, &mut session.child, &log) .await { - session.stop().await; - return Err(e); - } - - // The debugging endpoint answers as soon as the process is up, which is - // BEFORE the first window exists, so the target list is legitimately empty - // for a moment and a single read races it. Poll until a page target - // appears, and stop early if the process dies so a refused start is - // reported as its own reason rather than as a timeout. - let deadline = std::time::Instant::now() + Duration::from_secs(30); - loop { - if let Ok(Some(status)) = session.child.try_wait() { - tokio::time::sleep(Duration::from_millis(300)).await; - let said = log.lines().join("\n"); + Ok(targets) => targets, + Err(e) => { session.stop().await; - return Err(early_exit_error(&said, &status).into()); + return Err(e); } - let detail = match manager.get_cdp_targets(port).await { - Ok(targets) => { - if let Some(url) = targets - .into_iter() - .find(|t| t.target_type == "page") - .and_then(|t| t.websocket_debugger_url) - { - session.page_ws_url = url; - return Ok(session); - } - "the browser reported no page target".to_string() - } - Err(e) => e.to_string(), - }; - if std::time::Instant::now() >= deadline { - session.stop().await; - return Err(format!("No page target found for CDP: {detail}").into()); - } - tokio::time::sleep(Duration::from_millis(100)).await; - } + }; + session.page_ws_url = targets + .into_iter() + .find_map(|target| target.websocket_debugger_url) + .expect("CDP is ready only when a page has a socket"); + Ok(session) } pub(crate) fn page_ws_url(&self) -> &str { @@ -1576,7 +1550,7 @@ impl WayfernManager { port: u16, child: &mut tokio::process::Child, log: &BrowserLogTap, - ) -> Result<(), Box> { + ) -> Result, Box> { let url = format!("http://127.0.0.1:{port}/json/version"); let max_attempts = 120; let delay = Duration::from_millis(500); @@ -1591,8 +1565,10 @@ impl WayfernManager { } match self.http_client.get(&url).send().await { Ok(resp) if resp.status().is_success() => { - log::info!("CDP ready on port {port} after {attempt} attempts"); - return Ok(()); + log::info!("CDP port {port} is open after {attempt} attempts"); + return self + .wait_for_page_targets_while_running(port, child, log) + .await; } Ok(resp) => { last_error = Some(format!("HTTP {} from {url}", resp.status())); @@ -1610,12 +1586,60 @@ impl WayfernManager { Err(format!("CDP not ready after {max_attempts} attempts on port {port}: {detail}").into()) } + // The control port can open before the first tab. Both launch paths must + // wait for a page socket before they apply or check the fingerprint. + async fn wait_for_page_targets_while_running( + &self, + port: u16, + child: &mut tokio::process::Child, + log: &BrowserLogTap, + ) -> Result, Box> { + let deadline = std::time::Instant::now() + Duration::from_secs(30); + loop { + if let Ok(Some(status)) = child.try_wait() { + tokio::time::sleep(Duration::from_millis(300)).await; + return Err(early_exit_error(&log.lines().join("\n"), &status).into()); + } + let detail = match self.get_cdp_targets(port).await { + Ok(mut targets) => { + targets.retain(|target| { + target.target_type == "page" + && target + .websocket_debugger_url + .as_ref() + .is_some_and(|url| !url.is_empty()) + }); + if !targets.is_empty() { + return Ok(targets); + } + "the browser reported no page target with a socket".to_string() + } + Err(e) => e.to_string(), + }; + if std::time::Instant::now() >= deadline { + return Err( + crate::backend_error_with_detail( + "WAYFERN_FINGERPRINT_APPLY_FAILED", + format!("No page target found for CDP: {detail}"), + ) + .into(), + ); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + async fn get_cdp_targets( &self, port: u16, ) -> Result, Box> { let url = format!("http://127.0.0.1:{port}/json"); - let resp = self.http_client.get(&url).send().await?; + let resp = self + .http_client + .get(&url) + .send() + .await? + .error_for_status()?; let targets: Vec = resp.json().await?; Ok(targets) } @@ -2680,15 +2704,20 @@ impl WayfernManager { tap_browser_stderr(stderr, log_tap.clone(), profile.name.clone()); } - self + let page_targets = match self .wait_for_cdp_ready_while_running(port, &mut child, &log_tap) - .await?; + .await + { + Ok(targets) => targets, + Err(e) => { + if let Some(pid) = process_id { + kill_browser_process(pid); + } + return Err(e); + } + }; drop(child); - let targets = self.get_cdp_targets(port).await?; - log::info!("Found {} CDP targets", targets.len()); - - let page_targets: Vec<_> = targets.iter().filter(|t| t.target_type == "page").collect(); log::info!("Found {} page targets", page_targets.len()); // An identity-backed profile: the id, the user's overrides and the exit's @@ -4230,6 +4259,128 @@ mod tests { assert!(wait_for_exit(pid, Duration::from_secs(5)).await); } + fn cdp_test_process() -> tokio::process::Child { + #[cfg(unix)] + let mut command = { + let mut command = TokioCommand::new("sleep"); + command.arg("60"); + command + }; + #[cfg(windows)] + let mut command = { + let mut command = TokioCommand::new("ping"); + command.args(["-n", "60", "127.0.0.1"]); + command + }; + command + .kill_on_drop(true) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap() + } + + async fn cdp_startup_server( + replies: Vec<(axum::http::StatusCode, serde_json::Value)>, + ) -> ( + u16, + Arc, + tokio::task::JoinHandle<()>, + ) { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let requests = Arc::new(AtomicUsize::new(0)); + let seen = requests.clone(); + let app = axum::Router::new() + .route("/json/version", axum::routing::get(|| async { "{}" })) + .route( + "/json", + axum::routing::get(move || { + let attempt = seen.fetch_add(1, Ordering::SeqCst); + let (status, body) = replies[attempt.min(replies.len() - 1)].clone(); + async move { (status, axum::Json(body)) } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (port, requests, server) + } + + #[tokio::test] + async fn cdp_readiness_waits_for_a_page_socket_after_the_port_opens() { + use axum::http::StatusCode; + use std::sync::atomic::Ordering; + + let ready = json!([ + { "type": "page" }, + { "type": "page", "webSocketDebuggerUrl": "" }, + { "type": "service_worker", "webSocketDebuggerUrl": "ws://worker" }, + { "type": "page", "webSocketDebuggerUrl": "ws://ready-page" } + ]); + let (port, requests, server) = cdp_startup_server(vec![ + (StatusCode::OK, json!([])), + (StatusCode::SERVICE_UNAVAILABLE, ready.clone()), + (StatusCode::OK, json!([{ "type": "page" }])), + (StatusCode::OK, ready), + ]) + .await; + let mut child = cdp_test_process(); + let targets = tokio::time::timeout( + Duration::from_secs(5), + WayfernManager::new().wait_for_cdp_ready_while_running( + port, + &mut child, + &BrowserLogTap::default(), + ), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(requests.load(Ordering::SeqCst), 4); + assert_eq!(targets.len(), 1); + assert_eq!( + targets[0].websocket_debugger_url.as_deref(), + Some("ws://ready-page") + ); + child.kill().await.unwrap(); + server.abort(); + } + + #[tokio::test] + async fn cdp_page_wait_stops_when_the_browser_exits() { + use std::sync::atomic::Ordering; + + let (port, requests, server) = + cdp_startup_server(vec![(axum::http::StatusCode::OK, json!([]))]).await; + let mut child = cdp_test_process(); + let pid = child.id().unwrap(); + let stop = tokio::spawn(async move { + while requests.load(Ordering::SeqCst) == 0 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + force_kill_process(pid); + }); + let log = BrowserLogTap::default(); + log.push("Wayfern instance limit reached: 4 browsers are already running on this plan.".into()); + let error = tokio::time::timeout( + Duration::from_secs(5), + WayfernManager::new().wait_for_cdp_ready_while_running(port, &mut child, &log), + ) + .await + .unwrap() + .unwrap_err(); + assert_eq!( + code_of(&error.to_string()), + "WAYFERN_INSTANCE_LIMIT_REACHED" + ); + stop.await.unwrap(); + server.abort(); + } + #[test] fn identity_api_switch_follows_the_chromium_major() { // The profile version is the full Chromium version string. diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index 86517ce6..76b2c593 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Donut", - "version": "0.31.2", + "version": "0.31.3", "identifier": "com.donutbrowser", "build": { "beforeDevCommand": "pnpm copy-proxy-binary && pnpm dev", diff --git a/src-tauri/tests/donut_proxy_integration.rs b/src-tauri/tests/donut_proxy_integration.rs index ea96a473..1ada708c 100644 --- a/src-tauri/tests/donut_proxy_integration.rs +++ b/src-tauri/tests/donut_proxy_integration.rs @@ -104,6 +104,119 @@ struct ProxyTestTracker { binary_path: std::path::PathBuf, } +struct ProxyWorkerProcess { + child: std::process::Child, + config_id: String, +} + +impl Drop for ProxyWorkerProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + donutbrowser_lib::proxy_storage::delete_proxy_config(&self.config_id); + } +} + +#[tokio::test] +#[serial] +async fn test_worker_publishes_its_pid_with_readiness( +) -> Result<(), Box> { + use donutbrowser_lib::proxy_storage::{ + generate_proxy_id, get_proxy_config, save_proxy_config, ProxyConfig, + }; + + let binary_path = setup_test().await?; + let config_id = generate_proxy_id(); + let config = ProxyConfig::new(config_id.clone(), "DIRECT".to_string(), Some(0)) + .with_local_protocol(Some("socks5".to_string())); + save_proxy_config(&config).map_err(|error| error.to_string())?; + let worker = ProxyWorkerProcess { + child: std::process::Command::new(&binary_path) + .args(["proxy-worker", "start", "--id", &config_id]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn()?, + config_id: config_id.clone(), + }; + + let ready = tokio::time::timeout(Duration::from_secs(10), async { + loop { + if let Some(config) = get_proxy_config(&config_id) { + if config.local_url.is_some() { + break config; + } + } + sleep(Duration::from_millis(20)).await; + } + }) + .await?; + + assert_eq!(ready.pid, Some(worker.child.id())); + let port = ready.local_port.unwrap(); + assert_eq!(ready.local_url, Some(format!("socks5://127.0.0.1:{port}"))); + let mut stream = TcpStream::connect(("127.0.0.1", port)).await?; + stream.write_all(&[5, 1, 0]).await?; + let mut response = [0; 2]; + tokio::time::timeout(Duration::from_secs(2), stream.read_exact(&mut response)).await??; + assert_eq!(response, [5, 0]); + + let stopped = + TestUtils::execute_command(&binary_path, &["proxy", "stop", "--id", &config_id]).await?; + assert!(stopped.status.success()); + assert!(wait_for_port_closed(port, Duration::from_secs(5)).await); + assert!(get_proxy_config(&config_id).is_none()); + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_concurrent_proxy_starts_keep_worker_state( +) -> Result<(), Box> { + use donutbrowser_lib::proxy_storage::{get_proxy_config, is_process_running}; + + let binary_path = setup_test().await?; + let mut tracker = ProxyTestTracker::new(binary_path.clone()); + let results = futures_util::future::join_all((0..8).map(|_| { + TestUtils::execute_command( + &binary_path, + &["proxy", "start", "--local-protocol", "socks5"], + ) + })) + .await; + let mut failures = Vec::new(); + let mut workers = Vec::new(); + for result in results { + match result { + Ok(output) if output.status.success() => { + let config: Value = serde_json::from_slice(&output.stdout)?; + let id = config["id"].as_str().unwrap().to_string(); + tracker.track_proxy(id.clone()); + workers.push(id); + } + Ok(output) => failures.push(String::from_utf8_lossy(&output.stderr).into_owned()), + Err(error) => failures.push(error.to_string()), + } + } + assert!(failures.is_empty(), "{}", failures.join("\n")); + let mut pids = std::collections::HashSet::new(); + for id in workers { + let config = get_proxy_config(&id).unwrap(); + let pid = config.pid.unwrap(); + assert!(pids.insert(pid)); + assert!(is_process_running(pid)); + let port = config.local_port.unwrap(); + assert_eq!(config.local_url, Some(format!("socks5://127.0.0.1:{port}"))); + let mut stream = TcpStream::connect(("127.0.0.1", port)).await?; + stream.write_all(&[5, 1, 0]).await?; + let mut response = [0; 2]; + tokio::time::timeout(Duration::from_secs(2), stream.read_exact(&mut response)).await??; + assert_eq!(response, [5, 0]); + } + tracker.cleanup_all().await; + Ok(()) +} + impl ProxyTestTracker { fn new(binary_path: std::path::PathBuf) -> Self { Self { diff --git a/src/hooks/use-cookie-bot.ts b/src/hooks/use-cookie-bot.ts index 59ae1701..b064e52f 100644 --- a/src/hooks/use-cookie-bot.ts +++ b/src/hooks/use-cookie-bot.ts @@ -108,13 +108,25 @@ async function attachListeners() { // detach had already run, leaking a listener that no unsubscribe can reach // and double-emitting every session transition for the rest of the session. const generation = ++attachGeneration; + const recover = () => { + if (!enabled) return; + void attachStream(); + void refreshCookieBot(); + }; + const recoverOnFocus = () => { + if (document.visibilityState === "visible" && !snapshot.streamConnected) { + recover(); + } + }; const offs = await Promise.all([ onRemoteSessionState((session) => { if (!session.profile_id) return; const over = isSessionOver(session); const next = { ...snapshot.liveSessions }; if (over) { - delete next[session.profile_id]; + if (next[session.profile_id]?.session_id === session.session_id) { + delete next[session.profile_id]; + } } else { next[session.profile_id] = session; } @@ -125,11 +137,20 @@ async function attachListeners() { }), onRemoteSessionSnapshot((payload) => { emit({ liveSessions: indexOpenSessions(payload.sessions) }); + void refreshCookieBot(); }), onRemoteSessionStream((status) => { emit({ streamConnected: status.connected }); }), ]); + window.addEventListener("online", recover); + window.addEventListener("focus", recoverOnFocus); + document.addEventListener("visibilitychange", recoverOnFocus); + offs.push(() => { + window.removeEventListener("online", recover); + window.removeEventListener("focus", recoverOnFocus); + document.removeEventListener("visibilitychange", recoverOnFocus); + }); if (generation !== attachGeneration) { for (const off of offs) off(); return; diff --git a/src/hooks/use-team-locks.ts b/src/hooks/use-team-locks.ts index d53eeda0..338994bc 100644 --- a/src/hooks/use-team-locks.ts +++ b/src/hooks/use-team-locks.ts @@ -18,20 +18,13 @@ export function useTeamLocks(currentUserId?: string) { useEffect(() => { void fetchLocks(); - const unlistenAcquired = listen<{ profileId: string }>( - "team-lock-acquired", - () => void fetchLocks(), - ); - const unlistenReleased = listen<{ profileId: string }>( - "team-lock-released", - () => void fetchLocks(), - ); + const unlistenChanged = listen<{ + profileId: string | null; + action: string; + }>("profile-lock-changed", () => void fetchLocks()); return () => { - void unlistenAcquired.then((fn) => { - fn(); - }); - void unlistenReleased.then((fn) => { + void unlistenChanged.then((fn) => { fn(); }); };