Fix memory plateau after daily virus-database reload

Cargo.toml

@@ -33,6 +33,11 @@ tower-http = { version = "=0.6.2", features = ["limit", "trace", "timeout"] }
 tracing = "=0.1.41"
 tracing-subscriber = { version = "=0.3.19", features = ["env-filter"] }
 
+# jemalloc returns unused pages after a CVD hot-reload; glibc typically does not.
+[target.'cfg(target_os = "linux")'.dependencies]
+tikv-jemallocator = { version = "=0.7.0", features = ["background_threads", "unprefixed_malloc_on_supported_platforms"] }
+tikv-jemalloc-ctl = "=0.7.0"
+
 [dev-dependencies]
 criterion = { version = "=0.5.1", features = ["html_reports", "async_tokio"] }
 http-body-util = "=0.1.2"

Dockerfile

@@ -34,7 +34,8 @@ ENV DEFENDER_LISTEN=0.0.0.0:8080 \
     DEFENDER_UPDATE_INTERVAL_SECS=3600 \
     DEFENDER_MAX_BYTES=67108864 \
     DEFENDER_USER_AGENT="ClamAV/1.4.2 (defender; docker)" \
-    RUST_LOG=info
+    RUST_LOG=info \
+    MALLOC_CONF=background_thread:true,dirty_decay_ms:1000,muzzy_decay_ms:1000
 
 USER defender
 EXPOSE 8080

README.md

@@ -73,11 +73,13 @@ A background task (same process, independent Tokio task):
 1. `Range: bytes=0-511` against configured mirrors to read the remote CVD header
 2. Downloads `main.cvd` / `daily.cvd` when the version/MD5 is newer
 3. Verifies MD5 + RSA
-4. Compiles a **new** `Engine` off the request path
-5. Atomically swaps it with `arc-swap`
+4. Compiles a **new** `Engine` off the request path (streaming CVD members so the gzip body is not held alongside the compiled signatures)
+5. Atomically swaps it with `arc-swap`, then returns unused heap pages to the OS
 
 In-flight scans keep the previous `Arc<Engine>` until they finish. No connection drop, no restart.
 
+A daily `daily.cvd` publish is the usual trigger. RSS will rise while both engines exist, then fall back after the swap (`rss_before` / `rss_compiled` / `rss_after` on the reload log line).
+
 Default mirrors:
 
 - `https://database.clamav.net`
@@ -162,7 +164,7 @@ Official ClamAV CVD files as of 18 Aug 2026 (from `database.clamav.net`):
 | `bytecode.cvd` | **0.27 MiB** (281,702 B) | 1.24 MiB | 80 | Not executed (no bytecode VM) |
 | **Total baked** | **107.56 MiB** | **308.8 MiB** | | Image includes `main` + `daily` |
 
-Loaded into the scanner (main + daily, PUA off): ~540k file hashes, ~102k body signatures, ~307k logical signatures. Resident set with that engine is about **1.4 GiB**.
+Loaded into the scanner (main + daily, PUA off): ~540k file hashes, ~102k body signatures, ~307k logical signatures. Resident set with that engine is about **1.4 GiB**. The Aho-Corasick prefilter uses a contiguous NFA (not a DFA) so a database reload cannot balloon the automaton. After a hot-swap, jemalloc purges unused dirty pages so RSS does not stay at the two-engine peak.
 
 ## Development
 
@@ -220,7 +222,7 @@ Official `daily.cvd` over the same HTTP path:
 | `POST /scan` clean 64 KiB | 667 µs | 93.7 MiB/s |
 | `POST /scan` EICAR ×16 concurrent | 296 µs / batch | **54.0 k req/s** |
 
-Tiny requests are loopback/HTTP-latency bound; large bodies are bounded by MD5+SHA1+SHA256. RSS tests (`tests/memory.rs`) stay stable across 20k scans and 200 engine swaps.
+Tiny requests are loopback/HTTP-latency bound; large bodies are bounded by MD5+SHA1+SHA256. RSS tests (`tests/memory.rs`) stay stable across 20k scans and 200 engine swaps. Reloading official CVDs briefly overlaps the previous engine; unused pages are returned after the swap.
 
 ## Architecture
 

src/alloc.rs

