refactor: cleanup sync

This commit is contained in:
zhom
2026-08-10 15:47:31 +04:00
parent 929f5a0ead
commit 325d8fae31
16 changed files with 438 additions and 34 deletions
+9
View File
@@ -13,3 +13,12 @@ S3_ACCESS_KEY_ID=CHANGE_ME
S3_SECRET_ACCESS_KEY=CHANGE_ME
S3_BUCKET=donut-sync
S3_FORCE_PATH_STYLE=true
# The address Donut Browser is sent to for file transfers. Set this whenever
# S3_ENDPOINT is only reachable from the server — running MinIO in the same
# compose file makes S3_ENDPOINT a container name like http://minio:9000, which
# resolves on the container network and nowhere else. Presigned URLs are signed
# against the host they name, so leaving this unset there hands every client a
# URL it cannot open: /health and /readyz stay green while every transfer fails.
# Defaults to S3_ENDPOINT, which is correct when storage is already public.
# S3_PUBLIC_ENDPOINT=https://storage.example.com
+13 -3
View File
@@ -19,15 +19,25 @@ export class AppController {
return { status: "ok" };
}
// `storageEndpoint` is the host clients are handed in presigned URLs. The
// server cannot tell whether a client can reach it, so report it and let
// whoever is debugging a failing sync compare it against their network.
// Self-hosted only — see getDiagnosticStorageEndpoint.
@Get("readyz")
async getReadiness(): Promise<{ status: string; s3: boolean }> {
async getReadiness(): Promise<{
status: string;
s3: boolean;
storageEndpoint?: string;
}> {
const s3Ready = await this.syncService.checkS3Connectivity();
const storageEndpoint = this.syncService.getDiagnosticStorageEndpoint();
const diagnostic = storageEndpoint ? { storageEndpoint } : {};
if (!s3Ready) {
throw new HttpException(
{ status: "not ready", s3: false },
{ status: "not ready", s3: false, ...diagnostic },
HttpStatus.SERVICE_UNAVAILABLE,
);
}
return { status: "ready", s3: true };
return { status: "ready", s3: true, ...diagnostic };
}
}
+79 -8
View File
@@ -82,6 +82,10 @@ function sanitizeMetadata(
export class SyncService implements OnModuleInit {
private readonly logger = new Logger(SyncService.name);
private s3Client: S3Client;
// Signs the URLs handed to clients. Same instance as `s3Client` unless
// `S3_PUBLIC_ENDPOINT` names a different, client-reachable address.
private presignClient: S3Client;
private publicEndpoint: string;
private bucket: string;
// Upper bound on presign batch array length (DoS guard).
private static readonly MAX_BATCH_ITEMS = 1000;
@@ -112,16 +116,34 @@ export class SyncService implements OnModuleInit {
this.bucket = requireEnv("S3_BUCKET");
const credentials = { accessKeyId, secretAccessKey };
this.s3Client = new S3Client({
endpoint,
region,
credentials: {
accessKeyId,
secretAccessKey,
},
credentials,
forcePathStyle,
});
// Presigned URLs are handed to a desktop client on another machine, so they
// must name a host that client can reach. `S3_ENDPOINT` is often reachable
// only from the server: the documented compose file points it at
// `http://minio:9000`, a Docker service name that resolves on the compose
// network and nowhere else. Signing is bound to the host, so the presign
// client is a second client pinned to the public address rather than a
// string rewrite of the signed URL.
const publicEndpoint =
this.configService.get<string>("S3_PUBLIC_ENDPOINT") || endpoint;
this.publicEndpoint = publicEndpoint;
this.presignClient =
publicEndpoint === endpoint
? this.s3Client
: new S3Client({
endpoint: publicEndpoint,
region,
credentials,
forcePathStyle,
});
this.backendInternalUrl = this.configService.get<string>(
"BACKEND_INTERNAL_URL",
);
@@ -132,6 +154,51 @@ export class SyncService implements OnModuleInit {
async onModuleInit() {
await this.ensureBucketExists();
this.warnIfPresignEndpointIsServerOnly();
}
/**
* The address clients are sent to for object transfers, for `/readyz` to
* report when a self-hoster is debugging a failing sync.
*
* Withheld in cloud mode: `/readyz` is unauthenticated, and a managed
* deployment should not publish its storage host to anyone who can reach the
* probe. Self-hosters own both ends, and the value is the whole point of the
* diagnostic there.
*/
getDiagnosticStorageEndpoint(): string | undefined {
const isCloud = Boolean(
this.configService.get<string>("SYNC_JWT_PUBLIC_KEY"),
);
return isCloud ? undefined : this.publicEndpoint;
}
/**
* A single-label host (`minio`, `s3`) only resolves inside the container
* network, so every presigned URL built from it is unreachable for the
* desktop client even though the server's own S3 calls succeed. That failure
* shows up as healthy `/health` and `/readyz` with every file transfer
* failing at connect, which is near-impossible to diagnose from the client.
* Say it once at boot instead.
*/
private warnIfPresignEndpointIsServerOnly(): void {
let host: string;
try {
host = new URL(this.publicEndpoint).hostname;
} catch {
return;
}
const isSingleLabel =
!host.includes(".") && !host.includes(":") && host !== "localhost";
if (!isSingleLabel) return;
this.logger.warn(
`Storage endpoint '${this.publicEndpoint}' uses the container-only host '${host}'. ` +
"Presigned URLs built from it cannot be reached by Donut Browser, so every " +
"transfer will fail while /health and /readyz stay green. Set S3_PUBLIC_ENDPOINT " +
"to an address your devices can reach (and publish that port).",
);
}
private async ensureBucketExists(): Promise<void> {
@@ -332,7 +399,7 @@ export class SyncService implements OnModuleInit {
const metadataHeaders = new Set(
Object.keys(metadata ?? {}).map((name) => `x-amz-meta-${name}`),
);
const url = await getSignedUrl(this.s3Client, command, {
const url = await getSignedUrl(this.presignClient, command, {
expiresIn,
// The AWS presigner otherwise hoists user metadata into the query string.
// The client echoes the response metadata as headers, so those headers
@@ -374,7 +441,7 @@ export class SyncService implements OnModuleInit {
Key: key,
});
const url = await getSignedUrl(this.s3Client, command, { expiresIn });
const url = await getSignedUrl(this.presignClient, command, { expiresIn });
return {
url,
@@ -505,7 +572,9 @@ export class SyncService implements OnModuleInit {
ContentType: item.contentType || "application/octet-stream",
});
const url = await getSignedUrl(this.s3Client, command, { expiresIn });
const url = await getSignedUrl(this.presignClient, command, {
expiresIn,
});
return {
key: item.key,
@@ -565,7 +634,9 @@ export class SyncService implements OnModuleInit {
Key: key,
});
const url = await getSignedUrl(this.s3Client, command, { expiresIn });
const url = await getSignedUrl(this.presignClient, command, {
expiresIn,
});
return {
key: rawKey,
@@ -0,0 +1,201 @@
import { INestApplication } from "@nestjs/common";
import { ConfigModule } from "@nestjs/config";
import { Test, TestingModule } from "@nestjs/testing";
import request from "supertest";
import { App } from "supertest/types";
import { AppController } from "./../src/app.controller.js";
import { AppService } from "./../src/app.service.js";
import { SyncModule } from "./../src/sync/sync.module.js";
import {
configureTestEnv,
TEST_S3_ENDPOINT,
TEST_SYNC_TOKEN,
waitForTestS3,
} from "./test-env.js";
// Presigning is offline, so this host never has to accept a connection — the
// assertions are about which host ends up in the signed URL.
const PUBLIC_ENDPOINT = "https://storage.example.com";
// Only needs to be present for the server to consider itself cloud-mode; no
// token is verified against it in these assertions.
const CLOUD_PUBLIC_KEY =
"-----BEGIN PUBLIC KEY-----\nnot-a-real-key\n-----END PUBLIC KEY-----";
interface PresignResponse {
url: string;
}
interface PresignBatchResponse {
items: Array<{ key: string; url: string }>;
}
interface ReadyResponse {
status: string;
s3: boolean;
storageEndpoint: string;
}
async function bootstrap(publicEndpoint: string | undefined) {
configureTestEnv();
if (publicEndpoint) {
process.env.S3_PUBLIC_ENDPOINT = publicEndpoint;
} else {
delete process.env.S3_PUBLIC_ENDPOINT;
}
await waitForTestS3();
const moduleFixture: TestingModule = await Test.createTestingModule({
imports: [ConfigModule.forRoot({ isGlobal: true }), SyncModule],
controllers: [AppController],
providers: [AppService],
}).compile();
const app = moduleFixture.createNestApplication<INestApplication<App>>();
await app.listen(0);
return app;
}
// A self-hosted server usually reaches its storage over a private address the
// desktop client has no route to. Signing client URLs against that address
// handed every client a URL it could not open, so uploads failed at connect
// while /health and /readyz stayed green.
describe("presigned URL host", () => {
describe("with S3_PUBLIC_ENDPOINT set", () => {
let app: INestApplication<App>;
beforeAll(async () => {
app = await bootstrap(PUBLIC_ENDPOINT);
});
afterAll(async () => {
delete process.env.S3_PUBLIC_ENDPOINT;
await app.close();
});
it("signs single upload URLs against the public endpoint", async () => {
const response = await request(app.getHttpServer())
.post("/v1/objects/presign-upload")
.set("Authorization", `Bearer ${TEST_SYNC_TOKEN}`)
.send({ key: "endpoint/single.txt" })
.expect(200);
const { url } = response.body as PresignResponse;
expect(url.startsWith(PUBLIC_ENDPOINT)).toBe(true);
expect(url).not.toContain(TEST_S3_ENDPOINT);
});
it("signs batch upload URLs against the public endpoint", async () => {
const response = await request(app.getHttpServer())
.post("/v1/objects/presign-upload-batch")
.set("Authorization", `Bearer ${TEST_SYNC_TOKEN}`)
.send({ items: [{ key: "endpoint/a.txt" }, { key: "endpoint/b.txt" }] })
.expect(200);
const { items } = response.body as PresignBatchResponse;
expect(items).toHaveLength(2);
for (const item of items) {
expect(item.url.startsWith(PUBLIC_ENDPOINT)).toBe(true);
}
});
it("signs download URLs against the public endpoint", async () => {
const response = await request(app.getHttpServer())
.post("/v1/objects/presign-download")
.set("Authorization", `Bearer ${TEST_SYNC_TOKEN}`)
.send({ key: "endpoint/single.txt" })
.expect(200);
const { url } = response.body as PresignResponse;
expect(url.startsWith(PUBLIC_ENDPOINT)).toBe(true);
});
// The server's own S3 calls must keep using the private endpoint, or
// pointing clients at a public address would break the server itself.
it("still reaches storage over the private endpoint", async () => {
const response = await request(app.getHttpServer())
.post("/v1/objects/stat")
.set("Authorization", `Bearer ${TEST_SYNC_TOKEN}`)
.send({ key: "endpoint/does-not-exist" })
.expect(200);
expect(response.body).toEqual({ exists: false });
});
it("reports the client-facing endpoint from /readyz", async () => {
const response = await request(app.getHttpServer())
.get("/readyz")
.expect(200);
const body = response.body as ReadyResponse;
expect(body.s3).toBe(true);
expect(body.storageEndpoint).toBe(PUBLIC_ENDPOINT);
});
});
// /readyz has no auth, so a managed deployment must not publish its storage
// host to anyone who can reach the probe.
describe("in cloud mode", () => {
let app: INestApplication<App>;
const previousKey = process.env.SYNC_JWT_PUBLIC_KEY;
beforeAll(async () => {
process.env.SYNC_JWT_PUBLIC_KEY = CLOUD_PUBLIC_KEY;
app = await bootstrap(PUBLIC_ENDPOINT);
});
afterAll(async () => {
if (previousKey === undefined) {
delete process.env.SYNC_JWT_PUBLIC_KEY;
} else {
process.env.SYNC_JWT_PUBLIC_KEY = previousKey;
}
delete process.env.S3_PUBLIC_ENDPOINT;
await app.close();
});
it("withholds the storage endpoint from /readyz", async () => {
const response = await request(app.getHttpServer())
.get("/readyz")
.expect(200);
const body = response.body as ReadyResponse;
expect(body.s3).toBe(true);
expect(body.storageEndpoint).toBeUndefined();
expect(JSON.stringify(body)).not.toContain("storage.example.com");
});
});
describe("without S3_PUBLIC_ENDPOINT", () => {
let app: INestApplication<App>;
beforeAll(async () => {
app = await bootstrap(undefined);
});
afterAll(async () => {
await app.close();
});
it("falls back to S3_ENDPOINT", async () => {
const response = await request(app.getHttpServer())
.post("/v1/objects/presign-upload")
.set("Authorization", `Bearer ${TEST_SYNC_TOKEN}`)
.send({ key: "endpoint/fallback.txt" })
.expect(200);
const { url } = response.body as PresignResponse;
expect(url.startsWith(TEST_S3_ENDPOINT)).toBe(true);
});
it("reports the fallback endpoint from /readyz", async () => {
const response = await request(app.getHttpServer())
.get("/readyz")
.expect(200);
expect((response.body as ReadyResponse).storageEndpoint).toBe(
TEST_S3_ENDPOINT,
);
});
});
});
+71 -8
View File
@@ -109,6 +109,37 @@ fn is_critical_file(path: &str) -> bool {
.any(|pattern| path.contains(pattern))
}
/// How many failed paths to name before collapsing the rest into a count.
const MAX_LISTED_FAILURES: usize = 10;
/// Aggregate a batch of failed transfers into the message the user sees.
///
/// Whatever breaks a sync usually breaks every file the same way — one
/// unreachable storage host, one rejected signature — so the per-file causes
/// were dropped and only the paths survived into the message. That left users
/// staring at a list of filenames with nothing to act on. Carry the first
/// cause through, and stop pasting hundreds of paths into a toast.
fn critical_failure_message(action: &str, failures: &[(String, String)]) -> String {
let listed: Vec<&str> = failures
.iter()
.take(MAX_LISTED_FAILURES)
.map(|(path, _)| path.as_str())
.collect();
let hidden = failures.len().saturating_sub(listed.len());
let files = if hidden > 0 {
format!("{} (and {} more)", listed.join(", "), hidden)
} else {
listed.join(", ")
};
match failures.first() {
Some((_, cause)) => format!(
"Critical files failed to {action}: {files}. Cause: {cause}. Sync aborted to prevent data loss."
),
None => format!("Critical files failed to {action}: {files}. Sync aborted to prevent data loss."),
}
}
/// Validate that a manifest-supplied relative file path is safe to join onto a
/// profile directory before writing/deleting. The manifest is remote-controlled
/// (a self-hosted or compromised sync server, a MITM on a plaintext Regular-mode
@@ -1283,10 +1314,9 @@ impl SyncEngine {
}
if !critical_failures.is_empty() {
let file_list: Vec<&str> = critical_failures.iter().map(|(p, _)| p.as_str()).collect();
return Err(SyncError::IoError(format!(
"Critical files failed to upload: {}. Sync aborted to prevent data loss.",
file_list.join(", ")
return Err(SyncError::IoError(critical_failure_message(
"upload",
&critical_failures,
)));
}
@@ -1559,10 +1589,9 @@ impl SyncEngine {
}
if !critical_failures.is_empty() {
let file_list: Vec<&str> = critical_failures.iter().map(|(p, _)| p.as_str()).collect();
return Err(SyncError::IoError(format!(
"Critical files failed to download: {}. Sync aborted to prevent data loss.",
file_list.join(", ")
return Err(SyncError::IoError(critical_failure_message(
"download",
&critical_failures,
)));
}
@@ -4247,6 +4276,40 @@ pub async fn rollover_encryption_for_all_entities(
mod tests {
use super::*;
#[test]
fn test_critical_failure_message_carries_the_cause() {
// A self-hosted server that hands out unreachable presigned URLs fails
// every file with the same connect error. Naming only the files told the
// user nothing about why, which is what made this undiagnosable.
let failures = vec![
(
"Default/Cookies".to_string(),
"Failed to upload Default/Cookies after 3 retries: error sending request".to_string(),
),
("Local State".to_string(), "same".to_string()),
];
let message = critical_failure_message("upload", &failures);
assert!(message.contains("Default/Cookies"));
assert!(message.contains("Local State"));
assert!(message.contains("Cause: Failed to upload Default/Cookies"));
assert!(message.contains("Sync aborted to prevent data loss."));
}
#[test]
fn test_critical_failure_message_collapses_long_lists() {
let failures: Vec<(String, String)> = (0..25)
.map(|i| (format!("file-{i}"), "connect error".to_string()))
.collect();
let message = critical_failure_message("download", &failures);
assert!(message.contains("file-0"));
assert!(message.contains(&format!("file-{}", MAX_LISTED_FAILURES - 1)));
assert!(!message.contains(&format!("file-{MAX_LISTED_FAILURES}")));
assert!(message.contains("(and 15 more)"));
assert!(message.contains("failed to download"));
}
#[test]
fn test_is_safe_manifest_path() {
// Legitimate profile-relative paths are accepted.
+55 -15
View File
@@ -80,19 +80,49 @@ export function SyncConfigDialog({
const [connectionStatus, setConnectionStatus] = useState<
"unknown" | "testing" | "connected" | "error"
>("unknown");
const [storageEndpoint, setStorageEndpoint] = useState<string | null>(null);
const hasConfig = Boolean(serverUrl && token);
const testConnection = useCallback(async (url: string) => {
setConnectionStatus("testing");
try {
const healthUrl = `${url.replace(/\/$/, "")}/health`;
const response = await fetch(healthUrl);
setConnectionStatus(response.ok ? "connected" : "error");
} catch {
setConnectionStatus("error");
// `/health` is a bare liveness probe: it answers ok on a server whose storage
// is unreachable or misconfigured, which is how a green "connected" could sit
// next to a sync where every single file failed. `/readyz` checks storage and
// reports the endpoint clients are handed in presigned URLs, so surface that
// too — when transfers fail, it is the value worth checking first.
const probeServer = useCallback(async (url: string) => {
const base = url.replace(/\/$/, "");
const response = await fetch(`${base}/readyz`);
// A server old enough to predate /readyz is still a working server, so
// fall back rather than reporting a healthy setup as broken.
if (response.status === 404) {
const health = await fetch(`${base}/health`);
return { ok: health.ok, storageEndpoint: undefined };
}
if (!response.ok) {
return { ok: false as const, storageEndpoint: undefined };
}
const body = (await response.json()) as {
storageEndpoint?: string;
} | null;
return { ok: true as const, storageEndpoint: body?.storageEndpoint };
}, []);
const testConnection = useCallback(
async (url: string) => {
setConnectionStatus("testing");
try {
const result = await probeServer(url);
setStorageEndpoint(result.storageEndpoint ?? null);
setConnectionStatus(result.ok ? "connected" : "error");
} catch {
setStorageEndpoint(null);
setConnectionStatus("error");
}
},
[probeServer],
);
const loadSettings = useCallback(async () => {
setIsLoading(true);
try {
@@ -142,9 +172,9 @@ export function SyncConfigDialog({
setIsTesting(true);
setConnectionStatus("testing");
try {
const healthUrl = `${serverUrl.replace(/\/$/, "")}/health`;
const response = await fetch(healthUrl);
if (response.ok) {
const result = await probeServer(serverUrl);
setStorageEndpoint(result.storageEndpoint ?? null);
if (result.ok) {
setConnectionStatus("connected");
showSuccessToast(t("sync.config.connectionSuccess"));
} else {
@@ -152,12 +182,13 @@ export function SyncConfigDialog({
showErrorToast(t("sync.config.serverError"));
}
} catch {
setStorageEndpoint(null);
setConnectionStatus("error");
showErrorToast(t("sync.config.connectFailed"));
} finally {
setIsTesting(false);
}
}, [serverUrl, t]);
}, [serverUrl, t, probeServer]);
const handleSave = useCallback(async () => {
setIsSaving(true);
@@ -440,9 +471,18 @@ export function SyncConfigDialog({
</div>
)}
{connectionStatus === "connected" && (
<div className="flex items-center gap-2 text-sm text-muted-foreground">
<div className="size-2 rounded-full bg-success" />
{t("sync.status.connected")}
<div className="flex flex-col gap-1">
<div className="flex items-center gap-2 text-sm text-muted-foreground">
<div className="size-2 rounded-full bg-success" />
{t("sync.status.connected")}
</div>
{storageEndpoint && (
<span className="text-xs text-muted-foreground break-all">
{t("sync.config.storageEndpoint", {
endpoint: storageEndpoint,
})}
</span>
)}
</div>
)}
{connectionStatus === "error" && (
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "Connection successful!",
"serverError": "Server responded with an error",
"connectFailed": "Failed to connect to server",
"storageEndpoint": "Storage: {{endpoint}}",
"settingsSaved": "Sync settings saved",
"saveFailed": "Failed to save settings",
"disconnected": "Sync disconnected",
+1
View File
@@ -636,6 +636,7 @@
"connectionSuccess": "¡Conexión exitosa!",
"serverError": "El servidor respondió con un error",
"connectFailed": "Error al conectar con el servidor",
"storageEndpoint": "Almacenamiento: {{endpoint}}",
"settingsSaved": "Ajustes de sincronización guardados",
"saveFailed": "Error al guardar los ajustes",
"disconnected": "Sincronización desconectada",
+1
View File
@@ -636,6 +636,7 @@
"connectionSuccess": "Connexion réussie !",
"serverError": "Le serveur a répondu avec une erreur",
"connectFailed": "Échec de la connexion au serveur",
"storageEndpoint": "Stockage : {{endpoint}}",
"settingsSaved": "Paramètres de synchronisation enregistrés",
"saveFailed": "Échec de lenregistrement des paramètres",
"disconnected": "Synchronisation déconnectée",
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "接続に成功しました!",
"serverError": "サーバーがエラーで応答しました",
"connectFailed": "サーバーへの接続に失敗しました",
"storageEndpoint": "ストレージ: {{endpoint}}",
"settingsSaved": "同期設定を保存しました",
"saveFailed": "設定の保存に失敗しました",
"disconnected": "同期を切断しました",
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "연결 성공!",
"serverError": "서버가 오류로 응답했습니다",
"connectFailed": "서버에 연결하지 못했습니다",
"storageEndpoint": "스토리지: {{endpoint}}",
"settingsSaved": "동기화 설정이 저장되었습니다",
"saveFailed": "설정 저장 실패",
"disconnected": "동기화 연결 끊김",
+1
View File
@@ -636,6 +636,7 @@
"connectionSuccess": "Conexão bem-sucedida!",
"serverError": "O servidor respondeu com um erro",
"connectFailed": "Falha ao conectar ao servidor",
"storageEndpoint": "Armazenamento: {{endpoint}}",
"settingsSaved": "Configurações de sincronização salvas",
"saveFailed": "Falha ao salvar as configurações",
"disconnected": "Sincronização desconectada",
+1
View File
@@ -637,6 +637,7 @@
"connectionSuccess": "Подключение успешно!",
"serverError": "Сервер вернул ошибку",
"connectFailed": "Не удалось подключиться к серверу",
"storageEndpoint": "Хранилище: {{endpoint}}",
"settingsSaved": "Настройки синхронизации сохранены",
"saveFailed": "Не удалось сохранить настройки",
"disconnected": "Синхронизация отключена",
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "Bağlantı başarılı!",
"serverError": "Sunucu bir hatayla yanıt verdi",
"connectFailed": "Sunucuya bağlanılamadı",
"storageEndpoint": "Depolama: {{endpoint}}",
"settingsSaved": "Eşitleme ayarları kaydedildi",
"saveFailed": "Ayarlar kaydedilemedi",
"disconnected": "Eşitleme bağlantısı kesildi",
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "Kết nối thành công!",
"serverError": "Máy chủ trả về lỗi",
"connectFailed": "Kết nối máy chủ thất bại",
"storageEndpoint": "Bộ nhớ: {{endpoint}}",
"settingsSaved": "Đã lưu cài đặt đồng bộ",
"saveFailed": "Lưu cài đặt thất bại",
"disconnected": "Đã ngắt kết nối đồng bộ",
+1
View File
@@ -635,6 +635,7 @@
"connectionSuccess": "连接成功!",
"serverError": "服务器返回了错误",
"connectFailed": "连接服务器失败",
"storageEndpoint": "存储: {{endpoint}}",
"settingsSaved": "同步设置已保存",
"saveFailed": "保存设置失败",
"disconnected": "已断开同步",