diff --git a/Cargo.lock b/Cargo.lock index f9224567..35d51c94 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3991,6 +3991,7 @@ dependencies = [ "thiserror 2.0.20", "tokio", "tracing", + "tw-api", "tw-bedrock", "tw-config", "tw-pack", diff --git a/bin/twcore/src/sync_cmd.rs b/bin/twcore/src/sync_cmd.rs index a38cb948..81741e8e 100644 --- a/bin/twcore/src/sync_cmd.rs +++ b/bin/twcore/src/sync_cmd.rs @@ -93,6 +93,9 @@ fn print_view(v: &tw_api::SyncView) { if let Some(e) = &v.error { println!("problem: {}", e.text); } + for n in &v.notices { + println!("note: {}", n.text); + } if v.conflicts > 0 { println!( "conflicts: {} (resolve them in ThinkWatch Lite, Settings → Sync and backup)", @@ -242,13 +245,18 @@ fn decide(row: &tw_api::ImportRow) -> Result { name => format!("{} `{name}`", row.item.kind.slug()), }; eprintln!("{what}{differs} differs between this device and the sync data:"); - eprintln!(" u use the sync data (default)"); + eprintln!(" u use the sync data (default): this device takes the other devices' version"); eprintln!( - " k keep this device's version; it is uploaded and replaces the other devices' one" + " k keep this device's version: it is uploaded with this join and replaces the \ + other devices' version" ); let rename = row.suggested_name.clone().filter(|_| !row.global); if let Some(n) = &rename { - eprintln!(" r keep both: the sync data's one is added as {n}"); + eprintln!( + " r keep both: the sync data's version is added here as {n}; this device's \ + version keeps the name and is uploaded with this join, replacing the item of \ + that name on the other devices (they get {n} too)" + ); } loop { let a = ask("Choice", Some("u"))?; diff --git a/crates/tw-api/msg-codes.txt b/crates/tw-api/msg-codes.txt index 95d160d9..1b7a7fb4 100644 --- a/crates/tw-api/msg-codes.txt +++ b/crates/tw-api/msg-codes.txt @@ -586,6 +586,7 @@ sync.passphrase_changed sync.passphrase_short sync.passphrase_wrong sync.place_changed +sync.prefs_too_large sync.proxy_unusable sync.s3.bad_listing sync.s3.sign @@ -611,6 +612,7 @@ t.boom test t.broke test t.limited test t.plugin_threw test +t.prefs test t.upstream test t.x test yaml.anchor_or_alias diff --git a/crates/tw-api/src/lib.rs b/crates/tw-api/src/lib.rs index 762163e5..b59e6d4e 100644 --- a/crates/tw-api/src/lib.rs +++ b/crates/tw-api/src/lib.rs @@ -996,7 +996,8 @@ pub const MSG_CODES: &str = include_str!("../msg-codes.txt"); /// `POST /sync/disable`([`SyncDisable`])、`POST /sync/republish`(同步位置空了、keyring 不见了 /// 之后由人选的「重新上传本机配置」,→ [`SyncView`])。[`SyncState`] 里要人处理的有 /// `keyring_missing`、`emptied`、`misconfigured`,[`SyncView`] 的 `skipped` 是同步位置上跳过了的 -/// 文件数,[`SyncConflict`] 的 `options` 是能选的处理,[`SyncJoinPreview`] 的 `auto` 是加入时 +/// 文件数,`notices` 是同步成了但没照办的地方(`sync.prefs_too_large`),[`SyncConflict`] 的 +/// `options` 是能选的处理,[`SyncJoinPreview`] 的 `auto` 是加入时 /// 不用选的项(两边的默认网关密钥同名、值不同时采用同步数据的值)。状态变了发 [`Event::SyncStatus`];应用别的设备的 /// 版本时配置的来源是 `sync`,改了网关密钥时发 `GatewayKeysChanged`。[`ProviderView`] 多了 /// `sync_excluded`(不同步的上游)和 `sync_url_changed`(地址经同步变更,改之前的地址)。 diff --git a/crates/tw-api/src/sync.rs b/crates/tw-api/src/sync.rs index 1e95e604..82f6ae0d 100644 --- a/crates/tw-api/src/sync.rs +++ b/crates/tw-api/src/sync.rs @@ -136,6 +136,9 @@ pub struct SyncView { /// 同步位置上跳过了的文件:坏了的、不是这把密钥加密的、加密的位置上没加密的、等了 /// 30 分钟还没传完的。不是 0 时界面提一句 pub skipped: u32, + /// 同步成了,但有没照办的地方(上一轮的):界面提一句,**不是失败**,不计入连不上。现在 + /// 只有 `sync.prefs_too_large`:同步数据里的通用偏好超出上限,这台留着自己的,其余照常 + pub notices: Vec, } /// `POST /sync/test`、`POST /sync/enable/preview`:连上去看看。 diff --git a/crates/tw-control/src/sync.rs b/crates/tw-control/src/sync.rs index e0db8a45..ef29ecc7 100644 --- a/crates/tw-control/src/sync.rs +++ b/crates/tw-control/src/sync.rs @@ -132,8 +132,14 @@ async fn apply_pack( e, ) })?; - // 远程连接下偏好在 Lite 那边;同步这一路跑在 core 自己的任务里,不是远程连接 - let prefs_changed = after.client_prefs.is_some() && after.client_prefs != before.client_prefs; + // 远程连接下偏好在 Lite 那边;同步这一路跑在 core 自己的任务里,不是远程连接。太大的偏好 + // 不写(引擎已经不会给,也报了 `sync.prefs_too_large`):其余照常,不让整轮失败 + let prefs_changed = after.client_prefs.is_some() + && after.client_prefs != before.client_prefs + && !after + .client_prefs + .as_ref() + .is_some_and(tw_pack::disk::prefs_too_large); if prefs_changed && let Err(e) = tw_pack::disk::write_prefs(&path, after.client_prefs.as_ref()) { undo.restore(); @@ -613,6 +619,7 @@ impl Sync { excluded: Vec::new(), excluded_cascade: Vec::new(), skipped: 0, + notices: Vec::new(), }; let Some(mgr) = self.mgr() else { return off }; let cfg = mgr.gateway().config(); @@ -640,6 +647,11 @@ impl Sync { .map(|r| r.skipped) .or(prev.as_ref().map(|p| p.skipped)) .unwrap_or(0); + let notices = report + .as_ref() + .map(|r| r.notices.clone()) + .or(prev.as_ref().map(|p| p.notices.clone())) + .unwrap_or_default(); let devices = match &report { Some(r) => r .devices @@ -694,6 +706,7 @@ impl Sync { excluded: sync.exclude.clone(), excluded_cascade: cascade, skipped, + notices, } } diff --git a/crates/tw-control/tests/sync_regressions.rs b/crates/tw-control/tests/sync_regressions.rs index 7e8ca29e..74490dc4 100644 --- a/crates/tw-control/tests/sync_regressions.rs +++ b/crates/tw-control/tests/sync_regressions.rs @@ -449,3 +449,64 @@ async fn a_fresh_device_with_clients_set_up_adopts_the_default_key() { a.ok("POST", "/sync/now", Value::Null).await; assert_eq!(a.config().clients[0].key, "tw-default-a"); } + +/// 同步位置上来了一份超出上限的通用偏好(别处写进来的):这一轮不失败,偏好不应用、其余照常, +/// 视图里如实说(`notices`),不是错误、也不计入连不上 +#[tokio::test] +async fn oversized_preferences_in_the_sync_data_do_not_fail_the_round() { + let loc = tempfile::tempdir().unwrap(); + let (a, b) = (Dev::new('a'), Dev::new('b')); + a.enable(loc.path(), false, "A", Value::Null).await; + b.enable(loc.path(), false, "B", Value::Null).await; + // 依据最新的那个版本,写一个带着很大偏好的版本进去 + let mut names: Vec = std::fs::read_dir(loc.path().join("versions")) + .unwrap() + .map(|e| e.unwrap().file_name().to_string_lossy().to_string()) + .collect(); + names.sort(); + let head = names + .last() + .unwrap() + .trim_end_matches(".twconfig") + .to_string(); + let seq: u64 = head[..10].parse().unwrap(); + let text = add_raw( + &tw_pack::read( + &std::fs::read(loc.path().join("versions").join(format!("{head}.twconfig"))).unwrap(), + tw_pack::Unseal::None, + ) + .unwrap() + .config_yaml, + " - name: elsewhere\n base_url: https://e.example\n key: sk-e\n", + ); + let mut pack = tw_pack::Pack { + manifest: tw_pack::Manifest::new(tw_pack::PackKind::Full, tw_pack::PackOrigin::Sync), + config_yaml: text, + plugins: Default::default(), + client_prefs: Some(json!({ "notes": "x".repeat(70 * 1024) })), + }; + let meta = tw_sync::record::Meta { + seq: seq + 1, + device: "eeeeeeee-0000-4000-8000-00000000000e".into(), + device_name: "Elsewhere".into(), + parents: vec![head], + sha256: Some(tw_sync::engine::content_sha256(&pack)), + ..Default::default() + }; + pack.manifest.sync = Some(serde_json::to_value(&meta).unwrap()); + std::fs::write( + loc.path() + .join("versions") + .join(format!("{:010}-eeeeeeee.twconfig", seq + 1)), + tw_pack::write(&pack, tw_pack::Seal::None).unwrap(), + ) + .unwrap(); + for _ in 0..2 { + let v = b.ok("POST", "/sync/now", Value::Null).await; + assert_eq!(v["state"], "idle", "{v}"); + assert!(v["error"].is_null(), "{v}"); + assert!(v["failing_since_ms"].is_null(), "{v}"); + assert_eq!(v["notices"][0]["code"], "sync.prefs_too_large", "{v}"); + } + assert!(b.provider("elsewhere").is_some(), "其余照常应用"); +} diff --git a/crates/tw-sync/Cargo.toml b/crates/tw-sync/Cargo.toml index 085d9243..065e06bc 100644 --- a/crates/tw-sync/Cargo.toml +++ b/crates/tw-sync/Cargo.toml @@ -14,6 +14,8 @@ description = "Multi-device configuration sync: folder, WebDAV and S3 locations, # `sync` 一节的形状、地址规则 tw-config = { workspace = true } tw-types = { workspace = true } +# 通用偏好的上限 +tw-api = { workspace = true } # 配置包的格式与加密、把一份配置拆成项目与合并单位 tw-pack = { workspace = true } # S3 的 SigV4 和 Bedrock 用的是同一份 diff --git a/crates/tw-sync/src/engine.rs b/crates/tw-sync/src/engine.rs index 6ca71619..1122870c 100644 --- a/crates/tw-sync/src/engine.rs +++ b/crates/tw-sync/src/engine.rs @@ -233,6 +233,18 @@ pub struct Report { /// 同步位置上跳过了的文件:坏了的、不是这把密钥加密的、加密的位置上没加密的、等了太久 /// 还没传完的 pub skipped: u32, + /// 同步成了,但有没照办的地方,要告诉人(现在只有:通用偏好太大,没应用, + /// [`prefs_too_large_msg`]) + pub notices: Vec, +} + +/// 同步来的通用偏好超出上限、没有应用(其余照常) +pub fn prefs_too_large_msg() -> Msg { + msg!( + "sync.prefs_too_large", limit = tw_api::CLIENT_PREFS_MAX / 1024 => + "the general preferences in the sync data are larger than {limit} KiB, so this device \ + keeps its own; everything else was synced" + ) } /// 读出来的一个版本 @@ -353,7 +365,7 @@ fn local_err(e: impl std::fmt::Display) -> RoundError { } /// 内容的 SHA-256:配置原文、插件、通用偏好(不加密的版本带着它,发现传输损坏) -fn content_sha256(p: &Pack) -> String { +pub fn content_sha256(p: &Pack) -> String { let mut h = Sha256::new(); h.update(p.config_yaml.as_bytes()); for (id, f) in &p.plugins { @@ -634,6 +646,49 @@ pub fn adopted_default_key(local: &Snapshot, remote: &Snapshot) -> Option ItemId { + ItemId::new(ItemKind::ClientPrefs, "") +} + +/// 这一份里的通用偏好写不进本机:拼回 JSON 之后超出上限([`tw_pack::disk::prefs_too_large`], +/// 写偏好的地方都按它拒收) +fn prefs_oversized(s: &Snapshot) -> bool { + let Some(it) = s.items.get(&prefs_id()) else { + return false; + }; + let only = Snapshot { + items: BTreeMap::from([(prefs_id(), it.clone())]), + }; + let empty = Pack { + manifest: tw_pack::Manifest::new(PackKind::Full, PackOrigin::Sync), + config_yaml: format!("version: {}\n", tw_config::SCHEMA_VERSION), + plugins: Default::default(), + client_prefs: None, + }; + model::materialize(&empty, &only) + .ok() + .and_then(|p| p.client_prefs) + .is_some_and(|p| tw_pack::disk::prefs_too_large(&p)) +} + +/// 本机通用偏好此刻的指纹([`State::held_prefs`]) +fn prefs_print(s: &Snapshot) -> String { + let units = s.items.get(&prefs_id()).map(|it| &it.units); + let json = serde_json::to_string(&units).unwrap_or_default(); + let out: [u8; 32] = Sha256::digest(json.as_bytes()).into(); + out.iter().map(|b| format!("{b:02x}")).collect() +} + +/// 一个项目照 `from` 的样子(`from` 里没有就去掉) +fn put_item(dst: &mut Snapshot, from: &Snapshot, id: &ItemId) { + match from.items.get(id) { + Some(it) => dst.items.insert(id.clone(), it.clone()), + None => dst.items.remove(id), + }; +} + // ─────────────────────────────────────────────── 等登录的上游 /// 本机配置里的上游(带着这台设备自己的登录) @@ -1598,6 +1653,7 @@ impl Engine { return Ok(Report { encrypted: kr, conflicts: self.last_report.conflicts.clone(), + notices: self.last_report.notices.clone(), devices: self.devices().await?, skipped, ..Default::default() @@ -1645,6 +1701,16 @@ impl Engine { copy_disabled(it, &s); } } + // 同步位置上的偏好太大时这台留着自己的(见下)。之后同步位置上的变回能写的:这台的偏好 + // 一直没动过的话,不算本机改了(换回同步的样子再合并),同步位置上的照常落地 + let hold_prefs = prefs_oversized(&remote.snap); + if let Some(print) = st.held_prefs.take() + && !hold_prefs + && prefs_print(&local) == print + { + let synced = base.as_ref().map(|b| &b.snap).unwrap_or(&remote.snap); + put_item(&mut local, synced, &prefs_id()); + } // 别的设备处理掉了的冲突:照同步版本改 let open_ids: BTreeSet<&str> = remote.conflicts.iter().map(|c| c.id.as_str()).collect(); if let Some(b) = &base { @@ -1761,6 +1827,14 @@ impl Engine { for (o, _) in ORDERS { hold_items.insert(ItemId::new(*o, "")); } + // 同步位置上的通用偏好太大、写不进本机:两边各留各的 —— 本机不应用,上传的也不拿本机 + // 的去换掉别的设备的。其余照常 + let mut notices = Vec::new(); + if hold_prefs { + hold_items.insert(prefs_id()); + notices.push(prefs_too_large_msg()); + st.held_prefs = Some(prefs_print(&local_snap)); + } // 不加密时经同步停用的插件:本机启用了就放开 st.held_plugins.retain(|id, _| { let item = local.items.get(&ItemId::new(ItemKind::Plugin, id.clone())); @@ -1860,6 +1934,15 @@ impl Engine { }); } } + // 两边各自不大、合起来太大:本机留本机的,上传的留同步位置的 + if prefs_oversized(&target) { + put_item(&mut target, &local_snap, &prefs_id()); + put_item(&mut shared, &remote.snap, &prefs_id()); + if notices.is_empty() { + notices.push(prefs_too_large_msg()); + } + st.held_prefs = Some(prefs_print(&local_snap)); + } target.relink(); shared.relink(); // 没登录的上游在这台停用着 @@ -1937,6 +2020,7 @@ impl Engine { let mut report = Report { encrypted: kr, skipped, + notices, ..Default::default() }; let mut content_after = quiet.content.clone(); @@ -2407,6 +2491,14 @@ impl Engine { set_order(&mut target, *o, m.local); set_order(&mut shared, *o, m.shared); } + // 通用偏好太大、写不进本机:本机留本机的,上传的照同步数据 + let mut notices = Vec::new(); + if prefs_oversized(&target) { + put_item(&mut target, &local, &prefs_id()); + put_item(&mut shared, &remote.snap, &prefs_id()); + notices.push(prefs_too_large_msg()); + st.held_prefs = Some(prefs_print(&local)); + } target.relink(); shared.relink(); // 本机这一份:没登录的停用着;同步的样子是启用着的,记成等登录 @@ -2447,6 +2539,7 @@ impl Engine { let mut report = Report { encrypted: self.sealed, skipped: sc.skipped.len() as u32, + notices, ..Default::default() }; if model::normalized(&target) != model::normalized(&local) { diff --git a/crates/tw-sync/src/local.rs b/crates/tw-sync/src/local.rs index d1584626..955d1d60 100644 --- a/crates/tw-sync/src/local.rs +++ b/crates/tw-sync/src/local.rs @@ -64,6 +64,10 @@ pub struct State { /// 的分开算(等得更久) #[serde(skip_serializing_if = "BTreeMap::is_empty")] pub arriving: BTreeMap, + /// 同步位置上的通用偏好太大、这台留着自己的:留下时本机偏好的指纹。之后同步位置上的变回 + /// 能写的、本机的又没改过,就照同步的样子落地(本机的不算改动) + #[serde(skip_serializing_if = "Option::is_none")] + pub held_prefs: Option, /// 同步位置看上去是空的:从什么时候起、连着几轮 pub empty_since_ms: Option, pub empty_rounds: u32, diff --git a/crates/tw-sync/tests/regressions.rs b/crates/tw-sync/tests/regressions.rs index 992b5bdf..7b9bb4a7 100644 --- a/crates/tw-sync/tests/regressions.rs +++ b/crates/tw-sync/tests/regressions.rs @@ -8,7 +8,6 @@ use std::collections::BTreeSet; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; -use sha2::Digest; use tw_pack::{Pack, PackKind, PackOrigin, Seal}; use tw_sync::engine::{ ApplyError, Current, Decision, Engine, Host, RoundError, SKIP_AFTER_MS, Settings, Take, @@ -47,6 +46,11 @@ impl Dev { fn path(&self) -> PathBuf { self.dir.path().join("config.yaml") } + /// 通用偏好(和控制面一样存在配置旁边) + fn prefs(&self) -> Option { + let b = std::fs::read(self.dir.path().join("prefs.json")).ok()?; + serde_json::from_slice(&b).ok() + } fn text(&self) -> String { std::fs::read_to_string(self.path()).unwrap() } @@ -75,7 +79,7 @@ impl Host for Dev { manifest: tw_pack::Manifest::new(PackKind::Full, PackOrigin::Ui), config_yaml: text.clone(), plugins: Default::default(), - client_prefs: None, + client_prefs: self.prefs(), }, version: tw_config::version_of(&text), changed_ms: 1, @@ -86,7 +90,24 @@ impl Host for Dev { return Err(ApplyError::Stale); } tw_config::try_parse(&pack.config_yaml).map_err(|r| ApplyError::Rejected(r.msg()))?; + // 和控制面一样:写不进的偏好(超出上限)整个应用就失败 + if pack + .client_prefs + .as_ref() + .is_some_and(tw_pack::disk::prefs_too_large) + { + return Err(ApplyError::Rejected(tw_types::msg!( + "t.prefs" => "preferences too large" + ))); + } std::fs::write(self.path(), &pack.config_yaml).unwrap(); + let prefs = self.dir.path().join("prefs.json"); + match &pack.client_prefs { + Some(p) => std::fs::write(prefs, serde_json::to_vec(p).unwrap()).unwrap(), + None => { + let _ = std::fs::remove_file(prefs); + } + } Ok(()) } fn pinned_keys(&self) -> BTreeSet { @@ -183,6 +204,15 @@ fn read_version(loc: &Path, id: &str) -> Vec { /// 没有口令的人往同步位置里放一个不加密的版本(manifest.sync 照格式写、带 SHA-256),依据 /// 最新的那一个 fn forge_plain_version(loc: &Path, config_yaml: String) -> String { + forge_plain_version_with(loc, config_yaml, None) +} + +/// 同上,带着通用偏好 +fn forge_plain_version_with( + loc: &Path, + config_yaml: String, + client_prefs: Option, +) -> String { let head = versions(loc) .last() .unwrap() @@ -193,10 +223,9 @@ fn forge_plain_version(loc: &Path, config_yaml: String) -> String { manifest: tw_pack::Manifest::new(PackKind::Full, PackOrigin::Sync), config_yaml, plugins: Default::default(), - client_prefs: None, + client_prefs, }; - let sha: [u8; 32] = sha2::Sha256::digest(pack.config_yaml.as_bytes()).into(); - let sha: String = sha.iter().map(|b| format!("{b:02x}")).collect(); + let sha = tw_sync::engine::content_sha256(&pack); let meta = tw_sync::record::Meta { seq: seq + 1, device: "eeeeeeee-0000-4000-8000-00000000000e".into(), @@ -1008,6 +1037,42 @@ async fn an_icloud_placeholder_that_never_downloads_is_skipped_after_a_day() { assert!(r.uploaded.is_some()); } +// ─────────────────────────────────────────────── 太大的通用偏好 + +/// 同步位置上的通用偏好超出上限(别处写进来的,这台写不进):偏好不应用、其余照常,报 +/// `sync.prefs_too_large`,**不算失败**;之后上传的版本也不拿本机的偏好去换掉同步位置上的 +#[tokio::test] +async fn oversized_preferences_are_left_out_and_the_rest_applies() { + let mut t = bed(false).await; + let big = serde_json::json!({ "notes": "x".repeat(70 * 1024) }); + assert!(tw_pack::disk::prefs_too_large(&big)); + let cfg = add_provider(&t.a.text(), "elsewhere", "https://e.example", "sk-e"); + forge_plain_version_with(t.loc.path(), cfg, Some(big.clone())); + let r = t.eb.round().await.unwrap(); + let codes: Vec<&str> = r.notices.iter().map(|m| m.code.as_str()).collect(); + assert_eq!(codes, ["sync.prefs_too_large"]); + assert!(t.b.provider("elsewhere").is_some(), "其余照常应用"); + assert!(r.uploaded.is_none(), "不拿本机的偏好去换"); + // 下一轮(走捷径)还说着 + let r = t.eb.round().await.unwrap(); + assert_eq!(r.notices.len(), 1); + // B 改了别的:上传的版本里偏好还是同步位置上的那一份 + t.b.edit(|s| add_provider(s, "bnew", "https://bnew.example", "sk-bn")); + let up = t.eb.round().await.unwrap().uploaded.unwrap(); + let p = tw_pack::read(&read_version(t.loc.path(), &up), tw_pack::Unseal::None).unwrap(); + assert_eq!(p.client_prefs, Some(big)); + assert_eq!(t.b.prefs(), None, "B 的偏好没动"); + // 同步位置上的偏好变回不大的:照常落地(B 的偏好一直没动过,不算 B 改了),提示没了 + let small = serde_json::json!({ "notes": "short" }); + forge_plain_version_with(t.loc.path(), t.b.text(), Some(small.clone())); + let r = t.eb.round().await.unwrap(); + assert!(r.notices.is_empty(), "{:?}", r.notices); + assert!(r.conflicts.is_empty(), "{:?}", r.conflicts); + assert_eq!(t.b.prefs(), Some(small.clone())); + t.ea.round().await.unwrap(); + assert_eq!(t.a.prefs(), Some(small)); +} + // ─────────────────────────────────────────────── 空的同步位置 /// 文件夹位置的根目录还在、里面暂时是空的(网络盘没挂上时挂载点是个空目录):不当作清空,