fix(release): 完善发布事务和状态隔离

infrastructure/src/localized_patch.rs

@@ -43,7 +43,7 @@ pub const LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING: &str = "manual_proof
 pub const LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING_LABEL: &str = "人工校对中";
 
 #[derive(Debug)]
-struct LocalizedOutputLock {
+pub(crate) struct LocalizedOutputLock {
     file: std::fs::File,
 }
 
@@ -84,6 +84,10 @@ impl LocalizedOutputLock {
     }
 }
 
+pub(crate) fn acquire_localized_output_lock(root: &Path) -> anyhow::Result<LocalizedOutputLock> {
+    LocalizedOutputLock::acquire(root)
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct LocalizedReleaseTransaction {
     version: u32,
@@ -834,6 +838,7 @@ impl LocalizedPatchService {
             )
         })?;
         verify_localized_release_files(&version_path, &manifest)?;
+        verify_localized_distribution_manifest_at(&version_path, &current_release_id)?;
         if manifest.localized_release_id != current_release_id {
             return Err(anyhow::anyhow!(
                 "manifest release={} 与当前状态 release={} 不一致",
@@ -891,6 +896,10 @@ impl LocalizedPatchService {
                 })?,
             );
             verify_localized_release_files(&previous_path, restored_manifest.as_ref().unwrap())?;
+            verify_localized_distribution_manifest_at(
+                &previous_path,
+                restored_release_id.as_deref().unwrap_or_default(),
+            )?;
         }
 
         let previous_state_bytes = read_file_no_symlink(&state_path, "汉化版本状态")
@@ -2200,6 +2209,54 @@ fn verify_localized_release_files(
     Ok(())
 }
 
