refactor: cleanup

This commit is contained in:
zhom committed 2026-10-03 02:24:34 +04:00
1 parent 130cfda791
commit bddb5063ad
31 files changed
+2787 -399

No files matched your search

-16
View File
@@ -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:
+196
View File
@@ -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<string, string | undefined>): 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<string, string>; user?: UserContext };
} {
const request: { headers: Record<string, string>; user?: UserContext } = {
headers: { authorization: `Bearer ${token}` },
};
const context = {
switchToHttp: () => ({ getRequest: () => request }),
} as unknown as ExecutionContext;
return { context, request };
}
function sign(
payload: Record<string, unknown>,
privateKey: string = signingKeys.privateKey,
expiresIn = 300,
): string {
return jwt.sign(payload, privateKey, { algorithm: "RS256", expiresIn });
}
function syncToken(overrides: Record<string, unknown> = {}): 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,
});
});
});
+4
View File
@@ -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
+4 -4
View File
@@ -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",
+19 -1
View File
@@ -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"]');
+225 -1
View File
@@ -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");
+1 -1
View File
@@ -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",
+1 -1
View File
@@ -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",
+1 -1
View File
@@ -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"
+5 -7
View File
@@ -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;
+38 -5
View File
@@ -548,7 +548,7 @@ pub async fn update_recipe(
/// Delete one.
pub async fn delete_recipe(id: &str) -> Result<bool, String> {
let outcome: RecipeDeleted = request(
let outcome: Option<RecipeDeleted> = request(
reqwest::Method::DELETE,
format!("{}/recipes/{}", base(), urlencoding::encode(id)),
Vec::new(),
@@ -557,7 +557,11 @@ pub async fn delete_recipe(id: &str) -> Result<bool, String> {
)
.await
.map_err(|e| agent_error("recipe delete", e))?;
Ok(outcome.deleted.unwrap_or(true))
Ok(recipe_deleted(outcome))
}
fn recipe_deleted(outcome: Option<RecipeDeleted>) -> bool {
outcome.and_then(|reply| reply.deleted).unwrap_or(true)
}
#[derive(Debug, Default, Deserialize)]
@@ -925,16 +929,25 @@ async fn request<T: DeserializeOwned>(
return Err(format!("({status}) {text}"));
}
response
.json::<T>()
let bytes = response
.bytes()
.await
.map_err(|e| format!("decode response: {e}"))
.map_err(|e| format!("decode response: {e}"))?;
decode_body::<T>(&bytes).map_err(|e| format!("decode response: {e}"))
}
})
.await
.map_err(|e| AgentError(cloud_errors::classify_message(&e, codes)))
}
fn decode_body<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, serde_json::Error> {
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<String> {
mod tests {
use super::*;
#[test]
fn a_recipe_delete_with_no_body_is_a_success() {
let empty: Option<RecipeDeleted> = decode_body(b"").expect("a 204 has no body");
assert!(recipe_deleted(empty));
let blank: Option<RecipeDeleted> = decode_body(b" \n").expect("whitespace is no body");
assert!(recipe_deleted(blank));
let refused: Option<RecipeDeleted> =
decode_body(br#"{"deleted":false}"#).expect("an explicit answer");
assert!(!recipe_deleted(refused));
let bare: Option<RecipeDeleted> = 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::<RecipeDeleted>(b"").is_err());
assert!(decode_body::<AgentRecipe>(b"").is_err());
assert!(decode_body::<Option<RecipeDeleted>>(b"<html>").is_err());
}
#[test]
fn a_blank_goal_is_refused_before_anything_is_spent() {
for goal in ["", " ", "\n\t "] {
+1 -1
View File
@@ -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()
+416 -24
View File
@@ -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<Utc>) -> Option<Duration> {
let value = value.trim();
if let Ok(seconds) = value.parse::<u64>() {
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<Option<CloudAuthState>>,
refresh_lock: tokio::sync::Mutex<()>,
refresh_deferred_until: std::sync::Mutex<Option<Instant>>,
wayfern_token: Mutex<Option<String>>,
/// 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<i64> {
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<Duration> {
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<String, String> {
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(&current) {
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<String, String> {
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<String, String> {
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<F>(
&self,
build: F,
) -> Result<reqwest::Response, AuthorizedError>
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<Output = Result<T, String>> + 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::<i64>().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}");
}
}
}
+4
View File
@@ -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,
+348 -32
View File
@@ -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<Option<HashSet<String>>> = 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<CookieBotScheduleList
let query = scope
.map(|s| vec![("scope".to_string(), s.to_string())])
.unwrap_or_default();
request(
let list: CookieBotScheduleList = request(
reqwest::Method::GET,
format!("{}/schedules", base()),
query,
None,
SCHEDULE_CODES,
)
.await
.await?;
if scope.is_none_or(|requested| requested == "mine") {
record_own_schedules(&mut enrolled_profiles(), &list);
}
Ok(list)
}
/// This profile's enrolment, or `None` when there is none.
@@ -834,8 +843,14 @@ pub async fn get_schedule(profile_id: &str) -> Result<Option<CookieBotSchedule>,
.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<CookieBotSchedule, CookieBotError> {
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<BrowserProfile>,
) -> Vec<BrowserProfile> {
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<String>,
pub platform: Option<String>,
pub profile_name: Option<String>,
}
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<HashSet<String>>> {
ENROLLED_PROFILES
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn should_report(enrolled: &Option<HashSet<String>>, profile_id: &str) -> bool {
enrolled.as_ref().is_none_or(|ids| ids.contains(profile_id))
}
fn record_enrolment(enrolled: &mut Option<HashSet<String>>, 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<HashSet<String>>, 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<CookieBotScheduleDeleted, CookieBotError> {
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());
+4
View File
@@ -250,6 +250,10 @@ fn fingerprint_locale(profile: &BrowserProfile) -> (Option<String>, Option<Strin
(read("timezone"), read("language"))
}
pub(crate) fn fingerprint_timezone(profile: &BrowserProfile) -> Option<String> {
fingerprint_locale(profile).0
}
/// A mutex whose poison is not fatal.
///
/// A panic anywhere under this lock used to brick the check process-wide.
+5
View File
@@ -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")]
+38 -45
View File
@@ -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<bool, Box<dyn std::error::Error>> {
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)
+1
View File
@@ -1654,6 +1654,7 @@ pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box<dyn std::er
// Update config with actual port and local_url (scheme matches the protocol
// we serve, so the parent's readiness check and any consumer see the truth)
let mut updated_config = config.clone();
updated_config.pid = Some(std::process::id());
updated_config.local_port = Some(actual_port);
updated_config.local_url = Some(format!(
"{}://127.0.0.1:{}",
+232 -71
View File
@@ -31,28 +31,15 @@
use crate::log_redaction::ShortId;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::RwLock;
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;
use tokio::sync::Notify;
/// Emitted whenever the set of gated profiles changes.
pub const EVENT_REMOTE_HANDOFF: &str = "remote-handoff-changed";
/// Attempts at pulling a finished session's work before giving up for now.
///
/// The entry survives a failure, so "giving up" only means this burst stops;
/// the next stream event, app start or manual sync tries again. What the retries
/// buy is the common case: the profile lock is released a moment before this
/// machine's cached copy of it expires, and a single attempt would hit
/// `Skipped("profile is locked elsewhere")` and leave the user blocked for no
/// reason.
const PULL_ATTEMPTS: u32 = 5;
/// Delay before the second pull attempt. Doubles, capped by [`PULL_RETRY_MAX`].
const PULL_RETRY_BASE: Duration = Duration::from_secs(2);
/// Ceiling on the pull backoff. Above the 30s profile-lock refresh, so a run of
/// attempts is guaranteed to span at least one refresh of the lock cache.
const PULL_RETRY_MAX: Duration = Duration::from_secs(45);
const PULL_RETRY_BASE: Duration = Duration::from_secs(1);
const PULL_RETRY_MAX: Duration = Duration::from_secs(15);
/// Where a profile stands with respect to remote execution.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -77,6 +64,64 @@ struct HandoffEntry {
type Store = HashMap<String, HandoffEntry>;
static STORE: RwLock<Option<Store>> = RwLock::new(None);
static PULL_WORKERS: Mutex<Option<HashMap<String, Arc<Notify>>>> = Mutex::new(None);
fn with_workers<T>(f: impl FnOnce(&mut HashMap<String, Arc<Notify>>) -> 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<Notify>,
}
impl PullWorker {
fn claim(profile_id: String) -> Option<Self> {
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<String> {
// 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<HandoffState> {
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<String> {
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<String> = 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<F, Fut, E>(worker: PullWorker, mut pull: F, exists: E)
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<crate::sync::ProfileSyncOutcome, String>>,
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());
}
+87 -9
View File
@@ -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<String, RemoteSessionState> = 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<Option<tauri::async_runtime::JoinHandle<()>>> = 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<String> = 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::<serde_json::Value>(&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.
+131 -21
View File
@@ -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<TokioMutex<()>> {
static LOCKS: std::sync::OnceLock<StdMutex<HashMap<String, Weak<TokioMutex<()>>>>> =
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<StdMutex<HashMap<String, Arc<AtomicBool>>>> =
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<ProfileSyncOutcome> {
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 = &current_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
+33
View File
@@ -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 {
+16 -3
View File
@@ -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(
+224 -24
View File
@@ -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<Option<String>, 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<String> = 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,
&current,
&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(&current);
}
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<SyncWorkItem>,
running: &Arc<AtomicBool>,
_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<u64> = (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));
}
}
+415 -72
View File
@@ -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<String>,
}
#[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::<HeartbeatResponse>(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<Duration> {
RELEASE_RETRY_DELAYS.get(attempt).copied()
}
fn lock_signature(locks: &HashMap<String, ProfileLockInfo>) -> 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<Duration> {
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<Duration>) {
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<HashMap<String, ProfileLockInfo>>,
held: RwLock<HashSet<String>>,
heartbeat_handle: Mutex<Option<JoinHandle<()>>>,
connected: Mutex<bool>,
paused_until: std::sync::Mutex<Option<Instant>>,
}
pub static PROFILE_LOCK: std::sync::LazyLock<ProfileLockManager> =
@@ -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<String> {
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<String> = 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<ProfileLockInfo> {
@@ -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<ProfileLockInfo>) -> 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<Renewal, AuthorizedError> {
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<String> = {
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, "<html>"), 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<Duration> = (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.
+197 -46
View File
@@ -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<dyn std::error::Error + Send + Sync>> {
) -> Result<Vec<CdpTarget>, Box<dyn std::error::Error + Send + Sync>> {
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<Vec<CdpTarget>, Box<dyn std::error::Error + Send + Sync>> {
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<Vec<CdpTarget>, Box<dyn std::error::Error + Send + Sync>> {
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<CdpTarget> = 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<std::sync::atomic::AtomicUsize>,
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.
+1 -1
View File
@@ -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",
+113
View File
@@ -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<dyn std::error::Error + Send + Sync>> {
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<dyn std::error::Error + Send + Sync>> {
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 {
+22 -1
View File
@@ -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;
+5 -12
View File
@@ -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();
});
};