fix(release): 收紧分发热路径与事务边界

infrastructure/src/official_download.rs

@@ -25,6 +25,7 @@ use std::fs::{self, File};
 use std::io::Read;
 use std::path::{Path, PathBuf};
 use std::process::Command;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -74,6 +75,23 @@ const OFFICIAL_CAS_REUSE_REFERENCES_VERSION: u32 = 1;
 const DOWNLOAD_MANIFEST_VERSION: u32 = 1;
 const DOWNLOAD_QUARANTINE_VERSION: u32 = 1;
 const DEFAULT_RETRY_ATTEMPTS: usize = 3;
+static CAS_OWNERSHIP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
+
+fn new_cas_ownership_id(output_root: &Path) -> String {
+    let sequence = CAS_OWNERSHIP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_nanos())
+        .unwrap_or_default();
+    let material = format!(
+        "{}:{}:{}:{}",
+        output_root.display(),
+        std::process::id(),
+        now,
+        sequence
+    );
+    format!("cas-owner-{}", blake3::hash(material.as_bytes()).to_hex())
+}
 
 /// Outcome for one official resource pull item.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -353,6 +371,12 @@ pub struct OfficialResourceReuseWarning {
 pub struct OfficialCasReuseReferenceManifest {
     /// 引用清单版本。
     pub version: u32,
+    /// Persistent ownership identity for this release generation.
+    ///
+    /// `None` is an explicit legacy manifest marker. Legacy cleanup keeps the
+    /// historical basename key and never invents a new identity while loading.
+    #[serde(default)]
+    pub ownership_id: Option<String>,
     /// 每个 CAS 引用一项。允许重复，因为每个拉取项分别拥有一个引用。
     pub object_ids: Vec<String>,
 }
@@ -361,6 +385,7 @@ impl Default for OfficialCasReuseReferenceManifest {
     fn default() -> Self {
         Self {
             version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+            ownership_id: None,
             object_ids: Vec::new(),
         }
     }
@@ -393,15 +418,17 @@ pub fn read_cas_reuse_reference_manifest_at(
 
 /// Decrements and removes CAS references recorded for a release.
 ///
-/// Each decrement is committed together with a durable `(release, ordinal)`
-/// ownership record in CAS metadata. The release-local manifest remains a
-/// resumable progress cursor, so a crash before its rewrite cannot decrement
-/// the same ownership twice.
+/// Each decrement is committed together with a durable `(ownership, ordinal)`
+/// record in CAS metadata. The release-local manifest remains a resumable
+/// progress cursor, so a crash before its rewrite cannot decrement the same
+/// ownership twice. Legacy manifests without `ownership_id` intentionally use
+/// the historical release-basename key and are never assigned a new identity
+/// during cleanup.
 pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Result<usize, String> {
     let Some(mut manifest) = read_cas_reuse_reference_manifest_at(release_root)? else {
         return Ok(0);
     };
-    let release_id = release_root
+    let legacy_release_id = release_root
         .file_name()
         .and_then(|name| name.to_str())
         .filter(|name| !name.is_empty() && *name != "." && *name != "..")
@@ -412,6 +439,7 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res
             )
         })?
         .to_string();
+    let ownership_id = manifest.ownership_id.clone().unwrap_or(legacy_release_id);
     let objects_root = cas_root.join("objects");
     let metadata_path = cas_root.join("metadata.sqlite");
     require_existing_directory(cas_root, "CAS 根目录")?;
@@ -422,14 +450,14 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res
         let ordinal = manifest.object_ids.len() as u64;
         let cas_root = cas_root.to_path_buf();
         let object_id_for_runtime = object_id.clone();
-        let release_id_for_runtime = release_id.clone();
+        let ownership_id_for_runtime = ownership_id.clone();
         let runtime = tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()
             .map_err(|error| format!("创建 CAS 引用清理 runtime 失败：{error}"))?;
         let did_release = runtime.block_on(async move {
             let cas = crate::FileSystemCasRepository::new(cas_root);
-            cas.release_reference_once(&release_id_for_runtime, ordinal, &object_id_for_runtime)
+            cas.release_reference_once(&ownership_id_for_runtime, ordinal, &object_id_for_runtime)
                 .await
                 .map_err(|error| format!("减少 CAS release 引用失败 object={object_id}：{error}"))
         })?;