@@ -0,0 +1,79 @@
+//! Process allocator helpers.
+//!
+//! On Linux the binary uses jemalloc so that dropping a compiled [`crate::engine::Engine`]
+//! actually returns pages to the OS. glibc `malloc` commonly keeps the old
+//! arenas mapped, which is what a CVD hot-reload looks like as a permanent RSS
+//! step-up (old engine + new engine, then “stuck” at ~1.5×).
+
+#[cfg(target_os = "linux")]
+#[global_allocator]
+static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+
+/// Enable jemalloc’s background purge thread. Safe to call more than once.
+pub fn init() {
+    #[cfg(target_os = "linux")]
+    {
+        if let Err(e) = tikv_jemalloc_ctl::background_thread::write(true) {
+            tracing::debug!(error = ?e, "jemalloc background_thread not enabled");
+        }
+    }
+}
+
+/// Return unused heap pages to the OS.
+///
+/// Call after compiling an engine (scratch buffers from unpack/AC construction)
+/// and again after swapping so the previous engine’s pages can be unmapped.
+pub fn reclaim_unused_pages() {
+    #[cfg(target_os = "linux")]
+    {
+        let _ = tikv_jemalloc_ctl::epoch::advance();
+        // `MALLCTL_ARENAS_ALL` is `u32::MAX`: purge every arena, including the
+        // tokio blocking-pool thread that compiled the engine.
+        let rc = unsafe { tikv_jemalloc_ctl::raw::write(b"arena.4294967295.purge\0", ()) };
+        if let Err(e) = rc {
+            tracing::debug!(error = ?e, "jemalloc arena purge failed");
+        }
+    }
+}
+
+/// Current resident set in bytes (`VmRSS`), if `/proc` is available.
+pub fn rss_bytes() -> Option<u64> {
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    for line in status.lines() {
+        if let Some(rest) = line.strip_prefix("VmRSS:") {
+            let kb: u64 = rest.split_whitespace().next()?.parse().ok()?;
+            return Some(kb.saturating_mul(1024));
+        }
+    }
+    None
+}
+
+/// Format a byte count for logs (`123.4 MiB`).
+pub fn format_bytes(n: u64) -> String {
+    const MIB: f64 = 1024.0 * 1024.0;
+    if n >= 1024 * 1024 {
+        format!("{:.1} MiB", n as f64 / MIB)
+    } else if n >= 1024 {
+        format!("{:.1} KiB", n as f64 / 1024.0)
+    } else {
+        format!("{n} B")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn format_bytes_scales() {
+        assert_eq!(format_bytes(512), "512 B");
+        assert_eq!(format_bytes(2048), "2.0 KiB");
+        assert_eq!(format_bytes(2 * 1024 * 1024), "2.0 MiB");
+    }
+
+    #[test]
+    fn reclaim_does_not_panic() {
+        reclaim_unused_pages();
+        let _ = rss_bytes();
+    }
+}

src/cvd/header.rs

@@ -1,5 +1,9 @@
 //! 512-byte ClamAV CVD/CLD header parser.
 
+use std::fs::File;
+use std::io::Read;
+use std::path::Path;
+
 use crate::error::{Error, Result};
 
 /// Size of the fixed CVD header prefix.
@@ -101,6 +105,15 @@ impl CvdHeader {
         })
     }
 
+    /// Read only the 512-byte header from `path` (does not load the gzip body).
+    pub fn read_file(path: impl AsRef<Path>) -> Result<Self> {
+        let path = path.as_ref();
+        let mut file = File::open(path).map_err(|e| Error::io(path, e))?;
+        let mut buf = [0u8; CVD_HEADER_SIZE];
+        file.read_exact(&mut buf).map_err(|e| Error::io(path, e))?;
+        Self::parse(&buf)
+    }
+
     /// Serialize back to a 512-byte padded header.
     pub fn to_bytes(&self) -> [u8; CVD_HEADER_SIZE] {
         let s = format!(
@@ -197,4 +210,16 @@ mod tests {
         let h2 = CvdHeader::parse(&bytes).unwrap();
         assert_eq!(h, h2);
     }
+
+    #[test]
+    fn read_file_ignores_body() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("daily.cvd");
+        let h = CvdHeader::parse_str(SAMPLE).unwrap();
+        let mut bytes = h.to_bytes().to_vec();
+        bytes.extend_from_slice(&[0u8; 1024 * 1024]);
+        std::fs::write(&path, &bytes).unwrap();
+        let loaded = CvdHeader::read_file(&path).unwrap();
+        assert_eq!(loaded, h);
+    }
 }

src/cvd/mod.rs

@@ -15,7 +15,7 @@ pub mod unpack;
 pub mod verify;
 
 pub use header::CvdHeader;
-pub use unpack::{unpack_cvd, UnpackedDb};
+pub use unpack::{for_each_cvd_member, unpack_cvd, UnpackedDb};
 pub use verify::{verify_cvd, verify_cvd_bytes, VerifyMode};
 
 use crate::error::Result;

src/cvd/unpack.rs

@@ -1,13 +1,16 @@
 //! Unpack the gzip+tar body that follows a 512-byte CVD header.
 
 use std::collections::BTreeMap;