+fn verify_localized_distribution_manifest_at(
+    version_path: &Path,
+    localized_release_id: &str,
+) -> anyhow::Result<()> {
+    let path = version_path.join(LOCALIZED_DISTRIBUTION_MANIFEST_FILE);
+    let Some(bytes) = read_file_no_symlink(&path, "localized distribution manifest")
+        .map_err(anyhow::Error::msg)?
+    else {
+        // 旧 localized release 可能在 distribution manifest 引入前发布；
+        // 保留兼容 rollback，新的 publish 仍会在有 official manifest 时生成它。
+        return Ok(());
+    };
+    let manifest: LocalizedDistributionManifest = serde_json::from_slice(&bytes)?;
+    if manifest.version != 1 || manifest.localized_release_id != localized_release_id {
+        return Err(anyhow::anyhow!(
+            "localized distribution manifest identity 不一致：expected={} actual={}",
+            localized_release_id,
+            manifest.localized_release_id
+        ));
+    }
+    let mut destinations = BTreeSet::new();
+    for entry in &manifest.entries {
+        if !destinations.insert(entry.destination.as_str()) {
+            return Err(anyhow::anyhow!(
+                "localized distribution manifest 存在重复 destination：{}",
+                entry.destination
+            ));
+        }
+        let file_path = version_path.join(&entry.destination);
+        ensure_path_within_root(version_path, &file_path).map_err(anyhow::Error::msg)?;
+        ensure_safe_file_target(version_path, &file_path, "localized distribution 文件")
+            .map_err(anyhow::Error::msg)?;
+        let file_bytes = fs::read(&file_path)?;
+        let actual = blake3::hash(&file_bytes).to_hex().to_string();
+        if file_bytes.len() as u64 != entry.bytes || actual != entry.blake3 {
+            return Err(anyhow::anyhow!(
+                "localized distribution 文件完整性失败 {}：expected bytes={} blake3={} actual bytes={} blake3={}",
+                entry.destination,
+                entry.bytes,
+                entry.blake3,
+                file_bytes.len(),
+                actual
+            ));
+        }
+    }
+    Ok(())
+}
+
 fn write_localized_transaction(
     localized_output_root: &Path,
     transaction: &LocalizedReleaseTransaction,
@@ -2302,7 +2359,9 @@ fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result
         remove_owned_path(&transaction.version_path)?;
     } else {
         if let Some(target) = transaction.previous_current_target.as_deref() {
-            ensure_path_within_root(localized_output_root, target).map_err(anyhow::Error::msg)?;
+            let target_path = localized_output_root.join(target);
+            ensure_path_within_root(localized_output_root, &target_path)
+                .map_err(anyhow::Error::msg)?;
             restore_current_symlink(localized_output_root, &current_path, Some(target))?;
         } else if transaction.operation == "publish" {
             restore_current_symlink(localized_output_root, &current_path, None)?;
@@ -2328,6 +2387,19 @@ fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result
     remove_localized_transaction(localized_output_root)
 }
 
+pub(crate) fn recover_localized_output_transaction(root: &Path) -> anyhow::Result<()> {
+    recover_localized_transaction(root)
+}
+
+pub(crate) fn localized_output_transaction_pending(root: &Path) -> anyhow::Result<bool> {
+    Ok(read_file_no_symlink(
+        &root.join(LOCALIZED_TRANSACTION_FILE),
+        "localized release transaction",
+    )
+    .map_err(anyhow::Error::msg)?
+    .is_some())
+}
+
 /// Inspects one localized release without changing state, staging, current or
 /// any repair target.
 pub fn inspect_localized_release_artifact(
@@ -3359,6 +3431,10 @@ fn default_localized_version_state_version() -> u32 {
 #[cfg(test)]
 mod tests {
     use super::*;
+    #[cfg(unix)]
+    use std::sync::mpsc;
+    #[cfg(unix)]
+    use std::time::Duration;
     use tempfile::TempDir;
 
     fn push_u32_be(data: &mut Vec<u8>, value: u32) {
@@ -3381,6 +3457,82 @@ mod tests {
         data.extend_from_slice(&value.to_le_bytes());
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn localized_output_lock_serializes_independent_handles() {
+        let temp = TempDir::new().unwrap();
+        let first = LocalizedOutputLock::acquire(temp.path()).unwrap();
+        let (sender, receiver) = mpsc::channel();
+        let root = temp.path().to_path_buf();
+        let worker = std::thread::spawn(move || {
+            let second = LocalizedOutputLock::acquire(&root).unwrap();
+            sender.send(()).unwrap();
+            drop(second);
+        });
+
+        assert!(receiver.recv_timeout(Duration::from_millis(50)).is_err());
+        drop(first);
+        receiver.recv_timeout(Duration::from_secs(1)).unwrap();
+        worker.join().unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn interrupted_publish_transaction_is_recovered_before_next_mutation() {
+        use std::os::unix::fs::symlink;
+
+        let temp = TempDir::new().unwrap();
+        let root = temp.path().join("localized");
+        let old = root.join(LOCALIZED_VERSIONS_DIR).join("old");
+        let new = root.join(LOCALIZED_VERSIONS_DIR).join("new");
+        let staging = root.join(LOCALIZED_STAGING_DIR).join("new");
+        fs::create_dir_all(&old).unwrap();
+        fs::create_dir_all(&new).unwrap();
+        fs::create_dir_all(&staging).unwrap();
+        symlink(
+            Path::new(LOCALIZED_VERSIONS_DIR).join("new"),
+            root.join(LOCALIZED_CURRENT_LINK),
+        )
+        .unwrap();
+        let old_state = LocalizedVersionState {
+            state_version: LOCALIZED_VERSION_STATE_VERSION,
+            official_release_id: "official-old".to_string(),
+            current_release_id: Some("old".to_string()),
+            status: "localized".to_string(),
+            translation_workflow_status: None,
+            updated_unix_seconds: 1,
+        };
+        let old_state_bytes = serde_json::to_vec_pretty(&old_state).unwrap();
+        write_file_atomic(
+            &root.join(LOCALIZED_VERSION_STATE_FILE),
+            &old_state_bytes,
+            STATE_FILE_MODE,
+            "test state",
+        )
+        .unwrap();
+        write_localized_transaction(
+            &root,
+            &LocalizedReleaseTransaction::publish(
+                "new",
+                new.clone(),
+                staging.clone(),
+                Some(PathBuf::from("versions/old")),
+                Some(old_state_bytes),
+            ),
+        )
+        .unwrap();
+
+        write_localized_version_state(&root, &old_state).unwrap();
+
+        assert_eq!(
+            fs::read_link(root.join(LOCALIZED_CURRENT_LINK)).unwrap(),
+            PathBuf::from("versions/old")
+        );
+        assert!(!new.exists());
+        assert!(!staging.exists());
+        assert!(!localized_output_transaction_pending(&root).unwrap());
+    }
+
     fn push_i16_le(data: &mut Vec<u8>, value: i16) {
         data.extend_from_slice(&value.to_le_bytes());
     }