@@ -857,8 +885,14 @@ impl OfficialResourcePullService {
         if object_ids.is_empty() {
             return Ok(());
         }
-        let mut manifest =
-            read_cas_reuse_reference_manifest_at(&self.output_root)?.unwrap_or_default();
+        let mut manifest = match read_cas_reuse_reference_manifest_at(&self.output_root)? {
+            Some(manifest) => manifest,
+            None => OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: Some(new_cas_ownership_id(&self.output_root)),
+                object_ids: Vec::new(),
+            },
+        };
         manifest.object_ids.extend(object_ids);
         write_cas_reuse_reference_manifest(&self.output_root, &manifest)
     }
@@ -4662,7 +4696,9 @@ exit 22
         let references = read_cas_reuse_reference_manifest_at(&out_dir)
             .unwrap()
             .unwrap();
+        assert!(references.ownership_id.is_some());
         assert_eq!(references.object_ids, vec![object_id.clone()]);
+        let ownership_id = references.ownership_id.clone().unwrap();
         assert_eq!(cas_reference_count(&cas_root, &object_id), 2);
 
         assert_eq!(
@@ -4680,6 +4716,7 @@ exit 22
             out_dir.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
             serde_json::to_vec(&OfficialCasReuseReferenceManifest {
                 version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: Some(ownership_id),
                 object_ids: vec![object_id.clone()],
             })
             .unwrap(),
@@ -4695,6 +4732,75 @@ exit 22
             .is_none());
     }
 
+    #[test]
+    fn cas_ownership_identity_isolated_for_same_release_basename() {
+        let temp = TempDir::new().unwrap();
+        let cas_root = temp.path().join("cas");
+        let object_id = store_cas_object(&cas_root, b"shared object");
+        let cas = crate::FileSystemCasRepository::new(&cas_root);
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                cas.add_reference(&object_id).await.unwrap();
+                cas.add_reference(&object_id).await.unwrap();
+            });
+
+        let first = temp.path().join("first").join("release");
+        let second = temp.path().join("second").join("release");
+        fs::create_dir_all(&first).unwrap();
+        fs::create_dir_all(&second).unwrap();
+        for (root, ownership_id) in [(&first, "owner-first"), (&second, "owner-second")] {
+            fs::write(
+                root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+                serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                    version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                    ownership_id: Some(ownership_id.to_string()),
+                    object_ids: vec![object_id.clone()],
+                })
+                .unwrap(),
+            )
+            .unwrap();
+        }
+
+        assert_eq!(release_cas_reuse_references(&first, &cas_root).unwrap(), 1);
+        assert_eq!(cas_reference_count(&cas_root, &object_id), 2);
+        assert_eq!(release_cas_reuse_references(&second, &cas_root).unwrap(), 1);
+        assert_eq!(cas_reference_count(&cas_root, &object_id), 1);
+    }
+
+    #[test]
+    fn legacy_cas_manifest_uses_basename_key_without_migration() {
+        let temp = TempDir::new().unwrap();
+        let cas_root = temp.path().join("cas");
+        let object_id = store_cas_object(&cas_root, b"legacy object");
+        let release_root = temp.path().join("legacy-release");
+        fs::create_dir_all(&release_root).unwrap();
+        fs::write(
+            release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+            serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: None,
+                object_ids: vec![object_id.clone()],
+            })
+            .unwrap(),
+        )
+        .unwrap();
+
+        let cas = crate::FileSystemCasRepository::new(&cas_root);
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async { cas.add_reference(&object_id).await.unwrap() });
+        assert_eq!(
+            release_cas_reuse_references(&release_root, &cas_root).unwrap(),
+            1
+        );
+        assert_eq!(cas_reference_count(&cas_root, &object_id), 1);
+    }
+
     #[test]
     fn corrupted_cas_falls_back_to_network_with_diagnostic() {
         let temp = TempDir::new().unwrap();