perf improved hash plugin

2026-01-18 16:47:41 +01:00
parent d39a59c73f
commit 443fd3c6e6
23 changed files with 2851 additions and 0 deletions
@@ -0,0 +1,66 @@
+use axum::{extract::State, Json};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use crate::api::process::ProcessState;
+use crate::config::PluginConfig;
+
+#[derive(Deserialize)]
+pub struct ConfigureRequest {
+    pub config: HashMap<String, String>,
+}
+
+#[derive(Serialize)]
+pub struct ConfigureResponse {
+    pub status: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+pub async fn configure(
+    State(state): State<Arc<ProcessState>>,
+    Json(req): Json<ConfigureRequest>,
+) -> Json<ConfigureResponse> {
+    let mut cfg = state.config.write().await;
+
+    // Handle enabledHashes config
+    if let Some(hashes) = req.config.get("enabledHashes") {
+        let enabled = PluginConfig::parse_enabled_hashes(hashes);
+        if enabled.is_empty() {
+            return Json(ConfigureResponse {
+                status: "error",
+                error: Some("No valid hash algorithms specified".to_string()),
+            });
+        }
+        cfg.enabled_hashes = enabled;
+        tracing::info!("Updated enabled hashes: {:?}", cfg.enabled_hashes);
+    }
+
+    // Handle buffer size config
+    if let Some(size) = req.config.get("bufferSize") {
+        match size.parse::<usize>() {
+            Ok(s) if s >= 4096 => {
+                cfg.buffer_size = s;
+                tracing::info!("Updated buffer size: {}", s);
+            }
+            Ok(_) => {
+                return Json(ConfigureResponse {
+                    status: "error",
+                    error: Some("Buffer size must be at least 4096 bytes".to_string()),
+                });
+            }
+            Err(_) => {
+                return Json(ConfigureResponse {
+                    status: "error",
+                    error: Some("Invalid buffer size".to_string()),
+                });
+            }
+        }
+    }
+
+    Json(ConfigureResponse {
+        status: "ok",
+        error: None,
+    })
+}
@@ -0,0 +1,17 @@
+use axum::Json;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct HealthResponse {
+    pub status: &'static str,
+    pub ready: bool,
+    pub version: &'static str,
+}
+
+pub async fn health() -> Json<HealthResponse> {
+    Json(HealthResponse {
+        status: "healthy",
+        ready: true,
+        version: env!("CARGO_PKG_VERSION"),
+    })
+}
@@ -0,0 +1,99 @@
+use axum::Json;
+use serde::Serialize;
+use serde_json::{json, Value};
+
+#[derive(Serialize)]
+pub struct ManifestResponse {
+    pub id: &'static str,
+    pub name: &'static str,
+    pub version: &'static str,
+    pub description: &'static str,
+    pub author: &'static str,
+    pub dependencies: Vec<String>,
+    pub priority: u32,
+    pub color: &'static str,
+    #[serde(rename = "defaultQueue")]
+    pub default_queue: &'static str,
+    pub timeout: u64,
+    pub schema: Value,
+    pub config: Value,
+}
+
+pub async fn manifest() -> Json<ManifestResponse> {
+    Json(ManifestResponse {
+        id: "fast-full-hash",
+        name: "Fast Full File Hash (Rust)",
+        version: env!("CARGO_PKG_VERSION"),
+        description: "High-performance parallel hashing with optimized I/O",
+        author: "MetaMesh",
+        dependencies: vec![],
+        priority: 100,
+        color: "#FF5722",
+        default_queue: "background",
+        timeout: 300000,
+        schema: json!({
+            "cid_sha2-256": {
+                "label": "SHA-256 Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "SHA-256 hash in CID format (multicodec 0x12)"
+            },
+            "cid_sha1": {
+                "label": "SHA-1 Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "SHA-1 hash in CID format (multicodec 0x11)"
+            },
+            "cid_md5": {
+                "label": "MD5 Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "MD5 hash in CID format (multicodec 0xd5)"
+            },
+            "cid_crc32": {
+                "label": "CRC32 Checksum",
+                "type": "cid",
+                "readonly": true,
+                "hint": "CRC32 checksum in CID format (multicodec 0x0132)"
+            },
+            "cid_sha3-256": {
+                "label": "SHA3-256 Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "SHA3-256 hash in CID format (multicodec 0x16)"
+            },
+            "cid_sha3-384": {
+                "label": "SHA3-384 Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "SHA3-384 hash in CID format (multicodec 0x15)"
+            },
+            "cid_midhash256": {
+                "label": "MidHash256",
+                "type": "cid",
+                "readonly": true,
+                "hint": "O(1) file identification hash - SHA-256 of size + middle 1MB (custom multicodec 0x1000)"
+            },
+            "cid_bt_pieces_root": {
+                "label": "BitTorrent v2 Pieces Root",
+                "type": "cid",
+                "readonly": true,
+                "hint": "Merkle tree root of 16KB blocks (official multicodec 0xb702)"
+            },
+            "cid_bt_info_hash": {
+                "label": "BitTorrent v2 Info Hash",
+                "type": "cid",
+                "readonly": true,
+                "hint": "SHA-256 of bencoded info dict for magnet links (multicodec 0x12)"
+            }
+        }),
+        config: json!({
+            "enabledHashes": {
+                "type": "string",
+                "label": "Enabled hash algorithms (comma-separated)",
+                "default": "cid_sha2-256,cid_sha1,cid_md5,cid_crc32,cid_sha3-256,cid_sha3-384,cid_midhash256,cid_bt_pieces_root,cid_bt_info_hash",
+                "required": false
+            }
+        }),
+    })
+}
@@ -0,0 +1,9 @@
+mod configure;
+mod health;
+mod manifest;
+mod process;
+
+pub use configure::configure;
+pub use health::health;
+pub use manifest::manifest;
+pub use process::{process, ProcessState};
@@ -0,0 +1,249 @@
+use axum::{extract::State, Json};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Instant;
+
+use crate::cache::CacheManager;
+use crate::client::MetaCoreClient;
+use crate::config::{HashAlgorithm, SharedConfig};
+use crate::hasher::hash_file;
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProcessRequest {
+    pub task_id: String,
+    pub cid: String,
+    pub file_path: String,
+    pub callback_url: String,
+    pub meta_core_url: String,
+    #[serde(default)]
+    pub existing_meta: HashMap<String, String>,
+}
+
+#[derive(Serialize)]
+pub struct ProcessResponse {
+    pub status: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CallbackPayload {
+    pub task_id: String,
+    pub status: &'static str,
+    pub duration: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+}
+
+pub struct ProcessState {
+    pub config: SharedConfig,
+    pub cache: Arc<CacheManager>,
+}
+
+pub async fn process(
+    State(state): State<Arc<ProcessState>>,
+    Json(req): Json<ProcessRequest>,
+) -> Json<ProcessResponse> {
+    // Validate required fields
+    if req.task_id.is_empty() || req.file_path.is_empty() || req.callback_url.is_empty() {
+        return Json(ProcessResponse {
+            status: "rejected",
+            error: Some("Missing required fields".to_string()),
+        });
+    }
+
+    // Spawn async processing task
+    let state_clone = Arc::clone(&state);
+    tokio::spawn(async move {
+        process_file(state_clone, req).await;
+    });
+
+    Json(ProcessResponse {
+        status: "accepted",
+        error: None,
+    })
+}
+
+async fn process_file(state: Arc<ProcessState>, req: ProcessRequest) {
+    let start = Instant::now();
+    let task_id = req.task_id.clone();
+    let callback_url = req.callback_url.clone();
+
+    let result = do_process_file(state, req).await;
+
+    let duration = start.elapsed().as_millis() as u64;
+
+    let payload = match result {
+        Ok(ProcessResult::Completed) => CallbackPayload {
+            task_id,
+            status: "completed",
+            duration,
+            error: None,
+            reason: None,
+        },
+        Ok(ProcessResult::Skipped(reason)) => CallbackPayload {
+            task_id,
+            status: "skipped",
+            duration,
+            error: None,
+            reason: Some(reason),
+        },
+        Err(e) => CallbackPayload {
+            task_id,
+            status: "failed",
+            duration,
+            error: Some(e.to_string()),
+            reason: None,
+        },
+    };
+
+    // Send callback
+    if let Err(e) = send_callback(&callback_url, &payload).await {
+        tracing::error!("Failed to send callback: {}", e);
+    }
+}
+
+enum ProcessResult {
+    Completed,
+    Skipped(String),
+}
+
+async fn do_process_file(
+    state: Arc<ProcessState>,
+    req: ProcessRequest,
+) -> anyhow::Result<ProcessResult> {
+    let path = Path::new(&req.file_path);
+
+    // Check if file exists
+    if !path.exists() {
+        anyhow::bail!("File not found: {}", req.file_path);
+    }
+
+    // Get file metadata
+    let metadata = std::fs::metadata(path)?;
+    let size = metadata.len();
+    let mtime = metadata
+        .modified()
+        .ok()
+        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or(0);
+
+    let filename = path
+        .file_name()
+        .and_then(|n| n.to_str())
+        .unwrap_or("unknown")
+        .to_string();
+
+    // Get config
+    let config = state.config.read().await.clone();
+
+    // Determine which hashes we need to compute
+    let required_keys: Vec<&str> = config
+        .enabled_hashes
+        .iter()
+        .map(|a| a.key())
+        .collect();
+
+    // Check existing metadata - skip if all hashes already present
+    let missing_from_meta: Vec<&str> = required_keys
+        .iter()
+        .filter(|k| !req.existing_meta.contains_key(**k))
+        .copied()
+        .collect();
+
+    if missing_from_meta.is_empty() {
+        return Ok(ProcessResult::Skipped(
+            "All hashes already computed".to_string(),
+        ));
+    }
+
+    // Check cache for remaining hashes
+    let cached = state.cache.get(&filename, size, mtime).await;
+    let mut results: HashMap<String, String> = HashMap::new();
+
+    let to_compute: Vec<HashAlgorithm> = missing_from_meta
+        .iter()
+        .filter_map(|key| {
+            if let Some(ref cached_hashes) = cached {
+                if let Some(hash) = cached_hashes.get(*key) {
+                    results.insert(key.to_string(), hash.clone());
+                    return None;
+                }
+            }
+            HashAlgorithm::from_key(key)
+        })
+        .collect();
+
+    // If all hashes were in cache, we're done
+    if to_compute.is_empty() {
+        tracing::info!("All hashes from cache for {}", filename);
+    } else {
+        // Compute missing hashes
+        tracing::info!(
+            "Computing {} hashes for {} ({} bytes)",
+            to_compute.len(),
+            filename,
+            size
+        );
+
+        let enabled_set = to_compute.iter().copied().collect();
+        let computed = tokio::task::spawn_blocking({
+            let path = path.to_path_buf();
+            let buffer_size = config.buffer_size;
+            move || hash_file(&path, &enabled_set, buffer_size)
+        })
+        .await??;
+
+        // Merge computed hashes into results
+        for (key, value) in &computed {
+            results.insert(key.clone(), value.clone());
+        }
+
+        // Update cache with new hashes
+        let mut all_hashes = cached.unwrap_or_default();
+        all_hashes.extend(computed);
+        state
+            .cache
+            .insert(filename.clone(), size, mtime, all_hashes)
+            .await;
+
+        // Save cache periodically (could be optimized with debouncing)
+        if let Err(e) = state.cache.save().await {
+            tracing::warn!("Failed to save cache: {}", e);
+        }
+    }
+
+    // Store results to meta-core
+    if !results.is_empty() {
+        match MetaCoreClient::new(&req.meta_core_url) {
+            Ok(client) => {
+                if let Err(e) = client.merge_metadata(&req.cid, &results).await {
+                    tracing::warn!("Failed to store metadata to meta-core: {}", e);
+                }
+            }
+            Err(e) => {
+                tracing::warn!("Failed to create meta-core client: {}", e);
+            }
+        }
+    }
+
+    Ok(ProcessResult::Completed)
+}
+
+async fn send_callback(url: &str, payload: &CallbackPayload) -> anyhow::Result<()> {
+    let client = reqwest::Client::new();
+    let resp = client.post(url).json(payload).send().await?;
+
+    if !resp.status().is_success() {
+        anyhow::bail!("Callback failed with status {}", resp.status());
+    }
+
+    Ok(())
+}