ParrotXray/Mantis
1
0
Fork 0
mirror of https://github.com/ParrotXray/Mantis.git synced 2026-08-24 19:00:27 +09:00
Code Issues Packages Projects Releases Wiki Activity

Bound pending-flow backlog to avoid unbounded growth under sustained overload

Browse Source 
The backlog queue added to fix batch truncation only helps for short
bursts. Under sustained overload (drain rate consistently exceeding
inference_batch_size every tick), the queue would grow without bound,
costing memory and pushing detection latency ever higher.

Cap the backlog at batch_size * 10 and drop the oldest entries past
that cap -- they have waited longest and are the least time-relevant
for detection. Log at ERROR (FlowBacklogDropped) so sustained overload
is visible and distinguishable from the WARN-level transient backlog
log. The drop happens on the shared pending_flows queue before the
CSV/inference branch, so both sinks still see identical data.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138PxtKH73hqxv7h1oaoSdS
This commit is contained in:
Claude 2026-07-02 05:01:12 +00:00
parent af58b1cafd
commit 48f1747de3
No known key found for this signature in database
2 changed files with 25 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
mantis/src/detection/ml/engine.rs
View File

@ -39,6 +39,10 @@ pub struct Engine {
}
impl Engine {
// How many ticks' worth of batch_size the pending-flow backlog may hold
// before the oldest entries are dropped to bound memory and latency.
const PENDING_QUEUE_CAP_MULTIPLIER: usize = 10;
pub fn new(
models: Option<Arc<MLModels>>,
config: Arc<InferenceConfig>,
@ -119,6 +123,24 @@ impl Engine {
continue;
};
pending.extend(drained);
// Sustained overload (inflow consistently exceeds batch_size, not just
// a one-tick burst): the backlog would otherwise grow without bound,
// costing memory and pushing detection latency ever higher. Cap it and
// drop the oldest entries -- they have waited longest and are the
// least time-relevant for detection. Applies identically to the CSV
// and inference sinks since both read from this same queue.
let max_pending = self.batch_size.saturating_mul(Self::PENDING_QUEUE_CAP_MULTIPLIER);
if pending.len() > max_pending {
let overflow = pending.len() - max_pending;
pending.drain(..overflow);
log!(MLLog::FlowBacklogDropped {
dropped: overflow,
max_pending,
batch_size: self.batch_size,
});
}
let take = pending.len().min(self.batch_size);
let batch: Vec<FlowData> = pending.drain(..take).collect();
if !pending.is_empty() {

3
mantis/src/model/log/ml.rs
View File

@ -66,6 +66,9 @@ loggable! {
#[error("Flow backlog carried to next tick: {backlog} flows (batch_size={batch_size} too small for current load)")]
FlowBacklog { backlog: usize, batch_size: usize } => tracing::Level::WARN,
#[error("Sustained overload: dropped {dropped} oldest backlogged flows to stay under pending cap of {max_pending} (batch_size={batch_size} cannot keep up with inflow)")]
FlowBacklogDropped { dropped: usize, max_pending: usize, batch_size: usize } => tracing::Level::ERROR,
#[error("{model} inference failed: {error}")]
InferenceFailed { model: String, error: String } => tracing::Level::INFO,