+use std::fs::File;
 use std::io::{Cursor, Read};
+use std::path::Path;
 
 use flate2::read::GzDecoder;
 use tar::Archive;
 
 use super::header::CVD_HEADER_SIZE;
 use crate::error::{Error, Result};
+use crate::signatures::is_signature_member;
 
 /// Files extracted from a CVD/CLD archive, keyed by file name (no path).
 #[derive(Debug, Clone, Default)]
@@ -39,11 +42,43 @@ pub fn unpack_cvd(data: &[u8]) -> Result<UnpackedDb> {
 
 pub fn unpack_body(body: &[u8]) -> Result<UnpackedDb> {
     let gz = GzDecoder::new(Cursor::new(body));
-    let mut archive = Archive::new(gz);
+    let mut files = BTreeMap::new();
+    for_each_archive_file(gz, true, true, |name, data| {
+        files.insert(name, data);
+        Ok(())
+    })?;
+    Ok(UnpackedDb { files })
+}
+
+/// Stream CVD tar members from disk without retaining the compressed file or
+/// previously visited members. The 512-byte header is skipped; callers should
+/// already have authenticated the file.
+///
+/// Non-signature members (bytecode, YARA, …) and PUA files (when `load_pua` is
+/// false) are consumed and discarded without keeping their contents.
+pub fn for_each_cvd_member(
+    path: impl AsRef<Path>,
+    load_pua: bool,
+    mut visit: impl FnMut(&str, &[u8]) -> Result<()>,
+) -> Result<()> {
+    let path = path.as_ref();
+    let mut file = File::open(path).map_err(|e| Error::io(path, e))?;
+    let mut hdr = [0u8; CVD_HEADER_SIZE];
+    file.read_exact(&mut hdr).map_err(|e| Error::io(path, e))?;
+    let gz = GzDecoder::new(file);
+    for_each_archive_file(gz, false, load_pua, |name, data| visit(&name, &data))
+}
+
+fn for_each_archive_file<R: Read>(
+    reader: R,
+    load_all: bool,
+    load_pua: bool,
+    mut visit: impl FnMut(String, Vec<u8>) -> Result<()>,
+) -> Result<()> {
+    let mut archive = Archive::new(reader);
     archive.set_overwrite(false);
     archive.set_preserve_permissions(false);
 
-    let mut files = BTreeMap::new();
     let entries = archive
         .entries()
         .map_err(|e| Error::CvdUnpack(e.to_string()))?;
@@ -59,13 +94,18 @@ pub fn unpack_body(body: &[u8]) -> Result<UnpackedDb> {
             .and_then(|s| s.to_str())
             .ok_or_else(|| Error::CvdUnpack("non-utf8 member name".into()))?
             .to_string();
+        if !load_all && !is_signature_member(&name, load_pua) {
+            std::io::copy(&mut entry, &mut std::io::sink())
+                .map_err(|e| Error::CvdUnpack(e.to_string()))?;
+            continue;
+        }
         let mut buf = Vec::new();
         entry
             .read_to_end(&mut buf)
             .map_err(|e| Error::CvdUnpack(e.to_string()))?;
-        files.insert(name, buf);
+        visit(name, buf)?;
     }
-    Ok(UnpackedDb { files })
+    Ok(())
 }
 
 /// Build a synthetic CVD body (gzip tar) from name → contents. Header is not included.
@@ -140,4 +180,33 @@ mod tests {
         assert_eq!(unpacked.get("test.hdb").unwrap(), files[0].1);
         assert_eq!(unpacked.get("test.ndb").unwrap(), files[1].1);
     }
+
+    #[test]
+    fn for_each_member_matches_unpack() {
+        let files = [
+            (
+                "test.hdb",
+                b"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:1:Eicar-Test-File\n".as_slice(),
+            ),
+            ("test.ndb", b"Eicar:0:*:585530\n".as_slice()),
+            ("skip.cbc", b"not-a-signature".as_slice()),
+        ];
+        let cvd = pack_cvd(&files, 1, "unit").unwrap();
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("t.cvd");
+        std::fs::write(&path, &cvd).unwrap();
+
+        let mut seen = Vec::new();
+        for_each_cvd_member(&path, false, |name, data| {
+            seen.push((name.to_string(), data.to_vec()));
+            Ok(())
+        })
+        .unwrap();
+        seen.sort_by(|a, b| a.0.cmp(&b.0));
+        assert_eq!(seen.len(), 2);
+        assert_eq!(seen[0].0, "test.hdb");
+        assert_eq!(seen[1].0, "test.ndb");
+        assert_eq!(seen[0].1, files[0].1);
+        assert_eq!(seen[1].1, files[1].1);
+    }
 }