From 597b462d39f4ab7c375721833921e388e3f25dbf Mon Sep 17 00:00:00 2001 From: igor04091968 Date: Thu, 9 Jul 2026 00:26:40 +0300 Subject: [PATCH] Stabilize Hayabusa drop autoprocess --- adk-rust/Cargo.lock | 1 + adk-rust/crates/hayabusa-tools/Cargo.toml | 1 + .../hayabusa-tools/src/bin/autoprocess.rs | 213 +++++++++++++++++- aw-server/hayabusa/README.md | 2 +- 4 files changed, 207 insertions(+), 10 deletions(-) diff --git a/adk-rust/Cargo.lock b/adk-rust/Cargo.lock index 730dd19..e1556be 100644 --- a/adk-rust/Cargo.lock +++ b/adk-rust/Cargo.lock @@ -1211,6 +1211,7 @@ dependencies = [ "serde", "serde_json", "tempfile", + "zip", ] [[package]] diff --git a/adk-rust/crates/hayabusa-tools/Cargo.toml b/adk-rust/crates/hayabusa-tools/Cargo.toml index 1266bd1..b635241 100644 --- a/adk-rust/crates/hayabusa-tools/Cargo.toml +++ b/adk-rust/crates/hayabusa-tools/Cargo.toml @@ -31,6 +31,7 @@ regex.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true +zip.workspace = true [dev-dependencies] tempfile.workspace = true diff --git a/adk-rust/crates/hayabusa-tools/src/bin/autoprocess.rs b/adk-rust/crates/hayabusa-tools/src/bin/autoprocess.rs index f4502bf..8a6c186 100644 --- a/adk-rust/crates/hayabusa-tools/src/bin/autoprocess.rs +++ b/adk-rust/crates/hayabusa-tools/src/bin/autoprocess.rs @@ -1,12 +1,16 @@ use std::fs::{self, File}; use std::path::{Path, PathBuf}; use std::process::Command; +use std::thread; +use std::time::{Duration, Instant, SystemTime}; use anyhow::{Context, Result, bail}; +use chrono::Utc; use clap::Parser; use fs2::FileExt; use hayabusa_tools::{guess_host_from_filename, read_json_file}; use serde_json::{Value, json}; +use zip::ZipArchive; const LOCK_PATH: &str = "/opt/hayabusa/state/aw-hayabusa-autoprocess.lock"; const WRAPPER: &str = "/usr/local/bin/aw-hayabusa"; @@ -20,6 +24,9 @@ struct Cli { #[arg(long, default_value = "/opt/activitywatch/aw-rus-ops/drop")] drop_dir: PathBuf, + #[arg(long, default_value = "/opt/hayabusa/quarantine/drop")] + quarantine_dir: PathBuf, + #[arg(long, default_value_t = true)] once: bool, } @@ -66,15 +73,29 @@ fn run() -> Result { return Ok(0); } for zip_path in zips { - let result = process_one(&zip_path)?; - println!( - "{}", - serde_json::to_string_pretty(&json!({ - "processed": zip_path.display().to_string(), - "latest_intake": result.latest_intake, - "case_alert": result.case_alert, - }))? - ); + match process_one(&zip_path) { + Ok(result) => { + println!( + "{}", + serde_json::to_string_pretty(&json!({ + "processed": zip_path.display().to_string(), + "latest_intake": result.latest_intake, + "case_alert": result.case_alert, + }))? + ); + } + Err(err) => { + let quarantined = quarantine_drop_package(&cli.quarantine_dir, &zip_path, &err)?; + println!( + "{}", + serde_json::to_string_pretty(&json!({ + "quarantined": zip_path.display().to_string(), + "quarantine_dir": quarantined.display().to_string(), + "reason": err.to_string(), + }))? + ); + } + } } Ok(0) } @@ -97,6 +118,7 @@ fn list_zips(drop_dir: &Path) -> Result> { } fn process_one(zip_path: &Path) -> Result { + wait_for_stable_zip(zip_path)?; let sidecars = load_sidecars(zip_path)?; let host = guess_host(zip_path, &sidecars); let mode = if sidecars.mode.is_empty() { @@ -174,6 +196,85 @@ fn process_one(zip_path: &Path) -> Result { }) } +fn wait_for_stable_zip(zip_path: &Path) -> Result<()> { + wait_for_stable_zip_with( + zip_path, + Duration::from_secs(60), + Duration::from_secs(1), + Duration::from_secs(2), + 2, + ) +} + +fn wait_for_stable_zip_with( + zip_path: &Path, + max_wait: Duration, + interval: Duration, + min_modified_age: Duration, + required_stable_checks: u32, +) -> Result<()> { + let started = Instant::now(); + let mut last_len = None; + let mut stable_checks = 0; + let mut last_zip_error = None; + + loop { + let metadata = + fs::metadata(zip_path).with_context(|| format!("stat {}", zip_path.display()))?; + let len = metadata.len(); + let modified_age = metadata + .modified() + .ok() + .and_then(|modified| SystemTime::now().duration_since(modified).ok()) + .unwrap_or_default(); + + if len > 0 && last_len == Some(len) && modified_age >= min_modified_age { + stable_checks += 1; + } else { + stable_checks = 0; + } + + if stable_checks >= required_stable_checks { + match verify_zip_readable(zip_path) { + Ok(()) => return Ok(()), + Err(err) => { + last_zip_error = Some(err.to_string()); + stable_checks = 0; + } + } + } + + if started.elapsed() >= max_wait { + if let Some(err) = last_zip_error { + bail!( + "drop zip {} did not become readable: {err}", + zip_path.display() + ); + } + bail!( + "drop zip {} did not stabilize within {}s", + zip_path.display(), + max_wait.as_secs() + ); + } + + last_len = Some(len); + thread::sleep(interval); + } +} + +fn verify_zip_readable(zip_path: &Path) -> Result<()> { + let file = File::open(zip_path).with_context(|| format!("open {}", zip_path.display()))?; + let mut archive = + ZipArchive::new(file).with_context(|| format!("read zip {}", zip_path.display()))?; + for idx in 0..archive.len() { + let _entry = archive + .by_index(idx) + .with_context(|| format!("read zip entry {idx}"))?; + } + Ok(()) +} + fn load_sidecars(zip_path: &Path) -> Result { let base = zip_path.with_extension(""); let caseid_path = base.with_extension("caseid"); @@ -214,6 +315,49 @@ fn load_sidecars(zip_path: &Path) -> Result { }) } +fn quarantine_drop_package( + quarantine_dir: &Path, + zip_path: &Path, + err: &anyhow::Error, +) -> Result { + fs::create_dir_all(quarantine_dir) + .with_context(|| format!("create {}", quarantine_dir.display()))?; + let timestamp = Utc::now().format("%Y%m%dT%H%M%SZ"); + let zip_name = zip_path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("package.zip"); + let target_dir = quarantine_dir.join(format!("{timestamp}_{zip_name}")); + fs::create_dir_all(&target_dir).with_context(|| format!("create {}", target_dir.display()))?; + + let base = zip_path.with_extension(""); + for path in [ + zip_path.to_path_buf(), + base.with_extension("caseid"), + base.with_extension("meta.json"), + zip_path.with_extension("zip.sha256"), + ] { + if path.is_file() { + let target = target_dir.join(path.file_name().context("quarantine file name")?); + fs::rename(&path, &target) + .with_context(|| format!("move {} to {}", path.display(), target.display()))?; + } + } + + let reason = json!({ + "quarantined_at": Utc::now().to_rfc3339(), + "package": zip_path.display().to_string(), + "reason": err.to_string(), + "error_chain": format!("{err:#}"), + }); + fs::write( + target_dir.join("reason.json"), + serde_json::to_vec_pretty(&reason).context("serialize quarantine reason")?, + ) + .with_context(|| format!("write {}", target_dir.join("reason.json").display()))?; + Ok(target_dir) +} + fn archive_sidecars(report_dir: &Path, sidecars: &Sidecars) -> Result<()> { let target_dir = report_dir.join("input-sidecars"); fs::create_dir_all(&target_dir).with_context(|| format!("create {}", target_dir.display()))?; @@ -302,3 +446,54 @@ fn run_capture(program: &Path, args: &[String]) -> Result { stderr: String::from_utf8_lossy(&output.stderr).to_string(), }) } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use tempfile::tempdir; + use zip::write::SimpleFileOptions; + + #[test] + fn wait_for_stable_zip_accepts_complete_archive() { + let dir = tempdir().expect("tempdir"); + let zip_path = dir.path().join("HOST-20260709-000001.zip"); + write_test_zip(&zip_path); + + wait_for_stable_zip_with( + &zip_path, + Duration::from_secs(1), + Duration::from_millis(1), + Duration::from_secs(0), + 1, + ) + .expect("complete zip should be accepted"); + } + + #[test] + fn wait_for_stable_zip_rejects_unreadable_archive() { + let dir = tempdir().expect("tempdir"); + let zip_path = dir.path().join("HOST-20260709-000001.zip"); + fs::write(&zip_path, b"not a zip").expect("write partial zip"); + + let err = wait_for_stable_zip_with( + &zip_path, + Duration::from_millis(20), + Duration::from_millis(1), + Duration::from_secs(0), + 1, + ) + .expect_err("invalid zip should be rejected"); + assert!(err.to_string().contains("did not become readable")); + } + + fn write_test_zip(path: &Path) { + let file = File::create(path).expect("create zip"); + let mut zip = zip::ZipWriter::new(file); + zip.start_file("manifest.json", SimpleFileOptions::default()) + .expect("start manifest"); + zip.write_all(br#"{"host":"SHARKON2025"}"#) + .expect("write manifest"); + zip.finish().expect("finish zip"); + } +} diff --git a/aw-server/hayabusa/README.md b/aw-server/hayabusa/README.md index bf17b5d..bf7cfbb 100644 --- a/aw-server/hayabusa/README.md +++ b/aw-server/hayabusa/README.md @@ -118,4 +118,4 @@ Production scheduled task on `SHARKON2025`: Do not switch this task back to `SYSTEM` on the current RDP host: Task Scheduler starts `powershell.exe` under `SYSTEM`, but the process exits with `0xC0000142` before the upload script starts. -Server-side processing accepts Windows zip packages with backslash path separators and UTF-8 BOM in sidecar JSON. `aw-hayabusa-autoprocess` processes the full incoming queue after accepting a drop package, so stale incoming files from an earlier failed run are drained before the latest intake is recorded. +Server-side processing accepts Windows zip packages with backslash path separators and UTF-8 BOM in sidecar JSON. `aw-hayabusa-autoprocess` waits for a dropped ZIP to become size-stable and readable before accepting it, then processes the full incoming queue. This prevents `systemd.path` from quarantining an archive while SCP is still writing it. ZIPs that remain unreadable after the bounded wait are moved to `/opt/hayabusa/quarantine/drop` with `reason.json`.