TC-3071 Add support for the xz compression format.

Author: bxf12315

Cargo.lock

@@ -24,6 +24,7 @@ tokio = { version = "1.43.1", features = ["sync"] }
 urlencoding = "2"
 packageurl = "0.4.2"
 rand = "0.9.2"
+  xz2 = "0.1"
 
 [features]
 default = ["postgres"]

Cargo.toml

@@ -114,5 +114,5 @@
   ],
   "download_advisory": "24ae57c3-4b57-4f4e-82c1-83ae26059a89",
   "get_advisory": "24ae57c3-4b57-4f4e-82c1-83ae26059a89",
-  "upload_advisory_files": "./test-data/advisories/rhba-2024_11505.json"
+  "upload_advisory_files": "./test-data/advisories/rhba-2024_11505.json.xz"
 }

scenarios/full-20250604.json5

@@ -242,7 +242,7 @@ async fn main() -> Result<(), anyhow::Error> {
             .set_weight(1)?;
             if let Some(file) = &scenario.upload_advisory_files {
                 // Use sequential transaction execution to ensure immediate deletion after upload
-                s = s.register_transaction(tx!(upload_and_immediately_delete(file.clone()),name: format!("upload_and_immediately_delete")));
+                s = s.register_transaction(tx!(upload_and_immediately_delete(file.clone()),name: "upload_and_immediately_delete"));
             }
             s
         })

src/main.rs

@@ -2,15 +2,20 @@ use crate::utils::DisplayVec;
 use goose::goose::{GooseUser, TransactionError, TransactionResult};
 use rand::Rng;
 use serde_json::json;
-use std::sync::{
-    Arc,
-    atomic::{AtomicUsize, Ordering},
+use std::{
+    collections::HashMap,
+    io::Read,
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
 };
-use tokio::sync::OnceCell;
+use tokio::sync::{OnceCell, RwLock};
 use urlencoding::encode;
+use xz2::read::XzDecoder;
 
-/// Cache for base advisory content with file path
-static BASE_ADVISORY_CONTENT: OnceCell<(String, Vec<u8>)> = OnceCell::const_new();
+/// Thread-safe cache for multiple advisory files using async-aware RwLock
+static FILE_CACHE: OnceCell<RwLock<HashMap<String, Vec<u8>>>> = OnceCell::const_new();
 
 /// Generate a new advisory with modified ID
 fn generate_advisory_content(base_content: &[u8]) -> Result<Vec<u8>, Box<TransactionError>> {
@@ -42,32 +47,65 @@ fn generate_advisory_content(base_content: &[u8]) -> Result<Vec<u8>, Box<Transac
     })
 }
 
-/// Get base advisory content from file path - cached and initialized on first access
-async fn get_base_advisory_content(file_path: &str) -> Result<&'static Vec<u8>, String> {
-    let result = BASE_ADVISORY_CONTENT
-        .get_or_init(|| async {
-            match tokio::fs::read(file_path).await {
-                Ok(content) => (file_path.to_string(), content),
-                Err(e) => {
-                    panic!("Failed to load base advisory file {}: {}", file_path, e);
-                }
-            }
-        })
+/// Get base upload content from file path - cached and initialized on first access
+async fn get_base_upload_content(file_path: &str) -> Result<Vec<u8>, String> {
+    // Initialize cache if not already done
+    let cache = FILE_CACHE
+        .get_or_init(|| async { RwLock::new(HashMap::new()) })
         .await;
 
-    let (cached_path, content) = result;
+    // Check if the file is already cached
+    {
+        let cache_read = cache.read().await;
+        if let Some(content) = cache_read.get(file_path) {
+            return Ok(content.clone());
+        }
+    }
+
+    // Load the file content
+    let content = read_upload_file(file_path)
+        .await
+        .map_err(|e| format!("Failed to load base advisory file {}: {}", file_path, e))?;
 
-    // Verify that the cached path matches the requested path
-    if cached_path != file_path {
-        return Err(format!(
-            "Base advisory file path mismatch: expected {}, got {}",
-            file_path, cached_path
-        ));
+    // Insert into cache
+    {
+        let mut cache_write = cache.write().await;
+        cache_write.insert(file_path.to_string(), content.clone());
     }
 
     Ok(content)
 }
 
+/// Read upload file, handling both regular JSON and XZ compressed files
+async fn read_upload_file(file_path: &str) -> Result<Vec<u8>, String> {
+    let file_bytes = tokio::fs::read(file_path)
+        .await
+        .map_err(|e| format!("Failed to read file {}: {}", file_path, e))?;
+
+    // Check if file is XZ compressed by looking at the magic bytes
+    if file_path.ends_with(".xz") || is_xz_compressed(&file_bytes) {
+        decompress_xz(&file_bytes)
+    } else {
+        Ok(file_bytes)
+    }
+}
+
+/// Check if data is XZ compressed by examining magic bytes
+fn is_xz_compressed(data: &[u8]) -> bool {
+    data.len() >= 6 && data[0..6] == [0xFD, 0x37, 0x7A, 0x58, 0x5A, 0x00]
+}
+
+/// Decompress XZ compressed data
+fn decompress_xz(compressed_data: &[u8]) -> Result<Vec<u8>, String> {
+    let mut decoder = XzDecoder::new(compressed_data);
+    let mut decompressed = Vec::new();
+
+    decoder
+        .read_to_end(&mut decompressed)
+        .map_err(|e| format!("Failed to decompress XZ data: {}", e))?;
+    Ok(decompressed)
+}
+
 /// Upload advisory data and extract advisory ID from response
 async fn upload_advisory_and_extract_id(
     file_bytes: Vec<u8>,
@@ -120,16 +158,16 @@ pub async fn list_advisory_labels(user: &mut GooseUser) -> TransactionResult {
 
 /// Upload file and return advisory ID
 pub async fn upload_advisory_and_get_id(
-    advisory_file: String,
+    advisory_file: &str,
     user: &mut GooseUser,
 ) -> Result<String, Box<TransactionError>> {
     // Check cache first - only cache the base template
-    let base_content = get_base_advisory_content(&advisory_file)
+    let base_content = get_base_upload_content(advisory_file)
         .await
         .map_err(|e| Box::new(TransactionError::Custom(e)))?;
 
     // Generate new content with modified ID
-    let generated_content = generate_advisory_content(base_content)?;
+    let generated_content = generate_advisory_content(&base_content)?;
 
     // Upload the generated content and extract advisory ID
     upload_advisory_and_extract_id(generated_content, user).await
@@ -148,7 +186,7 @@ pub async fn upload_and_immediately_delete(
     user: &mut GooseUser,
 ) -> TransactionResult {
     // 1. Upload file and get ID
-    let advisory_id = upload_advisory_and_get_id(advisory_file.clone(), user).await?;
+    let advisory_id = upload_advisory_and_get_id(&advisory_file, user).await?;
 
     // 2. Immediately delete (no waiting required)
     delete_advisory_by_id(advisory_id, user).await?;

src/restapi.rs

@@ -409,7 +409,7 @@ FROM public.advisory order by modified desc limit 1;"#,
         let files: Vec<PathBuf> = entries
             .filter_map(|entry| {
                 let path = entry.ok()?.path();
-                if path.extension()? == "json" {
+                if path.extension()? == "xz" {
                     Some(path)
                 } else {
                     None