Sync: oversized preferences are left out instead of failing every round; CLI explains keep and keep-both

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<serde_json::Value> {
         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"))?;

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

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`（地址经同步变更，改之前的地址）。

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<Msg>,
 }
 
 /// `POST /sync/test`、`POST /sync/enable/preview`：连上去看看。

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,
         }
     }
 

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<String> = 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(), "其余照常应用");
+}

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 用的是同一份

crates/tw-sync/src/engine.rs

@@ -233,6 +233,18 @@ pub struct Report {
     /// 同步位置上跳过了的文件：坏了的、不是这把密钥加密的、加密的位置上没加密的、等了太久
     /// 还没传完的
     pub skipped: u32,
+    /// 同步成了，但有没照办的地方，要告诉人（现在只有：通用偏好太大，没应用，
+    /// [`prefs_too_large_msg`]）
+    pub notices: Vec<Msg>,
+}
+
+/// 同步来的通用偏好超出上限、没有应用（其余照常）
+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
     (key_value(l) != key_value(r)).then_some(id)
 }
 
+// ─────────────────────────────────────────────── 太大的通用偏好
+
+fn prefs_id() -> 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<S: Store, H: Host> Engine<S, H> {
             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<S: Store, H: Host> Engine<S, H> {
                 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<S: Store, H: Host> Engine<S, H> {
         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<S: Store, H: Host> Engine<S, H> {
                 });
             }
         }
+        // 两边各自不大、合起来太大：本机留本机的，上传的留同步位置的
+        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<S: Store, H: Host> Engine<S, H> {
         let mut report = Report {
             encrypted: kr,
             skipped,
+            notices,
             ..Default::default()
         };
         let mut content_after = quiet.content.clone();
@@ -2407,6 +2491,14 @@ impl<S: Store, H: Host> Engine<S, H> {
             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<S: Store, H: Host> Engine<S, H> {
         let mut report = Report {
             encrypted: self.sealed,
             skipped: sc.skipped.len() as u32,
+            notices,
             ..Default::default()
         };
         if model::normalized(&target) != model::normalized(&local) {

crates/tw-sync/src/local.rs

@@ -64,6 +64,10 @@ pub struct State {
     /// 的分开算（等得更久）
     #[serde(skip_serializing_if = "BTreeMap::is_empty")]
     pub arriving: BTreeMap<String, u64>,
+    /// 同步位置上的通用偏好太大、这台留着自己的：留下时本机偏好的指纹。之后同步位置上的变回
+    /// 能写的、本机的又没改过，就照同步的样子落地（本机的不算改动）
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub held_prefs: Option<String>,
     /// 同步位置看上去是空的：从什么时候起、连着几轮
     pub empty_since_ms: Option<u64>,
     pub empty_rounds: u32,