feat: harden official sync scheduling and integrity

infrastructure/src/official_download.rs

@@ -276,6 +276,22 @@ impl OfficialLocalManifestAuditReport {
     }
 }
 
+/// Existing local state for an official pull plan.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct OfficialLocalResourceState {
+    /// Number of expected URLs that already have download-manifest entries.
+    pub manifest_entry_count: usize,
+    /// Number of expected URL destinations that already exist on disk.
+    pub existing_file_count: usize,
+}
+
+impl OfficialLocalResourceState {
+    /// Returns true when this plan has any reusable or auditable local state.
+    pub fn has_any_resources(self) -> bool {
+        self.manifest_entry_count > 0 || self.existing_file_count > 0
+    }
+}
+
 /// Executes an official pull plan with the system `curl` command.
 #[derive(Debug, Clone)]
 pub struct OfficialResourcePullService {
@@ -357,6 +373,9 @@ impl OfficialResourcePullService {
         let urls = plan.all_urls()?;
         let total = urls.len();
         let mut items = Vec::new();
+        let mut verified_hashes = Vec::new();
+        let mut verified_hash_urls = HashSet::<String>::new();
+        let mut processed_urls = HashSet::<String>::new();
         for (offset, url) in urls.into_iter().enumerate() {
             if should_cancel() {
                 return Err("official resource pull interrupted by shutdown request".to_string());
@@ -406,6 +425,14 @@ impl OfficialResourcePullService {
                 result.bytes,
                 result.transferred_bytes,
             ));
+            processed_urls.insert(url.clone());
+            self.verify_ready_official_hashes(
+                &official_hash_pairs,
+                &processed_urls,
+                &mut verified_hash_urls,
+                &mut verified_hashes,
+                &mut manifest,
+            )?;
             items.push(OfficialResourcePullItem {
                 url,
                 destination,
@@ -418,8 +445,7 @@ impl OfficialResourcePullService {
         if should_cancel() {
             return Err("official resource pull interrupted by shutdown request".to_string());
         }
-
-        let verified_hashes = self.verify_downloaded_official_hashes(&official_hash_pairs)?;
+        self.verify_all_official_hashes_are_complete(&official_hash_pairs, &verified_hash_urls)?;
 
         Ok(OfficialResourcePullReport {
             items,
@@ -447,6 +473,34 @@ impl OfficialResourcePullService {
         Ok(OfficialLocalManifestAuditReport { items })
     }
 
+    /// Returns whether the current output root already contains local state for
+    /// this plan. This is intentionally cheaper than a full audit and is used
+    /// to choose between audit-then-repair and first-time full pull flows.
+    pub fn local_resource_state(
+        &self,
+        plan: &OfficialResourcePullPlan,
+    ) -> Result<OfficialLocalResourceState, String> {
+        let manifest = self.read_download_manifest()?;
+        let mut manifest_entry_count = 0;
+        let mut existing_file_count = 0;
+
+        for url in plan.all_urls()? {
+            if manifest.entries.contains_key(&url) {
+                manifest_entry_count += 1;
+            }
+
+            let destination = self.destination_for_url(&url)?;
+            if file_len_if_exists(&destination)?.is_some() {
+                existing_file_count += 1;
+            }
+        }
+
+        Ok(OfficialLocalResourceState {
+            manifest_entry_count,
+            existing_file_count,
+        })
+    }
+
     /// Fetches an official URL into memory.
     ///
     /// This is intended for small discovery inputs such as `server-info`,
@@ -604,37 +658,103 @@ impl OfficialResourcePullService {
         Ok(())
     }
 
-    fn verify_downloaded_official_hashes(
+    fn verify_ready_official_hashes(
         &self,
         pairs: &[OfficialSeedHashPair],
-    ) -> Result<Vec<OfficialResourceHashVerification>, String> {
-        let mut verified = Vec::new();
-
+        processed_urls: &HashSet<String>,
+        verified_hash_urls: &mut HashSet<String>,
+        verified_hashes: &mut Vec<OfficialResourceHashVerification>,
+        manifest: &mut OfficialDownloadManifest,
+    ) -> Result<(), String> {
         for pair in pairs {
-            let data_path = self.destination_for_url(&pair.data_url)?;
-            let hash_path = self.destination_for_url(&pair.hash_url)?;
-            let data = fs::read(&data_path).map_err(|error| {
-                format!(
-                    "Failed to read official hash data file {}: {error}",
-                    data_path.display()
-                )
-            })?;
-            let hash = fs::read(&hash_path).map_err(|error| {
-                format!(
-                    "Failed to read official hash sidecar {}: {error}",
-                    hash_path.display()
-                )
-            })?;
+            if verified_hash_urls.contains(&pair.hash_url) {
+                continue;
+            }
 
-            verified.push(verify_official_seed_catalog_hash(
-                &pair.data_url,
-                &data,
-                &pair.hash_url,
-                &hash,
-            )?);
+            if !processed_urls.contains(&pair.data_url) || !processed_urls.contains(&pair.hash_url)
+            {
+                continue;
+            }
+
+            if !self.hash_pair_files_exist(pair)? {
+                continue;
+            }
+
+            let verification = match self.verify_official_seed_hash_pair_from_disk(pair) {
+                Ok(verification) => verification,
+                Err(error) => {
+                    if let Err(cleanup_error) =
+                        self.invalidate_download_manifest_hash_pair(manifest, pair)
+                    {
+                        return Err(format!(
+                            "{error}; additionally failed to invalidate local manifest entries for official hash pair: {cleanup_error}"
+                        ));
+                    }
+
+                    return Err(error);
+                }
+            };
+
+            verified_hashes.push(verification);
+            verified_hash_urls.insert(pair.hash_url.clone());
         }
 
-        Ok(verified)
+        Ok(())
+    }
+
+    fn hash_pair_files_exist(&self, pair: &OfficialSeedHashPair) -> Result<bool, String> {
+        let data_path = self.destination_for_url(&pair.data_url)?;
+        let hash_path = self.destination_for_url(&pair.hash_url)?;
+        Ok(file_len_if_exists(&data_path)?.is_some() && file_len_if_exists(&hash_path)?.is_some())
+    }
+
+    fn verify_official_seed_hash_pair_from_disk(
+        &self,
+        pair: &OfficialSeedHashPair,
+    ) -> Result<OfficialResourceHashVerification, String> {
+        let data_path = self.destination_for_url(&pair.data_url)?;
+        let hash_path = self.destination_for_url(&pair.hash_url)?;
+        let data = fs::read(&data_path).map_err(|error| {
+            format!(
+                "Failed to read official hash data file {}: {error}",
+                data_path.display()
+            )
+        })?;
+        let hash = fs::read(&hash_path).map_err(|error| {
+            format!(
+                "Failed to read official hash sidecar {}: {error}",
+                hash_path.display()
+            )
+        })?;
+
+        verify_official_seed_catalog_hash(&pair.data_url, &data, &pair.hash_url, &hash)
+    }
+
+    fn invalidate_download_manifest_hash_pair(
+        &self,
+        manifest: &mut OfficialDownloadManifest,
+        pair: &OfficialSeedHashPair,
+    ) -> Result<(), String> {
+        manifest.entries.remove(&pair.data_url);
+        manifest.entries.remove(&pair.hash_url);
+        self.write_download_manifest(manifest)
+    }
+
+    fn verify_all_official_hashes_are_complete(
+        &self,
+        pairs: &[OfficialSeedHashPair],
+        verified_hash_urls: &HashSet<String>,
+    ) -> Result<(), String> {
+        for pair in pairs {
+            if !verified_hash_urls.contains(&pair.hash_url) {
+                return Err(format!(
+                    "official hash pair was not verified for {} using {}",
+                    pair.data_url, pair.hash_url
+                ));
+            }
+        }
+
+        Ok(())
     }
 
     fn read_download_manifest(&self) -> Result<OfficialDownloadManifest, String> {
@@ -1607,6 +1727,43 @@ fi
             .all(|item| item.status == OfficialResourcePullStatus::Downloaded));
     }
 
+    #[test]
+    fn local_resource_state_reports_manifest_entries_and_existing_files() {
+        let out_dir = TempDir::new().unwrap();
+        let bin_dir = TempDir::new().unwrap();
+        let curl_path = bin_dir.path().join("curl");
+        write_fake_curl(&curl_path);
+
+        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
+        let plan = build_official_pull_plan(discovery_plan(), inventory());
+        let urls = plan.all_urls().unwrap();
+
+        let empty_state = service.local_resource_state(&plan).unwrap();
+        assert_eq!(empty_state.manifest_entry_count, 0);
+        assert_eq!(empty_state.existing_file_count, 0);
+        assert!(!empty_state.has_any_resources());
+
+        let first_destination = service.destination_for_url(&urls[0]).unwrap();
+        fs::create_dir_all(first_destination.parent().unwrap()).unwrap();
+        fs::write(&first_destination, b"local-only").unwrap();
+
+        let file_only_state = service.local_resource_state(&plan).unwrap();
+        assert_eq!(file_only_state.manifest_entry_count, 0);
+        assert_eq!(file_only_state.existing_file_count, 1);
+        assert!(file_only_state.has_any_resources());
+
+        let mut manifest = OfficialDownloadManifest::default();
+        service
+            .record_download_manifest_entry(&mut manifest, &urls[0], &first_destination)
+            .unwrap();
+        service.write_download_manifest(&manifest).unwrap();
+
+        let manifest_state = service.local_resource_state(&plan).unwrap();
+        assert_eq!(manifest_state.manifest_entry_count, 1);
+        assert_eq!(manifest_state.existing_file_count, 1);
+        assert!(manifest_state.has_any_resources());
+    }
+
     #[test]
     fn redownloads_existing_file_when_manifest_hash_mismatches() {
         let out_dir = TempDir::new().unwrap();
@@ -1677,6 +1834,17 @@ fi
 
         let error = service.pull(&plan).unwrap_err();
         assert!(error.contains("official hash mismatch"));
+
+        let manifest = service.read_download_manifest().unwrap();
+        assert!(manifest.entries.is_empty());
+
+        let audit = service.audit_local_manifest(&plan).unwrap();
+        assert!(!audit.is_clean());
+        assert_eq!(audit.repair_needed_count(), 2);
+        assert!(audit
+            .items
+            .iter()
+            .all(|item| { item.status == OfficialLocalManifestAuditStatus::MissingManifestEntry }));
     }
 
     #[test]